[#304] Throw PersistitClosedException when a commit races Persistit.close()

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

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