From c73212012379da0457d4d8bc3b3be739cf86a10c Mon Sep 17 00:00:00 2001 From: Paolo Insogna Date: Mon, 5 Oct 2026 15:35:47 +0200 Subject: [PATCH] fix: Correctly drop memory. Signed-off-by: Paolo Insogna --- docs/cpp.md | 2 + docs/js.md | 2 + docs/rust.md | 2 + parser/src/lib.rs | 25 +++-- parser/src/parse.rs | 36 +++--- parser/tests/issues.rs | 189 ++++++++++++++++++++++++++++++++ parser/wasm/test/issues.test.js | 55 ++++++++++ 7 files changed, 284 insertions(+), 27 deletions(-) create mode 100644 parser/tests/issues.rs diff --git a/docs/cpp.md b/docs/cpp.md index 73779a5..776e11d 100644 --- a/docs/cpp.md +++ b/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. diff --git a/docs/js.md b/docs/js.md index db9210b..d43df97 100644 --- a/docs/js.md +++ b/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 diff --git a/docs/rust.md b/docs/rust.md index 648d2fa..1ad19ae 100644 --- a/docs/rust.md +++ b/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 diff --git a/parser/src/lib.rs b/parser/src/lib.rs index 4f74349..f5f0b50 100644 --- a/parser/src/lib.rs +++ b/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]); diff --git a/parser/src/parse.rs b/parser/src/parse.rs index 842246d..80105a0 100644 --- a/parser/src/parse.rs +++ b/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; } } diff --git a/parser/tests/issues.rs b/parser/tests/issues.rs new file mode 100644 index 0000000..e287557 --- /dev/null +++ b/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 = const { Cell::new(false) }; + static LIVE_BYTES: Cell = 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; + } + } +} diff --git a/parser/wasm/test/issues.test.js b/parser/wasm/test/issues.test.js index a7daa17..3ee83c1 100644 --- a/parser/wasm/test/issues.test.js +++ b/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]) {