diff --git a/README.md b/README.md index 0418fbc..c9ea8f3 100644 --- a/README.md +++ b/README.md @@ -48,6 +48,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..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. diff --git a/sidecar/Cargo.lock b/sidecar/Cargo.lock index fc6eb27..2a0a796 100644 --- a/sidecar/Cargo.lock +++ b/sidecar/Cargo.lock @@ -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" diff --git a/sidecar/Cargo.toml b/sidecar/Cargo.toml index 45ed6ab..f64eccd 100644 --- a/sidecar/Cargo.toml +++ b/sidecar/Cargo.toml @@ -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 diff --git a/sidecar/src/app.rs b/sidecar/src/app.rs new file mode 100644 index 0000000..05ef060 --- /dev/null +++ b/sidecar/src/app.rs @@ -0,0 +1,329 @@ +//! 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::future::Future; +use std::sync::atomic::AtomicI64; +use std::sync::Arc; +use std::time::Instant; + +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(config_path: Option<&str>, ready: R, shutdown: S) -> anyhow::Result<()> +where + R: FnOnce(), + S: Future, +{ + 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::(); + 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::(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; + } + } + }); + + // 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) +} diff --git a/sidecar/src/main.rs b/sidecar/src/main.rs index bbe8199..efb1ad3 100644 --- a/sidecar/src/main.rs +++ b/sidecar/src/main.rs @@ -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 @@ -32,8 +48,10 @@ use tracing_subscriber::EnvFilter; /// there is deliberately no feature-negotiation array: v3 implies all three kinds. pub const PROTOCOL_VERSION: u32 = 3; -#[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 +84,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::(); - 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::(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")), diff --git a/sidecar/src/unix.rs b/sidecar/src/unix.rs new file mode 100644 index 0000000..42749bf --- /dev/null +++ b/sidecar/src/unix.rs @@ -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"), + } +} diff --git a/sidecar/src/windows.rs b/sidecar/src/windows.rs new file mode 100644 index 0000000..a2628bf --- /dev/null +++ b/sidecar/src/windows.rs @@ -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> = 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) { + // 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 { + 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"); + } +}