Worker Tuning¶

TaskRunner can automatically resize its worker pool in response to load. When enabled, the leader periodically recomputes the ideal pool capacity to keep the predicted queue wait time within a target band, clamps it to your configured [MinWorkers, MaxWorkers] range, applies it locally, and publishes the new capacity so followers mirror it.

This is the closest thing TaskRunner has to an "autoscaler". It reacts to the same PredictedWaitTime signal surfaced in Stats.

Enabling tuning¶

Tuning is opt-in. It activates when all four of these are set on TaskRunnerConfig:

Field	Type	Meaning
`MinWorkers`	`int`	Lower bound for the pool. Also the initial capacity on `Start()`.
`MaxWorkers`	`int`	Upper bound.
`DesiredWaitTime`	`time.Duration`	Target predicted wait time.
`DesiredWaitTimeTolerance`	`time.Duration`	Allowed deviation around the target (the "band").

tr := runner.NewTaskRunner(runner.TaskRunnerConfig{
    MinWorkers:               2,
    MaxWorkers:               64,
    DesiredWaitTime:          500 * time.Millisecond,
    DesiredWaitTimeTolerance: 200 * time.Millisecond,
    LongQueueThreshold:       30 * time.Second, // also sets the tuning cadence
    NumFetchers:              4,
}, rdb, queue)

When tuning is disabled (any of the four unset), the pool is fixed at NumWorkers for the life of the process.

The decision function¶

The heart of tuning is a pure function, exhaustively unit-tested in runner/tuning_test.go:

func computeTunedWorkers(
    currentWorkers, replicationFactor, minWorkers, maxWorkers int,
    predictedWaitMs, desiredWaitMs, toleranceMs float64,
) (int, string)

It returns the next capacity and a reason string (visible in logs and useful for debugging). The logic, in order:

flowchart TD
    start([computeTunedWorkers]) --> inv{inputs valid?<br/>current>0, min>0,<br/>max>0, desired>0,<br/>tol>=0}
    inv -- no --> retcur[return current,<br/>invalid_inputs/invalid_wait_params]
    inv -- yes --> zero{predictedWait <= 0?}
    zero -- yes --> dec["reduce 10%, floored at min"]
    dec --> clamp
    zero -- no --> band{lower <= predicted <= upper?}
    band -- yes --> retcur2[return current,<br/>in_tolerance_band]
    band -- no --> dev{"deviation large?<br/>outside 0.5x..1.5x desired"}
    dev -- "no, just above upper" --> twUpper[targetWait = upper<br/>above_upper_boundary]
    dev -- "no, just below lower" --> twLower[targetWait = lower<br/>below_lower_boundary]
    dev -- "yes above" --> twCenter1[targetWait = desired<br/>deviation_large_above_center]
    dev -- "yes below" --> twCenter2[targetWait = desired<br/>deviation_large_below_center]
    twUpper --> math["desiredTotal = current*rep * predicted/target<br/>next = round(desiredTotal / rep)"]
    twLower --> math
    twCenter1 --> math
    twCenter2 --> math
    math --> clamp{"clamp to [min,max]"}
    clamp --> ret[return next, reason]

The math¶

Once a target wait time is chosen, the new capacity is derived from the proportionality baked into PredictedWaitTime:

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

Holding avgExecMs, queueLen, and replicationFactor constant, to move the predicted wait from predicted to target, scale workers by predicted / target:

desiredTotalWorkers = currentWorkers * replicationFactor * (predicted / target)
nextWorkers         = round(desiredTotalWorkers / replicationFactor)

Symmetric rounding (math.Round, not floor/ceil) is deliberate: with a replication factor > 1, asymmetric rounding can oscillate between two adjacent capacities on consecutive ticks. Symmetric rounding keeps it stable near the boundary -- see TestComputeTunedWorkers_ReplicationFactorGreaterThanOne_NoOscillationLikeFlip.

The empty-queue case¶

