diff --git a/Cargo.lock b/Cargo.lock
index 68e8bb5f..229c2fe2 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -4835,6 +4835,7 @@ dependencies = [
  "anyhow",
  "arrow",
  "bs58",
+ "libc",
  "serde",
  "serde_json",
  "sqd-array",
diff --git a/Dockerfile b/Dockerfile
index d9c757e8..48870e4f 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -63,6 +63,24 @@ COPY --from=hotblocks-retain-builder /out/sqd-hotblocks-retain .
 ENTRYPOINT ["/app/sqd-hotblocks-retain"]
 
 
+FROM builder AS flush-bench-builder
+ARG TARGETARCH
+# `cargo bench --no-run` emits target/release/deps/flush_spill-<hash>; the cache mount may
+# hold stale hashes, so take the newest non-.d artifact.
+RUN --mount=type=cache,target=/usr/local/cargo/registry,sharing=locked \
+    --mount=type=cache,target=/usr/local/cargo/git,sharing=locked \
+    --mount=type=cache,target=/app/target,id=cargo-target-${TARGETARCH},sharing=locked \
+    cargo bench -p sqd-data --bench flush_spill --no-run \
+    && mkdir -p /out \
+    && cp "$(ls -t target/release/deps/flush_spill-* | grep -v '\.d$' | head -1)" /out/flush_spill
+
+
+FROM debian:bookworm-slim AS flush-bench
+WORKDIR /app
+COPY --from=flush-bench-builder /out/flush_spill .
+ENTRYPOINT ["/app/flush_spill"]
+
+
 FROM builder AS archive-builder
 ARG TARGETARCH
 RUN --mount=type=cache,target=/usr/local/cargo/registry,sharing=locked \
@@ -73,6 +91,7 @@ RUN --mount=type=cache,target=/usr/local/cargo/registry,sharing=locked \
     && cp target/release/sqd-archive /out/
 
 
+# keep this stage last: it is the default target of a bare `docker build .`
 FROM debian:bookworm-slim AS sqd-archive
 RUN apt-get update && apt-get install ca-certificates -y
 WORKDIR /app
diff --git a/crates/data-core/src/chunk_builder.rs b/crates/data-core/src/chunk_builder.rs
index 5e050dc7..e796db00 100644
--- a/crates/data-core/src/chunk_builder.rs
+++ b/crates/data-core/src/chunk_builder.rs
@@ -2,7 +2,7 @@ use std::collections::BTreeMap;
 
 use sqd_array::slice::AnyTableSlice;
 
-use crate::ChunkProcessor;
+use crate::{ChunkProcessor, PreparedChunk};
 
 pub trait BlockChunkBuilder: ChunkBuilder {
     type Block;
@@ -24,6 +24,10 @@ pub trait ChunkBuilder {
     fn new_chunk_processor(&self) -> anyhow::Result<ChunkProcessor>;
 
     fn submit_to_processor(&self, processor: &mut ChunkProcessor) -> anyhow::Result<()>;
+
+    /// In-memory equivalent of `new_chunk_processor()` + submit + `finish()`; clears the
+    /// builder on success. No temp files.
+    fn prepare_in_memory(&mut self) -> anyhow::Result<PreparedChunk>;
 }
 
 #[macro_export]
@@ -98,6 +102,34 @@ macro_rules! chunk_builder {
                 Ok(())
             }
 
+            pub fn prepare_in_memory(&mut self) -> anyhow::Result<sqd_data_core::PreparedChunk> {
+                use sqd_array::slice::*;
+                let downcast = sqd_data_core::Downcast::new();
+                // downcast is chunk-wide: register all tables before building any
+                $(
+                sqd_data_core::register_downcast(
+                    &self.$table.as_slice(),
+                    &self.$table.schema(),
+                    $builder::table_description(),
+                    &downcast
+                )?;
+                )*
+                let mut tables = std::collections::BTreeMap::new();
+                $(
+                tables.insert(
+                    stringify!($table),
+                    sqd_data_core::PreparedTable::from_slice(
+                        &self.$table.as_slice(),
+                        self.$table.schema(),
+                        $builder::table_description(),
+                        downcast.clone()
+                    )?
+                );
+                )*
+                self.clear();
+                Ok(tables)
+            }
+
             pub fn dataset_description() -> sqd_dataset::DatasetDescriptionRef {
                 use sqd_dataset::*;
                 use std::sync::{Arc, LazyLock};
@@ -145,6 +177,10 @@ macro_rules! chunk_builder {
             fn submit_to_processor(&self, processor: &mut sqd_data_core::ChunkProcessor) -> anyhow::Result<()> {
                 self.submit_to_processor(processor)
             }
+
+            fn prepare_in_memory(&mut self) -> anyhow::Result<sqd_data_core::PreparedChunk> {
+                self.prepare_in_memory()
+            }
         }
 
         impl Default for $name {
diff --git a/crates/data-core/src/table_processor.rs b/crates/data-core/src/table_processor.rs
index 5b318930..9b66f8ec 100644
--- a/crates/data-core/src/table_processor.rs
+++ b/crates/data-core/src/table_processor.rs
@@ -1,5 +1,6 @@
 use std::{collections::HashMap, sync::Arc};
 
+use anyhow::bail;
 use arrow::{
     array::RecordBatch,
     datatypes::{DataType, Field, SchemaRef}
@@ -9,6 +10,7 @@ use sqd_array::{
     item_index_cast::cast_item_index,
     schema_patch::SchemaPatch,
     slice::{AnyTableSlice, AsSlice, Slice},
+    sort::sort_table_to_indexes,
     util::build_field_offsets,
     writer::ArrayWriter
 };
@@ -43,25 +45,39 @@ impl TableWriter {
 
 enum TableReader {
     Plain(TableFile),
-    Sort(SortedTable)
+    Sort(SortedTable),
+    Mem(MemTable)
 }
 
 impl TableReader {
     fn read_column(&mut self, dst: &mut impl ArrayWriter, i: usize, offset: usize, len: usize) -> anyhow::Result<()> {
         match self {
             TableReader::Plain(reader) => reader.read_column(dst, i, offset, len),
-            TableReader::Sort(reader) => reader.read_column(dst, i, offset, len)
+            TableReader::Sort(reader) => reader.read_column(dst, i, offset, len),
+            TableReader::Mem(reader) => reader.read_column(dst, i, offset, len)
         }
     }
 
     fn into_writer(self) -> anyhow::Result<TableWriter> {
         match self {
             TableReader::Plain(reader) => reader.into_writer().map(TableWriter::Plain),
-            TableReader::Sort(reader) => reader.into_sorter().map(TableWriter::Sort)
+            TableReader::Sort(reader) => reader.into_sorter().map(TableWriter::Sort),
+            TableReader::Mem(_) => bail!("processor reuse is not supported for in-memory prepared tables")
         }
     }
 }
 
+/// Columns materialized in output (sort-key) order.
+struct MemTable {
+    columns: Vec<AnyBuilder>
+}
+
+impl MemTable {
+    fn read_column(&self, dst: &mut impl ArrayWriter, i: usize, offset: usize, len: usize) -> anyhow::Result<()> {
+        self.columns[i].as_slice().slice(offset, len).write(dst)
+    }
+}
+
 pub struct TableProcessor {
     downcast: Downcast,
     schema: SchemaRef,
@@ -178,6 +194,72 @@ impl PreparedTable {
         })
     }
 
+    /// One-batch, in-memory equivalent of `TableProcessor` push + finish — no temp files.
+    ///
+    /// `downcast` must have every table of the chunk registered ([`register_downcast`])
+    /// before any table is built. Row order within equal full sort keys is unspecified and
+    /// may differ from the spill path (unstable sort); group contents are identical, and
+    /// queries re-sort output by a row-unique primary key, so it is not client-visible.
+    pub fn from_slice(
+        records: &AnyTableSlice<'_>,
+        schema: SchemaRef,
+        desc: &TableDescription,
+        downcast: Downcast
+    ) -> anyhow::Result<Self> {
+        let block_number_columns = desc
+            .downcast
+            .block_number
+            .iter()
+            .map(|name| schema.index_of(name))
+            .collect::<Result<Vec<_>, _>>()?;
+
+        let item_index_columns = desc
+            .downcast
+            .item_index
+            .iter()
+            .map(|name| schema.index_of(name))
+            .collect::<Result<Vec<_>, _>>()?;
+
+        let sort_key = desc
+            .sort_key
+            .iter()
+            .map(|name| schema.index_of(name))
+            .collect::<Result<Vec<_>, _>>()?;
+
+        let order = (!sort_key.is_empty() && records.len() > 0).then(|| sort_table_to_indexes(records, &sort_key));
+
+        let columns = (0..records.num_columns())
+            .map(|i| {
+                let mut b = AnyBuilder::new(schema.field(i).data_type());
+                match &order {
+                    Some(order) => records.column(i).write_indexes(&mut b, order.iter().copied())?,
+                    None => records.column(i).write(&mut b)?
+                }
+                Ok(b)
+            })
+            .collect::<anyhow::Result<Vec<_>>>()?;
+
+        let prepared_schema = downcast_schema(
+            schema.clone(),
+            &block_number_columns,
+            &item_index_columns,
+            downcast.get_block_number_type(),
+            downcast.get_item_index_type()
+        );
+
+        Ok(Self {
+            downcast,
+            block_number_columns,
+            item_index_columns,
+            column_offsets: build_field_offsets(0, schema.fields()),
+            writer_schema: schema,
+            prepared_schema,
+            reader: TableReader::Mem(MemTable { columns }),
+            buffers: HashMap::with_capacity(3),
+            num_rows: records.len()
+        })
+    }
+
     pub fn into_processor(self) -> anyhow::Result<TableProcessor> {
         self.downcast.reset();
         Ok(TableProcessor {
@@ -262,6 +344,23 @@ impl PreparedTable {
     }
 }
 
+/// Register a table's downcast columns without processing it — the in-memory path must
+/// register the whole chunk before building any [`PreparedTable`].
+pub fn register_downcast(
+    records: &AnyTableSlice<'_>,
+    schema: &SchemaRef,
+    desc: &TableDescription,
+    downcast: &Downcast
+) -> anyhow::Result<()> {
+    for name in desc.downcast.block_number.iter() {
+        downcast.reg_block_number(&records.column(schema.index_of(name)?));
+    }
+    for name in desc.downcast.item_index.iter() {
+        downcast.reg_item_index(&records.column(schema.index_of(name)?));
+    }
+    Ok(())
+}
+
 fn downcast_schema(
     schema: SchemaRef,
     block_number_columns: &[usize],
diff --git a/crates/data-core/tests/in_memory_prepare.rs b/crates/data-core/tests/in_memory_prepare.rs
new file mode 100644
index 00000000..39ef77f3
--- /dev/null
+++ b/crates/data-core/tests/in_memory_prepare.rs
@@ -0,0 +1,297 @@
+//! Differential tests: the in-memory prepare path must be observationally identical to the
+//! temp-file spill path — same prepared schemas (incl. chunk-wide downcast), same rows in
+//! the same order, for sorted/plain tables, strings, lists, nulls, and empty tables.
+
+use sqd_array::{
+    builder::{ListBuilder, StringBuilder, UInt32Builder, UInt64Builder},
+    slice::Slice
+};
+use sqd_data_core::{chunk_builder, table_builder, PreparedChunk};
+
+type TagsBuilder = ListBuilder<StringBuilder>;
+
+table_builder! {
+    SortedBuilder {
+        block_number: UInt64Builder,
+        item_index: UInt32Builder,
+        name: StringBuilder,
+        tags: TagsBuilder,
+    }
+
+    description(d) {
+        d.downcast.block_number = vec!["block_number"];
+        d.downcast.item_index = vec!["item_index"];
+        d.sort_key = vec!["name", "block_number", "item_index"];
+    }
+}
+
+table_builder! {
+    PlainBuilder {
+        block_number: UInt64Builder,
+        value: StringBuilder,
+    }
+
+    description(d) {
+        d.downcast.block_number = vec!["block_number"];
+    }
+}
+
+chunk_builder! {
+    TestChunkBuilder {
+        sorted: SortedBuilder,
+        plain: PlainBuilder,
+    }
+}
+
+/// Unsorted input with duplicate names and shuffled block numbers; the sort-key tail
+/// (`item_index`) stays unique so both paths agree on a total order.
+fn populate(b: &mut TestChunkBuilder, rows: u64, base_block: u64) {
+    let mut x = 7u64;
+    for i in 0..rows {
+        x = x.wrapping_mul(6364136223846793005).wrapping_add(1442695040888963407);
+        b.sorted.block_number.append(base_block + x % rows);
+        b.sorted.item_index.append(i as u32);
+        match i % 5 {
+            0 => b.sorted.name.append(""),
+            1 => b.sorted.name.append_null(),
+            _ => b.sorted.name.append(&format!("name-{}", x % 7))
+        }
+        if i % 4 == 3 {
+            b.sorted.tags.append_null();
+        } else {
+            for t in 0..i % 3 {
+                b.sorted.tags.values().append(&format!("tag-{t}-{}", x % 5));
+            }
+            b.sorted.tags.append();
+        }
+        b.plain.block_number.append(base_block + i);
+        b.plain.value.append(&format!("v{i}"));
+    }
+}
+
+fn disk_prepare(b: &TestChunkBuilder) -> PreparedChunk {
+    let mut p = b.new_chunk_processor().unwrap();
+    b.submit_to_processor(&mut p).unwrap();
+    p.finish().unwrap()
+}
+
+fn assert_chunks_equal(mut disk: PreparedChunk, mut mem: PreparedChunk) {
+    assert_eq!(disk.len(), mem.len());
+    for ((dn, d), (mn, m)) in disk.iter_mut().zip(mem.iter_mut()) {
+        assert_eq!(dn, mn);
+        assert_eq!(d.schema(), m.schema(), "schema mismatch for table {dn}");
+        assert_eq!(d.num_rows(), m.num_rows(), "row count mismatch for table {dn}");
+        let rows = d.num_rows();
+        assert_eq!(
+            d.read_record_batch(0, rows).unwrap(),
+            m.read_record_batch(0, rows).unwrap(),
+            "full read mismatch for table {dn}"
+        );
+        for (offset, len) in [
+            (0, rows.min(1)),
+            (rows / 2, rows - rows / 2),
+            (rows.saturating_sub(1), rows.min(1))
+        ] {
+            assert_eq!(
+                d.read_record_batch(offset, len).unwrap(),
+                m.read_record_batch(offset, len).unwrap(),
+                "window ({offset}, {len}) mismatch for table {dn}"
+            );
+        }
+    }
+}
+
+#[test]
+fn in_memory_prepare_matches_spill_path() {
+    let mut b = TestChunkBuilder::new();
+    populate(&mut b, 57, 1_000);
+    let disk = disk_prepare(&b);
+    let mem = b.prepare_in_memory().unwrap();
+    assert_chunks_equal(disk, mem);
+    assert_eq!(b.max_num_rows(), 0, "prepare_in_memory must clear the builder");
+}
+
+#[test]
+fn duplicate_sort_keys_match_spill_path() {
+    let mut b = TestChunkBuilder::new();
+    for i in 0..1_003u64 {
+        b.sorted.block_number.append(1_000 + (i.wrapping_mul(17) % 5));
+        b.sorted.item_index.append((i.wrapping_mul(7) % 11) as u32);
+        b.sorted.name.append(&format!("key-{}", i.wrapping_mul(13) % 3));
+        b.sorted.tags.values().append(&format!("payload-{i}"));
+        b.sorted.tags.append();
+
+        b.plain.block_number.append(1_000);
+        b.plain.value.append(&format!("value-{i}"));
+    }
+
+    let disk = disk_prepare(&b);
+    let mem = b.prepare_in_memory().unwrap();
+    assert_chunks_equal(disk, mem);
+}
+
+#[test]
+fn small_block_numbers_downcast_identically() {
+    // max block number fits u32 → both paths must downcast the u64 column to UInt32,
+    // exercising the cast-on-read path
+    let mut b = TestChunkBuilder::new();
+    populate(&mut b, 23, 1_000);
+    let disk = disk_prepare(&b);
+    let mem = b.prepare_in_memory().unwrap();
+    for t in disk.values() {
+        let f = t.schema().field_with_name("block_number").unwrap().data_type().clone();
+        assert_eq!(f, arrow::datatypes::DataType::UInt32);
+    }
+    assert_chunks_equal(disk, mem);
+}
+
+#[test]
+fn shared_downcast_is_consistent_across_tables() {
+    // huge block numbers only in `sorted` — `plain` must still widen to UInt64 in both
+    // paths, because the downcast is chunk-wide
+    let mut b = TestChunkBuilder::new();
+    populate(&mut b, 11, u32::MAX as u64 + 100);
+    let disk = disk_prepare(&b);
+    let mem = b.prepare_in_memory().unwrap();
+    for t in mem.values() {
+        let f = t.schema().field_with_name("block_number").unwrap().data_type().clone();
+        assert_eq!(f, arrow::datatypes::DataType::UInt64);
+    }
+    assert_chunks_equal(disk, mem);
+}
+
+#[test]
+fn empty_chunk() {
+    let mut b = TestChunkBuilder::new();
+    let disk = disk_prepare(&b);
+    let mem = b.prepare_in_memory().unwrap();
+    assert_chunks_equal(disk, mem);
+}
+
+table_builder! {
+    TiedBuilder {
+        name: StringBuilder,
+        id: UInt64Builder,
+    }
+
+    description(d) {
+        d.sort_key = vec!["name"];
+    }
+}
+
+chunk_builder! {
+    TiedChunkBuilder {
+        tied: TiedBuilder,
+    }
+}
+
+/// Duplicate full sort keys: intra-group row order is unspecified (unstable sort applied a
+/// different number of times per path), but groups must hold the same rows.
+#[test]
+fn duplicate_sort_keys_keep_group_contents() {
+    use arrow::{array::AsArray, datatypes::UInt64Type};
+
+    let mut b = TiedChunkBuilder::new();
+    for i in 0..40u64 {
+        b.tied.name.append(&format!("k{}", i % 4));
+        b.tied.id.append(i);
+    }
+
+    let mut p = b.new_chunk_processor().unwrap();
+    b.submit_to_processor(&mut p).unwrap();
+    let mut disk = p.finish().unwrap();
+    let mut mem = b.prepare_in_memory().unwrap();
+
+    let rows = |chunk: &mut PreparedChunk| -> Vec<(String, u64)> {
+        let t = chunk.get_mut("tied").unwrap();
+        let batch = t.read_record_batch(0, t.num_rows()).unwrap();
+        let names = batch.column(0).as_string::<i32>();
+        let ids = batch.column(1).as_primitive::<UInt64Type>();
+        (0..batch.num_rows())
+            .map(|i| (names.value(i).to_string(), ids.value(i)))
+            .collect()
+    };
+
+    let d = rows(&mut disk);
+    let m = rows(&mut mem);
+
+    assert!(d.windows(2).all(|w| w[0].0 <= w[1].0), "disk output not key-sorted");
+    assert!(m.windows(2).all(|w| w[0].0 <= w[1].0), "mem output not key-sorted");
+
+    let mut ds = d.clone();
+    let mut ms = m.clone();
+    ds.sort();
+    ms.sort();
+    assert_eq!(ds, ms, "row multisets differ");
+}
+
+#[test]
+fn multiple_spill_batches_match_one_in_memory_batch() {
+    let mut disk_builder = TestChunkBuilder::new();
+    let mut processor = disk_builder.new_chunk_processor().unwrap();
+
+    populate(&mut disk_builder, 31, 1_000);
+    disk_builder.submit_to_processor(&mut processor).unwrap();
+    disk_builder.clear();
+
+    populate(&mut disk_builder, 29, 2_000);
+    disk_builder.submit_to_processor(&mut processor).unwrap();
+    disk_builder.clear();
+    let disk = processor.finish().unwrap();
+
+    let mut mem_builder = TestChunkBuilder::new();
+    populate(&mut mem_builder, 31, 1_000);
+    populate(&mut mem_builder, 29, 2_000);
+    let mem = mem_builder.prepare_in_memory().unwrap();
+
+    assert_chunks_equal(disk, mem);
+}
+
+table_builder! {
+    ValidBuilder {
+        value: UInt64Builder,
+    }
+
+    description(_d) {}
+}
+
+table_builder! {
+    InvalidBuilder {
+        value: UInt64Builder,
+    }
+
+    description(d) {
+        d.sort_key = vec!["column_that_does_not_exist"];
+    }
+}
+
+chunk_builder! {
+    FailingChunkBuilder {
+        valid: ValidBuilder,
+        invalid: InvalidBuilder,
+    }
+}
+
+#[test]
+fn preparation_error_does_not_clear_buffered_rows() {
+    let mut b = FailingChunkBuilder::new();
+    b.valid.value.append(10);
+    b.invalid.value.append(20);
+
+    let err = b.prepare_in_memory().err().expect("invalid sort key must fail");
+    assert!(err.to_string().contains("column_that_does_not_exist"));
+    assert_eq!(b.max_num_rows(), 1, "failed prepare discarded buffered rows");
+
+    let slices = b.as_slice_map();
+    assert_eq!(slices["valid"].len(), 1);
+    assert_eq!(slices["invalid"].len(), 1);
+}
+
+#[test]
+fn in_memory_prepared_table_does_not_support_reuse() {
+    let mut b = TestChunkBuilder::new();
+    populate(&mut b, 3, 10);
+    let mut mem = b.prepare_in_memory().unwrap();
+    let (_, table) = mem.pop_first().unwrap();
+    assert!(table.into_processor().is_err());
+}
diff --git a/crates/data/Cargo.toml b/crates/data/Cargo.toml
index c2c08197..12c87a3f 100644
--- a/crates/data/Cargo.toml
+++ b/crates/data/Cargo.toml
@@ -15,5 +15,12 @@ sqd-data-core = { path = "../data-core" }
 sqd-dataset = { path = "../dataset" }
 sqd-primitives = { path = "../primitives" }
 
+[dev-dependencies]
+libc = "0.2"
+
+[[bench]]
+name = "flush_spill"
+harness = false
+
 [lints]
 workspace = true
diff --git a/crates/data/benches/flush_spill.rs b/crates/data/benches/flush_spill.rs
new file mode 100644
index 00000000..ce63ec50
--- /dev/null
+++ b/crates/data/benches/flush_spill.rs
@@ -0,0 +1,638 @@
+//! Decomposed cost of one hotblocks flush (docs/adr/0001-in-memory-chunk-prepare.md;
+//! results in docs/measurements/).
+//!
+//! Reproduces the ingest hot path stage by stage on a synthetic EVM block with N rows, in
+//! three modes:
+//!   fresh — the old prod path: push → new_chunk_processor (488 temp files) →
+//!           submit+finish → readback → drop
+//!   reuse — fix #3 idealized: construct once, then per flush push → submit+finish →
+//!           readback → reconstruct (per-table `into_processor`, replaces construct+drop)
+//!   mem   — fix #1: push → prepare_in_memory → readback → drop; no temp files
+//! Reports wall/user/sys time per stage and, on Linux, /proc/self/io deltas.
+//!
+//!   cargo bench -p sqd-data --bench flush_spill                 # full sweep, all modes
+//!   SQD_BENCH_QUICK=1 cargo bench -p sqd-data --bench flush_spill
+//!   TMPDIR=/dev/shm ...                                         # Bench C (Linux; docker: --shm-size=2g)
+//!
+//! Verdicts to extract: per-stage cost = fixed + marginal×rows (Bench A); fresh−reuse =
+//! idealized ceiling of processor reuse (Bench B); /tmp vs /dev/shm = fs sensitivity (Bench C).
+
+use std::{
+    collections::BTreeMap,
+    time::{Duration, Instant}
+};
+
+use anyhow::Context;
+use serde_json::{json, Value};
+use sqd_data::evm::{model::Block, tables::EvmChunkBuilder};
+use sqd_data_core::{BlockChunkBuilder, ChunkProcessor};
+
+const N_STAGES: usize = 5;
+
+#[derive(Clone, Copy, PartialEq)]
+enum Mode {
+    Fresh,
+    Reuse,
+    /// Fix #1: `prepare_in_memory` — no temp files at all.
+    Mem
+}
+
+impl Mode {
+    fn stages(self) -> [&'static str; N_STAGES] {
+        match self {
+            Mode::Fresh => ["push", "construct", "submit+finish", "readback", "drop"],
+            Mode::Reuse => ["push", "reconstruct", "submit+finish", "readback", ""],
+            Mode::Mem => ["push", "", "prepare(mem)", "readback", "drop"]
+        }
+    }
+
+    fn name(self) -> &'static str {
+        match self {
+            Mode::Fresh => "fresh",
+            Mode::Reuse => "reuse",
+            Mode::Mem => "mem"
+        }
+    }
+}
+
+fn main() -> anyhow::Result<()> {
+    let nofile_limit = raise_nofile_limit()?;
+
+    // rows = transactions per block (logs scale 1:1); iterations shrink as rows grow
+    let plan: &[(usize, usize)] = if std::env::var_os("SQD_BENCH_QUICK").is_some() {
+        &[(1, 20), (100, 10)]
+    } else {
+        &[(1, 200), (10, 200), (100, 100), (1_000, 50), (10_000, 10), (100_000, 3)]
+    };
+
+    println!("flush_spill: EVM chunk, temp dir = {}", std::env::temp_dir().display());
+    println!("RLIMIT_NOFILE soft limit: {nofile_limit}");
+    println!("CPU accounting scope: {}", cpu_scope());
+    println!(
+        "/proc/self/io: {}",
+        if io_snapshot().is_some() {
+            "available"
+        } else {
+            "NOT available (macOS?) — io columns will be zero"
+        }
+    );
+
+    let mut cases = Vec::new();
+    for &(rows, iters) in plan {
+        cases.push(run_case(rows, iters)?);
+    }
+    println!("\ncross-mode output equality: OK (every measured iteration)");
+
+    print_fixed_marginal(&cases);
+    print_mode_comparison(&cases);
+    Ok(())
+}
+
+struct CaseRun {
+    rows: usize,
+    // p50 wall per stage — median resists warmup/writeback outliers
+    p50_wall: [Duration; N_STAGES],
+    // p50 of per-iteration totals: median(A+B) != median(A)+median(B)
+    total_p50: Duration,
+    total_p95: Duration
+}
+
+/// Interleaved fresh+reuse execution needs ~1,000 fds; default soft limits (256 on
+/// macOS) are too low.
+fn raise_nofile_limit() -> anyhow::Result<libc::rlim_t> {
+    const TARGET: libc::rlim_t = 8192;
+    const REQUIRED: libc::rlim_t = 2048;
+
+    // SAFETY: `rlimit` is valid when zero-initialized and `getrlimit` receives a valid,
+    // writable pointer for the duration of the call.
+    let mut lim: libc::rlimit = unsafe { std::mem::zeroed() };
+    // SAFETY: `lim` points to initialized writable storage and RLIMIT_NOFILE is a valid
+    // resource selector on every supported Unix target.
+    let rc = unsafe { libc::getrlimit(libc::RLIMIT_NOFILE, &mut lim) };
+    if rc != 0 {
+        return Err(std::io::Error::last_os_error()).context("getrlimit(RLIMIT_NOFILE) failed");
+    }
+
+    if lim.rlim_cur < TARGET {
+        let requested = TARGET.min(lim.rlim_max);
+        let new_lim = libc::rlimit {
+            rlim_cur: requested,
+            rlim_max: lim.rlim_max
+        };
+        // SAFETY: `new_lim` is initialized, its soft limit does not exceed its hard limit,
+        // and the pointer remains valid for the duration of the call.
+        let rc = unsafe { libc::setrlimit(libc::RLIMIT_NOFILE, &new_lim) };
+        if rc != 0 {
+            return Err(std::io::Error::last_os_error()).context("setrlimit(RLIMIT_NOFILE) failed");
+        }
+        lim = new_lim;
+    }
+
+    anyhow::ensure!(
+        lim.rlim_cur >= REQUIRED,
+        "RLIMIT_NOFILE={} is below the {REQUIRED} required by the interleaved benchmark",
+        lim.rlim_cur
+    );
+    Ok(lim.rlim_cur)
+}
+
+const MODE_ORDERS: [[Mode; 3]; 6] = [
+    [Mode::Fresh, Mode::Reuse, Mode::Mem],
+    [Mode::Reuse, Mode::Mem, Mode::Fresh],
+    [Mode::Mem, Mode::Fresh, Mode::Reuse],
+    [Mode::Fresh, Mode::Mem, Mode::Reuse],
+    [Mode::Mem, Mode::Reuse, Mode::Fresh],
+    [Mode::Reuse, Mode::Fresh, Mode::Mem]
+];
+
+impl Mode {
+    const fn index(self) -> usize {
+        match self {
+            Mode::Fresh => 0,
+            Mode::Reuse => 1,
+            Mode::Mem => 2
+        }
+    }
+}
+
+struct ModeRun {
+    mode: Mode,
+    acc: [StageAcc; N_STAGES],
+    totals: Vec<Duration>,
+    spare: Option<ChunkProcessor>,
+    fds_baseline: usize,
+    fd_delta: usize
+}
+
+impl ModeRun {
+    fn new(mode: Mode, iters: usize) -> anyhow::Result<Self> {
+        let spare = match mode {
+            Mode::Reuse => Some(EvmChunkBuilder::new().new_chunk_processor()?),
+            _ => None
+        };
+        Ok(Self {
+            mode,
+            acc: Default::default(),
+            totals: Vec::with_capacity(iters),
+            spare,
+            fds_baseline: 0,
+            fd_delta: 0
+        })
+    }
+
+    fn reset_measurements(&mut self) {
+        self.acc = Default::default();
+        self.totals.clear();
+        self.fds_baseline = open_fds();
+        self.fd_delta = 0;
+    }
+
+    fn run_iteration(&mut self, block: &Block) -> anyhow::Result<BTreeMap<&'static str, arrow::array::RecordBatch>> {
+        let mut total = Duration::ZERO;
+        let mut builder = EvmChunkBuilder::new();
+
+        let (_, wall) = measured(&mut self.acc[0], || builder.push(block))?;
+        total += wall;
+
+        let mut prepared = match self.mode {
+            Mode::Mem => {
+                let (prepared, wall) = measured(&mut self.acc[2], || builder.prepare_in_memory())?;
+                total += wall;
+                prepared
+            }
+            Mode::Fresh | Mode::Reuse => {
+                let mut processor = match self.mode {
+                    Mode::Fresh => {
+                        let fds_before = open_fds();
+                        let (processor, wall) = measured(&mut self.acc[1], || builder.new_chunk_processor())?;
+                        total += wall;
+                        if self.fd_delta == 0 {
+                            self.fd_delta = open_fds().saturating_sub(fds_before);
+                        }
+                        processor
+                    }
+                    Mode::Reuse => self.spare.take().context("spare processor is missing")?,
+                    Mode::Mem => unreachable!()
+                };
+                let (prepared, wall) = measured(&mut self.acc[2], || {
+                    builder.submit_to_processor(&mut processor)?;
+                    builder.clear();
+                    processor.finish()
+                })?;
+                total += wall;
+                prepared
+            }
+        };
+
+        let (sample, wall) = measured(&mut self.acc[3], || {
+            prepared
+                .iter_mut()
+                .map(|(name, table)| {
+                    let batch = table.read_record_batch(0, table.num_rows())?;
+                    Ok((*name, batch))
+                })
+                .collect::<anyhow::Result<BTreeMap<_, _>>>()
+        })?;
+        total += wall;
+
+        match self.mode {
+            Mode::Fresh | Mode::Mem => {
+                let (_, wall) = measured(&mut self.acc[4], || {
+                    drop(prepared);
+                    Ok(())
+                })?;
+                total += wall;
+            }
+            Mode::Reuse => {
+                let (spare, wall) = measured(&mut self.acc[1], || {
+                    let tables = prepared
+                        .into_iter()
+                        .map(|(name, table)| table.into_processor().map(|processor| (name, processor)))
+                        .collect::<anyhow::Result<BTreeMap<_, _>>>()?;
+                    Ok(ChunkProcessor::new(tables))
+                })?;
+                total += wall;
+                self.spare = Some(spare);
+            }
+        }
+
+        self.totals.push(total);
+        Ok(sample)
+    }
+
+    fn report(&mut self, rows: usize) -> CaseRun {
+        println!("\n-- mode={} --", self.mode.name());
+        match self.mode {
+            Mode::Fresh => println!("temp files created in `construct` (fd delta): {}", self.fd_delta),
+            Mode::Mem => print_fd_stability("temp files: none; fds stable", self.fds_baseline),
+            Mode::Reuse => print_fd_stability("fds stable across iterations", self.fds_baseline)
+        }
+
+        println!(
+            "{:<15} {:>9} {:>9} {:>9} {:>9} {:>9} {:>8} {:>8} {:>10} {:>10} {:>8} {:>9} {:>9}",
+            "stage",
+            "p50 ms",
+            "p95 ms",
+            "mean ms",
+            "user ms",
+            "sys ms",
+            "syscr",
+            "syscw",
+            "wchar KB",
+            "rchar KB",
+            "rd MB",
+            "dirty MB",
+            "cancel MB"
+        );
+
+        let stages = self.mode.stages();
+        let mut p50_wall = [Duration::ZERO; N_STAGES];
+        for (idx, acc) in self.acc.iter_mut().enumerate() {
+            if acc.wall.is_empty() {
+                continue;
+            }
+            acc.wall.sort_unstable();
+            let p50 = percentile(&acc.wall, 50);
+            let p95 = percentile(&acc.wall, 95);
+            let mean = acc.wall.iter().sum::<Duration>() / acc.wall.len() as u32;
+            p50_wall[idx] = p50;
+            let n = acc.wall.len() as f64;
+            let per = |value: u64| value as f64 / n;
+            println!(
+                "{:<15} {:>9.3} {:>9.3} {:>9.3} {:>9.3} {:>9.3} {:>8.0} {:>8.0} {:>10.1} {:>10.1} {:>8.2} {:>9.2} {:>9.2}",
+                stages[idx],
+                p50.as_secs_f64() * 1e3,
+                p95.as_secs_f64() * 1e3,
+                mean.as_secs_f64() * 1e3,
+                acc.user.as_secs_f64() * 1e3 / n,
+                acc.sys.as_secs_f64() * 1e3 / n,
+                per(acc.io.syscr),
+                per(acc.io.syscw),
+                per(acc.io.wchar) / 1024.0,
+                per(acc.io.rchar) / 1024.0,
+                per(acc.io.read_bytes) / (1024.0 * 1024.0),
+                per(acc.io.write_bytes) / (1024.0 * 1024.0),
+                per(acc.io.cancelled_write_bytes) / (1024.0 * 1024.0)
+            );
+        }
+
+        self.totals.sort_unstable();
+        let total_p50 = percentile(&self.totals, 50);
+        let total_p95 = percentile(&self.totals, 95);
+        println!(
+            "per-iteration measured-stage total: p50 {:.3} ms, p95 {:.3} ms",
+            total_p50.as_secs_f64() * 1e3,
+            total_p95.as_secs_f64() * 1e3
+        );
+
+        CaseRun {
+            rows,
+            p50_wall,
+            total_p50,
+            total_p95
+        }
+    }
+}
+
+fn print_fd_stability(label: &str, baseline: usize) {
+    let end = open_fds();
+    println!(
+        "{label}: {} (start {baseline}, end {end})",
+        if end == baseline { "yes" } else { "NO — LEAK" }
+    );
+}
+
+fn run_case(rows: usize, iters: usize) -> anyhow::Result<(CaseRun, CaseRun, CaseRun)> {
+    let block: Block = serde_json::from_value(gen_block_json(rows, 0))?;
+
+    let mut builder = EvmChunkBuilder::new();
+    builder.push(&block)?;
+    let block_bytes = builder.byte_size();
+    builder.clear();
+
+    println!(
+        "\n== rows={rows}  iters={iters}  in-memory block ~{} KB ==",
+        block_bytes / 1024
+    );
+
+    let mut runs = [
+        ModeRun::new(Mode::Fresh, iters)?,
+        ModeRun::new(Mode::Reuse, iters)?,
+        ModeRun::new(Mode::Mem, iters)?
+    ];
+
+    for iteration in 0..2 {
+        let block: Block = serde_json::from_value(gen_block_json(rows, iteration + 10_000))?;
+        run_round(&mut runs, &block, iteration, rows)?;
+    }
+    for run in &mut runs {
+        run.reset_measurements();
+    }
+
+    for iteration in 0..iters {
+        let block: Block = serde_json::from_value(gen_block_json(rows, iteration))?;
+        run_round(&mut runs, &block, iteration, rows)?;
+    }
+
+    let mut reports = runs.iter_mut().map(|run| run.report(rows));
+    Ok((
+        reports.next().unwrap(),
+        reports.next().unwrap(),
+        reports.next().unwrap()
+    ))
+}
+
+fn run_round(runs: &mut [ModeRun; 3], block: &Block, iteration: usize, rows: usize) -> anyhow::Result<()> {
+    let mut outputs = std::array::from_fn::<_, 3, _>(|_| None);
+    for mode in MODE_ORDERS[iteration % MODE_ORDERS.len()] {
+        outputs[mode.index()] = Some(runs[mode.index()].run_iteration(block)?);
+    }
+
+    let fresh = outputs[Mode::Fresh.index()].as_ref().unwrap();
+    anyhow::ensure!(
+        Some(fresh) == outputs[Mode::Reuse.index()].as_ref(),
+        "fresh vs reuse output mismatch at rows={rows}, iteration={iteration}"
+    );
+    anyhow::ensure!(
+        Some(fresh) == outputs[Mode::Mem.index()].as_ref(),
+        "fresh vs mem output mismatch at rows={rows}, iteration={iteration}"
+    );
+    Ok(())
+}
+
+fn percentile(sorted: &[Duration], percent: usize) -> Duration {
+    let index = ((sorted.len() - 1) * percent + 50) / 100;
+    sorted[index]
+}
+
+fn print_fixed_marginal(cases: &[(CaseRun, CaseRun, CaseRun)]) {
+    let (first, last) = (&cases[0].0, &cases[cases.len() - 1].0);
+    if first.rows == last.rows {
+        return;
+    }
+    println!(
+        "\n== fresh mode: fixed vs marginal (fixed ≈ p50 at rows={}, marginal from rows={}) ==",
+        first.rows, last.rows
+    );
+    println!("{:<15} {:>12} {:>18}", "stage", "fixed ms", "marginal ms/1k rows");
+    let dr = (last.rows - first.rows) as f64;
+    for (idx, stage) in Mode::Fresh.stages().iter().enumerate() {
+        let fixed = first.p50_wall[idx].as_secs_f64() * 1e3;
+        let marginal = (last.p50_wall[idx].as_secs_f64() - first.p50_wall[idx].as_secs_f64()) * 1e3 / dr * 1000.0;
+        println!("{:<15} {:>12.3} {:>18.3}", stage, fixed, marginal);
+    }
+    let fixed_total = first.total_p50.as_secs_f64() * 1e3;
+    let marginal_total = (last.total_p50.as_secs_f64() - first.total_p50.as_secs_f64()) * 1e3 / dr * 1000.0;
+    println!("{:<15} {:>12.3} {:>18.3}", "TOTAL", fixed_total, marginal_total);
+    println!(
+        "rows where marginal overtakes fixed: ~{:.0}",
+        if marginal_total > 0.0 {
+            fixed_total / marginal_total * 1000.0
+        } else {
+            f64::INFINITY
+        }
+    );
+}
+
+fn print_mode_comparison(cases: &[(CaseRun, CaseRun, CaseRun)]) {
+    println!("\n== measured-stage total: fresh (prod) vs reuse (fix #3 ceiling) vs mem (fix #1) ==");
+    println!(
+        "{:>8} {:>10} {:>10} {:>10} {:>10} {:>8} {:>10} {:>10} {:>8}",
+        "rows", "fresh p50", "fresh p95", "reuse p50", "reuse p95", "saves", "mem p50", "mem p95", "saves"
+    );
+    for (fresh, reuse, mem) in cases {
+        let f = fresh.total_p50.as_secs_f64() * 1e3;
+        let r = reuse.total_p50.as_secs_f64() * 1e3;
+        let m = mem.total_p50.as_secs_f64() * 1e3;
+        println!(
+            "{:>8} {:>10.3} {:>10.3} {:>10.3} {:>10.3} {:>7.0}% {:>10.3} {:>10.3} {:>7.0}%",
+            fresh.rows,
+            f,
+            fresh.total_p95.as_secs_f64() * 1e3,
+            r,
+            reuse.total_p95.as_secs_f64() * 1e3,
+            (f - r) / f * 100.0,
+            m,
+            mem.total_p95.as_secs_f64() * 1e3,
+            (f - m) / f * 100.0
+        );
+    }
+}
+
+// -- measurement plumbing ---------------------------------------------------
+
+#[derive(Default)]
+struct StageAcc {
+    wall: Vec<Duration>,
+    user: Duration,
+    sys: Duration,
+    io: IoStats
+}
+
+fn measured<T>(acc: &mut StageAcc, f: impl FnOnce() -> anyhow::Result<T>) -> anyhow::Result<(T, Duration)> {
+    let io0 = io_snapshot();
+    let cpu0 = cpu_times()?;
+    let t0 = Instant::now();
+    let out = f();
+    let wall = t0.elapsed();
+    let cpu1 = cpu_times()?;
+    let io1 = io_snapshot();
+
+    acc.wall.push(wall);
+    acc.user += cpu1.0.saturating_sub(cpu0.0);
+    acc.sys += cpu1.1.saturating_sub(cpu0.1);
+    if let (Some(a), Some(b)) = (io0, io1) {
+        acc.io.add_delta(&a, &b);
+    }
+    Ok((out?, wall))
+}
+
+fn cpu_scope() -> &'static str {
+    if cfg!(target_os = "linux") {
+        "current thread (RUSAGE_THREAD)"
+    } else {
+        "process (RUSAGE_SELF; benchmark is single-threaded)"
+    }
+}
+
+fn cpu_times() -> anyhow::Result<(Duration, Duration)> {
+    #[cfg(target_os = "linux")]
+    const WHO: libc::c_int = libc::RUSAGE_THREAD;
+    #[cfg(not(target_os = "linux"))]
+    const WHO: libc::c_int = libc::RUSAGE_SELF;
+
+    // SAFETY: `rusage` is valid when zero-initialized and `getrusage` receives a valid,
+    // writable pointer for the duration of the call.
+    let mut ru: libc::rusage = unsafe { std::mem::zeroed() };
+    // SAFETY: `WHO` is a supported selector for this target and `ru` is writable.
+    anyhow::ensure!(unsafe { libc::getrusage(WHO, &mut ru) } == 0, "getrusage failed");
+    let tv = |t: libc::timeval| Duration::new(t.tv_sec as u64, t.tv_usec as u32 * 1000);
+    Ok((tv(ru.ru_utime), tv(ru.ru_stime)))
+}
+
+#[derive(Default, Clone, Copy)]
+struct IoStats {
+    rchar: u64,
+    wchar: u64,
+    syscr: u64,
+    syscw: u64,
+    read_bytes: u64,
+    write_bytes: u64,
+    cancelled_write_bytes: u64
+}
+
+impl IoStats {
+    fn add_delta(&mut self, before: &IoStats, after: &IoStats) {
+        self.rchar += after.rchar - before.rchar;
+        self.wchar += after.wchar - before.wchar;
+        self.syscr += after.syscr - before.syscr;
+        self.syscw += after.syscw - before.syscw;
+        self.read_bytes += after.read_bytes - before.read_bytes;
+        self.write_bytes += after.write_bytes - before.write_bytes;
+        // cancelled can regress vs a racing writeback; clamp instead of panicking in release
+        self.cancelled_write_bytes += after.cancelled_write_bytes.saturating_sub(before.cancelled_write_bytes);
+    }
+}
+
+fn io_snapshot() -> Option<IoStats> {
+    let s = std::fs::read_to_string("/proc/self/io").ok()?;
+    let get = |k: &str| {
+        s.lines()
+            .find(|l| l.starts_with(k))
+            .and_then(|l| l.split_whitespace().nth(1))
+            .and_then(|v| v.parse().ok())
+            .unwrap_or(0)
+    };
+    Some(IoStats {
+        rchar: get("rchar"),
+        wchar: get("wchar"),
+        syscr: get("syscr"),
+        syscw: get("syscw"),
+        read_bytes: get("read_bytes"),
+        write_bytes: get("write_bytes"),
+        cancelled_write_bytes: get("cancelled_write_bytes")
+    })
+}
+
+fn open_fds() -> usize {
+    std::fs::read_dir("/dev/fd").map(|d| d.count()).unwrap_or(0)
+}
+
+// -- synthetic block --------------------------------------------------------
+
+const TRANSFER_TOPIC: &str = "0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef";
+
+fn hex32(seed: u64) -> String {
+    format!("0x{seed:064x}")
+}
+
+fn addr(seed: u64) -> String {
+    format!("0x{seed:040x}")
+}
+
+/// One block, `n` transactions, one log per transaction. Field content varies by index so
+/// the sort path sees non-degenerate keys; payload sizes are mid-range mainnet-ish.
+fn gen_block_json(n: usize, iteration: usize) -> Value {
+    let series = (iteration as u64).wrapping_mul(1_000_003);
+    let transactions: Vec<Value> = (0..n)
+        .map(|i| {
+            let item = series.wrapping_add(i as u64);
+            json!({
+                "transactionIndex": i as u32,
+                "hash": hex32(0x1000_0000u64.wrapping_add(item)),
+                "nonce": item,
+                "from": addr(0x2222 + (item % 7)),
+                "to": addr(0x3333 + (item % 11)),
+                "input": format!("0x{item:0128x}"),
+                "value": "0xde0b6b3a7640000",
+                "type": 2,
+                "gas": "0x5208",
+                "gasPrice": "0x3b9aca00",
+                "cumulativeGasUsed": format!("0x{:x}", 21_000u64 * (i as u64 + 1)),
+                "effectiveGasPrice": "0x3b9aca00",
+                "gasUsed": "0x5208",
+                "logsBloom": "0x00",
+                "status": 1
+            })
+        })
+        .collect();
+
+    let logs: Vec<Value> = (0..n)
+        .map(|i| {
+            let item = series.wrapping_add(i as u64);
+            json!({
+                "logIndex": i as u32,
+                "transactionIndex": i as u32,
+                "transactionHash": hex32(0x1000_0000u64.wrapping_add(item)),
+                "address": addr(0x4444 + (item % 5)),
+                "data": format!("0x{:0128x}", item.wrapping_mul(31)),
+                "topics": [TRANSFER_TOPIC, hex32(item), hex32(item ^ 0xffff)]
+            })
+        })
+        .collect();
+
+    json!({
+        "header": {
+            "number": 20_000_000u64.wrapping_add(iteration as u64),
+            "hash": hex32(0xb10cu64.wrapping_add(series)),
+            "parentHash": hex32(0xb10bu64.wrapping_add(series)),
+            "timestamp": 1_760_000_000i64.saturating_add(iteration as i64),
+            "transactionsRoot": hex32(1),
+            "receiptsRoot": hex32(2),
+            "stateRoot": hex32(3),
+            "logsBloom": "0x00",
+            "sha3Uncles": hex32(4),
+            "extraData": "0x",
+            "miner": addr(0x1111),
+            "size": 100_000u64,
+            "gasLimit": "0x1c9c380",
+            "gasUsed": "0x5208"
+        },
+        "transactions": transactions,
+        "logs": logs,
+        // present-but-empty keeps the full data-availability mask → all 5 tables built,
+        // matching a full-mask prod dataset
+        "traces": [],
+        "stateDiffs": []
+    })
+}
diff --git a/crates/data/benches/results/2026-07-16-cc-par-02-e96c877.md b/crates/data/benches/results/2026-07-16-cc-par-02-e96c877.md
new file mode 100644
index 00000000..95a52c6d
--- /dev/null
+++ b/crates/data/benches/results/2026-07-16-cc-par-02-e96c877.md
@@ -0,0 +1,134 @@
+# `flush_spill` benchmark — cc-par-02, 2026-07-16
+
+## Provenance
+
+- Git commit: `e96c8774f1cfa7074a868f33a0f9afed29541e5c`
+- Image requested by the Job: `subsquid/data-flush-bench:e96c877`
+- Immutable image ID reported by kubelet:
+  `docker.io/subsquid/data-flush-bench@sha256:56de635f5e2cb82ef56a7b5cdde7bd24679a0e4bff8934fdc9eacd57688f611a`
+- Node: `cc-par-02`, Linux `6.1.0-40-amd64`
+- Job: `network-hotblocks-mainnet-internal/flush-spill-bench`
+- Pod: `flush-spill-bench-qnfkk`
+- Job interval: `2026-07-16T15:00:51Z`–`2026-07-16T15:02:34Z`
+- `RLIMIT_NOFILE`: 65,535
+- CPU accounting: current thread, `getrusage(RUSAGE_THREAD)`
+- `/dev/shm`: 4 GiB memory-backed `emptyDir`
+- Result: Job completed successfully; cross-mode equality passed on every measured
+  iteration in both runs; fd counts stayed stable in reuse and mem modes.
+
+The complete stdout was retrieved and checked before the Job was deleted, but the object
+disappeared before it could be copied into this directory. The exact aggregate tables and
+head-stage breakdown needed for the investigation are preserved below. This file is not a
+raw-log substitute.
+
+## Job shape
+
+```yaml
+apiVersion: batch/v1
+kind: Job
+metadata:
+  name: flush-spill-bench
+  namespace: network-hotblocks-mainnet-internal
+spec:
+  backoffLimit: 0
+  ttlSecondsAfterFinished: 172800
+  template:
+    spec:
+      nodeSelector:
+        kubernetes.io/hostname: cc-par-02
+      restartPolicy: Never
+      tolerations:
+        - key: dedicated
+          operator: Equal
+          value: paris-az-nodes
+          effect: NoSchedule
+      containers:
+        - name: bench
+          image: subsquid/data-flush-bench:e96c877
+          command: [bash, -c]
+          args:
+            - |
+              set -e
+              TMPDIR=/tmp /app/flush_spill
+              TMPDIR=/dev/shm /app/flush_spill
+          resources:
+            requests: {cpu: "2", memory: 4Gi}
+            limits: {memory: 16Gi}
+          volumeMounts:
+            - {name: shm, mountPath: /dev/shm}
+      volumes:
+        - name: shm
+          emptyDir: {medium: Memory, sizeLimit: 4Gi}
+```
+
+The two runs were sequential in one process. Mode order within each row-size case rotated
+through all six fresh/reuse/mem permutations. The tmpfs comparison can therefore still be
+affected by run-order and machine-load history; it is a filesystem-sensitivity bound, not
+a controlled physical-disk attribution.
+
+## `/tmp` (container overlay)
+
+All values are milliseconds. `saves` is relative to fresh p50.
+
+| rows | fresh p50 | fresh p95 | reuse p50 | reuse p95 | reuse saves | mem p50 | mem p95 | mem saves |
+|---:|---:|---:|---:|---:|---:|---:|---:|---:|
+| 1 | 52.433 | 81.269 | 11.774 | 14.701 | 78% | 0.224 | 0.314 | 99.6% |
+| 10 | 52.615 | 102.762 | 11.775 | 14.611 | 78% | 0.249 | 0.320 | 99.5% |
+| 100 | 53.410 | 65.539 | 11.852 | 14.075 | 78% | 0.453 | 0.574 | 99.2% |
+| 1,000 | 56.070 | 71.893 | 15.823 | 18.264 | 72% | 2.127 | 3.088 | 96.2% |
+| 10,000 | 116.691 | 372.746 | 59.492 | 76.858 | 49% | 22.097 | 31.148 | 81.1% |
+| 100,000 | 606.770 | 654.354 | 487.961 | 536.642 | 20% | 293.257 | 296.741 | 51.7% |
+
+Head (`rows=1`) stage detail:
+
+| mode/stage | p50 ms | p95 ms | mean ms | mean user ms | mean sys ms | dirty MB | cancelled MB |
+|---|---:|---:|---:|---:|---:|---:|---:|
+| fresh / push | 0.011 | 0.017 | 0.011 | 0.031 | 0.000 | 0.00 | 0.00 |
+| fresh / construct | 24.316 | 27.320 | 24.640 | 0.838 | 23.098 | 0.00 | 0.00 |
+| fresh / submit+finish | 6.726 | 8.939 | 6.875 | 0.625 | 6.175 | 1.46 | 0.00 |
+| fresh / readback | 0.701 | 0.935 | 0.767 | 0.195 | 0.554 | 0.00 | 0.00 |
+| fresh / drop | 20.216 | 46.453 | 26.504 | 0.582 | 14.592 | 0.00 | 0.00 |
+| reuse / reconstruct | 4.395 | 5.525 | 4.556 | 0.247 | 4.268 | 0.00 | 1.46 |
+| reuse / submit+finish | 6.476 | 8.955 | 6.708 | 0.832 | 4.680 | 1.46 | 0.00 |
+| mem / prepare | 0.110 | 0.152 | 0.116 | 0.063 | 0.021 | 0.00 | 0.00 |
+| mem / readback | 0.091 | 0.128 | 0.096 | 0.049 | 0.003 | 0.00 | 0.00 |
+
+Fresh fixed/marginal fit from the benchmark output:
+
+- Fixed total at one row: 52.433 ms.
+- Marginal total from 1 to 100,000 rows: 5.543 ms per 1,000 rows.
+- Crossover reported by the two-endpoint fit: about 9,459 rows.
+
+## `/dev/shm` (tmpfs)
+
+| rows | fresh p50 | fresh p95 | reuse p50 | reuse p95 | reuse saves | mem p50 | mem p95 | mem saves |
+|---:|---:|---:|---:|---:|---:|---:|---:|---:|
+| 1 | 7.912 | 9.874 | 4.295 | 5.441 | 46% | 0.192 | 0.261 | 97.6% |
+| 10 | 7.764 | 9.176 | 4.149 | 5.071 | 47% | 0.205 | 0.264 | 97.4% |
+| 100 | 8.281 | 9.769 | 4.286 | 5.424 | 48% | 0.386 | 0.440 | 95.3% |
+| 1,000 | 10.264 | 12.108 | 6.712 | 7.633 | 35% | 1.909 | 2.492 | 81.4% |
+| 10,000 | 38.777 | 42.367 | 33.879 | 37.179 | 13% | 20.950 | 27.954 | 46.0% |
+| 100,000 | 339.673 | 347.370 | 311.989 | 364.702 | 8% | 263.200 | 285.678 | 22.5% |
+
+At one row, moving fresh from overlay to tmpfs reduced p50 from 52.433 to 7.912 ms
+(84.9%). Mem was nearly filesystem-insensitive: 0.224 vs 0.192 ms. This supports the
+fixed filesystem-implementation diagnosis without proving that 84.9% was physical device
+I/O.
+
+## Interpretation limits
+
+- `/proc/self/io::write_bytes` is page-dirtying/storage-layer accounting, not observed
+  device traffic. `cancelled_write_bytes` records dirty bytes cancelled before writeback.
+- Reading `/proc/self/io` for instrumentation itself contributes the five `syscr` calls and
+  roughly 0.1 KiB `rchar` baseline shown on stages that otherwise perform no I/O.
+- CPU snapshots bracket the measured closure, but the wall timer excludes the second
+  `getrusage` call. At sub-millisecond scale, timer/accounting granularity can therefore
+  make mean CPU and mean wall values look slightly inconsistent; p50 wall is the primary
+  comparison.
+- The 100,000-row p95 is only the maximum of three iterations. The 10,000-row p95 has ten
+  samples. Treat both as diagnostics, not stable tail estimates.
+- Product `WriteStage::Prepare` times only `DataBuilder::finish()`. The benchmark total also
+  includes push, full readback, and drop to compare the complete prepared-chunk lifecycle.
+  For the new head prepare path itself, the rows=1 `prepare(mem)` p50 is 0.110 ms.
+- Peak RSS was not sampled. The 16 GiB Job limit only proves this synthetic sweep stayed
+  below that coarse bound.
diff --git a/crates/data/tests/in_memory_prepare_real_schemas.rs b/crates/data/tests/in_memory_prepare_real_schemas.rs
new file mode 100644
index 00000000..b7c5f10c
--- /dev/null
+++ b/crates/data/tests/in_memory_prepare_real_schemas.rs
@@ -0,0 +1,663 @@
+//! Losslessness checks for the production chunk builders.
+//!
+//! Every supported dataset kind is populated through its real JSON model and real table
+//! builders, then prepared through both the spill and in-memory paths. The complete Arrow
+//! schemas and record batches must be identical.
+
+use std::{collections::BTreeMap, sync::Mutex};
+
+use serde::de::DeserializeOwned;
+use serde_json::{json, Value};
+use sqd_data::{
+    bitcoin::{model as bitcoin, tables::BitcoinChunkBuilder},
+    evm::{model as evm, tables::EvmChunkBuilder},
+    hyperliquid_fills::{model as hyperliquid_fills, tables::HyperliquidFillsChunkBuilder},
+    hyperliquid_replica_cmds::{model as hyperliquid_replica_cmds, tables::HyperliquidReplicaCmdsChunkBuilder},
+    solana::{model as solana, tables::SolanaChunkBuilder},
+    tron::{model as tron, tables::TronChunkBuilder}
+};
+use sqd_data_core::{BlockChunkBuilder, PreparedChunk};
+
+static SPILL_LOCK: Mutex<()> = Mutex::new(());
+
+fn parse<T: DeserializeOwned>(value: Value) -> T {
+    serde_json::from_value(value).unwrap()
+}
+
+fn assert_lossless<B>(mut builder: B, block: B::Block, expected_rows: &[(&str, usize)])
+where
+    B: BlockChunkBuilder
+{
+    let _guard = SPILL_LOCK.lock().unwrap();
+    raise_nofile_limit();
+    builder.push(&block).unwrap();
+
+    let mut processor = builder.new_chunk_processor().unwrap();
+    builder.submit_to_processor(&mut processor).unwrap();
+    let disk = processor.finish().unwrap();
+    let mem = builder.prepare_in_memory().unwrap();
+
+    assert_eq!(builder.max_num_rows(), 0, "in-memory prepare did not clear the builder");
+    assert_chunks_equal(disk, mem, expected_rows);
+}
+
+fn raise_nofile_limit() {
+    const TARGET: libc::rlim_t = 8192;
+
+    // SAFETY: `rlimit` is valid when zero-initialized and the pointer remains writable for
+    // the duration of `getrlimit`.
+    let mut limit: libc::rlimit = unsafe { std::mem::zeroed() };
+    // SAFETY: `limit` points to valid writable storage and RLIMIT_NOFILE is supported on
+    // every Unix target on which this crate runs.
+    assert_eq!(unsafe { libc::getrlimit(libc::RLIMIT_NOFILE, &mut limit) }, 0);
+    if limit.rlim_cur >= TARGET {
+        return;
+    }
+
+    limit.rlim_cur = TARGET.min(limit.rlim_max);
+    // SAFETY: the initialized soft limit does not exceed the hard limit and the pointer is
+    // valid for the duration of the call.
+    assert_eq!(unsafe { libc::setrlimit(libc::RLIMIT_NOFILE, &limit) }, 0);
+    assert!(
+        limit.rlim_cur >= 1024,
+        "hard fd limit is too low for the EVM spill path"
+    );
+}
+
+fn assert_chunks_equal(mut disk: PreparedChunk, mut mem: PreparedChunk, expected_rows: &[(&str, usize)]) {
+    let expected: BTreeMap<_, _> = expected_rows.iter().copied().collect();
+    assert_eq!(
+        disk.len(),
+        expected.len(),
+        "spill path returned an unexpected table set"
+    );
+    assert_eq!(
+        mem.len(),
+        expected.len(),
+        "memory path returned an unexpected table set"
+    );
+
+    for (name, rows) in expected {
+        let disk_table = disk
+            .get_mut(name)
+            .unwrap_or_else(|| panic!("spill path lost table {name}"));
+        let mem_table = mem
+            .get_mut(name)
+            .unwrap_or_else(|| panic!("memory path lost table {name}"));
+
+        assert_eq!(disk_table.schema(), mem_table.schema(), "schema differs for {name}");
+        assert_eq!(disk_table.num_rows(), rows, "spill row count differs for {name}");
+        assert_eq!(mem_table.num_rows(), rows, "memory row count differs for {name}");
+
+        let disk_batch = disk_table.read_record_batch(0, rows).unwrap();
+        let mem_batch = mem_table.read_record_batch(0, rows).unwrap();
+        assert_eq!(disk_batch, mem_batch, "record batch differs for {name}");
+    }
+}
+
+fn hex(seed: u64) -> String {
+    format!("0x{seed:064x}")
+}
+
+fn address(seed: u64) -> String {
+    format!("0x{seed:040x}")
+}
+
+#[test]
+fn evm_real_schema_is_lossless() {
+    let block: evm::Block = parse(json!({
+        "header": {
+            "number": 20_000_000,
+            "hash": hex(1),
+            "parentHash": hex(0),
+            "timestamp": 1_760_000_000,
+            "transactionsRoot": hex(2),
+            "receiptsRoot": hex(3),
+            "stateRoot": hex(4),
+            "logsBloom": "0x00",
+            "sha3Uncles": hex(5),
+            "extraData": "0x1234",
+            "miner": address(1),
+            "size": 1234,
+            "gasLimit": "0x1c9c380",
+            "gasUsed": "0x5208",
+            "withdrawals": [{
+                "address": address(2),
+                "amount": "0x1",
+                "index": "0x2",
+                "validatorIndex": "0x3"
+            }],
+            "uncles": [hex(6)]
+        },
+        "transactions": [
+            {
+                "transactionIndex": 0,
+                "hash": hex(10),
+                "nonce": 7,
+                "from": address(3),
+                "to": address(4),
+                "input": "0x11223344aabbccdd",
+                "value": "0x5",
+                "type": 3,
+                "gas": "0x5208",
+                "gasPrice": "0x3b9aca00",
+                "maxFeePerGas": "0x4",
+                "maxPriorityFeePerGas": "0x2",
+                "accessList": [{
+                    "address": address(5),
+                    "storageKeys": [hex(11), hex(12)]
+                }],
+                "blobVersionedHashes": [hex(13)],
+                "authorizationList": [{
+                    "chainId": "1",
+                    "address": address(6),
+                    "nonce": "9",
+                    "yParity": 1,
+                    "r": hex(14),
+                    "s": hex(15)
+                }],
+                "cumulativeGasUsed": "0x5208",
+                "effectiveGasPrice": "0x3b9aca00",
+                "gasUsed": "0x5208",
+                "logsBloom": "0x00",
+                "status": 1
+            },
+            {
+                "transactionIndex": 1,
+                "hash": hex(16),
+                "nonce": 8,
+                "from": address(10),
+                "type": 118,
+                "gas": "0x8000",
+                "calls": [
+                    {"to": address(11), "value": "0x1", "input": "0xabcdef01"},
+                    {"value": "0x0", "input": "0x6000"}
+                ],
+                "nonceKey": "0x02",
+                "feeToken": address(12),
+                "feePayerSignature": {"v": 27, "r": hex(17), "s": hex(18)},
+                "signature": {
+                    "userAddress": address(13),
+                    "version": "v1",
+                    "signature": {
+                        "type": "p256",
+                        "r": hex(19),
+                        "s": hex(20),
+                        "pubKeyX": hex(21),
+                        "pubKeyY": hex(22),
+                        "preHash": true
+                    }
+                },
+                "validBefore": "0x100",
+                "validAfter": "0x10",
+                "aaAuthorizationList": [{
+                    "chainId": "0x1",
+                    "address": address(14),
+                    "nonce": 3,
+                    "signature": {
+                        "type": "secp256k1",
+                        "r": hex(23),
+                        "s": hex(24),
+                        "yParity": 1
+                    }
+                }],
+                "keyAuthorization": {
+                    "chainId": "0x1",
+                    "keyType": "webauthn",
+                    "keyId": hex(25),
+                    "expiry": "0x200",
+                    "limits": [{"token": address(15), "limit": "0xff"}],
+                    "signature": {
+                        "type": "webAuthn",
+                        "r": hex(26),
+                        "s": hex(27),
+                        "pubKeyX": hex(28),
+                        "pubKeyY": hex(29),
+                        "webauthnData": "0xaabbcc"
+                    }
+                },
+                "cumulativeGasUsed": "0xd208",
+                "gasUsed": "0x8000",
+                "logsBloom": "0x00",
+                "status": 1
+            }
+        ],
+        "logs": [{
+            "logIndex": 0,
+            "transactionIndex": 0,
+            "transactionHash": hex(10),
+            "address": address(7),
+            "data": "0xdeadbeef",
+            "topics": [hex(20), hex(21)]
+        }],
+        "traces": [
+            {
+                "transactionIndex": 0,
+                "traceAddress": [0],
+                "subtraces": 0,
+                "type": "call",
+                "action": {
+                    "from": address(3),
+                    "to": address(4),
+                    "value": "0x5",
+                    "gas": "0x5208",
+                    "input": "0x11223344",
+                    "callType": "call"
+                },
+                "result": {"gasUsed": "0x5100", "output": "0xaabb"}
+            },
+            {
+                "transactionIndex": 0,
+                "traceAddress": [1],
+                "subtraces": 0,
+                "type": "create",
+                "action": {
+                    "from": address(3),
+                    "value": "0x0",
+                    "gas": "0x10000",
+                    "init": "0x6000"
+                },
+                "result": {"gasUsed": "0x9000", "code": "0x6001", "address": address(8)}
+            },
+            {
+                "transactionIndex": 0,
+                "traceAddress": [2],
+                "subtraces": 0,
+                "type": "selfdestruct",
+                "action": {
+                    "address": address(16),
+                    "refundAddress": address(17),
+                    "balance": "0x42"
+                }
+            },
+            {
+                "transactionIndex": 0,
+                "traceAddress": [3],
+                "subtraces": 0,
+                "type": "reward",
+                "action": {
+                    "author": address(18),
+                    "value": "0x2a",
+                    "rewardType": "block"
+                }
+            }
+        ],
+        "stateDiffs": [
+            {
+                "transactionIndex": 0,
+                "address": address(9),
+                "key": "balance",
+                "kind": "+",
+                "next": "0x10"
+            },
+            {
+                "transactionIndex": 0,
+                "address": address(9),
+                "key": hex(22),
+                "kind": "*",
+                "prev": "0x01",
+                "next": "0x02"
+            }
+        ]
+    }));
+
+    assert_lossless(
+        EvmChunkBuilder::new(),
+        block,
+        &[
+            ("blocks", 1),
+            ("transactions", 2),
+            ("logs", 1),
+            ("traces", 4),
+            ("statediffs", 2)
+        ]
+    );
+}
+
+#[test]
+fn solana_real_schema_is_lossless() {
+    let accounts = [
+        "11111111111111111111111111111111",
+        "Vote111111111111111111111111111111111111111",
+        "SysvarRent111111111111111111111111111111111",
+        "SysvarC1ock11111111111111111111111111111111"
+    ];
+    let block: solana::Block = parse(json!({
+        "header": {
+            "number": 300_000_000,
+            "hash": "5HueCGU8rMjxEXxiPuD5BDu",
+            "parentNumber": 299_999_998,
+            "parentHash": "4vJ9JU1bJJE96FWSJKvHs",
+            "height": 280_000_000,
+            "timestamp": 1_760_000_000
+        },
+        "accounts": accounts,
+        "transactions": [{
+            "transactionIndex": 0,
+            "version": "legacy",
+            "accountKeys": [0, 1],
+            "addressTableLookups": [{
+                "accountKey": 2,
+                "readonlyIndexes": [1, 2],
+                "writableIndexes": [3]
+            }],
+            "numReadonlySignedAccounts": 0,
+            "numReadonlyUnsignedAccounts": 1,
+            "numRequiredSignatures": 1,
+            "recentBlockhash": "4vJ9JU1bJJE96FWSJKvHs",
+            "signatures": ["3Bxs4NN8M2Yn4TLb"],
+            "err": {"InstructionError": [0, "Custom"]},
+            "computeUnitsConsumed": "12345",
+            "costUnits": "13000",
+            "fee": "5000",
+            "loadedAddresses": {"readonly": [2], "writable": [3]},
+            "hasDroppedLogMessages": false
+        }],
+        "instructions": [{
+            "transactionIndex": 0,
+            "instructionAddress": [0, 1],
+            "programId": 1,
+            "accounts": [0, 1, 2, 3],
+            "data": "11111111111111111",
+            "computeUnitsConsumed": "1000",
+            "error": null,
+            "isCommitted": true,
+            "hasDroppedLogMessages": false
+        }],
+        "logs": [{
+            "transactionIndex": 0,
+            "logIndex": 0,
+            "instructionAddress": [0, 1],
+            "programId": 1,
+            "kind": "log",
+            "message": "Program log: all fields survive"
+        }],
+        "balances": [{"transactionIndex": 0, "account": 0, "pre": "100", "post": "90"}],
+        "tokenBalances": [{
+            "transactionIndex": 0,
+            "account": 0,
+            "preMint": 1,
+            "postMint": 1,
+            "preDecimals": 6,
+            "postDecimals": 6,
+            "preProgramId": 2,
+            "postProgramId": 2,
+            "preOwner": 3,
+            "postOwner": 3,
+            "preAmount": "10",
+            "postAmount": "9"
+        }],
+        "rewards": [{
+            "pubkey": 1,
+            "lamports": "42",
+            "postBalance": "1042",
+            "rewardType": "voting",
+            "commission": 7
+        }]
+    }));
+
+    assert_lossless(
+        SolanaChunkBuilder::new(),
+        block,
+        &[
+            ("blocks", 1),
+            ("transactions", 1),
+            ("instructions", 1),
+            ("logs", 1),
+            ("balances", 1),
+            ("token_balances", 1),
+            ("rewards", 1)
+        ]
+    );
+}
+
+#[test]
+fn bitcoin_real_schema_is_lossless() {
+    let block: bitcoin::Block = parse(json!({
+        "header": {
+            "number": 900_000,
+            "hash": hex(100),
+            "parentHash": hex(99),
+            "timestamp": 1_760_000_000,
+            "medianTime": 1_759_999_000,
+            "version": 2,
+            "merkleRoot": hex(101),
+            "nonce": 42,
+            "target": hex(102),
+            "bits": "0x1d00ffff",
+            "difficulty": 123.5,
+            "chainWork": hex(103),
+            "strippedSize": 900,
+            "size": 1000,
+            "weight": 3900
+        },
+        "transactions": [{
+            "hex": "0x010203",
+            "txid": hex(110),
+            "hash": hex(111),
+            "size": 200,
+            "vsize": 150,
+            "weight": 600,
+            "version": 2,
+            "locktime": 0,
+            "vin": [
+                {"coinbase": "0x03abcdef", "sequence": 4294967295u64, "txInWitness": ["0x01"]},
+                {
+                    "txid": hex(112),
+                    "vout": 1,
+                    "scriptSig": {"hex": "0x160014", "asm": "0 0011"},
+                    "sequence": 4294967294u64,
+                    "txInWitness": ["0xaa", "0xbb"],
+                    "prevout": {
+                        "generated": false,
+                        "height": 899_999,
+                        "value": 0.125,
+                        "scriptPubKey": {
+                            "hex": "0x0014",
+                            "asm": "0 abcd",
+                            "desc": "addr(test)",
+                            "type": "witness_v0_keyhash",
+                            "address": "bc1qexample"
+                        }
+                    }
+                }
+            ],
+            "vout": [{
+                "value": 0.124,
+                "n": 0,
+                "scriptPubKey": {
+                    "hex": "0x76a9",
+                    "asm": "OP_DUP OP_HASH160",
+                    "desc": "pkh(test)",
+                    "type": "pubkeyhash",
+                    "address": "1Example"
+                }
+            }]
+        }]
+    }));
+
+    assert_lossless(
+        BitcoinChunkBuilder::new(),
+        block,
+        &[("blocks", 1), ("transactions", 1), ("inputs", 2), ("outputs", 1)]
+    );
+}
+
+#[test]
+fn tron_real_schema_is_lossless() {
+    let block: tron::Block = parse(json!({
+        "header": {
+            "height": 70_000_000,
+            "hash": hex(200),
+            "parentHash": hex(199),
+            "txTrieRoot": hex(201),
+            "version": 29,
+            "timestamp": 1_760_000_000_000i64,
+            "witnessAddress": "0x41aa",
+            "witnessSignature": "0xbb"
+        },
+        "transactions": [{
+            "transactionIndex": 0,
+            "hash": hex(210),
+            "ret": [{"contractRet": "SUCCESS"}],
+            "signature": ["0xdead", "0xbeef"],
+            "type": "TriggerSmartContract",
+            "parameter": {"value": {
+                "owner_address": "0x41aa",
+                "contract_address": "0x41bb",
+                "data": "0xa9059cbb0011"
+            }},
+            "permissionId": 2,
+            "refBlockBytes": "0x1234",
+            "refBlockHash": "0xabcd",
+            "feeLimit": "1000000",
+            "expiration": 1_760_000_060_000i64,
+            "timestamp": "1760000000000",
+            "rawDataHex": "0x01020304",
+            "fee": "123",
+            "contractResult": "0x01",
+            "contractAddress": "0x41bb",
+            "result": "SUCCESS",
+            "energyFee": "10",
+            "energyUsage": "20",
+            "energyUsageTotal": "30",
+            "netUsage": "40",
+            "netFee": "50"
+        }],
+        "logs": [{
+            "transactionIndex": 0,
+            "logIndex": 0,
+            "address": "0x41bb",
+            "data": "0x0102",
+            "topics": [hex(211), hex(212)]
+        }],
+        "internalTransactions": [{
+            "transactionIndex": 0,
+            "internalTransactionIndex": 0,
+            "hash": hex(213),
+            "callerAddress": "0x41aa",
+            "transferToAddress": "0x41cc",
+            "callValueInfo": [{"callValue": "7", "tokenId": "1002000"}],
+            "note": "0x63616c6c",
+            "rejected": false,
+            "extra": "0x99"
+        }]
+    }));
+
+    assert_lossless(
+        TronChunkBuilder::new(),
+        block,
+        &[
+            ("blocks", 1),
+            ("transactions", 1),
+            ("logs", 1),
+            ("internal_transactions", 1)
+        ]
+    );
+}
+
+#[test]
+fn hyperliquid_fills_real_schema_is_lossless() {
+    let block: hyperliquid_fills::Block = parse(json!({
+        "header": {
+            "number": 10_000,
+            "hash": hex(300),
+            "parentHash": hex(299),
+            "timestamp": 1_760_000_000_000i64
+        },
+        "fills": [{
+            "fillIndex": 0,
+            "user": address(20),
+            "coin": "BTC",
+            "px": 123.25,
+            "sz": 0.5,
+            "side": "B",
+            "time": 1_760_000_000_000i64,
+            "startPosition": 1.25,
+            "dir": "Open Long",
+            "closedPnl": 0.25,
+            "hash": hex(301),
+            "oid": 42,
+            "crossed": true,
+            "fee": 0.01,
+            "builderFee": 0.001,
+            "tid": 43,
+            "cloid": "client-order-id",
+            "feeToken": "USDC",
+            "builder": address(21),
+            "twapId": 44
+        }]
+    }));
+
+    assert_lossless(
+        HyperliquidFillsChunkBuilder::new(),
+        block,
+        &[("blocks", 1), ("fills", 1)]
+    );
+}
+
+#[test]
+fn hyperliquid_replica_commands_real_schema_is_lossless() {
+    let block: hyperliquid_replica_cmds::Block = parse(json!({
+        "header": {
+            "height": 10_001,
+            "hash": hex(400),
+            "parentHash": hex(399),
+            "round": 12,
+            "parentRound": 11,
+            "proposer": address(30),
+            "timestamp": 1_760_000_000_000i64,
+            "hardfork": {"version": 3, "round": 10}
+        },
+        "actions": [
+            {
+                "actionIndex": 0,
+                "signature": {"r": hex(401), "s": hex(402), "v": 27},
+                "action": {"type": "order", "orders": [{"a": 1, "c": "cloid-a"}, {"a": 2}]},
+                "nonce": 100,
+                "vaultAddress": address(31),
+                "user": address(32),
+                "status": "ok",
+                "response": {"status": "ok", "data": {"statuses": ["resting"]}}
+            },
+            {
+                "actionIndex": 1,
+                "signature": {"r": hex(403), "s": hex(404), "v": 28},
+                "action": {"type": "cancelByCloid", "cancels": [{"asset": 1, "cloid": "cloid-a"}]},
+                "nonce": 101,
+                "status": "err",
+                "response": {"status": "err", "message": "already filled"}
+            },
+            {
+                "actionIndex": 2,
+                "signature": {"r": hex(405), "s": hex(406), "v": 27},
+                "action": {"type": "cancel", "cancels": [{"a": 3}, {"a": 3}, {"a": 4}]},
+                "nonce": 102,
+                "user": address(32),
+                "status": "ok",
+                "response": {"status": "ok"}
+            },
+            {
+                "actionIndex": 3,
+                "signature": {"r": hex(407), "s": hex(408), "v": 28},
+                "action": {
+                    "type": "batchModify",
+                    "modifies": [
+                        {"oid": 1, "order": {"a": 5, "c": "cloid-b"}},
+                        {"oid": 2, "order": {"a": 6}}
+                    ]
+                },
+                "nonce": 103,
+                "vaultAddress": address(31),
+                "status": "ok",
+                "response": {"status": "ok"}
+            }
+        ]
+    }));
+
+    assert_lossless(
+        HyperliquidReplicaCmdsChunkBuilder::new(),
+        block,
+        &[("blocks", 1), ("actions", 4)]
+    );
+}
diff --git a/crates/hotblocks-harness/src/sut.rs b/crates/hotblocks-harness/src/sut.rs
index 876eb1d0..40d3ad84 100644
--- a/crates/hotblocks-harness/src/sut.rs
+++ b/crates/hotblocks-harness/src/sut.rs
@@ -97,6 +97,10 @@ impl Sut {
         self.dir.path().join("sut.log")
     }
 
+    pub fn pid(&self) -> Option<u32> {
+        self.child.as_ref().and_then(|c| c.id())
+    }
+
     /// Boot a stopped process again against the same database — the CT-2 restart primitive.
     pub async fn restart(&mut self) -> Result<()> {
         if self.child.is_some() {
diff --git a/crates/hotblocks/src/dataset_controller/ingest_generic.rs b/crates/hotblocks/src/dataset_controller/ingest_generic.rs
index 621fd113..ebff077e 100644
--- a/crates/hotblocks/src/dataset_controller/ingest_generic.rs
+++ b/crates/hotblocks/src/dataset_controller/ingest_generic.rs
@@ -50,6 +50,9 @@ impl Display for NewChunk {
     }
 }
 
+/// `maybe_flush` spills builder contents to the processor beyond this size.
+const SPILL_BOUND_BYTES: usize = 30 * 1024 * 1024;
+
 struct DataBuilder<CB> {
     builder: CB,
     processor: Option<ChunkProcessor>
@@ -86,6 +89,12 @@ impl<CB: BlockChunkBuilder> DataBuilder<CB> {
     }
 
     pub fn finish(&mut self) -> anyhow::Result<PreparedChunk> {
+        if self.processor.is_none() && self.builder.byte_size() <= SPILL_BOUND_BYTES {
+            // no spill and within the spill bound — skip the temp files. The bound is
+            // re-checked here: a row-count-triggered flush can carry an oversized final
+            // block that maybe_flush's byte check never saw.
+            return self.builder.prepare_in_memory();
+        }
         self.flush_to_processor()?;
         self.processor.take().unwrap().finish()
     }
@@ -225,7 +234,7 @@ where
         if self.builder_ref().num_rows() > 200_000 {
             return self.flush().await;
         }
-        if self.builder_ref().in_memory_buffered_bytes() > 30 * 1024 * 1024 {
+        if self.builder_ref().in_memory_buffered_bytes() > SPILL_BOUND_BYTES {
             return self.with_blocking_builder(|b| b.flush_to_processor()).await;
         }
         Ok(())
@@ -315,3 +324,70 @@ where
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use sqd_data::hyperliquid_fills::{model::Block, tables::HyperliquidFillsChunkBuilder};
+
+    use super::*;
+
+    fn fills_block() -> Block {
+        serde_json::from_value(serde_json::json!({
+            "header": {
+                "number": 10_000,
+                "hash": "0xabc",
+                "parentHash": "0xabb",
+                "timestamp": 1_760_000_000_000i64
+            },
+            "fills": [{
+                "fillIndex": 0,
+                "user": "0x1111111111111111111111111111111111111111",
+                "coin": "BTC",
+                "px": 123.25,
+                "sz": 0.5,
+                "side": "B",
+                "time": 1_760_000_000_000i64,
+                "startPosition": 1.25,
+                "dir": "Open Long",
+                "closedPnl": 0.25,
+                "hash": "0x2222222222222222222222222222222222222222",
+                "oid": 42,
+                "crossed": true,
+                "fee": 0.01,
+                "tid": 43,
+                "feeToken": "USDC",
+                "cloid": "x".repeat(64 * 1024)
+            }]
+        }))
+        .unwrap()
+    }
+
+    fn unspilled(over_bytes: usize) -> DataBuilder<HyperliquidFillsChunkBuilder> {
+        let mut b = DataBuilder::new(HyperliquidFillsChunkBuilder::new());
+        let block = fills_block();
+        while b.in_memory_buffered_bytes() <= over_bytes {
+            b.push_block(&block).unwrap();
+        }
+        b
+    }
+
+    /// A row-count-triggered flush can hand `finish` an unspilled builder above the spill
+    /// bound — `maybe_flush`'s byte check never saw the final block. Such a chunk must
+    /// spill instead of being copied in memory. The two paths are told apart by
+    /// `into_processor`, which in-memory tables refuse.
+    #[test]
+    fn oversized_unspilled_chunk_spills_at_finish() {
+        let mut b = unspilled(SPILL_BOUND_BYTES);
+        let mut chunk = b.finish().unwrap();
+        let (_, table) = chunk.pop_first().unwrap();
+        assert!(table.into_processor().is_ok(), "expected the spill path");
+    }
+
+    #[test]
+    fn bounded_unspilled_chunk_is_prepared_in_memory() {
+        let mut b = unspilled(1);
+        let mut chunk = b.finish().unwrap();
+        let (_, table) = chunk.pop_first().unwrap();
+        assert!(table.into_processor().is_err(), "expected the in-memory path");
+    }
+}
diff --git a/crates/hotblocks/tests/ct1_happy_path.rs b/crates/hotblocks/tests/ct1_happy_path.rs
index a1cd6955..b32fe589 100644
--- a/crates/hotblocks/tests/ct1_happy_path.rs
+++ b/crates/hotblocks/tests/ct1_happy_path.rs
@@ -9,6 +9,7 @@
 use std::{sync::Arc, time::Duration};
 
 use anyhow::Result;
+use serde_json::Value;
 use sqd_hotblocks_harness::{
     chain::{Chain, Evm, HlFills, Solana},
     driver::FollowStep,
@@ -24,6 +25,49 @@ async fn ct1_evm() -> Result<()> {
     ct1(Arc::new(Evm), Numbering::Dense).await
 }
 
+struct EvmChangingOptionalMask;
+
+impl Chain for EvmChangingOptionalMask {
+    fn config_kind(&self) -> &'static str {
+        Evm.config_kind()
+    }
+
+    fn storage_kind(&self) -> &'static str {
+        Evm.storage_kind()
+    }
+
+    fn dialect(&self) -> &'static str {
+        Evm.dialect()
+    }
+
+    fn source_block(&self, block: &sqd_hotblocks_harness::types::Block) -> Value {
+        let mut value = Evm.source_block(block);
+        if block.number % 2 == 0 {
+            value
+                .as_object_mut()
+                .expect("EVM source block is an object")
+                .remove("traces");
+        }
+        value
+    }
+
+    fn scan_query(&self, from: u64, to: Option<u64>, expected_parent: Option<&str>) -> Value {
+        Evm.scan_query(from, to, expected_parent)
+    }
+
+    fn expected_emission(&self, block: &sqd_hotblocks_harness::types::Block) -> Value {
+        Evm.expected_emission(block)
+    }
+}
+
+/// Every block flips between a full EVM data mask and one without traces. The ingest path
+/// must flush at every transition without dropping or duplicating any blocks or required
+/// tables.
+#[tokio::test(flavor = "multi_thread")]
+async fn ct1_evm_changing_optional_mask() -> Result<()> {
+    ct1(Arc::new(EvmChangingOptionalMask), Numbering::Dense).await
+}
+
 /// Solana numbers blocks by time-based slots, and a slot that produced nothing leaves a hole.
 /// The window is still one chain — carried by `parentNumber`, not by the numbering (INV-1/2).
 /// The service links batches and chunks by hash and never by number, so it must not care.
diff --git a/crates/hotblocks/tests/peak_rss.rs b/crates/hotblocks/tests/peak_rss.rs
new file mode 100644
index 00000000..0d3e4fab
--- /dev/null
+++ b/crates/hotblocks/tests/peak_rss.rs
@@ -0,0 +1,105 @@
+//! Peak-RSS under head-path load (PR #98 checklist). Manual:
+//!
+//!   cargo test -p sqd-hotblocks --test peak_rss -- --ignored --nocapture
+//!
+//! `SQD_RSS_BIN=<path>` overrides the service binary (e.g. a master build for an
+//! old-vs-new comparison). Block production is paced so each response should carry one
+//! block and take the per-flush prepare path, but a binary slower than the pacing
+//! batches the accumulated tail into fewer, larger flushes — the report prints the
+//! achieved blocks/response so runs with different flush granularity are not compared
+//! silently. RSS is sampled from `ps` throughout.
+
+use std::{
+    sync::{
+        Arc,
+        atomic::{AtomicU64, Ordering}
+    },
+    time::Duration
+};
+
+use anyhow::Result;
+use sqd_hotblocks_harness::{
+    chain::Evm,
+    harness::{Harness, HarnessConfig}
+};
+
+const START: u64 = 1_000;
+
+fn blocks() -> u32 {
+    std::env::var("SQD_RSS_BLOCKS")
+        .ok()
+        .and_then(|v| v.parse().ok())
+        .unwrap_or(3_000)
+}
+
+#[tokio::test(flavor = "multi_thread")]
+#[ignore = "manual load test — run with --ignored --nocapture"]
+async fn peak_rss_under_head_load() -> Result<()> {
+    let bin = std::env::var("SQD_RSS_BIN").unwrap_or_else(|_| env!("CARGO_BIN_EXE_sqd-hotblocks").to_string());
+    println!("binary: {bin}");
+
+    let mut h = Harness::start(HarnessConfig::from_block(&bin, Arc::new(Evm), START)).await?;
+    let pid = h.sut.pid().expect("SUT pid");
+
+    let peak = Arc::new(AtomicU64::new(0));
+    let samples = Arc::new(std::sync::Mutex::new(Vec::<u64>::new()));
+    let sampler = tokio::spawn(sample_rss(pid, peak.clone(), samples.clone()));
+
+    let started = std::time::Instant::now();
+    let blocks = blocks();
+    for i in 0..blocks {
+        h.produce(1)?;
+        if i % 500 == 499 {
+            h.finalize_with_lag(5)?;
+            h.settle().await?;
+        }
+        tokio::time::sleep(Duration::from_millis(3)).await;
+    }
+    h.finalize_with_lag(5)?;
+    h.settle().await?;
+    h.assert_conforms().await?;
+    let elapsed = started.elapsed();
+
+    sampler.abort();
+    let samples = samples.lock().unwrap().clone();
+    let avg = |s: &[u64]| s.iter().sum::<u64>() / s.len().max(1) as u64;
+    let q = samples.len() / 4;
+    println!(
+        "blocks: {blocks} in {:.0}s ({:.0} blocks/s), samples: {}",
+        elapsed.as_secs_f64(),
+        blocks as f64 / elapsed.as_secs_f64(),
+        samples.len()
+    );
+    let stats = h.sim.stats(&h.dataset);
+    let data_responses = stats.stream_requests - stats.no_data - stats.fork_signals - stats.below_history;
+    println!(
+        "flush granularity: {} blocks over {data_responses} data responses ({:.2} blocks/response; 1.00 = one flush per block)",
+        stats.blocks_served,
+        stats.blocks_served as f64 / data_responses.max(1) as f64
+    );
+    println!(
+        "RSS MB: peak {:.0}, quartile avgs {:.0} → {:.0} → {:.0} → {:.0}",
+        peak.load(Ordering::Relaxed) as f64 / 1024.0,
+        avg(&samples[..q]) as f64 / 1024.0,
+        avg(&samples[q..2 * q]) as f64 / 1024.0,
+        avg(&samples[2 * q..3 * q]) as f64 / 1024.0,
+        avg(&samples[3 * q..]) as f64 / 1024.0
+    );
+    Ok(())
+}
+
+/// `ps -o rss=` is KiB on both macOS and Linux.
+async fn sample_rss(pid: u32, peak: Arc<AtomicU64>, samples: Arc<std::sync::Mutex<Vec<u64>>>) {
+    loop {
+        if let Ok(out) = tokio::process::Command::new("ps")
+            .args(["-o", "rss=", "-p", &pid.to_string()])
+            .output()
+            .await
+            && let Ok(kb) = String::from_utf8_lossy(&out.stdout).trim().parse::<u64>()
+        {
+            peak.fetch_max(kb, Ordering::Relaxed);
+            samples.lock().unwrap().push(kb);
+        }
+        tokio::time::sleep(Duration::from_millis(250)).await;
+    }
+}
diff --git a/docs/adr/0001-in-memory-chunk-prepare.md b/docs/adr/0001-in-memory-chunk-prepare.md
new file mode 100644
index 00000000..daa18cf4
--- /dev/null
+++ b/docs/adr/0001-in-memory-chunk-prepare.md
@@ -0,0 +1,63 @@
+# ADR 0001 — Prepare unspilled chunks in memory
+
+Status: proposed (PR #98) · Date: 2026-07-16 · Scope: `sqd-data-core`, `sqd-hotblocks`
+
+## Context
+
+Hotblocks ingest flushes a chunk on every upstream response, which at chain head means
+every block. Each flush built a full out-of-core `ChunkProcessor` — the spill machinery
+`crates/archive` uses to sort chunks of millions of rows — for what is usually a single
+block: several hundred temp files created, written, read back and torn down per flush
+(488 for the EVM schema, 229 for Solana; one per buffer per non-sort-key column).
+
+Measured on production-class hardware ([2026-07-16 flush bench]): a head flush costs
+~52 ms, ~90% of it kernel time, and the cost tracks the schema's file count, not the
+payload. A fleet of N datasets pays N × flush-rate × ~52 ms of CPU regardless of data
+volume — at 45 datasets roughly 10 cores of syscall and inode churn, growing linearly
+with the roster.
+
+The builder side already bounds nearly every unspilled chunk: `maybe_flush` moves
+builder contents into the processor once they exceed 30 MiB. The one exception is a
+flush triggered by the row-count bound (checked before the byte bound), whose final
+block can be arbitrarily large. At head every chunk fits whole.
+
+## Decision
+
+At `finish()`, if no processor exists and the builder is within the 30 MiB spill bound,
+prepare the chunk straight from the builder: sort and downcast in memory into a
+`TableReader::Mem`-backed `PreparedTable` (`PreparedTable::from_slice`,
+`ChunkBuilder::prepare_in_memory`). Chunks that did spill, or an unspilled chunk above
+the bound (the row-count-flush exception), keep the disk path bit-for-bit unchanged.
+The decision point is the existing spill bound — no new threshold, flag, or config.
+
+## Consequences
+
+- A head flush drops from 52.4 ms to 0.224 ms (−99.6%) on production hardware; no temp
+  files, no page-dirtying churn ([2026-07-16 flush bench]).
+- Differential tests pin observational equality with the spill path — schemas including
+  chunk-wide downcast, row content and order, partial reads; sorted/plain tables,
+  strings, lists, nulls, empty tables: `crates/data-core/tests/in_memory_prepare.rs`.
+- Chunk data now briefly occupies anon heap instead of page cache. Bounded by the
+  30 MiB spill bound × ≤3 in-flight chunks per dataset; a load probe at ~150× the
+  per-dataset head rate plateaus ~110 MB above the old path (jemalloc-retained churn,
+  no leak shape): `crates/hotblocks/tests/peak_rss.rs`.
+- Row order within equal full sort keys is unspecified and may differ from the spill
+  path. Ties exist in real schemas (EVM statediffs, Solana instructions) but are not
+  client-visible: queries re-sort output by a row-unique primary key. Today the orders
+  coincide byte-for-byte anyway (unstable-sort identity on already-sorted input, pinned
+  by test).
+- `PreparedTable::into_processor()` errors for in-memory tables — processor reuse
+  remains possible only on the disk path.
+
+## Alternatives considered
+
+- **Processor reuse** (restore the pre-`dbb896f` spare-cell): measured idealized ceiling
+  −78% at head, but the chunk-level reuse API was removed with optional-table support
+  (`0550396`), and naive reuse breaks on tables dropped after prepare. Kept as a
+  documented fallback, not chosen.
+- **Coalescing flushes at head**: proportional win, but trades freshness — hotblocks'
+  core value.
+- **Relocating TMPDIR / tmpfs**: a measurement tool, not a fix — syscall and allocation
+  overhead stays, and production temp storage is already NVMe.
+
+[2026-07-16 flush bench]: ../measurements/2026-07-16-flush-bench-e96c877.md
diff --git a/docs/measurements/2026-07-16-flush-bench-e96c877.md b/docs/measurements/2026-07-16-flush-bench-e96c877.md
new file mode 100644
index 00000000..c2dda951
--- /dev/null
+++ b/docs/measurements/2026-07-16-flush-bench-e96c877.md
@@ -0,0 +1,77 @@
+# 2026-07-16 — hotblocks flush cost decomposition (bench `e96c877`)
+
+Supporting measurements for [ADR 0001](../adr/0001-in-memory-chunk-prepare.md).
+
+**Environment:** one production-class node of the hotblocks fleet — Debian 12,
+Linux 6.1, NVMe-only storage, 128 cores; container `/tmp` on overlayfs. Bench:
+`crates/data/benches/flush_spill.rs` (`flush-bench` Dockerfile target), single-threaded,
+one flush at a time, synthetic EVM block with N rows. Modes are interleaved per
+iteration block; record-batch equality across all modes is asserted on every measured
+iteration. Numbers are p50 of per-iteration totals unless noted.
+
+Modes: **fresh** = the spill path building a new processor per flush (pre-fix
+production behavior); **reuse** = idealized processor recycling (fix #3 ceiling);
+**mem** = `prepare_in_memory` (ADR 0001).
+
+## `TMPDIR=/tmp` (overlay — production-equivalent)
+
+| rows | fresh p50 | fresh p95 | reuse p50 | saves | mem p50 | mem p95 | saves |
+|---|---|---|---|---|---|---|---|
+| 1 | 52.4 ms | 81.3 ms | 11.8 ms | 78% | **0.224 ms** | 0.314 ms | 100% |
+| 10 | 52.6 | 102.8 | 11.8 | 78% | 0.249 | 0.320 | 100% |
+| 100 | 53.4 | 65.5 | 11.9 | 78% | 0.453 | 0.574 | 99% |
+| 1 000 | 56.1 | 71.9 | 15.8 | 72% | 2.13 | 3.09 | 96% |
+| 10 000 | 116.7 | 372.7 | 59.5 | 49% | 22.1 | 31.1 | 81% |
+| 100 000 | 606.8 | 654.4 | 488.0 | 20% | 293.3 | 296.7 | 52% |
+
+## `TMPDIR=/dev/shm` (tmpfs — filesystem-sensitivity estimate)
+
+| rows | fresh p50 | reuse p50 | saves | mem p50 | saves |
+|---|---|---|---|---|---|
+| 1 | 7.9 ms | 4.3 ms | 46% | 0.192 ms | 98% |
+| 100 | 8.3 | 4.3 | 48% | 0.386 | 95% |
+| 10 000 | 38.8 | 33.9 | 13% | 21.0 | 46% |
+
+~86% of the head-flush cost is filesystem-implementation dependent (overlay vs tmpfs);
+what tmpfs keeps (syscalls, allocation, serialization) the mem path removes as well.
+
+## Stage split — head flush (rows = 1, `/tmp`, fresh mode)
+
+| stage | p50 | of which sys | notes |
+|---|---|---|---|
+| construct | 24.3 ms | 23.1 ms | 488 temp files (fd-delta measured) |
+| submit+finish | 6.7 ms | 6.2 ms | 1.46 MB page dirtying for a ~2.8 KB block (373 × 4 KiB) |
+| readback | 0.7 ms | 0.6 ms | page-cache hits, no physical reads |
+| drop | 20.2 ms (mean 26.5 — writeback tail) | 14.6 ms | close-only; unlinked inode teardown |
+
+Accounting caveat: `/proc/pid/io` credits `write_bytes` at page *dirtying* and
+`cancelled_write_bytes` at truncate-before-writeback, so these are dirtying figures,
+not proven disk traffic. In reuse mode the same 1.46 MB appears as `write_bytes` at
+submit and `cancelled_write_bytes` at reconstruct — dirtied, then cancelled by
+`set_len(0)`. Device-level attribution needs cgroup `io.stat` sampled concurrently.
+
+The fixed cost is flat in rows (construct marginal ≈ −0.01 ms per 1k rows) and
+dominates to ~9k rows; the fleet pays per file, not per byte. Cross-schema check on
+production metrics: the largest chain (Solana, 229 files) was the *cheapest* flush at
+147 ms while an idle EVM testnet (488 files) cost 232 ms — ~0.5–0.6 ms per file both.
+
+## Peak-RSS probe (macOS, release, jemalloc — `crates/hotblocks/tests/peak_rss.rs`)
+
+25 000 head blocks at ~177 blocks/s (≈150× the production per-dataset head rate)
+through the real binary, one block per flush, RSS sampled at 4 Hz:
+
+| binary | peak | quartile averages |
+|---|---|---|
+| mem path (fix) | 549 MB | 387 → 501 → 522 → 539 (plateau) |
+| spill path (old) | 435 MB | 200 → 309 → 401 → 425 |
+
+Steady-state delta ≈ 110 MB: chunk data briefly lives in jemalloc heap instead of page
+cache (which RSS never counted). No leak shape; black-box conformance asserted under
+load in every run.
+
+## Production symptom this explains (fleet, 2026-07-16)
+
+Pod at ~11.9 cores with **1.14 user / 10.65 sys** in ingest threads; `prepare` stage
+occupancy 12.1 at 59.2 flushes/s across 45 datasets (avg 204.6 ms/flush); ~51k
+read/write syscalls/s plus an estimated ~14–29k file creates/s (schema-weighted) that
+never appear in syscall counters; ~1.2k open deleted temp fds at any instant.