feat: run shards as separate processes behind a gateway

The server is now a gateway that starts SIM_SHARDS shard processes
(`sim shard --index I --count N ...`), each with its own database from
the SHARD_DATABASE_URL template. The gateway steps all shards in
lockstep over a /control WebSocket, publishes the entity directory and
market board between steps, merges the shards' views into one world
for the UI and routes edits to the shards they concern.

Shards send entity messages and transfers to each other directly over
/remote WebSockets; every step waits until its outgoing messages are
accepted, so nothing is in flight when the next step starts or a
snapshot is taken. Shard processes exit when the gateway goes away.

Update Picard to f88b9cb and register every other shard as a console
peer `shard-N`, so one console session can list, inspect, manage and
message the actors of all shards.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

.gitignore

@@ -3,3 +3,6 @@
 sim.db
 sim.db-shm
 sim.db-wal
+sim-shard-*.db
+sim-shard-*.db-shm
+sim-shard-*.db-wal

CHANGELOG.md

@@ -14,10 +14,14 @@ Notable changes to this project are documented here.
 - Carts are actors that add carrying capacity, wear out and can be bought at the market.
 - Merchants carry their market's surplus to markets on other shards and bring back bargains.
 - People earn and spend coins; workers buy tools, meals and food for the village.
+- The server runs as a gateway that starts `SIM_SHARDS` (default 3) shard processes, steps them in lockstep and shows them to the UI as one world. Shards exchange entities and messages directly over WebSockets.
+- Every shard keeps its own database, configured with `SHARD_DATABASE_URL` (default `sqlite://sim-shard-{shard}.db?mode=rwc`).
+- The Picard console reaches the actors of the other shards as `@shard-N` (`peers`, `@shard-1 actors`, `send @shard-1 PATH JSON`); `/api/actors/ws?shard=N` connects to shard `N` directly.
 
 ### Changed
 
 - 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 `054449c`.
+- Update Picard to `f88b9cb`.
 - The default world is larger: 192×128 tiles with 180 people, 1200 rabbits, 80 wolves, and twice the food and houses.
+- `DATABASE_URL` is replaced by `SHARD_DATABASE_URL`; existing `sim.db` snapshots are not migrated.

Cargo.lock

@@ -1939,7 +1939,7 @@ checksum = "9b4f627cb1b25917193a259e49bdad08f671f8d9708acfd5fe0a8c1455d87220"
 [[package]]
 name = "picard"
 version = "0.4.2"
