Brummel/Aura
1
0
Fork 0
Code Issues 31 Pull Requests Actions Packages Projects Releases Wiki Activity

spec: cycle 0004 firing policies and merge

Browse Source 
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) <noreply@anthropic.com>
This commit is contained in:
Brummel
2026-06-03 14:00:06 +02:00
parent 02fe89176f
commit 4cbb7795ef
1 changed files with 351 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/specs/0004-firing-policies-and-merge.md
+351
View File
@@ -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<Scalar> {
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<Scalar> {
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<Target> }
// 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<dyn Node>,
inputs: Vec<AnyColumn>,
firing: Vec<Firing>, // lifted from schema at bootstrap
slots: Vec<SlotState>,
}
// bootstrap: many sources instead of (source_targets, source_kind).
// before: bootstrap(nodes, source_targets: Vec<Target>, edges, source_kind: ScalarKind, observe)
// after:
pub fn bootstrap(
nodes: Vec<Box<dyn Node>>,
sources: Vec<SourceSpec>,
edges: Vec<Edge>,
observe: usize,
) -> Result<Harness, BootstrapError>
// run: many streams instead of one record iterator.
// before: run(&mut self, records: impl Iterator<Item = (Timestamp, Scalar)>)
// after:
pub fn run(&mut self, streams: Vec<Vec<(Timestamp, Scalar)>>) -> Vec<Option<Scalar>>
```
The `Harness` struct's `source_targets: Vec<Target>` field becomes
`sources: Vec<SourceSpec>`; 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).