feat: shard-aware world with transparent actor transfers

Add a router that delivers actor messages locally or to the shard that
currently hosts the entity. People and animals that leave their shard's
columns depart with their full state, forward messages while in flight
and are re-admitted on the destination shard. Enlarge the default world.

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

CHANGELOG.md

@@ -7,9 +7,11 @@ Notable changes to this project are documented here.
 ### Added
 
 - 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.
 
 ### Changed
 
 - People, rabbits, wolves, and partnerships are Picard actors that own their state; the world only reaches them through messages. Actors are named `person-<id>`, `rabbit-<id>`, `wolf-<id>`, and `partnership-<id>` instead of `entity-<id>`.
 - Rabbits killed by hunters no longer move or breed later in the same tick.
 - Update Picard to `054449c`.
+- The default world is larger: 192×128 tiles with 180 people, 1200 rabbits, 80 wolves, and twice the food and houses.

config.rn

@@ -2,8 +2,8 @@
 // `pub fn` items are entity behaviours. Every constant below is required.
 
 // World
-pub const WIDTH = 96;
-pub const HEIGHT = 64;
+pub const WIDTH = 192;
+pub const HEIGHT = 128;
 
 // Calendar
 pub const TICKS_PER_DAY = 1800;
@@ -21,20 +21,20 @@ pub const OLDER_ADULT = #{ min_days: 10, max_days: 20, fertile: true };
 pub const ELDERLY = #{ min_days: 4, max_days: 10, fertile: false };
 
 // Initial world
-pub const INITIAL_POPULATION = 90;
-pub const INITIAL_YOUNG_ADULTS = 30;
-pub const INITIAL_RABBITS = 600;
-pub const INITIAL_WOLVES = 40;
-pub const INITIAL_FOOD = 600;
+pub const INITIAL_POPULATION = 180;
+pub const INITIAL_YOUNG_ADULTS = 60;
+pub const INITIAL_RABBITS = 1200;
+pub const INITIAL_WOLVES = 80;
+pub const INITIAL_FOOD = 1200;
 pub const INITIAL_PERSON_HUNGER = 10.0;
 pub const INITIAL_ANIMAL_ENERGY = 70.0;
 pub const ANIMAL_ENERGY_DEATH_THRESHOLD = 0.0;
 pub const STARTING_TICK_FRACTION = 4;
 
 // Population and housing
-pub const POPULATION_CAPACITY = 12;
+pub const POPULATION_CAPACITY = 30;
 pub const HOUSE_CAPACITY = 3;
-pub const MAX_HOUSES = 10;
+pub const MAX_HOUSES = 20;
 
 // Daylight
 pub const DAYLIGHT_NIGHT_THRESHOLD = 0.2;
@@ -92,9 +92,9 @@ pub const STONE_REGROW_AMOUNT = 6;
 
 // Animal spawning
 pub const ANIMAL_POPULATION_INTERVAL = 50;
-pub const RABBIT_SPAWN_CAP = 15;
-pub const WOLF_SPAWN_CAP = 2;
-pub const WOLF_SPAWN_RABBIT_THRESHOLD = 40;
+pub const RABBIT_SPAWN_CAP = 30;
+pub const WOLF_SPAWN_CAP = 4;
+pub const WOLF_SPAWN_RABBIT_THRESHOLD = 80;
 pub const WOLF_MIN_HOME_DISTANCE = 15;
 
 // Rabbits
@@ -103,7 +103,7 @@ pub const RABBIT_ENERGY_REGEN = 0.2;
 pub const RABBIT_ENERGY_MAX = 100.0;
 pub const RABBIT_WOLF_DETECTION_RANGE = 4;
 pub const RABBIT_BREED_ENERGY = 80.0;
-pub const RABBIT_MAX_POPULATION = 120;
+pub const RABBIT_MAX_POPULATION = 300;
 pub const RABBIT_BREED_CHANCE = 0.01;
 pub const RABBIT_BREED_ENERGY_COST = 30.0;
 pub const RABBIT_NEWBORN_ENERGY = 50.0;
@@ -115,7 +115,7 @@ pub const WOLF_HUNT_SUCCESS_CHANCE = 0.4;
 pub const WOLF_HUNT_ENERGY_GAIN = 50.0;
 pub const WOLF_ENERGY_MAX = 100.0;
 pub const WOLF_BREED_ENERGY = 90.0;
