Harden delta dictionary test correctness

glasstiger · glasstiger · commit f3282a49563c · 2026-07-15T22:30:57.000+01:00
diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/DeltaDictCatchUpTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/DeltaDictCatchUpTest.java
@@ -97,6 +97,7 @@ public void testReconnectCatchUpRebuildsDictionary() throws Exception {
                 Assert.assertEquals("2nd connection dictionary size", 2, conn2.size());
                 Assert.assertEquals("alpha", conn2.get(0));
                 Assert.assertEquals("beta", conn2.get(1));
+                assertAckSequencesStartAtZero(handler.ackSequenceStarts());
             }
         });
     }
@@ -143,6 +144,7 @@ public void testReconnectPreservesMonotonicDeltaBaseline() throws Exception {
                                 + "baseline (deltaStart >= 1); a reset baseline re-ships the whole "
                                 + "dictionary from deltaStart 0",
                         handler.conn2SawDeltaAboveBaseline);
+                assertAckSequencesStartAtZero(handler.ackSequenceStarts());
             }
         });
     }
@@ -204,6 +206,7 @@ public void testFixedCapNearBoundarySymbolCatchesUpWithoutTerminal() throws Exce
                 Assert.assertEquals("2nd connection dictionary size", 2, conn2.size());
                 Assert.assertEquals(nearCapSymbol, conn2.get(0));
                 Assert.assertEquals("beta", conn2.get(1));
+                assertAckSequencesStartAtZero(handler.ackSequenceStarts());
             }
         });
     }
@@ -251,6 +254,7 @@ public void testForegroundCatchUpCapGapRetriesPastOrphanBudgetAndRecovers() thro
                     Assert.assertTrue("foreground sender must recover and drain after the cap is restored",
                             sender.awaitAckedFsn(targetFsn, 5_000));
                 }
+                assertAckSequencesStartAtZero(handler.ackSequenceStarts());
             }
         });
     }
@@ -307,10 +311,19 @@ public void testReconnectCatchUpSplitsLargeDictionaryAcrossFrames() throws Excep
                 Assert.assertTrue("catch-up split into multiple frames (saw "
                                 + handler.zeroTableFramesOnConn2 + ")",
                         handler.zeroTableFramesOnConn2 >= 2);
+                assertAckSequencesStartAtZero(handler.ackSequenceStarts());
             }
         });
     }
 
