Timing Metrics

TaskRunner collects lightweight per-task execution timings and derives useful aggregate statistics from them: average execution time, predicted queue wait time, and estimated TPS. These power both the worker tuner and your own observability via LongQueueHook.

Collection: hot path is cheap

Every worker, right before returning, records how long the action took:

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

Crucially, storeTiming does not touch Redis synchronously. It pushes a small struct onto a channel:

func (t *TaskRunner) storeTiming(taskName string, x time.Duration) {
    t.tasksTimingBulkWriter.write(timingDto{taskName: taskName, timing: x})
}

The TimingBulkWriter (runner/timing_bulk_writer.go) runs a single background goroutine that buffers writes and flushes them as a pipelined HINCRBY batch once per second:

p.HIncrBy(ctx, metricsHash, timing.taskName+"_sum", timing.timing.Milliseconds())
p.HIncrBy(ctx, metricsHash, timing.taskName+"_count", 1)

This keeps the per-message overhead to a single channel send -- measured at ~1.2 us/op in the benchmarks -- and amortizes Redis round-trips across the whole batch.

Storage layout

All timings live in a single Redis hash, keyed by consumer group:

Redis key	Field	Value
taskrunner:<group>:metrics	<taskName>_sum	cumulative execution time in ms
taskrunner:<group>:metrics	<taskName>_count	number of samples
taskrunner:<group>:metrics	delayed_tasks_schedule_sum	scheduler dispatch time, ms
taskrunner:<group>:metrics	delayed_tasks_schedule_count	scheduler dispatch count

From a single HGETALL (or per-field HGETs) you can reconstruct the average execution time for any task: avg = sum / count.

Aggregation and the Stats struct

The leader periodically (every LongQueueThreshold/2) calls GetTimingStatistics(), which returns:

type Stats struct {
    PerTaskTiming     map[string]int64  // avg exec ms per task name
    PredictedWaitTime float64           // estimated queue wait, ms
    AvgTiming         time.Duration     // overall avg exec time
    AvgScheduleTiming int64             // avg scheduler dispatch time, ms
    TPS               float64           // estimated throughput
}

PredictedWaitTime

This is an estimate, not a measured latency. It is computed from the current state of the queue and pool:

PredictedWaitTime = (avgExecMs * queueLen) / (workers * replicationFactor)

• avgExecMs -- overall average execution time across all tasks.
• queueLen -- XLEN of the stream (from queue.Len()).
• workers -- current workerPool.Cap() (or NumWorkers before Start).
• replicationFactor -- auto-detected count of live instances.

The intuition: "if each of workers * replicationFactor execution slots processes one message in avgExecMs, and there are queueLen messages ahead of you, you wait roughly (queueLen / (workers * replicationFactor)) * avgExecMs."

!!! warning "It's an estimate" PredictedWaitTime assumes the current queue drains at the current execution rate. It does not account for newly arriving messages or for tasks whose runtimes differ wildly from the average. Treat it as a control signal for the tuner, not as an SLA.

The zero-timing fallback

When a queue has items but no timing data yet (e.g. a fresh deploy), avgTiming is 0, which would make PredictedWaitTime 0 -- and the tuner would scale down to MinWorkers even under load. To avoid this, TaskRunner substitutes a 1ms placeholder:

effectiveAvgTiming := float64(avgTiming)
if queueLen > 0 && effectiveAvgTiming == 0 {
    effectiveAvgTiming = 1.0 // avoid dropping workers on empty metrics
}

TPS

tps = (1000 / avgTimingMs) * tpsWorkers * replicationFactor

The theoretical throughput if every worker processed a message back-to-back at the average rate. Like PredictedWaitTime, it's an estimate, not a measurement.

Reset interval

Timing hashes accumulate forever unless reset. The leader resets them on a schedule:

if t.cfg.MetricsResetInterval > 0 {
    if time.Since(t.lastMetricsResetTime) > t.cfg.MetricsResetInterval {
        if t.IsLeader() {
            t.resetTimingMetrics()
        }
        t.lastMetricsResetTime = time.Now()
    }
}

• Default: 24 hours. Set MetricsResetInterval to override.
• Set to 0 to disable resets entirely (timings accumulate for the lifetime of the Redis data).
• On reset, the leader pipelines HSET <task>_sum 0 and <task>_count 0 for every registered task, plus the scheduler fields.

!!! note "Followers update their local reset clock too" Even non-leaders update lastMetricsResetTime on each aggregator tick. This ensures that immediately after a leadership change, the new leader doesn't reset metrics the instant it takes over (which would lose data on every failover).

Surfacing the stats

LongQueueHook

tr := runner.NewTaskRunner(runner.TaskRunnerConfig{
    LongQueueThreshold: 30 * time.Second,
    LongQueueHook: func(s runner.Stats) {
        if s.PredictedWaitTime > 1000 {
            metrics.Gauge("taskrunner.predicted_wait_ms").Set(s.PredictedWaitTime)
            metrics.Gauge("taskrunner.tps").Set(s.TPS)
            log.Warn("queue backing up: predicted wait %.0fms", s.PredictedWaitTime)
        }
    },
}, rdb, queue)

LongQueueHook will be fired if the predicted wait time exceeds the LongQueueThreshold.

GetTimingStatistics

Call it on demand from anywhere (e.g. an HTTP /stats handler):

stats, err := tr.GetTimingStatistics()
if err != nil { /* ... */ }
for task, avgMs := range stats.PerTaskTiming {
    fmt.Printf("%s: avg %dms\n", task, avgMs)
}
fmt.Printf("predicted wait: %.0fms, est TPS: %.0f\n",
    stats.PredictedWaitTime, stats.TPS)

Note: GetTimingStatistics reads from Redis (one HGET per task plus XLEN), so it's cheap but not free. Don't call it in a hot loop.

Queue-level metrics (Redis backend)

The Redis Streams backend (redisstream) additionally tracks message-claim delays in a Redis-backed ring buffer (internal/ring, capacity 1000) and exposes them via GetMetrics:

q := queue.(*redisstream.RedisStreamMessageQueue)
m, err := q.GetMetrics(ctx)
// m["queue_size"], m["message_claim_delay_avg"],
// m["message_claim_delay_std"], m["message_claim_delay_max"],
// m["message_claim_delay_last"], per-group pending/lag, per-consumer idle

These complement the runner-level Stats -- use them together for a full picture.