diff --git a/TESTING.md b/TESTING.md index dbb492fe0f..7fca31d03c 100644 --- a/TESTING.md +++ b/TESTING.md @@ -217,6 +217,38 @@ keep it closed until it is active and repaired, then recheck zero before admitti traffic. After repair, timeouts/mutes only block temporarily; active bans still revoke permanently. A rollback to code that ignores `revoked_at` requires invitation admission paused, using the same traffic/writer controls if needed. +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 0067**, the complete +pre-change schema including channel join roles, for upgrade testing. Do not reuse +an older database that applied the invitation migration numbered 0067 or a +different 0068 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 0068 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 7d997a0791..d1d462dde3 100644 --- a/crates/buzz-db/TESTING.md +++ b/crates/buzz-db/TESTING.md @@ -67,5 +67,6 @@ transient disconnect races before reporting a warning. Invite upgrade tests start from the complete schema through migration 0067, including channel join roles. They exercise the invitation schema expansion (migration 0068) 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 54da3cf174..3f38b74ae3 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 2b48f1471b..7667352f01 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> { @@ -334,7 +336,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( @@ -377,6 +384,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 \ @@ -416,6 +487,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 @@ -500,7 +612,7 @@ async fn apply_pgschema_to_existing_database(pool: &PgPool) { #[tokio::test] #[ignore = "requires Postgres"] -async fn migration_schema_0068_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_eq!( @@ -533,6 +645,14 @@ async fn migration_schema_0068_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, @@ -551,6 +671,17 @@ async fn migration_schema_0068_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] @@ -559,6 +690,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]; @@ -619,7 +751,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" ); } @@ -667,4 +810,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 0000000000..96859417ae --- /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()) + ) + ) + );