From 2062c607c0ce50487c221ecd21cf58bdb0233ec1 Mon Sep 17 00:00:00 2001 From: venkat1701 Date: Wed, 14 Oct 2026 10:54:18 +0530 Subject: [PATCH 1/2] perf(engine): force the write-ahead log outside its append lock WriteAheadLog.sync() held the WAL lock for the whole fsync. A checkpoint syncs the log after appending its Checkpoint record, and a committing writer appends while it holds the commit lock, so every writer queued behind that fsync. The same happened to appends waiting on a group commit's sync under durability = sync. sync() now takes a separate sync lock, reads and clears the dirty flag under the WAL lock, and forces the active segment after releasing it. Syncs queue on the sync lock, so a sync that finds nothing dirty only returns after the force that covered its records. If the segment rolls while it is being forced, the roll has already forced and closed it, so the failed force on the closed file is ignored. truncateBefore() also deletes old segment files after releasing the lock. The new WAL_FORCE crash point sits between releasing the lock and the force. CheckpointConcurrencyTest commits from another thread while a checkpoint is held there, crashes checkpoints at CHECKPOINT_CATALOG, WAL_FORCE and CHECKPOINT_JOURNAL under three writers, and runs four writers through back-to-back checkpoints. --- docs/transactions/wal-and-recovery.md | 10 +- .../java/io/hstore/engine/StorageEngine.java | 2 +- .../java/io/hstore/engine/txn/CrashPoint.java | 3 +- .../io/hstore/engine/wal/WriteAheadLog.java | 68 +++++- .../engine/CheckpointConcurrencyTest.java | 225 ++++++++++++++++++ .../hstore/engine/wal/WriteAheadLogTest.java | 31 +++ 6 files changed, 320 insertions(+), 19 deletions(-) create mode 100644 engine/src/test/java/io/hstore/engine/CheckpointConcurrencyTest.java diff --git a/docs/transactions/wal-and-recovery.md b/docs/transactions/wal-and-recovery.md index 09ab2c5..c11f36d 100644 --- a/docs/transactions/wal-and-recovery.md +++ b/docs/transactions/wal-and-recovery.md @@ -76,7 +76,7 @@ The checksum covers the LSN, type, transaction id and payload. Embedding the LSN **Append.** `append(List)` encodes the whole list into one buffer and issues it as one positional write under the WAL lock, so the records of one call are contiguous. Before writing, the log rolls to a new segment if the active segment is non-empty and the batch would push it past `walSegmentBytes`. A batch is never split across segments; a batch larger than a segment gets an oversized segment of its own. Rolling `force(true)`s and closes the previous segment, creates the new file, and forces the `wal/segments` directory (`syncDirectory`; filesystems that cannot open a directory for sync are logged at `DEBUG` and skipped) so that the new segment's directory entry survives a power loss. -**Sync.** `sync()` calls `FileChannel.force(false)` on the active segment if anything was written since the last sync (`dirty`). Rolled segments were forced when they were closed, so one call makes the entire log durable. +**Sync.** `sync()` calls `FileChannel.force(false)` on the active segment if anything was written since the last sync (`dirty`). Rolled segments were forced when they were closed, so one call makes the entire log durable. The force runs outside the WAL lock: `sync()` takes a separate sync lock, reads and clears `dirty` and picks the active segment under the WAL lock, releases it, reaches `WAL_FORCE` and forces. Appends carry on during the force, and a record appended after `dirty` was cleared sets it again, so the next `sync()` forces once more. Because syncs queue on the sync lock, a `sync()` that finds `dirty` clear can only return after the force that covered its records has finished. If the log rolls while a force is running, the roll forces and closes that segment itself, so a force that then fails on the closed file is not an error. Any other failure sets `dirty` again and is thrown. **Open.** `restore()` lists `*.wal`, scans only the **last** segment frame by frame, truncates it to the end of the last valid frame and forces it. Recovery therefore always appends after a clean frame boundary. The scan checks each frame's header and CRC-32C but doesn't decode the records, and a reader reuses one header and one body buffer, growing the body only for a larger frame, so opening doesn't allocate per frame. @@ -89,7 +89,7 @@ The checksum covers the LSN, type, transaction id and payload. Embedding the LSN Only the tail can legitimately be torn, because every earlier segment was forced when the log rolled past it. Damage anywhere else means data loss or tampering, and the engine refuses to open rather than recover a silently shortened history. -**Truncate.** `truncateBefore(lsn)` deletes every segment whose *successor* starts at or before `lsn`, never the active segment. The segment containing `lsn` is always kept. +**Truncate.** `truncateBefore(lsn)` deletes every segment whose *successor* starts at or before `lsn`, never the active segment. The segment containing `lsn` is always kept. The segments are taken out of the log's list under the WAL lock and their files are deleted after it is released. ## 4. Record types @@ -300,7 +300,7 @@ LOG: [recovery] recovered generation 158: replayed 0 commits from lsn 156,490, ## 9. Checkpoints -A checkpoint writes the trees of one published generation to pages and makes that generation the starting point of the next recovery. `Checkpointer.checkpoint()` takes the commit lock for one short step only (step 9), so commits keep running while it writes and syncs. +A checkpoint writes the trees of one published generation to pages and makes that generation the starting point of the next recovery. `Checkpointer.checkpoint()` takes the commit lock for one short step only (step 9), and none of its file syncs hold the commit lock or the WAL lock, so commits keep running while it writes and syncs. 1. **Target.** `g = transactions.current()`, the newest published generation, and a new `Epoch` for this attempt. 2. **Write the pages.** A `Materializer` walks every branch's roots in `g` and writes each pending node it reaches, children first: `pages.allocate`, `pages.write`, then the frozen node is admitted to the node cache and the in-memory node is stamped with its stored reference and the epoch. A node stamped by the last published epoch was written by the previous checkpoint, or by a commit that logged its tree by reference, so its stored reference is reused and the subtree below it is not visited. The images carry `g` as their creation epoch. The result is `g` with every root stored. @@ -311,10 +311,10 @@ A checkpoint writes the trees of one published generation to pages and makes tha 7. `pages.checkpoint()` again, then `pages.journal(lsn)`: every count changed since the last checkpoint and every released directory entry, as absolute values, go to `pages.ref.journal` behind a header holding `lsn` and with a CRC-32C at the end, and the file is forced. `feed.sync()` (forces the active feed segment; earlier feed segments were forced when the feed rolled). 8. `CHECKPOINT_CATALOG`, then `catalog.publish(CatalogImage(written g, List.of(), segments, lsn, feed.size()))`. Retained history is not written into the image. The image is framed as `i32 magic 0x54414348`, `i32 crc32c(payload)`, `i64 length`, payload; written to `generations/%016x.cat.tmp`, forced, atomically renamed, the directory forced; then `manifest` is rewritten the same way. The three newest `.cat` files are kept; `loadLatest()` falls back to the newest valid file if the manifest is missing or names a corrupt file. 9. `TransactionManager.checkpointed(written g, epoch)` publishes the epoch and forgets the commit positions of generations before `g`. Then, holding the commit lock and with the group committer drained, it swaps the written pages into the head generation: every node there that this checkpoint wrote is replaced with its stored reference, and parents that only changed because of that are copied (`Materializer.rebound`). Head, `current` and head's history entry become the rebound generation. Last, it recounts the write buffer and wakes commits that were waiting for room. -10. `wal.append(Checkpoint(0, g, lsn))`, `wal.sync()`, `wal.truncateBefore(lsn)`. +10. `wal.append(Checkpoint(0, g, lsn))`, `wal.sync()`, `wal.truncateBefore(lsn)`. The sync and the file deletions run outside the WAL lock (section 3), so a commit that appends meanwhile does not wait for them. 11. `CHECKPOINT_JOURNAL`, then `pages.applyJournal()` writes the journaled values into `pages.ref` and `pages.dir`, forces both and truncates the journal to zero bytes. The checkpoint remembers `lsn` and resumes the counter. -Because the journal holds absolute values, applying it twice gives the same files. If the process dies after step 8 but before step 11 is finished, the next open finds a journal whose LSN matches the published catalog and applies it again. If it dies before step 8, the journal's LSN doesn't match and it is ignored: `pages.ref` and `pages.dir` were not touched, so they still match the previous catalog, and the pages this checkpoint wrote have addresses but no counts, so the next open hands them to its first checkpoint, which finds nothing that reaches them and marks them dead. If step 10 or 11 fails and the engine keeps running, the next checkpoint applies the old journal before writing a new one, and a failed step 8 marks `pages.ref` for rebuilding. `ReferenceCountTest` crashes at `CHECKPOINT_CATALOG` and at `CHECKPOINT_JOURNAL` and checks the counts after reopening, including after a checkpoint that failed at step 11 and a next one that crashed before step 8. `DeferredWritesTest` crashes a checkpoint at `CHECKPOINT_PAGES`, `CHECKPOINT_CATALOG` and `CHECKPOINT_JOURNAL` while another thread keeps committing, and checks that every acknowledged commit is there after reopening and that the counts match. +Because the journal holds absolute values, applying it twice gives the same files. If the process dies after step 8 but before step 11 is finished, the next open finds a journal whose LSN matches the published catalog and applies it again. If it dies before step 8, the journal's LSN doesn't match and it is ignored: `pages.ref` and `pages.dir` were not touched, so they still match the previous catalog, and the pages this checkpoint wrote have addresses but no counts, so the next open hands them to its first checkpoint, which finds nothing that reaches them and marks them dead. If step 10 or 11 fails and the engine keeps running, the next checkpoint applies the old journal before writing a new one, and a failed step 8 marks `pages.ref` for rebuilding. `ReferenceCountTest` crashes at `CHECKPOINT_CATALOG` and at `CHECKPOINT_JOURNAL` and checks the counts after reopening, including after a checkpoint that failed at step 11 and a next one that crashed before step 8. `DeferredWritesTest` crashes a checkpoint at `CHECKPOINT_PAGES`, `CHECKPOINT_CATALOG` and `CHECKPOINT_JOURNAL` while another thread keeps committing, and checks that every acknowledged commit is there after reopening and that the counts match. `CheckpointConcurrencyTest` does the same with three writers at `CHECKPOINT_CATALOG`, `WAL_FORCE` and `CHECKPOINT_JOURNAL`, reopening twice. A crash at `WAL_FORCE` can also land in a commit batch that the checkpoint drains in step 9, so there the writers may fail too, and the only commits allowed to come back without an acknowledgement are the ones in flight. It also commits from another thread while the checkpoint is held at `WAL_FORCE`, which must finish without waiting for the checkpoint and still be there after reopening, and it runs four writers through back-to-back checkpoints, each checking that its commit is visible right after it returns. The checkpoint LSN is the end of `g`'s `Commit` record, not the position of the `Checkpoint` record. The next recovery starts there, replays the commits published after `g`, and skips the `Checkpoint` record when it reaches it. Triggers: diff --git a/engine/src/main/java/io/hstore/engine/StorageEngine.java b/engine/src/main/java/io/hstore/engine/StorageEngine.java index ea0120d..7fbdab9 100644 --- a/engine/src/main/java/io/hstore/engine/StorageEngine.java +++ b/engine/src/main/java/io/hstore/engine/StorageEngine.java @@ -83,7 +83,7 @@ private StorageEngine(Path directory, EngineOptions options, FileChannel lockCha CatalogImage checkpoint = Recovery.discover(catalog); this.pages = PageStore.open(directory.resolve("data").resolve("segments"), options.pageSize(), options.pagesPerSegment(), options.maxOpenSegmentFiles(), checkpoint.segments(), checkpoint.checkpointLsn()); - this.wal = WriteAheadLog.open(directory.resolve("wal").resolve("segments"), options.walSegmentBytes()); + this.wal = WriteAheadLog.open(directory.resolve("wal").resolve("segments"), options.walSegmentBytes(), options.faults()); this.slots = new SlotRegistry(options.slots()); FeedCodec codec = new FeedCodec(slots); this.feed = ChangeFeed.open(directory.resolve("feed"), codec, options.walSegmentBytes()); diff --git a/engine/src/main/java/io/hstore/engine/txn/CrashPoint.java b/engine/src/main/java/io/hstore/engine/txn/CrashPoint.java index caa6c57..3458ccf 100644 --- a/engine/src/main/java/io/hstore/engine/txn/CrashPoint.java +++ b/engine/src/main/java/io/hstore/engine/txn/CrashPoint.java @@ -15,7 +15,8 @@ public enum CrashPoint { COMPACTION_RETIRE, CHECKPOINT_PAGES, CHECKPOINT_CATALOG, - CHECKPOINT_JOURNAL; + CHECKPOINT_JOURNAL, + WAL_FORCE; @FunctionalInterface public interface Injector { 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 bf8455c..e2a0158 100644 --- a/engine/src/main/java/io/hstore/engine/wal/WriteAheadLog.java +++ b/engine/src/main/java/io/hstore/engine/wal/WriteAheadLog.java @@ -7,6 +7,7 @@ import io.hstore.engine.page.ByteCursor; import io.hstore.engine.page.Checksums; import io.hstore.engine.page.FileHandle; +import io.hstore.engine.txn.CrashPoint; import java.io.IOException; import java.io.UncheckedIOException; @@ -42,7 +43,9 @@ public record Positioned(long lsn, long next, WalRecord record) { private final Path directory; private final long segmentBytes; + private final CrashPoint.Injector faults; private final ReentrantLock lock = new ReentrantLock(); + private final ReentrantLock syncing = new ReentrantLock(); private final NavigableMap segments = new TreeMap<>(); private final AtomicLong bytesAppended = new AtomicLong(); private FileHandle active; @@ -50,13 +53,18 @@ public record Positioned(long lsn, long next, WalRecord record) { private long end; private boolean dirty; - private WriteAheadLog(Path directory, long segmentBytes) { + private WriteAheadLog(Path directory, long segmentBytes, CrashPoint.Injector faults) { this.directory = directory; this.segmentBytes = segmentBytes; + this.faults = faults; } public static WriteAheadLog open(Path directory, long segmentBytes) { - WriteAheadLog log = new WriteAheadLog(directory, segmentBytes); + return open(directory, segmentBytes, CrashPoint.Injector.NONE); + } + + public static WriteAheadLog open(Path directory, long segmentBytes, CrashPoint.Injector faults) { + WriteAheadLog log = new WriteAheadLog(directory, segmentBytes, faults); log.restore(); return log; } @@ -166,14 +174,47 @@ private void syncDirectory() { } public void sync() { - lock.lock(); + syncing.lock(); try { - if (dirty && active != null) { - active.run(channel -> channel.force(false)); + FileHandle target; + lock.lock(); + try { + if (!dirty || active == null) { + return; + } + target = active; dirty = false; + } finally { + lock.unlock(); } + force(target); + } finally { + syncing.unlock(); + } + } + + private void force(FileHandle target) { + try { + faults.reach(CrashPoint.WAL_FORCE); + target.run(channel -> channel.force(false)); } catch (IOException e) { - throw HStoreException.io("write-ahead log sync failed", e); + if (unforced(target)) { + throw HStoreException.io("write-ahead log sync failed", e); + } + } catch (RuntimeException | Error failure) { + unforced(target); + throw failure; + } + } + + private boolean unforced(FileHandle target) { + lock.lock(); + try { + if (target != active) { + return false; + } + dirty = true; + return true; } finally { lock.unlock(); } @@ -198,22 +239,25 @@ public Stream read(long from) { } public void truncateBefore(long lsn) { + List obsolete = new ArrayList<>(); lock.lock(); try { - List obsolete = new ArrayList<>(); for (Map.Entry entry : segments.entrySet()) { Long next = segments.higherKey(entry.getKey()); if (next != null && next <= lsn && entry.getKey() != activeStart) { - obsolete.add(entry.getKey()); + obsolete.add(entry.getValue()); } } - for (long start : obsolete) { - Files.deleteIfExists(segments.remove(start)); + segments.values().removeAll(obsolete); + } finally { + lock.unlock(); + } + try { + for (Path path : obsolete) { + Files.deleteIfExists(path); } } catch (IOException e) { throw new UncheckedIOException(e); - } finally { - lock.unlock(); } } diff --git a/engine/src/test/java/io/hstore/engine/CheckpointConcurrencyTest.java b/engine/src/test/java/io/hstore/engine/CheckpointConcurrencyTest.java new file mode 100644 index 0000000..6acae4d --- /dev/null +++ b/engine/src/test/java/io/hstore/engine/CheckpointConcurrencyTest.java @@ -0,0 +1,225 @@ +// SPDX-FileCopyrightText: Metacog Labs +// SPDX-License-Identifier: PolyForm-Noncommercial-1.0.0 + +package io.hstore.engine; + +import io.hstore.engine.topology.EdgeKind; +import io.hstore.engine.txn.CrashPoint; +import io.hstore.engine.txn.Durability; + +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.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; +import java.util.stream.Collectors; +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 CheckpointConcurrencyTest { + + private static final int EDGES = 24; + + @TempDir + Path directory; + + private static EngineOptions options() { + return EngineTest.small().withCheckpointWalBytes(Long.MAX_VALUE).withMaintenanceInterval(Duration.ofHours(1)); + } + + private static long[] edges(StorageEngine engine) { + return engine.write(txn -> LongStream.range(0, EDGES).map(_ -> txn.createEdge(EngineTest.EDGE, EdgeKind.SET)).toArray()); + } + + private static void insert(StorageEngine engine, long edge, long member) { + engine.write(txn -> { + txn.insert(edge, member); + return null; + }); + } + + @ParameterizedTest + @EnumSource(value = CrashPoint.class, names = {"WAL_FORCE"}) + void aCommitFinishesWhileTheCheckpointIsAtThisStep(CrashPoint point) throws Exception { + AtomicReference opened = new AtomicReference<>(); + AtomicReference outcome = new AtomicReference<>(); + AtomicBoolean armed = new AtomicBoolean(); + Thread test = Thread.currentThread(); + long[] edges = new long[1]; + EngineOptions options = options().withDurability(Durability.ASYNC).withFaults(reached -> { + if (reached == point && Thread.currentThread() == test && armed.compareAndSet(true, false)) { + CompletableFuture commit = CompletableFuture.runAsync(() -> insert(opened.get(), edges[0], 7_777)); + try { + commit.get(5, TimeUnit.SECONDS); + outcome.set("committed"); + } catch (TimeoutException e) { + outcome.set("the commit waited for the checkpoint"); + } catch (InterruptedException | ExecutionException e) { + outcome.set(e); + } + } + }); + try (StorageEngine engine = StorageEngine.open(directory, options)) { + opened.set(engine); + edges[0] = edges(engine)[0]; + for (long member = 0; member < 50; member++) { + insert(engine, edges[0], member); + } + armed.set(true); + engine.checkpoint(); + assertEquals("committed", outcome.get()); + assertEquals(51L, (long) engine.read(snapshot -> snapshot.cardinality(edges[0]))); + engine.checkpoint(); + engine.verifyReferences(); + } + try (StorageEngine engine = StorageEngine.open(directory, options())) { + boolean kept = engine.read(snapshot -> snapshot.requireEdge(edges[0]).memberIds().anyMatch(member -> member == 7_777)); + assertTrue(kept); + engine.verifyReferences(); + } + } + + @ParameterizedTest + @EnumSource(value = CrashPoint.class, names = {"CHECKPOINT_CATALOG", "WAL_FORCE", "CHECKPOINT_JOURNAL"}) + void aCrashAtThisCheckpointStepLosesNoAcknowledgedCommit(CrashPoint point) throws Exception { + AtomicBoolean armed = new AtomicBoolean(); + Thread test = Thread.currentThread(); + EngineOptions options = options().withFaults(reached -> { + if (reached == point && Thread.currentThread() == test && armed.compareAndSet(true, false)) { + throw new CrashPoint.SimulatedCrash(point); + } + }); + StorageEngine engine = StorageEngine.open(directory, options); + long[] edges = edges(engine); + Map> acknowledged = new ConcurrentHashMap<>(); + AtomicBoolean writing = new AtomicBoolean(true); + List writers = new ArrayList<>(); + ConcurrentLinkedQueue failures = new ConcurrentLinkedQueue<>(); + Set inFlight = ConcurrentHashMap.newKeySet(); + for (int w = 0; w < 3; w++) { + long first = (w + 1) * 1_000_000L; + writers.add(Thread.ofPlatform().start(() -> { + for (long member = first; writing.get(); member++) { + long edge = edges[(int) (member % EDGES)]; + inFlight.add(member); + try { + insert(engine, edge, member); + } catch (Throwable e) { + failures.add(e); + return; + } + acknowledged.computeIfAbsent(edge, _ -> ConcurrentHashMap.newKeySet()).add(member); + inFlight.remove(member); + } + })); + } + try { + for (int i = 0; i < 10; i++) { + engine.checkpoint(); + } + armed.set(true); + assertThrows(CrashPoint.SimulatedCrash.class, engine::checkpoint); + } finally { + writing.set(false); + for (Thread writer : writers) { + writer.join(); + } + } + failures.forEach(failure -> assertTrue(point == CrashPoint.WAL_FORCE + && (failure instanceof CrashPoint.SimulatedCrash || failure instanceof HStoreException), "writer failed: " + failure)); + engine.halt(); + Map> first = null; + for (int round = 0; round < 2; round++) { + try (StorageEngine recovered = StorageEngine.open(directory, options())) { + Map> found = new HashMap<>(); + for (long edge : edges) { + Set members = recovered.read(snapshot -> snapshot.requireEdge(edge).memberIds().boxed().collect(Collectors.toSet())); + Set expected = acknowledged.getOrDefault(edge, Set.of()); + assertTrue(members.containsAll(expected), "edge " + edge + " lost acknowledged members after reopen " + round); + members.stream().filter(member -> !expected.contains(member)).forEach(member -> + assertTrue(inFlight.contains(member), "edge " + edge + " gained member " + member + " that was never written")); + found.put(edge, members); + } + if (first == null) { + first = found; + } else { + assertEquals(first, found); + } + recovered.verifyReferences(); + } + } + } + + @Test + void writersCommittingThroughBackToBackCheckpointsKeepEveryCommit() throws Exception { + EngineOptions options = options().withDurability(Durability.ASYNC); + StorageEngine engine = StorageEngine.open(directory, options); + long[] edges = edges(engine); + Map> acknowledged = new ConcurrentHashMap<>(); + ConcurrentLinkedQueue problems = new ConcurrentLinkedQueue<>(); + AtomicBoolean writing = new AtomicBoolean(true); + List writers = new ArrayList<>(); + for (int w = 0; w < 4; w++) { + long first = (w + 1) * 1_000_000L; + writers.add(Thread.ofPlatform().start(() -> { + for (long member = first; writing.get() && member < first + 2_000; member++) { + long edge = edges[(int) (member % EDGES)]; + long added = member; + insert(engine, edge, added); + acknowledged.computeIfAbsent(edge, _ -> ConcurrentHashMap.newKeySet()).add(added); + if (!engine.read(snapshot -> snapshot.requireEdge(edge).memberIds().anyMatch(m -> m == added))) { + problems.add("member " + added + " of edge " + edge + " vanished right after its commit"); + } + } + })); + } + int checkpoints = 0; + try { + while (writers.stream().anyMatch(Thread::isAlive)) { + engine.checkpoint(); + checkpoints++; + } + } finally { + writing.set(false); + for (Thread writer : writers) { + writer.join(); + } + } + assertTrue(checkpoints > 5, "only " + checkpoints + " checkpoints overlapped the writers"); + assertTrue(problems.isEmpty(), problems.stream().limit(5).toList().toString()); + assertMembers(engine, edges, acknowledged); + engine.checkpoint(); + engine.verifyReferences(); + engine.verifyChangeLog(); + engine.close(); + try (StorageEngine reopened = StorageEngine.open(directory, options)) { + assertMembers(reopened, edges, acknowledged); + reopened.verifyReferences(); + } + } + + private static void assertMembers(StorageEngine engine, long[] edges, Map> acknowledged) { + for (long edge : edges) { + Set members = engine.read(snapshot -> snapshot.requireEdge(edge).memberIds().boxed().collect(Collectors.toSet())); + assertEquals(acknowledged.getOrDefault(edge, Set.of()), members, "edge " + edge); + } + } +} diff --git a/engine/src/test/java/io/hstore/engine/wal/WriteAheadLogTest.java b/engine/src/test/java/io/hstore/engine/wal/WriteAheadLogTest.java index 2ce6949..88efeb3 100644 --- a/engine/src/test/java/io/hstore/engine/wal/WriteAheadLogTest.java +++ b/engine/src/test/java/io/hstore/engine/wal/WriteAheadLogTest.java @@ -3,6 +3,8 @@ package io.hstore.engine.wal; +import io.hstore.engine.txn.CrashPoint; + import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; @@ -14,6 +16,8 @@ import java.util.ArrayList; import java.util.HexFormat; import java.util.List; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; import java.util.random.RandomGenerator; import java.util.random.RandomGeneratorFactory; import java.util.stream.Stream; @@ -106,4 +110,31 @@ void aSmallFrameAfterALargeOneReadsBackIntact() { assertEquals(written.stream().map(WriteAheadLogTest::describe).toList(), replay(wal)); } } + + @Test + void aSyncWhoseSegmentRollsAwayDuringTheForceSucceeds() { + List written = new ArrayList<>(records(40, 17)); + AtomicReference opened = new AtomicReference<>(); + AtomicBoolean armed = new AtomicBoolean(); + WalRecord large = new WalRecord.Feed(99, new byte[(int) SEGMENT_BYTES]); + long end; + try (WriteAheadLog wal = WriteAheadLog.open(directory, SEGMENT_BYTES, point -> { + if (point == CrashPoint.WAL_FORCE && armed.compareAndSet(true, false)) { + opened.get().append(large); + } + })) { + opened.set(wal); + written.forEach(wal::append); + int segments = wal.segmentCount(); + armed.set(true); + wal.sync(); + written.add(large); + assertEquals(segments + 1, wal.segmentCount()); + end = wal.end(); + } + try (WriteAheadLog wal = WriteAheadLog.open(directory, SEGMENT_BYTES)) { + assertEquals(end, wal.end()); + assertEquals(written.stream().map(WriteAheadLogTest::describe).toList(), replay(wal)); + } + } } From 04c7bfc72ac4fb8c860c310066e163542b4732a5 Mon Sep 17 00:00:00 2001 From: venkat1701 Date: Wed, 14 Oct 2026 11:30:57 +0530 Subject: [PATCH 2/2] perf(engine): rebind the head mostly outside the commit lock After a checkpoint publishes, TransactionManager.checkpointed swapped the pages it wrote into the head generation while holding the commit lock. That walk visits every pending node the checkpoint did not write, and no commit could start until it finished. Materializer.Rebinding now remembers its result for each pending node it visits. The checkpoint runs one pass over the head without the lock, then takes the commit lock, drains the group committer as before, and runs a second pass over the current head. The second pass looks up nodes the first one visited and only walks what commits added in between, so a commit that lands between the passes is kept. A node a commit stamps with the epoch after the first pass saw it stays pending until the next checkpoint, which reuses its page. The new CHECKPOINT_REBIND crash point sits between the two passes. CheckpointConcurrencyTest commits while a checkpoint is held there and checks the commit survives the rebind and a reopen, and crashes there under three writers. --- docs/storage/persistent-tree.md | 2 +- docs/transactions/wal-and-recovery.md | 6 +- .../io/hstore/engine/tree/Materializer.java | 92 ++++++++++++------- .../java/io/hstore/engine/txn/CrashPoint.java | 1 + .../hstore/engine/txn/TransactionManager.java | 18 ++-- .../engine/CheckpointConcurrencyTest.java | 4 +- 6 files changed, 74 insertions(+), 49 deletions(-) diff --git a/docs/storage/persistent-tree.md b/docs/storage/persistent-tree.md index a0e1d4d..4acca08 100644 --- a/docs/storage/persistent-tree.md +++ b/docs/storage/persistent-tree.md @@ -183,7 +183,7 @@ A `WriteScope` built with a `PageSink` (every `Transaction` has one) can spill c A checkpoint that fails before it publishes leaves its epoch unpublished, so nothing trusts the pages it wrote, and the next checkpoint writes those nodes again. -**Rebinding.** Commits keep publishing while a checkpoint writes `g`, so by the time it publishes, the head generation may be several commits past `g` and still hold the pending nodes the checkpoint wrote. `Materializer.rebound(ref, schema, epoch)` walks the head generation's roots under the commit lock and replaces every node stamped by that epoch with its stored or inline reference. A pending parent whose children changed this way is copied with the new references; one whose children didn't change is kept as it is. Later commits then build on stored references, and the written nodes can be collected once no retained generation or open snapshot still holds them. +**Rebinding.** Commits keep publishing while a checkpoint writes `g`, so by the time it publishes, the head generation may be several commits past `g` and still hold the pending nodes the checkpoint wrote. `Materializer.Rebinding` walks the head generation's roots and replaces every node stamped by that epoch with its stored or inline reference. A pending parent whose children changed this way is copied with the new references; one whose children didn't change is kept as it is. The walk remembers its result for every pending node it visits, keyed by identity. The checkpoint walks once without the commit lock, then again under it over the head that commits have moved on to since, and the second walk only descends into nodes the first one never saw ([checkpoints](../transactions/wal-and-recovery.md#9-checkpoints)). A node a commit stamps with the epoch after the first walk saw it stays pending until the next checkpoint, which reuses its page. Later commits then build on stored references, and the written nodes can be collected once no retained generation or open snapshot still holds them. ### Inline trees diff --git a/docs/transactions/wal-and-recovery.md b/docs/transactions/wal-and-recovery.md index c11f36d..ec0c366 100644 --- a/docs/transactions/wal-and-recovery.md +++ b/docs/transactions/wal-and-recovery.md @@ -300,7 +300,7 @@ LOG: [recovery] recovered generation 158: replayed 0 commits from lsn 156,490, ## 9. Checkpoints -A checkpoint writes the trees of one published generation to pages and makes that generation the starting point of the next recovery. `Checkpointer.checkpoint()` takes the commit lock for one short step only (step 9), and none of its file syncs hold the commit lock or the WAL lock, so commits keep running while it writes and syncs. +A checkpoint writes the trees of one published generation to pages and makes that generation the starting point of the next recovery. `Checkpointer.checkpoint()` takes the commit lock for one short step only (the end of step 9), and none of its file syncs hold the commit lock or the WAL lock, so commits keep running while it writes and syncs. 1. **Target.** `g = transactions.current()`, the newest published generation, and a new `Epoch` for this attempt. 2. **Write the pages.** A `Materializer` walks every branch's roots in `g` and writes each pending node it reaches, children first: `pages.allocate`, `pages.write`, then the frozen node is admitted to the node cache and the in-memory node is stamped with its stored reference and the epoch. A node stamped by the last published epoch was written by the previous checkpoint, or by a commit that logged its tree by reference, so its stored reference is reused and the subtree below it is not visited. The images carry `g` as their creation epoch. The result is `g` with every root stored. @@ -310,11 +310,11 @@ A checkpoint writes the trees of one published generation to pages and makes tha 6. **Count.** `ReferenceCounter.checkpointed(previous, written)` compares the previously checkpointed generation with the one just written and updates the counts ([maintenance](../storage/maintenance.md#reference-counts)). Dead pages that no retained or pinned generation can reach are then released ([maintenance](../storage/maintenance.md#releasing-pages)). 7. `pages.checkpoint()` again, then `pages.journal(lsn)`: every count changed since the last checkpoint and every released directory entry, as absolute values, go to `pages.ref.journal` behind a header holding `lsn` and with a CRC-32C at the end, and the file is forced. `feed.sync()` (forces the active feed segment; earlier feed segments were forced when the feed rolled). 8. `CHECKPOINT_CATALOG`, then `catalog.publish(CatalogImage(written g, List.of(), segments, lsn, feed.size()))`. Retained history is not written into the image. The image is framed as `i32 magic 0x54414348`, `i32 crc32c(payload)`, `i64 length`, payload; written to `generations/%016x.cat.tmp`, forced, atomically renamed, the directory forced; then `manifest` is rewritten the same way. The three newest `.cat` files are kept; `loadLatest()` falls back to the newest valid file if the manifest is missing or names a corrupt file. -9. `TransactionManager.checkpointed(written g, epoch)` publishes the epoch and forgets the commit positions of generations before `g`. Then, holding the commit lock and with the group committer drained, it swaps the written pages into the head generation: every node there that this checkpoint wrote is replaced with its stored reference, and parents that only changed because of that are copied (`Materializer.rebound`). Head, `current` and head's history entry become the rebound generation. Last, it recounts the write buffer and wakes commits that were waiting for room. +9. `TransactionManager.checkpointed(written g, epoch)` publishes the epoch and forgets the commit positions of generations before `g`. It then swaps the written pages into the head generation: every node there that this checkpoint wrote is replaced with its stored reference, and parents that only changed because of that are copied (`Materializer.Rebinding`). The first pass runs over the head it reads without any lock and remembers what it did for every node it visited. `CHECKPOINT_REBIND` is reached next. Then, holding the commit lock and with the group committer drained, a second pass runs over the head as it is now: nodes the first pass already visited are looked up, so this pass only walks what commits added in between. Under `durability = sync` the drain can still wait for the sync of a commit batch that is in flight. Head, `current` and head's history entry become the rebound generation. Last, it recounts the write buffer and wakes commits that were waiting for room. 10. `wal.append(Checkpoint(0, g, lsn))`, `wal.sync()`, `wal.truncateBefore(lsn)`. The sync and the file deletions run outside the WAL lock (section 3), so a commit that appends meanwhile does not wait for them. 11. `CHECKPOINT_JOURNAL`, then `pages.applyJournal()` writes the journaled values into `pages.ref` and `pages.dir`, forces both and truncates the journal to zero bytes. The checkpoint remembers `lsn` and resumes the counter. -Because the journal holds absolute values, applying it twice gives the same files. If the process dies after step 8 but before step 11 is finished, the next open finds a journal whose LSN matches the published catalog and applies it again. If it dies before step 8, the journal's LSN doesn't match and it is ignored: `pages.ref` and `pages.dir` were not touched, so they still match the previous catalog, and the pages this checkpoint wrote have addresses but no counts, so the next open hands them to its first checkpoint, which finds nothing that reaches them and marks them dead. If step 10 or 11 fails and the engine keeps running, the next checkpoint applies the old journal before writing a new one, and a failed step 8 marks `pages.ref` for rebuilding. `ReferenceCountTest` crashes at `CHECKPOINT_CATALOG` and at `CHECKPOINT_JOURNAL` and checks the counts after reopening, including after a checkpoint that failed at step 11 and a next one that crashed before step 8. `DeferredWritesTest` crashes a checkpoint at `CHECKPOINT_PAGES`, `CHECKPOINT_CATALOG` and `CHECKPOINT_JOURNAL` while another thread keeps committing, and checks that every acknowledged commit is there after reopening and that the counts match. `CheckpointConcurrencyTest` does the same with three writers at `CHECKPOINT_CATALOG`, `WAL_FORCE` and `CHECKPOINT_JOURNAL`, reopening twice. A crash at `WAL_FORCE` can also land in a commit batch that the checkpoint drains in step 9, so there the writers may fail too, and the only commits allowed to come back without an acknowledgement are the ones in flight. It also commits from another thread while the checkpoint is held at `WAL_FORCE`, which must finish without waiting for the checkpoint and still be there after reopening, and it runs four writers through back-to-back checkpoints, each checking that its commit is visible right after it returns. +Because the journal holds absolute values, applying it twice gives the same files. If the process dies after step 8 but before step 11 is finished, the next open finds a journal whose LSN matches the published catalog and applies it again. If it dies before step 8, the journal's LSN doesn't match and it is ignored: `pages.ref` and `pages.dir` were not touched, so they still match the previous catalog, and the pages this checkpoint wrote have addresses but no counts, so the next open hands them to its first checkpoint, which finds nothing that reaches them and marks them dead. If step 10 or 11 fails and the engine keeps running, the next checkpoint applies the old journal before writing a new one, and a failed step 8 marks `pages.ref` for rebuilding. `ReferenceCountTest` crashes at `CHECKPOINT_CATALOG` and at `CHECKPOINT_JOURNAL` and checks the counts after reopening, including after a checkpoint that failed at step 11 and a next one that crashed before step 8. `DeferredWritesTest` crashes a checkpoint at `CHECKPOINT_PAGES`, `CHECKPOINT_CATALOG` and `CHECKPOINT_JOURNAL` while another thread keeps committing, and checks that every acknowledged commit is there after reopening and that the counts match. `CheckpointConcurrencyTest` does the same with three writers at `CHECKPOINT_CATALOG`, `CHECKPOINT_REBIND`, `WAL_FORCE` and `CHECKPOINT_JOURNAL`, reopening twice. A crash at `WAL_FORCE` can also land in a commit batch that the checkpoint drains in step 9, so there the writers may fail too, and the only commits allowed to come back without an acknowledgement are the ones in flight. It also commits from another thread while the checkpoint is held at `CHECKPOINT_REBIND` or `WAL_FORCE`, which must finish without waiting for the checkpoint and still be there after the rebind and after reopening, and it runs four writers through back-to-back checkpoints, each checking that its commit is visible right after it returns. The checkpoint LSN is the end of `g`'s `Commit` record, not the position of the `Checkpoint` record. The next recovery starts there, replays the commits published after `g`, and skips the `Checkpoint` record when it reaches it. Triggers: diff --git a/engine/src/main/java/io/hstore/engine/tree/Materializer.java b/engine/src/main/java/io/hstore/engine/tree/Materializer.java index f6f5e26..839cb57 100644 --- a/engine/src/main/java/io/hstore/engine/tree/Materializer.java +++ b/engine/src/main/java/io/hstore/engine/tree/Materializer.java @@ -7,6 +7,8 @@ import io.hstore.engine.page.PageId; import java.lang.foreign.MemorySegment; +import java.util.IdentityHashMap; +import java.util.Map; import java.util.function.Consumer; public final class Materializer { @@ -203,45 +205,65 @@ public static long unwrittenHeap(Ref ref) { return total[0]; } - public static Ref rebound(Ref ref, TreeSchema schema, Epoch epoch) { - if (!(ref instanceof Ref.Pending(Node node))) { - return ref; + public static final class Rebinding { + + private final Epoch epoch; + private final Map visited = new IdentityHashMap<>(); + + public Rebinding(Epoch epoch) { + this.epoch = epoch; } - Ref written = node.writtenIn(epoch); - if (written != null) { - return written; + + public Ref rebound(Ref ref, TreeSchema schema) { + if (!(ref instanceof Ref.Pending(Node node))) { + return ref; + } + Ref written = node.writtenIn(epoch); + if (written != null) { + return written; + } + Ref known = visited.get(node); + if (known != null) { + return known instanceof Ref.Pending(Node same) && same == node ? ref : known; + } + Ref result = switch (node) { + case Leaf leaf -> rebound(ref, leaf); + case Branch branch -> rebound(ref, branch); + }; + visited.put(node, result); + return result; } - return switch (node) { - case Leaf leaf -> { - if (!leaf.schema.codec().holdsRefs()) { - yield ref; - } - Object[] values = leaf.valueList().toArray(); - boolean changed = false; - for (int i = 0; i < values.length; i++) { - boolean[] moved = {false}; - Object mapped = mapRefs(leaf.schema, values[i], (nested, nestedSchema) -> { - Ref rebound = rebound(nested, nestedSchema, epoch); - moved[0] |= rebound != nested; - return rebound; - }); - if (moved[0]) { - values[i] = mapped; - changed = true; - } - } - yield changed ? new Ref.Pending(leaf.frozenWith(values)) : ref; + + private Ref rebound(Ref ref, Leaf leaf) { + if (!leaf.schema.codec().holdsRefs()) { + return ref; } - case Branch branch -> { - Ref[] children = branch.childArray(); - boolean changed = false; - for (int i = 0; i < children.length; i++) { - Ref rebound = rebound(children[i], branch.schema, epoch); - changed |= rebound != children[i]; - children[i] = rebound; + Object[] values = leaf.valueList().toArray(); + boolean changed = false; + for (int i = 0; i < values.length; i++) { + boolean[] moved = {false}; + Object mapped = mapRefs(leaf.schema, values[i], (nested, nestedSchema) -> { + Ref rebound = rebound(nested, nestedSchema); + moved[0] |= rebound != nested; + return rebound; + }); + if (moved[0]) { + values[i] = mapped; + changed = true; } - yield changed ? new Ref.Pending(branch.frozenWith(children)) : ref; } - }; + return changed ? new Ref.Pending(leaf.frozenWith(values)) : ref; + } + + private Ref rebound(Ref ref, Branch branch) { + Ref[] children = branch.childArray(); + boolean changed = false; + for (int i = 0; i < children.length; i++) { + Ref rebound = rebound(children[i], branch.schema); + changed |= rebound != children[i]; + children[i] = rebound; + } + return changed ? new Ref.Pending(branch.frozenWith(children)) : ref; + } } } diff --git a/engine/src/main/java/io/hstore/engine/txn/CrashPoint.java b/engine/src/main/java/io/hstore/engine/txn/CrashPoint.java index 3458ccf..93b265b 100644 --- a/engine/src/main/java/io/hstore/engine/txn/CrashPoint.java +++ b/engine/src/main/java/io/hstore/engine/txn/CrashPoint.java @@ -16,6 +16,7 @@ public enum CrashPoint { CHECKPOINT_PAGES, CHECKPOINT_CATALOG, CHECKPOINT_JOURNAL, + CHECKPOINT_REBIND, WAL_FORCE; @FunctionalInterface diff --git a/engine/src/main/java/io/hstore/engine/txn/TransactionManager.java b/engine/src/main/java/io/hstore/engine/txn/TransactionManager.java index 765f9da..d307b77 100644 --- a/engine/src/main/java/io/hstore/engine/txn/TransactionManager.java +++ b/engine/src/main/java/io/hstore/engine/txn/TransactionManager.java @@ -548,23 +548,25 @@ public void checkpointed(Generation materialized, Epoch epoch) { epoch.publish(); published = epoch; commitEnds.headMap(materialized.id(), false).clear(); + Materializer.Rebinding rebinding = new Materializer.Rebinding(epoch); + rebound(head.get(), rebinding); + storage.faults().reach(CrashPoint.CHECKPOINT_REBIND); exclusive(() -> { - rebind(epoch); + Generation rebound = rebound(head.get(), rebinding); + head.set(rebound); + current.set(rebound); + history.put(rebound.id(), rebound); return null; }); recountUnwritten(); signalRoom(); } - private void rebind(Epoch epoch) { - Generation latest = head.get(); + private Generation rebound(Generation latest, Materializer.Rebinding rebinding) { Map branches = new TreeMap<>(); latest.branches().forEach((id, branch) -> branches.put(id, branch.withRoots(branch.roots().map((slot, ref) -> - Materializer.rebound(ref, storage.slots().slot(slot).schema(), epoch))))); - Generation rebound = new Generation(latest.id(), latest.wallTime(), latest.txnId(), latest.nextAtom(), latest.nextTxn(), branches); - head.set(rebound); - current.set(rebound); - history.put(rebound.id(), rebound); + rebinding.rebound(ref, storage.slots().slot(slot).schema()))))); + return new Generation(latest.id(), latest.wallTime(), latest.txnId(), latest.nextAtom(), latest.nextTxn(), branches); } private void makeDurable(int batchSize, boolean wrotePages) { diff --git a/engine/src/test/java/io/hstore/engine/CheckpointConcurrencyTest.java b/engine/src/test/java/io/hstore/engine/CheckpointConcurrencyTest.java index 6acae4d..1181c85 100644 --- a/engine/src/test/java/io/hstore/engine/CheckpointConcurrencyTest.java +++ b/engine/src/test/java/io/hstore/engine/CheckpointConcurrencyTest.java @@ -57,7 +57,7 @@ private static void insert(StorageEngine engine, long edge, long member) { } @ParameterizedTest - @EnumSource(value = CrashPoint.class, names = {"WAL_FORCE"}) + @EnumSource(value = CrashPoint.class, names = {"CHECKPOINT_REBIND", "WAL_FORCE"}) void aCommitFinishesWhileTheCheckpointIsAtThisStep(CrashPoint point) throws Exception { AtomicReference opened = new AtomicReference<>(); AtomicReference outcome = new AtomicReference<>(); @@ -98,7 +98,7 @@ void aCommitFinishesWhileTheCheckpointIsAtThisStep(CrashPoint point) throws Exce } @ParameterizedTest - @EnumSource(value = CrashPoint.class, names = {"CHECKPOINT_CATALOG", "WAL_FORCE", "CHECKPOINT_JOURNAL"}) + @EnumSource(value = CrashPoint.class, names = {"CHECKPOINT_CATALOG", "CHECKPOINT_REBIND", "WAL_FORCE", "CHECKPOINT_JOURNAL"}) void aCrashAtThisCheckpointStepLosesNoAcknowledgedCommit(CrashPoint point) throws Exception { AtomicBoolean armed = new AtomicBoolean(); Thread test = Thread.currentThread();