Reclaim logical slot locks on fully-drained close

The store-and-forward logical slot lock lives in the shared
<sf_dir>/.slot-locks directory, outside the slot dir so it survives
a slot rename. Nothing reclaimed it: unlike the slot's .ack-watermark
and .symbol-dict side-files, which CursorSendEngine.close() unlinks
via removeOrphan on a fully-drained close, the logical lock and its
.lock.pid sidecar were left behind. Under rotating senderIds sharing
one sf_dir, .slot-locks accumulated one dead lock+pid pair per
distinct slot name for the lifetime of sf_dir.

SlotLock now exposes removeOrphanLogical, mirroring
AckWatermark.removeOrphan and PersistedSymbolDict.removeOrphan, and
CursorSendEngine's fully-drained cleanup calls it alongside the
sibling side-file removals. acquireLogical and removeOrphanLogical
share a resolveLogicalLock helper so the two can never target
different files. The unlink is best-effort: the retiring engine still
holds the directory-local lock, the real multi-writer guard, and a
fully-drained retirement performs no rename, so the logical lock is
not in use; a drainer momentarily mid-acquireLogical fails its
immediate candidacy and directory-lock check and backs off.

SlotLockTest covers the removal and its silent no-op on absent or
invalid input; EngineCloseSlotLockReleaseTest adds an end-to-end test
that a fully-drained engine close reclaims a build's orphaned logical
lock, verified to fail without the CursorSendEngine change.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: glasstiger

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

@@ -728,6 +728,17 @@ public synchronized void close() {
                     PersistedSymbolDict.removeOrphan(sfDir);
                 } catch (Throwable ignored) {
                 }
