From e5c826bcba2277ee383d503a5e46dd4e70c2f907 Mon Sep 17 00:00:00 2001
From: venkat1701 <thejeastdev@gmail.com>
Date: Sat, 10 Oct 2026 19:11:41 +0530
Subject: [PATCH 1/6] refactor(engine): let PageStore copy a page into the
 active segment

PageStore.copy reads a page, checks its header and checksum, and writes
the same bytes to space claimed in the active segment, returning the new
address. It doesn't touch the directory; the caller points the entry at
the copy once the bytes are synced. Allocation and copying share the code
that claims space and rolls segments.

Directory entries are now written with release and read with acquire, so
a reader that races a move sees the old address or the new one.
---
 .../io/hstore/engine/page/PageDirectory.java  |  6 ++--
 .../java/io/hstore/engine/page/PageStore.java | 35 +++++++++++++++----
 2 files changed, 32 insertions(+), 9 deletions(-)

diff --git a/engine/src/main/java/io/hstore/engine/page/PageDirectory.java b/engine/src/main/java/io/hstore/engine/page/PageDirectory.java
index 85af7ff..c7e2c50 100644
--- a/engine/src/main/java/io/hstore/engine/page/PageDirectory.java
+++ b/engine/src/main/java/io/hstore/engine/page/PageDirectory.java
@@ -7,6 +7,7 @@
 import java.lang.foreign.Arena;
 import java.lang.foreign.MemorySegment;
 import java.lang.foreign.ValueLayout;
+import java.lang.invoke.VarHandle;
 import java.nio.ByteOrder;
 import java.nio.channels.FileChannel;
 import java.nio.file.Path;
