perf(engine): keep checkpoint syncs and the rebind walk off the locks commits wait on

docs/storage/persistent-tree.md

@@ -183,7 +183,7 @@ A `WriteScope` built with a `PageSink` (every `Transaction` has one) can spill c
 
 A checkpoint that fails before it publishes leaves its epoch unpublished, so nothing trusts the pages it wrote, and the next checkpoint writes those nodes again.
 
-**Rebinding.** Commits keep publishing while a checkpoint writes `g`, so by the time it publishes, the head generation may be several commits past `g` and still hold the pending nodes the checkpoint wrote. `Materializer.rebound(ref, schema, epoch)` walks the head generation's roots under the commit lock and replaces every node stamped by that epoch with its stored or inline reference. A pending parent whose children changed this way is copied with the new references; one whose children didn't change is kept as it is. Later commits then build on stored references, and the written nodes can be collected once no retained generation or open snapshot still holds them.
+**Rebinding.** Commits keep publishing while a checkpoint writes `g`, so by the time it publishes, the head generation may be several commits past `g` and still hold the pending nodes the checkpoint wrote. `Materializer.Rebinding` walks the head generation's roots and replaces every node stamped by that epoch with its stored or inline reference. A pending parent whose children changed this way is copied with the new references; one whose children didn't change is kept as it is. The walk remembers its result for every pending node it visits, keyed by identity. The checkpoint walks once without the commit lock, then again under it over the head that commits have moved on to since, and the second walk only descends into nodes the first one never saw ([checkpoints](../transactions/wal-and-recovery.md#9-checkpoints)). A node a commit stamps with the epoch after the first walk saw it stays pending until the next checkpoint, which reuses its page. Later commits then build on stored references, and the written nodes can be collected once no retained generation or open snapshot still holds them.
 
 ### Inline trees
 

docs/transactions/wal-and-recovery.md

@@ -76,7 +76,7 @@ The checksum covers the LSN, type, transaction id and payload. Embedding the LSN
 
 **Append.** `append(List<WalRecord>)` encodes the whole list into one buffer and issues it as one positional write under the WAL lock, so the records of one call are contiguous. Before writing, the log rolls to a new segment if the active segment is non-empty and the batch would push it past `walSegmentBytes`. A batch is never split across segments; a batch larger than a segment gets an oversized segment of its own. Rolling `force(true)`s and closes the previous segment, creates the new file, and forces the `wal/segments` directory (`syncDirectory`; filesystems that cannot open a directory for sync are logged at `DEBUG` and skipped) so that the new segment's directory entry survives a power loss.
 
-**Sync.** `sync()` calls `FileChannel.force(false)` on the active segment if anything was written since the last sync (`dirty`). Rolled segments were forced when they were closed, so one call makes the entire log durable.
+**Sync.** `sync()` calls `FileChannel.force(false)` on the active segment if anything was written since the last sync (`dirty`). Rolled segments were forced when they were closed, so one call makes the entire log durable. The force runs outside the WAL lock: `sync()` takes a separate sync lock, reads and clears `dirty` and picks the active segment under the WAL lock, releases it, reaches `WAL_FORCE` and forces. Appends carry on during the force, and a record appended after `dirty` was cleared sets it again, so the next `sync()` forces once more. Because syncs queue on the sync lock, a `sync()` that finds `dirty` clear can only return after the force that covered its records has finished. If the log rolls while a force is running, the roll forces and closes that segment itself, so a force that then fails on the closed file is not an error. Any other failure sets `dirty` again and is thrown.
 
 **Open.** `restore()` lists `*.wal`, scans only the **last** segment frame by frame, truncates it to the end of the last valid frame and forces it. Recovery therefore always appends after a clean frame boundary. The scan checks each frame's header and CRC-32C but doesn't decode the records, and a reader reuses one header and one body buffer, growing the body only for a larger frame, so opening doesn't allocate per frame.
 
@@ -89,7 +89,7 @@ The checksum covers the LSN, type, transaction id and payload. Embedding the LSN
 
 Only the tail can legitimately be torn, because every earlier segment was forced when the log rolled past it. Damage anywhere else means data loss or tampering, and the engine refuses to open rather than recover a silently shortened history.
 
-**Truncate.** `truncateBefore(lsn)` deletes every segment whose *successor* starts at or before `lsn`, never the active segment. The segment containing `lsn` is always kept.
+**Truncate.** `truncateBefore(lsn)` deletes every segment whose *successor* starts at or before `lsn`, never the active segment. The segment containing `lsn` is always kept. The segments are taken out of the log's list under the WAL lock and their files are deleted after it is released.
 
 ## 4. Record types
 
@@ -300,7 +300,7 @@ LOG:  [recovery] recovered generation 158: replayed 0 commits from lsn 156,490,
 
 ## 9. Checkpoints
 
-A checkpoint writes the trees of one published generation to pages and makes that generation the starting point of the next recovery. `Checkpointer.checkpoint()` takes the commit lock for one short step only (step 9), so commits keep running while it writes and syncs.
+A checkpoint writes the trees of one published generation to pages and makes that generation the starting point of the next recovery. `Checkpointer.checkpoint()` takes the commit lock for one short step only (the end of step 9), and none of its file syncs hold the commit lock or the WAL lock, so commits keep running while it writes and syncs.
 
 1. **Target.** `g = transactions.current()`, the newest published generation, and a new `Epoch` for this attempt.
 2. **Write the pages.** A `Materializer` walks every branch's roots in `g` and writes each pending node it reaches, children first: `pages.allocate`, `pages.write`, then the frozen node is admitted to the node cache and the in-memory node is stamped with its stored reference and the epoch. A node stamped by the last published epoch was written by the previous checkpoint, or by a commit that logged its tree by reference, so its stored reference is reused and the subtree below it is not visited. The images carry `g` as their creation epoch. The result is `g` with every root stored.
@@ -310,11 +310,11 @@ A checkpoint writes the trees of one published generation to pages and makes tha
 6. **Count.** `ReferenceCounter.checkpointed(previous, written)` compares the previously checkpointed generation with the one just written and updates the counts ([maintenance](../storage/maintenance.md#reference-counts)). Dead pages that no retained or pinned generation can reach are then released ([maintenance](../storage/maintenance.md#releasing-pages)).
 7. `pages.checkpoint()` again, then `pages.journal(lsn)`: every count changed since the last checkpoint and every released directory entry, as absolute values, go to `pages.ref.journal` behind a header holding `lsn` and with a CRC-32C at the end, and the file is forced. `feed.sync()` (forces the active feed segment; earlier feed segments were forced when the feed rolled).
 8. `CHECKPOINT_CATALOG`, then `catalog.publish(CatalogImage(written g, List.of(), segments, lsn, feed.size()))`. Retained history is not written into the image. The image is framed as `i32 magic 0x54414348`, `i32 crc32c(payload)`, `i64 length`, payload; written to `generations/%016x.cat.tmp`, forced, atomically renamed, the directory forced; then `manifest` is rewritten the same way. The three newest `.cat` files are kept; `loadLatest()` falls back to the newest valid file if the manifest is missing or names a corrupt file.
-9. `TransactionManager.checkpointed(written g, epoch)` publishes the epoch and forgets the commit positions of generations before `g`. Then, holding the commit lock and with the group committer drained, it swaps the written pages into the head generation: every node there that this checkpoint wrote is replaced with its stored reference, and parents that only changed because of that are copied (`Materializer.rebound`). Head, `current` and head's history entry become the rebound generation. Last, it recounts the write buffer and wakes commits that were waiting for room.
-10. `wal.append(Checkpoint(0, g, lsn))`, `wal.sync()`, `wal.truncateBefore(lsn)`.
+9. `TransactionManager.checkpointed(written g, epoch)` publishes the epoch and forgets the commit positions of generations before `g`. It then swaps the written pages into the head generation: every node there that this checkpoint wrote is replaced with its stored reference, and parents that only changed because of that are copied (`Materializer.Rebinding`). The first pass runs over the head it reads without any lock and remembers what it did for every node it visited. `CHECKPOINT_REBIND` is reached next. Then, holding the commit lock and with the group committer drained, a second pass runs over the head as it is now: nodes the first pass already visited are looked up, so this pass only walks what commits added in between. Under `durability = sync` the drain can still wait for the sync of a commit batch that is in flight. Head, `current` and head's history entry become the rebound generation. Last, it recounts the write buffer and wakes commits that were waiting for room.
+10. `wal.append(Checkpoint(0, g, lsn))`, `wal.sync()`, `wal.truncateBefore(lsn)`. The sync and the file deletions run outside the WAL lock (section 3), so a commit that appends meanwhile does not wait for them.
 11. `CHECKPOINT_JOURNAL`, then `pages.applyJournal()` writes the journaled values into `pages.ref` and `pages.dir`, forces both and truncates the journal to zero bytes. The checkpoint remembers `lsn` and resumes the counter.
 
-Because the journal holds absolute values, applying it twice gives the same files. If the process dies after step 8 but before step 11 is finished, the next open finds a journal whose LSN matches the published catalog and applies it again. If it dies before step 8, the journal's LSN doesn't match and it is ignored: `pages.ref` and `pages.dir` were not touched, so they still match the previous catalog, and the pages this checkpoint wrote have addresses but no counts, so the next open hands them to its first checkpoint, which finds nothing that reaches them and marks them dead. If step 10 or 11 fails and the engine keeps running, the next checkpoint applies the old journal before writing a new one, and a failed step 8 marks `pages.ref` for rebuilding. `ReferenceCountTest` crashes at `CHECKPOINT_CATALOG` and at `CHECKPOINT_JOURNAL` and checks the counts after reopening, including after a checkpoint that failed at step 11 and a next one that crashed before step 8. `DeferredWritesTest` crashes a checkpoint at `CHECKPOINT_PAGES`, `CHECKPOINT_CATALOG` and `CHECKPOINT_JOURNAL` while another thread keeps committing, and checks that every acknowledged commit is there after reopening and that the counts match.
+Because the journal holds absolute values, applying it twice gives the same files. If the process dies after step 8 but before step 11 is finished, the next open finds a journal whose LSN matches the published catalog and applies it again. If it dies before step 8, the journal's LSN doesn't match and it is ignored: `pages.ref` and `pages.dir` were not touched, so they still match the previous catalog, and the pages this checkpoint wrote have addresses but no counts, so the next open hands them to its first checkpoint, which finds nothing that reaches them and marks them dead. If step 10 or 11 fails and the engine keeps running, the next checkpoint applies the old journal before writing a new one, and a failed step 8 marks `pages.ref` for rebuilding. `ReferenceCountTest` crashes at `CHECKPOINT_CATALOG` and at `CHECKPOINT_JOURNAL` and checks the counts after reopening, including after a checkpoint that failed at step 11 and a next one that crashed before step 8. `DeferredWritesTest` crashes a checkpoint at `CHECKPOINT_PAGES`, `CHECKPOINT_CATALOG` and `CHECKPOINT_JOURNAL` while another thread keeps committing, and checks that every acknowledged commit is there after reopening and that the counts match. `CheckpointConcurrencyTest` does the same with three writers at `CHECKPOINT_CATALOG`, `CHECKPOINT_REBIND`, `WAL_FORCE` and `CHECKPOINT_JOURNAL`, reopening twice. A crash at `WAL_FORCE` can also land in a commit batch that the checkpoint drains in step 9, so there the writers may fail too, and the only commits allowed to come back without an acknowledgement are the ones in flight. It also commits from another thread while the checkpoint is held at `CHECKPOINT_REBIND` or `WAL_FORCE`, which must finish without waiting for the checkpoint and still be there after the rebind and after reopening, and it runs four writers through back-to-back checkpoints, each checking that its commit is visible right after it returns.
 
 The checkpoint LSN is the end of `g`'s `Commit` record, not the position of the `Checkpoint` record. The next recovery starts there, replays the commits published after `g`, and skips the `Checkpoint` record when it reaches it. Triggers:
 

engine/src/main/java/io/hstore/engine/StorageEngine.java

@@ -83,7 +83,7 @@ private StorageEngine(Path directory, EngineOptions options, FileChannel lockCha
         CatalogImage checkpoint = Recovery.discover(catalog);
         this.pages = PageStore.open(directory.resolve("data").resolve("segments"), options.pageSize(), options.pagesPerSegment(),
                 options.maxOpenSegmentFiles(), checkpoint.segments(), checkpoint.checkpointLsn());
-        this.wal = WriteAheadLog.open(directory.resolve("wal").resolve("segments"), options.walSegmentBytes());
+        this.wal = WriteAheadLog.open(directory.resolve("wal").resolve("segments"), options.walSegmentBytes(), options.faults());
         this.slots = new SlotRegistry(options.slots());
         FeedCodec codec = new FeedCodec(slots);
         this.feed = ChangeFeed.open(directory.resolve("feed"), codec, options.walSegmentBytes());

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

@@ -7,6 +7,8 @@
 import io.hstore.engine.page.PageId;
 
 import java.lang.foreign.MemorySegment;
+import java.util.IdentityHashMap;
+import java.util.Map;
 import java.util.function.Consumer;
 
 public final class Materializer {
@@ -203,45 +205,65 @@ public static long unwrittenHeap(Ref ref) {
         return total[0];
     }
 
-    public static Ref rebound(Ref ref, TreeSchema<?> schema, Epoch epoch) {
-        if (!(ref instanceof Ref.Pending(Node node))) {
-            return ref;
+    public static final class Rebinding {
+
+        private final Epoch epoch;
+        private final Map<Node, Ref> visited = new IdentityHashMap<>();
+
+        public Rebinding(Epoch epoch) {
+            this.epoch = epoch;
         }
-        Ref written = node.writtenIn(epoch);
-        if (written != null) {
-            return written;
+
+        public Ref rebound(Ref ref, TreeSchema<?> schema) {
+            if (!(ref instanceof Ref.Pending(Node node))) {
+                return ref;
+            }
+            Ref written = node.writtenIn(epoch);
+            if (written != null) {
+                return written;
+            }
+            Ref known = visited.get(node);
+            if (known != null) {
+                return known instanceof Ref.Pending(Node same) && same == node ? ref : known;
+            }
+            Ref result = switch (node) {
+                case Leaf leaf -> rebound(ref, leaf);
+                case Branch branch -> rebound(ref, branch);
+            };
+            visited.put(node, result);
+            return result;
         }
-        return switch (node) {
-            case Leaf leaf -> {
-                if (!leaf.schema.codec().holdsRefs()) {
-                    yield ref;
-                }
-                Object[] values = leaf.valueList().toArray();
-                boolean changed = false;
-                for (int i = 0; i < values.length; i++) {
-                    boolean[] moved = {false};
-                    Object mapped = mapRefs(leaf.schema, values[i], (nested, nestedSchema) -> {
-                        Ref rebound = rebound(nested, nestedSchema, epoch);
-                        moved[0] |= rebound != nested;
-                        return rebound;
-                    });
-                    if (moved[0]) {
-                        values[i] = mapped;
-                        changed = true;
-                    }
-                }
-                yield changed ? new Ref.Pending(leaf.frozenWith(values)) : ref;
+
+        private Ref rebound(Ref ref, Leaf leaf) {
+            if (!leaf.schema.codec().holdsRefs()) {
+                return ref;
             }
-            case Branch branch -> {
-                Ref[] children = branch.childArray();
-                boolean changed = false;
-                for (int i = 0; i < children.length; i++) {
-                    Ref rebound = rebound(children[i], branch.schema, epoch);
-                    changed |= rebound != children[i];
-                    children[i] = rebound;
+            Object[] values = leaf.valueList().toArray();
+            boolean changed = false;
+            for (int i = 0; i < values.length; i++) {
+                boolean[] moved = {false};
+                Object mapped = mapRefs(leaf.schema, values[i], (nested, nestedSchema) -> {
+                    Ref rebound = rebound(nested, nestedSchema);
+                    moved[0] |= rebound != nested;
+                    return rebound;
+                });
+                if (moved[0]) {
+                    values[i] = mapped;
+                    changed = true;
                 }
-                yield changed ? new Ref.Pending(branch.frozenWith(children)) : ref;
             }
-        };
+            return changed ? new Ref.Pending(leaf.frozenWith(values)) : ref;
+        }
+
+        private Ref rebound(Ref ref, Branch branch) {
+            Ref[] children = branch.childArray();
+            boolean changed = false;
+            for (int i = 0; i < children.length; i++) {
+                Ref rebound = rebound(children[i], branch.schema);
+                changed |= rebound != children[i];
+                children[i] = rebound;
+            }
+            return changed ? new Ref.Pending(branch.frozenWith(children)) : ref;
+        }
     }
 }

engine/src/main/java/io/hstore/engine/txn/CrashPoint.java

@@ -15,7 +15,9 @@ public enum CrashPoint {
     COMPACTION_RETIRE,
     CHECKPOINT_PAGES,
     CHECKPOINT_CATALOG,
-    CHECKPOINT_JOURNAL;
+    CHECKPOINT_JOURNAL,
+    CHECKPOINT_REBIND,
+    WAL_FORCE;
 
     @FunctionalInterface
     public interface Injector {