From 2305660d5a99130923767c334ace8b5dbbde8220 Mon Sep 17 00:00:00 2001 From: Steve Biedermann Date: Wed, 21 Oct 2026 20:00:05 +0200 Subject: [PATCH] fix: exit shards when simulation threads fail Supervise the shard simulation thread and exit with status 1 when it panics or stops, allowing the container restart policy to recover instead of leaving a live control server with no simulation. Cover startup panics and actor-reply timeouts with subprocess regression tests, and document recovery behavior. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- CHANGELOG.md | 1 + README.md | 2 + src/cluster/shard.rs | 49 +++++++++----- tests/cluster.rs | 148 ++++++++++++++++++++++++++++++++++++++++++- 4 files changed, 182 insertions(+), 18 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 3727325..9ab72c6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -26,6 +26,7 @@ Notable changes to this project are documented here. ### Changed +- Terminate shard processes when their simulation thread fails, so Docker can restart them instead of leaving an unresponsive shard and a repeatedly restarting gateway. - People, rabbits, wolves, and partnerships are Picard actors that own their state; the world only reaches them through messages. Actors are named `person-`, `rabbit-`, `wolf-`, and `partnership-` instead of `entity-`. - Rabbits killed by hunters no longer move or breed later in the same tick. - Update Picard to `f88b9cb`. diff --git a/README.md b/README.md index 654b700..99b7297 100644 --- a/README.md +++ b/README.md @@ -81,6 +81,8 @@ docker compose up -d --remove-orphans Open [http://localhost:7878](http://localhost:7878). The `shard_0_data`, `shard_1_data`, and `shard_2_data` volumes survive container restarts, `just stop`, and `docker compose down`. +If a shard's simulation thread fails, the shard exits with an error so Compose's restart policy can restart it. The gateway also exits when it loses a shard and reconnects on restart. Recovery uses the saved snapshots; if the shards' saved ticks or seeds do not match, the gateway starts a fresh world. + Compose now uses SQLite instead of Postgres. Existing Postgres snapshots are not migrated; the old Postgres volume is left untouched, and the SQLite shards start fresh. To remove all shard volumes and saved simulation data as well: diff --git a/src/cluster/shard.rs b/src/cluster/shard.rs index d9cac97..6558df8 100644 --- a/src/cluster/shard.rs +++ b/src/cluster/shard.rs @@ -56,18 +56,17 @@ pub async fn run(options: ShardOptions) { tokio::spawn(save_snapshots(storage, snapshots)); let (commands, receiver) = mpsc::unbounded_channel(); - { - std::thread::Builder::new() - .name(format!("shard-{}", shard.index)) - .spawn(move || { - let (mut world, restored) = open_world(saved.as_deref(), seed, script, shard); - world - .start_remoting(&cluster_token) - .unwrap_or_else(|error| panic!("failed to start shard remoting: {error}")); - run_world(world, restored, receiver, &saves, &cluster_token); - }) - .expect("failed to start the simulation thread"); - } + let simulation = std::thread::Builder::new() + .name(format!("shard-{}", shard.index)) + .spawn(move || { + let (mut world, restored) = open_world(saved.as_deref(), seed, script, shard); + world + .start_remoting(&cluster_token) + .unwrap_or_else(|error| panic!("failed to start shard remoting: {error}")); + run_world(world, restored, receiver, &saves, &cluster_token); + }) + .expect("failed to start the simulation thread"); + let stopped = tokio::task::spawn_blocking(move || simulation.join()); if attached { std::thread::spawn(|| { let _ = std::io::copy(&mut std::io::stdin(), &mut std::io::sink()); @@ -87,9 +86,29 @@ pub async fn run(options: ShardOptions) { let _ = stdout.flush(); } tracing::info!(shard = shard.index, %address, "shard listening"); - axum::serve(listener, app) - .await - .expect("shard server failed"); + tokio::select! { + result = stopped => { + match result { + Ok(Ok(())) => tracing::error!( + shard = shard.index, + "shard simulation thread stopped unexpectedly; terminating shard" + ), + Ok(Err(_)) => tracing::error!( + shard = shard.index, + "shard simulation thread panicked; terminating shard" + ), + Err(error) => tracing::error!( + shard = shard.index, + %error, + "failed to monitor the shard simulation thread; terminating shard" + ), + } + std::process::exit(1); + } + result = async { axum::serve(listener, app).await } => { + result.expect("shard server failed"); + } + } } /// Restores the shard's saved world, or creates a new one when there is none or it doesn't fit. diff --git a/tests/cluster.rs b/tests/cluster.rs index 275226a..c656255 100644 --- a/tests/cluster.rs +++ b/tests/cluster.rs @@ -6,15 +6,157 @@ use std::net::{TcpListener, TcpStream}; use std::process::{Child, Command, Stdio}; use std::time::{Duration, Instant}; -struct Gateway(Child); +struct SimProcess(Child); -impl Drop for Gateway { +impl Drop for SimProcess { fn drop(&mut self) { let _ = self.0.kill(); let _ = self.0.wait(); } } +impl SimProcess { + fn wait_for_failure(&mut self) -> String { + let deadline = Instant::now() + Duration::from_secs(30); + let status = loop { + if let Some(status) = self.0.try_wait().unwrap() { + break status; + } + assert!( + Instant::now() < deadline, + "the shard stayed alive after its simulation thread panicked" + ); + std::thread::sleep(Duration::from_millis(50)); + }; + let mut stderr = String::new(); + self.0 + .stderr + .take() + .unwrap() + .read_to_string(&mut stderr) + .unwrap(); + assert_eq!(status.code(), Some(1), "{stderr}"); + assert!( + stderr.contains("shard simulation thread panicked; terminating shard"), + "{stderr}" + ); + stderr + } +} + +fn shard_command() -> Command { + let mut command = Command::new(env!("CARGO_BIN_EXE_sim")); + command + .args([ + "shard", + "--index", + "0", + "--count", + "1", + "--database", + "sqlite::memory:", + ]) + .env("SIM_REMOTING_BIND", "127.0.0.1:0") + .env_remove("SIM_REMOTING_ADVERTISE") + .env( + "SIM_CONFIG", + concat!(env!("CARGO_MANIFEST_DIR"), "/config.rn"), + ) + .env("RUST_LOG", "sim=info,picard=off") + .stdout(Stdio::piped()) + .stderr(Stdio::piped()); + command +} + +#[test] +fn shard_exits_when_simulation_startup_panics() { + let mut shard = SimProcess( + shard_command() + .env("SIM_REMOTING_BIND", "invalid-address") + .spawn() + .unwrap(), + ); + let stderr = shard.wait_for_failure(); + assert!( + stderr.contains("failed to start shard remoting"), + "{stderr}" + ); +} + +#[cfg(feature = "console")] +#[test] +fn shard_exits_when_actor_replies_time_out() { + use std::io::{BufRead, BufReader}; + use tungstenite::Message; + + let mut shard = SimProcess(shard_command().spawn().unwrap()); + let stdout = shard.0.stdout.take().unwrap(); + let (listening, address) = std::sync::mpsc::channel(); + let reader = std::thread::spawn(move || { + let mut lines = BufReader::new(stdout).lines(); + for line in lines.by_ref() { + let line = line.unwrap(); + if let Some(address) = line.strip_prefix("SHARD_LISTENING ") { + listening.send(address.to_owned()).unwrap(); + break; + } + } + for line in lines { + line.unwrap(); + } + }); + let address = address.recv_timeout(Duration::from_secs(30)).unwrap(); + let (mut socket, _) = tungstenite::connect(format!("ws://{address}/control")).unwrap(); + if let tungstenite::stream::MaybeTlsStream::Plain(stream) = socket.get_ref() { + stream + .set_read_timeout(Some(Duration::from_secs(30))) + .unwrap(); + } + + socket + .send(Message::Text( + serde_json::json!({ + "Console": serde_json::json!({ + "id": 1, + "op": "pause", + "path": "/user/world-inbox" + }).to_string() + }) + .to_string() + .into(), + )) + .unwrap(); + let response = socket.read().unwrap().into_text().unwrap(); + let reply: serde_json::Value = serde_json::from_str(&response).unwrap(); + assert!( + reply.get("Failed").is_none(), + "unexpected shard failure: {}", + shard.wait_for_failure() + ); + let console: serde_json::Value = serde_json::from_str( + reply["Console"] + .as_str() + .unwrap_or_else(|| panic!("expected a console reply: {reply}")), + ) + .unwrap(); + assert!(console.get("error").is_none(), "{console}"); + + socket + .send(Message::Text( + serde_json::json!({"View": {"terrain": false}}) + .to_string() + .into(), + )) + .unwrap(); + let stderr = shard.wait_for_failure(); + reader.join().unwrap(); + assert!( + stderr.contains("Picard entity actors did not answer request") + && stderr.contains("timed out waiting on channel"), + "{stderr}" + ); +} + fn get(address: &str, path: &str) -> Option { let mut stream = TcpStream::connect(address).ok()?; stream.set_read_timeout(Some(Duration::from_secs(5))).ok()?; @@ -41,7 +183,7 @@ fn gateway_presents_the_shards_as_one_world() { "sqlite://{}/shard-{{shard}}.db?mode=rwc", directory.display() ); - let _gateway = Gateway( + let _gateway = SimProcess( Command::new(env!("CARGO_BIN_EXE_sim")) .arg("5") .env("SIM_SHARDS", "2")