@@ -25,6 +26,7 @@ final class PageDirectory implements AutoCloseable {
     private static final long CHUNK_BYTES = 1L << CHUNK_SHIFT;
     private static final ValueLayout.OfInt INT = ValueLayout.JAVA_INT.withOrder(ByteOrder.LITTLE_ENDIAN);
     private static final ValueLayout.OfLong LONG = ValueLayout.JAVA_LONG.withOrder(ByteOrder.LITTLE_ENDIAN);
+    private static final VarHandle ENTRY = LONG.varHandle();
 
     private final Path path;
     private final FileChannel channel;
@@ -62,7 +64,7 @@ long get(long index) {
         long position = HEADER + index * Long.BYTES;
         MemorySegment[] mapped = chunks;
         int chunk = (int) (position >>> CHUNK_SHIFT);
-        return chunk < mapped.length ? mapped[chunk].get(LONG, position & (CHUNK_BYTES - 1)) : PageAddress.NONE;
+        return chunk < mapped.length ? (long) ENTRY.getAcquire(mapped[chunk], position & (CHUNK_BYTES - 1)) : PageAddress.NONE;
     }
 
     void put(long index, long address) {
@@ -71,7 +73,7 @@ void put(long index, long address) {
         if (chunk >= chunks.length) {
             mapThrough(chunk);
         }
-        chunks[chunk].set(LONG, position & (CHUNK_BYTES - 1), address);
+        ENTRY.setRelease(chunks[chunk], position & (CHUNK_BYTES - 1), address);
     }
 
     long allocate() {
diff --git a/engine/src/main/java/io/hstore/engine/page/PageStore.java b/engine/src/main/java/io/hstore/engine/page/PageStore.java
index 62a7521..bc96948 100644
--- a/engine/src/main/java/io/hstore/engine/page/PageStore.java
+++ b/engine/src/main/java/io/hstore/engine/page/PageStore.java
@@ -87,20 +87,41 @@ public long allocate(int length) {
         int units = PageId.unitsFor(length);
         allocation.lock();
         try {
-            if (activeSegment == 0 || nextUnit + units > segmentUnits) {
-                rollSegment();
-            }
-            int offset = nextUnit;
-            nextUnit += units;
-            segments.computeIfPresent(activeSegment, (_, info) -> info.withAllocation(info.pages() + 1, Math.max(info.units(), offset + (long) units)));
             long index = addresses.allocate();
-            addresses.put(index, new PageAddress(activeSegment, offset, units).pack());
+            addresses.put(index, claim(units));
             return PageId.pack(index, 0);
         } finally {
             allocation.unlock();
         }
     }
 
+    public long copy(long pageId) {
+        long from = locate(pageId);
+        MemorySegment image = read(pageId);
+        PageHeader.verify(image, pageId);
+        long to;
+        allocation.lock();
+        try {
+            to = claim(PageAddress.unitsOf(from));
+        } finally {
+            allocation.unlock();
+        }
+        files.get(PageAddress.segmentOf(to)).write((long) PageAddress.offsetOf(to) * PageId.UNIT_BYTES, image);
+        pagesWritten.incrementAndGet();
+        bytesWritten.addAndGet(image.byteSize());
+        return to;
+    }
+
+    private long claim(int units) {
+        if (activeSegment == 0 || nextUnit + units > segmentUnits) {
+            rollSegment();
+        }
+        int offset = nextUnit;
+        nextUnit += units;
+        segments.computeIfPresent(activeSegment, (_, info) -> info.withAllocation(info.pages() + 1, Math.max(info.units(), offset + (long) units)));
+        return new PageAddress(activeSegment, offset, units).pack();
+    }
+
     private void rollSegment() {
         if (activeSegment != 0) {
             segments.computeIfPresent(activeSegment, (_, info) -> info.withState(SegmentState.SEALED, 0));

From 198e7db52e72f3346d003131fd2885d3095cb3bc Mon Sep 17 00:00:00 2001
From: venkat1701 <thejeastdev@gmail.com>
Date: Sat, 10 Oct 2026 19:11:51 +0530
Subject: [PATCH 2/6] feat(engine): compact by moving pages instead of
 rewriting trees

Compaction used to rebuild every node on the path to each page it moved,
inside a commit that held the commit lock for the whole rewrite. At scale
4 a full pass allocated 6 to 9 GB and could not finish in a 4 GB heap.

Page numbers are stable now, so a live page in a victim segment is
copied byte for byte to the active segment and its directory entry is
pointed at the copy. Copies are synced in 4 MiB passes before their
entries change, so an entry never reaches the disk ahead of its bytes.
No root changes and nothing commits, so writers carry on during
compaction and the node cache stays valid.

Liveness is one bit per page number. It walks the generations taken in
a short cut under the commit lock, and only segments already sealed at
that cut can be victims. Without the cut, a segment sealed during the
walk could hold pages the walk never saw.

A retired segment is deleted after the next checkpoint, which makes the
new entries durable and drops the WAL records that still name the old
addresses, and once no reader pinned at or before the move remains.
History no longer holds segments back, since old generations reach moved
pages through the directory too.

Bulk loads that spilled pages before a compaction are still refused,
now by counting compactions instead of comparing generations, and on the
fast path as well. A crash mid-compaction leaves the segment SEALED for
the next pass. COMPACTION_MOVE and COMPACTION_RETIRE are new crash points,
kept out of the commit crash matrix.
---
 .../java/io/hstore/engine/StorageEngine.java  |   6 +-
 .../engine/maintenance/Checkpointer.java      |   6 +
 .../hstore/engine/maintenance/Compactor.java  | 149 ++++++++++++++----
 .../java/io/hstore/engine/page/PageStore.java |   3 +
 .../io/hstore/engine/tree/TreeWalker.java     |  67 ++------
 .../java/io/hstore/engine/txn/CrashPoint.java |   4 +-
 .../io/hstore/engine/txn/Transaction.java     |   6 +
 .../hstore/engine/txn/TransactionManager.java |  27 +---
 .../io/hstore/engine/CrashRecoveryTest.java   |   3 +-
 9 files changed, 160 insertions(+), 111 deletions(-)

diff --git a/engine/src/main/java/io/hstore/engine/StorageEngine.java b/engine/src/main/java/io/hstore/engine/StorageEngine.java
index fc2361f..7eab8a8 100644
--- a/engine/src/main/java/io/hstore/engine/StorageEngine.java
+++ b/engine/src/main/java/io/hstore/engine/StorageEngine.java
@@ -82,7 +82,7 @@ private StorageEngine(Path directory, EngineOptions options, FileChannel lockCha
                 options.derivations(), options.durability(), options.walMode(), options.faults(), options.historyLimit());
         this.transactions = new TransactionManager(storage, recovery.image().current(), recovery.image().history());
         this.checkpointer = new Checkpointer(transactions, pages, wal, feed, catalog, options.historyLimit(), checkpoint.checkpointLsn());
-        this.compactor = new Compactor(transactions, pages, slots);
+        this.compactor = new Compactor(transactions, pages, slots, checkpointer, options.faults());
         this.dictionary = new Dictionary(transactions);
         checkpointer.checkpoint();
         compactor.reclaim();
@@ -274,7 +274,7 @@ public Compactor.Report compact() {
             Compactor.Report report = compactor.compact(options.compactionLiveRatio());
             timedCheckpoint();
             List<Integer> reclaimed = compactor.reclaim();
-            LOG.log(System.Logger.Level.INFO, "compaction relocated segments {0} and reclaimed {1}", report.compacted(), reclaimed);
+            LOG.log(System.Logger.Level.INFO, "compaction moved the live pages out of segments {0} and reclaimed {1}", report.compacted(), reclaimed);
             return report;
         } finally {
             maintenance.unlock();
@@ -308,7 +308,7 @@ private void maintain() {
                         Compactor.Report report = compactor.compact(options.compactionLiveRatio(), 1);
                         if (!report.compacted().isEmpty()) {
                             timedCheckpoint();
-                            LOG.log(System.Logger.Level.INFO, "background compaction relocated segments {0}", report.compacted());
+                            LOG.log(System.Logger.Level.INFO, "background compaction moved the live pages out of segments {0}", report.compacted());
                         }
                     }
                     List<Integer> reclaimed = compactor.reclaim();
diff --git a/engine/src/main/java/io/hstore/engine/maintenance/Checkpointer.java b/engine/src/main/java/io/hstore/engine/maintenance/Checkpointer.java
index ab97b75..f4b627b 100644
--- a/engine/src/main/java/io/hstore/engine/maintenance/Checkpointer.java
+++ b/engine/src/main/java/io/hstore/engine/maintenance/Checkpointer.java
@@ -23,6 +23,7 @@ public record Result(long generation, long lsn, int walSegments) {
     private final CatalogStore catalog;
     private final int historyLimit;
     private volatile long lastLsn;
+    private volatile long completed;
 
     public Checkpointer(TransactionManager transactions, PageStore pages, WriteAheadLog wal, ChangeFeed feed,
                         CatalogStore catalog, int historyLimit, long recoveredLsn) {
@@ -39,6 +40,10 @@ public long walSinceCheckpoint() {
         return wal.end() - lastLsn;
     }
 
+    public long completed() {
+        return completed;
+    }
+
     public Result checkpoint() {
         return transactions.exclusive(() -> {
             long lsn = wal.end();
@@ -55,6 +60,7 @@ public Result checkpoint() {
             wal.sync();
             wal.truncateBefore(lsn);
             lastLsn = lsn;
+            completed++;
             return new Result(current.id(), lsn, wal.segmentCount());
         });
     }
diff --git a/engine/src/main/java/io/hstore/engine/maintenance/Compactor.java b/engine/src/main/java/io/hstore/engine/maintenance/Compactor.java
index 3bb7932..b39f87b 100644
--- a/engine/src/main/java/io/hstore/engine/maintenance/Compactor.java
+++ b/engine/src/main/java/io/hstore/engine/maintenance/Compactor.java
@@ -1,24 +1,25 @@
 package io.hstore.engine.maintenance;
 
-import io.hstore.engine.catalog.Branch;
 import io.hstore.engine.catalog.Generation;
 import io.hstore.engine.catalog.SlotRegistry;
+import io.hstore.engine.page.PageAddress;
 import io.hstore.engine.page.PageId;
 import io.hstore.engine.page.PageStore;
 import io.hstore.engine.page.SegmentInfo;
 import io.hstore.engine.page.SegmentState;
 import io.hstore.engine.tree.TreeWalker;
-import io.hstore.engine.tree.WriteScope;
+import io.hstore.engine.txn.CrashPoint;
 import io.hstore.engine.txn.TransactionManager;
 
+import java.util.Arrays;
+import java.util.BitSet;
 import java.util.Comparator;
-import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.OptionalLong;
 import java.util.Set;
 import java.util.TreeMap;
-import java.util.function.LongPredicate;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
@@ -27,25 +28,72 @@ public final class Compactor {
     public record Report(Map<Integer, Long> liveBytes, List<Integer> compacted, long generation) {
     }
 
+    private record Cut(List<Generation> generations, Set<Integer> sealed) {
+    }
+
+    private static final long PASS_BYTES = 4L << 20;
+
     private final TransactionManager transactions;
     private final PageStore pages;
     private final SlotRegistry slots;
+    private final Checkpointer checkpointer;
+    private final CrashPoint.Injector faults;
+    private final Map<Integer, Long> retiredBeforeCheckpoint = new ConcurrentHashMap<>();
 
-    public Compactor(TransactionManager transactions, PageStore pages, SlotRegistry slots) {
+    public Compactor(TransactionManager transactions, PageStore pages, SlotRegistry slots, Checkpointer checkpointer,
+                     CrashPoint.Injector faults) {
         this.transactions = transactions;
         this.pages = pages;
         this.slots = slots;
+        this.checkpointer = checkpointer;
+        this.faults = faults;
     }
 
     public Map<Integer, Long> liveness() {
-        Map<Long, Integer> visited = new HashMap<>();
+        return bytesBySegment(reachable(retained()));
+    }
+
+    private List<Generation> retained() {
+        return Stream.concat(transactions.history().stream(), Stream.of(transactions.current())).toList();
+    }
+
+    private Cut cut() {
+        return transactions.exclusive(() -> {
+            int active = pages.activeSegment();
+            Set<Integer> sealed = pages.segments().stream()
+                    .filter(segment -> segment.state() == SegmentState.SEALED && segment.id() != active)
+                    .map(SegmentInfo::id)
+                    .collect(Collectors.toUnmodifiableSet());
+            return new Cut(retained(), sealed);
+        });
+    }
+
+    private BitSet reachable(List<Generation> generations) {
+        BitSet live = new BitSet();
         TreeWalker walker = new TreeWalker(transactions.source());
-        Stream.concat(transactions.history().stream(), Stream.of(transactions.current()))
+        generations.stream()
                 .flatMap(generation -> generation.branches().values().stream())
                 .forEach(branch -> branch.roots().roots().forEach((slot, ref) ->
-                        walker.visit(ref, slots.slot(slot).schema(), visited)));
-        return visited.entrySet().stream().collect(Collectors.groupingBy(entry -> pages.segmentOf(entry.getKey()), TreeMap::new,
-                Collectors.summingLong(entry -> (long) entry.getValue() * PageId.UNIT_BYTES)));
+                        walker.visit(ref, slots.slot(slot).schema(), pageId -> mark(live, pageId))));
+        return live;
+    }
+
+    private static boolean mark(BitSet live, long pageId) {
+        int index = Math.toIntExact(PageId.indexOf(pageId));
+        if (live.get(index)) {
+            return false;
+        }
+        live.set(index);
+        return true;
+    }
+
+    private Map<Integer, Long> bytesBySegment(BitSet live) {
+        Map<Integer, Long> bytes = new TreeMap<>();
+        live.stream().forEach(index -> {
+            long address = pages.addressOf(PageId.pack(index, 0));
+            bytes.merge(PageAddress.segmentOf(address), (long) PageAddress.unitsOf(address) * PageId.UNIT_BYTES, Long::sum);
+        });
+        return bytes;
     }
 
     public Report compact(double liveThreshold) {
@@ -53,44 +101,81 @@ public Report compact(double liveThreshold) {
     }
 
     public Report compact(double liveThreshold, int maxVictims) {
-        Map<Integer, Long> live = liveness();
-        int active = pages.activeSegment();
+        Cut cut = cut();
+        BitSet live = reachable(cut.generations());
+        Map<Integer, Long> liveBytes = bytesBySegment(live);
         List<Integer> victims = pages.segments().stream()
-                .filter(segment -> segment.state() == SegmentState.SEALED && segment.id() != active)
-                .filter(segment -> live.getOrDefault(segment.id(), 0L) < segment.bytes() * liveThreshold)
-                .sorted(Comparator.comparingDouble(segment -> (double) live.getOrDefault(segment.id(), 0L) / Math.max(1, segment.bytes())))
+                .filter(segment -> segment.state() == SegmentState.SEALED && cut.sealed().contains(segment.id()))
+                .filter(segment -> liveBytes.getOrDefault(segment.id(), 0L) < segment.bytes() * liveThreshold)
+                .sorted(Comparator.comparingDouble(segment -> (double) liveBytes.getOrDefault(segment.id(), 0L) / Math.max(1, segment.bytes())))
                 .limit(maxVictims)
                 .map(SegmentInfo::id)
                 .toList();
-        long generation = transactions.current().id();
         if (victims.isEmpty()) {
-            return new Report(live, victims, generation);
+            return new Report(liveBytes, victims, transactions.current().id());
         }
         victims.forEach(id -> pages.transition(id, SegmentState.COMPACTING, 0));
-        Set<Integer> victimSet = Set.copyOf(victims);
-        LongPredicate moving = pageId -> victimSet.contains(pages.segmentOf(pageId));
-        TreeWalker walker = new TreeWalker(transactions.source());
-        List<Integer> branches = transactions.current().branches().values().stream()
-                .filter(Branch::isActive).map(Branch::id).toList();
-        for (int branch : branches) {
-            generation = transactions.rewrite(branch, roots -> roots.map((slot, ref) ->
-                    walker.relocate(ref, slots.slot(slot).schema(), moving, new WriteScope())));
+        move(live, Set.copyOf(victims));
+        faults.reach(CrashPoint.COMPACTION_MOVE);
+        long retiredAt = transactions.current().id();
+        transactions.compacted();
+        long checkpoints = checkpointer.completed();
+        victims.forEach(id -> {
+            pages.transition(id, SegmentState.RETIRED, retiredAt);
+            retiredBeforeCheckpoint.put(id, checkpoints);
+        });
+        faults.reach(CrashPoint.COMPACTION_RETIRE);
+        return new Report(liveBytes, victims, retiredAt);
+    }
+
+    private void move(BitSet live, Set<Integer> victims) {
+        long[] moved = new long[64];
+        int count = 0;
+        long bytes = 0;
+        for (int index = live.nextSetBit(0); index >= 0; index = live.nextSetBit(index + 1)) {
+            long pageId = PageId.pack(index, 0);
+            long from = pages.addressOf(pageId);
+            if (!victims.contains(PageAddress.segmentOf(from))) {
+                continue;
+            }
+            if (count + 2 > moved.length) {
+                moved = Arrays.copyOf(moved, moved.length * 2);
+            }
+            moved[count++] = pageId;
+            moved[count++] = pages.copy(pageId);
+            bytes += (long) PageAddress.unitsOf(from) * PageId.UNIT_BYTES;
+            if (bytes >= PASS_BYTES) {
+                repoint(moved, count);
+                count = 0;
+                bytes = 0;
+            }
+        }
+        repoint(moved, count);
+    }
+
+    private void repoint(long[] moved, int count) {
+        if (count == 0) {
+            return;
+        }
+        pages.sync();
+        for (int i = 0; i < count; i += 2) {
+            pages.place(moved[i], moved[i + 1]);
         }
-        long retiredAt = generation;
-        victims.forEach(id -> pages.transition(id, SegmentState.RETIRED, retiredAt));
-        return new Report(live, victims, generation);
     }
 
     public List<Integer> reclaim() {
         OptionalLong pinned = transactions.oldestPinned();
-        long oldestRetained = transactions.history().stream().mapToLong(Generation::id).min().orElse(Long.MAX_VALUE);
+        long checkpoints = checkpointer.completed();
         List<Integer> reclaimable = pages.segments().stream()
                 .filter(segment -> segment.state() == SegmentState.RETIRED)
-                .filter(segment -> pinned.isEmpty() || pinned.getAsLong() >= segment.retiredAt())
-                .filter(segment -> oldestRetained >= segment.retiredAt())
+                .filter(segment -> pinned.isEmpty() || pinned.getAsLong() > segment.retiredAt())
+                .filter(segment -> checkpoints > retiredBeforeCheckpoint.getOrDefault(segment.id(), -1L))
                 .map(SegmentInfo::id)
                 .toList();
-        reclaimable.forEach(pages::delete);
+        reclaimable.forEach(id -> {
+            pages.delete(id);
+            retiredBeforeCheckpoint.remove(id);
+        });
         return reclaimable;
     }
 }
diff --git a/engine/src/main/java/io/hstore/engine/page/PageStore.java b/engine/src/main/java/io/hstore/engine/page/PageStore.java
index bc96948..ebf4103 100644
--- a/engine/src/main/java/io/hstore/engine/page/PageStore.java
+++ b/engine/src/main/java/io/hstore/engine/page/PageStore.java
@@ -56,6 +56,9 @@ private void restore(Collection<SegmentInfo> known) {
                     files.put(id, segment);
                     long units = segment.unitsOnDisk();
                     SegmentInfo info = byId.getOrDefault(id, new SegmentInfo(id, SegmentState.SEALED, units, units, 0));
+                    if (info.state() == SegmentState.COMPACTING) {
+                        info = info.withState(SegmentState.SEALED, 0);
+                    }
                     segments.put(id, info.withAllocation(info.pages(), Math.max(info.units(), units)));
                 }
             }
diff --git a/engine/src/main/java/io/hstore/engine/tree/TreeWalker.java b/engine/src/main/java/io/hstore/engine/tree/TreeWalker.java
index 67233f5..8e2ac4a 100644
--- a/engine/src/main/java/io/hstore/engine/tree/TreeWalker.java
+++ b/engine/src/main/java/io/hstore/engine/tree/TreeWalker.java
@@ -1,6 +1,5 @@
 package io.hstore.engine.tree;
 
-import java.util.Map;
 import java.util.function.LongPredicate;
 
 public final class TreeWalker {
@@ -13,12 +12,12 @@ public TreeWalker(NodeSource source) {
         this.source = source;
     }
 
-    public void visit(Ref root, TreeSchema<?> schema, Map<Long, Integer> visited) {
-        visit(root, schema, UNKNOWN_HEIGHT, visited);
+    public void visit(Ref root, TreeSchema<?> schema, LongPredicate firstVisit) {
+        visit(root, schema, UNKNOWN_HEIGHT, firstVisit);
     }
 
-    private void visit(Ref ref, TreeSchema<?> schema, int height, Map<Long, Integer> visited) {
-        if (!(ref instanceof Ref.Stored(long pageId, int units, Summary _)) || visited.putIfAbsent(pageId, units) != null) {
+    private void visit(Ref ref, TreeSchema<?> schema, int height, LongPredicate firstVisit) {
+        if (!(ref instanceof Ref.Stored(long pageId, int _, Summary _)) || !firstVisit.test(pageId)) {
             return;
         }
         if (height == 0 && !schema.codec().holdsRefs()) {
@@ -27,65 +26,23 @@ private void visit(Ref ref, TreeSchema<?> schema, int height, Map<Long, Integer>
         switch (source.load(pageId, schema)) {
             case Leaf leaf -> {
                 for (int i = 0; i < leaf.size(); i++) {
-                    mapNested(schema, leaf.value(i), (nested, nestedSchema) -> {
-                        visit(nested, nestedSchema, UNKNOWN_HEIGHT, visited);
-                        return nested;
-                    });
+                    visitNested(schema, leaf.value(i), firstVisit);
                 }
             }
             case Branch branch -> {
                 for (int i = 0; i < branch.size(); i++) {
-                    visit(branch.child(i), schema, branch.height() - 1, visited);
+                    visit(branch.child(i), schema, branch.height() - 1, firstVisit);
                 }
             }
         }
     }
 
-    public Ref relocate(Ref root, TreeSchema<?> schema, LongPredicate moving, WriteScope scope) {
-        return relocate(root, schema, UNKNOWN_HEIGHT, moving, scope);
-    }
-
-    private Ref relocate(Ref ref, TreeSchema<?> schema, int height, LongPredicate inVictim, WriteScope scope) {
-        if (!(ref instanceof Ref.Stored(long pageId, int _, Summary _))) {
-            return ref;
-        }
-        boolean moving = inVictim.test(pageId);
-        if (height == 0 && !moving && !schema.codec().holdsRefs()) {
-            return ref;
-        }
-        return switch (source.load(pageId, schema)) {
-            case Leaf leaf -> {
-                Object[] values = leaf.valueList().toArray();
-                boolean changed = moving;
-                for (int i = 0; i < values.length; i++) {
-                    Object mapped = mapNested(schema, values[i], (nested, nestedSchema) ->
-                            relocate(nested, nestedSchema, UNKNOWN_HEIGHT, inVictim, scope));
-                    changed |= mapped != values[i];
-                    values[i] = mapped;
-                }
-                yield changed ? new Ref.Pending(Leaf.owned(schema, scope.token(), leaf.keyArray(), values, values.length)) : ref;
-            }
-            case Branch branch -> {
-                Ref[] children = branch.childArray();
-                boolean changed = moving;
-                for (int i = 0; i < children.length; i++) {
-                    Ref moved = relocate(children[i], schema, branch.height() - 1, inVictim, scope);
-                    changed |= moved != children[i];
-                    children[i] = moved;
-                }
-                yield changed
-                        ? new Ref.Pending(Branch.owned(schema, scope.token(), branch.separatorArray(), children, children.length, branch.height()))
-                        : ref;
-            }
-        };
-    }
-
-    private static <V> Object mapNested(TreeSchema<V> schema, Object value, ValueCodec.RefMapper mapper) {
-        if (!schema.codec().holdsRefs()) {
-            return value;
+    private <V> void visitNested(TreeSchema<V> schema, Object value, LongPredicate firstVisit) {
+        if (schema.codec().holdsRefs()) {
+            schema.codec().mapRefs(schema.cast(value), (nested, nestedSchema) -> {
+                visit(nested, nestedSchema, UNKNOWN_HEIGHT, firstVisit);
+                return nested;
+            });
         }
-        V typed = schema.cast(value);
-        V mapped = schema.codec().mapRefs(typed, mapper);
-        return mapped.equals(typed) ? value : mapped;
     }
 }
diff --git a/engine/src/main/java/io/hstore/engine/txn/CrashPoint.java b/engine/src/main/java/io/hstore/engine/txn/CrashPoint.java
index f6e4b18..56da57a 100644
--- a/engine/src/main/java/io/hstore/engine/txn/CrashPoint.java
+++ b/engine/src/main/java/io/hstore/engine/txn/CrashPoint.java
@@ -7,7 +7,9 @@ public enum CrashPoint {
     COMMIT_APPEND,
     DATA_SYNC,
     WAL_SYNC,
-    CATALOG_PUBLISH;
+    CATALOG_PUBLISH,
+    COMPACTION_MOVE,
+    COMPACTION_RETIRE;
 
     @FunctionalInterface
     public interface Injector {
diff --git a/engine/src/main/java/io/hstore/engine/txn/Transaction.java b/engine/src/main/java/io/hstore/engine/txn/Transaction.java
index ce0bfb2..1ea0b30 100644
--- a/engine/src/main/java/io/hstore/engine/txn/Transaction.java
+++ b/engine/src/main/java/io/hstore/engine/txn/Transaction.java
@@ -47,6 +47,7 @@ private record KeyRead(Slot<?> slot, long key) {
 
     private final TransactionManager manager;
     private final long id;
+    private final long compactionsAtStart;
     private final Generation snapshot;
     private final Branch branch;
     private final TxnOptions options;
@@ -62,6 +63,7 @@ private record KeyRead(Slot<?> slot, long key) {
     Transaction(TransactionManager manager, long id, Generation snapshot, TxnOptions options) {
         this.manager = manager;
         this.id = id;
+        this.compactionsAtStart = manager.compactions();
         this.snapshot = snapshot;
         this.branch = snapshot.branch(options.branch());
         this.options = options;
@@ -259,6 +261,10 @@ private void requireActive() {
         }
     }
 
+    long compactionsAtStart() {
+        return compactionsAtStart;
+    }
+
     boolean carriesStoredRoots() {
         return ops.stream().anyMatch(op -> op instanceof Op.EdgeOp(long _, EdgeAction.Load _));
     }
diff --git a/engine/src/main/java/io/hstore/engine/txn/TransactionManager.java b/engine/src/main/java/io/hstore/engine/txn/TransactionManager.java
index 16c39a5..8e9dae6 100644
--- a/engine/src/main/java/io/hstore/engine/txn/TransactionManager.java
+++ b/engine/src/main/java/io/hstore/engine/txn/TransactionManager.java
@@ -42,7 +42,6 @@
 import java.util.concurrent.locks.ReentrantLock;
 import java.util.function.Function;
 import java.util.function.Supplier;
-import java.util.function.UnaryOperator;
 import java.util.stream.Stream;
 
 public final class TransactionManager {
@@ -80,7 +79,7 @@ private record PageWrite(long pageId, MemorySegment image) {
     private final LongAdder readOnly = new LongAdder();
     private final LongAdder pagesWritten = new LongAdder();
     private final LongAdder walBytes = new LongAdder();
-    private volatile long relocationFence;
+    private final AtomicLong compactions = new AtomicLong();
 
     public TransactionManager(Storage storage, Generation recovered, Collection<Generation> retained) {
         this.storage = storage;
@@ -197,7 +196,7 @@ CommitResult commit(Transaction txn) {
                     || Ref.same(txn.baseRoots().get(EngineSlots.REQUESTS), branch.roots().get(EngineSlots.REQUESTS));
             boolean serializable = txn.options().isolation() == Isolation.SERIALIZABLE;
             fast = branch.roots().sameAs(txn.baseRoots()) || (requestsStable && txn.workspace().untouchedSince(branch.roots(), serializable));
-            if (!fast && txn.generation() < relocationFence && txn.carriesStoredRoots()) {
+            if (txn.compactionsAtStart() < compactions.get() && txn.carriesStoredRoots()) {
                 conflicts.increment();
                 throw HStoreException.conflict("bulk-loaded roots of transaction " + txn.id() + " predate a compaction");
             }
@@ -305,22 +304,12 @@ private static WalRecord meta(long txn, Branch branch) {
         return new WalRecord.BranchMeta(txn, branch.id(), branch.name(), branch.parent(), branch.baseGeneration(), branch.createdAt(), branch.state());
     }
 
-    public long rewrite(int branchId, UnaryOperator<RootVector> rewrite) {
-        Appended appended;
-        commitLock.lock();
-        try {
-            Generation latest = head.get();
-            Branch branch = latest.branch(branchId);
-            RootVector roots = rewrite.apply(branch.roots());
-            if (roots.sameAs(branch.roots())) {
-                return latest.id();
-            }
-            appended = append(latest, branch.withRoots(roots), txns.getAndIncrement(), List.of(), List.of(), List.of(), Long.MAX_VALUE);
-            relocationFence = appended.pending().generation().id();
-        } finally {
-            commitLock.unlock();
-        }
-        return committer.await(appended.pending()).id();
+    public void compacted() {
+        compactions.incrementAndGet();
+    }
+
+    long compactions() {
+        return compactions.get();
     }
 
     public <T> T exclusive(Supplier<T> action) {
diff --git a/engine/src/test/java/io/hstore/engine/CrashRecoveryTest.java b/engine/src/test/java/io/hstore/engine/CrashRecoveryTest.java
index a34fe4e..61cd5bf 100644
--- a/engine/src/test/java/io/hstore/engine/CrashRecoveryTest.java
+++ b/engine/src/test/java/io/hstore/engine/CrashRecoveryTest.java
@@ -18,6 +18,7 @@
 import java.nio.file.Path;
 import java.nio.file.StandardOpenOption;
 import java.util.Comparator;
+import java.util.EnumSet;
 import java.util.List;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
@@ -35,7 +36,7 @@ class CrashRecoveryTest {
     Path directory;
 
     static Stream<Arguments> crashMatrix() {
-        return Stream.of(WalMode.values()).flatMap(mode -> Stream.of(CrashPoint.values()).map(point -> Arguments.of(mode, point)));
+        return Stream.of(WalMode.values()).flatMap(mode -> EnumSet.range(CrashPoint.PAGE, CrashPoint.CATALOG_PUBLISH).stream().map(point -> Arguments.of(mode, point)));
     }
 
     @ParameterizedTest

From 8820f26ef160ec265d36876e06ed537ceaf2cc4e Mon Sep 17 00:00:00 2001
From: venkat1701 <thejeastdev@gmail.com>
Date: Sat, 10 Oct 2026 19:11:59 +0530
Subject: [PATCH 3/6] test(engine): cover compaction by moving pages

CompactionTest checks that compaction commits nothing and deletes its
victims, that a generation still in history reads back after the segments
holding its pages are gone, that a reader from before the move keeps the
old segment until it closes, that commits run during compaction without
conflicts, and that a bulk load begun before a compaction is refused and
then succeeds on retry. A crash after the moves or after retirement, with
or without losing the directory writes since the last checkpoint, loses
nothing. Each test reopens the engine before checking, so the reads come
from disk rather than the node cache.
---
 .../java/io/hstore/engine/CompactionTest.java | 252 ++++++++++++++++++
 1 file changed, 252 insertions(+)
 create mode 100644 engine/src/test/java/io/hstore/engine/CompactionTest.java

diff --git a/engine/src/test/java/io/hstore/engine/CompactionTest.java b/engine/src/test/java/io/hstore/engine/CompactionTest.java
new file mode 100644
index 0000000..4e16727
--- /dev/null
+++ b/engine/src/test/java/io/hstore/engine/CompactionTest.java
@@ -0,0 +1,252 @@
+package io.hstore.engine;
+
+import io.hstore.engine.maintenance.Compactor;
+import io.hstore.engine.page.SegmentInfo;
+import io.hstore.engine.page.SegmentState;
+import io.hstore.engine.topology.EdgeKind;
+import io.hstore.engine.topology.Incidence;
+import io.hstore.engine.txn.CrashPoint;
+import io.hstore.engine.txn.Snapshot;
+import io.hstore.engine.txn.Transaction;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class CompactionTest {
+
+    private static final int ROUNDS = 60;
+
+    @TempDir
+    Path directory;
+
+    private static EngineOptions options() {
+        return EngineTest.small().withPagesPerSegment(64).withHistoryLimit(2).withCheckpointWalBytes(Long.MAX_VALUE);
+    }
+
+    @Test
+    void compactionMovesPagesWithoutWritingAGeneration() {
+        try (StorageEngine engine = StorageEngine.open(directory, options())) {
+            long edge = churn(engine);
+            long generation = engine.stats().generation();
+            Set<Integer> before = segmentIds(engine);
+            Compactor.Report report = engine.compact();
+            assertFalse(report.compacted().isEmpty());
+            assertEquals(generation, engine.stats().generation(), "compaction must not commit");
+            Set<Integer> after = segmentIds(engine);
+            report.compacted().forEach(victim -> assertFalse(after.contains(victim), "segment " + victim + " still exists"));
+            assertTrue(after.size() < before.size(), "segments " + before + " -> " + after);
+            assertEquals(expected(), members(engine, edge));
+        }
+        try (StorageEngine engine = StorageEngine.open(directory, options())) {
+            assertTrue(engine.liveness().keySet().stream().allMatch(segmentIds(engine)::contains));
+            assertEquals(expected(), members(engine, onlyEdge(engine)));
+        }
+    }
+
+    @Test
+    void retainedHistoryStaysReadableOnceItsSegmentsAreDeleted() {
+        EngineOptions options = options().withHistoryLimit(20);
+        List<Long> written = Stream.iterate(500_000L, i -> i + 1).limit(2_000).toList();
+        long large;
+        long edge;
+        long earlier;
+        try (StorageEngine engine = StorageEngine.open(directory, options)) {
+            large = engine.write(txn -> {
+                long created = txn.createEdge(EngineTest.EDGE, EdgeKind.SET);
+                written.forEach(member -> txn.insert(created, member));
+                return created;
+            });
+            Set<Integer> early = segmentIds(engine);
+            edge = churn(engine);
+            earlier = engine.stats().generation() - 10;
+            Compactor.Report report = engine.compact();
+            assertTrue(report.compacted().stream().anyMatch(early::contains), "no early segment was compacted: " + report.compacted());
+            report.compacted().forEach(victim -> assertFalse(segmentIds(engine).contains(victim)));
+        }
+        try (StorageEngine engine = StorageEngine.open(directory, options);
+             Snapshot old = engine.snapshotAt(earlier, 0)) {
+            assertEquals(written, old.requireEdge(large).stream().map(Incidence::member).toList());
+            assertEquals(afterRound(ROUNDS - 11), old.requireEdge(edge).stream().map(Incidence::member).toList());
+        }
+    }
+
+    @Test
+    void aReaderFromBeforeTheMoveKeepsTheOldSegmentAround() {
+        try (StorageEngine engine = StorageEngine.open(directory, options())) {
+            long edge = churn(engine);
+            Compactor.Report report;
+            try (Snapshot reader = engine.snapshot()) {
+                report = engine.compact();
+                assertFalse(report.compacted().isEmpty());
+                for (int victim : report.compacted()) {
+                    SegmentInfo info = engine.stats().segments().stream().filter(segment -> segment.id() == victim).findFirst().orElseThrow();
+                    assertEquals(SegmentState.RETIRED, info.state());
+                }
+                assertEquals(expected(), reader.requireEdge(edge).stream().map(Incidence::member).toList());
+            }
+            engine.compact();
+            report.compacted().forEach(victim -> assertFalse(segmentIds(engine).contains(victim)));
+        }
+        try (StorageEngine engine = StorageEngine.open(directory, options())) {
+            assertEquals(expected(), members(engine, onlyEdge(engine)));
+        }
+    }
+
+    @Test
+    void commitsCarryOnWhilePagesMove() throws Exception {
+        List<Long> all;
+        try (StorageEngine engine = StorageEngine.open(directory, options())) {
+            long edge = churn(engine);
+            long conflicts = engine.stats().conflicts();
+            AtomicBoolean compacting = new AtomicBoolean(true);
+            List<Long> written = new ArrayList<>();
+            try (ExecutorService writer = Executors.newSingleThreadExecutor()) {
+                Future<?> writes = writer.submit(() -> {
+                    long next = 1_000_000;
+                    while (compacting.get() || written.size() < 50) {
+                        long member = next++;
+                        engine.write(txn -> {
+                            txn.insert(edge, member);
+                            return null;
+                        });
+                        written.add(member);
+                    }
+                    return null;
+                });
+                for (int i = 0; i < 3; i++) {
+                    engine.compact();
+                }
+                compacting.set(false);
+                writes.get();
+            }
+            assertEquals(conflicts, engine.stats().conflicts());
+            TreeSet<Long> expected = new TreeSet<>(expected());
+            expected.addAll(written);
+            all = List.copyOf(expected);
+            assertEquals(all, members(engine, edge));
+        }
+        try (StorageEngine engine = StorageEngine.open(directory, options())) {
+            assertEquals(all, members(engine, onlyEdge(engine)));
+        }
+    }
+
+    @Test
+    void aBulkLoadStartedBeforeCompactionHasToRetry() {
+        try (StorageEngine engine = StorageEngine.open(directory, options())) {
+            long edge = churn(engine);
+            long target = engine.write(txn -> txn.createEdge(EngineTest.EDGE, EdgeKind.SET));
+            List<Incidence> members = Stream.iterate(2_000_000L, i -> i + 1).limit(5_000).map(Incidence::of).toList();
+            Transaction early = engine.begin();
+            early.load(target, members);
+            assertFalse(engine.compact().compacted().isEmpty());
+            assertThrows(HStoreException.Conflict.class, early::commit);
+            engine.write(txn -> {
+                txn.load(target, members);
+                return null;
+            });
+            assertEquals(5_000L, (long) engine.read(snapshot -> snapshot.requireEdge(target).size()));
+            assertEquals(expected(), members(engine, edge));
+        }
+    }
+
+    static Stream<Arguments> crashes() {
+        return Stream.of(CrashPoint.COMPACTION_MOVE, CrashPoint.COMPACTION_RETIRE)
+                .flatMap(point -> Stream.of(Arguments.of(point, false), Arguments.of(point, true)));
+    }
+
+    @ParameterizedTest
+    @MethodSource("crashes")
+    void aCrashDuringCompactionLosesNothing(CrashPoint point, boolean directoryWritesLost) throws Exception {
+        AtomicBoolean armed = new AtomicBoolean(false);
+        EngineOptions options = options().withFaults(reached -> {
+            if (reached == point && armed.compareAndSet(true, false)) {
+                throw new CrashPoint.SimulatedCrash(point);
+            }
+        });
+        Path addresses = directory.resolve("data").resolve("segments").resolve("pages.dir");
+        long edge;
+        try (StorageEngine engine = StorageEngine.open(directory, options)) {
+            edge = churn(engine);
+        }
+        byte[] checkpointed = Files.readAllBytes(addresses);
+        StorageEngine engine = StorageEngine.open(directory, options);
+        armed.set(true);
+        assertThrows(CrashPoint.SimulatedCrash.class, engine::compact);
+        engine.halt();
+        if (directoryWritesLost) {
+            Files.write(addresses, checkpointed);
+        }
+        try (StorageEngine recovered = StorageEngine.open(directory, options())) {
+            assertEquals(expected(), members(recovered, edge));
+            recovered.compact();
+            assertEquals(expected(), members(recovered, edge));
+        }
+        try (StorageEngine reopened = StorageEngine.open(directory, options())) {
+            assertEquals(expected(), members(reopened, edge));
+        }
+    }
+
+    private static long churn(StorageEngine engine) {
+        long edge = engine.write(txn -> txn.createEdge(EngineTest.EDGE, EdgeKind.SET));
+        churn(engine, edge);
+        return edge;
+    }
+
+    private static void churn(StorageEngine engine, long edge) {
+        for (int round = 0; round < ROUNDS; round++) {
+            int base = round * 50;
+            engine.write(txn -> {
+                for (int i = 0; i < 50; i++) {
+                    txn.insert(edge, 10_000 + base + i);
+                }
+                if (base >= 100) {
+                    for (int i = 0; i < 50; i++) {
+                        txn.remove(edge, 10_000 + base - 100 + i);
+                    }
+                }
+                return null;
+            });
+        }
+    }
+
+    private static List<Long> expected() {
+        return afterRound(ROUNDS - 1);
+    }
+
+    private static List<Long> afterRound(int round) {
+        return Stream.iterate(10_000L + (round - 1) * 50L, i -> i + 1).limit(100).toList();
+    }
+
+    private static List<Long> members(StorageEngine engine, long edge) {
+        return engine.read(snapshot -> snapshot.requireEdge(edge).stream().map(Incidence::member).toList());
+    }
+
+    private static long onlyEdge(StorageEngine engine) {
+        return engine.read(snapshot -> snapshot.atomsOfType(0, EngineTest.EDGE).findFirst().orElseThrow());
+    }
+
+    private static Set<Integer> segmentIds(StorageEngine engine) {
+        return engine.stats().segments().stream().map(SegmentInfo::id).collect(Collectors.toSet());
+    }
+}

From 13a858a16ab411c465344746cc49f9c05dc4afd7 Mon Sep 17 00:00:00 2001
From: venkat1701 <thejeastdev@gmail.com>
Date: Sat, 10 Oct 2026 19:12:00 +0530
Subject: [PATCH 4/6] docs: describe compaction by moving pages

maintenance.md walks through the cut, the walk, the move and the rules
for deleting a segment, and why moves need no WAL records. The
transaction docs describe the compaction fence. Other pages drop the
word relocation where compaction no longer rewrites anything.
---
 docs/architecture/overview.md           |  2 +-
 docs/benchmarks.md                      |  2 +-
 docs/operations/configuration.md        |  2 +-
 docs/operations/logging.md              |  4 +-
 docs/storage/catalog-and-generations.md |  4 +-
 docs/storage/change-feed.md             |  2 +-
 docs/storage/maintenance.md             | 60 +++++++++++++------------
 docs/storage/pages.md                   |  6 +--
 docs/storage/persistent-tree.md         |  2 +-
 docs/storage/topology.md                |  4 +-
 docs/transactions/transactions.md       | 11 +++--
 docs/transactions/wal-and-recovery.md   |  2 +-
 12 files changed, 52 insertions(+), 49 deletions(-)

diff --git a/docs/architecture/overview.md b/docs/architecture/overview.md
index 53d6d31..bbe0f5d 100644
--- a/docs/architecture/overview.md
+++ b/docs/architecture/overview.md
@@ -91,7 +91,7 @@ The engine has no dependencies outside `java.base`. All three modules build into
 | `engine.txn` | `TransactionManager` (commit lock, append, publication, pins, history), `Transaction` (op log, read sets, rebase), `Snapshot`, `View` (read API shared by both), `Workspace` (mutable root vector plus derivations), `GroupCommitter`, `Op`/`EdgeAction`, `Derivation`/`EngineDerivations`, `CrashPoint`, `Durability`, `WalMode`, `Isolation` and `TxnOptions`. |
 | `engine.wal` | `WriteAheadLog` (segmented, checksummed records) and `WalRecord` (`Begin`, `Page`, `PageRef`, `Root`, `BranchMeta`, `Feed`, `Commit`, `Abort`, `Checkpoint`). |
 | `engine.feed` | `ChangeFeed` (segmented append-only `feed/*.feed` files, sparse generation index, retention, tailing subscriptions), `CommitEvent` and `FeedCodec`; see [change-feed.md](../storage/change-feed.md). |
-| `engine.maintenance` | `Checkpointer`, `Compactor` (liveness, relocation, reclamation) and `Recovery` (catalog discovery, WAL redo, durable-prefix rule). |
+| `engine.maintenance` | `Checkpointer`, `Compactor` (liveness, moving live pages out of victim segments, reclamation) and `Recovery` (catalog discovery, WAL redo, durable-prefix rule). |
 
 ### `io.hstore.db`
 
diff --git a/docs/benchmarks.md b/docs/benchmarks.md
index 87f3e80..f379651 100644
--- a/docs/benchmarks.md
+++ b/docs/benchmarks.md
@@ -229,7 +229,7 @@ not what drives HStore's footprint:
 * **Disk at scale 4 (0.54×).** Live data is comparable: about 1 GB of live node images in HStore against
   1.27 GB in JE. The difference is garbage that has not been reclaimed yet. Both engines are log-structured
   with a 50% minimum-utilisation cleaner: HStore's `compaction_live_ratio = 0.5`, JE's
-  `je.cleaner.minUtilization = 50`. A sealed HStore segment is only relocated once fewer than half of its
+  `je.cleaner.minUtilization = 50`. A sealed HStore segment is only compacted once fewer than half of its
   images are live, and the background pass runs after every `checkpoint_wal_mb` of writes
   ([maintenance.md](storage/maintenance.md)). Raising `compaction_live_ratio` trades write amplification for
   space.
diff --git a/docs/operations/configuration.md b/docs/operations/configuration.md
index 4a07679..72bc767 100644
--- a/docs/operations/configuration.md
+++ b/docs/operations/configuration.md
@@ -94,7 +94,7 @@ max_connections          = 200
 | `cache_nodes` | `HSTORE_CACHE_NODES` | `65536` | Decoded tree nodes kept in the node cache. |
 | `checkpoint_wal_mb` | `HSTORE_CHECKPOINT_WAL_MB` | `256` | WAL volume since the last checkpoint that triggers a background checkpoint. |
 | `history_limit` | `HSTORE_HISTORY_LIMIT` | `64` | Committed generations kept addressable for `AT GENERATION`, `AS OF`, `HISTORY` and the Studio time slider. |
-| `compaction_live_ratio` | `HSTORE_COMPACTION_LIVE_RATIO` | `0.5` | Sealed segments whose live bytes fall below this fraction of their allocated bytes are relocated by compaction: one segment (the emptiest) per background pass after every `checkpoint_wal_mb` of written node images, or all of them with `COMPACT`. Time-travel history is preserved; retired segments are deleted once `history_limit` has rolled past them. Must satisfy 0 < r < 1; other values fail at startup. |
+| `compaction_live_ratio` | `HSTORE_COMPACTION_LIVE_RATIO` | `0.5` | Sealed segments whose live bytes fall below this fraction of their allocated bytes are emptied by compaction, which moves their live pages to the active segment: one segment (the emptiest) per background pass after every `checkpoint_wal_mb` of written node images, or all of them with `COMPACT`. Time-travel history is preserved. Emptied segments are deleted after the next checkpoint, once no reader that started before the move is still open. Must satisfy 0 < r < 1; other values fail at startup. |
 | `feed_retention_generations` | `HSTORE_FEED_RETENTION_GENERATIONS` | `100000` | Committed generations kept in the [change feed](../storage/change-feed.md) for `HISTORY`, `DIFF` and subscribers. After every checkpoint, whole feed segments that lie entirely below `current − N` are deleted. *Holds* lower that floor: the semantic index's last persisted generation and the generation of each continuous materialised view. Feed segments roll at the WAL segment size (`EngineOptions.walSegmentBytes`, 64 MiB). Must be ≥ 1. |
 
 ### Resource budgets
diff --git a/docs/operations/logging.md b/docs/operations/logging.md
index 1f74d2c..3c01209 100644
--- a/docs/operations/logging.md
+++ b/docs/operations/logging.md
@@ -90,8 +90,8 @@ This is every message the server emits. Placeholders are in `{braces}`.
 
 | Severity | Component | Message |
 |---|---|---|
-| LOG | engine | `background compaction relocated segments {list}` (write-volume-triggered pass) |
-| LOG | engine | `compaction relocated segments {list} and reclaimed {list}` |
+| LOG | engine | `background compaction moved the live pages out of segments {list}` (write-volume-triggered pass) |
+| LOG | engine | `compaction moved the live pages out of segments {list} and reclaimed {list}` |
 | LOG | engine | `reclaimed retired segments {list}` |
 | WARNING | hstore | `background maintenance failed` (with DETAIL) |
 | LOG | txn | `created branch {name} (#{id}) from generation {g}` |
diff --git a/docs/storage/catalog-and-generations.md b/docs/storage/catalog-and-generations.md
index 4d5b465..7fbc7d6 100644
--- a/docs/storage/catalog-and-generations.md
+++ b/docs/storage/catalog-and-generations.md
@@ -100,7 +100,7 @@ Published generations are kept in `TransactionManager.history` (a `ConcurrentSki
 * `snapshotAt(generation, branch)` pins the generation and opens a snapshot over its root vector. A generation that has been trimmed fails with `generation N is no longer retained`.
 * `generationAsOf(wallTime)` returns the newest retained generation with `wallTime <= t`.
 
-Time travel needs no undo log. Copy-on-write never overwrites a page, so an old generation's root vector still points at valid pages, and reading the past is exactly as fast as reading the present. A historical page is only reclaimed once compaction has relocated the live data *and* both history trimming (`history_limit`) and every pinned reader have moved past the compaction generation ([maintenance.md](maintenance.md#reclaim)). Compaction therefore never shortens the time-travel window.
+Time travel needs no undo log. Copy-on-write never overwrites a page, so an old generation's root vector still points at valid pages, and reading the past is exactly as fast as reading the present. Compaction moves the pages a retained generation still needs and keeps their page numbers, so old root vectors stay valid after their segments are deleted ([maintenance.md](maintenance.md#reclaim)). Compaction therefore never shortens the time-travel window.
 
 The retained history is also persisted. `Checkpointer` writes up to `historyLimit` past generations into the catalog image, and `Recovery` appends every replayed generation, so time travel survives restarts.
 
@@ -124,7 +124,7 @@ Branches are named, independently writable lines of generations that share pages
 | merge | `markMerged(id)` → `closeBranch(id, MERGED)` | Sets the state to `MERGED` **and empties its roots**, so its pages are no longer reachable from the current generation. The actual three-way merge of contents happens in `db.temporal.BranchMerge` before this call. |
 | drop | `dropBranch(id)` → `closeBranch(id, DROPPED)` | The same, with state `DROPPED`. `Generation.branch` rejects a dropped id. |
 
-`main` (id 0) cannot be closed. `StorageEngine.branches()` lists only `ACTIVE` branches. Because closed branches carry empty root vectors, compaction only has to relocate the active ones.
+`main` (id 0) cannot be closed. `StorageEngine.branches()` lists only `ACTIVE` branches. Because closed branches carry empty root vectors, they keep no pages alive.
 
 ```sql
 CREATE BRANCH what_if;
diff --git a/docs/storage/change-feed.md b/docs/storage/change-feed.md
index 0cb2ac2..a956387 100644
--- a/docs/storage/change-feed.md
+++ b/docs/storage/change-feed.md
@@ -124,7 +124,7 @@ for each flagged column, in that order:
 
 A column is flagged only when it differs from the default (`roleSet != NO_ROLES`, `weight != Weight.ONE`, `validFrom != Long.MIN_VALUE`, `validTo != Long.MAX_VALUE`, `dataRef != 0`, `qualifier != 0`), so a plain membership costs 1 byte of flags plus the member id.
 
-Slot values that contain tree references (codecs with `holdsRefs()`, for example an `EdgeRecord` with member and order roots) are **detached** before encoding: every embedded `Ref` is replaced with `Ref.EMPTY`. Feed events therefore never pin pages, and compaction may relocate or reclaim any page without invalidating the feed. Consumers that need the topology of an edge read it from a snapshot, or use the event's `members` list.
+Slot values that contain tree references (codecs with `holdsRefs()`, for example an `EdgeRecord` with member and order roots) are **detached** before encoding: every embedded `Ref` is replaced with `Ref.EMPTY`. Feed events therefore never pin pages, and compaction may move or reclaim any page without invalidating the feed. Consumers that need the topology of an edge read it from a snapshot, or use the event's `members` list.
 
 ### Example
 
diff --git a/docs/storage/maintenance.md b/docs/storage/maintenance.md
index 5c15ff5..acedc26 100644
--- a/docs/storage/maintenance.md
+++ b/docs/storage/maintenance.md
@@ -13,12 +13,12 @@ Sources are under `engine/src/main/java/io/hstore/engine/maintenance/`, plus `St
 | Trigger | Action | Code |
 |---|---|---|
 | Engine open, after recovery | checkpoint, then reclaim | `StorageEngine` constructor |
-| `hstore-maintenance` virtual thread, every 500 ms (skipped when the `maintenance` lock is busy) | checkpoint if `walSinceCheckpoint() > checkpointWalBytes`; a compaction pass when more than `checkpointWalBytes` of node images were written since the previous pass (followed by a checkpoint if anything was relocated); then reclaim | `StorageEngine.maintain` |
+| `hstore-maintenance` virtual thread, every 500 ms (skipped when the `maintenance` lock is busy) | checkpoint if `walSinceCheckpoint() > checkpointWalBytes`; a compaction pass when more than `checkpointWalBytes` of node images were written since the previous pass (followed by a checkpoint if anything was moved); then reclaim | `StorageEngine.maintain` |
 | `CHECKPOINT;` (HQL), `StorageEngine.checkpoint()` | checkpoint | `timedCheckpoint` |
 | `COMPACT;` (HQL), `StorageEngine.compact()` | compact, checkpoint, reclaim | `StorageEngine.compact` |
 | `StorageEngine.close()` | final checkpoint, then shutdown | `close` |
 
-Checkpoint and compaction are serialized by `StorageEngine.maintenance`, a `ReentrantLock`. Background compaction plays the role of PostgreSQL's autovacuum. Its trigger is write volume, `pages.bytesWritten()` advancing by more than `checkpoint_wal_mb` since the last pass, so an idle database never pays for a liveness walk. The background pass relocates at most one segment, the emptiest one whose live bytes are below `compaction_live_ratio` of its allocated bytes. `COMPACT;` relocates every such segment on demand.
+Checkpoint and compaction are serialized by `StorageEngine.maintenance`, a `ReentrantLock`. Background compaction plays the role of PostgreSQL's autovacuum. Its trigger is write volume, `pages.bytesWritten()` advancing by more than `checkpoint_wal_mb` since the last pass, so an idle database never pays for a liveness walk. The background pass empties at most one segment, the emptiest one whose live bytes are below `compaction_live_ratio` of its allocated bytes. `COMPACT;` empties every such segment on demand.
 
 Every checkpoint is logged, and then the checkpoint listeners run. The database registers `SemanticPlane.persist`, which saves the HNSW index alongside the catalog. A sample log line:
 
@@ -51,9 +51,14 @@ Each commit writes fresh node images for every path it changes and references al
 Segments are append-only extents ([pages.md](pages.md#segment-files)), so garbage is never reused piecemeal. It is reclaimed one whole segment at a time:
 
 1. measure how many images in each segment are still reachable;
-2. copy the reachable ones out of mostly-dead segments;
-3. publish roots that no longer reference those segments;
-4. delete the files once no reader can still see the old roots.
+2. copy the reachable ones out of mostly-dead segments, byte for byte;
+3. point their entries in the page directory at the copies;
+4. delete the files once a checkpoint has made the new entries durable and no reader can still be using an old
+   address.
+
+No tree is rewritten and no generation is committed. Pages keep their numbers when they move
+([pages.md](pages.md#page-numbers-and-the-page-directory)), so every node that points at a moved page, in every
+generation and branch, still points at it.
 
 ### Segment state machine
 
@@ -61,13 +66,13 @@ Segments are append-only extents ([pages.md](pages.md#segment-files)), so garbag
 stateDiagram-v2
     [*] --> ACTIVE: rollSegment()
     ACTIVE --> SEALED: next segment rolled (capacity reached)
-    SEALED --> COMPACTING: chosen as victim (live < pages × ratio)
-    COMPACTING --> SEALED: allowed transition (abandoned compaction)
-    COMPACTING --> RETIRED: roots rewritten, retiredAt = generation
-    RETIRED --> [*]: reclaim(): file deleted once oldestPinned ≥ retiredAt
+    SEALED --> COMPACTING: chosen as victim (live < bytes × ratio)
+    COMPACTING --> SEALED: reopened after a crash mid-compaction
+    COMPACTING --> RETIRED: live pages moved, retiredAt = current generation
+    RETIRED --> [*]: reclaim(): file deleted after a checkpoint, once oldestPinned > retiredAt
 ```
 
-`SegmentState.canTransitionTo` allows only the next state, plus `COMPACTING → SEALED`. `SegmentInfo.withState` throws on anything else. Only one segment is `ACTIVE` at a time: the one `PageStore.allocate` appends to. It is never chosen as a victim.
+`SegmentState.canTransitionTo` allows only the next state, plus `COMPACTING → SEALED`. `SegmentInfo.withState` throws on anything else. `PageStore.open` turns a `COMPACTING` segment back into `SEALED`, so a compaction cut short by a crash leaves the segment a candidate for the next pass. Only one segment is `ACTIVE` at a time: the one `PageStore.allocate` appends to. It is never chosen as a victim.
 
 ### Liveness
 
@@ -77,7 +82,7 @@ stateDiagram-v2
 * **every branch** in each of those generations;
 * **every slot root** in each branch.
 
-`TreeWalker.visit(root, schema, visited)` records each `Ref.Stored` page id it meets, with the image size the reference carries, in one shared `visited` map, so an image shared by several generations, branches or trees is counted once. The size comes from the reference itself, so no page has to be read to learn it. It descends into branches and, for codecs that hold references (catalog edge records, promoted posting lists), into nested trees.
+`TreeWalker.visit(root, schema, firstVisit)` calls `firstVisit` with each `Ref.Stored` page id it meets and stops descending when that returns false. The compactor sets one bit per page number in a `BitSet`, so an image shared by several generations, branches or trees is visited once, and the whole walk costs one bit per page number. Sizes come from the page directory afterwards: each set bit is looked up and its length added to its segment. The walker descends into branches and, for codecs that hold references (catalog edge records, promoted posting lists), into nested trees.
 
 A detail that keeps this cheap: at height 0 of a tree whose codec has no nested references, the walker records the leaf's id **without loading it**. The summaries in the parent already prove that the leaf exists. Liveness is therefore proportional to the number of internal nodes plus the reference-holding leaves, not to total data size.
 
@@ -87,36 +92,35 @@ Liveness is exposed as `StorageEngine.liveness()` (live bytes per segment). The
 
 `Compactor.compact(liveThreshold)` (default threshold `compaction_live_ratio = 0.5`):
 
-1. **Choose victims.** A victim is a `SEALED` segment, not the active one, with `live bytes < allocated bytes × threshold`, so the ratio is *live bytes / bytes written*. Candidates are ordered from emptiest to fullest. The background pass takes only the first one (`compact(threshold, 1)`), which bounds how long a single relocation holds the commit lock; `COMPACT` takes all of them. If there are no victims, it reports and returns.
-2. Mark the victims `COMPACTING`.
-3. **Relocate.** For every **active** branch of the current generation, call `transactions.rewrite(branch, roots -> roots.map(relocate))`. `TreeWalker.relocate(ref, schema, moving, scope)` rebuilds as a fresh `Ref.Pending` copy:
-   * every stored node whose page id lies in a victim segment;
-   * every ancestor of such a node;
-   * every leaf whose nested trees changed.
+1. **Take a cut.** Inside `TransactionManager.exclusive`, which holds the commit lock and drains the group committer, record the retained generations (history plus `current`) and the set of segments that are already `SEALED`. Only those segments can become victims. A commit writes its pages under the commit lock but publishes its generation a moment later, and the active segment can fill up and be sealed while the walk runs. Taking the cut with nothing in flight means every page the walk doesn't see sits in a segment that wasn't sealed yet.
+2. **Walk** the recorded generations as described under *Liveness*.
+3. **Choose victims.** A victim is a segment from the cut, still `SEALED`, with `live bytes < allocated bytes × threshold`, so the ratio is *live bytes / bytes written*. Candidates are ordered from emptiest to fullest. The background pass takes only the first one (`compact(threshold, 1)`); `COMPACT` takes all of them. If there are no victims, it reports and returns.
+4. Mark the victims `COMPACTING`.
+5. **Move.** For every set bit whose directory entry points into a victim, `PageStore.copy` reads the image, checks it with `PageHeader.verify`, and writes the same bytes to the active segment. The header holds the page number, not the address, so neither the header nor the checksum changes. Every 4 MiB of copies, `PageStore.sync()` forces the segments and only then are the directory entries pointed at the copies. A directory entry therefore never reaches the disk pointing at bytes that are not already there.
+6. Count the compaction in `TransactionManager.compacted()` (see below) and mark the victims `RETIRED` with `retiredAt` set to the current generation.
 
-   Subtrees that touch no victim are returned unchanged and stay shared. `rewrite` publishes the new root vector as a normal commit: it takes the commit lock, materializes the pending nodes into the active segment, appends the WAL records and goes through the group-commit barrier. It also sets `relocationFence` to the new generation id.
-4. Mark the victims `RETIRED` with `retiredAt = generation`. Retained history is **not** truncated. Generations older than `retiredAt` keep pointing at the old images in the victims, which stay on disk and readable, so `AT GENERATION` and `AS OF` keep working across a compaction. The victims are reclaimed once `history_limit` has rolled past `retiredAt`.
+`StorageEngine.compact()` then checkpoints, which forces the page directory and persists the segment states, and calls `reclaim()`.
 
-`StorageEngine.compact()` then checkpoints, which persists the new segment states and roots, and calls `reclaim()`.
+Nothing in this takes the commit lock except the short cut in step 1, and nothing commits. Commits carry on while pages move, readers keep using the node cache, whose entries are keyed by page number and stay valid, and retained history keeps working because old generations reach the moved pages through the same directory.
 
-Closed branches (`MERGED`, `DROPPED`) carry empty root vectors ([catalog-and-generations.md](catalog-and-generations.md#branches)), so they hold nothing alive and are not rewritten.
+Moves are not written to the WAL. If the engine crashes before the checkpoint that follows a compaction, recovery replays the commits since the previous checkpoint and writes their original addresses back into the directory. Those addresses are in the victims, which are only deleted after that checkpoint, so they still hold the same bytes. The copies made before the crash become unreferenced space in the active segment, and the victims are picked again by a later pass. `CompactionTest` crashes at `COMPACTION_MOVE` and `COMPACTION_RETIRE`, with and without losing the directory writes made since the last checkpoint, and checks the data after reopening and after another compaction.
 
 #### Interaction with running transactions
 
-A transaction that started before the compaction holds roots from its base generation. At commit, one of these applies:
+Compaction doesn't change any root, so transactions that started before it commit exactly as they would have without it, on the fast path or by rebasing.
 
-* **Fast path** (`Workspace.untouchedSince`). Every slot the transaction changed (and, for `SERIALIZABLE`, every slot it read) still has the same root in the latest generation. Compaction rewrote none of those slots, which means they contained no images in victim segments, so the transaction's versions of them cannot reference victims either. `rebasedOnto` combines them with the latest, relocated roots for every other slot.
-* **Rebase path.** The op log is replayed onto the latest (relocated) roots, so the result references only relocated images.
-* **Fence.** A transaction whose base generation predates `relocationFence` and that carries already-stored roots, such as spilled bulk-load subtrees (`Transaction.carriesStoredRoots`), cannot be safely rebased onto relocated data. It fails with a retryable conflict (`bulk-loaded roots of transaction N predate a compaction`), and `HypergraphDatabase.write` retries it.
+The one exception is a bulk load. `Transaction.load` can spill finished subtrees to disk before commit (`Transaction.carriesStoredRoots`), and those pages are not reachable from any generation until the transaction commits, so the walk can't see them. If one of them sits in a victim, it isn't moved. Each transaction therefore records `TransactionManager.compactions()` when it begins, and a transaction that carries stored roots and saw fewer compactions than have since finished fails with a retryable conflict (`bulk-loaded roots of transaction N predate a compaction`). `HypergraphDatabase.write` retries it. The check applies on the fast path too, since compaction no longer changes the roots that would otherwise push such a transaction off it.
 
 ### Reclaim
 
 `Compactor.reclaim()` deletes the file and catalog entry of every `RETIRED` segment, but only when both of these hold:
 
-* no generation is pinned, or `oldestPinned() ≥ retiredAt`; and
-* the oldest generation still retained in history is `≥ retiredAt`, so no time-travel target can reference the segment.
+* a checkpoint has completed since the segment was retired (`Checkpointer.completed()`), so the new directory entries are durable and the WAL no longer holds commits whose addresses point into it; and
+* no generation is pinned, or `oldestPinned() > retiredAt`.
+
+Retained history no longer holds a segment back. Old generations reach moved pages through the directory like everything else, so `AT GENERATION` and `AS OF` keep working after the victims are gone.
 
-**Why it must wait.** A `Snapshot` opened before compaction pinned a generation older than `retiredAt`. That generation's roots still point into the victim segments, and copy-on-write promises the reader those images stay valid until it closes. Deleting the file early would turn its next read into `segment does not exist`. Pins are reference counts per generation (`TransactionManager.pins`). Every `Snapshot.close()` and transaction end decrements them, and the maintenance thread retries `reclaim` every 500 ms. A retired segment therefore disappears within half a second of its last reader closing.
+**Why it must wait for readers.** A reader looks up a page's address and then reads the page. One that started before the move may have looked up an old address and not read it yet. Every reader and writer pins its generation, and anyone pinned at `retiredAt` or earlier may have started before the moves finished. Deleting the file early would turn that reader's next read into `segment does not exist`. Pins are reference counts per generation (`TransactionManager.pins`). Every `Snapshot.close()` and transaction end decrements them, and the maintenance thread retries `reclaim` every 500 ms. A retired segment therefore disappears within half a second of its last reader closing.
 
 ### History retention and pins
 
diff --git a/docs/storage/pages.md b/docs/storage/pages.md
index 03b0152..908e3b5 100644
--- a/docs/storage/pages.md
+++ b/docs/storage/pages.md
@@ -156,9 +156,9 @@ no image. The file is memory-mapped in 8 MiB chunks, and a new chunk is mapped w
 lookup is a load from the OS page cache and the directory takes no Java heap. A million pages need 8 MB of it.
 
 Nothing outside `PageStore` sees an address. Trees, nested references, `NodeCache`, `Ref` and the WAL all use
-numbers. That is what will let compaction move an image by changing one directory entry, without rewriting the nodes
-that point at it. It doesn't do that yet; it still rebuilds the paths to the pages it moves
-([maintenance.md](maintenance.md)).
+numbers. That is what lets compaction move an image by copying its bytes and changing one directory entry, without
+rewriting the nodes that point at it ([maintenance.md](maintenance.md#compaction)). Entries are written and read with
+release and acquire ordering, so a reader racing a move sees either the old address or the new one, never a mix.
 
 Page `#1234`, stored as 3 units at unit 1234 of segment 1:
 
diff --git a/docs/storage/persistent-tree.md b/docs/storage/persistent-tree.md
index 66d4ba2..fe74b5a 100644
--- a/docs/storage/persistent-tree.md
+++ b/docs/storage/persistent-tree.md
@@ -20,7 +20,7 @@ This document describes the node structure, the write path, the summaries and th
 | `TreeCursor` | `TreeCursor.java` | Bidirectional cursor with seek and summary-based pruning. |
 | `TreeAlgebra`, `TreeDiff` | `TreeAlgebra.java`, `TreeDiff.java` | Intersection (leapfrog, synchronized descent, probing), union, difference, subset, equality and structural diff. |
 | `Materializer` | `Materializer.java` | Freezes and encodes pending nodes into pages, post-order. |
-| `TreeWalker`, `TreeVerifier` | | Reachability (liveness), relocation (compaction) and invariant checking. |
+| `TreeWalker`, `TreeVerifier` | | Reachability (liveness for compaction) and invariant checking. |
 
 ## Node structure
 
diff --git a/docs/storage/topology.md b/docs/storage/topology.md
index 09224a3..6de2322 100644
--- a/docs/storage/topology.md
+++ b/docs/storage/topology.md
@@ -36,9 +36,9 @@ EdgeRecord: u8 tag=1 | varint type | u8 kind (0=SET, 1=ORDERED) | Ref members |
 Ref:        i64le pageId (0 = empty) | [Summary encoding, only when pageId != 0]
 ```
 
-Because `AtomCodec.holdsRefs()` is true, the materializer writes each edge's member and order trees *before* the catalog leaf that references them. `TreeWalker` and `TreeVerifier` descend into them, and compaction relocates them. The catalog therefore forms a two-level tree of trees: catalog leaves hold edge records, and each edge record roots its own membership tree.
+Because `AtomCodec.holdsRefs()` is true, the materializer writes each edge's member and order trees *before* the catalog leaf that references them. `TreeWalker` and `TreeVerifier` descend into them, and compaction moves their pages like any other. The catalog therefore forms a two-level tree of trees: catalog leaves hold edge records, and each edge record roots its own membership tree.
 
-A record's `contentHash()` feeds the catalog's SET fingerprint. The hash of an edge record is computed from `(1, type, kind, members.count, members.fingerprint, version, dataRef, tenant, isolation)`. It is a function of the edge's member *contents*, never of page ids, so relocating pages during compaction does not change catalog fingerprints.
+A record's `contentHash()` feeds the catalog's SET fingerprint. The hash of an edge record is computed from `(1, type, kind, members.count, members.fingerprint, version, dataRef, tenant, isolation)`. It is a function of the edge's member *contents*, never of page ids, and compaction never changes page ids anyway, so it leaves catalog fingerprints alone.
 
 ## Hyperedges
 
diff --git a/docs/transactions/transactions.md b/docs/transactions/transactions.md
index 230e07f..fa66a8e 100644
--- a/docs/transactions/transactions.md
+++ b/docs/transactions/transactions.md
@@ -118,7 +118,7 @@ commit(txn):
     if txn has requestId and REQUESTS[hash(id)] exists -> DUPLICATE (prior txn id, prior generation)
     fast = branch.roots sameAs txn.baseRoots
         || (requestsStable && workspace.untouchedSince(branch.roots, isolation == SERIALIZABLE))
-    if !fast and txn.generation < relocationFence and txn carries Load ops -> Conflict
+    if txn.compactionsAtStart < compactions and txn carries Load ops -> Conflict
     if fast: workspace = txn.workspace
     else:    txn.validateReads(branch.roots); workspace = txn.replay(branch.roots)
     if requestId: workspace.write(REQUESTS, hash(id), RequestRecord(id, txn.id, latest.id + 1))
@@ -155,9 +155,9 @@ A footprint is *commutative* iff the edge was a `SET` edge at the base and every
 
 `TreeDiff.diff` uses the frontier algorithm over shared subtrees ([persistent tree](../storage/persistent-tree.md)), so the cost of the overlap check is proportional to the changed region, not the edge cardinality.
 
-### 5.3 Relocation fence
+### 5.3 Compaction fence
 
-Compaction rewrites roots through `TransactionManager.rewrite`, which records `relocationFence = new generation id`. A transaction that bulk-loaded stored subtrees (`carriesStoredRoots()`: any `EdgeOp` with `EdgeAction.Load`) began before the fence and cannot take the fast path is rejected with a conflict, because its stored roots may point into segments that compaction is about to retire. Retrying rebuilds the subtree.
+Compaction moves pages without changing any root, so it doesn't affect other commits. A bulk load is different: a transaction that bulk-loaded stored subtrees (`carriesStoredRoots()`: any `EdgeOp` with `EdgeAction.Load`) may have spilled pages that no generation reaches yet, so compaction can't see them and won't move them. Each transaction records `TransactionManager.compactions()` when it begins, and one that carries stored roots is rejected with a conflict if a compaction has finished since, on the fast path as well as the slow one. Retrying rebuilds the subtree.
 
 ### 5.4 Page budget
 
@@ -206,7 +206,7 @@ Under `SNAPSHOT` both commits succeed (R replays `Insert(p)` into B, which L nev
 
 | Code | Class | Retryable | Raised by |
 |---|---|---|---|
-| `RETRYABLE_CONFLICT` | `Conflict` | yes | replay preconditions, `verifyEdge`, `validateReads`, relocation fence, concurrent atom id/canonical key |
+| `RETRYABLE_CONFLICT` | `Conflict` | yes | replay preconditions, `verifyEdge`, `validateReads`, compaction fence, concurrent atom id/canonical key |
 | `RETRYABLE_IO` | `TransientIo` | yes | WAL/page/feed I/O failures; a broken commit pipeline (`"the commit pipeline stopped after a durability failure; restart the engine"`) |
 | `ABORTED_RESOURCE_LIMIT` | `ResourceLimit` | no | page budget, quotas |
 | `INVALID_SCHEMA` | `InvalidSchema` | no | caller errors |
@@ -344,10 +344,9 @@ Had T2 also touched `m1`, or had `E` been an `ORDERED` edge, the overlap check w
 
 ## 10. Branch operations and internal writes
 
-`system(branchId, work)`, `createBranch`, `closeBranch` and `rewrite` reuse `append` and the group committer, so they are durable and ordered exactly like user commits:
+`system(branchId, work)`, `createBranch` and `closeBranch` reuse `append` and the group committer, so they are durable and ordered exactly like user commits:
 
 * `createBranch` appends a `BranchMeta` record and a branch whose roots are the parent's current roots (copy-on-write makes this O(number of slots)). Names must be unique among active branches; the new id is one past the largest existing id.
 * `closeBranch` (`MERGED` or `DROPPED`, never for `MAIN`) empties the root vector so the branch's pages become unreachable.
-* `rewrite` (compaction) replaces roots without member or slot changes and sets the relocation fence.
 
 These commits carry no member or slot changes, so their `CommitEvent` is empty and nothing is written to the change feed (see [change feed](../storage/change-feed.md)).
diff --git a/docs/transactions/wal-and-recovery.md b/docs/transactions/wal-and-recovery.md
index c095b0d..6e1fdb8 100644
--- a/docs/transactions/wal-and-recovery.md
+++ b/docs/transactions/wal-and-recovery.md
@@ -275,7 +275,7 @@ The checkpoint LSN is the position of the `Checkpoint` record itself, so the nex
 |---|---|
 | Every open, right after recovery | `StorageEngine` constructor |
 | WAL grew by more than `checkpointWalBytes` (default 256 MiB) since the last checkpoint; checked every 500 ms | `StorageEngine.maintain` |
-| Compaction, between relocation and segment reclamation | `StorageEngine.compact` |
+| Compaction, after moving pages and before deleting segments | `StorageEngine.compact` |
 | Clean shutdown | `StorageEngine.close` |
 | On demand | `StorageEngine.checkpoint()`, HQL `CHECKPOINT` |
 

From 2772d621d856fb684dc64840cd0a1b0ff88de595 Mon Sep 17 00:00:00 2001
From: venkat1701 <thejeastdev@gmail.com>
Date: Sun, 11 Oct 2026 13:22:17 +0530
Subject: [PATCH 5/6] fix(engine): count a compaction from its cut so a load
 can't commit mid-move

The compaction count only went up after the pages had moved. A bulk load that
began before the cut and committed while pages were moving passed the check,
and its spilled pages, which the walk couldn't see, stayed in a victim that was
then retired and deleted. The count now goes up when the cut is taken, under
the commit lock, so that load fails with a retryable conflict.
---
 docs/storage/maintenance.md                   |  2 +-
 .../hstore/engine/maintenance/Compactor.java  |  2 +-
 .../java/io/hstore/engine/CompactionTest.java | 33 +++++++++++++++++++
 3 files changed, 35 insertions(+), 2 deletions(-)

diff --git a/docs/storage/maintenance.md b/docs/storage/maintenance.md
index acedc26..2b47e12 100644
--- a/docs/storage/maintenance.md
+++ b/docs/storage/maintenance.md
@@ -109,7 +109,7 @@ Moves are not written to the WAL. If the engine crashes before the checkpoint th
 
 Compaction doesn't change any root, so transactions that started before it commit exactly as they would have without it, on the fast path or by rebasing.
 
-The one exception is a bulk load. `Transaction.load` can spill finished subtrees to disk before commit (`Transaction.carriesStoredRoots`), and those pages are not reachable from any generation until the transaction commits, so the walk can't see them. If one of them sits in a victim, it isn't moved. Each transaction therefore records `TransactionManager.compactions()` when it begins, and a transaction that carries stored roots and saw fewer compactions than have since finished fails with a retryable conflict (`bulk-loaded roots of transaction N predate a compaction`). `HypergraphDatabase.write` retries it. The check applies on the fast path too, since compaction no longer changes the roots that would otherwise push such a transaction off it.
+The one exception is a bulk load. `Transaction.load` can spill finished subtrees to disk before commit (`Transaction.carriesStoredRoots`), and those pages are not reachable from any generation until the transaction commits, so the walk can't see them. If one of them sits in a victim, it isn't moved. Each transaction therefore records `TransactionManager.compactions()` when it begins. The count goes up when a compaction takes its cut, under the commit lock, so a transaction that carries stored roots and began before a cut that has since been taken fails with a retryable conflict, even when it commits while pages are still moving (`bulk-loaded roots of transaction N predate a compaction`). `HypergraphDatabase.write` retries it. The check applies on the fast path too, since compaction no longer changes the roots that would otherwise push such a transaction off it.
 
 ### Reclaim
 
diff --git a/engine/src/main/java/io/hstore/engine/maintenance/Compactor.java b/engine/src/main/java/io/hstore/engine/maintenance/Compactor.java
index b39f87b..6c7576b 100644
--- a/engine/src/main/java/io/hstore/engine/maintenance/Compactor.java
+++ b/engine/src/main/java/io/hstore/engine/maintenance/Compactor.java
@@ -59,6 +59,7 @@ private List<Generation> retained() {
 
     private Cut cut() {
         return transactions.exclusive(() -> {
+            transactions.compacted();
             int active = pages.activeSegment();
             Set<Integer> sealed = pages.segments().stream()
                     .filter(segment -> segment.state() == SegmentState.SEALED && segment.id() != active)
@@ -118,7 +119,6 @@ public Report compact(double liveThreshold, int maxVictims) {
         move(live, Set.copyOf(victims));
         faults.reach(CrashPoint.COMPACTION_MOVE);
         long retiredAt = transactions.current().id();
-        transactions.compacted();
         long checkpoints = checkpointer.completed();
         victims.forEach(id -> {
             pages.transition(id, SegmentState.RETIRED, retiredAt);
diff --git a/engine/src/test/java/io/hstore/engine/CompactionTest.java b/engine/src/test/java/io/hstore/engine/CompactionTest.java
index 4e16727..061a292 100644
--- a/engine/src/test/java/io/hstore/engine/CompactionTest.java
+++ b/engine/src/test/java/io/hstore/engine/CompactionTest.java
@@ -25,6 +25,7 @@
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
@@ -170,6 +171,38 @@ void aBulkLoadStartedBeforeCompactionHasToRetry() {
         }
     }
 
+    @Test
+    void aBulkLoadThatCommitsWhilePagesMoveHasToRetry() {
+        AtomicReference<Transaction> pending = new AtomicReference<>();
+        AtomicReference<RuntimeException> refused = new AtomicReference<>();
+        EngineOptions options = options().withFaults(point -> {
+            Transaction early = point == CrashPoint.COMPACTION_MOVE ? pending.getAndSet(null) : null;
+            if (early != null) {
+                refused.set(assertThrows(HStoreException.Conflict.class, early::commit));
+            }
+        });
+        List<Incidence> members = Stream.iterate(2_000_000L, i -> i + 1).limit(50_000).map(Incidence::of).toList();
+        long target;
+        long edge;
+        try (StorageEngine engine = StorageEngine.open(directory, options)) {
+            target = engine.write(txn -> txn.createEdge(EngineTest.EDGE, EdgeKind.SET));
+            Transaction early = engine.begin();
+            early.load(target, members);
+            edge = churn(engine);
+            pending.set(early);
+            assertFalse(engine.compact().compacted().isEmpty());
+            assertTrue(refused.get() != null, "the load committed while its pages were left behind");
+            engine.write(txn -> {
+                txn.load(target, members);
+                return null;
+            });
+        }
+        try (StorageEngine engine = StorageEngine.open(directory, options())) {
+            assertEquals(50_000L, (long) engine.read(snapshot -> snapshot.requireEdge(target).size()));
+            assertEquals(expected(), members(engine, edge));
+        }
+    }
+
     static Stream<Arguments> crashes() {
         return Stream.of(CrashPoint.COMPACTION_MOVE, CrashPoint.COMPACTION_RETIRE)
                 .flatMap(point -> Stream.of(Arguments.of(point, false), Arguments.of(point, true)));

From 6e59c72ab152505c7c32b34a14bdb42300f285bc Mon Sep 17 00:00:00 2001
From: venkat1701 <thejeastdev@gmail.com>
Date: Sun, 11 Oct 2026 13:22:19 +0530
Subject: [PATCH 6/6] fix(engine): put victims back to SEALED when moving their
 pages fails

A failed move left the victims in COMPACTING until a restart. Later passes only
pick SEALED segments and reclaim only deletes RETIRED ones, so those segments
were neither retried nor freed.
---
 .../main/java/io/hstore/engine/maintenance/Compactor.java  | 7 ++++++-
 1 file changed, 6 insertions(+), 1 deletion(-)

diff --git a/engine/src/main/java/io/hstore/engine/maintenance/Compactor.java b/engine/src/main/java/io/hstore/engine/maintenance/Compactor.java
index 6c7576b..ad8a5ab 100644
--- a/engine/src/main/java/io/hstore/engine/maintenance/Compactor.java
+++ b/engine/src/main/java/io/hstore/engine/maintenance/Compactor.java
@@ -116,7 +116,12 @@ public Report compact(double liveThreshold, int maxVictims) {
             return new Report(liveBytes, victims, transactions.current().id());
         }
         victims.forEach(id -> pages.transition(id, SegmentState.COMPACTING, 0));
-        move(live, Set.copyOf(victims));
+        try {
+            move(live, Set.copyOf(victims));
+        } catch (RuntimeException | Error failure) {
+            victims.forEach(id -> pages.transition(id, SegmentState.SEALED, 0));
+            throw failure;
+        }
         faults.reach(CrashPoint.COMPACTION_MOVE);
         long retiredAt = transactions.current().id();
         long checkpoints = checkpointer.completed();