Help us shape the future of Go! Take the 2026 Go Developer Survey

redis

package
v0.5.3 Latest Latest
Warning

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

 Go to latest Published: Nov 3, 2026 License: MIT Imports: 45 Imported by: 0

Documentation

Overview

Package redis owns an explicitly configured Redis connection pool. Feature packages consume its focused capabilities; application code uses typed handles.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func Module 

func Module(name foundation.ProviderID, key foundation.Key[*Client], config Config) foundation.Module

Module prepares a distinct client for each application and starts it at boot. Dependents resolve the typed key during construction; cleanup is installed before the connectivity check and participates in reverse-order shutdown.

Types

type Client 

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

Client owns its connections and bounded in-flight operations. A cache.Store borrows this client; it does not close it. Use Module for application ownership.

The per-operation path uses atomics and a FIFO admission semaphore only; the mutex guards lifecycle transitions and the subscription registry.

func Open 

func Open(ctx context.Context, config Config) (*Client, error)

Open prepares and starts a client, closing all owned resources on startup error.

func Prepare 

func Prepare(config Config) (*Client, error)

Prepare snapshots validated configuration without I/O or background goroutines.

func (*Client) Add 

func (c *Client) Add(ctx context.Context, key cache.EntryKey, data []byte, ttl cache.TTL) (bool, error)

Add atomically inserts an absent cache entry and reports whether it was added.

func (*Client) AddSnapshot 

func (c *Client) AddSnapshot(ctx context.Context, base cache.EntryKey, tags []cache.EntryKey, data []byte, ttl cache.TTL) (bool, error)

AddSnapshot fills an absent or invalidated entry under the current snapshot.

func (*Client) AddTagged 

func (c *Client) AddTagged(ctx context.Context, key cache.TaggedKey, data []byte, ttl cache.TTL) (bool, error)

AddTagged atomically fills an absent or invalidated entry under this snapshot. An existing current value retains its bytes and expiry.

func (*Client) AdvanceScheduleCursor 

func (c *Client) AdvanceScheduleCursor(ctx context.Context, key lease.Key, at time.Time, ttl time.Duration) error

AdvanceScheduleCursor raises one persisted cursor and refreshes its expiry.

func (*Client) ClearRateLimit 

func (c *Client) ClearRateLimit(ctx context.Context, key ratelimit.Key) (bool, error)

ClearRateLimit deletes the key's bucket, including unreadable state at its exact address. A lost reply leaves the deletion unconfirmed; it is not retried.

func (*Client) Close 

func (c *Client) Close(ctx context.Context) error

Close rejects new work and drains owned operations before closing connections. Caller cancellation only stops waiting; cleanup continues. In-flight commands retain their own bounded contexts. Close is safe to repeat concurrently.

func (*Client) DataCount 

func (c *Client) DataCount(ctx context.Context, key data.Key, l data.Limits) (uint64, error)

DataCount returns collection cardinality after checking its type and supplied bounds.

func (*Client) DataDeleteMany 

func (c *Client) DataDeleteMany(ctx context.Context, keys []data.Key) (uint64, error)

DataDeleteMany validates and atomically deletes a canonical batch, returning the number of removed keys.

func (*Client) DataExists 

func (c *Client) DataExists(ctx context.Context, key data.Key) (bool, error)

DataExists checks whether the declared collection exists with the expected native type.

func (*Client) DataExpire 

func (c *Client) DataExpire(ctx context.Context, key data.Key, ttl cache.TTL) (bool, error)

DataExpire atomically changes an existing collection's TTL while preserving its contents.

func (*Client) Done 

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

Done closes after all operations have drained and connections have closed. The zero client has no lifecycle and returns nil.

func (*Client) ExecuteRaw 

func (c *Client) ExecuteRaw(ctx context.Context, request raw.Request, limits raw.Limits) (raw.Reply, error)

ExecuteRaw performs one attempt. Missing Redis replies remain explicit nil values for a nullable decoder; server and transport errors never become misses.

func (*Client) ExecuteRawBatch 

func (c *Client) ExecuteRawBatch(ctx context.Context, requests []raw.Request, mode raw.Mode, limits raw.Limits) ([]raw.Reply, error)

