fix(reader): harden RunEnd/Constant/Zoned/Pco against malformed input

Part of TODO.md's "Per-encoding adversarial tests" security item (CLAUDE.md
§Security contract): a malformed file must always throw VortexException,
never a raw JDK exception.

- RunEndEncodingDecoder: reject negative num_runs and zero runs paired with a
  non-empty row count — previously decoded "successfully" into a lazy array
  backed by an empty ends/values child, then threw a raw
  IndexOutOfBoundsException/ArithmeticException on first read.
- ConstantEncodingDecoder: reject a Decimal-typed constant whose scalar oneof
  doesn't carry bytes_value (raw NullPointerException reading its length).
- ScanIterator.decodeZoneTable: bound the zone-map table's declared row count
  (an unvalidated layout field, deliberately decoupled from the data layout's
  chunk count) before it sizes an ArrayList — a negative value threw a raw
  IllegalArgumentException and a value just over Integer.MAX_VALUE wrapped
  negative on the int cast.
- PcoTansDecoder.build: size the degenerate (zero-bin) decode table to
  tableSize instead of a fixed 1-state table — a page's initial ANS state
  indices are read with ansSizeLog bits regardless of bin count, so a
  corrupt file pairing zero bins with a nonzero ansSizeLog indexed a stale
  1-entry array out of bounds.
- PcoEncodingDecoder: reject a bin offsetBits > 64 (wider than any latent);
  validate that declared per-page value counts are non-negative and sum to
  the expected valid row count before allocating/writing latent buffers
  sized or offset by them.

TODO.md: mark RunEnd, Constant, Zoned, and Pco done — all eleven
per-encoding gotchas now closed.

Author: dfa1

TODO.md

@@ -38,13 +38,7 @@ known gap, a contract audit, or supporting infra.
 Each encoding's `decode(DecodeContext)` should be exercised against crafted metadata that
 decodes but disagrees with the buffer payload. `bufferIndices[i] >= ctx.bufferCount()` (and the
 equivalent child-index check) is centralized in `DecodeContext.buffer(i)`/`decodeChild(i)`.
-VarBin, Dict, Bitpacked, ALP, Sparse, Chunked, and Struct are done — remaining gotchas:
-
-- [ ] **RLE / RunEnd**: `run_ends` non-monotonic; last `run_end` ≠ `row_count`.
-- [ ] **Constant**: protobuf scalar value missing or type-mismatched against declared `DType`.
-- [ ] **Zoned**: zone-map min > max; zone count ≠ child chunk count.
-- [ ] **Pco**: `bits_per_offset > 64`; `bin_count == 0` with non-empty page; per-page
-  `n` greater than `DEFAULT_MAX_PAGE_N`; ANS state values inconsistent with weight table.
+VarBin, Dict, Bitpacked, ALP, Sparse, Chunked, Struct, RunEnd, Constant, Zoned, and Pco are done.
 
 ### Resource caps
 

reader/src/main/java/io/github/dfa1/vortex/reader/ScanIterator.java

@@ -475,6 +475,15 @@ private List<ArrayStats> decodeZoneTable(ColumnName column) {
             return null;
         }
         long nZones = statsFlat.rowCount();
