install: wait for the isolated installer's pool tasks before it is freed

src/install/isolated_install.rs

@@ -37,6 +37,7 @@ use bun_paths::path_options::AssumeOk as _;
 use bun_paths::{self as paths, AutoAbsPath as AbsPath, AutoRelPath, PathBuffer};
 use bun_semver as semver;
 use bun_sys::{self as sys, Fd};
+use bun_threading::WaitGroup;
 use bun_wyhash::{Wyhash, Wyhash11};
 
 use crate::analytics;
@@ -2008,6 +2009,8 @@ pub(crate) fn install_isolated_packages(
         // TODO: delete
         let mut seen_workspace_ids: HashMap<PackageID, ()> = HashMap::default();
 
+        let tasks_in_flight = WaitGroup::init();
+
         // `installer::Task` carries `result: Result` (Drop via `TaskError`
         // payloads) and a non-nullable fn-ptr in `thread_pool::Task`, so
         // `assume_init()` on uninit memory is instant UB and a subsequent
@@ -2026,10 +2029,10 @@ pub(crate) fn install_isolated_packages(
                     installer: bun_ptr::BackRef::from(core::ptr::NonNull::dangling()),
                     result: installer::Result::None,
                     relink: installer::Relink::Off,
-                    task: bun_threading::thread_pool::Task {
-                        callback: installer::Task::callback,
-                        node: Default::default(),
-                    },
+                    task: bun_threading::thread_pool::CountedTask::new(
+                        installer::Task::callback,
+                        &tasks_in_flight,
+                    ),
                     next: bun_threading::Link::new(),
                 });
             }
@@ -2039,6 +2042,7 @@ pub(crate) fn install_isolated_packages(
 
         let show_progress = manager.options.log_level.show_progress();
         let installed = DynamicBitSet::init_empty(lockfile.packages.len())?;
+        let released = DynamicBitSet::init_empty(store.entries.len())?;
         let trusted_dependencies_from_update_requests =
             manager.find_trusted_dependencies_from_update_requests();
         // `Installer.manager` is a BACKREF raw pointer; copying `manager_ptr`
@@ -2057,6 +2061,8 @@ pub(crate) fn install_isolated_packages(
             },
             store: &store,
             tasks,
+            tasks_in_flight: &tasks_in_flight,
+            released,
             waiters_head: vec![store::entry::Id::INVALID; store.entries.len()].into_boxed_slice(),
             next_waiter: vec![store::entry::Id::INVALID; store.entries.len()].into_boxed_slice(),
             trusted_dependencies_mutex: Default::default(),

src/install/isolated_install/Installer.rs

@@ -7,7 +7,7 @@ use bun_core::{Environment, Global, Output};
 use bun_core::{ZStr, strings};
 use bun_paths::{self as paths, AbsPath, AutoAbsPath, AutoRelPath};
 use bun_sys::{self as sys, Fd};
-use bun_threading::{Mutex, UnboundedQueue, thread_pool};
+use bun_threading::{Mutex, UnboundedQueue, WaitGroup, thread_pool};
 
 use bun_semver::String as SemverString;
 use bun_sys::{FdDirExt as _, FdExt as _};
@@ -91,6 +91,13 @@ pub struct Installer<'a> {
     pub(crate) task_queue: UnboundedQueue<Task>, // intrusive via .next
     pub(crate) tasks: Box<[Task]>,
 
+    /// Tasks currently on the thread pool: `start_task` adds one, the
+    /// `CountedTask` in `Task::task` finishes it, and `Drop` waits for it.
+    pub(crate) tasks_in_flight: &'a WaitGroup,
+
+    /// Main thread only: entries whose pending-task slot has been released.
+    pub(crate) released: Bitset,
+
     /// Stable Rust has no
     /// generic atomic-enum, so store the `#[repr(u8)]` discriminant and
     /// round-trip via `Method::from_u8` at the load sites below.
@@ -119,6 +126,14 @@ pub struct Installer<'a> {
     pub(crate) next_waiter: Box<[StoreEntryId]>,
 }
 
+impl Drop for Installer<'_> {
+    /// Tasks on the pool read `self` and the store through `Task::installer`, so
+    /// neither may be freed until they have returned.
+    fn drop(&mut self) {
+        self.tasks_in_flight.wait();
+    }
+}
+
 impl<'a> Installer<'a> {
     // BACKREF accessors — `manager` points outside `Self`; see field doc.
     #[inline]
@@ -166,10 +181,36 @@ impl<'a> Installer<'a> {
             | Result::RunScripts(_)
         ));
 
+        if Environment::CI_ASSERT {
+            assert!(
+                !self.released.is_set(entry_id.get() as usize),
+                "isolated install: started entry {} after its pending task was released",
+                entry_id.get()
+            );
+        }
+
         task.result = Result::None;
