Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
31 changes: 18 additions & 13 deletions ARCHITECTURE.md
Original file line number Diff line number Diff line change
Expand Up @@ -503,24 +503,29 @@ shape is an explicit transition, not a flag day:
transitions take the matching exclusive lock. A supported serving transaction
never spans communities; unmigrated paths still rely on trigger/function
backstops.
- **Replica floor (transition foundation):** runtime now provides a shared
replica-floor lock helper for compliant channel-event writers, and the writer
probe handshake takes the exclusive counterpart before sampling `S`, scanning
activity, and committing the heartbeat token last. `probe_once` remains the
sole token/fence-wall publication path; there is no separately persisted active
floor cutoff. Commit-time trigger+GUC enforcement remains authoritative for all
paths.
- **Dual enforcement (current):** application-owned lock+precheck paths run in
front of the existing trigger/function enforcement; commit-time trigger checks
remain authoritative during this phase. Removing those backstops is a separate
gated migration, not part of the lock foundation.
- **Replica floor:** supported channel-event insertions check `created_at`
against `clock_timestamp()` on their admitted transaction using the shared
960-second floor. The ordered heartbeat handshake and activity scan retain
the same fence-wall calculation. Migration 0062 retires only the deferred
floor trigger/function; writer connections no longer arm its GUC and probe
startup no longer requires those objects.
- **Floor retirement deploy gate:** phase 1 must be deployed to every writer
pod/job, with no rollback expected, before applying the destructive contract.
Drain old connections/in-flight transactions, stop out-of-band writers, and
hold the replica breaker closed through the drop until replay catches up.
Already-running old pods do not reverify the trigger on each heartbeat.
- **Dual enforcement (community fence transition):** application-owned
lock+precheck paths run in front of the existing community trigger/function
enforcement. Those backstops remain in place until their own separately gated
retirement.
- **Role separation:** relay/runtime code owns admission and lock protocols;
operator maintenance/backfill workflows must either use those protocols or run
under explicit reviewed procedures that keep routing fences closed.
- **Unsupported shape:** direct owner SQL that bypasses supported transaction and
lock protocols is not a supported write path.
- **Fleet gates and reconciliation:** startup/fence probes verify guard catalog +
behavior, pgschema bootstraps run reconciliation, and lock/pool metrics remain
- **Fleet gates and reconciliation:** retained guards keep their catalog and
behavior checks; heartbeat/epoch and activity checks still fail closed.
pgschema bootstraps run reconciliation, and lock/pool metrics remain
the operational evidence path for rollout safety.

