diff --git a/.github/workflows/compact-stress.yaml b/.github/workflows/compact-stress.yaml
new file mode 100644
index 0000000000..54133224d3
--- /dev/null
+++ b/.github/workflows/compact-stress.yaml
@@ -0,0 +1,46 @@
+name: Compaction crash stress
+on:
+  workflow_dispatch:
+    inputs:
+      leaves:
+        description: "Number of MMR leaves"
+        required: false
+        default: "15000"
+      mode:
+        description: "native | weak | qemu-user"
+        required: false
+        default: "native"
+  # Optional nightly — keep off by default (long under QEMU).
+  # schedule:
+  #   - cron: "0 4 * * 1"
+
+jobs:
+  compact-stress:
+    name: Compact kill stress (${{ github.event.inputs.mode || 'native' }})
+    runs-on: ubuntu-latest
+    timeout-minutes: 90
+    steps:
+      - uses: actions/checkout@v3
+      - name: Install qemu-user (qemu-user mode only)
+        if: ${{ github.event.inputs.mode == 'qemu-user' }}
+        run: sudo apt-get update && sudo apt-get install -y qemu-user
+      - name: Run stress harness
+        env:
+          LEAVES: ${{ github.event.inputs.leaves || '15000' }}
+        run: |
+          mode="${{ github.event.inputs.mode || 'native' }}"
+          case "$mode" in
+            native)
+              ./etc/qemu-compact-stress/run.sh --all-phases
+              ;;
+            qemu-user)
+              ./etc/qemu-compact-stress/run.sh --qemu-user --all-phases
+              ;;
+            weak)
+              ./etc/qemu-compact-stress/run.sh --weak --all-phases
+              ;;
+            *)
+              echo "unknown mode $mode" >&2
+              exit 2
+              ;;
+          esac
diff --git a/chain/src/txhashset/txhashset.rs b/chain/src/txhashset/txhashset.rs
index 4f8446b2ac..f50a67daa2 100644
--- a/chain/src/txhashset/txhashset.rs
+++ b/chain/src/txhashset/txhashset.rs
@@ -714,9 +714,29 @@ impl TxHashSet {
 		let total_outputs = outputs_pos.len();
 		let max_height = batch.head()?.height;
 
+		// outputs_pos is sorted by pos, so binary search for the first header
+		// that can contain the first missing output. This avoids walking the
+		// entire chain when only a few recent index entries are missing.
+		let first_pos = outputs_pos[0].1;
+		let mut start_height = 1;
+		let mut end_height = max_height;
+		while start_height < end_height {
+			let search_height = (start_height + end_height) / 2;
+			let hash = header_pmmr.get_header_hash_by_height(search_height)?;
+			let h = batch.get_block_header(&hash)?;
+			if h.output_mmr_size < first_pos {
+				start_height = search_height + 1;
+			} else {
+				end_height = search_height;
+			}
+		}
+
 		let mut i = 0;
-		for search_height in 0..max_height {
-			let hash = header_pmmr.get_header_hash_by_height(search_height + 1)?;
+		for search_height in start_height..=max_height {
+			if i >= total_outputs {
+				break;
+			}
+			let hash = header_pmmr.get_header_hash_by_height(search_height)?;
 			let h = batch.get_block_header(&hash)?;
 			while i < total_outputs {
 				let (commit, pos1) = outputs_pos[i];
diff --git a/core/src/core/pmmr/segment.rs b/core/src/core/pmmr/segment.rs
index f9f555bd40..5e654b94e1 100644
--- a/core/src/core/pmmr/segment.rs
+++ b/core/src/core/pmmr/segment.rs
@@ -19,6 +19,7 @@ use crate::core::pmmr::{self, Backend, ReadablePMMR, ReadonlyPMMR};
 use crate::ser::{Error, PMMRIndexHashable, PMMRable, Readable, Reader, Writeable, Writer};
 use croaring::Bitmap;
 use std::cmp::min;
+use std::collections::HashMap;
 use std::fmt::Debug;
 
 const MAX_SEGMENT_READ_ITEMS: u64 = 1_000_000;
@@ -102,6 +103,9 @@ pub enum SegmentError {
 	/// Mismatch between expected and actual root hash
 	#[error("Root hash mismatch")]
 	Mismatch,
+	/// A hash included in the segment could not be verified against the segment root
+	#[error("Unverifiable hash at pos {0}")]
+	UnverifiableHash(u64),
 }
 
 /// Tuple that defines a segment of a given PMMR
@@ -415,6 +419,19 @@ where
 		mmr_size: u64,
 		bitmap: Option<&Bitmap>,
 	) -> Result<Option<Hash>, SegmentError> {
+		self.root_with_known(mmr_size, bitmap).map(|(root, _)| root)
+	}
+
+	/// As `root`, but also returns the hash of every position that was bound
+	/// during the reconstruction: leaf hashes, computed parent hashes and the
+	/// segment hashes consumed at pruning boundaries. Used to verify that the
+	/// remaining hashes carried by the segment are consistent with the root.
+	fn root_with_known(
+		&self,
+		mmr_size: u64,
+		bitmap: Option<&Bitmap>,
+	) -> Result<(Option<Hash>, HashMap<u64, Hash>), SegmentError> {
+		let mut known = HashMap::new();
 		let (segment_first_pos, segment_last_pos) = self.segment_pos_range(mmr_size);
 		let mut hashes = Vec::<Option<Hash>>::with_capacity(2 * (self.identifier.height as usize));
 		let mut leaves0 = self.leaf_pos.iter().zip(&self.leaf_data);
@@ -462,10 +479,12 @@ where
 						(Some(l), Some(r)) => Some((l, r).hash_with_index(pos0)),
 						(None, Some(r)) => {
 							let l = self.get_hash(left_child_pos - 1)?;
+							known.insert(left_child_pos - 1, l);
 							Some((l, r).hash_with_index(pos0))
 						}
 						(Some(l), None) => {
 							let r = self.get_hash(right_child_pos - 1)?;
+							known.insert(right_child_pos - 1, r);
 							Some((l, r).hash_with_index(pos0))
 						}
 					}
@@ -481,12 +500,15 @@ where
 					)
 				}
 			};
+			if let Some(h) = hash {
+				known.insert(pos0, h);
+			}
 			hashes.push(hash);
 		}
 
 		if self.full_segment(mmr_size) {
 			// Full segment: last position of segment is subtree root
-			Ok(hashes.pop().unwrap())
+			Ok((hashes.pop().unwrap(), known))
 		} else {
 			// Not full (only final segment): peaks in segment, bag them together
 			let peaks = pmmr::peaks(mmr_size)
@@ -500,7 +522,9 @@ where
 					.ok_or_else(|| SegmentError::MissingHash(1 + pos0))?;
 				if lhash.is_none() && bitmap.is_some() {
 					// If this entire peak is pruned, load it from the segment hashes
-					lhash = Some(self.get_hash(pos0)?);
+					let h = self.get_hash(pos0)?;
+					known.insert(pos0, h);
+					lhash = Some(h);
 				}
 				let lhash = lhash.ok_or_else(|| SegmentError::MissingHash(1 + pos0))?;
 
@@ -509,7 +533,7 @@ where
 					Some(rhash) => Some((lhash, rhash).hash_with_index(mmr_size)),
 				};
 			}
-			Ok(Some(hash.unwrap()))
+			Ok((Some(hash.unwrap()), known))
 		}
 	}
 
