bundler: route every cross-thread event to the bundle thread through one post()

src/bundler/ParseTask.rs

@@ -41,11 +41,7 @@
 
 declare_scope!(ParseTask, hidden);
 
 #[allow(non_snake_case)]
-mod EventLoop {
-    pub(super) type Task = bun_event_loop::ConcurrentTask::ConcurrentTask;
-}
-
 // the per-file parse arena is held as `bump: &'static Bump` (the
 // worker arena is pinned for the entire bundle pass — see `run_with_source_code`),
 // so `bump.alloc_*` / `ArenaString::into_bump_str` already yield `&'static`
@@ -64,7 +60,6 @@
 // ───────────────────────────────────────────────────────────────────────────
 // ContentsOrFd
 // ───────────────────────────────────────────────────────────────────────────
-
 #[derive(bun_core::EnumTag)]
 #[enum_tag(existing = ContentsOrFdTag)]
 pub enum ContentsOrFd {
@@ -130,7 +125,8 @@
 
 /// The information returned to the Bundler thread when a parse finishes.
 pub(crate) struct Result {
-    pub(crate) task: EventLoop::Task,
+    /// Mini-loop queue node for `crate::post` (see `post::Event::NODE`).
+    pub(crate) task: bun_event_loop::AnyTaskWithExtraContext::AnyTaskWithExtraContext,
     pub(crate) ctx: bun_ptr::ParentRef<BundleV2<'static>, bun_ptr::Mut>,
     pub(crate) value: ResultValue,
     pub(crate) watcher_data: WatcherData,
@@ -204,7 +200,7 @@
     /// (LIFETIMES.tsv) into the arena-allocated bundle, set at `init` time and
     /// valid until `BundleV2::deinit`. Prefer this over open-coded
     /// `unsafe { &*task.ctx }`; sites that mutate the bundle (e.g.
-    /// `on_complete`) must continue to deref the raw `ctx` field directly.
+    /// `ParseComplete::run`) must continue to deref the raw `ctx` field directly.
     ///
     /// # Safety
     ///
@@ -225,7 +221,7 @@
         resolve_result: &_resolver::Result,
         source_index: Index,
         // Take `*mut` so the stored BACKREF retains
-        // write provenance for `on_complete` (a `&BundleV2` param would shrink
+        // write provenance for `ParseComplete::run` (a `&BundleV2` param would shrink
         // provenance to read-only, making the later `&mut *ctx` UB).
         ctx: *mut BundleV2<'_>,
     ) -> ParseTask {
@@ -2807,7 +2803,7 @@
 
         let result = Box::new(Result {
             ctx: this.ctx.expect("ParseTask.ctx unset"),
-            task: EventLoop::Task::default(),
+            task: Default::default(),
             value,
             // `ExternalFreeFunction`
             // doesn't derive `Copy`, so move it out (task is consumed here).
@@ -2825,55 +2821,9 @@
         // `ParseTask` is arena-owned (no Drop); `jsx` may hold owned slices from tsconfig.
         drop(core::mem::take(&mut this.jsx));
 
-        // `worker.ctx` is a `BackRef<BundleV2>` (safe `Deref`); the BACKREF deref
-        // of `linker.r#loop` is centralised in `LinkerContext::any_loop_mut`.
-        //
-        // The loop is effectively non-optional — `BundleV2::init`
-        // always sets `linker.r#loop` before scheduling any ParseTask. Running
-        // `on_complete` inline on the worker thread would violate
-        // `BundleV2::on_parse_task_complete`'s threading contract (it mutates the
-        // bundler graph, which is owned by the main/bundler thread).
-        match worker
-            .ctx
-            .linker
-            .any_loop_mut()
-            .expect("BundleV2.linker.loop must be set before scheduling ParseTask")
-        {
-            bun_event_loop::AnyEventLoop::Js { .. } => {
-                let ct =
-                    bun_event_loop::ConcurrentTask::ConcurrentTask::from_callback(result, |p| {
-                        // SAFETY: `p` is the `result` Box leaked above; ownership
-                        // transfers to `on_complete`, which deallocates it.
-                        unsafe { on_complete(p) };
-                        Ok(())
-                    });
-                let poster = worker
-                    .ctx
-                    .js_poster
-                    .as_ref()
-                    .expect("JS-owned bundle has a poster");
-                if let bun_event_loop::Posted::Refused(ct) = poster.post(ct) {
-                    // Owning JS VM torn down mid-bundle: free the hop and the result.
-                    // SAFETY: refused ⇒ we own the task box and the leaked result.
-                    unsafe {
-                        bun_event_loop::ConcurrentTask::ConcurrentTask::release_refused(ct);
-                        drop(bun_core::heap::take(result));
-                    }
-                }
-            }
-            bun_event_loop::AnyEventLoop::Mini(mini) => {
-                // SAFETY: `result` is a valid heap pointer with `task` at the given offset;
-                // ownership transfers to the mini event loop which frees it after `on_complete_mini`.
-                unsafe {
-                    mini.enqueue_task_concurrent_with_extra_ctx::<Result, BundleV2<'static>>(
-                        result,
-                        on_complete_mini,
-                        offset_of!(Result, task),
-                    );
-                }
-            }
-        }
-        // Runs at function exit, i.e. after enqueue.
+        // SAFETY: `result` is a fresh heap allocation that `ParseComplete::run`
+        // (or `refused`) frees; `result.ctx` is the bundle this task belongs to.
+        unsafe { crate::post::post::<ParseComplete>(result) };
         worker.unget();
     }
 
@@ -2887,46 +2837,42 @@
         }
     }
 
