Configuration Reference¶

This is the complete reference for runner.TaskRunnerConfig and the redisstream queue options. For conceptual background on what each knob does, follow the links into the Concepts section.

TaskRunnerConfig¶

All fields live on runner.TaskRunnerConfig, passed to NewTaskRunner:

tr := runner.NewTaskRunner(runner.TaskRunnerConfig{
    // ...fields...
}, rdb, queue)

Identity & topology¶

Field	Type	Default	Description
`Host`	`string`	OS hostname	Per-instance identity. Used in consumer names (`<group>_<host>_<fetcherID>`) and the replication key. Falls back to a UUID if hostname lookup fails.
`ConsumerGroup`	`string`	(required)	Redis Streams consumer group. All instances that should share a workload MUST use the same value. Different groups each get a full copy of the stream.
`ConsumersPrefix`	`string`	(advisory)	Prefix for consumer names. Largely cosmetic; useful for distinguishing fleets in `XINFO CONSUMERS`.

Fetching & execution¶

Field	Type	Default	Description
`BatchSize`	`int`	(required if NumFetchers unset)	Messages read per `XREADGROUP` round-trip per fetcher. Larger = fewer round-trips, more memory per fetch. Typical: 10-100.
`NumFetchers`	`int`	= `BatchSize`	Concurrent fetcher goroutines. See Fetchers & Workers. Usually small (1-4).
`NumWorkers`	`int`	(required when tuning disabled)	Fixed worker-pool capacity. Ignored as a fixed value when tuning is enabled (becomes the fallback before `Start`).
`BlockDuration`	`time.Duration`	5s	How long `XREADGROUP` blocks waiting for new messages. Longer = lower idle CPU, slower reaction to shutdown.

Worker tuning (optional autoscaling)¶

Enable by setting all four. See Worker Tuning for the algorithm.

Field	Type	Default	Description
`MinWorkers`	`int`	(unset)	Lower bound for the pool. Also the initial capacity on `Start`.
`MaxWorkers`	`int`	(unset)	Upper bound.
`DesiredWaitTime`	`time.Duration`	(unset)	Target predicted wait time.
`DesiredWaitTimeTolerance`	`time.Duration`	(unset)	Allowed deviation (the band half-width).
`TuningCooldownSeconds`	`int`	10	Min seconds between actual capacity changes.

Reliability & callbacks¶

Field	Type	Default	Description
`FailedTaskHandler`	`FailedTaskHandler`	nil	Called when a task exhausts `MaxRetry`. If it returns nil, the message is acked (dropped); if it returns an error, the message is left for reclaim. See Error Handling.
`LongQueueHook`	`LongQueueHook`	nil	Called on every timing-aggregation tick with the current `Stats`. Despite the name, it fires regardless of queue length -- use it as a periodic stats callback.
`LongQueueThreshold`	`time.Duration`	0	Sets the aggregation/tuning cadence (`threshold/2`) and the threshold above which the queue is considered "long". 0 => cadence defaults to 1 minute.

Metrics¶

Field	Type	Default	Description
`MetricsResetInterval`	`time.Duration`	24h	How often the leader resets timing hashes. Set to 0 to disable resets. See Timing Metrics.

Deprecated¶

Field	Status	Note
`ReplicationFactor`	deprecated	Replication factor is now auto-detected from heartbeated keys. Kept for backward compat; ignored at runtime.

Task definition¶

Tasks are registered with tr.RegisterTask(&runner.Task{...}):

type Task struct {
    Name              string
    MaxRetry          int
    ReservationTimeout time.Duration
    Action            TaskAction    // func(ctx context.Context, payload any) error
    Unique            bool
    UniqueKey         UniqueKeyFunc // func(payload any) string
    UniqueFor         int64          // seconds
}

Field	Type	Default	Description
`Name`	`string`	(required)	Registry key. Must be unique per runner.
`MaxRetry`	`int`	1 (if 0)	Max delivery attempts before `FailedTaskHandler`.
`ReservationTimeout`	`time.Duration`	(set for long tasks)	Heartbeat interval. Set to a fraction of worst-case runtime.
`Action`	`TaskAction`	(required)	Your handler.
`Unique`	`bool`	false	Enables unique jobs.
`UniqueKey`	`UniqueKeyFunc`	nil	Per-payload scope. Nil => task-name-global uniqueness.
`UniqueFor`	`int64`	(required if Unique)	Lock TTL in seconds.

Redis Streams queue options¶

The shipped backend is constructed with an options pattern:

queue := redisstream.NewRedisStreamMessageQueueWithOptions(
    rdb,
    redisstream.WithPrefix("myapp"),
    redisstream.WithQueue("tasks"),
    // ...
)

There is also a positional constructor for backward compat:

queue := redisstream.NewRedisStreamMessageQueue(
    rdb, "myapp", "tasks", 30*time.Second, true)

Options¶

Option	Default	Description
`WithPrefix(prefix string)`	`"default"`	First segment of the stream key (`<prefix>:<queue>`). Namespace per environment/app.
`WithQueue(queue string)`	`"queue"`	Second segment of the stream key.
`WithReClaimDelay(d time.Duration)`	5 min	Visibility timeout. A pending message idle longer than this becomes eligible for `XCLAIM` by another consumer. Must be greater than your longest task.
`WithDeleteOnAck(bool)`	`true`	Whether to `XDEL` after `XACK`. `true` keeps the stream from growing; `false` retains history.
`WithRedisVersion(version string)`	auto-detected	Override the detected Redis version. Use to force the XPENDING `IDLE` filter path on or off.

Stream key¶

The Redis stream name is <prefix>:<queue>. The metrics ring key is <prefix>:metrics:<queue>. Choose prefix to avoid collisions across environments sharing a Redis (e.g. prod, staging).

Sensible starting points¶

Low-throughput, simple (a few tasks/sec)¶

TaskRunnerConfig{
    ConsumerGroup: "myapp",
    BatchSize:     10,
    NumFetchers:   1,
    NumWorkers:    8,
}

High-throughput batch (thousands/sec)¶

TaskRunnerConfig{
    ConsumerGroup:   "myapp",
    BatchSize:       100,
    NumFetchers:     4,
    MinWorkers:      10,
    MaxWorkers:      500,
    DesiredWaitTime: 100 * time.Millisecond,
    DesiredWaitTimeTolerance: 50 * time.Millisecond,
    LongQueueThreshold: 10 * time.Second,
}

Mixed short + long tasks¶

Give long tasks their own ReservationTimeout and keep ReClaimDelay well above it:

queue := redisstream.NewRedisStreamMessageQueueWithOptions(rdb,
    redisstream.WithReClaimDelay(10*time.Minute), // generous visibility window
    redisstream.WithDeleteOnAck(true),
)

tr.RegisterTask(&runner.Task{
    Name: "render-video",
    ReservationTimeout: 30 * time.Second, // heartbeat every 30s during a 10-min render
    MaxRetry: 2,
    Action:   render,
})

Common pitfalls¶

NumFetchers == 0 after construction defaults to BatchSize, which can be accidentally huge. Set both explicitly.
ReClaimDelay shorter than task runtime causes double-processing. Keep it comfortably above your worst case (including retries).
UniqueFor shorter than task runtime lets duplicates slip in. See Unique Jobs.
Forgetting LongQueueHook / not draining ErrorChannel() leaves errors buffered; since the channel has capacity 1, a full channel will block the next captureError and stall a worker. Always run a drainer.
DesiredWaitTimeTolerance == 0 makes the tuner chase a single point and never settle. Use 20-50% of DesiredWaitTime.