Compare commits
27 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| baa04e1a76 | |||
| 143f424867 | |||
| 60e6de55f3 | |||
| b92393d224 | |||
| 6d83df0a2c | |||
| a8f1804de9 | |||
| f39dfa4f84 | |||
| d83bb1748c | |||
| 5cdd80e694 | |||
| 5d909ca0a3 | |||
| d38a9e8a75 | |||
| 93411966d7 | |||
| d13ad11eb0 | |||
| 5612fba744 | |||
| 8b9dd0d9e8 | |||
| f41237392d | |||
| d0c2e7d6e1 | |||
| 4b8ea768b6 | |||
| 6fb063818a | |||
| 7499e099f4 | |||
| 2408d31ff7 | |||
| b00f2719a4 | |||
| 9216006208 | |||
| be0efd348c | |||
| 3dbc2f490c | |||
| 7b6584006e | |||
| 36141a23df |
@@ -17,10 +17,12 @@
|
|||||||
# so let this workflow run on one PR first. The `PR Checks / *` glob matches
|
# so let this workflow run on one PR first. The `PR Checks / *` glob matches
|
||||||
# without needing the dropdown.
|
# without needing the dropdown.
|
||||||
#
|
#
|
||||||
# Scope note: this gates PRs into `main` only. Feature work that lands on an
|
# Scope note: `edge` is gated as well as `main`. Multi-phase work lands there
|
||||||
# integration branch first (e.g. `edge`) is still caught on the branch's PR into
|
# first, so gating only the `main` hop would run these checks for the first time
|
||||||
# `main`. To gate that earlier hop too, add the branch to the `branches:` list
|
# at the cutover — the one moment a red build is most expensive to discover. This
|
||||||
# below — nothing else needs to change.
|
# is the same call `RunicGateway/installer` made for the same reason, and it was
|
||||||
|
# taken here after a nine-PR Android workstream landed on an ungated `edge` with
|
||||||
|
# no CI at all. Adding a branch to the `branches:` list is the whole change.
|
||||||
#
|
#
|
||||||
# Runner: the same self-hosted `ubuntu-latest` runner release.yml uses. Rust is
|
# Runner: the same self-hosted `ubuntu-latest` runner release.yml uses. Rust is
|
||||||
# not assumed to be preinstalled, so the toolchain step bootstraps it the same
|
# not assumed to be preinstalled, so the toolchain step bootstraps it the same
|
||||||
@@ -31,7 +33,7 @@ name: PR Checks
|
|||||||
|
|
||||||
on:
|
on:
|
||||||
pull_request:
|
pull_request:
|
||||||
branches: [main]
|
branches: [main, edge]
|
||||||
|
|
||||||
# A newer push to the same PR cancels the in-flight run.
|
# A newer push to the same PR cancels the in-flight run.
|
||||||
concurrency:
|
concurrency:
|
||||||
|
|||||||
@@ -149,6 +149,39 @@ jobs:
|
|||||||
fi
|
fi
|
||||||
fi
|
fi
|
||||||
|
|
||||||
|
# ── Orphan sweep ────────────────────────────────────────────────
|
||||||
|
#
|
||||||
|
# The check above is VERSION-SCOPED: it only ever asks about the one
|
||||||
|
# version this run computed. That is enough to recover an orphan on
|
||||||
|
# the very next run, and useless afterwards — once any releasable
|
||||||
|
# commit lands, the next run computes a NEW version, never looks at
|
||||||
|
# the old tag again, and the orphan becomes permanent and silent.
|
||||||
|
#
|
||||||
|
# servuo-plugins v0.1.0 is the proof, and the proof is pointed: the
|
||||||
|
# commit that ADDED the recovery above was itself typed
|
||||||
|
# `fix(release): ... recover the orphaned v0.1.0 tag`, so it bumped to
|
||||||
|
# v0.1.1 — and the run that introduced the recovery stepped straight
|
||||||
|
# past the tag it was written to rescue. That tag is still orphaned.
|
||||||
|
#
|
||||||
|
# So every v* tag is checked, and anything missing a release is
|
||||||
|
# WARNED about. Deliberately not recovered: publishing an old version
|
||||||
|
# would mean building today's tree and shipping it under a tag whose
|
||||||
|
# tree it is not, which is worse than the inconsistency it fixes.
|
||||||
|
# A human decides whether to recover or drop it.
|
||||||
|
#
|
||||||
|
# Never fails the run. A sweep that can break a good release is a
|
||||||
|
# sweep someone will delete.
|
||||||
|
ORPHANS=""
|
||||||
|
for T in $(git tag -l 'v*' --sort=-v:refname); do
|
||||||
|
T_HTTP="$(curl -s -o /dev/null -w '%{http_code}' \
|
||||||
|
-H "Authorization: token $(printf '%s' "${REGISTRY_TOKEN:-}" | tr -d '\r\n')" \
|
||||||
|
"https://${GITEA_HOST}/api/v1/repos/${REPO}/releases/tags/${T}" || echo 000)"
|
||||||
|
[ "$T_HTTP" = "404" ] && ORPHANS="${ORPHANS} ${T}"
|
||||||
|
done
|
||||||
|
if [ -n "${ORPHANS}" ]; then
|
||||||
|
echo "::warning::Tags with no release:${ORPHANS} — a run failed after tagging. Publish or delete them; this job will not do either."
|
||||||
|
fi
|
||||||
|
|
||||||
# Changelog range. A recovery run has nothing after the tag, so
|
# Changelog range. A recovery run has nothing after the tag, so
|
||||||
# summarize what the tag itself contains rather than emitting an empty
|
# summarize what the tag itself contains rather than emitting an empty
|
||||||
# list: the range that produced it, i.e. previous-tag..this-tag.
|
# list: the range that produced it, i.e. previous-tag..this-tag.
|
||||||
@@ -279,33 +312,43 @@ jobs:
|
|||||||
ls -l dist && echo "----" && cat dist/SHA256SUMS
|
ls -l dist && echo "----" && cat dist/SHA256SUMS
|
||||||
|
|
||||||
# ── RELEASE ENGINE: commit the bump, tag, push ───────────────────────
|
# ── RELEASE ENGINE: commit the bump, tag, push ───────────────────────
|
||||||
- name: Commit version bump and push tag
|
# Tag only — `main` is never pushed to.
|
||||||
|
#
|
||||||
|
# This step used to commit the version bump back to main first. Two things
|
||||||
|
# were wrong with that. It has never once executed: an EMPTY template
|
||||||
|
# expression written literally in a comment (the `$`+`{{ }}` token, which
|
||||||
|
# is why it is spelled out here) made the runner fail to build the script
|
||||||
|
# and skip the whole step silently, which is why sidecar/Cargo.toml still
|
||||||
|
# says 0.1.0 after six releases (the tags exist because the release API
|
||||||
|
# creates one when it publishes). And had it executed, it would have been
|
||||||
|
# declined — main is protected, and a release must not depend on a write
|
||||||
|
# to a protected branch.
|
||||||
|
#
|
||||||
|
# So the tag is the version, as it already is in servuo-plugins. The
|
||||||
|
# workflow still writes the real version into Cargo.toml before building,
|
||||||
|
# so a released binary self-reports correctly; what it no longer does is
|
||||||
|
# commit that edit back. The next version is computed from the newest tag,
|
||||||
|
# never from Cargo.toml, so nothing downstream depends on the file.
|
||||||
|
- name: Push the release tag
|
||||||
if: ${{ steps.plan.outputs.release == 'true' }}
|
if: ${{ steps.plan.outputs.release == 'true' }}
|
||||||
env:
|
env:
|
||||||
REGISTRY_USER: ${{ secrets.REGISTRY_USER }}
|
REGISTRY_USER: ${{ secrets.REGISTRY_USER }}
|
||||||
REGISTRY_TOKEN: ${{ secrets.REGISTRY_TOKEN }}
|
REGISTRY_TOKEN: ${{ secrets.REGISTRY_TOKEN }}
|
||||||
run: |
|
run: |
|
||||||
set -euo pipefail
|
set -euo pipefail
|
||||||
VERSION="${{ steps.plan.outputs.version }}"
|
|
||||||
TAG="${{ steps.plan.outputs.tag }}"
|
TAG="${{ steps.plan.outputs.tag }}"
|
||||||
# Secrets can arrive with a trailing newline (depending on how they were
|
# Secrets can arrive with a trailing newline (depending on how they were
|
||||||
# pasted); a stray CR/LF corrupts the remote URL ("credential url cannot
|
# pasted); a stray CR/LF corrupts the remote URL ("credential url cannot
|
||||||
# be parsed"). Strip line breaks before building the URL. Passing them via
|
# be parsed"). Strip line breaks before building the URL. They are passed
|
||||||
# env (not inline ${{ }}) also keeps a newline from breaking this script.
|
# via env rather than interpolated into this script, so a newline cannot
|
||||||
|
# break it — do NOT write a template token literally in a comment here,
|
||||||
|
# or the runner will skip this step without failing the job.
|
||||||
CI_USER="$(printf '%s' "${REGISTRY_USER}" | tr -d '\r\n')"
|
CI_USER="$(printf '%s' "${REGISTRY_USER}" | tr -d '\r\n')"
|
||||||
CI_TOKEN="$(printf '%s' "${REGISTRY_TOKEN}" | tr -d '\r\n')"
|
CI_TOKEN="$(printf '%s' "${REGISTRY_TOKEN}" | tr -d '\r\n')"
|
||||||
git config user.name "uo-link-ci"
|
git config user.name "uo-link-ci"
|
||||||
git config user.email "ci@whitlocktech.com"
|
git config user.email "ci@whitlocktech.com"
|
||||||
git remote set-url origin \
|
git remote set-url origin \
|
||||||
"https://${CI_USER}:${CI_TOKEN}@${GITEA_HOST}/${REPO}.git"
|
"https://${CI_USER}:${CI_TOKEN}@${GITEA_HOST}/${REPO}.git"
|
||||||
|
|
||||||
git add "${WORKDIR}/Cargo.toml" "${WORKDIR}/Cargo.lock"
|
|
||||||
if ! git diff --cached --quiet; then
|
|
||||||
git commit -m "chore(release): bump version to ${TAG} [skip ci]"
|
|
||||||
git push origin "HEAD:main"
|
|
||||||
else
|
|
||||||
echo "Version unchanged (first release) — no bump commit needed."
|
|
||||||
fi
|
|
||||||
# The tag may already exist when finishing a run that died after tagging
|
# The tag may already exist when finishing a run that died after tagging
|
||||||
# (see the plan step). `git tag` on an existing name fails under
|
# (see the plan step). `git tag` on an existing name fails under
|
||||||
# `set -e`; pushing an identical existing tag is a harmless no-op. A
|
# `set -e`; pushing an identical existing tag is a harmless no-op. A
|
||||||
@@ -332,18 +375,75 @@ jobs:
|
|||||||
# corrupt the Authorization header.
|
# corrupt the Authorization header.
|
||||||
CI_TOKEN="$(printf '%s' "${REGISTRY_TOKEN}" | tr -d '\r\n')"
|
CI_TOKEN="$(printf '%s' "${REGISTRY_TOKEN}" | tr -d '\r\n')"
|
||||||
|
|
||||||
REL_ID="$(curl -sSf -X POST "${API}/releases" \
|
PAYLOAD="$(jq -n --arg tag "$TAG" --arg body "$BODY" \
|
||||||
|
'{tag_name:$tag, name:$tag, body:$body, draft:false, prerelease:false}')"
|
||||||
|
|
||||||
|
# This POST is the step that orphaned tag v0.1.1 (run 75): it landed one
|
||||||
|
# second after the tag push and Gitea answered 500, having not finished
|
||||||
|
# processing the pushed tag. Re-running the workflow published the same
|
||||||
|
# four assets untouched, so the failure was a race, not a bad request.
|
||||||
|
#
|
||||||
|
# Two things went wrong there, and both are fixed here.
|
||||||
|
#
|
||||||
|
# 1. `curl -sSf` prints NO response body on an error status, so all the
|
||||||
|
# log carried was "curl: (22) ... error: 500" and the cause had to be
|
||||||
|
# inferred from timestamps. Capture the body and print it.
|
||||||
|
# 2. Nothing retried, so a transient 5xx became a permanent orphan tag.
|
||||||
|
# The plan step CAN recover one, but only on a run that reaches it --
|
||||||
|
# and a later push with no releasable commits stands down before it
|
||||||
|
# gets there, so in practice the tag sits until a human notices.
|
||||||
|
#
|
||||||
|
# 4xx is deliberately NOT retried: a bad token or a malformed body does
|
||||||
|
# not improve by being sent again, and retrying only turns a clear
|
||||||
|
# failure into a slow one.
|
||||||
|
REL_ID=""
|
||||||
|
for attempt in 1 2 3 4 5; do
|
||||||
|
HTTP="$(curl -s -o /tmp/rel.json -w '%{http_code}' -X POST "${API}/releases" \
|
||||||
-H "Authorization: token ${CI_TOKEN}" \
|
-H "Authorization: token ${CI_TOKEN}" \
|
||||||
-H "Content-Type: application/json" \
|
-H "Content-Type: application/json" \
|
||||||
-d "$(jq -n --arg tag "$TAG" --arg body "$BODY" \
|
-d "${PAYLOAD}" || echo 000)"
|
||||||
'{tag_name:$tag, name:$tag, body:$body, draft:false, prerelease:false}')" \
|
|
||||||
| jq -r '.id')"
|
if [ "$HTTP" = "201" ] || [ "$HTTP" = "200" ]; then
|
||||||
|
REL_ID="$(jq -r '.id' /tmp/rel.json)"
|
||||||
|
break
|
||||||
|
fi
|
||||||
|
|
||||||
|
echo "::warning::POST /releases attempt ${attempt} returned HTTP ${HTTP}"
|
||||||
|
echo "--- response body ---"
|
||||||
|
cat /tmp/rel.json || true
|
||||||
|
echo
|
||||||
|
echo "---------------------"
|
||||||
|
|
||||||
|
case "$HTTP" in
|
||||||
|
4*) echo "::error::HTTP ${HTTP} is a client error - not retrying."; exit 1 ;;
|
||||||
|
esac
|
||||||
|
|
||||||
|
if [ "$attempt" = 5 ]; then
|
||||||
|
echo "::error::POST /releases still failing after 5 attempts. Tag ${TAG} is pushed but has no release."
|
||||||
|
echo "::error::Re-run this workflow - the plan step detects the orphan tag and republishes it."
|
||||||
|
exit 1
|
||||||
|
fi
|
||||||
|
sleep $(( attempt * 5 ))
|
||||||
|
done
|
||||||
|
|
||||||
|
if [ -z "$REL_ID" ] || [ "$REL_ID" = "null" ]; then
|
||||||
|
echo "::error::Release created but no id came back; refusing to upload assets blind."
|
||||||
|
exit 1
|
||||||
|
fi
|
||||||
echo "Created release ${TAG} (id=${REL_ID})"
|
echo "Created release ${TAG} (id=${REL_ID})"
|
||||||
|
|
||||||
for f in "${BIN}-linux-x86_64" "${BIN}-linux-aarch64" "${BIN}-windows-x86_64.exe" SHA256SUMS; do
|
for f in "${BIN}-linux-x86_64" "${BIN}-linux-aarch64" "${BIN}-windows-x86_64.exe" SHA256SUMS; do
|
||||||
curl -sSf -X POST "${API}/releases/${REL_ID}/assets?name=${f}" \
|
# Same treatment. An upload that fails quietly leaves a release whose
|
||||||
|
# SHA256SUMS does not cover every binary it advertises, which is worse
|
||||||
|
# than no release at all -- that file IS the trust anchor.
|
||||||
|
HTTP="$(curl -s -o /tmp/asset.json -w '%{http_code}' -X POST "${API}/releases/${REL_ID}/assets?name=${f}" \
|
||||||
-H "Authorization: token ${CI_TOKEN}" \
|
-H "Authorization: token ${CI_TOKEN}" \
|
||||||
-F "attachment=@dist/${f}" >/dev/null
|
-F "attachment=@dist/${f}" || echo 000)"
|
||||||
|
if [ "$HTTP" != "201" ] && [ "$HTTP" != "200" ]; then
|
||||||
|
echo "::error::uploading ${f} returned HTTP ${HTTP}"
|
||||||
|
cat /tmp/asset.json || true
|
||||||
|
exit 1
|
||||||
|
fi
|
||||||
echo " uploaded ${f}"
|
echo " uploaded ${f}"
|
||||||
done
|
done
|
||||||
|
|
||||||
|
|||||||
30
README.md
30
README.md
@@ -12,11 +12,34 @@ ServUO plugin (C#, net48) ──loopback TCP, newline-JSON──► Rust sidec
|
|||||||
The shard never speaks WebSocket and exposes no port of its own — the sidecar is the only
|
The shard never speaks WebSocket and exposes no port of its own — the sidecar is the only
|
||||||
network-facing component, which is what keeps the game unreachable from the internet.
|
network-facing component, which is what keeps the game unreachable from the internet.
|
||||||
|
|
||||||
|
## Running a shard? Don't build this
|
||||||
|
|
||||||
|
The [**Runic Gateway installer**](https://gitea.whitlocktech.com/RunicGateway/installer) installs
|
||||||
|
this sidecar for you — the released binary, its config, a hardened service account and the service
|
||||||
|
registration — alongside the shard plugin, in one run, on Linux or Windows:
|
||||||
|
|
||||||
|
```bash
|
||||||
|
sudo ./runicgateway-installer-linux-x86_64 install
|
||||||
|
```
|
||||||
|
|
||||||
|
It ends by printing the base URL, WebSocket URL, protocol version and auth token to paste into
|
||||||
|
**Admin → Shard** on your site. Guide:
|
||||||
|
[installer/INSTALL.md](https://gitea.whitlocktech.com/RunicGateway/docs/src/branch/main/installer/INSTALL.md).
|
||||||
|
|
||||||
|
Installing it yourself is supported too — the release binaries on this repo's
|
||||||
|
[releases page](https://gitea.whitlocktech.com/RunicGateway/link/releases) are the same ones the
|
||||||
|
installer fetches, and
|
||||||
|
[INSTALL.md Appendix A3–A4](https://gitea.whitlocktech.com/RunicGateway/docs/src/branch/main/installer/INSTALL.md#a3-install-the-sidecar)
|
||||||
|
covers placing the binary and registering the service by hand.
|
||||||
|
|
||||||
|
Everything below this line is for **developing on the sidecar**.
|
||||||
|
|
||||||
## Related repos
|
## Related repos
|
||||||
|
|
||||||
| Repo | What |
|
| Repo | What |
|
||||||
|------|------|
|
|------|------|
|
||||||
| **this** — `RunicGateway/link` | The Rust sidecar (`sidecar/`). |
|
| **this** — `RunicGateway/link` | The Rust sidecar (`sidecar/`). |
|
||||||
|
| [RunicGateway/installer](https://gitea.whitlocktech.com/RunicGateway/installer) | The **installer** — deploys this sidecar and the plugin onto a shard host. The supported way to set one up. |
|
||||||
| [RunicGateway/servuo-plugins](https://gitea.whitlocktech.com/RunicGateway/servuo-plugins) | The **C# ServUO plugin** — the shard side of the bridge (`overlay/`, `patches/`, `deploy.ps1`, test scaffolding). |
|
| [RunicGateway/servuo-plugins](https://gitea.whitlocktech.com/RunicGateway/servuo-plugins) | The **C# ServUO plugin** — the shard side of the bridge (`overlay/`, `patches/`, `deploy.ps1`, test scaffolding). |
|
||||||
| [RunicGateway/docs](https://gitea.whitlocktech.com/RunicGateway/docs) | All project documentation — design docs, protocol spec, integration guide, research. |
|
| [RunicGateway/docs](https://gitea.whitlocktech.com/RunicGateway/docs) | All project documentation — design docs, protocol spec, integration guide, research. |
|
||||||
|
|
||||||
@@ -28,9 +51,10 @@ network-facing component, which is what keeps the game unreachable from the inte
|
|||||||
| `.gitea/workflows/pr-checks.yml` | Gates every PR into `main` on `cargo fmt --check`, `cargo clippy -D warnings`, and `cargo test`. |
|
| `.gitea/workflows/pr-checks.yml` | Gates every PR into `main` on `cargo fmt --check`, `cargo clippy -D warnings`, and `cargo test`. |
|
||||||
| `.gitea/workflows/release.yml` | Builds + releases the sidecar binary (Linux + Windows) on every merge to `main`. |
|
| `.gitea/workflows/release.yml` | Builds + releases the sidecar binary (Linux + Windows) on every merge to `main`. |
|
||||||
|
|
||||||
## Build & run
|
## Build & run (development)
|
||||||
|
|
||||||
The sidecar is a standard cargo crate:
|
Building from source is for working *on* the sidecar; a deployment gets its binary from a release,
|
||||||
|
via the installer or by hand. The sidecar is a standard cargo crate:
|
||||||
|
|
||||||
```bash
|
```bash
|
||||||
cd sidecar
|
cd sidecar
|
||||||
@@ -39,7 +63,7 @@ cp sidecar.toml.example sidecar.toml # then edit
|
|||||||
cargo run --release
|
cargo run --release
|
||||||
```
|
```
|
||||||
|
|
||||||
Deploying it rather than developing on it: `--config <PATH>` names the config file (as does
|
Deploying it by hand rather than developing on it: `--config <PATH>` names the config file (as does
|
||||||
`$UOLINK_CONFIG`), and `--print-config` prints the resolved settings — **including the auth token
|
`$UOLINK_CONFIG`), and `--print-config` prints the resolved settings — **including the auth token
|
||||||
the website needs** — as JSON, provisioning the config file on first run. That is the supported way
|
the website needs** — as JSON, provisioning the config file on first run. That is the supported way
|
||||||
to read the token back; it is not meant to be scraped from the log.
|
to read the token back; it is not meant to be scraped from the log.
|
||||||
|
|||||||
@@ -14,11 +14,13 @@
|
|||||||
//! - `shutdown` is whatever "stop" means on this host: Ctrl-C and `SIGTERM` on Unix, the SCM's
|
//! - `shutdown` is whatever "stop" means on this host: Ctrl-C and `SIGTERM` on Unix, the SCM's
|
||||||
//! `Stop` control on Windows.
|
//! `Stop` control on Windows.
|
||||||
|
|
||||||
|
use std::collections::HashMap;
|
||||||
use std::future::Future;
|
use std::future::Future;
|
||||||
use std::sync::atomic::AtomicI64;
|
use std::sync::atomic::AtomicI64;
|
||||||
use std::sync::Arc;
|
use std::sync::Arc;
|
||||||
use std::time::Instant;
|
use std::time::Instant;
|
||||||
|
|
||||||
|
use serde_json::Value;
|
||||||
use tokio::sync::{broadcast, mpsc};
|
use tokio::sync::{broadcast, mpsc};
|
||||||
use tracing::info;
|
use tracing::info;
|
||||||
|
|
||||||
@@ -88,6 +90,9 @@ where
|
|||||||
let last_event_ts = last_event.clone();
|
let last_event_ts = last_event.clone();
|
||||||
let replay_handle = handle.clone(); // re-push external news to the shard on (re)connect
|
let replay_handle = handle.clone(); // re-push external news to the shard on (re)connect
|
||||||
let mut total: u64 = 0;
|
let mut total: u64 = 0;
|
||||||
|
// Partly-received guild rosters, keyed by guild id. Lives in the event-loop task, so it needs
|
||||||
|
// no lock and dies with the loop. See `accumulate_roster`.
|
||||||
|
let mut roster_parts: HashMap<i64, RosterParts> = HashMap::new();
|
||||||
tokio::spawn(async move {
|
tokio::spawn(async move {
|
||||||
while let Some(ev) = event_rx.recv().await {
|
while let Some(ev) = event_rx.recv().await {
|
||||||
// Any line from the shard — including pong heartbeats — is a sign of life.
|
// Any line from the shard — including pong heartbeats — is a sign of life.
|
||||||
@@ -161,6 +166,43 @@ where
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
// Guild roster (Protocol 4): the member list that `guild.update`'s counts cannot
|
||||||
|
// express. It writes a *different column* of the same row, so it never races
|
||||||
|
// guild.update. `guild.leave` deliberately has no arm here — the sidecar
|
||||||
|
// forwards it (persisted and broadcast below, like any event) and the board's
|
||||||
|
// roster self-corrects on the next `guild.roster`, which the shard re-emits
|
||||||
|
// whenever the member set changes. Keeping the delta out of the board is what
|
||||||
|
// keeps the sidecar a forwarder rather than a thing that maintains state.
|
||||||
|
//
|
||||||
|
// A roster over the shard's per-line cap arrives as several frames, so it is
|
||||||
|
// reassembled before it is stored — see `accumulate_roster` for why that happens
|
||||||
|
// here rather than by appending to the column.
|
||||||
|
"guild.roster" => {
|
||||||
|
if let Some(id) = ev.value.get("id").and_then(|v| v.as_i64()) {
|
||||||
|
let seq = ev.value.get("seq").and_then(|v| v.as_i64()).unwrap_or(0);
|
||||||
|
let more = ev
|
||||||
|
.value
|
||||||
|
.get("more")
|
||||||
|
.and_then(|v| v.as_bool())
|
||||||
|
.unwrap_or(false);
|
||||||
|
let members = ev
|
||||||
|
.value
|
||||||
|
.get("members")
|
||||||
|
.and_then(|m| m.as_array())
|
||||||
|
.cloned()
|
||||||
|
.unwrap_or_default();
|
||||||
|
|
||||||
|
if let Some(complete) =
|
||||||
|
accumulate_roster(&mut roster_parts, id, seq, more, members)
|
||||||
|
{
|
||||||
|
let json = serde_json::Value::Array(complete).to_string();
|
||||||
|
if let Err(e) = event_store.upsert_guild_roster(id, &json, t).await
|
||||||
|
{
|
||||||
|
tracing::warn!(error = %e, "failed to upsert guild roster");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
"guild.remove" => {
|
"guild.remove" => {
|
||||||
if let Some(id) = ev.value.get("id").and_then(|v| v.as_i64()) {
|
if let Some(id) = ev.value.get("id").and_then(|v| v.as_i64()) {
|
||||||
if let Err(e) = event_store.delete_guild(id).await {
|
if let Err(e) = event_store.delete_guild(id).await {
|
||||||
@@ -279,6 +321,10 @@ where
|
|||||||
// with announce=false so a restart does not re-proclaim every article at once. news.add
|
// with announce=false so a restart does not re-proclaim every article at once. news.add
|
||||||
// is idempotent by id, so replaying to a still-populated shard is harmless.
|
// is idempotent by id, so replaying to a still-populated shard is harmless.
|
||||||
if ev.kind == "server.hello" {
|
if ev.kind == "server.hello" {
|
||||||
|
// A (re)connected shard restarts every roster from `seq` 0, so any half-received
|
||||||
|
// one belongs to the previous connection and can never be completed.
|
||||||
|
roster_parts.clear();
|
||||||
|
|
||||||
match event_store.news_all().await {
|
match event_store.news_all().await {
|
||||||
Ok(items) => {
|
Ok(items) => {
|
||||||
for mut item in items {
|
for mut item in items {
|
||||||
@@ -327,3 +373,199 @@ pub fn now_ms() -> i64 {
|
|||||||
.map(|d| d.as_millis() as i64)
|
.map(|d| d.as_millis() as i64)
|
||||||
.unwrap_or(0)
|
.unwrap_or(0)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// A guild roster that has arrived in part: the `seq` expected next, and what has accumulated.
|
||||||
|
struct RosterParts {
|
||||||
|
next_seq: i64,
|
||||||
|
members: Vec<Value>,
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Refuses to accumulate a roster past this many members. The shard caps its own frames, so
|
||||||
|
/// exceeding this means a shard that is buggy or not what it claims to be — and the one thing this
|
||||||
|
/// buffer must not do is grow without bound on its say-so.
|
||||||
|
const MAX_ROSTER_MEMBERS: usize = 50_000;
|
||||||
|
|
||||||
|
/// Reassembles a `guild.roster` that the shard split across frames, returning the whole member list
|
||||||
|
/// once the final frame arrives and `None` while one is still incomplete.
|
||||||
|
///
|
||||||
|
/// Reassembly happens **here, in memory, before the store** rather than by appending to the
|
||||||
|
/// `members` column, for two reasons. Appending would make the write a read-modify-write — the exact
|
||||||
|
/// thing the two-column board design exists to avoid — and it would publish a torn roster: a reader
|
||||||
|
/// hitting `GET /guilds` between frames would see a partial member list as though it were the truth.
|
||||||
|
/// Buffering keeps the store's write a single atomic upsert of a complete roster.
|
||||||
|
///
|
||||||
|
/// This is transport-level reassembly, not domain state: it is the same category of work as turning
|
||||||
|
/// bytes into a line, and it holds nothing once a roster is complete. That is what keeps it
|
||||||
|
/// compatible with the sidecar being a forwarder.
|
||||||
|
///
|
||||||
|
/// The ordinary case — a guild inside the shard's per-line cap, which is every realistic one —
|
||||||
|
/// arrives as `seq` 0 with `more` false and is returned immediately without ever touching the map.
|
||||||
|
fn accumulate_roster(
|
||||||
|
parts: &mut HashMap<i64, RosterParts>,
|
||||||
|
id: i64,
|
||||||
|
seq: i64,
|
||||||
|
more: bool,
|
||||||
|
members: Vec<Value>,
|
||||||
|
) -> Option<Vec<Value>> {
|
||||||
|
if seq == 0 {
|
||||||
|
// A fresh roster supersedes any partial one: the shard restarts at 0 every time it emits,
|
||||||
|
// so a leftover buffer is from an emission that was interrupted and will never finish.
|
||||||
|
parts.remove(&id);
|
||||||
|
|
||||||
|
if !more {
|
||||||
|
return Some(members);
|
||||||
|
}
|
||||||
|
|
||||||
|
parts.insert(
|
||||||
|
id,
|
||||||
|
RosterParts {
|
||||||
|
next_seq: 1,
|
||||||
|
members,
|
||||||
|
},
|
||||||
|
);
|
||||||
|
return None;
|
||||||
|
}
|
||||||
|
|
||||||
|
let entry = match parts.get_mut(&id) {
|
||||||
|
Some(entry) => entry,
|
||||||
|
// A continuation with nothing to continue: the sidecar started, or the shard reconnected,
|
||||||
|
// midway through an emission. Dropping it is right — the next full roster is complete.
|
||||||
|
None => {
|
||||||
|
tracing::debug!(
|
||||||
|
guild = id,
|
||||||
|
seq,
|
||||||
|
"roster continuation with no start; ignoring"
|
||||||
|
);
|
||||||
|
return None;
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
|
if entry.next_seq != seq {
|
||||||
|
tracing::warn!(
|
||||||
|
guild = id,
|
||||||
|
expected = entry.next_seq,
|
||||||
|
got = seq,
|
||||||
|
"roster frames out of order; discarding the partial roster"
|
||||||
|
);
|
||||||
|
parts.remove(&id);
|
||||||
|
return None;
|
||||||
|
}
|
||||||
|
|
||||||
|
entry.members.extend(members);
|
||||||
|
|
||||||
|
if entry.members.len() > MAX_ROSTER_MEMBERS {
|
||||||
|
tracing::warn!(
|
||||||
|
guild = id,
|
||||||
|
len = entry.members.len(),
|
||||||
|
"roster exceeded the reassembly cap; discarding"
|
||||||
|
);
|
||||||
|
parts.remove(&id);
|
||||||
|
return None;
|
||||||
|
}
|
||||||
|
|
||||||
|
if more {
|
||||||
|
entry.next_seq = seq + 1;
|
||||||
|
return None;
|
||||||
|
}
|
||||||
|
|
||||||
|
parts.remove(&id).map(|done| done.members)
|
||||||
|
}
|
||||||
|
|
||||||
|
#[cfg(test)]
|
||||||
|
mod tests {
|
||||||
|
use super::*;
|
||||||
|
|
||||||
|
fn members(names: &[&str]) -> Vec<Value> {
|
||||||
|
names
|
||||||
|
.iter()
|
||||||
|
.map(|n| serde_json::json!({"name": n}))
|
||||||
|
.collect()
|
||||||
|
}
|
||||||
|
|
||||||
|
fn names(vs: &[Value]) -> Vec<String> {
|
||||||
|
vs.iter()
|
||||||
|
.map(|v| v["name"].as_str().unwrap_or_default().to_string())
|
||||||
|
.collect()
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn a_single_frame_roster_is_returned_immediately() {
|
||||||
|
// Every realistic guild takes this path, and it must not depend on the buffer at all.
|
||||||
|
let mut parts = HashMap::new();
|
||||||
|
let out = accumulate_roster(&mut parts, 1, 0, false, members(&["Ada", "Bo"]));
|
||||||
|
|
||||||
|
assert_eq!(names(&out.expect("complete")), ["Ada", "Bo"]);
|
||||||
|
assert!(parts.is_empty(), "nothing should be buffered");
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn a_chunked_roster_reassembles_in_order() {
|
||||||
|
// The case the live rig caught: without this, only the final frame survived and a
|
||||||
|
// 155-member guild appeared on the board with 3 members.
|
||||||
|
let mut parts = HashMap::new();
|
||||||
|
|
||||||
|
assert!(accumulate_roster(&mut parts, 1, 0, true, members(&["Ada"])).is_none());
|
||||||
|
assert!(accumulate_roster(&mut parts, 1, 1, true, members(&["Bo"])).is_none());
|
||||||
|
let out = accumulate_roster(&mut parts, 1, 2, false, members(&["Cy"]));
|
||||||
|
|
||||||
|
assert_eq!(names(&out.expect("complete")), ["Ada", "Bo", "Cy"]);
|
||||||
|
assert!(parts.is_empty(), "buffer is released once complete");
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn a_restarted_roster_supersedes_a_partial_one() {
|
||||||
|
// A shard that reconnects mid-emission starts again at seq 0. The abandoned frames must not
|
||||||
|
// end up spliced onto the front of the new roster.
|
||||||
|
let mut parts = HashMap::new();
|
||||||
|
|
||||||
|
assert!(accumulate_roster(&mut parts, 1, 0, true, members(&["Stale"])).is_none());
|
||||||
|
let out = accumulate_roster(&mut parts, 1, 0, false, members(&["Fresh"]));
|
||||||
|
|
||||||
|
assert_eq!(names(&out.expect("complete")), ["Fresh"]);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn an_out_of_order_frame_discards_the_partial_roster() {
|
||||||
|
// Better to publish nothing and wait for the next full emission than to store a roster with
|
||||||
|
// a hole in it that nothing downstream could detect.
|
||||||
|
let mut parts = HashMap::new();
|
||||||
|
|
||||||
|
assert!(accumulate_roster(&mut parts, 1, 0, true, members(&["Ada"])).is_none());
|
||||||
|
assert!(accumulate_roster(&mut parts, 1, 2, false, members(&["Skipped"])).is_none());
|
||||||
|
assert!(parts.is_empty());
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn a_continuation_with_no_start_is_ignored() {
|
||||||
|
// The sidecar restarting midway through a shard's emission.
|
||||||
|
let mut parts = HashMap::new();
|
||||||
|
|
||||||
|
assert!(accumulate_roster(&mut parts, 1, 3, false, members(&["Orphan"])).is_none());
|
||||||
|
assert!(parts.is_empty());
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn two_guilds_reassemble_independently() {
|
||||||
|
// Rosters for different guilds interleave freely — the sweep emits one guild after another
|
||||||
|
// and nothing serialises them on the wire.
|
||||||
|
let mut parts = HashMap::new();
|
||||||
|
|
||||||
|
assert!(accumulate_roster(&mut parts, 1, 0, true, members(&["A1"])).is_none());
|
||||||
|
assert!(accumulate_roster(&mut parts, 2, 0, true, members(&["B1"])).is_none());
|
||||||
|
let g2 = accumulate_roster(&mut parts, 2, 1, false, members(&["B2"]));
|
||||||
|
let g1 = accumulate_roster(&mut parts, 1, 1, false, members(&["A2"]));
|
||||||
|
|
||||||
|
assert_eq!(names(&g2.expect("guild 2")), ["B1", "B2"]);
|
||||||
|
assert_eq!(names(&g1.expect("guild 1")), ["A1", "A2"]);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn an_empty_roster_is_a_complete_roster() {
|
||||||
|
// A guild whose last member left emits one frame with an empty array. Treating that as
|
||||||
|
// "nothing to store" would leave the board showing the roster it had before.
|
||||||
|
let mut parts = HashMap::new();
|
||||||
|
let out = accumulate_roster(&mut parts, 1, 0, false, vec![]);
|
||||||
|
|
||||||
|
assert_eq!(out.expect("complete").len(), 0);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -46,7 +46,85 @@ use tracing_subscriber::EnvFilter;
|
|||||||
/// `vendor.listing.remove`, with the `GET /ruleset`, `/points` and `/market` reads that serve them
|
/// `vendor.listing.remove`, with the `GET /ruleset`, `/points` and `/market` reads that serve them
|
||||||
/// from the store. Same shape as the v2 bump — the kinds are additive, the endpoints are not — and
|
/// from the store. Same shape as the v2 bump — the kinds are additive, the endpoints are not — and
|
||||||
/// there is deliberately no feature-negotiation array: v3 implies all three kinds.
|
/// there is deliberately no feature-negotiation array: v3 implies all three kinds.
|
||||||
pub const PROTOCOL_VERSION: u32 = 3;
|
///
|
||||||
|
/// v4 (Protocol 4): adds `guild.roster` and `guild.leave`, giving the guild board a real member list
|
||||||
|
/// instead of the member *count* that was all v2 could express. Additive in the same way again — the
|
||||||
|
/// kinds are new, `GET /guilds` grows a `roster` key, and nothing existing changed shape. This is the
|
||||||
|
/// first bump that also needed a **store migration** (`guilds.members`), because it is the first to
|
||||||
|
/// add a column to a table that already exists rather than a whole new table; see `store::migrate`.
|
||||||
|
///
|
||||||
|
/// v5 (Protocol 5): three enrichments that are additive in the same way again, bumped together
|
||||||
|
/// rather than one at a time because a protocol bump is not cheap here — it costs a sidecar
|
||||||
|
/// release, a republished bundle and an operator update on every shard, so a field left out costs
|
||||||
|
/// a whole second round of that rather than a follow-up commit. They are:
|
||||||
|
///
|
||||||
|
/// * `house.decay` gains `ownerName` and a decay SCHEDULE — `nextStage`, `decayPeriodSec`,
|
||||||
|
/// `dynamicDecay`, and `estimatedCollapse` only where it is exactly knowable (at IDOC under
|
||||||
|
/// dynamic decay; at any stage under static decay, which has no randomness to wait out).
|
||||||
|
/// * `vendor.listing` gains `ownerAcct` — without which the frame names an owner nobody can
|
||||||
|
/// resolve to a person — and a `fees` object carrying the charge, the funds, the pay interval
|
||||||
|
/// and the resolved `dismissalAt`.
|
||||||
|
/// * `account.login.result` is a NEW kind: the verdict of a login, which the pre-existing
|
||||||
|
/// `account.login.attempt` structurally cannot carry (its EventSink fires before the auth
|
||||||
|
/// decision is made).
|
||||||
|
///
|
||||||
|
/// **No store migration this time**, unlike v4. Every frame is persisted whole and the board tables
|
||||||
|
/// index only the columns they already had, so the new fields ride inside the stored JSON and the
|
||||||
|
/// new kind lands in `events` like any other. That is the dumb-forwarder property doing its job:
|
||||||
|
/// the sidecar defines no schema for a frame's contents and so needs no change when they grow.
|
||||||
|
///
|
||||||
|
/// v6 (Protocol 6): the first bump that is about a GUARANTEE rather than about data, and the first
|
||||||
|
/// the sidecar mostly gets for free. Two things:
|
||||||
|
///
|
||||||
|
/// * **`idempotencyKey` on inbound commands.** A command that carries one is executed by the shard
|
||||||
|
/// at most once; a repeat is answered with the original reply rather than re-run. That is what
|
||||||
|
/// makes a world-writing verb retryable at all — until now a lost acknowledgement was
|
||||||
|
/// indistinguishable from a command that never applied, so the website had to declare every
|
||||||
|
/// write un-retryable and accept losing one rather than risk doubling it. The sidecar's part is
|
||||||
|
/// to CARRY the key (it rides in the command body, which every write endpoint already passes
|
||||||
|
/// through verbatim) and to understand the one new answer the shard can now give: `bridge.busy`,
|
||||||
|
/// meaning a command under that key is still in flight. See `web::respond`.
|
||||||
|
/// * **`champ.boss.killed` is a new kind**: a champion's defeat, with the damage table only the
|
||||||
|
/// shard ever sees. It was previously inferable from `champ.update` going `bossUp` true then
|
||||||
|
/// false alongside a nearby `mob.killed`, which is fragile and says nothing about who did the
|
||||||
|
/// work. It lands in `events` and on the feed like any other kind, with no code here at all —
|
||||||
|
/// the dumb-forwarder property again.
|
||||||
|
///
|
||||||
|
/// **No store migration.** Nothing gains a column; the new kind is persisted whole like every other.
|
||||||
|
///
|
||||||
|
/// # Protocol 8 — the Asset Bridge (docs/link/v8.md)
|
||||||
|
///
|
||||||
|
/// The shard starts sending the operator's own **client assets** over this link: the cliloc string
|
||||||
|
/// table, creature and item art, player models. The point is that an operator stops having to run
|
||||||
|
/// a GUI converter on a desktop to make their site render a bestiary, and the shard is the only
|
||||||
|
/// host that already has the client files — a ServUO server cannot boot without them.
|
||||||
|
///
|
||||||
|
/// Phase 1 is the transport, and the sidecar's share of it is three things:
|
||||||
|
///
|
||||||
|
/// * **A new command family, `assets.*`, forwarded verbatim** like every other. The first of them
|
||||||
|
/// is `assets.sources` — stage 1 of the import gate: what the client files currently are, and
|
||||||
|
/// what version of the shard's extractor would read them. No pixels cross on this call.
|
||||||
|
/// * **An inbound line cap** — [`shard::MAX_INBOUND_LINE_BYTES`]. This is the one change that is
|
||||||
|
/// not additive. `read_line` had no bound at all, which was survivable while the shard had no
|
||||||
|
/// reason to send a large line; protocol 8 gives it one deliberately, and an unbounded read
|
||||||
|
/// facing a component that now sends megabytes is a memory-exhaustion shape we would be
|
||||||
|
/// inventing ourselves.
|
||||||
|
/// * **Nothing else.** Assets ride the request/reply path, so `rpc::try_route` consumes them
|
||||||
|
/// before `app.rs` can persist them to the store and fan them out to every WebSocket
|
||||||
|
/// subscriber — which is what keeps a 512 KiB reply from being written to SQLite and broadcast
|
||||||
|
/// to every connected client. The dumb-forwarder property is doing real work here: the sidecar
|
||||||
|
/// does not know what an asset is, and must not learn.
|
||||||
|
///
|
||||||
|
/// Phase 2 adds the first family that actually carries content: **`cliloc.table`**, served at
|
||||||
|
/// `GET /cliloc`. It pages — the shard cuts at a byte budget and the caller echoes a cursor back —
|
||||||
|
/// and the sidecar forwards it without keeping any of it, which matters more here than usual: the
|
||||||
|
/// payload is five megabytes of EA's strings out of the operator's own client, and the one copy of
|
||||||
|
/// it that should exist is the one the website imports. That phase also gave `assets.error` a
|
||||||
|
/// `code`, so the status a refusal maps to stops depending on the wording of a human-facing
|
||||||
|
/// sentence (see [`web::asset_error_status`]).
|
||||||
|
///
|
||||||
|
/// **No store migration**, again: nothing on this plane is an event, so nothing is persisted.
|
||||||
|
pub const PROTOCOL_VERSION: u32 = 8;
|
||||||
|
|
||||||
// Not `#[tokio::main]`: on Windows the SCM dispatcher takes over this thread and starts the runtime
|
// Not `#[tokio::main]`: on Windows the SCM dispatcher takes over this thread and starts the runtime
|
||||||
// itself, on its own thread, once the service actually begins. The runtime is built by whichever
|
// itself, on its own thread, once the service actually begins. The runtime is built by whichever
|
||||||
|
|||||||
@@ -7,15 +7,130 @@
|
|||||||
//! Framing is newline-delimited JSON, bidirectional: the shard sends events, we send commands. We
|
//! Framing is newline-delimited JSON, bidirectional: the shard sends events, we send commands. We
|
||||||
//! accept one shard connection at a time and re-accept when it drops (the shard reconnects on its
|
//! accept one shard connection at a time and re-accept when it drops (the shard reconnects on its
|
||||||
//! own, with a bounded backoff).
|
//! own, with a bounded backoff).
|
||||||
|
//!
|
||||||
|
//! Inbound lines are **capped** (see [`MAX_INBOUND_LINE_BYTES`]). Until protocol 8 they were not:
|
||||||
|
//! `read_line` will buffer a line of any length, which was survivable only because the shard had
|
||||||
|
//! never had a reason to send a large one. The Asset Bridge gives it one, so the gap had to close
|
||||||
|
//! before it became a memory-exhaustion shape we invented ourselves.
|
||||||
|
|
||||||
use std::sync::Arc;
|
use std::sync::Arc;
|
||||||
|
|
||||||
use serde_json::Value;
|
use serde_json::Value;
|
||||||
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
|
use tokio::io::{AsyncBufRead, AsyncBufReadExt, AsyncWriteExt, BufReader};
|
||||||
use tokio::net::TcpListener;
|
use tokio::net::TcpListener;
|
||||||
use tokio::sync::{mpsc, Mutex};
|
use tokio::sync::{mpsc, Mutex};
|
||||||
use tracing::{info, warn};
|
use tracing::{info, warn};
|
||||||
|
|
||||||
|
/// The longest line the sidecar will accept from the shard, in bytes.
|
||||||
|
///
|
||||||
|
/// Set above the largest legal batch rather than at it: the shard cuts a batch when the next item
|
||||||
|
/// would take it past `Bridge.AssetBatchBytes` (512 KiB), and always admits the first item of a
|
||||||
|
/// page even when that item alone is bigger than the budget — so one page can legitimately
|
||||||
|
/// overshoot by one item. Doubling the budget to get this cap is what makes that overshoot safe
|
||||||
|
/// instead of a dropped reply.
|
||||||
|
///
|
||||||
|
/// Over-long lines are **discarded, not buffered**, and the connection stays up. That is the same
|
||||||
|
/// disposition `BridgeLink.cs` has always had for its own 1 MiB inbound cap in the other
|
||||||
|
/// direction, and it is the right one here: a single malformed frame is not a reason to tear down
|
||||||
|
/// a link that live events are flowing over. The dropped reply simply times out and is
|
||||||
|
/// re-requested, which is safe because everything on the asset plane is idempotent.
|
||||||
|
pub const MAX_INBOUND_LINE_BYTES: usize = 1024 * 1024;
|
||||||
|
|
||||||
|
/// What one read off the shard socket produced.
|
||||||
|
#[derive(Debug)]
|
||||||
|
enum Line {
|
||||||
|
/// A complete line, within the cap.
|
||||||
|
Complete(String),
|
||||||
|
/// A line that ran past the cap. Carries how many bytes were thrown away, for the log.
|
||||||
|
TooLong(usize),
|
||||||
|
/// The shard closed the connection.
|
||||||
|
Eof,
|
||||||
|
}
|
||||||
|
|
||||||
|
/// A cancel-safe, capped, newline-delimited reader.
|
||||||
|
///
|
||||||
|
/// Every piece of state that must survive a partial read lives here rather than in a local,
|
||||||
|
/// because this is polled inside a `tokio::select!`: the loop below drops the future whenever a
|
||||||
|
/// command wins the race, and a `discarding` flag or a half-filled buffer held in a local would be
|
||||||
|
/// lost with it. Losing the buffer corrupts the *next* line; losing `discarding` turns the tail of
|
||||||
|
/// an over-long line into a line of its own. Both are silent.
|
||||||
|
///
|
||||||
|
/// The only await point is `fill_buf`, and nothing is consumed until after it returns, so a
|
||||||
|
/// cancellation between the two can lose at most the wakeup.
|
||||||
|
#[derive(Default)]
|
||||||
|
struct LineReader {
|
||||||
|
buf: Vec<u8>,
|
||||||
|
discarding: bool,
|
||||||
|
discarded: usize,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl LineReader {
|
||||||
|
async fn next<R: AsyncBufRead + Unpin>(&mut self, reader: &mut R) -> std::io::Result<Line> {
|
||||||
|
loop {
|
||||||
|
let consumed;
|
||||||
|
let outcome;
|
||||||
|
|
||||||
|
{
|
||||||
|
let available = reader.fill_buf().await?;
|
||||||
|
|
||||||
|
if available.is_empty() {
|
||||||
|
return Ok(Line::Eof);
|
||||||
|
}
|
||||||
|
|
||||||
|
match available.iter().position(|&b| b == b'\n') {
|
||||||
|
Some(at) => {
|
||||||
|
consumed = at + 1;
|
||||||
|
|
||||||
|
if self.discarding {
|
||||||
|
// The tail of a line we already gave up on. Swallow it, terminator
|
||||||
|
// included, and report the size once.
|
||||||
|
self.discarded += at;
|
||||||
|
let total = self.discarded;
|
||||||
|
self.discarding = false;
|
||||||
|
self.discarded = 0;
|
||||||
|
outcome = Some(Line::TooLong(total));
|
||||||
|
} else if self.buf.len() + at > MAX_INBOUND_LINE_BYTES {
|
||||||
|
// The cap is reached only now, on the chunk that also holds the
|
||||||
|
// terminator — so there is nothing left to discard.
|
||||||
|
let total = self.buf.len() + at;
|
||||||
|
self.buf.clear();
|
||||||
|
outcome = Some(Line::TooLong(total));
|
||||||
|
} else {
|
||||||
|
self.buf.extend_from_slice(&available[..at]);
|
||||||
|
let line = String::from_utf8_lossy(&self.buf).into_owned();
|
||||||
|
self.buf.clear();
|
||||||
|
outcome = Some(Line::Complete(line));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
None => {
|
||||||
|
consumed = available.len();
|
||||||
|
|
||||||
|
if self.discarding {
|
||||||
|
self.discarded += consumed;
|
||||||
|
} else if self.buf.len() + consumed > MAX_INBOUND_LINE_BYTES {
|
||||||
|
// Refuse rather than buffer: this is the whole point of the cap.
|
||||||
|
// Everything up to the next newline is now dropped on the floor.
|
||||||
|
self.discarded = self.buf.len() + consumed;
|
||||||
|
self.buf.clear();
|
||||||
|
self.discarding = true;
|
||||||
|
} else {
|
||||||
|
self.buf.extend_from_slice(available);
|
||||||
|
}
|
||||||
|
|
||||||
|
outcome = None;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
reader.consume(consumed);
|
||||||
|
|
||||||
|
if let Some(line) = outcome {
|
||||||
|
return Ok(line);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/// An event line received from the shard, parsed. `kind` is lifted out for routing.
|
/// An event line received from the shard, parsed. `kind` is lifted out for routing.
|
||||||
#[derive(Debug, Clone)]
|
#[derive(Debug, Clone)]
|
||||||
pub struct ShardEvent {
|
pub struct ShardEvent {
|
||||||
@@ -112,16 +227,25 @@ async fn handle_connection(
|
|||||||
handle.set(Some(cmd_tx)).await;
|
handle.set(Some(cmd_tx)).await;
|
||||||
|
|
||||||
let mut reader = BufReader::new(read_half);
|
let mut reader = BufReader::new(read_half);
|
||||||
let mut line = String::new();
|
let mut lines = LineReader::default();
|
||||||
|
|
||||||
loop {
|
loop {
|
||||||
tokio::select! {
|
tokio::select! {
|
||||||
// Inbound: a line from the shard.
|
// Inbound: a line from the shard.
|
||||||
result = reader.read_line(&mut line) => {
|
result = lines.next(&mut reader) => {
|
||||||
let n = result?;
|
match result? {
|
||||||
if n == 0 {
|
Line::Eof => return Ok(()), // clean EOF: shard closed
|
||||||
return Ok(()); // clean EOF: shard closed
|
Line::TooLong(bytes) => {
|
||||||
|
// Deliberately not a disconnect. See MAX_INBOUND_LINE_BYTES: a reply lost
|
||||||
|
// this way times out on the caller's side and is re-requested, and tearing
|
||||||
|
// the link down would take the live event feed with it.
|
||||||
|
warn!(
|
||||||
|
bytes,
|
||||||
|
cap = MAX_INBOUND_LINE_BYTES,
|
||||||
|
"inbound line over the cap; discarded"
|
||||||
|
);
|
||||||
}
|
}
|
||||||
|
Line::Complete(line) => {
|
||||||
let trimmed = line.trim_end();
|
let trimmed = line.trim_end();
|
||||||
if !trimmed.is_empty() {
|
if !trimmed.is_empty() {
|
||||||
match serde_json::from_str::<Value>(trimmed) {
|
match serde_json::from_str::<Value>(trimmed) {
|
||||||
@@ -136,7 +260,8 @@ async fn handle_connection(
|
|||||||
Err(e) => warn!(error = %e, line = %trimmed, "unparseable event"),
|
Err(e) => warn!(error = %e, line = %trimmed, "unparseable event"),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
line.clear();
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
// Outbound: a command to write to the shard.
|
// Outbound: a command to write to the shard.
|
||||||
cmd = cmd_rx.recv() => {
|
cmd = cmd_rx.recv() => {
|
||||||
@@ -152,3 +277,108 @@ async fn handle_connection(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[cfg(test)]
|
||||||
|
mod tests {
|
||||||
|
use super::*;
|
||||||
|
|
||||||
|
/// Drives `LineReader` over a byte slice, returning every outcome up to EOF.
|
||||||
|
async fn read_all(input: &[u8]) -> Vec<Line> {
|
||||||
|
let mut reader = BufReader::with_capacity(64, input);
|
||||||
|
let mut lines = LineReader::default();
|
||||||
|
let mut out = Vec::new();
|
||||||
|
|
||||||
|
loop {
|
||||||
|
match lines.next(&mut reader).await.unwrap() {
|
||||||
|
Line::Eof => break,
|
||||||
|
other => out.push(other),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
out
|
||||||
|
}
|
||||||
|
|
||||||
|
fn complete(lines: &[Line]) -> Vec<&str> {
|
||||||
|
lines
|
||||||
|
.iter()
|
||||||
|
.filter_map(|l| match l {
|
||||||
|
Line::Complete(s) => Some(s.as_str()),
|
||||||
|
_ => None,
|
||||||
|
})
|
||||||
|
.collect()
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn splits_on_newlines() {
|
||||||
|
let lines = read_all(b"{\"a\":1}\n{\"b\":2}\n").await;
|
||||||
|
assert_eq!(complete(&lines), vec!["{\"a\":1}", "{\"b\":2}"]);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// The reader's buffer is 64 bytes here, so every one of these lines spans several
|
||||||
|
/// `fill_buf` chunks. Reassembly across chunks is the thing `read_line` did for us.
|
||||||
|
#[tokio::test]
|
||||||
|
async fn reassembles_across_chunks() {
|
||||||
|
let long = "x".repeat(500);
|
||||||
|
let input = format!("{}\n{}\n", long, long);
|
||||||
|
let lines = read_all(input.as_bytes()).await;
|
||||||
|
|
||||||
|
assert_eq!(complete(&lines), vec![long.as_str(), long.as_str()]);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// The cap itself. The over-long line must be reported and thrown away, and — the part that
|
||||||
|
/// actually matters — the line *after* it must still arrive intact. A reader that lost its
|
||||||
|
/// `discarding` flag would emit the tail of the oversized line as a line of its own.
|
||||||
|
#[tokio::test]
|
||||||
|
async fn refuses_an_over_long_line_and_recovers() {
|
||||||
|
let mut input = Vec::new();
|
||||||
|
input.extend_from_slice(&b"a".repeat(MAX_INBOUND_LINE_BYTES + 10));
|
||||||
|
input.push(b'\n');
|
||||||
|
input.extend_from_slice(b"{\"kind\":\"pong\"}\n");
|
||||||
|
|
||||||
|
let lines = read_all(&input).await;
|
||||||
|
|
||||||
|
assert_eq!(lines.len(), 2);
|
||||||
|
assert!(
|
||||||
|
matches!(lines[0], Line::TooLong(n) if n >= MAX_INBOUND_LINE_BYTES),
|
||||||
|
"expected TooLong, got {:?}",
|
||||||
|
lines[0]
|
||||||
|
);
|
||||||
|
assert_eq!(complete(&lines), vec!["{\"kind\":\"pong\"}"]);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// A line of exactly the cap is legal; one byte more is not. Checking both sides is what says
|
||||||
|
/// the comparison is `>` rather than `>=`, which would silently cost a byte of the budget.
|
||||||
|
#[tokio::test]
|
||||||
|
async fn the_cap_is_inclusive() {
|
||||||
|
let at_cap = "b".repeat(MAX_INBOUND_LINE_BYTES);
|
||||||
|
let lines = read_all(format!("{}\n", at_cap).as_bytes()).await;
|
||||||
|
assert_eq!(complete(&lines).len(), 1);
|
||||||
|
|
||||||
|
let over = "b".repeat(MAX_INBOUND_LINE_BYTES + 1);
|
||||||
|
let lines = read_all(format!("{}\n", over).as_bytes()).await;
|
||||||
|
assert!(complete(&lines).is_empty());
|
||||||
|
assert!(matches!(lines[0], Line::TooLong(_)));
|
||||||
|
}
|
||||||
|
|
||||||
|
/// An over-long line whose terminator lands in the very chunk that crosses the cap: the
|
||||||
|
/// reader must not leave itself in `discarding` and eat the next line as well.
|
||||||
|
#[tokio::test]
|
||||||
|
async fn over_long_line_terminating_in_the_crossing_chunk() {
|
||||||
|
let mut input = Vec::new();
|
||||||
|
input.extend_from_slice(&b"c".repeat(MAX_INBOUND_LINE_BYTES + 1));
|
||||||
|
input.extend_from_slice(b"\n{\"kind\":\"pong\"}\n");
|
||||||
|
|
||||||
|
let lines = read_all(&input).await;
|
||||||
|
|
||||||
|
assert!(matches!(lines[0], Line::TooLong(_)));
|
||||||
|
assert_eq!(complete(&lines), vec!["{\"kind\":\"pong\"}"]);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// A partial line at EOF is dropped rather than delivered half-parsed. The shard reconnects
|
||||||
|
/// and re-sends; half a JSON object is not something to hand to the event fan-out.
|
||||||
|
#[tokio::test]
|
||||||
|
async fn trailing_partial_line_at_eof_is_dropped() {
|
||||||
|
let lines = read_all(b"{\"a\":1}\n{\"b\":").await;
|
||||||
|
assert_eq!(complete(&lines), vec!["{\"a\":1}"]);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -45,6 +45,7 @@ impl Store {
|
|||||||
.await?;
|
.await?;
|
||||||
|
|
||||||
sqlx::query(SCHEMA).execute(&pool).await?;
|
sqlx::query(SCHEMA).execute(&pool).await?;
|
||||||
|
migrate(&pool).await?;
|
||||||
info!(%path, "store ready");
|
info!(%path, "store ready");
|
||||||
Ok(Self { pool })
|
Ok(Self { pool })
|
||||||
}
|
}
|
||||||
@@ -236,12 +237,61 @@ impl Store {
|
|||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Upserts one guild's member roster (Protocol 4), keyed by guild id, touching **only** the
|
||||||
|
/// `members` column.
|
||||||
|
///
|
||||||
|
/// Deliberately not a write to `json`. That column holds the verbatim `guild.update` line, and a
|
||||||
|
/// roster arriving as its own event must not clobber the snapshot — name, abbreviation, leader,
|
||||||
|
/// online count — that `guild.update` owns. Splitting the two writers across two columns of one
|
||||||
|
/// row is what lets both be plain upserts: neither needs to read the other's value first, so
|
||||||
|
/// there is no read-modify-write and no ordering requirement between the two kinds.
|
||||||
|
///
|
||||||
|
/// The `INSERT` half is not redundant: a roster can arrive before the first `guild.update` for a
|
||||||
|
/// guild, and the row it creates then carries `'{}'` until that update fills it in.
|
||||||
|
pub async fn upsert_guild_roster(
|
||||||
|
&self,
|
||||||
|
id: i64,
|
||||||
|
members_json: &str,
|
||||||
|
t: i64,
|
||||||
|
) -> anyhow::Result<()> {
|
||||||
|
sqlx::query(
|
||||||
|
"INSERT INTO guilds (id, name, json, updated_t, members) VALUES (?, NULL, '{}', ?, ?)
|
||||||
|
ON CONFLICT(id) DO UPDATE SET members = excluded.members, updated_t = excluded.updated_t",
|
||||||
|
)
|
||||||
|
.bind(id)
|
||||||
|
.bind(t)
|
||||||
|
.bind(members_json)
|
||||||
|
.execute(&self.pool)
|
||||||
|
.await?;
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
/// The full guild board: every guild's latest snapshot, ordered by name.
|
/// The full guild board: every guild's latest snapshot, ordered by name.
|
||||||
|
///
|
||||||
|
/// The roster is stored in its own column (see [`Self::upsert_guild_roster`]) and folded into
|
||||||
|
/// the projected object as `roster` here, at read time. A guild that has had a `guild.update`
|
||||||
|
/// but no `guild.roster` yet simply has no `roster` key, which is the honest representation of
|
||||||
|
/// "not known" and distinct from a guild whose roster is genuinely empty.
|
||||||
pub async fn guilds_all(&self) -> anyhow::Result<Vec<Value>> {
|
pub async fn guilds_all(&self) -> anyhow::Result<Vec<Value>> {
|
||||||
let rows = sqlx::query("SELECT json FROM guilds ORDER BY name, id")
|
let rows = sqlx::query("SELECT json, members FROM guilds ORDER BY name, id")
|
||||||
.fetch_all(&self.pool)
|
.fetch_all(&self.pool)
|
||||||
.await?;
|
.await?;
|
||||||
Ok(parse_json_column(rows))
|
|
||||||
|
Ok(rows
|
||||||
|
.into_iter()
|
||||||
|
.filter_map(|r| {
|
||||||
|
let mut v: Value = serde_json::from_str(&r.get::<String, _>("json")).ok()?;
|
||||||
|
let members: Option<String> = r.get("members");
|
||||||
|
|
||||||
|
if let (Some(obj), Some(raw)) = (v.as_object_mut(), members) {
|
||||||
|
if let Ok(list) = serde_json::from_str::<Value>(&raw) {
|
||||||
|
obj.insert("roster".into(), list);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
Some(v)
|
||||||
|
})
|
||||||
|
.collect())
|
||||||
}
|
}
|
||||||
|
|
||||||
// ---- governor board (Protocol 2.0) ----
|
// ---- governor board (Protocol 2.0) ----
|
||||||
@@ -513,6 +563,75 @@ impl Store {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// The schema version this build expects. Bump it, and add the matching arm to [`migrate`], for
|
||||||
|
/// every change that `SCHEMA` alone cannot make to a database that already exists.
|
||||||
|
const SCHEMA_VERSION: i64 = 1;
|
||||||
|
|
||||||
|
/// Brings an existing database forward to [`SCHEMA_VERSION`].
|
||||||
|
///
|
||||||
|
/// `SCHEMA` is `CREATE TABLE IF NOT EXISTS` only, which is enough to *add a table* but cannot add a
|
||||||
|
/// column to a table that is already there. Every schema change up to and including Protocol 3.0
|
||||||
|
/// happened to add whole tables, so this never mattered and `ALTER TABLE` appears nowhere in this
|
||||||
|
/// repo's history. `guilds.members` (Protocol 4) is the first column added to an existing table, so
|
||||||
|
/// the mechanism has to exist now.
|
||||||
|
///
|
||||||
|
/// The version counter is SQLite's own `PRAGMA user_version`: an integer in the database header, so
|
||||||
|
/// it needs no table of its own and cannot be separated from the file it describes. Each step runs
|
||||||
|
/// in a transaction **together with** the bump that records it, so a step either lands completely or
|
||||||
|
/// not at all, and an interrupted run resumes at the right place rather than re-applying half of one.
|
||||||
|
///
|
||||||
|
/// A failure here propagates and aborts startup, deliberately. A half-migrated store answers the
|
||||||
|
/// website with confusing partial data, which is worse than being plainly absent — and the shard
|
||||||
|
/// dials *out* to the sidecar, so a sidecar that refuses to start never stalls the game.
|
||||||
|
async fn migrate(pool: &SqlitePool) -> anyhow::Result<()> {
|
||||||
|
let mut version: i64 = sqlx::query_scalar("PRAGMA user_version")
|
||||||
|
.fetch_one(pool)
|
||||||
|
.await?;
|
||||||
|
|
||||||
|
// A database written by a *newer* sidecar than this binary. This is not an error: every step
|
||||||
|
// here is additive, so a newer schema has only columns and tables an older reader ignores, and
|
||||||
|
// refusing to start would turn "roll the binary back" — a recovery path — into a dead end.
|
||||||
|
if version > SCHEMA_VERSION {
|
||||||
|
tracing::warn!(
|
||||||
|
found = version,
|
||||||
|
expected = SCHEMA_VERSION,
|
||||||
|
"store was written by a newer sidecar; continuing, as migrations are additive"
|
||||||
|
);
|
||||||
|
return Ok(());
|
||||||
|
}
|
||||||
|
|
||||||
|
while version < SCHEMA_VERSION {
|
||||||
|
let next = version + 1;
|
||||||
|
let mut tx = pool.begin().await?;
|
||||||
|
|
||||||
|
match next {
|
||||||
|
// Protocol 4: the guild board carries a member roster. Its own column rather than a
|
||||||
|
// field folded into `json`, because `json` holds the verbatim `guild.update` line and
|
||||||
|
// the two writers must not overwrite each other — see `upsert_guild_roster`.
|
||||||
|
1 => {
|
||||||
|
sqlx::query("ALTER TABLE guilds ADD COLUMN members TEXT")
|
||||||
|
.execute(&mut *tx)
|
||||||
|
.await?;
|
||||||
|
}
|
||||||
|
// Unreachable while SCHEMA_VERSION and this match are edited together, which is the
|
||||||
|
// point of failing loudly rather than silently leaving the counter short.
|
||||||
|
n => anyhow::bail!("no migration step defined for schema version {n}"),
|
||||||
|
}
|
||||||
|
|
||||||
|
// `PRAGMA` takes no bind parameters, so this is formatted — safe because `next` is an i64
|
||||||
|
// this loop produced, never anything from outside the process.
|
||||||
|
sqlx::query(&format!("PRAGMA user_version = {next}"))
|
||||||
|
.execute(&mut *tx)
|
||||||
|
.await?;
|
||||||
|
|
||||||
|
tx.commit().await?;
|
||||||
|
info!(version = next, "schema migration applied");
|
||||||
|
version = next;
|
||||||
|
}
|
||||||
|
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
fn parse_json_column(rows: Vec<sqlx::sqlite::SqliteRow>) -> Vec<Value> {
|
fn parse_json_column(rows: Vec<sqlx::sqlite::SqliteRow>) -> Vec<Value> {
|
||||||
rows.into_iter()
|
rows.into_iter()
|
||||||
.filter_map(|r| serde_json::from_str(&r.get::<String, _>("json")).ok())
|
.filter_map(|r| serde_json::from_str(&r.get::<String, _>("json")).ok())
|
||||||
@@ -611,3 +730,195 @@ CREATE TABLE IF NOT EXISTS ruleset (
|
|||||||
updated_t INTEGER NOT NULL
|
updated_t INTEGER NOT NULL
|
||||||
);
|
);
|
||||||
"#;
|
"#;
|
||||||
|
|
||||||
|
#[cfg(test)]
|
||||||
|
mod tests {
|
||||||
|
use super::*;
|
||||||
|
|
||||||
|
/// A unique scratch database path. Matches `config`'s idiom — `std::env::temp_dir()` plus the
|
||||||
|
/// test name — so the cases stay independent under the parallel test runner.
|
||||||
|
fn scratch(name: &str) -> String {
|
||||||
|
let dir = std::env::temp_dir().join(format!("uo-link-store-test-{name}"));
|
||||||
|
let _ = std::fs::remove_dir_all(&dir);
|
||||||
|
std::fs::create_dir_all(&dir).expect("create scratch dir");
|
||||||
|
dir.join("uo-link.db").to_string_lossy().into_owned()
|
||||||
|
}
|
||||||
|
|
||||||
|
/// The `guilds` table exactly as a pre-Protocol-4 sidecar left it: no `members` column, and
|
||||||
|
/// `user_version` still 0. This is the shape a real operator's database is in before an update,
|
||||||
|
/// and the only starting point where the migration does anything.
|
||||||
|
async fn legacy_db(path: &str) -> SqlitePool {
|
||||||
|
let pool = SqlitePoolOptions::new()
|
||||||
|
.max_connections(1)
|
||||||
|
.connect_with(
|
||||||
|
SqliteConnectOptions::new()
|
||||||
|
.filename(path)
|
||||||
|
.create_if_missing(true),
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
.expect("open legacy db");
|
||||||
|
|
||||||
|
sqlx::query(
|
||||||
|
"CREATE TABLE guilds (
|
||||||
|
id INTEGER PRIMARY KEY,
|
||||||
|
name TEXT,
|
||||||
|
json TEXT NOT NULL,
|
||||||
|
updated_t INTEGER NOT NULL
|
||||||
|
)",
|
||||||
|
)
|
||||||
|
.execute(&pool)
|
||||||
|
.await
|
||||||
|
.expect("create legacy guilds table");
|
||||||
|
|
||||||
|
pool.close().await;
|
||||||
|
pool
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn user_version(store: &Store) -> i64 {
|
||||||
|
sqlx::query_scalar("PRAGMA user_version")
|
||||||
|
.fetch_one(&store.pool)
|
||||||
|
.await
|
||||||
|
.expect("read user_version")
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn guild_columns(store: &Store) -> Vec<String> {
|
||||||
|
sqlx::query("PRAGMA table_info(guilds)")
|
||||||
|
.fetch_all(&store.pool)
|
||||||
|
.await
|
||||||
|
.expect("table_info")
|
||||||
|
.into_iter()
|
||||||
|
.map(|r| r.get::<String, _>("name"))
|
||||||
|
.collect()
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn an_existing_pre_protocol_4_database_gains_the_members_column() {
|
||||||
|
// The case that matters: `SCHEMA`'s CREATE TABLE IF NOT EXISTS is a no-op against a table
|
||||||
|
// that is already there, so without `migrate` this database would never get the column and
|
||||||
|
// every roster write would fail against a live install.
|
||||||
|
let path = scratch("legacy-upgrade");
|
||||||
|
legacy_db(&path).await;
|
||||||
|
|
||||||
|
let store = Store::open(&path).await.expect("open migrates");
|
||||||
|
|
||||||
|
assert!(
|
||||||
|
guild_columns(&store).await.contains(&"members".to_string()),
|
||||||
|
"the migration must add guilds.members to a database that already had the table"
|
||||||
|
);
|
||||||
|
assert_eq!(user_version(&store).await, SCHEMA_VERSION);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn a_fresh_database_lands_at_the_current_version() {
|
||||||
|
let path = scratch("fresh");
|
||||||
|
let store = Store::open(&path).await.expect("open");
|
||||||
|
|
||||||
|
assert!(guild_columns(&store).await.contains(&"members".to_string()));
|
||||||
|
assert_eq!(user_version(&store).await, SCHEMA_VERSION);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn reopening_an_already_migrated_database_is_a_no_op() {
|
||||||
|
// Every sidecar restart re-runs this path, so a second run must not attempt the ALTER again
|
||||||
|
// — which would fail with "duplicate column name" and, since a migration failure aborts
|
||||||
|
// startup, would leave the sidecar unable to start at all after its first upgrade.
|
||||||
|
let path = scratch("idempotent");
|
||||||
|
legacy_db(&path).await;
|
||||||
|
|
||||||
|
Store::open(&path).await.expect("first open");
|
||||||
|
let store = Store::open(&path).await.expect("second open must succeed");
|
||||||
|
|
||||||
|
assert_eq!(user_version(&store).await, SCHEMA_VERSION);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn a_roster_does_not_clobber_the_guild_update_snapshot() {
|
||||||
|
// The invariant the two-column split exists to give. `json` holds the verbatim guild.update
|
||||||
|
// line; if a roster write touched it, name/abbr/online would vanish from the board.
|
||||||
|
let path = scratch("no-clobber");
|
||||||
|
let store = Store::open(&path).await.expect("open");
|
||||||
|
|
||||||
|
store
|
||||||
|
.upsert_guild(
|
||||||
|
7,
|
||||||
|
Some("The Cartographers"),
|
||||||
|
r#"{"kind":"guild.update","id":7,"name":"The Cartographers","abbr":"MAP","members":2,"online":1}"#,
|
||||||
|
100,
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
.expect("upsert guild");
|
||||||
|
|
||||||
|
store
|
||||||
|
.upsert_guild_roster(
|
||||||
|
7,
|
||||||
|
r#"[{"serial":"0x1","name":"Ada"},{"serial":"0x2","name":"Bo"}]"#,
|
||||||
|
200,
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
.expect("upsert roster");
|
||||||
|
|
||||||
|
let guilds = store.guilds_all().await.expect("read board");
|
||||||
|
assert_eq!(guilds.len(), 1);
|
||||||
|
let g = &guilds[0];
|
||||||
|
|
||||||
|
assert_eq!(
|
||||||
|
g["name"], "The Cartographers",
|
||||||
|
"guild.update's name survived"
|
||||||
|
);
|
||||||
|
assert_eq!(g["abbr"], "MAP", "guild.update's abbr survived");
|
||||||
|
assert_eq!(g["online"], 1, "guild.update's online count survived");
|
||||||
|
assert_eq!(g["roster"].as_array().expect("roster is an array").len(), 2);
|
||||||
|
assert_eq!(g["roster"][0]["name"], "Ada");
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn the_two_writers_are_order_independent() {
|
||||||
|
// A roster can arrive before the first guild.update for a guild — on a reconnect the shard
|
||||||
|
// re-emits both and nothing orders them. Neither write may depend on the other's row.
|
||||||
|
let path = scratch("either-order");
|
||||||
|
let store = Store::open(&path).await.expect("open");
|
||||||
|
|
||||||
|
store
|
||||||
|
.upsert_guild_roster(9, r#"[{"serial":"0x3","name":"Cy"}]"#, 100)
|
||||||
|
.await
|
||||||
|
.expect("roster first");
|
||||||
|
store
|
||||||
|
.upsert_guild(
|
||||||
|
9,
|
||||||
|
Some("Late Arrivals"),
|
||||||
|
r#"{"kind":"guild.update","id":9,"name":"Late Arrivals","abbr":"LTE"}"#,
|
||||||
|
200,
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
.expect("update second");
|
||||||
|
|
||||||
|
let guilds = store.guilds_all().await.expect("read board");
|
||||||
|
assert_eq!(guilds.len(), 1, "one row, not two");
|
||||||
|
assert_eq!(guilds[0]["name"], "Late Arrivals");
|
||||||
|
assert_eq!(guilds[0]["roster"].as_array().expect("roster").len(), 1);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn a_guild_with_no_roster_yet_has_no_roster_key() {
|
||||||
|
// "Not known" and "known to be empty" are different, and the board must not conflate them:
|
||||||
|
// a website reading `roster: []` would render an empty roster as fact.
|
||||||
|
let path = scratch("absent-roster");
|
||||||
|
let store = Store::open(&path).await.expect("open");
|
||||||
|
|
||||||
|
store
|
||||||
|
.upsert_guild(
|
||||||
|
11,
|
||||||
|
Some("Unswept"),
|
||||||
|
r#"{"kind":"guild.update","id":11,"name":"Unswept"}"#,
|
||||||
|
100,
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
.expect("upsert guild");
|
||||||
|
|
||||||
|
let guilds = store.guilds_all().await.expect("read board");
|
||||||
|
assert!(
|
||||||
|
guilds[0].get("roster").is_none(),
|
||||||
|
"a guild with no roster event must not grow a roster key"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
1048
sidecar/src/web.rs
1048
sidecar/src/web.rs
File diff suppressed because it is too large
Load Diff
Reference in New Issue
Block a user