fix(qwp): move closed-mid-creation delegate teardown outside the pool lock

When borrow()/acquire() re-locks after building a fresh delegate and finds
the pool closed, both pools tore the delegate down while holding the pool
lock. A delegate close() can block for seconds (5s ack drain, ~3s drainer-
pool wind-down, 5s dispatch-thread join) or longer (unbounded I/O-thread
latch await behind an OS-level connect when connect_timeout=0), stalling
close(), giveBack/retireLease, reapIdle and watchdog cancels behind it --
violating retireLease's close-outside-the-lock contract on the ingest side
and cancelIfCurrent's held-only-briefly contract on the egress side.

SenderPool now mirrors retireLease: accounting under the lock (move the SF
slot reservation from inFlightCreations to closingSlots, bump
pendingLeaseTeardowns so close()'s outstanding-teardown wait sees the
in-flight close), delegate close outside, re-lock to reclaim the slot and
signal. QueryClientPool simply unlocks before shutdown(): the fresh worker
never entered `all`, so close()'s snapshot loop cannot race the teardown.

RED/GREEN regression tests park a fake delegate mid-close and prove the
pool lock stays free (probe close() completes), close() still waits for the
teardown to finish, and the closed-mid-creation worker teardown leaks no
native scratch.

Author: bluestreak01

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

