From 615095590b303a398eb92fc0796b724fe9d3ff06 Mon Sep 17 00:00:00 2001 From: Ian Oberst Date: Thu, 20 Aug 2026 14:08:46 -0700 Subject: [PATCH 1/2] fix(db): audit partition catalog before creation Co-authored-by: Ian Oberst Signed-off-by: Ian Oberst Co-authored-by: Codex Ai-assisted: true --- .env.example | 4 + crates/buzz-db/src/lib.rs | 16 +- crates/buzz-db/src/partition.rs | 2002 ++++++++++++++++++++++++++-- crates/buzz-relay/src/config.rs | 79 ++ crates/buzz-relay/src/main.rs | 97 +- crates/buzz-relay/src/mesh_boot.rs | 6 +- crates/buzz-relay/src/router.rs | 24 +- crates/buzz-relay/src/state.rs | 32 + 8 files changed, 2115 insertions(+), 145 deletions(-) diff --git a/.env.example b/.env.example index 0f7bbba6f13..b44f2997ae0 100644 --- a/.env.example +++ b/.env.example @@ -37,6 +37,10 @@ REDIS_URL=redis://localhost:6379 # Max connections in each of the relay's Postgres pools — writer and, when # READ_DATABASE_URL is set, reader (default 50). # BUZZ_DB_POOL_SIZE=50 +# Read-only partition catalog audit interval, clamped to 60s..24h (default 900s). +# BUZZ_PARTITION_AUDIT_INTERVAL_SECS=900 +# Emergency kill switch for monthly partition CREATE; audits remain enabled. +# BUZZ_PARTITION_MANAGER_CREATE_ENABLED=true # ----------------------------------------------------------------------------- # Typesense (search) diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 3ff230f9503..2f5c5566a9c 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -4232,10 +4232,20 @@ impl Db { .await } - /// Ensures monthly partitions exist for the next N months. + /// Audits the managed partition catalogs without making writes. + #[datastore_span(name = "audit_partitions", system = "postgresql")] + pub async fn audit_partitions(&self, months_ahead: u32) -> Result { + partition::audit_partition_catalog(&self.pool, months_ahead).await + } + + /// Ensures monthly partitions exist for the next N months when creation is enabled. #[datastore_span(name = "ensure_future_partitions", system = "postgresql")] - pub async fn ensure_future_partitions(&self, months_ahead: u32) -> Result<()> { - partition::ensure_future_partitions(&self.pool, months_ahead).await + pub async fn ensure_future_partitions( + &self, + months_ahead: u32, + create_enabled: bool, + ) -> Result { + partition::ensure_future_partitions(&self.pool, months_ahead, create_enabled).await } /// Backfill `d_tag` for existing NIP-33 events (kind 30000–39999) that have `d_tag IS NULL`. diff --git a/crates/buzz-db/src/partition.rs b/crates/buzz-db/src/partition.rs index b3803f1b34c..3582158695e 100644 --- a/crates/buzz-db/src/partition.rs +++ b/crates/buzz-db/src/partition.rs @@ -1,152 +1,1006 @@ -//! Monthly partition manager for `events` and `delivery_log`. -//! -//! Call `ensure_future_partitions` on startup and monthly via cron. +//! Read-only catalog audit and monthly partition manager for `events` and +//! `delivery_log`. -use chrono::{Datelike, TimeZone, Utc}; -use sqlx::{PgPool, Row}; +use std::collections::{HashMap, HashSet}; +use std::time::Instant; + +use chrono::{DateTime, Datelike, NaiveDate, TimeZone, Timelike, Utc}; +use sqlx::{PgConnection, PgPool, Row}; use tracing::info; use crate::error::{DbError, Result}; -/// Tables that may be partition-managed. Allowlist prevents DDL injection. +/// Tables that may be partition-managed. The allowlist prevents DDL injection. const PARTITIONED_TABLES: &[&str] = &["events", "delivery_log"]; -/// Ensures monthly partition tables exist for the next `months_ahead` months. -pub async fn ensure_future_partitions(pool: &PgPool, months_ahead: u32) -> Result<()> { - let now = Utc::now(); +/// A parsed endpoint from a PostgreSQL range-partition bound. +#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)] +pub enum PartitionBound { + /// The range has no lower limit. + MinValue, + /// A finite UTC timestamp. + Finite(DateTime), + /// The range has no upper limit. + MaxValue, +} - for i in 0..=(months_ahead as i32) { - let year = now.year(); - let month = now.month() as i32 + i; - let (target_year, target_month) = if month > 12 { - (year + (month - 1) / 12, ((month - 1) % 12 + 1) as u32) - } else { - (year, month as u32) - }; +/// The catalog classification of one attached child partition. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum PartitionChildKind { + /// The canonical `{parent}_pYYYY_MM` name agrees with exact month bounds. + CanonicalMonthly, + /// A finite lower bound extends through `MAXVALUE`. + CatchAll, + /// A well-formed monthly leaf uses a non-canonical name. + LegacyLeaf, + /// The canonical `{parent}_p_past` left-edge partition. + Past, + /// Bounds were unparseable, invalid, overlapping, or disagreed with the name. + Anomalous, +} - let (end_year, end_month) = if target_month == 12 { - (target_year + 1, 1u32) - } else { - (target_year, target_month + 1) - }; +/// How one target month is covered by the current partition catalog. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum MonthCoverageKind { + /// A bounded child covers the complete month. + CoveredByMonthly, + /// A right-edge catch-all covers the complete month. + CoveredByCatchAll, + /// No parseable child covers the complete month. + Uncovered, +} - let start = Utc - .with_ymd_and_hms(target_year, target_month, 1, 0, 0, 0) - .single() - .ok_or_else(|| { - DbError::InvalidData(format!("invalid date: {target_year}-{target_month:02}-01")) - })?; - let end = Utc - .with_ymd_and_hms(end_year, end_month, 1, 0, 0, 0) - .single() - .ok_or_else(|| { - DbError::InvalidData(format!("invalid date: {end_year}-{end_month:02}-01")) - })?; +/// Coverage for one target month. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct MonthCoverage { + /// Inclusive month start. + pub start: DateTime, + /// Catalog coverage classification. + pub kind: MonthCoverageKind, +} - let suffix = format!("{:04}_{:02}", target_year, target_month); - let start_str = start.format("%Y-%m-%d").to_string(); - let end_str = end.format("%Y-%m-%d").to_string(); +/// Audit details for one attached child partition. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct PartitionChildAudit { + /// Child relation name. + pub name: String, + /// PostgreSQL `pg_class.relkind` for the immediate child. + pub relation_kind: String, + /// Parsed inclusive lower range endpoint, when parsing succeeded. + pub lower: Option, + /// Parsed exclusive upper range endpoint, when parsing succeeded. + pub upper: Option, + /// Catalog classification. + pub kind: PartitionChildKind, + /// Parent trigger names absent from this child or a routable descendant leaf. + /// Nested-leaf entries are qualified as `{leaf}:{trigger}`. + pub missing_triggers: Vec, + /// Child trigger names absent from the parent. + pub extra_triggers: Vec, +} - for table in PARTITIONED_TABLES { - ensure_partition(pool, table, &start_str, &end_str, &suffix).await?; - } +/// Effective routable range for one leaf in the partition tree. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct PartitionLeafAudit { + /// Leaf relation name. + pub name: String, + /// Immediate child of the managed parent that owns this leaf. + pub root_child: String, + /// Effective inclusive lower bound after intersecting the ancestor path. + pub lower: PartitionBound, + /// Effective exclusive upper bound after intersecting the ancestor path. + pub upper: PartitionBound, + /// Whether this leaf is below an immediate partitioned child. + pub nested: bool, +} + +/// Read-only audit result for one managed parent table. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct PartitionTableAudit { + /// Parent relation name. + pub table: &'static str, + /// All attached children found via `pg_inherits`. + 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. + pub months: Vec, + /// Whether a parseable routable leaf covers the audit timestamp. + pub serving_safe: bool, +} + +impl PartitionTableAudit { + /// Number of children with anomalous catalog state. + pub fn anomalous_children(&self) -> usize { + self.children + .iter() + .filter(|child| child.kind == PartitionChildKind::Anomalous) + .count() } - Ok(()) + /// Number of parent triggers missing across all children. + pub fn missing_trigger_count(&self) -> usize { + self.children + .iter() + .map(|child| child.missing_triggers.len()) + .sum() + } + + /// Number of child-only row triggers across all children. + pub fn extra_trigger_count(&self) -> usize { + self.children + .iter() + .map(|child| child.extra_triggers.len()) + .sum() + } + + /// Whether the table is serving but has state requiring operator attention. + pub fn degraded(&self) -> bool { + self.children.iter().any(|child| { + matches!( + child.kind, + PartitionChildKind::LegacyLeaf | PartitionChildKind::Anomalous + ) || !child.missing_triggers.is_empty() + || !child.extra_triggers.is_empty() + }) || self + .months + .iter() + .any(|month| month.kind != MonthCoverageKind::CoveredByMonthly) + } } -/// Validate that a partition suffix is digits and underscores only. -fn validate_partition_suffix(suffix: &str) -> bool { - !suffix.is_empty() && suffix.chars().all(|c| c.is_ascii_digit() || c == '_') +/// Read-only audit result for every managed parent table. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct PartitionAudit { + /// Timestamp whose serving coverage was checked. + pub audited_at: DateTime, + /// Per-parent audit details. + pub tables: Vec, } -/// Validate that a date string matches YYYY-MM-DD format. -fn validate_date_str(s: &str) -> bool { - let bytes = s.as_bytes(); - bytes.len() == 10 - && bytes[4] == b'-' - && bytes[7] == b'-' - && bytes[..4].iter().all(|b| b.is_ascii_digit()) - && bytes[5..7].iter().all(|b| b.is_ascii_digit()) - && bytes[8..].iter().all(|b| b.is_ascii_digit()) +impl PartitionAudit { + /// Whether every managed parent can accept a row timestamped at `audited_at`. + pub fn serving_safe(&self) -> bool { + self.serving_safe_at(self.audited_at) + } + + /// Whether the cached catalog proves every managed parent covers `timestamp`. + pub fn serving_safe_at(&self, timestamp: DateTime) -> bool { + self.tables.iter().all(|table| { + table + .coverage_leaves + .iter() + .any(|leaf| leaf_covers_timestamp(leaf, ×tamp)) + }) + } } -async fn ensure_partition( +#[derive(Debug)] +struct CatalogChild { + name: String, + relation_kind: String, + lower: Option, + upper: Option, + kind: PartitionChildKind, +} + +/// Audit the managed partition catalogs without making any writes. +pub async fn audit_partition_catalog(pool: &PgPool, months_ahead: u32) -> Result { + audit_partition_catalog_at(pool, months_ahead, Utc::now()).await +} + +async fn audit_partition_catalog_at( pool: &PgPool, - table_name: &str, - start_date_str: &str, - end_date_str: &str, - suffix: &str, -) -> Result<()> { - // Allowlist check -- parameterized queries cannot be used for DDL identifiers. - if !PARTITIONED_TABLES.contains(&table_name) { - return Err(DbError::InvalidData(format!( - "table not in partition allowlist: {table_name:?}" - ))); + months_ahead: u32, + now: DateTime, +) -> Result { + let mut tables = Vec::with_capacity(PARTITIONED_TABLES.len()); + let mut errors = Vec::new(); + + for &table in PARTITIONED_TABLES { + let started = Instant::now(); + match audit_table(pool, table, months_ahead, now).await { + Ok(audit) => { + emit_audit_metrics(&audit, started.elapsed().as_secs_f64(), now); + tables.push(audit); + } + Err(error) => { + metrics::counter!( + "buzz_partition_audit_runs_total", + "table" => table, + "outcome" => "error" + ) + .increment(1); + metrics::histogram!( + "buzz_partition_audit_duration_seconds", + "table" => table + ) + .record(started.elapsed().as_secs_f64()); + errors.push(format!("{table}: {error}")); + } + } } - if !validate_partition_suffix(suffix) { - return Err(DbError::InvalidData(format!( - "partition suffix contains invalid characters: {suffix:?}" - ))); + + if errors.is_empty() { + Ok(PartitionAudit { + audited_at: now, + tables, + }) + } else { + Err(DbError::InvalidData(format!( + "partition catalog audit failed: {}", + errors.join("; ") + ))) } - if !validate_date_str(start_date_str) { - return Err(DbError::InvalidData(format!( - "start_date_str is not YYYY-MM-DD: {start_date_str:?}" - ))); +} + +/// Audit first, then create only months proven to be uncovered. +/// +/// Covered ranges are never probed with DDL. Creation failures are collected +/// across all managed parents and months before an aggregate error is returned. +pub async fn ensure_future_partitions( + pool: &PgPool, + months_ahead: u32, + create_enabled: bool, +) -> Result { + ensure_future_partitions_at(pool, months_ahead, create_enabled, Utc::now()).await +} + +async fn ensure_future_partitions_at( + pool: &PgPool, + months_ahead: u32, + create_enabled: bool, + now: DateTime, +) -> Result { + let audit = audit_partition_catalog_at(pool, months_ahead, now).await?; + let mut errors = Vec::new(); + let mut created_any = false; + + for table in &audit.tables { + for month in &table.months { + match month.kind { + MonthCoverageKind::CoveredByMonthly | MonthCoverageKind::CoveredByCatchAll => { + metrics::counter!( + "buzz_partition_create_attempts_total", + "table" => table.table, + "outcome" => "skipped_covered" + ) + .increment(1); + } + MonthCoverageKind::Uncovered if !create_enabled => {} + MonthCoverageKind::Uncovered => { + let expected_name = partition_name(table.table, month.start); + if table + .children + .iter() + .any(|child| child.name == expected_name) + { + metrics::counter!( + "buzz_partition_create_attempts_total", + "table" => table.table, + "outcome" => "error" + ) + .increment(1); + errors.push(format!( + "{} {}: canonical name {expected_name} exists with mismatched bounds", + table.table, + month.start.format("%Y-%m") + )); + continue; + } + match create_month_partition(pool, table.table, month.start).await { + Ok(name) => { + created_any = true; + metrics::counter!( + "buzz_partition_create_attempts_total", + "table" => table.table, + "outcome" => "created" + ) + .increment(1); + info!(table = table.table, partition = name, "added partition"); + } + Err(error) => { + metrics::counter!( + "buzz_partition_create_attempts_total", + "table" => table.table, + "outcome" => "error" + ) + .increment(1); + errors.push(format!( + "{} {}: {error}", + table.table, + month.start.format("%Y-%m") + )); + } + } + } + } + } } - if !validate_date_str(end_date_str) { - return Err(DbError::InvalidData(format!( - "end_date_str is not YYYY-MM-DD: {end_date_str:?}" - ))); + + if errors.is_empty() { + if created_any { + audit_partition_catalog_at(pool, months_ahead, now).await + } else { + Ok(audit) + } + } else { + Err(DbError::InvalidData(format!( + "partition creation failed: {}", + errors.join("; ") + ))) } +} - let partition_name = format!("{table_name}_p{suffix}"); +async fn audit_table( + pool: &PgPool, + table: &'static str, + months_ahead: u32, + now: DateTime, +) -> Result { + let mut transaction = pool.begin().await?; + sqlx::query("SET TRANSACTION READ ONLY") + .execute(&mut *transaction) + .await?; + pin_catalog_rendering(&mut transaction).await?; + let audit = audit_table_on(&mut transaction, table, months_ahead, now).await?; + transaction.commit().await?; + Ok(audit) +} - let row = sqlx::query( +async fn pin_catalog_rendering(connection: &mut PgConnection) -> Result<()> { + sqlx::query("SET LOCAL DateStyle TO 'ISO, YMD'") + .execute(&mut *connection) + .await?; + sqlx::query("SET LOCAL TimeZone TO 'UTC'") + .execute(&mut *connection) + .await?; + Ok(()) +} + +async fn audit_table_on( + connection: &mut PgConnection, + table: &'static str, + months_ahead: u32, + now: DateTime, +) -> Result { + let rows = sqlx::query( r#" - SELECT COUNT(*) as cnt - FROM pg_catalog.pg_class c - JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace - WHERE n.nspname = current_schema() - AND c.relname = $1 - AND c.relispartition = true + WITH RECURSIVE partition_tree AS ( + SELECT child.oid AS relation_oid, + parent.oid AS parent_oid, + child.relname, + child.relkind, + child.relpartbound, + child.relname AS root_child, + pg_catalog.pg_get_partkeydef(parent.oid) AS root_partition_key, + pg_catalog.pg_get_partkeydef(parent.oid) AS bound_partition_key, + 0 AS depth + FROM pg_catalog.pg_inherits inherited + JOIN pg_catalog.pg_class parent ON parent.oid = inherited.inhparent + JOIN pg_catalog.pg_namespace parent_ns ON parent_ns.oid = parent.relnamespace + JOIN pg_catalog.pg_class child ON child.oid = inherited.inhrelid + WHERE parent_ns.nspname = current_schema() + AND parent.relname = $1 + + UNION ALL + + SELECT descendant.oid, + tree.relation_oid, + descendant.relname, + descendant.relkind, + descendant.relpartbound, + tree.root_child, + tree.root_partition_key, + pg_catalog.pg_get_partkeydef(tree.relation_oid), + tree.depth + 1 + FROM partition_tree tree + JOIN pg_catalog.pg_inherits nested + ON nested.inhparent = tree.relation_oid + JOIN pg_catalog.pg_class descendant ON descendant.oid = nested.inhrelid + ) + SELECT tree.relation_oid::bigint AS relation_oid, + tree.parent_oid::bigint AS parent_oid, + tree.relname AS relation_name, + tree.relkind::text AS relation_kind, + tree.root_child, + tree.depth, + tree.root_partition_key, + tree.bound_partition_key, + pg_catalog.pg_get_expr(tree.relpartbound, tree.relation_oid) AS bound, + NOT EXISTS ( + SELECT 1 + FROM pg_catalog.pg_inherits child_edge + WHERE child_edge.inhparent = tree.relation_oid + ) AS is_leaf + FROM partition_tree tree + ORDER BY tree.depth, tree.relname "#, ) - .bind(&partition_name) - .fetch_one(pool) + .bind(table) + .fetch_all(&mut *connection) .await?; - let cnt: i64 = row.try_get("cnt")?; - if cnt > 0 { - return Ok(()); + let mut children = Vec::new(); + let mut coverage_leaves = Vec::new(); + let mut effective_ranges = HashMap::>::new(); + for row in rows { + let relation_oid: i64 = row.try_get("relation_oid")?; + let parent_oid: i64 = row.try_get("parent_oid")?; + let name: String = row.try_get("relation_name")?; + let relation_kind: String = row.try_get("relation_kind")?; + let root_child: String = row.try_get("root_child")?; + let depth: i32 = row.try_get("depth")?; + let is_leaf: bool = row.try_get("is_leaf")?; + let root_partition_key: Option = row.try_get("root_partition_key")?; + let bound_partition_key: Option = row.try_get("bound_partition_key")?; + let partition_key_compatible = + root_partition_key.is_some() && root_partition_key == bound_partition_key; + let expression: String = row.try_get("bound")?; + let own_range = parse_range_bounds(&expression); + let effective_range = if !partition_key_compatible { + None + } else if depth == 0 { + own_range.clone() + } else { + effective_ranges + .get(&parent_oid) + .and_then(|parent| parent.as_ref()) + .and_then(|parent| { + own_range + .as_ref() + .and_then(|own| intersect_ranges(parent, own)) + }) + }; + effective_ranges.insert(relation_oid, effective_range.clone()); + + if depth == 0 { + let (lower, upper) = match own_range { + Some(bounds) => (Some(bounds.0), Some(bounds.1)), + None => (None, None), + }; + let kind = if partition_key_compatible { + classify_child(&relation_kind, table, &name, lower.as_ref(), upper.as_ref()) + } else { + PartitionChildKind::Anomalous + }; + children.push(CatalogChild { + name: name.clone(), + relation_kind: relation_kind.clone(), + lower, + upper, + kind, + }); + } + + let routable_leaf = + (depth == 0 && relation_kind == "r") || (depth > 0 && is_leaf && relation_kind != "p"); + if routable_leaf { + if let Some((lower, upper)) = effective_range { + coverage_leaves.push(PartitionLeafAudit { + name, + root_child, + lower, + upper, + nested: depth > 0, + }); + } + } } + mark_overlaps_anomalous(&mut children); - // DDL identifiers cannot be parameterized -- all inputs are validated above. - let sql = format!( - "CREATE TABLE IF NOT EXISTS {partition_name} PARTITION OF {table_name} \ - FOR VALUES FROM ('{start_date_str}') TO ('{end_date_str}')" - ); + let parent_triggers = trigger_metadata_for_parent(connection, table).await?; + let descendant_triggers = trigger_metadata_for_descendants(connection, table).await?; + let mut child_audits = Vec::with_capacity(children.len()); + for child in children { + let triggers = descendant_triggers + .get(&child.name) + .map(|relation| &relation.triggers); + let mut missing_triggers = Vec::new(); + let routable_leaves: Vec<_> = descendant_triggers + .values() + .filter(|relation| { + relation.root_child == child.name + && ((relation.depth == 0 && relation.relation_kind == "r") + || (relation.depth > 0 + && relation.is_leaf + && relation.relation_kind != "p")) + }) + .collect(); + for leaf in routable_leaves { + for (name, parent_oid) in &parent_triggers { + let present = + trigger_lineage_reaches_parent(leaf, name, *parent_oid, &descendant_triggers) + && leaf + .triggers + .get(name) + .is_some_and(|metadata| matches!(metadata.enabled.as_str(), "O" | "A")); + if !present { + if leaf.depth == 0 { + missing_triggers.push(name.clone()); + } else { + missing_triggers.push(format!("{}:{name}", leaf.name)); + } + } + } + } + missing_triggers.sort(); + let mut extra_triggers: Vec<_> = triggers + .map(|triggers| { + triggers + .keys() + .filter(|name| !parent_triggers.contains_key(*name)) + .cloned() + .collect() + }) + .unwrap_or_default(); + extra_triggers.sort(); + child_audits.push(PartitionChildAudit { + name: child.name, + relation_kind: child.relation_kind, + lower: child.lower, + upper: child.upper, + kind: child.kind, + missing_triggers, + extra_triggers, + }); + } + + let mut months = Vec::with_capacity(months_ahead as usize + 1); + for offset in 0..=months_ahead as i32 { + 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 serving_safe = coverage_leaves + .iter() + .any(|leaf| leaf_covers_timestamp(leaf, &now)); + + Ok(PartitionTableAudit { + table, + children: child_audits, + coverage_leaves, + months, + serving_safe, + }) +} + +fn emit_audit_metrics(audit: &PartitionTableAudit, duration_seconds: f64, now: DateTime) { + let outcome = if audit.degraded() { "degraded" } else { "ok" }; + let uncovered = audit + .months + .iter() + .filter(|month| month.kind == MonthCoverageKind::Uncovered) + .count(); + let catch_all = audit + .months + .iter() + .filter(|month| month.kind == MonthCoverageKind::CoveredByCatchAll) + .count(); + + metrics::counter!( + "buzz_partition_audit_runs_total", + "table" => audit.table, + "outcome" => outcome + ) + .increment(1); + metrics::gauge!("buzz_partition_serving_safe", "table" => audit.table) + .set(if audit.serving_safe { 1.0 } else { 0.0 }); + metrics::gauge!("buzz_partition_uncovered_months", "table" => audit.table) + .set(uncovered as f64); + metrics::gauge!( + "buzz_partition_catch_all_covered_months", + "table" => audit.table + ) + .set(catch_all as f64); + metrics::gauge!("buzz_partition_anomalous_children", "table" => audit.table) + .set(audit.anomalous_children() as f64); + metrics::gauge!( + "buzz_partition_trigger_parity_missing", + "table" => audit.table + ) + .set(audit.missing_trigger_count() as f64); + metrics::gauge!( + "buzz_partition_trigger_parity_extra", + "table" => audit.table + ) + .set(audit.extra_trigger_count() as f64); + metrics::histogram!( + "buzz_partition_audit_duration_seconds", + "table" => audit.table + ) + .record(duration_seconds); + metrics::gauge!( + "buzz_partition_audit_last_success_timestamp_seconds", + "table" => audit.table + ) + .set(now.timestamp() as f64); +} + +#[derive(Debug, Clone)] +struct ChildTriggerMetadata { + oid: i64, + parent_oid: i64, + enabled: String, +} + +#[derive(Debug, Clone)] +struct DescendantTriggerMetadata { + name: String, + relation_oid: i64, + parent_relation_oid: i64, + root_child: String, + relation_kind: String, + depth: i32, + is_leaf: bool, + triggers: HashMap, +} + +async fn trigger_metadata_for_parent( + connection: &mut PgConnection, + table: &str, +) -> Result> { + let rows = sqlx::query( + r#" + SELECT trigger.tgname, trigger.oid::bigint AS trigger_oid + FROM pg_catalog.pg_class parent + JOIN pg_catalog.pg_namespace parent_ns ON parent_ns.oid = parent.relnamespace + JOIN pg_catalog.pg_trigger trigger ON trigger.tgrelid = parent.oid + WHERE parent_ns.nspname = current_schema() + AND parent.relname = $1 + AND NOT trigger.tgisinternal + AND (trigger.tgtype & 1) = 1 + "#, + ) + .bind(table) + .fetch_all(&mut *connection) + .await?; + rows.into_iter() + .map(|row| Ok((row.try_get("tgname")?, row.try_get("trigger_oid")?))) + .collect() +} + +async fn trigger_metadata_for_descendants( + connection: &mut PgConnection, + table: &str, +) -> Result> { + let rows = sqlx::query( + r#" + WITH RECURSIVE partition_tree AS ( + SELECT child.oid AS relation_oid, + parent.oid AS parent_relation_oid, + child.relname, + child.relkind, + child.relname AS root_child, + 0 AS depth + FROM pg_catalog.pg_inherits inherited + JOIN pg_catalog.pg_class parent ON parent.oid = inherited.inhparent + JOIN pg_catalog.pg_namespace parent_ns ON parent_ns.oid = parent.relnamespace + JOIN pg_catalog.pg_class child ON child.oid = inherited.inhrelid + WHERE parent_ns.nspname = current_schema() + AND parent.relname = $1 + + UNION ALL + + SELECT descendant.oid, + tree.relation_oid, + descendant.relname, + descendant.relkind, + tree.root_child, + tree.depth + 1 + FROM partition_tree tree + JOIN pg_catalog.pg_inherits nested + ON nested.inhparent = tree.relation_oid + JOIN pg_catalog.pg_class descendant ON descendant.oid = nested.inhrelid + ) + SELECT tree.relname AS relation_name, + tree.relation_oid::bigint AS relation_oid, + tree.parent_relation_oid::bigint AS parent_relation_oid, + tree.root_child, + tree.relkind::text AS relation_kind, + tree.depth, + NOT EXISTS ( + SELECT 1 + FROM pg_catalog.pg_inherits child_edge + WHERE child_edge.inhparent = tree.relation_oid + ) AS is_leaf, + trigger.tgname, + trigger.oid::bigint AS trigger_oid, + trigger.tgparentid::bigint AS trigger_parent_oid, + trigger.tgenabled::text AS trigger_enabled + FROM partition_tree tree + LEFT JOIN pg_catalog.pg_trigger trigger + ON trigger.tgrelid = tree.relation_oid + AND NOT trigger.tgisinternal + AND (trigger.tgtype & 1) = 1 + ORDER BY tree.depth, tree.relname, trigger.tgname + "#, + ) + .bind(table) + .fetch_all(&mut *connection) + .await?; + let mut descendants = HashMap::::new(); + for row in rows { + let name: String = row.try_get("relation_name")?; + let trigger: Option = row.try_get("tgname")?; + let entry = descendants + .entry(name.clone()) + .or_insert(DescendantTriggerMetadata { + name, + relation_oid: row.try_get("relation_oid")?, + parent_relation_oid: row.try_get("parent_relation_oid")?, + root_child: row.try_get("root_child")?, + relation_kind: row.try_get("relation_kind")?, + depth: row.try_get("depth")?, + is_leaf: row.try_get("is_leaf")?, + triggers: HashMap::new(), + }); + if let Some(trigger) = trigger { + entry.triggers.insert( + trigger, + ChildTriggerMetadata { + oid: row.try_get("trigger_oid")?, + parent_oid: row.try_get("trigger_parent_oid")?, + enabled: row.try_get("trigger_enabled")?, + }, + ); + } + } + Ok(descendants) +} + +fn trigger_lineage_reaches_parent( + leaf: &DescendantTriggerMetadata, + trigger_name: &str, + parent_trigger_oid: i64, + descendants: &HashMap, +) -> bool { + let mut relation = leaf; + let Some(mut trigger) = relation.triggers.get(trigger_name) else { + return false; + }; + while relation.depth > 0 { + let Some(parent_relation) = descendants + .values() + .find(|candidate| candidate.relation_oid == relation.parent_relation_oid) + else { + return false; + }; + let Some(parent_trigger) = parent_relation.triggers.get(trigger_name) else { + return false; + }; + if trigger.parent_oid != parent_trigger.oid { + return false; + } + relation = parent_relation; + trigger = parent_trigger; + } + trigger.parent_oid == parent_trigger_oid +} - match sqlx::query(sqlx::AssertSqlSafe(sql)).execute(pool).await { - Ok(_) => { - info!("added partition {partition_name}"); - Ok(()) +fn classify_child( + relation_kind: &str, + table: &str, + name: &str, + lower: Option<&PartitionBound>, + upper: Option<&PartitionBound>, +) -> PartitionChildKind { + if relation_kind != "r" { + return PartitionChildKind::Anomalous; + } + match (lower, upper) { + (Some(PartitionBound::Finite(_)), Some(PartitionBound::MaxValue)) => { + PartitionChildKind::CatchAll } - Err(sqlx::Error::Database(db_err)) - if db_err.code().as_deref() == Some("42P17") - && db_err.message().contains("would overlap partition") => + (Some(PartitionBound::MinValue), Some(PartitionBound::Finite(_))) + if name == format!("{table}_p_past") => { - // Fresh schemas include a right-edge catch-all partition (`*_p_future`). - // If it already covers this month, the table is still safe for writes; - // treat the overlap as "ensured" rather than failing startup. - info!( - partition_name, - "partition range already covered by an existing partition" - ); - Ok(()) + PartitionChildKind::Past + } + (Some(PartitionBound::Finite(lower)), Some(PartitionBound::Finite(upper))) + if is_exact_month(lower, upper) => + { + let canonical = format!("{table}_p{:04}_{:02}", lower.year(), lower.month()); + if name == canonical { + PartitionChildKind::CanonicalMonthly + } else if canonical_month_name(table, name).is_some() { + PartitionChildKind::Anomalous + } else { + PartitionChildKind::LegacyLeaf + } + } + _ => PartitionChildKind::Anomalous, + } +} + +fn canonical_month_name(table: &str, name: &str) -> Option<(i32, u32)> { + let suffix = name.strip_prefix(&format!("{table}_p"))?; + if suffix.len() != 7 || suffix.as_bytes().get(4) != Some(&b'_') { + return None; + } + let year = suffix[..4].parse::().ok()?; + let month = suffix[5..].parse::().ok()?; + (1..=12).contains(&month).then_some((year, month)) +} + +fn is_exact_month(lower: &DateTime, upper: &DateTime) -> bool { + if lower.day() != 1 + || lower.hour() != 0 + || lower.minute() != 0 + || lower.second() != 0 + || lower.nanosecond() != 0 + { + return false; + } + let Ok((year, month)) = add_months(lower.year(), lower.month(), 1) else { + return false; + }; + month_start(year, month).is_ok_and(|expected| expected == *upper) +} + +fn mark_overlaps_anomalous(children: &mut [CatalogChild]) { + let mut overlapping = HashSet::new(); + for left in 0..children.len() { + for right in (left + 1)..children.len() { + if ranges_overlap(&children[left], &children[right]) { + overlapping.insert(left); + overlapping.insert(right); + } + } + } + for index in overlapping { + children[index].kind = PartitionChildKind::Anomalous; + } +} + +fn ranges_overlap(left: &CatalogChild, right: &CatalogChild) -> bool { + let (Some(left_lower), Some(left_upper), Some(right_lower), Some(right_upper)) = ( + left.lower.as_ref(), + left.upper.as_ref(), + right.lower.as_ref(), + right.upper.as_ref(), + ) else { + return false; + }; + left_lower < right_upper && right_lower < left_upper +} + +fn coverage_for_range( + leaves: &[PartitionLeafAudit], + start: &DateTime, + end: &DateTime, +) -> MonthCoverageKind { + let start = PartitionBound::Finite(*start); + let end = PartitionBound::Finite(*end); + let mut ranges = leaves + .iter() + .filter(|leaf| leaf.upper > start && leaf.lower < end) + .collect::>(); + ranges.sort_by(|left, right| left.lower.cmp(&right.lower)); + + let mut cursor = start; + let mut catch_all_contributed = false; + for leaf in ranges { + if leaf.lower > cursor { + return MonthCoverageKind::Uncovered; + } + if leaf.upper > cursor { + catch_all_contributed |= leaf.upper == PartitionBound::MaxValue; + cursor = leaf.upper.clone(); + } + if cursor >= end { + return if catch_all_contributed { + MonthCoverageKind::CoveredByCatchAll + } else { + MonthCoverageKind::CoveredByMonthly + }; } - Err(e) => Err(e.into()), } + MonthCoverageKind::Uncovered +} + +fn leaf_covers_timestamp(leaf: &PartitionLeafAudit, timestamp: &DateTime) -> bool { + leaf.lower <= PartitionBound::Finite(*timestamp) + && leaf.upper > PartitionBound::Finite(*timestamp) +} + +fn intersect_ranges( + parent: &(PartitionBound, PartitionBound), + child: &(PartitionBound, PartitionBound), +) -> Option<(PartitionBound, PartitionBound)> { + let lower = std::cmp::max(parent.0.clone(), child.0.clone()); + let upper = std::cmp::min(parent.1.clone(), child.1.clone()); + (lower < upper).then_some((lower, upper)) +} + +fn parse_range_bounds(expression: &str) -> Option<(PartitionBound, PartitionBound)> { + let remainder = expression.strip_prefix("FOR VALUES FROM (")?; + let (lower, upper_with_suffix) = remainder.split_once(") TO (")?; + let upper = upper_with_suffix.strip_suffix(')')?; + Some((parse_bound(lower)?, parse_bound(upper)?)) +} + +fn parse_bound(input: &str) -> Option { + let input = input.trim(); + match input { + "MINVALUE" => Some(PartitionBound::MinValue), + "MAXVALUE" => Some(PartitionBound::MaxValue), + _ => { + let first_quote = input.find('\'')?; + let literal = &input[first_quote + 1..]; + let last_quote = literal.find('\'')?; + parse_timestamp_literal(&literal[..last_quote]).map(PartitionBound::Finite) + } + } +} + +fn parse_timestamp_literal(literal: &str) -> Option> { + if let Ok(timestamp) = DateTime::parse_from_rfc3339(literal) { + return Some(timestamp.with_timezone(&Utc)); + } + for format in ["%Y-%m-%d %H:%M:%S%.f%#z", "%Y-%m-%d %H:%M:%S%#z"] { + if let Ok(timestamp) = DateTime::parse_from_str(literal, format) { + return Some(timestamp.with_timezone(&Utc)); + } + } + let date = NaiveDate::parse_from_str(literal, "%Y-%m-%d").ok()?; + Some(Utc.from_utc_datetime(&date.and_hms_opt(0, 0, 0)?)) +} + +async fn create_month_partition( + pool: &PgPool, + table: &str, + start: DateTime, +) -> Result { + if !PARTITIONED_TABLES.contains(&table) { + return Err(DbError::InvalidData(format!( + "table not in partition allowlist: {table:?}" + ))); + } + let (end_year, end_month) = add_months(start.year(), start.month(), 1)?; + let end = month_start(end_year, end_month)?; + let partition_name = partition_name(table, start); + let start_date = start.format("%Y-%m-%d"); + let end_date = end.format("%Y-%m-%d"); + let sql = format!( + "CREATE TABLE IF NOT EXISTS {partition_name} PARTITION OF {table} \ + FOR VALUES FROM ('{start_date}') TO ('{end_date}')" + ); + sqlx::query(sqlx::AssertSqlSafe(sql)).execute(pool).await?; + Ok(partition_name) +} + +fn partition_name(table: &str, start: DateTime) -> String { + format!("{table}_p{:04}_{:02}", start.year(), start.month()) +} + +fn month_start(year: i32, month: u32) -> Result> { + Utc.with_ymd_and_hms(year, month, 1, 0, 0, 0) + .single() + .ok_or_else(|| DbError::InvalidData(format!("invalid date: {year}-{month:02}-01"))) +} + +fn add_months(year: i32, month: u32, offset: i32) -> Result<(i32, u32)> { + if !(1..=12).contains(&month) { + return Err(DbError::InvalidData(format!("invalid month: {month}"))); + } + let zero_based = year + .checked_mul(12) + .and_then(|value| value.checked_add(month as i32 - 1)) + .and_then(|value| value.checked_add(offset)) + .ok_or_else(|| { + DbError::InvalidData(format!("month arithmetic overflow: {year}-{month}")) + })?; + Ok(( + zero_based.div_euclid(12), + (zero_based.rem_euclid(12) + 1) as u32, + )) } #[cfg(test)] @@ -154,29 +1008,921 @@ mod tests { use super::*; #[test] - fn suffix_validation() { - assert!(validate_partition_suffix("2026_03")); - assert!(validate_partition_suffix("9999_12")); - assert!(!validate_partition_suffix("")); - assert!(!validate_partition_suffix("2026-03")); - assert!(!validate_partition_suffix("2026_03; DROP TABLE events--")); + fn parses_pg16_range_bound_formats() { + let expected = Utc.with_ymd_and_hms(2026, 7, 1, 0, 0, 0).unwrap(); + for expression in [ + "FOR VALUES FROM ('2026-07-01 00:00:00+00') TO (MAXVALUE)", + "FOR VALUES FROM ('2026-07-01 02:00:00+02') TO (MAXVALUE)", + "FOR VALUES FROM ('2026-07-01 00:00:00.000000+00'::timestamp with time zone) TO (MAXVALUE)", + "FOR VALUES FROM ('2026-07-01') TO (MAXVALUE)", + ] { + assert_eq!( + parse_range_bounds(expression), + Some((PartitionBound::Finite(expected), PartitionBound::MaxValue)), + "failed to parse {expression}" + ); + } + assert_eq!( + parse_range_bounds("FOR VALUES FROM (MINVALUE) TO ('2026-07-01')"), + Some((PartitionBound::MinValue, PartitionBound::Finite(expected))) + ); + assert!(parse_range_bounds("DEFAULT").is_none()); + } + + #[test] + fn classifies_names_and_bounds() { + let july = Utc.with_ymd_and_hms(2026, 7, 1, 0, 0, 0).unwrap(); + let august = Utc.with_ymd_and_hms(2026, 8, 1, 0, 0, 0).unwrap(); + let lower = PartitionBound::Finite(july); + let upper = PartitionBound::Finite(august); + assert_eq!( + classify_child("r", "events", "events_p2026_07", Some(&lower), Some(&upper)), + PartitionChildKind::CanonicalMonthly + ); + assert_eq!( + classify_child( + "r", + "events", + "events_july_repair", + Some(&lower), + Some(&upper) + ), + PartitionChildKind::LegacyLeaf + ); + assert_eq!( + classify_child("r", "events", "events_p2026_08", Some(&lower), Some(&upper)), + PartitionChildKind::Anomalous + ); + assert_eq!( + classify_child( + "r", + "events", + "events_p_future_next", + Some(&lower), + Some(&PartitionBound::MaxValue) + ), + PartitionChildKind::CatchAll + ); + assert_eq!( + classify_child("p", "events", "events_p2026_07", Some(&lower), Some(&upper)), + PartitionChildKind::Anomalous + ); + } + + #[test] + fn overlapping_ranges_are_anomalous() { + let july = Utc.with_ymd_and_hms(2026, 7, 1, 0, 0, 0).unwrap(); + let august = Utc.with_ymd_and_hms(2026, 8, 1, 0, 0, 0).unwrap(); + let september = Utc.with_ymd_and_hms(2026, 9, 1, 0, 0, 0).unwrap(); + let mut children = vec![ + CatalogChild { + name: "one".to_string(), + relation_kind: "r".to_string(), + lower: Some(PartitionBound::Finite(july)), + upper: Some(PartitionBound::Finite(september)), + kind: PartitionChildKind::LegacyLeaf, + }, + CatalogChild { + name: "two".to_string(), + relation_kind: "r".to_string(), + lower: Some(PartitionBound::Finite(august)), + upper: Some(PartitionBound::MaxValue), + kind: PartitionChildKind::CatchAll, + }, + ]; + mark_overlaps_anomalous(&mut children); + assert!(children + .iter() + .all(|child| child.kind == PartitionChildKind::Anomalous)); } #[test] - fn date_str_validation() { - assert!(validate_date_str("2026-03-01")); - assert!(validate_date_str("9999-12-31")); - assert!(!validate_date_str("2026-3-01")); - assert!(!validate_date_str("2026/03/01")); - assert!(!validate_date_str("20260301")); - assert!(!validate_date_str("2026-03-01; DROP TABLE events--")); + fn month_arithmetic_crosses_year_boundary() { + assert_eq!(add_months(2026, 12, 1).unwrap(), (2027, 1)); + assert_eq!(add_months(2026, 1, -1).unwrap(), (2025, 12)); + assert!(add_months(2026, 0, 1).is_err()); } #[test] - fn table_allowlist() { - assert!(PARTITIONED_TABLES.contains(&"events")); - assert!(PARTITIONED_TABLES.contains(&"delivery_log")); - assert!(!PARTITIONED_TABLES.contains(&"api_tokens")); - assert!(!PARTITIONED_TABLES.contains(&"users")); + fn month_coverage_can_span_multiple_nested_leaves() { + let start = Utc.with_ymd_and_hms(2026, 9, 1, 0, 0, 0).unwrap(); + let midpoint = Utc.with_ymd_and_hms(2026, 9, 15, 0, 0, 0).unwrap(); + let end = Utc.with_ymd_and_hms(2026, 10, 1, 0, 0, 0).unwrap(); + let leaves = vec![ + PartitionLeafAudit { + name: "first".to_string(), + root_child: "nested".to_string(), + lower: PartitionBound::Finite(start), + upper: PartitionBound::Finite(midpoint), + nested: true, + }, + PartitionLeafAudit { + name: "second".to_string(), + root_child: "nested".to_string(), + lower: PartitionBound::Finite(midpoint), + upper: PartitionBound::Finite(end), + nested: true, + }, + ]; + assert_eq!( + coverage_for_range(&leaves, &start, &end), + MonthCoverageKind::CoveredByMonthly + ); + } + + #[test] + fn extra_trigger_degradation_has_a_metric() { + let now = Utc.with_ymd_and_hms(2026, 9, 15, 12, 0, 0).unwrap(); + let start = Utc.with_ymd_and_hms(2026, 9, 1, 0, 0, 0).unwrap(); + let end = Utc.with_ymd_and_hms(2026, 10, 1, 0, 0, 0).unwrap(); + let audit = PartitionTableAudit { + table: "events", + children: vec![PartitionChildAudit { + name: "events_p2026_09".to_string(), + relation_kind: "r".to_string(), + lower: Some(PartitionBound::Finite(start)), + upper: Some(PartitionBound::Finite(end)), + kind: PartitionChildKind::CanonicalMonthly, + missing_triggers: Vec::new(), + extra_triggers: vec!["child_only_probe".to_string()], + }], + coverage_leaves: vec![PartitionLeafAudit { + name: "events_p2026_09".to_string(), + root_child: "events_p2026_09".to_string(), + lower: PartitionBound::Finite(start), + upper: PartitionBound::Finite(end), + nested: false, + }], + months: vec![MonthCoverage { + start, + kind: MonthCoverageKind::CoveredByMonthly, + }], + serving_safe: true, + }; + let recorder = metrics_util::debugging::DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + metrics::with_local_recorder(&recorder, || emit_audit_metrics(&audit, 0.01, now)); + + let extra = snapshotter + .snapshot() + .into_vec() + .into_iter() + .find_map(|(key, _, _, value)| { + (key.key().name() == "buzz_partition_trigger_parity_extra").then(|| { + let metrics_util::debugging::DebugValue::Gauge(value) = value else { + panic!("extra-trigger metric must be a gauge"); + }; + value.into_inner() + }) + }); + assert_eq!(extra, Some(1.0)); + } + + mod postgres { + use sqlx::postgres::PgPoolOptions; + use uuid::Uuid; + + use super::*; + + const TEST_DB_URL: &str = "postgres://buzz:buzz_dev@localhost:5432/buzz"; // sadscan:disable np.postgres.1 -- local test-only credentials + + async fn scratch_pool() -> (PgPool, PgPool, String) { + let url = std::env::var("BUZZ_TEST_DATABASE_URL") + .or_else(|_| std::env::var("DATABASE_URL")) + .unwrap_or_else(|_| TEST_DB_URL.to_string()); + let schema = format!("partition_audit_test_{}", Uuid::new_v4().simple()); + let admin = PgPool::connect(&url).await.expect("connect admin pool"); + sqlx::query(sqlx::AssertSqlSafe(format!("CREATE SCHEMA {schema}"))) + .execute(&admin) + .await + .expect("create scratch schema"); + let search_path_schema = schema.clone(); + let pool = PgPoolOptions::new() + .max_connections(1) + .after_connect(move |connection, _| { + let schema = search_path_schema.clone(); + Box::pin(async move { + sqlx::query(sqlx::AssertSqlSafe(format!("SET search_path TO {schema}"))) + .execute(connection) + .await?; + Ok(()) + }) + }) + .connect(&url) + .await + .expect("connect scratch pool"); + (pool, admin, schema) + } + + async fn drop_schema(admin: &PgPool, schema: &str) { + let _ = sqlx::query(sqlx::AssertSqlSafe(format!( + "DROP SCHEMA IF EXISTS {schema} CASCADE" + ))) + .execute(admin) + .await; + } + + async fn seed_parents(pool: &PgPool) { + sqlx::query( + "CREATE FUNCTION partition_test_trigger() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN RETURN NEW; END $$", + ) + .execute(pool) + .await + .expect("create trigger function"); + for (table, column) in [("events", "created_at"), ("delivery_log", "delivered_at")] { + let create = format!( + "CREATE TABLE {table} (id BIGSERIAL, {column} TIMESTAMPTZ NOT NULL, \ + alternate_at TIMESTAMPTZ NOT NULL, \ + PRIMARY KEY ({column}, alternate_at, id)) PARTITION BY RANGE ({column})" + ); + sqlx::query(sqlx::AssertSqlSafe(create)) + .execute(pool) + .await + .expect("create partitioned parent"); + let trigger = format!( + "CREATE TRIGGER partition_probe BEFORE INSERT ON {table} \ + FOR EACH ROW EXECUTE FUNCTION partition_test_trigger()" + ); + sqlx::query(sqlx::AssertSqlSafe(trigger)) + .execute(pool) + .await + .expect("create parent trigger"); + } + } + + async fn create_child(pool: &PgPool, table: &str, name: &str, lower: &str, upper: &str) { + let sql = format!( + "CREATE TABLE {name} PARTITION OF {table} FOR VALUES FROM ({lower}) TO ({upper})" + ); + sqlx::query(sqlx::AssertSqlSafe(sql)) + .execute(pool) + .await + .expect("create child"); + } + + async fn create_nested_child( + pool: &PgPool, + table: &str, + name: &str, + column: &str, + lower: &str, + upper: &str, + ) { + let sql = format!( + "CREATE TABLE {name} PARTITION OF {table} \ + FOR VALUES FROM ({lower}) TO ({upper}) PARTITION BY RANGE ({column})" + ); + sqlx::query(sqlx::AssertSqlSafe(sql)) + .execute(pool) + .await + .expect("create nested child"); + } + + async fn catalog_snapshot(pool: &PgPool) -> Vec<(String, String, String, i64)> { + let mut transaction = pool.begin().await.expect("begin catalog snapshot"); + sqlx::query("SET TRANSACTION READ ONLY") + .execute(&mut *transaction) + .await + .expect("make catalog snapshot read only"); + pin_catalog_rendering(&mut transaction) + .await + .expect("pin catalog rendering"); + let snapshot = sqlx::query_as( + r#" + SELECT child.relname, + child.relkind::text, + pg_catalog.pg_get_expr(child.relpartbound, child.oid), + count(trigger.oid) + FROM pg_catalog.pg_inherits inherited + JOIN pg_catalog.pg_class parent ON parent.oid = inherited.inhparent + JOIN pg_catalog.pg_namespace parent_ns ON parent_ns.oid = parent.relnamespace + JOIN pg_catalog.pg_class child ON child.oid = inherited.inhrelid + LEFT JOIN pg_catalog.pg_trigger trigger + ON trigger.tgrelid = child.oid AND NOT trigger.tgisinternal + WHERE parent_ns.nspname = current_schema() + AND child.relispartition + AND child.relkind IN ('r', 'p', 'f') + GROUP BY child.relname, child.relkind, child.relpartbound, child.oid + ORDER BY child.relname + "#, + ) + .fetch_all(&mut *transaction) + .await + .expect("catalog snapshot"); + transaction.commit().await.expect("commit catalog snapshot"); + snapshot + } + + fn fixed_now() -> DateTime { + Utc.with_ymd_and_hms(2026, 9, 15, 12, 0, 0).unwrap() + } + + async fn seed_fresh_layout(pool: &PgPool) { + seed_parents(pool).await; + for table in PARTITIONED_TABLES { + create_child( + pool, + table, + &format!("{table}_p_past"), + "MINVALUE", + "'2026-09-01'", + ) + .await; + create_child( + pool, + table, + &format!("{table}_p_future"), + "'2026-09-01'", + "MAXVALUE", + ) + .await; + } + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn fresh_layout_is_serving_safe_and_audit_is_read_only() { + let (pool, admin, schema) = scratch_pool().await; + seed_fresh_layout(&pool).await; + let before = catalog_snapshot(&pool).await; + let audit = audit_partition_catalog_at(&pool, 3, fixed_now()) + .await + .expect("audit"); + assert!(audit.serving_safe()); + assert!(audit.tables.iter().all(|table| { + table.anomalous_children() == 0 + && table.missing_trigger_count() == 0 + && table + .months + .iter() + .all(|month| month.kind == MonthCoverageKind::CoveredByCatchAll) + })); + assert_eq!(catalog_snapshot(&pool).await, before); + drop_schema(&admin, &schema).await; + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn repaired_layout_recognizes_bounds_not_catch_all_name() { + let (pool, admin, schema) = scratch_pool().await; + seed_parents(&pool).await; + for table in PARTITIONED_TABLES { + create_child( + &pool, + table, + &format!("{table}_p_past"), + "MINVALUE", + "'2026-07-01'", + ) + .await; + create_child( + &pool, + table, + &format!("{table}_july_repair"), + "'2026-07-01'", + "'2026-08-01'", + ) + .await; + create_child( + &pool, + table, + &format!("{table}_august_repair"), + "'2026-08-01'", + "'2026-09-01'", + ) + .await; + for month in 9..=12 { + let (end_year, end_month) = add_months(2026, month, 1).unwrap(); + create_child( + &pool, + table, + &format!("{table}_p2026_{month:02}"), + &format!("'2026-{month:02}-01'"), + &format!("'{end_year}-{end_month:02}-01'"), + ) + .await; + } + create_child( + &pool, + table, + &format!("{table}_p_future_next"), + "'2027-01-01'", + "MAXVALUE", + ) + .await; + } + let before = catalog_snapshot(&pool).await; + let audit = audit_partition_catalog_at(&pool, 3, fixed_now()) + .await + .expect("audit"); + assert!(audit.serving_safe()); + for table in &audit.tables { + assert_eq!( + table + .children + .iter() + .filter(|child| child.kind == PartitionChildKind::LegacyLeaf) + .count(), + 2 + ); + assert!(table + .children + .iter() + .any(|child| child.name.ends_with("p_future_next") + && child.kind == PartitionChildKind::CatchAll)); + assert!(table + .months + .iter() + .all(|month| month.kind == MonthCoverageKind::CoveredByMonthly)); + } + assert_eq!(catalog_snapshot(&pool).await, before); + drop_schema(&admin, &schema).await; + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn uncovered_months_are_created_with_trigger_parity() { + let (pool, admin, schema) = scratch_pool().await; + seed_parents(&pool).await; + for table in PARTITIONED_TABLES { + create_child( + &pool, + table, + &format!("{table}_p_past"), + "MINVALUE", + "'2026-09-01'", + ) + .await; + } + ensure_future_partitions_at(&pool, 1, true, fixed_now()) + .await + .expect("create gaps"); + let audit = audit_partition_catalog_at(&pool, 1, fixed_now()) + .await + .expect("audit"); + assert!(audit.serving_safe()); + assert!(audit.tables.iter().all(|table| table + .months + .iter() + .all(|month| month.kind == MonthCoverageKind::CoveredByMonthly) + && table.missing_trigger_count() == 0)); + drop_schema(&admin, &schema).await; + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn disabled_child_trigger_degrades_parity() { + let (pool, admin, schema) = scratch_pool().await; + seed_fresh_layout(&pool).await; + sqlx::query("ALTER TABLE events_p_future DISABLE TRIGGER partition_probe") + .execute(&pool) + .await + .expect("disable child trigger"); + + let audit = audit_partition_catalog_at(&pool, 3, fixed_now()) + .await + .expect("audit"); + let events = audit + .tables + .iter() + .find(|table| table.table == "events") + .expect("events audit"); + assert!(events.degraded()); + assert!(events.children.iter().any(|child| { + child.name == "events_p_future" && child.missing_triggers == ["partition_probe"] + })); + drop_schema(&admin, &schema).await; + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn always_enabled_parent_trigger_preserves_parity() { + let (pool, admin, schema) = scratch_pool().await; + seed_fresh_layout(&pool).await; + sqlx::query("ALTER TABLE events ENABLE ALWAYS TRIGGER partition_probe") + .execute(&pool) + .await + .expect("always-enable parent trigger"); + + let audit = audit_partition_catalog_at(&pool, 3, fixed_now()) + .await + .expect("audit"); + let events = audit + .tables + .iter() + .find(|table| table.table == "events") + .expect("events audit"); + assert_eq!(events.missing_trigger_count(), 0); + assert!(events + .children + .iter() + .all(|child| child.missing_triggers.is_empty())); + drop_schema(&admin, &schema).await; + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn disabled_nested_leaf_trigger_degrades_parity() { + let (pool, admin, schema) = scratch_pool().await; + seed_parents(&pool).await; + for (table, column) in [("events", "created_at"), ("delivery_log", "delivered_at")] { + let nested = format!("{table}_nested"); + create_nested_child( + &pool, + table, + &nested, + column, + "'2026-09-01'", + "'2026-10-01'", + ) + .await; + create_child( + &pool, + &nested, + &format!("{table}_nested_first"), + "'2026-09-01'", + "'2026-09-15'", + ) + .await; + create_child( + &pool, + &nested, + &format!("{table}_nested_second"), + "'2026-09-15'", + "'2026-10-01'", + ) + .await; + } + + let healthy = audit_partition_catalog_at(&pool, 0, fixed_now()) + .await + .expect("healthy nested audit"); + assert!(healthy + .tables + .iter() + .all(|table| table.missing_trigger_count() == 0)); + + sqlx::query("ALTER TABLE ONLY events_nested DISABLE TRIGGER partition_probe") + .execute(&pool) + .await + .expect("disable intermediate partitioned trigger"); + let intermediate_disabled = audit_partition_catalog_at(&pool, 0, fixed_now()) + .await + .expect("intermediate-disabled nested audit"); + assert!(intermediate_disabled + .tables + .iter() + .all(|table| table.missing_trigger_count() == 0)); + + sqlx::query("ALTER TABLE ONLY events_nested_first DISABLE TRIGGER partition_probe") + .execute(&pool) + .await + .expect("disable nested leaf trigger"); + + let audit = audit_partition_catalog_at(&pool, 0, fixed_now()) + .await + .expect("degraded nested audit"); + let events = audit + .tables + .iter() + .find(|table| table.table == "events") + .expect("events audit"); + assert_eq!(events.missing_trigger_count(), 1); + assert!(events.children.iter().any(|child| { + child.name == "events_nested" + && child.missing_triggers == ["events_nested_first:partition_probe"] + })); + drop_schema(&admin, &schema).await; + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn catch_all_skips_create_without_catalog_mutation() { + let (pool, admin, schema) = scratch_pool().await; + seed_fresh_layout(&pool).await; + let before = catalog_snapshot(&pool).await; + ensure_future_partitions_at(&pool, 3, true, fixed_now()) + .await + .expect("covered no-op"); + assert_eq!(catalog_snapshot(&pool).await, before); + drop_schema(&admin, &schema).await; + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn kill_switch_audits_but_does_not_create() { + let (pool, admin, schema) = scratch_pool().await; + seed_parents(&pool).await; + for table in PARTITIONED_TABLES { + create_child( + &pool, + table, + &format!("{table}_p_past"), + "MINVALUE", + "'2026-09-01'", + ) + .await; + } + let before = catalog_snapshot(&pool).await; + ensure_future_partitions_at(&pool, 1, false, fixed_now()) + .await + .expect("disabled create"); + assert_eq!(catalog_snapshot(&pool).await, before); + let audit = audit_partition_catalog_at(&pool, 1, fixed_now()) + .await + .expect("audit"); + assert!(audit.tables.iter().all(|table| !table.serving_safe + && table + .months + .iter() + .all(|month| month.kind == MonthCoverageKind::Uncovered))); + drop_schema(&admin, &schema).await; + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn nested_children_without_leaves_do_not_prove_coverage() { + let (pool, admin, schema) = scratch_pool().await; + seed_parents(&pool).await; + for (table, column) in [("events", "created_at"), ("delivery_log", "delivered_at")] { + create_nested_child( + &pool, + table, + &format!("{table}_nested"), + column, + "'2026-09-01'", + "'2026-10-01'", + ) + .await; + } + + let before = catalog_snapshot(&pool).await; + let audit = audit_partition_catalog_at(&pool, 0, fixed_now()) + .await + .expect("audit"); + assert!(!audit.serving_safe()); + assert!(audit.tables.iter().all(|table| { + table.degraded() + && table.anomalous_children() == 1 + && table.coverage_leaves.is_empty() + && table.months[0].kind == MonthCoverageKind::Uncovered + })); + assert_eq!(catalog_snapshot(&pool).await, before); + drop_schema(&admin, &schema).await; + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn nested_descendant_leaves_prove_coverage_but_remain_degraded() { + let (pool, admin, schema) = scratch_pool().await; + seed_parents(&pool).await; + for (table, column) in [("events", "created_at"), ("delivery_log", "delivered_at")] { + let nested = format!("{table}_nested"); + create_nested_child( + &pool, + table, + &nested, + column, + "'2026-09-01'", + "'2026-10-01'", + ) + .await; + create_child( + &pool, + &nested, + &format!("{table}_nested_leaf"), + "'2026-09-01'", + "'2026-10-01'", + ) + .await; + } + + let before = catalog_snapshot(&pool).await; + let audit = audit_partition_catalog_at(&pool, 0, fixed_now()) + .await + .expect("audit"); + assert!(audit.serving_safe()); + assert!(audit.tables.iter().all(|table| { + table.degraded() + && table.anomalous_children() == 1 + && table.coverage_leaves.len() == 1 + && table.coverage_leaves[0].nested + && table.months[0].kind == MonthCoverageKind::CoveredByMonthly + })); + assert_eq!(catalog_snapshot(&pool).await, before); + drop_schema(&admin, &schema).await; + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn nested_different_partition_key_does_not_prove_coverage() { + let (pool, admin, schema) = scratch_pool().await; + seed_parents(&pool).await; + for table in PARTITIONED_TABLES { + let nested = format!("{table}_nested"); + create_nested_child( + &pool, + table, + &nested, + "alternate_at", + "'2026-09-01'", + "'2026-10-01'", + ) + .await; + create_child( + &pool, + &nested, + &format!("{table}_nested_leaf"), + "'2026-09-01'", + "'2026-10-01'", + ) + .await; + } + + let audit = audit_partition_catalog_at(&pool, 0, fixed_now()) + .await + .expect("audit"); + assert!(!audit.serving_safe()); + assert!(audit.tables.iter().all(|table| { + table.degraded() + && table.anomalous_children() == 1 + && table.coverage_leaves.is_empty() + && table.months[0].kind == MonthCoverageKind::Uncovered + })); + drop_schema(&admin, &schema).await; + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn audit_pins_non_iso_session_rendering() { + let (pool, admin, schema) = scratch_pool().await; + seed_fresh_layout(&pool).await; + sqlx::query("SET DateStyle TO 'SQL, DMY'") + .execute(&pool) + .await + .expect("set non-ISO DateStyle"); + sqlx::query("SET TimeZone TO 'America/Los_Angeles'") + .execute(&pool) + .await + .expect("set non-UTC TimeZone"); + + let audit = audit_partition_catalog_at(&pool, 0, fixed_now()) + .await + .expect("audit"); + assert!(audit.serving_safe()); + assert!(audit + .tables + .iter() + .all(|table| !table.coverage_leaves.is_empty())); + let date_style: String = sqlx::query_scalar("SHOW DateStyle") + .fetch_one(&pool) + .await + .expect("show DateStyle"); + let time_zone: String = sqlx::query_scalar("SHOW TimeZone") + .fetch_one(&pool) + .await + .expect("show TimeZone"); + assert_eq!(date_style, "SQL, DMY"); + assert_eq!(time_zone, "America/Los_Angeles"); + drop_schema(&admin, &schema).await; + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn statement_triggers_are_not_part_of_leaf_parity() { + let (pool, admin, schema) = scratch_pool().await; + seed_parents(&pool).await; + for table in PARTITIONED_TABLES { + sqlx::query(sqlx::AssertSqlSafe(format!( + "CREATE TRIGGER statement_probe BEFORE INSERT ON {table} \ + FOR EACH STATEMENT EXECUTE FUNCTION partition_test_trigger()" + ))) + .execute(&pool) + .await + .expect("create parent statement trigger"); + create_child( + &pool, + table, + &format!("{table}_p_past"), + "MINVALUE", + "'2026-09-01'", + ) + .await; + create_child( + &pool, + table, + &format!("{table}_p2026_09"), + "'2026-09-01'", + "'2026-10-01'", + ) + .await; + } + + let audit = audit_partition_catalog_at(&pool, 0, fixed_now()) + .await + .expect("audit"); + assert!(audit.tables.iter().all(|table| { + !table.degraded() + && table.missing_trigger_count() == 0 + && table.extra_trigger_count() == 0 + })); + drop_schema(&admin, &schema).await; + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn canonical_name_with_wrong_bounds_is_a_real_creation_error() { + let (pool, admin, schema) = scratch_pool().await; + seed_parents(&pool).await; + for table in PARTITIONED_TABLES { + create_child( + &pool, + table, + &format!("{table}_p_past"), + "MINVALUE", + "'2026-09-01'", + ) + .await; + } + create_child( + &pool, + "events", + "events_p2026_09", + "'2026-10-01'", + "'2026-11-01'", + ) + .await; + + let result = ensure_future_partitions_at(&pool, 0, true, fixed_now()).await; + assert!( + matches!(result, Err(DbError::InvalidData(ref message)) if message.contains("mismatched bounds")), + "wrong-bound canonical name must not be counted as created: {result:?}" + ); + + let audit = audit_partition_catalog_at(&pool, 0, fixed_now()) + .await + .expect("audit"); + let events = audit + .tables + .iter() + .find(|table| table.table == "events") + .expect("events audit"); + assert_eq!(events.months[0].kind, MonthCoverageKind::Uncovered); + drop_schema(&admin, &schema).await; + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn anomalous_child_and_trigger_mismatch_degrade_without_aborting() { + let (pool, admin, schema) = scratch_pool().await; + seed_parents(&pool).await; + for table in PARTITIONED_TABLES { + create_child( + &pool, + table, + &format!("{table}_p_past"), + "MINVALUE", + "'2026-09-01'", + ) + .await; + create_child( + &pool, + table, + &format!("{table}_p2099_01"), + "'2026-09-01'", + "'2026-10-01'", + ) + .await; + create_child( + &pool, + table, + &format!("{table}_p_future"), + "'2026-10-01'", + "MAXVALUE", + ) + .await; + sqlx::query(sqlx::AssertSqlSafe(format!( + "CREATE TRIGGER child_only_probe BEFORE INSERT ON {table}_p2099_01 \ + FOR EACH ROW EXECUTE FUNCTION partition_test_trigger()" + ))) + .execute(&pool) + .await + .expect("create child-only trigger"); + } + let audit = audit_partition_catalog_at(&pool, 1, fixed_now()) + .await + .expect("audit"); + assert!(audit.serving_safe()); + assert!(audit.tables.iter().all(|table| { + table.degraded() + && table.anomalous_children() == 1 + && table.missing_trigger_count() == 0 + && table + .children + .iter() + .any(|child| child.extra_triggers == ["child_only_probe"]) + })); + drop_schema(&admin, &schema).await; + } } } diff --git a/crates/buzz-relay/src/config.rs b/crates/buzz-relay/src/config.rs index 037c6b1dd3d..6c9e5f9f855 100644 --- a/crates/buzz-relay/src/config.rs +++ b/crates/buzz-relay/src/config.rs @@ -133,6 +133,10 @@ pub struct Config { pub health_port: u16, /// TCP port for the Prometheus metrics exporter (`GET /metrics`). pub metrics_port: u16, + /// Interval between read-only partition catalog audits. + pub partition_audit_interval: Duration, + /// Whether the partition manager may create uncovered monthly partitions. + pub partition_manager_create_enabled: bool, /// When true, NIP-42 pubkey-only authentication (no API token) is /// restricted to pubkeys in the `pubkey_allowlist` table. Users with valid @@ -723,6 +727,18 @@ impl Config { .and_then(|v| v.parse().ok()) .unwrap_or(9102); + // Catalog-only and cheap, but clamp operator mistakes away from a hot + // loop or a multi-day observability gap. + let partition_audit_interval = Duration::from_secs( + std::env::var("BUZZ_PARTITION_AUDIT_INTERVAL_SECS") + .ok() + .and_then(|value| value.parse::().ok()) + .unwrap_or(15 * 60) + .clamp(60, 24 * 60 * 60), + ); + let partition_manager_create_enabled = + parse_bool("BUZZ_PARTITION_MANAGER_CREATE_ENABLED", true)?; + let s3_addressing_style = match std::env::var("BUZZ_S3_ADDRESSING_STYLE") { Ok(value) => value.parse().map_err(ConfigError::InvalidValue)?, Err(std::env::VarError::NotPresent) => buzz_media::config::S3AddressingStyle::default(), @@ -1010,6 +1026,8 @@ impl Config { uds_path, health_port, metrics_port, + partition_audit_interval, + partition_manager_create_enabled, pubkey_allowlist_enabled, require_relay_membership, huddle_audio_available, @@ -1119,6 +1137,8 @@ mod tests { assert!(config.max_connections > 0); assert!(config.send_buffer_size > 0); assert_eq!(config.max_frame_bytes, DEFAULT_MAX_FRAME_BYTES); + assert_eq!(config.partition_audit_interval, Duration::from_secs(900)); + assert!(config.partition_manager_create_enabled); assert!(config.slow_client_grace_limit > 0); assert!( !config.pubkey_allowlist_enabled, @@ -1159,6 +1179,65 @@ mod tests { ); } + #[test] + fn partition_manager_config_has_bounded_interval_and_create_kill_switch() { + let _guard = ENV_MUTEX.lock().unwrap(); + let previous_interval = std::env::var_os("BUZZ_PARTITION_AUDIT_INTERVAL_SECS"); + let previous_create = std::env::var_os("BUZZ_PARTITION_MANAGER_CREATE_ENABLED"); + + std::env::set_var("BUZZ_PARTITION_AUDIT_INTERVAL_SECS", "1"); + std::env::set_var("BUZZ_PARTITION_MANAGER_CREATE_ENABLED", "false"); + let minimum = Config::from_env().expect("minimum config"); + std::env::set_var("BUZZ_PARTITION_AUDIT_INTERVAL_SECS", "999999"); + let maximum = Config::from_env().expect("maximum config"); + + if let Some(value) = previous_interval { + std::env::set_var("BUZZ_PARTITION_AUDIT_INTERVAL_SECS", value); + } else { + std::env::remove_var("BUZZ_PARTITION_AUDIT_INTERVAL_SECS"); + } + if let Some(value) = previous_create { + std::env::set_var("BUZZ_PARTITION_MANAGER_CREATE_ENABLED", value); + } else { + std::env::remove_var("BUZZ_PARTITION_MANAGER_CREATE_ENABLED"); + } + + assert_eq!(minimum.partition_audit_interval, Duration::from_secs(60)); + assert!(!minimum.partition_manager_create_enabled); + assert_eq!( + maximum.partition_audit_interval, + Duration::from_secs(24 * 60 * 60) + ); + assert!(!maximum.partition_manager_create_enabled); + } + + #[test] + fn partition_manager_create_kill_switch_parses_false_values_strictly() { + let _guard = ENV_MUTEX.lock().unwrap(); + let previous = std::env::var_os("BUZZ_PARTITION_MANAGER_CREATE_ENABLED"); + + for value in ["FALSE", "off", " false "] { + std::env::set_var("BUZZ_PARTITION_MANAGER_CREATE_ENABLED", value); + let config = Config::from_env().expect("recognized false value"); + assert!(!config.partition_manager_create_enabled, "value: {value:?}"); + } + + std::env::set_var("BUZZ_PARTITION_MANAGER_CREATE_ENABLED", "disable-maybe"); + let invalid = Config::from_env(); + + if let Some(value) = previous { + std::env::set_var("BUZZ_PARTITION_MANAGER_CREATE_ENABLED", value); + } else { + std::env::remove_var("BUZZ_PARTITION_MANAGER_CREATE_ENABLED"); + } + + assert!(matches!( + invalid, + Err(ConfigError::InvalidValue(ref message)) + if message.contains("BUZZ_PARTITION_MANAGER_CREATE_ENABLED") + )); + } + #[test] fn s3_addressing_style_env_accepts_virtual_and_rejects_invalid_values() { let _guard = ENV_MUTEX.lock().unwrap(); diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index 566b684f830..a2f7d2f3270 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -83,6 +83,23 @@ impl EmissionScope { const USAGE_METRICS_LOCK_KEY: i64 = 0x4255_5A5A_4D45_5452; +fn partition_audit_interval(period: std::time::Duration) -> tokio::time::Interval { + let mut interval = tokio::time::interval(period); + interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + interval +} + +async fn wait_for_partition_audit_tick( + interval: &mut tokio::time::Interval, + delay_first_audit: &mut bool, +) { + if *delay_first_audit { + interval.tick().await; + *delay_first_audit = false; + } + interval.tick().await; +} + #[tokio::main] async fn main() -> anyhow::Result<()> { // Install the ring CryptoProvider for rustls. Required before any rustls @@ -154,12 +171,14 @@ async fn main() -> anyhow::Result<()> { ); let usage_interval_secs = usage_metrics_interval_secs(); - let usage_idle_timeout_secs = usage_metrics_idle_timeout_secs(usage_interval_secs); - relay_metrics::install(config.metrics_port, usage_idle_timeout_secs); + let metrics_refresh_interval_secs = + usage_interval_secs.max(config.partition_audit_interval.as_secs()); + let gauge_idle_timeout_secs = usage_metrics_idle_timeout_secs(metrics_refresh_interval_secs); + relay_metrics::install(config.metrics_port, gauge_idle_timeout_secs); metrics::gauge!("buzz_audit_enabled").set(if config.audit_enabled { 1.0 } else { 0.0 }); info!( port = config.metrics_port, - idle_timeout_secs = usage_idle_timeout_secs, + idle_timeout_secs = gauge_idle_timeout_secs, "Prometheus metrics exporter started" ); @@ -197,9 +216,22 @@ async fn main() -> anyhow::Result<()> { info!("Skipping database migrations because BUZZ_AUTO_MIGRATE is not enabled"); } - if let Err(e) = db.ensure_future_partitions(3).await { - error!("Failed to ensure partitions: {e}"); - } + let startup_partition_audit = match db + .ensure_future_partitions(3, config.partition_manager_create_enabled) + .await + { + Ok(audit) => Some(audit), + Err(error) => { + error!(%error, "Failed to ensure partitions"); + match db.audit_partitions(3).await { + Ok(audit) => Some(audit), + Err(error) => { + error!(%error, "Initial partition catalog audit failed"); + None + } + } + } + }; db.validate_deletion_serving_catalog().await.map_err(|e| { error!("Community deletion serving-fence validation failed: {e}"); @@ -466,6 +498,33 @@ async fn main() -> anyhow::Result<()> { media_storage, ); let state = Arc::new(app_state); + let mut delay_first_partition_audit = startup_partition_audit.is_some(); + if let Some(audit) = startup_partition_audit { + state.record_partition_audit(audit); + } + + // The periodic path is deliberately read-only. PR2 owns any catch-all + // advance; this task only refreshes metrics and the cached readiness term. + // Refresh failures retain the last-known-good audit by design; operators + // should alert when buzz_partition_audit_last_success_timestamp_seconds is stale. + { + let partition_state = Arc::clone(&state); + let audit_interval = state.config.partition_audit_interval; + tokio::spawn(async move { + let mut interval = partition_audit_interval(audit_interval); + loop { + wait_for_partition_audit_tick(&mut interval, &mut delay_first_partition_audit) + .await; + match partition_state.db.audit_partitions(3).await { + 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") + } + } + } + }); + } // Inter-relay mesh (BUZZ_MESH seam). `boot_mesh` returns None when the // kill switch is off — nothing is bound, published, or spawned, so the @@ -2037,7 +2096,8 @@ mod tests { use super::{ buzz_auto_migrate_enabled, dropped_in_memory_keys, idle_timeout_secs, - refresh_legacy_active_gauge_recency, run_periodic_until_cancelled, EmissionScope, + partition_audit_interval, refresh_legacy_active_gauge_recency, + run_periodic_until_cancelled, wait_for_partition_audit_tick, EmissionScope, InMemoryMetricKey, }; use metrics::GaugeFn; @@ -2071,6 +2131,26 @@ mod tests { assert!(tick_count.load(std::sync::atomic::Ordering::Relaxed) <= 1); } + #[tokio::test(start_paused = true)] + async fn partition_audit_first_tick_matches_startup_cache_state() { + let period = Duration::from_secs(60); + + let immediate_start = tokio::time::Instant::now(); + let mut immediate_interval = partition_audit_interval(period); + let mut delay_after_failed_startup = false; + wait_for_partition_audit_tick(&mut immediate_interval, &mut delay_after_failed_startup) + .await; + assert_eq!(tokio::time::Instant::now(), immediate_start); + + let delayed_start = tokio::time::Instant::now(); + let mut delayed_interval = partition_audit_interval(period); + let mut delay_after_successful_startup = true; + wait_for_partition_audit_tick(&mut delayed_interval, &mut delay_after_successful_startup) + .await; + assert_eq!(tokio::time::Instant::now() - delayed_start, period); + assert!(!delay_after_successful_startup); + } + #[test] fn buzz_auto_migrate_is_opt_in() { assert!(!buzz_auto_migrate_enabled(None)); @@ -2154,8 +2234,9 @@ mod tests { } #[test] - fn test_idle_timeout_is_at_least_three_usage_intervals() { + fn test_idle_timeout_is_at_least_three_metric_refresh_intervals() { assert_eq!(idle_timeout_secs(None, 300), 900); assert_eq!(idle_timeout_secs(Some(10), 1_000), 3_000); + assert_eq!(idle_timeout_secs(None, 86_400), 259_200); } } diff --git a/crates/buzz-relay/src/mesh_boot.rs b/crates/buzz-relay/src/mesh_boot.rs index cd7c427c72e..0e8f94b9f67 100644 --- a/crates/buzz-relay/src/mesh_boot.rs +++ b/crates/buzz-relay/src/mesh_boot.rs @@ -462,8 +462,8 @@ pub async fn boot_mesh( .map_err(|e| anyhow::anyhow!("mesh ready-registry publish failed: {e}"))?; tracing::info!(runtime_id = %runtime_id, "mesh ready record published"); - // Readiness-gated heartbeat: publishes while the relay would pass - // readiness, clears the record on ready→not-ready and on shutdown. + // The heartbeat is shutdown-gated only. Readiness terms deliberately do + // not gate mesh publication; shutdown clears the record. let hb_flag = Arc::clone(&shutting_down); buzz_relay_mesh::runtime::spawn_registry_heartbeat( registry.clone(), @@ -539,7 +539,7 @@ mod tests { let db = buzz_db::Db::from_pool( sqlx::postgres::PgPoolOptions::new() .max_connections(1) - .connect_lazy("postgres://unused:unused@127.0.0.1:1/unused") + .connect_lazy("postgres://unused:unused@127.0.0.1:1/unused") // sadscan:disable np.postgres.1 .expect("lazy database pool"), ); let handle = boot_mesh(&config, pool, db, &keys, Arc::new(AtomicBool::new(false))) diff --git a/crates/buzz-relay/src/router.rs b/crates/buzz-relay/src/router.rs index 1dce66e91e4..aca3196c2da 100644 --- a/crates/buzz-relay/src/router.rs +++ b/crates/buzz-relay/src/router.rs @@ -371,7 +371,7 @@ async fn liveness_handler() -> impl IntoResponse { (StatusCode::OK, "ok") } -/// Readiness probe — checks shutdown flag, Postgres, and Redis connectivity. +/// Readiness probe — checks shutdown, dependencies, and cached serving catalogs. async fn readiness_handler(State(state): State>) -> impl IntoResponse { use std::time::Duration; @@ -383,6 +383,7 @@ async fn readiness_handler(State(state): State>) -> impl IntoRespo .into_response(); } + let partition_catalog_ok = state.partition_serving_safe(); let check = async { let (pg_ok, redis_ok, deletion_catalog_ok) = tokio::join!( state.db.ping(), @@ -397,7 +398,7 @@ async fn readiness_handler(State(state): State>) -> impl IntoRespo .await .unwrap_or((false, false, false)); - if pg_ok && redis_ok && deletion_catalog_ok { + if readiness_checks_pass(pg_ok, redis_ok, deletion_catalog_ok, partition_catalog_ok) { (StatusCode::OK, Json(json!({"status": "ready"}))).into_response() } else { ( @@ -406,13 +407,23 @@ async fn readiness_handler(State(state): State>) -> impl IntoRespo "status": "not_ready", "postgres": pg_ok, "redis": redis_ok, - "deletion_catalog": deletion_catalog_ok + "deletion_catalog": deletion_catalog_ok, + "partition_catalog": partition_catalog_ok })), ) .into_response() } } +fn readiness_checks_pass( + postgres: bool, + redis: bool, + deletion_catalog: bool, + partition_catalog: bool, +) -> bool { + postgres && redis && deletion_catalog && partition_catalog +} + /// Status endpoint — service name, version, uptime. async fn status_handler(State(state): State>) -> impl IntoResponse { let uptime_secs = state.started_at.elapsed().as_secs(); @@ -476,6 +487,13 @@ mod tests { use super::*; + #[test] + fn readiness_requires_a_successful_serving_safe_partition_audit() { + assert!(readiness_checks_pass(true, true, true, true)); + assert!(!readiness_checks_pass(true, true, true, false)); + assert!(!readiness_checks_pass(false, true, true, true)); + } + #[test] fn invite_landing_path_requires_exactly_one_nonempty_code_segment() { assert!(is_invite_landing_path("/invite/payload.mac")); diff --git a/crates/buzz-relay/src/state.rs b/crates/buzz-relay/src/state.rs index 2f544e188c0..84e3e4780d3 100644 --- a/crates/buzz-relay/src/state.rs +++ b/crates/buzz-relay/src/state.rs @@ -713,6 +713,8 @@ pub struct AppState { pub audio_rooms: Arc, /// Set to `true` on SIGTERM — readiness probe returns 503. pub shutting_down: Arc, + /// Last successful read-only partition catalog audit. `None` fails readiness. + pub partition_audit: Arc>>, /// Process start time — used by `/_status` endpoint. pub started_at: Instant, /// Shared, community-scoped NIP-98 replay prevention. @@ -910,6 +912,7 @@ impl AppState { git_pack_cache, audio_rooms: Arc::new(AudioRoomManager::new()), shutting_down: Arc::new(AtomicBool::new(false)), + partition_audit: Arc::new(std::sync::RwLock::new(None)), started_at: Instant::now(), nip98_replay, admission_rate_limiter, @@ -956,6 +959,35 @@ impl AppState { self.mesh.get() } + /// Publish a successful partition audit for cached readiness checks. + pub fn record_partition_audit(&self, audit: buzz_db::partition::PartitionAudit) { + match self.partition_audit.write() { + Ok(mut cached) => *cached = Some(audit), + Err(poisoned) => *poisoned.into_inner() = Some(audit), + } + } + + /// Whether a successful cached audit proves every managed parent serves now. + /// + /// PR1 deliberately gates readiness on `now()` only. Ingestion accepts + /// timestamps within ±900 seconds, so a month boundary with no adjacent + /// partition remains a known accepted edge until PR2's bounded partition + /// advance closes it operationally. + /// Periodic refresh failures retain the last-known-good audit by design; + /// operators should alert on staleness of + /// `buzz_partition_audit_last_success_timestamp_seconds`. + /// This gates HTTP readiness only: catalog safety is cluster-global, so + /// removing every pod from mesh would not provide a safe reroute target. + pub fn partition_serving_safe(&self) -> bool { + let cached = match self.partition_audit.read() { + Ok(cached) => cached, + Err(poisoned) => poisoned.into_inner(), + }; + cached + .as_ref() + .is_some_and(|audit| audit.serving_safe_at(chrono::Utc::now())) + } + /// Record an event ID as locally-published for dedup, scoped to the /// community it was fanned out in. Called before Redis publish so the /// multi-node consumer can skip the echo for *this* community only — a From 94af7d3308ffd07e3f3efc8c0061aa36baa42d11 Mon Sep 17 00:00:00 2001 From: coder 0 Date: Fri, 21 Aug 2026 19:33:57 -0400 Subject: [PATCH 2/2] fix(db): audit nested leaf extra triggers Audit child-only row triggers on every routable descendant leaf, qualifying nested leaves the same way missing triggers are qualified. Add a Postgres regression test that fails when extra-trigger parity only looks at the immediate root child. Co-authored-by: coder 0 Signed-off-by: coder 0 --- crates/buzz-db/src/partition.rs | 85 +++++++++++++++++++++++++++------ 1 file changed, 71 insertions(+), 14 deletions(-) diff --git a/crates/buzz-db/src/partition.rs b/crates/buzz-db/src/partition.rs index 3582158695e..e2f1f183895 100644 --- a/crates/buzz-db/src/partition.rs +++ b/crates/buzz-db/src/partition.rs @@ -75,7 +75,8 @@ pub struct PartitionChildAudit { /// Parent trigger names absent from this child or a routable descendant leaf. /// Nested-leaf entries are qualified as `{leaf}:{trigger}`. pub missing_triggers: Vec, - /// Child trigger names absent from the parent. + /// Child-only row trigger names absent from the parent. + /// Nested-leaf entries are qualified as `{leaf}:{trigger}`. pub extra_triggers: Vec, } @@ -493,9 +494,6 @@ async fn audit_table_on( let descendant_triggers = trigger_metadata_for_descendants(connection, table).await?; let mut child_audits = Vec::with_capacity(children.len()); for child in children { - let triggers = descendant_triggers - .get(&child.name) - .map(|relation| &relation.triggers); let mut missing_triggers = Vec::new(); let routable_leaves: Vec<_> = descendant_triggers .values() @@ -507,7 +505,7 @@ async fn audit_table_on( && relation.relation_kind != "p")) }) .collect(); - for leaf in routable_leaves { + for leaf in &routable_leaves { for (name, parent_oid) in &parent_triggers { let present = trigger_lineage_reaches_parent(leaf, name, *parent_oid, &descendant_triggers) @@ -525,15 +523,20 @@ async fn audit_table_on( } } missing_triggers.sort(); - let mut extra_triggers: Vec<_> = triggers - .map(|triggers| { - triggers - .keys() - .filter(|name| !parent_triggers.contains_key(*name)) - .cloned() - .collect() - }) - .unwrap_or_default(); + let mut extra_triggers = Vec::new(); + for leaf in routable_leaves { + for name in leaf + .triggers + .keys() + .filter(|name| !parent_triggers.contains_key(*name)) + { + if leaf.depth == 0 { + extra_triggers.push(name.clone()); + } else { + extra_triggers.push(format!("{}:{name}", leaf.name)); + } + } + } extra_triggers.sort(); child_audits.push(PartitionChildAudit { name: child.name, @@ -1595,6 +1598,60 @@ mod tests { drop_schema(&admin, &schema).await; } + #[tokio::test] + #[ignore = "requires Postgres"] + async fn child_only_nested_leaf_trigger_degrades_parity() { + let (pool, admin, schema) = scratch_pool().await; + seed_parents(&pool).await; + create_nested_child( + &pool, + "events", + "events_nested", + "created_at", + "'2026-09-01'", + "'2026-10-01'", + ) + .await; + create_child( + &pool, + "events_nested", + "events_nested_first", + "'2026-09-01'", + "'2026-09-15'", + ) + .await; + create_child( + &pool, + "events_nested", + "events_nested_second", + "'2026-09-15'", + "'2026-10-01'", + ) + .await; + sqlx::query( + "CREATE TRIGGER child_only_probe BEFORE INSERT ON events_nested_first \ + FOR EACH ROW EXECUTE FUNCTION partition_test_trigger()", + ) + .execute(&pool) + .await + .expect("create nested child-only trigger"); + + let audit = audit_partition_catalog_at(&pool, 0, fixed_now()) + .await + .expect("audit"); + let events = audit + .tables + .iter() + .find(|table| table.table == "events") + .expect("events audit"); + assert_eq!(events.extra_trigger_count(), 1); + assert!(events.children.iter().any(|child| { + child.name == "events_nested" + && child.extra_triggers == ["events_nested_first:child_only_probe"] + })); + drop_schema(&admin, &schema).await; + } + #[tokio::test] #[ignore = "requires Postgres"] async fn catch_all_skips_create_without_catalog_mutation() {