+        self.tasks_in_flight.add_one();
+        let task: *mut Task = task;
+        // SAFETY: raw projection through the whole `tasks[entry_id]` slot, so the
+        // pointer `Task::callback` gets back keeps provenance over the whole `Task`.
+        let pool_task = unsafe { &raw mut (*task).task.task };
         manager
             .thread_pool
-            .schedule(thread_pool::Batch::from(&raw mut task.task));
+            .schedule(thread_pool::Batch::from(pool_task));
+    }
+
+    /// Main thread only. Every entry releases its pending-task slot exactly once.
+    fn release_pending_task(&mut self, entry_id: StoreEntryId) {
+        if Environment::CI_ASSERT {
+            assert!(
+                !self.released.is_set(entry_id.get() as usize),
+                "isolated install: released the pending task of entry {} twice",
+                entry_id.get()
+            );
+        }
+        self.released.set(entry_id.get() as usize);
+        self.manager_mut().decrement_pending_tasks();
     }
 
     /// Called from main thread, for an existing project-local store entry.
@@ -434,14 +475,10 @@ impl<'a> Installer<'a> {
 
         self.summary.fail += 1;
 
-        self.decrement_pending_tasks();
+        self.release_pending_task(entry_id);
         self.resume_unblocked_tasks(entry_id);
     }
 
-    pub(crate) fn decrement_pending_tasks(&mut self) {
-        self.manager_mut().decrement_pending_tasks();
-    }
-
     /// Called from main thread
     pub(crate) fn on_task_blocked(&mut self, entry_id: StoreEntryId) {
         // race condition (fixed now): task decides it is blocked because one of its dependencies
@@ -516,7 +553,7 @@ impl<'a> Installer<'a> {
             );
         }
 
-        self.decrement_pending_tasks();
+        self.release_pending_task(entry_id);
         self.resume_unblocked_tasks(entry_id);
 
         if let Some(node) = self.install_node.as_mut() {
@@ -651,7 +688,8 @@ pub struct Task {
     /// any `start_task` call — see `isolated_install.rs`.
     pub(crate) installer: bun_ptr::BackRef<Installer<'static>>,
 
-    pub(crate) task: thread_pool::Task,
+    /// Counted on `Installer::tasks_in_flight`; runs `Task::callback`.
+    pub(crate) task: thread_pool::CountedTask,
     pub(crate) next: bun_threading::Link<Task>, // INTRUSIVE: bun.UnboundedQueue(Task, .next) link
 
     pub(crate) result: Result,
@@ -1947,15 +1985,17 @@ impl Task {
             Err(_oom) => bun_core::out_of_memory(),
         };
 
-        // SAFETY: installer outlives all tasks (BACKREF). `callback` runs on the
-        // thread pool concurrently across many `Task`s sharing the same
-        // `*mut Installer` — never materialize `&mut Installer`. `task_queue.push`
-        // takes `&self` (lock-free); `store.entries` columns are atomic / per-entry;
-        // both are reached through a shared `&Installer`. `manager.wake()` is the
-        // cross-thread wakeup: route through `PackageManager::wake_raw` which never
-        // forms `&mut PackageManager`, so two threads finishing simultaneously do
-        // not hold aliased exclusive borrows ("deref it fresh per call" alone
-        // would not prevent the `&mut` lifetimes from overlapping).
+        // SAFETY: the installer is alive until this fn returns: `Task::task` counts
+        // on `Installer::tasks_in_flight` until then, and `Installer::drop` waits
+        // for that count. `callback` runs on the thread pool concurrently across
+        // many `Task`s sharing the same `*mut Installer` — never materialize
+        // `&mut Installer`. `task_queue.push` takes `&self` (lock-free);
+        // `store.entries` columns are atomic / per-entry; both are reached through
+        // a shared `&Installer`. `manager.wake()` is the cross-thread wakeup: route
+        // through `PackageManager::wake_raw` which never forms
+        // `&mut PackageManager`, so two threads finishing simultaneously do not
+        // hold aliased exclusive borrows ("deref it fresh per call" alone would
+        // not prevent the `&mut` lifetimes from overlapping).
         let installer_ptr = this.installer;
         let installer = installer_ptr.get();
         let manager_ptr: *mut PackageManager = installer.manager;

src/install/isolated_install/Store.rs

@@ -99,7 +99,7 @@ impl Drop for Store {
         use entry::EntryColumns as _;
         for slot in self.entries.items_scripts() {
             if let Some(list) = slot.take() {
-                // SAFETY: the installer returned only after every task finished, so nothing else references the list boxed in Installer's RunPreinstall step.
+                // SAFETY: `Installer::drop` waited for every task on the pool, so nothing else references the list boxed in Installer's RunPreinstall step.
                 unsafe { bun_core::heap::destroy(list) };
             }
         }