dev: cache Docker dependencies with cargo-chef

Cook native and Wasm dependencies in reusable Docker layers. Let the frontend build script use the cooked Wasm target directory while preserving local build defaults.

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

.gitignore

@@ -1,4 +1,5 @@
 /target
+/sim-admin-password
 /.NOCOMMIT
 sim.db
 sim.db-shm

AGENTS.md

@@ -3,7 +3,6 @@
 ## AGENT RULES
 - you **MUST NOT** change `AGENTS.md` or `CLAUDE.md` unless explicitly asked
 - you **MUST** never do `git push`, unless you're explicitly told to
-- you **MUST NOT** commit to `main`. If the current branch is `main`, ask the user before making changes whether to create a new branch or to make the changes without committing
 - you **MUST** always run `cargo fmt --all` after changing file contents. Never pass formatting options on the command line (`--config`, `--edition`, ...) or invoke `rustfmt` directly on files; when present, formatting settings come from `rustfmt.toml` and `Cargo.toml`
 - you **MUST** use conventional commits when committing (`feat`, `fix`, `perf`, `refactor`, `chore`, `docs`, `test`, `dev`), with `!` and a `BREAKING CHANGE:` footer for API breaks
   - `dev` is used for tooling, debugging aids and test-flakiness work that doesn't change shipped behaviour
@@ -11,7 +10,7 @@
 - you **MUST** end commit messages with a `Co-Authored-By:` trailer naming yourself as the agent (product name only, no model or version), e.g. `Co-Authored-By: Claude <noreply@anthropic.com>`
 - you **MUST** inform the user about commits or any other changes to git, including the commit message
 - you **SHOULD** commit your changes after every step, unless the user tells you to not commit or a file with the name `.NOCOMMIT` exists in the project root. This applies to all changes, including docs, `AGENTS.md`, CI and config files, not just Rust code
-  - changes that affect compilation or tests (Rust code, `Cargo.toml`/`Cargo.lock`, build scripts, test data) may only be committed after `cargo clippy` and `cargo test` pass
+  - changes that affect compilation or tests (Rust code, `Cargo.toml`/`Cargo.lock`, build scripts, test data) may only be committed after `cargo clippy` and `just test` pass
   - other changes (Markdown, CI, editor or deployment config) don't need those checks; commit them right away
   - check the current branch (`git branch --show-current`) before every commit, not just at the start of the session
 - you **SHOULD** commit formatting-only changes separately from functional changes: if `cargo fmt --all` reformats code you didn't change, commit that first as `chore: cargo fmt`; do not stage unrelated user changes
@@ -21,7 +20,7 @@
 
 ```sh
 cargo build --workspace --all-targets
-cargo test --workspace --no-default-features
+just test
 cargo clippy --all -- -D warnings   # must stay clean
 cargo fmt --all
 ```

CHANGELOG.md

@@ -6,6 +6,10 @@ Notable changes to this project are documented here.
 
 ### Added
 
+- Add an HTTP Basic Auth-protected `/admin` path with a random password persisted in the app data volume.
+- Add an About dialog explaining the simulation, shards, actors, and persistence.
+- Docker Compose runs the gateway and each shard in its own container, with one persistent SQLite volume per shard.
+- Public clients are view-only: the UI hides editing controls and the server rejects remote writes; localhost clients retain editing access.
 - Expose the Picard actor console over the `/api/actors/ws` WebSocket endpoint.
 - The world can be split into shards that each own a strip of columns. People and animals that walk across a shard border are transferred with their inventory and stay reachable at the same address.
 - Jobs and trading: every shard founds markets, farms, woodcutters, quarries, hunters, toolmakers and cartwrights. Adults are hired, commute to their job site each day, work their shift, sell the goods at the market, get paid and go home.
@@ -22,9 +26,10 @@ Notable changes to this project are documented here.
 
 ### Changed
 
+- Cache native and Wasm dependencies with cargo-chef in Docker builds to speed up source-only rebuilds.
+- 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`.
 - 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.
-- Docker Compose creates one Postgres database per shard (`sim_shard_0` to `sim_shard_2`); existing `postgres_data` volumes must be recreated with `docker compose down -v`.

Cargo.lock

@@ -1939,7 +1939,7 @@ checksum = "9b4f627cb1b25917193a259e49bdad08f671f8d9708acfd5fe0a8c1455d87220"
 [[package]]
 name = "picard"
 version = "0.4.2"
-source = "git+https://github.com/picard-rs/picard?rev=f88b9cbf078512f726c4c47dbce842fda878388e#f88b9cbf078512f726c4c47dbce842fda878388e"
+source = "git+https://github.com/picard-rs/picard?rev=85b51c3da6bde1a3bfa3d3e9043578ce90f4a6c3#85b51c3da6bde1a3bfa3d3e9043578ce90f4a6c3"
 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=f88b9cbf078512f726c4c47dbce842fda878388e#f88b9cbf078512f726c4c47dbce842fda878388e"
+source = "git+https://github.com/picard-rs/picard?rev=85b51c3da6bde1a3bfa3d3e9043578ce90f4a6c3#85b51c3da6bde1a3bfa3d3e9043578ce90f4a6c3"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -2798,6 +2798,7 @@ name = "sim"
 version = "0.1.0"
 dependencies = [
  "axum",
+ "base64",
  "include_dir",
  "picard",
  "rand 0.8.8",
@@ -2805,6 +2806,7 @@ dependencies = [
  "serde",
  "serde_json",
  "sqlx",
+ "subtle",
  "tokio",
  "tracing",
  "tracing-subscriber",

Cargo.toml

@@ -7,6 +7,8 @@ edition = "2024"
 members = ["ui"]
 
 [workspace.dependencies]
+base64 = "0.22"
+subtle = "2.6"
 tracing = "0.1"
 tungstenite = "0.29"
 uuid = { version = "1", features = ["v4", "serde"] }
@@ -17,6 +19,8 @@ console = ["picard/console", "serde"]
 serde = ["picard/serde"]
 
 [dependencies]
+base64.workspace = true
+subtle.workspace = true
 rune = "0.14.2"
 axum = { version = "0.8", features = ["ws"] }
 tokio = { version = "1", features = ["rt-multi-thread", "macros", "net", "sync", "time"] }
@@ -25,7 +29,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 = "f88b9cbf078512f726c4c47dbce842fda878388e", features = [
+picard = { git = "https://github.com/picard-rs/picard", rev = "85b51c3da6bde1a3bfa3d3e9043578ce90f4a6c3", features = [
     "remoting",
     "stats",
     "tracing",

Dockerfile

@@ -1,9 +1,23 @@
-FROM rust:1-bookworm AS builder
+FROM rust:1-bookworm AS chef
 
 WORKDIR /src
-RUN rustup target add wasm32-unknown-unknown
+RUN cargo install cargo-chef --version 0.1.78 --locked \
+    && rustup target add wasm32-unknown-unknown
+
+FROM chef AS planner
+
 COPY . .
-RUN cargo build --release --bin sim
+RUN cargo chef prepare --recipe-path recipe.json
+
+FROM chef AS builder
+
+ENV SIM_UI_TARGET_DIR=/src/target/ui
+COPY --from=planner /src/recipe.json recipe.json
+RUN cargo chef cook --release --locked --recipe-path recipe.json --bin sim
+RUN cargo chef cook --release --locked --recipe-path recipe.json \
+    --package ui --target wasm32-unknown-unknown --target-dir /src/target/ui
+COPY . .
+RUN cargo build --release --locked --bin sim
 
 FROM debian:bookworm-slim
 

Justfile

@@ -2,10 +2,17 @@ default:
     @just --list
 
 run:
-    docker compose down && docker compose up --build -d
+    docker compose build && docker compose up -d --remove-orphans
 
 stop:
-    docker compose down
+    docker compose down --remove-orphans
+
+admin:
+    docker compose exec -T app cat /data/admin-password
+
+test:
+    if ! cargo nextest --version >/dev/null 2>&1; then cargo install cargo-nextest --locked; fi
+    cargo nextest run --workspace
 
 dev:
     SHARD_DATABASE_URL="sqlite://sim-shard-{shard}.db?mode=rwc" cargo run

README.md

@@ -6,9 +6,15 @@ Every person, rabbit, wolf, and partnership is a Picard actor that owns its stat
 
 ## Cluster
 
-The world is split into vertical strips of columns called shards. Each shard is a separate process with its own Picard actor system and its own database. `sim` starts as a gateway: it spawns `SIM_SHARDS` (default 3) copies of itself with the `shard` subcommand, steps them in lockstep, and merges their views so the UI sees one world. Edits from the UI are routed to the shard that owns them.
+The world is split into vertical strips of columns called shards. Each shard is a separate process with its own Picard actor system and its own database. `sim` starts as a gateway: by default, it spawns `SIM_SHARDS` (default 3) copies of itself with the `shard` subcommand; when `SIM_SHARD_ADDRS` is set, it connects to externally managed shards instead. The gateway steps shards in lockstep and merges their views so the UI sees one world. Edits from the UI are routed to the shard that owns them.
 
-People, animals and carts that cross a shard border are transferred with their whole state and inventory to the neighbouring shard's actor system. They keep their address, and messages to them are forwarded transparently, so actors never know which shard they run on. Shards talk to each other directly over WebSockets (`/remote` for actor messages and transfers, `/control` for the gateway protocol).
+People, animals and carts that cross a shard border are migrated with their whole state and inventory into the neighbouring shard's Picard actor system. A cluster-wide directory maps each entity address to its current shard and actor path, so world messages continue to reach it after migration without actor-owned forwarding timers or transfer messages. Shards connect directly through Picard's authenticated remoting endpoint (`/remoting`); the gateway continues to use `/control` for simulation coordination. The gateway generates a shared token when it launches shard processes; external shard deployments supply the same `SIM_CLUSTER_TOKEN` to the gateway and every shard.
+
+## Public access
+
+Clients outside localhost can view the simulation but cannot edit it: the UI hides editing controls, and the server rejects remote write requests and WebSocket edit commands. When exposing the app through a reverse proxy, keep the published port bound to host loopback and ensure the proxy sets `X-Forwarded-For` or `X-Real-IP`. Compose enables `SIM_TRUST_PROXY_HEADERS=1` for this setup. Docker-host requests to `localhost` remain editable, while forwarded requests are classified by their client IP.
+
+The `/admin` path on the same host and port provides authenticated editing. On first start, the app generates a random password and persists it in the `app_data` Docker volume; it stays the same across container restarts. The default username is `admin` (override with `SIM_ADMIN_USER`). Run `just admin` to print the password from the running app container. The normal `/` routes remain view-only. Keep the admin route behind HTTPS through Caddy because Basic Auth credentials are only encoded, not encrypted, without TLS. Removing volumes with `docker compose down -v` also removes this password, so the next start generates a new one.
 
 ## Economy
 
@@ -16,9 +22,9 @@ Every shard founds markets and workplaces: farms, woodcutters, quarries, hunters
 
 ## Actor Console
 
-The server enables Picard's console by default and exposes its JSON request/response protocol over `ws://127.0.0.1:7878/api/actors/ws`. The connection reaches shard 0; use `?shard=N` to connect to shard `N`. Every shard registers the others as console peers named `shard-N`: `peers` lists them, `@shard-1 actors` forwards a command, and `send @shard-1 PATH JSON` sends a message to a remote actor. Send one JSON request per text frame; each receives one response. For example, `{"id":1,"op":"actors"}` lists actors, and `{"id":2,"op":"inspect","path":"/user/person-1"}` inspects one (actors are named `person-<id>`, `rabbit-<id>`, `wolf-<id>`, and `partnership-<id>`). Other operations include `messages`, `pause`, `resume`, `clear`, `stop`, `restart`, and `send`. Console requests are answered by the shard's actor system without waiting for the simulation thread. `cargo run --no-default-features` disables the endpoint; `--features console` enables it explicitly.
+The server enables Picard's console by default and exposes its JSON request/response protocol over `ws://127.0.0.1:7878/api/actors/ws`. The connection reaches shard 0; use `?shard=N` to connect to shard `N`. Every shard registers the others as console peers named `shard-N`: `peers` lists them, `@shard-1 actors` forwards a command, and `send @shard-1 PATH JSON` sends a message to a remote actor through Picard remoting. Send one JSON request per text frame; each receives one response. For example, `{"id":1,"op":"actors"}` lists actors, and `{"id":2,"op":"inspect","path":"/user/person-1"}` inspects one (actors are named `person-<id>`, `rabbit-<id>`, `wolf-<id>`, and `partnership-<id>`). Other operations include `messages`, `pause`, `resume`, `clear`, `stop`, `restart`, and `send`. Console requests are handled by the selected shard's Picard console. `cargo run --no-default-features` disables the endpoint; `--features console` enables it explicitly.
 
-The simulation script is the root [config.rn](./config.rn), a single Rune file. Its `pub const` items hold defaults and tuning values: map dimensions and terrain generation, initial population and resources, calendar and life-stage defaults, hunger, gathering, construction, partnerships and births, animal behavior, and edit limits. Its `pub fn person`, `pub fn rabbit`, and `pub fn wolf` items are the agent behaviours, and they can use the constants directly. Every constant the server reads must be defined; missing or invalid values are reported at startup. `SIM_CONFIG` can point to a different Rune file. Docker Compose mounts the repository's `config.rn` read-only into the app container.
+The simulation script is the root [config.rn](./config.rn), a single Rune file. Its `pub const` items hold defaults and tuning values: map dimensions and terrain generation, initial population and resources, calendar and life-stage defaults, hunger, gathering, construction, partnerships and births, animal behavior, and edit limits. Its `pub fn person`, `pub fn rabbit`, and `pub fn wolf` items are the agent behaviours, and they can use the constants directly. Every constant the server reads must be defined; missing or invalid values are reported at startup. `SIM_CONFIG` can point to a different Rune file. Docker Compose mounts the repository's `config.rn` read-only into the gateway and each shard container.
 
 The in-browser script editor edits the same file contents. Applied edits take effect immediately for behaviours and tuning values, while map dimensions, initial population, and calendar defaults apply when the world is reset. Editor changes are not written back to disk: on restart, `config.rn` from disk replaces them, even when a saved world snapshot is restored.
 
@@ -59,29 +65,34 @@ just
 Useful recipes:
 
 ```sh
-just run  # Stop Compose services, then build and start the app with Postgres
+just run  # Build images, then recreate the gateway and SQLite shards
-just stop # Stop Compose services; keep the database volume
+just stop # Stop Compose services; keep shard data volumes
 just dev  # Run locally with SQLite
 ```
 
-### Run with Docker Compose and Postgres
+### Run with Docker Compose and SQLite shards
 
