Skip to content

Architecture

TaskRunner is a library, not a daemon. Each instance of your application embeds it and shares a Redis instance for coordination. There is no peer-to-peer networking -- every instance communicates solely through Redis.

Component map

flowchart TB
    subgraph App["Your application process"]
        TR["TaskRunner instance"]
        subgraph FE["Fetchers (NumFetchers)"]
            F1["fetcher 1"]
            F2["fetcher N"]
        end
        WP["ants worker pool (NumWorkers)"]
        TA["timing aggregator loop"]
        EL["leader election loop"]
        RP["replication registrar"]
        DS["delayed scheduler (optional)"]
        TBW["TimingBulkWriter"]
        TR --> FE
        TR --> WP
        TR --> TA
        TR --> EL
        TR --> RP
        TR --> DS
        TR --> TBW
    end

    subgraph Redis
        RS[(Redis Stream)]
        CG["Consumer Group + PEL"]
        DL[(Delayed ZSET)]
        LK["leader:group"]
        HM["metrics hashes"]
        RF["replication keys"]
        UK["unique locks"]
    end

    F1 -->|XREADGROUP| RS
    RS --- CG
    F1 -->|XCLAIM pending| CG
    F1 -->|Invoke| WP
    WP -->|XACK / XDEL| RS
    WP -->|HINCRBY timings| HM
    WP -->|release redsync lock| UK
    TA -->|HGET timings, XLEN| HM
    EL <-->|SET NX / PEXPIRE Lua| LK
    RP -->|SET TTL| RF
    DS -->|ZRANGEBYSCORE -> XADD| DL
    DS --> RS

The fetcher / worker split

Fetching and execution are deliberately decoupled, and they scale independently:

  • Fetchers (runner/fetcher.go) read batches of messages from the queue. Each fetcher issues an XREADGROUP with Count = BatchSize and Block = BlockDuration, then hands each message to the worker pool via workerPool.Invoke(...). Fetchers never run your task code -- they only move messages.
  • Workers (runner/worker.go) run inside an ants PoolWithFunc. Each invocation unmarshals the message, looks up the task, runs its Action, and acks on success.

This split matters because the two have very different cost profiles. Fetching is I/O-bound (one Redis round-trip per batch) and should usually have a small fanout (1-4 fetchers). Execution is CPU/I/O-bound on your code and may need tens or hundreds of workers. Decoupling them lets the runner move a large batch into the pool in one call and let the pool drain it concurrently.

Adaptive batch sizing

When worker tuning is enabled, each fetcher also tracks the pool's free capacity. If free capacity drifts by more than BatchSize outside the [free - BatchSize, free + BatchSize] band for longer than TuningCooldownSeconds, the fetcher adjusts its Count to match free capacity -- so a large pool pulls larger batches and a small pool doesn't over-fetch.

The message lifecycle

sequenceDiagram
    participant P as Producer
    participant Q as Redis Stream
    participant F as Fetcher
    participant W as Worker
    participant PEL as Pending Entries List

    P->>Q: Dispatch -> XADD (payload)
    F->>Q: XREADGROUP group=... consumer=... count=N
    Q-->>F: messages (and adds to PEL)
    F->>W: workerPool.Invoke(msg)
    W->>W: json.Unmarshal -> lookup task
    alt action fast (< ReservationTimeout)
        W->>W: task.Action(payload)
        W->>Q: XACK + XDEL (deleteOnAck)
    else action slow
        loop every ReservationTimeout
            W->>Q: XCLAIM (heartbeat) - resets idle
        end
        W->>W: task.Action(payload)
        W->>Q: XACK + XDEL
    end
    Note over W,PEL: on crash, message stays in PEL<br/>after reClaimDelay another consumer XCLAIMs it

Visibility timeout and heartbeats

