test(relay,db): remove three timing races from Postgres and E2E tests

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

@@ -144,10 +144,13 @@ async fn reader_boot_ping_warms_pool_to_max_in_background() {
 /// At the production acquire budget, a reader that admits only some dials
 /// ends the warm pass bounded and partial, with no re-dialing. The pool is
 /// built by the production constructor, so each warm acquire, dial
-/// included, gets only [`Db::READER_ACQUIRE_TIMEOUT`]. The proxy relays the
-/// first 3 sessions and stalls every later dial mid-handshake: the first
-/// wave of 5 holds 3, the second wave's 5 dials all miss, and that wave's
-/// lack of progress ends the pass after exactly 10 dials.
+/// included, gets only [`Db::READER_ACQUIRE_TIMEOUT`]. The boot ping first
+/// fills the warm floor of 3, dialed outside that budget, and the proxy then
+/// stalls every later dial mid-handshake: the first wave of 5 holds the 3
+/// idle connections and its 2 dials miss, the second wave's 5 dials all
+/// miss, and that wave's lack of progress ends the pass after exactly 7
+/// dials. The pass needs no dial to beat the budget: on a loaded host every
+/// dial can miss it, which would end the pass after one empty wave.
 #[tokio::test]
 #[ignore = "requires Postgres"]
 async fn reader_warm_at_production_budget_ends_partial_when_dials_stall() {
@@ -156,7 +159,6 @@ async fn reader_warm_at_production_budget_ends_partial_when_dials_stall() {
     let writer_options = writer.connect_options();
     let (host, port) = (writer_options.get_host(), writer_options.get_port());
     let proxy = crate::test_support::PausingProxy::spawn(host.to_owned(), port).await;
-    proxy.stall_from(3);
     let scratch_url = scratch.url();
     let proxied_url = scratch_url.replacen(
         &format!("{host}:{port}"),
@@ -169,6 +171,13 @@ async fn reader_warm_at_production_budget_ends_partial_when_dials_stall() {
     );
     let read_pool = Db::connect_read_pool(&DbConfig::default(), &proxied_url, max)
         .expect("production reader pool");
+    assert!(
+        Db::read_pool_boot_ping_once(read_pool.clone(), Default::default()).await,
+        "boot ping must reach the reader"
+    );
+    await_idle_floor(&read_pool, 3).await;
+    let relayed = proxy.sessions();
+    proxy.stall_from(relayed);
 
     let started = Instant::now();
     let report = Db::warm_read_pool(&read_pool).await;
@@ -179,14 +188,19 @@ async fn reader_warm_at_production_budget_ends_partial_when_dials_stall() {
         (max, 3, 7),
         "one partial wave, then one wave that opens nothing"
     );
-    assert_eq!(proxy.sessions(), 10, "the pass must stop dialing");
     assert!(
         elapsed < Duration::from_secs(2),
         "two waves at the production budget must end promptly: {elapsed:?}"
     );
     await_idle_floor(&read_pool, 3).await;
     tokio::time::sleep(Duration::from_millis(300)).await;
-    assert_eq!(proxy.sessions(), 10, "nothing re-dials after the pass");
+    // Counted after the settle: the proxy accepts on this runtime, so a dial
+    // that missed its budget can still be queued for accept when the pass ends.
+    assert_eq!(
+        proxy.sessions() - relayed,
+        7,
+        "the pass must stop dialing, and nothing re-dials after it"
+    );
 
     read_pool.close().await;
 }

crates/buzz-relay/src/api/bridge/thread_window/postgres_tests.rs

@@ -180,8 +180,13 @@ impl Fixture {
     }
 }
 
-async fn production_db(mut config: buzz_db::DbConfig) -> buzz_db::Db {
-    config.database_url = crate::test_support::database_url();
+async fn production_db(config: buzz_db::DbConfig) -> buzz_db::Db {
+    production_db_at(crate::test_support::database_url(), config).await
+}
+
+/// [`production_db`] at another URL, such as a proxy in front of the test database.
+async fn production_db_at(database_url: String, mut config: buzz_db::DbConfig) -> buzz_db::Db {
+    config.database_url = database_url;
     config.max_connections = 5;
     buzz_db::Db::new(&config).await.unwrap()
 }

crates/buzz-relay/src/api/bridge/thread_window/postgres_tests/failure_postgres_tests.rs

@@ -10,10 +10,11 @@ use tracing_subscriber::{layer::Context, prelude::*, registry::LookupSpan, Layer
 
 // Pause the HTTP future after the first real auxiliary page completes. This
 // observes a production span rather than replacing the database or closure.
-// With a proxy, the first page's span close also pauses it. That runs inside
-// the request's poll, before the second page's query is written, so the second
-// page stays pending however fast the database answers.
-struct AuxPages(Arc<AtomicUsize>, Option<Arc<PausingProxy>>);
+// The first page's span close also pauses the proxy the request's database
+// connects through. That runs inside the request's poll, before the second
+// page's query is written, so the second page stays pending however fast the
+// database answers.
+struct AuxPages(Arc<AtomicUsize>, Arc<PausingProxy>);
 impl<S> Layer<S> for AuxPages
 where
     S: tracing::Subscriber + for<'a> LookupSpan<'a>,
@@ -22,16 +23,14 @@ where
         if ctx.span(&id).unwrap().metadata().name() == "thread_window_aux"
             && self.0.fetch_add(1, Ordering::SeqCst) == 0
         {
-            if let Some(proxy) = &self.1 {
-                proxy.set_paused(true);
-            }
+            self.1.set_paused(true);
         }
     }
 }
 
 async fn pause_after_aux_page<F: Future>(
     request: F,
-    proxy: Option<Arc<PausingProxy>>,
+    proxy: Arc<PausingProxy>,
 ) -> std::pin::Pin<Box<impl Future<Output = F::Output>>> {
     let pages = Arc::new(AtomicUsize::new(0));
     let subscriber = tracing_subscriber::registry().with(AuxPages(pages.clone(), proxy));
@@ -60,6 +59,23 @@ async fn pause_after_aux_page<F: Future>(
     request
 }
 
+// State for the paused request alone, whose production writer connects through
+// a proxy so `AuxPages` can hold its second writer page. The test still reads
+// through the fixture's direct `state.db` while the request is held; through
+// the proxy, that read would deadlock.
+async fn proxied_state(f: &Fixture) -> (Arc<AppState>, Arc<PausingProxy>) {
+    let database_url = crate::test_support::database_url();
+    let options: sqlx::postgres::PgConnectOptions = database_url.parse().unwrap();
+    let proxy =
+        Arc::new(PausingProxy::spawn(options.get_host().to_owned(), options.get_port()).await);
+    let mut url = url::Url::parse(&database_url).unwrap();
+    url.set_host(Some("127.0.0.1")).unwrap();
+    url.set_port(Some(proxy.port)).unwrap();
+    let mut state = (*f.state).clone();
+    state.db = production_db_at(url.into(), buzz_db::DbConfig::default()).await;
+    (Arc::new(state), proxy)
+}
+
 #[tokio::test]
 #[ignore = "requires Postgres"]
 async fn thread_window_bridge_restarts_both_aux_hops_after_replica_failure() {
@@ -92,11 +108,8 @@ async fn thread_window_bridge_restarts_both_aux_hops_after_replica_failure() {
     let mut filter = f.filter();
     filter["until"] = json!(f.root.created_at.as_secs() + 1);
     filter["before_id"] = json!("00".repeat(32));
-    let request = pause_after_aux_page(
-        f.post(&f.keys, "/query", json!([filter])),
-        Some(proxy.clone()),
-    )
-    .await;
+    let request =
+        pause_after_aux_page(f.post(&f.keys, "/query", json!([filter])), proxy.clone()).await;
     let held: i64 = sqlx::query_scalar("SELECT count(*) FROM pg_stat_activity WHERE application_name=$1 AND xact_start IS NOT NULL")
         .bind(&reader_name).fetch_one(&f.pool).await.unwrap();
     assert_eq!(held, 1, "must actually hold the proved replica transaction");
@@ -218,7 +231,12 @@ async fn assert_aux_access_change(grant: bool) {
         "pre-transition visibility control"
     );
 
-    let request = pause_after_aux_page(f.post(&f.keys, "/query", json!([filter])), None).await;
+    let (state, proxy) = proxied_state(&f).await;
+    let request = pause_after_aux_page(
+        post(state, &f.host, &f.keys, "/query", json!([filter])),
+        proxy.clone(),
+    )
+    .await;
     set_access(&f, aux_channel, grant).await;
     let current = f
         .state
@@ -231,6 +249,7 @@ async fn assert_aux_access_change(grant: bool) {
         "requested channel stays authorized"
     );
     assert_eq!(current.contains(&aux_channel), grant);
+    proxy.set_paused(false);
     let (status, interrupted) = request.await;
     assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE, "{interrupted}");
     assert_eq!(
@@ -355,8 +374,14 @@ async fn thread_window_batch_discards_earlier_windows_after_revocation() {
         let (status, before) = f.post(&f.keys, "/query", batch.clone()).await;
         assert_eq!(status, StatusCode::OK, "{before}");
         assert_eq!(ids(&before, Some(39007)).len(), 2);
-        let request = pause_after_aux_page(f.post(&f.keys, "/query", batch.clone()), None).await;
+        let (state, proxy) = proxied_state(&f).await;
+        let request = pause_after_aux_page(
+            post(state, &f.host, &f.keys, "/query", batch.clone()),
+            proxy.clone(),
+        )
+        .await;
         set_access(&f, f.channel, false).await;
+        proxy.set_paused(false);
         let (status, body) = request.await;
         assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE, "{body}");
         assert_eq!(

crates/buzz-test-client/tests/e2e_relay.rs

@@ -303,8 +303,14 @@ async fn test_invite_mint_and_claim_admits_new_pubkey() {
 }
 
 /// Claiming an invite joins the new member to the community's default
-/// channels: the relay owner flags one, the claimer receives its kind:44100
-/// and is on its kind:39002 roster, and a repeat claim is a no-op.
+/// channels: the relay owner flags one, the claim puts the claimer on its
+/// kind:39002 roster and stores one relay-signed kind:44100 for them, and a
+/// repeat claim is a no-op.
+///
+/// The claimer opens no WebSocket until the claim has answered. The e2e
+/// relay runs without `BUZZ_REQUIRE_RELAY_MEMBERSHIP`, so a NIP-42 admission
+/// would join them to the default channels itself and leave the claim
+/// nothing to join.
 #[tokio::test]
 #[ignore]
 async fn test_invite_claim_auto_joins_default_channel() {
@@ -319,23 +325,6 @@ async fn test_invite_claim_auto_joins_default_channel() {
         (true, String::new())
     );
 
-    let mut joiner_client = BuzzTestClient::connect(&url, &joiner)
-        .await
-        .expect("connect joiner");
-    let sid = sub_id("auto-join-notif");
-    let filter = Filter::new()
-        .kind(Kind::Custom(44100))
-        .custom_tag(SingleLetterTag::lowercase(Alphabet::P), joiner_hex.as_str())
-        .since(nostr::Timestamp::now() - 5u64);
-    joiner_client
-        .subscribe(&sid, vec![filter])
-        .await
-        .expect("subscribe to membership notifications");
-    joiner_client
-        .collect_until_eose(&sid, Duration::from_secs(5))
-        .await
-        .expect("EOSE for membership sub");
-
     let mint_response = invite_post(&relay_owner, "/api/invites", "{}").await;
     assert_eq!(mint_response.status(), reqwest::StatusCode::OK);
     let mint_json: serde_json::Value = mint_response.json().await.expect("mint JSON");
@@ -344,35 +333,14 @@ async fn test_invite_claim_auto_joins_default_channel() {
         .and_then(serde_json::Value::as_str)
         .expect("mint response includes code");
     let claim_body = serde_json::json!({ "code": code }).to_string();
+    let claimed_since = nostr::Timestamp::now() - 5u64;
     let claim_response = invite_post(&joiner, "/api/invites/claim", &claim_body).await;
     assert_eq!(claim_response.status(), reqwest::StatusCode::OK);
     let claim_json: serde_json::Value = claim_response.json().await.expect("claim JSON");
     assert_eq!(
         claim_json.get("status").and_then(serde_json::Value::as_str),
         Some("joined")
     );
-
-    // Other tests may leave their own default channels flagged on this
-    // shared relay; wait for the notification about ours.
-    let deadline = tokio::time::Instant::now() + Duration::from_secs(5);
-    loop {
-        let remaining = deadline.saturating_duration_since(tokio::time::Instant::now());
-        let msg = joiner_client
-            .recv_event(remaining)
-            .await
-            .expect("recv kind:44100 notification for the default channel");
-        if let RelayMessage::Event { event, .. } = msg {
-            let channel = event
-                .tags
-                .iter()
-                .find(|tag| tag.as_slice().first().map(String::as_str) == Some("h"))
-                .and_then(|tag| tag.as_slice().get(1).cloned());
-            if channel.as_deref() == Some(channel_id.as_str()) {
-                break;
-            }
-        }
-    }
-    joiner_client.disconnect().await.ok();
     assert_eq!(
         member_role(&url, &relay_owner, &channel_id, &joiner_hex)
             .await
@@ -381,6 +349,53 @@ async fn test_invite_claim_auto_joins_default_channel() {
         "the relay-signed kind:39002 roster lists the claimer"
     );
 
+    // The claim stores its notice before it answers, so the notice comes
+    // back as history, before EOSE.
+    let mut joiner_client = BuzzTestClient::connect(&url, &joiner)
+        .await
+        .expect("connect joiner");
+    let sid = sub_id("auto-join-notif");
+    let filter = Filter::new()
+        .kind(Kind::Custom(44100))
+        .custom_tag(SingleLetterTag::lowercase(Alphabet::P), joiner_hex.as_str())
+        .since(claimed_since);
+    joiner_client
+        .subscribe(&sid, vec![filter])
+        .await
+        .expect("subscribe to membership notifications");
+    let notices = joiner_client
+        .collect_until_eose(&sid, Duration::from_secs(5))
+        .await
+        .expect("EOSE for membership sub");
+    joiner_client.disconnect().await.ok();
+    // Other tests may leave their own default channels flagged on this
+    // shared relay; count only the notices about ours.
+    let ours: Vec<_> = notices
+        .iter()
+        .filter(|event| {
+            event.tags.iter().any(|tag| {
+                let tag = tag.as_slice();
+                tag.len() >= 2 && tag[0] == "h" && tag[1] == channel_id
+            })
+        })
+        .collect();
+    assert_eq!(
+        ours.len(),
+        1,
+        "one notice for the default channel: {notices:?}"
+    );
+    let info: serde_json::Value = reqwest::get(format!("{}/info", relay_http_url()))
+        .await
+        .expect("GET /info")
+        .json()
+        .await
+        .expect("NIP-11 JSON");
+    assert_eq!(
+        info.get("self").and_then(serde_json::Value::as_str),
+        Some(ours[0].pubkey.to_hex().as_str()),
+        "the relay signs the notice"
+    );
+
     let repeat_response = invite_post(&joiner, "/api/invites/claim", &claim_body).await;
     assert_eq!(repeat_response.status(), reqwest::StatusCode::OK);
     let repeat_json: serde_json::Value = repeat_response.json().await.expect("repeat claim JSON");