consensus_encoding: implement batched allocation for WitnessDecoder

diff --git a/.cargo/mutants.toml b/.cargo/mutants.toml
index 455bc8e7..e6cfccb1 100644
--- a/.cargo/mutants.toml
+++ b/.cargo/mutants.toml
@@ -52,6 +52,7 @@ exclude_re = [
     "primitives/.* <impl Encoder for .*Encoder<'_>>::advance", # Replacing the return with true causes an infinite loop.
     "primitives/.* <impl Decoder for WitnessDecoder>::push_bytes", # Replacing == with != causes an infinite loop
     "primitives/.* WitnessDecoder::resize_if_needed", # Replacing *= with += still resizes the buffer making the mutant untestable.
+    "primitives/.* WitnessDecoder::reserve_batch", # Mutations cause an infinite loop
     "primitives/.* replace \\+ with \\* in MerkleNode::calculate_root", # Replacing + with * causes an infinite loop
     "primitives/.* replace == with != in MerkleNode::calculate_root", # Replacing == with != isn't caught unless alloc is disabled.
 
diff --git a/primitives/src/witness.rs b/primitives/src/witness.rs
index 33166416..76f116f9 100644
--- a/primitives/src/witness.rs
+++ b/primitives/src/witness.rs
@@ -26,6 +26,10 @@ use crate::prelude::{Box, Vec};
 #[cfg(doc)]
 use crate::TxIn;
 
+/// Maximum amount of memory (in bytes) to allocate at once when deserializing vectors.
+#[cfg(feature = "alloc")]
+const MAX_VECTOR_ALLOCATE: usize = 1_000_000;
+
 /// The Witness is the data used to unlock bitcoin since the [SegWit upgrade].
 ///
 /// Can be logically seen as an array of bytestrings, i.e. `Vec<Vec<u8>>`, and it is serialized on the wire
@@ -316,15 +320,26 @@ impl WitnessDecoder {
         }
     }
 
-    /// Resizes the content buffer if needed, doubling the size each time.
-    fn resize_if_needed(&mut self, required_len: usize) {
-        if required_len >= self.content.len() {
-            let mut new_len = self.content.len().max(1);
-            while new_len <= required_len {
-                new_len *= 2;
-            }
-            self.content.resize(new_len, 0);
+    /// Allocates buffer space in ~1MB batches
+    /// Returns buffer length (may be less than `required_len` !!)
+    fn reserve_batch(&mut self, required_len: usize) -> usize {
+        if required_len <= self.content.len() {
+            return self.content.len();
         }
+
+        let bytes_needed = required_len - self.content.len();
+        let available_capacity = self.content.capacity() - self.content.len();
+
+        if available_capacity == 0 {
+            let batch_size = bytes_needed.min(MAX_VECTOR_ALLOCATE);
+            self.content.reserve_exact(batch_size);
+        }
+
+        // Only extend up to current capacity to limit batch allocation
+        let can_extend = (self.content.capacity() - self.content.len()).min(bytes_needed);
+        let new_len = self.content.len() + can_extend;
+        self.content.resize(new_len, 0);
+        new_len
     }
 }
 
@@ -364,6 +379,7 @@ impl Decoder for WitnessDecoder {
             // and some overhead (e.g. P2WPKH witness is ~100 bytes),
             // without reallocating.
             let witness_index_space = witness_elements * 4;
+            // Initially the index space is at the front of the buffer then we rotate left in `end`.
             self.cursor = witness_index_space;
             self.content = alloc::vec![0u8; self.cursor + 128];
         }
@@ -387,17 +403,17 @@ impl Decoder for WitnessDecoder {
             // If we have some bytes to read, then reading element data.
             // Else we are reading the element's length.
             if let Some(bytes_to_read) = self.element_bytes_remaining {
-                let copy_len = bytes.len().min(bytes_to_read);
+                let required_len = self.cursor + bytes.len().min(bytes_to_read);
+                let actual_len = self.reserve_batch(required_len);
 
-                // Ensure we have enough space.
-                let required_len = self.cursor + copy_len;
-                self.resize_if_needed(required_len);
+                let available_space = actual_len.saturating_sub(self.cursor);
+                let can_copy = available_space.min(bytes.len()).min(bytes_to_read);
 
-                self.content[self.cursor..self.cursor + copy_len]
-                    .copy_from_slice(&bytes[..copy_len]);
-                self.cursor += copy_len;
-                *bytes = &bytes[copy_len..];
-                let remaining = bytes_to_read - copy_len;
+                self.content[self.cursor..self.cursor + can_copy]
+                    .copy_from_slice(&bytes[..can_copy]);
+                self.cursor += can_copy;
+                *bytes = &bytes[can_copy..];
+                let remaining = bytes_to_read - can_copy;
 
                 if remaining == 0 {
                     // Element complete, move to next element.
@@ -426,7 +442,7 @@ impl Decoder for WitnessDecoder {
                 // Re-encode the length back into the buffer.
                 let encoded_size = CompactSizeEncoder::encoded_size(element_length);
                 let required_len = self.cursor + encoded_size + element_length;
-                self.resize_if_needed(required_len);
+                self.reserve_batch(required_len);
                 let encoded_compact_size = crate::compact_size_encode(element_length);
                 self.content[self.cursor..self.cursor + encoded_size]
                     .copy_from_slice(&encoded_compact_size);
@@ -1609,4 +1625,21 @@ mod test {
         let mut slice = [0xFE, 0xCD, 0xAB].as_slice();
         let _ = decode_unchecked(&mut slice);
     }
+
+    #[cfg(feature = "alloc")]
+    #[test]
+    fn test_dos_protection() {
+        let mut encoded = Vec::new();
+        encoded.extend_from_slice(&[0xFE, 0x00, 0x09, 0x3D, 0x00]); // 4_000_000 (witness count)
+        encoded.extend_from_slice(&[0xFE, 0x00, 0x09, 0x3D, 0x00]); // 4_000_000 (1st element length)
+
+        let mut slice = encoded.as_slice();
+        let mut dec = WitnessDecoder::new();
+
+        assert!(dec.push_bytes(&mut slice).unwrap());
+
+        let allocated = dec.content.len();
+
+        assert!(allocated >= 16_000_000 && allocated < 17_500_000);
+    }
 }