shared

package
v1.4.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 28, 2026 License: AGPL-3.0 Imports: 24 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	DefaultBufSize = 32768 // 32KB for write buffers - reduces write syscalls
	ReadBufSize    = 65536 // 64KB read buffer - better pipelining support

	ResponseBufSize = 8192 // 8KB response buffer initial size
)

Buffer sizes for optimal performance Larger buffers reduce syscall frequency at the cost of memory per connection

View Source 
const (
	RESPSimpleString = '+'
	RESPError        = '-'
	RESPInteger      = ':'
	RESPBulkString   = '$'
	RESPArray        = '*'
)

RESP2 data type prefixes

View Source 
const (
	RESP3Null           = '_' // Null: _\r\n
	RESP3Double         = ',' // Double: ,1.23\r\n
	RESP3Boolean        = '#' // Boolean: #t\r\n or #f\r\n
	RESP3BlobError      = '!' // Blob Error: !<len>\r\n<data>\r\n
	RESP3VerbatimString = '=' // Verbatim: =<len>\r\n<encoding>:<data>\r\n
	RESP3BigNumber      = '(' // Big Number: (<number>\r\n
	RESP3Map            = '%' // Map: %<count>\r\n<key><value>...
	RESP3Set            = '~' // Set: ~<count>\r\n<elements>...
	RESP3Push           = '>' // Push: ><count>\r\n<elements>...
	RESP3Attribute      = '|' // Attribute: |<count>\r\n<key><value>...<actual_reply>
)

RESP3 data type prefixes

View Source 
const (
	// PubsubMsgChanSize is the buffer size for the pub/sub message channel
	PubsubMsgChanSize = 1000
)

Variables

View Source 
var (
	EffectsNodeID pb.NodeID
	HlcClock      *crdt.HLC
)
View Source 
var (
	ErrInvalidProtocol   = errors.New("ERR invalid protocol")
	ErrInvalidBulkLength = errors.New("ERR invalid bulk length")
	ErrInvalidArrayCount = errors.New("ERR invalid array count")
	ErrLineTooLong       = errors.New("ERR line too long")
	ErrUnexpectedEOF     = errors.New("ERR unexpected end of input")
)

Protocol errors

View Source 
var (
	// MinStreamID is the minimum possible stream ID
	MinStreamID = StreamID{Ms: 0, Seq: 0}
	// MaxStreamID is the maximum possible stream ID
	MaxStreamID = StreamID{Ms: ^uint64(0), Seq: ^uint64(0)}

	ErrInvalidStreamID  = errors.New("invalid stream ID specified as stream command argument")
	ErrStreamIDTooSmall = errors.New("the ID specified in XADD is equal or smaller than the target stream top item")
)
View Source 
var ErrNoChannels = errors.New("register called with zero channels")

Functions

func ClearPubSubBroker 

func ClearPubSubBroker()

ClearPubSubBroker fully clears the global pub/sub broker pointer.

func ClearScriptingEngine 

func ClearScriptingEngine()

ClearScriptingEngine fully clears the global scripting engine pointer.

func CommandCount 

func CommandCount() int

CommandCount returns the number of registered command names.

func CommandHasCategory 

func CommandHasCategory(cmd CommandType, cat CommandCategory) bool

CommandHasCategory checks if a command belongs to a category

func CommandNamesList 

func CommandNamesList(fn func(name string, cmd CommandType) bool)

CommandNamesList calls fn for each registered command name (uppercase) and its type. If fn returns false, iteration stops.

func EncodeElementID 

func EncodeElementID(hlc time.Time, nodeID pb.NodeID) []byte

EncodeElementID encodes an (hlc, nodeID) pair into a 16-byte element ID.

func GetAllCategoryNames 

func GetAllCategoryNames() []string

GetAllCategoryNames returns all category names

func GetCategoryName 

func GetCategoryName(cat CommandCategory) string

GetCategoryName returns the name of a category

func GetDataBuf 

func GetDataBuf(n int) []byte

GetDataBuf gets a byte buffer of at least size n from the appropriate pool

func GetDefaultIDMPDuration 

func GetDefaultIDMPDuration() int64

func GetDefaultIDMPMaxSize 

func GetDefaultIDMPMaxSize() int64

func GetNodeID 

func GetNodeID() pb.NodeID

GetNodeID returns this node's ID.

func GetProtoMaxBulkLen 

func GetProtoMaxBulkLen() int64

GetProtoMaxBulkLen returns the current proto-max-bulk-len value.

func GetStreamNodeMaxEntries 

func GetStreamNodeMaxEntries() int64

GetStreamNodeMaxEntries returns the current stream-node-max-entries value.

func HLLHash 

func HLLHash(data []byte) uint64

HLLHash computes a 64-bit hash for the HyperLogLog algorithm using MurmurHash3

func HLLRho 

func HLLRho(bits uint64) uint8

HLLRho counts the number of leading zeros + 1 in the given bits This is used to determine the register value for HyperLogLog

func InitEffectsGlobals 

func InitEffectsGlobals(nodeID pb.NodeID)

InitEffectsGlobals initializes the global HLC clock and node ID. Must be called at server startup before any commands are processed.

func KeysAfterStreams 

func KeysAfterStreams(cmd *Command) []string

KeysAfterStreams extracts keys after the STREAMS keyword. Used by XREAD and XREADGROUP.

func KeysAll 

func KeysAll(cmd *Command) []string

KeysAll extracts all args as keys. Used by DEL, DELEX, PFMERGE, SINTERSTORE, SUNIONSTORE, SDIFFSTORE.

func KeysAllButLast 

func KeysAllButLast(cmd *Command) []string

KeysAllButLast extracts keys from all args except the last (timeout). Used by BLPOP, BRPOP, BZPOPMIN, BZPOPMAX.

func KeysFirst 

func KeysFirst(cmd *Command) []string

KeysFirst extracts the first arg as a key.

func KeysFirstTwo 

func KeysFirstTwo(cmd *Command) []string

KeysFirstTwo extracts the first two args as keys (source, destination). Used by BLMOVE, BRPOPLPUSH.

func KeysMSetStyle 

func KeysMSetStyle(cmd *Command) []string

KeysMSetStyle extracts keys from MSET-style commands where args are key-value pairs. Every other arg starting at index 0 is a key.

func KeysNumkeysAtOffset 

func KeysNumkeysAtOffset(cmd *Command, offset int) []string

KeysNumkeysAtOffset extracts keys using a numkeys argument at a given offset. Used by BLMPOP (offset=1), BZMPOP (offset=1), EVAL/EVALSHA/FCALL (offset=1).

func KeysSortCmd 

func KeysSortCmd(cmd *Command) []string

KeysSortCmd extracts keys from SORT/SORT_RO (source key + optional STORE destination).

func NextHLC 

func NextHLC() time.Time

NextHLC returns a monotonically increasing timestamp.

func NextToken 

func NextToken(b []byte) (token, rest []byte)

NextToken returns the next space-delimited token and the remaining bytes

func ParseBlockingTimeout 

func ParseBlockingTimeout(s string) (float64, string)

ParseBlockingTimeout parses a timeout string for blocking commands. Returns the timeout in seconds and an error message if invalid. Handles floats, integers, and hex values (0x...).

func ParseFloat64 

func ParseFloat64(b []byte) (float64, bool)

ParseFloat64 parses a float64 from a byte slice

func ParseInt64 

func ParseInt64(b []byte) (int64, bool)

ParseInt64 parses an int64 from a byte slice without allocation, rejecting overflow.

func ParseUint64 

func ParseUint64(b []byte) (uint64, bool)

ParseUint64 parses a uint64 from a byte slice without allocation

func PutCommand 

func PutCommand(cmd *Command)

PutCommand returns a Command to the pool

func PutDataBuf 

func PutDataBuf(buf []byte)

PutDataBuf returns a byte buffer to the appropriate pool. The caller yields ownership of buf's full capacity.

func PutWriter 

func PutWriter(w *Writer)

PutWriter returns a Writer to the pool

func ReconstructBytes 

func ReconstructBytes(snap *pb.ReducedEffect) ([]byte, bool)

ReconstructBytes attempts to convert a snapshot to raw bytes. Returns (bytes, true) if a reconstructor exists, (nil, false) otherwise.

func RegisterBytesReconstructor 

