Add QWP edge-case coverage

Author: glasstiger

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

@@ -52,6 +52,7 @@
 import io.questdb.client.std.Decimal256;
 import io.questdb.client.std.Decimal64;
 import io.questdb.client.std.Files;
+import io.questdb.client.std.FilesFacade;
 import io.questdb.client.std.IntList;
 import io.questdb.client.std.Numbers;
 import io.questdb.client.std.NumericException;
@@ -1012,6 +1013,8 @@ final class LineSenderBuilder {
         private static final int PROTOCOL_WEBSOCKET = 2;
         @TestOnly
         private static volatile Runnable quarantineAfterCloseHook;
+        @TestOnly
+        private static volatile FilesFacade quarantineFilesFacade = FilesFacade.INSTANCE;
         // Suffix for a slot set aside by quarantineTornSlot. Deliberately NOT the
         // sender's own slot name, so a restarted sender does not re-adopt it as its own;
         // quarantineTornSlot then marks it .failed, so the orphan drainer skips it too and
@@ -3061,15 +3064,16 @@ private static CursorSendEngine quarantineTornSlot(
                 hook.run();
             }
 
+            FilesFacade ff = quarantineFilesFacade;
             String quarantinePath = null;
             for (int i = 0; i < MAX_QUARANTINE_SLOT_ATTEMPTS; i++) {
                 String candidate = sfDir + "/" + senderId + QUARANTINE_SLOT_SUFFIX + i;
-                if (!Files.exists(candidate)) {
+                if (!ff.exists(candidate)) {
                     quarantinePath = candidate;
                     break;
                 }
             }
-            if (quarantinePath == null || Files.rename(slotPath, quarantinePath) != 0) {
+            if (quarantinePath == null || ff.rename(slotPath, quarantinePath) != 0) {
                 throw new LineSenderException(
                         detail + "; the affected data must be resent. The slot could not be set aside "
                                 + "automatically (" + (quarantinePath == null
@@ -3092,6 +3096,11 @@ public static void setQuarantineAfterCloseHookForTest(Runnable hook) {
             quarantineAfterCloseHook = hook;
         }
 
+        @TestOnly
+        public static void setQuarantineFilesFacadeForTest(FilesFacade ff) {
+            quarantineFilesFacade = ff == null ? FilesFacade.INSTANCE : ff;
+        }
+
         private static int resolveIPv4(String host) {
             try {
                 byte[] addr = InetAddress.getByName(host).getAddress();
@@ -4047,7 +4056,10 @@ public java.util.Map<String, Object> wsConfigSnapshotForTest() {
             m.put("max_background_drainers", maxBackgroundDrainers);
             m.put("max_frame_rejections", maxFrameRejections);
             m.put("poison_min_escalation_window_millis", poisonMinEscalationWindowMillis);
-            m.put("catchup_cap_gap_min_escalation_window_millis", catchUpCapGapMinEscalationWindowMillis);
+            m.put("catchup_cap_gap_min_escalation_window_millis",
+                    catchUpCapGapMinEscalationWindowMillis == PARAMETER_NOT_SET_EXPLICITLY
+                            ? CursorWebSocketSendLoop.DEFAULT_CATCHUP_CAP_GAP_MIN_ESCALATION_WINDOW_MILLIS
+                            : catchUpCapGapMinEscalationWindowMillis);
             m.put("error_inbox_capacity", errorInboxCapacity);
             m.put("connection_listener_inbox_capacity", connectionListenerInboxCapacity);
             m.put("token", httpToken);

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

@@ -467,6 +467,53 @@ public void testQuarantineFailsLoudlyWhenAllSlotNamesSaturated() throws Exceptio
         });
     }
 
+    @Test
+    public void testQuarantineRenameFailurePreservesOriginalSlot() throws Exception {
+        assertMemoryLeak(() -> {
+            writeAndTearUnreplayableSlot();
+            java.nio.file.Path slot = Paths.get(sfDir, "default");
+            int segmentCount = countSegmentFiles(slot);
+            Sender.LineSenderBuilder.setQuarantineFilesFacadeForTest(new DelegatingFilesFacade() {
+                @Override
+                public int rename(String oldPath, String newPath) {
+                    return -1;
+                }
+            });
+            try {
+                try (TestWebSocketServer good = new TestWebSocketServer(new DictReconstructingHandler())) {
+                    int port = good.getPort();
+                    good.start();
+                    Assert.assertTrue(good.awaitStart(5, TimeUnit.SECONDS));
+                    String cfg = "ws::addr=localhost:" + port + ";sf_dir=" + sfDir + ";";
+                    try {
+                        Sender.fromConfig(cfg).close();
+                        Assert.fail("build() must throw when the unreplayable slot rename fails");
+                    } catch (LineSenderException expected) {
+                        Assert.assertEquals(
+                                "recovered store-and-forward symbol dictionary is incomplete and cannot be rebuilt "
+                                        + "from the surviving frames (likely a host crash tore its unsynced tail): "
+                                        + "the frames reference symbol ids below their own delta start, which were "
+                                        + "introduced by frames since acked and trimmed away, so nothing still holds "
+                                        + "them; the recovered dictionary holds only 0 id(s) -- resend the affected "
+                                        + "data; the affected data must be resent. The slot could not be set aside "
+                                        + "automatically (rename to " + sfDir + "/default.unreplayable-0 failed), so "
+                                        + "this sender cannot start until " + sfDir + "/default is moved or removed "
+                                        + "by hand",
+                                expected.getMessage());
+                    }
+                }
+            } finally {
+                Sender.LineSenderBuilder.setQuarantineFilesFacadeForTest(null);
+            }
+            Assert.assertTrue("rename failure must preserve the original slot directory",
+                    java.nio.file.Files.isDirectory(slot));
+            Assert.assertEquals("rename failure must preserve every segment",
+                    segmentCount, countSegmentFiles(slot));
+            Assert.assertFalse("failed rename must not leave a quarantine directory",
+                    java.nio.file.Files.exists(Paths.get(sfDir, "default.unreplayable-0")));
+        });
+    }
+
     // Writes 12 delta frames (each introducing a new symbol) into the default slot
     // across several small segments, then makes the slot GENUINELY unreplayable: trims
     // the segment holding the earliest ids (munmap + unlink, exactly what SegmentManager

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

@@ -61,6 +61,23 @@ public void testAddRecoveredSymbol_appendsWithoutDeduplicating() {
         assertEquals(3, recovered.getOrAddSymbol("brand-new"));
     }
 
+    @Test
+    public void testAddRecoveredSymbol_rejectsNullWithoutMutation() {
+        GlobalSymbolDictionary dict = new GlobalSymbolDictionary();
+        assertEquals(0, dict.addRecoveredSymbol("AAPL"));
+
+        try {
+            dict.addRecoveredSymbol(null);
+            fail("expected IllegalArgumentException");
+        } catch (IllegalArgumentException expected) {
+            assertEquals("symbol cannot be null", expected.getMessage());
+        }
+
+        assertEquals(1, dict.size());
+        assertEquals("AAPL", dict.getSymbol(0));
+        assertEquals(0, dict.getId("AAPL"));
+    }
+
     @Test
     public void testAddSymbol_assignsSequentialIds() {
         GlobalSymbolDictionary dict = new GlobalSymbolDictionary();

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

@@ -222,6 +222,49 @@ public void testBuilderWithWebSocketTransportCreatesCorrectSenderType() throws E
         });
     }
 
+    @Test
+    public void testCatchUpCapGapMinEscalationWindowBuilderNegativeRejected() {
+        assertThrows("catchup_cap_gap_min_escalation_window_millis must be >= 0: -1",
+                () -> Sender.builder(Sender.Transport.WEBSOCKET)
+                        .address(LOCALHOST)
+                        .catchUpCapGapMinEscalationWindowMillis(-1));
+    }
+
+    @Test
+    public void testCatchUpCapGapMinEscalationWindowBuilderNotSupportedForTcp() {
+        assertThrows("catchup_cap_gap_min_escalation_window_millis is only supported for WebSocket transport",
+                () -> Sender.builder(Sender.Transport.TCP)
+                        .address(LOCALHOST)
+                        .catchUpCapGapMinEscalationWindowMillis(300_000));
+    }
+
+    @Test
+    public void testCatchUpCapGapMinEscalationWindowConfigStringNegativeRejected() {
+        assertThrows("catchup_cap_gap_min_escalation_window_millis must be >= 0: -1",
+                () -> Sender.builder("ws::addr=localhost:9000;"
+                        + "catchup_cap_gap_min_escalation_window_millis=-1;"));
+    }
+
+    @Test
+    public void testCatchUpCapGapMinEscalationWindowConfigStringNotSupportedForTcp() {
+        assertThrows("catchup_cap_gap_min_escalation_window_millis is only supported for WebSocket transport",
+                () -> Sender.builder("tcp::addr=localhost:9009;"
+                        + "catchup_cap_gap_min_escalation_window_millis=300000;"));
+    }
+
+    @Test
+    public void testCatchUpCapGapMinEscalationWindowDefaultsToFiveMinutes() {
+        Assert.assertEquals(300_000L,
+                Sender.builder(Sender.Transport.WEBSOCKET)
+                        .address(LOCALHOST)
+                        .wsConfigSnapshotForTest()
+                        .get("catchup_cap_gap_min_escalation_window_millis"));
+        Assert.assertEquals(300_000L,
+                Sender.builder("ws::addr=localhost:9000;")
+                        .wsConfigSnapshotForTest()
+                        .get("catchup_cap_gap_min_escalation_window_millis"));
+    }
+
     @Test
     public void testConnectionRefused() throws Exception {
         assertMemoryLeak(() -> {

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

@@ -273,6 +273,43 @@ public void testQwpBufferWriterUtf8LengthInvalidSurrogatePair() throws Exception
         });
     }
 
+    @Test
+    public void testRawVarintBoundariesAndReturnedAddress() throws Exception {
+        assertMemoryLeak(() -> {
+            try (NativeBufferWriter writer = new NativeBufferWriter(16)) {
+                assertRawVarint(writer, 0, 0x00);
+                assertRawVarint(writer, 1, 0x01);
+                assertRawVarint(writer, 127, 0x7F);
+                assertRawVarint(writer, 128, 0x80, 0x01);
+                assertRawVarint(writer, 16_383, 0xFF, 0x7F);
+                assertRawVarint(writer, 16_384, 0x80, 0x80, 0x01);
+                assertRawVarint(writer, Long.MAX_VALUE,
+                        0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0x7F);
+            }
+        });
+    }
+
+    @Test
+    public void testRawVarintRejectsNegativeWithoutWriting() throws Exception {
+        assertMemoryLeak(() -> {
+            try (NativeBufferWriter writer = new NativeBufferWriter(16)) {
+                long addr = writer.getBufferPtr();
+                Unsafe.getUnsafe().putByte(addr, (byte) 0x5A);
+                boolean rejected = false;
+                try {
+                    NativeBufferWriter.writeVarint(addr, -1);
+                } catch (AssertionError expected) {
+                    rejected = true;
+                    assertTrue(expected.getMessage().contains(
+                            "unsigned LEB128 varint requires a non-negative value: -1"));
+                }
+                assertTrue("negative raw varint must be rejected", rejected);
+                assertEquals("rejection must happen before the first write",
+                        (byte) 0x5A, Unsafe.getUnsafe().getByte(addr));
+            }
+        });
+    }
+
     @Test
     public void testReset() throws Exception {
         assertMemoryLeak(() -> {
@@ -618,4 +655,21 @@ public void testWriteVarintSmall() throws Exception {
             }
         });
     }
+
+    private static void assertRawVarint(NativeBufferWriter writer, long value, int... expectedBytes) {
+        long addr = writer.getBufferPtr();
+        for (int i = 0; i <= expectedBytes.length; i++) {
+            Unsafe.getUnsafe().putByte(addr + i, (byte) 0x5A);
+        }
+        long end = NativeBufferWriter.writeVarint(addr, value);
+        assertEquals("returned address for value=" + value, addr + expectedBytes.length, end);
+        assertEquals("varintSize for value=" + value,
+                expectedBytes.length, NativeBufferWriter.varintSize(value));
+        for (int i = 0; i < expectedBytes.length; i++) {
+            assertEquals("byte " + i + " for value=" + value,
+                    (byte) expectedBytes[i], Unsafe.getUnsafe().getByte(addr + i));
+        }
+        assertEquals("writer overran value=" + value,
+                (byte) 0x5A, Unsafe.getUnsafe().getByte(end));
+    }
 }