fix: guard callback replay reentrancy

docs/cpp.md

@@ -34,6 +34,8 @@ Callbacks are dispatched only when the corresponding `CALLBACK_ACTIVE_*` flag is
 
 Callbacks are disabled by default.
 
+Callback delivery uses an immutable snapshot of the current event batch, callback table, and activation mask. While that batch is being replayed, Milo functions that mutate the same parser are ignored and nested `milo_parse()` calls return `0`. Calling `milo_destroy()` is deferred until replay ends and stops delivery of the remaining callbacks in that batch. Direct field writes during callbacks remain unsupported.
+
 ## Constants
 
 The file `milo.h` defines several constants (`*` is used to denote a family prefix):

docs/js.md

@@ -22,6 +22,8 @@ Callbacks are dispatched only when enabled with `setActiveCallbacks`.
 
 Callbacks are disabled by default.
 
+Callback delivery uses an immutable snapshot of the current event batch and activation mask. While that batch is being replayed, operations that mutate the same parser are ignored and nested `parse()` calls return `0`. Calling `destroy()` is deferred until replay ends and stops delivery of the remaining callbacks in that batch. JavaScript callback exceptions are captured and rethrown only after Rust has cleaned up replay state.
+
 ### Constants
 
 The module exports several constants (`*` is used to denote a family prefix):

docs/rust.md

@@ -20,6 +20,8 @@ Callbacks are dispatched only when the corresponding `CALLBACK_ACTIVE_*` flag is
 
 Callbacks are disabled by default.
 
+Callback delivery uses an immutable snapshot of the current event batch, callback table, and activation mask. While that batch is being replayed, methods that mutate the same parser are ignored and nested `parse()` calls return `0`. FFI destruction is deferred until replay ends and stops delivery of the remaining callbacks in that batch.
+
 ## Constants
 
 The crate exports several constants (`*` is used to denote a family prefix):

macros/src/native.rs

