diff --git a/core/src/main/java/io/questdb/client/impl/QueryClientPool.java b/core/src/main/java/io/questdb/client/impl/QueryClientPool.java index 75323a12..ca3aec15 100644 --- a/core/src/main/java/io/questdb/client/impl/QueryClientPool.java +++ b/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 { diff --git a/core/src/main/java/io/questdb/client/impl/SenderPool.java b/core/src/main/java/io/questdb/client/impl/SenderPool.java index 2d01059e..91ed7019 100644 --- a/core/src/main/java/io/questdb/client/impl/SenderPool.java +++ b/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(); diff --git a/core/src/test/java/io/questdb/client/test/impl/QuestDBImplCloseLifecycleTest.java b/core/src/test/java/io/questdb/client/test/impl/QuestDBImplCloseLifecycleTest.java index 6344336b..3a92e25e 100644 --- a/core/src/test/java/io/questdb/client/test/impl/QuestDBImplCloseLifecycleTest.java +++ b/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 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 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, diff --git a/design/qwp-nack-policy-v2.md b/design/qwp-nack-policy-v2.md index fabe4507..2a9ff891 100644 --- a/design/qwp-nack-policy-v2.md +++ b/design/qwp-nack-policy-v2.md @@ -99,8 +99,13 @@ diagnostics only. The server already handles it at the right layer (Invariant B work): the read-only gate and the commit-path authorization refusal both set -`roleChangeClosePending` and close with a reconnect-eligible `NORMAL_CLOSURE` -instead of NACKing `SECURITY_ERROR` (`QwpIngressProcessorState`). The client +`roleChangeClosePending` and close with a reconnect-eligible +`NORMAL_CLOSURE` instead of NACKing `SECURITY_ERROR` +(`QwpIngressProcessorState`). A private-use code would let the server tell +the client's verbatim CLOSE echo apart from a voluntary client CLOSE that +crossed it on the wire, but deployed fleets classify anything outside +`NORMAL_CLOSURE`/`GOING_AWAY` as a poison strike, so that needs a +negotiated capability first. The client reconnects, hits the 421 role reject on the now-replica, and retries from SF until a primary is reachable. Consequently: