diff --git a/crates/buzz-test-client/src/loadgen/desktop.rs b/crates/buzz-test-client/src/loadgen/desktop.rs index 2575d16061e..31d464f13f4 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,12 +80,23 @@ 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. 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 { @@ -120,10 +136,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 +161,17 @@ 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, "since": since, "limit": 0}), + ], + route: Route::Presence, + }); + } + reqs } /// Channel batch subscriptions, which stay open. Up to 10 channels share @@ -215,11 +243,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 +395,7 @@ pub(super) fn establish_fetches( reads } Route::Channels => channel_recovery(windows, threads).collect(), - Route::Replay => Vec::new(), + Route::Presence | Route::Replay => Vec::new(), } } @@ -409,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); @@ -437,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) @@ -688,11 +717,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 +878,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 +938,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 +1029,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 +1106,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 +1156,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 +1203,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 +1234,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 +1257,16 @@ 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 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; + 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,20 +1308,145 @@ impl Client { waiting.insert(req.id, waited); } + let due = [ + activity.due(), + Some(self.next_tick), + next_renewal, + probe.as_ref().map(|p| p.sent + self.req_timeout), + ] + .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. + // 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_received(wait).await; + let backlogged = loop { + 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), + (_, 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); + } + } + } + } + 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; + } + } + 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); + } + } + } + } + } + // 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(|p| in_time(p.sent, at)) { + p.pending.ok(); + } + drained += 1; + if drained == DRAIN_FRAMES { + break true; + } + next = conn.next_event_received(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. - let expired: Vec = waiting + 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)| w.sent.elapsed() >= self.req_timeout) + .filter(|(_, w)| !deferring && overdue(w)) .map(|(id, _)| id.clone()) .collect(); - let mut established = Vec::new(); - for id in expired { - 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() { @@ -1268,77 +1461,11 @@ 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; - } - } - 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), + // A probe nothing answered: the socket stopped delivering. Any + // 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; } // `legacy` read everything when it connected. @@ -1358,6 +1485,13 @@ impl Client { } } + // 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 deferring { + continue; + } + let step = activity .step( &shared, @@ -1371,6 +1505,14 @@ impl Client { .await; match step { 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), } @@ -1403,6 +1545,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,11 +1575,26 @@ 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 } } +/// 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> { conn.send_raw(&json!(["CLOSE", id])).await } @@ -1448,7 +1626,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 +1638,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 +1658,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) @@ -1710,6 +1888,47 @@ 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] + 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:?}" + ); + // Live-only, as the app sends it: no history query. + assert_eq!( + reqs[2].filters, + [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)); + } + assert!(establish_fetches(Route::Presence, false, Owed::default(), 2, 1).is_empty()); } #[test] @@ -1797,17 +2016,22 @@ 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 { 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, + /// 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, + ok_delay: Duration, + ) -> (String, Frames) { use futures_util::SinkExt; use tokio_tungstenite::tungstenite::Message; @@ -1851,25 +2075,36 @@ 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" | "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()); 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]))); }); } _ => {} } - if verb == "REQ" || verb == "CLOSE" { + if matches!(verb.as_str(), "REQ" | "CLOSE" | "EVENT") { seen.lock().unwrap().push((index, verb, id)); } } @@ -1880,7 +2115,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. @@ -1947,11 +2182,14 @@ mod tests { "read_state" } else if filter["kinds"] == json!([30300]) { "reminders" + } else if filter["kinds"] == json!([9]) && filter.get("authors").is_some() { + "roots" } else { "" }; 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); @@ -1991,9 +2229,15 @@ 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 { + /// `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, messages, reactions) = ( + rig.presence_secs, + rig.messages_per_hour, + rig.reactions_per_hour, + ); Scenario::parse(&format!( r#" run_id = "d" @@ -2008,10 +2252,11 @@ 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 +messages_per_hour = {messages:?} +reactions_per_hour = {reactions:?} +thread_reply_fraction = 1.0 history_limit = 1 [desktop] fraction = 1.0 @@ -2101,6 +2346,15 @@ hold_secs = 1 fail: FailOnce, slow: Slow, req_timeout: Duration, + /// How late the relay acknowledges each EVENT. + ok_delay: Duration, + /// The humans' presence interval. + presence_secs: f64, + /// 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. @@ -2122,15 +2376,23 @@ hold_secs = 1 fail: &[], slow: None, req_timeout: REQ_TIMEOUT, + ok_delay: Duration::ZERO, + presence_secs: 3600.0, + messages_per_hour: 0.0, + reactions_per_hour: 0.0, + tick_secs: None, } } /// 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 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); @@ -2217,16 +2479,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 +2501,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 +2823,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 +2894,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 @@ -2647,7 +2916,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); @@ -2679,7 +2952,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); @@ -2701,7 +2979,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); @@ -2716,4 +2998,260 @@ 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()); + } + + /// 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(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 + } + + #[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"); + 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); + } + + #[test] + fn a_backlog_defers_deadlines_for_at_most_one_timeout() { + let t = Duration::from_secs(10); + 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, 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(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 + }) + .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_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\""); + 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); + 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_probe_answered_late_behind_a_slow_ok_reconnects() { + // 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 = 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-") && 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 + }) + .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-2").expect("probe-2 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 + // 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\""); + 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); + 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\""); + 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); + 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_secs(1), |o| { + o.queries + .iter() + .any(|f| f["kinds"] == json!([9]) && f.get("authors").is_some()) + }) + .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 6a35fb0dbfd..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 @@ -519,21 +523,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 @@ -691,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 @@ -719,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, } @@ -738,6 +775,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 +813,7 @@ impl Activity { .flatten(), split, recent: [None, None], + root_bound: None, } } @@ -810,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).await; + 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"])); } @@ -839,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. @@ -854,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 + }) } } @@ -868,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) } } } @@ -1125,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/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/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 d0a9df95cb7..5aaf7d2bee0 100644 --- a/perf/LOADGEN.md +++ b/perf/LOADGEN.md @@ -129,14 +129,18 @@ 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 (`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 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 +149,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 +189,21 @@ 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 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`), 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 +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 +273,9 @@ 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, 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.