diff --git a/AGENTS.md b/AGENTS.md index 18ffc7a..3a73d7d 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -17,7 +17,7 @@ the bus carries control and terminal metadata (`app/image/*` announcements, `evt It serves the canonical `southbound_health` metric plus operational families (`camera_captures`, `camera_queue`, `CameraCommand`), the standardized lifecycle verbs `sb/pause` / `sb/resume` / `sb/reconnect`, and 16 domain `sb/*` verbs (`sb/capture` and friends, `sb/ptz*`, queue verbs) on the -D-U28 component command inbox — SOUTHBOUND.md §2.2 sanctions `sb/capture`-style domain verbs. Runs on +D-U28 two-scope command inbox — SOUTHBOUND.md §2.2 sanctions `sb/capture`-style domain verbs. Runs on `GREENGRASS` / `HOST` / `KUBERNETES` via `edgecommons`, with no platform branching in this component. ## The seam @@ -56,7 +56,8 @@ the standard `edgecommons` envelope, owned by the canonical schema and not redec - Southbound routing/availability error codes are the standardized `BAD_ARGS` / `NO_SUCH_INSTANCE` / `DEVICE_UNAVAILABLE` (SOUTHBOUND.md §2.2); domain codes (`CAPTURE_*`, `PTZ_*`, …) are camera-specific. -- Instance routing is D-EIP-13/D-U28: body `instance`, optional iff exactly one camera is configured. +- Instance routing is D-EIP-13/D-U28: body `instance`, optional iff exactly one camera is configured; + an instance-addressed command topic routes by its token, which is authoritative (SOUTHBOUND.md §2.2). - Builders/facades are the construction path (`app()`, `events()`, `commands()`, `MetricBuilder`) — never hand-built topics or envelopes. - Runtime artifacts (durable state DBs, captured images, TLS fixtures, logs, build output) stay out of diff --git a/Cargo.lock b/Cargo.lock index dd52417..3bf44a9 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -790,8 +790,8 @@ checksum = "92773504d58c093f6de2459af4af33faa518c13451eb8f2b5698ed3d36e7c813" [[package]] name = "edgecommons" -version = "0.2.0" -source = "git+https://github.com/edgecommons/edgecommons.git?rev=36a70c48b65b35f77bfab70d3a73869debdfc407#36a70c48b65b35f77bfab70d3a73869debdfc407" +version = "0.4.0" +source = "git+https://github.com/edgecommons/edgecommons.git?rev=ef4c6248eec9cabc0acfc88eb0698528aa83a3ab#ef4c6248eec9cabc0acfc88eb0698528aa83a3ab" dependencies = [ "aes-gcm", "arc-swap", diff --git a/Cargo.toml b/Cargo.toml index e25b334..99bc38d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -48,10 +48,10 @@ native-all = ["genicam", "rtsp"] capacity-harness = [] [dependencies] -# Pinned to a published rev so CI (a standalone repo) can resolve it. Local dev builds against +# Pinned to a published rev (rust-lib/v0.4.0) so CI (a standalone repo) can resolve it. Local dev builds against # the sibling checkout via the gitignored `.cargo/config.toml` [patch] override, which replaces # this source outright. Same pattern as file-replicator. -edgecommons = { git = "https://github.com/edgecommons/edgecommons.git", rev = "36a70c48b65b35f77bfab70d3a73869debdfc407", default-features = false, features = ["credentials"] } +edgecommons = { git = "https://github.com/edgecommons/edgecommons.git", rev = "ef4c6248eec9cabc0acfc88eb0698528aa83a3ab", default-features = false, features = ["credentials"] } anyhow = "1" async-trait = "0.1" bytes = "1" @@ -101,7 +101,7 @@ rustls-pemfile = { version = "2", optional = true } tokio-rustls = { version = "0.26", optional = true } [dev-dependencies] -edgecommons = { git = "https://github.com/edgecommons/edgecommons.git", rev = "36a70c48b65b35f77bfab70d3a73869debdfc407", default-features = false, features = ["credentials", "standalone"] } +edgecommons = { git = "https://github.com/edgecommons/edgecommons.git", rev = "ef4c6248eec9cabc0acfc88eb0698528aa83a3ab", default-features = false, features = ["credentials", "standalone"] } proptest = "1" tokio = { version = "1", features = ["io-util", "macros", "net", "rt-multi-thread", "test-util"] } diff --git a/DESIGN.md b/DESIGN.md index 36babab..afe65fa 100644 --- a/DESIGN.md +++ b/DESIGN.md @@ -135,7 +135,7 @@ The words **MUST**, **MUST NOT**, **SHOULD**, **SHOULD NOT**, and **MAY** are no | D-CAM-15 | PTZ | Common normalized command contract mapped through backend capabilities | ONVIF provides the initial PTZ implementation; unsupported backends return a capability error. | | D-CAM-16 | Fleet safety | Layered bounded queues and byte-based admission | Camera count alone is not a safe memory or bandwidth bound. | | D-CAM-17 | Delivery | Integrate with `file-replicator` through disk and metadata, not code coupling | Keeps acquisition and delivery independently deployable. | -| D-CAM-18 | Command addressing | Use the shipped component `main` inbox and select camera `instance` in the body | Matches the shipped CommandInbox contract and both shipped adapters. **Resolved by core decision D-U28** (optional-instance UNS addressing: instance present ⇒ instance-scoped, absent ⇒ component/global-scoped, retiring the `main` sentinel), which supersedes the Phase 5 per-instance `cmd/sb/*` addressing in `core/docs/SOUTHBOUND.md` §2.2. The adapter currently ships the `main`-inbox + body-`instance` behavior; migration to optional-instance topic addressing (instance-scope `.../{instance}/cmd/sb/{verb}`, component/fleet `.../cmd/sb/{verb}`) is tracked under the D-U28 rollout, which also updates `core/docs/SOUTHBOUND.md`. | +| D-CAM-18 | Command addressing | Use the shipped component `main` inbox and select camera `instance` in the body | Matches the shipped CommandInbox contract and both shipped adapters. **Resolved by core decision D-U28** (optional-instance UNS addressing: instance present ⇒ instance-scoped, absent ⇒ component/global-scoped, retiring the `main` sentinel), which supersedes the Phase 5 per-instance `cmd/sb/*` addressing in `core/docs/SOUTHBOUND.md` §2.2. **Adopted with core 0.4.0** (rust-lib/v0.4.0, D-CAM-29): the inbox serves both scopes and the topic instance token is authoritative for every immediate-reply verb; `sb/capture`/`sb/capture-group` keep body-only routing pending a core scoped-outcome registration (the recorded gap in D-CAM-29). | | D-CAM-19 | Outbox acknowledgement | Withdrawn | There is no outbox and no acknowledgement to wait for. Terminal announcements publish once, best effort. Durable, acknowledged delivery is a generic messaging concern and belongs in the EdgeCommons messaging service as an opt-in augmentation across all four languages, available to any component, rather than being reimplemented inside one. | | D-CAM-20 | Group capture | `sb/capture-group` fans one request out as independent per-camera capture jobs sharing an adapter-generated `captureGroupId`; the single deferred reply aggregates every member's terminal result | One operator action often needs an evidence set from several cameras. Aggregation fits the shipped single-reply command model plus the D-CAM-10 deferred reply; a core scatter-gather exchange (one request, multiple replies) is not required for v1 and is raised as a core question (§27). | | D-CAM-21 | Capture thumbnail | Opt-in per capture profile (`thumbnail.size` = `small` 160px / `medium` 320px / `large` 640px, longest edge, aspect preserved, never upscaled); JPEG; carried in the ANNOUNCEMENT only, as native protobuf bytes; never in the durable record | A consumer on the bus can see the picture without fetching the file. It is bounded by the longest edge because cameras are 4:3 and 16:9 and a fixed W×H would distort or letterbox. It is announcement-only because the terminal body IS the committed document — the catalog's `terminal_result`, the metadata sidecar verbatim, and the body group replies embed — and a lossy, derived, disposable preview must not be durably stored per capture (D-CAM-13). It carries NO digest: a thumbnail is a lossy re-encode and a `sha256` beside the artifact's would invite a consumer to believe it is verifiable against it. A thumbnail that cannot be rendered, or will not fit the byte ceiling, is dropped and counted — it never fails a capture. The ceiling is the TRANSPORT's, resolved at startup: the Greengrass IPC client encodes a whole message into a static 10,000-byte buffer, so IPC carries `small` only (6 KiB budget) and a larger configured size is clamped down rather than rejected — the same config ships to Greengrass and to Kubernetes; MQTT carries all three (60 KiB budget, bounded by the library's 64 KiB binary-value cap, not by the broker). And a preview NEVER outranks the result: if an announcement carrying one cannot be published, the result is announced again without it. | @@ -145,6 +145,7 @@ The words **MUST**, **MUST NOT**, **SHOULD**, **SHOULD NOT**, and **MAY** are no | D-CAM-26 | Edge-console panel trio | Register `overview` / `signals` / `diagnostics` panels via `register_panel` (order 10/20/30, `scope: "instance"`), bound to the verbs the adapter serves: overview → `sb/status`/`sb/reconnect`/`sb/pause`/`sb/resume`; signals → `sb/list`/`sb/status`/`sb/capture`/`sb/capture-status`; diagnostics → `sb/discover`/`sb/queue-status`. Every panel meets the renderable-descriptor floor: `summary`/`keyValueList` widgets carry `rows`, `commandSummary` widgets carry `verbs`, widgets carry `id`/`title`, and no widget names a `writeVerb`. The signal-adapter widget kinds are absent by design — no `signalGrid` (no `sb/signals`/`sb/read` signal inventory; `cameraRoster`/`captureSurface` are the camera-domain equivalents) and no `treeBrowser` (no hierarchical `sb/browse`; `sb/discover` is an active scan, not a browse). | The baseline advertises a descriptor-driven panel surface to edge-console; registering the trio on the same inbox as the verbs, before the acknowledged subscription begins, advertises the panels atomically with the command surface they drive. | | D-CAM-27 | Committed `Cargo.lock`; `data-types.md` N/A | Commit `Cargo.lock` (drop the `/Cargo.lock` gitignore + rationale), regenerated with the local `.cargo` `[patch]` override inactive so it records the pinned git source and is valid on a fresh clone / in CI. The RTSP native-coverage harness (`simulators/run-rtsp-native-coverage.ps1`), which mounts the workspace read-only, keeps its **writable-overlay** lockfile — a single-file bind mount from outside the read-only source tree that masks the committed lock and is (re)generated in-container by a prep run — so cargo never has to rewrite the committed git-sourced lock on the read-only mount. The adapter's signal `data-types.md` page is **N/A**: this component publishes capture announcements (`app/image/*`), not `SouthboundSignalUpdate` envelopes, so the signal value-mapping page does not apply; the published shapes are documented in `messaging-interface.md`. | The three-way constraint (a committed lock must record the git source ⊕ a `[patch]`-ed local build rewrites it to a path source ⊕ the read-only validation mount cannot rewrite it) is resolved deliberately rather than by suppressing the `edgecommons component validate` warning: the committed lock is git-sourced and reproducible on a fresh clone/CI, and the two local frictions (a `git status` that shows lock churn while the patch is active, and the read-only harness) are handled by simply not committing that churn and by running the harness with the patch inactive. The manifest `license` field is reconciled to `BUSL-1.1` to match the `LICENSE` file (`Fixes #7`). | | D-CAM-28 | `config.schema.json` | Author a Draft 2020-12 `config.schema.json` whose **root models `component.global`** (with `$defs/camera` for one `component.instances[]` entry and `$defs` for every backend/profile/schedule/enum), derived field-for-field from `src/config.rs`. Strict (`additionalProperties:false`) everywhere the parser is (`deny_unknown_fields`); permissive only where the parser is (`featureOverrides`, `resourceGroups`). | `edgecommons component validate` could not check camera configs and warned on the missing schema. The CLI validates `component.global` against the schema root (`ec-validate/schema.rs` checks `/component/global` only), so the root is `component.global`; `$defs/camera` documents and lets external Draft 2020-12 validators check `instances[]`, matching the scaffold's `$defs/device`. Validated: the shipped `deploy/docker/simulator-config.json` and the §10.1 full example ACCEPT; a malformed global key, an ONVIF backend missing `mediaProfile`, a bad enum, an unknown backend field, and a camera missing `backend` all REJECT. `edgecommons component validate` reports **no findings** (schema valid, config clean, lockfile warning resolved). | +| D-CAM-29 | Core 0.4.0 adoption: scoped instance routing + conditional availability | Pin `edgecommons` at rust-lib/v0.4.0 (ef4c6248). Every immediate-reply verb is registered via `register_scoped` (SOUTHBOUND §2.2 / D-U28): the delivery topic’s `{instance}` token is authoritative — `scoped_request` refuses a conflicting `body.instance` with `BAD_ARGS`, injects a topic-only token as the routing selector (so the registry resolves it, `NO_SUCH_INSTANCE` for an unknown token), and passes component-scoped deliveries through to the existing body routing; verbs whose closed schemas carry no camera selector (`sb/list`, `sb/discover`, `sb/capture-group-submit`, `sb/capture-cancel`) refuse an instance-addressed delivery with `BAD_ARGS` rather than silently ignoring the token. `sb/discover`’s configuration-conditional availability is published into `describe` via `set_command_availability` (`disabled` with a reason while `global.discovery.enabled` is false; reapplied on committed reloads through the configuration listener). `receivedTs` is **N/A**: the adapter is a direct camera client with no upstream broker hop to stamp a receive time. | **Recorded gap, surfaced up front:** core 0.4.0 exposes the addressed-instance token only to the immediate-reply scoped registration; the two deferred verbs (`sb/capture`, `sb/capture-group`) must stay on `register_outcome` — trading deferred settlement (the dispatch permit is released for the capture’s duration and `sb/capture-cancel` settles the held reply) for token visibility would regress the command plane — so they route by the body on either topic and the topic token does not route them. Closing it needs a core scoped-outcome registration (a core follow-up, not an adapter workaround). PTZ capability is per-camera and runtime-discovered, so it is deliberately NOT mirrored into component-scope availability. | | D-CAM-22 | Bare-RTSP backend | A distinct `rtsp` backend addresses a camera by a raw `rtsp://`/`rtsps://` URL, with no ONVIF. It is still-image only, reuses the shared RTSP engine (`RtspCaptureController`) and the network/credential/TLS primitives, and advertises `capture_modes=[rtsp-frame]` with all PTZ/snapshot/discovery capabilities off. To make it buildable without ONVIF, the protocol-neutral net/auth primitives and the credential-resolution seam are lifted from the `onvif` module into a shared `backend::net` module, and the `rtsp` cargo feature is decoupled from `onvif`. | ONVIF gives identity, capability discovery, media profiles, snapshot, PTZ, and the governed stream URI; a raw RTSP URL gives none of these, so it is a genuinely different camera kind rather than a mode of `onvif-rtsp` — a separate backend keeps the ONVIF backend's required-field invariants (`deviceServiceUrl`/`mediaProfile`) intact. `connect()` performs the RTSP `DESCRIBE`/`SETUP` + auth + SDP/codec validation so a dead URL, bad auth, or unsupported codec fails at connect (the supervisor keeps such a camera OFFLINE rather than falsely ONLINE, since reachability is inferred from a successful connect). The URL carries no credentials (userinfo is rejected); credentials are `$secret` references resolved through the same bounded EdgeCommons path as ONVIF, and the same host-allowlist / DNS-pin / RTSPS-SNI / forbidden-address policy applies to the user-supplied URL. Decoupling the feature lets an operator ship an RTSP-only binary without the ONVIF surface. | ## 5. System context @@ -978,12 +979,14 @@ ecv1/{device}/{component}/{instance}/{class}[/{channel...}] Commands use at most two channel tokens (`sb/{verb}`), so they remain valid when `topic.includeRoot` reduces the channel budget. -The component command inbox plus body `instance` rule is the camera adapter's messaging contract. Org-level -core decision **D-U28** defines an optional-instance UNS grammar: the instance token is present for -instance-scoped traffic and absent for component/global-scoped traffic, and there is no `main` sentinel. The -camera adapter is single-instance at the component level, so its commands are component-scope -`ecv1/{device}/camera-adapter/cmd/sb/{verb}`; a specific camera is selected by the body `instance` field, -not by a per-instance command topic. +The two-scope command inbox is the camera adapter's messaging contract. Org-level core decision +**D-U28** defines an optional-instance UNS grammar: the instance token is present for instance-scoped +traffic and absent for component/global-scoped traffic, and there is no `main` sentinel. The adapter +serves both command scopes (core 0.4.0, D-CAM-29): component-scope +`ecv1/{device}/camera-adapter/cmd/sb/{verb}` selects a camera by the body `instance` field, and +instance-addressed `ecv1/{device}/camera-adapter/{instance}/cmd/sb/{verb}` routes by the topic token, +which is authoritative (a conflicting body `instance` is `BAD_ARGS`). The deferred verbs `sb/capture` / +`sb/capture-group` route by the body only (the D-CAM-29 recorded gap). ### 12.2 Core prerequisites @@ -2504,12 +2507,12 @@ Reviewers should explicitly decide: 6. Should PTZ preset mutation stay disabled by default as proposed? 7. Physical-camera validation is waived for this project because no hardware is available; a future hardware-certified release must select models for the compatibility matrix. -8. Command addressing — **resolved.** Core decision D-U28 adopts an optional-instance UNS grammar - (instance present ⇒ instance-scoped, absent ⇒ component/global-scoped, retiring the `main` sentinel), - superseding both the Phase 5 per-instance `cmd/sb/*` target in `core/docs/SOUTHBOUND.md` §2.2 and this - adapter's body-`instance` selection. The adapter ships the `main`-inbox + body-`instance` behavior - today; it migrates to optional-instance topic addressing under the D-U28 rollout, which updates - `core/docs/SOUTHBOUND.md`. +8. Command addressing — **resolved and adopted.** Core decision D-U28 adopts an optional-instance UNS + grammar (instance present ⇒ instance-scoped, absent ⇒ component/global-scoped, retiring the `main` + sentinel), and `core/docs/SOUTHBOUND.md` §2.2 now specifies addressed-instance routing. The adapter + serves both scopes via the core 0.4.0 scoped registration (D-CAM-29): the topic instance token is + authoritative for every immediate-reply verb; the deferred capture verbs keep body-only routing + pending a core scoped-outcome registration (the D-CAM-29 recorded gap). 9. Should the core add a scatter-gather message exchange pattern — one request producing multiple correlated replies or a streamed reply set? v1 group capture deliberately aggregates member results into one reply within the existing single-reply command model. A core pattern would also serve other diff --git a/docs/reference/messaging-interface.md b/docs/reference/messaging-interface.md index e215ea3..1764a14 100644 --- a/docs/reference/messaging-interface.md +++ b/docs/reference/messaging-interface.md @@ -1,14 +1,17 @@ # Messaging interface -Every command is a request/reply exchange on the component command inbox: +Every command is a request/reply exchange on the command inbox, which serves both command scopes: ```text -ecv1/{device}/camera-adapter/cmd/sb/{verb} +ecv1/{device}/camera-adapter/cmd/sb/{verb} (component scope) +ecv1/{device}/camera-adapter/{instance}/cmd/sb/{verb} (instance-addressed) ``` -Select a camera with the JSON body field `instance`; do not construct a per-instance command topic. The -reply is correlated with the incoming envelope. Normal capture *completion* is a separate terminal -application message, not the command reply — see [Terminal application messages](#terminal-application-messages). +On the component-scope topic, select a camera with the JSON body field `instance`. On an +instance-addressed topic the topic's `{instance}` token is authoritative: it routes the command, and a +body `instance` that disagrees with it is refused with `BAD_ARGS`. The reply is correlated with the +incoming envelope. Normal capture *completion* is a separate terminal application message, not the +command reply — see [Terminal application messages](#terminal-application-messages). ## Conventions @@ -16,10 +19,17 @@ These rules apply to every verb below. - **Closed request schema.** Bodies are parsed with `deny_unknown_fields`: any field not listed for a verb is rejected with `BAD_ARGS`. All field names are **camelCase** on the wire. -- **Selecting a camera.** Actuation verbs take an optional `instance`. Omit it only when exactly one - camera is configured — the sole camera is then used. With more than one camera, omission is - `BAD_ARGS`; an unknown name is `NO_SUCH_INSTANCE`; a disabled camera is `CAMERA_DISABLED`. An - `instance` token is non-empty, ≤128 bytes, ASCII letters/digits/`.`/`_`/`-`. +- **Selecting a camera.** Actuation verbs take an optional `instance`. On the component-scope topic, + omit it only when exactly one camera is configured — the sole camera is then used; with more than + one camera, omission is `BAD_ARGS`. An unknown name is `NO_SUCH_INSTANCE`; a disabled camera is + `CAMERA_DISABLED`. An `instance` token is non-empty, ≤128 bytes, ASCII + letters/digits/`.`/`_`/`-`. +- **Instance-addressed topics.** A verb that takes an `instance` selector also accepts the + instance-addressed topic form: the topic token routes it (no body `instance` needed), and a + conflicting body `instance` is `BAD_ARGS`. The component-scoped verbs `sb/list`, `sb/discover`, + `sb/capture-group-submit`, and `sb/capture-cancel` refuse an instance-addressed delivery with + `BAD_ARGS`. The two deferred verbs, `sb/capture` and `sb/capture-group`, select their target(s) + from the body only — the topic token does not route them; send them to the component-scope topic. - **Idempotency.** Every *mutating* verb requires a caller-owned `requestId` (1–256 bytes, no control characters). A retry with the same `requestId` and the same arguments returns the original outcome; a reused `requestId` with **different** arguments is `IDEMPOTENCY_CONFLICT`; an operation whose outcome @@ -322,7 +332,8 @@ snapshot. **What it does.** Runs one bounded, credential-free discovery pass across the compiled backends (or serves a continuation of a retained pass). Discovery must be enabled in config, or the verb replies -`UNSUPPORTED_CAPABILITY`. +`UNSUPPORTED_CAPABILITY` — and the built-in `describe` verb lists `sb/discover` as `disabled` with the +reason, so a console can grey it out instead of discovering the refusal. **Input payload** diff --git a/src/main.rs b/src/main.rs index 00786fd..b33550c 100644 --- a/src/main.rs +++ b/src/main.rs @@ -66,6 +66,14 @@ async fn main() -> anyhow::Result<()> { cameras = loaded.config.instances.len(), "camera-adapter configuration accepted" ); + // Publish `sb/discover`'s configuration-conditional availability into `describe` (a reload + // republishes it through the configuration listener below). + if let Some(inbox) = gg.commands() { + camera_adapter::runtime::apply_discover_availability( + &inbox, + loaded.config.global.discovery.enabled, + ); + } let credential_service = gg.credentials(); credential_service_available.store(credential_service.is_some(), Ordering::Release); let backend_context = @@ -149,6 +157,7 @@ async fn main() -> anyhow::Result<()> { Arc::downgrade(&runtime), app_factory, events_factory, + gg.commands(), )); gg.add_config_apply_listener(runtime_config_listener)?; readiness.complete_startup(); diff --git a/src/runtime.rs b/src/runtime.rs index 9a49d87..a4f421e 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -44,8 +44,8 @@ use std::time::Duration; use async_trait::async_trait; use edgecommons::commands::{ - CommandError, CommandInbox, CommandOutcome, DeferredReplyRegistry, DeferredReplyToken, - outcome_handler, + AVAILABILITY_AVAILABLE, AVAILABILITY_DISABLED, CommandError, CommandInbox, CommandOutcome, + DeferredReplyRegistry, DeferredReplyToken, outcome_handler, scoped_command_handler, }; use edgecommons::config::{ Config, ConfigurationApplicationError, ConfigurationApplicationResult, @@ -2247,6 +2247,37 @@ impl CommandVerb { pub fn parse(verb: &str) -> Option { Self::ALL.into_iter().find(|known| known.as_str() == verb) } + + /// True for the verbs whose reply may be deferred (`sb/capture`, `sb/capture-group`). + /// + /// These stay on the outcome registration: the core 0.4.0 scoped registration is + /// immediate-reply only, and giving up the deferred settlement (the dispatch permit is + /// released while a capture runs; `sb/capture-cancel` settles the pending reply) is not an + /// acceptable trade for topic-token visibility. See the D-CAM-18 register entry for the + /// recorded consequence. + #[must_use] + pub const fn defers(self) -> bool { + matches!(self, Self::Capture | Self::CaptureGroup) + } + + /// True when the verb's closed request schema carries a camera `instance` selector, so an + /// instance-addressed delivery topic (D-U28) can route it. + /// + /// The rest are component-scoped by design — `sb/list` and `sb/discover` answer for the + /// fleet, the group verbs target `instances[]`, and `sb/capture-cancel` targets a durable + /// component-scoped capture/group id — so an instance-addressed delivery of one of them is + /// refused rather than silently accepted with the token ignored. + #[must_use] + pub const fn instance_routable(self) -> bool { + !matches!( + self, + Self::List + | Self::Discover + | Self::CaptureGroup + | Self::CaptureGroupSubmit + | Self::CaptureCancel + ) + } } /// The verbs registered with the command inbox, derived from [`CommandVerb::ALL`]. @@ -2255,6 +2286,91 @@ pub fn camera_command_verbs() -> Vec<&'static str> { CommandVerb::ALL.iter().map(|verb| verb.as_str()).collect() } +/// SOUTHBOUND §2.2 addressed-instance routing (D-U28): reconcile the delivery topic's +/// `{instance}` token with the request body before dispatch. +/// +/// - **Topic instance is authoritative**: a `body.instance` that disagrees with the topic token +/// is refused with `BAD_ARGS`. +/// - **Topic-only**: the token is injected as the body's `instance` selector, so the existing +/// registry routing resolves it (an unknown token then answers `NO_SUCH_INSTANCE`). +/// - **Component scope** (`addressed` = `None`): the request passes through unchanged — the +/// existing body routing applies, including the single-camera default. +/// - A verb that is not [`CommandVerb::instance_routable`] refuses an instance-addressed +/// delivery with `BAD_ARGS` instead of silently ignoring the token. +fn scoped_request( + verb: CommandVerb, + mut request: Message, + addressed: Option<&str>, +) -> std::result::Result { + let Some(topic) = addressed else { + return Ok(request); + }; + if !verb.instance_routable() { + return Err(CommandError::new( + crate::ErrorCode::BadArgs.as_str(), + format!( + "{} is component-scoped; publish it to the component command topic, not a camera instance", + verb.as_str() + ), + )); + } + match request.body.get("instance").and_then(serde_json::Value::as_str) { + Some(in_body) if in_body != topic => Err(CommandError::new( + crate::ErrorCode::BadArgs.as_str(), + format!( + "body `instance` (`{in_body}`) conflicts with the topic-addressed instance (`{topic}`)" + ), + )), + Some(_) => Ok(request), + None => { + match &mut request.body { + serde_json::Value::Object(map) => { + map.insert( + "instance".to_string(), + serde_json::Value::String(topic.to_string()), + ); + } + body @ serde_json::Value::Null => { + *body = serde_json::json!({ "instance": topic }); + } + _ => { + return Err(CommandError::new( + crate::ErrorCode::BadArgs.as_str(), + "an instance-addressed command body must be a JSON object", + )); + } + } + Ok(request) + } + } +} + +/// Publishes `sb/discover`'s configuration-conditional availability into `describe`. +/// +/// `sb/discover` is the one verb a configuration can switch off outright +/// (`component.global.discovery.enabled: false` answers `UNSUPPORTED_CAPABILITY`), so its +/// describe entry is marked `disabled` with that reason — and reverts to `available` when a +/// configuration (re)load enables discovery again. PTZ capability is deliberately NOT mirrored +/// here: it is per-camera and discovered at runtime, and availability is component-scope. +/// +/// Best-effort: a failure is logged, never fatal — availability is advisory metadata on +/// `describe`, not a gate on the verb itself. +pub fn apply_discover_availability(inbox: &CommandInbox, discovery_enabled: bool) { + let (state, reason) = if discovery_enabled { + (AVAILABILITY_AVAILABLE, None) + } else { + ( + AVAILABILITY_DISABLED, + Some("camera discovery is disabled by configuration"), + ) + }; + if let Err(error) = + inbox.set_command_availability(CommandVerb::Discover.as_str(), state, reason) + { + tracing::warn!(error = %error, "sb/discover availability could not be published to describe"); + } +} + /// The three edge-console panel descriptors for the camera adapter. /// /// Core validates `id`/`title`/uniqueness; the widget kinds and bound verbs are console-interpreted, @@ -2733,6 +2849,9 @@ struct RuntimeReloadTransaction { apps: BTreeMap>, events: BTreeMap, checkpoint: Option, + /// The command inbox, when available, so a committed reload republishes the + /// configuration-conditional `sb/discover` availability into `describe`. + commands: Option>, } impl RuntimeReloadTransaction { @@ -2767,6 +2886,12 @@ impl PreparedConfigurationApply for RuntimeReloadTransaction { // A successful adapter transition is now waiting only for Core's infallible // ArcSwap store. Rollback is no longer permitted after this point. self.checkpoint = None; + if let Some(inbox) = &self.commands { + apply_discover_availability( + inbox, + self.replacement.global.discovery.enabled, + ); + } Ok(()) } Err(error) => { @@ -2800,6 +2925,9 @@ pub struct RuntimeConfigListener { runtime: Weak, app_factory: Arc, events_factory: Arc, + /// The command inbox, when available, so a committed reload republishes the + /// configuration-conditional `sb/discover` availability ([`apply_discover_availability`]). + commands: Option>, } impl RuntimeConfigListener { @@ -2810,11 +2938,13 @@ impl RuntimeConfigListener { runtime: Weak, app_factory: Arc, events_factory: Arc, + commands: Option>, ) -> Self { Self { runtime, app_factory, events_factory, + commands, } } } @@ -2885,6 +3015,7 @@ impl ConfigurationApplyListener for RuntimeConfigListener { apps, events, checkpoint: Some(checkpoint), + commands: self.commands.clone(), })) } } @@ -2909,20 +3040,53 @@ impl RuntimeCommandRouter { }) } - /// Registers every required adapter verb before the core subscribes to the command filter. + /// Registers every required adapter verb before the core subscribes to the command filters. + /// + /// Every verb whose reply is always immediate is registered through the **scoped** + /// registration (SOUTHBOUND §2.2 / D-U28): the handler receives the delivery topic's + /// `{instance}` token and [`scoped_request`] reconciles it with the request body before + /// dispatch — the topic token is authoritative. The two deferred-capable verbs + /// ([`CommandVerb::defers`]) must stay on the outcome registration, which core 0.4.0 does + /// not expose the topic token to; they keep body-only routing (recorded under D-CAM-18). /// /// A registration failure is fatal to component construction; a partial command surface is /// never exposed as active. pub fn register(self: &Arc, inbox: &CommandInbox) -> edgecommons::Result<()> { - for verb in camera_command_verbs() { + let deferred_registry = inbox.deferred_replies(); + for verb in CommandVerb::ALL { let router = Arc::clone(self); - inbox.register_outcome( - verb, - outcome_handler(move |request, deferred| { - let router = Arc::clone(&router); - async move { router.dispatch(verb, request, deferred).await } - }), - )?; + if verb.defers() { + inbox.register_outcome( + verb.as_str(), + outcome_handler(move |request, deferred| { + let router = Arc::clone(&router); + async move { router.dispatch(verb.as_str(), request, deferred).await } + }), + )?; + } else { + let deferred_registry = deferred_registry.clone(); + inbox.register_scoped( + verb.as_str(), + scoped_command_handler(move |request, addressed| { + let router = Arc::clone(&router); + let deferred_registry = deferred_registry.clone(); + async move { + let request = scoped_request(verb, request, addressed.as_deref())?; + match router + .dispatch(verb.as_str(), request, deferred_registry) + .await + { + CommandOutcome::ImmediateSuccess(value) => Ok(value), + CommandOutcome::ImmediateError(error) => Err(error), + _ => Err(CommandError::new( + crate::ErrorCode::BackendError.as_str(), + "verb settled through a deferred path it does not declare", + )), + } + } + }), + )?; + } } // The edge-console panel trio (overview / signals / diagnostics). Registered on the same inbox // as the verbs, before the acknowledged subscription begins, so the descriptor surface is diff --git a/src/runtime/tests/mod.rs b/src/runtime/tests/mod.rs index e80217f..04dc0c3 100644 --- a/src/runtime/tests/mod.rs +++ b/src/runtime/tests/mod.rs @@ -596,6 +596,7 @@ async fn runtime_config_listener_rejects_when_its_runtime_is_gone_before_factory Arc::new(|_instance, _config| -> edgecommons::Result { unreachable!("unavailable runtimes must not construct event facades") }), + None, ); let candidate = Arc::new( Config::from_value(COMPONENT_NAME, "gw-01", reload_config("C:/captures-a")) @@ -608,6 +609,104 @@ async fn runtime_config_listener_rejects_when_its_runtime_is_gone_before_factory assert_eq!(error.code, "CONFIG_APPLICATION_UNAVAILABLE"); } +// --- SOUTHBOUND §2.2 addressed-instance routing (D-U28) -------------------------------------- + +fn scoped_fixture_message(body: serde_json::Value) -> Message { + edgecommons::messaging::MessageBuilder::new("sb/status", "1.0") + .structured_payload(body) + .build() +} + +/// Topic-only (§2.2): the delivery topic's instance token becomes the routing selector the +/// registry resolves — no `body.instance` needed. +#[test] +fn topic_addressed_instance_is_injected_as_the_routing_selector() { + let request = scoped_request( + CommandVerb::Status, + scoped_fixture_message(json!({})), + Some("camera-b"), + ) + .expect("a topic-only instance routes the command"); + assert_eq!(request.body, json!({ "instance": "camera-b" })); + + // An agreeing `body.instance` passes through untouched. + let request = scoped_request( + CommandVerb::Status, + scoped_fixture_message(json!({ "instance": "camera-b" })), + Some("camera-b"), + ) + .expect("an agreeing body instance is accepted"); + assert_eq!(request.body, json!({ "instance": "camera-b" })); + + // A non-object body cannot carry the routing selector. + let error = scoped_request( + CommandVerb::Status, + scoped_fixture_message(json!("junk")), + Some("camera-b"), + ) + .expect_err("a non-object body cannot be instance-addressed"); + assert_eq!(error.code, crate::ErrorCode::BadArgs.as_str()); +} + +/// The topic token is authoritative (§2.2): a disagreeing `body.instance` is refused. +#[test] +fn conflicting_body_and_topic_instance_is_refused_bad_args() { + let error = scoped_request( + CommandVerb::Status, + scoped_fixture_message(json!({ "instance": "camera-a" })), + Some("camera-b"), + ) + .expect_err("a conflicting body instance must be refused"); + assert_eq!(error.code, crate::ErrorCode::BadArgs.as_str()); + assert!(error.message.contains("camera-a") && error.message.contains("camera-b")); +} + +/// Component scope (§2.2): no topic token means the body passes through unchanged and the +/// existing `body.instance` routing (including the single-camera default) applies. +#[test] +fn component_scope_keeps_body_instance_routing() { + let request = scoped_request( + CommandVerb::Status, + scoped_fixture_message(json!({ "instance": "camera-a" })), + None, + ) + .expect("component-scoped requests pass through"); + assert_eq!(request.body, json!({ "instance": "camera-a" })); + + let request = scoped_request(CommandVerb::Status, scoped_fixture_message(json!({})), None) + .expect("an empty component-scoped body passes through"); + assert_eq!(request.body, json!({})); +} + +/// The verbs whose schemas cannot name a camera refuse instance addressing instead of silently +/// ignoring the token; the deferred set is exactly the two capture verbs. +#[test] +fn verb_scoping_classification_matches_the_request_schemas() { + for verb in CommandVerb::ALL { + assert_eq!( + verb.defers(), + matches!(verb, CommandVerb::Capture | CommandVerb::CaptureGroup), + "{} deferral classification", + verb.as_str() + ); + } + for verb in [ + CommandVerb::List, + CommandVerb::Discover, + CommandVerb::CaptureGroupSubmit, + CommandVerb::CaptureCancel, + ] { + assert!(!verb.instance_routable()); + let error = scoped_request(verb, scoped_fixture_message(json!({})), Some("camera-a")) + .expect_err("a component-scoped verb refuses instance addressing"); + assert_eq!(error.code, crate::ErrorCode::BadArgs.as_str()); + } + // Component-scoped deliveries of those verbs are untouched. + let request = scoped_request(CommandVerb::List, scoped_fixture_message(json!({})), None) + .expect("component-scoped fleet verbs pass through"); + assert_eq!(request.body, json!({})); +} + #[cfg(test)] mod simulator_runtime; diff --git a/src/runtime/tests/simulator_runtime.rs b/src/runtime/tests/simulator_runtime.rs index fd633d9..85f6799 100644 --- a/src/runtime/tests/simulator_runtime.rs +++ b/src/runtime/tests/simulator_runtime.rs @@ -794,6 +794,61 @@ fn immediate_success(outcome: CommandOutcome) -> serde_json::Value { } } +/// SOUTHBOUND §2.2 through the production dispatch stack: an instance-ADDRESSED command (the +/// delivery topic's `{instance}` token, D-U28) routes to the addressed camera with no +/// `body.instance`, and an unknown token answers the standardized `NO_SUCH_INSTANCE` — proving +/// the scoped reconciliation composes with the registry routing the runtime already serves. +#[tokio::test] +async fn an_instance_addressed_command_routes_by_the_topic_token() { + let (port, _broker) = spawn_recording_mqtt_broker().await; + let directory = TempDir::new().unwrap(); + let configuration = config(directory.path(), &["camera-a", "camera-b"], false); + let runtime = runtime(configuration, &directory).await; + for instance in ["camera-a", "camera-b"] { + runtime + .start_supervisor(instance.to_string(), runtime.engine(instance).unwrap()) + .unwrap(); + wait_for_online(&runtime, instance).await; + } + let (_app, deferred) = command_deferred_registry(&directory, port).await; + + // Topic-only: the token routes the command even with two cameras configured. + let request = scoped_request( + CommandVerb::Status, + command_message("sb/status", "addressed-status", json!({})), + Some("camera-b"), + ) + .expect("a topic-only instance routes the command"); + let status = immediate_success( + runtime + .handle_camera_command("sb/status", request, deferred.clone()) + .await, + ); + assert_eq!( + status.get("instance"), + Some(&json!("camera-b")), + "the addressed camera answers its own status" + ); + + // An unknown topic token still routes by the token and is refused by the registry. + let request = scoped_request( + CommandVerb::Status, + command_message("sb/status", "addressed-ghost", json!({})), + Some("camera-ghost"), + ) + .expect("an unknown token is a routing question, not a schema error"); + let outcome = runtime + .handle_camera_command("sb/status", request, deferred.clone()) + .await; + match outcome { + CommandOutcome::ImmediateError(error) => { + assert_eq!(error.code, crate::ErrorCode::NoSuchInstance.as_str()); + } + other => panic!("an unknown addressed camera must answer NO_SUCH_INSTANCE, got {other:?}"), + } + runtime.shutdown().await; +} + fn queued_job(config: &AdapterConfig, capture_id: &str) -> crate::catalog::NewJob { let camera = config .instances @@ -3435,6 +3490,7 @@ async fn runtime_config_listener_rejects_invalid_candidates_and_factory_failures )) }) }, + None, ); let mut invalid_raw = core_config_value(directory.path(), &["camera-a"], false); @@ -3527,6 +3583,7 @@ async fn runtime_config_listener_does_not_mutate_when_event_facade_preparation_f )) }) }, + None, ); let error = match listener @@ -3974,6 +4031,7 @@ async fn rejected_reload_preflight_keeps_the_prior_supervisor_serving_captures() Arc::new(|_instance, _config| -> edgecommons::Result { unreachable!("the application facade rejection occurs first") }), + None, ); let candidate = Arc::new(core_config(directory.path(), &["camera-a"], false)); assert!( @@ -4067,6 +4125,7 @@ async fn failed_reload_transition_restores_prior_config_and_capture_service() { apps: BTreeMap::new(), events: BTreeMap::new(), checkpoint: Some(checkpoint), + commands: None, }; assert!( @@ -4172,6 +4231,7 @@ async fn runtime_config_listener_refreshes_retained_facades_and_applies_roster_a .events()) }) }, + None, ); let roster_candidate = Arc::new(core_config( @@ -4864,6 +4924,19 @@ async fn runtime_startup_router_and_deferred_capture_flows_use_real_core_facades "router registration omitted required camera verb {verb}" ); } + // `sb/discover` is registered through the scoped path, so its configuration-conditional + // describe availability can be published (disabled) and cleared (available) — the direct + // call proves the registered-verb precondition `apply_discover_availability` relies on. + inbox + .set_command_availability( + CommandVerb::Discover.as_str(), + AVAILABILITY_DISABLED, + Some("probe"), + ) + .expect("sb/discover is a registered verb - availability applies to it"); + apply_discover_availability(&inbox, true); + apply_discover_availability(&inbox, false); + apply_discover_availability(&inbox, true); let deferred = inbox.deferred_replies(); match router