func RegisterBytesReconstructor(vt pb.ValueType, fn BytesReconstructor)

func RegisterModuleCommands 

func RegisterModuleCommands(entries ...ModuleEntry)

RegisterModuleCommands appends command entries from an external module. Called from init() functions in module packages (e.g., redis/zset).

func SetDefaultIDMPDuration 

func SetDefaultIDMPDuration(v int64)

func SetDefaultIDMPMaxSize 

func SetDefaultIDMPMaxSize(v int64)

func SetProtoMaxBulkLen 

func SetProtoMaxBulkLen(v int64)

SetProtoMaxBulkLen updates the proto-max-bulk-len value atomically.

func SetPubSubBroker 

func SetPubSubBroker(b PubSubBroker)

SetPubSubBroker sets the pub/sub broker instance.

func SetPubSubClusterRouter added in v0.3.0 

func SetPubSubClusterRouter(r PubSubClusterRouter)

SetPubSubClusterRouter installs the cluster router. Called by the cluster wiring at startup; pass nil to clear.

func SetScriptingEngine 

func SetScriptingEngine(e ScriptingEngine)

SetScriptingEngine sets the scripting engine instance.

func SetServerStats 

func SetServerStats(s *Stats)

SetServerStats sets the server stats instance atomically.

func SetStreamNodeMaxEntries 

func SetStreamNodeMaxEntries(v int64)

SetStreamNodeMaxEntries updates the stream-node-max-entries value atomically.

func ToUpper 

func ToUpper(b []byte) []byte

ToUpper converts bytes to uppercase in-place style. Returns a new slice if conversion is needed (to avoid modifying original).

Types

type ACLSelector 

type ACLSelector struct {
	Commands    map[CommandType]bool
	Categories  map[CommandCategory]bool
	AllCommands bool
	KeyPatterns []KeyPattern
	AllKeys     bool
}

ACLSelector represents an additional permission set that can be applied conditionally

type ACLUser 

type ACLUser struct {
	Name           string
	Enabled        bool
	PasswordHashes [][]byte // bcrypt hashes (includes salt)
	NoPass         bool     // allow without password

	// Command permissions
	AllCommands bool                            // +@all
	Categories  map[CommandCategory]bool        // +@read, -@write, etc.
	Commands    map[CommandType]bool            // +get, -set, etc.
	Subcommands map[CommandType]map[string]bool // +config|get, etc.

	// Key permissions
	AllKeys     bool         // ~*
	KeyPatterns []KeyPattern // ~pattern, %R~pattern, %W~pattern

	// Channel permissions (for pub/sub)
	AllChannels     bool     // &*
	ChannelPatterns []string // &pattern

	// Selectors (additional permission sets, Redis 7.0+)
	Selectors []*ACLSelector
}

ACLUser represents a Redis ACL user

func (*ACLUser) GetUserForDisplay 

func (u *ACLUser) GetUserForDisplay() map[string]any

GetUserForDisplay returns user info suitable for ACL GETUSER

type BytesReconstructor 

type BytesReconstructor func(snap *pb.ReducedEffect) []byte

BytesReconstructor converts a non-SCALAR snapshot into raw bytes. Registered per ValueType by modules (bitmap, hll) at init time.

type Command 

type Command struct {
	Type        CommandType
	Args        [][]byte // All arguments after the command name
	RawName     []byte   // Original command name (preserved for unknown commands)
	Transaction any
	Context     *effects.Context
	Runtime     *effects.Engine
	// contains filtered or unexported fields
}

Command represents a parsed Redis command

func GetCommand 

func GetCommand() *Command

GetCommand gets a Command from the pool

func (*Command) Reset 

func (c *Command) Reset()

Reset clears the command for reuse, preserving slice capacity

func (*Command) SetSingleKey added in v1.3.0 

func (c *Command) SetSingleKey(arg []byte) string

SetSingleKey records arg as this command's one-key scratch value without copying the pooled parser bytes. The returned string and SingleKeySlice are valid only until the command is reset or returned to the pool. Code that retains a key beyond command execution must clone it first.

func (*Command) SingleKeySlice added in v1.3.0 

func (c *Command) SingleKeySlice() []string

SingleKeySlice returns allocation-free one-key scratch storage.

func (*Command) SingleKeyValue added in v1.3.0 

func (c *Command) SingleKeyValue() string

SingleKeyValue returns the value installed by SetSingleKey.

type CommandCategory 

type CommandCategory int

CommandCategory represents a category of Redis commands 

const (
	CatRead CommandCategory = 1 << iota
	CatWrite
	CatSet
	CatSortedSet
	CatList
	CatHash
	CatString
	CatBitmap
	CatHyperLogLog
	CatGeo
	CatStream
	CatPubSub
	CatAdmin
	CatFast
	CatSlow
	CatBlocking
	CatDangerous
	CatConnection
	CatTransaction
	CatScripting
	CatKeyspace
	CatAll // Special: represents all commands
)

func GetCategoryByName 

func GetCategoryByName(name string) (CommandCategory, bool)

GetCategoryByName returns a category by name

func GetCommandCategories 

func GetCommandCategories(cmd CommandType) []CommandCategory

GetCommandCategories returns the categories for a command

type CommandEntry 

type CommandEntry struct {
	// Handler is the standard handler: validate → return keys and a runner closure.
	// Used for both live execution and transaction queueing (when TxnPrepare is nil).
	Handler HandlerFunc

	// ConnHandler is for commands needing connection state (blocking, auth, pubsub, scripting).
	// When set, live dispatch calls this instead of Handler.
	ConnHandler ConnHandlerFunc

	// TxnPrepare overrides Handler for transaction queueing.
	// Used for blocking→non-blocking fallbacks and conn-needing commands in transactions.
	// If nil, Handler is used for transaction queueing.
	// Commands with neither Handler nor TxnPrepare cannot be queued (aborts transaction).
	TxnPrepare TxnPrepareFunc

	// Keys extracts data keys from a command for per-key ACL enforcement.
	// Required for ConnHandler entries that touch data keys (blocking commands, SORT, scripting).
	// Nil means the command does not touch data keys (auth, pubsub channels, server commands).
	Keys KeyExtractFunc

	Flags CommandFlags
}

CommandEntry holds everything needed to dispatch a command.

type CommandExecutor 

type CommandExecutor func(cmd *Command, w *Writer, conn *Connection)

CommandExecutor dispatches a command for execution (used by scripting for redis.call/pcall).

type CommandFlags 

type CommandFlags uint16

CommandFlags describes command behavior for dispatch. 

const (
	FlagWrite    CommandFlags = 1 << iota // Mutates data
	FlagNoAuth                            // Allowed without authentication (AUTH, QUIT, HELLO)
	FlagNoQueue                           // Not queueable in MULTI (MULTI, EXEC, DISCARD, WATCH, UNWATCH, QUIT)
	FlagPubSub                            // Allowed in RESP2 pub/sub mode
	FlagTrackGet                          // Track GET latency
	FlagTrackSet                          // Track SET latency
)

type CommandHandler 

type CommandHandler interface {
	// ExecuteInto processes a command and writes the response to the writer
	ExecuteInto(cmd *Command, w *Writer, conn *Connection)

	// Close releases resources used by the handler
	Close()

	// SetStats sets the shared stats tracker
	SetStats(stats *Stats)

	// GetAdaptiveStats returns per-shard adaptive cache statistics
	GetAdaptiveStats() []cache.AdaptiveStats

	// GetCacheBytes returns current bytes used by cached items
	GetCacheBytes() int64

	// GetItemCount returns the number of keys resident in the index (the set
	// eviction operates on), distinct from GetVertexCount
	GetItemCount() int

	// GetReleaseQueueDepth returns the number of cold-evicted keys whose
	// deferred ref-release walk has not yet run; a persistently deep queue
	// means reclaim is falling behind eviction
	GetReleaseQueueDepth() int

	// GetArenaBytes returns the critbit index's slot-array footprint (trie
	// skeleton), distinct from GetCacheBytes (vertex pool effect bytes)
	GetArenaBytes() int64

	// GetVertexCount returns the number of effects resident in the vertex pool
	GetVertexCount() int

	// GetCacheEvictions returns keys evicted under memory pressure (evicted_keys)
	GetCacheEvictions() uint64

	// GetVerticesReclaimed returns vertices freed by reclaim (storage churn)
	GetVerticesReclaimed() uint64

	// RequiresAuth returns true if authentication is required
	RequiresAuth() bool

	// DebugEnabled returns true if debug logging is enabled
	DebugEnabled() bool
}

