From b46005b333b930870f8787e781529a16497ca521 Mon Sep 17 00:00:00 2001 From: colby Date: Fri, 10 Jul 2026 15:58:04 -0500 Subject: [PATCH] Sidecar: shard link (Rust) First cut of the Rust sidecar, in sidecar/. It is the TCP listener the shard dials out to; that asymmetry is what keeps the game unreachable from the website. shard.rs: serve() binds 127.0.0.1:7788 and accepts shard connections in a loop, re-accepting on disconnect. Each connection splits read/write: the reader parses newline-JSON into ShardEvent { kind, value } and forwards over an mpsc; the writer drains a command mpsc. ShardHandle::send posts to whichever shard is connected and drops with a warning if none is -- a website query during an outage should fail fast, not queue; live events that must survive an outage are buffered by the shard. main.rs wires it up, logs events by kind, and runs a 15s heartbeat ping to exercise the command path. Later phases fan events out to a WebSocket broadcaster and SQLite, and turn REST calls into shard commands. Verified against the live shard: the sidecar received the shard's server.hello (parsed, fields intact), round-tripped its heartbeat ping -> pong, and after a sidecar restart the shard reconnected on its own and re-sent hello. Tokio + serde; axum/sqlx/tungstenite come with the WS and REST phases. Co-Authored-By: Claude Opus 4.8 --- README.md | 1 + sidecar/.gitignore | 1 + sidecar/Cargo.lock | 383 +++++++++++++++++++++++++++++++++++++++++++ sidecar/Cargo.toml | 16 ++ sidecar/README.md | 37 +++++ sidecar/src/main.rs | 67 ++++++++ sidecar/src/shard.rs | 154 +++++++++++++++++ 7 files changed, 659 insertions(+) create mode 100644 sidecar/.gitignore create mode 100644 sidecar/Cargo.lock create mode 100644 sidecar/Cargo.toml create mode 100644 sidecar/README.md create mode 100644 sidecar/src/main.rs create mode 100644 sidecar/src/shard.rs diff --git a/README.md b/README.md index 1f34aad..1c2866f 100644 --- a/README.md +++ b/README.md @@ -15,6 +15,7 @@ The shard never speaks WebSocket. Every world read happens on the Core thread; t |------|------| | `overlay/` | Mirrors the ServUO server root. Everything here — and **only** this — copies over an install. | | `patches/` | Unified diffs against stock ServUO for files we must modify rather than add. | +| `sidecar/` | The Rust sidecar: terminates the loopback link to the shard, exposes WS + REST to the website. See `sidecar/README.md`. | | `tools/` | Never deployed. Test scaffolding and anything else that must not reach a server. | | `docs/PLAN.md` | Implementation plan, measured performance budget, and the full data catalog. | | `docs/RESEARCH.md` | Original source-level research. Partly superseded — see the corrections table in `PLAN.md` §8. | diff --git a/sidecar/.gitignore b/sidecar/.gitignore new file mode 100644 index 0000000..ea8c4bf --- /dev/null +++ b/sidecar/.gitignore @@ -0,0 +1 @@ +/target diff --git a/sidecar/Cargo.lock b/sidecar/Cargo.lock new file mode 100644 index 0000000..7207aec --- /dev/null +++ b/sidecar/Cargo.lock @@ -0,0 +1,383 @@ +# This file is automatically @generated by Cargo. +# It is not intended for manual editing. +version = 4 + +[[package]] +name = "aho-corasick" +version = "1.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ddd31a130427c27518df266943a5308ed92d4b226cc639f5a8f1002816174301" +dependencies = [ + "memchr", +] + +[[package]] +name = "anyhow" +version = "1.0.103" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2a4385e2e34eb35d6b3efe798b9eb88096925d87726c0798709bf56d9ed84af3" + +[[package]] +name = "bytes" +version = "1.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04" + +[[package]] +name = "cfg-if" +version = "1.0.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801" + +[[package]] +name = "errno" +version = "0.3.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" +dependencies = [ + "libc", + "windows-sys", +] + +[[package]] +name = "itoa" +version = "1.0.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682" + +[[package]] +name = "lazy_static" +version = "1.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bbd2bcb4c963f2ddae06a2efc7e9f3591312473c50c6685e1f298068316e66fe" + +[[package]] +name = "libc" +version = "0.2.186" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "68ab91017fe16c622486840e4c83c9a37afeff978bd239b5293d61ece587de66" + +[[package]] +name = "log" +version = "0.4.33" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0ceec5bc11778974d1bcb055b18002eba7f4b3518b6a0081b3af5f21666da9ad" + +[[package]] +name = "matchers" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d1525a2a28c7f4fa0fc98bb91ae755d1e2d1505079e05539e35bc876b5d65ae9" +dependencies = [ + "regex-automata", +] + +[[package]] +name = "memchr" +version = "2.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98" + +[[package]] +name = "mio" +version = "1.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "02bd0af71c67b473010cbbc60715ee815645a4dc942899111f494b4b737d6fda" +dependencies = [ + "libc", + "wasi", + "windows-sys", +] + +[[package]] +name = "nu-ansi-term" +version = "0.50.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" +dependencies = [ + "windows-sys", +] + +[[package]] +name = "once_cell" +version = "1.21.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50" + +[[package]] +name = "pin-project-lite" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd" + +[[package]] +name = "proc-macro2" +version = "1.0.106" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8fd00f0bb2e90d81d1044c2b32617f68fcb9fa3bb7640c23e9c748e53fb30934" +dependencies = [ + "unicode-ident", +] + +[[package]] +name = "quote" +version = "1.0.46" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dfbc457d0c7a0759a614551b11a6409e5951f6c7537be1f1b7682b9ae9230368" +dependencies = [ + "proc-macro2", +] + +[[package]] +name = "regex-automata" +version = "0.4.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1f388202e4b80542a0921078cc23b6333bcf1409c1e3f86404cae4766a6131db" +dependencies = [ + "aho-corasick", + "memchr", + "regex-syntax", +] + +[[package]] +name = "regex-syntax" +version = "0.8.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d6f6ff9a378485b298a5286656da665ba74413d36db0979633275d2e708145d4" + +[[package]] +name = "serde" +version = "1.0.228" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9a8e94ea7f378bd32cbbd37198a4a91436180c5bb472411e48b5ec2e2124ae9e" +dependencies = [ + "serde_core", + "serde_derive", +] + +[[package]] +name = "serde_core" +version = "1.0.228" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "41d385c7d4ca58e59fc732af25c3983b67ac852c1a25000afe1175de458b67ad" +dependencies = [ + "serde_derive", +] + +[[package]] +name = "serde_derive" +version = "1.0.228" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d540f220d3187173da220f885ab66608367b6574e925011a9353e4badda91d79" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "serde_json" +version = "1.0.150" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e8014e44b4736ed0538adeecded0fce2a272f22dc9578a7eb6b2d9993c74cfb9" +dependencies = [ + "itoa", + "memchr", + "serde", + "serde_core", + "zmij", +] + +[[package]] +name = "sharded-slab" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f40ca3c46823713e0d4209592e8d6e826aa57e928f09752619fc696c499637f6" +dependencies = [ + "lazy_static", +] + +[[package]] +name = "signal-hook-registry" +version = "1.4.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c4db69cba1110affc0e9f7bcd48bbf87b3f4fc7c61fc9155afd4c469eb3d6c1b" +dependencies = [ + "errno", + "libc", +] + +[[package]] +name = "smallvec" +version = "1.15.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8ed6a63f02c8539c91a8685a86f4099661ba3da017932f6ebbea6de3f0fa7c90" + +[[package]] +name = "socket2" +version = "0.6.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "52d1cfed4120b4d927bf7c0f86d2087a4a7d6027c906d9f9d525a80573b9be51" +dependencies = [ + "libc", + "windows-sys", +] + +[[package]] +name = "syn" +version = "2.0.118" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1b9ae57f904213ebb649ce6895b8a66c66f0203b9319718f69a5612a065b1422" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + +[[package]] +name = "thread_local" +version = "1.1.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f60246a4944f24f6e018aa17cdeffb7818b76356965d03b07d6a9886e8962185" +dependencies = [ + "cfg-if", +] + +[[package]] +name = "tokio" +version = "1.52.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8fc7f01b389ac15039e4dc9531aa973a135d7a4135281b12d7c1bc79fd57fffe" +dependencies = [ + "bytes", + "libc", + "mio", + "pin-project-lite", + "signal-hook-registry", + "socket2", + "tokio-macros", + "windows-sys", +] + +[[package]] +name = "tokio-macros" +version = "2.7.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "385a6cb71ab9ab790c5fe8d67f1645e6c450a7ce006a33de03daa956cf70a496" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "tracing" +version = "0.1.44" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "63e71662fa4b2a2c3a26f570f037eb95bb1f85397f3cd8076caed2f026a6d100" +dependencies = [ + "pin-project-lite", + "tracing-attributes", + "tracing-core", +] + +[[package]] +name = "tracing-attributes" +version = "0.1.31" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7490cfa5ec963746568740651ac6781f701c9c5ea257c58e057f3ba8cf69e8da" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "tracing-core" +version = "0.1.36" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "db97caf9d906fbde555dd62fa95ddba9eecfd14cb388e4f491a66d74cd5fb79a" +dependencies = [ + "once_cell", + "valuable", +] + +[[package]] +name = "tracing-log" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ee855f1f400bd0e5c02d150ae5de3840039a3f54b025156404e34c23c03f47c3" +dependencies = [ + "log", + "once_cell", + "tracing-core", +] + +[[package]] +name = "tracing-subscriber" +version = "0.3.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cb7f578e5945fb242538965c2d0b04418d38ec25c79d160cd279bf0731c8d319" +dependencies = [ + "matchers", + "nu-ansi-term", + "once_cell", + "regex-automata", + "sharded-slab", + "smallvec", + "thread_local", + "tracing", + "tracing-core", + "tracing-log", +] + +[[package]] +name = "unicode-ident" +version = "1.0.24" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e6e4313cd5fcd3dad5cafa179702e2b244f760991f45397d14d4ebf38247da75" + +[[package]] +name = "uo-link-sidecar" +version = "0.1.0" +dependencies = [ + "anyhow", + "serde", + "serde_json", + "tokio", + "tracing", + "tracing-subscriber", +] + +[[package]] +name = "valuable" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ba73ea9cf16a25df0c8caa16c51acb937d5712a8429db78a3ee29d5dcacd3a65" + +[[package]] +name = "wasi" +version = "0.11.1+wasi-snapshot-preview1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b" + +[[package]] +name = "windows-link" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5" + +[[package]] +name = "windows-sys" +version = "0.61.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ae137229bcbd6cdf0f7b80a31df61766145077ddf49416a728b02cb3921ff3fc" +dependencies = [ + "windows-link", +] + +[[package]] +name = "zmij" +version = "1.0.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b8848ee67ecc8aedbaf3e4122217aff892639231befc6a1b58d29fff4c2cabaa" diff --git a/sidecar/Cargo.toml b/sidecar/Cargo.toml new file mode 100644 index 0000000..1fc64e1 --- /dev/null +++ b/sidecar/Cargo.toml @@ -0,0 +1,16 @@ +[package] +name = "uo-link-sidecar" +version = "0.1.0" +edition = "2021" +description = "Rust sidecar for the ServUO uo-link bridge: terminates the loopback link to the shard and exposes WebSocket + REST to the website." + +[dependencies] +tokio = { version = "1", features = ["rt-multi-thread", "macros", "net", "io-util", "sync", "time", "signal"] } +serde = { version = "1", features = ["derive"] } +serde_json = "1" +tracing = "0.1" +tracing-subscriber = { version = "0.3", features = ["env-filter"] } +anyhow = "1" + +[profile.release] +opt-level = 2 diff --git a/sidecar/README.md b/sidecar/README.md new file mode 100644 index 0000000..bed50b9 --- /dev/null +++ b/sidecar/README.md @@ -0,0 +1,37 @@ +# uo-link sidecar + +The Rust half of the bridge. It terminates the loopback link to the ServUO shard and (as it grows) exposes WebSocket + REST to the website. + +``` +website ──WS (live feed) / REST (queries)──► sidecar ──loopback TCP 127.0.0.1:7788──► shard + (this) newline-JSON, bidirectional +``` + +The sidecar is the TCP **listener**; the shard dials out to it. That is what keeps the game unreachable from the website — the game exposes no port of its own. See `../docs/PLAN.md` §2. + +## Run + +```bash +cargo run # info logging +RUST_LOG=debug cargo run # see every event, incl. pong heartbeats +``` + +Binds `127.0.0.1:7788` and waits for the shard to connect. Boot the shard (or it will reconnect on its own) and watch `server.hello` arrive. + +## Status + +| Piece | State | +|-------|-------| +| Shard link (`shard.rs`) | **done** — accepts the shard, reads events, sends commands, re-accepts on disconnect. Verified against the live shard: received `server.hello`, round-tripped a `ping`→`pong`, and reconnected after a sidecar restart. | +| WebSocket feed (website ← live events) | not started | +| REST queries (char profile, roster, vendor snapshot, link submit) | not started | +| SQLite persistence (event history, economy, cached profiles, link map) | not started | + +## Design + +- **`shard.rs`** — `serve()` binds the listener and accepts shard connections in a loop. Each connection splits into read/write halves: the read half parses newline-JSON into `ShardEvent { kind, value }` and forwards them; the write half drains an mpsc of command lines. `ShardHandle::send` posts a command to whichever shard is currently connected, and **drops with a warning if none is** — a website query during a shard outage should fail fast and retry, not queue behind a reconnect. Live *events* that must survive an outage are buffered by the shard, not here. +- **`main.rs`** — wires it together and, for now, logs events by kind. Later phases fan `ShardEvent`s out to the WebSocket broadcaster and SQLite, and turn REST calls into shard commands via `ShardHandle`. + +## Wire protocol + +Every line is one JSON object with `t` (epoch ms) and `kind`. The shard→sidecar events and sidecar→shard commands are catalogued in `../docs/PLAN.md` (§5 data catalog, §7 protocol) and were all validated end-to-end while building the plugin. Notable inbound commands the sidecar will issue: `char.request`, `account.roster`, `vendor.snapshot`, `link.confirm`, `towncrier.add`/`remove`, `ping`. diff --git a/sidecar/src/main.rs b/sidecar/src/main.rs new file mode 100644 index 0000000..77ab813 --- /dev/null +++ b/sidecar/src/main.rs @@ -0,0 +1,67 @@ +//! uo-link sidecar. +//! +//! Terminates the loopback link to the ServUO shard and (in later phases) exposes WebSocket + REST +//! to the website. This first cut proves the shard link: it receives real events and can send +//! commands back. + +mod shard; + +use tokio::sync::mpsc; +use tracing::info; +use tracing_subscriber::EnvFilter; + +const SHARD_ADDR: &str = "127.0.0.1:7788"; + +#[tokio::main] +async fn main() -> anyhow::Result<()> { + init_tracing(); + + info!("uo-link sidecar starting"); + + let (event_tx, mut event_rx) = mpsc::unbounded_channel::(); + let handle = shard::serve(SHARD_ADDR, event_tx).await?; + + // Phase 1: log every event and tally by kind. Later phases fan this out to WS + SQLite. + let mut total: u64 = 0; + tokio::spawn(async move { + while let Some(ev) = event_rx.recv().await { + total += 1; + match ev.kind.as_str() { + // These are the anchors worth surfacing at info; the rest are debug. + "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); + } + } + } + }); + + // A tiny demonstration that the command path works: ping the shard once it connects. + 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; + } + } + }); + + // Run until Ctrl-C. + tokio::signal::ctrl_c().await?; + info!("shutting down"); + Ok(()) +} + +fn init_tracing() { + tracing_subscriber::fmt() + .with_env_filter( + EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info")), + ) + .init(); +} diff --git a/sidecar/src/shard.rs b/sidecar/src/shard.rs new file mode 100644 index 0000000..daaa057 --- /dev/null +++ b/sidecar/src/shard.rs @@ -0,0 +1,154 @@ +//! The loopback link to the ServUO shard. +//! +//! The shard is the TCP *client*: it dials out to us. So the sidecar owns the listener, and the +//! shard'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. See docs/PLAN.md §2. +//! +//! 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). + +use std::sync::Arc; + +use serde_json::Value; +use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; +use tokio::net::TcpListener; +use tokio::sync::{mpsc, Mutex}; +use tracing::{info, warn}; + +/// An event line received from the shard, parsed. `kind` is lifted out for routing. +#[derive(Debug, Clone)] +pub struct ShardEvent { + pub kind: String, + pub value: Value, +} + +/// A handle for sending command lines to the shard. Cloneable and cheap. +/// +/// Commands are dropped (with a warning) when no shard is connected, rather than buffered: a +/// website query that arrives during a shard outage should fail fast and be retried, not silently +/// queue behind a reconnect. Live *events* are what must survive an outage, and those the shard +/// buffers on its side. +#[derive(Clone)] +pub struct ShardHandle { + tx: Arc>>>, +} + +impl ShardHandle { + fn new() -> Self { + Self { + tx: Arc::new(Mutex::new(None)), + } + } + + async fn set(&self, sender: Option>) { + *self.tx.lock().await = sender; + } + + /// Send one command line (a complete JSON object, no newline — we add the frame delimiter). + /// Returns false if no shard 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 shard connected"); + false + } + } + } + + pub async fn is_connected(&self) -> bool { + self.tx.lock().await.is_some() + } +} + +/// Binds the loopback listener and accepts shard 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 shard is currently connected. +pub async fn serve( + addr: &str, + event_tx: mpsc::UnboundedSender, +) -> std::io::Result { + let listener = TcpListener::bind(addr).await?; + info!(%addr, "shard link listening"); + + let handle = ShardHandle::new(); + let accept_handle = handle.clone(); + + tokio::spawn(async move { + loop { + match listener.accept().await { + Ok((stream, peer)) => { + info!(%peer, "shard connected"); + if let Err(e) = handle_connection(stream, &event_tx, &accept_handle).await { + warn!(error = %e, "shard connection ended"); + } else { + info!("shard disconnected"); + } + accept_handle.set(None).await; + } + Err(e) => { + warn!(error = %e, "accept failed"); + tokio::time::sleep(std::time::Duration::from_millis(500)).await; + } + } + } + }); + + Ok(handle) +} + +async fn handle_connection( + stream: tokio::net::TcpStream, + event_tx: &mpsc::UnboundedSender, + handle: &ShardHandle, +) -> 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::(); + handle.set(Some(cmd_tx)).await; + + let mut reader = BufReader::new(read_half); + let mut line = String::new(); + + loop { + tokio::select! { + // Inbound: a line from the shard. + result = reader.read_line(&mut line) => { + let n = result?; + if n == 0 { + return Ok(()); // clean EOF: shard closed + } + let trimmed = line.trim_end(); + if !trimmed.is_empty() { + match serde_json::from_str::(trimmed) { + Ok(value) => { + let kind = value + .get("kind") + .and_then(|k| k.as_str()) + .unwrap_or("") + .to_string(); + let _ = event_tx.send(ShardEvent { kind, value }); + } + Err(e) => warn!(error = %e, line = %trimmed, "unparseable event"), + } + } + line.clear(); + } + // Outbound: a command to write to the shard. + 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 + } + } + } + } +}