horizon-prepare: review fixes for the parallel preparation

- A failed exporter compile fails every lane at once instead of being
  retried by each lane that waits for it.
- The gameplay facts exporter streams to a .partial file and moves it into
  place: a run stopped halfway no longer leaves a short file the next run
  would keep.
- pool::map takes no per-thread state (a jar clone costs two Arc counts);
  pool::lanes and map are tested at 1 job and several. With 1 job the first
  failing lane stops the rest, as documented now.
- The lanes' log and stage messages go through relay(), which a test now
  checks; block behaviour reuses export_on; the stamp inputs are worked out
  once; the fifth lane exists only when it splits; dead checks are gone.
- check_prepare_exporters.py also prepares with 2 jobs (the default path
  below 6 cores, 5 JVMs) and repeats that run, which must start no JVM and
  change no file.
- The docs say what HORIZON_PREPARE_JOBS does, that the split is not yet
  timed on 6 cores, and the split's measured ~400 MB.

Co-Authored-By: Claude Opus 5.5 <[email protected]>

crates/horizon-prepare/src/artifacts.rs

@@ -654,20 +654,16 @@ fn unpack_server(
         checksum,
     )?;
     let entries = bundle_entries(&mut archive, "META-INF/libraries.list")?;
-    let libraries = crate::pool::map(
-        &entries,
-        || Ok(archive.clone()),
-        |archive, (checksum, _, path, relative)| {
-            let destination = directory.join("libraries/server").join(path);
-            extract_verified(
-                archive,
-                &format!("META-INF/libraries/{relative}"),
-                &destination,
-                checksum.as_str(),
-            )?;
-            Ok(destination)
-        },
-    )?;
+    let libraries = crate::pool::map(&entries, |(checksum, _, path, relative)| {
+        let destination = directory.join("libraries/server").join(path);
+        extract_verified(
+            &mut archive.clone(),
+            &format!("META-INF/libraries/{relative}"),
+            &destination,
+            checksum.as_str(),
+        )?;
+        Ok(destination)
+    })?;
     Ok((game, libraries))
 }
 
@@ -727,11 +723,9 @@ pub fn import(
         imports.push((source, destination, sha1));
     }
     let hint = files.hint(&lock.version);
