Merge pull request 'feat(sidecar): protocol 8 — the asset plane (Asset Bridge cutover, 1 of 5)' (#44) from edge into main

Reviewed-on: #44
Reviewed-by: Colby Whitlock <whitlocktech@gmail.com>

.gitea/workflows/release.yml

@@ -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.
@@ -342,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" \
+          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 "$(jq -n --arg tag "$TAG" --arg body "$BODY" \
+              -d "${PAYLOAD}" || echo 000)"
-                   '{tag_name:$tag, name:$tag, body:$body, draft:false, prerelease:false}')" \
+
-            | jq -r '.id')"
+            if [ "$HTTP" = "201" ] || [ "$HTTP" = "200" ]; then
+              REL_ID="$(jq -r '.id' /tmp/rel.json)"
+              break
+            fi
+
+            echo "::warning::POST /releases attempt ${attempt} returned HTTP ${HTTP}"
+            echo "--- response body ---"
+            cat /tmp/rel.json || true
+            echo
+            echo "---------------------"
+
+            case "$HTTP" in
+              4*) echo "::error::HTTP ${HTTP} is a client error - not retrying."; exit 1 ;;
+            esac
+
+            if [ "$attempt" = 5 ]; then
+              echo "::error::POST /releases still failing after 5 attempts. Tag ${TAG} is pushed but has no release."
+              echo "::error::Re-run this workflow - the plan step detects the orphan tag and republishes it."
+              exit 1
+            fi
+            sleep $(( attempt * 5 ))
+          done
+
+          if [ -z "$REL_ID" ] || [ "$REL_ID" = "null" ]; then
+            echo "::error::Release created but no id came back; refusing to upload assets blind."
+            exit 1
+          fi
           echo "Created release ${TAG} (id=${REL_ID})"
 
           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
 

sidecar/src/main.rs

@@ -52,7 +52,79 @@ use tracing_subscriber::EnvFilter;
 /// kinds are new, `GET /guilds` grows a `roster` key, and nothing existing changed shape. This is the
 /// first bump that also needed a **store migration** (`guilds.members`), because it is the first to
 /// add a column to a table that already exists rather than a whole new table; see `store::migrate`.
