Make final-probe regression deterministic

Release the retired slot from a test-only hook after the positive borrow wait has actually exhausted its budget. This removes the scheduler-dependent sleep and guarantees the test reaches the final post-wait probe.

Author: bluestreak01

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

@@ -148,6 +148,11 @@ public final class SenderPool implements AutoCloseable {
     private final Condition slotReleased;
     // True iff the configuration enables store-and-forward (sf_dir set).
     private final boolean storeAndForward;
+    // Test seam: runs after a capacity-starved borrow's condition wait has
+    // exhausted its positive timeout, before the loop's terminal pass. Null in
+    // production; regression tests release a retired slot here to prove that
+    // the terminal pass re-probes returned capacity before throwing.
+    private volatile Runnable borrowWaitExpiredHook;
     // 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
@@ -818,6 +823,12 @@ public PooledSender borrow() {
                 }
                 try {
                     remainingNanos = slotReleased.awaitNanos(remainingNanos);
+                    if (remainingNanos <= 0) {
+                        Runnable hook = borrowWaitExpiredHook;
+                        if (hook != null) {
+                            hook.run();
+                        }
+                    }
                 } catch (InterruptedException e) {
                     Thread.currentThread().interrupt();
                     throw new LineSenderException("interrupted while waiting for a Sender from the pool");
@@ -830,6 +841,11 @@ public PooledSender borrow() {
         }
     }
 
+    @TestOnly
+    public void setBorrowWaitExpiredHook(Runnable hook) {
+        this.borrowWaitExpiredHook = hook;
+    }
+
     /**
      * Raises the shutdown signal early -- without tearing down live delegates --
      * so an in-flight startup-recovery step driven on the {@link PoolHousekeeper}

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

@@ -1290,14 +1290,13 @@ public void testZeroTimeoutBorrowProbesRetiredSlotBeforeThrowing() throws Except
 
     @Test
     public void testParkedBorrowerGetsFinalProbeAfterBudgetExpiry() throws Exception {
-        // Release-during-wait twin of the zero-timeout test. A borrower parks
-        // in awaitNanos while the retired slot's flock is genuinely held; the
-        // deferred cleanup then releases the flock mid-wait. Nothing signals
-        // slotReleased (the release is delegate-side, volatile writes only), so
-        // the borrower sleeps out its full budget. Pre-fix its wake-up pass hit
-        // the terminal timeout check before reprobeRetiredSlots() and threw --
-        // missing capacity that had already come back. Post-fix the wake-up
-        // pass probes first, recovers the index, and the creation is admitted.
+        // Positive-timeout twin of the zero-timeout test. A borrower parks in
+        // awaitNanos while the retired slot's flock is genuinely held and
+        // sleeps out its full budget. A test hook releases the flock after the
+        // wait reports expiry but before the terminal loop pass. Pre-fix that
+        // pass hit the timeout check before reprobeRetiredSlots() and threw --
+        // missing capacity that had already come back. Post-fix the terminal
+        // pass probes first, recovers the index, and admits the creation.
         TestUtils.assertMemoryLeak(() -> {
             // Phase 1: strand unacked data under default-0.
             try (TestWebSocketServer silent = new TestWebSocketServer(new SilentHandler())) {
@@ -1318,8 +1317,9 @@ public void testParkedBorrowerGetsFinalProbeAfterBudgetExpiry() throws Exception
             Assert.assertTrue("unacked data must persist under default-0",
                     hasSegmentFile(slot("default-0")));
 
-            // Phase 2: maxSize=1, generous budget so the mid-wait release lands
-            // comfortably inside it.
+            // Phase 2: maxSize=1 and a positive acquire budget. A test hook
+            // releases the flock only after awaitNanos() has returned with that
+            // budget exhausted, so the terminal wake-up pass is deterministic.
             CountingAckHandler handler = new CountingAckHandler();
             try (TestWebSocketServer ack = new TestWebSocketServer(handler)) {
                 int ackPort = ack.getPort();
@@ -1340,49 +1340,38 @@ public void testParkedBorrowerGetsFinalProbeAfterBudgetExpiry() throws Exception
                     return real;
                 };
 
-                try (SenderPool pool = newPoolWithFactory(cfg2, 0, 1, 2_000, factory)) {
+                try (SenderPool pool = newPoolWithFactory(cfg2, 0, 1, 100, factory)) {
                     Assert.assertNotNull("recovery must have built slot 0", forged.get());
                     Assert.assertEquals("precondition: startup recovery must retire the slot",
                             1, pool.leakedSlotCount());
 
-                    // Borrower parks: its pre-park probe correctly fails while
-                    // the flock is genuinely held.
-                    AtomicReference<Throwable> failure = new AtomicReference<>();
-                    AtomicReference<PooledSender> borrowed = new AtomicReference<>();
-                    Thread borrower = new Thread(() -> {
+                    AtomicBoolean waitExpired = new AtomicBoolean();
+                    pool.setBorrowWaitExpiredHook(() -> {
+                        Assert.assertTrue("expired-wait hook must run exactly once",
+                                waitExpired.compareAndSet(false, true));
+                        Sender recoverer = forged.get();
                         try {
-                            borrowed.set(pool.borrow());
-                        } catch (Throwable e) {
-                            failure.set(e);
+                            setBooleanField(recoverer, "closed", false);
+                        } catch (Exception e) {
+                            throw new RuntimeException(e);
                         }
+                        // The flock drops only after the positive awaitNanos()
+                        // budget is exhausted. This delegate-side release does
+                        // not signal slotReleased.
+                        recoverer.close();
                     });
-                    borrower.start();
-
-                    // Let the borrower enter borrow() and park. If a CI stall
-                    // delays it past the release below, the pre-park probe
-                    // recovers instead and the test degrades to a pass -- never
-                    // a flaky failure.
-                    Thread.sleep(300);
-
-                    // Mid-wait release: flock drops, no signal reaches the pool.
-                    Sender recoverer = forged.get();
-                    setBooleanField(recoverer, "closed", false);
-                    recoverer.close();
-
-                    borrower.join(10_000);
-                    Assert.assertFalse("borrower must have finished", borrower.isAlive());
-                    if (failure.get() != null) {
-                        throw new AssertionError(
-                                "borrower must recover the retired slot on its wake-up pass, not time out",
-                                failure.get());
-                    }
-                    PooledSender b = borrowed.get();
-                    Assert.assertNotNull("borrower must have obtained a sender", b);
                     try {
-                        Assert.assertEquals("wake-up probe must recover the retired slot's capacity",
-                                0, pool.leakedSlotCount());
+                        PooledSender b = pool.borrow();
+                        try {
+                            Assert.assertTrue("borrow must exhaust its positive wait budget",
+                                    waitExpired.get());
+                            Assert.assertEquals("wake-up probe must recover the retired slot's capacity",
+                                    0, pool.leakedSlotCount());
+                        } finally {
+                            b.close();
+                        }
                     } finally {
-                        b.close();
+                        pool.setBorrowWaitExpiredHook(null);
                     }
                 }
             }