mirror of
https://github.com/block/buzz.git
synced 2026-10-28 04:51:40 +00:00
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 <noreply@openai.com> Signed-off-by: Brad Seiler <seiler@squareup.com> Signed-off-by: seiler <seiler@squareup.com>
This commit is contained in:
32
TESTING.md
32
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
|
||||
|
||||
@@ -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`.
|
||||
|
||||
@@ -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<chrono::DateTime<chrono::Utc>> =
|
||||
sqlx::query_scalar("SELECT revoked_at FROM relay_invites WHERE community_id=$1 AND id=$2")
|
||||
.bind(community.as_uuid())
|
||||
|
||||
@@ -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<LegacyInviteCase>) {
|
||||
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<DateTime<Utc>>, 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"
|
||||
);
|
||||
}
|
||||
|
||||
168
scripts/maintenance/backfill_invite_revocations.sql
Normal file
168
scripts/maintenance/backfill_invite_revocations.sql
Normal file
@@ -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())
|
||||
)
|
||||
)
|
||||
);
|
||||
Reference in New Issue
Block a user