From f7467079dbfc6da9b01f960ed7591fbea78577dc Mon Sep 17 00:00:00 2001 From: Ujjwaljain16 Date: Tue, 13 Oct 2026 02:38:36 +0530 Subject: [PATCH 1/3] Publish the recovered log end as the first checkpoint's lsn When recovery replayed no commit, the first checkpoint put the old lsn back, so WAL records of a transaction that never committed stayed in the replay window. A new transaction could be handed the same id, and after a second crash recovery applied both as one commit. That made a node that was never acknowledged visible, and let a half-written transaction delete committed data. Start the checkpointer from the lsn recovery already computed. Fixes #91 --- .../java/io/hstore/engine/StorageEngine.java | 2 +- .../java/io/hstore/engine/TxnIdReuseTest.java | 74 +++++++++++++++++++ 2 files changed, 75 insertions(+), 1 deletion(-) create mode 100644 engine/src/test/java/io/hstore/engine/TxnIdReuseTest.java diff --git a/engine/src/main/java/io/hstore/engine/StorageEngine.java b/engine/src/main/java/io/hstore/engine/StorageEngine.java index 964fdcc..9659006 100644 --- a/engine/src/main/java/io/hstore/engine/StorageEngine.java +++ b/engine/src/main/java/io/hstore/engine/StorageEngine.java @@ -93,7 +93,7 @@ private StorageEngine(Path directory, EngineOptions options, FileChannel lockCha options.writeBufferBytes()); this.transactions = new TransactionManager(storage, recovery.image().current(), recovery.image().history()); this.checkpointer = new Checkpointer(transactions, pages, source, wal, feed, catalog, references, slots, options.faults(), - checkpoint.current(), checkpoint.checkpointLsn()); + checkpoint.current(), recovery.image().checkpointLsn()); this.compactor = new Compactor(transactions, pages, checkpointer, options.faults()); this.dictionary = new Dictionary(transactions); checkpointer.checkpoint(); diff --git a/engine/src/test/java/io/hstore/engine/TxnIdReuseTest.java b/engine/src/test/java/io/hstore/engine/TxnIdReuseTest.java new file mode 100644 index 0000000..41edb77 --- /dev/null +++ b/engine/src/test/java/io/hstore/engine/TxnIdReuseTest.java @@ -0,0 +1,74 @@ +// SPDX-FileCopyrightText: Metacog Labs +// SPDX-License-Identifier: PolyForm-Noncommercial-1.0.0 + +package io.hstore.engine; + +import io.hstore.engine.txn.CrashPoint; +import io.hstore.engine.txn.Transaction; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.nio.file.Path; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.function.Consumer; + +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 TxnIdReuseTest { + + @TempDir + Path directory; + + @Test + void anUnacknowledgedTransactionStaysGoneAfterASecondCrash() { + crashDuringCommitThenCrashAgain(doomed -> doomed.createNode(EngineTest.NODE, "ghost")); + + try (StorageEngine third = StorageEngine.open(directory, EngineTest.small())) { + third.read(snapshot -> { + assertTrue(snapshot.resolve(0, EngineTest.NODE, "real").isPresent()); + assertTrue(snapshot.resolve(0, EngineTest.NODE, "ghost").isEmpty()); + assertEquals(3, snapshot.countOfType(0, EngineTest.NODE)); + return null; + }); + } + } + + @Test + void anUnacknowledgedDeleteDoesNotRemoveCommittedDataAfterASecondCrash() { + crashDuringCommitThenCrashAgain(doomed -> doomed.delete(1)); + + try (StorageEngine third = StorageEngine.open(directory, EngineTest.small())) { + third.read(snapshot -> { + assertTrue(snapshot.atom(1).isPresent()); + assertTrue(snapshot.resolve(0, EngineTest.NODE, "real").isPresent()); + return null; + }); + } + } + + private void crashDuringCommitThenCrashAgain(Consumer doomedWork) { + AtomicBoolean armed = new AtomicBoolean(false); + EngineOptions crashing = EngineTest.small().withFaults(reached -> { + if (reached == CrashPoint.WAL_APPEND && armed.compareAndSet(true, false)) { + throw new CrashPoint.SimulatedCrash(reached); + } + }); + try (StorageEngine seeded = StorageEngine.open(directory, EngineTest.small())) { + assertEquals(1L, (long) seeded.write(txn -> txn.createNode(EngineTest.NODE, "base"))); + seeded.write(txn -> txn.createNode(EngineTest.NODE, "other")); + } + StorageEngine first = StorageEngine.open(directory, crashing); + armed.set(true); + Transaction doomed = first.begin(); + doomedWork.accept(doomed); + assertThrows(CrashPoint.SimulatedCrash.class, doomed::commit); + first.halt(); + + StorageEngine second = StorageEngine.open(directory, EngineTest.small()); + second.write(txn -> txn.createNode(EngineTest.NODE, "real")); + second.halt(); + } +} From 82d8f516550073403d7d451937cc54ce37647482 Mon Sep 17 00:00:00 2001 From: Ujjwaljain16 Date: Tue, 13 Oct 2026 02:42:19 +0530 Subject: [PATCH 2/3] Test that commits after a discarded tail survive the next crash Covers the durable-prefix case of the same problem: after recovery drops commits whose pages are missing, a commit made afterwards was lost on the next unclean stop. Refs #91 --- .../io/hstore/engine/CrashRecoveryTest.java | 24 +++++++++++++++++++ 1 file changed, 24 insertions(+) diff --git a/engine/src/test/java/io/hstore/engine/CrashRecoveryTest.java b/engine/src/test/java/io/hstore/engine/CrashRecoveryTest.java index 1f6a301..09bf5ea 100644 --- a/engine/src/test/java/io/hstore/engine/CrashRecoveryTest.java +++ b/engine/src/test/java/io/hstore/engine/CrashRecoveryTest.java @@ -122,6 +122,30 @@ void recoveryStopsAtTheFirstCommitWhosePagesAreMissing() throws Exception { } } + @Test + void commitsAfterADiscardedTailSurviveTheNextCrash() throws Exception { + EngineOptions options = EngineTest.small().withWalMode(WalMode.PAGE_REFERENCES).withPagesPerSegment(100_000); + try (StorageEngine engine = StorageEngine.open(directory, options)) { + engine.write(txn -> txn.createNode(EngineTest.NODE, "durable")); + } + StorageEngine engine = StorageEngine.open(directory, options); + loadedNode(engine, "lost-1"); + long beforeSecond = Files.size(lastSegment()); + loadedNode(engine, "lost-2"); + engine.halt(); + try (FileChannel segment = FileChannel.open(lastSegment(), StandardOpenOption.WRITE)) { + segment.truncate(beforeSecond - 1); + } + StorageEngine recovered = StorageEngine.open(directory, options); + assertTrue(recovered.recovery().discardedCommits() >= 1); + recovered.write(txn -> txn.createNode(EngineTest.NODE, "afterwards")); + recovered.halt(); + try (StorageEngine reopened = StorageEngine.open(directory, options)) { + assertTrue(reopened.read(snapshot -> snapshot.resolve(0, EngineTest.NODE, "afterwards")).isPresent()); + assertTrue(reopened.read(snapshot -> snapshot.resolve(0, EngineTest.NODE, "durable")).isPresent()); + } + } + private static void loadedNode(StorageEngine engine, String key) { engine.write(txn -> { txn.createNode(EngineTest.NODE, key); From abe7cd0b6a2df7fabacd721cf045ce1df6518cbb Mon Sep 17 00:00:00 2001 From: Ujjwaljain16 Date: Tue, 13 Oct 2026 15:16:05 +0530 Subject: [PATCH 3/3] Skip past every transaction id recovery sees The next transaction id was only restored from Commit records, so the id of a transaction that died before committing was handed out again while its records were still in the log. Raise the recovered next id above the highest id found in the log. Refs #91 --- .../hstore/engine/maintenance/Recovery.java | 7 ++++++ .../java/io/hstore/engine/TxnIdReuseTest.java | 24 +++++++++++++++---- 2 files changed, 27 insertions(+), 4 deletions(-) diff --git a/engine/src/main/java/io/hstore/engine/maintenance/Recovery.java b/engine/src/main/java/io/hstore/engine/maintenance/Recovery.java index 6f54718..bb602ad 100644 --- a/engine/src/main/java/io/hstore/engine/maintenance/Recovery.java +++ b/engine/src/main/java/io/hstore/engine/maintenance/Recovery.java @@ -50,7 +50,9 @@ public static Outcome recover(CatalogImage checkpoint, WriteAheadLog wal, PageSt SlotRegistry slots, NodeSource source) { Map> pending = new HashMap<>(); List commits = new ArrayList<>(); + long[] highestTxn = {0}; wal.read(checkpoint.checkpointLsn()).forEach(positioned -> { + highestTxn[0] = Math.max(highestTxn[0], positioned.record().txnId()); switch (positioned.record()) { case WalRecord.Commit commit -> commits.add(commit); case WalRecord.Abort abort -> pending.remove(abort.txnId()); @@ -83,6 +85,11 @@ public static Outcome recover(CatalogImage checkpoint, WriteAheadLog wal, PageSt history.add(generation); replayed.add(generation); } + if (generation.nextTxn() <= highestTxn[0]) { + generation = new Generation(generation.id(), generation.wallTime(), generation.txnId(), generation.nextAtom(), + highestTxn[0] + 1, generation.branches()); + history.set(history.size() - 1, generation); + } pages.sync(); feed.truncateAfter(generation.id()); feed.sync(); diff --git a/engine/src/test/java/io/hstore/engine/TxnIdReuseTest.java b/engine/src/test/java/io/hstore/engine/TxnIdReuseTest.java index 41edb77..c27d27d 100644 --- a/engine/src/test/java/io/hstore/engine/TxnIdReuseTest.java +++ b/engine/src/test/java/io/hstore/engine/TxnIdReuseTest.java @@ -49,7 +49,26 @@ void anUnacknowledgedDeleteDoesNotRemoveCommittedDataAfterASecondCrash() { } } + @Test + void transactionIdsAreNotReusedAfterACrash() { + long doomedId = crashDuringCommit(doomed -> doomed.createNode(EngineTest.NODE, "ghost")); + + try (StorageEngine reopened = StorageEngine.open(directory, EngineTest.small())) { + Transaction next = reopened.begin(); + long nextId = next.id(); + next.abort(); + assertTrue(nextId > doomedId, "transaction " + nextId + " reuses an id already in the log"); + } + } + private void crashDuringCommitThenCrashAgain(Consumer doomedWork) { + crashDuringCommit(doomedWork); + StorageEngine second = StorageEngine.open(directory, EngineTest.small()); + second.write(txn -> txn.createNode(EngineTest.NODE, "real")); + second.halt(); + } + + private long crashDuringCommit(Consumer doomedWork) { AtomicBoolean armed = new AtomicBoolean(false); EngineOptions crashing = EngineTest.small().withFaults(reached -> { if (reached == CrashPoint.WAL_APPEND && armed.compareAndSet(true, false)) { @@ -66,9 +85,6 @@ private void crashDuringCommitThenCrashAgain(Consumer doomedWork) { doomedWork.accept(doomed); assertThrows(CrashPoint.SimulatedCrash.class, doomed::commit); first.halt(); - - StorageEngine second = StorageEngine.open(directory, EngineTest.small()); - second.write(txn -> txn.createNode(EngineTest.NODE, "real")); - second.halt(); + return doomed.id(); } }