Fix six moderate findings in the SF dictionary paths

Guards that don't guard. Already closed by the varint packing change:
readVarintAt returns -1 for a truncated varint instead of 0 with its end
at the limit, so a caller's `p > limit` bail-out can fire at an exact
boundary. It had no test; one now feeds a delta declaring two entries
while carrying one, ending precisely on the boundary that used to slip
through. Reverted to the indistinguishable shape it reports
expected:<0> but was:<2> -- the mirror claiming symbols it holds no
bytes for, which the next catch-up would ship as a chunk whose
deltaCount exceeds its payload.

DIR_MODE_SHARED. The javadoc claimed umask governed it "exactly as it
does for the sf_dir these live under", but sf_dir was created with an
explicit 0755 and umask can only CLEAR bits -- so under umask 002 the
lock directory became group-writable while sf_dir did not, and a second
uid still could not create its slot directory. build() failed one level
before the problem the mode existed to solve. sf_dir now gets the same
mode, and the mode is 01777 rather than 0777: without the sticky bit any
local user with write access can unlink another process's lock file,
which does not release that process's flock but does free the pathname,
so the next acquirer mints a second inode and locks it successfully.
umask masks only the rwx bits, so the restricted-deletion semantics
survive every umask and the usual 022 still yields 0755 plus sticky. The
slot directory itself stays 0755 -- only its creator writes it.

Foreground retry observability. Retrying an endpoint-policy rejection is
what the contract demands, but dispatchError ran only in the terminal
branches, so the retry was programmatically invisible: a revoked token
produced a throttled slf4j WARN and nothing else, in a library that
ships embedded in user applications. flush() kept returning success
until SF filled, and the failure then surfaced as "cursor ring
backpressured ... sf_max_total_bytes too small" -- pointing the operator
at disk sizing. The retry branches now dispatch a RETRIABLE SenderError,
and lastReconnectError is promoted from a connectLoop local that never
escaped to a volatile field, which close()'s drain-timeout throw uses to
name the real cause.

scanRecoveredChunks. It rejected entryCount <= 0 but never checked that
entryBytes actually holds entryCount well-formed entries. The CRC proves
the bytes are what was written, not that the header triple is
consistent, and the only write-side guard sits behind an assert that
production never runs. Such a chunk is now rejected during the scan, so
the trusted prefix ends there as it would on a CRC failure, instead of
decodeLoadedSymbols discovering it later and quarantining the whole slot.

ensureSentDictCapacity. The MAX_SENT_DICT_BYTES arm deliberately latches
via recordFatal before throwing, because accumulateSentDict runs AFTER
the frame's sendBinary and a bare throw unwinds into a reconnect that
replays the same frame. The allocation one line below had no such latch,
so a genuine out-of-memory took exactly the livelock the ceiling arm
exists to prevent. It now latches too.

PersistedSymbolDict.close(). It set closed = true and then ran munmap and
truncate unguarded, so a throw from either stranded the scratch buffer
and the fd for the process's lifetime -- a retry short-circuits on
`closed`. Each step is now in its own try/catch, as CursorSendEngine's
close already was.

Two tests were updated rather than added: the one asserting the scan
ACCEPTED an inconsistent chunk now asserts it ends the trusted prefix
(that path's exception-type change stays pinned on its reachable sibling
in CursorSendEngineTest), and SlotLockTest's mode assertion follows the
constant. The OOM latch and the close hardening have no test: neither an
Unsafe allocation failure nor a throwing munmap is injectable through
any existing seam.

Full client suite: 2702 tests green.

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

Author: glasstiger

core/src/main/java/io/questdb/client/Sender.java

