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

feat(relay): time post-metrics-bind startup steps (#8132)

Browse Source 
## Summary

Relay pods on large deployments can take over 5 minutes to open their
health port, and right now nothing shows which startup step is slow.
Startup timing ends at `metrics_bind`, and the slow work all happens
after it.

This PR times every startup step from Postgres connect through public
listener bind. It does not change startup behavior.

- Each step logs `Startup phase started` and `Startup phase finished`,
with `phase`, `elapsed_ms`, and `status` (`succeeded` / `degraded` /
`failed`, the same values `buzz_process_lifecycle` records use).
- Gauge `buzz_startup_phase_current{phase}` is `1` while a step runs and
`0` after it ends. A pod that is still starting shows exactly which step
it is on.
- Gauge `buzz_startup_phase_seconds{phase}` records each finished step's
duration.
- The startup NIP-43 membership reconcile logs `NIP-43 startup
reconciliation progress` every 5,000 communities, with `done`, `total`,
`republished`, `failed`, and `elapsed_ms`.

These are gauges rather than counters on purpose. Each value is written
once per boot, and a monotonic-counter scrape drops a series' first
sample, so a one-shot counter never shows up.

## Contract change

`deploy/charts/buzz/README.md` ("Early-startup telemetry contract") said
the early lifecycle phases are log-only. That still holds: `crypto_init`
through `metrics_bind` are unchanged and emit no metrics.

The README now also documents the post-bind steps. The guard in
`boot_lifecycle.rs` now allows exactly the two `buzz_startup_phase_*`
families. It still rejects any other startup, boot, or lifecycle family,
and it also rejects any exported label value that names an early phase.

## Design

- `crates/buzz-relay/src/startup_steps.rs` adds a closed `StartupStep`
vocabulary (18 steps) and a `StepTimer` guard.
- A step ends exactly once:
  - `finish()` → `succeeded`
  - `degrade()` then `finish()` → `degraded` (a non-fatal error branch)
- dropped without finishing → `failed` (covers `?` early returns and
panics)
- The NIP-43 and large-roster reconciles now return `ReconcileSummary {
repaired, failed }`. A sweep where any item fails ends its step
`degraded`, and the final log line includes the `failed` count. NIP-FI
warm ends `degraded` when any issuer did not warm. The deprecated
`reconcile_nip43_membership_snapshots` wrapper still returns `repaired`.
- Steps that are disabled by config (migrate, audit connect, NIP-FI
warm, git probe, NIP-43 reconcile) are not timed, so their series never
appear.
- The gauges are set once, so they age out after the existing gauge idle
timeout (2700 s with default intervals; logged as `idle_timeout_secs`).
A step that runs longer than that loses its `_current` series. The
README says so.
- Progress logging runs only for the `Bootstrap` reconcile. The 60 s
maintenance sweep stays quiet.

## What got simpler

Little got simpler in the code: this mostly adds instrumentation. One CI
gap closed along the way: the 54 `config` tests and the 9 non-ignored
`boot_lifecycle` tests, which ran in no CI lane, now run in `test-unit`.
The instrumentation is scoped as small as I could make it, and it is the
evidence needed for the follow-up that should delete code: moving the
per-community NIP-43 reconcile off the pre-bind path, after which the
900 s startup-probe override can go back to the chart default.

## Testing

- `startup_steps` unit tests (`DebuggingRecorder`):
  - a running step is current and has no duration
  - a finished step clears current and records its elapsed time
  - dropped and degraded steps clear current and record a duration
- log capture: `finish`, `degrade`+`finish`, drop, and panic each emit
exactly one `Startup phase finished` record with the right `status`.
Mutating `degrade()` to a no-op and removing the ends-once guard makes
these tests fail.
- the vocabulary is unique and shares no names with the early lifecycle
phases
- Tests that capture logs or check callsite interest now share
`test_support::tracing_dispatch_lock()`. These are the `startup_steps`
helpers, two `telemetry` tests, and the `config` and `bridge`
log-capture helpers. tracing-core caches each callsite's interest across
the whole process. Without the lock, the new panic test lost its record
in 60 of 150 parallel runs of `telemetry:: startup_steps:: config::`.
With the lock it failed 0 of 300. Removing the lock from `gauges()`
brings the failures back (53 of 100 runs).
- `nip43_progress_logs_every_interval_inside_the_startup_sweep_only`
checks the cadence at 79,942 communities (15 progress lines, none at the
start or end) and table-tests that the maintenance sweep never logs
progress.
-
`boot_lifecycle::successful_main_emits_complete_lifecycle_and_reports_current_startup_step`
boots the real relay binary against a TCP peer that never speaks
Postgres. It asserts that `/metrics` shows
`buzz_startup_phase_current{phase="db_connect"} 1` and no finished-step
durations, and that no early phase leaks into metrics.

- **CI routing.** `just test-unit` picks relay tests by hand-maintained
filter. Before this PR, CI's filters selected none of the
`startup_steps`, `telemetry`, `config` or non-ignored `boot_lifecycle`
tests above; the NIP-43 cadence test is picked up by the existing
`handlers::side_effects::tests::` clause. Its infra-free relay stanza
now selects `startup_steps::tests::`, `telemetry::tests::` and
`config::tests::` (anchored so `nip_fi_config` is not selected twice),
and a new `cargo nextest run -p buzz-relay --test boot_lifecycle` line
runs the non-ignored `boot_lifecycle` tests (they spawn the real binary
and need no Postgres or Redis; the Postgres cases stay `#[ignore]`d).
`scripts/run-tests.sh` mirrors both, running `telemetry` and `config`
single-threaded because they mutate process env.

### Manual check

1. `just relay` (or run `buzz-relay` against local Postgres and Redis).
2. Once the relay is up: `curl -s localhost:9102/metrics | grep
buzz_startup_phase`. Every step that ran has a `_seconds` value, and
every `_current` value is 0.
3. In the relay logs, `"Startup phase finished"` lines show each step's
`elapsed_ms`.

---------

Signed-off-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz>
Co-authored-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz>
This commit is contained in:
Luke Tornquist
2026-10-07 10:15:43 -04:00
committed by GitHub
parent bd1ff00e48
commit e097c42996
13 changed files with 655 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
Justfile
View File

@@ -481,8 +481,12 @@ test-unit:
# (JWKS source + refresh-loop composition), which `--lib` cannot reach
# because they live in the binary target; the nested
# `tests::postgres_tests::` stays in the PostgreSQL lane.
# The fourth clause adds the startup step timers (startup_steps) and the
# env- and log-capture tests in telemetry and config. All three modules
# are infra-free; config parses env into a struct and never connects.
# `^config::` is anchored so it does not also select nip_fi_config.
cargo nextest run -p buzz-relay --lib --bin buzz-relay \
-E 'test(/^api::admin::/) + test(/^handlers::channel_authz::/) + test(/^handlers::moderation_authz::/) + test(/^handlers::side_effects::tests::/) + test(/^storage_sweep::tests::/) + test(/^nip_fi_core::tests::/) + test(/^nip_fi_http::tests::/) + test(/^nip_fi_config::tests::/) + test(/^readiness::tests::/) + test(/^router::tests::/) + test(/^api::parse_query_tests::/) + test(/^api::git::transport::off_mode_precedence_tests::/) + test(/^audio::join::tests::/) + test(/^audio::handler::tests::/) + test(/^nip_fi_gate::tests::/) + test(/^nip_fi_session::tests::/) + test(/^nip_fi_shadow::tests::/) + test(/^nip_fi_shadow_session::tests::/) + test(=state::tests::neither_a_confirmed_inactive_community_nor_a_failed_lookup_admits_the_socket) + test(=handlers::req::tests::timed_out_historical_read_deregisters_before_closed) + test(=handlers::req::tests::superseded_timeout_leaves_replacement_intact) + test(=handlers::req::tests::search_claim_retires_live_and_yields_to_replacement) + test(=handlers::req::tests::concurrent_claims_and_stale_teardowns_keep_the_last_owner) + test(=handlers::req::tests::timeout_closed_is_emitted_before_a_replacement_can_claim) + test(=handlers::req::tests::revoke_then_replacement_keeps_replacement_whole) + test(=handlers::req::tests::claims_after_connection_cleanup_are_refused) + test(=handlers::req::tests::dropped_terminal_frame_cancels_connection) + test(=handlers::req::tests::revoke_dropped_terminal_frame_cancels_connection) + (kind(bin) & (test(/^tests::/) + test(/^composition_tests::/) + test(/^env_filter_tests::/)) - test(/^tests::postgres_tests::/))'
-E 'test(/^api::admin::/) + test(/^handlers::channel_authz::/) + test(/^handlers::moderation_authz::/) + test(/^handlers::side_effects::tests::/) + test(/^storage_sweep::tests::/) + test(/^nip_fi_core::tests::/) + test(/^nip_fi_http::tests::/) + test(/^nip_fi_config::tests::/) + test(/^readiness::tests::/) + test(/^router::tests::/) + test(/^api::parse_query_tests::/) + test(/^api::git::transport::off_mode_precedence_tests::/) + test(/^audio::join::tests::/) + test(/^audio::handler::tests::/) + test(/^nip_fi_gate::tests::/) + test(/^nip_fi_session::tests::/) + test(/^nip_fi_shadow::tests::/) + test(/^nip_fi_shadow_session::tests::/) + test(/^startup_steps::tests::/) + test(/^telemetry::tests::/) + test(/^config::tests::/) + test(=state::tests::neither_a_confirmed_inactive_community_nor_a_failed_lookup_admits_the_socket) + test(=handlers::req::tests::timed_out_historical_read_deregisters_before_closed) + test(=handlers::req::tests::superseded_timeout_leaves_replacement_intact) + test(=handlers::req::tests::search_claim_retires_live_and_yields_to_replacement) + test(=handlers::req::tests::concurrent_claims_and_stale_teardowns_keep_the_last_owner) + test(=handlers::req::tests::timeout_closed_is_emitted_before_a_replacement_can_claim) + test(=handlers::req::tests::revoke_then_replacement_keeps_replacement_whole) + test(=handlers::req::tests::claims_after_connection_cleanup_are_refused) + test(=handlers::req::tests::dropped_terminal_frame_cancels_connection) + test(=handlers::req::tests::revoke_dropped_terminal_frame_cancels_connection) + (kind(bin) & (test(/^tests::/) + test(/^composition_tests::/) + test(/^env_filter_tests::/)) - test(/^tests::postgres_tests::/))'
# Note on audio::join::tests scope: the full suite is infra-free (no DB,
# no Redis). The infra-free audio/FI regression witnesses — bootstrap
# ordering barrier, CommitConfirmed arm, pending-close invisibility,
@@ -545,6 +549,12 @@ test-unit:
+ test(=state::tests::disconnect_community_wins_reason_losing_nip_fi_does_not_enqueue_frame)
+ test(=state::tests::manager_disconnect_sets_reason_enqueues_frame_then_cancels)
+ test(/^api::nip_fi::/)'
# boot_lifecycle spawns the real relay binary and asserts its startup
# lifecycle, including that buzz_startup_phase_* reaches /metrics. Its
# non-ignored tests need no Postgres or Redis; the Postgres cases are
# #[ignore]d and stay in the PostgreSQL lane. Keep
# scripts/run-tests.sh in step.
cargo nextest run -p buzz-relay --test boot_lifecycle
# ACP author-gate and queue tests protect the trust boundary between
# relay events and agent prompts. They are infra-free; ignored lifecycle
# tests remain excluded and run in their dedicated integration lanes.

1
crates/buzz-relay/src/api/bridge.rs
View File

@@ -4831,6 +4831,7 @@ pub(crate) mod postgres_tests {
pubkey_hex: &str,
body: &[u8],
) -> (axum::http::StatusCode, String) {
let _tracing = crate::test_support::tracing_dispatch_lock();
let buf = Arc::new(Mutex::new(Vec::<u8>::new()));
let make_writer = CapturingMakeWriter {
buf: Arc::clone(&buf),

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

@@ -1627,6 +1627,8 @@ mod tests {
) -> (Result<Config, ConfigError>, String) {
use std::sync::{Arc, Mutex};
let _tracing = crate::test_support::tracing_dispatch_lock();
#[derive(Clone)]
struct CapturingMakeWriter {
buf: Arc<Mutex<Vec<u8>>>,

87
crates/buzz-relay/src/handlers/side_effects.rs
View File

@@ -3112,6 +3112,15 @@ pub enum Nip43ReconciliationPurpose {
Maintenance,
}
/// Per-item counts from one snapshot reconciliation sweep.
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
pub struct ReconcileSummary {
/// Snapshots rebuilt and republished.
pub repaired: usize,
/// Items whose reconciliation failed; each was logged and counted.
pub failed: usize,
}
/// Preserve the original maintenance reconciliation API for downstream callers.
#[deprecated(note = "use reconcile_nip43_membership_snapshots_with_purpose")]
pub async fn reconcile_nip43_membership_snapshots(state: &Arc<AppState>) -> anyhow::Result<usize> {
@@ -3120,20 +3129,34 @@ pub async fn reconcile_nip43_membership_snapshots(state: &Arc<AppState>) -> anyh
Nip43ReconciliationPurpose::Maintenance,
)
.await
.map(|summary| summary.repaired)
}
/// Reconcile NIP-43 snapshots with explicit startup or maintenance attribution.
pub async fn reconcile_nip43_membership_snapshots_with_purpose(
state: &Arc<AppState>,
purpose: Nip43ReconciliationPurpose,
) -> anyhow::Result<usize> {
) -> anyhow::Result<ReconcileSummary> {
let communities = match purpose {
Nip43ReconciliationPurpose::Bootstrap => state.db.bootstrap_community_hosts().await?,
Nip43ReconciliationPurpose::Maintenance => state.db.active_community_hosts().await?,
};
let mut reconciled = 0usize;
let mut failed = 0usize;
let total = communities.len();
let started = std::time::Instant::now();
for community in communities {
for (index, community) in communities.into_iter().enumerate() {
if nip43_progress_due(purpose, index, total) {
info!(
done = index,
total,
republished = reconciled,
failed,
elapsed_ms = u64::try_from(started.elapsed().as_millis()).unwrap_or(u64::MAX),
"NIP-43 startup reconciliation progress"
);
}
let community_id = buzz_core::CommunityId::from_uuid(community.id);
let host = community.host;
let result = async {
@@ -3171,6 +3194,7 @@ pub async fn reconcile_nip43_membership_snapshots_with_purpose(
Ok(true) => reconciled += 1,
Ok(false) => {}
Err(error) => {
failed += 1;
metrics::counter!("buzz_nip43_membership_reconciliation_failures_total")
.increment(1);
warn!(%community_id, %host, %error, "NIP-43 membership reconciliation failed");
@@ -3179,7 +3203,24 @@ pub async fn reconcile_nip43_membership_snapshots_with_purpose(
}
metrics::counter!("buzz_nip43_membership_reconciliations_total").increment(reconciled as u64);
Ok(reconciled)
Ok(ReconcileSummary {
repaired: reconciled,
failed,
})
}
/// Communities between startup NIP-43 reconciliation progress logs.
const NIP43_BOOTSTRAP_PROGRESS_INTERVAL: usize = 5_000;
/// Whether a sweep logs progress before processing community `done` of
/// `total`. Only the startup sweep logs (the 60 s maintenance sweep stays
/// quiet): every [`NIP43_BOOTSTRAP_PROGRESS_INTERVAL`] communities, never at
/// the start (nothing done yet) and never past the end.
fn nip43_progress_due(purpose: Nip43ReconciliationPurpose, done: usize, total: usize) -> bool {
matches!(purpose, Nip43ReconciliationPurpose::Bootstrap)
&& done > 0
&& done < total
&& done.is_multiple_of(NIP43_BOOTSTRAP_PROGRESS_INTERVAL)
}
/// Publish a kind:13534 relay membership list event (NIP-43).
@@ -3315,7 +3356,7 @@ pub async fn publish_nip43_member_removed(
/// never resolves a channel against a neighboring tenant.
pub async fn reconcile_large_channel_member_snapshots(
state: &Arc<AppState>,
) -> anyhow::Result<usize> {
) -> anyhow::Result<ReconcileSummary> {
const LEGACY_ROSTER_LIMIT: i64 = 1_000;
let relay_pubkey = state.relay_keypair.public_key();
@@ -3328,6 +3369,7 @@ pub async fn reconcile_large_channel_member_snapshots(
.await?;
let relay_pubkey_hex = relay_pubkey.to_hex();
let mut reconciled = 0usize;
let mut failed = 0usize;
for candidate in candidates {
let result = async {
@@ -3352,6 +3394,7 @@ pub async fn reconcile_large_channel_member_snapshots(
Ok(true) => reconciled += 1,
Ok(false) => {}
Err(error) => {
failed += 1;
metrics::counter!("buzz_channel_roster_reconciliation_failures_total").increment(1);
warn!(
community_id = %candidate.community_id,
@@ -3365,7 +3408,10 @@ pub async fn reconcile_large_channel_member_snapshots(
}
metrics::counter!("buzz_channel_roster_reconciliations_total").increment(reconciled as u64);
Ok(reconciled)
Ok(ReconcileSummary {
repaired: reconciled,
failed,
})
}
/// Reconcile channels that exist in the DB but don't have kind:39000 events.
@@ -3799,6 +3845,37 @@ pub async fn publish_nipia_unarchived(
mod tests {
use super::*;
#[test]
fn nip43_progress_logs_every_interval_inside_the_startup_sweep_only() {
use Nip43ReconciliationPurpose::{Bootstrap, Maintenance};
const STEP: usize = NIP43_BOOTSTRAP_PROGRESS_INTERVAL;
let total = 79_942;
let due: Vec<usize> = (0..total)
.filter(|&done| nip43_progress_due(Bootstrap, done, total))
.collect();
assert_eq!(due.first(), Some(&STEP));
assert_eq!(due.len(), total / STEP);
assert!(due.iter().all(|done| done.is_multiple_of(STEP)));
// (purpose, done, total, expected)
for (purpose, done, total, expected) in [
(Bootstrap, STEP, total, true),
(Maintenance, STEP, total, false),
(Bootstrap, 0, total, false),
(Bootstrap, STEP - 1, total, false),
(Bootstrap, STEP, STEP, false),
(Bootstrap, STEP, STEP + 1, true),
] {
assert_eq!(
nip43_progress_due(purpose, done, total),
expected,
"done={done} total={total} bootstrap={}",
matches!(purpose, Bootstrap)
);
}
assert!(!(0..total).any(|done| nip43_progress_due(Maintenance, done, total)));
}
#[test]
fn workflow_deletion_retry_matches_authorized_dispatch() {
let keys = nostr::Keys::generate();

2
crates/buzz-relay/src/lib.rs
View File

@@ -62,6 +62,8 @@ pub mod push_runtime;
pub mod readiness;
/// Axum router construction.
pub mod router;
/// Timing for startup work between metrics bind and serving.
pub mod startup_steps;
/// Shared application state.
pub mod state;
pub mod storage_sweep;

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

@@ -20,6 +20,7 @@ use buzz_relay::config::{Config, MAX_DRAIN_JITTER_MS};
use buzz_relay::lifecycle::{BootTracker, LifecycleReason, StartupPhase};
use buzz_relay::metrics as relay_metrics;
use buzz_relay::router::{build_health_router, build_router};
use buzz_relay::startup_steps::{StartupStep, StepTimer};
use buzz_relay::state::AppState;
use buzz_relay::storage_sweep;
use buzz_relay::telemetry;
@@ -305,10 +306,12 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> {
..DbConfig::default()
}
.with_session_timeouts_from_env();
let step = StepTimer::start(StartupStep::DbConnect);
let db = Db::new(&db_config).await.map_err(|e| {
error!("Failed to connect to Postgres: {e}");
anyhow::anyhow!("DB connection failed: {e}")
})?;
step.finish();
if db.has_read_pool() {
info!("Postgres connected (writer + lazy read replica pool)");
// Reader-down at boot must not crash or block the relay; this warn-only
@@ -322,21 +325,25 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> {
let auto_migrate =
buzz_auto_migrate_enabled(std::env::var("BUZZ_AUTO_MIGRATE").ok().as_deref());
if auto_migrate {
let step = StepTimer::start(StartupStep::DbMigrate);
db.migrate().await.map_err(|e| {
error!("Failed to run database migrations: {e}");
anyhow::anyhow!("Database migration failed: {e}")
})?;
step.finish();
info!("Database migrations complete");
} else {
info!("Skipping database migrations because BUZZ_AUTO_MIGRATE is not enabled");
}
let mut step = StepTimer::start(StartupStep::PartitionEnsure);
let startup_partition_audit = match db
.ensure_future_partitions(3, config.partition_manager_create_enabled)
.await
{
Ok(audit) => Some(audit),
Err(error) => {
step.degrade();
error!(%error, "Failed to ensure partitions");
match db.audit_partitions(3).await {
Ok(audit) => Some(audit),
@@ -347,12 +354,15 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> {
}
}
};
step.finish();
let step = StepTimer::start(StartupStep::DeletionFenceVerify);
db.validate_deletion_serving_catalog().await.map_err(|e| {
error!("Community deletion serving-fence validation failed: {e}");
anyhow::anyhow!("Community deletion serving fence is unsafe: {e}")
})?;
info!("Community deletion serving fences verified");
step.finish();
// Freshness fence probe: cursor pages route to the replica only for
// history the probe has verified as fully replayed. Deliberately AFTER
@@ -363,16 +373,19 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> {
// fence over an unenforced floor. Verification failure is loud but
// non-fatal: the fence stays closed and every cursor page routes to the
// writer.
let mut step = StepTimer::start(StartupStep::ReplicaFenceProbe);
match db.spawn_fence_probe().await {
Ok(true) => info!("Replica fence probe started (floor guard verified)"),
Ok(false) => {}
Err(e) => {
step.degrade();
error!(
"Replica fence disabled — floor guard verification failed: {e}. \
All cursor reads stay on the writer."
);
}
}
step.finish();
// NIP-43: if membership enforcement is on, a valid owner pubkey is required.
// config.rs already strips invalid values with a warning; catch the resulting
@@ -409,6 +422,7 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> {
// startup. An empty authority (unparseable `relay_url`)
// is a misconfiguration — fail fast when membership is enforced rather than
// seeding an empty-host community that no request can ever resolve to.
let mut step = StepTimer::start(StartupStep::MembershipBootstrap);
let deployment_community = {
let host = buzz_relay::tenant::relay_url_authority(&config.relay_url);
if host.is_empty() {
@@ -418,6 +432,7 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> {
config.relay_url
));
}
step.degrade();
error!(
relay_url = %config.relay_url,
"Could not derive a community host from relay_url; skipping membership backfill/bootstrap (non-fatal, membership not required)"
@@ -436,6 +451,7 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> {
"Failed to ensure deployment community (required when BUZZ_REQUIRE_RELAY_MEMBERSHIP=true): {e}"
));
}
step.degrade();
error!("Failed to ensure deployment community (non-fatal, membership not required): {e}");
None
}
@@ -460,6 +476,7 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> {
"Failed to backfill pubkey_allowlist (required when BUZZ_REQUIRE_RELAY_MEMBERSHIP=true): {e}"
));
} else {
step.degrade();
error!("Failed to backfill pubkey_allowlist (non-fatal): {e}");
}
}
@@ -483,6 +500,7 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> {
"Failed to bootstrap relay owner (required when BUZZ_REQUIRE_RELAY_MEMBERSHIP=true): {e}"
));
} else {
step.degrade();
error!(
"Failed to bootstrap relay owner (non-fatal, membership not required): {e}"
);
@@ -490,19 +508,27 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> {
}
}
}
step.finish();
// NIP-33: backfill d_tag for any existing parameterized replaceable events
// that predate the column addition. Idempotent — no-ops when fully populated.
let mut step = StepTimer::start(StartupStep::DTagBackfill);
match db.backfill_d_tags().await {
Ok(0) => {}
Ok(n) => info!("Backfilled d_tag for {n} NIP-33 events"),
Err(e) => error!("Failed to backfill d_tags: {e}"),
Err(e) => {
step.degrade();
error!("Failed to backfill d_tags: {e}");
}
}
step.finish();
let (audit, audit_metrics_pool) = if config.audit_enabled {
let step = StepTimer::start(StartupStep::AuditConnect);
let audit_pool = connect_audit_pool(&db_config)
.await
.map_err(|e| anyhow::anyhow!("Audit DB connection failed: {e}"))?;
step.finish();
info!("Audit service ready");
let metrics_pool = audit_pool.clone();
(Some(AuditService::new(audit_pool)), Some(metrics_pool))
@@ -511,6 +537,7 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> {
(None, None)
};
let step = StepTimer::start(StartupStep::RedisConnect);
let redis_pool = {
let mut cfg = deadpool_redis::Config::from_url(&config.redis_url);
cfg.pool = Some(deadpool_redis::PoolConfig::new(config.redis_pool_size));
@@ -530,6 +557,7 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> {
.map_err(|e| anyhow::anyhow!("PubSub init failed: {e}"))?,
);
info!("Redis pub/sub connected");
step.finish();
// Spawn Redis pub/sub subscriber for multi-node fan-out.
// Events published by other relay instances are received here and
@@ -575,10 +603,12 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> {
.read_database_url
.as_deref()
.unwrap_or(&config.database_url);
let step = StepTimer::start(StartupStep::SearchConnect);
let search_pool = sqlx::postgres::PgPoolOptions::new()
.connect(search_db_url)
.await
.map_err(|e| anyhow::anyhow!("Search DB connection failed: {e}"))?;
step.finish();
let search_metrics_pool = search_pool.clone();
let search = SearchService::new(search_pool);
info!(
@@ -679,7 +709,13 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> {
"NIP-FI: warming JWKS snapshots"
);
let issuer_ids: Vec<String> = jwks_configs.iter().map(|c| c.issuer.clone()).collect();
let mut step = StepTimer::start(StartupStep::NipFiJwksWarm);
let warmed = warm_nip_fi_jwks_snapshots(&jwks_source, &issuer_ids).await;
if warmed.iter().any(|ok| !ok) {
// FI ingress fails closed until the refresh loop lands a snapshot.
step.degrade();
}
step.finish();
let issuers = jwks_configs
.iter()
.zip(warmed)
@@ -709,6 +745,7 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> {
// relay behaves byte-identically to a build without the mesh. When
// enabled, a misconfigured mesh is fatal here (bind/Redis failure): an
// operator who asked for the mesh gets it or gets told why not.
let step = StepTimer::start(StartupStep::MeshBoot);
if let Some(handle) = buzz_relay::mesh_boot::boot_mesh(
&state.config,
state.redis_pool.clone(),
@@ -732,6 +769,7 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> {
}
info!(runtime_id = %runtime_id, "Inter-relay mesh started");
}
step.finish();
// Git-on-object-storage: admit the configured S3/MinIO backend against the
// linearizable conditional-write axiom (A3) before serving git traffic.
@@ -758,11 +796,13 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> {
race_rounds,
"running git object-store conformance probe (A3 gate)"
);
let step = StepTimer::start(StartupStep::GitConformanceProbe);
let report = state
.git_store
.run_conformance_probe(cfg)
.await
.map_err(|e| anyhow::anyhow!("git conformance probe failed: {e}"))?;
step.finish();
tracing::info!(
race_width = report.race_width,
race_rounds = report.race_rounds,
@@ -771,6 +811,7 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> {
);
}
let step = StepTimer::start(StartupStep::ChannelRosterFence);
match state.db.verify_channel_roster_fence().await {
Ok(()) => {
info!("Channel roster fence verified");
@@ -782,36 +823,62 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> {
));
}
}
step.finish();
// Repair legacy NIP-29 channel rosters that were persisted while the
// canonical member query still truncated at 1,000 rows. Validation above
// makes migration 0032 a code/schema compatibility gate before the new
// replacement protocol or listener can serve traffic.
let mut step = StepTimer::start(StartupStep::LargeRosterReconcile);
match buzz_relay::handlers::side_effects::reconcile_large_channel_member_snapshots(&state).await
{
Ok(count) if count > 0 => info!(count, "large channel member snapshots repaired"),
Ok(_) => {}
Ok(summary) => {
if summary.failed > 0 {
step.degrade();
}
if summary.repaired > 0 || summary.failed > 0 {
info!(
count = summary.repaired,
failed = summary.failed,
"large channel member snapshots reconciled on startup"
);
}
}
Err(error) => {
step.degrade();
tracing::warn!(%error, "large channel member snapshot startup reconciliation failed")
}
}
step.finish();
// NIP-43: reconcile the event-backed roster for every provisioned
// community before opening the listener. `relay_members` is canonical;
// this repairs pre-snapshot communities and any publication that failed
// after a membership transaction committed.
if config.require_relay_membership {
let mut step = StepTimer::start(StartupStep::Nip43Reconcile);
match buzz_relay::handlers::side_effects::reconcile_nip43_membership_snapshots_with_purpose(
&state,
buzz_relay::handlers::side_effects::Nip43ReconciliationPurpose::Bootstrap,
)
.await
{
Ok(count) => info!(count, "NIP-43 membership snapshots reconciled on startup"),
Ok(summary) => {
if summary.failed > 0 {
step.degrade();
}
info!(
count = summary.repaired,
failed = summary.failed,
"NIP-43 membership snapshots reconciled on startup"
);
}
Err(error) => {
step.degrade();
tracing::warn!(%error, "NIP-43 membership snapshot startup reconciliation failed")
}
}
step.finish();
let reconcile_state = Arc::clone(&state);
let interval_secs = std::env::var("BUZZ_NIP43_RECONCILE_INTERVAL_SECS")
@@ -830,8 +897,8 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> {
)
.await
{
Ok(count) if count > 0 => {
info!(count, "NIP-43 membership snapshots repaired")
Ok(summary) if summary.repaired > 0 => {
info!(count = summary.repaired, "NIP-43 membership snapshots repaired")
}
Ok(_) => {}
Err(error) => tracing::warn!(
@@ -1824,9 +1891,11 @@ async fn serve(
) -> anyhow::Result<()> {
let config = &state.config;
let step = StepTimer::start(StartupStep::HealthBind);
let health_listener = tokio::net::TcpListener::bind(("0.0.0.0", config.health_port))
.await
.map_err(|e| anyhow::anyhow!("Failed to bind health port {}: {e}", config.health_port))?;
step.finish();
info!(port = config.health_port, "Health probe listener started");
tokio::spawn(async move {
axum::serve(health_listener, health_router).await.ok();
@@ -1899,10 +1968,12 @@ async fn serve(
hard_shutdown_abort
});
let step = StepTimer::start(StartupStep::ListenerBind);
let tcp_listener = tokio::net::TcpListener::bind(&config.bind_addr)
.await
.map_err(|e| anyhow::anyhow!("Failed to bind {}: {e}", config.bind_addr))?;
info!(addr = %config.bind_addr, "buzz-relay TCP listening");
step.finish();
#[cfg(unix)]
if let Some(ref uds_path) = config.uds_path {

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

@@ -297,6 +297,7 @@ pub fn try_install(port: u16, gauge_idle_timeout_secs: u64) -> Result<(), Metric
describe_community_admission_metrics();
describe_db_pool_metrics();
describe_auth_metrics();
crate::startup_steps::describe_metrics();
initialize_auth_metric_series();
tokio::spawn(exporter);
Ok(())

391
crates/buzz-relay/src/startup_steps.rs Normal file
View File

@@ -0,0 +1,391 @@
//! Timing for the relay's startup work between metrics bind and serving.
//!
//! [`crate::lifecycle`] covers the earliest phases, which run before the
//! metrics exporter exists, as log-only records. Everything after
//! `metrics_bind` runs with a working recorder and structured logging, so
//! each step logs `Startup phase finished` (`phase`, `elapsed_ms`, `status`)
//! and sets two gauges:
//!
//! - `buzz_startup_phase_current{phase}` is `1` while the step runs and `0`
//! once it ends, so a pod that is still starting shows which step it is on.
//! - `buzz_startup_phase_seconds{phase}` is the step's wall-clock duration,
//! set once when the step ends.
//!
//! The `phase` label values are a closed vocabulary ([`StartupStep::ALL`]).
//! They never overlap the early lifecycle phase names.
use std::time::Instant;
use tracing::{info, warn};
/// Gauge that is `1` while a startup step is running.
pub const CURRENT_METRIC: &str = "buzz_startup_phase_current";
/// Gauge holding a finished startup step's duration in seconds.
pub const SECONDS_METRIC: &str = "buzz_startup_phase_seconds";
/// Register the startup-step gauge descriptions.
pub(crate) fn describe_metrics() {
metrics::describe_gauge!(
CURRENT_METRIC,
"1 while the labelled post-metrics-bind startup phase is running, 0 once it ends"
);
metrics::describe_gauge!(
SECONDS_METRIC,
metrics::Unit::Seconds,
"Wall-clock duration of the labelled post-metrics-bind startup phase"
);
}
/// One timed unit of startup work after the metrics listener is up.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum StartupStep {
/// Connect the Postgres writer (and lazy reader) pools.
DbConnect,
/// Apply database migrations (only when `BUZZ_AUTO_MIGRATE` is on).
DbMigrate,
/// Ensure or audit future event partitions.
PartitionEnsure,
/// Verify the community-deletion serving fences.
DeletionFenceVerify,
/// Verify the replica floor guard and start the fence probe.
ReplicaFenceProbe,
/// Ensure the deployment community, backfill the allowlist, and bootstrap the owner.
MembershipBootstrap,
/// Backfill NIP-33 `d_tag` values.
DTagBackfill,
/// Connect the audit database pool (only when audit is enabled).
AuditConnect,
/// Connect Redis commands and pub/sub.
RedisConnect,
/// Connect the search database pool.
SearchConnect,
/// Warm NIP-FI JWKS snapshots (only when NIP-FI is configured).
NipFiJwksWarm,
/// Start the inter-relay mesh (no-op when the mesh is off).
MeshBoot,
/// Run the git object-store conformance probe (when enabled).
GitConformanceProbe,
/// Verify the channel roster fence.
ChannelRosterFence,
/// Repair large NIP-29 channel roster snapshots.
LargeRosterReconcile,
/// Reconcile every community's NIP-43 membership snapshot.
Nip43Reconcile,
/// Bind the private health listener.
HealthBind,
/// Bind the public relay listener.
ListenerBind,
}
impl StartupStep {
/// The complete `phase` label vocabulary.
pub const ALL: [Self; 18] = [
Self::DbConnect,
Self::DbMigrate,
Self::PartitionEnsure,
Self::DeletionFenceVerify,
Self::ReplicaFenceProbe,
Self::MembershipBootstrap,
Self::DTagBackfill,
Self::AuditConnect,
Self::RedisConnect,
Self::SearchConnect,
Self::NipFiJwksWarm,
Self::MeshBoot,
Self::GitConformanceProbe,
Self::ChannelRosterFence,
Self::LargeRosterReconcile,
Self::Nip43Reconcile,
Self::HealthBind,
Self::ListenerBind,
];
/// Stable `phase` label value.
pub const fn as_str(self) -> &'static str {
match self {
Self::DbConnect => "db_connect",
Self::DbMigrate => "db_migrate",
Self::PartitionEnsure => "partition_ensure",
Self::DeletionFenceVerify => "deletion_fence_verify",
Self::ReplicaFenceProbe => "replica_fence_probe",
Self::MembershipBootstrap => "membership_bootstrap",
Self::DTagBackfill => "d_tag_backfill",
Self::AuditConnect => "audit_connect",
Self::RedisConnect => "redis_connect",
Self::SearchConnect => "search_connect",
Self::NipFiJwksWarm => "nip_fi_jwks_warm",
Self::MeshBoot => "mesh_boot",
Self::GitConformanceProbe => "git_conformance_probe",
Self::ChannelRosterFence => "channel_roster_fence",
Self::LargeRosterReconcile => "large_roster_reconcile",
Self::Nip43Reconcile => "nip43_reconcile",
Self::HealthBind => "health_bind",
Self::ListenerBind => "listener_bind",
}
}
}
/// How a startup step ended (same values as `buzz_process_lifecycle` `status`).
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum StepStatus {
Succeeded,
/// Non-fatal error; startup continued.
Degraded,
/// Dropped without [`StepTimer::finish`] — an early return or a panic.
Failed,
}
impl StepStatus {
const fn as_str(self) -> &'static str {
match self {
Self::Succeeded => "succeeded",
Self::Degraded => "degraded",
Self::Failed => "failed",
}
}
}
/// A running startup step. Ends exactly once: [`StepTimer::finish`] reports
/// `succeeded` (or `degraded` after [`StepTimer::degrade`]); dropping it
/// without finishing reports `failed`, which covers `?` returns and panics.
#[must_use = "a dropped step timer reports the step as failed"]
pub struct StepTimer {
step: StartupStep,
started: Instant,
degraded: bool,
ended: bool,
}
impl StepTimer {
/// Start timing `step` and mark it as the pod's current step.
pub fn start(step: StartupStep) -> Self {
metrics::gauge!(CURRENT_METRIC, "phase" => step.as_str()).set(1.0);
info!(phase = step.as_str(), "Startup phase started");
Self {
step,
started: Instant::now(),
degraded: false,
ended: false,
}
}
/// Record that the step hit a non-fatal error; it will end as `degraded`.
pub fn degrade(&mut self) {
self.degraded = true;
}
/// End the step as `succeeded`, or `degraded` if [`Self::degrade`] was called.
pub fn finish(mut self) {
let status = if self.degraded {
StepStatus::Degraded
} else {
StepStatus::Succeeded
};
self.end(status);
}
fn end(&mut self, status: StepStatus) {
if self.ended {
return;
}
self.ended = true;
let elapsed = self.started.elapsed();
let phase = self.step.as_str();
metrics::gauge!(SECONDS_METRIC, "phase" => phase).set(elapsed.as_secs_f64());
metrics::gauge!(CURRENT_METRIC, "phase" => phase).set(0.0);
let elapsed_ms = u64::try_from(elapsed.as_millis()).unwrap_or(u64::MAX);
match status {
StepStatus::Succeeded => {
info!(
phase,
elapsed_ms,
status = status.as_str(),
"Startup phase finished"
)
}
StepStatus::Degraded | StepStatus::Failed => {
warn!(
phase,
elapsed_ms,
status = status.as_str(),
"Startup phase finished"
)
}
}
}
}
impl Drop for StepTimer {
fn drop(&mut self) {
self.end(StepStatus::Failed);
}
}
#[cfg(test)]
mod tests {
use std::collections::{HashMap, HashSet};
use std::sync::{Arc, Mutex};
use metrics_util::debugging::{DebugValue, DebuggingRecorder};
use super::*;
use crate::lifecycle::StartupPhase;
/// Gauge values keyed by `(metric, phase)` after running `body`.
///
/// Holds the tracing dispatch lock: `body` fires the step log callsites
/// with no dispatcher, which must not race a `finished_records` capture.
fn gauges(body: impl FnOnce()) -> HashMap<(String, String), f64> {
let _tracing = crate::test_support::tracing_dispatch_lock();
let recorder = DebuggingRecorder::new();
let snapshotter = recorder.snapshotter();
metrics::with_local_recorder(&recorder, body);
snapshotter
.snapshot()
.into_vec()
.into_iter()
.filter_map(|(key, _, _, value)| {
let key = key.key();
let phase = key
.labels()
.find(|l| l.key() == "phase")?
.value()
.to_owned();
match value {
DebugValue::Gauge(v) => Some(((key.name().to_owned(), phase), v.into_inner())),
_ => None,
}
})
.collect()
}
#[derive(Clone, Default)]
struct CapturedLogs(Arc<Mutex<Vec<u8>>>);
impl std::io::Write for CapturedLogs {
fn write(&mut self, data: &[u8]) -> std::io::Result<usize> {
self.0
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner())
.extend_from_slice(data);
Ok(data.len())
}
fn flush(&mut self) -> std::io::Result<()> {
Ok(())
}
}
impl<'a> tracing_subscriber::fmt::MakeWriter<'a> for CapturedLogs {
type Writer = Self;
fn make_writer(&'a self) -> Self::Writer {
self.clone()
}
}
/// `(phase, status)` of every `Startup phase finished` record `body` emits.
fn finished_records(body: impl FnOnce()) -> Vec<(String, String)> {
let _tracing = crate::test_support::tracing_dispatch_lock();
let logs = CapturedLogs::default();
let subscriber = tracing_subscriber::fmt()
.json()
.with_writer(logs.clone())
.finish();
tracing::subscriber::with_default(subscriber, body);
let bytes = logs
.0
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner())
.clone();
String::from_utf8_lossy(&bytes)
.lines()
.filter_map(|line| serde_json::from_str::<serde_json::Value>(line).ok())
.filter(|record| record["fields"]["message"] == "Startup phase finished")
.map(|record| {
let field = |name: &str| record["fields"][name].as_str().unwrap_or("").to_owned();
(field("phase"), field("status"))
})
.collect()
}
fn get(gauges: &HashMap<(String, String), f64>, metric: &str, phase: &str) -> Option<f64> {
gauges.get(&(metric.to_owned(), phase.to_owned())).copied()
}
#[test]
fn running_step_is_current_and_has_no_duration() {
let mut held = None;
let gauges = gauges(|| held = Some(StepTimer::start(StartupStep::Nip43Reconcile)));
assert_eq!(get(&gauges, CURRENT_METRIC, "nip43_reconcile"), Some(1.0));
assert_eq!(get(&gauges, SECONDS_METRIC, "nip43_reconcile"), None);
std::mem::forget(held);
}
#[test]
fn finished_step_clears_current_and_records_duration() {
let gauges = gauges(|| {
let step = StepTimer::start(StartupStep::DbConnect);
std::thread::sleep(std::time::Duration::from_millis(5));
step.finish();
});
assert_eq!(get(&gauges, CURRENT_METRIC, "db_connect"), Some(0.0));
let seconds = get(&gauges, SECONDS_METRIC, "db_connect").expect("duration gauge");
assert!(seconds >= 0.005, "duration {seconds} shorter than the step");
}
#[test]
fn dropped_and_degraded_steps_clear_current_and_record_duration() {
let gauges = gauges(|| {
drop(StepTimer::start(StartupStep::RedisConnect));
let mut step = StepTimer::start(StartupStep::PartitionEnsure);
step.degrade();
step.finish();
});
for phase in ["redis_connect", "partition_ensure"] {
assert_eq!(get(&gauges, CURRENT_METRIC, phase), Some(0.0));
assert!(get(&gauges, SECONDS_METRIC, phase).is_some());
}
}
#[test]
fn each_terminal_path_logs_exactly_one_finished_record_with_its_status() {
let records = finished_records(|| {
StepTimer::start(StartupStep::DbConnect).finish();
let mut degraded = StepTimer::start(StartupStep::PartitionEnsure);
degraded.degrade();
degraded.finish();
drop(StepTimer::start(StartupStep::RedisConnect));
});
let expected = [
("db_connect", "succeeded"),
("partition_ensure", "degraded"),
("redis_connect", "failed"),
]
.map(|(phase, status)| (phase.to_owned(), status.to_owned()));
assert_eq!(records, expected);
}
#[test]
fn a_panic_inside_a_step_logs_it_failed() {
let records = finished_records(|| {
let result = std::panic::catch_unwind(|| {
let _step = StepTimer::start(StartupStep::MeshBoot);
panic!("step body panicked");
});
assert!(result.is_err());
});
assert_eq!(records, [("mesh_boot".to_owned(), "failed".to_owned())]);
}
#[test]
fn vocabulary_is_unique_and_disjoint_from_early_lifecycle_phases() {
let names: HashSet<_> = StartupStep::ALL.iter().map(|s| s.as_str()).collect();
assert_eq!(names.len(), StartupStep::ALL.len());
for phase in StartupPhase::ALL {
assert!(
!names.contains(phase.as_str()),
"{} overlaps",
phase.as_str()
);
}
}
}

2
crates/buzz-relay/src/telemetry.rs
View File

@@ -307,6 +307,7 @@ mod tests {
#[test]
fn trace_context_json_correlates_nested_span_logs() {
let _tracing = crate::test_support::tracing_dispatch_lock();
let output = Arc::new(Mutex::new(Vec::new()));
let output_writer = Arc::clone(&output);
let exporter = InMemorySpanExporter::default();
@@ -474,6 +475,7 @@ mod tests {
#[test]
fn trace_context_lookup_does_not_enable_callsites() {
let _tracing = crate::test_support::tracing_dispatch_lock();
let context_lookup = TraceContextLookup::default();
let subscriber = tracing_subscriber::registry().with(
context_lookup

15
crates/buzz-relay/src/test_support.rs
View File

@@ -1,5 +1,20 @@
const DEFAULT_DATABASE_URL: &str = "postgres://buzz:buzz_dev@localhost:5432/buzz"; // sadscan:disable np.postgres.1 -- local test-only credentials
/// Serialize lib tests whose assertions depend on which tracing dispatchers
/// are live: log-capture and callsite-interest tests, and tests that fire the
/// callsites they assert on. tracing-core caches each callsite's interest
/// process-wide from whichever dispatchers are live when it is first hit or
/// rebuilt, so a parallel test's dispatcher (or a thread with none) can enable
/// or disable another test's callsites. Not reentrant; take it once per test.
/// Two async sites install a dispatcher without it, because the guard would be
/// held across `.await`: `router.rs`'s otel-export test and the
/// `pause_after_aux_page` helper in
/// `api/bridge/thread_window/postgres_tests/failure_postgres_tests.rs`.
pub(crate) fn tracing_dispatch_lock() -> std::sync::MutexGuard<'static, ()> {
static LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(());
LOCK.lock().unwrap_or_else(|poisoned| poisoned.into_inner())
}
/// Resolve the database URL shared by PostgreSQL-backed relay tests.
pub(crate) fn database_url() -> String {
std::env::var("BUZZ_TEST_DATABASE_URL")

35
crates/buzz-relay/tests/boot_lifecycle.rs
View File

@@ -11,6 +11,7 @@ use std::{
use serde_json::Value;
use buzz_relay::lifecycle::StartupPhase;
use buzz_relay::startup_steps;
use buzz_relay::state::REDIS_BOOTSTRAP_FAILURE;
const VALID_RELAY_PRIVATE_KEY: &str =
@@ -186,8 +187,19 @@ fn wait_for_scraped_metric(process: &mut RelayProcess, port: u16, needle: &str)
}
}
fn assert_no_startup_lifecycle_metrics(scrape: &str) {
/// The early lifecycle phases (crypto through metrics bind) are log-only.
/// The only startup metric families are the post-metrics-bind step gauges,
/// and no metric family or label value may name an early phase.
fn assert_no_early_lifecycle_metrics(scrape: &str) {
let step_families = [startup_steps::CURRENT_METRIC, startup_steps::SECONDS_METRIC];
for line in scrape.lines() {
for phase in StartupPhase::ALL {
assert!(
!line.contains(&format!("\"{}\"", phase.as_str())),
"logs-only lifecycle contract exported early phase {}: {line}",
phase.as_str()
);
}
let Some(name) = line
.strip_prefix("# HELP ")
.or_else(|| line.strip_prefix("# TYPE "))
@@ -195,6 +207,9 @@ fn assert_no_startup_lifecycle_metrics(scrape: &str) {
else {
continue;
};
if step_families.contains(&name) {
continue;
}
assert!(
!["startup", "boot", "lifecycle"]
.iter()
@@ -465,7 +480,7 @@ fn otlp_build_failure_is_degraded_without_leaking_endpoint_credentials() {
("DATABASE_URL", &database_url),
]);
let scrape = wait_for_relay_metrics(&mut process, port);
assert_no_startup_lifecycle_metrics(&scrape);
assert_no_early_lifecycle_metrics(&scrape);
let output = process.terminate();
let events = lifecycle_events(&output);
assert_accounting(&events);
@@ -481,7 +496,7 @@ fn otlp_build_failure_is_degraded_without_leaking_endpoint_credentials() {
}
#[test]
fn successful_main_emits_complete_lifecycle_without_startup_metrics() {
fn successful_main_emits_complete_lifecycle_and_reports_current_startup_step() {
let fake_database = TcpListener::bind(("127.0.0.1", 0)).expect("bind fake database");
let database_url = format!(
"postgres://buzz@127.0.0.1:{}/buzz",
@@ -498,8 +513,20 @@ fn successful_main_emits_complete_lifecycle_without_startup_metrics() {
("DATABASE_URL", &database_url),
]);
let scrape = wait_for_relay_metrics(&mut process, port);
assert_no_startup_lifecycle_metrics(&scrape);
assert_no_early_lifecycle_metrics(&scrape);
assert_auth_metrics_have_stable_boot_zeros(&scrape);
// The fake database accepts TCP but never speaks Postgres, so the relay
// stays in its first post-metrics-bind step and must say so on /metrics.
let current = format!(
"{}{{phase=\"db_connect\"}} 1",
startup_steps::CURRENT_METRIC
);
let scrape = wait_for_scraped_metric(&mut process, port, &current);
assert!(
!scrape.contains(&format!("{}{{", startup_steps::SECONDS_METRIC)),
"no step has finished while the relay waits on Postgres"
);
assert_no_early_lifecycle_metrics(&scrape);
let output = process.terminate();
let events = lifecycle_events(&output);

24
deploy/charts/buzz/README.md
View File

@@ -148,6 +148,30 @@ These phases intentionally do not emit metrics. Most run before the Prometheus
exporter exists, and one uniform log-only contract preserves every phase's real
event time and failure without assigning an eventual scrape time to earlier work.
Startup work **after** `metrics_bind` (database connect through public listener
bind) runs with a live recorder, so each step logs `Startup phase started` and
`Startup phase finished` (`phase`, `elapsed_ms`, and `status` =
`succeeded`/`degraded`/`failed`, the same values lifecycle records use) and sets
two gauges labelled `phase`:
- `buzz_startup_phase_current` is `1` while the step runs and `0` once it ends.
A pod that has not opened its health port shows the step it is stuck on.
- `buzz_startup_phase_seconds` is the finished step's duration.
`degraded` means the step hit a non-fatal error and startup continued — for
example, some communities failed NIP-43 reconciliation or a NIP-FI issuer did
not warm. `failed` means startup aborted inside the step.
These are gauges, not counters, because each value is written once per boot
and a monotonic-counter scrape drops a series' first sample. Each is written
once, so it ages out after the gauge idle timeout: the `idle_timeout_secs`
field of the `Prometheus metrics exporter started` log (2700 s with default
intervals). A step that runs longer than that timeout loses its `_current`
series, so a pod stuck that long shows no current step. The `phase` values are
the closed `StartupStep` vocabulary in `crates/buzz-relay/src/startup_steps.rs`;
they never reuse an early lifecycle phase name. The NIP-43 startup reconcile
also logs `NIP-43 startup reconciliation progress` every 5,000 communities.
### Readiness contract
**`/_readiness` reports local process lifecycle only.** It performs no

16
scripts/run-tests.sh
View File

@@ -203,6 +203,19 @@ run_unit_tests() {
run_test_step "buzz-relay NIP-FI shadow session tests" \
cargo test -p buzz-relay --lib nip_fi_shadow_session::tests:: -- --nocapture
run_test_step "buzz-relay startup step tests" \
cargo test -p buzz-relay --lib startup_steps::tests:: -- --nocapture
# telemetry and config tests mutate process env. libtest runs one process
# with many threads (nextest, used by `just test-unit`, isolates each test),
# so run them single-threaded here. `--skip` keeps the substring filter
# `config::tests::` from also selecting nip_fi_config, which runs above.
run_test_step "buzz-relay telemetry tests" \
cargo test -p buzz-relay --lib telemetry::tests:: -- --test-threads=1 --nocapture
run_test_step "buzz-relay config tests" \
cargo test -p buzz-relay --lib config::tests:: -- --skip nip_fi_config:: --test-threads=1 --nocapture
# Mirror the NIP-FI (S3) stanza from `just test-unit`: module filters, then
# each exact name. Keep this list in step with that stanza's `test(=...)`s.
run_test_step "buzz-relay NIP-FI config tests" \
@@ -304,6 +317,9 @@ run_unit_tests() {
run_test_step "buzz-relay binary tests" \
cargo test -p buzz-relay --bin buzz-relay -- --nocapture
run_test_step "buzz-relay boot lifecycle tests" \
cargo test -p buzz-relay --test boot_lifecycle -- --nocapture
}
# ---- DB / integration tests (infra required) --------------------------------