Harden QWP client decoder, binds, and close paths

Tier 1 findings from the QWP client review, consolidated into one
change so client and server wire formats stay aligned.

Decoder (QwpResultBatchDecoder)
- Non-delta parseSymbolColumn now rejects a dict size that is negative
  or exceeds rowCount, and rejects an entry length that is negative or
  above Integer.MAX_VALUE. Previously dictSize*8 could overflow int and
  bypass ensureOwnedEntriesAddr; a signed-wrap entryLen could slip the
  p + entryLen > limit check and advance p backwards through already-
  consumed bytes.
- parseDeltaSymbolDict caps entryLen at Integer.MAX_VALUE before the
  int cast, computes newHeapPos in long space, and rejects growth
  beyond MAX_CONN_DICT_HEAP_BYTES. connDictSize is capped at
  MAX_CONN_DICT_SIZE. These guards keep a server that fails to emit
  CACHE_RESET from wrapping connDictHeapPos negative and letting
  copyMemory write past the heap.
- GEOHASH decode rejects precisionBits outside [1, 60], mirroring the
  server's own range check.

Bind encoder (QwpBindValues)
- checkScale split into per-width variants (DECIMAL64 <= 18,
  DECIMAL128 <= 38, DECIMAL256 <= 76) so out-of-range scales are
  rejected at the type boundary instead of silently passing through
  under the shared MAX_SCALE cap.
- setNull for DECIMAL64/128/256 emits the trailing scale byte and
  setNull for GEOHASH emits the precision_bits varint. The server
  reads both fields unconditionally before checking the null flag, so
  omitting them misframed every bind that followed in the batch.
- setDecimal128(Decimal128) and setDecimal256(Decimal256) now preserve
  value.getScale() on the null-sentinel path.
- New explicit helpers setNullDecimal64/128/256(int, int scale) and
  setNullGeohash(int, int precisionBits) let callers pin the type's
  scale/precision without a non-null value.
- setGeohash masks value to precisionBits before emitting, so bits
  above the declared precision cannot leak into the top wire byte
  when precisionBits is not a multiple of 8.

Buffer (QwpBatchBuffer)
- ensureCapacity rejects a negative required length, starts doubling
  at max(current, 1) so a zero initial capacity can still grow, and
  computes the doubling in long space with a clamp at
  Integer.MAX_VALUE. The old int doubling spun forever at 0 and
  wrapped negative above 2^30.

Concurrency (QwpQueryClient, QwpEgressIoThread)
- close() is gated by an AtomicBoolean so concurrent or repeat calls
  no longer walk the shutdown body twice.
- GenerationListener now owns its own AtomicReference<TerminalFailure>
  instead of sharing one across generations. An orphaned-but-in-flight
  listener's compareAndSet lands on a ref nobody reads, so a late
  callback from a dying I/O thread cannot poison the next connection.
- releaseBuffer re-checks closed after the freeBuffers.offer and
  removes the buffer back out if closePool raced it; the buffer is
  closed in place rather than stranded in a drained-and-abandoned
  pool. ArrayBlockingQueue.remove gives us the atomicity we need to
  avoid a double close when closePool's drain beat us to the same
  buffer.

Regression tests
- QwpBindEncoderTest: new coverage for per-width scale rejection
  (DECIMAL64 > 18, DECIMAL128 > 38, DECIMAL256 accepts 76), the four
  typed setNullDecimal*/setNullGeohash helpers, the geohash masking
  at sub-byte and 60-bit precisions, and updated expectations for
  the NULL path of the convenience Decimal128/256 overloads and the
  exhaustive null-types walk.
- QwpResultBatchDecoderHardeningTest: new frames for non-delta
  dict-size-above-rowCount, delta entry length > Integer.MAX_VALUE,
  and GEOHASH precision 0 / 61.
- QwpBatchBufferTest: new file covering grow-from-zero, reject-
  negative, and bounded-time growth.
- QwpQueryClientUnitTest: new concurrent-close stress test.
- QwpEgressIoThreadCloseRaceTest: new file; 200 iterations racing
  releaseBuffer against closePool under assertMemoryLeak so a
  stranded buffer surfaces as a native-memory leak.

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

Author: bluestreak01

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

