A sharded sensor pipeline

Three record types and three topics are the whole vocabulary. Frames is keyed by shard, so the routing table lives on the topic and the handlers carry none of it. Everything downstream is typed against these payloads, and nothing in the program deserializes and hopes.

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 annotations are contracts the compiler checks: it 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 with 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.

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 to that: the first draft of this page declared only {Metrics}, and the build failed with the path Frames -> ShardWorker -> Metrics -> Aggregator. The workers carry frame data into metrics, so Frames is a real dependency whether the declaration says 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 plain: subscribe, count, print. It runs on the main scheduler, and its one producer is the aggregator, so 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: every live run differs, which is what the closing step 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 keeps the process alive while the messages drain. The program constructs no channels, spawns no threads by hand and writes no shutdown sequence; 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 293

$ hale run pipeline.hl   # run 2
ALERT shard 1 value 292
ALERT shard 1 value 300

Two live runs of this program never match, because the probe reads real entropy. Record one, and the runtime journals every random read and every consumer's delivery order beneath 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, 85 consumes
ALERT shard 1 value 286
replay matches the recording (386 ring records). Compared, by category:
  public bus events:     162 across 3 consumers (publish + deliver, subject-aligned)
  payloads:              81 (sizes and identities; contents not canonicalised)
  queued consumes:       85 across 2 consumers (per consumer, in recorded order)
  async schedule steps:  not exercised: no async pool scheduling steps were recorded
  journal reads:         40 (kind, args, withheld state, value)
A category reported as not exercised was NOT verified by this match.