jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-10-27 04:30:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(relay): partition audit observability follow-ups (#7994)

Browse Source 
> [!IMPORTANT]
> **The alert rules changed in c3a314e71.** If you reviewed the alert
guidance earlier, please look again.
> - **The two range-based rules now apply only to live pods.** A range
selector such as `[30m]` keeps reading a terminated pod's samples until
they leave the window. So `BuzzPartitionAuditNotRunning` could fire for
about 10 minutes after normal pod turnover, and
`BuzzPartitionAuditFailing` could keep firing for up to an hour after
the pod was gone. Each rule now adds `and <counter>`, which requires the
counter to be exported right now:
>   ```yaml
>   expr: |
> sum without (outcome) (increase(buzz_partition_audit_runs_total[30m]))
== 0
>       and sum without (outcome) (buzz_partition_audit_runs_total)
>   ```
>   ```yaml
>   expr: |
>     increase(buzz_partition_audit_failures_total[1h]) >= 2
>       and buzz_partition_audit_failures_total
>   ```
> - **The audit counters start at 0.** When the relay starts, it exports
every `buzz_partition_audit_runs_total{table, outcome}` series and
`buzz_partition_audit_failures_total` at 0, before the first audit.
`increase()` ignores a new series' first sample. Before this change, it
could miss the first periodic failure, or read a table's first
`degraded` run as a stopped loop. This change is also what makes the
`and <counter>` gate work: the counters now exist for a relay's whole
life.
>
> The doc's old claim that pod turnover can't leave an alert firing was
true only of the instant-selector stale rule. The doc now explains the
difference.

## Why

These are the observability follow-ups deferred from #6515 ("fix(db):
audit partition catalog before creation"):

- **Timestamp-age alerts go silent** ([review
comment](https://github.com/block/buzz/pull/6515#discussion_r4127212141)).
Partition gauges are set only when a table audits successfully, and the
exporter evicts idle gauges after 2,700 s by default. After sustained
failures, `buzz_partition_audit_last_success_timestamp_seconds`
disappears. A table that has never audited successfully never has it. An
alert on `time() - last_success` then has no input.
- **Startup fallback failures are not counted** ([review
comment](https://github.com/block/buzz/pull/6515#discussion_r4127212158)).
When the startup fallback audit fails, it logs an error but does not
increment `buzz_partition_audit_failures_total`. Periodic failures do.
- **Repaired environments are always `degraded`**.
`PartitionTableAudit::degraded()` counts every legacy leaf, even one
wholly in the past. The `bb-block-staging` audit of the repaired catalog
reports `degraded` only because of `events_p2026_07_08_legacy`
`[2026-07-01, 2026-09-01)`. With this rule, every repaired environment
stays `degraded`, so nobody can alert on the outcome.
- **Partition config tests can see a poisoned env** ([review
comment](https://github.com/block/buzz/pull/6515#discussion_r4137416297)).
Two partition config tests take only `ENV_MUTEX`. A NIP-FI test running
in the same process can make them see its deliberately invalid
environment.

Partition health is still diagnostic only, through
`/_status.partition_catalog` and metrics. Nothing here affects
`/_readiness`. Handling concurrent creation (#6515 item 4) is out of
scope and stays with the catch-all advancement work.

## What

- **Success timestamp survives failures.** Each failed table audit
refreshes `buzz_partition_audit_last_success_timestamp_seconds` with
`increment(0.0)`, which leaves the value unchanged. It still resets the
exporter's idle tracking, the same mechanism as
`refresh_legacy_active_gauge_recency`. Sustained failures therefore keep
the last success exported and its age rising. A table that has never
succeeded exports `0`. A failed audit still never makes the last success
look fresh.
- **Alert guidance.** New `docs/partition-catalog-monitoring.md` covers
the metrics, how long each series lives, and the audit outcomes. It
notes that `runs_total` has no `unsafe` outcome. It gives three alert
rules:
- timestamp age, which also covers never-successful tables and sustained
failures;
- `sum without (outcome)
(increase(buzz_partition_audit_runs_total[30m])) == 0` for a stopped
audit loop. Counters are never idle-evicted, and this rule fires before
the stopped loop's timestamp is evicted, so there is no false resolve;
  - an aggregate failure-counter rule.

Both range-based rules also require their counter to be exported right
now (`and <counter>`), so a terminated pod's leftover samples can't
raise or hold an alert.
- **Zero counter baselines.** `try_install` exports each
`buzz_partition_audit_runs_total{table, outcome}` series (`ok`,
`degraded`, and `error` for both tables) and
`buzz_partition_audit_failures_total` at 0 before the startup audit, the
same way the auth series already start at 0. A failure in the startup
audit itself can still happen before the first scrape. The doc says so;
the stale rule covers that case.
- **Failure counter.** `buzz_partition_audit_failures_total` now counts
failed explicit audit attempts: the startup fallback audit and each
periodic audit. Both paths go through one helper,
`counted_partition_audit`. A failed `ensure_future_partitions` call,
such as a DDL error that the fallback then recovers, is not counted by
itself. A single attempt never counts twice.
- **Historical legacy leaves.** A legacy leaf is historical when its
finite upper bound is at or before the current month start. It is still
reported under `children` as `legacy_leaf`, but it no longer degrades
the table. A legacy leaf that still covers the current month or any
lookahead month stays degraded. All other degradation causes are
unchanged. The verdict now lives in `PartitionAuditReport::outcome()`,
which `buzz-admin partition-audit` uses. The JSON outcome names and exit
codes are the same as before.
- **Test hygiene.** Both partition config tests now use `env_guards()`.

## How it was tested

- `cargo fmt --all -- --check` and `cargo clippy -p buzz-db -p
buzz-relay -p buzz-admin --all-targets -- -D warnings`.
- New unit tests:
- The repaired incident layout with an injected clock: `degraded` in
August 2026 and `ok` on 2026-09-01T00:00Z, when the upper bound equals
the current month start, and on 2026-09-15. The legacy leaf is still in
the serialized report. The `outcome` label on the run counter follows
the verdict.
- A table test of the legacy-leaf predicate: the upper bound before, at,
and one second after the month start; lookahead and beyond-horizon
leaves; historical leaves with trigger drift; anomalous and `DEFAULT`
children; no audited month.
- Outcome precedence (error, then unsafe, then degraded) and the
`buzz-admin` exit-code mapping.
- Startup: a failed fallback counts once, a successful fallback after a
DDL failure counts nothing, and a successful ensure never runs the
fallback. Each periodic failure counts once. Against an unreachable
database, startup counts one aggregate failure while each table still
records two `error` runs.
- On the production exporter, never-successful audits leave run counters
and a `0` success timestamp for both tables.
- On the production exporter, all six run-counter series and the failure
counter read 0 before any audit. After one failed audit, `error` reads 1
and `ok`/`degraded` stay at 0. Removing the initialization fails this
test.
- PostgreSQL lane (`scripts/postgres-test-run.sh`) with new or extended
tests:
- The repaired layout is not degraded in September. The same catalog is
degraded in August.
- With a 1 s idle timeout, a failed audit leaves the success timestamp
unchanged, and sustained failures keep it exported past the idle
timeout. When audits stop, the timestamp is evicted but the run counters
remain. The next success exports a fresh timestamp.
- Mutation check: removing the failure-path refresh fails both alert
fixtures.
- These fixtures check the emitted series. They do not evaluate the
rules themselves, including the `and <counter>` gate. `promtool` isn't
available locally or in CI, and the doc points to `promtool test rules`
for that.
- `scripts/test-postgres-test-discovery.sh`.
- `config::` and `nip_fi_config::` tests run together in one process
five times.

## Related

This PR conflicts textually with #7993 (catch-all advancement) in
`crates/buzz-db/src/store/partition.rs` and
`crates/buzz-relay/src/main.rs`. Whichever PR merges second will be
rebased onto the other.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Signed-off-by: Ian Oberst <ioberst@block.xyz>
Co-authored-by: Ian Oberst <ioberst@block.xyz>
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Ian Oberst
2026-10-13 13:57:54 -07:00
committed by GitHub
parent 5dc693aeaf
commit 85b7331784
11 changed files with 1156 additions and 69 deletions
Download Patch File Download Diff File Expand all files Collapse all files

45
crates/buzz-admin/src/main.rs
View File

@@ -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<i32> {
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<i32> {
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<i32> {
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}");
}
}
}

457
crates/buzz-db/src/store/partition.rs
View File

@@ -186,7 +186,7 @@ pub struct PartitionTableAudit {
pub children: Vec<PartitionChildAudit>,
/// Cached effective bounds for every routable leaf in the catalog tree.
pub coverage_leaves: Vec<PartitionLeafAudit>,
/// Coverage of the current month and the requested future months.
/// Coverage of the current month, first, and the requested future months.
pub months: Vec<MonthCoverage>,
/// 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<PartitionAuditError>,
}
/// 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<PartitionAudit> {
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::<Vec<_>>();
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<Utc>,
) -> Result<Vec<MonthCoverage>> {
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<i32> {
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<Utc>) {
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<Utc>) -> 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::<Vec<_>>();
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::<Vec<_>>();
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::<Vec<_>>(),
[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::<Vec<_>>()
};
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;
}

4
crates/buzz-relay/src/config.rs
View File

@@ -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 "] {

229
crates/buzz-relay/src/main.rs
View File

@@ -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<Output = buzz_db::Result<PartitionAudit>>,
) -> buzz_db::Result<PartitionAudit> {
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<F>(
maintain: impl std::future::Future<Output = buzz_db::Result<PartitionAudit>>,
fallback_audit: impl FnOnce() -> F,
) -> Option<PartitionAudit>
where
F: std::future::Future<Output = buzz_db::Result<PartitionAudit>>,
{
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<buzz_db::partition::PartitionAudit> {
Ok(buzz_db::partition::PartitionAudit {
audited_at: chrono::Utc::now(),
tables: Vec::new(),
})
}
fn partition_audit_err(detail: &str) -> buzz_db::Result<buzz_db::partition::PartitionAudit> {
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<F: std::future::Future>(
future: F,
) -> (F::Output, Vec<CounterSample>) {
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));

172
crates/buzz-relay/src/metrics.rs
View File

@@ -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<String, f64> {
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<String> {
samples(scrape, metric).into_keys().collect()
}
fn managed_tables() -> Vec<String> {
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::<BTreeMap<_, _>>();
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::<Vec<_>>(), 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());
}
}
}

8
crates/buzz-relay/src/state.rs
View File

@@ -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<buzz_db::partition::PartitionAudit> {
let cached = match self.partition_audit.read() {
Ok(cached) => cached,