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

perf(relay): bound fleet usage metrics collection (#7176)

Browse Source 
## Why

The usage poller pays memory and query costs proportional to community
count even when per-community emission is disabled. Every pod loads the
full host map, and the leader materializes all per-community aggregates
before suppressing their labels. That makes telemetry collection itself
a scaling constraint.

## What changed

- Fleet-only usage metrics are now the default. `all` remains available
as an explicit rollback mode.
- In fleet-only mode, the poller skips the community host map and all
community-sized query results.
- Exact community counting is replaced with the PostgreSQL planner's
table-size estimate.
- Stock totals are collected as one fixed row, scanning each stock table
once per hour.
- 1d/7d/30d active-user totals come from one bounded 30-day scan per
day.
- Fleet database snapshots run only on a proved read replica, with a
telemetry-specific freshness budget and server-side statement timeouts.
If no such replica is available, the collection is skipped; it never
falls back to the writer.
- Successful fixed-row snapshots are cached and re-emitted on each poll
tick, so gauges stay scrape-visible without raising query frequency.
- `buzz_usage_snapshot_available{family}` reports whether the latest due
refresh of that family succeeded. After a skipped or failed refresh, the
last values stay visible (with `buzz_usage_snapshot_age_seconds`) but
the family reads unavailable until a 60-second retry succeeds.
- The `all` rollback collector publishes the same availability gauges,
so dashboards gated on availability keep working during a rollback.
- Each due refresh outcome is applied by `apply_stock_refresh` /
`apply_activity_refresh`, the single production mapping from `Ok(Some)`
/ `Ok(None)` / `Err` to cache, availability, skip counter, and retry;
the regression test drives that mapping directly.
- `buzz_usage_snapshot_age_seconds` is emitted in the `all` rollback
too, so age-based alerts keep firing through a rollback.
- Every fleet telemetry attempt that completes lands in
`buzz_db_route_decision`; a query error on a proved reader records
`skipped/replica_error`. An attempt abandoned at the relay deadline
records no route decision and appears only as
`buzz_usage_query_skipped_total{reason="timeout"}`. The shared
`proved_reader` warnings no longer claim "routing to writer", because
the telemetry caller never does (serving callers still record their
writer decision in the route counter).
- The PostgreSQL production-method matrix now exercises
`usage_fleet_stock_snapshot`, so the new `reader/maintenance` pool pair
is proven to be emitted by a real Db method.
- Each fleet collection, reader proof included, runs under a relay-side
`tokio::time::timeout` (10s stock, 20s activity, both above their
server-side `statement_timeout`). A reader that stops answering expires
into the same failure mapping with
`buzz_usage_query_skipped_total{reason="timeout"}`, so it cannot stall
the leader's poller, hold the advisory lock past a liveness check, or
leave a frozen snapshot reporting available.
- Fleet telemetry reader checkouts (`ReaderOperation::Maintenance`) opt
into SQLx `close_on_drop` before their first post-acquire await, so an
abandoned collection's connection is closed within SQLx's 5s
`CLOSE_ON_DROP_TIMEOUT` instead of going through the unbounded
return-to-pool ping while it holds a slot in the reader pool that
serving reads share. Serving reads keep pooled reuse; the cost is one
reader reconnect per fleet refresh.
- On leader demotion, cached snapshots and availability are cleared, and
a fresh collection is forced after the lock is reacquired.
- Fleet-only mode passes no host map to the worker-snapshot storage
emitter. It therefore omits community attribution instead of reporting
every byte as `buzz_storage_unmapped_community_bytes`.
- Lifetime stored-message and active-channel totals are no longer
queried in fleet-only mode. Event throughput remains available through
the existing stored-event counter.
- No test-selector changes: main's `config::tests::` selector in `just
test-unit` and the `run-tests.sh unit` fallback already runs the new
replica-budget config test, and the new fleet schedule, availability,
and storage tests run through the existing bin `tests::` and
`storage_sweep::tests::` selectors.
- Added seeded PostgreSQL contract tests and operator docs covering
rollout, rollback, availability, the no-replica behavior, and leader
election.

No migrations, tables, indexes, or new persistent storage are added.

### What got simpler

The relay-side storage half of the original branch (totals-only S3
listing and `buzz_storage_community_breakdown_available`) was dropped.
Main moved S3 listing out of the relay in #7543 and #7845, so that code
was dead on arrival. The rework adds no storage code path; it only gates
attribution on the presence of a host map.

## Risk

Medium. The default changes which telemetry is emitted:

- In fleet-only mode, `buzz_communities_estimated` replaces the exact
`buzz_communities_total`.
- `buzz_total_messages` and `buzz_total_active_channels` are not
refreshed in fleet-only mode (also called out in the rollout steps of
`docs/usage-metrics.md`).
- Deployments without a verified reader (including local compose and
single-node self-hosts) skip database-derived fleet snapshots instead of
querying the writer, and report both families unavailable.
- Multi-pod dashboards must filter database and storage series to
`buzz_usage_poller_is_leader == 1`.

Setting `BUZZ_USAGE_METRICS_PER_COMMUNITY=all` restores the legacy
collector for temporary rollback.

## Verification

At head `1879ee3aedf4227942120a3b521c8d35fc7839fe` (tree `78477e74`;
merges main `da651ad6`; the PR's diff against main is byte-identical to
`1613823f`'s, minus the Justfile and `run-tests.sh` selector lines that
main's `config::tests::` selector now subsumes):

- On Blox: `cargo fmt --all -- --check` and `cargo clippy -p buzz-db -p
buzz-relay --all-targets -- -D warnings` pass. Relay bin `tests::`
34/34, `config::tests::` 55/55, `storage_sweep::tests::` 10/10 and
`store::usage::tests` pass, as do the four `fleet_` and lock PostgreSQL
tests that use scratch databases, including
`abandoned_fleet_collection_releases_its_reader_slot`.
- Against a freshly migrated PostgreSQL database:
`store::usage::postgres_tests::` 14/14,
`storage_sweep::postgres_tests::` and `buzz-db --test
observability_source` 14/14 pass. The new reader code passes main's
crate-wide typed-acquisition and serving-table policy checks.
- E2E at `1613823f` is reused for this head. It ran in an isolated relay
against a real PostgreSQL primary and streaming standby. In 2/2 runs a
stock read went dark mid-result; the deadline abandoned it at about 10s,
the reader pool's active and size counts dropped to zero about 1s later,
and a fresh activity read and the scheduled stock retry succeeded on new
backends while the dark session stayed open. The `all`-mode and
healthy-replica controls also passed. Main's merged changes are on the
writer path and in startup step timing, and touch none of the reader
checkout, the `close_on_drop` seam, the fleet poller or the deadlines.

At head `1613823fa8ef1a423bc1a499c57bc40d692fcd4f` (tree `12bc6d1b`;
test/docs-only on top of `57749418`), on a Blox workstation:

- `cargo fmt --all -- --check` and `cargo clippy -p buzz-db -p
buzz-relay --all-targets -- -D warnings` pass. Relay bin `tests::` 34/34
pass.
- New PostgreSQL test
`abandoned_fleet_collection_releases_its_reader_slot`: a one-slot reader
pool behind a loopback proxy that goes dark mid-session; the collection
is cut at a deadline, then a later acquire must succeed in under 8s on a
fresh connection. The dark call runs on a fresh `Db` with a cold Aurora
capability cache, so the first post-acquire await is the capability
probe. It passes in 3.4s. With `close_on_drop` removed, or moved below
the probe await, it fails with `PoolTimedOut` after the 15s acquire
timeout. The other three `fleet_` PostgreSQL tests that use scratch
databases pass.
- `each_fleet_family_is_abandoned_at_its_own_deadline` now has an outer
timeout: replacing the stock deadline with a bare `.await` fails it
(`Elapsed`) instead of hanging.

At head `fd194ec79d6c6796d6764564de225ad56485cfc7` (tree `79e94727`,
main merged at `932228936`), on a Blox workstation:

- `cargo fmt --all -- --check` and `cargo clippy -p buzz-relay -p
buzz-db --all-targets -- -D warnings` pass.
- Relay bin `tests::` nextest selection: 27/27 pass (plus
`store::usage::tests`), including
`fleet_refresh_outcomes_drive_availability_through_production_mapping`
and the new `stalled_fleet_collection_times_out_into_failed_refresh`
(paused time: two never-completing collections are abandoned at exactly
10s + 20s; both families read unavailable with cached values kept,
`reason="timeout"` skips, retry scheduled; the next tick runs and
recovers; no route decision is recorded for an abandoned attempt) and
`each_fleet_family_is_abandoned_at_its_own_deadline`. Compile-time
asserts keep each deadline above its `statement_timeout`.
- Falsifiability: replacing the stock deadline with a bare `.await`
makes the new test fail (the tick stalls); swapping the two deadlines
fails the per-family test; deleting `schedule.mark_stock_failure(now)`
from the `Ok(None)` arm fails the mapping test. Trees restored and
re-verified.
- At `970414db`: CI PostgreSQL Tests 809/809 including the
production-method matrix with `reader/maintenance`; Unit Tests ran
`finish_usage_read_records_every_attempt_in_route_decisions`.
- `just ci` / `just test` at `5b18914a` on Blox: not green for
environmental reasons (`mobile-check` could not reach pub.dev;
`_ensure-services` failed so service-backed suites did not run;
`buzz-acp git_bootstrap` fails identically on `origin/main`). PostgreSQL
and integration coverage comes from GitHub CI at this head.

E2E at the previous head `0a01e5f22` (isolated relay, real Postgres
primary plus streaming replica, worker storage snapshot, desktop
frontend in relay mode): with no reader, both families report
skipped/unavailable and route `skipped/disabled` (no writer queries);
fleet-only storage emits totals without `buzz_community_storage_*` or
unmapped bytes; `all` mode reports both availability gauges `1` with
per-community storage series; with a fresh replica, both fleet paths
route `replica/fresh` on `reader/maintenance` acquisitions; desktop
send/receive works. The cached success → due failure → recovery
transition was not reached live (hourly/daily cadence); it is covered by
the seam test above.

The original branch also recorded live verification at `aa3e7023`,
before this rework: reduced-idle cache re-emission, two-relay leader
takeover with demoted availability clearing, the no-reader skip with
60-second retry, `all` rollback restoring per-community series, and a
desktop smoke test. Those runs predate the availability and storage
changes and are not claimed for this head.

## Linear tracking

Related work: BUZZ-167.

---------

Signed-off-by: tornquist <tornquist@squareup.com>
Signed-off-by: Luke Tornquist <tornquist@squareup.com>
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 15:46:18 -04:00
committed by GitHub
parent c96d707f6e
commit 6d07a29137
11 changed files with 1976 additions and 89 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
.env.example
View File

@@ -27,6 +27,16 @@ PGUSER=buzz
PGPASSWORD=buzz_dev
PGDATABASE=buzz
# Fleet usage telemetry is replica-only and independent of serving-read
# routing. This freshness budget defaults to 30000ms; 0 disables the database
# telemetry families. When the reader is absent, stale, or unavailable, the
# relay skips the collection and reports its availability as 0 instead of
# falling back to the writer.
# BUZZ_USAGE_METRICS_REPLICA_MAX_AGE_MS=30000
# Per-community metric series are disabled by default to bound relay memory and
# monitoring cardinality. `all` is the temporary rollback mode.
# BUZZ_USAGE_METRICS_PER_COMMUNITY=off
# -----------------------------------------------------------------------------
# Redis 7
# -----------------------------------------------------------------------------

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

@@ -36,7 +36,7 @@ pub use runtime::{
};
/// Valid low-cardinality `(pool_role, operation)` pairs for pool-acquisition telemetry.
pub const DB_POOL_ACQUIRE_VALID_PAIRS: [(&str, &str); 11] =
pub const DB_POOL_ACQUIRE_VALID_PAIRS: [(&str, &str); 12] =
runtime::observability::POOL_ACQUIRE_VALID_PAIRS;
/// Raw Prometheus series ceiling per relay pod for the operation-aware contract.

83
crates/buzz-db/src/runtime/mod.rs
View File

@@ -219,6 +219,10 @@ pub struct Db {
/// bounded-stale read semantics are a product decision, not an
/// invariant, so the gate ships off.
pub(crate) replica_read_max_age: Option<Duration>,
/// Replica freshness budget used only by the leader's fleet telemetry
/// queries. This is independent of serving-read rollout policy: telemetry
/// may be stale or skipped, but it must never fall back to the writer.
pub(crate) usage_metrics_replica_max_age: Option<Duration>,
/// Whether the reader endpoint supports the Aurora PostgreSQL identity
/// function ([`replica_fence::AURORA_IDENTITY_FN`]) — probed
/// once per process on the first routed read (on a plain autocommit
@@ -598,6 +602,10 @@ pub struct DbConfig {
/// than the staleness gate never routes anyway, so a larger budget
/// would only misrepresent the config.
pub replica_read_max_age_ms: u64,
/// Replica freshness budget for usage telemetry, in milliseconds
/// (`BUZZ_USAGE_METRICS_REPLICA_MAX_AGE_MS`). `0` disables fleet database
/// telemetry. This does not enable replica routing for serving reads.
pub usage_metrics_replica_max_age_ms: u64,
/// Session `lock_timeout` in milliseconds for writer connections (env
/// `BUZZ_DB_LOCK_TIMEOUT_MS`). `0` disables the timeout.
pub lock_timeout_ms: u64,
@@ -628,6 +636,7 @@ impl Default for DbConfig {
max_lifetime_secs: 1800,
idle_timeout_secs: 600,
replica_read_max_age_ms: 0,
usage_metrics_replica_max_age_ms: 30_000,
lock_timeout_ms: DEFAULT_LOCK_TIMEOUT_MS,
idle_txn_timeout_ms: DEFAULT_IDLE_TXN_TIMEOUT_MS,
statement_timeout_ms: 0,
@@ -690,6 +699,8 @@ impl Db {
None => None,
};
let replica_read_max_age = read_budget_from_ms(config.replica_read_max_age_ms);
let usage_metrics_replica_max_age =
read_budget_from_ms(config.usage_metrics_replica_max_age_ms);
Ok(Self {
pool,
max_connections: config.max_connections,
@@ -697,6 +708,7 @@ impl Db {
read_max_connections,
fence: std::sync::Arc::new(replica_fence::ReplicaFence::new()),
replica_read_max_age,
usage_metrics_replica_max_age,
reader_aurora_identity: std::sync::Arc::new(std::sync::OnceLock::new()),
})
}
@@ -931,6 +943,7 @@ impl Db {
read_pool: None,
fence: std::sync::Arc::new(replica_fence::ReplicaFence::new()),
replica_read_max_age: None,
usage_metrics_replica_max_age: None,
reader_aurora_identity: std::sync::Arc::new(std::sync::OnceLock::new()),
}
}
@@ -950,6 +963,7 @@ impl Db {
read_pool: Some(read_pool),
fence: std::sync::Arc::new(replica_fence::ReplicaFence::new()),
replica_read_max_age: None,
usage_metrics_replica_max_age: Some(replica_fence::FENCE_STALENESS),
reader_aurora_identity: std::sync::Arc::new(std::sync::OnceLock::new()),
}
}
@@ -1055,20 +1069,24 @@ impl Db {
// `read_pool` separately would spend a second budget whenever the
// capability is uncached — i.e. after a failed boot ping, which is
// precisely the reader-unavailable case the bound must hold for.
let conn = match observability::acquire_reader_with_legacy_metrics(read_pool, operation)
.await
{
Ok(conn) => conn,
Err(sqlx::Error::PoolTimedOut) => {
tracing::warn!("reader pool acquire timed out; routing to writer");
return Err("reader_acquire_timeout");
}
Err(e) => {
tracing::warn!(error = %e, "reader connection acquire failed; routing to writer");
return Err("reader_validation_error");
}
};
let conn =
match observability::acquire_reader_with_legacy_metrics(read_pool, operation).await {
Ok(conn) => conn,
Err(sqlx::Error::PoolTimedOut) => {
tracing::warn!("reader pool acquire timed out");
return Err("reader_acquire_timeout");
}
Err(e) => {
tracing::warn!(error = %e, "reader connection acquire failed");
return Err("reader_validation_error");
}
};
let mut conn = conn;
if operation.bounded_release() {
// Must precede the first post-acquire await: a caller deadline
// can drop this checkout at any of them.
conn.close_on_drop();
}
let aurora = self.reader_aurora_capability_on(&mut conn).await;
let mut tx = match sqlx::Transaction::begin(
conn,
@@ -1103,11 +1121,11 @@ impl Db {
// between samples entirely, so absence of elevated active is
// NOT evidence of a cold connect.
Err(sqlx::Error::PoolTimedOut) => {
tracing::warn!("reader pool acquire timed out; routing to writer");
tracing::warn!("reader pool acquire timed out");
return Err("reader_acquire_timeout");
}
Err(e) => {
tracing::warn!(error = %e, "reader transaction begin failed; routing to writer");
tracing::warn!(error = %e, "reader transaction begin failed");
return Err("reader_validation_error");
}
};
@@ -1115,7 +1133,7 @@ impl Db {
Ok(Some(observation)) => observation,
Ok(None) => return Err("reader_validation_error"),
Err(e) => {
tracing::warn!(error = %e, "heartbeat observation failed; routing to writer");
tracing::warn!(error = %e, "heartbeat observation failed");
return Err("reader_validation_error");
}
};
@@ -1368,6 +1386,39 @@ impl Db {
Ok(result)
}
/// Route a usage-telemetry query only to a proved, sufficiently fresh
/// reader. Unlike serving reads, an unavailable reader is recorded as
/// skipped and never represented as or executed on the writer.
pub(crate) async fn route_usage_read(
&self,
path: &'static str,
) -> Option<(sqlx::Transaction<'static, sqlx::Postgres>, &'static str)> {
let skip = |reason| {
Self::record_route(path, "skipped", reason);
None
};
let Some(read_pool) = &self.read_pool else {
return skip("disabled");
};
let Some(budget) = self.usage_metrics_replica_max_age else {
return skip("disabled");
};
let Some(newest) = self.fence.newest() else {
return skip("uninitialized");
};
if newest.committed_at.elapsed() > budget {
return skip("stale");
}
match self
.proved_reader(read_pool, observability::ReaderOperation::Maintenance)
.await
{
Ok((tx, entry)) if entry.committed_at.elapsed() <= budget => Some((tx, "fresh")),
Ok((_tx, _entry)) => skip("stale"),
Err(reason) => skip(reason),
}
}
/// Shared route decision for one read: evaluate the predicate against a
/// proved reader session and record the decision. Fail closed to the
/// writer everywhere.

41
crates/buzz-db/src/runtime/observability.rs
View File

@@ -28,6 +28,7 @@ enum PoolOperation {
ReaderSubscriptionHistory,
WriterEventWrite,
WriterMaintenance,
ReaderMaintenance,
}
/// Writer-pool operations. Reader-only combinations cannot be constructed.
@@ -76,14 +77,17 @@ pub(crate) enum ReaderOperation {
Bootstrap,
Authorization,
SubscriptionHistory,
/// Leader-only background reads such as fleet usage telemetry.
Maintenance,
}
impl ReaderOperation {
#[cfg(test)]
const ALL: [Self; 3] = [
const ALL: [Self; 4] = [
Self::Bootstrap,
Self::Authorization,
Self::SubscriptionHistory,
Self::Maintenance,
];
const fn pair(self) -> PoolOperation {
@@ -91,12 +95,26 @@ impl ReaderOperation {
Self::Bootstrap => PoolOperation::ReaderBootstrap,
Self::Authorization => PoolOperation::ReaderAuthorization,
Self::SubscriptionHistory => PoolOperation::ReaderSubscriptionHistory,
Self::Maintenance => PoolOperation::ReaderMaintenance,
}
}
/// Whether a dropped checkout must close the connection under SQLx's
/// bounded `close_on_drop` instead of returning it to the pool.
///
/// Maintenance reads run under relay-side deadlines that drop them
/// mid-query when a reader goes dark. SQLx's return-to-pool path pings
/// that connection with no timeout while still holding a pool slot, and
/// the reader pool is shared with serving reads, so each abandoned read
/// would strand a serving slot until the kernel gives up on the socket.
/// Serving reads keep pooled reuse.
pub(crate) const fn bounded_release(self) -> bool {
matches!(self, Self::Maintenance)
}
}
impl PoolOperation {
pub(crate) const ALL: [Self; 11] = [
pub(crate) const ALL: [Self; 12] = [
Self::WriterBootstrap,
Self::ReaderBootstrap,
Self::WriterReadiness,
@@ -108,13 +126,15 @@ impl PoolOperation {
Self::ReaderSubscriptionHistory,
Self::WriterEventWrite,
Self::WriterMaintenance,
Self::ReaderMaintenance,
];
pub(crate) const fn pool_role(self) -> &'static str {
match self {
Self::ReaderBootstrap | Self::ReaderAuthorization | Self::ReaderSubscriptionHistory => {
DbPoolRole::Reader.as_str()
}
Self::ReaderBootstrap
| Self::ReaderAuthorization
| Self::ReaderSubscriptionHistory
| Self::ReaderMaintenance => DbPoolRole::Reader.as_str(),
_ => DbPoolRole::Writer.as_str(),
}
}
@@ -130,7 +150,7 @@ impl PoolOperation {
"subscription_history"
}
Self::WriterEventWrite => "event_write",
Self::WriterMaintenance => "maintenance",
Self::WriterMaintenance | Self::ReaderMaintenance => "maintenance",
}
}
@@ -139,7 +159,7 @@ impl PoolOperation {
}
}
pub(crate) const POOL_ACQUIRE_VALID_PAIRS: [(&str, &str); 11] = [
pub(crate) const POOL_ACQUIRE_VALID_PAIRS: [(&str, &str); 12] = [
(DbPoolRole::Writer.as_str(), "bootstrap"),
(DbPoolRole::Reader.as_str(), "bootstrap"),
(DbPoolRole::Writer.as_str(), "readiness"),
@@ -151,9 +171,10 @@ pub(crate) const POOL_ACQUIRE_VALID_PAIRS: [(&str, &str); 11] = [
(DbPoolRole::Reader.as_str(), "subscription_history"),
(DbPoolRole::Writer.as_str(), "event_write"),
(DbPoolRole::Writer.as_str(), "maintenance"),
(DbPoolRole::Reader.as_str(), "maintenance"),
];
/// Eleven valid pairs × (12 histogram series + 1 start counter + 4 outcome counters + 1 gauge).
/// Twelve valid pairs × (12 histogram series + 1 start counter + 4 outcome counters + 1 gauge).
pub(crate) const POOL_ACQUIRE_RAW_SERIES_PER_POD: usize = POOL_ACQUIRE_VALID_PAIRS.len() * 18;
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
@@ -610,9 +631,10 @@ mod tests {
PoolOperation::ReaderBootstrap,
PoolOperation::ReaderAuthorization,
PoolOperation::ReaderSubscriptionHistory,
PoolOperation::ReaderMaintenance,
]
);
assert_eq!(super::POOL_ACQUIRE_RAW_SERIES_PER_POD, 198);
assert_eq!(super::POOL_ACQUIRE_RAW_SERIES_PER_POD, 216);
assert_eq!(
LockType::ALL.map(LockType::as_str),
[
@@ -1494,6 +1516,7 @@ mod tests {
let _ = routed_db
.query_events_routed("pool_operation_matrix_reader", &query)
.await;
let _ = routed_db.usage_fleet_stock_snapshot().await;
routed_db.refresh_pool_waiter_metrics();
let snapshot = snapshotter.snapshot().into_vec();

771
crates/buzz-db/src/store/usage.rs
View File

@@ -13,13 +13,184 @@
//! to Prometheus labels and calls `metrics::gauge!(...).set(...)`.
use buzz_datastore_tracing::datastore_span;
use chrono::{DateTime, Duration, Utc};
use sqlx::postgres::PgConnection;
use sqlx::{Connection as _, PgPool};
use sqlx::{Connection as _, Executor, PgPool, Postgres};
use uuid::Uuid;
use crate::error::Result;
use crate::{observability, Db};
/// Fixed-row fleet adoption snapshot used when per-community telemetry is disabled.
#[derive(Debug, Clone, Copy, Default, sqlx::FromRow)]
pub struct FleetStockSnapshot {
/// Planner estimate for the number of communities.
pub communities_estimated: i64,
/// Active human users.
pub users_human: i64,
/// Active agent users.
pub users_agent: i64,
/// Non-deleted stream channels.
pub channels_stream: i64,
/// Non-deleted forum channels.
pub channels_forum: i64,
/// Non-deleted direct-message channels.
pub channels_dm: i64,
/// Non-deleted workflow channels.
pub channels_workflow: i64,
/// Relay owners.
pub members_owner: i64,
/// Relay administrators.
pub members_admin: i64,
/// Relay members.
pub members_member: i64,
/// Active workflows.
pub workflows_active: i64,
/// Disabled workflows.
pub workflows_disabled: i64,
/// Archived workflows.
pub workflows_archived: i64,
/// Registered Git repositories.
pub git_repos: i64,
}
/// Fixed-row 1d/7d/30d active-user snapshot derived from one 30-day scan.
#[derive(Debug, Clone, Copy, Default, sqlx::FromRow)]
pub struct FleetActiveUsersSnapshot {
/// Human publishers active in the last day.
pub human_1d: i64,
/// Agent publishers active in the last day.
pub agent_1d: i64,
/// Unclassified publishers active in the last day.
pub unknown_1d: i64,
/// Human publishers active in the last seven days.
pub human_7d: i64,
/// Agent publishers active in the last seven days.
pub agent_7d: i64,
/// Unclassified publishers active in the last seven days.
pub unknown_7d: i64,
/// Human publishers active in the last thirty days.
pub human_30d: i64,
/// Agent publishers active in the last thirty days.
pub agent_30d: i64,
/// Unclassified publishers active in the last thirty days.
pub unknown_30d: i64,
}
/// Collect one fixed-row fleet adoption snapshot on the supplied executor.
pub async fn fleet_stock_snapshot_on<'e, E>(executor: E) -> Result<FleetStockSnapshot>
where
E: Executor<'e, Database = Postgres>,
{
Ok(sqlx::query_as::<_, FleetStockSnapshot>(
r#"
WITH user_counts AS (
SELECT
COUNT(*) FILTER (WHERE agent_owner_pubkey IS NULL) AS human,
COUNT(*) FILTER (WHERE agent_owner_pubkey IS NOT NULL) AS agent
FROM users
WHERE deactivated_at IS NULL
),
channel_counts AS (
SELECT
COUNT(*) FILTER (WHERE channel_type = 'stream') AS stream,
COUNT(*) FILTER (WHERE channel_type = 'forum') AS forum,
COUNT(*) FILTER (WHERE channel_type = 'dm') AS dm,
COUNT(*) FILTER (WHERE channel_type = 'workflow') AS workflow
FROM channels
WHERE deleted_at IS NULL
),
member_counts AS (
SELECT
COUNT(*) FILTER (WHERE role = 'owner') AS owner,
COUNT(*) FILTER (WHERE role = 'admin') AS admin,
COUNT(*) FILTER (WHERE role = 'member') AS member
FROM relay_members
),
workflow_counts AS (
SELECT
COUNT(*) FILTER (WHERE status = 'active') AS active,
COUNT(*) FILTER (WHERE status = 'disabled') AS disabled,
COUNT(*) FILTER (WHERE status = 'archived') AS archived
FROM workflows
)
SELECT
COALESCE((SELECT GREATEST(reltuples, 0)::bigint FROM pg_class WHERE oid = 'communities'::regclass), 0) AS communities_estimated,
users.human AS users_human,
users.agent AS users_agent,
channels.stream AS channels_stream,
channels.forum AS channels_forum,
channels.dm AS channels_dm,
channels.workflow AS channels_workflow,
members.owner AS members_owner,
members.admin AS members_admin,
members.member AS members_member,
workflows.active AS workflows_active,
workflows.disabled AS workflows_disabled,
workflows.archived AS workflows_archived,
(SELECT COUNT(*) FROM git_repo_names) AS git_repos
FROM user_counts users
CROSS JOIN channel_counts channels
CROSS JOIN member_counts members
CROSS JOIN workflow_counts workflows
"#,
)
.fetch_one(executor)
.await?)
}
/// Collect all fleet active-user windows with one bounded 30-day event scan.
pub async fn fleet_active_users_on<'e, E>(
executor: E,
observed_at: DateTime<Utc>,
) -> Result<FleetActiveUsersSnapshot>
where
E: Executor<'e, Database = Postgres>,
{
let start_30d = observed_at - Duration::days(30);
let start_7d = observed_at - Duration::days(7);
let start_1d = observed_at - Duration::days(1);
Ok(sqlx::query_as::<_, FleetActiveUsersSnapshot>(
r#"
WITH publishers AS (
SELECT
e.community_id,
e.pubkey,
MAX(e.created_at) AS last_active_at,
CASE
WHEN u.pubkey IS NULL THEN 'unknown'
WHEN u.agent_owner_pubkey IS NULL THEN 'human'
ELSE 'agent'
END AS author_type
FROM events e
LEFT JOIN users u
ON u.community_id = e.community_id AND u.pubkey = e.pubkey
WHERE e.created_at >= $1
AND e.created_at < $2
AND e.deleted_at IS NULL
GROUP BY e.community_id, e.pubkey, u.pubkey, u.agent_owner_pubkey
)
SELECT
COUNT(*) FILTER (WHERE last_active_at >= $4 AND author_type = 'human') AS human_1d,
COUNT(*) FILTER (WHERE last_active_at >= $4 AND author_type = 'agent') AS agent_1d,
COUNT(*) FILTER (WHERE last_active_at >= $4 AND author_type = 'unknown') AS unknown_1d,
COUNT(*) FILTER (WHERE last_active_at >= $3 AND author_type = 'human') AS human_7d,
COUNT(*) FILTER (WHERE last_active_at >= $3 AND author_type = 'agent') AS agent_7d,
COUNT(*) FILTER (WHERE last_active_at >= $3 AND author_type = 'unknown') AS unknown_7d,
COUNT(*) FILTER (WHERE author_type = 'human') AS human_30d,
COUNT(*) FILTER (WHERE author_type = 'agent') AS agent_30d,
COUNT(*) FILTER (WHERE author_type = 'unknown') AS unknown_30d
FROM publishers
"#,
)
.bind(start_30d)
.bind(observed_at)
.bind(start_7d)
.bind(start_1d)
.fetch_one(executor)
.await?)
}
/// Owns the detached Postgres session holding the relay usage-metrics advisory lock.
///
/// The connection deliberately does not return to the main pool: session advisory
@@ -458,6 +629,70 @@ impl Db {
}
}
/// Collect fleet adoption stocks from a proved read-replica snapshot.
///
/// Returns `None` instead of falling back to the writer when a proved
/// reader is unavailable. Usage telemetry is allowed to be stale or
/// skipped and must not add load to the serving database.
#[datastore_span(name = "usage_fleet_stock_snapshot", system = "postgresql")]
pub async fn usage_fleet_stock_snapshot(&self) -> Result<Option<FleetStockSnapshot>> {
let path = "usage_fleet_stock";
let Some((mut tx, reason)) = self.route_usage_read(path).await else {
return Ok(None);
};
let collected = async {
sqlx::query("SET LOCAL statement_timeout = '5s'")
.execute(&mut *tx)
.await?;
fleet_stock_snapshot_on(&mut *tx).await
}
.await;
Self::finish_usage_read(path, reason, collected)
}
/// Collect all fleet active-user windows from a proved read-replica snapshot.
///
/// Returns `None` instead of falling back to the writer when a proved
/// reader is unavailable.
#[datastore_span(name = "usage_fleet_active_users", system = "postgresql")]
pub async fn usage_fleet_active_users(
&self,
observed_at: DateTime<Utc>,
) -> Result<Option<FleetActiveUsersSnapshot>> {
let path = "usage_fleet_active_users";
let Some((mut tx, reason)) = self.route_usage_read(path).await else {
return Ok(None);
};
let collected = async {
sqlx::query("SET LOCAL statement_timeout = '15s'")
.execute(&mut *tx)
.await?;
fleet_active_users_on(&mut *tx, observed_at).await
}
.await;
Self::finish_usage_read(path, reason, collected)
}
/// Record the route outcome of a proved-reader telemetry query. A query
/// error is a skipped attempt (`replica_error`), never a writer fallback,
/// so every attempt that completes appears in `buzz_db_route_decision`.
fn finish_usage_read<T>(
path: &'static str,
reason: &'static str,
collected: Result<T>,
) -> Result<Option<T>> {
match collected {
Ok(snapshot) => {
Self::record_route(path, "replica", reason);
Ok(Some(snapshot))
}
Err(error) => {
Self::record_route(path, "skipped", "replica_error");
Err(error)
}
}
}
/// Return total number of communities on this relay.
#[datastore_span(name = "usage_community_count", system = "postgresql")]
pub async fn usage_community_count(&self) -> Result<i64> {
@@ -539,6 +774,79 @@ impl Db {
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::error::DbError;
use metrics_util::debugging::{DebugValue, DebuggingRecorder};
fn route_decisions(recorder: &DebuggingRecorder) -> Vec<(String, String, String, u64)> {
let mut decisions: Vec<_> = recorder
.snapshotter()
.snapshot()
.into_vec()
.into_iter()
.filter_map(|(key, _, _, value)| {
if key.key().name() != "buzz_db_route_decision" {
return None;
}
let label = |name: &str| {
key.key()
.labels()
.find(|label| label.key() == name)
.map(|label| label.value().to_owned())
.unwrap_or_default()
};
let DebugValue::Counter(count) = value else {
return None;
};
Some((label("path"), label("decision"), label("reason"), count))
})
.collect();
decisions.sort();
decisions
}
/// Every telemetry attempt on a proved reader lands in the route counter:
/// a completed query as `replica/<reason>`, a failed one as
/// `skipped/replica_error` with the error propagated (never a writer
/// fallback).
#[test]
fn finish_usage_read_records_every_attempt_in_route_decisions() {
let recorder = DebuggingRecorder::new();
let (ok, err) = metrics::with_local_recorder(&recorder, || {
(
Db::finish_usage_read("usage_fleet_stock", "fresh", Ok(7_u8)),
Db::finish_usage_read::<u8>(
"usage_fleet_active_users",
"fresh",
Err(DbError::AuthEventRejected),
),
)
});
assert!(matches!(ok, Ok(Some(7))));
assert!(matches!(err, Err(DbError::AuthEventRejected)));
assert_eq!(
route_decisions(&recorder),
vec![
(
"usage_fleet_active_users".to_owned(),
"skipped".to_owned(),
"replica_error".to_owned(),
1
),
(
"usage_fleet_stock".to_owned(),
"replica".to_owned(),
"fresh".to_owned(),
1
),
]
);
}
}
#[cfg(test)]
mod postgres_tests {
use super::*;
@@ -931,6 +1239,24 @@ mod postgres_tests {
assert!(after > before, "count should increase after insert");
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn fleet_snapshots_have_fixed_result_shapes() {
let pool = get_pool().await;
let stock = fleet_stock_snapshot_on(&pool)
.await
.expect("fleet stock snapshot");
assert!(stock.communities_estimated >= 0);
let now = chrono::Utc::now();
let activity = fleet_active_users_on(&pool, now)
.await
.expect("fleet activity snapshot");
assert!(activity.human_1d >= 0);
assert!(activity.human_7d >= activity.human_1d);
assert!(activity.human_30d >= activity.human_7d);
}
/// git_repo_counts queries git_repo_names (not git_repos) and is scoped per community.
#[tokio::test]
#[ignore = "requires Postgres"]
@@ -1067,4 +1393,447 @@ mod postgres_tests {
"no stream row after last channel deleted — poller will zero-fill"
);
}
async fn insert_metric_event(
pool: &PgPool,
community_id: Uuid,
pubkey: &[u8],
created_at: DateTime<Utc>,
deleted: bool,
) {
sqlx::query(
"INSERT INTO events \
(community_id, id, pubkey, created_at, kind, tags, content, sig, received_at, deleted_at) \
VALUES ($1, $2, $3, $4, 9, '[]', '', $5, $4, \
CASE WHEN $6 THEN $4 ELSE NULL END)",
)
.bind(community_id)
.bind(random_pubkey())
.bind(pubkey)
.bind(created_at)
.bind(vec![0u8; 64])
.bind(deleted)
.execute(pool)
.await
.expect("insert metric event");
}
/// Fixed-row fleet SQL must preserve every inclusion/exclusion rule while
/// keeping its result shape independent of community cardinality.
#[tokio::test]
#[ignore = "requires Postgres"]
async fn fleet_snapshots_match_seeded_stock_and_activity_exactly() {
let admin_url = crate::test_support::database_url();
let admin = PgPoolOptions::new()
.max_connections(1)
.connect(&admin_url)
.await
.expect("connect admin to create scratch db");
let (pool, scratch_name) = create_scratch_db(&admin, "usage_snapshot").await;
let (community_id, _, _) = make_community(&pool).await;
let human = random_pubkey();
let agent = random_pubkey();
let inactive = random_pubkey();
insert_user(&pool, community_id, &human, false).await;
sqlx::query(
"INSERT INTO users (community_id, pubkey, agent_owner_pubkey) VALUES ($1, $2, $3)",
)
.bind(community_id)
.bind(&agent)
.bind(&human)
.execute(&pool)
.await
.expect("insert agent");
insert_user(&pool, community_id, &inactive, false).await;
sqlx::query(
"UPDATE users SET deactivated_at = NOW() WHERE community_id = $1 AND pubkey = $2",
)
.bind(community_id)
.bind(&inactive)
.execute(&pool)
.await
.expect("deactivate user");
for channel_type in ["stream", "forum", "dm", "workflow"] {
sqlx::query(
"INSERT INTO channels (id, community_id, name, channel_type, visibility, created_by) \
VALUES ($1, $2, $3, $4::channel_type, 'open', $5)",
)
.bind(Uuid::new_v4())
.bind(community_id)
.bind(format!("metric-{channel_type}"))
.bind(channel_type)
.bind(&human)
.execute(&pool)
.await
.expect("insert live channel");
}
sqlx::query(
"INSERT INTO channels (id, community_id, name, channel_type, visibility, created_by, deleted_at) \
VALUES ($1, $2, 'deleted-stream', 'stream', 'open', $3, NOW())",
)
.bind(Uuid::new_v4())
.bind(community_id)
.bind(&human)
.execute(&pool)
.await
.expect("insert deleted channel");
for role in ["owner", "admin", "member"] {
sqlx::query(
"INSERT INTO relay_members (community_id, pubkey, role) VALUES ($1, $2, $3)",
)
.bind(community_id)
.bind(format!("{role}-pubkey"))
.bind(role)
.execute(&pool)
.await
.expect("insert relay member");
}
for status in ["active", "disabled", "archived"] {
sqlx::query(
"INSERT INTO workflows \
(community_id, id, name, owner_pubkey, definition, definition_hash, status, enabled) \
VALUES ($1, $2, $3, $4, '{}', $5, $6::workflow_status, $7)",
)
.bind(community_id)
.bind(Uuid::new_v4())
.bind(format!("metric-{status}"))
.bind(&human)
.bind(vec![0u8; 32])
.bind(status)
.bind(status == "active")
.execute(&pool)
.await
.expect("insert workflow");
}
sqlx::query(
"INSERT INTO git_repo_names (community_id, repo_id, owner_pubkey) VALUES ($1, 'metric-repo', $2)",
)
.bind(community_id)
.bind(hex::encode(&human))
.execute(&pool)
.await
.expect("insert git repo");
sqlx::query("ANALYZE communities")
.execute(&pool)
.await
.expect("refresh planner estimate");
let observed_at = Utc::now();
let unknown = random_pubkey();
insert_metric_event(
&pool,
community_id,
&human,
observed_at - Duration::hours(12),
false,
)
.await;
insert_metric_event(
&pool,
community_id,
&human,
observed_at - Duration::days(3),
false,
)
.await;
insert_metric_event(
&pool,
community_id,
&agent,
observed_at - Duration::days(3),
false,
)
.await;
insert_metric_event(
&pool,
community_id,
&unknown,
observed_at - Duration::days(20),
false,
)
.await;
insert_metric_event(
&pool,
community_id,
&unknown,
observed_at - Duration::days(31),
false,
)
.await;
insert_metric_event(
&pool,
community_id,
&unknown,
observed_at + Duration::hours(1),
false,
)
.await;
insert_metric_event(
&pool,
community_id,
&unknown,
observed_at - Duration::hours(1),
true,
)
.await;
let stock = fleet_stock_snapshot_on(&pool).await.expect("fleet stock");
assert_eq!(stock.communities_estimated, 1);
assert_eq!((stock.users_human, stock.users_agent), (1, 1));
assert_eq!(
(
stock.channels_stream,
stock.channels_forum,
stock.channels_dm,
stock.channels_workflow,
),
(1, 1, 1, 1)
);
assert_eq!(
(
stock.members_owner,
stock.members_admin,
stock.members_member
),
(1, 1, 1)
);
assert_eq!(
(
stock.workflows_active,
stock.workflows_disabled,
stock.workflows_archived,
),
(1, 1, 1)
);
assert_eq!(stock.git_repos, 1);
let activity = fleet_active_users_on(&pool, observed_at)
.await
.expect("fleet activity");
assert_eq!(
(activity.human_1d, activity.agent_1d, activity.unknown_1d),
(1, 0, 0)
);
assert_eq!(
(activity.human_7d, activity.agent_7d, activity.unknown_7d),
(1, 1, 0)
);
assert_eq!(
(activity.human_30d, activity.agent_30d, activity.unknown_30d,),
(1, 1, 1)
);
drop_scratch_db(&admin, pool, &scratch_name).await;
}
/// Fleet collection must read a proved replica and skip instead of
/// silently adding load to the writer when no reader is configured.
#[tokio::test]
#[ignore = "requires Postgres"]
async fn fleet_collection_is_replica_only_and_skips_without_reader() {
let admin_url = crate::test_support::database_url();
let admin = PgPoolOptions::new()
.max_connections(1)
.connect(&admin_url)
.await
.expect("connect admin to create scratch db");
let (writer, writer_name) = create_scratch_db(&admin, "usage_writer").await;
let (reader, reader_name) = create_scratch_db(&admin, "usage_reader").await;
let (writer_community, _, _) = make_community(&writer).await;
let (reader_community, _, _) = make_community(&reader).await;
insert_user(&writer, writer_community, &random_pubkey(), false).await;
insert_user(&reader, reader_community, &random_pubkey(), false).await;
insert_user(&reader, reader_community, &random_pubkey(), false).await;
let without_reader = Db::from_pool(writer.clone());
assert!(
without_reader
.usage_fleet_stock_snapshot()
.await
.expect("skip without reader")
.is_none(),
"telemetry must not fall back to the writer"
);
let db = Db::from_pools(writer.clone(), reader.clone());
db.fence().force_open_for_tests(Utc::now());
let snapshot = db
.usage_fleet_stock_snapshot()
.await
.expect("replica collection")
.expect("fresh proved reader");
assert_eq!(
snapshot.users_human, 2,
"fixture proves the reader served the query"
);
drop(db);
drop_scratch_db(&admin, reader, &reader_name).await;
drop_scratch_db(&admin, writer, &writer_name).await;
}
/// Loopback TCP proxy that can make the reader go dark. While dark, any
/// session that receives bytes stops relaying for good but keeps both
/// sockets open, like a replica that stops answering mid-query. Sessions
/// opened after the proxy comes back relay normally.
struct DarkeningProxy {
port: u16,
dark: std::sync::Arc<std::sync::atomic::AtomicBool>,
}
impl DarkeningProxy {
async fn spawn(upstream_host: String, upstream_port: u16) -> Self {
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
async fn pump(
mut from: tokio::net::tcp::OwnedReadHalf,
mut to: tokio::net::tcp::OwnedWriteHalf,
dark: Arc<AtomicBool>,
) {
let mut buf = [0u8; 8192];
loop {
let Ok(n) = from.read(&mut buf).await else {
return;
};
if n == 0 {
return;
}
if dark.load(Ordering::SeqCst) {
// Swallow the bytes and hold both sockets open.
let _held = (from, to);
std::future::pending::<()>().await;
return;
}
if to.write_all(&buf[..n]).await.is_err() {
return;
}
}
}
let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
.await
.expect("bind proxy");
let port = listener.local_addr().expect("proxy addr").port();
let dark = Arc::new(AtomicBool::new(false));
let accept_dark = dark.clone();
tokio::spawn(async move {
while let Ok((client, _)) = listener.accept().await {
let Ok(server) =
tokio::net::TcpStream::connect((upstream_host.as_str(), upstream_port))
.await
else {
continue;
};
let (client_read, client_write) = client.into_split();
let (server_read, server_write) = server.into_split();
tokio::spawn(pump(client_read, server_write, accept_dark.clone()));
tokio::spawn(pump(server_read, client_write, accept_dark.clone()));
}
});
Self { port, dark }
}
fn set_dark(&self, dark: bool) {
self.dark.store(dark, std::sync::atomic::Ordering::SeqCst);
}
}
/// A fleet collection abandoned at its relay-side deadline on a reader
/// that went dark mid-query must release its slot in the shared reader
/// pool within SQLx's bounded close-on-drop, not hold it until the kernel
/// gives up on the socket. SQLx's default return-to-pool path pings the
/// dark connection with no timeout while holding the slot.
#[tokio::test]
#[ignore = "requires Postgres"]
async fn abandoned_fleet_collection_releases_its_reader_slot() {
use sqlx::postgres::PgConnectOptions;
use std::time::{Duration, Instant};
let admin_url = crate::test_support::database_url();
let admin = PgPoolOptions::new()
.max_connections(1)
.connect(&admin_url)
.await
.expect("connect admin to create scratch db");
let (writer, writer_name) = create_scratch_db(&admin, "usage_writer").await;
let (reader, reader_name) = create_scratch_db(&admin, "usage_reader").await;
let (reader_community, _, _) = make_community(&reader).await;
insert_user(&reader, reader_community, &random_pubkey(), false).await;
let reader_options = reader.connect_options();
let proxy = DarkeningProxy::spawn(
reader_options.get_host().to_owned(),
reader_options.get_port(),
)
.await;
let proxied: PgConnectOptions =
(*reader_options).clone().host("127.0.0.1").port(proxy.port);
// One slot, so a stranded slot blocks every later acquire. No
// before-acquire ping, so the stall lands after the checkout is
// handed out, where the relay deadline drops it.
let read_pool = PgPoolOptions::new()
.max_connections(1)
.min_connections(0)
.test_before_acquire(false)
.acquire_timeout(Duration::from_secs(15))
.connect_with(proxied)
.await
.expect("connect reader through proxy");
let db = Db::from_pools(writer.clone(), read_pool.clone());
db.fence().force_open_for_tests(Utc::now());
db.usage_fleet_stock_snapshot()
.await
.expect("healthy collection")
.expect("fresh proved reader");
// Leave one established, idle connection for the collection to use.
drop(read_pool.acquire().await.expect("warm reader connection"));
let warm_deadline = Instant::now() + Duration::from_secs(5);
while read_pool.num_idle() != 1 {
assert!(
Instant::now() < warm_deadline,
"warm connection never returned to the pool"
);
tokio::time::sleep(Duration::from_millis(10)).await;
}
// A fresh Db has a cold Aurora capability cache, so the first
// post-acquire await is the capability probe, as in production after
// a failed boot ping. This pins close-on-drop ahead of that await.
let cold = Db::from_pools(writer.clone(), read_pool.clone());
cold.fence().force_open_for_tests(Utc::now());
proxy.set_dark(true);
tokio::time::timeout(Duration::from_secs(1), cold.usage_fleet_stock_snapshot())
.await
.expect_err("a dark reader must stall the collection until its deadline");
proxy.set_dark(false);
let started = Instant::now();
let mut replacement = read_pool
.acquire()
.await
.expect("the abandoned collection must release its reader slot");
assert!(
started.elapsed() < Duration::from_secs(8),
"slot release must be bounded by close-on-drop, took {:?}",
started.elapsed()
);
let one: i32 = sqlx::query_scalar("SELECT 1")
.fetch_one(&mut *replacement)
.await
.expect("replacement reader is usable");
assert_eq!(one, 1);
drop(replacement);
drop(cold);
drop(db);
read_pool.close().await;
drop_scratch_db(&admin, reader, &reader_name).await;
drop_scratch_db(&admin, writer, &writer_name).await;
}
}

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

@@ -127,6 +127,10 @@ pub struct Config {
/// `0` (the default) disables bounded-staleness replica routing; see
/// [`buzz_db::DbConfig::replica_read_max_age_ms`].
pub replica_read_max_age_ms: u64,
/// Replica freshness budget for fleet usage telemetry
/// (`BUZZ_USAGE_METRICS_REPLICA_MAX_AGE_MS`). Independent of serving-read
/// routing; defaults to 30 seconds and `0` disables telemetry DB queries.
pub usage_metrics_replica_max_age_ms: u64,
/// Upper bound, in milliseconds, of the per-connection random delay applied
/// when sending the `1012 Service Restart` close frame during graceful
@@ -644,6 +648,16 @@ impl Config {
})?,
Err(_) => 0,
};
let usage_metrics_replica_max_age_ms =
match std::env::var("BUZZ_USAGE_METRICS_REPLICA_MAX_AGE_MS") {
Ok(raw) => raw.trim().parse::<u64>().map_err(|_| {
ConfigError::InvalidValue(
"BUZZ_USAGE_METRICS_REPLICA_MAX_AGE_MS must be a non-negative integer"
.to_string(),
)
})?,
Err(_) => 30_000,
};
// Drain jitter: 0 = off (default). Clamp oversized values so every
// delayed close is initiated with ten seconds left in the relay's
@@ -1350,6 +1364,7 @@ impl Config {
database_url,
read_database_url,
replica_read_max_age_ms,
usage_metrics_replica_max_age_ms,
drain_jitter_ms,
redis_url,
redis_pool_size,
@@ -2264,6 +2279,35 @@ mod tests {
}
}
#[test]
fn usage_metrics_replica_budget_defaults_on_independently_of_serving_reads() {
let _guard = ENV_MUTEX.lock().unwrap();
let previous_usage = std::env::var_os("BUZZ_USAGE_METRICS_REPLICA_MAX_AGE_MS");
let previous_serving = std::env::var_os("BUZZ_REPLICA_READ_MAX_AGE_MS");
std::env::remove_var("BUZZ_USAGE_METRICS_REPLICA_MAX_AGE_MS");
std::env::remove_var("BUZZ_REPLICA_READ_MAX_AGE_MS");
let defaults = Config::from_env().expect("config");
std::env::set_var("BUZZ_USAGE_METRICS_REPLICA_MAX_AGE_MS", "0");
let disabled = Config::from_env().expect("config");
if let Some(value) = previous_usage {
std::env::set_var("BUZZ_USAGE_METRICS_REPLICA_MAX_AGE_MS", value);
} else {
std::env::remove_var("BUZZ_USAGE_METRICS_REPLICA_MAX_AGE_MS");
}
if let Some(value) = previous_serving {
std::env::set_var("BUZZ_REPLICA_READ_MAX_AGE_MS", value);
} else {
std::env::remove_var("BUZZ_REPLICA_READ_MAX_AGE_MS");
}
assert_eq!(defaults.replica_read_max_age_ms, 0);
assert_eq!(defaults.usage_metrics_replica_max_age_ms, 30_000);
assert_eq!(disabled.usage_metrics_replica_max_age_ms, 0);
}
#[test]
fn drain_jitter_defaults_off_and_rejects_junk() {
let _guards = env_guards();

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

@@ -65,11 +65,13 @@ fn relay_keypair_from_config(relay_private_key: Option<&str>) -> anyhow::Result<
/// communities would incur five-figure monthly costs if every community always
/// gets a full set of series. This knob is the cost lever.
///
/// Fleet-wide totals (`buzz_total_*`) always emit regardless of mode.
/// Fleet-wide totals are independent of this mode, but DB-backed families emit
/// only when a proved reader is available. Their fixed-cardinality availability
/// gauges distinguish a skipped collection from a real zero.
///
/// Set via `BUZZ_USAGE_METRICS_PER_COMMUNITY`:
/// - `all` — emit per-community series for every community (default)
/// - `off` — suppress all per-community series; fleet totals only
/// - `off` — suppress all per-community series; fleet totals only (default)
/// - `all` — emit per-community series for every community (rollback only)
///
/// A `top:<k>` mode (per-community series for the k most-active communities)
/// is planned as a fast-follow once the series-lifecycle (gauge idle-timeout
@@ -86,15 +88,19 @@ impl EmissionScope {
.unwrap_or_default()
.trim()
.to_ascii_lowercase();
match raw.as_str() {
"" | "all" => EmissionScope::All,
"off" => EmissionScope::Off,
Self::parse(&raw)
}
fn parse(raw: &str) -> Self {
match raw {
"all" => EmissionScope::All,
"" | "off" => EmissionScope::Off,
other => {
warn!(
value = other,
"BUZZ_USAGE_METRICS_PER_COMMUNITY: unknown value — defaulting to all"
"BUZZ_USAGE_METRICS_PER_COMMUNITY: unknown value — defaulting to off"
);
EmissionScope::All
EmissionScope::Off
}
}
}
@@ -102,6 +108,143 @@ impl EmissionScope {
fn allows(&self, _community_id: &Uuid) -> bool {
matches!(self, Self::All)
}
fn includes_per_community(&self) -> bool {
matches!(self, Self::All)
}
}
const FLEET_STOCK_INTERVAL: std::time::Duration = std::time::Duration::from_secs(60 * 60);
const FLEET_ACTIVITY_INTERVAL: std::time::Duration = std::time::Duration::from_secs(24 * 60 * 60);
const FLEET_RETRY_INTERVAL: std::time::Duration = std::time::Duration::from_secs(60);
// Relay-side deadlines for one complete fleet collection: reader acquire,
// replica proof, transaction begin, and the aggregate. The server-side
// `statement_timeout` (5s stock, 15s activity) cannot fire when a reader
// connection stops answering, so these bound the leader's poller tick. Each is
// above its statement timeout so the server-side error still wins normally.
const FLEET_STOCK_COLLECTION_DEADLINE: std::time::Duration = std::time::Duration::from_secs(10);
const FLEET_ACTIVITY_COLLECTION_DEADLINE: std::time::Duration = std::time::Duration::from_secs(20);
// Keep each deadline above its statement timeout in `buzz_db::store::usage`
// (5s stock, 15s activity) so a slow-but-healthy query reports the server-side
// `query_error`, not `timeout`.
const _: () = assert!(FLEET_STOCK_COLLECTION_DEADLINE.as_secs() > 5);
const _: () = assert!(FLEET_ACTIVITY_COLLECTION_DEADLINE.as_secs() > 15);
struct FleetUsageSchedule {
next_stock: std::time::Instant,
next_activity: std::time::Instant,
last_stock_success: Option<std::time::Instant>,
last_activity_success: Option<std::time::Instant>,
stock_snapshot: Option<buzz_db::usage::FleetStockSnapshot>,
activity_snapshot: Option<buzz_db::usage::FleetActiveUsersSnapshot>,
// Whether the most recent due refresh of each family succeeded. Cached
// snapshots outlive a failed refresh so their series stay scrape-visible,
// but availability reports only the latest outcome.
stock_refresh_ok: bool,
activity_refresh_ok: bool,
stock_interval: std::time::Duration,
activity_interval: std::time::Duration,
retry_interval: std::time::Duration,
}
impl FleetUsageSchedule {
fn new() -> Self {
Self::new_at(
std::time::Instant::now(),
FLEET_STOCK_INTERVAL,
FLEET_ACTIVITY_INTERVAL,
FLEET_RETRY_INTERVAL,
)
}
fn new_at(
now: std::time::Instant,
stock_interval: std::time::Duration,
activity_interval: std::time::Duration,
retry_interval: std::time::Duration,
) -> Self {
Self {
next_stock: now,
next_activity: now,
last_stock_success: None,
last_activity_success: None,
stock_snapshot: None,
activity_snapshot: None,
stock_refresh_ok: false,
activity_refresh_ok: false,
stock_interval,
activity_interval,
retry_interval,
}
}
fn stock_due(&self, now: std::time::Instant) -> bool {
now >= self.next_stock
}
fn activity_due(&self, now: std::time::Instant) -> bool {
now >= self.next_activity
}
fn mark_stock_success(
&mut self,
now: std::time::Instant,
snapshot: buzz_db::usage::FleetStockSnapshot,
) {
self.next_stock = now + self.stock_interval;
self.last_stock_success = Some(now);
self.stock_snapshot = Some(snapshot);
self.stock_refresh_ok = true;
}
fn mark_stock_failure(&mut self, now: std::time::Instant) {
self.next_stock = now + self.retry_interval;
self.stock_refresh_ok = false;
}
fn mark_activity_success(
&mut self,
now: std::time::Instant,
snapshot: buzz_db::usage::FleetActiveUsersSnapshot,
) {
self.next_activity = now + self.activity_interval;
self.last_activity_success = Some(now);
self.activity_snapshot = Some(snapshot);
self.activity_refresh_ok = true;
}
fn mark_activity_failure(&mut self, now: std::time::Instant) {
self.next_activity = now + self.retry_interval;
self.activity_refresh_ok = false;
}
/// The `all` rollback collects both families together on every leader
/// tick and emits their values directly, so only health state is kept.
fn mark_legacy_success(&mut self, now: std::time::Instant) {
self.last_stock_success = Some(now);
self.last_activity_success = Some(now);
self.stock_refresh_ok = true;
self.activity_refresh_ok = true;
}
fn stock_available(&self) -> bool {
self.stock_refresh_ok
}
fn activity_available(&self) -> bool {
self.activity_refresh_ok
}
fn reset_after_demotion(&mut self, now: std::time::Instant) {
self.next_stock = now;
self.next_activity = now;
self.last_stock_success = None;
self.last_activity_success = None;
self.stock_snapshot = None;
self.activity_snapshot = None;
self.stock_refresh_ok = false;
self.activity_refresh_ok = false;
}
}
const USAGE_METRICS_LOCK_KEY: i64 = 0x4255_5A5A_4D45_5452;
@@ -301,6 +444,7 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> {
database_url: config.database_url.clone(),
read_database_url: config.read_database_url.clone(),
replica_read_max_age_ms: config.replica_read_max_age_ms,
usage_metrics_replica_max_age_ms: config.usage_metrics_replica_max_age_ms,
max_connections: config.db_pool_size,
read_max_connections: config.db_read_pool_size,
..DbConfig::default()
@@ -1486,14 +1630,16 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> {
// are snapshotted from live in-memory state. Both avoid inc/dec drift.
//
// Multi-pod semantics:
// DB-derived: all pods export the same value → dashboard uses max()
// In-memory: each pod exports its partition → dashboard uses sum()
// DB/storage: leader-only; dashboards must filter by
// buzz_usage_poller_is_leader == 1 before aggregating.
// In-memory: each pod exports its partition → dashboard uses sum().
{
let usage_state = Arc::clone(&state);
let emission_scope = EmissionScope::from_env();
let interval_secs = usage_interval_secs;
let mut leader = None;
let mut emitted_in_memory = HashSet::new();
let mut fleet_schedule = FleetUsageSchedule::new();
tokio::spawn(async move {
// Jitter the first tick by a random fraction of the interval so
// that a rolling deploy with N pods doesn't hammer the DB
@@ -1515,6 +1661,7 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> {
&emission_scope,
&mut leader,
&mut emitted_in_memory,
&mut fleet_schedule,
)
.await
{
@@ -2228,28 +2375,40 @@ async fn run_usage_metrics_tick(
emission_scope: &EmissionScope,
leader: &mut Option<buzz_db::UsageMetricsLeader>,
emitted_in_memory: &mut HashSet<InMemoryMetricKey>,
fleet_schedule: &mut FleetUsageSchedule,
) -> anyhow::Result<()> {
let host_map: HashMap<Uuid, String> = match state.db.usage_community_hosts().await {
Ok(hosts) => hosts
.into_iter()
.map(|community| (community.id, community.host))
.collect(),
Err(error) => {
if leader.is_some() {
warn!("Usage metrics leader demoting: host map collection failed");
*leader = None;
let host_map: HashMap<Uuid, String> = if emission_scope.includes_per_community() {
match state.db.usage_community_hosts().await {
Ok(hosts) => hosts
.into_iter()
.map(|community| (community.id, community.host))
.collect(),
Err(error) => {
if leader.is_some() {
warn!("Usage metrics leader demoting: host map collection failed");
*leader = None;
mark_usage_leader_demoted(fleet_schedule);
}
emit_in_memory_usage_metrics(state, emission_scope, None, emitted_in_memory);
return Err(error.into());
}
emit_in_memory_usage_metrics(state, emission_scope, None, emitted_in_memory);
return Err(error.into());
}
} else {
HashMap::new()
};
emit_in_memory_usage_metrics(state, emission_scope, Some(&host_map), emitted_in_memory);
emit_in_memory_usage_metrics(
state,
emission_scope,
emission_scope.includes_per_community().then_some(&host_map),
emitted_in_memory,
);
let mut demoted = false;
if let Some(leader_guard) = leader.as_mut() {
if !leader_guard.is_live().await {
warn!("Usage metrics leader lock connection failed liveness check; demoting");
*leader = None;
mark_usage_leader_demoted(fleet_schedule);
demoted = true;
}
}
@@ -2263,10 +2422,16 @@ async fn run_usage_metrics_tick(
}
}
if leader.is_some() {
if let Err(error) = emit_db_usage_metrics(state, emission_scope, &host_map).await {
warn!("Usage metrics leader demoting: DB collection failed");
*leader = None;
return Err(error);
if emission_scope.includes_per_community() {
let result = emit_db_usage_metrics(state, emission_scope, &host_map).await;
record_legacy_usage_collection(result.is_ok(), fleet_schedule);
if let Err(error) = result {
warn!("Usage metrics leader demoting: DB collection failed");
*leader = None;
return Err(error);
}
} else {
emit_fleet_db_usage_metrics(state, fleet_schedule).await;
}
let invite_retention_cutoff = chrono::Utc::now() - chrono::Duration::days(30);
match state
@@ -2282,17 +2447,236 @@ async fn run_usage_metrics_tick(
warn!(error = %error, "failed to reap expired relay invites");
}
}
run_storage_sweep_tick(state, emission_scope, &host_map).await;
run_storage_sweep_tick(
state,
emission_scope,
emission_scope.includes_per_community().then_some(&host_map),
)
.await;
}
Ok(())
}
fn mark_usage_leader_demoted(schedule: &mut FleetUsageSchedule) {
schedule.reset_after_demotion(std::time::Instant::now());
emit_usage_snapshot_availability(schedule);
}
/// The `all` rollback collects every family on each leader tick, so a
/// successful collection makes both families available and a failure demotes
/// the leader. This keeps the documented availability gates valid in either
/// emission mode.
fn record_legacy_usage_collection(succeeded: bool, schedule: &mut FleetUsageSchedule) {
if succeeded {
schedule.mark_legacy_success(std::time::Instant::now());
emit_cached_fleet_usage_metrics(schedule);
} else {
mark_usage_leader_demoted(schedule);
}
}
fn emit_usage_snapshot_availability(schedule: &FleetUsageSchedule) {
let gauge = |available: bool| if available { 1.0 } else { 0.0 };
metrics::gauge!("buzz_usage_snapshot_available", "family" => "stock")
.set(gauge(schedule.stock_available()));
metrics::gauge!("buzz_usage_snapshot_available", "family" => "activity")
.set(gauge(schedule.activity_available()));
}
async fn emit_fleet_db_usage_metrics(state: &AppState, schedule: &mut FleetUsageSchedule) {
// Futures are lazy: a family that is not due is never polled.
refresh_due_fleet_usage(
schedule,
state.db.usage_fleet_stock_snapshot(),
state.db.usage_fleet_active_users(chrono::Utc::now()),
)
.await;
}
/// Run each due fleet collection under its relay-side deadline and apply the
/// outcome. An expired collection is dropped and treated as a failed refresh,
/// so a reader that stops answering cannot stall the leader's poller. The
/// dropped reader connection is closed in the background under SQLx's
/// bounded close-on-drop rather than returned to the shared reader pool.
async fn refresh_due_fleet_usage(
schedule: &mut FleetUsageSchedule,
stock: impl std::future::Future<Output = FleetStockRefresh>,
activity: impl std::future::Future<Output = FleetActivityRefresh>,
) {
let now = std::time::Instant::now();
if schedule.stock_due(now) {
let result = tokio::time::timeout(FLEET_STOCK_COLLECTION_DEADLINE, stock).await;
apply_stock_refresh(schedule, now, result);
}
if schedule.activity_due(now) {
let result = tokio::time::timeout(FLEET_ACTIVITY_COLLECTION_DEADLINE, activity).await;
apply_activity_refresh(schedule, now, result);
}
emit_cached_fleet_usage_metrics(schedule);
}
type FleetStockRefresh = buzz_db::Result<Option<buzz_db::usage::FleetStockSnapshot>>;
type FleetActivityRefresh = buzz_db::Result<Option<buzz_db::usage::FleetActiveUsersSnapshot>>;
/// Apply one due stock refresh. Success caches the snapshot and marks the
/// family available; a missing proved reader, a query error, or an expired
/// deadline keeps the cached values, marks the family unavailable, and
/// schedules a retry.
fn apply_stock_refresh(
schedule: &mut FleetUsageSchedule,
now: std::time::Instant,
result: Result<FleetStockRefresh, tokio::time::error::Elapsed>,
) {
match result {
Ok(Ok(Some(snapshot))) => schedule.mark_stock_success(now, snapshot),
Ok(Ok(None)) => {
schedule.mark_stock_failure(now);
record_usage_refresh_skip("stock", "reader_unavailable");
}
Ok(Err(error)) => {
schedule.mark_stock_failure(now);
warn!(error = %error, "fleet usage stock query failed");
record_usage_refresh_skip("stock", "query_error");
}
Err(_) => {
schedule.mark_stock_failure(now);
warn!(
deadline = ?FLEET_STOCK_COLLECTION_DEADLINE,
"fleet usage stock collection timed out"
);
record_usage_refresh_skip("stock", "timeout");
}
}
}
/// Apply one due activity refresh with the same contract as
/// [`apply_stock_refresh`].
fn apply_activity_refresh(
schedule: &mut FleetUsageSchedule,
now: std::time::Instant,
result: Result<FleetActivityRefresh, tokio::time::error::Elapsed>,
) {
match result {
Ok(Ok(Some(snapshot))) => schedule.mark_activity_success(now, snapshot),
Ok(Ok(None)) => {
schedule.mark_activity_failure(now);
record_usage_refresh_skip("activity", "reader_unavailable");
}
Ok(Err(error)) => {
schedule.mark_activity_failure(now);
warn!(error = %error, "fleet usage activity query failed");
record_usage_refresh_skip("activity", "query_error");
}
Err(_) => {
schedule.mark_activity_failure(now);
warn!(
deadline = ?FLEET_ACTIVITY_COLLECTION_DEADLINE,
"fleet usage activity collection timed out"
);
record_usage_refresh_skip("activity", "timeout");
}
}
}
fn record_usage_refresh_skip(family: &'static str, reason: &'static str) {
metrics::counter!(
"buzz_usage_query_skipped_total",
"family" => family,
"reason" => reason
)
.increment(1);
}
/// Refresh cached fixed-cardinality gauges on every poller tick. The
/// Prometheus recorder evicts gauges after several idle ticks, while the
/// underlying stock and activity queries intentionally run hourly/daily.
/// Re-emission preserves the query cadence without letting snapshots disappear
/// between collections; availability reports whether the latest refresh
/// succeeded, and age reports how old the cached values are.
fn emit_cached_fleet_usage_metrics(schedule: &FleetUsageSchedule) {
if let Some(snapshot) = schedule.stock_snapshot {
emit_fleet_stock_metrics(snapshot);
}
if let Some(snapshot) = schedule.activity_snapshot {
emit_fleet_active_user_metrics(snapshot);
}
emit_usage_snapshot_availability(schedule);
if let Some(last_success) = schedule.last_stock_success {
metrics::gauge!("buzz_usage_snapshot_age_seconds", "family" => "stock")
.set(last_success.elapsed().as_secs_f64());
}
if let Some(last_success) = schedule.last_activity_success {
metrics::gauge!("buzz_usage_snapshot_age_seconds", "family" => "activity")
.set(last_success.elapsed().as_secs_f64());
}
}
fn emit_fleet_stock_metrics(snapshot: buzz_db::usage::FleetStockSnapshot) {
metrics::gauge!("buzz_communities_estimated").set(snapshot.communities_estimated as f64);
metrics::gauge!("buzz_total_users", "type" => "human").set(snapshot.users_human as f64);
metrics::gauge!("buzz_total_users", "type" => "agent").set(snapshot.users_agent as f64);
for (kind, value) in [
("stream", snapshot.channels_stream),
("forum", snapshot.channels_forum),
("dm", snapshot.channels_dm),
("workflow", snapshot.channels_workflow),
] {
metrics::gauge!("buzz_total_channels", "type" => kind).set(value as f64);
}
for (role, value) in [
("owner", snapshot.members_owner),
("admin", snapshot.members_admin),
("member", snapshot.members_member),
] {
metrics::gauge!("buzz_total_relay_members", "role" => role).set(value as f64);
}
for (status, value) in [
("active", snapshot.workflows_active),
("disabled", snapshot.workflows_disabled),
("archived", snapshot.workflows_archived),
] {
metrics::gauge!("buzz_total_workflows", "status" => status).set(value as f64);
}
metrics::gauge!("buzz_total_git_repos").set(snapshot.git_repos as f64);
}
fn emit_fleet_active_user_metrics(snapshot: buzz_db::usage::FleetActiveUsersSnapshot) {
for (window, human, agent, unknown) in [
(
"1d",
snapshot.human_1d,
snapshot.agent_1d,
snapshot.unknown_1d,
),
(
"7d",
snapshot.human_7d,
snapshot.agent_7d,
snapshot.unknown_7d,
),
(
"30d",
snapshot.human_30d,
snapshot.agent_30d,
snapshot.unknown_30d,
),
] {
metrics::gauge!("buzz_total_active_users", "window" => window, "type" => "human")
.set(human as f64);
metrics::gauge!("buzz_total_active_users", "window" => window, "type" => "agent")
.set(agent as f64);
metrics::gauge!("buzz_total_active_users", "window" => window, "type" => "unknown")
.set(unknown as f64);
}
}
/// Read worker storage snapshots only on the existing leader metrics tick.
async fn run_storage_sweep_tick(
state: &AppState,
emission_scope: &EmissionScope,
host_map: &HashMap<Uuid, String>,
host_map: Option<&HashMap<Uuid, String>>,
) {
static MODE: std::sync::OnceLock<storage_sweep::StorageMetricsMode> =
std::sync::OnceLock::new();
@@ -2648,11 +3032,14 @@ mod tests {
use uuid::Uuid;
use super::{
buzz_auto_migrate_enabled, connect_audit_pool, dropped_in_memory_keys, idle_timeout_secs,
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,
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,
InMemoryMetricKey, PartitionAuditSchedule,
FleetUsageSchedule, InMemoryMetricKey, PartitionAuditSchedule,
FLEET_ACTIVITY_COLLECTION_DEADLINE, FLEET_STOCK_COLLECTION_DEADLINE,
};
use buzz_db::DbConfig;
use metrics::GaugeFn;
@@ -2853,6 +3240,443 @@ mod tests {
assert!(!EmissionScope::Off.allows(&Uuid::new_v4()));
}
#[test]
fn usage_metrics_default_to_fleet_only() {
assert!(matches!(EmissionScope::parse(""), EmissionScope::Off));
assert!(matches!(
EmissionScope::parse("not-a-mode"),
EmissionScope::Off
));
assert!(matches!(EmissionScope::parse("all"), EmissionScope::All));
}
#[test]
fn fleet_usage_families_have_independent_cadences() {
let now = std::time::Instant::now();
let mut schedule = FleetUsageSchedule::new_at(
now,
Duration::from_secs(60),
Duration::from_secs(600),
Duration::from_secs(10),
);
assert!(schedule.stock_due(now));
assert!(schedule.activity_due(now));
assert!(!schedule.stock_available());
assert!(!schedule.activity_available());
schedule.mark_stock_failure(now);
assert!(!schedule.stock_due(now + Duration::from_secs(9)));
assert!(schedule.stock_due(now + Duration::from_secs(10)));
assert!(schedule.activity_due(now));
let stock_success = now + Duration::from_secs(10);
let stock_snapshot = buzz_db::usage::FleetStockSnapshot {
communities_estimated: 7,
..Default::default()
};
schedule.mark_stock_success(stock_success, stock_snapshot);
assert!(schedule.stock_available());
assert_eq!(
schedule
.stock_snapshot
.expect("cached stock snapshot")
.communities_estimated,
7
);
assert!(!schedule.stock_due(stock_success + Duration::from_secs(59)));
assert!(schedule.stock_due(stock_success + Duration::from_secs(60)));
schedule.mark_activity_failure(now);
assert!(!schedule.activity_due(now + Duration::from_secs(9)));
assert!(schedule.activity_due(now + Duration::from_secs(10)));
let activity_success = now + Duration::from_secs(10);
let activity_snapshot = buzz_db::usage::FleetActiveUsersSnapshot {
human_1d: 3,
..Default::default()
};
schedule.mark_activity_success(activity_success, activity_snapshot);
assert!(schedule.activity_available());
assert_eq!(
schedule
.activity_snapshot
.expect("cached activity snapshot")
.human_1d,
3
);
assert!(!schedule.activity_due(activity_success + Duration::from_secs(599)));
assert!(schedule.activity_due(activity_success + Duration::from_secs(600)));
let demoted_at = activity_success + Duration::from_secs(20);
schedule.reset_after_demotion(demoted_at);
assert!(!schedule.stock_available());
assert!(!schedule.activity_available());
assert!(schedule.stock_due(demoted_at));
assert!(schedule.activity_due(demoted_at));
}
fn usage_metric_values(
recorder: &metrics_util::debugging::DebuggingRecorder,
) -> std::collections::HashMap<String, f64> {
recorder
.snapshotter()
.snapshot()
.into_vec()
.into_iter()
.filter_map(|(key, _, _, value)| {
let name = key.key().name();
let label = key
.key()
.labels()
.map(|label| format!("{}={}", label.key(), label.value()))
.collect::<Vec<_>>()
.join(",");
match value {
DebugValue::Gauge(value) => {
Some((format!("{name}{{{label}}}"), value.into_inner()))
}
DebugValue::Counter(value) => {
Some((format!("{name}{{{label}}}"), value as f64))
}
_ => None,
}
})
.collect()
}
/// Drives the production refresh-to-availability mapping
/// (`apply_*_refresh`) through success -> skipped/failed refresh ->
/// recovery and asserts the emitted gauges, the retained cached values,
/// the skip reason, and the retry schedule.
#[test]
fn fleet_refresh_outcomes_drive_availability_through_production_mapping() {
let now = std::time::Instant::now();
let retry = Duration::from_secs(10);
let stock = buzz_db::usage::FleetStockSnapshot {
communities_estimated: 7,
..Default::default()
};
let activity = buzz_db::usage::FleetActiveUsersSnapshot {
human_1d: 3,
..Default::default()
};
let stock_available = "buzz_usage_snapshot_available{family=stock}";
let activity_available = "buzz_usage_snapshot_available{family=activity}";
let cached_stock = "buzz_communities_estimated{}";
let cached_activity = "buzz_total_active_users{window=1d,type=human}";
for (query_error, reason) in [(false, "reader_unavailable"), (true, "query_error")] {
let mut schedule = FleetUsageSchedule::new_at(
now,
Duration::from_secs(60),
Duration::from_secs(600),
retry,
);
let tick = |schedule: &mut FleetUsageSchedule,
at: std::time::Instant,
stock_result,
activity_result| {
let recorder = metrics_util::debugging::DebuggingRecorder::new();
metrics::with_local_recorder(&recorder, || {
apply_stock_refresh(schedule, at, Ok(stock_result));
apply_activity_refresh(schedule, at, Ok(activity_result));
emit_cached_fleet_usage_metrics(schedule);
});
usage_metric_values(&recorder)
};
let values = tick(&mut schedule, now, Ok(Some(stock)), Ok(Some(activity)));
assert_eq!(values.get(stock_available), Some(&1.0), "{reason}");
assert_eq!(values.get(activity_available), Some(&1.0), "{reason}");
let failed_at = now + Duration::from_secs(600);
let values = if query_error {
tick(
&mut schedule,
failed_at,
Err(buzz_db::DbError::AuthEventRejected),
Err(buzz_db::DbError::AuthEventRejected),
)
} else {
tick(&mut schedule, failed_at, Ok(None), Ok(None))
};
assert_eq!(values.get(stock_available), Some(&0.0), "{reason}");
assert_eq!(values.get(activity_available), Some(&0.0), "{reason}");
assert_eq!(values.get(cached_stock), Some(&7.0), "{reason}");
assert_eq!(values.get(cached_activity), Some(&3.0), "{reason}");
for family in ["stock", "activity"] {
let key =
format!("buzz_usage_query_skipped_total{{family={family},reason={reason}}}");
assert_eq!(values.get(&key), Some(&1.0), "{key}");
}
assert!(!schedule.stock_due(failed_at + retry - Duration::from_millis(1)));
assert!(schedule.stock_due(failed_at + retry));
assert!(schedule.activity_due(failed_at + retry));
let recovered_at = failed_at + retry;
let values = tick(
&mut schedule,
recovered_at,
Ok(Some(stock)),
Ok(Some(activity)),
);
assert_eq!(values.get(stock_available), Some(&1.0), "{reason}");
assert_eq!(values.get(activity_available), Some(&1.0), "{reason}");
assert!(
!values
.keys()
.any(|key| key.starts_with("buzz_usage_query_skipped_total")),
"{reason}: recovery must not record a skip"
);
}
}
#[tokio::test(start_paused = true)]
async fn each_fleet_family_is_abandoned_at_its_own_deadline() {
let stock = || std::future::ready(Ok(Some(buzz_db::usage::FleetStockSnapshot::default())));
let activity = || {
std::future::ready(Ok(
Some(buzz_db::usage::FleetActiveUsersSnapshot::default()),
))
};
let schedule = || {
FleetUsageSchedule::new_at(
std::time::Instant::now(),
Duration::ZERO,
Duration::ZERO,
Duration::from_secs(60),
)
};
// The outer bound turns a missing deadline into a failure instead of
// a hang under paused time.
async fn bounded(refresh: impl std::future::Future<Output = ()>) {
tokio::time::timeout(Duration::from_secs(60 * 60), refresh)
.await
.expect("a stalled family must be abandoned at its deadline");
}
let started = tokio::time::Instant::now();
bounded(refresh_due_fleet_usage(
&mut schedule(),
std::future::pending(),
activity(),
))
.await;
assert_eq!(
tokio::time::Instant::now() - started,
FLEET_STOCK_COLLECTION_DEADLINE
);
let started = tokio::time::Instant::now();
bounded(refresh_due_fleet_usage(
&mut schedule(),
stock(),
std::future::pending(),
))
.await;
assert_eq!(
tokio::time::Instant::now() - started,
FLEET_ACTIVITY_COLLECTION_DEADLINE
);
}
#[tokio::test(start_paused = true)]
async fn stalled_fleet_collection_times_out_into_failed_refresh() {
let retry = Duration::from_secs(60);
let started = std::time::Instant::now();
// Zero cadences keep both families due after the seeded success.
let mut schedule =
FleetUsageSchedule::new_at(started, Duration::ZERO, Duration::ZERO, retry);
schedule.mark_stock_success(
started,
buzz_db::usage::FleetStockSnapshot {
communities_estimated: 7,
..Default::default()
},
);
schedule.mark_activity_success(
started,
buzz_db::usage::FleetActiveUsersSnapshot {
human_1d: 3,
..Default::default()
},
);
let recorder = metrics_util::debugging::DebuggingRecorder::new();
let _guard = metrics::set_default_local_recorder(&recorder);
let tick_started = tokio::time::Instant::now();
let before = std::time::Instant::now();
// A reader that never answers: neither collection ever completes.
tokio::time::timeout(
Duration::from_secs(60 * 60),
refresh_due_fleet_usage(
&mut schedule,
std::future::pending(),
std::future::pending(),
),
)
.await
.expect("a stalled reader must not stall the poller tick");
let after = std::time::Instant::now();
assert_eq!(
tokio::time::Instant::now() - tick_started,
FLEET_STOCK_COLLECTION_DEADLINE + FLEET_ACTIVITY_COLLECTION_DEADLINE,
"each family is abandoned at its own deadline"
);
let values = usage_metric_values(&recorder);
assert_eq!(
values.get("buzz_usage_snapshot_available{family=stock}"),
Some(&0.0)
);
assert_eq!(
values.get("buzz_usage_snapshot_available{family=activity}"),
Some(&0.0)
);
assert_eq!(values.get("buzz_communities_estimated{}"), Some(&7.0));
assert_eq!(
values.get("buzz_total_active_users{window=1d,type=human}"),
Some(&3.0)
);
for family in ["stock", "activity"] {
let key = format!("buzz_usage_query_skipped_total{{family={family},reason=timeout}}");
assert_eq!(values.get(&key), Some(&1.0), "{key}");
}
// An abandoned collection records no route decision; the documented
// signal is the timeout skip above.
assert!(
!values
.keys()
.any(|key| key.starts_with("buzz_db_route_decision")),
"a timed-out collection must not record a route decision"
);
assert!(!schedule.stock_due(before + retry - Duration::from_millis(1)));
assert!(!schedule.activity_due(before + retry - Duration::from_millis(1)));
assert!(schedule.stock_due(after + retry));
assert!(schedule.activity_due(after + retry));
// The next tick still runs and a successful collection recovers.
schedule.next_stock = after;
schedule.next_activity = after;
refresh_due_fleet_usage(
&mut schedule,
std::future::ready(Ok(Some(buzz_db::usage::FleetStockSnapshot {
communities_estimated: 8,
..Default::default()
}))),
std::future::ready(Ok(Some(buzz_db::usage::FleetActiveUsersSnapshot {
human_1d: 4,
..Default::default()
}))),
)
.await;
let values = usage_metric_values(&recorder);
assert_eq!(
values.get("buzz_usage_snapshot_available{family=stock}"),
Some(&1.0)
);
assert_eq!(
values.get("buzz_usage_snapshot_available{family=activity}"),
Some(&1.0)
);
assert_eq!(values.get("buzz_communities_estimated{}"), Some(&8.0));
assert_eq!(
values.get("buzz_total_active_users{window=1d,type=human}"),
Some(&4.0)
);
}
#[test]
fn failed_fleet_refresh_marks_cached_snapshot_unavailable_until_recovery() {
let now = std::time::Instant::now();
let mut schedule = FleetUsageSchedule::new_at(
now,
Duration::from_secs(60),
Duration::from_secs(600),
Duration::from_secs(10),
);
let stock = buzz_db::usage::FleetStockSnapshot {
communities_estimated: 7,
..Default::default()
};
let activity = buzz_db::usage::FleetActiveUsersSnapshot {
human_1d: 3,
..Default::default()
};
let stock_available = "buzz_usage_snapshot_available{family=stock}";
let activity_available = "buzz_usage_snapshot_available{family=activity}";
let emit = |schedule: &FleetUsageSchedule| {
let recorder = metrics_util::debugging::DebuggingRecorder::new();
metrics::with_local_recorder(&recorder, || emit_cached_fleet_usage_metrics(schedule));
usage_metric_values(&recorder)
};
schedule.mark_stock_success(now, stock);
schedule.mark_activity_success(now, activity);
let values = emit(&schedule);
assert_eq!(values.get(stock_available), Some(&1.0));
assert_eq!(values.get(activity_available), Some(&1.0));
// A failed due refresh keeps the last values visible, but no longer
// advertises them as available.
let failed_at = now + Duration::from_secs(600);
schedule.mark_stock_failure(failed_at);
schedule.mark_activity_failure(failed_at);
let values = emit(&schedule);
assert_eq!(values.get(stock_available), Some(&0.0));
assert_eq!(values.get(activity_available), Some(&0.0));
assert_eq!(values.get("buzz_communities_estimated{}"), Some(&7.0));
assert_eq!(
values.get("buzz_total_active_users{window=1d,type=human}"),
Some(&3.0)
);
assert!(schedule.stock_due(failed_at + Duration::from_secs(10)));
assert!(schedule.activity_due(failed_at + Duration::from_secs(10)));
let recovered_at = failed_at + Duration::from_secs(10);
schedule.mark_stock_success(recovered_at, stock);
schedule.mark_activity_success(recovered_at, activity);
let values = emit(&schedule);
assert_eq!(values.get(stock_available), Some(&1.0));
assert_eq!(values.get(activity_available), Some(&1.0));
}
#[test]
fn legacy_collection_publishes_coherent_availability() {
let mut schedule = FleetUsageSchedule::new_at(
std::time::Instant::now(),
Duration::from_secs(60),
Duration::from_secs(600),
Duration::from_secs(10),
);
let stock_available = "buzz_usage_snapshot_available{family=stock}";
let activity_available = "buzz_usage_snapshot_available{family=activity}";
let mut record = |succeeded: bool| {
let recorder = metrics_util::debugging::DebuggingRecorder::new();
metrics::with_local_recorder(&recorder, || {
record_legacy_usage_collection(succeeded, &mut schedule)
});
usage_metric_values(&recorder)
};
for (succeeded, expected) in [(true, 1.0), (false, 0.0), (true, 1.0)] {
let values = record(succeeded);
assert_eq!(values.get(stock_available), Some(&expected), "{succeeded}");
assert_eq!(
values.get(activity_available),
Some(&expected),
"{succeeded}"
);
// Age is the companion freshness signal; it must exist after a
// successful rollback collection and not after a demotion.
for family in ["stock", "activity"] {
let age = format!("buzz_usage_snapshot_age_seconds{{family={family}}}");
assert_eq!(values.contains_key(&age), succeeded, "{age}");
}
}
}
#[test]
fn test_dropped_in_memory_keys_preserves_resolved_host_label() {
let previous = HashSet::from([

113
crates/buzz-relay/src/storage_sweep.rs
View File

@@ -113,7 +113,7 @@ pub async fn run_storage_metrics_tick(
db: &Db,
state: &Mutex<StorageSweepState>,
mode: StorageMetricsMode,
host_map: &HashMap<Uuid, String>,
host_map: Option<&HashMap<Uuid, String>>,
allows: impl Fn(&Uuid) -> bool,
) -> anyhow::Result<()> {
if mode == StorageMetricsMode::Disabled {
@@ -179,7 +179,7 @@ async fn record_persisted_snapshot_load_failure(state: &Mutex<StorageSweepState>
async fn emit_storage_metrics(
state: &Mutex<StorageSweepState>,
mode: StorageMetricsMode,
host_map: &HashMap<Uuid, String>,
host_map: Option<&HashMap<Uuid, String>>,
allows: impl Fn(&Uuid) -> bool,
) {
if mode == StorageMetricsMode::Disabled {
@@ -195,7 +195,7 @@ async fn emit_storage_metrics(
fn emit_cached_storage_metrics(
state: &mut StorageSweepState,
host_map: &HashMap<Uuid, String>,
host_map: Option<&HashMap<Uuid, String>>,
allows: impl Fn(&Uuid) -> bool,
) {
let Some(cached) = &state.cached else {
@@ -229,24 +229,29 @@ fn emit_cached_storage_metrics(
metrics::gauge!("buzz_storage_snapshot_cap_utilization")
.set(snapshot.physical_objects as f64 / cached.max_objects as f64);
// Without a host map (fleet-only emission) the relay attributes nothing,
// so it emits neither community series nor the unmapped total: every byte
// would otherwise miss the empty map and read as a false unmapped anomaly.
let mut current = HashSet::new();
let mut unmapped_bytes = 0u64;
for (community_id, storage) in &snapshot.per_community {
let Some(host) = host_map.get(community_id) else {
unmapped_bytes += storage.bytes;
continue;
};
if !allows(community_id) {
continue;
if let Some(host_map) = host_map {
let mut unmapped_bytes = 0u64;
for (community_id, storage) in &snapshot.per_community {
let Some(host) = host_map.get(community_id) else {
unmapped_bytes += storage.bytes;
continue;
};
if !allows(community_id) {
continue;
}
metrics::gauge!("buzz_community_storage_bytes", "community" => host.clone())
.set(storage.bytes as f64);
metrics::gauge!("buzz_community_storage_objects", "community" => host.clone())
.set(storage.objects as f64);
current.insert(StorageEmittedKey::Bytes(host.clone()));
current.insert(StorageEmittedKey::Objects(host.clone()));
}
metrics::gauge!("buzz_community_storage_bytes", "community" => host.clone())
.set(storage.bytes as f64);
metrics::gauge!("buzz_community_storage_objects", "community" => host.clone())
.set(storage.objects as f64);
current.insert(StorageEmittedKey::Bytes(host.clone()));
current.insert(StorageEmittedKey::Objects(host.clone()));
metrics::gauge!("buzz_storage_unmapped_community_bytes").set(unmapped_bytes as f64);
}
metrics::gauge!("buzz_storage_unmapped_community_bytes").set(unmapped_bytes as f64);
// Zero series for communities that were emitted last tick but are no longer
// present in the current snapshot (community removed, host renamed, or
@@ -332,7 +337,7 @@ mod tests {
futures::executor::block_on(emit_storage_metrics(
&state,
StorageMetricsMode::Snapshot,
&HashMap::new(),
Some(&HashMap::new()),
|_| true,
));
});
@@ -353,7 +358,7 @@ mod tests {
&db,
&state,
StorageMetricsMode::Disabled,
&HashMap::new(),
Some(&HashMap::new()),
|_| true,
))
});
@@ -375,7 +380,7 @@ mod tests {
&db,
&state,
StorageMetricsMode::Snapshot,
&HashMap::new(),
Some(&HashMap::new()),
|_| true,
))
});
@@ -442,7 +447,7 @@ mod tests {
futures::executor::block_on(emit_storage_metrics(
&state,
StorageMetricsMode::Snapshot,
&host_map,
Some(&host_map),
|id| *id != excluded,
));
});
@@ -457,6 +462,60 @@ mod tests {
assert_eq!(values.get("buzz_community_storage_bytes"), Some(&100.0));
}
#[tokio::test]
async fn fleet_only_emission_omits_attribution_without_false_unmapped_bytes() {
let community = Uuid::new_v4();
let state = Mutex::new(StorageSweepState {
cached: Some(CachedSnapshot {
data: snapshot_with(community, 64, 2),
completed_at_wall: Utc::now(),
duration: Duration::from_millis(100),
max_objects: 1_000,
}),
persisted_load_ok: Some(true),
..Default::default()
});
let recorder = DebuggingRecorder::new();
metrics::with_local_recorder(&recorder, || {
futures::executor::block_on(emit_storage_metrics(
&state,
StorageMetricsMode::Snapshot,
None,
|_| true,
));
});
let values = gauge_snapshot(&recorder);
assert_eq!(values.get("buzz_total_storage_bytes"), Some(&64.0));
assert_eq!(values.get("buzz_storage_unmapped_community_bytes"), None);
assert!(labeled_community_gauges(&recorder).is_empty());
// A previously attributed series is zeroed once attribution stops.
let host_map = HashMap::from([(community, "host.example".to_string())]);
let recorder = DebuggingRecorder::new();
metrics::with_local_recorder(&recorder, || {
futures::executor::block_on(emit_storage_metrics(
&state,
StorageMetricsMode::Snapshot,
Some(&host_map),
|_| true,
));
futures::executor::block_on(emit_storage_metrics(
&state,
StorageMetricsMode::Snapshot,
None,
|_| true,
));
});
assert_eq!(
labeled_community_gauges(&recorder).get(&(
"buzz_community_storage_bytes".to_string(),
"host.example".to_string()
)),
Some(&0.0)
);
}
// --- F-EXT1 regression: stale per-community series are zeroed ---
/// Returns a map of `(metric_name, community_label_value) -> gauge_value`
@@ -555,7 +614,7 @@ mod tests {
futures::executor::block_on(emit_storage_metrics(
&state,
StorageMetricsMode::Snapshot,
&host_map_1,
Some(&host_map_1),
|_| true,
));
});
@@ -597,7 +656,7 @@ mod tests {
futures::executor::block_on(emit_storage_metrics(
&state,
StorageMetricsMode::Snapshot,
&host_map_2,
Some(&host_map_2),
|id| *id != community_c, // (c) scope-excluded
));
});
@@ -677,7 +736,7 @@ mod tests {
futures::executor::block_on(emit_storage_metrics(
&state,
StorageMetricsMode::Snapshot,
&host_map,
Some(&host_map),
|_| true,
));
});
@@ -721,7 +780,7 @@ mod tests {
futures::executor::block_on(emit_storage_metrics(
&state,
StorageMetricsMode::Snapshot,
&host_map,
Some(&host_map),
|_| true,
));
});
@@ -754,7 +813,7 @@ mod tests {
futures::executor::block_on(emit_storage_metrics(
&state,
StorageMetricsMode::Snapshot,
&host_map,
Some(&host_map),
|_| true,
));
});

8
crates/buzz-relay/src/storage_sweep/postgres_tests.rs
View File

@@ -12,7 +12,13 @@ fn observe_tick(
) -> (anyhow::Result<()>, HashMap<String, f64>) {
let recorder = DebuggingRecorder::new();
let result = metrics::with_local_recorder(&recorder, || {
futures::executor::block_on(run_storage_metrics_tick(db, state, mode, hosts, |_| true))
futures::executor::block_on(run_storage_metrics_tick(
db,
state,
mode,
Some(hosts),
|_| true,
))
});
let values = recorder
.snapshotter()

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

@@ -305,7 +305,7 @@ maximum gauges when diagnosing total capacity pressure.
Outcomes are `success`, `timeout`, `error`, and `cancelled`. Operations are
`bootstrap`, `readiness`, `tenant_resolution`, `authentication`,
`authorization`, `subscription_history`, `event_write`, and `maintenance`.
Only the following eleven pairs are valid:
Only the following twelve pairs are valid:
```text
writer/bootstrap reader/bootstrap
@@ -315,12 +315,12 @@ writer/authentication
writer/authorization reader/authorization
writer/subscription_history reader/subscription_history
writer/event_write
writer/maintenance
writer/maintenance reader/maintenance
```
Nine finite checkout buckets plus `+Inf`, sum, and count yield 12 histogram
series per valid pair. The new contract therefore has a hard ceiling of 198
raw Prometheus series per pod: `11 × (1 + 12 + 4 + 1)`. The two legacy acquisition
series per valid pair. The new contract therefore has a hard ceiling of 216
raw Prometheus series per pod: `12 × (1 + 12 + 4 + 1)`. The two legacy acquisition
families remain temporarily for dashboard compatibility and are not part of
that new-family budget. No `other` operation or request-controlled/sensitive
label is valid.

101
docs/usage-metrics.md Normal file
View File

@@ -0,0 +1,101 @@
# Usage metrics operations
The relay emits fleet-wide usage and storage gauges without retaining one
in-memory series per community. This is the default, bounded-cardinality mode:
```text
BUZZ_USAGE_METRICS_PER_COMMUNITY=off
BUZZ_USAGE_METRICS_REPLICA_MAX_AGE_MS=30000
```
Database-backed fleet gauges are replica-only. The replica freshness budget is
independent of `BUZZ_REPLICA_READ_MAX_AGE_MS`, which controls serving reads. A
missing, stale, or unavailable reader skips the telemetry query; the relay never
falls back to the writer for these aggregates. Set the telemetry budget to `0`
to disable the database-backed families explicitly.
A deployment with no `READ_DATABASE_URL`, or whose reader fails verification,
therefore emits none of the database-backed fleet totals (`buzz_total_users`,
`buzz_total_channels`, `buzz_total_active_users`, and the rest of the stock
family) and reports both families unavailable. That includes local compose and
single-node self-hosts. Configure a verified reader where those totals are
needed.
## Availability
Dashboards and monitors must pair values with these fixed-cardinality gauges:
- `buzz_usage_snapshot_available{family="stock"}`
- `buzz_usage_snapshot_available{family="activity"}`
Database and storage gauges are leader-only. In a multi-pod deployment, first
filter them to the pod where `buzz_usage_poller_is_leader == 1`; do not take an
unfiltered maximum across pods. A demoted pod clears its snapshot availability,
but previously exported value series remain scrape-visible until the recorder
evicts them.
Availability reports whether the latest due collection of that family
succeeded. A value of `0` means it was skipped or failed; it must not be
interpreted as all usage being zero. The last successful values stay
scrape-visible while availability is `0`, and
`buzz_usage_snapshot_age_seconds{family=...}` reports how old they are. Failed
stock or activity collections retry after 60 seconds; a successful retry
restores availability and resumes the normal hourly and daily cadences.
Each skipped or failed collection increments
`buzz_usage_query_skipped_total{family,reason}`: `reader_unavailable` when no
fresh proved reader was available, `query_error` when the reader query failed,
and `timeout` when the whole collection (reader proof included) exceeded its
relay-side deadline of 10 seconds for stock or 20 seconds for activity. The
deadline bounds a reader that stops answering, which the server-side statement
timeout cannot, so a hung reader cannot stall the leader's poller. Telemetry
reader connections are closed when the collection ends rather than returned to
the reader pool, and an abandoned one is closed within 5 seconds, so an
abandoned telemetry read cannot hold a slot that serving reads need.
Storage totals come from the `buzz-admin` worker snapshot and carry their own
`buzz_storage_snapshot_load_ok` and `buzz_storage_snapshot_age_seconds` health
gauges. In fleet-only mode the relay does not attribute storage to communities,
so it emits neither `buzz_community_storage_*` series nor
`buzz_storage_unmapped_community_bytes`.
Every fleet telemetry query attempt that completes is recorded in
`buzz_db_route_decision{path=~"usage_fleet_.*"}`: `replica/fresh` when it ran on
a proved reader, `skipped/<reason>` when no fresh proved reader was available,
and `skipped/replica_error` when the reader query itself failed. A collection
the relay abandons at its deadline records no route decision; it appears only
as `buzz_usage_query_skipped_total{reason="timeout"}`. None of these paths fall
back to the writer.
## Rollout and rollback
1. Deploy with per-community mode unset or `off` and the telemetry replica
freshness budget at its 30000ms default.
2. Update dashboards to use fleet gauges and gate alerts on the availability
gauges above.
3. Remove dependencies on the retired per-community series before increasing
community count.
`buzz_total_messages` and `buzz_total_active_channels` are emitted only by the
`all` collection. In the default fleet-only mode they stop updating, so move
dashboards off them before rollout.
`BUZZ_USAGE_METRICS_PER_COMMUNITY=all` temporarily restores the prior
per-community emission for rollback or dashboard migration. It also restores
the associated memory and monitoring-cardinality growth, so it is not the
steady-state configuration. In this mode every leader tick collects all
families, so a successful collection reports both availability gauges as `1`
and a failed one demotes the leader and reports `0`; dashboards gated on
availability or `buzz_usage_snapshot_age_seconds` keep working through a
rollback. It emits the exact
`buzz_communities_total` instead of `buzz_communities_estimated`.
## Leader election
Every relay process attempts the same PostgreSQL session advisory lock. Exactly
one live database session owns it and performs collection. If that process or
session exits, PostgreSQL releases the lock and another relay acquires it. The
lock only deduplicates collectors; it does not permit writer fallback for fleet
queries. A process that loses the lock clears its cached fleet snapshots, marks
their availability as zero, and forces fresh collection if it later becomes the
leader again.