Keep Discord member scrolling responsive with viewport-driven caching

crates/chatd/src/adapters/discord.rs

@@ -1,9 +1,10 @@
 //! Discord, signed in as the user: the gateway websocket for everything live, the HTTP API for
 //! history and for what the user does.
 //!
-//! Two threads. The socket thread (`gateway.rs`) holds the gateway, turns its events into
+//! The socket thread (`gateway.rs`) holds the gateway, turns its events into
 //! `Context` calls and is the only one that writes to the socket; the command thread serves
-//! `Command`s, makes every HTTP request and hands gateway sends to the socket thread's outbox.
+//! HTTP commands and hands gateway sends to the socket thread's outbox. A separate
+//! viewport thread serves cache reads and Watch updates without waiting for HTTP/backfill.
 //!
 //! What stays in memory is bounded and listed on [`State`]: roles and ids per server, the lists
 //! the user owns, one visible member window, the voice states of the servers. Trees, people and
@@ -32,9 +33,9 @@ use protocol::{Conversation, ConversationKind, Message, Presence, RelateAction,
 use rest::{Refused, Rest};
 use serde::Deserialize;
 use serde_json::{json, Value};
-use std::collections::{HashMap, HashSet, VecDeque};
+use std::collections::{HashMap, HashSet};
 use std::sync::atomic::{AtomicBool, AtomicU64};
-use std::sync::mpsc::Receiver;
+use std::sync::mpsc::{channel, Receiver};
 use std::sync::{Arc, Mutex};
 use std::time::{Duration, Instant};
 
@@ -193,7 +194,7 @@ const ACK_EVERY: Duration = Duration::from_secs(3);
 
 /// Frames the command thread wants sent, and a local socket pair that wakes the socket thread.
 struct Outbox {
-    queue: Mutex<VecDeque<String>>,
+    queue: Mutex<gateway::SendQueue>,
     wake: gateway::WakePair,
 }
 
@@ -246,7 +247,7 @@ pub fn start(ctx: Context, session: &str, inbox: Receiver<Command>) -> Result<()
         token: session.token,
         state: Mutex::new(State::default()),
         outbox: Outbox {
-            queue: Mutex::new(VecDeque::new()),
+            queue: Mutex::new(gateway::SendQueue::default()),
             wake,
         },
         sequence: AtomicU64::new(0),
@@ -255,6 +256,21 @@ pub fn start(ctx: Context, session: &str, inbox: Receiver<Command>) -> Result<()
     });
     let socket = discord.clone();
     std::thread::spawn(move || keep_connected(&socket.ctx, || socket.connect()));
+    let (commands, network_inbox) = channel();
+    let viewport = discord.clone();
+    std::thread::spawn(move || {
+        // Cache reads and desired viewport updates must not sit behind history
+        // backfill sleeps or an HTTP request. Preserve their order on one thread.
+        while let Ok(command) = inbox.recv() {
+            match command {
+                command @ (Command::Members { .. } | Command::Watch(_) | Command::OpenSpace(_) | Command::SearchPeople(_)) => viewport.command(command),
+                command => {
+                    let stop = matches!(&command, Command::Stop);
+                    if commands.send(command).is_err() || stop { break; }
+                }
+            }
+        }
+    });
     std::thread::spawn(move || {
         // A send with files uploads on a thread of its own, so the commands behind it (its own
         // cancel among them) are not held up for as long as the upload takes.
@@ -273,7 +289,7 @@ pub fn start(ctx: Context, session: &str, inbox: Receiver<Command>) -> Result<()
             Command::ProfileIn { person_id, space_id, reply } => Discord::profile_queued(&discord, person_id, Some(space_id), reply),
             command => discord.command(command),
         };
-        serve_commands(&discord.ctx, inbox, PACE, |id| discord.backfill(id), other)
+        serve_commands(&discord.ctx, network_inbox, PACE, |id| discord.backfill(id), other)
     });
     Ok(())
 }