ExecuteRawBatch executes a validated batch once, bounding combined replies. Transaction mode does not roll back successful commands when another command fails.

func (*Client) Exists 

func (c *Client) Exists(ctx context.Context, key cache.EntryKey) (bool, error)

Exists checks the stored type and byte bound without transferring the value.

func (*Client) ExistsTagged 

func (c *Client) ExistsTagged(ctx context.Context, key cache.TaggedKey) (bool, error)

ExistsTagged checks entry existence under a current tag snapshot without transferring the payload.

func (*Client) Expire 

func (c *Client) Expire(ctx context.Context, key cache.EntryKey, ttl cache.TTL) (bool, error)

Expire changes a live entry's expiry without replacing its bytes. Existing persistent entries return true when Forever is requested again.

func (*Client) ExpireRaw 

func (c *Client) ExpireRaw(ctx context.Context, key raw.Key, ttl cache.TTL) (bool, error)

ExpireRaw atomically changes an existing native key's TTL; Forever removes expiry.

func (*Client) ExpireSnapshot 

func (c *Client) ExpireSnapshot(ctx context.Context, base cache.EntryKey, tags []cache.EntryKey, ttl cache.TTL) (bool, error)

ExpireSnapshot changes the expiry of the current snapshot's entry.

func (*Client) ExpireTagged 

func (c *Client) ExpireTagged(ctx context.Context, key cache.TaggedKey, ttl cache.TTL) (bool, error)

ExpireTagged atomically checks the tag snapshot and changes a live entry's TTL.

func (*Client) Forget 

func (c *Client) Forget(ctx context.Context, key cache.EntryKey) (bool, error)

Forget atomically removes a cache entry and reports whether it existed.

func (*Client) ForgetMany 

func (c *Client) ForgetMany(ctx context.Context, keys []cache.EntryKey) (uint64, error)

ForgetMany atomically deletes a bounded canonical batch after validating every stored entry. A lost reply is an uncertain outcome, never a reason to retry.

func (*Client) ForgetManyTagged 

func (c *Client) ForgetManyTagged(ctx context.Context, keys []cache.TaggedKey) (uint64, error)

ForgetManyTagged checks one shared tag snapshot and every selected payload before one exact DEL. Stale physical entries are excluded from the live count.

func (*Client) ForgetSnapshot 

func (c *Client) ForgetSnapshot(ctx context.Context, base cache.EntryKey, tags []cache.EntryKey) (bool, error)

ForgetSnapshot removes the entry of the current snapshot; obsolete payloads are reclaimed but not counted.

func (*Client) ForgetTagged 

func (c *Client) ForgetTagged(ctx context.Context, key cache.TaggedKey) (bool, error)

ForgetTagged conditionally removes this tagged entry. A stale caller cannot remove a replacement written under a newer snapshot.

func (*Client) FoundryAdapter 

func (*Client) FoundryAdapter(frameworkadapter.Seal)

FoundryAdapter marks the client as framework-owned adapter I/O.

func (*Client) Get 

func (c *Client) Get(ctx context.Context, key cache.EntryKey) ([]byte, bool, error)

Get returns an owned value, distinguishing a present empty value from a miss.

func (*Client) GetMany 

func (c *Client) GetMany(ctx context.Context, keys []cache.EntryKey) ([]cache.BatchValue, error)

GetMany reads a bounded canonical batch in one script; results follow input order. Missing, expired, over-bound and wrong-type entries are misses.

func (*Client) GetManyTagged 

func (c *Client) GetManyTagged(ctx context.Context, keys []cache.TaggedKey) ([]cache.BatchValue, error)

GetManyTagged checks one shared snapshot and reads every payload in the same script. Obsolete payloads are misses and are reclaimed.

func (*Client) GetTagged 

func (c *Client) GetTagged(ctx context.Context, key cache.TaggedKey) ([]byte, bool, error)

GetTagged returns owned bytes under a current tag snapshot. A changed/missing snapshot returns Conflict; an older stored payload is removed conditionally.

func (*Client) HashDelete 

func (c *Client) HashDelete(ctx context.Context, key data.Key, field string, l data.Limits) (bool, error)

HashDelete removes a field and reports whether it existed.

func (*Client) HashGet 

func (c *Client) HashGet(ctx context.Context, key data.Key, field string, l data.Limits) (string, bool, error)

