Add QWP failure-path coverage

Author: glasstiger

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

@@ -172,8 +172,9 @@ public final class PersistedSymbolDict implements QuietCloseable {
     // tests inject a fault facade to exercise recovery I/O failures (a truncate
     // that cannot drop a torn tail, a short write) without a real broken disk.
     private final FilesFacade ff;
-    // Production writes directly into segmented append mappings. Custom facades retain the
-    // positioned-write path so fault tests can inject short writes through ff.write.
+    // Production writes directly into segmented append mappings. Wrapping facades retain the
+    // positioned-write path by default so fault tests can inject short writes through ff.write;
+    // mmap-specific fault facades opt in through FilesFacade.isMmapAllowed().
     private final boolean mappedAppend;
     // True only when recovery parsed the file through a temporary read-only
     // mmap instead of allocating a second native buffer as large as the file.
@@ -210,7 +211,7 @@ private PersistedSymbolDict(
     ) {
         this.ff = ff;
         this.fd = fd;
-        this.mappedAppend = ff == FilesFacade.INSTANCE;
+        this.mappedAppend = ff.isMmapAllowed();
         this.mappedRecoveryInput = mappedRecoveryInput;
         this.appendOffset = appendOffset;
         this.size = size;
@@ -492,7 +493,7 @@ public synchronized void close() {
             loadedEntriesLen = 0;
         }
         if (appendMapAddr != 0L) {
-            Files.munmap(appendMapAddr, appendMapCapacity, MemoryTag.MMAP_DEFAULT);
+            ff.munmap(appendMapAddr, appendMapCapacity, MemoryTag.MMAP_DEFAULT);
             appendMapAddr = 0L;
             appendMapCapacity = 0L;
             appendMapOffset = 0L;
@@ -639,13 +640,13 @@ private static PersistedSymbolDict openExisting(FilesFacade ff, String filePath,
             return null;
         }
         int len = (int) fileLen; // open() bounds fileLen to [HEADER_SIZE, Integer.MAX_VALUE)
-        boolean mappedInput = ff == FilesFacade.INSTANCE;
+        boolean mappedInput = ff.isMmapAllowed();
         long inputAddr = 0L;
         long entriesAddr = 0L;
         int entriesLen = 0;
         try {
             if (mappedInput) {
-                inputAddr = Files.mmap(fd, fileLen, 0L, Files.MAP_RO, MemoryTag.MMAP_DEFAULT);
+                inputAddr = ff.mmap(fd, fileLen, 0L, Files.MAP_RO, MemoryTag.MMAP_DEFAULT);
                 if (inputAddr == Files.FAILED_MMAP_ADDRESS) {
                     inputAddr = 0L;
                     throw new IllegalStateException("could not mmap symbol dictionary for recovery");
@@ -675,7 +676,7 @@ private static PersistedSymbolDict openExisting(FilesFacade ff, String filePath,
                 copyRecoveredEntries(inputAddr, scan.validLen, entriesAddr, entriesLen);
             }
             if (mappedInput) {
-                Files.munmap(inputAddr, fileLen, MemoryTag.MMAP_DEFAULT);
+                ff.munmap(inputAddr, fileLen, MemoryTag.MMAP_DEFAULT);
             } else {
                 Unsafe.free(inputAddr, len, MemoryTag.NATIVE_DEFAULT);
             }
@@ -695,7 +696,7 @@ private static PersistedSymbolDict openExisting(FilesFacade ff, String filePath,
         } catch (Throwable t) {
             if (inputAddr != 0L) {
                 if (mappedInput) {
-                    Files.munmap(inputAddr, fileLen, MemoryTag.MMAP_DEFAULT);
+                    ff.munmap(inputAddr, fileLen, MemoryTag.MMAP_DEFAULT);
                 } else {
                     Unsafe.free(inputAddr, len, MemoryTag.NATIVE_DEFAULT);
                 }
@@ -923,9 +924,9 @@ private void ensureAppendMap(long required) {
             appendMapAddr = 0L;
             appendMapCapacity = 0L;
             appendMapOffset = 0L;
-            Files.munmap(oldAddr, oldCapacity, MemoryTag.MMAP_DEFAULT);
+            ff.munmap(oldAddr, oldCapacity, MemoryTag.MMAP_DEFAULT);
         }
-        long newAddr = Files.mmap(
+        long newAddr = ff.mmap(
                 fd, newCapacity, newOffset, Files.MAP_RW, MemoryTag.MMAP_DEFAULT);
         if (newAddr == Files.FAILED_MMAP_ADDRESS) {
             throw new IllegalStateException("could not mmap append region for "

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

@@ -26,9 +26,11 @@
 
 import io.questdb.client.std.Compat;
 import io.questdb.client.std.Files;
+import io.questdb.client.std.FilesFacade;
 import io.questdb.client.std.MemoryTag;
 import io.questdb.client.std.QuietCloseable;
 import io.questdb.client.std.Unsafe;
+import org.jetbrains.annotations.TestOnly;
 
 import java.nio.charset.StandardCharsets;
 import java.nio.file.Path;
@@ -65,10 +67,12 @@ public final class SlotLock implements QuietCloseable {
     private static final String LOCK_FILE_NAME = ".lock";
     private static final String LOCK_PID_FILE_NAME = ".lock.pid";
     private static final String LOGICAL_LOCK_DIR_NAME = ".slot-locks";
+    private final FilesFacade ff;
     private final String slotDir;
     private int fd;
 
-    private SlotLock(String slotDir, int fd) {
+    private SlotLock(FilesFacade ff, String slotDir, int fd) {
+        this.ff = ff;
         this.slotDir = slotDir;
         this.fd = fd;
     }
@@ -83,10 +87,10 @@ private SlotLock(String slotDir, int fd) {
      */
     public static SlotLock acquire(String slotDir) {
         validateSlotDir(slotDir);
-        ensureDirectory(slotDir, "slot dir");
+        ensureDirectory(FilesFacade.INSTANCE, slotDir, "slot dir");
         String lockPath = slotDir + "/" + LOCK_FILE_NAME;
         String pidPath = slotDir + "/" + LOCK_PID_FILE_NAME;
-        return acquireAt(slotDir, lockPath, pidPath);
+        return acquireAt(FilesFacade.INSTANCE, slotDir, lockPath, pidPath);
     }
 
     /**
@@ -102,6 +106,12 @@ public static SlotLock acquire(String slotDir) {
      * the fresh directory through the old pathname.
      */
     public static SlotLock acquireLogical(String slotDir) {
+        return acquireLogical(FilesFacade.INSTANCE, slotDir);
+    }
+
+    /** Facade-aware variant used to exercise logical-lock I/O failures. */
+    @TestOnly
+    public static SlotLock acquireLogical(FilesFacade ff, String slotDir) {
         validateSlotDir(slotDir);
         Path slotPath = Paths.get(slotDir);
         Path parentPath = slotPath.getParent();
@@ -113,44 +123,44 @@ public static SlotLock acquireLogical(String slotDir) {
         String parentDir = parentPath.toString();
         String slotName = slotNamePath.toString();
         String logicalLockDir = parentDir + "/" + LOGICAL_LOCK_DIR_NAME;
-        ensureDirectory(logicalLockDir, "logical slot lock dir");
+        ensureDirectory(ff, logicalLockDir, "logical slot lock dir");
         String lockPath = logicalLockDir + "/" + slotName + ".lock";
         String pidPath = logicalLockDir + "/" + slotName + ".lock.pid";
-        return acquireAt(slotDir, lockPath, pidPath);
+        return acquireAt(ff, slotDir, lockPath, pidPath);
     }
 
-    private static SlotLock acquireAt(String slotDir, String lockPath, String pidPath) {
-        int fd = Files.openRW(lockPath);
+    private static SlotLock acquireAt(FilesFacade ff, String slotDir, String lockPath, String pidPath) {
+        int fd = ff.openRW(lockPath);
         if (fd < 0) {
             throw new IllegalStateException(
                     "could not open slot lock file: " + lockPath);
         }
         boolean ok = false;
         try {
-            int rc = Files.lock(fd);
+            int rc = ff.lock(fd);
             if (rc != 0) {
                 String holder = readHolder(pidPath);
                 throw new IllegalStateException(
                         "sf slot already in use by another process [slot="
                                 + slotDir + ", holder=" + holder + "]");
             }
-            writePid(pidPath);
+            writePid(ff, pidPath);
             ok = true;
-            return new SlotLock(slotDir, fd);
+            return new SlotLock(ff, slotDir, fd);
         } finally {
             if (!ok) {
-                Files.close(fd);
+                ff.close(fd);
             }
         }
     }
 
-    private static void ensureDirectory(String path, String description) {
-        if (!Files.exists(path)) {
-            int rc = Files.mkdir(path, Files.DIR_MODE_DEFAULT);
+    private static void ensureDirectory(FilesFacade ff, String path, String description) {
+        if (!ff.exists(path)) {
+            int rc = ff.mkdir(path, Files.DIR_MODE_DEFAULT);
             // Multiple senders may create the shared parent lock directory
             // concurrently. Treat EEXIST as success, just as the builder does
             // for the SF root itself.
-            if (rc != 0 && !Files.exists(path)) {
+            if (rc != 0 && !ff.exists(path)) {
                 throw new IllegalStateException(
                         "could not create " + description + ": " + path + " rc=" + rc);
             }
@@ -174,7 +184,7 @@ public void close() {
         // file or the .lock.pid sidecar — a stale PID is harmless (next
         // acquirer overwrites .lock.pid on success).
         if (fd >= 0) {
-            Files.close(fd);
+            ff.close(fd);
             fd = -1;
         }
     }
@@ -204,33 +214,33 @@ private static String readHolder(String pidPath) {
         }
     }
 
-    private static void writePid(String pidPath) {
+    private static void writePid(FilesFacade ff, String pidPath) {
         long pid;
         try {
             pid = Compat.currentPid();
         } catch (Throwable ignored) {
             // Diagnostic-only — never block lock acquisition on it.
             pid = -1L;
         }
-        int wfd = Files.openRW(pidPath);
+        int wfd = ff.openRW(pidPath);
         if (wfd < 0) {
             // Diagnostic-only — never block lock acquisition on it.
             return;
         }
         try {
-            Files.truncate(wfd, 0L);
+            ff.truncate(wfd, 0L);
             byte[] payload = (pid + "\n").getBytes(StandardCharsets.UTF_8);
             long addr = Unsafe.malloc(payload.length, MemoryTag.NATIVE_DEFAULT);
             try {
                 for (int i = 0; i < payload.length; i++) {
                     Unsafe.getUnsafe().putByte(addr + i, payload[i]);
                 }
-                Files.write(wfd, addr, payload.length, 0L);
+                ff.write(wfd, addr, payload.length, 0L);
             } finally {
                 Unsafe.free(addr, payload.length, MemoryTag.NATIVE_DEFAULT);
             }
         } finally {
-            Files.close(wfd);
+            ff.close(wfd);
         }
     }
 }

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

@@ -85,6 +85,16 @@ public interface FilesFacade {
 
     int fsync(int fd);
 
+    /**
+     * Whether callers should use this facade's mmap path. The production facade
+     * returns {@code true}; wrapping fault facades retain positioned I/O unless
+     * they explicitly opt in, preserving their ability to inject short reads and
+     * writes.
+     */
+    default boolean isMmapAllowed() {
+        return this == INSTANCE;
+    }
+
     long length(int fd);
 
     /**
@@ -107,6 +117,19 @@ public interface FilesFacade {
 
     int mkdir(String path, int mode);
 
+    /**
+     * Maps a file region. Kept on the facade so mmap failures can be injected
+     * without relying on platform-specific filesystem behavior.
+     */
+    default long mmap(int fd, long len, long offset, int flags, int memoryTag) {
+        return Files.mmap(fd, len, offset, flags, memoryTag);
+    }
+
+    /** Releases a region returned by {@link #mmap(int, long, long, int, int)}. */
+    default void munmap(long address, long len, int memoryTag) {
+        Files.munmap(address, len, memoryTag);
+    }
+
     int openCleanRW(String path);
 
     /**

core/src/test/java/io/questdb/client/test/cutlass/qwp/client/QwpWebSocketEncoderTest.java

@@ -1285,6 +1285,69 @@ public void testReset() throws Exception {
         });
     }
 
+    @Test
+    public void testSplitMessageValidatesArgumentsAndSizeOverflow() throws Exception {
+        assertMemoryLeak(() -> {
+            try (QwpWebSocketEncoder encoder = new QwpWebSocketEncoder();
+                 MicrobatchBuffer target = new MicrobatchBuffer(64);
+                 QwpTableBuffer table = new QwpTableBuffer("alpha")) {
+                GlobalSymbolDictionary dict = new GlobalSymbolDictionary();
+                int aapl = dict.getOrAddSymbol("AAPL");
+                dict.getOrAddSymbol("GOOG");
+                table.getOrCreateColumn("sym", TYPE_SYMBOL, false)
+                        .addSymbolWithGlobalId("AAPL", aapl);
+                table.nextRow();
+
+                encoder.beginMessage(1, dict, -1, 1);
+                int tableBodyOffset = encoder.getBuffer().getPosition();
+                encoder.addTable(table);
+                int tableBodyLength = encoder.getBuffer().getPosition() - tableBodyOffset;
+                encoder.finishMessage();
+
+                target.setBufferPos(1);
+                assertFailure(IllegalStateException.class,
+                        "split message target is not empty",
+                        () -> encoder.copySplitMessage(
+                                target, tableBodyOffset, tableBodyLength, false, -1, 1));
+                Assert.assertEquals("invalid target must not be modified", 1, target.getBufferPos());
+                target.reset();
+
+                assertFailure(IllegalArgumentException.class,
+                        "table body slice is outside the staged message",
+                        () -> encoder.copySplitMessage(
+                                target, tableBodyOffset - 1, tableBodyLength, false, -1, 1));
+                assertFailure(IllegalArgumentException.class,
+                        "table body slice is outside the staged message",
+                        () -> encoder.copySplitMessage(
+                                target, tableBodyOffset, -1, false, -1, 1));
+                assertFailure(IllegalArgumentException.class,
+                        "table body slice is outside the staged message",
+                        () -> encoder.copySplitMessage(
+                                target, tableBodyOffset, tableBodyLength + 1, false, -1, 1));
+                Assert.assertEquals("invalid slices must not write the target", 0, target.getBufferPos());
+
+                assertFailure(IllegalStateException.class,
+                        "split delta does not match the staged message"
+                                + " [stagedStart=0, stagedCount=2, splitStart=1, splitCount=1]",
+                        () -> encoder.copySplitMessage(
+                                target, tableBodyOffset, tableBodyLength, false, 0, 1));
+                Assert.assertEquals("delta mismatch must not write the target", 0, target.getBufferPos());
+
+                assertFailure(IllegalArgumentException.class,
+                        "tableBodyLength must be non-negative",
+                        () -> encoder.getSplitMessageSize(-1, -1, 1));
+                long overflowSize = (long) HEADER_SIZE
+                        + 1 // delta start 0
+                        + 1 // delta count 2
+                        + encoder.getDeltaEntriesLen()
+                        + Integer.MAX_VALUE;
+                assertFailure(OutOfMemoryError.class,
+                        "split QWP message size overflow: " + overflowSize,
+                        () -> encoder.getSplitMessageSize(Integer.MAX_VALUE, -1, 1));
+            }
+        });
+    }
+
     @Test
     public void testSplitMessageCopiesStagedTableBodies() throws Exception {
         assertMemoryLeak(() -> {
@@ -1388,6 +1451,22 @@ public void testVersionByteInHeader() throws Exception {
         });
     }
 
+    private static void assertFailure(
+            Class<? extends Throwable> expectedType,
+            String expectedMessage,
+            Runnable action
+    ) {
+        Throwable thrown = null;
+        try {
+            action.run();
+        } catch (Throwable t) {
+            thrown = t;
+        }
+        Assert.assertNotNull("expected " + expectedType.getSimpleName(), thrown);
+        Assert.assertEquals(expectedType, thrown.getClass());
+        Assert.assertEquals(expectedMessage, thrown.getMessage());
+    }
+
     private static final class Cursor {
         private long address;