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

feat(sidecar): protocol 1 — the transport

Browse Source 
The rust-link sidecar: it owns the loopback listener the Oxide bridge plugin
dials into, and serves the website a WebSocket feed plus store-backed reads.

Protocol 1 is deliberately three frames — server.hello, ping/pong, and one
correlated server.status — because phase 1's job is to get every seam working at
once with almost nothing in them.

What is load-bearing rather than incidental:

* The plugin is the TCP client and this process owns the listener, so a Rust
  server opens no extra port. Loopback is the trust boundary on that link and
  there is no token on it; the website-facing surface is the opposite, with auth
  always on and a token generated and persisted on first start.
* Inbound lines are capped at 1 MiB from the start rather than after the first
  large frame arrives. An over-long line is discarded and the connection stays
  up: one malformed frame is not a reason to drop a link live events flow over.
* Store-backed reads answer while the game is off, which is what lets a website
  render a server list during a wipe. /status is the one route that fails when
  the game is down, and /server answers 204 rather than a null when the game has
  never connected -- those are different answers and a client that cannot tell
  them apart renders a server that does not exist.
* The two RPC failures get distinct codes. 503 means the game is down; 504 means
  it is up and did not answer. Different fixes.
* rpc::REPLY_TIMEOUT is a ceiling every later command budget sits under: core
  classifies a budget overrun as retryable unconditionally, so an action whose
  budgetMs does not exceed it can never report retry:false.

One defect found while building, which no unit test would have caught: a
four-connection SQLite pool over :memory: hands out four separate empty
databases, because an in-memory database is per connection. It presents as
'no such table' from a random subset of queries. The pool is now capped at one
connection for an in-memory path, which is the only coherent reading of
:memory: and is what makes it usable at all.

Exercised end to end against a live Rust server: a server.hello travelled game
-> sidecar -> module -> the public website API, and killing this process left
the game untouched.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_016wDDVXWMDz82WqE1i969r4
This commit is contained in:
claude
2026-09-15 19:52:55 -05:00
parent d6a93506e8
commit e2a58f3455
12 changed files with 4651 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

415
sidecar/src/game.rs Normal file
View File

