From 85b7331784804fbf9bb07df4199d4dc6a2b42ade Mon Sep 17 00:00:00 2001 From: Ian Oberst Date: Tue, 13 Oct 2026 13:57:54 -0700 Subject: [PATCH] fix(relay): partition audit observability follow-ups (#7994) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit > [!IMPORTANT] > **The alert rules changed in c3a314e71.** If you reviewed the alert guidance earlier, please look again. > - **The two range-based rules now apply only to live pods.** A range selector such as `[30m]` keeps reading a terminated pod's samples until they leave the window. So `BuzzPartitionAuditNotRunning` could fire for about 10 minutes after normal pod turnover, and `BuzzPartitionAuditFailing` could keep firing for up to an hour after the pod was gone. Each rule now adds `and `, which requires the counter to be exported right now: > ```yaml > expr: | > sum without (outcome) (increase(buzz_partition_audit_runs_total[30m])) == 0 > and sum without (outcome) (buzz_partition_audit_runs_total) > ``` > ```yaml > expr: | > increase(buzz_partition_audit_failures_total[1h]) >= 2 > and buzz_partition_audit_failures_total > ``` > - **The audit counters start at 0.** When the relay starts, it exports every `buzz_partition_audit_runs_total{table, outcome}` series and `buzz_partition_audit_failures_total` at 0, before the first audit. `increase()` ignores a new series' first sample. Before this change, it could miss the first periodic failure, or read a table's first `degraded` run as a stopped loop. This change is also what makes the `and ` gate work: the counters now exist for a relay's whole life. > > The doc's old claim that pod turnover can't leave an alert firing was true only of the instant-selector stale rule. The doc now explains the difference. ## Why These are the observability follow-ups deferred from #6515 ("fix(db): audit partition catalog before creation"): - **Timestamp-age alerts go silent** ([review comment](https://github.com/block/buzz/pull/6515#discussion_r4127212141)). Partition gauges are set only when a table audits successfully, and the exporter evicts idle gauges after 2,700 s by default. After sustained failures, `buzz_partition_audit_last_success_timestamp_seconds` disappears. A table that has never audited successfully never has it. An alert on `time() - last_success` then has no input. - **Startup fallback failures are not counted** ([review comment](https://github.com/block/buzz/pull/6515#discussion_r4127212158)). When the startup fallback audit fails, it logs an error but does not increment `buzz_partition_audit_failures_total`. Periodic failures do. - **Repaired environments are always `degraded`**. `PartitionTableAudit::degraded()` counts every legacy leaf, even one wholly in the past. The `bb-block-staging` audit of the repaired catalog reports `degraded` only because of `events_p2026_07_08_legacy` `[2026-07-01, 2026-09-01)`. With this rule, every repaired environment stays `degraded`, so nobody can alert on the outcome. - **Partition config tests can see a poisoned env** ([review comment](https://github.com/block/buzz/pull/6515#discussion_r4137416297)). Two partition config tests take only `ENV_MUTEX`. A NIP-FI test running in the same process can make them see its deliberately invalid environment. Partition health is still diagnostic only, through `/_status.partition_catalog` and metrics. Nothing here affects `/_readiness`. Handling concurrent creation (#6515 item 4) is out of scope and stays with the catch-all advancement work. ## What - **Success timestamp survives failures.** Each failed table audit refreshes `buzz_partition_audit_last_success_timestamp_seconds` with `increment(0.0)`, which leaves the value unchanged. It still resets the exporter's idle tracking, the same mechanism as `refresh_legacy_active_gauge_recency`. Sustained failures therefore keep the last success exported and its age rising. A table that has never succeeded exports `0`. A failed audit still never makes the last success look fresh. - **Alert guidance.** New `docs/partition-catalog-monitoring.md` covers the metrics, how long each series lives, and the audit outcomes. It notes that `runs_total` has no `unsafe` outcome. It gives three alert rules: - timestamp age, which also covers never-successful tables and sustained failures; - `sum without (outcome) (increase(buzz_partition_audit_runs_total[30m])) == 0` for a stopped audit loop. Counters are never idle-evicted, and this rule fires before the stopped loop's timestamp is evicted, so there is no false resolve; - an aggregate failure-counter rule. Both range-based rules also require their counter to be exported right now (`and `), so a terminated pod's leftover samples can't raise or hold an alert. - **Zero counter baselines.** `try_install` exports each `buzz_partition_audit_runs_total{table, outcome}` series (`ok`, `degraded`, and `error` for both tables) and `buzz_partition_audit_failures_total` at 0 before the startup audit, the same way the auth series already start at 0. A failure in the startup audit itself can still happen before the first scrape. The doc says so; the stale rule covers that case. - **Failure counter.** `buzz_partition_audit_failures_total` now counts failed explicit audit attempts: the startup fallback audit and each periodic audit. Both paths go through one helper, `counted_partition_audit`. A failed `ensure_future_partitions` call, such as a DDL error that the fallback then recovers, is not counted by itself. A single attempt never counts twice. - **Historical legacy leaves.** A legacy leaf is historical when its finite upper bound is at or before the current month start. It is still reported under `children` as `legacy_leaf`, but it no longer degrades the table. A legacy leaf that still covers the current month or any lookahead month stays degraded. All other degradation causes are unchanged. The verdict now lives in `PartitionAuditReport::outcome()`, which `buzz-admin partition-audit` uses. The JSON outcome names and exit codes are the same as before. - **Test hygiene.** Both partition config tests now use `env_guards()`. ## How it was tested - `cargo fmt --all -- --check` and `cargo clippy -p buzz-db -p buzz-relay -p buzz-admin --all-targets -- -D warnings`. - New unit tests: - The repaired incident layout with an injected clock: `degraded` in August 2026 and `ok` on 2026-09-01T00:00Z, when the upper bound equals the current month start, and on 2026-09-15. The legacy leaf is still in the serialized report. The `outcome` label on the run counter follows the verdict. - A table test of the legacy-leaf predicate: the upper bound before, at, and one second after the month start; lookahead and beyond-horizon leaves; historical leaves with trigger drift; anomalous and `DEFAULT` children; no audited month. - Outcome precedence (error, then unsafe, then degraded) and the `buzz-admin` exit-code mapping. - Startup: a failed fallback counts once, a successful fallback after a DDL failure counts nothing, and a successful ensure never runs the fallback. Each periodic failure counts once. Against an unreachable database, startup counts one aggregate failure while each table still records two `error` runs. - On the production exporter, never-successful audits leave run counters and a `0` success timestamp for both tables. - On the production exporter, all six run-counter series and the failure counter read 0 before any audit. After one failed audit, `error` reads 1 and `ok`/`degraded` stay at 0. Removing the initialization fails this test. - PostgreSQL lane (`scripts/postgres-test-run.sh`) with new or extended tests: - The repaired layout is not degraded in September. The same catalog is degraded in August. - With a 1 s idle timeout, a failed audit leaves the success timestamp unchanged, and sustained failures keep it exported past the idle timeout. When audits stop, the timestamp is evicted but the run counters remain. The next success exports a fresh timestamp. - Mutation check: removing the failure-path refresh fails both alert fixtures. - These fixtures check the emitted series. They do not evaluate the rules themselves, including the `and ` gate. `promtool` isn't available locally or in CI, and the doc points to `promtool test rules` for that. - `scripts/test-postgres-test-discovery.sh`. - `config::` and `nip_fi_config::` tests run together in one process five times. ## Related This PR conflicts textually with #7993 (catch-all advancement) in `crates/buzz-db/src/store/partition.rs` and `crates/buzz-relay/src/main.rs`. Whichever PR merges second will be rebased onto the other. 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Signed-off-by: Ian Oberst Co-authored-by: Ian Oberst Co-authored-by: Claude Opus 5.5 --- Justfile | 12 +- crates/buzz-admin/src/main.rs | 45 +- crates/buzz-db/src/store/partition.rs | 457 +++++++++++++++++- crates/buzz-relay/src/config.rs | 4 +- crates/buzz-relay/src/main.rs | 229 +++++++-- crates/buzz-relay/src/metrics.rs | 172 +++++++ crates/buzz-relay/src/state.rs | 8 +- docs/partition-catalog-monitoring.md | 150 ++++++ ...artition-catalog-monitoring.rules.test.yml | 122 +++++ docs/partition-catalog-monitoring.rules.yml | 20 + scripts/run-tests.sh | 6 + 11 files changed, 1156 insertions(+), 69 deletions(-) create mode 100644 docs/partition-catalog-monitoring.md create mode 100644 docs/partition-catalog-monitoring.rules.test.yml create mode 100644 docs/partition-catalog-monitoring.rules.yml diff --git a/Justfile b/Justfile index 8d107af69e..d73a9cc547 100644 --- a/Justfile +++ b/Justfile @@ -448,10 +448,11 @@ test-unit: # suites are otherwise absent from the infra-free unit lane. cargo nextest run -p buzz-media --lib \ -E 'test(=bucket_index::tests::bucket_snapshot_json_round_trip_preserves_community_keys)' - # buzz-admin: storage-snapshot worker plus the community archive - # command parser, validation, and propagation-evidence tests. + # buzz-admin: storage-snapshot worker, the community archive command + # parser, validation, and propagation-evidence tests, and the partition + # audit outcome contract. cargo nextest run -p buzz-admin \ - -E 'test(storage_snapshot) + test(/^communities::tests::/) + test(/^tests::communities_/)' + -E 'test(storage_snapshot) + test(/^communities::tests::/) + test(/^tests::communities_/) + test(=tests::partition_audit_outcomes_keep_their_json_names_and_exit_codes)' # Multi-tenant conformance gate (buzz-conformance): the independent # replay checker + golden fixtures. No infra — pure in-process trace # replay — so it belongs in the unit job. Run all targets (lib + the @@ -539,7 +540,9 @@ test-unit: # DB-backed audio join tests use #[ignore] and run in the postgres lane. # NIP-FI (S3/S4) relay witnesses: the wholly-new nip_fi_upgrade and # api::nip_fi modules, - # the auth metrics contract module, plus the exact NIP-FI tests added, + # the auth metrics contract module, the DB-free partition alert series + # tests (their Postgres case is #[ignore]d and runs in the PostgreSQL + # lane), plus the exact NIP-FI tests added, # or whose assertions changed, in mixed modules (audio::room, # connection, handlers::*, state). nip_fi_config and router are # selected whole by the command above. Mixed modules are listed by exact @@ -550,6 +553,7 @@ test-unit: cargo nextest run -p buzz-relay --lib -E ' test(/^nip_fi_upgrade::/) + test(/^metrics::contract_tests::/) + + test(/^metrics::partition_alert_tests::/) + test(=audio::room::tests::roster_revisions_are_ordered_and_snapshot_is_authoritative) + test(=connection::tests::auth_lifecycle_reconciles_every_terminal_and_never_leaks_gauge) + test(=audio::room::tests::b1_pending_peer_removed_before_commit_emits_no_delta) diff --git a/crates/buzz-admin/src/main.rs b/crates/buzz-admin/src/main.rs index 66780e2041..33fc74b6d1 100644 --- a/crates/buzz-admin/src/main.rs +++ b/crates/buzz-admin/src/main.rs @@ -31,7 +31,10 @@ use std::time::Instant; use anyhow::Result; use buzz_core::kind::KIND_NIP43_MEMBERSHIP_LIST; use buzz_core::tenant::{relay_url_authority, TenantContext}; -use buzz_db::{partition::PartitionAuditReport, Db, DbConfig}; +use buzz_db::{ + partition::{PartitionAuditOutcome, PartitionAuditReport}, + Db, DbConfig, +}; use buzz_media::{BucketSnapshot, MediaConfig, MediaStorage, S3AddressingStyle, SweepError}; use buzz_pubsub::{EventTopic, PubSubManager}; use clap::{Parser, Subcommand}; @@ -358,7 +361,7 @@ struct PartitionAuditOutput { source_sha: &'static str, build_id: &'static str, build_url: &'static str, - outcome: &'static str, + outcome: PartitionAuditOutcome, months_ahead: u32, identity: PartitionAuditIdentity, report: PartitionAuditReport, @@ -399,20 +402,8 @@ async fn cmd_partition_audit(months_ahead: u32) -> Result { let report = Db::from_pool(pool) .audit_partitions_report(months_ahead) .await; - let outcome = if !report.errors.is_empty() { - "error" - } else if !report.serving_safe() { - "unsafe" - } else if report.tables.iter().any(|table| table.degraded()) { - "degraded" - } else { - "ok" - }; - let code = match outcome { - "error" => 5, - "unsafe" => 2, - _ => 0, - }; + let outcome = report.outcome(); + let code = partition_audit_exit_code(outcome); println!( "{}", serde_json::to_string_pretty(&PartitionAuditOutput { @@ -430,6 +421,15 @@ async fn cmd_partition_audit(months_ahead: u32) -> Result { Ok(code) } +/// Exit nonzero only when the audit is incomplete or a table cannot serve. +fn partition_audit_exit_code(outcome: PartitionAuditOutcome) -> i32 { + match outcome { + PartitionAuditOutcome::Error => 5, + PartitionAuditOutcome::Unsafe => 2, + PartitionAuditOutcome::Degraded | PartitionAuditOutcome::Ok => 0, + } +} + async fn cmd_add_member(pubkey_arg: String, role: String) -> Result { if let Err(msg) = validate_role(&role) { eprintln!("error: {msg}"); @@ -1090,4 +1090,17 @@ mod tests { assert!(command.is_err()); } + + #[test] + fn partition_audit_outcomes_keep_their_json_names_and_exit_codes() { + for (outcome, name, code) in [ + (PartitionAuditOutcome::Ok, "ok", 0), + (PartitionAuditOutcome::Degraded, "degraded", 0), + (PartitionAuditOutcome::Unsafe, "unsafe", 2), + (PartitionAuditOutcome::Error, "error", 5), + ] { + assert_eq!(serde_json::to_value(outcome).unwrap(), name); + assert_eq!(partition_audit_exit_code(outcome), code, "{name}"); + } + } } diff --git a/crates/buzz-db/src/store/partition.rs b/crates/buzz-db/src/store/partition.rs index 02dfa4e708..8322313de3 100644 --- a/crates/buzz-db/src/store/partition.rs +++ b/crates/buzz-db/src/store/partition.rs @@ -186,7 +186,7 @@ pub struct PartitionTableAudit { pub children: Vec, /// Cached effective bounds for every routable leaf in the catalog tree. pub coverage_leaves: Vec, - /// Coverage of the current month and the requested future months. + /// Coverage of the current month, first, and the requested future months. pub months: Vec, /// Whether a parseable routable leaf covers the audit timestamp. pub serving_safe: bool, @@ -234,15 +234,18 @@ impl PartitionTableAudit { } /// Whether the table is serving but has state requiring operator attention. + /// + /// A legacy leaf that ends at or before the current month start is + /// historical: it stays in `children` but no longer degrades the table. pub fn degraded(&self) -> bool { !self.partition_key_valid || self.children.iter().any(|child| { matches!( child.kind, - PartitionChildKind::LegacyLeaf - | PartitionChildKind::Default - | PartitionChildKind::Anomalous - ) || !child.missing_triggers.is_empty() + PartitionChildKind::Default | PartitionChildKind::Anomalous + ) || (child.kind == PartitionChildKind::LegacyLeaf + && !self.ends_by_current_month(child)) + || !child.missing_triggers.is_empty() || !child.extra_triggers.is_empty() || child.catch_all_nonempty == Some(true) || child.default_nonempty == Some(true) @@ -253,6 +256,18 @@ impl PartitionTableAudit { .any(|month| month.kind != MonthCoverageKind::CoveredByMonthly) } + /// Whether `child` has a finite upper bound at or before the audited month start. + /// + /// Without an audited month, no child is treated as historical. + fn ends_by_current_month(&self, child: &PartitionChildAudit) -> bool { + self.months.first().is_some_and(|current_month| { + matches!( + &child.upper, + Some(PartitionBound::Finite(upper)) if *upper <= current_month.start + ) + }) + } + /// Whether the catalog is understood well enough to authorize automatic DDL. fn structurally_safe_for_creation(&self) -> bool { self.partition_key_valid @@ -297,6 +312,20 @@ pub struct PartitionAuditReport { pub errors: Vec, } +/// Operator-facing verdict for a best-effort catalog audit report. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum PartitionAuditOutcome { + /// Every managed parent is serving safely without operator attention. + Ok, + /// Every managed parent is serving safely, but one needs operator attention. + Degraded, + /// The report is complete, but a managed parent is not serving safely. + Unsafe, + /// At least one managed parent could not be audited. + Error, +} + impl PartitionAuditReport { /// Whether the report is complete and every managed parent is serving safely. pub fn serving_safe(&self) -> bool { @@ -305,6 +334,19 @@ impl PartitionAuditReport { && self.tables.iter().all(|table| table.serving_safe) } + /// Most severe verdict across audit errors, serving safety, and degradation. + pub fn outcome(&self) -> PartitionAuditOutcome { + if !self.errors.is_empty() { + PartitionAuditOutcome::Error + } else if !self.serving_safe() { + PartitionAuditOutcome::Unsafe + } else if self.tables.iter().any(PartitionTableAudit::degraded) { + PartitionAuditOutcome::Degraded + } else { + PartitionAuditOutcome::Ok + } + } + fn into_complete_audit(self) -> Result { if self.errors.is_empty() { Ok(PartitionAudit { @@ -421,6 +463,7 @@ async fn audit_partition_catalog_report_at_with_timeout( "outcome" => "error" ) .increment(1); + refresh_last_success_recency(table); metrics::histogram!( "buzz_partition_audit_duration_seconds", "table" => table @@ -438,6 +481,7 @@ async fn audit_partition_catalog_report_at_with_timeout( "outcome" => "error" ) .increment(1); + refresh_last_success_recency(table); metrics::histogram!( "buzz_partition_audit_duration_seconds", "table" => table @@ -923,17 +967,7 @@ async fn audit_table_on( .map(|leaf| leaf.audit) .collect::>(); - let mut months = Vec::with_capacity(months_ahead as usize + 1); - for offset in 0..=months_ahead { - let (year, month) = add_months(now.year(), now.month(), offset)?; - let start = month_start(year, month)?; - let (end_year, end_month) = add_months(year, month, 1)?; - let end = month_start(end_year, end_month)?; - months.push(MonthCoverage { - start, - kind: coverage_for_range(&coverage_leaves, &start, &end), - }); - } + let months = month_coverage(&coverage_leaves, months_ahead, now)?; let serving_safe = coverage_leaves .iter() @@ -951,6 +985,26 @@ async fn audit_table_on( }) } +/// Classify the month containing `now`, then each lookahead month, in order. +fn month_coverage( + coverage_leaves: &[PartitionLeafAudit], + months_ahead: i32, + now: DateTime, +) -> Result> { + let mut months = Vec::with_capacity(months_ahead as usize + 1); + for offset in 0..=months_ahead { + let (year, month) = add_months(now.year(), now.month(), offset)?; + let start = month_start(year, month)?; + let (end_year, end_month) = add_months(year, month, 1)?; + let end = month_start(end_year, end_month)?; + months.push(MonthCoverage { + start, + kind: coverage_for_range(coverage_leaves, &start, &end), + }); + } + Ok(months) +} + fn validated_months_ahead(months_ahead: u32) -> Result { if months_ahead > MAX_PARTITION_MONTHS_AHEAD { return Err(DbError::InvalidData(format!( @@ -987,6 +1041,35 @@ fn qualified_relation_name(schema: &str, relation: &str) -> String { ) } +/// Export every `buzz_partition_audit_runs_total` series at zero. +/// +/// Prometheus `increase()` ignores the first sample of a new series, so an +/// outcome first seen mid-run would otherwise be invisible to rate-based +/// alerts. Call once at process start, before the first audit. +pub fn initialize_audit_metric_series() { + for &table in PARTITIONED_TABLES { + for outcome in ["ok", "degraded", "error"] { + metrics::counter!( + "buzz_partition_audit_runs_total", + "table" => table, + "outcome" => outcome + ) + .increment(0); + } + } +} + +/// Keep a table's last success timestamp exported through failed audits +/// without changing it, so the exporter's idle timeout cannot evict it. A +/// table that never succeeded exports 0, which reads as maximally stale. +fn refresh_last_success_recency(table: &'static str) { + metrics::gauge!( + "buzz_partition_audit_last_success_timestamp_seconds", + "table" => table + ) + .increment(0.0); +} + fn emit_audit_metrics(audit: &PartitionTableAudit, duration_seconds: f64, now: DateTime) { let outcome = if audit.degraded() { "degraded" } else { "ok" }; let uncovered = audit @@ -1788,6 +1871,335 @@ mod tests { assert_eq!(extra, Some(1.0)); } + fn month_bound(year: i32, month: u32) -> PartitionBound { + PartitionBound::Finite(Utc.with_ymd_and_hms(year, month, 1, 0, 0, 0).unwrap()) + } + + fn child_audit( + name: &str, + kind: PartitionChildKind, + lower: PartitionBound, + upper: PartitionBound, + ) -> PartitionChildAudit { + PartitionChildAudit { + schema: "public".to_string(), + name: name.to_string(), + relation_kind: "r".to_string(), + catch_all_nonempty: (upper == PartitionBound::MaxValue).then_some(false), + lower: Some(lower), + upper: Some(upper), + kind, + pending_detach: false, + default_nonempty: None, + missing_triggers: Vec::new(), + extra_triggers: Vec::new(), + } + } + + /// The repaired incident catalog, classified and covered by production code at `now`. + fn repaired_incident_audit(table: &'static str, now: DateTime) -> PartitionTableAudit { + let mut bounds = vec![ + ( + format!("{table}_p_past"), + PartitionBound::MinValue, + month_bound(2026, 7), + ), + ( + format!("{table}_p2026_07_08_legacy"), + month_bound(2026, 7), + month_bound(2026, 9), + ), + ]; + for month in 9..=12 { + let (end_year, end_month) = add_months(2026, month, 1).unwrap(); + bounds.push(( + format!("{table}_p2026_{month:02}"), + month_bound(2026, month), + month_bound(end_year, end_month), + )); + } + bounds.push(( + format!("{table}_p_future_next"), + month_bound(2027, 1), + PartitionBound::MaxValue, + )); + let children = bounds + .iter() + .map(|(name, lower, upper)| { + let kind = classify_child("r", table, name, Some(lower), Some(upper)); + child_audit(name, kind, lower.clone(), upper.clone()) + }) + .collect::>(); + let coverage_leaves = bounds + .into_iter() + .map(|(name, lower, upper)| PartitionLeafAudit { + schema: "public".to_string(), + root_child_schema: "public".to_string(), + root_child: name.clone(), + name, + catch_all_nonempty: (upper == PartitionBound::MaxValue).then_some(false), + lower, + upper, + nested: false, + is_default: false, + default_routed: false, + routing_exclusions: Vec::new(), + default_nonempty: None, + }) + .collect::>(); + let months = month_coverage(&coverage_leaves, 3, now).expect("month coverage"); + let serving_safe = coverage_leaves + .iter() + .any(|leaf| leaf_covers_timestamp(leaf, &now)); + PartitionTableAudit { + table, + partition_key: expected_partition_key(table).map(str::to_string), + expected_partition_key: expected_partition_key(table).unwrap(), + partition_key_valid: true, + children, + coverage_leaves, + months, + serving_safe, + } + } + + #[test] + fn repaired_incident_layout_is_historical_once_its_months_have_passed() { + for (now, degraded, outcome) in [ + ( + Utc.with_ymd_and_hms(2026, 8, 15, 12, 0, 0).unwrap(), + true, + PartitionAuditOutcome::Degraded, + ), + ( + // The legacy upper bound equals the current month start. + Utc.with_ymd_and_hms(2026, 9, 1, 0, 0, 0).unwrap(), + false, + PartitionAuditOutcome::Ok, + ), + ( + Utc.with_ymd_and_hms(2026, 9, 15, 12, 0, 0).unwrap(), + false, + PartitionAuditOutcome::Ok, + ), + ] { + let report = PartitionAuditReport { + audited_at: now, + tables: PARTITIONED_TABLES + .iter() + .map(|table| repaired_incident_audit(table, now)) + .collect(), + errors: Vec::new(), + }; + assert!(report.serving_safe(), "{now}"); + assert_eq!(report.outcome(), outcome, "{now}"); + for table in &report.tables { + assert_eq!(table.degraded(), degraded, "{} at {now}", table.table); + assert!(table + .months + .iter() + .all(|month| month.kind == MonthCoverageKind::CoveredByMonthly)); + // Historical leaves stay visible to operators. + assert_eq!( + table + .children + .iter() + .filter(|child| child.kind == PartitionChildKind::LegacyLeaf) + .map(|child| child.name.as_str()) + .collect::>(), + [format!("{}_p2026_07_08_legacy", table.table)] + ); + } + let json = serde_json::to_value(&report).expect("serialize report"); + assert!(json["tables"][0]["children"] + .as_array() + .expect("children") + .iter() + .any(|child| child["name"] == "events_p2026_07_08_legacy" + && child["kind"] == "legacy_leaf")); + + let recorder = metrics_util::debugging::DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + metrics::with_local_recorder(&recorder, || { + emit_audit_metrics(&report.tables[0], 0.01, now) + }); + let run_outcome = snapshotter + .snapshot() + .into_vec() + .into_iter() + .find_map(|(key, _, _, _)| { + (key.key().name() == "buzz_partition_audit_runs_total").then(|| { + key.key() + .labels() + .find(|label| label.key() == "outcome") + .map(|label| label.value().to_string()) + }) + }) + .flatten(); + let expected = if degraded { "degraded" } else { "ok" }; + assert_eq!(run_outcome.as_deref(), Some(expected), "{now}"); + } + } + + #[test] + fn report_outcome_ranks_errors_then_unsafe_then_degraded() { + let august = Utc.with_ymd_and_hms(2026, 8, 15, 12, 0, 0).unwrap(); + let degraded_tables = || { + PARTITIONED_TABLES + .iter() + .map(|table| repaired_incident_audit(table, august)) + .collect::>() + }; + let mut incomplete = degraded_tables(); + incomplete.pop(); + let error = PartitionAuditError { + table: "delivery_log", + error: "statement timeout".to_string(), + }; + for (tables, errors, outcome) in [ + ( + degraded_tables(), + Vec::new(), + PartitionAuditOutcome::Degraded, + ), + ( + incomplete.clone(), + Vec::new(), + PartitionAuditOutcome::Unsafe, + ), + (incomplete, vec![error], PartitionAuditOutcome::Error), + ] { + let report = PartitionAuditReport { + audited_at: august, + tables, + errors, + }; + assert_eq!(report.outcome(), outcome); + } + } + + #[test] + fn legacy_leaf_degrades_until_its_upper_bound_reaches_the_current_month() { + let now = Utc.with_ymd_and_hms(2026, 9, 15, 12, 0, 0).unwrap(); + let september = Utc.with_ymd_and_hms(2026, 9, 1, 0, 0, 0).unwrap(); + let historical = || { + child_audit( + "events_legacy", + PartitionChildKind::LegacyLeaf, + month_bound(2026, 6), + month_bound(2026, 8), + ) + }; + let with = |child: PartitionChildAudit| vec![child]; + let mut missing_trigger = historical(); + missing_trigger.missing_triggers = vec!["partition_probe".to_string()]; + let mut extra_trigger = historical(); + extra_trigger.extra_triggers = vec!["child_only_probe".to_string()]; + let mut anomalous = historical(); + anomalous.kind = PartitionChildKind::Anomalous; + let mut default = historical(); + default.kind = PartitionChildKind::Default; + let cases = [ + ( + "ends before the current month", + with(historical()), + 3, + false, + ), + ( + "ends exactly at the current month start", + with(child_audit( + "events_legacy", + PartitionChildKind::LegacyLeaf, + month_bound(2026, 7), + month_bound(2026, 9), + )), + 3, + false, + ), + ( + "ends one second into the current month", + with(child_audit( + "events_legacy", + PartitionChildKind::LegacyLeaf, + month_bound(2026, 7), + PartitionBound::Finite(september + chrono::Duration::seconds(1)), + )), + 3, + true, + ), + ( + "covers lookahead months", + with(child_audit( + "events_legacy", + PartitionChildKind::LegacyLeaf, + month_bound(2026, 10), + month_bound(2026, 12), + )), + 3, + true, + ), + ( + "starts beyond the lookahead", + with(child_audit( + "events_legacy", + PartitionChildKind::LegacyLeaf, + month_bound(2027, 6), + month_bound(2027, 8), + )), + 3, + true, + ), + ( + "historical but missing a trigger", + with(missing_trigger), + 3, + true, + ), + ( + "historical but has an extra trigger", + with(extra_trigger), + 3, + true, + ), + ( + "historical bounds on an anomalous child", + with(anomalous), + 3, + true, + ), + ( + "historical bounds on a default child", + with(default), + 3, + true, + ), + ("no audited month to compare", with(historical()), -1, true), + ]; + for (case, children, months_ahead, degraded) in cases { + let months = (0..=months_ahead) + .map(|offset| { + let (year, month) = add_months(2026, 9, offset).unwrap(); + MonthCoverage { + start: month_start(year, month).unwrap(), + kind: MonthCoverageKind::CoveredByMonthly, + } + }) + .collect(); + let audit = PartitionTableAudit { + table: "events", + partition_key: Some("RANGE (created_at)".to_string()), + expected_partition_key: "RANGE (created_at)", + partition_key_valid: true, + children, + coverage_leaves: Vec::new(), + months, + serving_safe: true, + }; + assert_eq!(audit.degraded(), degraded, "{case} at {now}"); + } + } + pub(super) mod postgres_tests { use sqlx::postgres::PgPoolOptions; use uuid::Uuid; @@ -2480,7 +2892,20 @@ mod tests { .months .iter() .all(|month| month.kind == MonthCoverageKind::CoveredByMonthly)); + assert!( + !table.degraded(), + "{} is historical in September", + table.table + ); } + + let september = audit_partition_catalog_report_at(&pool, 3, fixed_now()).await; + assert_eq!(september.outcome(), PartitionAuditOutcome::Ok); + let august = Utc.with_ymd_and_hms(2026, 8, 15, 12, 0, 0).unwrap(); + let august = audit_partition_catalog_report_at(&pool, 3, august).await; + assert!(august.serving_safe()); + assert!(august.tables.iter().all(PartitionTableAudit::degraded)); + assert_eq!(august.outcome(), PartitionAuditOutcome::Degraded); assert_eq!(catalog_snapshot(&pool).await, before); drop_schema(&admin, &schema).await; } diff --git a/crates/buzz-relay/src/config.rs b/crates/buzz-relay/src/config.rs index 44662bb9e1..d5c3cf60a2 100644 --- a/crates/buzz-relay/src/config.rs +++ b/crates/buzz-relay/src/config.rs @@ -2022,7 +2022,7 @@ mod tests { #[test] fn partition_manager_config_has_bounded_interval_and_create_kill_switch() { - let _guard = ENV_MUTEX.lock().unwrap(); + let _guards = env_guards(); let previous_interval = std::env::var_os("BUZZ_PARTITION_AUDIT_INTERVAL_SECS"); let previous_create = std::env::var_os("BUZZ_PARTITION_MANAGER_CREATE_ENABLED"); @@ -2054,7 +2054,7 @@ mod tests { #[test] fn partition_manager_create_kill_switch_parses_false_values_strictly() { - let _guard = ENV_MUTEX.lock().unwrap(); + let _guards = env_guards(); let previous = std::env::var_os("BUZZ_PARTITION_MANAGER_CREATE_ENABLED"); for value in ["FALSE", "off", " false "] { diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index e42c6a4975..4ed127dfff 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -12,7 +12,10 @@ use uuid::Uuid; use buzz_audit::AuditService; use buzz_auth::AuthService; use buzz_core::CommunityId; -use buzz_db::{partition::PARTITION_MANAGER_MONTHS_AHEAD, Db, DbConfig}; +use buzz_db::{ + partition::{PartitionAudit, PARTITION_MANAGER_MONTHS_AHEAD}, + Db, DbConfig, +}; use buzz_pubsub::PubSubManager; use buzz_search::SearchService; @@ -249,6 +252,49 @@ impl FleetUsageSchedule { const USAGE_METRICS_LOCK_KEY: i64 = 0x4255_5A5A_4D45_5452; +/// Run one explicit partition catalog audit attempt. +/// +/// `buzz_partition_audit_failures_total` counts failed explicit attempts: the +/// startup fallback audit and each periodic audit, or the periodic fallback +/// audit when advancement runs periodic maintenance. The audit inside +/// `maintain_partitions` is not an explicit attempt; its failure is counted +/// only through the fallback that follows it. +async fn counted_partition_audit( + attempt: impl std::future::Future>, +) -> buzz_db::Result { + let result = attempt.await; + if result.is_err() { + metrics::counter!("buzz_partition_audit_failures_total").increment(1); + } + result +} + +/// Maintain startup partitions, falling back to a read-only audit on failure. +/// +/// A maintenance failure alone never counts as an audit failure; a failed +/// fallback audit counts once. +async fn startup_partition_audit( + maintain: impl std::future::Future>, + fallback_audit: impl FnOnce() -> F, +) -> Option +where + F: std::future::Future>, +{ + match maintain.await { + Ok(audit) => Some(audit), + Err(error) => { + error!(%error, "Failed to ensure partitions"); + match counted_partition_audit(fallback_audit()).await { + Ok(audit) => Some(audit), + Err(error) => { + error!(%error, "Initial partition catalog audit failed"); + None + } + } + } + } +} + /// Retry missing diagnostics promptly, without tying audit recovery to readiness. /// Delays start after each attempt finishes, so slow audits cannot cause bursts. struct PartitionAuditSchedule { @@ -486,23 +532,15 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> { advance_enabled: config.partition_manager_advance_enabled, }; let mut step = StepTimer::start(StartupStep::PartitionEnsure); - let startup_partition_audit = match db - .maintain_partitions(PARTITION_MANAGER_MONTHS_AHEAD, partition_policy) - .await - { - Ok(audit) => Some(audit), - Err(error) => { - step.degrade(); - error!(%error, "Failed to ensure partitions"); - match db.audit_partitions(PARTITION_MANAGER_MONTHS_AHEAD).await { - Ok(audit) => Some(audit), - Err(error) => { - error!(%error, "Initial partition catalog audit failed"); - None - } - } - } - }; + let startup_partition_audit = startup_partition_audit( + async { + db.maintain_partitions(PARTITION_MANAGER_MONTHS_AHEAD, partition_policy) + .await + .inspect_err(|_| step.degrade()) + }, + || db.audit_partitions(PARTITION_MANAGER_MONTHS_AHEAD), + ) + .await; step.finish(); let step = StepTimer::start(StartupStep::DeletionFenceVerify); @@ -837,19 +875,20 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> { Ok(audit) => Ok(audit), Err(error) => { warn!(%error, "Periodic partition maintenance failed"); - db.audit_partitions(PARTITION_MANAGER_MONTHS_AHEAD).await + counted_partition_audit( + db.audit_partitions(PARTITION_MANAGER_MONTHS_AHEAD), + ) + .await } } } else { - db.audit_partitions(PARTITION_MANAGER_MONTHS_AHEAD).await + counted_partition_audit(db.audit_partitions(PARTITION_MANAGER_MONTHS_AHEAD)) + .await }; schedule.record_attempt(result.is_ok()); match result { Ok(audit) => partition_state.record_partition_audit(audit), - Err(error) => { - metrics::counter!("buzz_partition_audit_failures_total").increment(1); - warn!(%error, "Periodic partition catalog audit failed") - } + Err(error) => warn!(%error, "Periodic partition catalog audit failed"), } } }); @@ -3030,12 +3069,12 @@ mod tests { use super::{ apply_activity_refresh, apply_stock_refresh, buzz_auto_migrate_enabled, connect_audit_pool, - dropped_in_memory_keys, emit_cached_fleet_usage_metrics, idle_timeout_secs, - jwks_next_retry_after_failed_refresh, nip_fi_jwks_refresh_loop, + counted_partition_audit, dropped_in_memory_keys, emit_cached_fleet_usage_metrics, + idle_timeout_secs, jwks_next_retry_after_failed_refresh, nip_fi_jwks_refresh_loop, record_legacy_usage_collection, refresh_due_fleet_usage, refresh_legacy_active_gauge_recency, relay_keypair_from_config, - run_jwks_refresh_supervisor, run_periodic_until_cancelled, EmissionScope, - FleetUsageSchedule, InMemoryMetricKey, PartitionAuditSchedule, + run_jwks_refresh_supervisor, run_periodic_until_cancelled, startup_partition_audit, + EmissionScope, FleetUsageSchedule, InMemoryMetricKey, PartitionAuditSchedule, FLEET_ACTIVITY_COLLECTION_DEADLINE, FLEET_STOCK_COLLECTION_DEADLINE, }; use buzz_db::DbConfig; @@ -3204,6 +3243,140 @@ mod tests { ); } + fn partition_audit_ok() -> buzz_db::Result { + Ok(buzz_db::partition::PartitionAudit { + audited_at: chrono::Utc::now(), + tables: Vec::new(), + }) + } + + fn partition_audit_err(detail: &str) -> buzz_db::Result { + Err(buzz_db::DbError::InvalidData(detail.to_string())) + } + + /// One recorded counter: name, labels, and value. + type CounterSample = (String, Vec<(String, String)>, u64); + + /// Drive `future` under a local recorder and return its counter values. + fn with_partition_counters( + future: F, + ) -> (F::Output, Vec) { + let recorder = metrics_util::debugging::DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + let output = metrics::with_local_recorder(&recorder, || { + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("current-thread runtime") + .block_on(future) + }); + let counters = snapshotter + .snapshot() + .into_vec() + .into_iter() + .filter_map(|(key, _, _, value)| match value { + DebugValue::Counter(value) => Some(( + key.key().name().to_string(), + key.key() + .labels() + .map(|label| (label.key().to_string(), label.value().to_string())) + .collect(), + value, + )), + _ => None, + }) + .collect(); + (output, counters) + } + + fn aggregate_audit_failures(counters: &[CounterSample]) -> u64 { + counters + .iter() + .filter(|(name, _, _)| name == "buzz_partition_audit_failures_total") + .map(|(_, _, value)| value) + .sum() + } + + #[test] + fn startup_partition_audit_counts_only_a_failed_fallback_once() { + let (audit, counters) = with_partition_counters(startup_partition_audit( + async { partition_audit_err("partition creation failed") }, + || async { partition_audit_err("catalog audit failed") }, + )); + assert!(audit.is_none()); + assert_eq!(aggregate_audit_failures(&counters), 1); + + // A DDL failure recovered by the fallback audit is not an audit failure. + let (audit, counters) = with_partition_counters(startup_partition_audit( + async { partition_audit_err("partition creation failed") }, + || async { partition_audit_ok() }, + )); + assert!(audit.is_some()); + assert_eq!(aggregate_audit_failures(&counters), 0); + + let (audit, counters) = with_partition_counters(startup_partition_audit( + async { partition_audit_ok() }, + || async { panic!("successful ensure must not run the fallback audit") }, + )); + assert!(audit.is_some()); + assert_eq!(aggregate_audit_failures(&counters), 0); + } + + #[test] + fn periodic_partition_audit_counts_each_failed_attempt() { + let (result, counters) = + with_partition_counters(counted_partition_audit(async { partition_audit_ok() })); + assert!(result.is_ok()); + assert_eq!(aggregate_audit_failures(&counters), 0); + + let ((), counters) = with_partition_counters(async { + for _ in 0..2 { + assert!( + counted_partition_audit(async { partition_audit_err("timeout") }) + .await + .is_err() + ); + } + }); + assert_eq!(aggregate_audit_failures(&counters), 2); + } + + #[test] + fn unreachable_database_startup_counts_one_aggregate_audit_failure() { + let (audit, counters) = with_partition_counters(async { + let pool = sqlx::postgres::PgPoolOptions::new() + .acquire_timeout(Duration::from_millis(200)) + .connect_lazy("postgres://127.0.0.1:1/buzz") + .expect("lazy test pool"); + let db = buzz_db::Db::from_pool(pool); + let months_ahead = buzz_db::partition::PARTITION_MANAGER_MONTHS_AHEAD; + let policy = buzz_db::partition::PartitionMaintenancePolicy { + create_enabled: true, + advance_enabled: false, + }; + startup_partition_audit(db.maintain_partitions(months_ahead, policy), || { + db.audit_partitions(months_ahead) + }) + .await + }); + assert!(audit.is_none()); + assert_eq!(aggregate_audit_failures(&counters), 1); + // Each per-table attempt, including the one inside maintenance, still has + // its own error outcome. + for table in ["events", "delivery_log"] { + let errors: u64 = counters + .iter() + .filter(|(name, labels, _)| { + name == "buzz_partition_audit_runs_total" + && labels.contains(&("table".to_string(), table.to_string())) + && labels.contains(&("outcome".to_string(), "error".to_string())) + }) + .map(|(_, _, value)| value) + .sum(); + assert_eq!(errors, 2, "{table}"); + } + } + #[test] fn buzz_auto_migrate_is_opt_in() { assert!(!buzz_auto_migrate_enabled(None)); diff --git a/crates/buzz-relay/src/metrics.rs b/crates/buzz-relay/src/metrics.rs index 3566c0c019..67e4d3b35e 100644 --- a/crates/buzz-relay/src/metrics.rs +++ b/crates/buzz-relay/src/metrics.rs @@ -299,6 +299,7 @@ pub fn try_install(port: u16, gauge_idle_timeout_secs: u64) -> Result<(), Metric describe_auth_metrics(); crate::startup_steps::describe_metrics(); initialize_auth_metric_series(); + initialize_partition_audit_metric_series(); tokio::spawn(exporter); Ok(()) } @@ -522,6 +523,13 @@ pub(crate) fn initialize_auth_metric_series() { metrics::gauge!("buzz_ws_authenticated_connections_active").set(0.0); } +/// Export the partition audit counters at zero before the first audit, so +/// `increase()`-based alerts count the first failure and every new outcome. +pub(crate) fn initialize_partition_audit_metric_series() { + buzz_db::partition::initialize_audit_metric_series(); + metrics::counter!("buzz_partition_audit_failures_total").increment(0); +} + /// Record one issued NIP-42 challenge after it enters the connection writer. /// /// Before this lifecycle contract, `buzz_auth_attempts_total` incremented only @@ -1161,3 +1169,167 @@ mod tests { ); } } + +/// Fixtures for the partition audit alerts in +/// `docs/partition-catalog-monitoring.md`: the production exporter configuration +/// and audit metric emission export the series those rules read. +#[cfg(test)] +mod partition_alert_tests { + use std::collections::BTreeMap; + use std::time::Duration; + + const RUNS: &str = "buzz_partition_audit_runs_total"; + const LAST_SUCCESS: &str = "buzz_partition_audit_last_success_timestamp_seconds"; + const MANAGED_TABLES: [&str; 2] = ["delivery_log", "events"]; + + /// Sample value by `table` label for every sample of `metric` in a scrape. + /// Counters with several `outcome` labels keep the last sample per table. + fn samples(scrape: &str, metric: &str) -> BTreeMap { + scrape + .lines() + .filter_map(|line| line.strip_prefix(metric)?.strip_prefix('{')) + .filter_map(|labels| { + let (_, rest) = labels.split_once(r#"table=""#)?; + let (table, rest) = rest.split_once('"')?; + let (_, value) = rest.rsplit_once(' ')?; + Some((table.to_string(), value.parse().expect("numeric sample"))) + }) + .collect() + } + + fn tables(scrape: &str, metric: &str) -> Vec { + samples(scrape, metric).into_keys().collect() + } + + fn managed_tables() -> Vec { + MANAGED_TABLES + .iter() + .map(|table| table.to_string()) + .collect() + } + + fn unreachable_db() -> buzz_db::Db { + let pool = sqlx::postgres::PgPoolOptions::new() + .acquire_timeout(Duration::from_millis(200)) + .connect_lazy("postgres://127.0.0.1:1/buzz") + .expect("lazy test pool"); + buzz_db::Db::from_pool(pool) + } + + fn current_thread_runtime() -> tokio::runtime::Runtime { + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("current-thread runtime") + } + + #[test] + fn never_successful_audits_export_a_zero_success_timestamp() { + let (recorder, handle) = super::readiness_test_recorder_with_idle_timeout(1); + metrics::with_local_recorder(&recorder, || { + current_thread_runtime().block_on(async { + assert!(unreachable_db().audit_partitions(3).await.is_err()); + }); + }); + + let scrape = handle.render(); + assert_eq!(tables(&scrape, RUNS), managed_tables(), "{scrape}"); + let zero = MANAGED_TABLES + .iter() + .map(|table| (table.to_string(), 0.0)) + .collect::>(); + assert_eq!(samples(&scrape, LAST_SUCCESS), zero, "{scrape}"); + } + + #[test] + fn audit_counters_export_zero_baselines_before_the_first_audit() { + let (recorder, handle) = super::readiness_test_recorder_with_idle_timeout(1); + metrics::with_local_recorder(&recorder, super::initialize_partition_audit_metric_series); + + let scrape = handle.render(); + let zero_runs = scrape + .lines() + .filter(|line| line.starts_with(&format!("{RUNS}{{")) && line.ends_with(" 0")) + .count(); + assert_eq!(zero_runs, MANAGED_TABLES.len() * 3, "{scrape}"); + assert!( + scrape + .lines() + .any(|line| line == "buzz_partition_audit_failures_total 0"), + "{scrape}" + ); + + // A first failure then reads as an increase from the zero baseline. + metrics::with_local_recorder(&recorder, || { + current_thread_runtime().block_on(async { + assert!(unreachable_db().audit_partitions(3).await.is_err()); + }); + }); + let scrape = handle.render(); + for table in MANAGED_TABLES { + for (outcome, value) in [("ok", 0), ("degraded", 0), ("error", 1)] { + let line = scrape + .lines() + .find(|line| { + line.starts_with(&format!("{RUNS}{{")) + && line.contains(&format!(r#"table="{table}""#)) + && line.contains(&format!(r#"outcome="{outcome}""#)) + }) + .unwrap_or_else(|| panic!("{table}/{outcome} missing:\n{scrape}")); + assert!(line.ends_with(&format!(" {value}")), "{line}"); + } + } + } + + mod postgres_tests { + use super::*; + + #[test] + #[ignore = "requires Postgres"] + fn failed_audits_keep_the_last_success_past_the_idle_timeout() { + let idle_timeout = Duration::from_secs(1); + let (recorder, handle) = + crate::metrics::readiness_test_recorder_with_idle_timeout(idle_timeout.as_secs()); + let runtime = current_thread_runtime(); + let (working, failing) = metrics::with_local_recorder(&recorder, || { + runtime.block_on(async { + let pool = sqlx::PgPool::connect(&crate::test_support::database_url()) + .await + .expect("connect test database"); + let working = buzz_db::Db::from_pool(pool); + working.audit_partitions(3).await.expect("successful audit"); + (working, unreachable_db()) + }) + }); + + let scrape = handle.render(); + let first = samples(&scrape, LAST_SUCCESS); + assert_eq!(first.keys().cloned().collect::>(), managed_tables()); + + let audit = |db: &buzz_db::Db| { + metrics::with_local_recorder(&recorder, || { + runtime.block_on(async { db.audit_partitions(3).await.is_ok() }) + }) + }; + // A failed audit never makes the last success look fresh, and + // sustained failures keep it exported past the idle timeout, so its + // age keeps rising instead of the series disappearing. + assert!(!audit(&failing)); + assert_eq!(samples(&handle.render(), LAST_SUCCESS), first); + std::thread::sleep(idle_timeout + Duration::from_millis(250)); + assert!(!audit(&failing)); + assert_eq!(samples(&handle.render(), LAST_SUCCESS), first); + + // When audits stop running altogether the timestamp is evicted, + // but the run counters remain for the not-running rule. + std::thread::sleep(idle_timeout + Duration::from_millis(250)); + let scrape = handle.render(); + assert!(tables(&scrape, LAST_SUCCESS).is_empty(), "{scrape}"); + assert_eq!(tables(&scrape, RUNS), managed_tables(), "{scrape}"); + + // The next success exports a fresh timestamp again. + assert!(audit(&working)); + assert_eq!(tables(&handle.render(), LAST_SUCCESS), managed_tables()); + } + } +} diff --git a/crates/buzz-relay/src/state.rs b/crates/buzz-relay/src/state.rs index e762095ca3..7e015a5dc8 100644 --- a/crates/buzz-relay/src/state.rs +++ b/crates/buzz-relay/src/state.rs @@ -1649,9 +1649,11 @@ impl AppState { /// Snapshot the last completed audit without accessing the database. /// - /// Periodic refresh failures retain the last-known-good audit by design; - /// operators should alert on staleness of - /// `buzz_partition_audit_last_success_timestamp_seconds`. + /// Periodic refresh failures retain the last-known-good audit by design. + /// Alert on the age of + /// `buzz_partition_audit_last_success_timestamp_seconds`, which failed + /// audits keep exported, and on audit runs stopping. See + /// `docs/partition-catalog-monitoring.md`. pub fn partition_audit_snapshot(&self) -> Option { let cached = match self.partition_audit.read() { Ok(cached) => cached, diff --git a/docs/partition-catalog-monitoring.md b/docs/partition-catalog-monitoring.md new file mode 100644 index 0000000000..8371403d93 --- /dev/null +++ b/docs/partition-catalog-monitoring.md @@ -0,0 +1,150 @@ +# Partition catalog monitoring + +The relay audits the `events` and `delivery_log` partition catalogs at startup +and every `BUZZ_PARTITION_AUDIT_INTERVAL_SECS` (default 900 seconds). Results are +diagnostic only. They appear in `/_status.partition_catalog` and in the metrics +below, and they never change `/_readiness` or `/_liveness`. Alert on them; do not +gate traffic on them. + +## Outcomes + +`buzz-admin partition-audit` prints one outcome for the whole catalog. + +| Outcome | Meaning | `buzz-admin` exit code | +|---|---|---| +| `ok` | Every managed table covers now and needs no operator attention. | 0 | +| `degraded` | Every table covers now, but one needs attention. | 0 | +| `unsafe` | A table cannot route a row timestamped now. | 2 | +| `error` | At least one table could not be audited. | 5 | + +The relay's `buzz_partition_audit_runs_total{outcome}` is per table and is only +`ok`, `degraded`, or `error`. There is no `unsafe` outcome there: a table that +cannot route a row timestamped now is `degraded` and has +`buzz_partition_serving_safe` 0. + +A table is `degraded` when it has a partition-key mismatch, a `DEFAULT` or +anomalous child, trigger drift, an occupied catch-all or `DEFAULT` leaf, or a +current or lookahead month not covered by a bounded leaf. A bounded legacy leaf +with a non-canonical name degrades the table while it covers the current month or +later. Once its upper bound is at or before the start of the current month, it is +historical. It remains in the per-child report as `legacy_leaf` but no longer +degrades the table. + +## Metrics + +Per-table gauges are set only by a successful audit of that table: +`buzz_partition_audit_last_success_timestamp_seconds`, +`buzz_partition_serving_safe`, `buzz_partition_uncovered_months`, +`buzz_partition_catch_all_covered_months`, `buzz_partition_default_covered_months`, +`buzz_partition_anomalous_children`, `buzz_partition_catch_all_nonempty`, +`buzz_partition_default_nonempty`, `buzz_partition_trigger_parity_missing`, and +`buzz_partition_trigger_parity_extra`. A failed audit leaves them unchanged, so a +failure never makes the last success look fresh. + +The exporter removes a gauge that has not been updated within its idle timeout. +The default is 2,700 seconds: three times the larger of the usage-metrics and +partition-audit intervals. A failed audit of a table refreshes that table's +`buzz_partition_audit_last_success_timestamp_seconds` without changing its +value, so sustained failures keep the last success exported and its age keeps +rising. A table that has never been audited successfully exports `0`. The +timestamp disappears only when audits stop running for longer than the idle +timeout; the other per-table gauges can be evicted during sustained failures. + +Counters are not idle-evicted while the process runs. The relay exports every +run outcome for both managed tables, and the failure counter, at 0 when its +metrics exporter starts, so `increase()` counts the first failure and the first +run with a new outcome: + +- `buzz_partition_audit_runs_total{table, outcome}` increments once per table + per audit. The outcome is `ok`, `degraded`, or `error`, including the opening + audit of each partition maintenance pass. +- `buzz_partition_audit_failures_total` counts failed explicit audit attempts: + the fallback audit that runs when a maintenance pass fails, and each periodic + audit when maintenance does not run. A maintenance failure is not counted by + itself. If the fallback audit then fails, that attempt counts once. +- `buzz_partition_create_attempts_total{table, outcome}` counts per-month + decisions in each maintenance pass: `created`, `error`, or `skipped_covered` + for a month that was already covered. Maintenance runs at startup, and on + every periodic audit when `BUZZ_PARTITION_MANAGER_ADVANCE_ENABLED` is on, so + `skipped_covered` then grows on every pass in steady state. + +## Alerts + +Use all three rules below. Aggregate with `without (...)`, not `by (...)`, so +cluster, account, environment, namespace, and pod labels stay in the alert. + +When a pod terminates, its scrape target disappears and its series go stale. +Instant selectors drop a stale series at once, but range selectors such as +`[30m]` keep reading its earlier samples until they leave the window. Each +range-based rule therefore also requires the counter to be exported now, which +it is for the whole life of a relay, so a terminated pod cannot keep or start an +alert. + +`BuzzPartitionAuditNotRunning` also requires the run counter to have existed one +window ago. After a successful startup audit, the first periodic audit runs a +full interval later, so a new pod's counter stays flat until then and would +otherwise read as a stopped loop. + +```yaml +# The last success is older than two default audit intervals. This also fires +# for a table that has never audited successfully, whose timestamp is 0, and +# keeps firing through sustained failures. Scale 1800 with +# BUZZ_PARTITION_AUDIT_INTERVAL_SECS. +- alert: BuzzPartitionAuditStale + expr: time() - buzz_partition_audit_last_success_timestamp_seconds > 1800 + for: 10m + +# The relay is running but its audit loop has stopped. Use a window of about +# two audit intervals, and keep the window plus `for` below the gauge idle +# timeout (2,700 s by default), so this fires before the stopped loop's +# timestamp is evicted and the stale rule resolves. Use the same duration for +# the window and the offset. A loop that stops in a pod's first window is +# caught once the pod is a window old, at about 35 minutes. +- alert: BuzzPartitionAuditNotRunning + expr: | + sum without (outcome) (increase(buzz_partition_audit_runs_total[30m])) == 0 + and sum without (outcome) (buzz_partition_audit_runs_total) + and sum without (outcome) (buzz_partition_audit_runs_total offset 30m) + for: 5m + +# Explicit audit attempts keep failing, even if a later attempt succeeds. +- alert: BuzzPartitionAuditFailing + expr: | + increase(buzz_partition_audit_failures_total[1h]) >= 2 + and buzz_partition_audit_failures_total +``` + +The counters exist from relay start, and the success timestamp exists after the +first audit attempt, whether or not it succeeded. The startup audit runs within +seconds of the zero baselines, usually before the first scrape, so +`BuzzPartitionAuditFailing` can miss a failure in that one attempt; if it does +not recover, `BuzzPartitionAuditStale` fires. The run counters are exported +before the relay connects to the database, so a live relay that is stuck before +its first audit shows flat zero counters, which `BuzzPartitionAuditNotRunning` +catches. + +`degraded` asks for operator attention; it does not mean rows cannot be +written. Historical legacy leaves no longer count, so a repaired catalog +returns to `ok` once the current month passes the repaired range and every +current and lookahead month is a bounded monthly partition. Months still covered +by a catch-all stay `degraded` until advancement moves the catch-all. Coverage +problems show directly in `buzz_partition_serving_safe` and +`buzz_partition_uncovered_months`. + +The tests in `crates/buzz-relay/src/metrics.rs` (`partition_alert_tests`) check +that the production exporter and audit emit the series these rules read: zero +counter baselines before the first audit, a `0` timestamp before any success, +the last success kept through failures past the idle timeout, and run counters +that outlive an evicted timestamp. They do not +evaluate the rules, their `for` durations, or label matching. +`partition-catalog-monitoring.rules.test.yml` evaluates the rules in +`partition-catalog-monitoring.rules.yml`, including their `for` durations, +against a fresh pod, a stopped loop, a terminated pod, a table that never +succeeded, and repeated failures. CI does not run it. Keep the rules file +identical to the block above, and run it after changing either: + +```bash +promtool test rules docs/partition-catalog-monitoring.rules.test.yml +``` + +Then validate label matching against your own alerting configuration. diff --git a/docs/partition-catalog-monitoring.rules.test.yml b/docs/partition-catalog-monitoring.rules.test.yml new file mode 100644 index 0000000000..3a897c5b61 --- /dev/null +++ b/docs/partition-catalog-monitoring.rules.test.yml @@ -0,0 +1,122 @@ +# Run with: promtool test rules docs/partition-catalog-monitoring.rules.test.yml +# +# Series are sampled every minute and use the default 900 s audit interval. The +# startup audit finishes before the first scrape, so run counters start at 1. +rule_files: + - partition-catalog-monitoring.rules.yml + +evaluation_interval: 1m + +tests: + # A healthy new pod audits at startup and then every 15 minutes. Its counter + # is flat for the first interval, which must not read as a stopped loop. + - interval: 1m + input_series: + - series: 'buzz_partition_audit_runs_total{pod="fresh",table="events",outcome="ok"}' + values: '1x14 2x14 3x14 4x14 5x14' + - series: 'buzz_partition_audit_runs_total{pod="fresh",table="events",outcome="degraded"}' + values: '0x74' + - series: 'buzz_partition_audit_last_success_timestamp_seconds{pod="fresh",table="events"}' + values: '0x14 900x14 1800x14 2700x14 3600x14' + alert_rule_test: + - eval_time: 10m + alertname: BuzzPartitionAuditNotRunning + exp_alerts: [] + - eval_time: 14m + alertname: BuzzPartitionAuditNotRunning + exp_alerts: [] + - eval_time: 29m + alertname: BuzzPartitionAuditNotRunning + exp_alerts: [] + - eval_time: 70m + alertname: BuzzPartitionAuditNotRunning + exp_alerts: [] + - eval_time: 70m + alertname: BuzzPartitionAuditStale + exp_alerts: [] + + # The audit loop stops after the 15-minute audit while the pod keeps running. + # Both rules fire before the timestamp's 2,700 s idle eviction at 60 minutes. + - interval: 1m + input_series: + - series: 'buzz_partition_audit_runs_total{pod="stopped",table="events",outcome="ok"}' + values: '1x14 2x60' + - series: 'buzz_partition_audit_last_success_timestamp_seconds{pod="stopped",table="events"}' + values: '0x14 900x45' + alert_rule_test: + - eval_time: 40m + alertname: BuzzPartitionAuditNotRunning + exp_alerts: [] + - eval_time: 52m + alertname: BuzzPartitionAuditNotRunning + exp_alerts: + - exp_labels: {pod: stopped, table: events} + - eval_time: 57m + alertname: BuzzPartitionAuditStale + exp_alerts: + - exp_labels: {pod: stopped, table: events} + + # The audit loop stops right after the startup audit. It is caught once the + # pod is one window old. + - interval: 1m + input_series: + - series: 'buzz_partition_audit_runs_total{pod="early",table="events",outcome="ok"}' + values: '1x80' + alert_rule_test: + - eval_time: 33m + alertname: BuzzPartitionAuditNotRunning + exp_alerts: [] + - eval_time: 36m + alertname: BuzzPartitionAuditNotRunning + exp_alerts: + - exp_labels: {pod: early, table: events} + + # A pod terminates at 24 minutes. Range selectors still read its earlier + # samples, long enough to complete `for: 5m` at 50 minutes, but it must not + # start an alert. + - interval: 1m + input_series: + - series: 'buzz_partition_audit_runs_total{pod="gone",table="events",outcome="ok"}' + values: '1x14 2x9 stale' + - series: 'buzz_partition_audit_failures_total{pod="gone"}' + values: '0x10 1 2 3x7 stale' + alert_rule_test: + - eval_time: 45m + alertname: BuzzPartitionAuditNotRunning + exp_alerts: [] + - eval_time: 52m + alertname: BuzzPartitionAuditNotRunning + exp_alerts: [] + - eval_time: 60m + alertname: BuzzPartitionAuditNotRunning + exp_alerts: [] + - eval_time: 30m + alertname: BuzzPartitionAuditFailing + exp_alerts: [] + + # A table that has never audited successfully exports a 0 timestamp. + - interval: 1m + input_series: + - series: 'buzz_partition_audit_last_success_timestamp_seconds{pod="never",table="events"}' + values: '0x60' + alert_rule_test: + - eval_time: 35m + alertname: BuzzPartitionAuditStale + exp_alerts: [] + - eval_time: 45m + alertname: BuzzPartitionAuditStale + exp_alerts: + - exp_labels: {pod: never, table: events} + + # One failure does not alert; a second within an hour does. + - interval: 1m + input_series: + - series: 'buzz_partition_audit_failures_total{pod="once"}' + values: '0x10 1x50' + - series: 'buzz_partition_audit_failures_total{pod="twice"}' + values: '0x10 1x10 2x40' + alert_rule_test: + - eval_time: 40m + alertname: BuzzPartitionAuditFailing + exp_alerts: + - exp_labels: {pod: twice} diff --git a/docs/partition-catalog-monitoring.rules.yml b/docs/partition-catalog-monitoring.rules.yml new file mode 100644 index 0000000000..cef7430109 --- /dev/null +++ b/docs/partition-catalog-monitoring.rules.yml @@ -0,0 +1,20 @@ +# Alert rules from docs/partition-catalog-monitoring.md. Keep them identical to +# the doc; partition-catalog-monitoring.rules.test.yml tests them. +groups: + - name: buzz-partition-catalog + rules: + - alert: BuzzPartitionAuditStale + expr: time() - buzz_partition_audit_last_success_timestamp_seconds > 1800 + for: 10m + + - alert: BuzzPartitionAuditNotRunning + expr: | + sum without (outcome) (increase(buzz_partition_audit_runs_total[30m])) == 0 + and sum without (outcome) (buzz_partition_audit_runs_total) + and sum without (outcome) (buzz_partition_audit_runs_total offset 30m) + for: 5m + + - alert: BuzzPartitionAuditFailing + expr: | + increase(buzz_partition_audit_failures_total[1h]) >= 2 + and buzz_partition_audit_failures_total diff --git a/scripts/run-tests.sh b/scripts/run-tests.sh index b51e575973..9b8d613647 100755 --- a/scripts/run-tests.sh +++ b/scripts/run-tests.sh @@ -135,6 +135,9 @@ run_unit_tests() { run_test_step "buzz-admin community archive command tests" \ cargo test -p buzz-admin communities -- --nocapture + run_test_step "buzz-admin partition audit outcome contract test" \ + cargo test -p buzz-admin tests::partition_audit_outcomes_keep_their_json_names_and_exit_codes -- --exact --nocapture + # Multi-tenant conformance gate: independent replay checker + golden # fixtures (buzz-conformance). Pure in-process trace replay, no infra. run_test_step "buzz-conformance tests" \ @@ -287,6 +290,9 @@ run_unit_tests() { run_test_step "buzz-relay auth metrics contract tests" \ cargo test -p buzz-relay --lib metrics::contract_tests:: -- --nocapture + run_test_step "buzz-relay partition alert series tests" \ + cargo test -p buzz-relay --lib metrics::partition_alert_tests:: -- --nocapture + local nip_fi_exact_tests=( audio::room::tests::roster_revisions_are_ordered_and_snapshot_is_authoritative connection::tests::auth_lifecycle_reconciles_every_terminal_and_never_leaks_gauge