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

@@ -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-<id>`, `rabbit-<id>`, `wolf-<id>`, and `partnership-<id>` instead of `entity-<id>`.
 - Rabbits killed by hunters no longer move or breed later in the same tick.
 - Update Picard to `f88b9cb`.

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:

src/cluster/shard.rs

@@ -56,8 +56,7 @@ pub async fn run(options: ShardOptions) {
     tokio::spawn(save_snapshots(storage, snapshots));
 
     let (commands, receiver) = mpsc::unbounded_channel();
-    {
-        std::thread::Builder::new()
+    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);
@@ -67,7 +66,7 @@ pub async fn run(options: ShardOptions) {
             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.

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<serde_json::Value> {
     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")