Snapshot server batch cap once in sendRow

sendRow read the volatile serverMaxBatchSize three times while
checking one row against the cap. The I/O thread lowers that field
mid-stream on a failover -- clearing it to 0 when the new node
advertises no cap -- so a read torn across the guard and the throw
could reject the row against a "cap" of 0, which means "no cap": the
row rolled back and a retry succeeded, but the producer still saw a
spurious "row too large for server batch cap" error.

Snapshot the field into a local once, exactly as flushPendingRows
already does, so the guard, the comparison and the message all use
one consistent value.

Also add QwpWebSocketSenderTest coverage for the per-row cap guard,
which had none: an oversized row under a positive cap throws and
rolls back to zero rows, and a large row commits when the server
advertises no cap (the invariant the snapshot protects).

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

Author: glasstiger

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

@@ -4155,12 +4155,18 @@ private void sendRow() {
         // batch stay intact. (The check ignores the null-padding bytes
         // nextRow() will add; that's bounded by numColumns * elemSize and
         // far below any realistic cap.)
-        if (serverMaxBatchSize > 0) {
+        // Snapshot the volatile cap ONCE, as flushPendingRows does. The I/O thread
+        // lowers serverMaxBatchSize -- or clears it to 0 on a failover to a node that
+        // advertises no cap -- mid-stream via applyServerBatchSizeLimit. Re-reading the
+        // field across the guard and the throw could observe it drop to 0 between reads
+        // and reject the row against a "cap" of 0, which actually means "no cap".
+        int cap = serverMaxBatchSize;
+        if (cap > 0) {
             long rowBytes = currentTableBuffer.getBufferedBytes() - currentTableBufferSnapshotBytes;
-            if (rowBytes > serverMaxBatchSize) {
+            if (rowBytes > cap) {
                 throw new LineSenderException("row too large for server batch cap")
                         .put(" [rowBytes=").put(rowBytes)
-                        .put(", serverMaxBatchSize=").put(serverMaxBatchSize).put(']');
+                        .put(", serverMaxBatchSize=").put(cap).put(']');
             }
         }
 

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

@@ -624,6 +624,77 @@ public void testResetAfterCloseThrows() throws Exception {
         });
     }
 
+    @Test
+    public void testRowAcceptedWhenServerAdvertisesNoCap() throws Exception {
+        // serverMaxBatchSize == 0 means "no cap" (an older server, or a failover to a
+        // node that advertises none). The per-row guard in sendRow must be skipped, so
+        // even a large row is accepted rather than rejected against a "cap" of 0. This
+        // is the invariant the sendRow cap snapshot protects: the I/O thread can lower
+        // the volatile serverMaxBatchSize to 0 mid-row, and a torn read that observed
+        // the drop between the guard and the throw must not spuriously reject the row.
+        assertMemoryLeak(() -> {
+            try (QwpWebSocketSender sender = QwpWebSocketSender.createForTesting(
+                    "localhost", 9000,
+                    /*autoFlushRows*/ Integer.MAX_VALUE,
+                    /*autoFlushBytes*/ 0,
+                    /*autoFlushIntervalNanos*/ 0L)) {
+                sender.setConnectedForTest(true);
+                sender.applyServerBatchSizeLimit(0);
+                Assert.assertEquals(0, sender.getServerMaxBatchSize());
+
+                StringBuilder big = new StringBuilder();
+                for (int i = 0; i < 256; i++) {
+                    big.append('x');
+                }
+                // A row far larger than any small cap still commits when the server
+                // advertises no cap; the guard leaves it alone.
+                sender.table("t").stringColumn("s", big.toString()).atNow();
+
+                QwpTableBuffer buf = sender.getTableBuffer("t");
+                Assert.assertEquals(1, buf.getRowCount());
+            }
+        });
+    }
+
+    @Test
+    public void testRowExceedingServerBatchCapThrows() throws Exception {
+        // The per-row guard in sendRow rejects a single row whose encoded bytes
+        // already exceed the server's advertised cap, before nextRow() commits it, so
+        // the flush cannot build an oversize WS frame the server closes with 1009.
+        assertMemoryLeak(() -> {
+            try (QwpWebSocketSender sender = QwpWebSocketSender.createForTesting(
+                    "localhost", 9000,
+                    /*autoFlushRows*/ Integer.MAX_VALUE,
+                    /*autoFlushBytes*/ 0,
+                    /*autoFlushIntervalNanos*/ 0L)) {
+                sender.setConnectedForTest(true);
+                sender.applyServerBatchSizeLimit(64);
+                Assert.assertEquals(64, sender.getServerMaxBatchSize());
+
+                StringBuilder big = new StringBuilder();
+                for (int i = 0; i < 256; i++) {
+                    big.append('x');
+                }
+                sender.table("t").stringColumn("s", big.toString());
+                try {
+                    sender.atNow();
+                    Assert.fail("Expected LineSenderException");
+                } catch (LineSenderException e) {
+                    Assert.assertTrue(
+                            "expected 'row too large for server batch cap', got: " + e.getMessage(),
+                            e.getMessage().contains("row too large for server batch cap"));
+                    // The message reports the single snapshotted cap, not a torn value.
+                    Assert.assertTrue(
+                            "expected the snapshotted cap in the message, got: " + e.getMessage(),
+                            e.getMessage().contains("serverMaxBatchSize=64"));
+                }
+                // The rejected row was rolled back before commit, leaving no row behind.
+                QwpTableBuffer buf = sender.getTableBuffer("t");
+                Assert.assertEquals(0, buf.getRowCount());
+            }
+        });
+    }
+
     @Test
     public void testStringColumnAfterCloseThrows() throws Exception {
         assertMemoryLeak(() -> {