-pub const PROTOCOL_VERSION: u32 = 4;
+///
+/// v5 (Protocol 5): three enrichments that are additive in the same way again, bumped together
+/// rather than one at a time because a protocol bump is not cheap here — it costs a sidecar
+/// release, a republished bundle and an operator update on every shard, so a field left out costs
+/// a whole second round of that rather than a follow-up commit. They are:
+///
+///   * `house.decay` gains `ownerName` and a decay SCHEDULE — `nextStage`, `decayPeriodSec`,
+///     `dynamicDecay`, and `estimatedCollapse` only where it is exactly knowable (at IDOC under
+///     dynamic decay; at any stage under static decay, which has no randomness to wait out).
+///   * `vendor.listing` gains `ownerAcct` — without which the frame names an owner nobody can
+///     resolve to a person — and a `fees` object carrying the charge, the funds, the pay interval
+///     and the resolved `dismissalAt`.
+///   * `account.login.result` is a NEW kind: the verdict of a login, which the pre-existing
+///     `account.login.attempt` structurally cannot carry (its EventSink fires before the auth
+///     decision is made).
+///
+/// **No store migration this time**, unlike v4. Every frame is persisted whole and the board tables
+/// index only the columns they already had, so the new fields ride inside the stored JSON and the
+/// new kind lands in `events` like any other. That is the dumb-forwarder property doing its job:
+/// the sidecar defines no schema for a frame's contents and so needs no change when they grow.
+///
+/// v6 (Protocol 6): the first bump that is about a GUARANTEE rather than about data, and the first
+/// the sidecar mostly gets for free. Two things:
+///
+///   * **`idempotencyKey` on inbound commands.** A command that carries one is executed by the shard
+///     at most once; a repeat is answered with the original reply rather than re-run. That is what
+///     makes a world-writing verb retryable at all — until now a lost acknowledgement was
+///     indistinguishable from a command that never applied, so the website had to declare every
+///     write un-retryable and accept losing one rather than risk doubling it. The sidecar's part is
+///     to CARRY the key (it rides in the command body, which every write endpoint already passes
+///     through verbatim) and to understand the one new answer the shard can now give: `bridge.busy`,
+///     meaning a command under that key is still in flight. See `web::respond`.
+///   * **`champ.boss.killed` is a new kind**: a champion's defeat, with the damage table only the
+///     shard ever sees. It was previously inferable from `champ.update` going `bossUp` true then
+///     false alongside a nearby `mob.killed`, which is fragile and says nothing about who did the
+///     work. It lands in `events` and on the feed like any other kind, with no code here at all —
+///     the dumb-forwarder property again.
+///
+/// **No store migration.** Nothing gains a column; the new kind is persisted whole like every other.
+///
+/// # Protocol 8 — the Asset Bridge (docs/link/v8.md)
+///
+/// The shard starts sending the operator's own **client assets** over this link: the cliloc string
+/// table, creature and item art, player models. The point is that an operator stops having to run
+/// a GUI converter on a desktop to make their site render a bestiary, and the shard is the only
+/// host that already has the client files — a ServUO server cannot boot without them.
+///
+/// Phase 1 is the transport, and the sidecar's share of it is three things:
+///
+///   * **A new command family, `assets.*`, forwarded verbatim** like every other. The first of them
+///     is `assets.sources` — stage 1 of the import gate: what the client files currently are, and
+///     what version of the shard's extractor would read them. No pixels cross on this call.
+///   * **An inbound line cap** — [`shard::MAX_INBOUND_LINE_BYTES`]. This is the one change that is
+///     not additive. `read_line` had no bound at all, which was survivable while the shard had no
+///     reason to send a large line; protocol 8 gives it one deliberately, and an unbounded read
+///     facing a component that now sends megabytes is a memory-exhaustion shape we would be
+///     inventing ourselves.
+///   * **Nothing else.** Assets ride the request/reply path, so `rpc::try_route` consumes them
+///     before `app.rs` can persist them to the store and fan them out to every WebSocket
+///     subscriber — which is what keeps a 512 KiB reply from being written to SQLite and broadcast
+///     to every connected client. The dumb-forwarder property is doing real work here: the sidecar
+///     does not know what an asset is, and must not learn.
+///
+/// Phase 2 adds the first family that actually carries content: **`cliloc.table`**, served at
+/// `GET /cliloc`. It pages — the shard cuts at a byte budget and the caller echoes a cursor back —
+/// and the sidecar forwards it without keeping any of it, which matters more here than usual: the
+/// payload is five megabytes of EA's strings out of the operator's own client, and the one copy of
+/// it that should exist is the one the website imports. That phase also gave `assets.error` a
+/// `code`, so the status a refusal maps to stops depending on the wording of a human-facing
+/// sentence (see [`web::asset_error_status`]).
+///
+/// **No store migration**, again: nothing on this plane is an event, so nothing is persisted.
+pub const PROTOCOL_VERSION: u32 = 8;
 
 // Not `#[tokio::main]`: on Windows the SCM dispatcher takes over this thread and starts the runtime
 // itself, on its own thread, once the service actually begins. The runtime is built by whichever

sidecar/src/shard.rs

@@ -7,15 +7,130 @@
 //! Framing is newline-delimited JSON, bidirectional: the shard sends events, we send commands. We
 //! accept one shard connection at a time and re-accept when it drops (the shard reconnects on its
 //! own, with a bounded backoff).
+//!
+//! Inbound lines are **capped** (see [`MAX_INBOUND_LINE_BYTES`]). Until protocol 8 they were not:
+//! `read_line` will buffer a line of any length, which was survivable only because the shard had
+//! never had a reason to send a large one. The Asset Bridge gives it one, so the gap had to close
+//! before it became a memory-exhaustion shape we invented ourselves.
 
 use std::sync::Arc;
 
 use serde_json::Value;
-use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
+use tokio::io::{AsyncBufRead, AsyncBufReadExt, AsyncWriteExt, BufReader};
 use tokio::net::TcpListener;
 use tokio::sync::{mpsc, Mutex};
 use tracing::{info, warn};
 
