Documentation
¶
Overview ¶
Package redis is the Redis-backed queue driver. Compared to the SQL drivers, this one trades a relational source-of-truth for Redis' server-side collections: each op id flows from the `pending` SORTED SET into the `running` LIST and on to `done` | `failed` | `cancelled`, while the canonical fields live in a parallel HASH at `<prefix>:data:<id>`.
⚠ The pending set is a ZSET, not a LIST, and that is the whole point. A LIST is positional: consumed with BLMOVE it hands out ops in arrival order and there is nowhere for Op.Priority to be expressed, so the field was persisted, returned by Get, rendered by the admin UI — and ignored by the one operation it exists for. The SQL drivers claim with `ORDER BY priority DESC, enqueued_at ASC`; on Redis a scan the storage sync discovered still went out ahead of a person's upload, and the 41 seconds an interactive scan waited behind a 20 000-file first import were still being paid. The ZSET score encodes exactly that ordering — see pendingScore.
Claiming is one Lua script (dequeueScript). Redis runs it atomically, which buys two things a LIST could not:
- type filtering with no mutation. A LIST cannot be searched, so the old code popped the head, looked at it, and pushed mismatches back at the far end. A ZRANGE walks candidates in priority order and leaves the ones it does not want exactly where they are.
- pop and claim in a single step. BLMOVE moved the id to `running` and a separate transaction then flipped the hash to `running`; a crash in between left an op no list owned and no recovery pass could interpret. Now the ZREM, the LPUSH, the status write, the attempt bump and the release of the coalescing claim either all happen or none do.
The cost is that the claim is no longer a blocking Redis command. A worker that finds nothing waits on `<prefix>:signal`, a capped LIST every push into pending writes one token to, so an arriving op still wakes a worker in about a millisecond rather than at the next poll. See Dequeue.
Delayed dispatch (Op.NotBefore) lives in a second ZSET keyed by the unix epoch deadline. A promoter goroutine, started during Init, sweeps it every second and atomically moves due ids into pending.
Index ¶
- type Driver
- func (d *Driver) Ack(ctx context.Context, id string) error
- func (d *Driver) Cancel(ctx context.Context, id string) error
- func (d *Driver) Close() error
- func (d *Driver) Dequeue(ctx context.Context, types []string) (queue.Op, error)
- func (d *Driver) Enqueue(ctx context.Context, op queue.Op) (string, error)
- func (d *Driver) Fail(ctx context.Context, id, errMsg string, retry bool) error
- func (d *Driver) Get(ctx context.Context, id string) (queue.Op, error)
- func (d *Driver) Init(ctx context.Context, cfg map[string]any) error
- func (d *Driver) List(ctx context.Context, status string, limit, offset int) ([]queue.Op, int64, error)
- func (*Driver) Name() string
- func (d *Driver) RecoverOrphans(ctx context.Context, olderThan time.Duration) (int64, error)
- func (d *Driver) Retry(ctx context.Context, id string) error
- func (d *Driver) Stats(ctx context.Context) (queue.Stats, error)
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Driver ¶
type Driver struct {
// contains filtered or unexported fields
}
Driver is the Redis-backed queue.
func (*Driver) Ack ¶
Ack marks the op completed. The id moves from running → done; older done entries fall off via LTRIM so the list stays bounded.
func (*Driver) Cancel ¶
Cancel transitions a pending op to cancelled. Running ops cannot be cancelled in v0.1 — same contract as the SQL drivers.
func (*Driver) Close ¶
Close releases resources. The connection is only closed when the driver opened it (cfg["url"] path) — borrowed clients keep running. The promoter goroutine is always cancelled.
func (*Driver) Dequeue ¶
Dequeue claims the highest-priority runnable op whose type the caller handles, atomically, and blocks for up to blockTimeout when there is nothing to claim.
One Lua script does the claim (see dequeueScript). It walks the head of the pending ZSET — which is priority order, then arrival order within a priority — skipping ops of types this worker has no handler for and leaving them exactly where they are, and on the first match performs the whole claim in one indivisible step: out of pending, into running, hash flipped to `running` with started_at stamped, attempts bumped, and the coalescing claim released.
⚠ The atomicity is not decoration. The BLMOVE this replaced moved the id to `running` and then a SECOND round trip flipped the hash; a crash in the gap left an id sitting in the running list whose hash still said `pending`, and RecoverOrphans reads exactly that as a stale list entry and drops it — the op was gone from every list, permanently pending in its own hash, and nothing would ever run it. There is no such window now, and none in the other direction either: the ZREM is inside the same script as the claim, so no two workers can be handed the same op.
⚠⚠ What this costs: the claim is no longer a blocking Redis command. A script cannot block, so "wait for work" moved to a doorbell — every push into pending writes a token to a capped LIST and a worker with nothing to do waits on BLPOP for one. The observable behaviour is the same (an arriving op wakes a worker in about a millisecond, not at the next poll), with two honest differences: a token may be consumed by a worker whose type filter does not match the op that produced it, and a drained burst can leave up to signalCap stale tokens behind. Both cost one extra script call and then the worker's own poll interval — never a missed op, because the script is re-run after the wait and the pool re-enters Dequeue immediately after every successful claim.
func (*Driver) Enqueue ¶
Enqueue persists op and returns its assigned id. NotBefore set in the future routes the id into the scheduled ZSET; otherwise it lands in the pending set immediately, at the position its priority and arrival earn.
func (*Driver) Fail ¶
Fail records the failure. retry=true sends the op back through the scheduled ZSET with a 30s hold-off so a flapping upstream doesn't burn through the attempt budget instantly. retry=false parks the op in the failed list for operator inspection.
func (*Driver) Init ¶
Init wires the driver to its backing store. cfg keys:
url — Redis connection URL (redis://[:pass@]host:port/db).
Required when cfg["client"] is absent.
key_prefix — Optional namespace, default "filex:queue". The driver
appends `:pending`, `:running`, … to derive concrete
key names.
client — *redis.Client (preferred). When set, url is ignored
and Close() will not Close() the borrowed handle —
same lifecycle contract as the SQL drivers.
logger — *slog.Logger override; defaults to slog.Default().
func (*Driver) List ¶
func (d *Driver) List(ctx context.Context, status string, limit, offset int) ([]queue.Op, int64, error)
List returns ops filtered by status (or empty for any) with pagination. When status is empty we union all five status lists for totals — call Stats() instead if you only need counters.
func (*Driver) Name ¶
Name implements queue.Driver. Pointer receiver because Driver embeds sync.Once via the promoter plumbing; copying it would copy the lock.
func (*Driver) RecoverOrphans ¶
RecoverOrphans flips long-running ids back to pending. Called on boot to handle ungraceful shutdowns. The check pairs LRANGE running with HGET status + started_at — anything still claiming `running` whose started_at is older than `olderThan` ago is considered orphaned.
func (*Driver) Retry ¶
Retry transitions a failed op back to pending. The attempts counter is preserved so the operator can see how many tries it took before the manual intervention.
func (*Driver) Stats ¶
Stats returns the dashboard counters. Done24h is approximate — the done list is capped at doneCap and we have no per-entry timestamp in the list itself, so we fall back to "size of done list, capped at doneCap" which matches what the SQL drivers report when their retention windows align.
Source Files