Decode on the loader's own threads and only what is still claimed, and time the age of the presented state.

Image decode takes the pool it runs on as a parameter and every lane hands it the loader's. `ComputeTaskPool` is the frame's pool, with its own thread count and its own affinity, so a load stage that decoded on it ran on the frame's threads and competed with the frame for them. Material compile chunks move onto the same load pool instead of raw `std::thread::scope` threads: the pool's workers set their affinity to the process CPUs when they spawn, while a thread spawned from an `AsyncComputeTaskPool` worker inherits that pool's narrower set. The compile scope is explicitly non-cooperative, so the calling thread waits for its chunks rather than ticking the pool's queue. The `common_mp` FPV plan is held rather than enqueued, and decodes once the donor plans have merged. `prune_to` drops every demand whose rows the merge would not fill — the same predicate `apply` uses, so pruning changes when the question is asked and not what the catalog ends up holding — and `release_claims` hands the rows back when nothing survives, which is otherwise `apply`'s job and would leave later stages skipping every name the plan had claimed. The census counts `pruned_variants` and `pruned_rows` on their own, apart from duplicate claims and from discards, and the report line carries what was cut, what is left and what asking cost. `presented_state_age` and `presented_frames_behind` measure from a render frame's extract to the render graph's Finish set: how old the state a frame drew was, and how many main frames had opened since it was taken. They get their own table in the counter report, because an age divided by a frame is not a share of one, and a line in the frames section beside main/render overlap — what pipelining costs, next to what it buys. Both say where they stop: bevy presents after the graph schedule returns, so the present, the compositor and the panel are outside, and neither is an input-to-photon latency. `IW4L_FRAME_LATENCY` sets the surface's `desired_maximum_frame_latency` without a rebuild. The backend may clamp it, so the manifest records the number asked for and never a grant this process cannot read back. Env switches now read through one rule in `perf::switch`: absent, empty, `0`, `false` and `off` are off. `IW4L_MULTI_DRAW=0` therefore turns the indirect path off instead of enabling it, which is what a paired run across it needs. A lightmap page counts the bytes of all six of its images in `handed_bytes`, `bytes_total` and against the budget pacing the overlay frames, and a slot that takes another slot's handle no longer counts as having handed anything over. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-10-26 02:11:39 -04:00 · 2026-09-17 00:44:05 +03:00
parent 9856f91a98
commit 2fb3e4558b
20 changed files with 761 additions and 182 deletions
@@ -14,7 +14,7 @@ use bevy::prelude::*;
 use bevy::render::render_resource::{
    Extent3d, TextureDimension, TextureFormat, TextureViewDescriptor, TextureViewDimension,
 };
-use bevy::tasks::{ComputeTaskPool, TaskPool};
+use bevy::tasks::TaskPool;

 use crate::material_catalog::TS_2D;
 use crate::progress::LoadStage;
