diff --git a/lang/perl/lib/Avro/BinaryDecoder.pm b/lang/perl/lib/Avro/BinaryDecoder.pm
index d2bb65e14f7..0fa7f8e17db 100644
--- a/lang/perl/lib/Avro/BinaryDecoder.pm
+++ b/lang/perl/lib/Avro/BinaryDecoder.pm
@@ -124,7 +124,21 @@ sub skip_bytes {
     my $class = shift;
     my $reader = pop;
     my $size = decode_long($class, undef, undef, $reader);
-    $reader->seek($size, 0);
+    if ($size < 0) {
+        throw Avro::Schema::Error::Parse(
+            "Invalid negative bytes/string length: $size");
+    }
+    # Reject a declared length that exceeds the bytes actually remaining before
+    # skipping, mirroring decode_bytes: a seek past EOF silently succeeds on
+    # many handles, so without this a truncated input would later read zeros
+    # instead of failing.
+    _ensure_available($reader, $size, 1);
+    # Skip forward by $size bytes relative to the current position (SEEK_CUR);
+    # whence 0 (SEEK_SET) would incorrectly seek to the absolute offset $size.
+    # A failed seek means the reader cannot skip and is treated as fatal.
+    unless ($reader->seek($size, 1)) {
+        throw Avro::Schema::Error::Parse("Failed to skip $size bytes");
+    }
     return;
 }
 
@@ -132,10 +146,161 @@ sub decode_bytes {
     my $class = shift;
     my $reader = pop;
     my $size = decode_long($class, undef, undef, $reader);
-    $reader->read(my $buf, $size);
+    if ($size < 0) {
+        throw Avro::Schema::Error::Parse(
+            "Invalid negative bytes/string length: $size");
+    }
+    _ensure_available($reader, $size);
+    my $nread = $reader->read(my $buf, $size);
+    if (!defined $nread || $nread != $size) {
+        # A short read means the input is truncated; failing here avoids
+        # returning an under-sized buffer and misaligning the decoder.
+        throw Avro::Schema::Error::Parse(
+            "Expected $size bytes but read " . (defined $nread ? $nread : 0));
+    }
     return $buf;
 }
 
