From 2a0caddfcbf628772ac8d2299f91af2d816695fc Mon Sep 17 00:00:00 2001 From: chillin_since_26 Date: Thu, 5 Nov 2026 19:44:01 +0100 Subject: [PATCH] feat(node): keep refusal counts across restarts --- CHANGELOG.md | 11 +-- api/components/schemas/access/NodeHealth.yaml | 14 ++-- api/paths/vault/vault_api_v1_node_health.yaml | 10 +-- crates/dpp-common/src/metrics.rs | 49 +----------- crates/dpp-dal/src/pg/mod.rs | 2 + crates/dpp-dal/src/pg/repo_node_failure.rs | 66 +++++++++++++++++ crates/dpp-dal/tests/pg_node_failure.rs | 36 +++++++++ crates/dpp-node/src/boot/db.rs | 5 +- crates/dpp-node/src/boot/tasks.rs | 8 +- crates/dpp-node/src/infra/ruleset.rs | 15 +++- crates/dpp-node/src/main.rs | 14 +++- crates/dpp-node/tests/openapi_contract.rs | 4 +- crates/dpp-node/tests/smoke.rs | 9 ++- crates/dpp-types/src/lib.rs | 1 + crates/dpp-types/src/node_failure.rs | 38 ++++++++++ crates/dpp-vault/src/domain/service/mod.rs | 14 ++++ .../dpp-vault/src/domain/service/publish.rs | 74 +++++++++++-------- crates/dpp-vault/src/handlers/node_health.rs | 41 ++++++---- crates/dpp-vault/src/handlers/ruleset.rs | 16 +++- crates/dpp-vault/tests/helpers/mod.rs | 8 +- crates/dpp-vault/tests/node_failure_counts.rs | 67 +++++++++++++++++ ops/pg/0048_node_failure_counter.sql | 30 ++++++++ 22 files changed, 407 insertions(+), 125 deletions(-) create mode 100644 crates/dpp-dal/src/pg/repo_node_failure.rs create mode 100644 crates/dpp-dal/tests/pg_node_failure.rs create mode 100644 crates/dpp-types/src/node_failure.rs create mode 100644 crates/dpp-vault/tests/node_failure_counts.rs create mode 100644 ops/pg/0048_node_failure_counter.sql diff --git a/CHANGELOG.md b/CHANGELOG.md index 57d3e09e..f9159960 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -63,11 +63,12 @@ under the pre-1.0 conventions in [VERSIONING.md](docs/governance/VERSIONING.md): - **An authenticated node health route reports the counters an operator pages on.** `GET /vault/api/v1/node/health` returns the publishes refused for a failed - signature and the ruleset bundles refused on load, both since the node process - started, and the registry-sync outbox's pending, submitted, stalled and rejected - counts. A monitor that cannot scrape `/metrics` can read them with an API key. - The unauthenticated `/health` still answers liveness only. The two process - counters reset on restart; the outbox does not. (#138) + signature and the ruleset bundles refused on load, and the registry-sync + outbox's pending, submitted, stalled and rejected counts. A monitor that cannot + scrape `/metrics` can read them with an API key. The unauthenticated `/health` + still answers liveness only. The two refusal counts are kept in the database and + survive a restart, as the outbox counts always have. The Prometheus counters on + `/metrics` still reset on restart. (#138, #590) - **An authority or a person with a legitimate interest can read the passport as it stood at a past moment.** EN 18221:2026 clause 4.2 expects the version of diff --git a/api/components/schemas/access/NodeHealth.yaml b/api/components/schemas/access/NodeHealth.yaml index 4f455f57..3ee708fd 100644 --- a/api/components/schemas/access/NodeHealth.yaml +++ b/api/components/schemas/access/NodeHealth.yaml @@ -1,24 +1,22 @@ type: object -required: - - signingFailuresTotal - - rulesetLoadFailuresTotal properties: signingFailuresTotal: type: integer format: int64 minimum: 0 description: | - Publishes refused because a passport could not be signed, since the node - process started. A restart resets it, so a restart hides failures from - before it. + Publishes refused because a passport could not be signed, since the node was + first started. Kept in the database, so a restart does not reset it. Absent + on a deployment with no durable counter wired. example: 0 rulesetLoadFailuresTotal: type: integer format: int64 minimum: 0 description: | - Compliance Current ruleset bundles refused on load, since the node process - started. A restart resets it. + Compliance Current ruleset bundles refused on load, since the node was first + started. Kept in the database, so a restart does not reset it. Absent on a + deployment with no durable counter wired. example: 0 registryOutbox: $ref: ./RegistryOutboxHealth.yaml diff --git a/api/paths/vault/vault_api_v1_node_health.yaml b/api/paths/vault/vault_api_v1_node_health.yaml index e3ae4ca0..1d92875c 100644 --- a/api/paths/vault/vault_api_v1_node_health.yaml +++ b/api/paths/vault/vault_api_v1_node_health.yaml @@ -3,11 +3,11 @@ get: summary: Node operational health counters description: | Reports the counters an operator pages on: publishes refused for a failed - signature and ruleset bundles refused on load, both since the node process - started, and the registry-sync outbox's pending, submitted, stalled and - rejected counts. Authenticated, so a monitor that cannot scrape `/metrics` - can read them with an API key. The unauthenticated `/health` answers - liveness only. + signature and ruleset bundles refused on load, counted since the node was + first started and kept across restarts, and the registry-sync outbox's + pending, submitted, stalled and rejected counts. Authenticated, so a monitor + that cannot scrape `/metrics` can read them with an API key. The + unauthenticated `/health` answers liveness only. tags: - Node security: diff --git a/crates/dpp-common/src/metrics.rs b/crates/dpp-common/src/metrics.rs index 4f7e3ca9..e8806c2f 100644 --- a/crates/dpp-common/src/metrics.rs +++ b/crates/dpp-common/src/metrics.rs @@ -17,7 +17,6 @@ //! breaking change for anyone already scraping it, so this is flagged, not //! fixed, here. -use std::sync::atomic::{AtomicU64, Ordering}; use std::time::Instant; use axum::{extract::Request, middleware::Next, response::Response}; @@ -64,56 +63,14 @@ pub async fn http_metrics_middleware(request: Request, next: Next) -> Response { response } -/// Signing failures since this process started. The Prometheus recorder cannot be -/// read back, so the node-health route reads these instead. -static SIGNING_FAILURES: AtomicU64 = AtomicU64::new(0); -/// Ruleset bundles refused on load since this process started, as above. -static RULESET_LOAD_FAILURES: AtomicU64 = AtomicU64::new(0); - -/// Count one publish refused because a passport could not be signed. -/// -/// Every increment goes through here, not `metrics::counter!` directly, so the -/// count `/metrics` exports and the one the node-health route reports cannot drift. +/// Count one publish refused because a passport could not be signed, on the +/// Prometheus counter `/metrics` exports. The durable count is kept beside it by +/// the publish path, see `dpp_types::node_failure`. pub fn record_signing_failure() { - SIGNING_FAILURES.fetch_add(1, Ordering::Relaxed); metrics::counter!("signing_failures_total").increment(1); } /// Count one ruleset bundle refused on load. See [`record_signing_failure`]. pub fn record_ruleset_load_failure() { - RULESET_LOAD_FAILURES.fetch_add(1, Ordering::Relaxed); metrics::counter!("ruleset_load_failures_total").increment(1); } - -/// Signing failures since process start. A restart resets it. -#[must_use] -pub fn signing_failures() -> u64 { - SIGNING_FAILURES.load(Ordering::Relaxed) -} - -/// Ruleset bundles refused since process start. A restart resets it. -#[must_use] -pub fn ruleset_load_failures() -> u64 { - RULESET_LOAD_FAILURES.load(Ordering::Relaxed) -} - -#[cfg(test)] -mod tests { - use super::*; - - /// A refused signature shows up on the readable count, not only on the - /// exported counter the route cannot read. - #[test] - fn a_recorded_signing_failure_is_counted_where_the_route_reads_it() { - let before = signing_failures(); - record_signing_failure(); - assert_eq!(signing_failures(), before + 1); - } - - #[test] - fn a_recorded_ruleset_failure_is_counted_where_the_route_reads_it() { - let before = ruleset_load_failures(); - record_ruleset_load_failure(); - assert_eq!(ruleset_load_failures(), before + 1); - } -} diff --git a/crates/dpp-dal/src/pg/mod.rs b/crates/dpp-dal/src/pg/mod.rs index d8d82e00..2f9378e6 100644 --- a/crates/dpp-dal/src/pg/mod.rs +++ b/crates/dpp-dal/src/pg/mod.rs @@ -14,6 +14,7 @@ pub mod repo_api_key; pub mod repo_audit; pub mod repo_evidence; pub mod repo_idempotency; +pub mod repo_node_failure; pub mod repo_operator_config; pub mod repo_passport; pub mod repo_passport_version; @@ -37,6 +38,7 @@ pub use repo_api_key::PgApiKeyRepo; pub use repo_audit::PgAuditRepo; pub use repo_evidence::PgEvidenceDossierRepo; pub use repo_idempotency::PgIdempotencyRepo; +pub use repo_node_failure::PgNodeFailureRepo; pub use repo_operator_config::PgOperatorConfigRepo; pub use repo_passport::PgPassportRepo; pub use repo_passport_version::PgPassportVersionRepo; diff --git a/crates/dpp-dal/src/pg/repo_node_failure.rs b/crates/dpp-dal/src/pg/repo_node_failure.rs new file mode 100644 index 00000000..5377b77c --- /dev/null +++ b/crates/dpp-dal/src/pg/repo_node_failure.rs @@ -0,0 +1,66 @@ +//! `NodeFailureCounter` on PostgreSQL: the refusals this node has counted, kept +//! across restarts (`ops/pg/0048`). + +use async_trait::async_trait; +use sqlx::Row; + +use dpp_domain::DppError; +use dpp_types::node_failure::{FailureCounts, FailureKind, NodeFailureCounter}; + +use super::{PgDal, db_err}; + +/// PostgreSQL implementation of [`NodeFailureCounter`]. +pub struct PgNodeFailureRepo { + dal: PgDal, +} + +impl PgNodeFailureRepo { + /// Construct a repo sharing the given pool handle. + pub fn new(dal: PgDal) -> Self { + Self { dal } + } +} + +/// The name a kind is stored under. The CHECK in `ops/pg/0048` lists the same two. +fn name_of(kind: FailureKind) -> &'static str { + match kind { + FailureKind::Signing => "signing", + FailureKind::RulesetLoad => "ruleset_load", + } +} + +#[async_trait] +impl NodeFailureCounter for PgNodeFailureRepo { + async fn record(&self, kind: FailureKind) -> Result<(), DppError> { + // One statement: the first refusal creates the row and later ones add to + // it, so two refusals of the same kind racing each other cannot lose one. + sqlx::query( + "INSERT INTO odal.node_failure_counter (name, refusals) VALUES ($1, 1) \ + ON CONFLICT (name) DO UPDATE \ + SET refusals = odal.node_failure_counter.refusals + 1", + ) + .bind(name_of(kind)) + .execute(self.dal.pool()) + .await + .map_err(db_err)?; + Ok(()) + } + + async fn counts(&self) -> Result { + let rows = sqlx::query("SELECT name, refusals FROM odal.node_failure_counter") + .fetch_all(self.dal.pool()) + .await + .map_err(db_err)?; + let mut counts = FailureCounts::default(); + for row in rows { + let name: String = row.get("name"); + let refusals: i64 = row.get("refusals"); + match name.as_str() { + "signing" => counts.signing = refusals, + "ruleset_load" => counts.ruleset_load = refusals, + _ => {} + } + } + Ok(counts) + } +} diff --git a/crates/dpp-dal/tests/pg_node_failure.rs b/crates/dpp-dal/tests/pg_node_failure.rs new file mode 100644 index 00000000..ab5ec70b --- /dev/null +++ b/crates/dpp-dal/tests/pg_node_failure.rs @@ -0,0 +1,36 @@ +//! `PgNodeFailureRepo`: refusals are counted per kind, and the count outlives the +//! repo that wrote it, as it does for a node that restarts. + +#![cfg(feature = "integration-tests")] + +use dpp_dal::pg::PgNodeFailureRepo; +use dpp_dal::test_harness::start_pg; +use dpp_types::node_failure::{FailureCounts, FailureKind, NodeFailureCounter}; + +#[tokio::test(flavor = "multi_thread")] +async fn refusals_are_counted_per_kind_and_survive_a_restart() { + let pg = start_pg().await; + let before = PgNodeFailureRepo::new(pg.dal.clone()) + .counts() + .await + .unwrap(); + assert_eq!(before, FailureCounts::default(), "nothing refused yet"); + + let repo = PgNodeFailureRepo::new(pg.dal.clone()); + repo.record(FailureKind::Signing).await.unwrap(); + repo.record(FailureKind::Signing).await.unwrap(); + repo.record(FailureKind::RulesetLoad).await.unwrap(); + + // A new repo over the same database, as a node that has restarted holds. + let after_restart = PgNodeFailureRepo::new(pg.dal.clone()) + .counts() + .await + .unwrap(); + assert_eq!( + after_restart, + FailureCounts { + signing: 2, + ruleset_load: 1, + } + ); +} diff --git a/crates/dpp-node/src/boot/db.rs b/crates/dpp-node/src/boot/db.rs index 47163f21..4cebbb9a 100644 --- a/crates/dpp-node/src/boot/db.rs +++ b/crates/dpp-node/src/boot/db.rs @@ -7,7 +7,7 @@ use anyhow::Context; use async_trait::async_trait; use dpp_dal::pg::{ - PgApiKeyRepo, PgAuditRepo, PgDal, PgEvidenceDossierRepo, PgIdempotencyRepo, + PgApiKeyRepo, PgAuditRepo, PgDal, PgEvidenceDossierRepo, PgIdempotencyRepo, PgNodeFailureRepo, PgOperatorConfigRepo, PgPassportRepo, PgRegistryIdentityRepo, PgRegistrySyncRepo, PgRegistryTransferRepo, PgScanTelemetryRepo, PgSealOutboxRepo, PgSnapshotOutboxRepo, PgTransferRepo, PgWebhookRepo, @@ -45,6 +45,8 @@ pub struct DbComponents { pub api_key_repo: Arc, pub registry_repo: Arc, pub registry_outbox: Arc, + /// Durable refusal counts the node-health route reports across restarts. + pub failure_counter: Arc, pub transfer_store: Arc, pub transfer_outbox: Arc, pub evidence_store: Arc, @@ -141,6 +143,7 @@ pub async fn init_db(cfg: &NodeConfig) -> anyhow::Result { api_key_repo: Arc::new(PgApiKeyRepo::new(dal.clone())), registry_repo: Arc::new(PgRegistryIdentityRepo::new(dal.clone())), registry_outbox: Arc::new(PgRegistrySyncRepo::new(dal.clone())), + failure_counter: Arc::new(PgNodeFailureRepo::new(dal.clone())), transfer_store: Arc::new(PgTransferRepo::new(dal.clone())), transfer_outbox: Arc::new(PgRegistryTransferRepo::new(dal.clone())), evidence_store: Arc::new(PgEvidenceDossierRepo::new(dal.clone())), diff --git a/crates/dpp-node/src/boot/tasks.rs b/crates/dpp-node/src/boot/tasks.rs index 897f6ef8..31fc9ec6 100644 --- a/crates/dpp-node/src/boot/tasks.rs +++ b/crates/dpp-node/src/boot/tasks.rs @@ -971,6 +971,7 @@ pub async fn spawn_snapshot_refresh(outbox: Arc) { pub fn spawn_ruleset_poll( active: Arc, interval: Option, + failure_counter: Arc, ) { let Some(interval) = interval.filter(|_| active.has_channel()) else { return; @@ -982,7 +983,12 @@ pub fn spawn_ruleset_poll( tokio::spawn(async move { loop { tokio::time::sleep(interval).await; - dpp_node::infra::ruleset::reload_and_report(&active, false).await; + dpp_node::infra::ruleset::reload_and_report( + &active, + false, + Some(failure_counter.as_ref()), + ) + .await; } }); } diff --git a/crates/dpp-node/src/infra/ruleset.rs b/crates/dpp-node/src/infra/ruleset.rs index 524b3d38..75c4cdba 100644 --- a/crates/dpp-node/src/infra/ruleset.rs +++ b/crates/dpp-node/src/infra/ruleset.rs @@ -76,6 +76,7 @@ pub use dpp_rules::bundle::{ AcceptancePolicy, RulesetAcceptance, RulesetError, RulesetManifest, RulesetProvenance, SignedBundle, content_hash, verify_bundle, }; +use dpp_types::node_failure::{FailureKind, NodeFailureCounter}; /// How often the poller re-reads the channel when `RULESET_POLL_INTERVAL_SECS` /// is unset. Five minutes: a regulatory ruleset changes on the order of @@ -500,7 +501,11 @@ fn poll_interval_from_env() -> Option { /// `at_boot` only changes the wording: at boot a refusal means the node is /// serving its baseline, later it means the node is still serving the last /// bundle that was good. -pub async fn reload_and_report(active: &ActiveRuleset, at_boot: bool) { +pub async fn reload_and_report( + active: &ActiveRuleset, + at_boot: bool, + failure_counter: Option<&dyn NodeFailureCounter>, +) { match active.reload().await { Ok(r) if r.changed => { metrics::counter!("ruleset_swaps_total").increment(1); @@ -515,6 +520,14 @@ pub async fn reload_and_report(active: &ActiveRuleset, at_boot: bool) { Err(RulesetReloadError::NotConfigured) => {} Err(e) => { dpp_common::metrics::record_ruleset_load_failure(); + if let Some(counter) = failure_counter + && let Err(count_err) = counter.record(FailureKind::RulesetLoad).await + { + tracing::error!( + error = %count_err, + "refused ruleset bundle could not be counted durably" + ); + } tracing::error!( code = event_codes::RULESET_LOAD_FAILED, error = %e, diff --git a/crates/dpp-node/src/main.rs b/crates/dpp-node/src/main.rs index b53eb251..bcc332bf 100644 --- a/crates/dpp-node/src/main.rs +++ b/crates/dpp-node/src/main.rs @@ -203,7 +203,12 @@ async fn main() -> anyhow::Result<()> { // every start and cannot drift into a second copy of read-verify-swap. let ruleset_wiring = dpp_node::infra::ruleset::from_env(); let active_ruleset = ruleset_wiring.active.clone(); - dpp_node::infra::ruleset::reload_and_report(&active_ruleset, true).await; + dpp_node::infra::ruleset::reload_and_report( + &active_ruleset, + true, + Some(db.failure_counter.as_ref()), + ) + .await; tracing::info!(version = %active_ruleset.version(), "active ruleset"); // ── Vault service state ──────────────────────────────────────────────────── @@ -266,6 +271,7 @@ async fn main() -> anyhow::Result<()> { ) .with_registry_reader(db.operator_repo.clone()) .with_registry_outbox(db.registry_outbox.clone()) + .with_failure_counter(db.failure_counter.clone()) .with_versions(db.version_store.clone()) .with_successors(db.successors.clone()) .with_transfer_store(db.transfer_store.clone()) @@ -428,7 +434,11 @@ async fn main() -> anyhow::Result<()> { boot::tasks::spawn_job_cleanup(db.job_store.clone()); boot::tasks::spawn_idempotency_purge(db.idempotency.clone()); boot::tasks::spawn_scan_prune(db.scan_repo.clone()); - boot::tasks::spawn_ruleset_poll(active_ruleset.clone(), ruleset_wiring.poll_interval); + boot::tasks::spawn_ruleset_poll( + active_ruleset.clone(), + ruleset_wiring.poll_interval, + db.failure_counter.clone(), + ); boot::tasks::spawn_registry_drain( db.registry_outbox.clone(), registry_sync_for_drain.clone(), diff --git a/crates/dpp-node/tests/openapi_contract.rs b/crates/dpp-node/tests/openapi_contract.rs index ef6a4065..3ecebc23 100644 --- a/crates/dpp-node/tests/openapi_contract.rs +++ b/crates/dpp-node/tests/openapi_contract.rs @@ -3866,8 +3866,8 @@ mod fixtures { /// serves. The values are fixed and non-zero so each field is exercised. pub fn node_health() -> NodeHealth { NodeHealth { - signing_failures_total: 1, - ruleset_load_failures_total: 0, + signing_failures_total: Some(1), + ruleset_load_failures_total: Some(0), registry_outbox: Some(registry_outbox_health()), } } diff --git a/crates/dpp-node/tests/smoke.rs b/crates/dpp-node/tests/smoke.rs index 7e91a4ca..859e7e3b 100644 --- a/crates/dpp-node/tests/smoke.rs +++ b/crates/dpp-node/tests/smoke.rs @@ -18,9 +18,9 @@ use metrics_exporter_prometheus::{PrometheusBuilder, PrometheusHandle}; use base64::Engine as _; use dpp_crypto::keystore::KeyStore; use dpp_dal::pg::{ - PgApiKeyRepo, PgAuditRepo, PgDal, PgEvidenceDossierRepo, PgOperatorConfigRepo, PgPassportRepo, - PgPassportVersionRepo, PgRegistryIdentityRepo, PgScanTelemetryRepo, PgTransferRepo, - PgWebhookRepo, + PgApiKeyRepo, PgAuditRepo, PgDal, PgEvidenceDossierRepo, PgNodeFailureRepo, + PgOperatorConfigRepo, PgPassportRepo, PgPassportVersionRepo, PgRegistryIdentityRepo, + PgScanTelemetryRepo, PgTransferRepo, PgWebhookRepo, }; use dpp_dal::test_harness::{TestPg, start_pg}; use dpp_domain::passport::PassportRef; @@ -188,6 +188,7 @@ async fn start_node_with_ruleset( .with_registry_reader(operator_repo.clone()) // Wired as `boot::db` wires it, so the node-health route reads a real outbox. .with_registry_outbox(Arc::new(dpp_dal::pg::PgRegistrySyncRepo::new(dal.clone()))) + .with_failure_counter(Arc::new(PgNodeFailureRepo::new(dal.clone()))) .with_versions(version_store) // Wired as `boot::db` wires it, so the public-read branch that resolves // a superseded carrier on to its successor is actually exercised here. @@ -430,7 +431,7 @@ async fn a_running_node_adopts_a_newly_published_ruleset_without_a_restart() { ActiveRuleset::baseline() .with_channel(Box::new(FileRulesetSource::new(&bundle_path)), &pubkey), ); - dpp_node::infra::ruleset::reload_and_report(&ruleset, true).await; + dpp_node::infra::ruleset::reload_and_report(&ruleset, true, None).await; let pg = start_pg().await; let base = start_node_with_ruleset(pg.dal.clone(), ruleset).await; diff --git a/crates/dpp-types/src/lib.rs b/crates/dpp-types/src/lib.rs index 0d87f5eb..467f7ea8 100644 --- a/crates/dpp-types/src/lib.rs +++ b/crates/dpp-types/src/lib.rs @@ -29,6 +29,7 @@ pub mod api_key; pub mod audit; pub mod auth; pub mod evidence; +pub mod node_failure; pub mod operator; pub mod passport_request; pub mod qualification; diff --git a/crates/dpp-types/src/node_failure.rs b/crates/dpp-types/src/node_failure.rs new file mode 100644 index 00000000..c1829a93 --- /dev/null +++ b/crates/dpp-types/src/node_failure.rs @@ -0,0 +1,38 @@ +//! Refusals this node has counted, kept in the database so a restart does not +//! clear them. +//! +//! The Prometheus counters for the same refusals reset when the process restarts. +//! The node-health route answers a monitor that reads the value rather than its +//! change, so the count it serves is kept where a restart cannot reach it. The +//! PostgreSQL implementation is `dpp-dal`'s `PgNodeFailureRepo` (`ops/pg/0048`). + +use async_trait::async_trait; +use dpp_domain::DppError; + +/// The refusal a count is kept for. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum FailureKind { + /// A publish refused because a passport could not be signed. + Signing, + /// A Compliance Current ruleset bundle refused on load. + RulesetLoad, +} + +/// Every kind's count since the node was first started. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub struct FailureCounts { + /// Publishes refused for a signature. + pub signing: i64, + /// Ruleset bundles refused on load. + pub ruleset_load: i64, +} + +/// The durable refusal counts. +#[async_trait] +pub trait NodeFailureCounter: Send + Sync { + /// Count one refusal of the given kind. + async fn record(&self, kind: FailureKind) -> Result<(), DppError>; + + /// The counts so far, for every kind. + async fn counts(&self) -> Result; +} diff --git a/crates/dpp-vault/src/domain/service/mod.rs b/crates/dpp-vault/src/domain/service/mod.rs index a9ddac9f..1fc75840 100644 --- a/crates/dpp-vault/src/domain/service/mod.rs +++ b/crates/dpp-vault/src/domain/service/mod.rs @@ -64,6 +64,7 @@ use dpp_types::{ STANDALONE_OPERATOR_ID, audit::AuditRepository, evidence::EvidenceDossierRepository, + node_failure::NodeFailureCounter, operator::OperatorConfigRepository, registry_sync::{RegistryStatusIntent, RegistrySyncOutbox}, registry_transfer::RegistryTransferOutbox, @@ -110,6 +111,10 @@ pub struct PassportService { /// `None` (test doubles / in-memory repo), publish falls back to the legacy /// inline path. pub registry_outbox: Option>, + /// Durable count of publishes refused for a signature. Read by the node-health + /// route, so the count survives a restart. `None` (test doubles) leaves the + /// refusals on the Prometheus counter alone, and the route omits the count. + pub failure_counter: Option>, /// Archived versions of passports — EN 18221:2026 clause 4.2. /// /// This is now the only thing in this struct called an archive, and it is @@ -244,6 +249,7 @@ impl PassportService { registry_sync, backup, registry_outbox: None, + failure_counter: None, versions: None, successors: None, transfer_store: None, @@ -309,6 +315,14 @@ impl PassportService { self } + /// Count the refusals this service makes in the database as well as on the + /// Prometheus counter, so the node-health route reports them across restarts. + #[must_use] + pub fn with_failure_counter(mut self, counter: Arc) -> Self { + self.failure_counter = Some(counter); + self + } + /// Declare the public base URL of this deployment's continuity snapshots, /// so registrations can carry the passport's back-up link. #[must_use] diff --git a/crates/dpp-vault/src/domain/service/publish.rs b/crates/dpp-vault/src/domain/service/publish.rs index ae97ea23..9355b86b 100644 --- a/crates/dpp-vault/src/domain/service/publish.rs +++ b/crates/dpp-vault/src/domain/service/publish.rs @@ -19,6 +19,7 @@ use dpp_types::{ STANDALONE_OPERATOR_ID, audit::PassportAuditEntry, auth::AuthContext, + node_failure::FailureKind, snapshot::snapshot_json_key, state_change::{PassportWrite, StateChange}, }; @@ -148,6 +149,19 @@ impl PassportService { /// /// Returns `DppError::InvalidTransition` if the passport is not in a publishable /// state, or `DppError::Signing` if the identity service fails. + /// Count one refused signature on the Prometheus counter and, when a durable + /// counter is wired, in the database. A failure to count it durably is logged + /// rather than returned: the publish is already refused, and that refusal is the + /// error the caller needs. + async fn record_signing_refusal(&self) { + dpp_common::metrics::record_signing_failure(); + if let Some(counter) = &self.failure_counter + && let Err(e) = counter.record(FailureKind::Signing).await + { + tracing::error!(error = %e, "refused signature could not be counted durably"); + } + } + #[tracing::instrument(skip(self), fields(passport_id = %id))] pub async fn publish(&self, id: PassportId, auth: &AuthContext) -> Result { let mut passport = self.find_by_id(id).await?; @@ -390,21 +404,19 @@ impl PassportService { let payload = serde_json::to_value(&passport).map_err(|e| DppError::Serialisation(e.to_string()))?; - let jws = self - .identity - .sign_passport(passport.id, &payload) - .await - .map(|c| c.jws) - .map_err(|e| { - dpp_common::metrics::record_signing_failure(); + let jws = match self.identity.sign_passport(passport.id, &payload).await { + Ok(c) => c.jws, + Err(e) => { + self.record_signing_refusal().await; metrics::counter!("passport_publish_rejected_total", "reason" => REASON_SIGNING_FAILED).increment(1); tracing::error!( code = event_codes::JWS_UNSIGNED_PUBLISH_BLOCKED, error = %e, "publish aborted — signing failed; passport remains draft" ); - DppError::Signing(e.to_string()) - })?; + return Err(DppError::Signing(e.to_string())); + } + }; passport.jws_signature = Some(jws); // Public verifiability: also sign the *public (redacted) view* — the exact @@ -418,21 +430,19 @@ impl PassportService { // over itself; the full-payload `jws_signature` above stays Confidential // for authenticated full-passport verification. let public_view = crate::public_view::public_view(&passport); - let public_jws = self - .identity - .sign_passport(passport.id, &public_view) - .await - .map(|c| c.jws) - .map_err(|e| { - dpp_common::metrics::record_signing_failure(); + let public_jws = match self.identity.sign_passport(passport.id, &public_view).await { + Ok(c) => c.jws, + Err(e) => { + self.record_signing_refusal().await; metrics::counter!("passport_publish_rejected_total", "reason" => REASON_SIGNING_FAILED).increment(1); tracing::error!( code = event_codes::JWS_UNSIGNED_PUBLISH_BLOCKED, error = %e, "publish aborted — public-view signing failed; passport remains draft" ); - DppError::Signing(e.to_string()) - })?; + return Err(DppError::Signing(e.to_string())); + } + }; passport.public_jws_signature = Some(public_jws); // Each non-public disclosure set gets its own proof over its own view. @@ -443,22 +453,26 @@ impl PassportService { // siblings so all three describe the same moment, and keyed by // disclosure set so the artefact names the data it covers rather than // the actor vocabulary of one regulation. - passport.disclosure_signatures = crate::public_view::sign_disclosure_views( + let disclosure_signatures = match crate::public_view::sign_disclosure_views( self.identity.as_ref(), &passport, ) .await - .map_err(|e| { - dpp_common::metrics::record_signing_failure(); - metrics::counter!("passport_publish_rejected_total", "reason" => REASON_SIGNING_FAILED) - .increment(1); - tracing::error!( - code = event_codes::JWS_UNSIGNED_PUBLISH_BLOCKED, - error = %e, - "publish aborted — disclosure-view signing failed; passport remains draft" - ); - DppError::Signing(e.to_string()) - })?; + { + Ok(signatures) => signatures, + Err(e) => { + self.record_signing_refusal().await; + metrics::counter!("passport_publish_rejected_total", "reason" => REASON_SIGNING_FAILED) + .increment(1); + tracing::error!( + code = event_codes::JWS_UNSIGNED_PUBLISH_BLOCKED, + error = %e, + "publish aborted — disclosure-view signing failed; passport remains draft" + ); + return Err(DppError::Signing(e.to_string())); + } + }; + passport.disclosure_signatures = disclosure_signatures; // The EU-registry registration this publish owes, if one can be built. // diff --git a/crates/dpp-vault/src/handlers/node_health.rs b/crates/dpp-vault/src/handlers/node_health.rs index 998a8aa5..3ba33528 100644 --- a/crates/dpp-vault/src/handlers/node_health.rs +++ b/crates/dpp-vault/src/handlers/node_health.rs @@ -31,11 +31,14 @@ pub struct RegistryOutboxHealth { #[derive(Serialize)] #[serde(rename_all = "camelCase")] pub struct NodeHealth { - /// Publishes refused because a passport could not be signed, since the - /// process started. - pub signing_failures_total: u64, - /// Ruleset bundles refused on load, since the process started. - pub ruleset_load_failures_total: u64, + /// Publishes refused because a passport could not be signed, since the node + /// was first started. Absent on a deployment with no durable counter wired. + #[serde(skip_serializing_if = "Option::is_none")] + pub signing_failures_total: Option, + /// Ruleset bundles refused on load, since the node was first started. Absent + /// on a deployment with no durable counter wired. + #[serde(skip_serializing_if = "Option::is_none")] + pub ruleset_load_failures_total: Option, /// Absent on a deployment with no registry-sync outbox wired, rather than /// reported as zeros a monitor would read as an empty queue. #[serde(skip_serializing_if = "Option::is_none")] @@ -47,12 +50,19 @@ pub struct NodeHealth { /// as `/node/state`: a failing signer or a stalled queue is a description of /// where the node is weak. The unauthenticated `/health` answers liveness only. /// -/// The two process counters reset when the node restarts, so a restart hides -/// failures from before it. The outbox is read from the database and does not. +/// The two refusal counts are kept in the database, so a restart does not clear +/// them. The Prometheus counters on `/metrics` still reset on restart. pub async fn node_health_handler( State(state): State, Extension(_auth): Extension, ) -> impl IntoResponse { + let refusals = match &state.service.failure_counter { + Some(counter) => match counter.counts().await { + Ok(c) => Some(c), + Err(e) => return internal_error(e), + }, + None => None, + }; let registry_outbox = match &state.service.registry_outbox { Some(outbox) => match outbox.status_counts(STALL_THRESHOLD).await { Ok(c) => Some(RegistryOutboxHealth { @@ -69,8 +79,8 @@ pub async fn node_health_handler( ( StatusCode::OK, Json(NodeHealth { - signing_failures_total: dpp_common::metrics::signing_failures(), - ruleset_load_failures_total: dpp_common::metrics::ruleset_load_failures(), + signing_failures_total: refusals.map(|c| c.signing), + ruleset_load_failures_total: refusals.map(|c| c.ruleset_load), registry_outbox, }), ) @@ -81,18 +91,17 @@ pub async fn node_health_handler( mod tests { use super::*; - /// With no outbox wired the block is omitted entirely; the counters stay. + /// With no counter or outbox wired, each block is omitted rather than reported + /// as zeros a monitor would read as a healthy node. #[test] - fn an_unwired_outbox_is_omitted_rather_than_reported_as_empty() { + fn an_unwired_counter_and_outbox_are_omitted_rather_than_reported_as_zero() { let json = serde_json::to_value(NodeHealth { - signing_failures_total: 0, - ruleset_load_failures_total: 0, + signing_failures_total: None, + ruleset_load_failures_total: None, registry_outbox: None, }) .expect("serialises"); - assert!(json.get("registryOutbox").is_none(), "{json}"); - assert_eq!(json["signingFailuresTotal"], 0); - assert_eq!(json["rulesetLoadFailuresTotal"], 0); + assert_eq!(json, serde_json::json!({}), "{json}"); } } diff --git a/crates/dpp-vault/src/handlers/ruleset.rs b/crates/dpp-vault/src/handlers/ruleset.rs index bc865c40..aeca24ee 100644 --- a/crates/dpp-vault/src/handlers/ruleset.rs +++ b/crates/dpp-vault/src/handlers/ruleset.rs @@ -16,6 +16,7 @@ use axum::{ response::{IntoResponse, Response}, }; use dpp_common::ruleset_admin::RulesetReloadError; +use dpp_types::node_failure::FailureKind; use crate::{middleware::auth::AuthContext, state::AppState}; @@ -45,7 +46,20 @@ pub async fn reload_ruleset_handler( match admin.reload().await { Ok(report) => (StatusCode::OK, Json(report)).into_response(), - Err(e) => reload_error(e), + Err(e) => { + if !matches!(e, RulesetReloadError::NotConfigured) { + dpp_common::metrics::record_ruleset_load_failure(); + if let Some(counter) = &state.service.failure_counter + && let Err(count_err) = counter.record(FailureKind::RulesetLoad).await + { + tracing::error!( + error = %count_err, + "refused ruleset bundle could not be counted durably" + ); + } + } + reload_error(e) + } } } diff --git a/crates/dpp-vault/tests/helpers/mod.rs b/crates/dpp-vault/tests/helpers/mod.rs index 969408fa..e782dc06 100644 --- a/crates/dpp-vault/tests/helpers/mod.rs +++ b/crates/dpp-vault/tests/helpers/mod.rs @@ -13,9 +13,10 @@ use async_trait::async_trait; use base64::Engine; use dpp_dal::pg::{ - PgApiKeyRepo, PgAuditRepo, PgDal, PgEvidenceDossierRepo, PgOperatorConfigRepo, PgPassportRepo, - PgPassportVersionRepo, PgRegistryIdentityRepo, PgRegistrySyncRepo, PgRegistryTransferRepo, - PgScanTelemetryRepo, PgSealOutboxRepo, PgTransferRepo, PgWebhookRepo, + PgApiKeyRepo, PgAuditRepo, PgDal, PgEvidenceDossierRepo, PgNodeFailureRepo, + PgOperatorConfigRepo, PgPassportRepo, PgPassportVersionRepo, PgRegistryIdentityRepo, + PgRegistrySyncRepo, PgRegistryTransferRepo, PgScanTelemetryRepo, PgSealOutboxRepo, + PgTransferRepo, PgWebhookRepo, }; use dpp_domain::{ DppError, GhostBackup, GhostRegistrySync, PassthroughRegistry, @@ -603,6 +604,7 @@ async fn start_vault_with_identity( // because nothing was. .with_transfer_store(Arc::new(PgTransferRepo::new(dal.clone()))) .with_evidence_store(Arc::new(PgEvidenceDossierRepo::new(dal.clone()))) + .with_failure_counter(Arc::new(PgNodeFailureRepo::new(dal.clone()))) // The node commits every state change through the state-change log, so a // harness without it would write the parts one after another, a path no // real deployment takes. diff --git a/crates/dpp-vault/tests/node_failure_counts.rs b/crates/dpp-vault/tests/node_failure_counts.rs new file mode 100644 index 00000000..9065accb --- /dev/null +++ b/crates/dpp-vault/tests/node_failure_counts.rs @@ -0,0 +1,67 @@ +//! A publish refused for its signature is counted in the database, where a restart +//! cannot clear it. The count is read through a fresh repo over the same database, +//! as a node that has restarted would read it. + +#![cfg(feature = "integration-tests")] + +mod helpers; + +use dpp_dal::pg::PgNodeFailureRepo; +use dpp_types::node_failure::{FailureCounts, NodeFailureCounter}; +use helpers::{ + TestClient, make_jwt, seed_complete_operator, start_postgres, start_vault_failing_signer, +}; + +fn op() -> String { + "00000000-0000-0000-0000-000000000001".to_owned() +} + +fn sample() -> serde_json::Value { + serde_json::json!({ + "productName": "Test Widget", + "manufacturer": {"name": "ACME Corp", "address": "Berlin, DE"}, + "materials": [], + }) +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_publish_refused_for_its_signature_is_counted_durably() { + let pg = start_postgres().await; + let base = start_vault_failing_signer(pg.dal.clone()).await; + seed_complete_operator(&pg.dal).await; + let client = TestClient::new(&base, make_jwt(&op())); + + // Creation does not involve the signer, so the draft is made first. + let resp = client.post_json("/api/v1/dpp", sample()).await; + assert_eq!( + resp.status(), + 201, + "create must succeed regardless of signer" + ); + let id = resp.json::().await.unwrap()["id"] + .as_str() + .unwrap() + .to_owned(); + + let publish = client + .post_json(&format!("/api/v1/dpp/{id}/publish"), serde_json::json!({})) + .await; + assert!( + publish.status().is_server_error(), + "publish with a broken signer must fail, got {}", + publish.status() + ); + + let counts = PgNodeFailureRepo::new(pg.dal.clone()) + .counts() + .await + .unwrap(); + assert_eq!( + counts, + FailureCounts { + signing: 1, + ruleset_load: 0, + }, + "one refused signature, counted once" + ); +} diff --git a/ops/pg/0048_node_failure_counter.sql b/ops/pg/0048_node_failure_counter.sql new file mode 100644 index 00000000..92652a6c --- /dev/null +++ b/ops/pg/0048_node_failure_counter.sql @@ -0,0 +1,30 @@ +-- ============================================================================ +-- 0048 — node_failure_counter: refusals this node has counted, kept across +-- restarts. +-- +-- Two refusals are counted for the node-health route: +-- +-- * `signing` — a publish refused because a passport could not be +-- signed. The publish path returns before any write, so +-- the refusal leaves no row this table could be rebuilt +-- from; +-- * `ruleset_load` — a Compliance Current ruleset bundle refused on load. +-- +-- The Prometheus counters with the same meaning reset when the process +-- restarts, which is the convention for a series a scraper reads over time. A +-- monitor that reads the value rather than its change would be told that a node +-- whose signing failed for a week is healthy after one restart, so the count the +-- route serves is kept here. +-- +-- One row per kind, named by the code that increments it. The first refusal of a +-- kind creates its row, so the table starts empty. +-- +-- Single-tenant: no `operator_id` column. +-- ============================================================================ + +CREATE TABLE odal.node_failure_counter ( + name TEXT PRIMARY KEY CHECK (name IN ('signing', 'ruleset_load')), + refusals BIGINT NOT NULL DEFAULT 0 CHECK (refusals >= 0) +); + +GRANT SELECT, INSERT, UPDATE ON odal.node_failure_counter TO odal_app;