Deflake close interrupt lifecycle tests

The lifecycle tests inferred that close had left its creation wait whenever
ReentrantLock.hasWaiters() briefly returned false. Interrupt delivery moves a
condition waiter to the lock queue while awaitNanos() reacquires the lock, so
the observer could mistake that transition for deadline expiry.

Add test-only retry hooks that fire only after each pool catches an interrupt
and confirms the original deadline still permits another wait. The tests now
send each next interrupt after that acknowledgement, preventing coalescing and
removing the transient queue-membership check while retaining deadline-restart
coverage.

Author: bluestreak01

core/src/main/java/io/questdb/client/impl/QueryClientPool.java

@@ -97,6 +97,11 @@ public final class QueryClientPool implements AutoCloseable {
     // DEFAULT_CLOSE_QUERY_TIMEOUT_MILLIS. Volatile because QuestDBImpl sets it
     // once at build time on a different thread than the borrowers that read it.
     private volatile long closeQueryTimeoutMillis = DEFAULT_CLOSE_QUERY_TIMEOUT_MILLIS;
+    // Test seam invoked after close() handles an interrupt and confirms the
+    // original creation-wait deadline still permits another wait. Null in
+    // production; lifecycle tests use it to acknowledge distinct retries
+    // without inspecting transient Condition queue membership.
+    private volatile Runnable creationWaitRetryHook;
     private int inFlightCreations;
 
     public QueryClientPool(
@@ -340,12 +345,20 @@ public void close() {
             long creationRemainingNanos = creationWaitNanos;
             boolean creationWaitInterrupted = false;
             while (inFlightCreations > 0 && creationRemainingNanos > 0) {
+                boolean isRetryingAfterInterrupt = false;
                 try {
                     creationFinished.awaitNanos(creationRemainingNanos);
                 } catch (InterruptedException e) {
                     creationWaitInterrupted = true;
+                    isRetryingAfterInterrupt = true;
                 }
                 creationRemainingNanos = creationWaitDeadlineNanos - System.nanoTime();
+                if (isRetryingAfterInterrupt && inFlightCreations > 0 && creationRemainingNanos > 0) {
+                    Runnable hook = creationWaitRetryHook;
+                    if (hook != null) {
+                        hook.run();
+                    }
+                }
             }
             if (creationWaitInterrupted) {
                 Thread.currentThread().interrupt();
@@ -557,6 +570,11 @@ public boolean isClosedForTesting() {
         return closed;
     }
 
+    @TestOnly
+    public void setCreationWaitRetryHookForTesting(Runnable hook) {
+        this.creationWaitRetryHook = hook;
+    }
+
     private QueryWorker createUnlocked() {
         QwpQueryClient client = QwpQueryClient.fromConfig(configurationString);
         try {

core/src/main/java/io/questdb/client/impl/SenderPool.java

@@ -222,6 +222,11 @@ public final class SenderPool implements AutoCloseable {
     // production; regression tests release a retired slot here to prove that
     // the terminal pass re-probes returned capacity before throwing.
     private volatile Runnable borrowWaitExpiredHook;
+    // Test seam invoked after close() handles an interrupt and confirms the
+    // original creation-wait deadline still permits another wait. Null in
+    // production; lifecycle tests use it to acknowledge distinct retries
+    // without inspecting transient Condition queue membership.
+    private volatile Runnable creationWaitRetryHook;
     // Slots removed from `all` whose delegate is still releasing its flock.
     // They keep reserving capacity (and their slotInUse mark) until the
     // flock drops, so the cap check and the slot allocator stay consistent
@@ -1410,6 +1415,11 @@ public void setBorrowWaitExpiredHook(Runnable hook) {
         this.borrowWaitExpiredHook = hook;
     }
 
+    @TestOnly
+    public void setCreationWaitRetryHookForTesting(Runnable hook) {
+        this.creationWaitRetryHook = hook;
+    }
+
     /**
      * Raises the shutdown signal early -- without tearing down live delegates --
      * so an in-flight startup-recovery step stops promptly between slots. Direct
@@ -1484,12 +1494,20 @@ public void close() {
             long creationRemainingNanos = creationWaitNanos;
             boolean creationWaitInterrupted = false;
             while (inFlightCreations > 0 && creationRemainingNanos > 0) {
+                boolean isRetryingAfterInterrupt = false;
                 try {
                     creationFinished.awaitNanos(creationRemainingNanos);
                 } catch (InterruptedException e) {
                     creationWaitInterrupted = true;
+                    isRetryingAfterInterrupt = true;
                 }
                 creationRemainingNanos = creationWaitDeadlineNanos - System.nanoTime();
+                if (isRetryingAfterInterrupt && inFlightCreations > 0 && creationRemainingNanos > 0) {
+                    Runnable hook = creationWaitRetryHook;
+                    if (hook != null) {
+                        hook.run();
+                    }
+                }
             }
             if (creationWaitInterrupted) {
                 Thread.currentThread().interrupt();

core/src/test/java/io/questdb/client/test/impl/QuestDBImplCloseLifecycleTest.java

@@ -39,11 +39,11 @@
 
 import java.lang.reflect.Proxy;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Semaphore;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
-import java.util.function.BooleanSupplier;
 import java.util.function.Consumer;
 import java.util.function.IntFunction;
 
@@ -186,18 +186,21 @@ public void facadeCloseIsBoundedUnderRepeatedInterruptsDuringQueryCreation() thr
                 inCreation.countDown();
                 awaitOrFail(releaseCreation, "test never released query creation");
             };
-            // A 1s creation-wait budget, not 100ms: the interrupt storm below must land at least
-            // twice inside this window for the deadline-restart property to be exercised at all,
-            // and a freshly started, yielding interrupter thread is not guaranteed two scheduler
-            // quanta within 100ms on a saturated CI agent (observed on hosted 3-core mac agents,
-            // where the post-join count assert failed with the product deadline honored exactly).
+            // A 1s creation-wait budget gives the closer scheduler margin. Each next interrupt
+            // is sent only after the pool acknowledges that it caught the previous one and will
+            // retry against the original deadline, so interrupts cannot coalesce.
             QuestDBImpl db = newQuestDB(
                     SENDER_CFG, 0, 0, 1000, slotIndex -> fakeSender(null, null, null), connectHook);
             QueryClientPool pool = db.getQueryPoolForTesting();
             AtomicReference<Throwable> borrowOutcome = new AtomicReference<>();
-            AtomicBoolean closeReturnedInterrupted = new AtomicBoolean();
-            AtomicBoolean keepInterrupting = new AtomicBoolean(true);
-            AtomicInteger interruptCount = new AtomicInteger();
+            AtomicBoolean isCloseComplete = new AtomicBoolean();
+            AtomicBoolean isCloseReturnedInterrupted = new AtomicBoolean();
+            AtomicInteger interruptRetryCount = new AtomicInteger();
+            Semaphore closeProgress = new Semaphore(0);
+            pool.setCreationWaitRetryHookForTesting(() -> {
+                interruptRetryCount.incrementAndGet();
+                closeProgress.release();
+            });
             Thread borrower = new Thread(() -> {
                 try {
                     db.borrowQuery();
@@ -206,16 +209,14 @@ public void facadeCloseIsBoundedUnderRepeatedInterruptsDuringQueryCreation() thr
                 }
             }, "interrupted-query-borrower");
             Thread closer = new Thread(() -> {
-                db.close();
-                closeReturnedInterrupted.set(Thread.currentThread().isInterrupted());
-            }, "interrupted-query-closer");
-            Thread interrupter = new Thread(() -> {
-                while (keepInterrupting.get()) {
-                    interruptCount.incrementAndGet();
-                    closer.interrupt();
-                    Thread.yield();
+                try {
+                    db.close();
+                    isCloseReturnedInterrupted.set(Thread.currentThread().isInterrupted());
+                } finally {
+                    isCloseComplete.set(true);
+                    closeProgress.release();
                 }
-            }, "query-close-interrupter");
+            }, "interrupted-query-closer");
             long nativeBaseline = Unsafe.getMemUsedByTag(MemoryTag.NATIVE_DEFAULT);
             try {
                 borrower.start();
@@ -226,16 +227,15 @@ public void facadeCloseIsBoundedUnderRepeatedInterruptsDuringQueryCreation() thr
                 closer.start();
                 awaitCreationWaiter(pool,
                         "facade close did not wait while query construction was internally owned");
-                interrupter.start();
-                awaitRepeatedInterrupts(interruptCount, pool::hasCreationWaiterForTesting,
-                        "query close left its creation wait before the interrupt storm landed twice");
-                closer.join(TimeUnit.SECONDS.toMillis(5));
-                Assert.assertFalse(
+                Assert.assertTrue(
                         "repeated interrupts restarted the query creation-wait deadline",
-                        closer.isAlive());
-                Assert.assertTrue("test did not repeatedly interrupt query close", interruptCount.get() > 1);
+                        interruptUntilClose(closer, closeProgress, isCloseComplete));
+                closer.join(TimeUnit.SECONDS.toMillis(5));
+                Assert.assertFalse("facade query closer did not terminate", closer.isAlive());
+                Assert.assertTrue("test did not observe repeated query close interrupt retries",
+                        interruptRetryCount.get() > 1);
                 Assert.assertTrue("facade close must restore query closer interruption",
-                        closeReturnedInterrupted.get());
+                        isCloseReturnedInterrupted.get());
                 Assert.assertEquals(
                         "close must retain late-completion cleanup ownership",
                         1, pool.inFlightCreations());
@@ -252,9 +252,7 @@ public void facadeCloseIsBoundedUnderRepeatedInterruptsDuringQueryCreation() thr
                         borrowOutcome.get() instanceof QueryException
                                 && String.valueOf(borrowOutcome.get().getMessage()).contains("closed"));
             } finally {
-                keepInterrupting.set(false);
                 releaseCreation.countDown();
-                interrupter.join(TimeUnit.SECONDS.toMillis(10));
                 db.close();
                 borrower.join(TimeUnit.SECONDS.toMillis(10));
                 closer.join(TimeUnit.SECONDS.toMillis(10));
@@ -275,15 +273,20 @@ public void facadeCloseIsBoundedUnderRepeatedInterruptsDuringSenderCreation() th
             };
             String senderConfig = "ws::addr=localhost:1;sf_dir="
                     + System.getProperty("java.io.tmpdir") + "/qdb-interrupted-pool-" + System.nanoTime() + ";";
-            // 1s creation-wait budget for the same reason as the query-interrupt test above: the
-            // interrupt storm must land at least twice inside the window even on a saturated agent.
+            // 1s creation-wait budget for the same acknowledged-interrupt retry protocol as the
+            // query test above.
             QuestDBImpl db = newQuestDB(senderConfig, 0, 0, 1000, senderFactory, client -> {
             });
             SenderPool pool = db.getSenderPoolForTesting();
             AtomicReference<Throwable> borrowOutcome = new AtomicReference<>();
-            AtomicBoolean closeReturnedInterrupted = new AtomicBoolean();
-            AtomicBoolean keepInterrupting = new AtomicBoolean(true);
-            AtomicInteger interruptCount = new AtomicInteger();
+            AtomicBoolean isCloseComplete = new AtomicBoolean();
+            AtomicBoolean isCloseReturnedInterrupted = new AtomicBoolean();
+            AtomicInteger interruptRetryCount = new AtomicInteger();
+            Semaphore closeProgress = new Semaphore(0);
+            pool.setCreationWaitRetryHookForTesting(() -> {
+                interruptRetryCount.incrementAndGet();
+                closeProgress.release();
+            });
             Thread borrower = new Thread(() -> {
                 try {
                     db.borrowSender();
@@ -292,16 +295,14 @@ public void facadeCloseIsBoundedUnderRepeatedInterruptsDuringSenderCreation() th
                 }
             }, "interrupted-sender-borrower");
             Thread closer = new Thread(() -> {
-                db.close();
-                closeReturnedInterrupted.set(Thread.currentThread().isInterrupted());
-            }, "interrupted-sender-closer");
-            Thread interrupter = new Thread(() -> {
-                while (keepInterrupting.get()) {
-                    interruptCount.incrementAndGet();
-                    closer.interrupt();
-                    Thread.yield();
+                try {
+                    db.close();
+                    isCloseReturnedInterrupted.set(Thread.currentThread().isInterrupted());
+                } finally {
+                    isCloseComplete.set(true);
+                    closeProgress.release();
                 }
-            }, "sender-close-interrupter");
+            }, "interrupted-sender-closer");
             try {
                 borrower.start();
                 Assert.assertTrue("sender borrow never reached construction",
@@ -313,16 +314,15 @@ public void facadeCloseIsBoundedUnderRepeatedInterruptsDuringSenderCreation() th
                 closer.start();
                 awaitCreationWaiter(pool,
                         "facade close did not wait while sender construction was internally owned");
-                interrupter.start();
-                awaitRepeatedInterrupts(interruptCount, pool::hasCreationWaiterForTesting,
-                        "sender close left its creation wait before the interrupt storm landed twice");
-                closer.join(TimeUnit.SECONDS.toMillis(5));
-                Assert.assertFalse(
+                Assert.assertTrue(
                         "repeated interrupts restarted the sender creation-wait deadline",
-                        closer.isAlive());
-                Assert.assertTrue("test did not repeatedly interrupt sender close", interruptCount.get() > 1);
+                        interruptUntilClose(closer, closeProgress, isCloseComplete));
+                closer.join(TimeUnit.SECONDS.toMillis(5));
+                Assert.assertFalse("facade sender closer did not terminate", closer.isAlive());
+                Assert.assertTrue("test did not observe repeated sender close interrupt retries",
+                        interruptRetryCount.get() > 1);
                 Assert.assertTrue("facade close must restore sender closer interruption",
-                        closeReturnedInterrupted.get());
+                        isCloseReturnedInterrupted.get());
                 Assert.assertEquals(
                         "close must retain late-completion cleanup ownership",
                         1, pool.getInFlightCreationsForTesting());
@@ -341,9 +341,7 @@ public void facadeCloseIsBoundedUnderRepeatedInterruptsDuringSenderCreation() th
                         borrowOutcome.get() instanceof LineSenderException
                                 && String.valueOf(borrowOutcome.get().getMessage()).contains("closed"));
             } finally {
-                keepInterrupting.set(false);
                 releaseCreation.countDown();
-                interrupter.join(TimeUnit.SECONDS.toMillis(10));
                 db.close();
                 borrower.join(TimeUnit.SECONDS.toMillis(10));
                 closer.join(TimeUnit.SECONDS.toMillis(10));
@@ -529,34 +527,6 @@ private static void awaitCreationWaiter(SenderPool pool, String message) {
         Assert.fail(message);
     }
 
-    /**
-     * Holds the test until the interrupt storm has landed at least twice while the facade close is
-     * still inside its bounded creation wait. The deadline-restart property is only exercised by
-     * interrupts that arrive during that wait, and the scheduler owes the interrupter thread
-     * nothing: with a post-join count assert alone, the run races the close budget against thread
-     * scheduling and can fail with the product invariant intact. Failing here instead separates
-     * "interrupter starved before the budget expired" from a genuine deadline bug.
-     */
-    private static void awaitRepeatedInterrupts(
-            AtomicInteger interruptCount,
-            BooleanSupplier closerStillWaiting,
-            String message
-    ) {
-        long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(10);
-        while (System.nanoTime() < deadline) {
-            // Count first: two interrupts observed while polling means the storm landed no matter
-            // how quickly the wait ends afterwards, so a budget expiry seen next is not a failure.
-            if (interruptCount.get() > 1) {
-                return;
-            }
-            if (!closerStillWaiting.getAsBoolean()) {
-                Assert.fail(message + "; interrupts landed: " + interruptCount.get());
-            }
-            Thread.yield();
-        }
-        Assert.fail(message + "; interrupts landed: " + interruptCount.get());
-    }
-
     private static void awaitOrFail(CountDownLatch latch, String message) {
         try {
             if (!latch.await(10, TimeUnit.SECONDS)) {
@@ -610,6 +580,26 @@ private static Sender fakeSender(
                 });
     }
 
+    private static boolean interruptUntilClose(
+            Thread closer,
+            Semaphore closeProgress,
+            AtomicBoolean isCloseComplete
+    ) throws InterruptedException {
+        long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5);
+        closer.interrupt();
+        while (!isCloseComplete.get()) {
+            long remainingNanos = deadline - System.nanoTime();
+            if (remainingNanos <= 0
+                    || !closeProgress.tryAcquire(remainingNanos, TimeUnit.NANOSECONDS)) {
+                return false;
+            }
+            if (!isCloseComplete.get()) {
+                closer.interrupt();
+            }
+        }
+        return true;
+    }
+
     private static QuestDBImpl newQuestDB(
             int senderMin,
             int queryMin,