+        // statsFlat.rowCount() is an unvalidated field straight from the layout FlatBuffer
+        // (PostscriptParser never bounds it — zone count is deliberately decoupled from the
+        // data layout's chunk count, see this method's Javadoc). Below it sizes an ArrayList
+        // and drives a per-zone loop via an `(int) nZones` cast: a negative value throws a raw
+        // IllegalArgumentException from the ArrayList constructor instead of degrading to "no
+        // zone map" like every other unusable shape this method already falls back on.
+        if (nZones < 0 || nZones > Integer.MAX_VALUE) {
+            return null;
+        }
         SegmentSpec spec = file.footer().segmentSpecs().get(segIdx);
         try (Arena tableArena = Arena.ofConfined()) {
             Array decoded = file.decodeSegment(spec, statsDtype, nZones, tableArena);

reader/src/main/java/io/github/dfa1/vortex/reader/decode/ConstantEncodingDecoder.java

@@ -114,6 +114,13 @@ private static Array constantPrimitive(DType outDtype, PType ptype, ProtoScalarV
 
     private static Array decodeDecimal(DType dtype, ProtoScalarValue scalar, long n) {
         byte[] elemBytes = scalar.bytes_value();
+        if (elemBytes == null) {
+            // A scalar whose oneof tag doesn't match the declared Decimal dtype (e.g. only
+            // int64_value set) leaves bytes_value() null; without this guard the length read
+            // below is a raw NullPointerException instead of a VortexException (ADR 0003).
+            throw new VortexException(EncodingId.VORTEX_CONSTANT,
+                    "constant decimal scalar missing bytes_value");
+        }
         int elemLen = elemBytes.length;
         // Decode the single scalar value via LazyDecimalArray (reuses its LE byte-order logic),
         // then wrap in a constant array — O(1) allocation regardless of row count.

reader/src/main/java/io/github/dfa1/vortex/reader/decode/PcoEncodingDecoder.java

@@ -7,6 +7,7 @@
 import io.github.dfa1.vortex.core.io.PTypeIO;
 import io.github.dfa1.vortex.core.proto.ProtoPcoChunkInfo;
 import io.github.dfa1.vortex.core.proto.ProtoPcoMetadata;
+import io.github.dfa1.vortex.core.proto.ProtoPcoPageInfo;
 import io.github.dfa1.vortex.reader.array.Array;
 import io.github.dfa1.vortex.reader.array.BoolArray;
 import io.github.dfa1.vortex.reader.array.MaskedArray;
@@ -67,6 +68,28 @@ public Array decode(DecodeContext ctx) {
             }
         }
 
+        // Pages declare their own value counts (ProtoPcoPageInfo.n_values), independent of
+        // validCount. A crafted file can pair a huge or negative per-page count with a small
+        // rowCount: without this check, a negative count silently no-ops its loop while a
+        // desynced total either writes past rawLatents/compactOut (raw IndexOutOfBounds) or
+        // sizes rawAdjs from an attacker-controlled chunkN unrelated to any real buffer
+        // (OutOfMemoryError). Validating the total up front keeps every per-page/per-chunk
+        // access below implicitly bounded by validCount.
+        long totalPageValues = 0L;
+        for (ProtoPcoChunkInfo chunkInfo : meta.chunks()) {
+            for (ProtoPcoPageInfo page : chunkInfo.pages()) {
+                if (page.n_values() < 0) {
+                    throw new VortexException(EncodingId.VORTEX_PCO,
+                            "pco page n_values " + page.n_values() + " is negative");
+                }
+                totalPageValues += page.n_values();
+            }
+        }
+        if (totalPageValues != validCount) {
+            throw new VortexException(EncodingId.VORTEX_PCO,
+                    "pco total page values " + totalPageValues + " != expected valid row count " + validCount);
+        }
+
         MemorySegment rawLatents = ctx.arena().allocate(validCount * Long.BYTES);
 
         int nChunks = meta.chunks().size();
@@ -727,6 +750,14 @@ private static PcoBin[] readBins(LeBitReader r, int nBins, int ansSizeLog, int d
             int weight = (int) r.readBits(ansSizeLog) + 1;
             long lower = r.readBits(dtypeSize);
             int offsetBits = (int) r.readBits(offsetBitsWidth);
+            if (offsetBits > 64) {
+                // offsetBitsWidth is 5/6/7 bits wide (max value 31/63/127), wider than the
+                // 64-bit latent an offset can ever legally span; a page later reads this many
+                // bits per value via LeBitReader#readBits(int), whose own <=64 contract this
+                // would otherwise violate.
+                throw new VortexException(EncodingId.VORTEX_PCO,
+                        "pco bin offsetBits " + offsetBits + " exceeds max 64");
+            }
             bins[b] = new PcoBin(weight, lower, offsetBits);
         }
         return bins;

reader/src/main/java/io/github/dfa1/vortex/reader/decode/PcoTansDecoder.java

@@ -33,12 +33,17 @@ private PcoTansDecoder(int[] nextStateIdxBase, int[] bitsToRead,
     ///
     /// Port of `Spec::from_weights` + `Decoder::new` from pcodec.
     public static PcoTansDecoder build(int ansSizeLog, PcoBin[] bins) {
+        int tableSize = 1 << ansSizeLog;
         if (bins.length == 0) {
-            // Degenerate: no bins → 1-state table, all offsets zero.
-            return new PcoTansDecoder(new int[]{0}, new int[]{0}, new int[]{0}, new long[]{0L});
+            // Degenerate: no bins → every state decodes to offset zero. Sized to tableSize
+            // (not a fixed 1-state table): the initial ANS state indices a page carries are
+            // read with ansSizeLog bits (so any value in [0, tableSize) is possible) before
+            // this decoder is consulted — a corrupt file pairing zero bins with a nonzero
+            // ansSizeLog previously indexed a real 1-entry table out of bounds, a raw
+            // ArrayIndexOutOfBoundsException instead of a VortexException (ADR 0003).
+            return new PcoTansDecoder(new int[tableSize], new int[tableSize], new int[tableSize], new long[tableSize]);
         }
 
-        int tableSize = 1 << ansSizeLog;
         int[] weights = new int[bins.length];
         for (int i = 0; i < bins.length; i++) {
             weights[i] = bins[i].weight();

reader/src/main/java/io/github/dfa1/vortex/reader/decode/RunEndEncodingDecoder.java

@@ -52,6 +52,18 @@ public Array decode(DecodeContext ctx) {
         long offset = meta.offset();
 
         long n = ctx.rowCount();
+        if (numRuns < 0) {
+            throw new VortexException(EncodingId.VORTEX_RUNEND, "runend: negative num_runs " + numRuns);
+        }
+        if (numRuns == 0 && n > 0) {
+            // Zero runs cover no rows — a crafted file pairing that with a non-empty row
+            // count previously decoded "successfully" into a LazyRunEndXxxArray backed by
+            // an empty ends/values child, then threw a raw IndexOutOfBoundsException (or
+            // ArithmeticException via the % elementCount broadcast path) on first read
+            // instead of failing here as a VortexException.
+            throw new VortexException(EncodingId.VORTEX_RUNEND,
+                    "runend: zero runs cannot cover " + n + " row(s)");
+        }
         DType endsDtype = new DType.Primitive(endsPtype, false);
         Array endsArr = ctx.decodeChild(0, endsDtype, numRuns);
         Array endsData = endsArr instanceof MaskedArray m ? m.inner() : endsArr;

reader/src/test/java/io/github/dfa1/vortex/reader/ScanIteratorZoneCountAdversarialTest.java

@@ -0,0 +1,104 @@
+package io.github.dfa1.vortex.reader;
+
+import io.github.dfa1.vortex.core.model.ColumnName;
+import io.github.dfa1.vortex.core.model.DType;
+import io.github.dfa1.vortex.core.model.LayoutId;
+import io.github.dfa1.vortex.reader.layout.Layout;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import java.lang.foreign.Arena;
+import java.lang.foreign.MemorySegment;
+import java.lang.foreign.ValueLayout;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.BDDMockito.given;
+
+/// A `vortex.stats` (zoned) layout's zone-map table row count is its own layout metadata field
+/// (`statsFlat.rowCount()`), never bounds-checked at parse time or cross-checked against the
+/// data layout's actual chunk count (see [ScanIterator#columnZoneStats] Javadoc — the two are
+/// deliberately decoupled). [ScanIterator] previously cast that attacker-controlled row count
+/// straight to `int` to size an `ArrayList`: a negative value threw a raw
+/// `IllegalArgumentException` and a value just over `Integer.MAX_VALUE` wrapped to negative on
+/// the cast, both instead of the documented "fall back to per-chunk stats" behavior.
+@ExtendWith(MockitoExtension.class)
+class ScanIteratorZoneCountAdversarialTest {
+
+    private static final ColumnName COLUMN = ColumnName.of("v");
+    private static final DType.Struct SCHEMA = new DType.Struct(List.of(COLUMN), List.of(DType.I64), false);
+
+    @Mock
+    private VortexHandle file;
+
+    @ParameterizedTest
+    @ValueSource(longs = {-1L, Long.MIN_VALUE, ((long) Integer.MAX_VALUE) + 1L, Long.MAX_VALUE})
+    void corruptZoneCount_fallsBackInsteadOfCrashing(long corruptZoneCount) {
+        // Given — a one-chunk file whose zone-map table declares a corrupt row count
+        Layout root = rootLayout(corruptZoneCount);
+        Footer footer = new Footer(List.of(), List.of(),
+                List.of(new SegmentSpec(0, 0, (byte) 0, CompressionScheme.NONE),
+                        new SegmentSpec(0, 0, (byte) 0, CompressionScheme.NONE)),
+                List.of());
+        given(file.dtype()).willReturn(SCHEMA);
+        given(file.layout()).willReturn(root);
+        given(file.footer()).willReturn(footer);
+
+        // When
+        List<ArrayStats> result;
+        try (ScanIterator sut = new ScanIterator(file, ScanOptions.columns("v"))) {
+            result = sut.columnZoneStats("v");
+        }
+
+        // Then — degrades to the per-chunk fallback (one empty entry per chunk), no raw exception
+        assertThat(result).hasSize(1);
+        assertThat(result.getFirst()).isEqualTo(ArrayStats.empty());
+    }
+
+    @Test
+    void plausibleZoneCount_isNotRejected() {
+        // Given — a small, legitimate-looking zone count on an otherwise-corrupt (headerless)
+        // stats segment, which still degrades gracefully once decoding is attempted
+        Layout root = rootLayout(1L);
+        Footer footer = new Footer(List.of(), List.of(),
+                List.of(new SegmentSpec(0, 0, (byte) 0, CompressionScheme.NONE),
+                        new SegmentSpec(0, 0, (byte) 0, CompressionScheme.NONE)),
+                List.of());
+        given(file.dtype()).willReturn(SCHEMA);
+        given(file.layout()).willReturn(root);
+        given(file.footer()).willReturn(footer);
+
+        // When
+        List<ArrayStats> result;
+        try (ScanIterator sut = new ScanIterator(file, ScanOptions.columns("v"))) {
+            result = sut.columnZoneStats("v");
+        }
+
+        // Then — the guard only rejects implausible counts; this one reaches the normal decode
+        // path (which itself falls back gracefully on the segment's missing content)
+        assertThat(result).hasSize(1);
+    }
+
+    /// Builds `Struct(v) -> Zoned[Flat(data, empty segment 0), Flat(stats, rowCount=zoneCount,
+    /// segment 1)]`. The data flat's zero-length segment makes the per-chunk fallback resolve to
+    /// [ArrayStats#empty()] without needing real FlatBuffer bytes.
+    private static Layout rootLayout(long zoneCount) {
+        Layout dataFlat = new Layout(LayoutId.FLAT, 5, null, List.of(), List.of(0));
+        Layout statsFlat = new Layout(LayoutId.FLAT, zoneCount, minStatBitset(), List.of(), List.of(1));
+        Layout zoned = new Layout(LayoutId.STATS, 5, null, List.of(dataFlat, statsFlat), List.of());
+        return new Layout(LayoutId.STRUCT, 5, null, List.of(zoned), List.of());
+    }
+
+    /// `vortex.stats` metadata: 4-byte zone length (unused here) + a bitset with the `MIN` bit
+    /// (ordinal 4) set, so [io.github.dfa1.vortex.reader.layout.ZonedStatsSchema#statsTableDtype]
+    /// resolves a non-empty schema and the code under test proceeds past its early-return guards.
+    private static MemorySegment minStatBitset() {
+        MemorySegment seg = Arena.ofAuto().allocate(5);
+        seg.set(ValueLayout.JAVA_BYTE, 4, (byte) 0x10);
+        return seg;
+    }
+}

reader/src/test/java/io/github/dfa1/vortex/reader/decode/ConstantEncodingDecoderTest.java

@@ -0,0 +1,80 @@
+package io.github.dfa1.vortex.reader.decode;
+
+import io.github.dfa1.vortex.core.error.VortexException;
+import io.github.dfa1.vortex.core.model.DType;
+import io.github.dfa1.vortex.core.model.EncodingId;
+import io.github.dfa1.vortex.core.proto.ProtoScalarValue;
+import io.github.dfa1.vortex.reader.ReadRegistry;
+import io.github.dfa1.vortex.reader.array.Array;
+import io.github.dfa1.vortex.reader.array.LongArray;
+import org.junit.jupiter.api.Test;
+
+import java.lang.foreign.Arena;
+import java.lang.foreign.MemorySegment;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+class ConstantEncodingDecoderTest {
+
+    private static final ConstantEncodingDecoder SUT = new ConstantEncodingDecoder();
+
+    @Test
+    void encodingId_isVortexConstant() {
+        // Given / When / Then
+        assertThat(SUT.encodingId()).isEqualTo(EncodingId.VORTEX_CONSTANT);
+    }
+
+    @Test
+    void primitiveScalar_missingAllValueFields_decodesAsZero() {
+        // Given — a scalar with every oneof field null (no tag matched the declared I64
+        // dtype); scalarToRawBits() has an explicit fallback for this, so it must not crash.
+        ProtoScalarValue scalar = new ProtoScalarValue(null, null, null, null, null, null, null, null, null, null, null);
+
+        // When
+        Array result = decode(scalar, DType.I64, 3);
+
+        // Then
+        LongArray longs = (LongArray) result;
+        assertThat(longs.getLong(0)).isZero();
+        assertThat(longs.getLong(2)).isZero();
+    }
+
+    /// A scalar whose oneof tag doesn't match the declared Decimal dtype (e.g. only
+    /// int64_value set, `bytes_value` absent) previously threw a raw NullPointerException
+    /// reading `bytes_value().length` instead of a [VortexException] (ADR 0003).
+    @Test
+    void decimalScalar_missingBytesValue_throwsVortexException() {
+        // Given — int64_value set, bytes_value absent, for a Decimal-typed constant
+        ProtoScalarValue scalar = new ProtoScalarValue(null, null, 42L, null, null, null, null, null, null, null, null);
+        DType decimalDtype = new DType.Decimal((byte) 10, (byte) 2, false);
+
+        // When / Then
+        assertThatThrownBy(() -> decode(scalar, decimalDtype, 1))
+                .isInstanceOf(VortexException.class)
+                .hasMessageContaining("bytes_value");
+    }
+
+    @Test
+    void decimalScalar_withBytesValue_decodes() {
+        // Given — a 4-byte little-endian two's-complement decimal, scale 2 → 12345 / 100
+        ProtoScalarValue scalar = new ProtoScalarValue(
+                null, null, null, null, null, null, null,
+                new byte[]{(byte) 0x39, (byte) 0x30, (byte) 0x00, (byte) 0x00}, null, null, null);
+        DType decimalDtype = new DType.Decimal((byte) 9, (byte) 2, false);
+
+        // When
+        Array result = decode(scalar, decimalDtype, 2);
+
+        // Then
+        assertThat(result.length()).isEqualTo(2);
+    }
+
+    private static Array decode(ProtoScalarValue scalar, DType dtype, long n) {
+        MemorySegment scalarBuf = MemorySegment.ofArray(scalar.encode());
+        ArrayNode node = new ArrayNode(EncodingId.VORTEX_CONSTANT, null, new ArrayNode[0], new int[]{0});
+        DecodeContext ctx = new DecodeContext(node, dtype, n, new MemorySegment[]{scalarBuf},
+                ReadRegistry.empty(), Arena.ofAuto());
+        return SUT.decode(ctx);
+    }
+}

reader/src/test/java/io/github/dfa1/vortex/reader/decode/PcoEncodingDecoderTest.java

@@ -146,6 +146,28 @@ private static MemorySegment chunkMetaConv1(int quantization, long biasLatent,
         return segmentOf(buf);
     }
 
+    /// Packs `values[i]` into `widths[i]` bits, LSB-first per field then concatenated — the
+    /// same layout [LeBitReader#readBits(int)] consumes.
+    private static byte[] packBitsLsbFirst(int[] widths, long[] values) {
+        java.util.BitSet bits = new java.util.BitSet();
+        int pos = 0;
+        for (int f = 0; f < widths.length; f++) {
+            for (int i = 0; i < widths[f]; i++) {
+                if (((values[f] >>> i) & 1L) != 0L) {
+                    bits.set(pos);
+                }
+                pos++;
+            }
+        }
+        byte[] buf = new byte[Math.max((pos + 7) / 8, 1)];
+        for (int i = 0; i < pos; i++) {
+            if (bits.get(i)) {
+                buf[i / 8] |= (byte) (1 << (i % 8));
+            }
+        }
+        return buf;
+    }
+
     private static MemorySegment chunkMetaLookback() {
         return segmentOf((byte) 0x20, (byte) 0x00, (byte) 0x00, (byte) 0x00, (byte) 0x00, (byte) 0x00, (byte) 0x00);
     }
@@ -509,5 +531,45 @@ void conv1Delta_with64BitDtype_throwsVortexException(PType ptype) {
                     .isInstanceOf(VortexException.class)
                     .hasMessageContaining("Conv1");
         }
+
+        @Test
+        void binOffsetBitsExceeds64_throwsVortexException() {
+            // Given — mode=Classic(0), delta=NoOp(0), ansSizeLog=0, nBins=1, one bin whose
+            // offsetBits (100) exceeds the 64-bit latent it would be read into.
+            byte[] chunkMeta = packBitsLsbFirst(
+                    new int[]{4, 4, 4, 15, 64, 7},
+                    new long[]{0, 0, 0, 1, 0, 100});
+            DecodeContext ctx = ctxWith(metaWithOneChunk(1), DType.U64, 1,
+                    new MemorySegment[]{segmentOf(chunkMeta), segmentOf((byte) 0x00)});
+
+            // When / Then
+            assertThatThrownBy(() -> SUT.decode(ctx))
+                    .isInstanceOf(VortexException.class)
+                    .hasMessageContaining("offsetBits");
+        }
+
+        @Test
+        void pageValuesTotalMismatchesRowCount_throwsVortexException() {
+            // Given — one page declares 5 values but the context row count is 3
+            DecodeContext ctx = ctxWith(metaWithOneChunk(5), DType.U64, 3,
+                    new MemorySegment[]{segmentOf((byte) 0x00), segmentOf((byte) 0x00)});
+
+            // When / Then
+            assertThatThrownBy(() -> SUT.decode(ctx))
+                    .isInstanceOf(VortexException.class)
+                    .hasMessageContaining("total page values");
+        }
+
+        @Test
+        void negativePageNValues_throwsVortexException() {
+            // Given — a page whose n_values decodes to -1 (a valid varint32 on the wire)
+            DecodeContext ctx = ctxWith(metaWithOneChunk(-1), DType.U64, 0,
+                    new MemorySegment[]{segmentOf((byte) 0x00), segmentOf((byte) 0x00)});
+
+            // When / Then
+            assertThatThrownBy(() -> SUT.decode(ctx))
+                    .isInstanceOf(VortexException.class)
+                    .hasMessageContaining("negative");
+        }
     }
 }