From 52bd78a10cd6b9c2016ff8f36f5fef423af8b097 Mon Sep 17 00:00:00 2001 From: venkat1701 Date: Wed, 14 Oct 2026 11:52:29 +0530 Subject: [PATCH 1/6] test(engine): record writes and syncs per file so a test can cut the power Every file the engine writes or syncs is now opened through FileHandle.channel, and memory-mapped files are forced through FileHandle.force. Both go through FileHandle.channels, a package-private static that defaults to FileChannel::open and only tests replace. The catalog, the reference journal and the WAL and feed directory syncs used FileChannel.open directly; CatalogStore and ReferenceJournal now share FileHandle.syncDirectory instead of their own copies. PowerLoss, a test helper, installs itself there for one directory. It records every write, truncation and force per file, every force of a mapping (pages.dir, pages.ref) and of a directory, can throw at a chosen event, and then rewrites the directory as a power failure could leave it: only forced bytes, all bytes, the last unsynced write cut in half or with a flipped byte, or a random mix. Files created since their directory was last forced may vanish. --- .../hstore/engine/catalog/CatalogStore.java | 12 +- .../io/hstore/engine/feed/ChangeFeed.java | 2 +- .../io/hstore/engine/page/FileHandle.java | 34 +- .../io/hstore/engine/page/MappedLongs.java | 6 +- .../hstore/engine/page/ReferenceJournal.java | 13 +- .../io/hstore/engine/wal/WriteAheadLog.java | 2 +- .../java/io/hstore/engine/page/PowerLoss.java | 538 ++++++++++++++++++ .../engine/page/PowerLossModelTest.java | 136 +++++ 8 files changed, 716 insertions(+), 27 deletions(-) create mode 100644 engine/src/test/java/io/hstore/engine/page/PowerLoss.java create mode 100644 engine/src/test/java/io/hstore/engine/page/PowerLossModelTest.java 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..9f957a5f 100644 --- a/engine/src/main/java/io/hstore/engine/feed/ChangeFeed.java +++ b/engine/src/main/java/io/hstore/engine/feed/ChangeFeed.java @@ -210,7 +210,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); 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..a98ccc02 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,7 @@ 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); } 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 +87,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/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/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/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)); + } +} From 9f12c74bdb9db6327b9943f076d7043cd81fa0ef Mon Sep 17 00:00:00 2001 From: venkat1701 Date: Wed, 14 Oct 2026 11:55:04 +0530 Subject: [PATCH 2/6] fix(engine): write FORMAT durably when a database is created verifyFormat wrote FORMAT with Files.newOutputStream and never forced it or the directory. A power loss soon after a database was created could leave FORMAT empty while the WAL already held acknowledged commits. The next open then read no format property, took it for format 1 and refused the database ("uses storage format 1"). FORMAT is now written to FORMAT.tmp, forced, renamed over FORMAT and the directory forced, as the catalog files already are. It happens once per database. --- docs/transactions/wal-and-recovery.md | 3 +- .../java/io/hstore/engine/StorageEngine.java | 15 ++++++- .../java/io/hstore/engine/PowerLossTest.java | 43 +++++++++++++++++++ 3 files changed, 58 insertions(+), 3 deletions(-) create mode 100644 engine/src/test/java/io/hstore/engine/PowerLossTest.java diff --git a/docs/transactions/wal-and-recovery.md b/docs/transactions/wal-and-recovery.md index dd16a697..2c56c57b 100644 --- a/docs/transactions/wal-and-recovery.md +++ b/docs/transactions/wal-and-recovery.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) 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/test/java/io/hstore/engine/PowerLossTest.java b/engine/src/test/java/io/hstore/engine/PowerLossTest.java new file mode 100644 index 00000000..3be7a4ec --- /dev/null +++ b/engine/src/test/java/io/hstore/engine/PowerLossTest.java @@ -0,0 +1,43 @@ +// SPDX-FileCopyrightText: Metacog Labs +// SPDX-License-Identifier: PolyForm-Noncommercial-1.0.0 + +package io.hstore.engine; + +import io.hstore.engine.page.PowerLoss; +import io.hstore.engine.txn.Durability; +import io.hstore.engine.txn.WalMode; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.nio.file.Path; +import java.time.Duration; +import java.util.Random; + +import static org.junit.jupiter.api.Assertions.assertTrue; + +class PowerLossTest { + + @TempDir + Path directory; + + @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()); + } + } + + 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)); + } +} From 26a7fa8d84243c091dc56667db579e846feb281c Mon Sep 17 00:00:00 2001 From: venkat1701 Date: Wed, 14 Oct 2026 11:57:31 +0530 Subject: [PATCH 3/6] fix(engine): force the segments directory when a data file is created Nothing ever forced data/segments. A new segment file, and pages.dir and pages.ref themselves, could vanish after a power loss even though their contents had been forced, because the directory entry was never made durable. With PAGE_REFERENCES a commit whose pages sat in such a segment was acknowledged after pages.sync and wal.sync, but after the power loss recovery found its pages missing and discarded it and every commit after it. SegmentFile.open now forces the directory when it creates the file, and MappedLongs does the same when it creates pages.dir or pages.ref. That is one directory sync per new segment (256 MiB by default) and two per database. --- docs/storage/pages.md | 5 +++- .../io/hstore/engine/page/MappedLongs.java | 1 + .../io/hstore/engine/page/SegmentFile.java | 7 ++++- .../java/io/hstore/engine/PowerLossTest.java | 29 +++++++++++++++++++ 4 files changed, 40 insertions(+), 2 deletions(-) diff --git a/docs/storage/pages.md b/docs/storage/pages.md index 96322d36..4e615ea5 100644 --- a/docs/storage/pages.md +++ b/docs/storage/pages.md @@ -196,6 +196,9 @@ writes each logged address back into the directory before it checks or replays t 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. +`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 `/data/segments/pages.ref` (`page/ReferenceCounts.java`) has the same layout as the directory: a 64-byte @@ -274,7 +277,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/engine/src/main/java/io/hstore/engine/page/MappedLongs.java b/engine/src/main/java/io/hstore/engine/page/MappedLongs.java index a98ccc02..71dce66e 100644 --- a/engine/src/main/java/io/hstore/engine/page/MappedLongs.java +++ b/engine/src/main/java/io/hstore/engine/page/MappedLongs.java @@ -48,6 +48,7 @@ static MappedLongs open(Path path, int magic, int format, String what) { header.set(INT, 0, magic); header.set(INT, 4, format); 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"); 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/test/java/io/hstore/engine/PowerLossTest.java b/engine/src/test/java/io/hstore/engine/PowerLossTest.java index 3be7a4ec..b07c0f6a 100644 --- a/engine/src/test/java/io/hstore/engine/PowerLossTest.java +++ b/engine/src/test/java/io/hstore/engine/PowerLossTest.java @@ -4,16 +4,22 @@ package io.hstore.engine; import io.hstore.engine.page.PowerLoss; +import io.hstore.engine.topology.EdgeKind; +import io.hstore.engine.topology.Incidence; import io.hstore.engine.txn.Durability; 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.EnumSource; import java.nio.file.Path; import java.time.Duration; import java.util.Random; +import java.util.stream.LongStream; +import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertTrue; class PowerLossTest { @@ -36,6 +42,29 @@ void aDatabaseCreatedJustBeforeAPowerLossStillOpens() { } } + @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(); + } + } + 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)); From 9d44e20944535df5dd6e7f21b1e0ab8c1d5f865c Mon Sep 17 00:00:00 2001 From: venkat1701 Date: Wed, 14 Oct 2026 12:00:26 +0530 Subject: [PATCH 4/6] fix(engine): read a page directory's next number 0 as 1 MappedLongs forces the header of a new pages.dir with only its magic and format. PageDirectory.open sets the next page number to 1 after that and the first checkpoint forces it. If the power went out in between, the directory came back with next number 0, which PageDirectory.allocate handed out, and every checkpoint from then on failed in PageId.pack with "page number out of range: 0". The database could no longer write a page. Page number 0 is never used, so open now reads a stored 0 as 1. That also repairs a directory already left in this state. --- docs/storage/pages.md | 4 +++- .../io/hstore/engine/page/PageDirectory.java | 2 +- .../java/io/hstore/engine/PowerLossTest.java | 18 ++++++++++++++++++ 3 files changed, 22 insertions(+), 2 deletions(-) diff --git a/docs/storage/pages.md b/docs/storage/pages.md index 4e615ea5..dedcdd99 100644 --- a/docs/storage/pages.md +++ b/docs/storage/pages.md @@ -194,7 +194,9 @@ 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. 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/test/java/io/hstore/engine/PowerLossTest.java b/engine/src/test/java/io/hstore/engine/PowerLossTest.java index b07c0f6a..26df337a 100644 --- a/engine/src/test/java/io/hstore/engine/PowerLossTest.java +++ b/engine/src/test/java/io/hstore/engine/PowerLossTest.java @@ -20,6 +20,7 @@ import java.util.stream.LongStream; 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 { @@ -65,6 +66,23 @@ void pagesInASegmentCreatedSinceTheLastCheckpointSurvive(WalMode mode) { } } + @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(); + } + } + 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)); From f894ecfeec6b7b13102aa59c8469c5846f30d6c1 Mon Sep 17 00:00:00 2001 From: venkat1701 Date: Wed, 14 Oct 2026 12:58:34 +0530 Subject: [PATCH 5/6] fix(engine): reject feed frames whose header disagrees with their payload A feed frame is [length][crc32c of payload][generation][payload], so the CRC does not cover the length or the generation. When power was lost with unsynced feed frames, two things got past ChangeFeed.restore: * A run of zeros, where a frame never reached the disk but a later one did, reads as frames of length 0, CRC 0 and generation 0 (the CRC of an empty payload is 0). restore set lastGeneration back to 0, so recovery re-appended the Feed records of every replayed commit after events that were already there, and the feed listed those commits twice. * A damaged generation field passed the check. A frame of generation 3 came back as generation 5898243, recovery skipped the Feed records of the replayed commits as already present, and truncateAfter(3) then dropped the bad frame, so the event of commit 3, which had been acknowledged, was gone for good. FeedCodec always writes the generation first in the payload, which the CRC does cover. A frame is now valid only if its payload is not empty and starts with the header's generation, and restore stops at a frame that is not newer than the one before it. The file format is unchanged. --- docs/storage/change-feed.md | 2 +- .../io/hstore/engine/feed/ChangeFeed.java | 9 ++- .../java/io/hstore/engine/PowerLossTest.java | 63 +++++++++++++++++++ 3 files changed, 72 insertions(+), 2 deletions(-) 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/engine/src/main/java/io/hstore/engine/feed/ChangeFeed.java b/engine/src/main/java/io/hstore/engine/feed/ChangeFeed.java index 9f957a5f..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(); } @@ -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/test/java/io/hstore/engine/PowerLossTest.java b/engine/src/test/java/io/hstore/engine/PowerLossTest.java index 26df337a..52823dc0 100644 --- a/engine/src/test/java/io/hstore/engine/PowerLossTest.java +++ b/engine/src/test/java/io/hstore/engine/PowerLossTest.java @@ -3,6 +3,7 @@ package io.hstore.engine; +import io.hstore.engine.feed.CommitEvent; import io.hstore.engine.page.PowerLoss; import io.hstore.engine.topology.EdgeKind; import io.hstore.engine.topology.Incidence; @@ -14,8 +15,14 @@ import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.EnumSource; +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.Random; import java.util.stream.LongStream; @@ -83,6 +90,62 @@ void aPageDirectoryWhoseFirstPageNumberWasNeverForcedStillHandsOutNumbers() { } } + @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 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)); From e7489507142267402ac45d65f1aa589e1f865dd7 Mon Sep 17 00:00:00 2001 From: venkat1701 Date: Wed, 14 Oct 2026 12:29:08 +0530 Subject: [PATCH 6/6] test(engine): cut the power at random I/O events and check recovery PowerLossTest runs a seeded workload of small commits, bulk loads that spill and write pages at commit, deletes, aborted transactions, checkpoints and compactions, for both durability levels and both WAL modes. It cuts the power at four random I/O events per configuration with a random tear of the unsynced writes, recovers, cuts the power again somewhere in recovery and the commit and checkpoint after it, and recovers again. Every time the surviving commits must be a prefix of the attempted ones, holding every acknowledged commit under sync and every checkpointed one under async, with no aborted work, one feed event per commit, the same prefix after both recoveries, and trees and reference counts that pass TreeVerifier and verifyReferences. Four smaller tests pin the cases from the issue: async commits after the last checkpoint are lost together, a Commit record cut in half or with a flipped byte loses only the unacknowledged commit, and a torn page discards its commit although the log records survived. -Dhstore.powerloss.seeds=N runs more seeds. --- docs/transactions/wal-and-recovery.md | 39 +- .../java/io/hstore/engine/PowerLossTest.java | 360 ++++++++++++++++++ 2 files changed, 398 insertions(+), 1 deletion(-) diff --git a/docs/transactions/wal-and-recovery.md b/docs/transactions/wal-and-recovery.md index 2c56c57b..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). @@ -354,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/test/java/io/hstore/engine/PowerLossTest.java b/engine/src/test/java/io/hstore/engine/PowerLossTest.java index 52823dc0..925cc800 100644 --- a/engine/src/test/java/io/hstore/engine/PowerLossTest.java +++ b/engine/src/test/java/io/hstore/engine/PowerLossTest.java @@ -3,17 +3,24 @@ 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; @@ -23,8 +30,11 @@ 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; @@ -32,9 +42,57 @@ 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); @@ -90,6 +148,86 @@ void aPageDirectoryWhoseFirstPageNumberWasNeverForcedStillHandsOutNumbers() { } } + @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(); @@ -146,6 +284,228 @@ private void assertFeedHoldsOneEventPerCommit() { } } + 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));