CommandHandler defines the interface for Redis command handlers. Implementations can use different storage backends:

  • Handler: in-memory CloxCache (fast, volatile)
  • CRDTHandler: persistent CRDT data plane (durable, replicated) [future]

type CommandRunner 

type CommandRunner func()

CommandRunner is a closure that executes a pre-validated command.

type CommandStats 

type CommandStats struct {
	Calls         atomic.Uint64
	Usec          atomic.Uint64
	RejectedCalls atomic.Uint64
	FailedCalls   atomic.Uint64
}

CommandStats tracks statistics for a single command

type CommandStatsSnapshot 

type CommandStatsSnapshot struct {
	Calls         uint64
	Usec          uint64
	RejectedCalls uint64
	FailedCalls   uint64
}

CommandStatsSnapshot is a non-atomic snapshot of command stats

type CommandType 

type CommandType int

CommandType represents the type of Redis command 

const (
	CmdUnknown CommandType = iota
	CmdNoop

	// Connection commands
	CmdPing
	CmdEcho
	CmdAuth
	CmdSelect
	CmdQuit
	CmdHello

	// String commands
	CmdGet
	CmdSet
	CmdDel
	CmdDelEx
	CmdExists
	CmdExpire
	CmdExpireAt
	CmdPExpire
	CmdPExpireAt
	CmdTTL
	CmdPTTL
	CmdExpireTime
	CmdPExpireTime
	CmdPersist
	CmdRename
	CmdRenameNX
	CmdIncr
	CmdDecr
	CmdIncrBy
	CmdDecrBy
	CmdIncrByFloat
	CmdAppend
	CmdGetSet
	CmdGetDel
	CmdGetEx
	CmdMGet
	CmdMSet
	CmdMSetNX
	CmdMSetEX
	CmdSetNX
	CmdSetEX
	CmdPSetEX
	CmdStrLen
	CmdSetRange
	CmdGetRange
	CmdSubstr // Alias for GETRANGE (deprecated)
	CmdLcs    // Longest Common Subsequence
	CmdDigest // XXH3 hash digest of string value
	CmdType

	// Bitmap commands
	CmdSetBit
	CmdGetBit
	CmdBitCount
	CmdBitPos
	CmdBitOp
	CmdBitField
	CmdBitFieldRO

	// List commands
	CmdLPush
	CmdLPushX
	CmdRPush
	CmdRPushX
	CmdLPop
	CmdRPop
	CmdBLPop
	CmdBRPop
	CmdLMPop
	CmdBLMPop
	CmdLMove
	CmdBLMove
	CmdRPopLPush
	CmdBRPopLPush
	CmdLLen
	CmdLRange
	CmdLIndex
	CmdLSet
	CmdLRem
	CmdLTrim
	CmdLInsert
	CmdLPos

	// Hash commands
	CmdHGet
	CmdHSet
	CmdHSetNX
	CmdHDel
	CmdHExists
	CmdHGetAll
	CmdHKeys
	CmdHVals
	CmdHLen
	CmdHMGet
	CmdHMSet
	CmdHIncrBy
	CmdHIncrByFloat
	CmdHStrLen
	CmdHRandField
	CmdHScan
	CmdHGetDel
	CmdHExpire
	CmdHPExpire
	CmdHExpireAt
	CmdHPExpireAt
	CmdHTTL
	CmdHPTTL
	CmdHExpireTime
	CmdHPExpireTime
	CmdHPersist
	CmdHSetEx
	CmdHGetEx

	// Set commands
	CmdSAdd
	CmdSCard
	CmdSIsMember
	CmdSMIsMember
	CmdSMembers
	CmdSPop
	CmdSRem
	CmdSRandMember
	CmdSInter
	CmdSInterStore
	CmdSInterCard
	CmdSUnion
	CmdSUnionStore
	CmdSDiff
	CmdSDiffStore
	CmdSMove

	// Sorted set commands
	CmdZAdd
	CmdZRem
	CmdZScore
	CmdZCard
	CmdZRank
	CmdZRevRank
	CmdZRange
	CmdZRevRange
	CmdZRangeByScore
	CmdZRevRangeByScore
	CmdZCount
	CmdZIncrBy
	CmdZPopMin
	CmdZPopMax
	CmdZMPop
	CmdZMScore
	CmdBZPopMin
	CmdBZPopMax
	CmdBZMPop
	CmdZRangeByLex
	CmdZRevRangeByLex
	CmdZLexCount
	CmdZRemRangeByScore
	CmdZRemRangeByRank
	CmdZRemRangeByLex
	CmdZUnionStore
	CmdZInterStore
	CmdZUnion
	CmdZInter
	CmdZInterCard
	CmdZDiff
	CmdZDiffStore
	CmdZRangeStore
	CmdZRandMember

	// Generic commands
	CmdCopy
	CmdMove
	CmdRandomKey
	CmdSort
	CmdSortRO

	// Server commands
	CmdInfo
	CmdDBSize
	CmdFlushDB
	CmdFlushAll
	CmdTime
	CmdKeys
	CmdScan
	CmdCommand
	CmdConfig
	CmdClient
	CmdDebug
	CmdMemory
	CmdShutdown

	// Transaction commands
	CmdMulti
	CmdExec
	CmdDiscard
	CmdWatch
	CmdUnwatch

	// Scripting commands
	CmdFunction
	CmdScript
	CmdEval
	CmdEvalSha
	CmdEvalRO
	CmdEvalShaRO
	CmdFcall
	CmdFcallRO

	// Replication commands (stubbed - replication not supported)
	CmdSync
	CmdPSync
	CmdReplConf
	CmdWait
	CmdWaitAof

	// Pub/Sub commands
	CmdSubscribe
	CmdUnsubscribe
	CmdPSubscribe
	CmdPUnsubscribe
	CmdPublish
	CmdPubSub

	// Database commands
	CmdSwapDB

	// Stream commands
	CmdXAdd
	CmdXLen
	CmdXRange
	CmdXRevRange
	CmdXRead
	CmdXGroup
	CmdXReadGroup
	CmdXAck
	CmdXPending
	CmdXTrim
	CmdXDel
	CmdXInfo
	CmdXSetId
	CmdXDelEx
	CmdXAckDel
	CmdXClaim
	CmdXAutoClaim
	CmdXIdmpRecord
	CmdXCfgSet

	// HyperLogLog commands
	CmdPFAdd
	CmdPFCount
	CmdPFMerge
	CmdPFSelfTest
	CmdPFDebug

	// Geo commands
	CmdGeoAdd
	CmdGeoDist
	CmdGeoHash
	CmdGeoPos
	CmdGeoRadius
	CmdGeoRadiusRO
	CmdGeoRadiusByMember
	CmdGeoRadiusByMemberRO
	CmdGeoSearch
	CmdGeoSearchStore

	// ACL commands
	CmdAcl

	// RedisJSON (JSON.*) commands
	CmdJSONSet
	CmdJSONGet
	CmdJSONMGet
	CmdJSONMSet
	CmdJSONMerge
	CmdJSONDel
	CmdJSONForget
	CmdJSONClear
	CmdJSONType
	CmdJSONNumIncrBy
	CmdJSONNumMultBy
	CmdJSONNumPowBy
	CmdJSONArrLen
	CmdJSONArrAppend
	CmdJSONArrInsert
	CmdJSONArrPop
	CmdJSONArrTrim
	CmdJSONArrIndex
	CmdJSONStrLen
	CmdJSONStrAppend
	CmdJSONObjKeys
	CmdJSONObjLen
	CmdJSONToggle
	CmdJSONResp
	CmdJSONDebug

	// CmdMax is a sentinel marking the end of the CommandType enum.
	// Must remain last. Used to size the registry array.
	CmdMax
)

func GetCommandsInCategory 

func GetCommandsInCategory(cat CommandCategory) []CommandType

GetCommandsInCategory returns all commands in a category

func LookupCommandName 

func LookupCommandName(name string) (CommandType, bool)