@@ -519,10 +543,21 @@ where
 		mmr_size: u64,
 		bitmap: Option<&Bitmap>,
 	) -> Result<(Hash, u64), SegmentError> {
-		let root = self.root(mmr_size, bitmap)?;
+		self.first_unpruned_parent_with_known(mmr_size, bitmap)
+			.map(|(res, _)| res)
+	}
+
+	/// As `first_unpruned_parent`, but also returns the hashes bound during
+	/// root reconstruction (see `root_with_known`).
+	fn first_unpruned_parent_with_known(
+		&self,
+		mmr_size: u64,
+		bitmap: Option<&Bitmap>,
+	) -> Result<((Hash, u64), HashMap<u64, Hash>), SegmentError> {
+		let (root, mut known) = self.root_with_known(mmr_size, bitmap)?;
 		let (_, last) = self.segment_pos_range(mmr_size);
 		if let Some(root) = root {
-			return Ok((root, 1 + last));
+			return Ok(((root, 1 + last), known));
 		}
 		let bitmap = bitmap.unwrap();
 		let n_leaves = pmmr::n_leaves(mmr_size);
@@ -533,10 +568,11 @@ where
 		let mut family_branch = pmmr::family_branch(last, mmr_size).into_iter();
 		while cardinality == 0 {
 			hash = self.get_hash(pos0).map(|h| (h, 1 + pos0));
-			if hash.is_ok() {
+			if let Ok((h, pos1)) = hash {
 				// Return early in case a lower level hash is already present
 				// This can occur if both child trees are pruned but compaction hasn't run yet
-				return hash;
+				known.insert(pos1 - 1, h);
+				return Ok(((h, pos1), known));
 			}
 
 			if let Some((p0, _)) = family_branch.next() {
@@ -548,7 +584,61 @@ where
 				break;
 			}
 		}
-		hash
+		let (h, pos1) = hash?;
+		known.insert(pos1 - 1, h);
+		Ok(((h, pos1), known))
+	}
+
+	/// Verify that every hash carried by this segment is bound to the segment
+	/// root that the merkle proof validates. Root reconstruction only consumes
+	/// hashes at pruning boundaries: any additional hashes would otherwise be
+	/// applied to the local PMMR unchecked, allowing a peer to hand us a
+	/// segment that passes root validation but corrupts the reconstructed MMR,
+	/// which only surfaces (without attribution) at full state validation.
+	/// A hash is accepted if it was consumed during reconstruction, or if it
+	/// hashes into an already accepted parent together with its sibling.
+	fn verify_carried_hashes(&self, mut known: HashMap<u64, Hash>) -> Result<(), SegmentError> {
+		// Descending position order: in postorder a parent position is always
+		// greater than its children, so a bound parent can bind its children
+		// before they are visited.
+		for (&pos0, hash) in self.hash_pos.iter().zip(&self.hashes).rev() {
+			let bound = match known.get(&pos0) {
+				Some(h) => {
+					if h != hash {
+						return Err(SegmentError::UnverifiableHash(pos0));
+					}
+					*h
+				}
+				// Not bound (yet): if no parent binds it, we fail below.
+				None => continue,
+			};
+			let height = pmmr::bintree_postorder_height(pos0);
+			if height == 0 {
+				continue;
+			}
+			let left_pos0 = pos0 - (1 << height);
+			let right_pos0 = pos0 - 1;
+			let left = known
+				.get(&left_pos0)
+				.copied()
+				.or_else(|| self.get_hash(left_pos0).ok());
+			let right = known
+				.get(&right_pos0)
+				.copied()
+				.or_else(|| self.get_hash(right_pos0).ok());
+			if let (Some(l), Some(r)) = (left, right) {
+				if (l, r).hash_with_index(pos0) == bound {
+					known.insert(left_pos0, l);
+					known.insert(right_pos0, r);
+				}
+			}
+		}
+		for &pos0 in &self.hash_pos {
+			if !known.contains_key(&pos0) {
+				return Err(SegmentError::UnverifiableHash(pos0));
+			}
+		}
+		Ok(())
 	}
 
 	/// Check validity of the segment by calculating its root and validating the merkle proof
@@ -559,7 +649,8 @@ where
 		mmr_root: Hash,
 	) -> Result<(), SegmentError> {
 		let (first, last) = self.segment_pos_range(mmr_size);
-		let (segment_root, segment_unpruned_pos) = self.first_unpruned_parent(mmr_size, bitmap)?;
+		let ((segment_root, segment_unpruned_pos), known) =
+			self.first_unpruned_parent_with_known(mmr_size, bitmap)?;
 		self.proof.validate(
 			mmr_size,
 			mmr_root,
@@ -567,7 +658,8 @@ where
 			last,
 			segment_root,
 			segment_unpruned_pos,
-		)
+		)?;
+		self.verify_carried_hashes(known)
 	}
 
 	/// Check validity of the segment by calculating its root and validating the merkle proof
@@ -582,7 +674,8 @@ where
 		other_is_left: bool,
 	) -> Result<(), SegmentError> {
 		let (first, last) = self.segment_pos_range(mmr_size);
-		let (segment_root, segment_unpruned_pos) = self.first_unpruned_parent(mmr_size, bitmap)?;
+		let ((segment_root, segment_unpruned_pos), known) =
+			self.first_unpruned_parent_with_known(mmr_size, bitmap)?;
 		self.proof.validate_with(
 			mmr_size,
 			mmr_root,
@@ -593,7 +686,8 @@ where
 			hash_last_pos,
 			other_root,
 			other_is_left,
-		)
+		)?;
+		self.verify_carried_hashes(known)
 	}
 }
 
