Publish the recovered log end as the first checkpoint's lsn

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();

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<Long, List<WalRecord>> pending = new HashMap<>();
         List<WalRecord.Commit> 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();

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);

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<Transaction> doomedWork) {
+        crashDuringCommit(doomedWork);
+        StorageEngine second = StorageEngine.open(directory, EngineTest.small());
+        second.write(txn -> txn.createNode(EngineTest.NODE, "real"));
+        second.halt();
+    }
+
+    private long crashDuringCommit(Consumer<Transaction> 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();
+    }
+}