When predictedWait <= 0 (empty queue or no timing data), tuning scales the pool down by 10% per tick rather than slamming to MinWorkers instantly:

if predictedWaitMs <= 0 {
    newWorkers := int(float64(currentWorkers) * 0.9)
    if currentWorkers > minWorkers && newWorkers > minWorkers {
        return newWorkers, "predicted_wait_zero"
    }
    return minWorkers, "predicted_wait_zero"
}

This avoids a thundering-herd scale-up the moment traffic returns: the pool decays gradually, leaving headroom.

Cooldown¶

Tuning runs on the timing-aggregation loop (every LongQueueThreshold/2, or every minute if LongQueueThreshold == 0). But after an actual capacity change, a cooldown blocks further changes for TuningCooldownSeconds (default 10s):

cooldown := time.Duration(t.cfg.TuningCooldownSeconds) * time.Second
if cooldown > 0 {
    lastUnixNano := t.lastWorkerTuningChangeAt.Load()
    if lastUnixNano != 0 {
        if time.Since(time.Unix(0, lastUnixNano)) < cooldown {
            return
        }
    }
}

This prevents rapid oscillation: the new capacity needs time to affect PredictedWaitTime before we decide to change it again.

Leader publishes, followers apply¶

Only the leader decides the capacity, but every instance needs to honor it -- otherwise a fleet of N instances would each tune independently and overshoot. The mechanism:

Leader computes next, calls workerPool.Tune(next), and publishes:
```
SET taskrunner:<group>:worker_tuning_capacity <next> EX 20
```
Followers (and the leader itself, at the start of each timingAggregator tick) call applyRemoteWorkerTuning(), which reads that key, clamps to local [MinWorkers, MaxWorkers], and calls Tune(remote) if it differs from current.

The published value has a 20s TTL, so if the leader dies the followers keep the last-known capacity until a new leader is elected and starts publishing again.

Reason strings¶

computeTunedWorkers returns a reason for every decision. You'll see it in the leader's log line on each capacity change:

level=info msg="worker pool tuned" old_total=20 new_total=35 \
    old_predicted_wait_ms=248 reason=above_upper_boundary

Reason	Meaning
`in_tolerance_band`	Predicted wait already within `[desired-tol, desired+tol]`; no change.
`above_upper_boundary`	Predicted slightly above band; target = upper boundary.
`below_lower_boundary`	Predicted slightly below band; target = lower boundary.
`deviation_large_above_center`	Predicted > 1.5x desired; target = desired (center).
`deviation_large_below_center`	Predicted < 0.5x desired; target = desired (center).
`predicted_wait_zero`	Empty queue / no data; decay by 10% toward min.
`clamp_to_max` / implicit min	Result exceeded bounds; clamped.
`invalid_inputs` / `invalid_wait_params`	Bad config; passthrough current.

Practical tuning advice¶

Start with a realistic wait target. DesiredWaitTime is the maximum queueing delay you can tolerate in normal operation. For user-facing work this might be 100-500ms; for batch it could be seconds.
Use a non-zero tolerance (20-50% of DesiredWaitTime). A zero tolerance means the pool chases a single point and never settles.
Clamp sensibly. MinWorkers keeps latency bounded during low load (don't scale to zero). MaxWorkers protects your CPU/memory budget during spikes.
Mind the cadence. Tuning runs every LongQueueThreshold/2. A 30s threshold means tuning checks every 15s. Faster cadence = faster reaction but more churn.
Watch the LongQueueHook. It receives Stats on the same cadence; log PredictedWaitTime and TPS to validate the pool is staying in band.
Set TuningCooldownSeconds to a few multiples of your typical task runtime, so a capacity change has time to take effect before re-evaluation.

What about replication factor?¶

PredictedWaitTime divides by workers * replicationFactor, where replicationFactor is auto-detected from the number of registered replication keys (taskrunner:replication:*, refreshed every 10s). The deprecated TaskRunnerConfig.ReplicationFactor field is kept only for backward compatibility. You no longer need to set it.