RunicGateway/link
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 12 Wiki Activity

feat(sidecar): start as a real Windows service
All checks were successful
PR Checks / rust-gates (pull_request) Successful in 2m22s
Details

Browse Source 
`sc.exe start RunicGatewayLink` failed with 1053 on every Windows install:
"a timeout was reached (30000 milliseconds) while waiting for the service to
connect", with SERVICE_EXIT_CODE 0. Nothing had crashed. The sidecar was a
plain console program, and the Windows service control manager only supervises
a process that calls StartServiceCtrlDispatcher and identifies itself within
~30 seconds.

The installer's design assumed symmetry with systemd, which supervises any
foreground process. Windows has no equivalent: it is a service-aware binary or
a shim, and a shim was already rejected as a third binary to keep current.

Split the entry point so the platform only owns starting and stopping:

  systemd --> main --> unix::run ---------------+
                                                +--> app::run
  SCM ------> main --> windows::run --> ServiceMain
                                    \-> console fallback

- app.rs is the whole sidecar, unchanged and shared. No #[cfg] on the data path.
- windows.rs speaks the SCM handshake. The dispatcher is tried first and failing
  is expected: ERROR_FAILED_SERVICE_CONTROLLER_CONNECT (1063) means "not started
  by the SCM" and falls through to a normal foreground run, so one binary does
  both with no --service flag to forget.
- Running is reported only once the shard port is bound and the store is open, so
  a bad config fails the start instead of flapping Running -> Stopped, and a
  failed run leaves a nonzero SERVICE_EXIT_CODE instead of the misleading 0.
- A service has no stdout, so service mode logs to uo-link-sidecar.log.<date>
  beside its config, rolled daily, seven kept.
- unix.rs additionally handles SIGTERM, which is what systemctl stop sends and
  which previously took the default disposition mid-write.

The Windows crates are declared under [target.'cfg(windows)'.dependencies].
Verified: a Linux build in rust:1-slim-bookworm succeeds and resolves neither
windows-service nor tracing-appender.

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
claude
2026-08-07 13:35:52 -05:00
parent 915f0296a9
commit 96af2afa68
7 changed files with 768 additions and 300 deletions
Download Patch File Download Diff File Expand all files Collapse all files

334
sidecar/src/main.rs
View File