@@ -0,0 +1,415 @@
//! The loopback link to the Rust server's Oxide bridge plugin.
//!
//! The plugin is the TCP *client*: it dials out to us. So the sidecar owns the listener, and the
//! plugin's outbound socket is the only thing that ever connects. This is the whole reason the game
//! is never directly reachable from the website — it exposes no port for us.
//!
//! Framing is newline-delimited JSON, bidirectional: the plugin sends events (`kind`), we send
//! commands (`cmd`). We accept one plugin connection at a time and re-accept when it drops (the
//! plugin reconnects on its own, with a bounded backoff).
//!
//! **Loopback is the trust boundary.** There is no token on this link, exactly as on the ServUO
//! bridge: the plugin and the sidecar share a host, and the address to bind is `127.0.0.1`. Binding
//! `[game].bind` to anything routable puts an unauthenticated command channel on the network, and
//! the operator documentation says so in as many words.
//!
//! Inbound lines are **capped** (see [`MAX_INBOUND_LINE_BYTES`]) from the start rather than after
//! the first large frame arrives: an unbounded `read_line` facing a peer that will one day send a
//! map image is a memory-exhaustion shape we would be inventing ourselves.
use std::sync::Arc;
use serde_json::Value;
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 plugin, in bytes.
///
/// Over-long lines are **discarded, not buffered**, and the connection stays up: a single malformed
/// frame is not a reason to tear down a link that live events are flowing over. A dropped reply
/// simply times out on the caller's side and is re-requested.
pub const MAX_INBOUND_LINE_BYTES: usize = 1024 * 1024;
/// What one read off the plugin 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 plugin 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 plugin, parsed. `kind` is lifted out for routing.
#[derive(Debug, Clone)]
pub struct GameEvent {
pub kind: String,
pub value: Value,
}
/// A handle for sending command lines to the plugin. Cloneable and cheap.
///
/// Commands are dropped (with a warning) when no plugin is connected, rather than buffered: a
/// website query that arrives during an outage should fail fast and be retried, not silently queue
/// behind a reconnect. Live *events* are what must survive an outage, and those the plugin buffers
/// on its side.
#[derive(Clone)]
pub struct GameHandle {
tx: Arc<Mutex<Option<mpsc::UnboundedSender<String>>>>,
}
impl GameHandle {
fn new() -> Self {
Self {
tx: Arc::new(Mutex::new(None)),
}
}
async fn set(&self, sender: Option<mpsc::UnboundedSender<String>>) {
*self.tx.lock().await = sender;
}
/// Send one command line (a complete JSON object, no newline — we add the frame delimiter).
/// Returns false if no plugin is currently connected.
pub async fn send(&self, line: String) -> bool {
let guard = self.tx.lock().await;
match guard.as_ref() {
Some(sender) => sender.send(line).is_ok(),
None => {
warn!("dropping command; no plugin connected");
false
}
}
}
pub async fn is_connected(&self) -> bool {
self.tx.lock().await.is_some()
}
}
/// Binds the loopback listener and accepts plugin connections forever. Each accepted connection
/// runs until it drops, then we loop back to accept the next one. Events are forwarded to
/// `event_tx`; the returned handle sends commands to whichever plugin is currently connected.
///
/// The **bound** address is returned alongside the handle rather than assumed to be the one asked
/// for: `127.0.0.1:0` is a legitimate thing to configure (and what the tests use), and a log line
/// echoing the request rather than the result is the kind that is wrong exactly when it matters.
pub async fn serve(
addr: &str,
event_tx: mpsc::UnboundedSender<GameEvent>,
) -> std::io::Result<(GameHandle, std::net::SocketAddr)> {
let listener = TcpListener::bind(addr).await?;
let bound = listener.local_addr()?;
info!(addr = %bound, "game link listening");
let handle = GameHandle::new();
let accept_handle = handle.clone();
tokio::spawn(async move {
loop {
match listener.accept().await {
Ok((stream, peer)) => {
info!(%peer, "plugin connected");
if let Err(e) = handle_connection(stream, &event_tx, &accept_handle).await {
warn!(error = %e, "plugin connection ended");
} else {
info!("plugin disconnected");
}
accept_handle.set(None).await;
// A disconnect is a fact the website should see without polling, so it rides
// the same channel every other fact does. Nothing persists it.
let _ = event_tx.send(GameEvent {
kind: "link.down".to_string(),
value: serde_json::json!({ "kind": "link.down" }),
});
}
Err(e) => {
warn!(error = %e, "accept failed");
tokio::time::sleep(std::time::Duration::from_millis(500)).await;
}
}
}
});
Ok((handle, bound))
}
async fn handle_connection(
stream: tokio::net::TcpStream,
event_tx: &mpsc::UnboundedSender<GameEvent>,
handle: &GameHandle,
) -> std::io::Result<()> {
stream.set_nodelay(true).ok();
let (read_half, mut write_half) = stream.into_split();
// Install the outbound command channel for this connection.
let (cmd_tx, mut cmd_rx) = mpsc::unbounded_channel::<String>();
handle.set(Some(cmd_tx)).await;
let mut reader = BufReader::new(read_half);
let mut lines = LineReader::default();
loop {
tokio::select! {
// Inbound: a line from the plugin.
result = lines.next(&mut reader) => {
match result? {
Line::Eof => return Ok(()), // clean EOF: plugin closed
Line::TooLong(bytes) => {
// Deliberately not a disconnect. See MAX_INBOUND_LINE_BYTES.
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) {
Ok(value) => {
let kind = value
.get("kind")
.and_then(|k| k.as_str())
.unwrap_or("")
.to_string();
let _ = event_tx.send(GameEvent { kind, value });
}
Err(e) => warn!(error = %e, line = %trimmed, "unparseable event"),
}
}
}
}
}
// Outbound: a command to write to the plugin.
cmd = cmd_rx.recv() => {
match cmd {
Some(mut c) => {
c.push('\n');
write_half.write_all(c.as_bytes()).await?;
write_half.flush().await?;
}
None => return Ok(()), // channel closed
}
}
}
}
}
#[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` would have done 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.
#[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 plugin 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}"]);
}
/// The end-to-end shape of the link, over a real socket: the plugin dials in, sends a hello,
/// and receives a command. This is the criterion phase 1 is judged on, minus the two peers.
#[tokio::test]
async fn a_dialling_plugin_is_accepted_and_can_be_commanded() {
use tokio::io::AsyncReadExt;
let (tx, mut rx) = mpsc::unbounded_channel();
let (handle, addr) = serve("127.0.0.1:0", tx).await.unwrap();
let mut client = tokio::net::TcpStream::connect(addr).await.unwrap();
client
.write_all(b"{\"kind\":\"server.hello\",\"serverId\":\"main\"}\n")
.await
.unwrap();
let ev = rx.recv().await.unwrap();
assert_eq!(ev.kind, "server.hello");
assert_eq!(ev.value["serverId"], "main");
assert!(handle.is_connected().await);
assert!(handle.send("{\"cmd\":\"ping\"}".to_string()).await);
let mut buf = [0u8; 64];
let n = client.read(&mut buf).await.unwrap();
assert_eq!(&buf[..n], b"{\"cmd\":\"ping\"}\n");
}
}