From 6fa84d465380b776605b50c098d7ec3ce99a0b2d Mon Sep 17 00:00:00 2001 From: Vincent Bernardoff Date: Sun, 18 Oct 2026 16:13:20 +0200 Subject: [PATCH] fix: decouple Discord member scrolling from network requests Serve member cache reads and viewport changes independently of HTTP backfill. Keep a bounded GUI page cache, expose pending snapshots, and let the latest visible range drive one coalesced gateway subscription. Document the cache contract and cover rapid viewport changes in tests. Co-Authored-By: GPT-6 --- crates/chatd/src/adapters/discord.rs | 31 +- crates/chatd/src/adapters/discord/gateway.rs | 78 +++- crates/chatd/src/adapters/discord/members.rs | 172 ++++---- crates/chatd/src/adapters/mod.rs | 1 + crates/chatd/src/adapters/slack/people.rs | 1 + crates/chatd/src/adapters/teams/people.rs | 1 + crates/chatd/src/adapters/telegram.rs | 1 + crates/chatd/src/demo/adapter.rs | 2 +- crates/chatd/src/server.rs | 1 + crates/chatd/tests/discord.rs | 81 +++- crates/protocol/src/lib.rs | 17 + crates/tinyclient/src/gui/people.rs | 3 + crates/tinyclient/src/gui/people/hub.rs | 20 +- crates/tinyclient/src/gui/people/info.rs | 13 +- crates/tinyclient/src/gui/people/members.rs | 406 +++++++++++++++++-- docs/architecture.md | 2 +- docs/design/polish/07-panes.md | 42 ++ 17 files changed, 707 insertions(+), 165 deletions(-) diff --git a/crates/chatd/src/adapters/discord.rs b/crates/chatd/src/adapters/discord.rs index 404c82d..4102fa7 100644 --- a/crates/chatd/src/adapters/discord.rs +++ b/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>, + queue: Mutex, wake: gateway::WakePair, } @@ -246,7 +247,7 @@ pub fn start(ctx: Context, session: &str, inbox: Receiver) -> 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) -> 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) -> 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(_) diff --git a/crates/chatd/src/adapters/discord/gateway.rs b/crates/chatd/src/adapters/discord/gateway.rs index d49c29c..bc2283d 100644 --- a/crates/chatd/src/adapters/discord/gateway.rs +++ b/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, + subscriptions: serde_json::Map, +} + +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 { + 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"); } } diff --git a/crates/chatd/src/adapters/discord/members.rs b/crates/chatd/src/adapters/discord/members.rs index 1600f33..2ecaea1 100644 --- a/crates/chatd/src/adapters/discord/members.rs +++ b/crates/chatd/src/adapters/discord/members.rs @@ -75,15 +75,6 @@ struct Roster { rows: BTreeMap, } -struct Asked { - offset: u32, - limit: u32, - reply: Reply, - since: Instant, - /// The ranges were corrected once the sections were known; not again. - corrected: bool, -} - #[derive(Default)] pub struct Watched { /// The server whose tree or channel is on screen. @@ -94,7 +85,6 @@ pub struct Watched { people: Vec, sent: Option, roster: Option, - asked: Option, /// The shell's screen as last acted on, so a heartbeat that finds it unchanged does nothing. seen: Arc, /// When the shell last asked for the Space's tree. @@ -125,7 +115,6 @@ impl Watched { self.list = channel.map(str::to_owned); self.ranges = if channel.is_some() { vec![(0, WINDOW - 1)] } else { Vec::new() }; self.roster = None; - self.asked = None; } } } @@ -215,6 +204,16 @@ pub struct ListUpdate { ops: Vec, } +impl ListUpdate { + fn changes_roster(&self, groups: &[(String, u32)]) -> bool { + // SYNC delivers a requested window and INVALIDATE discards an old one. + // Neither edits the roster; treating them as edits makes readers + // invalidate the page that just arrived and starves later queued pages. + !self.groups.iter().map(|group| (&group.id, group.count)).eq(groups.iter().map(|(id, count)| (id, *count))) + || self.ops.iter().any(|op| matches!(op.op.as_str(), "UPDATE" | "INSERT" | "DELETE")) + } +} + #[derive(Deserialize)] pub struct Chunk { #[serde(default)] @@ -328,11 +327,22 @@ impl Discord { (None, None, _) if **watching == Watching::default() => state.members.show(None), (None, None, _) => state.members.list(None), } + if state.members.list.is_some() { + if let Some((offset, count)) = watching.members { + // Round to GUI cache pages so both edge pages are available. + let first = offset / 50 * 50; + let end = offset.saturating_add(count).saturating_add(49) / 50 * 50; + state.members.ranges = ranges(state.members.roster.as_ref(), first, end.saturating_sub(first)); + } + } + let listed: HashSet = state.members.roster.as_ref().into_iter().flat_map(|roster| roster.rows.iter()) + .filter(|(row, _)| state.members.ranges.iter().any(|(from, to)| *row >= from && *row <= to)) + .filter_map(|(_, entry)| match entry { Entry::Member(person) => Some(person.key.id.clone()), _ => None }).collect(); // Friends' presence arrives unasked. Anyone else on screen is asked for by id, and // only as part of the server that is on screen anyway. let relations = &state.relations; let strangers = watching.people.iter().filter(|key| ours(key)).map(|key| &key.id); - let strangers = strangers.filter(|id| relations.get(*id) != Some(&Relation::Friend) && **id != state.me); + let strangers = strangers.filter(|id| !listed.contains(*id) && relations.get(*id) != Some(&Relation::Friend) && **id != state.me); state.members.people = match state.members.on_screen { Some(_) => strangers.take(PEOPLE).cloned().collect(), None => Vec::new(), @@ -358,81 +368,39 @@ impl Discord { named.collect() } - /// Answers the waiting `Members` request if the rows it asked for are all here. If the - /// sections show they are in another window, asks for that one, once. - fn settle(&self) { - let mut again = false; - let answered = { - let mut state = self.state.lock().unwrap(); - let watched = &mut state.members; - let (Some(asked), Some(roster), Some(guild)) = (&mut watched.asked, &watched.roster, &watched.on_screen) else { - return; - }; - match roster.window(asked.offset, asked.limit) { - Some(members) => { - let groups = roster.groups.clone(); - let guild = guild.clone(); - watched.asked.take().map(|asked| (asked.reply, members, groups, guild)) - } - None => { - let needed = ranges(Some(roster), asked.offset, asked.limit); - if !asked.corrected && needed != watched.ranges { - asked.corrected = true; - watched.ranges = needed; - again = true; - } - None - } - } - }; - if again { - self.sync_subscription(); - } - if let Some((reply, members, groups, guild)) = answered { - let roster = Roster { - channel: String::new(), - groups, - rows: BTreeMap::new(), - }; - let sections = self.sections(&guild, &roster); - answer(reply, Ok(Members { sections, members })); - } - } - - /// The visible rows of a server channel's member list. + /// Reads the current cache immediately. Only Watch controls the live viewport; + /// a cache miss is pending data, never a request that blocks later scrolling. pub(super) fn members(&self, channel: String, offset: u32, limit: u32, reply: Reply) { let conversation = self.ctx.known(&channel).ok().flatten(); let guild = conversation.as_ref().and_then(|conversation| self.guild_of(conversation)); let in_tree = conversation.as_ref().is_some_and(|conversation| conversation.space.is_some()); - let (Some(guild), true) = (guild, in_tree) else { - // A direct conversation's members are its title; a group's are not kept anywhere. - return answer(reply, Err(Refusal::Unsupported)); - }; + let (Some(guild), true) = (guild, in_tree) else { return answer(reply, Err(Refusal::Unsupported)); }; if conversation.is_some_and(|conversation| conversation.kind == ConversationKind::Thread) { return answer(reply, Err(Refusal::Unsupported)); } - { + let (groups, members, pending) = { let mut state = self.state.lock().unwrap(); - if !state.guilds.contains_key(&guild) { - return answer(reply, Err(Refusal::Unsupported)); - } + if !state.guilds.contains_key(&guild) { return answer(reply, Err(Refusal::Unsupported)); } let watched = &mut state.members; + if watched.seen.members.is_some() && watched.list.as_deref() != Some(&channel) { + // A delayed query from a previous conversation cannot take over + // a viewport already selected by Watch. + return answer(reply, Ok(Members { pending: true, sections: Vec::new(), members: Vec::new() })); + } + let opening = watched.list.as_deref() != Some(&channel); watched.show(Some(&guild)); watched.list(Some(&channel)); - watched.ranges = ranges(watched.roster.as_ref(), offset, limit.clamp(1, WINDOW)); - // An earlier request still waiting is superseded; its sender sees the channel close. - watched.asked = Some(Asked { - offset, - limit: limit.clamp(1, WINDOW), - reply, - since: Instant::now(), - corrected: false, - }); - } - // Rows already held answer at once; otherwise Discord is asked and its update settles it. - self.settle(); - let waiting = self.state.lock().unwrap().members.asked.is_some(); - self.subscribe(waiting); + // Bootstrap the first window before its sections make a viewport possible. + // Subsequent cache queries must never move the subscribed window. + if opening { watched.ranges = ranges(watched.roster.as_ref(), offset, limit.clamp(1, WINDOW)); } + let groups = watched.roster.as_ref().map_or_else(Vec::new, |roster| roster.groups.clone()); + let found = watched.roster.as_ref().and_then(|roster| roster.window(offset, limit.clamp(1, WINDOW))); + let pending = found.is_none(); + (groups, found.unwrap_or_default(), pending) + }; + self.sync_subscription(); + let roster = Roster { channel, groups, rows: BTreeMap::new() }; + answer(reply, Ok(Members { pending, sections: self.sections(&guild, &roster), members })); } fn person(&self, member: &Member, files: &mut Vec<(MediaFile, String)>) -> Option<(Person, Presence, String)> { @@ -462,6 +430,7 @@ impl Discord { let mut files = Vec::new(); let mut seen = Vec::new(); let shown; + let changed; let mut entry = |item: &ListItem| match item.member.as_ref().and_then(|member| self.person(member, &mut files)) { Some((person, presence, status)) => { seen.push((person.key.id.clone(), presence, status)); @@ -497,6 +466,7 @@ impl Discord { return; } shown = channel.clone(); + changed = update.changes_roster(&roster.groups); roster.groups = update.groups.iter().map(|group| (group.id.clone(), group.count)).collect(); for (op, range, index, mut entries) in ops { match (op.as_str(), range, index) { @@ -543,28 +513,12 @@ impl Discord { for (id, presence, status) in seen { self.ctx.presence(&id, presence, &status, false); } - self.settle(); - // Rows were added, removed or moved: the shell asks again for the ones on screen. - if !update.ops.is_empty() { + // Wake cache readers when their subscribed data arrives, as well as on edits. + if changed || update.ops.iter().any(|op| op.op == "SYNC") { self.ctx.members_changed(&shown); } } - /// Gives up on a `Members` request Discord never answered. True while one is waiting. - pub(super) fn members_tick(&self) -> bool { - let mut state = self.state.lock().unwrap(); - let late = state.members.asked.as_ref().is_some_and(|asked| asked.since.elapsed() > PATIENCE); - if late { - if let Some(asked) = state.members.asked.take() { - drop(state); - let why = "Discord did not send the member list.".to_owned(); - answer(asked.reply, Err(Refusal::Refused(why))); - return false; - } - } - state.members.asked.is_some() - } - /// Members of the server on screen whose name starts with what the user typed. The result /// arrives as a chunk and goes to People; at most 25. Discord has no directory outside a /// server, so with no server on screen there is nothing to ask. @@ -666,3 +620,33 @@ impl Discord { Ok(()) } } + +#[cfg(test)] +mod paging_tests { + use super::*; + + #[test] + fn requested_windows_do_not_invalidate_the_pages_they_deliver() { + let mut update: ListUpdate = serde_json::from_value(json!({ + "guild_id": "guild", "groups": [{"id": "online", "count": 200}], + "ops": [{"op": "SYNC", "range": [100, 199], "items": []}] + })).unwrap(); + let groups = vec![("online".to_owned(), 200)]; + assert!(!update.changes_roster(&groups)); + // Every page in a multi-page viewport must be able to land without + // forcing the reader back to the first page on each delivery. + for range in [(0, 99), (100, 199), (200, 299)] { + update.ops[0].range = Some(range); + assert!(!update.changes_roster(&groups)); + } + update.ops[0].op = "INVALIDATE".to_owned(); + assert!(!update.changes_roster(&groups), "discarding an old window must not restart visible reads"); + update.groups[0].count = 201; + assert!(update.changes_roster(&groups), "section counts still invalidate cached offsets"); + update.groups[0].count = 200; + for op in ["UPDATE", "INSERT", "DELETE"] { + update.ops[0].op = op.to_owned(); + assert!(update.changes_roster(&groups), "{op} must refresh cached members"); + } + } +} diff --git a/crates/chatd/src/adapters/mod.rs b/crates/chatd/src/adapters/mod.rs index b027d49..77c84bd 100644 --- a/crates/chatd/src/adapters/mod.rs +++ b/crates/chatd/src/adapters/mod.rs @@ -65,6 +65,7 @@ pub struct Download { /// One window of a member list. #[derive(Debug, Clone)] pub struct Members { + pub pending: bool, pub sections: Vec, pub members: Vec, } diff --git a/crates/chatd/src/adapters/slack/people.rs b/crates/chatd/src/adapters/slack/people.rs index 20f909a..a8e28d9 100644 --- a/crates/chatd/src/adapters/slack/people.rs +++ b/crates/chatd/src/adapters/slack/people.rs @@ -468,6 +468,7 @@ impl Slack { self.users(&window); let members: Vec = window.iter().map(|id| self.person(id, &self.user(id))).collect(); Ok(Members { + pending: false, // Slack's member list has no sections. sections: vec![MemberSection { name: "Members".to_owned(), diff --git a/crates/chatd/src/adapters/teams/people.rs b/crates/chatd/src/adapters/teams/people.rs index aa0828d..aadc04f 100644 --- a/crates/chatd/src/adapters/teams/people.rs +++ b/crates/chatd/src/adapters/teams/people.rs @@ -276,6 +276,7 @@ impl Teams { }) .collect(); Ok(Members { + pending: false, // Teams' roster has no sections. sections: vec![MemberSection { name: "Members".to_owned(), diff --git a/crates/chatd/src/adapters/telegram.rs b/crates/chatd/src/adapters/telegram.rs index 9a67d2c..25084f6 100644 --- a/crates/chatd/src/adapters/telegram.rs +++ b/crates/chatd/src/adapters/telegram.rs @@ -801,6 +801,7 @@ fn members(account: &Account, conversation: &str, offset: u32, limit: u32) -> Re } }; Ok(Members { + pending: false, sections: vec![MemberSection { name: "Members".to_owned(), count: total }], members: ids.into_iter().map(|id| person(account, Who::User(id), relation(account, id))).collect(), }) diff --git a/crates/chatd/src/demo/adapter.rs b/crates/chatd/src/demo/adapter.rs index a97618b..d83e4d9 100644 --- a/crates/chatd/src/demo/adapter.rs +++ b/crates/chatd/src/demo/adapter.rs @@ -464,7 +464,7 @@ impl Demo { MemberSection { name: "Offline".to_owned(), count: people.len() as u32 - online }, ]; let members = people.into_iter().skip(offset as usize).take(limit as usize).collect(); - Ok(Members { sections, members }) + Ok(Members { pending: false, sections, members }) } fn start_direct(&self, person_id: &str) -> Result { diff --git a/crates/chatd/src/server.rs b/crates/chatd/src/server.rs index 13c1694..17b0dd9 100644 --- a/crates/chatd/src/server.rs +++ b/crates/chatd/src/server.rs @@ -1487,6 +1487,7 @@ impl Chat { }); match asked { Ok(found) => Response::Members { + pending: found.pending, conversation, sections: found.sections, offset, diff --git a/crates/chatd/tests/discord.rs b/crates/chatd/tests/discord.rs index 81df53b..7fdb491 100644 --- a/crates/chatd/tests/discord.rs +++ b/crates/chatd/tests/discord.rs @@ -1302,15 +1302,12 @@ fn the_space_on_screen_is_the_only_one_subscribed() { sleep(Duration::from_millis(300)); assert_eq!(subscriptions().len(), 1, "looking again changes nothing"); - // The visible rows of a member list: asked for by range, answered from Discord's sync. - let asking = { - let (dir, conversation) = (world.dir.clone(), world.key("4101")); - std::thread::spawn(move || { - let mut stream = ChatStream::connect(&dir.join("chat.sock")).unwrap(); - write_frame(&mut stream, &Request::Members { conversation, offset: 0, limit: 50 }).unwrap(); - read_frame::(&mut stream).unwrap() - }) - }; + // A missing window returns immediately; no network request holds the caller. + let mut stream = ChatStream::connect(&world.dir.join("chat.sock")).unwrap(); + stream.set_read_timeout(Some(Duration::from_secs(1))).unwrap(); + write_frame(&mut stream, &Request::Members { conversation: world.key("4101"), offset: 0, limit: 50 }).unwrap(); + let pending: Response = read_frame(&mut stream).unwrap(); + assert!(matches!(pending, Response::Members { pending: true, ref members, .. } if members.is_empty()), "{pending:?}"); let ranged = wait("the range request", || subscriptions().pop().filter(|frame| frame["d"]["subscriptions"]["4000"]["channels"]["4101"].is_array())); assert_eq!(ranged["d"]["subscriptions"]["4000"]["channels"]["4101"], json!([[0, 99]])); let member = |id: &str, username: &str, name: &str, nick: Value, status: &str| { @@ -1335,7 +1332,11 @@ fn the_space_on_screen_is_the_only_one_subscribed() { member("2002", "dev", "Dev Patel", Value::Null, "offline"), ]}], })); - match asking.join().unwrap() { + let delivered = wait("the cached member window", || { + let response = world.ask(Request::Members { conversation: world.key("4101"), offset: 0, limit: 50 }); + matches!(response, Response::Members { pending: false, .. }).then_some(response) + }); + match delivered { Response::Members { sections, members, offset, .. } => { let sections: Vec<(&str, u32)> = sections.iter().map(|section| (section.name.as_str(), section.count)).collect(); assert_eq!(sections, [("mods", 1), ("Online", 1), ("Offline", 1)]); @@ -1354,6 +1355,15 @@ fn the_space_on_screen_is_the_only_one_subscribed() { other => panic!("{other:?}"), } assert_eq!(subscriptions().len(), asked); + // Two uncached reads do not replace the live viewport or wait behind one another. + let before = subscriptions().len(); + for offset in [500, 1000, 1500] { + let response = world.ask(Request::Members { conversation: world.key("4101"), offset, limit: 50 }); + // Outside this tiny roster is a valid empty page; either way no gateway send. + assert!(matches!(response, Response::Members { .. })); + } + sleep(Duration::from_millis(100)); + assert_eq!(subscriptions().len(), before, "cache reads must not move the live subscription"); // A direct conversation has no member list. let none = world.ask(Request::Members { conversation: world.key("3001"), offset: 0, limit: 50 }); assert!(matches!(none, Response::Error { code: ErrorCode::Unsupported, .. }), "{none:?}"); @@ -1397,6 +1407,57 @@ fn the_space_on_screen_is_the_only_one_subscribed() { assert_eq!(closed["d"]["subscriptions"]["4000"]["channels"], json!({})); } +#[test] +fn missing_member_pages_never_block_a_new_viewport_or_move_its_subscription() { + let world = World::start(); + world.tree("4000"); + let cached = |offset| { + let mut stream = ChatStream::connect(&world.dir.join("chat.sock")).unwrap(); + stream.set_read_timeout(Some(Duration::from_secs(1))).unwrap(); + write_frame(&mut stream, &Request::Members { conversation: world.key("4101"), offset, limit: 50 }).unwrap(); + read_frame::(&mut stream).unwrap() + }; + assert!(matches!(cached(0), Response::Members { pending: true, .. })); + let groups = json!([{"id": "online", "count": 4000}]); + let member = |row: u32| json!({"member": {"user": { + "id": (10000 + row).to_string(), "username": format!("member-{row}"), + "discriminator": "0", "avatar": null + }, "roles": []}}); + let mut first = vec![json!({"group": {"id": "online", "count": 4000}})]; + first.extend((1..100).map(member)); + world.discord.push("GUILD_MEMBER_LIST_UPDATE", json!({ + "guild_id": "4000", "groups": groups, + "ops": [{"op": "SYNC", "range": [0, 99], "items": first}] + })); + wait("the initial cache", || matches!(cached(0), Response::Members { pending: false, .. }).then_some(())); + let before = world.discord.frames_of(37).len(); + for offset in [500, 1000, 1500] { assert!(matches!(cached(offset), Response::Members { pending: true, .. })); } + sleep(Duration::from_millis(100)); + assert_eq!(world.discord.frames_of(37).len(), before, "misses must not enqueue network reads"); + + let mut watch = Watching { space: Some(world.key("4000")), conversation: Some(world.key("4101")), members: Some((1200, 50)), ..Watching::default() }; + world.ask(Request::Watch(watch.clone())); + wait("the latest viewport", || world.discord.frames_of(37).pop().filter(|frame| + frame["d"]["subscriptions"]["4000"]["channels"]["4101"] == json!([[0, 99], [1200, 1299]]))); + world.discord.push("GUILD_MEMBER_LIST_UPDATE", json!({ + "guild_id": "4000", "groups": groups, + "ops": [{"op": "SYNC", "range": [1200, 1299], "items": (1200..1300).map(member).collect::>()}] + })); + let ready = wait("the new viewport cache", || { + let response = cached(1200); + matches!(response, Response::Members { pending: false, .. }).then_some(response) + }); + let Response::Members { members, .. } = ready else { unreachable!() }; + assert_eq!(members.len(), 50); + assert!(matches!(cached(500), Response::Members { pending: true, .. }), "old pending page need not finish first"); + let before = world.discord.frames_of(37).len(); + watch.people = members.iter().map(|person| person.key.clone()).collect(); + world.ask(Request::Watch(watch)); + for _ in 0..20 { assert!(matches!(cached(1200), Response::Members { pending: false, .. })); } + sleep(Duration::from_millis(100)); + assert_eq!(world.discord.frames_of(37).len(), before, "cached reads and listed presence must not spend subscription budget"); +} + #[test] fn servers_channels_people_and_settings_follow_live_events() { let world = World::start(); diff --git a/crates/protocol/src/lib.rs b/crates/protocol/src/lib.rs index 2286437..8d598df 100644 --- a/crates/protocol/src/lib.rs +++ b/crates/protocol/src/lib.rs @@ -1488,6 +1488,9 @@ pub enum Response { total: u32, }, Members { + /// Discord cache miss: rows will arrive through the active viewport subscription. + #[serde(default, skip_serializing_if = "is_false")] + pending: bool, conversation: Key, sections: Vec, offset: u32, @@ -1665,6 +1668,20 @@ pub fn media_process_needed(active_calls: usize) -> bool { #[cfg(test)] mod tests { + #[test] + fn members_pending_is_explicit_and_old_complete_responses_still_decode() { + let mut value = serde_json::json!({"op": "members", "conversation": { + "network": "discord", "account": "account", "id": "channel" + }, "offset": 0, "sections": [], "members": []}); + let complete: super::Response = serde_json::from_value(value.clone()).unwrap(); + assert!(matches!(complete, super::Response::Members { pending: false, .. })); + assert!(serde_json::to_value(complete).unwrap().get("pending").is_none()); + value["pending"] = serde_json::json!(true); + let pending: super::Response = serde_json::from_value(value).unwrap(); + assert!(matches!(pending, super::Response::Members { pending: true, .. })); + assert_eq!(serde_json::to_value(pending).unwrap()["pending"], true); + } + use super::*; fn row(kind: ConversationKind, unread: u32, mentions: u32, muted: bool) -> Conversation { diff --git a/crates/tinyclient/src/gui/people.rs b/crates/tinyclient/src/gui/people.rs index 86f2c3e..7a0b13a 100644 --- a/crates/tinyclient/src/gui/people.rs +++ b/crates/tinyclient/src/gui/people.rs @@ -420,6 +420,9 @@ impl Shell { if self.right_kind() != Some(RightKind::Members) { hub.watch(members::SOURCE, [], cx); hub.watch_members(None, cx); + if let Some(members) = self.people_state.members.get() { + members.update(cx, |members, _| members.pause()); + } } if self.right_kind() != Some(RightKind::Info) { if let Some(info) = self.people_state.info.get().filter(|info| info.read(cx).person().is_some()) { diff --git a/crates/tinyclient/src/gui/people/hub.rs b/crates/tinyclient/src/gui/people/hub.rs index 71485d9..7e0d117 100644 --- a/crates/tinyclient/src/gui/people/hub.rs +++ b/crates/tinyclient/src/gui/people/hub.rs @@ -186,8 +186,11 @@ impl WatchSet { } let quiet = now.saturating_duration_since(last); let waited = now.saturating_duration_since(first); - if quiet < QUIET && waited < LATEST { - return Due::Wait((QUIET - quiet).min(LATEST - waited)); + let viewport_changed = self.sent.as_ref().map_or(wanted.members.is_some(), |sent| sent.members != wanted.members); + let quiet_limit = if viewport_changed { Duration::from_millis(16) } else { QUIET }; + let latest_limit = if viewport_changed { Duration::from_millis(50) } else { LATEST }; + if quiet < quiet_limit && waited < latest_limit { + return Due::Wait((quiet_limit - quiet).min(latest_limit - waited)); } self.changing = None; self.sent = Some(wanted.clone()); @@ -577,6 +580,19 @@ fn live(typists: &[Typist], heard: i64, now: i64) -> impl Iterator) { let this = cx.weak_entity(); - let request = Request::Members { conversation, offset: 0, limit: facts::INLINE_MEMBERS }; + let request = Request::Members { conversation: conversation.clone(), offset: 0, limit: facts::INLINE_MEMBERS }; super::ask_later(cx, request, move |answer, cx| { - let Ok(Response::Members { members, .. }) = answer else { + let Ok(Response::Members { members, pending, .. }) = answer else { return; }; let _ = this.update(cx, |this, cx| { if this.asked == asked { + if pending { + cx.spawn(async move |this, cx| { + cx.background_executor().timer(std::time::Duration::from_millis(250)).await; + let _ = this.update(cx, |this, cx| { + if this.asked == asked { this.ask_members(conversation, asked, cx); } + }); + }).detach(); + return; + } this.hub.learn(&members); this.hub.watch("info", members.iter().map(|person| person.key.clone()).collect::>(), cx); this.members = members; diff --git a/crates/tinyclient/src/gui/people/members.rs b/crates/tinyclient/src/gui/people/members.rs index 0384d18..f25c09c 100644 --- a/crates/tinyclient/src/gui/people/members.rs +++ b/crates/tinyclient/src/gui/people/members.rs @@ -1,10 +1,10 @@ //! The member list of the open conversation, in the right pane. Only the rows on screen are -//! ever asked for: the sections and their counts say how tall the list is, and each visible -//! stretch is one `Members` page. +//! served from a bounded page cache; missing visible ranges wait for scrolling +//! to settle and use one backend request at a time. use super::presence::Ring; use super::{in_call, person_avatar, shown_presence, Hub, Scene}; -use crate::gui::components::{empty_state, skeleton}; +use crate::gui::components::{button, empty_state, skeleton, ButtonKind}; use super::info::Subject; use crate::gui::pane::Layer; use crate::gui::state::Shell; @@ -17,12 +17,13 @@ use protocol::{Key, MemberSection, Person, Presence, Request, Response}; use std::collections::{HashMap, HashSet}; use std::ops::Range; use std::rc::Rc; +use std::time::{Duration, Instant}; pub const SOURCE: &str = "members"; const ROW: f32 = theme::ROW_ONE; const PAGE: u32 = 50; /// Pages of members held; the ones farthest from the viewport go first. -const PAGES_KEPT: usize = 8; +const PAGES_KEPT: usize = 64; /// Placeholder rows before the first answer says how many there are. const WAITING_ROWS: usize = 12; @@ -56,6 +57,82 @@ pub fn slot(sections: &[MemberSection], ix: usize) -> Option { None } +/// One immediate cache query at a time; the newest viewport replaces obsolete queued work. +#[derive(Default)] +struct PageRequests { + wanted: Vec, + ready: HashSet, + stale: HashSet, + waiting: HashSet, + failed: HashSet, + attempts: HashMap, + active: Option<(u64, u32)>, +} + +impl PageRequests { + fn viewport(&mut self, first: u32, last: u32, total: u32) -> bool { + // Discord already delivers 100-row windows around each 50-member read. + // Speculative neighboring reads can displace those live windows and make + // every roster update resubscribe back and forth without any scrolling. + let wanted: Vec = (first..=last).filter(|page| page.saturating_mul(PAGE) < total).take(PAGES_KEPT).collect(); + if self.wanted == wanted { return false; } + self.wanted = wanted; + self.waiting.retain(|page| self.wanted.contains(page)); + self.failed.retain(|page| self.wanted.contains(page)); + self.attempts.retain(|page, _| self.wanted.contains(page)); + true + } + + fn next(&mut self, epoch: u64) -> Option { + if self.active.is_some() { return None; } + // Fetch holes before refreshing retained rows, even during frequent roster edits. + let eligible = |page: &u32| !self.failed.contains(page) && !self.waiting.contains(page); + let page = self.wanted.iter().copied().find(|page| !self.ready.contains(page) && eligible(page)) + .or_else(|| self.wanted.iter().copied().find(|page| self.stale.contains(page) && eligible(page)))?; + self.active = Some((epoch, page)); + Some(page) + } + + fn pending(&mut self, epoch: u64, page: u32, current: u64) -> bool { + if self.active != Some((epoch, page)) { return false; } + self.active = None; + if epoch != current { return false; } + if self.wanted.contains(&page) { self.waiting.insert(page); } + true + } + + fn complete(&mut self, epoch: u64, page: u32, current: u64, success: bool) -> bool { + if self.active != Some((epoch, page)) { return false; } + self.active = None; + if epoch != current { return false; } + if success { + self.ready.insert(page); + self.stale.remove(&page); + self.attempts.remove(&page); + } else if self.wanted.contains(&page) { + self.failed.insert(page); + *self.attempts.entry(page).or_default() += 1; + } + true + } + + fn retry_delay(&self, page: u32) -> Duration { + let attempts = self.attempts.get(&page).copied().unwrap_or(1); + Duration::from_secs(1 << attempts.min(5)) + } + + fn reset(&mut self) { + // Reads are independent cache queries; navigation never waits for an old reply. + self.active = None; + self.wanted = vec![0]; + self.ready.clear(); + self.stale.clear(); + self.waiting.clear(); + self.failed.clear(); + self.attempts.clear(); + } +} + pub struct MemberList { shell: WeakEntity, hub: Hub, @@ -63,8 +140,13 @@ pub struct MemberList { conversation: Option, sections: Vec, pages: HashMap>>, - asked: HashSet, + requests: PageRequests, + errors: HashMap, + viewport_tick: u64, + settled: bool, + scroll_task: Option>, loaded: bool, + waiting_since: Option, refused: Option, epoch: u64, scroll: UniformListScrollHandle, @@ -79,8 +161,13 @@ impl MemberList { conversation: None, sections: Vec::new(), pages: HashMap::new(), - asked: HashSet::new(), + requests: PageRequests::default(), + errors: HashMap::new(), + viewport_tick: 0, + settled: true, + scroll_task: None, loaded: false, + waiting_since: None, refused: None, epoch: 0, scroll: UniformListScrollHandle::new(), @@ -96,19 +183,43 @@ impl MemberList { pub fn see(&mut self, scene: Scene, cx: &mut Context) { let moved = self.conversation != scene.open; self.scene = scene; - if moved { + if moved || self.requests.wanted.is_empty() { self.conversation = self.scene.open.clone(); - self.reset(); - self.want(0, cx); + if moved { self.reset(); } else { self.requests.wanted = vec![0]; } + // Declare the initial viewport before rows exist, so an unrelated + // presence watch cannot unsubscribe the bootstrap window. + self.hub.watch_members(self.conversation.as_ref().map(|_| (0, PAGE)), cx); + self.pump(cx); } } + pub fn pause(&mut self) { + if self.requests.wanted.is_empty() { return; } + self.epoch += 1; + self.requests.active = None; + self.requests.wanted.clear(); + self.requests.waiting.clear(); + self.requests.failed.clear(); + self.requests.attempts.clear(); + self.errors.clear(); + self.refused = None; + self.waiting_since = None; + self.viewport_tick += 1; + self.scroll_task = None; + self.settled = true; + } + fn reset(&mut self) { self.epoch += 1; self.sections.clear(); self.pages.clear(); - self.asked.clear(); + self.requests.reset(); + self.errors.clear(); + self.viewport_tick += 1; + self.scroll_task = None; + self.settled = true; self.loaded = false; + self.waiting_since = None; self.refused = None; } @@ -119,59 +230,142 @@ impl MemberList { } } - /// Asks again for the pages held, which are the ones around the viewport. The rows stay - /// as they are until the answers land. + /// Refreshes only the latest viewport against retained rows, one request at a time. fn read_again(&mut self, cx: &mut Context) { - if !self.loaded { - return; - } - self.epoch += 1; - let pages: Vec = self.asked.drain().collect(); - for page in pages { - self.want(page, cx); - } + if !self.loaded && self.requests.waiting.is_empty() { return; } + self.requests.waiting.clear(); + // Retain the cached rows; refresh only the latest viewport and its neighbors. + self.requests.stale.extend(self.requests.ready.iter().copied()); + self.pump(cx); } - fn want(&mut self, page: u32, cx: &mut Context) { - let Some(conversation) = self.conversation.clone() else { - return; - }; - if !self.asked.insert(page) { - return; - } + fn viewport(&mut self, first: u32, last: u32, cx: &mut Context) { + let total = self.sections.iter().map(|section| section.count).sum(); + if !self.requests.viewport(first, last, total) { return; } + let errors = self.errors.len(); + self.errors.retain(|page, _| self.requests.failed.contains(page)); + if self.errors.len() != errors { cx.notify(); } + self.viewport_tick += 1; + let tick = self.viewport_tick; + self.settled = false; + self.scroll_task = Some(cx.spawn(async move |this, cx| { + cx.background_executor().timer(Duration::from_millis(16)).await; + let _ = this.update(cx, |this, cx| { + if this.viewport_tick == tick { + this.settled = true; + this.pump(cx); + } + }); + })); + } + + fn pump(&mut self, cx: &mut Context) { + if !self.settled { return; } + let Some(conversation) = self.conversation.clone() else { return; }; + let Some(page) = self.requests.next(self.epoch) else { return; }; let request = Request::Members { conversation, offset: page * PAGE, limit: PAGE }; let (epoch, this) = (self.epoch, cx.weak_entity()); super::ask_later(cx, request, move |answer, cx| { let _ = this.update(cx, |this, cx| { - if this.epoch == epoch { + let pending = matches!(&answer, Ok(Response::Members { pending: true, .. })); + let success = matches!(&answer, Ok(Response::Members { pending: false, .. })); + let accepted = if pending { this.requests.pending(epoch, page, this.epoch) } + else { this.requests.complete(epoch, page, this.epoch, success) }; + if accepted { this.landed(page, answer, cx); + } else { + this.pump(cx); } }); }); } fn landed(&mut self, page: u32, answer: std::io::Result, cx: &mut Context) { + if let Ok(Response::Members { pending: true, sections, .. }) = &answer { + self.waiting_since.get_or_insert_with(Instant::now); + if !sections.is_empty() { self.sections = sections.clone(); self.loaded = true; } + // A cache miss never holds up the next viewport. SYNC notifications + // normally wake it; a cheap cache poll also covers missed notifications. + let epoch = self.epoch; + cx.spawn(async move |this, cx| { + cx.background_executor().timer(Duration::from_millis(250)).await; + let _ = this.update(cx, |this, cx| { + if this.epoch == epoch && this.requests.wanted.contains(&page) && this.requests.waiting.remove(&page) { + this.pump(cx); + } + }); + }).detach(); + self.pump(cx); + cx.notify(); + return; + } + let retryable = match &answer { + Err(err) => crate::client::is_slow(err) || crate::client::is_gone(err), + Ok(Response::Error { code: protocol::ErrorCode::Offline, .. }) => true, + Ok(Response::Error { detail, .. }) => detail == "Discord did not send the member list.", + _ => false, + }; match answer { Ok(Response::Members { sections, members, .. }) => { self.hub.learn(&members); self.sections = sections; self.pages.insert(page, members.into_iter().map(Rc::new).collect()); self.refused = None; + self.errors.remove(&page); while self.pages.len() > PAGES_KEPT { - let farthest = self.pages.keys().copied().max_by_key(|held| held.abs_diff(page)); + let anchor = self.requests.wanted.first().copied().unwrap_or(page); + let farthest = self.pages.keys().copied().max_by_key(|held| (!self.requests.wanted.contains(held), held.abs_diff(anchor))); let Some(farthest) = farthest else { break; }; self.pages.remove(&farthest); - self.asked.remove(&farthest); + self.requests.ready.remove(&farthest); + self.requests.stale.remove(&farthest); } } - Ok(Response::Error { detail, .. }) => self.refused = Some(detail), - Ok(_) => self.refused = Some(crate::client::OTHER_ANSWER.to_owned()), - Err(err) => self.refused = Some(crate::client::trouble_words(&err)), + Ok(Response::Error { detail, .. }) => self.failed(page, detail), + Ok(_) => self.failed(page, crate::client::OTHER_ANSWER.to_owned()), + Err(err) => self.failed(page, crate::client::trouble_words(&err)), + } + if retryable && self.requests.failed.contains(&page) { + // Transient gateway congestion must not leave a stationary viewport + // blocked forever. One delayed attempt per failed read, never per render. + let epoch = self.epoch; + let delay = self.requests.retry_delay(page); + cx.spawn(async move |this, cx| { + cx.background_executor().timer(delay).await; + let _ = this.update(cx, |this, cx| { + if this.epoch == epoch && this.requests.wanted.contains(&page) && this.requests.failed.remove(&page) { + this.pump(cx); + } + }); + }).detach(); + } + if self.requests.wanted.iter().all(|page| self.requests.ready.contains(page)) || !self.errors.is_empty() { + self.waiting_since = None; } self.loaded = true; // The counts live in the section headers, never beside the pane's title. + self.pump(cx); + cx.notify(); + } + + fn failed(&mut self, page: u32, detail: String) { + if self.sections.is_empty() { + self.refused = Some(detail.clone()); + } + if self.requests.wanted.contains(&page) { + self.errors.insert(page, detail); + } + } + + fn retry(&mut self, cx: &mut Context) { + self.requests.failed.clear(); + self.errors.clear(); + self.refused = None; + self.loaded = !self.sections.is_empty(); + self.settled = true; + self.pump(cx); cx.notify(); } @@ -190,9 +384,7 @@ impl MemberList { .collect(); if let (Some(first), Some(last)) = (offsets.first().copied(), offsets.last().copied()) { self.hub.watch_members(Some((first, last - first + 1)), cx); - for page in first / PAGE..=last / PAGE { - self.want(page, cx); - } + self.viewport(first / PAGE, last / PAGE, cx); } let on_screen = offsets.iter().filter_map(|offset| self.member(*offset)).filter(|person| !person.bot); self.hub.watch(SOURCE, on_screen.map(|person| person.key.clone()).collect::>(), cx); @@ -281,9 +473,13 @@ fn waiting_row() -> Div { impl Render for MemberList { fn render(&mut self, _: &mut Window, cx: &mut Context) -> impl IntoElement { - let body = div().size_full().pt(px(4.)); + let body = div().size_full().pt(px(4.)).flex().flex_col().when( + self.waiting_since.is_some_and(|at| at.elapsed() >= Duration::from_secs(1)) && !self.requests.waiting.is_empty(), + |body| body.child(div().flex_none().px(px(12.)).py(px(4.)).type_small().text_color(theme::text_soft()) + .child("Waiting for Discord to send these members…")), + ); if let Some(refused) = &self.refused { - return body.child(empty_state("icons/users.svg", "No member list", refused).px(px(16.))); + return body.child(empty_state("icons/users.svg", "Couldn’t load members", refused).px(px(16.))).child(button("retry-members", ButtonKind::Ghost, "Try again").on_click(cx.listener(|this, _, _, cx| this.retry(cx)))); } if !self.loaded { return body.children((0..WAITING_ROWS).map(|_| waiting_row())); @@ -292,7 +488,11 @@ impl Render for MemberList { if rows == 0 { return body.child(empty_state("icons/users.svg", "Nobody here", "This conversation has no members to show.").px(px(16.))); } - body.child(uniform_list("members", rows, cx.processor(Self::rows)).track_scroll(&self.scroll).size_full()) + body.when(!self.errors.is_empty(), |body| body.child( + div().flex_none().px(px(12.)).flex().items_center().gap(px(8.)) + .child(div().flex_1().type_small().text_color(theme::text_soft()).child("Some members couldn’t load")) + .child(button("retry-member-pages", ButtonKind::Ghost, "Retry").on_click(cx.listener(|this, _, _, cx| this.retry(cx)))) + )).child(div().flex_1().min_h_0().child(uniform_list("members", rows, cx.processor(Self::rows)).track_scroll(&self.scroll).size_full())) } } @@ -309,7 +509,7 @@ impl MemberList { self.hub.learn(&all); for (page, people) in all.chunks(PAGE as usize).enumerate() { self.pages.insert(page as u32, people.iter().cloned().map(Rc::new).collect()); - self.asked.insert(page as u32); + self.requests.ready.insert(page as u32); } self.loaded = true; cx.notify(); @@ -366,6 +566,132 @@ mod tests { counts.iter().map(|count| MemberSection { name: "Role".into(), count: *count }).collect() } + #[test] + fn a_missing_discord_window_does_not_hold_the_next_viewport_hostage() { + let mut requests = PageRequests::default(); + requests.viewport(23, 24, 4000); + assert_eq!(requests.next(1), Some(23)); + assert!(requests.pending(1, 23, 1)); + assert_eq!(requests.next(1), Some(24)); + assert!(requests.pending(1, 24, 1)); + assert_eq!(requests.next(1), None, "wait for delivery rather than spin on cache misses"); + requests.viewport(40, 41, 4000); + assert_eq!(requests.next(1), Some(40), "no timeout or completion of old network reads needed"); + assert!(requests.pending(1, 40, 1)); + requests.waiting.clear(); // notification that the subscribed rows arrived + assert_eq!(requests.next(1), Some(40)); + assert!(requests.complete(1, 40, 1, true)); + } + + #[test] + fn busy_large_roster_refreshes_do_not_read_neighbors_that_displace_live_windows() { + let mut requests = PageRequests::default(); + // Captured viewport: members 1197..1246; five groups, about 3900 people. + // Pages 23 and 24 cover it. Reading neighbor 25 requests a third remote + // window, evicting the window containing page 23 from the subscription. + requests.viewport(23, 24, 3931); + assert_eq!(requests.wanted, [23, 24]); + for _ in 0..100 { + requests.stale.extend(requests.ready.iter().copied()); + for page in [23, 24] { + assert_eq!(requests.next(1), Some(page)); + assert!(requests.complete(1, page, 1, true)); + } + assert_eq!(requests.next(1), None, "no requests for off-screen pages 22 or 25"); + } + } + + #[test] + fn fast_scrolling_coalesces_missing_pages_behind_one_active_request() { + let mut requests = PageRequests::default(); + requests.viewport(0, 1, 10_000); + assert_eq!(requests.next(1), Some(0)); + requests.viewport(20, 21, 10_000); + requests.viewport(40, 41, 10_000); + assert_eq!(requests.next(1), None, "never supersede an active Discord reply"); + assert!(requests.complete(1, 0, 1, true)); + assert_eq!(requests.next(1), Some(40), "discard intermediate scroll ranges"); + assert!(requests.complete(1, 40, 1, true)); + assert_eq!(requests.next(1), Some(41), "continue with the other visible page"); + assert!(requests.complete(1, 41, 1, true)); + assert_eq!(requests.next(1), None); + } + + #[test] + fn transient_failures_can_recover_without_another_scroll_and_back_off() { + let mut requests = PageRequests::default(); + requests.viewport(40, 40, 10_000); + for seconds in [2, 4, 8, 16, 32, 32] { + assert_eq!(requests.next(1), Some(40)); + assert!(requests.complete(1, 40, 1, false)); + assert_eq!(requests.retry_delay(40), Duration::from_secs(seconds)); + assert!(requests.failed.remove(&40)); // delayed retry, same viewport + } + assert_eq!(requests.next(1), Some(40)); + assert!(requests.complete(1, 40, 1, true)); + assert!(requests.attempts.is_empty()); + assert_eq!(requests.next(1), None); + } + + #[test] + fn roster_updates_cannot_starve_missing_visible_pages() { + let mut requests = PageRequests::default(); + requests.viewport(40, 42, 10_000); + // Each window's arrival can coincide with a real presence/roster edit. + // Already loaded rows must remain lower priority than the next hole. + for page in [40, 41, 42] { + assert_eq!(requests.next(1), Some(page)); + assert!(requests.complete(1, page, 1, true)); + requests.stale.extend(requests.ready.iter().copied()); + } + assert!([40, 41, 42].iter().all(|page| requests.ready.contains(page))); + assert_eq!(requests.next(1), Some(40), "refresh cached rows only after filling holes"); + } + + #[test] + fn cached_pages_are_reused_and_failed_pages_do_not_create_retry_loops() { + let mut requests = PageRequests::default(); + requests.viewport(2, 2, 250); + requests.ready.extend([1, 2, 3]); + assert_eq!(requests.next(1), None); + requests.viewport(3, 3, 250); + assert_eq!(requests.next(1), None, "scrolling to a cached page needs no speculative reads"); + requests.viewport(4, 4, 250); + assert_eq!(requests.next(1), Some(4)); + assert!(requests.complete(1, 4, 1, false)); + assert_eq!(requests.next(1), None); + assert!(requests.ready.contains(&2) && requests.ready.contains(&3), "failure preserves cached pages"); + requests.failed.clear(); // explicit Retry + assert_eq!(requests.next(1), Some(4)); + } + + #[test] + fn old_conversation_replies_cannot_fill_the_new_conversations_cache() { + let mut requests = PageRequests::default(); + requests.reset(); + assert_eq!(requests.next(1), Some(0)); + requests.reset(); + assert_eq!(requests.next(2), Some(0), "navigation does not wait for the old query"); + assert!(!requests.complete(1, 0, 2, true)); + assert!(requests.ready.is_empty()); + assert_eq!(requests.next(2), None, "the new query remains active after the old reply"); + assert!(requests.complete(2, 0, 2, true)); + requests.stale.extend(requests.ready.iter().copied()); // refresh after a member update + assert_eq!(requests.next(2), Some(0)); + } + + #[test] + fn visible_requests_stay_inside_the_roster_and_the_cache_capacity() { + let mut requests = PageRequests::default(); + requests.viewport(0, 0, 50); + assert_eq!(requests.wanted, [0]); + requests.viewport(2, 2, 150); + assert_eq!(requests.wanted, [2]); + requests.viewport(0, 100, 10_000); + assert_eq!(requests.wanted.len(), PAGES_KEPT); + assert_eq!(requests.wanted, (0..PAGES_KEPT as u32).collect::>()); + } + #[test] fn rows_are_a_running_sum_of_the_section_counts() { let list = sections(&[2, 0, 3]); diff --git a/docs/architecture.md b/docs/architecture.md index f1cffae..fd281ef 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -41,7 +41,7 @@ How the processes fit together and who may edit what. The contract itself is the | `Spaces {network}`, `Tree {space, offset, limit, peek}` | `Spaces`; `Tree {groups (kind, count), conversations, total}` | store; `Command::OpenSpace` | | `Conversation`, `Collapse`, `Star`, `SetDraft` | `Conversation`; `Ok` | store (local only) | | `People {network, relation, query, offset, limit}` | `People {people, total}` | store; `Command::SearchPeople`, debounced | -| `Members`, `StartDirect`, `Handoff` | `Members`; `Conversation`; `Link` | adapter only (`Unsupported` until built) | +| `Members`, `StartDirect`, `Handoff` | `Members`; `Conversation`; `Link` | adapter only; Discord `Members.pending` returns a cache miss immediately, while `Watch.members` owns the live viewport | | `Profile {person, space}` | `Profile` (typed, kept 5 min) | `Command::ProfileIn`, falling back to `Command::Profile` | | `ConversationInfo {conversation}`, `SpaceInfo {space}` | `ConversationInfo {info}`; `SpaceInfo {info}` (kept 5 min) | `Command::ConversationInfo` / `SpaceInfo` | | `SetMute {conversation, notify}`, `Leave`, `CloseDirect` | `Ok` once the network accepted | `Command::SetMute` / `Leave` / `CloseDirect` | diff --git a/docs/design/polish/07-panes.md b/docs/design/polish/07-panes.md index c4f55f9..5705a5e 100644 --- a/docs/design/polish/07-panes.md +++ b/docs/design/polish/07-panes.md @@ -172,3 +172,45 @@ Glyphs: `info`, `user`, `users`, `messages-square`, `search`, `panel-left`, `ell - Auto-revealing the open row in the sidebar; reorder under the pointer; opening on arrow keys. - Slack's "jump to unread" banner at the top: landing at the divider already does it. - A second window, tabs inside the pane, a pane deeper than 2. + + +### Member viewport and cache + +The member pane keeps 64 pages of 50 members. Cached rows display immediately. +Only visible pages are read, one local cache query at a time; the latest viewport +replaces queued ranges after one frame (16 ms). Missing pages precede refreshes +of retained rows. Replies from a different conversation are discarded. Closing the pane cancels +pending polls without discarding cached rows; reopening resumes from that cache. + +Discord uses one live viewport subscription, driven by `Watch.members`. The +viewport rounds to the UI's 50-member pages and subscribes to Discord's own +100-row windows. Page queries never change that subscription or wait for the +network: `Members.pending: true` means the requested rows are not cached yet, +not an empty roster or a refusal. Pending reads do not block a later viewport. +Discord cache reads, viewport changes and viewport-bound member searches run +on their own ordered thread, +independent of history backfill sleeps and HTTP requests. Window `SYNC` and roster edits emit `Change::Members`; the GUI then reads the +cache again. A 250 ms cache poll covers missed notifications without sending +another subscription. Conversation details also handle pending previews. + +The GUI does not request speculative adjacent pages, which would displace live +windows and cause subscription ping-pong. Discord's windows already hold nearby +rows. Presence for members inside subscribed windows is not requested a second +time by individual ID. Viewport watches settle separately from the slower +presence watches, normally within one frame, and while changing within 50 ms +(though an already sleeping watch dispatcher can add up to 200 ms). + +Unsent gateway subscriptions coalesce by guild, preserving unsubscribe +transitions and only the newest window/presence state. Ready snapshots send +immediately; there is no artificial two-second timer or burst-credit budget. +The gateway keeps a rolling 100-frame/minute command budget, reserving 20 of +Discord's documented 120 for heartbeat/handshake traffic. Only exhaustion of +that actual budget delays queued snapshots. + +Cache eviction prefers pages outside the viewport. Pending data retains cached +rows; after one second the pane explains that Discord has not sent the requested +members yet. Transient connection/time-out failures retry while wanted, with +2/4/8/16/32-second capped backoff. Other refusals offer Retry and preserve the +loaded roster. No failure retries per render, and old-conversation timers have +no effect after navigation. GUI and daemon must both be updated for the pending +cache response contract.