Finding (critic #5)

An OnEvent side-effect consumer whose handler permanently fails retried head-of-line forever on a flat 10ms backoff. That wedges the consumer and pins the Compactor floor at the poison record's offset, so the event log grows unboundedly — a single bad record can take down a pod's disk over time.

Fix

• Exponential backoff + full jitter (default 10ms → 30s) replaces the flat 10ms retry. Per-pod attempt counter, reset to 0 on any advance.
• Block-forever stays the default. WithMaxAttempts(0) (the default) preserves today's zero-data-loss semantics: a permanently-failing handler keeps the record head-of-line and never drops the side effect. Skipping a poison record is strictly opt-in — dropping a side effect by default is the wrong default.
• Loud while blocking. Each retry emits a via.consumer.stuck gauge (the attempt count); a one-shot WARN fires once attempts cross a threshold, so the floor-pin is observable rather than silent.

New options (variadic — existing 2-arg call sites still compile)

• WithMaxAttempts(n int) — after n handler-error attempts on a record, treat it as poison: emit via.consumer.poisoned + WARN, dead-letter it (if a hook is set) or drop it, then commit/advance past it. Un-wedges the consumer and unpins the floor.
• WithRetryBackoff(base, max time.Duration) — backoff bounds (default 10ms → 30s).
• WithDeadLetter(fn) — invoked just before a record is poisoned. Returns nil → advance; returns an error → do not advance, keep retrying (a record the operator opted to DLQ is never lost when the sink is down).

deliver() was refactored to return a three-way outcome (advance / retry / block) so a forward-incompatible record always blocks and is never counted toward poison; an undecodable record still skips via the decode path (via.consumer.undecodable), distinct from the poison path.

Tests (red→green, internal pkg, -race)

• default maxAttempts=0: always-erroring handler never advances; via.consumer.error + via.consumer.stuck keep firing; never poisons.
• WithMaxAttempts(N): advances past the record after N attempts, emits via.consumer.poisoned, then delivers the next record (proves un-wedged).
• errors-then-succeeds: advances, attempts reset (no premature poison on a later record).
• WithDeadLetter returning nil: hook gets correct key/off/data/cause, then advances + via.consumer.poisoned.
• WithDeadLetter returning err: does not advance (stays retrying).
• forward-incompatible still blocks under WithMaxAttempts (never poisoned); undecodable still skips (not poisoned).

Docs

docs/production.md metrics catalogue: added via.consumer.poisoned and via.consumer.stuck rows, amended the via.consumer.error note (bounded-retry/backoff + WithMaxAttempts opt-in skip). docs/distributed-state.md OnEvent section documents the new options.

ci-check.sh passes (gofmt, vet, golangci v2.12.2, govulncheck, -race, alloc benches).