-    let client_libraries = crate::pool::map(
-        &imports,
-        || Ok(()),
-        |_, (source, destination, sha1)| import_copy(source, destination, sha1, "sha1", &hint),
-    )?;
+    let client_libraries = crate::pool::map(&imports, |(source, destination, sha1)| {
+        import_copy(source, destination, sha1, "sha1", &hint)
+    })?;
     write_import_metadata(
         root,
         lock,
@@ -753,11 +747,7 @@ fn write_import_metadata(
     log: Log,
 ) -> Result<(), String> {
     let sides = [("client", client), ("server", server)];
-    let inventories = crate::pool::map(
-        &sides,
-        || Ok(()),
-        |_, (_, jar)| crate::inventory::inventory(jar),
-    )?;
+    let inventories = crate::pool::map(&sides, |(_, jar)| crate::inventory::inventory(jar))?;
     for ((side, _), data) in sides.into_iter().zip(inventories) {
         let count = |key| data.get(key).cloned().unwrap_or(Value::Null);
         write_json(
@@ -929,22 +919,18 @@ pub fn verify(lock: &Lock) -> Result<Artifacts, String> {
     }
     let mut jars = HashMap::new();
     let sides = ["client", "server"];
-    let checked = crate::pool::map(
-        &sides,
-        || Ok(()),
-        |_, side| {
-            let jar = lock.directory.join(format!("jars/{side}.jar"));
-            let pinned = manifest
-                .get("downloads")
-                .and_then(|d| d.get(side))
-                .and_then(|d| d.get("sha1"))
-                .and_then(Value::as_str);
-            let good = pinned == Some(lock.pin(&format!("{side}_sha1"))?)
-                && jar.exists()
-                && Some(digest(&jar, "sha1")?.as_str()) == pinned;
-            Ok((good, jar))
-        },
-    )?;
+    let checked = crate::pool::map(&sides, |side| {
+        let jar = lock.directory.join(format!("jars/{side}.jar"));
+        let pinned = manifest
+            .get("downloads")
+            .and_then(|d| d.get(side))
+            .and_then(|d| d.get("sha1"))
+            .and_then(Value::as_str);
+        let good = pinned == Some(lock.pin(&format!("{side}_sha1"))?)
+            && jar.exists()
+            && Some(digest(&jar, "sha1")?.as_str()) == pinned;
+        Ok((good, jar))
+    })?;
     for (side, (good, jar)) in sides.into_iter().zip(checked) {
         if !good {
             return Err(format!(
@@ -964,17 +950,13 @@ pub fn verify(lock: &Lock) -> Result<Artifacts, String> {
     }
     jars.insert("server", game);
     let entries = bundle_entries(&mut archive, "META-INF/libraries.list")?;
-    let libraries = crate::pool::map(
-        &entries,
-        || Ok(()),
-        |_, (checksum, _, path, relative)| {
-            let library = lock.directory.join("libraries/server").join(path);
-            if !library.exists() || digest(&library, "sha256")? != *checksum {
-                return Err(format!("Dependency checksum changed: {relative}"));
-            }
-            Ok(library)
-        },
-    )?;
+    let libraries = crate::pool::map(&entries, |(checksum, _, path, relative)| {
+        let library = lock.directory.join("libraries/server").join(path);
+        if !library.exists() || digest(&library, "sha256")? != *checksum {
+            return Err(format!("Dependency checksum changed: {relative}"));
+        }
+        Ok(library)
+    })?;
     Ok(Artifacts {
         manifest,
         jars,
@@ -987,22 +969,18 @@ pub fn verify(lock: &Lock) -> Result<Artifacts, String> {
 /// player's launcher did not have is left out.
 pub fn client_libraries(lock: &Lock, manifest: &Value) -> Result<Vec<PathBuf>, String> {
     let artifacts = client_library_artifacts(manifest)?;
-    let libraries = crate::pool::map(
-        &artifacts,
-        || Ok(()),
-        |_, (name, artifact)| {
-            let (path, sha1) = artifact_fields(artifact)?;
-            let relative = safe_relative(path).ok_or("Unsafe client library path")?;
-            let library = lock.directory.join("libraries/client").join(&relative);
-            if runs_without(name) && !library.exists() {
-                return Ok(None);
-            }
-            if !library.exists() || digest(&library, "sha1")? != sha1 {
-                return Err(format!("Client dependency checksum changed: {path}"));
-            }
-            Ok(Some(library))
-        },
-    )?;
+    let libraries = crate::pool::map(&artifacts, |(name, artifact)| {
+        let (path, sha1) = artifact_fields(artifact)?;
+        let relative = safe_relative(path).ok_or("Unsafe client library path")?;
+        let library = lock.directory.join("libraries/client").join(&relative);
+        if runs_without(name) && !library.exists() {
+            return Ok(None);
+        }
+        if !library.exists() || digest(&library, "sha1")? != sha1 {
+            return Err(format!("Client dependency checksum changed: {path}"));
+        }
+        Ok(Some(library))
+    })?;
     Ok(libraries.into_iter().flatten().collect())
 }
 

crates/horizon-prepare/src/exporters.rs

@@ -123,49 +123,45 @@ pub(crate) fn classes(run: &Run) -> Result<PathBuf, String> {
         ("server", SERVER, &run.server_classpath),
         ("client", CLIENT, &run.client_classpath),
     ];
-    crate::pool::map(
-        &sides,
-        || Ok(()),
-        |_, (side, mains, cp)| {
-            (run.log)(&format!("Compiling {side} exporters (shared helpers once)"));
-            let started = std::time::Instant::now();
-            let classes = cached.join(side);
-            mkdir(&classes)?;
-            let log = run.log_file(&cached.join(format!("{side}.log")))?;
-            let mut args = vec![
-                "-encoding".into(),
-                "UTF-8".into(),
-                "-cp".into(),
-                cp.to_string(),
-                "-sourcepath".into(),
-                arg(&run.root.join("mods/horizon/src/test/java")),
-                "-d".into(),
-                arg(&classes),
-            ];
-            args.extend(mains.iter().map(|n| arg(&run.oracle(&format!("{n}.java")))));
-            jdk.javac(run.java_major, &args, &log)?;
-            let validate = vec![
-                "-Xverify:all".into(),
-                "-cp".into(),
-                format!(
-                    "{}{}{}",
-                    classes.display(),
-                    if cfg!(windows) { ";" } else { ":" },
-                    cp
-                ),
-                "dev.horizonlab.parity.PrepareWorker".into(),
-                "validate".into(),
-                arg(&classes),
-            ];
-            // No exporter executes here: resolve every class and its signatures against pinned jars.
-            run.jdk.java(&validate, Some(&run.local), &log)?;
-            (run.log)(&format!(
-                "Compiled {side} exporters{}",
-                crate::took(started)
-            ));
-            Ok(())
-        },
-    )?;
+    crate::pool::map(&sides, |(side, mains, cp)| {
+        (run.log)(&format!("Compiling {side} exporters (shared helpers once)"));
+        let started = std::time::Instant::now();
+        let classes = cached.join(side);
+        mkdir(&classes)?;
+        let log = run.log_file(&cached.join(format!("{side}.log")))?;
+        let mut args = vec![
+            "-encoding".into(),
+            "UTF-8".into(),
+            "-cp".into(),
+            cp.to_string(),
+            "-sourcepath".into(),
+            arg(&run.root.join("mods/horizon/src/test/java")),
+            "-d".into(),
+            arg(&classes),
+        ];
+        args.extend(mains.iter().map(|n| arg(&run.oracle(&format!("{n}.java")))));
+        jdk.javac(run.java_major, &args, &log)?;
+        let validate = vec![
+            "-Xverify:all".into(),
+            "-cp".into(),
+            format!(
+                "{}{}{}",
+                classes.display(),
+                if cfg!(windows) { ";" } else { ":" },
+                cp
+            ),
+            "dev.horizonlab.parity.PrepareWorker".into(),
+            "validate".into(),
+            arg(&classes),
+        ];
+        // No exporter executes here: resolve every class and its signatures against pinned jars.
+        run.jdk.java(&validate, Some(&run.local), &log)?;
+        (run.log)(&format!(
+            "Compiled {side} exporters{}",
+            crate::took(started)
+        ));
+        Ok(())
+    })?;
     let manifest = Value::Object(vec![
         ("key".into(), Value::Str(key.clone())),
         ("classes".into(), outputs(&cached)?),

crates/horizon-prepare/src/lib.rs

@@ -417,7 +417,8 @@ struct Run<'a> {
     server_classpath: String,
     /// The client jar and its libraries: the classpath of the client-side exporters.
     client_classpath: String,
-    bundles: Mutex<Option<PathBuf>>,
+    /// The compiled exporter classes, or why they didn't compile: one attempt for all lanes.
+    bundles: Mutex<Option<Result<PathBuf, String>>>,
     /// The exporter JVMs, which run at the same time: the server's, the client's, and on a
     /// machine with cores to spare a second server one for the generator's block behaviour.
     workers: [Mutex<Option<worker::Worker>>; 3],
@@ -438,13 +439,14 @@ impl Run<'_> {
         (self.stage)();
     }
 
-    /// The exporter classes, compiled once: a lane that needs them while another compiles waits.
+    /// The exporter classes, compiled once: a lane that needs them while another compiles waits,
+    /// and a failed compile fails every lane without compiling again.
     fn bundle(&self) -> Result<PathBuf, String> {
-        let mut bundle = self.bundles.lock().unwrap();
-        if bundle.is_none() {
-            *bundle = Some(exporters::classes(self)?);
-        }
-        Ok(bundle.clone().unwrap())
+        self.bundles
+            .lock()
+            .unwrap()
+            .get_or_insert_with(|| exporters::classes(self))
+            .clone()
     }
 
     /// Stops an exporter JVM once its lane needs it no more, to free its memory.
@@ -539,16 +541,13 @@ impl Run<'_> {
         what: &str,
         classpath: &str,
         classes: &Path,
-        sources: &[&Path],
         mains: &[&str],
         heap: &str,
         args: &[String],
         log: &Path,
     ) -> Result<(), String> {
         let worker = self.side(classpath);
-        self.export_on(
-            worker, what, classpath, classes, sources, mains, heap, args, log,
-        )
+        self.export_on(worker, what, classpath, classes, mains, heap, args, log)
     }
 
     /// `export`, on exporter JVM `worker`.
@@ -559,7 +558,6 @@ impl Run<'_> {
         what: &str,
         classpath: &str,
         classes: &Path,
-        _sources: &[&Path],
         mains: &[&str],
         heap: &str,
         args: &[String],
@@ -657,11 +655,9 @@ fn prepare_impl(
     };
     let manifest_path = resources.join("horizon-resources.json");
     let oracle = root.join(PARITY).join("VanillaHeightmapOracle.java");
-    let jar_hashes = pool::map(
-        &[&found.jars["client"], &found.jars["server"]],
-        || Ok(()),
-        |_, jar| digest(jar, "sha256"),
-    )?;
+    let jar_hashes = pool::map(&[&found.jars["client"], &found.jars["server"]], |jar| {
+        digest(jar, "sha256")
+    })?;
     let origin = Value::Object(vec![
         ("minecraft".into(), Value::Str(lock.version.clone())),
         ("client_sha256".into(), Value::Str(jar_hashes[0].clone())),
@@ -718,17 +714,7 @@ fn prepare_impl(
             check::write_stamp(&resources)?;
             Ok(resources)
         });
-        // The fraction done is the stages opened before the one opening.
-        let mut stages = 0;
-        for line in lines {
-            match line {
-                Some(line) => log(&line),
-                None => {
-                    progress(stages.min(STAGES) as f32 / STAGES as f32);
-                    stages += 1;
-                }
-            }
-        }
+        relay(lines, log, progress);
         work.join()
             .unwrap_or_else(|_| Err("The preparation stopped unexpectedly".into()))
     })?;
@@ -743,6 +729,21 @@ fn prepare_impl(
     Ok(result)
 }
 
+/// Hands the lanes' messages to the caller's `log` and `progress`: a line, or None for a stage
+/// opened. The fraction done is the stages opened before the one opening.
+fn relay(messages: std::sync::mpsc::Receiver<Option<String>>, log: Log, progress: Progress) {
+    let mut stages = 0;
+    for message in messages {
+        match message {
+            Some(line) => log(&line),
+            None => {
+                progress(stages.min(STAGES) as f32 / STAGES as f32);
+                stages += 1;
+            }
+        }
+    }
+}
+
 fn arg(path: &Path) -> String {
     path.display().to_string()
 }
@@ -751,11 +752,11 @@ fn arg(path: &Path) -> String {
 type Entries = Vec<(&'static str, Value)>;
 
 /// The exporters, the data generator and the asset copy, on four lanes that run at once:
-/// the server exporter JVM's steps in their old order (they share its state), the client
+/// the server exporter JVM's steps in a fixed order (they share its state), the client
 /// exporter JVM's then the command registries' JVM, the data generator's then the creative
 /// inventory's, and the copying, which needs no Java. At most three JVMs run at a time.
 /// From this many cores (`pool::jobs`) the generator's block behaviour gets its own server JVM.
-/// It pays a second Bootstrap and ~300 MB, which only pays off when the other lanes leave cores
+/// It pays a second Bootstrap and ~400 MB, which only pays off when the other lanes leave cores
 /// idle: a 6-core laptop finished its other JVMs 20 s before the server lane.
 const SPLIT_BLOCKS_JOBS: usize = 6;
 
@@ -840,7 +841,6 @@ fn steps(
                     server_cp
                 },
                 &run.local.join(classes),
-                &sources,
                 &[main],
                 heap,
                 &args,
@@ -870,7 +870,6 @@ fn steps(
                 "block states",
                 server_cp,
                 &root.join("classes"),
-                &[oracle],
                 &["VanillaHeightmapOracle"],
                 "512M",
                 &["export".into(), arg(&facts)],
@@ -891,7 +890,6 @@ fn steps(
                 "movement and attributes",
                 server_cp,
                 &export.join("classes"),
-                &[&movement, &attributes, oracle, &landing, &collision],
                 &["VanillaBlockLandingOracle", "VanillaEntityAttributeOracle"],
                 "768M",
                 &["export".into(), arg(&export)],
@@ -931,7 +929,6 @@ fn steps(
                 "damage and health",
                 server_cp,
                 &export.join("classes"),
-                &[&damage, &health, &attributes],
                 &["VanillaPlayerHealthOracle", "VanillaDamageSourceOracle"],
                 "768M",
                 &["export".into(), arg(&export)],
@@ -960,7 +957,6 @@ fn steps(
                 "collision shapes",
                 server_cp,
                 &export.join("classes"),
-                &[&collision],
                 &["VanillaCollisionOracle"],
                 "768M",
                 &["generate".into(), arg(&export)],
@@ -982,7 +978,6 @@ fn steps(
                 "face visibility",
                 server_cp,
                 &export.join("classes"),
-                &[&visibility, &collision, oracle],
                 &["VanillaFaceVisibilityOracle"],
                 "768M",
                 &["export".into(), arg(&export)],
@@ -1059,7 +1054,6 @@ fn steps(
                 "model bindings",
                 client_cp,
                 &root.join("client-classes"),
-                &[&binding, &helper],
                 &["VanillaBlockStateBindingOracle"],
                 "512M",
                 &["export".into(), arg(&model_states)],
@@ -1107,7 +1101,6 @@ fn steps(
                 "command registries",
                 server_cp,
                 &export.join("classes"),
-                &[&command_oracle],
                 &["VanillaCommandParseOracle"],
                 "1G",
                 &["registries".into(), arg(&export)],
@@ -1187,10 +1180,9 @@ fn steps(
         run.stage();
         (run.log)("Copying the client's assets and data");
         let started = Instant::now();
-        let log = |line: &str| (run.log)(line);
-        let mut indexed = sounds::Assets::load(repo, Some(root), minecraft, &run.inputs, &log);
+        let mut indexed = sounds::Assets::load(repo, Some(root), minecraft, &run.inputs, run.log);
         let (version, count) = extract_client_resources(&run.client, root, indexed.as_mut().ok())?;
-        sounds::optional_imports(indexed.as_mut().map_err(|e| e.as_str()), &log);
+        sounds::optional_imports(indexed.as_mut().map_err(|e| e.as_str()), run.log);
         (run.log)(&format!(
             "Copied the client's assets and data{}",
             took(started)
@@ -1208,20 +1200,17 @@ fn steps(
         ])
     });
 
-    // With cores to spare, the generator's block behaviour (a quarter of the server lane's
-    // time) runs on a second server JVM beside the first; it pays its own Bootstrap.
-    let blocks: pool::Lane<Entries> = Box::new(|| {
-        if split_blocks {
+    let mut lanes = vec![server, client, reports, files];
+    if split_blocks {
+        // With cores to spare, the generator's block behaviour (a quarter of the server lane's
+        // time) runs on a second server JVM beside the first; it pays its own Bootstrap.
+        lanes.push(Box::new(|| {
             worldgen_blocks(run, 2)?;
             run.close_worker(2);
-        }
-        Ok(Vec::new())
-    });
-
-    let entries: Entries = pool::lanes(vec![server, client, reports, files, blocks])?
-        .into_iter()
-        .flatten()
-        .collect();
+            Ok(Vec::new())
+        }));
+    }
+    let entries: Entries = pool::lanes(lanes)?.into_iter().flatten().collect();
     // Biome parameter tables come from the data generator reports above.
     run.stage();
     worldgen_tables(run)?;
@@ -1332,16 +1321,14 @@ fn extract_client_resources(
         });
         groups[group].push((name, target));
     }
-    pool::map(
-        &groups,
-        || Ok(client.clone()),
-        |client, group| {
-            for (name, target) in group {
-                extract_member(client, name, target)?;
-            }
-            Ok(())
-        },
-    )?;
+    pool::map(&groups, |group| {
+        // A clone shares the jar's parsed member list and reads at its own offset.
+        let mut client = client.clone();
+        for (name, target) in group {
+            extract_member(&mut client, name, target)?;
+        }
+        Ok(())
+    })?;
     Ok((version, count))
 }
 
@@ -1456,42 +1443,29 @@ fn worldgen_bundles(run: &Run) -> Result<(), String> {
     // Structure templates: gzip NBT files under structure/<namespace>/<path>.nbt.
     let prefix = "data/minecraft/structure/";
     let (server, client) = (open_zip(&run.server)?, open_zip(&run.client)?);
+    // zip_names gives each name once.
     let templates: Vec<String> = zip_names(&server)?
         .into_iter()
         .filter(|name| name.starts_with(prefix) && name.ends_with(".nbt"))
         .collect();
-    // Each name's file once (a repeated name gives the same member).
-    let mut seen = std::collections::HashSet::new();
-    let templates: Vec<String> = templates
-        .into_iter()
-        .filter(|name| seen.insert(name.clone()))
-        .collect();
-    pool::map(
-        &templates,
-        || Ok(server.clone()),
-        |server, name| {
-            let relative = member_path(&name[prefix.len()..], "Invalid structure path")?;
-            let data = zip_bytes(server, name)?;
-            write_if_changed(&out.join("structure/minecraft").join(relative), &data)
-        },
-    )?;
-    pool::map(
-        &WORLDGEN_BUNDLES,
-        || Ok((server.clone(), client.clone())),
-        |(server, client), (name, path)| {
-            let prefix = format!("data/minecraft/{path}/");
-            let resources = json_resources(server, &prefix)?;
-            if resources.as_object().is_none_or(<[_]>::is_empty)
-                || resources != json_resources(client, &prefix)?
-            {
-                return Err(format!("Reference worldgen resources disagree: {path}"));
-            }
-            write_text(
-                &out.join(format!("{name}.json")),
-                &pyjson::dumps(&resources, &pyjson::COMPACT),
-            )
-        },
-    )?;
+    pool::map(&templates, |name| {
+        let relative = member_path(&name[prefix.len()..], "Invalid structure path")?;
+        let data = zip_bytes(&mut server.clone(), name)?;
+        write_if_changed(&out.join("structure/minecraft").join(relative), &data)
+    })?;
+    pool::map(&WORLDGEN_BUNDLES, |(name, path)| {
+        let prefix = format!("data/minecraft/{path}/");
+        let resources = json_resources(&mut server.clone(), &prefix)?;
+        if resources.as_object().is_none_or(<[_]>::is_empty)
+            || resources != json_resources(&mut client.clone(), &prefix)?
+        {
+            return Err(format!("Reference worldgen resources disagree: {path}"));
+        }
+        write_text(
+            &out.join(format!("{name}.json")),
+            &pyjson::dumps(&resources, &pyjson::COMPACT),
+        )
+    })?;
     (run.log)(&format!(
         "Wrote the world generator's bundle{}",
         took(started)
@@ -1506,6 +1480,7 @@ fn worldgen_blocks(run: &Run, worker: usize) -> Result<(), String> {
     // Block behavior used by feature placement checks (survival, support, collision).
     // These come from the Java exporters of the simple-block parity run, or a direct export.
     let all_present = |dir: &Path| BLOCK_BEHAVIOR_FILES.iter().all(|n| dir.join(n).exists());
+    let inputs = stamp_inputs(run)?;
     let mut behavior = local.join("rust-simple-block-parity");
     if !all_present(&behavior) {
         behavior = local.join("horizon-client-block-behavior");
@@ -1514,30 +1489,24 @@ fn worldgen_blocks(run: &Run, worker: usize) -> Result<(), String> {
             .map(|n| behavior.join(n))
             .collect();
         let stamp = behavior.join("stamp.json");
-        let inputs = stamp_inputs(run)?;
         if !stamp_matches(&stamp, &inputs, &outputs) {
             export_block_behavior(run, worker, &behavior)?;
             write_stamp(&stamp, &inputs, &outputs)?;
         }
     }
-    if all_present(&behavior) {
-        mkdir(&blocks)?;
-        for name in BLOCK_BEHAVIOR_FILES {
-            copy(&behavior.join(name), &blocks.join(name))?;
-        }
-    }
     mkdir(&blocks)?;
+    for name in BLOCK_BEHAVIOR_FILES {
+        copy(&behavior.join(name), &blocks.join(name))?;
+    }
     let structures = local.join("horizon-client-structures");
     let output = blocks.join("structure-blocks.json");
     let stamp = structures.join("stamp.json");
-    let inputs = stamp_inputs(run)?;
     if !stamp_matches(&stamp, &inputs, std::slice::from_ref(&output)) {
         run.export_on(
             worker,
             "structure blocks",
             &run.server_classpath,
             &structures.join("classes"),
-            &[&run.oracle("VanillaStructureBlockExporter.java")],
             &["VanillaStructureBlockExporter"],
             "1G",
             &[arg(&output)],
@@ -1575,7 +1544,7 @@ fn stamp_matches(stamp: &Path, inputs: &Value, outputs: &[PathBuf]) -> bool {
 }
 
 fn stamp_outputs(outputs: &[PathBuf]) -> Result<Value, String> {
-    let hashes = pool::map(outputs, || Ok(()), |_, path| digest(path, "sha256"))?;
+    let hashes = pool::map(outputs, |path| digest(path, "sha256"))?;
     Ok(Value::Object(
         outputs
             .iter()
@@ -1650,31 +1619,16 @@ fn worldgen_tables(run: &Run) -> Result<(), String> {
 /// Feature placement needs them: without these files the generator skips every placed feature
 /// whose blocks are checked for survival or support (trees, flowers, grass, kelp, ...).
 fn export_block_behavior(run: &Run, worker: usize, export: &Path) -> Result<(), String> {
-    (run.log)("Exporting block behavior (VanillaSimpleBlockOracle)");
-    let fail = |e: String| format!("{e} while exporting block behavior");
-    // Compiled first, so the time is this step's (the JVM's start included).
-    run.bundle().map_err(fail)?;
-    let started = Instant::now();
-    let classes = export.join("classes");
-    mkdir(&classes)?;
-    let log_path = export.join("export.log");
-    let log = run.log_file(&log_path)?;
-    let fail = |e: String| {
-        format!(
-            "{e} while exporting block behavior; see {}",
-            log_path.display()
-        )
-    };
-    run.run(
+    run.export_on(
         worker,
-        "1G",
+        "block behavior",
         &run.server_classpath,
-        "VanillaSimpleBlockOracle",
+        &export.join("classes"),
+        &["VanillaSimpleBlockOracle"],
+        "1G",
         &["export".into(), arg(export)],
-        &log,
-        &log_path,
-    )
-    .map_err(fail)?;
+        &export.join("export.log"),
+    )?;
     let rules = survival::bind_definitions(
         &read_json(&export.join("survival-classes.json"))?,
         &read_json(&export.join("survival-placement-classes.json"))?,
@@ -1683,9 +1637,7 @@ fn export_block_behavior(run: &Run, worker: usize, export: &Path) -> Result<(),
     write_text(
         &export.join("survival-rules.json"),
         &pyjson::dumps(&rules, &pyjson::COMPACT),
-    )?;
-    (run.log)(&format!("Exported block behavior{}", took(started)));
-    Ok(())
+    )
 }
 
 #[cfg(test)]
@@ -1751,6 +1703,31 @@ mod tests {
         fs::remove_dir_all(dir).unwrap();
     }
 
+    #[test]
+    fn lanes_messages_reach_the_log_and_the_fraction_only_grows() {
+        let (send, messages) = std::sync::mpsc::channel();
+        for stage in 0..STAGES {
+            send.send(Some(format!("line {stage}"))).unwrap();
+            send.send(None).unwrap();
+        }
+        drop(send);
+        let lines = std::cell::RefCell::new(Vec::new());
+        let fractions = std::cell::RefCell::new(Vec::new());
+        relay(
+            messages,
+            &|line| lines.borrow_mut().push(line.to_string()),
+            &|done| fractions.borrow_mut().push(done),
+        );
+        let lines = lines.into_inner();
+        assert_eq!(lines.len() as u32, STAGES);
+        assert_eq!(lines[3], "line 3");
+        // Each stage reports the ones before it: the last stops short of 1, which only the end
+        // of `prepare` reports.
+        let fractions = fractions.into_inner();
+        let expected: Vec<f32> = (0..STAGES).map(|n| n as f32 / STAGES as f32).collect();
+        assert_eq!(fractions, expected);
+    }
+
     /// `--progress json` reaches 1 only at the end: `STAGES` counts the stages `steps` opens.
     #[test]
     fn steps_open_every_stage() {

crates/horizon-prepare/src/pool.rs

@@ -1,5 +1,5 @@
 //! Running independent work at once: the exporter JVMs of each side, the asset copy, hashing.
-//! `HORIZON_PREPARE_JOBS=1` runs everything one step after another, as before.
+//! `HORIZON_PREPARE_JOBS=1` runs everything one step after another.
 
 use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
 use std::sync::Mutex;
@@ -17,21 +17,23 @@ pub(crate) fn jobs() -> usize {
         })
 }
 
-/// `work` over every item on up to `jobs()` threads, each with its own `state` (an open jar),
-/// the results in the items' order. The first item that fails, in that order, gives the error;
-/// the threads stop taking items once one failed.
-pub(crate) fn map<T: Sync, S, R: Send>(
+/// `work` over every item on up to `jobs()` threads, the results in the items' order. The first
+/// item that fails, in that order, gives the error; the threads stop taking items once one failed.
+pub(crate) fn map<T: Sync, R: Send>(
     items: &[T],
-    state: impl Fn() -> Result<S, String> + Sync,
-    work: impl Fn(&mut S, &T) -> Result<R, String> + Sync,
+    work: impl Fn(&T) -> Result<R, String> + Sync,
 ) -> Result<Vec<R>, String> {
-    let threads = jobs().min(items.len());
+    map_with(jobs(), items, work)
+}
+
+fn map_with<T: Sync, R: Send>(
+    jobs: usize,
+    items: &[T],
+    work: impl Fn(&T) -> Result<R, String> + Sync,
+) -> Result<Vec<R>, String> {
+    let threads = jobs.min(items.len());
     if threads <= 1 {
-        let mut state = match items.is_empty() {
-            true => return Ok(Vec::new()),
-            false => state()?,
-        };
-        return items.iter().map(|item| work(&mut state, item)).collect();
+        return items.iter().map(work).collect();
     }
     let next = AtomicUsize::new(0);
     let failed = AtomicBool::new(false);
@@ -40,38 +42,35 @@ pub(crate) fn map<T: Sync, S, R: Send>(
         for _ in 0..threads {
             scope.spawn(|| {
                 let mut results = Vec::new();
-                match state() {
-                    Ok(mut state) => {
-                        while !failed.load(Ordering::Relaxed) {
-                            let i = next.fetch_add(1, Ordering::Relaxed);
-                            let Some(item) = items.get(i) else { break };
-                            let result = work(&mut state, item);
-                            failed.fetch_or(result.is_err(), Ordering::Relaxed);
-                            results.push((i, result));
-                        }
-                    }
-                    Err(e) => {
-                        failed.store(true, Ordering::Relaxed);
-                        results.push((0, Err(e)));
-                    }
+                while !failed.load(Ordering::Relaxed) {
+                    let i = next.fetch_add(1, Ordering::Relaxed);
+                    let Some(item) = items.get(i) else { break };
+                    let result = work(item);
+                    failed.fetch_or(result.is_err(), Ordering::Relaxed);
+                    results.push((i, result));
                 }
                 done.lock().unwrap().extend(results);
             });
         }
     });
     let mut done = done.into_inner().unwrap();
-    done.sort_by_key(|(i, result)| (*i, result.is_ok()));
+    done.sort_by_key(|(i, _)| *i);
     done.into_iter().map(|(_, result)| result).collect()
 }
 
 /// One sequence of steps that needs nothing from the others.
 pub(crate) type Lane<'a, R> = Box<dyn FnOnce() -> Result<R, String> + Send + 'a>;
 
-/// Runs every lane to its end, at once unless `jobs()` is 1; the first lane's error, in the
-/// lanes' order, wins. A lane that panics fails the preparation instead of the program.
+/// Runs the lanes at once, each to its end; the first lane's error, in the lanes' order, wins.
+/// With `jobs()` 1 they run one after another and the first error stops them. A lane that
+/// panics fails the preparation instead of the program.
 pub(crate) fn lanes<R: Send>(lanes: Vec<Lane<'_, R>>) -> Result<Vec<R>, String> {
+    lanes_with(jobs(), lanes)
+}
+
+fn lanes_with<R: Send>(jobs: usize, lanes: Vec<Lane<'_, R>>) -> Result<Vec<R>, String> {
     let stopped = |_| Err("A preparation step stopped unexpectedly".to_string());
-    if jobs() <= 1 {
+    if jobs <= 1 {
         return lanes
             .into_iter()
             .map(|lane| {
@@ -96,36 +95,23 @@ mod tests {
     #[test]
     fn results_keep_the_items_order_and_the_first_failure_wins() {
         let items: Vec<u32> = (0..500).collect();
-        let squares = map(&items, || Ok(()), |_, n| Ok(n * n)).unwrap();
-        assert_eq!(squares, items.iter().map(|n| n * n).collect::<Vec<_>>());
-        let failure = map(
-            &items,
-            || Ok(()),
-            |_, n| {
+        for jobs in [1, 4] {
+            let squares = map_with(jobs, &items, |n| Ok(n * n)).unwrap();
+            assert_eq!(squares, items.iter().map(|n| n * n).collect::<Vec<_>>());
+            let failure = map_with(jobs, &items, |n| {
                 if n % 7 == 3 {
                     Err(format!("bad {n}"))
                 } else {
                     Ok(*n)
                 }
-            },
-        );
-        assert_eq!(failure, Err("bad 3".into()));
-        assert_eq!(
-            map(&[1], || Err::<(), _>("no jar".into()), |_, n| Ok(*n)),
-            Err("no jar".into())
-        );
-        assert_eq!(
-            map(
-                &[] as &[u32],
-                || Err::<(), _>("unused".into()),
-                |_, n| Ok(*n)
-            ),
-            Ok(vec![])
-        );
+            });
+            assert_eq!(failure, Err("bad 3".into()));
+            assert_eq!(map_with(jobs, &[] as &[u32], |n| Ok(*n)), Ok(vec![]));
+        }
     }
 
     #[test]
-    fn every_lane_finishes_and_the_first_error_is_reported() {
+    fn lanes_run_at_once_to_their_end_and_the_first_error_is_reported() {
         let finished = AtomicUsize::new(0);
         let lane = |result: Result<u8, String>| -> Lane<'_, u8> {
             Box::new(|| {
@@ -133,16 +119,26 @@ mod tests {
                 result
             })
         };
-        assert_eq!(lanes(vec![lane(Ok(1)), lane(Ok(2))]), Ok(vec![1, 2]));
-        let result = lanes(vec![
-            lane(Ok(1)),
-            lane(Err("second".into())),
-            lane(Err("third".into())),
-            lane(Ok(4)),
-        ]);
-        assert_eq!(result, Err("second".into()));
-        assert_eq!(finished.load(Ordering::Relaxed), 6);
-        let panics: Lane<'_, u8> = Box::new(|| panic!("exporter thread"));
-        assert!(lanes(vec![lane(Ok(1)), panics]).is_err());
+        let four = || {
+            vec![
+                lane(Ok(1)),
+                lane(Err("second".into())),
+                lane(Err("third".into())),
+                lane(Ok(4)),
+            ]
+        };
+        assert_eq!(
+            lanes_with(4, vec![lane(Ok(1)), lane(Ok(2))]),
+            Ok(vec![1, 2])
+        );
+        assert_eq!(lanes_with(4, four()), Err("second".into()));
+        assert_eq!(finished.swap(0, Ordering::Relaxed), 6);
+        // One after another, the first error stops the rest.
+        assert_eq!(lanes_with(1, four()), Err("second".into()));
+        assert_eq!(finished.load(Ordering::Relaxed), 2);
+        for jobs in [1, 4] {
+            let panics: Lane<'_, u8> = Box::new(|| panic!("exporter thread"));
+            assert!(lanes_with(jobs, vec![lane(Ok(1)), panics]).is_err());
+        }
     }
 }

docs/launcher-preparation.md

@@ -25,7 +25,8 @@ pinned dependency checksums and compiler settings. Class checksums detect missin
 outputs. Compilation happens in a private staging folder before publishing a complete cache
 entry. Sources and paths are normalized so one release bundle works on all three platforms.
 
-Compatible exporters run sequentially through one server JVM and one client JVM. Both keep
+Compatible exporters run through one server JVM (two with 6 cores or more, below) and one
+client JVM, each worker's exporters in a fixed order, the workers side by side. They keep
 Minecraft's initialized built-in registries. Every task restores the original registry tags
 before replying; failures stop preparation. Collision exports share the first generated
 registry and collision files, encoded block states and shape grids within the server worker.
@@ -37,27 +38,33 @@ components as well as tags. The data generator also remains separate. Inspection
 its providers. Calling it inside a worker would shut down shared executors; incorporating
 its providers directly needs a separate comparison of registry and executor lifetimes.
 Fresh release preparation therefore starts five exporter/data-generator JVMs, down from 18
-(six with 6 cores or more, below; the release check runs that way), and invokes no compiler. `HORIZON_PREPARE_ISOLATED=1` retains separate exporter JVMs for
-output comparisons and diagnosis.
+(six with 6 cores or more, below), and invokes no compiler. `HORIZON_PREPARE_ISOLATED=1`
+retains separate exporter JVMs for output comparisons and diagnosis.
 
 Those JVMs run on four lanes at once, so at most three run at a time: the server worker's
 steps in their fixed order; the client worker's, then the command registries'; the data
 generator, then the creative tabs'; and the Rust-only copying (client assets and data, the
 world generator's bundle), spread over several threads. The world generator's tables, which
 need the block facts and the reports, and the manifest come after all four. A worker stops
-when its lane is done. With 6 cores or more (`HORIZON_PREPARE_JOBS`), the world generator's
-block behaviour and structure blocks, a quarter of the server lane, run on a fifth lane in a
-second server JVM: it costs a second Bootstrap and ~300 MB, and only pays off when the other
-lanes would leave cores idle. The two sides compile at once too. Imports and checks hash the pinned
-files on several threads; every file is still checked in full. `HORIZON_PREPARE_JOBS=1` runs
-it all one step after another (for a machine short on memory).
+when its lane is done. The two sides compile at once too. Imports and checks hash the pinned
+files on several threads; every file is still checked in full.
+
+`HORIZON_PREPARE_JOBS` is how many things run at once: the processors by default, at most 8.
+`1` runs it all one step after another (for a machine short on memory). From 6, the world
+generator's block behaviour and structure blocks, a quarter of the server lane, run on a fifth
+lane in a second server JVM: it costs a second Bootstrap and ~400 MB, and only pays off when
+the other lanes would leave cores idle. On a 6-core laptop they finished ~20 s before the server
+lane; the split itself is not yet timed on such a machine (on 3 cores it is slower, 44.9 s
+against 40.5 s, so it stays off there). The release check prepares both ways.
+
 Every JVM preparing starts (javac too, through `-J`) runs with `-XX:TieredStopAtLevel=1
 -XX:+UseSerialGC`: short batch jobs side by side lose more CPU to the C2 compiler than it wins
-back, and the serial collector keeps their heaps small (fresh: 105 s before, 63–66 s after, same
-output). Inside the server worker the collision exporter skips its 62 MB verification inputs,
-which only the parity harness reads, and the gameplay facts exporter streams its 56 MB file:
-the worker peaks at about 480 MB instead of 1.25 GB, and a fresh preparation at about 1.06 GB
-for all its processes instead of 1.47 GB. Threads reading one jar share its parsed member list.
+back, and the serial collector keeps their heaps small (fresh: 105 s before, 63–66 s after,
+same output). Inside the server worker the collision exporter skips its 62 MB verification
+inputs, which only the parity harness reads, and the gameplay facts exporter streams its 56 MB
+file: the worker peaks at about 480 MB instead of 1.25 GB, and a fresh preparation at about
+1.06 GB for all its processes instead of 1.47 GB (about 400 MB more with the second server
+JVM). Threads reading one jar share its parsed member list.
 
 A repeat run starts no JVM when nothing changed: each exporter is skipped by the hashes of its
 sources, its jar and its outputs. Most of those are in `horizon-resources.json`; the block
@@ -69,6 +76,8 @@ outputs' sha256), so the prepared files stay the same.
 for every file stable across the two isolated runs. For the three known noisy JSON files,
 it also requires semantic equality after normalizing object key order, opaque JVM lambda
 addresses and creative-tab tag set order. Other variation fails the release check.
+It compares two batched runs with them, with 6 jobs and with 2 (six JVMs and five), then
+repeats the second: that run must start no JVM and leave every file as it was.
 
 On Linux with Java 25, the output comparisons used 23,204 resource files per run.
 All runs used precompiled exporters:

docs/prepare-performance.md

@@ -2,7 +2,9 @@
 
 The reported preparation timings put sounds at 20–29 seconds with little CPU use, versus
 0.9–1.4 seconds for fresh client asset extraction. Prefer fewer file opens, reads and writes
-over parallelizing decompression.
+first. The exporter JVMs dominate a fresh preparation now; they and the asset copy run side by
+side, and extraction and hashing are spread over threads that share one parsed jar
+([launcher-preparation.md](launcher-preparation.md)).
 
 Preparation now shares one parsed asset index between panorama, font, pack and language imports.
 Its destination verification results live only for that preparation. A subsequent invocation

mods/horizon/src/test/java/dev/horizonlab/parity/VanillaGameplayFactsExporter.java

@@ -98,10 +98,15 @@ public final class VanillaGameplayFactsExporter {
         Bootstrap.bootStrap();
         // The same bytes Gson.toJson(JsonElement) gives for {"states": [...]}.
         var gson = new Gson();
-        try (var out = gson.newJsonWriter(Files.newBufferedWriter(Path.of(args[0]), StandardCharsets.UTF_8))) {
+        // Streamed beside the target and moved over it when whole: a run stopped halfway leaves
+        // no short file for the next preparation to keep.
+        var target = Path.of(args[0]);
+        var partial = target.resolveSibling(target.getFileName() + ".partial");
+        try (var out = gson.newJsonWriter(Files.newBufferedWriter(partial, StandardCharsets.UTF_8))) {
             out.beginObject().name("states").beginArray();
             states(gson, out);
             out.endArray().endObject();
         }
+        Files.move(partial, target, StandardCopyOption.REPLACE_EXISTING);
     }
 }

tests/test_prepare_exporters.py

@@ -12,7 +12,8 @@ from tools.check_prepare_exporters import verify
 class PrepareExportersTests(unittest.TestCase):
     def test_known_noise_does_not_mask_changed_resources(self):
         with tempfile.TemporaryDirectory() as temporary:
-            folders = [Path(temporary) / str(i) for i in range(3)]
+            # Two isolated runs, then two batched ones (with and without the second server JVM).
+            folders = [Path(temporary) / str(i) for i in range(4)]
             for i, folder in enumerate(folders):
                 folder.mkdir()
                 (folder / 'stable.nbt').write_bytes(b'unchanged')
@@ -29,9 +30,10 @@ class PrepareExportersTests(unittest.TestCase):
                     ('creative-inventory.json', b'{"tabs":[{},{"tags":[["a","b"]],"items":["dirt"]}]}'),
                     ('horizon-resources.json', b'{"schema":8}'),
                 ]:
-                    path = folders[2] / name
-                    original = path.read_bytes()
-                    path.write_bytes(changed)
-                    with self.subTest(resource=name), self.assertRaises(SystemExit):
-                        verify(folders)
-                    path.write_bytes(original)
+                    for folder in folders[2:]:
+                        path = folder / name
+                        original = path.read_bytes()
+                        path.write_bytes(changed)
+                        with self.subTest(resource=name, run=folder.name), self.assertRaises(SystemExit):
+                            verify(folders)
+                        path.write_bytes(original)

tools/check_prepare_exporters.py

@@ -1,5 +1,5 @@
 #!/usr/bin/env python3
-"""Compare two isolated preparations with a batched, runtime-only release preparation.
+"""Compare two isolated preparations with batched, runtime-only release preparations.
 
 Uses only locally imported pinned Minecraft files. Run after compile-exporters and copy its
 server/, client/ and manifest.json to exporters/. Byte differences are accepted only where
@@ -48,22 +48,24 @@ def normalize(path):
 
 
 def verify(folders):
-    a, b, batched = [hashes(folder) for folder in folders]
-    if a.keys() != b.keys() or a.keys() != batched.keys():
+    """`folders`: the two isolated runs' resources, then each batched run's."""
+    raw = [hashes(folder) for folder in folders]
+    a = raw[0]
+    if any(other.keys() != a.keys() for other in raw):
         raise SystemExit('Resource file sets differ')
     known = {'render-definitions.json', 'block-behaviours.json', 'creative-inventory.json'}
-    noise = [name for name in a if a[name] != b[name] or a[name] != batched[name]]
+    noise = [name for name in a if any(other[name] != a[name] for other in raw)]
     differences = []
     for name in noise:
         if name not in known:
             differences.append(name)
         else:
-            x, y, z = [normalize(folder / name) for folder in folders]
-            if x != y or x != z:
+            first, *rest = [normalize(folder / name) for folder in folders]
+            if any(value != first for value in rest):
                 differences.append(name)
     manifests = []
-    for folder, raw in zip(folders, [a, b, batched]):
-        replacements = {raw[name]: f'normalized:{name}' for name in noise}
+    for folder, hashed in zip(folders, raw):
+        replacements = {hashed[name]: f'normalized:{name}' for name in noise}
         def clean(node):
             if isinstance(node, str):
                 return replacements.get(node, node)
@@ -73,7 +75,7 @@ def verify(folders):
                 return {key: clean(item) for key, item in node.items()}
             return node
         manifests.append(clean(json.loads((folder / 'horizon-resources.json').read_text())))
-    if manifests[0] != manifests[1] or manifests[0] != manifests[2]:
+    if any(manifest != manifests[0] for manifest in manifests):
         differences.append('horizon-resources.json')
     print(f'Normalized exporter noise: {noise}')
     if differences:
@@ -96,7 +98,11 @@ def main():
         binary = frozen
         results = []
         times = []
-        for name in ['isolated-1', 'isolated-2', 'batched']:
+        # (name, HORIZON_PREPARE_JOBS, exporter/data-generator JVMs): with 6 jobs or more, block
+        # behaviour gets a second server JVM; below, most machines' path, it stays on the first.
+        runs = [('isolated-1', None, None), ('isolated-2', None, None),
+                ('batched', '6', 6), ('batched-2-jobs', '2', 5)]
+        for name, jobs, expected in runs:
             folder = work / name
             folder.mkdir()
             shutil.copyfile(ROOT / 'research-lock.json', folder / 'research-lock.json')
@@ -121,17 +127,17 @@ def main():
                 wrapper.chmod(0o755)
                 env['JAVA_HOME'] = str(runtime)
                 env['PATH'] = str(runtime / 'bin')
-                # As a machine with 6 cores or more runs it: block behaviour on a second server JVM.
-                env['HORIZON_PREPARE_JOBS'] = '6'
+                env['HORIZON_PREPARE_JOBS'] = jobs
             started = time.monotonic()
             with (work / f'{name}.log').open('w') as log:
                 subprocess.run([str(binary), '--root', str(folder)], cwd=folder, env=env,
                                stdout=log, stderr=subprocess.STDOUT, check=True, timeout=1200)
-            if name == 'batched':
+            if expected:
                 launches = [line for line in trace.read_text().splitlines() if line != '-version']
-                if len(launches) != 6:
-                    raise SystemExit(f'Expected 6 exporter/data-generator JVMs, got {len(launches)}')
-                print(f'Runtime-only preparation: {len(launches)} JVM launches', flush=True)
+                if len(launches) != expected:
+                    raise SystemExit(f'{name}: expected {expected} exporter/data-generator JVMs,'
+                                     f' got {len(launches)}')
+                print(f'{name}: {len(launches)} JVM launches', flush=True)
             text = (work / f'{name}.log').read_text()
             if 'Compiling ' in text:
                 raise SystemExit(f'{name} unexpectedly needed a compiler')
@@ -140,7 +146,19 @@ def main():
             times.append(round(time.monotonic() - started, 2))
             print(f'{name}: {times[-1]}s, prepared', flush=True)
         verify(results)
-        print(f'Isolated {times[:2]}s, batched {times[2]}s')
+        # Nothing changed since the last run: it starts no JVM and rewrites nothing.
+        before = hashes(resources)
+        trace.write_text('')
+        with (work / 'repeat.log').open('w') as log:
+            subprocess.run([str(binary), '--root', str(folder)], cwd=folder, env=env,
+                           stdout=log, stderr=subprocess.STDOUT, check=True, timeout=1200)
+        launches = [line for line in trace.read_text().splitlines() if line != '-version']
+        if launches:
+            raise SystemExit(f'A repeat preparation started {len(launches)} JVMs')
+        if hashes(resources) != before:
+            raise SystemExit('A repeat preparation changed the resources')
+        print('Repeat preparation: no JVM, same files')
+        print(f'Isolated {times[:2]}s, batched {times[2:]}s')
 
 
 if __name__ == '__main__':