Skip to content

Stop requiring the 0032 roster trigger at relay startup #8611

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
loganj merged 2 commits into from
Oct 28, 2026
Merged

Stop requiring the 0032 roster trigger at relay startup #8611

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
3f81a0f
Stop requiring the 0032 roster trigger at startup
loganj Oct 26, 2026
6e6d290
test(db): drop the stale roster probe note from the guard exceptions
loganj Oct 26, 2026
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
1 change: 0 additions & 1 deletion crates/buzz-db/src/runtime/migration.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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(())
}

Expand Down
6 changes: 3 additions & 3 deletions crates/buzz-db/src/store/channel.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
242 changes: 19 additions & 223 deletions crates/buzz-db/src/store/channel_members.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down Expand Up @@ -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<u8>, 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.
Expand Down Expand Up @@ -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(
Expand Down Expand Up @@ -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;
}
Expand Down
31 changes: 10 additions & 21 deletions crates/buzz-db/tests/observability_source.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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(",
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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<String> {
Expand Down Expand Up @@ -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() {
Expand Down Expand Up @@ -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}"
);
}

Expand Down Expand Up @@ -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}"
);
}
Loading
Loading