diff --git a/.gitea/workflows/pr-checks.yml b/.gitea/workflows/pr-checks.yml index d1d8939..1681efb 100644 --- a/.gitea/workflows/pr-checks.yml +++ b/.gitea/workflows/pr-checks.yml @@ -17,10 +17,12 @@ # so let this workflow run on one PR first. The `PR Checks / *` glob matches # without needing the dropdown. # -# Scope note: this gates PRs into `main` only. Feature work that lands on an -# integration branch first (e.g. `edge`) is still caught on the branch's PR into -# `main`. To gate that earlier hop too, add the branch to the `branches:` list -# below — nothing else needs to change. +# Scope note: `edge` is gated as well as `main`. Multi-phase work lands there +# first, so gating only the `main` hop would run these checks for the first time +# at the cutover — the one moment a red build is most expensive to discover. This +# is the same call `RunicGateway/installer` made for the same reason, and it was +# taken here after a nine-PR Android workstream landed on an ungated `edge` with +# no CI at all. Adding a branch to the `branches:` list is the whole change. # # Runner: the same self-hosted `ubuntu-latest` runner release.yml uses. Rust is # not assumed to be preinstalled, so the toolchain step bootstraps it the same @@ -31,7 +33,7 @@ name: PR Checks on: pull_request: - branches: [main] + branches: [main, edge] # A newer push to the same PR cancels the in-flight run. concurrency: diff --git a/sidecar/src/app.rs b/sidecar/src/app.rs index 05ef060..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. @@ -161,6 +166,43 @@ where } } } + // Guild roster (Protocol 4): the member list that `guild.update`'s counts cannot + // express. It writes a *different column* of the same row, so it never races + // guild.update. `guild.leave` deliberately has no arm here — the sidecar + // forwards it (persisted and broadcast below, like any event) and the board's + // 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_default(); + + if let Some(complete) = + accumulate_roster(&mut roster_parts, id, seq, more, members) + { + 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 { @@ -279,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 { @@ -327,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); + } +} diff --git a/sidecar/src/main.rs b/sidecar/src/main.rs index efb1ad3..347869f 100644 --- a/sidecar/src/main.rs +++ b/sidecar/src/main.rs @@ -46,7 +46,13 @@ use tracing_subscriber::EnvFilter; /// `vendor.listing.remove`, with the `GET /ruleset`, `/points` and `/market` reads that serve them /// from the store. Same shape as the v2 bump — the kinds are additive, the endpoints are not — and /// there is deliberately no feature-negotiation array: v3 implies all three kinds. -pub const PROTOCOL_VERSION: u32 = 3; +/// +/// v4 (Protocol 4): adds `guild.roster` and `guild.leave`, giving the guild board a real member list +/// instead of the member *count* that was all v2 could express. Additive in the same way again — the +/// kinds are new, `GET /guilds` grows a `roster` key, and nothing existing changed shape. This is the +/// first bump that also needed a **store migration** (`guilds.members`), because it is the first to +/// add a column to a table that already exists rather than a whole new table; see `store::migrate`. +pub const PROTOCOL_VERSION: u32 = 4; // Not `#[tokio::main]`: on Windows the SCM dispatcher takes over this thread and starts the runtime // itself, on its own thread, once the service actually begins. The runtime is built by whichever diff --git a/sidecar/src/store.rs b/sidecar/src/store.rs index 0079e87..cbb2c8c 100644 --- a/sidecar/src/store.rs +++ b/sidecar/src/store.rs @@ -45,6 +45,7 @@ impl Store { .await?; sqlx::query(SCHEMA).execute(&pool).await?; + migrate(&pool).await?; info!(%path, "store ready"); Ok(Self { pool }) } @@ -236,12 +237,61 @@ impl Store { Ok(()) } + /// Upserts one guild's member roster (Protocol 4), keyed by guild id, touching **only** the + /// `members` column. + /// + /// Deliberately not a write to `json`. That column holds the verbatim `guild.update` line, and a + /// roster arriving as its own event must not clobber the snapshot — name, abbreviation, leader, + /// online count — that `guild.update` owns. Splitting the two writers across two columns of one + /// row is what lets both be plain upserts: neither needs to read the other's value first, so + /// there is no read-modify-write and no ordering requirement between the two kinds. + /// + /// The `INSERT` half is not redundant: a roster can arrive before the first `guild.update` for a + /// guild, and the row it creates then carries `'{}'` until that update fills it in. + pub async fn upsert_guild_roster( + &self, + id: i64, + members_json: &str, + t: i64, + ) -> anyhow::Result<()> { + sqlx::query( + "INSERT INTO guilds (id, name, json, updated_t, members) VALUES (?, NULL, '{}', ?, ?) + ON CONFLICT(id) DO UPDATE SET members = excluded.members, updated_t = excluded.updated_t", + ) + .bind(id) + .bind(t) + .bind(members_json) + .execute(&self.pool) + .await?; + Ok(()) + } + /// The full guild board: every guild's latest snapshot, ordered by name. + /// + /// The roster is stored in its own column (see [`Self::upsert_guild_roster`]) and folded into + /// the projected object as `roster` here, at read time. A guild that has had a `guild.update` + /// but no `guild.roster` yet simply has no `roster` key, which is the honest representation of + /// "not known" and distinct from a guild whose roster is genuinely empty. pub async fn guilds_all(&self) -> anyhow::Result> { - let rows = sqlx::query("SELECT json FROM guilds ORDER BY name, id") + let rows = sqlx::query("SELECT json, members FROM guilds ORDER BY name, id") .fetch_all(&self.pool) .await?; - Ok(parse_json_column(rows)) + + Ok(rows + .into_iter() + .filter_map(|r| { + let mut v: Value = serde_json::from_str(&r.get::("json")).ok()?; + let members: Option = r.get("members"); + + if let (Some(obj), Some(raw)) = (v.as_object_mut(), members) { + if let Ok(list) = serde_json::from_str::(&raw) { + obj.insert("roster".into(), list); + } + } + + Some(v) + }) + .collect()) } // ---- governor board (Protocol 2.0) ---- @@ -513,6 +563,75 @@ impl Store { } } +/// The schema version this build expects. Bump it, and add the matching arm to [`migrate`], for +/// every change that `SCHEMA` alone cannot make to a database that already exists. +const SCHEMA_VERSION: i64 = 1; + +/// Brings an existing database forward to [`SCHEMA_VERSION`]. +/// +/// `SCHEMA` is `CREATE TABLE IF NOT EXISTS` only, which is enough to *add a table* but cannot add a +/// column to a table that is already there. Every schema change up to and including Protocol 3.0 +/// happened to add whole tables, so this never mattered and `ALTER TABLE` appears nowhere in this +/// repo's history. `guilds.members` (Protocol 4) is the first column added to an existing table, so +/// the mechanism has to exist now. +/// +/// The version counter is SQLite's own `PRAGMA user_version`: an integer in the database header, so +/// it needs no table of its own and cannot be separated from the file it describes. Each step runs +/// in a transaction **together with** the bump that records it, so a step either lands completely or +/// not at all, and an interrupted run resumes at the right place rather than re-applying half of one. +/// +/// A failure here propagates and aborts startup, deliberately. A half-migrated store answers the +/// website with confusing partial data, which is worse than being plainly absent — and the shard +/// dials *out* to the sidecar, so a sidecar that refuses to start never stalls the game. +async fn migrate(pool: &SqlitePool) -> anyhow::Result<()> { + let mut version: i64 = sqlx::query_scalar("PRAGMA user_version") + .fetch_one(pool) + .await?; + + // A database written by a *newer* sidecar than this binary. This is not an error: every step + // here is additive, so a newer schema has only columns and tables an older reader ignores, and + // refusing to start would turn "roll the binary back" — a recovery path — into a dead end. + if version > SCHEMA_VERSION { + tracing::warn!( + found = version, + expected = SCHEMA_VERSION, + "store was written by a newer sidecar; continuing, as migrations are additive" + ); + return Ok(()); + } + + while version < SCHEMA_VERSION { + let next = version + 1; + let mut tx = pool.begin().await?; + + match next { + // Protocol 4: the guild board carries a member roster. Its own column rather than a + // field folded into `json`, because `json` holds the verbatim `guild.update` line and + // the two writers must not overwrite each other — see `upsert_guild_roster`. + 1 => { + sqlx::query("ALTER TABLE guilds ADD COLUMN members TEXT") + .execute(&mut *tx) + .await?; + } + // Unreachable while SCHEMA_VERSION and this match are edited together, which is the + // point of failing loudly rather than silently leaving the counter short. + n => anyhow::bail!("no migration step defined for schema version {n}"), + } + + // `PRAGMA` takes no bind parameters, so this is formatted — safe because `next` is an i64 + // this loop produced, never anything from outside the process. + sqlx::query(&format!("PRAGMA user_version = {next}")) + .execute(&mut *tx) + .await?; + + tx.commit().await?; + info!(version = next, "schema migration applied"); + version = next; + } + + Ok(()) +} + fn parse_json_column(rows: Vec) -> Vec { rows.into_iter() .filter_map(|r| serde_json::from_str(&r.get::("json")).ok()) @@ -611,3 +730,195 @@ CREATE TABLE IF NOT EXISTS ruleset ( updated_t INTEGER NOT NULL ); "#; + +#[cfg(test)] +mod tests { + use super::*; + + /// A unique scratch database path. Matches `config`'s idiom — `std::env::temp_dir()` plus the + /// test name — so the cases stay independent under the parallel test runner. + fn scratch(name: &str) -> String { + let dir = std::env::temp_dir().join(format!("uo-link-store-test-{name}")); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).expect("create scratch dir"); + dir.join("uo-link.db").to_string_lossy().into_owned() + } + + /// The `guilds` table exactly as a pre-Protocol-4 sidecar left it: no `members` column, and + /// `user_version` still 0. This is the shape a real operator's database is in before an update, + /// and the only starting point where the migration does anything. + async fn legacy_db(path: &str) -> SqlitePool { + let pool = SqlitePoolOptions::new() + .max_connections(1) + .connect_with( + SqliteConnectOptions::new() + .filename(path) + .create_if_missing(true), + ) + .await + .expect("open legacy db"); + + sqlx::query( + "CREATE TABLE guilds ( + id INTEGER PRIMARY KEY, + name TEXT, + json TEXT NOT NULL, + updated_t INTEGER NOT NULL + )", + ) + .execute(&pool) + .await + .expect("create legacy guilds table"); + + pool.close().await; + pool + } + + async fn user_version(store: &Store) -> i64 { + sqlx::query_scalar("PRAGMA user_version") + .fetch_one(&store.pool) + .await + .expect("read user_version") + } + + async fn guild_columns(store: &Store) -> Vec { + sqlx::query("PRAGMA table_info(guilds)") + .fetch_all(&store.pool) + .await + .expect("table_info") + .into_iter() + .map(|r| r.get::("name")) + .collect() + } + + #[tokio::test] + async fn an_existing_pre_protocol_4_database_gains_the_members_column() { + // The case that matters: `SCHEMA`'s CREATE TABLE IF NOT EXISTS is a no-op against a table + // that is already there, so without `migrate` this database would never get the column and + // every roster write would fail against a live install. + let path = scratch("legacy-upgrade"); + legacy_db(&path).await; + + let store = Store::open(&path).await.expect("open migrates"); + + assert!( + guild_columns(&store).await.contains(&"members".to_string()), + "the migration must add guilds.members to a database that already had the table" + ); + assert_eq!(user_version(&store).await, SCHEMA_VERSION); + } + + #[tokio::test] + async fn a_fresh_database_lands_at_the_current_version() { + let path = scratch("fresh"); + let store = Store::open(&path).await.expect("open"); + + assert!(guild_columns(&store).await.contains(&"members".to_string())); + assert_eq!(user_version(&store).await, SCHEMA_VERSION); + } + + #[tokio::test] + async fn reopening_an_already_migrated_database_is_a_no_op() { + // Every sidecar restart re-runs this path, so a second run must not attempt the ALTER again + // — which would fail with "duplicate column name" and, since a migration failure aborts + // startup, would leave the sidecar unable to start at all after its first upgrade. + let path = scratch("idempotent"); + legacy_db(&path).await; + + Store::open(&path).await.expect("first open"); + let store = Store::open(&path).await.expect("second open must succeed"); + + assert_eq!(user_version(&store).await, SCHEMA_VERSION); + } + + #[tokio::test] + async fn a_roster_does_not_clobber_the_guild_update_snapshot() { + // The invariant the two-column split exists to give. `json` holds the verbatim guild.update + // line; if a roster write touched it, name/abbr/online would vanish from the board. + let path = scratch("no-clobber"); + let store = Store::open(&path).await.expect("open"); + + store + .upsert_guild( + 7, + Some("The Cartographers"), + r#"{"kind":"guild.update","id":7,"name":"The Cartographers","abbr":"MAP","members":2,"online":1}"#, + 100, + ) + .await + .expect("upsert guild"); + + store + .upsert_guild_roster( + 7, + r#"[{"serial":"0x1","name":"Ada"},{"serial":"0x2","name":"Bo"}]"#, + 200, + ) + .await + .expect("upsert roster"); + + let guilds = store.guilds_all().await.expect("read board"); + assert_eq!(guilds.len(), 1); + let g = &guilds[0]; + + assert_eq!( + g["name"], "The Cartographers", + "guild.update's name survived" + ); + assert_eq!(g["abbr"], "MAP", "guild.update's abbr survived"); + assert_eq!(g["online"], 1, "guild.update's online count survived"); + assert_eq!(g["roster"].as_array().expect("roster is an array").len(), 2); + assert_eq!(g["roster"][0]["name"], "Ada"); + } + + #[tokio::test] + async fn the_two_writers_are_order_independent() { + // A roster can arrive before the first guild.update for a guild — on a reconnect the shard + // re-emits both and nothing orders them. Neither write may depend on the other's row. + let path = scratch("either-order"); + let store = Store::open(&path).await.expect("open"); + + store + .upsert_guild_roster(9, r#"[{"serial":"0x3","name":"Cy"}]"#, 100) + .await + .expect("roster first"); + store + .upsert_guild( + 9, + Some("Late Arrivals"), + r#"{"kind":"guild.update","id":9,"name":"Late Arrivals","abbr":"LTE"}"#, + 200, + ) + .await + .expect("update second"); + + let guilds = store.guilds_all().await.expect("read board"); + assert_eq!(guilds.len(), 1, "one row, not two"); + assert_eq!(guilds[0]["name"], "Late Arrivals"); + assert_eq!(guilds[0]["roster"].as_array().expect("roster").len(), 1); + } + + #[tokio::test] + async fn a_guild_with_no_roster_yet_has_no_roster_key() { + // "Not known" and "known to be empty" are different, and the board must not conflate them: + // a website reading `roster: []` would render an empty roster as fact. + let path = scratch("absent-roster"); + let store = Store::open(&path).await.expect("open"); + + store + .upsert_guild( + 11, + Some("Unswept"), + r#"{"kind":"guild.update","id":11,"name":"Unswept"}"#, + 100, + ) + .await + .expect("upsert guild"); + + let guilds = store.guilds_all().await.expect("read board"); + assert!( + guilds[0].get("roster").is_none(), + "a guild with no roster event must not grow a roster key" + ); + } +}