diff --git a/etc/qemu-compact-stress/README.md b/etc/qemu-compact-stress/README.md
new file mode 100644
index 0000000000..9a80426899
--- /dev/null
+++ b/etc/qemu-compact-stress/README.md
@@ -0,0 +1,99 @@
+# QEMU / weak-VPS / testnet-like compaction stress
+
+Reproduces slow compaction and hard-kill mid-compact (issue #3872).
+
+## Why testnet is worse than mainnet
+
+Cut-through horizon is the same on both chains (one week of blocks), but in
+practice **testnet is usually the easier and heavier repro**, even on a local PC:
+
+| Factor | Testnet | Long-running mainnet |
+|--------|---------|----------------------|
+| Spent / UTXO churn | High (faucets, testing, spam) | Lower relative churn |
+| Typical compact delta | Often large | Incremental small rewrites |
+| After PIBD / fresh node | First compact rewrites almost all hash+data against a huge prune set | Same worst case if you never compacted, but less often |
+| Desktoplocal timing | Easy to notice multi-second / multi-minute runs | Often mild if node has been compacting for months |
+
+So: **reproduce on testnet first**, then use this harness to isolate the PMMR
+rewrite + kill/recovery path without running a full node.
+
+## Real testnet repro (full node)
+
+```bash
+# Sync testnet (PIBD or existing chain_data), then force/wait for compact.
+# In another shell, when logs show compact starting:
+kill -9 $(pgrep -f 'grin.*testnet')   # hard kill mid-compact
+# Restart; with the journal fix the node should recover without a bad root.
+grin --testnet
+```
+
+Watch for `txhashset: starting compaction` / `check_compact` and for a long
+`Loading database` / high iowait during rewrite (same class of load as #3872).
+
+## Harness pieces
+
+| Path | Role |
+|------|------|
+| `store/examples/compact_crash_stress.rs` | prepare / compact / verify / kill-test |
+| `etc/qemu-compact-stress/run.sh` | native · testnet-like · qemu-user · Docker weak VPS |
+| `GRIN_COMPACT_*` env hooks in `store/src/pmmr.rs` | pause at a phase so parent can SIGKILL |
+
+Hooks are **no-ops** unless env vars are set.
+
+## Do I need to kill anything?
+
+| Command | Kill? |
+|---------|--------|
+| `./run.sh --testnet-like` | **No** — timing only. Wait for progress lines and final `OK:` |
+| `./run.sh --slow-only` | **No** |
+| `./run.sh` / `--all-phases` | **No** — harness auto-`SIGKILL`s a child when the phase is ready |
+| Real `grin --testnet` node | Optional: `kill -9` only if you are testing full-node recovery by hand |
+
+`--testnet-like` used to look “stuck” because prepare only printed at the end.
+It now prints progress every 10k leaves on **stderr**.
+
+## Quick start
+
+```bash
+# Testnet-like: dense prune set, time first vs second compact (no kill)
+./etc/qemu-compact-stress/run.sh --testnet-like
+
+# Kill recovery (automatic): dense prune + SIGKILL at journal
+LEAVES=50000 PRUNE_PCT=92 ./etc/qemu-compact-stress/run.sh
+
+# All kill windows
+./etc/qemu-compact-stress/run.sh --all-phases
+
+# Weak VPS: Docker with 0.5 CPU + 512MB on the *host* arch (not forced amd64)
+./etc/qemu-compact-stress/run.sh --weak --testnet-like
+
+# Direct example
+cargo run -p grin_store --example compact_crash_stress --release -- \
+  prepare ./target/tmp/tn 100000 92
+cargo run -p grin_store --example compact_crash_stress --release -- \
+  compact ./target/tmp/tn
+```
+
+`compact` prints **first** and **second** pass times:
+
+- **first** ≈ testnet / first compact after a dense prune set  
+- **second** ≈ already-compacted rewrite  
+
+Expect first to dominate on large/dense datasets (why testnet “feels heavier”).
+
+## Kill phases
+
+| Phase | When | Expected recovery |
+|-------|------|-------------------|
+| `hash_tmp` | After hash staging | Discard orphans; keep pre-compact live files |
+| `data_tmp` | After data staging | Same |
+| `journal` | After journal fsync, before renames | Roll renames forward; post-compact state |
+
+## Manual env
+
+```bash
+export GRIN_COMPACT_READY_FILE=/tmp/compact_ready
+export GRIN_COMPACT_PAUSE_PHASE=journal
+export GRIN_COMPACT_CRASH_PAUSE_MS=60000
+# run compact in one terminal; kill -9 when ready file says "journal"
+```
diff --git a/etc/qemu-compact-stress/run.sh b/etc/qemu-compact-stress/run.sh
new file mode 100755
index 0000000000..eac2fa8b08
--- /dev/null
+++ b/etc/qemu-compact-stress/run.sh
@@ -0,0 +1,224 @@
+#!/usr/bin/env bash
+# Simulate a weak VPS (slow CPU / tight RAM) and hard-kill mid PMMR compaction.
+#
+# Testnet is usually the best real-world repro (often heavier than mainnet even
+# on a desktop): high spend churn + first compact after PIBD rewrites almost
+# the entire hash/data files. Use --testnet-like for that profile.
+#
+# Modes:
+#   ./run.sh                  # native: auto kill-test at journal (you do NOT kill)
+#   ./run.sh --testnet-like   # timing only (no kill): large leaves + ~92% prune
+#   ./run.sh --weak           # docker cgroup limits (0.5 CPU, 512MB), host arch
+#   ./run.sh --qemu-user      # linux: run under qemu-x86_64 (TCG)
+#   ./run.sh --all-phases     # kill-test at hash_tmp, data_tmp, journal
+#   ./run.sh --slow-only      # prepare + compact only (no kill)
+#
+# You never need to Ctrl-C / kill by hand for the harness — kill-test sends
+# SIGKILL to a child automatically. Only a real grin --testnet node would use
+# manual kill -9 while logs show compaction.
+#
+# Env overrides:
+#   LEAVES=100000      number of leaves
+#   PRUNE_PCT=90       percent pruned (testnet-like: 90-95)
+#   PHASE=journal      kill phase
+#   WORK_DIR=...
+#   COMPACT_STRESS_IMAGE=rust:1.83-bookworm
+#
+# Exit 0 on success.
+set -euo pipefail
+
+ROOT="$(cd "$(dirname "$0")/../.." && pwd)"
+cd "$ROOT"
+
+MODE="native"
+ALL_PHASES=0
+SLOW_ONLY=0
+TESTNET_LIKE=0
+PHASE="${PHASE:-journal}"
+WORK_DIR="${WORK_DIR:-$ROOT/target/tmp/qemu-compact-stress}"
+
+while [[ $# -gt 0 ]]; do
+	case "$1" in
+		--weak) MODE="weak"; shift ;;
+		--qemu-user) MODE="qemu-user"; shift ;;
+		--all-phases) ALL_PHASES=1; shift ;;
+		--slow-only) SLOW_ONLY=1; shift ;;
+		--testnet-like) TESTNET_LIKE=1; SLOW_ONLY=1; shift ;;
+		--phase) PHASE="$2"; shift 2 ;;
+		--leaves) LEAVES="$2"; shift 2 ;;
+		--prune-pct) PRUNE_PCT="$2"; shift 2 ;;
+		-h|--help)
+			sed -n '2,32p' "$0"
+			exit 0
+			;;
+		*)
+			echo "unknown arg: $1" >&2
+			exit 2
+			;;
+	esac
+done
+
+if [[ "$TESTNET_LIKE" -eq 1 ]]; then
+	# First compact after a dense prune set (post-PIBD / high-churn testnet).
+	# Smaller default under --weak so Docker builds finish in reasonable time.
+	if [[ "$MODE" == "weak" ]]; then
+		LEAVES="${LEAVES:-20000}"
+	else
+		LEAVES="${LEAVES:-100000}"
+	fi
+	PRUNE_PCT="${PRUNE_PCT:-92}"
+elif [[ -z "${LEAVES:-}" ]]; then
+	if [[ "$MODE" == "weak" ]]; then
+		LEAVES=8000
+	else
+		LEAVES=20000
+	fi
+fi
+PRUNE_PCT="${PRUNE_PCT:-50}"
+
+echo "==> compact crash stress"
+echo "    leaves=${LEAVES} prune_pct=${PRUNE_PCT} mode=${MODE} phase=${PHASE}"
+if [[ "$SLOW_ONLY" -eq 1 ]]; then
+	echo "    mode=timing only — do NOT kill; wait for prepare/compact progress on stderr"
+else
+	echo "    mode=kill-test — harness auto-SIGKILLs a child at phase=${PHASE} (no manual kill)"
+fi
+
+echo "==> building compact_crash_stress (release) on host"
+cargo build -p grin_store --example compact_crash_stress --release
+BIN="$ROOT/target/release/examples/compact_crash_stress"
+test -x "$BIN"
+
+run_bin() {
+	if [[ "$MODE" == "qemu-user" ]]; then
+		local qemubin=""
+		if command -v qemu-x86_64 >/dev/null 2>&1; then
+			qemubin="qemu-x86_64"
+		elif command -v qemu-x86_64-static >/dev/null 2>&1; then
+			qemubin="qemu-x86_64-static"
+		else
+			echo "qemu-x86_64 not found (install qemu-user)" >&2
+			exit 1
+		fi
+		"$qemubin" -cpu qemu64 "$BIN" "$@"
+	else
+		"$BIN" "$@"
+	fi
+}
+
+run_kill_phase() {
+	local phase="$1"
+	local dir="${WORK_DIR}/${phase}"
+	rm -rf "$dir"
+	mkdir -p "$dir"
+	echo ""
+	echo "========== kill-test phase=${phase} leaves=${LEAVES} prune_pct=${PRUNE_PCT} =========="
+	echo "    (parent waits for phase, then SIGKILLs child — no action needed from you)"
+	run_bin kill-test "$dir" "$LEAVES" "$phase" "$PRUNE_PCT"
+}
+
+run_slow_only() {
+	local dir="${WORK_DIR}/slow"
+	rm -rf "$dir"
+	mkdir -p "$dir"
+	echo ""
+	echo "========== slow compact leaves=${LEAVES} prune_pct=${PRUNE_PCT} =========="
+	echo "    first compact ≈ testnet post-PIBD; second ≈ already-compacted delta"
+	echo "    No kill: leave this running until it prints 'OK: compact crash stress finished'"
+	run_bin prepare "$dir" "$LEAVES" "$PRUNE_PCT"
+	run_bin compact "$dir"
+	run_bin verify "$dir"
+}
+
+run_native_or_qemu() {
+	if [[ "$SLOW_ONLY" -eq 1 ]]; then
+		run_slow_only
+		return
+	fi
+	if [[ "$ALL_PHASES" -eq 1 ]]; then
+		for p in hash_tmp data_tmp journal; do
+			run_kill_phase "$p"
+		done
+	else
+		run_kill_phase "$PHASE"
+	fi
+}
+
+run_weak_docker() {
+	if ! command -v docker >/dev/null 2>&1; then
+		echo "docker not found; cannot run --weak mode" >&2
+		exit 1
+	fi
+
+	# Use the host's native platform (linux/arm64 on Apple Silicon, linux/amd64
+	# on typical CI). Forcing linux/amd64 on arm runs the whole toolchain under
+	# QEMU TCG and often breaks PATH / takes hours. Cgroup limits still model a
+	# weak VPS; for extra CPU drag use --qemu-user on an x86_64 Linux host.
+	local image="${COMPACT_STRESS_IMAGE:-rust:1.83-bookworm}"
+	local platform
+	platform="$(docker version -f '{{.Server.Os}}/{{.Server.Arch}}' 2>/dev/null || echo linux/amd64)"
+
+	echo "==> weak VPS via docker"
+	echo "    platform=${platform} (native)  cpus=0.5  memory=512m  image=${image}"
+	echo "    building inside container (first run downloads the image + deps)"
+
+	# Explicit cargo PATH: login shells under some images do not load rustup env.
+	docker run --rm \
+		--platform "$platform" \
+		--cpus="0.5" \
+		--memory="512m" \
+		--memory-swap="512m" \
+		-v "$ROOT:/src:rw" \
+		-w /src \
+		-e CARGO_TARGET_DIR=/src/target/qemu-compact-docker \
+		-e PATH="/usr/local/cargo/bin:/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin" \
+		"$image" \
+		bash -c "
+			set -euo pipefail
+			export PATH=\"/usr/local/cargo/bin:\$PATH\"
+			if [[ -f /usr/local/cargo/env ]]; then
+				# shellcheck disable=SC1091
+				source /usr/local/cargo/env
+			fi
+			if ! command -v cargo >/dev/null 2>&1; then
+				echo \"cargo not found in image PATH=\$PATH\" >&2
+				ls -la /usr/local/cargo/bin 2>/dev/null || true
+				exit 1
+			fi
+			echo \"==> docker: cargo=\$(command -v cargo) rustc=\$(rustc --version)\"
+			# Keep apt light — only if clang headers missing for bindgen crates.
+			if ! dpkg -s libclang-dev >/dev/null 2>&1; then
+				apt-get update -qq
+				DEBIAN_FRONTEND=noninteractive apt-get install -y -qq build-essential pkg-config libclang-dev
+			fi
+			echo '==> docker: cargo build (release example) — may take a while under cgroup limits'
+			cargo build -p grin_store --example compact_crash_stress --release
+			BIN=/src/target/qemu-compact-docker/release/examples/compact_crash_stress
+			WORK=/src/target/tmp/qemu-compact-stress-docker
+			rm -rf \"\$WORK\"
+			mkdir -p \"\$WORK\"
+			if [[ '${SLOW_ONLY}' -eq 1 ]]; then
+				echo '==> docker: prepare+compact (no kill)'
+				\"\$BIN\" prepare \"\$WORK/slow\" ${LEAVES} ${PRUNE_PCT}
+				\"\$BIN\" compact \"\$WORK/slow\"
+				\"\$BIN\" verify \"\$WORK/slow\"
+			elif [[ '${ALL_PHASES}' -eq 1 ]]; then
+				for p in hash_tmp data_tmp journal; do
+					echo \"========== docker kill-test phase=\$p ==========\"
+					\"\$BIN\" kill-test \"\$WORK/\$p\" ${LEAVES} \"\$p\" ${PRUNE_PCT}
+				done
+			else
+				echo \"==> docker: kill-test phase=${PHASE} (auto SIGKILL)\"
+				\"\$BIN\" kill-test \"\$WORK/${PHASE}\" ${LEAVES} ${PHASE} ${PRUNE_PCT}
+			fi
+		"
+}
+
+case "$MODE" in
+	native|qemu-user) run_native_or_qemu ;;
+	weak) run_weak_docker ;;
+	*) echo "bad mode $MODE" >&2; exit 2 ;;
+esac
+
+echo ""
+echo "OK: compact crash stress finished (mode=${MODE} leaves=${LEAVES} prune_pct=${PRUNE_PCT})"
diff --git a/store/Cargo.toml b/store/Cargo.toml
index bb594abe34..c93f5b21fc 100644
--- a/store/Cargo.toml
+++ b/store/Cargo.toml
@@ -28,3 +28,7 @@ chrono = "0.4.11"
 rand = "0.6"
 filetime = "0.2"
 env_logger = "0.7"
