fix(qwp): probe the slot flock O(1) before paying a recovery build on parked slots

A managed slot whose flock is held by another live owner is parked as
CONTENDED and re-probed on every startup-recovery retry cycle -- every
second on the direct pool's private driver, for as long as the owner
lives. Each re-probe paid a full recovery build just to reach
SlotLock.acquire and throw: two isCandidateOrphan dir enumerations, a
complete config re-parse plus builder graph, parent-dir fsync barriers
in periodic durability mode, and an owned SegmentManager allocation
torn straight back down.

Add SlotLock.probeHolder: a side-effect-light non-blocking flock probe
that never creates dirs or files, never throws, and routes its
momentary hold through the standard close() so an unconfirmed unlock
can never leak from a probe. The recovery scan asks the probe first and
parks on a held flock for a few syscalls; an indeterminate probe falls
through to the full build, which keeps sole ownership of error
classification. Races are benign both ways: a free probe can still
lose the acquire (parks exactly as before) and a stale held probe is
re-observed next cycle.

New SenderPoolSfTest proves the arc with a real externally held flock:
three parked cycles cost zero builds, and the first cycle after release
recovers the slot with exactly one build.

Author: bluestreak01

core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/SlotLock.java

@@ -135,6 +135,57 @@ public static SlotLock acquire(String slotDir, boolean syncParentDirectory) {
         }
     }
 
