sidecar/src/app.rs

@@ -14,11 +14,13 @@
 //! - `shutdown` is whatever "stop" means on this host: Ctrl-C and `SIGTERM` on Unix, the SCM's
 //!   `Stop` control on Windows.
 
+use std::collections::HashMap;
 use std::future::Future;
 use std::sync::atomic::AtomicI64;
 use std::sync::Arc;
 use std::time::Instant;
 
+use serde_json::Value;
 use tokio::sync::{broadcast, mpsc};
 use tracing::info;
 
@@ -88,6 +90,9 @@ where
     let last_event_ts = last_event.clone();
     let replay_handle = handle.clone(); // re-push external news to the shard on (re)connect
     let mut total: u64 = 0;
+    // Partly-received guild rosters, keyed by guild id. Lives in the event-loop task, so it needs
+    // no lock and dies with the loop. See `accumulate_roster`.
+    let mut roster_parts: HashMap<i64, RosterParts> = HashMap::new();
     tokio::spawn(async move {
         while let Some(ev) = event_rx.recv().await {
             // Any line from the shard — including pong heartbeats — is a sign of life.
@@ -168,22 +173,36 @@ where
                     // roster self-corrects on the next `guild.roster`, which the shard re-emits
                     // whenever the member set changes. Keeping the delta out of the board is what
                     // keeps the sidecar a forwarder rather than a thing that maintains state.
+                    //
+                    // A roster over the shard's per-line cap arrives as several frames, so it is
+                    // reassembled before it is stored — see `accumulate_roster` for why that happens
+                    // here rather than by appending to the column.
                     "guild.roster" => {
                         if let Some(id) = ev.value.get("id").and_then(|v| v.as_i64()) {
+                            let seq = ev.value.get("seq").and_then(|v| v.as_i64()).unwrap_or(0);
+                            let more = ev
+                                .value
+                                .get("more")
+                                .and_then(|v| v.as_bool())
+                                .unwrap_or(false);
                             let members = ev
                                 .value
                                 .get("members")
+                                .and_then(|m| m.as_array())
                                 .cloned()
-                                .unwrap_or_else(|| serde_json::json!([]));
+                                .unwrap_or_default();
 
-                            if let Err(e) = event_store
+                            if let Some(complete) =
-                                .upsert_guild_roster(id, &members.to_string(), t)
+                                accumulate_roster(&mut roster_parts, id, seq, more, members)
-                                .await
+                            {
+                                let json = serde_json::Value::Array(complete).to_string();
+                                if let Err(e) = event_store.upsert_guild_roster(id, &json, t).await
                                 {
                                     tracing::warn!(error = %e, "failed to upsert guild roster");
                                 }
                             }
                         }
+                    }
                     "guild.remove" => {
                         if let Some(id) = ev.value.get("id").and_then(|v| v.as_i64()) {
                             if let Err(e) = event_store.delete_guild(id).await {
@@ -302,6 +321,10 @@ where
             // with announce=false so a restart does not re-proclaim every article at once. news.add
             // is idempotent by id, so replaying to a still-populated shard is harmless.
             if ev.kind == "server.hello" {
+                // A (re)connected shard restarts every roster from `seq` 0, so any half-received
+                // one belongs to the previous connection and can never be completed.
+                roster_parts.clear();
+
                 match event_store.news_all().await {
                     Ok(items) => {
                         for mut item in items {
@@ -350,3 +373,199 @@ pub fn now_ms() -> i64 {
         .map(|d| d.as_millis() as i64)
         .unwrap_or(0)
 }
+
+/// A guild roster that has arrived in part: the `seq` expected next, and what has accumulated.
+struct RosterParts {
+    next_seq: i64,
+    members: Vec<Value>,
+}
+
+/// Refuses to accumulate a roster past this many members. The shard caps its own frames, so
+/// exceeding this means a shard that is buggy or not what it claims to be — and the one thing this
+/// buffer must not do is grow without bound on its say-so.
+const MAX_ROSTER_MEMBERS: usize = 50_000;
+
+/// Reassembles a `guild.roster` that the shard split across frames, returning the whole member list
+/// once the final frame arrives and `None` while one is still incomplete.
+///
+/// Reassembly happens **here, in memory, before the store** rather than by appending to the
+/// `members` column, for two reasons. Appending would make the write a read-modify-write — the exact
+/// thing the two-column board design exists to avoid — and it would publish a torn roster: a reader
+/// hitting `GET /guilds` between frames would see a partial member list as though it were the truth.
+/// Buffering keeps the store's write a single atomic upsert of a complete roster.
+///
+/// This is transport-level reassembly, not domain state: it is the same category of work as turning
+/// bytes into a line, and it holds nothing once a roster is complete. That is what keeps it
+/// compatible with the sidecar being a forwarder.
+///
+/// The ordinary case — a guild inside the shard's per-line cap, which is every realistic one —
+/// arrives as `seq` 0 with `more` false and is returned immediately without ever touching the map.
+fn accumulate_roster(
+    parts: &mut HashMap<i64, RosterParts>,
+    id: i64,
+    seq: i64,
+    more: bool,
+    members: Vec<Value>,
+) -> Option<Vec<Value>> {
+    if seq == 0 {
+        // A fresh roster supersedes any partial one: the shard restarts at 0 every time it emits,
+        // so a leftover buffer is from an emission that was interrupted and will never finish.
+        parts.remove(&id);
+
+        if !more {
+            return Some(members);
+        }
+
+        parts.insert(
+            id,
+            RosterParts {
+                next_seq: 1,
+                members,
+            },
+        );
+        return None;
+    }
+
+    let entry = match parts.get_mut(&id) {
+        Some(entry) => entry,
+        // A continuation with nothing to continue: the sidecar started, or the shard reconnected,
+        // midway through an emission. Dropping it is right — the next full roster is complete.
+        None => {
+            tracing::debug!(
+                guild = id,
+                seq,
+                "roster continuation with no start; ignoring"
+            );
+            return None;
+        }
+    };
+
+    if entry.next_seq != seq {
+        tracing::warn!(
+            guild = id,
+            expected = entry.next_seq,
+            got = seq,
+            "roster frames out of order; discarding the partial roster"
+        );
+        parts.remove(&id);
+        return None;
+    }
+
+    entry.members.extend(members);
+
+    if entry.members.len() > MAX_ROSTER_MEMBERS {
+        tracing::warn!(
+            guild = id,
+            len = entry.members.len(),
+            "roster exceeded the reassembly cap; discarding"
+        );
+        parts.remove(&id);
+        return None;
+    }
+
+    if more {
+        entry.next_seq = seq + 1;
+        return None;
+    }
+
+    parts.remove(&id).map(|done| done.members)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn members(names: &[&str]) -> Vec<Value> {
+        names
+            .iter()
+            .map(|n| serde_json::json!({"name": n}))
+            .collect()
+    }
+
+    fn names(vs: &[Value]) -> Vec<String> {
+        vs.iter()
+            .map(|v| v["name"].as_str().unwrap_or_default().to_string())
+            .collect()
+    }
+
+    #[test]
+    fn a_single_frame_roster_is_returned_immediately() {
+        // Every realistic guild takes this path, and it must not depend on the buffer at all.
+        let mut parts = HashMap::new();
+        let out = accumulate_roster(&mut parts, 1, 0, false, members(&["Ada", "Bo"]));
+
+        assert_eq!(names(&out.expect("complete")), ["Ada", "Bo"]);
+        assert!(parts.is_empty(), "nothing should be buffered");
+    }
+
+    #[test]
+    fn a_chunked_roster_reassembles_in_order() {
+        // The case the live rig caught: without this, only the final frame survived and a
+        // 155-member guild appeared on the board with 3 members.
+        let mut parts = HashMap::new();
+
+        assert!(accumulate_roster(&mut parts, 1, 0, true, members(&["Ada"])).is_none());
+        assert!(accumulate_roster(&mut parts, 1, 1, true, members(&["Bo"])).is_none());
+        let out = accumulate_roster(&mut parts, 1, 2, false, members(&["Cy"]));
+
+        assert_eq!(names(&out.expect("complete")), ["Ada", "Bo", "Cy"]);
+        assert!(parts.is_empty(), "buffer is released once complete");
+    }
+
+    #[test]
+    fn a_restarted_roster_supersedes_a_partial_one() {
+        // A shard that reconnects mid-emission starts again at seq 0. The abandoned frames must not
+        // end up spliced onto the front of the new roster.
+        let mut parts = HashMap::new();
+
+        assert!(accumulate_roster(&mut parts, 1, 0, true, members(&["Stale"])).is_none());
+        let out = accumulate_roster(&mut parts, 1, 0, false, members(&["Fresh"]));
+
+        assert_eq!(names(&out.expect("complete")), ["Fresh"]);
+    }
+
+    #[test]
+    fn an_out_of_order_frame_discards_the_partial_roster() {
+        // Better to publish nothing and wait for the next full emission than to store a roster with
+        // a hole in it that nothing downstream could detect.
+        let mut parts = HashMap::new();
+
+        assert!(accumulate_roster(&mut parts, 1, 0, true, members(&["Ada"])).is_none());
+        assert!(accumulate_roster(&mut parts, 1, 2, false, members(&["Skipped"])).is_none());
+        assert!(parts.is_empty());
+    }
+
+    #[test]
+    fn a_continuation_with_no_start_is_ignored() {
+        // The sidecar restarting midway through a shard's emission.
+        let mut parts = HashMap::new();
+
+        assert!(accumulate_roster(&mut parts, 1, 3, false, members(&["Orphan"])).is_none());
+        assert!(parts.is_empty());
+    }
+
+    #[test]
+    fn two_guilds_reassemble_independently() {
+        // Rosters for different guilds interleave freely — the sweep emits one guild after another
+        // and nothing serialises them on the wire.
+        let mut parts = HashMap::new();
+
+        assert!(accumulate_roster(&mut parts, 1, 0, true, members(&["A1"])).is_none());
+        assert!(accumulate_roster(&mut parts, 2, 0, true, members(&["B1"])).is_none());
+        let g2 = accumulate_roster(&mut parts, 2, 1, false, members(&["B2"]));
+        let g1 = accumulate_roster(&mut parts, 1, 1, false, members(&["A2"]));
+
+        assert_eq!(names(&g2.expect("guild 2")), ["B1", "B2"]);
+        assert_eq!(names(&g1.expect("guild 1")), ["A1", "A2"]);
+    }
+
+    #[test]
+    fn an_empty_roster_is_a_complete_roster() {
+        // A guild whose last member left emits one frame with an empty array. Treating that as
+        // "nothing to store" would leave the board showing the roster it had before.
+        let mut parts = HashMap::new();
+        let out = accumulate_roster(&mut parts, 1, 0, false, vec![]);
+
+        assert_eq!(out.expect("complete").len(), 0);
+    }
+}