From 6bb151d19c4d56d2e25c6632f80e150fddcecc8e Mon Sep 17 00:00:00 2001 From: Luke Tornquist Date: Fri, 6 Nov 2026 09:26:41 -0500 Subject: [PATCH] fix(loadgen): agent socket publishes don't wait for their OK An agent awaited each presence, typing and frame OK inside its loop, so a slow OK froze the HTTP action in flight beside it. Socket publishes now go out without waiting; each OK is matched as it arrives, one not back within the client's publish timeout is an error, and at most 128 wait at once. Agent queries and submits no longer start after the stop, so a turn under way at the stop sends nothing past the request in flight. The guard moves into query() and submit(), replacing the call-site checks. LOADGEN.md now says the turn request mix approximates sustained turns. 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/run.rs | 323 +++++++++++++++++---- perf/LOADGEN.md | 15 +- 2 files changed, 283 insertions(+), 55 deletions(-) diff --git a/crates/buzz-test-client/src/loadgen/run.rs b/crates/buzz-test-client/src/loadgen/run.rs index af24527264..9f9203fb66 100644 --- a/crates/buzz-test-client/src/loadgen/run.rs +++ b/crates/buzz-test-client/src/loadgen/run.rs @@ -1,6 +1,6 @@ //! Drives the simulated population through the scenario's ramp. -use std::collections::{HashMap, HashSet}; +use std::collections::{HashMap, HashSet, VecDeque}; use std::future::Future; use std::pin::Pin; use std::sync::Arc; @@ -16,7 +16,8 @@ use buzz_core::kind::{ use buzz_core::observer::{ encrypt_observer_payload, OBSERVER_AGENT_TAG, OBSERVER_FRAME_TAG, OBSERVER_FRAME_TELEMETRY, }; -use buzz_ws_client::{NostrWsConnection, RelayMessage, WsClientError}; +use buzz_ws_client::connection::PUBLISH_OK_TIMEOUT_SECS; +use buzz_ws_client::{NostrWsConnection, OkResponse, RelayMessage, WsClientError}; use nostr::{ Alphabet, Event, EventBuilder, Filter, Keys, Kind, PublicKey, SingleLetterTag, Tag, Timestamp, }; @@ -28,7 +29,7 @@ use super::identity::{self, Role}; use super::net::{http_reason, Net}; use super::scenario::{AgentTurn, Community, HumanProfile, Members, Scenario, WideSplit}; use super::seed::{find_roots, hot_root_keys, reply_tags, HOT_ROOT_PREFIX, ROOT_PREFIX}; -use super::stats::{abort_reason, ops_json, Op, Stats}; +use super::stats::{abort_reason, ops_json, Op, Pending, Stats}; /// Content prefix of timed messages: `lg:`. const TIMED_PREFIX: &str = "lg:"; @@ -1171,11 +1172,107 @@ fn every(secs: f64, from: Instant) -> Option { (secs > 0.0).then(|| from + Duration::from_secs_f64(secs)) } +/// The most socket publishes an agent leaves awaiting their `OK`; past it, a +/// publish is an error instead of sent. A relay that answers within the +/// publish timeout keeps an agent far below it. +const MAX_UNACKED: usize = 128; + +/// An agent's socket publishes awaiting their `OK`, oldest first. An agent +/// sends a frame and moves on, as a harness does, so a slow `OK` holds up +/// neither its HTTP action nor its next frame. Each `OK` is matched as it +/// arrives; one not back within [`PUBLISH_OK_TIMEOUT_SECS`] is an error. +struct Unacked<'a> { + stats: &'a Stats, + /// Event id, the publish's timing and its deadline. Identical frames sent + /// within a second share an id; each `OK` settles the oldest. + waiting: VecDeque<(String, Pending<'a>, Instant)>, +} + +impl<'a> Unacked<'a> { + fn new(stats: &'a Stats) -> Self { + Self { + stats, + waiting: VecDeque::new(), + } + } + + /// Sends `event` as `op` without waiting for its `OK`. + async fn send( + &mut self, + conn: &mut NostrWsConnection, + op: Op, + event: Result, + ) -> Result<(), WsClientError> { + let event = match event { + Ok(event) => event, + Err(e) => { + self.stats.err(op, &e); + return Ok(()); + } + }; + if self.waiting.len() >= MAX_UNACKED { + self.stats.err(op, "too many publishes awaiting OK"); + return Ok(()); + } + let pending = self.stats.begin(op); + if let Err(e) = conn.send_raw(&json!(["EVENT", event])).await { + // The caller records the broken socket. + pending.forget(); + return Err(e); + } + let deadline = Instant::now() + Duration::from_secs(PUBLISH_OK_TIMEOUT_SECS); + self.waiting + .push_back((event.id.to_hex(), pending, deadline)); + Ok(()) + } + + /// Settles the oldest publish `ok` answers, if any. + fn answer(&mut self, ok: &OkResponse) { + let Some(at) = self.waiting.iter().position(|(id, ..)| *id == ok.event_id) else { + return; + }; + if let Some((_, pending, _)) = self.waiting.remove(at) { + if ok.accepted { + pending.ok(); + } else { + pending.err(&ok.message); + } + } + } + + /// When the oldest publish times out. + fn deadline(&self) -> Option { + self.waiting.front().map(|(.., deadline)| *deadline) + } + + /// Fails each publish whose `OK` is overdue at `now`. + fn expire(&mut self, now: Instant) { + while let Some((.., deadline)) = self.waiting.front() { + if *deadline > now { + break; + } + if let Some((_, pending, _)) = self.waiting.pop_front() { + pending.err("OK timeout"); + } + } + } + + /// Leaves every publish unjudged once its socket is gone: a broken + /// socket is recorded as such, and an `OK` cut off by the stop is not + /// judged, as a read abandoned at the stop is not. + fn forget(&mut self) { + for (_, pending, _) in self.waiting.drain(..) { + pending.forget(); + } + } +} + /// An agent client: NIP-OA authenticated socket with its mentions, then /// either HTTP polling, posts and observer frames on Poisson schedules, or /// turns of work (see [`AgentTurn`]), or both. One HTTP action at a time -/// runs beside the socket, so presence, typing and frames keep their -/// schedule while a request is slow, as a harness's do. +/// runs beside the socket, and socket publishes do not wait for their `OK` +/// (see [`Unacked`]), so presence, typing, frames and requests keep their +/// schedule while a request or an `OK` is slow, as a harness's do. async fn agent(shared: Arc, global: usize) { let scenario = &shared.scenario; let profile = &scenario.agent; @@ -1248,6 +1345,7 @@ async fn agent(shared: Arc, global: usize) { // connects together does not heartbeat together. let mut next_presence = every(profile.presence_interval_secs * rng.unit(), Instant::now()) .or_else(|| every(profile.presence_interval_secs, Instant::now())); + let mut unacked = Unacked::new(&shared.stats); while ok && !stopped(&shared) { // While an HTTP action runs, only socket work is waited for. @@ -1259,6 +1357,7 @@ async fn agent(shared: Arc, global: usize) { turn.as_ref().map_or(if_idle(next_turn), |t| t.due(idle)), if_idle(next_query), if_idle(next_post), + unacked.deadline(), ] .into_iter() .flatten() @@ -1276,6 +1375,7 @@ async fn agent(shared: Arc, global: usize) { } }; match received { + Some(Ok(RelayMessage::Ok(answer))) => unacked.answer(&answer), Some(Ok(message)) => { if live(&shared.stats, &message, subscribed, &mut last_message).is_err() { ok = false; @@ -1292,6 +1392,7 @@ async fn agent(shared: Arc, global: usize) { } } let now = Instant::now(); + unacked.expire(now); let idle = busy.is_none(); // One action per wakeup; anything else due runs on the next pass. // An HTTP action only starts with none in flight. @@ -1299,19 +1400,22 @@ async fn agent(shared: Arc, global: usize) { next_presence = every(profile.presence_interval_secs, now); let event = EventBuilder::new(kind(KIND_PRESENCE_UPDATE), "online").sign_with_keys(&keys); - send(&shared.stats, &mut conn, Op::PublishEphemeral, event).await + let event = event.map_err(|e| format!("sign: {e}")); + unacked.send(&mut conn, Op::PublishEphemeral, event).await } else if let Some(step) = turn.as_mut().and_then(|t| t.step(now, &profile.turn, idle)) { match step { TurnStep::Typing(channel) => { let event = EventBuilder::new(kind(KIND_TYPING_INDICATOR), "") .tags([tag(["h", &channel])]) - .sign_with_keys(&keys); - send(&shared.stats, &mut conn, Op::PublishEphemeral, event).await + .sign_with_keys(&keys) + .map_err(|e| format!("sign: {e}")); + unacked.send(&mut conn, Op::PublishEphemeral, event).await } TurnStep::Frame => { let frame = observer_frame(&keys, &owner, profile.turn.frame_bytes); - publish_frame(&shared.stats, &mut conn, frame).await + let frame = frame.map_err(|e| format!("observer: {e}")); + unacked.send(&mut conn, Op::PublishEphemeral, frame).await } TurnStep::Read(channel) => { let filters = json!([{ @@ -1323,7 +1427,7 @@ async fn agent(shared: Arc, global: usize) { query(sh, Op::HttpQuery, k, community, &filters, a).await; Outcome::Done })); - Ok(true) + Ok(()) } TurnStep::End => { if let Some(ended) = turn.take() { @@ -1332,7 +1436,7 @@ async fn agent(shared: Arc, global: usize) { Outcome::Ended })); } - Ok(true) + Ok(()) } } } else if idle && turn.is_none() && next_turn.is_some_and(|t| t <= now) { @@ -1341,7 +1445,7 @@ async fn agent(shared: Arc, global: usize) { busy = Some(Box::pin(async move { Outcome::Started(Box::new(start_turn(sh, community, k, a, plan).await)) })); - Ok(true) + Ok(()) } else if idle && next_query.is_some_and(|t| t <= now) { next_query = rng.after(query_per_hour); let channel = &channels[rng.below(channels.len())]; @@ -1354,7 +1458,7 @@ async fn agent(shared: Arc, global: usize) { query(sh, Op::HttpQuery, k, community, &filters, a).await; Outcome::Done })); - Ok(true) + Ok(()) } else if idle && next_post.is_some_and(|t| t <= now) { next_post = rng.after(profile.events_per_hour); let channel = &channels[rng.below(channels.len())]; @@ -1367,13 +1471,14 @@ async fn agent(shared: Arc, global: usize) { submit(sh, community, k, a, event, Op::HttpEvents).await; Outcome::Done })); - Ok(true) + Ok(()) } else if next_frame.is_some_and(|t| t <= now) { next_frame = rng.after(profile.observer_frames_per_hour); let frame = observer_frame(&keys, &owner, 0); - publish_frame(&shared.stats, &mut conn, frame).await + let frame = frame.map_err(|e| format!("observer: {e}")); + unacked.send(&mut conn, Op::PublishEphemeral, frame).await } else { - Ok(true) + Ok(()) }; if let Err(e) = sent { shared @@ -1382,6 +1487,7 @@ async fn agent(shared: Arc, global: usize) { ok = false; } } + unacked.forget(); let _ = conn.disconnect().await; // The action in flight finishes before the agent reconnects or // stops; each checks the stop between its requests. @@ -1428,7 +1534,7 @@ async fn start_turn( .as_ref() .and_then(|t| buzz_core::nip10::parse_thread_markers(&t.tags).resolve()) .map(|(root, _)| root); - if let Some(root) = root.filter(|_| !stopped(shared)) { + if let Some(root) = root { thread_context(shared, community, keys, auth, &channel, &root).await; } let author = target.as_ref().map(|e| e.pubkey.to_hex()); @@ -1438,15 +1544,10 @@ async fn start_turn( json!([{ "kinds": [KIND_GIT_REPO_ANNOUNCEMENT], "#buzz-channel": [channel] }]), json!([{ "kinds": [KIND_PROFILE], "authors": [author.unwrap_or_else(|| keys.public_key().to_hex())] }]), ]; - // Each request checks the stop, so a turn starting at the stop sends - // nothing more than the request under way. for filters in reads { - if stopped(shared) { - break; - } query(shared, Op::TurnContext, keys, community, &filters, auth).await; } - if let Some(target) = target.as_ref().filter(|_| !stopped(shared)) { + if let Some(target) = &target { for emoji in TURN_REACTIONS { let event = EventBuilder::new(kind(KIND_REACTION), emoji) .tags([tag(["h", &channel]), tag(["e", &target.id.to_hex()])]); @@ -1545,7 +1646,7 @@ async fn end_turn(shared: &Shared, community: &Community, keys: &Keys, auth: &Ta ) .tags(tags); submit(shared, community, keys, auth, reply, Op::HttpEvents).await; - let Some(target) = turn.target.filter(|_| !stopped(shared)) else { + let Some(target) = turn.target else { return; }; let mine = json!([{ @@ -1554,9 +1655,6 @@ async fn end_turn(shared: &Shared, community: &Community, keys: &Keys, auth: &Ta "#e": [target.id.to_hex()], }]); for _ in TURN_REACTIONS { - if stopped(shared) { - return; - } let found = query(shared, Op::HttpQuery, keys, community, &mine, auth) .await .unwrap_or_default(); @@ -1590,23 +1688,10 @@ fn observer_frame(keys: &Keys, owner: &Keys, pad: usize) -> Result, -) -> Result { - match frame { - Ok(event) => publish(stats, conn, Op::PublishEphemeral, event).await, - Err(e) => { - stats.err(Op::PublishEphemeral, &format!("observer: {e}")); - Ok(false) - } - } -} - -/// Signs and submits `event` through HTTP `/events` as `op`. A reaction the -/// relay already holds counts as answered: an earlier turn cut short by the -/// stop left it, and the next removal deletes it. +/// Signs and submits `event` through HTTP `/events` as `op`; as with +/// [`query`], none starts after the stop. A reaction the relay already holds +/// counts as answered: an earlier turn cut short by the stop left it, and +/// the next removal deletes it. async fn submit( shared: &Shared, community: &Community, @@ -1615,6 +1700,9 @@ async fn submit( event: EventBuilder, op: Op, ) { + if stopped(shared) { + return; + } let event = match event.sign_with_keys(keys) { Ok(event) => event, Err(e) => return shared.stats.err(op, &format!("sign: {e}")), @@ -1764,7 +1852,9 @@ async fn http( } } -/// An agent's HTTP `/query` as `op`; the events it returned. +/// An agent's HTTP `/query` as `op`; the events it returned. None starts +/// after the stop, so an action under way at the stop sends nothing more +/// than the request already in flight. async fn query( shared: &Shared, op: Op, @@ -1773,6 +1863,9 @@ async fn query( filters: &Value, auth: &Tag, ) -> Option> { + if stopped(shared) { + return None; + } http(shared, op, keys, community, "/query", filters, Some(auth)).await } @@ -2079,6 +2172,9 @@ hold_secs = 1 history_delay: Duration, /// From this long after the handshake, each `OK` takes this long. slow_ok: Option<(Duration, Duration)>, + /// Each EVENT's `OK` follows this long after `ok_delay`, without + /// holding up the frames after it. + ok_lag: Duration, } impl FakeRelay { @@ -2087,6 +2183,7 @@ hold_secs = 1 ok_delay: Some(Duration::ZERO), history_delay: Duration::ZERO, slow_ok: None, + ok_lag: Duration::ZERO, }; } @@ -2117,6 +2214,16 @@ hold_secs = 1 if ws.send(Message::Text(challenge.into())).await.is_err() { return; } + let (mut sink, mut ws) = ws.split(); + let (out, mut replies) = tokio::sync::mpsc::unbounded_channel::(); + tokio::spawn(async move { + while let Some(reply) = replies.recv().await { + let text = Message::Text(reply.to_string().into()); + if sink.send(text).await.is_err() { + return; + } + } + }); while let Some(Ok(Message::Text(text))) = ws.next().await { let frame: Value = serde_json::from_str(&text).unwrap(); let reply = match frame[0].as_str() { @@ -2147,15 +2254,20 @@ hold_secs = 1 } } tokio::time::sleep(delay).await; - json!(["OK", frame[1]["id"], true, ""]) + let ok = json!(["OK", frame[1]["id"], true, ""]); + if !relay.ok_lag.is_zero() { + let out = out.clone(); + tokio::spawn(async move { + tokio::time::sleep(relay.ok_lag).await; + let _ = out.send(ok); + }); + continue; + } + ok } _ => continue, }; - if ws - .send(Message::Text(reply.to_string().into())) - .await - .is_err() - { + if out.send(reply).is_err() { return; } } @@ -2606,7 +2718,30 @@ hold_secs = 3 /// live reactions, and is slow to answer tool reads. #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn agent_turns_read_react_stream_reply_and_remove_their_reactions() { - let (ws, seen) = fake_ws_relay(FakeRelay::PROMPT).await; + agent_turns_against(FakeRelay::PROMPT).await; + } + + /// The same turns when every socket `OK` is slower than the agent's + /// heartbeat, though well inside the abort ceiling: nothing waits for an + /// `OK`, so tool reads, typing and frames all keep their schedule. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn agent_turns_keep_moving_while_socket_oks_lag() { + let lag = Duration::from_millis(1500); + let report = agent_turns_against(FakeRelay { + ok_lag: lag, + ..FakeRelay::PROMPT + }) + .await; + let acked = &report["steps"][0]["ops"]["publish_ephemeral_ok"]; + assert!(acked["count"].as_u64() > Some(0), "{report}"); + assert_eq!(acked["errors"], 0, "{report}"); + let p50 = acked["p50_ms"].as_f64().unwrap(); + assert!(p50 >= lag.as_millis() as f64, "{report}"); + } + + /// Runs one agent's turns against `relay` and checks what they sent. + async fn agent_turns_against(relay: FakeRelay) -> Value { + let (ws, seen) = fake_ws_relay(relay).await; let channel = identity::channel_id("t", 0, 0); let me = identity::keys("t", Role::Agent, 0).public_key().to_hex(); // A human's thread reply: only a `reply` marker, naming the root. @@ -2783,6 +2918,7 @@ hold_secs = 3 queries.iter().any(|q| q[0]["limit"] == json!(50)), "{queries:?}" ); + report } /// A relay that answers every `/query` with 200 and a body that is not @@ -2822,6 +2958,89 @@ hold_secs = 3 ); } + /// Runs one agent's turns against a relay that holds one top-level + /// channel message and the agent's live reactions, and holds the first + /// request `held` picks past the stop. Returns every request's path and + /// body, in order, and where the held one is among them. + async fn turns_stopped_during(held: fn(&str, &Value) -> bool) -> (Vec<(String, Value)>, usize) { + let (ws, _) = fake_ws_relay(FakeRelay::PROMPT).await; + let channel = identity::channel_id("t", 0, 0); + let target = EventBuilder::new(kind(KIND_STREAM_MESSAGE), "question") + .tags([tag(["h", &channel])]) + .sign_with_keys(&Keys::generate()) + .unwrap(); + let log: Log<(String, Value)> = Default::default(); + let at: Arc>> = Default::default(); + let (requests, held_at) = (log.clone(), at.clone()); + // Released after the 3 s hold has stopped the run, well inside the + // shutdown grace. + let release = Instant::now() + Duration::from_millis(4500); + let http = super::super::seed::tests::fake_http(move |_, path, body| { + let body: Value = serde_json::from_slice(body).unwrap(); + let hold = { + let mut requests = requests.lock().unwrap(); + requests.push((path.to_string(), body.clone())); + let mut held_at = held_at.lock().unwrap(); + let hold = held_at.is_none() && held(path, &body); + if hold { + *held_at = Some(requests.len() - 1); + } + hold + }; + if hold { + tokio::task::block_in_place(|| { + std::thread::sleep(release.saturating_duration_since(Instant::now())); + }); + } + if path == "/events" { + return (200, json!({"accepted": true, "message": ""}).to_string()); + } + let found = match body[0]["kinds"][0].as_u64() { + Some(k) if k == u64::from(KIND_STREAM_MESSAGE) => json!([target]), + Some(k) if k == u64::from(KIND_REACTION) => { + let posted = requests.lock().unwrap(); + let reactions = posted + .iter() + .filter(|(path, e)| path == "/events" && e["kind"] == json!(KIND_REACTION)); + json!(reactions.map(|(_, e)| e).collect::>()) + } + _ => json!([]), + }; + (200, found.to_string()) + }) + .await; + let scenario = agent_turn_scenario(&ws, &http, ""); + let net = Net::new(&scenario).await.unwrap(); + let shard = Shard::parse("0/1").unwrap(); + let report = run_with(scenario, net, shard, Duration::from_secs(5)) + .await + .expect("a turn cut by the stop does not fail the run"); + assert_eq!(report["clients_cancelled"], 0, "{report}"); + let log = log.lock().unwrap().clone(); + let at = at.lock().unwrap().expect("the request was made and held"); + (log, at) + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn a_turn_stopped_while_adding_a_reaction_sends_nothing_more() { + let (log, at) = turns_stopped_during(|path, body| { + path == "/events" && body["kind"] == json!(KIND_REACTION) + }) + .await; + assert_eq!(at + 1, log.len(), "requests after the stop: {log:?}"); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn a_turn_stopped_while_reading_its_reactions_deletes_none() { + let (log, at) = turns_stopped_during(|path, body| { + path == "/query" && body[0]["kinds"][0] == json!(KIND_REACTION) + }) + .await; + assert_eq!(at + 1, log.len(), "requests after the stop: {log:?}"); + let deleted = log.iter().any(|(_, e)| e["kind"] == json!(KIND_DELETION)); + assert!(!deleted, "{log:?}"); + } + /// Holds `/events` responses until opened, counting the requests held. #[derive(Default)] struct Gate { diff --git a/perf/LOADGEN.md b/perf/LOADGEN.md index d7b9a695d6..3f679f4cff 100644 --- a/perf/LOADGEN.md +++ b/perf/LOADGEN.md @@ -129,9 +129,18 @@ does around a prompt: the message and a kind:5 deletion of each one found. An agent makes one HTTP request sequence at a time, and the socket keeps -running beside it: presence, typing and frames stay on schedule while a -request is slow, as a harness's do. A read counts as an error when its -response is not valid JSON of the expected shape. +running beside it: presence, typing and frames are sent without waiting for +their `OK`, so they stay on schedule while a request or an `OK` is slow, as a +harness's do. An `OK` not back within 30 seconds is an error. A read counts +as an error when its response is not valid JSON of the expected shape. +Requests do not start after the stop; one under way finishes. + +The request mix is an approximation of sustained turns, not a replay of the +harness. Every turn reads as a first prompt does. A harness that keeps +working in one thread reads 25 replies instead of 13, batches profile +lookups across prompts and caches project-home lookups for 30 seconds, so +the turn context here reads profiles and project homes more often than a +busy harness does. For a harness-like agent, also set `agent.query_interval_secs = 0` (no polling) and `events_per_hour = observer_frames_per_hour = 0`, so turns are