Crash Recovery

Under exactly-once, an output produced from a record is never duplicated — but a hard crash can leave the local state (windowed accumulators, keyed values held in RocksDB) behind the output. By default the worker rebuilds that state by replaying the missing input before it resumes. This page explains the shape of that gap and the knobs that control how a worker recovers. Timer output has separate commit recovery, described below.

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. After a hard crash, the state the worker restarts from — the windowed accumulators and keyed values held in your local RocksDB — can be behind the output topic: the state gap. How far behind depends on how the worker stopped:

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

By default the worker replays the gap — it re-reads those events to rebuild the state, without producing anything — and only then resumes. Either recovery policy preserves the output guarantee; they differ on the state (below). Recovery proceeds only if timer commit reconciliation also succeeds; an incompatible stale snapshot can require operator intervention.

Outputs produced by a timer — a window closing, an early fire, a session ending — are reconciled with the sink before timers resume. With the pending local changes intact, a committed timer batch is finalized without running its callback again. If a stale checkpoint lacks those changes, or the broker commit history is missing or incompatible, the worker stops. The record-state-gap policies below do not permit replaying an already committed timer. See Timer output after a hard crash.

Checkpoints after timer-only progress

A worker can emit timer results without consuming another input message. On restart or a clean handoff back to a previous machine, Turbine checks committed timer history as well as the input offset before choosing local state. If the local copy is stale and a compatible checkpoint is available, it restores that checkpoint even when both copies have the same input offset.

A replacement checkpoint is validated separately before it replaces local state. If it cannot prove compatible history, startup stops and preserves the local copy. Rescale imports for exactly-once state remain unsupported and are rejected before installation. This does not prevent recovery from a compatible full checkpoint.

Retention and reset flags

If retained Kafka input no longer covers an enrolled exactly-once worker's resume position (or the topic was recreated), startup stops and preserves local state. from_earliest and from_latest do not authorize deleting that history: it proves which timer outputs have already committed. Restore a compatible checkpoint whose position is still retained, if one exists. To deliberately start a separate output history, use a new app_id and a fresh state directory; keep the old state and broker offsets until you have resolved the incident. At-least-once subscriptions retain their existing automatic reset behavior.

Where the gap comes from

The Kafka transaction covers (input offsets, output messages). The batch's state changes are written to disk just after, and the two writes are not a single atomic operation. If a crash happens between them, the broker records offsets up to N as processed, while local RocksDB only reflects offsets up to M < N.

That window is the time it takes to write the batch's state durably, so it grows with how much state a batch changes. With small state it is short; with large state it is a sizeable share of each batch cycle — on a high-cardinality windowed aggregation, about one random hard crash in five landed in it. Plan for the gap to happen, not as a rare corner case.

The N−M events in the gap are never re-emitted: their outputs are already committed. What is missing is their effect on the state, and that is what the recovery policy decides about — rebuild it (replay, the default) or skip it (accept). Skipping leaves a counter, a windowed accumulator or a keyed value without those events for good.

The gap is observable via turbine_eos_state_gap_events{worker} — a counter incremented at boot by the size of the gap, whichever policy handles it. 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: replay the gap

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

The worker boots, sees the gap, and silently re-consumes the missing events to rebuild state. During that replay, nothing reaches Kafka — no output, no dead-letter record, no transaction — so the output topic stays never-twice. Once it has caught up with the broker, the worker goes live and outputs resume. The result is the state a crash-free run would have had.

What it costs is the time to re-read the gap, at your normal processing speed:

• Local disk intact (the common case): the gap is at most one batch, so replay is a batch's worth of work — not measurable next to a restart.
• Recovering from a snapshot (a lost disk, or another node taking over): the gap is everything since that snapshot, and the partition starts producing again only once it has replayed it. halt_if_gap_exceeds caps how much it will replay.

A window whose close comes due during the replay closes once, after the replay, with its output produced normally.

Your handler runs again on the replayed events. Its Kafka output is suppressed, but side effects outside Kafka — a database write, an HTTP call — happen a second time. They already happen again whenever a crash aborts a batch's transaction, so under exactly-once they need to be idempotent anyway; see Idempotency contract.

When the gap is no longer in Kafka

Replay reads the gap from the broker. If retention has already deleted part of it, the state cannot be rebuilt, and the worker refuses to start instead of skipping those events silently: that partition halts with an error stating how many events are gone (the rest of the app keeps running, and /health reports degraded — see Error handling). To start anyway, with the state of the whole gap missing, set on_crash_recovery="accept" on that subscription.

Accept the gap (lossy)

@app.subscribe(
    kafka.topic("events"),
    output=kafka.topic("enriched"),
    processing_guarantee="exactly_once",
    on_crash_recovery="accept",   # ← skip the gap; its state updates are lost
)
def enrich(batch): ...

The worker boots, seeks forward to the broker's committed offset and resumes. Outputs are never duplicated, but the events in the gap never reach the state: an aggregate missing them stays wrong, and a window they belonged to closes with a wrong value or not at all. The skipped events are counted in turbine_eos_state_gap_skipped_events{worker}.

Choose it only when getting back to live data matters more than an exact state:

• The gap has left Kafka retention, so replay is impossible (see above).
• The gap is too large to replay in time — typically after restoring an old snapshot — and the state heals itself: short windows that will soon be replaced, or a "latest value per key" that the next event overwrites.

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, that worker halts with a fatal error (the rest of the app keeps running, and /health reports degraded — see Error handling) 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 replay (or skip) a huge backlog.

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

Combining the knobs

The options compose freely:

# Production EOS: rebuild state after a crash (the default), but page someone
# instead of replaying more than 10 M events
@app.subscribe(
    kafka.topic("events"),
    output=kafka.topic("enriched"),
    processing_guarantee="exactly_once",
    halt_if_gap_exceeds=10_000_000,
)
def enrich(batch): ...

There's deliberately no alert_if_gap_exceeds parameter — the counters already let 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

# Alert on any state lost to on_crash_recovery="accept"
increase(turbine_eos_state_gap_skipped_events[1h]) > 0

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