fix: Correctly drop memory.

docs/cpp.md

@@ -324,6 +324,8 @@ Parses `data` up to `limit` characters.
 
 It returns the number of consumed characters.
 
+After a parser error, subsequent calls return `0`, emit no events or callbacks, and do not copy or retain new input. The original error and any previously retained input remain available until reset or destruction. Call `milo_reset()` before reusing the parser.
+
 ### `void milo_set_active_events(Parser *parser, uint64_t value)`
 
 Sets the active event bitmask on the parser.

docs/js.md

@@ -298,6 +298,8 @@ Parses `data` up to `limit` characters.
 
 It returns the number of consumed characters.
 
+After a parser error, subsequent calls return `0`, emit no events or callbacks, and do not copy or retain new input. The original error and any previously retained input remain available until reset or destruction. Call `reset()` before reusing the parser.
+
 #### `reset(parser)`
 
 Resets a parser. The second parameters specifies if to also reset the

docs/rust.md

@@ -267,6 +267,8 @@ Parses `data` up to `limit` characters.
 
 It returns the number of consumed characters.
 
+After a parser error, subsequent calls return `0`, emit no events or callbacks, and do not copy or retain new input. The original error and any previously retained input remain available until reset or destruction. Call `reset()` before reusing the parser.
+
 #### `Parser::reset(&mut self, keep_parsed: bool)`
 
 Resets a parser. The second parameters specifies if to also reset the

parser/src/lib.rs

@@ -173,14 +173,7 @@ impl Parser {
     self.error_description[0] = 0;
     self.error_description_len = 0;
 
-    if self.unconsumed_len > 0 {
-      unsafe {
-        let _ = slice::from_raw_parts(self.unconsumed, self.unconsumed_len);
-      }
-
-      self.unconsumed = ptr::null();
-      self.unconsumed_len = 0;
-    }
+    self.clear_unconsumed();
 
     self.clear();
     self.skip_body = false;
@@ -189,6 +182,21 @@ impl Parser {
     }
   }
 
