diff --git a/AGENTS.md b/AGENTS.md index 14cd741..b4fa659 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -71,6 +71,13 @@ first. minted by the library's `commands()`/`events()` facades — no hand-rolled topic builder, no configurable prefix, no legacy alias (all retired in the UNS-core migration; see `docs/reference/messaging-interface.md`). `state`/`cfg`/`metric` are reserved, library-owned classes. +- **Command verbs declare their addressing scope** (core 0.5.0, `DESIGN-scoped-commands.md` D-SC-2; + register entry in DESIGN §16): `get-status` and `trigger` are `CommandScope::Both` — no instance named + means the component-wide answer, one named means that instance — and `set-activation` is + `CommandScope::Instance` (it has no "all" form; `INSTANCE_REQUIRED`/`UNKNOWN_INSTANCE` stay + component-side per D-SC-4). An instance is named by the topic (the library's per-instance inbox) or by + the `instance` body field; the topic wins, and `src/control.rs`'s `address()` folds it into the body + selector so every handler reads one selector. - Durable **write-ahead** state → crash-safe move+delete with checksum-verify-before-complete. - **Long-outage tolerant** (hours–~2d): time-based `giveUpAfter` (default 7d, not attempt caps, shipped) and resume in-flight (shipped). The **disconnection circuit-breaker** (§13.4) is still **not implemented** — diff --git a/Cargo.lock b/Cargo.lock index be97054..8573c32 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1771,8 +1771,8 @@ dependencies = [ [[package]] name = "edgecommons" -version = "0.2.0" -source = "git+https://github.com/edgecommons/edgecommons.git?rev=36a70c48b65b35f77bfab70d3a73869debdfc407#36a70c48b65b35f77bfab70d3a73869debdfc407" +version = "0.5.0" +source = "git+https://github.com/edgecommons/edgecommons.git?rev=a14a3285573ef2bb6a531e1e1936c6dc40a85ef4#a14a3285573ef2bb6a531e1e1936c6dc40a85ef4" dependencies = [ "aes-gcm 0.10.3", "arc-swap", diff --git a/Cargo.toml b/Cargo.toml index 7e9d747..05ff585 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -74,15 +74,15 @@ dest-azure = ["dep:azure_storage_blobs", "dep:azure_storage", "dep:azure_core", dest-gcs = ["dep:reqwest", "reqwest/json", "dep:base64", "edgecommons/credentials"] [dependencies] -# edgecommons Rust library, pinned to a commit on edgecommons/edgecommons (private). Building standalone needs -# read access to that repo. The fetch goes through the git CLI (see .cargo/config.toml); locally use your -# gh/git credential helper, and CI's reusable workflow rewrites the https URL with EDGECOMMONS_READ_TOKEN. +# edgecommons Rust library, pinned to the `rust-lib/v0.5.0` release tag on edgecommons/edgecommons — a +# public repo, so the fetch needs no credentials in CI or locally. # -# UNS MIGRATION NOTE: this code REQUIRES the UNS core (top-level `identity`, `gg.uns()`, `gg.commands()`, -# the reserved-class guard, the `data()`/`events()`/`app()` class facades) plus the rebranded Rust crate -# metadata. The pin below points at the current core `main`; LOCAL DEV may still build against the sibling -# checkout via the gitignored `.cargo/config.toml` `paths` override, which replaces this source outright. -edgecommons = { git = "https://github.com/edgecommons/edgecommons.git", rev = "36a70c48b65b35f77bfab70d3a73869debdfc407", default-features = false } +# This code REQUIRES the UNS core (top-level `identity`, `gg.uns()`, `gg.commands()`, the reserved-class +# guard, the `data()`/`events()`/`app()` class facades) and 0.5.0's scoped command registration +# (`register(verb, scope, handler)`). LOCAL DEV may build against the sibling checkout via the gitignored +# `.cargo/config.toml` `[patch]` override, which replaces this source outright — never commit a +# `Cargo.lock` regenerated while that override is active. +edgecommons = { git = "https://github.com/edgecommons/edgecommons.git", rev = "a14a3285573ef2bb6a531e1e1936c6dc40a85ef4", default-features = false } tokio = { version = "1", features = ["rt-multi-thread", "macros", "signal", "time", "sync", "fs"] } serde = { version = "1", features = ["derive"] } serde_json = "1" diff --git a/DESIGN.md b/DESIGN.md index 84a97b8..39a96d1 100644 --- a/DESIGN.md +++ b/DESIGN.md @@ -1045,10 +1045,33 @@ the UNS core this validity guarantee is enforced by the library that mints the t > The hand-rolled local control dispatcher below was replaced by verb registrations on the edgecommons > **command inbox** (`gg.commands()`, `ecv1/{device}/FileReplicator/cmd/#`), which now provides the > generic control framework core previously lacked. Concretely: **`get-config` is retired** (the library's -> built-in `get-configuration` verb answers it, redacted — plus `ping` / `reload-config`); the -> instance-in-topic scoping (`…/cmd/instances/{id}/…`) became an instance **body field** -> (`cmd/get-status` + `{instance?}`); verbs are `cmd/get-status`, `cmd/trigger`, `cmd/set-activation`. The -> table below is the ORIGINAL design; the `get-status` reply *shape* it describes is still accurate. +> built-in `get-configuration` verb answers it, redacted — plus `ping` / `reload-config`); verbs are +> `cmd/get-status`, `cmd/trigger`, `cmd/set-activation`. The table below is the ORIGINAL design; the +> `get-status` reply *shape* it describes is still accurate. +> +> **Instance addressing (core 0.5.0 scoped commands, core `DESIGN-scoped-commands.md`).** Each verb +> declares an addressing scope at registration, which the library enforces before dispatch: +> +> | Verb | Scope | No instance named | One instance named | +> |---|---|---|---| +> | `get-status` | `Both` | component-wide roster + summary | that instance's document | +> | `trigger` | `Both` | every active instance | that instance | +> | `set-activation` | `Instance` | `INSTANCE_REQUIRED` | that instance | +> +> `get-status`/`trigger` are `Both` because "no instance named" has always carried a meaningful +> *component-wide* answer here rather than being a default to resolve — D-SC-3's dual-semantics use. +> `set-activation` is `Instance`: it never had an "all" form, and per D-SC-4 the "no instance named" +> policy (`INSTANCE_REQUIRED`) and the unknown-id check (`UNKNOWN_INSTANCE`) stay component-side, +> because the library owns addressing and has no configuration knowledge. +> +> An instance is named **either** by the delivery topic +> (`ecv1/{device}/FileReplicator/{instance}/cmd/{verb}`, the library's per-instance inbox) **or** by the +> `instance` body field over the component topic. The topic wins (D-SC-4), and the library rejects a +> request whose topic and `body.instance` disagree with `BAD_ARGS`. `src/control.rs`'s `address()` folds +> the topic-addressed instance into the body selector, so the handlers keep reading exactly one +> selector. The original design's instance-in-topic scoping is therefore back — as the library's UNS +> instance inbox rather than a hand-rolled route — alongside the body field that replaced it in the UNS +> migration. Uses the edgecommons request/reply primitive (`request`/`reply`, `reply_to`). At design time core had **no generic control framework** — each component wired its own handlers — so this speced a small local **control diff --git a/docs/how-to-guides.md b/docs/how-to-guides.md index 373e70e..6bb9fe2 100644 --- a/docs/how-to-guides.md +++ b/docs/how-to-guides.md @@ -159,16 +159,21 @@ lean on resume: ## Activate / deactivate an instance from the control plane -Pause or resume one instance at runtime without a redeploy, via the `set-activation` command (it has **no** -"all" form — `instance` is required): +Pause or resume one instance at runtime without a redeploy, via the `set-activation` command. It is an +**instance-scoped** verb with **no** "all" form: name the instance in the topic, or in the `instance` +body field over the component topic. ``` -request topic ecv1//FileReplicator/cmd/set-activation +request topic ecv1//FileReplicator/plant-csv-to-s3/cmd/set-activation header.name set-activation -body { "instance": "plant-csv-to-s3", "active": false, "persist": true } +body { "active": false, "persist": true } reply body { "ok": true, "result": { "instance": "plant-csv-to-s3", "active": false, "persisted": true } } ``` +The equivalent over the component topic — `ecv1//FileReplicator/cmd/set-activation` with +`{ "instance": "plant-csv-to-s3", "active": false, "persist": true }` — is accepted unchanged. Naming +no instance at all answers `INSTANCE_REQUIRED`. + Send the request through an edgecommons client API or another protobuf-aware producer. The body and reply above are decoded JSON content inside the EdgeCommons command envelope; raw MQTT JSON is not accepted as a normal command message. diff --git a/docs/reference/data-types.md b/docs/reference/data-types.md index 00ca4e2..d75db36 100644 --- a/docs/reference/data-types.md +++ b/docs/reference/data-types.md @@ -14,11 +14,13 @@ unsigned 64-bit (a JavaScript consumer may lose precision above 2^53). Timestamp ## The `get-status` document -`get-status` (`…/cmd/get-status`) returns different shapes depending on the request body: +`get-status` returns different shapes depending on which instance, if any, the request names: -- **`{}`** (no `instance`) → a **component-wide** document: a roster of every instance plus a summary. -- **`{ "instance": "" }`** → that one instance's **per-instance** document (or its **disabled** document - if it was disabled at startup). An unknown id is the error `UNKNOWN_INSTANCE`. +- **No instance named** — `…/FileReplicator/cmd/get-status` with no `instance` body field → a + **component-wide** document: a roster of every instance plus a summary. +- **One instance named** — `…/FileReplicator/{id}/cmd/get-status`, or the component topic with + `{ "instance": "" }` → that one instance's **per-instance** document (or its **disabled** + document if it was disabled at startup). An unknown id is the error `UNKNOWN_INSTANCE`. The reply is always wrapped by the command contract: `{ "ok": true, "result": }`. diff --git a/docs/reference/messaging-interface.md b/docs/reference/messaging-interface.md index 2f2b5a0..c2afdb4 100644 --- a/docs/reference/messaging-interface.md +++ b/docs/reference/messaging-interface.md @@ -16,9 +16,11 @@ ecv1/{device}/{component}/[{instance}/]{class}[/{channel…}] `.` — `com.mbreissi.edgecommons.FileReplicator` → **`FileReplicator`**), not the `file-replicator` registry slug. - `{instance}` — OPTIONAL. Present (a `component.instances[].id`) for instance-scoped traffic — a - replication instance's own events (`file-ready`, `replication-*`, …) ride `gg.instance(id).events()`. - **Absent** for component-scope traffic: the built-in command verbs, `ComponentReady`, the scope-`"all"` - `trigger`/`get-status` events, and the library's own `state`/`cfg`/`metric` keepalives. + replication instance's own events (`file-ready`, `replication-*`, …) ride `gg.instance(id).events()`, + and a command addressed at one instance rides that instance's own command inbox. + **Absent** for component-scope traffic: a command addressed at the whole component, `ComponentReady`, + the scope-`"all"` `trigger`/`get-status` events, and the library's own `state`/`cfg`/`metric` + keepalives. - `{class}` ∈ `cmd` (inbound commands, request/reply) · `evt` (event stream) · the **reserved**, library-owned `state`/`cfg`/`metric`/`log` (this component never publishes to them directly). @@ -27,24 +29,37 @@ There is no configurable topic prefix and no legacy alias — the UNS grammar ab ## Commands (`cmd`, request/reply via `reply_to`) -Registered on the component-scope command inbox (`ecv1/{device}/FileReplicator/cmd/#`); an instance token -is optional and present only for explicit multi-instance addressing. Scoping an instance is a request-body -field, not a topic segment (mirroring how opcua-adapter/ -modbus-adapter address their multi-instance `sb/*` verbs and telemetry-processor's `route`/`pause`/ -`resume`). Publish commands with the edgecommons client APIs (`MessageBuilder` + `MessagingService` -request/reply, or an equivalent protobuf producer), not by sending JSON text to MQTT. Every decoded reply -body is `{"ok": true, "result": }` or `{"ok": false, "error": {"code", "message"}}` (the -edgecommons command-inbox contract — the request's `header.name` MUST equal the verb). +Two inboxes carry commands: the component one, `ecv1/{device}/FileReplicator/cmd/#`, and one per +replication instance, `ecv1/{device}/FileReplicator/{instance}/cmd/#`. Publish commands with the +edgecommons client APIs (`MessageBuilder` + `MessagingService` request/reply, or an equivalent protobuf +producer), not by sending JSON text to MQTT. Every decoded reply body is +`{"ok": true, "result": }` or `{"ok": false, "error": {"code", "message"}}` (the edgecommons +command-inbox contract — the request's `header.name` MUST equal the verb). -| Verb | Topic | Body | Result / error codes | +| Verb | Scope | Body | Result / error codes | |---|---|---|---| -| `get-status` | `…/cmd/get-status` | `{ "instance"?: string }` | omitted `instance` → component-wide roster+summary; present → that instance's document; `UNKNOWN_INSTANCE` | -| `trigger` | `…/cmd/trigger` | `{ "instance"?: string, "ignoreWindow"?: bool }` | accepted + counts; `UNKNOWN_INSTANCE` | -| `set-activation` | `…/cmd/set-activation` | `{ "instance": string, "active"?: bool, "persist"?: bool, "reset"?: bool }` | new state; `INSTANCE_REQUIRED` (no "all" form), `UNKNOWN_INSTANCE`, `INVALID_REQUEST` (neither `active` nor `reset`), `ACTIVATION_FAILED` | +| `get-status` | both | `{ "instance"?: string }` | no instance named → component-wide roster+summary; one named → that instance's document; `UNKNOWN_INSTANCE` | +| `trigger` | both | `{ "instance"?: string, "ignoreWindow"?: bool }` | accepted + counts; `UNKNOWN_INSTANCE` | +| `set-activation` | instance | `{ "instance": string, "active"?: bool, "persist"?: bool, "reset"?: bool }` | new state; `INSTANCE_REQUIRED` (no "all" form), `UNKNOWN_INSTANCE`, `INVALID_REQUEST` (neither `active` nor `reset`), `ACTIVATION_FAILED` | -The library's own built-in verbs are also available: `ping` (liveness), `reload-config` (re-fetch + -re-apply), and **`get-configuration`** — returns the **redacted** effective config (`{"config": -}`, with secrets replaced rather than left as unresolved `$secret` refs). +### Addressing an instance + +Each verb declares an addressing **scope**, which the library enforces before the verb runs: + +- **`both`** (`get-status`, `trigger`) — either inbox is meaningful. Address the component + (`…/FileReplicator/cmd/{verb}`) for the fleet-wide answer, or one instance + (`…/FileReplicator/{instance}/cmd/{verb}`) for that instance alone. +- **`instance`** (`set-activation`) — this verb has no "all" form. It answers `INSTANCE_REQUIRED` + when neither the topic nor the body names an instance. + +The `instance` body field remains the way to target one instance over the component inbox. When the +topic names an instance it wins; a request that names one instance in the topic and a different one +in `body.instance` is rejected with `BAD_ARGS` before the verb runs. + +The library's own built-in verbs are also available on both inboxes: `ping` (liveness), +`reload-config` (re-fetch + re-apply), and **`get-configuration`** — returns the **redacted** +effective config (`{"config": }`, with secrets replaced rather than left as unresolved +`$secret` refs). ## Events (`evt`, `evt/{severity}/{type}`) diff --git a/src/control.rs b/src/control.rs index 3fe2c52..8f0276e 100644 --- a/src/control.rs +++ b/src/control.rs @@ -17,13 +17,27 @@ //! message}}` body shape. Handlers return `Result, CommandError>` instead of building //! their own reply. //! -//! **Scoping — body field, not topic segment.** The old scheme split `cmd/status` (all instances) -//! from `cmd/instances/{id}/status` (one instance) by TOPIC. The command inbox is one subscription -//! per component (component scope, no instance token), so scoping now rides an optional `instance` field in -//! the request body — the same convention opcua-adapter/modbus-adapter use for their multi-instance -//! `sb/*` verbs and telemetry-processor uses for `pause`/`resume`'s `route` field. `get-status`/ -//! `trigger` omit `instance` for "all"; `set-activation` always requires it (it never had an "all" -//! form). +//! **Scoping — declared per verb, topic first, body second.** The old scheme split `cmd/status` +//! (all instances) from `cmd/instances/{id}/status` (one instance) by a hand-rolled TOPIC layout. +//! Under the UNS core the library owns both inboxes — the component's +//! `ecv1/{device}/file-replicator/cmd/#` and each instance's +//! `ecv1/{device}/file-replicator/{instance}/cmd/#` — and core 0.5.0 makes each verb declare its +//! addressing [`CommandScope`] (`DESIGN-scoped-commands.md` D-SC-2), enforced before dispatch: +//! +//! | Verb | Scope | Component-addressed | Instance-addressed | +//! |---|---|---|---| +//! | `get-status` | [`CommandScope::Both`] | every instance + the summary | that instance's status | +//! | `trigger` | [`CommandScope::Both`] | every active instance | that instance | +//! | `set-activation` | [`CommandScope::Instance`] | the `instance` body field, else `INSTANCE_REQUIRED` | that instance | +//! +//! `get-status`/`trigger` are `Both` because "no instance named" has always been a meaningful +//! *component-wide* answer here, not a default to resolve — D-SC-3's dual-semantics use. +//! `set-activation` is `Instance`: it never had an "all" form, and D-SC-4 leaves the +//! "no instance named" policy (this component's `INSTANCE_REQUIRED`) and the unknown-id check +//! (`UNKNOWN_INSTANCE`) on the component side, since the library does not know the configuration. +//! +//! The optional `instance` body field is unchanged for component-addressed callers; when the topic +//! names an instance it wins ([`address`]). //! //! ## get-config — retired, not migrated //! The custom `get-config` verb (and its optional `legacyConfigTopic` alias) is dropped outright: the @@ -62,7 +76,9 @@ use std::sync::Arc; use async_trait::async_trait; -use edgecommons::commands::{command_handler, CommandError, CommandHandler, CommandInbox}; +use edgecommons::commands::{ + command_handler, CommandError, CommandHandler, CommandInbox, CommandScope, +}; use edgecommons::messaging::Message; use edgecommons::prelude::Config; use serde_json::{json, Value}; @@ -199,9 +215,10 @@ impl ControlPlane { try_register( commands, "get-status", - command_handler(move |req| { + CommandScope::Both, + command_handler(move |req, addressed| { let me = me.clone(); - async move { me.on_get_status(req).await } + async move { me.on_get_status(address(req, addressed)).await } }), ); } @@ -210,9 +227,10 @@ impl ControlPlane { try_register( commands, "trigger", - command_handler(move |req| { + CommandScope::Both, + command_handler(move |req, addressed| { let me = me.clone(); - async move { me.on_trigger(req).await } + async move { me.on_trigger(address(req, addressed)).await } }), ); } @@ -221,9 +239,10 @@ impl ControlPlane { try_register( commands, "set-activation", - command_handler(move |req| { + CommandScope::Instance, + command_handler(move |req, addressed| { let me = me.clone(); - async move { me.on_set_activation(req).await } + async move { me.on_set_activation(address(req, addressed)).await } }), ); } @@ -417,16 +436,45 @@ impl ControlPlane { } } -/// Registers a verb, logging (not failing) if the inbox rejects it — mirrors -/// telemetry-processor's `try_register`. -fn try_register(commands: &CommandInbox, verb: &str, handler: Arc) { - if let Err(e) = commands.register(verb, handler) { +/// Registers a verb at its declared [`CommandScope`], logging (not failing) if the inbox rejects it +/// — mirrors telemetry-processor's `try_register`. +fn try_register( + commands: &CommandInbox, + verb: &str, + scope: CommandScope, + handler: Arc, +) { + if let Err(e) = commands.register(verb, scope, handler) { tracing::warn!(verb, error = %e, "failed to register command verb"); } } -/// The request body's `"instance"` selector (absent → "all", except `set-activation`, which requires -/// it — see [`ControlPlane::on_set_activation`]). +/// Fold the delivery's **topic-addressed** instance (core 0.5.0's `addressed_instance`, D-SC-4) into +/// the request's `"instance"` body selector, so every handler below reads exactly one selector — +/// [`instance_of`] — no matter which of the two inboxes the request arrived on. +/// +/// The topic is authoritative: an instance-addressed delivery overwrites whatever the body said. +/// That can only ever *fill in* a missing value in practice, because the library already refuses a +/// request whose topic token and `body.instance` disagree (`BAD_ARGS`) before dispatch. A +/// component-addressed delivery leaves the request untouched, so the established +/// "`instance` in the body, absent = all" convention keeps working unchanged. +fn address(mut req: Message, addressed_instance: Option) -> Message { + let Some(id) = addressed_instance else { + return req; + }; + match req.body.as_object_mut() { + Some(obj) => { + obj.insert("instance".to_string(), Value::String(id)); + } + // A non-object body carries no selector to preserve; the addressing must not be lost. + None => req.body = json!({ "instance": id }), + } + req +} + +/// The request's `"instance"` selector — the topic-addressed instance when the delivery named one +/// (folded in by [`address`]), else the body field (absent → "all", except `set-activation`, which +/// requires it — see [`ControlPlane::on_set_activation`]). fn instance_of(req: &Message) -> Option<&str> { req.body.get("instance").and_then(Value::as_str) } @@ -900,6 +948,32 @@ mod tests { assert_eq!(v["failed"]["items"][1]["quarantinedAt"], json!("2026-07-01T09:15:22Z")); } + // ---- addressing (core 0.5.0 scoped commands, D-SC-4) ---------------------------------------- + + #[test] + fn the_topic_addressed_instance_becomes_the_selector_and_wins_over_the_body() { + // Instance-addressed delivery, no body selector: the addressing must not be lost. + let r = address(req("get-status", json!({ "ignoreWindow": true })), Some("i2".into())); + assert_eq!(instance_of(&r), Some("i2")); + assert_eq!(r.body["ignoreWindow"], json!(true), "other body fields survive"); + + // The topic wins (the library has already refused an outright conflicting pair). + let r = address(req("trigger", json!({ "instance": "i1" })), Some("i1".into())); + assert_eq!(instance_of(&r), Some("i1")); + + // Component-addressed delivery: the body selector is untouched, absent still means "all". + let r = address(req("get-status", json!({ "instance": "i1" })), None); + assert_eq!(instance_of(&r), Some("i1")); + let r = address(req("get-status", json!({})), None); + assert_eq!(instance_of(&r), None); + } + + #[test] + fn a_non_object_body_still_carries_the_addressed_instance() { + let r = address(req("trigger", json!("not-an-object")), Some("i3".into())); + assert_eq!(instance_of(&r), Some("i3")); + } + // ---- get-status ----------------------------------------------------------------------------- fn seed_status(store: &Arc) {