@@ -93,10 +93,31 @@ public long getScratchAddr() {
     }
 
     private void ensureCapacity(int required) {
+        if (required < 0) {
+            // A negative request cannot be honoured. Reject loudly rather than
+            // silently wrapping through the doubling loop.
+            throw new IllegalArgumentException("QwpBatchBuffer required capacity must be non-negative: " + required);
+        }
         if (required <= scratchCapacity) return;
-        int newCap = scratchCapacity;
-        while (newCap < required) newCap *= 2;
-        scratchAddr = Unsafe.realloc(scratchAddr, scratchCapacity, newCap, MemoryTag.NATIVE_DEFAULT);
-        scratchCapacity = newCap;
+        // Start the doubling at max(current, 1) so a buffer constructed with
+        // initialCapacity=0 can still grow (0 *= 2 would spin forever). Cap the
+        // double step against Integer.MAX_VALUE because `newCap *= 2` wraps
+        // negative once newCap passes 2^30, at which point the while loop
+        // could never reach `required` and would spin indefinitely.
+        long newCap = Math.max(scratchCapacity, 1);
+        while (newCap < required) {
+            newCap <<= 1;
+            if (newCap > Integer.MAX_VALUE) {
+                newCap = Integer.MAX_VALUE;
+                break;
+            }
+        }
+        if (newCap < required) {
+            throw new OutOfMemoryError("QwpBatchBuffer required capacity " + required
+                    + " exceeds Integer.MAX_VALUE");
+        }
+        int capped = (int) newCap;
+        scratchAddr = Unsafe.realloc(scratchAddr, scratchCapacity, capped, MemoryTag.NATIVE_DEFAULT);
+        scratchCapacity = capped;
     }
 }

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

