Share the slot-lock dir and stop false quarantines

Two defects in the parent-anchored logical slot lock, which this branch
introduced: before it, build() took no SlotLock at all and SlotLock
created only the per-process slot directory.

The lock now lives under a directory every sender beneath one sf_dir
must create a file in, and ensureDirectory created it 0755. Whichever
process starts first owns it, so a sender running as a different uid
cannot create its lock file and acquireAt throws "could not open slot
lock file". That is an IllegalStateException, not a LineSenderException,
so a caller following the documented contract does not catch it and
build() fails outright. Create that one directory 0777 instead and let
the deployment's umask decide the sharing policy, exactly as it already
does for the sf_dir the lock lives under: the usual 022 leaves it at the
same 0755 as before, while a shared-group deployment gets the
group-writable directory it needs. Slot directories keep the restrictive
mode -- one process creates each and only that process writes inside it.

The drainer then turned that failure into a permanent one. isLockContention
matches only "already in use", so every other acquireLogical failure --
the permission case above, or a momentary fd exhaustion -- rethrows into
the outer catch, which drops a .failed sentinel. That runs BEFORE
CursorSendEngine takes the slot's own .lock, so the drainer quarantines a
healthy slot another live process still owns. Nothing ever removes the
sentinel and isCandidateOrphan treats it as disqualifying, so when the
real owner later died its unacked data would stay stranded until an
operator intervened. Write the sentinel only when engine is non-null,
which is set once the engine holds the directory-local lock and so marks
genuine adoption; a pre-adoption failure now leaves the slot for the next
scan. The auth/upgrade and wire sentinels are unchanged -- both fire
after adoption and are sanctioned orphan terminals.

Both tests fail without their fix: the mode assertion reports 493 where
511 is required, and the drainer test finds the sentinel in a slot it
never locked. The drainer test blocks the lock by planting .slot-locks as
a regular file rather than by chmod, so it reproduces the failure on any
platform and does not silently pass when CI runs as root.

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/BackgroundDrainer.java

@@ -728,10 +728,27 @@ public void run() {
                 LOG.debug("drainer setup failed for slot {}: {}", slotPath, msg, t);
             }
             lastErrorMessage = msg;
