Skip to content

Fan-out Engine

Fan-out looks like magic from outside (attach a child, copies appear), but it's a deliberately boring machine: per-partition cursors tailing the parent's committed log, with durable positions and an at-least-once commit protocol. No double-publish from producers, no broker-side subscriptions; just log readers that never lose their place.

Where the work runs

For each (parent partition × child), one cursor goroutine runs on the node that owns that parent partition, so slab reads are always local disk, and the cursor's durable state lives in the same directory (same durability domain) as the log it tails:

flowchart LR
    subgraph owner["owner of orders/p3"]
        LOG[("orders/p3 log")]
        CUR1["cursor → analytics"] --> LOG
        CUR2["cursor → retry (1h delay)"] --> LOG
        OFF[("fanout-analytics.offset<br/>fanout-retry.offset")]
    end
    CUR1 -->|"commit batch"| A[("analytics/p* owners")]
    CUR2 -->|"commit batch"| R[("retry/p* owners")]

A reconciler on every node diffs desired cursors (from the metastore: links × owned partitions) against running cursors once a second: cursors spawn on attach, stop on detach/delete/ownership change.

The cursor loop: commit-before-advance

flowchart TD
    READ["read slab of committed parent records<br/>(fill-or-linger batching)"] --> REKEY["re-key each record with the<br/>child's partitioner (key preserved)"]
    REKEY --> COMMIT["commit per-child-partition batches<br/>(local or one RPC to the owner)"]
    COMMIT -->|all batches acked| PERSIST["persist cursor offset"]
    PERSIST --> READ
    COMMIT -->|any failure| RETRY["back off, re-read from<br/>unadvanced offset"] --> READ

The invariant is the whole guarantee: the cursor's durable offset only advances past records whose child commits were acknowledged (and child commits are the same fsync-and-verify as any produce). A crash mid-flight re-commits the last slab: duplicates into the child, never a gap.

Attach epochs: why re-attach never replays

Each attachment gets a fresh epoch ID, stamped into the cursor's offset file. Detach + re-attach must start from the parent's current tail (the client contract says "no backfill"), so a cursor that finds an offset file from a different epoch refuses to resume it. Epochs turn "is this my state?" from a guess into an equality check.

Tail-anchoring (the act of skipping to the tail and overwriting the offset file) is the engine's only destructive move, and chaos testing showed a stale metastore replica can fabricate exactly the epoch mismatch that triggers it. So an anchor now requires the Raft leader to confirm the epoch (with the barrier rule for self-leaders); anything unconfirmed defers, and the reconciler retries a second later. Cursor offset files get the same protection before deletion. The incident that forced this is told in Cluster Lifecycle.

The delay gate

A delay child's cursor adds one filter: only read records whose parent commit time is ≤ now − delay.

flowchart LR
    subgraph parent log
        r1["r₁ committed 12:00:00"] --> r2["r₂ committed 12:00:01"] --> r3["r₃ committed 12:00:02"]
    end
    GATE{"now − delay ≥ commit time?"} -->|"yes: fan out"| r1
    GATE -->|"not yet: sleep until r₂ is due"| r2

Because commit times are monotonic per partition (assigned under the partition lock, a storage-engine property), the first not-yet-due record proves everything behind it isn't due either. So an idle delay cursor is O(1): peek the head, sleep until its due time (capped so gauges stay fresh). No timer wheels, no scan-the-backlog polling: a million pending delayed messages cost the same as one.

Delivery is therefore never early (the gate is checked against the owner's clock at read time) and usually lands within a second of due (long-poll wakeups + the linger window).

Edge behaviors

  • Drop-behind: if a cursor falls behind the parent's retention (child down for days), aged-out offsets are skipped and counted on an explicit loss metric: bounded, alarmed loss instead of a wedged parent. The retention floor (≥ delay + 1h for delay children) makes this unreachable in sane configs.
  • Dead child-partition owner: fan-out never reroutes to a sibling child partition (unlike produce, a cursor can afford to wait; rerouting would scatter a key's records across child partitions for no availability gain); the cursor stalls on that bucket and retries until the owner returns.
  • Lag observability: fanout_lag_messages (parent HWM − cursor) is the health signal for normal children; fanout_due_lag_seconds (how far behind the due frontier) is the one for delay children; raw offset lag on a delay child is permanently ≈ rate × delay by design.

The numbers

Constant Value
Reconcile interval (desired vs running cursors) 1s
Slab long-poll / retry backoff 1s / 1s
Batch caps 4,096 records / 4 MiB (fanout.max_batch_records/bytes)
Linger to fatten a partial batch 25ms
Delay cursor max sleep (metadata freshness bound) 30s (defaultFanoutDueWakeCap)
Orphan cursor-file sweep every 30th reconcile pass
Max children per parent / max delay 108 / 1 year
Retention floor for a delay child's parent delay + 1h (topic.MinRetentionMs)
Attach epoch 8 random bytes, hex (e.g. 67953471cc57a32a)

The cursor's durable state, in full

One JSON file per (parent partition, child), living next to the log it indexes:

topics/orders/p00003/fanout-analytics.offset
{"epoch":"67953471cc57a32a","next_offset":98332}

That's the entire recovery story: epoch says which attachment this position belongs to; next_offset says where to resume. Written via atomic temp+rename only after the batch below it is committed to the child (commit-before-advance). Everything else (running goroutines, batches in flight, lag gauges) is disposable.

Reading a slab: fill-or-linger

readBatch long-polls the parent for up to 1s, then tops up until the batch hits 4,096 records / 4 MiB or a 25ms linger expires, so a busy parent produces fat child commits (one fsync each on the child side) while a trickling parent still ships within ~25ms. For a delay child, every read carries MaxCommittedAt = now − delay; the reader stops at the first undue record and reports when it becomes due, which is what lets the cursor sleep instead of spin.