fix(qwp): retain startup-retired recoverers so a late flock release restores pool capacity

isSlotLockReleased() is no longer a one-shot snapshot: deferred engine
cleanup on a worker/I/O-thread exit path can release the SF slot flock
after close() returned. The runtime reclaim paths (discardBroken/reapIdle
via reclaimSlot) already keep such slots in retiredSlots and re-probe
them, but the in-range startup-recovery pass only ticked leakedSlots and
dropped the recoverer, making the retirement permanent even after the
release -- fatal at maxSize=1, where every later borrow timed out until
process restart.

Hand the retained recoverer out of drainCandidateSlotForRecovery
(retainedOut replaces the flockHeld boolean) and add it to retiredSlots
alongside the leakedSlots tick, so the existing reprobeRetiredSlots()
drivers (capacity-starved borrow, housekeeper tick) recover the capacity
once the worker finally exits. Out-of-range recoverers stay excluded:
they carry no leakedSlots tick and freeSlotIndex(idx) would index past
the slotInUse array.

Author: bluestreak01

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

@@ -179,12 +179,15 @@ public final class SenderPool implements AutoCloseable {
     // retiredSlots and returns any index whose flock has since dropped.
     // Guarded by lock; only ever ticks for SF slots.
     private int leakedSlots;
-    // The retired slots behind the leakedSlots count (runtime reclaim paths
-    // only; startup-recovery retirements stay permanent — their transient
-    // recoverer sender is out of scope by retire time). Re-probed by
-    // reprobeRetiredSlots() so a late flock release (deferred engine cleanup
-    // on a worker exit path) restores the pool's capacity instead of
-    // ratcheting it down until process exit. Guarded by lock.
+    // The retired slots behind the leakedSlots count: runtime reclaim paths
+    // (discardBroken/reapIdle via reclaimSlot) and the in-range startup-
+    // recovery pass (recoverOneSlotStep, which retains the recoverer slot for
+    // exactly this purpose). Re-probed by reprobeRetiredSlots() so a late
+    // flock release (deferred engine cleanup on a worker exit path) restores
+    // the pool's capacity instead of ratcheting it down until process exit.
+    // Out-of-range startup recoverers are NEVER added: they carry no
+    // leakedSlots tick and their index has no slotInUse entry to free.
+    // Guarded by lock.
     private final ArrayList<SenderSlot> retiredSlots = new ArrayList<>();
     // SF slots currently held by the in-range startup-recovery pass
     // (recoverOneSlotStep): each is reserved under `lock` for the
@@ -480,7 +483,7 @@ private boolean recoverOneSlotStep(long stepBudgetMillis) {
             recoveryComplete = true;
             return false;
         }
-        final boolean[] flockHeld = new boolean[1];
+        final SenderSlot[] retained = new SenderSlot[1];
 
         // Pass 1: in-range managed slots [0, maxSize). Skip live and empty slots
         // cheaply; spend the step on the first slot that actually holds data.
@@ -526,23 +529,29 @@ private boolean recoverOneSlotStep(long stepBudgetMillis) {
             // A real candidate -> spend the step on it. Advance the cursor first
             // so a resume never reprocesses this index.
             recoveryInRangeNext++;
-            boolean stopScan = drainCandidateSlotForRecovery(i, slotPath, stepBudgetMillis, flockHeld);
+            boolean stopScan = drainCandidateSlotForRecovery(i, slotPath, stepBudgetMillis, retained);
             lock.lock();
             try {
                 // Release the recovery reservation accounting; from here either
                 // leakedSlots (retire) or the freed index carries the cap math.
                 recoveringSlots--;
-                if (flockHeld[0]) {
+                if (retained[0] != null) {
                     // close() retained the flock because an I/O or manager
-                    // worker did not stop. Retire the slot permanently (mirror
+                    // worker did not stop. Retire the slot (mirror
                     // discardBroken/reapIdle): keep slotInUse[i] set and count it
                     // in leakedSlots so the borrow() cap math accounts for the
                     // lost capacity and no later borrow ever reuses the
-                    // still-locked dir.
+                    // still-locked dir. Keep the recoverer in retiredSlots so
+                    // reprobeRetiredSlots() restores the capacity once the
+                    // deferred engine cleanup releases the flock — without it
+                    // the retirement would be permanent even after the release
+                    // (fatal at maxSize=1: every later borrow would time out).
                     leakedSlots++;
-                    LOG.warn("startup SF recovery: slot {} retired permanently: delegate close() returned with "
+                    retiredSlots.add(retained[0]);
+                    LOG.warn("startup SF recovery: slot {} retired: delegate close() returned with "
                                     + "the flock still held (I/O or manager worker did not stop); pool capacity reduced by 1, "
-                                    + "now {} of {} usable [leakedSlots={}]",
+                                    + "now {} of {} usable [leakedSlots={}]; the slot is re-probed and recovered "
+                                    + "if the worker releases the flock later",
                             i, maxSize - leakedSlots, maxSize, leakedSlots);
                 } else {
                     slotInUse[i] = false;
@@ -586,15 +595,17 @@ private boolean recoverOneSlotStep(long stepBudgetMillis) {
             if (!OrphanScanner.isCandidateOrphan(slotPath)) {
                 continue;
             }
-            boolean stopScan = drainCandidateSlotForRecovery(idx, slotPath, stepBudgetMillis, flockHeld);
-            if (flockHeld[0]) {
+            boolean stopScan = drainCandidateSlotForRecovery(idx, slotPath, stepBudgetMillis, retained);
+            if (retained[0] != null) {
                 // Out of the pool's [0, maxSize) capacity range: there is no
                 // slotInUse entry to retire and no future borrow targets this
                 // dir, so a still-held flock only leaks this recoverer's
                 // worker-reachable resources (a best-effort teardown loss,
                 // logged). Crucially we do
                 // NOT touch leakedSlots -- that would wrongly shrink the
-                // in-range pool capacity.
+                // in-range pool capacity -- and we do NOT add to retiredSlots:
+                // there is no capacity to recover, and freeSlotIndex(idx)
+                // would index past the slotInUse array (sized maxSize).
                 LOG.warn("startup SF recovery: out-of-range slot {} closed with the flock still held "
                                 + "(I/O or manager worker did not stop); its data is durable on disk for a later attempt",
                         slotPath);
@@ -616,17 +627,20 @@ private boolean recoverOneSlotStep(long stepBudgetMillis) {
      * (whose {@link #defaultSender} derives the dir {@code <base>-slotIndex}),
      * drains its unacked data, and closes the delegate. Shared by both recovery
      * passes -- the in-range pass and the out-of-range pass -- which differ only
-     * in their slot bookkeeping, handled by the caller via {@code flockHeld}.
+     * in their slot bookkeeping, handled by the caller via {@code retainedOut}.
      *
-     * @param flockHeld single-element out-param set to {@code true} iff a
-     *                  recoverer was built and its {@code close()} returned with
-     *                  the flock still held because a worker did not stop
+     * @param retainedOut single-element out-param set to the recoverer iff one
+     *                    was built and its {@code close()} returned with the
+     *                    flock still held because a worker did not stop; the
+     *                    in-range caller keeps it in {@link #retiredSlots} so a
+     *                    late flock release can be re-probed. {@code null} when
+     *                    the flock was released (or no recoverer was built).
      * @return {@code true} if a build/drain failure occurred that will very
      * likely repeat for every remaining slot, so the caller should stop scanning
      */
     private boolean drainCandidateSlotForRecovery(int slotIndex, String slotPath,
-                                                  long remainingMillis, boolean[] flockHeld) {
-        flockHeld[0] = false;
+                                                  long remainingMillis, SenderSlot[] retainedOut) {
+        retainedOut[0] = null;
         // Hoisted so the flock check after the try can consult it:
         // createRecoverer() takes the slot flock on <base>-slotIndex, and
         // delegate().close() can retain it when an I/O or manager worker does
@@ -678,8 +692,8 @@ private boolean drainCandidateSlotForRecovery(int slotIndex, String slotPath,
             LOG.warn("startup SF recovery: scan failed for slot {} ({})",
                     slotPath, scanErr.toString());
         }
-        if (recoverer != null) {
-            flockHeld[0] = !flockReleased(recoverer);
+        if (recoverer != null && !flockReleased(recoverer)) {
+            retainedOut[0] = recoverer;
         }
         return stopScan;
     }

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

@@ -875,7 +875,7 @@ public void testStartupRecoveryRetiresSlotWhenRecovererCloseLeavesFlockHeld() th
         // C1 regression: the startup recovery loop MUST mirror discardBroken /
         // reapIdle. When a recoverer's delegate close() returns with the SF
         // flock still held (the I/O thread refused to stop), the recovered slot
-        // index must be retired permanently (leakedSlots++, slotInUse stays
+        // index must be retired (leakedSlots++, slotInUse stays
         // set) -- NOT freed. Freeing it would let a later borrow re-pick the
         // still-locked dir and resurrect "sf slot already in use", the exact
         // failure class this PR exists to kill. Pre-fix the recovery finally
@@ -951,7 +951,7 @@ public void testStartupRecoveryRetiresSlotWhenRecovererCloseLeavesFlockHeld() th
                     try {
                         Assert.assertTrue("borrow must use a fresh slot dir",
                                 Files.exists(slot("default-1")));
-                        // Capacity is permanently reduced by the leaked slot:
+                        // Capacity stays reduced while the flock is held:
                         // max=2, one leaked + one live => the next borrow times
                         // out rather than colliding on the locked dir.
                         try {
@@ -979,6 +979,108 @@ public void testStartupRecoveryRetiresSlotWhenRecovererCloseLeavesFlockHeld() th
         });
     }
 
+    @Test
+    public void testStartupRetiredSlotRecoveredAfterLateFlockRelease() throws Exception {
+        // Recovery twin of
+        // testStartupRecoveryRetiresSlotWhenRecovererCloseLeavesFlockHeld: a
+        // slot retired by STARTUP recovery (recoverer close() returned with
+        // the flock still held) must not stay lost until process exit.
+        // isSlotLockReleased() is no longer a one-shot snapshot -- deferred
+        // engine cleanup on a worker exit path can release the flock later --
+        // so the pool must keep the recoverer in retiredSlots and re-probe it.
+        // Pre-fix, startup recovery only ticked leakedSlots and dropped the
+        // recoverer: at maxSize=1 every later borrow timed out forever even
+        // after the flock dropped. This test is RED until the recoverer is
+        // retained.
+        TestUtils.assertMemoryLeak(() -> {
+            // Phase 1: strand unacked data under default-0 so startup recovery
+            // treats it as a candidate orphan and builds a recoverer.
+            try (TestWebSocketServer silent = new TestWebSocketServer(new SilentHandler())) {
+                int silentPort = silent.getPort();
+                silent.start();
+                Assert.assertTrue(silent.awaitStart(5, TimeUnit.SECONDS));
+                String cfg1 = "ws::addr=localhost:" + silentPort + ";sf_dir=" + sfDir
+                        + ";close_flush_timeout_millis=500;";
+                try (SenderPool pool = new SenderPool(cfg1, 1, 1, 1_000, Long.MAX_VALUE, Long.MAX_VALUE)) {
+                    PooledSender s = pool.borrow();
+                    for (int i = 0; i < 3; i++) {
+                        s.table("recover").longColumn("v", i).atNow();
+                        s.flush();
+                    }
+                    s.close();
+                }
+            }
+            Assert.assertTrue("unacked data must persist under default-0",
+                    hasSegmentFile(slot("default-0")));
+
+            // Phase 2: maxSize=1 -- the worst case, where the single slot's
+            // retirement starves the whole pool. Forge the retention exactly
+            // like the retire test (closed=true makes the recovery drain and
+            // close a no-op, so the real flock stays held and slotLockReleased
+            // stays false), but only for the FIRST build of slot 0: the
+            // post-recovery borrow below must get a working delegate on the
+            // recovered index.
+            CountingAckHandler handler = new CountingAckHandler();
+            try (TestWebSocketServer ack = new TestWebSocketServer(handler)) {
+                int ackPort = ack.getPort();
+                ack.start();
+                Assert.assertTrue(ack.awaitStart(5, TimeUnit.SECONDS));
+                String cfg2 = "ws::addr=localhost:" + ackPort + ";sf_dir=" + sfDir + ";";
+
+                AtomicReference<Sender> forged = new AtomicReference<>();
+                IntFunction<Sender> factory = idx -> {
+                    Sender real = Sender.builder(cfg2).senderId("default-" + idx).build();
+                    if (idx == 0 && forged.compareAndSet(null, real)) {
+                        try {
+                            setBooleanField(real, "closed", true);
+                        } catch (Exception e) {
+                            throw new RuntimeException(e);
+                        }
+                    }
+                    return real;
+                };
+
+                try (SenderPool pool = newPoolWithFactory(cfg2, 0, 1, 500, factory)) {
+                    Assert.assertNotNull("recovery must have built slot 0", forged.get());
+                    Assert.assertEquals("precondition: startup recovery must retire the slot",
+                            1, pool.leakedSlotCount());
+
+                    // While the flock is genuinely held, borrows time out: the
+                    // cap-check re-probe finds the flock still reported held.
+                    try {
+                        pool.borrow();
+                        Assert.fail("borrow must time out while the slot is retired");
+                    } catch (LineSenderException e) {
+                        Assert.assertTrue(e.getMessage(), e.getMessage().contains("timed out"));
+                    }
+
+                    // The late release: the "wedged worker" finishes and the
+                    // flock genuinely drops (un-forge and close the recoverer
+                    // for real; in production this flip comes from
+                    // isSlotLockReleased() re-probing the retained engine after
+                    // the worker exited).
+                    Sender recoverer = forged.get();
+                    setBooleanField(recoverer, "closed", false);
+                    recoverer.close();
+
+                    // The capacity-starved borrow must re-probe the startup-
+                    // retired slot, recover its capacity, and create on the
+                    // freed index -- no housekeeper tick required.
+                    PooledSender b = pool.borrow();
+                    try {
+                        Assert.assertEquals("borrow must recover the startup-retired slot's capacity",
+                                0, pool.leakedSlotCount());
+                        boolean[] slotInUse = (boolean[]) getField(pool, "slotInUse");
+                        Assert.assertTrue("recovered index 0 must carry the new borrow", slotInUse[0]);
+                        Assert.assertEquals(1, countSlotDirs());
+                    } finally {
+                        b.close();
+                    }
+                }
+            }
+        });
+    }
+
     // ----------------------------------------------------------------------
     // Concurrency stress: borrow/return churn must never collide on a slot.
     // ----------------------------------------------------------------------