fix(engine): power-loss crash tests and the five durability bugs they found

docs/storage/change-feed.md

@@ -75,7 +75,7 @@ offset  size  field
 
 **Append and roll.** `append` writes the frame to the last segment. Before writing, it rolls when there is no segment yet or the active segment already holds at least `segmentBytes` (`end - lastKey >= segmentBytes`). A frame is never split, so a segment may exceed `segmentBytes` by at most one frame, and every segment begins at a frame boundary. Rolling forces the previous segment, creates `%016x.feed` at the current `end`, and forces the `feed/` directory.
 
-**Open.** `restore()` opens every `*.feed`, scans all frames from the first segment's start to the end of the last segment, and stops at the first frame that is short, has a negative or overlong length, or fails its CRC. `cutAt(position)` then deletes every segment that starts after the failure point and truncates the segment containing it, so the feed always ends at a clean frame.
+**Open.** `restore()` opens every `*.feed`, scans all frames from the first segment's start to the end of the last segment, and stops at the first frame that is short, has a negative or overlong length, fails its CRC, or is not newer than the frame before it. The CRC covers only the payload, so a frame also counts as damaged unless its payload is non-empty and starts with the same generation as its header (`FeedCodec` always writes the generation first); this keeps a damaged header, or a run of zeros left where an unsynced frame never reached the disk, from passing as a frame. `cutAt(position)` then deletes every segment that starts after the failure point and truncates the segment containing it, so the feed always ends at a clean frame.
 
 **Index.** Each `Segment` remembers `firstGeneration`, the generation of its first frame. An in-memory sparse index `generation -> global position` holds the first frame of **every segment** and every 64th frame thereafter (`INDEX_STRIDE = 64`). `startFor(g)` seeks to the floor entry (never before the first retained segment) and scans forward, so locating a generation costs at most 63 frame reads plus segment boundaries.
 

docs/storage/pages.md

@@ -194,7 +194,12 @@ checkpoint writes have no WAL record; the checkpoint forces the directory before
 refers to them. Recovery
 writes each logged address back into the directory before it checks or replays the page, so an entry that a crash
 lost comes back from the log. The header's next index is updated on every allocation, and recovery raises it past
-every number it restores, so a restored number is never handed out a second time.
+every number it restores, so a restored number is never handed out a second time. Number 0 is never used, so a
+next index of 0 (a directory whose header was forced when the file was created, before the first checkpoint forced
+it again) is read as 1.
+
+`pages.dir` and `pages.ref` force the `data/segments` directory when they are created, and so does every new segment
+file, so that a power loss can't drop the file itself after its contents were forced.
 
 ### Reference counts
 
@@ -274,7 +279,7 @@ sequenceDiagram
 
 * `allocate` rejects `length > pageSize`.
 * Allocation is serialized by a `ReentrantLock`, and only the counter and directory updates run under it.
-* Rolling a segment marks the previous one `SEALED`, creates the next id, and resets `nextUnit` to 0.
+* Rolling a segment marks the previous one `SEALED`, creates the next id, forces the `data/segments` directory so the new file's entry is durable before any page in it is, and resets `nextUnit` to 0.
 * `PageStore.write(pageId, image)` looks the number up in the directory and performs a positional `FileChannel.write` at `offset × 64`, then marks the file dirty. It refuses an image longer than the units allocated for it. A checkpoint writes each image as soon as it has allocated it. For the pages a commit writes, the write happens later, inside `TransactionManager.append`, after the WAL records have been appended (see [wal-and-recovery.md](../transactions/wal-and-recovery.md)).
 * `PageStore.sync()` calls `FileChannel.force(false)` on every dirty segment. The group committer calls it once per batch in `PAGE_REFERENCES` WAL mode, before the WAL sync, if a commit in the batch wrote pages. `PageStore.checkpoint()` does the same and then forces the page directory; only `Checkpointer` calls it.
 

docs/transactions/wal-and-recovery.md