@@ -18,30 +18,30 @@ pub fn generate_callbacks(callbacks: &[String]) -> TokenStream {
     if callback_name == "on_error" {
       quote! {
         #event_const => {
-          let at = unsafe { core::ptr::read_unaligned(self.events.add(cursor + 1) as *const u32) }.to_le() as usize;
-          if self.active_callbacks & #active_const != 0 {
-            (self.callbacks.#callback)(self, at, 0);
+          let at = unsafe { core::ptr::read_unaligned(events.as_ptr().add(cursor + 1) as *const u32) }.to_le() as usize;
+          if active_callbacks & #active_const != 0 {
+            (callbacks.#callback)(self, at, 0);
           }
           cursor += EVENT_ERROR_SIZE;
         }
       }
     } else if callback_name == "on_headers" {
       quote! {
         #event_const => {
-          let at = unsafe { core::ptr::read_unaligned(self.events.add(cursor + 1) as *const u32) }.to_le() as usize;
-          if self.active_callbacks & #active_const != 0 {
-            (self.callbacks.#callback)(self, at, 0);
+          let at = unsafe { core::ptr::read_unaligned(events.as_ptr().add(cursor + 1) as *const u32) }.to_le() as usize;
+          if active_callbacks & #active_const != 0 {
+            (callbacks.#callback)(self, at, 0);
           }
           cursor += EVENT_METADATA_SIZE;
         }
       }
     } else {
       quote! {
         #event_const => {
-          let at = unsafe { core::ptr::read_unaligned(self.events.add(cursor + 1) as *const u32) }.to_le() as usize;
-          let len = unsafe { core::ptr::read_unaligned(self.events.add(cursor + 5) as *const u32) }.to_le() as usize;
-          if self.active_callbacks & #active_const != 0 {
-            (self.callbacks.#callback)(self, at, len);
+          let at = unsafe { core::ptr::read_unaligned(events.as_ptr().add(cursor + 1) as *const u32) }.to_le() as usize;
+          let len = unsafe { core::ptr::read_unaligned(events.as_ptr().add(cursor + 5) as *const u32) }.to_le() as usize;
+          if active_callbacks & #active_const != 0 {
+            (callbacks.#callback)(self, at, len);
           }
           cursor += EVENT_RANGE_SIZE;
         }
@@ -72,17 +72,36 @@ pub fn generate_callbacks(callbacks: &[String]) -> TokenStream {
     #[cfg(not(target_family = "wasm"))]
     impl Parser {
       #[inline]
-      fn invoke_callbacks(&mut self) {
+      fn invoke_callbacks(&mut self, event_len: usize) {
+        if self.callback_replay || event_len == 0 || event_len > EVENTS_BUFFER_SIZE {
+          return;
+        }
+
+        let events = unsafe { core::slice::from_raw_parts(self.events, event_len) }.to_vec();
+        let active_callbacks = self.active_callbacks;
+        let callbacks = self.callbacks.clone();
+        let replay_flag = core::ptr::addr_of_mut!(self.callback_replay);
+        unsafe { core::ptr::write_volatile(replay_flag, true); }
+
+        struct ReplayGuard(*mut bool);
+        impl Drop for ReplayGuard {
+          fn drop(&mut self) { unsafe { core::ptr::write_volatile(self.0, false); } }
+        }
+        let _guard = ReplayGuard(replay_flag);
         let mut cursor = 0usize;
 
-        loop {
-          let event_type = unsafe { *self.events.add(cursor) };
+        while cursor < events.len() {
+          let event_type = events[cursor];
 
           match event_type {
             EVENT_END => break,
             #(#replay_arms)*
             _ => break,
           }
+
+          if unsafe { core::ptr::read_volatile(core::ptr::addr_of!(self.destroy_pending)) } {
+            break;
+          }
         }
       }
     }

macros/src/parser_fields.rs

@@ -46,6 +46,8 @@ struct ParserStub {
   unconsumed: WasmPointer,
   unconsumed_len: WasmUsize,
   error_description_len: u8,
+  callback_replay: bool,
+  destroy_pending: bool,
   events: WasmPointer,
 }
 

macros/src/wasm.rs

@@ -24,10 +24,10 @@ pub fn generate_callbacks(callbacks: &[String]) -> TokenStream {
 
       quote! {
         #event_const => {
-          let at = unsafe { core::ptr::read_unaligned(self.events.add(cursor + 1) as *const u32) }.to_le() as usize;
-          let len = unsafe { core::ptr::read_unaligned(self.events.add(cursor + 5) as *const u32) }.to_le() as usize;
-          if self.active_callbacks & #active_const != 0 {
-            unsafe { #callback(self.ptr, at, len); }
+          let at = unsafe { core::ptr::read_unaligned(events.as_ptr().add(cursor + 1) as *const u32) }.to_le() as usize;
+          let len = unsafe { core::ptr::read_unaligned(events.as_ptr().add(cursor + 5) as *const u32) }.to_le() as usize;
+          if active_callbacks & #active_const != 0 {
+            unsafe { #callback(parser_ptr, at, len); }
           }
           cursor += EVENT_RANGE_SIZE;
         }
@@ -65,34 +65,49 @@ pub fn generate_callbacks(callbacks: &[String]) -> TokenStream {
     #[cfg(target_family = "wasm")]
     impl Parser {
       #[inline]
-      fn invoke_callbacks(&mut self) {
+      fn invoke_callbacks(&mut self, event_len: usize) {
+        if self.callback_replay || event_len == 0 || event_len > EVENTS_BUFFER_SIZE {
+          return;
+        }
+
+        let events = unsafe { core::slice::from_raw_parts(self.events, event_len) }.to_vec();
+        let active_callbacks = self.active_callbacks;
+        let parser_ptr = self as *mut Parser as *mut c_void;
+        let replay_flag = core::ptr::addr_of_mut!(self.callback_replay);
+        unsafe { core::ptr::write_volatile(replay_flag, true); }
+
+        struct ReplayGuard(*mut bool);
+        impl Drop for ReplayGuard {
+          fn drop(&mut self) { unsafe { core::ptr::write_volatile(self.0, false); } }
+        }
+        let _guard = ReplayGuard(replay_flag);
         let mut cursor = 0usize;
 
-        loop {
-          let event_type = unsafe { *self.events.add(cursor) };
+        while cursor < events.len() {
+          let event_type = events[cursor];
 
           match event_type {
             EVENT_END => break,
             EVENT_ERROR => {
-              let at = unsafe { core::ptr::read_unaligned(self.events.add(cursor + 1) as *const u32) }.to_le() as usize;
-              if self.active_callbacks & CALLBACK_ACTIVE_ON_ERROR != 0 {
-                unsafe { on_error(self.ptr, at, 0); }
+              let at = unsafe { core::ptr::read_unaligned(events.as_ptr().add(cursor + 1) as *const u32) }.to_le() as usize;
+              if active_callbacks & CALLBACK_ACTIVE_ON_ERROR != 0 {
+                unsafe { on_error(parser_ptr, at, 0); }
               }
               cursor += EVENT_ERROR_SIZE;
             }
             EVENT_HEADERS => {
-              let at = unsafe { core::ptr::read_unaligned(self.events.add(cursor + 1) as *const u32) }.to_le() as usize;
-              let method_or_status = unsafe { core::ptr::read_unaligned(self.events.add(cursor + 5) as *const u16) }.to_le() as u32;
-              let should_keep_alive = unsafe { *self.events.add(cursor + 7) } != 0;
-              let should_upgrade = unsafe { *self.events.add(cursor + 8) } != 0;
-              let has_trailers = unsafe { *self.events.add(cursor + 9) } != 0;
-              let body_kind = unsafe { *self.events.add(cursor + 10) };
-              let content_length = unsafe { core::ptr::read_unaligned(self.events.add(cursor + 11) as *const u64) }.to_le() as f64;
+              let at = unsafe { core::ptr::read_unaligned(events.as_ptr().add(cursor + 1) as *const u32) }.to_le() as usize;
+              let method_or_status = unsafe { core::ptr::read_unaligned(events.as_ptr().add(cursor + 5) as *const u16) }.to_le() as u32;
+              let should_keep_alive = events[cursor + 7] != 0;
+              let should_upgrade = events[cursor + 8] != 0;
+              let has_trailers = events[cursor + 9] != 0;
+              let body_kind = events[cursor + 10];
+              let content_length = unsafe { core::ptr::read_unaligned(events.as_ptr().add(cursor + 11) as *const u64) }.to_le() as f64;
 
-              if self.active_callbacks & CALLBACK_ACTIVE_ON_HEADERS != 0 {
+              if active_callbacks & CALLBACK_ACTIVE_ON_HEADERS != 0 {
                 unsafe {
                   on_headers(
-                    self.ptr,
+                    parser_ptr,
                     at,
                     method_or_status,
                     should_keep_alive,
@@ -108,6 +123,10 @@ pub fn generate_callbacks(callbacks: &[String]) -> TokenStream {
             #(#replay_arms)*
             _ => break,
           }
+
+          if unsafe { core::ptr::read_volatile(core::ptr::addr_of!(self.destroy_pending)) } {
+            break;
+          }
         }
       }
     }

parser/src/lib.rs

@@ -70,6 +70,10 @@ pub struct Parser {
   pub unconsumed_len: usize,
   pub error_description_len: u8,
 
+  // Callback replay state. These fields are managed internally.
+  callback_replay: bool,
+  destroy_pending: bool,
+
   // Event buffer. Keep this at the end of the struct for external readers.
   pub events: *mut c_uchar,
 }
@@ -143,10 +147,35 @@ impl Parser {
       unconsumed: ptr::null(),
       unconsumed_len: 0,
       error_description_len: 0,
+      callback_replay: false,
+      destroy_pending: false,
       events: events_ptr,
     }
   }
 
+  #[inline(always)]
+  pub(crate) fn can_mutate(&self) -> bool { !self.callback_replay }
+
+  #[inline(always)]
+  pub(crate) unsafe fn can_mutate_raw(parser: *const Parser) -> bool {
+    !unsafe { core::ptr::read_volatile(core::ptr::addr_of!((*parser).callback_replay)) }
+  }
+
+  #[inline(always)]
+  pub(crate) unsafe fn request_destroy_raw(parser: *mut Parser) -> bool {
+    if unsafe { Self::can_mutate_raw(parser) } {
+      true
+    } else {
+      unsafe { core::ptr::write_volatile(core::ptr::addr_of_mut!((*parser).destroy_pending), true) };
+      false
+    }
+  }
+
+  #[inline(always)]
+  pub(crate) unsafe fn should_destroy_after_call_raw(parser: *const Parser) -> bool {
+    unsafe { Self::can_mutate_raw(parser) && core::ptr::read_volatile(core::ptr::addr_of!((*parser).destroy_pending)) }
+  }
+
   /// Resets a parser. The second parameters specifies if to also reset the
   /// parsed counter.
   ///
@@ -161,6 +190,10 @@ impl Parser {
   ///   * debug
   ///   * context
   pub fn reset(&mut self, keep_parsed: bool) {
+    if !self.can_mutate() {
+      return;
+    }
+
     self.state = STATE_START;
     self.paused = false;
 
@@ -199,6 +232,10 @@ impl Parser {
 
   /// Clears all values about the message in the parser.
   pub fn clear(&mut self) {
+    if !self.can_mutate() {
+      return;
+    }
+
     self.is_connect = false;
     self.method = 0;
     self.status = 0;
@@ -256,14 +293,26 @@ impl Parser {
   }
 
   /// Pauses the parser. It will have to be resumed via `resume`.
-  pub fn pause(&mut self) { self.paused = true; }
+  pub fn pause(&mut self) {
+    if self.can_mutate() {
+      self.paused = true;
+    }
+  }
 
   /// Resumes the parser.
-  pub fn resume(&mut self) { self.paused = false; }
+  pub fn resume(&mut self) {
+    if self.can_mutate() {
+      self.paused = false;
+    }
+  }
 
   /// Marks the parser as finished. Any new data received via `parse` will
   /// put the parser in the error state.
   pub fn finish(&mut self) {
+    if !self.can_mutate() {
+      return;
+    }
+
     match self.state {
       // If the parser is one of the initial states, simply jump to finish
       STATE_START | STATE_REQUEST_LINE | STATE_STATUS_LINE | STATE_FINISH => {
@@ -294,6 +343,10 @@ impl Parser {
   /// Marks parsing as failed and starts a standalone error event batch.
   #[inline(always)]
   pub fn fail(&mut self, code: u8, description: &str) {
+    if !self.can_mutate() {
+      return;
+    }
+
     self.set_error(code, description);
     let mut event_cursor = 0usize;
     if (self.active_events | self.active_callbacks) & EVENT_ACTIVE_ON_ERROR != 0 {