diff --git a/crates/buzz-db/src/runtime/tests/reader_pool_postgres_tests.rs b/crates/buzz-db/src/runtime/tests/reader_pool_postgres_tests.rs index 41e84cd2d4..d3ac5877ff 100644 --- a/crates/buzz-db/src/runtime/tests/reader_pool_postgres_tests.rs +++ b/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; } diff --git a/crates/buzz-relay/src/api/bridge/thread_window/postgres_tests.rs b/crates/buzz-relay/src/api/bridge/thread_window/postgres_tests.rs index 651805587e..35318b0c61 100644 --- a/crates/buzz-relay/src/api/bridge/thread_window/postgres_tests.rs +++ b/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() } diff --git a/crates/buzz-relay/src/api/bridge/thread_window/postgres_tests/failure_postgres_tests.rs b/crates/buzz-relay/src/api/bridge/thread_window/postgres_tests/failure_postgres_tests.rs index d795917cb6..3f67c7d0ea 100644 --- a/crates/buzz-relay/src/api/bridge/thread_window/postgres_tests/failure_postgres_tests.rs +++ b/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, Option>); +// 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, Arc); impl Layer 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( request: F, - proxy: Option>, + proxy: Arc, ) -> std::pin::Pin>> { 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( 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, Arc) { + 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!( diff --git a/crates/buzz-test-client/tests/e2e_relay.rs b/crates/buzz-test-client/tests/e2e_relay.rs index 39d7d74a6d..6a214860a7 100644 --- a/crates/buzz-test-client/tests/e2e_relay.rs +++ b/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,6 +333,7 @@ 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"); @@ -351,28 +341,6 @@ async fn test_invite_claim_auto_joins_default_channel() { 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");