+
+[[example]]
+name = "compact_crash_stress"
+path = "examples/compact_crash_stress.rs"
diff --git a/store/examples/compact_crash_stress.rs b/store/examples/compact_crash_stress.rs
new file mode 100644
index 0000000000..4b9811235d
--- /dev/null
+++ b/store/examples/compact_crash_stress.rs
@@ -0,0 +1,417 @@
+// Copyright 2021 The Grin Developers
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Stress harness for PMMR compaction on weak hardware and crash recovery.
+//!
+//! Used by `etc/qemu-compact-stress/run.sh` to simulate a low-end VPS
+//! (slow CPU via QEMU, tight RAM) and hard-kill mid-compaction.
+//!
+//! **Testnet is usually the easiest real-world repro** — often heavier than
+//! mainnet even on a desktop: more spent/UTXO churn, and the first compact
+//! after PIBD rewrites almost the entire hash/data files against a huge prune
+//! set. Long-running mainnet nodes compact incrementally (smaller deltas).
+//! Use a high `prune_pct` (runner `--testnet-like`) to model that case.
+//!
+//! Subcommands:
+//! - `prepare <dir> <leaves> [prune_pct]` — build PMMR, prune pct% of leaves
+//! - `compact <dir>` — run `check_compact`, print duration (and a 2nd pass)
+//! - `verify <dir>` — open and check root matches `root.txt`
+//! - `kill-test <dir> <leaves> <phase> [prune_pct]` — SIGKILL mid-compact
+//!
+//! Env (also used by check_compact test hooks):
+//! - `GRIN_COMPACT_READY_FILE`, `GRIN_COMPACT_PAUSE_PHASE`,
+//!   `GRIN_COMPACT_CRASH_PAUSE_MS`, `GRIN_COMPACT_PRUNE_PCT`
+
+use croaring::Bitmap;
+use grin_core::core::hash::{DefaultHashable, Hash};
+use grin_core::core::pmmr::{ReadablePMMR, PMMR};
+use grin_core::ser::{Error, PMMRable, ProtocolVersion, Readable, Reader, Writeable, Writer};
+use grin_store::pmmr::PMMRBackend;
+use grin_util::ToHex;
+use std::env;
+use std::fs::{self, File};
+use std::io::Read;
+use std::path::{Path, PathBuf};
+use std::process::{Command, Stdio};
+use std::thread;
+use std::time::{Duration, Instant};
+
+#[derive(Copy, Clone, Debug, PartialEq, Eq)]
+struct TestElem(u32);
+
+impl DefaultHashable for TestElem {}
+
+impl PMMRable for TestElem {
+	type E = Self;
+
+	fn as_elmt(&self) -> Self::E {
+		*self
+	}
+
+	fn elmt_size() -> Option<u16> {
+		Some(4)
+	}
+}
+
+impl Writeable for TestElem {
+	fn write<W: Writer>(&self, writer: &mut W) -> Result<(), Error> {
+		writer.write_u32(self.0)
+	}
+}
+
+impl Readable for TestElem {
+	fn read<R: Reader>(reader: &mut R) -> Result<TestElem, Error> {
+		Ok(TestElem(reader.read_u32()?))
+	}
+}
+
+fn usage() -> ! {
+	eprintln!(
+		"usage:
+  compact_crash_stress prepare <data_dir> <n_leaves> [prune_pct]
+  compact_crash_stress compact <data_dir>
+  compact_crash_stress verify  <data_dir>
+  compact_crash_stress kill-test <data_dir> <n_leaves> <phase> [prune_pct]
+
+prune_pct: 0-99 percent of leaves to prune before compact (default 50).
+  Testnet-like first compact after heavy spend/PIBD: try 90-95.
+
+phases for kill-test: hash_tmp | data_tmp | journal
+"
+	);
+	std::process::exit(2);
+}
+
+fn parse_prune_pct(arg: Option<String>) -> u32 {
+	let pct = arg
+		.or_else(|| env::var("GRIN_COMPACT_PRUNE_PCT").ok())
+		.map(|s| s.parse::<u32>().expect("prune_pct u32"))
+		.unwrap_or(50);
+	assert!(
+		pct < 100,
+		"prune_pct must be 0..99 (keep at least some UTXOs)"
+	);
+	pct
+}
+
+fn root_path(dir: &Path) -> PathBuf {
+	dir.join("root.txt")
+}
+fn size_path(dir: &Path) -> PathBuf {
+	dir.join("mmr_size.txt")
+}
+
+fn write_meta(dir: &Path, root: Hash, mmr_size: u64) {
+	// Full 32-byte hex (Display truncates for logging).
+	fs::write(root_path(dir), root.as_bytes().to_hex()).unwrap();
+	fs::write(size_path(dir), format!("{}", mmr_size)).unwrap();
+}
+
+fn read_meta(dir: &Path) -> (Hash, u64) {
+	let mut root_s = String::new();
+	File::open(root_path(dir))
+		.unwrap()
+		.read_to_string(&mut root_s)
+		.unwrap();
+	let root = Hash::from_hex(root_s.trim()).expect("valid root hex");
+	let size: u64 = fs::read_to_string(size_path(dir))
+		.unwrap()
+		.trim()
+		.parse()
+		.unwrap();
+	(root, size)
+}
+
+fn open_backend(dir: &Path) -> PMMRBackend<TestElem> {
+	PMMRBackend::new(dir, true, ProtocolVersion(1), None).expect("open backend")
+}
+
+/// Build a prunable PMMR with `n_leaves`, prune `prune_pct`% of leaves
+/// (stable root), sync, and persist root/size metadata.
+///
+/// High prune_pct (90–95) models testnet / post-PIBD first compact: a large
+/// prune set and a full hash+data file rewrite, which is much heavier than
+/// the small deltas a continuously compacting mainnet node typically sees.
+fn progress(label: &str, done: u32, total: u32, t0: Instant) {
+	if done == 0 || done == total || done % 10_000 == 0 {
+		eprintln!(
+			"  ... {}: {}/{} ({:.0}%) elapsed={:.1}s",
+			label,
+			done,
+			total,
+			100.0 * f64::from(done) / f64::from(total.max(1)),
+			t0.elapsed().as_secs_f64()
+		);
+	}
+}
+
+fn cmd_prepare(dir: &Path, n_leaves: u32, prune_pct: u32) {
+	if dir.exists() {
+		fs::remove_dir_all(dir).unwrap();
+	}
+	fs::create_dir_all(dir).unwrap();
+
+	let t0 = Instant::now();
+	eprintln!(
+		"prepare: building {} leaves (prune_pct={}) under {} — do not kill, wait for 'prepare: leaves=...'",
+		n_leaves,
+		prune_pct,
+		dir.display()
+	);
+
+	let mut backend = open_backend(dir);
+	let mut mmr = PMMR::new(&mut backend);
+	for i in 0..n_leaves {
+		mmr.push(&TestElem(i)).unwrap();
+		progress("push", i + 1, n_leaves, t0);
+	}
+	let mmr_size = mmr.unpruned_size();
+	let root = mmr.root().unwrap();
+	drop(mmr);
+	backend.sync().unwrap();
+	eprintln!(
+		"prepare: push done in {:.1}s, pruning...",
+		t0.elapsed().as_secs_f64()
+	);
+
+	// Keep every Nth leaf so (prune_pct)% are removed. N = 100 / (100 - pct).
+	// e.g. prune_pct=50 → keep every 2nd; prune_pct=90 → keep every 10th.
+	let keep_every = (100u32 / (100 - prune_pct)).max(1);
+	let mut pruned = 0u32;
+	{
+		let mut mmr = PMMR::at(&mut backend, mmr_size);
+		for i in 0..n_leaves {
+			if i % keep_every != 0 {
+				let pos0 = grin_core::core::pmmr::insertion_to_pmmr_index(i as u64);
+				if mmr.prune(pos0).unwrap_or(false) {
+					pruned += 1;
+				}
+			}
+			progress("prune", i + 1, n_leaves, t0);
+		}
+	}
+	backend.sync().unwrap();
+	write_meta(dir, root, mmr_size);
+
+	let hash_bytes = fs::metadata(dir.join("pmmr_hash.bin"))
+		.map(|m| m.len())
+		.unwrap_or(0);
+	let data_bytes = fs::metadata(dir.join("pmmr_data.bin"))
+		.map(|m| m.len())
+		.unwrap_or(0);
+	println!(
+		"prepare: leaves={} pruned={} (~{}%, keep_every={}) mmr_size={} hash_bytes={} data_bytes={} root={} took={:.2}s",
+		n_leaves,
+		pruned,
+		prune_pct,
+		keep_every,
+		mmr_size,
+		hash_bytes,
+		data_bytes,
+		root,
+		t0.elapsed().as_secs_f64()
+	);
+}
+
+fn cmd_compact(dir: &Path) {
+	let (expected_root, mmr_size) = read_meta(dir);
+	let mut backend = open_backend(dir);
+
+	// First compact: full rewrite against the accumulated prune set
+	// (testnet / post-PIBD worst case).
+	eprintln!(
+		"compact: first pass (dense prune rewrite) mmr_size={} — no kill needed in --slow-only/--testnet-like",
+		mmr_size
+	);
+	let t0 = Instant::now();
+	backend
+		.check_compact(mmr_size, &Bitmap::new())
+		.expect("check_compact");
+	backend.sync().unwrap();
+	let first = t0.elapsed();
+	eprintln!("compact: first pass done in {:.3}s", first.as_secs_f64());
+
+	// Second compact with no new prunes: should be much cheaper — closer to
+	// a long-running mainnet node that only rewrites a small delta.
+	eprintln!("compact: second pass (incremental / already compacted)...");
+	let t1 = Instant::now();
+	backend
+		.check_compact(mmr_size, &Bitmap::new())
+		.expect("check_compact second");
+	backend.sync().unwrap();
+	let second = t1.elapsed();
+
+	let pmmr = PMMR::at(&mut backend, mmr_size);
+	let root = pmmr.root().unwrap();
+	assert_eq!(root, expected_root, "root must be unchanged by compact");
+	println!(
+		"compact: ok root={} first={:.3}s ({:.0} ms) second={:.3}s ({:.0} ms) ratio={:.1}x",
+		root,
+		first.as_secs_f64(),
+		first.as_secs_f64() * 1000.0,
+		second.as_secs_f64(),
+		second.as_secs_f64() * 1000.0,
+		if second.as_secs_f64() > 0.0 {
+			first.as_secs_f64() / second.as_secs_f64()
+		} else {
+			0.0
+		}
+	);
+}
+
+fn cmd_verify(dir: &Path) {
+	let (expected_root, mmr_size) = read_meta(dir);
+	let mut backend = open_backend(dir);
+	let pmmr = PMMR::at(&mut backend, mmr_size);
+	let root = pmmr.root().unwrap();
+	assert_eq!(
+		root, expected_root,
+		"root mismatch after open/recovery: got {} want {}",
+		root, expected_root
+	);
+	// Spot-check: leaf insertion index 0 is always kept by prepare (keep_every).
+	assert!(
+		pmmr.n_unpruned_leaves() > 0,
+		"expected unpruned leaves after compact/recovery"
+	);
+	assert!(
+		pmmr.get_data(0).is_some(),
+		"expected first leaf (insertion 0) still readable"
+	);
+	assert!(
+		!dir.join("pmmr_compact.journal").exists(),
+		"journal must not remain after successful open"
+	);
+	println!("verify: ok root={}", root);
+}
+
+fn wait_for_phase(ready_file: &Path, phase: &str, timeout: Duration) -> bool {
+	let start = Instant::now();
+	while start.elapsed() < timeout {
+		if let Ok(contents) = fs::read_to_string(ready_file) {
+			if contents.trim() == phase {
+				return true;
+			}
+		}
+		thread::sleep(Duration::from_millis(20));
+	}
+	false
+}
+
+/// Prepare dataset, spawn `compact` child paused at `phase`, SIGKILL it, verify.
+fn cmd_kill_test(dir: &Path, n_leaves: u32, phase: &str, prune_pct: u32) {
+	match phase {
+		"hash_tmp" | "data_tmp" | "journal" => {}
+		_ => {
+			eprintln!("unknown phase '{}'", phase);
+			usage();
+		}
+	}
+
+	cmd_prepare(dir, n_leaves, prune_pct);
+	let (expected_root, _) = read_meta(dir);
+
+	let ready_file = dir.join("compact_ready");
+	let _ = fs::remove_file(&ready_file);
+
+	let exe = env::current_exe().expect("current_exe");
+	let mut child = Command::new(&exe)
+		.arg("compact")
+		.arg(dir)
+		.env("GRIN_COMPACT_READY_FILE", &ready_file)
+		.env("GRIN_COMPACT_PAUSE_PHASE", phase)
+		.env("GRIN_COMPACT_CRASH_PAUSE_MS", "30000")
+		.stdout(Stdio::inherit())
+		.stderr(Stdio::inherit())
+		.spawn()
+		.expect("spawn compact child");
+
+	println!(
+		"kill-test: child pid={} waiting for phase '{}' ...",
+		child.id(),
+		phase
+	);
+	let saw = wait_for_phase(&ready_file, phase, Duration::from_secs(600));
+	if !saw {
+		let _ = child.kill();
+		let _ = child.wait();
+		panic!(
+			"timed out waiting for compact phase '{}' (ready file: {:?})",
+			phase, ready_file
+		);
+	}
+
+	// Hard kill: no graceful shutdown (simulates OOM killer / kill -9 / power loss).
+	#[cfg(unix)]
+	{
+		let pid = child.id() as i32;
+		let rc = unsafe { libc::kill(pid, libc::SIGKILL) };
+		assert_eq!(rc, 0, "SIGKILL failed");
+	}
+	#[cfg(not(unix))]
+	{
+		child.kill().expect("kill child");
+	}
+	let status = child.wait().expect("wait child");
+	println!(
+		"kill-test: child exited {:?} after SIGKILL at '{}'",
+		status, phase
+	);
+
+	// Parent re-opens: recovery must leave a consistent store with the same root.
+	cmd_verify(dir);
+	let (root, _) = read_meta(dir);
+	assert_eq!(root, expected_root);
+	println!(
+		"kill-test: PASS phase={} root recovered={}",
+		phase, expected_root
+	);
+}
+
+fn main() {
+	let mut args = env::args().skip(1);
+	let cmd = args.next().unwrap_or_else(|| usage());
+	match cmd.as_str() {
+		"prepare" => {
+			let dir = PathBuf::from(args.next().unwrap_or_else(|| usage()));
+			let n: u32 = args
+				.next()
+				.unwrap_or_else(|| usage())
+				.parse()
+				.expect("n_leaves");
+			let prune_pct = parse_prune_pct(args.next());
+			cmd_prepare(&dir, n, prune_pct);
+		}
+		"compact" => {
+			let dir = PathBuf::from(args.next().unwrap_or_else(|| usage()));
+			cmd_compact(&dir);
+		}
+		"verify" => {
+			let dir = PathBuf::from(args.next().unwrap_or_else(|| usage()));
+			cmd_verify(&dir);
+		}
+		"kill-test" => {
+			let dir = PathBuf::from(args.next().unwrap_or_else(|| usage()));
+			let n: u32 = args
+				.next()
+				.unwrap_or_else(|| usage())
+				.parse()
+				.expect("n_leaves");
+			let phase = args.next().unwrap_or_else(|| usage());
+			let prune_pct = parse_prune_pct(args.next());
+			cmd_kill_test(&dir, n, &phase, prune_pct);
+		}
+		_ => usage(),
+	}
+}
diff --git a/store/src/leaf_set.rs b/store/src/leaf_set.rs
index d0cc25531f..f2e1486def 100644
--- a/store/src/leaf_set.rs
+++ b/store/src/leaf_set.rs
@@ -181,6 +181,23 @@ impl LeafSet {
 		Ok(())
 	}
 