@@ -899,6 +915,7 @@ impl Discord {
         let unasked = match &command {
             Command::Watch(watching) => !ours(&watching.conversation) && !ours(&watching.space),
             Command::Warm(_)
+            | Command::Members { .. }
             | Command::Download { .. }
             | Command::Handoff { .. }
             | Command::CallLeave(_)

crates/chatd/src/adapters/discord/gateway.rs

@@ -40,12 +40,36 @@ 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);
-const SENDS_A_MINUTE: usize = 60;
+const SENDS_A_MINUTE: usize = 100;
 /// 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.
 const HELD: Duration = Duration::from_secs(30);
 
+#[derive(Default)]
+pub(super) struct SendQueue {
+    commands: VecDeque<String>,
+    subscriptions: serde_json::Map<String, Value>,
+}
+
+impl SendQueue {
+    fn push(&mut self, op: u8, data: Value) {
+        if op == 37 {
+            if let Some(guilds) = data.get("subscriptions").and_then(Value::as_object) {
+                self.subscriptions.extend(guilds.iter().map(|(guild, state)| (guild.clone(), state.clone())));
+            }
+        } else { self.commands.push_back(json!({"op": op, "d": data}).to_string()); }
+    }
+
+    fn pop(&mut self) -> Option<String> {
+        if let Some(command) = self.commands.pop_front() { return Some(command); }
+        if self.subscriptions.is_empty() { return None; }
+        Some(json!({"op": 37, "d": {"subscriptions": std::mem::take(&mut self.subscriptions)}}).to_string())
+    }
+
+    fn clear(&mut self) { *self = Self::default(); }
+}
+
 /// A socket handle for `poll` / `WSAPoll`.
 #[derive(Clone, Copy)]
 pub struct Pollable {
@@ -302,7 +326,7 @@ impl Discord {
 
     /// Queues a gateway frame for the socket thread.
     pub(super) fn tell(&self, op: u8, d: Value) {
-        self.outbox.queue.lock().unwrap().push_back(json!({"op": op, "d": d}).to_string());
+        self.outbox.queue.lock().unwrap().push(op, d);
         self.wake();
     }
 
@@ -419,13 +443,13 @@ impl Discord {
                     Err(lost) => return Err(self.lost(lost)),
                 }
             }
-            // Discord closes a socket that sends more than 120 frames a minute. Half of that is
-            // ours to use; anything beyond waits in the queue.
+            // 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_front() else {
+                let Some(frame) = self.outbox.queue.lock().unwrap().pop() else {
                     break;
                 };
                 wire.send(frame)?;
@@ -447,7 +471,12 @@ impl Discord {
             }
             let pending = self.tick();
             let until_beat = beat.next.saturating_duration_since(Instant::now());
-            let limit = if pending { Duration::from_millis(500) } else { IDLE };
+            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()));
+                }
+            }
             wire.wait(wake, until_beat.min(limit));
             self.outbox.wake.drain();
         }
@@ -545,10 +574,43 @@ impl Discord {
     /// Timers that ride on the socket thread. True while one is running, so the loop looks again
     /// soon.
     fn tick(&self) -> bool {
-        let members = self.members_tick();
         let voice = self.voice_tick();
         let search = self.search_tick();
         self.unmute_tick();
-        members || voice || search
+        voice || search
+    }
+}
+
+#[cfg(test)]
+mod send_queue_tests {
+    use super::*;
+
+    #[test]
+    fn scrolling_coalesces_only_unsent_snapshots_without_an_artificial_delay() {
+        let mut queue = SendQueue::default();
+        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 {
+            queue.push(37, json!({"subscriptions": {"guild": {"window": step}}}));
+            assert!(queue.pop().is_some(), "ready snapshots do not wait for a timer");
+        }
+    }
+
+    #[test]
+    fn coalescing_preserves_guild_unsubscribes_and_command_order() {
+        let mut queue = SendQueue::default();
+        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();
+        assert_eq!(command["op"], 4);
+        let subscription: Value = serde_json::from_str(&queue.pop().unwrap()).unwrap();
+        assert_eq!(subscription["d"]["subscriptions"], json!({"a": {"typing": false}, "b": {"typing": true}}));
+        assert!(queue.pop().is_none());
+        queue.clear();
+        queue.push(37, json!({"subscriptions": {"c": {}}}));
+        assert!(queue.pop().is_some(), "a new connection sends its first subscription immediately");
     }
 }