A sharded sensor pipeline

Three record types and three topics are the whole vocabulary. `Frames` is keyed by `shard` — the routing table lives on the topic, not in handler code. Everything downstream is typed against these payloads; there is no deserialize-and-hope step anywhere in this program.

type RawFrame { shard: Int; seq: Int; sample: Int; }
type Metric   { shard: Int; value: Int; }
type Alert    { shard: Int; value: Int; }

topic Frames  { payload: RawFrame; keyed_by shard; }
topic Metrics { payload: Metric; }
topic Alerts  { payload: Alert; }

The decode step is the hot path, so it says so. These are contracts, not comments: the compiler follows every callee and rejects the build if the function ever reaches a syscall, nondeterminism, or an allocation. Add one string concatenation and see.

@no_syscall @deterministic
@budget(alloc_per_call = 0)
fn decode(f: RawFrame) -> Int {
    return f.sample * 3 + f.seq % 7;
}

Build a label on the hot path.
-     return f.sample * 3 + f.seq % 7;
+     let tag = "frame-" + to_string(f.seq);

$ hale check pipeline.hl
pipeline.hl:11:4: type error: hot-path budget exceeded: `decode` declares `@budget(alloc_per_call = 0)` but the compiler counts 1 arena allocation(s) per call. For a zero-alloc hot path: hoist per-iteration allocations to reused fields, read with `recv_into` into a reused `std::bytes::BytesBuilder`, and keep locus instantiation out of the fn. If the allocation is intentional, drop `@budget` for `@unbounded fn`.
    fn decode(f: RawFrame) -> Int {
       ^^^^^^
pipeline.hl:12:15: type error: counts against `decode`'s alloc budget: a string concatenation (each `+` builds a fresh String)
        let tag = "frame-" + to_string(f.seq);
                  ^^^^^^^^^^^^^^^^^^^^^^^^^^^

Each worker subscribes to its own shard — `where key == self.shard` — so the fan-out is declarative and a worker never sees another shard's frames. Both workers are placed on one cooperative pool below, which makes the pair a single consumer: one thread, deterministic order, no locks in sight.

locus ShardWorker {
    params { shard: Int = 0; handled: Int = 0; }
    bus {
        subscribe Frames as ingest where key == self.shard;
        publish Metrics;
    }
    fn ingest(f: RawFrame) {
        self.handled = self.handled + 1;
        Metrics <- Metric { shard: f.shard, value: decode(f) };
    }
}

The aggregator declares what may influence it. The set is transitive through the bus graph, and the compiler holds it honestly: the first draft of this very page declared only `{Metrics}`, and the build failed with the path `Frames -> ShardWorker -> Metrics -> Aggregator` — the workers launder frame data into metrics, so `Frames` is a real dependency whether we say so or not. Try hiding it again:

@effects(depends: {Metrics, Frames})
locus Aggregator {
    params { peak: Int = 0; count: Int = 0; }
    bus {
        subscribe Metrics as fold;
        publish Alerts;
    }
    fn fold(m: Metric) {
        self.count = self.count + 1;
        if m.value > self.peak {
            self.peak = m.value;
            if m.value > 250 {
                Alerts <- Alert { shard: m.shard, value: m.value };
            }
        }
    }
}

Declare less than the truth.
- @effects(depends: {Metrics, Frames})
+ @effects(depends: {Metrics})

$ hale check pipeline.hl
pipeline.hl:27:10: type error: declared dependency set violated: `Aggregator` can transitively depend on `Frames` through the bus, which its `@effects(depends: …)` does not declare. Path: subject `Frames` -> `ShardWorker` -> subject `Metrics` -> `Aggregator`. Add the subject to the set, or route the input through a subject this locus doesn't reach.
    @effects(depends: {Metrics})
             ^^^^^^^

The sink is deliberately boring — subscribe, count, print. It runs on the main scheduler, and because its one producer is the aggregator, its output order is the aggregator's fold order.

locus AlertSink {
    params { fired: Int = 0; }
    bus { subscribe Alerts as raise_alarm; }
    fn raise_alarm(a: Alert) {
        self.fired = self.fired + 1;
        println("ALERT shard ", a.shard, " value ", a.value);
    }
}

The probe is pinned to its own OS thread and publishes forty frames with random samples. The randomness is the point: it makes every live run different, which is what the closing act needs.

locus Probe {
    bus { publish Frames; }
    run() {
        let mut seq = 0;
        while seq < 40 {
            let sample = std::rand::next_int(100);
            Frames <- RawFrame { shard: seq % 2, seq: seq, sample: sample };
            seq = seq + 1;
        }
    }
}

Assembly is declarative: the tree owns the loci, `placement` puts the probe on core 0 and both workers on one pool, and `run` just keeps the process alive while the messages drain. No channels are constructed, no threads are spawned by hand, no shutdown choreography — dissolve order is the ownership tree.

main locus Pipeline {
    params {
        w0: ShardWorker = ShardWorker { shard: 0 };
        w1: ShardWorker = ShardWorker { shard: 1 };
        agg: Aggregator = Aggregator { };
        sink: AlertSink = AlertSink { };
        probe: Probe = Probe { };
    }
    placement {
        probe: pinned(core = 0);
        w0: cooperative(pool = shards);
        w1: cooperative(pool = shards);
    }
    run() { std::time::sleep(900ms); }
}
fn main() { Pipeline { }; }

$ hale run pipeline.hl   # run 1
ALERT shard 0 value 301

$ hale run pipeline.hl   # run 2
ALERT shard 0 value 280

Now the part no other language's example page can do. Two live runs of this program never match — the probe reads real entropy. Record one, and the runtime journals every random read and every consumer's delivery order below the application. Replay serves the same values back in the same order, and the comparison signs off: the run became a file, and the file runs again.

$ LOTUS_OBS_RECORD=pipeline.halerec hale run pipeline.hl
$ hale replay pipeline.halerec pipeline.hl --allow-live-effects --diff
hale replay: journal served fully — 0 divergences, 83 consumes
ALERT shard 1 value 282
ALERT shard 0 value 295
ALERT shard 1 value 302
replay matches the recording: 83 consumes across 2 consumers; canonical payloads identical, raw ABI payload sizes matched (390 ring records, 40 journal reads)