LookupCommandName returns the CommandType for a given uppercase command name. Returns CmdUnknown and false if the name is not recognized.

func ParseCommandType 

func ParseCommandType(name []byte) CommandType

ParseCommandType converts a command name byte slice to CommandType

func (CommandType) String 

func (c CommandType) String() string

String returns the command name

func (CommandType) StringPtr added in v1.2.0 

func (c CommandType) StringPtr() *string

StringPtr returns a stable pointer to the command's display name.

type ConnHandlerFunc 

type ConnHandlerFunc func(cmd *Command, w *Writer, db *Database, conn *Connection)

ConnHandlerFunc is for commands that need connection state (blocking, auth, transactions, pubsub, scripting).

type Connection 

type Connection struct {
	User          *ACLUser        // nil = unauthenticated
	Username      string          // Cached username for logging
	Protocol      ProtocolVersion // RESP protocol version (default RESP2)
	ClientName    string          // Client name set via CLIENT SETNAME or HELLO
	ClientLibName string          // Client library name (from HELLO)
	ClientLibVer  string          // Client library version (from HELLO)
	RemoteAddr    string          // Client remote address for logging

	// Transaction state
	InTransaction bool            // True if inside MULTI block
	TxnAborted    bool            // True if watched key was modified
	QueuedCmds    []QueuedCommand // Commands queued for EXEC
	WatchedKeys   []WatchedKey    // Keys being watched with their versions

	// Script state
	InScript bool // True if executing commands from a Lua script (blocking commands become non-blocking)

	// SI Transaction state (for CRDT handler with Snapshot Isolation)
	SITransaction   any      // Active SI transaction (nil if not in SI mode)
	BufferedResults [][]byte // Results captured during non-SI MULTI for EXEC

	// Blocking command support
	Ctx   context.Context // Context for blocking operations (cancellation on disconnect)
	Stats *Stats          // Server stats (set at connection creation)

	// Pub/Sub state
	PubSubClient *PubSubClient // nil if not in pubsub mode

	// Effects engine context for migrated modules
	EffectsCtx *effects.Context

	// ReadOnlyCtx serves read-only, non-transactional commands. It answers
	// read-misses from the cluster key filters without subscribing (free
	// misses). Lazily created alongside EffectsCtx.
	ReadOnlyCtx *effects.Context
	// contains filtered or unexported fields
}

Connection represents the state of a client connection

func (*Connection) Block 

func (c *Connection) Block(timeout float64) (context.Context, func())

Block sets up a blocking context with an optional timeout. Returns (ctx, cleanup) — caller MUST defer cleanup() immediately.

func (*Connection) ClearWatches 

func (c *Connection) ClearWatches()

ClearWatches clears watched keys without affecting transaction state

func (*Connection) InPubSubMode 

func (c *Connection) InPubSubMode() bool

InPubSubMode returns true if the connection is in pub/sub mode

func (*Connection) IsBlocked 

func (c *Connection) IsBlocked() bool

IsBlocked returns true if the connection is currently blocked.

func (*Connection) IsRESP3 

func (c *Connection) IsRESP3() bool

IsRESP3 returns true if the connection is using RESP3

func (*Connection) ResetTransaction 

func (c *Connection) ResetTransaction()

ResetTransaction clears transaction state

func (*Connection) Unblock 

func (c *Connection) Unblock(withError bool) bool

Unblock cancels a blocked operation. If withError is true, the client receives an error message instead of nil.

func (*Connection) WasUnblockedWithError 

func (c *Connection) WasUnblockedWithError() bool

WasUnblockedWithError returns true if CLIENT UNBLOCK ERROR was used.

type Consumer 

type Consumer struct {
	Name         string
	SeenTime     int64 // last interaction time (Unix ms)
	PendingCount int64 // entries owned by this consumer
	ActiveTime   int64 // last message fetched time
}

Consumer represents a consumer within a group

type ConsumerGroup 

type ConsumerGroup struct {
	Name            string
	LastDeliveredID StreamID // for ">" - next delivery starts after this ID
	EntriesRead     int64    // for lag calculation (XINFO)

	// Pending Entries List (PEL)
	// Memory bounded by gap size, not total volume
	Pending map[StreamID]*PendingEntry

	// Consumers
	Consumers map[string]*Consumer

	// Per-consumer pending limit (0 = unlimited)
	MaxPending int64

	CreatedAt int64 // Unix milliseconds
}

ConsumerGroup represents a consumer group with TCP-style ack tracking

func NewConsumerGroup 

func NewConsumerGroup(name string, lastDeliveredID StreamID) *ConsumerGroup

NewConsumerGroup creates a new consumer group

func (*ConsumerGroup) GetOrCreateConsumer 

func (g *ConsumerGroup) GetOrCreateConsumer(name string) *Consumer

GetOrCreateConsumer gets or creates a consumer in the group

type Database 

type Database struct {
	// contains filtered or unexported fields
}

Database is the handle threaded through command handlers. Its only remaining job is reaching the blocking-command subscription registry; actual state lives in the effects engine.

func (*Database) Manager 

func (db *Database) Manager() *DatabaseManager

Manager returns the owning DatabaseManager.

type DatabaseManager 

type DatabaseManager struct {

	// Subscriptions manages blocking command wake signals. Waiters register
	// locally; wake signals arrive through the engine's OnKeyDataAdded /
	// OnKeyDeleted / OnFlushAll callbacks, which fire for both local flushes
	// and remote effect arrivals.
	Subscriptions *SubscriptionManager[struct{}]
	// contains filtered or unexported fields
}

DatabaseManager owns the Database and the blocking-command wake registry.

func NewDatabaseManager added in v1.2.0 

func NewDatabaseManager() *DatabaseManager

NewDatabaseManager creates a new database manager.

func (*DatabaseManager) DB added in v1.2.0 

func (dm *DatabaseManager) DB() *Database

DB returns the database.

type HLLValue 

type HLLValue struct {
	Registers [16384]uint8 // Each register stores a 6-bit value (0-63)
}

HLLValue stores HyperLogLog data for probabilistic cardinality estimation Uses 16384 6-bit registers (2^14 registers) for ~0.81% standard error

func (*HLLValue) Add 

func (h *HLLValue) Add(element []byte) bool

Add adds an element to the HyperLogLog Returns true if the internal registers were modified (cardinality might have increased)

func (*HLLValue) Count 

func (h *HLLValue) Count() int64

Count estimates the cardinality of the set

func (*HLLValue) Merge 

func (h *HLLValue) Merge(others ...*HLLValue)

Merge merges other HyperLogLogs into this one Takes the maximum value for each register position

type HandlerFunc 

type HandlerFunc func(cmd *Command, w *Writer, db *Database) (valid bool, keys []string, runner CommandRunner)

HandlerFunc is the standard handler signature: validate a command, return keys and a runner.

func WrapHandler 

func WrapHandler(fn func()) HandlerFunc

WrapHandler wraps a direct-execute function as a HandlerFunc (no keys, always valid).

type HashValue 

type HashValue struct {
	Fields    map[string][]byte
	FieldsTTL map[string]int64 // field -> Unix millisecond timestamp, 0 = no expiration
}

HashValue stores hash field/value pairs

type KeyExtractFunc 

type KeyExtractFunc func(cmd *Command) []string

KeyExtractFunc extracts keys from a command for ACL checking. Used by ConnHandler entries that touch data keys but bypass the standard Handler path.

type KeyPattern 

type KeyPattern struct {
	Pattern string
	Type    KeyPatternType
}

KeyPattern represents a key access pattern

type KeyPatternType 

type KeyPatternType int

KeyPatternType indicates what operations are permitted by a key pattern 

const (
	KeyPatternReadWrite KeyPatternType = iota // ~pattern (read + write)
	KeyPatternReadOnly                        // %R~pattern (read only)
	KeyPatternWriteOnly                       // %W~pattern (write only)
)

type ListValue 

type ListValue struct {

	// Sequence tracking for persistent storage (CRDT handler)
	// HeadSeq is the sequence number of the first element (decrements on LPUSH)
	// TailSeq is the sequence number of the last element (increments on RPUSH)
	// For an empty list: HeadSeq=0, TailSeq=-1
	HeadSeq int64
	TailSeq int64
	// contains filtered or unexported fields
}

ListValue stores list data using a doubly-linked list for O(1) push/pop