@@ -60,6 +60,22 @@
  */
 public final class QwpBindValues implements QuietCloseable {
 
+    /**
+     * Maximum scale for a DECIMAL128 bind. DECIMAL128 stores up to 38 digits of
+     * precision, so scale above 38 is mathematically invalid even though the
+     * wire format can carry any byte value.
+     */
+    private static final int DECIMAL128_MAX_SCALE = 38;
+    /**
+     * Maximum scale for a DECIMAL256 bind. DECIMAL256 is the widest decimal
+     * type and its scale cap matches {@link Decimals#MAX_SCALE} (76).
+     */
+    private static final int DECIMAL256_MAX_SCALE = Decimals.MAX_SCALE;
+    /**
+     * Maximum scale for a DECIMAL64 bind. DECIMAL64 stores up to 18 digits of
+     * precision, so scale above 18 is mathematically invalid.
+     */
+    private static final int DECIMAL64_MAX_SCALE = 18;
     /**
      * Maximum GEOHASH precision in bits, matching the server's
      * {@code ColumnType.GEOLONG_MAX_BITS} check.
@@ -108,7 +124,7 @@ public QwpBindValues setDate(int index, long millisSinceEpoch) {
     }
 
     public QwpBindValues setDecimal128(int index, int scale, long lo, long hi) {
-        checkScale(scale);
+        checkScale128(scale);
         advance(index);
         writeHeader(QwpConstants.TYPE_DECIMAL128, false);
         writer.putByte((byte) scale);
@@ -120,18 +136,21 @@ public QwpBindValues setDecimal128(int index, int scale, long lo, long hi) {
     /**
      * Convenience overload that takes the scale and limbs from the supplied
      * {@link Decimal128}. If the value is NULL (Decimal128's canonical NULL
-     * sentinel), an explicit DECIMAL128 NULL is encoded; otherwise the scale
-     * and 128-bit unscaled value are encoded as-is.
+     * sentinel), an explicit DECIMAL128 NULL is encoded preserving
+     * {@code value.getScale()}; a {@code null} reference encodes with scale 0.
      */
     public QwpBindValues setDecimal128(int index, Decimal128 value) {
-        if (value == null || Decimal128.isNull(value.getHigh(), value.getLow())) {
-            return setNull(index, QwpConstants.TYPE_DECIMAL128);
+        if (value == null) {
+            return setNullDecimal128(index, 0);
+        }
+        if (Decimal128.isNull(value.getHigh(), value.getLow())) {
+            return setNullDecimal128(index, value.getScale());
         }
         return setDecimal128(index, value.getScale(), value.getLow(), value.getHigh());
     }
 
     public QwpBindValues setDecimal256(int index, int scale, long ll, long lh, long hl, long hh) {
-        checkScale(scale);
+        checkScale256(scale);
         advance(index);
         writeHeader(QwpConstants.TYPE_DECIMAL256, false);
         writer.putByte((byte) scale);
@@ -145,17 +164,21 @@ public QwpBindValues setDecimal256(int index, int scale, long ll, long lh, long
     /**
      * Convenience overload that takes the scale and limbs from the supplied
      * {@link Decimal256}. If the value is NULL (Decimal256's canonical NULL
-     * sentinel), an explicit DECIMAL256 NULL is encoded.
+     * sentinel), an explicit DECIMAL256 NULL is encoded preserving
+     * {@code value.getScale()}; a {@code null} reference encodes with scale 0.
      */
     public QwpBindValues setDecimal256(int index, Decimal256 value) {
-        if (value == null || Decimal256.isNull(value.getHh(), value.getHl(), value.getLh(), value.getLl())) {
-            return setNull(index, QwpConstants.TYPE_DECIMAL256);
+        if (value == null) {
+            return setNullDecimal256(index, 0);
+        }
+        if (Decimal256.isNull(value.getHh(), value.getHl(), value.getLh(), value.getLl())) {
+            return setNullDecimal256(index, value.getScale());
         }
         return setDecimal256(index, value.getScale(), value.getLl(), value.getLh(), value.getHl(), value.getHh());
     }
 
     public QwpBindValues setDecimal64(int index, int scale, long unscaledValue) {
-        checkScale(scale);
+        checkScale64(scale);
         advance(index);
         writeHeader(QwpConstants.TYPE_DECIMAL64, false);
         writer.putByte((byte) scale);
@@ -181,19 +204,21 @@ public QwpBindValues setFloat(int index, float value) {
      * Encodes a GEOHASH bind with the given precision (in bits) and packed
      * value. The value is stored little-endian in {@code ceil(precisionBits / 8)}
      * bytes. Precision must be in {@code [1, 60]}.
+     * <p>
+     * {@code value} is masked to {@code precisionBits} before encoding, so bits
+     * above the declared precision cannot leak into the top wire byte (which
+     * would otherwise pass through when {@code precisionBits} is not a multiple
+     * of 8).
      */
     public QwpBindValues setGeohash(int index, int precisionBits, long value) {
-        if (precisionBits < GEOHASH_MIN_BITS || precisionBits > GEOHASH_MAX_BITS) {
-            throw new IllegalArgumentException(
-                    "GEOHASH precision must be in [" + GEOHASH_MIN_BITS + ", " + GEOHASH_MAX_BITS
-                            + "], got " + precisionBits);
-        }
+        checkGeohashPrecision(precisionBits);
         advance(index);
         writeHeader(QwpConstants.TYPE_GEOHASH, false);
         writer.putVarint(precisionBits);
+        long masked = maskGeohashBits(value, precisionBits);
         int byteCount = (precisionBits + 7) >>> 3;
         for (int b = 0; b < byteCount; b++) {
-            writer.putByte((byte) (value >>> (b * 8)));
+            writer.putByte((byte) (masked >>> (b * 8)));
         }
         return this;
     }
@@ -226,14 +251,83 @@ public QwpBindValues setLong256(int index, long l0, long l1, long l2, long l3) {
      * Binds an explicit NULL with the given QWP wire type. The type code must
      * be one of the supported scalar bind types; ARRAY, BINARY, and IPv4 are
      * rejected because the server decoder does not accept them as binds.
+     * <p>
+     * DECIMAL64/128/256 NULLs are encoded with scale 0 and GEOHASH NULLs with
+     * precision {@value #GEOHASH_MIN_BITS} bit; use {@link #setNullDecimal64},
+     * {@link #setNullDecimal128}, {@link #setNullDecimal256}, or
+     * {@link #setNullGeohash} when the scale/precision matters (it becomes part
+     * of the bound variable's type on the server).
      */
     public QwpBindValues setNull(int index, byte qwpTypeCode) {
         checkBindType(qwpTypeCode);
+        if (qwpTypeCode == QwpConstants.TYPE_DECIMAL64) {
+            return setNullDecimal64(index, 0);
+        }
+        if (qwpTypeCode == QwpConstants.TYPE_DECIMAL128) {
+            return setNullDecimal128(index, 0);
+        }
+        if (qwpTypeCode == QwpConstants.TYPE_DECIMAL256) {
+            return setNullDecimal256(index, 0);
+        }
+        if (qwpTypeCode == QwpConstants.TYPE_GEOHASH) {
+            return setNullGeohash(index, GEOHASH_MIN_BITS);
+        }
         advance(index);
         writeHeader(qwpTypeCode, true);
         return this;
     }
 
+    /**
+     * Binds an explicit NULL with DECIMAL128 type and the given scale. The
+     * server reads the scale byte regardless of null, so the scale must be
+     * supplied even for NULL (it becomes part of the bound variable's type).
+     */
+    public QwpBindValues setNullDecimal128(int index, int scale) {
+        checkScale128(scale);
+        advance(index);
+        writeHeader(QwpConstants.TYPE_DECIMAL128, true);
+        writer.putByte((byte) scale);
+        return this;
+    }
+
+    /**
+     * Binds an explicit NULL with DECIMAL256 type and the given scale. See
+     * {@link #setNullDecimal128} for the rationale.
+     */
+    public QwpBindValues setNullDecimal256(int index, int scale) {
+        checkScale256(scale);
+        advance(index);
+        writeHeader(QwpConstants.TYPE_DECIMAL256, true);
+        writer.putByte((byte) scale);
+        return this;
+    }
+
+    /**
+     * Binds an explicit NULL with DECIMAL64 type and the given scale. See
+     * {@link #setNullDecimal128} for the rationale.
+     */
+    public QwpBindValues setNullDecimal64(int index, int scale) {
+        checkScale64(scale);
+        advance(index);
+        writeHeader(QwpConstants.TYPE_DECIMAL64, true);
+        writer.putByte((byte) scale);
+        return this;
+    }
+
+    /**
+     * Binds an explicit NULL with GEOHASH type and the given precision (bits).
+     * The server reads the precision_bits varint regardless of null, so
+     * precision must be supplied even for NULL (it becomes part of the bound
+     * variable's type).
+     */
+    public QwpBindValues setNullGeohash(int index, int precisionBits) {
+        checkGeohashPrecision(precisionBits);
+        advance(index);
+        writeHeader(QwpConstants.TYPE_GEOHASH, true);
+        writer.putVarint(precisionBits);
+        return this;
+    }
+
     public QwpBindValues setShort(int index, short value) {
         advance(index);
         writeHeader(QwpConstants.TYPE_SHORT, false);
@@ -351,6 +445,14 @@ public void reset() {
         expectedNextIndex = 0;
     }
 
+    private static void checkGeohashPrecision(int precisionBits) {
+        if (precisionBits < GEOHASH_MIN_BITS || precisionBits > GEOHASH_MAX_BITS) {
+            throw new IllegalArgumentException(
+                    "GEOHASH precision must be in [" + GEOHASH_MIN_BITS + ", " + GEOHASH_MAX_BITS
+                            + "], got " + precisionBits);
+        }
+    }
+
     private static boolean isAscii(CharSequence value, int charLen) {
         for (int i = 0; i < charLen; i++) {
             if (value.charAt(i) >= 0x80) {
@@ -360,6 +462,10 @@ private static boolean isAscii(CharSequence value, int charLen) {
         return true;
     }
 
+    private static long maskGeohashBits(long value, int precisionBits) {
+        return precisionBits >= 64 ? value : value & ((1L << precisionBits) - 1L);
+    }
+
     private void advance(int index) {
         if (index != expectedNextIndex) {
             throw new IllegalStateException(
@@ -400,10 +506,24 @@ private void checkBindType(byte type) {
         }
     }
 
-    private void checkScale(int scale) {
-        if (scale < 0 || scale > Decimals.MAX_SCALE) {
+    private void checkScale128(int scale) {
+        if (scale < 0 || scale > DECIMAL128_MAX_SCALE) {
+            throw new IllegalArgumentException(
+                    "DECIMAL128 scale must be in [0, " + DECIMAL128_MAX_SCALE + "], got " + scale);
+        }
+    }
+
+    private void checkScale256(int scale) {
+        if (scale < 0 || scale > DECIMAL256_MAX_SCALE) {
+            throw new IllegalArgumentException(
+                    "DECIMAL256 scale must be in [0, " + DECIMAL256_MAX_SCALE + "], got " + scale);
+        }
+    }
+
+    private void checkScale64(int scale) {
+        if (scale < 0 || scale > DECIMAL64_MAX_SCALE) {
             throw new IllegalArgumentException(
-                    "scale must be in [0, " + Decimals.MAX_SCALE + "], got " + scale);
+                    "DECIMAL64 scale must be in [0, " + DECIMAL64_MAX_SCALE + "], got " + scale);
         }
     }
 

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

@@ -247,8 +247,19 @@ public void releaseBuffer(QwpBatchBuffer buffer) {
         // surfaces as a broken buffer rather than a slow native-memory leak.
         if (!freeBuffers.offer(buffer)) {
             buffer.close();
+            return;
         }
         pendingRelease.offer(RELEASE_TOKEN);
+        // Close race: closePool may have set closed AND drained freeBuffers
+        // between our closed-check above and our offer. In that window the
+        // buffer landed in a cleared-and-abandoned queue with no consumer.
+        // Re-check after offer: if closed is now true and the buffer is
+        // still in freeBuffers, remove it and close in place. remove() is
+        // atomic on ArrayBlockingQueue, so if closePool's drain beat us to
+        // it, remove returns false and we skip the double-close.
+        if (closed && freeBuffers.remove(buffer)) {
+            buffer.close();
+        }
     }
 
     /**