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/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/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); 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..c27d27d --- /dev/null +++ b/engine/src/test/java/io/hstore/engine/TxnIdReuseTest.java @@ -0,0 +1,90 @@ +// 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; + }); + } + } + + @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)) { + 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(); + return doomed.id(); + } +}