Merge pull request #351 from NanCunChild/fix/lmdb-reopen-dbi

Fix/lmdb reopen dbi

Author: NanCunChild

Cargo.toml

@@ -12,6 +12,11 @@ cargo-features = ["profile-rustflags"]
 [workspace]
 resolver = "2"
 
+# `tools/stress-bot` is a dev-only load-testing tool (azalea-based swarm client). It is its own
+# standalone workspace and excluded here so its large dependency tree never compiles as part of the
+# main build, CI, or release. See `tools/stress-bot/README.md`.
+exclude = ["tools/stress-bot"]
+
 #================= Members =================#
 members = [
     "src/bin",

src/bin/src/systems/fluids/mod.rs

@@ -249,14 +249,31 @@ pub fn seed_fluid_tick(
         state: state.clone(),
         dim,
     };
+    seed_fluid_tick_with_view(scheduler, &view, current_tick, pos);
+}
+
+/// Like [`seed_fluid_tick`] but reuses an existing [`WorldBlockView`] instead of constructing one.
+///
+/// Building a [`WorldBlockView`] clones the [`GlobalState`] handle (an `Arc` bump). Hot paths seed
+/// many positions back to back — applying a batch of changes wakes the six neighbours of every
+/// changed block — so building the view once and passing it in turns what was an `Arc` clone per
+/// seeded position into a single clone for the whole batch. The reads themselves are unaffected: the
+/// view always reflects the live world (it reads the chunk cache on each call), so reusing it across
+/// the intervening `apply_change` writes is equivalent to rebuilding it every time.
+fn seed_fluid_tick_with_view(
+    scheduler: &mut BlockTickScheduler,
+    view: &WorldBlockView,
+    current_tick: u64,
+    pos: BlockPos,
+) {
     let block = view.block_at(pos);
-    let Some(rules) = rules_for_block(block, dim.0) else {
+    let Some(rules) = rules_for_block(block, view.dim.0) else {
         debug_assert!(!is_fluid(block), "rules_for_block disagreed with is_fluid");
         return;
     };
     // A lava block already touching water solidifies almost immediately rather than waiting out its
     // spread cadence.
-    let delay = if would_react(pos, &view) {
+    let delay = if would_react(pos, view) {
         REACTION_DELAY
     } else {
         rules.tick_delay
@@ -290,12 +307,18 @@ pub fn seed_on_block_break(
 ) {
     let current = tick.get();
     let dim = *dim;
+    // One view for the whole batch of break events: each seed below would otherwise clone the state
+    // handle, and a break wakes seven positions (the block plus its six neighbours).
+    let view = WorldBlockView {
+        state: state.0.clone(),
+        dim,
+    };
     for event in events.read() {
         let pos = event.position;
         // The broken position itself (in case it is now exposed to a fluid) and its neighbours.
-        seed_fluid_tick(&mut scheduler.0, &state.0, dim, current, pos);
+        seed_fluid_tick_with_view(&mut scheduler.0, &view, current, pos);
         for neighbour in neighbours(pos) {
-            seed_fluid_tick(&mut scheduler.0, &state.0, dim, current, neighbour);
+            seed_fluid_tick_with_view(&mut scheduler.0, &view, current, neighbour);
         }
     }
 }
@@ -551,6 +574,14 @@ fn apply_changes(
     // Fallback cadence for re-ticking a changed block whose new state is not itself fluid.
     let fallback_delay = FluidRules::for_kind(FluidKind::Water, dim.0).tick_delay;
 
+    // One view for the whole batch. Neighbour waking below reads through it after each
+    // `apply_change` write; because the view reads the live chunk cache (never a snapshot), reusing
+    // it is equivalent to rebuilding it per seed but without the per-seed state-handle clone.
+    let view = WorldBlockView {
+        state: state.clone(),
+        dim,
+    };
+
     for change in changes {
         if !apply_change(state, dim, change) {
             continue;
@@ -569,10 +600,10 @@ fn apply_changes(
             scheduler.schedule(change.pos, TickKind::FluidSpread, current, delay);
         }
 
-        // Wake fluid neighbours so the update propagates outward. `seed_fluid_tick` skips
+        // Wake fluid neighbours so the update propagates outward. `seed_fluid_tick_with_view` skips
         // non-fluid neighbours and picks the reaction cadence for lava that now touches water.
         for neighbour in fluid_neighbours(change.pos) {
-            seed_fluid_tick(scheduler, state, dim, current, neighbour);
+            seed_fluid_tick_with_view(scheduler, &view, current, neighbour);
         }
     }
 }

src/lib/derive_macros/src/registries_packets/mod.rs

@@ -1,6 +1,9 @@
 use indexmap::IndexMap;
 use quote::quote;
 use serde_json::Value;
+use std::collections::HashMap;
+
+use craftflow_nbt::DynNBT;
 
 pub(crate) fn build_mapping(_: proc_macro::TokenStream) -> proc_macro::TokenStream {
     let json_file = include_bytes!("../../../../../assets/data/registry_packets.json");
@@ -12,7 +15,20 @@ pub(crate) fn build_mapping(_: proc_macro::TokenStream) -> proc_macro::TokenStre
         let mut packets = vec![];
         for (value_name, value) in &value_set {
             let mut nbt_data_buf = Vec::new();
-            craftflow_nbt::to_writer(&mut nbt_data_buf, &value).unwrap();
+            // The registry data is sourced from JSON, which cannot express NBT's distinct numeric
+            // tags: every JSON integer would otherwise serialise as a `Long` and every real as a
+            // `Double`. The vanilla client coerces numeric tags leniently and tolerates that, but
+            // strict clients deserialise the registry into typed structs and reject a field whose
+            // tag is not exactly what the schema expects (e.g. `dimension_type.height` must be an
+            // `Int`, not a `Long`). `dimension_type` is encoded through a schema-aware converter so
+            // every field carries its correct tag; all other registries keep the byte-for-byte
+            // output of the previous generic path until they, too, need a schema.
+            if reg_entry == "minecraft:dimension_type" {
+                let nbt = dimension_type_to_nbt(value);
+                craftflow_nbt::to_writer(&mut nbt_data_buf, &nbt).unwrap();
+            } else {
+                craftflow_nbt::to_writer(&mut nbt_data_buf, &value).unwrap();
+            }
             let kv = (value_name.clone(), nbt_data_buf);
             packets.push(kv);
         }
@@ -35,3 +51,93 @@ pub(crate) fn build_mapping(_: proc_macro::TokenStream) -> proc_macro::TokenStre
     }
     .into()
 }
+
+/// The NBT numeric tag a value must use, when it differs from the generic default (integers → Int,
+/// reals → Double). Only the tags actually needed by the current schema overrides are listed.
+#[derive(Clone, Copy)]
+enum NumTag {
+    Long,
+    Float,
+    Double,
+}
+
+/// The `dimension_type` fields whose vanilla NBT tag differs from the generic default. Every other
+/// field is an `Int` (e.g. `height`, `min_y`, `logical_height`), a `Byte` boolean, a `String`, or a
+/// nested compound, all of which the generic conversion already produces correctly.
+fn dimension_type_field_tag(field: &str) -> Option<NumTag> {
+    match field {
+        // Stored as `0`/`0.x` in JSON but a float in the dimension codec.
+        "ambient_light" => Some(NumTag::Float),
+        // A double in the dimension codec; JSON carries it as the integer `1`.
+        "coordinate_scale" => Some(NumTag::Double),
+        // A long in the dimension codec (optional; present for the Nether and the End).
+        "fixed_time" => Some(NumTag::Long),
+        _ => None,
+    }
+}
+
+/// Converts one `dimension_type` entry's JSON into correctly-typed NBT, applying the per-field tag
+/// overrides above to the entry's top-level fields. Nested values (e.g. the
+/// `monster_spawn_light_level` int-provider compound) use the generic conversion, which already
+/// yields `Int` for their integers.
+fn dimension_type_to_nbt(entry: &Value) -> DynNBT {
+    let obj = entry
+        .as_object()
+        .expect("dimension_type entry must be a JSON object");
+    let mut map = HashMap::with_capacity(obj.len());
+    for (key, value) in obj {
+        map.insert(
+            key.clone(),
+            json_to_nbt(value, dimension_type_field_tag(key)),
+        );
+    }
+    DynNBT::Compound(map)
+}
+
+/// Generic JSON → NBT conversion with Minecraft-appropriate defaults: integers become `Int` (not
+/// `Long`), reals become `Double`, and booleans become `Byte`. `force` overrides the numeric tag
+/// for a single scalar where the schema demands a non-default tag; it does not propagate into
+/// nested lists or compounds.
+fn json_to_nbt(value: &Value, force: Option<NumTag>) -> DynNBT {
+    match value {
+        Value::Bool(b) => DynNBT::Byte(i8::from(*b)),
+        Value::Number(n) => match force {
+            Some(NumTag::Float) => DynNBT::Float(num_f64(n) as f32),
+            Some(NumTag::Double) => DynNBT::Double(num_f64(n)),
+            Some(NumTag::Long) => DynNBT::Long(num_i64(n)),
+            None => {
+                if let Some(i) = n.as_i64() {
+                    // Default integers to Int (the common registry tag), widening to Long only when
+                    // the value genuinely does not fit in an i32.
+                    match i32::try_from(i) {
+                        Ok(v) => DynNBT::Int(v),
+                        Err(_) => DynNBT::Long(i),
+                    }
+                } else {
+                    DynNBT::Double(num_f64(n))
+                }
+            }
+        },
+        Value::String(s) => DynNBT::String(s.clone()),
+        Value::Array(items) => DynNBT::List(items.iter().map(|v| json_to_nbt(v, None)).collect()),
+        Value::Object(obj) => {
+            let mut map = HashMap::with_capacity(obj.len());
+            for (key, value) in obj {
+                map.insert(key.clone(), json_to_nbt(value, None));
+            }
+            DynNBT::Compound(map)
+        }
+        // Registries contain no JSON nulls; encode defensively as a zero byte rather than panicking.
+        Value::Null => DynNBT::Byte(0),
+    }
+}
+
+fn num_f64(n: &serde_json::Number) -> f64 {
+    n.as_f64()
+        .expect("registry numeric value is representable as f64")
+}
+
+fn num_i64(n: &serde_json::Number) -> i64 {
+    n.as_i64()
+        .expect("registry numeric value forced to an integer tag must be an integer")
+}

src/lib/storage/src/lmdb.rs

@@ -93,12 +93,18 @@ impl LmdbBackend {
         if let Some(db) = cache.get(table) {
             return Ok(Some(*db));
         }
-        let opened = {
-            let rtxn = self.env.read_txn()?;
-            self.env
-                .open_database::<U128<BigEndian>, Bytes>(&rtxn, Some(table))?
-            // `rtxn` is dropped at the end of this block, before the handle is cached or returned.
-        };
+        // The handle must be opened inside a *write* transaction that is then committed. LMDB ties a
+        // newly opened database handle to the transaction that opened it: it stays private to that
+        // transaction until commit, and is closed automatically if the transaction is aborted. A
+        // read transaction has no commit (dropping it aborts), so a handle opened in one would be
+        // closed the moment the transaction is dropped — leaving a dangling `dbi` that fails with
+        // `EINVAL` when reused in a later transaction. Opening in a committed write transaction
+        // promotes the handle into the shared environment so it remains valid for the env's lifetime.
+        let wtxn = self.env.write_txn()?;
+        let opened = self
+            .env
+            .open_database::<U128<BigEndian>, Bytes>(&wtxn, Some(table))?;
+        wtxn.commit()?;
         if let Some(db) = opened {
             cache.insert(table.to_string(), db);
         }
@@ -304,6 +310,36 @@ mod tests {
         remove_dir_all(path).unwrap();
     }
 
+    /// Reopening an existing environment must be able to read tables that were created by a
+    /// previous process. This exercises [`LmdbBackend::open_table`], the path taken for a table that
+    /// already exists on disk but is not yet in the in-process handle cache. A handle opened inside
+    /// a read transaction is closed when that transaction is aborted, so caching and reusing it
+    /// later fails with `EINVAL`; opening it in a committed write transaction keeps it valid.
+    #[test]
+    fn test_reopen_reads_existing_table() {
+        let path = tempdir().unwrap().keep();
+        let key = 12345678901234567890u128;
+        let value = vec![9, 8, 7, 6];
+        {
+            let backend =
+                LmdbBackend::initialize(Some(path.clone()), 10 * 1024 * 1024 * 1024).unwrap();
+            backend.create_table("test_table".to_string()).unwrap();
+            backend
+                .insert("test_table".to_string(), key, value.clone())
+                .unwrap();
+            backend.flush().unwrap();
+        }
+        // Fresh backend over the same path with an empty handle cache, mirroring a server restart.
+        {
+            let backend =
+                LmdbBackend::initialize(Some(path.clone()), 10 * 1024 * 1024 * 1024).unwrap();
+            assert!(backend.exists("test_table".to_string(), key).unwrap());
+            let retrieved_value = backend.get("test_table".to_string(), key).unwrap();
+            assert_eq!(retrieved_value, Some(value));
+        }
+        remove_dir_all(path).unwrap();
+    }
+
     #[test]
     fn test_batch_insert() {
         let path = tempdir().unwrap().keep();

src/lib/world/src/scheduler/mod.rs

@@ -64,17 +64,20 @@ impl ChunkTickQueue {
     }
 
     fn drain_due(&mut self, current_tick: u64, out: &mut Vec<ScheduledTick>) {
-        // Partition into due / not-yet-due, retaining the latter.
-        let mut remaining = Vec::with_capacity(self.pending.len());
-        for tick in self.pending.drain(..) {
+        // Compact in place: move due ticks into `out` and keep the rest. `retain` shifts the kept
+        // elements down without allocating, where the previous implementation allocated a fresh
+        // full-size buffer on every call — for every chunk, every tick, even when nothing was due.
+        // `seen` is borrowed separately from `pending` so the closure does not capture all of `self`.
+        let seen = &mut self.seen;
+        self.pending.retain(|tick| {
             if tick.target_tick <= current_tick {
-                self.seen.remove(&(tick.pos, tick.kind, tick.target_tick));
-                out.push(tick);
+                seen.remove(&(tick.pos, tick.kind, tick.target_tick));
+                out.push(*tick);
+                false
             } else {
-                remaining.push(tick);
+                true
             }
-        }
-        self.pending = remaining;
+        });
     }
 
     /// Drains at most `budget` due ticks into `out`, leaving any remaining due ticks queued (they
@@ -85,18 +88,20 @@ impl ChunkTickQueue {
         out: &mut Vec<ScheduledTick>,
         budget: usize,
     ) -> usize {
+        // Compact in place (see `drain_due`). Order is preserved, so once the budget is exhausted
+        // every still-due tick is simply kept and picked up by a later drain.
+        let seen = &mut self.seen;
         let mut taken = 0;
-        let mut remaining = Vec::with_capacity(self.pending.len());
-        for tick in self.pending.drain(..) {
+        self.pending.retain(|tick| {
             if taken < budget && tick.target_tick <= current_tick {
-                self.seen.remove(&(tick.pos, tick.kind, tick.target_tick));
-                out.push(tick);
+                seen.remove(&(tick.pos, tick.kind, tick.target_tick));
+                out.push(*tick);
                 taken += 1;
+                false
             } else {
-                remaining.push(tick);
+                true
             }
-        }
-        self.pending = remaining;
+        });
         taken
     }