From 4cbb7795eff04ea373fe12c25122d02119243803 Mon Sep 17 00:00:00 2001 From: Brummel Date: Wed, 3 Jun 2026 14:00:06 +0200 Subject: [PATCH] spec: cycle 0004 firing policies and merge MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fourth walking-skeleton cycle: the reactive-semantics layer — both firing policies (C6), the k-way ingestion merge (C3), the monotonic cycle_id clock (C4), and freshness-gated recompute / sample-and-hold (C5). This is the first point where the engine departs from RustAst rather than sharpening it. Both rails ship together (user directive: "must sit right from the start"): - A = fire-on-any-fresh + hold (as-of join); - B = all-fresh barrier (synchronizing join), accumulating across cycles. Per-input-group, mixable on one node. Encoded as Firing::{Any, Barrier(u8)} on InputSpec (where C8 places firing); default Any keeps Sma/Sub unchanged. Key design resolution (studied against RustAst src/ast/rtl/streams/): the barrier token is the data *timestamp*, not cycle_id. RustAst's last_cycle_per_input.all(== cycle_id) synchronizes one source's fan-out but cannot synchronize independent sources — under C4 four same-timestamp sources are four different cycle_ids, so the cycle_id barrier never fires for C6's own OHLC example. Substituting timestamp is the minimal faithful generalization (fires both the diamond rejoin and the multi-source bar). Mode A and the k-way merge are both new (RustAst had neither). Two read clocks, both load-bearing (resolves the 0003 audit's "0004 owns cycle_id" with no write-only field): cycle_id is the freshness epoch (C5 reads it via per-slot fresh_at == cycle_id); timestamp is the barrier token (C6 reads it via last_ts == T). Sample-and-hold falls out of the existing push model — a non-firing node pushes nothing, so consumers see the held value via window[0]. Scope (user-chosen): reactive semantics only — in-memory (timestamp, value) sources; the merge is the permanent ingestion core, the Source trait (C11) + data-server IO are a later orthogonal cycle. bootstrap/run/InputSpec signatures change (the 0003 tests re-express on the multi-source API). Headline test pair (identical sources, firing-only difference): A emits [None,None,120,130,140,240] (holds); B emits [None,None,120,None,None,240] (waits for timestamp coincidence). Plus determinism (C1), mixed A+B, the 0003 fan-out/join compat baseline, and the bootstrap rejections. Grounding-check PASS: 10 load-bearing substrate assumptions ratified by green tests; the spec's "before" shapes match disk exactly. refs walking-skeleton Co-Authored-By: Claude Opus 4.8 (1M context) --- docs/specs/0004-firing-policies-and-merge.md | 351 +++++++++++++++++++ 1 file changed, 351 insertions(+) create mode 100644 docs/specs/0004-firing-policies-and-merge.md diff --git a/docs/specs/0004-firing-policies-and-merge.md b/docs/specs/0004-firing-policies-and-merge.md new file mode 100644 index 0000000..dc3117b --- /dev/null +++ b/docs/specs/0004-firing-policies-and-merge.md @@ -0,0 +1,351 @@ +# Firing Policies + Ingestion Merge — Design Spec + +**Date:** 2026-06-03 +**Status:** Draft — awaiting user spec review +**Authors:** orchestrator + Claude + +## Goal + +Make the reactive semantics of the engine *sit right from the start*: a node +re-evaluates only when its inputs warrant it, driven by **two firing policies** +that must both ship and both be fully tested: + +- **A — fire-on-any-fresh + hold** (as-of join): the node fires whenever any of + its A-inputs is fresh this cycle; stale inputs contribute their last held value + (C5 sample-and-hold). +- **B — all-fresh barrier** (synchronizing join): the node fires only when every + member of a barrier group shares the current cycle's timestamp — e.g. O/H/L/C + assembled from four separate sources, complete only when all four have ticked + at the same bar time. + +Firing is only *observable* when inputs tick at heterogeneous rates, so this +cycle also delivers the enabling substrate: a **k-way merge of multiple sources** +into one chronological cycle stream (C3), the **monotonic cycle_id** clock (C4), +and **freshness-gated recompute** (C5). Per the chosen scope, sources are +in-memory `(timestamp, value)` streams declared by the test/caller; the merge is +the permanent ingestion core, while the `Source` trait + data-server IO are a +later, orthogonal cycle. + +This is the first place the engine departs from RustAst rather than sharpening +it: RustAst had one clock per source and a `cycle_id`-equality barrier that only +synchronizes a single source's fan-out; it never merged heterogeneous sources and +never implemented mode A (see the design notes below). + +## Architecture + +Three contracts compose into one mechanism that turns out to need **two** read +clocks, not one: + +- **`cycle_id` is the freshness epoch (C4 → C5).** The merge advances a monotonic + `cycle_id` per emitted record. Each push into an input slot stamps that slot + with the current `cycle_id`. A slot is *fresh this cycle* iff its stamp equals + the current `cycle_id`. This is the genuine first reader of C4's `cycle_id` — + it is not a write-only field. +- **`timestamp` is the barrier token (C4 → C6).** Each push also stamps the slot + with the cycle's data timestamp `T`. A barrier group is complete iff all its + members carry `last_ts == T`. This is RustAst's `last_cycle_per_input.all(==)` + barrier with the synchronization token changed from `cycle_id` to `timestamp` — + the change forced by C4's tie rule (see design notes). +- **Push *is* freshness propagation.** A node that fires pushes its output into + consumers' input columns, stamping them fresh; a node that does not fire pushes + nothing, so consumers keep seeing the held value via `window[0]`. Sample-and- + hold (C5) falls out of the existing push model with no new storage — only + engine-side per-slot bookkeeping (`fresh_at`, `last_ts`). + +The graph wiring (flat node array, index edge table, Kahn topo order), the +node-owns-its-input-columns shape (cycle 0002), and `Ctx` are all unchanged. What +changes: `bootstrap` takes many sources instead of one; `run` takes many streams +and merges them; each node carries per-input firing + freshness state; the eval +step is gated by the firing predicate instead of running unconditionally. + +## Concrete code shapes + +### User-facing: the two rails, authored as nodes + +The two firing modes are exercised by two fixture nodes that are **identical +except for the firing annotation in their schema** — a 2-input f64 sum. This is +the empirical evidence for the feature-acceptance criterion: the only thing a +node author writes to choose a rail is `Firing::Any` vs `Firing::Barrier(g)`. + +```rust +// Mode A — as-of join: fires whenever either input is fresh; holds the other. +struct AsOfSum; +impl Node for AsOfSum { + fn schema(&self) -> NodeSchema { + NodeSchema { + inputs: vec![ + InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Any }, + InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Any }, + ], + output: ScalarKind::F64, + } + } + fn eval(&mut self, ctx: Ctx<'_>) -> Option { + let a = ctx.f64_in(0); + let b = ctx.f64_in(1); + if a.is_empty() || b.is_empty() { return None; } // warm-up + Some(Scalar::F64(a[0] + b[0])) + } +} + +// Mode B — barrier: fires only when both inputs share the cycle timestamp T. +struct BarrierSum; +impl Node for BarrierSum { + fn schema(&self) -> NodeSchema { + NodeSchema { + inputs: vec![ + InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Barrier(0) }, + InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Barrier(0) }, + ], + output: ScalarKind::F64, + } + } + fn eval(&mut self, ctx: Ctx<'_>) -> Option { + Some(Scalar::F64(ctx.f64_in(0)[0] + ctx.f64_in(1)[0])) + } +} +``` + +### User-facing: wiring two heterogeneous sources and running + +```rust +// source 0 ticks at t=1,2,3,4; source 1 ticks at t=2,4 (slower). +let s0 = vec![(Timestamp(1), Scalar::F64(10.0)), (Timestamp(2), Scalar::F64(20.0)), + (Timestamp(3), Scalar::F64(30.0)), (Timestamp(4), Scalar::F64(40.0))]; +let s1 = vec![(Timestamp(2), Scalar::F64(100.0)), (Timestamp(4), Scalar::F64(200.0))]; + +let mut h = Harness::bootstrap( + vec![Box::new(AsOfSum)], + vec![ + SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] }, + SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 1 }] }, + ], + vec![], // no producer->producer edges + 0, // observe node 0 +).expect("valid"); + +let out = h.run(vec![s0, s1]); +// Mode A: [None, None, Some(120.0), Some(130.0), Some(140.0), Some(240.0)] +// ^ holds s1=100 across t=3 and t=4-on-s0; emits on every tick. +// Mode B (same wiring, BarrierSum instead of AsOfSum): +// [None, None, Some(120.0), None, None, Some(240.0)] +// ^ emits ONLY at t=2 and t=4 where both inputs share the timestamp; holds (no +// emit) at t=1, t=3, and the t=4 s0-cycle until s1 completes the bar. +``` + +The merged cycle order (one record = one cycle, ties by source index, C4) is: +`(t1,s0) (t2,s0) (t2,s1) (t3,s0) (t4,s0) (t4,s1)` — six cycles. The two emitted +vectors above differ exactly at the held cycles, which *is* the difference +between the two rails. + +### Implementation shape (secondary) + +**aura-core — `Firing` enum + `InputSpec.firing` (before → after):** + +```rust +// before +pub struct InputSpec { pub kind: ScalarKind, pub lookback: usize } + +// after +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum Firing { + Any, // mode A: fires the node whenever this input is fresh this cycle + Barrier(u8), // mode B: member of barrier group N; the group fires when all + // its members share the current cycle's timestamp +} +pub struct InputSpec { pub kind: ScalarKind, pub lookback: usize, pub firing: Firing } +``` + +`Sma` (1 input) and `Sub` (2 inputs) in aura-std gain `firing: Firing::Any` on +each declared input — under a single source ticking every cycle, an `Any` input +is fresh every cycle, so their behavior is unchanged (backward compatible). + +**aura-engine — `SourceSpec`, per-slot state, signature changes (before → after):** + +```rust +// new: a declared source — its scalar kind and the input slots it feeds. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct SourceSpec { pub kind: ScalarKind, pub targets: Vec } + +// per-input engine bookkeeping (not visible to nodes). +struct SlotState { fresh_at: u64, last_ts: Timestamp } // init { 0, Timestamp(i64::MIN) } + +// NodeBox gains the lifted firing vector + the per-input slot state. +struct NodeBox { + node: Box, + inputs: Vec, + firing: Vec, // lifted from schema at bootstrap + slots: Vec, +} + +// bootstrap: many sources instead of (source_targets, source_kind). +// before: bootstrap(nodes, source_targets: Vec, edges, source_kind: ScalarKind, observe) +// after: +pub fn bootstrap( + nodes: Vec>, + sources: Vec, + edges: Vec, + observe: usize, +) -> Result + +// run: many streams instead of one record iterator. +// before: run(&mut self, records: impl Iterator) +// after: +pub fn run(&mut self, streams: Vec>) -> Vec> +``` + +The `Harness` struct's `source_targets: Vec` field becomes +`sources: Vec`; no other field is added (`cycle_id` is a loop local, +slot state lives in `NodeBox`). + +**The firing predicate (shape):** + +```rust +fn fires(firing: &[Firing], slots: &[SlotState], cycle_id: u64, ts: Timestamp) -> bool { + // mode A: any A-input fresh this cycle + for (i, f) in firing.iter().enumerate() { + if matches!(f, Firing::Any) && slots[i].fresh_at == cycle_id { return true; } + } + // mode B: for each distinct barrier group, fire iff >=1 member fresh this + // cycle AND all members carry last_ts == ts. The ">=1 fresh" clause is the + // once-per-T guard: a group already completed in a prior cycle has no fresh + // member now, so it does not re-fire. + for g in distinct_barrier_groups(firing) { + let members = barrier_members(firing, g); + let any_fresh = members.iter().any(|&i| slots[i].fresh_at == cycle_id); + let all_at_ts = members.iter().all(|&i| slots[i].last_ts == ts); + if any_fresh && all_at_ts { return true; } + } + false +} +``` + +**The run loop (shape):** maintain a cursor per source; per cycle, pick the live +source head with the smallest `(timestamp, source_index)` (C4 k-way merge); +increment `cycle_id`; push the value into that source's target slots (stamping +`fresh_at = cycle_id`, `last_ts = ts`); walk nodes in topo order, and for each +node that `fires(...)`, `eval` it, capture the observed node's result, and +forward each `Some` output into consumers' slots (stamping them likewise). A node +that does not fire is skipped (held). Allocation-free on the per-cycle path +(cursors + counters only). + +## Components + +- **`Firing` (aura-core)** — the per-input policy enum; part of `InputSpec`, so + it is declared in `schema()` exactly where C8 places it ("declares each input's + … firing group"). +- **`SourceSpec` (aura-engine)** — a declared source: its kind + the input slots + it feeds. The multi-source generalization of cycle 0003's `(source_targets, + source_kind)` pair. +- **`SlotState` (aura-engine, private)** — per-input `{ fresh_at, last_ts }`; the + two read clocks (freshness epoch + barrier token). +- **`Harness::bootstrap` / `run` (aura-engine)** — extended to many sources + + the merge + the firing gate. Topo order, edge kind-checks, and cycle rejection + are unchanged from 0003. +- **`AsOfSum` / `BarrierSum` (test fixtures in aura-engine tests)** — the two + rails as authored nodes; not library nodes (no speculative aura-std surface + until a real strategy needs a reusable join). + +## Data flow + +1. The caller hands `run` one `(Timestamp, Scalar)` stream per source, each + ascending in time (guaranteed by C3 ingestion; the test supplies sorted vecs). +2. The k-way merge emits one record per cycle in global `(timestamp, source + index)` order, advancing `cycle_id` monotonically (C3/C4). +3. The record's value is pushed into its source's target slots; those slots are + stamped `fresh_at = cycle_id`, `last_ts = ts`. +4. Nodes evaluate in topo order. Each node's firing predicate reads its slots' + `fresh_at` (freshness, C5) and `last_ts` (barrier, C6). A firing node evals + and forwards `Some` outputs into consumers (stamping them fresh); a non-firing + node holds. +5. The observed node's emitted value (`Some` when it fired and produced output, + `None` when it held or filtered) is collected per cycle and returned. + +## Error handling + +- **Bootstrap (`BootstrapError`, unchanged variants):** + - `KindMismatch { producer, consumer }` — a source target's slot kind ≠ the + source's `kind`, or an edge's producer output ≠ consumer slot kind. Checked + per source and per edge, as in 0003. + - `BadIndex` — a source target, edge endpoint, or `observe` index out of range. + - `Cycle` — the wiring contains a directed cycle (Kahn leftover). +- **`run` precondition:** `streams.len()` must equal `sources.len()`, and each + stream must be ascending by timestamp. A mismatch in length is a caller bug; the + shape (panic with a clear message vs. a typed error) is a planner decision, but + it is *not* a silent no-op. Stream monotonicity is a documented precondition of + the C3 ingestion boundary, not re-validated in the hot path. +- **Cold barrier member:** a slot never pushed has `last_ts = Timestamp(i64::MIN)`, + which equals no real timestamp, so a barrier with an un-ticked member never + completes — correct warm-up behavior. +- **Zero-input node:** never has a fresh input, so never fires; harmless (no such + node exists today). Noted, not special-cased. + +## Testing strategy + +All on both rails, all deterministic (C1): + +1. **Mode A — as-of hold** (`aura-engine`): the worked `AsOfSum` example above; + assert the emitted vector `[None, None, 120, 130, 140, 240]`, proving the held + s1 value (100) is reused across t=3 and the t=4 s0-cycle. +2. **Mode B — barrier wait** (`aura-engine`): same wiring with `BarrierSum`; + assert `[None, None, 120, None, None, 240]`, proving the node emits only when + both inputs share the timestamp and holds (no emit) otherwise. +3. **Determinism** (C1): a second identical `run` of each rail is bit-identical to + the first. +4. **Mixed A+B on one node:** a 3-input fixture `[Barrier(0), Barrier(0), Any]` + fed by three sources; assert it fires both when the barrier pair completes + (held A) and when the A-input ticks (held barrier pair) — the OR-combine. +5. **Backward-compat fan-out/join** (single source, all `Any`): the 0003 + `source -> {Sma(2), Sma(4)} -> Sub` DAG re-expressed on the new API still emits + `[None, None, None, 2, 2, 2]`, proving the firing gate is transparent when one + source ticks every cycle. +6. **Bootstrap rejections** (adapted to `SourceSpec`): kind-mismatch (i64 source + into an f64 slot), bad index (observe out of range), and a directed cycle — + the 0003 reject tests re-expressed. +7. **aura-core:** `InputSpec`/schema carry `firing`; `Sma`/`Sub` declare + `Firing::Any`; existing aura-core/aura-std tests stay green. + +## Acceptance criteria + +Per aura's feature-acceptance criterion (the audience reaches for it; it improves +correctness; it reintroduces no failure class the core constraints forbid): + +- **Reached-for:** authoring a strategy that combines a fast price with a held + daily bias (mode A) or assembles a bar from separate O/H/L/C feeds (mode B) is + exactly C6's named motivation; the only authoring act is the `Firing` + annotation, shown above. +- **Correctness / no look-ahead (C2):** freshness reads only past pushes; the + barrier waits for completeness rather than peeking ahead; held values are + explicit last values, never future ones. +- **Determinism (C1):** the merge tie-break (source order) and the firing + predicate (pure function of `fresh_at`/`last_ts`, which evolve deterministically) + make every run reproducible — asserted in test 3. +- **No new failure class:** no interior concurrency, no per-cycle allocation, no + merge inside the graph (C3 — the single merge is at ingestion), no `dyn Any` / + `Rc` / `RefCell` on the surface. +- Gates green: `cargo build/test/clippy --workspace -- -D warnings`; surface- + purity grep clean. + +## Design notes — why aura diverges from RustAst here + +RustAst (`src/ast/rtl/streams/`) is the only prior implementation, and studying it +fixed two decisions: + +- **Barrier token: `timestamp`, not `cycle_id`.** RustAst's `PipeStream` barrier + is `last_cycle_per_input.iter().all(|&c| c == cycle_id)` — all inputs reported + the same `cycle_id`. That synchronizes one source's fan-out/rejoin (every path + carries the originating tick's `cycle_id`), but it **cannot** synchronize + independent sources: under C4, four sources at the same bar timestamp become + four *different* `cycle_id`s (ties broken by source order), so `all(== cycle_id)` + would never fire for C6's own O/H/L/C example. Substituting `timestamp` for + `cycle_id` is the minimal faithful generalization: it fires the diamond rejoin + (paths share the cycle timestamp) *and* the multi-source bar (separate cycles, + shared timestamp). +- **Mode A is new.** RustAst implemented only the barrier (C6: "RustAst + implemented only B"); fire-on-any-fresh + hold has no prior art and is built + here from the freshness primitive. +- **The merge is new.** RustAst gives each source its own `RootStream` clock + (`data_streams.rs`) with no k-way merge; aura's single merged timeline (C3) with + one shared `cycle_id`/`timestamp` sequence is precisely what makes a cross-source + barrier meaningful. RustAst's `total_count` push counter is aura's already-built + `Column::run_count` (glossary: run-count, Avoid total_count).