[buzz] Recover interrupted default-channel auto-joins

crates/buzz-db/src/runtime/migration.rs

@@ -898,7 +898,11 @@ mod postgres_tests {
         let mut migrations: Vec<_> = MIGRATOR.iter().collect();
         migrations.sort_by_key(|migration| migration.version);
 
-        assert_eq!(migrations.len(), 83);
+        assert_eq!(migrations.len(), 84);
+        assert!(migrations[83]
+            .sql
+            .as_str()
+            .contains("ADD COLUMN auto_join_pending UUID"));
         for (index, migration) in migrations.iter().enumerate() {
             assert_eq!(migration.version, (index + 1) as i64);
         }
@@ -3026,6 +3030,7 @@ mod postgres_tests {
             .expect("apply migrations for schema parity");
 
         for table in [
+            "relay_members",
             "relay_admin_actions",
             "relay_admin_outbox",
             "relay_operator_audit",

crates/buzz-db/src/runtime/tests.rs

@@ -2090,6 +2090,65 @@ async fn is_relay_member_is_bounded_routed_and_fails_closed() {
     );
 }
 
+/// The per-AUTH pending default-channel read is bounded-routed like
+/// membership; the roster writers' read stays on the writer. A different
+/// token on each pool proves which pool answered.
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn pending_default_channel_join_routed_reads_the_bounded_replica() {
+    let (writer, _writer_db) = ScratchDb::migrated(&admin_url().await, "pend_w").await;
+    let (replica, _replica_db) = ScratchDb::migrated(&admin_url().await, "pend_r").await;
+
+    let community = Uuid::new_v4();
+    let pubkey = "cc".repeat(32);
+    let (writer_token, replica_token) = (Uuid::new_v4(), Uuid::new_v4());
+    for (pool, token) in [(&writer, writer_token), (&replica, replica_token)] {
+        sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)")
+            .bind(community)
+            .bind(format!("pending-routing-{}.example", community.simple()))
+            .execute(pool)
+            .await
+            .expect("insert community");
+        sqlx::query(
+            "INSERT INTO relay_members (community_id, pubkey, role, auto_join_pending) \
+             VALUES ($1, $2, 'member', $3)",
+        )
+        .bind(community)
+        .bind(&pubkey)
+        .bind(token)
+        .execute(pool)
+        .await
+        .expect("seed pending work");
+    }
+    let cid = CommunityId::from_uuid(community);
+    let token = |pending: Option<relay_members::PendingDefaultChannelJoin>| {
+        pending.expect("pending work").token
+    };
+
+    let mut db = Db::from_pools(writer.clone(), replica.clone());
+    db.fence().force_open_for_tests(chrono::Utc::now());
+
+    // Budget unset ⇒ writer.
+    let pending = db.pending_default_channel_join_routed(cid, &pubkey).await;
+    assert_eq!(token(pending.expect("gate off")), writer_token);
+
+    // Budget set + fresh entry ⇒ replica; the writer-only read ignores it.
+    db.set_replica_read_max_age_for_tests(Some(std::time::Duration::from_secs(5)));
+    let pending = db.pending_default_channel_join_routed(cid, &pubkey).await;
+    assert_eq!(token(pending.expect("gate on")), replica_token);
+    let pending = db.pending_default_channel_join(cid, &pubkey).await;
+    assert_eq!(token(pending.expect("writer read")), writer_token);
+
+    // Entry older than the budget ⇒ fail closed to the writer.
+    db.fence().close();
+    db.fence().force_open_for_tests_at(
+        chrono::Utc::now(),
+        std::time::Instant::now() - std::time::Duration::from_secs(10),
+    );
+    let pending = db.pending_default_channel_join_routed(cid, &pubkey).await;
+    assert_eq!(token(pending.expect("entry too old")), writer_token);
+}
+
 /// Community separation across every routed seam, verified on
 /// REPLICA-SERVED reads.
 ///

crates/buzz-db/src/runtime/tests/listed_channel_name_postgres_tests.rs

@@ -505,6 +505,16 @@ async fn ledger_version(pool: &PgPool) -> i64 {
         .expect("read migration ledger")
 }
 
+async fn listed_name_migration_recorded(pool: &PgPool) -> bool {
+    sqlx::query_scalar(
+        "SELECT EXISTS (SELECT 1 FROM _sqlx_migrations WHERE version = $1 AND success)",
+    )
+    .bind(BEFORE_INDEX + 1)
+    .fetch_one(pool)
+    .await
+    .expect("read listed-name migration ledger entry")
+}
+
 async fn index_oid(pool: &PgPool) -> Option<i64> {
     sqlx::query_scalar("SELECT to_regclass('public.idx_channels_listed_name')::oid::bigint")
         .fetch_one(pool)
@@ -556,7 +566,7 @@ async fn migration_schema_listed_name_index_needs_the_renames_then_builds_inline
     migration::run_migrations(&pool)
         .await
         .expect("migrate a populated table after the renames");
-    assert_eq!(ledger_version(&pool).await, BEFORE_INDEX + 1);
+    assert!(listed_name_migration_recorded(&pool).await);
     let mut expected = vec![
         (kept, "square-ios".to_owned()),
         renamed(newer, "Square-iOS"),
@@ -633,6 +643,6 @@ async fn migration_schema_listed_name_index_validates_a_prebuild() {
         .await
         .expect("accept the prebuild without waiting for channel writers");
     writer.rollback().await.unwrap();
-    assert_eq!(ledger_version(&pool).await, BEFORE_INDEX + 1);
+    assert!(listed_name_migration_recorded(&pool).await);
     assert_eq!(index_oid(&pool).await, oid, "the prebuild is not replaced");
 }

crates/buzz-db/src/store/channel_members.rs

@@ -737,26 +737,61 @@ pub async fn put_first_member(
     community_id: CommunityId,
     channel_id: Uuid,
     pubkey: &[u8],
+) -> Result<Option<MemberRecord>> {
+    put_first_member_with_pending(pool, community_id, channel_id, pubkey, None).await
+}
+
+async fn put_first_member_with_pending(
+    pool: &PgPool,
+    community_id: CommunityId,
+    channel_id: Uuid,
+    pubkey: &[u8],
+    pending: Option<Uuid>,
 ) -> Result<Option<MemberRecord>> {
     let mut tx = crate::begin_community_write_transaction(
         pool,
         community_id,
         crate::observability::WriterOperation::EventWrite,
     )
     .await?;
+    let admitted_at = if let Some(pending) = pending {
+        // Fence removal/re-admission and completion against an older worker.
+        // Keep this lock through the channel write, on the same connection.
+        let current = sqlx::query_scalar::<_, DateTime<Utc>>(
+            "SELECT created_at FROM relay_members \
+             WHERE community_id = $1 AND pubkey = $2 AND auto_join_pending = $3 FOR SHARE",
+        )
+        .bind(community_id.as_uuid())
+        .bind(hex::encode(pubkey))
+        .bind(pending)
+        .fetch_optional(tx.conn())
+        .await?;
+        let Some(created_at) = current else {
+            tx.rollback().await?;
+            return Ok(None);
+        };
+        Some(created_at)
+    } else {
+        None
+    };
     lock_channel_for_member_put(&mut tx, channel_id).await?;
-    let seen: bool = sqlx::query_scalar(
-        "SELECT EXISTS (SELECT 1 FROM channel_members \
-         WHERE community_id = $1 AND channel_id = $2 AND pubkey = $3)",
+    let history: Option<(Option<DateTime<Utc>>,)> = sqlx::query_as(
+        "SELECT removed_at FROM channel_members \
+         WHERE community_id = $1 AND channel_id = $2 AND pubkey = $3",
     )
     .bind(community_id.as_uuid())
     .bind(channel_id)
     .bind(pubkey)
-    .fetch_one(tx.conn())
+    .fetch_optional(tx.conn())
     .await?;
-    if seen {
-        tx.rollback().await?;
-        return Ok(None);
+    if let Some((removed_at,)) = history {
+        // Ordinary first joins preserve all history. Recovery preserves a
+        // leave from this admission, but fresh roster admissions retain the
+        // original behavior of restoring earlier removed memberships.
+        if admitted_at.is_none() || removed_at.is_none() || removed_at >= admitted_at {
+            tx.rollback().await?;
+            return Ok(None);
+        }
     }
     let target = MemberPut {
         pubkey,
@@ -773,7 +808,11 @@ pub async fn put_first_member(
         }
         Err(refused) => {
             tx.rollback().await?;
-            Err(refused)
+            if pending.is_some() && matches!(refused, DbError::AccessDenied(_)) {
+                Ok(None)
+            } else {
+                Err(refused)
+            }
         }
     }
 }
@@ -2260,6 +2299,19 @@ impl Db {
         .await
     }
 
+    /// Complete one pending default-channel join, preserving leaves since admission.
+    /// A stale token or a removed relay member writes nothing.
+    #[datastore_span(name = "put_pending_default_member", system = "postgresql")]
+    pub async fn put_pending_default_member(
+        &self,
+        community: CommunityId,
+        channel: Uuid,
+        pubkey: &[u8],
+        pending: Uuid,
+    ) -> Result<Option<MemberRecord>> {
+        put_first_member_with_pending(&self.pool, community, channel, pubkey, Some(pending)).await
+    }
+
     /// Self-joins `pubkey` with the channel's `join_role` only if they have
     /// never been a member of the channel (see [`put_first_member`]).
     #[datastore_span(name = "put_first_member", system = "postgresql")]

crates/buzz-db/src/store/relay_invite.rs

@@ -611,8 +611,8 @@ pub async fn claim_relay_invite_with_owner(
     }
 
     let inserted = sqlx::query(
-        "INSERT INTO relay_members (community_id, pubkey, role, added_by) \
-         VALUES ($1, $2, 'member', 'invite') \
+        "INSERT INTO relay_members (community_id, pubkey, role, added_by, auto_join_pending) \
+         VALUES ($1, $2, 'member', 'invite', gen_random_uuid()) \
          ON CONFLICT (community_id, pubkey) DO NOTHING",
     )
     .bind(community.as_uuid())

crates/buzz-db/src/store/relay_invite/claim_owner.rs

@@ -116,8 +116,8 @@ pub async fn claim_relay_membership_with_owner(
         return Err(DbError::InviteRestricted);
     }
     let inserted = sqlx::query(
-        "INSERT INTO relay_members (community_id, pubkey, role, added_by) \
-         VALUES ($1, $2, $3, 'invite') \
+        "INSERT INTO relay_members (community_id, pubkey, role, added_by, auto_join_pending) \
+         VALUES ($1, $2, $3, 'invite', gen_random_uuid()) \
          ON CONFLICT (community_id, pubkey) DO NOTHING",
     )
     .bind(community.as_uuid())