+/// The longest line the sidecar will accept from the shard, in bytes.
+///
+/// Set above the largest legal batch rather than at it: the shard cuts a batch when the next item
+/// would take it past `Bridge.AssetBatchBytes` (512 KiB), and always admits the first item of a
+/// page even when that item alone is bigger than the budget — so one page can legitimately
+/// overshoot by one item. Doubling the budget to get this cap is what makes that overshoot safe
+/// instead of a dropped reply.
+///
+/// Over-long lines are **discarded, not buffered**, and the connection stays up. That is the same
+/// disposition `BridgeLink.cs` has always had for its own 1 MiB inbound cap in the other
+/// direction, and it is the right one here: a single malformed frame is not a reason to tear down
+/// a link that live events are flowing over. The dropped reply simply times out and is
+/// re-requested, which is safe because everything on the asset plane is idempotent.
+pub const MAX_INBOUND_LINE_BYTES: usize = 1024 * 1024;
+
+/// What one read off the shard socket produced.
+#[derive(Debug)]
+enum Line {
+    /// A complete line, within the cap.
+    Complete(String),
+    /// A line that ran past the cap. Carries how many bytes were thrown away, for the log.
+    TooLong(usize),
+    /// The shard closed the connection.
+    Eof,
+}
+
+/// A cancel-safe, capped, newline-delimited reader.
+///
+/// Every piece of state that must survive a partial read lives here rather than in a local,
+/// because this is polled inside a `tokio::select!`: the loop below drops the future whenever a
+/// command wins the race, and a `discarding` flag or a half-filled buffer held in a local would be
+/// lost with it. Losing the buffer corrupts the *next* line; losing `discarding` turns the tail of
+/// an over-long line into a line of its own. Both are silent.
+///
+/// The only await point is `fill_buf`, and nothing is consumed until after it returns, so a
+/// cancellation between the two can lose at most the wakeup.
+#[derive(Default)]
+struct LineReader {
+    buf: Vec<u8>,
+    discarding: bool,
+    discarded: usize,
+}
+
+impl LineReader {
+    async fn next<R: AsyncBufRead + Unpin>(&mut self, reader: &mut R) -> std::io::Result<Line> {
+        loop {
+            let consumed;
+            let outcome;
+
+            {
+                let available = reader.fill_buf().await?;
+
+                if available.is_empty() {
+                    return Ok(Line::Eof);
+                }
+
+                match available.iter().position(|&b| b == b'\n') {
+                    Some(at) => {
+                        consumed = at + 1;
+
+                        if self.discarding {
+                            // The tail of a line we already gave up on. Swallow it, terminator
+                            // included, and report the size once.
+                            self.discarded += at;
+                            let total = self.discarded;
+                            self.discarding = false;
+                            self.discarded = 0;
+                            outcome = Some(Line::TooLong(total));
+                        } else if self.buf.len() + at > MAX_INBOUND_LINE_BYTES {
+                            // The cap is reached only now, on the chunk that also holds the
+                            // terminator — so there is nothing left to discard.
+                            let total = self.buf.len() + at;
+                            self.buf.clear();
+                            outcome = Some(Line::TooLong(total));
+                        } else {
+                            self.buf.extend_from_slice(&available[..at]);
+                            let line = String::from_utf8_lossy(&self.buf).into_owned();
+                            self.buf.clear();
+                            outcome = Some(Line::Complete(line));
+                        }
+                    }
+                    None => {
+                        consumed = available.len();
+
+                        if self.discarding {
+                            self.discarded += consumed;
+                        } else if self.buf.len() + consumed > MAX_INBOUND_LINE_BYTES {
+                            // Refuse rather than buffer: this is the whole point of the cap.
+                            // Everything up to the next newline is now dropped on the floor.
+                            self.discarded = self.buf.len() + consumed;
+                            self.buf.clear();
+                            self.discarding = true;
+                        } else {
+                            self.buf.extend_from_slice(available);
+                        }
+
+                        outcome = None;
+                    }
+                }
+            }
+
+            reader.consume(consumed);
+
+            if let Some(line) = outcome {
+                return Ok(line);
+            }
+        }
+    }
+}
+
 /// An event line received from the shard, parsed. `kind` is lifted out for routing.
 #[derive(Debug, Clone)]
 pub struct ShardEvent {
@@ -112,16 +227,25 @@ async fn handle_connection(
     handle.set(Some(cmd_tx)).await;
 
     let mut reader = BufReader::new(read_half);
-    let mut line = String::new();
+    let mut lines = LineReader::default();
 
     loop {
         tokio::select! {
             // Inbound: a line from the shard.
-            result = reader.read_line(&mut line) => {
+            result = lines.next(&mut reader) => {
-                let n = result?;
+                match result? {
-                if n == 0 {
+                    Line::Eof => return Ok(()), // clean EOF: shard closed
-                    return Ok(()); // clean EOF: shard closed
+                    Line::TooLong(bytes) => {
+                        // Deliberately not a disconnect. See MAX_INBOUND_LINE_BYTES: a reply lost
+                        // this way times out on the caller's side and is re-requested, and tearing
+                        // the link down would take the live event feed with it.
+                        warn!(
+                            bytes,
+                            cap = MAX_INBOUND_LINE_BYTES,
+                            "inbound line over the cap; discarded"
+                        );
                     }
+                    Line::Complete(line) => {
                         let trimmed = line.trim_end();
                         if !trimmed.is_empty() {
                             match serde_json::from_str::<Value>(trimmed) {
@@ -136,7 +260,8 @@ async fn handle_connection(
                                 Err(e) => warn!(error = %e, line = %trimmed, "unparseable event"),
                             }
                         }
-                line.clear();
+                    }
+                }
             }
             // Outbound: a command to write to the shard.
             cmd = cmd_rx.recv() => {
@@ -152,3 +277,108 @@ async fn handle_connection(
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// Drives `LineReader` over a byte slice, returning every outcome up to EOF.
+    async fn read_all(input: &[u8]) -> Vec<Line> {
+        let mut reader = BufReader::with_capacity(64, input);
+        let mut lines = LineReader::default();
+        let mut out = Vec::new();
+
+        loop {
+            match lines.next(&mut reader).await.unwrap() {
+                Line::Eof => break,
+                other => out.push(other),
+            }
+        }
+
+        out
+    }
+
+    fn complete(lines: &[Line]) -> Vec<&str> {
+        lines
+            .iter()
+            .filter_map(|l| match l {
+                Line::Complete(s) => Some(s.as_str()),
+                _ => None,
+            })
+            .collect()
+    }
+
+    #[tokio::test]
+    async fn splits_on_newlines() {
+        let lines = read_all(b"{\"a\":1}\n{\"b\":2}\n").await;
+        assert_eq!(complete(&lines), vec!["{\"a\":1}", "{\"b\":2}"]);
+    }
+
+    /// The reader's buffer is 64 bytes here, so every one of these lines spans several
+    /// `fill_buf` chunks. Reassembly across chunks is the thing `read_line` did for us.
+    #[tokio::test]
+    async fn reassembles_across_chunks() {
+        let long = "x".repeat(500);
+        let input = format!("{}\n{}\n", long, long);
+        let lines = read_all(input.as_bytes()).await;
+
+        assert_eq!(complete(&lines), vec![long.as_str(), long.as_str()]);
+    }
+
+    /// The cap itself. The over-long line must be reported and thrown away, and — the part that
+    /// actually matters — the line *after* it must still arrive intact. A reader that lost its
+    /// `discarding` flag would emit the tail of the oversized line as a line of its own.
+    #[tokio::test]
+    async fn refuses_an_over_long_line_and_recovers() {
+        let mut input = Vec::new();
+        input.extend_from_slice(&b"a".repeat(MAX_INBOUND_LINE_BYTES + 10));
+        input.push(b'\n');
+        input.extend_from_slice(b"{\"kind\":\"pong\"}\n");
+
+        let lines = read_all(&input).await;
+
+        assert_eq!(lines.len(), 2);
+        assert!(
+            matches!(lines[0], Line::TooLong(n) if n >= MAX_INBOUND_LINE_BYTES),
+            "expected TooLong, got {:?}",
+            lines[0]
+        );
+        assert_eq!(complete(&lines), vec!["{\"kind\":\"pong\"}"]);
+    }
+
+    /// A line of exactly the cap is legal; one byte more is not. Checking both sides is what says
+    /// the comparison is `>` rather than `>=`, which would silently cost a byte of the budget.
+    #[tokio::test]
+    async fn the_cap_is_inclusive() {
+        let at_cap = "b".repeat(MAX_INBOUND_LINE_BYTES);
+        let lines = read_all(format!("{}\n", at_cap).as_bytes()).await;
+        assert_eq!(complete(&lines).len(), 1);
+
+        let over = "b".repeat(MAX_INBOUND_LINE_BYTES + 1);
+        let lines = read_all(format!("{}\n", over).as_bytes()).await;
+        assert!(complete(&lines).is_empty());
+        assert!(matches!(lines[0], Line::TooLong(_)));
+    }
+
+    /// An over-long line whose terminator lands in the very chunk that crosses the cap: the
+    /// reader must not leave itself in `discarding` and eat the next line as well.
+    #[tokio::test]
+    async fn over_long_line_terminating_in_the_crossing_chunk() {
+        let mut input = Vec::new();
+        input.extend_from_slice(&b"c".repeat(MAX_INBOUND_LINE_BYTES + 1));
+        input.extend_from_slice(b"\n{\"kind\":\"pong\"}\n");
+
+        let lines = read_all(&input).await;
+
+        assert!(matches!(lines[0], Line::TooLong(_)));
+        assert_eq!(complete(&lines), vec!["{\"kind\":\"pong\"}"]);
+    }
+
+    /// A partial line at EOF is dropped rather than delivered half-parsed. The shard reconnects
+    /// and re-sends; half a JSON object is not something to hand to the event fan-out.
+    #[tokio::test]
+    async fn trailing_partial_line_at_eof_is_dropped() {
+        let lines = read_all(b"{\"a\":1}\n{\"b\":").await;
+        assert_eq!(complete(&lines), vec!["{\"a\":1}"]);
+    }
+}