RunicGateway/Rust-Link
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(sidecar): protocol 2 — file by type, a cursor feed, and bounded history #2

Merged
whitlocktech merged 1 commits from feat/phase-3-read-path into main 2026-09-16 16:37:28 +00:00
Conversation 0 Commits 1 Files Changed 8 +910 -116
8 changed files with 910 additions and 116 deletions
Download Patch File Download Diff File Expand all files Collapse all files

99
.gitea/workflows/pr-checks.yml Normal file
View File

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

32
sidecar/README.md
View File

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

186
sidecar/src/app.rs
View File

@@ -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 ev.kind.as_str() {
"server.hello" => {
match Filing::of(&frame_type) {
Filing::Persist => {
total += 1;
tracing::debug!(kind = %ev.kind, n = total, "{}", line);
if let Err(e) = event_store
.insert_event(
frame_t(&ev.value),
&ev.kind,
field(&ev.value, "serverId"),
field(&ev.value, "wipeId"),
&line,
)
.await
{
store::warn_write("event persist failed", &e);
}
}
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, n = total, "{}", line);
if let Err(e) = event_store
.put_server_state(frame_t(&ev.value), &line)
.await
{
warn!(error = %e, "server board write failed");
}
}
"link.down" => info!("game link down"),
_ => tracing::debug!(kind = %ev.kind, n = total, "{}", line),
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)
.put_board(&ev.kind, frame_t(&ev.value), &line)
.await
{
warn!(error = %e, "event persist failed");
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]

29
sidecar/src/config.rs
View File

@@ -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::<i64>() {
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
"#
)
}

7
sidecar/src/game.rs
View File

@@ -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) => {

18
sidecar/src/main.rs
View File

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

548
sidecar/src/store.rs
View File

@@ -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,
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<bool> {
// `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::<String, _>("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 (?, ?, ?)")
///
/// `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<Vec<Value>> {
/// 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<Vec<Value>> {
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?
// 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 = ?");
}
None => {
sqlx::query("SELECT json FROM events ORDER BY id DESC LIMIT ?")
.bind(limit)
.fetch_all(&self.pool)
.await?
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<FeedItem>, 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<i64> {
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<Option<Value>> {
let row = sqlx::query("SELECT json FROM server_state WHERE id = 1")
pub async fn board(&self, kind: &str) -> anyhow::Result<Option<Value>> {
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::<String, _>("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<serde_json::Map<String, Value>> {
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::<Value>(&row.get::<String, _>("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<u64> {
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<sqlx::sqlite::SqliteRow>) -> Vec<Value> {
.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);
}
}

85
sidecar/src/web.rs
View File

@@ -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<AtomicI64>,
/// Frames this build could not file, because they carried no usable `type`. See `/health`.
pub untyped: Arc<AtomicU64>,
}
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<AppState>) -> 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<AppState>) -> 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<String> {
/// 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<AppState>) -> 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<AppState>) -> 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<AppState>) -> 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<String>,
wipe: Option<String>,
limit: Option<i64>,
}
async fn events(State(st): State<AppState>, Query(q): Query<EventsQuery>) -> 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<AppState>, Query(q): Query<EventsQuery>) -> Res
}
}
#[derive(Debug, Deserialize)]
struct FeedQuery {
since: Option<i64>,
limit: Option<i64>,
}
/// 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<AppState>, Query(q): Query<FeedQuery>) -> 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<Value> = 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,