From a4218630e170912bb0ccf25108c889f29ba75b52 Mon Sep 17 00:00:00 2001 From: Brad Seiler Date: Wed, 21 Oct 2026 12:51:43 -0700 Subject: [PATCH 1/2] Add invitation history repair as a separate maintenance change Revoke historical legacy candidates and clear examined admission markers under the deletion/admission locks. Require separate approval before core deployment and a controlled rollout that prevents new timeout/mute history. Co-authored-by: Codex Signed-off-by: Brad Seiler Signed-off-by: seiler Signed-off-by: Brad Seiler --- TESTING.md | 32 + crates/buzz-db/TESTING.md | 5 +- .../relay_invite/revocation_postgres_tests.rs | 1 + .../relay_invite/upgrade_postgres_tests.rs | 556 +++++++++++++++++- .../backfill_invite_revocations.sql | 168 ++++++ 5 files changed, 757 insertions(+), 5 deletions(-) create mode 100644 scripts/maintenance/backfill_invite_revocations.sql diff --git a/TESTING.md b/TESTING.md index cc41319f642..035fe059c58 100644 --- a/TESTING.md +++ b/TESTING.md @@ -222,6 +222,38 @@ invitation admission paused, using the same traffic/writer controls if needed. It also requires `BUZZ_AUTO_MIGRATE` off: with auto-migration on (the Helm default), the old binary fails at startup with SQLx `VersionMissing(75)`. See [docs/database.md](docs/database.md) Rule 5. +The separately approved companion provides the operator command: + +```bash +psql -X -v ON_ERROR_STOP=1 -f scripts/maintenance/backfill_invite_revocations.sql +``` + +Run it as a top-level script, outside any caller transaction. The final query +must report `0` unrepaired invites in active communities. Run it twice; the +second run must keep zero and leave revocation timestamps and markers unchanged. +Review fenced communities separately and keep their admission closed until they +can be repaired under the same controls. + +Use a **fresh disposable database through migration 0074**, the complete +pre-change schema including channel join roles, for upgrade testing. Do not reuse +an older database that applied the invitation migration numbered 0074 or a +different 0075 checksum; reconcile it separately without resetting a user's +database. Seed legacy links for a restricted issuer and an unrelated issuer; +lift the first restriction before +upgrade. Apply schema-only 0075 and confirm it rewrites no invite rows. Before +repair, signed claims reject historical legacy candidates lazily. Run the actual +repair script twice, checking selective revocation, cleared markers, zero +remaining active-community legacy rows and idempotence. The populated automated +fixtures also exercise real `bin/pgschema apply --file schema/schema.sql`, then +`scripts/reconcile-schema-after-pgschema.sql` twice, for outcome parity. + +After repair, a newly applied timeout/mute must temporarily block claims without +durable revocation; after expiry the same link works. Ban via the authorized +owner path and verify permanent rejection, even after unban and relay restart. +A fresh replacement minted by an unrestricted issuer must work. Include current +owner restrictions, noncanonical issuers, community/deletion fences, and the +marker-clearing lock-serialization tests in the full PostgreSQL lane. + Legacy owner attribution uses the agent's currently recorded `agent_owner_pubkey`; deleting an owner user row out of band clears that link through its foreign key, so a later repair cannot recover the former diff --git a/crates/buzz-db/TESTING.md b/crates/buzz-db/TESTING.md index 17b42b8bb91..39ad9d83a2d 100644 --- a/crates/buzz-db/TESTING.md +++ b/crates/buzz-db/TESTING.md @@ -68,5 +68,6 @@ Invite upgrade tests start from the complete schema through migration 0074, including channel join roles, entity deletion approval, the unlisted channel setting and the user deletion target and fences. They exercise the invitation schema expansion (migration 0075) through both migration and desired-state paths against populated -data, retaining a non-default channel join role. Historical repair is validated -separately from the application/schema change. +data, retaining a non-default channel join role. They also execute the operator +repair script with `psql`; make sure the PostgreSQL client is installed and +resolves on `PATH`. diff --git a/crates/buzz-db/src/store/relay_invite/revocation_postgres_tests.rs b/crates/buzz-db/src/store/relay_invite/revocation_postgres_tests.rs index 54da3cf1749..3f38b74ae34 100644 --- a/crates/buzz-db/src/store/relay_invite/revocation_postgres_tests.rs +++ b/crates/buzz-db/src/store/relay_invite/revocation_postgres_tests.rs @@ -317,6 +317,7 @@ async fn stalled_production_mint_ignores_lifted_history_and_later_audit_touches( .unwrap(); } + super::upgrade_postgres_tests::run_invite_repair_script(&pool).await; let revoked: Option> = sqlx::query_scalar("SELECT revoked_at FROM relay_invites WHERE community_id=$1 AND id=$2") .bind(community.as_uuid()) diff --git a/crates/buzz-db/src/store/relay_invite/upgrade_postgres_tests.rs b/crates/buzz-db/src/store/relay_invite/upgrade_postgres_tests.rs index 7a8b8fb3442..a09b6acfd7d 100644 --- a/crates/buzz-db/src/store/relay_invite/upgrade_postgres_tests.rs +++ b/crates/buzz-db/src/store/relay_invite/upgrade_postgres_tests.rs @@ -4,6 +4,8 @@ use uuid::Uuid; struct LegacyInviteCase { community: Uuid, + revoke: bool, + active: bool, } struct RelayAdminActionFixture<'a> { @@ -335,7 +337,12 @@ async fn legacy_fixture() -> (PgPool, Vec) { set_deletion_state(&pool, community, "tombstone").await; } - cases.push(LegacyInviteCase { community }); + let active = !matches!(index, 1 | 2); + cases.push(LegacyInviteCase { + community, + revoke: index < 8 && active, + active, + }); } // Preserve a populated, non-default join role across both upgrade paths. sqlx::query( @@ -378,6 +385,70 @@ async fn assert_join_role_preserved(pool: &PgPool, community: Uuid) { ); } +async fn start_invite_repair_script( + pool: &PgPool, + application_name: Option<&str>, +) -> std::process::Child { + let database_url = crate::test_support::database_url(); + let options: sqlx::postgres::PgConnectOptions = database_url.parse().unwrap(); + let host = options.get_host().to_owned(); + let port = options.get_port().to_string(); + let user = options.get_username().to_owned(); + let password = database_url + .split_once("://") + .and_then(|(_, rest)| rest.split_once('@')) + .and_then(|(authority, _)| authority.split_once(':')) + .map(|(_, password)| password.to_owned()) + .or_else(|| std::env::var("PGPASSWORD").ok()) + .unwrap_or_default(); + let database: String = sqlx::query_scalar("SELECT current_database()") + .fetch_one(pool) + .await + .unwrap(); + let script = std::path::Path::new(env!("CARGO_MANIFEST_DIR")) + .join("../../scripts/maintenance/backfill_invite_revocations.sql"); + let mut command = std::process::Command::new("psql"); + command + .args(["-X", "-v", "ON_ERROR_STOP=1"]) + .arg("--host") + .arg(host) + .arg("--port") + .arg(port) + .arg("--username") + .arg(user) + .arg("--dbname") + .arg(database) + .arg("--file") + .arg(script) + .env("PGPASSWORD", password) + .stdout(std::process::Stdio::piped()) + .stderr(std::process::Stdio::piped()); + if let Some(application_name) = application_name { + command.env("PGAPPNAME", application_name); + } + command.spawn().expect("start invite repair script") +} + +pub(super) async fn run_invite_repair_script(pool: &PgPool) -> String { + let child = start_invite_repair_script(pool, None).await; + let output = tokio::task::spawn_blocking(move || child.wait_with_output()) + .await + .unwrap() + .unwrap(); + assert!( + output.status.success(), + "invite repair script failed:\n{}\n{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr), + ); + let stdout = String::from_utf8_lossy(&output.stdout).into_owned(); + assert!( + stdout.lines().any(|line| line.trim() == "0"), + "repair validation did not return zero unrepaired invites:\n{stdout}" + ); + stdout +} + async fn invite_revocation_state(pool: &PgPool) -> Vec<(Uuid, Option>, bool)> { sqlx::query_as( "SELECT community_id, revoked_at, legacy_admission \ @@ -417,6 +488,47 @@ async fn assert_admission_column(pool: &PgPool) { ); } +async fn assert_repaired_and_fenced(pool: &PgPool, cases: &[LegacyInviteCase]) { + assert_admission_column(pool).await; + for case in cases { + let (revoked, legacy): (bool, bool) = sqlx::query_as( + "SELECT revoked_at IS NOT NULL, legacy_admission \ + FROM relay_invites WHERE community_id = $1", + ) + .bind(case.community) + .fetch_one(pool) + .await + .unwrap(); + assert_eq!(revoked, case.revoke, "tenant {}", case.community); + assert_eq!(legacy, !case.active, "legacy marker for {}", case.community); + } + let unfenced_error = + sqlx::query("UPDATE relay_invites SET use_count = use_count + 1 WHERE community_id = $1") + .bind(cases[1].community) + .execute(pool) + .await + .unwrap_err(); + assert!( + unfenced_error + .to_string() + .contains("community write fenced"), + "{unfenced_error}" + ); + let invitation_triggers: i64 = sqlx::query_scalar( + "SELECT count(*) FROM pg_trigger WHERE NOT tgisinternal \ + AND tgname IN ('invite_restriction_lock', 'invite_issuer_revocation', \ + 'invite_owner_lock', 'invite_owner_revocation', \ + 'community_write_fence_relay_invites')", + ) + .fetch_one(pool) + .await + .unwrap(); + assert_eq!( + invitation_triggers, 1, + "only the community deletion fence remains" + ); +} + async fn apply_pgschema_to_existing_database(pool: &PgPool) { // The pinned migration history has retired triggers on the partitioned // events parent and auto-named deletion-request checks that the desired @@ -503,7 +615,7 @@ async fn apply_pgschema_to_existing_database(pool: &PgPool) { #[tokio::test] #[ignore = "requires Postgres"] -async fn migration_schema_0075_adds_invite_columns_without_rewriting_history() { +async fn migration_schema_backfill_repairs_retained_restrictions_and_ignores_audit_only_history() { let (pool, cases) = legacy_fixture().await; crate::migration::run_migrations(&pool).await.unwrap(); assert!(sqlx::query_scalar::<_, bool>( @@ -535,6 +647,14 @@ async fn migration_schema_0075_adds_invite_columns_without_rewriting_history() { "pre-migration rows remain legacy-admitted" ); + run_invite_repair_script(&pool).await; + let repaired = invite_revocation_state(&pool).await; + assert_repaired_and_fenced(&pool, &cases).await; + run_invite_repair_script(&pool).await; + assert_eq!(invite_revocation_state(&pool).await, repaired); + assert_repaired_and_fenced(&pool, &cases).await; + assert_join_role_preserved(&pool, cases[0].community).await; + insert_legacy_invite( &pool, cases[8].community, @@ -553,6 +673,17 @@ async fn migration_schema_0075_adds_invite_columns_without_rewriting_history() { .await .unwrap(); assert!(old_code_insert_default); + run_invite_repair_script(&pool).await; + let old_code_insert_repaired: bool = sqlx::query_scalar( + "SELECT legacy_admission FROM relay_invites \ + WHERE community_id=$1 AND token_hash=$2", + ) + .bind(cases[8].community) + .bind([100u8; 32].as_slice()) + .fetch_one(&pool) + .await + .unwrap(); + assert!(!old_code_insert_repaired); } #[tokio::test] @@ -561,6 +692,7 @@ async fn migration_schema_legacy_claim_lazily_revokes_history_after_a_fenced_com { let (pool, cases) = legacy_fixture().await; crate::migration::run_migrations(&pool).await.unwrap(); + run_invite_repair_script(&pool).await; let community = cases[1].community; let token_hash = [2u8; 32]; @@ -621,7 +753,18 @@ async fn migration_schema_legacy_claim_lazily_revokes_history_after_a_fenced_com .unwrap(); assert!( legacy_admission, - "claim does not stand in for maintenance repair" + "fenced communities retain the legacy marker until repaired" + ); + run_invite_repair_script(&pool).await; + let legacy_admission: bool = + sqlx::query_scalar("SELECT legacy_admission FROM relay_invites WHERE community_id = $1") + .bind(community) + .fetch_one(&pool) + .await + .unwrap(); + assert!( + !legacy_admission, + "active community is repaired on the next run" ); } @@ -669,4 +812,411 @@ async fn migration_schema_desired_pgschema_reconcile_matches_populated_upgrade() 0, "pgschema preserves the legacy default for populated rows" ); + + run_invite_repair_script(&pool).await; + let repaired = invite_revocation_state(&pool).await; + assert_repaired_and_fenced(&pool, &cases).await; + run_invite_repair_script(&pool).await; + assert_eq!(invite_revocation_state(&pool).await, repaired); + assert_repaired_and_fenced(&pool, &cases).await; + assert_join_role_preserved(&pool, cases[0].community).await; +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn migration_schema_repaired_marker_survives_reconnect_and_timeout_expires_without_revocation( +) { + let (pool, cases) = legacy_fixture().await; + crate::migration::run_migrations(&pool).await.unwrap(); + run_invite_repair_script(&pool).await; + + let community = CommunityId::from_uuid(cases[8].community); + let token_hash = [9u8; 32]; + let legacy_admission: bool = sqlx::query_scalar( + "SELECT legacy_admission FROM relay_invites \ + WHERE community_id=$1 AND token_hash=$2", + ) + .bind(cases[8].community) + .bind(token_hash.as_slice()) + .fetch_one(&pool) + .await + .unwrap(); + assert!( + !legacy_admission, + "repair clears a spared active legacy row" + ); + + pool.close().await; + let pool = PgPool::connect(&crate::test_support::database_url()) + .await + .unwrap(); + let legacy_admission: bool = sqlx::query_scalar( + "SELECT legacy_admission FROM relay_invites \ + WHERE community_id=$1 AND token_hash=$2", + ) + .bind(cases[8].community) + .bind(token_hash.as_slice()) + .fetch_one(&pool) + .await + .unwrap(); + assert!( + !legacy_admission, + "repaired marker is persisted in the database" + ); + + let issuer = [1u8; 32]; + crate::moderation::timeout_member( + &pool, + community, + &issuer, + &[3; 32], + Utc::now() + Duration::hours(1), + None, + ) + .await + .unwrap(); + let claimant = hex::encode([9u8; 32]); + assert_eq!( + super::revocation_postgres_tests::claim_relay_invite_outcome( + &pool, + community, + &token_hash, + &claimant, + None, + ) + .await + .unwrap(), + ClaimOutcome::IssuerRestricted + ); + let revoked_while_timed_out: bool = sqlx::query_scalar( + "SELECT revoked_at IS NOT NULL FROM relay_invites \ + WHERE community_id=$1 AND token_hash=$2", + ) + .bind(cases[8].community) + .bind(token_hash.as_slice()) + .fetch_one(&pool) + .await + .unwrap(); + assert!( + !revoked_while_timed_out, + "a post-repair timeout is temporary" + ); + + sqlx::query( + "UPDATE community_bans SET muted_until=clock_timestamp()-interval '1 second' \ + WHERE community_id=$1 AND pubkey=$2", + ) + .bind(community.as_uuid()) + .bind(issuer.as_slice()) + .execute(&pool) + .await + .unwrap(); + assert!(matches!( + super::revocation_postgres_tests::claim_relay_invite_outcome( + &pool, + community, + &token_hash, + &claimant, + None, + ) + .await + .unwrap(), + ClaimOutcome::Joined { .. } + )); + let revoked_after_timeout: bool = sqlx::query_scalar( + "SELECT revoked_at IS NOT NULL FROM relay_invites \ + WHERE community_id=$1 AND token_hash=$2", + ) + .bind(cases[8].community) + .bind(token_hash.as_slice()) + .fetch_one(&pool) + .await + .unwrap(); + assert!(!revoked_after_timeout); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn invite_repair_revokes_new_protocol_invites_for_active_bans() { + let (pool, community, issuer, claimant) = super::revocation_postgres_tests::fixture().await; + let invite = mint_relay_invite(&pool, community, &issuer, 3600, None) + .await + .unwrap(); + sqlx::query( + "INSERT INTO community_bans (community_id, pubkey, banned, actor_pubkey) \ + VALUES ($1, $2, true, $3)", + ) + .bind(community.as_uuid()) + .bind([1u8; 32].as_slice()) + .bind([3u8; 32].as_slice()) + .execute(&pool) + .await + .unwrap(); + + run_invite_repair_script(&pool).await; + let (revoked, legacy): (bool, bool) = sqlx::query_as( + "SELECT revoked_at IS NOT NULL, legacy_admission \ + FROM relay_invites WHERE community_id=$1 AND id=$2", + ) + .bind(community.as_uuid()) + .bind(invite.invite_id) + .fetch_one(&pool) + .await + .unwrap(); + assert!( + revoked, + "repair preserves active-ban revocation for new rows" + ); + assert!(!legacy, "new-protocol rows remain outside legacy admission"); + assert_eq!( + super::revocation_postgres_tests::claim_relay_invite_outcome( + &pool, + community, + &hash_v2_code(&invite.code), + &claimant, + None, + ) + .await + .unwrap(), + ClaimOutcome::Revoked + ); +} + +async fn assert_repair_waits_for_admission(pool: &PgPool, application_name: &str) { + let has_shared_deletion_lock: bool = sqlx::query_scalar( + "SELECT EXISTS (\ + SELECT 1 FROM pg_locks AS lock \ + JOIN pg_stat_activity AS activity ON activity.pid=lock.pid \ + WHERE activity.application_name=$1 \ + AND lock.locktype='advisory' AND lock.granted \ + AND lock.mode='ShareLock'\ + )", + ) + .bind(application_name) + .fetch_one(pool) + .await + .unwrap(); + assert!( + has_shared_deletion_lock, + "repair takes the shared deletion lock before waiting for admission" + ); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn invite_repair_serializes_with_ban_before_clearing_legacy_marker() { + let (pool, community, issuer, _) = super::revocation_postgres_tests::fixture().await; + insert_legacy_invite(&pool, *community.as_uuid(), 81, &issuer, Utc::now()).await; + insert_restriction(&pool, *community.as_uuid(), &[1; 32], false, None, None).await; + + let mut restriction_row = pool.begin().await.unwrap(); + sqlx::query("SELECT pubkey FROM community_bans WHERE community_id=$1 FOR UPDATE") + .bind(community.as_uuid()) + .fetch_one(&mut *restriction_row) + .await + .unwrap(); + let (ban_pool, ban_name) = super::revocation_postgres_tests::named_pool().await; + let ban = tokio::spawn(async move { + crate::moderation::ban_member(&ban_pool, community, &[1; 32], &[3; 32], None, None).await + }); + super::revocation_postgres_tests::wait_for_lock(&pool, &ban_name, "transactionid", &ban).await; + + let repair_name = format!("invite-repair-{}", Uuid::new_v4()); + let child = start_invite_repair_script(&pool, Some(&repair_name)).await; + let repair = tokio::task::spawn_blocking(move || child.wait_with_output()); + super::revocation_postgres_tests::wait_for_lock(&pool, &repair_name, "advisory", &repair).await; + assert_repair_waits_for_admission(&pool, &repair_name).await; + + restriction_row.commit().await.unwrap(); + ban.await.unwrap().unwrap(); + let output = repair.await.unwrap().unwrap(); + assert!( + output.status.success(), + "repair failed:\n{}\n{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr), + ); + let (revoked, legacy): (bool, bool) = sqlx::query_as( + "SELECT revoked_at IS NOT NULL, legacy_admission FROM relay_invites \ + WHERE community_id=$1 AND token_hash=$2", + ) + .bind(community.as_uuid()) + .bind([81u8; 32].as_slice()) + .fetch_one(&pool) + .await + .unwrap(); + assert!(revoked); + assert!(!legacy); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn invite_repair_serializes_with_owner_change_before_clearing_legacy_marker() { + let (pool, community, _, _) = super::revocation_postgres_tests::fixture().await; + let agent = [4u8; 32]; + let owner = [5u8; 32]; + sqlx::query("INSERT INTO users (community_id,pubkey) VALUES ($1,$2),($1,$3)") + .bind(community.as_uuid()) + .bind(agent.as_slice()) + .bind(owner.as_slice()) + .execute(&pool) + .await + .unwrap(); + insert_restriction(&pool, *community.as_uuid(), &owner, false, None, None).await; + let issuer = hex::encode(agent); + insert_legacy_invite(&pool, *community.as_uuid(), 82, &issuer, Utc::now()).await; + + let mut user_row = pool.begin().await.unwrap(); + sqlx::query("SELECT pubkey FROM users WHERE community_id=$1 AND pubkey=$2 FOR UPDATE") + .bind(community.as_uuid()) + .bind(agent.as_slice()) + .fetch_one(&mut *user_row) + .await + .unwrap(); + let (owner_pool, owner_name) = super::revocation_postgres_tests::named_pool().await; + let owner_write = tokio::spawn(async move { + crate::user::set_agent_owner(&owner_pool, community, &agent, &owner).await + }); + super::revocation_postgres_tests::wait_for_lock( + &pool, + &owner_name, + "transactionid", + &owner_write, + ) + .await; + + let repair_name = format!("invite-repair-{}", Uuid::new_v4()); + let child = start_invite_repair_script(&pool, Some(&repair_name)).await; + let repair = tokio::task::spawn_blocking(move || child.wait_with_output()); + super::revocation_postgres_tests::wait_for_lock(&pool, &repair_name, "advisory", &repair).await; + assert_repair_waits_for_admission(&pool, &repair_name).await; + + user_row.commit().await.unwrap(); + assert!(owner_write.await.unwrap().unwrap()); + let output = repair.await.unwrap().unwrap(); + assert!( + output.status.success(), + "repair failed:\n{}\n{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr), + ); + let (revoked, legacy): (bool, bool) = sqlx::query_as( + "SELECT revoked_at IS NOT NULL, legacy_admission FROM relay_invites \ + WHERE community_id=$1 AND token_hash=$2", + ) + .bind(community.as_uuid()) + .bind([82u8; 32].as_slice()) + .fetch_one(&pool) + .await + .unwrap(); + assert!( + revoked, + "repair sees the committed current owner restriction" + ); + assert!(!legacy); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn invite_repair_waits_for_deletion_transition_before_clearing_legacy_marker() { + let (pool, community, issuer, _) = super::revocation_postgres_tests::fixture().await; + insert_legacy_invite(&pool, *community.as_uuid(), 83, &issuer, Utc::now()).await; + insert_restriction(&pool, *community.as_uuid(), &[1; 32], false, None, None).await; + + let mut deletion = pool.begin().await.unwrap(); + sqlx::query("SELECT pg_advisory_xact_lock(community_deletion_lock_key($1))") + .bind(community.as_uuid()) + .execute(&mut *deletion) + .await + .unwrap(); + sqlx::query( + "SELECT set_config('buzz.deletion_executor_community',$1,true), \ + set_config('buzz.deletion_fence_generation','7',true)", + ) + .bind(community.to_string()) + .execute(&mut *deletion) + .await + .unwrap(); + sqlx::query( + "UPDATE communities SET deletion_state='quiescing', \ + deletion_fence_generation=7 WHERE id=$1", + ) + .bind(community.as_uuid()) + .execute(&mut *deletion) + .await + .unwrap(); + + let repair_name = format!("invite-repair-{}", Uuid::new_v4()); + let child = start_invite_repair_script(&pool, Some(&repair_name)).await; + let repair = tokio::task::spawn_blocking(move || child.wait_with_output()); + super::revocation_postgres_tests::wait_for_lock(&pool, &repair_name, "advisory", &repair).await; + let waiting_for_deletion_before_admission: bool = sqlx::query_scalar( + "SELECT EXISTS (\ + SELECT 1 FROM pg_locks AS lock \ + JOIN pg_stat_activity AS activity ON activity.pid=lock.pid \ + WHERE activity.application_name=$1 AND lock.locktype='advisory' \ + AND NOT lock.granted AND lock.mode='ShareLock'\ + ) AND NOT EXISTS (\ + SELECT 1 FROM pg_locks AS lock \ + JOIN pg_stat_activity AS activity ON activity.pid=lock.pid \ + WHERE activity.application_name=$1 AND lock.locktype='advisory' \ + AND lock.granted AND lock.mode='ExclusiveLock'\ + )", + ) + .bind(&repair_name) + .fetch_one(&pool) + .await + .unwrap(); + assert!( + waiting_for_deletion_before_admission, + "repair waits for the deletion lock before taking admission" + ); + let (revoked, legacy): (bool, bool) = sqlx::query_as( + "SELECT revoked_at IS NOT NULL, legacy_admission FROM relay_invites \ + WHERE community_id=$1 AND token_hash=$2", + ) + .bind(community.as_uuid()) + .bind([83u8; 32].as_slice()) + .fetch_one(&pool) + .await + .unwrap(); + assert!( + !revoked && legacy, + "repair must not write through deletion lock" + ); + + deletion.commit().await.unwrap(); + let output = repair.await.unwrap().unwrap(); + assert!( + output.status.success(), + "repair failed:\n{}\n{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr), + ); + let (revoked, legacy): (bool, bool) = sqlx::query_as( + "SELECT revoked_at IS NOT NULL, legacy_admission FROM relay_invites \ + WHERE community_id=$1 AND token_hash=$2", + ) + .bind(community.as_uuid()) + .bind([83u8; 32].as_slice()) + .fetch_one(&pool) + .await + .unwrap(); + assert!(!revoked && legacy, "quiescing communities remain fenced"); + + set_deletion_state(&pool, *community.as_uuid(), "active").await; + run_invite_repair_script(&pool).await; + let (revoked, legacy): (bool, bool) = sqlx::query_as( + "SELECT revoked_at IS NOT NULL, legacy_admission FROM relay_invites \ + WHERE community_id=$1 AND token_hash=$2", + ) + .bind(community.as_uuid()) + .bind([83u8; 32].as_slice()) + .fetch_one(&pool) + .await + .unwrap(); + assert!( + revoked && !legacy, + "active community is repaired on the next run" + ); } diff --git a/scripts/maintenance/backfill_invite_revocations.sql b/scripts/maintenance/backfill_invite_revocations.sql new file mode 100644 index 00000000000..96859417ae2 --- /dev/null +++ b/scripts/maintenance/backfill_invite_revocations.sql @@ -0,0 +1,168 @@ +-- This companion must be separately approved before core deployment and run +-- in that same controlled rollout, after schema migration 0068 and draining +-- every old writer. Prevent invite mint/claim, ban lifts, and NEW timeouts/mutes +-- throughout drain/repair. Without controls covering all writer paths, remove +-- public traffic and stop all relay/admin/background/direct writers; run only +-- this operator connection. Reopen only on new code after zero remaining and +-- an idempotent rerun. Review fenced communities before reopening them. +-- +-- Each community batch commits independently, takes the same lock order as the +-- application writers, and skips a community unless it is active and has no +-- deletion timestamp. For active communities it records historical revocation +-- before clearing legacy_admission on every invite row. Skipped communities +-- retain the marker and remain fail-closed until a later run. The script is +-- idempotent; rerun it after a failure. +\set ON_ERROR_STOP on +SET lock_timeout = '5s'; +SET statement_timeout = '2min'; + +DO $$ +DECLARE + last_community UUID; + target_community UUID; + community_active BOOLEAN; + affected BIGINT; +BEGIN + LOOP + SELECT id INTO target_community + FROM communities + WHERE last_community IS NULL OR id > last_community + ORDER BY id + LIMIT 1; + EXIT WHEN NOT FOUND; + last_community := target_community; + + LOOP + PERFORM pg_advisory_xact_lock_shared( + community_deletion_lock_key(target_community) + ); + PERFORM pg_advisory_xact_lock( + hashtextextended('invite-admission:' || target_community::text, 0) + ); + SELECT deletion_state = 'active' AND deleted_at IS NULL + INTO community_active + FROM communities + WHERE id = target_community; + + IF community_active IS DISTINCT FROM TRUE THEN + COMMIT; + EXIT; + END IF; + + WITH candidates AS ( + SELECT invite.id, + ( + (invite.legacy_admission AND ( + issuer.pubkey IS NULL + OR restriction_state.has_retained_restriction + )) + OR restriction_state.has_active_ban + ) AS should_revoke + FROM relay_invites AS invite + CROSS JOIN LATERAL ( + SELECT CASE + WHEN octet_length(invite.created_by) = 64 + AND invite.created_by !~ '[^0-9a-f]' + THEN decode(invite.created_by, 'hex') + END AS pubkey + ) AS issuer + CROSS JOIN LATERAL ( + SELECT + EXISTS ( + SELECT 1 + FROM community_bans AS restriction + WHERE restriction.community_id = invite.community_id + AND ( + restriction.pubkey = issuer.pubkey + OR restriction.pubkey = ( + SELECT user_row.agent_owner_pubkey + FROM users AS user_row + WHERE user_row.community_id = invite.community_id + AND user_row.pubkey = issuer.pubkey + ) + ) + ) AS has_retained_restriction, + EXISTS ( + SELECT 1 + FROM community_bans AS restriction + WHERE restriction.community_id = invite.community_id + AND ( + restriction.pubkey = issuer.pubkey + OR restriction.pubkey = ( + SELECT user_row.agent_owner_pubkey + FROM users AS user_row + WHERE user_row.community_id = invite.community_id + AND user_row.pubkey = issuer.pubkey + ) + ) + AND restriction.banned + AND (restriction.ban_expires_at IS NULL + OR restriction.ban_expires_at > clock_timestamp()) + ) AS has_active_ban + ) AS restriction_state + WHERE invite.community_id = target_community + AND ( + invite.legacy_admission + OR (invite.revoked_at IS NULL AND restriction_state.has_active_ban) + ) + ORDER BY invite.id + LIMIT 500 + FOR UPDATE OF invite + ) + UPDATE relay_invites AS invite + SET revoked_at = CASE + WHEN candidates.should_revoke AND invite.revoked_at IS NULL + THEN clock_timestamp() + ELSE invite.revoked_at + END, + legacy_admission = false + FROM candidates + WHERE invite.community_id = target_community + AND invite.id = candidates.id; + GET DIAGNOSTICS affected = ROW_COUNT; + + COMMIT; + EXIT WHEN affected = 0; + END LOOP; + END LOOP; +END +$$; + +-- Expected result: 0. Every legacy marker in an active community has been +-- examined and cleared; every currently active ban has durably revoked its +-- invite. Markers in fenced communities remain for a later repair run. +SELECT count(*) AS unrepaired_invites +FROM relay_invites AS invite +JOIN communities AS community ON community.id = invite.community_id +CROSS JOIN LATERAL ( + SELECT CASE + WHEN octet_length(invite.created_by) = 64 + AND invite.created_by !~ '[^0-9a-f]' + THEN decode(invite.created_by, 'hex') + END AS pubkey +) AS issuer +WHERE community.deletion_state = 'active' + AND community.deleted_at IS NULL + AND ( + invite.legacy_admission + OR ( + invite.revoked_at IS NULL + AND EXISTS ( + SELECT 1 + FROM community_bans AS restriction + WHERE restriction.community_id = invite.community_id + AND ( + restriction.pubkey = issuer.pubkey + OR restriction.pubkey = ( + SELECT user_row.agent_owner_pubkey + FROM users AS user_row + WHERE user_row.community_id = invite.community_id + AND user_row.pubkey = issuer.pubkey + ) + ) + AND restriction.banned + AND (restriction.ban_expires_at IS NULL + OR restriction.ban_expires_at > clock_timestamp()) + ) + ) + ); From 96a2686bd15e628ef6c1d2d6bfd9dbbe8140a360 Mon Sep 17 00:00:00 2001 From: Brad Seiler Date: Thu, 29 Oct 2026 19:57:05 -0400 Subject: [PATCH 2/2] docs(db): align invitation repair with schema 0075 Co-authored-by: Sarnow <1894dcbafaa4df1ac607a9bf879ac0b7d77fbccb0e0f4619ecc252c8adbad8a2@seiler-test.communities.public.builderlabstaging.xyz> Signed-off-by: Brad Seiler --- TESTING.md | 4 ++-- scripts/maintenance/backfill_invite_revocations.sql | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/TESTING.md b/TESTING.md index 035fe059c58..fa3a9a3a82d 100644 --- a/TESTING.md +++ b/TESTING.md @@ -236,8 +236,8 @@ can be repaired under the same controls. Use a **fresh disposable database through migration 0074**, the complete pre-change schema including channel join roles, for upgrade testing. Do not reuse -an older database that applied the invitation migration numbered 0074 or a -different 0075 checksum; reconcile it separately without resetting a user's +an older database that applied an invitation migration under an earlier number +or a different 0075 checksum; reconcile it separately without resetting a user's database. Seed legacy links for a restricted issuer and an unrelated issuer; lift the first restriction before upgrade. Apply schema-only 0075 and confirm it rewrites no invite rows. Before diff --git a/scripts/maintenance/backfill_invite_revocations.sql b/scripts/maintenance/backfill_invite_revocations.sql index 96859417ae2..286da2578ba 100644 --- a/scripts/maintenance/backfill_invite_revocations.sql +++ b/scripts/maintenance/backfill_invite_revocations.sql @@ -1,5 +1,5 @@ -- This companion must be separately approved before core deployment and run --- in that same controlled rollout, after schema migration 0068 and draining +-- in that same controlled rollout, after schema migration 0075 and draining -- every old writer. Prevent invite mint/claim, ban lifts, and NEW timeouts/mutes -- throughout drain/repair. Without controls covering all writer paths, remove -- public traffic and stop all relay/admin/background/direct writers; run only