@@ -17,7 +17,7 @@ Source of truth:
 | Checkpoints | `engine/src/main/java/io/hstore/engine/maintenance/Checkpointer.java`, `tree/Materializer.java`, `catalog/CatalogStore.java` |
 | Reference counts | `engine/src/main/java/io/hstore/engine/maintenance/ReferenceCounter.java`, `page/ReferenceCounts.java`, `page/ReferenceJournal.java` |
 | Byte encodings | `engine/src/main/java/io/hstore/engine/page/ByteCursor.java`, `tree/Ref.java`, `tree/Summary.java` |
-| Crash test matrix | `engine/src/test/java/io/hstore/engine/CrashRecoveryTest.java`, `DeferredWritesTest.java` |
+| Crash test matrix | `engine/src/test/java/io/hstore/engine/CrashRecoveryTest.java`, `DeferredWritesTest.java`, `PowerLossTest.java` |
 
 Related: [transactions](transactions.md), [change feed](../storage/change-feed.md), [pages](../storage/pages.md), [catalog and generations](../storage/catalog-and-generations.md), [maintenance](../storage/maintenance.md), [configuration](../operations/configuration.md).
 
@@ -27,7 +27,8 @@ Related: [transactions](transactions.md), [change feed](../storage/change-feed.m
 
 ```text
 <db>/
-  FORMAT                         properties: format=8, page-size=<bytes>
+  FORMAT                         properties: format=8, page-size=<bytes>; written to FORMAT.tmp, forced,
+                                 renamed and the directory forced, before anything else is created
   LOCK                           exclusive FileLock held while open
   catalog/manifest               name of the current catalog image file
   catalog/generations/%016x.cat  checkpointed catalog images (3 newest retained)
@@ -353,3 +354,40 @@ After each completed checkpoint, `StorageEngine.timedCheckpoint` logs `checkpoin
 `recoveryStopsAtTheFirstCommitWhosePagesAreMissing` covers the power-loss reordering that a process crash cannot produce. With `PAGE_REFERENCES` it commits `durable`, reopens, commits `lost-1` and `lost-2`, each of which creates a node and bulk-loads a 20,000-member edge (the load spills pages, and the commit writes the rest of the tree because it grew by more than 4,096 entries), halts, and truncates the last data segment to one byte short of its size before `lost-2`, so `lost-1`'s last page fails its checksum. Recovery must report `discardedCommits >= 1`, keep `durable`, drop `lost-2` (which itself has intact pages but follows a broken commit), and accept new writes that survive another reopen.
 
 `concurrentCommitsShareDurabilityBarriers` runs 400 concurrent writers and checks that all 400 commits are visible and that `averageGroupCommit >= 1`.
+
+### Power loss
+
+The tests above stop the engine but keep every byte it wrote, which is what a killed process leaves behind. `PowerLossTest` models a power failure instead, where only what was forced is certain to be on disk.
+
+The engine opens every file it writes or syncs through `FileHandle.channel`, and forces its memory-mapped files (`pages.dir`, `pages.ref`) through `FileHandle.force`. Both go through a package-private `FileHandle.Channels` that defaults to `FileChannel::open`. The test helper `PowerLoss` (`engine/src/test/java/io/hstore/engine/page/PowerLoss.java`) replaces it for one directory and records, per file, every write, truncation and `force`, every force of a mapping, and every force of a directory. It can cut the power before any of these events: the call throws `PowerLoss.Cut`, and later I/O still reaches the files but is no longer recorded. `restore` then rewrites the directory as the disk could hold it after the failure:
+
+| `Tear` | Writes made since a file's last `force` |
+|---|---|
+| `NOTHING` | all lost; every file holds what its last `force` saw |
+| `ALL` | all kept, as after a process crash |
+| `HALF_LAST` | all kept except the last one in each file, which is cut in half: a WAL record cut short, half a page |
+| `FLIP_LAST` | all kept, with one byte flipped in the middle of the last one in each file, so its checksum fails |
+| `RANDOM` | per file a prefix, a random subset or all of them, with a write sometimes applied only in part; a mapped file mixes 4 KiB blocks from its last `force` and from the moment of the crash; sometimes one byte of one unsynced write is flipped |
+
+A file created since its directory was last forced may be gone (always, under `NOTHING`). A file the engine wrote without going through `FileHandle` is left empty, unless it was renamed from a `.tmp` file that was forced first, which keeps its contents.
+
+`powerCutsAtRandomEventsRecoverAPrefixOfTheCommits` runs a seeded workload for both durability levels and both WAL modes: small commits that append a new node to one ordered edge, commits that also bulk-load 300 or 5,000 members into a new edge (spilling pages, and writing pages at commit for trees past 4,096 changes), commits that delete an earlier bulk edge, aborted transactions with and without spilled pages, checkpoints and compactions. It first runs without a crash to count the I/O events, then cuts the power at four random events, each with a random `Tear`. One seed per configuration runs by default; `-Dhstore.powerloss.seeds=N` runs seeds 1 to N. After restoring, it opens the database, checks it, commits once more and checkpoints, with a second power cut at a random event anywhere in that sequence, including recovery itself. After restoring again it checks once more. Each check requires that:
+
+* the commits that survived are a prefix of the commits attempted;
+* under `SYNC` every acknowledged commit is in it, and under `ASYNC` every commit a completed checkpoint covered;
+* no aborted transaction appears, and every bulk edge exists with its full size exactly when the prefix created it and did not delete it;
+* the change feed holds one event per surviving commit and nothing else, generations 1 to the last one;
+* the second recovery found the same prefix as the first, and the commit after it survived if it was acknowledged under `SYNC` or checkpointed;
+* `TreeVerifier` accepts every tree of every branch and the reference counts match (`verifyReferences`), as `hstore check` does.
+
+The other tests in the class each pin one case: unsynced `ASYNC` commits after a checkpoint are lost together; a `Commit` record cut short or with a flipped byte loses only the commit that was not acknowledged; a commit whose `PAGE_REFERENCES` page was torn is discarded although its log records survived; and five cases that failed before the fixes that came with them, listed in the table below.
+
+| Test | What went wrong before |
+|---|---|
+| `aDatabaseCreatedJustBeforeAPowerLossStillOpens` | `FORMAT` was written without a `force`, so it could come back empty, and the database then refused to open (`uses storage format 1`) although its log held acknowledged commits |
+| `pagesInASegmentCreatedSinceTheLastCheckpointSurvive` | the `data/segments` directory was never forced, so a new segment file, or `pages.dir` and `pages.ref` themselves, could disappear with the pages of acknowledged commits in them |
+| `aPageDirectoryWhoseFirstPageNumberWasNeverForcedStillHandsOutNumbers` | `pages.dir` forces its header with next index 0 when it is created and only sets it to 1 afterwards; losing power before the first checkpoint left next index 0, and every later checkpoint failed with `page number out of range: 0` |
+| `aZeroedFeedFrameBeforeASurvivingOneDoesNotRepeatEvents` | zeros where an unsynced feed frame never reached the disk read as frames of generation 0; recovery then appended the events of replayed commits a second time |
+| `aFeedFrameWhoseUnsyncedGenerationWasDamagedDoesNotHideCommittedEvents` | the feed CRC does not cover the header's generation; a damaged generation made recovery skip the events of acknowledged commits, which were then lost from the feed |
+
+The model does not cover directories created with `Files.createDirectories` (their own entries are assumed durable), deletions (a deleted file stays deleted), or the files of the database module (`PayloadStore`, `SemanticPlane`).

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

@@ -14,6 +14,7 @@
 import io.hstore.engine.maintenance.Compactor;
 import io.hstore.engine.maintenance.Recovery;
 import io.hstore.engine.maintenance.ReferenceCounter;
+import io.hstore.engine.page.FileHandle;
 import io.hstore.engine.page.PageStore;
 import io.hstore.engine.tree.NodeCache;
 import io.hstore.engine.tree.NodeSource;
@@ -24,12 +25,15 @@
 import io.hstore.engine.txn.TxnOptions;
 import io.hstore.engine.wal.WriteAheadLog;
 
+import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.io.UncheckedIOException;
+import java.nio.ByteBuffer;
 import java.nio.channels.FileChannel;
 import java.nio.channels.FileLock;
 import java.nio.file.Files;
 import java.nio.file.Path;
+import java.nio.file.StandardCopyOption;
 import java.nio.file.StandardOpenOption;
 import java.time.Duration;
 import java.util.List;
@@ -145,9 +149,16 @@ private static void verifyFormat(Path directory, EngineOptions options) throws I
         }
         format.setProperty("format", FORMAT_VERSION);
         format.setProperty("page-size", Integer.toString(options.pageSize()));