type ModuleEntry 

type ModuleEntry struct {
	Cmd   CommandType
	Entry *CommandEntry
}

ModuleEntry pairs a CommandType with its CommandEntry for module registration.

func GetModuleRegistrations 

func GetModuleRegistrations() []ModuleEntry

GetModuleRegistrations returns all module-registered command entries.

type Parser 

type Parser struct {
	// contains filtered or unexported fields
}

Parser handles RESP2 protocol parsing

func NewParser 

func NewParser(r io.Reader) *Parser

NewParser creates a new RESP2 protocol parser

func NewParserWithReader 

func NewParserWithReader(r *bufio.Reader) *Parser

NewParserWithReader creates a parser using an existing buffered reader

func (*Parser) Buffered 

func (p *Parser) Buffered() int

Buffered returns the number of bytes currently buffered

func (*Parser) ReadCommand 

func (p *Parser) ReadCommand() (*Command, error)

ReadCommand reads and parses the next Redis command Returns a pooled Command - caller should call putCommand when done

func (*Parser) ReadCommandInto 

func (p *Parser) ReadCommandInto(cmd *Command) (*Command, error)

ReadCommandInto reads a command into the provided Command struct

func (*Parser) Reset 

func (p *Parser) Reset(r io.Reader)

Reset resets the parser with a new reader

type PendingEntry 

type PendingEntry struct {
	ID            StreamID
	Consumer      string // owning consumer name
	DeliveryTime  int64  // Unix ms of last delivery
	DeliveryCount int64  // for retry logic / dead-letter detection
}

PendingEntry tracks delivery information for Redis compatibility

type ProtocolVersion 

type ProtocolVersion int

ProtocolVersion represents the RESP protocol version 

const (
	RESP2 ProtocolVersion = 2
	RESP3 ProtocolVersion = 3
)

type PubSubBroker 

type PubSubBroker interface {
	Subscribe(client *PubSubClient, channels ...string) []int
	Unsubscribe(client *PubSubClient, channels ...string) ([]string, []int)
	PSubscribe(client *PubSubClient, patterns ...string) []int
	PUnsubscribe(client *PubSubClient, patterns ...string) ([]string, []int)
	Publish(channel string, message []byte) int
	Cleanup(client *PubSubClient)
	SubscriptionCount(client *PubSubClient) int
	Channels(pattern string) []string
	NumSub(channels ...string) map[string]int
	NumPat() int
}

PubSubBroker is the interface for the pub/sub subsystem. This allows the pubsub module to register commands via init() without circular imports.

func GetPubSubBroker 

func GetPubSubBroker() PubSubBroker

GetPubSubBroker returns the current pub/sub broker instance (may be nil).

type PubSubClient 

type PubSubClient struct {
	// contains filtered or unexported fields
}

PubSubClient represents a client in pub/sub mode

func NewPubSubClient 

func NewPubSubClient(conn *Connection, protocol ProtocolVersion) *PubSubClient

NewPubSubClient creates a new pub/sub client

func (*PubSubClient) Close 

func (c *PubSubClient) Close()

Close closes the client's message channel

func (*PubSubClient) Done 

func (c *PubSubClient) Done() <-chan struct{}

Done returns a channel that is closed when the client is closed

func (*PubSubClient) IsBusy 

func (c *PubSubClient) IsBusy() bool

IsBusy returns true if the client is currently executing a command

func (*PubSubClient) MsgChan 

func (c *PubSubClient) MsgChan() <-chan *PubSubMessage

MsgChan returns the message channel for reading

func (*PubSubClient) Send 

func (c *PubSubClient) Send(msg *PubSubMessage) bool

Send sends a message to the client's message channel (non-blocking)

func (*PubSubClient) SetBusy 

func (c *PubSubClient) SetBusy(busy bool)

SetBusy marks the client as busy (during command execution)

type PubSubClusterRouter added in v0.3.0 

type PubSubClusterRouter interface {
	// AnnounceSub broadcasts an ephemeral SubscriptionEffect to every
	// peer so they record this node's interest in the channel / pattern.
	// Called on the 0→1 transition for a local subscription.
	AnnounceSub(channel string, isPattern bool)

	// AnnounceUnsub is the unsubscribe counterpart of AnnounceSub.
	// Called on the last→0 transition for a local subscription.
	AnnounceUnsub(channel string, isPattern bool)

	// RouteMessage delivers a PUBLISH to remote peers whose recorded
	// subscriptions match the channel (literal channel match or glob
	// match against a recorded pattern). Returns the number of peers
	// the message was sent to (one count per peer regardless of how
	// many of their local clients are subscribed).
	RouteMessage(channel string, payload []byte) int

	// ClusterChannels returns the union of channels remote peers have
	// announced interest in, optionally filtered by pattern.
	ClusterChannels(pattern string) []string

	// ClusterNumSub returns, per channel, the count of remote peers
	// announcing interest in that exact channel name. Patterns are
	// not counted (matching Redis NUMSUB semantics).
	ClusterNumSub(channels []string) map[string]int

	// ClusterPatterns returns every distinct pattern announced by a
	// remote peer. Returned to the caller (not as a count) so the
	// local broker can union with its own pattern set and dedupe.
	ClusterPatterns() []string
}

PubSubClusterRouter is the cluster-side surface the local pub/sub broker uses to fan out subscriptions and messages to peer nodes. A nil router (standalone mode) means every call is a no-op.

func GetPubSubClusterRouter added in v0.3.0 

func GetPubSubClusterRouter() PubSubClusterRouter

GetPubSubClusterRouter returns the installed router, or nil in standalone mode.

type PubSubMessage 

type PubSubMessage struct {
	Type    string // "message", "pmessage", "subscribe", "unsubscribe", "psubscribe", "punsubscribe"
	Pattern string // for pmessage only
	Channel string
	Payload []byte
	Count   int // subscription count for sub/unsub responses
}

PubSubMessage represents a message to be delivered to subscribers

type QueuedCommand 

type QueuedCommand struct {
	Type   CommandType
	Args   [][]byte
	Keys   []string      // Keys extracted during validation (for ACL checking at EXEC)
	Runner CommandRunner // Pre-validated runner to execute at EXEC time
}

QueuedCommand stores a command queued during a transaction

type RedisValue 

type RedisValue struct {
	Type              ValueType
	Data              []byte       // For strings
	List              *ListValue   // For lists
	Hash              *HashValue   // For hashes
	Set               *SetValue    // For sets
	ZSet              *ZSetValue   // For sorted sets
	Stream            *StreamValue // For streams
	HLL               *HLLValue    // For HyperLogLog
	CounterDeltaInt   int64
	CounterDeltaFloat float64
	Exptime           int64 // Unix millisecond timestamp, 0 = no expiration
	CreatedAt         int64 // Unix nanosecond timestamp when created
}

RedisValue represents a Redis value with expiration support

type RefMode 

type RefMode int

RefMode specifies how to handle consumer group references during trimming 

const (
	RefModeKeepRef RefMode = iota // Default: keep PEL references for trimmed entries
	RefModeDelRef                 // Delete PEL references when trimming
	RefModeAcked                  // Only trim entries acknowledged by all groups
)

type Registration 

type Registration[T any] struct {
	// contains filtered or unexported fields
}

Registration represents a pending subscription

func (*Registration[T]) Cancel 

func (r *Registration[T]) Cancel()

Cancel marks the waiter as dead and removes it from all topics. Safe to call multiple times. Always call this (use defer).

func (*Registration[T]) CancelAndRenotify 

func (r *Registration[T]) CancelAndRenotify()

CancelAndRenotify cancels this registration and wakes the next waiter on each channel. Use this when the waiter could not consume the data (e.g. WRONGTYPE on destination) so the next blocked client gets a chance.

func (*Registration[T]) IsOldestWaiter 

func (r *Registration[T]) IsOldestWaiter(channel []byte) bool

IsOldestWaiter checks if this waiter is first in line for the given channel.

func (*Registration[T]) NotifyChan 

func (r *Registration[T]) NotifyChan() <-chan struct{}

NotifyChan returns the channel that receives wake signals. Use this when you need to select on multiple channels along with the wake signal.

func (*Registration[T]) Wait 

func (r *Registration[T]) Wait() ([]byte, error)

