feat(engine): FlatGraph::bind_tap + TapBindError (#282)
The run-mode-aware bind seam. bind_tap attaches a CALLER-BUILT Box<dyn Node> sink to a declared tap — appending the sink node + an edge from the tap's (node, field) to its input, before bootstrap so the Kahn sort orders it. The engine constructs no aura-std type (it takes a Box<dyn Node>), keeping its production dependency aura-core-only. bind_tap raises UndeclaredTap; duplicate detection across binds is the caller's (the engine keeps no cross-call state), mirroring how bind_sources' DuplicateFeed is a supply-level fault. TapBindError is exported for the CLI. A bound tap records exactly the producer's per-cycle output through the full public pipeline; an undeclared name is refused without mutating the graph. refs #282
This commit is contained in:
@@ -312,6 +312,62 @@ pub struct FlatGraph {
|
||||
pub taps: Vec<FlatTap>,
|
||||
}
|
||||
|
||||
/// A tap-binding fault caught before the run. The output-side twin of
|
||||
/// `SourceBindError`. `UndeclaredTap` is raised by `bind_tap` itself;
|
||||
/// `DuplicateBind` is raised by the **caller** (the run-mode-owning layer) when
|
||||
/// it dedups a tap name across binds — the engine method binds a single tap and
|
||||
/// keeps no cross-call state, so duplicate detection is the caller's, exactly as
|
||||
/// `bind_sources`' `DuplicateFeed` is a supply-level, not a per-source, fault.
|
||||
#[derive(Debug, PartialEq, Eq)]
|
||||
pub enum TapBindError {
|
||||
/// A binding names a tap the graph does not declare (raised by `bind_tap`).
|
||||
UndeclaredTap { name: String },
|
||||
/// The same tap name was bound more than once (raised by the caller dedup).
|
||||
DuplicateBind { name: String },
|
||||
}
|
||||
|
||||
impl core::fmt::Display for TapBindError {
|
||||
fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
|
||||
match self {
|
||||
TapBindError::UndeclaredTap { name } => {
|
||||
write!(f, "bind_tap names tap \"{name}\" but the graph declares no such tap")
|
||||
}
|
||||
TapBindError::DuplicateBind { name } => {
|
||||
write!(f, "tap \"{name}\" was bound more than once")
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl FlatGraph {
|
||||
/// Attach a caller-built sink to a declared tap: append `sink` + `sig` to the
|
||||
/// flat node/signature arrays and an `Edge` from the tap's `(node, field)` to
|
||||
/// the sink's input slot 0. Constructs no sink itself (the engine has no
|
||||
/// `aura-std` production dep) — the caller hands in a `Recorder`. Call BEFORE
|
||||
/// `bootstrap`; the Kahn sort then orders the appended sink. Binding a tap
|
||||
/// the graph does not declare is a typed `UndeclaredTap` refusal; duplicate
|
||||
/// detection across binds is the caller's (this method keeps no cross-call
|
||||
/// state — see `TapBindError`).
|
||||
pub fn bind_tap(
|
||||
&mut self,
|
||||
tap_name: &str,
|
||||
sink: Box<dyn Node>,
|
||||
sig: NodeSchema,
|
||||
) -> Result<(), TapBindError> {
|
||||
let tap = self
|
||||
.taps
|
||||
.iter()
|
||||
.find(|t| t.name == tap_name)
|
||||
.ok_or_else(|| TapBindError::UndeclaredTap { name: tap_name.to_string() })?;
|
||||
let (from, from_field) = (tap.node, tap.field);
|
||||
let sink_idx = self.nodes.len();
|
||||
self.nodes.push(sink);
|
||||
self.signatures.push(sig);
|
||||
self.edges.push(Edge { from, to: sink_idx, slot: 0, from_field });
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
/// A wiring fault caught once, at bootstrap — C7's "type check paid at wiring"
|
||||
/// generalized to the whole topology.
|
||||
#[derive(Debug, PartialEq, Eq)]
|
||||
@@ -2774,4 +2830,44 @@ mod tests {
|
||||
fired.sort_unstable();
|
||||
assert_eq!(fired, vec![0, 1]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn bind_tap_attaches_a_caller_built_sink_and_records_the_tapped_series() {
|
||||
// A tap on a producer; bind_tap attaches a caller-built Recorder; the run
|
||||
// records exactly the producer's per-cycle output. An undeclared tap name
|
||||
// is a typed refusal.
|
||||
let prices = f64_stream(&[(1, 10.0), (2, 20.0), (3, 30.0)]);
|
||||
// graph: source -> Sma(1) [node 0, producer], declared as tap "d" on (0,0).
|
||||
let mut flat = FlatGraph {
|
||||
nodes: vec![Box::new(Sma::new(1))],
|
||||
signatures: vec![Sma::builder().schema().clone()],
|
||||
sources: vec![SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot: 0 }])],
|
||||
edges: vec![],
|
||||
taps: vec![FlatTap { name: "d".into(), node: 0, field: 0 }],
|
||||
};
|
||||
// undeclared name refused
|
||||
let (tx_bad, _rx_bad) = mpsc::channel();
|
||||
assert!(matches!(
|
||||
flat.bind_tap("nope", Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx_bad)),
|
||||
recorder_sig(&[ScalarKind::F64], Firing::Any)),
|
||||
Err(TapBindError::UndeclaredTap { .. })
|
||||
));
|
||||
// declared name binds
|
||||
let (tx, rx) = mpsc::channel();
|
||||
flat.bind_tap("d", Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)),
|
||||
recorder_sig(&[ScalarKind::F64], Firing::Any))
|
||||
.expect("declared tap binds");
|
||||
let mut h = Harness::bootstrap(flat).expect("bootstraps with the appended recorder");
|
||||
h.run(vec![Box::new(VecSource::new(prices))]);
|
||||
let recorded: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
|
||||
assert_eq!(
|
||||
recorded,
|
||||
vec![
|
||||
(Timestamp(1), vec![Scalar::f64(10.0)]),
|
||||
(Timestamp(2), vec![Scalar::f64(20.0)]),
|
||||
(Timestamp(3), vec![Scalar::f64(30.0)]),
|
||||
],
|
||||
"the bound tap records exactly the producer's per-cycle output"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -66,7 +66,7 @@ pub use construction::{replay, GraphSession, Op, OpError};
|
||||
pub use graph_model::model_to_json;
|
||||
pub use harness::{
|
||||
window_of, bind_sources, BootstrapError, Edge, FlatGraph, FlatTap, Harness, Source, SourceBindError,
|
||||
SourceSpec, SplitMix64, SyntheticSpec, Target, VecSource,
|
||||
SourceSpec, SplitMix64, SyntheticSpec, TapBindError, Target, VecSource,
|
||||
};
|
||||
pub use report::{
|
||||
derive_position_events, expected_max_of_normals, f64_field, inv_norm_cdf, join_on_ts,
|
||||
|
||||
@@ -0,0 +1,111 @@
|
||||
//! End-to-end coverage for `FlatGraph::bind_tap` (#282, task 5): the caller-side
|
||||
//! half of declared taps. `flat_taps_e2e.rs` pins that `compile()` resolves a
|
||||
//! declared `Tap` into `FlatGraph.taps`; these tests pin the other half — that
|
||||
//! `bind_tap` actually wires a caller-built sink onto the resolved wire and that
|
||||
//! it runs through the FULL public pipeline (`Composite::with_taps` ->
|
||||
//! `compile()` -> `bind_tap` -> `Harness::bootstrap` -> `run`), not just the
|
||||
//! hand-built `FlatGraph` the in-crate unit test exercises.
|
||||
|
||||
use std::sync::mpsc;
|
||||
|
||||
use aura_engine::{
|
||||
Composite, Firing, Harness, NodeSchema, PortSpec, Role, ScalarKind, Scalar, Tap, TapWire,
|
||||
Target, Timestamp, VecSource,
|
||||
};
|
||||
use aura_std::{Recorder, Sma};
|
||||
|
||||
/// A single-node composite: one root-bound `price` role feeds `Sma(length=1)`
|
||||
/// (an identity passthrough), declaring `taps` on its one producer (node 0,
|
||||
/// field 0). No `OutField` is exported — the tap is the composite's only
|
||||
/// measurement surface, matching the "declared tap on an interior producer
|
||||
/// with no consuming edge" shape #282 targets.
|
||||
fn producer(taps: Vec<Tap>) -> Composite {
|
||||
Composite::new(
|
||||
"producer",
|
||||
vec![Sma::builder().bind("length", Scalar::i64(1)).into()],
|
||||
vec![],
|
||||
vec![Role { name: "price".into(), targets: vec![Target { node: 0, slot: 0 }], source: Some(ScalarKind::F64) }],
|
||||
vec![],
|
||||
)
|
||||
.with_taps(taps)
|
||||
}
|
||||
|
||||
/// The `NodeSchema` a `Recorder::new(kinds, firing, tx)` sink satisfies — the
|
||||
/// caller-side construction contract `bind_tap` requires alongside the boxed
|
||||
/// node (mirrors `aura_std::Recorder::builder`'s schema, built here directly
|
||||
/// since `bind_tap` takes a bare `Box<dyn Node>`, not a `PrimitiveBuilder`).
|
||||
fn recorder_sig(kinds: &[ScalarKind]) -> NodeSchema {
|
||||
NodeSchema {
|
||||
inputs: kinds
|
||||
.iter()
|
||||
.enumerate()
|
||||
.map(|(i, &kind)| PortSpec { kind, firing: Firing::Any, name: format!("col[{i}]") })
|
||||
.collect(),
|
||||
output: vec![],
|
||||
params: vec![],
|
||||
}
|
||||
}
|
||||
|
||||
fn prices(values: &[f64]) -> Vec<(Timestamp, Scalar)> {
|
||||
values.iter().enumerate().map(|(i, &v)| (Timestamp(i as i64 + 1), Scalar::f64(v))).collect()
|
||||
}
|
||||
|
||||
/// Property: **`bind_tap` wires a caller-built sink onto the exact flat
|
||||
/// producer a declared `Tap` resolved to, all the way through the public
|
||||
/// `Composite::with_taps -> compile() -> bind_tap -> Harness::bootstrap ->
|
||||
/// run` pipeline.** The recorded trace must be exactly the tapped producer's
|
||||
/// per-cycle output — not empty (which a wire pointed at the wrong node/field,
|
||||
/// or a bind that silently no-ops, would also produce for an unrelated
|
||||
/// column), and not some other node's output.
|
||||
#[test]
|
||||
fn bind_tap_records_the_tapped_producers_output_through_the_public_pipeline() {
|
||||
let mut flat = producer(vec![Tap { name: "d".into(), from: TapWire { node: 0, field: 0 } }])
|
||||
.compile()
|
||||
.expect("a tap-bearing composite compiles");
|
||||
|
||||
let (tx, rx) = mpsc::channel();
|
||||
flat.bind_tap(
|
||||
"d",
|
||||
Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)),
|
||||
recorder_sig(&[ScalarKind::F64]),
|
||||
)
|
||||
.expect("the declared tap name binds");
|
||||
|
||||
let mut h = Harness::bootstrap(flat).expect("bootstraps with the appended tap sink");
|
||||
h.run(vec![Box::new(VecSource::new(prices(&[10.0, 20.0, 30.0])))]);
|
||||
|
||||
let recorded: Vec<f64> = rx.try_iter().map(|(_, row)| row[0].as_f64()).collect();
|
||||
assert_eq!(
|
||||
recorded,
|
||||
vec![10.0, 20.0, 30.0],
|
||||
"the bound tap must record exactly the tapped producer's per-cycle output"
|
||||
);
|
||||
}
|
||||
|
||||
/// Property: **`bind_tap` refuses a name the graph did not declare as a tap,
|
||||
/// and does so without mutating the graph** — the caller can retry with the
|
||||
/// correct name rather than being left with a half-attached sink or a
|
||||
/// silently-corrupted node list. `TapBindError` is not part of the crate's
|
||||
/// public re-export surface (yet), so this asserts the shape observable from
|
||||
/// outside the crate: an `Err`, and an unmodified flat graph.
|
||||
#[test]
|
||||
fn bind_tap_refuses_an_undeclared_tap_name_without_mutating_the_graph() {
|
||||
let mut flat = producer(vec![Tap { name: "d".into(), from: TapWire { node: 0, field: 0 } }])
|
||||
.compile()
|
||||
.expect("a tap-bearing composite compiles");
|
||||
let nodes_before = flat.nodes.len();
|
||||
|
||||
let (tx, _rx) = mpsc::channel();
|
||||
let result = flat.bind_tap(
|
||||
"nope",
|
||||
Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)),
|
||||
recorder_sig(&[ScalarKind::F64]),
|
||||
);
|
||||
|
||||
assert!(result.is_err(), "binding an undeclared tap name must be refused");
|
||||
assert_eq!(
|
||||
flat.nodes.len(),
|
||||
nodes_before,
|
||||
"a refused bind must not append the sink to the flat node list"
|
||||
);
|
||||
}
|
||||
Reference in New Issue
Block a user