-Install Docker with the Compose plugin, then run `just run`. This stops any existing Compose stack before rebuilding and starting the app with Postgres. You can also use the equivalent Docker command directly:
+Install Docker with the Compose plugin, then run `just run`. It builds the images while the current stack is still running, then asks Compose to recreate changed services and remove obsolete service containers. This minimizes downtime by keeping the build outside the service interruption. The stack consists of one gateway container and three shard containers; each shard stores its snapshot in its own SQLite database and persistent volume. You can also use the equivalent Docker commands directly:
 
 ```sh
-docker compose up --build -d
+docker compose build
+docker compose up -d --remove-orphans
 ```
 
-Open [http://localhost:7878](http://localhost:7878). Postgres creates one database per shard (`sim_shard_0` to `sim_shard_2`) from [docker/postgres-init.sql](./docker/postgres-init.sql) when the volume is first created; add databases there when raising `SIM_SHARDS`, and recreate the volume after changing the script. Compose waits for Postgres to become healthy before starting the app. Postgres data is stored in the `postgres_data` volume and survives container restarts, `just stop`, and `docker compose down`.
+The Docker build uses cargo-chef to cache native server and Wasm frontend dependencies separately. Source-only changes reuse these dependency layers; dependency or toolchain changes rebuild them. Inside the builder, `SIM_UI_TARGET_DIR` points the frontend build script at the cooked Wasm artifacts. Local builds retain their default frontend output directory unless this variable is set.
 
-To remove the database volume and all saved simulation data as well:
+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:
 
 ```sh
 docker compose down -v
 ```
 
-The Compose file uses development-only database credentials. Change them before exposing the service beyond a trusted local environment.
-
 ## Database Configuration
 
 Each shard selects its database through `SHARD_DATABASE_URL`. `{shard}` in the URL is replaced with the shard index and is required when there is more than one shard:
@@ -97,9 +108,9 @@ For example, run locally against a reachable Postgres server that has the shard
 SHARD_DATABASE_URL='postgres://USER:PASSWORD@localhost:5432/sim_shard_{shard}' cargo run
 ```
 
-Every shard saves its part of the world as a snapshot about once per second and restores it at startup. SQLite is convenient for local testing; Compose configures the app to use Postgres.
+Every shard saves its part of the world as a snapshot about once per second and restores it at startup. Compose uses a separate SQLite database and volume for each shard; Postgres remains available for deployments that configure it explicitly.
 
-`SIM_SHARDS` sets the number of shards (default 3). `SIM_ADDR` sets the gateway's listen address. It defaults to `127.0.0.1:7878`; containers set it to `0.0.0.0:7878`.
+`SIM_SHARDS` sets the number of shards (default 3). `SIM_SHARD_ADDRS` optionally provides comma-separated shard control addresses in index order; set `SIM_CLUSTER_TOKEN` to the same cluster token on the gateway and every external shard. `SIM_ADDR` sets the gateway's listen address. It defaults to `127.0.0.1:7878`; containers set it to `0.0.0.0:7878`.
 
 ## Project Layout
 
@@ -109,7 +120,7 @@ Every shard saves its part of the world as a snapshot about once per second and
 - `src/storage.rs` provides SQLite and Postgres snapshot storage.
 - `config.rn` is the Rune simulation script: `pub const` defaults and tuning values plus `pub fn` behaviours.
 - `ui/` contains the Leptos frontend, compiled to Wasm by `build.rs`.
-- `Dockerfile`, `compose.yaml`, `docker/postgres-init.sql` and `Justfile` provide container and developer workflows.
+- `Dockerfile`, `compose.yaml`, and `Justfile` provide container and developer workflows.
 
 ## Tests
 

build.rs

@@ -8,6 +8,7 @@ fn main() {
     for p in ["ui/src", "ui/js", "ui/style.css", "ui/Cargo.toml"] {
         println!("cargo:rerun-if-changed={p}");
     }
+    println!("cargo:rerun-if-env-changed=SIM_UI_TARGET_DIR");
     let out = PathBuf::from(env::var_os("OUT_DIR").expect("OUT_DIR"));
     let pkg = out.join("ui-pkg");
     let release = env::var("PROFILE").is_ok_and(|p| p == "release");
@@ -15,7 +16,9 @@ fn main() {
     ensure_wasm_target();
 
     // A separate target dir avoids deadlocking on the outer build's lock.
-    let target_dir = out.join("ui-target");
+    let target_dir = env::var_os("SIM_UI_TARGET_DIR")
+        .map(PathBuf::from)
+        .unwrap_or_else(|| out.join("ui-target"));
     let mut cargo = Command::new(env::var_os("CARGO").unwrap_or_else(|| "cargo".into()));
     cargo
         .args([

compose.yaml

@@ -1,35 +1,74 @@
 services:
   app:
+    cpus: 2
     build: .
     environment:
       SIM_ADDR: 0.0.0.0:7878
       SIM_CONFIG: /etc/sim/config.rn
       SIM_SHARDS: "3"
-      SHARD_DATABASE_URL: postgres://sim:sim@db:5432/sim_shard_{shard}
+      SIM_SHARD_ADDRS: shard-0:9000,shard-1:9000,shard-2:9000
+      SIM_CLUSTER_TOKEN: ${SIM_CLUSTER_TOKEN:-compose-dev-token}
+      SIM_TRUST_PROXY_HEADERS: "1"
+      SIM_ADMIN_USER: ${SIM_ADMIN_USER:-admin}
+      SIM_ADMIN_PASSWORD_FILE: /data/admin-password
     volumes:
       - ./config.rn:/etc/sim/config.rn:ro
+      - app_data:/data
     ports:
-      - "7878:7878"
+      - "127.0.0.1:7878:7878"
     depends_on:
-      db:
+      shard-0:
-        condition: service_healthy
+        condition: service_started
+      shard-1:
+        condition: service_started
+      shard-2:
+        condition: service_started
     restart: unless-stopped
 
-  db:
+  shard-0:
-    image: postgres:17-alpine
+    build: .
+    command: ["sim", "shard", "--index", "0", "--count", "3", "--seed", "42", "--listen", "0.0.0.0:9000"]
     environment:
-      POSTGRES_DB: sim
+      SIM_CONFIG: /etc/sim/config.rn
-      POSTGRES_USER: sim
+      SIM_CLUSTER_TOKEN: ${SIM_CLUSTER_TOKEN:-compose-dev-token}
-      POSTGRES_PASSWORD: sim
+      SIM_REMOTING_BIND: 0.0.0.0:9001
-    healthcheck:
+      SIM_REMOTING_ADVERTISE: ws://shard-0:9001/remoting
-      test: [ "CMD-SHELL", "pg_isready -U sim -d sim" ]
+      SHARD_DATABASE_URL: sqlite:///data/shard-0.db?mode=rwc
-      interval: 5s
-      timeout: 3s
-      retries: 10
     volumes:
-      - postgres_data:/var/lib/postgresql/data
+      - ./config.rn:/etc/sim/config.rn:ro
-      - ./docker/postgres-init.sql:/docker-entrypoint-initdb.d/shards.sql:ro
+      - shard_0_data:/data
+    restart: unless-stopped
+
+  shard-1:
+    build: .
+    command: ["sim", "shard", "--index", "1", "--count", "3", "--seed", "42", "--listen", "0.0.0.0:9000"]
+    environment:
+      SIM_CONFIG: /etc/sim/config.rn
+      SIM_CLUSTER_TOKEN: ${SIM_CLUSTER_TOKEN:-compose-dev-token}
+      SIM_REMOTING_BIND: 0.0.0.0:9001
+      SIM_REMOTING_ADVERTISE: ws://shard-1:9001/remoting
+      SHARD_DATABASE_URL: sqlite:///data/shard-1.db?mode=rwc
+    volumes:
+      - ./config.rn:/etc/sim/config.rn:ro
+      - shard_1_data:/data
+    restart: unless-stopped
+
+  shard-2:
+    build: .
+    command: ["sim", "shard", "--index", "2", "--count", "3", "--seed", "42", "--listen", "0.0.0.0:9000"]
+    environment:
+      SIM_CONFIG: /etc/sim/config.rn
+      SIM_CLUSTER_TOKEN: ${SIM_CLUSTER_TOKEN:-compose-dev-token}
+      SIM_REMOTING_BIND: 0.0.0.0:9001
+      SIM_REMOTING_ADVERTISE: ws://shard-2:9001/remoting
+      SHARD_DATABASE_URL: sqlite:///data/shard-2.db?mode=rwc
+    volumes:
+      - ./config.rn:/etc/sim/config.rn:ro
+      - shard_2_data:/data
     restart: unless-stopped
 
 volumes:
-  postgres_data:
+  app_data:
+  shard_0_data:
+  shard_1_data:
+  shard_2_data:

src/cluster/gateway.rs

@@ -1,4 +1,4 @@
-//! The gateway: starts the shard nodes, steps them in lockstep and presents their parts of the
+//! The gateway: connects to 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};
@@ -39,6 +39,10 @@ pub struct GatewayOptions {
     pub seed: u64,
     /// Database URL of each shard; `{shard}` is replaced by the shard's index.
     pub database_template: String,
+    /// Control addresses of externally managed shard processes, in index order.
+    pub external_shards: Option<Vec<String>>,
+    /// Shared cluster token; generated when shards are managed by this gateway.
+    pub cluster_token: Option<String>,
 }
 
 /// Where the gateway publishes the unified world for the HTTP handlers.
@@ -71,7 +75,7 @@ struct ShardProcess {
 }
 
 impl ShardProcess {
-    fn start(index: usize, options: &GatewayOptions) -> Result<Self, String> {
+    fn start(index: usize, options: &GatewayOptions, cluster_token: &str) -> Result<Self, String> {
         let executable = std::env::current_exe().map_err(|e| e.to_string())?;
         let database = options
             .database_template
@@ -81,6 +85,7 @@ impl ShardProcess {
             .args(["--index", &index.to_string()])
             .args(["--count", &options.shards.to_string()])
             .args(["--seed", &options.seed.to_string()])
+            .args(["--cluster-token", cluster_token])
             .args(["--database", &database])
             .stdin(Stdio::piped())
             .stdout(Stdio::piped())
@@ -128,8 +133,20 @@ struct ShardLink {
 impl ShardLink {
     fn connect(index: usize, address: &str) -> Result<Self, String> {
         let url = format!("ws://{address}/control");
-        let (socket, _) = tungstenite::connect(&url)
+        let started = Instant::now();
-            .map_err(|e| format!("failed to connect to shard {index} at {url}: {e}"))?;
+        let (socket, _) = loop {
+            match tungstenite::connect(&url) {
+                Ok(connection) => break connection,
+                Err(_) if started.elapsed() < SHARD_TIMEOUT => {
+                    std::thread::sleep(Duration::from_millis(250));
+                }
+                Err(error) => {
+                    return Err(format!(
+                        "failed to connect to shard {index} at {url}: {error}"
+                    ));
+                }
+            }
+        };
         if let MaybeTlsStream::Plain(stream) = socket.get_ref() {
             let _ = stream.set_nodelay(true);
             let _ = stream.set_read_timeout(Some(SHARD_TIMEOUT));
@@ -174,6 +191,7 @@ impl ShardLink {
 
 struct Ready {
     system_id: Uuid,
+    remote_address: String,
     seed: u64,
     tick: u64,
     restored: bool,
@@ -183,11 +201,13 @@ fn ready(reply: ControlReply) -> Result<Ready, String> {
     match reply {
         ControlReply::Ready {
             system_id,
+            remote_address,
             seed,
             tick,
             restored,
         } => Ok(Ready {
             system_id,
+            remote_address,
             seed,
             tick,
             restored,
@@ -202,10 +222,11 @@ fn unexpected(reply: &ControlReply) -> String {
 
 struct Cluster {
     links: Vec<ShardLink>,
-    processes: Vec<ShardProcess>,
+    _processes: Vec<ShardProcess>,
     seed: u64,
+    cluster_token: String,
     /// Directory changes to publish with the next step.
-    directory: Vec<(EntityKey, Option<usize>)>,
+    directory: Vec<(EntityKey, Option<(usize, String)>)>,
     board: Vec<Quote>,
     paused: bool,
     speed: u64,
@@ -215,18 +236,40 @@ struct Cluster {
 
 impl Cluster {
     fn start(options: &GatewayOptions) -> Result<Self, String> {
+        let cluster_token = options
+            .cluster_token
+            .clone()
+            .unwrap_or_else(|| Uuid::new_v4().to_string());
+        let (processes, links) = if let Some(addresses) = &options.external_shards {
+            if addresses.len() != options.shards {
+                return Err(format!(
+                    "SIM_SHARD_ADDRS lists {} shards, but SIM_SHARDS is {}",
+                    addresses.len(),
+                    options.shards
+                ));
+            }
+            let links = addresses
+                .iter()
+                .enumerate()
+                .map(|(index, address)| ShardLink::connect(index, address))
+                .collect::<Result<Vec<_>, _>>()?;
+            (Vec::new(), links)
+        } else {
             let processes = (0..options.shards)
-            .map(|index| ShardProcess::start(index, options))
+                .map(|index| ShardProcess::start(index, options, &cluster_token))
                 .collect::<Result<Vec<_>, _>>()?;
             let links = processes
                 .iter()
                 .enumerate()
                 .map(|(index, process)| ShardLink::connect(index, &process.address))
                 .collect::<Result<Vec<_>, _>>()?;
+            (processes, links)
+        };
         let mut cluster = Self {
             links,
-            processes,
+            _processes: processes,
             seed: options.seed,
+            cluster_token,
             directory: Vec::new(),
             board: Vec::new(),
             paused: false,
@@ -288,9 +331,9 @@ impl Cluster {
     fn connect_peers(&mut self, readies: &[Ready]) -> Result<(), String> {
         let peers: Vec<Peer> = readies
             .iter()
-            .zip(&self.processes)
+            .map(|ready| Peer {
-            .map(|(ready, process)| Peer {
+                address: ready.remote_address.clone(),
-                address: process.address.clone(),
+                token: self.cluster_token.clone(),
                 system_id: ready.system_id,
             })
             .collect();
@@ -300,8 +343,10 @@ impl Cluster {
             let ControlReply::Roster(keys) = reply else {
                 return Err(unexpected(&reply));
             };
-            self.directory
+            self.directory.extend(
-                .extend(keys.into_iter().map(|key| (key, Some(index))));
+                keys.into_iter()
+                    .map(|(key, path)| (key, Some((index, path)))),
+            );
         }
         self.board.clear();
         Ok(())
@@ -320,10 +365,18 @@ impl Cluster {
                 return Err(unexpected(&reply));
             };
             self.board.extend(quotes);
-            self.directory
+            self.directory.extend(
-                .extend(report.moved.into_iter().map(|(key, to)| (key, Some(to))));
+                report
-            self.directory
+                    .moved
-                .extend(report.arrived.into_iter().map(|key| (key, Some(index))));
+                    .into_iter()
+                    .map(|(key, to, path)| (key, Some((to, path)))),
+            );
+            self.directory.extend(
+                report
+                    .arrived
+                    .into_iter()
+                    .map(|(key, path)| (key, Some((index, path)))),
+            );
             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.

src/cluster/protocol.rs

@@ -4,12 +4,14 @@ use crate::world::{EntityKey, Quote, StepReport};
 use serde::{Deserialize, Serialize};
 use uuid::Uuid;
 
-/// Where a shard node accepts entity messages from the other shards.
+/// How another shard can authenticate and connect to this shard's Picard remoting listener.
 #[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Eq)]
 pub struct Peer {
-    /// The `host:port` of the shard node, which serves `/remote` and `/console`.
+    /// Picard remote endpoint.
     pub address: String,
-    /// The current actor system of the shard, which remote requests are addressed to.
+    /// Shared authentication token for this cluster.
+    pub token: String,
+    /// The current actor system of the shard.
     pub system_id: Uuid,
 }
 
@@ -21,7 +23,7 @@ pub enum Control {
     Peers(Vec<Peer>),
     /// Runs one step after applying the gateway's directory changes and market board.
     Step {
-        directory: Vec<(EntityKey, Option<usize>)>,
+        directory: Vec<(EntityKey, Option<(usize, String)>)>,
         board: Vec<Quote>,
     },
     /// Asks for the shard's part of the world, and for the terrain when `terrain` is set.
@@ -44,6 +46,7 @@ pub enum Control {
 pub enum ControlReply {
     Ready {
         system_id: Uuid,
+        remote_address: String,
         seed: u64,
         tick: u64,
         restored: bool,
@@ -57,7 +60,7 @@ pub enum ControlReply {
         state: String,
         terrain: Option<String>,
     },
-    Roster(Vec<EntityKey>),
+    Roster(Vec<(EntityKey, String)>),
     Script(Result<(), String>),
     Console(String),
     Failed(String),

src/cluster/shard.rs

@@ -1,17 +1,14 @@
-//! A shard node: simulates one strip of the world, answers the gateway on `/control` and takes
+//! A shard node: simulates one strip of the world, answers the gateway on `/control`, and accepts
-//! entity messages from the other shards on `/remote`.
+//! actor messages and migrations from other shards through Picard remoting.
 
 use super::protocol::{Control, ControlReply, Peer};
-use super::transport::{self, ClusterTransport};
 use crate::storage::Storage;
-use crate::world::{Router, Shard, SimScript, World};
+use crate::world::{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.
@@ -20,6 +17,7 @@ pub const LISTENING: &str = "SHARD_LISTENING";
 pub struct ShardOptions {
     pub shard: Shard,
     pub seed: u64,
+    pub cluster_token: String,
     pub database_url: String,
     pub listen: String,
     pub script: SimScript,
@@ -28,38 +26,17 @@ pub struct ShardOptions {
 }
 
 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,
+        cluster_token,
         database_url,
         listen,
         script,
@@ -78,19 +55,18 @@ pub async fn run(options: ShardOptions) {
     let (saves, snapshots) = watch::channel(None::<String>);
     tokio::spawn(save_snapshots(storage, snapshots));
 
-    let current = Current::default();
     let (commands, receiver) = mpsc::unbounded_channel();
-    {
+    let simulation = std::thread::Builder::new()
-        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);
+            let (mut world, restored) = open_world(saved.as_deref(), seed, script, shard);
-                current.publish(&world);
+            world
-                run_world(world, restored, receiver, &current, &saves);
+                .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());
@@ -98,12 +74,8 @@ pub async fn run(options: ShardOptions) {
         });
     }
 
-    let app = axum::Router::new()
+    let app = axum::Router::new().route("/control", get(control));
-        .route("/control", get(control))
+    let app = app.with_state(Node { commands });
-        .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));
@@ -114,9 +86,29 @@ pub async fn run(options: ShardOptions) {
         let _ = stdout.flush();
     }
     tracing::info!(shard = shard.index, %address, "shard listening");
-    axum::serve(listener, app)
+    tokio::select! {
-        .await
+        result = stopped => {
-        .expect("shard server failed");
+            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.
@@ -149,8 +141,8 @@ fn run_world(
     mut world: World,
     mut restored: bool,
     mut commands: mpsc::UnboundedReceiver<Command>,
-    current: &Current,
     saves: &watch::Sender<Option<String>>,
+    cluster_token: &str,
 ) {
     while let Some((control, reply)) = commands.blocking_recv() {
         let answer = match control {
@@ -191,8 +183,10 @@ fn run_world(
             Control::Script(source) => ControlReply::Script(world.set_script(&source)),
             Control::Restart { seed } => {
                 world = world.restarted(seed);
+                world
+                    .start_remoting(cluster_token)
+                    .unwrap_or_else(|error| panic!("failed to restart shard remoting: {error}"));
                 restored = false;
-                current.publish(&world);
                 ready(&world, restored)
             }
             Control::Console(request) => console(&world, &request),
@@ -204,6 +198,9 @@ fn run_world(
 fn ready(world: &World, restored: bool) -> ControlReply {
     ControlReply::Ready {
         system_id: world.router().system_id(),
+        remote_address: world
+            .remote_address()
+            .expect("the shard remote listener is started"),
         seed: world.seed(),
         tick: world.tick(),
         restored,
@@ -211,25 +208,9 @@ fn ready(world: &World, restored: bool) -> ControlReply {
 }
 
 fn connect(world: &World, peers: Vec<Peer>) {
-    let this = world.shard().index;
+    world
-    let transport = ClusterTransport::new(this, peers.clone());
+        .connect_peers(peers)
-    world.router().set_remote(Arc::clone(&transport) as _);
+        .unwrap_or_else(|error| panic!("failed to connect shard peers: {error}"));
-    #[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")]
@@ -273,58 +254,3 @@ async fn serve_control(mut socket: WebSocket, node: Node) {
         }
     }
 }
-
-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

@@ -1,255 +1,85 @@
-//! Delivers entity messages between shard nodes over persistent WebSockets.
+//! Adapts Picard's authenticated cluster transport to the world's location router.
-//!
-//! 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 crate::world::RemoteSender;
-use picard::remoting::{ActorAddress, RemoteRequest, RemoteResponse};
+use picard::actors::ActorSystem;
-use std::collections::VecDeque;
+use picard::remoting::{
-use std::net::TcpStream;
+    ActorAddress, ClusterTransport as PicardClusterTransport, RemoteConnection, RemotePeer,
-use std::sync::{Arc, Condvar, Mutex};
+};
-use std::thread::JoinHandle;
+use serde_json::Value;
-use std::time::{Duration, Instant};
+use std::collections::HashMap;
-use tungstenite::stream::MaybeTlsStream;
+use std::sync::Arc;
-use tungstenite::{Message, WebSocket};
+use uuid::Uuid;
 
-const MAX_BATCH: usize = 512;
+#[derive(Clone)]
-const FLUSH_TIMEOUT: Duration = Duration::from_secs(10);
+pub(crate) struct PeerConnection {
-const CONNECT_ATTEMPTS: u32 = 50;
+    pub system_id: Uuid,
-
+    pub endpoint: String,
-#[derive(Default)]
+    pub connection: RemoteConnection,
-struct Queue {
-    waiting: VecDeque<RemoteRequest>,
-    /// Messages queued or sent but not yet answered.
-    unanswered: usize,
-    closed: bool,
 }
 
-struct Link {
+pub(crate) struct ClusterTransport {
-    queue: Mutex<Queue>,
+    peers: HashMap<usize, PeerConnection>,
-    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 connect(
-    pub fn new(this: usize, peers: Vec<Peer>) -> Arc<Self> {
+        system: &ActorSystem,
-        let connections = peers
+        this: usize,
-            .into_iter()
+        peers: Vec<Peer>,
-            .enumerate()
+        transport: Arc<PicardClusterTransport>,
-            .map(|(index, peer)| {
+    ) -> Result<Arc<Self>, String> {
-                (index != this).then(|| {
+        let mut connections = HashMap::new();
-                    let link = Arc::new(Link {
+        for (index, peer) in peers.into_iter().enumerate() {
-                        queue: Mutex::default(),
+            if index == this {
-                        changed: Condvar::new(),
+                continue;
-                    });
-                    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),
             }
-                })
+            let name = format!("shard-{index}");
-            })
+            let remote = RemotePeer::new(name.clone(), peer.address.clone(), peer.token)
-            .collect();
+                .map_err(|error| error.to_string())?;
-        Arc::new(Self { connections })
+            let connection = system.connect_remote_with(remote, transport.clone());
+            #[cfg(feature = "console")]
+            system.register_console_remote_peer(name, connection.clone());
+            connections.insert(
+                index,
+                PeerConnection {
+                    system_id: peer.system_id,
+                    endpoint: peer.address,
+                    connection,
+                },
+            );
+        }
+        Ok(Arc::new(Self { peers: connections }))
+    }
+
+    pub fn peers(&self) -> HashMap<usize, PeerConnection> {
+        self.peers.clone()
     }
 }
 
 impl RemoteSender for ClusterTransport {
-    fn send(&self, shard: usize, path: &str, payload: serde_json::Value) {
+    fn send(&self, shard: usize, path: &str, payload: Value) {
-        let Some(Some(connection)) = self.connections.get(shard) else {
+        let Some(peer) = self.peers.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 destination = ActorAddress::remote(peer.system_id, peer.endpoint.clone(), path);
-        let mut queue = connection.link.lock();
+        match picard::remoting::block_on(peer.connection.send(destination, None, payload)) {
-        queue
+            Ok(response) if response.accepted => {}
-            .waiting
+            Ok(response) => tracing::warn!(
-            .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,
+                path,
-                        "shard flush timed out"
+                error = ?response.error,
-                    );
+                "Picard rejected a remote actor message"
-                    break;
+            ),
-                }
+            Err(error) => tracing::warn!(
-                queue = connection
+                shard,
-                    .link
+                path,
-                    .changed
+                %error,
-                    .wait_timeout(queue, left)
+                "failed to send a remote actor message"
-                    .expect("transport queue lock")
+            ),
-                    .0;
         }
     }
-    }
-}
 
-impl Drop for ClusterTransport {
+    fn flush(&self) {}
-    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

@@ -4,14 +4,21 @@ mod world;
 
 use axum::Router;
 use axum::extract::ws::{Message, WebSocket, WebSocketUpgrade};
-use axum::extract::{DefaultBodyLimit, Path, Request, State};
+use axum::extract::{ConnectInfo, DefaultBodyLimit, OriginalUri, Path, Request, State};
-use axum::http::{StatusCode, header};
+use axum::http::{HeaderMap, StatusCode, header};
 use axum::middleware::{self, Next};
-use axum::response::{Html, IntoResponse, Response};
+use axum::response::{Html, IntoResponse, Json, Response};
 use axum::routing::{get, post};
+use base64::Engine;
+use base64::engine::general_purpose::STANDARD as BASE64;
 use cluster::gateway::{GatewayOptions, Msg, Outputs};
 use cluster::shard::ShardOptions;
 use include_dir::{Dir, include_dir};
+use rand::Rng;
+use std::fs::{self, OpenOptions};
+use std::io::{self, Write};
+use std::net::{IpAddr, SocketAddr};
+use std::path::{Path as FilePath, PathBuf};
 use std::sync::{Arc, Mutex};
 use std::time::Duration;
 use tokio::sync::{mpsc, oneshot, watch};
@@ -24,21 +31,44 @@ static UI: Dir = include_dir!("$OUT_DIR/ui-pkg");
 const INDEX: &str = r#"<!doctype html>
 <html><head><meta charset="utf-8"><meta name="viewport" content="width=device-width, initial-scale=1"><title>World Simulation</title></head>
 <body style="background:#111;color:#ddd;font:13px monospace"><script type="module">
-try { const m = await import('/pkg/ui.js'); await m.default({ module_or_path: '/pkg/ui_bg.wasm' }); }
+try {
+    const base = location.pathname === '/admin' || location.pathname.startsWith('/admin/') ? '/admin' : '';
+    const m = await import(`${base}/pkg/ui.js`);
+    await m.default({ module_or_path: `${base}/pkg/ui_bg.wasm` });
+}
 catch (e) { document.body.textContent = 'Failed to load the UI: ' + e; }
 </script></body></html>"#;
 
+#[derive(Clone)]
+struct AdminCredentials {
+    username: String,
+    password: String,
+}
+
 #[derive(Clone)]
 struct App {
     terrain: Arc<Mutex<String>>,
     script: Arc<Mutex<String>>,
     state: watch::Receiver<Arc<str>>,
     tx: mpsc::UnboundedSender<Msg>,
+    trust_proxy_headers: bool,
+    admin_credentials: AdminCredentials,
 }
 
-/// Rejects cross-site requests (including WebSocket upgrades), which could otherwise edit the world.
+/// Rejects cross-site requests and remote write requests.
-async fn same_origin(req: Request, next: Next) -> Response {
+async fn same_origin(State(app): State<App>, req: Request, next: Next) -> Response {
     let h = req.headers();
+    let admin_path = is_admin_path(req.uri().path());
+    if admin_path && !has_valid_admin_credentials(h, &app.admin_credentials) {
+        let mut response = StatusCode::UNAUTHORIZED.into_response();
+        response.headers_mut().insert(
+            header::WWW_AUTHENTICATE,
+            "Basic realm=\"World Simulation Admin\", charset=\"UTF-8\""
+                .parse()
+                .expect("valid authentication challenge"),
+        );
+        return response;
+    }
     if let Some(origin) = h.get(header::ORIGIN).and_then(|v| v.to_str().ok()) {
         let host = h
             .get(header::HOST)
@@ -49,9 +79,141 @@ async fn same_origin(req: Request, next: Next) -> Response {
             return StatusCode::FORBIDDEN.into_response();
         }
     }
+    let peer = req
+        .extensions()
+        .get::<ConnectInfo<SocketAddr>>()
+        .map(|ConnectInfo(address)| *address);
+    if !admin_path
+        && !can_edit(peer, h, app.trust_proxy_headers)
+        && (req.method() != axum::http::Method::GET || req.uri().path() == "/api/actors/ws")
+    {
+        return StatusCode::FORBIDDEN.into_response();
+    }
     next.run(req).await
 }
 
+fn is_admin_path(path: &str) -> bool {
+    path == "/admin" || path.starts_with("/admin/")
+}
+
+fn has_valid_admin_credentials(headers: &HeaderMap, credentials: &AdminCredentials) -> bool {
+    let Some(encoded) = headers
+        .get(header::AUTHORIZATION)
+        .and_then(|value| value.to_str().ok())
+        .and_then(|value| value.strip_prefix("Basic "))
+    else {
+        return false;
+    };
+    let Ok(decoded) = BASE64.decode(encoded) else {
+        return false;
+    };
+    let Some(separator) = decoded.iter().position(|byte| *byte == b':') else {
+        return false;
+    };
+    use subtle::ConstantTimeEq;
+    let username_matches = decoded[..separator]
+        .ct_eq(credentials.username.as_bytes())
+        .unwrap_u8()
+        == 1;
+    let password_matches = decoded[separator + 1..]
+        .ct_eq(credentials.password.as_bytes())
+        .unwrap_u8()
+        == 1;
+    username_matches & password_matches
+}
+
+fn load_admin_credentials() -> AdminCredentials {
+    let username = std::env::var("SIM_ADMIN_USER").unwrap_or_else(|_| "admin".into());
+    assert!(
+        !username.is_empty() && !username.contains(':'),
+        "SIM_ADMIN_USER must be non-empty and must not contain ':'"
+    );
+    let path = std::env::var_os("SIM_ADMIN_PASSWORD_FILE")
+        .map(PathBuf::from)
+        .unwrap_or_else(|| PathBuf::from("sim-admin-password"));
+    let password = load_or_create_admin_password(&path).unwrap_or_else(|error| {
+        panic!(
+            "failed to load admin password at {}: {error}",
+            path.display()
+        )
+    });
+    AdminCredentials { username, password }
+}
+
+fn load_or_create_admin_password(path: &FilePath) -> io::Result<String> {
+    match read_admin_password(path) {
+        Ok(password) => Ok(password),
+        Err(error) if error.kind() == io::ErrorKind::NotFound => create_admin_password(path),
+        Err(error) => Err(error),
+    }
+}
+
+fn read_admin_password(path: &FilePath) -> io::Result<String> {
+    let password = fs::read_to_string(path)?;
+    let password = password.trim_end_matches(['\r', '\n']).to_owned();
+    if password.is_empty() {
+        return Err(io::Error::new(
+            io::ErrorKind::InvalidData,
+            "admin password file is empty",
+        ));
+    }
+    Ok(password)
+}
+
+fn create_admin_password(path: &FilePath) -> io::Result<String> {
+    if let Some(parent) = path
+        .parent()
+        .filter(|parent| !parent.as_os_str().is_empty())
+    {
+        fs::create_dir_all(parent)?;
+    }
+    let password: String = rand::thread_rng()
+        .sample_iter(&rand::distributions::Alphanumeric)
+        .take(32)
+        .map(char::from)
+        .collect();
+    let mut options = OpenOptions::new();
+    options.write(true).create_new(true);
+    #[cfg(unix)]
+    {
+        use std::os::unix::fs::OpenOptionsExt;
+        options.mode(0o600);
+    }
+    match options.open(path) {
+        Ok(mut file) => {
+            writeln!(file, "{password}")?;
+            file.sync_all()?;
+            Ok(password)
+        }
+        Err(error) if error.kind() == io::ErrorKind::AlreadyExists => read_admin_password(path),
+        Err(error) => Err(error),
+    }
+}
+
+fn can_edit(peer: Option<SocketAddr>, headers: &HeaderMap, trust_proxy_headers: bool) -> bool {
+    let client_ips = headers
+        .get("x-forwarded-for")
+        .or_else(|| headers.get("x-real-ip"));
+    if let Some(client_ips) = client_ips {
+        return (trust_proxy_headers || peer.is_some_and(|address| address.ip().is_loopback()))
+            && client_ips
+                .to_str()
+                .ok()
+                .and_then(|ips| ips.split(',').next())
+                .and_then(|ip| ip.trim().parse::<IpAddr>().ok())
+                .is_some_and(|ip| ip.is_loopback());
+    }
+
+    peer.is_some_and(|address| address.ip().is_loopback())
+        || headers
+            .get(header::HOST)
+            .and_then(|host| host.to_str().ok())
+            .and_then(|host| host.parse::<axum::http::uri::Authority>().ok())
+            .is_some_and(|authority| {
+                matches!(authority.host(), "localhost" | "127.0.0.1" | "[::1]")
+            })
+}
+
 fn json(body: String) -> impl IntoResponse {
     ([(header::CONTENT_TYPE, "application/json")], body)
 }
@@ -80,6 +242,15 @@ async fn state(State(app): State<App>) -> impl IntoResponse {
     json(app.state.borrow().to_string())
 }
 
+async fn access(
+    State(app): State<App>,
+    ConnectInfo(peer): ConnectInfo<SocketAddr>,
+    headers: HeaderMap,
+    OriginalUri(uri): OriginalUri,
+) -> Json<bool> {
+    Json(is_admin_path(uri.path()) || can_edit(Some(peer), &headers, app.trust_proxy_headers))
+}
+
 async fn get_script(State(app): State<App>) -> String {
     app.script.lock().unwrap().clone()
 }
@@ -121,8 +292,14 @@ async fn post_edit(State(app): State<App>, body: String) -> StatusCode {
     StatusCode::OK
 }
 
-async fn ws(upgrade: WebSocketUpgrade, State(app): State<App>) -> Response {
+async fn ws(
-    upgrade.on_upgrade(move |socket| client(socket, app))
+    upgrade: WebSocketUpgrade,
+    State(app): State<App>,
+    ConnectInfo(peer): ConnectInfo<SocketAddr>,
+    headers: HeaderMap,
+) -> Response {
+    let can_edit = can_edit(Some(peer), &headers, app.trust_proxy_headers);
+    upgrade.on_upgrade(move |socket| client(socket, app, can_edit))
 }
 
 #[cfg(feature = "console")]
@@ -188,7 +365,7 @@ async fn picard_client(mut socket: WebSocket, app: App, shard: usize) {
 }
 
 /// Pushes every new snapshot to the client; text frames from the client are edit commands.
-async fn client(mut socket: WebSocket, app: App) {
+async fn client(mut socket: WebSocket, app: App, can_edit: bool) {
     let mut rx = app.state.clone();
     loop {
         let snapshot = rx.borrow_and_update().to_string();
@@ -204,7 +381,7 @@ async fn client(mut socket: WebSocket, app: App) {
                     break;
                 }
                 msg = socket.recv() => match msg {
-                    Some(Ok(Message::Text(t))) if t.len() <= 256 => {
+                    Some(Ok(Message::Text(t))) if can_edit && t.len() <= 256 => {
                         let _ = app.tx.send(Msg::Edit(t.as_str().to_owned()));
                     }
                     Some(Ok(Message::Close(_)) | Err(_)) | None => return,
@@ -229,7 +406,8 @@ 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 cluster_token = std::env::var("SIM_CLUSTER_TOKEN").ok();
+    let mut database_url = std::env::var("SHARD_DATABASE_URL").ok();
     let mut listen = "127.0.0.1:0".to_owned();
     let mut attached = false;
     let mut args = args.iter();
@@ -254,6 +432,7 @@ fn shard_options(args: &[String], script: SimScript) -> ShardOptions {
                     .parse()
                     .unwrap_or_else(|_| panic!("invalid seed: {value}"))
             }
+            "--cluster-token" => cluster_token = Some(value.clone()),
             "--database" => database_url = Some(value.clone()),
             "--listen" => listen.clone_from(value),
             _ => panic!("unknown shard option {flag}"),
@@ -267,6 +446,7 @@ fn shard_options(args: &[String], script: SimScript) -> ShardOptions {
     ShardOptions {
         shard,
         seed,
+        cluster_token: cluster_token.unwrap_or_else(|| uuid::Uuid::new_v4().to_string()),
         database_url: database_url.expect("--database is required"),
         listen,
         script,
@@ -307,6 +487,28 @@ async fn main() {
         shards == 1 || database_template.contains("{shard}"),
         "SHARD_DATABASE_URL must contain {{shard}} so that every shard gets its own database"
     );
+    let external_shards = std::env::var("SIM_SHARD_ADDRS").ok().map(|addresses| {
+        addresses
+            .split(',')
+            .map(str::trim)
+            .map(str::to_owned)
+            .collect::<Vec<_>>()
+    });
+    if let Some(addresses) = &external_shards {
+        assert_eq!(
+            addresses.len(),
+            shards,
+            "SIM_SHARD_ADDRS must list one control address per shard"
+        );
+        assert!(
+            addresses.iter().all(|address| !address.is_empty()),
+            "SIM_SHARD_ADDRS contains an empty address"
+        );
+        assert!(
+            std::env::var("SIM_CLUSTER_TOKEN").is_ok(),
+            "SIM_CLUSTER_TOKEN is required with SIM_SHARD_ADDRS"
+        );
+    }
     let addr = std::env::var("SIM_ADDR").unwrap_or_else(|_| "127.0.0.1:7878".into());
     let listener = tokio::net::TcpListener::bind(&addr)
         .await
@@ -326,6 +528,8 @@ async fn main() {
             shards,
             seed,
             database_template,
+            external_shards,
+            cluster_token: std::env::var("SIM_CLUSTER_TOKEN").ok(),
         };
         std::thread::Builder::new()
             .name("gateway".into())
@@ -338,9 +542,13 @@ async fn main() {
         script,
         state: state_rx,
         tx,
+        trust_proxy_headers: std::env::var("SIM_TRUST_PROXY_HEADERS")
+            .is_ok_and(|value| value == "1"),
+        admin_credentials: load_admin_credentials(),
     };
-    let router = Router::new()
+    let mut router = Router::new()
         .route("/", get(index))
+        .route("/api/access", get(access))
         .route("/api/terrain", get(terrain))
         .route("/api/state", get(state))
         .route("/api/script", get(get_script).post(post_script))
@@ -350,12 +558,117 @@ async fn main() {
         .route("/ws", get(ws))
         .route("/pkg/{*path}", get(pkg));
     #[cfg(feature = "console")]
-    let router = router.route("/api/actors/ws", get(picard_ws));
+    {
+        router = router.route("/api/actors/ws", get(picard_ws));
+    }
+    let admin_routes = router.clone();
+    router = router.nest("/admin", admin_routes);
     let router = router
         .layer(DefaultBodyLimit::max(65536))
-        .layer(middleware::from_fn(same_origin))
+        .layer(middleware::from_fn_with_state(app.clone(), same_origin))
         .with_state(app);
 
     println!("Simulation running at http://{addr} with {shards} shards");
-    axum::serve(listener, router).await.unwrap();
+    axum::serve(
+        listener,
+        router.into_make_service_with_connect_info::<SocketAddr>(),
+    )
+    .await
+    .unwrap();
+}
+
+#[cfg(test)]
+mod access_tests {
+    use super::*;
+
+    #[test]
+    fn local_clients_can_edit_without_a_proxy_header() {
+        let peer = "127.0.0.1:7878".parse().unwrap();
+        assert!(can_edit(Some(peer), &HeaderMap::new(), false));
+    }
+
+    #[test]
+    fn forwarded_remote_clients_are_read_only() {
+        let peer = "127.0.0.1:7878".parse().unwrap();
+        let mut headers = HeaderMap::new();
+        headers.insert("x-forwarded-for", "203.0.113.9".parse().unwrap());
+        assert!(!can_edit(Some(peer), &headers, false));
+    }
+
+    #[test]
+    fn localhost_host_is_local_through_docker_port_mapping() {
+        let peer = "172.18.0.1:7878".parse().unwrap();
+        let mut headers = HeaderMap::new();
+        headers.insert(header::HOST, "localhost:7878".parse().unwrap());
+        assert!(can_edit(Some(peer), &headers, false));
+    }
+
+    #[test]
+    fn public_host_without_a_forwarded_local_address_is_read_only() {
+        let peer = "172.18.0.1:7878".parse().unwrap();
+        let mut headers = HeaderMap::new();
+        headers.insert(header::HOST, "sim.example.com".parse().unwrap());
+        assert!(!can_edit(Some(peer), &headers, false));
+    }
+
+    #[test]
+    fn remote_peers_cannot_spoof_a_loopback_forwarded_address() {
+        let peer = "203.0.113.8:7878".parse().unwrap();
+        let mut headers = HeaderMap::new();
+        headers.insert("x-forwarded-for", "127.0.0.1".parse().unwrap());
+        assert!(!can_edit(Some(peer), &headers, false));
+    }
+
+    #[test]
+    fn trusted_proxy_loopback_client_is_local() {
+        let peer = "172.18.0.1:7878".parse().unwrap();
+        let mut headers = HeaderMap::new();
+        headers.insert("x-forwarded-for", "127.0.0.1".parse().unwrap());
+        assert!(can_edit(Some(peer), &headers, true));
+    }
+
+    #[test]
+    fn admin_path_matches_only_its_route_tree() {
+        assert!(is_admin_path("/admin"));
+        assert!(is_admin_path("/admin/api/edit"));
+        assert!(!is_admin_path("/administer"));
+        assert!(!is_admin_path("/api/edit"));
+    }
+
+    #[test]
+    fn admin_path_requires_matching_basic_credentials() {
+        let credentials = AdminCredentials {
+            username: "operator".into(),
+            password: "long-secret".into(),
+        };
+        let mut headers = HeaderMap::new();
+        let encoded = BASE64.encode("operator:long-secret");
+        headers.insert(
+            header::AUTHORIZATION,
+            format!("Basic {encoded}").parse().unwrap(),
+        );
+        assert!(has_valid_admin_credentials(&headers, &credentials));
+
+        headers.insert(
+            header::AUTHORIZATION,
+            format!("Basic {}", BASE64.encode("operator:wrong"))
+                .parse()
+                .unwrap(),
+        );
+        assert!(!has_valid_admin_credentials(&headers, &credentials));
+    }
+
+    #[test]
+    fn generated_admin_password_is_persisted_and_reused() {
+        let directory = std::env::temp_dir().join(format!("sim-admin-{}", uuid::Uuid::new_v4()));
+        let path = directory.join("admin-password");
+
+        let first = load_or_create_admin_password(&path).unwrap();
+        assert_eq!(first.len(), 32);
+        assert!(first.bytes().all(|byte| byte.is_ascii_alphanumeric()));
+        assert_eq!(read_admin_password(&path).unwrap(), first);
+        assert_eq!(load_or_create_admin_password(&path).unwrap(), first);
+
+        std::fs::remove_dir_all(directory).unwrap();
+    }
 }

src/world.rs

@@ -9,8 +9,8 @@ use actors::{
 };
 pub use economy::Quote;
 use economy::{Cart, CartMsg, Job, Market, MarketMsg, Workplace, WorkplaceMsg};
-use picard::actors::{ActorRef, ActorSystem, Sender, Tell};
+use picard::actors::{ActorRef, ActorReference, ActorSystem, Sender, Tell};
-use router::{Arrival, LocalActor};
+use router::{Admission, Arrival, LocalActor};
 pub use router::{EntityKey, Kind, RemoteSender, Router, Shard};
 use rune::runtime::RuntimeContext;
 use rune::termcolor::Buffer;
@@ -1134,24 +1134,13 @@ impl Species {
     }
 }
 
-/// How long a departed entity's actor keeps forwarding messages to the shard it moved to.
-const FORWARDING_TICKS: u64 = 100;
-
-/// An entity that moved to another shard and whose old actor still forwards its messages.
-struct Departure {
-    key: EntityKey,
-    shard: usize,
-    tick: u64,
-    actor: LocalActor,
-}
-
 /// What changed in a shard during its steps, for the gateway to maintain the cluster directory.
 #[derive(Clone, Debug, Default, Deserialize, Serialize)]
 pub struct StepReport {
     /// Entities handed to another shard.
-    pub moved: Vec<(EntityKey, usize)>,
+    pub moved: Vec<(EntityKey, usize, String)>,
     /// Entities that were born on, spawned on or moved into this shard.
-    pub arrived: Vec<EntityKey>,
+    pub arrived: Vec<(EntityKey, String)>,
     /// Entities that died or were removed.
     pub left: Vec<EntityKey>,
 }
@@ -1211,9 +1200,11 @@ pub struct World {
     next_partnership_id: u32,
     mailbox: Mailbox,
     actor_system: ActorSystem,
+    remote_transport: Option<Arc<picard::remoting::ClusterTransport>>,
+    remote_listener: Option<Box<dyn picard::remoting::RemoteListener>>,
+    remote_address: Option<String>,
     shard: Shard,
     router: Arc<Router>,
-    departed: Vec<Departure>,
     report: StepReport,
     paths: PathScratch,
     houses: Vec<(usize, usize)>,
@@ -1493,8 +1484,10 @@ impl World {
             mailbox: Mailbox::new(&actor_system),
             router: Router::new(shard, &actor_system),
             actor_system,
+            remote_transport: None,
+            remote_listener: None,
+            remote_address: None,
             shard,
-            departed: Vec::new(),
             report: StepReport::default(),
             paths: PathScratch::default(),
             houses: vec![],
@@ -1593,8 +1586,10 @@ impl World {
 
     /// Records an entity appearing on this shard, unless it is the only one.
     fn report_arrival(&mut self, key: EntityKey) {
-        if self.shard.count > 1 {
+        if self.shard.count > 1
-            self.report.arrived.push(key);
+            && let Some(path) = self.router.actor_path(key)
+        {
+            self.report.arrived.push((key, path));
         }
     }
 
@@ -1716,46 +1711,31 @@ impl World {
                     EntityKey::new(Kind::Cart, id)
                 }
             };
-            self.departed.retain(|departure| departure.key != key);
             self.report_arrival(key);
         }
     }
 
-    /// Stops the actors of entities that left long enough ago for every shard to know where
+    /// Migrates every entity standing outside this shard's columns to its owning shard.
-    /// they went; the router keeps forwarding their messages from then on.
-    fn retire_departures(&mut self) {
-        let tick = self.tick;
-        let (expired, waiting): (Vec<_>, Vec<_>) = self
-            .departed
-            .drain(..)
-            .partition(|departure| tick.saturating_sub(departure.tick) >= FORWARDING_TICKS);
-        self.departed = waiting;
-        for departure in expired {
-            if self.router.retire_forwarder(departure.key, departure.shard) {
-                departure.actor.stop();
-            }
-        }
-    }
-
-    /// Hands every entity standing outside this shard's columns to the shard that owns its
-    /// position. The entities' actors pass their whole state on themselves.
     fn transfer_out(&mut self, census: &Census) {
         if self.shard.count == 1 {
             return;
         }
-        let request = self.mailbox.request();
-        let mut leaving = Vec::new();
         for person in &census.people {
-            if !self.owns(person.position())
+            if !self.owns(person.position()) && self.person_actors.contains_key(&person.id) {
-                && let Some(actor) = self.person_actors.remove(&person.id)
-            {
                 let shard = self.shard.owner(person.x, self.width);
-                actor.tell(PersonMsg::Depart(request, shard), self.mailbox.reply_to());
+                let key = EntityKey::person(person.id);
-                leaving.push((
+                match self
-                    EntityKey::person(person.id),
+                    .router
-                    shard,
+                    .migrate(key, shard, Admission::Person(Box::new(person.clone())))
-                    LocalActor::Person(actor),
+                {
-                ));
+                    Ok(path) => {
+                        self.person_actors.remove(&person.id);
+                        self.report.moved.push((key, shard, path));
+                    }
+                    Err(error) => {
+                        tracing::error!(entity = %key.name(), shard, %error, "failed to migrate entity");
+                    }
+                }
             }
         }
         for species in [Species::Rabbit, Species::Wolf] {
@@ -1764,34 +1744,42 @@ impl World {
                 Species::Wolf => &census.wolves,
             };
             for (id, animal) in animals {
-                if !self.owns(animal.position())
+                if !self.owns(animal.position()) && self.animal_actors(species).contains_key(id) {
-                    && let Some(actor) = self.animal_actors_mut(species).remove(id)
-                {
                     let shard = self.shard.owner(animal.x, self.width);
-                    actor.tell(AnimalMsg::Depart(request, shard), self.mailbox.reply_to());
                     let key = EntityKey::new(species.kind(), *id);
-                    leaving.push((key, shard, LocalActor::Animal(actor)));
+                    match self.router.migrate(
+                        key,
+                        shard,
+                        Admission::Animal(species.kind(), *id, animal.clone()),
+                    ) {
+                        Ok(path) => {
+                            self.animal_actors_mut(species).remove(id);
+                            self.report.moved.push((key, shard, path));
+                        }
+                        Err(error) => {
+                            tracing::error!(entity = %key.name(), shard, %error, "failed to migrate entity");
+                        }
+                    }
                 }
             }
         }
         for cart in &census.carts {
-            if !self.owns(cart.position())
+            if !self.owns(cart.position()) && self.cart_actors.contains_key(&cart.id) {
-                && let Some(actor) = self.cart_actors.remove(&cart.id)
-            {
                 let shard = self.shard.owner(cart.x, self.width);
-                actor.tell(CartMsg::Depart(request, shard), self.mailbox.reply_to());
+                let key = cart.key();
-                leaving.push((cart.key(), shard, LocalActor::Cart(actor)));
+                match self
+                    .router
+                    .migrate(key, shard, Admission::Cart(cart.clone()))
+                {
+                    Ok(path) => {
+                        self.cart_actors.remove(&cart.id);
+                        self.report.moved.push((key, shard, path));
+                    }
+                    Err(error) => {
+                        tracing::error!(entity = %key.name(), shard, %error, "failed to migrate entity");
+                    }
                 }
             }
-        self.mailbox.collect(request, leaving.len());
-        for (key, shard, actor) in leaving {
-            self.report.moved.push((key, shard));
-            self.departed.push(Departure {
-                key,
-                shard,
-                tick: self.tick,
-                actor,
-            });
         }
     }
 
@@ -2124,7 +2112,6 @@ impl World {
     pub fn step(&mut self) {
         self.tick += 1;
         self.receive_arrivals();
-        self.retire_departures();
         self.regrow();
         let new_day = self.tick.is_multiple_of(self.calendar.ticks_per_day);
         if new_day {
@@ -3503,18 +3490,6 @@ impl World {
     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.
@@ -3528,14 +3503,64 @@ impl World {
         self.receive_arrivals();
     }
 
-    pub fn shard(&self) -> Shard {
-        self.shard
-    }
-
     pub fn router(&self) -> &Arc<Router> {
         &self.router
     }
 
+    pub fn start_remoting(&mut self, token: &str) -> Result<(), String> {
+        let transport = Arc::new(picard::remoting::ClusterTransport::new());
+        let authenticator = picard::remoting::PeerAuthenticator::new(
+            (0..self.shard.count)
+                .filter(|index| *index != self.shard.index)
+                .map(|index| (format!("shard-{index}"), token.to_owned())),
+        );
+        let bind = std::env::var("SIM_REMOTING_BIND")
+            .unwrap_or_else(|_| "127.0.0.1:0".into())
+            .parse()
+            .map_err(|error: std::net::AddrParseError| error.to_string())?;
+        let listener = self
+            .actor_system
+            .listen_remote_with(
+                bind,
+                authenticator,
+                self.actor_system.remote_handler(),
+                transport.clone(),
+            )
+            .map_err(|error| error.to_string())?;
+        let address = listener
+            .local_addr()
+            .ok_or_else(|| "Picard remoting listener did not report its address".to_owned())?;
+        self.remote_address = Some(
+            std::env::var("SIM_REMOTING_ADVERTISE")
+                .unwrap_or_else(|_| format!("ws://{address}/remoting")),
+        );
+        self.remote_transport = Some(transport);
+        self.remote_listener = Some(listener);
+        self.router.register_migration_handler();
+        Ok(())
+    }
+
+    pub fn connect_peers(&self, peers: Vec<crate::cluster::protocol::Peer>) -> Result<(), String> {
+        let transport = self
+            .remote_transport
+            .as_ref()
+            .ok_or_else(|| "Picard remoting has not been started".to_owned())?
+            .clone();
+        let peers = crate::cluster::transport::ClusterTransport::connect(
+            &self.actor_system,
+            self.shard.index,
+            peers,
+            transport,
+        )?;
+        self.router.set_remote(peers.clone());
+        self.router.set_peers(peers.peers());
+        Ok(())
+    }
+
+    pub fn remote_address(&self) -> Option<String> {
+        self.remote_address.clone()
+    }
+
     /// Prices of this shard's markets, for the gateway to publish to every shard.
     pub fn quotes(&self) -> Vec<Quote> {
         let local: HashSet<EntityKey> = self
@@ -3567,29 +3592,34 @@ impl World {
         std::mem::take(&mut self.report)
     }
 
-    /// Every person and animal currently simulated by this shard.
+    /// Every entity currently simulated by this shard and its current actor path.
-    pub fn roster(&self) -> Vec<EntityKey> {
+    pub fn roster(&self) -> Vec<(EntityKey, String)> {
-        let people = self.person_actors.keys().map(|&id| EntityKey::person(id));
+        let people = self
+            .person_actors
+            .iter()
+            .map(|(&id, actor)| (EntityKey::person(id), actor.path().to_string()));
         let rabbits = self
             .rabbit_actors
-            .keys()
+            .iter()
-            .map(|&id| EntityKey::new(Kind::Rabbit, id));
+            .map(|(&id, actor)| (EntityKey::new(Kind::Rabbit, id), actor.path().to_string()));
         let wolves = self
             .wolf_actors
-            .keys()
+            .iter()
-            .map(|&id| EntityKey::new(Kind::Wolf, id));
+            .map(|(&id, actor)| (EntityKey::new(Kind::Wolf, id), actor.path().to_string()));
-        let workplaces = self
+        let workplaces = self.workplace_actors.iter().map(|(&id, actor)| {
-            .workplace_actors
+            (
-            .keys()
+                EntityKey::new(Kind::Workplace, id),
-            .map(|&id| EntityKey::new(Kind::Workplace, id));
+                actor.path().to_string(),
+            )
+        });
         let markets = self
             .market_actors
-            .keys()
+            .iter()
-            .map(|&id| EntityKey::new(Kind::Market, id));
+            .map(|(&id, actor)| (EntityKey::new(Kind::Market, id), actor.path().to_string()));
         let carts = self
             .cart_actors
-            .keys()
+            .iter()
-            .map(|&id| EntityKey::new(Kind::Cart, id));
+            .map(|(&id, actor)| (EntityKey::new(Kind::Cart, id), actor.path().to_string()));
         people
             .chain(rabbits)
             .chain(wolves)
@@ -3602,6 +3632,9 @@ impl World {
 
 impl Drop for World {
     fn drop(&mut self) {
+        if let Some(listener) = &self.remote_listener {
+            listener.shutdown();
+        }
         // Shutdown starts eagerly; Drop must not block waiting for completion.
         drop(self.actor_system.shutdown());
     }
@@ -3613,47 +3646,32 @@ mod tests {
     use super::economy::{Cart, Good, Job, MarketMsg, Phase, Trade, Workplace};
     use super::router::Kind;
     use super::{
-        Animal, CROP, DEFAULT_SCRIPT, EntityKey, FORWARDING_TICKS, H, LifeStage, Person, Router,
+        Animal, CROP, DEFAULT_SCRIPT, EntityKey, H, LifeStage, Person, Shard, SimScript, W, World,
-        Shard, SimScript, W, World, cheb, default_config,
+        cheb, default_config,
     };
-    use crate::world::router::RemoteSender;
-    use picard::remoting::{ActorAddress, RemoteRequest};
-    use std::sync::{Arc, Mutex, Weak};
     use std::time::{Duration, Instant};
 
-    /// Delivers remote messages by calling the receiving shard's router directly.
-    #[derive(Default)]
-    struct Loopback {
-        routers: Mutex<Vec<Weak<Router>>>,
-    }
-
-    impl RemoteSender for Loopback {
-        fn send(&self, shard: usize, path: &str, payload: serde_json::Value) {
-            let router = self.routers.lock().unwrap()[shard].upgrade().unwrap();
-            let destination = ActorAddress::local(router.system_id(), path);
-            let response =
-                router.handle("loopback", RemoteRequest::new(destination, None, payload));
-            assert!(response.accepted, "{path} rejected: {:?}", response.error);
-        }
-
-        fn flush(&self) {}
-    }
-
     fn connected_shards(count: usize) -> Vec<World> {
-        let loopback = Arc::new(Loopback::default());
+        let mut shards: Vec<World> = (0..count)
-        let shards: Vec<World> = (0..count)
             .map(|index| {
                 let shard = Shard::new(index, count).unwrap();
                 World::new_shard(42, SimScript::load(DEFAULT_SCRIPT).unwrap(), shard)
             })
             .collect();
+        let token = uuid::Uuid::new_v4().to_string();
+        for world in &mut shards {
+            world.start_remoting(&token).unwrap();
+        }
+        let peers = shards
+            .iter()
+            .map(|world| crate::cluster::protocol::Peer {
+                address: world.remote_address().unwrap(),
+                token: token.clone(),
+                system_id: world.router().system_id(),
+            })
+            .collect::<Vec<_>>();
         for world in &shards {
-            loopback
+            world.connect_peers(peers.clone()).unwrap();
-                .routers
-                .lock()
-                .unwrap()
-                .push(Arc::downgrade(world.router()));
-            world.router().set_remote(loopback.clone());
         }
         shards
     }
@@ -3663,8 +3681,18 @@ mod tests {
         let mut changes = Vec::new();
         for (index, world) in shards.iter_mut().enumerate() {
             let report = world.take_report();
-            changes.extend(report.moved.into_iter().map(|(key, to)| (key, Some(to))));
+            changes.extend(
-            changes.extend(report.arrived.into_iter().map(|key| (key, Some(index))));
+                report
+                    .moved
+                    .into_iter()
+                    .map(|(key, to, path)| (key, Some((to, path)))),
+            );
+            changes.extend(
+                report
+                    .arrived
+                    .into_iter()
+                    .map(|(key, path)| (key, Some((index, path)))),
+            );
             changes.extend(report.left.into_iter().map(|key| (key, None)));
         }
         for world in shards {
@@ -3708,7 +3736,7 @@ mod tests {
     }
 
     /// The default script at the size of the original single village, for long-running tests.
-    fn village_script() -> SimScript {
+    fn village_script_source() -> String {
         let source = [
             ("WIDTH", "96"),
             ("HEIGHT", "64"),
@@ -3729,6 +3757,11 @@ mod tests {
         .fold(DEFAULT_SCRIPT.to_string(), |source, (name, value)| {
             with_constant(&source, name, value)
         });
+        source
+    }
+
+    fn village_script() -> SimScript {
+        let source = village_script_source();
         SimScript::load(&source).unwrap()
     }
 
@@ -4095,7 +4128,11 @@ mod tests {
                 assert!(position == expected.home || expected.res[index] == *res);
             }
         }
-        let mut ids: Vec<EntityKey> = shards.iter().flat_map(World::roster).collect();
+        let mut ids: Vec<EntityKey> = shards
+            .iter()
+            .flat_map(World::roster)
+            .map(|(key, _)| key)
+            .collect();
         let total = ids.len();
         ids.sort_unstable();
         ids.dedup();
@@ -4118,7 +4155,10 @@ mod tests {
         shards[0].step();
         assert!(!shards[0].person_actors.contains_key(&id));
         let report = shards[0].take_report();
-        assert_eq!(report.moved, vec![(EntityKey::person(id), 1)]);
+        assert_eq!(report.moved.len(), 1);
+        assert_eq!(report.moved[0].0, EntityKey::person(id));
+        assert_eq!(report.moved[0].1, 1);
+        assert!(report.moved[0].2.starts_with("/user/person-"));
 
         shards[1].receive_arrivals();
         assert!(shards[1].person_actors.contains_key(&id));
@@ -4130,21 +4170,14 @@ mod tests {
         assert_eq!(arrived.wood, 7);
         assert!(shards[1].owns(arrived.position()));
 
-        // The old actor forwards messages sent to the person on the shard they left.
+        // Picard's migration forwarder keeps the old address usable before the directory update.
         shards[0].tell_person(id, PersonMsg::Edit(PersonEdit::Stone(3)));
         wait_for_person(&shards[1], id, |person| person.stone == 3);
 
-        // Once it stops, the router forwards them itself until the directory catches up.
+        publish_directory(&mut shards);
-        shards[0].tick += FORWARDING_TICKS;
-        shards[0].retire_departures();
-        assert!(shards[0].departed.is_empty());
         shards[0].tell_person(id, PersonMsg::Edit(PersonEdit::Stone(4)));
         wait_for_person(&shards[1], id, |person| person.stone == 4);
 
-        publish_directory(&mut shards);
-        shards[0].tell_person(id, PersonMsg::Edit(PersonEdit::Stone(5)));
-        wait_for_person(&shards[1], id, |person| person.stone == 5);
-
         // Walking back resumes life on the first shard.
         let back = shards[0].home;
         shards[1].tell_person(id, PersonMsg::MoveTo(back));
@@ -4153,8 +4186,8 @@ mod tests {
         assert!(shards[0].person_actors.contains_key(&id));
         assert!(!shards[1].person_actors.contains_key(&id));
         publish_directory(&mut shards);
-        shards[1].tell_person(id, PersonMsg::Edit(PersonEdit::Stone(6)));
+        shards[1].tell_person(id, PersonMsg::Edit(PersonEdit::Stone(5)));
-        wait_for_person(&shards[0], id, |person| person.stone == 6);
+        wait_for_person(&shards[0], id, |person| person.stone == 5);
     }
 
     #[test]
@@ -4298,7 +4331,17 @@ mod tests {
 
     #[test]
     fn village_survives_until_its_first_weekly_birth_check() {
-        let mut world = World::new_with_script(42, village_script());
+        let script = village_script_source();
+        let script = with_constant(&script, "TICKS_PER_DAY", "60");
+        let script = with_constant(&script, "INITIAL_RABBITS", "0");
+        let script = with_constant(&script, "INITIAL_WOLVES", "0");
+        let script = with_constant(&script, "INITIAL_POPULATION", "18");
+        let script = with_constant(&script, "INITIAL_YOUNG_ADULTS", "18");
+        let script = with_constant(&script, "INITIAL_FOOD", "120");
+        let script = with_constant(&script, "HUNGER_OTHER", "0.08");
+        let script = with_constant(&script, "HUNGER_NIGHT_AT_HOME", "0.04");
+        let script = SimScript::load(&script).unwrap();
+        let mut world = World::new_with_script(42, script);
         let first_week_tick =
             world.calendar.ticks_per_day * u64::from(world.calendar.days_per_week);
         let mut daily = Vec::new();

src/world/actors.rs

@@ -5,7 +5,7 @@
 //! lets them apply entity-local rules (ageing, hunger, eating, energy, dying) themselves.
 
 use super::economy::{Cart, Job, Market, Traded, Workplace};
-use super::router::{Admission, EntityKey, Router};
+use super::router::EntityKey;
 use super::{Animal, CalendarSettings, LifeStage, Partnership, Person, Rng, sample_stage_duration};
 use picard::actors::{
     Actor, ActorFactoryArgs, ActorRef, ActorRefFactory, ActorSystem, Context as ActorContext,
@@ -14,7 +14,6 @@ use picard::actors::{
 use serde::{Deserialize, Serialize};
 use std::cell::{Cell, RefCell};
 use std::collections::HashMap;
-use std::sync::Arc;
 use std::sync::mpsc::{Receiver, Sender as MpscSender, channel};
 use std::time::Duration;
 
@@ -125,11 +124,6 @@ pub(super) enum PersonMsg {
     },
     SetJob(Option<Job>),
     SetCart(Option<u64>),
-    /// Hands the person to `shard`, whose region they walked into; this actor then forwards
-    /// every message to them until it is stopped.
-    Depart(RequestId, usize),
-    /// The person walked back into this shard while this actor was still forwarding.
-    Readmit(Person),
 }
 
 /// One animal tick: the world supplies movement and opportunities, the animal its metabolism.
@@ -159,8 +153,6 @@ pub(super) enum AnimalMsg {
     /// A successful attack; only the first one kills the animal.
     Hunted(RequestId),
     SetEnergyWithin(Area, f32),
-    Depart(RequestId, usize),
-    Readmit(Animal),
 }
 
 #[derive(Clone, Debug, Deserialize, Serialize)]
@@ -188,7 +180,6 @@ pub(super) enum ReplyBody {
     Hunted(AnimalId, bool),
     Partnership(Partnership),
     BirthDue(Partnership, bool),
-    Departed,
     Market(Market),
     Workplace(Workplace),
     Cart(Cart),
@@ -215,22 +206,15 @@ pub(super) fn reply(sender: Sender, request: RequestId, body: ReplyBody) {
 
 pub(super) struct PersonActor {
     person: Person,
-    router: Arc<Router>,
-    /// The shard this person moved to; set while this actor only forwards messages.
-    forward_to: Option<usize>,
 }
 
-impl ActorFactoryArgs<(Person, Arc<Router>)> for PersonActor {
+impl ActorFactoryArgs<Person> for PersonActor {
-    fn create_args((mut person, router): (Person, Arc<Router>)) -> Self {
+    fn create_args(mut person: Person) -> Self {
         if person.rng == 0 {
             // Snapshots from before people had their own random stream.
             person.rng = u64::from(person.id).wrapping_mul(0x9E37_79B9_7F4A_7C15) | 1;
         }
-        Self {
+        Self { person }
-            person,
-            router,
-            forward_to: None,
-        }
     }
 }
 
@@ -320,16 +304,6 @@ impl Actor for PersonActor {
     type Msg = PersonMsg;
 
     fn recv(&mut self, _ctx: &ActorContext<Self::Msg>, msg: Self::Msg, sender: Sender) {
-        if let Some(shard) = self.forward_to {
-            if let PersonMsg::Readmit(person) = msg {
-                self.person = person;
-                self.forward_to = None;
-            } else {
-                let key = EntityKey::person(self.person.id);
-                self.router.forward(shard, key, &msg);
-            }
-            return;
-        }
         match msg {
             PersonMsg::Observe(request) => {
                 reply(sender, request, ReplyBody::Person(self.person.clone()));
@@ -367,34 +341,18 @@ impl Actor for PersonActor {
                 self.person.unpaid_days = 0;
             }
             PersonMsg::SetCart(cart) => self.person.cart = cart,
-            PersonMsg::Depart(request, shard) => {
-                self.router
-                    .admit_at(shard, Admission::Person(Box::new(self.person.clone())));
-                self.forward_to = Some(shard);
-                reply(sender, request, ReplyBody::Departed);
-            }
-            PersonMsg::Readmit(person) => self.person = person,
         }
     }
 }
 
 pub(super) struct AnimalActor {
-    key: EntityKey,
     id: AnimalId,
     animal: Animal,
-    router: Arc<Router>,
-    forward_to: Option<usize>,
 }
 
-impl ActorFactoryArgs<(EntityKey, Animal, Arc<Router>)> for AnimalActor {
+impl ActorFactoryArgs<(EntityKey, Animal)> for AnimalActor {
-    fn create_args((key, animal, router): (EntityKey, Animal, Arc<Router>)) -> Self {
+    fn create_args((key, animal): (EntityKey, Animal)) -> Self {
-        Self {
+        Self { id: key.id, animal }
-            key,
-            id: key.id,
-            animal,
-            router,
-            forward_to: None,
-        }
     }
 }
 
@@ -436,15 +394,6 @@ impl Actor for AnimalActor {
     type Msg = AnimalMsg;
 
     fn recv(&mut self, _ctx: &ActorContext<Self::Msg>, msg: Self::Msg, sender: Sender) {
-        if let Some(shard) = self.forward_to {
-            if let AnimalMsg::Readmit(animal) = msg {
-                self.animal = animal;
-                self.forward_to = None;
-            } else {
-                self.router.forward(shard, self.key, &msg);
-            }
-            return;
-        }
         match msg {
             AnimalMsg::Observe(request) => {
                 reply(
@@ -475,13 +424,6 @@ impl Actor for AnimalActor {
                     self.animal.energy = energy;
                 }
             }
-            AnimalMsg::Depart(request, shard) => {
-                let admission = Admission::Animal(self.key.kind, self.id, self.animal.clone());
-                self.router.admit_at(shard, admission);
-                self.forward_to = Some(shard);
-                reply(sender, request, ReplyBody::Departed);
-            }
-            AnimalMsg::Readmit(animal) => self.animal = animal,
         }
     }
 }

src/world/economy.rs

@@ -6,7 +6,7 @@
 //! between people, workplaces and markets.
 
 use super::actors::{PersonMsg, ReplyBody, RequestId, reply};
-use super::router::{Admission, EntityKey, Kind, Router};
+use super::router::{EntityKey, Kind, Router};
 use picard::actors::{Actor, ActorFactoryArgs, Context as ActorContext, Sender};
 use serde::{Deserialize, Serialize};
 use std::sync::Arc;
@@ -722,23 +722,15 @@ pub(super) enum CartMsg {
     },
     SetOwner(Option<u32>),
     MoveTo((usize, usize)),
-    Depart(RequestId, usize),
-    Readmit(Cart),
 }
 
 pub(super) struct CartActor {
     cart: Cart,
-    router: Arc<Router>,
-    forward_to: Option<usize>,
 }
 
-impl ActorFactoryArgs<(Cart, Arc<Router>)> for CartActor {
+impl ActorFactoryArgs<Cart> for CartActor {
-    fn create_args((cart, router): (Cart, Arc<Router>)) -> Self {
+    fn create_args(cart: Cart) -> Self {
-        Self {
+        Self { cart }
-            cart,
-            router,
-            forward_to: None,
-        }
     }
 }
 
@@ -746,15 +738,6 @@ impl Actor for CartActor {
     type Msg = CartMsg;
 
     fn recv(&mut self, _ctx: &ActorContext<Self::Msg>, msg: Self::Msg, sender: Sender) {
-        if let Some(shard) = self.forward_to {
-            if let CartMsg::Readmit(cart) = msg {
-                self.cart = cart;
-                self.forward_to = None;
-            } else {
-                self.router.forward(shard, self.cart.key(), &msg);
-            }
-            return;
-        }
         let cart = &mut self.cart;
         match msg {
             CartMsg::Observe(request) => reply(sender, request, ReplyBody::Cart(cart.clone())),
@@ -773,12 +756,6 @@ impl Actor for CartActor {
             }
             CartMsg::SetOwner(owner) => cart.owner = owner,
             CartMsg::MoveTo(position) => (cart.x, cart.y) = position,
-            CartMsg::Depart(request, shard) => {
-                self.router.admit_at(shard, Admission::Cart(cart.clone()));
-                self.forward_to = Some(shard);
-                reply(sender, request, ReplyBody::Departed);
-            }
-            CartMsg::Readmit(readmitted) => *cart = readmitted,
         }
     }
 }

src/world/router.rs

@@ -15,17 +15,14 @@ use super::economy::{
     WorkplaceMsg,
 };
 use super::{Animal, Partnership, Person};
+use crate::cluster::transport::PeerConnection;
 use picard::actors::{ActorRef, ActorRefFactory, ActorReference, ActorSystem, CreateError, Tell};
-use picard::remoting::{RemoteError, RemoteHandler, RemoteRequest, RemoteResponse};
 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
 use std::ops::Range;
 use std::sync::{Arc, Mutex, RwLock, Weak};
 use std::time::Duration;
 
-/// Remote path that accepts entities moving into this shard's region.
-pub const ADMIT_PATH: &str = "/shard/admit";
-
 #[derive(Clone, Copy, Debug, Deserialize, Serialize, PartialEq, Eq, Hash, PartialOrd, Ord)]
 #[serde(rename_all = "snake_case")]
 pub enum Kind {
@@ -39,16 +36,6 @@ pub enum Kind {
 }
 
 impl Kind {
-    const ALL: [Kind; 7] = [
-        Kind::Person,
-        Kind::Rabbit,
-        Kind::Wolf,
-        Kind::Partnership,
-        Kind::Workplace,
-        Kind::Market,
-        Kind::Cart,
-    ];
-
     fn name(self) -> &'static str {
         match self {
             Kind::Person => "person",
@@ -82,16 +69,6 @@ impl EntityKey {
     pub fn name(self) -> String {
         format!("{}-{}", self.kind.name(), self.id)
     }
-
-    pub fn path(self) -> String {
-        format!("/user/{}", self.name())
-    }
-
-    pub fn from_path(path: &str) -> Option<Self> {
-        let (kind, id) = path.strip_prefix("/user/")?.rsplit_once('-')?;
-        let kind = Kind::ALL.into_iter().find(|k| k.name() == kind)?;
-        Some(Self::new(kind, id.parse().ok()?))
-    }
 }
 
 /// This process's part of the world: shard `index` of `count` owns a vertical strip of columns.
@@ -142,11 +119,10 @@ impl Shard {
 
 /// Delivers serialised messages to other shards; implemented by the cluster transport.
 pub trait RemoteSender: Send + Sync {
-    /// Queues `payload` for the actor at `path` on shard `shard`. Messages to one shard are
+    /// Sends `payload` to the actor at `path` on shard `shard`.
-    /// delivered in the order they were queued.
     fn send(&self, shard: usize, path: &str, payload: serde_json::Value);
 
-    /// Waits until every queued message has been accepted by its receiving shard.
+    /// Waits until all sent messages have been accepted by their receiving actors.
     fn flush(&self);
 }
 
@@ -172,6 +148,17 @@ impl LocalActor {
             LocalActor::Cart(actor) => actor.stop(),
         }
     }
+
+    pub(super) fn path(&self) -> String {
+        match self {
+            LocalActor::Person(actor) => actor.path().to_string(),
+            LocalActor::Animal(actor) => actor.path().to_string(),
+            LocalActor::Partnership(actor) => actor.path().to_string(),
+            LocalActor::Workplace(actor) => actor.path().to_string(),
+            LocalActor::Market(actor) => actor.path().to_string(),
+            LocalActor::Cart(actor) => actor.path().to_string(),
+        }
+    }
 }
 
 /// A message type that can be routed to the entities whose actors accept it.
@@ -241,16 +228,6 @@ pub(super) enum Admission {
     Cart(Cart),
 }
 
-impl Admission {
-    fn key(&self) -> EntityKey {
-        match self {
-            Admission::Person(person) => EntityKey::person(person.id),
-            Admission::Animal(kind, id, _) => EntityKey::new(*kind, *id),
-            Admission::Cart(cart) => cart.key(),
-        }
-    }
-}
-
 /// An admitted entity, waiting for the world to pick it up at the start of its next step.
 pub(super) enum Arrival {
     Person(Box<Person>, ActorRef<PersonMsg>),
@@ -258,26 +235,14 @@ pub(super) enum Arrival {
     Cart(u64, ActorRef<CartMsg>),
 }
 
-impl Arrival {
-    fn key(&self) -> EntityKey {
-        match self {
-            Arrival::Person(person, _) => EntityKey::person(person.id),
-            Arrival::Animal(kind, id, ..) => EntityKey::new(*kind, *id),
-            Arrival::Cart(id, _) => EntityKey::new(Kind::Cart, *id),
-        }
-    }
-}
-
 pub struct Router {
     shard: Shard,
     system: ActorSystem,
     this: Weak<Router>,
-    dispatch: RemoteHandler,
     local: RwLock<HashMap<EntityKey, LocalActor>>,
-    /// Shard of every entity elsewhere in the cluster, as last published by the gateway.
+    /// Shard and current actor path of every entity elsewhere in the cluster.
-    directory: RwLock<HashMap<EntityKey, usize>>,
+    directory: RwLock<HashMap<EntityKey, (usize, String)>>,
-    /// Where entities that left this shard went, until the directory catches up.
+    peers: RwLock<HashMap<usize, PeerConnection>>,
-    forwarding: RwLock<HashMap<EntityKey, usize>>,
     remote: RwLock<Option<Arc<dyn RemoteSender>>>,
     arrivals: Mutex<Vec<Arrival>>,
 }
@@ -288,10 +253,9 @@ impl Router {
             shard,
             system: system.clone(),
             this: this.clone(),
-            dispatch: system.remote_handler(),
             local: RwLock::default(),
             directory: RwLock::default(),
-            forwarding: RwLock::default(),
+            peers: RwLock::default(),
             remote: RwLock::default(),
             arrivals: Mutex::default(),
         })
@@ -306,12 +270,15 @@ impl Router {
         *self.remote.write().expect("router lock") = Some(remote);
     }
 
+    pub(super) fn set_peers(&self, peers: HashMap<usize, PeerConnection>) {
+        *self.peers.write().expect("router peer lock") = peers;
+    }
+
     fn arc(&self) -> Arc<Router> {
         self.this.upgrade().expect("the router outlives its actors")
     }
 
     pub(super) fn register(&self, key: EntityKey, actor: LocalActor) {
-        self.forwarding.write().expect("router lock").remove(&key);
         self.local.write().expect("router lock").insert(key, actor);
     }
 
@@ -319,26 +286,16 @@ impl Router {
         self.local.write().expect("router lock").remove(&key);
     }
 
-    /// Remembers that `key` moved to `shard` once its forwarding actor is gone.
-    pub(super) fn set_forwarding(&self, key: EntityKey, shard: usize) {
-        self.forwarding
-            .write()
-            .expect("router lock")
-            .insert(key, shard);
-    }
-
     /// Applies directory changes published by the gateway; `None` means the entity is gone.
-    pub fn apply_directory(&self, changes: &[(EntityKey, Option<usize>)]) {
+    pub fn apply_directory(&self, changes: &[(EntityKey, Option<(usize, String)>)]) {
         let mut directory = self.directory.write().expect("router lock");
-        let mut forwarding = self.forwarding.write().expect("router lock");
+        for (key, location) in changes {
-        for &(key, shard) in changes {
+            match location {
-            forwarding.remove(&key);
+                Some((shard, path)) if *shard != self.shard.index => {
-            match shard {
+                    directory.insert(*key, (*shard, path.clone()));
-                Some(shard) if shard != self.shard.index => {
-                    directory.insert(key, shard);
                 }
                 _ => {
-                    directory.remove(&key);
+                    directory.remove(key);
                 }
             }
         }
@@ -350,22 +307,21 @@ impl Router {
             || self.remote_location(key).is_some()
     }
 
-    fn remote_location(&self, key: EntityKey) -> Option<usize> {
+    pub(super) fn actor_path(&self, key: EntityKey) -> Option<String> {
-        let forwarded = self
+        self.local
-            .forwarding
             .read()
             .expect("router lock")
             .get(&key)
-            .copied();
+            .map(LocalActor::path)
-        forwarded
+    }
-            .or_else(|| {
+
+    fn remote_location(&self, key: EntityKey) -> Option<(usize, String)> {
         self.directory
             .read()
             .expect("router lock")
             .get(&key)
-                    .copied()
+            .filter(|(shard, _)| *shard != self.shard.index)
-            })
+            .cloned()
-            .filter(|&shard| shard != self.shard.index)
     }
 
     /// Delivers `message` to the entity, wherever in the cluster it is.
@@ -379,17 +335,17 @@ impl Router {
             .cloned();
         if let Some(actor) = local {
             actor.tell(message, None);
-        } else if let Some(shard) = self.remote_location(key) {
+        } else if let Some((shard, path)) = self.remote_location(key) {
-            self.forward(shard, key, &message);
+            self.forward(shard, &path, &message);
         } else {
             tracing::debug!(entity = %key.name(), "dropped a message for an unknown entity");
         }
     }
 
     /// Sends `message` to the entity on `shard`, bypassing the local registry.
-    pub(super) fn forward<M: Routed>(&self, shard: usize, key: EntityKey, message: &M) {
+    pub(super) fn forward<M: Routed>(&self, shard: usize, path: &str, message: &M) {
         match serde_json::to_value(message) {
-            Ok(payload) => self.send_remote(shard, &key.path(), payload),
+            Ok(payload) => self.send_remote(shard, path, payload),
             Err(error) => tracing::warn!(%error, "failed to serialise a routed message"),
         }
     }
@@ -402,72 +358,37 @@ impl Router {
         }
     }
 
-    /// Hands an entity to the shard owning its new position.
+    /// Recreates an entity actor at the destination path of a Picard migration.
-    pub(super) fn admit_at(&self, shard: usize, admission: Admission) {
+    pub(super) fn admit(&self, admission: Admission, path: &str) -> Result<(), String> {
-        if shard == self.shard.index {
-            self.admit(admission);
-            return;
-        }
-        match serde_json::to_value(&admission) {
-            Ok(payload) => self.send_remote(shard, ADMIT_PATH, payload),
-            Err(error) => tracing::warn!(%error, "failed to serialise an admission"),
-        }
-    }
-
-    /// Takes in an entity: an actor still forwarding for it resumes, otherwise a new one starts.
-    pub(super) fn admit(&self, admission: Admission) {
-        // Holding the arrivals lock keeps `retire_forwarder` from stopping a resumed actor.
         let mut arrivals = self.arrivals.lock().expect("router lock");
-        let key = admission.key();
+        let name = path
-        let existing = self.local.read().expect("router lock").get(&key).cloned();
+            .rsplit('/')
-        let arrival = match (admission, existing) {
+            .next()
-            (Admission::Person(person), Some(LocalActor::Person(actor))) => {
+            .filter(|name| !name.is_empty())
-                actor.tell(PersonMsg::Readmit((*person).clone()), None);
+            .ok_or_else(|| format!("invalid migrated actor path: {path}"))?;
+        let arrival = match admission {
+            Admission::Person(person) => {
+                let actor = self
+                    .spawn_person_named((*person).clone(), name)
+                    .ok_or_else(|| format!("failed to recreate person actor at {path}"))?;
                 Arrival::Person(person, actor)
             }
-            (Admission::Person(person), _) => {
+            Admission::Animal(kind, id, animal) => {
-                let Some(actor) = self.spawn_person((*person).clone()) else {
+                let actor = self
-                    return;
+                    .spawn_animal_named(kind, id, animal, name)
-                };
+                    .ok_or_else(|| format!("failed to recreate animal actor at {path}"))?;
-                Arrival::Person(person, actor)
-            }
-            (Admission::Animal(kind, id, animal), Some(LocalActor::Animal(actor))) => {
-                actor.tell(AnimalMsg::Readmit(animal), None);
                 Arrival::Animal(kind, id, actor)
             }
-            (Admission::Animal(kind, id, animal), _) => {
+            Admission::Cart(cart) => {
-                let Some(actor) = self.spawn_animal(kind, id, animal) else {
-                    return;
-                };
-                Arrival::Animal(kind, id, actor)
-            }
-            (Admission::Cart(cart), Some(LocalActor::Cart(actor))) => {
                 let id = cart.id;
-                actor.tell(CartMsg::Readmit(cart), None);
+                let actor = self
-                Arrival::Cart(id, actor)
+                    .spawn_cart_named(cart, name)
-            }
+                    .ok_or_else(|| format!("failed to recreate cart actor at {path}"))?;
-            (Admission::Cart(cart), _) => {
-                let id = cart.id;
-                let Some(actor) = self.spawn_cart(cart) else {
-                    return;
-                };
                 Arrival::Cart(id, actor)
             }
         };
-        self.forwarding.write().expect("router lock").remove(&key);
         arrivals.push(arrival);
-    }
+        Ok(())
-
-    /// Replaces the forwarding actor of an entity that moved to `shard` by a forwarding entry;
-    /// returns `false` when the entity has come back in the meantime and must keep its actor.
-    pub(super) fn retire_forwarder(&self, key: EntityKey, shard: usize) -> bool {
-        let arrivals = self.arrivals.lock().expect("router lock");
-        let returned = arrivals.iter().any(|arrival| arrival.key() == key);
-        if !returned {
-            self.unregister(key);
-            self.set_forwarding(key, shard);
-        }
-        !returned
     }
 
     pub(super) fn take_arrivals(&self) -> Vec<Arrival> {
@@ -477,13 +398,12 @@ impl Router {
     /// Starts an actor, retrying while a stopped actor with the same name is still shutting down.
     fn start<M: picard::LocalMessage>(
         &self,
-        key: EntityKey,
+        name: &str,
         create: impl Fn(&str) -> Result<ActorRef<M>, CreateError>,
     ) -> Option<ActorRef<M>> {
-        let name = key.name();
         let mut attempts = 0;
         loop {
-            match create(&name) {
+            match create(name) {
                 Ok(actor) => return Some(actor),
                 Err(CreateError::AlreadyExists(_)) if attempts < 200 => {
                     attempts += 1;
@@ -499,9 +419,14 @@ impl Router {
 
     pub(super) fn spawn_person(&self, person: Person) -> Option<ActorRef<PersonMsg>> {
         let key = EntityKey::person(person.id);
-        let actor = self.start(key, |name| {
+        self.spawn_person_named(person, &key.name())
+    }
+
+    fn spawn_person_named(&self, person: Person, name: &str) -> Option<ActorRef<PersonMsg>> {
+        let key = EntityKey::person(person.id);
+        let actor = self.start(name, |name| {
             self.system
-                .actor_of_args::<PersonActor, _>(name, (person.clone(), self.arc()))
+                .actor_of_args::<PersonActor, _>(name, person.clone())
         })?;
         self.register(key, LocalActor::Person(actor.clone()));
         Some(actor)
@@ -514,9 +439,20 @@ impl Router {
         animal: Animal,
     ) -> Option<ActorRef<AnimalMsg>> {
         let key = EntityKey::new(kind, id);
-        let actor = self.start(key, |name| {
+        self.spawn_animal_named(kind, id, animal, &key.name())
+    }
+
+    fn spawn_animal_named(
+        &self,
+        kind: Kind,
+        id: AnimalId,
+        animal: Animal,
+        name: &str,
+    ) -> Option<ActorRef<AnimalMsg>> {
+        let key = EntityKey::new(kind, id);
+        let actor = self.start(name, |name| {
             self.system
-                .actor_of_args::<AnimalActor, _>(name, (key, animal.clone(), self.arc()))
+                .actor_of_args::<AnimalActor, _>(name, (key, animal.clone()))
         })?;
         self.register(key, LocalActor::Animal(actor.clone()));
         Some(actor)
@@ -527,7 +463,7 @@ impl Router {
         partnership: Partnership,
     ) -> Option<ActorRef<PartnershipMsg>> {
         let key = EntityKey::new(Kind::Partnership, u64::from(partnership.id));
-        let actor = self.start(key, |name| {
+        let actor = self.start(&key.name(), |name| {
             self.system
                 .actor_of_args::<PartnershipActor, _>(name, partnership.clone())
         })?;
@@ -537,7 +473,7 @@ impl Router {
 
     pub(super) fn spawn_workplace(&self, workplace: Workplace) -> Option<ActorRef<WorkplaceMsg>> {
         let key = workplace.key();
-        let actor = self.start(key, |name| {
+        let actor = self.start(&key.name(), |name| {
             self.system
                 .actor_of_args::<WorkplaceActor, _>(name, (workplace.clone(), self.arc()))
         })?;
@@ -547,7 +483,7 @@ impl Router {
 
     pub(super) fn spawn_market(&self, market: Market) -> Option<ActorRef<MarketMsg>> {
         let key = market.key();
-        let actor = self.start(key, |name| {
+        let actor = self.start(&key.name(), |name| {
             self.system
                 .actor_of_args::<MarketActor, _>(name, (market.clone(), self.arc()))
         })?;
@@ -557,53 +493,19 @@ impl Router {
 
     pub(super) fn spawn_cart(&self, cart: Cart) -> Option<ActorRef<CartMsg>> {
         let key = cart.key();
-        let actor = self.start(key, |name| {
+        self.spawn_cart_named(cart, &key.name())
+    }
+
+    fn spawn_cart_named(&self, cart: Cart, name: &str) -> Option<ActorRef<CartMsg>> {
+        let key = cart.key();
+        let actor = self.start(name, |name| {
             self.system
-                .actor_of_args::<CartActor, _>(name, (cart.clone(), self.arc()))
+                .actor_of_args::<CartActor, _>(name, cart.clone())
         })?;
         self.register(key, LocalActor::Cart(actor.clone()));
         Some(actor)
     }
 
-    /// Answers a remoting request from another shard: admissions are taken in directly, entity
-    /// messages go to Picard's remote dispatch, and messages for entities that are no longer
-    /// here are forwarded to where they went.
-    pub fn handle(&self, peer: &str, request: RemoteRequest) -> RemoteResponse {
-        let id = request.id;
-        if request.destination.system_id != self.system.id() {
-            return RemoteResponse::rejected(
-                id,
-                RemoteError::Rejected("destination system ID does not match this shard".into()),
-            );
-        }
-        if request.destination.path == ADMIT_PATH {
-            return match serde_json::from_value::<Admission>(request.payload) {
-                Ok(admission) => {
-                    self.admit(admission);
-                    RemoteResponse::accepted(id)
-                }
-                Err(error) => {
-                    RemoteResponse::rejected(id, RemoteError::InvalidMessage(error.to_string()))
-                }
-            };
-        }
-        let path = request.destination.path.clone();
-        let payload = request.payload.clone();
-        let response = (self.dispatch)(peer, request);
-        if !response.accepted
-            && matches!(
-                response.error,
-                Some(RemoteError::ActorNotFound(_) | RemoteError::Rejected(_))
-            )
-            && let Some(key) = EntityKey::from_path(&path)
-            && let Some(shard) = self.remote_location(key)
-        {
-            self.send_remote(shard, &path, payload);
-            return RemoteResponse::accepted(id);
-        }
-        response
-    }
-
     /// Waits until every message queued for other shards has been accepted.
     pub fn flush(&self) {
         let remote = self.remote.read().expect("router lock").clone();
@@ -611,11 +513,59 @@ impl Router {
             remote.flush();
         }
     }
+
+    pub(super) fn migrate(
+        &self,
+        key: EntityKey,
+        shard: usize,
+        admission: Admission,
+    ) -> Result<String, String> {
+        let peer = self
+            .peers
+            .read()
+            .expect("router peer lock")
+            .get(&shard)
+            .cloned()
+            .ok_or_else(|| format!("shard {shard} is not connected"))?;
+        let local = self
+            .local
+            .read()
+            .expect("router lock")
+            .get(&key)
+            .cloned()
+            .ok_or_else(|| format!("actor {} is not local", key.name()))?;
+        let destination_path = format!("/user/{}-m-{}", key.name(), uuid::Uuid::new_v4());
+        let state = serde_json::to_value(admission).map_err(|error| error.to_string())?;
+        self.system
+            .migrate_actor(
+                &local.path(),
+                peer.system_id,
+                Some(peer.endpoint),
+                Some(&destination_path),
+                "sim-entity",
+                state,
+                &peer.connection,
+            )
+            .map_err(|error| error.to_string())?;
+        Ok(destination_path)
+    }
+
+    pub(super) fn register_migration_handler(&self) {
+        let router = self.arc();
+        self.system.register_migration_handler(
+            "sim-entity",
+            Arc::new(move |_system, payload| {
+                let admission: Admission = serde_json::from_value(payload.state.clone())
+                    .map_err(|error| format!("invalid migrated entity: {error}"))?;
+                router.admit(admission, &payload.destination_path)
+            }),
+        );
+    }
 }
 
 #[cfg(test)]
 mod tests {
-    use super::{EntityKey, Kind, Shard};
+    use super::Shard;
 
     #[test]
     fn shards_partition_the_columns_and_hand_out_distinct_ids() {
@@ -643,12 +593,4 @@ mod tests {
         assert_eq!(Shard::new(1, 3).unwrap().share(10), 3);
         assert_eq!(Shard::new(0, 3).unwrap().share(10), 4);
     }
-
-    #[test]
-    fn entity_keys_round_trip_through_actor_paths() {
-        let key = EntityKey::new(Kind::Partnership, 42);
-        assert_eq!(key.path(), "/user/partnership-42");
-        assert_eq!(EntityKey::from_path(&key.path()), Some(key));
-        assert_eq!(EntityKey::from_path("/user/world-inbox"), None);
-    }
 }

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")

ui/Cargo.toml

@@ -11,7 +11,7 @@ leptos = { version = "0.8", features = ["csr"] }
 wasm-bindgen = "=0.2.129"
 wasm-bindgen-futures = "0.4"
 js-sys = "0.3"
-web-sys = { version = "0.3", features = ["CanvasRenderingContext2d", "HtmlCanvasElement", "HtmlElement", "MouseEvent", "WheelEvent", "AddEventListenerOptions", "EventTarget", "WebSocket", "MessageEvent", "Location", "Element", "DomRect", "HtmlDivElement", "Window", "Storage"] }
+web-sys = { version = "0.3", features = ["CanvasRenderingContext2d", "HtmlCanvasElement", "HtmlElement", "HtmlDialogElement", "MouseEvent", "WheelEvent", "AddEventListenerOptions", "EventTarget", "WebSocket", "MessageEvent", "Location", "Element", "DomRect", "HtmlDivElement", "Window", "Storage"] }
 gloo-net = { version = "0.6", default-features = false, features = ["http", "json"] }
 gloo-timers = { version = "0.3", features = ["futures"] }
 serde = { version = "1", features = ["derive"] }

ui/src/api.rs

@@ -12,6 +12,17 @@ thread_local! {
     static SOCKET: RefCell<Option<WebSocket>> = const { RefCell::new(None) };
 }
 
+fn endpoint(path: &str) -> String {
+    let admin = web_sys::window()
+        .and_then(|window| window.location().pathname().ok())
+        .is_some_and(|pathname| pathname == "/admin" || pathname.starts_with("/admin/"));
+    if admin {
+        format!("/admin{path}")
+    } else {
+        path.to_owned()
+    }
+}
+
 /// Opens the live socket and calls `on_state` with each pushed snapshot; reconnects after a drop.
 pub fn connect(on_state: impl Fn(String) + 'static) {
     let Some(loc) = web_sys::window().map(|w| w.location()) else {
@@ -24,7 +35,10 @@ pub fn connect(on_state: impl Fn(String) + 'static) {
         "ws"
     };
     let Ok(host) = loc.host() else { return };
-    open(&format!("{proto}://{host}/ws"), on_state.clone());
+    open(
+        &format!("{proto}://{host}{}", endpoint("/ws")),
+        on_state.clone(),
+    );
 
     spawn_local(async move {
         loop {
@@ -71,16 +85,19 @@ fn open(url: &str, on_state: Rc<dyn Fn(String)>) {
 }
 
 pub async fn get_json<T: DeserializeOwned>(url: &str) -> Option<T> {
-    Request::get(url).send().await.ok()?.json().await.ok()
+    let url = endpoint(url);
+    Request::get(&url).send().await.ok()?.json().await.ok()
 }
 
 pub async fn get_text(url: &str) -> Option<String> {
-    Request::get(url).send().await.ok()?.text().await.ok()
+    let url = endpoint(url);
+    Request::get(&url).send().await.ok()?.text().await.ok()
 }
 
 /// Returns (success, response body).
 pub async fn post(url: &str, body: String) -> Option<(bool, String)> {
-    let r = Request::post(url).body(body).ok()?.send().await.ok()?;
+    let url = endpoint(url);
+    let r = Request::post(&url).body(body).ok()?.send().await.ok()?;
     Some((r.ok(), r.text().await.ok()?))
 }
 

ui/src/lib.rs

@@ -39,9 +39,15 @@ fn App() -> impl IntoView {
     let tool = RwSignal::new("inspect".to_string());
     let param = RwSignal::new(String::new());
     let amount = RwSignal::new("5".to_string());
+    let can_edit = RwSignal::new(false);
+    let about_dialog = NodeRef::<leptos::html::Dialog>::new();
     // (selection counter, people at the time of selection); the counter forces editor rows to rebuild.
     let editing = RwSignal::new((0u64, Vec::<Person>::new()));
 
+    spawn_local(async move {
+        can_edit.set(api::get_json::<bool>("/api/access").await.unwrap_or(false));
+    });
+
     // The server pushes a snapshot per tick; terrain is refetched only when its revision changes.
     let last_rev = StoredValue::new(None::<u32>);
     api::connect(move |text| {
@@ -288,7 +294,49 @@ fn App() -> impl IntoView {
             ></canvas>
             </div>
             <div class="side">
+                <div class="panel-heading">
                     <h3>"World"</h3>
+                    <button on:click=move |_| {
+                        if let Some(dialog) = about_dialog.get() {
+                            let _ = dialog.show_modal();
+                        }
+                    } aria-label="About World Simulation">"About"</button>
+                </div>
+                <dialog class="about-dialog" node_ref=about_dialog>
+                    <div class="about-header">
+                        <div>
+                            <p class="about-kicker">"WORLD SIMULATION"</p>
+                            <h2>"A living world, running in shards"</h2>
+                        </div>
+                        <button
+                            class="about-close"
+                            on:click=move |_| {
+                                if let Some(dialog) = about_dialog.get() {
+                                    dialog.close();
+                                }
+                            }
+                            aria-label="Close about dialog"
+                        >"Close"</button>
+                    </div>
+                    <p>
+                        "This is an ongoing simulation of a small world. People and wildlife move, gather resources, work, trade, form families, and respond to the rules in the world's Rune behavior script."
+                    </p>
+                    <h3>"One world, several shards"</h3>
+                    <p>
+                        "The map is divided into vertical regions called shards. Each shard runs its own part of the simulation, and entities that cross a boundary are handed to the shard that owns their new position. The gateway coordinates their steps and combines their views into this map."
+                    </p>
+                    <h3>"Actors and persistence"</h3>
+                    <p>
+                        "Each person, animal, partnership, market, workplace, and cart is managed by an actor. Shards save their own world snapshots, so the simulation can resume after a restart."
+                    </p>
+                    <div class="about-actions">
+                        <button on:click=move |_| {
+                            if let Some(dialog) = about_dialog.get() {
+                                dialog.close();
+                            }
+                        } autofocus=true>"Done"</button>
+                    </div>
+                </dialog>
                 {move || state.with(|s| s.as_ref().map(|s| {
                     let (h, m) = (s.hour.floor(), (s.hour.fract() * 60.0).floor());
                     view! {
@@ -333,6 +381,10 @@ fn App() -> impl IntoView {
                         )}</div>
                     }
                 }))}
+                <Show
+                    when=move || can_edit.get()
+                    fallback=|| view! { <p class="readonly">"View-only access"</p> }
+                >
                 <div>
                     <button on:click=move |_| {
                         api::edit(format!("pause {}", u8::from(!paused())));
@@ -465,6 +517,7 @@ fn App() -> impl IntoView {
 
                 <h3>"Behaviour script (Rune)"</h3>
                 <ScriptEditor />
+                </Show>
 
                 <h3>"Events"</h3>
                 <div class="log">

ui/style.css

@@ -48,6 +48,73 @@ canvas {
     box-sizing: border-box;
 }
 
+.panel-heading {
+    display: flex;
+    align-items: center;
+    justify-content: space-between;
+}
+
+.panel-heading h3 {
+    margin: 1em 0;
+}
+
+.about-dialog {
+    width: min(560px, calc(100vw - 40px));
+    max-height: min(640px, calc(100vh - 40px));
+    padding: 24px;
+    overflow: auto;
+    border: 1px solid #56615c;
+    border-top: 3px solid #9bbf8a;
+    border-radius: 4px;
+    background: #171b19;
+    color: #ddd;
+    font: inherit;
+    line-height: 1.6;
+}
+
+.about-dialog::backdrop {
+    background: rgb(0 0 0 / 72%);
+}
+
+.about-dialog h2 {
+    margin: 4px 0 12px;
+    font-size: 20px;
+    line-height: 1.3;
+}
+
+.about-dialog h3 {
+    margin: 22px 0 4px;
+    color: #b7d1a8;
+    font-size: 14px;
+}
+
+.about-dialog p {
+    margin: 8px 0;
+    color: #c2c9c4;
+}
+
+.about-header {
+    display: flex;
+    justify-content: space-between;
+    align-items: flex-start;
+    gap: 16px;
+}
+
+.about-kicker {
+    color: #a7c895 !important;
+    font-size: 11px;
+}
+
+.about-close {
+    flex: none;
+}
+
+.about-actions {
+    display: flex;
+    justify-content: flex-end;
+    margin-top: 20px;
+}
+
 .side input[type=number] {
     width: 56px;
 }