Skip to content

Crash Recovery

Under exactly-once, the output topic is never duplicated — but the local state (windowed accumulators, keyed values held in RocksDB) can briefly lag the output after a hard crash. This page explains the shape of that gap and the knobs that control how a worker recovers.

It applies only to exactly-once subscriptions: at-least-once handlers re-process the last batch on recovery, so there is no state gap (and no knobs).

State gap on crash recovery

The exactly-once contract is outputs are never duplicated. State — the windowed accumulators and key-keyed values held in your local RocksDB — may lag the output topic by a small amount after a hard crash. How small depends on how the worker stopped:

Scenario Output topic Local state
Steady state, no crash exactly-once up-to-date with output
Graceful stop (deploy, scale, rebalance) exactly-once up-to-date — the worker drains and snapshots before handing off
Hard crash (host failure), local RocksDB intact exactly-once lags by up to one batch (the in-flight one when the crash hit)
Hard crash, recovering from a stale snapshot exactly-once lags by however many events the snapshot is behind the broker

In every row, the output topic stays exactly-once. The only thing that varies is how stale the local state is when the worker resumes.

Why state can lag

The Kafka transaction covers (input offsets, output messages). RocksDB is written to disk just after — typically within milliseconds, but the two writes are not a single atomic operation. If a crash happens between them, the broker thinks offsets up to N have been processed, but local RocksDB thinks it only processed up to M < N. On restart, Turbine trusts the broker, seeks forward to N+1, and resumes. The N−M events between the two are not re-emitted (no duplicates on output) but their state mutations are skipped — so a counter, a windowed accumulator, or a keyed value that should have observed those events doesn't.

For most analytics workloads this is the right trade-off: a counter that's off by a small bounded amount is easier to live with than duplicate output messages that downstream consumers have to dedup.

The gap is observable via turbine_eos_state_gap_events{worker} — a counter incremented at boot by the number of skipped events. Your monitoring stack owns the alert threshold; no code change needed to tune it.

Tuning crash recovery

For an EOS subscription, two optional parameters let you choose what happens when a non-zero state gap is detected at boot.

Default: accept the gap and keep going

@app.subscribe(
    kafka.topic("events"),
    output=kafka.topic("enriched"),
    processing_guarantee="exactly_once",
    # on_crash_recovery="accept" is the default — no need to set it
)
def enrich(batch): ...

The worker boots, seeks forward to the broker's committed offset, increments turbine_eos_state_gap_events by the gap size, and resumes. Outputs are never duplicated. State is stale for the skipped events.

This is right for most analytics: aggregations that tolerate a small undercount, enrichments where the missed window is acceptable, anything where "duplicate outputs would be worse than slightly-off state".

Rebuild state instead of accepting the gap

@app.subscribe(
    kafka.topic("events"),
    output=kafka.topic("enriched"),
    processing_guarantee="exactly_once",
    on_crash_recovery="replay",   # ← rebuild state silently from the broker
)
def enrich(batch): ...

The worker boots, sees the gap, and silently re-consumes the missing events to rebuild state. During that replay phase, every Kafka-output side effect is suppressed — no produces, no transactions — so the output topic stays never-twice. Once local_offset catches up to the broker, the worker flips to LIVE and outputs resume.

Use replay when:

  • Your state is mission-critical and "slightly off" isn't acceptable (e.g., financial counters, security event aggregates).
  • Your Kafka retention covers the gap. The replay reads from the broker; if the messages are no longer there, replay can't reconstruct anything.
  • Your handler is pure with respect to external side effects (or already idempotent). The replay re-runs your processor on the gap, and produce_batch is the only path Turbine can suppress. If your handler also writes to a non-Kafka database or calls an HTTP endpoint, those will fire again during replay. See Idempotency contract.

Circuit-breaker: refuse to boot on anomalously large gaps

@app.subscribe(
    kafka.topic("events"),
    output=kafka.topic("enriched"),
    processing_guarantee="exactly_once",
    halt_if_gap_exceeds=10_000_000,   # ← 10 M events ⇒ refuse to start
)
def enrich(batch): ...

When the gap exceeds the threshold at boot, the worker exits with a fatal error and the metric turbine_eos_state_gap_halt_total{worker} ticks. Use this to catch "the snapshot pipeline is broken" or "we're recovering from a far-too-old snapshot" — situations where you'd rather page an operator than let the worker silently absorb (or replay) a huge backlog.

halt_if_gap_exceeds works with both accept and replay. It runs first; if the gap exceeds the threshold, no recovery is attempted.

Combining the knobs

The three options compose freely:

# Production EOS: rebuild state if needed, but never start with a >10M gap
@app.subscribe(
    kafka.topic("events"),
    output=kafka.topic("enriched"),
    processing_guarantee="exactly_once",
    on_crash_recovery="replay",
    halt_if_gap_exceeds=10_000_000,
)
def enrich(batch): ...

There's deliberately no alert_if_gap_exceeds parameter — the turbine_eos_state_gap_events{worker} counter already lets your monitoring stack express any alert you want:

# Alert if any worker observes >10k gap in the last 5 minutes
increase(turbine_eos_state_gap_events[5m]) > 10000

Keeping the threshold in your monitoring config (instead of in the subscribe call) means you can tune it without a redeploy.