-        try (var out = Files.newOutputStream(file)) {
-            format.store(out, null);
+        ByteArrayOutputStream bytes = new ByteArrayOutputStream();
+        format.store(bytes, null);
+        Path temporary = directory.resolve("FORMAT.tmp");
+        try (FileChannel channel = FileHandle.channel(temporary, StandardOpenOption.CREATE, StandardOpenOption.WRITE,
+                StandardOpenOption.TRUNCATE_EXISTING)) {
+            channel.write(ByteBuffer.wrap(bytes.toByteArray()));
+            channel.force(true);
         }
+        Files.move(temporary, file, StandardCopyOption.ATOMIC_MOVE);
+        FileHandle.syncDirectory(directory);
     }
 
     public Path directory() {

engine/src/main/java/io/hstore/engine/catalog/CatalogStore.java

@@ -5,6 +5,7 @@
 
 import io.hstore.engine.page.ByteCursor;
 import io.hstore.engine.page.Checksums;
+import io.hstore.engine.page.FileHandle;
 
 import java.io.IOException;
 import java.io.UncheckedIOException;
@@ -114,7 +115,7 @@ private Stream<Path> files() {
     private void writeDurably(Path target, byte[] bytes) {
         Path temporary = target.resolveSibling(target.getFileName() + ".tmp");
         try {
-            try (FileChannel channel = FileChannel.open(temporary, StandardOpenOption.CREATE, StandardOpenOption.WRITE,
+            try (FileChannel channel = FileHandle.channel(temporary, StandardOpenOption.CREATE, StandardOpenOption.WRITE,
                     StandardOpenOption.TRUNCATE_EXISTING)) {
                 channel.write(ByteBuffer.wrap(bytes));
                 channel.force(true);
@@ -124,19 +125,12 @@ private void writeDurably(Path target, byte[] bytes) {
             } catch (AtomicMoveNotSupportedException e) {
                 Files.move(temporary, target, StandardCopyOption.REPLACE_EXISTING);
             }
-            syncDirectory(target.getParent());
+            FileHandle.syncDirectory(target.getParent());
         } catch (IOException e) {
             throw new UncheckedIOException("cannot publish catalog file " + target, e);
         }
     }
 
-    private static void syncDirectory(Path directory) {
-        try (FileChannel channel = FileChannel.open(directory, StandardOpenOption.READ)) {
-            channel.force(true);
-        } catch (IOException _) {
-        }
-    }
-
     private void deleteQuietly(Path path) {
         try {
             Files.deleteIfExists(path);

engine/src/main/java/io/hstore/engine/feed/ChangeFeed.java

@@ -122,6 +122,9 @@ private void restore() throws IOException {
         long position = segments.firstKey();
         for (Iterator<Frame> frames = new FrameReader(position, last.start + last.file.size()); frames.hasNext(); ) {
             Frame frame = frames.next();
+            if (frame.generation() <= lastGeneration) {
+                break;
+            }
             remember(frame.generation(), position);
             position = frame.next();
         }
@@ -210,7 +213,7 @@ private Segment roll() throws IOException {
         Path path = directory.resolve("%016x.feed".formatted(end));
         Segment segment = new Segment(end, path, FileHandle.open(path));
         segments.put(end, segment);
-        try (FileChannel listing = FileChannel.open(directory, StandardOpenOption.READ)) {
+        try (FileChannel listing = FileHandle.channel(directory, StandardOpenOption.READ)) {
             listing.force(true);
         } catch (IOException unsupported) {
             LOG.log(System.Logger.Level.DEBUG, "directory sync is not supported for " + directory, unsupported);
@@ -400,14 +403,18 @@ private Frame readLocked() {
                     return null;
                 }
                 byte[] payload = body.array();
-                return Checksums.crc32c(payload) == crc
+                return Checksums.crc32c(payload) == crc && carries(payload, generation)
                         ? new Frame(position, position + FRAME_HEADER + length, generation, payload)
                         : null;
             } catch (IOException e) {
                 throw new UncheckedIOException(e);
             }
         }
 
+        private static boolean carries(byte[] payload, long generation) {
+            return payload.length > 0 && ByteCursor.wrap(payload).getVarLong() == generation;
+        }
+
         private boolean readFully(FileHandle file, ByteBuffer buffer, long at) throws IOException {
             return file.call(channel -> {
                 while (buffer.hasRemaining()) {

engine/src/main/java/io/hstore/engine/page/FileHandle.java

@@ -4,9 +4,11 @@
 package io.hstore.engine.page;
 
 import java.io.IOException;
+import java.lang.foreign.MemorySegment;
 import java.nio.channels.ClosedByInterruptException;
 import java.nio.channels.ClosedChannelException;
 import java.nio.channels.FileChannel;
+import java.nio.file.OpenOption;
 import java.nio.file.Path;
 import java.nio.file.StandardOpenOption;
 import java.util.concurrent.locks.ReentrantLock;
@@ -26,6 +28,16 @@ public interface Action {
         void run(FileChannel channel) throws IOException;
     }
 
+    @FunctionalInterface
+    interface Channels {
+        FileChannel open(Path path, OpenOption... options) throws IOException;
+
+        default void forced(Path path) {
+        }
+    }
+
+    static volatile Channels channels = FileChannel::open;
+
     private final Path path;
     private final ReentrantLock reopening = new ReentrantLock();
     private volatile FileChannel channel;
@@ -37,7 +49,18 @@ private FileHandle(Path path, FileChannel channel) {
     }
 
     public static FileHandle open(Path path) throws IOException {
-        return new FileHandle(path, FileChannel.open(path, StandardOpenOption.CREATE, StandardOpenOption.READ, StandardOpenOption.WRITE));
+        return new FileHandle(path, channel(path, StandardOpenOption.CREATE, StandardOpenOption.READ, StandardOpenOption.WRITE));
+    }
+
+    public static FileChannel channel(Path path, OpenOption... options) throws IOException {
+        return channels.open(path, options);
+    }
+
+    public static void syncDirectory(Path directory) {
+        try (FileChannel listing = channel(directory, StandardOpenOption.READ)) {
+            listing.force(true);
+        } catch (IOException _) {
+        }
     }
 
     public static <T> T uninterrupted(Supplier<T> work) {
@@ -90,11 +113,18 @@ public long size() throws IOException {
         return call(FileChannel::size);
     }
 
+    void force(MemorySegment... mappings) {
+        for (MemorySegment mapping : mappings) {
+            mapping.force();
+        }
+        channels.forced(path);
+    }
+
     private void reopen(FileChannel failed) throws IOException {
         reopening.lock();
         try {
             if (!closed && channel == failed) {
-                channel = FileChannel.open(path, StandardOpenOption.READ, StandardOpenOption.WRITE);
+                channel = channel(path, StandardOpenOption.READ, StandardOpenOption.WRITE);
             }
         } finally {
             reopening.unlock();