Wait blocks until a wake signal is received or ctx is canceled. Returns the channel that was signaled (nil if canceled) and error. Can be called multiple times - waiter remains subscribed until Cancel().

type Registry 

type Registry [CmdMax]*CommandEntry

Registry is the command dispatch table, indexed by CommandType for O(1) lookup. Stored as a field on Handler since entries reference handler methods.

func (*Registry) Lookup 

func (r *Registry) Lookup(cmd CommandType) *CommandEntry

Lookup returns the entry for a command type, or nil if unregistered.

func (*Registry) Register 

func (r *Registry) Register(cmd CommandType, entry *CommandEntry)

Register adds a command entry to the Registry.

type ScriptingEngine 

type ScriptingEngine interface {
	HandleEval(cmd *Command, w *Writer, conn *Connection, db *Database, readOnly bool)
	HandleScript(cmd *Command, w *Writer)
	HandleFunction(cmd *Command, w *Writer)
	HandleFcall(cmd *Command, w *Writer, conn *Connection, db *Database, readOnly bool)
	IsScriptTimedOut() bool
	Close()
}

ScriptingEngine is the interface for the scripting subsystem. This allows the scripting module to register commands via init() without circular imports.

func GetScriptingEngine 

func GetScriptingEngine() ScriptingEngine

GetScriptingEngine returns the current scripting engine instance (may be nil).

type SetValue 

type SetValue struct {
	Members map[string]struct{}
}

SetValue stores set members

type Snapshot 

type Snapshot struct {
	Time             time.Time
	Uptime           time.Duration
	CurrConnections  int64
	TotalConnections uint64
	CmdGet           uint64
	CmdSet           uint64
	CmdDel           uint64
	CmdGetHits       uint64
	CmdGetMisses     uint64
	HitRate          float64
	GetLatencyP50    time.Duration
	GetLatencyP99    time.Duration
	SetLatencyP50    time.Duration
	SetLatencyP99    time.Duration
	CmdLatencyP50    time.Duration
	CmdLatencyP99    time.Duration
	BytesRead        uint64
	BytesWritten     uint64
}

Snapshot represents a point-in-time snapshot of statistics

type Stats 

type Stats struct {

	// Connection stats
	CurrConnections  atomic.Int64
	TotalConnections atomic.Uint64
	BlockedClients   atomic.Int64

	// Command counts
	CmdGet atomic.Uint64
	CmdSet atomic.Uint64
	CmdDel atomic.Uint64

	// Hit/miss tracking
	CmdGetHits   atomic.Uint64
	CmdGetMisses atomic.Uint64

	// Bytes read/written
	BytesRead    atomic.Uint64
	BytesWritten atomic.Uint64

	// Total error replies
	TotalErrorReplies atomic.Uint64
	// contains filtered or unexported fields
}

Stats tracks Redis server statistics

func GetServerStats 

func GetServerStats() *Stats

GetServerStats returns the current server stats instance (may be nil).

func NewStats 

func NewStats() *Stats

NewStats creates a new Stats instance

func (*Stats) ClientBlocked 

func (s *Stats) ClientBlocked()

ClientBlocked records a client entering a blocking state

func (*Stats) ClientUnblocked 

func (s *Stats) ClientUnblocked()

ClientUnblocked records a client leaving a blocking state

func (*Stats) CmdLatencyP50 

func (s *Stats) CmdLatencyP50() time.Duration

CmdLatencyP50 returns the 50th percentile command latency (all commands)

func (*Stats) CmdLatencyP99 

func (s *Stats) CmdLatencyP99() time.Duration

CmdLatencyP99 returns the 99th percentile command latency (all commands)

func (*Stats) ConnectionClosed 

func (s *Stats) ConnectionClosed()

ConnectionClosed records a closed connection

func (*Stats) ConnectionOpened 

func (s *Stats) ConnectionOpened()

ConnectionOpened records a new connection

func (*Stats) GetCommandStats 

func (s *Stats) GetCommandStats() map[string]CommandStatsSnapshot

GetCommandStats returns a copy of command statistics, keyed by command name, for commands that have been called at least once.

func (*Stats) GetErrorStats 

func (s *Stats) GetErrorStats() map[string]uint64

GetErrorStats returns a copy of error statistics

func (*Stats) GetLatencyP50 

func (s *Stats) GetLatencyP50() time.Duration

GetLatencyP50 returns the 50th percentile GET latency

func (*Stats) GetLatencyP99 

func (s *Stats) GetLatencyP99() time.Duration

GetLatencyP99 returns the 99th percentile GET latency

func (*Stats) HitRate 

func (s *Stats) HitRate() float64

HitRate returns the cache hit rate (0.0 to 1.0)

func (*Stats) RecordCommand 

func (s *Stats) RecordCommand(cmd CommandType, usec uint64, failed bool)

RecordCommand records a command execution

func (*Stats) RecordError 

func (s *Stats) RecordError(prefix string)

RecordError records an error by its prefix (e.g., "NOGROUP", "ERR", "WRONGTYPE")

func (*Stats) RecordGetLatency 

func (s *Stats) RecordGetLatency(d time.Duration)

RecordGetLatency records a GET command latency

func (*Stats) RecordSetLatency 

func (s *Stats) RecordSetLatency(d time.Duration)

RecordSetLatency records a SET command latency

func (*Stats) Reset 

func (s *Stats) Reset()

Reset resets all statistics

func (*Stats) SetLatencyP50 

func (s *Stats) SetLatencyP50() time.Duration

SetLatencyP50 returns the 50th percentile SET latency

func (*Stats) SetLatencyP99 

func (s *Stats) SetLatencyP99() time.Duration

SetLatencyP99 returns the 99th percentile SET latency

func (*Stats) Snapshot 

func (s *Stats) Snapshot() Snapshot

Snapshot takes a point-in-time snapshot of statistics

func (*Stats) Uptime 

func (s *Stats) Uptime() time.Duration

Uptime returns the server uptime

type StreamEntry 

type StreamEntry struct {
	ID     StreamID
	Fields [][]byte // alternating [key1, val1, key2, val2, ...]
}

StreamEntry represents a single entry in a stream

type StreamID 

type StreamID struct {
	Ms  uint64 // Unix timestamp in milliseconds
	Seq uint64 // Sequence number within the millisecond
}

StreamID represents a Redis stream entry ID (milliseconds-sequence)

func ParseStreamID 

func ParseStreamID(s string) (id StreamID, exclusive bool, isSpecial bool, err error)

ParseStreamID parses a stream ID from a string. Supported formats:

  • "*" - auto-generate (only for XADD)
  • "$" - last entry ID (for XREAD)
  • ">" - only new entries (for XREADGROUP)
  • "-" - minimum ID (for XRANGE)
  • "+" - maximum ID (for XRANGE)
  • "0" or "0-0" - minimum ID
  • "1234567890123-0" - full ID
  • "1234567890123" - partial ID (seq defaults to 0 for start, max for end)
  • "(1234567890123-0" - exclusive (for XRANGE)

func ParseStreamIDForRange 

func ParseStreamIDForRange(s string, isEnd bool) (id StreamID, exclusive bool, err error)

ParseStreamIDForRange parses an ID for XRANGE/XREVRANGE, handling partial IDs appropriately. For start IDs, partial IDs get seq=0. For end IDs, partial IDs get seq=MaxUint64.

func (StreamID) Compare 

func (id StreamID) Compare(other StreamID) int

Compare compares two stream IDs. Returns -1 if id < other, 0 if id == other, 1 if id > other.

func (StreamID) IsZero 

func (id StreamID) IsZero() bool

IsZero returns true if the stream ID is 0-0

func (StreamID) Key 

func (id StreamID) Key() string

Key returns a 16-byte big-endian encoding suitable for lexicographic ordering. Big-endian ensures that numeric order matches byte order.

func (StreamID) String 

func (id StreamID) String() string

String returns the string representation of the stream ID

type StreamValue 

type StreamValue struct {
	// Entry storage - trie for ordered iteration, map for O(1) lookup
	Index   keytrie.KeyIndex        // ordered index of ID strings (rebuilt on deserialize)
	Entries map[string]*StreamEntry // ID string -> entry data (serialized)

	// Stream metadata
	LastID       StreamID // highest ID ever generated (for auto-ID)
	FirstID      StreamID // first entry ID (updated on trim)
	EntriesAdded int64    // total entries ever added (for XINFO)
	MaxDeletedID StreamID // highest deleted ID (for XSETID)

	// Consumer groups
	Groups map[string]*ConsumerGroup
	// contains filtered or unexported fields
}