A Redis Stream entry handed to a consumer via XREADGROUP enters the Pending Entries List (PEL) for that group. It stays there until acked. If the consumer never acks (because it crashed, OOMed, or lost network), the message is not lost -- another consumer can reclaim it with XCLAIM once it has been idle long enough.

The risk is that a legitimately slow task gets reclaimed while it is still running, causing double-processing. TaskRunner solves this with heartbeats:

  • Each task has a ReservationTimeout. While the action runs, the worker loops on a select. Every ReservationTimeout, if the queue backend requires heartbeats (Redis Streams does -- RequireHeartHeartBeat() == true), the worker re-claims the message with XCLAIM, resetting its idle timer.
  • This extends the visibility window for as long as the task genuinely runs.
  • If the worker process dies, no more heartbeats are sent, idle time climbs past reClaimDelay, and another consumer reclaims it.

!!! warning "Don't thread the cancel context into the action" The worker deliberately uses context.Background() for the per-message action, not the Start(ctx) cancel context. This is intentional: tying cancellation to process shutdown would cause in-flight tasks to be aborted and requeued, which interacts badly with the visibility-timeout model. Let tasks complete; shutdown drains gracefully.

Pending reclaims and retries

On each Receive, the queue backend first checks for pending messages that have been idle longer than reClaimDelay. If any exist, it XCLAIMs them and hands them to the worker. The reclaimed message's retry count is incremented; when it exceeds Task.MaxRetry, the FailedTaskHandler is invoked and the message is no longer retried.

On Redis 6.2+, the reclaim uses XPENDING's IDLE filter server-side for efficiency. On older versions it falls back to fetching the full PEL and filtering client-side.

Leader election

Only one instance at a time may be the leader. Leadership is needed for the "global" chores:

  • Running the delayed scheduler (StartDelayedSchedule).
  • Computing and resetting timing metrics.
  • Publishing the worker-pool capacity (so followers can mirror it).

See Leader Election for the full mechanism. The short version: a Lua-scripted atomic SET NX PX wins the lease, PEXPIRE (guarded by a value check) renews it, and DEL (also value-guarded) releases it. Election is cooperative -- there is no fencing token; correctness relies on the lease TTL expiring before a split-brain instance can act.

Timing and metrics

Every worker records how long its action took:

defer func() {
    t.storeTiming(task.Name, time.Since(timeStart))
}()

storeTiming does not write to Redis synchronously. It pushes a timingDto{taskName, duration} onto an unbuffered channel consumed by the TimingBulkWriter, which flushes a pipelined HINCRBY batch every second:

Redis key Field Value
taskrunner:<group>:metrics <task>_sum total ms
taskrunner:<group>:metrics <task>_count number of samples

From these the leader derives PerTaskTiming averages and a PredictedWaitTime estimate. See Timing Metrics.

Concurrency model

State Guard
TaskRunner.status (Init -> Starting -> Started -> Stopped) atomic.Uint64 CAS
isLeader atomic.Bool
inFlight, processed, fails counters atomic.Int64
Task registry safemap.SafeMap (embedded sync.RWMutex)
Replication-factor cache TTL cache.Cache (sync.RWMutex + janitor)
Last-tuning-change timestamp atomic.Int64 (unix nano)

Start() is guarded by a sync.Mutex (startLock) so that concurrent Start() calls deterministically return ErrTaskRunnerAlreadyStarted / ErrRaceOccuredOnStart rather than racing.

Error channel

TaskRunner does not return per-message errors from Start. Instead, runtime errors (fetch failures, ack failures, election errors, lock-release failures) are pushed onto a buffered errorChannel (capacity 1). You drain it via TaskRunner.ErrorChannel():

go func() {
    for err := range tr.ErrorChannel() {
        log.Printf("taskrunner: %v", err)
    }
}()

!!! caution "Drain the error channel" Because the channel has capacity 1, a captureError call will block if the channel is full and nobody is draining it. In a long-running service you should always run a drainer goroutine.