From 0ce6e1c485bdce356478d8a9ad4f00dd38f85f0d Mon Sep 17 00:00:00 2001 From: Justin Scroggins Date: Sun, 18 Oct 2026 16:27:14 -0400 Subject: [PATCH] fix: pace Discord member subscriptions so a scroll cannot spend the Gateway budget The send loop drained every coalesced subscription snapshot as soon as it was queued, so a burst of viewport changes spent the whole rolling allowance on intermediate positions and the final viewport waited for the minute to roll over. The send queue now owns the rolling window. The first eight subscription frames a minute go at once; past them each waits a spacing of 60 s / (80 - frames this minute), which settles a sustained scroll near forty frames a minute with the last viewport about a second behind. Subscriptions stop at 80 frames a minute, leaving twenty of the queued hundred for commands, and a guild not subscribed this minute skips the spacing. Heartbeats, identify and resume still go straight to the socket. Fixes #44 Co-Authored-By: Claude Opus 5.5 --- crates/chatd/src/adapters/discord/gateway.rs | 177 ++++++++++++++++--- 1 file changed, 153 insertions(+), 24 deletions(-) diff --git a/crates/chatd/src/adapters/discord/gateway.rs b/crates/chatd/src/adapters/discord/gateway.rs index bc2283d..d0f1a3f 100644 --- a/crates/chatd/src/adapters/discord/gateway.rs +++ b/crates/chatd/src/adapters/discord/gateway.rs @@ -13,7 +13,7 @@ use protocol::Link; use serde::Deserialize; use serde_json::value::RawValue; use serde_json::{json, Value}; -use std::collections::VecDeque; +use std::collections::{HashMap, VecDeque}; use std::io::{self, ErrorKind, Read, Write}; use std::net::{TcpListener, TcpStream}; use std::sync::atomic::Ordering; @@ -40,7 +40,17 @@ const GATEWAY: &str = "wss://gateway.discord.gg/?encoding=json&v=9"; const CAPABILITIES: u64 = 1_734_653; /// How often the loop looks up when nothing happens, to notice it was stopped. const IDLE: Duration = Duration::from_secs(5); +/// Queued frames a minute. Discord allows 120; the rest is kept for heartbeats and the handshake, +/// which go straight to the socket. const SENDS_A_MINUTE: usize = 100; +/// Member subscriptions go only while the minute holds fewer frames than this, so scrolling never +/// leaves a voice join or a presence change waiting for the minute to roll over. +const SUBSCRIPTIONS_A_MINUTE: usize = 80; +/// This many subscription frames a minute go the moment they are asked for. Past them each waits +/// a spacing that grows as the minute's allowance runs down, so a long scroll settles near half +/// of it and the last position is never more than a second or two behind. +const SUBSCRIPTION_BURST: usize = 8; +const MINUTE: Duration = Duration::from_secs(60); /// This many sessions refused in a row and no more are asked for until the user does something. const IDENTIFIES: u32 = 3; /// How often a held adapter looks up, to notice it was stopped. @@ -50,6 +60,10 @@ const HELD: Duration = Duration::from_secs(30); pub(super) struct SendQueue { commands: VecDeque, subscriptions: 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. + subscribed: HashMap, } impl SendQueue { @@ -61,12 +75,40 @@ impl SendQueue { } else { self.commands.push_back(json!({"op": op, "d": data}).to_string()); } } - fn pop(&mut self) -> Option { - if let Some(command) = self.commands.pop_front() { return Some(command); } - if self.subscriptions.is_empty() { return None; } + /// The next frame the allowance lets go at `now`. + fn pop(&mut self, now: Instant) -> Option { + if self.due(now)? > now { return None; } + let subscription = self.commands.is_empty(); + 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()) } + /// When the next waiting frame may go, or `None` with nothing waiting. Commands go first and + /// may take the whole allowance; subscriptions are spaced and capped below it. + fn due(&mut self, now: Instant) -> Option { + while self.sent.front().is_some_and(|(at, _)| now.saturating_duration_since(*at) >= MINUTE) { self.sent.pop_front(); } + self.subscribed.retain(|_, at| now.saturating_duration_since(*at) < MINUTE); + let free = match self.sent.front() { + Some((at, _)) if self.sent.len() >= SENDS_A_MINUTE => *at + MINUTE, + _ => now, + }; + if !self.commands.is_empty() { return Some(free); } + if self.subscriptions.is_empty() { return None; } + let used = self.sent.len(); + let spaced = if used >= SUBSCRIPTIONS_A_MINUTE { + self.sent[used - SUBSCRIPTIONS_A_MINUTE].0 + MINUTE + } else if self.sent.iter().filter(|(_, subscription)| *subscription).count() < SUBSCRIPTION_BURST { + now + } else { + // A guild not subscribed this minute, a server just opened, goes at once. + let gap = MINUTE / (SUBSCRIPTIONS_A_MINUTE - used) as u32; + self.subscriptions.keys().map(|guild| self.subscribed.get(guild).map_or(now, |at| *at + gap)).min().unwrap_or(now) + }; + Some(spaced.max(free).max(now)) + } + fn clear(&mut self) { *self = Self::default(); } } @@ -427,7 +469,6 @@ impl Discord { ready: false, identified: false, }; - let mut sent = VecDeque::new(); let wake = self.outbox.wake.poll_read(); loop { if self.ctx.stopped() { @@ -443,17 +484,12 @@ impl Discord { Err(lost) => return Err(self.lost(lost)), } } - // Discord allows 120 frames per minute. Twenty are - // reserved for heartbeats/handshake; anything beyond waits in the queue. - while sent.front().is_some_and(|at: &Instant| at.elapsed() > Duration::from_secs(60)) { - sent.pop_front(); - } - while beat.ready && sent.len() < SENDS_A_MINUTE { - let Some(frame) = self.outbox.queue.lock().unwrap().pop() else { + // 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)?; - sent.push_back(Instant::now()); } if Instant::now() >= beat.next { if !beat.acked { @@ -472,9 +508,9 @@ impl Discord { let pending = self.tick(); let until_beat = beat.next.saturating_duration_since(Instant::now()); let mut limit = if pending { Duration::from_millis(500) } else { IDLE }; - if sent.len() >= SENDS_A_MINUTE { - if let Some(at) = sent.front() { - limit = limit.min(Duration::from_secs(60).saturating_sub(at.elapsed())); + if beat.ready { + if let Some(due) = self.outbox.queue.lock().unwrap().due(Instant::now()) { + limit = limit.min(due.saturating_duration_since(Instant::now())); } } wire.wait(wake, until_beat.min(limit)); @@ -585,32 +621,125 @@ impl Discord { mod send_queue_tests { use super::*; + fn window(frame: &str) -> Value { + let frame: Value = serde_json::from_str(frame).unwrap(); + frame["d"]["subscriptions"]["guild"]["window"].clone() + } + + /// Everything the queue lets go at `now`, as the socket loop would send it. + fn flush(queue: &mut SendQueue, now: Instant, sent: &mut Vec<(Instant, bool)>) { + while let Some(frame) = queue.pop(now) { + sent.push((now, serde_json::from_str::(&frame).unwrap()["op"] == 37)); + } + } + + /// No rolling minute holds more than the allowance, of everything or of subscriptions. + fn within_allowance(sent: &[(Instant, bool)]) { + for (index, (start, _)) in sent.iter().enumerate() { + let minute = sent[index..].iter().take_while(|(at, _)| *at < *start + MINUTE); + let (all, subscriptions) = minute.fold((0, 0), |(all, subs), (_, sub)| (all + 1, subs + usize::from(*sub))); + assert!(all <= SENDS_A_MINUTE && subscriptions <= SUBSCRIPTIONS_A_MINUTE, "{all} frames, {subscriptions} subscriptions in a minute"); + } + } + #[test] fn scrolling_coalesces_only_unsent_snapshots_without_an_artificial_delay() { let mut queue = SendQueue::default(); + let now = Instant::now(); for step in 0..600 { queue.push(37, json!({"subscriptions": {"guild": {"window": step}}})); } assert_eq!(queue.subscriptions.len(), 1); - let frame: Value = serde_json::from_str(&queue.pop().unwrap()).unwrap(); - assert_eq!(frame["d"]["subscriptions"]["guild"]["window"], 599); - for step in 0..100 { + assert_eq!(window(&queue.pop(now).unwrap()), 599); + for step in 1..SUBSCRIPTION_BURST { queue.push(37, json!({"subscriptions": {"guild": {"window": step}}})); - assert!(queue.pop().is_some(), "ready snapshots do not wait for a timer"); + assert!(queue.pop(now).is_some(), "a short scroll does not wait for a timer"); } + queue.push(37, json!({"subscriptions": {"guild": {"window": "next"}}})); + assert!(queue.pop(now).is_none(), "past the burst a snapshot is spaced"); + let due = queue.due(now).unwrap(); + assert!(due > now && due - now < Duration::from_secs(1), "{:?}", due - now); + assert_eq!(window(&queue.pop(due).unwrap()), "next"); } #[test] fn coalescing_preserves_guild_unsubscribes_and_command_order() { let mut queue = SendQueue::default(); + let now = Instant::now(); queue.push(37, json!({"subscriptions": {"a": {"typing": true}}})); queue.push(37, json!({"subscriptions": {"a": {"typing": false}, "b": {"typing": true}}})); queue.push(4, json!({"channel_id": "voice"})); - let command: Value = serde_json::from_str(&queue.pop().unwrap()).unwrap(); + let command: Value = serde_json::from_str(&queue.pop(now).unwrap()).unwrap(); assert_eq!(command["op"], 4); - let subscription: Value = serde_json::from_str(&queue.pop().unwrap()).unwrap(); + let subscription: Value = serde_json::from_str(&queue.pop(now).unwrap()).unwrap(); assert_eq!(subscription["d"]["subscriptions"], json!({"a": {"typing": false}, "b": {"typing": true}})); - assert!(queue.pop().is_none()); + assert!(queue.pop(now).is_none()); + assert!(queue.due(now).is_none(), "nothing waiting, nothing to wake for"); queue.clear(); queue.push(37, json!({"subscriptions": {"c": {}}})); - assert!(queue.pop().is_some(), "a new connection sends its first subscription immediately"); + assert!(queue.pop(now).is_some(), "a new connection sends its first subscription immediately"); + } + + /// Five minutes of a viewport moving every 10 ms with a presence change every ten seconds + /// among it, on a clock the test moves. + #[test] + fn a_sustained_scroll_keeps_to_the_allowance_and_ends_on_the_last_viewport() { + let mut queue = SendQueue::default(); + let start = Instant::now(); + let mut now = start; + let mut sent = Vec::new(); + for step in 0..30_000u32 { + queue.push(37, json!({"subscriptions": {"guild": {"window": step}}})); + if step % 1000 == 0 { + queue.push(3, json!({"status": "online"})); + let before = sent.len(); + flush(&mut queue, now, &mut sent); + assert!(sent[before..].iter().any(|(_, sub)| !sub), "a command waited behind scrolling at {:?}", now - start); + } + flush(&mut queue, now, &mut sent); + assert!(queue.subscriptions.len() <= 1 && queue.commands.is_empty() && queue.subscribed.len() <= 1, "the queue grows"); + assert!(queue.sent.len() <= SENDS_A_MINUTE); + now += Duration::from_millis(10); + } + within_allowance(&sent); + let delivered = sent.iter().filter(|(_, sub)| *sub).count(); + assert!(delivered < 5 * SUBSCRIPTIONS_A_MINUTE, "{delivered} subscription frames in five minutes"); + // The pace settles with a reserve left: well under the share was spent in the last minute. + let recent = sent.iter().filter(|(at, sub)| *sub && *at + MINUTE > now).count(); + assert!(recent <= SUBSCRIPTIONS_A_MINUTE * 3 / 4, "{recent} subscriptions in the last minute"); + + // The scroll stops: the last viewport goes when the loop next wakes, within a second or two. + let due = queue.due(now).expect("the last viewport is waiting"); + assert!(due - now <= Duration::from_secs(2), "the last viewport waits {:?}", due - now); + assert_eq!(window(&queue.pop(due).unwrap()), 29_999); + assert!(queue.due(due).is_none()); + + // A server opened right after still goes at once. + queue.push(37, json!({"subscriptions": {"other": {"typing": true}}})); + assert!(queue.pop(due).is_some(), "navigation waited behind scrolling"); + } + + #[test] + fn a_spent_allowance_waits_for_the_minute_and_keeps_room_for_commands() { + let mut queue = SendQueue::default(); + let start = Instant::now(); + let mut sent = Vec::new(); + // Servers not yet subscribed each go at once, until the subscription share is spent. + for guild in 0..SUBSCRIPTIONS_A_MINUTE + 10 { + queue.push(37, json!({"subscriptions": {guild.to_string(): {"typing": true}}})); + flush(&mut queue, start + Duration::from_millis(guild as u64), &mut sent); + } + assert_eq!(sent.len(), SUBSCRIPTIONS_A_MINUTE); + assert_eq!(queue.subscriptions.len(), 10, "the rest wait as one frame"); + let now = start + Duration::from_secs(1); + 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, "commands take what subscriptions leave, and no more"); + assert_eq!(queue.commands.len(), SUBSCRIPTIONS_A_MINUTE); + within_allowance(&sent); + // The oldest frame ages out after a minute, and commands go before subscriptions. + let due = queue.due(now).unwrap(); + assert_eq!(due, start + MINUTE); + assert!(queue.pop(due - Duration::from_millis(1)).is_none()); + let command: Value = serde_json::from_str(&queue.pop(due).unwrap()).unwrap(); + assert_eq!(command["op"], 3); } }