HashGet returns a bounded encoded field value and distinguishes absence.

func (*Client) HashGetAll 

func (c *Client) HashGetAll(ctx context.Context, key data.Key, l data.Limits) ([]data.StoredField, error)

HashGetAll returns every field sorted by field bytes, after bounding cardinality, field sizes and the value reply budget on the server. The script returns fields unordered (Lua comparison follows the server locale); the adapter sorts them bytewise and rejects duplicates.

func (*Client) HashGetMany 

func (c *Client) HashGetMany(ctx context.Context, key data.Key, fields []string, l data.Limits) ([]value.Optional[string], error)

HashGetMany reads the requested fields atomically after checking every stored size against the value and reply bounds.

func (*Client) HashIncrement 

func (c *Client) HashIncrement(ctx context.Context, key data.Key, field string, delta int64, l data.Limits) (int64, error)

HashIncrement uses Redis signed 64-bit arithmetic on a canonical integer field.

func (*Client) HashSet 

func (c *Client) HashSet(ctx context.Context, key data.Key, field, value string, l data.Limits) (bool, error)

HashSet stores a bounded encoded value and reports whether the field was new, preserving existing expiry.

func (*Client) Increment 

func (c *Client) Increment(ctx context.Context, key cache.EntryKey, delta int64, initialTTL cache.TTL) (int64, error)

Increment uses server integer arithmetic and retains an existing entry's TTL. Missing entries begin at zero and receive initialTTL. No mutation is retried.

func (*Client) IncrementSnapshot 

func (c *Client) IncrementSnapshot(ctx context.Context, base cache.EntryKey, tags []cache.EntryKey, delta int64, ttl cache.TTL) (int64, error)

IncrementSnapshot applies exact signed arithmetic under the current snapshot.

func (*Client) IncrementTagged 

func (c *Client) IncrementTagged(ctx context.Context, key cache.TaggedKey, delta int64, ttl cache.TTL) (int64, error)

IncrementTagged combines snapshot validation with exact signed arithmetic. Existing expiry is retained; an absent/invalidated counter starts at zero.

func (*Client) InvalidateTags 

func (c *Client) InvalidateTags(ctx context.Context, keys []cache.EntryKey) error

InvalidateTags atomically rotates the entire selected set. An uncertain remote error may hide an applied rotation; callers must not infer rollback or retry.

func (*Client) LeaseAcquire 

func (c *Client) LeaseAcquire(ctx context.Context, key lease.Key, owner lease.Owner, ttl time.Duration) (bool, error)

LeaseAcquire attempts once against this Redis authority. No mutation is retried.

func (*Client) LeaseForceRelease 

func (c *Client) LeaseForceRelease(ctx context.Context, key lease.Key) (bool, error)

LeaseForceRelease administratively deletes one exact lease address whatever its owner, type or expiry, repairing metadata that ordinary operations reject as corrupt (such as a lease key without TTL). It is never used implicitly.

func (*Client) LeaseRelease 

func (c *Client) LeaseRelease(ctx context.Context, key lease.Key, owner lease.Owner) (bool, error)

LeaseRelease deletes only the matching live owner. Expiry/replacement returns false.

func (*Client) LeaseRenew 

func (c *Client) LeaseRenew(ctx context.Context, key lease.Key, owner lease.Owner, ttl time.Duration) (bool, error)

LeaseRenew extends only the matching live owner; it cannot recreate an expired key.

func (*Client) LeaseTransfer 

func (c *Client) LeaseTransfer(ctx context.Context, key lease.Key, from, to lease.Owner, ttl time.Duration) (bool, error)

LeaseTransfer atomically replaces the live owner from with to and renews the lease; a different, expired or absent owner returns false without change.

func (*Client) ListPop 

func (c *Client) ListPop(ctx context.Context, key data.Key, end data.End, l data.Limits) (string, bool, error)

ListPop removes one element after checking its size.

func (*Client) ListPush 

func (c *Client) ListPush(ctx context.Context, key data.Key, values []string, end data.End, l data.Limits) (uint64, error)

ListPush inserts every value in one command after checking sizes and the resulting length, and returns the new length.

func (*Client) ListRange 

