Copy compaction pages outside the commit lock

docs/storage/change-feed.md

@@ -26,7 +26,7 @@ record CommitEvent(long generation, long txnId, long wallTime, int branch,
 * `members` is the workspace's `memberChanges()` list in emission order: `MemberChange.Added(edge, incidence, locator)`, `Removed(edge, incidence, locator)`, `Updated(edge, before, after, locatorBefore, locatorAfter)`. The locator is the member id for `SET` edges and the order label (position key in the order tree) for `ORDERED` edges (`Hyperedge.locator`).
 * `slots` is the workspace's `slotChanges()`: one `SlotChange(slot, key, before, after)` per effective write to a **primary** slot (`Workspace.write` records nothing when `before.equals(after)` and nothing for derived slots). Derived indexes are not in the feed; consumers that need them recompute or read the tree.
 
-`CommitEvent.isEmpty()` is true when both lists are empty. Empty events are never logged: `createBranch`, `closeBranch` and compaction (`rewrite`) commits advance the generation but leave no feed entry. Consumers must therefore treat generation numbers in the feed as **increasing, not contiguous**.
+`CommitEvent.isEmpty()` is true when both lists are empty. Empty events are never logged: `createBranch`, `closeBranch` and compaction (`relocate`) commits advance the generation but leave no feed entry. Consumers must therefore treat generation numbers in the feed as **increasing, not contiguous**.
 
 ## 2. When events are written
 

docs/storage/maintenance.md

@@ -87,14 +87,14 @@ 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.
+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)`); `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.
+3. **Relocate** (`TransactionManager.relocate(moving)`), in three steps so that commits keep going while pages are copied:
+   1. **Snapshot**, under the commit lock for a moment: take the latest appended generation and reserve a transaction id.
+   2. **Copy**, without the commit lock. For every **active** branch of the snapshot, `TreeWalker.relocate(ref, schema, moving, scope, reached)` rebuilds as a fresh `Ref.Pending` copy every stored node whose page id lies in a victim segment, every ancestor of such a node, and every leaf whose nested trees changed. Subtrees that touch no victim are returned unchanged and stay shared, and every page the walk reaches is recorded. The copies are materialized and written to the active segment right away, with their WAL records kept for later under the reserved transaction id. `TreeWalker.pairMoves` then walks old and new trees side by side to map each old page id to its copy.
+   3. **Swap**, under the commit lock, one branch at a time. `TreeWalker.adopt` walks the branch's **latest** roots: a page with a copy is replaced by the copy; a page that existed at the snapshot without a copy is kept as it is, since nothing below it moved; and a page written after the snapshot, by a commit that ran during the copy, is descended into and rebuilt if anything below it moved. Only that last kind costs work under the lock, and there are only as many as commits wrote during the copy. The result is committed like any other commit, carrying the copies' WAL records, and `relocationFence` moves to its generation. Branches created during the copy are swapped the same way.
 
-   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.
+   Without the swap, a crash leaves the copies as unreferenced pages in the active segment, which a later compaction reclaims; the data is unchanged.
 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 persists the new segment states and roots, and calls `reclaim()`.

docs/transactions/transactions.md

