feat(sidecar)!: guild rosters on the guild board, and a way to migrate the store #31

Merged
whitlocktech merged 2 commits from feat/teams-phase1-guild-roster into edge 2026-08-17 19:28:57 +00:00
4 changed files with 569 additions and 8 deletions

View File

@@ -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:

View File

@@ -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.
@@ -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<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);
}
}

View File

@@ -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

View File

@@ -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<Vec<Value>> {
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::<String, _>("json")).ok()?;
let members: Option<String> = r.get("members");
if let (Some(obj), Some(raw)) = (v.as_object_mut(), members) {
if let Ok(list) = serde_json::from_str::<Value>(&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<sqlx::sqlite::SqliteRow>) -> Vec<Value> {
rows.into_iter()
.filter_map(|r| serde_json::from_str(&r.get::<String, _>("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<String> {
sqlx::query("PRAGMA table_info(guilds)")
.fetch_all(&store.pool)
.await
.expect("table_info")
.into_iter()
.map(|r| r.get::<String, _>("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"
);
}
}