+  pub(crate) fn clear_unconsumed(&mut self) {
+    if self.unconsumed_len > 0 {
+      // SAFETY: Retained input is allocated as a boxed slice with exactly this
+      // length, and the parser owns it until replacement, reset, or destruction.
+      unsafe {
+        drop(Box::from_raw(ptr::slice_from_raw_parts_mut(
+          self.unconsumed as *mut c_uchar,
+          self.unconsumed_len,
+        )));
+      }
+    }
+    self.unconsumed = ptr::null();
+    self.unconsumed_len = 0;
+  }
+
   /// Clears all values about the message in the parser.
   pub fn clear(&mut self) {
     self.is_connect = false;
@@ -333,6 +341,7 @@ impl Parser {
 
 impl Drop for Parser {
   fn drop(&mut self) {
+    self.clear_unconsumed();
     if !self.events.is_null() {
       unsafe {
         let _ = Box::from_raw(self.events as *mut [u8; 65536]);

parser/src/parse.rs

@@ -29,9 +29,16 @@ impl Parser {
     let mut failed = false;
     let active_events = self.active_events | self.active_callbacks;
 
-    // If the self.is paused, this is a no-op
-    if self.paused {
+    // Error states cannot resume without reset. Do not aggregate or retain new
+    // input on retries, which would otherwise grow the pending buffer forever.
+    if self.state == STATE_ERROR {
+      self.position = 0;
+    }
+
+    // Paused and failed parsers consume nothing and produce an empty event batch.
+    if self.paused || self.state == STATE_ERROR {
       if active_events != 0 {
+        // SAFETY: The live parser owns the event buffer and its first byte.
         unsafe {
           *self.events = EVENT_END;
         }
@@ -1444,23 +1451,14 @@ impl Parser {
     self.parsed += consumed as u64;
 
     if self.manage_unconsumed {
-      unsafe {
-        // Drop any previous retained data
-        if unconsumed_len > 0 {
-          let _ = from_raw_parts(self.unconsumed, unconsumed_len);
-        }
-
-        // If less bytes were consumed than requested, copy the unconsumed portion in
-        // the self.for the next iteration
-        if consumed < limit {
-          let (ptr, len, _) = data.to_vec().into_raw_parts();
-
-          self.unconsumed = ptr;
-          self.unconsumed_len = len;
-        } else {
-          self.unconsumed = ptr::null();
-          self.unconsumed_len = 0;
-        }
+      self.clear_unconsumed();
+
+      // Use a boxed slice so the length also determines the allocation layout
+      // when releasing retained input; Vec capacity is not stored in the ABI.
+      if consumed < limit {
+        let retained = data.to_vec().into_boxed_slice();
+        self.unconsumed_len = retained.len();
+        self.unconsumed = Box::into_raw(retained) as *const c_uchar;
       }
     }
 

parser/tests/issues.rs

@@ -0,0 +1,189 @@
+use std::alloc::{GlobalAlloc, Layout, System};
+use std::cell::Cell;
+use std::sync::atomic::{AtomicUsize, Ordering};
+
+use milo_parser::{ERROR_NONE, Parser, STATE_ERROR};
+
+struct TrackingAllocator;
+
+static WATCHED_POINTER: AtomicUsize = AtomicUsize::new(0);
+static FREED_SIZE: AtomicUsize = AtomicUsize::new(0);
+
+thread_local! {
+  // Count only this test thread's allocations, excluding test-runner activity.
+  static TRACK_MEMORY: Cell<bool> = const { Cell::new(false) };
+  static LIVE_BYTES: Cell<isize> = const { Cell::new(0) };
+}
+
+// Keep the allocator in this integration-test binary so other suites cannot
+// interfere with the retained-buffer lifetime checks.
+#[global_allocator]
+static ALLOCATOR: TrackingAllocator = TrackingAllocator;
+
+// SAFETY: Every allocation and deallocation is forwarded unchanged to System.
+unsafe impl GlobalAlloc for TrackingAllocator {
+  unsafe fn alloc(&self, layout: Layout) -> *mut u8 {
+    // SAFETY: The caller supplies the valid allocation layout.
+    let pointer = unsafe { System.alloc(layout) };
+    if !pointer.is_null() && TRACK_MEMORY.try_with(Cell::get).unwrap_or(false) {
+      let _ = LIVE_BYTES.try_with(|bytes| bytes.set(bytes.get() + layout.size() as isize));
+    }
+    pointer
+  }
+
+  unsafe fn dealloc(&self, pointer: *mut u8, layout: Layout) {
+    if TRACK_MEMORY.try_with(Cell::get).unwrap_or(false) {
+      let _ = LIVE_BYTES.try_with(|bytes| bytes.set(bytes.get() - layout.size() as isize));
+    }
+    if WATCHED_POINTER
+      .compare_exchange(pointer as usize, 0, Ordering::SeqCst, Ordering::SeqCst)
+      .is_ok()
+    {
+      FREED_SIZE.store(layout.size(), Ordering::SeqCst);
+    }
+    // SAFETY: The pointer and layout are forwarded from the allocator caller.
+    unsafe { System.dealloc(pointer, layout) }
+  }
+}
+
+#[test]
+#[allow(non_snake_case)]
+fn issue_16__repeated_input_has_bounded_memory() {
+  for input in [
+    b"".as_slice(),
+    b"GET / HTTP/1.1\r\nX-Test: ",
+    b"GET / HTTP/1.1\r\nX-Test: \0",
+    // Declared body lengths must not cause proportional allocations.
+    b"POST / HTTP/1.1\r\nContent-Length: 9999999999999999999\r\n\r\n",
+    b"POST / HTTP/1.1\r\nTransfer-Encoding: chunked\r\n\r\nffffffffffffffff\r\n",
+  ] {
+    for managed in [false, true] {
+      TRACK_MEMORY.set(true);
+      let baseline = LIVE_BYTES.get();
+      for _ in 0..1024 {
+        let mut parser = Parser::new();
+        parser.manage_unconsumed = managed;
+        parser.parse(input.as_ptr(), input.len());
+
+        // Empty retries must not leak copies of a pending line or repeat errors.
+        let retained_len = parser.unconsumed_len;
+        let parsed = parser.parsed;
+        let state = parser.state;
+        let live = LIVE_BYTES.get();
+        for _ in 0..16 {
+          assert_eq!(parser.parse(b"".as_ptr(), 0), 0);
+          assert_eq!(parser.unconsumed_len, retained_len);
+          assert_eq!(parser.parsed, parsed);
+          assert_eq!(parser.state, state);
+          assert_eq!(LIVE_BYTES.get(), live);
+        }
+
+        parser.finish();
+        parser.reset(false);
+        assert_eq!(parser.error_code, ERROR_NONE);
+        assert!(parser.unconsumed.is_null());
+        parser.parse(input.as_ptr(), input.len());
+        drop(parser);
+        assert_eq!(LIVE_BYTES.get(), baseline, "input={input:?}, managed={managed}");
+      }
+      TRACK_MEMORY.set(false);
+    }
+  }
+}
+
+#[test]
+#[allow(non_snake_case)]
+fn issue_16__error_retries_do_not_accumulate_input() {
+  for managed in [false, true] {
+    let mut parser = Parser::new();
+    parser.manage_unconsumed = managed;
+    let invalid = b"GET / HTTP/1.1\r\nX-Test: \0";
+    parser.parse(invalid.as_ptr(), invalid.len());
+    assert_eq!(parser.state, STATE_ERROR);
+    let code = parser.error_code;
+    let description = parser.error_description_str().to_string();
+    let retained = parser.unconsumed;
+    let retained_len = parser.unconsumed_len;
+    let parsed = parser.parsed;
+
+    TRACK_MEMORY.set(true);
+    let baseline = LIVE_BYTES.get();
+    for _ in 0..1024 {
+      assert_eq!(parser.parse(invalid.as_ptr(), invalid.len()), 0);
+      assert_eq!(parser.state, STATE_ERROR);
+      assert_eq!(parser.error_code, code);
+      assert_eq!(parser.error_description_str(), description);
+      assert_eq!(parser.unconsumed_len, retained_len);
+      assert_eq!(parser.unconsumed, retained);
+      assert_eq!(parser.parsed, parsed);
+      assert_eq!(parser.position, 0);
+      assert_eq!(LIVE_BYTES.get(), baseline);
+    }
+    TRACK_MEMORY.set(false);
+
+    parser.reset(false);
+    let valid = b"GET / HTTP/1.1\r\n\r\n";
+    assert_eq!(parser.parse(valid.as_ptr(), valid.len()), valid.len());
+    assert_eq!(parser.error_code, ERROR_NONE);
+  }
+}
+
+#[test]
+#[allow(non_snake_case)]
+fn issue_16__release_retained_input() {
+  let mut parser = Parser::new();
+  parser.manage_unconsumed = true;
+  let prefix = b"GET / HTTP/1.1\r\nX-Test: ";
+  assert_eq!(parser.parse(prefix.as_ptr(), prefix.len()), 16);
+  assert_eq!(parser.unconsumed_len, 8);
+
+  // Replacing an incomplete header must free the previous allocation, even
+  // when no new input arrives.
+  for input in [b"a".as_slice(), b"", b"b"] {
+    let previous_len = parser.unconsumed_len;
+    FREED_SIZE.store(0, Ordering::SeqCst);
+    WATCHED_POINTER.store(parser.unconsumed as usize, Ordering::SeqCst);
+    assert_eq!(parser.parse(input.as_ptr(), input.len()), 0);
+    assert_eq!(WATCHED_POINTER.load(Ordering::SeqCst), 0);
+    assert_eq!(FREED_SIZE.load(Ordering::SeqCst), previous_len);
+    assert_eq!(parser.unconsumed_len, previous_len + input.len());
+  }
+
+  let previous_len = parser.unconsumed_len;
+  WATCHED_POINTER.store(parser.unconsumed as usize, Ordering::SeqCst);
+  let suffix = b"\r\n\r\n";
+  assert_eq!(parser.parse(suffix.as_ptr(), suffix.len()), previous_len + suffix.len());
+  assert_eq!(WATCHED_POINTER.load(Ordering::SeqCst), 0);
+  assert_eq!(FREED_SIZE.load(Ordering::SeqCst), previous_len);
+  assert!(parser.unconsumed.is_null());
+  assert_eq!(parser.unconsumed_len, 0);
+  assert_eq!(parser.error_code, ERROR_NONE);
+
+  // Reset and destruction must also release pending input, including after
+  // malformed input has moved the parser into its terminal error state.
+  for input in [prefix.as_slice(), b"GET / HTTP/1.1\r\nX-Test: \0"] {
+    for reset in [false, true] {
+      parser.reset(false);
+      parser.parse(input.as_ptr(), input.len());
+      assert!(parser.unconsumed_len > 0);
+      if input.ends_with(b"\0") {
+        assert_eq!(parser.state, STATE_ERROR);
+      }
+      let previous_len = parser.unconsumed_len;
+      WATCHED_POINTER.store(parser.unconsumed as usize, Ordering::SeqCst);
+      // Disabling retention must not lose ownership of an existing buffer.
+      parser.manage_unconsumed = false;
+      if reset {
+        parser.reset(false);
+        assert!(parser.unconsumed.is_null());
+        assert_eq!(parser.unconsumed_len, 0);
+      } else {
+        drop(parser);
+        parser = Parser::new();
+      }
+      assert_eq!(WATCHED_POINTER.load(Ordering::SeqCst), 0);
+      assert_eq!(FREED_SIZE.load(Ordering::SeqCst), previous_len);
+      parser.manage_unconsumed = true;
+    }
+  }
+}

parser/wasm/test/issues.test.js

@@ -2,6 +2,61 @@ import assert from 'node:assert/strict'
 import { it } from 'node:test'
 import { createParser, setup } from './helpers.js'
 
+it('issue-16 - error_retries_do_not_accumulate_input', () => {
+  for (const managed of [false, true]) {
+    for (const callbacks of [false, true]) {
+      let errors = 0
+      const milo = setup({
+        on_error () {
+          errors++
+        }
+      })
+      const parser = milo.create()
+      const invalid = Buffer.from('GET / HTTP/1.1\r\nX-Test: \0')
+      const valid = Buffer.from('GET / HTTP/1.1\r\n\r\n')
+      const ptr = milo.alloc(invalid.length)
+      try {
+        milo.setShouldManageUnconsumed(parser, managed)
+        milo.setActiveCallbacks(parser, callbacks ? milo.CALLBACK_ACTIVE_ON_ERROR : 0n)
+        milo.setActiveEvents(parser, callbacks ? 0n : milo.EVENT_ACTIVE_ON_ERROR)
+        new Uint8Array(milo.memory.buffer, ptr, invalid.length).set(invalid)
+        milo.parse(parser, ptr, invalid.length)
+        assert.equal(milo.getState(parser), milo.STATE_ERROR)
+        const code = milo.getErrorCode(parser)
+        const description = milo.getErrorDescription(parser)
+        const parsed = milo.getParsed(parser)
+        const view = new DataView(milo.memory.buffer)
+        const retained = view.getUint32(parser + milo.ParserFields.UNCONSUMED, true)
+        const retainedLength = view.getUint32(parser + milo.ParserFields.UNCONSUMED_LEN, true)
+        const events = view.getUint32(parser + milo.ParserFields.EVENTS, true)
+        assert.equal(view.getUint8(events), milo.EVENT_ERROR)
+        const memorySize = milo.memory.buffer.byteLength
+        for (let i = 0; i < 1024; i++) {
+          // Alternate empty retries with new malformed input on the failed parser.
+          assert.equal(milo.parse(parser, ptr, i % 2 ? invalid.length : 0), 0)
+          assert.equal(milo.getState(parser), milo.STATE_ERROR)
+          assert.equal(milo.getErrorCode(parser), code)
+          assert.equal(milo.getErrorDescription(parser), description)
+          assert.equal(milo.getParsed(parser), parsed)
+          assert.equal(milo.getPosition(parser), 0)
+          assert.equal(milo.memory.buffer.byteLength, memorySize)
+          assert.equal(view.getUint32(parser + milo.ParserFields.UNCONSUMED, true), retained)
+          assert.equal(view.getUint32(parser + milo.ParserFields.UNCONSUMED_LEN, true), retainedLength)
+          assert.equal(view.getUint8(events), milo.EVENT_END)
+          assert.equal(errors, callbacks ? 1 : 0)
+        }
+        milo.reset(parser, false)
+        new Uint8Array(milo.memory.buffer, ptr, valid.length).set(valid)
+        assert.equal(milo.parse(parser, ptr, valid.length), valid.length)
+        assert.equal(milo.getErrorCode(parser), milo.ERROR_NONE)
+      } finally {
+        milo.dealloc(ptr, invalid.length)
+        milo.destroy(parser)
+      }
+    }
+  }
+})
+
 it('issue-26 - preserve_events_before_error', () => {
   const suffix = 'HTTP/9.9 garbage\r\n\r\n'
   for (const chunked of [false, true]) {