Guard the sent-dict mirror against int overflow

The I/O thread's lifetime-monotonic symbol-dictionary mirror is sized with
int math: accumulateSentDict passed sentDictBytesLen + regionBytes (an int
sum) to ensureSentDictCapacity, and the grow step doubled capacity*2, also
int. On a pathological, very-high-cardinality connection the sum overflows
negative -- so the capacity check passes and copyMemory scribbles past the
buffer (silent heap corruption) -- and capacity*2 overflows negative near
1 GB, degrading the doubling to exact-fit reallocs. Reaching this needs
~200M+ distinct symbols on one connection, far past any real workload, but
the failure mode is silent corruption.

ensureSentDictCapacity now takes a long, the caller passes a long sum, and
the method throws a LineSenderException above an int-addressable ceiling
(Integer.MAX_VALUE - 8) instead of overflowing, growing in long math
clamped to that ceiling. Defensive only -- not reachable at realistic
symbol cardinality, so there is no scale test.

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

Author: glasstiger

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

@@ -133,6 +133,12 @@ public final class CursorWebSocketSendLoop implements QuietCloseable {
      */
     public static final long DEFAULT_POISON_MIN_ESCALATION_WINDOW_MILLIS = 5_000L;
     private static final Logger LOG = LoggerFactory.getLogger(CursorWebSocketSendLoop.class);
+    // Hard ceiling for the lifetime-monotonic sent-dictionary mirror. The mirror
+    // fields are int, so it cannot exceed Integer.MAX_VALUE bytes; reaching even
+    // this needs ~200M+ distinct symbols on a single connection, far past any real
+    // workload. The guard exists so that pathological growth fails loudly instead
+    // of overflowing the int capacity math into a heap-corrupting copyMemory.
+    private static final int MAX_SENT_DICT_BYTES = Integer.MAX_VALUE - 8;
     /**
      * Throttle "reconnect attempt N failed" WARN logs to one per 5 s.
      */
@@ -2035,19 +2041,32 @@ private void accumulateSentDict(long payloadAddr, int payloadLen) {
             }
         }
         int regionBytes = (int) (p - regionStart);
-        ensureSentDictCapacity(sentDictBytesLen + regionBytes);
+        // long sum: sentDictBytesLen + regionBytes can exceed Integer.MAX_VALUE on
+        // a very-high-cardinality lifetime connection; ensureSentDictCapacity then
+        // fails loudly rather than overflowing to a negative int (which would make
+        // the capacity check pass and copyMemory scribble past the buffer).
+        ensureSentDictCapacity((long) sentDictBytesLen + regionBytes);
         Unsafe.getUnsafe().copyMemory(regionStart, sentDictBytesAddr + sentDictBytesLen, regionBytes);
         sentDictBytesLen += regionBytes;
         sentDictCount += (int) deltaCount;
     }
 
-    private void ensureSentDictCapacity(int required) {
+    private void ensureSentDictCapacity(long required) {
         if (sentDictBytesCapacity >= required) {
             return;
         }
-        int newCap = Math.max(sentDictBytesCapacity * 2, Math.max(4096, required));
-        sentDictBytesAddr = Unsafe.realloc(sentDictBytesAddr, sentDictBytesCapacity, newCap, MemoryTag.NATIVE_DEFAULT);
-        sentDictBytesCapacity = newCap;
+        if (required > MAX_SENT_DICT_BYTES) {
+            throw new LineSenderException("symbol dictionary mirror exceeds the maximum size ["
+                    + "required=" + required + ", max=" + MAX_SENT_DICT_BYTES + ']');
+        }
+        // Grow in long to avoid the capacity*2 int overflow (negative) that would
+        // otherwise degrade the doubling near 1 GB; clamp to the int ceiling.
+        long newCap = Math.max((long) sentDictBytesCapacity * 2, Math.max(4096L, required));
+        if (newCap > MAX_SENT_DICT_BYTES) {
+            newCap = MAX_SENT_DICT_BYTES;
+        }
+        sentDictBytesAddr = Unsafe.realloc(sentDictBytesAddr, sentDictBytesCapacity, (int) newCap, MemoryTag.NATIVE_DEFAULT);
+        sentDictBytesCapacity = (int) newCap;
     }
 
     private long readVarintAt(long p, long limit) {