@@ -289,6 +289,7 @@ pub fn decode_material_color_maps(
    zone_ff: &Path,
    catalog: &mut MaterialDefinitions,
    stage: &LoadStage,
+    pool: &TaskPool,
 ) -> Result<MaterialImageStats, String> {
    let main = game_main_for_zone(zone_ff)?;
    let index = IwdIndex::open(&main)?;
@@ -302,7 +303,7 @@ pub fn decode_material_color_maps(
    stage.total(work.len() as u64);

    let images = &catalog.images;
-    let decoded = decode_requests_in_parallel(images, index.as_ref(), &work, stage);
+    let decoded = decode_requests_in_parallel(images, index.as_ref(), &work, stage, pool);

    for outcome in decoded {
        match outcome {
@@ -384,6 +385,7 @@ pub fn decode_color_or_2d_for_names(
    catalog: &mut MaterialDefinitions,
    names: impl IntoIterator<Item = impl AsRef<str>>,
    stage: &LoadStage,
+    pool: &TaskPool,
 ) -> Result<usize, String> {
    let work = requested_named_2d_slots(catalog, names);
    if work.is_empty() {
@@ -393,7 +395,7 @@ pub fn decode_color_or_2d_for_names(
    let index = IwdIndex::open(&main)?;
    stage.total(work.len() as u64);
    let images = &catalog.images;
-    let decoded = decode_requests_in_parallel(images, index.as_ref(), &work, stage);
+    let decoded = decode_requests_in_parallel(images, index.as_ref(), &work, stage, pool);
    let mut n = 0usize;
    for outcome in decoded {
        if let ImageOutcome::Decoded {
@@ -470,6 +472,7 @@ pub fn decode_catalog_images_from_iwd(
    catalog: &mut MaterialDefinitions,
    images: impl IntoIterator<Item = (usize, u8)>,
    stage: &LoadStage,
+    pool: &TaskPool,
 ) -> Result<usize, String> {
    let mut seen = std::collections::BTreeSet::new();
    let mut work: Vec<ImageRequest> = Vec::new();
@@ -493,7 +496,7 @@ pub fn decode_catalog_images_from_iwd(
    let index = IwdIndex::open(&main)?;
    stage.total(work.len() as u64);
    let catalog_images = &catalog.images;
-    let decoded = decode_requests_in_parallel(catalog_images, index.as_ref(), &work, stage);
+    let decoded = decode_requests_in_parallel(catalog_images, index.as_ref(), &work, stage, pool);
    let mut n = 0usize;
    for outcome in decoded {
        if let ImageOutcome::Decoded {
@@ -788,17 +791,23 @@ impl Drop for Flight {
    }
 }

+/// Decode this work on the pool the caller names.
+///
+/// The pool is a parameter because the right one is the loader's, and
+/// `asset_material` cannot ask for it: the dependency runs the other way.
+/// `ComputeTaskPool` is the frame's pool, with its own thread count and its
+/// own affinity; a load stage that decodes on it runs on the frame's threads
+/// and competes with the frame for them.
 fn decode_requests_in_parallel(
    images: &[AuthoredImage],
    index: &IwdIndex,
    work: &[ImageRequest],
    stage: &LoadStage,
+    pool: &TaskPool,
 ) -> Vec<ImageOutcome> {
    if work.is_empty() {
        return Vec::new();
    }
-    let pool = ComputeTaskPool::get_or_init(TaskPool::default);
-
    let chunk_size = work.len().div_ceil(pool.thread_num().max(1)).clamp(1, 64);

    pool.scope(|scope| {
@@ -2036,6 +2045,13 @@ pub struct ImageDemandPlan {
    /// is the normal case — several materials share an image — and the
    /// difference is what `duplicate_claims` reports.
    claimed_rows: usize,
+    /// Demands [`Self::prune_to`] dropped because the merged catalog had
+    /// already been answered for them. They are still claims this plan made,
+    /// so they stay in `canonical_variants`; what they are not is work, and
+    /// the census names them on their own rather than letting them fall into
+    /// the gap between two other numbers.
+    pruned_variants: usize,
+    pruned_rows: usize,
 }

 /// One image this plan has ready, and how it got it.
@@ -2069,6 +2085,8 @@ pub struct DecodedImageBatch {
    reused_variants: usize,
    claimed_rows: usize,
    canonical_variants: usize,
+    pruned_variants: usize,
+    pruned_rows: usize,
 }

 impl Drop for DecodedImageBatch {
@@ -2099,6 +2117,13 @@ pub struct ImageMergeCensus {
    pub canonical_variants: usize,
    pub prepared_variants: usize,
    pub duplicate_claims: usize,
+    /// Claims the plan gave up before decoding, because the merged catalog was
+    /// already answered for the name. Its own number: it is neither a
+    /// duplicate claim — two rows wanting one image — nor a discard, which is
+    /// a decode that happened and was thrown away. `canonical_variants` still
+    /// counts them, so `canonical - pruned` is what the decode was asked for.
+    pub pruned_variants: usize,
+    pub pruned_rows: usize,
    pub filled_rows: usize,
    pub already_decoded: usize,
    pub discarded_variants: usize,
@@ -2282,6 +2307,8 @@ impl ImageDemandPlan {
            demands: Vec::new(),
            requests: Vec::new(),
            claimed_rows: 0,
+            pruned_variants: 0,
+            pruned_rows: 0,
        }
    }

@@ -2293,10 +2320,78 @@ impl ImageDemandPlan {
        self.demands.is_empty()
    }

-    /// Distinct images this plan will decode. `len` is the same number; this
-    /// is the name the census uses for it.
+    /// Distinct images this plan claimed. Not what it will decode once it has
+    /// been pruned — `len` is that — because the claims it gave up are what a
+    /// reader is comparing the decode against.
    pub fn canonical_variants(&self) -> usize {
-        self.demands.len()
+        self.demands.len() + self.pruned_variants
+    }
+
+    /// Demands and rows dropped before the decode, because the merged catalog
+    /// already holds an answer for the name.
+    pub fn pruned(&self) -> (usize, usize) {
+        (self.pruned_variants, self.pruned_rows)
+    }
+
+    /// Drop every demand whose claim the merge will not take.
+    ///
+    /// `apply` fills exactly the rows that still carry this plan's id and have
+    /// nothing decoded in them; a name with no such row left is a decode whose
+    /// result `apply` would count and throw away. The two predicates are the
+    /// same one, deliberately: what survives here is what would have survived
+    /// there, so pruning changes when the question is asked and not what the
+    /// merged catalog ends up holding.
+    ///
+    /// It is only true of a catalog that is finished with this plan's rivals.
+    /// Called against a catalog another plan is still going to apply into, it
+    /// would cut names that plan is about to release.
+    pub fn prune_to(&mut self, catalog: &MaterialDefinitions) {
+        let id = self.id;
+        let mut wanted: HashMap<&str, usize> = HashMap::new();
+        for image in &catalog.images {
+            if image.pending_decode == Some(id) && image.decoded.is_none() {
+                *wanted.entry(image.name.as_str()).or_default() += 1;
+            }
+        }
+        let mut kept_rows = 0usize;
+        let mut index = 0usize;
+        while index < self.demands.len() {
+            match wanted.get(self.demands[index].name.as_str()) {
+                Some(rows) => {
+                    kept_rows += rows;
+                    index += 1;
+                }
+                None => {
+                    self.demands.swap_remove(index);
+                    self.requests.swap_remove(index);
+                    self.pruned_variants += 1;
+                }
+            }
+        }
+        self.pruned_rows = self.claimed_rows.saturating_sub(kept_rows);
+    }
+
+    /// Hand back every row this plan still holds a claim on, and say how many
+    /// there were.
+    ///
+    /// `apply` is what normally does this: it clears `pending_decode` on every
+    /// row carrying the plan's id, whether or not the plan had anything to put
+    /// there. A plan that [`Self::prune_to`] emptied never reaches `apply`, so
+    /// without this its rows keep a mark saying a decode is owed on them —
+    /// and [`requested_color_map_slots`] and [`requested_named_2d_slots`] read
+    /// that mark as "somebody else is already handling this name" and skip the
+    /// row. Every row still marked here was answered by another plan, which is
+    /// why the plan is empty; the count is that plan's `already_decoded`.
+    pub fn release_claims(&self, catalog: &mut MaterialDefinitions) -> usize {
+        let id = self.id;
+        let mut released = 0usize;
+        for image in &mut catalog.images {
+            if image.pending_decode == Some(id) {
+                image.pending_decode = None;
+                released += 1;
+            }
+        }
+        released
    }

    /// Catalog rows that asked for one of those images.
@@ -2331,7 +2426,12 @@ impl ImageDemandPlan {
    /// Decode this plan's images. `job` is the row this work reports into, and
    /// it is stamped *after* the memory ceiling lets the plan through, so the
    /// decode timer measures decoding and the wait before it is its own column.
-    pub fn run(self, stage: &LoadStage, job: asset_transport::Job) -> DecodedImageBatch {
+    pub fn run(
+        self,
+        stage: &LoadStage,
+        job: asset_transport::Job,
+        pool: &TaskPool,
+    ) -> DecodedImageBatch {
        if self.demands.is_empty() {
            return DecodedImageBatch::default();
        }
@@ -2354,7 +2454,9 @@ impl ImageDemandPlan {
                    reused_bytes: 0,
                    reused_variants: 0,
                    claimed_rows: self.claimed_rows,
-                    canonical_variants: self.demands.len(),
+                    canonical_variants: self.canonical_variants(),
+                    pruned_variants: self.pruned_variants,
+                    pruned_rows: self.pruned_rows,
                };
            }
        };
@@ -2370,7 +2472,8 @@ impl ImageDemandPlan {
            ..Default::default()
        };
        stage.total(work.len() as u64);
-        let outcomes = decode_requests_in_parallel(&self.demands, index.as_ref(), &work, stage);
+        let outcomes =
+            decode_requests_in_parallel(&self.demands, index.as_ref(), &work, stage, pool);
        let mut decoded = Vec::with_capacity(outcomes.len());
        let mut newly_prepared_bytes = 0u64;
        let mut reused_bytes = 0u64;
@@ -2426,7 +2529,9 @@ impl ImageDemandPlan {
            reused_bytes,
            reused_variants,
            claimed_rows: self.claimed_rows,
-            canonical_variants: self.demands.len(),
+            canonical_variants: self.canonical_variants(),
+            pruned_variants: self.pruned_variants,
+            pruned_rows: self.pruned_rows,
        }
    }
 }
@@ -2446,6 +2551,8 @@ impl DecodedImageBatch {
            canonical_variants: self.canonical_variants,
            prepared_variants: self.stats.decoded,
            duplicate_claims: self.claimed_rows.saturating_sub(self.canonical_variants),
+            pruned_variants: self.pruned_variants,
+            pruned_rows: self.pruned_rows,
            newly_prepared_bytes: self.newly_prepared_bytes,
            reused_bytes: self.reused_bytes,
            reused_variants: self.reused_variants,
@@ -2661,11 +2768,12 @@ pub fn plan_material_color_maps(
    zone_ff: &Path,
    catalog: &mut MaterialDefinitions,
    stage: &LoadStage,
+    pool: &TaskPool,
 ) -> (MaterialImageStats, ImageDemandPlan) {
    let mut plan = ImageDemandPlan::new(zone_ff);
    let requested = requested_color_map_slots(catalog);
    let inline = claim(catalog, &mut plan, requested);
-    let stats = decode_inline(catalog, &inline, stage);
+    let stats = decode_inline(catalog, &inline, stage, pool);
    (stats, plan)
 }

@@ -2697,16 +2805,18 @@ pub fn plan_color_or_2d_for_names(
    catalog: &mut MaterialDefinitions,
    names: impl IntoIterator<Item = impl AsRef<str>>,
    stage: &LoadStage,
+    pool: &TaskPool,
 ) -> usize {
    let requested = requested_named_2d_slots(catalog, names);
    let inline = claim(catalog, plan, requested);
-    decode_inline(catalog, &inline, stage).decoded
+    decode_inline(catalog, &inline, stage, pool).decoded
 }

 fn decode_inline(
    catalog: &mut MaterialDefinitions,
    work: &[ImageRequest],
    stage: &LoadStage,
+    pool: &TaskPool,
 ) -> MaterialImageStats {
    let mut stats = MaterialImageStats {
        requested: work.len(),
@@ -2718,7 +2828,7 @@ fn decode_inline(

    let index = IwdIndex(Arc::new(asset_transport::IwdIndex::default()));
    let images = &catalog.images;
-    for outcome in decode_requests_in_parallel(images, &index, work, stage) {
+    for outcome in decode_requests_in_parallel(images, &index, work, stage, pool) {
        match outcome {
            ImageOutcome::Decoded {
                image_index,
@@ -2870,4 +2980,112 @@ mod tests {
            (Winner::OtherSource, _)
        ));
    }
+    /// A row in each of the four states a claim can be in by the time the
+    /// merged catalog is finished with this plan's rivals, and the one
+    /// question `prune_to` is allowed to answer: would `apply` have filled it?
+    ///
+    /// The two predicates are meant to be the same one — `apply` fills the
+    /// rows still carrying the plan's id with nothing decoded in them, and
+    /// `prune_to` keeps exactly the demands those rows name. Written as an
+    /// equivalence rather than as four expected names so that an edit to
+    /// either side that drifts from the other fails here instead of showing up
+    /// as an image that never arrives.
+    #[test]
+    fn pruning_keeps_exactly_what_the_merge_would_have_filled() {
+        let mut catalog = MaterialDefinitions::default();
+        let mut plan = ImageDemandPlan::new(Path::new("zone.ff"));
+        let id = plan.id;
+
+        // Claimed by this plan and still unanswered: the decode is this plan's
+        // to do.
+        catalog.images.push(undecoded_row("wanted"));
+        // Two rows share one name. One of them was answered elsewhere; the
+        // other still wants it, so the name survives and only the row is lost.
+        catalog.images.push(undecoded_row("shared"));
+        catalog.images.push(undecoded_row("shared"));
+        // Claimed by this plan, then decoded by a donor that got there first:
+        // `apply` would count it `already_decoded` and throw this decode away.
+        catalog.images.push(undecoded_row("beaten"));
+        // A rival plan overwrote the claim, so this row is not this plan's to
+        // fill at all.
+        catalog.images.push(undecoded_row("stolen"));
+
+        // The same two steps `claim` takes: mark the row as owed to this plan,
+        // and hand the plan a copy of it to decode from.
+        for image in &mut catalog.images {
+            image.pending_decode = Some(id);
+            let mut owned = image.clone();
+            owned.decoded = None;
+            plan.push(owned, (0, false, false, false));
+        }
+        assert_eq!(
+            plan.len(),
+            4,
+            "four distinct names claimed across five rows"
+        );
+        assert_eq!(plan.claimed_rows(), 5);
+
+        catalog.images[2].decoded = Some(Image::default());
+        catalog.images[3].decoded = Some(Image::default());
+        catalog.images[4].pending_decode = Some(id + 1);
+
+        plan.prune_to(&catalog);
+
+        let would_fill: std::collections::BTreeSet<String> = catalog
+            .images
+            .iter()
+            .filter(|image| image.pending_decode == Some(id) && image.decoded.is_none())
+            .map(|image| image.name.as_str().to_owned())
+            .collect();
+        let survived: std::collections::BTreeSet<String> = plan
+            .demands
+            .iter()
+            .map(|demand| demand.name.as_str().to_owned())
+            .collect();
+        assert_eq!(
+            survived, would_fill,
+            "the decode and the merge disagree about who owns which name"
+        );
+
+        // `canonical_variants` still counts the claims the plan gave up, so a
+        // reader comparing the decode against what was asked for has both.
+        assert_eq!(plan.canonical_variants(), 4);
+        assert_eq!(
+            plan.pruned(),
+            (2, 3),
+            "beaten and stolen, plus the answered `shared` row"
+        );
+        assert_eq!(plan.len(), 2);
+    }
+
+    /// Nothing survives, so nobody ever calls `apply` — and `apply` is what
+    /// hands the rows back. Without `release_claims` they keep a mark that the
+    /// stages after this one read as "somebody else is handling this name".
+    #[test]
+    fn a_plan_that_prunes_to_nothing_still_releases_its_rows() {
+        let mut catalog = MaterialDefinitions::default();
+        let mut plan = ImageDemandPlan::new(Path::new("zone.ff"));
+        let id = plan.id;
+
+        catalog.images.push(undecoded_row("beaten"));
+        let mut owned = catalog.images[0].clone();
+        owned.decoded = None;
+        catalog.images[0].pending_decode = Some(id);
+        plan.push(owned, (0, false, false, false));
+
+        catalog.images[0].decoded = Some(Image::default());
+        plan.prune_to(&catalog);
+        assert!(
+            plan.is_empty(),
+            "a name a donor answered is not this plan's work"
+        );
+
+        assert_eq!(plan.release_claims(&mut catalog), 1);
+        assert_eq!(catalog.images[0].pending_decode, None);
+        assert_eq!(
+            plan.release_claims(&mut catalog),
+            0,
+            "releasing twice must not double-count"
+        );
+    }
 }
@@ -1076,6 +1076,7 @@ impl ZoneLane for Iw4Lane {
                path,
                &mut material_population,
                &stage,
+                crate::session_load::load_pool(),
            );
            report.push(format!(
                "common_mp materials: {} claimed for the merged pool, {} in-zone bodies decoded here ({} missing, {} unsupported)",
@@ -1089,6 +1090,7 @@ impl ZoneLane for Iw4Lane {
                &mut material_population,
                sink.tracers.named_materials(),
                &stage,
+                crate::session_load::load_pool(),
            );
            report.push(format!(
                "common_mp tracer beam images: {tracer_inline} in-zone TS_COLOR_MAP/TS_2D decoded, rest claimed"
@@ -1104,6 +1106,7 @@ impl ZoneLane for Iw4Lane {
                &mut material_population,
                fx_2d_names,
                &stage,
+                crate::session_load::load_pool(),
            );
            report.push(format!(
                "common_mp fx elem 2d images: {fx_inline} in-zone TS_COLOR_MAP/TS_2D decoded, rest claimed"
@@ -1304,7 +1307,7 @@ fn decode_map_material_images(
    glass_names: Vec<String>,
 ) -> Vec<String> {
    let mut report = Vec::new();
-    match decode_material_color_maps(path, catalog, &stage) {
+    match decode_material_color_maps(path, catalog, &stage, crate::session_load::load_pool()) {
        Ok(stats) => {
            report.push(format!(
                "IWD color/normal maps: {}/{} decoded, {} missing, {} unsupported from {} archives",
@@ -1321,7 +1324,13 @@ fn decode_map_material_images(
    }
    let mut names = fx_name_hints;
    names.extend(glass_names.iter().cloned());
-    match crate::material_images::decode_color_or_2d_for_names(path, catalog, names, &stage) {
+    match crate::material_images::decode_color_or_2d_for_names(
+        path,
+        catalog,
+        names,
+        &stage,
+        crate::session_load::load_pool(),
+    ) {
        Ok(n) => report.push(format!(
            "fx elem 2d images: {n} TS_COLOR_MAP/TS_2D decoded (unique Bound names)"
        )),
@@ -719,8 +719,12 @@ impl ZoneLane for Iw5Lane {
        let mut pending_images = None;
        if decode_color_maps {
            let stage = progress.stage("planning IW5 common_mp material images");
-            let (inline, plan) =
-                crate::material_images::plan_material_color_maps(path, &mut materials, &stage);
+            let (inline, plan) = crate::material_images::plan_material_color_maps(
+                path,
+                &mut materials,
+                &stage,
+                crate::session_load::load_pool(),
+            );
            report.push(format!(
                "common_mp IW5 IWD color maps: {} claimed for the merged pool, {} in-zone bodies decoded here ({} missing, {} unsupported)",
                plan.len(),
@@ -212,7 +212,12 @@ impl ZoneLane for T5Lane {
        let map_xmodels = std::mem::take(&mut sink.map_xmodels);
        let bodies = std::mem::take(&mut sink.bodies);
        let fpv_meshes = std::mem::take(&mut sink.fpv_meshes);
-        match decode_material_color_maps(path, &mut materials, &stage) {
+        match decode_material_color_maps(
+            path,
+            &mut materials,
+            &stage,
+            crate::session_load::load_pool(),
+        ) {
            Ok(stats) => {
                report.push(format!(
                "IWD color/normal maps: {}/{} decoded, {} missing, {} unsupported from {} archives",
@@ -696,8 +701,12 @@ impl ZoneLane for T5Lane {
        let mut pending_images = None;
        if decode_color_maps {
            let stage = progress.stage("planning T5 common_mp material images");
-            let (inline, plan) =
-                crate::material_images::plan_material_color_maps(path, &mut materials, &stage);
+            let (inline, plan) = crate::material_images::plan_material_color_maps(
+                path,
+                &mut materials,
+                &stage,
+                crate::session_load::load_pool(),
+            );
            report.push(format!(
                "common_mp T5 IWD color maps: {} claimed for the merged pool, {} in-zone bodies decoded here ({} missing, {} unsupported)",
                plan.len(),
@@ -471,10 +471,13 @@ pub async fn load_prepared_match(
            // The runtime common_mp walk is synchronous here, so its plan is
            // handed over the instant the walk puts it down rather than at the
            // bottom of the function.
-            common_images = spawn_image_plan(
+            // Held, not enqueued: every one of this plan's disputed claims
+            // was answered by a donor, and the donors have not finished. It
+            // decodes once the merged catalog can say which of its claims are
+            // still its own.
+            common_images = hold_image_plan(
                "common_mp FPV",
                census.pending_images.take(),
-                &progress,
                common_images_job,
            );
            shared_surfaces = census.shared_surfaces;
@@ -847,79 +850,24 @@ pub async fn load_prepared_match(
            .unwrap_or_else(|| "missing".into()),
    ));

-    for pending in [t5_images, foreign_images, bundle_images, common_images]
+    // The donors first, and only then the plan that claims the same names
+    // they do. Its claims are resolved against the catalog they leave behind
+    // rather than against the one it was built from, which is why it waited.
+    for pending in [t5_images, foreign_images, bundle_images]
        .into_iter()
        .flatten()
    {
-        let job = pending.job;
-        let (label, batch) = pending.join().await;
-        let requested = batch.stats.requested;
-        let missing = batch.stats.missing;
-        let unsupported = batch.stats.unsupported;
-        let first_gap = batch.stats.first_gap.clone();
-        let census = batch.apply(&mut global);
-        job.bytes(None, Some(census.final_cpu_bytes))
-            // What the plan prepared and what it served out of another plan's
-            // work, kept apart: only the first is work this plan did. What
-            // survived is `retained + discarded`, a check on the pair and never
-            // its definition.
-            .prepared(census.newly_prepared_bytes, census.reused_bytes)
-            .merged(
-                census.final_cpu_bytes,
-                census.discarded_decoded_bytes,
-                census.discard_line(),
-            );
-        report.push(format!(
-                "{label} claimed images: {}/{requested} into the merged pool ({} already decoded by an earlier source, {missing} missing, {unsupported} unsupported, {} claimed rows dropped by the merge)",
-                census.filled_rows, census.already_decoded, census.discarded_variants,
-            ));
-        report.push(format!(
-            "{label} image demand: claimed_rows={} canonical_variants={} prepared_variants={} duplicate_claims={} final_cpu_bytes={} newly_prepared_bytes={} reused_variants={} reused_bytes={} discarded_decoded_bytes={} discarded_same_payload={}",
-            census.claimed_rows,
-            census.canonical_variants,
-            census.prepared_variants,
-            census.duplicate_claims,
-            census.final_cpu_bytes,
-            census.newly_prepared_bytes,
-            census.reused_variants,
-            census.reused_bytes,
-            census.discarded_decoded_bytes,
-            census.discarded_same_payload,
-        ));
-        if let Some(line) = census.discard_line() {
-            report.push(format!("{label} image discard: {line}"));
-        }
-        // Who answered for the claims this plan prepared and the merge threw
-        // away. A count by reason says how much was wasted; this says by
-        // whom, which is what decides whether the claim could have been
-        // resolved before it was prepared.
-        if !census.disputed_winners.is_empty() {
-            report.push(format!(
-                "{label} image dispute: {}",
-                census
-                    .disputed_winners
-                    .iter()
-                    .map(|winner| format!(
-                        "{}={} won_by={} first={}",
-                        winner.kind,
-                        winner.claims,
-                        winner
-                            .plan
-                            .map_or_else(|| "none".to_owned(), |plan| format!("plan{plan}")),
-                        winner.first,
-                    ))
-                    .collect::<Vec<_>>()
-                    .join("; "),
-            ));
-        }
-        if let Some(gap) = first_gap {
-            report.push(format!("{label} claimed image gap: {gap}"));
-        }
+        apply_image_batch(&mut global, pending, &mut report).await;
+    }
+    if let Some(held) = common_images
+        && let Some(pending) = held.prune_then_enqueue(&mut global, &progress, &mut report)
+    {
+        apply_image_batch(&mut global, pending, &mut report).await;
    }

    if let Ok(path) = &zone_ff {
        let stage = progress.stage("decoding merged material images");
-        match crate::decode_material_color_maps(path, &mut global, &stage) {
+        match crate::decode_material_color_maps(path, &mut global, &stage, load_pool()) {
            Ok(stats) => report.push(format!(
                "merged material images: {}/{} decoded, {} missing, {} unsupported",
                stats.decoded, stats.requested, stats.missing, stats.unsupported
@@ -997,6 +945,7 @@ pub async fn load_prepared_match(
                &mut global,
                requested,
                &stage,
+                load_pool(),
            ) {
                Ok(n) => report.push(format!(
                    "IWD light attenuation: decoded {n} of {want} GfxLightDef images (Image_LoadFromIwi; empty payload is not a host ramp)"
@@ -1020,6 +969,7 @@ pub async fn load_prepared_match(
            &mut global,
            common_tracers.named_materials(),
            &stage,
+            load_pool(),
        ) {
            Ok(n) => report.push(format!(
                "tracer beam images after absorb: {n} TS_COLOR_MAP/TS_2D decoded"
@@ -1212,6 +1162,7 @@ pub async fn load_prepared_match(
                    &mut global,
                    missing.iter(),
                    &stage,
+                    load_pool(),
                ) {
                    Ok(n) => report.push(format!(
                        "fx elem 2d images after absorb: {n} TS_COLOR_MAP/TS_2D decoded"
@@ -1720,8 +1671,8 @@ fn walk_foreign_material_common(
    ) else {
        return (MaterialCatalog::default(), None, report);
    };
-    let pending_images =
-        spawn_image_plan("IW5 common_mp", census.pending_images.take(), progress, job);
+    let pending_images = hold_image_plan("IW5 common_mp", census.pending_images.take(), job)
+        .map(|held| held.enqueue(progress));
    let materials = census.material_population;
    report.extend(census.report);
    report.push(format!(
@@ -1762,6 +1713,85 @@ fn capture_common_zone(
    }
 }

+/// Join one image plan, merge it into the catalog and write its census down.
+///
+/// Separate from the caller because the plans are no longer one list: the
+/// donors are applied first so that the plan claiming the same names can be
+/// pruned against the result, and both halves merge a batch exactly the same
+/// way.
+async fn apply_image_batch(
+    global: &mut crate::MaterialDefinitions,
+    pending: PendingImages,
+    report: &mut Vec<String>,
+) {
+    let job = pending.job;
+    let (label, batch) = pending.join().await;
+    let requested = batch.stats.requested;
+    let missing = batch.stats.missing;
+    let unsupported = batch.stats.unsupported;
+    let first_gap = batch.stats.first_gap.clone();
+    let census = batch.apply(global);
+    job.bytes(None, Some(census.final_cpu_bytes))
+        // What the plan prepared and what it served out of another plan's
+        // work, kept apart: only the first is work this plan did. What
+        // survived is `retained + discarded`, a check on the pair and never
+        // its definition.
+        .prepared(census.newly_prepared_bytes, census.reused_bytes)
+        .merged(
+            census.final_cpu_bytes,
+            census.discarded_decoded_bytes,
+            census.discard_line(),
+        );
+    report.push(format!(
+            "{label} claimed images: {}/{requested} into the merged pool ({} already decoded by an earlier source, {missing} missing, {unsupported} unsupported, {} claimed rows dropped by the merge)",
+            census.filled_rows, census.already_decoded, census.discarded_variants,
+        ));
+    report.push(format!(
+        "{label} image demand: claimed_rows={} canonical_variants={} prepared_variants={} duplicate_claims={} pruned_variants={} pruned_rows={} final_cpu_bytes={} newly_prepared_bytes={} reused_variants={} reused_bytes={} discarded_decoded_bytes={} discarded_same_payload={}",
+        census.claimed_rows,
+        census.canonical_variants,
+        census.prepared_variants,
+        census.duplicate_claims,
+        census.pruned_variants,
+        census.pruned_rows,
+        census.final_cpu_bytes,
+        census.newly_prepared_bytes,
+        census.reused_variants,
+        census.reused_bytes,
+        census.discarded_decoded_bytes,
+        census.discarded_same_payload,
+    ));
+    if let Some(line) = census.discard_line() {
+        report.push(format!("{label} image discard: {line}"));
+    }
+    // Who answered for the claims this plan prepared and the merge threw
+    // away. A count by reason says how much was wasted; this says by
+    // whom, which is what decides whether the claim could have been
+    // resolved before it was prepared.
+    if !census.disputed_winners.is_empty() {
+        report.push(format!(
+            "{label} image dispute: {}",
+            census
+                .disputed_winners
+                .iter()
+                .map(|winner| format!(
+                    "{}={} won_by={} first={}",
+                    winner.kind,
+                    winner.claims,
+                    winner
+                        .plan
+                        .map_or_else(|| "none".to_owned(), |plan| format!("plan{plan}")),
+                    winner.first,
+                ))
+                .collect::<Vec<_>>()
+                .join("; "),
+        ));
+    }
+    if let Some(gap) = first_gap {
+        report.push(format!("{label} claimed image gap: {gap}"));
+    }
+}
+
 /// An image plan already on the pool, and the job row that records when it got
 /// there. The producer owns both from the moment its plan is ready: the walk
 /// that discovers the demand hands it over itself rather than carrying it back
@@ -1779,18 +1809,27 @@ impl PendingImages {
    }
 }

-/// Put a ready plan on the load pool now.
+/// A plan whose walk has finished, holding the job row that records when it
+/// did.
 ///
 /// `job` is opened by the producer when it starts looking, so the row carries
 /// the whole story: discovered → ready → enqueued → started → finished →
 /// joined. The gap this exists to expose is `ready → enqueued`; a timer that
-/// starts when a worker picks the job up cannot see it at all.
-fn spawn_image_plan(
+/// starts when a worker picks the job up cannot see it at all — and a plan
+/// that waits for the catalog before it decodes spends that whole wait here,
+/// in the open, rather than inside a decode timer.
+struct HeldImagePlan {
+    label: &'static str,
+    plan: ImageDemandPlan,
+    job: load_jobs::Job,
+}
+
+/// Take a finished walk's plan, if it wants anything at all.
+fn hold_image_plan(
    label: &'static str,
    plan: Option<ImageDemandPlan>,
-    progress: &LoadProgress,
    job: load_jobs::Job,
-) -> Option<PendingImages> {
+) -> Option<HeldImagePlan> {
    let plan = plan.filter(|plan| !plan.is_empty())?;
    // The census names the winner of a disputed claim by plan id, so the id
    // and the label are printed together once, here, where both are known.
@@ -1804,19 +1843,76 @@ fn spawn_image_plan(
    let job = job
        .canonical(label)
        .items(plan.canonical_variants() as u64)
-        .plan_ready()
-        .enqueued();
-    let progress = progress.clone();
+        .plan_ready();
+    Some(HeldImagePlan { label, plan, job })
+}

-    let task = load_pool().spawn(async move {
-        job.started();
-        let stage = progress.stage(format!("decoding {label} material images"));
-        let batch = plan.run(&stage, job);
-        drop(stage);
-        job.finished();
-        (label, batch)
-    });
-    Some(PendingImages { job, task })
+impl HeldImagePlan {
+    /// Put the plan on the load pool now, with every claim it made.
+    fn enqueue(self, progress: &LoadProgress) -> PendingImages {
+        let Self { label, plan, job } = self;
+        let job = job.enqueued();
+        let progress = progress.clone();
+        let task = load_pool().spawn(async move {
+            job.started();
+            let stage = progress.stage(format!("decoding {label} material images"));
+            // The plan splits itself across this same pool. The worker this
+            // task is on joins that scope rather than blocking on it, so a
+            // plan does not cost a load worker to supervise it.
+            let batch = plan.run(&stage, job, load_pool());
+            drop(stage);
+            job.finished();
+            (label, batch)
+        });
+        PendingImages { job, task }
+    }
+
+    /// Resolve the plan's claims against the merged catalog, then decode what
+    /// is left of it.
+    ///
+    /// The catalog has to be one every rival plan has already applied into: a
+    /// row still carrying this plan's id with nothing decoded in it is a row
+    /// this plan will fill, and a name with no such row left is a decode whose
+    /// result the merge would count and throw away. Asked earlier, the same
+    /// question cuts names a plan still in flight is about to release.
+    ///
+    /// Returns `None` when nothing survives, which is a plan that never runs
+    /// rather than an empty one that does.
+    fn prune_then_enqueue(
+        mut self,
+        catalog: &mut crate::MaterialDefinitions,
+        progress: &LoadProgress,
+        report: &mut Vec<String>,
+    ) -> Option<PendingImages> {
+        let resolving = std::time::Instant::now();
+        self.plan.prune_to(catalog);
+        let resolved_ms = resolving.elapsed().as_secs_f32() * 1000.0;
+        let (variants, rows) = self.plan.pruned();
+        // The cost of asking is on the line beside what it saved: this runs
+        // on the consumer, between the last donor's apply and the decode, so
+        // it is serial time the walk pays whatever the decode then skips.
+        report.push(format!(
+            "{} image claims resolved before decode: {} of {} variants cut ({rows} claimed rows), {} left to decode, {resolved_ms:.1}ms to resolve",
+            self.label,
+            variants,
+            self.plan.canonical_variants(),
+            self.plan.len(),
+        ));
+        if self.plan.is_empty() {
+            // Nothing left to decode means nothing will ever call `apply`, and
+            // `apply` is what hands the claimed rows back. Do it here instead,
+            // or the rows keep a mark saying a decode is owed on them and the
+            // stages after this one skip every name the plan had claimed.
+            let released = self.plan.release_claims(catalog);
+            report.push(format!(
+                "{} image plan dropped before decode: every claim was answered elsewhere, {released} claimed rows handed back",
+                self.label,
+            ));
+            self.job.items(0).enqueued().started().finished().joined();
+            return None;
+        }
+        Some(self.enqueue(progress))
+    }
 }

 enum ForeignCommonWork {
@@ -1884,12 +1980,8 @@ fn walk_iw5_weapon_bundle(
    // Patch A. The bundle's images used to wait here until the consumer had
    // finished the synchronous `common_mp` walk and got round to unpacking
    // this tuple — 2.4 s of ready work with nobody holding it.
-    let pending_images = spawn_image_plan(
-        "IW5 weapon bundle",
-        census.pending_images.take(),
-        progress,
-        job,
-    );
+    let pending_images = hold_image_plan("IW5 weapon bundle", census.pending_images.take(), job)
+        .map(|held| held.enqueue(progress));
    report.extend(census.report);
    report.push(format!(
        "iw5 weapons: path={} ids={} gun_named={} fpv={} world_guns={} xanims={} materials={}",
@@ -1942,8 +2034,8 @@ fn walk_shared_iw5_common(
    // One capture serves the material seed and the weapon bundle, so there is
    // one plan here, not two: the donor is walked once and its images are
    // claimed once.
-    let pending_images =
-        spawn_image_plan("IW5 common_mp", census.pending_images.take(), progress, job);
+    let pending_images = hold_image_plan("IW5 common_mp", census.pending_images.take(), job)
+        .map(|held| held.enqueue(progress));
    report.extend(census.report);
    let materials = census.material_population;
    report.push(format!(
@@ -2073,8 +2165,8 @@ fn walk_t5_weapon_common(
        }
    };
    let mut census = lane(image.game).load_common_mp(&donor, &image, progress, true, material_seed);
-    let pending_images =
-        spawn_image_plan("T5 common_mp", census.pending_images.take(), progress, job);
+    let pending_images = hold_image_plan("T5 common_mp", census.pending_images.take(), job)
+        .map(|held| held.enqueue(progress));
    report.extend(census.report);
    report.push(format!(
        "t5 weapon common: path={} weapons={} fpv={} world_guns={} materials={} xanims={} fx={}",
@@ -116,6 +116,8 @@ pub(crate) fn render(wall: &SpanStats, out: &mut Vec<String>) {
            .to_owned(),
    );

+    age_table(&stats, frames, out);
+
    out.push(String::new());
    out.push("  inside Present and Ui, CPU (ms)".to_owned());
    ms_table(&stats, &PHASE_MS, wall, frames, Share::Frame, out);
@@ -163,6 +165,69 @@ pub(crate) fn render(wall: &SpanStats, out: &mut Vec<String>) {
    never_sampled(out);
 }

+/// How old the state on the surface was, in its own table because every column
+/// the other tables carry would be a lie here.
+///
+/// `presented_state_age` is when the work happened, not how long anything
+/// took, so it has no `per frame` and no `%frame`: an age divided by a frame
+/// is not a share of one, and among the render stages it reads as one — which
+/// is the input-to-photon reading the frame section forbids.
+/// `presented_frames_behind` is the same quantity counted in frames, so it
+/// belongs beside it rather than in the work table, where its `total` would be
+/// a sum of the number one.
+///
+/// Silent when the counter never fired at all, which is a run that presented
+/// nothing rather than a run that presented in line with the main world: a
+/// serialised run still has an age here, it just answers zero frames behind.
+/// A MISS row would read as a measurement that came back empty rather than as
+/// a question the run never asked.
+fn age_table(stats: &[CounterStats], frames: u64, out: &mut Vec<String>) {
+    let Some(age) = find(stats, Counter::RenderPresentedStateAgeMs) else {
+        return;
+    };
+    out.push(String::new());
+    out.push("  presented state age (ms) — when, not how long".to_owned());
+    let mut table = Table::new(
+        4,
+        &[
+            ("counter", Align::Left),
+            ("n", Align::Right),
+            ("n/frame", Align::Right),
+            ("avg", Align::Right),
+            ("p50", Align::Right),
+            ("p95", Align::Right),
+            ("p99", Align::Right),
+            ("max", Align::Right),
+        ],
+    );
+    debug_assert_eq!(age.unit, Unit::Milliseconds);
+    table.row([
+        age.counter.name().to_owned(),
+        age.samples.to_string(),
+        opt2(age.per_frame_samples(frames)),
+        num2(age.avg()),
+        num2(age.p50),
+        num2(age.p95),
+        num2(age.p99),
+        num2(age.max),
+    ]);
+    table.render(out);
+    out.push(
+        "    the wall from this frame's extract — the main world stalled, so nothing moved across it — to the render graph's Finish set. It is not input-to-photon: the compositor, the queue behind it and the panel are all outside this process, and `desired_maximum_frame_latency` is a hint the backend may clamp, so how many frames the GPU is really allowed in flight is not in here either."
+            .to_owned(),
+    );
+    match find(stats, Counter::RenderPresentedFramesBehind) {
+        Some(behind) => out.push(format!(
+            "    `presented_frames_behind` on the same frames: p50 {:.0}, max {:.0}. It is the age in frames rather than in milliseconds — zero is a render world running in line with the main one, and one or more is a pipelined render world presenting an older state. Read it as the arm the run took, not as a distribution: with pipelining on it is the same number nearly every frame.",
+            behind.p50, behind.max,
+        )),
+        None => out.push(
+            "    `presented_frames_behind` MISS — nobody counted, which is not the same as zero frames behind."
+                .to_owned(),
+        ),
+    }
+}
+
 /// Which HUD slice was the largest, not only how large. An index is not a
 /// duration and has no business in a millisecond table, so it gets a sentence:
 /// the reader who has just seen a HUD stage take a third of the frame should
@@ -191,6 +256,10 @@ fn hud_stage(stats: &[CounterStats], out: &mut Vec<String>) {
 fn untabled(stats: &[CounterStats], out: &mut Vec<String>) {
    let tabled: Vec<Counter> = CPU_MS
        .into_iter()
+        .chain([
+            Counter::RenderPresentedStateAgeMs,
+            Counter::RenderPresentedFramesBehind,
+        ])
        .chain(PHASE_MS)
        .chain(SCHEDULE_MS)
        .chain(GPU_MS)
@@ -118,6 +118,7 @@ pub(crate) fn collect(
        // threads the compute pool was built with.
        compute_threads: ComputeTaskPool::try_get().map(|pool| pool.thread_num()),
        bevy_tracing: cfg!(feature = "bevy-trace"),
+        frame_latency_requested: crate::plugins::frame_latency(),
    };
    let _ = FACTS.set(facts);
 }
@@ -108,6 +108,48 @@ fn overlap(live: &[&FrameRow], out: &mut Vec<String>) {
        "  measured from the intervals, not from the frame a render stage names as its origin: the second says where the work belongs, not that it ran inside that frame's wall. Serialised rendering reads zero here whatever else is true."
            .to_owned(),
    );
+    presented_age(out);
+}
+
+/// How old the state on the screen was, as far as the process can see it.
+///
+/// The other half of the pipelining pair. Overlap is what it buys; this is
+/// what it costs: the render world presents a frame it extracted while the
+/// main world was somewhere earlier, and the image that reaches the surface
+/// was built from that older state. Read at the render graph's Finish set,
+/// the last point in the schedule that still belongs to this image: bevy
+/// presents after the whole graph schedule returns, so the present itself is
+/// outside this measurement and so is everything after it — the compositor,
+/// the queue behind it and the panel. It is not input-to-photon, and
+/// `desired_maximum_frame_latency` is a hint the backend may clamp, so how
+/// many frames the GPU is really allowed in flight is not in here either.
+fn presented_age(out: &mut Vec<String>) {
+    let stats = perf::stats::counter_snapshot(perf::Phase::Live);
+    let find = |counter: perf::Counter| {
+        stats
+            .iter()
+            .find(|row| row.counter == counter)
+            .filter(|row| row.samples > 0)
+    };
+    let (Some(age), Some(behind)) = (
+        find(perf::Counter::RenderPresentedStateAgeMs),
+        find(perf::Counter::RenderPresentedFramesBehind),
+    ) else {
+        return;
+    };
+    out.push(format!(
+        "  presented state age: p50 {:.2} ms, p99 {:.2} ms, max {:.2} ms from extract to the end of the render graph, {:.2} main frames behind on average (p99 {:.0}), over {} presented frames.",
+        age.p50,
+        age.p99,
+        age.max,
+        behind.avg(),
+        behind.p99,
+        age.samples,
+    ));
+    out.push(
+        "  it ends where the render graph does, one step before bevy presents: the present call itself, the compositor and the display are all outside it, so this is the age of the state the graph finished with and never an input-to-photon latency."
+            .to_owned(),
+    );
 }

 /// The first seconds against the rest. Exact, not histogram midpoints: these
@@ -62,6 +62,12 @@ pub(crate) struct RuntimeFacts {
    /// different answers here are not comparable.
    pub(crate) pipelined_rendering: Option<bool>,
    pub(crate) compute_threads: Option<usize>,
+    /// Frames the surface was *asked* to let the CPU run ahead of the GPU.
+    /// wgpu treats it as a hint and a backend may clamp it — on Vulkan it is
+    /// tied to the swapchain image count — so this is the request and not the
+    /// grant, and a pair of runs across it is only comparable on what each one
+    /// asked for.
+    pub(crate) frame_latency_requested: u32,
    /// Whether this binary was built with Bevy's own `tracing` spans on. Off
    /// is the normal build and the one a timed run uses; on is a diagnostic
    /// build where the spans inside the render graph — the `queue_submit`
@@ -143,6 +149,7 @@ pub(crate) fn build(facts: &RuntimeFacts, run_id: Option<&str>, artifacts: &Path
        "scheduling": {
            "pipelined_rendering": facts.pipelined_rendering,
            "compute_threads": facts.compute_threads,
+            "frame_latency_requested": facts.frame_latency_requested,
            "bevy_tracing": facts.bevy_tracing,
        },
        "threads": threads(),
@@ -65,7 +65,7 @@ pub fn assemble_listen_app() -> App {

 pub fn default_plugins_with_quiet_log(mut window: WindowPlugin) -> bevy::app::PluginGroupBuilder {
    if let Some(primary) = window.primary_window.as_mut() {
-        primary.desired_maximum_frame_latency = core::num::NonZeroU32::new(1);
+        primary.desired_maximum_frame_latency = core::num::NonZeroU32::new(frame_latency());
    }
    let mut wgpu = WgpuSettings::default();
    wgpu.features |= WgpuFeatures::TEXTURE_FORMAT_16BIT_NORM
@@ -109,14 +109,41 @@ const PIPELINED_RENDERING_ENV: &str = "IW4L_PIPELINED_RENDERING";
 /// The manifest records which branch a run took, so a report never has to be
 /// read against a guess about it.
 fn pipelined_rendering() -> bool {
-    match std::env::var(PIPELINED_RENDERING_ENV) {
-        Ok(value) => {
-            let value = value.trim();
-            !(value.is_empty()
-                || value == "0"
-                || value.eq_ignore_ascii_case("false")
-                || value.eq_ignore_ascii_case("off"))
-        }
-        Err(_) => false,
-    }
+    perf::switch(PIPELINED_RENDERING_ENV)
+}
+
+const FRAME_LATENCY_ENV: &str = "IW4L_FRAME_LATENCY";
+
+/// How many frames the surface is asked to let the CPU run ahead of the GPU.
+///
+/// wgpu calls this a hint and the backend is free to clamp it: on Vulkan it is
+/// bound to the number of swapchain images, so a run that asked for two did
+/// not necessarily get two, and only a measurement says which. One is the
+/// default because it is what the runtime shipped; the variable exists so the
+/// other arm needs no rebuild, and the manifest records the number that was
+/// asked for — never the number the driver granted, which this process cannot
+/// read back.
+///
+/// The value is a count, not a switch: anything unparseable or zero is the
+/// default, and says so rather than silently picking an arm. Read once, so the
+/// window and the manifest cannot disagree and the complaint is made once.
+pub(crate) fn frame_latency() -> u32 {
+    const DEFAULT: u32 = 1;
+    static FRAMES: std::sync::OnceLock<u32> = std::sync::OnceLock::new();
+    *FRAMES.get_or_init(|| {
+        let Some(asked) = std::env::var_os(FRAME_LATENCY_ENV) else {
+            return DEFAULT;
+        };
+        match asked.to_str().map(str::trim).and_then(|v| v.parse().ok()) {
+            Some(frames) if frames > 0 => frames,
+            _ => {
+                diag::warn!(
+                    Launch,
+                    "{FRAME_LATENCY_ENV}={} is not a frame count; using {DEFAULT}",
+                    asked.to_string_lossy(),
+                );
+                DEFAULT
+            }
+        }
+    })
 }
@@ -11,6 +11,7 @@ mod session;
 #[path = "session_noop.rs"]
 mod session;
 pub mod stats;
+mod switch;
 #[cfg(all(unix, feature = "native"))]
 mod vocabulary;
 #[cfg(not(all(unix, feature = "native")))]
@@ -25,6 +26,7 @@ pub use event::{
    world_ready,
 };
 pub use session::{RunMetadata, enabled, flush, start};
+pub use switch::on as switch;

 /// Whether *any* recorder is listening: the Perfetto session (`IW4L_PERF`) or
 /// the in-process bench statistics (`IW4L_BENCH`). Census code that only asked
@@ -45,18 +45,7 @@ static STATE: Mutex<Option<SessionState>> = Mutex::new(None);
 static ATEXIT_ARMED: AtomicBool = AtomicBool::new(false);

 pub fn enabled() -> bool {
-    match std::env::var(ENV) {
-        Ok(value)
-            if value.is_empty()
-                || value == "0"
-                || value.eq_ignore_ascii_case("false")
-                || value.eq_ignore_ascii_case("off") =>
-        {
-            false
-        }
-        Ok(_) => true,
-        Err(_) => false,
-    }
+    crate::switch::on(ENV)
 }

 pub fn start(metadata: RunMetadata) -> Result<Option<PathBuf>, String> {
@@ -10,12 +10,7 @@ pub struct RunMetadata {
 }

 pub fn enabled() -> bool {
-    std::env::var(ENV).is_ok_and(|value| {
-        !value.is_empty()
-            && value != "0"
-            && !value.eq_ignore_ascii_case("false")
-            && !value.eq_ignore_ascii_case("off")
-    })
+    crate::switch::on(ENV)
 }

 pub fn start(_metadata: RunMetadata) -> Result<Option<PathBuf>, String> {
@@ -183,18 +183,7 @@ pub fn enabled() -> bool {
 }

 fn env_enabled() -> bool {
-    match std::env::var(ENV) {
-        Ok(value)
-            if value.is_empty()
-                || value == "0"
-                || value.eq_ignore_ascii_case("false")
-                || value.eq_ignore_ascii_case("off") =>
-        {
-            false
-        }
-        Ok(_) => true,
-        Err(_) => false,
-    }
+    crate::switch::on(ENV)
 }

 /// Everything recorded from here on belongs to [`Phase::Live`]. Called once the
@@ -0,0 +1,21 @@
+//! One reading of an environment switch, for every switch a run is alternated
+//! across.
+//!
+//! Absent, empty, `0`, `false` and `off` are off; anything else is on. The
+//! rule lives here once so that two switches cannot disagree about what `=0`
+//! means: a pair of runs across a switch that reads `=0` as on is two runs of
+//! the same arm, and nothing in the report says so.
+
+/// Whether `name` asks for the behaviour behind it.
+pub fn on(name: &str) -> bool {
+    match std::env::var(name) {
+        Ok(value) => {
+            let value = value.trim();
+            !(value.is_empty()
+                || value == "0"
+                || value.eq_ignore_ascii_case("false")
+                || value.eq_ignore_ascii_case("off"))
+        }
+        Err(_) => false,
+    }
+}
@@ -67,6 +67,15 @@ pub enum Counter {
    /// began: what the one number above had to finish and submit.
    RenderGraphSubmitPendingN,
    RenderGraphPresentMs,
+    /// How long the state this render frame drew had been extracted by the
+    /// time the render graph finished with it, and how many main frames the
+    /// main world had opened since. Zero frames behind is a render world
+    /// running in line with the main one; one or more is a pipelined render
+    /// world drawing an older state. Neither is input-to-photon: it stops
+    /// where the graph does, one step before the present call, and the
+    /// compositor and the display are further outside still.
+    RenderPresentedStateAgeMs,
+    RenderPresentedFramesBehind,

    /// Code-constant writes the material overlay made for the colour list.
    /// Work the frame did to fill values a shader may or may not read; it
@@ -254,6 +263,8 @@ impl Counter {
        Self::RenderGraphSubmitIntervalMs,
        Self::RenderGraphSubmitPendingN,
        Self::RenderGraphPresentMs,
+        Self::RenderPresentedStateAgeMs,
+        Self::RenderPresentedFramesBehind,
        Self::CounterOverlayConstWrites,
        Self::RenderSubmitSunMs,
        Self::RenderSubmitGatherMs,
@@ -305,6 +316,8 @@ impl Counter {
            Self::RenderGraphSubmitIntervalMs => "graph_submit_schedule_interval",
            Self::RenderGraphSubmitPendingN => "graph_submit_pending",
            Self::RenderGraphPresentMs => "graph_present",
+            Self::RenderPresentedStateAgeMs => "presented_state_age",
+            Self::RenderPresentedFramesBehind => "presented_frames_behind",
            Self::CounterOverlayConstWrites => "overlay_const_writes",
            Self::RenderSubmitSunMs => "submit_sun",
            Self::RenderSubmitGatherMs => "submit_gather",
@@ -345,6 +358,7 @@ impl Counter {
            Self::RenderGraphRenderMs
            | Self::RenderGraphSubmitIntervalMs
            | Self::RenderGraphPresentMs
+            | Self::RenderPresentedStateAgeMs
            | Self::RenderSubmitSunMs
            | Self::RenderSubmitGatherMs
            | Self::RenderSubmitPrepareMs
@@ -49,6 +49,22 @@ fn compile_one_job(catalog: &RuntimeMaterialCatalog, job: CompileJob) -> PassOut
    }
 }

+/// Compile every job, in chunks, on the load pool.
+///
+/// The load pool's workers set their affinity to the process CPUs when they
+/// spawn, and they already exist. A thread spawned here would inherit the
+/// coordinator's affinity instead — on the paced path below that is an
+/// `AsyncComputeTaskPool` worker, confined to that pool's share of the CPUs.
+///
+/// The results keep the order of `jobs` whatever order the chunks finish in:
+/// `TaskPool::scope` returns one `Vec` per task in the order the tasks were
+/// spawned, and every task here is spawned directly from this closure.
+///
+/// The cooperative policy is explicit and off: the calling thread waits for
+/// the chunks instead of ticking the pool's queue. It is the narrow-affinity
+/// thread in the paced case, and in the synchronous one it is the main thread
+/// in the middle of world spawn — neither is somewhere unrelated load work
+/// should be pulled onto.
 fn compile_jobs_parallel(
    catalog: Arc<RuntimeMaterialCatalog>,
    jobs: Vec<CompileJob>,
@@ -60,27 +76,24 @@ fn compile_jobs_parallel(
    let nthreads = assets::load_workers().clamp(1, jobs.len());
    let chunk_len = jobs.len().div_ceil(nthreads);

-    std::thread::scope(|scope| {
-        let handles: Vec<_> = jobs
-            .chunks(chunk_len)
-            .map(|chunk| {
+    assets::load_pool()
+        .scope_with_executor(false, None, |scope| {
+            for chunk in jobs.chunks(chunk_len) {
                let catalog = Arc::clone(&catalog);
                let progress = Arc::clone(&progress);
-                scope.spawn(move || {
+                scope.spawn(async move {
                    let mut out = Vec::with_capacity(chunk.len());
                    for &job in chunk {
                        out.push(compile_one_job(&catalog, job));
                        progress.fetch_add(1, Ordering::Relaxed);
                    }
                    out
-                })
-            })
-            .collect();
-        handles
-            .into_iter()
-            .flat_map(|handle| handle.join().expect("compile worker"))
-            .collect()
-    })
+                });
+            }
+        })
+        .into_iter()
+        .flatten()
+        .collect()
 }

 fn log_catalog_generation(catalog: &RuntimeMaterialCatalog) {
@@ -134,6 +134,18 @@ fn image_bytes(image: &Image) -> u64 {
        .unwrap_or(0)
 }

+/// Every image one lightmap page hands over, summed. A page is six `add`s, and
+/// `handed_bytes`, `bytes_total` and the budget that paces the overlay frames
+/// are all owed all six of them.
+fn lightmap_page_bytes(page: &super::world::WorldLightmap) -> u64 {
+    page.primary_image.as_ref().map_or(0, image_bytes)
+        + page.secondary_image.as_ref().map_or(0, image_bytes)
+        + page.secondary_b_image.as_ref().map_or(0, image_bytes)
+        + image_bytes(&page.ambient_image)
+        + image_bytes(&page.directional_image)
+        + image_bytes(&page.sun_mask_image)
+}
+
 #[derive(Default)]
 pub struct WorldImageUpload {
    pub done: u32,
@@ -142,6 +154,9 @@ pub struct WorldImageUpload {
    /// server takes the image here, the render world turns it into a texture
    /// some frames later, and the driver copies it later still. Calling this
    /// "uploaded" is how a handoff rate gets reported as a transfer rate.
+    ///
+    /// A slot that takes another slot's handle hands nothing over and is not
+    /// in here; its bytes are `reused_handle_bytes`.
    pub handed_bytes: u64,
    pub bytes_total: u64,
    /// The longest single handoff, and how big it was. A frame budget can only
@@ -297,7 +312,13 @@ impl WorldImageUpload {
            .flatten()
            .map(image_bytes)
            .sum::<u64>()
-            + self.probes.iter().flatten().map(image_bytes).sum::<u64>();
+            + self.probes.iter().flatten().map(image_bytes).sum::<u64>()
+            + self
+                .lightmaps
+                .iter()
+                .flatten()
+                .map(lightmap_page_bytes)
+                .sum::<u64>();
        self.exact_at = 0;
        self.probe_at = 0;
        self.lightmap_at = 0;
@@ -359,8 +380,8 @@ impl WorldImageUpload {
                continue;
            }
            let bytes = image.as_ref().map(image_bytes).unwrap_or(0);
-            self.handed_bytes += bytes;
            let step = std::time::Instant::now();
+            let mut handed = bytes;
            let variant = self.exact_variants[self.exact_at];
            self.exact_handles[self.exact_at] = image.map(|image| {
                // A slot whose variant is already an asset takes that handle
@@ -372,13 +393,17 @@ impl WorldImageUpload {
                if let Some(handle) = self.exact_by_variant.get(&variant) {
                    self.reused_handles = self.reused_handles.saturating_add(1);
                    self.reused_handle_bytes = self.reused_handle_bytes.saturating_add(bytes);
+                    // Nothing was handed to the asset server: the slot took a
+                    // handle that already existed.
+                    handed = 0;
                    return handle.clone();
                }
                let handle = images.add(image);
                self.exact_by_variant.insert(variant, handle.clone());
                handle
            });
-            self.note_step(step.elapsed(), bytes);
+            self.handed_bytes += handed;
+            self.note_step(step.elapsed(), handed);
            self.exact_at += 1;
            self.done = self.done.saturating_add(1);
            stepped = true;
@@ -409,6 +434,7 @@ impl WorldImageUpload {
                return false;
            }
            let page = self.lightmaps[self.lightmap_at].take();
+            let bytes = page.as_ref().map_or(0, lightmap_page_bytes);
            let step = std::time::Instant::now();
            self.lightmap_handles[self.lightmap_at] = page.map(|lightmap| {
                diag::info!(
@@ -432,8 +458,11 @@ impl WorldImageUpload {
                }
            });
            // One lightmap page is six adds and they cannot be split, so the
-            // page is the indivisible unit here, not the image.
-            self.note_step(step.elapsed(), 0);
+            // page is the indivisible unit here, not the image — and its bytes
+            // are the six summed, which is what the budget has to be told
+            // about even though it cannot stop in the middle of them.
+            self.handed_bytes += bytes;
+            self.note_step(step.elapsed(), bytes);
            if self.lightmap_handles[self.lightmap_at].is_some() {
                self.done = self.done.saturating_add(1);
            }
@@ -17,6 +17,21 @@ pub(crate) struct RenderFrameSample {
    /// finished sample a frame or more later, so without this the stage
    /// timings land on whichever row happened to be open when they arrived.
    origin_frame: u64,
+    /// When this render frame's extract began: the moment the state it
+    /// presents was taken out of the main world, with the main world stalled
+    /// so nothing moved between the two.
+    origin_extracted_at: Option<Instant>,
+    /// How old that state was when the render graph finished with it, and how
+    /// many main frames had been opened since it was taken. Serialised
+    /// rendering answers zero frames behind; a pipelined render world answers
+    /// one or more, which is the cost the throughput result is paid for in.
+    ///
+    /// It ends where the graph does. Bevy presents after the `RenderGraph`
+    /// schedule returns, so the present call is already outside this, and the
+    /// compositor and the display are outside the process entirely: it is not
+    /// input-to-photon and must never be read as it.
+    pub(crate) presented_state_age_ms: Option<f32>,
+    pub(crate) presented_frames_behind: Option<u32>,
    /// Which render frame this is. The main world reads the published sample
    /// whether or not a new one was published, and emitting the same
    /// measurement again would count one piece of work twice.
@@ -243,6 +258,7 @@ impl SharedRenderStages {
        self.working = RenderFrameSample::default();
        self.working.sequence = sequence;
        self.working.origin_frame = origin_frame;
+        self.working.origin_extracted_at = Some(Instant::now());
        self.working.wgpu_floor_encode_ms = floor.0;
        self.working.wgpu_floor_finish_ms = floor.1;
        self.working.wgpu_floor_submit_ms = floor.2;
@@ -385,6 +401,10 @@ pub struct RenderFrameDiag {
    pub graph_submit_pending_n: Option<u32>,

    pub graph_present_ms: Option<f32>,
+    /// How old the drawn state was when the render graph finished with it, and
+    /// how many main frames the main world had opened since it was extracted.
+    pub presented_state_age_ms: Option<f32>,
+    pub presented_frames_behind: Option<u32>,

    pub wgpu_floor_encode_ms: Option<f32>,
    pub wgpu_floor_finish_ms: Option<f32>,
@@ -814,8 +834,23 @@ fn graph_submit_end(timer: Res<GraphSubmitStageTimer>, slot: Res<SharedRenderSta
 #[derive(Resource, Default)]
 struct GraphPresentMark(Option<Instant>);

-fn graph_present_mark(mut mark: ResMut<GraphPresentMark>) {
-    mark.0 = Some(Instant::now());
+fn graph_present_mark(mut mark: ResMut<GraphPresentMark>, slot: Res<SharedRenderStagesSlot>) {
+    let now = Instant::now();
+    mark.0 = Some(now);
+    // The last set of the graph, which is the latest point in the schedule
+    // that still belongs to *this* image: bevy presents after the whole
+    // `RenderGraph` schedule returns, from `render_system`, where no system of
+    // ours runs. So this is close to the present and deliberately not it.
+    let open = perf::frames::open_index();
+    if let Ok(mut guard) = slot.0.lock() {
+        let origin = guard.working.origin_frame;
+        guard.working.presented_frames_behind =
+            Some(u32::try_from(open.saturating_sub(origin)).unwrap_or(u32::MAX));
+        if let Some(extracted) = guard.working.origin_extracted_at {
+            guard.working.presented_state_age_ms =
+                Some(now.duration_since(extracted).as_secs_f32() * 1000.0);
+        }
+    }
 }

 fn stamp_graph_present(mark: Res<GraphPresentMark>, slot: Res<SharedRenderStagesSlot>) {
@@ -1158,6 +1193,8 @@ pub fn sample_render_frame_diag(
            diag.graph_submit_ms = sample.graph_submit_ms;
            diag.graph_submit_pending_n = sample.graph_submit_pending_n;
            diag.graph_present_ms = sample.graph_present_ms;
+            diag.presented_state_age_ms = sample.presented_state_age_ms;
+            diag.presented_frames_behind = sample.presented_frames_behind;

            if fresh {
                for (counter, value) in [
@@ -1167,6 +1204,10 @@ pub fn sample_render_frame_diag(
                        sample.graph_submit_ms,
                    ),
                    (perf::Counter::RenderGraphPresentMs, sample.graph_present_ms),
+                    (
+                        perf::Counter::RenderPresentedStateAgeMs,
+                        sample.presented_state_age_ms,
+                    ),
                ] {
                    if let Some(value) = value {
                        counter.emit_at(f64::from(value), sample.origin_frame);
@@ -1176,6 +1217,10 @@ pub fn sample_render_frame_diag(
                    perf::Counter::RenderGraphSubmitPendingN
                        .emit_at(f64::from(pending), sample.origin_frame);
                }
+                if let Some(behind) = sample.presented_frames_behind {
+                    perf::Counter::RenderPresentedFramesBehind
+                        .emit_at(f64::from(behind), sample.origin_frame);
+                }
            }
            diag.wgpu_floor_encode_ms = sample.wgpu_floor_encode_ms;
            diag.wgpu_floor_finish_ms = sample.wgpu_floor_finish_ms;
@@ -6,9 +6,13 @@ use super::residency::GpuStream;

 const ARG_WORDS: usize = 5;

+/// Whether this run asked for the indirect path.
+///
+/// Read once, through the same rule as every other switch a paired run is
+/// alternated across: `IW4L_MULTI_DRAW=0` turns it off.
 pub(super) fn multi_draw_requested() -> bool {
    static REQUESTED: std::sync::OnceLock<bool> = std::sync::OnceLock::new();
-    *REQUESTED.get_or_init(|| std::env::var_os("IW4L_MULTI_DRAW").is_some())
+    *REQUESTED.get_or_init(|| perf::switch("IW4L_MULTI_DRAW"))
 }

 #[derive(bevy::prelude::Resource, Default)]