StreamValue holds all entries and consumer groups for a stream

func NewStreamValue 

func NewStreamValue() *StreamValue

NewStreamValue creates a new empty stream

func (*StreamValue) ForEach 

func (s *StreamValue) ForEach(fn func(entry *StreamEntry) bool)

ForEach iterates over all entries in order, calling fn for each. If fn returns false, iteration stops.

func (*StreamValue) Trim 

func (s *StreamValue) Trim(opts TrimOptions) int64

Trim trims the stream according to the options and returns the count of trimmed entries

type SubscriptionManager 

type SubscriptionManager[T any] struct {
	// contains filtered or unexported fields
}

SubscriptionManager manages subscriptions for blocking commands

func NewSubscriptionManager 

func NewSubscriptionManager[T any]() *SubscriptionManager[T]

NewSubscriptionManager creates a new subscription manager

func (*SubscriptionManager[T]) BlockingKeyCount 

func (m *SubscriptionManager[T]) BlockingKeyCount() int64

BlockingKeyCount returns the number of keys that have at least one active waiter.

func (*SubscriptionManager[T]) BlockingKeyCounts 

func (m *SubscriptionManager[T]) BlockingKeyCounts() (total int64, nokey int64)

BlockingKeyCounts returns the total number of blocking keys and the number of blocking keys where at least one client wants to be unblocked on key deletion (nokey).

func (*SubscriptionManager[T]) Enqueue 

func (m *SubscriptionManager[T]) Enqueue(channel []byte, _ T) bool

Enqueue is an alias for Notify (for compatibility)

func (*SubscriptionManager[T]) Notify 

func (m *SubscriptionManager[T]) Notify(channel []byte) bool

Notify wakes the oldest waiting subscriber for this channel. Waiters remain in the queue until they call Cancel(). Returns true if a waiter was signaled, false if no waiters.

func (*SubscriptionManager[T]) NotifyAll 

func (m *SubscriptionManager[T]) NotifyAll()

NotifyAll wakes ALL waiters across all topics. Used for operations like SWAPDB that may affect any blocked client.

func (*SubscriptionManager[T]) NotifyAllWaiters 

func (m *SubscriptionManager[T]) NotifyAllWaiters(channel []byte)

NotifyAllWaiters wakes ALL waiters for a specific channel. Used for operations like DEL, XGROUP DESTROY that invalidate the key/group.

func (*SubscriptionManager[T]) Register 

func (m *SubscriptionManager[T]) Register(ctx context.Context, channels ...[]byte) (*Registration[T], error)

Register creates a waiter for the given channels without blocking. Call Cancel() when done, or use defer reg.Cancel().

func (*SubscriptionManager[T]) RegisterWithNoKey 

func (m *SubscriptionManager[T]) RegisterWithNoKey(ctx context.Context, channels ...[]byte) (*Registration[T], error)

RegisterWithNoKey creates a waiter that should be unblocked when the key is deleted. This is used for XREADGROUP with ">" where the client wants to know if the stream is deleted.

func (*SubscriptionManager[T]) RegisterWithOptions 

func (m *SubscriptionManager[T]) RegisterWithOptions(ctx context.Context, unblockOnNoKey bool, channels ...[]byte) (*Registration[T], error)

RegisterWithOptions creates a waiter with configurable options. unblockOnNoKey: if true, the client wants to be unblocked when the key is deleted.

type Topic 

type Topic[T any] struct {
	// contains filtered or unexported fields
}

Topic holds the waiting subscribers for a single key

type TrimOptions 

type TrimOptions struct {
	MaxLen           int64     // Trim to this length (only used if HasMaxLen is true)
	MinID            *StreamID // Trim entries with ID < MinID (nil = no limit)
	Approx           bool      // Use approximate trimming (~)
	Limit            int64     // Max entries to trim per operation (0 = unlimited)
	NodeSize         int64     // Node size for approximate trimming (default 100)
	RefMode          RefMode   // How to handle consumer group references
	HasMaxLen        bool      // Whether MAXLEN was specified (allows MaxLen=0 to mean "trim to 0")
	HasExplicitLimit bool      // Whether LIMIT was explicitly specified (affects NodeSize rounding)
}

TrimOptions specifies trimming parameters

type TxnPrepareFunc 

type TxnPrepareFunc func(cmd *Command, w *Writer, db *Database, conn *Connection) (valid bool, keys []string, runner CommandRunner)

TxnPrepareFunc validates and prepares a command for transaction queueing. It has access to conn for commands that need connection state (scripting, pubsub, etc.).

type ValueType 

type ValueType byte

ValueType represents the type of Redis value 

const (
	TypeNone   ValueType = iota // Key doesn't exist
	TypeString                  // String value
	TypeList                    // List value
	TypeHash                    // Hash value
	TypeSet                     // Set value (future)
	TypeZSet                    // Sorted set value (future)
	TypeCounter
	TypeStream // Stream value
	TypeHLL    // HyperLogLog value
)

func (ValueType) String 

func (t ValueType) String() string

String returns the Redis type name

type Waiter 

type Waiter[T any] struct {
	// contains filtered or unexported fields
}

Waiter represents a subscriber waiting for a wake signal

type WatchedKey 

type WatchedKey struct {
	DB      int // Database index
	Key     string
	Version int64 // CreatedAt timestamp when watched (0 if key didn't exist)
}

WatchedKey stores a key being watched with its version

type Writer 

type Writer struct {
	// contains filtered or unexported fields
}

Writer handles RESP response formatting (supports RESP2 and RESP3)

func GetWriter 

func GetWriter() *Writer

GetWriter gets a Writer from the pool

func NewWriter 

func NewWriter(buf *bytes.Buffer) *Writer

NewWriter creates a new RESP response writer (defaults to RESP2)

func (*Writer) Buffer 

func (w *Writer) Buffer() *bytes.Buffer

Buffer returns the underlying buffer

func (*Writer) Protocol 

func (w *Writer) Protocol() ProtocolVersion

Protocol returns the current protocol version

func (*Writer) Reset 

func (w *Writer) Reset(buf *bytes.Buffer)

Reset resets the writer with a new buffer

func (*Writer) ResetErrorFlag 

func (w *Writer) ResetErrorFlag()

ResetErrorFlag resets the error tracking flag (call before each command)

func (*Writer) Scratch added in v1.2.0 

func (w *Writer) Scratch() *bytes.Buffer

Scratch returns the writer's reusable response-staging buffer, reset and ready for use. Retaining the buffer on the writer means the per-command stage-then-commit in ExecuteInto reuses one high-water allocation instead of growing a fresh buffer every command. If the writer is already staging into its scratch (a nested ExecuteInto on the same writer), a fresh buffer is returned so the outer stage's pending bytes aren't clobbered.

func (*Writer) SetProtocol 

func (w *Writer) SetProtocol(p ProtocolVersion)

SetProtocol sets the protocol version for subsequent writes

func (*Writer) SetStats 

func (w *Writer) SetStats(s *Stats)

SetStats sets the stats object for error tracking

func (*Writer) WriteArray 

func (w *Writer) WriteArray(count int)

WriteArray writes a RESP array header (*<count>\r\n)

func (*Writer) WriteAttribute 

func (w *Writer) WriteAttribute(count int)

WriteAttribute writes RESP3 attribute metadata (|<count>\r\n) No RESP2 equivalent - silently ignored for RESP2

func (*Writer) WriteBigNumber 

func (w *Writer) WriteBigNumber(n string)