+	/// Write the current bitmap to an arbitrary path (used for journaled compact).
+	/// Does not update the in-memory backup; call `mark_flushed` after the write
+	/// is committed via rename.
+	pub fn write_to<P: AsRef<Path>>(&self, path: P) -> io::Result<()> {
+		let mut bitmap = self.bitmap.clone();
+		bitmap.run_optimize();
+		let mut file = File::create(path.as_ref())?;
+		file.write_all(&bitmap.serialize::<Portable>())?;
+		file.sync_all()?;
+		Ok(())
+	}
+
+	/// Mark the in-memory bitmap as matching what is on disk.
+	pub fn mark_flushed(&mut self) {
+		self.bitmap_bak = self.bitmap.clone();
+	}
+
 	/// Discard any pending changes.
 	pub fn discard(&mut self) {
 		self.bitmap = self.bitmap_bak.clone();
diff --git a/store/src/pmmr.rs b/store/src/pmmr.rs
index 53c6871e13..d8f48a2d89 100644
--- a/store/src/pmmr.rs
+++ b/store/src/pmmr.rs
@@ -13,8 +13,10 @@
 
 //! Implementation of the persistent Backend for the prunable MMR tree.
 
-use std::fs;
-use std::{io, time};
+use std::fs::{self, File};
+use std::io::{self, Write};
+use std::thread;
+use std::time::{self, Duration};
 
 use crate::grin_core::core::hash::{Hash, Hashed};
 use crate::grin_core::core::pmmr::{self, family, Backend};
@@ -27,11 +29,50 @@ use croaring::Bitmap;
 use std::convert::TryInto;
 use std::path::{Path, PathBuf};
 
+/// Test-only hooks for simulating a hard kill mid-compaction (weak-VPS /
+/// qemu stress harness). Production runs leave these env vars unset so this
+/// is a pure no-op.
+///
+/// - `GRIN_COMPACT_READY_FILE`: path written with the current phase name when
+///   that phase is reached (`hash_tmp`, `data_tmp`, `journal`, `applied`).
+/// - `GRIN_COMPACT_PAUSE_PHASE`: if set to a phase name, sleep while at that
+///   phase so an external watchdog can SIGKILL the process.
+/// - `GRIN_COMPACT_CRASH_PAUSE_MS`: sleep duration in ms (default 5000 when
+///   pause phase is set).
+fn compact_test_hook(phase: &str) {
+	if let Ok(path) = std::env::var("GRIN_COMPACT_READY_FILE") {
+		if !path.is_empty() {
+			let _ = fs::write(&path, phase.as_bytes());
+		}
+	}
+	let pause_phase = match std::env::var("GRIN_COMPACT_PAUSE_PHASE") {
+		Ok(p) if p == phase => p,
+		_ => return,
+	};
+	let ms = std::env::var("GRIN_COMPACT_CRASH_PAUSE_MS")
+		.ok()
+		.and_then(|s| s.parse::<u64>().ok())
+		.unwrap_or(5_000);
+	if ms > 0 {
+		debug!(
+			"compact: test hook pausing {}ms at phase '{}' (env)",
+			ms, pause_phase
+		);
+		thread::sleep(Duration::from_millis(ms));
+	}
+}
+
 const PMMR_HASH_FILE: &str = "pmmr_hash.bin";
 const PMMR_DATA_FILE: &str = "pmmr_data.bin";
 const PMMR_LEAF_FILE: &str = "pmmr_leaf.bin";
 const PMMR_PRUN_FILE: &str = "pmmr_prun.bin";
 const PMMR_SIZE_FILE: &str = "pmmr_size.bin";
+/// Marker written after all compact temps are durable and before any live
+/// file is replaced. Presence means an interrupted compact must roll forward.
+const PMMR_COMPACT_JOURNAL: &str = "pmmr_compact.journal";
+/// Suffix appended to leaf/prune paths for pre-commit compact temps
+/// (`pmmr_prun.bin.compact`). Distinct from `.tmp` used by single-file flushes.
+const PMMR_COMPACT_SUFFIX: &str = ".compact";
 const REWIND_FILE_CLEANUP_DURATION_SECONDS: u64 = 60 * 60 * 24; // 24 hours as seconds
 
 /// The list of PMMR_Files for internal purposes
@@ -42,6 +83,105 @@ pub const PMMR_FILES: [&str; 4] = [
 	PMMR_PRUN_FILE,
 ];
 
+/// Path for a leaf/prune compact staging file: `foo.bin` → `foo.bin.compact`.
+fn compact_staging_path(path: &Path) -> PathBuf {
+	let mut os = path.as_os_str().to_os_string();
+	os.push(PMMR_COMPACT_SUFFIX);
+	PathBuf::from(os)
+}
+
+/// Rename `from` over `to` if `from` exists. No-op if the temp is absent
+/// (already applied during a previous partial recovery).
+fn replace_if_present(from: &Path, to: &Path) -> io::Result<()> {
+	if !from.exists() {
+		return Ok(());
+	}
+	// Windows cannot rename over an existing file; Unix replace is atomic.
+	if to.exists() {
+		fs::remove_file(to)?;
+	}
+	fs::rename(from, to)?;
+	Ok(())
+}
+
+/// Apply any durable compact temps into place and clear the journal.
+/// Safe to call when some temps are already gone (partial prior apply).
+fn apply_compact_temps(data_dir: &Path) -> io::Result<()> {
+	replace_if_present(
+		&data_dir.join(PMMR_HASH_FILE).with_extension("tmp"),
+		&data_dir.join(PMMR_HASH_FILE),
+	)?;
+	replace_if_present(
+		&data_dir.join(PMMR_DATA_FILE).with_extension("tmp"),
+		&data_dir.join(PMMR_DATA_FILE),
+	)?;
+	// Size file is optional (fixed-size backends never create one).
+	replace_if_present(
+		&data_dir.join(PMMR_SIZE_FILE).with_extension("tmp"),
+		&data_dir.join(PMMR_SIZE_FILE),
+	)?;
+
+	let prune_path = data_dir.join(PMMR_PRUN_FILE);
+	replace_if_present(&compact_staging_path(&prune_path), &prune_path)?;
+
+	let leaf_path = data_dir.join(PMMR_LEAF_FILE);
+	replace_if_present(&compact_staging_path(&leaf_path), &leaf_path)?;
+
+	let journal = data_dir.join(PMMR_COMPACT_JOURNAL);
+	if journal.exists() {
+		fs::remove_file(&journal)?;
+	}
+	Ok(())
+}
+
+/// Drop orphaned compact staging files left by a crash during prep (no journal).
+fn cleanup_orphan_compact_temps(data_dir: &Path) -> io::Result<()> {
+	let orphans = [
+		data_dir.join(PMMR_HASH_FILE).with_extension("tmp"),
+		data_dir.join(PMMR_DATA_FILE).with_extension("tmp"),
+		data_dir.join(PMMR_SIZE_FILE).with_extension("tmp"),
+		compact_staging_path(&data_dir.join(PMMR_PRUN_FILE)),
+		compact_staging_path(&data_dir.join(PMMR_LEAF_FILE)),
+	];
+	for path in &orphans {
+		if path.exists() {
+			fs::remove_file(path)?;
+		}
+	}
+	Ok(())
+}
+
+/// On open: roll forward a committed compact, or discard uncommitted prep temps.
+fn recover_compact(data_dir: &Path) -> io::Result<()> {
+	let journal = data_dir.join(PMMR_COMPACT_JOURNAL);
+	if journal.exists() {
+		debug!(
+			"pmmr: completing interrupted compaction via journal in {:?}",
+			data_dir
+		);
+		apply_compact_temps(data_dir)?;
+	} else {
+		cleanup_orphan_compact_temps(data_dir)?;
+	}
+	Ok(())
+}
+
+fn write_compact_journal(data_dir: &Path) -> io::Result<()> {
+	let path = data_dir.join(PMMR_COMPACT_JOURNAL);
+	let mut file = File::create(&path)?;
+	// Version byte — existence of the file is the commit signal.
+	file.write_all(b"1")?;
+	file.sync_all()?;
+	// Best-effort directory fsync so the journal entry itself is durable.
+	#[cfg(unix)]
+	{
+		if let Ok(dir) = File::open(data_dir) {
+			let _ = dir.sync_all();
+		}
+	}
+	Ok(())
+}
+
 /// PMMR persistent backend implementation. Relies on multiple facilities to
 /// handle writing, reading and pruning.
 ///
@@ -289,6 +429,12 @@ impl<T: PMMRable> PMMRBackend<T> {
 	) -> io::Result<PMMRBackend<T>> {
 		let data_dir = data_dir.as_ref();
 
+		// Complete or abandon an interrupted compaction before opening files so
+		// hash/data/prune/leaf never disagree after a hard kill mid-compact.
+		if prunable {
+			recover_compact(data_dir)?;
+		}
+
 		// Are we dealing with "fixed size" data elements or "variable size" data elements
 		// maintained in an associated size file?
 		let size_info = if let Some(fixed_size) = T::elmt_size() {
@@ -413,6 +559,12 @@ impl<T: PMMRable> PMMRBackend<T> {
 	/// aligned. The block_marker in the db/index for the particular block
 	/// will have a suitable output_pos. This is used to enforce a horizon
 	/// after which the local node should have all the data to allow rewinding.
+	///
+	/// Crash safety: all new files are written to staging paths first, then a
+	/// journal marker is fsynced. Only after the journal is durable are live
+	/// files replaced. If the process is killed mid-replace, the next
+	/// `PMMRBackend::new` rolls the renames forward from the journal so
+	/// hash/data/prune/leaf stay consistent.
 	pub fn check_compact(&mut self, cutoff_pos: u64, rewind_rm_pos: &Bitmap) -> io::Result<bool> {
 		assert!(self.prunable, "Trying to compact a non-prunable PMMR");
 
@@ -420,6 +572,8 @@ impl<T: PMMRable> PMMRBackend<T> {
 		// on the cutoff_pos provided.
 		let (leaves_removed, pos_to_rm) = self.pos_to_rm(cutoff_pos, rewind_rm_pos);
 
+		// --- Phase 1: write all staging files (live files untouched) ---
+
 		// Save compact copy of the hash file, skipping removed data.
 		{
 			let pos_to_rm = map_vec!(pos_to_rm, |pos1| {
@@ -429,6 +583,7 @@ impl<T: PMMRable> PMMRBackend<T> {
 
 			self.hash_file.write_tmp_pruned(&pos_to_rm)?;
 		}
+		compact_test_hook("hash_tmp");
 
 		// Save compact copy of the data file, skipping removed leaves.
 		{
@@ -446,27 +601,39 @@ impl<T: PMMRable> PMMRBackend<T> {
 
 			self.data_file.write_tmp_pruned(&pos_to_rm)?;
 		}
+		compact_test_hook("data_tmp");
+
+		// Stage the post-compact prune list and leaf set next to the data temps.
+		let mut new_prune_bitmap = self.prune_list.bitmap();
+		new_prune_bitmap.or_inplace(&leaves_removed);
+		let prune_path = self.data_dir.join(PMMR_PRUN_FILE);
+		let prune_staging = compact_staging_path(&prune_path);
+		let new_prune = PruneList::new(Some(prune_path), new_prune_bitmap);
+		new_prune.write_to(&prune_staging)?;
+
+		let leaf_staging = compact_staging_path(&self.data_dir.join(PMMR_LEAF_FILE));
+		self.leaf_set.write_to(&leaf_staging)?;
+
+		// --- Phase 2: journal commit point (temps are durable) ---
+		write_compact_journal(&self.data_dir)?;
+		compact_test_hook("journal");
+
+		// --- Phase 3: replace live files, then clear journal ---
+		debug!("compact: applying journaled file replacements...");
+		// Release mmaps/handles so renames succeed on all platforms.
+		self.hash_file.release();
+		self.data_file.release();
+		apply_compact_temps(&self.data_dir)?;
 
-		// Replace hash and data files with compact copies.
-		// Rebuild and intialize from the new files.
-		{
-			debug!("compact: about to replace hash and data files and rebuild...");
-			self.hash_file.replace_with_tmp()?;
-			self.data_file.replace_with_tmp()?;
-			debug!("compact: ...finished replacing and rebuilding");
-		}
-
-		// Update the prune list and write to disk.
-		{
-			let mut bitmap = self.prune_list.bitmap();
-			bitmap.or_inplace(&leaves_removed);
-			self.prune_list = PruneList::new(Some(self.data_dir.join(PMMR_PRUN_FILE)), bitmap);
-			self.prune_list.flush()?;
-		}
+		// Re-open hash/data from the new files (rebuilds size file if needed).
+		self.hash_file.reinit_from_disk()?;
+		self.data_file.reinit_from_disk()?;
+		debug!("compact: ...finished applying replacements");
+		compact_test_hook("applied");
 
-		// Write the leaf_set to disk.
-		// Optimize the bitmap storage in the process.
-		self.leaf_set.flush()?;
+		// In-memory state matches the files we just installed.
+		self.prune_list = new_prune;
+		self.leaf_set.mark_flushed();
 
 		self.clean_rewind_files()?;
 
diff --git a/store/src/prune_list.rs b/store/src/prune_list.rs
index 1066016e86..f8d7496a22 100644
--- a/store/src/prune_list.rs
+++ b/store/src/prune_list.rs
@@ -129,6 +129,16 @@ impl PruneList {
 		Ok(())
 	}
 
+	/// Write the current prune bitmap to an arbitrary path (used for journaled compact).
+	pub fn write_to<P: AsRef<Path>>(&self, path: P) -> io::Result<()> {
+		let mut bitmap = self.bitmap.clone();
+		bitmap.run_optimize();
+		let mut file = std::fs::File::create(path.as_ref())?;
+		file.write_all(&bitmap.serialize::<Portable>())?;
+		file.sync_all()?;
+		Ok(())
+	}
+
 	/// Discard in-memory changes and restore the last flushed prune_list state.
 	pub fn discard(&mut self) -> io::Result<()> {
 		let path = match self.path.clone() {
diff --git a/store/src/types.rs b/store/src/types.rs
index 70cde93a2d..9e881369b2 100644
--- a/store/src/types.rs
+++ b/store/src/types.rs
@@ -165,6 +165,16 @@ where
 	pub fn replace_with_tmp(&mut self) -> io::Result<()> {
 		self.file.replace_with_tmp()
 	}
+
+	/// Re-open from disk after an external file replace (journaled compact).
+	pub fn reinit_from_disk(&mut self) -> io::Result<()> {
+		self.file.reinit_from_disk()
+	}
+
+	/// Path of the pruned temporary file written by `write_tmp_pruned`.
+	pub fn tmp_path(&self) -> PathBuf {
+		self.file.tmp_path()
+	}
 }
 
 /// Wrapper for a file that can be read at any position (random read) but for
@@ -490,10 +500,6 @@ where
 		Ok(file)
 	}
 
-	fn tmp_path(&self) -> PathBuf {
-		self.path.with_extension("tmp")
-	}
-
 	/// Saves a copy of the current file content, skipping data at the provided
 	/// prune positions. prune_pos must be ordered.
 	pub fn write_tmp_pruned(&self, prune_pos: &[u64]) -> io::Result<()> {
@@ -507,7 +513,9 @@ where
 		let mut current_pos = 0;
 		let mut prune_pos = prune_pos;
 		while let Ok(elmt) = T::read(&mut streaming_reader) {
-			if prune_pos.contains(&current_pos) {
+			// prune_pos is ordered so we only ever need to check the head,
+			// avoiding a scan of the remaining positions for every element.
+			if prune_pos.first() == Some(&current_pos) {
 				// Pruned pos, moving on.
 				prune_pos = &prune_pos[1..];
 			} else {
@@ -517,10 +525,17 @@ where
 			}
 			current_pos += 1;
 		}
-		buf_writer.flush()?;
+		// Flush and fsync so the temp is durable before a compact journal is written.
+		let file = buf_writer.into_inner()?;
+		file.sync_all()?;
 		Ok(())
 	}
 
+	/// Path of the pruned temporary file written by `write_tmp_pruned`.
+	pub fn tmp_path(&self) -> PathBuf {
+		self.path.with_extension("tmp")
+	}
+
 	/// Replace the underlying file with the file at tmp path.
 	/// Rebuild and initialize from the new file.
 	pub fn replace_with_tmp(&mut self) -> io::Result<()> {
@@ -540,6 +555,26 @@ where
 		Ok(())
 	}
 
+	/// Re-open this file from disk after an external replace (e.g. journaled compact).
+	/// Rebuilds the associated size file if it is inconsistent with the data.
+	pub fn reinit_from_disk(&mut self) -> io::Result<()> {
+		self.buffer.clear();
+		self.buffer_start_pos_bak = 0;
+		self.release();
+		self.init()?;
+
+		// Same consistency repair as open(): size_file may lag the data file
+		// if we crashed between replacing data and rebuilding sizes.
+		let expected_size = self.size()?;
+		if let SizeInfo::VariableSize(ref mut size_file) = &mut self.size_info {
+			if size_file.sum_sizes()? != expected_size {
+				self.rebuild_size_file()?;
+				self.init()?;
+			}
+		}
+		Ok(())
+	}
+
 	fn rebuild_size_file(&mut self) -> io::Result<()> {
 		if let SizeInfo::VariableSize(ref mut size_file) = &mut self.size_info {
 			// Note: Reading from data file and writing sizes to the associated (tmp) size_file.
diff --git a/store/tests/pmmr.rs b/store/tests/pmmr.rs
index 935b39a6b4..b8b2acb519 100644
--- a/store/tests/pmmr.rs
+++ b/store/tests/pmmr.rs
@@ -813,6 +813,170 @@ fn compact_twice() {
 	teardown(data_dir);
 }
 
+/// Compaction must leave no journal or staging files behind on success.
+#[test]
+fn compact_clears_journal() {
+	let (data_dir, elems) = setup("compact_clears_journal");
+	{
+		let mut backend =
+			store::pmmr::PMMRBackend::new(data_dir.to_string(), true, ProtocolVersion(1), None)
+				.unwrap();
+		let mmr_size = load(0, &elems[0..8], &mut backend);
+		backend.sync().unwrap();
+		{
+			let mut pmmr: PMMR<'_, TestElem, _> = PMMR::at(&mut backend, mmr_size);
+			pmmr.prune(0).unwrap();
+			pmmr.prune(1).unwrap();
+		}
+		backend.sync().unwrap();
+		backend.check_compact(mmr_size, &Bitmap::new()).unwrap();
+
+		let dir = std::path::Path::new(&data_dir);
+		assert!(!dir.join("pmmr_compact.journal").exists());
+		assert!(!dir.join("pmmr_hash.tmp").exists());
+		assert!(!dir.join("pmmr_data.tmp").exists());
+		assert!(!dir.join("pmmr_prun.bin.compact").exists());
+		assert!(!dir.join("pmmr_leaf.bin.compact").exists());
+	}
+	teardown(data_dir);
+}
+
+/// If a journal is present with staged files, open rolls renames forward and
+/// yields a consistent PMMR (simulates hard kill after journal, before apply).
+#[test]
+fn compact_journal_recovery_roll_forward() {
+	let (data_dir, elems) = setup("compact_journal_recovery");
+	let dir = std::path::Path::new(&data_dir);
+
+	let (root, mmr_size) = {
+		let mut backend =
+			store::pmmr::PMMRBackend::new(data_dir.to_string(), true, ProtocolVersion(1), None)
+				.unwrap();
+		let mmr_size = load(0, &elems[0..8], &mut backend);
+		backend.sync().unwrap();
+		let root = {
+			let pmmr: PMMR<'_, TestElem, _> = PMMR::at(&mut backend, mmr_size);
+			pmmr.root().unwrap()
+		};
+		{
+			let mut pmmr: PMMR<'_, TestElem, _> = PMMR::at(&mut backend, mmr_size);
+			pmmr.prune(0).unwrap();
+			pmmr.prune(1).unwrap();
+			pmmr.prune(3).unwrap();
+		}
+		backend.sync().unwrap();
+
+		// Snapshot pre-compact live files.
+		let pre_hash = fs::read(dir.join("pmmr_hash.bin")).unwrap();
+		let pre_data = fs::read(dir.join("pmmr_data.bin")).unwrap();
+		let pre_prun = fs::read(dir.join("pmmr_prun.bin")).unwrap_or_default();
+		let pre_leaf = fs::read(dir.join("pmmr_leaf.bin")).unwrap();
+
+		backend.check_compact(mmr_size, &Bitmap::new()).unwrap();
+
+		// Snapshot post-compact live files (the committed new state).
+		let post_hash = fs::read(dir.join("pmmr_hash.bin")).unwrap();
+		let post_data = fs::read(dir.join("pmmr_data.bin")).unwrap();
+		let post_prun = fs::read(dir.join("pmmr_prun.bin")).unwrap();
+		let post_leaf = fs::read(dir.join("pmmr_leaf.bin")).unwrap();
+
+		// Simulate crash after journal write, before renames:
+		// live files are still pre-compact; temps hold post-compact content.
+		fs::write(dir.join("pmmr_hash.bin"), &pre_hash).unwrap();
+		fs::write(dir.join("pmmr_data.bin"), &pre_data).unwrap();
+		fs::write(dir.join("pmmr_prun.bin"), &pre_prun).unwrap();
+		fs::write(dir.join("pmmr_leaf.bin"), &pre_leaf).unwrap();
+		fs::write(dir.join("pmmr_hash.tmp"), &post_hash).unwrap();
+		fs::write(dir.join("pmmr_data.tmp"), &post_data).unwrap();
+		fs::write(dir.join("pmmr_prun.bin.compact"), &post_prun).unwrap();
+		fs::write(dir.join("pmmr_leaf.bin.compact"), &post_leaf).unwrap();
+		fs::write(dir.join("pmmr_compact.journal"), b"1").unwrap();
+
+		(root, mmr_size)
+	};
+
+	// Re-open: recovery should roll forward and leave a consistent store.
+	{
+		let mut backend =
+			store::pmmr::PMMRBackend::new(data_dir.to_string(), true, ProtocolVersion(1), None)
+				.unwrap();
+		assert!(!dir.join("pmmr_compact.journal").exists());
+		assert!(!dir.join("pmmr_hash.tmp").exists());
+		let pmmr: PMMR<'_, TestElem, _> = PMMR::at(&mut backend, mmr_size);
+		assert_eq!(root, pmmr.root().unwrap());
+		assert_eq!(pmmr.get_data(4).unwrap(), TestElem(4));
+	}
+	teardown(data_dir);
+}
+
+/// Staging files without a journal are abandoned prep — discarded on open,
+/// leaving the pre-compact live files intact.
+#[test]
+fn compact_orphan_temps_discarded() {
+	let (data_dir, elems) = setup("compact_orphan_temps");
+	let dir = std::path::Path::new(&data_dir);
+
+	let (root, mmr_size, pre_hash_len) = {
+		let mut backend =
+			store::pmmr::PMMRBackend::new(data_dir.to_string(), true, ProtocolVersion(1), None)
+				.unwrap();
+		let mmr_size = load(0, &elems[0..8], &mut backend);
+		backend.sync().unwrap();
+		let root = {
+			let pmmr: PMMR<'_, TestElem, _> = PMMR::at(&mut backend, mmr_size);
+			pmmr.root().unwrap()
+		};
+		{
+			let mut pmmr: PMMR<'_, TestElem, _> = PMMR::at(&mut backend, mmr_size);
+			pmmr.prune(0).unwrap();
+			pmmr.prune(1).unwrap();
+		}
+		backend.sync().unwrap();
+
+		let pre_hash = fs::read(dir.join("pmmr_hash.bin")).unwrap();
+		let pre_data = fs::read(dir.join("pmmr_data.bin")).unwrap();
+		let pre_prun = fs::read(dir.join("pmmr_prun.bin")).unwrap_or_default();
+		let pre_leaf = fs::read(dir.join("pmmr_leaf.bin")).unwrap();
+		let pre_hash_len = pre_hash.len();
+
+		backend.check_compact(mmr_size, &Bitmap::new()).unwrap();
+
+		let post_hash = fs::read(dir.join("pmmr_hash.bin")).unwrap();
+		let post_data = fs::read(dir.join("pmmr_data.bin")).unwrap();
+		let post_prun = fs::read(dir.join("pmmr_prun.bin")).unwrap();
+		let post_leaf = fs::read(dir.join("pmmr_leaf.bin")).unwrap();
+
+		// Full pre-compact live state + post content staged as temps, no journal.
+		fs::write(dir.join("pmmr_hash.bin"), &pre_hash).unwrap();
+		fs::write(dir.join("pmmr_data.bin"), &pre_data).unwrap();
+		fs::write(dir.join("pmmr_prun.bin"), &pre_prun).unwrap();
+		fs::write(dir.join("pmmr_leaf.bin"), &pre_leaf).unwrap();
+		fs::write(dir.join("pmmr_hash.tmp"), &post_hash).unwrap();
+		fs::write(dir.join("pmmr_data.tmp"), &post_data).unwrap();
+		fs::write(dir.join("pmmr_prun.bin.compact"), &post_prun).unwrap();
+		fs::write(dir.join("pmmr_leaf.bin.compact"), &post_leaf).unwrap();
+
+		(root, mmr_size, pre_hash_len)
+	};
+
+	{
+		let mut backend =
+			store::pmmr::PMMRBackend::new(data_dir.to_string(), true, ProtocolVersion(1), None)
+				.unwrap();
+		assert!(!dir.join("pmmr_hash.tmp").exists());
+		assert!(!dir.join("pmmr_data.tmp").exists());
+		assert!(!dir.join("pmmr_prun.bin.compact").exists());
+		assert!(!dir.join("pmmr_leaf.bin.compact").exists());
+		assert_eq!(
+			fs::metadata(dir.join("pmmr_hash.bin")).unwrap().len() as usize,
+			pre_hash_len
+		);
+		let pmmr: PMMR<'_, TestElem, _> = PMMR::at(&mut backend, mmr_size);
+		assert_eq!(root, pmmr.root().unwrap());
+	}
+	teardown(data_dir);
+}
+
 #[test]
 fn cleanup_rewind_files_test() {
 	let expected = 10;
diff --git a/store/tests/segment.rs b/store/tests/segment.rs
index a24c1b8c5c..823a8970be 100644
--- a/store/tests/segment.rs
+++ b/store/tests/segment.rs
@@ -12,9 +12,9 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use crate::core::core::hash::DefaultHashable;
+use crate::core::core::hash::{DefaultHashable, Hash};
 use crate::core::core::pmmr;
-use crate::core::core::pmmr::segment::{Segment, SegmentIdentifier};
+use crate::core::core::pmmr::segment::{Segment, SegmentError, SegmentIdentifier};
 use crate::core::core::pmmr::{Backend, ReadablePMMR, ReadonlyPMMR, PMMR};
 use crate::core::ser::{
 	BinReader, BinWriter, DeserializationMode, Error, PMMRable, ProtocolVersion, Readable, Reader,
@@ -327,6 +327,103 @@ fn pruned_segment() {
 	fs::remove_dir_all(&data_dir).unwrap();
 }
 
+#[test]
+fn segment_carried_hash_verification() {
+	let t = Utc::now();
+	let data_dir = format!(
+		"./target/tmp/{}.{}-carried_hashes",
+		t.timestamp(),
+		t.timestamp_subsec_nanos()
+	);
+	fs::create_dir_all(&data_dir).unwrap();
+
+	let n_leaves = 16;
+	let mut ba = PMMRBackend::new(&data_dir, true, ProtocolVersion(1), None).unwrap();
+	let mut mmr = PMMR::new(&mut ba);
+	for i in 0..n_leaves {
+		mmr.push(&TestElem([i / 7, i / 5, i / 3, i])).unwrap();
+	}
+	let last_pos = mmr.unpruned_size();
+	let root = mmr.root().unwrap();
+
+	let mut bitmap = Bitmap::new();
+	bitmap.add_range(0..n_leaves);
+
+	// An unpruned segment carries hashes that root reconstruction recomputes
+	// from the leaves. Corrupting one of them must be rejected even though
+	// it does not affect the reconstructed root.
+	let id = SegmentIdentifier { height: 2, idx: 0 };
+	let ro_mmr = ReadonlyPMMR::at(&mut ba, last_pos);
+	let segment = Segment::from_pmmr(id, &ro_mmr, true).unwrap();
+	segment.validate(last_pos, Some(&bitmap), root).unwrap();
+	let (sid, hash_pos, mut hashes, leaf_pos, leaf_data, proof) = segment.parts();
+	let idx = hash_pos.iter().position(|&p| p == 2).unwrap();
+	hashes[idx] = Hash::default();
+	let tampered = Segment::from_parts(sid, hash_pos, hashes, leaf_pos, leaf_data, proof);
+	assert_eq!(
+		tampered.validate(last_pos, Some(&bitmap), root),
+		Err(SegmentError::UnverifiableHash(2))
+	);
+
+	// Prune all leaves of segment 1, remembering the interior hashes as an
+	// uncompacted peer would still serve them, then compact so the segment
+	// carries a single hash: the pruned subtree root at pos 13.
+	let mut mmr = PMMR::at(&mut ba, last_pos);
+	prune(&mut mmr, &mut bitmap, &[4, 5, 6, 7]);
+	ba.sync().unwrap();
+	let h9 = ba.get_hash(9).unwrap();
+	let h12 = ba.get_hash(12).unwrap();
+	ba.check_compact(last_pos, &Bitmap::new()).unwrap();
+	ba.sync().unwrap();
+
+	let id = SegmentIdentifier { height: 2, idx: 1 };
+	let ro_mmr = ReadonlyPMMR::at(&mut ba, last_pos);
+	let segment = Segment::from_pmmr(id, &ro_mmr, true).unwrap();
+	assert_eq!(segment.hash_iter().count(), 1);
+	segment.validate(last_pos, Some(&bitmap), root).unwrap();
+
+	// A segment from an uncompacted peer additionally carries the children
+	// of the pruned subtree root. They hash into the (proof-bound) root at
+	// pos 13, so the segment must validate.
+	let (sid, hash_pos, hashes, leaf_pos, leaf_data, proof) = segment.parts();
+	let mut dense_hash_pos = hash_pos.clone();
+	let mut dense_hashes = hashes.clone();
+	dense_hash_pos.splice(0..0, [9, 12]);
+	dense_hashes.splice(0..0, [h9, h12]);
+	let dense = Segment::from_parts(
+		sid,
+		dense_hash_pos,
+		dense_hashes,
+		leaf_pos.clone(),
+		leaf_data.clone(),
+		proof.clone(),
+	);
+	dense.validate(last_pos, Some(&bitmap), root).unwrap();
+
+	// Root reconstruction never looks beneath the pruned subtree root, but
+	// apply pushes every carried hash into the local PMMR. An extra hash
+	// beneath the root that cannot be verified against it must be rejected.
+	let mut tampered_hash_pos = hash_pos.clone();
+	let mut tampered_hashes = hashes.clone();
+	tampered_hash_pos.insert(0, 9);
+	tampered_hashes.insert(0, Hash::default());
+	let tampered = Segment::from_parts(
+		sid,
+		tampered_hash_pos,
+		tampered_hashes,
+		leaf_pos,
+		leaf_data,
+		proof,
+	);
+	assert_eq!(
+		tampered.validate(last_pos, Some(&bitmap), root),
+		Err(SegmentError::UnverifiableHash(9))
+	);
+
+	std::mem::drop(ba);
+	fs::remove_dir_all(&data_dir).unwrap();
+}
+
 #[test]
 fn ser_round_trip() {
 	let t = Utc::now();