+    /**
+     * Side-effect-light contention probe: reports the current holder of the
+     * slot flock without creating the slot dir or lock file and without
+     * paying a full engine build. Opens the existing {@code .lock} file
+     * (absent means nothing can hold a flock on it), try-locks it
+     * non-blocking, and releases immediately on success.
+     * <p>
+     * Returns a non-null holder description (the {@code .lock.pid} payload,
+     * or {@code "unknown"}) when the flock is currently held by a live
+     * owner; {@code null} when the lock is free or the probe could not
+     * determine state (missing lock file, open failure). Callers must treat
+     * {@code null} as "proceed to a full acquire", which owns real error
+     * classification -- the probe never throws.
+     * <p>
+     * Races are benign in both directions: a free probe can still lose the
+     * subsequent acquire to a concurrent owner (the caller handles that
+     * contention exactly as before), and a held probe that goes stale the
+     * moment the owner exits is simply re-observed on the caller's next
+     * cycle. The probe's momentary hold can make a concurrent acquirer see
+     * spurious contention -- the same class of race two real contenders
+     * already have.
+     */
+    public static String probeHolder(String slotDir) {
+        if (slotDir == null || slotDir.isEmpty()) {
+            return null;
+        }
+        // Same pre-step as acquire(): a lock retained by THIS process after
+        // an unconfirmed unlock would otherwise read as a live holder for as
+        // long as the retry list carries it.
+        retryPendingReleases();
+        String lockPath = slotDir + "/" + LOCK_FILE_NAME;
+        if (!Files.exists(lockPath)) {
+            return null;
+        }
+        int fd = Files.openRW(lockPath);
+        if (fd < 0) {
+            return null;
+        }
+        if (Files.lock(fd) != 0) {
+            String holder = readHolder(slotDir + "/" + LOCK_PID_FILE_NAME);
+            Files.close(fd);
+            return holder;
+        }
+        // The flock was free and is momentarily ours. Route the release
+        // through the standard close() so an unconfirmed unlock is retained
+        // on the retry list exactly like a normal owner's -- a probe must
+        // never leak a held flock.
+        new SlotLock(slotDir, fd).close();
+        return null;
+    }
+
     /**
      * Replaces the live descriptor with a known-dead value until the returned
      * guard closes. Test-only: exercises release retry paths without exposing

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

@@ -31,6 +31,7 @@
 import io.questdb.client.cutlass.qwp.client.QwpWebSocketSender;
 import io.questdb.client.cutlass.qwp.client.sf.cursor.BackgroundDrainerListener;
 import io.questdb.client.cutlass.qwp.client.sf.cursor.OrphanScanner;
+import io.questdb.client.cutlass.qwp.client.sf.cursor.SlotLock;
 import io.questdb.client.cutlass.qwp.client.sf.cursor.SlotLockContentionException;
 import io.questdb.client.std.Files;
 import io.questdb.client.std.IntList;
@@ -934,6 +935,29 @@ private RecoveryDrainOutcome drainCandidateSlotForRecovery(int slotIndex, String
             if (!OrphanScanner.isCandidateOrphan(slotPath)) {
                 return RecoveryDrainOutcome.DRAINED;
             }
+            // O(1) contention pre-probe (flock only). A slot parked as
+            // CONTENDED is re-probed on every retry cycle for as long as its
+            // live owner runs -- potentially that owner's whole lifetime --
+            // and the full recovery build below (config re-parse, builder
+            // graph, parent-dir fsync barriers in periodic durability, owned
+            // SegmentManager allocation) would exist only to reach
+            // SlotLock.acquire and throw. Ask the flock directly first so
+            // the steady-state cost of a held slot is a few syscalls per
+            // cycle, not a build. Races are benign in both directions: a
+            // free probe can still lose the acquire inside the build (the
+            // contention catch below parks exactly as before), and a held
+            // probe that goes stale is re-observed on the next cycle. An
+            // indeterminate probe (null) falls through to the build, which
+            // owns real error classification.
+            String probedHolder = SlotLock.probeHolder(slotPath);
+            if (probedHolder != null) {
+                if (warnSlotOnce(slotIndex)) {
+                    LOG.warn("startup SF recovery: slot {} is held by another live owner (holder={}); "
+                            + "parking it and continuing with the remaining slots",
+                            slotPath, probedHolder);
+                }
+                return RecoveryDrainOutcome.CONTENDED;
+            }
             try {
                 // Recovery delegate: forced OFF-mode initial connect (see
                 // createRecoverer / defaultRecoverySender), so this build does

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

@@ -36,6 +36,8 @@
 import io.questdb.client.cutlass.qwp.client.sf.cursor.SegmentManager;
 import io.questdb.client.cutlass.qwp.client.sf.cursor.SlotLock;
 import io.questdb.client.cutlass.qwp.client.sf.cursor.SlotLockContentionException;
+import io.questdb.client.std.MemoryTag;
+import io.questdb.client.std.Unsafe;
 import io.questdb.client.impl.PooledSender;
 import io.questdb.client.impl.SenderPool;
 import io.questdb.client.std.Files;
@@ -3840,6 +3842,86 @@ private static boolean awaitAtLeast(AtomicInteger counter, int target, long time
         return counter.get() >= target;
     }
 
+    @Test
+    public void testContendedSlotReprobeUsesFlockProbeNotFullBuild() throws Exception {
+        // A slot whose flock is held by another LIVE owner is parked and
+        // re-probed on every retry cycle -- potentially for the owner's whole
+        // lifetime. The re-probe must ask the flock directly (O(1) probe,
+        // a few syscalls), not pay a full recovery build (config re-parse,
+        // builder graph, parent-dir fsync barriers in periodic durability,
+        // owned SegmentManager allocation) per cycle just to reach
+        // SlotLock.acquire and throw.
+        TestUtils.assertMemoryLeak(() -> {
+            String config = "ws::addr=localhost:1;sf_dir=" + sfDir + ";";
+            String slot0 = slot("default-0");
+
+            // Seed one unacked frame so slot 0 is a candidate orphan. The
+            // group root normally comes from Sender.build(); this test seeds
+            // the slot directly, so create the parent first (mkdir in
+            // SlotLock.acquire is non-recursive).
+            Assert.assertEquals(0, Files.mkdir(sfDir, Files.DIR_MODE_DEFAULT));
+            long buf = Unsafe.malloc(16, MemoryTag.NATIVE_DEFAULT);
+            try (CursorSendEngine seed = new CursorSendEngine(slot0, 1 << 20)) {
+                Unsafe.getUnsafe().setMemory(buf, 16, (byte) 1);
+                Assert.assertEquals(0L, seed.appendBlocking(buf, 16));
+            } finally {
+                Unsafe.free(buf, 16, MemoryTag.NATIVE_DEFAULT);
+            }
+            Assert.assertTrue("seeded slot must be a candidate orphan",
+                    OrphanScanner.isCandidateOrphan(slot0));
+
+            AtomicInteger builds = new AtomicInteger();
+            IntFunction<Sender> factory = idx -> {
+                builds.incrementAndGet();
+                // Fidelity with the production build: the recovery build's
+                // fate is decided by the real slot flock, exactly like
+                // defaultRecoverySender's engine construction.
+                SlotLock lock = SlotLock.acquire(slot("default-" + idx));
+                return (Sender) Proxy.newProxyInstance(
+                        Sender.class.getClassLoader(),
+                        new Class[]{Sender.class},
+                        (proxy, method, args) -> {
+                            if ("drain".equals(method.getName())) {
+                                return Boolean.TRUE;
+                            }
+                            if ("close".equals(method.getName())) {
+                                lock.close();
+                                return null;
+                            }
+                            throw new AssertionError(
+                                    "unexpected recovery sender call: " + method.getName());
+                        });
+            };
+
+            SlotLock held = SlotLock.acquire(slot0);
+            try (SenderPool pool = newDeferredPoolWithFactory(config, 0, 1, 5_000, factory)) {
+                // Steady-state re-probe of a parked contended slot: three
+                // full cycles while the flock is held.
+                for (int cycle = 0; cycle < 3; cycle++) {
+                    Assert.assertFalse("a cycle with only a contended slot must defer",
+                            pool.runStartupRecoveryStepForTesting());
+                }
+                Assert.assertEquals("re-probing a contended slot must be a flock probe, "
+                        + "not a full recovery build per cycle", 0, builds.get());
+                Assert.assertTrue("parked slot must keep its durable data",
+                        hasSegmentFile(slot0));
+
+                // The probe must not dampen recovery: once the owner lets
+                // go, the very next cycle pays exactly one real build and
+                // drains the slot.
+                held.close();
+                Assert.assertTrue("released slot must be recovered on the next cycle",
+                        pool.runStartupRecoveryStepForTesting());
+                Assert.assertEquals("released slot must be recovered with exactly one build",
+                        1, builds.get());
+                Assert.assertFalse("final scan step must mark recovery complete",
+                        pool.runStartupRecoveryStepForTesting());
+            } finally {
+                held.close(); // idempotent when already released
+            }
+        });
+    }
+
     private static boolean awaitNoSegmentFile(String slotPath, long timeoutMillis)
             throws InterruptedException {
         long deadline = System.currentTimeMillis() + timeoutMillis;