fix(qwp): count a retired slot with stranded data as a recovery deferral

The in-range recovery scan skipped any reserved index as "live" without a
deferral tick. A RETIRED index (wedged close() kept the slot flock) aliases
into that branch: the scan could finish a cycle with zero deferrals and latch
recoveryComplete while the retired slot's dir still held unacked durable
data. From there nothing in-process ever drained it -- reprobeRetiredSlots()
only restores capacity -- so the data waited for a restart or a lucky borrow
of that exact index, which steady low load may never produce.

Treat a retired index whose dir is still a candidate orphan as a deferral,
the same rule as a CONTENDED park: the end-of-scan rewind keeps the cycle
alive, and once the deferred cleanup releases the flock the scan reserves
the freed index and drains it itself -- no borrow required. Data-inert
(durable on disk; a restart already rescanned); this closes the drain-
liveness gap for the life of the pool.

The regression test drives the scan by hand: forge the wedged retire, walk
the scan (pre-fix it latched complete right here), heal the flock, and
assert the scan itself delivers the stranded data and only then completes.

Author: bluestreak01

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

@@ -287,6 +287,9 @@ public final class SenderPool implements AutoCloseable {
     // remaining slots, recoveryDeferredThisCycle records the park, and once
     // both passes finish with parked slots outstanding the cursors rewind for
     // another cycle on a later tick instead of latching recoveryComplete.
+    // A RETIRED index whose dir still holds data is deferred the same way on
+    // every walk (see the reserved-skip branch in recoverOneSlotStep) so the
+    // latch can never strand a retired slot's data while the pool lives.
     // recoveryFailStreak/-Slot track consecutive failures on one candidate;
     // recoveryWarnedSlots dedups the per-slot WARNs so an indefinitely
     // retried slot logs once per failure episode, not once per retry. All of
@@ -649,7 +652,12 @@ boolean runStartupRecoveryStep() {
      * can neither target the dir being recovered nor over-allocate past
      * {@code maxSize}. Prewarmed/borrowed slots (already live, holding their
      * flock) are skipped, as are empty slots (a cheap directory probe); only a
-     * slot that actually holds stranded data spends the step's single drain. The
+     * slot that actually holds stranded data spends the step's single drain. A
+     * RETIRED index (a wedged close() kept its flock; see {@link #reclaimSlot}
+     * and the retire branch below) is likewise never drained in place, but
+     * while its dir still holds data it counts as a deferral, so the scan keeps
+     * cycling instead of latching {@code recoveryComplete} past stranded data
+     * that only a restart or a lucky borrow of that index would ever deliver. The
      * out-of-range pass needs no reservation: those indices have no
      * {@code slotInUse} entry and are never allocated by borrow().
      * <p>
@@ -702,26 +710,49 @@ private boolean recoverOneSlotStep(long stepBudgetMillis) {
                 return false;
             }
             int i = recoveryInRangeNext;
+            String slotPath = sfDir + "/" + slotBaseId + "-" + i;
             // Reserve this index unless prewarm (or a concurrent borrow after
             // publication) already holds it live. Count the reservation in
             // recoveringSlots so the borrow() cap check cannot over-allocate
             // while this slot is held for recovery.
             boolean reserved;
+            boolean retired = false;
             lock.lock();
             try {
                 reserved = slotInUse[i];
                 if (!reserved) {
                     slotInUse[i] = true;
                     recoveringSlots++;
+                } else {
+                    retired = isRetiredSlotIndex(i);
                 }
             } finally {
                 lock.unlock();
             }
             if (reserved) {
+                // A reserved index is normally LIVE (prewarm or a concurrent
+                // borrow owns it) and is none of recovery's business. A RETIRED
+                // index is different: its flock is held by this pool's own
+                // wedged former delegate, and any unacked data in its dir is
+                // exactly as stranded as a CONTENDED slot's. Without counting
+                // it as a deferral the scan would latch recoveryComplete and
+                // abandon that data until a restart or a lucky borrow of this
+                // exact index -- which steady low load may never produce. Count
+                // it -- the same rule as a CONTENDED park -- so the end-of-scan
+                // rewind keeps the cycle alive; once the deferred engine
+                // cleanup releases the flock, reprobeRetiredSlots()/the release
+                // callback frees the index and a later cycle reserves and
+                // drains it right here. The isCandidateOrphan probe (a few
+                // syscalls, outside the lock) keeps an already-clean retired
+                // dir from cycling the scan forever, and racing a concurrent
+                // recover/borrow can only over-count -- costing one extra
+                // cheap rewound walk, never a missed candidate.
+                if (retired && OrphanScanner.isCandidateOrphan(slotPath)) {
+                    recoveryDeferredThisCycle++;
+                }
                 recoveryInRangeNext++;
                 continue;
             }
-            String slotPath = sfDir + "/" + slotBaseId + "-" + i;
             if (!OrphanScanner.isCandidateOrphan(slotPath)) {
                 // No stranded data: release the reservation and keep scanning;
                 // an empty slot must not cost a whole step.
@@ -868,8 +899,8 @@ private boolean recoverOneSlotStep(long stepBudgetMillis) {
         }
 
         if (recoveryDeferredThisCycle > 0) {
-            // At least one slot was parked this cycle (contended or
-            // persistently failing). Its data stays durable on disk, so
+            // At least one slot was parked this cycle (contended, persistently
+            // failing, or retired with data still on disk). Its data stays durable on disk, so
             // instead of latching recoveryComplete -- which would abandon it
             // until a restart or a lucky borrow of that index -- rewind the
             // scan and let the driver's retry cadence run another cycle.
@@ -2073,6 +2104,24 @@ private void addRetiredSlot(SenderSlot s) {
         retiredSlots.add(s);
     }
 
+    /**
+     * Whether {@code idx} is currently held by a RETIRED slot (see
+     * {@link #reclaimSlot} and the in-range recovery retire branch) rather
+     * than a live one. Lets the recovery scan tell "reserved because
+     * borrowed/prewarmed" apart from "reserved because a wedged close() left
+     * the flock held" for its deferral accounting: only the latter's dir can
+     * hold stranded data no borrow is coming for. Caller must hold
+     * {@code lock}; retiredSlots is bounded by maxSize, so the walk is cheap.
+     */
+    private boolean isRetiredSlotIndex(int idx) {
+        for (int i = 0, n = retiredSlots.size(); i < n; i++) {
+            if (retiredSlots.get(i).slotIndex() == idx) {
+                return true;
+            }
+        }
+        return false;
+    }
+
     private void recoverRetiredSlotAt(int retiredIndex) {
         SenderSlot s = retiredSlots.get(retiredIndex);
         int last = retiredSlots.size() - 1;

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

@@ -1753,6 +1753,124 @@ public void testStartupRetiredSlotRecoveredAfterLateFlockRelease() throws Except
         });
     }
 
+    @Test
+    public void testRecoveryScanStaysAliveWhileRetiredSlotHoldsStrandedData() throws Exception {
+        // Drain-liveness regression: a slot RETIRED mid-scan (recoverer close()
+        // returned with the flock still held) keeps slotInUse[i] set, so the
+        // next walk over its index took the "reserved => live" skip WITHOUT
+        // counting a deferral and latched recoveryComplete with the slot's
+        // unacked data still on disk. From there the data was stranded until a
+        // restart or a lucky borrow of that exact index -- which steady low
+        // load may never produce (testStartupRetiredSlotRecoveredAfterLate-
+        // FlockRelease passes only because its borrow IS that lucky borrow).
+        // The scan must instead treat a retired index whose dir is still a
+        // candidate orphan as a deferral -- the same rule as a CONTENDED park:
+        // the cycle keeps rewinding, and once the late flock release restores
+        // the index the SCAN itself drains the data, no borrow required. This
+        // test is RED (latches complete) until that deferral is counted.
+        TestUtils.assertMemoryLeak(() -> {
+            // Phase 1: strand unacked data under default-0.
+            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: ack-ing server; the FIRST recovery build of slot 0 is
+            // forged into the wedged shape (closed=true makes the recovery
+            // drain throw and close() a no-op, so the real flock stays held
+            // and slotLockReleased stays false) so the scan's own drain FAILs
+            // and the slot is retired. deferStartupRecovery=true keeps the
+            // scan entirely under this test's control -- no background driver.
+            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 {
+                            ((QwpWebSocketSender) real).setClosedForTesting(true);
+                        } catch (Exception e) {
+                            throw new RuntimeException(e);
+                        }
+                    }
+                    return real;
+                };
+
+                try (SenderPool pool = new SenderPool(cfg2, 0, 1, 500,
+                        Long.MAX_VALUE, Long.MAX_VALUE, factory, true)) {
+                    // Step 1: builds the forged recoverer, fails its drain, and
+                    // retires the slot (leakedSlots=1, slotInUse[0] stays set).
+                    // The FAILED outcome stops the drive without advancing.
+                    long deadline = System.currentTimeMillis() + 10_000;
+                    while (pool.leakedSlotCount() == 0
+                            && System.currentTimeMillis() < deadline) {
+                        pool.runStartupRecoveryStepForTesting();
+                    }
+                    Assert.assertNotNull("recovery must have built slot 0", forged.get());
+                    Assert.assertEquals("scan must retire the wedged slot",
+                            1, pool.leakedSlotCount());
+
+                    // Re-drive the scan: the retired index is skipped as
+                    // reserved. Pre-fix the first full walk latched
+                    // recoveryComplete; post-fix the still-candidate retired
+                    // dir counts as a deferral and the cycle keeps rewinding.
+                    for (int k = 0; k < 8; k++) {
+                        pool.runStartupRecoveryStepForTesting();
+                    }
+                    Assert.assertTrue("retired dir must still hold the stranded data",
+                            hasSegmentFile(slot("default-0")));
+                    Assert.assertFalse(
+                            "scan must NOT latch recoveryComplete while a retired slot still "
+                                    + "holds stranded durable data -- that abandons the data "
+                                    + "until a restart or a lucky borrow of that exact index",
+                            pool.isRecoveryCompleteForTesting());
+
+                    // The late flock release: the "wedged worker" finishes
+                    // (un-forge + real close). The housekeeper tick re-probes
+                    // retiredSlots and frees the index; the still-alive scan
+                    // must then reserve it and drain the stranded data ITSELF
+                    // -- no borrow anywhere in this phase.
+                    Sender recoverer = forged.get();
+                    ((QwpWebSocketSender) recoverer).setClosedForTesting(false);
+                    recoverer.close();
+                    pool.reapIdle();
+                    Assert.assertEquals("late release must restore the retired capacity",
+                            0, pool.leakedSlotCount());
+
+                    deadline = System.currentTimeMillis() + 10_000;
+                    while (!pool.isRecoveryCompleteForTesting()
+                            && System.currentTimeMillis() < deadline) {
+                        pool.runStartupRecoveryStepForTesting();
+                    }
+                    Assert.assertTrue("scan must complete once the stranded data is drained",
+                            pool.isRecoveryCompleteForTesting());
+                    Assert.assertTrue("the scan itself must have drained the stranded data",
+                            awaitNoSegmentFile(slot("default-0"), 15_000));
+                    Assert.assertTrue("replayed frames must reach the server",
+                            awaitAtLeast(handler.frames, 1, 15_000));
+                }
+            }
+        });
+    }
+
     @Test
     public void testZeroTimeoutBorrowProbesRetiredSlotBeforeThrowing() throws Exception {
         // Boundary twin of testStartupRetiredSlotRecoveredAfterLateFlockRelease: