test: cross parallel tick helper thresholds with real resources

Compare deferred and in-place metadata for 2,085 varied mobs, including pack-selected effect particles and switch-off vector reuse. Exercise live climbing, snapshot order and nearest-player invalidation for 293 mobs. Build 132 random-tick section masks ahead and retain exact event, state and random comparisons through edits and reloads.

Co-Authored-By: Kacper <[email protected]>
Co-Authored-By: GPT 6.1 Sol <[email protected]>
(cherry picked from commit 41a579d428)

Cargo.lock

@@ -706,6 +706,25 @@ dependencies = [
  "cfg-if",
 ]
 
+[[package]]
+name = "crossbeam-deque"
+version = "0.8.8"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "622f3fc73690be383c7214310406f28a90e6edeadc3cea882f9d71e495b9711a"
+dependencies = [
+ "crossbeam-epoch",
+ "crossbeam-utils",
+]
+
+[[package]]
+name = "crossbeam-epoch"
+version = "0.9.21"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "dc74980687109a3b14c72fd458107bf0baa1da1a1a805e178d15501ba9b86d9d"
+dependencies = [
+ "crossbeam-utils",
+]
+
 [[package]]
 name = "crossbeam-queue"
 version = "0.3.14"
@@ -926,6 +945,12 @@ version = "1.0.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "92773504d58c093f6de2459af4af33faa518c13451eb8f2b5698ed3d36e7c813"
 
+[[package]]
+name = "either"
+version = "1.19.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0e9c71c2167ca323c882b99918929403426e2373ea17242ff5653e0d5e1058be"
+
 [[package]]
 name = "embedded-io"
 version = "0.4.0"
@@ -1554,6 +1579,7 @@ dependencies = [
  "open",
  "png",
  "pollster",
+ "rayon",
  "rustix 1.1.5",
  "serde",
  "serde_json",
@@ -3654,6 +3680,26 @@ dependencies = [
  "objc2-quartz-core 0.3.2",
 ]
 
+[[package]]
+name = "rayon"
+version = "1.12.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "fb39b166781f92d482534ef4b4b1b2568f42613b53e5b6c160e24cfbfa30926d"
+dependencies = [
+ "either",
+ "rayon-core",
+]
+
+[[package]]
+name = "rayon-core"
+version = "1.13.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "22e18b0f0062d30d4230b2e85ff77fdfe4326feb054b9783a3460d8435c8ab91"
+dependencies = [
+ "crossbeam-deque",
+ "crossbeam-utils",
+]
+
 [[package]]
 name = "rc2"
 version = "0.8.1"

crates/horizon-client-lib/Cargo.toml

@@ -27,6 +27,9 @@ horizon-mcp = { path = "../horizon-mcp" }
 # The server thread ahead of the workers (MinecraftServer.spin's Thread.setPriority).
 horizon-optifine = { path = "../horizon-optifine" }
 horizon-threads = { path = "../horizon-threads" }
+# One pool of helper threads for the server tick's parallel phases (`parallel.rs`): threads made
+# once, not for every phase of every tick.
+rayon = "1.10"
 horizon-hdr = { path = "../horizon-hdr" }
 horizon-world = { path = "../horizon-world" }
 horizon-protocol = { path = "../horizon-protocol" }

crates/horizon-client-render/src/options_ui_pages.rs

@@ -171,7 +171,7 @@ impl OptionsUi {
                 added::Row::Text { text, .. } => {
                     out.extend(wrap(&text, 64).into_iter().map(Row::Note))
                 }
-                added::Row::Space(_) | added::Row::Field { .. } => continue,
+                added::Row::Space(_) => continue,
                 added::Row::Button {
                     label,
                     active,

crates/horizon-client-render/src/poi_sections.rs

@@ -7,7 +7,10 @@
 //! sections are immutable behind `Arc` and edited through `Arc::make_mut`, so
 //! a weak reference tells an edited or replaced section from the one the list
 //! was built from (see `random_tick_sections`), without keeping its blocks.
-use crate::{chunk::Section, fast_hash::FxHashMap};
+use crate::{
+    chunk::{Chunk, Section},
+    fast_hash::FxHashMap,
+};
 use std::sync::{Arc, Mutex, Weak};
 
 /// A section's POIs: the block index in the section (`y << 8 | z << 4 | x`)
@@ -18,13 +21,16 @@ struct Cached<T> {
     section: Weak<Section>,
     pois: SectionPois<T>,
 }
-/// Cached POI lists by section, `(chunk x, section index, chunk z)`.
+/// Cached POI lists by chunk column `(chunk x, chunk z)`, each by section
+/// index: a search through a column looks it up once.
 pub struct PoiSections<T> {
     sections: Mutex<Lists<T>>,
 }
 struct Lists<T> {
-    by_section: FxHashMap<(i32, i32, i32), Cached<T>>,
-    /// Size above which lists of sections no longer resident are dropped.
+    by_column: FxHashMap<(i32, i32), Vec<Option<Cached<T>>>>,
+    /// How many section lists are kept, and the count above which the lists
+    /// of sections no longer resident are dropped.
+    len: usize,
     prune_at: usize,
 }
 const MIN_PRUNE: usize = 4096;
@@ -32,27 +38,61 @@ impl<T> Default for PoiSections<T> {
     fn default() -> Self {
         Self {
             sections: Mutex::new(Lists {
-                by_section: FxHashMap::default(),
+                by_column: FxHashMap::default(),
+                len: 0,
                 prune_at: MIN_PRUNE,
             }),
         }
     }
 }
-impl<T: Copy> PoiSections<T> {
-    /// The POIs of `section`, whose block states `kind` classifies. `kind`
-    /// must give the same answer for a state on every call.
-    pub fn get(
-        &self,
-        key: (i32, i32, i32),
-        section: &Arc<Section>,
-        mut kind: impl FnMut(u32) -> Option<T>,
-    ) -> SectionPois<T> {
-        let mut lists = self.sections.lock().unwrap_or_else(|e| e.into_inner());
-        if let Some(cached) = lists.by_section.get(&key)
-            && cached.section.as_ptr() == Arc::as_ptr(section)
-        {
-            return cached.pois.clone();
+impl<T: Copy> Lists<T> {
+    /// The column's lists, for sections up to `last`, after dropping the
+    /// lists of sections no longer resident if there are too many.
+    fn column(
+        &mut self,
+        column: (i32, i32),
+        last: usize,
+    ) -> (&mut [Option<Cached<T>>], &mut usize) {
+        if self.len >= self.prune_at {
+            self.prune();
         }
+        let lists = self.by_column.entry(column).or_default();
+        if lists.len() <= last {
+            lists.resize_with(last + 1, || None);
+        }
+        (lists, &mut self.len)
+    }
+
+    /// Drops the lists of sections no longer resident.
+    fn prune(&mut self) {
+        let mut len = 0;
+        self.by_column.retain(|_, lists| {
+            for c in lists.iter_mut() {
+                if c.as_ref().is_some_and(|c| c.section.strong_count() == 0) {
+                    *c = None;
+                }
+            }
+            let kept = lists.iter().filter(|c| c.is_some()).count();
+            len += kept;
+            kept > 0
+        });
+        self.len = len;
+        self.prune_at = (len * 2).max(MIN_PRUNE);
+    }
+}
+
+/// The POIs of `section` from `slot`, listed again (and kept there) unless
+/// `slot` was listed from this very section; `len` counts the slots filled.
+fn list<'a, T: Copy>(
+    slot: &'a mut Option<Cached<T>>,
+    len: &mut usize,
+    section: &Arc<Section>,
+    kind: &mut impl FnMut(u32) -> Option<T>,
+) -> &'a SectionPois<T> {
+    let fresh = slot
+        .as_ref()
+        .is_some_and(|c| c.section.as_ptr() == Arc::as_ptr(section));
+    if !fresh {
         let pois: SectionPois<T> = if section.blocks.maybe_has(|v| kind(v).is_some()) {
             let mut states = [0u32; 4096];
             section.unpack(&mut states);
@@ -64,18 +104,56 @@ impl<T: Copy> PoiSections<T> {
         } else {
             Arc::from([])
         };
-        if lists.by_section.len() >= lists.prune_at {
-            lists.by_section.retain(|_, c| c.section.strong_count() > 0);
-            lists.prune_at = (lists.by_section.len() * 2).max(MIN_PRUNE);
+        *len += usize::from(slot.is_none());
+        *slot = Some(Cached {
+            section: Arc::downgrade(section),
+            pois,
+        });
+    }
+    &slot.as_ref().expect("just listed").pois
+}
+
+impl<T: Copy> PoiSections<T> {
+    /// The POIs of `section`, the `key.1`-th section of the column
+    /// `(key.0, key.2)`, whose block states `kind` classifies. `kind` must give
+    /// the same answer for a state on every call.
+    pub fn get(
+        &self,
+        key: (i32, i32, i32),
+        section: &Arc<Section>,
+        mut kind: impl FnMut(u32) -> Option<T>,
+    ) -> SectionPois<T> {
+        let mut lists = self.sections.lock().unwrap_or_else(|e| e.into_inner());
+        let s = usize::try_from(key.1).expect("a section index");
+        let (column, len) = lists.column((key.0, key.2), s);
+        list(&mut column[s], len, section, &mut kind).clone()
+    }
+
+    /// The POIs of `chunk` from block Y `y0` to `y1`, its first section at
+    /// `min_y`, as [`Self::get`] gives them: each section that has any, with
+    /// the Y of its lowest blocks, in order, put in `out`. One lookup of the
+    /// column.
+    pub fn column(
+        &self,
+        chunk: &Chunk,
+        min_y: i32,
+        (y0, y1): (i32, i32),
+        mut kind: impl FnMut(u32) -> Option<T>,
+        out: &mut Vec<(i32, SectionPois<T>)>,
+    ) {
+        out.clear();
+        let (first, last) = ((y0 - min_y) >> 4, (y1 - min_y) >> 4);
+        if first > last {
+            return;
+        }
+        let mut lists = self.sections.lock().unwrap_or_else(|e| e.into_inner());
+        let (slots, len) = lists.column((chunk.x, chunk.z), last as usize);
+        for s in first as usize..=last as usize {
+            let pois = list(&mut slots[s], len, &chunk.sections[s], &mut kind);
+            if !pois.is_empty() {
+                out.push((min_y + s as i32 * 16, pois.clone()));
+            }
         }
-        lists.by_section.insert(
-            key,
-            Cached {
-                section: Arc::downgrade(section),
-                pois: pois.clone(),
-            },
-        );
-        pois
     }
 }
 
@@ -143,6 +221,48 @@ mod tests {
         assert!(index.get((0, 0, 0), &chunk.sections[0], kind).is_empty());
     }
 
+    /// A column gives each section's list as `get` does, the sections
+    /// without any left out, and keeps them until a section changes.
+    #[test]
+    fn a_column_gives_what_each_section_gives() {
+        let index = PoiSections::<u8>::default();
+        let kind = |v: u32| (v == 5).then_some(5);
+        let mut chunk = Arc::new(Chunk::empty(2, -3));
+        for (x, y, z) in [(1, MIN_Y + 2, 3), (4, MIN_Y + 40, 0), (0, MIN_Y + 41, 15)] {
+            Arc::make_mut(&mut chunk).set(x, y, z, 5);
+        }
+        let mut out = Vec::new();
+        index.column(&chunk, MIN_Y, (MIN_Y, MIN_Y + 63), kind, &mut out);
+        let bases: Vec<i32> = out.iter().map(|(base, _)| *base).collect();
+        assert_eq!(bases, [MIN_Y, MIN_Y + 32]);
+        for (base, pois) in &out {
+            let s = (base - MIN_Y) / 16;
+            let one = index.get((2, s, -3), &chunk.sections[s as usize], kind);
+            assert!(Arc::ptr_eq(pois, &one));
+        }
+        // Kept: no rescan.
+        index.column(
+            &chunk,
+            MIN_Y,
+            (MIN_Y, MIN_Y + 63),
+            |_| panic!("rescanned"),
+            &mut out,
+        );
+        assert_eq!(out.len(), 2);
+        // An edit lists that section again.
+        Arc::make_mut(&mut chunk).set(4, MIN_Y + 40, 0, 1);
+        let mut rescans = 0;
+        let counted = |v: u32| {
+            rescans += 1;
+            kind(v)
+        };
+        index.column(&chunk, MIN_Y, (MIN_Y + 20, MIN_Y + 63), counted, &mut out);
+        assert!(rescans > 0);
+        assert_eq!(out.len(), 1);
+        assert_eq!(out[0].0, MIN_Y + 32);
+        assert_eq!(&*out[0].1, &[(9 << 8 | 15 << 4, 5)]);
+    }
+
     #[test]
     fn unloaded_sections_are_dropped() {
         let index = PoiSections::<u8>::default();
@@ -153,7 +273,7 @@ mod tests {
             index.get((1, i, 0), &Arc::new(Section::default()), kind);
         }
         let lists = index.sections.lock().unwrap();
-        assert!(lists.by_section.len() <= 2);
-        assert!(lists.by_section.contains_key(&(0, 0, 0)));
+        assert!(lists.len <= 2);
+        assert!(lists.by_column[&(0, 0)][0].is_some());
     }
 }

crates/horizon-client/examples/probe/mob_ai_probe.rs

@@ -1208,6 +1208,7 @@ pub fn main() -> horizon_client::Result<()> {
                     dragon: None,
                     leash_boats: &[],
                     who_lookup: None,
+                    foreign: None,
                     others_grid: $grid,
                     others_index: None,
                     others_villagers: None,

crates/horizon-client/src/client_lib.rs

@@ -8,6 +8,7 @@ pub mod animate_tick;
 pub mod arrows;
 pub mod bed_geometry;
 pub mod block_contact;
+pub mod parallel;
 pub mod powder_snow;
 pub use horizon_client_render::{
     assets, atlas_sources, audio, auth, autotune, ban_notice, bidi, biome, block_edit,

crates/horizon-client/src/entities.rs

@@ -84,7 +84,9 @@ fn slime_at(world: &World, facts: &BlockFacts, position: BlockPos) -> bool {
 mod fluids;
 mod item_fall;
 mod item_no_physics;
+mod obstacles;
 pub use fluids::{fall_reset_hit, fluid_flow, update_entity_fluids, water_clip};
+pub use obstacles::ObstacleGrid;
 
 #[cfg(test)]
 mod drop_motion_tests {
@@ -1111,110 +1113,6 @@ fn lava_cauldron_contacts(
     }
     contacts
 }
-/// The boxes that stop a dropped item (boats, minecarts, shulkers, a held-still happy ghast),
-/// binned by 8-block cell, so an item looks only at the ones near it: a farm with many drops and
-/// many vehicles checks a handful of boxes per item, not every box in the level.
-#[derive(Default)]
-pub struct ObstacleGrid {
-    cells: crate::fast_hash::FxHashMap<(i32, i32, i32), Vec<u32>>,
-    boxes: Vec<Aabb>,
-    /// Each box's entity's `getRootVehicle` when it rides something; empty
-    /// when none was given.
-    roots: Vec<Option<u32>>,
-    found: Vec<u32>,
-}
-impl ObstacleGrid {
-    const CELL: f64 = 8.0;
-    pub fn new(boxes: Vec<Aabb>) -> Self {
-        Self::with_roots(boxes, Vec::new())
-    }
-    /// [`Self::new`] with the vehicle at the bottom of what each box's entity
-    /// rides (`roots[i]` for `boxes[i]`), so [`Self::near_except`] can leave a
-    /// mover's own riders out.
-    pub fn with_roots(boxes: Vec<Aabb>, roots: Vec<Option<u32>>) -> Self {
-        let mut cells: crate::fast_hash::FxHashMap<(i32, i32, i32), Vec<u32>> = Default::default();
-        for (i, b) in boxes.iter().enumerate() {
-            let (lo, hi) = Self::span(*b);
-            for x in lo.0..=hi.0 {
-                for y in lo.1..=hi.1 {
-                    for z in lo.2..=hi.2 {
-                        cells.entry((x, y, z)).or_default().push(i as u32);
-                    }
-                }
-            }
-        }
-        Self {
-            cells,
-            boxes,
-            roots,
-            found: Vec::new(),
-        }
-    }
-    fn span(b: Aabb) -> ((i32, i32, i32), (i32, i32, i32)) {
-        let cell = |v: Vec3| {
-            (
-                (v.x / Self::CELL).floor() as i32,
-                (v.y / Self::CELL).floor() as i32,
-                (v.z / Self::CELL).floor() as i32,
-            )
-        };
-        (cell(b.min_position()), cell(b.max_position()))
-    }
-    pub fn is_empty(&self) -> bool {
-        self.boxes.is_empty()
-    }
-    /// The boxes meeting `area` (`getEntityCollisions`'s query box), in the order they were given.
-    pub fn near(&mut self, area: Aabb, out: &mut Vec<Aabb>) {
-        self.near_except(area, None, out);
-    }
-    /// [`Self::near`] without the boxes of what rides `vehicle` (given with
-    /// [`Self::with_roots`]): `Entity.canCollideWith` leaves out
-    /// `isPassengerOfSameVehicle`, so a stand's riders do not stop it.
-    pub fn near_except(&mut self, area: Aabb, vehicle: Option<u32>, out: &mut Vec<Aabb>) {
-        out.clear();
-        if self.boxes.is_empty() {
-            return;
-        }
-        let kept = |i: usize| vehicle.is_none() || self.roots.get(i).copied().flatten() != vehicle;
-        let (lo, hi) = Self::span(area);
-        let side = |lo: i32, hi: i32| (i64::from(hi) - i64::from(lo) + 1).max(0);
-        let cells = side(lo.0, hi.0)
-            .saturating_mul(side(lo.1, hi.1))
-            .saturating_mul(side(lo.2, hi.2));
-        // An area over more cells than there are boxes (a stand loaded with a
-        // huge Motion asks for one millions of blocks long): test each box
-        // once rather than walk the cells.
-        if cells > self.boxes.len() as i64 {
-            out.extend(
-                self.boxes
-                    .iter()
-                    .enumerate()
-                    .filter(|&(i, b)| kept(i) && b.intersects(area))
-                    .map(|(_, b)| *b),
-            );
-            return;
-        }
-        self.found.clear();
-        for x in lo.0..=hi.0 {
-            for y in lo.1..=hi.1 {
-                for z in lo.2..=hi.2 {
-                    if let Some(list) = self.cells.get(&(x, y, z)) {
-                        self.found.extend(list);
-                    }
-                }
-            }
-        }
-        self.found.sort_unstable();
-        self.found.dedup();
-        for &i in &self.found {
-            let b = self.boxes[i as usize];
-            if kept(i as usize) && b.intersects(area) {
-                out.push(b);
-            }
-        }
-    }
-}
-
 /// Dropped items by block cell, for `ItemEntity.mergeWithNeighbours`'s
 /// search around an item. Holds indexes into the tick's item list.
 #[derive(Default)]

crates/horizon-client/src/entities/obstacles.rs

@@ -0,0 +1,122 @@
+//! The boxes that stop a dropped item, by cell (`ObstacleGrid`).
+use crate::fast_hash::FxHashMap;
+use horizon_core::geometry::{Aabb, Vec3};
+
+/// The boxes that stop a dropped item (boats, minecarts, shulkers, a held-still happy ghast),
+/// binned by 8-block cell, so an item looks only at the ones near it: a farm with many drops and
+/// many vehicles checks a handful of boxes per item, not every box in the level.
+#[derive(Default)]
+pub struct ObstacleGrid {
+    cells: FxHashMap<(i32, i32, i32), Vec<u32>>,
+    boxes: Vec<Aabb>,
+    /// Each box's entity's `getRootVehicle` when it rides something; empty
+    /// when none was given.
+    roots: Vec<Option<u32>>,
+    found: Vec<u32>,
+}
+impl ObstacleGrid {
+    const CELL: f64 = 8.0;
+    pub fn new(boxes: Vec<Aabb>) -> Self {
+        Self::with_roots(boxes, Vec::new())
+    }
+    /// [`Self::new`] with the vehicle at the bottom of what each box's entity
+    /// rides (`roots[i]` for `boxes[i]`), so [`Self::near_except`] can leave a
+    /// mover's own riders out.
+    pub fn with_roots(boxes: Vec<Aabb>, roots: Vec<Option<u32>>) -> Self {
+        let mut cells: FxHashMap<(i32, i32, i32), Vec<u32>> = Default::default();
+        for (i, b) in boxes.iter().enumerate() {
+            let (lo, hi) = Self::span(*b);
+            for x in lo.0..=hi.0 {
+                for y in lo.1..=hi.1 {
+                    for z in lo.2..=hi.2 {
+                        cells.entry((x, y, z)).or_default().push(i as u32);
+                    }
+                }
+            }
+        }
+        Self {
+            cells,
+            boxes,
+            roots,
+            found: Vec::new(),
+        }
+    }
+    fn span(b: Aabb) -> ((i32, i32, i32), (i32, i32, i32)) {
+        let cell = |v: Vec3| {
+            (
+                (v.x / Self::CELL).floor() as i32,
+                (v.y / Self::CELL).floor() as i32,
+                (v.z / Self::CELL).floor() as i32,
+            )
+        };
+        (cell(b.min_position()), cell(b.max_position()))
+    }
+    pub fn is_empty(&self) -> bool {
+        self.boxes.is_empty()
+    }
+    /// The boxes meeting `area` (`getEntityCollisions`'s query box), in the order they were given.
+    pub fn near(&mut self, area: Aabb, out: &mut Vec<Aabb>) {
+        self.near_except(area, None, out);
+    }
+    /// [`Self::near`] without the boxes of what rides `vehicle` (given with
+    /// [`Self::with_roots`]): `Entity.canCollideWith` leaves out
+    /// `isPassengerOfSameVehicle`, so a stand's riders do not stop it.
+    pub fn near_except(&mut self, area: Aabb, vehicle: Option<u32>, out: &mut Vec<Aabb>) {
+        let mut found = std::mem::take(&mut self.found);
+        self.near_with(area, vehicle, &mut found, out);
+        self.found = found;
+    }
+    /// [`Self::near`] with the caller's scratch, so several threads can ask one grid.
+    pub fn near_shared(&self, area: Aabb, found: &mut Vec<u32>, out: &mut Vec<Aabb>) {
+        self.near_with(area, None, found, out);
+    }
+    fn near_with(
+        &self,
+        area: Aabb,
+        vehicle: Option<u32>,
+        found: &mut Vec<u32>,
+        out: &mut Vec<Aabb>,
+    ) {
+        out.clear();
+        if self.boxes.is_empty() {
+            return;
+        }
+        let kept = |i: usize| vehicle.is_none() || self.roots.get(i).copied().flatten() != vehicle;
+        let (lo, hi) = Self::span(area);
+        let side = |lo: i32, hi: i32| (i64::from(hi) - i64::from(lo) + 1).max(0);
+        let cells = side(lo.0, hi.0)
+            .saturating_mul(side(lo.1, hi.1))
+            .saturating_mul(side(lo.2, hi.2));
+        // An area over more cells than there are boxes (a stand loaded with a
+        // huge Motion asks for one millions of blocks long): test each box
+        // once rather than walk the cells.
+        if cells > self.boxes.len() as i64 {
+            out.extend(
+                self.boxes
+                    .iter()
+                    .enumerate()
+                    .filter(|&(i, b)| kept(i) && b.intersects(area))
+                    .map(|(_, b)| *b),
+            );
+            return;
+        }
+        found.clear();
+        for x in lo.0..=hi.0 {
+            for y in lo.1..=hi.1 {
+                for z in lo.2..=hi.2 {
+                    if let Some(list) = self.cells.get(&(x, y, z)) {
+                        found.extend(list);
+                    }
+                }
+            }
+        }
+        found.sort_unstable();
+        found.dedup();
+        for &i in found.iter() {
+            let b = self.boxes[i as usize];
+            if kept(i as usize) && b.intersects(area) {
+                out.push(b);
+            }
+        }
+    }
+}

crates/horizon-client/src/game.rs

@@ -1499,7 +1499,7 @@ impl LevelAccess for GameLevel<'_> {
             };
         }
         if let Some(signal) = books::lectern_signal(self.world, pos) {
-            return signal;
+            return signal * i32::from(books::has_book_state(self.registry, state));
         }
         if let Some(signal) =
             machines::shelf_signal(self.world, self.registry, state, pos, direction)

crates/horizon-client/src/game/block_ticks.rs

@@ -239,6 +239,19 @@ impl Game<'_> {
             self.tick_scheduled()?;
         }
         self.with_level(|b, level, ticks| {
+            // The masks the random ticks will read, built on helper threads first while the
+            // parallel tick is on (`HORIZON_PARALLEL_TICK`): the draws and ticks don't depend on them.
+            if !debug && horizon_client::parallel::lanes() > 1 {
+                horizon_client::spans::time("random_tick_masks", || {
+                    let chunks = &level.world.chunks;
+                    level.random_tick_sections.prepare_masks(
+                        &scratch.ticking,
+                        |key| chunks.get(&key).cloned(),
+                        level.facts,
+                        &|id| b.random_tick_candidate(BlockStateId(id)),
+                    );
+                });
+            }
             horizon_client::spans::time("random_ticks", || {
                 for (x, z) in scratch.ticking.drain(..).filter(|_| !debug) {
                     // The last answer, asked from Y `from` at a world revision:

crates/horizon-client/src/game/boats.rs

@@ -682,19 +682,14 @@ impl Game<'_> {
     /// of it (`ClipContext.Block.FALLDAMAGE_RESETTING`, `ClipContext.Fluid.WATER`)
     /// and a hit resets the fall, before the move's own drop counts.
     pub(super) fn vehicle_fall_reset(&self, fall: f64, from: Vec3, moved: Vec3) -> f64 {
-        let length = moved.length_squared();
-        if fall == 0.0 || length < 1.0 {
-            return fall;
-        }
-        let reset = &self.res.motion().fall_reset;
-        let hit = horizon_client::entities::fall_reset_hit(
+        fall_reset(
             &self.world,
-            &self.res.motion().fluids,
-            |p| reset.contains(&*self.res.facts().get(self.world.block(p.x, p.y, p.z)).name),
+            self.res.motion(),
+            self.res.facts(),
+            fall,
             from,
             moved,
-        );
-        if hit { 0.0 } else { fall }
+        )
     }
     /// `AbstractBoat.checkFallDamage` after a move that went `moved_y` (with
     /// `b` already where it landed): on the ground the fall resets; a move
@@ -2071,3 +2066,26 @@ mod tests {
     }
 }
 horizon_client::impl_joined!(Boat);
+
+/// [`Game::vehicle_fall_reset`] from the tables it reads.
+pub(super) fn fall_reset(
+    world: &horizon_client::world::World,
+    motion: &horizon_client::motion::MotionResources,
+    facts: &horizon_client::blocks::BlockFacts,
+    fall: f64,
+    from: Vec3,
+    moved: Vec3,
+) -> f64 {
+    if fall == 0.0 || moved.length_squared() < 1.0 {
+        return fall;
+    }
+    let reset = &motion.fall_reset;
+    let hit = horizon_client::entities::fall_reset_hit(
+        world,
+        &motion.fluids,
+        |p| reset.contains(&*facts.get(world.block(p.x, p.y, p.z)).name),
+        from,
+        moved,
+    );
+    if hit { 0.0 } else { fall }
+}

crates/horizon-client/src/game/books.rs

@@ -1608,6 +1608,16 @@ impl Game<'_> {
         }
     }
 }
+/// `LecternBlock.getAnalogOutputSignal`'s `HAS_BOOK` check.
+pub(super) fn has_book_state(
+    registry: &'static horizon_world::registry::BlockRegistry,
+    state: horizon_world::registry::BlockStateId,
+) -> bool {
+    registry
+        .state(state)
+        .is_some_and(|s| s.properties.get("has_book").is_some_and(|v| v == "true"))
+}
+
 /// `LecternBlockEntity.getRedstoneSignal`.
 pub(super) fn lectern_signal(world: &horizon_client::world::World, pos: BlockPos) -> Option<i32> {
     match world.block_entities.get(&(pos.x, pos.y, pos.z)) {
@@ -1721,6 +1731,7 @@ mod lectern_signal_tests {
         items::ItemStack,
     };
     use horizon_core::position::BlockPos;
+    use horizon_world::level::LevelAccess;
     use horizon_world::registry::BlockStateId;
 
     fn signal(game: &mut super::Game<'_>, book: Option<ItemStack>, page: i32) -> i32 {
@@ -1738,6 +1749,55 @@ mod lectern_signal_tests {
         lectern_signal(&game.world, BlockPos::new(key.0, key.1, key.2)).unwrap()
     }
 
+    /// `LecternBlock.getAnalogOutputSignal` reads the block entity only while
+    /// the state's `HAS_BOOK` is true.
+    #[test]
+    #[ignore = "needs prepared resources in HORIZON_TEST_RESOURCES"]
+    fn a_lectern_without_the_has_book_state_feeds_no_comparator() {
+        let mut game = server_game();
+        let key = (8, 64, 8);
+        let state = game
+            .pack
+            .assets
+            .registry
+            .default_state("minecraft:lectern")
+            .unwrap();
+        game.world.set(key.0, key.1, key.2, state).unwrap();
+        game.world.block_entities.insert(
+            key,
+            BlockEntity::Lectern(Lectern {
+                book: Some(ItemStack::new("minecraft:writable_book", 1)),
+                page: 0,
+            }),
+        );
+        let signal = |game: &mut super::Game<'_>, state| {
+            game.with_level(|_, level, _| {
+                level.block_entity_analog_signal(
+                    BlockPos::new(key.0, key.1, key.2),
+                    state,
+                    horizon_core::geometry::Direction::North,
+                )
+            })
+        };
+        assert_eq!(
+            signal(&mut game, state),
+            0,
+            "the block state says the lectern holds no book"
+        );
+        let registry = game.pack.assets.registry;
+        let mut with_book = registry.state(state).unwrap().clone();
+        with_book
+            .properties
+            .insert("has_book".into(), "true".into());
+        let with_book = registry.id(&with_book).unwrap();
+        game.world.set(key.0, key.1, key.2, with_book).unwrap();
+        assert_eq!(
+            signal(&mut game, with_book),
+            15,
+            "with HAS_BOOK the book's page signal"
+        );
+    }
+
     /// `LecternBlockEntity.getRedstoneSignal`: `floor(page / (pageCount - 1)
     /// * 14) + (hasBook() ? 1 : 0)`. `hasBook()` asks for a book-content
     /// component, so a stack that only sits in the lectern (a non-book item a

crates/horizon-client/src/game/command_harness.rs

@@ -380,7 +380,7 @@ impl Game<'_> {
                         let (mut meta, mut defaults) = (Vec::new(), Vec::new());
                         super::lan::mob_meta(
                             m,
-                            hub,
+                            hub.sync_data(),
                             None,
                             g.world.game_time,
                             &mut meta,

crates/horizon-client/src/game/creatures.rs

@@ -3,6 +3,8 @@
 //! breeding, conversions, drops), projectiles, player interactions
 //! (feeding, shearing, milking, spawn eggs, throwing) and rendering.
 mod mob_geometry;
+mod mob_group;
+mod prep;
 mod projectile_world;
 mod pufferfish_player;
 mod spawn_eggs;
@@ -1048,18 +1050,11 @@ impl Game<'_> {
             teams: std::sync::Arc::new(self.mob_teams()),
             items: Some(self.items()),
             lid_openers: &self.lid_openers,
-            mod_effects: if self.remote.is_none() {
-                Some(self)
-            } else {
-                None
-            },
+            mod_effects: self.remote.is_none().then_some(self as _),
             // A warden's away suspects are UUIDs: on the server side, where the
             // entity UUIDs live, they resolve back to the mobs that are here.
-            who_lookup: if self.remote.is_none() {
-                Some(self)
-            } else {
-                None
-            },
+            who_lookup: self.remote.is_none().then_some(self as _),
+            foreign: None,
             border: Some(self.border.geometry()),
             // The first player in the level: a server has no player of its own.
             difficulty: self.local_difficulty(
@@ -1187,50 +1182,32 @@ impl Game<'_> {
         self.init_sulfur_attribute_maps();
         let uuid = self.level.player_uuid;
         let level_id = self.dim.id();
-        // LivingEntity.isPushable reads onClimbable live; a mob's cached one
-        // is only refreshed by its own push step, so one that has not ticked
-        // yet (the first tick after a load, a fresh spawn) or has moved since
-        // would be pushed out of a ladder or scaffolding. Refreshed here, from
-        // where each mob stands now, before the snapshot others push it by.
-        let climbing: Vec<bool> = {
-            let w = self.mob_world_base(&[]);
-            self.mobs.iter().map(|m| m.on_climbable(&w)).collect()
-        };
-        for (m, climbing) in self.mobs.iter_mut().zip(climbing) {
-            m.on_climbable = climbing;
-        }
+        // Each mob's climbing, refreshed, and its snapshot entry (creatures/prep.rs).
         let mut snapshot = std::mem::take(&mut self.creatures.seen_buf);
-        snapshot.extend(self.mobs.iter().map(|m| seen(m, uuid)));
+        self.prepare_mobs(uuid, &mut snapshot);
         let mut events = std::mem::take(&mut self.creatures.event_buf);
         events.clear();
         // SulfurCube.primeWhenOnPoweredPosition runs before super.tick adopts
         // the body: the cubes whose adopted body explodes, as this tick starts.
         let primable = self.sulfur_cubes_adopted_explosive();
-        let mut mobs = std::mem::take(&mut self.mobs);
+        let mobs = std::mem::take(&mut self.mobs);
         // Each mob's target as the mods last heard it, by its place in `mobs`
         // (`native_living_change_target`).
         let mut targets = std::mem::take(&mut self.creatures.target_buf);
         targets.clear();
         targets.extend(mobs.iter().map(|m| m.ai.target));
-        let mut dead = Vec::new();
-        let mut corpses = Vec::new();
-        let mut died = std::mem::take(&mut self.died);
+        let died = std::mem::take(&mut self.died);
         let mut ticked = std::mem::take(&mut self.creatures.ticked);
         ticked.clear();
         let ticking = self.entity_ticking_area();
-        // With guests on, each mob sees its nearest player (game/lan_mobs.rs).
-        let mut guests = self.lan_guest_views();
-        // What stands in a guest's slot while a mob holds that guest's view for
-        // its tick (see `lent` below).
-        let mut spare = guests.first().map(|g| g.1.clone());
         let players = self.level_players();
         let nearest = super::natural_spawning::NearestPlayers::new(&players);
-        let mut bindings = match self.lan.as_mut() {
+        // With guests on, each mob sees its nearest player (game/lan_mobs.rs).
+        let guests = self.lan_guest_views();
+        let bindings = match self.lan.as_mut() {
             Some(host) if !guests.is_empty() => std::mem::take(&mut host.mob_players),
             _ => Default::default(),
         };
-        // The guests' entity ids by connection, for the shulker bullets fired
-        // at the guest a mob sees.
         let guest_ids: Vec<(u32, u32)> = match self.lan.as_ref() {
             Some(host) if !guests.is_empty() => host
                 .hub
@@ -1240,23 +1217,9 @@ impl Game<'_> {
                 .collect(),
             _ => Vec::new(),
         };
-        let mut guest_events = Vec::new();
         // The block writes since the last tick: mobs re-path around them.
         let path_writes = std::mem::take(self.world.path_writes.get_or_insert_with(Vec::new));
-        // The mobs whose bodies go this tick: (id, feet, width, height, split).
-        // 🤖 Agent: the last field is the grown slime, magma cube or sulfur
-        // cube the body leaves behind (`AbstractCubeMob.remove`), captured as
-        // the body goes because it has left the list by the time it splits.
-        let mut gone = Vec::new();
-        let level_random;
-        {
-            // The level random, lent to the mobs (brain behaviours draw from
-            // it) and taken back after their ticks.
-            let mut scratch = self.world.random.clone();
-            let mut pushes = horizon_client::mobs::PendingPushes::default();
-            let mut keep = Vec::with_capacity(mobs.len());
-            let mut grid =
-                horizon_client::mobs::seen_grid::SeenGrid::new(snapshot.iter().map(|s| s.bounds));
+        let group = {
             // All but the snapshot is the same for every mob of the tick, so
             // build it once: the player view and the teams scan every UUID.
             self.leash_boats = self
@@ -1266,87 +1229,52 @@ impl Game<'_> {
                 .collect();
             let mut base = self.mob_world_base(&[]);
             let path_changes = horizon_client::mobs::collision_changes(&base, &path_writes);
-            // Each mob keeps its snapshot slot all tick; the first slot of an
-            // id answers, as a scan would.
-            let mut index = horizon_client::fast_hash::FxHashMap::with_capacity_and_hasher(
-                snapshot.len(),
-                Default::default(),
-            );
-            let mut villager_slots = Vec::new();
-            let mut solid_slots = Vec::new();
-            for (i, s) in snapshot.iter().enumerate() {
-                index.entry(s.id).or_insert(i as u32);
-                if s.species == horizon_client::mobs::Species::Villager {
-                    villager_slots.push(i as u32);
-                }
-                if s.may_block_mobs() {
-                    solid_slots.push(i as u32);
-                }
-            }
             laps.lap("creature_prepare");
             let single_player = horizon_client::mobs::single_player_pos(
                 players.iter().map(|p| (p.position, p.spectator)),
             );
+            let nearest = prep::MobsNearest::new(&nearest, &mobs, single_player.is_none());
             let mut despawn_w = base.despawn_view();
             let mut cached_difficulty: Option<((i32, i32), spawning::LocalDifficulty)> = None;
-            for i in 0..mobs.len() {
-                let m = &mut mobs[i];
-                let (bx, bz) = (m.position.x.floor() as i32, m.position.z.floor() as i32);
-                let seen_guest = if guests.is_empty() {
-                    None
-                } else {
-                    let bound = bindings.get(&m.id).copied();
-                    let conn = super::lan_mobs::player_for(m, base.player.as_ref(), &guests, bound);
-                    match conn {
-                        Some(conn) => {
-                            bindings.insert(m.id, conn);
-                        }
-                        None => {
-                            bindings.remove(&m.id);
-                        }
-                    }
-                    conn
-                };
+            let shared = mob_group::Shared {
+                uuid,
+                level_id,
+                path_changes: &path_changes,
+                guest_ids: &guest_ids,
+            };
+            let start = mob_group::Start {
+                events,
+                // The level random, lent to the mobs and taken back after their ticks.
+                random: self.world.random.clone(),
+                guests,
+                player: base.player.take(),
+                bindings,
+                died,
+                ticked,
+                targets,
+            };
+            let mut group = mob_group::Group::new(mobs.into(), snapshot, start);
+            for i in 0..group.mobs.len() {
+                let position = group.mobs[i].position;
+                let (bx, bz) = (position.x.floor() as i32, position.z.floor() as i32);
+                let seen_guest = group.bind_guest(i);
                 if !base.world.loaded(bx, bz) {
-                    keep.push(!m.hostile() || m.persistent);
+                    group.unloaded(i);
                     continue;
                 }
-                let nearest_player = single_player.unwrap_or_else(|| {
-                    nearest
-                        .nearest(m.position.x, m.position.y, m.position.z)
-                        .map(|p| p.position)
-                });
+                let nearest_player = single_player.unwrap_or_else(|| nearest.at(i, position));
                 // ServerLevel.tick: every loaded entity checks despawning, but
                 // only those in entity-ticking chunks (within the simulation
-                // distance) tick. The despawn check reads the nearest player
-                // and the difficulty, not the player the mob sees.
+                // distance) tick.
                 if !ticking(bx, bz) {
                     despawn_w.nearest_player = nearest_player;
-                    let kept = horizon_client::mobs::ai::despawn_check(m, &despawn_w);
-                    if !kept {
-                        died.remove(&m.id);
-                        snapshot[i].alive = false;
-                    }
-                    keep.push(kept);
+                    group.rest(i, &despawn_w);
                     continue;
                 }
-                let controlling_player_movement = horizon_client::mobs::equine::controlled(m)
-                    .then(|| {
-                        base.player
-                            .as_ref()
-                            .filter(|p| p.riding_vehicle == Some(m.id))
-                            .map(|p| p.velocity)
-                            .or_else(|| {
-                                guests
-                                    .iter()
-                                    .find(|(_, p)| p.riding_vehicle == Some(m.id))
-                                    .map(|(_, p)| p.velocity)
-                            })
-                    })
-                    .flatten();
+                let controlling = group.controlling(i);
                 // A mod canceled its tick: it stays as it is.
-                if round.canceled(m.id) {
-                    keep.push(true);
+                if round.canceled(group.mobs[i].id) {
+                    group.keep.push(true);
                     continue;
                 }
                 // getCurrentDifficultyAt(blockPosition()): the mob's own chunk,
@@ -1356,175 +1284,34 @@ impl Game<'_> {
                     Some((key, difficulty)) if key == chunk_key => difficulty,
                     _ => {
                         let difficulty =
-                            self.local_difficulty([m.position.x, m.position.y, m.position.z]);
+                            self.local_difficulty([position.x, position.y, position.z]);
                         cached_difficulty = Some((chunk_key, difficulty));
                         difficulty
                     }
                 };
-                horizon_client::mobs::apply_mob_pushes(m, &mut pushes);
-                horizon_client::mobs::blocks_changed(m, &path_changes);
-                let from = events.len();
-                // The guest's view is lent to the mob for its tick, not copied:
-                // the tick only reads it (`&MobWorld`), and it goes back to the
-                // guest's slot straight after.
-                let lent = seen_guest.and_then(|conn| guests.iter().position(|g| g.0 == conn));
-                let active_player = match (seen_guest, lent) {
-                    (None, _) => base.player.take(),
-                    (Some(_), Some(k)) => spare
-                        .take()
-                        .map(|stand_in| std::mem::replace(&mut guests[k].1, stand_in)),
-                    (Some(_), None) => None,
-                };
-                let mut w = MobWorld {
-                    others: &snapshot,
-                    others_grid: Some(&grid),
-                    others_index: Some(&index),
-                    others_villagers: Some(&villager_slots),
-                    others_solid: Some(&solid_slots),
-                    difficulty,
-                    nearest_player,
-                    player: active_player,
-                    teams: std::sync::Arc::clone(&base.teams),
-                    ..base
-                };
-                w.controlling_player_movement = controlling_player_movement;
-                ticked.insert(mobs[i].id);
-                // `horizon-report`'s cost by mob type (tick_profile.rs).
-                let sampled = super::tick_profile::start().map(|t| (t, mobs[i].kind.clone()));
-                let kept =
-                    horizon_client::mobs::riders::tick(&mut mobs, i, &w, &mut scratch, &mut events);
-                if let Some((started, kind)) = sampled {
-                    super::tick_profile::add(level_id, &kind, started);
-                }
-                let m = &mut mobs[i];
-                if seen_guest.is_none() {
-                    base.player = w.player;
-                } else if let Some(k) = lent {
-                    spare = w.player.map(|v| std::mem::replace(&mut guests[k].1, v));
-                }
-                // Its tick changed its target: the mods may keep the one it had
-                // (`native_living_change_target`), and then what the tick did to
-                // the new one is not dealt.
-                if let Some(&before) = targets.get(i)
-                    && m.ai.target != before
-                {
-                    let named = target_named(
-                        m.ai.target,
-                        seen_guest,
-                        &guests,
-                        &guest_ids,
-                        base.player.as_ref(),
-                    );
-                    if let Some((player, mob)) = named {
-                        let kind = super::native_hooks::mob_kind(m);
-                        let mut cancel = false;
-                        Self::native_living_change_target(
-                            level_id,
-                            m.id,
-                            &kind,
-                            player,
-                            mob,
-                            &mut cancel,
-                        );
-                        if cancel {
-                            if let Some(new) = m.ai.target {
-                                let mut k = from;
-                                while k < events.len() {
-                                    if hits_target(&events[k], new) {
-                                        events.remove(k);
-                                    } else {
-                                        k += 1;
-                                    }
-                                }
-                            }
-                            m.ai.target = before;
-                        }
-                    }
-                    targets[i] = m.ai.target;
-                }
-                // A killed mob's body is removed: its effects that act on
-                // death run (`LivingEntity.remove` with KILLED).
-                if !kept && !m.alive() && !m.removed && !m.extra.effects.is_empty() {
-                    corpses.push(m.clone());
-                }
-                horizon_client::mobs::take_mob_pushes(&mut events, from, &mut pushes);
-                // LivingEntity.tickDeath: after twenty ticks the body is removed.
-                if !kept && !m.removed && !m.alive() && m.death_time >= 20 {
-                    let (width, height) = (f64::from(m.width()), f64::from(m.height()));
-                    // 🤖 Agent: `AbstractCubeMob.remove` splits a grown cube into
-                    // its smaller halves when it is removed, twenty ticks after
-                    // the death. The parent leaves the list before the split
-                    // runs, so its size, position, kind and persistence are
-                    // captured here, as they are at the removal.
-                    let split = if matches!(
-                        m.species(),
-                        Species::Slime | Species::MagmaCube | Species::SulfurCube
-                    ) && m.extra.size > 1
-                    {
-                        Some(m.clone())
-                    } else {
-                        None
-                    };
-                    gone.push((m.id, m.position, width, height, split));
-                }
-                if let Some(conn) = seen_guest {
-                    // ShulkerBullet.finalTarget is the guest's entity: the
-                    // bullet homes on it while the projectiles tick, where
-                    // `Who::Player` is the host's own player.
-                    if let Some(&(_, id)) = guest_ids.iter().find(|g| g.0 == conn) {
-                        for e in &mut events[from..] {
-                            if let MobEvent::Projectile(p) = e
-                                && let ProjectileKind::ShulkerBullet { target, .. } = &mut p.kind
-                                && *target == Some(Who::Player)
-                            {
-                                *target = Some(Who::Mob(id));
-                            }
-                        }
-                    }
-                    let mut k = from;
-                    while k < events.len() {
-                        if super::lan_mobs::acts_on_player(&events[k]) {
-                            guest_events.push((conn, m.id, events.remove(k)));
-                        } else {
-                            k += 1;
-                        }
-                    }
-                }
-                // Killed during its own tick (drowning, burning, falling...).
-                if !m.alive() && !m.removed && died.insert(m.id) {
-                    dead.push(m.clone());
-                }
-                if !kept {
-                    died.remove(&m.id);
-                }
-                keep.push(kept);
-                // ServerLevel ticks entities one after another: the ones
-                // after this mob see it as it is now.
-                snapshot[i] = seen(m, uuid);
-                snapshot[i].alive &= kept;
-                grid.update(i, snapshot[i].bounds);
-                horizon_client::mobs::apply_brain_events(&mut mobs, &mut events, from);
+                let player = (seen_guest, nearest_player, controlling);
+                group.tick(i, player, difficulty, &base, &shared);
             }
             // Targets changed after the mob's own tick (another mob's alert):
             // heard now, and a cancel puts the old one back before its next tick.
-            for (i, m) in mobs.iter_mut().enumerate() {
-                let Some(&before) = targets.get(i) else {
+            for (i, m) in group.mobs.iter_mut().enumerate() {
+                let Some(&before) = group.targets.get(i) else {
                     break;
                 };
                 if m.ai.target == before {
                     continue;
                 }
-                let seen_guest = if guests.is_empty() {
+                let seen_guest = if group.guests.is_empty() {
                     None
                 } else {
-                    bindings.get(&m.id).copied()
+                    group.bindings.get(&m.id).copied()
                 };
                 let named = target_named(
                     m.ai.target,
                     seen_guest,
-                    &guests,
+                    &group.guests,
                     &guest_ids,
-                    base.player.as_ref(),
+                    group.player.as_ref(),
                 );
                 if let Some((player, mob)) = named {
                     let kind = super::native_hooks::mob_kind(m);
@@ -1543,13 +1330,29 @@ impl Game<'_> {
                 }
             }
             laps.lap("creature_loop");
-            let mut flags = keep.into_iter();
-            mobs.retain(|_| flags.next().unwrap_or(true));
-            for m in &mut mobs {
-                horizon_client::mobs::apply_mob_pushes(m, &mut pushes);
+            let mut flags = std::mem::take(&mut group.keep).into_iter();
+            group.mobs.retain(|_| flags.next().unwrap_or(true));
+            for m in &mut group.mobs {
+                horizon_client::mobs::apply_mob_pushes(m, &mut group.pushes);
             }
-            level_random = scratch;
-        }
+            group
+        };
+        let mob_group::Group {
+            mut mobs,
+            mut snapshot,
+            mut events,
+            random: level_random,
+            guests,
+            mut bindings,
+            died,
+            ticked,
+            mut targets,
+            dead,
+            corpses,
+            gone,
+            guest_events,
+            ..
+        } = group;
         self.world.random = level_random;
         if let Some(host) = self.lan.as_mut()
             && !guests.is_empty()
@@ -1565,7 +1368,7 @@ impl Game<'_> {
         // first, as `append` put them; with none, the ticked list goes back
         // as it is instead of being copied into a new one, every mob a tick.
         if self.mobs.is_empty() {
-            self.mobs = mobs;
+            self.mobs = mobs.into();
         } else {
             self.mobs.append(&mut mobs);
         }
@@ -8376,6 +8179,7 @@ mod pearl_tests {
             dragon: None,
             leash_boats: &[],
             who_lookup: None,
+            foreign: None,
             others_grid: None,
             others_index: None,
             others_villagers: None,

crates/horizon-client/src/game/creatures/mob_group.rs

@@ -0,0 +1,338 @@
+//! The mob loop of `tick_creature_entities` over a group of mobs, in their list order, with
+//! everything the loop keeps for them: their snapshot and its indexes, the pushes and events
+//! their ticks give, the level random they draw from, the players they see and what dies. The
+//! serial loop is one group holding every mob; a group of the parallel mob tick (#14457) holds
+//! the mobs no mob outside it can reach this tick.
+use super::*;
+use horizon_client::{
+    fast_hash::{FxHashMap, FxHashSet},
+    mobs::{MobSlot, PendingPushes, seen_grid::SeenGrid, spawning::LocalDifficulty},
+};
+use horizon_core::random::LegacyRandom;
+
+/// What every mob of the loop reads and none writes.
+pub(super) struct Shared<'a> {
+    pub uuid: Option<[i32; 4]>,
+    pub level_id: &'static str,
+    /// The blocks whose collision changed since the last tick: mobs re-path around them.
+    pub path_changes: &'a [[i32; 3]],
+    /// The guests' entity ids by connection, for the shulker bullets fired at the guest a mob
+    /// sees.
+    pub guest_ids: &'a [(u32, u32)],
+}
+
+/// A mob whose body goes this tick: (id, feet, width, height, the grown cube it splits into).
+/// The last field is the grown slime, magma cube or sulfur cube the body leaves behind
+/// (`AbstractCubeMob.remove`), captured as the body goes because it has left the list by the
+/// time it splits.
+pub(super) type Gone = (u32, Vec3, f64, f64, Option<Mob>);
+
+/// A group's mobs and everything the loop keeps for them, in their list order.
+pub(super) struct Group<M: MobSlot> {
+    pub mobs: Vec<M>,
+    /// Each mob's snapshot entry: as it was before the tick until its turn, as its turn left it
+    /// after.
+    pub snapshot: Vec<Seen>,
+    grid: SeenGrid,
+    /// The first slot of each id, as a scan would find it.
+    index: FxHashMap<u32, u32>,
+    villager_slots: Vec<u32>,
+    solid_slots: Vec<u32>,
+    pub pushes: PendingPushes,
+    pub events: Vec<MobEvent>,
+    /// The level random, lent to the mobs (brain behaviours draw from it).
+    pub random: LegacyRandom,
+    pub guests: super::super::lan_mobs::GuestViews,
+    /// What stands in a guest's slot while a mob holds that guest's view for its tick.
+    spare: Option<PlayerView>,
+    /// The host's own player, lent to each mob that sees it for its tick.
+    pub player: Option<PlayerView>,
+    /// The guest each mob sees, by mob id.
+    pub bindings: FxHashMap<u32, u32>,
+    pub died: FxHashSet<u32>,
+    pub ticked: FxHashSet<u32>,
+    /// Each mob's target as the mods last heard it (`native_living_change_target`).
+    pub targets: Vec<Option<Who>>,
+    /// Whether each mob stays in the list, pushed at its turn.
+    pub keep: Vec<bool>,
+    pub dead: Vec<Mob>,
+    pub corpses: Vec<Mob>,
+    pub gone: Vec<Gone>,
+    pub guest_events: Vec<(u32, u32, MobEvent)>,
+}
+
+/// What a group starts from besides its mobs and their snapshot.
+pub(super) struct Start {
+    pub events: Vec<MobEvent>,
+    pub random: LegacyRandom,
+    pub guests: super::super::lan_mobs::GuestViews,
+    pub player: Option<PlayerView>,
+    pub bindings: FxHashMap<u32, u32>,
+    pub died: FxHashSet<u32>,
+    pub ticked: FxHashSet<u32>,
+    pub targets: Vec<Option<Who>>,
+}
+
+impl<M: MobSlot> Group<M> {
+    /// A group of `mobs`, whose snapshot entries are `snapshot`.
+    pub(super) fn new(mobs: Vec<M>, snapshot: Vec<Seen>, start: Start) -> Self {
+        let grid = SeenGrid::new(snapshot.iter().map(|s| s.bounds));
+        let mut index = FxHashMap::with_capacity_and_hasher(snapshot.len(), Default::default());
+        let (mut villager_slots, mut solid_slots) = (Vec::new(), Vec::new());
+        for (i, s) in snapshot.iter().enumerate() {
+            index.entry(s.id).or_insert(i as u32);
+            if s.species == Species::Villager {
+                villager_slots.push(i as u32);
+            }
+            if s.may_block_mobs() {
+                solid_slots.push(i as u32);
+            }
+        }
+        let n = mobs.len();
+        Self {
+            spare: start.guests.first().map(|g| g.1.clone()),
+            mobs,
+            snapshot,
+            grid,
+            index,
+            villager_slots,
+            solid_slots,
+            pushes: PendingPushes::default(),
+            events: start.events,
+            random: start.random,
+            guests: start.guests,
+            player: start.player,
+            bindings: start.bindings,
+            died: start.died,
+            ticked: start.ticked,
+            targets: start.targets,
+            keep: Vec::with_capacity(n),
+            dead: Vec::new(),
+            corpses: Vec::new(),
+            gone: Vec::new(),
+            guest_events: Vec::new(),
+        }
+    }
+
+    /// The guest mob `i` sees now, kept in the bindings: none while no guest is on.
+    pub(super) fn bind_guest(&mut self, i: usize) -> Option<u32> {
+        if self.guests.is_empty() {
+            return None;
+        }
+        let m = self.mobs[i].mob();
+        let bound = self.bindings.get(&m.id).copied();
+        let conn = super::super::lan_mobs::player_for(m, self.player.as_ref(), &self.guests, bound);
+        match conn {
+            Some(conn) => self.bindings.insert(m.id, conn),
+            None => self.bindings.remove(&m.id),
+        };
+        conn
+    }
+
+    /// Mob `i`'s turn when its chunk isn't loaded: it is kept as it is, unless it would despawn.
+    pub(super) fn unloaded(&mut self, i: usize) {
+        let m = self.mobs[i].mob();
+        self.keep.push(!m.hostile() || m.persistent);
+    }
+
+    /// Mob `i`'s turn outside the entity-ticking area: only its despawn check, which reads the
+    /// nearest player and the difficulty, not the player the mob sees.
+    pub(super) fn rest(&mut self, i: usize, despawn_w: &MobWorld) {
+        let m = self.mobs[i].mob_mut();
+        let kept = horizon_client::mobs::ai::despawn_check(m, despawn_w);
+        if !kept {
+            self.died.remove(&m.id);
+            self.snapshot[i].alive = false;
+        }
+        self.keep.push(kept);
+    }
+
+    /// Who controls mob `i`'s moves this tick: the velocity of the player riding it, when it is
+    /// a controlled mount.
+    pub(super) fn controlling(&self, i: usize) -> Option<Vec3> {
+        let m = self.mobs[i].mob();
+        horizon_client::mobs::equine::controlled(m)
+            .then(|| {
+                self.player
+                    .as_ref()
+                    .filter(|p| p.riding_vehicle == Some(m.id))
+                    .map(|p| p.velocity)
+                    .or_else(|| {
+                        self.guests
+                            .iter()
+                            .find(|(_, p)| p.riding_vehicle == Some(m.id))
+                            .map(|(_, p)| p.velocity)
+                    })
+            })
+            .flatten()
+    }
+
+    /// Mob `i`'s own tick, then what the loop does with what it did. `players`: the guest it
+    /// sees (none: the host's player), its nearest player, and the velocity of the player
+    /// controlling it as a mount ([`Self::controlling`]).
+    pub(super) fn tick(
+        &mut self,
+        i: usize,
+        (seen_guest, nearest_player, controlling_player_movement): (
+            Option<u32>,
+            Option<Vec3>,
+            Option<Vec3>,
+        ),
+        difficulty: LocalDifficulty,
+        base: &MobWorld,
+        shared: &Shared,
+    ) {
+        let m = self.mobs[i].mob_mut();
+        horizon_client::mobs::apply_mob_pushes(m, &mut self.pushes);
+        horizon_client::mobs::blocks_changed(m, shared.path_changes);
+        let from = self.events.len();
+        // The guest's view is lent to the mob for its tick, not copied:
+        // the tick only reads it (`&MobWorld`), and it goes back to the
+        // guest's slot straight after.
+        let lent = seen_guest.and_then(|conn| self.guests.iter().position(|g| g.0 == conn));
+        let active_player = match (seen_guest, lent) {
+            (None, _) => self.player.take(),
+            (Some(_), Some(k)) => self
+                .spare
+                .take()
+                .map(|stand_in| std::mem::replace(&mut self.guests[k].1, stand_in)),
+            (Some(_), None) => None,
+        };
+        let mut w = MobWorld {
+            others: &self.snapshot,
+            others_grid: Some(&self.grid),
+            others_index: Some(&self.index),
+            others_villagers: Some(&self.villager_slots),
+            others_solid: Some(&self.solid_slots),
+            difficulty,
+            nearest_player,
+            player: active_player,
+            teams: std::sync::Arc::clone(&base.teams),
+            ..*base
+        };
+        w.controlling_player_movement = controlling_player_movement;
+        self.ticked.insert(self.mobs[i].mob().id);
+        // `horizon-report`'s cost by mob type (tick_profile.rs).
+        let sampled =
+            super::super::tick_profile::start().map(|t| (t, self.mobs[i].mob().kind.clone()));
+        let kept = horizon_client::mobs::riders::tick(
+            &mut self.mobs,
+            i,
+            &w,
+            &mut self.random,
+            &mut self.events,
+        );
+        if let Some((started, kind)) = sampled {
+            super::super::tick_profile::add(shared.level_id, &kind, started);
+        }
+        let m = self.mobs[i].mob_mut();
+        if seen_guest.is_none() {
+            self.player = w.player;
+        } else if let Some(k) = lent {
+            self.spare = w
+                .player
+                .map(|v| std::mem::replace(&mut self.guests[k].1, v));
+        }
+        // Its tick changed its target: the mods may keep the one it had
+        // (`native_living_change_target`), and then what the tick did to
+        // the new one is not dealt.
+        if let Some(&before) = self.targets.get(i)
+            && m.ai.target != before
+        {
+            let named = target_named(
+                m.ai.target,
+                seen_guest,
+                &self.guests,
+                shared.guest_ids,
+                self.player.as_ref(),
+            );
+            if let Some((player, mob)) = named {
+                let kind = super::super::native_hooks::mob_kind(m);
+                let mut cancel = false;
+                Game::native_living_change_target(
+                    shared.level_id,
+                    m.id,
+                    &kind,
+                    player,
+                    mob,
+                    &mut cancel,
+                );
+                if cancel {
+                    if let Some(new) = m.ai.target {
+                        let mut k = from;
+                        while k < self.events.len() {
+                            if hits_target(&self.events[k], new) {
+                                self.events.remove(k);
+                            } else {
+                                k += 1;
+                            }
+                        }
+                    }
+                    m.ai.target = before;
+                }
+            }
+            self.targets[i] = m.ai.target;
+        }
+        // A killed mob's body is removed: its effects that act on
+        // death run (`LivingEntity.remove` with KILLED).
+        if !kept && !m.alive() && !m.removed && !m.extra.effects.is_empty() {
+            self.corpses.push(m.clone());
+        }
+        horizon_client::mobs::take_mob_pushes(&mut self.events, from, &mut self.pushes);
+        // LivingEntity.tickDeath: after twenty ticks the body is removed.
+        if !kept && !m.removed && !m.alive() && m.death_time >= 20 {
+            let (width, height) = (f64::from(m.width()), f64::from(m.height()));
+            // `AbstractCubeMob.remove` splits a grown cube into its smaller halves when it is
+            // removed, twenty ticks after the death. The parent leaves the list before the split
+            // runs, so its size, position, kind and persistence are captured here, as they are
+            // at the removal.
+            let split = if matches!(
+                m.species(),
+                Species::Slime | Species::MagmaCube | Species::SulfurCube
+            ) && m.extra.size > 1
+            {
+                Some(m.clone())
+            } else {
+                None
+            };
+            self.gone.push((m.id, m.position, width, height, split));
+        }
+        if let Some(conn) = seen_guest {
+            // ShulkerBullet.finalTarget is the guest's entity: the
+            // bullet homes on it while the projectiles tick, where
+            // `Who::Player` is the host's own player.
+            if let Some(&(_, id)) = shared.guest_ids.iter().find(|g| g.0 == conn) {
+                for e in &mut self.events[from..] {
+                    if let MobEvent::Projectile(p) = e
+                        && let ProjectileKind::ShulkerBullet { target, .. } = &mut p.kind
+                        && *target == Some(Who::Player)
+                    {
+                        *target = Some(Who::Mob(id));
+                    }
+                }
+            }
+            let mut k = from;
+            while k < self.events.len() {
+                if super::super::lan_mobs::acts_on_player(&self.events[k]) {
+                    self.guest_events.push((conn, m.id, self.events.remove(k)));
+                } else {
+                    k += 1;
+                }
+            }
+        }
+        // Killed during its own tick (drowning, burning, falling...).
+        if !m.alive() && !m.removed && self.died.insert(m.id) {
+            self.dead.push(m.clone());
+        }
+        if !kept {
+            self.died.remove(&m.id);
+        }
+        self.keep.push(kept);
+        // ServerLevel ticks entities one after another: the ones
+        // after this mob see it as it is now.
+        self.snapshot[i] = seen(m, shared.uuid);
+        self.snapshot[i].alive &= kept;
+        self.grid.update(i, self.snapshot[i].bounds);
+        horizon_client::mobs::apply_brain_events(&mut self.mobs, &mut self.events, from);
+    }
+}

crates/horizon-client/src/game/creatures/prep.rs

@@ -0,0 +1,247 @@
+//! What the mob loop (`tick_creature_entities`) prepares for every mob before it runs, on the tick's
+//! helper threads while the parallel tick is on (`horizon_client::parallel`): each answer reads only
+//! that mob, the blocks and the players, so the helpers give what one thread gives.
+use super::super::{Game, natural_spawning::NearestPlayers};
+use horizon_client::mobs::{Mob, MobWorld, Seen};
+use horizon_client::parallel::par_map;
+use horizon_core::geometry::Vec3;
+
+/// From this many mobs a preparation goes to the helper threads.
+const MIN_MOBS: usize = 256;
+
+impl Game<'_> {
+    /// `LivingEntity.isPushable` reads onClimbable live; a mob's cached one is only refreshed by its
+    /// own push step, so one that has not ticked yet (the first tick after a load, a fresh spawn) or
+    /// has moved since would be pushed out of a ladder or scaffolding. Refreshed here, from where
+    /// each mob stands now; then each mob's entry of the snapshot others push it by goes in
+    /// `snapshot`.
+    pub(super) fn prepare_mobs(&mut self, player: Option<[i32; 4]>, snapshot: &mut Vec<Seen>) {
+        let climbing: Vec<bool> = {
+            let w = self.mob_world_base(&[]);
+            let rebuild = hookless(&w);
+            par_map(&self.mobs, MIN_MOBS, |part, out| {
+                let w = rebuild();
+                out.extend(part.iter().map(|m| m.on_climbable(&w)));
+            })
+        };
+        for (m, climbing) in self.mobs.iter_mut().zip(climbing) {
+            m.on_climbable = climbing;
+        }
+        snapshot.extend(par_map(&self.mobs, MIN_MOBS, |part, out| {
+            out.extend(part.iter().map(|m| super::seen(m, player)));
+        }));
+    }
+}
+
+/// The nearest player each mob's despawn check reads (`Mob.checkDespawn`), found for every mob before
+/// the loop. A mob that moved before its turn is asked again where it stands.
+pub(super) struct MobsNearest<'p> {
+    players: &'p NearestPlayers<'p>,
+    /// Each mob's position as the loop started and its nearest player there; empty while the
+    /// parallel tick is off (every mob asks at its turn).
+    found: Vec<(Vec3, Option<Vec3>)>,
+}
+
+impl<'p> MobsNearest<'p> {
+    /// `wanted`: whether the loop asks (not with one player, whose position answers).
+    pub(super) fn new(players: &'p NearestPlayers<'p>, mobs: &[Mob], wanted: bool) -> Self {
+        let found = if wanted && horizon_client::parallel::lanes() > 1 {
+            par_map(mobs, MIN_MOBS, |part, out| {
+                out.extend(
+                    part.iter()
+                        .map(|m| (m.position, Self::ask(players, m.position))),
+                );
+            })
+        } else {
+            Vec::new()
+        };
+        Self { players, found }
+    }
+    fn ask(players: &NearestPlayers, at: Vec3) -> Option<Vec3> {
+        players.nearest(at.x, at.y, at.z).map(|p| p.position)
+    }
+    /// The nearest player to mob `i` of the loop, standing at `at`.
+    pub(super) fn at(&self, i: usize, at: Vec3) -> Option<Vec3> {
+        match self.found.get(i) {
+            Some(&(was, nearest)) if was == at => nearest,
+            _ => Self::ask(self.players, at),
+        }
+    }
+}
+
+/// `base` without its hooks into the game, which can't leave the server thread, as a recipe a
+/// helper thread builds its own world from: native mod effects (the phases on helpers don't run
+/// them) and the UUID lookup.
+fn hookless<'w>(base: &MobWorld<'w>) -> impl Fn() -> MobWorld<'w> + Sync + use<'w> {
+    // Every field but the two hooks, named, so that a field added to `MobWorld` is decided here.
+    let MobWorld {
+        world,
+        registry,
+        facts,
+        fluid_tables,
+        movement,
+        collision,
+        suffocation,
+        data,
+        player,
+        controlling_player_movement,
+        nearest_player,
+        others,
+        dragon,
+        leash_boats,
+        others_grid,
+        others_index,
+        others_villagers,
+        others_solid,
+        daylight,
+        monsters_burn,
+        game_time,
+        day_time,
+        raining,
+        thundering,
+        level,
+        block_tags,
+        landing,
+        biomes,
+        mob_griefing,
+        universal_anger,
+        mob_drops,
+        max_entity_cramming,
+        difficulty,
+        dimension_type,
+        world_seed,
+        sea_level,
+        teams,
+        items,
+        lid_openers,
+        mod_effects: _,
+        who_lookup: _,
+        foreign: _,
+        border,
+    } = base.clone();
+    move || MobWorld {
+        world,
+        registry,
+        facts,
+        fluid_tables,
+        movement,
+        collision,
+        suffocation,
+        data,
+        player: player.clone(),
+        controlling_player_movement,
+        nearest_player,
+        others,
+        dragon,
+        leash_boats,
+        others_grid,
+        others_index,
+        others_villagers,
+        others_solid,
+        daylight,
+        monsters_burn,
+        game_time,
+        day_time,
+        raining,
+        thundering,
+        level,
+        block_tags,
+        landing,
+        biomes,
+        mob_griefing,
+        universal_anger,
+        mob_drops,
+        max_entity_cramming,
+        difficulty,
+        dimension_type,
+        world_seed,
+        sea_level,
+        teams: std::sync::Arc::clone(&teams),
+        items,
+        lid_openers,
+        mod_effects: None,
+        who_lookup: None,
+        foreign: None,
+        border,
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::super::super::natural_spawning::LevelPlayer;
+    use super::*;
+    use horizon_core::geometry::Aabb;
+
+    #[test]
+    #[ignore = "needs prepared resources and scratch homes/cwd"]
+    fn preparation_refreshes_climbing_and_nearest_after_a_move() {
+        let mut game = super::super::super::native_machines::tests::server_game();
+        game.build_registries();
+        game.set_allow_commands(true);
+        for command in [
+            "setblock 8 80 9 stone",
+            "setblock 8 80 8 ladder[facing=north]",
+            "summon pig 8.5 80 8.5 {NoAI:1b}",
+            "summon pig 9.5 80 8.5 {NoAI:1b}",
+        ] {
+            assert!(game.run_command(command), "{command}");
+        }
+        assert_eq!(game.mobs.len(), 2);
+        let examples = game.mobs.clone();
+        game.mobs = (0..MIN_MOBS + 37)
+            .map(|i| {
+                let mut mob = examples[i % 2].clone();
+                mob.id = 10_000 + i as u32;
+                mob.on_climbable = i % 2 != 0; // deliberately stale in both cases
+                mob
+            })
+            .collect::<Vec<_>>()
+            .into();
+        let mut snapshot = Vec::new();
+        game.prepare_mobs(None, &mut snapshot);
+        assert_eq!(snapshot.len(), game.mobs.len());
+        for (i, (m, seen)) in game.mobs.iter().zip(&snapshot).enumerate() {
+            assert_eq!(m.on_climbable, i % 2 == 0, "mob {i}");
+            assert_eq!(seen.id, m.id);
+            assert_eq!(seen.position, m.position);
+            assert_eq!(seen.pushable, i % 2 != 0, "snapshot {i}");
+        }
+
+        let player = |x, spectator| {
+            let position = Vec3::new(x, 80.0, 8.5);
+            LevelPlayer {
+                uuid: 0,
+                position,
+                bounds: Aabb::new(position, position),
+                alive: false, // dead players still count for checkDespawn
+                creative: false,
+                spectator,
+            }
+        };
+        let players = [player(8.5, true), player(7.5, false), player(9.5, false)];
+        let index = NearestPlayers::new(&players);
+        let nearest = MobsNearest::new(&index, &game.mobs, true);
+        for (i, mob) in game.mobs.iter().enumerate() {
+            let expected = if i % 2 == 0 {
+                players[1].position
+            } else {
+                players[2].position
+            };
+            assert_eq!(nearest.at(i, mob.position), Some(expected));
+            // A previous mob's push moved this mob across the nearest-player tie.
+            assert_eq!(
+                nearest.at(i, Vec3::new(10.0, 80.0, 8.5)),
+                Some(players[2].position)
+            );
+        }
+        let no_cache = MobsNearest::new(&index, &game.mobs, false);
+        assert_eq!(
+            no_cache.at(0, game.mobs[0].position),
+            Some(players[1].position)
+        );
+        let spectators = [player(8.5, true)];
+        let index = NearestPlayers::new(&spectators);
+        let nearest = MobsNearest::new(&index, &game.mobs, true);
+        assert_eq!(nearest.at(0, game.mobs[0].position), None);
+    }
+}

crates/horizon-client/src/game/entity.rs

@@ -594,24 +594,9 @@ impl Game<'_> {
         splash: &mut impl FnMut(horizon_client::entities::ItemSplash),
     ) -> (bool, bool) {
         let no_physics = escape_speed.is_some();
-        let heights = item.push_by_fluid(&self.world, &self.res.motion().fluids);
-        // Entity.baseTick splashes before moving, except on the first tick.
-        item.water_entry(heights.0 > 0.0, item.ticks == 0, splash);
-        item.lava_fall(heights.1 > 0.0);
-        let step = item.tick_move_before_block_effects(
-            &self.world,
-            self.res.facts(),
-            self.collision,
-            no_physics,
-            escape_speed,
-            heights,
-            entity_boxes,
-            Some(self.border.geometry()),
-            Some(&self.res.motion().blocks),
-        );
-        item.check_fall(&step, |fall, from, moved| {
-            self.vehicle_fall_reset(fall, from, moved)
-        });
+        let step = self
+            .item_tables()
+            .move_item(item, escape_speed, entity_boxes, splash);
         if let Some(pos) = step.vertical_contact {
             let state = self.world.block(pos.x, pos.y, pos.z);
             let position = item.position;
@@ -630,23 +615,9 @@ impl Game<'_> {
                 return (false, true);
             }
         }
-        item.finish_move_block_effects(
-            &self.world,
-            self.res.facts(),
-            no_physics,
-            Some(&self.res.motion().blocks),
-            step,
-        );
-        item.push_by_bubble_column(
-            &self.world,
-            self.pack.assets.registry,
-            self.res.facts(),
-            self.collision,
-        );
-        let alive = item.tick_settle(&self.world, self.res.facts(), no_physics, step);
-        // ItemEntity.tick refreshes immersion and splashes on water entry.
-        let (water, _) = item.push_by_fluid(&self.world, &self.res.motion().fluids);
-        item.water_entry(water > 0.0, false, splash);
+        let alive = self
+            .item_tables()
+            .settle_item(item, no_physics, step, splash);
         (alive, false)
     }
 

crates/horizon-client/src/game/item_entities.rs

@@ -2,6 +2,8 @@
 //! `tick_experience`), the stacks a destroyed container lets fall, and spawning and merging orbs.
 use super::*;
 
+mod ahead;
+
 impl Game<'_> {
     pub(super) fn tick_items(&mut self) {
         // EntityTickEvent.Pre for the items about to tick, and the Post after them.
@@ -43,6 +45,13 @@ impl Game<'_> {
             horizon_client::entities::ObstacleGrid::new(self.item_entity_collisions());
         let mut nearby = Vec::new();
         let mut bubble_tops = Vec::new();
+        let mut moves = ahead::Moves::work_out(self, &items, &obstacles, round, |item| {
+            !self.joined_during_entity_pass(item.id)
+                && ticking(
+                    item.position.x.floor() as i32,
+                    item.position.z.floor() as i32,
+                )
+        });
         // `ItemUtils.onContainerDestroyed`: what a destroyed item held, and
         // where it lay, for adding to the level once its own tick is done.
         let mut destroyed: Vec<(ItemStack, Vec3)> = Vec::new();
@@ -65,33 +74,35 @@ impl Game<'_> {
             {
                 continue;
             }
-            // `getEntityCollisions(entity, bounds.expandTowards(movement))`: the boxes around
-            // where the item can reach this tick (a block of slack for the next velocity).
-            let reach = Vec3::new(
-                item.velocity.x.abs() + 1.0,
-                item.velocity.y.abs() + 1.0,
-                item.velocity.z.abs() + 1.0,
-            );
-            obstacles.near(item.bounds().inflate(reach), &mut nearby);
-            // ItemEntity.tick's noPhysics (servers only): out of a block, a solid entity
-            // or the border, placed or closed on it, towards the closest open side (#1252).
-            let no_physics = merging
-                && item.inside_block(
-                    &self.world,
-                    self.collision,
+            let mut at_turn = |item: &mut ItemEntity| {
+                // `getEntityCollisions(entity, bounds.expandTowards(movement))`: the boxes around
+                // where the item can reach this tick.
+                obstacles.near(ahead::reach(item), &mut nearby);
+                // ItemEntity.tick's noPhysics (servers only): out of a block, a solid entity
+                // or the border, placed or closed on it, towards the closest open side (#1252).
+                let no_physics = merging
+                    && item.inside_block(
+                        &self.world,
+                        self.collision,
+                        &nearby,
+                        Some(self.border.geometry()),
+                    );
+                // `moveTowardsClosestSpace` draws the speed from the item's own random.
+                let escape_speed = no_physics.then(|| item.random.next_float() * 0.2 + 0.1);
+                let mut entered = Vec::new();
+                let (alive, discarded) = self.tick_item_with_block_contacts(
+                    item,
+                    escape_speed,
                     &nearby,
-                    Some(self.border.geometry()),
+                    false,
+                    &mut |splash| entered.push(splash),
                 );
-            // `moveTowardsClosestSpace` draws the speed from the item's own random.
-            let escape_speed = no_physics.then(|| item.random.next_float() * 0.2 + 0.1);
-            let mut entered = Vec::new();
-            let (alive, discarded) = self.tick_item_with_block_contacts(
-                item,
-                escape_speed,
-                &nearby,
-                false,
-                &mut |splash| entered.push(splash),
-            );
+                (no_physics, alive, discarded, entered)
+            };
+            let (no_physics, alive, discarded, entered) = match moves.take(i, item, &mut at_turn) {
+                Some(moved) => moved,
+                None => at_turn(item),
+            };
             // Entity.doWaterSplashEffect: the server plays the sound (broadcast
             // to clients). Its addParticle is empty.
             for splash in entered {

crates/horizon-client/src/game/item_entities/ahead.rs

@@ -0,0 +1,336 @@
+//! The dropped items' own moves, worked out ahead of the item loop on the tick's helper threads
+//! while the parallel tick is on (`horizon_client::parallel`).
+//!
+//! An item's move (`ItemEntity.tick` up to its merge) reads only itself, the level's blocks, the
+//! obstacle boxes and the border, and the item loop changes none of them: so an item's move worked
+//! out on a copy before the loop is the move its turn would make. Its escape from inside a block
+//! draws the item's own random, which the copy carries. A merge before its turn can change its
+//! count, age and pickup delay: such an item moves at its turn instead. With a native mod loaded
+//! (its hooks see every item) or a mod listening to the round, every item moves at its turn.
+use super::super::{boats::fall_reset, entity::Entity as _, native_ticks};
+use super::*;
+use horizon_client::{
+    blocks::BlockFacts,
+    entities::{ItemEntity, ItemSplash, ItemStep, ObstacleGrid},
+    motion::MotionResources,
+    world::World,
+};
+use horizon_core::geometry::{Aabb, Vec3, border::BorderGeometry};
+use horizon_world::{block_collision::BlockCollisionRegistry, registry::BlockRegistry};
+
+/// From this many items their moves go to the helper threads.
+const MIN_ITEMS: usize = 256;
+
+/// What an item's own move reads.
+pub(in crate::game) struct ItemTables<'a> {
+    world: &'a World,
+    facts: &'a BlockFacts,
+    collision: &'a BlockCollisionRegistry<'a>,
+    motion: &'a MotionResources,
+    registry: &'a BlockRegistry,
+    border: &'a BorderGeometry,
+}
+
+impl<'w> Game<'w> {
+    pub(in crate::game) fn item_tables(&self) -> ItemTables<'_> {
+        ItemTables {
+            world: &self.world,
+            facts: self.res.facts(),
+            collision: self.collision,
+            motion: self.res.motion(),
+            registry: self.pack.assets.registry,
+            border: self.border.geometry(),
+        }
+    }
+}
+
+impl ItemTables<'_> {
+    /// `ItemEntity.tick` up to its move's vertical contact: the fluids, the water splash, the move
+    /// (pushed out at `escape_speed` when inside a block) and its fall.
+    pub(in crate::game) fn move_item(
+        &self,
+        item: &mut ItemEntity,
+        escape_speed: Option<f32>,
+        entity_boxes: &[Aabb],
+        splash: &mut impl FnMut(ItemSplash),
+    ) -> ItemStep {
+        let heights = item.push_by_fluid(self.world, &self.motion.fluids);
+        // Entity.baseTick splashes before moving, except on the first tick.
+        item.water_entry(heights.0 > 0.0, item.ticks == 0, splash);
+        item.lava_fall(heights.1 > 0.0);
+        let step = item.tick_move_before_block_effects(
+            self.world,
+            self.facts,
+            self.collision,
+            escape_speed.is_some(),
+            escape_speed,
+            heights,
+            entity_boxes,
+            Some(self.border),
+            Some(&self.motion.blocks),
+        );
+        item.check_fall(&step, |fall, from, moved| {
+            fall_reset(self.world, self.motion, self.facts, fall, from, moved)
+        });
+        step
+    }
+    /// The rest of the move after its vertical contact: block effects, bubble columns, settling
+    /// and the water entry. Whether the item lives on.
+    pub(in crate::game) fn settle_item(
+        &self,
+        item: &mut ItemEntity,
+        no_physics: bool,
+        step: ItemStep,
+        splash: &mut impl FnMut(ItemSplash),
+    ) -> bool {
+        item.finish_move_block_effects(
+            self.world,
+            self.facts,
+            no_physics,
+            Some(&self.motion.blocks),
+            step,
+        );
+        item.push_by_bubble_column(self.world, self.registry, self.facts, self.collision);
+        let alive = item.tick_settle(self.world, self.facts, no_physics, step);
+        // ItemEntity.tick refreshes immersion and splashes on water entry.
+        let (water, _) = item.push_by_fluid(self.world, &self.motion.fluids);
+        item.water_entry(water > 0.0, false, splash);
+        alive
+    }
+}
+
+/// `getEntityCollisions(entity, bounds.expandTowards(movement))`: the box around an item that its
+/// move reads others' boxes in (a block of slack for the next velocity).
+pub(super) fn reach(item: &ItemEntity) -> Aabb {
+    let v = item.velocity;
+    item.bounds()
+        .inflate(Vec3::new(v.x.abs() + 1.0, v.y.abs() + 1.0, v.z.abs() + 1.0))
+}
+
+/// An item's move as its turn makes it: without physics (inside a block), alive, discarded, and
+/// its splashes.
+pub(super) type Moved = (bool, bool, bool, Vec<ItemSplash>);
+
+/// One item's move, worked out on a copy.
+pub(super) struct Ahead {
+    moved: ItemEntity,
+    result: Moved,
+    /// What a merge before its turn can change: its count, age and pickup delay.
+    started: (u32, i32, i64),
+}
+
+fn started(item: &ItemEntity) -> (u32, i32, i64) {
+    (item.stack.count, item.age, item.pickup_delay)
+}
+
+/// `item`'s whole move on a server with no native mod, from the tables alone.
+fn move_alone(
+    tables: &ItemTables,
+    obstacles: &ObstacleGrid,
+    item: &mut ItemEntity,
+    found: &mut Vec<u32>,
+    near: &mut Vec<Aabb>,
+) -> Moved {
+    obstacles.near_shared(reach(item), found, near);
+    let no_physics = item.inside_block(tables.world, tables.collision, near, Some(tables.border));
+    // `moveTowardsClosestSpace` draws the speed from the item's own random.
+    let escape_speed = no_physics.then(|| item.random.next_float() * 0.2 + 0.1);
+    let mut splashes = Vec::new();
+    let step = tables.move_item(item, escape_speed, near, &mut |s| splashes.push(s));
+    if step.vertical_contact.is_some() && item.stack.count == 0 {
+        return (no_physics, false, true, splashes);
+    }
+    let alive = tables.settle_item(item, no_physics, step, &mut |s| splashes.push(s));
+    (no_physics, alive, false, splashes)
+}
+
+#[cfg(test)]
+thread_local! {
+    /// A test's choice of the path, over the parallel tick's switch.
+    static AHEAD: std::cell::Cell<Option<bool>> = const { std::cell::Cell::new(None) };
+    /// The moves a test's tick put in place, and those a merge made it refuse.
+    static TAKEN: std::cell::Cell<(usize, usize)> = const { std::cell::Cell::new((0, 0)) };
+}
+
+/// Whether the moves are worked out ahead: while the parallel tick is on.
+fn ahead() -> bool {
+    #[cfg(test)]
+    if let Some(ahead) = AHEAD.with(std::cell::Cell::get) {
+        return ahead;
+    }
+    horizon_client::parallel::lanes() > 1
+}
+
+/// The moves worked out ahead, by the item's place in the loop.
+pub(super) struct Moves(Vec<Option<Ahead>>);
+
+impl Moves {
+    /// The moves of the items that will tick (`ticks`), on the helper threads; none while the
+    /// parallel tick is off, for few items, or with a mod that could see an item's move.
+    pub(super) fn work_out(
+        game: &Game,
+        items: &[ItemEntity],
+        obstacles: &ObstacleGrid,
+        round: &native_ticks::TickRound,
+        ticks: impl Fn(&ItemEntity) -> bool,
+    ) -> Self {
+        if !ahead()
+            || items.len() < MIN_ITEMS
+            || game.remote.is_some()
+            || round.listening()
+            || !horizon_client::mod_api::mods().is_empty()
+        {
+            return Self(Vec::new());
+        }
+        let todo: Vec<usize> = (0..items.len()).filter(|&i| ticks(&items[i])).collect();
+        let tables = game.item_tables();
+        let done = horizon_client::parallel::par_map(&todo, MIN_ITEMS / 4, |part, out| {
+            let (mut found, mut near) = (Vec::new(), Vec::new());
+            for &i in part {
+                let mut moved = items[i].clone();
+                let result = move_alone(&tables, obstacles, &mut moved, &mut found, &mut near);
+                let started = started(&items[i]);
+                out.push((
+                    i,
+                    Ahead {
+                        moved,
+                        result,
+                        started,
+                    },
+                ));
+            }
+        });
+        let mut moves: Vec<Option<Ahead>> = (0..items.len()).map(|_| None).collect();
+        for (i, a) in done {
+            moves[i] = Some(a);
+        }
+        Self(moves)
+    }
+    /// Item `i`'s move, put in place, when it was worked out and no merge changed the item since.
+    /// Under `HORIZON_PARALLEL_TICK=verify` the move is made again at its turn, on a copy, and a
+    /// difference is logged.
+    pub(super) fn take(
+        &mut self,
+        i: usize,
+        item: &mut ItemEntity,
+        again: impl FnOnce(&mut ItemEntity) -> Moved,
+    ) -> Option<Moved> {
+        let a = self.0.get_mut(i)?.take()?;
+        let fits = a.started == started(item);
+        #[cfg(test)]
+        TAKEN.with(|t| {
+            let (taken, refused) = t.get();
+            t.set((taken + usize::from(fits), refused + usize::from(!fits)));
+        });
+        if !fits {
+            return None;
+        }
+        if horizon_client::parallel::parallel_tick() == 2 {
+            let mut serial = item.clone();
+            let expected = again(&mut serial);
+            check(item.id, &serial, &expected, &a.moved, &a.result);
+        }
+        *item = a.moved;
+        Some(a.result)
+    }
+}
+
+/// `HORIZON_PARALLEL_TICK=verify`: an item's move worked out ahead against the one at its turn.
+fn check(id: u32, serial: &ItemEntity, expected: &Moved, ahead: &ItemEntity, got: &Moved) {
+    use std::sync::atomic::{AtomicU64, Ordering};
+    static CHECKED: AtomicU64 = AtomicU64::new(0);
+    static DIFFERED: AtomicU64 = AtomicU64::new(0);
+    if format!("{serial:?} {expected:?}") != format!("{ahead:?} {got:?}") {
+        DIFFERED.fetch_add(1, Ordering::Relaxed);
+        log::error!("parallel tick: item {id}'s move worked out ahead differs from its turn's");
+    }
+    let n = CHECKED.fetch_add(1, Ordering::Relaxed) + 1;
+    if n == 1 || n.is_multiple_of(10_000) {
+        log::info!(
+            "parallel tick: {n} item moves worked out ahead checked, {} differed",
+            DIFFERED.load(Ordering::Relaxed)
+        );
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::super::super::native_machines::tests::server_game;
+    use super::{AHEAD, TAKEN};
+    use horizon_client::{chunk::Chunk, items::ItemStack};
+    use std::sync::Arc;
+
+    /// One item tick with the moves worked out ahead leaves every item, and the level's random,
+    /// as the tick that moves each item at its turn: items escaping a block (their own random),
+    /// floating in water, falling, and merging into a later item before its turn (whose move
+    /// is then made at its turn).
+    #[test]
+    #[ignore = "needs prepared resources in HORIZON_TEST_RESOURCES"]
+    fn moves_worked_out_ahead_match_those_made_at_each_turn() {
+        let mut game = server_game();
+        let feet = horizon_core::geometry::Vec3::new(8.5, 70.0, 8.5);
+        game.player = Some(horizon_client::motion::PlayerMotion::new(game.motion, feet).unwrap());
+        // `ChunkMap.prepareEntityTickingChunk`: the chunks within 2 are loaded too.
+        for x in -2..=2 {
+            for z in -2..=2 {
+                game.world
+                    .chunks
+                    .insert((x, z), Arc::new(Chunk::empty(x, z)));
+            }
+        }
+        game.update_chunk_tickets();
+        let registry = game.pack.assets.registry;
+        let stone = registry.default_state("minecraft:stone").unwrap();
+        let water = registry.default_state("minecraft:water").unwrap();
+        for x in 0..16 {
+            for z in 0..16 {
+                let (fx, fz) = (f64::from(x) + 0.5, f64::from(z) + 0.5);
+                let stack = |count| ItemStack::new("minecraft:stone", count);
+                match (x + z) % 5 {
+                    0 => {
+                        game.world.set(x, 65, z, stone).unwrap();
+                        game.spawn_item(stack(1), [fx, 65.3, fz], [0.0; 3], 10);
+                    }
+                    1 => {
+                        game.world.set(x, 65, z, water).unwrap();
+                        game.spawn_item(stack(1), [fx, 65.5, fz], [0.0; 3], 10);
+                    }
+                    2 => {
+                        game.spawn_item(stack(1), [fx, 66.0, fz], [0.0; 3], 0);
+                        game.spawn_item(stack(5), [fx, 66.0, fz], [0.0; 3], 0);
+                    }
+                    _ => game.spawn_item(stack(1), [fx, 67.0, fz], [0.1, 0.2, -0.1], 10),
+                }
+            }
+        }
+        for item in game.dropped.iter_mut() {
+            item.ticks = 39;
+        }
+        game.ticked_entities_through = u32::MAX;
+        let (items, random) = (game.dropped.clone(), game.world.random.clone());
+        let mut runs = Vec::new();
+        for ahead in [false, true] {
+            game.dropped = items.clone();
+            game.world.random = random.clone();
+            AHEAD.with(|a| a.set(Some(ahead)));
+            TAKEN.with(|t| t.set((0, 0)));
+            game.tick_items();
+            let left: Vec<String> = game.dropped.iter().map(|i| format!("{i:?}")).collect();
+            runs.push((
+                left,
+                format!("{:?}", game.world.random),
+                TAKEN.with(|t| t.get()),
+            ));
+        }
+        AHEAD.with(|a| a.set(None));
+        assert!(runs[0].0.len() < items.len(), "some items merged");
+        assert_eq!(runs[0].0, runs[1].0);
+        assert_eq!(runs[0].1, runs[1].1);
+        // The path under test ran: most moves were put in place, and a merge refused some.
+        let (taken, refused) = runs[1].2;
+        assert!(
+            taken >= 200 && refused > 0,
+            "{taken} put in place, {refused} refused"
+        );
+    }
+}

crates/horizon-client/src/game/lan.rs

@@ -4,6 +4,7 @@
 //! costs one `Option` check per tick.
 mod event_timings;
 mod level_event_targets;
+mod mob_states;
 
 use super::{Game, shield::OFFHAND_SLOT};
 use event_timings::EventTimings;
@@ -363,12 +364,12 @@ pub(super) fn baby_index(species: Species) -> Option<u8> {
 /// own: `mob_meta`'s arms hold these mobs' other entries.
 fn mob_variant_meta(
     m: &Mob,
-    hub: &Hub,
+    sync: &SyncData,
     meta: &mut Vec<(u8, i32, Value)>,
     defaults: &mut Vec<(u8, i32, Value)>,
 ) {
     let (kind, e) = (m.kind.id.as_str(), &m.extra);
-    let id = |registry: &str, name: &str| hub.registry_id(registry, &format!("minecraft:{name}"));
+    let id = |registry: &str, name: &str| sync.id(registry, &format!("minecraft:{name}"));
     match kind {
         // Axolotl.DATA_VARIANT (after AgeableMob's two).
         "minecraft:axolotl" => {
@@ -501,7 +502,7 @@ fn mob_variant_meta(
 /// `defaults` gets the defaults of the entries that can go back to theirs.
 pub(super) fn mob_meta(
     m: &Mob,
-    hub: &Hub,
+    sync: &SyncData,
     player_entity: Option<i32>,
     game_time: i64,
     meta: &mut Vec<(u8, i32, Value)>,
@@ -558,7 +559,7 @@ pub(super) fn mob_meta(
     // LivingEntity.DATA_EFFECT_PARTICLES and DATA_EFFECT_AMBIENCE_ID: the
     // swirls of its effects (the ones a mob has are visible, not ambient).
     meta.extend(
-        hub.effect_meta(
+        sync.effect_meta(
             (m.extra.effects.iter())
                 .filter(|e| !m.extra.hidden_effects.contains(&e.0))
                 .map(|e| (e.0, m.extra.ambient_effects.contains(&e.0))),
@@ -590,7 +591,7 @@ pub(super) fn mob_meta(
         }
     }
     let e = &m.extra;
-    let id = |registry: &str, name: &str| hub.registry_id(registry, &format!("minecraft:{name}"));
+    let id = |registry: &str, name: &str| sync.id(registry, &format!("minecraft:{name}"));
     // Pig.DATA_BOOST_TIME and Strider.DATA_BOOST_TIME (18, the first entry
     // after AgeableMob's two). The total `boost()` stored in `counter2`.
     // The default 0 is not registered: vanilla never writes it back, and the
@@ -598,7 +599,7 @@ pub(super) fn mob_meta(
     if matches!(kind, "minecraft:pig" | "minecraft:strider") && e.counter2 > 0 {
         meta.push((18, serializer::INT, Value::Int(e.counter2)));
     }
-    mob_variant_meta(m, hub, meta, defaults);
+    mob_variant_meta(m, sync, meta, defaults);
     mob_raider_meta(m, meta, defaults);
     match kind {
         "minecraft:guardian" | "minecraft:elder_guardian" => {
@@ -1948,14 +1949,15 @@ impl Game<'_> {
         mut kept: Option<&mut super::lan_idle::IdlePass<'_>>,
     ) {
         out.clear();
-        let mut speeds = super::lan_attributes::SpeedByType::default();
+        let mut states = mob_states::Later::new(self.world.game_time);
         // ChunkMap.tick sends the changes of an entity outside the
         // entity-ticking range only when it changed section or needs sync.
         let levels = self.tickets.levels();
         let idle_at = |p: horizon_core::geometry::Vec3| {
             !levels.in_entity_ticking_range((p.x.floor() as i32) >> 4, (p.z.floor() as i32) >> 4)
         };
-        for m in self.mobs.iter().filter(|m| !m.removed && m.id > after) {
+        for (index, m) in (self.mobs.iter().enumerate()).filter(|(_, m)| !m.removed && m.id > after)
+        {
             // The lead's holder: the player the mob sees, another mob or a
             // fence knot (game/lan_entities.rs).
             let player_entity = || {
@@ -1986,6 +1988,7 @@ impl Game<'_> {
             let bare = idle
                 && (kept.as_deref_mut()).is_some_and(|k| k.bare(hub, m.id as i32, kind, &place))
                 && !m.needs_sync;
+            let slot = out.as_slice().len();
             let e = out.add(m.id as i32, entity_uuid(m.id), kind);
             e.position = place.position;
             (e.yaw, e.pitch, e.head_yaw) = place.rotation;
@@ -1999,21 +2002,11 @@ impl Game<'_> {
             if bare {
                 continue;
             }
-            mob_meta(
-                m,
-                hub,
-                player_entity(),
-                self.world.game_time,
-                &mut e.meta,
-                &mut e.defaults,
-            );
-            // HappyGhast.aiStep: requiresPrecisePosition while on a still timeout.
-            e.precise = m.species() == horizon_client::mobs::Species::HappyGhast
-                && horizon_client::mobs::happy_ghast::on_still_timeout(m);
-            mob_freeze_meta(m, &mut e.meta, &mut e.defaults);
+            let (sync, sources) = (hub.sync_data(), self.attr_sources());
+            states.state((slot, index), m, player_entity(), sync, sources, e);
             mob_equipment(m, hub, &mut e.equipment);
-            self.mob_attributes(m, &mut speeds, &mut e.attributes);
         }
+        states.fill(&self.mobs, hub.sync_data(), self.attr_sources(), out);
         for d in self.dropped.iter().filter(|d| d.id > after) {
             let idle = idle_at(d.position);
             let place = super::lan_idle::Place {
@@ -4116,7 +4109,7 @@ mod tests {
         let hub = meta_hub("RaiderMetaTest");
         let meta_of = |m: &Mob| {
             let (mut meta, mut defaults) = (Vec::new(), Vec::new());
-            super::mob_meta(m, &hub, None, 0, &mut meta, &mut defaults);
+            super::mob_meta(m, hub.sync_data(), None, 0, &mut meta, &mut defaults);
             (meta, defaults)
         };
 
@@ -4145,7 +4138,7 @@ mod tests {
             // The same health: the ghoul's own maximum is 40, the zombie's 20.
             m.health = 10.0;
             let (mut meta, mut defaults) = (Vec::new(), Vec::new());
-            super::mob_meta(&m, &hub, None, 0, &mut meta, &mut defaults);
+            super::mob_meta(&m, hub.sync_data(), None, 0, &mut meta, &mut defaults);
             (meta, defaults)
         };
         for (template, id) in [

crates/horizon-client/src/game/lan/mob_states.rs

@@ -0,0 +1,286 @@
+//! The part of each mob's state for its clients (`Game::lan_entities`) that reads only the mob and
+//! the synced tables: its entity data and attribute instances. While the parallel tick is on
+//! (`horizon_client::parallel`) it is built after the pass, on the tick's helper threads, and put in
+//! place; otherwise in place during the pass, as one thread always did.
+use super::super::lan_attributes::{AttrSources, SpeedByType};
+use super::{mob_freeze_meta, mob_meta};
+use horizon_client::lan::{
+    data::SyncData,
+    entity::{EntityState, States},
+};
+use horizon_client::mobs::Mob;
+use horizon_protocol::{entity_data::Value, packets::play::AttributeSnapshot};
+
+/// From this many mobs the states go to the helper threads.
+const MIN_MOBS: usize = 1024;
+
+/// One mob's entity data (`meta`, and the `defaults` its entries can go back to) and attribute
+/// instances; whether its position is sent precisely.
+#[allow(clippy::too_many_arguments)]
+pub(super) fn mob_state(
+    m: &Mob,
+    sync: &SyncData,
+    sources: AttrSources,
+    player_entity: Option<i32>,
+    game_time: i64,
+    speeds: &mut SpeedByType,
+    meta: &mut Vec<(u8, i32, Value)>,
+    defaults: &mut Vec<(u8, i32, Value)>,
+    attributes: &mut Vec<AttributeSnapshot>,
+) -> bool {
+    mob_meta(m, sync, player_entity, game_time, meta, defaults);
+    mob_freeze_meta(m, meta, defaults);
+    sources.mob_attributes(m, speeds, attributes);
+    // HappyGhast.aiStep: requiresPrecisePosition while on a still timeout.
+    m.species() == horizon_client::mobs::Species::HappyGhast
+        && horizon_client::mobs::happy_ghast::on_still_timeout(m)
+}
+
+#[cfg(test)]
+thread_local! {
+    /// A test's choice of the path, over the parallel tick's switch.
+    static DEFER: std::cell::Cell<Option<bool>> = const { std::cell::Cell::new(None) };
+}
+
+/// Whether the states wait for the helper threads: while the parallel tick is on.
+fn deferring() -> bool {
+    #[cfg(test)]
+    if let Some(defer) = DEFER.with(std::cell::Cell::get) {
+        return defer;
+    }
+    horizon_client::parallel::lanes() > 1
+}
+
+/// The mobs' states of one pass: built in place, or (while the parallel tick is on) the mobs whose
+/// states wait for the helper threads: (their slot in the states, their index in the mobs, the
+/// player entity their lead may name).
+pub(super) struct Later {
+    waiting: Option<Vec<(usize, usize, Option<i32>)>>,
+    game_time: i64,
+    speeds: SpeedByType,
+}
+
+impl Later {
+    pub(super) fn new(game_time: i64) -> Self {
+        Self {
+            waiting: deferring().then(Vec::new),
+            game_time,
+            speeds: SpeedByType::default(),
+        }
+    }
+    /// Mob `index`'s state `e`, in `slot` of the states: built now, or left for [`Self::fill`].
+    pub(super) fn state(
+        &mut self,
+        (slot, index): (usize, usize),
+        m: &Mob,
+        player_entity: Option<i32>,
+        sync: &SyncData,
+        sources: AttrSources,
+        e: &mut EntityState,
+    ) {
+        if let Some(waiting) = &mut self.waiting {
+            waiting.push((slot, index, player_entity));
+            return;
+        }
+        let (meta, defaults, attributes) = (&mut e.meta, &mut e.defaults, &mut e.attributes);
+        e.precise = mob_state(
+            m,
+            sync,
+            sources,
+            player_entity,
+            self.game_time,
+            &mut self.speeds,
+            meta,
+            defaults,
+            attributes,
+        );
+    }
+    /// The waiting states, built on the helper threads and put in place.
+    pub(super) fn fill(
+        self,
+        mobs: &[Mob],
+        sync: &SyncData,
+        sources: AttrSources,
+        out: &mut States,
+    ) {
+        let Some(waiting) = self.waiting else {
+            return;
+        };
+        let game_time = self.game_time;
+        let built = horizon_client::parallel::par_map(&waiting, MIN_MOBS, |part, done| {
+            let mut speeds = SpeedByType::default();
+            for &(slot, index, player) in part {
+                let (mut meta, mut defaults, mut attributes) = (Vec::new(), Vec::new(), Vec::new());
+                let precise = mob_state(
+                    &mobs[index],
+                    sync,
+                    sources,
+                    player,
+                    game_time,
+                    &mut speeds,
+                    &mut meta,
+                    &mut defaults,
+                    &mut attributes,
+                );
+                done.push((slot, meta, defaults, attributes, precise));
+            }
+        });
+        let states = out.as_mut_slice();
+        for (slot, meta, defaults, attributes, precise) in built {
+            let e = &mut states[slot];
+            (e.meta, e.defaults, e.attributes, e.precise) = (meta, defaults, attributes, precise);
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use horizon_client::lan::{NetConfig, hub::HubConfig};
+
+    /// The states put in place after the pass are those built in place during it, for mobs whose
+    /// entity data and attributes vary: effects, armor, a horse's rolled jump, variants, a baby,
+    /// a happy ghast, and what `/attribute` changed.
+    #[test]
+    #[ignore = "needs prepared resources and scratch homes/cwd"]
+    fn deferred_states_match_those_built_in_place() {
+        crate::mod_api::in_own_process(
+            "game::lan::mob_states::tests::deferred_states_match_those_built_in_place",
+            deferred_states_match_those_built_in_place_in_own_process,
+        );
+    }
+
+    fn deferred_states_match_those_built_in_place_in_own_process() {
+        let mut game = super::super::super::native_machines::tests::server_game();
+        game.build_registries();
+        game.set_allow_commands(true);
+        for command in [
+            "summon piglin 8.5 80 8.5 {NoAI:1b,PersistenceRequired:1b}",
+            "effect give @e[type=piglin] speed 30 1",
+            "item replace entity @e[type=piglin,limit=1] armor.head with iron_helmet",
+            "summon horse 10.5 80 8.5 {NoAI:1b}",
+            "summon sheep 12.5 80 8.5 {NoAI:1b,Color:3b,Age:-24000}",
+            "summon axolotl 14.5 80 8.5 {NoAI:1b,Variant:2}",
+            "summon happy_ghast 16.5 84 8.5 {NoAI:1b}",
+            "summon pig 18.5 80 8.5 {NoAI:1b}",
+            "summon frog 20.5 80 8.5 {NoAI:1b}",
+            "attribute @e[type=pig,limit=1] minecraft:scale base set 1.5",
+        ] {
+            assert!(game.run_command(command), "{command}");
+        }
+        let examples = game.mobs.clone();
+        game.mobs = (0..MIN_MOBS * 2 + 37)
+            .map(|i| {
+                let mut mob = examples[i % examples.len()].clone();
+                mob.id = 10_000 + i as u32;
+                mob.removed = i % 11 == 0;
+                mob
+            })
+            .collect::<Vec<_>>()
+            .into();
+        game.lan_host(
+            super::super::LanSettings {
+                port: 0,
+                guest_commands: true,
+                force_game_mode: false,
+            },
+            NetConfig::loopback(0, String::new(), 2),
+            HubConfig {
+                default_game_mode: 0,
+                guest_commands: true,
+                force_game_mode: false,
+                hardcore: false,
+                view_distance: 2,
+                simulation_distance: 2,
+                dimension: "minecraft:overworld".into(),
+                allow_flight: false,
+                seed: 0,
+                max_players: 20,
+                integrated: false,
+                levels: Default::default(),
+            },
+        )
+        .unwrap();
+        let mut lan = game.lan.take().unwrap();
+        // The shared tables must retain pack-selected effect colors, not just
+        // vanilla's registry color. This process owns the global pack palette.
+        horizon_client::brewing::set_pack_colors(vec![("speed".into(), 0xAA12_3456)]);
+        let players = Default::default();
+        let mut runs = Vec::new();
+        struct Reset;
+        impl Drop for Reset {
+            fn drop(&mut self) {
+                DEFER.with(|d| d.set(None));
+            }
+        }
+        let _reset = Reset;
+        for after in [0, 10_012] {
+            runs.clear();
+            for defer in [false, true, false] {
+                DEFER.with(|d| d.set(Some(defer)));
+                let mut states = States::default();
+                game.lan_entities(&players, &mut lan.hub, &mut states, false, after, None);
+                runs.push(states.as_slice().to_vec());
+            }
+            let expected_ids: Vec<_> = game
+                .mobs
+                .iter()
+                .filter(|m| !m.removed && m.id > after)
+                .map(|m| m.id as i32)
+                .collect();
+            assert_eq!(
+                runs[0].iter().map(|e| e.id).collect::<Vec<_>>(),
+                expected_ids
+            );
+            assert!(runs[0].iter().filter(|e| !e.meta.is_empty()).count() >= MIN_MOBS);
+            assert!(runs[0].iter().any(|e| !e.attributes.is_empty()));
+            let colored: Vec<_> = runs[0]
+                .iter()
+                .filter(|e| e.kind == "minecraft:piglin")
+                .collect();
+            assert!(!colored.is_empty());
+            for e in colored {
+                let particles = e.meta.iter().find(|e| e.0 == 10).unwrap();
+                let Value::Particles(particles) = &particles.2 else {
+                    panic!("effect particle metadata");
+                };
+                assert_eq!(particles.len(), 1);
+                assert_eq!(
+                    particles[0].data,
+                    Some(horizon_protocol::entity_data::ParticleData::Color(
+                        0xFF12_3456_u32 as i32
+                    ))
+                );
+            }
+            assert_eq!(runs[0], runs[1], "deferred, after {after}");
+            assert_eq!(runs[0], runs[2], "switch off, after {after}");
+
+            // With deferral off the caller can read the metadata immediately,
+            // before fill, and fill must not replace the vectors it already owns.
+            let m = game
+                .mobs
+                .iter()
+                .find(|m| !m.removed && m.id > after)
+                .unwrap();
+            let mut states = States::default();
+            let e = states.add(m.id as i32, 0, &m.kind.id);
+            let mut later = Later::new(game.world.game_time);
+            later.state((0, 0), m, None, lan.hub.sync_data(), game.attr_sources(), e);
+            assert_eq!(e.meta, runs[0][0].meta);
+            assert_eq!(e.defaults, runs[0][0].defaults);
+            assert_eq!(e.attributes, runs[0][0].attributes);
+            let pointers = (e.meta.as_ptr(), e.defaults.as_ptr(), e.attributes.as_ptr());
+            later.fill(
+                &game.mobs,
+                lan.hub.sync_data(),
+                game.attr_sources(),
+                &mut states,
+            );
+            let e = &states.as_slice()[0];
+            assert_eq!(
+                pointers,
+                (e.meta.as_ptr(), e.defaults.as_ptr(), e.attributes.as_ptr())
+            );
+        }
+    }
+}

crates/horizon-client/src/game/lan/sulfur_cube_meta_tests.rs

@@ -22,7 +22,7 @@ fn a_sulfur_cube_sends_its_fuse_at_19_and_from_bucket_at_20() {
     let hub = meta_hub("SulfurCubeMetaTest");
     let meta_of = |m: &Mob| {
         let (mut meta, mut defaults) = (Vec::new(), Vec::new());
-        super::mob_meta(m, &hub, None, 0, &mut meta, &mut defaults);
+        super::mob_meta(m, hub.sync_data(), None, 0, &mut meta, &mut defaults);
         (meta, defaults)
     };
 

crates/horizon-client/src/game/lan_attributes.rs

@@ -103,10 +103,32 @@ impl Game<'_> {
         }
     }
 
+    /// What a mob's synced attribute instances are read from (`mob_attributes`).
+    pub(super) fn attr_sources(&self) -> AttrSources<'_> {
+        AttrSources {
+            catalog: &self.motion.attributes,
+            items: self.items(),
+            commands: &self.cmds.attributes,
+        }
+    }
+}
+
+/// What the synced attribute instances are read from: the attribute catalog, the item table and
+/// what `/attribute` changed. Plain tables, so the tick's helper threads can read them
+/// (`Game::lan_entities`).
+#[derive(Clone, Copy)]
+pub(super) struct AttrSources<'a> {
+    catalog: &'a horizon_world::attributes::entity::Catalog,
+    items: &'a horizon_client::items::Items,
+    commands:
+        &'a std::collections::HashMap<Target, horizon_world::attributes::entity::EntityAttributes>,
+}
+
+impl AttrSources<'_> {
     /// The registry id of a client-syncable attribute the entity type has
     /// (`AttributeMap.getInstance` is null for the others), by name.
     fn synced_attribute(&self, entity: &str, name: &str) -> Option<i32> {
-        let catalog = &self.motion.attributes;
+        let catalog = self.catalog;
         let id = catalog.id(name)?;
         (catalog.attribute(id).syncable && catalog.default_base(entity, id).is_some())
             .then(|| id.registry_id())
@@ -114,7 +136,7 @@ impl Game<'_> {
 
     /// The type's base value of an attribute, as the instance starts.
     fn attribute_default(&self, entity: &str, name: &str) -> f64 {
-        let catalog = &self.motion.attributes;
+        let catalog = self.catalog;
         catalog
             .id(name)
             .and_then(|id| catalog.default_base(entity, id))
@@ -132,7 +154,7 @@ impl Game<'_> {
         out: &mut Vec<AttributeSnapshot>,
     ) {
         for (slot, item) in slots {
-            let Some(info) = item.and_then(|i| self.items().get(i)) else {
+            let Some(info) = item.and_then(|i| self.items.get(i)) else {
                 continue;
             };
             for (attribute, id, amount, op, group) in &info.modifiers {
@@ -306,6 +328,16 @@ impl Game<'_> {
         }
     }
 
+    /// The instances `/attribute` created in an entity's generic map
+    /// replace the ones derived from its state.
+    fn generic_attributes_of(&self, t: Target, out: &mut Vec<AttributeSnapshot>) {
+        if let Some(map) = self.commands.get(&t) {
+            generic_attributes_into(map, out);
+        }
+    }
+}
+
+impl Game<'_> {
     /// An armor stand's synced instances: movement speed (its first
     /// `aiStep`), its equipment's modifiers and what `/attribute` changed.
     pub(super) fn stand_attributes(
@@ -315,18 +347,19 @@ impl Game<'_> {
     ) -> Vec<AttributeSnapshot> {
         const KIND: &str = "minecraft:armor_stand";
         let mut out = Vec::new();
-        if let Some(a) = self.synced_attribute(KIND, "minecraft:movement_speed") {
+        let sources = self.attr_sources();
+        if let Some(a) = sources.synced_attribute(KIND, "minecraft:movement_speed") {
             instance(
                 &mut out,
                 a,
-                self.attribute_default(KIND, "minecraft:movement_speed"),
+                sources.attribute_default(KIND, "minecraft:movement_speed"),
             );
         }
         let slots = horizon_client::entities::ArmorStand::SLOTS
             .into_iter()
             .zip(equipment.iter().map(|s| s.as_ref().map(|s| &s.item)));
-        self.equipment_attributes(KIND, |_| None, slots, &mut out);
-        self.generic_attributes_of(Target::Stand(id), &mut out);
+        sources.equipment_attributes(KIND, |_| None, slots, &mut out);
+        sources.generic_attributes_of(Target::Stand(id), &mut out);
         out
     }
 
@@ -372,14 +405,6 @@ impl Game<'_> {
             }
         }
     }
-
-    /// The instances `/attribute` created in an entity's generic map
-    /// replace the ones derived from its state.
-    fn generic_attributes_of(&self, t: Target, out: &mut Vec<AttributeSnapshot>) {
-        if let Some(map) = self.cmds.attributes.get(&t) {
-            generic_attributes_into(map, out);
-        }
-    }
 }
 
 #[cfg(test)]

crates/horizon-client/src/game/piston_push.rs

@@ -738,14 +738,11 @@ impl Game<'_> {
         });
         self.set_pushed_on_ground(target, limited, resolved);
 
-        // Entity.move updates ground before its tiny-collision return. This
-        // native slice covers only unoccupied server NoAI horizontal moves.
-        if let Target::Entity(EntityHandle::Mob(i)) = target {
+        // shortcut: unoccupied server mobs only, expand when controller/passenger authority is bound.
+        let mob = if let Target::Entity(EntityHandle::Mob(i)) = target {
             let m = &self.mobs[i];
-            let eligible = !self.options.execution_role.is_client()
+            (!self.options.execution_role.is_client()
                 && self.remote.is_none()
-                && limited.y == 0.0
-                && m.no_ai
                 && !m.extra.passenger
                 && m.extra.mount.is_none()
                 && m.rider.is_none()
@@ -757,16 +754,22 @@ impl Game<'_> {
                         .guests
                         .iter()
                         .any(|g| host.hub.here(g) && g.vehicle == Some(m.id))
-                });
-            if eligible {
-                self.mobs[i].on_ground = limited.y < 0.0 && limited.y != resolved.y;
-            }
+                }))
+            .then_some(i)
+        } else {
+            None
+        };
+        // Entity.move updates ground before its tiny-collision return.
+        if let Some(i) = mob
+            && self.mobs[i].no_ai
+        {
+            self.mobs[i].on_ground = limited.y < 0.0 && limited.y != resolved.y;
+        }
+        let tnt_after_move =
+            tnt.is_some() && !self.options.execution_role.is_client() && self.remote.is_none();
+        if mob.is_none() && !tnt_after_move {
+            self.restitute_piston(target, limited, resolved);
         }
-        // `Entity.move(MoverType.PISTON, …)` restitution, which runs whether or
-        // not the step is applied: so a primed TNT in a chunk that does not tick
-        // entities still loses its velocity into walls and ceilings and
-        // bounces on slime (#12022).
-        self.restitute_piston(target, limited, resolved);
         let moved = resolved.length_squared() > 1.0e-7
             || limited.length_squared() - resolved.length_squared() < 1.0e-7;
         if moved {
@@ -775,6 +778,12 @@ impl Game<'_> {
             // A move the collisions cut to nothing still ends in the check.
             self.release_after_block_move(target);
         }
+        // Restitution reads the new landing block even when the push was blocked.
+        if let Some(i) = mob {
+            self.mobs[i].restitute_movement(&self.world, &self.motion.blocks, limited, resolved);
+        } else if tnt_after_move {
+            self.restitute_piston(target, limited, resolved);
+        }
         if let Some((i, from, ..)) = tnt {
             self.finish_tnt_piston_move(i, from, resolved);
         }
@@ -1053,6 +1062,7 @@ mod tests {
 mod order_tests {
     use crate::game::{Audio, Game, Options, Progress, Settings, load_resources};
     use horizon_client::chunk::Chunk;
+    use horizon_core::geometry::Vec3;
     use horizon_core::{geometry::Direction, position::BlockPos};
     use horizon_world::level::pistons::MovingPiston;
     use std::sync::Arc;
@@ -1151,8 +1161,9 @@ mod order_tests {
         }
         // The near stone's box reaches x 5.5 into a TNT from x 5.01: 0.49 plus
         // PUSH_OFFSET; the far stone's old box stops it at x 6.
+        // Vanilla's TNT width is 0.98F, not the exact decimal 0.98.
         assert!(
-            (moved[0] - 0.5).abs() < 1e-9,
+            (moved[0] - 0.500000009536743).abs() < 1e-9,
             "far stone gone first: {}",
             moved[0]
         );
@@ -1251,7 +1262,7 @@ mod order_tests {
     }
 
     /// A primed TNT pushed down by a piston onto a slime block bounces up,
-    /// matching vanilla's slime restitution (~+0.64) even in a lazy chunk (#12022).
+    /// matching vanilla's restitution even in a lazy chunk (#12022).
     #[test]
     #[ignore = "needs prepared resources in HORIZON_TEST_RESOURCES"]
     fn a_pushed_tnt_bounces_on_slime() {
@@ -1312,32 +1323,36 @@ mod order_tests {
             death_ticks: 0,
             id: 1,
         };
-        game.block_ticks.moving_pistons = vec![(BlockPos::new(5, 200, 5), piston)];
-        game.sync_moving_pistons();
-        game.tnts.clear();
-        assert!(game.run_command("summon tnt 5.5 200.5 5.5 {fuse:80,Motion:[0.0d,-0.4d,0.0d]}"));
-        assert!(
-            (game.tnts[0].velocity.y - (-0.4)).abs() < 1.0e-9,
-            "TNT starts with vy -0.4, got {:?}",
-            game.tnts[0].velocity
-        );
-        // Piston advances from progress 0 to 0.5 and pushes the TNT down.
-        let pushes: Vec<_> = game
-            .block_ticks
-            .moving_pistons
-            .iter()
-            .map(|&(pos, old)| (pos, old, old.progress + 0.5))
-            .collect();
-        for (_, piston) in &mut game.block_ticks.moving_pistons {
-            piston.progress += 0.5;
+        // Pinned 26.3 PistonMovingBlockEntity.moveCollidedEntities, without entity ticks.
+        for (floor, y, vy, final_y, final_vy) in [
+            (slime, 200.5, -0.4, 200.0, 0.3510000138133766_f64),
+            (stone, 200.5, -0.4, 200.0, 0.0),
+            (slime, 200.0, -0.4, 200.0, 0.3200000047683716),
+            (slime, 200.00001, -0.4, 200.00001, 0.32000064476813395),
+            (slime, 200.5, 0.3, 200.0, 0.0),
+        ] {
+            game.world.set(5, 199, 5, floor).unwrap();
+            game.block_ticks.moving_pistons = vec![(BlockPos::new(5, 200, 5), piston)];
+            game.sync_moving_pistons();
+            game.tnts.clear();
+            assert!(game.run_command(&format!(
+                "summon tnt 5.5 {y} 5.5 {{fuse:80,Motion:[0.0d,{vy}d,0.0d]}}"
+            )));
+            assert_eq!(game.tnts[0].velocity.y, vy);
+            let pushes: Vec<_> = game
+                .block_ticks
+                .moving_pistons
+                .iter()
+                .map(|&(pos, old)| (pos, old, old.progress + 0.5))
+                .collect();
+            for (_, piston) in &mut game.block_ticks.moving_pistons {
+                piston.progress += 0.5;
+            }
+            game.push_piston_entities(pushes);
+            assert_eq!(game.tnts[0].position, Vec3::new(5.5, final_y, 5.5));
+            assert_eq!(game.tnts[0].velocity.y.to_bits(), final_vy.to_bits());
+            assert!(game.tnts[0].on_ground);
         }
-        game.push_piston_entities(pushes);
-        // The slime block at y=199 bounces the TNT up; vertical velocity must be positive.
-        assert!(
-            game.tnts[0].velocity.y > 0.0,
-            "TNT should bounce up on slime, got vy {:?}",
-            game.tnts[0].velocity
-        );
     }
 }
 
@@ -1602,6 +1617,54 @@ mod h255_mob_piston_ground_tests {
     }
     #[test]
     #[ignore = "needs prepared resources in HORIZON_TEST_RESOURCES"]
+    fn h255_added_tnt_authority_and_exhausted_budget() {
+        use crate::game::execution_role::ExecutionRole;
+        for role in [
+            ExecutionRole::Client,
+            ExecutionRole::IntegratedServer,
+            ExecutionRole::DedicatedServer,
+        ] {
+            with_role_game(role, |game| {
+                assert!(game.run_command("setblock 5 199 5 slime_block"));
+                assert!(
+                    game.run_command("summon tnt 5.5 200.5 5.5 {fuse:80,Motion:[0d,-0.4d,0d]}")
+                );
+                let mut budget = std::collections::HashMap::new();
+                let push = |game: &mut Game<'_>,
+                            budget: &mut std::collections::HashMap<_, _>,
+                            movement| {
+                    let tnt = &game.tnts[0];
+                    game.move_pushed(
+                        Target::Entity(EntityHandle::Tnt(0)),
+                        tnt.bounds(),
+                        [tnt.position.x, tnt.position.z],
+                        0.0,
+                        tnt.on_ground,
+                        movement,
+                        Direction::Down,
+                        budget,
+                    );
+                };
+                push(game, &mut budget, Vec3::ZERO);
+                assert_eq!(game.tnts[0].position.y, 200.5);
+                assert_eq!(game.tnts[0].velocity.y, -0.4);
+                push(game, &mut budget, Vec3::new(0.0, -0.51, 0.0));
+                let expected = if role.is_client() {
+                    0.0_f64
+                } else {
+                    0.3510000138133766
+                };
+                assert_eq!(game.tnts[0].velocity.y.to_bits(), expected.to_bits());
+                game.tnts[0].position.y = 200.5;
+                game.tnts[0].velocity.y = -0.4;
+                push(game, &mut budget, Vec3::new(0.0, -0.51, 0.0));
+                assert_eq!(game.tnts[0].position.y, 200.5);
+                assert_eq!(game.tnts[0].velocity.y, -0.4);
+            });
+        }
+    }
+    #[test]
+    #[ignore = "needs prepared resources in HORIZON_TEST_RESOURCES"]
     fn h255_added_no_request_and_exhausted_budget_keep_state() {
         with_game(|game| {
             let i = added_mob(game, 20, "horse", true, 8.5);
@@ -1613,6 +1676,16 @@ mod h255_mob_piston_ground_tests {
             added_push(game, i, Vec3::new(0.51, 0.0, 0.0));
             assert!(game.mobs[i].on_ground, "occupied path remains excluded");
             assert_eq!(game.mobs[i].position, Vec3::new(9.01, 202.0, 4.5));
+            assert!(game.run_command("setblock 9 201 4 slime_block"));
+            game.mobs[i].position.y = 202.25;
+            game.mobs[i].velocity = Vec3::new(0.0, -1.0, 0.0);
+            game.mobs[i].sync_position = false;
+            added_push(game, i, Vec3::new(0.0, -0.5, 0.0));
+            assert_eq!(
+                game.mobs[i].velocity.y, 0.0,
+                "occupied restitution stays outside this slice"
+            );
+            assert!(!game.mobs[i].sync_position);
             assert!(game.run_command("ride @s dismount"));
             assert!(game.riding.is_none());
             assert!(game.mobs[i].rider.is_none());
@@ -1655,23 +1728,46 @@ mod h255_mob_piston_ground_tests {
     }
     #[test]
     #[ignore = "needs prepared resources in HORIZON_TEST_RESOURCES"]
-    fn h255_added_client_role_ai_and_vertical_paths_keep_native_state() {
+    fn h255_added_client_role_ai_and_vertical_ground_state() {
         use crate::game::execution_role::ExecutionRole as Role;
-        for (role, no_ai, movement) in [
-            (Role::Client, true, Vec3::new(0.25, 0.0, 0.0)),
-            (Role::IntegratedServer, false, Vec3::new(0.25, 0.0, 0.0)),
-            (Role::IntegratedServer, true, Vec3::new(0.0, 0.25, 0.0)),
+        for (role, no_ai, movement, on_ground) in [
+            (Role::Client, true, Vec3::new(0.25, 0.0, 0.0), true),
+            (
+                Role::IntegratedServer,
+                false,
+                Vec3::new(0.25, 0.0, 0.0),
+                true,
+            ),
+            (
+                Role::IntegratedServer,
+                true,
+                Vec3::new(0.0, 0.25, 0.0),
+                false,
+            ),
         ] {
             with_role_game(role, |game| {
                 let i = added_mob(game, 22, "husk", no_ai, 8.5);
                 let before = game.mobs[i].position;
                 added_push(game, i, movement);
                 assert_eq!(game.mobs[i].position, before.add_vector(movement));
-                assert!(
-                    game.mobs[i].on_ground,
-                    "excluded native path remains unchanged; not Java parity"
-                );
+                assert_eq!(game.mobs[i].on_ground, on_ground);
             });
         }
     }
+
+    #[test]
+    #[ignore = "needs prepared resources in HORIZON_TEST_RESOURCES"]
+    fn h255_added_partial_push_reads_slime_after_the_position_changes() {
+        with_game(|game| {
+            assert!(game.run_command("setblock 8 201 4 slime_block"));
+            let i = added_mob(game, 23, "iron_golem", true, 8.5);
+            game.mobs[i].position.y = 202.25;
+            game.mobs[i].velocity = Vec3::new(0.0, -1.0, 0.0);
+            game.mobs[i].sync_position = false;
+            added_push(game, i, Vec3::new(0.0, -0.5, 0.0));
+            assert_eq!(game.mobs[i].position, Vec3::new(8.5, 202.0, 4.5));
+            assert!(game.mobs[i].velocity.y > 0.9);
+            assert!(game.mobs[i].sync_position);
+        });
+    }
 }

crates/horizon-client/src/lan/data.rs

@@ -4,6 +4,7 @@
 //! (`data/minecraft/...`) and the registries report.
 use crate::chunk::LevelShapes;
 use horizon_protocol::{
+    entity_data,
     packets::configuration::{RegistryData, RegistryEntry, RegistryTags, TagEntry, UpdateTags},
     registry::{Builtin, Dynamic},
 };
@@ -506,6 +507,58 @@ impl SyncData {
             .any(|(_, e)| e.from_pack.contains(&true))
     }
 
+    /// `LivingEntity.updateSynchronizedMobEffectParticles`: DATA_EFFECT_PARTICLES
+    /// (10) and DATA_EFFECT_AMBIENCE_ID (11) for the visible `effects`, as
+    /// (effect, ambient). The entries are left out when there is nothing to
+    /// show, which is their default.
+    pub fn effect_meta<'a>(
+        &self,
+        effects: impl IntoIterator<Item = (&'a str, bool)>,
+    ) -> Vec<(u8, i32, entity_data::Value)> {
+        let mut particles = Vec::new();
+        let mut all_ambient = true;
+        for (effect, ambient) in effects {
+            let effect = effect.trim_start_matches("minecraft:");
+            // MobEffect.particleFactory: the omens have their own particles,
+            // the others an entity effect in their colour (faint if ambient).
+            let (name, data) = match effect {
+                "trial_omen" | "raid_omen" => (effect, None),
+                _ => {
+                    let Some(def) = crate::brewing::effect(effect) else {
+                        continue;
+                    };
+                    let alpha = if ambient { 38 } else { 255 };
+                    let argb = (alpha << 24) | def.shown_color() as i32;
+                    (
+                        "entity_effect",
+                        Some(entity_data::ParticleData::Color(argb)),
+                    )
+                }
+            };
+            let Some(id) = self.id("minecraft:particle_type", &format!("minecraft:{name}")) else {
+                continue;
+            };
+            particles.push(entity_data::Particle { id, data });
+            all_ambient &= ambient;
+        }
+        if particles.is_empty() {
+            return Vec::new();
+        }
+        let mut meta = vec![(
+            10,
+            entity_data::serializer::PARTICLES,
+            entity_data::Value::Particles(particles),
+        )];
+        if all_ambient {
+            meta.push((
+                11,
+                entity_data::serializer::BOOLEAN,
+                entity_data::Value::Bool(true),
+            ));
+        }
+        meta
+    }
+
     /// The id of a name in a built-in or synchronized registry.
     pub fn id(&self, registry: &str, name: &str) -> Option<i32> {
         if registry == "minecraft:data_component_type"

crates/horizon-client/src/lan/hub.rs

@@ -24,7 +24,7 @@ use horizon_core::{
 };
 use horizon_protocol::ids::play::clientbound as ids;
 use horizon_protocol::{
-    entity_data::{Particle, ParticleData, Value, serializer},
+    entity_data::{Value, serializer},
     packets::play::{self, Clientbound as C, Serverbound as S},
     text::{literal, translatable},
 };
@@ -3250,39 +3250,12 @@ impl Hub {
         &self,
         effects: impl IntoIterator<Item = (&'a str, bool)>,
     ) -> Vec<(u8, i32, Value)> {
-        let mut particles = Vec::new();
-        let mut all_ambient = true;
-        for (effect, ambient) in effects {
-            let effect = effect.trim_start_matches("minecraft:");
-            // MobEffect.particleFactory: the omens have their own particles,
-            // the others an entity effect in their colour (faint if ambient).
-            let (name, data) = match effect {
-                "trial_omen" | "raid_omen" => (effect, None),
-                _ => {
-                    let Some(def) = crate::brewing::effect(effect) else {
-                        continue;
-                    };
-                    let alpha = if ambient { 38 } else { 255 };
-                    let argb = (alpha << 24) | def.shown_color() as i32;
-                    ("entity_effect", Some(ParticleData::Color(argb)))
-                }
-            };
-            let Some(id) =
-                self.registry_id("minecraft:particle_type", &format!("minecraft:{name}"))
-            else {
-                continue;
-            };
-            particles.push(Particle { id, data });
-            all_ambient &= ambient;
-        }
-        if particles.is_empty() {
-            return Vec::new();
-        }
-        let mut meta = vec![(10, serializer::PARTICLES, Value::Particles(particles))];
-        if all_ambient {
-            meta.push((11, serializer::BOOLEAN, Value::Bool(true)));
-        }
-        meta
+        self.sync.effect_meta(effects)
+    }
+
+    /// The registries and tables the clients are synced with (read by helper threads).
+    pub fn sync_data(&self) -> &SyncData {
+        &self.sync
     }
 
     fn player_state(

crates/horizon-client/src/mobs.rs

@@ -5,6 +5,10 @@
 //! fly and amphibious node evaluators (`path`, `nav`), move/look/jump/body
 //! controls (`control`), `LivingEntity` physics (`living`) and per-species
 //! behaviour, models, animations and sounds.
+mod mob_slot;
+mod pending_pushes;
+pub use mob_slot::MobSlot;
+pub use pending_pushes::{PendingPushes, apply_mob_pushes};
 pub mod ai;
 pub mod allay;
 pub mod ambient;
@@ -2923,28 +2927,6 @@ pub fn take_mob_pushes(events: &mut Vec<MobEvent>, from: usize, pending: &mut Pe
     }
     events.truncate(kept);
 }
-/// Pushes waiting for their mob's tick, in push order. A list while short;
-/// past [`PendingPushes::LIST_MAX`] (a crowd pushing each other) one list
-/// per mob, as scanning the whole list for each mob grows with its square.
-#[derive(Default)]
-pub struct PendingPushes {
-    list: Vec<(u32, Vec3)>,
-    by_mob: crate::fast_hash::FxHashMap<u32, Vec<Vec3>>,
-}
-impl PendingPushes {
-    const LIST_MAX: usize = 64;
-    pub fn push(&mut self, id: u32, v: Vec3) {
-        if self.by_mob.is_empty() && self.list.len() < Self::LIST_MAX {
-            self.list.push((id, v));
-            return;
-        }
-        // Older pushes first, so each mob's stay in order.
-        for (id, v) in self.list.drain(..) {
-            self.by_mob.entry(id).or_default().push(v);
-        }
-        self.by_mob.entry(id).or_default().push(v);
-    }
-}
 /// `ServerLevel.sendBlockUpdated`: for each write that changed a block's collision
 /// shape, a mob whose path runs near it re-paths (`PathNavigation.shouldRecomputePath`).
 /// The writes whose shape changed are found once per tick ([`collision_changes`]).
@@ -2972,50 +2954,28 @@ pub fn collision_changes(w: &MobWorld, writes: &[((i32, i32, i32), u32, u32)]) -
         })
         .collect()
 }
-/// Adds the pending pushes aimed at `m` to its velocity, in push order.
-pub fn apply_mob_pushes(m: &mut Mob, pending: &mut PendingPushes) {
-    // Entity.push(x, y, z) sets needsSync.
-    let mut add = |v: Vec3| {
-        m.velocity = Vec3::new(m.velocity.x + v.x, m.velocity.y + v.y, m.velocity.z + v.z);
-        m.needs_sync = true;
-    };
-    if !pending.list.is_empty() {
-        pending.list.retain(|(id, v)| {
-            if *id != m.id {
-                return true;
-            }
-            add(*v);
-            false
-        });
-    }
-    if !pending.by_mob.is_empty()
-        && let Some(pushes) = pending.by_mob.remove(&m.id)
-    {
-        pushes.into_iter().for_each(add);
-    }
-}
 /// Applies the brain memories a mob's tick set on other mobs (from event
 /// index `from` on), at once: vanilla sets them on the other entity during
 /// the tick, before the mobs after it tick.
-pub fn apply_brain_events(mobs: &mut [Mob], events: &mut Vec<MobEvent>, from: usize) {
+pub fn apply_brain_events<M: MobSlot>(mobs: &mut [M], events: &mut Vec<MobEvent>, from: usize) {
     let mut i = from;
     while i < events.len() {
         match events[i] {
             MobEvent::BreedTarget { mob, partner } => {
-                if let Some(m) = mobs.iter_mut().find(|m| m.id == mob) {
+                if let Some(m) = mobs.iter_mut().map(MobSlot::mob_mut).find(|m| m.id == mob) {
                     animal_brain::set_breed_partner(m, partner);
                 }
                 events.remove(i);
             }
             MobEvent::LockGaze { mob, partner } => {
-                if let Some(m) = mobs.iter_mut().find(|m| m.id == mob) {
+                if let Some(m) = mobs.iter_mut().map(MobSlot::mob_mut).find(|m| m.id == mob) {
                     animal_brain::lock_gaze(m, partner);
                     villager_brain::lock_gaze_half(m, Who::Mob(partner));
                 }
                 events.remove(i);
             }
             MobEvent::Retreat { mob, from, at } => {
-                if let Some(m) = mobs.iter_mut().find(|m| m.id == mob) {
+                if let Some(m) = mobs.iter_mut().map(MobSlot::mob_mut).find(|m| m.id == mob) {
                     piglin::receive_retreat(m, from, at);
                 }
                 events.remove(i);
@@ -3029,7 +2989,11 @@ pub fn apply_brain_events(mobs: &mut [Mob], events: &mut Vec<MobEvent>, from: us
                 events.remove(i);
             }
             MobEvent::SchoolSize { leader, by } => {
-                if let Some(m) = mobs.iter_mut().find(|m| m.id == leader) {
+                if let Some(m) = mobs
+                    .iter_mut()
+                    .map(MobSlot::mob_mut)
+                    .find(|m| m.id == leader)
+                {
                     m.extra.followers += by;
                 }
                 events.remove(i);
@@ -3039,7 +3003,11 @@ pub fn apply_brain_events(mobs: &mut [Mob], events: &mut Vec<MobEvent>, from: us
             MobEvent::Breed {
                 parent, partner, ..
             } => {
-                if let Some(m) = mobs.iter_mut().find(|m| m.id == partner) {
+                if let Some(m) = mobs
+                    .iter_mut()
+                    .map(MobSlot::mob_mut)
+                    .find(|m| m.id == partner)
+                {
                     m.extra.age = 6000;
                     m.extra.in_love = 0;
                     animal_brain::forget_breed_target(m, parent);
@@ -3061,6 +3029,7 @@ pub fn apply_brain_events(mobs: &mut [Mob], events: &mut Vec<MobEvent>, from: us
                 let mut raised = Vec::new();
                 let (aim, fish_pos, fish_silent) = mobs
                     .iter()
+                    .map(MobSlot::mob)
                     .find(|m| m.id == mob)
                     .map(|m| {
                         (
@@ -3075,7 +3044,10 @@ pub fn apply_brain_events(mobs: &mut [Mob], events: &mut Vec<MobEvent>, from: us
                     })
                     .map(|(aim, pos, silent)| (Some(aim), Some(pos), silent))
                     .unwrap_or((None, None, true));
-                if let Some(t) = mobs.iter_mut().find(|m| m.id == target)
+                if let Some(t) = mobs
+                    .iter_mut()
+                    .map(MobSlot::mob_mut)
+                    .find(|m| m.id == target)
                     && t.damage_aimed(amount, source, aim)
                 {
                     t.add_effect("minecraft:poison", poison_ticks, 0);
@@ -3290,6 +3262,9 @@ pub struct MobWorld<'a> {
     /// not in the level (a warden's anger, `AngerManagement.angerByUuid`).
     /// None: this world keeps no entity UUIDs, so nothing is resolved.
     pub who_lookup: Option<&'a dyn WhoLookup>,
+    /// Another thread's mobs, for a lookup by id that misses `others`: a group of the server's
+    /// parallel mob loop (`game/mob_islands.rs`) answers as the serial loop would.
+    pub foreign: Option<&'a dyn ForeignSeen>,
     /// The level's world border, whose walls stop a mob's moves
     /// (`Entity.collectColliders`); None: nothing stops them there.
     pub border: Option<&'a horizon_core::geometry::border::BorderGeometry>,
@@ -3304,6 +3279,10 @@ pub trait ModEffectTicks {
 /// names while it is in the level, never some other entity as a stand-in. Only
 /// the game layer keeps entity UUIDs, so it answers this (`Game::who_by_uuid`)
 /// and the mob layer reaches it through `MobWorld`.
+/// A mob outside this thread's part of the mob loop, as the serial loop would see it now.
+pub trait ForeignSeen {
+    fn seen(&self, id: u32) -> Option<&Seen>;
+}
 pub trait WhoLookup {
     fn who_by_uuid(&self, uuid: [i32; 4]) -> Option<Who>;
 }

crates/horizon-client/src/mobs/aquatic.rs

@@ -749,18 +749,35 @@ fn can_be_followed(s: Species, followers: i32) -> bool {
 }
 /// `startFollowing` on a recruit, applied after the recruiter's tick. A fish
 /// another leader took this tick (its snapshot was older) stays with it.
-pub fn start_following(mobs: &mut [Mob], fish: u32, leader: u32) {
-    let alive = |mobs: &[Mob], id: u32| mobs.iter().any(|m| m.id == id && m.alive() && !m.removed);
-    let Some(current) = mobs.iter().find(|m| m.id == fish).map(|m| m.extra.leader) else {
+pub fn start_following<M: super::MobSlot>(mobs: &mut [M], fish: u32, leader: u32) {
+    let alive = |mobs: &[M], id: u32| {
+        mobs.iter()
+            .map(super::MobSlot::mob)
+            .any(|m| m.id == id && m.alive() && !m.removed)
+    };
+    let Some(current) = mobs
+        .iter()
+        .map(super::MobSlot::mob)
+        .find(|m| m.id == fish)
+        .map(|m| m.extra.leader)
+    else {
         return;
     };
     if current.is_some_and(|l| alive(mobs, l)) {
         return;
     }
-    if let Some(m) = mobs.iter_mut().find(|m| m.id == fish) {
+    if let Some(m) = mobs
+        .iter_mut()
+        .map(super::MobSlot::mob_mut)
+        .find(|m| m.id == fish)
+    {
         m.extra.leader = Some(leader);
     }
-    if let Some(m) = mobs.iter_mut().find(|m| m.id == leader) {
+    if let Some(m) = mobs
+        .iter_mut()
+        .map(super::MobSlot::mob_mut)
+        .find(|m| m.id == leader)
+    {
         m.extra.followers += 1;
     }
 }

crates/horizon-client/src/mobs/goals.rs

@@ -21,7 +21,7 @@ pub struct Cx<'a, 'w> {
     pub lr: &'a mut LegacyRandom,
     pub events: &'a mut Vec<MobEvent>,
 }
-pub trait Goal: std::fmt::Debug {
+pub trait Goal: std::fmt::Debug + Send + Sync {
     fn flags(&self) -> u8;
     /// A goal that draws its first start delay in its constructor (a schooling fish's
     /// `FollowFlockLeaderGoal`): the delay again, when the mob's random is seeded afterwards.

crates/horizon-client/src/mobs/living.rs

@@ -849,7 +849,6 @@ impl Mob {
         self.horizontal_collision = x_hit || z_hit;
         // A server's mobs are authoritative (`isLocalInstanceAuthoritative`):
         // the vertical flags update even without vertical movement.
-        let moved_vertically = wanted.y.abs() > 0.0;
         let vertical = wanted.y != moved.y;
         self.vertical_collision_below = vertical && wanted.y < 0.0;
         self.on_ground = self.vertical_collision_below;
@@ -857,42 +856,51 @@ impl Mob {
         // checkFallDamage reads the landing block (getOnPosLegacy).
         self.supporting = self.supporting_block(w, moved);
         self.species_check_fall_damage(w, moved, events);
-        // Entity.restituteMovementAfterCollisions: the colliding components
-        // bounce back by the entity's bounciness or the block's (a slime
-        // block, a bed), whichever is larger. A mob never sneaks, so nothing
-        // suppresses its bounce there.
-        if moved_vertically && vertical || self.horizontal_collision {
-            // Entity.getOnPos(0.2F): the block whose bounce this move sees.
+        self.restitute_movement(w.world, w.movement, wanted, moved);
+        self.movement_emission(w, moved, events);
+        // Entity.move's last step: soul sand, honey and the like slow the
+        // horizontal speed of whatever the mob stands in or on.
+        let factor = self.block_speed_factor(w);
+        self.velocity = horizon_world::entity_move::scale_by_block_speed(self.velocity, factor);
+    }
+    /// `Entity.restituteMovementAfterCollisions`, shared by travel and piston pushes.
+    pub fn restitute_movement(
+        &mut self,
+        world: &crate::world::World,
+        movement: &horizon_world::movement_properties::BoundMovementProperties,
+        wanted: Vec3,
+        moved: Vec3,
+    ) {
+        let differs = |a: f64, b: f64| (b - a).abs() >= f64::from(1.0e-5_f32);
+        let x_hit = differs(wanted.x, moved.x);
+        let z_hit = differs(wanted.z, moved.z);
+        let vertical = wanted.y != moved.y;
+        if wanted.y.abs() > 0.0 && vertical || x_hit || z_hit {
             let on = (
                 self.position.x.floor() as i32,
                 (self.position.y - f64::from(0.2_f32)).floor() as i32,
                 self.position.z.floor() as i32,
             );
             let input = RestitutionInput {
-                block: w.world.block(on.0, on.1, on.2),
+                block: world.block(on.0, on.1, on.2),
                 collided_x: x_hit,
                 collided_z: z_hit,
                 vertical_collision: vertical,
-                collision_below: self.vertical_collision_below,
+                collision_below: vertical && wanted.y < 0.0,
                 living: true,
                 resolved_movement: moved,
                 fall_distance: self.fall_distance,
             };
-            // The port's phases cannot fail for a mob: a block outside the
-            // registry reads as one with no bounce, never as an error.
+            // Unknown block states have no bounce; the provider cannot fail.
             let _ = restitute_movement(
                 &mut MobRestitution {
                     mob: self,
-                    world: w,
+                    world,
+                    movement,
                 },
                 input,
             );
         }
-        self.movement_emission(w, moved, events);
-        // Entity.move's last step: soul sand, honey and the like slow the
-        // horizontal speed of whatever the mob stands in or on.
-        let factor = self.block_speed_factor(w);
-        self.velocity = horizon_world::entity_move::scale_by_block_speed(self.velocity, factor);
     }
     /// `Entity.getBlockSpeedFactor` at the mob's feet.
     fn block_speed_factor(&self, w: &MobWorld) -> f32 {
@@ -2103,20 +2111,21 @@ pub fn client_water_splash(mob: &mut Mob, out: &mut Vec<MobEvent>) {
 /// mob that never sneaks. The phases themselves are
 /// `movement_restitution::restitute_movement`, the same port the player and
 /// dropped items use.
-struct MobRestitution<'a, 'w> {
+struct MobRestitution<'a> {
     mob: &'a mut Mob,
-    world: &'a MobWorld<'w>,
+    world: &'a crate::world::World,
+    movement: &'a horizon_world::movement_properties::BoundMovementProperties,
 }
-impl<'a, 'w> MobRestitution<'a, 'w> {
+impl<'a> MobRestitution<'a> {
     /// The block's movement data: its coefficients and its bounce tag.
     fn movement(
         &self,
         state: BlockStateId,
     ) -> Result<horizon_world::movement_properties::MovementBlock, ()> {
-        self.world.movement.get(state).map_err(|_| ())
+        self.movement.get(state).map_err(|_| ())
     }
 }
-impl RestitutionWorld for MobRestitution<'_, '_> {
+impl RestitutionWorld for MobRestitution<'_> {
     // The block data cannot fail here: a state the registry does not know
     // reads as a block with no bounce, never as an error.
     type Error = ();
@@ -2162,7 +2171,7 @@ impl RestitutionWorld for MobRestitution<'_, '_> {
                 position.z.floor() as i32,
             ));
         };
-        let state = self.world.world.block(support[0], support[1], support[2]);
+        let state = self.world.block(support[0], support[1], support[2]);
         let keeps_column = self
             .movement(state)
             .is_ok_and(|b| b.wall || b.coefficients.fence_gate);

crates/horizon-client/src/mobs/living_tests.rs

@@ -174,6 +174,7 @@ fn a_mob_bounces_off_a_bouncy_block_and_not_off_stone() {
             dragon: None,
             leash_boats: &[],
             who_lookup: None,
+            foreign: None,
             others_grid: None,
             others_index: None,
             others_villagers: None,
@@ -291,6 +292,7 @@ fn a_ridden_horse_floats_in_deep_water() {
         dragon: None,
         leash_boats: &[],
         who_lookup: None,
+        foreign: None,
         others_grid: None,
         others_index: None,
         others_villagers: None,
@@ -411,6 +413,7 @@ fn a_falling_mob_lands_on_a_shulker_or_a_held_happy_ghast() {
             nearest_player: None,
             others: &others,
             who_lookup: None,
+            foreign: None,
             dragon: None,
             leash_boats: &[],
             others_grid: None,

crates/horizon-client/src/mobs/mob_slot.rs

@@ -0,0 +1,29 @@
+//! [`MobSlot`]: a mob a list hands out.
+use super::Mob;
+
+/// A mob a list hands out: the list's own, or a borrow of one in a larger list (a group of the
+/// server's parallel mob loop, `game/mob_islands.rs`, works on its mobs in place).
+pub trait MobSlot {
+    fn mob(&self) -> &Mob;
+    fn mob_mut(&mut self) -> &mut Mob;
+}
+impl MobSlot for Mob {
+    #[inline]
+    fn mob(&self) -> &Mob {
+        self
+    }
+    #[inline]
+    fn mob_mut(&mut self) -> &mut Mob {
+        self
+    }
+}
+impl MobSlot for &mut Mob {
+    #[inline]
+    fn mob(&self) -> &Mob {
+        self
+    }
+    #[inline]
+    fn mob_mut(&mut self) -> &mut Mob {
+        self
+    }
+}

crates/horizon-client/src/mobs/pending_pushes.rs

@@ -0,0 +1,103 @@
+//! The pushes a mob's tick gives others, waiting for each one's own tick.
+use super::Mob;
+use horizon_core::geometry::Vec3;
+
+/// Pushes waiting for their mob's tick, in push order. A list while short;
+/// past [`PendingPushes::LIST_MAX`] (a crowd pushing each other) one list
+/// per mob, as scanning the whole list for each mob grows with its square.
+#[derive(Default)]
+pub struct PendingPushes {
+    list: Vec<(u32, Vec3)>,
+    by_mob: crate::fast_hash::FxHashMap<u32, Vec<Vec3>>,
+}
+impl PendingPushes {
+    const LIST_MAX: usize = 64;
+    /// `other`'s pushes after these, each mob's in their order.
+    pub fn append(&mut self, other: PendingPushes) {
+        for (id, v) in other.list {
+            self.push(id, v);
+        }
+        for (id, list) in other.by_mob {
+            for v in list {
+                self.push(id, v);
+            }
+        }
+    }
+    pub fn push(&mut self, id: u32, v: Vec3) {
+        if self.by_mob.is_empty() && self.list.len() < Self::LIST_MAX {
+            self.list.push((id, v));
+            return;
+        }
+        // Older pushes first, so each mob's stay in order.
+        for (id, v) in self.list.drain(..) {
+            self.by_mob.entry(id).or_default().push(v);
+        }
+        self.by_mob.entry(id).or_default().push(v);
+    }
+}
+/// Adds the pending pushes aimed at `m` to its velocity, in push order.
+pub fn apply_mob_pushes(m: &mut Mob, pending: &mut PendingPushes) {
+    // Entity.push(x, y, z) sets needsSync.
+    let mut add = |v: Vec3| {
+        m.velocity = Vec3::new(m.velocity.x + v.x, m.velocity.y + v.y, m.velocity.z + v.z);
+        m.needs_sync = true;
+    };
+    if !pending.list.is_empty() {
+        pending.list.retain(|(id, v)| {
+            if *id != m.id {
+                return true;
+            }
+            add(*v);
+            false
+        });
+    }
+    if !pending.by_mob.is_empty()
+        && let Some(pushes) = pending.by_mob.remove(&m.id)
+    {
+        pushes.into_iter().for_each(add);
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::PendingPushes;
+    use horizon_core::geometry::Vec3;
+
+    /// Each mob's pushes, in the order they apply.
+    fn per_mob(p: &PendingPushes) -> std::collections::BTreeMap<u32, Vec<[f64; 3]>> {
+        let mut out = std::collections::BTreeMap::<u32, Vec<[f64; 3]>>::new();
+        for (id, v) in &p.list {
+            out.entry(*id).or_default().push([v.x, v.y, v.z]);
+        }
+        for (id, list) in &p.by_mob {
+            out.entry(*id)
+                .or_default()
+                .extend(list.iter().map(|v| [v.x, v.y, v.z]));
+        }
+        out
+    }
+
+    /// Two lists of pushes, the second appended to the first, apply as one list of them all in
+    /// that order, on either side of the switch from one list to a list per mob.
+    #[test]
+    fn appended_pushes_apply_in_order() {
+        let pushes: Vec<(u32, Vec3)> = (0..200u32)
+            .map(|k| (k * 7 % 11, Vec3::new(f64::from(k), 0.0, -f64::from(k))))
+            .collect();
+        let mut all = PendingPushes::default();
+        for &(id, v) in &pushes {
+            all.push(id, v);
+        }
+        for split in [0, 30, 64, 70, 150, 200] {
+            let (mut first, mut second) = (PendingPushes::default(), PendingPushes::default());
+            for &(id, v) in &pushes[..split] {
+                first.push(id, v);
+            }
+            for &(id, v) in &pushes[split..] {
+                second.push(id, v);
+            }
+            first.append(second);
+            assert_eq!(per_mob(&first), per_mob(&all), "split at {split}");
+        }
+    }
+}

crates/horizon-client/src/mobs/riders.rs

@@ -186,16 +186,23 @@ pub fn seat(attachment: [f64; 3], rider_offset: f64, at: Vec3, yaw: f32) -> Vec3
 
 /// `Entity.startRiding`: the link it makes, `rider` on `vehicle` as its first
 /// passenger (`addPassenger`).
-pub fn link(mobs: &mut [Mob], rider: u32, vehicle: u32) {
+pub fn link<M: super::MobSlot>(mobs: &mut [M], rider: u32, vehicle: u32) {
     if rider == vehicle {
         return;
     }
-    if let Some(m) = mobs.iter_mut().find(|m| m.id == rider) {
+    if let Some(m) = mobs
+        .iter_mut()
+        .map(super::MobSlot::mob_mut)
+        .find(|m| m.id == rider)
+    {
         super::shulker::on_start_riding(m);
         m.extra.mount = Some(vehicle);
         m.extra.passenger = true;
     }
-    if let Some(m) = mobs.iter_mut().find(|m| m.id == vehicle)
+    if let Some(m) = mobs
+        .iter_mut()
+        .map(super::MobSlot::mob_mut)
+        .find(|m| m.id == vehicle)
         && m.extra.rider.is_none()
     {
         m.extra.rider = Some(rider);
@@ -213,9 +220,10 @@ pub fn can_add_passenger(v: &Mob, aboard: usize) -> bool {
 }
 /// `Piglin.getTopPassenger`: down the chain of first passengers from
 /// `vehicle`, stopping at the `counter`th seat.
-fn top_passenger(mobs: &[Mob], vehicle: u32, counter: i32) -> u32 {
+fn top_passenger<M: super::MobSlot>(mobs: &[M], vehicle: u32, counter: i32) -> u32 {
     let first = mobs
         .iter()
+        .map(super::MobSlot::mob)
         .find(|m| m.id == vehicle)
         .and_then(|m| m.extra.rider);
     match first {
@@ -228,17 +236,22 @@ fn top_passenger(mobs: &[Mob], vehicle: u32, counter: i32) -> u32 {
 /// Not the vehicle it is on already, nor one it carries itself, and only
 /// after its boarding cooldown, onto a vehicle with no passenger
 /// (`canRide`, `canAddPassenger`). Returns whether it boarded.
-pub fn start_riding(mobs: &mut [Mob], rider: u32, vehicle: u32) -> bool {
+pub fn start_riding<M: super::MobSlot>(mobs: &mut [M], rider: u32, vehicle: u32) -> bool {
     // `Shulker.startRiding` puts its attach face back before the ride is
     // decided, so a shulker that is refused still ends up lid up.
-    if let Some(m) = mobs.iter_mut().find(|m| m.id == rider) {
+    if let Some(m) = mobs
+        .iter_mut()
+        .map(super::MobSlot::mob_mut)
+        .find(|m| m.id == rider)
+    {
         super::shulker::on_start_riding(m);
     }
-    let Some(r) = mobs.iter().find(|m| m.id == rider) else {
+    let Some(r) = mobs.iter().map(super::MobSlot::mob).find(|m| m.id == rider) else {
         return false;
     };
     let on_hoglin = mobs
         .iter()
+        .map(super::MobSlot::mob)
         .any(|m| m.id == vehicle && m.species() == Species::Hoglin);
     let vehicle = if r.species() == Species::Piglin && r.is_baby() && on_hoglin {
         top_passenger(mobs, vehicle, MAX_PASSENGERS_ON_ONE_HOGLIN)
@@ -252,17 +265,26 @@ pub fn start_riding(mobs: &mut [Mob], rider: u32, vehicle: u32) -> bool {
     // its vehicle.
     let mut up = Some(vehicle);
     while let Some(v) = up {
-        up = mobs.iter().find(|m| m.id == v).and_then(|m| m.extra.mount);
+        up = mobs
+            .iter()
+            .map(super::MobSlot::mob)
+            .find(|m| m.id == v)
+            .and_then(|m| m.extra.mount);
         if up == Some(rider) {
             return false;
         }
     }
-    let Some(v) = mobs.iter().find(|m| m.id == vehicle) else {
+    let Some(v) = mobs
+        .iter()
+        .map(super::MobSlot::mob)
+        .find(|m| m.id == vehicle)
+    else {
         return false;
     };
     let count = usize::from(v.rider.is_some())
         + mobs
             .iter()
+            .map(super::MobSlot::mob)
             .filter(|m| m.extra.mount == Some(vehicle))
             .count()
             .max(usize::from(v.extra.rider.is_some()));
@@ -470,25 +492,26 @@ pub fn solve_body_rot(head: f32, vehicle_body: f32) -> f32 {
 /// Tick a passenger with Mob.getNavigation's recursive vehicle redirection.
 /// Navigation is lent, not copied: goal start/stop/path progress all address
 /// the same object the vehicle ticks. The pathfinder sees the vehicle body.
-pub fn tick(
-    mobs: &mut [Mob],
+pub fn tick<M: super::MobSlot>(
+    mobs: &mut [M],
     index: usize,
     w: &MobWorld,
     random: &mut horizon_core::random::LegacyRandom,
     events: &mut Vec<super::MobEvent>,
 ) -> bool {
     let mut owner = index;
-    if !mobs[index].no_ai {
+    if !mobs[index].mob().no_ai {
         for _ in 0..mobs.len() {
             let Some(vehicle) = mobs[owner]
+                .mob()
                 .extra
                 .mount
-                .and_then(|id| mobs.iter().position(|m| m.id == id))
+                .and_then(|id| mobs.iter().position(|m| m.mob().id == id))
             else {
                 break;
             };
             if vehicle == index
-                || controller(&mobs[vehicle], w).is_none_or(|r| r.id != mobs[owner].id)
+                || controller(mobs[vehicle].mob(), w).is_none_or(|r| r.id != mobs[owner].mob().id)
             {
                 break;
             }
@@ -496,14 +519,14 @@ pub fn tick(
         }
     }
     if owner == index {
-        return mobs[index].tick(w, random, events);
+        return mobs[index].mob_mut().tick(w, random, events);
     }
     let (rider, vehicle) = if index < owner {
         let (left, right) = mobs.split_at_mut(owner);
-        (&mut left[index], &mut right[0])
+        (left[index].mob_mut(), right[0].mob_mut())
     } else {
         let (left, right) = mobs.split_at_mut(index);
-        (&mut right[0], &mut left[owner])
+        (right[0].mob_mut(), left[owner].mob_mut())
     };
     let body = vehicle.path_mob();
     std::mem::swap(&mut rider.ai.nav, &mut vehicle.ai.nav);

crates/horizon-client/src/mobs/seen_grid.rs

@@ -15,6 +15,9 @@ pub struct SeenGrid {
     /// The widest box (along x or z) the grid has held: no box reaches
     /// further than this past its minimum corner.
     reach: f64,
+    /// The entries a query counts when it decides to scan them all instead (at least the grid's
+    /// own): a grid of part of a list answers as the whole list's would.
+    entries: usize,
 }
 
 fn column(x: f64, z: f64) -> (i32, i32) {
@@ -39,6 +42,22 @@ impl SeenGrid {
         }
         grid
     }
+    /// The widest box and the entries it has, for a grid of part of its entries that must meet
+    /// the same entries in the same order (the server's parallel mob tick).
+    pub fn bounds(&self) -> (f64, usize) {
+        (self.reach, self.cell_of.len().max(self.entries))
+    }
+    /// [`Self::bounds`] of the grid of `boxes`, without building it.
+    pub fn bounds_of(boxes: impl Iterator<Item = Aabb>) -> (f64, usize) {
+        boxes.fold((0.0_f64, 0), |(reach, n), b| (reach.max(width(b)), n + 1))
+    }
+    /// Answer queries as a grid of `bounds` would: as wide a margin, and a scan past as many
+    /// entries.
+    pub fn answering_as(mut self, (reach, entries): (f64, usize)) -> Self {
+        self.reach = self.reach.max(reach);
+        self.entries = entries;
+        self
+    }
     /// Entry `i`'s box is now `b` (the mob moved or resized).
     pub fn update(&mut self, i: usize, b: Aabb) {
         self.reach = self.reach.max(width(b));
@@ -78,7 +97,7 @@ impl SeenGrid {
         let hi = column(max_x, max_z);
         let span = |a: i32, b: i32| i64::from(b) - i64::from(a) + 1;
         let (nx, nz) = (span(lo.0, hi.0), span(lo.1, hi.1));
-        let max_cells = (self.cell_of.len() as i64).max(64);
+        let max_cells = (self.cell_of.len().max(self.entries) as i64).max(64);
         if nx < 1 || nz < 1 || nx.saturating_mul(nz) > max_cells {
             return false;
         }
@@ -337,4 +356,46 @@ mod tests {
             rounds * n
         );
     }
+
+    /// A grid of part of a list's boxes that answers as the whole list's grid decides when to scan
+    /// every entry and when to look in columns: the part meets its boxes as the whole would.
+    #[test]
+    fn a_part_answers_as_the_whole_list() {
+        let mut seed = 0x9e37_79b9_7f4a_7c15u64;
+        let mut next = |range: f64| {
+            seed = seed.wrapping_mul(6_364_136_223_846_793_005).wrapping_add(1);
+            (seed >> 11) as f64 / (1u64 << 53) as f64 * range
+        };
+        let boxes: Vec<Aabb> = (0..400)
+            .map(|_| boxed(next(400.0) - 200.0, next(8.0), next(400.0) - 200.0, 0.6))
+            .collect();
+        let whole = SeenGrid::new(boxes.iter().copied());
+        assert_eq!(SeenGrid::bounds_of(boxes.iter().copied()), whole.bounds());
+        let part: Vec<usize> = (0..boxes.len()).step_by(4).collect();
+        let grid = SeenGrid::new(part.iter().map(|&i| boxes[i])).answering_as(whole.bounds());
+        let (mut a, mut b) = (Vec::new(), Vec::new());
+        let mut scanned = 0;
+        for _ in 0..500 {
+            let side = next(100.0);
+            let area = boxed(next(400.0) - 200.0, next(8.0), next(400.0) - 200.0, side);
+            let indexed = whole.candidates(area, &mut a);
+            assert_eq!(grid.candidates(area, &mut b), indexed, "{side} blocks");
+            scanned += usize::from(!indexed);
+            if indexed {
+                let met = |out: &[u32]| -> Vec<usize> {
+                    out.iter()
+                        .map(|&j| part[j as usize])
+                        .filter(|&i| boxes[i].intersects(area))
+                        .collect()
+                };
+                let expected: Vec<usize> = a
+                    .iter()
+                    .map(|&j| j as usize)
+                    .filter(|i| i % 4 == 0 && boxes[*i].intersects(area))
+                    .collect();
+                assert_eq!(met(&b), expected);
+            }
+        }
+        assert!(scanned > 0 && scanned < 500, "{scanned} of 500 scanned");
+    }
 }

crates/horizon-client/src/mobs/villager_brain.rs

@@ -928,12 +928,12 @@ fn chunk_pois(
         return;
     };
     let kind = |state: u32| poi_of_state(w, horizon_world::registry::BlockStateId(state));
-    let min = w.world.bottom_y();
-    for s in (y0 - min) >> 4..=(y1 - min) >> 4 {
-        let section = &chunk.sections[s as usize];
-        for &(i, poi) in w.world.pois.get((qx, s, qz), section, kind).iter() {
+    let (min, mut lists) = (w.world.bottom_y(), Vec::new());
+    w.world.pois.column(chunk, min, (y0, y1), kind, &mut lists);
+    for (base, pois) in &lists {
+        for &(i, poi) in pois.iter() {
             let i = i32::from(i);
-            let y = min + s * 16 + (i >> 8);
+            let y = base + (i >> 8);
             if (y0..=y1).contains(&y) {
                 visit(poi, [qx * 16 + (i & 15), y, qz * 16 + (i >> 4 & 15)]);
             }

crates/horizon-client/src/mobs/world.rs

@@ -604,13 +604,14 @@ impl MobWorld<'_> {
         }
     }
     pub fn seen(&self, id: u32) -> Option<&Seen> {
-        match self.others_index {
+        let own = match self.others_index {
             Some(index) => index
                 .get(&id)
                 .and_then(|&i| self.others.get(i as usize))
                 .filter(|s| s.id == id),
             None => self.others.iter().find(|s| s.id == id),
-        }
+        };
+        own.or_else(|| self.foreign?.seen(id))
     }
     /// Living mobs of the given species within `range` of `from`, nearest first.
     pub fn nearby<'s>(
@@ -773,6 +774,7 @@ pub(crate) fn with_resource_test_world(
         dragon: None,
         leash_boats: &[],
         who_lookup: None,
+        foreign: None,
         others_grid: None,
         others_index: None,
         others_villagers: None,

crates/horizon-client/src/mobs/zombie_villager.rs

@@ -261,6 +261,7 @@ mod tests {
             dragon: None,
             leash_boats: &[],
             who_lookup: None,
+            foreign: None,
             others_grid: None,
             others_index: None,
             others_villagers: None,

crates/horizon-client/src/multiplayer.rs

@@ -7,7 +7,6 @@ use crate::lan::discovery::Scanner;
 use crate::text::{DARK_GRAY, GRAY, Text, format_translation};
 use crate::title::confirm_top;
 use horizon_protocol::servers::{self, ServerEntry};
-pub use page::Form;
 use std::{
     collections::HashMap,
     path::{Path, PathBuf},
@@ -139,8 +138,6 @@ pub enum Screen {
         relay: String,
         focus: u8,
     },
-    /// A page a crate registered at the title screen's Friends icon, as a form.
-    Page(Form),
     Disconnected {
         reason: String,
         /// Auto-reconnect's next attempt, until it is cancelled.
@@ -149,21 +146,6 @@ pub enum Screen {
         lines: std::cell::Cell<usize>,
     },
 }
-mod page;
-
-/// What the title screen's Friends icon opens: the page a crate registered there
-/// (`Place::Title`), else Join a Friend when an invite code can be joined. None: it is inactive.
-pub fn title_entry() -> Option<Screen> {
-    match page::title_page() {
-        Some(page) => Some(Screen::Page(Form::open(page))),
-        None => crate::net::joins_friends().then(|| Screen::friend(true)),
-    }
-}
-/// Whether the title screen's Friends icon opens something ([`title_entry`]).
-pub fn has_title_entry() -> bool {
-    page::title_page().is_some() || crate::net::joins_friends()
-}
-
 /// What a click asks the title screen to do.
 pub enum Outcome {
     None,
@@ -194,7 +176,6 @@ impl Screen {
             Self::Direct { .. } => "direct_connect",
             Self::ConfirmDelete { .. } => "server_delete",
             Self::Friend { .. } => "friends_join",
-            Self::Page(form) => form.id(),
             Self::Disconnected { .. } => "disconnected",
         }
     }
@@ -763,7 +744,6 @@ impl Screen {
                     53,
                 ),
             ],
-            Self::Page(form) => form.buttons(w, h),
             Self::Friend { code, .. } => vec![
                 b(
                     half(w) - 100.0,
@@ -884,7 +864,6 @@ impl Screen {
                     ]),
                 )
             }
-            Self::Page(form) => form.narration(),
             Self::Friend { .. } => ScreenInfo::new(
                 "multiplayer/friend",
                 get("horizon.multiplayer.join_friend", "Join a Friend"),
@@ -1339,7 +1318,6 @@ impl Screen {
                     centered(batch, &line.plain(), top + 17.0 + i as f32 * 9.0, u32::MAX);
                 }
             }
-            Self::Page(form) => form.draw(atlas, batch, caret, focused_at),
             Self::Friend {
                 code, relay, focus, ..
             } => {
@@ -1405,7 +1383,6 @@ impl Screen {
     /// selected server or LAN game, as a click on it does.
     pub fn confirm(&self) -> Outcome {
         match self {
-            Self::Page(form) => form.confirm(),
             Self::Direct { address } if servers::parse_address(address).is_some() => {
                 Outcome::Connect {
                     address: direct_address(address),
@@ -1454,9 +1431,6 @@ impl Screen {
         saves: &Path,
     ) -> Outcome {
         let (w, h) = size;
-        if let Self::Page(form) = self {
-            return form.click(id, x, y);
-        }
         match (&mut *self, id) {
             (
                 Self::List {
@@ -1849,7 +1823,6 @@ impl Screen {
             Self::Friend { focus: 0, .. } => 64.0,
             Self::Friend { focus: 1, .. } => 105.0,
             Self::Friend { .. } => return None,
-            Self::Page(form) => return form.focused_box(),
             _ => return None,
         };
         Some([(w / 2.0).floor() - 100.0, y, 200.0, 20.0])
@@ -1878,13 +1851,6 @@ impl Screen {
             _ => None,
         }
     }
-    /// Whether a press off the buttons may land on one of its text fields.
-    pub fn has_fields(&self) -> bool {
-        matches!(
-            self,
-            Self::Edit { .. } | Self::Friend { .. } | Self::Page(_)
-        )
-    }
     /// Join a friend, with the relay the player set before.
     pub fn friend(from_title: bool) -> Self {
         Self::Friend {
@@ -1944,9 +1910,6 @@ impl Screen {
     }
     /// The focused text field, its id and maximum length (copy and paste).
     pub fn focused_text(&mut self) -> Option<(crate::clipboard::FieldId, &mut String, usize)> {
-        if let Self::Page(form) = self {
-            return form.focused_text();
-        }
         let id = match self {
             Self::Edit { focus: 0, .. } => SERVER_NAME,
             Self::Edit { .. } => SERVER_ADDRESS,
@@ -1971,9 +1934,6 @@ impl Screen {
     /// round to the first field. `button` is the focused button, if any; a field
     /// has the focus (`focus` below 2) only while no button does.
     pub fn tab_dir(&mut self, reverse: bool, buttons: &[u8], button: &mut Option<u8>) {
-        if let Self::Page(form) = self {
-            return form.tab_dir(reverse, buttons, button);
-        }
         let (Self::Edit { focus, .. } | Self::Friend { focus, .. }) = self else {
             return;
         };
@@ -2000,7 +1960,6 @@ impl Screen {
     pub fn escape(&mut self, saves: &Path) -> bool {
         match self {
             Self::List { .. }
-            | Self::Page(_)
             | Self::Friend {
                 from_title: true, ..
             } => true,

crates/horizon-client/src/multiplayer/page.rs

@@ -1,467 +0,0 @@
-//! A page a crate registers at `horizon_hooks::pages::Place::Title`, which the title screen's
-//! Friends icon opens: a form drawn as Join a Friend and `ManageServerScreen` are. The title at
-//! the top, then the rows from y 53 (a field's grey label, its 200-wide `EditBox` 11 below, the
-//! next 41 lower; text 10 a line), and the page's buttons in a column under them from a quarter
-//! of the height + 114, 24 apart, the one that leaves it last.
-//!
-//! The engine keeps each field's text while the form is open (the title screen's caret edits
-//! it, as every field of its own) and tells the page each change before anything else reaches
-//! it ([`Act::Edit`]), then Enter in a field ([`Act::Submit`]) and the presses. A page joins a
-//! server with `horizon_hooks::pages::connect`.
-use super::{ButtonSpec, Outcome, centered_x, half, quarter};
-use crate::hud::{HudAtlas, HudBatch, rgba};
-use horizon_hooks::pages::{self, Act, Page, Place, Row};
-use std::cell::RefCell;
-use std::time::Instant;
-
-/// The first button's id; button `k` of the page is `FIRST + k`.
-const FIRST: u8 = 100;
-/// The text a line of the form wraps at.
-const WIDTH: f32 = 300.0;
-
-/// The page the title screen's Friends icon opens: the first registered there.
-pub(super) fn title_page() -> Option<&'static dyn Page> {
-    pages::engine::all()
-        .iter()
-        .copied()
-        .find(|page| page.place() == Place::Title)
-}
-
-/// An open form: the page, and what the engine keeps while it is open.
-#[derive(Clone)]
-pub struct Form {
-    page: &'static dyn Page,
-    /// Each field's text, in the order the rows have them.
-    texts: Vec<String>,
-    /// The text the page last heard of each field.
-    sent: RefCell<Vec<String>>,
-    /// The focused field; `texts.len()` (or more) while a button has the focus.
-    pub(super) focus: usize,
-    /// Each field's box where the last draw put it.
-    boxes: RefCell<Vec<[f32; 4]>>,
-}
-
-impl std::fmt::Debug for Form {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        f.debug_struct("Form")
-            .field("page", &self.page.id())
-            .field("texts", &self.texts)
-            .field("focus", &self.focus)
-            .finish()
-    }
-}
-
-/// A button of the form: its row (None: the one that leaves), label key, and whether it acts.
-struct Button {
-    row: Option<usize>,
-    label: &'static str,
-    active: bool,
-    half: bool,
-}
-
-impl Form {
-    /// The form of `page`, its fields holding the text its rows start with, the first focused.
-    pub fn open(page: &'static dyn Page) -> Self {
-        let mut rows = Vec::new();
-        page.rows(&mut rows);
-        let texts: Vec<String> = rows
-            .iter()
-            .filter_map(|row| match row {
-                Row::Field { text, .. } => Some(text.clone()),
-                _ => None,
-            })
-            .collect();
-        Self {
-            page,
-            sent: RefCell::new(texts.clone()),
-            boxes: RefCell::new(vec![[0.0; 4]; texts.len()]),
-            texts,
-            focus: 0,
-        }
-    }
-
-    pub fn id(&self) -> &'static str {
-        self.page.id()
-    }
-
-    /// The page's rows, after it heard every change to its fields.
-    fn rows(&self) -> Vec<Row> {
-        let mut rows = Vec::new();
-        self.page.rows(&mut rows);
-        let fields = field_rows(&rows);
-        let mut sent = self.sent.borrow_mut();
-        let mut told = false;
-        for ((text, sent), row) in self.texts.iter().zip(sent.iter_mut()).zip(fields) {
-            if text != sent {
-                sent.clone_from(text);
-                self.page.act(row, Act::Edit(text.clone()));
-                told = true;
-            }
-        }
-        if told {
-            rows.clear();
-            self.page.rows(&mut rows);
-        }
-        rows
-    }
-
-    fn buttons_of(rows: &[Row], back: &'static str) -> Vec<Button> {
-        let mut buttons: Vec<Button> = rows
-            .iter()
-            .enumerate()
-            .filter_map(|(r, row)| match row {
-                Row::Button {
-                    label,
-                    active,
-                    half,
-                    ..
-                } => Some(Button {
-                    row: Some(r),
-                    label,
-                    active: *active,
-                    half: *half,
-                }),
-                _ => None,
-            })
-            .collect();
-        buttons.push(Button {
-            row: None,
-            label: back,
-            active: true,
-            half: false,
-        });
-        buttons
-    }
-
-    /// The id of the button that leaves: Cancel's or Done's, for the controller.
-    fn back_id(&self) -> u8 {
-        if self.page.back() == "gui.cancel" {
-            57
-        } else {
-            50
-        }
-    }
-
-    pub fn buttons(&self, w: f32, h: f32) -> Vec<ButtonSpec> {
-        let rows = self.rows();
-        let cx = half(w);
-        let mut y = quarter(h) + 114.0;
-        let mut beside = false;
-        let mut out = Vec::new();
-        for (k, b) in Self::buttons_of(&rows, self.page.back())
-            .into_iter()
-            .enumerate()
-        {
-            let id = match b.row {
-                Some(_) => FIRST + k as u8,
-                None => self.back_id(),
-            };
-            let label = crate::lang::get(b.label);
-            let (x, bw) = match (b.half, beside) {
-                (true, true) => {
-                    y -= 24.0;
-                    beside = false;
-                    (cx + 2.0, 98.0)
-                }
-                (true, false) => {
-                    beside = true;
-                    (cx - 100.0, 98.0)
-                }
-                (false, _) => {
-                    beside = false;
-                    (cx - 100.0, 200.0)
-                }
-            };
-            out.push((x, y, bw, label, b.active, id));
-            y += 24.0;
-        }
-        out
-    }
-
-    pub fn draw(
-        &self,
-        atlas: &HudAtlas,
-        batch: &mut HudBatch,
-        caret: &mut crate::ui::Caret,
-        focused_at: Instant,
-    ) {
-        let w = batch.width;
-        let rows = self.rows();
-        let title = crate::lang::get(self.page.title());
-        let tw = HudBatch::text_width(atlas, &title);
-        batch.text(atlas, &title, centered_x(w, tw), 20.0, u32::MAX, true);
-        let mut y = 53.0;
-        let mut field = 0;
-        let mut boxes = self.boxes.borrow_mut();
-        for (r, row) in rows.iter().enumerate() {
-            let (text, color) = match row {
-                Row::Field { label, max, .. } => {
-                    let Some(value) = self.texts.get(field) else {
-                        break;
-                    };
-                    super::Screen::field(
-                        atlas,
-                        batch,
-                        caret,
-                        focused_at,
-                        &crate::lang::get(label),
-                        value,
-                        ("page.field", r),
-                        *max,
-                        (y, y + 11.0),
-                        self.focus == field,
-                    );
-                    boxes[field] = [half(w) - 100.0, y + 11.0, 200.0, 20.0];
-                    field += 1;
-                    y += 41.0;
-                    continue;
-                }
-                Row::Space(px) => {
-                    y += f32::from(*px);
-                    continue;
-                }
-                Row::Heading(key) => (crate::lang::get(key), u32::MAX),
-                Row::Note(text) => (text.clone(), rgba(160, 160, 160, 255)),
-                Row::Text { text, quiet, .. } => {
-                    let grey = rgba(160, 160, 160, 255);
-                    (text.clone(), if *quiet { grey } else { u32::MAX })
-                }
-                _ => continue,
-            };
-            for line in super::wrap(&text, WIDTH, atlas) {
-                let lw = HudBatch::text_width(atlas, &line);
-                batch.text(atlas, &line, centered_x(w, lw), y, color, true);
-                y += 10.0;
-            }
-        }
-    }
-
-    pub fn narration(&self) -> crate::widgets::ScreenInfo {
-        crate::widgets::ScreenInfo::new(self.page.id(), crate::lang::get(self.page.title()))
-    }
-
-    /// A press on button `id`, or on a field (`id` None).
-    pub fn click(&mut self, id: Option<u8>, x: f32, y: f32) -> Outcome {
-        match id {
-            Some(id) if id == self.back_id() => Outcome::Back,
-            Some(id) if id >= FIRST => {
-                let rows = self.rows();
-                let buttons = Self::buttons_of(&rows, self.page.back());
-                match buttons.get(usize::from(id - FIRST)) {
-                    Some(Button {
-                        row: Some(r),
-                        active: true,
-                        ..
-                    }) => self.act(*r, Act::Press),
-                    _ => Outcome::None,
-                }
-            }
-            Some(_) => Outcome::None,
-            None => {
-                let hit = self
-                    .boxes
-                    .borrow()
-                    .iter()
-                    .position(|&[bx, by, bw, bh]| crate::ui::hovered((x, y), bx, by, bw, bh));
-                if let Some(field) = hit {
-                    self.focus = field;
-                }
-                Outcome::None
-            }
-        }
-    }
-
-    /// Enter: the focused field submits.
-    pub fn confirm(&self) -> Outcome {
-        let rows = self.rows();
-        match field_rows(&rows).nth(self.focus) {
-            Some(r) => self.act(r, Act::Submit),
-            None => Outcome::None,
-        }
-    }
-
-    /// The page acts; a server it asked to join is joined.
-    fn act(&self, row: usize, act: Act) -> Outcome {
-        self.page.act(row, act);
-        match pages::engine::take_connect() {
-            Some(address) => Outcome::Connect {
-                address,
-                lan: false,
-            },
-            None => Outcome::None,
-        }
-    }
-
-    pub fn focused_box(&self) -> Option<[f32; 4]> {
-        self.boxes.borrow().get(self.focus).copied()
-    }
-
-    /// The focused field, its id and maximum length.
-    pub fn focused_text(&mut self) -> Option<(crate::clipboard::FieldId, &mut String, usize)> {
-        let mut rows = Vec::new();
-        self.page.rows(&mut rows);
-        let (r, max) = rows
-            .iter()
-            .enumerate()
-            .filter_map(|(r, row)| match row {
-                Row::Field { max, .. } => Some((r, *max)),
-                _ => None,
-            })
-            .nth(self.focus)?;
-        let text = self.texts.get_mut(self.focus)?;
-        Some((("page.field", r), text, max))
-    }
-
-    /// Tab through the fields, then the enabled `buttons`, then round to the first field.
-    pub fn tab_dir(&mut self, reverse: bool, buttons: &[u8], button: &mut Option<u8>) {
-        let fields = self.texts.len();
-        let stops = fields + buttons.len();
-        if stops == 0 {
-            return;
-        }
-        let at = match *button {
-            Some(id) => buttons.iter().position(|&b| b == id).map(|i| fields + i),
-            None => Some(self.focus.min(fields.saturating_sub(1))),
-        }
-        .unwrap_or(0);
-        let next = if reverse {
-            (at + stops - 1) % stops
-        } else {
-            (at + 1) % stops
-        };
-        if next < fields {
-            self.focus = next;
-            *button = None;
-        } else {
-            self.focus = fields;
-            *button = Some(buttons[next - fields]);
-        }
-    }
-}
-
-/// The rows that are fields, in order.
-fn field_rows(rows: &[Row]) -> impl Iterator<Item = usize> + '_ {
-    rows.iter()
-        .enumerate()
-        .filter(|(_, row)| matches!(row, Row::Field { .. }))
-        .map(|(r, _)| r)
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use std::sync::Mutex;
-
-    /// A join form: a code and a relay, Join while the code has 4 characters or more.
-    struct Join {
-        heard: Mutex<Vec<(usize, Act)>>,
-        code: Mutex<String>,
-    }
-
-    impl Page for Join {
-        fn id(&self) -> &'static str {
-            "test.join"
-        }
-        fn title(&self) -> &'static str {
-            "horizon.multiplayer.join_friend"
-        }
-        fn place(&self) -> Place {
-            Place::Title
-        }
-        fn rows(&self, rows: &mut Vec<Row>) {
-            let code = self.code.lock().unwrap().clone();
-            rows.push(Row::field("horizon.multiplayer.invite_code", code.clone()).max(8));
-            rows.push(Row::field(
-                "horizon.multiplayer.relay",
-                "relay.example:25599",
-            ));
-            rows.push(Row::button("horizon.multiplayer.join").active(code.len() >= 4));
-        }
-        fn act(&self, row: usize, act: Act) {
-            if let (0, Act::Edit(text)) = (row, &act) {
-                text.clone_into(&mut self.code.lock().unwrap());
-            }
-            if matches!((row, &act), (2, Act::Press) | (0, Act::Submit)) {
-                pages::connect(format!("friend:{}", self.code.lock().unwrap()));
-            }
-            self.heard.lock().unwrap().push((row, act));
-        }
-        fn back(&self) -> &'static str {
-            "gui.cancel"
-        }
-    }
-
-    fn join() -> &'static Join {
-        Box::leak(Box::new(Join {
-            heard: Mutex::new(Vec::new()),
-            code: Mutex::new(String::new()),
-        }))
-    }
-
-    #[test]
-    fn a_form_lays_out_as_join_a_friend() {
-        let form = Form::open(join());
-        let buttons = form.buttons(427.0, 240.0);
-        // Join at a quarter of the height + 114, Cancel 24 below, both 200 wide and centred.
-        assert_eq!(
-            buttons
-                .iter()
-                .map(|(x, y, w, _, on, id)| (*x, *y, *w, *on, *id))
-                .collect::<Vec<_>>(),
-            [
-                (113.0, 174.0, 200.0, false, FIRST),
-                (113.0, 198.0, 200.0, true, 57)
-            ]
-        );
-        assert_eq!(form.texts, ["", "relay.example:25599"]);
-    }
-
-    #[test]
-    fn typing_reaches_the_page_before_a_press_and_enter_submits() {
-        let page = join();
-        let mut form = Form::open(page);
-        let (id, text, max) = form.focused_text().unwrap();
-        assert_eq!((id, max), (("page.field", 0), 8));
-        text.push_str("AB");
-        // Join is greyed: the page heard "AB" before the buttons were laid out.
-        assert!(!form.buttons(427.0, 240.0)[0].4);
-        form.focused_text().unwrap().1.push_str("CD");
-        assert!(form.buttons(427.0, 240.0)[0].4);
-        match form.click(Some(FIRST), 0.0, 0.0) {
-            Outcome::Connect { address, lan } => {
-                assert_eq!((address.as_str(), lan), ("friend:ABCD", false))
-            }
-            _ => panic!("Join did not connect"),
-        }
-        match form.confirm() {
-            Outcome::Connect { address, .. } => assert_eq!(address, "friend:ABCD"),
-            _ => panic!("Enter did not submit"),
-        }
-        let heard = page.heard.lock().unwrap();
-        assert_eq!(
-            *heard,
-            [
-                (0, Act::Edit("AB".into())),
-                (0, Act::Edit("ABCD".into())),
-                (2, Act::Press),
-                (0, Act::Submit)
-            ]
-        );
-        assert!(matches!(form.click(Some(57), 0.0, 0.0), Outcome::Back));
-    }
-
-    #[test]
-    fn tab_goes_through_the_fields_then_the_buttons() {
-        let mut form = Form::open(join());
-        let mut button = None;
-        form.tab_dir(false, &[57], &mut button);
-        assert_eq!((form.focus, button), (1, None));
-        form.tab_dir(false, &[57], &mut button);
-        assert_eq!((form.focus, button), (2, Some(57)));
-        form.tab_dir(false, &[57], &mut button);
-        assert_eq!((form.focus, button), (0, None));
-        form.tab_dir(true, &[57], &mut button);
-        assert_eq!(button, Some(57));
-        assert!(form.focused_text().is_none());
-    }
-}

crates/horizon-client/src/net.rs

@@ -523,22 +523,6 @@ pub fn is_invite(code: &str, full: bool) -> bool {
     horizon_hooks::engine::tunnel().is_some_and(|r| r.is_code(code, full))
 }
 
-/// Whether an invite code can be joined: a resolver is registered (horizon-official registers
-/// horizon-friends'). The title screen's Friends icon, which opens Join a Friend, is inactive
-/// without one, as the pause menu's is without a page to share from.
-pub fn joins_friends() -> bool {
-    #[cfg(test)]
-    if JOINS_FRIENDS_IN_TEST.get() {
-        return true;
-    }
-    horizon_hooks::engine::tunnel().is_some()
-}
-#[cfg(test)]
-thread_local! {
-    /// A test's stand-in for a registered resolver: the registry is the process's and sealed.
-    pub(crate) static JOINS_FRIENDS_IN_TEST: std::cell::Cell<bool> = const { std::cell::Cell::new(false) };
-}
-
 /// An address as players and logs see it: a friend's invite code stays
 /// out of it (its secret is in there).
 pub fn shown_address(address: &str) -> String {

crates/horizon-client/src/parallel.rs

@@ -0,0 +1,146 @@
+//! The helper threads of the server tick's parallel phases (`HORIZON_PARALLEL_TICK`): one pool, made
+//! once, instead of threads started for every phase of every tick. A phase hands it work that gives
+//! the same result in whatever order the threads take it, and each part's results stay in order.
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::{Mutex, OnceLock, PoisonError};
+
+/// `HORIZON_PARALLEL_TICK`: off when unset (0). `1` turns the server tick's parallel phases on (1),
+/// and `verify` does too (2), also checking the phases that can be checked against the serial code.
+/// Each phase gives the serial result either way.
+pub fn parallel_tick() -> u8 {
+    static ON: OnceLock<u8> = OnceLock::new();
+    *ON.get_or_init(|| match std::env::var("HORIZON_PARALLEL_TICK").as_deref() {
+        Ok("1" | "on" | "true") => 1,
+        Ok("verify") => 2,
+        _ => 0,
+    })
+}
+
+/// How many threads a phase may share its work among: `horizon-threads`' count (at most 8) while
+/// the parallel tick is on, else 1.
+pub fn lanes() -> usize {
+    if parallel_tick() > 0 {
+        horizon_threads::worker_count().clamp(1, 8)
+    } else {
+        1
+    }
+}
+
+/// The pool: one thread fewer than a phase's lanes (the server thread takes parts too).
+fn pool() -> &'static rayon::ThreadPool {
+    static POOL: OnceLock<rayon::ThreadPool> = OnceLock::new();
+    POOL.get_or_init(|| {
+        rayon::ThreadPoolBuilder::new()
+            .num_threads(
+                horizon_threads::worker_count()
+                    .clamp(1, 8)
+                    .saturating_sub(1)
+                    .max(1),
+            )
+            .thread_name(|i| format!("tick-helper-{i}"))
+            .build()
+            .expect("the tick's helper threads start")
+    })
+}
+
+/// `f` over `items` in parts on the pool, or all on this thread while the parallel tick is off or
+/// for fewer than `min` items: every part's results, in the order of `items`, as one thread would
+/// give them.
+pub fn par_map<T: Sync, R: Send>(
+    items: &[T],
+    min: usize,
+    f: impl Fn(&[T], &mut Vec<R>) + Sync,
+) -> Vec<R> {
+    par_map_on(items, min, lanes(), f)
+}
+
+/// [`par_map`] on `lanes` threads.
+///
+/// The parts are small (four a thread) and go to whichever thread is free, this one too: a helper
+/// the machine is slow to run (the chunk workers compete for the same cores) holds the phase up by
+/// one part at most instead of an eighth of the work, and the server thread works instead of only
+/// waiting.
+fn par_map_on<T: Sync, R: Send>(
+    items: &[T],
+    min: usize,
+    lanes: usize,
+    f: impl Fn(&[T], &mut Vec<R>) + Sync,
+) -> Vec<R> {
+    if lanes <= 1 || items.len() < min.max(2) {
+        let mut out = Vec::with_capacity(items.len());
+        f(items, &mut out);
+        return out;
+    }
+    let size = items.len().div_ceil(lanes * 4);
+    let count = items.len().div_ceil(size);
+    let next = AtomicUsize::new(0);
+    let slots: Vec<Mutex<Vec<R>>> = (0..count).map(|_| Mutex::new(Vec::new())).collect();
+    let work = || {
+        loop {
+            let k = next.fetch_add(1, Ordering::Relaxed);
+            if k >= count {
+                break;
+            }
+            let part = &items[k * size..((k + 1) * size).min(items.len())];
+            let mut done = Vec::with_capacity(part.len());
+            f(part, &mut done);
+            *slots[k].lock().unwrap_or_else(PoisonError::into_inner) = done;
+        }
+    };
+    pool().in_place_scope(|s| {
+        for _ in 1..lanes {
+            s.spawn(|_| work());
+        }
+        work();
+    });
+    let mut out = Vec::with_capacity(items.len());
+    for slot in slots {
+        out.extend(slot.into_inner().unwrap_or_else(PoisonError::into_inner));
+    }
+    out
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// On several threads, every item's result comes back once, in the items' order, for item
+    /// counts that don't divide into the parts evenly; a part may give several results or none.
+    #[test]
+    fn parts_come_back_in_order() {
+        for n in [0usize, 1, 7, 33, 1000, 4097] {
+            let items: Vec<u64> = (0..n as u64).collect();
+            let f = |part: &[u64], out: &mut Vec<u64>| {
+                for &x in part {
+                    // Spin a little so the threads overlap.
+                    let mut h = x;
+                    for _ in 0..200 {
+                        h = h.wrapping_mul(6364136223846793005).wrapping_add(1);
+                    }
+                    if x % 3 != 0 {
+                        out.push(x * 10);
+                    }
+                    if x % 5 == 0 {
+                        out.push(h);
+                    }
+                }
+            };
+            let mut serial = Vec::new();
+            f(&items, &mut serial);
+            assert_eq!(par_map_on(&items, 1, 4, f), serial, "{n} items");
+        }
+    }
+
+    /// A panic in a part reaches the caller instead of being lost on a helper.
+    #[test]
+    fn a_panic_reaches_the_caller() {
+        let items: Vec<u32> = (0..256).collect();
+        let caught = std::panic::catch_unwind(|| {
+            par_map_on(&items, 1, 4, |part, out: &mut Vec<u32>| {
+                assert!(!part.contains(&200), "item 200");
+                out.extend(part);
+            })
+        });
+        assert!(caught.is_err());
+    }
+}

crates/horizon-client/src/random_tick_sections.rs

@@ -117,6 +117,44 @@ impl RandomTickSections {
         });
         Some((y, mask.clone()))
     }
+    /// Builds, on the tick's helper threads (`crate::parallel`), the masks [`Self::next_masked`]
+    /// will build at the first draw in each ticking section of the chunks `keys` names, from their
+    /// blocks now (`chunk_of`), so the random tick loop finds them built. A mask only tells the loop
+    /// which drawn blocks to read: the draws and the ticks are the same either way, and an edit
+    /// before the loop reaches a section drops its mask as usual.
+    pub fn prepare_masks(
+        &mut self,
+        keys: &[(i32, i32)],
+        chunk_of: impl Fn((i32, i32)) -> Option<Arc<Chunk>>,
+        facts: &BlockFacts,
+        candidate: &(impl Fn(u32) -> bool + Sync),
+    ) {
+        let mut todo: Vec<((i32, i32), usize, Arc<Chunk>)> = Vec::new();
+        for &key in keys {
+            let Some(chunk) = chunk_of(key) else {
+                continue;
+            };
+            let cached = Self::refresh(&mut self.chunks, key, &chunk, &mut |id| {
+                facts.get(BlockStateId(id)).random_ticks
+            });
+            for i in cached.sections.mask.iter() {
+                if cached.masks.get(i).is_some_and(Option::is_none) {
+                    todo.push((key, i, Arc::clone(&chunk)));
+                }
+            }
+        }
+        let masks = crate::parallel::par_map(&todo, 128, |part, out| {
+            out.extend(
+                (part.iter())
+                    .map(|(_, i, chunk)| Self::candidate_mask(&chunk.sections[*i], candidate)),
+            );
+        });
+        for ((key, i, _), mask) in todo.into_iter().zip(masks) {
+            if let Some(cached) = self.chunks.get_mut(&key) {
+                cached.masks[i] = Some(Rc::new(mask));
+            }
+        }
+    }
     fn candidate_mask(section: &Section, mut candidate: impl FnMut(u32) -> bool) -> RandomTickMask {
         let blocks = &section.blocks;
         let mut ids = [0u16; 4096];
@@ -481,25 +519,41 @@ mod tick_parity {
             time: 0,
             speed: 3,
         };
+        // More than 128 ticking sections: prepare_masks must exercise its
+        // helper path when HORIZON_PARALLEL_TICK is enabled, not just defer work.
+        for x in 2..=65 {
+            original.chunks.insert((x, 0), chunk(x));
+        }
         Arc::make_mut(original.chunks.get_mut(&(1, 0)).unwrap()).lit = true; // precipitation-only chunk
         let mut cached = original.clone();
         // The masked path reads only the drawn blocks that can tick.
         let mut masked = original.clone();
+        // The masked path again, its masks built ahead of each tick (`prepare_masks`).
+        let mut prepared = original.clone();
+        let mut prep_cache = RandomTickSections::default();
+        let mut d = Ticks::new(12345);
         let mut cache = RandomTickSections::default();
         let mut mask_cache = RandomTickSections::default();
         let (mut a, mut b, mut c) = (Ticks::new(12345), Ticks::new(12345), Ticks::new(12345));
-        for x in -1..=1 {
+        for x in -1..=65 {
             a.add_empty_chunk(ChunkPos::new(x, 0)).unwrap();
             b.add_empty_chunk(ChunkPos::new(x, 0)).unwrap();
             c.add_empty_chunk(ChunkPos::new(x, 0)).unwrap();
+            d.add_empty_chunk(ChunkPos::new(x, 0)).unwrap();
         }
         for step in 0..128 {
-            for level in [&mut original, &mut cached, &mut masked] {
+            for level in [&mut original, &mut cached, &mut masked, &mut prepared] {
                 level.time = step;
                 level.speed = [0, 1, 3, 9][step as usize % 4];
                 level.reads.borrow_mut().clear();
                 level.events.clear();
                 match step {
+                    4 => {
+                        // The large fixture has exercised every tick speed and
+                        // helper-mask invalidation; keep the remaining edit,
+                        // unload and reload cases small as before.
+                        level.chunks.retain(|&(x, _), _| x <= 1);
+                    }
                     12 => {
                         level.chunks.remove(&(-1, 0));
                     }
@@ -528,15 +582,17 @@ mod tick_parity {
             if step == 12 {
                 cache.remove(&(-1, 0));
                 mask_cache.remove(&(-1, 0));
+                prep_cache.remove(&(-1, 0));
             }
             if step == 50 || step == 60 {
                 facts.states[grass.0 as usize].random_ticks = step == 60;
                 cache.clear();
                 mask_cache.clear();
+                prep_cache.clear();
             }
             // A write between ticks.
             if step == 20 {
-                for level in [&mut original, &mut cached, &mut masked] {
+                for level in [&mut original, &mut cached, &mut masked, &mut prepared] {
                     level.set_raw(BlockPos::new(0, MIN_Y + 16, 0), grass);
                 }
             }
@@ -589,6 +645,55 @@ mod tick_parity {
                 };
                 behaviours.tick_chunk_random_masked(&mut masked, &mut c, ChunkPos::new(x, z), next);
             }
+            let keys = lit(&prepared);
+            let candidate = |id| behaviours.random_tick_candidate(BlockStateId(id));
+            prep_cache.prepare_masks(
+                &keys,
+                |key| prepared.chunks.get(&key).cloned(),
+                &facts,
+                &candidate,
+            );
+            for &(x, z) in &keys {
+                // Every ticking section's mask is there before the tick asks.
+                let now = prepared.chunks[&(x, z)].clone();
+                let mut from = MIN_Y;
+                while let Some((y, _)) = prep_cache.next_masked(
+                    (x, z),
+                    &now,
+                    &facts,
+                    |_| unreachable!("built ahead"),
+                    from,
+                ) {
+                    from = y + 16;
+                }
+            }
+            for (x, z) in keys {
+                let next = |level: &mut Level, from| {
+                    let now = level.chunks[&(x, z)].clone();
+                    prep_cache
+                        .next_masked((x, z), &now, &facts, candidate, from)
+                        .map(|(y, mask)| (y, Some(mask)))
+                };
+                behaviours.tick_chunk_random_masked(
+                    &mut prepared,
+                    &mut d,
+                    ChunkPos::new(x, z),
+                    next,
+                );
+            }
+            assert_eq!(
+                original.events, prepared.events,
+                "prepared events at tick {step}"
+            );
+            assert_eq!(
+                a.random_tick_value, d.random_tick_value,
+                "prepared position RNG at tick {step}"
+            );
+            assert_eq!(
+                original.random.clone().next_long(),
+                prepared.random.next_long(),
+                "prepared level RNG at tick {step}"
+            );
             assert_eq!(original.reads, cached.reads, "ordered reads at tick {step}");
             assert_eq!(
                 original.events, masked.events,
@@ -629,6 +734,9 @@ mod tick_parity {
             for (s, t) in c.sections.iter().zip(&masked.chunks[key].sections) {
                 assert_eq!(s.blocks, t.blocks);
             }
+            for (s, t) in c.sections.iter().zip(&prepared.chunks[key].sections) {
+                assert_eq!(s.blocks, t.blocks);
+            }
         }
     }
 }

crates/horizon-client/src/title.rs

@@ -1940,8 +1940,8 @@ impl Title {
                     )
                     .does(crate::widgets::Action::Exit),
                 ];
-                // CommonButtons' icon buttons: vanilla's Friends button opens a crate's page or
-                // Join a Friend, inactive with neither (`multiplayer::title_entry`).
+                // CommonButtons' icon buttons; Horizon's Join a Friend (an
+                // invite code) sits behind vanilla's Friends button.
                 for (i, (key, plain, id, icon)) in [
                     ("gui.friends.open", "Friends", 5, "friends/friends"),
                     ("options.language", "Language...", 6, "icon/language"),
@@ -1960,7 +1960,7 @@ impl Title {
                         y + 72.0,
                         20.0,
                         &lang::get_or(key, plain),
-                        id != 5 || crate::multiplayer::has_title_entry(),
+                        true,
                         id,
                     );
                     button.icon = Some((icon, 15.0));
@@ -3658,7 +3658,12 @@ Would you like to enable the Narrator or visit the Accessibility Settings?",
             .map(|b| b.id);
         if hit.is_some() {
             self.focused_button = hit;
-        } else if matches!(&self.mode, Mode::Multiplayer(screen) if screen.has_fields()) {
+        } else if matches!(
+            self.mode,
+            Mode::Multiplayer(
+                crate::multiplayer::Screen::Edit { .. } | crate::multiplayer::Screen::Friend { .. }
+            )
+        ) {
             // A press on a text field takes the focus from the button.
             self.focused_button = None;
         }
@@ -3777,7 +3782,7 @@ Would you like to enable the Narrator or visit the Accessibility Settings?",
                 crate::multiplayer::Outcome::None => Action::None,
             },
             (Mode::Main, Some(5)) => {
-                self.mode = crate::multiplayer::title_entry().map_or(Mode::Main, Mode::Multiplayer);
+                self.mode = Mode::Multiplayer(crate::multiplayer::Screen::friend(true));
                 Action::None
             }
             (Mode::Onboarding(_), Some(onboarding::SOUNDS)) => Action::Sounds,
@@ -5301,8 +5306,6 @@ mod classic_tests {
 
     #[test]
     fn classic_main_menu_icons_route_language_and_accessibility() {
-        // Join a Friend needs a resolver (title/tests.rs tests the icon without one).
-        crate::net::JOINS_FRIENDS_IN_TEST.set(true);
         let mut title = Title::new(
             PathBuf::new(),
             Path::new("/nonexistent/horizon-test-resources"),

crates/horizon-client/src/title/tests.rs

@@ -100,61 +100,13 @@ impl Drop for Fixture {
     }
 }
 
-/// Join a Friend needs a resolver; these tests stand one in (`net::JOINS_FRIENDS_IN_TEST`).
-fn with_resolver() -> Fixture {
-    crate::net::JOINS_FRIENDS_IN_TEST.set(true);
-    Fixture::new()
-}
-
-#[test]
-fn without_a_resolver_the_friends_icon_is_inactive_and_opens_nothing() {
-    crate::net::JOINS_FRIENDS_IN_TEST.set(false);
-    let mut f = Fixture::new();
-    let icons: Vec<_> = f
-        .title
-        .buttons(480.0, 262.0)
-        .into_iter()
-        .filter(|b| b.icon.is_some())
-        .map(|b| (b.label.clone(), b.enabled, b.x))
-        .collect();
-    // Same place and look as with one, only greyed; Language and Accessibility stay active.
-    let friends = icons.iter().find(|(label, ..)| label == "Friends").unwrap();
-    assert!(!friends.1);
-    assert!(
-        icons
-            .iter()
-            .filter(|(label, ..)| label != "Friends")
-            .all(|(_, on, _)| *on)
-    );
-    assert_eq!(f.click("Friends"), Action::None);
-    assert!(matches!(f.title.mode, Mode::Main));
-    crate::net::JOINS_FRIENDS_IN_TEST.set(true);
-    let with: Vec<_> = f
-        .title
-        .buttons(480.0, 262.0)
-        .into_iter()
-        .filter(|b| b.icon.is_some())
-        .map(|b| (b.label.clone(), b.x, b.y))
-        .collect();
-    assert_eq!(
-        with.iter()
-            .map(|(l, x, _)| (l.clone(), *x))
-            .collect::<Vec<_>>(),
-        icons
-            .iter()
-            .map(|(l, _, x)| (l.clone(), *x))
-            .collect::<Vec<_>>()
-    );
-    crate::net::JOINS_FRIENDS_IN_TEST.set(false);
-}
-
 mod multiplayer_parent_navigation {
     use super::*;
     use crate::multiplayer::{Outcome, Screen};
 
     #[test]
     fn friend_cancel_returns_to_title() {
-        let mut f = with_resolver();
+        let mut f = Fixture::new();
         f.click("Friends");
         assert!(matches!(
             f.title.mode,
@@ -166,7 +118,7 @@ mod multiplayer_parent_navigation {
 
     #[test]
     fn friend_escape_returns_to_title() {
-        let mut f = with_resolver();
+        let mut f = Fixture::new();
         f.click("Friends");
         assert!(matches!(
             f.title.mode,

crates/horizon-core/src/random.rs

@@ -1,8 +1,10 @@
 //! Vanilla's two seeded random streams. No platform/default Rust RNG is used.
 mod hash;
+mod ordered;
 mod positional;
 mod worldgen;
 pub use hash::{seed_from_hash, seed_from_hash_utf16};
+pub use ordered::{LaneDone, OrderedDraws};
 pub use positional::{
     LegacyPositionalRandom, PositionalRandom, XoroshiroPositionalRandom, java_string_hash,
     position_seed,
@@ -104,10 +106,23 @@ pub trait RandomSource: Sized {
     }
 }
 
-#[derive(Clone, Debug)]
+#[derive(Debug)]
 pub struct LegacyRandom {
     state: u64,
     gaussian: Option<f64>,
+    /// A lane of an [`OrderedDraws`] stream: each item's first draw waits for its turn.
+    gate: Option<Box<ordered::Gate>>,
+}
+
+/// A copy draws on its own, from the state it was copied at: it is never gated.
+impl Clone for LegacyRandom {
+    fn clone(&self) -> Self {
+        Self {
+            state: self.state,
+            gaussian: self.gaussian,
+            gate: None,
+        }
+    }
 }
 
 impl LegacyRandom {
@@ -118,6 +133,7 @@ impl LegacyRandom {
         Self {
             state: (seed as u64 ^ Self::MULTIPLIER) & Self::MASK,
             gaussian: None,
+            gate: None,
         }
     }
 
@@ -126,6 +142,7 @@ impl LegacyRandom {
         Self {
             state,
             gaussian: None,
+            gate: None,
         }
     }
 
@@ -134,7 +151,18 @@ impl LegacyRandom {
         self.state
     }
 
+    /// A gated lane's first draw of an item waits for its turn and takes the shared state.
+    #[inline]
+    fn sync(&mut self) {
+        if let Some(gate) = self.gate.as_deref_mut()
+            && !gate.held()
+        {
+            (self.state, self.gaussian) = gate.take();
+        }
+    }
+
     fn bits(&mut self, count: u32) -> i32 {
+        self.sync();
         self.state = self.state.wrapping_mul(Self::MULTIPLIER).wrapping_add(11) & Self::MASK;
         let value = (self.state >> (48 - count)) as i32;
         #[cfg(feature = "rng-trace")]
@@ -149,7 +177,11 @@ impl RandomSource for LegacyRandom {
         LegacyPositionalRandom::new(self.next_long())
     }
     fn set_seed(&mut self, seed: i64) {
-        *self = Self::new(seed);
+        // LegacyRandomSource.setSeed resets the stream and its gaussian cache, not the lane.
+        // Reseeding is an ordered mutation even when this item makes no draws afterwards.
+        self.sync();
+        self.state = (seed as u64 ^ Self::MULTIPLIER) & Self::MASK;
+        self.gaussian = None;
     }
 
     fn next_int(&mut self) -> i32 {
@@ -190,6 +222,7 @@ impl RandomSource for LegacyRandom {
         value as f64 * (1.0 / 9_007_199_254_740_992.0)
     }
     fn next_gaussian(&mut self) -> f64 {
+        self.sync();
         let mut cache = self.gaussian.take();
         let value = marsaglia_polar(&mut cache, || self.next_double());
         self.gaussian = cache;
@@ -352,6 +385,24 @@ mod tests {
         invalid_bounds_do_not_consume_state(XoroshiroRandom::new(0));
     }
 
+    #[test]
+    fn legacy_reseed_resets_the_gaussian_cache_and_stream() {
+        let mut rng = LegacyRandom::new(42);
+        for seed in [i64::MIN, -1, 0, 42, i64::MAX] {
+            // Leave a cached second gaussian before reseeding.
+            rng.next_gaussian();
+            rng.set_seed(seed);
+            let mut fresh = LegacyRandom::new(seed);
+            for _ in 0..3 {
+                assert_eq!(
+                    rng.next_gaussian().to_bits(),
+                    fresh.next_gaussian().to_bits()
+                );
+                assert_eq!(rng.next_long(), fresh.next_long());
+            }
+        }
+    }
+
     #[test]
     fn zero_state_fallback_and_reseed() {
         let mut a = XoroshiroRandom::from_state(0, 0);

crates/horizon-core/src/random/ordered.rs

@@ -0,0 +1,401 @@
+//! One [`LegacyRandom`] stream shared by several threads that each run their part of a list in list
+//! order (the server's parallel mob tick, `game/mob_islands.rs`): every draw comes out exactly as if
+//! the whole list had run on one thread.
+//!
+//! Each thread (a lane) runs a subset of the list's items, in ascending item order, and says which
+//! item it is on ([`LegacyRandom::begin_item`]). The first draw or reseed waits until every other
+//! lane has passed that item, then takes the stream's state; the item's end gives it back
+//! ([`LegacyRandom::end_item`]). So the items before it have made all their draws and the items after
+//! it none. The lane on the smallest item never waits, so the lanes can't deadlock, as long as every
+//! lane has its own thread. An item that never draws or reseeds never waits.
+use super::LegacyRandom;
+use std::sync::{
+    Arc, Mutex, PoisonError,
+    atomic::{AtomicUsize, Ordering},
+};
+
+/// The shared stream and where each lane is.
+#[derive(Debug)]
+pub struct OrderedDraws {
+    /// The stream's state between holders: the generator state and the gaussian cache.
+    state: Mutex<(u64, Option<f64>)>,
+    /// The item each lane is on; `usize::MAX` once it has finished.
+    lanes: Box<[AtomicUsize]>,
+}
+
+/// A lane's hold on the shared stream, inside its [`LegacyRandom`].
+#[derive(Debug)]
+pub(super) struct Gate {
+    draws: Arc<OrderedDraws>,
+    lane: usize,
+    item: usize,
+    held: bool,
+}
+
+impl OrderedDraws {
+    /// `random`'s stream shared by `lanes` lanes, none of them started.
+    pub fn new(random: &LegacyRandom, lanes: usize) -> Arc<Self> {
+        Arc::new(Self {
+            state: Mutex::new((random.state, random.gaussian)),
+            lanes: (0..lanes).map(|_| AtomicUsize::new(0)).collect(),
+        })
+    }
+    /// The random lane `lane` draws from: its draws wait for their turn.
+    pub fn lane(self: &Arc<Self>, lane: usize) -> LegacyRandom {
+        let mut random = LegacyRandom::from_state(0);
+        random.gate = Some(Box::new(Gate {
+            draws: Arc::clone(self),
+            lane,
+            item: 0,
+            held: false,
+        }));
+        random
+    }
+    /// The stream as the lanes left it, once they have all finished.
+    pub fn finish(&self) -> LegacyRandom {
+        let (state, gaussian) = *self.state.lock().unwrap_or_else(PoisonError::into_inner);
+        let mut random = LegacyRandom::from_state(state);
+        random.gaussian = gaussian;
+        random
+    }
+    /// Wait until lane `lane` has finished item `item` (it is on a later one, or done).
+    pub fn wait_past(&self, lane: usize, item: usize) {
+        let mut spins = 0u32;
+        while self.lanes[lane].load(Ordering::Acquire) <= item {
+            spins += 1;
+            if spins < 64 {
+                std::hint::spin_loop();
+            } else {
+                std::thread::yield_now();
+            }
+        }
+    }
+    fn wait_for_turn(&self, lane: usize, item: usize) {
+        let mut spins = 0u32;
+        loop {
+            let ready = self
+                .lanes
+                .iter()
+                .enumerate()
+                .all(|(k, at)| k == lane || at.load(Ordering::Acquire) > item);
+            if ready {
+                return;
+            }
+            spins += 1;
+            if spins < 64 {
+                std::hint::spin_loop();
+            } else {
+                std::thread::yield_now();
+            }
+        }
+    }
+}
+
+impl Gate {
+    /// The shared state for this item's first draw, after the items before it.
+    pub(super) fn take(&mut self) -> (u64, Option<f64>) {
+        self.draws.wait_for_turn(self.lane, self.item);
+        self.held = true;
+        *self
+            .draws
+            .state
+            .lock()
+            .unwrap_or_else(PoisonError::into_inner)
+    }
+    pub(super) fn held(&self) -> bool {
+        self.held
+    }
+}
+
+impl LegacyRandom {
+    /// A gated random's lane starts item `item` (ascending within the lane). Nothing for an
+    /// ungated one.
+    pub fn begin_item(&mut self, item: usize) {
+        if let Some(gate) = self.gate.as_deref_mut() {
+            gate.item = item;
+            gate.draws.lanes[gate.lane].store(item, Ordering::Release);
+        }
+    }
+    /// The item's draws are done: the shared stream goes on from where it left it.
+    pub fn end_item(&mut self) {
+        if let Some(gate) = self.gate.as_deref_mut()
+            && gate.held
+        {
+            *gate
+                .draws
+                .state
+                .lock()
+                .unwrap_or_else(PoisonError::into_inner) = (self.state, self.gaussian);
+            gate.held = false;
+        }
+    }
+    /// The lane has run all its items: no other lane waits for it any more.
+    pub fn finish_lane(&mut self) {
+        self.end_item();
+        if let Some(gate) = self.gate.as_deref_mut() {
+            gate.draws.lanes[gate.lane].store(usize::MAX, Ordering::Release);
+        }
+    }
+    /// A gated random's [`LaneDone`], for its lane's thread to hold while it runs; None for an
+    /// ungated one.
+    pub fn lane_done(&self) -> Option<LaneDone> {
+        self.gate.as_deref().map(|gate| LaneDone {
+            draws: Arc::clone(&gate.draws),
+            lane: gate.lane,
+        })
+    }
+}
+
+/// Marks a lane finished when dropped, also while a panic unwinds its thread: the other lanes
+/// then run to their end instead of waiting for a lane that stopped, and the panic reaches the
+/// caller when it joins the threads.
+#[must_use = "the lane is marked finished when this is dropped"]
+pub struct LaneDone {
+    draws: Arc<OrderedDraws>,
+    lane: usize,
+}
+
+impl Drop for LaneDone {
+    fn drop(&mut self) {
+        self.draws.lanes[self.lane].store(usize::MAX, Ordering::Release);
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::random::RandomSource;
+
+    /// The timeout lives outside the scoped joins: a broken gate must fail, not hang the test suite.
+    fn bounded<T: Send + 'static>(run: impl FnOnce() -> T + Send + 'static) -> T {
+        let (sent, finished) = std::sync::mpsc::channel();
+        std::thread::spawn(move || {
+            let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(run));
+            let _ = sent.send(result);
+        });
+        match finished
+            .recv_timeout(std::time::Duration::from_secs(20))
+            .expect("ordered lanes must finish within the timeout")
+        {
+            Ok(value) => value,
+            Err(panic) => std::panic::resume_unwind(panic),
+        }
+    }
+
+    #[test]
+    fn reseeding_lanes_preserve_the_serial_stream() {
+        let mut initial = LegacyRandom::new(42);
+        initial.next_gaussian(); // A reseed before the first draw must also clear this cache.
+        // Independent serial expectation: replace an ungated generator, without calling set_seed.
+        let mut serial = LegacyRandom::new(11);
+        let mut expected = vec![vec![serial.next_gaussian().to_bits()]];
+        let before_reseed = serial.next_gaussian().to_bits();
+        serial = LegacyRandom::new(-7);
+        expected.push(vec![before_reseed, serial.next_gaussian().to_bits()]);
+        serial = LegacyRandom::new(i64::MIN);
+        expected.push(vec![]); // Reseed-only item must publish the state and clear the cache.
+        expected.push(vec![serial.next_gaussian().to_bits()]);
+        expected.push(vec![
+            serial.next_gaussian().to_bits(),
+            serial.next_int() as u64,
+        ]);
+        let before_reseed = serial.next_int() as u64;
+        serial = LegacyRandom::new(0);
+        expected.push(vec![
+            before_reseed,
+            serial.next_long() as u64,
+            serial.next_gaussian().to_bits(),
+        ]);
+
+        let (got, mut final_random) = bounded(move || {
+            let shared = OrderedDraws::new(&initial, 3);
+            let mut got: Vec<_> = std::thread::scope(|s| {
+                let handles: Vec<_> = (0..3)
+                    .map(|lane| {
+                        let shared = Arc::clone(&shared);
+                        s.spawn(move || {
+                            let mut random = shared.lane(lane);
+                            let mut out = Vec::new();
+                            // Intentionally no LaneDone: finish_lane itself must survive reseeding.
+                            for item in (lane..6).step_by(3) {
+                                random.begin_item(item);
+                                let values = match item {
+                                    0 => {
+                                        random.set_seed(11);
+                                        vec![random.next_gaussian().to_bits()]
+                                    }
+                                    1 => {
+                                        let before = random.next_gaussian().to_bits();
+                                        random.set_seed(-7);
+                                        vec![before, random.next_gaussian().to_bits()]
+                                    }
+                                    2 => {
+                                        random.set_seed(i64::MIN);
+                                        vec![]
+                                    }
+                                    3 => vec![random.next_gaussian().to_bits()],
+                                    4 => vec![
+                                        random.next_gaussian().to_bits(),
+                                        random.next_int() as u64,
+                                    ],
+                                    5 => {
+                                        let before = random.next_int() as u64;
+                                        random.set_seed(0);
+                                        vec![
+                                            before,
+                                            random.next_long() as u64,
+                                            random.next_gaussian().to_bits(),
+                                        ]
+                                    }
+                                    _ => unreachable!(),
+                                };
+                                random.end_item();
+                                out.push((item, values));
+                            }
+                            random.finish_lane();
+                            out
+                        })
+                    })
+                    .collect();
+                handles
+                    .into_iter()
+                    .flat_map(|h| h.join().unwrap())
+                    .collect()
+            });
+            got.sort_by_key(|(item, _)| *item);
+            (
+                got.into_iter().map(|(_, v)| v).collect::<Vec<_>>(),
+                shared.finish(),
+            )
+        });
+        assert_eq!(got, expected);
+        assert_eq!(final_random.state(), serial.state());
+        // finish() must keep the cached gaussian as well as the LCG state.
+        assert_eq!(
+            final_random.next_gaussian().to_bits(),
+            serial.next_gaussian().to_bits()
+        );
+        assert_eq!(final_random.next_long(), serial.next_long());
+    }
+
+    #[test]
+    fn a_reseeded_lane_keeps_its_panic_guard_and_published_stream() {
+        bounded(|| {
+            let shared = OrderedDraws::new(&LegacyRandom::new(42), 2);
+            std::thread::scope(|s| {
+                let panicking = Arc::clone(&shared);
+                let handle = s.spawn(move || {
+                    let mut random = panicking.lane(0);
+                    random.begin_item(0);
+                    random.set_seed(7);
+                    let _done = random.lane_done().expect("reseed must keep the lane guard");
+                    random.next_gaussian();
+                    random.end_item();
+                    panic!("after publishing the reseeded item");
+                });
+                let mut random = shared.lane(1);
+                random.begin_item(1);
+                let mut serial = LegacyRandom::new(7);
+                serial.next_gaussian();
+                assert_eq!(
+                    random.next_gaussian().to_bits(),
+                    serial.next_gaussian().to_bits()
+                );
+                random.finish_lane();
+                assert!(handle.join().is_err(), "the panic must reach the caller");
+                assert_eq!(shared.finish().next_long(), serial.next_long());
+            });
+        });
+    }
+
+    /// Items 0..40 over three lanes, each drawing on some items: the same values as one thread.
+    #[test]
+    fn lanes_draw_in_list_order() {
+        let draws_on = |item: usize| match item % 5 {
+            0 => 0,
+            1 => 1,
+            2 => 3,
+            _ => (item % 3) as i32,
+        };
+        let mut serial = LegacyRandom::new(42);
+        let expected: Vec<Vec<i32>> = (0..40)
+            .map(|i| {
+                (0..draws_on(i))
+                    .map(|_| serial.next_int_bound(1000).unwrap())
+                    .collect()
+            })
+            .collect();
+        let shared = OrderedDraws::new(&LegacyRandom::new(42), 3);
+        let got: Vec<(usize, Vec<i32>)> = std::thread::scope(|s| {
+            let handles: Vec<_> = (0..3)
+                .map(|lane| {
+                    let shared = Arc::clone(&shared);
+                    s.spawn(move || {
+                        let mut random = shared.lane(lane);
+                        let mut out = Vec::new();
+                        for item in (0..40).filter(|i| (i * 7 + 3) % 3 == lane) {
+                            random.begin_item(item);
+                            let values = (0..draws_on(item))
+                                .map(|_| random.next_int_bound(1000).unwrap())
+                                .collect();
+                            random.end_item();
+                            out.push((item, values));
+                        }
+                        random.finish_lane();
+                        out
+                    })
+                })
+                .collect();
+            handles
+                .into_iter()
+                .flat_map(|h| h.join().unwrap())
+                .collect()
+        });
+        let mut got = got;
+        got.sort_by_key(|(i, _)| *i);
+        assert_eq!(
+            got.into_iter().map(|(_, v)| v).collect::<Vec<_>>(),
+            expected
+        );
+        assert_eq!(shared.finish().state(), serial.state());
+    }
+
+    /// A lane that panics part way, holding its [`LaneDone`], doesn't hold up a lane whose later
+    /// items wait for it: that lane finishes, and the panic reaches the caller.
+    #[test]
+    fn a_panicking_lane_lets_the_others_finish() {
+        let (sent, finished) = std::sync::mpsc::channel();
+        std::thread::spawn(move || {
+            let shared = OrderedDraws::new(&LegacyRandom::new(7), 2);
+            let caught = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
+                std::thread::scope(|s| {
+                    let panicking = Arc::clone(&shared);
+                    s.spawn(move || {
+                        let mut random = panicking.lane(0);
+                        let _done = random.lane_done();
+                        for item in [0, 2, 4] {
+                            random.begin_item(item);
+                            random.next_int_bound(10).unwrap();
+                            assert_ne!(item, 2, "item 2");
+                            random.end_item();
+                        }
+                        random.finish_lane();
+                    });
+                    let mut random = shared.lane(1);
+                    let _done = random.lane_done();
+                    for item in [1, 3, 5] {
+                        random.begin_item(item);
+                        random.next_int_bound(10).unwrap();
+                        random.end_item();
+                    }
+                    random.finish_lane();
+                });
+            }));
+            let _ = sent.send(caught.is_err());
+        });
+        let panicked = finished
+            .recv_timeout(std::time::Duration::from_secs(20))
+            .expect("the lanes finished instead of waiting for the one that panicked");
+        assert!(panicked, "the panic reached the caller");
+    }
+}

crates/horizon-hooks/src/pages.rs

@@ -19,11 +19,6 @@ pub enum Place {
     /// start a group of their own after the page's. A [`Row::Heading`] starts another group. A
     /// [`Row::Key`] shows its key there, without changing it: give a page of its own to a key.
     Extras,
-    /// The title screen's Friends icon slot (vanilla's `CommonButtons.friends`, beside Language
-    /// and Accessibility): with a page registered here the icon opens the first one as a form of
-    /// its own, as Join a Friend is drawn ([`Row::Field`]). It shows heading, note, text, space,
-    /// field and button rows; the others belong in the options shell.
-    Title,
 }
 
 /// One page. Its values live in the crate (statics, often its settings [`crate::settings::Store`]);
@@ -45,11 +40,6 @@ pub trait Page: Sync {
     fn tooltip(&self, _row: usize) -> Option<Label> {
         None
     }
-    /// The label of the button the engine adds to leave a page drawn as its own screen: a
-    /// translation key, `gui.done` by default (`gui.cancel` for a form that does something).
-    fn back(&self) -> &'static str {
-        "gui.done"
-    }
 }
 
 /// A row of a page. Labels are translation keys.
@@ -95,16 +85,6 @@ pub enum Row {
     /// A key binding, as Key Binds shows one: `key` is its key name (`key.keyboard.z`,
     /// `key.mouse.left`). A press waits for the next key or mouse button ([`Act::Bind`]).
     Key { label: &'static str, key: String },
-    /// A text field ([`Row::field`]) on a page drawn as its own screen, as vanilla's `EditBox`:
-    /// its label above it, focus by click or Tab, typing, the caret keys, selection, copy and
-    /// paste, at most `max` characters. `text` is its value when the page opens; the engine then
-    /// keeps it, and sends each change ([`Act::Edit`]) and Enter ([`Act::Submit`]).
-    #[non_exhaustive]
-    Field {
-        label: &'static str,
-        text: String,
-        max: usize,
-    },
 }
 
 impl Row {
@@ -127,23 +107,6 @@ impl Row {
         }
     }
 
-    /// A text field starting with `text`, at most 32 characters (`EditBox`'s default).
-    pub fn field(label: &'static str, text: impl Into<String>) -> Self {
-        Self::Field {
-            label,
-            text: text.into(),
-            max: 32,
-        }
-    }
-
-    /// A text field taking at most `chars` characters.
-    pub fn max(mut self, chars: usize) -> Self {
-        if let Self::Field { max, .. } = &mut self {
-            *max = chars;
-        }
-        self
-    }
-
     /// A button greyed out (`false`): drawn, never pressed.
     pub fn active(mut self, on: bool) -> Self {
         if let Self::Button { active, .. } = &mut self {
@@ -204,19 +167,6 @@ pub enum Act {
     Slide(f32),
     /// A key row got its new key: its key name (`key.keyboard.z`).
     Bind(String),
-    /// A field's text changed: all of it.
-    Edit(String),
-    /// Enter in a field.
-    Submit,
-}
-
-/// The server a page asked to join ([`connect`]).
-static CONNECT: std::sync::Mutex<Option<String>> = std::sync::Mutex::new(None);
-
-/// From a [`Place::Title`] page's [`Page::act`]: join the server at `address` as Direct
-/// Connection does (the engine's addresses, `friend:<code>` among them), once the act returns.
-pub fn connect(address: impl Into<String>) {
-    *CONNECT.lock().unwrap_or_else(|e| e.into_inner()) = Some(address.into());
 }
 
 static PAGES: Registry<dyn Page> = Registry::new();
@@ -235,9 +185,4 @@ pub mod engine {
     pub fn all() -> &'static [&'static dyn Page] {
         PAGES.all()
     }
-
-    /// The server a page asked to join since the last call ([`connect`]).
-    pub fn take_connect() -> Option<String> {
-        CONNECT.lock().unwrap_or_else(|e| e.into_inner()).take()
-    }
 }

crates/horizon-platform/src/lib.rs

@@ -65,7 +65,7 @@ pub mod settings {
 
 /// Settings pages a crate adds to the options screens (docs/layers.md, "Platform registries").
 pub mod pages {
-    pub use horizon_hooks::pages::{Act, Label, Page, Place, Row, connect, register};
+    pub use horizon_hooks::pages::{Act, Label, Page, Place, Row, register};
 }
 
 /// Drawing on the HUD (docs/layers.md, "Platform registries").

docs/layers.md

@@ -363,7 +363,7 @@ pub trait Store: Sync {
 A page lives in the options screen's shell, and its entry sits on one of the game's menus.
 
 ```rust
-pub enum Place { HorizonSettings, Pause, Extras, Title }
+pub enum Place { HorizonSettings, Title, Pause }
 pub trait Page: Sync {
     fn id(&self) -> &'static str;            // stable, for horizon-ctl and the page stack
     fn title(&self) -> &'static str;         // translation key
@@ -371,7 +371,6 @@ pub trait Page: Sync {
     fn rows(&self, rows: &mut Vec<Row>);     // asked when the page is drawn, not every frame elsewhere
     fn act(&self, row: usize, act: Act);     // a click, a reverse click, a slider's value, a key
     fn tooltip(&self, row: usize) -> Option<Label> { None }   // hover text, may follow the value
-    fn back(&self) -> &'static str { "gui.done" }            // the leave button of a page drawn as a screen
 }
 pub enum Row {
     Heading(&'static str),                   // translation key
@@ -383,11 +382,9 @@ pub enum Row {
     Key { label: &'static str, key: String },                 // a binding: `key.keyboard.z`
     Text { text: String, quiet: bool, groups: bool },         // Row::text(..): white or grey, wrapped
     Space(u8),                                                // GUI pixels, on a page drawn as a screen
-    Field { label: &'static str, text: String, max: usize },  // Row::field(..).max(n): an EditBox, on a page drawn as a screen
 }
 pub enum Label { Key(&'static str), Value(String) }
-pub enum Act { Press, PressBack, Slide(f32), Bind(String), Edit(String), Submit }
-pub fn connect(address: impl Into<String>);                   // from a Title page's act: join that server
+pub enum Act { Press, PressBack, Slide(f32), Bind(String) }
 ```
 
 - `HorizonSettings`: a row of the Horizon Settings page, after the engine's own pages. `Title` and `Pause`: a
@@ -398,17 +395,6 @@ pub fn connect(address: impl Into<String>);                   // from a Title pa
   button, `.half()` puts two side by side, `.footer()` sets one beside Done on a page drawn as its own screen,
   `.quiet()` greys text, `.groups()` breaks it only between `-`-separated groups (an invite code). In the options
   shell, text is notes and space is nothing.
-- `Title` is the title screen's Friends icon slot (vanilla's `CommonButtons.friends`), not a new row: the icon
-  keeps its place and look at every GUI scale, and opens the first page registered there as a form drawn as Join a
-  Friend is (`multiplayer/page.rs`): the title, the rows from y 53 (a field's grey label, its 200-wide box 11
-  below, 41 a field), the buttons in a column from a quarter of the height + 114, the back button last
-  (`Page::back`, `gui.cancel` for a form). It shows heading, note, text, space, field and button rows. With no page
-  there the icon opens the engine's Join a Friend while an invite-code resolver is registered, else it is
-  inactive.
-- `Field` is vanilla's `EditBox` on a page drawn as its own screen: focus by click or Tab, typing, the caret keys,
-  selection, copy and paste, at most `max` characters (32 by default). The engine keeps its text while the page is
-  open and sends each change (`Act::Edit`, before any other act or redraw reaches the page) and Enter
-  (`Act::Submit`). The options shell has no field.
 - Rows are rebuilt when the page is drawn or acted on, like the engine's own pages: no cost while no page is
   open.
 - The engine keeps native mods' pages (`Page::NativeMod`) as they are. Both kinds sit in one module that

docs/play-with-friends.md

@@ -53,8 +53,7 @@ router that opened the port, or with a forwarded port, that's it: no server took
 (`horizon_hooks::tunnel`, `horizon_platform::hooks::tunnel` for the Horizon layer): the resolver a hook offers
 reads codes and opens the tunnel, which the join (`net.rs`) then reaches on a loopback port like any server.
 horizon-friends' `game` side offers it, and the official executable (`horizon-official`) registers it;
-horizon-client's own executable has none, so there a `friend:` address is refused and the title screen's
-Friends icon, which opens Join a Friend, is inactive (`net::joins_friends`). Hosting needs no hook: the
+horizon-client's own executable has none, so there a `friend:` address is refused. Hosting needs no hook: the
 host end connects to the world's LAN port over loopback, so tunnelled guests are LAN guests. It still lives in
 horizon-client (`game/friends.rs`, `game/friends/screen.rs`, the `friends` feature, which comes with
 `horizon-extras`) until the pause menu and `horizon-ctl` offer entries to crates above the engine.

tools/command_exec_corpus.py

@@ -6291,6 +6291,14 @@ def moving_piston_presentation_diagnostic():
     return moving_piston_presentation(cmd, setup)
 
 
+def piston_mob_bounce():
+    try:
+        from .piston_bounce_corpus import piston_bounce
+    except ImportError:
+        from piston_bounce_corpus import piston_bounce
+    return piston_bounce(cmd, setup)
+
+
 def piston_entity_tick_order():
     """#9337: entities travel before moving pistons push them, not afterward."""
     ops = []
@@ -7481,6 +7489,7 @@ def groups():
         ('wither spawn', wither_spawn()),
         ('riding teleport', riding_teleport()),
         ('piston entity tick order', piston_entity_tick_order()),
+        ('piston mob bounce', piston_mob_bounce()),
         ('primed tnt box', primed_tnt_box()),
         ('primed tnt fall', primed_tnt_fall()),
         ('command block last output', command_block_last_output()),

tools/piston_bounce_corpus.py

@@ -0,0 +1,64 @@
+"""Actual downward piston pushes onto slime, including the blocked move (#12091)."""
+
+
+def piston_bounce(cmd, setup):
+    ops = setup('/tick freeze', '/gamerule spawn_mobs false',
+                '/gamerule advance_time false', '/forceload add 0 0')
+    ops.append({'op': 'settle', 'required_chunks': [[0, 0]]})
+    ops += setup('/tp @s 1.5 200 1.5', '/gamemode creative @s',
+                 '/kill @e[type=!player]', '/fill 0 199 0 15 199 15 stone',
+                 '/fill 0 200 0 15 208 15 air',
+                 '/scoreboard objectives add piston_bounce dummy',
+                 '/execute store result score start piston_bounce run time query gametime')
+    cases = [('slime_block', 'slime_block', False),
+             ('slime_block', 'stone', False),
+             ('stone', 'slime_block', False),
+             ('slime_block', 'slime_block', True)]
+    for index, (floor, top, no_ai) in enumerate(cases):
+        x, z = 4 + index % 2 * 6, 4 + index // 2 * 6
+        selector = f'@e[tag=piston_bounce_{index},limit=1]'
+        for command in (f'/setblock {x} 200 {z} {floor}',
+                        f'/setblock {x} 204 {z} {top}',
+                        f'/setblock {x} 205 {z} piston[facing=down]',
+                        f'/summon wither_skeleton {x}.5 201 {z}.5 '
+                        '{PersistenceRequired:1b,Invulnerable:1b,Motion:[0d,0d,0d],'
+                        f'NoAI:{int(no_ai)}b,Tags:["piston_bounce_{index}"]' + '}'):
+            ops.append(cmd(command, expect_result=1))
+        ops += setup(f'/attribute {selector} movement_speed base set 0')
+        fields = (f'Pos:[{x}.5d,201d,{z}.5d],Motion:[0d,0d,0d],'
+                  'Health:20f' + (',NoAI:1b' if no_ai else ''))
+        ops.append(cmd(f'/execute if entity @e[tag=piston_bounce_{index},limit=1,type=wither_skeleton,nbt={{' + fields + '}]',
+                       expect_result=1))
+        ops.append(cmd(f'/execute unless entity @e[tag=piston_bounce_{index},nbt={{NoGravity:1b}}]',
+                       expect_result=1))
+        if not no_ai:
+            ops.append(cmd(f'/execute unless entity @e[tag=piston_bounce_{index},nbt={{NoAI:1b}}]',
+                           expect_result=1))
+    for tick in range(7):
+        if tick == 1:
+            for x, z in [(4, 4), (10, 4), (4, 10), (10, 10)]:
+                ops.append(cmd(f'/setblock {x + 1} 205 {z} redstone_block', expect_result=1))
+        if tick == 3:
+            # Remove the ceiling so the bounce becomes observable travel.
+            for x, z in [(4, 4), (10, 4), (4, 10), (10, 10)]:
+                ops += setup(f'/fill {x} 203 {z} {x} 205 {z} air')
+                for y in (203, 204, 205):
+                    ops.append(cmd(f'/execute if block {x} {y} {z} air', expect_result=1))
+        if tick:
+            ops += setup('/tick step 1')
+            ops.append({'op': 'tick', 'n': 1})
+        ops += setup('/execute store result score now piston_bounce run time query gametime',
+                     '/scoreboard players operation now piston_bounce -= start piston_bounce')
+        ops.append(cmd('/scoreboard players get now piston_bounce', expect_result=tick))
+        for index in range(len(cases)):
+            selector = f'@e[tag=piston_bounce_{index},limit=1]'
+            for field in ('Pos', 'Motion', 'OnGround'):
+                ops.append(cmd(f'/data get entity {selector} {field}'))
+            if tick in (1, 2):
+                x, z = 4 + index % 2 * 6, 4 + index // 2 * 6
+                ops.append(cmd(f'/execute if block {x} 205 {z} piston[facing=down,extended=true]',
+                               expect_result=1))
+                ops.append(cmd(f'/data get block {x} 204 {z} progress'))
+    return ops + setup('/kill @e[type=!player]', '/fill 0 200 0 15 208 15 air',
+                       '/scoreboard objectives remove piston_bounce', '/forceload remove 0 0',
+                       '/tick unfreeze', '/tp @s 8.5 230 8.5 0 0')