perf(push): avoid empty push delivery claims (#8524)

Each enabled push delivery worker currently opens a claim transaction
for every active community, even when its queue is empty. Discover
eligible communities in bounded pages first, then use the existing claim
and delivery path.

No schema or index change is needed. Per-delivery optimizations (sharing
serving-write leases across a batch and combining revalidation or
membership checks), enqueue batching, and outbox pruning are out of
scope and are being investigated separately, including their deletion,
revocation, and stale-worker safety guarantees.

### Related issue

No matching existing issue or PR found.

### Validation

- 25 focused PostgreSQL tests passed, covering real-worker delivery,
eligibility filtering, pagination fairness, concurrent claims,
expired-claim recovery, and stale-worker fencing. The [full PostgreSQL
CI
suite](https://github.com/block/buzz/actions/runs/39675698081/job/119490943190)
passed all 1,090 tests, including all seven new regressions.
- Discovery was benchmarked with **43,000 communities**, existing
indexes, and 30 warm executions per scenario. Empty queues required no
community lookups; eight ready communities spread across the ID range
required eight community index lookups. Results and the
alternative-query comparison are below.
- In the separate real-worker test with 1,000 empty communities, the
original worker made 1,401 claim admissions in 1.8 seconds; the fixed
worker made zero. One and four workers made 3 and 12 discovery reads
respectively.
- In the delivery test with 65 ready communities plus 1,000 empty
communities, all 81 notifications reached an HTTP test gateway in 0.83
seconds versus 1.55 seconds with the original loop.
- Independent standards and specification reviews, plus [Codex
review](https://github.com/block/buzz/pull/8524#issuecomment-6326568721),
found no remaining blockers. Final Rust linting, the full local
unit/integration run, desktop JavaScript tests/build, web build, and
mobile analysis/tests passed. The full local desktop test run had one
unrelated timing failure in `reaps_descendant_on_timeout`; its isolated
recheck passed.
[CI](https://github.com/block/buzz/actions/runs/39675698081) passed,
including the Windows retry after an unrelated subprocess timeout.

These are isolated local measurements, not production latency guarantees
or a simulation of full production concurrency. Production delivery
validation remains part of the release process; this PR does not
re-enable push or deploy the change.

### Alternative considered

An outbox-first query selects due pending jobs and expired sending jobs
with `UNION ALL`, applies the same eligibility checks, then returns
distinct community IDs. This could avoid scanning communities with no
ready work. However, the current query's `EXISTS` clause does not force
PostgreSQL to probe the outbox once per community: the measured plans
already avoid that work when queues are empty.

Both query shapes were tested against the same **43,000-community**
dataset using existing indexes. Median database execution times over 30
warm runs:

| Scenario | Current query | Outbox-first query |
| --- | ---: | ---: |
| Empty outbox | 0.038 ms | 0.041 ms |
| No ready work, 27,336 retained jobs | 0.060 ms | 0.066 ms |
| Eight ready communities spread across the ID range, including expired
sending jobs, plus retained history | 0.109 ms | 0.126 ms |
| Eight ready communities, 128 invalid jobs, and 27,200 terminal jobs |
0.225 ms | 0.335 ms |

Both queries returned identical community IDs in every scenario. Forced
generic prepared plans produced similar results. The current query was
retained because the alternative did not reduce measured execution time
or eliminate a community scan present in the empty-queue plan. No query
rewrite or additional index was justified by these measurements.

---------

Signed-off-by: Tom Brow <tomb@block.xyz>

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

@@ -14,6 +14,9 @@ use crate::error::Result;
 use crate::Db;
 use buzz_datastore_tracing::datastore_span;
 
+mod delivery;
+pub use delivery::due_wake_communities;
+
 async fn acquire_operation_connection(
     pool: &PgPool,
     operation: crate::observability::WriterOperation,
@@ -1482,6 +1485,18 @@ impl Db {
         crate::push::enqueue_wakes(&self.pool, community, requests).await
     }
 
+    /// Discover the next bounded page of communities with eligible delivery work.
+    ///
+    /// Continue after the last returned ID until empty, then start a new sweep.
+    /// Discovery is advisory: claims and send-time validation remain authoritative.
+    #[datastore_span(name = "due_push_wake_communities", system = "postgresql")]
+    pub async fn due_push_wake_communities(
+        &self,
+        after: Option<CommunityId>,
+    ) -> Result<Vec<CommunityId>> {
+        due_wake_communities(&self.pool, after).await
+    }
+
     /// Exclusively claim due wake jobs for one community.
     #[datastore_span(name = "claim_due_push_wakes", system = "postgresql")]
     pub async fn claim_due_push_wakes(
@@ -1582,6 +1597,10 @@ impl Db {
     }
 }
 
+#[cfg(test)]
+#[path = "push/delivery_postgres_tests.rs"]
+mod delivery_postgres_tests;
+
 #[cfg(test)]
 mod postgres_tests {
     use super::*;
@@ -1638,7 +1657,7 @@ mod postgres_tests {
             .expect("sign lease event")
     }
 
-    async fn make_community(pool: &PgPool) -> CommunityId {
+    pub(super) async fn make_community(pool: &PgPool) -> CommunityId {
         let id = Uuid::new_v4();
         sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)")
             .bind(id)
@@ -1658,7 +1677,7 @@ mod postgres_tests {
         }
     }
 
-    async fn activate(
+    pub(super) async fn activate(
         pool: &PgPool,
         community: CommunityId,
         author: &[u8],
@@ -2087,7 +2106,7 @@ mod postgres_tests {
         assert_eq!(total, 3, "one row per distinct (endpoint, event) key");
     }
 
-    async fn enqueue_one(
+    pub(super) async fn enqueue_one(
         pool: &PgPool,
         community: CommunityId,
         author: &[u8],
@@ -2627,7 +2646,7 @@ mod postgres_tests {
         .expect("seed matcher row");
     }
 
-    async fn quiesce_test_community(pool: &PgPool, community: CommunityId) {
+    pub(super) async fn quiesce_test_community(pool: &PgPool, community: CommunityId) {
         let mut lifecycle = pool.begin().await.expect("begin lifecycle fixture");
         sqlx::query(
             "SELECT set_config('buzz.deletion_executor_community', $1, true), \

crates/buzz-db/src/store/push/delivery.rs

@@ -0,0 +1,56 @@
+use super::*;
+
+// Use the existing due/recovery indexes. Filtering before LIMIT prevents
+// invalid jobs from starving live work.
+const DUE_COMMUNITIES_SQL: &str = r#"
+    SELECT c.id
+    FROM communities c
+    WHERE ($1::uuid IS NULL OR c.id > $1)
+      AND c.archived_at IS NULL
+      AND c.deleted_at IS NULL
+      AND c.deletion_state = 'active'
+      AND EXISTS (
+          SELECT 1
+          FROM push_wake_outbox o
+          JOIN push_leases l
+            ON l.community_id = o.community_id
+           AND l.author = o.author
+           AND l.installation_id = o.installation_id
+           AND l.generation = o.lease_generation
+           AND l.endpoint_hash = o.endpoint_hash
+          JOIN events e
+            ON e.community_id = o.community_id
+           AND e.id = o.event_id
+           AND e.deleted_at IS NULL
+          WHERE o.community_id = c.id
+            AND o.expires_at > EXTRACT(EPOCH FROM now())::bigint
+            AND o.next_attempt_at <= now()
+            AND (o.state = 'pending' OR (o.state = 'sending' AND o.lease_until < now()))
+            AND l.active
+            AND l.endpoint_enabled
+            AND l.expires_at > EXTRACT(EPOCH FROM now())::bigint
+      )
+    ORDER BY c.id
+    LIMIT 64
+"#;
+
+/// Read up to 64 communities with eligible due wakes, ordered after `after`.
+///
+/// This read takes no community admission locks and releases its connection
+/// before the caller claims work. A full sweep advances after each page's last
+/// ID, so a busy community cannot hide quieter communities beyond the first page.
+/// An empty page ends the sweep; newly due work behind the cursor is found on
+/// the next sweep. Callers must still claim and revalidate before sending.
+pub async fn due_wake_communities(
+    pool: &PgPool,
+    after: Option<CommunityId>,
+) -> Result<Vec<CommunityId>> {
+    let mut connection =
+        acquire_operation_connection(pool, crate::observability::WriterOperation::Maintenance)
+            .await?;
+    let ids: Vec<Uuid> = sqlx::query_scalar(DUE_COMMUNITIES_SQL)
+        .bind(after.map(|community| *community.as_uuid()))
+        .fetch_all(&mut *connection)
+        .await?;
+    Ok(ids.into_iter().map(CommunityId::from_uuid).collect())
+}

crates/buzz-db/src/store/push/delivery_postgres_tests.rs

@@ -0,0 +1,297 @@
+use super::postgres_tests::{activate, enqueue_one, make_community, quiesce_test_community};
+use super::*;
+
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn empty_communities_do_not_require_wake_claims() {
+    let pool = PgPool::connect(&crate::test_support::database_url())
+        .await
+        .expect("isolated database");
+    sqlx::query(
+        "INSERT INTO communities (id, host) \
+         SELECT gen_random_uuid(), 'empty-' || n || '.example' FROM generate_series(1, 1000) n",
+    )
+    .execute(&pool)
+    .await
+    .expect("seed empty communities");
+
+    assert!(due_wake_communities(&pool, None)
+        .await
+        .expect("discover ready work")
+        .is_empty());
+}
+
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn discovery_preserves_claim_eligibility_and_recovers_expired_workers() {
+    let pool = PgPool::connect(&crate::test_support::database_url())
+        .await
+        .unwrap();
+    let author = [21; 32];
+    let mut expected = Vec::new();
+    for case in [
+        "pending",
+        "expired_worker",
+        "live_worker",
+        "future_retry",
+        "expired_wake",
+        "inactive_lease",
+        "disabled_endpoint",
+        "expired_lease",
+        "old_generation",
+        "old_endpoint",
+        "deleted_event",
+        "missing_event",
+        "archived",
+        "quiescing",
+    ] {
+        let community = make_community(&pool).await;
+        activate(&pool, community, &author, "install", &[22; 32], 1).await;
+        let id = enqueue_one(&pool, community, &author, &[23; 32], 1).await;
+        let update = match case {
+            "pending" => {
+                expected.push(community);
+                continue;
+            }
+            "expired_worker" => {
+                expected.push(community);
+                "UPDATE push_wake_outbox SET state = 'sending', lease_until = now() - interval '1 second' WHERE community_id = $1"
+            }
+            "live_worker" => "UPDATE push_wake_outbox SET state = 'sending', lease_until = now() + interval '1 hour' WHERE community_id = $1",
+            "future_retry" => "UPDATE push_wake_outbox SET next_attempt_at = now() + interval '1 hour' WHERE community_id = $1",
+            "expired_wake" => "UPDATE push_wake_outbox SET expires_at = 1 WHERE community_id = $1",
+            "inactive_lease" => "UPDATE push_leases SET active = false, endpoint_hash = NULL, endpoint_grant = NULL, max_class = NULL, subscriptions = NULL WHERE community_id = $1",
+            "disabled_endpoint" => "UPDATE push_leases SET endpoint_enabled = false WHERE community_id = $1",
+            "expired_lease" => "UPDATE push_leases SET expires_at = 1 WHERE community_id = $1",
+            "old_generation" => "UPDATE push_wake_outbox SET lease_generation = 2 WHERE community_id = $1",
+            "old_endpoint" => "UPDATE push_wake_outbox SET endpoint_hash = decode(repeat('ff', 32), 'hex') WHERE community_id = $1",
+            "deleted_event" => "UPDATE events SET deleted_at = now() WHERE community_id = $1",
+            "missing_event" => "DELETE FROM events WHERE community_id = $1",
+            "archived" => "UPDATE communities SET archived_at = now() WHERE id = $1",
+            "quiescing" => {
+                quiesce_test_community(&pool, community).await;
+                continue;
+            },
+            _ => unreachable!(),
+        };
+        sqlx::query(update)
+            .bind(community.as_uuid())
+            .execute(&pool)
+            .await
+            .unwrap();
+        if case == "expired_worker" {
+            // Discovery does not claim: the original request ID is still recovered
+            // by the unchanged claim API, fenced with a fresh claim ID.
+            let discovered = due_wake_communities(&pool, None).await.unwrap();
+            assert!(discovered.contains(&community));
+            let wake = claim_due_wakes(
+                &pool,
+                community,
+                16,
+                Utc::now() + chrono::Duration::seconds(30),
+            )
+            .await
+            .unwrap()
+            .pop()
+            .unwrap();
+            assert_eq!(wake.id, id);
+            assert!(retry_wake(
+                &pool,
+                community,
+                id,
+                wake.claim_id,
+                Utc::now() - chrono::Duration::seconds(1)
+            )
+            .await
+            .unwrap());
+        }
+    }
+    expected.sort_by_key(|community| *community.as_uuid());
+    assert_eq!(due_wake_communities(&pool, None).await.unwrap(), expected);
+
+    // A lifecycle transition after discovery is still rejected at admission.
+    let community = expected[0];
+    quiesce_test_community(&pool, community).await;
+    assert!(matches!(
+        claim_due_wakes(
+            &pool,
+            community,
+            16,
+            Utc::now() + chrono::Duration::seconds(30)
+        )
+        .await,
+        Err(crate::DbError::AccessDenied(_))
+    ));
+}
+
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn discovery_pages_past_busy_communities_without_claiming_them() {
+    // One connection also proves discovery releases its checkout before claims.
+    let pool = sqlx::postgres::PgPoolOptions::new()
+        .max_connections(1)
+        .connect(&crate::test_support::database_url())
+        .await
+        .unwrap();
+    let mut communities = Vec::new();
+    for _ in 0..70 {
+        let community = make_community(&pool).await;
+        activate(&pool, community, &[24; 32], "install", &[25; 32], 1).await;
+        enqueue_one(&pool, community, &[24; 32], &[26; 32], 1).await;
+        communities.push(community);
+    }
+    communities.sort_by_key(|community| *community.as_uuid());
+    let first = due_wake_communities(&pool, None).await.unwrap();
+    assert_eq!(first, communities[..64]);
+    // Leave the first page completely busy: later communities still get a turn.
+    let second = due_wake_communities(&pool, first.last().copied())
+        .await
+        .unwrap();
+    assert_eq!(second, communities[64..]);
+    assert!(due_wake_communities(&pool, second.last().copied())
+        .await
+        .unwrap()
+        .is_empty());
+    assert_eq!(due_wake_communities(&pool, None).await.unwrap(), first);
+    let wake = claim_due_wakes(
+        &pool,
+        second[0],
+        16,
+        Utc::now() + chrono::Duration::seconds(30),
+    )
+    .await
+    .unwrap();
+    assert_eq!(wake.len(), 1);
+}
+
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn invalid_backlog_cannot_hide_ready_communities() {
+    let pool = PgPool::connect(&crate::test_support::database_url())
+        .await
+        .unwrap();
+    sqlx::query("INSERT INTO communities (host) SELECT 'empty-' || n || '.example' FROM generate_series(1,1000) n")
+        .execute(&pool).await.unwrap();
+    let mut expected = Vec::new();
+    for n in 0..136 {
+        // Fixed ordering puts more than one page of invalid work before live work.
+        let community = CommunityId::from_uuid(Uuid::from_u128(n + 1));
+        sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)")
+            .bind(community.as_uuid())
+            .bind(format!("ready-{n}.example"))
+            .execute(&pool)
+            .await
+            .unwrap();
+        activate(&pool, community, &[31; 32], "install", &[32; 32], 1).await;
+        enqueue_one(&pool, community, &[31; 32], &[33; 32], 1).await;
+        if n < 128 {
+            sqlx::query("UPDATE push_wake_outbox SET lease_generation = 2 WHERE community_id = $1")
+                .bind(community.as_uuid())
+                .execute(&pool)
+                .await
+                .unwrap();
+        } else {
+            expected.push(community);
+        }
+    }
+    // Retained terminal jobs must not turn this into an all-history scan. These
+    // rows have no source event, which is legal after event retention/deletion.
+    sqlx::query(
+        r#"
+        INSERT INTO push_wake_outbox (community_id, author, installation_id, lease_generation,
+            endpoint_hash, event_id, class, expires_at, state)
+        SELECT community_id, author, installation_id, generation, endpoint_hash,
+               decode(lpad(to_hex(n),64,'0'),'hex'), 'default', expires_at, 'delivered'
+        FROM push_leases CROSS JOIN generate_series(1,200) n
+    "#,
+    )
+    .execute(&pool)
+    .await
+    .unwrap();
+    sqlx::raw_sql(
+        "ANALYZE communities; ANALYZE push_wake_outbox; ANALYZE push_leases; ANALYZE events;",
+    )
+    .execute(&pool)
+    .await
+    .unwrap();
+    let started = std::time::Instant::now();
+    for _ in 0..10 {
+        assert_eq!(due_wake_communities(&pool, None).await.unwrap(), expected);
+    }
+    eprintln!("10 discoveries with 1,136 communities, 128 invalid jobs, 27,200 terminal jobs, 8 ready jobs: {:?}", started.elapsed());
+}
+
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn concurrent_discovery_preserves_exclusive_claims_and_recovery_fencing() {
+    let pool = PgPool::connect(&crate::test_support::database_url())
+        .await
+        .unwrap();
+    let community = make_community(&pool).await;
+    activate(&pool, community, &[41; 32], "install", &[42; 32], 1).await;
+    for n in 0..32 {
+        enqueue_one(&pool, community, &[41; 32], &[n; 32], 1).await;
+    }
+    // Discovery is deliberately not exclusive: both workers can see the same
+    // candidate, while the existing claim transaction owns exclusivity.
+    let (a, b) = tokio::join!(
+        due_wake_communities(&pool, None),
+        due_wake_communities(&pool, None)
+    );
+    assert_eq!(a.unwrap(), vec![community]);
+    assert_eq!(b.unwrap(), vec![community]);
+    let until = Utc::now() + chrono::Duration::seconds(30);
+    let (a, b) = tokio::join!(
+        claim_due_wakes(&pool, community, 16, until),
+        claim_due_wakes(&pool, community, 16, until),
+    );
+    let mut claims = a.unwrap();
+    claims.extend(b.unwrap());
+    assert_eq!(claims.len(), 32);
+    assert_eq!(
+        claims
+            .iter()
+            .map(|wake| wake.id)
+            .collect::<std::collections::HashSet<_>>()
+            .len(),
+        32
+    );
+    assert!(due_wake_communities(&pool, None).await.unwrap().is_empty());
+
+    let old = &claims[0];
+    sqlx::query("UPDATE push_wake_outbox SET lease_until = now() - interval '1 second' WHERE community_id = $1 AND id = $2")
+        .bind(community.as_uuid()).bind(old.id).execute(&pool).await.unwrap();
+    assert_eq!(
+        due_wake_communities(&pool, None).await.unwrap(),
+        vec![community]
+    );
+    let recovered = claim_due_wakes(&pool, community, 16, until).await.unwrap();
+    assert_eq!(recovered.len(), 1);
+    assert_eq!(
+        recovered[0].id, old.id,
+        "gateway idempotency key survives recovery"
+    );
+    assert_ne!(recovered[0].claim_id, old.claim_id);
+    assert!(matches!(
+        revalidate_wake_for_send(&pool, community, old.id, old.claim_id)
+            .await
+            .unwrap(),
+        RevalidateWakeOutcome::Suppressed
+    ));
+    assert!(!complete_wake(&pool, community, old.id, old.claim_id)
+        .await
+        .unwrap());
+    assert!(
+        !retry_wake(&pool, community, old.id, old.claim_id, Utc::now())
+            .await
+            .unwrap()
+    );
+    assert!(!fail_wake(&pool, community, old.id, old.claim_id)
+        .await
+        .unwrap());
+    assert!(
+        complete_wake(&pool, community, old.id, recovered[0].claim_id)
+            .await
+            .unwrap()
+    );
+}

crates/buzz-relay/src/push_runtime.rs

@@ -422,10 +422,20 @@ async fn delivery_worker_loop(state: &AppState) {
     let mut idle_delay = Duration::from_millis(500);
     loop {
         let mut found = false;
-        match state.db.active_community_hosts().await {
+        let mut after = None;
-            Ok(communities) => {
+        loop {
+            let communities = match state.db.due_push_wake_communities(after).await {
+                Ok(communities) if communities.is_empty() => break,
+                Ok(communities) => communities,
+                Err(e) => {
+                    error!("push worker discovery failed: {e}");
+                    break;
+                }
+            };
+            // Advance even when a candidate loses its work or refuses admission.
+            // Busy early communities must not prevent later pages getting a turn.
+            after = communities.last().copied();
             for community in communities {
-                    let community = buzz_core::CommunityId::from_uuid(community.id);
                 let until = Utc::now() + TimeDelta::seconds(CLAIM_SECS);
                 match state.db.claim_due_push_wakes(community, 16, until).await {
                     Ok(wakes) => {
@@ -434,8 +444,7 @@ async fn delivery_worker_loop(state: &AppState) {
                             deliver_one(state, &http, wake).await;
                         }
                     }
-                        // The community left `active` after the host scan; the
+                    // Discovery is advisory; admission still enforces lifecycle.
-                        // next scan skips it, so this is not worth a warning.
                     Err(e @ buzz_db::DbError::AccessDenied(_)) => {
                         debug!(%community, "push wake claim refused: {e}")
                     }
@@ -443,13 +452,10 @@ async fn delivery_worker_loop(state: &AppState) {
                 }
             }
         }
-            Err(e) => error!("push worker community scan failed: {e}"),
-        }
         if found {
             idle_delay = Duration::from_millis(500);
         } else {
-            // Empty sweeps back off so an idle worker stops paying a full
+            // Preserve the existing idle cadence without per-community claims.
-            // per-community claim scan every 500ms forever.
             tokio::time::sleep(idle_delay).await;
             idle_delay = (idle_delay * 2).min(IDLE_POLL_CEILING);
         }
@@ -941,6 +947,10 @@ mod tests {
     }
 }
 
+#[cfg(test)]
+#[path = "push_runtime/delivery_postgres_tests.rs"]
+mod delivery_postgres_tests;
+
 #[cfg(test)]
 mod shutdown_tests {
     use super::run_push_task;

crates/buzz-relay/src/push_runtime/delivery_postgres_tests.rs

@@ -0,0 +1,178 @@
+use super::*;
+use metrics_util::debugging::{DebugValue, DebuggingRecorder};
+
+#[tokio::test(flavor = "current_thread")]
+#[ignore = "requires Postgres"]
+async fn idle_delivery_worker_does_not_admit_empty_communities() {
+    let url = std::env::var("BUZZ_TEST_DATABASE_URL").expect("isolated database URL");
+    let pool = sqlx::postgres::PgPoolOptions::new()
+        .max_connections(1)
+        .connect(&url)
+        .await
+        .unwrap();
+    sqlx::query(
+        "INSERT INTO communities (id, host) \
+         SELECT gen_random_uuid(), 'empty-' || n || '.example' FROM generate_series(1, 1000) n",
+    )
+    .execute(&pool)
+    .await
+    .unwrap();
+    let mut state = crate::state::tests::test_state_with_database_pool(pool).await;
+    Arc::get_mut(&mut Arc::get_mut(&mut state).unwrap().config)
+        .unwrap()
+        .push_enabled = true;
+
+    for workers in [1, 4] {
+        let recorder = DebuggingRecorder::new();
+        let snapshotter = recorder.snapshotter();
+        let _guard = metrics::set_default_local_recorder(&recorder);
+        // The real worker, real database, and real backoff: idle checks at 0, 0.5,
+        // and 1.5 seconds. Slower machines may complete fewer checks, never more.
+        assert!(tokio::time::timeout(
+            Duration::from_millis(1800),
+            futures::future::join_all((0..workers).map(|_| run_delivery_worker(state.clone()))),
+        )
+        .await
+        .is_err());
+
+        let snapshot = snapshotter.snapshot().into_vec();
+        let mut discoveries = 0;
+        let mut admissions = 0;
+        for (key, _, _, value) in snapshot {
+            let DebugValue::Counter(count) = value else {
+                continue;
+            };
+            if key.key().name() == "buzz_db_pool_acquire_attempts_total"
+                && key
+                    .key()
+                    .labels()
+                    .any(|label| label.key() == "operation" && label.value() == "maintenance")
+            {
+                discoveries += count;
+            }
+            if key.key().name() == "buzz_db_advisory_lock_acquisitions_total"
+                && key
+                    .key()
+                    .labels()
+                    .any(|label| label.key() == "lock_type" && label.value() == "deletion")
+            {
+                admissions += count;
+            }
+        }
+        assert_eq!(
+            admissions, 0,
+            "idle communities must not open claim transactions"
+        );
+        assert!(
+            (workers..=3 * workers).contains(&discoveries),
+            "idle polling stays bounded: {discoveries}"
+        );
+        eprintln!("{workers} workers, 1,000 empty communities, 1.8s: {discoveries} discovery reads, {admissions} claim admissions");
+    }
+    state.begin_shutdown();
+}
+
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn delivery_worker_reaches_later_pages_before_revisiting_busy_communities() {
+    use axum::{extract::State, routing::post, Json, Router};
+    use std::{collections::HashMap, future::IntoFuture};
+    use tokio::sync::mpsc;
+
+    async fn accept(
+        State(seen): State<mpsc::UnboundedSender<uuid::Uuid>>,
+        Json(body): Json<serde_json::Value>,
+    ) -> Json<serde_json::Value> {
+        seen.send(body["request_id"].as_str().unwrap().parse().unwrap())
+            .unwrap();
+        Json(serde_json::json!({"status": "accepted"}))
+    }
+
+    let pool = sqlx::PgPool::connect(&std::env::var("BUZZ_TEST_DATABASE_URL").unwrap())
+        .await
+        .unwrap();
+    // 1,000 empty communities and 65 ready communities cross the discovery page.
+    // The earliest ready community has more work than one claim batch can take.
+    sqlx::raw_sql(
+        r#"
+        INSERT INTO communities (id, host)
+        SELECT ('00000000-0000-0000-0000-' || lpad(n::text, 12, '0'))::uuid,
+               'delivery-' || n || '.example' FROM generate_series(1, 1065) n;
+        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)
+        SELECT id, decode(repeat('21',32),'hex'), 'install', decode(repeat('22',32),'hex'),
+               1, 1, true, decode(repeat('23',32),'hex'), 'opaque-grant', 'default', '[]',
+               4102444800 FROM communities WHERE host IN
+               (SELECT 'delivery-' || n || '.example' FROM generate_series(1,65) n);
+        INSERT INTO events (community_id, id, pubkey, created_at, kind, tags, content, sig)
+        SELECT l.community_id, decode(lpad(to_hex(n),64,'0'),'hex'), l.author,
+               now(), 9, '[]', '', decode(repeat('24',64),'hex')
+        FROM push_leases l CROSS JOIN generate_series(1,17) n
+        WHERE n = 1 OR l.community_id = '00000000-0000-0000-0000-000000000001';
+        INSERT INTO push_wake_outbox (community_id, author, installation_id, lease_generation,
+            endpoint_hash, event_id, class, expires_at)
+        SELECT l.community_id, l.author, l.installation_id, l.generation,
+               l.endpoint_hash, e.id, 'default', l.expires_at
+        FROM push_leases l JOIN events e ON e.community_id = l.community_id;
+    "#,
+    )
+    .execute(&pool)
+    .await
+    .unwrap();
+    let jobs: HashMap<uuid::Uuid, uuid::Uuid> = sqlx::query_as::<_, (uuid::Uuid, uuid::Uuid)>(
+        "SELECT id, community_id FROM push_wake_outbox",
+    )
+    .fetch_all(&pool)
+    .await
+    .unwrap()
+    .into_iter()
+    .collect();
+    let (seen, mut received) = mpsc::unbounded_channel();
+    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let address = listener.local_addr().unwrap();
+    let server = tokio::spawn(
+        axum::serve(
+            listener,
+            Router::new()
+                .route("/deliver", post(accept))
+                .with_state(seen),
+        )
+        .into_future(),
+    );
+    let mut state = crate::state::tests::test_state_with_database_pool(pool).await;
+    let config = Arc::get_mut(&mut Arc::get_mut(&mut state).unwrap().config).unwrap();
+    config.push_enabled = true;
+    config.push_gateway_delivery_url = Some(format!("http://{address}/deliver").parse().unwrap());
+    let started = Instant::now();
+    let worker = run_delivery_worker(state.clone());
+    let observe = async {
+        let mut order = Vec::new();
+        for _ in 0..81 {
+            let id = received.recv().await.unwrap();
+            order.push(jobs[&id]);
+        }
+        order
+    };
+    let order = tokio::select! {
+        _ = worker => panic!("worker exited early"),
+        result = tokio::time::timeout(Duration::from_secs(20), observe) => result.expect("deliver every community"),
+    };
+    state.begin_shutdown();
+    server.abort();
+    let hot: uuid::Uuid = "00000000-0000-0000-0000-000000000001".parse().unwrap();
+    let last: uuid::Uuid = "00000000-0000-0000-0000-000000000065".parse().unwrap();
+    assert_eq!(order[..16], [hot; 16]);
+    assert_eq!(
+        order[79], last,
+        "later discovery page progresses in the first sweep"
+    );
+    assert_eq!(
+        order[80], hot,
+        "busy community is revisited only after the full sweep"
+    );
+    eprintln!(
+        "81 deliveries across 65 ready / 1,065 total communities: {:?}",
+        started.elapsed()
+    );
+}