Error Handling¶

TaskRunner distinguishes three classes of failure: dispatch errors (returned from Dispatch), task execution errors (from your Action), and runtime/infrastructure errors (pushed to the error channel). This page covers each and how to handle them robustly.

1. Dispatch errors¶

Dispatch, DispatchDelayed, and ScheduleFor return errors directly. Handle them at the call site.

Error	Constant	When	Typical handling
Task not registered	`runner.ErrTaskNotFound`	The task name isn't in the registry.	Bug -- fix the registration.
Unique lock not acquired	`runner.ErrTaskAlreadyDispatched`	A unique job is already in flight for this key.	Expected -- log and move on; do not retry blindly.
`UniqueFor` missing	`runner.ErrUniqueForIsRequired`	`Task.Unique == true` but `UniqueFor == 0`.	Bug -- set `UniqueFor`.
Delay too short	`runner.ErrDurationIsSmallerThanCheckCycle`	`DispatchDelayed` / `ScheduleFor` with delay < 5s.	Use the regular queue, or bump the delay.
Marshal failure	(wraps `encoding/json`)	The payload can't be JSON-marshalled.	Make the payload serializable.
Queue add failure	(wraps Redis error)	`XADD` failed (Redis down, OOM, etc.).	Retry with backoff; alert.
Already started	`runner.ErrTaskRunnerAlreadyStarted`	`Start()` called twice on the same runner.	Bug -- call `Start` once.

Use errors.Is for sentinel comparison:

err := tr.Dispatch(ctx, "sendEmail", orderID)
switch {
case errors.Is(err, runner.ErrTaskAlreadyDispatched):
    log.Printf("email for %s already queued", orderID) // fine
case errors.Is(err, runner.ErrTaskNotFound):
    return fmt.Errorf("bug: sendEmail not registered: %w", err)
case err != nil:
    return fmt.Errorf("dispatch failed: %w", err) // retry upstream
}

2. Task execution errors¶

When your Action returns an error, the worker:

Increments the failure counter.
Pushes a TaskExecutionError onto the error channel (wrapped with the task name).
Does not ack -- the message stays in the PEL and will be redelivered after reClaimDelay, incrementing its retry count.

This gives you at-least-once retry for free. The redelivery cycle continues until the message's receive count exceeds Task.MaxRetry, at which point the FailedTaskHandler runs.

MaxRetry semantics¶

tr.RegisterTask(&runner.Task{
    Name:     "charge-card",
    MaxRetry: 5,   // delivered up to 6 times total (1 initial + 5 retries)
    Action:   chargeCard,
})

MaxRetry == 0 is treated as 1 (one delivery, no retries). The check is if task.MaxRetry < int(m.GetReceiveCount()), so a message that has been delivered MaxRetry+1 times is considered exhausted.

Receive count on Redis Streams

A freshly XREADGROUP-ed message reports a receive count of 1. Reclaimed (pending) messages carry their accumulated retry count from the PEL.

FailedTaskHandler¶

When retries are exhausted, the configured handler runs:

type FailedTaskHandler func(ctx context.Context, task TaskMessage, err error) error

tr := runner.NewTaskRunner(runner.TaskRunnerConfig{
    FailedTaskHandler: func(ctx context.Context, msg runner.TaskMessage, err error) error {
        // Persist to a dead-letter store, emit an alert, etc.
        if saveErr := dlq.Save(ctx, msg, err); saveErr != nil {
            return saveErr // returning non-nil leaves the message for reclaim
        }
        return nil // returning nil acks the message (drops it from the queue)
    },
}, rdb, queue)

The return value controls the message's fate:

Return	Effect
`nil`	The worker acks (and deletes, if `deleteOnAck`) the message. It will not be retried.
non-nil error	The message is not acked. It stays in the PEL and the handler will run again on the next reclaim. Use this to retry the DLQ write itself.