**Does NOT:** cache queries, implement connection pooling logic (delegated to sqlx), or make network calls outside Postgres.
Expand Down
7 changes: 7 additions & 0 deletions crates/buzz-db/src/error.rs
Original file line number Diff line number Diff line change
Expand Up @@ -81,6 +81,13 @@ pub enum DbError {
#[error("invalid timestamp: {0}")]
InvalidTimestamp(i64),

/// A channel event is too old for replica-fence completeness. Retrying the
/// same signed event cannot make its timestamp newer. Like the mixed-version
/// trigger's SQLx error, this remains a database error at the relay boundary
/// (WS OK false / HTTP 500), rather than a different ingest rejection class.
#[error("database error: events.created_at is below the replica-fence floor")]
CreatedAtFloorRejected,

/// A roster mutation would remove the last effective relay Operator,
/// leaving no one able to administer the deployment through the API.
/// The transaction is rolled back and the caller must add a replacement
Expand Down
6 changes: 3 additions & 3 deletions crates/buzz-db/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -45,15 +45,15 @@ pub const DB_POOL_ACQUIRE_RAW_SERIES_PER_POD: usize =
runtime::observability::POOL_ACQUIRE_RAW_SERIES_PER_POD;

/// Valid database connection role/step pairs with a start counter.
pub const DB_CONNECTION_STARTED_STEPS: [(DbPoolRole, DbConnectionStep); 4] =
pub const DB_CONNECTION_STARTED_STEPS: [(DbPoolRole, DbConnectionStep); 3] =
runtime::CONNECTION_STARTED_STEPS;

/// Valid database connection role/step pairs with a duration histogram.
pub const DB_CONNECTION_DURATION_STEPS: [(DbPoolRole, DbConnectionStep); 4] =
pub const DB_CONNECTION_DURATION_STEPS: [(DbPoolRole, DbConnectionStep); 3] =
runtime::CONNECTION_DURATION_STEPS;

/// Valid database connection role/step/outcome terminal combinations.
pub const DB_CONNECTION_TERMINALS: [(DbPoolRole, DbConnectionStep, DbConnectionOutcome); 15] =
pub const DB_CONNECTION_TERMINALS: [(DbPoolRole, DbConnectionStep, DbConnectionOutcome); 12] =
runtime::CONNECTION_TERMINALS;

/// Raw Prometheus series ceiling per pod for connection-step telemetry.
Expand Down
39 changes: 9 additions & 30 deletions crates/buzz-db/src/runtime/connection_observability.rs
Original file line number Diff line number Diff line change
Expand Up @@ -17,8 +17,6 @@ pub enum DbConnectionStep {
WriterPool,
/// A physical connection has completed DNS/network/TLS/authentication.
PhysicalConnect,
/// Install the created-at replica-fence floor.
CreatedAtFloor,
/// Install lock, idle-transaction, and statement timeouts.
SessionTimeouts,
/// Verify READ COMMITTED transaction isolation.
Expand All @@ -29,10 +27,9 @@ pub enum DbConnectionStep {

impl DbConnectionStep {
/// Complete metric-label vocabulary.
pub const ALL: [Self; 6] = [
pub const ALL: [Self; 5] = [
Self::WriterPool,
Self::PhysicalConnect,
Self::CreatedAtFloor,
Self::SessionTimeouts,
Self::Isolation,
Self::Ready,
Expand All @@ -43,7 +40,6 @@ impl DbConnectionStep {
match self {
Self::WriterPool => "writer_pool",
Self::PhysicalConnect => "physical_connect",
Self::CreatedAtFloor => "created_at_floor",
Self::SessionTimeouts => "session_timeouts",
Self::Isolation => "isolation",
Self::Ready => "ready",
Expand Down Expand Up @@ -77,18 +73,17 @@ impl DbConnectionOutcome {
}

/// Valid role/step pairs with an explicit start counter.
pub const CONNECTION_STARTED_STEPS: [(DbPoolRole, DbConnectionStep); 4] = [
pub const CONNECTION_STARTED_STEPS: [(DbPoolRole, DbConnectionStep); 3] = [
(DbPoolRole::Writer, DbConnectionStep::WriterPool),
(DbPoolRole::Writer, DbConnectionStep::CreatedAtFloor),
(DbPoolRole::Writer, DbConnectionStep::SessionTimeouts),
(DbPoolRole::Writer, DbConnectionStep::Isolation),
];

/// Valid role/step pairs with a duration histogram.
pub const CONNECTION_DURATION_STEPS: [(DbPoolRole, DbConnectionStep); 4] = CONNECTION_STARTED_STEPS;
pub const CONNECTION_DURATION_STEPS: [(DbPoolRole, DbConnectionStep); 3] = CONNECTION_STARTED_STEPS;

/// Valid role/step/outcome terminal combinations.
pub const CONNECTION_TERMINALS: [(DbPoolRole, DbConnectionStep, DbConnectionOutcome); 15] = [
pub const CONNECTION_TERMINALS: [(DbPoolRole, DbConnectionStep, DbConnectionOutcome); 12] = [
(
DbPoolRole::Writer,
DbConnectionStep::WriterPool,
Expand All @@ -114,21 +109,6 @@ pub const CONNECTION_TERMINALS: [(DbPoolRole, DbConnectionStep, DbConnectionOutc
DbConnectionStep::PhysicalConnect,
DbConnectionOutcome::Succeeded,
),
(
DbPoolRole::Writer,
DbConnectionStep::CreatedAtFloor,
DbConnectionOutcome::Succeeded,
),
(
DbPoolRole::Writer,
DbConnectionStep::CreatedAtFloor,
DbConnectionOutcome::Failed,
),
(
DbPoolRole::Writer,
DbConnectionStep::CreatedAtFloor,
DbConnectionOutcome::Cancelled,
),
(
DbPoolRole::Writer,
DbConnectionStep::SessionTimeouts,
Expand Down Expand Up @@ -166,8 +146,8 @@ pub const CONNECTION_TERMINALS: [(DbPoolRole, DbConnectionStep, DbConnectionOutc
),
];

/// Four start counters + four 13-series histograms + fifteen terminal counters.
pub const CONNECTION_RAW_SERIES_PER_POD: usize = 4 + (4 * 13) + 15;
/// Three start counters + three 13-series histograms + twelve terminal counters.
pub const CONNECTION_RAW_SERIES_PER_POD: usize = 3 + (3 * 13) + 12;

pub(crate) struct DbConnectionStepAttempt {
pool_role: DbPoolRole,
Expand Down Expand Up @@ -282,13 +262,12 @@ mod tests {
[
"writer_pool",
"physical_connect",
"created_at_floor",
"session_timeouts",
"isolation",
"ready",
]
);
assert_eq!(CONNECTION_RAW_SERIES_PER_POD, 71);
assert_eq!(CONNECTION_RAW_SERIES_PER_POD, 54);
}

#[test]
Expand All @@ -297,7 +276,7 @@ mod tests {
let snapshotter = recorder.snapshotter();
let _guard = metrics::set_default_local_recorder(&recorder);
let attempt =
DbConnectionStepAttempt::start(DbPoolRole::Writer, DbConnectionStep::CreatedAtFloor);
DbConnectionStepAttempt::start(DbPoolRole::Writer, DbConnectionStep::SessionTimeouts);
drop(attempt);

let metrics = snapshotter.snapshot().into_vec();
Expand All @@ -309,7 +288,7 @@ mod tests {
.map(|label| (label.key(), label.value()))
.collect::<std::collections::BTreeMap<_, _>>();
assert_eq!(labels.get("pool_role"), Some(&"writer"));
assert_eq!(labels.get("step"), Some(&"created_at_floor"));
assert_eq!(labels.get("step"), Some(&"session_timeouts"));
assert!(!labels.contains_key("reason"));
assert!(!labels.contains_key("connection_ordinal"));
match value {
Expand Down
17 changes: 2 additions & 15 deletions crates/buzz-db/src/runtime/migration.rs
Original file line number Diff line number Diff line change
Expand Up @@ -50,14 +50,6 @@ pub(crate) async fn run_migrations_through(pool: &PgPool, target: i64) -> Result
async fn run_migrations_locked(conn: &mut PgConnection) -> Result<()> {
reject_legacy_nip_rs_cardinality_ambiguity(conn).await?;
MIGRATOR.run(&mut *conn).await?;
// The replica-fence proof (see `replica_fence`) requires the commit-time
// `created_at` floor trigger from migration 0021 — correctly shaped — on
// the `events` parent and every partition. `CREATE TABLE .. PARTITION OF`
// clones parent triggers, but a partition attached with `ATTACH
// PARTITION` or created by an older code path would silently escape the
// guard, so migration fails closed if any is missing. (The fence probe
// re-runs this same check at startup on non-migrating relays.)
crate::replica_fence::verify_floor_guard_catalog(&mut *conn).await?;
crate::channel::verify_channel_roster_fence_catalog(&mut *conn).await?;
Ok(())
}
Expand Down Expand Up @@ -705,10 +697,11 @@ mod postgres_tests {
let mut migrations: Vec<_> = MIGRATOR.iter().collect();
migrations.sort_by_key(|migration| migration.version);

assert_eq!(migrations.len(), 60);
assert_eq!(migrations.len(), 61);
assert_eq!(migrations[58].version, 59);
assert!(migrations[58].sql.as_str().contains("buzz.push_enabled"));
assert_eq!(migrations[59].version, 60);
assert_eq!(migrations[60].version, 62);
assert!(migrations[59]
.sql
.as_str()
Expand Down Expand Up @@ -1044,12 +1037,6 @@ mod postgres_tests {
.as_str()
.contains("CREATE TABLE join_policy_acceptances"));

// Replica-fence commit-time floor guard on channel-bearing events.
assert_eq!(migrations[20].version, 21);
assert!(migrations[20]
.sql
.as_str()
.contains("events_created_at_floor_guard"));
assert!(!migrations[0]
.sql
.as_str()
Expand Down
84 changes: 21 additions & 63 deletions crates/buzz-db/src/runtime/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -356,8 +356,8 @@ pub(crate) mod route_proof {
use uuid::Uuid;

/// Proof that a query/page can only return rows with
/// `channel_id IS NOT NULL` — the domain of the commit-time floor guard
/// (migration 0021). `channel_ids` (retains channel-NULL rows) and
/// `channel_id IS NOT NULL` — the domain of the application floor check
/// (see [`replica_fence`]). `channel_ids` (retains channel-NULL rows) and
/// `global_only = false` are explicitly NOT proofs.
///
/// Each constructor keys off *how* its path proves channel-bearing-ness:
Expand Down Expand Up @@ -415,18 +415,18 @@ pub(crate) enum RoutePredicate {
/// read budget `B` (default off). Bounds TIME — the page misses at most
/// the freshest `B` of writes. Sound for ANY query shape, including
/// global (channel-NULL) rows: it relies only on heartbeat commit order,
/// not the floor guard.
/// not the floor check.
Bounded,
/// Completeness: the proved wall must cover the page's upper bound.
/// Bounds CONTENT — every row at/below `upper` is present, meaningful
/// even when the cursor is hours old, where `B`-freshness says nothing.
/// Sound ONLY on the floor guard's domain (channel-bearing rows), hence
/// Sound ONLY on the application floor's domain (channel-bearing rows), hence
/// the proof token. `upper` is non-optional: the no-upper-bound
/// post-verifying case is [`RoutePredicate::CoveredPostVerified`].
///
/// Bounds INSERT-completeness only — "no missing rows", not "no extra
/// rows". Soft deletes are `UPDATE .. SET deleted_at` commits outside
/// the floor guard and never touch `created_at`, so a covered page can
/// the floor check and never touch `created_at`, so a covered page can
/// briefly serve a row the writer already excludes; deletion visibility
/// is bounded by replication lag under `FENCE_STALENESS` (30s), not by
/// `upper` or `B`. Do not extend the covered arm to a surface that
Expand Down Expand Up @@ -708,10 +708,6 @@ impl Db {
/// When `config.read_database_url` is set, a second pool with the same
/// sizing is connected to it for lag-tolerant reads (see [`Db::read`]).
///
/// The writer pool arms the commit-time `created_at` floor guard
/// (migration 0021) on every connection by setting the
/// `buzz.created_at_floor` GUC — this is what makes the replica fence
/// proof hold for every insert path that goes through this pool.
pub async fn new(config: &DbConfig) -> Result<Self> {
let pool = Self::connect_writer_pool(config).await?;
let read_max_connections = config
Expand Down Expand Up @@ -742,11 +738,10 @@ impl Db {

/// Connect the writer pool with all session-level safety premises.
///
/// SQLx stores one `after_connect` hook, so the floor guard and transaction
/// isolation assertion must remain in this single closure. Registering a
/// second hook replaces the first and silently disarms the floor trigger.
/// SQLx stores one `after_connect` hook, so session timeouts and the
/// transaction isolation assertion must remain in this single closure.
/// Additional writer pools, including the relay audit pool, must use this
/// constructor so they inherit the timeout, floor-guard, and isolation
/// constructor so they inherit the timeout and isolation
/// policy installed by [`Db::new`].
pub async fn connect_writer_pool(config: &DbConfig) -> Result<PgPool> {
use connection_observability::{
Expand All @@ -771,22 +766,6 @@ impl Db {
// point at which DNS/network/TLS/authentication succeeded.
record_milestone(DbPoolRole::Writer, DbConnectionStep::PhysicalConnect);

// `SET` cannot take bind parameters; `set_config` can.
let floor = DbConnectionStepAttempt::start(
DbPoolRole::Writer,
DbConnectionStep::CreatedAtFloor,
);
if let Err(error) =
sqlx::query("SELECT set_config('buzz.created_at_floor', $1, false)")
.bind(replica_fence::CREATED_AT_FLOOR_SECS.to_string())
.execute(&mut *conn)
.await
{
floor.fail();
return Err(error);
}
floor.succeed();

// `lock_timeout` fails the waiting statement; it does not
// cancel the holder. `idle_in_transaction_session_timeout`
// reaps only holders idling inside an open transaction,
Expand Down Expand Up @@ -893,9 +872,8 @@ impl Db {
/// are dialed only on first acquire; the ~10-minute reaper never tops
/// the pool back up, which is fine — routed reads re-fill it on demand.
///
/// No floor guard or writer-isolation assertion: replica sessions are
/// read-only, so the commit-time trigger from migration 0021 never fires
/// here and the write fence that depends on READ COMMITTED is never reached.
/// No writer-isolation assertion: replica sessions are read-only, so the
/// write fence that depends on READ COMMITTED is never reached here.
fn connect_read_pool(config: &DbConfig, url: &str, max_connections: u32) -> Result<PgPool> {
Ok(PgPoolOptions::new()
.max_connections(max_connections)
Expand Down Expand Up @@ -1015,43 +993,19 @@ impl Db {
&self.fence
}

/// Verify the floor guard end-to-end, then spawn the background fence
/// probe. Returns `Ok(false)` when no replica is configured.
///
/// Ordering matters (Perci, PR #2084 review): this must run **after**
/// the migration decision. On a relay with `BUZZ_AUTO_MIGRATE` off, the
/// writer pool arms the GUC regardless, but if migration 0021 has not
/// been applied there is no trigger enforcing it — and a heartbeat probe
/// would open the fence over an unenforced floor. So the probe is gated
/// on an unconditional two-part verification against the live schema:
/// catalog shape ([`replica_fence::verify_floor_guard_catalog`]) and
/// observed semantics through this exact pool
/// ([`replica_fence::verify_floor_guard_behavior`]).
///
/// On any verification failure the probe is never spawned and the fence
/// stays closed: every cursor page routes to the writer. The relay keeps
/// serving — degraded capacity, never holes.
pub async fn spawn_fence_probe(&self) -> Result<bool> {
/// Spawn the background fence probe after the migration decision.
/// Returns `false` when no replica is configured. Probe failures keep
/// the fence closed and cursor reads on the writer. Application floor
/// coverage is enforced by the supported-writer contract and fleet gate.
pub fn spawn_fence_probe(&self) -> bool {
if self.read_pool.is_none() {
return Ok(false);
return false;
}
self.verify_replica_fence_at_boot().await?;
tokio::spawn(replica_fence::run_probe(
self.pool.clone(),
std::sync::Arc::clone(&self.fence),
));
Ok(true)
}

/// Verify replica-fence catalog shape and behavior through attributed
/// writer/bootstrap acquisitions without starting the recurring probe.
pub(crate) async fn verify_replica_fence_at_boot(&self) -> Result<()> {
let mut connection =
observability::acquire_writer(&self.pool, observability::WriterOperation::Bootstrap)
.await?;
replica_fence::verify_floor_guard_catalog(&mut *connection).await?;
drop(connection);
replica_fence::verify_floor_guard_behavior(&self.pool).await
true
}

/// The pool for lag-tolerant reads: the read replica when configured,
Expand Down Expand Up @@ -1613,3 +1567,7 @@ mod pool_role_tests {
#[cfg(test)]
#[path = "tests.rs"]
mod postgres_tests;

#[cfg(test)]
#[path = "tests/floor_retirement.rs"]
mod floor_retirement_postgres_tests;
Loading
Loading