diff --git a/docs/storage/change-feed.md b/docs/storage/change-feed.md index 8418b94e..00cecc24 100644 --- a/docs/storage/change-feed.md +++ b/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. diff --git a/docs/storage/pages.md b/docs/storage/pages.md index 96322d36..dedcdd99 100644 --- a/docs/storage/pages.md +++ b/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. diff --git a/docs/transactions/wal-and-recovery.md b/docs/transactions/wal-and-recovery.md index dd16a697..45539d5b 100644 --- a/docs/transactions/wal-and-recovery.md +++ b/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 / - FORMAT properties: format=8, page-size= + FORMAT properties: format=8, page-size=; 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`). diff --git a/engine/src/main/java/io/hstore/engine/StorageEngine.java b/engine/src/main/java/io/hstore/engine/StorageEngine.java index 8c04f7da..36f7717d 100644 --- a/engine/src/main/java/io/hstore/engine/StorageEngine.java +++ b/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() { diff --git a/engine/src/main/java/io/hstore/engine/catalog/CatalogStore.java b/engine/src/main/java/io/hstore/engine/catalog/CatalogStore.java index 9d866c5f..ba0d870e 100644 --- a/engine/src/main/java/io/hstore/engine/catalog/CatalogStore.java +++ b/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 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); diff --git a/engine/src/main/java/io/hstore/engine/feed/ChangeFeed.java b/engine/src/main/java/io/hstore/engine/feed/ChangeFeed.java index 0782bb25..831bbffe 100644 --- a/engine/src/main/java/io/hstore/engine/feed/ChangeFeed.java +++ b/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 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,7 +403,7 @@ 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) { @@ -408,6 +411,10 @@ private Frame readLocked() { } } + 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()) { diff --git a/engine/src/main/java/io/hstore/engine/page/FileHandle.java b/engine/src/main/java/io/hstore/engine/page/FileHandle.java index 57961f73..86c09f07 100644 --- a/engine/src/main/java/io/hstore/engine/page/FileHandle.java +++ b/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 uninterrupted(Supplier 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(); diff --git a/engine/src/main/java/io/hstore/engine/page/MappedLongs.java b/engine/src/main/java/io/hstore/engine/page/MappedLongs.java index 2f32c32c..71dce66e 100644 --- a/engine/src/main/java/io/hstore/engine/page/MappedLongs.java +++ b/engine/src/main/java/io/hstore/engine/page/MappedLongs.java @@ -47,7 +47,8 @@ static MappedLongs open(Path path, int magic, int format, String what) { if (longs.fresh) { header.set(INT, 0, magic); header.set(INT, 4, format); - header.force(); + file.force(header); + FileHandle.syncDirectory(path.getParent()); } else if (header.get(INT, 0) != magic || header.get(INT, 4) != format) { file.close(); throw HStoreException.invalid(what + " " + path + " has an unknown format"); @@ -87,9 +88,7 @@ void put(long index, long value) { } void force() { - for (MemorySegment chunk : chunks) { - chunk.force(); - } + file.force(chunks); } private synchronized void mapThrough(int last) { 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 8d9689d3..785e06c7 100644 --- a/engine/src/main/java/io/hstore/engine/page/PageDirectory.java +++ b/engine/src/main/java/io/hstore/engine/page/PageDirectory.java @@ -28,7 +28,7 @@ static PageDirectory open(Path path) { entries.header(NEXT_OFFSET, 1L); } PageDirectory directory = new PageDirectory(entries); - directory.next = entries.header(NEXT_OFFSET); + directory.next = Math.max(1L, entries.header(NEXT_OFFSET)); return directory; } diff --git a/engine/src/main/java/io/hstore/engine/page/ReferenceJournal.java b/engine/src/main/java/io/hstore/engine/page/ReferenceJournal.java index 7726d63b..baef6fb8 100644 --- a/engine/src/main/java/io/hstore/engine/page/ReferenceJournal.java +++ b/engine/src/main/java/io/hstore/engine/page/ReferenceJournal.java @@ -32,7 +32,7 @@ private ReferenceJournal() { static void write(Path path, long lsn, ReferenceCounts counts, long[] cleared, int clearedCount) { boolean created = !Files.exists(path); - try (FileChannel channel = FileChannel.open(path, StandardOpenOption.CREATE, StandardOpenOption.WRITE, + try (FileChannel channel = FileHandle.channel(path, StandardOpenOption.CREATE, StandardOpenOption.WRITE, StandardOpenOption.TRUNCATE_EXISTING)) { Writer out = new Writer(channel); out.putInt(MAGIC).putInt(FORMAT).putLong(lsn).putInt(counts.changed()).putInt(clearedCount); @@ -46,7 +46,7 @@ static void write(Path path, long lsn, ReferenceCounts counts, long[] cleared, i throw HStoreException.io("cannot write reference journal " + path, e); } if (created) { - syncDirectory(path.getParent()); + FileHandle.syncDirectory(path.getParent()); } } @@ -86,7 +86,7 @@ static void clear(Path path) { if (!Files.exists(path)) { return; } - try (FileChannel channel = FileChannel.open(path, StandardOpenOption.WRITE)) { + try (FileChannel channel = FileHandle.channel(path, StandardOpenOption.WRITE)) { channel.truncate(0); channel.force(true); } catch (IOException e) { @@ -94,13 +94,6 @@ static void clear(Path path) { } } - private static void syncDirectory(Path directory) { - try (FileChannel channel = FileChannel.open(directory, StandardOpenOption.READ)) { - channel.force(true); - } catch (IOException _) { - } - } - private static final class Writer { private final FileChannel channel; private final ByteBuffer buffer = ByteBuffer.allocate(BUFFER).order(ByteOrder.LITTLE_ENDIAN); diff --git a/engine/src/main/java/io/hstore/engine/page/SegmentFile.java b/engine/src/main/java/io/hstore/engine/page/SegmentFile.java index 815a6409..c37be92c 100644 --- a/engine/src/main/java/io/hstore/engine/page/SegmentFile.java +++ b/engine/src/main/java/io/hstore/engine/page/SegmentFile.java @@ -39,7 +39,12 @@ private SegmentFile(int id, Path path, FileHandle file) { static SegmentFile open(int id, Path path) { try { - return new SegmentFile(id, path, FileHandle.open(path)); + boolean created = !Files.exists(path); + SegmentFile segment = new SegmentFile(id, path, FileHandle.open(path)); + if (created) { + FileHandle.syncDirectory(path.getParent()); + } + return segment; } catch (IOException e) { throw new UncheckedIOException("cannot open segment " + path, e); } diff --git a/engine/src/main/java/io/hstore/engine/wal/WriteAheadLog.java b/engine/src/main/java/io/hstore/engine/wal/WriteAheadLog.java index 9df854a9..bf8455c9 100644 --- a/engine/src/main/java/io/hstore/engine/wal/WriteAheadLog.java +++ b/engine/src/main/java/io/hstore/engine/wal/WriteAheadLog.java @@ -158,7 +158,7 @@ private void roll() throws IOException { } private void syncDirectory() { - 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); diff --git a/engine/src/test/java/io/hstore/engine/PowerLossTest.java b/engine/src/test/java/io/hstore/engine/PowerLossTest.java new file mode 100644 index 00000000..925cc800 --- /dev/null +++ b/engine/src/test/java/io/hstore/engine/PowerLossTest.java @@ -0,0 +1,513 @@ +// SPDX-FileCopyrightText: Metacog Labs +// SPDX-License-Identifier: PolyForm-Noncommercial-1.0.0 + +package io.hstore.engine; + +import io.hstore.engine.catalog.Branch; +import io.hstore.engine.catalog.Slot; +import io.hstore.engine.feed.CommitEvent; +import io.hstore.engine.page.PowerLoss; +import io.hstore.engine.topology.EdgeKind; +import io.hstore.engine.topology.Incidence; +import io.hstore.engine.tree.TreeVerifier; +import io.hstore.engine.txn.Durability; +import io.hstore.engine.txn.Snapshot; +import io.hstore.engine.txn.Transaction; +import io.hstore.engine.txn.WalMode; + +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.EnumSource; +import org.junit.jupiter.params.provider.MethodSource; + +import java.nio.ByteBuffer; +import java.nio.ByteOrder; +import java.nio.channels.FileChannel; +import java.nio.file.Path; +import java.nio.file.StandardOpenOption; +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.Random; +import java.util.stream.IntStream; +import java.util.stream.LongStream; +import java.util.stream.Stream; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class PowerLossTest { + + private static final int STEPS = 16; + private static final int CRASHES = 4; + private static final int SECOND_CRASH_ODDS = 80; + + @TempDir + Path directory; + + private int runs; + + private record Commit(String key, long node, long bulk, int members, long dropped) { + } + + private static final class History { + final List commits = new ArrayList<>(); + final List abandoned = new ArrayList<>(); + long edge; + int acknowledged; + int checkpointed; + } + + static Stream configurations() { + long seeds = Long.getLong("hstore.powerloss.seeds", 1); + return Stream.of(Durability.values()).flatMap(durability -> Stream.of(WalMode.values()).flatMap(mode -> + LongStream.rangeClosed(1, seeds).mapToObj(seed -> Arguments.of(durability, mode, seed)))); + } + + @ParameterizedTest + @MethodSource("configurations") + void powerCutsAtRandomEventsRecoverAPrefixOfTheCommits(Durability durability, WalMode mode, long seed) { + EngineOptions options = options(durability, mode); + long events = dryRun(options, seed); + Random random = new Random(seed * 31 + durability.ordinal() * 7 + mode.ordinal()); + for (int crash = 0; crash < CRASHES; crash++) { + long event = 1 + random.nextLong(events + 1); + PowerLoss.Tear tear = tear(random); + String context = "%s %s seed %d, power lost at event %d of %d, unsynced writes %s".formatted(durability, mode, seed, event, events, tear); + Path database = directory.resolve("run-" + runs++); + History history = new History(); + try { + try (PowerLoss loss = PowerLoss.watch(database)) { + loss.crashAt(event); + drive(database, options, loss, history, seed); + loss.restore(tear, random); + } + recoverTwice(database, options, history, durability, random, context); + } catch (RuntimeException failure) { + throw new AssertionError(context, failure); + } + } + } + + @Test + void aDatabaseCreatedJustBeforeAPowerLossStillOpens() { + EngineOptions options = options(Durability.SYNC, WalMode.PAGE_REFERENCES); + try (PowerLoss loss = PowerLoss.watch(directory)) { + StorageEngine engine = StorageEngine.open(directory, options); + engine.write(txn -> txn.createNode(EngineTest.NODE, "acknowledged")); + loss.powerOff(); + engine.halt(); + loss.restore(PowerLoss.Tear.NOTHING, new Random(1)); + } + try (StorageEngine engine = StorageEngine.open(directory, options)) { + assertTrue(engine.read(snapshot -> snapshot.resolve(0, EngineTest.NODE, "acknowledged")).isPresent()); + } + } + + @ParameterizedTest + @EnumSource(WalMode.class) + void pagesInASegmentCreatedSinceTheLastCheckpointSurvive(WalMode mode) { + EngineOptions options = options(Durability.SYNC, mode); + long edge; + try (PowerLoss loss = PowerLoss.watch(directory)) { + StorageEngine engine = StorageEngine.open(directory, options); + edge = engine.write(txn -> { + long created = txn.createEdge(EngineTest.EDGE, EdgeKind.SET); + txn.load(created, LongStream.range(0, 6_000).mapToObj(Incidence::of).toList()); + return created; + }); + loss.powerOff(); + engine.halt(); + loss.restore(PowerLoss.Tear.NOTHING, new Random(1)); + } + try (StorageEngine engine = StorageEngine.open(directory, options)) { + assertEquals(0, engine.recovery().discardedCommits()); + assertEquals(6_000, (long) engine.read(snapshot -> snapshot.requireEdge(edge).size())); + engine.verifyReferences(); + } + } + + @Test + void aPageDirectoryWhoseFirstPageNumberWasNeverForcedStillHandsOutNumbers() { + EngineOptions options = options(Durability.SYNC, WalMode.PAGE_REFERENCES); + try (PowerLoss loss = PowerLoss.watch(directory)) { + loss.crashOn(event -> event.kind() == PowerLoss.Kind.MAPPED_FORCE && event.file().endsWith("pages.ref")); + assertThrows(PowerLoss.Cut.class, () -> StorageEngine.open(directory, options)); + loss.restore(PowerLoss.Tear.NOTHING, new Random(1)); + } + try (StorageEngine engine = StorageEngine.open(directory, options)) { + engine.write(txn -> txn.createNode(EngineTest.NODE, "first")); + } + try (StorageEngine engine = StorageEngine.open(directory, options)) { + assertTrue(engine.read(snapshot -> snapshot.resolve(0, EngineTest.NODE, "first")).isPresent()); + engine.verifyReferences(); + } + } + + @ParameterizedTest + @EnumSource(WalMode.class) + void asyncCommitsAfterTheLastCheckpointAreLostTogether(WalMode mode) { + EngineOptions options = options(Durability.ASYNC, mode); + try (PowerLoss loss = PowerLoss.watch(directory)) { + StorageEngine engine = StorageEngine.open(directory, options); + for (int i = 0; i < 10; i++) { + String key = "n" + i; + engine.write(txn -> txn.createNode(EngineTest.NODE, key)); + if (i == 4) { + engine.checkpoint(); + } + } + loss.powerOff(); + engine.halt(); + loss.restore(PowerLoss.Tear.NOTHING, new Random(1)); + } + try (StorageEngine engine = StorageEngine.open(directory, options)) { + engine.read(snapshot -> { + assertEquals(5, snapshot.countOfType(0, EngineTest.NODE)); + IntStream.range(0, 5).forEach(i -> assertTrue(snapshot.resolve(0, EngineTest.NODE, "n" + i).isPresent())); + return null; + }); + engine.verifyReferences(); + } + } + + @ParameterizedTest + @EnumSource(value = PowerLoss.Tear.class, names = {"HALF_LAST", "FLIP_LAST"}) + void aTornOrCorruptCommitRecordDropsOnlyTheUnacknowledgedCommit(PowerLoss.Tear tear) { + EngineOptions options = options(Durability.SYNC, WalMode.PAGE_REFERENCES); + try (PowerLoss loss = PowerLoss.watch(directory)) { + StorageEngine engine = StorageEngine.open(directory, options); + engine.write(txn -> txn.createNode(EngineTest.NODE, "acknowledged")); + loss.crashOn(event -> event.kind() == PowerLoss.Kind.FORCE && event.file().toString().endsWith(".wal")); + assertThrows(PowerLoss.Cut.class, () -> engine.write(txn -> txn.createNode(EngineTest.NODE, "unacknowledged"))); + engine.halt(); + loss.restore(file -> file.toString().endsWith(".wal") ? tear : PowerLoss.Tear.ALL, new Random(1)); + } + reopenWithOnly("acknowledged", options); + } + + @ParameterizedTest + @EnumSource(value = PowerLoss.Tear.class, names = {"HALF_LAST", "FLIP_LAST"}) + void aCommitWhosePageWasTornIsDroppedEvenThoughItsLogRecordsSurvived(PowerLoss.Tear tear) { + EngineOptions options = options(Durability.SYNC, WalMode.PAGE_REFERENCES); + try (PowerLoss loss = PowerLoss.watch(directory)) { + StorageEngine engine = StorageEngine.open(directory, options); + engine.write(txn -> txn.createNode(EngineTest.NODE, "acknowledged")); + loss.crashOn(event -> event.kind() == PowerLoss.Kind.FORCE && event.file().toString().endsWith(".seg")); + assertThrows(PowerLoss.Cut.class, () -> engine.write(txn -> { + txn.createNode(EngineTest.NODE, "unacknowledged"); + long edge = txn.createEdge(EngineTest.EDGE, EdgeKind.SET); + txn.load(edge, LongStream.range(0, 6_000).mapToObj(Incidence::of).toList()); + return edge; + })); + engine.halt(); + loss.restore(file -> file.toString().endsWith(".seg") ? tear : PowerLoss.Tear.ALL, new Random(1)); + } + try (StorageEngine engine = StorageEngine.open(directory, options)) { + assertEquals(1, engine.recovery().discardedCommits()); + } + reopenWithOnly("acknowledged", options); + } + + private void reopenWithOnly(String key, EngineOptions options) { + try (StorageEngine engine = StorageEngine.open(directory, options)) { + engine.read(snapshot -> { + assertTrue(snapshot.resolve(0, EngineTest.NODE, key).isPresent()); + assertEquals(1, snapshot.countOfType(0, EngineTest.NODE)); + return null; + }); + engine.verifyReferences(); + engine.write(txn -> txn.createNode(EngineTest.NODE, "later")); + } + try (StorageEngine engine = StorageEngine.open(directory, options)) { + assertEquals(2L, (long) engine.read(snapshot -> snapshot.countOfType(0, EngineTest.NODE))); + } + } + + @Test + void aFeedFrameWhoseUnsyncedGenerationWasDamagedDoesNotHideCommittedEvents() throws Exception { + List frames = threeCommitsThenPowerLoss(); + try (FileChannel feed = FileChannel.open(feedSegment(), StandardOpenOption.READ, StandardOpenOption.WRITE)) { + ByteBuffer generation = ByteBuffer.allocate(1); + feed.read(generation, frames.get(2) + 10); + generation.put(0, (byte) (generation.get(0) ^ 0x5A)); + feed.write(generation.rewind(), frames.get(2) + 10); + } + assertFeedHoldsOneEventPerCommit(); + } + + @Test + void aZeroedFeedFrameBeforeASurvivingOneDoesNotRepeatEvents() throws Exception { + List frames = threeCommitsThenPowerLoss(); + try (FileChannel feed = FileChannel.open(feedSegment(), StandardOpenOption.WRITE)) { + feed.write(ByteBuffer.allocate(Math.toIntExact(frames.get(2) - frames.get(1))), frames.get(1)); + } + assertFeedHoldsOneEventPerCommit(); + } + + private List threeCommitsThenPowerLoss() throws Exception { + EngineOptions options = options(Durability.SYNC, WalMode.PAGE_REFERENCES); + try (PowerLoss loss = PowerLoss.watch(directory)) { + StorageEngine engine = StorageEngine.open(directory, options); + for (int i = 0; i < 3; i++) { + String key = "n" + i; + engine.write(txn -> txn.createNode(EngineTest.NODE, key)); + } + loss.powerOff(); + engine.halt(); + loss.restore(PowerLoss.Tear.ALL, new Random(1)); + } + List frames = new ArrayList<>(); + try (FileChannel feed = FileChannel.open(feedSegment(), StandardOpenOption.READ)) { + ByteBuffer length = ByteBuffer.allocate(Integer.BYTES).order(ByteOrder.LITTLE_ENDIAN); + for (long position = 0; position < feed.size(); position += 16 + length.getInt(0)) { + frames.add(position); + feed.read(length.clear(), position); + } + } + assertEquals(3, frames.size()); + return frames; + } + + private Path feedSegment() { + return directory.resolve("feed").resolve("%016x.feed".formatted(0)); + } + + private void assertFeedHoldsOneEventPerCommit() { + try (StorageEngine engine = StorageEngine.open(directory, options(Durability.SYNC, WalMode.PAGE_REFERENCES))) { + assertEquals(3L, (long) engine.read(snapshot -> snapshot.countOfType(0, EngineTest.NODE))); + assertEquals(List.of(1L, 2L, 3L), engine.feed().replay(0).map(CommitEvent::generation).toList()); + } + } + + private long dryRun(EngineOptions options, long seed) { + Path database = directory.resolve("run-" + runs++); + try (PowerLoss loss = PowerLoss.watch(database)) { + return drive(database, options, loss, new History(), seed); + } + } + + private static long drive(Path database, EngineOptions options, PowerLoss loss, History history, long seed) { + StorageEngine engine = null; + long events; + try { + engine = StorageEngine.open(database, options); + workload(engine, loss, history, seed); + } catch (Throwable failure) { + if (!loss.crashed()) { + throw failure; + } + } finally { + events = loss.events(); + loss.powerOff(); + if (engine != null) { + engine.halt(); + } + } + return events; + } + + private static void workload(StorageEngine engine, PowerLoss loss, History history, long seed) { + Random random = new Random(seed); + for (int step = 0; step < STEPS && !loss.crashed(); step++) { + int roll = step == 0 ? 0 : random.nextInt(100); + try { + if (roll < 50) { + commit(engine, loss, history, random, 0, false); + } else if (roll < 65) { + commit(engine, loss, history, random, random.nextBoolean() ? 300 : 5_000, false); + } else if (roll < 73) { + commit(engine, loss, history, random, 0, true); + } else if (roll < 80) { + abandon(engine, history, random.nextBoolean() ? 0 : 5_000); + } else if (roll < 92) { + engine.checkpoint(); + if (!loss.crashed()) { + history.checkpointed = history.acknowledged; + } + } else { + engine.compact(); + if (!loss.crashed()) { + history.checkpointed = history.acknowledged; + } + } + } catch (Throwable failure) { + if (!loss.crashed()) { + throw failure; + } + } + } + } + + private static void commit(StorageEngine engine, PowerLoss loss, History history, Random random, int members, boolean drop) { + int index = history.commits.size(); + List alive = alive(history.commits, index); + long victim = drop && !alive.isEmpty() ? alive.get(random.nextInt(alive.size())).bulk() : 0; + try (Transaction txn = engine.begin()) { + String key = "c-" + index; + long node = txn.createNode(EngineTest.NODE, key); + if (index == 0) { + history.edge = txn.createEdge(EngineTest.EDGE, EdgeKind.ORDERED); + } + txn.append(history.edge, Incidence.of(node)); + long bulk = 0; + if (members > 0) { + bulk = txn.createEdge(EngineTest.EDGE, EdgeKind.SET); + long first = 1_000_000L + index * 10_000L; + txn.load(bulk, LongStream.range(first, first + members).mapToObj(Incidence::of).toList()); + } + if (victim != 0) { + txn.delete(victim); + } + history.commits.add(new Commit(key, node, bulk, members, victim)); + txn.commit(); + } + if (!loss.crashed()) { + history.acknowledged = history.commits.size(); + } + } + + private static void abandon(StorageEngine engine, History history, int members) { + String key = "x-" + history.abandoned.size(); + history.abandoned.add(key); + try (Transaction txn = engine.begin()) { + txn.createNode(EngineTest.NODE, key); + if (members > 0) { + long bulk = txn.createEdge(EngineTest.EDGE, EdgeKind.SET); + txn.load(bulk, LongStream.range(5_000_000L, 5_000_000L + members).mapToObj(Incidence::of).toList()); + } + txn.abort(); + } + } + + private static List alive(List commits, int prefix) { + List alive = new ArrayList<>(); + for (Commit commit : commits.subList(0, prefix)) { + if (commit.dropped() != 0) { + alive.removeIf(earlier -> earlier.bulk() == commit.dropped()); + } + if (commit.bulk() != 0) { + alive.add(commit); + } + } + return alive; + } + + private static void recoverTwice(Path database, EngineOptions options, History history, Durability durability, Random random, + String context) { + PowerLoss.Tear tear = tear(random); + Random chance = new Random(random.nextLong()); + int observed = -1; + boolean acknowledged = false; + boolean checkpointed = false; + try (PowerLoss loss = PowerLoss.watch(database)) { + loss.crashOn(_ -> chance.nextInt(SECOND_CRASH_ODDS) == 0); + StorageEngine engine = null; + try { + engine = StorageEngine.open(database, options); + observed = verify(engine, history, durability, context); + engine.write(txn -> txn.createNode(EngineTest.NODE, "after")); + acknowledged = !loss.crashed(); + engine.checkpoint(); + checkpointed = !loss.crashed(); + } catch (Throwable failure) { + if (!loss.crashed()) { + throw failure; + } + } finally { + loss.powerOff(); + if (engine != null) { + engine.halt(); + } + } + loss.restore(tear, random); + } + String again = context + ", recovered and lost power again with unsynced writes " + tear; + try (StorageEngine engine = StorageEngine.open(database, options)) { + int recovered = verify(engine, history, durability, again); + if (observed >= 0) { + assertEquals(observed, recovered, again + ": commits seen after the first recovery"); + } + if (checkpointed || acknowledged && durability == Durability.SYNC) { + assertTrue(engine.read(snapshot -> snapshot.resolve(0, EngineTest.NODE, "after")).isPresent(), again + ": commit after recovery"); + } + } + } + + private static PowerLoss.Tear tear(Random random) { + return PowerLoss.Tear.values()[random.nextInt(PowerLoss.Tear.values().length)]; + } + + private static int verify(StorageEngine engine, History history, Durability durability, String context) { + int recovered = engine.read(snapshot -> { + int present = 0; + while (snapshot.resolve(0, EngineTest.NODE, "c-" + present).isPresent()) { + present++; + } + for (int i = present; i <= history.commits.size(); i++) { + assertTrue(snapshot.resolve(0, EngineTest.NODE, "c-" + i).isEmpty(), context + ": c-" + i + " survived without c-" + present); + } + for (String key : history.abandoned) { + assertTrue(snapshot.resolve(0, EngineTest.NODE, key).isEmpty(), context + ": abandoned " + key + " appeared"); + } + int after = snapshot.resolve(0, EngineTest.NODE, "after").isPresent() ? 1 : 0; + assertEquals(present + after, snapshot.countOfType(0, EngineTest.NODE), context + ": node count"); + assertEquals(LongStream.rangeClosed(1, present + after).boxed().toList(), + engine.feed().replay(0).map(CommitEvent::generation).toList(), context + ": change feed generations"); + return present; + }); + int floor = durability == Durability.SYNC ? history.acknowledged : history.checkpointed; + assertTrue(recovered >= floor, context + ": recovered " + recovered + " commits, " + floor + " were durable (" + + history.acknowledged + " acknowledged, " + history.checkpointed + " checkpointed)"); + assertTrue(recovered <= history.commits.size(), context + ": recovered " + recovered + " of " + history.commits.size() + " attempted"); + engine.read(snapshot -> { + List prefix = history.commits.subList(0, recovered); + if (recovered > 0) { + assertEquals(prefix.stream().map(Commit::node).toList(), + snapshot.requireEdge(history.edge).stream().map(Incidence::member).toList(), context + ": ordered edge"); + } + for (Commit commit : prefix) { + assertEquals(commit.node(), snapshot.resolve(0, EngineTest.NODE, commit.key()).orElseThrow(), context); + assertEquals(1, snapshot.degree(commit.node()), context + ": degree of " + commit.key()); + } + List alive = alive(history.commits, recovered); + for (Commit commit : history.commits) { + if (commit.bulk() == 0) { + continue; + } + if (alive.contains(commit)) { + assertEquals(commit.members(), snapshot.requireEdge(commit.bulk()).size(), context + ": members of " + commit.key()); + } else { + assertTrue(snapshot.edge(commit.bulk()).isEmpty(), context + ": edge of " + commit.key() + " should not exist"); + } + } + return null; + }); + verifyTrees(engine); + engine.verifyReferences(); + return recovered; + } + + private static void verifyTrees(StorageEngine engine) { + try (Snapshot snapshot = engine.snapshot()) { + TreeVerifier verifier = new TreeVerifier(engine.source()); + for (Branch branch : snapshot.catalog().branches().values()) { + if (branch.isActive()) { + for (Map.Entry root : branch.roots().roots().entrySet()) { + Slot slot = engine.slots().slot(root.getKey()); + verifier.verify(branch.roots().get(slot), slot.schema()); + } + } + } + } + } + + private static EngineOptions options(Durability durability, WalMode mode) { + return EngineTest.small().withDurability(durability).withWalMode(mode).withWalSegmentBytes(48 << 10) + .withCheckpointWalBytes(Long.MAX_VALUE).withMaintenanceInterval(Duration.ofHours(1)); + } +} diff --git a/engine/src/test/java/io/hstore/engine/page/PowerLoss.java b/engine/src/test/java/io/hstore/engine/page/PowerLoss.java new file mode 100644 index 00000000..c2a1ca38 --- /dev/null +++ b/engine/src/test/java/io/hstore/engine/page/PowerLoss.java @@ -0,0 +1,538 @@ +// SPDX-FileCopyrightText: Metacog Labs +// SPDX-License-Identifier: PolyForm-Noncommercial-1.0.0 + +package io.hstore.engine.page; + +import java.io.IOException; +import java.io.UncheckedIOException; +import java.lang.foreign.Arena; +import java.lang.foreign.MemorySegment; +import java.nio.ByteBuffer; +import java.nio.MappedByteBuffer; +import java.nio.channels.FileChannel; +import java.nio.channels.FileLock; +import java.nio.channels.ReadableByteChannel; +import java.nio.channels.WritableByteChannel; +import java.nio.file.Files; +import java.nio.file.OpenOption; +import java.nio.file.Path; +import java.nio.file.StandardOpenOption; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Random; +import java.util.Set; +import java.util.TreeMap; +import java.util.TreeSet; +import java.util.concurrent.ConcurrentHashMap; +import java.util.function.Function; +import java.util.function.Predicate; +import java.util.stream.Collectors; +import java.util.stream.Stream; + +public final class PowerLoss implements AutoCloseable { + + public enum Tear { + NOTHING, + ALL, + HALF_LAST, + FLIP_LAST, + RANDOM + } + + public enum Kind { + WRITE, + TRUNCATE, + FORCE, + MAPPED_FORCE, + DIRECTORY_FORCE + } + + public record Event(Kind kind, Path file) { + } + + public static final class Cut extends Error { + Cut(Event event) { + super("power lost before " + event.kind() + " of " + event.file(), null, false, false); + } + } + + @FunctionalInterface + private interface Io { + T run() throws IOException; + } + + private sealed interface Op { + } + + private record Write(long position, byte[] bytes) implements Op { + } + + private record Truncate(long size) implements Op { + } + + private record Image(byte[] bytes, List fresh) { + } + + private static final class Tracked { + private byte[] durable; + private final List pending = new ArrayList<>(); + private boolean entryDurable; + private boolean mapped; + private byte[] current; + + Tracked(byte[] durable, boolean entryDurable) { + this.durable = durable; + this.entryDurable = entryDurable; + } + } + + private static final int BLOCK = 4096; + private static final Map WATCHED = new ConcurrentHashMap<>(); + + static { + FileHandle.channels = new FileHandle.Channels() { + @Override + public FileChannel open(Path path, OpenOption... options) throws IOException { + PowerLoss owner = owner(path); + return owner == null ? FileChannel.open(path, options) : owner.open(key(path), options); + } + + @Override + public void forced(Path path) { + PowerLoss owner = owner(path); + if (owner != null) { + owner.mappingForced(key(path)); + } + } + }; + } + + private final Path root; + private final Set baseline; + private final Map files = new HashMap<>(); + private Predicate trigger = _ -> false; + private long events; + private boolean crashed; + private Set atCrash; + + private PowerLoss(Path root, Set baseline) { + this.root = root; + this.baseline = baseline; + } + + public static PowerLoss watch(Path root) { + PowerLoss loss = new PowerLoss(key(root), list(key(root))); + if (WATCHED.putIfAbsent(loss.root, loss) != null) { + throw new IllegalStateException(root + " is already watched"); + } + return loss; + } + + public synchronized void crashAt(long event) { + trigger = _ -> events == event; + } + + public synchronized void crashOn(Predicate when) { + trigger = when; + } + + public synchronized long events() { + return events; + } + + public synchronized boolean crashed() { + return crashed; + } + + public synchronized void powerOff() { + if (!crashed) { + cut(); + } + } + + public void restore(Tear tear, Random random) { + restore(_ -> tear, random); + } + + public synchronized void restore(Function tears, Random random) { + powerOff(); + try { + for (Path path : list(root)) { + if (!atCrash.contains(path)) { + Files.delete(path); + } + } + Map images = new TreeMap<>(); + for (Path path : new TreeSet<>(atCrash)) { + Tracked tracked = files.get(path); + Tear tear = tears.apply(path); + if (tracked == null) { + if (!baseline.contains(path) && !files.containsKey(path.resolveSibling(path.getFileName() + ".tmp"))) { + Files.write(path, new byte[0]); + } + } else if (!tracked.entryDurable && (tear == Tear.NOTHING || random.nextBoolean())) { + Files.deleteIfExists(path); + } else { + images.put(path, tracked.mapped ? mixed(tracked, tear, random) : replayed(tracked, tear, random)); + } + } + if (images.keySet().stream().anyMatch(path -> tears.apply(path) == Tear.RANDOM) && random.nextInt(3) == 0) { + List candidates = images.values().stream().filter(image -> !image.fresh().isEmpty()).toList(); + if (!candidates.isEmpty()) { + Image chosen = candidates.get(random.nextInt(candidates.size())); + long[] range = chosen.fresh().get(random.nextInt(chosen.fresh().size())); + flip(chosen.bytes(), range[0] + random.nextLong(range[1])); + } + } + for (Map.Entry entry : images.entrySet()) { + Files.write(entry.getKey(), entry.getValue().bytes()); + } + } catch (IOException e) { + throw new UncheckedIOException(e); + } + } + + @Override + public void close() { + WATCHED.remove(root, this); + } + + private static Image replayed(Tracked tracked, Tear tear, Random random) { + List ops = tracked.pending; + Image image = new Image(tracked.durable.clone(), new ArrayList<>()); + int last = ops.size() - 1; + int mode = random.nextInt(3); + int prefix = random.nextInt(ops.size() + 1); + for (int i = 0; i < ops.size(); i++) { + Op op = ops.get(i); + image = switch (tear) { + case NOTHING -> image; + case ALL -> apply(image, op, 1); + case HALF_LAST -> i < last ? apply(image, op, 1) : apply(image, op, 0.5); + case FLIP_LAST -> { + Image applied = apply(image, op, 1); + if (i == last && op instanceof Write(long position, byte[] bytes) && bytes.length > 0) { + flip(applied.bytes(), position + bytes.length / 2); + } + yield applied; + } + case RANDOM -> switch (mode) { + case 0 -> i < prefix ? apply(image, op, 1) : i == prefix ? apply(image, op, random.nextDouble()) : image; + case 1 -> switch (random.nextInt(3)) { + case 0 -> image; + case 1 -> apply(image, op, 1); + default -> apply(image, op, random.nextDouble()); + }; + default -> apply(image, op, 1); + }; + }; + } + return image; + } + + private static Image apply(Image image, Op op, double fraction) { + return switch (op) { + case Truncate(long size) -> new Image(size < image.bytes().length ? Arrays.copyOf(image.bytes(), Math.toIntExact(size)) : image.bytes(), + image.fresh()); + case Write(long position, byte[] bytes) -> { + int length = (int) Math.floor(bytes.length * fraction); + if (length == 0) { + yield image; + } + int end = Math.toIntExact(position + length); + byte[] grown = end > image.bytes().length ? Arrays.copyOf(image.bytes(), end) : image.bytes(); + System.arraycopy(bytes, 0, grown, Math.toIntExact(position), length); + image.fresh().add(new long[]{position, length}); + yield new Image(grown, image.fresh()); + } + }; + } + + private static Image mixed(Tracked tracked, Tear tear, Random random) { + byte[] durable = tracked.durable; + byte[] current = tracked.current == null ? durable : tracked.current; + return switch (tear) { + case NOTHING -> new Image(durable.clone(), List.of()); + case ALL, HALF_LAST, FLIP_LAST -> new Image(current.clone(), List.of()); + case RANDOM -> { + byte[] mixed = new byte[random.nextBoolean() ? durable.length : current.length]; + for (int block = 0; block < mixed.length; block += BLOCK) { + byte[] source = random.nextBoolean() ? durable : current; + int length = Math.max(0, Math.min(BLOCK, Math.min(mixed.length, source.length) - block)); + System.arraycopy(source, Math.min(block, source.length), mixed, block, length); + } + yield new Image(mixed, List.of()); + } + }; + } + + private static void flip(byte[] bytes, long position) { + if (position < bytes.length) { + bytes[Math.toIntExact(position)] ^= (byte) 0x5A; + } + } + + private synchronized FileChannel open(Path path, OpenOption... options) throws IOException { + if (Files.isDirectory(path)) { + return new Recorded(this, path, FileChannel.open(path, options), true); + } + if (crashed) { + return FileChannel.open(path, options); + } + boolean existed = Files.exists(path); + Tracked tracked = files.get(path); + if (!existed) { + files.put(path, new Tracked(new byte[0], false)); + } else if (tracked == null) { + files.put(path, new Tracked(Files.readAllBytes(path), true)); + } + if (existed && Arrays.asList(options).contains(StandardOpenOption.TRUNCATE_EXISTING)) { + event(Kind.TRUNCATE, path); + files.get(path).pending.add(new Truncate(0)); + } + return new Recorded(this, path, FileChannel.open(path, options), false); + } + + private synchronized int write(Path file, long position, ByteBuffer source, Io io) throws IOException { + if (crashed) { + return io.run(); + } + event(Kind.WRITE, file); + ByteBuffer copy = source.duplicate(); + int written = io.run(); + byte[] bytes = new byte[written]; + copy.get(bytes); + files.get(file).pending.add(new Write(position, bytes)); + return written; + } + + private synchronized void truncate(Path file, long size, Io io) throws IOException { + if (crashed) { + io.run(); + return; + } + event(Kind.TRUNCATE, file); + io.run(); + files.get(file).pending.add(new Truncate(size)); + } + + private synchronized void force(Path file, boolean directory, Io io) throws IOException { + if (crashed) { + io.run(); + return; + } + event(directory ? Kind.DIRECTORY_FORCE : Kind.FORCE, file); + io.run(); + if (directory) { + files.forEach((path, tracked) -> tracked.entryDurable |= file.equals(path.getParent())); + } else { + Tracked tracked = files.get(file); + Image image = new Image(tracked.durable, new ArrayList<>()); + for (Op op : tracked.pending) { + image = apply(image, op, 1); + } + tracked.durable = image.bytes().clone(); + tracked.pending.clear(); + } + } + + private synchronized void mapped(Path file) { + Tracked tracked = files.get(file); + if (tracked != null) { + tracked.mapped = true; + } + } + + private synchronized void mappingForced(Path file) { + if (crashed) { + return; + } + event(Kind.MAPPED_FORCE, file); + Tracked tracked = files.get(file); + try (FileChannel channel = FileChannel.open(file, StandardOpenOption.READ)) { + if (tracked.durable.length != channel.size()) { + tracked.durable = new byte[Math.toIntExact(channel.size())]; + } + ByteBuffer buffer = ByteBuffer.wrap(tracked.durable); + while (buffer.hasRemaining()) { + channel.read(buffer, buffer.position()); + } + } catch (IOException e) { + throw new UncheckedIOException(e); + } + } + + private void event(Kind kind, Path file) { + events++; + Event event = new Event(kind, file); + if (trigger.test(event)) { + cut(); + throw new Cut(event); + } + } + + private void cut() { + crashed = true; + atCrash = list(root); + files.forEach((path, tracked) -> { + if (tracked.mapped && Files.exists(path)) { + tracked.current = read(path); + } + }); + } + + private static byte[] read(Path file) { + try { + return Files.readAllBytes(file); + } catch (IOException e) { + throw new UncheckedIOException(e); + } + } + + private static Set list(Path root) { + if (!Files.exists(root)) { + return Set.of(); + } + try (Stream walk = Files.walk(root)) { + return walk.filter(Files::isRegularFile).map(PowerLoss::key).collect(Collectors.toSet()); + } catch (IOException e) { + throw new UncheckedIOException(e); + } + } + + private static Path key(Path path) { + return path.toAbsolutePath().normalize(); + } + + private static PowerLoss owner(Path path) { + Path absolute = key(path); + for (Map.Entry watched : WATCHED.entrySet()) { + if (absolute.startsWith(watched.getKey())) { + return watched.getValue(); + } + } + return null; + } + + private static final class Recorded extends FileChannel { + private final PowerLoss owner; + private final Path file; + private final FileChannel delegate; + private final boolean directory; + + Recorded(PowerLoss owner, Path file, FileChannel delegate, boolean directory) { + this.owner = owner; + this.file = file; + this.delegate = delegate; + this.directory = directory; + } + + @Override + public int read(ByteBuffer destination) throws IOException { + return delegate.read(destination); + } + + @Override + public long read(ByteBuffer[] destinations, int offset, int length) throws IOException { + return delegate.read(destinations, offset, length); + } + + @Override + public int read(ByteBuffer destination, long position) throws IOException { + return delegate.read(destination, position); + } + + @Override + public int write(ByteBuffer source) throws IOException { + return owner.write(file, delegate.position(), source, () -> delegate.write(source)); + } + + @Override + public long write(ByteBuffer[] sources, int offset, int length) throws IOException { + long total = 0; + for (int i = offset; i < offset + length; i++) { + total += write(sources[i]); + } + return total; + } + + @Override + public int write(ByteBuffer source, long position) throws IOException { + return owner.write(file, position, source, () -> delegate.write(source, position)); + } + + @Override + public long position() throws IOException { + return delegate.position(); + } + + @Override + public FileChannel position(long position) throws IOException { + delegate.position(position); + return this; + } + + @Override + public long size() throws IOException { + return delegate.size(); + } + + @Override + public FileChannel truncate(long size) throws IOException { + owner.truncate(file, size, () -> delegate.truncate(size)); + return this; + } + + @Override + public void force(boolean metaData) throws IOException { + owner.force(file, directory, () -> { + delegate.force(metaData); + return null; + }); + } + + @Override + public long transferTo(long position, long count, WritableByteChannel target) { + throw new UnsupportedOperationException(); + } + + @Override + public long transferFrom(ReadableByteChannel source, long position, long count) { + throw new UnsupportedOperationException(); + } + + @Override + public MappedByteBuffer map(MapMode mode, long position, long size) throws IOException { + if (mode == MapMode.READ_WRITE) { + owner.mapped(file); + } + return delegate.map(mode, position, size); + } + + @Override + public MemorySegment map(MapMode mode, long offset, long size, Arena arena) throws IOException { + if (mode == MapMode.READ_WRITE) { + owner.mapped(file); + } + return delegate.map(mode, offset, size, arena); + } + + @Override + public FileLock lock(long position, long size, boolean shared) throws IOException { + return delegate.lock(position, size, shared); + } + + @Override + public FileLock tryLock(long position, long size, boolean shared) throws IOException { + return delegate.tryLock(position, size, shared); + } + + @Override + protected void implCloseChannel() throws IOException { + delegate.close(); + } + } +} diff --git a/engine/src/test/java/io/hstore/engine/page/PowerLossModelTest.java b/engine/src/test/java/io/hstore/engine/page/PowerLossModelTest.java new file mode 100644 index 00000000..f27a1367 --- /dev/null +++ b/engine/src/test/java/io/hstore/engine/page/PowerLossModelTest.java @@ -0,0 +1,136 @@ +// SPDX-FileCopyrightText: Metacog Labs +// SPDX-License-Identifier: PolyForm-Noncommercial-1.0.0 + +package io.hstore.engine.page; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.nio.ByteBuffer; +import java.nio.channels.FileChannel; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.StandardOpenOption; +import java.util.Random; + +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +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 PowerLossModelTest { + + @TempDir + Path directory; + + @Test + void onlyForcedBytesSurvive() throws Exception { + Path file = directory.resolve("log"); + try (PowerLoss loss = PowerLoss.watch(directory)) { + try (FileHandle handle = FileHandle.open(file)) { + FileHandle.syncDirectory(directory); + write(handle, 0, "synced"); + handle.run(channel -> channel.force(false)); + write(handle, 6, "-lost"); + } + loss.restore(PowerLoss.Tear.NOTHING, new Random(1)); + } + assertEquals("synced", Files.readString(file)); + } + + @Test + void theLastUnsyncedWriteCanBeCutInHalfOrFlipped() throws Exception { + Path half = directory.resolve("half"); + Path flipped = directory.resolve("flipped"); + for (PowerLoss.Tear tear : new PowerLoss.Tear[]{PowerLoss.Tear.HALF_LAST, PowerLoss.Tear.FLIP_LAST}) { + Path file = tear == PowerLoss.Tear.HALF_LAST ? half : flipped; + try (PowerLoss loss = PowerLoss.watch(directory)) { + try (FileHandle handle = FileHandle.open(file)) { + FileHandle.syncDirectory(directory); + write(handle, 0, "first"); + write(handle, 5, "12345678"); + } + loss.restore(tear, new Random(1)); + } + } + assertEquals("first1234", Files.readString(half)); + byte[] expected = "first12345678".getBytes(StandardCharsets.UTF_8); + expected[9] ^= 0x5A; + assertArrayEquals(expected, Files.readAllBytes(flipped)); + } + + @Test + void aFileWhoseDirectoryWasNeverSyncedDisappears() throws Exception { + Path file = directory.resolve("new"); + try (PowerLoss loss = PowerLoss.watch(directory)) { + try (FileHandle handle = FileHandle.open(file)) { + write(handle, 0, "data"); + handle.run(channel -> channel.force(false)); + } + loss.restore(PowerLoss.Tear.NOTHING, new Random(1)); + } + assertFalse(Files.exists(file)); + } + + @Test + void aFileRenamedAfterItWasForcedKeepsItsContents() throws Exception { + Path temporary = directory.resolve("manifest.tmp"); + Path target = directory.resolve("manifest"); + Path unsynced = directory.resolve("written-elsewhere"); + try (PowerLoss loss = PowerLoss.watch(directory)) { + try (FileChannel channel = FileHandle.channel(temporary, StandardOpenOption.CREATE, StandardOpenOption.WRITE)) { + channel.write(ByteBuffer.wrap("generation".getBytes(StandardCharsets.UTF_8))); + channel.force(true); + } + Files.move(temporary, target); + FileHandle.syncDirectory(directory); + Files.writeString(unsynced, "never forced"); + loss.restore(PowerLoss.Tear.NOTHING, new Random(1)); + } + assertEquals("generation", Files.readString(target)); + assertEquals("", Files.readString(unsynced)); + } + + @Test + void mappedFilesKeepWhatTheirLastForceSaw() throws Exception { + Path file = directory.resolve("pages.dir"); + try (PowerLoss loss = PowerLoss.watch(directory)) { + PageDirectory addresses = PageDirectory.open(file); + FileHandle.syncDirectory(directory); + addresses.put(1, 11); + addresses.force(); + addresses.put(2, 22); + addresses.close(); + loss.restore(PowerLoss.Tear.NOTHING, new Random(1)); + } + PageDirectory reopened = PageDirectory.open(file); + assertEquals(11, reopened.get(1)); + assertEquals(0, reopened.get(2)); + reopened.close(); + } + + @Test + void theChosenEventCutsThePowerAndLaterWritesAreNotRecorded() throws Exception { + Path file = directory.resolve("log"); + try (PowerLoss loss = PowerLoss.watch(directory)) { + loss.crashOn(event -> event.kind() == PowerLoss.Kind.FORCE); + try (FileHandle handle = FileHandle.open(file)) { + FileHandle.syncDirectory(directory); + write(handle, 0, "unsynced"); + assertThrows(PowerLoss.Cut.class, () -> handle.run(channel -> channel.force(false))); + assertTrue(loss.crashed()); + write(handle, 8, "-after"); + handle.run(channel -> channel.force(false)); + } + loss.restore(PowerLoss.Tear.NOTHING, new Random(1)); + } + assertEquals("", Files.readString(file)); + } + + private static void write(FileHandle handle, long position, String text) throws Exception { + ByteBuffer buffer = ByteBuffer.wrap(text.getBytes(StandardCharsets.UTF_8)); + handle.run(channel -> channel.write(buffer, position)); + } +}