Delayed Scheduling

Sometimes you need a task to run later -- "send a reminder 24 hours from now", "expire this session at its TTL", "retry after a backoff window". TaskRunner supports this via a Redis sorted set (ZSET) polled by the leader.

The model

flowchart LR
    P[Producer] -->|"DispatchDelayed(d=10s)"| Z[(ZSET<br/>score=run_at_unix<br/>member=DelayedTask JSON)]
    Z -.polled every 5s.-> L[Leader]
    L -->|ZRANGEBYSCORE 0..now| Z
    L -->|for each due job: Dispatch| S[(Stream)]
    L -->|ZREM after dispatch| Z
    S --> W[Workers]

1. DispatchDelayed(ctx, taskName, payload, d) writes a DelayedTask JSON into the ZSET with score = time.Now().Add(d).Unix().
2. The leader (and only the leader) runs StartDelayedSchedule, which ticks every 5 seconds.
3. On each tick it pages through ZSET members whose score is <= now, and for each one it calls Dispatch (which XADDs to the stream) then ZREMs the ZSET entry.
4. Workers process the now-due message exactly like any other.

Minimum effective delay

Because the scheduler ticks every 5s, any delay shorter than 5s is rejected:

if d.Seconds() < 5 {
    return ErrDurationIsSmallerThanCheckCycle
}

In practice, expect a 1-5 second error on top of your requested delay: a job scheduled for now+10s may run anywhere from now+10s to now+15s depending on where in the tick cycle it lands. For sub-second precision, use the regular queue.

The API

DispatchDelayed -- relative

// Run "cleanup" with payload "temp" roughly 5 minutes from now.
err := tr.DispatchDelayed(ctx, "cleanup", "temp", 5*time.Minute)

ScheduleFor -- absolute

// Run "invoice" at a specific wall-clock time.
when := time.Date(2026, 1, 1, 9, 0, 0, 0, time.UTC)
err := tr.ScheduleFor(ctx, "invoice", invoiceID, when)

Both return ErrTaskNotFound if the task isn't registered, or ErrDurationIsSmallerThanCheckCycle if the requested time is less than 5s in the future.

Starting the scheduler

You must start the scheduler explicitly alongside the worker loop. Only the leader will actually dispatch due jobs, but every instance should call it -- that way leadership failover is seamless (the new leader is already running the loop).

ctx, cancel := context.WithCancel(context.Background())
defer cancel()

go func() { _ = tr.Start(ctx) }()                  // workers + election
go func() { _ = tr.StartDelayedSchedule(ctx, 1000) }() // scheduler; 1000 = batch size

StartDelayedSchedule(ctx, batchSize) blocks until ctx is canceled. The batchSize controls how many due jobs are dispatched per page and the size of the scheduler's internal dispatch pool.

Internals: paging through due jobs

When a backlog of delayed jobs all come due at once (e.g. a scheduled maintenance window), the scheduler pages through them rather than loading the whole ZSET into memory:

pages := math.Ceil(float64(count) / float64(batchSize))
for page := 0; page < int(pages); page++ {
    tasks, _ := t.redisClient.ZRangeByScoreWithScores(ctx, key, &redis.ZRangeBy{
        Min: "-inf", Max: now,
        Count:  batchSize,
        Offset: int64(page * batchSize),
    }).Result()
    for _, payload := range tasks {
        // unmarshal, Dispatch, ZREM
        workerPool.Invoke(map[string]any{"task": task, "payload": payload})
    }
}

Dispatching happens inside a bounded ants pool (batchSize workers) so a large backlog doesn't spawn unbounded goroutines. The total dispatch time per tick is recorded as a timing sample under the taskrunner:<group>:delayed_tasks_schedule metrics key.

Error handling on dispatch

When the scheduler calls Dispatch for a due job, dispatch can fail. Two cases are handled explicitly:

• ErrTaskAlreadyDispatched -- silently swallowed. This happens if the job was a unique task whose lock is still held; it'll be retried on a future tick (or expire naturally). The ZSET entry is not removed, so it remains due.
• Any other error -- captured to the error channel. The ZSET entry is also left in place, so the next tick will retry it.

A successful dispatch removes the ZSET entry with ZREM keyed on the original JSON member.

At-least-once, even for delayed jobs

Because the ZREM happens after the XADD, a crash between the two means the job could be dispatched twice on recovery. Your task Action should be idempotent if this matters. This is the standard at-least-once tradeoff and applies to the regular queue too.

Redis key layout

Key	Type	Contents
taskrunner:<group>:delayed_tasks	ZSET	score = run_at_unix, member = DelayedTask JSON
taskrunner:<group>:metrics (fields delayed_tasks_schedule_sum/_count)	HASH	scheduling latency samples

See Redis Key Layout for the complete map.

When to use delayed vs. regular retry

Delayed scheduling is not the same as the retry-on-failure mechanism. The two are orthogonal:

• Retry (via MaxRetry + the pending reclaim path) handles failures -- your action returned an error or the worker crashed. The message stays in the PEL and is redelivered after reClaimDelay.
• Delayed handles intentional future execution -- you know at dispatch time that the job should run later.

For exponential backoff on failure, either return an error and let the reclaim/retry cycle handle it, or have your Action dispatch a new delayed job for the next attempt with a computed backoff.