+    private static void assertAckSequencesStartAtZero(List<Long> ackSequenceStarts) {
+        Assert.assertFalse("the fixture must ACK at least one connection", ackSequenceStarts.isEmpty());
+        for (int i = 0; i < ackSequenceStarts.size(); i++) {
+            Assert.assertEquals("connection " + (i + 1) + " must begin ACK sequencing at zero",
+                    0L, ackSequenceStarts.get(i).longValue());
+        }
+    }
+
     private static String symbolName(int i) {
         // 10-char symbols so 40 of them clearly exceed the advertised 160-byte cap.
         return "symbol" + (1000 + i);
@@ -371,10 +384,15 @@ private static class CatchUpHandler implements TestWebSocketServer.WebSocketServ
         // post-reconnect frame trips it.
         volatile boolean conn2SawDeltaAboveBaseline;
         volatile boolean sawZeroTableFrameOnConn2;
+        private final List<Long> ackSequenceStarts = new CopyOnWriteArrayList<>();
         private final List<List<String>> dictsByConn = new CopyOnWriteArrayList<>();
         private TestWebSocketServer.ClientHandler currentClient;
         private final AtomicLong nextSeq = new AtomicLong(0);
 
+        List<Long> ackSequenceStarts() {
+            return new ArrayList<>(ackSequenceStarts);
+        }
+
         synchronized List<String> dictFor(int connNumber) {
             return connNumber <= dictsByConn.size()
                     // Copy under the lock: the caller iterates it unlocked while the
@@ -390,6 +408,7 @@ public synchronized void onBinaryMessage(TestWebSocketServer.ClientHandler clien
                 currentClient = client;
                 connectionsAccepted.incrementAndGet();
                 dictsByConn.add(new ArrayList<>()); // fresh dictionary per connection
+                nextSeq.set(0);
             }
             int connNumber = dictsByConn.size();
             List<String> dict = dictsByConn.get(connNumber - 1);
@@ -414,7 +433,11 @@ public synchronized void onBinaryMessage(TestWebSocketServer.ClientHandler clien
                 }
             }
             try {
-                client.sendBinary(buildAck(nextSeq.getAndIncrement()));
+                long ackSequence = nextSeq.getAndIncrement();
+                if (newConnection) {
+                    ackSequenceStarts.add(ackSequence);
+                }
+                client.sendBinary(buildAck(ackSequence));
                 // Drop the first connection right after ACKing its only frame,
                 // forcing the sender to reconnect onto a fresh dictionary.
                 if (connNumber == 1) {
@@ -474,23 +497,33 @@ private static int tableCount(byte[] frame) {
      */
     private static class CapShrinkHandler implements TestWebSocketServer.WebSocketServerHandler {
         final AtomicInteger connectionsAccepted = new AtomicInteger();
+        private final List<Long> ackSequenceStarts = new CopyOnWriteArrayList<>();
         private final AtomicLong nextSeq = new AtomicLong(0);
         private TestWebSocketServer.ClientHandler currentClient;
         private volatile TestWebSocketServer server;
 
+        List<Long> ackSequenceStarts() {
+            return new ArrayList<>(ackSequenceStarts);
+        }
+
         void setServer(TestWebSocketServer server) {
             this.server = server;
         }
 
         @Override
         public synchronized void onBinaryMessage(TestWebSocketServer.ClientHandler client, byte[] data) {
-            if (currentClient != client) {
+            boolean newConnection = currentClient != client;
+            if (newConnection) {
                 currentClient = client;
                 connectionsAccepted.incrementAndGet();
                 nextSeq.set(0);
             }
             try {
-                client.sendBinary(CatchUpHandler.buildAck(nextSeq.getAndIncrement()));
+                long ackSequence = nextSeq.getAndIncrement();
+                if (newConnection) {
+                    ackSequenceStarts.add(ackSequence);
+                }
+                client.sendBinary(CatchUpHandler.buildAck(ackSequence));
                 if (connectionsAccepted.get() == 1) {
                     // Connection 1 registered the big symbol. Shrink the cap so the
                     // reconnect's catch-up budget can't fit it, then drop to force
@@ -518,6 +551,7 @@ private static class SplitCatchUpHandler implements TestWebSocketServer.WebSocke
         // Set once the server has closed connection 1 (see CatchUpHandler.conn1Closed).
         volatile boolean conn1Closed;
         volatile int zeroTableFramesOnConn2;
+        private final List<Long> ackSequenceStarts = new CopyOnWriteArrayList<>();
         private final List<List<String>> dictsByConn = new CopyOnWriteArrayList<>();
         private final int dropConn1AtDictSize;
         private final AtomicLong nextSeq = new AtomicLong(0);
@@ -528,6 +562,10 @@ private static class SplitCatchUpHandler implements TestWebSocketServer.WebSocke
             this.dropConn1AtDictSize = dropConn1AtDictSize;
         }
 
+        List<Long> ackSequenceStarts() {
+            return new ArrayList<>(ackSequenceStarts);
+        }
+
         synchronized List<String> dictFor(int connNumber) {
             return connNumber <= dictsByConn.size()
                     // Copy under the lock: the caller iterates it unlocked while the
@@ -543,6 +581,7 @@ public synchronized void onBinaryMessage(TestWebSocketServer.ClientHandler clien
                 currentClient = client;
                 connectionsAccepted.incrementAndGet();
                 dictsByConn.add(new ArrayList<>()); // fresh dictionary per connection
+                nextSeq.set(0);
             }
             int connNumber = dictsByConn.size();
             List<String> dict = dictsByConn.get(connNumber - 1);
@@ -551,7 +590,11 @@ public synchronized void onBinaryMessage(TestWebSocketServer.ClientHandler clien
                 zeroTableFramesOnConn2++;
             }
             try {
-                client.sendBinary(CatchUpHandler.buildAck(nextSeq.getAndIncrement()));
+                long ackSequence = nextSeq.getAndIncrement();
+                if (newConnection) {
+                    ackSequenceStarts.add(ackSequence);
+                }
+                client.sendBinary(CatchUpHandler.buildAck(ackSequence));
                 // Drop connection 1 only once it has learned the entire batch, so
                 // the sender's sent-dictionary mirror is complete and the reconnect
                 // catch-up must replay a dictionary larger than the batch cap.
diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/DeltaDictRecoveryTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/DeltaDictRecoveryTest.java
@@ -789,8 +789,9 @@ public void testPersistFailureSurfacesAsLineSenderException() throws Exception {
                 CursorSendEngine engine = new CursorSendEngine(
                         slot, 4L * 1024 * 1024, CursorSendEngine.DEFAULT_APPEND_DEADLINE_NANOS,
                         CursorSendEngine.DEFAULT_APPEND_DEADLINE_NANOS, ff);
-                try (Sender sender = QwpWebSocketSender.connect(
-                        "localhost", port, null, 0, 0, 0L, null, false, engine)) {
+                Sender sender = QwpWebSocketSender.connect(
+                        "localhost", port, null, 0, 0, 0L, null, false, engine);
+                try {
                     ff.armed = true; // the next dictionary append short-writes
                     sender.table("m").symbol("s", "boom").longColumn("v", 1L).atNow();
                     try {
@@ -801,14 +802,42 @@ public void testPersistFailureSurfacesAsLineSenderException() throws Exception {
                                         + "not leak a raw IllegalStateException: " + expected.getMessage(),
                                 expected.getMessage().contains("failed to persist symbol dictionary before publish"));
                     }
-                } catch (LineSenderException ignored) {
-                    // close() re-flushes the still-buffered row; the facade has disarmed, so
-                    // this normally succeeds. Either way it is not what we assert.
+                } finally {
+                    try {
+                        sender.close();
+                    } catch (LineSenderException ignored) {
+                        // close() re-flushes the still-buffered row; the facade has disarmed, so
+                        // this normally succeeds. Either way it is not what we assert.
+                    }
                 }
             }
         });
     }
 
+    @Test
+    public void testReconstructingFixtureResetsAckSequencePerConnection() throws Exception {
+        assertMemoryLeak(() -> {
+            DictReconstructingHandler handler = new DictReconstructingHandler();
+            try (TestWebSocketServer server = new TestWebSocketServer(handler)) {
+                int port = server.getPort();
+                server.start();
+                Assert.assertTrue(server.awaitStart(5, TimeUnit.SECONDS));
+
+                for (int i = 0; i < 2; i++) {
+                    try (Sender sender = Sender.fromConfig("ws::addr=localhost:" + port + ";")) {
+                        sender.table("m").symbol("s", "sym-" + i).longColumn("v", i).atNow();
+                        long fsn = sender.flushAndGetSequence();
+                        Assert.assertTrue("connection " + (i + 1) + " must receive ACK zero",
+                                sender.awaitAckedFsn(fsn, 5_000));
+                    }
+                }
+
+                Assert.assertEquals("each fresh connection must restart wire ACK sequencing",
+                        Arrays.asList(0L, 0L), handler.ackSequenceStarts());
+            }
+        });
+    }
+
     @Test
     public void testFailedPublishDoesNotDuplicatePersistedSymbols() throws Exception {
         // Regression: persistNewSymbolsBeforePublish is a write-ahead -- it runs
@@ -1385,10 +1414,15 @@ private static int readVarint(byte[] buf, int[] pos) {
      */
     private static class DictReconstructingHandler implements TestWebSocketServer.WebSocketServerHandler {
         volatile boolean sawCatchUpFrame;
+        private final List<Long> ackSequenceStarts = new ArrayList<>();
         private final List<String> dict = new ArrayList<>();
         private final AtomicLong nextSeq = new AtomicLong(0);
         private TestWebSocketServer.ClientHandler currentClient;
 
+        synchronized List<Long> ackSequenceStarts() {
+            return new ArrayList<>(ackSequenceStarts);
+        }
+
         synchronized List<String> dictSnapshot() {
             return new ArrayList<>(dict);
         }
@@ -1399,16 +1433,22 @@ synchronized int maxDictSize() {
 
         @Override
         public synchronized void onBinaryMessage(TestWebSocketServer.ClientHandler client, byte[] data) {
-            if (currentClient != client) {
+            boolean newConnection = currentClient != client;
+            if (newConnection) {
                 currentClient = client;
                 dict.clear(); // fresh server dictionary per connection
+                nextSeq.set(0);
             }
             accumulate(data);
             if (tableCount(data) == 0 && hasDelta(data)) {
                 sawCatchUpFrame = true;
             }
             try {
-                client.sendBinary(buildAck(nextSeq.getAndIncrement()));
+                long ackSequence = nextSeq.getAndIncrement();
+                if (newConnection) {
+                    ackSequenceStarts.add(ackSequence);
+                }
+                client.sendBinary(buildAck(ackSequence));
             } catch (IOException e) {
                 throw new RuntimeException(e);
             }
