Fetchers & Workers

TaskRunner separates reading messages from the queue (fetching) from running your task code (execution). The two run on independent goroutine pools and scale separately. This page covers what each does, how they cooperate, and the knobs that control them.

The two pools

Fetchers

A fetcher is a goroutine that loops forever, calling queue.Receive(...) which translates to XREADGROUP. Each fetcher reads up to BatchSize messages per round-trip and hands each one to the worker pool via workerPool.Invoke(...).

// runner/fetcher.go (simplified)
for {
    messages, err := t.queue.Receive(ctx, blockDuration, batchSize, group, consumerName)
    if err != nil { t.captureError(err); continue }
    for _, message := range messages {
        t.workerPool.Invoke(fetchedMessage{message: message, consumer: consumerName})
    }
}

Key properties:

• I/O-bound. A fetcher spends almost all its time blocked on XREADGROUP with Block = BlockDuration (default 5s). You rarely need more than a few.
• Consumer names are per-fetcher. The consumer name is <group>_<host>_<fetcherID>, so a multi-fetcher instance shows up as multiple consumers in XINFO CONSUMERS. This is normal and lets Redis spread the PEL.
• Self-healing. If a fetcher panics, the deferred recover logs it and a new fetcher is spawned to keep NumFetchers constant (unless the context is canceled).

Workers

Workers run inside an ants PoolWithFunc. Each Invoke triggers one call to TaskRunner.worker(...), which:

1. Unmarshals the message payload from JSON into a TaskMessage.
2. Looks up the *Task by TaskMessage.TaskName in the registry.
3. Enforces MaxRetry -- if the message's receive count exceeds it, the FailedTaskHandler runs and the message is dropped.
4. Runs task.Action(ctx, payload) in a separate goroutine (executeTask) so it can be heartbeated.
5. On success, XACKs (and XDELs if deleteOnAck). On error, captures the error and leaves the message unacked for reclaim.

flowchart LR
    subgraph Fetcher
        XR["XREADGROUP<br/>(batch)"]
    end
    subgraph "ants PoolWithFunc"
        W1[worker 1]
        W2[worker 2]
        WN[worker N]
    end
    XR -->|"Invoke per msg"| W1
    XR --> W2
    XR --> WN
    W1 --> Action1["task.Action"]
    W2 --> Action2["task.Action"]
    WN --> ActionN["task.Action"]
    Action1 -.heartbeat.-> HB["XCLAIM<br/>every ReservationTimeout"]
    Action1 -->|"on success"| Ack["XACK + XDEL"]

Heartbeats: preventing double-processing

The trickiest part of any at-least-once queue is handling slow tasks. If a worker is still running a 60-second task, but the visibility timeout is 30 seconds, Redis will hand the same message to another consumer after 30s -- causing two workers to process it simultaneously.

TaskRunner's solution: while the action runs, the worker re-claims the message on a timer, resetting its idle time:

for {
    select {
    case result := <-resultChannel:
        // action finished -> ack or fail
    case <-time.After(task.ReservationTimeout):
        if t.queue.RequireHeartHeartBeat() {
            t.queue.HeartBeat(ctx, group, consumer, messageID) // XCLAIM
        }
    }
}

Because XCLAIM resets the message's idle time, the pending-reclaimer (which looks for entries idle longer than reClaimDelay) will not steal it.

Set ReservationTimeout to a fraction of your task's worst case

If your task normally runs in 2s but occasionally in 30s, set ReservationTimeout to ~5s. Heartbeats fire every 5s, well under the reClaimDelay (default 5min), so the message is never stolen while the task is genuinely running. A shorter ReservationTimeout means more frequent Redis round-trips during long tasks; a longer one means slower recovery if the worker actually dies.

Configuration knobs

Field	Default	Effect
NumFetchers	= BatchSize	Number of concurrent fetcher goroutines.
NumWorkers	--	Capacity of the ants worker pool. With tuning disabled, this is fixed.
BatchSize	--	Messages read per XREADGROUP per fetcher.
BlockDuration	5s	How long XREADGROUP blocks waiting for new messages.
Task.MaxRetry	1 if 0	Max times a message is delivered before FailedTaskHandler runs.
Task.ReservationTimeout	--	Heartbeat interval for long-running tasks.
Task.FailedTaskHandler (cfg)	nil	Called on terminal failure; can ack to stop retries.

Adaptive batch sizing (tuning mode)

When worker tuning is enabled (MinWorkers/MaxWorkers/ DesiredWaitTime set), each fetcher also tracks the worker pool's free capacity and adjusts its batch size to follow it:

shouldUpdateBatchSize := func() (bool, int) {
    poolFreeCapacity := t.workerPool.Free()
    if time.Since(batchSizeLastUpdate) < TuningCooldownSeconds { return false, batchSize }
    if poolFreeCapacity < BatchSize { return false, batchSize }
    lowerBand := poolFreeCapacity - BatchSize
    upperBand := poolFreeCapacity + BatchSize
    if (batchSize < lowerBand || batchSize > upperBand) && poolFreeCapacity > BatchSize {
        batchSizeLastUpdate = time.Now()
        return true, poolFreeCapacity
    }
    // ...
}

The intent: a pool that just grew to 200 free workers should pull batches of ~200, not the configured BatchSize of 10. This keeps the pool fed without over-fetching when the pool shrinks. Updates are rate-limited by TuningCooldownSeconds to avoid churn.

Choosing NumFetchers vs NumWorkers

A common mistake is to set them equal. They serve different roles:

• NumFetchers is about saturating the Redis round-trip. One fetcher per core is usually plenty; 1-4 covers almost all workloads. More fetchers than that just adds XREADGROUP contention.
• NumWorkers is about your task code. If each task does 200ms of CPU work and you want 1000 req/s, you need ~200 workers. The ants pool multiplexes them onto GOMAXPROCS OS threads.

Rule of thumb: NumFetchers small, NumWorkers large. If you enable worker tuning, set NumFetchers to a small constant and let NumWorkers float between MinWorkers and MaxWorkers.