-    fn on_complete_mini(result: *mut Result, ctx: *mut BundleV2<'static>) {
-        // SAFETY: callback contract — `result` was heap-allocated above; `ctx` is
-        // the BACKREF stashed in `result.ctx`.
-        BundleV2::on_parse_task_complete(unsafe { &mut *result }, unsafe { &mut *ctx });
-        // SAFETY: `result` is uniquely owned (callback contract).
-        drop_result_owned_fields(unsafe { &mut *result });
-        // `drop(heap::take(result))` would run full Drop glue:
-        // `on_parse_task_complete` SWAPS `result.value.Success.source` with the
-        // graph's placeholder and moves `result.ast` out, so post-swap
-        // `result.value` holds the *placeholder* `Source` whose
-        // `contents: Cow::Borrowed` may alias plugin-/loader-provided bytes the
-        // graph's swapped-in Source still references (asan use-after-poison at
-        // process_files_to_copy:4241 in bundler_loader/_plugin tests). So:
-        // dealloc the box without running Drop.
-        // SAFETY: `result` came from `bun_core::heap::into_raw(Box<Result>)`
-        // above; uniquely owned. Dealloc with the same layout, no field Drop.
-        unsafe { std::alloc::dealloc(result.cast::<u8>(), std::alloc::Layout::new::<Result>()) };
-    }
+    /// A worker finished parsing one file. `Result` is heap-allocated by the
+    /// worker and freed here.
+    pub(crate) struct ParseComplete;
 
-    /// # Safety
-    /// `result` must be a live, uniquely-owned heap allocation produced by
-    /// `bun_core::heap::into_raw(Box<Result>)` in `run_from_thread_pool_impl`
-    /// (or `ServerComponentParseTask`'s equivalent). Ownership transfers to
-    /// this fn, which deallocates `result` before returning. Must run on the
-    /// main/bundler thread (it dereferences `result.ctx` mutably).
-    pub(crate) unsafe fn on_complete(result: *mut Result) {
-        // SAFETY: result allocated via heap::alloc above; uniquely owned here.
-        let r = unsafe { &mut *result };
-        let ctx = r.ctx;
-        // SAFETY: `ctx` is a ParentRef<BundleV2> stored with write provenance
-        // (`from_raw_mut` in `ParseTask::init`); the BundleV2 outlives the bundle
-        // pass and no other `&mut BundleV2` is live on this (main) thread when the
-        // event-loop callback fires. `r` and `*ctx` are disjoint allocations.
-        BundleV2::on_parse_task_complete(r, unsafe { ctx.assume_mut() });
-        drop_result_owned_fields(r);
-        // See `on_complete_mini` for why this is `dealloc`, not `drop(take(_))`.
-        // SAFETY: `result` came from `bun_core::heap::into_raw(Box<Result>)`
-        // above; uniquely owned. Dealloc with the same layout, no field Drop.
-        unsafe { std::alloc::dealloc(result.cast::<u8>(), std::alloc::Layout::new::<Result>()) };
+    impl crate::post::Event for ParseComplete {
+        type Item = Result;
+        const NODE: usize = offset_of!(Result, task);
+
+        fn bundle(result: *mut Result) -> *mut BundleV2<'static> {
+            // SAFETY: `result` is live (`post`'s contract).
+            unsafe { (*result).ctx }.as_mut_ptr()
+        }
+
+        unsafe fn run(result: *mut Result, bv2: &mut BundleV2<'static>) {
+            // SAFETY: `post`'s contract — uniquely owned heap allocation, disjoint from `*bv2`.
+            let r = unsafe { &mut *result };
+            BundleV2::on_parse_task_complete(r, bv2);
+            drop_result_owned_fields(r);
+            // Not `drop(heap::take(result))`: `on_parse_task_complete` SWAPS
+            // `result.value.Success.source` with the graph's placeholder and moves
+            // `result.ast` out, so post-swap `result.value` holds the *placeholder*
+            // `Source` whose `contents: Cow::Borrowed` may alias plugin-/loader-provided
+            // bytes the graph's swapped-in Source still references (asan
+            // use-after-poison in bundler_loader/_plugin tests). Dealloc without Drop.
+            // SAFETY: `result` came from `heap::into_raw(Box<Result>)`; uniquely owned.
+            unsafe {
+                std::alloc::dealloc(result.cast::<u8>(), std::alloc::Layout::new::<Result>())
+            };
+        }
+
+        unsafe fn refused(result: *mut Result) {
+            // Nothing was swapped out of it, so full Drop is right here.
+            // SAFETY: `result` came from `heap::into_raw(Box<Result>)`; uniquely owned.
+            drop(unsafe { bun_core::heap::take(result) });
+        }
     }
 } // end mod parse_worker
 
