From 7e46559f290f01a36936ac163da832a6847ec46d Mon Sep 17 00:00:00 2001 From: Luke Tornquist Date: Tue, 3 Nov 2026 15:26:11 -0500 Subject: [PATCH 1/9] fix(loadgen): bound desktop reads at stop, keep timely EOSEs, model presence and probes Follow-up to #9005 review findings: - Desktop HTTP reads race the run's stop and are abandoned there, like EOSE waits; Fetcher::finish keeps the read task's handle through the join so a cancelled client still aborts it. - Desktop thread-root lookups (thread recovery and replies) run under the desktop read timeout and stop; the basic human is unchanged. - The socket loop drains frames already in hand before judging EOSE deadlines, so an EOSE buffered behind a slow publish OK is not expired. - Both modes open the app's live-only presence route and send its silence probe after a quiet tick; an unanswered probe ends the session. - The seed log says memberships ensured, not added. - LOADGEN.md notes the read limits as simplifications and legacy as a lower bound. Co-authored-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Co-Authored-By: Claude Opus 5.5 Signed-off-by: Luke Tornquist --- .../buzz-test-client/src/loadgen/desktop.rs | 467 +++++++++++++----- crates/buzz-test-client/src/loadgen/run.rs | 41 +- crates/buzz-test-client/src/loadgen/seed.rs | 4 +- crates/buzz-test-client/src/loadgen/stats.rs | 6 +- perf/LOADGEN.md | 26 +- 5 files changed, 422 insertions(+), 122 deletions(-) diff --git a/crates/buzz-test-client/src/loadgen/desktop.rs b/crates/buzz-test-client/src/loadgen/desktop.rs index 2575d16061e..befe9b6d07e 100644 --- a/crates/buzz-test-client/src/loadgen/desktop.rs +++ b/crates/buzz-test-client/src/loadgen/desktop.rs @@ -12,6 +12,7 @@ use std::collections::{HashMap, HashSet, VecDeque}; use std::sync::Arc; use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; +use buzz_core::kind::KIND_PRESENCE_UPDATE; use buzz_ws_client::{NostrWsConnection, RelayMessage, WsClientError}; use futures_util::stream::{self, StreamExt}; use nostr::Keys; @@ -20,8 +21,8 @@ use serde_json::{json, Value}; use super::identity::{self, Role}; use super::net::http_reason; use super::run::{ - connect, observe, roots, stopped, Activity, Backoff, Directory, Rng, Seen, Shared, Step, - MAX_IDLE, + bounded, connect, observe, roots, stopped, Activity, Backoff, Directory, Rng, Seen, Shared, + Step, MAX_IDLE, }; use super::scenario::{DesktopMode, WideSplit}; use super::stats::{Op, Pending}; @@ -41,8 +42,10 @@ const REPLAY_BATCH: usize = 10; const LIVE_BATCH: usize = 2048; /// REQs awaiting EOSE at once while a connection establishes. const SETUP_CONCURRENCY: usize = 4; -/// How long a REQ may wait for its EOSE. +/// How long a REQ may wait for its EOSE, and a silence probe for any frame. const REQ_TIMEOUT: Duration = Duration::from_secs(10); +/// Frames handled in one pass before deadlines are judged. +const DRAIN_FRAMES: usize = 256; /// A resumed subscription starts this long before its cursor. const RESUME_LAG_SECS: u64 = 960; /// A first subscription starts this long before now. @@ -59,6 +62,8 @@ const MAX_ROSTER_PAGES: usize = 20; const PROFILE_CAP: usize = 1024; /// Status authors read. const STATUS_AUTHORS: usize = 100; +/// Presence authors watched: the app's cap. +const PRESENCE_AUTHORS: usize = 256; /// Channels per unread repair read. const UNREAD_CHANNELS: usize = 128; /// Events per channel window page. @@ -75,6 +80,8 @@ pub(super) enum Route { Profiles, /// The global membership route: grants, read state and reminders. Membership, + /// The global live-only presence route, which replays nothing. + Presence, /// A channel batch, which stays open. Channels, /// A replay of channels behind a live-only batch, closed at its EOSE. @@ -120,10 +127,12 @@ fn channel_filter(ids: &[String], since: u64, limit: usize) -> Value { json!({"kinds": CHANNEL_KINDS, "#h": ids, "since": since, "limit": limit}) } -/// The two global subscriptions, profiles and membership. -fn global_reqs(viewer: &str, since: u64) -> [Req; 2] { +/// The global subscriptions: profiles, membership and, while the client +/// watches anyone's presence, a live-only presence route over those +/// `presence` authors. +fn global_reqs(viewer: &str, presence: &[String], since: u64) -> Vec { let limit = REPLAY_LIMIT; - [ + let mut reqs = vec![ Req { id: "profiles".into(), filters: vec![ @@ -143,7 +152,15 @@ fn global_reqs(viewer: &str, since: u64) -> [Req; 2] { ], route: Route::Membership, }, - ] + ]; + if !presence.is_empty() { + reqs.push(Req { + id: "presence".into(), + filters: vec![json!({"kinds": [KIND_PRESENCE_UPDATE], "authors": presence})], + route: Route::Presence, + }); + } + reqs } /// Channel batch subscriptions, which stay open. Up to 10 channels share @@ -215,11 +232,12 @@ pub(super) fn replay_reqs( pub(super) fn connect_reqs( mode: DesktopMode, viewer: &str, + presence: &[String], channels: &[String], since: u64, generation: u64, ) -> Vec { - let mut reqs = Vec::from(global_reqs(viewer, since)); + let mut reqs = global_reqs(viewer, presence, since); reqs.extend(batch_reqs(mode, channels, since, REPLAY_LIMIT, generation)); reqs.extend(replay_reqs(mode, channels, since, generation)); reqs @@ -366,7 +384,7 @@ pub(super) fn establish_fetches( reads } Route::Channels => channel_recovery(windows, threads).collect(), - Route::Replay => Vec::new(), + Route::Presence | Route::Replay => Vec::new(), } } @@ -688,11 +706,16 @@ impl Reader { let base = &shared.communities[self.community].http_url; let body = filters.to_string().into_bytes(); let pending = shared.stats.begin(op); - let response = tokio::time::timeout( + let Some(response) = bounded( + shared, HTTP_TIMEOUT, shared.net.post(&self.keys, base, "/query", body, None), ) - .await; + .await + else { + pending.forget(); + return None; + }; match response { Err(_) => pending.err("read timeout"), Ok(Err(e)) => pending.err(&e.to_string()), @@ -844,7 +867,12 @@ impl Reader { let shared = &self.shared; let base = &shared.communities[self.community].http_url; let pending = shared.stats.begin(Op::DesktopNip11); - match tokio::time::timeout(HTTP_TIMEOUT, shared.net.relay_info(base)).await { + let Some(response) = bounded(shared, HTTP_TIMEOUT, shared.net.relay_info(base)).await + else { + pending.forget(); + return; + }; + match response { Err(_) => pending.err("read timeout"), Ok(Err(e)) => pending.err(&e.to_string()), Ok(Ok((status, text))) if !status.is_success() => { @@ -899,7 +927,7 @@ impl Reader { let channel = &self.channels[i % windows]; let shared = &self.shared; let community = &shared.communities[self.community]; - let found = roots(shared, community, &self.keys, channel).await; + let found = roots(shared, community, &self.keys, channel, Some(HTTP_TIMEOUT)).await; let Some(root) = found.get((i / windows) % found.len().max(1)) else { return; }; @@ -990,11 +1018,14 @@ impl Fetcher { self.task = Some(tokio::spawn(task)); } - /// Waits for the reads under way and queued. + /// Waits for the reads under way and queued. The handle stays here + /// until the task ends, so a client cancelled meanwhile still aborts it + /// on drop instead of detaching it. async fn finish(&mut self) { - if let Some(task) = self.task.take() { + if let Some(task) = &mut self.task { let _ = task.await; } + self.task = None; } } @@ -1064,6 +1095,12 @@ struct Waiting<'a> { sent: Instant, } +/// A silence probe awaiting any frame. +struct Probe<'a> { + pending: Pending<'a>, + sent: Instant, +} + /// A desktop client. Its sessions differ by [`Situation`]: the first is /// cold, a renewal follows [`renewal`], and any other reconnect is a /// [`Situation::Reconnect`]. @@ -1108,6 +1145,7 @@ async fn run_client(shared: Arc, global: usize, req_timeout: Duration) { unread_at: None, caught_up: Vec::new(), generation: 0, + probes: 0, first_renewal: true, last_message: None, split: scenario.wide_split(Role::Human, global), @@ -1154,6 +1192,8 @@ struct Client { /// The routes the last connection had caught up. caught_up: Vec, generation: u64, + /// Silence probes sent, across connections. + probes: u64, first_renewal: bool, last_message: Option, split: Option, @@ -1183,7 +1223,15 @@ impl Client { let (windows, threads) = (reader.windows, profile.threads); self.generation += 1; - let reqs = connect_reqs(mode, &reader.viewer, channels, since, self.generation); + let presence = &reader.profiles[..reader.profiles.len().min(PRESENCE_AUTHORS)]; + let reqs = connect_reqs( + mode, + &reader.viewer, + presence, + channels, + since, + self.generation, + ); let mut batches: Vec = reqs .iter() .filter(|r| r.route == Route::Channels) @@ -1198,7 +1246,13 @@ impl Client { let mut waiting: HashMap> = HashMap::new(); let mut open: HashSet = HashSet::new(); + // Whether any frame came since the last tick, the probe a silent + // tick sends, and that probe's subscription, closed by the next one. + let mut heard = false; + let mut probe: Option> = None; + let mut probe_id: Option = None; let mut activity = Activity::new(&scenario.human, &mut self.rng, self.split); + activity.root_bound = Some(HTTP_TIMEOUT); let tick_every = Duration::from_secs_f64(profile.tick_secs); let backstop = Duration::from_secs_f64(profile.backstop_secs); let next_renewal = (profile.renewal_secs > 0.0).then(|| { @@ -1240,16 +1294,115 @@ impl Client { waiting.insert(req.id, waited); } + let probe_deadline = probe.as_ref().map(|p| p.sent + self.req_timeout); + let due = [ + activity.due(), + Some(self.next_tick), + next_renewal, + probe_deadline, + ] + .into_iter() + .flatten() + .min(); + let wait = due + .map(|d| d.saturating_duration_since(Instant::now())) + .unwrap_or(MAX_IDLE) + .min(MAX_IDLE); + // The first frame waits; after it, only frames already in hand + // are handled (at most DRAIN_FRAMES), so an EOSE buffered while a + // publish waited for its OK is seen before deadlines are judged. + let mut established = Vec::new(); + let mut drained = 0; + let mut next = conn.next_event(wait).await; + let backlogged = loop { + match next { + Ok(RelayMessage::Eose { subscription_id }) => { + if let Some(w) = waiting.remove(&subscription_id) { + w.pending.ok(); + let retired = match (w.route, w.slot) { + (Route::Replay, _) => Some(subscription_id), + (_, Some(slot)) => { + open.insert(subscription_id.clone()); + let old = + std::mem::replace(&mut batches[slot], subscription_id); + open.remove(&old); + Some(old) + } + (route, None) => { + open.insert(subscription_id); + let cut = limited(w.cut, w.events); + established.extend(establishment.eose(route, cut)); + None + } + }; + if let Some(id) = retired { + if let Err(e) = close(conn, &id).await { + break 'session disconnected(e); + } + } + } + } + Ok(RelayMessage::Closed { + subscription_id, + message, + }) => { + // Live subscriptions the relay refuses end the session, + // as for the basic human; a refused replay is an error. + if let Some(w) = waiting.remove(&subscription_id) { + w.pending.err(&message); + if w.route != Route::Replay { + break 'session End::Failed; + } + } else if open.remove(&subscription_id) { + stats.err( + Op::DesktopLive, + &format!("CLOSED {subscription_id}: {message}"), + ); + break 'session End::Failed; + } + } + Ok(message) => { + // An event before EOSE is replay, which the app counts + // against its limit; only events after EOSE are delivery. + if let RelayMessage::Event { + subscription_id, .. + } = &message + { + if let Some(w) = waiting.get_mut(subscription_id) { + w.events += 1; + } else if open.contains(subscription_id) { + if let Some(target) = observe(stats, &message) { + self.last_message = Some(target); + } + } + } + } + Err(WsClientError::Timeout) => break false, + Err(e) => break 'session disconnected(e), + } + // Any frame proves the socket still reaches the relay. + heard = true; + if let Some(p) = probe.take() { + p.pending.ok(); + } + drained += 1; + if drained == DRAIN_FRAMES { + break true; + } + next = conn.next_event(Duration::ZERO).await; + }; + // A REQ past its deadline: a replay or a replacement is closed (a // replacement keeps the batch it would have replaced, and the next // tick tries again); a connection's live route stays open, and - // establishes as cut, since it may have missed history. + // establishes as cut, since it may have missed history. A backlog + // still in hand may hold an EOSE in time, so its deadlines wait + // for the next pass. let expired: Vec = waiting .iter() - .filter(|(_, w)| w.sent.elapsed() >= self.req_timeout) + .filter(|(_, w)| !backlogged && w.sent.elapsed() >= self.req_timeout) .map(|(id, _)| id.clone()) .collect(); - let mut established = Vec::new(); for id in expired { let Some(w) = waiting.remove(&id) else { continue; @@ -1268,77 +1421,12 @@ impl Client { } } - let due = [activity.due(), Some(self.next_tick), next_renewal] - .into_iter() - .flatten() - .min(); - let wait = due - .map(|d| d.saturating_duration_since(Instant::now())) - .unwrap_or(MAX_IDLE) - .min(MAX_IDLE); - match conn.next_event(wait).await { - Ok(RelayMessage::Eose { subscription_id }) => { - if let Some(w) = waiting.remove(&subscription_id) { - w.pending.ok(); - let retired = match (w.route, w.slot) { - (Route::Replay, _) => Some(subscription_id), - (_, Some(slot)) => { - open.insert(subscription_id.clone()); - let old = std::mem::replace(&mut batches[slot], subscription_id); - open.remove(&old); - Some(old) - } - (route, None) => { - open.insert(subscription_id); - let cut = limited(w.cut, w.events); - established.extend(establishment.eose(route, cut)); - None - } - }; - if let Some(id) = retired { - if let Err(e) = close(conn, &id).await { - break 'session disconnected(e); - } - } - } - } - Ok(RelayMessage::Closed { - subscription_id, - message, - }) => { - // Live subscriptions the relay refuses end the session, - // as for the basic human; a refused replay is an error. - if let Some(w) = waiting.remove(&subscription_id) { - w.pending.err(&message); - if w.route != Route::Replay { - break End::Failed; - } - } else if open.remove(&subscription_id) { - stats.err( - Op::DesktopLive, - &format!("CLOSED {subscription_id}: {message}"), - ); - break End::Failed; - } + // A probe nothing answered: the socket stopped delivering. + if !backlogged && probe_deadline.is_some_and(|d| Instant::now() >= d) { + if let Some(p) = probe.take() { + p.pending.err("probe timeout"); } - Ok(message) => { - // An event before EOSE is replay, which the app counts - // against its limit; only events after EOSE are delivery. - if let RelayMessage::Event { - subscription_id, .. - } = &message - { - if let Some(w) = waiting.get_mut(subscription_id) { - w.events += 1; - } else if open.contains(subscription_id) { - if let Some(target) = observe(stats, &message) { - self.last_message = Some(target); - } - } - } - } - Err(WsClientError::Timeout) => {} - Err(e) => break disconnected(e), + break End::Failed; } // `legacy` read everything when it connected. @@ -1358,6 +1446,8 @@ impl Client { } } + // A publish's OK is a frame too, though the step consumes it. + let before = activity.due(); let step = activity .step( &shared, @@ -1370,7 +1460,7 @@ impl Client { ) .await; match step { - Ok(Step::Continue) => {} + Ok(Step::Continue) => heard |= activity.due() != before, Ok(Step::SessionEnd) => break End::Dropped, Err(e) => break disconnected(e), } @@ -1403,6 +1493,27 @@ impl Client { batches_at = Some(now); } self.fetch(tick_fetches(due)); + // A silent minute asks the relay for an empty answer, in + // both modes; no frame by the deadline ends the session. + if !std::mem::take(&mut heard) && probe.is_none() { + if let Some(old) = probe_id.take() { + if let Err(e) = close(conn, &old).await { + break disconnected(e); + } + } + self.probes += 1; + let id = format!("probe-{}", self.probes); + let req = + json!(["REQ", id, {"kinds": [44100], "#p": [reader.viewer], "limit": 0}]); + if let Err(e) = conn.send_raw(&req).await { + break disconnected(e); + } + probe = Some(Probe { + pending: stats.begin(Op::DesktopProbe), + sent: Instant::now(), + }); + probe_id = Some(id); + } } if next_renewal.is_some_and(|t| now >= t) { break End::Renewal; @@ -1412,6 +1523,9 @@ impl Client { for (_, w) in waiting.drain() { w.pending.forget(); } + if let Some(p) = probe { + p.pending.forget(); + } self.caught_up = establishment.caught_up(); end } @@ -1448,7 +1562,7 @@ mod tests { #[test] fn up_to_ten_channels_share_one_replaying_batch_in_both_modes() { for mode in [DesktopMode::Legacy, DesktopMode::Current] { - let reqs = connect_reqs(mode, "v", &ids(5), 100, 1); + let reqs = connect_reqs(mode, "v", &[], &ids(5), 100, 1); let names: Vec<&str> = reqs.iter().map(|r| r.id.as_str()).collect(); assert_eq!(names, ["profiles", "membership", "batch-1-0"], "{mode:?}"); assert_eq!(shape(&reqs[2..]), [(5, 500, 1, false)], "{mode:?}"); @@ -1460,13 +1574,13 @@ mod tests { #[test] fn more_than_ten_channels_batch_by_ten_before_and_replay_separately_after() { - let legacy = connect_reqs(DesktopMode::Legacy, "v", &ids(25), 100, 1); + let legacy = connect_reqs(DesktopMode::Legacy, "v", &[], &ids(25), 100, 1); assert_eq!( shape(&legacy[2..]), [(10, 500, 1, false), (10, 500, 1, false), (5, 500, 1, false)] ); assert!(legacy.iter().all(|r| !r.cut_by_design())); - let current = connect_reqs(DesktopMode::Current, "v", &ids(25), 100, 1); + let current = connect_reqs(DesktopMode::Current, "v", &[], &ids(25), 100, 1); assert_eq!( shape(¤t[2..]), [ @@ -1480,7 +1594,7 @@ mod tests { let cut: Vec<_> = current.iter().map(Req::cut_by_design).collect(); assert_eq!(cut, [false, false, true, false, false, false]); // One live-only filter holds up to 2048 channels. - let wide = connect_reqs(DesktopMode::Current, "v", &ids(2049), 100, 1); + let wide = connect_reqs(DesktopMode::Current, "v", &[], &ids(2049), 100, 1); let live: Vec<_> = wide[2..] .iter() .filter(|r| r.route == Route::Channels) @@ -1712,6 +1826,32 @@ mod tests { ); } + #[test] + fn presence_is_a_live_only_global_only_while_someone_is_watched() { + for mode in [DesktopMode::Legacy, DesktopMode::Current] { + let watched = ["a".to_string(), "b".to_string()]; + let reqs = connect_reqs(mode, "v", &watched, &ids(5), 100, 1); + let routes: Vec = reqs.iter().map(|r| r.route).collect(); + assert_eq!( + routes, + [ + Route::Profiles, + Route::Membership, + Route::Presence, + Route::Channels + ], + "{mode:?}" + ); + assert_eq!( + reqs[2].filters, + [json!({"kinds": [20001], "authors": ["a", "b"]})] + ); + let alone = connect_reqs(mode, "v", &[], &ids(5), 100, 1); + assert!(alone.iter().all(|r| r.route != Route::Presence)); + } + assert!(establish_fetches(Route::Presence, false, Owed::default(), 2, 1).is_empty()); + } + #[test] fn resumed_subscriptions_start_before_the_cursor_but_not_before_the_floor() { assert_eq!(resume_since(10_000, None, 20_000), 19_700); @@ -1804,10 +1944,15 @@ mod tests { Arc::new(|_, _| (0, Duration::ZERO)) } - /// A loopback relay that challenges, accepts AUTH and every EVENT, answers - /// each REQ per `plan` and records REQs and CLOSEs. With `drop_first`, it - /// closes its first connection that long after accepting it. - async fn fake_ws_relay(plan: Plan, drop_first: Option) -> (String, Frames) { + /// A loopback relay that challenges, accepts AUTH and every EVENT (its OK + /// `ok_delay` late), answers each REQ per `plan` and records REQs and + /// CLOSEs. With `drop_first`, it closes its first connection that long + /// after accepting it. + async fn fake_ws_relay( + plan: Plan, + drop_first: Option, + ok_delay: Duration, + ) -> (String, Frames) { use futures_util::SinkExt; use tokio_tungstenite::tungstenite::Message; @@ -1853,9 +1998,17 @@ mod tests { let verb = frame[0].as_str().unwrap_or_default().to_string(); let id = frame[1].as_str().unwrap_or_default().to_string(); match verb.as_str() { - "AUTH" | "EVENT" => { + "AUTH" => { let _ = tx.send(text(json!(["OK", frame[1]["id"], true, ""]))); } + "EVENT" => { + let (tx, ok) = + (tx.clone(), json!(["OK", frame[1]["id"], true, ""])); + tokio::spawn(async move { + tokio::time::sleep(ok_delay).await; + let _ = tx.send(text(ok)); + }); + } "REQ" => { let (events, delay) = plan(index, &id); let (tx, replayed, id) = (tx.clone(), replayed.clone(), id.clone()); @@ -1993,7 +2146,9 @@ mod tests { /// `humans` desktop humans in one community of 25 channels, each in /// `joined`, with two windows and one thread open and no traffic of /// their own. - fn desktops(ws: &str, http: &str, humans: usize, joined: usize, desktop: &str) -> Scenario { + fn desktops(rig: &Rig, ws: &str, http: &str) -> Scenario { + let (humans, joined, desktop) = (rig.humans, rig.joined, &rig.desktop); + let presence = rig.presence_secs; Scenario::parse(&format!( r#" run_id = "d" @@ -2008,7 +2163,7 @@ humans = {humans} agents = 0 channels_per_member = {joined} [human] -presence_interval_secs = 3600.0 +presence_interval_secs = {presence:?} typing_per_hour = 0.0 messages_per_hour = 0.0 reactions_per_hour = 0.0 @@ -2101,6 +2256,10 @@ hold_secs = 1 fail: FailOnce, slow: Slow, req_timeout: Duration, + /// How late the relay acknowledges each EVENT. + ok_delay: Duration, + /// The humans' presence interval; their only traffic of their own. + presence_secs: f64, } /// One client, with a co-member whose profile it reads. @@ -2122,15 +2281,17 @@ hold_secs = 1 fail: &[], slow: None, req_timeout: REQ_TIMEOUT, + ok_delay: Duration::ZERO, + presence_secs: 3600.0, } } /// Runs the clients for at least `min`, until `done` holds, then stops /// them and returns what they did. Fails after 15 s without `done`. async fn drive(rig: Rig, min: Duration, done: impl Fn(&Outcome) -> bool) -> Outcome { - let (ws, frames) = fake_ws_relay(rig.plan, rig.drop_first).await; + let (ws, frames) = fake_ws_relay(rig.plan.clone(), rig.drop_first, rig.ok_delay).await; let (http, queries) = fake_api("d", rig.fail, rig.slow).await; - let scenario = desktops(&ws, &http, rig.humans, rig.joined, &rig.desktop); + let scenario = desktops(&rig, &ws, &http); let net = Net::new(&scenario).await.unwrap(); let stats = Arc::new(Stats::default()); let (stop_tx, stop) = watch::channel(false); @@ -2217,16 +2378,18 @@ hold_secs = 1 #[tokio::test] async fn a_cold_start_sends_the_modeled_requests_once() { // Legacy: per-channel batches of ten for 25 channels; current: one - // live-only batch and three replays. Both read the same over HTTP. + // live-only batch and three replays. Both open the three globals + // (each client watches the other's presence) and read the same over + // HTTP. let cases = [ - ("legacy", 25, vec![("desktop_live_eose", 2 * 5)]), + ("legacy", 25, vec![("desktop_live_eose", 2 * 6)]), ( "current", 25, - vec![("desktop_live_eose", 2 * 3), ("desktop_replay_eose", 2 * 3)], + vec![("desktop_live_eose", 2 * 4), ("desktop_replay_eose", 2 * 3)], ), - ("legacy", 5, vec![("desktop_live_eose", 2 * 3)]), - ("current", 5, vec![("desktop_live_eose", 2 * 3)]), + ("legacy", 5, vec![("desktop_live_eose", 2 * 4)]), + ("current", 5, vec![("desktop_live_eose", 2 * 4)]), ]; for (mode, joined, subs) in cases { let want = with(cold_reads(2), &subs); @@ -2237,6 +2400,7 @@ hold_secs = 1 .await; no_errors(&out); assert_eq!(out.desktop(), want, "{mode} J={joined}"); + assert!(out.saw("REQ", "presence"), "{mode} J={joined}"); assert_eq!(out.count("ws_connect_auth"), 2, "{mode} J={joined}"); } } @@ -2558,7 +2722,8 @@ hold_secs = 1 "the roster retry waited for reminders" ); assert_eq!(out.errors("desktop_roster"), 1); - assert_eq!(out.count("desktop_reminders"), 1); + // Still slow at the stop, the reminders read is abandoned there. + assert_eq!(out.ops.get("desktop_reminders"), None); } #[tokio::test] @@ -2628,11 +2793,14 @@ hold_secs = 1 ) .await; no_errors(&before); + // Silent ticks probe, as many times as the run had silent ticks. + let mut ops = before.desktop(); + ops.remove("desktop_probe"); assert_eq!( - before.desktop(), + ops, with( cold_reads(2), - &[("desktop_live_eose", 6), ("desktop_replay_eose", 6)] + &[("desktop_live_eose", 8), ("desktop_replay_eose", 6)] ) ); assert!(!before @@ -2716,4 +2884,81 @@ hold_secs = 1 assert_eq!(reads, cold); assert!(small.count("desktop_unread") <= 4); } + + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn a_read_in_flight_at_the_stop_is_abandoned_not_waited_for() { + // The roster read takes far longer than the stop may: `drive` gives + // a stopped client 5 s to return. + let mut rig = solo(5, &format!("mode = \"current\"\n{QUIET}")); + rig.slow = Some(("roster", Duration::from_secs(8))); + let out = drive(rig, Duration::ZERO, |o| { + o.queries.iter().any(|f| f["kinds"] == json!([39002])) + }) + .await; + // Abandoned, like an EOSE wait at the stop: neither a success nor + // an error. + assert_eq!(out.count("desktop_roster"), 0); + assert_eq!(out.errors("desktop_roster"), 0); + } + + #[tokio::test] + async fn a_cancelled_finish_still_aborts_the_read_task() { + let (alive, ended) = tokio::sync::oneshot::channel::<()>(); + let mut fetcher = Fetcher::default(); + fetcher.task = Some(tokio::spawn(async move { + let _alive = alive; + std::future::pending::<()>().await; + })); + let finishing = tokio::spawn(async move { fetcher.finish().await }); + tokio::task::yield_now().await; + finishing.abort(); + let _ = finishing.await; + let ended = tokio::time::timeout(Duration::from_secs(2), ended).await; + assert!(ended.expect("the read task ended").is_err()); + } + + #[tokio::test] + async fn an_eose_in_time_is_not_expired_behind_a_slow_ok() { + // Every EOSE comes 1.2 s after its REQ, well within the deadline, + // but while the first presence heartbeat (within a second of the + // connect) waits 3.5 s for its OK, past the deadline. + let mut rig = solo(5, &format!("mode = \"current\"\n{QUIET}")); + rig.presence_secs = 1.0; + rig.ok_delay = Duration::from_millis(3_500); + rig.req_timeout = Duration::from_millis(2_500); + rig.plan = Arc::new(|_, _| (0, Duration::from_millis(1_200))); + let out = drive(rig, Duration::ZERO, |o| { + o.count("publish_ephemeral_ok") >= 1 && o.count("desktop_live_eose") >= 4 + }) + .await; + assert_eq!(out.errors("desktop_live_eose"), 0, "{:?}", out.ops); + assert_eq!(out.count("desktop_live_eose"), 4, "{:?}", out.ops); + assert!(!out.saw("CLOSE", "batch-1-0"), "{:?}", out.frames); + } + + #[tokio::test] + async fn a_silent_tick_probes_and_an_unanswered_probe_reconnects() { + let mut rig = solo(5, "mode = \"current\"\ntick_secs = 1.0"); + rig.req_timeout = Duration::from_millis(500); + // The first connection never answers its probe. + rig.plan = Arc::new(|conn, id| { + let late = conn == 0 && id.starts_with("probe-"); + (0, Duration::from_secs(if late { 30 } else { 0 })) + }); + let out = drive(rig, Duration::ZERO, |o| o.count("desktop_probe") >= 1).await; + assert_eq!(out.errors("desktop_probe"), 1, "{:?}", out.ops); + let probes: Vec<_> = out + .frames + .iter() + .filter(|(_, v, id)| v == "REQ" && id.starts_with("probe-")) + .map(|(conn, _, id)| (*conn, id.as_str())) + .collect(); + assert_eq!(probes[0], (0, "probe-1"), "{:?}", out.frames); + assert!( + probes.iter().any(|(conn, _)| *conn == 1), + "{:?}", + out.frames + ); + assert_eq!(out.count("ws_connect_auth"), 2, "{:?}", out.ops); + } } diff --git a/crates/buzz-test-client/src/loadgen/run.rs b/crates/buzz-test-client/src/loadgen/run.rs index 6a35fb0dbfd..98f5ebaaccd 100644 --- a/crates/buzz-test-client/src/loadgen/run.rs +++ b/crates/buzz-test-client/src/loadgen/run.rs @@ -519,21 +519,52 @@ pub(super) async fn connect( None } +/// `read` under `limit`, abandoned once the run stops: `None` when it was, +/// as EOSE waits are abandoned at stop. A read still running then would +/// outlive the shutdown grace and fail a healthy run. +pub(super) async fn bounded( + shared: &Shared, + limit: Duration, + read: impl std::future::Future, +) -> Option> { + if stopped(shared) { + return None; + } + tokio::select! { + result = tokio::time::timeout(limit, read) => Some(result), + () = stopping(shared) => None, + } +} + /// Thread roots of `channel`, cached once found. Concurrent first callers may /// each query; a failed query is not cached, so a later reply retries it. +/// With `bound`, the lookup is [`bounded`] by it; one abandoned at stop is +/// not recorded. pub(super) async fn roots( shared: &Shared, community: &Community, reader: &Keys, channel: &str, + bound: Option, ) -> Arc> { if let Some(found) = shared.roots.lock().await.get(channel) { return found.clone(); } let owner = identity::keys(&shared.scenario.run_id, Role::Owner, community.index); let pending = shared.stats.begin(Op::HttpQuery); - let found = match find_roots(&shared.net, community, reader, &owner, channel, ROOT_PREFIX).await - { + let lookup = find_roots(&shared.net, community, reader, &owner, channel, ROOT_PREFIX); + let result = match bound { + None => lookup.await, + Some(limit) => match bounded(shared, limit, lookup).await { + Some(Ok(result)) => result, + Some(Err(_)) => Err(anyhow::anyhow!("read timeout")), + None => { + pending.forget(); + return Arc::new(Vec::new()); + } + }, + }; + let found = match result { Ok(found) => { pending.ok(); found @@ -738,6 +769,9 @@ pub(super) struct Activity { /// With a split, the latest message seen in a wide (0) and a regular (1) /// channel, so reactions follow the split too. recent: [Option; 2], + /// The bound of a reply's thread root lookup ([`roots`]); `None` for the + /// basic human. + pub(super) root_bound: Option, } /// Index into `channels` of the channel a member's next typing indicator, @@ -773,6 +807,7 @@ impl Activity { .flatten(), split, recent: [None, None], + root_bound: None, } } @@ -827,7 +862,7 @@ impl Activity { let channel = &channels[pick_channel(self.split, channels, rng)]; let mut tags = vec![tag(["h", channel])]; if rng.unit() < profile.thread_reply_fraction { - let roots = roots(shared, community, keys, channel).await; + let roots = roots(shared, community, keys, channel, self.root_bound).await; if !roots.is_empty() { tags.push(tag(["e", &roots[rng.below(roots.len())], "", "reply"])); } diff --git a/crates/buzz-test-client/src/loadgen/seed.rs b/crates/buzz-test-client/src/loadgen/seed.rs index 9742d001fde..20f12b9a062 100644 --- a/crates/buzz-test-client/src/loadgen/seed.rs +++ b/crates/buzz-test-client/src/loadgen/seed.rs @@ -43,9 +43,9 @@ pub async fn seed(scenario: &Scenario, net: &Net, stats: &Arc) -> anyhow: } create_channels(scenario, net, &community, &owner).await?; if let Some(members) = &members { - let added = add_members(scenario, net, &community, &owner, members).await?; + let ensured = add_members(scenario, net, &community, &owner, members).await?; eprintln!( - "seed: community {} memberships added ({added} people)", + "seed: community {} memberships ensured ({ensured} people)", community.index ); } diff --git a/crates/buzz-test-client/src/loadgen/stats.rs b/crates/buzz-test-client/src/loadgen/stats.rs index aacf4928b96..6da0c1e46f4 100644 --- a/crates/buzz-test-client/src/loadgen/stats.rs +++ b/crates/buzz-test-client/src/loadgen/stats.rs @@ -27,6 +27,8 @@ pub enum Op { DesktopLive, /// Desktop client: EOSE of a closing channel replay subscription. DesktopReplay, + /// Desktop client: a silence probe, answered by any frame. + DesktopProbe, /// Desktop client HTTP `/query` reads, one per request type. DesktopRoster, DesktopMetadata, @@ -44,7 +46,7 @@ pub enum Op { } impl Op { - pub const ALL: [Op; 27] = [ + pub const ALL: [Op; 28] = [ Op::WsConnect, Op::History, Op::PublishEphemeral, @@ -59,6 +61,7 @@ impl Op { Op::DesktopNip11, Op::DesktopLive, Op::DesktopReplay, + Op::DesktopProbe, Op::DesktopRoster, Op::DesktopMetadata, Op::DesktopSidebar, @@ -90,6 +93,7 @@ impl Op { Op::DesktopNip11 => "desktop_nip11", Op::DesktopLive => "desktop_live_eose", Op::DesktopReplay => "desktop_replay_eose", + Op::DesktopProbe => "desktop_probe", Op::DesktopRoster => "desktop_roster", Op::DesktopMetadata => "desktop_metadata", Op::DesktopSidebar => "desktop_sidebar", diff --git a/perf/LOADGEN.md b/perf/LOADGEN.md index d0a9df95cb7..f5fcee7bf6d 100644 --- a/perf/LOADGEN.md +++ b/perf/LOADGEN.md @@ -129,14 +129,17 @@ the report. On a **cold start** it: 1. reads the relay's NIP-11 document (`desktop_nip11`); -2. opens its profile and membership subscriptions, and its channel - subscriptions (`desktop_live_eose`). With up to ten channels both modes +2. opens its profile and membership subscriptions, a live-only presence + subscription over up to 256 co-members, and its channel subscriptions + (`desktop_live_eose`). With up to ten channels both modes open one per-channel batch that replays the newest 500 events. With more, `legacy` opens one such batch per ten channels; `current` opens one live-only subscription per 2048 channels and replays the channels ten at a time in separate subscriptions it closes at EOSE (`desktop_replay_eose`). At most four are set up at once, each with a 10 s EOSE timeout; -3. reads over HTTP, at most six at a time with a 30 s timeout: its roster +3. reads over HTTP, at most six at a time with a 30 s timeout (a + simplification: the app has no read limiter of its own, and its reads + time out after 10 s or 20 s): its roster (`desktop_roster`, paged), channel names (`desktop_metadata`), sidebar preferences (`desktop_sidebar`), co-member profiles (`desktop_profiles`) and statuses (`desktop_statuses`), the custom emoji set @@ -145,7 +148,10 @@ On a **cold start** it: (`desktop_reminders`), and for each of `windows` open channels a window catch-up (`desktop_window`) and its notices (`desktop_notices`), and for each of `threads` open threads a seeded root lookup and its first thread - page (`desktop_thread_root`, `desktop_thread_window`). A window catch-up + page (`desktop_thread_root`, `desktop_thread_window`). The root lookup is + the generator's own search for seeded roots (`http_query`), cached per + channel; a reply's lookup runs on the socket, outside the six, with the + same timeout. A window catch-up reads the head page, then follows the cursor in the relay's window bounds (kind 39006) for up to five older pages, until the bounds say history is exhausted. @@ -182,6 +188,15 @@ profile read runs again when the profile subscription next establishes. A replacement is live-only and is opened before the subscription it replaces is closed. A replacement that misses its EOSE deadline is closed, the subscription it would have replaced stays, and the next tick tries again. +In both modes, a tick with no frame from the relay since the previous tick +sends a silence probe, a live-only REQ for the viewer's membership notices +(`desktop_probe`). Any frame answers it; none within 10 s ends the session +as failed, and the client reconnects with backoff. + +`legacy` is a lower bound on the app before its fixes. That app also gave +each open channel, and each channel it watched without being a member, a +subscription of its own beside the batches, and re-sent every subscription +on each tick; the generator models only the batches. Every `renewal_secs` (0 disables it; the first is jittered) the client renews its identity and reconnects. In `legacy` that is an ordinary reconnect, which @@ -251,7 +266,8 @@ are recorded as `unfinished at stop` errors, and the cancellation fails the run whatever the limits. Keep `target.connect_timeout_secs` below 15, or a stalled connect is cancelled instead of timing out. Publishes, history fetches and HTTP requests can outlast the grace period (30 s, 30 s and 60 s deadlines), in -which case the client is cancelled and the run fails. +which case the client is cancelled and the run fails. Desktop reads, like +EOSE waits, are abandoned at the stop and not recorded. The process exits non-zero when a step aborts, a client panics, a client had to be cancelled, or `shutdown.abort_reason` is set. From 63ae433bcdcc7844b6af49c18b806cfe8cc64571 Mon Sep 17 00:00:00 2001 From: Luke Tornquist Date: Tue, 3 Nov 2026 15:43:41 -0500 Subject: [PATCH 2/9] fix(loadgen): judge probes and deferred deadlines by frames actually read Review follow-ups on the desktop session loop: - A probe's expiry reads the probe still outstanding after the drain, so a probe a drained frame answered no longer fails the session. - A reply whose root lookup the stop cut short is not published. - Only received frames count as heard: a publish reports whether the relay answered, and an OK in time answers an outstanding probe. A rescheduled targetless reaction or an OK timeout no longer does. - A backlog defers an EOSE deadline by at most one more timeout, and while it does the pass only drains, so neither sustained traffic nor a second OK wait inside a finite backlog misjudges it. Co-authored-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Co-Authored-By: Claude Opus 5.5 Signed-off-by: Luke Tornquist --- .../buzz-test-client/src/loadgen/desktop.rs | 194 +++++++++++++++--- crates/buzz-test-client/src/loadgen/run.rs | 41 ++-- perf/LOADGEN.md | 8 +- 3 files changed, 199 insertions(+), 44 deletions(-) diff --git a/crates/buzz-test-client/src/loadgen/desktop.rs b/crates/buzz-test-client/src/loadgen/desktop.rs index befe9b6d07e..4624198f620 100644 --- a/crates/buzz-test-client/src/loadgen/desktop.rs +++ b/crates/buzz-test-client/src/loadgen/desktop.rs @@ -1294,12 +1294,11 @@ impl Client { waiting.insert(req.id, waited); } - let probe_deadline = probe.as_ref().map(|p| p.sent + self.req_timeout); let due = [ activity.due(), Some(self.next_tick), next_renewal, - probe_deadline, + probe.as_ref().map(|p| p.sent + self.req_timeout), ] .into_iter() .flatten() @@ -1395,15 +1394,13 @@ impl Client { // A REQ past its deadline: a replay or a replacement is closed (a // replacement keeps the batch it would have replaced, and the next // tick tries again); a connection's live route stays open, and - // establishes as cut, since it may have missed history. A backlog - // still in hand may hold an EOSE in time, so its deadlines wait - // for the next pass. - let expired: Vec = waiting + // establishes as cut, since it may have missed history. + let timed_out: Vec = waiting .iter() - .filter(|(_, w)| !backlogged && w.sent.elapsed() >= self.req_timeout) + .filter(|(_, w)| expired(w.sent.elapsed(), self.req_timeout, backlogged)) .map(|(id, _)| id.clone()) .collect(); - for id in expired { + for id in timed_out { let Some(w) = waiting.remove(&id) else { continue; }; @@ -1421,11 +1418,10 @@ impl Client { } } - // A probe nothing answered: the socket stopped delivering. - if !backlogged && probe_deadline.is_some_and(|d| Instant::now() >= d) { - if let Some(p) = probe.take() { - p.pending.err("probe timeout"); - } + // A probe nothing answered: the socket stopped delivering. Any + // frame drained above has answered it. + if let Some(p) = probe.take_if(|p| p.sent.elapsed() >= self.req_timeout) { + p.pending.err("probe timeout"); break End::Failed; } @@ -1446,8 +1442,17 @@ impl Client { } } - // A publish's OK is a frame too, though the step consumes it. - let before = activity.due(); + // While a deadline waits on the backlog, the pass only drains: a + // publish waiting for its OK, or a tick, could outlast the + // deferral and expire an EOSE already in hand. + if backlogged + && waiting + .values() + .any(|w| w.sent.elapsed() >= self.req_timeout) + { + continue; + } + let step = activity .step( &shared, @@ -1460,7 +1465,15 @@ impl Client { ) .await; match step { - Ok(Step::Continue) => heard |= activity.due() != before, + Ok(Step::Continue) => {} + // A publish's OK is a frame too, though the step consumed it; + // one in time answers the probe. + Ok(Step::Answered) => { + heard = true; + if let Some(p) = probe.take_if(|p| p.sent.elapsed() < self.req_timeout) { + p.pending.ok(); + } + } Ok(Step::SessionEnd) => break End::Dropped, Err(e) => break disconnected(e), } @@ -1531,6 +1544,14 @@ impl Client { } } +/// Whether a REQ that has waited `waited` for its EOSE is judged expired. A +/// backlog still in hand may hold the EOSE in time, so its deadline waits for +/// the next pass, but at most one more `timeout`: traffic that keeps every +/// pass full does not defer it forever. +fn expired(waited: Duration, timeout: Duration, backlogged: bool) -> bool { + waited >= timeout && (!backlogged || waited >= timeout * 2) +} + async fn close(conn: &mut NostrWsConnection, id: &str) -> Result<(), WsClientError> { conn.send_raw(&json!(["CLOSE", id])).await } @@ -1945,9 +1966,9 @@ mod tests { } /// A loopback relay that challenges, accepts AUTH and every EVENT (its OK - /// `ok_delay` late), answers each REQ per `plan` and records REQs and - /// CLOSEs. With `drop_first`, it closes its first connection that long - /// after accepting it. + /// `ok_delay` late), answers each REQ per `plan` and records REQs, + /// CLOSEs and EVENTs (by kind). With `drop_first`, it closes its first + /// connection that long after accepting it. async fn fake_ws_relay( plan: Plan, drop_first: Option, @@ -1996,7 +2017,10 @@ mod tests { while let Some(Ok(Message::Text(raw))) = stream.next().await { let frame: Value = serde_json::from_str(&raw).unwrap(); let verb = frame[0].as_str().unwrap_or_default().to_string(); - let id = frame[1].as_str().unwrap_or_default().to_string(); + let id = match frame[1].get("kind") { + Some(kind) => kind.to_string(), + None => frame[1].as_str().unwrap_or_default().to_string(), + }; match verb.as_str() { "AUTH" => { let _ = tx.send(text(json!(["OK", frame[1]["id"], true, ""]))); @@ -2022,7 +2046,7 @@ mod tests { } _ => {} } - if verb == "REQ" || verb == "CLOSE" { + if matches!(verb.as_str(), "REQ" | "CLOSE" | "EVENT") { seen.lock().unwrap().push((index, verb, id)); } } @@ -2033,7 +2057,7 @@ mod tests { } /// Reads the fake API fails once each, by name: `roster`, `read_state`, - /// `unread`, `reminders`. + /// `unread`, `reminders`, `roots`. type FailOnce = &'static [&'static str]; /// A read the fake API answers late, by name, and how late. @@ -2100,6 +2124,8 @@ mod tests { "read_state" } else if filter["kinds"] == json!([30300]) { "reminders" + } else if filter["kinds"] == json!([9]) && filter.get("authors").is_some() { + "roots" } else { "" }; @@ -2144,11 +2170,12 @@ mod tests { } /// `humans` desktop humans in one community of 25 channels, each in - /// `joined`, with two windows and one thread open and no traffic of - /// their own. + /// `joined`, with two windows and one thread open and only the traffic + /// of their own that `rig` sets. fn desktops(rig: &Rig, ws: &str, http: &str) -> Scenario { let (humans, joined, desktop) = (rig.humans, rig.joined, &rig.desktop); - let presence = rig.presence_secs; + let (presence, messages, reactions) = + (rig.presence_secs, rig.messages_per_hour, rig.reactions_per_hour); Scenario::parse(&format!( r#" run_id = "d" @@ -2165,8 +2192,9 @@ channels_per_member = {joined} [human] presence_interval_secs = {presence:?} typing_per_hour = 0.0 -messages_per_hour = 0.0 -reactions_per_hour = 0.0 +messages_per_hour = {messages:?} +reactions_per_hour = {reactions:?} +thread_reply_fraction = 1.0 history_limit = 1 [desktop] fraction = 1.0 @@ -2258,8 +2286,11 @@ hold_secs = 1 req_timeout: Duration, /// How late the relay acknowledges each EVENT. ok_delay: Duration, - /// The humans' presence interval; their only traffic of their own. + /// The humans' presence interval. presence_secs: f64, + /// Their messages, each a thread reply, and reactions. + messages_per_hour: f64, + reactions_per_hour: f64, } /// One client, with a co-member whose profile it reads. @@ -2283,6 +2314,8 @@ hold_secs = 1 req_timeout: REQ_TIMEOUT, ok_delay: Duration::ZERO, presence_secs: 3600.0, + messages_per_hour: 0.0, + reactions_per_hour: 0.0, } } @@ -2961,4 +2994,109 @@ hold_secs = 1 ); assert_eq!(out.count("ws_connect_auth"), 2, "{:?}", out.ops); } + + #[test] + fn a_backlog_defers_a_deadline_by_at_most_one_more_timeout() { + let t = Duration::from_secs(10); + let s = Duration::from_secs; + assert!(!expired(s(9), t, false)); + assert!(expired(s(10), t, false)); + // A backlog may hold the EOSE: wait for it, but not forever. + assert!(!expired(s(10), t, true)); + assert!(!expired(s(19), t, true)); + assert!(expired(s(20), t, true)); + } + + #[tokio::test] + async fn a_long_backlog_behind_slow_oks_keeps_its_timely_eoses() { + // Every REQ replays 600 events, more than one pass drains, with its + // EOSE 1.2 s late, while heartbeats each second wait 3.5 s for OKs. + // No heartbeat may run between the backlog's passes: a second OK + // wait would outlast the deferral. + let mut rig = solo(5, &format!("mode = \"current\"\n{QUIET}")); + rig.presence_secs = 1.0; + rig.ok_delay = Duration::from_millis(3_500); + rig.req_timeout = Duration::from_millis(2_500); + rig.plan = Arc::new(|_, _| (600, Duration::from_millis(1_200))); + let out = drive(rig, Duration::ZERO, |o| { + o.count("publish_ephemeral_ok") >= 2 && o.count("desktop_live_eose") >= 4 + }) + .await; + assert_eq!(out.errors("desktop_live_eose"), 0, "{:?}", out.ops); + assert_eq!(out.count("desktop_live_eose"), 4, "{:?}", out.ops); + } + + #[tokio::test] + async fn a_probe_answered_behind_a_late_ok_does_not_reconnect() { + // Ticks every 0.25 s; a heartbeat every 1 s waits 0.5 s for its OK. + // The tick after an OK hears it, so the next one, 0.25 s before the + // next heartbeat, probes. The probe's EOSE comes 0.3 s later, while + // that heartbeat waits, and is read once its OK comes, after the + // probe's 0.7 s deadline. + let mut rig = solo(5, "mode = \"current\"\ntick_secs = 0.25"); + rig.presence_secs = 1.0; + rig.ok_delay = Duration::from_millis(500); + rig.req_timeout = Duration::from_millis(700); + rig.plan = Arc::new(|_, id| { + let late = if id.starts_with("probe-") { 300 } else { 0 }; + (0, Duration::from_millis(late)) + }); + let out = drive(rig, Duration::ZERO, |o| o.saw("REQ", "probe-3")).await; + assert_eq!(out.errors("desktop_probe"), 0, "{:?}", out.ops); + assert!(out.count("desktop_probe") >= 1, "{:?}", out.ops); + assert_eq!(out.count("ws_connect_auth"), 1, "{:?}", out.ops); + assert!(out.frames.iter().all(|(conn, ..)| *conn == 0), "{:?}", out.frames); + } + + #[tokio::test] + async fn a_publish_ok_in_time_answers_an_outstanding_probe() { + // No probe ever gets its EOSE. Ticks every 0.5 s; a heartbeat every + // second waits 0.2 s for its OK. A probe follows a silent tick, and + // the next heartbeat's OK comes well inside its 3 s deadline. + let mut rig = solo(5, "mode = \"current\"\ntick_secs = 0.5"); + rig.presence_secs = 1.0; + rig.ok_delay = Duration::from_millis(200); + rig.req_timeout = Duration::from_secs(3); + rig.plan = Arc::new(|_, id| { + let late = if id.starts_with("probe-") { 60 } else { 0 }; + (0, Duration::from_secs(late)) + }); + let out = drive(rig, Duration::ZERO, |o| o.saw("REQ", "probe-2")).await; + assert_eq!(out.errors("desktop_probe"), 0, "{:?}", out.ops); + assert!(out.count("desktop_probe") >= 1, "{:?}", out.ops); + assert_eq!(out.count("ws_connect_auth"), 1, "{:?}", out.ops); + } + + #[tokio::test] + async fn a_reaction_with_no_target_is_not_heard_as_a_frame() { + // Reactions come several times a second but have nothing to react + // to, so nothing is sent and nothing comes back: a silent tick + // still probes. + let mut rig = solo(5, "mode = \"current\"\ntick_secs = 0.5"); + rig.reactions_per_hour = 36_000.0; + let out = drive(rig, Duration::ZERO, |o| o.count("desktop_probe") >= 1).await; + assert_eq!(out.errors("desktop_probe"), 0, "{:?}", out.ops); + assert!(!out.saw("EVENT", "7"), "{:?}", out.frames); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn a_reply_whose_root_lookup_the_stop_cut_short_is_not_published() { + // The first message is a reply whose root search outlasts the stop + // may; its OK would never come in time either. + let mut rig = solo(5, &format!("mode = \"current\"\n{QUIET}")); + rig.messages_per_hour = 36_000.0; + rig.slow = Some(("roots", Duration::from_secs(8))); + rig.ok_delay = Duration::from_secs(60); + let out = drive(rig, Duration::from_millis(500), |o| { + o.queries + .iter() + .filter(|f| f["kinds"] == json!([9]) && f.get("authors").is_some()) + .count() + >= 2 + }) + .await; + assert!(!out.saw("EVENT", "9"), "{:?}", out.frames); + assert_eq!(out.count("http_query"), 0, "{:?}", out.ops); + assert_eq!(out.errors("http_query"), 0, "{:?}", out.ops); + } } diff --git a/crates/buzz-test-client/src/loadgen/run.rs b/crates/buzz-test-client/src/loadgen/run.rs index 98f5ebaaccd..d1e68ebe58f 100644 --- a/crates/buzz-test-client/src/loadgen/run.rs +++ b/crates/buzz-test-client/src/loadgen/run.rs @@ -464,25 +464,29 @@ pub(super) fn observe(stats: &Stats, message: &RelayMessage) -> Option { seen(message) } -/// Publishes over the socket and records send→OK latency. +/// Publishes over the socket and records send→OK latency. `true` when the +/// relay answered, accepting or not. async fn publish( stats: &Stats, conn: &mut NostrWsConnection, op: Op, event: Event, -) -> Result<(), WsClientError> { +) -> Result { let pending = stats.begin(op); match conn.send_event(event).await { Ok(ok) if ok.accepted => pending.ok(), Ok(ok) => pending.err(&ok.message), - Err(WsClientError::Timeout) => pending.err("OK timeout"), + Err(WsClientError::Timeout) => { + pending.err("OK timeout"); + return Ok(false); + } Err(e) => { // The caller records the broken socket. pending.forget(); return Err(e); } } - Ok(()) + Ok(true) } /// Connects with retry until connected or stopped. Each attempt is bounded @@ -722,7 +726,7 @@ async fn human(shared: Arc, global: usize) { ) .await; match outcome { - Ok(Step::Continue) => {} + Ok(Step::Continue | Step::Answered) => {} Ok(Step::SessionEnd) => break, Err(e) => { shared @@ -750,6 +754,8 @@ async fn human(shared: Arc, global: usize) { /// What [`Activity::step`] leaves the session to do. pub(super) enum Step { Continue, + /// The relay answered a publish: a frame arrived. + Answered, /// The session reached its scheduled end; reconnect. SessionEnd, } @@ -845,24 +851,29 @@ impl Activity { } } let now = Instant::now(); + let mut answered = false; if self.next_presence.is_some_and(|t| t <= now) { self.next_presence = Some(now + Duration::from_secs_f64(self.interval)); let event = EventBuilder::new(kind(KIND_PRESENCE_UPDATE), "online").sign_with_keys(keys); - send(&shared.stats, conn, Op::PublishEphemeral, event).await?; + answered = send(&shared.stats, conn, Op::PublishEphemeral, event).await?; } else if self.next_typing.is_some_and(|t| t <= now) { self.next_typing = rng.after(profile.typing_per_hour); let channel = &channels[pick_channel(self.split, channels, rng)]; let event = EventBuilder::new(kind(KIND_TYPING_INDICATOR), "") .tags([tag(["h", channel])]) .sign_with_keys(keys); - send(&shared.stats, conn, Op::PublishEphemeral, event).await?; + answered = send(&shared.stats, conn, Op::PublishEphemeral, event).await?; } else if self.next_message.is_some_and(|t| t <= now) { self.next_message = rng.after(profile.messages_per_hour); let channel = &channels[pick_channel(self.split, channels, rng)]; let mut tags = vec![tag(["h", channel])]; if rng.unit() < profile.thread_reply_fraction { let roots = roots(shared, community, keys, channel, self.root_bound).await; + // A lookup the stop cut short leaves nothing to publish. + if stopped(shared) { + return Ok(Step::Continue); + } if !roots.is_empty() { tags.push(tag(["e", &roots[rng.below(roots.len())], "", "reply"])); } @@ -874,7 +885,7 @@ impl Activity { ) .tags(tags) .sign_with_keys(keys); - send(&shared.stats, conn, Op::PublishMessage, event).await?; + answered = send(&shared.stats, conn, Op::PublishMessage, event).await?; } else if self.next_reaction.is_some_and(|t| t <= now) { self.next_reaction = rng.after(profile.reactions_per_hour); // React to each seen message at most once; the relay rejects duplicates. @@ -889,12 +900,16 @@ impl Activity { let event = EventBuilder::new(kind(KIND_REACTION), "+") .tags([tag(["h", &target.channel]), tag(["e", &target.id])]) .sign_with_keys(keys); - send(&shared.stats, conn, Op::PublishMessage, event).await?; + answered = send(&shared.stats, conn, Op::PublishMessage, event).await?; } } else if self.session_end.is_some_and(|t| t <= now) { return Ok(Step::SessionEnd); } - Ok(Step::Continue) + Ok(if answered { + Step::Answered + } else { + Step::Continue + }) } } @@ -903,12 +918,12 @@ pub(super) async fn send( conn: &mut NostrWsConnection, op: Op, event: Result, -) -> Result<(), WsClientError> { +) -> Result { match event { Ok(event) => publish(stats, conn, op, event).await, Err(e) => { stats.err(op, &format!("sign: {e}")); - Ok(()) + Ok(false) } } } @@ -1160,7 +1175,7 @@ async fn agent(shared: Arc, global: usize) { shared .stats .err(Op::PublishEphemeral, &format!("observer: {e}")); - Ok(()) + Ok(false) } }; if let Err(e) = result { diff --git a/perf/LOADGEN.md b/perf/LOADGEN.md index f5fcee7bf6d..7271cdd9e67 100644 --- a/perf/LOADGEN.md +++ b/perf/LOADGEN.md @@ -190,8 +190,9 @@ is closed. A replacement that misses its EOSE deadline is closed, the subscription it would have replaced stays, and the next tick tries again. In both modes, a tick with no frame from the relay since the previous tick sends a silence probe, a live-only REQ for the viewer's membership notices -(`desktop_probe`). Any frame answers it; none within 10 s ends the session -as failed, and the client reconnects with backoff. +(`desktop_probe`). Any frame answers it, a publish's `OK` included; none +within 10 s ends the session as failed, and the client reconnects with +backoff. `legacy` is a lower bound on the app before its fixes. That app also gave each open channel, and each channel it watched without being a member, a @@ -267,7 +268,8 @@ whatever the limits. Keep `target.connect_timeout_secs` below 15, or a stalled connect is cancelled instead of timing out. Publishes, history fetches and HTTP requests can outlast the grace period (30 s, 30 s and 60 s deadlines), in which case the client is cancelled and the run fails. Desktop reads, like -EOSE waits, are abandoned at the stop and not recorded. +EOSE waits, are abandoned at the stop and not recorded, and a reply whose +thread-root lookup the stop cuts short is not published. The process exits non-zero when a step aborts, a client panics, a client had to be cancelled, or `shutdown.abort_reason` is set. From 8e9114d1cefd95041c1673a6ee6e266821d5ed1b Mon Sep 17 00:00:00 2001 From: Luke Tornquist Date: Tue, 3 Nov 2026 15:47:04 -0500 Subject: [PATCH 3/9] test(loadgen): set sub-second ticks after parsing; fmt The new probe tests need ticks shorter than a scenario may ask for, so the rig sets them on the parsed profile. The stop-cut reply test waits for its first root search rather than two. Co-authored-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Co-Authored-By: Claude Opus 5.5 Signed-off-by: Luke Tornquist --- .../buzz-test-client/src/loadgen/desktop.rs | 36 +++++++++++++------ 1 file changed, 25 insertions(+), 11 deletions(-) diff --git a/crates/buzz-test-client/src/loadgen/desktop.rs b/crates/buzz-test-client/src/loadgen/desktop.rs index 4624198f620..bafb6a239da 100644 --- a/crates/buzz-test-client/src/loadgen/desktop.rs +++ b/crates/buzz-test-client/src/loadgen/desktop.rs @@ -2174,8 +2174,11 @@ mod tests { /// of their own that `rig` sets. fn desktops(rig: &Rig, ws: &str, http: &str) -> Scenario { let (humans, joined, desktop) = (rig.humans, rig.joined, &rig.desktop); - let (presence, messages, reactions) = - (rig.presence_secs, rig.messages_per_hour, rig.reactions_per_hour); + let (presence, messages, reactions) = ( + rig.presence_secs, + rig.messages_per_hour, + rig.reactions_per_hour, + ); Scenario::parse(&format!( r#" run_id = "d" @@ -2291,6 +2294,8 @@ hold_secs = 1 /// Their messages, each a thread reply, and reactions. messages_per_hour: f64, reactions_per_hour: f64, + /// A tick shorter than a scenario may ask for. + tick_secs: Option, } /// One client, with a co-member whose profile it reads. @@ -2316,6 +2321,7 @@ hold_secs = 1 presence_secs: 3600.0, messages_per_hour: 0.0, reactions_per_hour: 0.0, + tick_secs: None, } } @@ -2324,7 +2330,10 @@ hold_secs = 1 async fn drive(rig: Rig, min: Duration, done: impl Fn(&Outcome) -> bool) -> Outcome { let (ws, frames) = fake_ws_relay(rig.plan.clone(), rig.drop_first, rig.ok_delay).await; let (http, queries) = fake_api("d", rig.fail, rig.slow).await; - let scenario = desktops(&rig, &ws, &http); + let mut scenario = desktops(&rig, &ws, &http); + if let Some(secs) = rig.tick_secs { + scenario.desktop.tick_secs = secs; + } let net = Net::new(&scenario).await.unwrap(); let stats = Arc::new(Stats::default()); let (stop_tx, stop) = watch::channel(false); @@ -3033,7 +3042,8 @@ hold_secs = 1 // next heartbeat, probes. The probe's EOSE comes 0.3 s later, while // that heartbeat waits, and is read once its OK comes, after the // probe's 0.7 s deadline. - let mut rig = solo(5, "mode = \"current\"\ntick_secs = 0.25"); + let mut rig = solo(5, "mode = \"current\""); + rig.tick_secs = Some(0.25); rig.presence_secs = 1.0; rig.ok_delay = Duration::from_millis(500); rig.req_timeout = Duration::from_millis(700); @@ -3045,7 +3055,11 @@ hold_secs = 1 assert_eq!(out.errors("desktop_probe"), 0, "{:?}", out.ops); assert!(out.count("desktop_probe") >= 1, "{:?}", out.ops); assert_eq!(out.count("ws_connect_auth"), 1, "{:?}", out.ops); - assert!(out.frames.iter().all(|(conn, ..)| *conn == 0), "{:?}", out.frames); + assert!( + out.frames.iter().all(|(conn, ..)| *conn == 0), + "{:?}", + out.frames + ); } #[tokio::test] @@ -3053,7 +3067,8 @@ hold_secs = 1 // No probe ever gets its EOSE. Ticks every 0.5 s; a heartbeat every // second waits 0.2 s for its OK. A probe follows a silent tick, and // the next heartbeat's OK comes well inside its 3 s deadline. - let mut rig = solo(5, "mode = \"current\"\ntick_secs = 0.5"); + let mut rig = solo(5, "mode = \"current\""); + rig.tick_secs = Some(0.5); rig.presence_secs = 1.0; rig.ok_delay = Duration::from_millis(200); rig.req_timeout = Duration::from_secs(3); @@ -3072,7 +3087,8 @@ hold_secs = 1 // Reactions come several times a second but have nothing to react // to, so nothing is sent and nothing comes back: a silent tick // still probes. - let mut rig = solo(5, "mode = \"current\"\ntick_secs = 0.5"); + let mut rig = solo(5, "mode = \"current\""); + rig.tick_secs = Some(0.5); rig.reactions_per_hour = 36_000.0; let out = drive(rig, Duration::ZERO, |o| o.count("desktop_probe") >= 1).await; assert_eq!(out.errors("desktop_probe"), 0, "{:?}", out.ops); @@ -3087,12 +3103,10 @@ hold_secs = 1 rig.messages_per_hour = 36_000.0; rig.slow = Some(("roots", Duration::from_secs(8))); rig.ok_delay = Duration::from_secs(60); - let out = drive(rig, Duration::from_millis(500), |o| { + let out = drive(rig, Duration::from_secs(1), |o| { o.queries .iter() - .filter(|f| f["kinds"] == json!([9]) && f.get("authors").is_some()) - .count() - >= 2 + .any(|f| f["kinds"] == json!([9]) && f.get("authors").is_some()) }) .await; assert!(!out.saw("EVENT", "9"), "{:?}", out.frames); From 0a7a653cf6e49ebcc7ba3ff2f792ebb444e35e87 Mon Sep 17 00:00:00 2001 From: Luke Tornquist Date: Tue, 3 Nov 2026 15:52:51 -0500 Subject: [PATCH 4/9] test(loadgen): keep the fake API's slow reads off the worker; answer REQs after their delay A slow read slept on a runtime worker and stalled the tasks queued behind it, so the reply-stop test saw the stop late. The fake relay now sends a REQ's replayed events after its delay too, so the long-backlog test's events arrive while an OK is awaited and pile up as a backlog. Every existing plan with events has no delay. Co-authored-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Co-Authored-By: Claude Opus 5.5 Signed-off-by: Luke Tornquist --- crates/buzz-test-client/src/loadgen/desktop.rs | 15 ++++++++------- 1 file changed, 8 insertions(+), 7 deletions(-) diff --git a/crates/buzz-test-client/src/loadgen/desktop.rs b/crates/buzz-test-client/src/loadgen/desktop.rs index bafb6a239da..ad3790cbbba 100644 --- a/crates/buzz-test-client/src/loadgen/desktop.rs +++ b/crates/buzz-test-client/src/loadgen/desktop.rs @@ -1958,7 +1958,7 @@ mod tests { type Frames = Shared2>; /// How the fake relay answers a REQ on a connection: events replayed - /// before its EOSE, and how long the EOSE waits. + /// before its EOSE, and how long the answer waits. type Plan = Arc (usize, Duration) + Send + Sync>; fn instant() -> Plan { @@ -2037,10 +2037,10 @@ mod tests { let (events, delay) = plan(index, &id); let (tx, replayed, id) = (tx.clone(), replayed.clone(), id.clone()); tokio::spawn(async move { + tokio::time::sleep(delay).await; for _ in 0..events { let _ = tx.send(text(json!(["EVENT", id, replayed]))); } - tokio::time::sleep(delay).await; let _ = tx.send(text(json!(["EOSE", id]))); }); } @@ -2130,7 +2130,8 @@ mod tests { "" }; if let Some((_, delay)) = slow.filter(|(slow, _)| *slow == name) { - std::thread::sleep(delay); + // Off the worker, so tasks queued behind this one still run. + tokio::task::block_in_place(|| std::thread::sleep(delay)); } if fail.contains(&name) && !failed.lock().unwrap().contains(&name) { failed.lock().unwrap().push(name); @@ -3018,10 +3019,10 @@ hold_secs = 1 #[tokio::test] async fn a_long_backlog_behind_slow_oks_keeps_its_timely_eoses() { - // Every REQ replays 600 events, more than one pass drains, with its - // EOSE 1.2 s late, while heartbeats each second wait 3.5 s for OKs. - // No heartbeat may run between the backlog's passes: a second OK - // wait would outlast the deferral. + // Every REQ answers 1.2 s late with 600 events, more than one pass + // drains, and its EOSE, while heartbeats each second wait 3.5 s for + // OKs. No heartbeat may run between the backlog's passes: a second + // OK wait would outlast the deferral. let mut rig = solo(5, &format!("mode = \"current\"\n{QUIET}")); rig.presence_secs = 1.0; rig.ok_delay = Duration::from_millis(3_500); From 5d2b8991751302a17886b017a7406c0ab4eeea2b Mon Sep 17 00:00:00 2001 From: Luke Tornquist Date: Tue, 3 Nov 2026 15:59:01 -0500 Subject: [PATCH 5/9] fix(loadgen): time a backlog's deferral from the first pass that defers A deadline deferred to a backlog was bounded at twice its timeout from the REQ, so an OK wait longer than that left no time to drain a backlog holding timely EOSEs. The deferral now runs one timeout from the first pass that defers, and resets once the backlog drains. Co-authored-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Co-Authored-By: Claude Opus 5.5 Signed-off-by: Luke Tornquist --- .../buzz-test-client/src/loadgen/desktop.rs | 68 ++++++++++++------- 1 file changed, 44 insertions(+), 24 deletions(-) diff --git a/crates/buzz-test-client/src/loadgen/desktop.rs b/crates/buzz-test-client/src/loadgen/desktop.rs index ad3790cbbba..1b9ecfbc733 100644 --- a/crates/buzz-test-client/src/loadgen/desktop.rs +++ b/crates/buzz-test-client/src/loadgen/desktop.rs @@ -1249,6 +1249,8 @@ impl Client { // Whether any frame came since the last tick, the probe a silent // tick sends, and that probe's subscription, closed by the next one. let mut heard = false; + // When the deadlines past due were first deferred to a backlog. + let mut deferred_since: Option = None; let mut probe: Option> = None; let mut probe_id: Option = None; let mut activity = Activity::new(&scenario.human, &mut self.rng, self.split); @@ -1395,9 +1397,16 @@ impl Client { // replacement keeps the batch it would have replaced, and the next // tick tries again); a connection's live route stays open, and // establishes as cut, since it may have missed history. + let overdue = |w: &Waiting| w.sent.elapsed() >= self.req_timeout; + let deferring = defer( + &mut deferred_since, + backlogged, + waiting.values().any(overdue), + self.req_timeout, + ); let timed_out: Vec = waiting .iter() - .filter(|(_, w)| expired(w.sent.elapsed(), self.req_timeout, backlogged)) + .filter(|(_, w)| !deferring && overdue(w)) .map(|(id, _)| id.clone()) .collect(); for id in timed_out { @@ -1442,14 +1451,10 @@ impl Client { } } - // While a deadline waits on the backlog, the pass only drains: a + // While deadlines wait on the backlog, the pass only drains: a // publish waiting for its OK, or a tick, could outlast the // deferral and expire an EOSE already in hand. - if backlogged - && waiting - .values() - .any(|w| w.sent.elapsed() >= self.req_timeout) - { + if deferring { continue; } @@ -1544,12 +1549,16 @@ impl Client { } } -/// Whether a REQ that has waited `waited` for its EOSE is judged expired. A -/// backlog still in hand may hold the EOSE in time, so its deadline waits for -/// the next pass, but at most one more `timeout`: traffic that keeps every -/// pass full does not defer it forever. -fn expired(waited: Duration, timeout: Duration, backlogged: bool) -> bool { - waited >= timeout && (!backlogged || waited >= timeout * 2) +/// Whether a pass defers the deadlines already past. A backlog still in hand +/// may hold their EOSEs in time, so they wait for the next pass, but for at +/// most `timeout` from the first pass that deferred them (`since`): traffic +/// that keeps every pass full does not defer them forever. +fn defer(since: &mut Option, backlogged: bool, overdue: bool, timeout: Duration) -> bool { + if !(backlogged && overdue) { + *since = None; + return false; + } + since.get_or_insert_with(Instant::now).elapsed() < timeout } async fn close(conn: &mut NostrWsConnection, id: &str) -> Result<(), WsClientError> { @@ -3006,27 +3015,37 @@ hold_secs = 1 } #[test] - fn a_backlog_defers_a_deadline_by_at_most_one_more_timeout() { + fn a_backlog_defers_deadlines_for_at_most_one_timeout() { let t = Duration::from_secs(10); - let s = Duration::from_secs; - assert!(!expired(s(9), t, false)); - assert!(expired(s(10), t, false)); - // A backlog may hold the EOSE: wait for it, but not forever. - assert!(!expired(s(10), t, true)); - assert!(!expired(s(19), t, true)); - assert!(expired(s(20), t, true)); + let mut since = None; + // Nothing past due, or nothing in hand: nothing to defer. + assert!(!defer(&mut since, true, false, t)); + assert!(!defer(&mut since, false, true, t)); + assert!(since.is_none()); + // A backlog may hold the EOSEs: wait for them, however late the + // first deferring pass comes, but not forever. + assert!(defer(&mut since, true, true, t)); + assert!(since.is_some()); + since = Instant::now().checked_sub(Duration::from_secs(9)); + assert!(defer(&mut since, true, true, t)); + since = Instant::now().checked_sub(t); + assert!(!defer(&mut since, true, true, t)); + // A drained backlog starts the next deferral afresh. + assert!(!defer(&mut since, false, true, t)); + assert!(since.is_none()); } #[tokio::test] async fn a_long_backlog_behind_slow_oks_keeps_its_timely_eoses() { // Every REQ answers 1.2 s late with 600 events, more than one pass // drains, and its EOSE, while heartbeats each second wait 3.5 s for - // OKs. No heartbeat may run between the backlog's passes: a second - // OK wait would outlast the deferral. + // OKs, more than twice the 1.5 s deadline. The deferral runs from + // the first pass that defers, not from the REQ, and no heartbeat may + // run between the backlog's passes: a second OK wait would outlast it. let mut rig = solo(5, &format!("mode = \"current\"\n{QUIET}")); rig.presence_secs = 1.0; rig.ok_delay = Duration::from_millis(3_500); - rig.req_timeout = Duration::from_millis(2_500); + rig.req_timeout = Duration::from_millis(1_500); rig.plan = Arc::new(|_, _| (600, Duration::from_millis(1_200))); let out = drive(rig, Duration::ZERO, |o| { o.count("publish_ephemeral_ok") >= 2 && o.count("desktop_live_eose") >= 4 @@ -3034,6 +3053,7 @@ hold_secs = 1 .await; assert_eq!(out.errors("desktop_live_eose"), 0, "{:?}", out.ops); assert_eq!(out.count("desktop_live_eose"), 4, "{:?}", out.ops); + assert!(!out.saw("CLOSE", "batch-1-0"), "{:?}", out.frames); } #[tokio::test] From 91f556946a616f2f4d6c9d3c57157ca8935b43ec Mon Sep 17 00:00:00 2001 From: Luke Tornquist Date: Tue, 3 Nov 2026 15:59:27 -0500 Subject: [PATCH 6/9] docs(loadgen): note how long a backlog may defer an EOSE deadline Co-authored-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Co-Authored-By: Claude Opus 5.5 Signed-off-by: Luke Tornquist --- perf/LOADGEN.md | 3 +++ 1 file changed, 3 insertions(+) diff --git a/perf/LOADGEN.md b/perf/LOADGEN.md index 7271cdd9e67..f47df0e95a0 100644 --- a/perf/LOADGEN.md +++ b/perf/LOADGEN.md @@ -188,6 +188,9 @@ profile read runs again when the profile subscription next establishes. A replacement is live-only and is opened before the subscription it replaces is closed. A replacement that misses its EOSE deadline is closed, the subscription it would have replaced stays, and the next tick tries again. +A deadline that passes while frames received in time are still unread waits +for them, for up to one more timeout, so under sustained traffic a missing +EOSE times out up to one timeout late. In both modes, a tick with no frame from the relay since the previous tick sends a silence probe, a live-only REQ for the viewer's membership notices (`desktop_probe`). Any frame answers it, a publish's `OK` included; none From a5c9618b340dc0d89092e3b3cc97c47bbf6333fa Mon Sep 17 00:00:00 2001 From: Luke Tornquist Date: Tue, 3 Nov 2026 16:18:11 -0500 Subject: [PATCH 7/9] test(loadgen): wait for every lower bound a desktop test asserts Three desktop journeys stopped as soon as part of what they assert held and then asserted the rest, which the flake check's repeats caught lagging. Each now waits for everything it asserts as a lower bound. Co-authored-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Co-Authored-By: Claude Opus 5.5 Signed-off-by: Luke Tornquist --- .../buzz-test-client/src/loadgen/desktop.rs | 19 ++++++++++++++++--- 1 file changed, 16 insertions(+), 3 deletions(-) diff --git a/crates/buzz-test-client/src/loadgen/desktop.rs b/crates/buzz-test-client/src/loadgen/desktop.rs index 1b9ecfbc733..05c75c8ae0c 100644 --- a/crates/buzz-test-client/src/loadgen/desktop.rs +++ b/crates/buzz-test-client/src/loadgen/desktop.rs @@ -2867,7 +2867,11 @@ hold_secs = 1 "mode = \"current\"\ntick_secs = 1.0\nbackstop_secs = 1.0", ), Duration::ZERO, - |o| o.count("desktop_roster") >= 4 && o.saw("CLOSE", "batch-1-0"), + |o| { + o.count("desktop_roster") >= 4 + && o.count("desktop_unread") >= 4 + && o.saw("CLOSE", "batch-1-0") + }, ) .await; no_errors(&after); @@ -2899,7 +2903,12 @@ hold_secs = 1 let large = drive( rig(2, 25, "mode = \"current\"\nrenewal_secs = 1.0\n"), Duration::ZERO, - |o| o.count("desktop_replay_eose") >= 12 && o.count("desktop_unread") >= 4, + |o| { + o.count("desktop_replay_eose") >= 12 + && o.count("desktop_unread") >= 4 + && o.count("ws_connect_auth") >= 4 + && o.count("desktop_window") >= 16 + }, ) .await; no_errors(&large); @@ -2921,7 +2930,11 @@ hold_secs = 1 let small = drive( rig(2, 5, "mode = \"current\"\nrenewal_secs = 1.0\n"), Duration::ZERO, - |o| o.count("desktop_live_eose") >= 12 && o.count("desktop_thread_window") == 2, + |o| { + o.count("desktop_live_eose") >= 12 + && o.count("ws_connect_auth") >= 4 + && o.count("desktop_thread_window") == 2 + }, ) .await; no_errors(&small); From 319a03f708cbbe0610cd008449b181459f9fada8 Mon Sep 17 00:00:00 2001 From: Luke Tornquist Date: Wed, 4 Nov 2026 10:10:42 -0500 Subject: [PATCH 8/9] fix(loadgen): judge desktop frames by arrival and match the app's presence route Review follow-ups on the desktop model: - The presence REQ now carries `since` and `limit: 0`, as the app sends its live-only routes, so it runs no history query the app never asks for. - `NostrWsConnection` keeps when each buffered frame arrived, and `next_event_received` returns it. The desktop session judges EOSEs and the silence probe by that instant, so a frame that arrived after its deadline while a publish waited for its OK is a timeout, not an answer. - Presence takes no part in establishment, as in the app: its timeout or EOSE no longer cuts a renewal's carry and releases the other routes. - A probe is closed at its EOSE, not when the next probe goes out. Co-authored-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Co-Authored-By: Claude Opus 5.5 Signed-off-by: Luke Tornquist --- .../buzz-test-client/src/loadgen/desktop.rs | 167 ++++++++++++++---- crates/buzz-ws-client/src/connection.rs | 43 +++-- perf/LOADGEN.md | 19 +- 3 files changed, 174 insertions(+), 55 deletions(-) diff --git a/crates/buzz-test-client/src/loadgen/desktop.rs b/crates/buzz-test-client/src/loadgen/desktop.rs index 05c75c8ae0c..56ebc8e97b7 100644 --- a/crates/buzz-test-client/src/loadgen/desktop.rs +++ b/crates/buzz-test-client/src/loadgen/desktop.rs @@ -88,6 +88,15 @@ pub(super) enum Route { Replay, } +impl Route { + /// Whether the route takes part in [`Establishment`]. A replay serves + /// its batch, and presence replays nothing, so neither is resumed, + /// carried nor recovered, and neither cuts a renewal's carry. + fn establishes(self) -> bool { + !matches!(self, Route::Presence | Route::Replay) + } +} + /// One REQ frame. #[derive(Clone, Debug, PartialEq)] pub(super) struct Req { @@ -156,7 +165,9 @@ fn global_reqs(viewer: &str, presence: &[String], since: u64) -> Vec { if !presence.is_empty() { reqs.push(Req { id: "presence".into(), - filters: vec![json!({"kinds": [KIND_PRESENCE_UPDATE], "authors": presence})], + filters: vec![ + json!({"kinds": [KIND_PRESENCE_UPDATE], "authors": presence, "since": since, "limit": 0}), + ], route: Route::Presence, }); } @@ -427,7 +438,7 @@ impl Establishment { /// `route` reached its EOSE, its replay `cut` or not. Returns the routes /// that establish now, each with its continuity. pub(super) fn eose(&mut self, route: Route, cut: bool) -> Vec<(Route, bool)> { - if route == Route::Replay { + if !route.establishes() { return Vec::new(); } let cut = cut || !self.resumed.contains(&route); @@ -455,7 +466,7 @@ impl Establishment { /// were cut, but only an EOSE proves a route live: it is not caught up, /// so the next connection cannot resume or carry it. pub(super) fn timeout(&mut self, route: Route) -> Vec<(Route, bool)> { - if route != Route::Replay { + if route.establishes() { self.timed_out.push(route); } self.eose(route, true) @@ -1247,7 +1258,8 @@ impl Client { let mut waiting: HashMap> = HashMap::new(); let mut open: HashSet = HashSet::new(); // Whether any frame came since the last tick, the probe a silent - // tick sends, and that probe's subscription, closed by the next one. + // tick sends, and that probe's subscription, closed at its EOSE or, + // failing that, by the next probe. let mut heard = false; // When the deadlines past due were first deferred to a backlog. let mut deferred_since: Option = None; @@ -1312,13 +1324,35 @@ impl Client { // The first frame waits; after it, only frames already in hand // are handled (at most DRAIN_FRAMES), so an EOSE buffered while a // publish waited for its OK is seen before deadlines are judged. + // Each frame is judged by when it arrived, not when it is read: + // one that came after its deadline is no answer. let mut established = Vec::new(); + // REQs whose EOSE came after their deadline. + let mut late = Vec::new(); + let req_timeout = self.req_timeout; + let in_time = move |sent: Instant, at: Instant| at.duration_since(sent) < req_timeout; let mut drained = 0; - let mut next = conn.next_event(wait).await; + let mut next = conn.next_event_received(wait).await; let backlogged = loop { - match next { - Ok(RelayMessage::Eose { subscription_id }) => { - if let Some(w) = waiting.remove(&subscription_id) { + let (at, message) = match next { + Ok(received) => received, + Err(WsClientError::Timeout) => break false, + Err(e) => break 'session disconnected(e), + }; + match message { + RelayMessage::Eose { subscription_id } => { + if probe_id.as_deref() == Some(subscription_id.as_str()) { + // The app closes its probe at its EOSE. + probe_id = None; + if let Err(e) = close(conn, &subscription_id).await { + break 'session disconnected(e); + } + } else if waiting + .get(&subscription_id) + .is_some_and(|w| !in_time(w.sent, at)) + { + late.extend(waiting.remove_entry(&subscription_id)); + } else if let Some(w) = waiting.remove(&subscription_id) { w.pending.ok(); let retired = match (w.route, w.slot) { (Route::Replay, _) => Some(subscription_id), @@ -1343,10 +1377,10 @@ impl Client { } } } - Ok(RelayMessage::Closed { + RelayMessage::Closed { subscription_id, message, - }) => { + } => { // Live subscriptions the relay refuses end the session, // as for the basic human; a refused replay is an error. if let Some(w) = waiting.remove(&subscription_id) { @@ -1362,7 +1396,7 @@ impl Client { break 'session End::Failed; } } - Ok(message) => { + message => { // An event before EOSE is replay, which the app counts // against its limit; only events after EOSE are delivery. if let RelayMessage::Event { @@ -1378,19 +1412,18 @@ impl Client { } } } - Err(WsClientError::Timeout) => break false, - Err(e) => break 'session disconnected(e), } - // Any frame proves the socket still reaches the relay. + // Any frame proves the socket still reaches the relay, but + // answers the probe only if it came by the probe's deadline. heard = true; - if let Some(p) = probe.take() { + if let Some(p) = probe.take_if(|p| in_time(p.sent, at)) { p.pending.ok(); } drained += 1; if drained == DRAIN_FRAMES { break true; } - next = conn.next_event(Duration::ZERO).await; + next = conn.next_event_received(Duration::ZERO).await; }; // A REQ past its deadline: a replay or a replacement is closed (a @@ -1409,10 +1442,11 @@ impl Client { .filter(|(_, w)| !deferring && overdue(w)) .map(|(id, _)| id.clone()) .collect(); - for id in timed_out { - let Some(w) = waiting.remove(&id) else { - continue; - }; + // A REQ whose EOSE came late timed out, whatever the backlog + // still holds. + let mut expired = late; + expired.extend(timed_out.iter().filter_map(|id| waiting.remove_entry(id))); + for (id, w) in expired { w.pending.err("EOSE timeout"); if w.route == Route::Replay || w.slot.is_some() { if w.slot.is_some() { @@ -1428,7 +1462,7 @@ impl Client { } // A probe nothing answered: the socket stopped delivering. Any - // frame drained above has answered it. + // frame drained above that came in time has answered it. if let Some(p) = probe.take_if(|p| p.sent.elapsed() >= self.req_timeout) { p.pending.err("probe timeout"); break End::Failed; @@ -1854,6 +1888,20 @@ mod tests { early.eose(Route::Membership, false), [(Route::Membership, false)] ); + + // Presence, like a replay, establishes nothing: neither its timeout + // nor its EOSE after one cuts a renewal's carry, and it is never + // caught up. + let mut presence = Establishment::new(Situation::Renewal, &all); + assert!(presence.timeout(Route::Presence).is_empty()); + assert!(presence.eose(Route::Presence, false).is_empty()); + for route in all { + assert!(presence.eose(route, false).is_empty()); + } + assert_eq!(presence.caught_up(), all); + let mut after = Establishment::new(Situation::Renewal, &presence.caught_up()); + assert!(after.eose(Route::Presence, false).is_empty()); + assert!(after.eose(Route::Profiles, false).is_empty()); } #[test] @@ -1872,9 +1920,10 @@ mod tests { ], "{mode:?}" ); + // Live-only, as the app sends it: no history query. assert_eq!( reqs[2].filters, - [json!({"kinds": [20001], "authors": ["a", "b"]})] + [json!({"kinds": [20001], "authors": ["a", "b"], "since": 100, "limit": 0})] ); let alone = connect_reqs(mode, "v", &[], &ids(5), 100, 1); assert!(alone.iter().all(|r| r.route != Route::Presence)); @@ -2982,25 +3031,39 @@ hold_secs = 1 assert!(ended.expect("the read task ended").is_err()); } - #[tokio::test] - async fn an_eose_in_time_is_not_expired_behind_a_slow_ok() { - // Every EOSE comes 1.2 s after its REQ, well within the deadline, - // but while the first presence heartbeat (within a second of the - // connect) waits 3.5 s for its OK, past the deadline. + /// One client whose four routes each get their EOSE `eose_after` the + /// REQ, while the first presence heartbeat (within a second of the + /// connect) waits 3.5 s for its OK, past the 2.5 s deadline. Every EOSE + /// arrives while the client waits for that OK. + async fn eoses_behind_a_slow_ok(eose_after: Duration) -> Outcome { let mut rig = solo(5, &format!("mode = \"current\"\n{QUIET}")); rig.presence_secs = 1.0; rig.ok_delay = Duration::from_millis(3_500); rig.req_timeout = Duration::from_millis(2_500); - rig.plan = Arc::new(|_, _| (0, Duration::from_millis(1_200))); - let out = drive(rig, Duration::ZERO, |o| { - o.count("publish_ephemeral_ok") >= 1 && o.count("desktop_live_eose") >= 4 + rig.plan = Arc::new(move |_, _| (0, eose_after)); + drive(rig, Duration::ZERO, |o| { + let eoses = o.count("desktop_live_eose") + o.errors("desktop_live_eose"); + o.count("publish_ephemeral_ok") >= 1 && eoses >= 4 }) - .await; + .await + } + + #[tokio::test] + async fn an_eose_in_time_is_not_expired_behind_a_slow_ok() { + let out = eoses_behind_a_slow_ok(Duration::from_millis(1_200)).await; assert_eq!(out.errors("desktop_live_eose"), 0, "{:?}", out.ops); assert_eq!(out.count("desktop_live_eose"), 4, "{:?}", out.ops); assert!(!out.saw("CLOSE", "batch-1-0"), "{:?}", out.frames); } + #[tokio::test] + async fn a_late_eose_is_not_answered_behind_a_slow_ok() { + // Read only once the OK comes, but they came after the deadline. + let out = eoses_behind_a_slow_ok(Duration::from_millis(3_000)).await; + assert_eq!(out.errors("desktop_live_eose"), 4, "{:?}", out.ops); + assert_eq!(out.count("desktop_live_eose"), 0, "{:?}", out.ops); + } + #[tokio::test] async fn a_silent_tick_probes_and_an_unanswered_probe_reconnects() { let mut rig = solo(5, "mode = \"current\"\ntick_secs = 1.0"); @@ -3096,6 +3159,48 @@ hold_secs = 1 ); } + #[tokio::test] + async fn a_probe_answered_late_behind_a_slow_ok_reconnects() { + // Ticks every 0.25 s; a heartbeat every 2 s waits 1.5 s for its OK. + // The tick after an OK hears it, so the next one probes, 0.25 s + // before the next heartbeat. On the first connection, the probe's + // EOSE comes 1.3 s after it, past its 1 s deadline, while that + // heartbeat waits; the OK itself comes later still. Neither answers. + let mut rig = solo(5, "mode = \"current\""); + rig.tick_secs = Some(0.25); + rig.presence_secs = 2.0; + rig.ok_delay = Duration::from_millis(1_500); + rig.req_timeout = Duration::from_secs(1); + rig.plan = Arc::new(|conn, id| { + let late = conn == 0 && id.starts_with("probe-"); + (0, Duration::from_millis(if late { 1_300 } else { 0 })) + }); + let out = drive(rig, Duration::ZERO, |o| { + o.errors("desktop_probe") >= 1 && o.count("ws_connect_auth") >= 2 + }) + .await; + assert_eq!(out.errors("desktop_probe"), 1, "{:?}", out.ops); + // The probe that failed went out before a heartbeat on its + // connection, whose OK wait held its EOSE. + let first = out.at("REQ", "probe-1").expect("probe-1 was sent"); + assert!( + out.frames[first..] + .iter() + .any(|(conn, verb, id)| *conn == 0 && verb == "EVENT" && id == "20001"), + "{:?}", + out.frames + ); + } + + #[tokio::test] + async fn an_answered_probe_is_closed_at_its_eose() { + // The app closes a probe at its EOSE, not when it sends the next. + let rig = solo(5, "mode = \"current\"\ntick_secs = 1.0"); + let out = drive(rig, Duration::ZERO, |o| o.saw("CLOSE", "probe-1")).await; + assert_eq!(out.count("desktop_probe"), 1, "{:?}", out.ops); + assert!(!out.saw("REQ", "probe-2"), "{:?}", out.frames); + } + #[tokio::test] async fn a_publish_ok_in_time_answers_an_outstanding_probe() { // No probe ever gets its EOSE. Ticks every 0.5 s; a heartbeat every diff --git a/crates/buzz-ws-client/src/connection.rs b/crates/buzz-ws-client/src/connection.rs index d5836aa7f5a..c2b1cc8772a 100644 --- a/crates/buzz-ws-client/src/connection.rs +++ b/crates/buzz-ws-client/src/connection.rs @@ -1,5 +1,5 @@ use std::collections::VecDeque; -use std::time::Duration; +use std::time::{Duration, Instant}; use futures_util::{SinkExt, StreamExt}; use nostr::{Event, Keys, Tag}; @@ -29,7 +29,9 @@ pub const PUBLISH_OK_TIMEOUT_SECS: u64 = 30; /// A NIP-42-capable WebSocket connection to a Nostr relay. pub struct NostrWsConnection { ws: WsStream, - buffer: VecDeque, + /// Messages read while waiting for something else, each with when it + /// was read. + buffer: VecDeque<(Instant, RelayMessage)>, pending_challenge: Option, relay_url: String, } @@ -164,10 +166,23 @@ impl NostrWsConnection { &mut self, timeout_dur: Duration, ) -> Result { - if let Some(msg) = self.buffer.pop_front() { - return Ok(msg); + Ok(self.next_event_received(timeout_dur).await?.1) + } + + /// [`Self::next_event`], with when the message was read off the socket. + /// + /// A message that arrived while another call, such as + /// [`Self::send_event`], waited for something else is returned later, + /// with the instant it arrived rather than the instant it is returned. + pub async fn next_event_received( + &mut self, + timeout_dur: Duration, + ) -> Result<(Instant, RelayMessage), WsClientError> { + if let Some(entry) = self.buffer.pop_front() { + return Ok(entry); } - self.recv_one(timeout_dur).await + let msg = self.recv_one(timeout_dur).await?; + Ok((Instant::now(), msg)) } /// Closes the WebSocket connection gracefully. @@ -185,10 +200,6 @@ impl NostrWsConnection { } async fn recv_one(&mut self, timeout_dur: Duration) -> Result { - if let Some(msg) = self.buffer.pop_front() { - return Ok(msg); - } - loop { let raw = timeout(timeout_dur, self.ws.next()) .await @@ -224,9 +235,9 @@ impl NostrWsConnection { if let Some(idx) = self .buffer .iter() - .position(|m| matches!(m, RelayMessage::Auth { .. })) + .position(|(_, m)| matches!(m, RelayMessage::Auth { .. })) { - match self.buffer.remove(idx).unwrap() { + match self.buffer.remove(idx).unwrap().1 { RelayMessage::Auth { challenge } => return Ok(challenge), _ => unreachable!(), } @@ -261,7 +272,7 @@ impl NostrWsConnection { } return Ok(challenge); } - other => self.buffer.push_back(other), + other => self.buffer.push_back((Instant::now(), other)), } } Message::Ping(data) => { @@ -283,9 +294,9 @@ impl NostrWsConnection { if let Some(idx) = self .buffer .iter() - .position(|m| matches!(m, RelayMessage::Ok(ok) if ok.event_id == event_id)) + .position(|(_, m)| matches!(m, RelayMessage::Ok(ok) if ok.event_id == event_id)) { - match self.buffer.remove(idx).unwrap() { + match self.buffer.remove(idx).unwrap().1 { RelayMessage::Ok(ok) => return Ok(ok), _ => unreachable!(), } @@ -313,9 +324,9 @@ impl NostrWsConnection { RelayMessage::Ok(ok) if ok.event_id == event_id => return Ok(ok), RelayMessage::Auth { ref challenge } => { self.pending_challenge = Some(challenge.clone()); - self.buffer.push_back(msg); + self.buffer.push_back((Instant::now(), msg)); } - other => self.buffer.push_back(other), + other => self.buffer.push_back((Instant::now(), other)), } } Message::Ping(data) => { diff --git a/perf/LOADGEN.md b/perf/LOADGEN.md index f47df0e95a0..5aaf7d2bee0 100644 --- a/perf/LOADGEN.md +++ b/perf/LOADGEN.md @@ -129,8 +129,9 @@ the report. On a **cold start** it: 1. reads the relay's NIP-11 document (`desktop_nip11`); -2. opens its profile and membership subscriptions, a live-only presence - subscription over up to 256 co-members, and its channel subscriptions +2. opens its profile and membership subscriptions, a live-only (`limit: 0`) + presence subscription over up to 256 co-members, which replays nothing + and recovers nothing, and its channel subscriptions (`desktop_live_eose`). With up to ten channels both modes open one per-channel batch that replays the newest 500 events. With more, `legacy` opens one such batch per ten channels; `current` opens one @@ -188,14 +189,16 @@ profile read runs again when the profile subscription next establishes. A replacement is live-only and is opened before the subscription it replaces is closed. A replacement that misses its EOSE deadline is closed, the subscription it would have replaced stays, and the next tick tries again. -A deadline that passes while frames received in time are still unread waits -for them, for up to one more timeout, so under sustained traffic a missing -EOSE times out up to one timeout late. +A frame is judged by when it arrived, even if it is read later because a +publish was waiting for its `OK`: an EOSE that arrived after its deadline is +a timeout. A deadline that passes while frames received in time are still +unread waits for them, for up to one more timeout, so under sustained +traffic a missing EOSE times out up to one timeout late. In both modes, a tick with no frame from the relay since the previous tick sends a silence probe, a live-only REQ for the viewer's membership notices -(`desktop_probe`). Any frame answers it, a publish's `OK` included; none -within 10 s ends the session as failed, and the client reconnects with -backoff. +(`desktop_probe`), closed at its EOSE. Any frame that arrives within 10 s +answers it, a publish's `OK` included; if none does, the session ends as +failed and the client reconnects with backoff. `legacy` is a lower bound on the app before its fixes. That app also gave each open channel, and each channel it watched without being a member, a From 446d31fe3d3b408a63bb89a325d123881fec994d Mon Sep 17 00:00:00 2001 From: Luke Tornquist Date: Wed, 4 Nov 2026 10:20:14 -0500 Subject: [PATCH 9/9] test(loadgen): pin the late probe to one that follows a heartbeat The first probe can come before any heartbeat, where no OK wait holds its EOSE, so only later probes answer late. Co-authored-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Co-Authored-By: Claude Opus 5.5 Signed-off-by: Luke Tornquist --- .../buzz-test-client/src/loadgen/desktop.rs | 24 ++++++++++--------- 1 file changed, 13 insertions(+), 11 deletions(-) diff --git a/crates/buzz-test-client/src/loadgen/desktop.rs b/crates/buzz-test-client/src/loadgen/desktop.rs index 56ebc8e97b7..31d464f13f4 100644 --- a/crates/buzz-test-client/src/loadgen/desktop.rs +++ b/crates/buzz-test-client/src/loadgen/desktop.rs @@ -3161,19 +3161,21 @@ hold_secs = 1 #[tokio::test] async fn a_probe_answered_late_behind_a_slow_ok_reconnects() { - // Ticks every 0.25 s; a heartbeat every 2 s waits 1.5 s for its OK. - // The tick after an OK hears it, so the next one probes, 0.25 s - // before the next heartbeat. On the first connection, the probe's - // EOSE comes 1.3 s after it, past its 1 s deadline, while that - // heartbeat waits; the OK itself comes later still. Neither answers. + // Ticks every 0.25 s; a heartbeat every 1 s waits 0.6 s for its OK. + // The tick after an OK hears it, so the next one probes, 0.15 s + // before the next heartbeat. The first probe may come before any + // heartbeat and is answered at once. On the first connection, every + // later probe's EOSE comes 0.6 s after it, past its 0.4 s deadline, + // while that heartbeat waits; the OK itself comes later still. + // Neither answers. let mut rig = solo(5, "mode = \"current\""); rig.tick_secs = Some(0.25); - rig.presence_secs = 2.0; - rig.ok_delay = Duration::from_millis(1_500); - rig.req_timeout = Duration::from_secs(1); + rig.presence_secs = 1.0; + rig.ok_delay = Duration::from_millis(600); + rig.req_timeout = Duration::from_millis(400); rig.plan = Arc::new(|conn, id| { - let late = conn == 0 && id.starts_with("probe-"); - (0, Duration::from_millis(if late { 1_300 } else { 0 })) + let late = conn == 0 && id.starts_with("probe-") && id != "probe-1"; + (0, Duration::from_millis(if late { 600 } else { 0 })) }); let out = drive(rig, Duration::ZERO, |o| { o.errors("desktop_probe") >= 1 && o.count("ws_connect_auth") >= 2 @@ -3182,7 +3184,7 @@ hold_secs = 1 assert_eq!(out.errors("desktop_probe"), 1, "{:?}", out.ops); // The probe that failed went out before a heartbeat on its // connection, whose OK wait held its EOSE. - let first = out.at("REQ", "probe-1").expect("probe-1 was sent"); + let first = out.at("REQ", "probe-2").expect("probe-2 was sent"); assert!( out.frames[first..] .iter()