fix(db): admit fleet sweeps per tenant instead of in-SQL fence predicate

crates/buzz-db/src/error.rs

@@ -99,7 +99,42 @@ impl DbError {
             DbError::Sqlx(sqlx::Error::Database(db)) if db.code().as_deref() == Some("57014")
         )
     }
+
+    /// Whether the failure is about reaching the database at all (pool
+    /// acquire timeout, closed pool, or socket I/O) rather than about the
+    /// statement or its tenant. Every later query on the pool is expected to
+    /// fail the same way.
+    pub fn is_connection_unavailable(&self) -> bool {
+        matches!(
+            self,
+            DbError::Sqlx(sqlx::Error::PoolTimedOut | sqlx::Error::PoolClosed | sqlx::Error::Io(_))
+        )
+    }
 }
 
 /// Convenience alias for `Result<T, DbError>`.
 pub type Result<T> = std::result::Result<T, DbError>;
+
+#[cfg(test)]
+mod tests {
+    use super::DbError;
+
+    #[test]
+    fn connection_unavailable_covers_only_pool_and_io_failures() {
+        for error in [
+            sqlx::Error::PoolTimedOut,
+            sqlx::Error::PoolClosed,
+            sqlx::Error::Io(std::io::Error::from(std::io::ErrorKind::ConnectionReset)),
+        ] {
+            assert!(DbError::Sqlx(error).is_connection_unavailable());
+        }
+        for error in [
+            DbError::Sqlx(sqlx::Error::RowNotFound),
+            DbError::Sqlx(sqlx::Error::Protocol("bad frame".into())),
+            DbError::AccessDenied("write-fenced".into()),
+            DbError::InvalidData("bad row".into()),
+        ] {
+            assert!(!error.is_connection_unavailable(), "{error:?}");
+        }
+    }
+}

crates/buzz-db/src/lib.rs

