From c7fa3cbe77df655c2a1db32431b1dbdf207dd90e Mon Sep 17 00:00:00 2001 From: Justin Scroggins Date: Sun, 18 Oct 2026 16:26:36 -0400 Subject: [PATCH 1/2] fix: deliver a Discord member viewport chosen while offline after the session resumes A new gateway connection cleared the whole send queue, so a subscription queued while the socket was down was lost while Watched.sent already named it, and nothing said it again on the resumed session. A resumed connection now keeps every subscription change Discord may not have heard: those never written, and those written since the last heartbeat Discord answered, since a frame written as the socket died may never have arrived. They go as one frame once RESUMED comes. Commands are still dropped, and a new session still starts empty. Every connection also starts a minute of its own, as Discord counts the allowance per connection, so the old connection's sends, or a clock that stood still through a suspend, never hold the kept frame back. Fixes #42 Co-Authored-By: Claude Opus 5.5 --- crates/chatd/src/adapters/discord/gateway.rs | 101 ++++++++++++++++++- crates/chatd/tests/discord.rs | 43 ++++++++ 2 files changed, 139 insertions(+), 5 deletions(-) diff --git a/crates/chatd/src/adapters/discord/gateway.rs b/crates/chatd/src/adapters/discord/gateway.rs index d0f1a3f..921040b 100644 --- a/crates/chatd/src/adapters/discord/gateway.rs +++ b/crates/chatd/src/adapters/discord/gateway.rs @@ -60,6 +60,10 @@ const HELD: Duration = Duration::from_secs(30); pub(super) struct SendQueue { commands: VecDeque, subscriptions: serde_json::Map, + /// Subscription changes written since the last heartbeat went, and those written before the + /// heartbeat still unanswered: what a socket that died may have lost on the way. + written: serde_json::Map, + unanswered: serde_json::Map, /// When each frame of the last minute went, and whether it was a subscription. sent: VecDeque<(Instant, bool)>, /// When each guild's subscription last went, for the spacing. @@ -82,7 +86,9 @@ impl SendQueue { self.sent.push_back((now, subscription)); if !subscription { return self.commands.pop_front(); } for guild in self.subscriptions.keys() { self.subscribed.insert(guild.clone(), now); } - Some(json!({"op": 37, "d": {"subscriptions": std::mem::take(&mut self.subscriptions)}}).to_string()) + let frame = json!({"op": 37, "d": {"subscriptions": &self.subscriptions}}).to_string(); + self.written.extend(std::mem::take(&mut self.subscriptions)); + Some(frame) } /// When the next waiting frame may go, or `None` with nothing waiting. Commands go first and @@ -109,7 +115,22 @@ impl SendQueue { Some(spaced.max(free).max(now)) } - fn clear(&mut self) { *self = Self::default(); } + /// A heartbeat went. Discord reads in order, so its answer means it heard what was written + /// before it. + fn beat(&mut self) { self.unanswered.extend(std::mem::take(&mut self.written)); } + + fn answered(&mut self) { self.unanswered.clear(); } + + /// A new connection, with a minute of its own: Discord's allowance is per connection. Commands + /// were for the old one. A resumed session still needs every subscription change Discord may + /// not have heard, written or not, as one frame; a new session starts with none. + fn reconnected(&mut self, resuming: bool) { + let mut kept = serde_json::Map::new(); + if resuming { + for changes in [&mut self.unanswered, &mut self.written, &mut self.subscriptions] { kept.extend(std::mem::take(changes)); } + } + *self = Self { subscriptions: kept, ..Self::default() }; + } } /// A socket handle for `poll` / `WSAPoll`. @@ -461,7 +482,7 @@ impl Discord { None => address("TINYCLIENT_DISCORD_GATEWAY", GATEWAY), }; let mut wire = Wire::connect(&url, &self.rest.identity().user_agent)?; - self.outbox.queue.lock().unwrap().clear(); + self.outbox.queue.lock().unwrap().reconnected(session.is_some()); let mut beat = Beat { every: Duration::from_millis(41_250), next: Instant::now() + Duration::from_secs(60), @@ -498,6 +519,7 @@ impl Discord { let sequence = self.sequence.load(Ordering::Relaxed); let sequence = if sequence == 0 { Value::Null } else { json!(sequence) }; wire.send(json!({"op": 1, "d": sequence}).to_string())?; + self.outbox.queue.lock().unwrap().beat(); beat.acked = false; beat.next = Instant::now() + beat.every; if beat.ready { @@ -559,7 +581,10 @@ impl Discord { }; wire.send(hello.to_string())?; } - (11, _) => beat.acked = true, + (11, _) => { + beat.acked = true; + self.outbox.queue.lock().unwrap().answered(); + } // The server asked for a heartbeat now. (1, _) => beat.next = Instant::now(), (7, _) => return Err(Failure::Dropped("Discord asked to reconnect".to_owned())), @@ -673,7 +698,7 @@ mod send_queue_tests { assert_eq!(subscription["d"]["subscriptions"], json!({"a": {"typing": false}, "b": {"typing": true}})); assert!(queue.pop(now).is_none()); assert!(queue.due(now).is_none(), "nothing waiting, nothing to wake for"); - queue.clear(); + queue.reconnected(false); queue.push(37, json!({"subscriptions": {"c": {}}})); assert!(queue.pop(now).is_some(), "a new connection sends its first subscription immediately"); } @@ -742,4 +767,70 @@ mod send_queue_tests { let command: Value = serde_json::from_str(&queue.pop(due).unwrap()).unwrap(); assert_eq!(command["op"], 3); } + + #[test] + fn a_resumed_connection_keeps_unheard_subscriptions_and_a_new_session_does_not() { + let mut queue = SendQueue::default(); + let now = Instant::now(); + let subscriptions = |frame: Option| serde_json::from_str::(&frame.unwrap()).unwrap()["d"]["subscriptions"].clone(); + // Heard: a heartbeat written after it was answered. + queue.push(37, json!({"subscriptions": {"a": {"typing": true}}})); + assert!(queue.pop(now).is_some()); + queue.beat(); + queue.answered(); + // Written and not known to be heard, on either side of a heartbeat still unanswered. + queue.push(37, json!({"subscriptions": {"a": {"typing": false}, "b": {"typing": true}}})); + assert!(queue.pop(now).is_some()); + queue.beat(); + queue.push(37, json!({"subscriptions": {"b": {"window": 1}}})); + assert!(queue.pop(now).is_some()); + // Never written. + queue.push(37, json!({"subscriptions": {"b": {"window": 2}, "c": {"typing": false}}})); + queue.push(4, json!({"channel_id": "voice"})); + queue.reconnected(true); + assert_eq!(subscriptions(queue.pop(now)), json!({"a": {"typing": false}, "b": {"window": 2}, "c": {"typing": false}})); + assert!(queue.pop(now).is_none(), "commands are not carried to the next connection"); + + // What was heard before the drop is not said again. + queue.beat(); + queue.answered(); + queue.reconnected(true); + assert!(queue.pop(now).is_none()); + queue.push(37, json!({"subscriptions": {"b": {"typing": true}}})); + assert!(queue.pop(now).is_some()); + queue.push(37, json!({"subscriptions": {"b": {"window": 3}}})); + queue.reconnected(false); + assert!(queue.pop(now).is_none(), "a new session starts with nothing subscribed"); + } + + /// Discord counts each connection's minute on its own, and a suspended machine's clock stands + /// still, so the last connection's sends must not hold up the next one's. + #[test] + fn a_resumed_connection_starts_its_own_minute() { + let mut queue = SendQueue::default(); + let now = Instant::now(); + let mut sent = Vec::new(); + for guild in 0..SUBSCRIPTIONS_A_MINUTE { + queue.push(37, json!({"subscriptions": {guild.to_string(): {"typing": true}}})); + flush(&mut queue, now, &mut sent); + } + for _ in 0..SENDS_A_MINUTE { queue.push(3, json!({"status": "idle"})); } + flush(&mut queue, now, &mut sent); + assert_eq!(sent.len(), SENDS_A_MINUTE); + queue.beat(); + queue.answered(); + queue.push(37, json!({"subscriptions": {"guild": {"window": "kept"}}})); + assert!(queue.due(now).is_some_and(|due| due - now > MINUTE / 2), "the old connection's minute is spent"); + + queue.reconnected(true); + let kept: Value = serde_json::from_str(&queue.pop(now).expect("the kept viewport waits for nothing")).unwrap(); + assert_eq!(kept["d"]["subscriptions"], json!({"guild": {"window": "kept"}})); + assert!(queue.pop(now).is_none(), "commands are not carried to the next connection"); + queue.push(3, json!({"status": "online"})); + assert!(queue.pop(now).is_some(), "a command waits for nothing either"); + for step in 1..SUBSCRIPTION_BURST { + queue.push(37, json!({"subscriptions": {"guild": {"window": step}}})); + assert!(queue.pop(now).is_some(), "the new connection has its own burst"); + } + } } diff --git a/crates/chatd/tests/discord.rs b/crates/chatd/tests/discord.rs index 8ee83b3..fd5c224 100644 --- a/crates/chatd/tests/discord.rs +++ b/crates/chatd/tests/discord.rs @@ -1284,6 +1284,49 @@ fn a_dropped_socket_resumes_and_a_refused_token_expires_the_account() { assert_eq!(world.link(), Link::Expired); } +#[test] +fn member_viewport_changed_while_offline_is_delivered_after_resume() { + let world = World::start(); + world.tree("4000"); + let watch = |channel: &str| Watching { + space: Some(world.key("4000")), + conversation: Some(world.key(channel)), + members: Some((0, 50)), + ..Watching::default() + }; + let listed = |frame: &Value, channel: &str| frame["d"]["subscriptions"]["4000"]["channels"][channel] == json!([[0, 99]]); + world.ask(Request::Watch(watch("4101"))); + wait("the first viewport", || world.discord.frames_of(37).pop().filter(|frame| listed(frame, "4101"))); + + world.discord.control.lock().unwrap().send(Control::Drop).unwrap(); + wait("the link going down", || (world.link() == Link::Offline).then_some(())); + // Queued while there is no socket to write it to. + world.ask(Request::Watch(watch("4105"))); + wait("the link coming back", || (world.link() == Link::Connected).then_some(())); + assert_eq!(world.discord.frames_of(6).len(), 1, "the session was resumed"); + + // Discord still holds the old viewport, so the new one has to be said on the resumed session. + wait("the viewport chosen while offline", || { + let frames = world.discord.frames.lock().unwrap(); + let resumed = frames.iter().position(|frame| frame["op"] == 6)?; + frames[resumed + 1..].iter().find(|frame| frame["op"] == 37 && listed(frame, "4105")).cloned() + }); + world.discord.push("GUILD_MEMBER_LIST_UPDATE", json!({ + "guild_id": "4000", + "groups": [{"id": "online", "count": 1}], + "ops": [{"op": "SYNC", "range": [0, 99], "items": [ + {"group": {"id": "online", "count": 1}}, + {"member": {"user": {"id": "2006", "username": "sam", "global_name": "Sam Whitlock", "discriminator": "0", "avatar": null}, "roles": []}}, + ]}], + })); + let filled = wait("the new viewport's members", || { + let response = world.ask(Request::Members { conversation: world.key("4105"), offset: 0, limit: 50 }); + matches!(response, Response::Members { pending: false, .. }).then_some(response) + }); + let Response::Members { members, .. } = filled else { unreachable!() }; + assert_eq!(members.iter().map(|person| person.name.as_str()).collect::>(), ["Sam Whitlock"]); +} + #[test] fn the_space_on_screen_is_the_only_one_subscribed() { let world = World::start(); From d3382ef7fa97bbb2157f31739ee066fd2fe217d5 Mon Sep 17 00:00:00 2001 From: Justin Scroggins Date: Tue, 20 Oct 2026 09:43:21 -0400 Subject: [PATCH 2/2] Send what the queue lets go in a loop that ends on its own word, so clippy passes The `while beat.ready` loop added with the send pacing never changes its condition and ends only on the queue's `break`, which clippy denies as `while_immutable_condition`. An `if` around a `loop` sends the same frames and still lets the queue go before each write. Co-Authored-By: Claude Opus 5.5 --- crates/chatd/src/adapters/discord/gateway.rs | 15 +++++++++------ 1 file changed, 9 insertions(+), 6 deletions(-) diff --git a/crates/chatd/src/adapters/discord/gateway.rs b/crates/chatd/src/adapters/discord/gateway.rs index 921040b..cd0502a 100644 --- a/crates/chatd/src/adapters/discord/gateway.rs +++ b/crates/chatd/src/adapters/discord/gateway.rs @@ -505,12 +505,15 @@ impl Discord { Err(lost) => return Err(self.lost(lost)), } } - // What the allowance does not let go yet waits in the queue: see `SendQueue::due`. - while beat.ready { - let Some(frame) = self.outbox.queue.lock().unwrap().pop(Instant::now()) else { - break; - }; - wire.send(frame)?; + // What the allowance does not let go yet waits in the queue: see `SendQueue::due`. The + // queue is let go before each frame is written. + if beat.ready { + loop { + let Some(frame) = self.outbox.queue.lock().unwrap().pop(Instant::now()) else { + break; + }; + wire.send(frame)?; + } } if Instant::now() >= beat.next { if !beat.acked {