@@ -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::<shard::ShardEvent>();
let handle = shard::serve(&cfg.shard.bind, event_tx).await?;
// Live feed: every shard event fans out to all connected website WebSocket clients.
let (bcast_tx, _) = broadcast::channel::<String>(1024);
// Request/reply correlation for REST queries.
let rpc = rpc::Rpc::new();
// Durable store: event history, economy series, cached profiles, link map.
let store = store::Store::open(&cfg.store.path).await?;
// Health/observability state.
let started = Instant::now();
let last_event = Arc::new(AtomicI64::new(0));
// Website-facing HTTP server.
let web_state = web::AppState {
events: bcast_tx.clone(),
shard: handle.clone(),
rpc: rpc.clone(),
store: store.clone(),
token: Arc::new(cfg.web.auth_token.clone()),
started,
last_event: last_event.clone(),
};
let web_bind = cfg.web.bind.clone();
tokio::spawn(async move {
if let Err(e) = web::serve(&web_bind, web_state).await {
tracing::error!(error = %e, "web server exited");
}
});
// Event loop: a line that correlates to a pending REST call is a reply — route it to the
// waiting caller and stop. Everything else is a live event: log it, persist it, broadcast it.
let feed_tx = bcast_tx.clone();
let route_rpc = rpc.clone();
let event_store = store.clone();
let last_event_ts = last_event.clone();
let replay_handle = handle.clone(); // re-push external news to the shard on (re)connect
let mut total: u64 = 0;
tokio::spawn(async move {
while let Some(ev) = event_rx.recv().await {
// Any line from the shard — including pong heartbeats — is a sign of life.
last_event_ts.store(now_ms(), std::sync::atomic::Ordering::Relaxed);
if route_rpc.try_route(&ev.value).await {
continue; // consumed as a reply
}
total += 1;
match ev.kind.as_str() {
"server.hello" | "mob.login" | "mob.logout" | "player.death" | "vendor.sale"
| "house.decay" | "link.request" => {
info!(kind = %ev.kind, n = total, "{}", ev.value);
}
_ => tracing::debug!(kind = %ev.kind, n = total, "{}", ev.value),
}
// Persist, then broadcast. `pong` and `ws.hello` are ephemeral chatter, not history.
if ev.kind != "pong" {
let t = ev
.value
.get("t")
.and_then(|v| v.as_i64())
.unwrap_or_else(now_ms);
let text = ev.value.to_string();
if let Err(e) = event_store.insert_event(t, &ev.kind, &text).await {
tracing::warn!(error = %e, "failed to persist event");
}
// The champ board is a live projection: champ.update folds in the latest state (one
// row per spawn), champ.remove drops a spawn that despawned or was slain.
match ev.kind.as_str() {
"champ.update" => {
if let Some(serial) = ev.value.get("serial").and_then(|s| s.as_str()) {
if let Err(e) = event_store
.upsert_champ(
serial,
ev.value.get("status").and_then(|s| s.as_str()),
ev.value.get("name").and_then(|n| n.as_str()),
&text,
t,
)
.await
{
tracing::warn!(error = %e, "failed to upsert champ board");
}
}
}
"champ.remove" => {
if let Some(serial) = ev.value.get("serial").and_then(|s| s.as_str()) {
if let Err(e) = event_store.delete_champ(serial).await {
tracing::warn!(error = %e, "failed to remove champ board row");
}
}
}
// Guild board (Protocol 2.0): guild.update folds in the latest roster (one row
// per guild id); guild.remove drops a disbanded guild.
"guild.update" => {
if let Some(id) = ev.value.get("id").and_then(|v| v.as_i64()) {
if let Err(e) = event_store
.upsert_guild(
id,
ev.value.get("name").and_then(|n| n.as_str()),
&text,
t,
)
.await
{
tracing::warn!(error = %e, "failed to upsert guild board");
}
}
}
"guild.remove" => {
if let Some(id) = ev.value.get("id").and_then(|v| v.as_i64()) {
if let Err(e) = event_store.delete_guild(id).await {
tracing::warn!(error = %e, "failed to remove guild board row");
}
}
}
// Governor board (Protocol 2.0): city.update folds in each city's latest
// governance state (one row per city).
"city.update" => {
if let Some(city) = ev.value.get("city").and_then(|c| c.as_str()) {
if let Err(e) = event_store.upsert_governor(city, &text, t).await {
tracing::warn!(error = %e, "failed to upsert governor board");
}
}
}
// House registry (Protocol 2.0): house.update folds in each house's latest state
// (one row per serial); house.remove drops a demolished/traded house.
"house.update" => {
if let Some(serial) = ev.value.get("serial").and_then(|s| s.as_str()) {
if let Err(e) = event_store
.upsert_house(
serial,
ev.value.get("name").and_then(|n| n.as_str()),
&text,
t,
)
.await
{
tracing::warn!(error = %e, "failed to upsert house registry");
}
}
}
"house.remove" => {
if let Some(serial) = ev.value.get("serial").and_then(|s| s.as_str()) {
if let Err(e) = event_store.delete_house(serial).await {
tracing::warn!(error = %e, "failed to remove house registry row");
}
}
}
// Points/loyalty boards (Protocol 3.0): one row per point system, keyed by
// the shard's own PointsType name. The plugin only emits a system whose top N
// actually moved, so this is a sparse stream of overwrites — and there is no
// `points.remove` to handle, because the shard's set of systems is fixed at
// startup and cannot shrink.
"points.board" => {
if let Some(system) = ev.value.get("system").and_then(|s| s.as_str()) {
if let Err(e) = event_store
.upsert_points_board(
system,
ev.value.get("nameString").and_then(|n| n.as_str()),
&text,
t,
)
.await
{
tracing::warn!(error = %e, "failed to upsert points board");
}
}
}
// Player-vendor market index (Protocol 3.0). Each frame is authoritative for
// one vendor — the shard's round-robin sweep only emits a shop whose contents,
// prices or location actually moved — so this is a whole-row overwrite.
//
// Unlike the boards above there IS a remove: a vendor is dismissed, expires, or
// its owner switches off the in-game Vendor Search flag, and any of those must
// take the shop off the site. The last of the three is a privacy control, so
// dropping the row promptly is the point rather than housekeeping.
"vendor.listing" => {
if let Some(serial) = ev.value.get("serial").and_then(|s| s.as_str()) {
let loc = ev.value.get("location");
let field = |k: &str| loc.and_then(|l| l.get(k));
if let Err(e) = event_store
.upsert_vendor(
serial,
ev.value.get("shopName").and_then(|v| v.as_str()),
ev.value.get("ownerName").and_then(|v| v.as_str()),
field("map").and_then(|v| v.as_str()),
field("x").and_then(|v| v.as_i64()),
field("y").and_then(|v| v.as_i64()),
field("region").and_then(|v| v.as_str()),
ev.value.get("count").and_then(|v| v.as_i64()),
&text,
t,
)
.await
{
tracing::warn!(error = %e, "failed to upsert vendor listing");
}
}
}
"vendor.listing.remove" => {
if let Some(serial) = ev.value.get("serial").and_then(|s| s.as_str()) {
if let Err(e) = event_store.delete_vendor(serial).await {
tracing::warn!(error = %e, "failed to remove vendor listing");
}
}
}
// Shard ruleset (Protocol 3.0): a singleton projection. The shard re-emits
// world.ruleset on every connect, so this row is simply overwritten; `rev`
// lets a reader tell a re-send from an actual config change.
"world.ruleset" => {
if let Err(e) = event_store
.upsert_ruleset(ev.value.get("rev").and_then(|r| r.as_str()), &text, t)
.await
{
tracing::warn!(error = %e, "failed to upsert ruleset");
}
}
_ => {}
}
}
// On a shard (re)connect, re-push the stored external news: the shard rebuilds
// TownCryerSystem.NewsEntries from scratch each boot and does not persist ours. Replay
// with announce=false so a restart does not re-proclaim every article at once. news.add
// is idempotent by id, so replaying to a still-populated shard is harmless.
if ev.kind == "server.hello" {
match event_store.news_all().await {
Ok(items) => {
for mut item in items {
if let Some(obj) = item.as_object_mut() {
obj.insert("announce".to_string(), serde_json::json!(false));
}
if !replay_handle.send(item.to_string()).await {
break; // shard went away mid-replay
}
}
}
Err(e) => tracing::warn!(error = %e, "news replay: could not read stored news"),
}
}
let _ = feed_tx.send(ev.value.to_string());
}
});
// Heartbeat to the shard, exercising the command path.
let ping_handle = handle.clone();
tokio::spawn(async move {
loop {
tokio::time::sleep(std::time::Duration::from_secs(15)).await;
if ping_handle.is_connected().await {
let _ = ping_handle
.send(r#"{"kind":"ping","id":"sidecar-heartbeat"}"#.to_string())
.await;
}
}
});
tokio::signal::ctrl_c().await?;
info!("shutting down");
Ok(())
#[cfg(unix)]
return unix::run(args.config.as_deref());
}
fn now_ms() -> i64 {
use std::time::{SystemTime, UNIX_EPOCH};
SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.as_millis() as i64)
.unwrap_or(0)
}
fn init_tracing() {
/// Logging for a foreground run: human-readable, on stdout.
pub fn init_console_tracing() {
tracing_subscriber::fmt()
.with_env_filter(
EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info")),