func (c *Client) ListRange(ctx context.Context, key data.Key, start, stop int64, l data.Limits) ([]string, error)

ListRange returns elements start..stop inclusive within the reply bounds.

func (*Client) ListTrim 

func (c *Client) ListTrim(ctx context.Context, key data.Key, start, stop int64, l data.Limits) error

ListTrim keeps elements start..stop inclusive (Redis LTRIM semantics).

func (*Client) LockoutBegin 

func (c *Client) LockoutBegin(ctx context.Context, key lockout.Key, policy lockout.Policy, candidate lockout.Generation) (lockout.Admission, error)

LockoutBegin atomically checks the shared failure window using Redis TIME. This client owns no process-local account state and never retries a lost reply.

func (*Client) LockoutFinish 

func (c *Client) LockoutFinish(ctx context.Context, key lockout.Key, policy lockout.Policy, snapshot lockout.Snapshot, outcome lockout.Outcome) (lockout.Decision, error)

LockoutFinish atomically records the outcome against the admitted snapshot and returns the current lockout decision.

func (*Client) LockoutReset 

func (c *Client) LockoutReset(ctx context.Context, key lockout.Key, policy lockout.Policy) (bool, error)

LockoutReset removes lockout state and reports whether state existed.

func (*Client) PeekRateLimit 

func (c *Client) PeekRateLimit(ctx context.Context, key ratelimit.Key, limit ratelimit.Limit, cost uint32) (ratelimit.Decision, error)

PeekRateLimit runs the same atomic script read-only: it reports the decision cost would receive now without consuming capacity or changing expiry.

func (*Client) Ping 

func (c *Client) Ping(ctx context.Context) error

Ping verifies the running client under the same operation bounds as features.

func (*Client) Publish 

func (c *Client) Publish(ctx context.Context, channel pubsub.Channel, data []byte) (uint64, error)

Publish attempts one ephemeral publication. The count is server subscriptions, not delivery/handler completion. Uncertain publications are never retried.

func (*Client) Put 

func (c *Client) Put(ctx context.Context, key cache.EntryKey, data []byte, ttl cache.TTL) error

Put atomically replaces a cache entry and its TTL without retrying an uncertain mutation.

func (*Client) PutLeased 

func (c *Client) PutLeased(ctx context.Context, key cache.EntryKey, data []byte, ttl cache.TTL, proof lease.Proof) error

PutLeased atomically checks this fill's current lease and publishes the value. Old owners cannot write after a successor has acquired the same fill address.

func (*Client) PutSnapshot 

func (c *Client) PutSnapshot(ctx context.Context, base cache.EntryKey, tags []cache.EntryKey, data []byte, ttl cache.TTL) error

PutSnapshot resolves tag metadata (creating fresh versions for missing metadata) and writes under the resolved snapshot in one atomic script.

func (*Client) PutTagged 

func (c *Client) PutTagged(ctx context.Context, key cache.TaggedKey, data []byte, ttl cache.TTL) error

PutTagged validates the snapshot and writes an owned value with explicit expiry. Corruption, stale snapshots and denied mutation permissions preserve live data.

func (*Client) PutTaggedLeased 

func (c *Client) PutTaggedLeased(ctx context.Context, key cache.TaggedKey, data []byte, ttl cache.TTL, proof lease.Proof) error

PutTaggedLeased checks the lease and tag snapshot in the same atomic operation. It reuses tagged payload/expiry validation and permission preflight.

func (*Client) RateLimit 

func (c *Client) RateLimit(ctx context.Context, key ratelimit.Key, limit ratelimit.Limit, cost uint32) (ratelimit.Decision, error)

RateLimit makes one atomic decision using Redis TIME and the key's phased fixed windows. A live bucket with a different policy is converted, keeping its admitted usage; corrupt metadata fails without mutation. Lost replies never trigger retry; their capacity consumption remains unknown to the caller.

func (*Client) ReadSnapshot 

func (c *Client) ReadSnapshot(ctx context.Context, base cache.EntryKey, tags []cache.EntryKey, payload bool) (cache.TaggedKey, []byte, bool, error)

ReadSnapshot resolves tag/namespace metadata (creating fresh versions for missing metadata) and reads the tagged entry in one atomic script, so every typed read costs one round trip. Obsolete, over-bound or malformed payloads are misses and are reclaimed. The reply never includes a stale value.