+## Reject a declared length that exceeds the bytes actually remaining before
+## allocating for it, to guard against an out-of-memory attack from a malicious
+## or truncated input. Only enforced for larger reads and when the reader can
+## report its size via seek/tell; smaller reads and non-seekable readers fall
+## through to a direct read.
+use constant _MAX_UNCHECKED_READ => 1024 * 1024;
+
+## Maximum number of zero-byte-encoded collection elements (e.g. an array of
+## nulls) to allocate from a single decode. 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 via the
+## AVRO_MAX_COLLECTION_ITEMS environment variable, or by setting
+## $Avro::BinaryDecoder::MAX_COLLECTION_ITEMS directly.
+our $DEFAULT_MAX_COLLECTION_ITEMS = 10_000_000;
+## 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.
+our $DEFAULT_MAX_COLLECTION_STRUCTURAL = (2 ** 31) - 1 - 8;
+
+## AVRO_MAX_COLLECTION_ITEMS, when set, caps both limits; otherwise zero-byte
+## elements use the tighter default and all collections use the structural cap.
+## Both may also be overridden directly (e.g. in tests).
+our $MAX_COLLECTION_ITEMS;
+our $MAX_COLLECTION_STRUCTURAL;
+if ( defined $ENV{AVRO_MAX_COLLECTION_ITEMS} && $ENV{AVRO_MAX_COLLECTION_ITEMS} =~ /\A[0-9]+\z/ ) {
+    $MAX_COLLECTION_ITEMS = $MAX_COLLECTION_STRUCTURAL = $ENV{AVRO_MAX_COLLECTION_ITEMS} + 0;
+}
+else {
+    $MAX_COLLECTION_ITEMS      = $DEFAULT_MAX_COLLECTION_ITEMS;
+    $MAX_COLLECTION_STRUCTURAL = $DEFAULT_MAX_COLLECTION_STRUCTURAL;
+}
+
+## Number of bytes still available to read, or undef when the reader cannot
+## report its size. Used to reject a declared length or collection block count
+## that exceeds the data actually available before allocating for it. Readers
+## that do not support seeking to the end (e.g. a streaming decompressor) return
+## undef, so the check is simply skipped for them.
+sub _bytes_remaining {
+    my ($reader) = @_;
+    my $current = eval { $reader->tell };
+    return if !defined $current || $current < 0;
+
+    # Attempt to seek to the end. Readers that cannot (e.g. a streaming
+    # decompressor) leave the position unchanged; skip the check for them.
+    my $moved = eval { $reader->seek(0, 2) };   # SEEK_END
+    return if !$moved;
+
+    # We are now at the end. Record the end offset, then ALWAYS restore the
+    # original position. A restore failure leaves the reader corrupted for
+    # subsequent decoding, so treat it as fatal rather than continuing.
+    my $end = eval { $reader->tell };
+    unless (eval { $reader->seek($current, 0) }) {   # SEEK_SET
+        throw Avro::Schema::Error::Parse(
+            "Failed to restore reader position after size check");
+    }
+    return if !defined $end || $end < 0;
+    return $end - $current;
+}
+
+sub _ensure_available {
+    my ($reader, $size, $force) = @_;
+    # Small reads normally skip the check to avoid per-value overhead, since the
+    # decode paths verify the actual read length. The skip paths use seek(),
+    # which can silently succeed past EOF and does not verify anything, so they
+    # pass $force to always validate against the bytes remaining.
+    return if !$force && $size <= _MAX_UNCHECKED_READ;
+    my $remaining = _bytes_remaining($reader);
+    return unless defined $remaining;
+    if ($size > $remaining) {
+        throw Avro::Schema::Error::Parse(
+            "Cannot read $size bytes, only $remaining remaining");
+    }
+    return;
+}
+
+## 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. A type that can encode to zero bytes (null) returns 0, which
+## disables the collection check for it (so an array of nulls is not falsely
+## rejected).
+sub _min_bytes_per_element {
+    my ($schema, $visited) = @_;
+    $visited ||= {};
+    my $type = $schema->type;
+    return 0 if $type eq 'null';
+    return 4 if $type eq 'float';
+    return 8 if $type eq 'double';
+    return $schema->size if $type eq 'fixed';
+    if ($type eq 'record' || $type eq 'error') {
+        my $id = "$schema"; # stringified reference as identity
+        return 0 if $visited->{$id};
+        $visited->{$id} = 1;
+        my $total = 0;
+        for my $field (@{ $schema->fields }) {
+            $total += _min_bytes_per_element($field->{type}, $visited);
+        }
+        delete $visited->{$id};
+        return $total;
+    }
+    # boolean, int, long, bytes, string, enum, union, array, map: >= 1 byte
+    # (a union encodes at least a 1-byte branch index).
+    return 1;
+}
+
+## Reject a collection (array or map) block whose declared element count could
+## not be backed by the bytes actually remaining, before iterating. Skipped when
+## the per-element minimum is zero or when the reader cannot report how many
+## bytes remain.
+sub _ensure_collection_available {
+    my ($reader, $existing, $count, $min_bytes) = @_;
+    return if $count <= 0;
+    my $total = $existing + $count;
+    # The structural cap bounds every array/map, including zero-byte element
+    # collections and non-seekable readers where the bytes check cannot run.
+    if ($total > $MAX_COLLECTION_STRUCTURAL) {
+        throw Avro::BinaryDecoder::Error::CollectionSize(
+            "Cannot read a collection of more than $MAX_COLLECTION_STRUCTURAL "
+          . "elements (declared $total)");
+    }
+    if ($min_bytes > 0) {
+        my $remaining = _bytes_remaining($reader);
+        # Compare via integer division rather than multiplying, so $count * $min_bytes
+        # cannot overflow/wrap (notably on 32-bit builds) and defeat the check.
+        if (defined $remaining && $count > int($remaining / $min_bytes)) {
+            throw Avro::Schema::Error::Parse(
+                "Collection claims $count elements with at least $min_bytes bytes each, "
+              . "but only $remaining bytes are available");
+        }
+    }
+    # Zero-byte elements (e.g. null) consume no input, so they cannot be bounded
+    # by the bytes remaining; additionally cap their cumulative count.
+    elsif ($total > $MAX_COLLECTION_ITEMS) {
+        throw Avro::BinaryDecoder::Error::CollectionSize(
+            "Cannot read a collection of more than $MAX_COLLECTION_ITEMS zero-byte "
+          . "elements (declared $total); raise "
+          . "\$Avro::BinaryDecoder::MAX_COLLECTION_ITEMS if this is legitimate");
+    }
+    return;
+}
+
 sub skip_string { &skip_bytes }
 sub decode_string {
     my $class = shift;
@@ -206,7 +371,14 @@ sub decode_enum {
     my ($writer_schema, $reader_schema, $reader) = @_;
     my $index = decode_int($class, @_);
 
-    my $w_data = $writer_schema->symbols->[$index];
+    my $symbols = $writer_schema->symbols;
+    ## A negative or out-of-range index is malformed; reject it before indexing
+    ## (Perl's negative indexing would otherwise select the wrong symbol).
+    if ($index < 0 || $index >= scalar @$symbols) {
+        throw Avro::Schema::Error::Parse(
+            "Enum symbol index $index out of range for " . scalar(@$symbols) . " symbols");
+    }
+    my $w_data = $symbols->[$index];
     ## 1.3.2 if the writer's symbol is not present in the reader's enum,
     ## then an error is signalled.
     throw Avro::Schema::Error::Mismatch("enum unknown")
@@ -215,27 +387,86 @@ sub decode_enum {
 }
 
 sub skip_block {
-    my $class = shift;
-    my ($reader, $block_content) = @_;
-    my $block_count = decode_long($class, undef, undef, $reader);
+    # Called as a plain function by skip_array/skip_map as
+    # skip_block($reader, $min_bytes, $coderef); it is not a method, so there is
+    # no leading class argument. decode_long ignores its (shifted) first argument
+    # and reads from the last, so pass __PACKAGE__ for it and $reader last.
+    my ($reader, $min_bytes, $block_content) = @_;
+    # Zero-byte elements loop with no per-item input, so bound them tightly;
+    # other elements consume bytes (bounded by EOF) so only the structural cap
+    # applies. Every collection is capped by the structural limit, so for
+    # zero-byte elements use whichever of the two limits is tighter.
+    my $limit = $MAX_COLLECTION_STRUCTURAL;
+    if ($min_bytes <= 0 && $MAX_COLLECTION_ITEMS < $limit) {
+        $limit = $MAX_COLLECTION_ITEMS;
+    }
+    my $block_count = decode_long(__PACKAGE__, undef, undef, $reader);
+    my $skipped = 0;
     while ($block_count) {
         if ($block_count < 0) {
-            $reader->seek($block_count, 0);
-            next;
+            # A byte-sized (negative-count) block still declares an element
+            # count; bound it cumulatively too, so a huge declared count with a
+            # small or zero block size cannot bypass the cap (e.g. zero-byte
+            # elements whose block size is 0).
+            my $count = -$block_count;
+            if ($skipped + $count > $limit) {
+                throw Avro::BinaryDecoder::Error::CollectionSize(
+                    "Cannot skip a collection of more than $limit "
+                  . "elements (declared @{[ $skipped + $count ]})");
+            }
+            $skipped += $count;
+            # A negative count is followed by a long block size in bytes, which
+            # lets the whole block be skipped without decoding each item. Skip
+            # forward by that many bytes relative to the current position
+            # (SEEK_CUR); whence 0 (SEEK_SET) would seek to an absolute (here
+            # nonsensical) offset. A failed seek is treated as fatal.
+            my $block_size = decode_long(__PACKAGE__, undef, undef, $reader);
+            if ($block_size < 0) {
+                # A negative block size would seek the reader backwards,
+                # risking an infinite loop or mis-decoding of a corrupt input.
+                throw Avro::Schema::Error::Parse(
+                    "Invalid negative block size: $block_size");
+            }
+            # Reject a block size that exceeds the bytes actually remaining
+            # before skipping, so a truncated input fails instead of seeking
+            # past EOF.
+            _ensure_available($reader, $block_size, 1);
+            # Reject a block size too small to contain the declared element
+            # count (at the element's minimum on-wire size), so a malformed
+            # size cannot leave the reader mid-element and misalign decoding.
+            if ($min_bytes > 0 && $count > int($block_size / $min_bytes)) {
+                throw Avro::Schema::Error::Parse(
+                    "Block size $block_size is too small for $count elements "
+                  . "of >= $min_bytes bytes");
+            }
+            unless ($reader->seek($block_size, 1)) {
+                throw Avro::Schema::Error::Parse(
+                    "Failed to skip block of $block_size bytes");
+            }
         }
         else {
+            # Bound the cumulative element count: skipping a huge block of
+            # zero-byte elements would otherwise loop unboundedly (a CPU
+            # exhaustion) even though it allocates nothing.
+            if ($skipped + $block_count > $limit) {
+                throw Avro::BinaryDecoder::Error::CollectionSize(
+                    "Cannot skip a collection of more than $limit "
+                  . "elements (declared @{[ $skipped + $block_count ]})");
+            }
+            $skipped += $block_count;
             for (1..$block_count) {
                 $block_content->();
             }
         }
-        $block_count = decode_long($class, undef, undef, $reader);
+        $block_count = decode_long(__PACKAGE__, undef, undef, $reader);
     }
 }
 
 sub skip_array {
     my $class = shift;
     my ($schema, $reader) = @_;
-    skip_block($reader, sub { $class->skip($schema->items, $reader) });
+    skip_block($reader, _min_bytes_per_element($schema->items),
+        sub { $class->skip($schema->items, $reader) });
 }
 
 ## 1.3.2 Arrays are encoded as a series of blocks. Each block consists of a
@@ -251,13 +482,19 @@ sub decode_array {
     my @array;
     my $writer_items = $writer_schema->items;
     my $reader_items = $reader_schema->items;
+    my $min_bytes = _min_bytes_per_element($writer_items);
     while ($block_count) {
         my $block_size;
         if ($block_count < 0) {
             $block_count = -$block_count;
             $block_size = decode_long($class, @_);
+            if ($block_size < 0) {
+                throw Avro::Schema::Error::Parse(
+                    "Invalid negative array block size: $block_size");
+            }
             ## XXX we can skip with $reader_schema?
         }
+        _ensure_collection_available($reader, scalar(@array), $block_count, $min_bytes);
         for (1..$block_count) {
             push @array, $class->decode(
                 writer_schema => $writer_items,
@@ -273,7 +510,8 @@ sub decode_array {
 sub skip_map {
     my $class = shift;
     my ($schema, $reader) = @_;
-    skip_block($reader, sub {
+    # Map entries always carry a >= 1 byte key, so pass a positive minimum.
+    skip_block($reader, 1 + _min_bytes_per_element($schema->values), sub {
         skip_string($class, $reader);
         $class->skip($schema->values, $reader);
     });
@@ -296,13 +534,27 @@ sub decode_map {
     my $block_count = decode_long($class, @_);
     my $writer_values = $writer_schema->values;
     my $reader_values = $reader_schema->values;
+    # A map key is a non-empty string (decode_map rejects empty keys below), so
+    # its minimum on-wire size is 2 bytes: a 1-byte length prefix plus at least
+    # 1 byte of key data, in addition to the value.
+    my $min_bytes = 2 + _min_bytes_per_element($writer_values);
+    # Track the number of decoded entries separately: scalar(keys %hash)
+    # undercounts when duplicate keys overwrite earlier entries, which would let
+    # a multi-block map exceed the cumulative caps without being rejected.
+    my $decoded = 0;
     while ($block_count) {
         my $block_size;
         if ($block_count < 0) {
             $block_count = -$block_count;
             $block_size = decode_long($class, @_);
+            if ($block_size < 0) {
+                throw Avro::Schema::Error::Parse(
+                    "Invalid negative map block size: $block_size");
+            }
             ## XXX we can skip with $reader_schema?
         }
+        _ensure_collection_available($reader, $decoded, $block_count, $min_bytes);
+        $decoded += $block_count;
         for (1..$block_count) {
             my $key = decode_string($class, @_);
             unless (defined $key && length $key) {
@@ -323,8 +575,12 @@ sub skip_union {
     my $class = shift;
     my ($schema, $reader) = @_;
     my $idx = decode_long($class, undef, undef, $reader);
-    my $union_schema = $schema->schemas->[$idx]
-        or throw Avro::Schema::Error::Parse("union union member");
+    my $schemas = $schema->schemas;
+    if ($idx < 0 || $idx >= scalar @$schemas) {
+        throw Avro::Schema::Error::Parse(
+            "Union branch index $idx out of range for " . scalar(@$schemas) . " branches");
+    }
+    my $union_schema = $schemas->[$idx];
     $class->skip($union_schema, $reader);
 }
 
@@ -335,7 +591,14 @@ sub decode_union {
     my $class = shift;
     my ($writer_schema, $reader_schema, $reader) = @_;
     my $idx = decode_long($class, @_);
-    my $union_schema = $writer_schema->schemas->[$idx];
+    my $schemas = $writer_schema->schemas;
+    ## A negative or out-of-range branch index is malformed; reject it before
+    ## indexing (skip_union performs the same check).
+    if ($idx < 0 || $idx >= scalar @$schemas) {
+        throw Avro::Schema::Error::Parse(
+            "Union branch index $idx out of range for " . scalar(@$schemas) . " branches");
+    }
+    my $union_schema = $schemas->[$idx];
     ## XXX TODO: schema resolution
     # The first schema in the reader's union that matches the selected writer's
     # union schema is recursively resolved against it. if none match, an error
@@ -350,7 +613,18 @@ sub decode_union {
 sub skip_fixed {
     my $class = shift;
     my ($schema, $reader) = @_;
-    $reader->seek($schema->size, 0);
+    # Reject a fixed size that exceeds the bytes actually remaining before
+    # skipping, mirroring skip_bytes: a seek past EOF silently succeeds on many
+    # handles, so without this a truncated input would later read zeros instead
+    # of failing.
+    _ensure_available($reader, $schema->size, 1);
+    # Skip the fixed-size payload relative to the current position (SEEK_CUR);
+    # whence 0 (SEEK_SET) would incorrectly seek to the absolute offset. A
+    # failed seek is treated as fatal.
+    unless ($reader->seek($schema->size, 1)) {
+        throw Avro::Schema::Error::Parse(
+            "Failed to skip " . $schema->size . " bytes");
+    }
 }
 
 ## 1.3.2 Fixed instances are encoded using the number of bytes declared in the
@@ -380,7 +654,18 @@ sub unsigned_varint {
     my $more;
     my $shift = 0;
     do {
-        $reader->read(my $buf, 1);
+        # A 64-bit value uses at most 10 bytes (shifts 0..63); reject an overlong
+        # varint rather than reading an unbounded continuation chain and letting
+        # $shift overflow into a garbage value.
+        if ($shift >= 70) {
+            throw Avro::Schema::Error::Parse("Varint is too long");
+        }
+        my $got = $reader->read(my $buf, 1);
+        # A short read (EOF) would otherwise make `ord $buf` == 0 and silently
+        # decode truncated input as a valid 0 byte; treat it as a parse error.
+        if (!$got) {
+            throw Avro::Schema::Error::Parse("Unexpected end of input while reading varint");
+        }
         my $byte = ord $buf;
         my $value = $byte & 0x7F;
         $int |= $value << $shift;
@@ -390,4 +675,9 @@ sub unsigned_varint {
     return $int;
 }
 
+package Avro::BinaryDecoder::Error::CollectionSize;
+use parent -norequire, 'Error::Simple';
+
+package Avro::BinaryDecoder;
+
 1;
diff --git a/lang/perl/t/03_bin_decode.t b/lang/perl/t/03_bin_decode.t
index e5da35d9983..1aadfad79f1 100644
--- a/lang/perl/t/03_bin_decode.t
+++ b/lang/perl/t/03_bin_decode.t
@@ -26,6 +26,15 @@ use Test::Exception;
 
 use_ok 'Avro::BinaryDecoder';
 
+## BinaryDecoder initializes MAX_COLLECTION_ITEMS / MAX_COLLECTION_STRUCTURAL
+## from AVRO_MAX_COLLECTION_ITEMS at load time. Reset both to their defaults so
+## the whole suite is deterministic regardless of the environment; individual
+## tests still `local`-override them where they exercise a specific limit.
+$Avro::BinaryDecoder::MAX_COLLECTION_ITEMS =
+    $Avro::BinaryDecoder::DEFAULT_MAX_COLLECTION_ITEMS;
+$Avro::BinaryDecoder::MAX_COLLECTION_STRUCTURAL =
+    $Avro::BinaryDecoder::DEFAULT_MAX_COLLECTION_STRUCTURAL;
+
 ## spec examples
 {
     my $enc = "\x06\x66\x6f\x6f";
@@ -84,6 +93,68 @@ EOJ
     is $dec, "a", "Binary_Encodings.Complex_Types.Unions-a";
 }
 
+## union and enum index bounds
+{
+    my $union = Avro::Schema->parse(q(["string","null"]));
+    # Index 2 (zig-zag int 0x04) is out of range for a 2-branch union.
+    open my $reader, '<', \"\x04" or die "Can't open memory file: $!";
+    throws_ok {
+        Avro::BinaryDecoder->decode(
+            writer_schema => $union,
+            reader_schema => $union,
+            reader        => $reader,
+        );
+    } 'Avro::Schema::Error::Parse', "union branch index out of range is rejected";
+
+    # Index -1 (zig-zag int 0x01) must not wrap to a valid branch.
+    open $reader, '<', \"\x01" or die "Can't open memory file: $!";
+    throws_ok {
+        Avro::BinaryDecoder->decode(
+            writer_schema => $union,
+            reader_schema => $union,
+            reader        => $reader,
+        );
+    } 'Avro::Schema::Error::Parse', "negative union branch index is rejected";
+
+    my $enum = Avro::Schema->parse(
+        q({ "type": "enum", "name": "e", "symbols": [ "a", "b" ] }));
+    # Index 9 (zig-zag int 0x12) is out of range for a 2-symbol enum.
+    open $reader, '<', \"\x12" or die "Can't open memory file: $!";
+    throws_ok {
+        Avro::BinaryDecoder->decode(
+            writer_schema => $enum,
+            reader_schema => $enum,
+            reader        => $reader,
+        );
+    } 'Avro::Schema::Error::Parse', "enum symbol index out of range is rejected";
+
+    # Index -1 (zig-zag int 0x01) must not wrap to a valid symbol.
+    open $reader, '<', \"\x01" or die "Can't open memory file: $!";
+    throws_ok {
+        Avro::BinaryDecoder->decode(
+            writer_schema => $enum,
+            reader_schema => $enum,
+            reader        => $reader,
+        );
+    } 'Avro::Schema::Error::Parse', "negative enum symbol index is rejected";
+}
+
+## overlong varint
+{
+    # A 64-bit value uses at most 10 bytes; an 11th byte (10 continuation bytes 0x80 then a terminator) is a
+    # malformed overlong varint and must be rejected.
+    my $long = Avro::Schema->parse(q({ "type": "long" }));
+    my $data = ("\x80" x 10) . "\x01";
+    open my $reader, '<', \$data or die "Can't open memory file: $!";
+    throws_ok {
+        Avro::BinaryDecoder->decode(
+            writer_schema => $long,
+            reader_schema => $long,
+            reader        => $reader,
+        );
+    } 'Avro::Schema::Error::Parse', "overlong varint is rejected";
+}
+
 ## enum schema resolution
 {
 
@@ -255,4 +326,362 @@ EOP
     is $dec->{one}[0], 1.0, "kind of dumb test";
 }
 
+## A bytes/string value declares a length prefix; a malicious or truncated
+## input can declare far more bytes than actually exist. On a seekable reader
+## that is rejected before allocating for it.
+{
+    my $bytes_schema = Avro::Schema->parse(q({ "type": "bytes" }));
+
+    ## A length prefix declaring 100 MiB, with no payload following it.
+    my $enc = '';
+    Avro::BinaryEncoder->encode(
+        schema  => Avro::Schema->parse(q({ "type": "long" })),
+        data    => 100 * 1024 * 1024,
+        emit_cb => sub { $enc .= ${ $_[0] } },
+    );
+    open my $reader, '<', \$enc or die "Can't open memory file: $!";
+    throws_ok {
+        Avro::BinaryDecoder->decode(
+            writer_schema => $bytes_schema,
+            reader_schema => $bytes_schema,
+            reader        => $reader,
+        );
+    } qr/Cannot read/, "oversized bytes length is rejected before allocating";
+
+    ## A well-formed value above the check threshold whose data is present
+    ## still decodes.
+    my $payload = 'x' x (2 * 1024 * 1024);
+    my $enc2 = '';
+    Avro::BinaryEncoder->encode(
+        schema  => $bytes_schema,
+        data    => $payload,
+        emit_cb => sub { $enc2 .= ${ $_[0] } },
+    );
+    open my $reader2, '<', \$enc2 or die "Can't open memory file: $!";
+    my $dec = Avro::BinaryDecoder->decode(
+        writer_schema => $bytes_schema,
+        reader_schema => $bytes_schema,
+        reader        => $reader2,
+    );
+    is $dec, $payload, "within-limit bytes value still decodes";
+}
+
+## 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).
+sub encode_long {
+    my ($value) = @_;
+    my $enc = '';
+    Avro::BinaryEncoder->encode(
+        schema  => Avro::Schema->parse(q({ "type": "long" })),
+        data    => $value,
+        emit_cb => sub { $enc .= ${ $_[0] } },
+    );
+    return $enc;
+}
+
+{
+    my $array_schema = Avro::Schema->parse(q({ "type": "array", "items": "long" }));
+    my $enc = encode_long(1_000_000); # 1,000,000 longs, no element data
+    open my $reader, '<', \$enc or die "Can't open memory file: $!";
+    throws_ok {
+        Avro::BinaryDecoder->decode(
+            writer_schema => $array_schema,
+            reader_schema => $array_schema,
+            reader        => $reader,
+        );
+    } qr/Collection claims/, "oversized array count is rejected before iterating";
+}
+
+{
+    my $map_schema = Avro::Schema->parse(q({ "type": "map", "values": "long" }));
+    my $enc = encode_long(1_000_000);
+    open my $reader, '<', \$enc or die "Can't open memory file: $!";
+    throws_ok {
+        Avro::BinaryDecoder->decode(
+            writer_schema => $map_schema,
+            reader_schema => $map_schema,
+            reader        => $reader,
+        );
+    } qr/Collection claims/, "oversized map count is rejected before iterating";
+}
+
+{
+    ## An array of nulls: null elements occupy zero bytes, so a large count is
+    ## legitimate and must not be rejected.
+    my $array_schema = Avro::Schema->parse(q({ "type": "array", "items": "null" }));
+    my $count = 100_000;
+    my $enc = encode_long($count) . encode_long(0); # one block + end marker
+    open my $reader, '<', \$enc or die "Can't open memory file: $!";
+    my $dec = Avro::BinaryDecoder->decode(
+        writer_schema => $array_schema,
+        reader_schema => $array_schema,
+        reader        => $reader,
+    );
+    is scalar(@$dec), $count, "array of nulls is not falsely rejected";
+}
+
+## Zero-byte elements (null, all-null records) consume no input, so the
+## bytes-remaining check cannot bound them. A huge declared block count of such
+## elements is capped instead. (Fixed is not zero-byte here: this SDK's schema
+## parser requires a positive Fixed size.)
+sub decode_zero_byte_array {
+    my ($items_schema, $enc) = @_;
+    my $schema = Avro::Schema->parse(qq({ "type": "array", "items": $items_schema }));
+    open my $reader, '<', \$enc or die "Can't open memory file: $!";
+    return Avro::BinaryDecoder->decode(
+        writer_schema => $schema,
+        reader_schema => $schema,
+        reader        => $reader,
+    );
+}
+
+{
+    ## The reported exploit: a ~6 byte payload declaring 200,000,000 nulls is
+    ## rejected by the default limit, before allocating. Pin the limit to the
+    ## default so the test is deterministic even if AVRO_MAX_COLLECTION_ITEMS was
+    ## set in the environment when the module was loaded.
+    local $Avro::BinaryDecoder::MAX_COLLECTION_ITEMS =
+        $Avro::BinaryDecoder::DEFAULT_MAX_COLLECTION_ITEMS;
+    throws_ok {
+        decode_zero_byte_array('"null"', encode_long(200_000_000) . encode_long(0));
+    } qr/more than \d+ zero-byte/, "array of 200M nulls rejected by default limit";
+}
+
+{
+    ## Within a lowered limit, a legitimate small null array still decodes.
+    local $Avro::BinaryDecoder::MAX_COLLECTION_ITEMS = 1000;
+    my $dec = decode_zero_byte_array('"null"', encode_long(1000) . encode_long(0));
+    is scalar(@$dec), 1000, "array of nulls within configured limit still reads";
+}
+
+{
+    ## INT64_MIN as a block count is the pathological negation case. Negating it
+    ## yields 2**63, which the cap rejects. INT64_MIN zig-zag encodes as the
+    ## 10-byte varint below, followed by a block byte-size (0).
+    my $int64_min = "\xFF\xFF\xFF\xFF\xFF\xFF\xFF\xFF\xFF\x01";
+    throws_ok {
+        decode_zero_byte_array('"null"', $int64_min . encode_long(0));
+    } qr/Cannot read a collection|zero-byte|Invalid/, "INT64_MIN array block count rejected";
+}
+
+{
+    local $Avro::BinaryDecoder::MAX_COLLECTION_ITEMS = 1000;
+    throws_ok {
+        decode_zero_byte_array('"null"', encode_long(1001) . encode_long(0));
+    } qr/more than 1000 zero-byte/, "array of nulls over configured limit rejected";
+}
+
+{
+    ## Cumulative across blocks: two blocks of 600 (1200 > 1000).
+    local $Avro::BinaryDecoder::MAX_COLLECTION_ITEMS = 1000;
+    throws_ok {
+        decode_zero_byte_array('"null"', encode_long(600) . encode_long(600) . encode_long(0));
+    } qr/more than 1000 zero-byte/, "cumulative null blocks over limit rejected";
+}
+
+{
+    ## A negative block count (abs value + block byte-size) is normalized and
+    ## must still be bounded.
+    local $Avro::BinaryDecoder::MAX_COLLECTION_ITEMS = 1000;
+    throws_ok {
+        decode_zero_byte_array('"null"', encode_long(-200_000) . encode_long(0));
+    } qr/more than 1000 zero-byte/, "negative null block count rejected";
+}
+
+{
+    ## A record whose only field is null encodes to zero bytes.
+    local $Avro::BinaryDecoder::MAX_COLLECTION_ITEMS = 1000;
+    throws_ok {
+        decode_zero_byte_array('{ "type": "record", "name": "R", "fields": [{"name":"n","type":"null"}] }',
+            encode_long(2000) . encode_long(0));
+    } qr/more than 1000 zero-byte/, "array of all-null records over limit rejected";
+}
+
+{
+    ## A huge map<null> is bounded by the bytes-remaining check (each entry has a
+    ## >= 1 byte key), not the zero-byte cap.
+    my $map_schema = Avro::Schema->parse(q({ "type": "map", "values": "null" }));
+    my $enc = encode_long(200_000_000);
+    open my $reader, '<', \$enc or die "Can't open memory file: $!";
+    throws_ok {
+        Avro::BinaryDecoder->decode(
+            writer_schema => $map_schema,
+            reader_schema => $map_schema,
+            reader        => $reader,
+        );
+    } qr/Collection claims/, "oversized map<null> rejected by available-bytes check";
+}
+
+{
+    ## Skipping a huge array<null> writer field absent from the reader schema is
+    ## bounded (a CPU exhaustion otherwise), via skip_block.
+    local $Avro::BinaryDecoder::MAX_COLLECTION_ITEMS = 1000;
+    my $w_schema = Avro::Schema->parse(<<'EOJ');
+      { "type": "record", "name": "test",
+        "fields" : [
+            {"name": "arr", "type": {"type": "array", "items": "null"}},
+            {"name": "a", "type": "long"} ]}
+EOJ
+    my $r_schema = Avro::Schema->parse(<<'EOJ');
+      { "type": "record", "name": "test",
+        "fields" : [ {"name": "a", "type": "long"} ]}
+EOJ
+    # Hand-crafted: array block of 2000 nulls + end marker, then a = 42.
+    my $enc = encode_long(2000) . encode_long(0) . encode_long(42);
+    open my $reader, '<', \$enc or die "Can't open memory file: $!";
+    throws_ok {
+        Avro::BinaryDecoder->decode(
+            writer_schema => $w_schema,
+            reader_schema => $r_schema,
+            reader        => $reader,
+        );
+    } qr/more than 1000/, "skipping oversized array<null> field is bounded";
+}
+
+{
+    ## The zero-byte cap raises the dedicated exception class.
+    local $Avro::BinaryDecoder::MAX_COLLECTION_ITEMS = 1000;
+    my $err;
+    eval {
+        decode_zero_byte_array('"null"', encode_long(2000) . encode_long(0));
+        1;
+    } or $err = $@;
+    isa_ok $err, 'Avro::BinaryDecoder::Error::CollectionSize',
+        "zero-byte cap throws CollectionSize";
+}
+
+{
+    ## Non-zero-byte collections are bounded by a structural cap in addition to
+    ## the bytes-remaining check. A small backed array<long> that passes the
+    ## bytes check is still rejected once its count exceeds the structural limit.
+    local $Avro::BinaryDecoder::MAX_COLLECTION_STRUCTURAL = 5;
+    my $schema = Avro::Schema->parse(q({ "type": "array", "items": "long" }));
+    my $enc = '';
+    Avro::BinaryEncoder->encode(
+        schema  => $schema,
+        data    => [ 0 .. 9 ],   # 10 real longs, enough bytes to pass the byte check
+        emit_cb => sub { $enc .= ${ $_[0] } },
+    );
+    open my $reader, '<', \$enc or die "Can't open memory file: $!";
+    throws_ok {
+        Avro::BinaryDecoder->decode(
+            writer_schema => $schema,
+            reader_schema => $schema,
+            reader        => $reader,
+        );
+    } qr/more than 5 elements/, "non-zero collection bounded by structural cap";
+}
+
+{
+    ## A backed array<long> within the structural cap still decodes.
+    local $Avro::BinaryDecoder::MAX_COLLECTION_STRUCTURAL = 100;
+    my $schema = Avro::Schema->parse(q({ "type": "array", "items": "long" }));
+    my $enc = '';
+    Avro::BinaryEncoder->encode(
+        schema  => $schema,
+        data    => [ 0 .. 9 ],
+        emit_cb => sub { $enc .= ${ $_[0] } },
+    );
+    open my $reader, '<', \$enc or die "Can't open memory file: $!";
+    my $dec = Avro::BinaryDecoder->decode(
+        writer_schema => $schema,
+        reader_schema => $schema,
+        reader        => $reader,
+    );
+    is_deeply $dec, [ 0 .. 9 ], "backed array within structural cap still reads";
+}
+
+## Skipping writer fields absent from the reader schema must advance the reader
+## by the correct relative amount (SEEK_CUR), including fixed and bytes fields.
+{
+    my $w_schema = Avro::Schema->parse(<<'EOJ');
+      { "type": "record", "name": "test",
+        "fields" : [
+            {"name": "f", "type": {"type": "fixed", "name": "fix8", "size": 8}},
+            {"name": "b", "type": "bytes"},
+            {"name": "a", "type": "long"} ]}
+EOJ
+    my $r_schema = Avro::Schema->parse(<<'EOJ');
+      { "type": "record", "name": "test",
+        "fields" : [ {"name": "a", "type": "long"} ]}
+EOJ
+    my $data = { f => "01234567", b => "payload", a => 42 };
+    my $enc = '';
+    Avro::BinaryEncoder->encode(
+        schema  => $w_schema,
+        data    => $data,
+        emit_cb => sub { $enc .= ${ $_[0] } },
+    );
+    open my $reader, '<', \$enc or die "Cannot open memory file: $!";
+    my $dec = Avro::BinaryDecoder->decode(
+        writer_schema => $w_schema,
+        reader_schema => $r_schema,
+        reader        => $reader,
+    );
+    is $dec->{a}, 42, "long read correctly after skipping fixed and bytes fields";
+    ok ! exists $dec->{f}, "skipped fixed field absent";
+    ok ! exists $dec->{b}, "skipped bytes field absent";
+}
+
+## Skipping writer array/map fields absent from the reader schema exercises
+## skip_block, which is called as a plain function (not a method) by
+## skip_array/skip_map.
+{
+    my $w_schema = Avro::Schema->parse(<<'EOJ');
+      { "type": "record", "name": "test",
+        "fields" : [
+            {"name": "arr", "type": {"type": "array", "items": "long"}},
+            {"name": "map", "type": {"type": "map", "values": "string"}},
+            {"name": "a", "type": "long"} ]}
+EOJ
+    my $r_schema = Avro::Schema->parse(<<'EOJ');
+      { "type": "record", "name": "test",
+        "fields" : [ {"name": "a", "type": "long"} ]}
+EOJ
+    my $data = { arr => [ 1, 2, 3 ], map => { x => "y", z => "w" }, a => 99 };
+    my $enc = '';
+    Avro::BinaryEncoder->encode(
+        schema  => $w_schema,
+        data    => $data,
+        emit_cb => sub { $enc .= ${ $_[0] } },
+    );
+    open my $reader, '<', \$enc or die "Cannot open memory file: $!";
+    my $dec = Avro::BinaryDecoder->decode(
+        writer_schema => $w_schema,
+        reader_schema => $r_schema,
+        reader        => $reader,
+    );
+    is $dec->{a}, 99, "long read correctly after skipping array and map fields";
+    ok ! exists $dec->{arr}, "skipped array field absent";
+    ok ! exists $dec->{map}, "skipped map field absent";
+}
+
+## A skipped bytes field whose declared length far exceeds the remaining data
+## must be rejected rather than silently seeking past EOF.
+{
+    my $w_schema = Avro::Schema->parse(<<'EOJ');
+      { "type": "record", "name": "test",
+        "fields" : [
+            {"name": "b", "type": "bytes"},
+            {"name": "a", "type": "long"} ]}
+EOJ
+    my $r_schema = Avro::Schema->parse(<<'EOJ');
+      { "type": "record", "name": "test",
+        "fields" : [ {"name": "a", "type": "long"} ]}
+EOJ
+    # Declare a bytes length of 100 MiB but provide no data for it.
+    my $enc = encode_long(100 * 1024 * 1024);
+    open my $reader, '<', \$enc or die "Cannot open memory file: $!";
+    throws_ok {
+        Avro::BinaryDecoder->decode(
+            writer_schema => $w_schema,
+            reader_schema => $r_schema,
+            reader        => $reader,
+        );
+    } qr/Cannot read/, "oversized skipped bytes length is rejected";
+}
+
 done_testing;