Exactly-once

A checkpoint is one atomic unit: every operator's state plus every source offset, written as a single file. Recovery loads the last complete checkpoint and replays the log from the offsets it recorded. There is no other recovery path. A kill -9 in the middle of an open window is a test case, not an edge case.

This page shows the guarantee on a file source, so you can run every command below and get the numbers below. The Kafka sink side, where each window is a transaction, is on the Kafka page.

What a checkpoint holds

One file, written to a temporary path, fsynced, renamed over the target, then the directory is fsynced. A partial checkpoint cannot exist on disk: a run either finds a complete one or none. A flipped byte is rejected rather than loaded.

A sibling <path>.lock file guards the checkpoint against a second engine. It is an OS flock, held for the process lifetime and released by the kernel on any exit, including kill -9.

Stop mid-window, resume, no drift

Six one-minute windows, two merchants, 700 rows each per window, in time order. The rows for a single window span thousands of lines, so a checkpoint taken mid-stream lands inside an open window.

python3 - orders.ndjson <<'PY'
MERCH = [("acme", 10), ("globex", 3)]
with open("orders.ndjson", "w") as f:
    for w in range(6):
        ts = "2026-07-18 10:%02d:00" % w
        for _ in range(700):
            for m, amt in MERCH:
                f.write('{"merchant":"%s","amount":%d,"event_time":"%s"}\n' % (m, amt, ts))
PY

The query is the same for every run below, so keep it in a variable.

SQL="SELECT merchant,
            TUMBLE(event_time, INTERVAL '1 MINUTE') AS w,
            count(*) AS n,
            sum(amount) AS total
     FROM orders
     GROUP BY merchant, TUMBLE(event_time, INTERVAL '1 MINUTE')"

Run it clean, to completion. This is the ground truth: twelve rows, two merchants over six windows.

ubik --from file://orders.ndjson "$SQL"

{"merchant":"acme","w":"2026-07-18 10:00:00","n":700,"total":7000}
{"merchant":"acme","w":"2026-07-18 10:01:00","n":700,"total":7000}
{"merchant":"acme","w":"2026-07-18 10:02:00","n":700,"total":7000}
{"merchant":"acme","w":"2026-07-18 10:03:00","n":700,"total":7000}
{"merchant":"acme","w":"2026-07-18 10:04:00","n":700,"total":7000}
{"merchant":"acme","w":"2026-07-18 10:05:00","n":700,"total":7000}
{"merchant":"globex","w":"2026-07-18 10:00:00","n":700,"total":2100}
{"merchant":"globex","w":"2026-07-18 10:01:00","n":700,"total":2100}
{"merchant":"globex","w":"2026-07-18 10:02:00","n":700,"total":2100}
{"merchant":"globex","w":"2026-07-18 10:03:00","n":700,"total":2100}
{"merchant":"globex","w":"2026-07-18 10:04:00","n":700,"total":2100}
{"merchant":"globex","w":"2026-07-18 10:05:00","n":700,"total":2100}

Now cut the run short. --stop-after 1 checkpoints after one chunk (about 2048 rows) and stops. That boundary falls inside the 10:01 window: the first window is closed and emitted, the second is folded into state but held.

ubik --from file://orders.ndjson "$SQL" --checkpoint ck --stop-after 1

The status and stats print on stderr; the closed window prints on stdout.

{"status":{"state":"stopped","checkpoint":"ck"}}
{"stats":{"rows_in":2048,"rows_out":2,"late_dropped":0,"windows_closed":1,"rows_folded":2048}}
{"merchant":"acme","w":"2026-07-18 10:00:00","n":700,"total":7000}
{"merchant":"globex","w":"2026-07-18 10:00:00","n":700,"total":2100}

The checkpoint is on disk, with its lock.

ls ck ck.lock

The 10:01 window emitted nothing. Its rows read before the cut are inside the checkpoint's operator state, not lost and not yet counted. Resume from it.

ubik --from file://orders.ndjson "$SQL" --resume ck

{"status":{"state":"done"}}
{"stats":{"rows_in":6352,"rows_out":10,"late_dropped":0,"windows_closed":5,"rows_folded":6352}}
{"merchant":"acme","w":"2026-07-18 10:01:00","n":700,"total":7000}
{"merchant":"acme","w":"2026-07-18 10:02:00","n":700,"total":7000}
{"merchant":"acme","w":"2026-07-18 10:03:00","n":700,"total":7000}
{"merchant":"acme","w":"2026-07-18 10:04:00","n":700,"total":7000}
{"merchant":"acme","w":"2026-07-18 10:05:00","n":700,"total":7000}
{"merchant":"globex","w":"2026-07-18 10:01:00","n":700,"total":2100}
{"merchant":"globex","w":"2026-07-18 10:02:00","n":700,"total":2100}
{"merchant":"globex","w":"2026-07-18 10:03:00","n":700,"total":2100}
{"merchant":"globex","w":"2026-07-18 10:04:00","n":700,"total":2100}
{"merchant":"globex","w":"2026-07-18 10:05:00","n":700,"total":2100}

The resume read 6352 rows. Added to the partial run's 2048, that is 8400, the whole file and not one line more: the source picked up at the offset the checkpoint stored, so nothing was replayed. The 10:01 window comes out with n:700, assembled across the cut from the rows folded into the checkpoint and the rows read after it. Concatenate the two runs' output and it is byte-identical to the ground truth: no row lost, none double-counted.

What a kill -9 guarantees

A crash is the same as --stop-after, minus the clean stop. The last complete checkpoint on disk is the recovery point; everything after it is replayed on resume from the recorded offsets, deterministically. The flock is released by the kernel on the kill, so the resuming engine takes it with nothing to clean up.

--checkpoint-every N sets how much a crash costs. It checkpoints every N chunks of roughly 2048 rows; a smaller N replays less after a crash and writes more often. The crash harness in CI runs this loop, killing the engine with SIGKILL mid-stream across several cycles, and compares the result to a DuckDB batch over the same events at rest. Green means byte-identical.

One command for first run and restart

--pipeline PATH is the auto-resume form: it checkpoints to PATH and resumes from it if it already exists, so one command serves both the first start and every restart after a crash. It needs --follow for a live source or --checkpoint-every for a periodic checkpoint.

That is one ExecStart line, and the supervisor is the OS.

[Service]
ExecStart=/usr/local/bin/ubik --from kafka://broker/orders "<SQL>" \
  --to kafka://broker/orders_agg --pipeline /var/lib/ubik/orders --follow
Restart=on-failure

Restart=on-failure plus auto-resume is the whole crash-recovery story: systemd restarts the process, --pipeline finds the checkpoint and continues from it.

The Kafka sink is transactional

When --to is a Kafka topic, the output is a changelog keyed by (window, group), upserted. Each closed window's rows are wrapped in one Kafka transaction, so a read_committed consumer sees a window atomically or not at all. A crashed producer is fenced on resume by a stable transactional.id, so its in-flight transaction is aborted and the window is re-emitted. Exactly-once across a crash is deterministic replay plus an idempotent key plus checkpoint-owned offsets, not two-phase commit: a re-emitted window upserts byte-identical rows over the first. Connection and offset details are on the Kafka page.

Flags

When a resume refuses

A resume that cannot guarantee exactly-once fails loudly rather than producing a wrong result.