WriteBigNumber writes a RESP3 big number ((<number>\r\n) Falls back to bulk string for RESP2

func (*Writer) WriteBlobError 

func (w *Writer) WriteBlobError(errType, msg string)

WriteBlobError writes a RESP3 blob error (!<len>\r\n<data>\r\n) Falls back to simple error for RESP2

func (*Writer) WriteBoolean 

func (w *Writer) WriteBoolean(b bool)

WriteBoolean writes a RESP3 boolean (#t\r\n or #f\r\n) Falls back to integer (1 or 0) for RESP2

func (*Writer) WriteBulkString 

func (w *Writer) WriteBulkString(b []byte)

WriteBulkString writes a RESP bulk string ($5\r\nhello\r\n)

func (*Writer) WriteBulkStringStr 

func (w *Writer) WriteBulkStringStr(s string)

WriteBulkStringStr writes a string as a RESP bulk string

func (*Writer) WriteDouble 

func (w *Writer) WriteDouble(f float64)

WriteDouble writes a RESP3 double (,<value>\r\n) Falls back to bulk string for RESP2

func (*Writer) WriteError 

func (w *Writer) WriteError(msg string)

WriteError writes a RESP error (-ERR message\r\n)

func (*Writer) WriteErrorf 

func (w *Writer) WriteErrorf(format string, args ...any)

WriteErrorf writes a formatted RESP error

func (*Writer) WriteInteger 

func (w *Writer) WriteInteger(n int64)

WriteInteger writes a RESP integer (:1000\r\n)

func (*Writer) WriteMap 

func (w *Writer) WriteMap(count int)

WriteMap writes a RESP3 map header (%<count>\r\n) Falls back to array with 2*count elements for RESP2

func (*Writer) WriteNoAuth 

func (w *Writer) WriteNoAuth()

WriteNoAuth writes the NOAUTH error

func (*Writer) WriteNotInteger 

func (w *Writer) WriteNotInteger()

WriteNotInteger writes the "not an integer" error

func (*Writer) WriteNull 

func (w *Writer) WriteNull()

WriteNull writes RESP3 null (_\r\n) or RESP2 null bulk string ($-1\r\n)

func (*Writer) WriteNullArray 

func (w *Writer) WriteNullArray()

WriteNullArray writes a RESP null array (*-1\r\n) for RESP2, or null (_\r\n) for RESP3

func (*Writer) WriteNullBulkString 

func (w *Writer) WriteNullBulkString()

WriteNullBulkString writes a RESP null bulk string ($-1\r\n) for RESP2, or null (_\r\n) for RESP3

func (*Writer) WriteOK 

func (w *Writer) WriteOK()

WriteOK writes +OK\r\n

func (*Writer) WriteOne 

func (w *Writer) WriteOne()

WriteOne writes :1\r\n

func (*Writer) WriteOutOfRange 

func (w *Writer) WriteOutOfRange()

WriteOutOfRange writes the "out of range" error

func (*Writer) WritePong 

func (w *Writer) WritePong()

WritePong writes +PONG\r\n

func (*Writer) WritePush 

func (w *Writer) WritePush(count int)

WritePush writes a RESP3 push message header (><count>\r\n) Only valid for RESP3 - no RESP2 equivalent (used for pub/sub, invalidation)

func (*Writer) WriteQueued 

func (w *Writer) WriteQueued()

WriteQueued writes +QUEUED\r\n (for transactions)

func (*Writer) WriteRaw 

func (w *Writer) WriteRaw(data []byte)

WriteRaw writes raw bytes directly to the buffer (pre-formatted RESP)

func (*Writer) WriteScore 

func (w *Writer) WriteScore(f float64)

WriteScore writes a Redis score value In RESP3: uses double type (,<value>\r\n) with proper inf/nan handling In RESP2: uses bulk string with formatted score

func (*Writer) WriteSet 

func (w *Writer) WriteSet(count int)

WriteSet writes a RESP3 set header (~<count>\r\n) Falls back to array for RESP2

func (*Writer) WriteSimpleString 

func (w *Writer) WriteSimpleString(s string)

WriteSimpleString writes a RESP simple string (+OK\r\n)

func (*Writer) WriteSyntaxError 

func (w *Writer) WriteSyntaxError()

WriteSyntaxError writes a syntax error

func (*Writer) WriteUnknownCommand 

func (w *Writer) WriteUnknownCommand(cmd string, args [][]byte)

WriteUnknownCommand writes the unknown command error

func (*Writer) WriteVerbatimString 

func (w *Writer) WriteVerbatimString(encoding string, data []byte)

WriteVerbatimString writes a RESP3 verbatim string (=<len>\r\n<enc>:<data>\r\n) Falls back to bulk string for RESP2

func (*Writer) WriteWrongNumArgs 

func (w *Writer) WriteWrongNumArgs(cmd string)

WriteWrongNumArgs writes the wrong number of arguments error

func (*Writer) WriteWrongNumArguments 

func (w *Writer) WriteWrongNumArguments(cmd string)

WriteWrongNumArguments writes the wrong number of arguments error

func (*Writer) WriteWrongType 

func (w *Writer) WriteWrongType()

WriteWrongType writes the WRONGTYPE error

func (*Writer) WriteZero 

func (w *Writer) WriteZero()

WriteZero writes :0\r\n

func (*Writer) WroteError 

func (w *Writer) WroteError() bool

WroteError returns true if WriteError was called since last ResetErrorFlag

type ZSetEntry 

type ZSetEntry struct {
	Member string
	Score  float64
}

ZSetEntry represents a member-score pair for sorted results

type ZSetValue 

type ZSetValue struct {
	Members map[string]float64 // member -> score
}

ZSetValue stores sorted set members with scores

func (*ZSetValue) Sorted 

func (z *ZSetValue) Sorted() []ZSetEntry

Sorted returns all members sorted by score (ascending), then by member name

func (*ZSetValue) SortedReverse 

func (z *ZSetValue) SortedReverse() []ZSetEntry

SortedReverse returns all members sorted by score (descending), then by member name (descending)

func (*ZSetValue) ZAdd 

func (z *ZSetValue) ZAdd(members map[string]float64) int

func (*ZSetValue) ZAddWithOptions 

func (z *ZSetValue) ZAddWithOptions(member string, score float64, nx, xx, gt, lt bool) (added, changed bool)

ZAddWithOptions adds/updates members with ZADD options Returns (added, changed) counts based on options Options: nx (only add new), xx (only update), gt (update if greater), lt (update if less), ch (count changed)

func (*ZSetValue) ZCard 

func (z *ZSetValue) ZCard() int

ZCard returns the number of members in the sorted set

func (*ZSetValue) ZCount 

func (z *ZSetValue) ZCount(min, max float64, minExclusive, maxExclusive bool) int

ZCount counts members with scores between min and max (inclusive)

func (*ZSetValue) ZIncrBy 

func (z *ZSetValue) ZIncrBy(member string, delta float64) float64

ZIncrBy increments the score of a member by delta Returns the new score

func (*ZSetValue) ZPopMax 

func (z *ZSetValue) ZPopMax() (ZSetEntry, bool)

ZPopMax removes and returns the member with the highest score

func (*ZSetValue) ZPopMaxN 

func (z *ZSetValue) ZPopMaxN(n int) []ZSetEntry

ZPopMaxN removes and returns up to n members with the highest scores

func (*ZSetValue) ZPopMin 

func (z *ZSetValue) ZPopMin() (ZSetEntry, bool)

ZPopMin removes and returns the member with the lowest score

func (*ZSetValue) ZPopMinN 

func (z *ZSetValue) ZPopMinN(n int) []ZSetEntry

ZPopMinN removes and returns up to n members with the lowest scores

func (*ZSetValue) ZRange 

func (z *ZSetValue) ZRange(start, stop int, reverse bool) []ZSetEntry

ZRange returns members in the given rank range (0-indexed, inclusive)

func (*ZSetValue) ZRangeByScore 

func (z *ZSetValue) ZRangeByScore(min, max float64, minExclusive, maxExclusive, reverse bool, offset, count int) []ZSetEntry

ZRangeByScore returns members with scores in the given range

func (*ZSetValue) ZRank 

func (z *ZSetValue) ZRank(member string) int

ZRank returns the rank (0-indexed) of a member in ascending order Returns -1 if member not found

func (*ZSetValue) ZRem 

func (z *ZSetValue) ZRem(members ...string) int

ZRem removes members from the sorted set Returns the number of members removed

func (*ZSetValue) ZRevRank 

func (z *ZSetValue) ZRevRank(member string) int

ZRevRank returns the rank (0-indexed) of a member in descending order Returns -1 if member not found

func (*ZSetValue) ZScore 

func (z *ZSetValue) ZScore(member string) (float64, bool)

ZScore returns the score of a member

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.