diff --git a/sidecar/src/app.rs b/sidecar/src/app.rs index 5f3f1d8..74a2350 100644 --- a/sidecar/src/app.rs +++ b/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 = 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,19 +173,33 @@ 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 - .upsert_guild_roster(id, &members.to_string(), t) - .await + if let Some(complete) = + accumulate_roster(&mut roster_parts, id, seq, more, members) { - tracing::warn!(error = %e, "failed to upsert guild roster"); + 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"); + } } } } @@ -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, +} + +/// 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, + id: i64, + seq: i64, + more: bool, + members: Vec, +) -> Option> { + 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 { + names + .iter() + .map(|n| serde_json::json!({"name": n})) + .collect() + } + + fn names(vs: &[Value]) -> Vec { + 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); + } +}