-source = "git+https://github.com/picard-rs/picard?rev=054449c9ab5a634c24fca81bcd4f9b20bbdc60b7#054449c9ab5a634c24fca81bcd4f9b20bbdc60b7"
+source = "git+https://github.com/picard-rs/picard?rev=f88b9cbf078512f726c4c47dbce842fda878388e#f88b9cbf078512f726c4c47dbce842fda878388e"
 dependencies = [
  "better_default",
  "chrono",
@@ -1968,7 +1968,7 @@ dependencies = [
 [[package]]
 name = "picard-macros"
 version = "0.2.0"
-source = "git+https://github.com/picard-rs/picard?rev=054449c9ab5a634c24fca81bcd4f9b20bbdc60b7#054449c9ab5a634c24fca81bcd4f9b20bbdc60b7"
+source = "git+https://github.com/picard-rs/picard?rev=f88b9cbf078512f726c4c47dbce842fda878388e#f88b9cbf078512f726c4c47dbce842fda878388e"
 dependencies = [
  "proc-macro2",
  "quote",

Cargo.toml

@@ -25,7 +25,7 @@ serde = { version = "1", features = ["derive"] }
 serde_json = "1"
 sqlx = { version = "0.8", default-features = false, features = ["runtime-tokio-rustls", "any", "sqlite", "postgres"] }
 rand = "0.8"
-picard = { git = "https://github.com/picard-rs/picard", rev = "054449c9ab5a634c24fca81bcd4f9b20bbdc60b7", features = [
+picard = { git = "https://github.com/picard-rs/picard", rev = "f88b9cbf078512f726c4c47dbce842fda878388e", features = [
     "remoting",
     "stats",
     "tracing",

src/cluster.rs

@@ -0,0 +1,6 @@
+//! Runs the world as several shard processes behind one gateway.
+
+pub mod gateway;
+pub mod protocol;
+pub mod shard;
+pub mod transport;

src/cluster/gateway.rs

@@ -0,0 +1,590 @@
+//! The gateway: starts the shard nodes, steps them in lockstep and presents their parts of the
+//! world to the UI as one world.
+
+use super::protocol::{Control, ControlReply, Peer};
+use super::shard::LISTENING;
+use crate::world::{EntityKey, Quote, Shard, regenerated_seed};
+use serde_json::{Value, json};
+use std::io::{BufRead, BufReader};
+use std::net::TcpStream;
+use std::process::{Child, Command, Stdio};
+use std::sync::{Arc, Mutex};
+use std::time::{Duration, Instant};
+use tokio::sync::{mpsc, oneshot, watch};
+use tungstenite::stream::MaybeTlsStream;
+use tungstenite::{Message, WebSocket};
+use uuid::Uuid;
+
+const FRAME: Duration = Duration::from_millis(100);
+const SAVE_INTERVAL: Duration = Duration::from_secs(1);
+const SHARD_TIMEOUT: Duration = Duration::from_secs(120);
+
+pub enum Msg {
+    Edit(String),
+    Script(String, oneshot::Sender<Result<(), String>>),
+    #[cfg(feature = "console")]
+    PicardConsole {
+        request: String,
+        shard: usize,
+        reply: oneshot::Sender<String>,
+    },
+    Restart {
+        regenerate: bool,
+        reply: oneshot::Sender<u64>,
+    },
+}
+
+pub struct GatewayOptions {
+    pub shards: usize,
+    pub seed: u64,
+    /// Database URL of each shard; `{shard}` is replaced by the shard's index.
+    pub database_template: String,
+}
+
+/// Where the gateway publishes the unified world for the HTTP handlers.
+pub struct Outputs {
+    pub terrain: Arc<Mutex<String>>,
+    pub script: Arc<Mutex<String>>,
+    pub state: watch::Sender<Arc<str>>,
+}
+
+/// Runs the cluster until the process exits; a failing shard takes the whole cluster down.
+pub fn conduct(
+    options: &GatewayOptions,
+    outputs: &Outputs,
+    mut messages: mpsc::UnboundedReceiver<Msg>,
+) {
+    let result = Cluster::start(options).and_then(|mut cluster| {
+        let result = cluster.run(outputs, &mut messages);
+        drop(cluster);
+        result
+    });
+    if let Err(error) = result {
+        tracing::error!(%error, "the cluster stopped");
+        std::process::exit(1);
+    }
+}
+
+struct ShardProcess {
+    child: Child,
+    address: String,
+}
+
+impl ShardProcess {
+    fn start(index: usize, options: &GatewayOptions) -> Result<Self, String> {
+        let executable = std::env::current_exe().map_err(|e| e.to_string())?;
+        let database = options
+            .database_template
+            .replace("{shard}", &index.to_string());
+        let mut child = Command::new(executable)
+            .args(["shard", "--attached"])
+            .args(["--index", &index.to_string()])
+            .args(["--count", &options.shards.to_string()])
+            .args(["--seed", &options.seed.to_string()])
+            .args(["--database", &database])
+            .stdin(Stdio::piped())
+            .stdout(Stdio::piped())
+            .stderr(Stdio::inherit())
+            .spawn()
+            .map_err(|e| format!("failed to start shard {index}: {e}"))?;
+        let stdout = child.stdout.take().expect("piped shard stdout");
+        let mut lines = BufReader::new(stdout).lines();
+        let address = loop {
+            match lines.next() {
+                Some(Ok(line)) => match line.strip_prefix(LISTENING) {
+                    Some(address) => break address.trim().to_owned(),
+                    None => println!("[shard {index}] {line}"),
+                },
+                _ => {
+                    let _ = child.kill();
+                    return Err(format!("shard {index} exited before it was ready"));
+                }
+            }
+        };
+        std::thread::spawn(move || {
+            for line in lines.map_while(Result::ok) {
+                println!("[shard {index}] {line}");
+            }
+        });
+        Ok(Self { child, address })
+    }
+}
+
+impl Drop for ShardProcess {
+    fn drop(&mut self) {
+        let _ = self.child.kill();
+        let _ = self.child.wait();
+    }
+}
+
+type Socket = WebSocket<MaybeTlsStream<TcpStream>>;
+
+/// The gateway's `/control` connection to one shard; commands are answered in order.
+struct ShardLink {
+    index: usize,
+    socket: Socket,
+}
+
+impl ShardLink {
+    fn connect(index: usize, address: &str) -> Result<Self, String> {
+        let url = format!("ws://{address}/control");
+        let (socket, _) = tungstenite::connect(&url)
+            .map_err(|e| format!("failed to connect to shard {index} at {url}: {e}"))?;
+        if let MaybeTlsStream::Plain(stream) = socket.get_ref() {
+            let _ = stream.set_nodelay(true);
+            let _ = stream.set_read_timeout(Some(SHARD_TIMEOUT));
+        }
+        Ok(Self { index, socket })
+    }
+
+    fn send(&mut self, control: &Control) -> Result<(), String> {
+        let frame = serde_json::to_string(control).map_err(|e| e.to_string())?;
+        self.socket
+            .send(Message::Text(frame.into()))
+            .map_err(|e| format!("lost shard {}: {e}", self.index))
+    }
+
+    fn receive(&mut self) -> Result<ControlReply, String> {
+        loop {
+            match self.socket.read() {
+                Ok(Message::Text(text)) => {
+                    return match serde_json::from_str(text.as_str()) {
+                        Ok(ControlReply::Failed(error)) => {
+                            Err(format!("shard {} failed: {error}", self.index))
+                        }
+                        Ok(reply) => Ok(reply),
+                        Err(error) => Err(format!(
+                            "unreadable reply from shard {}: {error}",
+                            self.index
+                        )),
+                    };
+                }
+                Ok(Message::Ping(_) | Message::Pong(_)) => {}
+                Ok(_) => return Err(format!("shard {} closed the connection", self.index)),
+                Err(error) => return Err(format!("lost shard {}: {error}", self.index)),
+            }
+        }
+    }
+
+    fn call(&mut self, control: &Control) -> Result<ControlReply, String> {
+        self.send(control)?;
+        self.receive()
+    }
+}
+
+struct Ready {
+    system_id: Uuid,
+    seed: u64,
+    tick: u64,
+    restored: bool,
+}
+
+fn ready(reply: ControlReply) -> Result<Ready, String> {
+    match reply {
+        ControlReply::Ready {
+            system_id,
+            seed,
+            tick,
+            restored,
+        } => Ok(Ready {
+            system_id,
+            seed,
+            tick,
+            restored,
+        }),
+        other => Err(unexpected(&other)),
+    }
+}
+
+fn unexpected(reply: &ControlReply) -> String {
+    format!("unexpected reply from a shard: {reply:?}")
+}
+
+struct Cluster {
+    links: Vec<ShardLink>,
+    processes: Vec<ShardProcess>,
+    seed: u64,
+    /// Directory changes to publish with the next step.
+    directory: Vec<(EntityKey, Option<usize>)>,
+    board: Vec<Quote>,
+    paused: bool,
+    speed: u64,
+    width: usize,
+    terrain_rev: Option<u64>,
+}
+
+impl Cluster {
+    fn start(options: &GatewayOptions) -> Result<Self, String> {
+        let processes = (0..options.shards)
+            .map(|index| ShardProcess::start(index, options))
+            .collect::<Result<Vec<_>, _>>()?;
+        let links = processes
+            .iter()
+            .enumerate()
+            .map(|(index, process)| ShardLink::connect(index, &process.address))
+            .collect::<Result<Vec<_>, _>>()?;
+        let mut cluster = Self {
+            links,
+            processes,
+            seed: options.seed,
+            directory: Vec::new(),
+            board: Vec::new(),
+            paused: false,
+            speed: 1,
+            width: 0,
+            terrain_rev: None,
+        };
+        let readies = cluster
+            .broadcast(|_| Control::Hello)?
+            .into_iter()
+            .map(ready)
+            .collect::<Result<Vec<_>, _>>()?;
+        let consistent = readies
+            .windows(2)
+            .all(|pair| pair[0].tick == pair[1].tick && pair[0].seed == pair[1].seed);
+        let restored = readies.iter().filter(|ready| ready.restored).count();
+        if consistent && (restored == 0 || restored == readies.len()) {
+            cluster.seed = readies[0].seed;
+            cluster.connect_peers(&readies)?;
+        } else {
+            tracing::warn!("the shards' saved states don't match; starting a new world");
+            cluster.restart(options.seed)?;
+        }
+        Ok(cluster)
+    }
+
+    /// Sends one command to every shard before waiting for any answer, so they work in parallel.
+    fn broadcast(
+        &mut self,
+        control: impl Fn(usize) -> Control,
+    ) -> Result<Vec<ControlReply>, String> {
+        for link in &mut self.links {
+            link.send(&control(link.index))?;
+        }
+        self.links.iter_mut().map(ShardLink::receive).collect()
+    }
+
+    fn broadcast_done(&mut self, control: impl Fn(usize) -> Control) -> Result<(), String> {
+        for reply in self.broadcast(control)? {
+            if !matches!(reply, ControlReply::Done) {
+                return Err(unexpected(&reply));
+            }
+        }
+        Ok(())
+    }
+
+    fn restart(&mut self, seed: u64) -> Result<(), String> {
+        let readies = self
+            .broadcast(|_| Control::Restart { seed })?
+            .into_iter()
+            .map(ready)
+            .collect::<Result<Vec<_>, _>>()?;
+        self.seed = seed;
+        self.terrain_rev = None;
+        self.connect_peers(&readies)
+    }
+
+    /// Tells every shard where the others are and publishes where every entity lives.
+    fn connect_peers(&mut self, readies: &[Ready]) -> Result<(), String> {
+        let peers: Vec<Peer> = readies
+            .iter()
+            .zip(&self.processes)
+            .map(|(ready, process)| Peer {
+                address: process.address.clone(),
+                system_id: ready.system_id,
+            })
+            .collect();
+        self.broadcast_done(|_| Control::Peers(peers.clone()))?;
+        self.directory.clear();
+        for (index, reply) in self.broadcast(|_| Control::Roster)?.into_iter().enumerate() {
+            let ControlReply::Roster(keys) = reply else {
+                return Err(unexpected(&reply));
+            };
+            self.directory
+                .extend(keys.into_iter().map(|key| (key, Some(index))));
+        }
+        self.board.clear();
+        Ok(())
+    }
+
+    fn step(&mut self) -> Result<(), String> {
+        let directory = std::mem::take(&mut self.directory);
+        let board = std::mem::take(&mut self.board);
+        let replies = self.broadcast(|_| Control::Step {
+            directory: directory.clone(),
+            board: board.clone(),
+        })?;
+        let mut gone = Vec::new();
+        for (index, reply) in replies.into_iter().enumerate() {
+            let ControlReply::Stepped { report, quotes } = reply else {
+                return Err(unexpected(&reply));
+            };
+            self.board.extend(quotes);
+            self.directory
+                .extend(report.moved.into_iter().map(|(key, to)| (key, Some(to))));
+            self.directory
+                .extend(report.arrived.into_iter().map(|key| (key, Some(index))));
+            gone.extend(report.left.into_iter().map(|key| (key, None)));
+        }
+        // Departures last, so that an entity that moved and died in the same step stays gone.
+        self.directory.extend(gone);
+        Ok(())
+    }
+
+    fn run(
+        &mut self,
+        outputs: &Outputs,
+        messages: &mut mpsc::UnboundedReceiver<Msg>,
+    ) -> Result<(), String> {
+        let mut last_saved = Instant::now();
+        self.publish(outputs)?;
+        loop {
+            let frame_started = Instant::now();
+            while let Ok(message) = messages.try_recv() {
+                self.handle(message, outputs)?;
+            }
+            if !self.paused {
+                for _ in 0..self.speed {
+                    self.step()?;
+                }
+            }
+            self.publish(outputs)?;
+            if last_saved.elapsed() >= SAVE_INTERVAL {
+                self.broadcast_done(|_| Control::Save)?;
+                last_saved = Instant::now();
+            }
+            std::thread::sleep(FRAME.saturating_sub(frame_started.elapsed()));
+        }
+    }
+
+    fn handle(&mut self, message: Msg, outputs: &Outputs) -> Result<(), String> {
+        match message {
+            Msg::Edit(command) => {
+                for (index, command) in route_edit(&command, self.links.len()) {
+                    let reply = self.links[index].call(&Control::Edit(command))?;
+                    if !matches!(reply, ControlReply::Done) {
+                        return Err(unexpected(&reply));
+                    }
+                }
+            }
+            Msg::Script(source, reply) => {
+                let mut result = Ok(());
+                for answer in self.broadcast(|_| Control::Script(source.clone()))? {
+                    let ControlReply::Script(answer) = answer else {
+                        return Err(unexpected(&answer));
+                    };
+                    if result.is_ok() {
+                        result = answer;
+                    }
+                }
+                if result.is_ok() {
+                    *outputs.script.lock().expect("script lock") = source;
+                }
+                let _ = reply.send(result);
+            }
+            #[cfg(feature = "console")]
+            Msg::PicardConsole {
+                request,
+                shard,
+                reply,
+            } => {
+                let response = match self.links.get_mut(shard) {
+                    Some(link) => match link.call(&Control::Console(request))? {
+                        ControlReply::Console(response) => response,
+                        other => return Err(unexpected(&other)),
+                    },
+                    None => json!({
+                        "id": null,
+                        "error": {"code": "invalid_request", "message": "no such shard"}
+                    })
+                    .to_string(),
+                };
+                let _ = reply.send(response);
+            }
+            Msg::Restart { regenerate, reply } => {
+                let seed = if regenerate {
+                    regenerated_seed(self.seed)
+                } else {
+                    self.seed
+                };
+                self.restart(seed)?;
+                let _ = reply.send(seed);
+            }
+        }
+        Ok(())
+    }
+
+    /// Merges the shards' views into one world for the UI.
+    fn publish(&mut self, outputs: &Outputs) -> Result<(), String> {
+        let mut states = Vec::with_capacity(self.links.len());
+        for reply in self.broadcast(|_| Control::View { terrain: false })? {
+            let ControlReply::View { state, .. } = reply else {
+                return Err(unexpected(&reply));
+            };
+            states.push(serde_json::from_str::<Value>(&state).map_err(|e| e.to_string())?);
+        }
+        let first = &states[0];
+        self.paused = first["paused"].as_bool().unwrap_or(false);
+        self.speed = first["speed"].as_u64().unwrap_or(1);
+        let rev = first["rev"].as_u64();
+        if rev != self.terrain_rev {
+            let reply = self.links[0].call(&Control::View { terrain: true })?;
+            let ControlReply::View {
+                terrain: Some(terrain),
+                ..
+            } = reply
+            else {
+                return Err(unexpected(&reply));
+            };
+            let parsed: Value = serde_json::from_str(&terrain).map_err(|e| e.to_string())?;
+            self.width = parsed["w"]
+                .as_u64()
+                .and_then(|w| usize::try_from(w).ok())
+                .unwrap_or(0);
+            *outputs.terrain.lock().expect("terrain lock") = terrain;
+            self.terrain_rev = rev;
+        }
+        let merged = merge_views(&states, self.width);
+        outputs.state.send_replace(Arc::from(merged.to_string()));
+        Ok(())
+    }
+}
+
+/// Which shards an edit command goes to: people are found through shard 0's router, stock is
+/// shared out, and everything else (including terrain, pause and speed) applies everywhere.
+fn route_edit(command: &str, shards: usize) -> Vec<(usize, String)> {
+    let words: Vec<&str> = command.split_whitespace().collect();
+    match words.as_slice() {
+        ["person", ..] => vec![(0, command.to_owned())],
+        ["stock", name, value] => {
+            let total = value
+                .parse::<i64>()
+                .map_or(0, |v| usize::try_from(v.max(0)).unwrap_or(usize::MAX));
+            (0..shards)
+                .filter_map(|index| Shard::new(index, shards))
+                .map(|shard| (shard.index, format!("stock {name} {}", shard.share(total))))
+                .collect()
+        }
+        _ => (0..shards)
+            .map(|index| (index, command.to_owned()))
+            .collect(),
+    }
+}
+
+/// Combines the shards' state JSON: settings come from shard 0, entities are concatenated,
+/// totals are summed, and `homes` and `shards` tell the UI where each shard lives.
+fn merge_views(states: &[Value], width: usize) -> Value {
+    let Some(Value::Object(first)) = states.first() else {
+        return json!({});
+    };
+    let mut merged = first.clone();
+    for (key, value) in &mut merged {
+        match key.as_str() {
+            "home" | "log" => {}
+            "stock" | "actor_stats" => {
+                if let Value::Object(totals) = value {
+                    for (name, total) in totals.iter_mut() {
+                        let sum: u64 = states
+                            .iter()
+                            .filter_map(|state| state[key.as_str()][name.as_str()].as_u64())
+                            .sum();
+                        *total = sum.into();
+                    }
+                }
+            }
+            _ => {
+                if let Value::Array(items) = value {
+                    for state in &states[1..] {
+                        if let Some(more) = state[key.as_str()].as_array() {
+                            items.extend(more.iter().cloned());
+                        }
+                    }
+                }
+            }
+        }
+    }
+    let limit = states
+        .iter()
+        .filter_map(|state| state["log"].as_array().map(Vec::len))
+        .max()
+        .unwrap_or(0);
+    let mut log: Vec<Value> = states
+        .iter()
+        .filter_map(|state| state["log"].as_array())
+        .flatten()
+        .cloned()
+        .collect();
+    log.sort_by_key(|entry| entry.as_str().and_then(log_day).unwrap_or(0));
+    let log = log.split_off(log.len().saturating_sub(limit));
+    merged.insert("log".into(), Value::Array(log));
+    merged.insert(
+        "homes".into(),
+        states.iter().map(|state| state["home"].clone()).collect(),
+    );
+    let layout = Shard::new(0, states.len()).unwrap_or(Shard::SINGLE);
+    merged.insert(
+        "shards".into(),
+        (0..states.len())
+            .map(|index| {
+                let columns = layout.columns_of(index, width);
+                json!([columns.start, columns.end])
+            })
+            .collect(),
+    );
+    Value::Object(merged)
+}
+
+fn log_day(entry: &str) -> Option<u64> {
+    entry.strip_prefix("[day ")?.split(']').next()?.parse().ok()
+}
+
+#[cfg(test)]
+mod tests {
+    use super::{merge_views, route_edit};
+    use serde_json::json;
+
+    #[test]
+    fn views_merge_into_one_world() {
+        let left = json!({
+            "paused": false, "speed": 2, "tick": 10, "home": [5, 5],
+            "stock": {"food": 3, "wood": 1},
+            "actor_stats": {"live_actors": 4},
+            "people": [[1, 5, 5]], "markets": [[1, 6, 6]],
+            "log": ["[day 1] a", "[day 3] c"]
+        });
+        let right = json!({
+            "paused": true, "speed": 9, "tick": 10, "home": [50, 5],
+            "stock": {"food": 4, "wood": 2},
+            "actor_stats": {"live_actors": 6},
+            "people": [[2, 50, 5], [4, 51, 5]], "markets": [],
+            "log": ["[day 2] b"]
+        });
+        let merged = merge_views(&[left, right], 96);
+        assert_eq!(merged["speed"], 2);
+        assert_eq!(merged["paused"], false);
+        assert_eq!(merged["stock"], json!({"food": 7, "wood": 3}));
+        assert_eq!(merged["actor_stats"]["live_actors"], 10);
+        assert_eq!(merged["people"].as_array().unwrap().len(), 3);
+        assert_eq!(merged["markets"].as_array().unwrap().len(), 1);
+        assert_eq!(merged["log"], json!(["[day 2] b", "[day 3] c"]));
+        assert_eq!(merged["homes"], json!([[5, 5], [50, 5]]));
+        assert_eq!(merged["shards"], json!([[0, 48], [48, 96]]));
+    }
+
+    #[test]
+    fn edits_go_where_they_apply() {
+        assert_eq!(
+            route_edit("person 4 coins 9", 3),
+            vec![(0, "person 4 coins 9".into())]
+        );
+        assert_eq!(
+            route_edit("stock food 10", 3),
+            vec![
+                (0, "stock food 4".into()),
+                (1, "stock food 3".into()),
+                (2, "stock food 3".into())
+            ]
+        );
+        assert_eq!(route_edit("pause 1", 2).len(), 2);
+    }
+}

src/cluster/protocol.rs

@@ -0,0 +1,64 @@
+//! Messages between the gateway and its shard nodes on the `/control` WebSocket.
+
+use crate::world::{EntityKey, Quote, StepReport};
+use serde::{Deserialize, Serialize};
+use uuid::Uuid;
+
+/// Where a shard node accepts entity messages from the other shards.
+#[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Eq)]
+pub struct Peer {
+    /// The `host:port` of the shard node, which serves `/remote` and `/console`.
+    pub address: String,
+    /// The current actor system of the shard, which remote requests are addressed to.
+    pub system_id: Uuid,
+}
+
+#[derive(Debug, Deserialize, Serialize)]
+pub enum Control {
+    /// Asks a freshly connected shard what it is running.
+    Hello,
+    /// Connects the shard to the other shards; sent again whenever any of them restarts.
+    Peers(Vec<Peer>),
+    /// Runs one step after applying the gateway's directory changes and market board.
+    Step {
+        directory: Vec<(EntityKey, Option<usize>)>,
+        board: Vec<Quote>,
+    },
+    /// Asks for the shard's part of the world, and for the terrain when `terrain` is set.
+    View {
+        terrain: bool,
+    },
+    /// Takes in admitted entities and persists a snapshot.
+    Save,
+    /// Every entity the shard simulates, for the gateway's directory.
+    Roster,
+    Edit(String),
+    Script(String),
+    Restart {
+        seed: u64,
+    },
+    Console(String),
+}
+
+#[derive(Debug, Deserialize, Serialize)]
+pub enum ControlReply {
+    Ready {
+        system_id: Uuid,
+        seed: u64,
+        tick: u64,
+        restored: bool,
+    },
+    Done,
+    Stepped {
+        report: StepReport,
+        quotes: Vec<Quote>,
+    },
+    View {
+        state: String,
+        terrain: Option<String>,
+    },
+    Roster(Vec<EntityKey>),
+    Script(Result<(), String>),
+    Console(String),
+    Failed(String),
+}

src/cluster/shard.rs

@@ -0,0 +1,330 @@
+//! A shard node: simulates one strip of the world, answers the gateway on `/control` and takes
+//! entity messages from the other shards on `/remote`.
+
+use super::protocol::{Control, ControlReply, Peer};
+use super::transport::{self, ClusterTransport};
+use crate::storage::Storage;
+use crate::world::{Router, Shard, SimScript, World};
+use axum::extract::State;
+use axum::extract::ws::{Message, WebSocket, WebSocketUpgrade};
+use axum::response::Response;
+use axum::routing::get;
+use std::io::Write;
+use std::sync::{Arc, RwLock};
+use std::time::Duration;
+use tokio::sync::{mpsc, oneshot, watch};
+
+/// The line a shard prints to stdout once it accepts connections, followed by its address.
+pub const LISTENING: &str = "SHARD_LISTENING";
+
+pub struct ShardOptions {
+    pub shard: Shard,
+    pub seed: u64,
+    pub database_url: String,
+    pub listen: String,
+    pub script: SimScript,
+    /// Exit when stdin closes, so that a shard never outlives the gateway that started it.
+    pub attached: bool,
+}
+
+type Command = (Control, oneshot::Sender<ControlReply>);
+type CurrentRouter = Arc<RwLock<Option<Arc<Router>>>>;
+#[cfg(feature = "console")]
+type CurrentConsole = Arc<RwLock<Option<picard::console::Console>>>;
+
+/// What the request handlers need from the current world, which changes on restart.
+#[derive(Clone, Default)]
+struct Current {
+    router: CurrentRouter,
+    #[cfg(feature = "console")]
+    console: CurrentConsole,
+}
+
+impl Current {
+    fn publish(&self, world: &World) {
+        *self.router.write().expect("router lock") = Some(Arc::clone(world.router()));
+        #[cfg(feature = "console")]
+        {
+            *self.console.write().expect("console lock") = Some(world.console());
+        }
+    }
+}
+
+#[derive(Clone)]
+struct Node {
+    commands: mpsc::UnboundedSender<Command>,
+    current: Current,
+}
+
+pub async fn run(options: ShardOptions) {
+    let ShardOptions {
+        shard,
+        seed,
+        database_url,
+        listen,
+        script,
+        attached,
+    } = options;
+    let storage = Storage::connect(&database_url).await.unwrap_or_else(|e| {
+        panic!(
+            "failed to connect to the database of shard {}: {e}",
+            shard.index
+        )
+    });
+    let saved = storage
+        .load()
+        .await
+        .unwrap_or_else(|e| panic!("failed to load the snapshot of shard {}: {e}", shard.index));
+    let (saves, snapshots) = watch::channel(None::<String>);
+    tokio::spawn(save_snapshots(storage, snapshots));
+
+    let current = Current::default();
+    let (commands, receiver) = mpsc::unbounded_channel();
+    {
+        let current = current.clone();
+        std::thread::Builder::new()
+            .name(format!("shard-{}", shard.index))
+            .spawn(move || {
+                let (world, restored) = open_world(saved.as_deref(), seed, script, shard);
+                current.publish(&world);
+                run_world(world, restored, receiver, &current, &saves);
+            })
+            .expect("failed to start the simulation thread");
+    }
+    if attached {
+        std::thread::spawn(|| {
+            let _ = std::io::copy(&mut std::io::stdin(), &mut std::io::sink());
+            std::process::exit(0);
+        });
+    }
+
+    let app = axum::Router::new()
+        .route("/control", get(control))
+        .route("/remote", get(remote));
+    #[cfg(feature = "console")]
+    let app = app.route("/console", get(console_socket));
+    let app = app.with_state(Node { commands, current });
+    let listener = tokio::net::TcpListener::bind(&listen)
+        .await
+        .unwrap_or_else(|e| panic!("shard {} failed to bind {listen}: {e}", shard.index));
+    let address = listener.local_addr().expect("shard listener address");
+    {
+        let mut stdout = std::io::stdout().lock();
+        let _ = writeln!(stdout, "{LISTENING} {address}");
+        let _ = stdout.flush();
+    }
+    tracing::info!(shard = shard.index, %address, "shard listening");
+    axum::serve(listener, app)
+        .await
+        .expect("shard server failed");
+}
+
+/// Restores the shard's saved world, or creates a new one when there is none or it doesn't fit.
+fn open_world(saved: Option<&str>, seed: u64, script: SimScript, shard: Shard) -> (World, bool) {
+    match saved
+        .map(|snapshot| World::from_snapshot_with_script(snapshot, seed, script.clone(), shard))
+    {
+        Some(Ok(world)) => (world, true),
+        Some(Err(error)) => {
+            tracing::warn!(shard = shard.index, %error, "discarding an unusable snapshot");
+            (World::new_shard(seed, script, shard), false)
+        }
+        None => (World::new_shard(seed, script, shard), false),
+    }
+}
+
+async fn save_snapshots(storage: Storage, mut snapshots: watch::Receiver<Option<String>>) {
+    while snapshots.changed().await.is_ok() {
+        let Some(snapshot) = snapshots.borrow_and_update().clone() else {
+            continue;
+        };
+        if let Err(error) = storage.save(&snapshot).await {
+            tracing::error!(%error, "failed to persist the shard snapshot");
+        }
+    }
+}
+
+/// The Rune VM is not Send, so the world lives on this thread and only acts on gateway commands.
+fn run_world(
+    mut world: World,
+    mut restored: bool,
+    mut commands: mpsc::UnboundedReceiver<Command>,
+    current: &Current,
+    saves: &watch::Sender<Option<String>>,
+) {
+    while let Some((control, reply)) = commands.blocking_recv() {
+        let answer = match control {
+            Control::Hello => ready(&world, restored),
+            Control::Peers(peers) => {
+                connect(&world, peers);
+                ControlReply::Done
+            }
+            Control::Step { directory, board } => {
+                world.router().apply_directory(&directory);
+                world.set_board(board);
+                world.step();
+                world.router().flush();
+                ControlReply::Stepped {
+                    report: world.take_report(),
+                    quotes: world.quotes(),
+                }
+            }
+            Control::View { terrain } => ControlReply::View {
+                state: world.state_json(),
+                terrain: terrain.then(|| world.terrain_json()),
+            },
+            Control::Save => {
+                world.admit_arrivals();
+                match world.snapshot_json() {
+                    Ok(snapshot) => {
+                        saves.send_replace(Some(snapshot));
+                        ControlReply::Done
+                    }
+                    Err(error) => ControlReply::Failed(error.to_string()),
+                }
+            }
+            Control::Roster => ControlReply::Roster(world.roster()),
+            Control::Edit(command) => {
+                world.apply(&command);
+                ControlReply::Done
+            }
+            Control::Script(source) => ControlReply::Script(world.set_script(&source)),
+            Control::Restart { seed } => {
+                world = world.restarted(seed);
+                restored = false;
+                current.publish(&world);
+                ready(&world, restored)
+            }
+            Control::Console(request) => console(&world, &request),
+        };
+        let _ = reply.send(answer);
+    }
+}
+
+fn ready(world: &World, restored: bool) -> ControlReply {
+    ControlReply::Ready {
+        system_id: world.router().system_id(),
+        seed: world.seed(),
+        tick: world.tick(),
+        restored,
+    }
+}
+
+fn connect(world: &World, peers: Vec<Peer>) {
+    let this = world.shard().index;
+    let transport = ClusterTransport::new(this, peers.clone());
+    world.router().set_remote(Arc::clone(&transport) as _);
+    #[cfg(feature = "console")]
+    for (index, peer) in peers
+        .into_iter()
+        .enumerate()
+        .filter(|&(index, _)| index != this)
+    {
+        let transport = Arc::clone(&transport);
+        world.register_console_peer(
+            &format!("shard-{index}"),
+            Arc::new(move |path, message| {
+                crate::world::RemoteSender::send(&*transport, index, path, message);
+                Ok(())
+            }),
+            Arc::new(move |request| transport::console_query(&peer.address, request)),
+        );
+    }
+}
+
+#[cfg(feature = "console")]
+fn console(world: &World, request: &str) -> ControlReply {
+    ControlReply::Console(world.console().handle_text(request))
+}
+
+#[cfg(not(feature = "console"))]
+fn console(_: &World, _: &str) -> ControlReply {
+    ControlReply::Failed("this build has no Picard console".into())
+}
+
+async fn control(upgrade: WebSocketUpgrade, State(node): State<Node>) -> Response {
+    upgrade
+        .max_message_size(1 << 30)
+        .on_upgrade(move |socket| serve_control(socket, node))
+}
+
+async fn serve_control(mut socket: WebSocket, node: Node) {
+    while let Some(Ok(message)) = socket.recv().await {
+        let Message::Text(text) = message else {
+            continue;
+        };
+        let answer = match serde_json::from_str::<Control>(text.as_str()) {
+            Ok(control) => {
+                let (reply, result) = oneshot::channel();
+                if node.commands.send((control, reply)).is_err() {
+                    return;
+                }
+                result
+                    .await
+                    .unwrap_or_else(|_| ControlReply::Failed("the simulation stopped".into()))
+            }
+            Err(error) => ControlReply::Failed(format!("unreadable command: {error}")),
+        };
+        let Ok(frame) = serde_json::to_string(&answer) else {
+            return;
+        };
+        if socket.send(Message::Text(frame.into())).await.is_err() {
+            return;
+        }
+    }
+}
+
+async fn remote(upgrade: WebSocketUpgrade, State(node): State<Node>) -> Response {
+    upgrade
+        .max_message_size(1 << 30)
+        .on_upgrade(move |socket| serve_remote(socket, node))
+}
+
+async fn serve_remote(mut socket: WebSocket, node: Node) {
+    while let Some(Ok(message)) = socket.recv().await {
+        let Message::Text(text) = message else {
+            continue;
+        };
+        let router = node.current.router.read().expect("router lock").clone();
+        let Some(router) = router else {
+            return;
+        };
+        let answer = tokio::task::spawn_blocking(move || {
+            transport::receive(&router, "shard", text.as_str())
+        });
+        let Ok(Ok(frame)) = tokio::time::timeout(Duration::from_secs(30), answer).await else {
+            return;
+        };
+        if socket.send(Message::Text(frame.into())).await.is_err() {
+            return;
+        }
+    }
+}
+
+#[cfg(feature = "console")]
+async fn console_socket(upgrade: WebSocketUpgrade, State(node): State<Node>) -> Response {
+    upgrade.on_upgrade(move |socket| serve_console(socket, node))
+}
+
+/// Answers console requests forwarded by another shard's console; it never waits for the
+/// simulation thread, which may itself be waiting for this answer.
+#[cfg(feature = "console")]
+async fn serve_console(mut socket: WebSocket, node: Node) {
+    while let Some(Ok(message)) = socket.recv().await {
+        let Message::Text(text) = message else {
+            continue;
+        };
+        let console = node.current.console.read().expect("console lock").clone();
+        let Some(console) = console else {
+            return;
+        };
+        let Ok(response) =
+            tokio::task::spawn_blocking(move || console.handle_text(text.as_str())).await
+        else {
+            return;
+        };
+        if socket.send(Message::Text(response.into())).await.is_err() {
+            return;
+        }
+    }
+}

src/cluster/transport.rs

@@ -0,0 +1,255 @@
+//! Delivers entity messages between shard nodes over persistent WebSockets.
+//!
+//! Every peer gets one connection and one worker thread. Messages are queued in order and sent
+//! in batches; the receiving shard answers every batch with one response per message, so
+//! [`RemoteSender::flush`] can wait until everything sent so far has reached its mailbox.
+
+use super::protocol::Peer;
+use crate::world::{RemoteSender, Router};
+use picard::remoting::{ActorAddress, RemoteRequest, RemoteResponse};
+use std::collections::VecDeque;
+use std::net::TcpStream;
+use std::sync::{Arc, Condvar, Mutex};
+use std::thread::JoinHandle;
+use std::time::{Duration, Instant};
+use tungstenite::stream::MaybeTlsStream;
+use tungstenite::{Message, WebSocket};
+
+const MAX_BATCH: usize = 512;
+const FLUSH_TIMEOUT: Duration = Duration::from_secs(10);
+const CONNECT_ATTEMPTS: u32 = 50;
+
+#[derive(Default)]
+struct Queue {
+    waiting: VecDeque<RemoteRequest>,
+    /// Messages queued or sent but not yet answered.
+    unanswered: usize,
+    closed: bool,
+}
+
+struct Link {
+    queue: Mutex<Queue>,
+    changed: Condvar,
+}
+
+impl Link {
+    fn lock(&self) -> std::sync::MutexGuard<'_, Queue> {
+        self.queue.lock().expect("transport queue lock")
+    }
+}
+
+struct Connection {
+    peer: Peer,
+    link: Arc<Link>,
+    worker: Option<JoinHandle<()>>,
+}
+
+pub struct ClusterTransport {
+    connections: Vec<Option<Connection>>,
+}
+
+impl ClusterTransport {
+    /// Connects lazily to every peer except `this` shard.
+    pub fn new(this: usize, peers: Vec<Peer>) -> Arc<Self> {
+        let connections = peers
+            .into_iter()
+            .enumerate()
+            .map(|(index, peer)| {
+                (index != this).then(|| {
+                    let link = Arc::new(Link {
+                        queue: Mutex::default(),
+                        changed: Condvar::new(),
+                    });
+                    let worker = {
+                        let (link, url) =
+                            (Arc::clone(&link), format!("ws://{}/remote", peer.address));
+                        std::thread::Builder::new()
+                            .name(format!("shard-link-{index}"))
+                            .spawn(move || run_link(&url, &link))
+                            .expect("failed to start a shard link")
+                    };
+                    Connection {
+                        peer,
+                        link,
+                        worker: Some(worker),
+                    }
+                })
+            })
+            .collect();
+        Arc::new(Self { connections })
+    }
+}
+
+impl RemoteSender for ClusterTransport {
+    fn send(&self, shard: usize, path: &str, payload: serde_json::Value) {
+        let Some(Some(connection)) = self.connections.get(shard) else {
+            tracing::warn!(shard, path, "dropped a message for an unknown shard");
+            return;
+        };
+        let destination = ActorAddress::local(connection.peer.system_id, path);
+        let mut queue = connection.link.lock();
+        queue
+            .waiting
+            .push_back(RemoteRequest::new(destination, None, payload));
+        queue.unanswered += 1;
+        connection.link.changed.notify_all();
+    }
+
+    fn flush(&self) {
+        let deadline = Instant::now() + FLUSH_TIMEOUT;
+        for (shard, connection) in self.connections.iter().enumerate() {
+            let Some(connection) = connection else {
+                continue;
+            };
+            let mut queue = connection.link.lock();
+            while queue.unanswered > 0 && !queue.closed {
+                let left = deadline.saturating_duration_since(Instant::now());
+                if left.is_zero() {
+                    tracing::warn!(
+                        shard,
+                        unanswered = queue.unanswered,
+                        "shard flush timed out"
+                    );
+                    break;
+                }
+                queue = connection
+                    .link
+                    .changed
+                    .wait_timeout(queue, left)
+                    .expect("transport queue lock")
+                    .0;
+            }
+        }
+    }
+}
+
+impl Drop for ClusterTransport {
+    fn drop(&mut self) {
+        for connection in self.connections.iter_mut().flatten() {
+            connection.link.lock().closed = true;
+            connection.link.changed.notify_all();
+            if let Some(worker) = connection.worker.take() {
+                let _ = worker.join();
+            }
+        }
+    }
+}
+
+type Socket = WebSocket<MaybeTlsStream<TcpStream>>;
+
+fn connect(url: &str) -> Option<Socket> {
+    for attempt in 0..CONNECT_ATTEMPTS {
+        match tungstenite::connect(url) {
+            Ok((socket, _)) => {
+                if let MaybeTlsStream::Plain(stream) = socket.get_ref() {
+                    let _ = stream.set_nodelay(true);
+                }
+                return Some(socket);
+            }
+            Err(error) if attempt + 1 == CONNECT_ATTEMPTS => {
+                tracing::error!(url, %error, "failed to connect to a shard");
+            }
+            Err(_) => std::thread::sleep(Duration::from_millis(100)),
+        }
+    }
+    None
+}
+
+/// Sends queued messages to one peer until the transport is dropped.
+fn run_link(url: &str, link: &Link) {
+    let mut socket: Option<Socket> = None;
+    loop {
+        let batch: Vec<RemoteRequest> = {
+            let mut queue = link.lock();
+            while queue.waiting.is_empty() && !queue.closed {
+                queue = link.changed.wait(queue).expect("transport queue lock");
+            }
+            if queue.closed {
+                if let Some(mut socket) = socket {
+                    let _ = socket.close(None);
+                }
+                return;
+            }
+            let count = queue.waiting.len().min(MAX_BATCH);
+            queue.waiting.drain(..count).collect()
+        };
+        let sent = batch.len();
+        if socket.is_none() {
+            socket = connect(url);
+        }
+        let delivered = socket
+            .as_mut()
+            .is_some_and(|socket| exchange(socket, &batch));
+        if !delivered {
+            tracing::warn!(url, dropped = sent, "lost messages to a shard");
+            socket = None;
+        }
+        let mut queue = link.lock();
+        queue.unanswered = queue.unanswered.saturating_sub(sent);
+        link.changed.notify_all();
+    }
+}
+
+fn exchange(socket: &mut Socket, batch: &[RemoteRequest]) -> bool {
+    let Ok(frame) = serde_json::to_string(batch) else {
+        return false;
+    };
+    if socket.send(Message::Text(frame.into())).is_err() {
+        return false;
+    }
+    loop {
+        match socket.read() {
+            Ok(Message::Text(text)) => {
+                let Ok(responses) = serde_json::from_str::<Vec<RemoteResponse>>(&text) else {
+                    return false;
+                };
+                for response in responses.iter().filter(|response| !response.accepted) {
+                    tracing::debug!(error = ?response.error, "a shard rejected a message");
+                }
+                return true;
+            }
+            Ok(Message::Ping(_) | Message::Pong(_)) => {}
+            _ => return false,
+        }
+    }
+}
+
+/// Answers one batch of messages from another shard.
+pub fn receive(router: &Router, peer: &str, frame: &str) -> String {
+    let responses: Vec<RemoteResponse> = match serde_json::from_str::<Vec<RemoteRequest>>(frame) {
+        Ok(requests) => requests
+            .into_iter()
+            .map(|request| router.handle(peer, request))
+            .collect(),
+        Err(error) => {
+            tracing::warn!(%error, "received an unreadable batch from a shard");
+            Vec::new()
+        }
+    };
+    serde_json::to_string(&responses).unwrap_or_else(|_| "[]".into())
+}
+
+/// Runs one Picard console request on the shard at `address` and returns its JSON response.
+#[cfg(feature = "console")]
+pub fn console_query(address: &str, request: &str) -> Result<String, String> {
+    let url = format!("ws://{address}/console");
+    let (mut socket, _) =
+        tungstenite::connect(&url).map_err(|e| format!("cannot reach {url}: {e}"))?;
+    if let MaybeTlsStream::Plain(stream) = socket.get_ref() {
+        let _ = stream.set_read_timeout(Some(FLUSH_TIMEOUT));
+    }
+    socket
+        .send(Message::Text(request.to_owned().into()))
+        .map_err(|e| e.to_string())?;
+    loop {
+        match socket.read() {
+            Ok(Message::Text(text)) => {
+                let _ = socket.close(None);
+                return Ok(text.as_str().to_owned());
+            }
+            Ok(Message::Ping(_) | Message::Pong(_)) => {}
+            Ok(_) => return Err("the shard closed the console".into()),
+            Err(error) => return Err(error.to_string()),
+        }
+    }
+}

src/main.rs

@@ -1,3 +1,4 @@
+mod cluster;
 mod storage;
 mod world;
 
@@ -8,12 +9,13 @@ use axum::http::{StatusCode, header};
 use axum::middleware::{self, Next};
 use axum::response::{Html, IntoResponse, Response};
 use axum::routing::{get, post};
+use cluster::gateway::{GatewayOptions, Msg, Outputs};
+use cluster::shard::ShardOptions;
 use include_dir::{Dir, include_dir};
 use std::sync::{Arc, Mutex};
-use std::time::{Duration, Instant};
-use storage::Storage;
+use std::time::Duration;
 use tokio::sync::{mpsc, oneshot, watch};
-use world::{Shard, SimScript, World};
+use world::{Shard, SimScript};
 
 // The UI bundle that build.rs produced from ui/.
 static UI: Dir = include_dir!("$OUT_DIR/ui-pkg");
@@ -26,17 +28,6 @@ try { const m = await import('/pkg/ui.js'); await m.default({ module_or_path: '/
 catch (e) { document.body.textContent = 'Failed to load the UI: ' + e; }
 </script></body></html>"#;
 
-enum Msg {
-    Edit(String),
-    Script(String, oneshot::Sender<Result<(), String>>),
-    #[cfg(feature = "console")]
-    PicardConsole(String, oneshot::Sender<String>),
-    Restart {
-        regenerate: bool,
-        reply: oneshot::Sender<u64>,
-    },
-}
-
 #[derive(Clone)]
 struct App {
     terrain: Arc<Mutex<String>>,
@@ -45,95 +36,6 @@ struct App {
     tx: mpsc::UnboundedSender<Msg>,
 }
 
-struct SimStart {
-    seed: u64,
-    script: SimScript,
-    snapshot: Option<String>,
-}
-
-/// The Rune VM is not Send, so the world lives on this thread and publishes JSON snapshots.
-fn run_sim(
-    SimStart {
-        seed,
-        script: sim_script,
-        snapshot,
-    }: SimStart,
-    mut rx: mpsc::UnboundedReceiver<Msg>,
-    terrain: &Mutex<String>,
-    script: &Mutex<String>,
-    state_tx: &watch::Sender<Arc<str>>,
-    snapshot_tx: &watch::Sender<Option<String>>,
-) {
-    let mut world = match snapshot {
-        Some(snapshot) => {
-            World::from_snapshot_with_script(&snapshot, seed, sim_script, Shard::SINGLE)
-                .unwrap_or_else(|e| panic!("failed to restore simulation snapshot: {e}"))
-        }
-        None => World::new_shard(seed, sim_script, Shard::SINGLE),
-    };
-    *script.lock().unwrap() = world.script().to_string();
-    let mut rev = u32::MAX;
-    let mut last_saved = Instant::now();
-    snapshot_tx.send_replace(Some(
-        world
-            .snapshot_json()
-            .expect("failed to serialize initial simulation snapshot"),
-    ));
-    loop {
-        while let Ok(msg) = rx.try_recv() {
-            match msg {
-                Msg::Edit(cmd) => world.apply(&cmd),
-                Msg::Script(src, reply) => {
-                    let r = world.set_script(&src);
-                    if r.is_ok() {
-                        *script.lock().unwrap() = src;
-                    }
-                    let _ = reply.send(r);
-                }
-                #[cfg(feature = "console")]
-                Msg::PicardConsole(request, reply) => {
-                    let _ = reply.send(world.handle_picard_console_text(&request));
-                }
-                Msg::Restart { regenerate, reply } => {
-                    let seed = if regenerate {
-                        world.regenerated_seed()
-                    } else {
-                        world.seed()
-                    };
-                    world = world.restarted(seed);
-                    *script.lock().unwrap() = world.script().to_string();
-                    rev = u32::MAX;
-                    snapshot_tx.send_replace(Some(
-                        world
-                            .snapshot_json()
-                            .expect("failed to serialize restarted simulation"),
-                    ));
-                    let _ = reply.send(seed);
-                }
-            }
-        }
-        if !world.paused {
-            for _ in 0..world.speed {
-                world.step();
-            }
-        }
-        if world.terrain_rev != rev {
-            rev = world.terrain_rev;
-            *terrain.lock().unwrap() = world.terrain_json();
-        }
-        state_tx.send_replace(Arc::from(world.state_json()));
-        if last_saved.elapsed() >= Duration::from_secs(1) {
-            snapshot_tx.send_replace(Some(
-                world
-                    .snapshot_json()
-                    .expect("failed to serialize simulation snapshot"),
-            ));
-            last_saved = Instant::now();
-        }
-        std::thread::sleep(Duration::from_millis(100));
-    }
-}
-
 /// Rejects cross-site requests (including WebSocket upgrades), which could otherwise edit the world.
 async fn same_origin(req: Request, next: Next) -> Response {
     let h = req.headers();
@@ -224,19 +126,34 @@ async fn ws(upgrade: WebSocketUpgrade, State(app): State<App>) -> Response {
 }
 
 #[cfg(feature = "console")]
-async fn picard_ws(upgrade: WebSocketUpgrade, State(app): State<App>) -> Response {
-    upgrade.on_upgrade(move |socket| picard_client(socket, app))
+#[derive(serde::Deserialize)]
+struct ConsoleTarget {
+    #[serde(default)]
+    shard: usize,
 }
 
 #[cfg(feature = "console")]
-async fn picard_client(mut socket: WebSocket, app: App) {
+async fn picard_ws(
+    upgrade: WebSocketUpgrade,
+    axum::extract::Query(target): axum::extract::Query<ConsoleTarget>,
+    State(app): State<App>,
+) -> Response {
+    upgrade.on_upgrade(move |socket| picard_client(socket, app, target.shard))
+}
+
+#[cfg(feature = "console")]
+async fn picard_client(mut socket: WebSocket, app: App, shard: usize) {
     while let Some(message) = socket.recv().await {
         match message {
             Ok(Message::Text(text)) if text.len() <= picard::console::MAX_REQUEST_BYTES => {
                 let (reply, result) = oneshot::channel();
                 if app
                     .tx
-                    .send(Msg::PicardConsole(text.to_string(), reply))
+                    .send(Msg::PicardConsole {
+                        request: text.to_string(),
+                        shard,
+                        reply,
+                    })
                     .is_err()
                 {
                     return;
@@ -298,78 +215,122 @@ async fn client(mut socket: WebSocket, app: App) {
     }
 }
 
-#[tokio::main]
-async fn main() {
-    let filter = tracing_subscriber::EnvFilter::try_from_default_env()
-        .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("picard=info"));
-    tracing_subscriber::fmt()
-        .with_env_filter(filter)
-        .with_target(false)
-        .init();
-
-    let seed = std::env::args()
-        .nth(1)
-        .and_then(|s| s.parse().ok())
-        .unwrap_or(42u64);
+fn load_script() -> (String, SimScript) {
     let config_path = std::env::var("SIM_CONFIG").unwrap_or_else(|_| "config.rn".into());
     let config_source = std::fs::read_to_string(&config_path)
         .unwrap_or_else(|e| panic!("failed to read simulation config at {config_path}: {e}"));
     let sim_script = SimScript::load(&config_source)
         .unwrap_or_else(|e| panic!("failed to load simulation config at {config_path}: {e}"));
-    let addr = std::env::var("SIM_ADDR").unwrap_or_else(|_| "127.0.0.1:7878".into());
-    let database_url =
-        std::env::var("DATABASE_URL").unwrap_or_else(|_| "sqlite://sim.db?mode=rwc".into());
-    let storage = Storage::connect(&database_url)
-        .await
-        .expect("failed to connect to simulation database");
-    let saved_snapshot = storage
-        .load()
-        .await
-        .expect("failed to load simulation snapshot");
-    let (snapshot_tx, snapshot_rx) = watch::channel(None::<String>);
-    let snapshot_storage = storage.clone();
-    tokio::spawn(async move {
-        let mut snapshots = snapshot_rx;
-        let mut interval = tokio::time::interval(Duration::from_secs(1));
-        interval.tick().await;
-        loop {
-            tokio::select! {
-                _ = interval.tick() => {}
-                changed = snapshots.changed() => {
-                    if changed.is_err() {
-                        break;
-                    }
-                }
-            }
-            let Some(snapshot) = snapshots.borrow_and_update().clone() else {
-                continue;
-            };
-            if let Err(error) = snapshot_storage.save(&snapshot).await {
-                eprintln!("failed to persist simulation snapshot: {error}");
-            }
+    (config_source, sim_script)
+}
+
+/// Parses `shard --index I --count N --seed S --database URL [--listen ADDR] [--attached]`.
+fn shard_options(args: &[String], script: SimScript) -> ShardOptions {
+    let mut index = None;
+    let mut count = None;
+    let mut seed = 42;
+    let mut database_url = None;
+    let mut listen = "127.0.0.1:0".to_owned();
+    let mut attached = false;
+    let mut args = args.iter();
+    while let Some(flag) = args.next() {
+        if flag == "--attached" {
+            attached = true;
+            continue;
         }
+        let value = args
+            .next()
+            .unwrap_or_else(|| panic!("missing value for {flag}"));
+        let number = || {
+            value
+                .parse()
+                .unwrap_or_else(|_| panic!("invalid value for {flag}: {value}"))
+        };
+        match flag.as_str() {
+            "--index" => index = Some(number()),
+            "--count" => count = Some(number()),
+            "--seed" => {
+                seed = value
+                    .parse()
+                    .unwrap_or_else(|_| panic!("invalid seed: {value}"))
+            }
+            "--database" => database_url = Some(value.clone()),
+            "--listen" => listen.clone_from(value),
+            _ => panic!("unknown shard option {flag}"),
+        }
+    }
+    let shard = Shard::new(
+        index.expect("--index is required"),
+        count.expect("--count is required"),
+    )
+    .expect("--index must be less than --count");
+    ShardOptions {
+        shard,
+        seed,
+        database_url: database_url.expect("--database is required"),
+        listen,
+        script,
+        attached,
+    }
+}
+
+#[tokio::main]
+async fn main() {
+    let args: Vec<String> = std::env::args().skip(1).collect();
+    let is_shard = args.first().is_some_and(|mode| mode == "shard");
+    let filter = tracing_subscriber::EnvFilter::try_from_default_env()
+        .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("picard=info,sim=info"));
+    // Shards keep stdout for announcing their address to the gateway.
+    tracing_subscriber::fmt()
+        .with_env_filter(filter)
+        .with_target(false)
+        .with_writer(std::io::stderr)
+        .init();
+
+    let (config_source, sim_script) = load_script();
+    if is_shard {
+        cluster::shard::run(shard_options(&args[1..], sim_script)).await;
+        return;
+    }
+
+    let seed = args.first().and_then(|s| s.parse().ok()).unwrap_or(42u64);
+    let shards = std::env::var("SIM_SHARDS").ok().map_or(3, |count| {
+        count
+            .parse::<usize>()
+            .ok()
+            .filter(|&count| count > 0)
+            .unwrap_or_else(|| panic!("SIM_SHARDS must be a positive number, not {count}"))
     });
+    let database_template = std::env::var("SHARD_DATABASE_URL")
+        .unwrap_or_else(|_| "sqlite://sim-shard-{shard}.db?mode=rwc".into());
+    assert!(
+        shards == 1 || database_template.contains("{shard}"),
+        "SHARD_DATABASE_URL must contain {{shard}} so that every shard gets its own database"
+    );
+    let addr = std::env::var("SIM_ADDR").unwrap_or_else(|_| "127.0.0.1:7878".into());
+    let listener = tokio::net::TcpListener::bind(&addr)
+        .await
+        .expect("failed to bind");
 
     let terrain_buf = Arc::new(Mutex::new(String::new()));
     let script = Arc::new(Mutex::new(config_source));
     let (tx, rx) = mpsc::unbounded_channel();
     let (state_tx, state_rx) = watch::channel(Arc::<str>::from("{}"));
     {
-        let (terrain_buf, script) = (Arc::clone(&terrain_buf), Arc::clone(&script));
-        std::thread::spawn(move || {
-            run_sim(
-                SimStart {
-                    seed,
-                    script: sim_script,
-                    snapshot: saved_snapshot,
-                },
-                rx,
-                &terrain_buf,
-                &script,
-                &state_tx,
-                &snapshot_tx,
-            );
-        });
+        let outputs = Outputs {
+            terrain: Arc::clone(&terrain_buf),
+            script: Arc::clone(&script),
+            state: state_tx,
+        };
+        let options = GatewayOptions {
+            shards,
+            seed,
+            database_template,
+        };
+        std::thread::Builder::new()
+            .name("gateway".into())
+            .spawn(move || cluster::gateway::conduct(&options, &outputs, rx))
+            .expect("failed to start the gateway");
     }
 
     let app = App {
@@ -395,9 +356,6 @@ async fn main() {
         .layer(middleware::from_fn(same_origin))
         .with_state(app);
 
-    let listener = tokio::net::TcpListener::bind(&addr)
-        .await
-        .expect("failed to bind");
-    println!("Simulation running at http://{addr}");
+    println!("Simulation running at http://{addr} with {shards} shards");
     axum::serve(listener, router).await.unwrap();
 }

src/world.rs

@@ -1,9 +1,5 @@
 mod actors;
 mod economy;
-#[allow(
-    dead_code,
-    reason = "the cluster gateway and shard nodes use the remote side"
-)]
 mod router;
 mod work;
 
@@ -11,10 +7,11 @@ use actors::{
     Activity, AnimalId, AnimalMsg, AnimalTurn, Area, BirthCost, Goods, Mailbox, Needs,
     PartnershipMsg, PersonEdit, PersonMsg, ReplyBody, RequestId,
 };
-use economy::{Cart, CartMsg, Job, Market, MarketMsg, Quote, Workplace, WorkplaceMsg};
+pub use economy::Quote;
+use economy::{Cart, CartMsg, Job, Market, MarketMsg, Workplace, WorkplaceMsg};
 use picard::actors::{ActorRef, ActorSystem, Sender, Tell};
 use router::{Arrival, LocalActor};
-pub use router::{EntityKey, Kind, Router, Shard};
+pub use router::{EntityKey, Kind, RemoteSender, Router, Shard};
 use rune::runtime::RuntimeContext;
 use rune::termcolor::Buffer;
 use rune::{Any, Context, Diagnostics, Module, Source, Sources, Unit, Vm};
@@ -29,6 +26,15 @@ pub const W: usize = 96;
 pub const H: usize = 64;
 pub const DEFAULT_SCRIPT: &str = include_str!("../config.rn");
 
+/// A random seed for a new world that differs from `current`.
+pub fn regenerated_seed(current: u64) -> u64 {
+    let mut seed = rand::random::<u64>();
+    if seed == current {
+        seed = seed.wrapping_add(1);
+    }
+    seed
+}
+
 #[derive(Clone, Copy, Debug, Deserialize, Serialize, rune::FromValue)]
 pub struct StageDuration {
     pub min_days: u32,
@@ -681,6 +687,7 @@ fn config_accessor_source() -> String {
 /// A compiled simulation script: `pub const` defaults plus `pub fn` behaviours.
 #[derive(Clone)]
 pub struct SimScript {
+    #[cfg_attr(not(test), expect(dead_code, reason = "tests read the source back"))]
     source: String,
     config: SimulationConfig,
     runtime: Arc<RuntimeContext>,
@@ -737,6 +744,7 @@ impl SimScript {
         })
     }
 
+    #[cfg(test)]
     pub fn source(&self) -> &str {
         &self.source
     }
@@ -2862,14 +2870,6 @@ impl World {
         world
     }
 
-    pub fn regenerated_seed(&self) -> u64 {
-        let mut seed = rand::random::<u64>();
-        if seed == self.seed {
-            seed = seed.wrapping_add(1);
-        }
-        seed
-    }
-
     pub fn seed(&self) -> u64 {
         self.seed
     }
@@ -2910,6 +2910,7 @@ impl World {
         })
     }
 
+    #[cfg(test)]
     pub fn script(&self) -> &str {
         self.script.source()
     }
@@ -3498,14 +3499,34 @@ impl World {
     }
 
     #[cfg(feature = "console")]
-    pub fn handle_picard_console_text(&self, request: &str) -> String {
-        self.actor_system.console().handle_text(request)
+    pub fn console(&self) -> picard::console::Console {
+        self.actor_system.console()
+    }
+
+    /// Lets the Picard console reach the actors of another shard as `@name`.
+    #[cfg(feature = "console")]
+    pub fn register_console_peer(
+        &self,
+        name: &str,
+        sender: picard::console::ConsolePeerSender,
+        query: picard::console::ConsolePeerQuery,
+    ) {
+        self.actor_system.register_console_peer(name, sender);
+        self.actor_system.register_console_peer_query(name, query);
     }
 }
 
 /// The interface the cluster uses to drive a shard.
-#[allow(dead_code, reason = "the cluster gateway and shard nodes use these")]
 impl World {
+    pub fn tick(&self) -> u64 {
+        self.tick
+    }
+
+    /// Takes in entities handed over by other shards, so that a snapshot includes them.
+    pub fn admit_arrivals(&mut self) {
+        self.receive_arrivals();
+    }
+
     pub fn shard(&self) -> Shard {
         self.shard
     }
@@ -4441,7 +4462,7 @@ mod tests {
     #[test]
     fn regenerated_seed_differs_from_current_seed() {
         let world = World::new(42);
-        assert_ne!(world.regenerated_seed(), world.seed());
+        assert_ne!(super::regenerated_seed(world.seed()), world.seed());
     }
 
     #[test]

tests/cluster.rs

@@ -0,0 +1,86 @@
+//! Runs the real binary as a gateway with two shard processes and checks that the UI sees one
+//! world spanning both shards.
+
+use std::io::{Read, Write};
+use std::net::{TcpListener, TcpStream};
+use std::process::{Child, Command, Stdio};
+use std::time::{Duration, Instant};
+
+struct Gateway(Child);
+
+impl Drop for Gateway {
+    fn drop(&mut self) {
+        let _ = self.0.kill();
+        let _ = self.0.wait();
+    }
+}
+
+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()?;
+    write!(
+        stream,
+        "GET {path} HTTP/1.1\r\nHost: {address}\r\nConnection: close\r\n\r\n"
+    )
+    .ok()?;
+    let mut response = String::new();
+    stream.read_to_string(&mut response).ok()?;
+    let (_, body) = response.split_once("\r\n\r\n")?;
+    serde_json::from_str(body).ok()
+}
+
+#[test]
+fn gateway_presents_the_shards_as_one_world() {
+    let address = {
+        let probe = TcpListener::bind("127.0.0.1:0").unwrap();
+        probe.local_addr().unwrap().to_string()
+    };
+    let directory = std::env::temp_dir().join(format!("sim-cluster-test-{}", std::process::id()));
+    std::fs::create_dir_all(&directory).unwrap();
+    let database = format!(
+        "sqlite://{}/shard-{{shard}}.db?mode=rwc",
+        directory.display()
+    );
+    let _gateway = Gateway(
+        Command::new(env!("CARGO_BIN_EXE_sim"))
+            .arg("5")
+            .env("SIM_SHARDS", "2")
+            .env("SIM_ADDR", &address)
+            .env("SHARD_DATABASE_URL", &database)
+            .env(
+                "SIM_CONFIG",
+                concat!(env!("CARGO_MANIFEST_DIR"), "/config.rn"),
+            )
+            .env("RUST_LOG", "warn")
+            .stdout(Stdio::null())
+            .spawn()
+            .unwrap(),
+    );
+
+    let deadline = Instant::now() + Duration::from_secs(120);
+    let mut first_tick = None;
+    loop {
+        assert!(Instant::now() < deadline, "the cluster never got going");
+        std::thread::sleep(Duration::from_millis(500));
+        let Some(state) = get(&address, "/api/state") else {
+            continue;
+        };
+        let Some(tick) = state["tick"].as_u64() else {
+            continue;
+        };
+        let start = *first_tick.get_or_insert(tick);
+        let shards = state["shards"].as_array().unwrap();
+        assert_eq!(shards.len(), 2);
+        let border = shards[1][0].as_u64().unwrap();
+        let people = state["people"].as_array().unwrap();
+        let west = people.iter().any(|p| p[1].as_u64() < Some(border));
+        let east = people.iter().any(|p| p[1].as_u64() >= Some(border));
+        if tick > start + 5 && west && east {
+            assert_eq!(state["homes"].as_array().unwrap().len(), 2);
+            break;
+        }
+    }
+    let terrain = get(&address, "/api/terrain").unwrap();
+    assert!(terrain["w"].as_u64().unwrap() > 0);
+    let _ = std::fs::remove_dir_all(&directory);
+}