func (*Client) ResolveTags 

func (c *Client) ResolveTags(ctx context.Context, keys []cache.EntryKey) ([]cache.TagVersion, error)

ResolveTags atomically creates fresh versions for missing metadata and returns the current versions in canonical input order. Redis tags require Redis 7+.

func (*Client) ScheduleCursor 

func (c *Client) ScheduleCursor(ctx context.Context, key lease.Key) (time.Time, bool, error)

ScheduleCursor reads one persisted catch-up cursor.

func (*Client) SetAdd 

func (c *Client) SetAdd(ctx context.Context, key data.Key, member string, l data.Limits) (bool, error)

SetAdd inserts an encoded member within the cardinality bound and reports whether it was new.

func (*Client) SetContains 

func (c *Client) SetContains(ctx context.Context, key data.Key, member string, l data.Limits) (bool, error)

SetContains tests exact encoded member equality after checking collection bounds.

func (*Client) SetMembers 

func (c *Client) SetMembers(ctx context.Context, key data.Key, l data.Limits) ([]string, error)

SetMembers returns owned, unique member strings in byte order within the reply bounds.

func (*Client) SetRemove 

func (c *Client) SetRemove(ctx context.Context, key data.Key, member string, l data.Limits) (bool, error)

SetRemove removes an encoded member and reports whether it existed.

func (*Client) SortedSetAdd 

func (c *Client) SortedSetAdd(ctx context.Context, key data.Key, member string, score float64, l data.Limits) (bool, error)

SortedSetAdd inserts or re-scores one member, rejecting growth when full.

func (*Client) SortedSetCountByScore 

func (c *Client) SortedSetCountByScore(ctx context.Context, key data.Key, scores data.ScoreRange, l data.Limits) (uint64, error)

SortedSetCountByScore counts members matching the validated score range within the collection bound.

func (*Client) SortedSetIncrement 

func (c *Client) SortedSetIncrement(ctx context.Context, key data.Key, member string, delta float64, l data.Limits) (float64, error)

SortedSetIncrement adds delta to one member's score (missing starts at zero).

func (*Client) SortedSetRange 

func (c *Client) SortedSetRange(ctx context.Context, key data.Key, window data.Window, l data.Limits) ([]data.StoredMember, error)

SortedSetRange returns a rank window with scores.

func (*Client) SortedSetRangeByScore 

func (c *Client) SortedSetRangeByScore(ctx context.Context, key data.Key, scores data.ScoreRange, window data.Window, l data.Limits) ([]data.StoredMember, error)

SortedSetRangeByScore returns a score range, windowed with Redis LIMIT.

func (*Client) SortedSetRank 

func (c *Client) SortedSetRank(ctx context.Context, key data.Key, member string, order data.Order, l data.Limits) (uint64, bool, error)

SortedSetRank returns a member's zero-based rank in the selected order and distinguishes absence.

func (*Client) SortedSetRemove 

func (c *Client) SortedSetRemove(ctx context.Context, key data.Key, member string, l data.Limits) (bool, error)

SortedSetRemove removes an encoded member and reports whether it existed.

func (*Client) SortedSetScore 

func (c *Client) SortedSetScore(ctx context.Context, key data.Key, member string, l data.Limits) (float64, bool, error)

SortedSetScore returns a member's score and distinguishes absence.

func (*Client) Start 

func (c *Client) Start(ctx context.Context) error

Start checks connectivity once. The first caller owns that attempt; concurrent callers can cancel their wait. A failed start is terminal for this instance.

func (*Client) Stats 

func (c *Client) Stats() Stats

Stats returns current operation, subscription, and pool counts, treating an uninitialized client as closing.

func (*Client) Subscribe 

func (c *Client) Subscribe(ctx context.Context, channels []pubsub.Channel, limits pubsub.Limits) (pubsub.Stream, error)

Subscribe confirms every exact channel before returning. All subscriptions of one Client share one dedicated connection with one reader and one heartbeat; a channel already confirmed on that connection is ready immediately, and a new channel waits for its own SUBSCRIBE acknowledgement. Establishment context cancellation never leaves an unowned reader. Any connection failure (receive error, heartbeat loss or unexpected protocol frame) terminates EVERY stream on that connection with pubsub.ErrDisconnected; vendor reconnect attempts never imply continuous delivery or cause automatic public resubscription.