@@ -60,9 +60,9 @@ pub const DB_CONNECTION_TERMINALS: [(DbPoolRole, DbConnectionStep, DbConnectionO
 pub const DB_CONNECTION_RAW_SERIES_PER_POD: usize = runtime::CONNECTION_RAW_SERIES_PER_POD;
 pub(crate) use runtime::{
     begin_community_event_write_transaction,
-    begin_community_event_write_transaction_with_legacy_metrics, begin_community_write_transaction,
-    insert_mentions_in_transaction, observability, route_proof, ReadSessionInner, RouteDecision,
-    RoutePredicate,
+    begin_community_event_write_transaction_with_legacy_metrics, begin_community_sweep_transaction,
+    begin_community_write_transaction, insert_mentions_in_transaction, observability, route_proof,
+    sweep_admitted_communities, ReadSessionInner, RouteDecision, RoutePredicate,
 };
 pub use store::{
     admin_moderation, allowlist, api_token, archived_identities, artifact, channel,

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

@@ -202,6 +202,60 @@ pub(crate) async fn begin_community_write_transaction(
     .await
 }
 
+/// Begin a fleet sweep's tenant-local mutation, skipping communities whose
+/// deletion lifecycle no longer admits writes. All other failures propagate.
+pub(crate) async fn begin_community_sweep_transaction(
+    pool: &PgPool,
+    community: CommunityId,
+    operation: observability::WriterOperation,
+) -> Result<Option<AdmittedTx>> {
+    match begin_community_write_transaction(pool, community, operation).await {
+        Ok(tx) => Ok(Some(tx)),
+        Err(DbError::AccessDenied(_)) => Ok(None),
+        Err(error) => Err(error),
+    }
+}
+
+/// Run one admitted mutation per community for a periodic fleet sweep and
+/// return the committed results in order. Fenced communities are skipped.
+/// A tenant-local failure is logged and the sweep moves on; that tenant's
+/// rows retry next tick. A connection-level failure stops the sweep instead,
+/// because every remaining tenant would wait out the same acquire timeout.
+pub(crate) async fn sweep_admitted_communities<T, F, Fut>(
+    pool: &PgPool,
+    communities: impl IntoIterator<Item = Uuid>,
+    operation: observability::WriterOperation,
+    sweep: &'static str,
+    mut mutate: F,
+) -> Vec<T>
+where
+    F: FnMut(Uuid, AdmittedTx) -> Fut,
+    Fut: std::future::Future<Output = Result<T>>,
+{
+    let mut committed = Vec::new();
+    for id in communities {
+        let result =
+            match begin_community_sweep_transaction(pool, CommunityId::from_uuid(id), operation)
+                .await
+            {
+                Ok(None) => continue,
+                Ok(Some(tx)) => mutate(id, tx).await,
+                Err(error) => Err(error),
+            };
+        match result {
+            Ok(value) => committed.push(value),
+            Err(error) if error.is_connection_unavailable() => {
+                tracing::warn!(sweep, community_id = %id, error = %error,
+                    "Fleet sweep stopped: database unavailable; remaining tenants retry next tick");
+                break;
+            }
+            Err(error) => tracing::warn!(sweep, community_id = %id, error = %error,
+                "Fleet sweep tenant failed; retained rows retry next tick"),
+        }
+    }
+    committed
+}
+
 /// Database handle. Clone is cheap (Arc-backed pool).
 #[derive(Clone, Debug)]
 pub struct Db {

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

@@ -206,7 +206,6 @@ async fn enqueue(
         "INSERT INTO push_match_queue (community_id, event_id) \
          SELECT $1, $2 FROM events e \
          WHERE e.community_id = $1 AND e.id = $2 AND e.deleted_at IS NULL \
-           AND community_write_allowed($1) \
            AND EXISTS (SELECT 1 FROM push_leases l \
              WHERE l.community_id = $1 AND l.active AND l.endpoint_enabled \
                AND l.updated_at <= e.received_at \
@@ -244,4 +243,70 @@ mod tests {
         assert_eq!(receiver.len(), CAPACITY - 1);
         producer.join().await;
     }
+
+    mod postgres_tests {
+        use super::*;
+
+        #[tokio::test]
+        #[ignore = "requires Postgres"]
+        async fn enqueue_admits_active_tenant_and_rejects_quiescing_tenant() {
+            let (db, communities) = crate::test_support::admission_fixture().await;
+            crate::test_support::reject_legacy_sql_fence_predicate(&db.pool).await;
+            for (index, community) in communities[..2].iter().enumerate() {
+                sqlx::query(
+                    "INSERT INTO events \
+                     (community_id, id, pubkey, created_at, kind, tags, content, sig) \
+                     VALUES ($1, $2, $3, to_timestamp(1), 9, '[]', '', $4)",
+                )
+                .bind(community.as_uuid())
+                .bind(vec![index as u8; 32])
+                .bind(vec![1_u8; 32])
+                .bind(vec![2_u8; 64])
+                .execute(&db.pool)
+                .await
+                .expect("seed event");
+                sqlx::query(
+                    "INSERT INTO push_leases (community_id, author, installation_id, \
+                     source_event_id, source_created_at, generation, active, endpoint_hash, \
+                     endpoint_grant, max_class, subscriptions, expires_at, updated_at) \
+                     VALUES ($1, $2, 'test', $3, 1, 1, true, $4, 'grant', 'default', '[]', \
+                             EXTRACT(EPOCH FROM now())::bigint + 3600, now() - interval '1 hour')",
+                )
+                .bind(community.as_uuid())
+                .bind(vec![3_u8; 32])
+                .bind(vec![4_u8; 32])
+                .bind(vec![5_u8; 32])
+                .execute(&db.pool)
+                .await
+                .expect("seed lease");
+            }
+            let active = Job {
+                community: communities[0],
+                event_id: vec![0_u8; 32],
+                barrier: None,
+            };
+            let quiescing = Job {
+                community: communities[1],
+                event_id: vec![1_u8; 32],
+                barrier: None,
+            };
+            let err = enqueue(&db.pool, &quiescing, None)
+                .await
+                .expect_err("quiescing enqueue");
+            assert!(crate::test_support::is_admission_rejection(&err));
+            enqueue(&db.pool, &active, None)
+                .await
+                .expect("admitted enqueue");
+            for (community, expected) in [(communities[0], 1), (communities[1], 0)] {
+                let count: i64 = sqlx::query_scalar(
+                    "SELECT count(*) FROM push_match_queue WHERE community_id = $1",
+                )
+                .bind(community.as_uuid())
+                .fetch_one(&db.pool)
+                .await
+                .expect("read enqueue state");
+                assert_eq!(count, expected, "queue state for {community}");
+            }
+        }
+    }
 }

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

@@ -77,6 +77,33 @@ async fn setup_db() -> Db {
     Db::from_pool(pool)
 }
 
+/// A pool that becomes unavailable after the candidate read must end the
+/// sweep at the first tenant, not cost one acquire attempt per tenant.
+#[tokio::test]
+async fn fleet_sweep_stops_at_first_unavailable_connection() {
+    let pool = sqlx::postgres::PgPoolOptions::new()
+        .connect_lazy(&crate::test_support::database_url())
+        .expect("construct lazy sweep pool");
+    pool.close().await;
+    let attempted = std::cell::Cell::new(0);
+    let candidates = (0..3).map(|_| {
+        attempted.set(attempted.get() + 1);
+        Uuid::new_v4()
+    });
+
+    let committed = sweep_admitted_communities(
+        &pool,
+        candidates,
+        observability::WriterOperation::Maintenance,
+        "test_sweep",
+        |_, _| async { Ok(()) },
+    )
+    .await;
+
+    assert!(committed.is_empty());
+    assert_eq!(attempted.get(), 1, "sweep must stop after the outage");
+}
+
 #[tokio::test]
 async fn event_write_transaction_preserves_legacy_acquisition_metrics() {
     use metrics_util::debugging::{DebugValue, DebuggingRecorder};