@@ -255,6 +255,16 @@ public QueryWorker acquire() {
                     lock.lock();
                     inFlightCreations--;
                     if (closed) {
+                        // Pool was closed mid-creation -- tear the fresh worker
+                        // down rather than leaking it, but OUTSIDE the lock:
+                        // shutdown() joins the dispatch thread for up to
+                        // SHUTDOWN_JOIN_MILLIS, and close()/release()/discard()/
+                        // cancelIfCurrent() all contend on this lock (whose
+                        // contract is "held only briefly"). The accounting above
+                        // already ran under the lock, and the worker never
+                        // entered `all`, so close()'s snapshot loop cannot race
+                        // this teardown.
+                        lock.unlock();
                         try {
                             created.shutdown();
                         } catch (Throwable ignored) {

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

@@ -726,13 +726,41 @@ public PooledSender borrow() {
                     if (closed) {
                         // Pool was closed mid-creation -- destroy the new connection
                         // rather than leaking it. Other waiters have been signaled
-                        // by close() already.
-                        freeSlotIndex(slotIndex);
+                        // by close() already. The delegate is closed OUTSIDE the
+                        // lock (mirroring retireLease): its close() can block for
+                        // seconds (bounded ack drain, drainer-pool wind-down) or
+                        // longer (unbounded I/O-thread latch await behind an
+                        // OS-level connect), which held here would stall close(),
+                        // giveBack/retireLease and reapIdle behind the pool lock.
+                        // Accounting first, under the lock: for an SF slot the
+                        // index reservation moves from inFlightCreations to
+                        // closingSlots until the close below releases the flock,
+                        // and pendingLeaseTeardowns keeps the out-of-lock close
+                        // visible to close()'s outstanding-teardown wait.
+                        boolean reserved = created.slotIndex() >= 0;
+                        if (reserved) {
+                            closingSlots++;
+                        }
+                        pendingLeaseTeardowns++;
+                        lock.unlock();
                         try {
                             created.delegate().close();
                         } catch (Throwable ignored) {
                             // Best-effort: an Error (e.g. -ea AssertionError)
                             // from teardown must not mask the closed-pool signal.
+                        } finally {
+                            // Re-lock to reclaim the SF slot index and signal a
+                            // close() waiting on this teardown. MUST run even if
+                            // the delegate close threw, otherwise the slot stays
+                            // reserved forever and close() waits out its full
+                            // budget on a teardown that already happened.
+                            lock.lock();
+                            pendingLeaseTeardowns--;
+                            if (reserved) {
+                                reclaimSlot(created, " after closed-mid-creation teardown");
+                            }
+                            slotReleased.signalAll();
+                            lock.unlock();
                         }
                         throw new LineSenderException("QuestDB handle is closed");
                     }

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

@@ -34,7 +34,10 @@
 import org.junit.Assert;
 import org.junit.Test;
 
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Consumer;
 
 // Error-safety of the three QueryClientPool creation paths the teardown-hardening
@@ -221,6 +224,75 @@ public void preWarmDoesNotLeakNativeScratchOnErrorFromStart() throws Exception {
         });
     }
 
+    // Site: acquire()'s closed-mid-creation branch. When acquire() re-locks
+    // after building a fresh worker and finds the pool closed, it must tear the
+    // worker down on its own thread -- outside the pool lock (shutdown() joins
+    // the dispatch thread for up to SHUTDOWN_JOIN_MILLIS; cancelIfCurrent's
+    // contract is that this lock is "held only briefly") -- and still reclaim
+    // everything: the client's NATIVE_DEFAULT scratch and the creation slot.
+    // RED (teardown dropped in the restructure): the scratch leaks and the
+    //      baseline assertion fails.
+    // GREEN: shutdown() runs after the lock is released -> no leak, accounting
+    //        restored, acquire() surfaces the closed pool.
+    @Test(timeout = 30_000)
+    public void closedMidCreationTearsDownFreshWorkerWithoutLeak() throws Exception {
+        TestUtils.assertMemoryLeak(() -> {
+            CountDownLatch inConnect = new CountDownLatch(1);
+            CountDownLatch releaseConnect = new CountDownLatch(1);
+            Consumer<QwpQueryClient> connectHook = client -> {
+                inConnect.countDown();
+                try {
+                    if (!releaseConnect.await(10, TimeUnit.SECONDS)) {
+                        throw new IllegalStateException("test never released the parked connect");
+                    }
+                } catch (InterruptedException e) {
+                    Thread.currentThread().interrupt();
+                    throw new IllegalStateException("interrupted in parked connect", e);
+                }
+            };
+            // No-op startHook: the dispatch thread stays unstarted, so the
+            // closed-mid-creation shutdown()'s interrupt+join is instant.
+            QueryClientPool pool = newPool(CFG, 0, 1, 10_000, connectHook, w -> {
+            });
+            try {
+                long baseline = Unsafe.getMemUsedByTag(MemoryTag.NATIVE_DEFAULT);
+                AtomicReference<Throwable> acquireOutcome = new AtomicReference<>();
+                Thread acquirer = new Thread(() -> {
+                    try {
+                        pool.acquire();
+                    } catch (Throwable t) {
+                        acquireOutcome.set(t);
+                    }
+                }, "mid-creation-acquirer");
+                acquirer.start();
+                Assert.assertTrue("acquirer never reached connect()",
+                        inConnect.await(10, TimeUnit.SECONDS));
+
+                // Close the pool while the worker build is in flight (the fresh
+                // worker never entered `all`, so close()'s snapshot skips it),
+                // then let the build finish: acquire() must observe `closed`,
+                // tear the worker down on its own thread and throw.
+                pool.close();
+                releaseConnect.countDown();
+                acquirer.join(TimeUnit.SECONDS.toMillis(10));
+                Assert.assertFalse("acquirer did not finish", acquirer.isAlive());
+
+                Assert.assertTrue(
+                        "acquire() must surface the closed pool, got: " + acquireOutcome.get(),
+                        acquireOutcome.get() instanceof QueryException
+                                && String.valueOf(acquireOutcome.get().getMessage()).contains("closed"));
+                long after = Unsafe.getMemUsedByTag(MemoryTag.NATIVE_DEFAULT);
+                Assert.assertEquals(
+                        "closed-mid-creation acquire() leaked the fresh worker's NATIVE_DEFAULT scratch",
+                        baseline, after);
+                Assert.assertEquals("in-flight creation accounting must be restored",
+                        0, inFlightCreations(pool));
+            } finally {
+                pool.close();
+            }
+        });
+    }
+
     private static Consumer<QwpQueryClient> alwaysThrow() {
         return client -> {
             throw new AssertionError("injected native connect failure");