type Config 

type Config struct {
	Host           string
	Port           uint16
	Database       int
	User           string
	Password       secret.String
	TLS            TLSMode
	TLSConfig      *tls.Config
	MaxConnections int
	MaxOperations  int
	// MaxSubscriptions separately bounds dedicated pub/sub connections.
	MaxSubscriptions int
	ConnectTimeout   time.Duration
	OperationTimeout time.Duration
	PoolTimeout      time.Duration
	MaxIdleTime      time.Duration
	MaxLifetime      time.Duration
	// MaxValueBytes bounds cache writes and server-side reads before GET returns
	// the value. It is not a bound on an untrusted server's RESP parser allocations.
	MaxValueBytes int
}

Config addresses one standalone Redis server. It performs no environment or credential-file discovery. Cluster and Sentinel routing are not implemented. TLS certificate/key material must remain immutable after Prepare.

func DefaultConfig 

func DefaultConfig() Config

DefaultConfig sizes the pool like go-redis's own default (10 connections per CPU) on a typical 4–8 core host, with a fixed value so configuration validation stays deterministic across machines. MaxOperations bounds admitted commands, including those waiting for a pooled connection; a burst beyond it queues briefly and then reports fault.Overloaded.

func (Config) Validate 

func (c Config) Validate() error

Validate checks the explicit endpoint, credentials, capacity, timeouts, and TLS policy without connecting.

type Connection 

type Connection struct {
	Name  ConnectionName
	Value *Client
}

Connection binds a typed name to an existing borrowed instance.

type ConnectionName 

type ConnectionName string

ConnectionName identifies one configured redis instance, distinct from other service families.

func (ConnectionName) Validate 

func (n ConnectionName) Validate() error

Validate checks the configured service name.

type Connections 

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

Connections is immutable. Default and named access alias the same owned instance. Resolve dependencies during construction; pass concrete services to domain code.

func NewConnections 

func NewConnections(selected ConnectionName, entries ...Connection) (*Connections, error)

NewConnections validates named clients and the selected default while borrowing their lifecycles.

func (*Connections) Connection 

func (r *Connections) Connection(name ConnectionName) (*Client, error)

Connection returns the registered client or an error if name is unknown.

func (*Connections) Default 

func (r *Connections) Default() (*Client, error)

Default returns the selected registered client or an error for an uninitialized registry.

func (*Connections) DefaultName 

func (r *Connections) DefaultName() ConnectionName

DefaultName returns the selected name, or an empty name for a nil registry.

func (*Connections) Names 

func (r *Connections) Names() []ConnectionName

Names returns an owned, sorted list of names, or nil for a nil registry.

type JobBackend 

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

JobBackend borrows the existing Redis client and owns no connections. Limits apply per queue and must match across processes. Accepted work persists until terminal retention expires or retained-record eviction; durability depends on Redis persistence/failover. Stop workers and publication before closing the client's owning module.

Queues use storage layout 2: the immutable envelope is stored apart from small mutable state. A layout-1 queue is never migrated implicitly: every operation returns jobs.ErrLegacyLayout until an operator stops or drains the previous release and runs JobMigrateLayout (`jobs migrate-layout`). The previous release cannot read a migrated queue.

func NewJobBackend 

func NewJobBackend(client *Client, config jobs.QueueConfig) (*JobBackend, error)

NewJobBackend validates queue configuration and borrows a prepared Redis client without I/O.

func (*JobBackend) DurableAcceptance 

func (*JobBackend) DurableAcceptance() bool

DurableAcceptance describes Redis acknowledgement under the deployment's persistence/failover policy; it is not an fsync or replication guarantee.

func (*JobBackend) JobCancel 

func (b *JobBackend) JobCancel(ctx context.Context, key jobs.Key, target jobs.Target) (bool, error)

JobCancel requests cancellation of the matching target and reports whether state changed.

func (*JobBackend) JobCancelWorkflow 

func (b *JobBackend) JobCancelWorkflow(ctx context.Context, key jobs.Key, id jobs.WorkflowID) (bool, error)

JobCancelWorkflow requests cancellation of a workflow and reports whether state changed.