@@ -228,7 +228,7 @@ Under `SNAPSHOT` both commits succeed (R replays `Insert(p)` into B, which L nev
 | Data page writes into the OS page cache | Waking the committing thread |
 | WAL append of `Commit`, `head.set(next)`, `committer.submit(pending)` | |
 
-The same lock guards `system(...)` (internal writes such as the dictionary), `createBranch`, `closeBranch` and `rewrite` (compaction). `exclusive(action)` takes the lock **and** drains the committer, which is how `Checkpointer.checkpoint()` obtains a quiescent point where `current == head` and every appended commit is durable and published.
+The same lock guards `system(...)` (internal writes such as the dictionary), `createBranch`, `closeBranch` and the swap step of `relocate` (compaction), which holds it only to point the latest roots at the already-written copies. `exclusive(action)` takes the lock **and** drains the committer, which is how `Checkpointer.checkpoint()` obtains a quiescent point where `current == head` and every appended commit is durable and published.
 
 Because the lock is released before waiting for durability, the next committer can append while the previous batch is being synced. This pipelining is what lets group commit amortise `fsync`.
 
@@ -344,10 +344,10 @@ 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`, `closeBranch` and `relocate` 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.
+* `relocate` (compaction) copies pages outside the lock, then replaces roots without member or slot changes and sets the relocation fence ([maintenance.md](../storage/maintenance.md#compaction)).
 
 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)).

engine/src/main/java/io/hstore/engine/maintenance/Compactor.java

@@ -1,14 +1,12 @@
 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.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.TransactionManager;
 
 import java.util.Comparator;
@@ -69,16 +67,9 @@ public Report compact(double liveThreshold, int maxVictims) {
         victims.forEach(id -> pages.transition(id, SegmentState.COMPACTING, 0));
         Set<Integer> victimSet = Set.copyOf(victims);
         LongPredicate moving = pageId -> victimSet.contains(PageId.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())));
-        }
-        long retiredAt = generation;
+        long retiredAt = transactions.relocate(moving);
         victims.forEach(id -> pages.transition(id, SegmentState.RETIRED, retiredAt));
-        return new Report(live, victims, generation);
+        return new Report(live, victims, retiredAt);
     }
 
     public List<Integer> reclaim() {

engine/src/main/java/io/hstore/engine/tree/TreeWalker.java

@@ -1,6 +1,9 @@
 package io.hstore.engine.tree;
 
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Map;
+import java.util.function.LongConsumer;
 import java.util.function.LongPredicate;
 
 public final class TreeWalker {
@@ -42,36 +45,112 @@ private void visit(Ref ref, TreeSchema<?> schema, int height, Map<Long, Integer>
     }
 
     public Ref relocate(Ref root, TreeSchema<?> schema, LongPredicate moving, WriteScope scope) {
-        return relocate(root, schema, UNKNOWN_HEIGHT, moving, scope);
+        return relocate(root, schema, moving, scope, _ -> {
+        });
     }
 
-    private Ref relocate(Ref ref, TreeSchema<?> schema, int height, LongPredicate inVictim, WriteScope scope) {
+    public Ref relocate(Ref root, TreeSchema<?> schema, LongPredicate moving, WriteScope scope, LongConsumer reached) {
+        return relocate(root, schema, UNKNOWN_HEIGHT, moving, scope, reached);
+    }
+
+    private Ref relocate(Ref ref, TreeSchema<?> schema, int height, LongPredicate inVictim, WriteScope scope, LongConsumer reached) {
         if (!(ref instanceof Ref.Stored(long pageId, int _, Summary _))) {
             return ref;
         }
+        reached.accept(pageId);
         boolean moving = inVictim.test(pageId);
         if (height == 0 && !moving && !schema.codec().holdsRefs()) {
             return ref;
         }
+        return rebuild(ref, pageId, schema, moving, scope, (child, childSchema, childHeight) ->
+                relocate(child, childSchema, childHeight, inVictim, scope, reached));
+    }
+
+    public void pairMoves(Ref before, Ref after, TreeSchema<?> schema, Map<Long, Ref> moved) {
+        pairMoves(before, after, schema, UNKNOWN_HEIGHT, moved);
+    }
+
+    private void pairMoves(Ref before, Ref after, TreeSchema<?> schema, int height, Map<Long, Ref> moved) {
+        if (!(before instanceof Ref.Stored(long oldId, int _, Summary _)) || !(after instanceof Ref.Stored(long newId, int _, Summary _))
+                || oldId == newId || moved.putIfAbsent(oldId, after) != null) {
+            return;
+        }
+        if (height == 0 && !schema.codec().holdsRefs()) {
+            return;
+        }
+        switch (source.load(oldId, schema)) {
+            case Branch branch -> {
+                Branch copy = (Branch) source.load(newId, schema);
+                for (int i = 0; i < branch.size(); i++) {
+                    pairMoves(branch.child(i), copy.child(i), schema, branch.height() - 1, moved);
+                }
+            }
+            case Leaf leaf -> {
+                Leaf copy = (Leaf) source.load(newId, schema);
+                for (int i = 0; i < leaf.size(); i++) {
+                    List<Ref> oldRefs = new ArrayList<>();
+                    List<Ref> newRefs = new ArrayList<>();
+                    List<TreeSchema<?>> schemas = new ArrayList<>();
+                    mapNested(schema, leaf.value(i), (nested, nestedSchema) -> {
+                        oldRefs.add(nested);
+                        schemas.add(nestedSchema);
+                        return nested;
+                    });
+                    mapNested(schema, copy.value(i), (nested, _) -> {
+                        newRefs.add(nested);
+                        return nested;
+                    });
+                    for (int k = 0; k < oldRefs.size(); k++) {
+                        pairMoves(oldRefs.get(k), newRefs.get(k), schemas.get(k), UNKNOWN_HEIGHT, moved);
+                    }
+                }
+            }
+        }
+    }
+
+    public Ref adopt(Ref root, TreeSchema<?> schema, Map<Long, Ref> moved, LongPredicate existed, WriteScope scope) {
+        return adopt(root, schema, UNKNOWN_HEIGHT, moved, existed, scope);
+    }
+
+    private Ref adopt(Ref ref, TreeSchema<?> schema, int height, Map<Long, Ref> moved, LongPredicate existed, WriteScope scope) {
+        if (!(ref instanceof Ref.Stored(long pageId, int _, Summary _))) {
+            return ref;
+        }
+        Ref copy = moved.get(pageId);
+        if (copy != null) {
+            return copy;
+        }
+        if (existed.test(pageId) || (height == 0 && !schema.codec().holdsRefs())) {
+            return ref;
+        }
+        return rebuild(ref, pageId, schema, false, scope, (child, childSchema, childHeight) ->
+                adopt(child, childSchema, childHeight, moved, existed, scope));
+    }
+
+    @FunctionalInterface
+    private interface ChildMapper {
+        Ref map(Ref child, TreeSchema<?> schema, int height);
+    }
+
+    private Ref rebuild(Ref ref, long pageId, TreeSchema<?> schema, boolean force, WriteScope scope, ChildMapper mapper) {
         return switch (source.load(pageId, schema)) {
             case Leaf leaf -> {
                 Object[] values = leaf.valueList().toArray();
-                boolean changed = moving;
+                boolean changed = force;
                 for (int i = 0; i < values.length; i++) {
-                    Object mapped = mapNested(schema, values[i], (nested, nestedSchema) ->
-                            relocate(nested, nestedSchema, UNKNOWN_HEIGHT, inVictim, scope));
+                    Object mapped = mapNested(schema, values[i], (nested, nestedSchema) -> mapper.map(nested, nestedSchema, UNKNOWN_HEIGHT));
                     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;
+                boolean changed = force;
                 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;
+                    Ref mapped = mapper.map(children[i], schema, branch.height() - 1);
+                    changed |= mapped != children[i];
+                    children[i] = mapped;
                 }
                 yield changed
                         ? new Ref.Pending(Branch.owned(schema, scope.token(), branch.separatorArray(), children, children.length, branch.height()))