+                try {
+                    // The logical slot lock lives OUTSIDE the slot dir (in the
+                    // shared .slot-locks dir) so it survives a slot rename; the
+                    // fully-drained retirement that removes this slot's other
+                    // side-files must remove it too, or .slot-locks accumulates a
+                    // dead lock+pid pair per distinct slot name for the lifetime of
+                    // sf_dir. This engine still holds the directory-local lock, so
+                    // the best-effort unlink is safe (see SlotLock.removeOrphanLogical).
+                    SlotLock.removeOrphanLogical(sfDir);
+                } catch (Throwable ignored) {
+                }
             }
         } finally {
             if (slotLock != null) {

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

@@ -113,20 +113,51 @@ public static SlotLock acquireLogical(String slotDir) {
     @TestOnly
     public static SlotLock acquireLogical(FilesFacade ff, String slotDir) {
         validateSlotDir(slotDir);
-        Path slotPath = Paths.get(slotDir);
-        Path parentPath = slotPath.getParent();
-        Path slotNamePath = slotPath.getFileName();
-        if (parentPath == null || slotNamePath == null || slotNamePath.toString().isEmpty()) {
+        String[] paths = resolveLogicalLock(slotDir);
+        if (paths == null) {
             throw new IllegalArgumentException(
                     "slotDir must contain a parent and slot name: " + slotDir);
         }
-        String parentDir = parentPath.toString();
-        String slotName = slotNamePath.toString();
-        String logicalLockDir = parentDir + "/" + LOGICAL_LOCK_DIR_NAME;
-        ensureDirectory(ff, logicalLockDir, "logical slot lock dir");
-        String lockPath = logicalLockDir + "/" + slotName + ".lock";
-        String pidPath = logicalLockDir + "/" + slotName + ".lock.pid";
-        return acquireAt(ff, slotDir, lockPath, pidPath);
+        ensureDirectory(ff, paths[0], "logical slot lock dir");
+        return acquireAt(ff, slotDir, paths[1], paths[2]);
+    }
+
+    /**
+     * Best-effort removal of the parent-anchored logical lock files
+     * ({@code <parent>/.slot-locks/<slotName>.lock} and its {@code .lock.pid}
+     * sidecar) for {@code slotDir}, mirroring {@link AckWatermark#removeOrphan}
+     * and {@link PersistedSymbolDict#removeOrphan} for the slot's in-directory
+     * side-files. The fully-drained close that permanently retires a slot calls
+     * this: the logical lock lives OUTSIDE the slot dir (so it survives a slot
+     * rename), so nothing else reclaims it, and without this
+     * {@code <sf_dir>/.slot-locks} accumulates one dead lock+pid pair per
+     * distinct slot name for the lifetime of {@code sf_dir} -- unbounded under
+     * rotating {@code senderId}s.
+     * <p>
+     * The unlink is best-effort and safe: the retiring engine still holds the
+     * slot's directory-local {@link #acquire} lock (the real multi-writer guard)
+     * across this cleanup, and fully-drained retirement performs no rename, so
+     * the logical lock -- which only guards the close/rename/recreate transition
+     * -- is not in use. An orphan drainer momentarily mid-{@link #acquireLogical}
+     * fails its immediately-following candidacy / directory-lock check (the slot
+     * is gone) and backs off. Unlike {@link #acquireLogical}, an unusable
+     * {@code slotDir} is a silent no-op here rather than a throw.
+     */
+    public static void removeOrphanLogical(String slotDir) {
+        removeOrphanLogical(FilesFacade.INSTANCE, slotDir);
+    }
+
+    /** Facade-aware variant of {@link #removeOrphanLogical(String)}. */
+    public static void removeOrphanLogical(FilesFacade ff, String slotDir) {
+        if (slotDir == null || slotDir.isEmpty()) {
+            return;
+        }
+        String[] paths = resolveLogicalLock(slotDir);
+        if (paths == null) {
+            return;
+        }
+        ff.remove(paths[1]);
+        ff.remove(paths[2]);
     }
 
     private static SlotLock acquireAt(FilesFacade ff, String slotDir, String lockPath, String pidPath) {
@@ -167,6 +198,30 @@ private static void ensureDirectory(FilesFacade ff, String path, String descript
         }
     }
 
+    /**
+     * Resolves the parent-anchored logical lock layout for {@code slotDir}:
+     * {@code [0]} the {@code .slot-locks} directory, {@code [1]} the
+     * {@code <slotName>.lock} path, {@code [2]} the {@code .lock.pid} path.
+     * Returns {@code null} when {@code slotDir} has no usable parent or name.
+     * Shared by {@link #acquireLogical} and {@link #removeOrphanLogical} so the
+     * two can never target different files.
+     */
+    private static String[] resolveLogicalLock(String slotDir) {
+        Path slotPath = Paths.get(slotDir);
+        Path parentPath = slotPath.getParent();
+        Path slotNamePath = slotPath.getFileName();
+        if (parentPath == null || slotNamePath == null || slotNamePath.toString().isEmpty()) {
+            return null;
+        }
+        String logicalLockDir = parentPath + "/" + LOGICAL_LOCK_DIR_NAME;
+        String slotName = slotNamePath.toString();
+        return new String[]{
+                logicalLockDir,
+                logicalLockDir + "/" + slotName + LOCK_FILE_NAME,
+                logicalLockDir + "/" + slotName + LOCK_PID_FILE_NAME
+        };
+    }
+
     private static void validateSlotDir(String slotDir) {
         if (slotDir == null || slotDir.isEmpty()) {
             throw new IllegalArgumentException("slotDir must not be empty");

core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/EngineCloseSlotLockReleaseTest.java

@@ -38,6 +38,8 @@
 import java.nio.file.Paths;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
 /**
@@ -84,22 +86,34 @@ public void setUp() {
     @After
     public void tearDown() {
         if (sfDir == null) return;
-        long find = Files.findFirst(sfDir);
+        rmDirRecursive(sfDir);
+    }
+
+    private static void rmDirRecursive(String dir) {
+        if (!Files.exists(dir)) return;
+        long find = Files.findFirst(dir);
         if (find > 0) {
             try {
                 int rc = 1;
                 while (rc > 0) {
                     String name = Files.utf8ToString(Files.findName(find));
                     if (name != null && !".".equals(name) && !"..".equals(name)) {
-                        Files.remove(sfDir + "/" + name);
+                        String child = dir + "/" + name;
+                        long probe = Files.findFirst(child);
+                        if (probe > 0) {
+                            Files.findClose(probe);
+                            rmDirRecursive(child);
+                        } else {
+                            Files.remove(child);
+                        }
                     }
                     rc = Files.findNext(find);
                 }
             } finally {
                 Files.findClose(find);
             }
         }
-        Files.remove(sfDir);
+        Files.remove(dir);
     }
 
     @Test(timeout = 10_000L)
@@ -176,4 +190,33 @@ public void testSlotLockReleasedEvenIfRingCloseThrows() throws Exception {
             }
         });
     }
+
+    @Test(timeout = 10_000L)
+    public void testFullyDrainedCloseRemovesLogicalSlotLock() throws Exception {
+        TestUtils.assertMemoryLeak(() -> {
+            // The engine's slot lives one level under sfDir, so the parent-anchored
+            // logical lock lands at <sfDir>/.slot-locks/slot0.lock -- inside the tree
+            // tearDown cleans.
+            String slotDir = sfDir + "/slot0";
+
+            // Simulate the build's transient logical lock: acquire and release it,
+            // leaving the shared-dir lock file behind exactly as a real build does
+            // (close() releases the flock but keeps the file so it survives a rename).
+            SlotLock.acquireLogical(slotDir).close();
+            String lockFile = sfDir + "/.slot-locks/slot0.lock";
+            assertTrue("precondition: a build's logical lock leaves a file behind",
+                    Files.exists(lockFile));
+
+            // A fresh engine that never publishes is fully drained on close, so its
+            // close() runs the retirement cleanup that must reclaim the logical lock.
+            try (CursorSendEngine engine = new CursorSendEngine(slotDir, 4L * 1024 * 1024)) {
+                assertEquals(slotDir, engine.sfDir());
+            }
+
+            assertFalse("fully-drained engine close must reclaim the orphaned logical "
+                            + "slot lock; otherwise .slot-locks grows unbounded under "
+                            + "rotating senderIds",
+                    Files.exists(lockFile));
+        });
+    }
 }

core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/SlotLockTest.java

@@ -167,6 +167,40 @@ public void testLogicalLockReportsLockDirectoryCreationFailure() throws Exceptio
         });
     }
 
+    @Test
+    public void testRemoveOrphanLogicalDeletesLockAndPidFiles() throws Exception {
+        TestUtils.assertMemoryLeak(() -> {
+            String slot = parentDir + "/alpha";
+            assertEquals(0, Files.mkdir(slot, Files.DIR_MODE_DEFAULT));
+            // acquireLogical anchors the lock under the shared parent .slot-locks dir.
+            String lockFile = parentDir + "/.slot-locks/alpha.lock";
+            String pidFile = parentDir + "/.slot-locks/alpha.lock.pid";
+            try (SlotLock ignored = SlotLock.acquireLogical(slot)) {
+                assertTrue("logical .lock created", Files.exists(lockFile));
+                assertTrue("logical .lock.pid created", Files.exists(pidFile));
+            }
+            // close() releases the flock but deliberately keeps the file (it must
+            // outlast a slot rename); only the fully-drained retirement reclaims it.
+            assertTrue("logical .lock survives close", Files.exists(lockFile));
+            SlotLock.removeOrphanLogical(slot);
+            assertFalse("logical .lock removed on retirement", Files.exists(lockFile));
+            assertFalse("logical .lock.pid removed on retirement", Files.exists(pidFile));
+        });
+    }
+
+    @Test
+    public void testRemoveOrphanLogicalIsSilentNoOpWhenAbsentOrInvalid() throws Exception {
+        TestUtils.assertMemoryLeak(() -> {
+            // Never-locked slot: nothing to remove, must not throw.
+            SlotLock.removeOrphanLogical(parentDir + "/never-locked");
+            // Unlike acquireLogical (which throws on these), the retirement cleanup
+            // is best-effort and tolerates unusable input silently.
+            SlotLock.removeOrphanLogical(null);
+            SlotLock.removeOrphanLogical("");
+            SlotLock.removeOrphanLogical("slot"); // no parent component
+        });
+    }
+
     @Test
     public void testTwoDifferentSlotsCoexist() throws Exception {
         TestUtils.assertMemoryLeak(() -> {