If FailedTaskHandler is nil, exhausted messages are simply dropped (no ack is attempted either -- they remain in the PEL until the stream is trimmed or the group is deleted). Always set a handler in production.

Invalid payloads¶

If a message's JSON can't be unmarshalled into a TaskMessage, the worker treats it as a terminal failure (retrying a malformed payload is pointless):

if err := json.Unmarshal([]byte(messagePayload), &taskMessage); err != nil {
    failed()
    taskMessage.Payload = m.Payload
    handleFailedTask(ctx, taskMessage, ErrInvalidTaskPayload)
    return
}

ErrInvalidTaskPayload is passed to FailedTaskHandler.

3. Runtime / infrastructure errors¶

These are not tied to a specific task; they're emitted by the engine itself:

Fetch failures (XREADGROUP errors).
Ack/delete failures (XACK/XDEL errors).
Heartbeat failures (during long tasks).
Lock-release failures.
Election errors (Redis errors during set/renew).
Scheduler dispatch errors.

They are pushed onto TaskRunner.ErrorChannel() -- a buffered channel of capacity 1. Drain it in its own goroutine for the lifetime of the runner:

go func() {
    for err := range tr.ErrorChannel() {
        log.WithError(err).Error("taskrunner")
        metrics.Counter("taskrunner.errors").Inc()
    }
    // channel closed -> runner has shut down
}()

Always drain the error channel

captureError does a blocking send onto a capacity-1 channel. If nobody drains it and it fills up, the next captureError will block the worker or fetcher that called it, stalling progress. A drainer goroutine is mandatory in production, not optional.

Error types reference¶

Type / constant	Package	Purpose
`TaskRunnerError` (string)	`runner`	Base type for runner sentinels. Implements `error`.
`ErrTaskNotFound`	`runner`	Dispatch / delayed dispatch of unknown task.
`ErrTaskAlreadyDispatched`	`runner`	Unique lock contention.
`ErrTaskMaxRetryExceed`	`runner`	Passed to `FailedTaskHandler` on exhaustion.
`ErrInvalidTaskPayload`	`runner`	Bad message JSON.
`ErrUniqueForIsRequired`	`runner`	Unique task missing `UniqueFor`.
`ErrDurationIsSmallerThanCheckCycle`	`runner`	Delay < 5s.
`ErrTaskRunnerAlreadyStarted`	`runner`	Double `Start()`.
`ErrRaceOccuredOnStart`	`runner`	CAS race during `Start()`.
`ErrTaskUnlockFailed`	`runner`	Lock release returned false.
`TaskExecutionError`	`runner`	Wraps an `Action` error with the task name. Use `GetTaskName()` / `GetError()`.
`QueueError` (string)	`contracts`	Base type for queue sentinels.
`ErrNoNewMessage`	`contracts`	`Receive` found nothing (often benign).
`MessageNotFoundError`	`contracts`	Operation on a missing message.

Use errors.Is for sentinels and errors.As for typed errors:

var execErr runner.TaskExecutionError
if errors.As(err, &execErr) {
    log.Printf("task %s failed: %v", execErr.GetTaskName(), execErr.GetError())
}

Designing idempotent tasks¶

Because TaskRunner is at-least-once, a message can be processed more than once (worker crash between action-completion and ack, scheduler duplicate dispatch, pending reclaim after a slow task). Design Actions to be safe under retry:

Use a deterministic ID in the payload and check a "have I already done this?" store (DB row, Redis key) before doing the work.
Make side effects commutative where possible (increment a counter, not "set to current+1").
For emails / webhooks / payments, attach an idempotency key and let the downstream dedupe.

Action: func(ctx context.Context, payload any) error {
    id := payload.(string)
    done, err := redis.SetArgs(ctx, "done:"+id, 1, redis.SetArgs{
        NX: true, TTL: 24 * time.Hour,
    }).Result()
    if err != nil { return err }
    if done == nil { return nil } // already processed
    return doTheWork(ctx, id)
},