From 06fa5d73301ec5c5d0805ff37cbe98500b8d4ba7 Mon Sep 17 00:00:00 2001 From: wtclaude Date: Wed, 16 Sep 2026 08:18:43 -0500 Subject: [PATCH] =?UTF-8?q?feat(sidecar):=20protocol=202=20=E2=80=94=20fil?= =?UTF-8?q?e=20by=20type,=20a=20cursor=20feed,=20and=20bounded=20history?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The sidecar now files a frame by its `type` and never by its `kind`. That is the dumb-forwarder property made structural: `event` is appended to history, `snapshot` replaces the board of its kind, `reply` is routed by `reqId`, `control` is broadcast and kept nowhere. Ten new event kinds are no change here at all, which is the whole point when the thing that grows fastest is the catalogue. A frame whose `type` this build does not know is dropped and counted, never guessed at. Defaulting an absent one to `event` would file a BOARD as history — the presence board appended a few thousand times, which nothing reports. The count is on `/health` as `untyped_frames`, because the failure it diagnoses (a plugin and a sidecar on different protocol versions, which the game link has no handshake to catch) otherwise presents as a website showing nothing while the game is plainly up. It caught exactly that within three seconds of first running, against a protocol 1 plugin still live on a retired rig. `boards` generalises protocol 1's single `server_state` row, and a database made by protocol 1 is migrated in place: the two indexed columns are added by a guarded `ALTER`, and the old board is carried across. Without that carry-over an upgraded sidecar answers `204` until the game next connects, and the website reads that as "never heard from" — losing a server it has rendered for weeks at the exact moment somebody upgraded the bridge. `GET /feed` is the ingest cursor: oldest first, strictly after an id, with `lastId` and `more`. It is a separate route rather than a flag on `/events` because one route with two orderings serves the other one to every caller that forgets the parameter — and for the ingesting caller that means advancing its cursor past rows it never read. Omitting `since` asks where the END is; `since=0` is the other question entirely, and the two must not be separated by whether somebody typed a parameter. `[store].retain_days` (default 14) prunes events hourly. Boards are never pruned: history grows and the present does not, and a pruned board is a server that has never connected. The repository also had no CI. `pr-checks.yml` runs the fmt, clippy and test gates phases 1 and 3 have both been running by hand — a guard nothing invokes is a guard whose state nobody knows. 44 tests pass, clippy clean at `-D warnings`. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_016wDDVXWMDz82WqE1i969r4 --- .gitea/workflows/pr-checks.yml | 99 ++++++ sidecar/README.md | 32 +- sidecar/src/app.rs | 186 +++++++++-- sidecar/src/config.rs | 29 ++ sidecar/src/game.rs | 7 +- sidecar/src/main.rs | 18 +- sidecar/src/store.rs | 570 ++++++++++++++++++++++++++++----- sidecar/src/web.rs | 85 ++++- 8 files changed, 910 insertions(+), 116 deletions(-) create mode 100644 .gitea/workflows/pr-checks.yml diff --git a/.gitea/workflows/pr-checks.yml b/.gitea/workflows/pr-checks.yml new file mode 100644 index 0000000..ce89ea4 --- /dev/null +++ b/.gitea/workflows/pr-checks.yml @@ -0,0 +1,99 @@ +# Gate every pull request into `main` on the checks this repository already had +# and nobody ran automatically. +# +# Phases 1 and 3 both wrote `cargo fmt`, `cargo clippy -D warnings` and a test +# suite, and both ran them BY HAND from a workstation. That is the whole gap +# this file closes: a guard nothing invokes is a guard whose state nobody knows, +# and the repository that gets released had nothing gating it at all. +# +# Adapted from RunicGateway/installer's pr-checks.yml, which is the other Rust +# crate in this project and already solved the toolchain-on-a-shared-runner +# problem. Two differences, both because of where the crate sits: +# +# • The crate is in `sidecar/`, not at the repo root, so every cargo step runs +# with that working directory and the cache key reads that lockfile. +# • There is no "does a crate exist yet" detection. The installer needed it +# because its CI landed before its code; here the code came first. +# +# Enforcement (one-time, in the Gitea UI): +# Repository Settings → Branches → Branch Protection (rule for `main`) +# • Enable Status Check +# • Status check patterns: PR Checks / * +# Gitea only lists a context in its dropdown after it has reported once, so let +# this run on one PR first; the glob matches without the dropdown and keeps +# matching as jobs are added. +# +# Scope note: `edge` is gated as well as `main` though this repo has no `edge` +# branch. Multi-phase work lands there first everywhere else in this project, and +# 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 find. + +name: PR Checks + +on: + pull_request: + branches: [main, edge] + +concurrency: + group: pr-checks-${{ github.ref }} + cancel-in-progress: true + +jobs: + rust-gates: + runs-on: ubuntu-latest + timeout-minutes: 30 + defaults: + run: + working-directory: sidecar + steps: + - uses: actions/checkout@v4 + + # One job runs all three gates on purpose: installing the toolchain costs + # far more than the checks do, so splitting fmt/clippy/test into parallel + # jobs would pay that cost three times for no wall-clock win. + - name: Install Rust toolchain (rustfmt + clippy) + run: | + set -euo pipefail + SUDO=""; [ "$(id -u)" -ne 0 ] && SUDO="sudo" + $SUDO apt-get update + $SUDO apt-get install -y --no-install-recommends \ + build-essential curl ca-certificates git + + if ! command -v cargo >/dev/null 2>&1; then + curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs \ + | sh -s -- -y --profile minimal --default-toolchain stable + fi + echo "${HOME}/.cargo/bin" >> "$GITHUB_PATH" + export PATH="${HOME}/.cargo/bin:${PATH}" + rustup component add rustfmt clippy + cargo --version && cargo fmt --version && cargo clippy --version + + # Keyed on Cargo.lock: dependency builds are reused until a dep actually + # changes. A cache miss only makes the run slower, never wrong. + - name: Cache cargo registry and build dir + uses: actions/cache@v4 + with: + path: | + ~/.cargo/registry + ~/.cargo/git + sidecar/target + key: ${{ runner.os }}-cargo-${{ hashFiles('sidecar/Cargo.lock') }} + restore-keys: | + ${{ runner.os }}-cargo- + + # Cheapest gate first — parses only, no compile, so a formatting slip fails + # in seconds instead of after a full build. + - name: cargo fmt --check + run: cargo fmt --check + + # --all-targets covers the tests too, which is where most of this crate's + # interesting code is. -D warnings makes a lint a failure, so the crate + # starts clean at this bar and anything new is a regression from the PR. + - name: cargo clippy + run: cargo clippy --locked --all-targets -- -D warnings + + # --locked also proves Cargo.lock is in sync with Cargo.toml rather than + # letting the build silently update it. + - name: cargo test + run: cargo test --locked + diff --git a/sidecar/README.md b/sidecar/README.md index ce85119..dac655e 100644 --- a/sidecar/README.md +++ b/sidecar/README.md @@ -15,6 +15,7 @@ Configuration reference and endpoint list. For what this component *is*, see the | `[web].bind` | `RUSTLINK_WEB_BIND` | `127.0.0.1:8090` | Where the website reaches this sidecar | | `[web].auth_token` | `RUSTLINK_WEB_TOKEN` | *(generated)* | The shared secret the website presents | | `[store].path` | `RUSTLINK_DB_PATH` | `rust-link.db` | SQLite file | +| `[store].retain_days` | `RUSTLINK_RETAIN_DAYS` | `14` | Days of event history to keep. `0` keeps everything | Two things about those defaults are load-bearing: @@ -51,7 +52,9 @@ still authenticate). Every response carries `X-RustLink-Version`. |---|---|---| | `GET /health` | — | Unauthenticated, so monitoring can reach it | | `GET /server` | store | The last `server.hello`. **`204` when the game has never connected** | -| `GET /events?kind=&limit=` | store | Newest first; `limit` clamped to 1–1000 | +| `GET /boards` | store | Every board, keyed by kind. `200` with an empty object when there are none | +| `GET /events?kind=&wipe=&limit=` | store | Newest first; `limit` clamped to 1–1000. For a human | +| `GET /feed?since=&limit=` | store | **Oldest first, from a cursor.** For a consumer that must not miss a row. Omitting `since` asks where the end is | | `GET /status` | plugin (RPC) | A live round trip. `503` with no plugin, `504` on no reply | | `GET /ws` | broadcast | The live feed. Sends `ws.hello` on connect | @@ -63,20 +66,41 @@ the game is down, because "what is it doing right now" has no stale answer worth heard from this server" and "this server reports nothing" are different answers, and a client that cannot tell them apart renders a server that does not exist. -## Protocol 1 +## The protocol Newline-delimited JSON over TCP, both directions. Outbound frames (plugin → sidecar) carry `kind`; inbound frames (sidecar → plugin) carry `cmd`. Lines are capped at 1 MiB; an over-long line is discarded and the connection stays up. +**This process files a frame by its `type`, and never by its `kind`** — which is what keeps it a +dumb forwarder while the catalogue grows. Ten new event kinds are no change here at all. + +| `type` | Kept | Broadcast | Example | +|---|---|---|---| +| `event` | appended to the history | yes | `player.death` | +| `snapshot` | **replaces** the board of that kind | yes | `players.online` | +| `reply` | no | no | `server.status`, routed by `reqId` | +| `control` | no | yes | `pong`, `link.down` | + +A frame with no `type` this build knows is **dropped and counted**, never guessed at, and the count +is on `/health` as `untyped_frames`. The game link has no version handshake, so a plugin and a +sidecar on different protocol versions show up there and nowhere else. + | Frame | Direction | Purpose | |---|---|---| -| `server.hello` | plugin → sidecar | Sent on **every connect**, not once at server start — this process restarts independently of the game. Carries `serverId` and `bootId` | +| `server.hello` | plugin → sidecar | A board. Sent on **every connect**, not once at server start — this process restarts independently of the game. Carries `serverId`, `bootId` and `wipeId` | +| `players.online` | plugin → sidecar | The other board: who is connected, re-sent on connect and every 60s | | `ping` / `pong` | sidecar → plugin → sidecar | The heartbeat, every 30s. A `pong` is never persisted; it only moves `last_event` | | `server.status` | sidecar → plugin → sidecar | The one request/reply verb, correlated by `reqId` | +| the read path | plugin → sidecar | Presence, deaths, chat, tallies, moderation, the wipe — the catalogue is `PROTOCOL.md` §8.4 | `bootId` is how a game restart is told apart from a sidecar reconnect — the distinction the event -system's `reconcile` hangs off later. +system's `reconcile` hangs off later. `wipeId` is how a wipe splits the history instead of ending +it; the plugin derives it, and every frame carries it. + +**History is bounded, boards are not.** `[store].retain_days` (default 14) prunes events hourly; a +board is one row per kind holding what is true now, and pruning it would make a server the site has +rendered for weeks look like one that has never connected. The permanent record is the website's. **The RPC reply timeout (`rpc::REPLY_TIMEOUT`, 10s) is a ceiling every later command budget sits under.** Core classifies a budget overrun as retryable unconditionally, because it cannot ask the diff --git a/sidecar/src/app.rs b/sidecar/src/app.rs index a834788..57763f1 100644 --- a/sidecar/src/app.rs +++ b/sidecar/src/app.rs @@ -12,7 +12,7 @@ //! supervision later is a new module rather than a restructuring of this one. use std::future::Future; -use std::sync::atomic::AtomicI64; +use std::sync::atomic::{AtomicI64, AtomicU64}; use std::sync::Arc; use std::time::{Duration, Instant}; @@ -28,6 +28,41 @@ use crate::{config, game, rpc, store, web}; /// indistinguishable from "the link died six hours ago". const HEARTBEAT: Duration = Duration::from_secs(30); +/// What this process does with one frame, decided by the frame's `type` and by nothing else. +/// +/// Lifting it out of the event loop is not tidiness. This is the whole of protocol 2's filing +/// rule (PROTOCOL.md §8.1) and the reason a new event kind costs this repository nothing — so it +/// is worth being a thing that can be asserted about, rather than five arms of a `match` inside a +/// spawned task that no test can reach. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum Filing { + /// History. Stored, then broadcast. + Persist, + /// Current state. Replaces the board of its kind, then broadcast. + Board, + /// Neither history nor state: broadcast only. The heartbeat, and our own `link.down`. + Announce, + /// A reply whose caller has already timed out. Storing it would put a point-in-time answer + /// into the history as though the game had volunteered it; broadcasting it would show it to a + /// website as a live event. Neither is true, so it goes nowhere. + Ignore, + /// Not something this build knows how to file. Almost always a plugin on a different protocol + /// version — the game link has no handshake to catch that earlier. + Unfilable, +} + +impl Filing { + fn of(frame_type: &str) -> Self { + match frame_type { + "event" => Filing::Persist, + "snapshot" => Filing::Board, + "control" => Filing::Announce, + "reply" => Filing::Ignore, + _ => Filing::Unfilable, + } + } +} + /// Runs the sidecar until `shutdown` resolves. /// /// `config_path` is the `--config` argument, or `None` to resolve `$RUSTLINK_CONFIG` and the @@ -67,6 +102,11 @@ where let started = Instant::now(); let last_event = Arc::new(AtomicI64::new(0)); + // Frames this build could not file. Surfaced on /health rather than only in the log, because + // the shape it diagnoses — a plugin and a sidecar on different protocol versions — presents to + // an operator as "the website shows nothing" and nothing else. + let untyped = Arc::new(AtomicU64::new(0)); + // Website-facing HTTP server. let web_state = web::AppState { events: bcast_tx.clone(), @@ -76,6 +116,7 @@ where token: Arc::new(cfg.web.auth_token.clone()), started, last_event: last_event.clone(), + untyped: untyped.clone(), }; let web_bind = cfg.web.bind.clone(); tokio::spawn(async move { @@ -101,14 +142,18 @@ where }); // The event loop. A line that correlates to a pending REST call is a reply — route it to the - // waiting caller and stop. Everything else is a live event: persist it, then broadcast it. + // waiting caller and stop. Everything else is filed by its `type`, and by its `type` alone: + // that is what keeps this process a dumb forwarder while the catalogue grows (PROTOCOL.md + // §8.1). Ten new event kinds are no change here. let configured_server_id = cfg.game.server_id.clone(); let feed_tx = bcast_tx.clone(); let route_rpc = rpc.clone(); let event_store = store.clone(); let last_event_ts = last_event.clone(); + let untyped_count = untyped.clone(); tokio::spawn(async move { let mut total: u64 = 0; + let mut warned_untyped = false; while let Some(ev) = event_rx.recv().await { // Any line from the plugin — a pong included — is a sign of life. @@ -118,32 +163,72 @@ where continue; // consumed as a reply } - total += 1; let line = ev.value.to_string(); + let frame_type = ev + .value + .get("type") + .and_then(|v| v.as_str()) + .unwrap_or("") + .to_string(); + + match Filing::of(&frame_type) { + Filing::Persist => { + total += 1; + tracing::debug!(kind = %ev.kind, n = total, "{}", line); - match ev.kind.as_str() { - "server.hello" => { - check_server_id(&configured_server_id, &ev.value); - info!(kind = %ev.kind, n = total, "{}", line); if let Err(e) = event_store - .put_server_state(frame_t(&ev.value), &line) + .insert_event( + frame_t(&ev.value), + &ev.kind, + field(&ev.value, "serverId"), + field(&ev.value, "wipeId"), + &line, + ) .await { - warn!(error = %e, "server board write failed"); + store::warn_write("event persist failed", &e); } } - "link.down" => info!("game link down"), - _ => tracing::debug!(kind = %ev.kind, n = total, "{}", line), - } + Filing::Board => { + // A board REPLACES rather than appends. Filing one as history is the mistake + // `type` exists to prevent, and it is invisible until somebody wonders why the + // presence board has four thousand rows. + if ev.kind == store::SERVER_BOARD { + check_server_id(&configured_server_id, &ev.value); + info!(kind = %ev.kind, "{}", line); + } else { + tracing::debug!(kind = %ev.kind, "board {}", line); + } - // `pong` and `link.down` are ephemeral: one is heartbeat chatter and the other is this - // process's own observation, not something the game said. Neither is history. - if ev.kind != "pong" && ev.kind != "link.down" { - if let Err(e) = event_store - .insert_event(frame_t(&ev.value), &ev.kind, &line) - .await - { - warn!(error = %e, "event persist failed"); + if let Err(e) = event_store + .put_board(&ev.kind, frame_t(&ev.value), &line) + .await + { + store::warn_write("board write failed", &e); + } + } + Filing::Announce => { + if ev.kind == "link.down" { + info!("game link down"); + } + } + Filing::Ignore => { + tracing::debug!(kind = %ev.kind, "unrouted reply discarded"); + continue; + } + Filing::Unfilable => { + untyped_count.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + + if !warned_untyped { + warned_untyped = true; + warn!( + kind = %ev.kind, + frame_type = %frame_type, + "frame with no usable `type`; dropping. The plugin and this sidecar are almost certainly on different protocol versions — the game link has no version handshake, so this is where that shows up. Counted on /health as untyped_frames." + ); + } + + continue; // not stored, and not broadcast: nobody downstream can file it either } } @@ -153,6 +238,20 @@ where } }); + // Retention. Hourly, and once at startup so a store that grew while an operator had this + // turned off does not wait an hour to shrink. + let prune_store = store.clone(); + let retain_days = cfg.store.retain_days; + tokio::spawn(async move { + let mut tick = tokio::time::interval(Duration::from_secs(3600)); + loop { + tick.tick().await; + if let Err(e) = prune_store.prune(retain_days).await { + store::warn_write("prune failed", &e); + } + } + }); + ready(); info!("rust-link sidecar ready"); @@ -188,6 +287,18 @@ fn frame_t(value: &serde_json::Value) -> i64 { .unwrap_or_else(now_ms) } +/// Lifts an optional string field out of a frame, for the columns the store indexes on. +/// +/// Absent and empty are the same answer here — `None` — because the plugin omits a field it cannot +/// answer (a server that has never saved has no wipe) and an empty string in an indexed column +/// would group every such row together as though they shared something. +fn field<'a>(value: &'a serde_json::Value, key: &str) -> Option<&'a str> { + value + .get(key) + .and_then(|v| v.as_str()) + .filter(|s| !s.is_empty()) +} + fn now_ms() -> i64 { std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) @@ -208,6 +319,41 @@ mod tests { ); } + /// The filing rule, which is the whole of protocol 2 in this file. + #[test] + fn a_frame_is_filed_by_its_type_and_by_nothing_else() { + assert_eq!(Filing::of("event"), Filing::Persist); + assert_eq!(Filing::of("snapshot"), Filing::Board); + assert_eq!(Filing::of("control"), Filing::Announce); + assert_eq!(Filing::of("reply"), Filing::Ignore); + } + + /// The two ways a frame arrives unfilable, and the reason neither may be guessed at. + /// + /// Defaulting an absent `type` to `event` would file a BOARD as history — the presence board + /// appended a few thousand times, which nothing reports and nobody notices until they wonder + /// why the database is large. A frame nobody can file is dropped loudly instead, and counted + /// where an operator can see it. + #[test] + fn an_absent_or_unknown_type_is_never_guessed_at() { + assert_eq!(Filing::of(""), Filing::Unfilable); + assert_eq!(Filing::of("events"), Filing::Unfilable); + assert_eq!(Filing::of("Event"), Filing::Unfilable); + assert_eq!(Filing::of("boards"), Filing::Unfilable); + } + + /// The columns the store indexes on come out of the frame here, and "absent" and "empty" have + /// to be the same answer: a server that has never saved sends no `wipeId`, and an empty string + /// in an indexed column would group every such row together as though they shared a wipe. + #[test] + fn an_empty_string_is_not_a_value() { + let frame = json!({"serverId": "main", "wipeId": "", "other": 3}); + assert_eq!(field(&frame, "serverId"), Some("main")); + assert_eq!(field(&frame, "wipeId"), None); + assert_eq!(field(&frame, "missing"), None); + assert_eq!(field(&frame, "other"), None); + } + /// A frame with no `t` must still land with a usable timestamp rather than at the epoch, or /// every un-stamped event sorts to the beginning of history forever. #[test] diff --git a/sidecar/src/config.rs b/sidecar/src/config.rs index 9d91374..d5e66f5 100644 --- a/sidecar/src/config.rs +++ b/sidecar/src/config.rs @@ -64,6 +64,15 @@ pub struct WebCfg { pub struct StoreCfg { #[serde(default = "default_db_path")] pub path: String, + /// How many days of event history to keep. `0` keeps everything. + /// + /// The store sits on a game host, and protocol 2 gave it a catalogue that produces real volume + /// — every death, every chat line, every connect. The *permanent* record is the website's: + /// per-wipe rollups in the module's own tables (R12). So this bounds the sidecar's copy, and + /// the default is generous enough that nobody needs to think about it and small enough that a + /// busy month is not a wipe-day outage. + #[serde(default = "default_retain_days")] + pub retain_days: i64, } /// A loaded configuration plus what loading it *did* — an installer re-running the binary needs to @@ -89,6 +98,9 @@ fn default_web_bind() -> String { fn default_db_path() -> String { "rust-link.db".into() } +fn default_retain_days() -> i64 { + 14 +} impl Default for GameCfg { fn default() -> Self { @@ -110,6 +122,7 @@ impl Default for StoreCfg { fn default() -> Self { Self { path: default_db_path(), + retain_days: default_retain_days(), } } } @@ -189,6 +202,15 @@ impl Config { if let Ok(v) = env::var("RUSTLINK_DB_PATH") { self.store.path = v; } + if let Ok(v) = env::var("RUSTLINK_RETAIN_DAYS") { + // A malformed value is ignored rather than fatal: this reaches the process as a panel + // variable somebody typed (R22), and refusing to start over a stray character would + // take the bridge down for a setting that has a perfectly good default. + match v.trim().parse::() { + Ok(days) if days >= 0 => self.store.retain_days = days, + _ => tracing::warn!(value = %v, "ignoring an unreadable RUSTLINK_RETAIN_DAYS"), + } + } } /// Resolves `[store].path` against the config file's directory (see the module docs). Absolute @@ -373,6 +395,13 @@ auth_token = "{token}" # Relative paths resolve against the directory holding THIS FILE, not the # working directory of whatever started the process. path = "rust-link.db" + +# How many days of event history to keep. 0 keeps everything. +# +# The permanent record is the website's — it holds per-wipe rollups that survive +# a wipe. This database is the recent copy the site reads to catch up, and it +# lives on the game host, so it is bounded. +retain_days = 14 "# ) } diff --git a/sidecar/src/game.rs b/sidecar/src/game.rs index 3553aca..7d3faa3 100644 --- a/sidecar/src/game.rs +++ b/sidecar/src/game.rs @@ -204,10 +204,13 @@ pub async fn serve( } accept_handle.set(None).await; // A disconnect is a fact the website should see without polling, so it rides - // the same channel every other fact does. Nothing persists it. + // the same channel every other fact does. Nothing persists it: it is this + // process's own observation, not something the game said — which is exactly + // what `type: "control"` means (PROTOCOL.md §8.1). It is synthesised here + // rather than anywhere else because this is the only place that knows. let _ = event_tx.send(GameEvent { kind: "link.down".to_string(), - value: serde_json::json!({ "kind": "link.down" }), + value: serde_json::json!({ "kind": "link.down", "type": "control" }), }); } Err(e) => { diff --git a/sidecar/src/main.rs b/sidecar/src/main.rs index 0e411bb..acd9163 100644 --- a/sidecar/src/main.rs +++ b/sidecar/src/main.rs @@ -57,7 +57,23 @@ use tracing_subscriber::EnvFilter; /// /// Both directions are newline-delimited JSON over TCP. Outbound frames (plugin -> sidecar) carry /// `kind`; inbound frames (sidecar -> plugin) carry `cmd`. -pub const PROTOCOL_VERSION: u32 = 1; +/// +/// # Protocol 2 — the read path +/// +/// The catalogue: presence, deaths, chat, gathering, moderation and the wipe. Three things about +/// it reach this file rather than only the plugin's: +/// +/// * **Every frame carries `type`** — `event`, `snapshot`, `reply` or `control` — and this +/// process files on THAT, never on `kind`. It is what keeps the sidecar a dumb forwarder while +/// the catalogue grows: ten new event kinds are no change here at all. +/// * **Every frame carries `serverId` and `wipeId`**, and both are lifted into indexed columns +/// (the one migration shape the store's header predicted). +/// * **`GET /feed`** is the ingest cursor, oldest-first, separate from `/events` so that no +/// caller can get the other ordering by forgetting a parameter. +/// +/// `docs/rust-link/PROTOCOL.md` §8 is the specification; this constant is one of its four +/// declaration sites. +pub const PROTOCOL_VERSION: u32 = 2; fn main() -> anyhow::Result<()> { let args = match cli::parse(std::env::args().skip(1)) { diff --git a/sidecar/src/store.rs b/sidecar/src/store.rs index f863776..2d6145a 100644 --- a/sidecar/src/store.rs +++ b/sidecar/src/store.rs @@ -1,33 +1,51 @@ -//! SQLite persistence: the event history, and the last thing the game said about itself. +//! SQLite persistence: the event history, and the boards holding what is true right now. //! //! This is what lets the website read the past without asking the game, and what survives a sidecar //! restart. The event loop writes every live event here as it broadcasts it; REST reads query here //! instead of round-tripping the plugin. //! //! **The sidecar defines no schema for a frame's contents.** Events are persisted whole, as the -//! JSON text that arrived, with only `t` and `kind` lifted out for indexing. That is the +//! JSON text that arrived, with only the columns it must *index* lifted out. That is the //! dumb-forwarder property doing real work: a protocol version that adds fields to an event needs -//! no change here, and only a version that adds a *new indexed column* ever needs a migration. +//! no change here, and only a version that adds a new indexed column ever needs a migration. +//! +//! Protocol 2 is the first version that needed one — `server_id` and `wipe_id` (PROTOCOL.md §8.9) +//! — and it is applied the way this project applies every schema change: as an `ALTER` guarded by a +//! column check, never as an edit to the `CREATE`, because `CREATE TABLE IF NOT EXISTS` does +//! nothing at all against a database that already has the table and an edited column would reach +//! fresh installs only. use std::path::Path; -use serde_json::Value; +use serde_json::{json, Value}; use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions}; use sqlx::{Row, SqlitePool}; -use tracing::info; +use tracing::{info, warn}; const SCHEMA: &str = " CREATE TABLE IF NOT EXISTS events ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - t INTEGER NOT NULL, - kind TEXT NOT NULL, - json TEXT NOT NULL + id INTEGER PRIMARY KEY AUTOINCREMENT, + t INTEGER NOT NULL, + kind TEXT NOT NULL, + server_id TEXT, + wipe_id TEXT, + json TEXT NOT NULL ); CREATE INDEX IF NOT EXISTS idx_events_kind_id ON events (kind, id DESC); CREATE INDEX IF NOT EXISTS idx_events_t ON events (t); --- Exactly one row, id 1: the most recent server.hello. A board, in the sense chapter 4 uses the --- word — current state with one producer, re-sent on every connect — rather than a history. +-- Boards: current state, one row per kind, replaced whole. A board in chapter 4's sense — state +-- with exactly one producer, re-sent on every connect — rather than a history. Protocol 1 had one +-- of these hard-coded as `server_state`; protocol 2 has two and will have more, so the kind is a +-- key rather than a table name. +CREATE TABLE IF NOT EXISTS boards ( + kind TEXT PRIMARY KEY, + t INTEGER NOT NULL, + json TEXT NOT NULL +); + +-- Protocol 1's single board. Kept so that a sidecar upgraded in place can carry its contents over +-- (see `migrate`); nothing writes to it any more. CREATE TABLE IF NOT EXISTS server_state ( id INTEGER PRIMARY KEY CHECK (id = 1), t INTEGER NOT NULL, @@ -35,6 +53,25 @@ CREATE TABLE IF NOT EXISTS server_state ( ); "; +/// The board holding the last `server.hello`. Named once here rather than spelled at four call +/// sites, because it is the one board with a REST route of its own. +pub const SERVER_BOARD: &str = "server.hello"; + +/// One row of the ingest feed: a stored event, with the identity a cursor needs. +#[derive(Debug, Clone)] +pub struct FeedItem { + pub id: i64, + pub t: i64, + pub kind: String, + pub frame: Value, +} + +impl FeedItem { + pub fn to_json(&self) -> Value { + json!({ "id": self.id, "t": self.t, "kind": self.kind, "frame": self.frame }) + } +} + #[derive(Clone)] pub struct Store { pool: SqlitePool, @@ -74,8 +111,65 @@ impl Store { .await?; sqlx::query(SCHEMA).execute(&pool).await?; + + let store = Self { pool }; + store.migrate().await?; + info!(%path, "store ready"); - Ok(Self { pool }) + Ok(store) + } + + /// Brings a database created by an older protocol up to this one. + /// + /// Two steps, both idempotent, both safe to run on a fresh database where they do nothing: + /// add the columns protocol 2 indexes on, and carry protocol 1's single board into `boards`. + /// + /// The board carry-over matters more than it looks: without it, an upgraded sidecar answers + /// `204` for `/server` until the game next connects, and the website reads that as *this + /// server has never been heard from* — the site loses a server it has been rendering for + /// weeks, at the exact moment somebody upgraded the bridge. + async fn migrate(&self) -> anyhow::Result<()> { + for (column, ddl) in [ + ("server_id", "ALTER TABLE events ADD COLUMN server_id TEXT"), + ("wipe_id", "ALTER TABLE events ADD COLUMN wipe_id TEXT"), + ] { + if !self.has_column("events", column).await? { + sqlx::query(ddl).execute(&self.pool).await?; + info!(column, "events: column added"); + } + } + + // Indexed after the columns exist, and in the same idempotent spirit. + sqlx::query("CREATE INDEX IF NOT EXISTS idx_events_wipe_id ON events (wipe_id, id DESC)") + .execute(&self.pool) + .await?; + + let carried: Option<(i64, String)> = sqlx::query_as( + "SELECT t, json FROM server_state WHERE id = 1 + AND NOT EXISTS (SELECT 1 FROM boards WHERE kind = ?)", + ) + .bind(SERVER_BOARD) + .fetch_optional(&self.pool) + .await?; + + if let Some((t, json)) = carried { + self.put_board(SERVER_BOARD, t, &json).await?; + info!("carried the protocol 1 server board into boards"); + } + + Ok(()) + } + + async fn has_column(&self, table: &str, column: &str) -> anyhow::Result { + // `PRAGMA table_info` does not take a bind parameter for the table name, which is why this + // is formatted. Both call sites pass a literal; nothing here is reachable from a request. + let rows = sqlx::query(&format!("PRAGMA table_info({table})")) + .fetch_all(&self.pool) + .await?; + + Ok(rows + .iter() + .any(|r| r.get::("name").eq_ignore_ascii_case(column))) } /// Cheap liveness check for the health endpoint. @@ -86,44 +180,128 @@ impl Store { /// Appends one live event. Failures are logged by the caller; persistence must never block the /// live feed. - pub async fn insert_event(&self, t: i64, kind: &str, json: &str) -> anyhow::Result<()> { - sqlx::query("INSERT INTO events (t, kind, json) VALUES (?, ?, ?)") - .bind(t) - .bind(kind) - .bind(json) - .execute(&self.pool) - .await?; + /// + /// `server_id` and `wipe_id` are lifted out of the frame by the caller and stored as columns as + /// well as remaining in the JSON. Duplicated deliberately: the column is what an index and a + /// `WHERE` can reach, and the JSON is what stays correct when the columns change. + pub async fn insert_event( + &self, + t: i64, + kind: &str, + server_id: Option<&str>, + wipe_id: Option<&str>, + json: &str, + ) -> anyhow::Result<()> { + sqlx::query( + "INSERT INTO events (t, kind, server_id, wipe_id, json) VALUES (?, ?, ?, ?, ?)", + ) + .bind(t) + .bind(kind) + .bind(server_id) + .bind(wipe_id) + .bind(json) + .execute(&self.pool) + .await?; Ok(()) } - /// Most-recent events, newest first, optionally filtered by kind. - pub async fn recent(&self, kind: Option<&str>, limit: i64) -> anyhow::Result> { + /// Most-recent events, **newest first**, optionally filtered by kind and by wipe. + /// + /// For a human, an admin screen, or a point-in-time look. A consumer that must not miss a row + /// wants [`Store::feed`] instead — see its documentation for why these are two functions and + /// not one with a flag. + pub async fn recent( + &self, + kind: Option<&str>, + wipe: Option<&str>, + limit: i64, + ) -> anyhow::Result> { let limit = limit.clamp(1, 1000); - let rows = match kind { - Some(k) => { - sqlx::query("SELECT json FROM events WHERE kind = ? ORDER BY id DESC LIMIT ?") - .bind(k) - .bind(limit) - .fetch_all(&self.pool) - .await? - } - None => { - sqlx::query("SELECT json FROM events ORDER BY id DESC LIMIT ?") - .bind(limit) - .fetch_all(&self.pool) - .await? - } - }; + + // Built rather than branched four ways: two optional filters is four combinations, and the + // fourth is always the one nobody tested. The bindings stay parameterised. + let mut sql = String::from("SELECT json FROM events WHERE 1 = 1"); + + if kind.is_some() { + sql.push_str(" AND kind = ?"); + } + if wipe.is_some() { + sql.push_str(" AND wipe_id = ?"); + } + + sql.push_str(" ORDER BY id DESC LIMIT ?"); + + let mut query = sqlx::query(&sql); + + if let Some(k) = kind { + query = query.bind(k); + } + if let Some(w) = wipe { + query = query.bind(w); + } + + let rows = query.bind(limit).fetch_all(&self.pool).await?; Ok(parse_json_column(rows)) } - /// Replaces the one `server_state` row. Called for every `server.hello`, which the plugin sends - /// on every connect — so this is an upsert by construction, not by accident. - pub async fn put_server_state(&self, t: i64, json: &str) -> anyhow::Result<()> { - sqlx::query( - "INSERT INTO server_state (id, t, json) VALUES (1, ?, ?) - ON CONFLICT(id) DO UPDATE SET t = excluded.t, json = excluded.json", + /// The ingest cursor: events **after** `since`, **oldest first**. + /// + /// This is a separate function from [`Store::recent`], and the route on top of it is a separate + /// route, for one reason: a single route whose ordering depends on a query parameter serves the + /// other ordering to every caller that forgets it, and for the ingesting caller that means + /// advancing its cursor past rows it never read. Silently, and once per deployment mistake. + /// + /// Returns the page and whether it filled — a consumer an hour behind drains at its own pace + /// rather than guessing from a count. + pub async fn feed(&self, since: i64, limit: i64) -> anyhow::Result<(Vec, bool)> { + let limit = limit.clamp(1, 1000); + + let rows = sqlx::query( + "SELECT id, t, kind, json FROM events WHERE id > ? ORDER BY id ASC LIMIT ?", ) + .bind(since) + .bind(limit) + .fetch_all(&self.pool) + .await?; + + let more = rows.len() as i64 == limit; + + let items = rows + .into_iter() + .filter_map(|r| { + let json: String = r.get("json"); + serde_json::from_str(&json).ok().map(|frame| FeedItem { + id: r.get("id"), + t: r.get("t"), + kind: r.get("kind"), + frame, + }) + }) + .collect(); + + Ok((items, more)) + } + + /// The highest event id in the store, or 0 when it is empty. + /// + /// A consumer starting from nothing uses this to begin at the *end* rather than replaying the + /// whole history it has no use for — a fresh module against a sidecar that has been running for + /// a month wants what happens next, not a fortnight of deaths. + pub async fn last_event_id(&self) -> anyhow::Result { + let row = sqlx::query("SELECT COALESCE(MAX(id), 0) AS id FROM events") + .fetch_one(&self.pool) + .await?; + Ok(row.get("id")) + } + + /// Replaces one board. Called for every snapshot frame, which the plugin re-sends on every + /// connect and on a cadence — so this is an upsert by construction, not by accident. + pub async fn put_board(&self, kind: &str, t: i64, json: &str) -> anyhow::Result<()> { + sqlx::query( + "INSERT INTO boards (kind, t, json) VALUES (?, ?, ?) + ON CONFLICT(kind) DO UPDATE SET t = excluded.t, json = excluded.json", + ) + .bind(kind) .bind(t) .bind(json) .execute(&self.pool) @@ -131,16 +309,74 @@ impl Store { Ok(()) } - /// The last thing the game said about itself, or `None` if it has never connected. + /// One board, or `None` if the game has never sent it. /// /// This is the read that makes the website render while the game is off, which is the whole /// reason the sidecar holds a database at all. - pub async fn server_state(&self) -> anyhow::Result> { - let row = sqlx::query("SELECT json FROM server_state WHERE id = 1") + pub async fn board(&self, kind: &str) -> anyhow::Result> { + let row = sqlx::query("SELECT json FROM boards WHERE kind = ?") + .bind(kind) .fetch_optional(&self.pool) .await?; Ok(row.and_then(|r| serde_json::from_str(&r.get::("json")).ok())) } + + /// Every board, keyed by kind. What a consumer reads once on connect to know the present + /// before it starts following the story. + pub async fn all_boards(&self) -> anyhow::Result> { + let rows = sqlx::query("SELECT kind, json FROM boards ORDER BY kind") + .fetch_all(&self.pool) + .await?; + + let mut out = serde_json::Map::new(); + + for row in rows { + let kind: String = row.get("kind"); + if let Ok(v) = serde_json::from_str::(&row.get::("json")) { + out.insert(kind, v); + } + } + + Ok(out) + } + + /// Deletes events older than `retain_days`, returning how many went. + /// + /// **Boards are never pruned**, and that asymmetry is the design rather than an oversight: a + /// board is one row per kind holding what is true now, and deleting it would make a server the + /// site has rendered for weeks look like one that has never connected. History is bounded + /// because it grows; the present is not, because it does not. + /// + /// Safe to be aggressive here because the *permanent* record lives on the website — per-wipe + /// rollups in the module's own tables (R12) — and this database sits on a game host whose disk + /// belongs to the operator. + pub async fn prune(&self, retain_days: i64) -> anyhow::Result { + if retain_days <= 0 { + return Ok(0); // retention off; an operator who wants everything keeps everything + } + + let cutoff = now_ms() - retain_days * 86_400_000; + + let done = sqlx::query("DELETE FROM events WHERE t < ?") + .bind(cutoff) + .execute(&self.pool) + .await?; + + let n = done.rows_affected(); + + if n > 0 { + info!(pruned = n, retain_days, "pruned old events"); + } + + Ok(n) + } +} + +fn now_ms() -> i64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_millis() as i64) + .unwrap_or(0) } /// Whether this path names an in-memory database rather than a file. Covers the bare `:memory:` @@ -155,6 +391,12 @@ fn parse_json_column(rows: Vec) -> Vec { .collect() } +/// Warns once about a store write that failed. Persistence failures must never stop the live feed, +/// so every caller logs and carries on; this keeps them saying the same thing. +pub fn warn_write(what: &str, e: &anyhow::Error) { + warn!(error = %e, "{what}"); +} + #[cfg(test)] mod tests { use super::*; @@ -164,6 +406,18 @@ mod tests { Store::open(":memory:").await.unwrap() } + async fn insert(s: &Store, t: i64, kind: &str, wipe: Option<&str>) { + s.insert_event( + t, + kind, + Some("main"), + wipe, + &json!({"kind": kind, "t": t}).to_string(), + ) + .await + .unwrap(); + } + /// The trap this file's pool sizing exists for: a multi-connection pool over `:memory:` hands /// out empty databases. Asserting the *pool* is what makes the reason visible; asserting only /// that a query works would pass again the moment someone "tidied" the sizing back. @@ -179,8 +433,8 @@ mod tests { } /// `sqlx::query` over a multi-statement string is the kind of thing that quietly runs only the - /// first statement. Both tables and both reads have to work on a real file, under the pool - /// size production uses. + /// first statement. Every table and both reads have to work on a real file, under the pool size + /// production uses. #[tokio::test] async fn the_whole_schema_is_created_on_a_pooled_file_store() { let dir = std::env::temp_dir().join(format!("rust-link-test-{}", std::process::id())); @@ -190,10 +444,10 @@ mod tests { let s = Store::open(path.to_str().unwrap()).await.unwrap(); assert_eq!(s.pool.options().get_max_connections(), 4); - s.insert_event(1, "k", "{}").await.unwrap(); - s.put_server_state(1, "{}").await.unwrap(); - assert_eq!(s.recent(None, 10).await.unwrap().len(), 1); - assert!(s.server_state().await.unwrap().is_some()); + insert(&s, 1, "k", None).await; + s.put_board(SERVER_BOARD, 1, "{}").await.unwrap(); + assert_eq!(s.recent(None, None, 10).await.unwrap().len(), 1); + assert!(s.board(SERVER_BOARD).await.unwrap().is_some()); drop(s); let _ = std::fs::remove_dir_all(&dir); @@ -202,70 +456,216 @@ mod tests { #[tokio::test] async fn events_come_back_newest_first_and_filter_by_kind() { let s = store().await; - s.insert_event(1, "server.hello", &json!({"n": 1}).to_string()) - .await - .unwrap(); - s.insert_event(2, "other", &json!({"n": 2}).to_string()) - .await - .unwrap(); - s.insert_event(3, "server.hello", &json!({"n": 3}).to_string()) - .await - .unwrap(); + insert(&s, 1, "server.hello", None).await; + insert(&s, 2, "other", None).await; + insert(&s, 3, "server.hello", None).await; - let all = s.recent(None, 10).await.unwrap(); + let all = s.recent(None, None, 10).await.unwrap(); assert_eq!(all.len(), 3); - assert_eq!(all[0]["n"], 3); + assert_eq!(all[0]["t"], 3); - let hellos = s.recent(Some("server.hello"), 10).await.unwrap(); + let hellos = s.recent(Some("server.hello"), None, 10).await.unwrap(); assert_eq!(hellos.len(), 2); - assert_eq!(hellos[0]["n"], 3); + assert_eq!(hellos[0]["t"], 3); + } + + /// R12 in one test: a wipe splits the history without erasing any of it. + #[tokio::test] + async fn events_filter_by_wipe_without_losing_the_other_wipe() { + let s = store().await; + insert(&s, 1, "player.death", Some("w-a")).await; + insert(&s, 2, "player.death", Some("w-a")).await; + insert(&s, 3, "player.death", Some("w-b")).await; + + assert_eq!(s.recent(None, Some("w-a"), 10).await.unwrap().len(), 2); + assert_eq!(s.recent(None, Some("w-b"), 10).await.unwrap().len(), 1); + assert_eq!(s.recent(None, None, 10).await.unwrap().len(), 3); + // Both filters at once is the combination that is easy to build wrong. + assert_eq!( + s.recent(Some("player.death"), Some("w-b"), 10) + .await + .unwrap() + .len(), + 1 + ); + } + + /// The cursor's two properties, and they are the ones a consumer's correctness rests on: + /// oldest first, and strictly after the id it was given. + #[tokio::test] + async fn the_feed_is_a_cursor_and_runs_oldest_first() { + let s = store().await; + for i in 1..=5 { + insert(&s, i, "player.death", None).await; + } + + let (page, more) = s.feed(0, 2).await.unwrap(); + assert_eq!(page.len(), 2); + assert!(more, "a full page must say there is more"); + assert_eq!(page[0].t, 1); + assert_eq!(page[1].t, 2); + + let (page, more) = s.feed(page[1].id, 10).await.unwrap(); + assert_eq!(page.len(), 3); + assert!(!more, "a short page is the end of the queue"); + assert_eq!(page[0].t, 3); + + // The cursor is exclusive; re-reading from the last id delivers nothing twice. + let (page, _) = s.feed(page[2].id, 10).await.unwrap(); + assert!(page.is_empty()); + } + + #[tokio::test] + async fn a_fresh_consumer_can_start_at_the_end() { + let s = store().await; + assert_eq!(s.last_event_id().await.unwrap(), 0); + + for i in 1..=3 { + insert(&s, i, "k", None).await; + } + + let last = s.last_event_id().await.unwrap(); + assert_eq!(last, 3); + assert!(s.feed(last, 10).await.unwrap().0.is_empty()); } /// An absent board reads as `None`, not as an empty object. A caller must be able to tell /// "the game has never connected" from "the game connected and said nothing" — collapsing the /// two is how a site ends up rendering a server that does not exist. #[tokio::test] - async fn server_state_is_absent_until_a_hello_arrives() { + async fn a_board_is_absent_until_a_snapshot_arrives() { let s = store().await; - assert!(s.server_state().await.unwrap().is_none()); + assert!(s.board(SERVER_BOARD).await.unwrap().is_none()); - s.put_server_state(1, &json!({"serverId": "main", "players": 0}).to_string()) + s.put_board(SERVER_BOARD, 1, &json!({"serverId": "main"}).to_string()) .await .unwrap(); - assert_eq!(s.server_state().await.unwrap().unwrap()["serverId"], "main"); + assert_eq!( + s.board(SERVER_BOARD).await.unwrap().unwrap()["serverId"], + "main" + ); } - /// The board holds exactly one row however many times the plugin reconnects. + /// A board holds exactly one row however many times the plugin reconnects, and the boards are + /// independent of one another. #[tokio::test] - async fn a_second_hello_replaces_the_first() { + async fn a_second_snapshot_replaces_the_first_of_its_own_kind_only() { let s = store().await; - s.put_server_state(1, &json!({"bootId": "a"}).to_string()) + s.put_board(SERVER_BOARD, 1, &json!({"bootId": "a"}).to_string()) .await .unwrap(); - s.put_server_state(2, &json!({"bootId": "b"}).to_string()) + s.put_board(SERVER_BOARD, 2, &json!({"bootId": "b"}).to_string()) + .await + .unwrap(); + s.put_board("players.online", 2, &json!({"count": 4}).to_string()) .await .unwrap(); - assert_eq!(s.server_state().await.unwrap().unwrap()["bootId"], "b"); - let count: i64 = sqlx::query("SELECT COUNT(*) AS c FROM server_state") - .fetch_one(&s.pool) - .await - .unwrap() - .get("c"); - assert_eq!(count, 1); + assert_eq!(s.board(SERVER_BOARD).await.unwrap().unwrap()["bootId"], "b"); + assert_eq!( + s.board("players.online").await.unwrap().unwrap()["count"], + 4 + ); + + let all = s.all_boards().await.unwrap(); + assert_eq!(all.len(), 2); } #[tokio::test] async fn the_limit_is_clamped_rather_than_trusted() { let s = store().await; for i in 0..5 { - s.insert_event(i, "k", &json!({"i": i}).to_string()) - .await - .unwrap(); + insert(&s, i, "k", None).await; } // 0 and negatives would otherwise mean "no rows" and "SQLite's unlimited" respectively. - assert_eq!(s.recent(None, 0).await.unwrap().len(), 1); - assert_eq!(s.recent(None, -1).await.unwrap().len(), 1); - assert_eq!(s.recent(None, 100_000).await.unwrap().len(), 5); + assert_eq!(s.recent(None, None, 0).await.unwrap().len(), 1); + assert_eq!(s.recent(None, None, -1).await.unwrap().len(), 1); + assert_eq!(s.recent(None, None, 100_000).await.unwrap().len(), 5); + assert_eq!(s.feed(0, 0).await.unwrap().0.len(), 1); + } + + /// Retention deletes history and leaves the present alone. The second half is the half worth + /// asserting: a pruned board is a server that has "never connected". + #[tokio::test] + async fn pruning_bounds_the_history_and_never_touches_a_board() { + let s = store().await; + let old = now_ms() - 30 * 86_400_000; + + insert(&s, old, "player.death", None).await; + insert(&s, now_ms(), "player.death", None).await; + s.put_board(SERVER_BOARD, old, &json!({"serverId": "main"}).to_string()) + .await + .unwrap(); + + assert_eq!(s.prune(14).await.unwrap(), 1); + assert_eq!(s.recent(None, None, 10).await.unwrap().len(), 1); + assert!(s.board(SERVER_BOARD).await.unwrap().is_some()); + + // Retention off keeps everything, which is a supported configuration rather than a bug. + insert(&s, old, "player.death", None).await; + assert_eq!(s.prune(0).await.unwrap(), 0); + assert_eq!(s.recent(None, None, 10).await.unwrap().len(), 2); + } + + /// The upgrade path, on a real file because that is the only place it can happen: a protocol 1 + /// database has `server_state` and no `wipe_id`, and opening it with this build must produce a + /// store that still knows which server it is holding. + #[tokio::test] + async fn a_protocol_1_database_is_migrated_in_place() { + let dir = std::env::temp_dir().join(format!("rust-link-migrate-{}", std::process::id())); + let path = dir.join("old.db"); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).unwrap(); + + // Exactly protocol 1's schema, written by hand so the test does not depend on this file + // still being able to produce it. + let opts = SqliteConnectOptions::new() + .filename(&path) + .create_if_missing(true); + let pool = SqlitePoolOptions::new() + .max_connections(1) + .connect_with(opts) + .await + .unwrap(); + sqlx::query( + "CREATE TABLE events (id INTEGER PRIMARY KEY AUTOINCREMENT, t INTEGER NOT NULL, + kind TEXT NOT NULL, json TEXT NOT NULL); + CREATE TABLE server_state (id INTEGER PRIMARY KEY CHECK (id = 1), t INTEGER NOT NULL, + json TEXT NOT NULL);", + ) + .execute(&pool) + .await + .unwrap(); + sqlx::query("INSERT INTO events (t, kind, json) VALUES (1, 'server.hello', '{\"t\":1}')") + .execute(&pool) + .await + .unwrap(); + sqlx::query( + "INSERT INTO server_state (id, t, json) VALUES (1, 1, '{\"serverId\":\"legacy\"}')", + ) + .execute(&pool) + .await + .unwrap(); + pool.close().await; + + let s = Store::open(path.to_str().unwrap()).await.unwrap(); + + // The columns arrived, the old rows survived with them empty, and the board came across — + // so an upgraded sidecar does not report a server it has been serving for weeks as one it + // has never heard of. + assert!(s.has_column("events", "wipe_id").await.unwrap()); + assert_eq!(s.recent(None, None, 10).await.unwrap().len(), 1); + assert_eq!( + s.board(SERVER_BOARD).await.unwrap().unwrap()["serverId"], + "legacy" + ); + + // And it is idempotent: opening again must not fail on an ALTER that already ran. + drop(s); + let again = Store::open(path.to_str().unwrap()).await.unwrap(); + assert!(again.board(SERVER_BOARD).await.unwrap().is_some()); + + drop(again); + let _ = std::fs::remove_dir_all(&dir); } } diff --git a/sidecar/src/web.rs b/sidecar/src/web.rs index ef040b1..969919f 100644 --- a/sidecar/src/web.rs +++ b/sidecar/src/web.rs @@ -15,7 +15,7 @@ //! route that fails when the game is down — which is the honest answer to "what is it doing *right //! now*". -use std::sync::atomic::{AtomicI64, Ordering}; +use std::sync::atomic::{AtomicI64, AtomicU64, Ordering}; use std::sync::Arc; use std::time::{Duration, Instant}; @@ -57,6 +57,8 @@ pub struct AppState { pub started: Instant, /// Epoch ms of the last line received from the plugin, 0 if none yet. pub last_event: Arc, + /// Frames this build could not file, because they carried no usable `type`. See `/health`. + pub untyped: Arc, } pub async fn serve(addr: &str, state: AppState) -> anyhow::Result<()> { @@ -64,8 +66,14 @@ pub async fn serve(addr: &str, state: AppState) -> anyhow::Result<()> { .route(WS_PATH, get(ws_upgrade)) // The board: the last `server.hello`. Store-backed, so it answers while the game is off. .route("/server", get(server_board)) - // Event history, newest first, optionally filtered by kind. + // Every board at once — what a consumer reads on connect to know the present before it + // starts following the story. + .route("/boards", get(boards)) + // Event history, newest first, optionally filtered by kind and wipe. For a human. .route("/events", get(events)) + // The ingest cursor: oldest first, strictly after an id. For a consumer that must not miss + // a row. Deliberately a second route rather than a flag on the first — see `feed`. + .route("/feed", get(feed)) // Live: a correlated round trip to the plugin. Fails when the game is down, by design. .route("/status", get(status)) .route_layer(middleware::from_fn_with_state(state.clone(), gate)); @@ -95,6 +103,8 @@ async fn health(State(st): State) -> impl IntoResponse { let status = if plugin && db_ok { "ok" } else { "degraded" }; + let untyped = st.untyped.load(Ordering::Relaxed); + Json(json!({ "status": status, "protocol": PROTOCOL_VERSION, @@ -102,6 +112,10 @@ async fn health(State(st): State) -> impl IntoResponse { "database": if db_ok { "ok" } else { "error" }, "uptime": format_uptime(st.started.elapsed()), "last_event": iso_ms(last_ms), + // Non-zero means the plugin is speaking a protocol this build cannot file. It is reported + // here rather than only in the log because the symptom an operator sees — a website that + // shows nothing while the game is plainly up — names nothing at all. + "untyped_frames": untyped, })) } @@ -133,7 +147,7 @@ fn iso_ms(ms: i64) -> Option { /// reports nothing" are different answers, and a client that cannot tell them apart renders a /// server that does not exist. async fn server_board(State(st): State) -> Response { - match st.store.server_state().await { + match st.store.board(crate::store::SERVER_BOARD).await { Ok(Some(v)) => Json(v).into_response(), Ok(None) => StatusCode::NO_CONTENT.into_response(), Err(e) => { @@ -143,16 +157,30 @@ async fn server_board(State(st): State) -> Response { } } +/// Every board, keyed by kind. `200` with an empty object when the game has never connected — +/// unlike `/server`, which answers `204`, because "no boards yet" is a complete answer to "give me +/// all of them" and an empty map renders correctly where a `204` has to be special-cased. +async fn boards(State(st): State) -> Response { + match st.store.all_boards().await { + Ok(map) => Json(json!({ "boards": map })).into_response(), + Err(e) => { + warn!(error = %e, "board read failed"); + store_error() + } + } +} + #[derive(Debug, Deserialize)] struct EventsQuery { kind: Option, + wipe: Option, limit: Option, } async fn events(State(st): State, Query(q): Query) -> Response { match st .store - .recent(q.kind.as_deref(), q.limit.unwrap_or(50)) + .recent(q.kind.as_deref(), q.wipe.as_deref(), q.limit.unwrap_or(50)) .await { Ok(rows) => Json(json!({ "events": rows })).into_response(), @@ -163,6 +191,55 @@ async fn events(State(st): State, Query(q): Query) -> Res } } +#[derive(Debug, Deserialize)] +struct FeedQuery { + since: Option, + limit: Option, +} + +/// The ingest cursor: everything after `since`, oldest first. +/// +/// **Omitting `since` asks where the end is** — it answers with no rows and the current `lastId`, +/// which is what a consumer with no cursor of its own needs. `since=0` is the other question, and +/// the one nobody should ask by accident: replay everything retained. A module installed today +/// against a sidecar that has been running a month wants what happens next, not a fortnight of +/// deaths it has no rollups for, and the difference between those two intentions must not be the +/// difference between typing a parameter and forgetting it. +/// +/// The response always carries `lastId`, so a caller advances without inspecting the rows, and +/// `more`, so one that has fallen behind comes straight back rather than waiting out its poll +/// interval. +async fn feed(State(st): State, Query(q): Query) -> Response { + let since = match q.since { + Some(n) => n.max(0), + None => match st.store.last_event_id().await { + Ok(id) => { + return Json(json!({ "items": [], "lastId": id, "more": false })).into_response() + } + Err(e) => { + warn!(error = %e, "feed tail read failed"); + return store_error(); + } + }, + }; + + match st.store.feed(since, q.limit.unwrap_or(200)).await { + Ok((items, more)) => { + // The cursor a caller should send next. When the page is empty that is the cursor it + // sent — never 0, which would silently replay the whole retained history on the next + // poll of a quiet server. + let last_id = items.last().map(|i| i.id).unwrap_or(since); + let rows: Vec = items.iter().map(|i| i.to_json()).collect(); + + Json(json!({ "items": rows, "lastId": last_id, "more": more })).into_response() + } + Err(e) => { + warn!(error = %e, "feed read failed"); + store_error() + } + } +} + fn store_error() -> Response { ( StatusCode::INTERNAL_SERVER_ERROR, -- 2.49.1