test(pool): make close interrupt deadline checks deterministic

Track the bounded creation wait and handled interrupts in one atomic test witness. Handshake repeated interrupts until the original wait exits so the regression tests reject deadline resets without depending on scheduler timing.

Author: jerrinot

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

@@ -97,6 +97,12 @@ 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;
+    // Atomic test witness for the bounded creation-wait loop. The sign bit
+    // means active; the remaining bits count handled InterruptedExceptions.
+    // Condition.hasWaiters() cannot serve this purpose because an interrupt
+    // transiently transfers the closer out of the condition queue. Volatile
+    // lets white-box tests observe active/count as one coherent snapshot.
+    private volatile long creationWaitState;
     private int inFlightCreations;
 
     public QueryClientPool(
@@ -339,13 +345,21 @@ public void close() {
             final long creationWaitDeadlineNanos = System.nanoTime() + creationWaitNanos;
             long creationRemainingNanos = creationWaitNanos;
             boolean creationWaitInterrupted = false;
-            while (inFlightCreations > 0 && creationRemainingNanos > 0) {
-                try {
-                    creationFinished.awaitNanos(creationRemainingNanos);
-                } catch (InterruptedException e) {
-                    creationWaitInterrupted = true;
+            creationWaitState = inFlightCreations > 0 && creationRemainingNanos > 0
+                    ? Long.MIN_VALUE
+                    : 0;
+            try {
+                while (inFlightCreations > 0 && creationRemainingNanos > 0) {
+                    try {
+                        creationFinished.awaitNanos(creationRemainingNanos);
+                    } catch (InterruptedException e) {
+                        creationWaitInterrupted = true;
+                        creationWaitState++;
+                    }
+                    creationRemainingNanos = creationWaitDeadlineNanos - System.nanoTime();
                 }
-                creationRemainingNanos = creationWaitDeadlineNanos - System.nanoTime();
+            } finally {
+                creationWaitState &= Long.MAX_VALUE;
             }
             if (creationWaitInterrupted) {
                 Thread.currentThread().interrupt();
@@ -528,11 +542,16 @@ void release(QueryWorker w, long gen) {
         }
     }
 
+    /**
+     * Returns whether {@link #close()} is inside its bounded wait for
+     * in-flight creations. This remains true while an interrupt transfers the
+     * closer out of the condition queue and it re-enters with the same deadline.
+     */
     @TestOnly
     public boolean hasCreationWaiterForTesting() {
         lock.lock();
         try {
-            return lock.hasWaiters(creationFinished);
+            return creationWaitState < 0;
         } finally {
             lock.unlock();
         }

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

@@ -238,6 +238,12 @@ public final class SenderPool implements AutoCloseable {
     // (cancelling recovery) WITHOUT making a later close() short-circuit the
     // teardown. Guarded by lock.
     private boolean closeStarted;
+    // Atomic test witness for the bounded creation-wait loop. The sign bit
+    // means active; the remaining bits count handled InterruptedExceptions.
+    // Condition.hasWaiters() cannot serve this purpose because an interrupt
+    // transiently transfers the closer out of the condition queue. Volatile
+    // lets white-box tests observe active/count as one coherent snapshot.
+    private volatile long creationWaitState;
     private int inFlightCreations;
     // Lease teardowns currently running on borrower threads (retireLease's
     // delegate-close section, outside the lock). close() counts these as
@@ -1339,11 +1345,16 @@ public Thread getStartupRecoveryThreadForTesting() {
         return startupRecoveryThread;
     }
 
+    /**
+     * Returns whether {@link #close()} is inside its bounded wait for
+     * in-flight creations. This remains true while an interrupt transfers the
+     * closer out of the condition queue and it re-enters with the same deadline.
+     */
     @TestOnly
     public boolean hasCreationWaiterForTesting() {
         lock.lock();
         try {
-            return lock.hasWaiters(creationFinished);
+            return creationWaitState < 0;
         } finally {
             lock.unlock();
         }
@@ -1483,13 +1494,21 @@ public void close() {
             final long creationWaitDeadlineNanos = System.nanoTime() + creationWaitNanos;
             long creationRemainingNanos = creationWaitNanos;
             boolean creationWaitInterrupted = false;
-            while (inFlightCreations > 0 && creationRemainingNanos > 0) {
-                try {
-                    creationFinished.awaitNanos(creationRemainingNanos);
-                } catch (InterruptedException e) {
-                    creationWaitInterrupted = true;
+            creationWaitState = inFlightCreations > 0 && creationRemainingNanos > 0
+                    ? Long.MIN_VALUE
+                    : 0;
+            try {
+                while (inFlightCreations > 0 && creationRemainingNanos > 0) {
+                    try {
+                        creationFinished.awaitNanos(creationRemainingNanos);
+                    } catch (InterruptedException e) {
+                        creationWaitInterrupted = true;
+                        creationWaitState++;
+                    }
+                    creationRemainingNanos = creationWaitDeadlineNanos - System.nanoTime();
                 }
-                creationRemainingNanos = creationWaitDeadlineNanos - System.nanoTime();
+            } finally {
+                creationWaitState &= Long.MAX_VALUE;
             }
             if (creationWaitInterrupted) {
                 Thread.currentThread().interrupt();

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

@@ -43,13 +43,18 @@
 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.concurrent.locks.LockSupport;
 import java.util.function.Consumer;
 import java.util.function.IntFunction;
+import java.util.function.LongSupplier;
 
 public class QuestDBImplCloseLifecycleTest {
 
+    private static final long QUERY_CREATION_WAIT_STATE_OFFSET =
+            Unsafe.getFieldOffset(QueryClientPool.class, "creationWaitState");
     private static final String QUERY_CFG = "ws::addr=127.0.0.1:9000;";
+    private static final long SENDER_CREATION_WAIT_STATE_OFFSET =
+            Unsafe.getFieldOffset(SenderPool.class, "creationWaitState");
     private static final String SENDER_CFG = "http::addr=127.0.0.1:1;protocol_version=2;auto_flush=off;";
 
     @Test(timeout = 30_000)
@@ -186,18 +191,14 @@ 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, not 100ms, leaves enough room to
+            // establish the handshaked interrupt loop before asserting the
+            // original absolute deadline.
             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();
             Thread borrower = new Thread(() -> {
                 try {
                     db.borrowQuery();
@@ -209,13 +210,6 @@ public void facadeCloseIsBoundedUnderRepeatedInterruptsDuringQueryCreation() thr
                 db.close();
                 closeReturnedInterrupted.set(Thread.currentThread().isInterrupted());
             }, "interrupted-query-closer");
-            Thread interrupter = new Thread(() -> {
-                while (keepInterrupting.get()) {
-                    interruptCount.incrementAndGet();
-                    closer.interrupt();
-                    Thread.yield();
-                }
-            }, "query-close-interrupter");
             long nativeBaseline = Unsafe.getMemUsedByTag(MemoryTag.NATIVE_DEFAULT);
             try {
                 borrower.start();
@@ -226,14 +220,17 @@ 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");
+                long handledInterrupts = interruptUntilCreationWaitEnds(
+                        closer,
+                        () -> getCreationWaitState(pool),
+                        "query creation wait did not honor its absolute deadline");
                 closer.join(TimeUnit.SECONDS.toMillis(5));
                 Assert.assertFalse(
                         "repeated interrupts restarted the query creation-wait deadline",
                         closer.isAlive());
-                Assert.assertTrue("test did not repeatedly interrupt query close", interruptCount.get() > 1);
+                Assert.assertTrue(
+                        "query close did not handle both interrupts",
+                        handledInterrupts > 1);
                 Assert.assertTrue("facade close must restore query closer interruption",
                         closeReturnedInterrupted.get());
                 Assert.assertEquals(
@@ -252,9 +249,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 +270,13 @@ 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.
+            // Use the same 1s budget and handshaked interrupt loop as the query
+            // lifecycle 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();
             Thread borrower = new Thread(() -> {
                 try {
                     db.borrowSender();
@@ -295,13 +288,6 @@ public void facadeCloseIsBoundedUnderRepeatedInterruptsDuringSenderCreation() th
                 db.close();
                 closeReturnedInterrupted.set(Thread.currentThread().isInterrupted());
             }, "interrupted-sender-closer");
-            Thread interrupter = new Thread(() -> {
-                while (keepInterrupting.get()) {
-                    interruptCount.incrementAndGet();
-                    closer.interrupt();
-                    Thread.yield();
-                }
-            }, "sender-close-interrupter");
             try {
                 borrower.start();
                 Assert.assertTrue("sender borrow never reached construction",
@@ -313,14 +299,17 @@ 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");
+                long handledInterrupts = interruptUntilCreationWaitEnds(
+                        closer,
+                        () -> getCreationWaitState(pool),
+                        "sender creation wait did not honor its absolute deadline");
                 closer.join(TimeUnit.SECONDS.toMillis(5));
                 Assert.assertFalse(
                         "repeated interrupts restarted the sender creation-wait deadline",
                         closer.isAlive());
-                Assert.assertTrue("test did not repeatedly interrupt sender close", interruptCount.get() > 1);
+                Assert.assertTrue(
+                        "sender close did not handle both interrupts",
+                        handledInterrupts > 1);
                 Assert.assertTrue("facade close must restore sender closer interruption",
                         closeReturnedInterrupted.get());
                 Assert.assertEquals(
@@ -341,9 +330,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,32 +516,69 @@ private static void awaitCreationWaiter(SenderPool pool, String message) {
         Assert.fail(message);
     }
 
+    private static long getCreationWaitState(QueryClientPool pool) {
+        return Unsafe.getUnsafe().getLongVolatile(pool, QUERY_CREATION_WAIT_STATE_OFFSET);
+    }
+
+    private static long getCreationWaitState(SenderPool pool) {
+        return Unsafe.getUnsafe().getLongVolatile(pool, SENDER_CREATION_WAIT_STATE_OFFSET);
+    }
+
     /**
-     * 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.
+     * Delivers one interrupt at a time, waiting until the creation-wait loop
+     * handles each one before delivering the next. This prevents interrupt
+     * coalescing while keeping pressure on the wait until its original absolute
+     * deadline expires. The state snapshot encodes active in the sign bit and
+     * the handled-interrupt count in the remaining bits.
      */
-    private static void awaitRepeatedInterrupts(
-            AtomicInteger interruptCount,
-            BooleanSupplier closerStillWaiting,
+    private static long interruptUntilCreationWaitEnds(
+            Thread closer,
+            LongSupplier creationWaitState,
             String message
     ) {
-        long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(10);
+        long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5);
         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;
+            long state = creationWaitState.getAsLong();
+            long handledInterrupts = state & Long.MAX_VALUE;
+            if (state >= 0) {
+                Assert.assertTrue(
+                        message + "; wait ended after handling only " + handledInterrupts + " interrupt(s)",
+                        handledInterrupts > 1);
+                return handledInterrupts;
             }
-            if (!closerStillWaiting.getAsBoolean()) {
-                Assert.fail(message + "; interrupts landed: " + interruptCount.get());
+
+            closer.interrupt();
+            long countBeforeInterrupt = handledInterrupts;
+            while (System.nanoTime() < deadline) {
+                state = creationWaitState.getAsLong();
+                handledInterrupts = state & Long.MAX_VALUE;
+                if (state >= 0) {
+                    Assert.assertTrue(
+                            message + "; wait ended after handling only " + handledInterrupts + " interrupt(s)",
+                            handledInterrupts > 1);
+                    return handledInterrupts;
+                }
+                if (handledInterrupts > countBeforeInterrupt) {
+                    // Keep enough interrupt pressure to expose a restarted
+                    // deadline without monopolising a CI worker core.
+                    LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(1));
+                    break;
+                }
+                Thread.yield();
             }
-            Thread.yield();
         }
-        Assert.fail(message + "; interrupts landed: " + interruptCount.get());
+
+        long state = creationWaitState.getAsLong();
+        long handledInterrupts = state & Long.MAX_VALUE;
+        if (state >= 0) {
+            Assert.assertTrue(
+                    message + "; wait ended after handling only " + handledInterrupts + " interrupt(s)",
+                    handledInterrupts > 1);
+            return handledInterrupts;
+        }
+        Assert.fail(message + "; creation wait still active after handling "
+                + handledInterrupts + " interrupt(s)");
+        return 0;
     }
 
     private static void awaitOrFail(CountDownLatch latch, String message) {