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