func (*JobBackend) JobEnqueue 

func (b *JobBackend) JobEnqueue(ctx context.Context, key jobs.Key, envelope jobs.Envelope) (bool, error)

JobEnqueue atomically accepts an envelope and reports whether it was newly inserted. A new insertion wakes this process's waiting workers.

func (*JobBackend) JobFinish 

func (b *JobBackend) JobFinish(ctx context.Context, key jobs.Key, proof jobs.Ownership, result jobs.Result) (bool, error)

JobFinish applies the validated result to owned work and reports whether state changed.

func (*JobBackend) JobForget 

func (b *JobBackend) JobForget(ctx context.Context, key jobs.Key, target jobs.Target) (bool, error)

JobForget removes one retained terminal independent record, its envelope and its deduplication identity. Live and workflow records are never removed.

func (*JobBackend) JobInspect 

func (b *JobBackend) JobInspect(ctx context.Context, key jobs.Key, id jobs.ExecutionID) (value.Optional[jobs.Record], error)

JobInspect returns a decoded retained record, or an unset result when the identity is absent.

func (*JobBackend) JobList 

func (b *JobBackend) JobList(ctx context.Context, key jobs.Key, options jobs.ListOptions) (jobs.Page, error)

JobList returns a bounded filtered page of decoded job records and its continuation identity.

func (*JobBackend) JobMigrateLayout 

func (b *JobBackend) JobMigrateLayout(ctx context.Context, key jobs.Key) (bool, error)

JobMigrateLayout moves a layout-1 queue to layout 2 in one atomic script. Records keep their IDs, state and history; embedded envelopes move on each record's next write. It is idempotent.

func (*JobBackend) JobRenew 

func (b *JobBackend) JobRenew(ctx context.Context, key jobs.Key, proof jobs.Ownership, ttl time.Duration) (jobs.LeaseStatus, error)

JobRenew extends a job lease and returns ownership and cancellation status.

func (*JobBackend) JobReserve 

func (b *JobBackend) JobReserve(ctx context.Context, key jobs.Key, owner lease.Owner, ttl time.Duration) (value.Optional[jobs.Reservation], error)

JobReserve atomically leases eligible work to owner, returning an unset result when none is available.

func (*JobBackend) JobRetry 

func (b *JobBackend) JobRetry(ctx context.Context, key jobs.Key, request jobs.RetryRequest) (bool, error)

JobRetry requeues an eligible failed job using the retry token and an atomically checked record fingerprint. An already applied token reports no change.

func (*JobBackend) JobStart 

func (b *JobBackend) JobStart(ctx context.Context, key jobs.Key, proof jobs.Ownership) (uint32, error)

JobStart marks owned work as running and returns its attempt number.

func (*JobBackend) JobStats 

func (b *JobBackend) JobStats(ctx context.Context, key jobs.Key) (jobs.QueueStats, error)

JobStats reports queue depth from indexes and counters, without payloads.

func (*JobBackend) JobWakeup 

func (b *JobBackend) JobWakeup(key jobs.Key) <-chan struct{}

JobWakeup wakes idle workers of this process subscribed to key after this process's own accepted enqueue in that queue. Workers in other processes rely on their adaptive idle polling.

func (*JobBackend) JobWorkflow 

func (b *JobBackend) JobWorkflow(ctx context.Context, key jobs.Key, workflow jobs.WorkflowEnvelope) (bool, error)

JobWorkflow atomically accepts the workflow and its members and reports whether it was newly inserted.

func (*JobBackend) JobWorkflowStatus 

func (b *JobBackend) JobWorkflowStatus(ctx context.Context, key jobs.Key, id jobs.WorkflowID) (value.Optional[jobs.WorkflowStatus], error)

JobWorkflowStatus reports one workflow's progress without payloads.

type Stats 

type Stats struct {
	Open, Idle, Operations int
	// Subscriptions includes establishment/drain; SubscriptionConnections is the
	// driver's live dedicated connection count, separate from Open.
	Subscriptions, SubscriptionConnections int
	Ready, Closing                         bool
}

Stats is a concurrent snapshot of owned operations and connection use.

type TLSMode 

type TLSMode string

TLSMode selects verified TLS or explicitly unencrypted transport. There is no automatic downgrade or fallback to a different endpoint. 

