Module dataflow 

THE DATAFLOW RUNTIME (Phase 2): a pipeline as tasks on OS threads connected by bounded in-memory edges that carry Arrow batches and the two control signals — watermarks and checkpoint barriers — in band (design/2026-09-15-phase2-dataflow-scoping.md).

• A task owns one operator and runs a loop over its single inbox, into which every upstream edge sends tagged messages; FIFO per edge is what per-key order rests on.
• An edge is sync_channel-bounded: a full inbox backpressures the sender.
• Shuffle hashes the routing columns of a batch (create_hashes) into vnodes and sends one sub-batch (take) per downstream task by its vnode range; forward sends whole batches to one task; broadcast sends the same batch to every downstream task.
• Watermarks: each source emits Watermark { source, ts } (None = idle); a task keeps the latest per (edge, source), fires its operator on the minimum over the active ones, and emits its own watermark downstream under its own id, so the whole graph shares one time.
• Barriers and epochs: the coordinator issues epochs — periodically from the runtime’s own ticker, and one last time on stop — as a Control::Barrier to every source; a source records its position, reports a Snapshot, and forwards Barrier { epoch } behind the data it already sent. A task with several input edges aligns — batches arriving on an edge that already delivered the barrier are held until every edge has — then snapshots its operator, forwards the barrier, and hands the snapshot to the coordinator. Sinks make their output durable and Ack. When every task has reported an epoch (or finished), the epoch is complete: the coordinator sends Control::Commit to the sources, which commit the positions they recorded at that barrier. Offsets therefore never run ahead of durable output — the at-least-once gate, now across task boundaries.
• Eos flows in order behind the last data and the final barrier; a task finishes when every input reached it.

The runtime is Kafka-free: sources and sinks are traits, tested here with in-memory ones.

Structs

Chained

A sink CHAINED into a source’s task: the source’s batches go straight to the sink on the same thread — no edge, no hop, no second thread to wake (the scoping note’s chaining, for the stage that needs no reshuffle). At a barrier the sink makes its output durable FIRST, then the source records its position; a sink that cannot (a failed flush) leaves the epoch without a position, so the commit that follows commits nothing newer. The task reports one Snapshot per epoch and no Ack: the ack is implied by the order.

Completed

An epoch every task reported (or had finished before): its snapshots, by task.

EpochStaging

Commit-gated output staging for an exactly-once sink that has no external transaction of its own (the Arrow file sink, the Flight sink) — the software analogue of the Kafka sink’s producer transaction. on_data buffers into the open epoch; on_barrier seals it (pre-commit); commit releases every sealed epoch up to the committed one for the sink to make visible. Nothing is visible before its epoch commits, so a restart from the last committed checkpoint re-produces the same rows (deterministic keys) and each committed row appears exactly once. Uncommitted staged rows are dropped on stop — a replay re-emits them. (At-least-once sinks skip this and emit in on_data, unchanged.)

EpochTracker

The coordinator’s book: which tasks reported which epoch, which tasks finished. Epochs complete in order. An epoch completes when every expected task has reported it (or finished before it) — for a single-process run that is every task 0..n; for one worker of a multi- process run it is only the tasks placed on this worker (the leader aggregates across workers).

Graph

The graph under construction: add tasks, connect routes, then start.

OpSnapshot

What a task checkpoints at a barrier: a small head written every epoch, and zero or more immutable files uploaded once. A source, or an operator whose whole state is the head, has no files (OpSnapshot::whole).

Out

An operator’s outputs for one input event: batches to route, and its own watermark if it advanced. The task routes them.

Probe

One task’s probe: its state, the target of a send, and when it entered the state.

Running

A started dataflow: issue epochs, commit them, stop it, join it.

StateFile

An immutable local state file an operator produced at a barrier: content-stable, so a backend that stores files (the object store) uploads it once and later epochs that still reference it re-upload nothing. min_time/max_time are its event-time range, for the manifest and rescale.

WorkerExchange

This worker’s exchange endpoints for a multi-process run: one LinkSender per peer worker it sends to, and one LinkReceiver per peer it receives from (established by the membership handshake before start). Graph::start runs with none of this — a single-worker run where every edge is a local SyncSender; the multi-process path is Graph::start_worker, which derives the actual edge wiring from the placement.

Enums

Control

What the coordinator tells a source.

Event

What the coordinator hears from tasks.

Msg

What flows on an edge.

Poll

Route

Where a task’s output goes.

TaskState

What a task is doing right now — the runtime’s own answer to “where is it stuck”: every task keeps one, the coordinator reads them (Running::task_states).

Traits

Operator

What a task runs. Operators are single-threaded and own their state; the runtime never calls two methods concurrently.

Sink

A sink consumes batches and acks barriers.

Source

A source drives itself: the runtime calls poll until it returns Poll::Done (or the coordinator stops it).

Functions

position_record

The position record of a single-partition source (partition is its task index): the one shape the checkpoint reads a source’s positions from — topic keys the restore’s lookup, in maps a partition to the next offset. Every in-process source writes its barrier record with this.

process_cpu_by_thread

The CPU time of every thread in this process so far, summed by thread name (milliseconds): the engine’s tasks by their fv-task-N names, librdkafka’s rdk:* threads, the rest. Empty where /proc is absent. This is where a build’s cores go, at a glance.

shuffle

Split a batch by vnode: hash(columns) % vnodes → the target whose range holds it. Returns one optional sub-batch per target, in target order. Missing routing columns hash as null.

thread_cpu_ms

The CPU time this thread has used so far (user + system), from /proc/thread-self/stat; None where that file does not exist. Linux reports it in USER_HZ ticks, which is 100 on every architecture regardless of the kernel’s HZ.

vnode_ranges

Contiguous vnode ranges for tasks tasks over vnodes vnodes.

Type Aliases

Acker

Acks an epoch’s barrier from off the sink’s thread: a sink whose durability lands elsewhere (a file committer syncing parts, a table commit) withholds the ack in on_barrier and calls this once the epoch is durable — under its own task id, which the runtime binds in.

DeferredOperator

An operator built once its task’s input edges are known: (task id, in-edges with the upstream task each comes from) — a two-input operator learns which edges are which side. Err when it cannot be built (a checkpoint it refuses to restore): the whole start fails, before any task runs, with that message.

DeferredSink

A sink built at start, on the worker that owns its task, knowing its task id — for a sink that touches a shared resource (a durable file sink’s directory) and must be opened by its owner only, and one that acks its barriers from another thread under its own id. A builder that fails fails the start before any task runs, exactly as a deferred operator’s does.

EdgeId

An edge’s id: unique within one running graph, and the same on every worker.

Hasher

Hashes the routing columns of a batch into out (one u64 per row) — the shuffle’s key hash, supplied by the planner (the engine’s DataFusion hash, fixed-seed: a key’s vnode is the same on every run, every worker and every restart). The runtime never chooses a hash function.

TaskId

A task’s id: unique within one running graph.