From af8e606cb63448ab3a02821abfeef4902a03e1d9 Mon Sep 17 00:00:00 2001 From: breis Date: Mon, 27 Jul 2026 20:58:13 -0400 Subject: [PATCH] Adopt core 0.5.0: scoped command registration Bump the edgecommons pin to the rust-lib/v0.5.0 tag (a14a328) and migrate the command surface to 0.5.0's breaking two-form registration (DESIGN-scoped-commands D-SC-1/D-SC-2): every verb declares a CommandScope and every handler receives the addressed instance. All four processor verbs declare CommandScope::Both (D-TP-9). A route is a component.instances[] entry, so a route-addressed delivery names exactly one route while a component-addressed one keeps its established "every route" meaning -- D-SC-3's dual-semantics use. Resolution is `addressed_instance ?? body.route`: the topic token is authoritative (D-SC-4) and the legacy `route` body selector is unchanged for component-addressed callers. get-stats and flush, which never had a body selector, now honor the topic token instead of ignoring it, so an addressed flush can no longer fan out to every route. Docs: the messaging-interface and how-to reference pages state each verb's scope and the per-route inbox topic; DESIGN.md carries the register entry and AGENTS.md the convention. The D-TP-8 panel rationale is rewritten -- both panels stay component-scoped because they render fleet-wide aggregates, not because routes are unaddressable. Validation: cargo test 101 (default features) / 114 (coverage feature set), clippy -D warnings clean, coverage 93.22% lines against the 90% gate. --- AGENTS.md | 6 ++ CLAUDE.md | 8 +- Cargo.lock | 6 +- Cargo.toml | 13 ++- DESIGN.md | 40 +++++--- docs/how-to-guides.md | 15 ++- docs/reference/messaging-interface.md | 40 +++++--- src/dispatch.rs | 135 ++++++++++++++++++++------ 8 files changed, 189 insertions(+), 74 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 1fe3d09..bf00fc9 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -94,6 +94,12 @@ reference. restamps `local` output with the processor's own identity and the subscribe fan-out drops any inbound message whose identity matches that — without both halves, a `local` route would loop. - A full route queue drops and counts; it never blocks the transport's dispatch task. +- Command verbs declare their addressing scope at registration (core 0.5.0, + `DESIGN-scoped-commands.md` D-SC-2). All four processor verbs — `get-stats`, `flush`, `pause`, + `resume` — declare `CommandScope::Both` (DESIGN.md D-TP-9): a route **is** a + `component.instances[]` entry, so a route-addressed delivery acts on that route alone while a + component-addressed one keeps its established "every route" meaning. The topic's instance token + wins over the legacy `route` body selector. - Four-way parity: if this repo's Java/Python/TypeScript siblings exist, observable behavior should match — same config shape, same metric names, same command verbs. - Builders/facades are the construction path (`messaging()`, `streams()`, `events()`, `commands()`, diff --git a/CLAUDE.md b/CLAUDE.md index 79097d6..ce3a9ea 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -8,10 +8,10 @@ inherits — lives in `AGENTS.md` and is shared with every agent tool. It is imp ## Local-dev notes -- **Default (committed pin):** `Cargo.toml`'s `edgecommons` dependency is a git `rev` pin, and - `Cargo.lock` is committed against that pin — a plain `cargo build`/`clone` needs only read access - to `edgecommons/edgecommons` (private; the fetch goes through the git CLI, and CI rewrites the URL - with the `EDGECOMMONS_READ_TOKEN` PAT). +- **Default (committed pin):** `Cargo.toml`'s `edgecommons` dependency is a git `rev` pin — currently + the `rust-lib/v0.5.0` release tag — and `Cargo.lock` is committed against that pin. + `edgecommons/edgecommons` is public, so a plain `cargo build`/`clone` fetches it anonymously, in CI + and locally. - **Building against an unpushed sibling change:** add a gitignored `.cargo/config.toml` next to this file with a `[patch]` block pointing the git URL at your local `../core/libs/rust` checkout (see the comment above the `edgecommons` dependency in `Cargo.toml` for the exact form). This is local-dev diff --git a/Cargo.lock b/Cargo.lock index 6f980ed..f48196d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1282,8 +1282,8 @@ checksum = "92773504d58c093f6de2459af4af33faa518c13451eb8f2b5698ed3d36e7c813" [[package]] name = "edgecommons" -version = "0.2.0" -source = "git+https://github.com/edgecommons/edgecommons.git?rev=36a70c48b65b35f77bfab70d3a73869debdfc407#36a70c48b65b35f77bfab70d3a73869debdfc407" +version = "0.5.0" +source = "git+https://github.com/edgecommons/edgecommons.git?rev=a14a3285573ef2bb6a531e1e1936c6dc40a85ef4#a14a3285573ef2bb6a531e1e1936c6dc40a85ef4" dependencies = [ "arc-swap", "async-trait", @@ -1316,7 +1316,7 @@ dependencies = [ [[package]] name = "edgestreamlog" version = "0.1.0" -source = "git+https://github.com/edgecommons/edgecommons.git?rev=36a70c48b65b35f77bfab70d3a73869debdfc407#36a70c48b65b35f77bfab70d3a73869debdfc407" +source = "git+https://github.com/edgecommons/edgecommons.git?rev=a14a3285573ef2bb6a531e1e1936c6dc40a85ef4#a14a3285573ef2bb6a531e1e1936c6dc40a85ef4" dependencies = [ "apache-avro", "arrow", diff --git a/Cargo.toml b/Cargo.toml index bfa097f..7c4f3d0 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -42,13 +42,12 @@ streaming-file-avro = ["streaming", "edgecommons/streaming-file-avro"] scripting-lua = ["dep:mlua"] [dependencies] -# The edgecommons Rust library, pinned to a commit on edgecommons/edgecommons (private). Building this -# repo standalone needs read access to that repo. The fetch goes through the git CLI (see -# .cargo/config.toml); locally it uses your gh/git credential helper, and in CI the reusable workflow -# rewrites this https URL with the EDGECOMMONS_READ_TOKEN PAT. For local development against an -# unpushed change, add a gitignored `.cargo/config.toml` `[patch]` override pointing at -# `../core/libs/rust` (the sibling checkout) instead of editing this pin — see CLAUDE.md. -edgecommons = { git = "https://github.com/edgecommons/edgecommons.git", rev = "36a70c48b65b35f77bfab70d3a73869debdfc407", default-features = false } +# The edgecommons Rust library, pinned to the `rust-lib/v0.5.0` release tag on +# edgecommons/edgecommons (a public repo — the fetch needs no credentials, in CI or locally). For +# local development against an unpushed change, add a gitignored `.cargo/config.toml` `[patch]` +# override pointing at `../core/libs/rust` (the sibling checkout) instead of editing this pin — see +# CLAUDE.md. +edgecommons = { git = "https://github.com/edgecommons/edgecommons.git", rev = "a14a3285573ef2bb6a531e1e1936c6dc40a85ef4", default-features = false } tokio = { version = "1", features = ["rt-multi-thread", "macros", "signal", "time", "sync"] } serde = { version = "1", features = ["derive"] } serde_json = "1" diff --git a/DESIGN.md b/DESIGN.md index cf1118e..998cd19 100644 --- a/DESIGN.md +++ b/DESIGN.md @@ -91,10 +91,24 @@ Numbered `D-TP-` so later sessions can cite them. panels (`overview`: fleet totals + flush/pause/resume; `routes`: per-route counters via `get-stats`) were straightforward to add via the library's existing `commands.register_panel` (used identically by the protocol-adapter templates) and ride the command surface that already shipped — - so they were implemented rather than deferred. `scope: "component"`, not `"instance"`: the - processor has no console-facing UNS instance dimension (a route is internal wiring, addressed by an - optional `body.route` field on the command verbs, not a topic segment), unlike a southbound - adapter's per-device instances. + so they were implemented rather than deferred. `scope: "component"`, not `"instance"`: both panels + render fleet-wide aggregates over every route (the overview totals, the routes table). Individual + routes stay reachable through the verbs' own `Both` scope (D-TP-9), which the console derives from + `describe` rather than from a panel descriptor. +- **D-TP-9 (core 0.5.0 adoption). Every command verb declares `CommandScope::Both`.** Core 0.5.0's + scoped registration (`register(verb, scope, handler)`, core `DESIGN-scoped-commands.md` D-SC-2) + requires each verb to declare its addressing. A route **is** a `component.instances[]` entry, so + the library subscribes a per-route command inbox and an instance-addressed delivery names exactly + one route — while the established meaning of a component-addressed delivery is "every route". That + is precisely D-SC-3's dual-semantics use of `Both`, so all four verbs (`get-stats`, `flush`, + `pause`, `resume`) declare it: `None` = every route, a token = that route. `Component` would refuse + a legitimate per-route request, and `Instance` would make the fleet-wide form an error. + Resolution is `addressed_instance ?? body.route` — the topic token is authoritative (D-SC-4) and + the legacy `route` body selector is retained, unchanged, for component-addressed callers. + `get-stats` and `flush`, which never had a body selector, now honor the topic token rather than + ignoring it: silently fanning a mutating `flush` out to every route when one was addressed is + exactly the mis-targeting class D-SC-1 closes. Unknown/unconfigured route ids stay non-fatal + (they simply select nothing), matching the pre-existing `pause`/`resume` behavior. ## Config @@ -109,15 +123,15 @@ field-by-field prose it was derived from; treat that page, not this one, as the Beyond the library's automatic `ping` / `reload-config` / `get-configuration`: -| Verb | Body | Result | -|------|------|--------| -| `get-stats` | — | Per-route counters (`in`/`out`/`dropped`/`streamAppends`/`publishFailures`/`queueDepth`/`paused`). | -| `flush` | — | Force-closes every route's open **time** windows now; `{flushed: n}`. Count windows are unaffected. | -| `pause` | `{route?}` | Stops enqueuing to a route (or all routes when omitted); `{paused: [ids]}`. | -| `resume` | `{route?}` | The inverse of `pause`; `{resumed: [ids]}`. | +| Verb | Scope | Body | Result | +|------|-------|------|--------| +| `get-stats` | `Both` | — | Counters for the addressed route, or every route (`in`/`out`/`dropped`/`streamAppends`/`publishFailures`/`queueDepth`/`paused`). | +| `flush` | `Both` | — | Force-closes the addressed route's (or every route's) open **time** windows now; `{flushed: n}`. Count windows are unaffected. | +| `pause` | `Both` | `{route?}` | Stops enqueuing to the addressed route (or all routes when neither the topic nor the body names one); `{paused: [ids]}`. | +| `resume` | `Both` | `{route?}` | The inverse of `pause`; `{resumed: [ids]}`. | -Two edge-console panels (`overview`, `routes`; see D-TP-8) bind to these verbs. Full wire contract in -`docs/reference/messaging-interface.md`. +Every verb declares `CommandScope::Both` (D-TP-9). Two edge-console panels (`overview`, `routes`; +see D-TP-8) bind to these verbs. Full wire contract in `docs/reference/messaging-interface.md`. ## Metrics @@ -127,7 +141,7 @@ summed across routes and emitted as interval deltas every 30s via `gg.metrics()` ## Validation -- `cargo test` (107 tests as of this remediation) — pipeline mechanics, route config parsing, +- `cargo test` (114 tests with the coverage job's feature set) — pipeline mechanics, route config parsing, route-build decisions (target/filter/publish/script-output-topic resolution, the restamp policy — `src/route_build.rs`), the fan-out handler + command/panel registration, the route dispatcher (local/northbound/stream targets, restamp, failure→evt), the metric/event surface. No broker diff --git a/docs/how-to-guides.md b/docs/how-to-guides.md index 1432c01..267b914 100644 --- a/docs/how-to-guides.md +++ b/docs/how-to-guides.md @@ -477,17 +477,22 @@ filter to specific adapters instead when you don't need the whole fleet, e.g. **Goal:** inspect and control a running processor from the console / any MQTT client. -The processor answers its command inbox at `ecv1/{device}/telemetry-processor/cmd/`. Send a -`cmd` envelope (`header.name` = the verb) with `header.reply_to` set to get a structured reply. +The processor answers its command inbox at `ecv1/{device}/telemetry-processor/cmd/` and, per +route, at `ecv1/{device}/telemetry-processor/{route}/cmd/`. Send a `cmd` envelope +(`header.name` = the verb) with `header.reply_to` set to get a structured reply. + +Every verb below declares the scope `both`: address the component to act on **every route**, or +address one route to act on that route alone. The topic's route token wins over a `route` field in +the body. | Verb | What it does | |------|--------------| | `ping` | liveness — `{status:"RUNNING", uptimeSecs}` (library built-in) | | `reload-config` | re-fetch + re-apply the config (library built-in) | | `get-configuration` | the redacted effective config (library built-in) | -| `get-stats` | per-route counters `{routes:[{id,in,out,dropped,streamAppends,publishFailures,queueDepth,paused}]}` | -| `flush` | force-close every route's open **time** windows now → `{flushed:n}` | -| `pause` / `resume` | stop / restart enqueuing to a route (`{route}`) or all routes (body omitted) | +| `get-stats` | counters for the addressed route, or every route: `{routes:[{id,in,out,dropped,streamAppends,publishFailures,queueDepth,paused}]}` | +| `flush` | force-close the addressed route's (or every route's) open **time** windows now → `{flushed:n}` | +| `pause` / `resume` | stop / restart enqueuing to the addressed route (or the body's `{route}`), or all routes when neither names one | The processor also publishes, without any request: its `state` keepalive (`ecv1/{device}/telemetry-processor/state`), a `metric/pipeline` throughput metric (when diff --git a/docs/reference/messaging-interface.md b/docs/reference/messaging-interface.md index 7d5e916..f513531 100644 --- a/docs/reference/messaging-interface.md +++ b/docs/reference/messaging-interface.md @@ -226,27 +226,43 @@ the triggering message, which continues on the source topic. ## Command verbs -The processor subscribes its own command inbox `ecv1/{device}/telemetry-processor/cmd/#` (wired +The processor subscribes its own command inbox `ecv1/{device}/telemetry-processor/cmd/#` and, for each +configured route, the per-route inbox `ecv1/{device}/telemetry-processor/{route}/cmd/#` (both wired automatically by the library). A `cmd` request whose `header.reply_to` is set gets a structured reply `{"ok": true, "result": …}` or `{"ok": false, "error": {"code", "message"}}`; a request without `reply_to` is fire-and-forget. **Built-in verbs** (library-provided, cannot be shadowed): -| Verb | Result | -|------|--------| -| `ping` | `{ "status": "RUNNING", "uptimeSecs": n }` — liveness/echo | -| `reload-config` | re-fetch + re-apply the config from the active source → `{ "reloaded": true }` | -| `get-configuration` | the current **redacted effective config** → `{ "config": … }` | +| Verb | Scope | Result | +|------|-------|--------| +| `ping` | both | `{ "status": "RUNNING", "uptimeSecs": n }` — liveness/echo | +| `reload-config` | both | re-fetch + re-apply the config from the active source → `{ "reloaded": true }` | +| `get-configuration` | both | the current **redacted effective config** → `{ "config": … }` | **Custom verbs** (registered by the processor): -| Verb | Body | Result | -|------|------|--------| -| `get-stats` | — | `{ "routes": [ { id, in, out, dropped, streamAppends, publishFailures, queueDepth, paused } ] }` — per-route counters | -| `flush` | — | force-close every route's open **time** windows now → `{ "flushed": n }` (messages emitted). Count windows keep their count semantics. | -| `pause` | `{ "route"? }` | stop enqueuing to a route (or all routes when omitted) → `{ "paused": [ids] }` | -| `resume` | `{ "route"? }` | the inverse of `pause` → `{ "resumed": [ids] }` | +| Verb | Scope | Body | Result | +|------|-------|------|--------| +| `get-stats` | both | — | `{ "routes": [ { id, in, out, dropped, streamAppends, publishFailures, queueDepth, paused } ] }` — counters for the addressed route, or every route | +| `flush` | both | — | force-close the addressed route's (or every route's) open **time** windows now → `{ "flushed": n }` (messages emitted). Count windows keep their count semantics. | +| `pause` | both | `{ "route"? }` | stop enqueuing to a route → `{ "paused": [ids] }` | +| `resume` | both | `{ "route"? }` | the inverse of `pause` → `{ "resumed": [ids] }` | + +### Verb scope and addressing + +A route is a `component.instances[]` entry, so every verb above is addressable two ways and each one +declares the scope `both`: + +- **Component-addressed** (`ecv1/{device}/telemetry-processor/cmd/{verb}`, no route token) means + **every route** — the fleet-wide form. +- **Route-addressed** (`ecv1/{device}/telemetry-processor/{route}/cmd/{verb}`) acts on that route + alone. + +The topic's route token is authoritative: when a request is route-addressed, `pause`/`resume` ignore +a `route` field in the body. The `route` body field remains the way to target a single route over the +component-addressed topic. A request that names one route in the topic and a different instance in +`body.instance` is rejected by the library with `BAD_ARGS` before the verb runs. > **Known limitation.** The built-in `reload-config` hot-swaps the config snapshot but the routes are > wired once at startup, so a route topology change needs a component restart; there is no dynamic diff --git a/src/dispatch.rs b/src/dispatch.rs index a019535..d605d5a 100644 --- a/src/dispatch.rs +++ b/src/dispatch.rs @@ -12,7 +12,10 @@ //! per-route delivery honoring `paused` and tallying `messages_in`/`messages_dropped`/`queue_depth`, //! with a rate-limited `evt/warning/queue-overflow` on backpressure. //! - [`register_commands`] wires the custom command verbs (`get-stats`/`flush`/`pause`/`resume`) and -//! the two edge-console panel descriptors onto the library's command inbox. +//! the two edge-console panel descriptors onto the library's command inbox. All four verbs are +//! registered at [`CommandScope::Both`] (D-SC-2/D-SC-3): a route **is** a +//! `component.instances[]` entry, so an instance-addressed delivery names exactly one route, +//! while a component-addressed one means "every route" — the dual-semantics use. //! - [`validate_script_output_topic`] rejects a script stage's `output.topic` that targets a reserved //! UNS class, feeds back into the route's own subscribe filters, or collides with a route-level //! `publish.topic`. @@ -110,24 +113,34 @@ pub(crate) async fn self_subscribe( /// Register the processor's custom command verbs on the library command inbox (a no-op when no /// messaging transport wired an inbox). The built-in `ping`/`reload-config`/`get-configuration` /// verbs are registered by the library and complement these. +/// +/// Every verb declares [`CommandScope::Both`]: a route is a `component.instances[]` entry, so an +/// instance-addressed delivery (`ecv1/{d}/{c}/{route}/cmd/{verb}`) names exactly one route, and a +/// component-addressed one keeps the established "every route" meaning (D-SC-3's dual-semantics +/// use). The topic's token is authoritative over the legacy `route` body field — see +/// [`selected_route`]. pub(crate) fn register_commands(cmds: &Arc, handles: Arc>) { - // get-stats — per-route counters snapshot. + // get-stats — counters for the addressed route, or every route. { let handles = handles.clone(); - try_register(cmds, "get-stats", command_handler(move |_req| { + try_register(cmds, "get-stats", CommandScope::Both, command_handler(move |req, addressed| { let handles = handles.clone(); - async move { Ok(Some(stats_json(&handles))) } + async move { + Ok(Some(stats_json(&handles, selected_route(&req, addressed.as_deref())))) + } })); } - // flush — force-close every route's open time windows now; report the total emitted. + // flush — force-close the addressed route's (or every route's) open time windows now; report + // the total emitted. { let handles = handles.clone(); - try_register(cmds, "flush", command_handler(move |_req| { + try_register(cmds, "flush", CommandScope::Both, command_handler(move |req, addressed| { let handles = handles.clone(); async move { + let selected = selected_route(&req, addressed.as_deref()); let mut flushed = 0u64; - for route in handles.iter() { + for route in handles.iter().filter(|r| is_selected(r, selected)) { let (reply_tx, reply_rx) = oneshot::channel(); if route.control.send(Control::Flush(reply_tx)).await.is_ok() { if let Ok(n) = reply_rx.await { @@ -140,28 +153,30 @@ pub(crate) fn register_commands(cmds: &Arc, handles: Arc Vec { ] } -/// Register a verb, logging (not failing) if the inbox rejects it. -fn try_register(cmds: &Arc, verb: &str, handler: Arc) { - if let Err(e) = cmds.register(verb, handler) { +/// Register a verb at its declared [`CommandScope`], logging (not failing) if the inbox rejects it. +fn try_register( + cmds: &Arc, + verb: &str, + scope: CommandScope, + handler: Arc, +) { + if let Err(e) = cmds.register(verb, scope, handler) { tracing::warn!(verb, error = %e, "failed to register command verb"); } } +/// The route one command delivery selects: the **topic-addressed instance wins** over the legacy +/// `route` body field (D-SC-4 — the topic is authoritative), and `None` means "every route". +/// +/// The library has already rejected a delivery whose topic token and `body.instance` disagree; this +/// component's own selector has always been named `route`, so the precedence between the two is +/// resolved here rather than by the library. +pub(crate) fn selected_route<'a>(request: &'a Message, addressed_instance: Option<&'a str>) -> Option<&'a str> { + addressed_instance.or_else(|| request.body.get("route").and_then(Value::as_str)) +} + +/// Whether `route` is selected by `selected` (`None` selects every route). +fn is_selected(route: &RouteHandle, selected: Option<&str>) -> bool { + selected.is_none() || selected == Some(route.id.as_str()) +} + /// The `get-stats` reply body: `{routes: [{id, in, out, dropped, streamAppends, publishFailures, -/// queueDepth, paused}]}`. -pub(crate) fn stats_json(handles: &[RouteHandle]) -> Value { +/// queueDepth, paused}]}` — the selected route only, or every route when `selected` is `None`. +pub(crate) fn stats_json(handles: &[RouteHandle], selected: Option<&str>) -> Value { let routes: Vec = handles .iter() + .filter(|r| is_selected(r, selected)) .map(|r| { json!({ "id": r.id, @@ -216,13 +252,18 @@ pub(crate) fn stats_json(handles: &[RouteHandle]) -> Value { json!({ "routes": routes }) } -/// Apply `paused` to the route named in `request.body.route` (or all routes when absent). Returns -/// `{paused|resumed: [ids...]}`. -pub(crate) fn set_paused(handles: &[RouteHandle], request: &Message, paused: bool) -> Value { - let route = request.body.get("route").and_then(Value::as_str); +/// Apply `paused` to the route this delivery selects ([`selected_route`]: the topic-addressed +/// instance, else `request.body.route`, else every route). Returns `{paused|resumed: [ids...]}`. +pub(crate) fn set_paused( + handles: &[RouteHandle], + request: &Message, + addressed_instance: Option<&str>, + paused: bool, +) -> Value { + let route = selected_route(request, addressed_instance); let mut affected = Vec::new(); for r in handles { - if route.is_none() || route == Some(r.id.as_str()) { + if is_selected(r, route) { r.stats.paused.store(paused, Ordering::Relaxed); affected.push(r.id.clone()); } @@ -280,7 +321,7 @@ mod tests { let orders: Vec = ps.iter().map(|p| p["order"].as_u64().unwrap()).collect(); assert_eq!(orders, vec![10, 20]); for p in &ps { - assert_eq!(p["scope"], json!("component"), "the processor has no console instance dimension"); + assert_eq!(p["scope"], json!("component"), "both panels render fleet-wide route aggregates"); assert!(!p["title"].as_str().unwrap().is_empty()); } assert_eq!(ps[0]["verbs"], json!(["get-stats", "flush", "pause", "resume"])); @@ -352,7 +393,7 @@ mod tests { h.stats.queue_depth.store(5, Ordering::Relaxed); h.stats.paused.store(true, Ordering::Relaxed); - let out = stats_json(&[h]); + let out = stats_json(&[h], None); let routes = out["routes"].as_array().unwrap(); assert_eq!(routes.len(), 1); assert_eq!(routes[0]["id"], json!("r1")); @@ -369,25 +410,59 @@ mod tests { MessageBuilder::new("cmd", "1.0").command(body).build() } + #[test] + fn stats_json_reports_only_the_addressed_route() { + let handles = vec![a_handle("r1"), a_handle("r2")]; + + let out = stats_json(&handles, Some("r2")); + let routes = out["routes"].as_array().unwrap(); + assert_eq!(routes.len(), 1, "an instance-addressed get-stats answers for that route only"); + assert_eq!(routes[0]["id"], json!("r2")); + + assert_eq!(stats_json(&handles, None)["routes"].as_array().unwrap().len(), 2); + assert!( + stats_json(&handles, Some("nope"))["routes"].as_array().unwrap().is_empty(), + "an unconfigured route selects nothing" + ); + } + #[test] fn set_paused_targets_one_route_when_named_and_all_when_omitted() { let handles = vec![a_handle("r1"), a_handle("r2")]; - let out = set_paused(&handles, &a_request(json!({ "route": "r1" })), true); + let out = set_paused(&handles, &a_request(json!({ "route": "r1" })), None, true); assert_eq!(out["paused"], json!(["r1"])); assert!(handles[0].stats.is_paused()); assert!(!handles[1].stats.is_paused()); - let out = set_paused(&handles, &a_request(json!({})), true); + let out = set_paused(&handles, &a_request(json!({})), None, true); assert_eq!(out["paused"].as_array().unwrap().len(), 2, "no route -> every route"); assert!(handles[1].stats.is_paused()); - let out = set_paused(&handles, &a_request(json!({})), false); + let out = set_paused(&handles, &a_request(json!({})), None, false); assert_eq!(out["resumed"].as_array().unwrap().len(), 2); assert!(!handles[0].stats.is_paused()); assert!(!handles[1].stats.is_paused()); } + #[test] + fn the_topic_addressed_route_wins_over_the_body_selector() { + // D-SC-4: the delivery topic is authoritative; `route` in the body is the legacy selector. + let handles = vec![a_handle("r1"), a_handle("r2")]; + + let out = set_paused(&handles, &a_request(json!({ "route": "r1" })), Some("r2"), true); + assert_eq!(out["paused"], json!(["r2"])); + assert!(!handles[0].stats.is_paused()); + assert!(handles[1].stats.is_paused()); + + // Component-addressed with no body selector = every route (D-SC-3's "the whole component"). + let req = a_request(json!({})); + assert_eq!(selected_route(&req, None), None); + assert_eq!(selected_route(&req, Some("r2")), Some("r2")); + let req = a_request(json!({ "route": "r1" })); + assert_eq!(selected_route(&req, None), Some("r1")); + } + // ---- self_subscribe (the fan-out handler) ----------------------------------------------------- fn southbound(signal_id: &str) -> Message {