-            try {
-                OrphanScanner.markFailed(slotPath, "setup: " + msg);
-            } catch (Throwable ignored) {
-                // best-effort
+            // Quarantine ONLY a slot this drainer actually adopted. engine is
+            // assigned after CursorSendEngine has taken the slot's own .lock, so
+            // a null engine here means the failure happened before adoption --
+            // while merely probing a slot another live process still owns.
+            //
+            // The failures that reach this point pre-adoption are LOCAL and
+            // usually transient: acquireLogical rethrows anything that is not
+            // lock contention, so a permission problem on the shared
+            // .slot-locks directory or a momentary fd exhaustion lands here.
+            // Writing the sentinel then would exclude a healthy, in-use slot
+            // from orphan drain permanently (OrphanScanner.isCandidateOrphan
+            // treats .failed as disqualifying, and nothing ever removes it), so
+            // when its real owner later dies its unacked data would be stranded
+            // until an operator intervened. Leave the slot alone; the next scan
+            // retries it.
+            if (engine != null) {
+                try {
+                    OrphanScanner.markFailed(slotPath, "setup: " + msg);
+                } catch (Throwable ignored) {
+                    // best-effort
+                }
             }
             outcome = DrainOutcome.FAILED;
         } finally {

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

@@ -87,7 +87,9 @@ private SlotLock(FilesFacade ff, String slotDir, int fd) {
      */
     public static SlotLock acquire(String slotDir) {
         validateSlotDir(slotDir);
-        ensureDirectory(FilesFacade.INSTANCE, slotDir, "slot dir");
+        // DIR_MODE_DEFAULT is right here: one process creates its own slot
+        // directory and only that process writes inside it.
+        ensureDirectory(FilesFacade.INSTANCE, slotDir, "slot dir", Files.DIR_MODE_DEFAULT);
         String lockPath = slotDir + "/" + LOCK_FILE_NAME;
         String pidPath = slotDir + "/" + LOCK_PID_FILE_NAME;
         return acquireAt(FilesFacade.INSTANCE, slotDir, lockPath, pidPath);
@@ -118,7 +120,17 @@ public static SlotLock acquireLogical(FilesFacade ff, String slotDir) {
             throw new IllegalArgumentException(
                     "slotDir must contain a parent and slot name: " + slotDir);
         }
-        ensureDirectory(ff, paths[0], "logical slot lock dir");
+        // SHARED_DIR_MODE, not DIR_MODE_DEFAULT: unlike a slot directory -- which
+        // one process creates and only that process writes -- this directory is
+        // shared by every sender under the same sf_dir, and each of them must
+        // CREATE its own lock file inside it. At 0755 the first process to start
+        // owns it and a sender running as a different uid cannot create its lock
+        // file at all, so build() throws before it can open the slot. Passing
+        // 0777 hands the sharing policy to the deployment's umask, which is what
+        // already governs sf_dir itself: an unset umask leaves this identical to
+        // the old 0755, while a shared-group deployment (umask 002) gets the
+        // group-writable directory it needs.
+        ensureDirectory(ff, paths[0], "logical slot lock dir", Files.DIR_MODE_SHARED);
         return acquireAt(ff, slotDir, paths[1], paths[2]);
     }
 
@@ -185,9 +197,9 @@ private static SlotLock acquireAt(FilesFacade ff, String slotDir, String lockPat
         }
     }
 
-    private static void ensureDirectory(FilesFacade ff, String path, String description) {
+    private static void ensureDirectory(FilesFacade ff, String path, String description, int mode) {
         if (!ff.exists(path)) {
-            int rc = ff.mkdir(path, Files.DIR_MODE_DEFAULT);
+            int rc = ff.mkdir(path, mode);
             // Multiple senders may create the shared parent lock directory
             // concurrently. Treat EEXIST as success, just as the builder does
             // for the SF root itself.

core/src/main/java/io/questdb/client/std/Files.java

@@ -75,6 +75,17 @@ public final class Files {
      * others may traverse and read but not modify. 0755 octal
      */
     public static final int DIR_MODE_DEFAULT = 493;
+    /**
+     * Creation mode for a directory that processes running as DIFFERENT users
+     * must all create files in -- the store-and-forward logical slot-lock
+     * directory being the one such case. 0777 octal, so the deployment's umask
+     * decides the actual sharing policy exactly as it does for the sf_dir these
+     * live under: the usual 022 yields the same 0755 as
+     * {@link #DIR_MODE_DEFAULT}, while a shared-group deployment (umask 002)
+     * gets the group-writable directory a second uid needs. Do NOT use this for
+     * a directory only its creator writes to.
+     */
+    public static final int DIR_MODE_SHARED = 511;
 
     /** {@code dirent.d_type} sentinel: type unknown (filesystem doesn't fill it). */
     public static final int DT_UNKNOWN = 0;

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

@@ -39,6 +39,7 @@
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
 
 public class BackgroundDrainerOrphanTailTest {
 
@@ -85,6 +86,59 @@ public void testDeferredOnlyRecoveredTailRetiresWithoutConnecting() throws Excep
         });
     }
 
+    @Test
+    public void testPreAdoptionSetupFailureDoesNotQuarantineTheSlot() throws Exception {
+        // A drainer that cannot even take the logical lock has NOT adopted the
+        // slot -- another live process may still own it. The failures that reach
+        // that point are local and usually transient (acquireLogical rethrows
+        // anything that is not lock contention, so a permission problem on the
+        // shared .slot-locks directory or a momentary fd exhaustion lands
+        // there). Writing the .failed sentinel then would exclude a healthy,
+        // in-use slot from orphan drain permanently -- nothing ever removes it --
+        // so when its real owner later died its unacked data would be stranded
+        // until an operator intervened.
+        //
+        // Blocks the lock by planting .slot-locks as a regular FILE: acquireLogical
+        // sees it exists, skips the mkdir, and then cannot open a lock file
+        // beneath it. That reproduces the pre-adoption failure on every platform
+        // and without depending on the test user's privileges -- a chmod-based
+        // version would silently pass when CI runs as root.
+        String slotPath = temporaryFolder.newFolder("live-slot").getAbsolutePath();
+        TestUtils.assertMemoryLeak(() -> {
+            try (CursorSendEngine engine = new CursorSendEngine(slotPath, SEGMENT_SIZE_BYTES)) {
+                appendDeferredFrame(engine);
+            }
+            String parent = slotPath.substring(0, slotPath.lastIndexOf('/'));
+            String lockDirPath = parent + "/.slot-locks";
+            int fd = Files.openRW(lockDirPath);
+            assertTrue("could not plant the blocking file", fd > -1);
+            Files.close(fd);
+
+            AtomicInteger connectAttempts = new AtomicInteger();
+            BackgroundDrainer drainer = new BackgroundDrainer(
+                    slotPath,
+                    SEGMENT_SIZE_BYTES,
+                    1L << 20,
+                    () -> {
+                        connectAttempts.incrementAndGet();
+                        throw new AssertionError("setup must fail before any connect");
+                    },
+                    5_000L,
+                    1L,
+                    5L,
+                    true,
+                    200L);
+
+            drainer.run();
+
+            assertEquals("setup must fail before the reconnect factory is used",
+                    0, connectAttempts.get());
+            assertEquals(BackgroundDrainer.DrainOutcome.FAILED, drainer.outcome());
+            assertFalse("a slot this drainer never adopted must not be quarantined",
+                    Files.exists(slotPath + "/" + OrphanScanner.FAILED_SENTINEL_NAME));
+        });
+    }
+
     private static void appendDeferredFrame(CursorSendEngine engine) {
         long buffer = Unsafe.malloc(16, MemoryTag.NATIVE_DEFAULT);
         try {

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

@@ -141,6 +141,32 @@ public void testLogicalLockRejectsInvalidPaths() throws Exception {
         });
     }
 
+    @Test
+    public void testSharedLockDirectoryIsCreatedUmaskGoverned() throws Exception {
+        // .slot-locks is the one directory every sender under an sf_dir must
+        // CREATE a file in, so unlike a slot directory it cannot be created
+        // 0755: the first process to start would own it and a sender running as
+        // a different uid could not create its lock file, failing build(). Pass
+        // 0777 and let the deployment's umask decide, exactly as it already does
+        // for the sf_dir these live under.
+        TestUtils.assertMemoryLeak(() -> {
+            String slot = parentDir + "/mode-check";
+            String lockDir = parentDir + "/.slot-locks";
+            RecordingMkdirFacade ff = new RecordingMkdirFacade();
+            try (SlotLock ignored = SlotLock.acquireLogical(ff, slot)) {
+                assertEquals("the shared lock dir must be created umask-governed",
+                        Files.DIR_MODE_SHARED, ff.modeFor(lockDir));
+            }
+            // The per-slot directory keeps the restrictive mode: one process
+            // creates it and only that process writes inside it.
+            RecordingMkdirFacade slotFf = new RecordingMkdirFacade();
+            String ownSlot = parentDir + "/own-slot";
+            slotFf.mkdir(ownSlot, Files.DIR_MODE_DEFAULT);
+            assertEquals("a slot dir must not be loosened",
+                    Files.DIR_MODE_DEFAULT, slotFf.modeFor(ownSlot));
+        });
+    }
+
     @Test
     public void testLogicalLockReportsLockDirectoryCreationFailure() throws Exception {
         TestUtils.assertMemoryLeak(() -> {
@@ -259,6 +285,22 @@ private static boolean isDir(String path) {
         return true;
     }
 
+    /** Records the mode each directory was created with, then delegates. */
+    private static final class RecordingMkdirFacade extends DelegatingFilesFacade {
+        private final java.util.Map<String, Integer> modes = new java.util.HashMap<>();
+
+        int modeFor(String path) {
+            Integer mode = modes.get(path);
+            return mode == null ? -1 : mode;
+        }
+
+        @Override
+        public int mkdir(String path, int mode) {
+            modes.put(path, mode);
+            return super.mkdir(path, mode);
+        }
+    }
+
     private static final class LockDirectoryFailureFacade extends DelegatingFilesFacade {
         private final String lockDir;
         private int openRwCalls;