diff --git a/CHANGELOG.md b/CHANGELOG.md index 7ac171074..fff3a0a37 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -117,6 +117,19 @@ layout) per [`STABILITY_POLICY.md`](STABILITY_POLICY.md). default — the block is the fallback). The outputs are dropped by the on-disk index (no size increase, no reindex) and re-derived at serve time. SDK: `TweakEntry::taproot_outputs` and `SubscribeOptions::tweak_outputs`. +- Alerting: node-health events on the streaming API. A new `status` category + (bit 16 — explicit-request only, so a `categories=0` subscriber is unaffected) + carries `StatusEvent` bodies describing the node itself (`ibd_complete`, + `tip_stall`, `disk_low`, `mempool_congested`, `peer_floor`, `deep_reorg`), + level-triggered with paired `raised`/`cleared` states and an additive + `details` string map. Served on gRPC and WS/SSE; **not** on the ZMQ + `nodeevent` topic, which has no per-subscriber category mask (use `alertfile` + webhooks or `-alertnotify` for health over ZMQ). Requires **`rpc:read` as well + as `stream:subscribe`** where `-authfile` is in use — the bodies carry host + telemetry (disk, peers, mempool occupancy, tip height) that the same + capability gates on the RPC surface. Not replayable (no cursor — + detectors re-raise standing conditions after a restart). Wire schema only in + this change; the detectors that emit them land next. ## Releases diff --git a/docs/api/streaming.md b/docs/api/streaming.md index c3f4e651b..c726474d7 100644 --- a/docs/api/streaming.md +++ b/docs/api/streaming.md @@ -126,6 +126,7 @@ message NodeEvent { BlockTweaks block_tweaks = 28; // BIP 352 per-block tweak data (§7.7) SilentPaymentMatched silent_payment_matched = 29; // BIP 352 scan-key watch match (§7.7) MempoolTweak mempool_tweak = 30; // BIP 352 mempool-time tweak, Tier 1.5 (§7.7) + StatusEvent status = 31; // node-health condition (§7.8) } } ``` @@ -913,6 +914,75 @@ deliver `Event::SilentPaymentMatched`; `examples/sp_wallet.rs` re-derives each match's spending key offline from `tweak` + `k`. The scan-key helpers need the crate's default `bitcoin` feature (to derive the removal identity `b_scan·G`). +### 7.8 Node-health status events + +Every other body on this surface describes the *chain* or the *mempool*. A +`StatusEvent` describes the **node itself** — a stalled tip, a filling disk, a +peer-starved daemon — so an operator can drive alerting off the same stream they +already consume, instead of polling `getblockchaininfo` on a timer. + +The `status` category is **bit 16**. Like `tweaks`, it is **not** part of the +`categories = 0` ("all") default: a subscriber written against an older node +never begins receiving a body it has no parser for. Unlike `tweaks` it has no +index prerequisite. + +**Authorization: `status` requires `rpc:read` in addition to +`stream:subscribe`.** Every other category describes public chain data, which is +what `stream:subscribe` exists to hand out. `StatusEvent` describes the host — +free bytes on the node's volume, connected peers split inbound/outbound, the +mempool byte cap with its occupancy and `mempoolminfee`, tip height, IBD state, +reorg depth and fork heights. That is the content of `getblockchaininfo`, +`getpeerinfo`, `getmempoolinfo` and `getwarnings`, and `satd-auth` separates the +two capabilities precisely so an operator can issue a streaming token to a +wallet backend or tenant *without* granting node read RPC. A token holding only +`stream:subscribe` is refused with `PERMISSION_DENIED` (gRPC) or `403` (WS/SSE) +when it requests bit 16, and a mid-stream `SetCategories` / `SetWatchSet` from +such a token has the bit stripped. This applies only where `-authfile` is +configured: with no token store the node behaves exactly as before, and the +operator principal (cookie / `rpcuser` / `rpcauth`) holds every capability. + +It is served on the carriers that have a per-subscriber category mask to opt in +with: gRPC `Subscribe`, gRPC `Watch`, WS, and SSE. It is **not** emitted on the +ZMQ `nodeevent` topic. ZMQ has no per-subscriber mask — only an all-or-nothing +topic toggle — so honoring "explicit-only" there is impossible: emitting status +would hand a new body type to every already-deployed `nodeevent` consumer on +upgrade, and their only escape would be `eventszmqnodeevent=0`, which silences +every other envelope too. Operators consuming health over ZMQ should use the +`alertfile` webhook dispatcher or `-alertnotify` instead. + +```proto +message StatusEvent { + StatusKind kind = 1; // ibd_complete | tip_stall | disk_low | + // mempool_congested | peer_floor | deep_reorg + StatusState state = 2; // raised | cleared | edge + StatusSeverity severity = 3; // info | warning | critical + string message = 4; // human-readable; log it, do not parse it + map details = 5; // kind-specific fields (see below) +} +``` + +**Level-triggered.** A standing condition emits exactly one `raised` on entry and +exactly one `cleared` on recovery, with hysteresis between the two thresholds so +a value hovering at the line does not flap. Observations with no recovered state +(IBD finishing, a deep reorg landing) are `edge` and never produce a `cleared`. + +**Not replayable.** A status event carries no `cursor` and is excluded from the +replay ring, so a `from_cursor` resume never yields one. Durability comes from +re-evaluation instead: the detectors re-examine every condition at startup and +re-raise the ones still standing, which is what makes health alerting +at-least-once across a restart. A condition that both raised and fully cleared +while a consumer was away is stale by definition and is not reconstructed. + +**Additive by construction.** `details` is a string map and `StatusKind` is an +open enum: new kinds and new detail keys ship without a schema bump (§4). A +consumer must tolerate an unrecognized `kind` — `message` and `severity` remain +meaningful — and must not require any particular `details` key to be present. + +The same detections also drive the node's warnings registry, so `getwarnings`, +the Core-compatible `-alertnotify` hook, and this stream can never disagree about +node state. Operator-facing configuration (detector thresholds, the `alertfile` +webhook dispatcher) is documented in the Operator Manual, not here. + ## 8. Reorg events Reorgs are **first-class**, emitted in-process as consensus ground truth — a diff --git a/docs/release-notes/0.5.0-pre.md b/docs/release-notes/0.5.0-pre.md index a5fd91681..437993716 100644 --- a/docs/release-notes/0.5.0-pre.md +++ b/docs/release-notes/0.5.0-pre.md @@ -204,6 +204,57 @@ and matching against it arrive in the following changes. helpers use the crate's default `bitcoin` feature to derive a target's removal identity. +## Alerting & node health + +Every body on the streaming API so far described the *chain* or the *mempool*. +None of them described the **node**, so an operator who wanted to know that their +daemon had stalled, filled its disk, or lost its peers had to poll +`getblockchaininfo` on a timer and diff the results themselves. + +- **Node-health events are now first-class on the firehose.** A new `status` + category carries `StatusEvent` bodies for six conditions the daemon detects + about itself: `ibd_complete`, `tip_stall`, `disk_low`, `mempool_congested`, + `peer_floor`, and `deep_reorg`. Each event carries a machine-readable `kind`, + a `severity`, a human-readable `message`, and a `details` map of structured + fields as strings — mostly decimal numbers (thresholds, counts, byte totals), + but not exclusively: a `cleared` event can carry a `reason` token such as + `detector_disabled`. Do not parse the map uniformly as integers. +- **Level-triggered, not edge-spammed.** A standing condition emits exactly one + `raised` when it is entered and exactly one `cleared` when it recovers, with + hysteresis between the two thresholds so a value hovering at the line does not + flap a pager. Conditions with no recovered state — IBD finishing, a deep reorg + landing — are `edge` and never produce a `cleared`. +- **Opt-in by construction.** The category is **bit 16** and, like the + silent-payment `tweaks` category, is deliberately excluded from the + `categories = 0` ("all") default: a subscriber written against an older node + never begins receiving a body it has no parser for after an upgrade. Unlike + `tweaks` there is no index prerequisite. Served on the three mask-bearing + transports — gRPC `Subscribe`/`Watch`, WS, and SSE. It is deliberately **not** + emitted on the ZMQ `nodeevent` topic: ZMQ has no per-subscriber category mask, + so putting it there would push a new body shape at every existing `nodeevent` + consumer. For health over ZMQ, use `alertfile` webhooks or `-alertnotify`. +- **`status` needs `rpc:read` on top of `stream:subscribe`.** Every other + category describes the chain, which is public; this one describes your host — + free disk, peer topology, mempool occupancy and `mempoolminfee`, tip height, + IBD state. That is the content of `getblockchaininfo`, `getpeerinfo`, + `getmempoolinfo` and `getwarnings`, and the whole point of a separate + `stream:subscribe` capability is that you can give a wallet backend or a + tenant a streaming token *without* node read RPC. So a token holding only + `stream:subscribe` is refused (`PERMISSION_DENIED` / `403`) when it asks for + bit 16, and a mid-stream category update from one has the bit stripped rather + than honored. This narrows nothing for a node that has not enabled + `-authfile`: with no token store the behavior is unchanged, and the operator + principal (cookie, `rpcuser`, `rpcauth`) holds every capability already. +- **Durability is by re-evaluation, not replay.** A status event carries no + resume cursor and is excluded from the replay ring, so a `from_cursor` resume + never yields one. Instead the detectors re-examine every condition at startup + and re-raise the ones still standing — which is what makes health alerting + at-least-once across a restart. A condition that both raised and fully cleared + while a consumer was away is stale by definition and is not reconstructed. +- Additive on the wire: the schema version does not bump, `StatusKind` is an open + enum, and `details` is a string map, so future conditions and fields ship + without breaking a deployed consumer. + ## Storage / Core compatibility ### Core v28+ XOR-obfuscated block files (`blocksxor`) diff --git a/events/src/catauth.rs b/events/src/catauth.rs new file mode 100644 index 000000000..7d29ac2ff --- /dev/null +++ b/events/src/catauth.rs @@ -0,0 +1,120 @@ +//! Per-category authorization for the streaming carriers. +//! +//! Most categories carry data that is *about the chain* — blocks, transactions, +//! the mempool — which is public by nature and which `stream:subscribe` exists +//! to hand out. The `status` category is different: it carries data about the +//! **host**, and every other surface keeps that behind `rpc:read`. +//! +//! A `StatusEvent` body reports free bytes on the node's volume, connected peer +//! counts split inbound/outbound, the mempool byte cap with its current +//! occupancy and `mempoolminfee`, tip height, IBD state, and reorg depth with +//! fork heights. Those are the answers to `getblockchaininfo`, `getpeerinfo`, +//! `getmempoolinfo` and `getwarnings`. +//! +//! `satd-auth` separates `rpc:read` from `stream:subscribe` precisely so an +//! operator can issue a streaming token to a wallet backend, an indexer, or a +//! tenant *without* also handing over the node's read RPC. Serving `status` on +//! `stream:subscribe` alone would route around that split rather than extend +//! it: the token would deliver, over the stream, the host telemetry it was +//! issued to withhold. So the bit needs both capabilities. +//! +//! This is not a new capability in the `auth.toml` vocabulary. Adding one would +//! be a Tier 1 surface change, and it is not needed — the data already has an +//! owner in the existing vocabulary, and `status` is new in this release, so no +//! deployed token loses access to something it has today. + +/// Whether `principal` may receive the `status` category. +/// +/// `None` means the token store is not configured at all, which is the +/// Core-compatible default (loopback trust) and grants everything — this gate +/// only ever narrows a `-authfile` deployment. The operator principal +/// (cookie / userpass / rpcauth) carries `CapabilitySet::ALL` and so passes. +pub(crate) fn may_receive_status(principal: Option<&satd_auth::Principal>) -> bool { + principal.is_none_or(|p| p.has(satd_auth::Capability::RpcRead)) +} + +/// Strip `status` from `mask` when `principal` may not receive it. +/// +/// Used on the mid-stream control paths (`SetCategories`, `SetWatchSet`), which +/// have no per-message error channel of their own. The handshake paths reject +/// outright instead — see `may_receive_status`'s callers in `grpc`/`ws` — so a +/// client that asks for `status` up front is told plainly rather than left +/// wondering why a category it requested never arrives. +#[cfg_attr(not(feature = "grpc"), allow(dead_code))] +pub(crate) fn strip_unauthorized(principal: Option<&satd_auth::Principal>, mask: u32) -> u32 { + if may_receive_status(principal) { + return mask; + } + if mask & node::events::CATEGORY_STATUS != 0 { + tracing::debug!( + target: "events::auth", + "dropping the status category from a mid-stream category update: \ + the token lacks rpc:read", + ); + } + mask & !node::events::CATEGORY_STATUS +} + +#[cfg(test)] +mod tests { + use super::*; + use satd_auth::{Capability, CapabilitySet, Principal}; + + /// `has()` reads only `caps`, so the principal kind is immaterial here and + /// `loopback` is the cheapest constructor that takes an explicit set. + fn principal_with(caps: CapabilitySet) -> Principal { + Principal::loopback(caps) + } + + /// The Core-compatible default: no `-authfile`, no token store, nothing to + /// narrow. This gate must never change behavior for a node that has not + /// opted into bearer auth. + #[test] + fn auth_disabled_receives_status() { + assert!(may_receive_status(None)); + assert_eq!( + strip_unauthorized(None, node::events::CATEGORY_STATUS), + node::events::CATEGORY_STATUS, + ); + } + + /// The finding: a token issued `stream:subscribe` *in order to* withhold + /// `getblockchaininfo` / `getpeerinfo` / `getwarnings` must not get the same + /// telemetry back over the stream. + #[test] + fn stream_subscribe_alone_does_not_carry_status() { + let p = principal_with(CapabilitySet::EMPTY.with(Capability::StreamSubscribe)); + assert!(!may_receive_status(Some(&p))); + + let asked = node::events::CATEGORY_CHAIN | node::events::CATEGORY_STATUS; + let got = strip_unauthorized(Some(&p), asked); + assert_eq!(got & node::events::CATEGORY_STATUS, 0, "status is stripped"); + assert_eq!( + got & node::events::CATEGORY_CHAIN, + node::events::CATEGORY_CHAIN, + "and nothing else is", + ); + } + + #[test] + fn rpc_read_carries_status() { + let p = principal_with( + CapabilitySet::EMPTY + .with(Capability::StreamSubscribe) + .with(Capability::RpcRead), + ); + assert!(may_receive_status(Some(&p))); + assert_eq!( + strip_unauthorized(Some(&p), node::events::CATEGORY_STATUS), + node::events::CATEGORY_STATUS, + ); + } + + /// The operator principal holds everything, so an operator-authenticated + /// stream is unaffected. + #[test] + fn the_operator_principal_carries_status() { + let p = principal_with(CapabilitySet::ALL); + assert!(may_receive_status(Some(&p))); + } +} diff --git a/events/src/grpc.rs b/events/src/grpc.rs index 51d5a1034..9c125698f 100644 --- a/events/src/grpc.rs +++ b/events/src/grpc.rs @@ -825,6 +825,10 @@ impl NodeEventStream for NodeEventStreamSvc { &self, request: Request, ) -> Result, Status> { + // Read before `into_inner()`, which drops the extensions the auth + // interceptor stashed the principal in. `None` means the token store is + // not configured at all (loopback trust), the Core-compatible default. + let principal = request.extensions().get::().cloned(); let req = request.into_inner(); // Enforce the global concurrent-subscription cap before attaching a @@ -866,6 +870,26 @@ impl NodeEventStream for NodeEventStreamSvc { } else { req.categories }; + + // `status` (bit 16) additionally requires `rpc:read`: it reports host + // telemetry — free disk, peer topology, mempool occupancy and + // `mempoolminfee`, tip height, IBD state — that every other surface + // keeps behind that capability. Refused here rather than stripped, + // because the client asked for it explicitly (bit 16 is never in the + // `0` default) and a silently absent category reads as "the node is + // healthy". See `crate::catauth`. + if category_mask & node::events::CATEGORY_STATUS != 0 + && !crate::catauth::may_receive_status(principal.as_ref()) + { + debug!( + target: "events::grpc", + code = "permission_denied", + "rejecting Subscribe: the status category requires rpc:read", + ); + return Err(Status::permission_denied( + "the status category requires the rpc:read capability", + )); + } let since_seq = req.since_seq.unwrap_or(0); // Silent-payment `tweaks` category (bit 8) is explicit-request only and @@ -1451,8 +1475,15 @@ impl NodeEventStream for NodeEventStreamSvc { // control update cannot forward shared-broadcast // `BlockTweaks` past the Subscribe path's // index/completeness/dust-limit validation. + // `status` is stripped for a token without + // `rpc:read` for the same reason it is refused + // on Subscribe — a control message must not be + // a way around the handshake gate. category_mask.store( - mask & !node::events::CATEGORY_TWEAKS, + crate::catauth::strip_unauthorized( + principal.as_ref(), + mask & !node::events::CATEGORY_TWEAKS, + ), Ordering::Relaxed, ); } @@ -2150,8 +2181,16 @@ fn apply_control( c.categories }; // Strip the tweaks bit: Watch never serves the firehose (see the - // SetWatchSet path above). - category_mask.store(mask & !node::events::CATEGORY_TWEAKS, Ordering::Relaxed); + // SetWatchSet path above). `status` goes the same way for a token + // without `rpc:read`, so a mid-stream update is not a route around + // the handshake gate. + category_mask.store( + crate::catauth::strip_unauthorized( + principal, + mask & !node::events::CATEGORY_TWEAKS, + ), + Ordering::Relaxed, + ); } Some(Msg::SetWatchOptions(o)) => { // Per-stream raw-tx inlining. Store the encoder-side flag AND toggle @@ -3019,6 +3058,7 @@ fn body_to_proto(body: &NodeEventBody) -> pb::node_event::Body { // dropped before conversion); this maps the survivor to its full proto // form. `tweaks_only` never applies — the txid is always present. NodeEventBody::MempoolTweak(entry) => Body::MempoolTweak(mempool_tweak_to_proto(entry)), + NodeEventBody::Status(ev) => Body::Status(status_event_to_proto(ev)), NodeEventBody::Heartbeat { uptime_ns } => Body::Heartbeat(pb::Heartbeat { uptime_ns: *uptime_ns, }), @@ -3035,6 +3075,39 @@ fn body_to_proto(body: &NodeEventBody) -> pb::node_event::Body { } } +fn status_event_to_proto(ev: &node::events::StatusEvent) -> pb::StatusEvent { + use node::events::{StatusKind as K, StatusSeverity as Sev, StatusState as St}; + let kind = match ev.kind { + K::IbdComplete => pb::StatusKind::IbdComplete, + K::TipStall => pb::StatusKind::TipStall, + K::DiskLow => pb::StatusKind::DiskLow, + K::MempoolCongested => pb::StatusKind::MempoolCongested, + K::PeerFloor => pb::StatusKind::PeerFloor, + K::DeepReorg => pb::StatusKind::DeepReorg, + }; + let state = match ev.state { + St::Raised => pb::StatusState::Raised, + St::Cleared => pb::StatusState::Cleared, + St::Edge => pb::StatusState::Edge, + }; + let severity = match ev.severity { + Sev::Info => pb::StatusSeverity::Info, + Sev::Warning => pb::StatusSeverity::Warning, + Sev::Critical => pb::StatusSeverity::Critical, + }; + pb::StatusEvent { + kind: kind as i32, + state: state as i32, + severity: severity as i32, + message: ev.message.clone(), + // `details` is a `BTreeMap` on the Rust side (deterministic order for + // the HMAC-signed webhook body); proto maps are unordered, so the + // ordering guarantee simply does not carry here — no consumer of the + // proto surface depends on it. + details: ev.details.iter().map(|(k, v)| (k.clone(), v.clone())).collect(), + } +} + fn set_cursor_result_to_proto(outcome: &node::events::SetCursorOutcome) -> pb::SetCursorResult { use node::events::{CursorRejectReason as R, SetCursorOutcome as O}; use pb::set_cursor_result::Outcome; @@ -3220,6 +3293,62 @@ mod tests { ); } + #[test] + fn status_event_maps_to_proto_enums_and_details() { + use node::events::{StatusEvent, StatusKind, StatusSeverity, StatusState}; + + let ev = StatusEvent::raised(StatusKind::TipStall, "no block for 3612s") + .with_detail("seconds_since_block", 3612u64) + .with_detail("threshold_seconds", 3600u64); + let pb = status_event_to_proto(&ev); + + assert_eq!(pb.kind, pb::StatusKind::TipStall as i32); + assert_eq!(pb.state, pb::StatusState::Raised as i32); + assert_eq!(pb.severity, pb::StatusSeverity::Critical as i32); + assert_eq!(pb.message, "no block for 3612s"); + assert_eq!(pb.details.get("seconds_since_block").map(String::as_str), Some("3612")); + assert_eq!(pb.details.get("threshold_seconds").map(String::as_str), Some("3600")); + + // Every kind and state has a distinct, non-UNSPECIFIED mapping — a + // zero-valued enum on the wire would be indistinguishable from a field + // the server never set. + let mut seen = std::collections::HashSet::new(); + for kind in StatusKind::ALL { + let ev = if kind.is_edge() { + StatusEvent::edge(kind, "x") + } else { + StatusEvent::raised(kind, "x") + }; + let mapped = status_event_to_proto(&ev).kind; + assert_ne!(mapped, pb::StatusKind::Unspecified as i32, "{kind:?}"); + assert!(seen.insert(mapped), "duplicate proto mapping for {kind:?}"); + } + for (state, want) in [ + (StatusState::Raised, pb::StatusState::Raised), + (StatusState::Cleared, pb::StatusState::Cleared), + (StatusState::Edge, pb::StatusState::Edge), + ] { + let ev = StatusEvent { + kind: StatusKind::DiskLow, + state, + severity: StatusSeverity::Info, + message: String::new(), + details: Default::default(), + }; + assert_eq!(status_event_to_proto(&ev).state, want as i32); + } + } + + #[test] + fn status_envelope_rides_the_status_body_field() { + use node::events::{StatusEvent, StatusKind}; + let body = NodeEventBody::Status(StatusEvent::edge(StatusKind::IbdComplete, "synced")); + assert!(matches!( + body_to_proto(&body), + pb::node_event::Body::Status(_) + )); + } + #[test] fn block_tweaks_to_proto_dust_limit_sets_filtered() { // A floor above the smaller entry drops it and flags `filtered`. @@ -4284,6 +4413,7 @@ mod tests { pb::node_event::Body::BlockTweaks(_) => "block_tweaks", pb::node_event::Body::SilentPaymentMatched(_) => "silent_payment_matched", pb::node_event::Body::MempoolTweak(_) => "mempool_tweak", + pb::node_event::Body::Status(_) => "status", }) .collect(); assert!( diff --git a/events/src/lib.rs b/events/src/lib.rs index 8ab38f544..b56704ca8 100644 --- a/events/src/lib.rs +++ b/events/src/lib.rs @@ -43,3 +43,7 @@ pub mod descriptor; // per-remove release) — shared by the gRPC and WS watch surfaces. #[cfg(any(feature = "grpc", feature = "ws"))] pub(crate) mod watchset; + +// Category-level authorization, shared by the gRPC and WS/SSE carriers. +#[cfg(any(feature = "grpc", feature = "ws"))] +pub(crate) mod catauth; diff --git a/events/src/ws.rs b/events/src/ws.rs index df9d6ad98..767f27221 100644 --- a/events/src/ws.rs +++ b/events/src/ws.rs @@ -283,6 +283,32 @@ impl FirehoseQuery { } } +/// Refuse a handshake that asks for `status` without `rpc:read`. +/// +/// Refused rather than quietly stripped: bit 16 is never in the `0` default, so +/// asking for it is always deliberate, and a category that silently never +/// arrives reads to a health dashboard as "nothing is wrong". See +/// [`crate::catauth`] for why `rpc:read` is the capability that owns this data. +#[allow(clippy::result_large_err)] +fn require_status_capability( + mask: u32, + principal: Option<&satd_auth::Principal>, +) -> Result<(), Response> { + if mask & node::events::CATEGORY_STATUS != 0 && !crate::catauth::may_receive_status(principal) { + debug!( + target: "events::ws", + status = 403, + "rejecting streamws connection: the status category requires rpc:read", + ); + return Err(( + StatusCode::FORBIDDEN, + "the status category requires the rpc:read capability", + ) + .into_response()); + } + Ok(()) +} + fn mask_from(categories: Option) -> u32 { // The WS/SSE firehose does not serve the tweaks category this release, so // the tweaks bit is stripped from EVERY mask, default or explicit: @@ -295,6 +321,11 @@ fn mask_from(categories: Option) -> u32 { // any gRPC client attaches, so without this strip the WS live filter // would forward them (its replay already returns empty). Tier 1 tweak // consumption is gRPC-only. + // + // The other explicit-only category, `status` (bit 16), IS served here: it is + // low-volume JSON with no index prerequisite, so a WS/SSE client that asks + // for it by bit gets it (and, being explicit-only, a `0`-subscriber still + // does not). match categories { None | Some(0) => node::events::ALL_CATEGORIES_DEFAULT, Some(c) => c & !node::events::CATEGORY_TWEAKS, @@ -393,9 +424,10 @@ async fn sse_firehose( headers: HeaderMap, Query(q): Query, ) -> Response { - if let Err(resp) = authorize(&state, &headers) { - return resp; - } + let principal = match authorize(&state, &headers) { + Ok(p) => p, + Err(resp) => return resp, + }; // Admission control: hold a connection permit for the stream's lifetime. let Ok(permit) = state.conn_sem.clone().try_acquire_owned() else { return ( @@ -405,6 +437,9 @@ async fn sse_firehose( .into_response(); }; let mask = mask_from(q.categories); + if let Err(resp) = require_status_capability(mask, principal.as_ref()) { + return resp; + } // Subscribe to the live broadcast FIRST so nothing is missed between the // replay snapshot and the live tail (the snapshot→live handoff ordering). let rx = state.publisher.subscribe(); @@ -496,6 +531,9 @@ async fn ws_upgrade( .into_response(); }; let initial_mask = mask_from(q.categories); + if let Err(resp) = require_status_capability(initial_mask, principal.as_ref()) { + return resp; + } let cursor = q.cursor(); // Bound a single inbound frame/message — control frames are tiny. let max_bytes = state.max_message_bytes; @@ -1029,7 +1067,16 @@ fn apply_ws_control( // category); strip the bit so a control update cannot opt this // stream into shared-broadcast `BlockTweaks` and bypass the // Subscribe path's index/completeness/dust-limit validation. - category_mask.store(mask & !node::events::CATEGORY_TWEAKS, Ordering::Relaxed); + // `status` goes the same way for a token without `rpc:read`, + // so a control message is not a route around the handshake + // gate that already refused it. + category_mask.store( + crate::catauth::strip_unauthorized( + principal, + mask & !node::events::CATEGORY_TWEAKS, + ), + Ordering::Relaxed, + ); } outcome } @@ -1068,8 +1115,16 @@ fn apply_ws_control( categories }; // Strip the tweaks bit: WS/SSE never serves the firehose (see the - // SetWatchSet path above). - category_mask.store(mask & !node::events::CATEGORY_TWEAKS, Ordering::Relaxed); + // SetWatchSet path above). `status` goes the same way for a token + // without `rpc:read` — the handshake already refused it, and a + // control message must not be a second door. + category_mask.store( + crate::catauth::strip_unauthorized( + principal, + mask & !node::events::CATEGORY_TWEAKS, + ), + Ordering::Relaxed, + ); } WsControl::SetWatchOptions { include_raw_tx: want } => { // Store the encoder-side flag AND toggle the registry gate counter so @@ -1546,6 +1601,20 @@ mod tests { assert_ne!(mask_from(None) & node::events::CATEGORY_MEMPOOL, 0); } + #[test] + fn status_is_served_on_ws_but_only_when_asked_for() { + // Unlike tweaks, the status category is NOT stripped from a WS mask: + // health events are low-volume JSON this surface can serve. It is still + // explicit-only, so the `0`/absent default excludes it. + assert_ne!( + mask_from(Some(node::events::CATEGORY_STATUS)) & node::events::CATEGORY_STATUS, + 0, + "an explicit status request must survive the WS mask", + ); + assert_eq!(mask_from(None) & node::events::CATEGORY_STATUS, 0); + assert_eq!(mask_from(Some(0)) & node::events::CATEGORY_STATUS, 0); + } + #[test] fn parses_scripthash_hex() { let hexstr = "ff".repeat(32); @@ -1637,6 +1706,75 @@ mod tests { assert_eq!(ws.len(), 3, "cap 0 disables the per-connection limit"); } + /// The handshake gate. `status` carries host telemetry that `rpc:read` + /// owns everywhere else, so a `stream:subscribe`-only token — the exact + /// token an operator issues *to withhold* `getblockchaininfo` — is refused + /// rather than served, and rather than silently given an empty category. + #[test] + fn the_status_category_needs_rpc_read_at_the_handshake() { + use satd_auth::{Capability, CapabilitySet, Principal}; + let stream_only = Principal::loopback(CapabilitySet::EMPTY.with(Capability::StreamSubscribe)); + let with_read = Principal::loopback( + CapabilitySet::EMPTY + .with(Capability::StreamSubscribe) + .with(Capability::RpcRead), + ); + let status = node::events::CATEGORY_STATUS; + + assert!(require_status_capability(status, Some(&stream_only)).is_err()); + assert!(require_status_capability(status, Some(&with_read)).is_ok()); + // Mixed masks are refused too — the bit is what matters, not whether it + // was asked for alone. + assert!( + require_status_capability(status | node::events::CATEGORY_CHAIN, Some(&stream_only)) + .is_err(), + ); + // Everything else is untouched for the same token. + assert!(require_status_capability(node::events::CATEGORY_CHAIN, Some(&stream_only)).is_ok()); + // The default mask does not contain status, so an ordinary subscriber + // is unaffected — this gate must not break `categories=0`. + assert!( + require_status_capability(mask_from(None), Some(&stream_only)).is_ok(), + "the default mask must stay reachable without rpc:read", + ); + // Auth disabled entirely (Core-compatible default): unchanged. + assert!(require_status_capability(status, None).is_ok()); + } + + /// …and the mid-stream control path is not a second door around it. + #[test] + fn set_categories_cannot_add_status_without_rpc_read() { + use satd_auth::{Capability, CapabilitySet, Principal}; + let reg = Arc::new(WatchRegistry::new()); + let (handle, _rx) = reg.register(WATCH_CHANNEL_CAPACITY); + let p = Principal::loopback(CapabilitySet::EMPTY.with(Capability::StreamSubscribe)); + let mask = AtomicU32::new(node::events::CATEGORY_CHAIN); + let mut ws = WatchSet::default(); + + let ctrl = format!( + r#"{{"type":"set_categories","categories":{}}}"#, + node::events::CATEGORY_STATUS | node::events::CATEGORY_CHAIN, + ); + apply_ws_control(&ctrl, &handle, Some(&p), &mask, &AtomicBool::new(false), &mut ws, 0, (8, 32)); + let got = mask.load(Ordering::Relaxed); + assert_eq!(got & node::events::CATEGORY_STATUS, 0, "status must not be added"); + assert_ne!(got & node::events::CATEGORY_CHAIN, 0, "the rest of the update applies"); + + // With rpc:read the same message is honored, so the gate is the + // capability and not the code path. + let p2 = Principal::loopback( + CapabilitySet::EMPTY + .with(Capability::StreamSubscribe) + .with(Capability::RpcRead), + ); + apply_ws_control(&ctrl, &handle, Some(&p2), &mask, &AtomicBool::new(false), &mut ws, 0, (8, 32)); + assert_ne!( + mask.load(Ordering::Relaxed) & node::events::CATEGORY_STATUS, + 0, + "rpc:read receives status", + ); + } + #[test] fn set_watch_set_applies_categories_only_when_accepted() { use satd_auth::{Accounting, Capability, CapabilitySet, LocalAccounting, Principal}; diff --git a/events/src/zmq.rs b/events/src/zmq.rs index df6094456..10f0b9f90 100644 --- a/events/src/zmq.rs +++ b/events/src/zmq.rs @@ -209,8 +209,29 @@ async fn publish( topics: &ZmqTopicConfig, counters: &TopicCounters, ) -> Result<(), zeromq::ZmqError> { - // Catch-all envelope topic (always emitted unless disabled). - if ZmqTopicConfig::enabled(topics.nodeevent) { + // Catch-all envelope topic (always emitted unless disabled), minus the + // `status` category. + // + // `status` is explicit-only: a subscriber must name the bit to receive it, + // so it never reaches a client that did not ask. gRPC/WS/SSE can honor that + // because each subscriber carries a category mask; the ZMQ fanout has no + // per-subscriber mask, only an all-or-nothing topic toggle. Emitting status + // here would hand a new body type to every already-deployed `nodeevent` + // consumer the moment they upgrade — health detectors are on by default, so + // `peer_floor` alone would start frames arriving — and their only remedy + // would be `eventszmqnodeevent=0`, which kills every other envelope too. + // + // This is not the same situation as the tweak categories, despite both + // being explicit-only: `BlockTweaks`/`MempoolTweak` reach the shared + // broadcast only while a gRPC tweaks subscriber is attached, so a ZMQ-only + // deployment never sees them either way. Status has no such gate. + // + // Status is served on the three mask-bearing transports, on webhooks, and + // through `-alertnotify`. If a ZMQ-only consumer ever needs it, the additive + // fix is an opt-in `eventszmqstatus` topic, not widening this one. + let explicit_only_on_unmasked_transport = + env.category_bit() == node::events::CATEGORY_STATUS; + if ZmqTopicConfig::enabled(topics.nodeevent) && !explicit_only_on_unmasked_transport { match serde_json::to_vec(env) { Ok(payload) => { let seq = counters.next(&counters.nodeevent); @@ -305,6 +326,13 @@ async fn publish( // full envelope is still available on `nodeevent` for consumers who // subscribe to the tweaks category there. } + NodeEventBody::Status(_) => { + // Node-health conditions are a satd-native category with no Bitcoin + // Core-compat ZMQ topic (Core surfaces the equivalent through + // `-alertnotify`, which satd also drives from the same detectors). + // Nor does it ride the catch-all `nodeevent` topic — see the + // explicit-only note at the top of `publish`. + } NodeEventBody::MempoolTweak(_) => { // Mempool-time tweak (Tier 1.5): like BlockTweaks, no Core-compat ZMQ // topic. It reaches the shared broadcast only while a gRPC @@ -638,6 +666,57 @@ mod tests { } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn nodeevent_topic_omits_the_explicit_only_status_category() { + // `status` is explicit-only, and ZMQ has no per-subscriber category + // mask to opt in with. An already-deployed `nodeevent` consumer must + // therefore not start receiving a new body type just because it + // upgraded to a build whose health detectors are on by default. + // + // Publish a status event first, then a mempool event. If status were + // emitted the first frame would be the status one; asserting the first + // frame is the mempool event proves status was skipped rather than + // merely delayed. + let publisher = EventPublisher::new(edge(), 64); + let (mp_tx, _) = broadcast::channel::(16); + let (ch_tx, _) = broadcast::channel::(16); + let (shutdown_tx, shutdown_rx) = watch::channel(false); + publisher.spawn_bridges(mp_tx.subscribe(), ch_tx.subscribe(), shutdown_rx.clone()); + + let (endpoint, _) = ephemeral_endpoint().await; + spin_up_sink( + endpoint.clone(), + ZmqTopicConfig::default(), + publisher.clone(), + shutdown_rx.clone(), + ) + .await; + + let mut sub = SubSocket::new(); + sub.connect(&endpoint).await.expect("sub connect"); + sub.subscribe(topics::NODEEVENT).await.expect("subscribe"); + tokio::time::sleep(Duration::from_millis(150)).await; + + publisher.publish_status(node::events::StatusEvent::raised( + node::events::StatusKind::PeerFloor, + "0 peers connected", + )); + mp_tx.send(enter_event(0x55)).unwrap(); + + let msg = tokio::time::timeout(Duration::from_secs(3), sub.recv()) + .await + .expect("nodeevent timeout") + .expect("recv"); + let frames: Vec<_> = msg.into_vec().into_iter().map(|b| b.to_vec()).collect(); + let parsed: serde_json::Value = serde_json::from_slice(&frames[1]).unwrap(); + assert_eq!( + parsed["body"]["category"], "mempool", + "status must not reach the unmasked ZMQ fanout" + ); + + let _ = shutdown_tx.send(true); + } + + #[tokio::test] async fn nodeevent_topic_carries_full_envelope() { let publisher = EventPublisher::new(edge(), 64); let (mp_tx, _) = broadcast::channel::(16); diff --git a/node/src/events/envelope.rs b/node/src/events/envelope.rs index 8922c3718..7922acc6e 100644 --- a/node/src/events/envelope.rs +++ b/node/src/events/envelope.rs @@ -14,6 +14,8 @@ use serde::{Serialize, Serializer}; use crate::chain::events::ChainEvent; use crate::mempool::events::MempoolEvent; +use super::status::StatusEvent; + use super::schema::SCHEMA_VERSION; /// Filename inside `` that holds the auto-generated UUIDv4 if @@ -340,11 +342,16 @@ pub const CATEGORY_HEARTBEAT: u32 = 4; /// **not** part of the `categories = 0` default, so a pre-existing subscriber /// never begins receiving tweak volume after a node upgrade. pub const CATEGORY_TWEAKS: u32 = 8; +/// Node-health conditions the daemon detects about itself ([`StatusEvent`]). +/// Explicit-request only — **not** part of the `categories = 0` default, so a +/// pre-existing subscriber never begins receiving a new event type it has no +/// parser for after a node upgrade. +pub const CATEGORY_STATUS: u32 = 16; /// Categories excluded from the `categories = 0` ("all") default: opt-in, /// high-volume, or custody-adjacent streams a legacy subscriber must not begin -/// receiving after a node upgrade. Currently just [`CATEGORY_TWEAKS`]. -pub const EXPLICIT_ONLY_CATEGORIES: u32 = CATEGORY_TWEAKS; +/// receiving after a node upgrade. +pub const EXPLICIT_ONLY_CATEGORIES: u32 = CATEGORY_TWEAKS | CATEGORY_STATUS; /// The mask a `categories = 0` request expands to: every bit (forward-compat /// for rolling upgrades — a client may request a category a mixed-version @@ -477,6 +484,13 @@ pub enum NodeEventBody { /// is excluded from the replay ring, so a `from_cursor` resume never yields /// one (missed admissions surface at confirmation via [`BlockTweaks`]). MempoolTweak(SpTweakEntry), + /// A node-health condition the daemon detected about itself (stalled tip, + /// low disk, peer starvation …); category bit [`CATEGORY_STATUS`]. + /// Explicit-request only, like tweaks. Carries no durable cursor and is + /// excluded from the replay ring — durability comes from detectors + /// re-evaluating and re-raising standing conditions after a restart + /// (see [`status`](super::status)). + Status(StatusEvent), Heartbeat { /// Nanoseconds since the [`super::EventPublisher`] was /// constructed. Lets downstream consumers measure end-to-end @@ -684,6 +698,7 @@ impl NodeEvent { NodeEventBody::Chain(_) => CATEGORY_CHAIN, NodeEventBody::BlockTweaks(_) => CATEGORY_TWEAKS, NodeEventBody::MempoolTweak(_) => CATEGORY_TWEAKS, + NodeEventBody::Status(_) => CATEGORY_STATUS, NodeEventBody::Heartbeat { .. } => CATEGORY_HEARTBEAT, // A lag notice is a control signal, not a content category: it // must reach every subscriber regardless of the category mask, so @@ -870,6 +885,50 @@ mod tests { assert_ne!(ALL_CATEGORIES_DEFAULT & CATEGORY_HEARTBEAT, 0); } + fn status_env() -> NodeEvent { + NodeEvent::new( + stamp(), + NodeEventBody::Status( + crate::events::status::StatusEvent::raised( + crate::events::status::StatusKind::DiskLow, + "free space below floor", + ) + .with_detail("free_bytes", 1_234u64), + ), + ) + } + + #[test] + fn status_category_is_explicit_only() { + // Health events carry bit 16 and, like tweaks, are excluded from the + // `categories = 0` default: a subscriber written against an older node + // must never start receiving an event body it has no parser for. + assert_eq!(CATEGORY_STATUS, 16); + assert_eq!(status_env().category_bit(), CATEGORY_STATUS); + assert_eq!( + ALL_CATEGORIES_DEFAULT & CATEGORY_STATUS, + 0, + "status must NOT be in the all-categories default", + ); + assert_ne!(EXPLICIT_ONLY_CATEGORIES & CATEGORY_STATUS, 0); + } + + #[test] + fn status_carries_no_durable_cursor() { + // Status events are not replayable: durability is by re-evaluation on + // restart, so they must never anchor a resume position. + assert!(status_env().body.derive_cursor(1, 42).is_none()); + } + + #[test] + fn status_serializes_flat_under_the_category_tag() { + let json = serde_json::to_string(&status_env().body).unwrap(); + assert_eq!( + json, + r#"{"category":"status","kind":"disk_low","state":"raised","severity":"critical","message":"free space below floor","details":{"free_bytes":"1234"}}"# + ); + } + fn tweak_entry_fixture() -> node_sp_index::TweakEntry { use bitcoin::secp256k1::{PublicKey, Secp256k1, SecretKey}; let secp = Secp256k1::new(); diff --git a/node/src/events/mod.rs b/node/src/events/mod.rs index 8c54fe85e..4f0a26dfd 100644 --- a/node/src/events/mod.rs +++ b/node/src/events/mod.rs @@ -16,6 +16,7 @@ //! the internal broadcasts into envelopes, drives a 1 Hz heartbeat, //! and supervises sink tasks. //! - [`sink`] — `EventSink` trait, runtime helpers. +//! - [`status`] — `StatusEvent` node-health conditions (`status` category). //! - [`schema`] — schema-version constant and evolution rules. pub mod envelope; @@ -23,14 +24,16 @@ pub mod publisher; pub mod replay; pub mod schema; pub mod sink; +pub mod status; pub mod watch; pub use envelope::{ BlockTweaks, Cursor, CursorRejectReason, EdgeIdentity, EdgeStamp, NodeEvent, NodeEventBody, SetCursorOutcome, SpTaprootOutput, SpTweakEntry, ALL_CATEGORIES_DEFAULT, CATEGORY_CHAIN, CATEGORY_HEARTBEAT, - CATEGORY_MEMPOOL, CATEGORY_TWEAKS, EXPLICIT_ONLY_CATEGORIES, TWEAKS_TX_INDEX, + CATEGORY_MEMPOOL, CATEGORY_STATUS, CATEGORY_TWEAKS, EXPLICIT_ONLY_CATEGORIES, TWEAKS_TX_INDEX, }; +pub use status::{StatusEvent, StatusKind, StatusSeverity, StatusState}; pub use publisher::{ EventPublisher, MempoolTweakSource, MempoolTweakSubscriberGuard, TweakSubscriberGuard, ENVELOPE_BROADCAST_CAPACITY, diff --git a/node/src/events/publisher.rs b/node/src/events/publisher.rs index 3dda37035..7e9a3cb6f 100644 --- a/node/src/events/publisher.rs +++ b/node/src/events/publisher.rs @@ -26,6 +26,7 @@ use crate::mempool::events::MempoolEvent; use super::envelope::{ BlockTweaks, Cursor, EdgeIdentity, EdgeStamp, NodeEvent, NodeEventBody, SpTweakEntry, }; +use super::status::StatusEvent; /// Read handle for the mempool's admission-cached silent-payment tweaks, keyed /// by txid. Implemented by [`crate::mempool::pool::Mempool`]; the mempool bridge @@ -228,6 +229,22 @@ impl EventPublisher { } } + /// Publish a node-health [`StatusEvent`] onto the envelope bus. + /// + /// The only publicly-callable emitter on the publisher: health conditions + /// are detected by a task outside this module (unlike mempool/chain events, + /// which are bridged from the internal broadcasts). Best-effort like every + /// other publish — with no subscribers the envelope is dropped, which is the + /// steady state on a node whose operator configured neither a `status` + /// subscriber nor a webhook. + /// + /// Status envelopes carry no durable cursor and are kept out of the replay + /// ring; see [`status`](super::status) for why re-evaluation, not replay, is + /// what makes them at-least-once across a restart. + pub fn publish_status(&self, event: StatusEvent) { + self.publish(NodeEventBody::Status(event)); + } + /// Number of envelopes published since startup. Useful for tests. pub fn published(&self) -> u64 { self.seq.load(Ordering::Relaxed) @@ -285,9 +302,15 @@ impl EventPublisher { // exists to retain. `MempoolTweak` is excluded too: it is ephemeral and // best-effort (no durable cursor), so it is never replayed — a client // that missed an admission catches the payment at confirmation. + // `Status` is excluded for the same reason: health events are not + // replayable at all (detectors re-raise standing conditions after a + // restart instead), so retaining them would only evict the mempool + // transitions the ring exists for. if !matches!( env.body, - NodeEventBody::BlockTweaks(_) | NodeEventBody::MempoolTweak(_) + NodeEventBody::BlockTweaks(_) + | NodeEventBody::MempoolTweak(_) + | NodeEventBody::Status(_) ) { let mut ring = self .replay_ring @@ -590,6 +613,39 @@ mod tests { } } + #[test] + fn status_events_stay_out_of_the_replay_ring() { + use crate::events::status::{StatusEvent, StatusKind}; + + // Ring big enough for the Enter plus every status event that follows — + // so if status events were retained they would still fit, and the only + // thing this can detect is the exclusion itself. Then a second pass with + // a ring of 1 proves they cannot evict a retained mempool transition. + for cap in [8usize, 1] { + let publisher = EventPublisher::new(edge(), cap); + publisher.publish(NodeEventBody::Mempool(enter_event(1))); + for _ in 0..4 { + publisher.publish_status(StatusEvent::raised(StatusKind::DiskLow, "low")); + } + assert_eq!( + publisher.replay_mempool_since(0).len(), + 1, + "the Enter must survive a burst of status events (ring cap {cap})", + ); + } + } + + #[test] + fn published_status_carries_no_cursor() { + use crate::events::status::{StatusEvent, StatusKind}; + let publisher = EventPublisher::new(edge(), 8); + let mut rx = publisher.subscribe(); + publisher.publish_status(StatusEvent::edge(StatusKind::IbdComplete, "synced")); + let env = rx.try_recv().expect("status envelope"); + assert!(env.cursor.is_none(), "status must not anchor a resume position"); + assert!(matches!(env.body, NodeEventBody::Status(_))); + } + fn block_event(byte: u8, height: u32) -> ChainEvent { ChainEvent::BlockConnected { hash: BlockHash::from_raw_hash( diff --git a/node/src/events/replay.rs b/node/src/events/replay.rs index c5dbd7238..961a9b309 100644 --- a/node/src/events/replay.rs +++ b/node/src/events/replay.rs @@ -261,7 +261,8 @@ pub struct CursorReplay { /// /// `tweak_source` is the silent-payment index read handle, consulted only when /// the `tweaks` bit is set. A **tweaks-only** subscription (`category_mask == -/// tweaks`) against a complete index is exempt from the `max_blocks` clamp — +/// tweaks`, ignoring the non-replayable `status` bit) against a complete index +/// is exempt from the `max_blocks` clamp — /// each `sp_tweaks` row embeds the hash of the block it describes (§3.2), so a /// phone can cold-sync from taproot activation in one subscription without a /// height→hash lookup. A **mixed** subscription keeps the clamp for every @@ -287,7 +288,14 @@ pub fn build_cursor_replay( // index replays unclamped from the requested cursor. A partial index is // never treated as authoritative (the carrier rejects a replay request in // that state before calling us), so we still gate on `is_complete()`. - let tweaks_only_sub = category_mask == CATEGORY_TWEAKS; + // `status` is masked out before the test: it contributes nothing replayable + // (never synthesized here, and excluded from the replay ring), so a client + // that asks for "my tweaks, and tell me if the node is sick" is still a + // tweaks-only subscription for replay purposes. Testing the raw mask for + // equality would silently drop the exemption and clamp a cold-syncing + // wallet to the last `max_blocks` — a truncation the Subscribe path has no + // way to signal in-band. + let tweaks_only_sub = category_mask & !super::CATEGORY_STATUS == CATEGORY_TWEAKS; let deep_exempt = tweaks_only_sub && sp.map(|s| s.is_complete()).unwrap_or(false); let requested_start = from.height.saturating_add(1); @@ -759,6 +767,30 @@ mod tests { assert!(r.events.is_empty(), "deep tweaks are streamed lazily, not buffered"); } + #[test] + fn adding_the_status_bit_does_not_forfeit_the_deep_replay_exemption() { + // "Give me my tweaks, and tell me if the node is sick" — tweaks|status + // (8|16). `status` carries nothing replayable, so this is still a + // tweaks-only subscription as far as replay is concerned and must stay + // exempt. Testing the raw mask for equality would clamp a cold-syncing + // BIP-352 wallet to the most recent `max_blocks` and, because the + // Subscribe path never surfaces `clamped` in-band, the wallet would + // silently miss every earlier block. + let src = FlatChain { tip: 50 }; + let pubr = publisher(); + let sp = MockSp { complete: true, activation: 1, tip: 50 }; + let r = build_cursor_replay( + &src, + &pubr, + cursor(0), + CATEGORY_TWEAKS | crate::events::CATEGORY_STATUS, + 10, + Some(&sp), + ); + assert!(!r.clamped, "the status bit must not forfeit the exemption"); + assert_eq!(r.deep_tweak_range, Some((1, 50)), "full span, unclamped"); + } + #[test] fn tweaks_only_replay_clamps_when_index_incomplete() { // Without completeness the exemption does not apply, so the shared diff --git a/node/src/events/status.rs b/node/src/events/status.rs new file mode 100644 index 000000000..14d1c6451 --- /dev/null +++ b/node/src/events/status.rs @@ -0,0 +1,363 @@ +//! Node-health status events (`status` category, bit +//! [`CATEGORY_STATUS`](super::CATEGORY_STATUS)). +//! +//! A [`StatusEvent`] is the daemon reporting a condition about *itself* — +//! stalled tip, low disk, congested mempool, peer starvation — rather than +//! about the chain or the mempool's contents. They are the substrate for +//! operator alerting: the same event feeds the streaming carriers, the +//! `-alertnotify` shell hook (via the warnings registry), and the webhook +//! dispatcher, so those three can never disagree about node state. +//! +//! Firing semantics are **level-triggered**: a standing condition emits +//! exactly one [`StatusState::Raised`] on entry and exactly one +//! [`StatusState::Cleared`] on recovery, with hysteresis between the two +//! thresholds so a value hovering at the line does not flap. Observations with +//! no "recovered" state (IBD finishing, a deep reorg landing) are +//! [`StatusState::Edge`] and never produce a clear. +//! +//! Status events carry no durable cursor and are not retained in the replay +//! ring: they are not replayable. Durability comes from re-evaluation instead +//! — detectors re-examine every condition at startup and re-raise the ones +//! still standing, which is what makes health alerting at-least-once across a +//! restart without any replay machinery. A condition that both raised and +//! fully cleared while a consumer was away is stale by definition and is not +//! reconstructed. + +use std::collections::BTreeMap; + +use serde::Serialize; + +/// Which health condition a [`StatusEvent`] describes. +/// +/// Additive: consumers switch on the kind and **must** tolerate an +/// unrecognized one (a newer node talking to an older client during a rolling +/// upgrade) — `message` and `severity` stay meaningful in that case. +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum StatusKind { + /// Initial block download finished (edge, at most once per process). + IbdComplete, + /// No block connected for longer than the configured window. Clears when + /// the next block connects. + /// + /// Deliberately *not* suppressed during initial block download: + /// `is_initial_block_download()` compares the tip header's timestamp + /// against the wall clock rather than tracking sync progress, so a node + /// that was caught up and then wedged re-enters it exactly when the + /// operator most needs paging. A node that is genuinely syncing connects + /// blocks continuously and never crosses the threshold on its own. + TipStall, + /// Free space on the data (or blocks) directory fell below the configured + /// floor. Clears with hysteresis above it. + DiskLow, + /// Mempool occupancy crossed the configured percentage of its byte cap. + MempoolCongested, + /// Connected-peer count sat below the configured floor for the hold time. + PeerFloor, + /// A reorg at least `alertreorgdepth` blocks deep was applied (edge). + DeepReorg, +} + +impl StatusKind { + /// The snake_case wire name, also used as the stable + /// [`NodeWarnings`](crate::warnings::NodeWarnings) id suffix + /// (`alert.`) and the `satd_alert_active{kind=...}` label value. + pub const fn as_str(self) -> &'static str { + match self { + StatusKind::IbdComplete => "ibd_complete", + StatusKind::TipStall => "tip_stall", + StatusKind::DiskLow => "disk_low", + StatusKind::MempoolCongested => "mempool_congested", + StatusKind::PeerFloor => "peer_floor", + StatusKind::DeepReorg => "deep_reorg", + } + } + + /// Every kind, for config validation and metric pre-registration. + /// + /// Kept exhaustive by the compile-time guard below the impl block. A kind + /// missing from here compiles clean and fails only at runtime, invisibly: + /// `from_str_exact` scans `ALL`, so the alertfile would reject + /// `kinds = ["the_new_kind"]` as unknown even though the streaming docs + /// list it, and the metric would never be pre-registered. + pub const ALL: [StatusKind; 6] = [ + StatusKind::IbdComplete, + StatusKind::TipStall, + StatusKind::DiskLow, + StatusKind::MempoolCongested, + StatusKind::PeerFloor, + StatusKind::DeepReorg, + ]; + + /// Parse a wire name (the inverse of [`as_str`](Self::as_str)). Used by + /// the alertfile's per-hook `kinds` filter, which rejects unknown names + /// rather than silently matching nothing. + pub fn from_str_exact(s: &str) -> Option { + Self::ALL.into_iter().find(|k| k.as_str() == s) + } + + /// Whether this kind is a one-shot observation (no `cleared` will follow). + pub const fn is_edge(self) -> bool { + matches!(self, StatusKind::IbdComplete | StatusKind::DeepReorg) + } + + /// The severity a detector reports this kind at. Fixed per kind in v1 (a + /// hook filters with `min_severity`, it does not reassign severities). + pub const fn severity(self) -> StatusSeverity { + match self { + StatusKind::IbdComplete => StatusSeverity::Info, + StatusKind::TipStall | StatusKind::DiskLow | StatusKind::DeepReorg => { + StatusSeverity::Critical + } + StatusKind::MempoolCongested | StatusKind::PeerFloor => StatusSeverity::Warning, + } + } + + /// The stable [`NodeWarnings`](crate::warnings::NodeWarnings) id for this + /// kind, so `getwarnings` / `-alertnotify` and the status stream agree. + pub fn warning_id(self) -> String { + format!("alert.{}", self.as_str()) + } +} + +/// Compile-time guard that [`StatusKind::ALL`] lists every variant. +/// +/// Adding a variant makes this `match` non-exhaustive and fails the build here, +/// pointing at the array that needs updating — rather than shipping a kind that +/// the alertfile parser rejects and the metrics registry never sees. +const _: () = { + const fn every_variant_is_in_all(k: StatusKind) -> usize { + match k { + StatusKind::IbdComplete => 0, + StatusKind::TipStall => 1, + StatusKind::DiskLow => 2, + StatusKind::MempoolCongested => 3, + StatusKind::PeerFloor => 4, + StatusKind::DeepReorg => 5, + } + } + // Also pins the array's length to the variant count. + assert!(StatusKind::ALL.len() == every_variant_is_in_all(StatusKind::DeepReorg) + 1); +}; + +/// Level-triggered lifecycle of a condition. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum StatusState { + /// Condition entered; stands until a matching [`Cleared`](Self::Cleared). + Raised, + /// Condition recovered. + Cleared, + /// One-shot observation; no clear will follow. + Edge, +} + +/// How loud a condition is. Ordered, so a hook's `min_severity` filter is a +/// simple comparison. +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum StatusSeverity { + Info, + Warning, + Critical, +} + +impl StatusSeverity { + pub const fn as_str(self) -> &'static str { + match self { + StatusSeverity::Info => "info", + StatusSeverity::Warning => "warning", + StatusSeverity::Critical => "critical", + } + } + + /// Parse a wire name; used by the alertfile's `min_severity` key, which + /// rejects unknown names. + pub fn from_str_exact(s: &str) -> Option { + match s { + "info" => Some(StatusSeverity::Info), + "warning" => Some(StatusSeverity::Warning), + "critical" => Some(StatusSeverity::Critical), + _ => None, + } + } +} + +/// One node-health observation, carried by +/// [`NodeEventBody::Status`](super::NodeEventBody::Status). +/// +/// Serializes flat under the envelope's `category` tag: +/// ```json +/// {"category":"status","kind":"tip_stall","state":"raised", +/// "severity":"critical","message":"no block connected for 3612s", +/// "details":{"seconds_since_block":"3612","threshold_seconds":"3600"}} +/// ``` +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +pub struct StatusEvent { + pub kind: StatusKind, + pub state: StatusState, + pub severity: StatusSeverity, + /// Human-readable one-liner. Safe to log or page on, **not** to parse: + /// machine consumers switch on `kind` and read `details`. + pub message: String, + /// Kind-specific structured fields, as strings. + /// + /// Mostly decimal numbers, but **not** universally: a `cleared` event can + /// carry a `reason` token (`detector_disabled`, `mempool_cap_zero`). A + /// consumer must not parse the map uniformly as integers. + /// + /// A `BTreeMap` (not a `HashMap`) so the rendered JSON has a deterministic + /// key order — the webhook body is HMAC-signed, and golden signature + /// vectors would be unreproducible under map-iteration order. + pub details: BTreeMap, +} + +impl StatusEvent { + /// Build an event, taking the severity from the kind and the state from + /// whether the kind is an edge observation. + fn build(kind: StatusKind, state: StatusState, message: impl Into) -> Self { + Self { + kind, + state, + severity: kind.severity(), + message: truncate(&message.into(), MAX_MESSAGE_LEN), + details: BTreeMap::new(), + } + } + + /// A condition entering. + /// + /// An edge kind has no standing state, so it is coerced to + /// [`StatusState::Edge`] rather than shipping a `raised` a consumer would + /// wait forever to see cleared. These constructors are total on purpose: + /// the kind is a runtime parameter at both call sites in `health.rs`, so a + /// mismatch is reachable, and a `debug_assert!` would let a release build + /// emit the malformed event silently — leaving `satd_alert_active{kind=…}` + /// stuck at 1 and a receiver waiting on a `cleared` that cannot come. + pub fn raised(kind: StatusKind, message: impl Into) -> Self { + Self::build(kind, Self::state_for(kind, StatusState::Raised), message) + } + + /// A condition recovering. Coerced to [`StatusState::Edge`] for edge kinds, + /// which never clear — see [`raised`](Self::raised). + pub fn cleared(kind: StatusKind, message: impl Into) -> Self { + Self::build(kind, Self::state_for(kind, StatusState::Cleared), message) + } + + /// The state an event of this kind may actually carry. + fn state_for(kind: StatusKind, requested: StatusState) -> StatusState { + if kind.is_edge() { + StatusState::Edge + } else if requested == StatusState::Edge { + // A standing kind cannot be an edge; `raised` is the honest + // reading of "this just happened" for one. + StatusState::Raised + } else { + requested + } + } + + /// A one-shot observation. + pub fn edge(kind: StatusKind, message: impl Into) -> Self { + Self::build(kind, Self::state_for(kind, StatusState::Edge), message) + } + + /// Attach a structured detail field (builder style). Values are stringified + /// by the caller so the map stays additive forever. + /// + /// Both key and value are truncated to [`MAX_DETAIL_LEN`]. Every producer + /// is in-tree today and emits short tokens, but this body rides a 4096-slot + /// broadcast to every subscriber *and* goes inside an HMAC-signed webhook + /// payload, so the first detector to interpolate peer-supplied text (a user + /// agent, a reject reason) should not be able to size either. + #[must_use] + pub fn with_detail(mut self, key: &str, value: impl ToString) -> Self { + self.details + .insert(truncate(key, MAX_DETAIL_LEN), truncate(&value.to_string(), MAX_DETAIL_LEN)); + self + } +} + +/// Cap on any single `details` key or value. +pub const MAX_DETAIL_LEN: usize = 256; + +/// Cap on the human-readable `message`. +pub const MAX_MESSAGE_LEN: usize = 1024; + +/// Truncate on a UTF-8 boundary (slicing mid-codepoint would panic). +fn truncate(s: &str, max: usize) -> String { + if s.len() <= max { + return s.to_string(); + } + let mut end = max; + while end > 0 && !s.is_char_boundary(end) { + end -= 1; + } + s[..end].to_string() +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn kind_names_round_trip() { + for k in StatusKind::ALL { + assert_eq!(StatusKind::from_str_exact(k.as_str()), Some(k)); + } + assert_eq!(StatusKind::from_str_exact("no_such_kind"), None); + } + + #[test] + fn severity_names_round_trip() { + for s in [ + StatusSeverity::Info, + StatusSeverity::Warning, + StatusSeverity::Critical, + ] { + assert_eq!(StatusSeverity::from_str_exact(s.as_str()), Some(s)); + } + assert_eq!(StatusSeverity::from_str_exact("fatal"), None); + } + + #[test] + fn severity_is_ordered_for_min_severity_filters() { + assert!(StatusSeverity::Critical > StatusSeverity::Warning); + assert!(StatusSeverity::Warning > StatusSeverity::Info); + } + + #[test] + fn warning_ids_are_namespaced() { + assert_eq!(StatusKind::DiskLow.warning_id(), "alert.disk_low"); + } + + #[test] + fn edge_kinds_have_no_clear() { + assert!(StatusKind::IbdComplete.is_edge()); + assert!(StatusKind::DeepReorg.is_edge()); + assert!(!StatusKind::TipStall.is_edge()); + } + + #[test] + fn details_serialize_in_deterministic_order() { + // The webhook body is HMAC-signed; a non-deterministic key order would + // make golden signature vectors unreproducible. + let ev = StatusEvent::raised(StatusKind::DiskLow, "low") + .with_detail("z_last", 1) + .with_detail("a_first", 2) + .with_detail("m_middle", 3); + let json = serde_json::to_string(&ev.details).unwrap(); + assert_eq!(json, r#"{"a_first":"2","m_middle":"3","z_last":"1"}"#); + } + + #[test] + fn serde_shape_is_flat_snake_case() { + let ev = StatusEvent::raised(StatusKind::TipStall, "no block connected for 3612s") + .with_detail("seconds_since_block", 3612u64); + let json = serde_json::to_string(&ev).unwrap(); + assert_eq!( + json, + r#"{"kind":"tip_stall","state":"raised","severity":"critical","message":"no block connected for 3612s","details":{"seconds_since_block":"3612"}}"# + ); + } +} diff --git a/satd-events-proto/proto/satd/events/v1/events.proto b/satd-events-proto/proto/satd/events/v1/events.proto index 4ddf68097..01d74a41b 100644 --- a/satd-events-proto/proto/satd/events/v1/events.proto +++ b/satd-events-proto/proto/satd/events/v1/events.proto @@ -469,10 +469,12 @@ message MempoolTweak { // Filter knobs for a subscription. All optional; an unset field means // "no filter" (the server applies its own defaults). message SubscribeRequest { - // Bitfield: mempool=1, chain=2, heartbeat=4, tweaks=8 (BIP 352, §3.5). - // Zero means "all" — EXCEPT `tweaks`, which is explicit-request only and is - // NOT included in the "all" expansion, so a pre-existing categories=0 - // subscriber does not begin receiving tweak volume after a node upgrade. The + // Bitfield: mempool=1, chain=2, heartbeat=4, tweaks=8 (BIP 352, §3.5), + // status=16 (node health, §8). + // Zero means "all" — EXCEPT `tweaks` and `status`, which are + // explicit-request only and are NOT included in the "all" expansion, so a + // pre-existing categories=0 subscriber does not begin receiving tweak volume + // or a new event type after a node upgrade. The // `tweaks` bit requires silentpaymentindex=1; a request that sets it against // a node without the index is rejected in-band. Unknown bits are silently // ignored — the server treats the request as "any of the known categories I @@ -643,6 +645,79 @@ message Heartbeat { uint64 uptime_ns = 1; } +// A node-health condition the daemon detected about itself (`status` +// category, bit 16). Explicit-request only — NOT part of the categories=0 +// "all" expansion, so a pre-existing subscriber never begins receiving a new +// event type after a node upgrade. +// +// Standing conditions are level-triggered: exactly one RAISED when the +// condition is entered and exactly one CLEARED when it recovers (with +// hysteresis so a value hovering at the threshold does not flap). One-shot +// observations that have no "recovered" state are EDGE. Detectors +// re-evaluate on startup and re-raise conditions that are still standing, so +// a consumer that was disconnected across a restart still learns about them +// (at-least-once by re-evaluation rather than by replay — a status event +// carries no durable cursor and is never replayed). +message StatusEvent { + StatusKind kind = 1; + StatusState state = 2; + StatusSeverity severity = 3; + // Human-readable one-liner. Stable-ish: safe to log or forward to a pager, + // NOT to parse — machine consumers switch on `kind` and read `details`. + string message = 4; + // Kind-specific structured fields (heights, byte counts, peer counts, + // thresholds). A string map so every future taxonomy addition stays + // additive on both the proto and JSON surfaces. + // + // Most values are decimal numbers rendered as strings, but NOT all of them: + // some kinds carry non-numeric values (e.g. a reason token). Parse per key, + // not by assuming the whole map is numeric. + // + // Keys present for a given kind are documented in docs/api/streaming.md + // §7.8; a consumer must tolerate unknown keys and missing optional ones. + map details = 5; +} + +// Which health condition a StatusEvent describes. Additive: a client MUST +// tolerate an unrecognized kind (forward-compat during a rolling upgrade) — +// `message` and `severity` remain meaningful. +enum StatusKind { + STATUS_KIND_UNSPECIFIED = 0; + // Initial block download finished (edge, at most once per process). + STATUS_KIND_IBD_COMPLETE = 1; + // No block connected for longer than the configured window. Clears when the + // next block connects (or when the window is widened past the current tip + // age). Deliberately NOT suppressed during initial block download: a node + // that was caught up and then wedged re-enters IBD a day later — exactly when + // the alert is wanted — while a node genuinely syncing connects blocks + // continuously and so never reaches the window. + STATUS_KIND_TIP_STALL = 2; + // Free space on the data (or blocks) directory fell below the configured + // floor. Clears with hysteresis above it. + STATUS_KIND_DISK_LOW = 3; + // Mempool occupancy crossed the configured percentage of its byte cap. + STATUS_KIND_MEMPOOL_CONGESTED = 4; + // Connected-peer count sat below the configured floor for the hold time. + STATUS_KIND_PEER_FLOOR = 5; + // A reorg at least `alertreorgdepth` blocks deep was applied (edge). + STATUS_KIND_DEEP_REORG = 6; +} + +// Level-triggered lifecycle of a condition. +enum StatusState { + STATUS_STATE_UNSPECIFIED = 0; + STATUS_STATE_RAISED = 1; // condition entered (standing until CLEARED) + STATUS_STATE_CLEARED = 2; // condition recovered + STATUS_STATE_EDGE = 3; // one-shot observation; no CLEARED will follow +} + +enum StatusSeverity { + STATUS_SEVERITY_UNSPECIFIED = 0; + STATUS_SEVERITY_INFO = 1; + STATUS_SEVERITY_WARNING = 2; + STATUS_SEVERITY_CRITICAL = 3; +} + // In-band notice that the server dropped events for this subscriber because // it could not keep up with the broadcast (the slow-consumer "lag"). The // stream then continues live. `resume_cursor` is the position of the last @@ -829,5 +904,6 @@ message NodeEvent { BlockTweaks block_tweaks = 28; // BIP 352 per-block tweak data (Subscribe `tweaks` category, §3.5) SilentPaymentMatched silent_payment_matched = 29; // BIP 352 scan-key watch match (Watch stream, §4) MempoolTweak mempool_tweak = 30; // BIP 352 mempool-time tweak (Subscribe `tweaks` + mempool_tweaks, §3.5) + StatusEvent status = 31; // node-health condition (Subscribe `status` category, bit 16) } }