feat(aura-runner): TapPlan + layered fold registry; run_signal_r re-wired onto the shared pair
The #283 registry model's assembly seam: tap_plan.rs carries TapSubscription (Named{label, params} — fully data-expressible — | Live(closure), the single non-data variant), TapPlan (name → subscription + default; the plan travels with the FoldRegistry it resolves against, the Env-at-resolve pattern, so layered hosts inject vocabulary as data), and the registry itself: eight core entries (record, count, sum, mean, min, max, first, last), each with a doc line (help + roster-enumerating refusals) and a scalar-typed param schema (all v1 entries param-less; the seam ships now because it sits in every entry's build signature). Typed TapPlanError refusals fire pre-bootstrap, before any store I/O; the terminal Store variant keeps the established "writing tap traces failed" register bytes. bind_tap_plan/BoundTaps is the shared wiring pair both declared-tap entry points call; run_signal_r is re-wired onto it here (Recorder bind + post-run try_iter drain replaced — no O(cycles) retention, no per-cycle heap), with all six callers threaded (both CLI run-arm and the four example-identity test calls pass TapPlan::record_all()). run_measurement follows in the next commit. Registry growth stays a new Rust entry (C25); a layer-registered entry is pinned by the injection test, param validation by a schema-carrying test entry (missing/mismatched/unknown all refused with named messages). Verification: cargo build --workspace clean; aura-runner 85+ tests green incl. both example-identity anchors; tap_recording byte-compat anchors green (recorded traces byte-identical, tap-free writes nothing, duplicate-tap refusal unchanged); clippy -D warnings clean. refs #283 refs #77
This commit is contained in:
@@ -45,6 +45,7 @@ use aura_measurement::information_coefficient;
|
||||
// (a sibling module reaching it via `crate::blueprint_axis_probe_reopened`,
|
||||
// the crate-root re-export this `use` gives it), not from this module itself.
|
||||
use aura_runner::member::{blueprint_axis_probe, blueprint_axis_probe_reopened, run_signal_r, wrapped_bound_names, RunData};
|
||||
use aura_runner::TapPlan;
|
||||
#[cfg(test)]
|
||||
use aura_runner::member::{run_blueprint_member, wrap_r, SYNTHETIC_PIP_SIZE};
|
||||
// The family builders (blueprint sweep / walk-forward / MC), the shared
|
||||
@@ -1765,7 +1766,7 @@ fn dispatch_run(a: RunCmd, env: &aura_runner::project::Env) {
|
||||
None => Vec::new(),
|
||||
};
|
||||
let data = run_data_from(a.real.as_deref(), a.from, a.to);
|
||||
let report = run_signal_r(signal, ¶ms, data, a.seed.unwrap_or(0), env);
|
||||
let report = run_signal_r(signal, ¶ms, data, a.seed.unwrap_or(0), env, TapPlan::record_all());
|
||||
println!("{}", report.to_json());
|
||||
} else if has_tap {
|
||||
// Measurement path: wrap_r-free closed guard via the signal's own
|
||||
@@ -3099,6 +3100,7 @@ mod tests {
|
||||
RunData::Synthetic,
|
||||
0,
|
||||
&env,
|
||||
TapPlan::record_all(),
|
||||
);
|
||||
let member4 = &family.points[0].report; // slow=4 is the first odometer point
|
||||
assert_eq!(member4.metrics, single.metrics, "loaded sweep member == single run");
|
||||
|
||||
@@ -18,11 +18,16 @@ pub mod member;
|
||||
pub mod project;
|
||||
pub mod reproduce;
|
||||
pub mod runner;
|
||||
pub mod tap_plan;
|
||||
pub mod tap_recorder;
|
||||
pub mod translate;
|
||||
|
||||
pub use project::Env;
|
||||
pub use runner::DefaultMemberRunner;
|
||||
pub use tap_plan::{
|
||||
bind_tap_plan, BoundTaps, FoldBuildCtx, FoldEntry, FoldOutput, FoldRegistry, FoldSink, TapPlan,
|
||||
TapPlanError, TapSubscription,
|
||||
};
|
||||
pub use tap_recorder::TapRecorder;
|
||||
|
||||
/// A refusal a library function reports instead of exiting the process
|
||||
|
||||
@@ -9,14 +9,14 @@
|
||||
//! `CliMemberRunner`, and any downstream World program) drive real data
|
||||
//! through.
|
||||
|
||||
use std::collections::{BTreeMap, BTreeSet, HashSet};
|
||||
use std::collections::{BTreeMap, HashSet};
|
||||
use std::sync::{mpsc, Arc, LazyLock};
|
||||
|
||||
use aura_composites::{cost_graph, risk_executor, StopRule};
|
||||
use aura_core::{zip_params, Cell, Firing, ParamSpec, PrimitiveBuilder, Scalar, ScalarKind, Timestamp};
|
||||
use aura_engine::{
|
||||
blueprint_from_json, window_of, BlueprintNode, ColumnarTrace, Composite, GraphBuilder, Harness,
|
||||
RunManifest, VecSource,
|
||||
blueprint_from_json, window_of, BlueprintNode, Composite, GraphBuilder, Harness, RunManifest,
|
||||
VecSource,
|
||||
};
|
||||
use aura_backtest::{
|
||||
summarize, summarize_r, RunMetrics, RunReport, SimBroker, PM_FIELD_NAMES, PM_RECORD_KINDS,
|
||||
@@ -26,6 +26,7 @@ use aura_strategy::{cost_port, GEOMETRY_WIDTH};
|
||||
|
||||
use crate::binding::ResolvedBinding;
|
||||
use crate::project::Env;
|
||||
use crate::tap_plan::{bind_tap_plan, TapPlan};
|
||||
use crate::translate::{R_SMA_STOP_LENGTH, R_SMA_STOP_K};
|
||||
|
||||
/// The single build-time commit provenance (`option_env!("AURA_COMMIT")`,
|
||||
@@ -480,6 +481,7 @@ pub fn resolve_run_data(
|
||||
#[allow(clippy::type_complexity)]
|
||||
pub fn run_signal_r(
|
||||
signal: Composite, params: &[Scalar], data: RunData, seed: u64, env: &Env,
|
||||
plan: TapPlan,
|
||||
) -> RunReport {
|
||||
// topology_hash's own two-line body, inlined: `content_id_of` over the
|
||||
// canonical (#164) blueprint JSON — the CLI shell's `topology_hash`
|
||||
@@ -517,25 +519,14 @@ pub fn run_signal_r(
|
||||
eprintln!("aura: this blueprint does not compile to a runnable harness: {e:?}");
|
||||
std::process::exit(1);
|
||||
});
|
||||
// Bind each declared tap to a fresh `Recorder` + channel, before bootstrap
|
||||
// — `flat.taps` already carries the signal's declared taps hoisted to the
|
||||
// root. Dedup is the caller's per `TapBindError::DuplicateBind`'s doc (the
|
||||
// engine keeps no cross-call state).
|
||||
let mut seen: BTreeSet<String> = BTreeSet::new();
|
||||
let mut tap_drains: Vec<(String, ScalarKind, mpsc::Receiver<(Timestamp, Vec<Scalar>)>)> = Vec::new();
|
||||
let declared: Vec<aura_engine::FlatTap> = flat.taps.clone();
|
||||
for tap in &declared {
|
||||
if !seen.insert(tap.name.clone()) {
|
||||
eprintln!("aura: {}", aura_engine::TapBindError::DuplicateBind { name: tap.name.clone() });
|
||||
std::process::exit(1);
|
||||
}
|
||||
let kind = flat.signatures[tap.node].output[tap.field].kind;
|
||||
let (tx, rx) = mpsc::channel();
|
||||
let sig = Recorder::builder(vec![kind], Firing::Any, tx.clone()).schema().clone();
|
||||
flat.bind_tap(&tap.name, Box::new(Recorder::new(&[kind], Firing::Any, tx)), sig)
|
||||
.expect("declared tap binds (name from flat.taps, kind from its own signature)");
|
||||
tap_drains.push((tap.name.clone(), kind, rx));
|
||||
}
|
||||
// Bind each declared tap per the plan's subscription, before bootstrap
|
||||
// (#283): typed refusals for bad plans, record consumers hold their
|
||||
// streaming writer in-graph, folds accumulate O(1), live closures run
|
||||
// inline. Dedup stays caller-owned per TapBindError::DuplicateBind's doc.
|
||||
let bound = bind_tap_plan(&mut flat, plan, env, &run_name).unwrap_or_else(|e| {
|
||||
eprintln!("aura: {e}");
|
||||
std::process::exit(1);
|
||||
});
|
||||
let mut h = Harness::bootstrap(flat).expect("valid r-sma harness");
|
||||
// `sources` were opened via `resolve_run_data(&data, env, &binding)` against
|
||||
// this SAME `binding`, and `key_supply` keys them by that binding's own role
|
||||
@@ -557,23 +548,13 @@ pub fn run_signal_r(
|
||||
manifest.project = env.provenance();
|
||||
let mut metrics = summarize(&aura_engine::f64_field(&eq_rows, 0), &aura_engine::f64_field(&ex_rows, 0));
|
||||
metrics.r = Some(summarize_r(&r_rows, &[]));
|
||||
// Drain + persist each declared tap's series, guarded so a tap-free
|
||||
// `aura run` stays byte-identical to today (no `runs/` write at all).
|
||||
// `manifest` is built above so the persisted `index.json` carries this
|
||||
// run's own provenance.
|
||||
if !tap_drains.is_empty() {
|
||||
let tap_traces: Vec<ColumnarTrace> = tap_drains
|
||||
.into_iter()
|
||||
.map(|(name, kind, rx)| {
|
||||
let rows: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
|
||||
ColumnarTrace::from_rows(&name, &[kind], &rows)
|
||||
})
|
||||
.collect();
|
||||
env.trace_store().write(&run_name, &manifest, &tap_traces).unwrap_or_else(|e| {
|
||||
eprintln!("aura: writing tap traces failed: {e}");
|
||||
std::process::exit(1);
|
||||
});
|
||||
}
|
||||
// Close the tap plan: drain the ≤1-message channels, write fold rows,
|
||||
// then `index.json` — nothing was buffered during the run (#283). A
|
||||
// tap-free (or nothing-persisting) plan wrote nothing at all.
|
||||
bound.finish(&manifest).unwrap_or_else(|e| {
|
||||
eprintln!("aura: {e}");
|
||||
std::process::exit(1);
|
||||
});
|
||||
RunReport { manifest, metrics }
|
||||
}
|
||||
|
||||
@@ -1158,8 +1139,8 @@ mod tests {
|
||||
let env = Env::std();
|
||||
let loaded = blueprint_from_json(include_str!("../../aura-cli/examples/r_breakout.json"), &|t| std_vocabulary(t))
|
||||
.expect("shipped r_breakout example loads");
|
||||
let via_file = run_signal_r(loaded, &[], RunData::Synthetic, 0, &env);
|
||||
let via_carve = run_signal_r(r_breakout_signal(Some(R_BREAKOUT_CHANNEL)), &[], RunData::Synthetic, 0, &env);
|
||||
let via_file = run_signal_r(loaded, &[], RunData::Synthetic, 0, &env, TapPlan::record_all());
|
||||
let via_carve = run_signal_r(r_breakout_signal(Some(R_BREAKOUT_CHANNEL)), &[], RunData::Synthetic, 0, &env, TapPlan::record_all());
|
||||
assert_eq!(via_file.metrics, via_carve.metrics, "loaded example grades identically to the carve");
|
||||
}
|
||||
|
||||
@@ -1311,13 +1292,14 @@ mod tests {
|
||||
let env = Env::std();
|
||||
let loaded = blueprint_from_json(include_str!("../../aura-cli/examples/r_meanrev.json"), &|t| std_vocabulary(t))
|
||||
.expect("shipped r_meanrev example loads");
|
||||
let via_file = run_signal_r(loaded, &[], RunData::Synthetic, 0, &env);
|
||||
let via_file = run_signal_r(loaded, &[], RunData::Synthetic, 0, &env, TapPlan::record_all());
|
||||
let via_carve = run_signal_r(
|
||||
r_meanrev_signal(Some(R_MEANREV_WINDOW), Some(R_MEANREV_BAND_K)),
|
||||
&[],
|
||||
RunData::Synthetic,
|
||||
0,
|
||||
&env,
|
||||
TapPlan::record_all(),
|
||||
);
|
||||
assert_eq!(via_file.metrics, via_carve.metrics, "loaded example grades identically to the carve");
|
||||
}
|
||||
|
||||
@@ -0,0 +1,639 @@
|
||||
//! The per-run tap subscription plan (#283, 2026-07-21 design revision) and
|
||||
//! the layered fold registry it resolves against. A tap stays a pure
|
||||
//! declaration (C27); this module is where the run/measurement side declares
|
||||
//! what consumes it. Record, fold, and live are ONE mechanism — a consumer
|
||||
//! of the tap's `(Timestamp, Cell)` stream; the only real axis is data-layer
|
||||
//! serializability: a `Named` subscription (label + scalar params) is fully
|
||||
//! expressible as data, a `Live` closure is the single deliberately non-data
|
||||
//! variant. The registry mirrors the node-vocabulary pattern
|
||||
//! (`std_vocabulary` + `Env::resolve`): a closed, typed vocabulary whose
|
||||
//! growth is a new Rust entry (C25) — core seeds here, higher layers
|
||||
//! (e.g. trading) register their own entries without bleeding into the core.
|
||||
|
||||
use std::collections::{BTreeMap, BTreeSet};
|
||||
use std::fmt;
|
||||
use std::sync::mpsc::{self, Receiver, Sender};
|
||||
|
||||
use aura_core::{Cell, Firing, Node, NodeSchema, ParamSpec, PortSpec, Scalar, ScalarKind, Timestamp};
|
||||
use aura_engine::{ColumnarTrace, FlatGraph, RunManifest, TapBindError};
|
||||
use aura_registry::{TapWriterOpener, TraceStoreError, TraceStreamer};
|
||||
use aura_std::{fold_binds_at, fold_output_kind, FoldKind, TapFold, TapLive};
|
||||
|
||||
use crate::project::Env;
|
||||
use crate::tap_recorder::TapRecorder;
|
||||
|
||||
/// What consumes one declared tap during a run — the drain policy, declared
|
||||
/// on the run/measurement side (#283), never on the tap.
|
||||
pub enum TapSubscription {
|
||||
/// A consumer from the plan's fold registry, by label, with scalar-typed
|
||||
/// construction parameters (v1 core entries take none).
|
||||
Named { label: String, params: Vec<(String, Scalar)> },
|
||||
/// An in-process consumer; loss policy is the closure's own.
|
||||
Live(Box<dyn FnMut(Timestamp, Cell) + Send>),
|
||||
}
|
||||
|
||||
impl TapSubscription {
|
||||
/// A named subscription with no construction parameters.
|
||||
pub fn named(label: &str) -> Self {
|
||||
TapSubscription::Named { label: label.to_string(), params: Vec::new() }
|
||||
}
|
||||
|
||||
/// A named subscription with construction parameters.
|
||||
pub fn named_with(label: &str, params: Vec<(String, Scalar)>) -> Self {
|
||||
TapSubscription::Named { label: label.to_string(), params }
|
||||
}
|
||||
|
||||
/// Sugar for a live subscription around a closure.
|
||||
pub fn live(consumer: impl FnMut(Timestamp, Cell) + Send + 'static) -> Self {
|
||||
TapSubscription::Live(Box::new(consumer))
|
||||
}
|
||||
}
|
||||
|
||||
/// Per-run tap plan: name → subscription; unnamed taps get the default. The
|
||||
/// plan carries the registry it resolves against (the `Env`-at-resolve
|
||||
/// pattern), so a layered host injects its extended vocabulary as data. The
|
||||
/// only defaults are `named("record")` (`record_all` — the one entry that
|
||||
/// binds at every kind and takes no params) and `None` (empty plan; unnamed
|
||||
/// taps stay unbound, inert per C27) — fold/live subscriptions are per-tap,
|
||||
/// never a default, structurally (the field is private and no constructor
|
||||
/// sets another).
|
||||
pub struct TapPlan {
|
||||
pub(crate) registry: FoldRegistry,
|
||||
pub(crate) default_named: Option<(String, Vec<(String, Scalar)>)>,
|
||||
pub(crate) by_name: BTreeMap<String, TapSubscription>,
|
||||
}
|
||||
|
||||
impl TapPlan {
|
||||
/// Record every declared tap — the CLI's plan on both verbs.
|
||||
pub fn record_all() -> Self {
|
||||
Self::record_all_with(FoldRegistry::core())
|
||||
}
|
||||
|
||||
/// `record_all` over an injected (layered) registry.
|
||||
pub fn record_all_with(registry: FoldRegistry) -> Self {
|
||||
TapPlan {
|
||||
registry,
|
||||
default_named: Some(("record".to_string(), Vec::new())),
|
||||
by_name: BTreeMap::new(),
|
||||
}
|
||||
}
|
||||
|
||||
/// No default: only explicitly subscribed taps are consumed.
|
||||
pub fn empty() -> Self {
|
||||
TapPlan { registry: FoldRegistry::core(), default_named: None, by_name: BTreeMap::new() }
|
||||
}
|
||||
|
||||
/// Subscribe one tap by name (replaces an earlier subscription for it).
|
||||
pub fn subscribe(&mut self, tap: &str, sub: TapSubscription) {
|
||||
self.by_name.insert(tap.to_string(), sub);
|
||||
}
|
||||
}
|
||||
|
||||
/// Persistence shape of an entry's consumer.
|
||||
pub enum FoldOutput {
|
||||
/// The full series, streamed (the `record` entry).
|
||||
Series,
|
||||
/// One summary row at finalize, of this kind.
|
||||
Row(ScalarKind),
|
||||
}
|
||||
|
||||
/// The sink a build call receives — matching the entry's declared
|
||||
/// [`FoldOutput`]: `Row` entries get the one-row channel, `Series` entries
|
||||
/// the deferred writer opener + terminal-outcome channel.
|
||||
pub enum FoldSink {
|
||||
Row(Sender<(Timestamp, Vec<Scalar>)>),
|
||||
Series(TapWriterOpener, Sender<Result<(), TraceStoreError>>),
|
||||
}
|
||||
|
||||
/// What the wiring hands a registry entry's `build`.
|
||||
pub struct FoldBuildCtx {
|
||||
/// The tap's column kind (already `binds_at`-validated).
|
||||
pub kind: ScalarKind,
|
||||
/// Construction parameters, already validated against the entry's schema.
|
||||
pub params: Vec<(String, Scalar)>,
|
||||
pub sink: FoldSink,
|
||||
}
|
||||
|
||||
/// One registry entry: a labelled consumer constructor with docs, a param
|
||||
/// schema, bind rules, and its persistence shape.
|
||||
pub struct FoldEntry {
|
||||
pub label: &'static str,
|
||||
/// One line for help generation and roster-enumerating refusals.
|
||||
pub doc: &'static str,
|
||||
/// Scalar-typed construction parameters (the node vocabulary's param
|
||||
/// plane). All v1 core entries: empty — the seam ships now because it
|
||||
/// sits in every entry's build signature (C25: a param that would carry
|
||||
/// logic is a new entry, never a freetext hole).
|
||||
pub params: Vec<ParamSpec>,
|
||||
/// May this entry bind a tap column of this kind?
|
||||
pub binds_at: Box<dyn Fn(ScalarKind) -> bool>,
|
||||
/// What the consumer persists, given the tap's column kind.
|
||||
pub output: Box<dyn Fn(ScalarKind) -> FoldOutput>,
|
||||
/// Build the consumer node. `ctx.params` arrive validated; `ctx.sink`
|
||||
/// matches `output` (the wiring consults `output` first — an entry may
|
||||
/// `panic!` on the wrong variant as a wiring-contract violation).
|
||||
pub build: Box<dyn Fn(FoldBuildCtx) -> Box<dyn Node>>,
|
||||
}
|
||||
|
||||
/// The label → entry map. Layered: `core()` seeds the standard vocabulary;
|
||||
/// `register` adds (or deliberately shadows) an entry — dependency
|
||||
/// injection, no core edit.
|
||||
pub struct FoldRegistry {
|
||||
entries: BTreeMap<&'static str, FoldEntry>,
|
||||
}
|
||||
|
||||
impl FoldRegistry {
|
||||
/// The core vocabulary: `record`, `count`, `sum`, `mean`, `min`, `max`,
|
||||
/// `first`, `last`.
|
||||
pub fn core() -> Self {
|
||||
let mut r = FoldRegistry { entries: BTreeMap::new() };
|
||||
r.register(FoldEntry {
|
||||
label: "record",
|
||||
doc: "persist the full series, lossless, at constant memory (any kind)",
|
||||
params: Vec::new(),
|
||||
binds_at: Box::new(|_| true),
|
||||
output: Box::new(|_| FoldOutput::Series),
|
||||
build: Box::new(|ctx| {
|
||||
let FoldSink::Series(opener, outcome_tx) = ctx.sink else {
|
||||
panic!("wiring contract: record is a Series entry");
|
||||
};
|
||||
Box::new(TapRecorder::new(ctx.kind, opener, outcome_tx))
|
||||
}),
|
||||
});
|
||||
for (label, doc, fold) in [
|
||||
("count", "number of warm rows (any kind; i64 row)", FoldKind::Count),
|
||||
("sum", "sum of the series (f64 taps; f64 row)", FoldKind::Sum),
|
||||
("mean", "arithmetic mean of the series (f64 taps; f64 row)", FoldKind::Mean),
|
||||
("min", "minimum of the series (f64 taps; f64 row)", FoldKind::Min),
|
||||
("max", "maximum of the series (f64 taps; f64 row)", FoldKind::Max),
|
||||
(
|
||||
"first",
|
||||
"first warm value, at its own timestamp (any kind; kind-preserving row)",
|
||||
FoldKind::First,
|
||||
),
|
||||
("last", "last warm value (any kind; kind-preserving row)", FoldKind::Last),
|
||||
] {
|
||||
r.register(FoldEntry {
|
||||
label,
|
||||
doc,
|
||||
params: Vec::new(),
|
||||
binds_at: Box::new(move |k| fold_binds_at(fold, k)),
|
||||
output: Box::new(move |k| FoldOutput::Row(fold_output_kind(fold, k))),
|
||||
build: Box::new(move |ctx| {
|
||||
let FoldSink::Row(tx) = ctx.sink else {
|
||||
panic!("wiring contract: {fold:?} is a Row entry");
|
||||
};
|
||||
Box::new(TapFold::new(ctx.kind, fold, tx))
|
||||
}),
|
||||
});
|
||||
}
|
||||
r
|
||||
}
|
||||
|
||||
/// Add an entry. A later registration under an existing label replaces
|
||||
/// it (layers may deliberately shadow).
|
||||
pub fn register(&mut self, entry: FoldEntry) {
|
||||
self.entries.insert(entry.label, entry);
|
||||
}
|
||||
|
||||
pub fn get(&self, label: &str) -> Option<&FoldEntry> {
|
||||
self.entries.get(label)
|
||||
}
|
||||
|
||||
/// `(label, doc)` in label order — the help surface and the refusal
|
||||
/// roster.
|
||||
pub fn roster(&self) -> Vec<(&'static str, &'static str)> {
|
||||
self.entries.values().map(|e| (e.label, e.doc)).collect()
|
||||
}
|
||||
|
||||
/// The labels alone, label order (refusal messages).
|
||||
pub fn labels(&self) -> Vec<&'static str> {
|
||||
self.entries.keys().copied().collect()
|
||||
}
|
||||
}
|
||||
|
||||
/// A typed tap-plan fault — the pre-bootstrap refusals plus the terminal
|
||||
/// store fault. Entry points map every variant to the established
|
||||
/// `aura: ` + exit-1 refusal register via `Display`.
|
||||
pub enum TapPlanError {
|
||||
/// The plan names a tap the blueprint does not declare.
|
||||
UnknownTap { name: String },
|
||||
/// The plan names a label the registry does not carry.
|
||||
UnknownLabel { label: String, roster: Vec<&'static str> },
|
||||
/// The entry's bind rule rejects the tap's column kind.
|
||||
KindMismatch { tap: String, label: String, kind: ScalarKind },
|
||||
/// A param binding names no schema param (`takes` = the schema's names).
|
||||
UnknownParam { label: String, name: String, takes: Vec<String> },
|
||||
/// A schema param is unbound.
|
||||
MissingParam { label: String, name: String, kind: ScalarKind },
|
||||
/// A param binding's kind mismatches its schema declaration.
|
||||
ParamKind { label: String, name: String, expected: ScalarKind, got: ScalarKind },
|
||||
/// Two declared taps share a name (the caller-owned dedup guard).
|
||||
Bind(aura_engine::TapBindError),
|
||||
/// The trace store failed (begin/write/finish) — the terminal register.
|
||||
Store(TraceStoreError),
|
||||
}
|
||||
|
||||
impl fmt::Display for TapPlanError {
|
||||
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||||
match self {
|
||||
TapPlanError::UnknownTap { name } => {
|
||||
write!(f, "the tap plan names '{name}', but the blueprint declares no such tap")
|
||||
}
|
||||
TapPlanError::UnknownLabel { label, roster } => {
|
||||
write!(f, "unknown fold '{label}' — available: {}", roster.join(", "))
|
||||
}
|
||||
TapPlanError::KindMismatch { tap, label, kind } => {
|
||||
write!(f, "fold '{label}' cannot bind tap '{tap}' of kind {kind:?}")
|
||||
}
|
||||
TapPlanError::UnknownParam { label, name, takes } => {
|
||||
if takes.is_empty() {
|
||||
write!(f, "fold '{label}' takes no parameters (got '{name}')")
|
||||
} else {
|
||||
write!(f, "fold '{label}' has no parameter '{name}' — takes: {}", takes.join(", "))
|
||||
}
|
||||
}
|
||||
TapPlanError::MissingParam { label, name, kind } => {
|
||||
write!(f, "fold '{label}' requires parameter '{name}' ({kind:?})")
|
||||
}
|
||||
TapPlanError::ParamKind { label, name, expected, got } => {
|
||||
write!(f, "fold '{label}' parameter '{name}' is {expected:?}, got {got:?}")
|
||||
}
|
||||
TapPlanError::Bind(e) => write!(f, "{e}"),
|
||||
TapPlanError::Store(e) => write!(f, "writing tap traces failed: {e}"),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl fmt::Debug for TapPlanError {
|
||||
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||||
write!(f, "TapPlanError({self})")
|
||||
}
|
||||
}
|
||||
|
||||
impl From<TraceStoreError> for TapPlanError {
|
||||
fn from(e: TraceStoreError) -> Self {
|
||||
TapPlanError::Store(e)
|
||||
}
|
||||
}
|
||||
|
||||
/// Validate one `Named` subscription's param bindings against the entry's
|
||||
/// schema: every binding names a schema param of the right kind; every
|
||||
/// schema param is bound.
|
||||
pub(crate) fn validate_params(
|
||||
entry: &FoldEntry,
|
||||
bound: &[(String, Scalar)],
|
||||
) -> Result<(), TapPlanError> {
|
||||
let takes: Vec<String> = entry.params.iter().map(|p| p.name.clone()).collect();
|
||||
for (name, value) in bound {
|
||||
match entry.params.iter().find(|p| &p.name == name) {
|
||||
None => {
|
||||
return Err(TapPlanError::UnknownParam {
|
||||
label: entry.label.to_string(),
|
||||
name: name.clone(),
|
||||
takes,
|
||||
})
|
||||
}
|
||||
Some(p) if p.kind != value.kind() => {
|
||||
return Err(TapPlanError::ParamKind {
|
||||
label: entry.label.to_string(),
|
||||
name: name.clone(),
|
||||
expected: p.kind,
|
||||
got: value.kind(),
|
||||
})
|
||||
}
|
||||
Some(_) => {}
|
||||
}
|
||||
}
|
||||
for p in &entry.params {
|
||||
if !bound.iter().any(|(n, _)| n == &p.name) {
|
||||
return Err(TapPlanError::MissingParam {
|
||||
label: entry.label.to_string(),
|
||||
name: p.name.clone(),
|
||||
kind: p.kind,
|
||||
});
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// The one-input sink schema every tap consumer binds with (empty output =
|
||||
/// pure consumer, C8; the port name is a non-load-bearing debug symbol).
|
||||
fn tap_sink_schema(kind: ScalarKind) -> NodeSchema {
|
||||
NodeSchema {
|
||||
inputs: vec![PortSpec { kind, firing: Firing::Any, name: "in".to_string() }],
|
||||
output: vec![],
|
||||
params: vec![],
|
||||
}
|
||||
}
|
||||
|
||||
/// The bound half of a tap plan: what `bind_tap_plan` leaves for the caller
|
||||
/// to drain after the run. Both declared-tap entry points call the pair, so
|
||||
/// the mechanism cannot drift between them.
|
||||
#[allow(clippy::type_complexity)]
|
||||
pub struct BoundTaps {
|
||||
streamer: Option<TraceStreamer>,
|
||||
declared: Vec<String>,
|
||||
/// Persisting taps (Series and Row alike), declared-tap order — the
|
||||
/// `index.json` tap order.
|
||||
persisted: Vec<String>,
|
||||
rows: Vec<(String, ScalarKind, Receiver<(Timestamp, Vec<Scalar>)>)>,
|
||||
outcomes: Vec<(String, Receiver<Result<(), TraceStoreError>>)>,
|
||||
}
|
||||
|
||||
impl BoundTaps {
|
||||
/// Every declared tap, declared order (the measurement report's roster).
|
||||
pub fn declared_names(&self) -> &[String] {
|
||||
&self.declared
|
||||
}
|
||||
|
||||
/// Drain the ≤1-message-per-tap channels and close the run: record
|
||||
/// outcomes first (declared order), then fold rows written through the
|
||||
/// same streamer, then `index.json` last. Any fault returns before the
|
||||
/// index is written — the store's crash discipline (no `index.json` →
|
||||
/// treat-as-absent) covers the partial directory, exactly like today's
|
||||
/// failed `write`. A plan that persisted nothing began no streamer and
|
||||
/// writes nothing.
|
||||
pub fn finish(mut self, manifest: &RunManifest) -> Result<(), TapPlanError> {
|
||||
for (_, rx) in &self.outcomes {
|
||||
// finalize ran inside the harness's end-of-stream flush, so the
|
||||
// single outcome is already in the channel.
|
||||
if let Ok(Err(e)) = rx.try_recv() {
|
||||
return Err(TapPlanError::Store(e));
|
||||
}
|
||||
}
|
||||
let streamer = self.streamer.take();
|
||||
for (name, kind, rx) in self.rows {
|
||||
let rows: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect(); // ≤1
|
||||
let trace = ColumnarTrace::from_rows(&name, &[kind], &rows);
|
||||
streamer
|
||||
.as_ref()
|
||||
.expect("a Row tap implies a begun streamer")
|
||||
.write_full(&trace)?;
|
||||
}
|
||||
if let Some(s) = streamer {
|
||||
s.finish(manifest, &self.persisted)?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
/// Validate the plan against `flat.taps`, begin the streamed run write when
|
||||
/// anything persists, and bind one consumer per subscribed tap — the shared
|
||||
/// pre-bootstrap half of both declared-tap entry points. Refusals are typed
|
||||
/// and complete BEFORE any store I/O (a refused run never half-writes).
|
||||
pub fn bind_tap_plan(
|
||||
flat: &mut FlatGraph,
|
||||
mut plan: TapPlan,
|
||||
env: &Env,
|
||||
run_name: &str,
|
||||
) -> Result<BoundTaps, TapPlanError> {
|
||||
let declared_taps: Vec<aura_engine::FlatTap> = flat.taps.clone();
|
||||
|
||||
// Dedup guard (caller-owned per TapBindError::DuplicateBind's doc).
|
||||
let mut seen: BTreeSet<String> = BTreeSet::new();
|
||||
for tap in &declared_taps {
|
||||
if !seen.insert(tap.name.clone()) {
|
||||
return Err(TapPlanError::Bind(TapBindError::DuplicateBind { name: tap.name.clone() }));
|
||||
}
|
||||
}
|
||||
|
||||
// Unknown-tap guard: every plan name must be declared.
|
||||
for name in plan.by_name.keys() {
|
||||
if !declared_taps.iter().any(|t| &t.name == name) {
|
||||
return Err(TapPlanError::UnknownTap { name: name.clone() });
|
||||
}
|
||||
}
|
||||
|
||||
// Resolve each declared tap to its subscription; validate all Named
|
||||
// subscriptions fully before any store I/O.
|
||||
enum Resolved {
|
||||
Named { label: String, params: Vec<(String, Scalar)> },
|
||||
Live(Box<dyn FnMut(Timestamp, Cell) + Send>),
|
||||
}
|
||||
let mut resolved: Vec<(String, ScalarKind, Resolved)> = Vec::new();
|
||||
for tap in &declared_taps {
|
||||
let kind = flat.signatures[tap.node].output[tap.field].kind;
|
||||
let sub = match plan.by_name.remove(&tap.name) {
|
||||
Some(TapSubscription::Named { label, params }) => Resolved::Named { label, params },
|
||||
Some(TapSubscription::Live(consumer)) => Resolved::Live(consumer),
|
||||
None => match &plan.default_named {
|
||||
Some((label, params)) => {
|
||||
Resolved::Named { label: label.clone(), params: params.clone() }
|
||||
}
|
||||
None => continue, // unbound, inert (C27)
|
||||
},
|
||||
};
|
||||
if let Resolved::Named { label, params } = &sub {
|
||||
let entry = plan.registry.get(label).ok_or_else(|| TapPlanError::UnknownLabel {
|
||||
label: label.clone(),
|
||||
roster: plan.registry.labels(),
|
||||
})?;
|
||||
if !(entry.binds_at)(kind) {
|
||||
return Err(TapPlanError::KindMismatch {
|
||||
tap: tap.name.clone(),
|
||||
label: label.clone(),
|
||||
kind,
|
||||
});
|
||||
}
|
||||
validate_params(entry, params)?;
|
||||
}
|
||||
resolved.push((tap.name.clone(), kind, sub));
|
||||
}
|
||||
|
||||
// Anything persisting? Then begin the run (directory creation is the
|
||||
// pre-run refusal point).
|
||||
let persists = resolved.iter().any(|(_, _, s)| matches!(s, Resolved::Named { .. }));
|
||||
let streamer = if persists { Some(env.trace_store().begin_run(run_name)?) } else { None };
|
||||
|
||||
// Bind one consumer per resolved tap.
|
||||
let mut bound = BoundTaps {
|
||||
streamer,
|
||||
declared: declared_taps.iter().map(|t| t.name.clone()).collect(),
|
||||
persisted: Vec::new(),
|
||||
rows: Vec::new(),
|
||||
outcomes: Vec::new(),
|
||||
};
|
||||
for (name, kind, sub) in resolved {
|
||||
let node: Box<dyn Node> = match sub {
|
||||
Resolved::Named { label, params } => {
|
||||
let entry = plan.registry.get(&label).expect("validated above");
|
||||
bound.persisted.push(name.clone());
|
||||
match (entry.output)(kind) {
|
||||
FoldOutput::Series => {
|
||||
let opener = bound
|
||||
.streamer
|
||||
.as_ref()
|
||||
.expect("persisting tap implies a begun streamer")
|
||||
.register_tap(&name, kind);
|
||||
let (tx, rx) = mpsc::channel();
|
||||
bound.outcomes.push((name.clone(), rx));
|
||||
(entry.build)(FoldBuildCtx { kind, params, sink: FoldSink::Series(opener, tx) })
|
||||
}
|
||||
FoldOutput::Row(out_kind) => {
|
||||
let (tx, rx) = mpsc::channel();
|
||||
bound.rows.push((name.clone(), out_kind, rx));
|
||||
(entry.build)(FoldBuildCtx { kind, params, sink: FoldSink::Row(tx) })
|
||||
}
|
||||
}
|
||||
}
|
||||
Resolved::Live(consumer) => Box::new(TapLive::new(kind, consumer)),
|
||||
};
|
||||
flat.bind_tap(&name, node, tap_sink_schema(kind))
|
||||
.expect("declared tap binds (name from flat.taps, kind from its own signature)");
|
||||
}
|
||||
Ok(bound)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use aura_core::{AnyColumn, Ctx};
|
||||
use std::sync::mpsc;
|
||||
|
||||
#[test]
|
||||
fn roster_lists_the_eight_core_entries_with_docs() {
|
||||
let r = FoldRegistry::core();
|
||||
let labels: Vec<&str> = r.roster().iter().map(|(l, _)| *l).collect();
|
||||
// BTreeMap order — alphabetical.
|
||||
assert_eq!(
|
||||
labels,
|
||||
vec!["count", "first", "last", "max", "mean", "min", "record", "sum"]
|
||||
);
|
||||
assert!(r.roster().iter().all(|(_, doc)| !doc.is_empty()), "every entry documents itself");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn unknown_label_refusal_enumerates_the_roster() {
|
||||
let r = FoldRegistry::core();
|
||||
let e = TapPlanError::UnknownLabel { label: "p95".to_string(), roster: r.labels() };
|
||||
let msg = e.to_string();
|
||||
assert!(msg.starts_with("unknown fold 'p95'"), "{msg}");
|
||||
assert!(msg.contains("record") && msg.contains("mean"), "roster enumerated: {msg}");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn arithmetic_folds_bind_only_at_f64_record_and_count_anywhere() {
|
||||
let r = FoldRegistry::core();
|
||||
for label in ["sum", "mean", "min", "max"] {
|
||||
let e = r.get(label).expect(label);
|
||||
assert!((e.binds_at)(ScalarKind::F64), "{label} binds f64");
|
||||
assert!(!(e.binds_at)(ScalarKind::I64), "{label} refuses i64");
|
||||
assert!(!(e.binds_at)(ScalarKind::Bool), "{label} refuses bool");
|
||||
}
|
||||
for label in ["record", "count", "first", "last"] {
|
||||
let e = r.get(label).expect(label);
|
||||
for k in [ScalarKind::F64, ScalarKind::I64, ScalarKind::Bool, ScalarKind::Timestamp] {
|
||||
assert!((e.binds_at)(k), "{label} binds {k:?}");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_param_on_a_param_less_entry_is_refused() {
|
||||
let r = FoldRegistry::core();
|
||||
let e = validate_params(
|
||||
r.get("mean").expect("mean"),
|
||||
&[("q".to_string(), Scalar::f64(0.95))],
|
||||
)
|
||||
.expect_err("core entries take no params");
|
||||
assert_eq!(e.to_string(), "fold 'mean' takes no parameters (got 'q')");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_layer_registered_entry_resolves_and_builds() {
|
||||
// The dependency-injection proof without a trading layer: a test-only
|
||||
// entry registered on top of core resolves by label and its consumer
|
||||
// runs (here: a re-labelled count over the committed TapFold core).
|
||||
let mut r = FoldRegistry::core();
|
||||
r.register(FoldEntry {
|
||||
label: "tally",
|
||||
doc: "test-only: warm-row count under a layer label",
|
||||
params: Vec::new(),
|
||||
binds_at: Box::new(|_| true),
|
||||
output: Box::new(|_| FoldOutput::Row(ScalarKind::I64)),
|
||||
build: Box::new(|ctx| {
|
||||
let FoldSink::Row(tx) = ctx.sink else {
|
||||
panic!("wiring contract: tally is a Row entry");
|
||||
};
|
||||
Box::new(TapFold::new(ctx.kind, FoldKind::Count, tx))
|
||||
}),
|
||||
});
|
||||
assert_eq!(r.labels().len(), 9, "core plus the layer entry");
|
||||
let entry = r.get("tally").expect("layer entry resolves");
|
||||
|
||||
let (tx, rx) = mpsc::channel();
|
||||
let mut node = (entry.build)(FoldBuildCtx {
|
||||
kind: ScalarKind::F64,
|
||||
params: Vec::new(),
|
||||
sink: FoldSink::Row(tx),
|
||||
});
|
||||
let mut inputs = vec![AnyColumn::with_capacity(ScalarKind::F64, 1)];
|
||||
node.initialize();
|
||||
for (t, v) in [(1_i64, 5.0_f64), (2, 7.0)] {
|
||||
inputs[0].push(Scalar::f64(v)).unwrap();
|
||||
assert_eq!(node.eval(Ctx::new(&inputs, Timestamp(t))), None);
|
||||
}
|
||||
node.finalize();
|
||||
let rows: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
|
||||
assert_eq!(rows, vec![(Timestamp(2), vec![Scalar::i64(2)])]);
|
||||
}
|
||||
|
||||
/// Exercises the `record` entry's Series-sink build path (destructure
|
||||
/// `FoldSink::Series` -> construct `TapRecorder`), left uncovered by
|
||||
/// `a_layer_registered_entry_resolves_and_builds` above, which only
|
||||
/// drives a Row-sink entry.
|
||||
#[test]
|
||||
fn record_entry_builds_a_series_sink_that_persists_the_stream() {
|
||||
let root = std::env::temp_dir()
|
||||
.join(format!("aura-runner-tap_plan-record-{}", std::process::id()));
|
||||
let _ = std::fs::remove_dir_all(&root);
|
||||
std::fs::create_dir_all(&root).expect("create temp store root");
|
||||
let store = aura_registry::TraceStore::open(&root);
|
||||
let streamer = store.begin_run("plan").expect("begin run");
|
||||
let opener = streamer.register_tap("t", ScalarKind::F64);
|
||||
|
||||
let r = FoldRegistry::core();
|
||||
let entry = r.get("record").expect("record");
|
||||
let (outcome_tx, outcome_rx) = mpsc::channel();
|
||||
let mut node = (entry.build)(FoldBuildCtx {
|
||||
kind: ScalarKind::F64,
|
||||
params: Vec::new(),
|
||||
sink: FoldSink::Series(opener, outcome_tx),
|
||||
});
|
||||
let mut inputs = vec![AnyColumn::with_capacity(ScalarKind::F64, 1)];
|
||||
node.initialize();
|
||||
for (t, v) in [(1_i64, 5.0_f64), (2, 7.0)] {
|
||||
inputs[0].push(Scalar::f64(v)).unwrap();
|
||||
assert_eq!(node.eval(Ctx::new(&inputs, Timestamp(t))), None);
|
||||
}
|
||||
node.finalize();
|
||||
|
||||
let outcomes: Vec<Result<(), TraceStoreError>> = outcome_rx.try_iter().collect();
|
||||
assert_eq!(outcomes.len(), 1, "exactly one terminal outcome");
|
||||
assert!(outcomes[0].is_ok(), "record's Series build path writes cleanly: {outcomes:?}");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn schema_params_are_validated_missing_mismatched_and_unknown() {
|
||||
// A schema-carrying test entry reaches the validation branches v1's
|
||||
// param-less core entries cannot: missing, kind-mismatched, and
|
||||
// unknown-against-a-non-empty-schema.
|
||||
let entry = FoldEntry {
|
||||
label: "windowed",
|
||||
doc: "test-only: carries a param schema",
|
||||
params: vec![ParamSpec { name: "length".to_string(), kind: ScalarKind::I64 }],
|
||||
binds_at: Box::new(|_| true),
|
||||
output: Box::new(|_| FoldOutput::Row(ScalarKind::F64)),
|
||||
build: Box::new(|_| panic!("validation-only entry is never built")),
|
||||
};
|
||||
let missing = validate_params(&entry, &[]).expect_err("length is required");
|
||||
assert_eq!(missing.to_string(), "fold 'windowed' requires parameter 'length' (I64)");
|
||||
let mismatched = validate_params(&entry, &[("length".to_string(), Scalar::f64(2.0))])
|
||||
.expect_err("kind mismatch");
|
||||
assert_eq!(mismatched.to_string(), "fold 'windowed' parameter 'length' is I64, got F64");
|
||||
let unknown = validate_params(&entry, &[("q".to_string(), Scalar::i64(1))])
|
||||
.expect_err("unknown param");
|
||||
assert_eq!(unknown.to_string(), "fold 'windowed' has no parameter 'q' — takes: length");
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user