const (
	VerifyTLS  TLSMode = "verify-full"
	DisableTLS TLSMode = "disable"
)

type WebSocketBackend 

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

WebSocketBackend borrows the existing Redis pool and pub/sub implementation. It owns no server/connection lifecycle and never retries a state mutation. Keep its Client alive until every borrowing Hub/Publisher has actually stopped.

func NewWebSocketBackend 

func NewWebSocketBackend(client *Client) (*WebSocketBackend, error)

NewWebSocketBackend borrows a prepared Redis client without I/O or a separate connection lifecycle.

func (*WebSocketBackend) Publish 

func (b *WebSocketBackend) Publish(ctx context.Context, channel pubsub.Channel, data []byte) (uint64, error)

Publish delegates ephemeral publication to the borrowed client and returns the server subscription count.

func (*WebSocketBackend) Subscribe 

func (b *WebSocketBackend) Subscribe(ctx context.Context, channels []pubsub.Channel, limits pubsub.Limits) (pubsub.Stream, error)

Subscribe establishes an exact-channel stream through the borrowed client.

func (*WebSocketBackend) WebSocketAppend 

func (b *WebSocketBackend) WebSocketAppend(ctx context.Context, key websocket.ClusterKey, channel websocket.ChannelID, policy websocket.ReplayConfig, data []byte) error

WebSocketAppend stores a validated publication in bounded replay history without publishing it.

func (*WebSocketBackend) WebSocketCheck 

func (b *WebSocketBackend) WebSocketCheck(ctx context.Context, key websocket.ClusterKey) error

WebSocketCheck validates or establishes the cluster policy under Redis authority.

func (*WebSocketBackend) WebSocketClose 

func (b *WebSocketBackend) WebSocketClose(ctx context.Context, key websocket.ClusterKey, instance websocket.InstanceID, connection websocket.ConnectionID) error

WebSocketClose releases an owned connection and its memberships; an absent connection is a no-op.

func (*WebSocketBackend) WebSocketHistory 

func (b *WebSocketBackend) WebSocketHistory(ctx context.Context, key websocket.ClusterKey, channel websocket.ChannelID, policy websocket.ReplayConfig) ([][]byte, error)

WebSocketHistory returns owned retained frames within the replay count, byte, and retention bounds.

func (*WebSocketBackend) WebSocketJoin 

func (b *WebSocketBackend) WebSocketJoin(ctx context.Context, key websocket.ClusterKey, instance websocket.InstanceID, connection websocket.ConnectionID, membership websocket.ClusterMembership) (websocket.PresenceSnapshot, error)

WebSocketJoin registers membership for an owned live connection and returns its presence snapshot.

func (*WebSocketBackend) WebSocketLeave 

func (b *WebSocketBackend) WebSocketLeave(ctx context.Context, key websocket.ClusterKey, instance websocket.InstanceID, connection websocket.ConnectionID, scope websocket.Scope) error

WebSocketLeave removes a scope membership for the owned connection; an absent connection is a no-op.

func (*WebSocketBackend) WebSocketMembers 

func (b *WebSocketBackend) WebSocketMembers(ctx context.Context, key websocket.ClusterKey, scope websocket.Scope) (websocket.PresenceSnapshot, error)

WebSocketMembers returns an owned presence snapshot ordered by member identity.

func (*WebSocketBackend) WebSocketOpen 

func (b *WebSocketBackend) WebSocketOpen(ctx context.Context, key websocket.ClusterKey, instance websocket.InstanceID, connection websocket.ConnectionID) error

WebSocketOpen registers a connection owned by instance within cluster capacity.

func (*WebSocketBackend) WebSocketTouch 

func (b *WebSocketBackend) WebSocketTouch(ctx context.Context, key websocket.ClusterKey, instance websocket.InstanceID, connection websocket.ConnectionID) error

WebSocketTouch renews the owned connection and its membership leases.

Source Files

View all Source files

Directories

Path Synopsis
Package data provides typed Redis hashes, sets, sorted sets and lists.
 Package data provides typed Redis hashes, sets, sorted sets and lists.
Package raw is the explicit trusted Redis command boundary.
 Package raw is the explicit trusted Redis command boundary.

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.