Compare commits
24 Commits
v1.1.1
...
6c8a247761
| Author | SHA1 | Date | |
|---|---|---|---|
| 6c8a247761 | |||
| f39dfa4f84 | |||
| d83bb1748c | |||
| 5cdd80e694 | |||
| 5d909ca0a3 | |||
| d38a9e8a75 | |||
| 93411966d7 | |||
| d13ad11eb0 | |||
| 5612fba744 | |||
| 8b9dd0d9e8 | |||
| f41237392d | |||
| d0c2e7d6e1 | |||
| 4b8ea768b6 | |||
| 6fb063818a | |||
| 7499e099f4 | |||
| 2408d31ff7 | |||
| b00f2719a4 | |||
| 9216006208 | |||
| be0efd348c | |||
| 3dbc2f490c | |||
| 7b6584006e | |||
| 67d7800300 | |||
| 96af2afa68 | |||
| 36141a23df |
@@ -17,10 +17,12 @@
|
||||
# so let this workflow run on one PR first. The `PR Checks / *` glob matches
|
||||
# without needing the dropdown.
|
||||
#
|
||||
# Scope note: this gates PRs into `main` only. Feature work that lands on an
|
||||
# integration branch first (e.g. `edge`) is still caught on the branch's PR into
|
||||
# `main`. To gate that earlier hop too, add the branch to the `branches:` list
|
||||
# below — nothing else needs to change.
|
||||
# Scope note: `edge` is gated as well as `main`. Multi-phase work lands there
|
||||
# first, so gating only the `main` hop would run these checks for the first time
|
||||
# at the cutover — the one moment a red build is most expensive to discover. This
|
||||
# is the same call `RunicGateway/installer` made for the same reason, and it was
|
||||
# taken here after a nine-PR Android workstream landed on an ungated `edge` with
|
||||
# no CI at all. Adding a branch to the `branches:` list is the whole change.
|
||||
#
|
||||
# Runner: the same self-hosted `ubuntu-latest` runner release.yml uses. Rust is
|
||||
# not assumed to be preinstalled, so the toolchain step bootstraps it the same
|
||||
@@ -31,7 +33,7 @@ name: PR Checks
|
||||
|
||||
on:
|
||||
pull_request:
|
||||
branches: [main]
|
||||
branches: [main, edge]
|
||||
|
||||
# A newer push to the same PR cancels the in-flight run.
|
||||
concurrency:
|
||||
|
||||
@@ -149,6 +149,39 @@ jobs:
|
||||
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
|
||||
# summarize what the tag itself contains rather than emitting an empty
|
||||
# list: the range that produced it, i.e. previous-tag..this-tag.
|
||||
@@ -279,33 +312,43 @@ jobs:
|
||||
ls -l dist && echo "----" && cat dist/SHA256SUMS
|
||||
|
||||
# ── 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' }}
|
||||
env:
|
||||
REGISTRY_USER: ${{ secrets.REGISTRY_USER }}
|
||||
REGISTRY_TOKEN: ${{ secrets.REGISTRY_TOKEN }}
|
||||
run: |
|
||||
set -euo pipefail
|
||||
VERSION="${{ steps.plan.outputs.version }}"
|
||||
TAG="${{ steps.plan.outputs.tag }}"
|
||||
# Secrets can arrive with a trailing newline (depending on how they were
|
||||
# pasted); a stray CR/LF corrupts the remote URL ("credential url cannot
|
||||
# be parsed"). Strip line breaks before building the URL. Passing them via
|
||||
# env (not inline ${{ }}) also keeps a newline from breaking this script.
|
||||
# be parsed"). Strip line breaks before building the URL. They are passed
|
||||
# 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_TOKEN="$(printf '%s' "${REGISTRY_TOKEN}" | tr -d '\r\n')"
|
||||
git config user.name "uo-link-ci"
|
||||
git config user.email "ci@whitlocktech.com"
|
||||
git remote set-url origin \
|
||||
"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
|
||||
# (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
|
||||
@@ -332,18 +375,75 @@ jobs:
|
||||
# corrupt the Authorization header.
|
||||
CI_TOKEN="$(printf '%s' "${REGISTRY_TOKEN}" | tr -d '\r\n')"
|
||||
|
||||
REL_ID="$(curl -sSf -X POST "${API}/releases" \
|
||||
-H "Authorization: token ${CI_TOKEN}" \
|
||||
-H "Content-Type: application/json" \
|
||||
-d "$(jq -n --arg tag "$TAG" --arg body "$BODY" \
|
||||
'{tag_name:$tag, name:$tag, body:$body, draft:false, prerelease:false}')" \
|
||||
| jq -r '.id')"
|
||||
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 "Content-Type: application/json" \
|
||||
-d "${PAYLOAD}" || echo 000)"
|
||||
|
||||
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})"
|
||||
|
||||
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}" \
|
||||
-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}"
|
||||
done
|
||||
|
||||
|
||||
53
README.md
53
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
|
||||
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
|
||||
|
||||
| Repo | What |
|
||||
|------|------|
|
||||
| **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/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/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
|
||||
cd sidecar
|
||||
@@ -39,7 +63,7 @@ cp sidecar.toml.example sidecar.toml # then edit
|
||||
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
|
||||
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.
|
||||
@@ -48,6 +72,29 @@ to read the token back; it is not meant to be scraped from the log.
|
||||
uo-link-sidecar --print-config --config /etc/runicgateway/sidecar.toml
|
||||
```
|
||||
|
||||
### Running as a service
|
||||
|
||||
The same binary runs in the foreground and as a system service — there is no `--service` flag to
|
||||
remember, because the process can tell how it was started.
|
||||
|
||||
- **Linux/systemd** supervises any foreground process, so the unit just runs the binary. `SIGTERM`
|
||||
(what `systemctl stop` sends) and `SIGINT` both unwind it cleanly; logs go to the journal.
|
||||
- **Windows** cannot. The service control manager only supervises a process that connects back to
|
||||
it within ~30 seconds via `StartServiceCtrlDispatcher`; a plain console program registered with
|
||||
`sc.exe create` is killed with **error 1053** despite running perfectly. So on Windows the sidecar
|
||||
speaks that handshake: started by the SCM it runs as a service, started from a shell the connect
|
||||
fails with `ERROR_FAILED_SERVICE_CONTROLLER_CONNECT` and it falls through to an ordinary
|
||||
foreground run. It reports `Running` only once the shard port is bound and the store is open, and
|
||||
— having no console — logs to `uo-link-sidecar.<date>.log` beside its config, rolled daily.
|
||||
|
||||
Only the starting and stopping is platform-specific: `src/app.rs` is the entire sidecar and is
|
||||
shared, while `src/windows.rs` and `src/unix.rs` do nothing but start it and tell it when to stop.
|
||||
The Windows crates are declared under `[target.'cfg(windows)'.dependencies]`, so Cargo neither
|
||||
resolves nor builds them for a Linux target.
|
||||
|
||||
Registering the service is the installer's job; to do it by hand see
|
||||
[INSTALL.md Appendix A4](https://gitea.whitlocktech.com/RunicGateway/docs/src/branch/main/installer/INSTALL.md).
|
||||
|
||||
`.gitea/workflows/release.yml` cross-compiles Linux + Windows binaries and cuts a Gitea release on
|
||||
every merge to `main` (conventional-commit versioning). See [`sidecar/README.md`](sidecar/README.md)
|
||||
for configuration and the wire protocol.
|
||||
|
||||
95
sidecar/Cargo.lock
generated
95
sidecar/Cargo.lock
generated
@@ -242,6 +242,15 @@ version = "2.5.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "217698eaf96b4a3f0bc4f3662aaa55bdf913cd54d7204591faa790070c6d0853"
|
||||
|
||||
[[package]]
|
||||
name = "crossbeam-channel"
|
||||
version = "0.5.16"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "d85363c37faeca707aef026efa9f3b34d077bce547e48f770770625c6013679e"
|
||||
dependencies = [
|
||||
"crossbeam-utils",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "crossbeam-queue"
|
||||
version = "0.3.13"
|
||||
@@ -284,6 +293,12 @@ dependencies = [
|
||||
"zeroize",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "deranged"
|
||||
version = "0.5.8"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "7cd812cc2bc1d69d4764bd80df88b4317eaef9e773c75226407d9bc0876b211c"
|
||||
|
||||
[[package]]
|
||||
name = "digest"
|
||||
version = "0.10.7"
|
||||
@@ -921,6 +936,12 @@ dependencies = [
|
||||
"zeroize",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "num-conv"
|
||||
version = "0.2.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "521739c6d2bac4aa25192232afe6841231376b2b26d4d9fae5ecf8ca5772e441"
|
||||
|
||||
[[package]]
|
||||
name = "num-integer"
|
||||
version = "0.1.46"
|
||||
@@ -1048,6 +1069,12 @@ dependencies = [
|
||||
"zerovec",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "powerfmt"
|
||||
version = "0.2.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "439ee305def115ba05938db6eb1644ff94165c5ab5e9420d1c1bcedbba909391"
|
||||
|
||||
[[package]]
|
||||
name = "ppv-lite86"
|
||||
version = "0.2.21"
|
||||
@@ -1565,6 +1592,12 @@ version = "2.6.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "13c2bddecc57b384dee18652358fb23172facb8a2c51ccc10d74c157bdea3292"
|
||||
|
||||
[[package]]
|
||||
name = "symlink"
|
||||
version = "0.1.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "a7973cce6668464ea31f176d85b13c7ab3bba2cb3b77a2ed26abd7801688010a"
|
||||
|
||||
[[package]]
|
||||
name = "syn"
|
||||
version = "2.0.118"
|
||||
@@ -1642,6 +1675,36 @@ dependencies = [
|
||||
"cfg-if",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "time"
|
||||
version = "0.3.55"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "cdb87b95ec50ddfa440816d227a17b2ccbdda963a316a727fda0fc4334f7d134"
|
||||
dependencies = [
|
||||
"deranged",
|
||||
"num-conv",
|
||||
"powerfmt",
|
||||
"serde_core",
|
||||
"time-core",
|
||||
"time-macros",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "time-core"
|
||||
version = "0.1.9"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "9e1c906769ad99c88eaa54e728060edef082f8e358ff32030cb7c7d315e81109"
|
||||
|
||||
[[package]]
|
||||
name = "time-macros"
|
||||
version = "0.2.32"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "7e689342a48d2ea927c87ea50cabf8594854bf940e9310208848d680d668ed85"
|
||||
dependencies = [
|
||||
"num-conv",
|
||||
"time-core",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "tinystr"
|
||||
version = "0.8.3"
|
||||
@@ -1798,6 +1861,19 @@ dependencies = [
|
||||
"tracing-core",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "tracing-appender"
|
||||
version = "0.2.5"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "050686193eb999b4bb3bc2acfa891a13da00f79734704c4b8b4ef1a10b368a3c"
|
||||
dependencies = [
|
||||
"crossbeam-channel",
|
||||
"symlink",
|
||||
"thiserror 2.0.18",
|
||||
"time",
|
||||
"tracing-subscriber",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "tracing-attributes"
|
||||
version = "0.1.31"
|
||||
@@ -1913,7 +1989,9 @@ dependencies = [
|
||||
"tokio",
|
||||
"toml",
|
||||
"tracing",
|
||||
"tracing-appender",
|
||||
"tracing-subscriber",
|
||||
"windows-service",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -2025,6 +2103,12 @@ dependencies = [
|
||||
"wasite",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "widestring"
|
||||
version = "1.2.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "72069c3113ab32ab29e5584db3c6ec55d416895e60715417b5b883a357c3e471"
|
||||
|
||||
[[package]]
|
||||
name = "windows-core"
|
||||
version = "0.62.2"
|
||||
@@ -2075,6 +2159,17 @@ dependencies = [
|
||||
"windows-link",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "windows-service"
|
||||
version = "0.8.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "857224b3b211c6f3616921f081ee54721ee3ad2ace2fac6a6337e032f7b4dcf2"
|
||||
dependencies = [
|
||||
"bitflags",
|
||||
"widestring",
|
||||
"windows-sys 0.61.2",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "windows-strings"
|
||||
version = "0.5.1"
|
||||
|
||||
@@ -17,5 +17,12 @@ toml = "0.8"
|
||||
getrandom = "0.2"
|
||||
chrono = { version = "0.4", default-features = false, features = ["std", "clock"] }
|
||||
|
||||
# Speaking the Windows Service Control Manager's startup handshake, and logging somewhere other
|
||||
# than the stdout a service does not have. Declared per target so Cargo neither resolves nor builds
|
||||
# either crate for Linux — the Linux binary is byte-for-byte unaffected by Windows service support.
|
||||
[target.'cfg(windows)'.dependencies]
|
||||
windows-service = "0.8"
|
||||
tracing-appender = "0.2"
|
||||
|
||||
[profile.release]
|
||||
opt-level = 2
|
||||
|
||||
571
sidecar/src/app.rs
Normal file
571
sidecar/src/app.rs
Normal file
@@ -0,0 +1,571 @@
|
||||
//! The sidecar itself: everything that happens between "we have a config path" and "we were told
|
||||
//! to stop". Identical on every platform.
|
||||
//!
|
||||
//! This module exists so that *how the process is started and stopped* — a bare `main` under
|
||||
//! systemd, or a `ServiceMain` under the Windows SCM — is the only thing that differs between
|
||||
//! hosts. The shard listener, the config, the store, the web server and the event loop are shared
|
||||
//! code with no `#[cfg]` in sight.
|
||||
//!
|
||||
//! [`run`] is parameterised on the two things a supervisor cares about:
|
||||
//!
|
||||
//! - `ready` is called once the sidecar is actually up (listener bound, store open). The Windows
|
||||
//! service reports `Running` to the SCM there, so a config or bind failure surfaces as a *start*
|
||||
//! failure rather than a service that reports Running and then dies.
|
||||
//! - `shutdown` is whatever "stop" means on this host: Ctrl-C and `SIGTERM` on Unix, the SCM's
|
||||
//! `Stop` control on Windows.
|
||||
|
||||
use std::collections::HashMap;
|
||||
use std::future::Future;
|
||||
use std::sync::atomic::AtomicI64;
|
||||
use std::sync::Arc;
|
||||
use std::time::Instant;
|
||||
|
||||
use serde_json::Value;
|
||||
use tokio::sync::{broadcast, mpsc};
|
||||
use tracing::info;
|
||||
|
||||
use crate::{config, rpc, shard, store, web};
|
||||
|
||||
/// Runs the sidecar until `shutdown` resolves.
|
||||
///
|
||||
/// `config_path` is the `--config` argument, or `None` to resolve `$UOLINK_CONFIG` and the default
|
||||
/// as usual.
|
||||
pub async fn run<R, S>(config_path: Option<&str>, ready: R, shutdown: S) -> anyhow::Result<()>
|
||||
where
|
||||
R: FnOnce(),
|
||||
S: Future<Output = ()>,
|
||||
{
|
||||
info!("uo-link sidecar starting");
|
||||
|
||||
let loaded = config::Config::load(config_path)?;
|
||||
let cfg = loaded.cfg;
|
||||
info!(
|
||||
config = %loaded.path.display(),
|
||||
shard = %cfg.shard.bind,
|
||||
web = %cfg.web.bind,
|
||||
db = %cfg.store.path,
|
||||
auth = cfg.auth_required(),
|
||||
"configuration loaded"
|
||||
);
|
||||
|
||||
// Shard link: events in, commands out.
|
||||
let (event_tx, mut event_rx) = mpsc::unbounded_channel::<shard::ShardEvent>();
|
||||
let handle = shard::serve(&cfg.shard.bind, event_tx).await?;
|
||||
|
||||
// Live feed: every shard event fans out to all connected website WebSocket clients.
|
||||
let (bcast_tx, _) = broadcast::channel::<String>(1024);
|
||||
|
||||
// Request/reply correlation for REST queries.
|
||||
let rpc = rpc::Rpc::new();
|
||||
|
||||
// Durable store: event history, economy series, cached profiles, link map.
|
||||
let store = store::Store::open(&cfg.store.path).await?;
|
||||
|
||||
// Health/observability state.
|
||||
let started = Instant::now();
|
||||
let last_event = Arc::new(AtomicI64::new(0));
|
||||
|
||||
// Website-facing HTTP server.
|
||||
let web_state = web::AppState {
|
||||
events: bcast_tx.clone(),
|
||||
shard: handle.clone(),
|
||||
rpc: rpc.clone(),
|
||||
store: store.clone(),
|
||||
token: Arc::new(cfg.web.auth_token.clone()),
|
||||
started,
|
||||
last_event: last_event.clone(),
|
||||
};
|
||||
let web_bind = cfg.web.bind.clone();
|
||||
tokio::spawn(async move {
|
||||
if let Err(e) = web::serve(&web_bind, web_state).await {
|
||||
tracing::error!(error = %e, "web server exited");
|
||||
}
|
||||
});
|
||||
|
||||
// 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: log it, persist it, broadcast it.
|
||||
let feed_tx = bcast_tx.clone();
|
||||
let route_rpc = rpc.clone();
|
||||
let event_store = store.clone();
|
||||
let last_event_ts = last_event.clone();
|
||||
let replay_handle = handle.clone(); // re-push external news to the shard on (re)connect
|
||||
let mut total: u64 = 0;
|
||||
// Partly-received guild rosters, keyed by guild id. Lives in the event-loop task, so it needs
|
||||
// no lock and dies with the loop. See `accumulate_roster`.
|
||||
let mut roster_parts: HashMap<i64, RosterParts> = HashMap::new();
|
||||
tokio::spawn(async move {
|
||||
while let Some(ev) = event_rx.recv().await {
|
||||
// Any line from the shard — including pong heartbeats — is a sign of life.
|
||||
last_event_ts.store(now_ms(), std::sync::atomic::Ordering::Relaxed);
|
||||
|
||||
if route_rpc.try_route(&ev.value).await {
|
||||
continue; // consumed as a reply
|
||||
}
|
||||
|
||||
total += 1;
|
||||
match ev.kind.as_str() {
|
||||
"server.hello" | "mob.login" | "mob.logout" | "player.death" | "vendor.sale"
|
||||
| "house.decay" | "link.request" => {
|
||||
info!(kind = %ev.kind, n = total, "{}", ev.value);
|
||||
}
|
||||
_ => tracing::debug!(kind = %ev.kind, n = total, "{}", ev.value),
|
||||
}
|
||||
|
||||
// Persist, then broadcast. `pong` and `ws.hello` are ephemeral chatter, not history.
|
||||
if ev.kind != "pong" {
|
||||
let t = ev
|
||||
.value
|
||||
.get("t")
|
||||
.and_then(|v| v.as_i64())
|
||||
.unwrap_or_else(now_ms);
|
||||
let text = ev.value.to_string();
|
||||
if let Err(e) = event_store.insert_event(t, &ev.kind, &text).await {
|
||||
tracing::warn!(error = %e, "failed to persist event");
|
||||
}
|
||||
|
||||
// The champ board is a live projection: champ.update folds in the latest state (one
|
||||
// row per spawn), champ.remove drops a spawn that despawned or was slain.
|
||||
match ev.kind.as_str() {
|
||||
"champ.update" => {
|
||||
if let Some(serial) = ev.value.get("serial").and_then(|s| s.as_str()) {
|
||||
if let Err(e) = event_store
|
||||
.upsert_champ(
|
||||
serial,
|
||||
ev.value.get("status").and_then(|s| s.as_str()),
|
||||
ev.value.get("name").and_then(|n| n.as_str()),
|
||||
&text,
|
||||
t,
|
||||
)
|
||||
.await
|
||||
{
|
||||
tracing::warn!(error = %e, "failed to upsert champ board");
|
||||
}
|
||||
}
|
||||
}
|
||||
"champ.remove" => {
|
||||
if let Some(serial) = ev.value.get("serial").and_then(|s| s.as_str()) {
|
||||
if let Err(e) = event_store.delete_champ(serial).await {
|
||||
tracing::warn!(error = %e, "failed to remove champ board row");
|
||||
}
|
||||
}
|
||||
}
|
||||
// Guild board (Protocol 2.0): guild.update folds in the latest roster (one row
|
||||
// per guild id); guild.remove drops a disbanded guild.
|
||||
"guild.update" => {
|
||||
if let Some(id) = ev.value.get("id").and_then(|v| v.as_i64()) {
|
||||
if let Err(e) = event_store
|
||||
.upsert_guild(
|
||||
id,
|
||||
ev.value.get("name").and_then(|n| n.as_str()),
|
||||
&text,
|
||||
t,
|
||||
)
|
||||
.await
|
||||
{
|
||||
tracing::warn!(error = %e, "failed to upsert guild board");
|
||||
}
|
||||
}
|
||||
}
|
||||
// Guild roster (Protocol 4): the member list that `guild.update`'s counts cannot
|
||||
// express. It writes a *different column* of the same row, so it never races
|
||||
// guild.update. `guild.leave` deliberately has no arm here — the sidecar
|
||||
// forwards it (persisted and broadcast below, like any event) and the board's
|
||||
// roster self-corrects on the next `guild.roster`, which the shard re-emits
|
||||
// whenever the member set changes. Keeping the delta out of the board is what
|
||||
// keeps the sidecar a forwarder rather than a thing that maintains state.
|
||||
//
|
||||
// A roster over the shard's per-line cap arrives as several frames, so it is
|
||||
// reassembled before it is stored — see `accumulate_roster` for why that happens
|
||||
// here rather than by appending to the column.
|
||||
"guild.roster" => {
|
||||
if let Some(id) = ev.value.get("id").and_then(|v| v.as_i64()) {
|
||||
let seq = ev.value.get("seq").and_then(|v| v.as_i64()).unwrap_or(0);
|
||||
let more = ev
|
||||
.value
|
||||
.get("more")
|
||||
.and_then(|v| v.as_bool())
|
||||
.unwrap_or(false);
|
||||
let members = ev
|
||||
.value
|
||||
.get("members")
|
||||
.and_then(|m| m.as_array())
|
||||
.cloned()
|
||||
.unwrap_or_default();
|
||||
|
||||
if let Some(complete) =
|
||||
accumulate_roster(&mut roster_parts, id, seq, more, members)
|
||||
{
|
||||
let json = serde_json::Value::Array(complete).to_string();
|
||||
if let Err(e) = event_store.upsert_guild_roster(id, &json, t).await
|
||||
{
|
||||
tracing::warn!(error = %e, "failed to upsert guild roster");
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
"guild.remove" => {
|
||||
if let Some(id) = ev.value.get("id").and_then(|v| v.as_i64()) {
|
||||
if let Err(e) = event_store.delete_guild(id).await {
|
||||
tracing::warn!(error = %e, "failed to remove guild board row");
|
||||
}
|
||||
}
|
||||
}
|
||||
// Governor board (Protocol 2.0): city.update folds in each city's latest
|
||||
// governance state (one row per city).
|
||||
"city.update" => {
|
||||
if let Some(city) = ev.value.get("city").and_then(|c| c.as_str()) {
|
||||
if let Err(e) = event_store.upsert_governor(city, &text, t).await {
|
||||
tracing::warn!(error = %e, "failed to upsert governor board");
|
||||
}
|
||||
}
|
||||
}
|
||||
// House registry (Protocol 2.0): house.update folds in each house's latest state
|
||||
// (one row per serial); house.remove drops a demolished/traded house.
|
||||
"house.update" => {
|
||||
if let Some(serial) = ev.value.get("serial").and_then(|s| s.as_str()) {
|
||||
if let Err(e) = event_store
|
||||
.upsert_house(
|
||||
serial,
|
||||
ev.value.get("name").and_then(|n| n.as_str()),
|
||||
&text,
|
||||
t,
|
||||
)
|
||||
.await
|
||||
{
|
||||
tracing::warn!(error = %e, "failed to upsert house registry");
|
||||
}
|
||||
}
|
||||
}
|
||||
"house.remove" => {
|
||||
if let Some(serial) = ev.value.get("serial").and_then(|s| s.as_str()) {
|
||||
if let Err(e) = event_store.delete_house(serial).await {
|
||||
tracing::warn!(error = %e, "failed to remove house registry row");
|
||||
}
|
||||
}
|
||||
}
|
||||
// Points/loyalty boards (Protocol 3.0): one row per point system, keyed by
|
||||
// the shard's own PointsType name. The plugin only emits a system whose top N
|
||||
// actually moved, so this is a sparse stream of overwrites — and there is no
|
||||
// `points.remove` to handle, because the shard's set of systems is fixed at
|
||||
// startup and cannot shrink.
|
||||
"points.board" => {
|
||||
if let Some(system) = ev.value.get("system").and_then(|s| s.as_str()) {
|
||||
if let Err(e) = event_store
|
||||
.upsert_points_board(
|
||||
system,
|
||||
ev.value.get("nameString").and_then(|n| n.as_str()),
|
||||
&text,
|
||||
t,
|
||||
)
|
||||
.await
|
||||
{
|
||||
tracing::warn!(error = %e, "failed to upsert points board");
|
||||
}
|
||||
}
|
||||
}
|
||||
// Player-vendor market index (Protocol 3.0). Each frame is authoritative for
|
||||
// one vendor — the shard's round-robin sweep only emits a shop whose contents,
|
||||
// prices or location actually moved — so this is a whole-row overwrite.
|
||||
//
|
||||
// Unlike the boards above there IS a remove: a vendor is dismissed, expires, or
|
||||
// its owner switches off the in-game Vendor Search flag, and any of those must
|
||||
// take the shop off the site. The last of the three is a privacy control, so
|
||||
// dropping the row promptly is the point rather than housekeeping.
|
||||
"vendor.listing" => {
|
||||
if let Some(serial) = ev.value.get("serial").and_then(|s| s.as_str()) {
|
||||
let loc = ev.value.get("location");
|
||||
let field = |k: &str| loc.and_then(|l| l.get(k));
|
||||
if let Err(e) = event_store
|
||||
.upsert_vendor(
|
||||
serial,
|
||||
ev.value.get("shopName").and_then(|v| v.as_str()),
|
||||
ev.value.get("ownerName").and_then(|v| v.as_str()),
|
||||
field("map").and_then(|v| v.as_str()),
|
||||
field("x").and_then(|v| v.as_i64()),
|
||||
field("y").and_then(|v| v.as_i64()),
|
||||
field("region").and_then(|v| v.as_str()),
|
||||
ev.value.get("count").and_then(|v| v.as_i64()),
|
||||
&text,
|
||||
t,
|
||||
)
|
||||
.await
|
||||
{
|
||||
tracing::warn!(error = %e, "failed to upsert vendor listing");
|
||||
}
|
||||
}
|
||||
}
|
||||
"vendor.listing.remove" => {
|
||||
if let Some(serial) = ev.value.get("serial").and_then(|s| s.as_str()) {
|
||||
if let Err(e) = event_store.delete_vendor(serial).await {
|
||||
tracing::warn!(error = %e, "failed to remove vendor listing");
|
||||
}
|
||||
}
|
||||
}
|
||||
// Shard ruleset (Protocol 3.0): a singleton projection. The shard re-emits
|
||||
// world.ruleset on every connect, so this row is simply overwritten; `rev`
|
||||
// lets a reader tell a re-send from an actual config change.
|
||||
"world.ruleset" => {
|
||||
if let Err(e) = event_store
|
||||
.upsert_ruleset(ev.value.get("rev").and_then(|r| r.as_str()), &text, t)
|
||||
.await
|
||||
{
|
||||
tracing::warn!(error = %e, "failed to upsert ruleset");
|
||||
}
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
|
||||
// On a shard (re)connect, re-push the stored external news: the shard rebuilds
|
||||
// TownCryerSystem.NewsEntries from scratch each boot and does not persist ours. Replay
|
||||
// with announce=false so a restart does not re-proclaim every article at once. news.add
|
||||
// is idempotent by id, so replaying to a still-populated shard is harmless.
|
||||
if ev.kind == "server.hello" {
|
||||
// A (re)connected shard restarts every roster from `seq` 0, so any half-received
|
||||
// one belongs to the previous connection and can never be completed.
|
||||
roster_parts.clear();
|
||||
|
||||
match event_store.news_all().await {
|
||||
Ok(items) => {
|
||||
for mut item in items {
|
||||
if let Some(obj) = item.as_object_mut() {
|
||||
obj.insert("announce".to_string(), serde_json::json!(false));
|
||||
}
|
||||
if !replay_handle.send(item.to_string()).await {
|
||||
break; // shard went away mid-replay
|
||||
}
|
||||
}
|
||||
}
|
||||
Err(e) => tracing::warn!(error = %e, "news replay: could not read stored news"),
|
||||
}
|
||||
}
|
||||
|
||||
let _ = feed_tx.send(ev.value.to_string());
|
||||
}
|
||||
});
|
||||
|
||||
// Heartbeat to the shard, exercising the command path.
|
||||
let ping_handle = handle.clone();
|
||||
tokio::spawn(async move {
|
||||
loop {
|
||||
tokio::time::sleep(std::time::Duration::from_secs(15)).await;
|
||||
if ping_handle.is_connected().await {
|
||||
let _ = ping_handle
|
||||
.send(r#"{"kind":"ping","id":"sidecar-heartbeat"}"#.to_string())
|
||||
.await;
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
// Everything that can fail at startup has now either failed or succeeded: the shard port is
|
||||
// bound and the store is open. A supervisor may call this "started".
|
||||
ready();
|
||||
|
||||
shutdown.await;
|
||||
info!("shutting down");
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub fn now_ms() -> i64 {
|
||||
use std::time::{SystemTime, UNIX_EPOCH};
|
||||
SystemTime::now()
|
||||
.duration_since(UNIX_EPOCH)
|
||||
.map(|d| d.as_millis() as i64)
|
||||
.unwrap_or(0)
|
||||
}
|
||||
|
||||
/// A guild roster that has arrived in part: the `seq` expected next, and what has accumulated.
|
||||
struct RosterParts {
|
||||
next_seq: i64,
|
||||
members: Vec<Value>,
|
||||
}
|
||||
|
||||
/// Refuses to accumulate a roster past this many members. The shard caps its own frames, so
|
||||
/// exceeding this means a shard that is buggy or not what it claims to be — and the one thing this
|
||||
/// buffer must not do is grow without bound on its say-so.
|
||||
const MAX_ROSTER_MEMBERS: usize = 50_000;
|
||||
|
||||
/// Reassembles a `guild.roster` that the shard split across frames, returning the whole member list
|
||||
/// once the final frame arrives and `None` while one is still incomplete.
|
||||
///
|
||||
/// Reassembly happens **here, in memory, before the store** rather than by appending to the
|
||||
/// `members` column, for two reasons. Appending would make the write a read-modify-write — the exact
|
||||
/// thing the two-column board design exists to avoid — and it would publish a torn roster: a reader
|
||||
/// hitting `GET /guilds` between frames would see a partial member list as though it were the truth.
|
||||
/// Buffering keeps the store's write a single atomic upsert of a complete roster.
|
||||
///
|
||||
/// This is transport-level reassembly, not domain state: it is the same category of work as turning
|
||||
/// bytes into a line, and it holds nothing once a roster is complete. That is what keeps it
|
||||
/// compatible with the sidecar being a forwarder.
|
||||
///
|
||||
/// The ordinary case — a guild inside the shard's per-line cap, which is every realistic one —
|
||||
/// arrives as `seq` 0 with `more` false and is returned immediately without ever touching the map.
|
||||
fn accumulate_roster(
|
||||
parts: &mut HashMap<i64, RosterParts>,
|
||||
id: i64,
|
||||
seq: i64,
|
||||
more: bool,
|
||||
members: Vec<Value>,
|
||||
) -> Option<Vec<Value>> {
|
||||
if seq == 0 {
|
||||
// A fresh roster supersedes any partial one: the shard restarts at 0 every time it emits,
|
||||
// so a leftover buffer is from an emission that was interrupted and will never finish.
|
||||
parts.remove(&id);
|
||||
|
||||
if !more {
|
||||
return Some(members);
|
||||
}
|
||||
|
||||
parts.insert(
|
||||
id,
|
||||
RosterParts {
|
||||
next_seq: 1,
|
||||
members,
|
||||
},
|
||||
);
|
||||
return None;
|
||||
}
|
||||
|
||||
let entry = match parts.get_mut(&id) {
|
||||
Some(entry) => entry,
|
||||
// A continuation with nothing to continue: the sidecar started, or the shard reconnected,
|
||||
// midway through an emission. Dropping it is right — the next full roster is complete.
|
||||
None => {
|
||||
tracing::debug!(
|
||||
guild = id,
|
||||
seq,
|
||||
"roster continuation with no start; ignoring"
|
||||
);
|
||||
return None;
|
||||
}
|
||||
};
|
||||
|
||||
if entry.next_seq != seq {
|
||||
tracing::warn!(
|
||||
guild = id,
|
||||
expected = entry.next_seq,
|
||||
got = seq,
|
||||
"roster frames out of order; discarding the partial roster"
|
||||
);
|
||||
parts.remove(&id);
|
||||
return None;
|
||||
}
|
||||
|
||||
entry.members.extend(members);
|
||||
|
||||
if entry.members.len() > MAX_ROSTER_MEMBERS {
|
||||
tracing::warn!(
|
||||
guild = id,
|
||||
len = entry.members.len(),
|
||||
"roster exceeded the reassembly cap; discarding"
|
||||
);
|
||||
parts.remove(&id);
|
||||
return None;
|
||||
}
|
||||
|
||||
if more {
|
||||
entry.next_seq = seq + 1;
|
||||
return None;
|
||||
}
|
||||
|
||||
parts.remove(&id).map(|done| done.members)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
fn members(names: &[&str]) -> Vec<Value> {
|
||||
names
|
||||
.iter()
|
||||
.map(|n| serde_json::json!({"name": n}))
|
||||
.collect()
|
||||
}
|
||||
|
||||
fn names(vs: &[Value]) -> Vec<String> {
|
||||
vs.iter()
|
||||
.map(|v| v["name"].as_str().unwrap_or_default().to_string())
|
||||
.collect()
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_single_frame_roster_is_returned_immediately() {
|
||||
// Every realistic guild takes this path, and it must not depend on the buffer at all.
|
||||
let mut parts = HashMap::new();
|
||||
let out = accumulate_roster(&mut parts, 1, 0, false, members(&["Ada", "Bo"]));
|
||||
|
||||
assert_eq!(names(&out.expect("complete")), ["Ada", "Bo"]);
|
||||
assert!(parts.is_empty(), "nothing should be buffered");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_chunked_roster_reassembles_in_order() {
|
||||
// The case the live rig caught: without this, only the final frame survived and a
|
||||
// 155-member guild appeared on the board with 3 members.
|
||||
let mut parts = HashMap::new();
|
||||
|
||||
assert!(accumulate_roster(&mut parts, 1, 0, true, members(&["Ada"])).is_none());
|
||||
assert!(accumulate_roster(&mut parts, 1, 1, true, members(&["Bo"])).is_none());
|
||||
let out = accumulate_roster(&mut parts, 1, 2, false, members(&["Cy"]));
|
||||
|
||||
assert_eq!(names(&out.expect("complete")), ["Ada", "Bo", "Cy"]);
|
||||
assert!(parts.is_empty(), "buffer is released once complete");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_restarted_roster_supersedes_a_partial_one() {
|
||||
// A shard that reconnects mid-emission starts again at seq 0. The abandoned frames must not
|
||||
// end up spliced onto the front of the new roster.
|
||||
let mut parts = HashMap::new();
|
||||
|
||||
assert!(accumulate_roster(&mut parts, 1, 0, true, members(&["Stale"])).is_none());
|
||||
let out = accumulate_roster(&mut parts, 1, 0, false, members(&["Fresh"]));
|
||||
|
||||
assert_eq!(names(&out.expect("complete")), ["Fresh"]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn an_out_of_order_frame_discards_the_partial_roster() {
|
||||
// Better to publish nothing and wait for the next full emission than to store a roster with
|
||||
// a hole in it that nothing downstream could detect.
|
||||
let mut parts = HashMap::new();
|
||||
|
||||
assert!(accumulate_roster(&mut parts, 1, 0, true, members(&["Ada"])).is_none());
|
||||
assert!(accumulate_roster(&mut parts, 1, 2, false, members(&["Skipped"])).is_none());
|
||||
assert!(parts.is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_continuation_with_no_start_is_ignored() {
|
||||
// The sidecar restarting midway through a shard's emission.
|
||||
let mut parts = HashMap::new();
|
||||
|
||||
assert!(accumulate_roster(&mut parts, 1, 3, false, members(&["Orphan"])).is_none());
|
||||
assert!(parts.is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn two_guilds_reassemble_independently() {
|
||||
// Rosters for different guilds interleave freely — the sweep emits one guild after another
|
||||
// and nothing serialises them on the wire.
|
||||
let mut parts = HashMap::new();
|
||||
|
||||
assert!(accumulate_roster(&mut parts, 1, 0, true, members(&["A1"])).is_none());
|
||||
assert!(accumulate_roster(&mut parts, 2, 0, true, members(&["B1"])).is_none());
|
||||
let g2 = accumulate_roster(&mut parts, 2, 1, false, members(&["B2"]));
|
||||
let g1 = accumulate_roster(&mut parts, 1, 1, false, members(&["A2"]));
|
||||
|
||||
assert_eq!(names(&g2.expect("guild 2")), ["B1", "B2"]);
|
||||
assert_eq!(names(&g1.expect("guild 1")), ["A1", "A2"]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn an_empty_roster_is_a_complete_roster() {
|
||||
// A guild whose last member left emits one frame with an empty array. Treating that as
|
||||
// "nothing to store" would leave the board showing the roster it had before.
|
||||
let mut parts = HashMap::new();
|
||||
let out = accumulate_roster(&mut parts, 1, 0, false, vec![]);
|
||||
|
||||
assert_eq!(out.expect("complete").len(), 0);
|
||||
}
|
||||
}
|
||||
@@ -1,21 +1,37 @@
|
||||
//! uo-link sidecar.
|
||||
//!
|
||||
//! Terminates the loopback link to the ServUO shard and exposes a website-facing HTTP surface. So
|
||||
//! far: the shard link (bidirectional) and a WebSocket live feed. REST queries and SQLite come next.
|
||||
//! Terminates the loopback link to the ServUO shard and exposes a website-facing HTTP surface: the
|
||||
//! shard link (bidirectional), a WebSocket live feed, and REST queries backed by SQLite.
|
||||
//!
|
||||
//! # Layout
|
||||
//!
|
||||
//! `main` does argument handling and nothing else; the sidecar proper lives in [`app`] and is the
|
||||
//! same code on every platform. Only *how the process is started and stopped* is
|
||||
//! platform-specific:
|
||||
//!
|
||||
//! ```text
|
||||
//! systemd ──▶ main ──▶ unix::run ─────────────────────────┐
|
||||
//! ├──▶ app::run
|
||||
//! SCM ─────▶ main ──▶ windows::run ──▶ ServiceMain ───────┘
|
||||
//! └─▶ console fallback ──┘
|
||||
//! ```
|
||||
//!
|
||||
//! The Windows half is not optional politeness: the SCM refuses to supervise a program that does
|
||||
//! not speak its startup handshake (see [`windows`]). The platform modules are gated with `#[cfg]`
|
||||
//! and their dependencies are declared per target, so none of it reaches a Linux build.
|
||||
|
||||
mod app;
|
||||
mod cli;
|
||||
mod config;
|
||||
mod rpc;
|
||||
mod shard;
|
||||
mod store;
|
||||
#[cfg(unix)]
|
||||
mod unix;
|
||||
mod web;
|
||||
#[cfg(windows)]
|
||||
mod windows;
|
||||
|
||||
use std::sync::atomic::AtomicI64;
|
||||
use std::sync::Arc;
|
||||
use std::time::Instant;
|
||||
|
||||
use tokio::sync::{broadcast, mpsc};
|
||||
use tracing::info;
|
||||
use tracing_subscriber::EnvFilter;
|
||||
|
||||
/// Wire-protocol version between the website and the sidecar. Bump this whenever an event or
|
||||
@@ -30,10 +46,57 @@ use tracing_subscriber::EnvFilter;
|
||||
/// `vendor.listing.remove`, with the `GET /ruleset`, `/points` and `/market` reads that serve them
|
||||
/// from the store. Same shape as the v2 bump — the kinds are additive, the endpoints are not — and
|
||||
/// there is deliberately no feature-negotiation array: v3 implies all three kinds.
|
||||
pub const PROTOCOL_VERSION: u32 = 3;
|
||||
///
|
||||
/// v4 (Protocol 4): adds `guild.roster` and `guild.leave`, giving the guild board a real member list
|
||||
/// instead of the member *count* that was all v2 could express. Additive in the same way again — the
|
||||
/// kinds are new, `GET /guilds` grows a `roster` key, and nothing existing changed shape. This is the
|
||||
/// first bump that also needed a **store migration** (`guilds.members`), because it is the first to
|
||||
/// add a column to a table that already exists rather than a whole new table; see `store::migrate`.
|
||||
///
|
||||
/// 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.
|
||||
pub const PROTOCOL_VERSION: u32 = 7;
|
||||
|
||||
#[tokio::main]
|
||||
async fn main() -> anyhow::Result<()> {
|
||||
// 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
|
||||
// platform module ends up running.
|
||||
fn main() -> anyhow::Result<()> {
|
||||
let args = match cli::parse(std::env::args().skip(1)) {
|
||||
Ok(args) => args,
|
||||
Err(msg) => {
|
||||
@@ -66,299 +129,15 @@ async fn main() -> anyhow::Result<()> {
|
||||
cli::Mode::Run => {}
|
||||
}
|
||||
|
||||
init_tracing();
|
||||
info!("uo-link sidecar starting");
|
||||
#[cfg(windows)]
|
||||
return windows::run(args.config.as_deref());
|
||||
|
||||
let loaded = config::Config::load(args.config.as_deref())?;
|
||||
let cfg = loaded.cfg;
|
||||
info!(
|
||||
config = %loaded.path.display(),
|
||||
shard = %cfg.shard.bind,
|
||||
web = %cfg.web.bind,
|
||||
db = %cfg.store.path,
|
||||
auth = cfg.auth_required(),
|
||||
"configuration loaded"
|
||||
);
|
||||
|
||||
// Shard link: events in, commands out.
|
||||
let (event_tx, mut event_rx) = mpsc::unbounded_channel::<shard::ShardEvent>();
|
||||
let handle = shard::serve(&cfg.shard.bind, event_tx).await?;
|
||||
|
||||
// Live feed: every shard event fans out to all connected website WebSocket clients.
|
||||
let (bcast_tx, _) = broadcast::channel::<String>(1024);
|
||||
|
||||
// Request/reply correlation for REST queries.
|
||||
let rpc = rpc::Rpc::new();
|
||||
|
||||
// Durable store: event history, economy series, cached profiles, link map.
|
||||
let store = store::Store::open(&cfg.store.path).await?;
|
||||
|
||||
// Health/observability state.
|
||||
let started = Instant::now();
|
||||
let last_event = Arc::new(AtomicI64::new(0));
|
||||
|
||||
// Website-facing HTTP server.
|
||||
let web_state = web::AppState {
|
||||
events: bcast_tx.clone(),
|
||||
shard: handle.clone(),
|
||||
rpc: rpc.clone(),
|
||||
store: store.clone(),
|
||||
token: Arc::new(cfg.web.auth_token.clone()),
|
||||
started,
|
||||
last_event: last_event.clone(),
|
||||
};
|
||||
let web_bind = cfg.web.bind.clone();
|
||||
tokio::spawn(async move {
|
||||
if let Err(e) = web::serve(&web_bind, web_state).await {
|
||||
tracing::error!(error = %e, "web server exited");
|
||||
}
|
||||
});
|
||||
|
||||
// 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: log it, persist it, broadcast it.
|
||||
let feed_tx = bcast_tx.clone();
|
||||
let route_rpc = rpc.clone();
|
||||
let event_store = store.clone();
|
||||
let last_event_ts = last_event.clone();
|
||||
let replay_handle = handle.clone(); // re-push external news to the shard on (re)connect
|
||||
let mut total: u64 = 0;
|
||||
tokio::spawn(async move {
|
||||
while let Some(ev) = event_rx.recv().await {
|
||||
// Any line from the shard — including pong heartbeats — is a sign of life.
|
||||
last_event_ts.store(now_ms(), std::sync::atomic::Ordering::Relaxed);
|
||||
|
||||
if route_rpc.try_route(&ev.value).await {
|
||||
continue; // consumed as a reply
|
||||
}
|
||||
|
||||
total += 1;
|
||||
match ev.kind.as_str() {
|
||||
"server.hello" | "mob.login" | "mob.logout" | "player.death" | "vendor.sale"
|
||||
| "house.decay" | "link.request" => {
|
||||
info!(kind = %ev.kind, n = total, "{}", ev.value);
|
||||
}
|
||||
_ => tracing::debug!(kind = %ev.kind, n = total, "{}", ev.value),
|
||||
}
|
||||
|
||||
// Persist, then broadcast. `pong` and `ws.hello` are ephemeral chatter, not history.
|
||||
if ev.kind != "pong" {
|
||||
let t = ev
|
||||
.value
|
||||
.get("t")
|
||||
.and_then(|v| v.as_i64())
|
||||
.unwrap_or_else(now_ms);
|
||||
let text = ev.value.to_string();
|
||||
if let Err(e) = event_store.insert_event(t, &ev.kind, &text).await {
|
||||
tracing::warn!(error = %e, "failed to persist event");
|
||||
}
|
||||
|
||||
// The champ board is a live projection: champ.update folds in the latest state (one
|
||||
// row per spawn), champ.remove drops a spawn that despawned or was slain.
|
||||
match ev.kind.as_str() {
|
||||
"champ.update" => {
|
||||
if let Some(serial) = ev.value.get("serial").and_then(|s| s.as_str()) {
|
||||
if let Err(e) = event_store
|
||||
.upsert_champ(
|
||||
serial,
|
||||
ev.value.get("status").and_then(|s| s.as_str()),
|
||||
ev.value.get("name").and_then(|n| n.as_str()),
|
||||
&text,
|
||||
t,
|
||||
)
|
||||
.await
|
||||
{
|
||||
tracing::warn!(error = %e, "failed to upsert champ board");
|
||||
}
|
||||
}
|
||||
}
|
||||
"champ.remove" => {
|
||||
if let Some(serial) = ev.value.get("serial").and_then(|s| s.as_str()) {
|
||||
if let Err(e) = event_store.delete_champ(serial).await {
|
||||
tracing::warn!(error = %e, "failed to remove champ board row");
|
||||
}
|
||||
}
|
||||
}
|
||||
// Guild board (Protocol 2.0): guild.update folds in the latest roster (one row
|
||||
// per guild id); guild.remove drops a disbanded guild.
|
||||
"guild.update" => {
|
||||
if let Some(id) = ev.value.get("id").and_then(|v| v.as_i64()) {
|
||||
if let Err(e) = event_store
|
||||
.upsert_guild(
|
||||
id,
|
||||
ev.value.get("name").and_then(|n| n.as_str()),
|
||||
&text,
|
||||
t,
|
||||
)
|
||||
.await
|
||||
{
|
||||
tracing::warn!(error = %e, "failed to upsert guild board");
|
||||
}
|
||||
}
|
||||
}
|
||||
"guild.remove" => {
|
||||
if let Some(id) = ev.value.get("id").and_then(|v| v.as_i64()) {
|
||||
if let Err(e) = event_store.delete_guild(id).await {
|
||||
tracing::warn!(error = %e, "failed to remove guild board row");
|
||||
}
|
||||
}
|
||||
}
|
||||
// Governor board (Protocol 2.0): city.update folds in each city's latest
|
||||
// governance state (one row per city).
|
||||
"city.update" => {
|
||||
if let Some(city) = ev.value.get("city").and_then(|c| c.as_str()) {
|
||||
if let Err(e) = event_store.upsert_governor(city, &text, t).await {
|
||||
tracing::warn!(error = %e, "failed to upsert governor board");
|
||||
}
|
||||
}
|
||||
}
|
||||
// House registry (Protocol 2.0): house.update folds in each house's latest state
|
||||
// (one row per serial); house.remove drops a demolished/traded house.
|
||||
"house.update" => {
|
||||
if let Some(serial) = ev.value.get("serial").and_then(|s| s.as_str()) {
|
||||
if let Err(e) = event_store
|
||||
.upsert_house(
|
||||
serial,
|
||||
ev.value.get("name").and_then(|n| n.as_str()),
|
||||
&text,
|
||||
t,
|
||||
)
|
||||
.await
|
||||
{
|
||||
tracing::warn!(error = %e, "failed to upsert house registry");
|
||||
}
|
||||
}
|
||||
}
|
||||
"house.remove" => {
|
||||
if let Some(serial) = ev.value.get("serial").and_then(|s| s.as_str()) {
|
||||
if let Err(e) = event_store.delete_house(serial).await {
|
||||
tracing::warn!(error = %e, "failed to remove house registry row");
|
||||
}
|
||||
}
|
||||
}
|
||||
// Points/loyalty boards (Protocol 3.0): one row per point system, keyed by
|
||||
// the shard's own PointsType name. The plugin only emits a system whose top N
|
||||
// actually moved, so this is a sparse stream of overwrites — and there is no
|
||||
// `points.remove` to handle, because the shard's set of systems is fixed at
|
||||
// startup and cannot shrink.
|
||||
"points.board" => {
|
||||
if let Some(system) = ev.value.get("system").and_then(|s| s.as_str()) {
|
||||
if let Err(e) = event_store
|
||||
.upsert_points_board(
|
||||
system,
|
||||
ev.value.get("nameString").and_then(|n| n.as_str()),
|
||||
&text,
|
||||
t,
|
||||
)
|
||||
.await
|
||||
{
|
||||
tracing::warn!(error = %e, "failed to upsert points board");
|
||||
}
|
||||
}
|
||||
}
|
||||
// Player-vendor market index (Protocol 3.0). Each frame is authoritative for
|
||||
// one vendor — the shard's round-robin sweep only emits a shop whose contents,
|
||||
// prices or location actually moved — so this is a whole-row overwrite.
|
||||
//
|
||||
// Unlike the boards above there IS a remove: a vendor is dismissed, expires, or
|
||||
// its owner switches off the in-game Vendor Search flag, and any of those must
|
||||
// take the shop off the site. The last of the three is a privacy control, so
|
||||
// dropping the row promptly is the point rather than housekeeping.
|
||||
"vendor.listing" => {
|
||||
if let Some(serial) = ev.value.get("serial").and_then(|s| s.as_str()) {
|
||||
let loc = ev.value.get("location");
|
||||
let field = |k: &str| loc.and_then(|l| l.get(k));
|
||||
if let Err(e) = event_store
|
||||
.upsert_vendor(
|
||||
serial,
|
||||
ev.value.get("shopName").and_then(|v| v.as_str()),
|
||||
ev.value.get("ownerName").and_then(|v| v.as_str()),
|
||||
field("map").and_then(|v| v.as_str()),
|
||||
field("x").and_then(|v| v.as_i64()),
|
||||
field("y").and_then(|v| v.as_i64()),
|
||||
field("region").and_then(|v| v.as_str()),
|
||||
ev.value.get("count").and_then(|v| v.as_i64()),
|
||||
&text,
|
||||
t,
|
||||
)
|
||||
.await
|
||||
{
|
||||
tracing::warn!(error = %e, "failed to upsert vendor listing");
|
||||
}
|
||||
}
|
||||
}
|
||||
"vendor.listing.remove" => {
|
||||
if let Some(serial) = ev.value.get("serial").and_then(|s| s.as_str()) {
|
||||
if let Err(e) = event_store.delete_vendor(serial).await {
|
||||
tracing::warn!(error = %e, "failed to remove vendor listing");
|
||||
}
|
||||
}
|
||||
}
|
||||
// Shard ruleset (Protocol 3.0): a singleton projection. The shard re-emits
|
||||
// world.ruleset on every connect, so this row is simply overwritten; `rev`
|
||||
// lets a reader tell a re-send from an actual config change.
|
||||
"world.ruleset" => {
|
||||
if let Err(e) = event_store
|
||||
.upsert_ruleset(ev.value.get("rev").and_then(|r| r.as_str()), &text, t)
|
||||
.await
|
||||
{
|
||||
tracing::warn!(error = %e, "failed to upsert ruleset");
|
||||
}
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
|
||||
// On a shard (re)connect, re-push the stored external news: the shard rebuilds
|
||||
// TownCryerSystem.NewsEntries from scratch each boot and does not persist ours. Replay
|
||||
// with announce=false so a restart does not re-proclaim every article at once. news.add
|
||||
// is idempotent by id, so replaying to a still-populated shard is harmless.
|
||||
if ev.kind == "server.hello" {
|
||||
match event_store.news_all().await {
|
||||
Ok(items) => {
|
||||
for mut item in items {
|
||||
if let Some(obj) = item.as_object_mut() {
|
||||
obj.insert("announce".to_string(), serde_json::json!(false));
|
||||
}
|
||||
if !replay_handle.send(item.to_string()).await {
|
||||
break; // shard went away mid-replay
|
||||
}
|
||||
}
|
||||
}
|
||||
Err(e) => tracing::warn!(error = %e, "news replay: could not read stored news"),
|
||||
}
|
||||
}
|
||||
|
||||
let _ = feed_tx.send(ev.value.to_string());
|
||||
}
|
||||
});
|
||||
|
||||
// Heartbeat to the shard, exercising the command path.
|
||||
let ping_handle = handle.clone();
|
||||
tokio::spawn(async move {
|
||||
loop {
|
||||
tokio::time::sleep(std::time::Duration::from_secs(15)).await;
|
||||
if ping_handle.is_connected().await {
|
||||
let _ = ping_handle
|
||||
.send(r#"{"kind":"ping","id":"sidecar-heartbeat"}"#.to_string())
|
||||
.await;
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
tokio::signal::ctrl_c().await?;
|
||||
info!("shutting down");
|
||||
Ok(())
|
||||
#[cfg(unix)]
|
||||
return unix::run(args.config.as_deref());
|
||||
}
|
||||
|
||||
fn now_ms() -> i64 {
|
||||
use std::time::{SystemTime, UNIX_EPOCH};
|
||||
SystemTime::now()
|
||||
.duration_since(UNIX_EPOCH)
|
||||
.map(|d| d.as_millis() as i64)
|
||||
.unwrap_or(0)
|
||||
}
|
||||
|
||||
fn init_tracing() {
|
||||
/// Logging for a foreground run: human-readable, on stdout.
|
||||
pub fn init_console_tracing() {
|
||||
tracing_subscriber::fmt()
|
||||
.with_env_filter(
|
||||
EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info")),
|
||||
|
||||
@@ -45,6 +45,7 @@ impl Store {
|
||||
.await?;
|
||||
|
||||
sqlx::query(SCHEMA).execute(&pool).await?;
|
||||
migrate(&pool).await?;
|
||||
info!(%path, "store ready");
|
||||
Ok(Self { pool })
|
||||
}
|
||||
@@ -236,12 +237,61 @@ impl Store {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Upserts one guild's member roster (Protocol 4), keyed by guild id, touching **only** the
|
||||
/// `members` column.
|
||||
///
|
||||
/// Deliberately not a write to `json`. That column holds the verbatim `guild.update` line, and a
|
||||
/// roster arriving as its own event must not clobber the snapshot — name, abbreviation, leader,
|
||||
/// online count — that `guild.update` owns. Splitting the two writers across two columns of one
|
||||
/// row is what lets both be plain upserts: neither needs to read the other's value first, so
|
||||
/// there is no read-modify-write and no ordering requirement between the two kinds.
|
||||
///
|
||||
/// The `INSERT` half is not redundant: a roster can arrive before the first `guild.update` for a
|
||||
/// guild, and the row it creates then carries `'{}'` until that update fills it in.
|
||||
pub async fn upsert_guild_roster(
|
||||
&self,
|
||||
id: i64,
|
||||
members_json: &str,
|
||||
t: i64,
|
||||
) -> anyhow::Result<()> {
|
||||
sqlx::query(
|
||||
"INSERT INTO guilds (id, name, json, updated_t, members) VALUES (?, NULL, '{}', ?, ?)
|
||||
ON CONFLICT(id) DO UPDATE SET members = excluded.members, updated_t = excluded.updated_t",
|
||||
)
|
||||
.bind(id)
|
||||
.bind(t)
|
||||
.bind(members_json)
|
||||
.execute(&self.pool)
|
||||
.await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// The full guild board: every guild's latest snapshot, ordered by name.
|
||||
///
|
||||
/// The roster is stored in its own column (see [`Self::upsert_guild_roster`]) and folded into
|
||||
/// the projected object as `roster` here, at read time. A guild that has had a `guild.update`
|
||||
/// but no `guild.roster` yet simply has no `roster` key, which is the honest representation of
|
||||
/// "not known" and distinct from a guild whose roster is genuinely empty.
|
||||
pub async fn guilds_all(&self) -> anyhow::Result<Vec<Value>> {
|
||||
let rows = sqlx::query("SELECT json FROM guilds ORDER BY name, id")
|
||||
let rows = sqlx::query("SELECT json, members FROM guilds ORDER BY name, id")
|
||||
.fetch_all(&self.pool)
|
||||
.await?;
|
||||
Ok(parse_json_column(rows))
|
||||
|
||||
Ok(rows
|
||||
.into_iter()
|
||||
.filter_map(|r| {
|
||||
let mut v: Value = serde_json::from_str(&r.get::<String, _>("json")).ok()?;
|
||||
let members: Option<String> = r.get("members");
|
||||
|
||||
if let (Some(obj), Some(raw)) = (v.as_object_mut(), members) {
|
||||
if let Ok(list) = serde_json::from_str::<Value>(&raw) {
|
||||
obj.insert("roster".into(), list);
|
||||
}
|
||||
}
|
||||
|
||||
Some(v)
|
||||
})
|
||||
.collect())
|
||||
}
|
||||
|
||||
// ---- governor board (Protocol 2.0) ----
|
||||
@@ -513,6 +563,75 @@ impl Store {
|
||||
}
|
||||
}
|
||||
|
||||
/// The schema version this build expects. Bump it, and add the matching arm to [`migrate`], for
|
||||
/// every change that `SCHEMA` alone cannot make to a database that already exists.
|
||||
const SCHEMA_VERSION: i64 = 1;
|
||||
|
||||
/// Brings an existing database forward to [`SCHEMA_VERSION`].
|
||||
///
|
||||
/// `SCHEMA` is `CREATE TABLE IF NOT EXISTS` only, which is enough to *add a table* but cannot add a
|
||||
/// column to a table that is already there. Every schema change up to and including Protocol 3.0
|
||||
/// happened to add whole tables, so this never mattered and `ALTER TABLE` appears nowhere in this
|
||||
/// repo's history. `guilds.members` (Protocol 4) is the first column added to an existing table, so
|
||||
/// the mechanism has to exist now.
|
||||
///
|
||||
/// The version counter is SQLite's own `PRAGMA user_version`: an integer in the database header, so
|
||||
/// it needs no table of its own and cannot be separated from the file it describes. Each step runs
|
||||
/// in a transaction **together with** the bump that records it, so a step either lands completely or
|
||||
/// not at all, and an interrupted run resumes at the right place rather than re-applying half of one.
|
||||
///
|
||||
/// A failure here propagates and aborts startup, deliberately. A half-migrated store answers the
|
||||
/// website with confusing partial data, which is worse than being plainly absent — and the shard
|
||||
/// dials *out* to the sidecar, so a sidecar that refuses to start never stalls the game.
|
||||
async fn migrate(pool: &SqlitePool) -> anyhow::Result<()> {
|
||||
let mut version: i64 = sqlx::query_scalar("PRAGMA user_version")
|
||||
.fetch_one(pool)
|
||||
.await?;
|
||||
|
||||
// A database written by a *newer* sidecar than this binary. This is not an error: every step
|
||||
// here is additive, so a newer schema has only columns and tables an older reader ignores, and
|
||||
// refusing to start would turn "roll the binary back" — a recovery path — into a dead end.
|
||||
if version > SCHEMA_VERSION {
|
||||
tracing::warn!(
|
||||
found = version,
|
||||
expected = SCHEMA_VERSION,
|
||||
"store was written by a newer sidecar; continuing, as migrations are additive"
|
||||
);
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
while version < SCHEMA_VERSION {
|
||||
let next = version + 1;
|
||||
let mut tx = pool.begin().await?;
|
||||
|
||||
match next {
|
||||
// Protocol 4: the guild board carries a member roster. Its own column rather than a
|
||||
// field folded into `json`, because `json` holds the verbatim `guild.update` line and
|
||||
// the two writers must not overwrite each other — see `upsert_guild_roster`.
|
||||
1 => {
|
||||
sqlx::query("ALTER TABLE guilds ADD COLUMN members TEXT")
|
||||
.execute(&mut *tx)
|
||||
.await?;
|
||||
}
|
||||
// Unreachable while SCHEMA_VERSION and this match are edited together, which is the
|
||||
// point of failing loudly rather than silently leaving the counter short.
|
||||
n => anyhow::bail!("no migration step defined for schema version {n}"),
|
||||
}
|
||||
|
||||
// `PRAGMA` takes no bind parameters, so this is formatted — safe because `next` is an i64
|
||||
// this loop produced, never anything from outside the process.
|
||||
sqlx::query(&format!("PRAGMA user_version = {next}"))
|
||||
.execute(&mut *tx)
|
||||
.await?;
|
||||
|
||||
tx.commit().await?;
|
||||
info!(version = next, "schema migration applied");
|
||||
version = next;
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn parse_json_column(rows: Vec<sqlx::sqlite::SqliteRow>) -> Vec<Value> {
|
||||
rows.into_iter()
|
||||
.filter_map(|r| serde_json::from_str(&r.get::<String, _>("json")).ok())
|
||||
@@ -611,3 +730,195 @@ CREATE TABLE IF NOT EXISTS ruleset (
|
||||
updated_t INTEGER NOT NULL
|
||||
);
|
||||
"#;
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
/// A unique scratch database path. Matches `config`'s idiom — `std::env::temp_dir()` plus the
|
||||
/// test name — so the cases stay independent under the parallel test runner.
|
||||
fn scratch(name: &str) -> String {
|
||||
let dir = std::env::temp_dir().join(format!("uo-link-store-test-{name}"));
|
||||
let _ = std::fs::remove_dir_all(&dir);
|
||||
std::fs::create_dir_all(&dir).expect("create scratch dir");
|
||||
dir.join("uo-link.db").to_string_lossy().into_owned()
|
||||
}
|
||||
|
||||
/// The `guilds` table exactly as a pre-Protocol-4 sidecar left it: no `members` column, and
|
||||
/// `user_version` still 0. This is the shape a real operator's database is in before an update,
|
||||
/// and the only starting point where the migration does anything.
|
||||
async fn legacy_db(path: &str) -> SqlitePool {
|
||||
let pool = SqlitePoolOptions::new()
|
||||
.max_connections(1)
|
||||
.connect_with(
|
||||
SqliteConnectOptions::new()
|
||||
.filename(path)
|
||||
.create_if_missing(true),
|
||||
)
|
||||
.await
|
||||
.expect("open legacy db");
|
||||
|
||||
sqlx::query(
|
||||
"CREATE TABLE guilds (
|
||||
id INTEGER PRIMARY KEY,
|
||||
name TEXT,
|
||||
json TEXT NOT NULL,
|
||||
updated_t INTEGER NOT NULL
|
||||
)",
|
||||
)
|
||||
.execute(&pool)
|
||||
.await
|
||||
.expect("create legacy guilds table");
|
||||
|
||||
pool.close().await;
|
||||
pool
|
||||
}
|
||||
|
||||
async fn user_version(store: &Store) -> i64 {
|
||||
sqlx::query_scalar("PRAGMA user_version")
|
||||
.fetch_one(&store.pool)
|
||||
.await
|
||||
.expect("read user_version")
|
||||
}
|
||||
|
||||
async fn guild_columns(store: &Store) -> Vec<String> {
|
||||
sqlx::query("PRAGMA table_info(guilds)")
|
||||
.fetch_all(&store.pool)
|
||||
.await
|
||||
.expect("table_info")
|
||||
.into_iter()
|
||||
.map(|r| r.get::<String, _>("name"))
|
||||
.collect()
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn an_existing_pre_protocol_4_database_gains_the_members_column() {
|
||||
// The case that matters: `SCHEMA`'s CREATE TABLE IF NOT EXISTS is a no-op against a table
|
||||
// that is already there, so without `migrate` this database would never get the column and
|
||||
// every roster write would fail against a live install.
|
||||
let path = scratch("legacy-upgrade");
|
||||
legacy_db(&path).await;
|
||||
|
||||
let store = Store::open(&path).await.expect("open migrates");
|
||||
|
||||
assert!(
|
||||
guild_columns(&store).await.contains(&"members".to_string()),
|
||||
"the migration must add guilds.members to a database that already had the table"
|
||||
);
|
||||
assert_eq!(user_version(&store).await, SCHEMA_VERSION);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_fresh_database_lands_at_the_current_version() {
|
||||
let path = scratch("fresh");
|
||||
let store = Store::open(&path).await.expect("open");
|
||||
|
||||
assert!(guild_columns(&store).await.contains(&"members".to_string()));
|
||||
assert_eq!(user_version(&store).await, SCHEMA_VERSION);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn reopening_an_already_migrated_database_is_a_no_op() {
|
||||
// Every sidecar restart re-runs this path, so a second run must not attempt the ALTER again
|
||||
// — which would fail with "duplicate column name" and, since a migration failure aborts
|
||||
// startup, would leave the sidecar unable to start at all after its first upgrade.
|
||||
let path = scratch("idempotent");
|
||||
legacy_db(&path).await;
|
||||
|
||||
Store::open(&path).await.expect("first open");
|
||||
let store = Store::open(&path).await.expect("second open must succeed");
|
||||
|
||||
assert_eq!(user_version(&store).await, SCHEMA_VERSION);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_roster_does_not_clobber_the_guild_update_snapshot() {
|
||||
// The invariant the two-column split exists to give. `json` holds the verbatim guild.update
|
||||
// line; if a roster write touched it, name/abbr/online would vanish from the board.
|
||||
let path = scratch("no-clobber");
|
||||
let store = Store::open(&path).await.expect("open");
|
||||
|
||||
store
|
||||
.upsert_guild(
|
||||
7,
|
||||
Some("The Cartographers"),
|
||||
r#"{"kind":"guild.update","id":7,"name":"The Cartographers","abbr":"MAP","members":2,"online":1}"#,
|
||||
100,
|
||||
)
|
||||
.await
|
||||
.expect("upsert guild");
|
||||
|
||||
store
|
||||
.upsert_guild_roster(
|
||||
7,
|
||||
r#"[{"serial":"0x1","name":"Ada"},{"serial":"0x2","name":"Bo"}]"#,
|
||||
200,
|
||||
)
|
||||
.await
|
||||
.expect("upsert roster");
|
||||
|
||||
let guilds = store.guilds_all().await.expect("read board");
|
||||
assert_eq!(guilds.len(), 1);
|
||||
let g = &guilds[0];
|
||||
|
||||
assert_eq!(
|
||||
g["name"], "The Cartographers",
|
||||
"guild.update's name survived"
|
||||
);
|
||||
assert_eq!(g["abbr"], "MAP", "guild.update's abbr survived");
|
||||
assert_eq!(g["online"], 1, "guild.update's online count survived");
|
||||
assert_eq!(g["roster"].as_array().expect("roster is an array").len(), 2);
|
||||
assert_eq!(g["roster"][0]["name"], "Ada");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn the_two_writers_are_order_independent() {
|
||||
// A roster can arrive before the first guild.update for a guild — on a reconnect the shard
|
||||
// re-emits both and nothing orders them. Neither write may depend on the other's row.
|
||||
let path = scratch("either-order");
|
||||
let store = Store::open(&path).await.expect("open");
|
||||
|
||||
store
|
||||
.upsert_guild_roster(9, r#"[{"serial":"0x3","name":"Cy"}]"#, 100)
|
||||
.await
|
||||
.expect("roster first");
|
||||
store
|
||||
.upsert_guild(
|
||||
9,
|
||||
Some("Late Arrivals"),
|
||||
r#"{"kind":"guild.update","id":9,"name":"Late Arrivals","abbr":"LTE"}"#,
|
||||
200,
|
||||
)
|
||||
.await
|
||||
.expect("update second");
|
||||
|
||||
let guilds = store.guilds_all().await.expect("read board");
|
||||
assert_eq!(guilds.len(), 1, "one row, not two");
|
||||
assert_eq!(guilds[0]["name"], "Late Arrivals");
|
||||
assert_eq!(guilds[0]["roster"].as_array().expect("roster").len(), 1);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_guild_with_no_roster_yet_has_no_roster_key() {
|
||||
// "Not known" and "known to be empty" are different, and the board must not conflate them:
|
||||
// a website reading `roster: []` would render an empty roster as fact.
|
||||
let path = scratch("absent-roster");
|
||||
let store = Store::open(&path).await.expect("open");
|
||||
|
||||
store
|
||||
.upsert_guild(
|
||||
11,
|
||||
Some("Unswept"),
|
||||
r#"{"kind":"guild.update","id":11,"name":"Unswept"}"#,
|
||||
100,
|
||||
)
|
||||
.await
|
||||
.expect("upsert guild");
|
||||
|
||||
let guilds = store.guilds_all().await.expect("read board");
|
||||
assert!(
|
||||
guilds[0].get("roster").is_none(),
|
||||
"a guild with no roster event must not grow a roster key"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
41
sidecar/src/unix.rs
Normal file
41
sidecar/src/unix.rs
Normal file
@@ -0,0 +1,41 @@
|
||||
//! Unix startup and shutdown.
|
||||
//!
|
||||
//! There is no supervisor protocol to speak: systemd starts the process, and stops it by sending
|
||||
//! `SIGTERM`. All this module does is translate the two signals that mean "stop" into the future
|
||||
//! [`crate::app::run`] waits on, so a `systemctl stop` unwinds the same way a Ctrl-C does instead
|
||||
//! of being killed by the default `SIGTERM` disposition mid-write.
|
||||
|
||||
use tokio::signal::unix::{signal, SignalKind};
|
||||
|
||||
pub fn run(config_path: Option<&str>) -> anyhow::Result<()> {
|
||||
crate::init_console_tracing();
|
||||
|
||||
tokio::runtime::Runtime::new()?.block_on(crate::app::run(config_path, || {}, shutdown_signal()))
|
||||
}
|
||||
|
||||
/// Resolves on the first `SIGINT` or `SIGTERM`.
|
||||
async fn shutdown_signal() {
|
||||
// A failure to install a handler is not worth aborting a running sidecar for: fall back to
|
||||
// pending, which leaves that signal's default disposition (terminate) in place.
|
||||
let mut term = match signal(SignalKind::terminate()) {
|
||||
Ok(s) => s,
|
||||
Err(e) => {
|
||||
tracing::warn!(error = %e, "could not listen for SIGTERM");
|
||||
std::future::pending::<()>().await;
|
||||
unreachable!()
|
||||
}
|
||||
};
|
||||
let mut int = match signal(SignalKind::interrupt()) {
|
||||
Ok(s) => s,
|
||||
Err(e) => {
|
||||
tracing::warn!(error = %e, "could not listen for SIGINT");
|
||||
term.recv().await;
|
||||
return;
|
||||
}
|
||||
};
|
||||
|
||||
tokio::select! {
|
||||
_ = term.recv() => tracing::info!("SIGTERM received"),
|
||||
_ = int.recv() => tracing::info!("SIGINT received"),
|
||||
}
|
||||
}
|
||||
@@ -72,6 +72,41 @@ pub async fn serve(addr: &str, state: AppState) -> anyhow::Result<()> {
|
||||
.route("/admin/ban", post(admin_ban))
|
||||
.route("/admin/unban", post(admin_unban))
|
||||
.route("/admin/broadcast", post(admin_broadcast))
|
||||
// The event plane (protocol 6, EVENTS_PLAN.md Phase 11b). Leases are a live config value
|
||||
// the website holds for a bounded time; the shard restores baseline when the deadline
|
||||
// passes whether or not anyone asks it to. GET lists the whole catalog with current values,
|
||||
// which is the one read both `read()` and `inForce()` on the website's side are served by.
|
||||
.route("/lease", get(lease_list).post(lease_apply))
|
||||
.route("/lease/release", post(lease_release))
|
||||
// The run-scoped participation ledger. `snapshot` is a POST despite being a read: it
|
||||
// carries the caller's `idempotencyKey`, and on a well-attended run the shard walks its
|
||||
// members across ticks rather than in one inbound call -- so a repeat arriving mid-walk is
|
||||
// answered `bridge.busy`, and a read that can be refused as a repeat is not a GET.
|
||||
.route("/participation", post(participation_open))
|
||||
.route(
|
||||
"/participation/:run_id/snapshot",
|
||||
post(participation_snapshot),
|
||||
)
|
||||
.route("/participation/:run_id/close", post(participation_close))
|
||||
// The world verbs (protocol 7, EVENTS_PLAN.md Phase 12a). Five things an event author
|
||||
// can place -- creatures, an enhanced "boss", an oracle NPC, a temporary gate,
|
||||
// decoration -- and ONE command family, because each of them ends in "an object exists
|
||||
// and this run owns it". POST places, GET says what the run still owns, POST .../despawn
|
||||
// gives it back. Ownership is held on the shard, so despawn cannot be pointed at a serial
|
||||
// the run did not create.
|
||||
.route("/world", post(world_spawn))
|
||||
.route("/world/:run_id", get(world_owned))
|
||||
.route("/world/:run_id/despawn", post(world_despawn))
|
||||
// The one-shots (protocol 7 part b, EVENTS_PLAN.md Phase 12b). Neither owned nor
|
||||
// borrowed: an item put into somebody's hands, and a world save. Both are `done is
|
||||
// done`, which is why they are not in the world family -- there is nothing to give
|
||||
// back and no ledger row core would come back for.
|
||||
//
|
||||
// `GET /items` is the shard's own grant allowlist, so the website's dropdown offers
|
||||
// what this shard will actually build rather than what a module guessed.
|
||||
.route("/items", get(item_catalog))
|
||||
.route("/items/grant", post(item_grant))
|
||||
.route("/world/save", post(world_save))
|
||||
// Help-page (support) queue: snapshot the open queue, respond to / close a page.
|
||||
.route("/pages", get(pages_list))
|
||||
.route("/pages/:id/respond", post(page_respond))
|
||||
@@ -246,13 +281,31 @@ fn constant_time_eq(a: &[u8], b: &[u8]) -> bool {
|
||||
|
||||
// ---- shared reply handling ----
|
||||
|
||||
/// Protocol 6. `bridge.busy` says a command carrying this `idempotencyKey` is already in flight on
|
||||
/// the shard: nothing was run, and the caller should come back.
|
||||
///
|
||||
/// It maps to **425 Too Early**, which is what that status is for — a server unwilling to risk
|
||||
/// processing a request that might be a replay. The obvious alternative, 409, is already the
|
||||
/// protocol-version gate's answer, and those two want opposite dispositions from a client: a version
|
||||
/// mismatch is a deployment fault nobody should retry, and a busy shard is a retry that should
|
||||
/// succeed on its own. Sharing a status would have made the difference readable only by inspecting
|
||||
/// the body, which is exactly how a retry loop ends up hiding a mismatched deployment.
|
||||
///
|
||||
/// It is checked BEFORE the `.error` suffix test in each responder below, and it is deliberately not
|
||||
/// spelled `bridge.busy.error`: nothing is wrong. The work is happening.
|
||||
const BUSY_KIND: &str = "bridge.busy";
|
||||
const BUSY_STATUS: StatusCode = StatusCode::TOO_EARLY;
|
||||
|
||||
/// Turns an RPC result into an HTTP response. A `bridge.error` reply from the shard becomes a 4xx;
|
||||
/// a real reply is returned as-is; transport failures map to 503/504.
|
||||
/// a `bridge.busy` reply becomes a 425; a real reply is returned as-is; transport failures map to
|
||||
/// 503/504.
|
||||
fn respond(result: Result<Value, RpcError>) -> (StatusCode, Json<Value>) {
|
||||
match result {
|
||||
Ok(value) => {
|
||||
let kind = value.get("kind").and_then(|k| k.as_str()).unwrap_or("");
|
||||
if kind == "bridge.error" || kind.ends_with(".error") {
|
||||
if kind == BUSY_KIND {
|
||||
(BUSY_STATUS, Json(value))
|
||||
} else if kind == "bridge.error" || kind.ends_with(".error") {
|
||||
let reason = value
|
||||
.get("reason")
|
||||
.and_then(|r| r.as_str())
|
||||
@@ -286,7 +339,9 @@ fn respond_admin(result: Result<Value, RpcError>) -> (StatusCode, Json<Value>) {
|
||||
match result {
|
||||
Ok(value) => {
|
||||
let kind = value.get("kind").and_then(|k| k.as_str()).unwrap_or("");
|
||||
if kind == "admin.error" {
|
||||
if kind == BUSY_KIND {
|
||||
(BUSY_STATUS, Json(value))
|
||||
} else if kind == "admin.error" {
|
||||
let reason = value
|
||||
.get("reason")
|
||||
.and_then(|r| r.as_str())
|
||||
@@ -317,6 +372,70 @@ fn respond_admin(result: Result<Value, RpcError>) -> (StatusCode, Json<Value>) {
|
||||
}
|
||||
}
|
||||
|
||||
/// Like `respond`, but for the event plane: leases and the participation ledger.
|
||||
///
|
||||
/// Two mappings are the point of it existing rather than reusing `respond`.
|
||||
///
|
||||
/// **`lease.drifted` is a 200.** The shard was asked to compare and set, it compared, and it
|
||||
/// refused to overwrite somebody's deliberate change -- that is the mechanism working, not a
|
||||
/// failure, and `cleanup.js` on the website treats `drifted` as a distinct successful outcome
|
||||
/// rather than an error. It is also why this is not a 409: 409 is the protocol-version gate's, and
|
||||
/// a version mismatch and a drifted lease want opposite dispositions from a caller. The same
|
||||
/// argument protocol 6 made for `bridge.busy` being a 425.
|
||||
///
|
||||
/// **The event plane being switched off is a 403**, not the 400 the generic responder's
|
||||
/// reason-sniffing would produce. `Bridge.EventsEnabled` is an operator's deliberate refusal to let
|
||||
/// the website change the world on a schedule, and telling the website it sent a bad request would
|
||||
/// send an administrator hunting a bug in a step that is written correctly.
|
||||
fn respond_event(result: Result<Value, RpcError>) -> (StatusCode, Json<Value>) {
|
||||
match result {
|
||||
Ok(value) => {
|
||||
let kind = value.get("kind").and_then(|k| k.as_str()).unwrap_or("");
|
||||
if kind == BUSY_KIND {
|
||||
(BUSY_STATUS, Json(value))
|
||||
} else if kind.ends_with(".error") {
|
||||
let reason = value
|
||||
.get("reason")
|
||||
.and_then(|r| r.as_str())
|
||||
.unwrap_or("request rejected");
|
||||
let code = if reason.contains("disabled") {
|
||||
StatusCode::FORBIDDEN
|
||||
} else if reason.contains("no lease is offered")
|
||||
|| reason.contains("not counting")
|
||||
// Phase 12b. A grant against a run this shard has never been told to count
|
||||
// is the same shape as an unknown lease key: the caller named something that
|
||||
// does not exist here, which is a 404 and never a retry. It is deliberately
|
||||
// NOT the same as a run whose ledger is open and empty -- that is a 200 with
|
||||
// `granted: 0`, because "nobody came" is a result rather than a mistake.
|
||||
|| reason.contains("no participation ledger")
|
||||
{
|
||||
StatusCode::NOT_FOUND
|
||||
} else if reason.contains("saves at most every") {
|
||||
// A save refused because one just happened is the shard's rate limit, and it
|
||||
// is TRANSIENT in a way nothing else on this plane is: the same request will
|
||||
// succeed once the interval passes. 429 says exactly that, and keeps it out of
|
||||
// the module's permanent-status set so a phase boundary is retried rather than
|
||||
// abandoned.
|
||||
StatusCode::TOO_MANY_REQUESTS
|
||||
} else {
|
||||
StatusCode::BAD_REQUEST
|
||||
};
|
||||
(code, Json(value))
|
||||
} else {
|
||||
(StatusCode::OK, Json(value))
|
||||
}
|
||||
}
|
||||
Err(RpcError::NoShard) => (
|
||||
StatusCode::SERVICE_UNAVAILABLE,
|
||||
Json(json!({"error": "shard not connected"})),
|
||||
),
|
||||
Err(RpcError::Timeout) => (
|
||||
StatusCode::GATEWAY_TIMEOUT,
|
||||
Json(json!({"error": "shard did not reply in time"})),
|
||||
),
|
||||
}
|
||||
}
|
||||
|
||||
/// Like `respond`, but for the account-provisioning plane. Maps an `account.error` reply to a
|
||||
/// status by its reason: a name clash is a 409, the per-IP cap is a 429, a disabled/protected/
|
||||
/// refused action is a 403, an unknown target or "not linked" is a 404, anything else a 400.
|
||||
@@ -324,7 +443,9 @@ fn respond_account(result: Result<Value, RpcError>) -> (StatusCode, Json<Value>)
|
||||
match result {
|
||||
Ok(value) => {
|
||||
let kind = value.get("kind").and_then(|k| k.as_str()).unwrap_or("");
|
||||
if kind == "account.error" {
|
||||
if kind == BUSY_KIND {
|
||||
(BUSY_STATUS, Json(value))
|
||||
} else if kind == "account.error" {
|
||||
let reason = value
|
||||
.get("reason")
|
||||
.and_then(|r| r.as_str())
|
||||
@@ -454,6 +575,16 @@ async fn link_delete(
|
||||
/// Forwards a staff moderation command to the shard, correlated on a fresh reqId. Injects `kind`
|
||||
/// and `reqId`, requiring the caller-supplied `actor` up front (the shard enforces it too). The
|
||||
/// body's remaining fields (account/serial/durationSec/reason/text/hue) pass straight through.
|
||||
///
|
||||
/// **Protocol 6: `idempotencyKey` is one of those remaining fields**, and passing it through is the
|
||||
/// whole of the sidecar's part in the guarantee. It is worth stating rather than leaving to the
|
||||
/// word "remaining", because a later refactor that narrowed this to a known field list would quietly
|
||||
/// turn every retried world write back into a possible duplicate, and nothing here would fail.
|
||||
///
|
||||
/// The key belongs to the CALLER's unit of work — the website's event step — so the sidecar neither
|
||||
/// generates one nor validates it. Note also that `reqId` is regenerated on every call: a retry
|
||||
/// carries the same idempotency key under a NEW correlation id, which is exactly why the shard
|
||||
/// re-stamps a replayed reply rather than echoing the id the first attempt used.
|
||||
async fn admin_call(st: &AppState, kind: &str, body: Value) -> (StatusCode, Json<Value>) {
|
||||
let mut obj = match body {
|
||||
Value::Object(m) => m,
|
||||
@@ -504,6 +635,241 @@ async fn admin_broadcast(State(st): State<AppState>, Json(body): Json<Value>) ->
|
||||
admin_call(&st, "admin.broadcast", body).await
|
||||
}
|
||||
|
||||
// ---- event plane handlers (protocol 6, Phase 11b) ----
|
||||
|
||||
/// Forwards an event-plane command to the shard, correlated on a fresh reqId.
|
||||
///
|
||||
/// Deliberately NOT `admin_call`: that one requires an `actor`, because every verb behind it is a
|
||||
/// staff member pressing a button and the shard's audit trail has to name them. An event verb's
|
||||
/// author is a RUN, which the body already carries as `runId` -- and demanding an actor here would
|
||||
/// have the runner inventing a human name for something no human is doing.
|
||||
///
|
||||
/// Everything else about it is the same, and the `idempotencyKey` passthrough matters for the same
|
||||
/// reason it does there: the key is one of the body's remaining fields, and a refactor that
|
||||
/// narrowed this to a known field list would silently make every retried lease a possible duplicate.
|
||||
async fn event_call(st: &AppState, kind: &str, body: Value) -> (StatusCode, Json<Value>) {
|
||||
let mut obj = match body {
|
||||
Value::Object(m) => m,
|
||||
Value::Null => serde_json::Map::new(),
|
||||
_ => {
|
||||
return (
|
||||
StatusCode::BAD_REQUEST,
|
||||
Json(json!({"error": "body must be a JSON object"})),
|
||||
)
|
||||
}
|
||||
};
|
||||
|
||||
let req_id = st.rpc.next_req_id();
|
||||
obj.insert("kind".to_string(), json!(kind));
|
||||
obj.insert("reqId".to_string(), json!(req_id));
|
||||
|
||||
respond_event(st.rpc.call(&st.shard, Value::Object(obj), &req_id).await)
|
||||
}
|
||||
|
||||
/// Every lease this shard offers, with what each is worth right now and what is holding it.
|
||||
///
|
||||
/// One read answers both questions the website asks about a lease: `read()` wants the current value
|
||||
/// before it applies anything, and `inForce()` wants to know whether the shard still has a record
|
||||
/// of the hold. Splitting them would be two round trips for one key.
|
||||
///
|
||||
/// **`held` means "the shard still has a record of this lease", not "the value is still
|
||||
/// overridden".** A lease whose deadline has already fired stays listed, with `expired: true`,
|
||||
/// until teardown collects its verdict -- otherwise a reconcile in that window would report it gone
|
||||
/// and the website would write off a correctly-working backstop as an orphaned resource.
|
||||
/// **`?key=` and `?target=` narrow it to one row, and a targeted key needs them** (protocol 7 part
|
||||
/// b). `Spawner.MaxCount` is one capability over thousands of spawners, so it has no single
|
||||
/// "current" and the catalog walk cannot fill one in -- while the website's `read()` needs exactly
|
||||
/// one value for exactly one target before it applies anything. Naming both answers that.
|
||||
///
|
||||
/// The frame also carries `holds`: every lease this shard is actually holding, whatever key or
|
||||
/// target it is on. A catalog walk can enumerate the KEYS but never the holds on a targeted one --
|
||||
/// there is no list of spawners to walk -- so without it a reconcile after an outage would have no
|
||||
/// way to ask "what are you still holding?".
|
||||
async fn lease_list(State(st): State<AppState>, Query(q): Query<LeaseQuery>) -> impl IntoResponse {
|
||||
let mut body = serde_json::Map::new();
|
||||
if let Some(key) = q.key {
|
||||
body.insert("key".to_string(), json!(key));
|
||||
}
|
||||
if let Some(target) = q.target {
|
||||
body.insert("target".to_string(), json!(target));
|
||||
}
|
||||
let arg = if body.is_empty() {
|
||||
Value::Null
|
||||
} else {
|
||||
Value::Object(body)
|
||||
};
|
||||
event_call(&st, "lease.list", arg).await
|
||||
}
|
||||
|
||||
/// Narrowing for `GET /lease`. Both optional: absent means the whole catalog, as before.
|
||||
#[derive(Debug, Deserialize)]
|
||||
struct LeaseQuery {
|
||||
key: Option<String>,
|
||||
target: Option<String>,
|
||||
}
|
||||
|
||||
/// Body: {"key":"...","value":"...","holdMs":<ms>,"untilMs":<opt>,"runId":<opt>,"idempotencyKey":<opt>}.
|
||||
///
|
||||
/// **`holdMs` is authoritative and `untilMs` is carried for display.** An absolute deadline computed
|
||||
/// on the website and honoured on the shard is a deadline measured against two clocks, and a shard
|
||||
/// running ten minutes fast would restore a ten-minute lease the moment it took it. A duration is
|
||||
/// immune to that; the absolute time is still worth sending so a console can say when the hold ends.
|
||||
///
|
||||
/// Values cross as TEXT whatever the lease's declared type, because JSON would otherwise decide for
|
||||
/// us: `1200` and `1200.0` are one number to a parser and two strings to a compare-and-set.
|
||||
async fn lease_apply(State(st): State<AppState>, Json(body): Json<Value>) -> impl IntoResponse {
|
||||
event_call(&st, "lease.apply", body).await
|
||||
}
|
||||
|
||||
/// Body: {"key":"...","expected":"...","baseline":"...","idempotencyKey":<opt>}.
|
||||
///
|
||||
/// `expected` is what the event applied and `baseline` is what to put back, both out of the
|
||||
/// website's ledger rather than the shard's memory -- so a release still works after a reconnect,
|
||||
/// and a shard that has forgotten the lease entirely (a restart, which reverts every lease anyway)
|
||||
/// can answer honestly instead of refusing.
|
||||
///
|
||||
/// A mismatch comes back `lease.drifted` with a **200**: see `respond_event`.
|
||||
async fn lease_release(State(st): State<AppState>, Json(body): Json<Value>) -> impl IntoResponse {
|
||||
event_call(&st, "lease.release", body).await
|
||||
}
|
||||
|
||||
/// Body: {"runId":"...","map":"Felucca","x":N,"y":N,"radius":N,"holdMs":<opt>}.
|
||||
///
|
||||
/// Declares where a run happens and starts counting who is there. The area is a map, a point and a
|
||||
/// radius rather than a region name, because protocol 6's own live walk established that the most
|
||||
/// specific region containing an event is routinely anonymous.
|
||||
async fn participation_open(
|
||||
State(st): State<AppState>,
|
||||
Json(body): Json<Value>,
|
||||
) -> impl IntoResponse {
|
||||
event_call(&st, "participation.open", body).await
|
||||
}
|
||||
|
||||
/// Body: {"idempotencyKey":<opt>}. Answers the run's tally, best-effort resolved to accounts.
|
||||
///
|
||||
/// A POST for a read, and the reason is worth keeping: on a well-attended run the shard walks its
|
||||
/// members in chunks across Core ticks rather than handing the whole resolve to one inbound call,
|
||||
/// so the handler completes after its call returned and a repeat arriving in between is answered
|
||||
/// `bridge.busy`. A read that can legitimately be refused as a repeat in flight is not a GET.
|
||||
async fn participation_snapshot(
|
||||
State(st): State<AppState>,
|
||||
Path(run_id): Path<String>,
|
||||
Json(body): Json<Value>,
|
||||
) -> impl IntoResponse {
|
||||
let mut obj = match body {
|
||||
Value::Object(m) => m,
|
||||
_ => serde_json::Map::new(),
|
||||
};
|
||||
obj.insert("runId".to_string(), json!(run_id));
|
||||
event_call(&st, "participation.snapshot", Value::Object(obj)).await
|
||||
}
|
||||
|
||||
/// Body: {"idempotencyKey":<opt>}. Stops counting; the tally stays readable through the shard's
|
||||
/// grace window, because closing an event and collecting its results are two steps and either can
|
||||
/// be retried.
|
||||
async fn participation_close(
|
||||
State(st): State<AppState>,
|
||||
Path(run_id): Path<String>,
|
||||
Json(body): Json<Value>,
|
||||
) -> impl IntoResponse {
|
||||
let mut obj = match body {
|
||||
Value::Object(m) => m,
|
||||
_ => serde_json::Map::new(),
|
||||
};
|
||||
obj.insert("runId".to_string(), json!(run_id));
|
||||
event_call(&st, "participation.close", Value::Object(obj)).await
|
||||
}
|
||||
|
||||
/// Body: {"runId":"...","what":"creature|boss|npc|gate|decor","map":"...","x":N,"y":N,...}.
|
||||
///
|
||||
/// One route for five author-facing verbs. The `what` discriminator is a wire detail: the
|
||||
/// differences between them -- a boss's multipliers, an oracle's lines, a gate's destination and
|
||||
/// `holdMs` -- are fields on one command rather than five commands, so there is one ledger shape,
|
||||
/// one teardown path and one reconcile instead of five near-identical ones in three repos.
|
||||
///
|
||||
/// The shard registers every serial it places against the run and PERSISTS that registry beside
|
||||
/// the world save, which is what makes `world_despawn` below safe: a spawned creature survives a
|
||||
/// restart, so an in-memory registry would leave the website holding serials the shard would not
|
||||
/// vouch for.
|
||||
async fn world_spawn(State(st): State<AppState>, Json(body): Json<Value>) -> impl IntoResponse {
|
||||
event_call(&st, "world.spawn", body).await
|
||||
}
|
||||
|
||||
/// What the run still owns, and the answer the website's `reconcile()` is built on.
|
||||
///
|
||||
/// A GET, unlike `participation_snapshot`: it carries no idempotency key and the shard answers it
|
||||
/// in one pass, pruning rows whose object the world has already lost as it walks. Anything not
|
||||
/// listed is gone -- which is the shape core wants, because it takes a row out of its ledger only
|
||||
/// on an explicit reply and this is that reply.
|
||||
async fn world_owned(State(st): State<AppState>, Path(run_id): Path<String>) -> impl IntoResponse {
|
||||
event_call(&st, "world.owned", json!({ "runId": run_id })).await
|
||||
}
|
||||
|
||||
/// Body: {"serials":[...]} -- or no serials at all, which means everything the run owns and is the
|
||||
/// call teardown actually makes.
|
||||
///
|
||||
/// Three answers, and the split is why the shard keeps a registry at all. `removed` was found and
|
||||
/// deleted; `gone` was owned but already absent, which is what happens when a player kills an event
|
||||
/// creature and is a SUCCESS; `refused` was never this run's to delete, and is the only answer here
|
||||
/// that means somebody asked for something they should not have.
|
||||
async fn world_despawn(
|
||||
State(st): State<AppState>,
|
||||
Path(run_id): Path<String>,
|
||||
Json(body): Json<Value>,
|
||||
) -> impl IntoResponse {
|
||||
let mut obj = match body {
|
||||
Value::Object(m) => m,
|
||||
_ => serde_json::Map::new(),
|
||||
};
|
||||
obj.insert("runId".to_string(), json!(run_id));
|
||||
event_call(&st, "world.despawn", Value::Object(obj)).await
|
||||
}
|
||||
|
||||
// ---- the one-shots (protocol 7 part b) ----
|
||||
|
||||
/// What this shard is willing to grant, and the bounds it will grant within.
|
||||
///
|
||||
/// A read, so the website's option source offers what this shard will actually build. The module
|
||||
/// holds the same list, which is two copies of a short allowlist on purpose and exactly how the
|
||||
/// lease bounds are already carried: the module's copy is what makes a bad value a refusal on a
|
||||
/// form, and this one is what is true when the website is wrong.
|
||||
async fn item_catalog(State(st): State<AppState>) -> impl IntoResponse {
|
||||
event_call(&st, "item.catalog", Value::Null).await
|
||||
}
|
||||
|
||||
/// Body: {"runId":"...","item":"gold","amount":N,"hue":<opt>,"name":<opt>,"where":<opt>,"idempotencyKey":<opt>}.
|
||||
///
|
||||
/// **The recipients are not in the body, and that is the design.** The shard already holds the
|
||||
/// run's participation ledger (protocol 6 part b), keyed by the same character serials the
|
||||
/// website's `member_key` holds, so the grant names a run and the shard resolves who was there.
|
||||
/// Sending a list would mean the same list crossing the wire twice with a window in which the two
|
||||
/// disagree -- and it would have needed a core surface handing a module core's own participants.
|
||||
///
|
||||
/// A run with no ledger open is a 404, not an empty success: "nobody came" and "you never told me
|
||||
/// to count" are different facts, and only the first is a result a run should record.
|
||||
///
|
||||
/// **Retryable, and protocol 6 is why.** `EVENTS.md` §G called a grant un-retryable because a lost
|
||||
/// acknowledgement and a grant that never applied looked the same -- exactly the argument that made
|
||||
/// `uo.broadcast` answer `retry: false` in Phase 9. An `idempotencyKey` closes that: a repeat is
|
||||
/// answered by the original reply, so a retried grant cannot be one winner receiving two.
|
||||
async fn item_grant(State(st): State<AppState>, Json(body): Json<Value>) -> impl IntoResponse {
|
||||
event_call(&st, "item.grant", body).await
|
||||
}
|
||||
|
||||
/// Body: {"idempotencyKey":<opt>}. Starts a world save, useful as a phase boundary.
|
||||
///
|
||||
/// The reply says the save was STARTED and nothing more. What actually happened rides
|
||||
/// `world.save.before` / `world.save.after`, which have been on the event stream since protocol 2 --
|
||||
/// so this route asserts nothing it cannot know, and a caller that needs the completion watches the
|
||||
/// stream it is already connected to.
|
||||
///
|
||||
/// **A save too soon after the last one is refused, not queued**, and the shard counts ServUO's own
|
||||
/// autosave as the last one. A save stops the world; a queued one would land at a moment nobody
|
||||
/// chose, in the middle of whatever the next step is doing.
|
||||
async fn world_save(State(st): State<AppState>, Json(body): Json<Value>) -> impl IntoResponse {
|
||||
event_call(&st, "world.save", body).await
|
||||
}
|
||||
|
||||
// ---- help-page queue handlers ----
|
||||
|
||||
/// The open help-page queue, correlated on reqId. Returns a pages.list.
|
||||
@@ -978,3 +1344,272 @@ async fn ws_client(mut socket: WebSocket, state: AppState) {
|
||||
|
||||
info!("ws client disconnected");
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
fn reply(kind: &str) -> Result<Value, RpcError> {
|
||||
Ok(json!({"t": 1, "kind": kind, "reqId": "r-9"}))
|
||||
}
|
||||
|
||||
/// Protocol 6. Every responder must recognise `bridge.busy`, because every write plane can be
|
||||
/// retried: the staff plane, the account plane and the plain command plane all reach handlers
|
||||
/// that a keyed retry can arrive at. A responder that missed it would return 200 with a body
|
||||
/// saying nothing happened, which is the worst of the three possible answers.
|
||||
#[test]
|
||||
fn busy_maps_to_425_on_every_plane() {
|
||||
assert_eq!(respond(reply("bridge.busy")).0, StatusCode::TOO_EARLY);
|
||||
assert_eq!(respond_admin(reply("bridge.busy")).0, StatusCode::TOO_EARLY);
|
||||
assert_eq!(
|
||||
respond_account(reply("bridge.busy")).0,
|
||||
StatusCode::TOO_EARLY
|
||||
);
|
||||
assert_eq!(respond_event(reply("bridge.busy")).0, StatusCode::TOO_EARLY);
|
||||
}
|
||||
|
||||
/// The event plane is the FIRST place `bridge.busy` is reachable on a live shard rather than
|
||||
/// only in a unit test: `participation.snapshot` walks a well-attended run's members across
|
||||
/// Core ticks, so it completes after its inbound call returned and a repeat can genuinely land
|
||||
/// mid-flight. 11a built the door and had nothing to walk through it.
|
||||
#[test]
|
||||
fn a_drifted_lease_is_a_200_not_a_409() {
|
||||
let value =
|
||||
json!({"kind": "lease.drifted", "key": "PlayerCaps.SkillCap", "current": "1300"});
|
||||
let (status, body) = respond_event(Ok(value));
|
||||
|
||||
// The shard was asked to compare and set, it compared, and it declined to overwrite
|
||||
// somebody's deliberate change. That is the mechanism working; the website records
|
||||
// `drifted` as a distinct successful outcome rather than an error.
|
||||
assert_eq!(status, StatusCode::OK);
|
||||
assert_eq!(body.0.get("current").and_then(|v| v.as_str()), Some("1300"));
|
||||
|
||||
// And explicitly not the version gate's status, for the reason 425 is not either: a
|
||||
// mismatched deployment and a moved value want opposite dispositions from a caller.
|
||||
assert_ne!(status, StatusCode::CONFLICT);
|
||||
}
|
||||
|
||||
/// The event plane being switched off is an operator's refusal, not a malformed request. A 400
|
||||
/// would send an administrator hunting a bug in a step that is written correctly.
|
||||
#[test]
|
||||
fn the_event_gate_being_off_is_a_403() {
|
||||
assert_eq!(
|
||||
respond_event(Ok(json!({
|
||||
"kind": "lease.error",
|
||||
"reason": "the event plane is disabled on this shard (Bridge.EventsEnabled)"
|
||||
})))
|
||||
.0,
|
||||
StatusCode::FORBIDDEN
|
||||
);
|
||||
assert_eq!(
|
||||
respond_event(Ok(json!({
|
||||
"kind": "participation.error",
|
||||
"reason": "the event plane is disabled on this shard (Bridge.EventsEnabled)"
|
||||
})))
|
||||
.0,
|
||||
StatusCode::FORBIDDEN
|
||||
);
|
||||
}
|
||||
|
||||
/// Protocol 7's world verbs go through the same responder, and this pins the two mappings
|
||||
/// they depend on rather than trusting that the reason-sniffing above keeps covering a kind
|
||||
/// it was written before.
|
||||
///
|
||||
/// A CEILING refusal is a 400 on purpose. It is permanent -- retrying "you asked for 80
|
||||
/// creatures and this shard places 30" gets the same answer forever -- and it is the module's
|
||||
/// `PERMANENT_STATUSES` that has to see it as such, so classifying it as anything retryable
|
||||
/// would put a run in a loop against a limit that will never move.
|
||||
#[test]
|
||||
fn a_world_refusal_is_a_400_and_the_gate_is_still_a_403() {
|
||||
assert_eq!(
|
||||
respond_event(Ok(json!({
|
||||
"kind": "world.error",
|
||||
"action": "spawn",
|
||||
"reason": "this shard places 1 to 30 of 'creature' at a time, and 80 was asked for"
|
||||
})))
|
||||
.0,
|
||||
StatusCode::BAD_REQUEST
|
||||
);
|
||||
assert_eq!(
|
||||
respond_event(Ok(json!({
|
||||
"kind": "world.error",
|
||||
"action": "spawn",
|
||||
"reason": "the event plane is disabled on this shard (Bridge.EventsEnabled)"
|
||||
})))
|
||||
.0,
|
||||
StatusCode::FORBIDDEN
|
||||
);
|
||||
}
|
||||
|
||||
/// A run the shard has no registry rows for answers with an EMPTY hand, not a 404, and the
|
||||
/// distinction is load-bearing for reconcile.
|
||||
///
|
||||
/// "This run owns nothing" and "I have never heard of this run" are the same fact once the
|
||||
/// registry is the only record of ownership, and they stay the same fact across a restart:
|
||||
/// the registry is written by `EventSink.WorldSave`, so it and the objects it describes are
|
||||
/// saved and lost together. A 404 here would make the website treat a run that legitimately
|
||||
/// owns nothing as a shard it could not reach.
|
||||
#[test]
|
||||
fn a_run_owning_nothing_is_an_empty_list_not_a_404() {
|
||||
let (status, body) = respond_event(Ok(json!({
|
||||
"kind": "world.owned.ok",
|
||||
"runId": "77",
|
||||
"owned": [],
|
||||
"pruned": 0
|
||||
})));
|
||||
|
||||
assert_eq!(status, StatusCode::OK);
|
||||
assert_eq!(
|
||||
body.0
|
||||
.get("owned")
|
||||
.and_then(|v| v.as_array())
|
||||
.map(|a| a.len()),
|
||||
Some(0)
|
||||
);
|
||||
}
|
||||
|
||||
/// An unknown lease key and an unknown run are not-founds; anything else the shard refuses is a
|
||||
/// bad request. The catalog is short and a typo in a step is the likely cause of both.
|
||||
#[test]
|
||||
fn unknown_lease_and_run_are_404s() {
|
||||
assert_eq!(
|
||||
respond_event(Ok(json!({
|
||||
"kind": "lease.error",
|
||||
"reason": "no lease is offered for key 'Loot.MaxProps'"
|
||||
})))
|
||||
.0,
|
||||
StatusCode::NOT_FOUND
|
||||
);
|
||||
assert_eq!(
|
||||
respond_event(Ok(json!({
|
||||
"kind": "participation.error",
|
||||
"reason": "this shard is not counting run '42'"
|
||||
})))
|
||||
.0,
|
||||
StatusCode::NOT_FOUND
|
||||
);
|
||||
assert_eq!(
|
||||
respond_event(Ok(json!({
|
||||
"kind": "lease.error",
|
||||
"reason": "a lease needs a positive holdMs"
|
||||
})))
|
||||
.0,
|
||||
StatusCode::BAD_REQUEST
|
||||
);
|
||||
}
|
||||
|
||||
/// A run this shard was never told to count is a 404; a run that WAS counted and had no
|
||||
/// attendees is a 200. Protocol 7 part b.
|
||||
#[test]
|
||||
fn an_uncounted_run_is_a_404_and_an_empty_one_is_not() {
|
||||
assert_eq!(
|
||||
respond_event(Ok(json!({
|
||||
"kind": "oneshot.error",
|
||||
"reason": "run 42 has no participation ledger open on this shard"
|
||||
})))
|
||||
.0,
|
||||
StatusCode::NOT_FOUND
|
||||
);
|
||||
// The distinction the 404 exists to preserve. "Nobody came" is a RESULT -- an event
|
||||
// nobody attended still happened -- and answering it as a failure would have the module
|
||||
// retry a grant against a ledger that will be just as empty next time.
|
||||
assert_eq!(
|
||||
respond_event(Ok(json!({
|
||||
"kind": "item.grant.ok",
|
||||
"runId": "42",
|
||||
"granted": 0,
|
||||
"missed": []
|
||||
})))
|
||||
.0,
|
||||
StatusCode::OK
|
||||
);
|
||||
}
|
||||
|
||||
/// The save rate limit is the one refusal on this plane that the same request will get past
|
||||
/// by waiting, so it is a 429 rather than the 400 every other refusal is.
|
||||
#[test]
|
||||
fn a_save_refused_for_coming_too_soon_is_a_429() {
|
||||
assert_eq!(
|
||||
respond_event(Ok(json!({
|
||||
"kind": "oneshot.error",
|
||||
"reason": "this shard saves at most every 300 seconds, and the last save was 12 seconds ago"
|
||||
})))
|
||||
.0,
|
||||
StatusCode::TOO_MANY_REQUESTS
|
||||
);
|
||||
// And an ordinary refusal on the same plane is still a 400, so the 429 is not swallowing
|
||||
// the class it sits beside: a grant this shard does not offer will never succeed, however
|
||||
// long the caller waits.
|
||||
assert_eq!(
|
||||
respond_event(Ok(json!({
|
||||
"kind": "oneshot.error",
|
||||
"reason": "this shard does not grant 'castle'"
|
||||
})))
|
||||
.0,
|
||||
StatusCode::BAD_REQUEST
|
||||
);
|
||||
// The event gate being off stays a 403 on this plane too -- it is an operator's deliberate
|
||||
// refusal, not a bad request.
|
||||
assert_eq!(
|
||||
respond_event(Ok(json!({
|
||||
"kind": "oneshot.error",
|
||||
"reason": "the event plane is disabled on this shard (Bridge.EventsEnabled)"
|
||||
})))
|
||||
.0,
|
||||
StatusCode::FORBIDDEN
|
||||
);
|
||||
}
|
||||
|
||||
/// A lease taken, a tally answered: an ordinary success carries straight through.
|
||||
#[test]
|
||||
fn event_successes_are_200s() {
|
||||
assert_eq!(respond_event(reply("lease.ok")).0, StatusCode::OK);
|
||||
assert_eq!(respond_event(reply("lease.list.ok")).0, StatusCode::OK);
|
||||
assert_eq!(respond_event(reply("participation.ok")).0, StatusCode::OK);
|
||||
assert_eq!(
|
||||
respond_event(reply("participation.snapshot.ok")).0,
|
||||
StatusCode::OK
|
||||
);
|
||||
}
|
||||
|
||||
/// 425 must not collide with the protocol-version gate's 409: a mismatch is a deployment fault
|
||||
/// nobody should retry, a busy shard is a retry that will succeed. Same-status would make the
|
||||
/// two readable only by inspecting the body.
|
||||
#[test]
|
||||
fn busy_is_not_the_version_gates_status() {
|
||||
assert_ne!(BUSY_STATUS, StatusCode::CONFLICT);
|
||||
}
|
||||
|
||||
/// A replayed reply is an ordinary success. The shard marks it `replayed: true` for the log, and
|
||||
/// the caller must be able to treat it exactly as it would have treated the answer it lost.
|
||||
#[test]
|
||||
fn a_replayed_reply_is_still_a_200() {
|
||||
let value = json!({"t": 1, "kind": "admin.ok", "reqId": "r-9", "replayed": true});
|
||||
let (status, body) = respond_admin(Ok(value));
|
||||
assert_eq!(status, StatusCode::OK);
|
||||
assert_eq!(body.0.get("replayed").and_then(|v| v.as_bool()), Some(true));
|
||||
}
|
||||
|
||||
/// The error mapping the busy arm is threaded in front of must be untouched by it.
|
||||
#[test]
|
||||
fn errors_still_map_as_before() {
|
||||
assert_eq!(
|
||||
respond(Ok(
|
||||
json!({"kind": "bridge.error", "reason": "unknown account"})
|
||||
))
|
||||
.0,
|
||||
StatusCode::NOT_FOUND
|
||||
);
|
||||
assert_eq!(
|
||||
respond_admin(Ok(json!({"kind": "admin.error", "reason": "protected"}))).0,
|
||||
StatusCode::FORBIDDEN
|
||||
);
|
||||
assert_eq!(
|
||||
respond_account(Ok(
|
||||
json!({"kind": "account.error", "reason": "already exists"})
|
||||
))
|
||||
.0,
|
||||
StatusCode::CONFLICT
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
239
sidecar/src/windows.rs
Normal file
239
sidecar/src/windows.rs
Normal file
@@ -0,0 +1,239 @@
|
||||
//! Windows startup and shutdown: the SCM handshake.
|
||||
//!
|
||||
//! Unlike systemd, the Windows Service Control Manager cannot supervise an arbitrary console
|
||||
//! program. A binary registered with `sc.exe create` has ~30 seconds to call
|
||||
//! `StartServiceCtrlDispatcher` and connect back to the SCM; one that never does is killed with
|
||||
//! **error 1053, "the service did not respond to the start request in a timely fashion"** — even
|
||||
//! though the process itself started perfectly and is sitting there serving traffic. That is the
|
||||
//! entire reason this module exists.
|
||||
//!
|
||||
//! ## One binary, two ways in
|
||||
//!
|
||||
//! The dispatcher is tried first and *failing is expected*: when the process was started from a
|
||||
//! shell rather than by the SCM, the connect fails with `ERROR_FAILED_SERVICE_CONTROLLER_CONNECT`
|
||||
//! (1063), and that — and only that — falls through to a normal foreground run. So
|
||||
//! `uo-link-sidecar.exe --config ...` stays an ordinary console app you can Ctrl-C, `cargo run`
|
||||
//! still works, and the same binary can be registered as a service with no `--service` flag for an
|
||||
//! operator to forget. Any other dispatcher error is a real failure and is reported.
|
||||
//!
|
||||
//! ## Logging goes to a file, because a service has no stdout
|
||||
//!
|
||||
//! Under the SCM there is no console attached, so the normal stdout subscriber writes into the
|
||||
//! void. In service mode the sidecar logs to a daily-rolled file next to its config instead
|
||||
//! (`uo-link-sidecar.YYYY-MM-DD.log`, seven kept). A service whose start fails leaves a reason
|
||||
//! behind rather than only an SCM error code.
|
||||
|
||||
use std::ffi::OsString;
|
||||
use std::path::PathBuf;
|
||||
use std::sync::Arc;
|
||||
use std::sync::OnceLock;
|
||||
use std::time::Duration;
|
||||
|
||||
use tokio::sync::Notify;
|
||||
use tracing_subscriber::EnvFilter;
|
||||
use windows_service::service::{
|
||||
ServiceControl, ServiceControlAccept, ServiceExitCode, ServiceState, ServiceStatus, ServiceType,
|
||||
};
|
||||
use windows_service::service_control_handler::{self, ServiceControlHandlerResult};
|
||||
use windows_service::{define_windows_service, service_dispatcher};
|
||||
|
||||
/// Must match the name the installer registers (`installer/src/service.rs::WINDOWS_SERVICE`). For
|
||||
/// an own-process service the SCM ignores it, but a mismatch would be a trap for whoever converts
|
||||
/// this to a shared-process service later.
|
||||
pub const SERVICE_NAME: &str = "RunicGatewayLink";
|
||||
|
||||
const SERVICE_TYPE: ServiceType = ServiceType::OWN_PROCESS;
|
||||
|
||||
/// `ERROR_FAILED_SERVICE_CONTROLLER_CONNECT` — "this process was not started by the SCM", which is
|
||||
/// the normal answer when a human runs the binary.
|
||||
const ERROR_FAILED_SERVICE_CONTROLLER_CONNECT: i32 = 1063;
|
||||
|
||||
/// `service_main` is called through an `extern "system"` trampoline and so can capture nothing.
|
||||
/// The parsed `--config` is handed over here instead of being re-parsed, so the service and a
|
||||
/// console run resolve their configuration through exactly the same code path.
|
||||
static CONFIG_PATH: OnceLock<Option<String>> = OnceLock::new();
|
||||
|
||||
pub fn run(config_path: Option<&str>) -> anyhow::Result<()> {
|
||||
let _ = CONFIG_PATH.set(config_path.map(str::to_string));
|
||||
|
||||
match service_dispatcher::start(SERVICE_NAME, ffi_service_main) {
|
||||
Ok(()) => Ok(()),
|
||||
// Not started by the SCM: this is a foreground run, which is not an error.
|
||||
Err(windows_service::Error::Winapi(e))
|
||||
if e.raw_os_error() == Some(ERROR_FAILED_SERVICE_CONTROLLER_CONNECT) =>
|
||||
{
|
||||
console_run(config_path)
|
||||
}
|
||||
Err(e) => Err(anyhow::Error::new(e)
|
||||
.context("could not connect to the Windows service control manager")),
|
||||
}
|
||||
}
|
||||
|
||||
/// A normal foreground run: stdout logging, Ctrl-C to stop.
|
||||
fn console_run(config_path: Option<&str>) -> anyhow::Result<()> {
|
||||
crate::init_console_tracing();
|
||||
tokio::runtime::Runtime::new()?.block_on(crate::app::run(config_path, || {}, async {
|
||||
let _ = tokio::signal::ctrl_c().await;
|
||||
}))
|
||||
}
|
||||
|
||||
define_windows_service!(ffi_service_main, service_main);
|
||||
|
||||
fn service_main(_arguments: Vec<OsString>) {
|
||||
// Arguments are deliberately ignored: for an own-process service the `binPath=` arguments
|
||||
// arrive on the process command line and have already been parsed in `main`. What lands here
|
||||
// is whatever was typed after `sc start`, which nothing in this deployment uses.
|
||||
if let Err(e) = serve() {
|
||||
// Nowhere left to report to but the log: the status handle is gone or was never obtained.
|
||||
tracing::error!(error = %e, "service exited with an error");
|
||||
}
|
||||
}
|
||||
|
||||
fn serve() -> anyhow::Result<()> {
|
||||
let config_path = CONFIG_PATH.get().cloned().flatten();
|
||||
// Held for the life of the service: dropping the guard stops the background log writer.
|
||||
let _log_guard = init_service_tracing(config_path.as_deref());
|
||||
|
||||
// The SCM calls the control handler on its own thread, so the stop signal crosses a thread
|
||||
// boundary into the async world. `notify_one` stores a permit if nothing is waiting yet, so a
|
||||
// stop that arrives during startup is not lost.
|
||||
let stop = Arc::new(Notify::new());
|
||||
let handler_stop = stop.clone();
|
||||
let status_handle =
|
||||
service_control_handler::register(SERVICE_NAME, move |control| match control {
|
||||
ServiceControl::Interrogate => ServiceControlHandlerResult::NoError,
|
||||
ServiceControl::Stop | ServiceControl::Shutdown => {
|
||||
handler_stop.notify_one();
|
||||
ServiceControlHandlerResult::NoError
|
||||
}
|
||||
_ => ServiceControlHandlerResult::NotImplemented,
|
||||
})?;
|
||||
|
||||
// Registering the handler is the handshake 1053 was about. Everything after this point gets to
|
||||
// take as long as it credibly needs, as long as the state keeps being reported.
|
||||
status_handle.set_service_status(ServiceStatus {
|
||||
service_type: SERVICE_TYPE,
|
||||
current_state: ServiceState::StartPending,
|
||||
controls_accepted: ServiceControlAccept::empty(),
|
||||
exit_code: ServiceExitCode::Win32(0),
|
||||
checkpoint: 0,
|
||||
wait_hint: Duration::from_secs(30),
|
||||
process_id: None,
|
||||
})?;
|
||||
|
||||
let ready_handle = status_handle;
|
||||
let result = tokio::runtime::Runtime::new()?.block_on(crate::app::run(
|
||||
config_path.as_deref(),
|
||||
// Reported only once the shard port is bound and the store is open, so a bad config or a
|
||||
// taken port fails the *start* instead of flapping Running → Stopped a moment later.
|
||||
move || {
|
||||
let _ = ready_handle.set_service_status(ServiceStatus {
|
||||
service_type: SERVICE_TYPE,
|
||||
current_state: ServiceState::Running,
|
||||
controls_accepted: ServiceControlAccept::STOP | ServiceControlAccept::SHUTDOWN,
|
||||
exit_code: ServiceExitCode::Win32(0),
|
||||
checkpoint: 0,
|
||||
wait_hint: Duration::default(),
|
||||
process_id: None,
|
||||
});
|
||||
},
|
||||
async move { stop.notified().await },
|
||||
));
|
||||
|
||||
// A failed run must leave a nonzero SERVICE_EXIT_CODE behind: `sc query` reporting STOPPED with
|
||||
// exit code 0 is what made the original failure look like a clean stop.
|
||||
let exit_code = match &result {
|
||||
Ok(()) => ServiceExitCode::Win32(0),
|
||||
Err(e) => {
|
||||
tracing::error!(error = %e, "sidecar failed");
|
||||
ServiceExitCode::ServiceSpecific(1)
|
||||
}
|
||||
};
|
||||
status_handle.set_service_status(ServiceStatus {
|
||||
service_type: SERVICE_TYPE,
|
||||
current_state: ServiceState::Stopped,
|
||||
controls_accepted: ServiceControlAccept::empty(),
|
||||
exit_code,
|
||||
checkpoint: 0,
|
||||
wait_hint: Duration::default(),
|
||||
process_id: None,
|
||||
})?;
|
||||
|
||||
result
|
||||
}
|
||||
|
||||
/// Where the service writes its log: beside the config it was pointed at, which is the directory
|
||||
/// the installer already provisions and grants the service account write access to.
|
||||
fn log_dir(config_path: Option<&str>) -> PathBuf {
|
||||
if let Some(parent) = config_path
|
||||
.map(PathBuf::from)
|
||||
.as_deref()
|
||||
.and_then(|p| p.parent())
|
||||
.filter(|p| !p.as_os_str().is_empty())
|
||||
{
|
||||
return parent.to_path_buf();
|
||||
}
|
||||
match std::env::var_os("ProgramData") {
|
||||
Some(program_data) => PathBuf::from(program_data).join("RunicGateway"),
|
||||
None => std::env::temp_dir(),
|
||||
}
|
||||
}
|
||||
|
||||
/// Returns `None` if the log file could not be opened — a service that cannot write a log is still
|
||||
/// a service worth running, and the SCM start must not fail over it.
|
||||
fn init_service_tracing(
|
||||
config_path: Option<&str>,
|
||||
) -> Option<tracing_appender::non_blocking::WorkerGuard> {
|
||||
let appender = tracing_appender::rolling::Builder::new()
|
||||
.rotation(tracing_appender::rolling::Rotation::DAILY)
|
||||
.filename_prefix("uo-link-sidecar")
|
||||
.filename_suffix("log")
|
||||
.max_log_files(7)
|
||||
.build(log_dir(config_path))
|
||||
.ok()?;
|
||||
|
||||
let (writer, guard) = tracing_appender::non_blocking(appender);
|
||||
tracing_subscriber::fmt()
|
||||
.with_env_filter(
|
||||
EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info")),
|
||||
)
|
||||
.with_ansi(false) // a log file is not a terminal
|
||||
.with_writer(writer)
|
||||
.init();
|
||||
Some(guard)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn log_dir_follows_the_config_file() {
|
||||
assert_eq!(
|
||||
log_dir(Some(r"C:\ProgramData\RunicGateway\sidecar.toml")),
|
||||
PathBuf::from(r"C:\ProgramData\RunicGateway")
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_bare_filename_does_not_become_the_filesystem_root() {
|
||||
// `--config sidecar.toml` has a parent of "", which as a path means the root of the current
|
||||
// drive — somewhere a service account cannot write. Fall back instead.
|
||||
let dir = log_dir(Some("sidecar.toml"));
|
||||
assert_ne!(dir, PathBuf::from(""));
|
||||
assert!(dir.is_absolute(), "{}", dir.display());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn no_config_falls_back_to_program_data() {
|
||||
let dir = log_dir(None);
|
||||
assert!(dir.is_absolute(), "{}", dir.display());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn service_name_matches_the_installer() {
|
||||
// installer/src/service.rs::WINDOWS_SERVICE. Kept as a literal on both sides — the two
|
||||
// repos are released independently and do not share a crate.
|
||||
assert_eq!(SERVICE_NAME, "RunicGatewayLink");
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user