-pub const WOLF_MAX_POPULATION = 10;
+pub const WOLF_MAX_POPULATION = 24;
 pub const WOLF_BREED_CHANCE = 0.02;
 pub const WOLF_BREED_ENERGY_COST = 40.0;
 pub const WOLF_NEWBORN_ENERGY = 50.0;

src/main.rs

@@ -13,7 +13,7 @@ use std::sync::{Arc, Mutex};
 use std::time::{Duration, Instant};
 use storage::Storage;
 use tokio::sync::{mpsc, oneshot, watch};
-use world::{SimScript, World};
+use world::{Shard, SimScript, World};
 
 // The UI bundle that build.rs produced from ui/.
 static UI: Dir = include_dir!("$OUT_DIR/ui-pkg");
@@ -65,9 +65,11 @@ fn run_sim(
     snapshot_tx: &watch::Sender<Option<String>>,
 ) {
     let mut world = match snapshot {
-        Some(snapshot) => World::from_snapshot_with_script(&snapshot, seed, sim_script)
-            .unwrap_or_else(|e| panic!("failed to restore simulation snapshot: {e}")),
-        None => World::new_with_script(seed, sim_script),
+        Some(snapshot) => {
+            World::from_snapshot_with_script(&snapshot, seed, sim_script, Shard::SINGLE)
+                .unwrap_or_else(|e| panic!("failed to restore simulation snapshot: {e}"))
+        }
+        None => World::new_shard(seed, sim_script, Shard::SINGLE),
     };
     *script.lock().unwrap() = world.script().to_string();
     let mut rev = u32::MAX;

src/world/actors.rs

@@ -4,6 +4,7 @@
 //! sends them the effects of world-level decisions (movement, gathering, hunting, births) and
 //! lets them apply entity-local rules (ageing, hunger, eating, energy, dying) themselves.
 
+use super::router::{Admission, EntityKey, Router};
 use super::{Animal, CalendarSettings, LifeStage, Partnership, Person, Rng, sample_stage_duration};
 use picard::actors::{
     Actor, ActorFactoryArgs, ActorRef, ActorRefFactory, ActorSystem, Context as ActorContext,
@@ -12,6 +13,7 @@ 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;
 
@@ -93,6 +95,11 @@ pub(super) enum PersonMsg {
     Edit(PersonEdit),
     SetPartner(Option<u32>),
     MoveTo((usize, usize)),
+    /// 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.
@@ -122,6 +129,8 @@ 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)]
@@ -149,6 +158,7 @@ pub(super) enum ReplyBody {
     Hunted(AnimalId, bool),
     Partnership(Partnership),
     BirthDue(Partnership, bool),
+    Departed,
 }
 
 #[derive(Clone, Debug, Deserialize, Serialize)]
@@ -166,15 +176,22 @@ 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> for PersonActor {
-    fn create_args(mut person: Person) -> Self {
+impl ActorFactoryArgs<(Person, Arc<Router>)> for PersonActor {
+    fn create_args((mut person, router): (Person, Arc<Router>)) -> 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 { person }
+        Self {
+            person,
+            router,
+            forward_to: None,
+        }
     }
 }
 
@@ -254,6 +271,16 @@ 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()));
@@ -277,18 +304,34 @@ impl Actor for PersonActor {
             PersonMsg::Edit(edit) => self.edit(edit),
             PersonMsg::SetPartner(partner_id) => self.person.partner_id = partner_id,
             PersonMsg::MoveTo(position) => (self.person.x, self.person.y) = position,
+            PersonMsg::Depart(request, shard) => {
+                self.router
+                    .admit_at(shard, Admission::Person(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<(AnimalId, Animal)> for AnimalActor {
-    fn create_args((id, animal): (AnimalId, Animal)) -> Self {
-        Self { id, animal }
+impl ActorFactoryArgs<(EntityKey, Animal, Arc<Router>)> for AnimalActor {
+    fn create_args((key, animal, router): (EntityKey, Animal, Arc<Router>)) -> Self {
+        Self {
+            key,
+            id: key.id,
+            animal,
+            router,
+            forward_to: None,
+        }
     }
 }
 
@@ -330,6 +373,15 @@ 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(
@@ -360,6 +412,13 @@ 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/router.rs

@@ -0,0 +1,557 @@
+//! Location-transparent addressing of entity actors.
+//!
+//! Actors never talk to each other through raw actor references. They send messages to an
+//! [`EntityKey`] through the [`Router`], which delivers them to the local actor or, when the
+//! entity lives on another shard, serialises them and sends them over Picard remoting. Entities
+//! that leave a shard's region are handed to the shard that owns their new position as an
+//! [`Admission`]; the departed actor keeps forwarding messages until every shard knows where the
+//! entity went.
+
+use super::actors::{
+    AnimalActor, AnimalId, AnimalMsg, PartnershipActor, PartnershipMsg, PersonActor, PersonMsg,
+};
+use super::{Animal, Partnership, Person};
+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 {
+    Person,
+    Rabbit,
+    Wolf,
+    Partnership,
+}
+
+impl Kind {
+    const ALL: [Kind; 4] = [Kind::Person, Kind::Rabbit, Kind::Wolf, Kind::Partnership];
+
+    fn name(self) -> &'static str {
+        match self {
+            Kind::Person => "person",
+            Kind::Rabbit => "rabbit",
+            Kind::Wolf => "wolf",
+            Kind::Partnership => "partnership",
+        }
+    }
+}
+
+/// Cluster-wide identity of an entity actor; ids are unique per kind across all shards.
+#[derive(Clone, Copy, Debug, Deserialize, Serialize, PartialEq, Eq, Hash, PartialOrd, Ord)]
+pub struct EntityKey {
+    pub kind: Kind,
+    pub id: u64,
+}
+
+impl EntityKey {
+    pub fn new(kind: Kind, id: u64) -> Self {
+        Self { kind, id }
+    }
+
+    pub fn person(id: u32) -> Self {
+        Self::new(Kind::Person, u64::from(id))
+    }
+
+    /// The actor name, which is also the last segment of its path.
+    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.
+#[derive(Clone, Copy, Debug, Deserialize, Serialize, PartialEq, Eq)]
+pub struct Shard {
+    pub index: usize,
+    pub count: usize,
+}
+
+impl Shard {
+    pub const SINGLE: Shard = Shard { index: 0, count: 1 };
+
+    pub fn new(index: usize, count: usize) -> Option<Self> {
+        (count > 0 && index < count).then_some(Self { index, count })
+    }
+
+    /// The columns of shard `index` in a world `width` tiles wide.
+    pub fn columns_of(&self, index: usize, width: usize) -> Range<usize> {
+        width * index / self.count..width * (index + 1) / self.count
+    }
+
+    pub fn columns(&self, width: usize) -> Range<usize> {
+        self.columns_of(self.index, width)
+    }
+
+    /// The shard owning column `x`.
+    pub fn owner(&self, x: usize, width: usize) -> usize {
+        (0..self.count)
+            .find(|&index| self.columns_of(index, width).contains(&x))
+            .unwrap_or(self.count - 1)
+    }
+
+    /// First id of this shard's id sequences; ids advance by [`Shard::count`] so that every
+    /// shard hands out distinct ids without coordination.
+    pub fn first_id(&self) -> u64 {
+        self.index as u64 + 1
+    }
+
+    pub fn id_stride(&self) -> u64 {
+        self.count as u64
+    }
+
+    /// This shard's share of a cluster-wide `total`.
+    pub fn share(&self, total: usize) -> usize {
+        total / self.count + usize::from(self.index < total % self.count)
+    }
+}
+
+/// 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
+    /// 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.
+    fn flush(&self);
+}
+
+/// The live actor behind a local entity.
+#[derive(Clone)]
+pub(super) enum LocalActor {
+    Person(ActorRef<PersonMsg>),
+    Animal(ActorRef<AnimalMsg>),
+    Partnership(ActorRef<PartnershipMsg>),
+}
+
+impl LocalActor {
+    pub(super) fn stop(self) {
+        match self {
+            LocalActor::Person(actor) => actor.stop(),
+            LocalActor::Animal(actor) => actor.stop(),
+            LocalActor::Partnership(actor) => actor.stop(),
+        }
+    }
+}
+
+/// A message type that can be routed to the entities whose actors accept it.
+pub(super) trait Routed: picard::Message + Clone {
+    fn local(actor: &LocalActor) -> Option<&ActorRef<Self>>;
+}
+
+impl Routed for PersonMsg {
+    fn local(actor: &LocalActor) -> Option<&ActorRef<Self>> {
+        match actor {
+            LocalActor::Person(actor) => Some(actor),
+            _ => None,
+        }
+    }
+}
+
+impl Routed for AnimalMsg {
+    fn local(actor: &LocalActor) -> Option<&ActorRef<Self>> {
+        match actor {
+            LocalActor::Animal(actor) => Some(actor),
+            _ => None,
+        }
+    }
+}
+
+impl Routed for PartnershipMsg {
+    fn local(actor: &LocalActor) -> Option<&ActorRef<Self>> {
+        match actor {
+            LocalActor::Partnership(actor) => Some(actor),
+            _ => None,
+        }
+    }
+}
+
+/// An entity's complete state, including its inventory, moving into another shard's region.
+#[derive(Clone, Debug, Deserialize, Serialize)]
+pub(super) enum Admission {
+    Person(Person),
+    Animal(Kind, AnimalId, Animal),
+}
+
+impl Admission {
+    fn key(&self) -> EntityKey {
+        match self {
+            Admission::Person(person) => EntityKey::person(person.id),
+            Admission::Animal(kind, id, _) => EntityKey::new(*kind, *id),
+        }
+    }
+}
+
+/// An admitted entity, waiting for the world to pick it up at the start of its next step.
+pub(super) enum Arrival {
+    Person(Person, ActorRef<PersonMsg>),
+    Animal(Kind, AnimalId, ActorRef<AnimalMsg>),
+}
+
+impl Arrival {
+    fn key(&self) -> EntityKey {
+        match self {
+            Arrival::Person(person, _) => EntityKey::person(person.id),
+            Arrival::Animal(kind, id, ..) => EntityKey::new(*kind, *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.
+    directory: RwLock<HashMap<EntityKey, usize>>,
+    /// Where entities that left this shard went, until the directory catches up.
+    forwarding: RwLock<HashMap<EntityKey, usize>>,
+    remote: RwLock<Option<Arc<dyn RemoteSender>>>,
+    arrivals: Mutex<Vec<Arrival>>,
+}
+
+impl Router {
+    pub fn new(shard: Shard, system: &ActorSystem) -> Arc<Self> {
+        Arc::new_cyclic(|this| Self {
+            shard,
+            system: system.clone(),
+            this: this.clone(),
+            dispatch: system.remote_handler(),
+            local: RwLock::default(),
+            directory: RwLock::default(),
+            forwarding: RwLock::default(),
+            remote: RwLock::default(),
+            arrivals: Mutex::default(),
+        })
+    }
+
+    /// Identity of this shard's actor system, which remote requests must be addressed to.
+    pub fn system_id(&self) -> uuid::Uuid {
+        self.system.id()
+    }
+
+    pub fn set_remote(&self, remote: Arc<dyn RemoteSender>) {
+        *self.remote.write().expect("router lock") = Some(remote);
+    }
+
+    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);
+    }
+
+    pub(super) fn unregister(&self, key: EntityKey) {
+        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>)]) {
+        let mut directory = self.directory.write().expect("router lock");
+        let mut forwarding = self.forwarding.write().expect("router lock");
+        for &(key, shard) in changes {
+            forwarding.remove(&key);
+            match shard {
+                Some(shard) if shard != self.shard.index => {
+                    directory.insert(key, shard);
+                }
+                _ => {
+                    directory.remove(&key);
+                }
+            }
+        }
+    }
+
+    /// Whether the entity exists anywhere in the cluster, as far as this shard knows.
+    pub fn knows(&self, key: EntityKey) -> bool {
+        self.local.read().expect("router lock").contains_key(&key)
+            || self.remote_location(key).is_some()
+    }
+
+    fn remote_location(&self, key: EntityKey) -> Option<usize> {
+        let forwarded = self
+            .forwarding
+            .read()
+            .expect("router lock")
+            .get(&key)
+            .copied();
+        forwarded
+            .or_else(|| {
+                self.directory
+                    .read()
+                    .expect("router lock")
+                    .get(&key)
+                    .copied()
+            })
+            .filter(|&shard| shard != self.shard.index)
+    }
+
+    /// Delivers `message` to the entity, wherever in the cluster it is.
+    pub(super) fn tell<M: Routed>(&self, key: EntityKey, message: M) {
+        let local = self
+            .local
+            .read()
+            .expect("router lock")
+            .get(&key)
+            .and_then(M::local)
+            .cloned();
+        if let Some(actor) = local {
+            actor.tell(message, None);
+        } else if let Some(shard) = self.remote_location(key) {
+            self.forward(shard, key, &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) {
+        match serde_json::to_value(message) {
+            Ok(payload) => self.send_remote(shard, &key.path(), payload),
+            Err(error) => tracing::warn!(%error, "failed to serialise a routed message"),
+        }
+    }
+
+    fn send_remote(&self, shard: usize, path: &str, payload: serde_json::Value) {
+        let remote = self.remote.read().expect("router lock").clone();
+        match remote {
+            Some(remote) => remote.send(shard, path, payload),
+            None => tracing::debug!(path, "dropped a message for a shard without a connection"),
+        }
+    }
+
+    /// Hands an entity to the shard owning its new position.
+    pub(super) fn admit_at(&self, shard: usize, admission: Admission) {
+        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 existing = self.local.read().expect("router lock").get(&key).cloned();
+        let arrival = match (admission, existing) {
+            (Admission::Person(person), Some(LocalActor::Person(actor))) => {
+                actor.tell(PersonMsg::Readmit(person.clone()), None);
+                Arrival::Person(person, actor)
+            }
+            (Admission::Person(person), _) => {
+                let Some(actor) = self.spawn_person(person.clone()) else {
+                    return;
+                };
+                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), _) => {
+                let Some(actor) = self.spawn_animal(kind, id, animal) else {
+                    return;
+                };
+                Arrival::Animal(kind, id, actor)
+            }
+        };
+        self.forwarding.write().expect("router lock").remove(&key);
+        arrivals.push(arrival);
+    }
+
+    /// 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> {
+        std::mem::take(&mut *self.arrivals.lock().expect("router lock"))
+    }
+
+    /// Starts an actor, retrying while a stopped actor with the same name is still shutting down.
+    fn start<M: picard::LocalMessage>(
+        &self,
+        key: EntityKey,
+        create: impl Fn(&str) -> Result<ActorRef<M>, CreateError>,
+    ) -> Option<ActorRef<M>> {
+        let name = key.name();
+        let mut attempts = 0;
+        loop {
+            match create(&name) {
+                Ok(actor) => return Some(actor),
+                Err(CreateError::AlreadyExists(_)) if attempts < 200 => {
+                    attempts += 1;
+                    std::thread::sleep(Duration::from_millis(5));
+                }
+                Err(error) => {
+                    tracing::error!(entity = %name, %error, "failed to start an entity actor");
+                    return None;
+                }
+            }
+        }
+    }
+
+    pub(super) fn spawn_person(&self, person: Person) -> Option<ActorRef<PersonMsg>> {
+        let key = EntityKey::person(person.id);
+        let actor = self.start(key, |name| {
+            self.system
+                .actor_of_args::<PersonActor, _>(name, (person.clone(), self.arc()))
+        })?;
+        self.register(key, LocalActor::Person(actor.clone()));
+        Some(actor)
+    }
+
+    pub(super) fn spawn_animal(
+        &self,
+        kind: Kind,
+        id: AnimalId,
+        animal: Animal,
+    ) -> Option<ActorRef<AnimalMsg>> {
+        let key = EntityKey::new(kind, id);
+        let actor = self.start(key, |name| {
+            self.system
+                .actor_of_args::<AnimalActor, _>(name, (key, animal.clone(), self.arc()))
+        })?;
+        self.register(key, LocalActor::Animal(actor.clone()));
+        Some(actor)
+    }
+
+    pub(super) fn spawn_partnership(
+        &self,
+        partnership: Partnership,
+    ) -> Option<ActorRef<PartnershipMsg>> {
+        let key = EntityKey::new(Kind::Partnership, u64::from(partnership.id));
+        let actor = self.start(key, |name| {
+            self.system
+                .actor_of_args::<PartnershipActor, _>(name, partnership.clone())
+        })?;
+        self.register(key, LocalActor::Partnership(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();
+        if let Some(remote) = remote {
+            remote.flush();
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::{EntityKey, Kind, Shard};
+
+    #[test]
+    fn shards_partition_the_columns_and_hand_out_distinct_ids() {
+        for (width, count) in [(192, 3), (10, 3), (97, 4), (8, 1)] {
+            let mut owners = Vec::new();
+            for index in 0..count {
+                let shard = Shard::new(index, count).unwrap();
+                for x in shard.columns(width) {
+                    assert_eq!(shard.owner(x, width), index);
+                    owners.push(x);
+                }
+            }
+            assert_eq!(owners, (0..width).collect::<Vec<_>>());
+        }
+        let ids: Vec<u64> = (0..3)
+            .flat_map(|index| {
+                let shard = Shard::new(index, 3).unwrap();
+                (0..4).map(move |n| shard.first_id() + n * shard.id_stride())
+            })
+            .collect();
+        let mut unique = ids.clone();
+        unique.sort_unstable();
+        unique.dedup();
+        assert_eq!(unique.len(), ids.len());
+        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);
+    }
+}