From 63a355bbbedc56c603fc66f5b34dff1aa653fd6f Mon Sep 17 00:00:00 2001 From: breis Date: Mon, 27 Jul 2026 17:20:44 -0400 Subject: [PATCH] Adopt core 0.5.0: declared verb scope; scoped deferred captures close D-CAM-29 Pin edgecommons at rust-lib/v0.5.0 (a14a3285) and regenerate the committed git-sourced Cargo.lock. Every verb registers through the breaking two-form surface register/register_outcome with a declared CommandScope (SOUTHBOUND 2.2 / D-SC-2), derived from its closed request schema: - Component: sb/list, sb/discover, sb/capture-group, sb/capture-group-submit, sb/capture-cancel. The library refuses any instance addressing, replacing the adapter's hand-rolled refusal. - Instance: sb/capture, sb/capture-submit, sb/reconnect, sb/ptz, sb/ptz-presets, sb/pause, sb/resume. - Both: sb/status, sb/capture-status, sb/queue-status, sb/queue-clear - no addressing means the whole component. The deferred verbs are scoped too, closing the D-CAM-29 recorded gap: sb/capture (Instance) and sb/capture-group (Component) register through the scoped outcome form, so the topic token routes a deferred capture while deferred settlement, permit release, and sb/capture-cancel settling the held reply are unchanged. The scoped_request enforcement layer is deleted - addressing (conflict-first BAD_ARGS, Component-scope rejection) is library-owned ahead of dispatch (D-SC-4). addressed_request keeps only the seeding into the body selector that the optional-iff-one configured default and the NO_SUCH_INSTANCE existence check read. Keepalive instance state (D-SC-7): the state keepalive's instances[] state comes from the single state model that answers sb/status - a camera paused with sb/pause reports PAUSED while connected keeps reporting reachability. The exact wire element is pinned via the now-public InstanceConnectivity::to_json. Tests: scope-declaration and seeding units, an addressed deferred-capture routing test, a PAUSED keepalive wire pin, named dispatch-pipeline coverage, and a live-inbox integration over a MessagingService loopback that asserts the byte-pinned library refusals through the adapter's real registrations. The k8s deployment-config contract test now normalizes CRLF so it also runs on Windows autocrlf checkouts. Docs: messaging-interface scope table and addressing rules, metrics keepalive state vocabulary incl. PAUSED, DESIGN D-CAM-29 closed by new D-CAM-30, AGENTS invariants. --- AGENTS.md | 8 +- Cargo.lock | 4 +- Cargo.toml | 6 +- DESIGN.md | 28 +- docs/reference/messaging-interface.md | 29 +- docs/reference/metrics.md | 2 +- src/runtime.rs | 213 ++++++------ src/runtime/command.rs | 18 +- src/runtime/tests/mod.rs | 110 +++--- src/runtime/tests/simulator_runtime.rs | 267 ++++++++++++++- .../simulator_runtime/coverage_command.rs | 324 ++++++++++++++++++ tests/deployment_config.rs | 3 +- 12 files changed, 816 insertions(+), 196 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 3a73d7d..140ea77 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -56,8 +56,12 @@ 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; - an instance-addressed command topic routes by its token, which is authoritative (SOUTHBOUND.md §2.2). +- Instance routing is D-EIP-13/D-U28 with declared verb scope (SOUTHBOUND.md §2.2 / D-SC-2): every + verb — deferred captures included — registers a `CommandScope` (`Component`/`Instance`/`Both`) and + the library enforces the addressing before dispatch (conflict-first `BAD_ARGS`; a `Component` verb + refuses any instance addressing). The topic token is authoritative; a body `instance` is optional + iff exactly one camera is configured, and that default plus the `NO_SUCH_INSTANCE` existence check + stay adapter-side (D-SC-4). - 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 3bf44a9..2c0beea 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -790,8 +790,8 @@ checksum = "92773504d58c093f6de2459af4af33faa518c13451eb8f2b5698ed3d36e7c813" [[package]] name = "edgecommons" -version = "0.4.0" -source = "git+https://github.com/edgecommons/edgecommons.git?rev=ef4c6248eec9cabc0acfc88eb0698528aa83a3ab#ef4c6248eec9cabc0acfc88eb0698528aa83a3ab" +version = "0.5.0" +source = "git+https://github.com/edgecommons/edgecommons.git?rev=a14a3285573ef2bb6a531e1e1936c6dc40a85ef4#a14a3285573ef2bb6a531e1e1936c6dc40a85ef4" dependencies = [ "aes-gcm", "arc-swap", diff --git a/Cargo.toml b/Cargo.toml index 99bc38d..ee4978f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -48,10 +48,10 @@ native-all = ["genicam", "rtsp"] capacity-harness = [] [dependencies] -# Pinned to a published rev (rust-lib/v0.4.0) so CI (a standalone repo) can resolve it. Local dev builds against +# Pinned to a published rev (rust-lib/v0.5.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 = "ef4c6248eec9cabc0acfc88eb0698528aa83a3ab", default-features = false, features = ["credentials"] } +edgecommons = { git = "https://github.com/edgecommons/edgecommons.git", rev = "a14a3285573ef2bb6a531e1e1936c6dc40a85ef4", 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 = "ef4c6248eec9cabc0acfc88eb0698528aa83a3ab", default-features = false, features = ["credentials", "standalone"] } +edgecommons = { git = "https://github.com/edgecommons/edgecommons.git", rev = "a14a3285573ef2bb6a531e1e1936c6dc40a85ef4", 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 afe65fa..9a6852f 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. **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-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. **Superseded in turn by the declared-scope model** (core 0.5.0, D-SC-1..4): every verb registers a `CommandScope` and every handler — deferred included — receives the library-resolved addressed instance; see D-CAM-30. | | 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,7 +145,8 @@ 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-29 | Core 0.4.0 adoption: scoped instance routing + conditional availability | Adopt core 0.4.0's immediate-reply scoped registration: the delivery topic's `{instance}` token authoritative for every immediate-reply verb, adapter-side conflict refusal, and `sb/discover`'s configuration-conditional availability 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 — this part of the decision stands unchanged). `receivedTs` is **N/A**: the adapter is a direct camera client with no upstream broker hop to stamp a receive time. PTZ capability is per-camera and runtime-discovered, so it is deliberately NOT mirrored into component-scope availability. | **The recorded gap is CLOSED by D-CAM-30 (core 0.5.0).** Core 0.4.0 exposed the addressed-instance token only to the immediate-reply registration, so `sb/capture`/`sb/capture-group` routed by the body and the topic token did not route them — surfaced up front here as a gap needing a core scoped-outcome registration. Core 0.5.0's breaking two-form surface delivers exactly that: the deferred verbs now receive the addressed instance like every other verb, with deferred settlement unchanged. The adapter-side routing/conflict layer this entry introduced (`register_scoped` + `scoped_request`) is deleted — addressing enforcement is library-owned. | +| D-CAM-30 | Core 0.5.0 adoption: declared verb scope + scoped deferred captures + keepalive instance state | Pin `edgecommons` at rust-lib/v0.5.0 (a14a3285). Every verb registers through the two-form surface `register(verb, scope, handler)` / `register_outcome(verb, scope, handler)` with a declared `CommandScope` (SOUTHBOUND §2.2 / D-SC-2), derived from its closed request schema: **`Component`** for the selector-less verbs (`sb/list`, `sb/discover`, `sb/capture-group`, `sb/capture-group-submit`, `sb/capture-cancel` — fleet answers, `instances[]` targets, durable capture/group ids), **`Instance`** for per-camera actuation (`sb/capture`, `sb/capture-submit`, `sb/reconnect`, `sb/ptz`, `sb/ptz-presets`, `sb/pause`, `sb/resume`), **`Both`** for the dual-semantics verbs where no addressing means the whole component (`sb/status` every camera, `sb/queue-status` the fleet, `sb/capture-status` component-wide lookups, `sb/queue-clear` the `allCameras` drain). The library enforces addressing ahead of dispatch (conflict-first `BAD_ARGS`, `Component`-scope rejection, D-SC-4); the adapter's hand-rolled `scoped_request` layer is deleted, keeping only the D-SC-4 component-side policies: the optional-iff-one configured-camera default and `NO_SUCH_INSTANCE` for an unknown name (`addressed_request` seeds the library-resolved token into the body selector those policies read). **The deferred verbs are scoped too — closing the D-CAM-29 gap:** `sb/capture` (`Instance`) and `sb/capture-group` (`Component`) register through the scoped outcome form, so the topic token routes a deferred capture while deferred settlement, permit release, and `sb/capture-cancel` settling the held reply are unchanged. Companion (D-SC-7): the state keepalive's `instances[]` `state` comes from the single instance state model that answers `sb/status` — a paused camera reports `PAUSED` (shared `CONNECTING`/`ONLINE`/`BACKOFF`/`PAUSED` vocabulary) while `connected` keeps reporting reachability; the exact wire element is pinned via the now-public `InstanceConnectivity::to_json`. | The 0.4.0 model needed an adapter-side enforcement layer and still left the two most consequential verbs blind to the envelope. With the declared scope the library owns addressing for every registration form, the camera class of gap is structurally impossible (D-SC-1), and `describe` advertises each verb's scope for the console. Dual-meaning verbs gain first-class component-wide semantics instead of overloading "no instance named". `PAUSED` in the keepalive lets a console distinguish expected-quiet from silently-stale (D-SC-8) without a second bookkeeping path. | | 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 @@ -981,12 +982,15 @@ reduces the channel budget. 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). +traffic and absent for component/global-scoped traffic, and there is no `main` sentinel. Every verb +declares a `CommandScope` at registration (core 0.5.0, D-SC-2, D-CAM-30) and the library enforces the +addressing ahead of dispatch: component-scope `ecv1/{device}/camera-adapter/cmd/sb/{verb}` selects a +camera by the body `instance` field, 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`), and a `Component`-scoped verb refuses any +instance addressing. The deferred verbs are scoped exactly like the immediate ones: `sb/capture` +routes by the addressed instance and `sb/capture-group` is component-scoped, both through the scoped +outcome registration. ### 12.2 Core prerequisites @@ -2509,10 +2513,10 @@ Reviewers should explicitly decide: hardware-certified release must select models for the compatibility matrix. 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). + sentinel), and `core/docs/SOUTHBOUND.md` §2.2 specifies addressed-instance routing with a declared + per-verb `CommandScope` (D-SC-2). The adapter serves both scopes via the core 0.5.0 two-form + registration (D-CAM-30): the topic instance token is authoritative for every verb, deferred + captures included — the D-CAM-29 recorded gap is closed. 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 1764a14..c6a5ae7 100644 --- a/docs/reference/messaging-interface.md +++ b/docs/reference/messaging-interface.md @@ -9,9 +9,11 @@ ecv1/{device}/camera-adapter/{instance}/cmd/sb/{verb} (instance-addressed) 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). +body `instance` that disagrees with it is refused with `BAD_ARGS`. Every verb declares a **scope** — +`component`, `instance`, or `both` — advertised in the built-in `describe` verb's per-command `scope` +field and enforced before the verb's handler runs. 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 @@ -24,12 +26,19 @@ These rules apply to every verb below. 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. +- **Command scope.** Each verb's declared scope decides which addressing it accepts: + + | Scope | Verbs | Addressing | + |---|---|---| + | `component` | `sb/list`, `sb/discover`, `sb/capture-group`, `sb/capture-group-submit`, `sb/capture-cancel` | Component-scope topic only. Any instance addressing — a topic `{instance}` token or a body `instance` — is refused with `BAD_ARGS`. | + | `instance` | `sb/capture`, `sb/capture-submit`, `sb/reconnect`, `sb/ptz`, `sb/ptz-presets`, `sb/pause`, `sb/resume` | Targets one camera: the topic token, else the body `instance`, else the single-camera omission rule. | + | `both` | `sb/status`, `sb/capture-status`, `sb/queue-status`, `sb/queue-clear` | An addressed camera narrows the answer; no addressing at all means the whole component (every camera / the whole fleet). | + +- **Instance addressing.** A topic `{instance}` token and a body `instance` that are both present + and different are refused with `BAD_ARGS` — checked first, for every scope. At an `instance` or + `both` verb the topic token is authoritative and routes the command with no body `instance` + needed; an unknown addressed camera is `NO_SUCH_INSTANCE`. This applies to the deferred verbs + (`sb/capture`) exactly as to the immediate ones. - **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 @@ -60,7 +69,7 @@ reaches a terminal state, then settles with the full terminal body. | Field | Type | Required | Meaning | |---|---|---|---| -| `instance` | string | optional* | Target camera (*single-camera omission rule). | +| `instance` | string | optional* | Target camera (*single-camera omission rule). On an instance-addressed topic the topic token selects the camera and no body `instance` is needed. | | `requestId` | string | **yes** | Durable idempotency key, 1–256 bytes. | | `captureProfile` | string | optional | Named profile (≤128 bytes); defaults to the camera's `defaultCaptureProfile`. Unknown → `UNKNOWN_CAPTURE_PROFILE`. | | `timeoutMs` | u64 | optional | 1000–1800000. Defaults to the profile's `timeoutMs`, else `global.timeouts.jobTerminalMs`. | diff --git a/docs/reference/metrics.md b/docs/reference/metrics.md index 510b1bf..e3034b8 100644 --- a/docs/reference/metrics.md +++ b/docs/reference/metrics.md @@ -84,7 +84,7 @@ camera has dropped from the keepalive rather than by polling `sb/list` or `sb/st |---|---| | `instance` | The camera ID. | | `connected` | True only while the camera's protocol session is online. The normalized flag any consumer can act on. | -| `state` | The camera's own condition token: `ONLINE`, `CONNECTING`, `BACKOFF`, `OFFLINE`, `DEGRADED`, `DISABLED`, `STOPPING`. `BACKOFF` and `CONNECTING` are both `connected: false`, and they call for different responses. | +| `state` | The camera's condition token: `ONLINE`, `CONNECTING`, `BACKOFF`, `PAUSED`, `OFFLINE`, `DEGRADED`, `DISABLED`, `STOPPING`. `BACKOFF` and `CONNECTING` are both `connected: false`, and they call for different responses. A camera paused with `sb/pause` reports `PAUSED` — deliberately quiet, not stale — while `connected` keeps reporting reachability, because pause suspends capture workload, not the session. The token comes from the same state model that answers `sb/status`, so the pushed and the pulled answer cannot disagree. | | `detail` | Why the camera is down, in its own words, when it has reported an error. A healthy camera carries none. | | `attributes` | Camera-specific data: `backend`, the connection `generation`, and `lastErrorCode` when an error is known. | diff --git a/src/runtime.rs b/src/runtime.rs index a4f421e..45a603c 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -45,7 +45,7 @@ use std::time::Duration; use async_trait::async_trait; use edgecommons::commands::{ AVAILABILITY_AVAILABLE, AVAILABILITY_DISABLED, CommandError, CommandInbox, CommandOutcome, - DeferredReplyRegistry, DeferredReplyToken, outcome_handler, scoped_command_handler, + CommandScope, DeferredReplyRegistry, DeferredReplyToken, command_handler, outcome_handler, }; use edgecommons::config::{ Config, ConfigurationApplicationError, ConfigurationApplicationResult, @@ -2250,33 +2250,47 @@ impl CommandVerb { /// 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. + /// These register through `register_outcome`, keeping the deferred settlement (the dispatch + /// permit is released while a capture runs; `sb/capture-cancel` settles the pending reply). + /// The outcome registration receives the addressed instance exactly like the immediate one + /// (core 0.5.0, D-SC-1), so deferral costs no addressing fidelity. #[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 [`CommandScope`] this verb declares at registration (D-SC-2), derived from its closed + /// request schema. The library enforces it ahead of dispatch: a topic/body instance conflict + /// and any instance addressing at a `Component` verb are refused with `BAD_ARGS` before a + /// handler runs. /// - /// 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. + /// - `Component` — the schema carries no camera selector: `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. + /// - `Instance` — per-camera actuation; an unaddressed delivery falls back to the adapter's + /// optional-iff-one configured-camera default (D-SC-4). + /// - `Both` — dual-semantics: an addressed camera narrows the answer, no addressing means the + /// whole component (`sb/status` answers every camera, `sb/queue-status` the whole fleet, + /// `sb/capture-status` component-wide lookups, `sb/queue-clear` the `allCameras` drain). #[must_use] - pub const fn instance_routable(self) -> bool { - !matches!( - self, + pub const fn scope(self) -> CommandScope { + match self { Self::List - | Self::Discover - | Self::CaptureGroup - | Self::CaptureGroupSubmit - | Self::CaptureCancel - ) + | Self::Discover + | Self::CaptureGroup + | Self::CaptureGroupSubmit + | Self::CaptureCancel => CommandScope::Component, + Self::Capture + | Self::CaptureSubmit + | Self::Reconnect + | Self::Ptz + | Self::PtzPresets + | Self::Pause + | Self::Resume => CommandScope::Instance, + Self::Status | Self::CaptureStatus | Self::QueueStatus | Self::QueueClear => { + CommandScope::Both + } + } } } @@ -2286,63 +2300,41 @@ 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. +/// Seeds the library-resolved `addressed_instance` (SOUTHBOUND §2.2 / D-SC-4) into the request +/// body's `instance` selector, so the adapter's existing configured-default and unknown-camera +/// resolution (`resolve_actuation_instance` / `resolve_instance`, answering `NO_SUCH_INSTANCE` +/// for an unknown name) routes the command. /// -/// - **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, +/// The library owns addressing ahead of dispatch: a `body.instance` conflicting with the topic +/// token, and any instance addressing at a `Component`-scoped verb, are refused with `BAD_ARGS` +/// before a handler runs — neither can reach this function. What remains adapter-side is exactly +/// the split D-SC-4 assigns it: the optional-iff-one configured-camera default and the +/// instance-existence check, both applied by the body routing this seeds. +fn addressed_request( mut request: Message, addressed: Option<&str>, ) -> std::result::Result { - let Some(topic) = addressed else { + let Some(instance) = 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) + match &mut request.body { + serde_json::Value::Object(map) => { + // An existing `body.instance` is identical to the addressed one — the library has + // already refused a conflict — so the insert only fills an absent selector. + map.entry("instance".to_string()) + .or_insert_with(|| serde_json::Value::String(instance.to_string())); + } + body @ serde_json::Value::Null => { + *body = serde_json::json!({ "instance": instance }); + } + _ => { + 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`. @@ -3042,12 +3034,12 @@ impl RuntimeCommandRouter { /// 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). + /// Every verb declares its [`CommandScope`] at registration (SOUTHBOUND §2.2 / D-SC-2), and + /// every handler — immediate and deferred alike — receives the library-resolved addressed + /// instance (D-SC-1): the delivery topic's `{instance}` token is authoritative, a + /// conflicting `body.instance` and any instance addressing at a `Component` verb are refused + /// by the library before dispatch, and [`addressed_request`] seeds the resolved token into + /// the body so the adapter's configured-default/unknown-camera resolution routes it. /// /// A registration failure is fatal to component construction; a partial command surface is /// never exposed as active. @@ -3058,32 +3050,23 @@ impl RuntimeCommandRouter { 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 } + verb.scope(), + outcome_handler(move |request, deferred, addressed| { + Arc::clone(&router).dispatch_outcome(verb, request, deferred, addressed) }), )?; } else { let deferred_registry = deferred_registry.clone(); - inbox.register_scoped( + inbox.register( 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", - )), - } - } + verb.scope(), + command_handler(move |request, addressed| { + Arc::clone(&router).dispatch_immediate( + verb, + request, + deferred_registry.clone(), + addressed, + ) }), )?; } @@ -3125,6 +3108,44 @@ impl RuntimeCommandRouter { self.stopping.store(true, Ordering::Release); } + /// The registered pipeline for a deferred-capable verb: seed the library-resolved addressed + /// instance into the body selector ([`addressed_request`]), then delegate. Named rather than + /// inlined in the registration closure so the pipeline the inbox invokes is directly + /// testable. + async fn dispatch_outcome( + self: Arc, + verb: CommandVerb, + request: Message, + deferred: DeferredReplyRegistry, + addressed: Option, + ) -> CommandOutcome { + match addressed_request(request, addressed.as_deref()) { + Ok(request) => self.dispatch(verb.as_str(), request, deferred).await, + Err(error) => CommandOutcome::ImmediateError(error), + } + } + + /// The registered pipeline for an immediate verb: seed the addressing, delegate, and map the + /// runtime's [`CommandOutcome`] onto the immediate handler contract. A deferred settlement + /// out of a verb that does not declare one is a wiring fault, answered as `BACKEND_ERROR`. + async fn dispatch_immediate( + self: Arc, + verb: CommandVerb, + request: Message, + deferred: DeferredReplyRegistry, + addressed: Option, + ) -> std::result::Result, CommandError> { + let request = addressed_request(request, addressed.as_deref())?; + match self.dispatch(verb.as_str(), request, deferred).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", + )), + } + } + async fn dispatch( &self, verb: &'static str, diff --git a/src/runtime/command.rs b/src/runtime/command.rs index ba060af..5407440 100644 --- a/src/runtime/command.rs +++ b/src/runtime/command.rs @@ -708,6 +708,13 @@ impl CameraRuntime { /// /// The same element shape answers core's built-in `status` verb, so one sampler serves both the /// push and the pull. + /// + /// D-SC-7: the `state` token comes from the single instance state model that answers + /// `sb/status` — the registry's connection lifecycle plus the operator pause flag, never a + /// second bookkeeping path. A deliberately paused camera reports `PAUSED` (the shared + /// `CONNECTING`/`ONLINE`/`BACKOFF`/`PAUSED` keepalive vocabulary), so a console can tell + /// expected-quiet from silently-stale; `connected` still reports reachability, because pause + /// suspends capture workload, not the session. #[must_use] pub fn camera_connectivity(&self) -> Vec { let Ok(snapshots) = self.registry.snapshots(MAX_CONNECTIVITY_INSTANCES) else { @@ -716,6 +723,7 @@ impl CameraRuntime { snapshots .into_iter() .map(|snapshot| { + let paused = self.is_paused(&snapshot.instance); let connected = snapshot.state == CameraConnectionState::Online; let mut attributes = serde_json::Map::new(); attributes.insert( @@ -732,9 +740,13 @@ impl CameraRuntime { serde_json::Value::from(error.code.clone()), ); } - let state = serde_json::to_value(snapshot.state) - .ok() - .and_then(|token| token.as_str().map(str::to_owned)); + let state = if paused { + Some("PAUSED".to_owned()) + } else { + serde_json::to_value(snapshot.state) + .ok() + .and_then(|token| token.as_str().map(str::to_owned)) + }; let detail = snapshot .last_error .as_ref() diff --git a/src/runtime/tests/mod.rs b/src/runtime/tests/mod.rs index 04dc0c3..3c38723 100644 --- a/src/runtime/tests/mod.rs +++ b/src/runtime/tests/mod.rs @@ -609,7 +609,7 @@ 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) -------------------------------------- +// --- SOUTHBOUND §2.2 addressed-instance routing (D-U28 / D-SC-1..4) -------------------------- fn scoped_fixture_message(body: serde_json::Value) -> Message { edgecommons::messaging::MessageBuilder::new("sb/status", "1.0") @@ -617,71 +617,59 @@ fn scoped_fixture_message(body: serde_json::Value) -> Message { .build() } -/// Topic-only (§2.2): the delivery topic's instance token becomes the routing selector the -/// registry resolves — no `body.instance` needed. +/// The library-resolved addressed instance becomes the routing selector the registry resolves — +/// no `body.instance` needed. Conflict refusal and component-scope rejection are library-owned +/// (core 0.5.0, D-SC-4) and never reach this seeding step. #[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"); +fn the_addressed_instance_is_seeded_as_the_routing_selector() { + let request = addressed_request(scoped_fixture_message(json!({})), Some("camera-b")) + .expect("an addressed 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, + // An agreeing `body.instance` (the only kind the library lets through) passes untouched. + let request = addressed_request( 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")), + // A null body grows an object carrying the selector. + let request = addressed_request( + scoped_fixture_message(serde_json::Value::Null), Some("camera-b"), ) - .expect_err("a non-object body cannot be instance-addressed"); - assert_eq!(error.code, crate::ErrorCode::BadArgs.as_str()); -} + .expect("a null body can be instance-addressed"); + assert_eq!(request.body, json!({ "instance": "camera-b" })); -/// 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"); + // A non-object body cannot carry the routing selector. + let error = addressed_request(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()); - 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. +/// An unaddressed delivery passes through unchanged: the existing `body`-driven routing +/// (including the optional-iff-one configured-camera default) applies. #[test] -fn component_scope_keeps_body_instance_routing() { - let request = scoped_request( - CommandVerb::Status, +fn an_unaddressed_delivery_keeps_body_instance_routing() { + let request = addressed_request( scoped_fixture_message(json!({ "instance": "camera-a" })), None, ) - .expect("component-scoped requests pass through"); + .expect("unaddressed 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"); + let request = addressed_request(scoped_fixture_message(json!({})), None) + .expect("an empty unaddressed 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. +/// Every verb's declared scope (D-SC-2) matches its closed request schema, and the deferred set +/// is exactly the two capture verbs. The library enforces the declaration ahead of dispatch, so +/// this classification IS the addressing contract: `Component` verbs refuse any instance +/// addressing, `Instance`/`Both` verbs route by the topic token first. #[test] -fn verb_scoping_classification_matches_the_request_schemas() { +fn verb_scope_declarations_match_the_request_schemas() { for verb in CommandVerb::ALL { assert_eq!( verb.defers(), @@ -689,22 +677,32 @@ fn verb_scoping_classification_matches_the_request_schemas() { "{} deferral classification", verb.as_str() ); + let expected = match verb { + // No camera selector in the schema: fleet answers, `instances[]` targets, and + // durable capture/group ids. + CommandVerb::List + | CommandVerb::Discover + | CommandVerb::CaptureGroup + | CommandVerb::CaptureGroupSubmit + | CommandVerb::CaptureCancel => CommandScope::Component, + // Dual-semantics: an addressed camera narrows the answer; no addressing means the + // whole component. + CommandVerb::Status + | CommandVerb::CaptureStatus + | CommandVerb::QueueStatus + | CommandVerb::QueueClear => CommandScope::Both, + // Per-camera actuation, deferred captures included (core 0.5.0 hands the outcome + // registration the same addressing). + CommandVerb::Capture + | CommandVerb::CaptureSubmit + | CommandVerb::Reconnect + | CommandVerb::Ptz + | CommandVerb::PtzPresets + | CommandVerb::Pause + | CommandVerb::Resume => CommandScope::Instance, + }; + assert_eq!(verb.scope(), expected, "{} declared scope", 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)] diff --git a/src/runtime/tests/simulator_runtime.rs b/src/runtime/tests/simulator_runtime.rs index 85f6799..2dfc8ff 100644 --- a/src/runtime/tests/simulator_runtime.rs +++ b/src/runtime/tests/simulator_runtime.rs @@ -795,9 +795,10 @@ 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. +/// delivery topic's `{instance}` token, D-U28, resolved by the library and seeded via +/// `addressed_request`) routes to the addressed camera with no `body.instance`, and an unknown +/// token answers the standardized `NO_SUCH_INSTANCE` — proving the library-resolved addressing +/// 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; @@ -812,13 +813,12 @@ async fn an_instance_addressed_command_routes_by_the_topic_token() { } 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, + // Addressed: the token routes the command even with two cameras configured. + let request = addressed_request( command_message("sb/status", "addressed-status", json!({})), Some("camera-b"), ) - .expect("a topic-only instance routes the command"); + .expect("an addressed instance routes the command"); let status = immediate_success( runtime .handle_camera_command("sb/status", request, deferred.clone()) @@ -830,9 +830,8 @@ async fn an_instance_addressed_command_routes_by_the_topic_token() { "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, + // An unknown addressed token still routes by the token and is refused by the registry. + let request = addressed_request( command_message("sb/status", "addressed-ghost", json!({})), Some("camera-ghost"), ) @@ -849,6 +848,190 @@ async fn an_instance_addressed_command_routes_by_the_topic_token() { runtime.shutdown().await; } +/// D-CAM-29 closed: the deferred `sb/capture` is scoped too. The addressed camera routes the +/// capture with no `body.instance` — through the production deferred dispatch, keeping the +/// deferred settlement — and the durable job lands on the ADDRESSED camera. +#[tokio::test] +async fn an_instance_addressed_deferred_capture_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; + + // The body names no camera; only the library-resolved topic token selects camera-b. With two + // cameras configured, body-only routing would have refused this as ambiguous. + let request = addressed_request( + command_message( + "sb/capture", + "addressed-capture", + json!({ "requestId": "addressed-capture-1" }), + ), + Some("camera-b"), + ) + .expect("an addressed instance routes the deferred capture"); + let outcome = runtime + .handle_camera_command("sb/capture", request, deferred.clone()) + .await; + let CommandOutcome::DeferredWithContinuation { continuation, .. } = outcome else { + panic!("an addressed capture must keep the deferred settlement path"); + }; + continuation + .await + .expect("the addressed capture must be durably accepted"); + + let job = runtime + .catalog + .job_by_ledger( + crate::catalog::LedgerKey::new("camera-b", "sb/capture", "addressed-capture-1") + .unwrap(), + ) + .await + .unwrap() + .expect("the capture must be ledgered under the addressed camera"); + assert_eq!( + job.instance, "camera-b", + "the topic token, not a body default, selected the camera" + ); + runtime.shutdown().await; +} + +/// The registered per-verb pipelines — the exact functions the inbox invokes for every delivery — +/// seed the library-resolved addressing before dispatch and map the runtime's outcome onto each +/// registration contract: immediate success/error onto the immediate handler result, an +/// addressing seed fault onto an immediate refusal on both forms, and a deferred settlement out +/// of an immediate verb onto the `BACKEND_ERROR` wiring fault. +#[tokio::test] +async fn the_registered_dispatch_pipelines_seed_addressing_and_map_outcomes() { + let (port, _broker) = spawn_recording_mqtt_broker().await; + let directory = TempDir::new().unwrap(); + let (_app, deferred) = command_deferred_registry(&directory, port).await; + + // An uninstalled router answers the stable startup-unavailable error through the immediate + // mapping. + let router = RuntimeCommandRouter::new(); + let error = Arc::clone(&router) + .dispatch_immediate( + CommandVerb::List, + command_message("sb/list", "pipeline-uninstalled", json!({})), + deferred.clone(), + None, + ) + .await + .expect_err("an uninstalled router must answer an immediate error"); + assert_eq!(error.code, crate::ErrorCode::DeviceUnavailable.as_str()); + + // An addressing seed fault (a non-object body under an addressed delivery) is refused before + // dispatch, on the immediate and the outcome pipeline alike. + let error = Arc::clone(&router) + .dispatch_immediate( + CommandVerb::Status, + command_message("sb/status", "pipeline-junk", json!("junk")), + deferred.clone(), + Some("camera-a".to_string()), + ) + .await + .expect_err("a non-object body cannot be instance-addressed"); + assert_eq!(error.code, crate::ErrorCode::BadArgs.as_str()); + match Arc::clone(&router) + .dispatch_outcome( + CommandVerb::Capture, + command_message("sb/capture", "pipeline-junk-deferred", json!("junk")), + deferred.clone(), + Some("camera-a".to_string()), + ) + .await + { + CommandOutcome::ImmediateError(error) => { + assert_eq!(error.code, crate::ErrorCode::BadArgs.as_str()); + } + other => panic!("a non-object deferred body must be refused, got {other:?}"), + } + + // A service that echoes what it was dispatched proves the pipelines seed the addressed + // instance into the body selector before delegating. + struct EchoService; + #[async_trait] + impl CameraCommandService for EchoService { + async fn handle_camera_command( + &self, + verb: &'static str, + request: Message, + _deferred: DeferredReplyRegistry, + ) -> CommandOutcome { + CommandOutcome::ImmediateSuccess(Some(json!({ "verb": verb, "body": request.body }))) + } + } + router.install(Arc::new(EchoService)).unwrap(); + let value = Arc::clone(&router) + .dispatch_immediate( + CommandVerb::Status, + command_message("sb/status", "pipeline-seeded", json!({})), + deferred.clone(), + Some("camera-b".to_string()), + ) + .await + .expect("the immediate pipeline maps success through") + .expect("the echo service always answers a value"); + assert_eq!(value["verb"], json!("sb/status")); + assert_eq!(value["body"], json!({ "instance": "camera-b" })); + match Arc::clone(&router) + .dispatch_outcome( + CommandVerb::Capture, + command_message( + "sb/capture", + "pipeline-seeded-deferred", + json!({ "requestId": "pipeline-1" }), + ), + deferred.clone(), + Some("camera-b".to_string()), + ) + .await + { + CommandOutcome::ImmediateSuccess(Some(value)) => { + assert_eq!(value["body"]["instance"], json!("camera-b")); + assert_eq!(value["body"]["requestId"], json!("pipeline-1")); + } + other => panic!("the outcome pipeline must hand the seeded request through, got {other:?}"), + } + + // A verb registered immediate must never settle through a deferred path; the mapping answers + // the wiring fault as BACKEND_ERROR instead of leaving an open token behind. + struct DeferringService; + #[async_trait] + impl CameraCommandService for DeferringService { + async fn handle_camera_command( + &self, + _verb: &'static str, + request: Message, + deferred: DeferredReplyRegistry, + ) -> CommandOutcome { + match deferred.defer(&request, Duration::from_secs(5)) { + Ok(token) => CommandOutcome::Deferred(token), + Err(error) => CommandOutcome::ImmediateError(error), + } + } + } + let wired = RuntimeCommandRouter::new(); + wired.install(Arc::new(DeferringService)).unwrap(); + let error = Arc::clone(&wired) + .dispatch_immediate( + CommandVerb::List, + command_message("sb/list", "pipeline-deferred-fault", json!({})), + deferred.clone(), + None, + ) + .await + .expect_err("a deferred settlement out of an immediate verb is a wiring fault"); + assert_eq!(error.code, crate::ErrorCode::BackendError.as_str()); +} + fn queued_job(config: &AdapterConfig, capture_id: &str) -> crate::catalog::NewJob { let camera = config .instances @@ -7362,6 +7545,70 @@ async fn every_camera_reports_its_reachability_to_the_heartbeat() { runtime.shutdown().await; } +/// D-SC-7: a deliberately paused camera is distinguishable on the passive surface. The keepalive +/// `state` comes from the same instance state model that answers `sb/status` — the registry's +/// lifecycle plus the operator pause flag — so a paused camera reports `PAUSED` (the shared +/// `CONNECTING`/`ONLINE`/`BACKOFF`/`PAUSED` vocabulary) while `connected` keeps reporting +/// reachability, because pause suspends capture workload, not the session. The exact wire +/// element is pinned through the public `InstanceConnectivity::to_json`. +#[tokio::test] +async fn a_paused_camera_reports_paused_in_the_keepalive_instance_state() { + let directory = TempDir::new().unwrap(); + let runtime = runtime( + config(directory.path(), &["camera-a", "camera-b"], false), + &directory, + ) + .await; + runtime + .start_supervisor("camera-a".to_string(), runtime.engine("camera-a").unwrap()) + .unwrap(); + wait_for_online(&runtime, "camera-a").await; + + assert!(runtime.set_paused("camera-a", true)); + let samples = runtime.camera_connectivity(); + let paused = samples + .iter() + .find(|camera| camera.instance == "camera-a") + .expect("the paused camera must still be reported"); + assert!( + paused.connected, + "pause suspends capture workload, not the session — reachability is unchanged" + ); + assert_eq!( + paused.state.as_deref(), + Some("PAUSED"), + "the keepalive state must say the quiet is deliberate" + ); + + // The exact published element (the shape a console consumes), byte-pinned via the public + // to_json: a healthy-but-paused camera carries no detail and no lastErrorCode. + let generation = runtime.registry.snapshot("camera-a").unwrap().generation; + assert_eq!( + paused.to_json(), + json!({ + "instance": "camera-a", + "connected": true, + "state": "PAUSED", + "attributes": { "backend": "sim", "generation": generation }, + }), + "the wire element must carry PAUSED exactly where the state token rides" + ); + + // Resume restores the single state model's lifecycle token on the same surface. + assert!(runtime.set_paused("camera-a", false)); + let resumed = runtime.camera_connectivity(); + assert_eq!( + resumed + .iter() + .find(|camera| camera.instance == "camera-a") + .and_then(|camera| camera.state.as_deref()), + Some("ONLINE"), + "resume must hand the keepalive back to the connection lifecycle" + ); + + runtime.shutdown().await; +} + /// Q2: the component emitted no metrics at all. /// /// There was not one call site for `metrics()`, `MetricBuilder`, or `MetricService` anywhere diff --git a/src/runtime/tests/simulator_runtime/coverage_command.rs b/src/runtime/tests/simulator_runtime/coverage_command.rs index 209ba4c..54e17af 100644 --- a/src/runtime/tests/simulator_runtime/coverage_command.rs +++ b/src/runtime/tests/simulator_runtime/coverage_command.rs @@ -4935,3 +4935,327 @@ fn the_panel_trio_is_registered_with_the_right_ids_orders_and_scope() { } } } + +// --- Live-inbox declared-scope integration (core 0.5.0, D-CAM-30) ---------------------------- + +/// An in-process `MessagingService` double for driving the REAL command inbox: it captures the +/// inbox's acknowledged subscriptions so a test can deliver command messages through the +/// production dispatch — library scope enforcement, addressing resolution, and the registered +/// per-verb closures — and records every reply the inbox sends back. +#[derive(Default)] +struct InboxLoopbackMessaging { + subscriptions: Mutex)>>, + replies: Mutex>, +} + +impl InboxLoopbackMessaging { + fn unsupported(&self) -> edgecommons::Result { + Err(edgecommons::EdgeCommonsError::Messaging( + "not supported by the inbox loopback double".to_string(), + )) + } + + /// Delivers one message to the handler subscribed under the first filter that matches + /// `topic` (MQTT `+`/`#` semantics — enough for the two D-U28 inbox filters). + async fn deliver(&self, topic: &str, message: Message) { + let handler = { + let subscriptions = self.subscriptions.lock().unwrap(); + subscriptions + .iter() + .find(|(filter, _)| mqtt_filter_matches(filter, topic)) + .map(|(_, handler)| Arc::clone(handler)) + .expect("the inbox must have subscribed a filter matching the delivery topic") + }; + handler.handle(topic.to_string(), message).await; + } + + /// The reply the inbox sent for `correlation_id`. + fn reply_for(&self, correlation_id: &str) -> Message { + self.replies + .lock() + .unwrap() + .iter() + .find(|(correlation, _)| correlation == correlation_id) + .map(|(_, reply)| reply.clone()) + .expect("the inbox must have replied to the delivered request") + } + + fn filters(&self) -> Vec { + self.subscriptions + .lock() + .unwrap() + .iter() + .map(|(filter, _)| filter.clone()) + .collect() + } +} + +/// Minimal MQTT filter matching for the loopback double. +fn mqtt_filter_matches(filter: &str, topic: &str) -> bool { + let mut filter_parts = filter.split('/'); + let mut topic_parts = topic.split('/'); + loop { + match (filter_parts.next(), topic_parts.next()) { + (Some("#"), _) => return true, + (Some("+"), Some(_)) => {} + (Some(expected), Some(actual)) if expected == actual => {} + (None, None) => return true, + _ => return false, + } + } +} + +#[async_trait] +impl edgecommons::messaging::MessagingService for InboxLoopbackMessaging { + async fn publish(&self, _topic: &str, _msg: &Message) -> edgecommons::Result<()> { + Ok(()) + } + async fn publish_northbound( + &self, + _topic: &str, + _msg: &Message, + _qos: edgecommons::messaging::Qos, + ) -> edgecommons::Result<()> { + Ok(()) + } + async fn publish_raw( + &self, + _topic: &str, + _payload: &serde_json::Value, + ) -> edgecommons::Result<()> { + Ok(()) + } + async fn publish_northbound_raw( + &self, + _topic: &str, + _payload: &serde_json::Value, + _qos: edgecommons::messaging::Qos, + ) -> edgecommons::Result<()> { + Ok(()) + } + async fn subscribe( + &self, + filter: &str, + handler: Arc, + _max_messages: usize, + _max_concurrency: usize, + ) -> edgecommons::Result<()> { + self.subscriptions + .lock() + .unwrap() + .push((filter.to_string(), handler)); + Ok(()) + } + async fn subscribe_acknowledged( + &self, + filter: &str, + handler: Arc, + max_messages: usize, + max_concurrency: usize, + _timeout: Duration, + ) -> edgecommons::Result<()> { + self.subscribe(filter, handler, max_messages, max_concurrency) + .await + } + async fn subscribe_northbound( + &self, + _filter: &str, + _handler: Arc, + _qos: edgecommons::messaging::Qos, + _max_messages: usize, + _max_concurrency: usize, + ) -> edgecommons::Result<()> { + Ok(()) + } + async fn unsubscribe(&self, filter: &str) -> edgecommons::Result<()> { + self.subscriptions + .lock() + .unwrap() + .retain(|(subscribed, _)| subscribed != filter); + Ok(()) + } + async fn unsubscribe_northbound(&self, _filter: &str) -> edgecommons::Result<()> { + Ok(()) + } + async fn request( + &self, + _topic: &str, + _msg: Message, + ) -> edgecommons::Result { + self.unsupported() + } + async fn request_northbound( + &self, + _topic: &str, + _msg: Message, + ) -> edgecommons::Result { + self.unsupported() + } + async fn request_with_timeout( + &self, + _topic: &str, + _msg: Message, + _timeout: Option, + ) -> edgecommons::Result { + self.unsupported() + } + async fn request_northbound_with_timeout( + &self, + _topic: &str, + _msg: Message, + _timeout: Option, + ) -> edgecommons::Result { + self.unsupported() + } + async fn reply(&self, request: &Message, reply: Message) -> edgecommons::Result<()> { + self.replies + .lock() + .unwrap() + .push((request.header.correlation_id.clone(), reply)); + Ok(()) + } + async fn reply_northbound( + &self, + _request: &Message, + _reply: Message, + ) -> edgecommons::Result<()> { + Ok(()) + } + fn cancel_request(&self, _reply_future: edgecommons::messaging::ReplyFuture) {} + fn cancel_request_northbound(&self, _reply_future: edgecommons::messaging::ReplyFuture) {} + fn connected(&self) -> bool { + true + } +} + +/// The declared-scope contract through the REAL inbox over the adapter's actual registrations +/// (D-CAM-30): the library resolves the delivery topic's instance token and enforces the scope +/// ahead of dispatch — the byte-pinned conflict and component-scope refusals — and the +/// registered per-verb closures (immediate and deferred alike) receive the addressing and seed +/// it into the body the runtime routes by. +#[tokio::test] +async fn the_live_inbox_enforces_declared_scope_over_the_registered_verbs() { + let messaging = Arc::new(InboxLoopbackMessaging::default()); + let config = Arc::new( + edgecommons::config::Config::from_value(crate::COMPONENT_NAME, "inbox-e2e", json!({})) + .expect("an empty component config is valid"), + ); + let inbox = edgecommons::commands::CommandInbox::new( + Arc::clone(&messaging) as Arc, + config, + Arc::new(|| 0), + Arc::new(|| Box::pin(async { true })), + Arc::new(|| None), + Arc::new(Vec::new), + ); + let router = RuntimeCommandRouter::new(); + router + .register(&inbox) + .expect("every camera verb registers with its declared scope"); + + /// Echoes the dispatched body back, so the reply proves what the pipeline seeded. + struct EchoService; + #[async_trait] + impl CameraCommandService for EchoService { + async fn handle_camera_command( + &self, + verb: &'static str, + request: Message, + _deferred: DeferredReplyRegistry, + ) -> CommandOutcome { + CommandOutcome::ImmediateSuccess(Some(json!({ "verb": verb, "body": request.body }))) + } + } + router.install(Arc::new(EchoService)).unwrap(); + let status = Arc::clone(&inbox).start().await; + assert_eq!( + status.state, + edgecommons::commands::CommandInboxStartupState::Active, + "the loopback inbox must activate" + ); + let instance_prefix = messaging + .filters() + .into_iter() + .find_map(|filter| filter.strip_suffix("+/cmd/#").map(str::to_string)) + .expect("the inbox subscribes the D-U28 instance-scope filter"); + + let request = |verb: &str, suffix: &str, body: serde_json::Value| { + MessageBuilder::new(verb, "1.0") + .correlation_id(format!("live-inbox-{suffix}")) + .reply_to("live-inbox/replies") + .structured_payload(body) + .build() + }; + + // An instance-addressed immediate verb: the token is seeded into the body the runtime + // routes by — exercising the registered immediate closure end to end. + messaging + .deliver( + &format!("{instance_prefix}camera-b/cmd/sb/status"), + request("sb/status", "status", json!({})), + ) + .await; + let reply = messaging.reply_for("live-inbox-status"); + assert_eq!(reply.body["ok"], json!(true)); + assert_eq!(reply.body["result"]["body"], json!({ "instance": "camera-b" })); + + // An instance-addressed DEFERRED verb routes by the token too (the closed D-CAM-29 gap), + // through the registered outcome closure. + messaging + .deliver( + &format!("{instance_prefix}camera-b/cmd/sb/capture"), + request("sb/capture", "capture", json!({ "requestId": "live-1" })), + ) + .await; + let reply = messaging.reply_for("live-inbox-capture"); + assert_eq!(reply.body["ok"], json!(true)); + assert_eq!( + reply.body["result"]["body"], + json!({ "instance": "camera-b", "requestId": "live-1" }) + ); + + // Conflict-first, byte-pinned, library-owned: the handler never runs. + messaging + .deliver( + &format!("{instance_prefix}camera-b/cmd/sb/status"), + request("sb/status", "conflict", json!({ "instance": "camera-a" })), + ) + .await; + let reply = messaging.reply_for("live-inbox-conflict"); + assert_eq!(reply.body["ok"], json!(false)); + assert_eq!(reply.body["error"]["code"], json!("BAD_ARGS")); + assert_eq!( + reply.body["error"]["message"], + json!("instance in body conflicts with the addressed instance") + ); + + // A COMPONENT-scoped verb refuses an instance-addressed delivery... + messaging + .deliver( + &format!("{instance_prefix}camera-b/cmd/sb/list"), + request("sb/list", "component-topic", json!({})), + ) + .await; + let reply = messaging.reply_for("live-inbox-component-topic"); + assert_eq!(reply.body["error"]["code"], json!("BAD_ARGS")); + assert_eq!( + reply.body["error"]["message"], + json!("verb 'sb/list' is component-scoped") + ); + + // ...and a body-named instance at component scope. + messaging + .deliver( + &format!("{instance_prefix}cmd/sb/list"), + request("sb/list", "component-body", json!({ "instance": "camera-a" })), + ) + .await; + let reply = messaging.reply_for("live-inbox-component-body"); + assert_eq!(reply.body["error"]["code"], json!("BAD_ARGS")); + assert_eq!( + reply.body["error"]["message"], + json!("verb 'sb/list' is component-scoped - the body must not name an instance") + ); + + let _ = inbox.stop().await; +} diff --git a/tests/deployment_config.rs b/tests/deployment_config.rs index 52fc217..ea6089a 100644 --- a/tests/deployment_config.rs +++ b/tests/deployment_config.rs @@ -21,7 +21,8 @@ fn docker_simulator_config_is_a_valid_initial_configuration() { #[test] fn kubernetes_configmap_embeds_a_valid_initial_configuration() { - let document = include_str!("../k8s/configmap.yaml"); + // Normalized so the contract check also runs on a CRLF checkout (Windows autocrlf). + let document = include_str!("../k8s/configmap.yaml").replace("\r\n", "\n"); let marker = " config.json: |-\n"; let (_, body) = document .split_once(marker)