-pub(crate) use parse_worker::on_complete;
+pub(crate) use parse_worker::ParseComplete;

src/bundler/ServerComponentParseTask.rs

@@ -2,7 +2,6 @@
 //! running through the js_parser. It emits a ParseTask.Result and joins
 //! with the same logic that it runs though.
 
-use core::mem::offset_of;
 use std::fmt::Write as _;
 
 use bun_alloc::{AllocError as OOM, Arena}; // bumpalo::Bump re-export
@@ -22,7 +21,7 @@
 use crate::bundle_v2::BundleV2;
 use crate::cache::ExternalFreeFunction;
 use crate::options::{Loader, Target};
-use crate::parse_task::{self, ResultValue, Success, WatcherData, on_complete};
+use crate::parse_task::{self, ParseComplete, ResultValue, Success, WatcherData};
 
 pub(crate) struct ServerComponentParseTask {
     pub task: ThreadPoolTask,
@@ -107,59 +106,15 @@
     //
     // The loop is effectively non-optional — `BundleV2::init`
     // always sets `linker.r#loop` before scheduling any ServerComponentParseTask.
     // Running `on_complete` inline on the worker thread would violate
     // `BundleV2::on_parse_task_complete`'s threading contract (it mutates the
     // bundler graph, which is owned by the main/bundler thread).
-    match worker
-        .ctx
-        .linker
-        .any_loop_mut()
-        .expect("BundleV2.linker.loop must be set before scheduling ServerComponentParseTask")
-    {
-        bun_event_loop::AnyEventLoop::Js { .. } => {
-            let ct = bun_event_loop::ConcurrentTask::ConcurrentTask::from_callback(result, |p| {
-                // SAFETY: `p` is the `result` Box leaked above; ownership
-                // transfers to `on_complete`, which deallocates it.
-                unsafe { on_complete(p) };
-                Ok(())
-            });
-            let poster = worker
-                .ctx
-                .js_poster
-                .as_ref()
-                .expect("JS-owned bundle has a poster");
-            if let bun_event_loop::Posted::Refused(ct) = poster.post(ct) {
-                // Owning JS VM torn down mid-bundle: free the hop and the result.
-                // SAFETY: refused ⇒ we own the task box and the leaked result.
-                unsafe {
-                    bun_event_loop::ConcurrentTask::ConcurrentTask::release_refused(ct);
-                    drop(bun_core::heap::take(result));
-                }
-            }
-        }
-        bun_event_loop::AnyEventLoop::Mini(mini) => {
-            // SAFETY: `result` is a freshly Box-leaked `parse_task::Result` (above) and
-            // `offset_of!(parse_task::Result, task)` is the intrusive task field within it.
-            unsafe {
-                mini.enqueue_task_concurrent_with_extra_ctx::<parse_task::Result, BundleV2<'static>>(
-                    result,
-                    on_complete_mini,
-                    offset_of!(parse_task::Result, task),
-                );
-            }
-        }
-    }
-    // Runs at function exit, i.e. after enqueue.
+    // SAFETY: `result` is a fresh heap allocation that `ParseComplete::run`
+    // (or `refused`) frees; `result.ctx` is the bundle this task belongs to.
+    unsafe { crate::post::post::<ParseComplete>(result) };
     worker.unget();
 }
 
-fn on_complete_mini(result: *mut parse_task::Result, _ctx: *mut BundleV2<'static>) {
-    // `on_complete` already recovers `ctx` from `result.ctx`.
-    // SAFETY: callback contract — `result` is the uniquely-owned Box leaked in
-    // `run_from_thread_pool`; ownership transfers to `on_complete`.
-    unsafe { on_complete(result) };
-}
-
 fn task_callback(
     task: &mut ServerComponentParseTask,
     log: &mut Log,