diff --git a/crates/buzz-db/src/runtime/migration.rs b/crates/buzz-db/src/runtime/migration.rs index ccbd8615ab3..9365ea10d7a 100644 --- a/crates/buzz-db/src/runtime/migration.rs +++ b/crates/buzz-db/src/runtime/migration.rs @@ -58,7 +58,6 @@ async fn run_migrations_locked(conn: &mut PgConnection) -> Result<()> { // 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(()) } diff --git a/crates/buzz-db/src/store/channel.rs b/crates/buzz-db/src/store/channel.rs index e577113ef9f..8907b224a93 100644 --- a/crates/buzz-db/src/store/channel.rs +++ b/crates/buzz-db/src/store/channel.rs @@ -27,9 +27,9 @@ pub use crate::channel_members::{ add_member, get_accessible_channel_ids, get_accessible_channels, get_bot_members, get_member_count, get_member_counts_bulk, get_member_role, get_members, get_members_bulk, get_users_bulk, is_member, list_large_channel_rosters_needing_reconciliation, - lock_member_snapshot, membership_pairs, remove_member, verify_channel_roster_fence_behavior, - verify_channel_roster_fence_catalog, AccessibleChannel, BotChannelEntry, BotMemberRecord, - ChannelIdentity, LargeChannelRoster, LockedMemberSnapshot, MemberRecord, UserRecord, + lock_member_snapshot, membership_pairs, remove_member, AccessibleChannel, BotChannelEntry, + BotMemberRecord, ChannelIdentity, LargeChannelRoster, LockedMemberSnapshot, MemberRecord, + UserRecord, }; /// A channel row as returned from the database. diff --git a/crates/buzz-db/src/store/channel_members.rs b/crates/buzz-db/src/store/channel_members.rs index 0922be59487..01ce4a5d590 100644 --- a/crates/buzz-db/src/store/channel_members.rs +++ b/crates/buzz-db/src/store/channel_members.rs @@ -50,11 +50,11 @@ const CHANNEL_MEMBERSHIP_LOCK_NAMESPACE: &str = "buzz_channel_membership:"; /// Verify that migration 0032's roster fence is active on the partitioned /// `events` parent and every attached partition. /// -/// New roster publishers depend on this database-side guard to serialize with -/// legacy publishers during a rolling deployment. If the migration has not -/// been applied, publishing with the new lock protocol would falsely appear -/// safe while an old pod could still overwrite it with stale membership. -pub async fn verify_channel_roster_fence_catalog<'e>( +/// Only tests use this. The relay no longer requires the trigger at startup or +/// after migrating: [`validate_channel_roster_snapshot`] enforces the same rule +/// in every roster writer, so the trigger can be dropped. +#[cfg(test)] +pub(crate) async fn verify_channel_roster_fence_catalog<'e>( executor: impl sqlx::PgExecutor<'e>, ) -> Result<()> { // tgtype bits: 1 = ROW, 2 = BEFORE, 4 = INSERT, 16 = UPDATE, 64 = INSTEAD. @@ -94,87 +94,6 @@ pub async fn verify_channel_roster_fence_catalog<'e>( Ok(()) } -/// Prove migration 0032's roster fence semantics through the live writer pool. -/// -/// The catalog check cannot detect a no-op or otherwise corrupted trigger -/// function. This rolled-back probe verifies that a canonical empty roster is -/// accepted while a stale roster member is rejected with `check_violation`. -pub async fn verify_channel_roster_fence_behavior(pool: &sqlx::PgPool) -> Result<()> { - let connection = crate::observability::acquire_writer( - pool, - crate::observability::WriterOperation::Bootstrap, - ) - .await?; - let mut tx = sqlx::Transaction::begin(connection, None).await?; - let community_id = Uuid::new_v4(); - let channel_id = Uuid::new_v4(); - sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)") - .bind(community_id) - .bind(format!( - "roster-fence-verify-{}.invalid", - community_id.simple() - )) - .execute(&mut *tx) - .await?; - - let insert = |id: Vec, tags: serde_json::Value| { - sqlx::query( - "INSERT INTO events (community_id, id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id, d_tag) \ - VALUES ($1, $2, $3, NOW(), 39002, $4, '', $5, NOW(), $6, $7)", - ) - .bind(community_id) - .bind(id) - .bind(vec![0u8; 32]) - .bind(tags) - .bind(vec![0u8; 64]) - .bind(channel_id) - .bind(channel_id.to_string()) - }; - - insert( - vec![0u8; 32], - serde_json::json!([["d", channel_id.to_string()]]), - ) - .execute(&mut *tx) - .await - .map_err(|error| { - DbError::InvalidData(format!( - "channel roster fence rejected a canonical probe roster: {error}" - )) - })?; - - sqlx::query("SAVEPOINT roster_fence_probe") - .execute(&mut *tx) - .await?; - let stale = insert( - vec![1u8; 32], - serde_json::json!([ - ["d", channel_id.to_string()], - ["p", hex::encode([2u8; 32]), "", "member"] - ]), - ) - .execute(&mut *tx) - .await; - match stale { - Err(sqlx::Error::Database(error)) if error.code().as_deref() == Some("23514") => {} - Ok(_) => { - return Err(DbError::InvalidData( - "channel roster fence is inert: a stale probe roster was accepted".into(), - )); - } - Err(error) => { - return Err(DbError::InvalidData(format!( - "channel roster fence probe failed unexpectedly: {error}" - ))); - } - } - sqlx::query("ROLLBACK TO SAVEPOINT roster_fence_probe") - .execute(&mut *tx) - .await?; - tx.rollback().await?; - Ok(()) -} - /// Take the per-channel membership lock. MUST be the first statement in the /// transaction that then reads roles/owner counts and writes membership, so the /// whole check-then-write sequence is atomic against a concurrent one. @@ -2047,20 +1966,6 @@ pub async fn get_member_role( } impl Db { - /// Verify the mixed-version channel-roster database fence end to end. - #[datastore_span(name = "verify_channel_roster_fence", system = "postgresql")] - pub async fn verify_channel_roster_fence(&self) -> Result<()> { - { - let mut connection = crate::observability::acquire_writer( - &self.pool, - crate::observability::WriterOperation::Bootstrap, - ) - .await?; - verify_channel_roster_fence_catalog(&mut *connection).await?; - } - verify_channel_roster_fence_behavior(&self.pool).await - } - /// Capture the active roster while holding the membership-writer lock. #[datastore_span(name = "lock_member_snapshot", system = "postgresql")] pub async fn lock_member_snapshot( @@ -5073,140 +4978,31 @@ mod postgres_tests { .expect("create channel"); } + /// The relay must run without the 0032 roster trigger: the app-side check + /// in every roster writer is the fence, and a later migration drops the + /// trigger. Migrating a database that has no trigger must succeed. + /// + /// Mutation: restore the catalog check after `MIGRATOR.run` in + /// `run_migrations_locked` → migrating fails → RED. #[tokio::test] #[ignore = "requires Postgres"] - async fn unmigrated_roster_fence_blocks_startup_until_0032_is_applied() { - let admin = PgPool::connect(&admin_url().await) - .await - .expect("connect admin"); - let (pool, scratch_name) = - create_scratch_db_through(&admin, "roster_fence_unmigrated", Some(31)).await; - let db = Db::from_pool(pool.clone()); - - let error = db - .verify_channel_roster_fence() - .await - .expect_err("pre-0032 schema must block roster publishers"); - assert!( - error.to_string().contains("channel roster fence trigger"), - "startup gate must report the missing schema fence: {error}" - ); - let rows_before: i64 = sqlx::query_scalar("SELECT count(*) FROM events WHERE kind = 39002") - .fetch_one(&pool) - .await - .expect("count pre-migration rosters"); - assert_eq!( - rows_before, 0, - "failed startup gate must not publish a roster" - ); - - migration::run_migrations(&pool) - .await - .expect("apply migration 0032"); - db.verify_channel_roster_fence() - .await - .expect("0032 must open the startup gate"); - - drop_scratch_db(&admin, pool, &scratch_name).await; - } - - #[tokio::test] - #[ignore = "requires Postgres"] - async fn channel_roster_fence_behavior_verification_detects_inert_function() { + async fn migrations_run_without_the_roster_trigger() { let admin = PgPool::connect(&admin_url().await) .await .expect("connect admin"); - let (pool, scratch_name) = create_scratch_db(&admin, "roster_fence_inert").await; - let db = Db::from_pool(pool.clone()); - + let (pool, scratch_name) = create_scratch_db(&admin, "roster_fence_absent").await; sqlx::raw_sql( - "CREATE OR REPLACE FUNCTION guard_channel_roster_snapshot() \ - RETURNS TRIGGER AS $$ BEGIN RETURN NEW; END; $$ LANGUAGE plpgsql;", - ) - .execute(&pool) - .await - .expect("replace roster fence with inert body"); - let error = db - .verify_channel_roster_fence() - .await - .expect_err("inert roster fence must fail closed"); - assert!( - error - .to_string() - .contains("stale probe roster was accepted"), - "behavior probe must identify inert semantics: {error}" - ); - - drop_scratch_db(&admin, pool, &scratch_name).await; - } - - #[tokio::test] - #[ignore = "requires Postgres"] - async fn channel_roster_fence_catalog_verification_fails_closed() { - let admin = PgPool::connect(&admin_url().await) - .await - .expect("connect admin"); - let (pool, scratch_name) = create_scratch_db(&admin, "roster_fence_catalog").await; - let db = Db::from_pool(pool.clone()); - - db.verify_channel_roster_fence() - .await - .expect("migrated roster fence must verify"); - - let child: String = sqlx::query_scalar( - "SELECT n.nspname || '.' || c.relname \ - FROM pg_inherits i JOIN pg_class c ON c.oid = i.inhrelid \ - JOIN pg_namespace n ON n.oid = c.relnamespace \ - WHERE i.inhparent = 'public.events'::regclass ORDER BY i.inhrelid LIMIT 1", + "DROP TRIGGER trg_events_guard_channel_roster_snapshot ON events; \ + DROP FUNCTION guard_channel_roster_snapshot();", ) - .fetch_one(&pool) - .await - .expect("load event partition"); - sqlx::query(sqlx::AssertSqlSafe(format!( - "ALTER TABLE {child} DISABLE TRIGGER trg_events_guard_channel_roster_snapshot" - ))) .execute(&pool) .await - .expect("disable partition roster trigger"); - let error = db - .verify_channel_roster_fence() - .await - .expect_err("disabled partition roster fence must fail closed"); - assert!( - error.to_string().contains(&child), - "verification must identify the unfenced partition: {error}" - ); - - drop_scratch_db(&admin, pool, &scratch_name).await; - } - - #[tokio::test] - #[ignore = "requires Postgres"] - async fn channel_roster_fence_verification_supports_size_one_pool() { - let admin = PgPool::connect(&admin_url().await) - .await - .expect("connect admin"); - let (seed_pool, scratch_name) = create_scratch_db(&admin, "roster_fence_size_one").await; - seed_pool.close().await; + .expect("drop the roster trigger"); + assert!(verify_channel_roster_fence_catalog(&pool).await.is_err()); - let base_url = admin_url().await; - let path = base_url.rfind('/').expect("database URL path"); - let scratch_url = format!("{}/{}", &base_url[..path], scratch_name); - let pool = PgPoolOptions::new() - .max_connections(1) - .acquire_timeout(std::time::Duration::from_secs(1)) - .connect(&scratch_url) + migration::run_migrations(&pool) .await - .expect("connect size-one writer pool"); - let db = Db::from_pool(pool.clone()); - - tokio::time::timeout( - std::time::Duration::from_secs(2), - db.verify_channel_roster_fence(), - ) - .await - .expect("roster verification must not self-deadlock on its second checkout") - .expect("migrated roster fence verifies on a size-one pool"); + .expect("migrating must not require the roster trigger"); drop_scratch_db(&admin, pool, &scratch_name).await; } diff --git a/crates/buzz-db/tests/observability_source.rs b/crates/buzz-db/tests/observability_source.rs index b1c95d7f1c7..cc11238d4fe 100644 --- a/crates/buzz-db/tests/observability_source.rs +++ b/crates/buzz-db/tests/observability_source.rs @@ -884,10 +884,7 @@ fn has_any_tenant_local_chokepoint(source: &str) -> bool { const ADMITTED_TX_SIGNATURE_MARKERS: [&str; 2] = ["&mut AdmittedTx", "&mut crate::AdmittedTx"]; // Narrow reviewed exceptions for non-serving verification probes only. -const GUARDED_WRITE_FUNCTION_EXCEPTIONS: [&str; 2] = [ - "pub async fn verify_floor_guard_behavior(", - "pub async fn verify_channel_roster_fence_behavior(", -]; +const GUARDED_WRITE_FUNCTION_EXCEPTIONS: [&str; 1] = ["pub async fn verify_floor_guard_behavior("]; const GUARDED_TX_ADAPTER_METHOD_PINS: [(&str, &str); 1] = [( "pub async fn replace_member_event(", @@ -1390,10 +1387,7 @@ fn serving_table_writes_expose_syntactic_chokepoint_or_guarded_tx_routes() { // Rolled-back verification probes write `events` rows that never commit, so // they owe no push job or TTL refresh. -const EVENT_INSERT_FOLLOW_UP_EXCEPTIONS: [&str; 2] = [ - "pub async fn verify_floor_guard_behavior(", - "pub async fn verify_channel_roster_fence_behavior(", -]; +const EVENT_INSERT_FOLLOW_UP_EXCEPTIONS: [&str; 1] = ["pub async fn verify_floor_guard_behavior("]; /// `source` with every `#[cfg(test)]` item removed, wherever it sits in the /// file. Each item runs to the `;` that ends it or to the brace that closes @@ -1925,19 +1919,17 @@ fn every_production_event_insert_runs_the_follow_up_hook() { ); } -// Fixed channel-less kinds cannot be channel rosters. The startup probes must -// exercise the retained DB backstops directly and roll back all their writes. +// Fixed channel-less kinds cannot be channel rosters, and the replica-floor +// startup probe rolls back all its writes. // Exemptions match by function header, so a kind 39002 insert added inside one // of these functions would pass the policy; keep them single-purpose. -const EVENT_INSERT_ROSTER_GUARD_EXCEPTIONS: [&str; 4] = [ +const EVENT_INSERT_ROSTER_GUARD_EXCEPTIONS: [&str; 3] = [ // NIP-43 membership is kind 13534 with channel_id NULL. "pub(crate) async fn write_nip43_membership_snapshot(", // Push leases are kind 30350 with channel_id NULL. "pub async fn accept_lease_event(", // Rolled-back replica-floor startup probe. "pub async fn verify_floor_guard_behavior(", - // Rolled-back roster-fence startup probe. - "pub async fn verify_channel_roster_fence_behavior(", ]; fn event_insert_roster_guard_violations(production_source: &str) -> Vec { @@ -1992,9 +1984,6 @@ pub async fn accept_lease_event(tx: &mut AdmittedTx) { pub async fn verify_floor_guard_behavior(pool: &PgPool) { sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)"); } -pub async fn verify_channel_roster_fence_behavior(pool: &PgPool) { - sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)"); -} #[cfg(test)] mod tests { fn fixture() { @@ -2038,10 +2027,10 @@ fn every_production_event_insert_runs_the_roster_guard() { violations.is_empty(), "production `INSERT INTO events` writers must run the app-side roster guard: {violations:?}" ); - // Five guarded writers plus the four explicitly exempt writers/probes. + // Five guarded writers plus the three explicitly exempt writers/probes. assert!( - checked_writers >= 9, - "expected at least nine event writers, found {checked_writers}" + checked_writers >= 8, + "expected at least eight event writers, found {checked_writers}" ); } @@ -2161,7 +2150,7 @@ fn every_production_event_insert_names_d_tag() { // The same writer set as the roster-guard scan. A lower count means the // scan stopped seeing a writer, not that one was removed safely. assert!( - checked_writers >= 9, - "expected at least nine event writers, found {checked_writers}" + checked_writers >= 8, + "expected at least eight event writers, found {checked_writers}" ); } diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index 9cb14040715..e4b70c5059b 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -1113,24 +1113,8 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> { ); } - let step = StepTimer::start(StartupStep::ChannelRosterFence); - match state.db.verify_channel_roster_fence().await { - Ok(()) => { - info!("Channel roster fence verified"); - } - Err(error) => { - error!(%error, "Channel roster fence validation failed"); - return Err(anyhow::anyhow!( - "Channel roster fence is unsafe; apply or repair migration 0032 before starting this relay: {error}" - )); - } - } - step.finish(); - // Repair legacy NIP-29 channel rosters that were persisted while the - // canonical member query still truncated at 1,000 rows. Validation above - // makes migration 0032 a code/schema compatibility gate before the new - // replacement protocol or listener can serve traffic. + // canonical member query still truncated at 1,000 rows. let mut step = StepTimer::start(StartupStep::LargeRosterReconcile); match buzz_relay::handlers::side_effects::reconcile_large_channel_member_snapshots(&state).await { diff --git a/crates/buzz-relay/src/startup_steps.rs b/crates/buzz-relay/src/startup_steps.rs index 1ef069ccd79..55253bbf4e2 100644 --- a/crates/buzz-relay/src/startup_steps.rs +++ b/crates/buzz-relay/src/startup_steps.rs @@ -75,8 +75,6 @@ pub enum StartupStep { MeshInitialReconcile, /// Run the git object-store conformance probe (when enabled). GitConformanceProbe, - /// Verify the channel roster fence. - ChannelRosterFence, /// Repair large NIP-29 channel roster snapshots. LargeRosterReconcile, /// Bind the private health listener. @@ -104,7 +102,6 @@ impl StartupStep { Self::MeshPublish, Self::MeshInitialReconcile, Self::GitConformanceProbe, - Self::ChannelRosterFence, Self::LargeRosterReconcile, Self::HealthBind, Self::ListenerBind, @@ -129,7 +126,6 @@ impl StartupStep { Self::MeshPublish => "mesh_publish", Self::MeshInitialReconcile => "mesh_initial_reconcile", Self::GitConformanceProbe => "git_conformance_probe", - Self::ChannelRosterFence => "channel_roster_fence", Self::LargeRosterReconcile => "large_roster_reconcile", Self::HealthBind => "health_bind", Self::ListenerBind => "listener_bind", diff --git a/deploy/charts/buzz/README.md b/deploy/charts/buzz/README.md index 5f5870e804c..4c664dd2db6 100644 --- a/deploy/charts/buzz/README.md +++ b/deploy/charts/buzz/README.md @@ -616,7 +616,7 @@ default so long-lived WebSocket connections have time to drain. Schema migrations are embedded in the relay binary via `sqlx::migrate!` and run at startup, gated by `BUZZ_AUTO_MIGRATE` (default `true`). Multiple replicas race-safely behind a Postgres advisory lock. `helm upgrade` is the entire upgrade procedure. -Migration 0032 is a hard compatibility boundary for relay versions that publish repaired channel rosters. The relay verifies the roster-fence trigger catalog and behavior before opening listeners and refuses to start if 0032 is missing or inert. Apply migrations before rolling the relay; for large installations, prefer a controlled `buzz-admin migrate` job with PostgreSQL lock monitoring before the code rollout. +Migration 0032 is a hard compatibility boundary for relay versions that publish repaired channel rosters. Older relays verify its roster-fence trigger before opening listeners and refuse to start if the trigger is missing or inert. Newer relays run the same check in every roster writer and do not require the trigger. A later release drops it, and that release requires every relay to run a version without the startup check first. Apply migrations before rolling the relay; for large installations, prefer a controlled `buzz-admin migrate` job with PostgreSQL lock monitoring before the code rollout. If you prefer decoupling migrations from serving, set `migrate.autoMigrate=false`. **In that mode the chart does not run migrations for you** — you own running `buzz-admin migrate` (separate Pod / one-shot Job) against the database before every `helm install` / `helm upgrade`. Readiness probes only verify DB connectivity, not schema freshness, so a pod will appear healthy against an unmigrated schema and fail under load. A pre-upgrade Helm Job for this is on the chart roadmap; the values knob `migrate.preUpgradeJob.enabled` is reserved.