diff --git a/lang/php/lib/Datum/AvroIOBinaryDecoder.php b/lang/php/lib/Datum/AvroIOBinaryDecoder.php
index ac5fcf7e929..da7888d96ca 100644
--- a/lang/php/lib/Datum/AvroIOBinaryDecoder.php
+++ b/lang/php/lib/Datum/AvroIOBinaryDecoder.php
@@ -38,6 +38,14 @@
  */
 class AvroIOBinaryDecoder
 {
+    /**
+     * Reads with a declared length above this many bytes are validated against
+     * the number of bytes actually remaining before allocating, to guard
+     * against an out-of-memory attack from a malicious or truncated input.
+     * Smaller reads skip the check to avoid per-value overhead.
+     */
+    private const MAX_UNCHECKED_READ = 1048576; // 1 MiB
+
     /**
      * @param AvroIO $io object from which to read.
      */
@@ -65,9 +73,39 @@ public function readBoolean(): bool
      */
     public function read(int $len): string
     {
+        if ($len < 0) {
+            // AvroStringIO::read() accepts a negative length and moves the
+            // pointer backwards; reject it before delegating.
+            throw new AvroException("Cannot read a negative number of bytes: {$len}");
+        }
+        if ($len > self::MAX_UNCHECKED_READ) {
+            $remaining = $this->bytesRemaining();
+            if ($len > $remaining) {
+                throw new AvroException("Cannot read {$len} bytes, only {$remaining} remaining.");
+            }
+        }
+
         return $this->io->read($len);
     }
 
+    /**
+     * Number of bytes still available to read, determined by seeking to the end
+     * and restoring the position (which both the string and file IO
+     * implementations support). Used to reject a declared length or collection
+     * block count that exceeds the data actually available before allocating.
+     */
+    public function bytesRemaining(): int
+    {
+        $current = $this->io->tell();
+        $this->io->seek(0, AvroIO::SEEK_END);
+        $end = $this->io->tell();
+        $this->io->seek($current, AvroIO::SEEK_SET);
+
+        // Clamp to 0: AvroStringIO::seek() allows seeking past EOF, which would
+        // otherwise yield a confusing negative "remaining" count.
+        return max(0, $end - $current);
+    }
+
     public function readInt(): int
     {
         return (int) $this->readLong();
@@ -78,6 +116,12 @@ public function readLong(): string|int
         $byte = ord($this->nextByte());
         $bytes = [$byte];
         while (0 != ($byte & 0x80)) {
+            // A 64-bit value uses at most 10 bytes; reject an overlong varint
+            // rather than reading an unbounded continuation chain and silently
+            // corrupting the value. Bounds both the native and GMP decode paths.
+            if (count($bytes) >= 10) {
+                throw new AvroException('Varint is too long');
+            }
             $byte = ord($this->nextByte());
             $bytes[] = $byte;
         }
@@ -235,13 +279,40 @@ public function skipRecord(AvroRecordSchema $writersSchema, AvroIOBinaryDecoder
 
     public function skipArray(AvroArraySchema $writersSchema, AvroIOBinaryDecoder $decoder): void
     {
+        $minBytes = AvroIODatumReader::collectionElementMinBytes($writersSchema->items());
+        $skipped = 0;
         $blockCount = $decoder->readLong();
-        while (0 !== $blockCount) {
+        while (0 != $blockCount) {
+            $blockSize = null;
             if ($blockCount < 0) {
-                $decoder->skip($this->readLong());
+                if (PHP_INT_MIN == $blockCount) {
+                    throw new AvroException('Invalid array block count');
+                }
+                $blockCount = -$blockCount;
+                $blockSize = $decoder->readLong();
+                if ($blockSize < 0) {
+                    throw new AvroException('Invalid negative array block size');
+                }
             }
-            for ($i = 0; $i < $blockCount; $i++) {
-                AvroIODatumReader::skipData($writersSchema->items(), $decoder);
+            // Bound the (normalized) count on both the sized and unsized paths so
+            // a negative block count cannot bypass the skip limit.
+            AvroIODatumReader::checkSkipCollectionCount($skipped, $blockCount, $minBytes);
+            $skipped += $blockCount;
+            if (null !== $blockSize) {
+                // seek() can move past EOF, so a truncated/oversized block would
+                // otherwise be "skipped" silently, hiding truncation. Reject a
+                // block size larger than the bytes actually remaining.
+                if ($blockSize > $decoder->bytesRemaining()) {
+                    throw new AvroException('Array block size exceeds the remaining input');
+                }
+                if ($minBytes > 0 && $blockCount > intdiv($blockSize, $minBytes)) {
+                    throw new AvroException('Array block size too small for the declared element count');
+                }
+                $decoder->skip($blockSize);
+            } else {
+                for ($i = 0; $i < $blockCount; $i++) {
+                    AvroIODatumReader::skipData($writersSchema->items(), $decoder);
+                }
             }
             $blockCount = $decoder->readLong();
         }
@@ -249,14 +320,39 @@ public function skipArray(AvroArraySchema $writersSchema, AvroIOBinaryDecoder $d
 
     public function skipMap(AvroMapSchema $writersSchema, AvroIOBinaryDecoder $decoder): void
     {
+        // Map entries always carry a >= 1 byte key, so the minimum is positive.
+        $minBytes = 1 + AvroIODatumReader::collectionElementMinBytes($writersSchema->values());
+        $skipped = 0;
         $blockCount = $decoder->readLong();
-        while (0 !== $blockCount) {
+        while (0 != $blockCount) {
+            $blockSize = null;
             if ($blockCount < 0) {
-                $decoder->skip($this->readLong());
+                if (PHP_INT_MIN == $blockCount) {
+                    throw new AvroException('Invalid map block count');
+                }
+                $blockCount = -$blockCount;
+                $blockSize = $decoder->readLong();
+                if ($blockSize < 0) {
+                    throw new AvroException('Invalid negative map block size');
+                }
             }
-            for ($i = 0; $i < $blockCount; $i++) {
-                $decoder->skipString();
-                AvroIODatumReader::skipData($writersSchema->values(), $decoder);
+            AvroIODatumReader::checkSkipCollectionCount($skipped, $blockCount, $minBytes);
+            $skipped += $blockCount;
+            if (null !== $blockSize) {
+                // seek() can move past EOF; reject a block size larger than the
+                // bytes remaining so a truncated block isn't silently skipped.
+                if ($blockSize > $decoder->bytesRemaining()) {
+                    throw new AvroException('Map block size exceeds the remaining input');
+                }
+                if ($minBytes > 0 && $blockCount > intdiv($blockSize, $minBytes)) {
+                    throw new AvroException('Map block size too small for the declared element count');
+                }
+                $decoder->skip($blockSize);
+            } else {
+                for ($i = 0; $i < $blockCount; $i++) {
+                    $decoder->skipString();
+                    AvroIODatumReader::skipData($writersSchema->values(), $decoder);
+                }
             }
             $blockCount = $decoder->readLong();
         }
diff --git a/lang/php/lib/Datum/AvroIOCollectionSizeException.php b/lang/php/lib/Datum/AvroIOCollectionSizeException.php
new file mode 100644
index 00000000000..6aa9b0bce0a
--- /dev/null
+++ b/lang/php/lib/Datum/AvroIOCollectionSizeException.php
@@ -0,0 +1,43 @@
+<?php
+
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+declare(strict_types=1);
+
+namespace Apache\Avro\Datum;
+
+use Apache\Avro\AvroException;
+
+/**
+ * Raised when an array or map declares more items than the configured maximum.
+ *
+ * The block count of an array or map is read from the (potentially untrusted or
+ * truncated) input and drives allocation of the resulting collection. This
+ * exception guards against unbounded memory allocation from a very large or
+ * malformed block count.
+ */
+class AvroIOCollectionSizeException extends AvroException
+{
+    public function __construct(int $maxItems)
+    {
+        parent::__construct(
+            sprintf('Cannot read collections larger than %d items.', $maxItems)
+        );
+    }
+}
diff --git a/lang/php/lib/Datum/AvroIODatumReader.php b/lang/php/lib/Datum/AvroIODatumReader.php
index 48f193f4c02..78d087c1665 100644
--- a/lang/php/lib/Datum/AvroIODatumReader.php
+++ b/lang/php/lib/Datum/AvroIODatumReader.php
@@ -43,6 +43,29 @@
  */
 class AvroIODatumReader
 {
+    /**
+     * Name of the environment variable overriding the maximum number of
+     * zero-byte-encoded collection elements (e.g. an array of nulls) to
+     * allocate from a single decode.
+     */
+    public const MAX_COLLECTION_ITEMS_ENV = 'AVRO_MAX_COLLECTION_ITEMS';
+
+    /**
+     * Default maximum number of zero-byte-encoded collection elements to
+     * allocate. Such elements consume no input, so the bytes-remaining check
+     * cannot bound their count; without a cap a tiny payload can declare a huge
+     * block count and exhaust memory. Overridable with the
+     * {@see self::MAX_COLLECTION_ITEMS_ENV} environment variable.
+     */
+    public const DEFAULT_MAX_COLLECTION_ITEMS = 10000000;
+
+    /**
+     * Structural cap on the number of elements in any array or map (an overflow
+     * / defense-in-depth guard), matching the historical Integer.MAX_VALUE - 8
+     * limit. Non-zero-byte elements are also bounded by the bytes remaining.
+     */
+    public const DEFAULT_MAX_COLLECTION_STRUCTURAL = 2147483639;
+
     public function __construct(
         private ?AvroSchema $writersSchema = null,
         private ?AvroSchema $readersSchema = null
@@ -284,12 +307,22 @@ public function readArray(
         AvroIOBinaryDecoder $decoder
     ): array {
         $items = [];
+        $minBytes = self::minBytesPerElement($writersSchema->items());
         $blockCount = $decoder->readLong();
-        while (0 !== $blockCount) {
+        while (0 != $blockCount) {
             if ($blockCount < 0) {
+                // PHP_INT_MIN cannot be negated: -PHP_INT_MIN promotes to a
+                // float, so reject it rather than propagating a non-int count.
+                if (PHP_INT_MIN == $blockCount) {
+                    throw new AvroException('Invalid array block count');
+                }
                 $blockCount = -$blockCount;
-                $decoder->readLong(); // Read (and ignore) block size
+                $blockSize = $decoder->readLong(); // block byte-size
+                if ($blockSize < 0) {
+                    throw new AvroException('Invalid negative array block size');
+                }
             }
+            self::ensureCollectionAvailable($decoder, count($items), $blockCount, $minBytes);
             for ($i = 0; $i < $blockCount; $i++) {
                 $items[] = $this->readData(
                     $writersSchema->items(),
@@ -312,14 +345,29 @@ public function readMap(
         AvroIOBinaryDecoder $decoder
     ): array {
         $items = [];
+        $minBytes = 1 + self::minBytesPerElement($writersSchema->values());
+        $read = 0; // Cumulative pairs read; count($items) would undercount duplicate keys.
         $pair_count = $decoder->readLong();
         while (0 != $pair_count) {
             if ($pair_count < 0) {
+                // PHP_INT_MIN cannot be negated: -PHP_INT_MIN promotes to a
+                // float, so reject it rather than propagating a non-int count.
+                if (PHP_INT_MIN == $pair_count) {
+                    throw new AvroException('Invalid map block count');
+                }
                 $pair_count = -$pair_count;
-                // Note: we're not doing anything with block_size other than skipping it
-                $decoder->readLong();
+                $blockSize = $decoder->readLong(); // block byte-size
+                if ($blockSize < 0) {
+                    throw new AvroException('Invalid negative map block size');
+                }
             }
 
+            // Map keys are strings (>= 1 byte length prefix) plus the value.
+            // Bound against the cumulative pairs read, not count($items): a
+            // stream repeating the same key would otherwise shrink count($items)
+            // and slip past the cumulative cap.
+            self::ensureCollectionAvailable($decoder, $read, $pair_count, $minBytes);
+            $read += $pair_count;
             for ($i = 0; $i < $pair_count; $i++) {
                 $key = $decoder->readString();
                 $items[$key] = $this->readData(
@@ -530,6 +578,34 @@ public function readDefaultValue(AvroSchema $fieldSchema, mixed $defaultValue):
         }
     }
 
+    /**
+     * Minimum on-wire size of a collection element schema. Exposed for the skip
+     * path, which lives in the decoder.
+     */
+    public static function collectionElementMinBytes(AvroSchema $schema): int
+    {
+        return self::minBytesPerElement($schema);
+    }
+
+    /**
+     * Bounds the cumulative number of elements skipped in an array or map, so a
+     * huge block of zero-byte elements cannot loop unboundedly (a CPU
+     * exhaustion) even though skipping allocates nothing.
+     *
+     * @throws AvroIOCollectionSizeException if the limit is exceeded
+     */
+    public static function checkSkipCollectionCount(int $existing, int $count, int $minBytes): void
+    {
+        if ($count <= 0) {
+            return;
+        }
+        [$zeroByteLimit, $structuralLimit] = self::collectionLimits();
+        $limit = $minBytes > 0 ? $structuralLimit : $zeroByteLimit;
+        if ($count > $limit || $existing > $limit - $count) {
+            throw new AvroIOCollectionSizeException($limit);
+        }
+    }
+
     private function readDecimal(string $bytes, int $scale): string
     {
         $mostSignificantBit = ord($bytes[0]) & 0x80;
@@ -538,4 +614,123 @@ private function readDecimal(string $bytes, int $scale): string
 
         return (string) ($scale > 0 ? ($int / (10 ** $scale)) : $int);
     }
+
+    /**
+     * Minimum number of bytes a single value of the given schema can occupy on
+     * the wire. Used to reject an array/map block count that could not be backed
+     * by the bytes remaining. It returns 0 for any schema that can encode to
+     * zero bytes: the null primitive, but also a record with no fields or whose
+     * fields all encode to zero bytes. A zero return disables the collection
+     * check for that element type (so, e.g., an array of nulls is not falsely
+     * rejected). Types that cannot be resolved cheaply default to 1.
+     *
+     * @param array<int, bool> $visited
+     */
+    private static function minBytesPerElement(mixed $schema, array $visited = []): int
+    {
+        $type = $schema instanceof AvroSchema ? $schema->type() : $schema;
+        // Named/complex field types may nest a schema object; unwrap one level.
+        if ($type instanceof AvroSchema) {
+            return self::minBytesPerElement($type, $visited);
+        }
+        if (!is_string($type)) {
+            return 1;
+        }
+        switch ($type) {
+            case AvroSchema::NULL_TYPE:
+                return 0;
+            case AvroSchema::FLOAT_TYPE:
+                return 4;
+            case AvroSchema::DOUBLE_TYPE:
+                return 8;
+            case AvroSchema::FIXED_SCHEMA:
+                // Clamp to >= 0: a malformed fixed schema could have a negative
+                // size (AvroSchema::parse does not reject it), which would make
+                // this negative and cause ensureCollectionAvailable to treat the
+                // element as zero-byte.
+                return $schema instanceof AvroFixedSchema ? max(0, $schema->size()) : 1;
+            case AvroSchema::RECORD_SCHEMA:
+            case AvroSchema::ERROR_SCHEMA:
+                if (!$schema instanceof AvroRecordSchema) {
+                    return 1;
+                }
+                $id = spl_object_id($schema);
+                if (isset($visited[$id])) {
+                    return 1; // self-referencing schema: safe lower bound of 1 byte
+                }
+                $visited[$id] = true;
+                $total = 0;
+                foreach ($schema->fields() as $field) {
+                    $total += self::minBytesPerElement($field->type(), $visited);
+                }
+
+                return $total;
+            default:
+                // boolean, int, long, bytes, string, enum, union, array, map:
+                // all encode to at least one byte.
+                return 1;
+        }
+    }
+
+    /**
+     * Returns the configured collection limits as [zeroByteLimit, structuralLimit].
+     * AVRO_MAX_COLLECTION_ITEMS, when a non-negative integer, overrides both
+     * limits to that value (which may raise or lower them).
+     *
+     * @return array{int, int}
+     */
+    private static function collectionLimits(): array
+    {
+        $value = getenv(self::MAX_COLLECTION_ITEMS_ENV);
+        if (false !== $value && '' !== $value && preg_match('/^-?\d+$/', $value)) {
+            $parsed = (int) $value;
+            if ($parsed >= 0) {
+                return [$parsed, $parsed];
+            }
+        }
+
+        return [self::DEFAULT_MAX_COLLECTION_ITEMS, self::DEFAULT_MAX_COLLECTION_STRUCTURAL];
+    }
+
+    /**
+     * Rejects a collection (array or map) block that could not be backed by the
+     * input, before iterating.
+     *
+     * For elements with a positive minimum on-wire size, the declared count is
+     * checked against the bytes remaining and a structural cap. For zero-byte
+     * elements (e.g. an array of nulls), which consume no input and so cannot be
+     * bounded by the bytes remaining, the cumulative count is checked against the
+     * tighter zero-byte limit.
+     *
+     * @throws AvroException                 if the bytes-remaining check fails
+     * @throws AvroIOCollectionSizeException if a size limit is exceeded
+     */
+    private static function ensureCollectionAvailable(
+        AvroIOBinaryDecoder $decoder,
+        int $existing,
+        int $count,
+        int $minBytesPerElement
+    ): void {
+        if ($count <= 0) {
+            return;
+        }
+        [$zeroByteLimit, $structuralLimit] = self::collectionLimits();
+        if ($minBytesPerElement > 0) {
+            $remaining = $decoder->bytesRemaining();
+            if ($count > intdiv($remaining, $minBytesPerElement)) {
+                throw new AvroException(
+                    "Collection claims {$count} elements with at least {$minBytesPerElement} "
+                    ."bytes each, but only {$remaining} bytes are available."
+                );
+            }
+            if ($count > $structuralLimit || $existing > $structuralLimit - $count) {
+                throw new AvroIOCollectionSizeException($structuralLimit);
+            }
+
+            return;
+        }
+        if ($count > $zeroByteLimit || $existing > $zeroByteLimit - $count) {
+            throw new AvroIOCollectionSizeException($zeroByteLimit);
+        }
+    }
 }
diff --git a/lang/php/lib/Schema/AvroUnionSchema.php b/lang/php/lib/Schema/AvroUnionSchema.php
index c6b6cbf279e..dcf80640a85 100644
--- a/lang/php/lib/Schema/AvroUnionSchema.php
+++ b/lang/php/lib/Schema/AvroUnionSchema.php
@@ -98,7 +98,7 @@ public function schemas(): array
      */
     public function schemaByIndex($index): AvroSchema
     {
-        if (count($this->schemas) > $index) {
+        if ($index >= 0 && count($this->schemas) > $index) {
             return $this->schemas[$index];
         }
 
diff --git a/lang/php/lib/autoload.php b/lang/php/lib/autoload.php
index de1a863513b..12c9475e794 100644
--- a/lang/php/lib/autoload.php
+++ b/lang/php/lib/autoload.php
@@ -33,6 +33,7 @@
 
 include __DIR__.'/Datum/AvroIOBinaryDecoder.php';
 include __DIR__.'/Datum/AvroIOBinaryEncoder.php';
+include __DIR__.'/Datum/AvroIOCollectionSizeException.php';
 include __DIR__.'/Datum/AvroIODatumReader.php';
 include __DIR__.'/Datum/AvroIODatumWriter.php';
 include __DIR__.'/Datum/AvroIOSchemaMatchException.php';
diff --git a/lang/php/test/DatumIOTest.php b/lang/php/test/DatumIOTest.php
index 8614cf61365..8a840c00adb 100644
--- a/lang/php/test/DatumIOTest.php
+++ b/lang/php/test/DatumIOTest.php
@@ -24,6 +24,7 @@
 use Apache\Avro\AvroException;
 use Apache\Avro\Datum\AvroIOBinaryDecoder;
 use Apache\Avro\Datum\AvroIOBinaryEncoder;
+use Apache\Avro\Datum\AvroIOCollectionSizeException;
 use Apache\Avro\Datum\AvroIODatumReader;
 use Apache\Avro\Datum\AvroIODatumWriter;
 use Apache\Avro\Datum\Type\AvroDuration;
@@ -34,6 +35,19 @@
 
 class DatumIOTest extends TestCase
 {
+    /** @var false|string Value of AVRO_MAX_COLLECTION_ITEMS captured before each test. */
+    private string|false $originalMaxCollectionItems = false;
+
+    protected function setUp(): void
+    {
+        parent::setUp();
+        // Capture any pre-existing value so tests that override
+        // AVRO_MAX_COLLECTION_ITEMS restore it rather than unconditionally
+        // unsetting it, which would break isolation when the variable is already
+        // set in the environment (CI or a developer shell).
+        $this->originalMaxCollectionItems = getenv('AVRO_MAX_COLLECTION_ITEMS');
+    }
+
     #[DataProvider('data_provider')]
     public function test_datum_round_trip(string $schema_json, mixed $datum, string $binary): void
     {
@@ -261,6 +275,306 @@ public function test_invalid_bytes_logical_type_out_of_range(): void
         $writer->write("10000", $encoder);
     }
 
+    // A bytes/string value declares a length prefix; a malicious or truncated
+    // input can declare far more bytes than actually exist. On a reader that
+    // can report its size, that is rejected before allocating for it.
+    public function test_read_bytes_rejects_length_beyond_stream(): void
+    {
+        $io = new AvroStringIO();
+        (new AvroIOBinaryEncoder($io))->writeLong(100 * 1024 * 1024);
+        $io->seek(0);
+        $decoder = new AvroIOBinaryDecoder($io);
+
+        $this->expectException(AvroException::class);
+        $decoder->readBytes();
+    }
+
+    public function test_read_bytes_within_stream_still_reads(): void
+    {
+        $payload = str_repeat('x', 2 * 1024 * 1024);
+        $io = new AvroStringIO();
+        (new AvroIOBinaryEncoder($io))->writeBytes($payload);
+        $io->seek(0);
+        $decoder = new AvroIOBinaryDecoder($io);
+
+        $this->assertEquals($payload, $decoder->readBytes());
+    }
+
+    public function test_read_array_rejects_count_beyond_stream(): void
+    {
+        $io = new AvroStringIO();
+        (new AvroIOBinaryEncoder($io))->writeLong(1000000); // 1,000,000 longs, no data
+        $this->expectException(AvroException::class);
+        $this->decodeWith('{"type":"array","items":"long"}', $io);
+    }
+
+    public function test_read_map_rejects_count_beyond_stream(): void
+    {
+        $io = new AvroStringIO();
+        (new AvroIOBinaryEncoder($io))->writeLong(1000000);
+        $this->expectException(AvroException::class);
+        $this->decodeWith('{"type":"map","values":"long"}', $io);
+    }
+
+    public function test_read_union_rejects_out_of_range_index(): void
+    {
+        // Index 5 (zig-zag long) is out of range for a 2-branch union.
+        $io = new AvroStringIO();
+        (new AvroIOBinaryEncoder($io))->writeLong(5);
+        $this->expectException(AvroException::class);
+        $this->decodeWith('["null","long"]', $io);
+    }
+
+    public function test_read_union_rejects_negative_index(): void
+    {
+        // A negative index must not wrap to a valid branch.
+        $io = new AvroStringIO();
+        (new AvroIOBinaryEncoder($io))->writeLong(-1);
+        $this->expectException(AvroException::class);
+        $this->decodeWith('["null","long"]', $io);
+    }
+
+    public function test_read_long_rejects_overlong_varint(): void
+    {
+        // A 64-bit value uses at most 10 bytes; an 11th continuation byte is a
+        // malformed (overlong) varint and must be rejected.
+        $io = new AvroStringIO(str_repeat("\x80", 10)."\x01");
+        $io->seek(0);
+        $decoder = new AvroIOBinaryDecoder($io);
+        $this->expectException(AvroException::class);
+        $decoder->readLong();
+    }
+
+    public function test_read_array_of_null_not_falsely_rejected(): void
+    {
+        $count = 100000;
+        $io = new AvroStringIO();
+        $encoder = new AvroIOBinaryEncoder($io);
+        $encoder->writeLong($count); // one block of `count` nulls (zero bytes each)
+        $encoder->writeLong(0);      // end-of-array marker
+        $result = $this->decodeWith('{"type":"array","items":"null"}', $io);
+        $this->assertCount($count, $result);
+    }
+
+    public function test_read_array_within_stream_still_reads(): void
+    {
+        $schema = AvroSchema::parse('{"type":"array","items":"long"}');
+        $io = new AvroStringIO();
+        (new AvroIODatumWriter($schema))->write([1, 2, 3], new AvroIOBinaryEncoder($io));
+        $this->assertEquals([1, 2, 3], $this->decodeWith('{"type":"array","items":"long"}', $io));
+    }
+
+    public function test_array_of_null_exceeds_default_limit(): void
+    {
+        // The reported exploit: a ~6 byte payload declaring 200,000,000 nulls.
+        $this->expectException(AvroIOCollectionSizeException::class);
+        $this->decodeWith('{"type":"array","items":"null"}', self::zeroByteBlock(200000000));
+    }
+
+    public function test_array_of_null_int64_min_block_count(): void
+    {
+        // PHP_INT_MIN as a block count cannot be negated (-PHP_INT_MIN promotes
+        // to a float), so it must be rejected explicitly. It zig-zag encodes as
+        // the 10-byte varint below, followed by a block byte-size (0).
+        $io = new AvroStringIO(
+            "\xFF\xFF\xFF\xFF\xFF\xFF\xFF\xFF\xFF\x01\x00"
+        );
+        $this->expectException(AvroException::class);
+        $this->decodeWith('{"type":"array","items":"null"}', $io);
+    }
+
+    public function test_array_of_null_within_configured_limit_still_reads(): void
+    {
+        putenv('AVRO_MAX_COLLECTION_ITEMS=1000');
+
+        try {
+            $result = $this->decodeWith('{"type":"array","items":"null"}', self::zeroByteBlock(1000));
+            $this->assertCount(1000, $result);
+        } finally {
+            $this->restoreMaxCollectionItems();
+        }
+    }
+
+    public function test_array_of_null_exceeds_configured_limit(): void
+    {
+        putenv('AVRO_MAX_COLLECTION_ITEMS=1000');
+
+        try {
+            $this->expectException(AvroIOCollectionSizeException::class);
+            $this->decodeWith('{"type":"array","items":"null"}', self::zeroByteBlock(1001));
+        } finally {
+            $this->restoreMaxCollectionItems();
+        }
+    }
+
+    public function test_array_of_null_cumulative_across_blocks(): void
+    {
+        $io = new AvroStringIO();
+        $encoder = new AvroIOBinaryEncoder($io);
+        $encoder->writeLong(600);
+        $encoder->writeLong(600);
+        $encoder->writeLong(0);
+        putenv('AVRO_MAX_COLLECTION_ITEMS=1000');
+
+        try {
+            $this->expectException(AvroIOCollectionSizeException::class);
+            $this->decodeWith('{"type":"array","items":"null"}', $io);
+        } finally {
+            $this->restoreMaxCollectionItems();
+        }
+    }
+
+    public function test_array_of_null_negative_block_count(): void
+    {
+        putenv('AVRO_MAX_COLLECTION_ITEMS=1000');
+
+        try {
+            $this->expectException(AvroIOCollectionSizeException::class);
+            $this->decodeWith('{"type":"array","items":"null"}', self::zeroByteBlock(200000, true));
+        } finally {
+            $this->restoreMaxCollectionItems();
+        }
+    }
+
+    public function test_map_cumulative_limit_not_bypassed_by_duplicate_keys(): void
+    {
+        // Two blocks of 600 entries all reusing the same key: count($items) stays
+        // 1, but the cumulative pairs read (1200) must still exceed the 1000 cap.
+        $io = new AvroStringIO();
+        $encoder = new AvroIOBinaryEncoder($io);
+        foreach ([600, 600] as $blockCount) {
+            $encoder->writeLong($blockCount);
+            for ($i = 0; $i < $blockCount; $i++) {
+                $encoder->writeString('k'); // 1-byte key; null value is 0 bytes
+            }
+        }
+        $encoder->writeLong(0);
+        putenv('AVRO_MAX_COLLECTION_ITEMS=1000');
+
+        try {
+            $this->expectException(AvroIOCollectionSizeException::class);
+            $this->decodeWith('{"type":"map","values":"null"}', $io);
+        } finally {
+            $this->restoreMaxCollectionItems();
+        }
+    }
+
+    public function test_array_of_zero_length_fixed_exceeds_limit(): void
+    {
+        putenv('AVRO_MAX_COLLECTION_ITEMS=1000');
+
+        try {
+            $this->expectException(AvroIOCollectionSizeException::class);
+            $this->decodeWith('{"type":"array","items":{"type":"fixed","name":"empty","size":0}}', self::zeroByteBlock(2000));
+        } finally {
+            $this->restoreMaxCollectionItems();
+        }
+    }
+
+    public function test_array_of_all_null_record_exceeds_limit(): void
+    {
+        putenv('AVRO_MAX_COLLECTION_ITEMS=1000');
+
+        try {
+            $this->expectException(AvroIOCollectionSizeException::class);
+            $this->decodeWith(
+                '{"type":"array","items":{"type":"record","name":"R","fields":[{"name":"n","type":"null"}]}}',
+                self::zeroByteBlock(2000)
+            );
+        } finally {
+            $this->restoreMaxCollectionItems();
+        }
+    }
+
+    public function test_map_of_null_rejected_by_available_bytes(): void
+    {
+        // Map entries always carry a >= 1 byte key, so a huge map<null> is bounded
+        // by the bytes-remaining check.
+        $this->expectException(AvroException::class);
+        $this->decodeWith('{"type":"map","values":"null"}', self::zeroByteBlock(200000000));
+    }
+
+    public function test_invalid_env_override_falls_back_to_default(): void
+    {
+        // An invalid limit is ignored, so the default (10,000,000) still rejects
+        // the 200,000,000 exploit.
+        putenv('AVRO_MAX_COLLECTION_ITEMS=not-a-number');
+
+        try {
+            $this->expectException(AvroIOCollectionSizeException::class);
+            $this->decodeWith('{"type":"array","items":"null"}', self::zeroByteBlock(200000000));
+        } finally {
+            $this->restoreMaxCollectionItems();
+        }
+    }
+
+    public function test_env_override_raises_limit(): void
+    {
+        // Raising the limit above the count lets a large null array decode.
+        putenv('AVRO_MAX_COLLECTION_ITEMS=20000');
+
+        try {
+            $result = $this->decodeWith('{"type":"array","items":"null"}', self::zeroByteBlock(15000));
+            $this->assertCount(15000, $result);
+        } finally {
+            $this->restoreMaxCollectionItems();
+        }
+    }
+
+    public function test_non_zero_collection_bounded_by_structural_cap(): void
+    {
+        // A backed array<long> that passes the bytes check is still rejected once
+        // its count exceeds the (env-lowered) structural cap.
+        $io = new AvroStringIO();
+        $encoder = new AvroIOBinaryEncoder($io);
+        $encoder->writeLong(10);           // block count 10
+        for ($i = 0; $i < 10; $i++) {
+            $encoder->writeLong($i);       // 10 real longs
+        }
+        $encoder->writeLong(0);
+        putenv('AVRO_MAX_COLLECTION_ITEMS=5');
+
+        try {
+            $this->expectException(AvroIOCollectionSizeException::class);
+            $this->decodeWith('{"type":"array","items":"long"}', $io);
+        } finally {
+            $this->restoreMaxCollectionItems();
+        }
+    }
+
+    public function test_skip_array_of_null_respects_limit(): void
+    {
+        // Skipping a huge zero-byte array (e.g. during projection) is bounded.
+        $schema = AvroSchema::parse('{"type":"array","items":"null"}');
+        $io = self::zeroByteBlock(200000000);
+        $io->seek(0);
+        putenv('AVRO_MAX_COLLECTION_ITEMS=1000');
+
+        try {
+            $this->expectException(AvroIOCollectionSizeException::class);
+            AvroIODatumReader::skipData($schema, new AvroIOBinaryDecoder($io));
+        } finally {
+            $this->restoreMaxCollectionItems();
+        }
+    }
+
+    public function test_skip_array_of_null_negative_block_respects_limit(): void
+    {
+        // The negative (byte-sized) block form must also be bounded when skipping,
+        // so it cannot bypass the skip limit.
+        $schema = AvroSchema::parse('{"type":"array","items":"null"}');
+        $io = self::zeroByteBlock(200000000, true);
+        $io->seek(0);
+        putenv('AVRO_MAX_COLLECTION_ITEMS=1000');
+
+        try {
+            $this->expectException(AvroIOCollectionSizeException::class);
+            AvroIODatumReader::skipData($schema, new AvroIOBinaryDecoder($io));
+        } finally {
+            $this->restoreMaxCollectionItems();
+        }
+    }
+
     public static function validDurationLogicalTypes(): array
     {
         return [
@@ -420,6 +734,48 @@ public function test_field_default_value(
         }
     }
 
+    protected function restoreMaxCollectionItems(): void
+    {
+        if (false === $this->originalMaxCollectionItems) {
+            putenv('AVRO_MAX_COLLECTION_ITEMS');
+        } else {
+            putenv('AVRO_MAX_COLLECTION_ITEMS='.$this->originalMaxCollectionItems);
+        }
+    }
+
+    /**
+     * Builds one array/map block of `$count` zero-byte elements plus an end
+     * marker (a negative count is preceded by a block byte-size).
+     */
+    private static function zeroByteBlock(int $count, bool $negative = false): AvroStringIO
+    {
+        $io = new AvroStringIO();
+        $encoder = new AvroIOBinaryEncoder($io);
+        if ($negative) {
+            $encoder->writeLong(-$count);
+            $encoder->writeLong(0); // block byte-size
+        } else {
+            $encoder->writeLong($count);
+        }
+        $encoder->writeLong(0); // end-of-collection marker
+
+        return $io;
+    }
+
+    // An array/map block declares an element count; a malicious or truncated
+    // input can declare far more elements than the remaining bytes could hold.
+    // The count is validated against the bytes remaining before iterating, using
+    // the minimum on-wire size of the element schema (so 0-byte elements like
+    // null are not falsely rejected).
+    private function decodeWith(string $schemaJson, AvroStringIO $io): mixed
+    {
+        $schema = AvroSchema::parse($schemaJson);
+        $io->seek(0);
+        $reader = new AvroIODatumReader($schema);
+
+        return $reader->read(new AvroIOBinaryDecoder($io));
+    }
+
     /**
      * @param string $datum
      * @param string $expected