Checkpointing

• The position is a set, not a number. One offset per source partition; a checkpoint missing any of them describes no reachable state.
• The state is everything derived from consumed input — open windows, keyed aggregates, join buffers, session state, deduplication sets. Anything held in memory that affects future output belongs in it (Streaming State).
• Emitted output is not in the checkpoint. Rows already written to the sink survive the restart and will be written again during replay, which is why sink idempotency is a separate and mandatory concern.
• Checkpoint storage must outlive the node. Local disk covers a process crash and not a machine loss, which is the failure most likely to need it.
• Several checkpoints, not one. The most recent one can be corrupt or incomplete; the fallback is only available if it was retained.

A timer writes the in-memory state to durable storage; the consumer commits its offsets on its own schedule. Two mechanisms, two cadences, two storage systems, no relationship between them beyond both being roughly current.

The source positions are a field of the checkpointed state. One conditional write, or one transaction, makes the pair durable. Recovery loads the object and derives the resume position from it, so there is no way for the two to disagree.

Two independent durable writes have a crash window between them by construction, and no ordering removes it — it only chooses whether the failure appears as duplication or as loss. Making the pair a single write eliminates the window rather than narrowing it, which is the same reason the transactional outbox exists on the operational side (The Transactional Outbox).

1UNSAFE — two independent durable writes
2--------------------------------------
3loop:
4    batch = source.read(from = position)
5    state = fold(state, batch)
6    position = batch.end_position
7 
8    every checkpoint_interval:
9        checkpoint_store.put(state)        # write 1
10        source.commit_offsets(position)    # write 2
11        # crash between write 1 and write 2  -> gap is reprocessed into
12        #   state that already contains it  -> double counting
13        # swap the two lines and a crash between them -> gap is skipped
14        #   entirely and its contribution is lost, silently
15 
16SAFE — one durable write containing both
17----------------------------------------
18loop:
19    batch = source.read(from = checkpoint.position)
20    state = fold(state, batch)
21 
22    every checkpoint_interval:
23        checkpoint_store.put_atomic({
24            state:    state,
25            position: batch.end_position,   # same object, same write
26            version:  logic_version,        # so a restore knows what it means
27        })
28 
29on restart:
30    checkpoint = checkpoint_store.latest_complete()
31    state      = checkpoint.state
32    resume_from(checkpoint.position)
33    # input between checkpoint.position and the crash is read AGAIN.
34    # that is at-least-once consumption, and it is fine ONLY because
35    # the sink is keyed and idempotent:
36    #     MERGE INTO agg USING new ON agg.key = new.key
37    #       WHEN MATCHED THEN UPDATE ... WHEN NOT MATCHED THEN INSERT ...

The safe version does not eliminate reprocessing — it makes reprocessing harmless. That is the actual goal, and it is why the sink appears in a lesson about checkpoints: a consistent checkpoint plus a non-idempotent sink is still a broken pipeline, just one that fails on the output side instead of the state side.

The replay interval is exactly the span between the last complete checkpoint and the crash — here 10:03 to 10:06. Shortening the checkpoint interval shrinks that span and nothing else; it does not make the replay safe. Only the sink can do that.