diff --git a/crates/buzz-db/src/store/push.rs b/crates/buzz-db/src/store/push.rs index 24ae312521..ce3a734a43 100644 --- a/crates/buzz-db/src/store/push.rs +++ b/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, + ) -> Result> { + 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), \ diff --git a/crates/buzz-db/src/store/push/delivery.rs b/crates/buzz-db/src/store/push/delivery.rs new file mode 100644 index 0000000000..eb6acae33f --- /dev/null +++ b/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, +) -> Result> { + let mut connection = + acquire_operation_connection(pool, crate::observability::WriterOperation::Maintenance) + .await?; + let ids: Vec = 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()) +} diff --git a/crates/buzz-db/src/store/push/delivery_postgres_tests.rs b/crates/buzz-db/src/store/push/delivery_postgres_tests.rs new file mode 100644 index 0000000000..eeecb888a9 --- /dev/null +++ b/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::>() + .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() + ); +} diff --git a/crates/buzz-relay/src/push_runtime.rs b/crates/buzz-relay/src/push_runtime.rs index 699a8a2634..7de9cdb9ca 100644 --- a/crates/buzz-relay/src/push_runtime.rs +++ b/crates/buzz-relay/src/push_runtime.rs @@ -422,34 +422,40 @@ 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 { - Ok(communities) => { - 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) => { - for wake in wakes { - found = true; - deliver_one(state, &http, wake).await; - } + let mut after = None; + 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 until = Utc::now() + TimeDelta::seconds(CLAIM_SECS); + match state.db.claim_due_push_wakes(community, 16, until).await { + Ok(wakes) => { + for wake in wakes { + found = true; + deliver_one(state, &http, wake).await; } - // The community left `active` after the host scan; the - // next scan skips it, so this is not worth a warning. - Err(e @ buzz_db::DbError::AccessDenied(_)) => { - debug!(%community, "push wake claim refused: {e}") - } - Err(e) => warn!(%community, "push wake claim failed: {e}"), } + // Discovery is advisory; admission still enforces lifecycle. + Err(e @ buzz_db::DbError::AccessDenied(_)) => { + debug!(%community, "push wake claim refused: {e}") + } + Err(e) => warn!(%community, "push wake claim failed: {e}"), } } - 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 - // per-community claim scan every 500ms forever. + // Preserve the existing idle cadence without per-community claims. 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; diff --git a/crates/buzz-relay/src/push_runtime/delivery_postgres_tests.rs b/crates/buzz-relay/src/push_runtime/delivery_postgres_tests.rs new file mode 100644 index 0000000000..ec40e59241 --- /dev/null +++ b/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>, + Json(body): Json, + ) -> Json { + 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 = 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() + ); +}