From 8bb9817a819576a9695e24f4a8adf75444f7f8ae Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Fri, 24 Jul 2026 12:27:42 +0300 Subject: [PATCH] [#304] Throw PersistitClosedException when a commit races Persistit.close() A transaction commit racing Persistit.close() could reach JournalManager.waitForDurability after close() cleared the JOURNAL_FLUSHER reference and fail with a raw IllegalStateException, bypassing callers' closed-state handling (seen as a TransactionTest2.transactionsConcurrentWithPersistitClose flake). A commit that instead captured the flusher reference just before it was cleared would poll forever against the stopped thread. Throw PersistitClosedException in both cases: when the flusher reference is already null, and inside the JournalFlusher wait loop once the flusher is stopped and its final durability marks cannot cover the caller's timestamp. The final marks are published before the flusher stops, so a single re-read after observing the stop distinguishes a durable commit from one that can never be confirmed. New JournalManagerTest cases cover both windows; without the fix the first fails with the IllegalStateException and the second hangs past its join deadline. Fixes #304 --- .../java/com/persistit/JournalManager.java | 20 ++++++++- .../com/persistit/JournalManagerTest.java | 45 +++++++++++++++++++ 2 files changed, 64 insertions(+), 1 deletion(-) diff --git a/persistit/core/src/main/java/com/persistit/JournalManager.java b/persistit/core/src/main/java/com/persistit/JournalManager.java index 0a8874405..99b7f01a1 100644 --- a/persistit/core/src/main/java/com/persistit/JournalManager.java +++ b/persistit/core/src/main/java/com/persistit/JournalManager.java @@ -61,6 +61,7 @@ import com.persistit.Persistit.FatalErrorException; import com.persistit.TransactionPlayer.TransactionPlayerListener; import com.persistit.exception.CorruptJournalException; +import com.persistit.exception.PersistitClosedException; import com.persistit.exception.PersistitException; import com.persistit.exception.PersistitIOException; import com.persistit.exception.PersistitInterruptedException; @@ -1849,6 +1850,8 @@ void pruneObsoleteTransactions(final boolean rollbackPruningEnabled) { * complete the I/O when requested. In particular, a value of * zero indicates the I/O should start immediately. * @throws PersistitInterruptedException if the thread is interrupted while waiting + * @throws PersistitClosedException if Persistit was closed or crashed before + * durability could be confirmed */ void waitForDurability(final long flushedTimestamp, final long leadTime, final long stallTime) @@ -1857,7 +1860,7 @@ void waitForDurability(final long flushedTimestamp, final long leadTime, final l if (flusher != null) { flusher.waitForDurability(flushedTimestamp, leadTime, stallTime); } else { - throw new IllegalStateException("JOURNAL_FLUSHER is not running"); + throw new PersistitClosedException("JOURNAL_FLUSHER is not running"); } } @@ -2349,6 +2352,21 @@ private void waitForDurability(final long flushedTimestamp, final long leadTime, break; } + if (isStopped()) { + /* + * The flusher publishes its final _startTimestamp and + * _endTimestamp before stopping, so one re-read after + * observing the stop is exact: either the final flush + * cycle covered flushedTimestamp, or no future cycle ever + * will and polling would spin forever against a stopped + * thread. + */ + if (_endTimestamp > flushedTimestamp && _startTimestamp > flushedTimestamp) { + break; + } + throw new PersistitClosedException("JOURNAL_FLUSHER stopped before durability was confirmed"); + } + long remainingSleepNanos; if (estimatedRemainingIoNanos == -1) { remainingSleepNanos = Math.max(0, _flushInterval - (now - endTime)); diff --git a/persistit/core/src/test/java/com/persistit/JournalManagerTest.java b/persistit/core/src/test/java/com/persistit/JournalManagerTest.java index 0350e9578..3d23220bb 100644 --- a/persistit/core/src/test/java/com/persistit/JournalManagerTest.java +++ b/persistit/core/src/test/java/com/persistit/JournalManagerTest.java @@ -22,6 +22,7 @@ import com.persistit.CheckpointManager.Checkpoint; import com.persistit.JournalManager.PageNode; import com.persistit.TransactionPlayer.TransactionPlayerListener; +import com.persistit.exception.PersistitClosedException; import com.persistit.exception.PersistitException; import com.persistit.unit.ConcurrentUtil.ThrowingRunnable; import com.persistit.util.Util; @@ -678,6 +679,50 @@ public void run() throws Exception { disableSequencer(); } + @Test + public void waitForDurabilityAfterCloseThrowsPersistitClosedException() throws Exception { + final JournalManager jman = _persistit.getJournalManager(); + final long flushedTimestamp = _persistit.getTimestampAllocator().updateTimestamp(); + _persistit.close(); + try { + jman.waitForDurability(flushedTimestamp, 0, 0); + fail("Expected PersistitClosedException"); + } catch (final PersistitClosedException expected) { + /* + * A commit racing Persistit.close() must observe the documented + * closed-state exception, not a raw IllegalStateException that + * bypasses callers' shutdown handling (issue #304). + */ + } + } + + @Test + public void waitForDurabilityOnStoppedFlusherThrowsInsteadOfSpinning() throws Exception { + final JournalManager jman = _persistit.getJournalManager(); + _persistit.crash(); + /* + * crash() stops JOURNAL_FLUSHER without clearing the JournalManager's + * reference to it, and a timestamp allocated after the flusher's final + * cycle can never be covered by its durability marks - the shape of + * the shutdown race in which a committing thread captured the flusher + * reference just before it was cleared. Without the stopped-flusher + * check this wait polls forever. + */ + final long flushedTimestamp = _persistit.getTimestampAllocator().updateTimestamp(); + final Thread waiter = createThread("WAIT_FOR_DURABILITY", new ThrowingRunnable() { + @Override + public void run() throws Exception { + try { + jman.waitForDurability(flushedTimestamp, 0, 0); + fail("Expected PersistitClosedException"); + } catch (final PersistitClosedException expected) { + // expected + } + } + }); + startAndJoinAssertSuccess(10000, waiter); + } + private int countKeys(final boolean mvcc) throws PersistitException { final Exchange exchange = _persistit.getExchange(_volumeName, "JournalManagerTest1", false); exchange.ignoreMVCCFetch(!mvcc);