diff --git a/docs/transactions/wal-and-recovery.md b/docs/transactions/wal-and-recovery.md index 60f684f..3af29e5 100644 --- a/docs/transactions/wal-and-recovery.md +++ b/docs/transactions/wal-and-recovery.md @@ -70,7 +70,7 @@ The checksum covers the LSN, type, transaction id and payload. Embedding the LSN **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. -**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. +**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. **Read.** `read(from)` iterates frames starting at LSN `from` across segments. When fewer than 25 header bytes remain in a segment, the reader treats it as the clean end of that segment and continues with the next one. A *damaged* frame is one whose length is out of range, whose embedded LSN differs from the expected one, whose body is short, or whose CRC does not match. What happens next depends on where it is: 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 af9c72b..197f8c4 100644 --- a/engine/src/main/java/io/hstore/engine/wal/WriteAheadLog.java +++ b/engine/src/main/java/io/hstore/engine/wal/WriteAheadLog.java @@ -69,7 +69,7 @@ private void restore() { } Map.Entry last = segments.lastEntry(); long validEnd = last.getKey(); - for (Positioned positioned : (Iterable) () -> new Reader(last.getKey())) { + for (Positioned positioned : (Iterable) () -> new Reader(last.getKey(), false)) { validEnd = positioned.next(); } activeStart = last.getKey(); @@ -184,7 +184,7 @@ public long bytesAppended() { } public Stream read(long from) { - return StreamSupport.stream(Spliterators.spliteratorUnknownSize(new Reader(from), + return StreamSupport.stream(Spliterators.spliteratorUnknownSize(new Reader(from, true), Spliterator.ORDERED | Spliterator.NONNULL), false); } @@ -240,8 +240,12 @@ private final class Reader implements Iterator { private long lsn; private Positioned next; private boolean finished; + private final ByteBuffer header = ByteBuffer.allocate(HEADER); + private final boolean decode; + private ByteBuffer body = ByteBuffer.allocate(4096); - Reader(long from) { + Reader(long from, boolean decode) { + this.decode = decode; Long floor = segments.floorKey(from); this.files = segments.tailMap(floor == null ? Long.MIN_VALUE : floor, true).entrySet().iterator(); this.lsn = from; @@ -293,7 +297,7 @@ private boolean openNext() { private Positioned readFrame() { try { - ByteBuffer header = ByteBuffer.allocate(HEADER); + header.clear(); if (!readFully(header, lsn - segmentStart)) { return header.position() == 0 ? null : damaged("frame header is truncated"); } @@ -306,7 +310,11 @@ private Positioned readFrame() { if (payload < 0 || payload > MAX_RECORD || frameLsn != lsn) { return damaged("frame header is invalid"); } - ByteBuffer body = ByteBuffer.allocate(HEADER - 8 + payload); + int size = HEADER - 8 + payload; + if (body.capacity() < size) { + body = ByteBuffer.allocate(Math.max(size, body.capacity() * 2)); + } + body.clear().limit(size); body.put(header.position(8)); if (!readFully(body, lsn - segmentStart + HEADER)) { return damaged("frame is truncated"); @@ -315,7 +323,7 @@ private Positioned readFrame() { if (Checksums.crc32c(frame) != crc) { return damaged("frame checksum does not match"); } - WalRecord record = WalRecord.readPayload(type, txnId, ByteCursor.over(frame).position(HEADER - 8)); + WalRecord record = decode ? WalRecord.readPayload(type, txnId, ByteCursor.over(frame).position(HEADER - 8)) : null; long start = lsn; lsn += HEADER + payload; return new Positioned(start, lsn, record); diff --git a/engine/src/test/java/io/hstore/engine/wal/WriteAheadLogTest.java b/engine/src/test/java/io/hstore/engine/wal/WriteAheadLogTest.java new file mode 100644 index 0000000..f7d7589 --- /dev/null +++ b/engine/src/test/java/io/hstore/engine/wal/WriteAheadLogTest.java @@ -0,0 +1,106 @@ +package io.hstore.engine.wal; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.io.IOException; +import java.nio.channels.FileChannel; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.StandardOpenOption; +import java.util.ArrayList; +import java.util.HexFormat; +import java.util.List; +import java.util.random.RandomGenerator; +import java.util.random.RandomGeneratorFactory; +import java.util.stream.Stream; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class WriteAheadLogTest { + + private static final long SEGMENT_BYTES = 64 * 1024; + + @TempDir + Path directory; + + private static List records(int count, long seed) { + RandomGenerator random = RandomGeneratorFactory.of("L64X128MixRandom").create(seed); + List records = new ArrayList<>(); + for (int i = 0; i < count; i++) { + byte[] bytes = new byte[random.nextInt(random.nextInt(10) == 0 ? 20_000 : 200)]; + random.nextBytes(bytes); + records.add(switch (i % 4) { + case 0 -> new WalRecord.Begin(i, random.nextInt(4), random.nextLong(1_000_000)); + case 1 -> new WalRecord.Page(i, random.nextLong(), bytes); + case 2 -> new WalRecord.Feed(i, bytes); + default -> new WalRecord.Commit(i, i, random.nextLong(), random.nextLong(1 << 30), random.nextLong(1 << 30)); + }); + } + return records; + } + + private static String describe(WalRecord record) { + return switch (record) { + case WalRecord.Page(long txn, long page, byte[] image) -> "page " + txn + " " + page + " " + HexFormat.of().formatHex(image); + case WalRecord.Feed(long txn, byte[] payload) -> "feed " + txn + " " + HexFormat.of().formatHex(payload); + default -> record.toString(); + }; + } + + private static List replay(WriteAheadLog wal) { + try (Stream stream = wal.read(0)) { + return stream.map(positioned -> describe(positioned.record())).toList(); + } + } + + @Test + void reopenedLogReplaysRecordsOfMixedSizesAcrossSegments() { + List written = records(3_000, 7); + long end; + try (WriteAheadLog wal = WriteAheadLog.open(directory, SEGMENT_BYTES)) { + written.forEach(wal::append); + end = wal.end(); + assertTrue(wal.segmentCount() > 3, "segments " + wal.segmentCount()); + } + try (WriteAheadLog wal = WriteAheadLog.open(directory, SEGMENT_BYTES)) { + assertEquals(end, wal.end()); + assertEquals(written.stream().map(WriteAheadLogTest::describe).toList(), replay(wal)); + } + } + + @Test + void aTornFinalFrameIsDroppedAndEverythingBeforeItKept() throws IOException { + List written = records(400, 11); + long lastStart; + try (WriteAheadLog wal = WriteAheadLog.open(directory, SEGMENT_BYTES)) { + written.subList(0, written.size() - 1).forEach(wal::append); + lastStart = wal.end(); + wal.append(written.getLast()); + } + Path last; + try (Stream files = Files.list(directory)) { + last = files.filter(path -> path.toString().endsWith(".wal")).max(Path::compareTo).orElseThrow(); + } + try (FileChannel channel = FileChannel.open(last, StandardOpenOption.WRITE)) { + channel.truncate(channel.size() - 3); + } + try (WriteAheadLog wal = WriteAheadLog.open(directory, SEGMENT_BYTES)) { + assertEquals(lastStart, wal.end()); + assertEquals(written.subList(0, written.size() - 1).stream().map(WriteAheadLogTest::describe).toList(), replay(wal)); + } + } + + @Test + void aSmallFrameAfterALargeOneReadsBackIntact() { + List written = List.of(new WalRecord.Feed(1, new byte[]{1, 2, 3}), new WalRecord.Page(2, 9, new byte[30_000]), + new WalRecord.Feed(3, new byte[]{4, 5}), new WalRecord.Commit(4, 4, 5, 6, 7), new WalRecord.Feed(5, new byte[0])); + try (WriteAheadLog wal = WriteAheadLog.open(directory, 1 << 20)) { + written.forEach(wal::append); + } + try (WriteAheadLog wal = WriteAheadLog.open(directory, 1 << 20)) { + assertEquals(written.stream().map(WriteAheadLogTest::describe).toList(), replay(wal)); + } + } +}