@@ -1550,7 +1550,11 @@ public Sender build() {
                     slotPath = null;
                 } else {
                     if (!Files.exists(sfDir)) {
-                        int rc = Files.mkdir(sfDir, Files.DIR_MODE_DEFAULT);
+                        // DIR_MODE_SHARED, matching .slot-locks: a second uid has to create
+                        // its own slot directory in here, so restricting this to 0755 makes
+                        // the shared lock directory pointless (see Files.DIR_MODE_SHARED).
+                        // Under the usual umask 022 this is still 0755, plus sticky.
+                        int rc = Files.mkdir(sfDir, Files.DIR_MODE_SHARED);
                         // mkdir is non-zero on failure, but "already exists"
                         // is one such failure. Multiple SF senders sharing one
                         // sf_dir can be built concurrently (the pool calls

core/src/main/java/io/questdb/client/cutlass/qwp/client/QwpWebSocketSender.java

@@ -3347,14 +3347,24 @@ private void drainOnClose(boolean errorOwnedByCustomHandler) {
             }
             if (System.nanoTime() >= deadlineNanos) {
                 long acked = cursorEngine.ackedFsn();
-                LOG.warn("close() drain timed out after {}ms [target={} acked={}], pending data may be lost",
-                        closeFlushTimeoutMillis, target, acked);
+                // Name the outage the I/O thread is riding out, when there is one. A
+                // foreground sender now retries endpoint-policy rejections indefinitely,
+                // so a revoked token reaches the operator HERE -- and blaming timeout
+                // tuning for what is actually an auth failure is how the review found
+                // this path misdirecting.
+                CursorWebSocketSendLoop loop = cursorSendLoop;
+                Throwable outage = loop == null ? null : loop.lastReconnectError();
+                LOG.warn("close() drain timed out after {}ms [target={} acked={}], pending data may be lost{}",
+                        closeFlushTimeoutMillis, target, acked,
+                        outage == null ? "" : "; wire is not draining: " + outage.getMessage());
                 throw new LineSenderException("close() drain timed out after ")
                         .put(closeFlushTimeoutMillis).put(" ms [targetFsn=")
                         .put(target).put(", ackedFsn=").put(acked)
                         .put("] - server did not acknowledge ")
                         .put(target - acked)
-                        .put(" pending batches; data may be lost (use larger closeFlushTimeoutMillis or smaller batches)");
+                        .put(outage == null
+                                ? " pending batches; data may be lost (use larger closeFlushTimeoutMillis or smaller batches)"
+                                : " pending batches; the wire is not draining: " + outage.getMessage());
             }
             java.util.concurrent.locks.LockSupport.parkNanos(50_000L);
         }

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

@@ -376,6 +376,12 @@ public final class CursorWebSocketSendLoop implements QuietCloseable {
     // successful connect attempt in any mode). Once true, stays true. This is
     // connection-state observability only; reconnect policy is role-based.
     private volatile boolean hasEverConnected;
+    // Cause of the outage the reconnect loop is currently riding out, or null once a
+    // connect succeeds. Written by the I/O thread, read by the producer thread so a
+    // backpressure or drain-timeout failure can NAME the reason the wire is not draining
+    // instead of blaming disk sizing. Was a connectLoop local that never escaped, which
+    // is why a revoked token surfaced to operators as "sf_max_total_bytes too small".
+    private volatile Throwable lastReconnectError;
     private volatile Thread ioThread;
     // Typed marker for a durable-ack CAPABILITY-GAP terminal: set (before the
     // terminalError latch, so a checkError() caller that observes the latch is
@@ -1599,7 +1605,7 @@ private void connectLoop(Throwable initial, String phase, long paceFirstAttemptM
         }
         int attempts = 0;
         long lastLogNanos = 0L;
-        Throwable lastReconnectError = initial;
+        lastReconnectError = initial;
         while (running) {
             attempts++;
             totalReconnectAttempts.incrementAndGet();
@@ -1665,6 +1671,8 @@ private void connectLoop(Throwable initial, String phase, long paceFirstAttemptM
                 }
                 resetCatchUpCapGapEpisode();
                 lastReconnectError = e;
+                dispatchRetriedEndpointPolicyFailure(
+                        SenderError.Category.SECURITY_ERROR, "ws-upgrade-failed: " + e.getMessage());
                 long now = System.nanoTime();
                 if (now - lastLogNanos >= RECONNECT_LOG_THROTTLE_NANOS) {
                     LOG.warn("{} attempt {}: foreground auth/upgrade policy rejected the connection; "
@@ -1703,6 +1711,9 @@ private void connectLoop(Throwable initial, String phase, long paceFirstAttemptM
                 }
                 resetCatchUpCapGapEpisode();
                 lastReconnectError = e;
+                dispatchRetriedEndpointPolicyFailure(
+                        SenderError.Category.PROTOCOL_VIOLATION,
+                        "durable-ack-mismatch: " + e.getMessage());
                 long now = System.nanoTime();
                 if (now - lastLogNanos >= RECONNECT_LOG_THROTTLE_NANOS) {
                     LOG.warn("{} attempt {}: foreground durable-ack capability is temporarily "
@@ -1821,6 +1832,45 @@ private void connectLoop(Throwable initial, String phase, long paceFirstAttemptM
      * failure is then a transient the drainer must ride out, never a producer-fatal
      * terminal (Invariant B).
      */
+    /**
+     * Reports an endpoint-policy rejection a FOREGROUND sender is riding out.
+     * <p>
+     * Retrying is what the store-and-forward contract demands, but until now the retry was
+     * programmatically invisible: dispatchError ran only in the terminal branches, so a
+     * revoked token produced nothing but a throttled slf4j WARN -- in a library that ships
+     * embedded in user applications, frequently with no binding configured. flush() kept
+     * returning success until SF filled, and the failure then surfaced as "cursor ring
+     * backpressured ... sf_max_total_bytes too small", pointing the operator at disk
+     * sizing. That is the same complaint this PR makes to justify keeping STARTUP
+     * terminal; post-start it was merely delayed by SF capacity, not avoided.
+     * <p>
+     * RETRIABLE, not TERMINAL: the handler learns the wire is being rejected while the
+     * producer stays alive and no data is at risk.
+     */
+    private void dispatchRetriedEndpointPolicyFailure(SenderError.Category category, String message) {
+        long fromFsn = engine.ackedFsn() + 1L;
+        dispatchError(new SenderError(
+                category,
+                SenderError.Policy.RETRIABLE,
+                SenderError.NO_STATUS_BYTE,
+                message,
+                SenderError.NO_MESSAGE_SEQUENCE,
+                fromFsn,
+                Math.max(fromFsn, engine.publishedFsn()),
+                null,
+                System.nanoTime()
+        ));
+    }
+
+    /**
+     * Cause of the outage the reconnect loop is riding out, or {@code null} when the wire
+     * is up. Lets a producer-side failure name the real reason rather than blaming disk
+     * sizing -- see the field.
+     */
+    public Throwable lastReconnectError() {
+        return lastReconnectError;
+    }
+
     private boolean endpointPolicyFailureIsTerminal() {
         return reconnectPolicy == ReconnectPolicy.ORPHAN || !hasEverConnected;
     }
@@ -2341,6 +2391,7 @@ private void swapClient(WebSocketClient newClient) {
         // failures on a foreground sender and rides them out instead, because
         // store-and-forward now owns the buffered data (Invariant B).
         this.hasEverConnected = true;
+        this.lastReconnectError = null;
         if (old != null) {
             try {
                 old.close();
@@ -2552,19 +2603,32 @@ private void ensureSentDictCapacity(long required) {
         if (newCap > MAX_SENT_DICT_BYTES) {
             newCap = MAX_SENT_DICT_BYTES;
         }
-        if (sentDictBytesOwned) {
-            sentDictBytesAddr = Unsafe.realloc(
-                    sentDictBytesAddr,
-                    sentDictBytesCapacity,
-                    (int) newCap,
-                    MemoryTag.NATIVE_DEFAULT);
-        } else {
-            long newAddr = Unsafe.malloc((int) newCap, MemoryTag.NATIVE_DEFAULT);
-            if (sentDictBytesLen > 0) {
-                Unsafe.getUnsafe().copyMemory(sentDictBytesAddr, newAddr, sentDictBytesLen);
+        try {
+            if (sentDictBytesOwned) {
+                sentDictBytesAddr = Unsafe.realloc(
+                        sentDictBytesAddr,
+                        sentDictBytesCapacity,
+                        (int) newCap,
+                        MemoryTag.NATIVE_DEFAULT);
+            } else {
+                long newAddr = Unsafe.malloc((int) newCap, MemoryTag.NATIVE_DEFAULT);
+                if (sentDictBytesLen > 0) {
+                    Unsafe.getUnsafe().copyMemory(sentDictBytesAddr, newAddr, sentDictBytesLen);
+                }
+                sentDictBytesAddr = newAddr;
+                sentDictBytesOwned = true;
             }
-            sentDictBytesAddr = newAddr;
-            sentDictBytesOwned = true;
+        } catch (Throwable t) {
+            // Latch for exactly the reason the MAX_SENT_DICT_BYTES arm above does, and it
+            // was asymmetric until now: accumulateSentDict runs AFTER the frame's
+            // sendBinary, so a bare throw unwinds to ioLoop -> fail() -> connectLoop,
+            // which (running still true) reconnects and replays the same frame, which
+            // re-attempts the same allocation. A genuine out-of-memory therefore became an
+            // unbounded reconnect livelock instead of the loud failure the ceiling arm
+            // promises. recordFatal flips running=false so connectLoop winds down and
+            // checkError() surfaces it; the throw still unwinds past the pending copy.
+            recordFatal(t);
+            throw t;
         }
         sentDictBytesCapacity = (int) newCap;
     }

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

@@ -499,35 +499,58 @@ public synchronized void close() {
             return;
         }
         closed = true;
+        // Each step in its own try/catch, as CursorSendEngine.close() already does.
+        // `closed` is set first, so a throw from any step short-circuits every retry --
+        // stranding the scratch buffer and the fd for the process's lifetime. Unreachable
+        // with FilesFacade.INSTANCE (munmap and truncate are native calls returning
+        // int/boolean), but the ff seam exists precisely so a test CAN inject a throw,
+        // and a close path must not depend on its own steps never failing.
         if (loadedEntriesAddr != 0L) {
-            Unsafe.free(loadedEntriesAddr, loadedEntriesLen, MemoryTag.NATIVE_DEFAULT);
+            try {
+                Unsafe.free(loadedEntriesAddr, loadedEntriesLen, MemoryTag.NATIVE_DEFAULT);
+            } catch (Throwable ignored) {
+            }
             // Null after freeing (like scratchAddr below) so a future accessor that
             // reads loadedEntriesAddr()/loadedEntriesLen() post-close cannot
             // dereference freed native memory; the getters are not closed-guarded.
             loadedEntriesAddr = 0L;
             loadedEntriesLen = 0;
         }
         if (appendMapAddr != 0L) {
-            ff.munmap(appendMapAddr, appendMapCapacity, MemoryTag.MMAP_DEFAULT);
+            long mapAddr = appendMapAddr;
+            long mapCapacity = appendMapCapacity;
             appendMapAddr = 0L;
             appendMapCapacity = 0L;
             appendMapOffset = 0L;
-            // The active window reserves space past the logical end. Return that tail on
-            // orderly close; after a crash open() treats the zero-filled reserve as
-            // a torn trailing chunk and truncates it to the same appendOffset.
-            if (!ff.truncate(fd, appendOffset)) {
-                LOG.warn("symbol dict {} could not trim mmap reserve to {}; recovery will "
-                                + "discard the zero-filled tail on the next open",
-                        FILE_NAME, appendOffset);
+            try {
+                ff.munmap(mapAddr, mapCapacity, MemoryTag.MMAP_DEFAULT);
+            } catch (Throwable ignored) {
+            }
+            try {
+                // The active window reserves space past the logical end. Return that tail on
+                // orderly close; after a crash open() treats the zero-filled reserve as
+                // a torn trailing chunk and truncates it to the same appendOffset.
+                if (!ff.truncate(fd, appendOffset)) {
+                    LOG.warn("symbol dict {} could not trim mmap reserve to {}; recovery will "
+                                    + "discard the zero-filled tail on the next open",
+                            FILE_NAME, appendOffset);
+                }
+            } catch (Throwable ignored) {
             }
         }
         if (scratchAddr != 0L) {
-            Unsafe.free(scratchAddr, scratchCap, MemoryTag.NATIVE_DEFAULT);
+            try {
+                Unsafe.free(scratchAddr, scratchCap, MemoryTag.NATIVE_DEFAULT);
+            } catch (Throwable ignored) {
+            }
             scratchAddr = 0L;
             scratchCap = 0;
         }
         if (fd >= 0) {
-            ff.close(fd);
+            try {
+                ff.close(fd);
+            } catch (Throwable ignored) {
+            }
         }
     }
 
@@ -800,6 +823,9 @@ private static RecoveryScan scanAndCopyRecoveredChunks(long inputAddr, int len,
                     || entriesLen + entryBytes > Integer.MAX_VALUE) {
                 break;
             }
+            if (!isConsistentEntryRegion(inputAddr, entriesStart, entryBytes, entryCount)) {
+                break;
+            }
             Unsafe.getUnsafe().copyMemory(inputAddr + entriesStart, dstAddr + entriesLen, entryBytes);
             entriesLen += entryBytes;
             diskPos = chunkEndI + CRC_SIZE;
@@ -808,6 +834,51 @@ private static RecoveryScan scanAndCopyRecoveredChunks(long inputAddr, int len,
         return new RecoveryScan(count, (int) entriesLen, diskPos);
     }
 
+    /**
+     * Whether {@code [start, start + bytes)} holds exactly {@code count} well-formed
+     * {@code [len varint][utf8]} entries, consuming the region exactly.
+     * <p>
+     * The chunk CRC proves the bytes are what was WRITTEN; it says nothing about whether
+     * the header triple is self-consistent. The only write-side guard,
+     * {@code validateRawEntries}, sits behind an {@code assert} -- and this library ships
+     * embedded in user applications, which run without {@code -ea}. So a producer bug or
+     * a torn write that happens to re-checksum could record a chunk whose stored
+     * entryCount disagrees with its entries, shifting the dense id-&gt;symbol map for
+     * everything above it.
+     * <p>
+     * Checking here ends the trusted prefix at that chunk -- the same treatment a CRC
+     * failure gets -- instead of letting decodeLoadedSymbols discover it later and throw
+     * two layers up, which quarantines the whole slot rather than salvaging its intact
+     * prefix. It costs one varint decode per entry on a cold path that already walks
+     * every entry immediately afterwards.
+     */
+    private static boolean isConsistentEntryRegion(long addr, int start, long bytes, long count) {
+        long p = start;
+        long limit = start + bytes;
+        for (long i = 0; i < count; i++) {
+            long len = 0;
+            int shift = 0;
+            boolean terminated = false;
+            while (p < limit) {
+                byte b = Unsafe.getUnsafe().getByte(addr + p++);
+                len |= (long) (b & 0x7F) << shift;
+                if ((b & 0x80) == 0) {
+                    terminated = true;
+                    break;
+                }
+                shift += 7;
+                if (shift > 35) {
+                    return false; // over-long run: a canonical length varint is <= 5 bytes
+                }
+            }
+            if (!terminated || len < 0 || p + len > limit) {
+                return false;
+            }
+            p += len;
+        }
+        return p == limit;
+    }
+
     private static PersistedSymbolDict openFresh(FilesFacade ff, String filePath) {
         int fd = ff.openCleanRW(filePath);
         if (fd < 0) {

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

@@ -120,7 +120,7 @@ public static SlotLock acquireLogical(FilesFacade ff, String slotDir) {
             throw new IllegalArgumentException(
                     "slotDir must contain a parent and slot name: " + slotDir);
         }
-        // SHARED_DIR_MODE, not DIR_MODE_DEFAULT: unlike a slot directory -- which
+        // DIR_MODE_SHARED, 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

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

@@ -76,16 +76,32 @@ public final class Files {
      */
     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;
+     * Creation mode for a directory that processes running as DIFFERENT users must all
+     * create entries in: the store-and-forward root and its logical slot-lock directory.
+     * <b>01777 octal -- 0777 plus the sticky bit.</b>
+     * <p>
+     * The deployment's umask decides the sharing policy: the usual 022 yields 01755, i.e.
+     * exactly {@link #DIR_MODE_DEFAULT} plus sticky, so the common case is unchanged;
+     * a shared-group deployment (umask 002) gets the 01775 a second uid needs. umask
+     * masks only the rwx bits, never the sticky bit, so the restricted-deletion semantics
+     * survive every umask.
+     * <p>
+     * The sticky bit is not optional here. Without it, on a multi-tenant host any local
+     * user with write access to the directory can unlink another process's lock file --
+     * which does not release that process's flock, but does free the pathname, so the
+     * next acquirer creates a SECOND inode and locks it successfully. Two owners of a
+     * lock whose entire job is mutual exclusion. Sticky restricts deletion to the file's
+     * owner and closes that.
+     * <p>
+     * This must be applied to EVERY directory on the path a foreign uid has to create in.
+     * Applying it only to {@code .slot-locks} -- while {@code sf_dir} itself stays 0755 --
+     * is a half-measure: umask can only clear bits, so under umask 002 the lock directory
+     * becomes group-writable while sf_dir does not, and the second uid still cannot create
+     * its slot directory. build() then fails one level before the problem this mode
+     * exists to solve. Do NOT use it for a directory only its creator writes to; the slot
+     * directory itself stays {@link #DIR_MODE_DEFAULT}.
+     */
+    public static final int DIR_MODE_SHARED = 1023;
 
     /** {@code dirent.d_type} sentinel: type unknown (filesystem doesn't fill it). */
     public static final int DT_UNKNOWN = 0;