diff --git a/src/install/PackageManager.rs b/src/install/PackageManager.rs index e2c97f58844d..0bba0746783a 100644 --- a/src/install/PackageManager.rs +++ b/src/install/PackageManager.rs @@ -361,6 +361,10 @@ pub struct PackageManager { pub(crate) appended_task_packages: AppendedTaskPackageMap, pub(crate) network_dedupe_map: crate::network_task::DedupeMap, + /// Manifest tasks that failed, until `forget_failed_manifest_tasks`. + pub(crate) failed_manifest_tasks: Vec, + /// Runtime lookups inside `sleep_until`, where JS runs and can resolve again. + pub(crate) root_lookups_waiting: u32, pub(crate) async_network_task_queue: AsyncNetworkTaskQueue, pub(crate) network_tarball_batch: thread_pool::Batch, pub(crate) network_resolve_batch: thread_pool::Batch, @@ -2108,6 +2112,8 @@ pub fn init( wr!(running_git_tasks, AtomicU32::new(0)); wr!(appended_task_packages, AppendedTaskPackageMap::default()); wr!(network_dedupe_map, Default::default()); + wr!(failed_manifest_tasks, Vec::new()); + wr!(root_lookups_waiting, 0); wr!(async_network_task_queue, AsyncNetworkTaskQueue::default()); wr!(network_tarball_batch, thread_pool::Batch::default()); wr!(network_resolve_batch, thread_pool::Batch::default()); @@ -2569,6 +2575,8 @@ fn init_with_runtime_once( wr!(running_git_tasks, AtomicU32::new(0)); wr!(appended_task_packages, AppendedTaskPackageMap::default()); wr!(network_dedupe_map, Default::default()); + wr!(failed_manifest_tasks, Vec::new()); + wr!(root_lookups_waiting, 0); wr!(async_network_task_queue, AsyncNetworkTaskQueue::default()); wr!(network_tarball_batch, thread_pool::Batch::default()); wr!(network_resolve_batch, thread_pool::Batch::default()); diff --git a/src/install/PackageManager/PackageManagerEnqueue.rs b/src/install/PackageManager/PackageManagerEnqueue.rs index 990321e3d219..01bc120afa21 100644 --- a/src/install/PackageManager/PackageManagerEnqueue.rs +++ b/src/install/PackageManager/PackageManagerEnqueue.rs @@ -640,6 +640,7 @@ pub fn enqueue_dependency_to_root( this.start_progress_bar_if_none(); } + this.root_lookups_waiting += 1; let mgr: *mut PackageManager = this; let mut closure = Closure { err: None, @@ -651,6 +652,7 @@ pub fn enqueue_dependency_to_root( // `Closure::is_done`, so the callback's `&mut *closure.manager` // is the unique live borrow. unsafe { PackageManager::sleep_until(mgr, &mut closure, Closure::is_done) }; + this.root_lookups_waiting -= 1; if this.options.log_level.show_progress() { this.end_progress_bar(); diff --git a/src/install/PackageManager/runTasks.rs b/src/install/PackageManager/runTasks.rs index 318e0b6ef2e4..7939cdebddd4 100644 --- a/src/install/PackageManager/runTasks.rs +++ b/src/install/PackageManager/runTasks.rs @@ -453,9 +453,9 @@ fn run_tasks_erased( // pointer (`StringOrTinyString` is self-referential and not // `Clone`) so the loop body can read `name` after the // `&mut task.callback` borrow ends. - // SAFETY: `name` lives in `task.callback` which outlives this - // match arm (the task is only `put` back to the pool by a later - // resolve-task pass, never inside this loop iteration). + // SAFETY: `name` lives in `task.callback`. The task goes back to + // the pool in a later resolve-task pass, or at a failure exit of + // this arm, which does so after its last use of `name`. let name = unsafe { bun_ptr::detach_lifetime(name.slice()) }; let is_extended_manifest = *is_extended_manifest; if log_level.show_progress() { @@ -554,6 +554,10 @@ fn run_tasks_erased( } } + manifest_task_failed(manager, task.task_id); + // SAFETY: the request ran and will not be retried; nothing uses the + // task, or `name`, after this. + unsafe { release_network_task(&manager.preallocated_network_tasks, task_ptr) }; continue; }; let response = &metadata.response; @@ -577,6 +581,12 @@ fn run_tasks_erased( &task.url_buf, ); + manifest_task_failed(manager, task.task_id); + // SAFETY: the request ran and will not be retried; nothing uses the + // task, or `name`, after this. + unsafe { + release_network_task(&manager.preallocated_network_tasks, task_ptr) + }; continue; } @@ -610,6 +620,10 @@ fn run_tasks_erased( } } + manifest_task_failed(manager, task.task_id); + // SAFETY: the request ran and will not be retried; nothing uses the + // task, or `name`, after this. + unsafe { release_network_task(&manager.preallocated_network_tasks, task_ptr) }; continue; } @@ -1042,15 +1056,7 @@ fn run_tasks_erased( // SAFETY: see the put-task `defer!` above — `manager_ptr` is the // function-scope provenance root; `net_ptr` is the network task // owned by this resolve task and is returned to the pool here. - // `unsafe_http_client` is `MaybeUninit` so `put()`'s - // `drop_in_place` skips it — drop manually - // (HTTP completed, so it IS init) so the inner - // `AsyncHTTP.{request,response}_headers: EntryList` don't - // leak per put/get cycle. - unsafe { - (*net_ptr).unsafe_http_client.assume_init_drop(); - (*manager_ptr).preallocated_network_tasks.put(net_ptr); - } + unsafe { release_network_task(&(*manager_ptr).preallocated_network_tasks, net_ptr) }; }; if task.status == Task::Status::Fail { let req = task.request_package_manifest(); @@ -1075,6 +1081,7 @@ fn run_tasks_erased( ); } + manifest_task_failed(manager, task.id); continue; } debug_assert!(task.tag == Task::Tag::PackageManifest); @@ -1135,12 +1142,7 @@ fn run_tasks_erased( // SAFETY: see the put-task `defer!` above — `manager_ptr` is the // function-scope provenance root; `net_ptr` (checked non-null) is // the network task owned by this resolve task. - // `unsafe_http_client` is `MaybeUninit` so `put()`'s drop - // skips it — drop manually so headers don't leak. - unsafe { - (*net_ptr).unsafe_http_client.assume_init_drop(); - (*manager_ptr).preallocated_network_tasks.put(net_ptr); - } + unsafe { release_network_task(&(*manager_ptr).preallocated_network_tasks, net_ptr) }; } }; @@ -1946,6 +1948,35 @@ pub(crate) fn network_task_has_failed(this: &PackageManager, task_id: Task::Id) .is_some_and(|e| e.failed) } +/// Its waiters never run. `bun install` keeps the `network_dedupe_map` entry to ask once per run. +fn manifest_task_failed(this: &mut PackageManager, task_id: Task::Id) { + let _ = this.task_queue.remove(&task_id); + this.failed_manifest_tasks.push(task_id); +} + +/// `put()` skips the `MaybeUninit` HTTP client, so drop it first or its header lists leak. +/// SAFETY: the request ran, which initialized the client, and nothing uses `task` after this. +unsafe fn release_network_task(pool: &super::PreallocatedNetworkTasks, task: *mut NetworkTask) { + // SAFETY: fn contract. + unsafe { + (*task).unsafe_http_client.assume_init_drop(); + pool.put(task); + } +} + +/// For the runtime, whose manager outlives a resolve: the next resolve asks the registry again. +pub(crate) fn forget_failed_manifest_tasks(this: &mut PackageManager) { + // A resolve nested in a lookup's wait ends first; the waiting lookup's second pass needs its record. + if this.root_lookups_waiting > 0 { + return; + } + for task_id in this.failed_manifest_tasks.drain(..) { + let _ = this.network_dedupe_map.remove(&task_id); + // A resolver pass that ran after the failure queued its waiter again. + let _ = this.task_queue.remove(&task_id); + } +} + /// The first failed download in a `run_tasks` pass halves the number of /// concurrent requests (down to the configured minimum). fn throttle_after_network_error(manager: &PackageManager, has_network_error: &mut bool) { @@ -2173,6 +2204,10 @@ impl PackageManager { network_task_has_failed(self, task_id) } #[inline] + pub fn forget_failed_manifest_tasks(&mut self) { + forget_failed_manifest_tasks(self) + } + #[inline] pub(crate) fn get_network_task(&mut self) -> *mut NetworkTask { get_network_task(self) } diff --git a/src/jsc/VirtualMachine.rs b/src/jsc/VirtualMachine.rs index 053879072380..13250530e1dd 100644 --- a/src/jsc/VirtualMachine.rs +++ b/src/jsc/VirtualMachine.rs @@ -4868,6 +4868,13 @@ impl VirtualMachine { mode.is_esm(), IS_A_FILE_PATH, ); + // Both passes of `_resolve` share one failed registry lookup; the next resolve asks again. + if let Some(pm) = jsc_vm.transpiler.resolver.package_manager { + // SAFETY: sole `dyn AutoInstaller` impl is `PackageManager`. + unsafe { + (*pm.cast::().as_ptr()).forget_failed_manifest_tasks(); + } + } if let Err(err_) = resolve_result { let err = err_; let import_kind = mode.import_kind(); diff --git a/test/cli/run/run-autoinstall.test.ts b/test/cli/run/run-autoinstall.test.ts index bdf63a8690f8..0d7e000b1a54 100644 --- a/test/cli/run/run-autoinstall.test.ts +++ b/test/cli/run/run-autoinstall.test.ts @@ -1,6 +1,6 @@ import { describe, expect, test } from "bun:test"; import { mkdirSync } from "fs"; -import { bunEnv, bunExe, tempDir, tmpdirSync } from "harness"; +import { bunEnv, bunExe, isASAN, tempDir, tmpdirSync } from "harness"; import { join } from "path"; // --install= Configure auto-install behavior. One of "auto" (default, auto-installs when no node_modules), "fallback" (missing packages only), "force" (always). @@ -121,3 +121,244 @@ test("--install=fallback to install missing packages", async () => { expect(stderr?.toString("utf8")).not.toContain("error: Cannot find package 'is-odd'"); expect(stdout?.toString("utf8")).toBe("true false\n"); }); + +// Without node_modules the runtime keeps one package manager for the life of +// the process. A manifest request that failed stayed recorded in it as "already +// requested", so every later resolve of that name failed without a request, +// also after the registry was healthy again. +describe.concurrent("a failed registry lookup is asked again by the next resolve", () => { + // Fails every manifest request with `respond` until the script asks for /__heal. + // The script asks for /__next between its resolves: `lookups` is one list of requests per resolve. + function flakyRegistry(respond: (deadPort: number) => Response) { + const dead = Bun.listen({ + hostname: "127.0.0.1", + port: 0, + socket: { + open(socket) { + socket.end(); + }, + data() {}, + }, + }); + const lookups: string[][] = [[]]; + const tarballs = new Map(); + let healthy = false; + const server = Bun.serve({ + port: 0, + hostname: "127.0.0.1", + async fetch(req) { + const { pathname, origin } = new URL(req.url); + if (pathname === "/__next") return (lookups.push([]), new Response("ok")); + if (pathname === "/__heal") return ((healthy = true), new Response("ok")); + if (pathname === "/__break") return ((healthy = false), new Response("ok")); + lookups.at(-1)!.push(pathname); + if (pathname.endsWith(".tgz")) return new Response(tarballs.get(pathname)); + if (!healthy) return respond(dead.port); + const name = pathname.slice(1); + const tarball = await new Bun.Archive( + { + "package/package.json": JSON.stringify({ name, version: "1.0.0", main: "index.js" }), + "package/index.js": "module.exports = { version: '1.0.0' };", + }, + { compress: "gzip" }, + ).bytes(); + tarballs.set(`/${name}/-/${name}-1.0.0.tgz`, tarball); + return Response.json({ + name, + "dist-tags": { latest: "1.0.0" }, + versions: { + "1.0.0": { + name, + version: "1.0.0", + main: "index.js", + dist: { + tarball: `${origin}/${name}/-/${name}-1.0.0.tgz`, + integrity: "sha512-" + new Bun.CryptoHasher("sha512").update(tarball).digest("base64"), + }, + }, + }, + }); + }, + }); + return { + url: server.url.href, + lookups, + [Symbol.dispose]() { + server.stop(true); + dead.stop(true); + }, + }; + } + + async function run(registry: { url: string }, script: string, env: Record = {}) { + using dir = tempDir("autoinstall-failed-lookup", { + "bunfig.toml": `[install]\nregistry = "${registry.url}"\n`, + "index.js": `const registry = ${JSON.stringify(registry.url)};\n${script}`, + }); + await using proc = Bun.spawn({ + cmd: [bunExe(), "index.js"], + cwd: String(dir), + env: { + ...bunEnv, + BUN_INSTALL_CACHE_DIR: join(String(dir), ".bun-cache"), + // A 5xx or a dropped connection is retried: one lookup is then two requests. + BUN_CONFIG_HTTP_RETRY_COUNT: "1", + ...env, + }, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + return { stdout: stdout.trim(), stderr, exitCode }; + } + + const failures: Record Response; requestsPerLookup: number; busy?: true }> = + { + "404": { respond: () => new Response("{}", { status: 404 }), requestsPerLookup: 1 }, + "500": { respond: () => new Response("{}", { status: 500 }), requestsPerLookup: 2 }, + "a manifest that does not parse": { respond: () => new Response("not json"), requestsPerLookup: 1 }, + // the connection closes before a response arrives + "a dropped connection": { + respond: port => Response.redirect(`http://127.0.0.1:${port}/flaky-pkg`, 302), + requestsPerLookup: 2, + }, + // The wait for the registry runs the event loop. While a microtask keeps that turn busy, the + // response lands, and the module queue's poll handles the failure in place of the waiter. + // The spin only steers which of the two handles it: every assertion holds for both. + "404 while the event loop is busy": { + respond: () => new Response("{}", { status: 404 }), + requestsPerLookup: 1, + busy: true, + }, + }; + + test.each(Object.entries(failures))("%s", async (_, { respond, requestsPerLookup, busy }) => { + using registry = flakyRegistry(respond); + const { stdout, exitCode } = await run( + registry, + ` + const attempt = () => { + ${busy ? "queueMicrotask(() => { const end = performance.now() + 100; while (performance.now() < end); });" : ""} + try { return require("flaky-pkg").version; } catch (e) { return e.code; } + }; + const results = [attempt()]; + await fetch(registry + "__next"); + results.push(attempt()); + await fetch(registry + "__heal"); + await fetch(registry + "__next"); + results.push(attempt()); + results.push(await import("flaky-pkg").then(m => m.default.version, e => e.code)); + console.log(JSON.stringify(results)); + `, + ); + + expect(stdout).toBe(JSON.stringify(["MODULE_NOT_FOUND", "MODULE_NOT_FOUND", "1.0.0", "1.0.0"])); + // Each failing resolve is one lookup, and so is the resolve after the registry recovers. + const lookup = Array(requestsPerLookup).fill("/flaky-pkg"); + expect(registry.lookups).toEqual([lookup, lookup, ["/flaky-pkg", "/flaky-pkg/-/flaky-pkg-1.0.0.tgz"]]); + expect(exitCode).toBe(0); + }); + + // The wait for the registry runs the event loop, so a microtask can resolve inside it. That + // nested resolve ends first. It must not drop the record the outer resolve's second pass needs. + test("a resolve nested in the wait does not make the outer resolve ask twice", async () => { + using registry = flakyRegistry(() => new Response("{}", { status: 404 })); + const { stdout, exitCode } = await run( + registry, + ` + const attempt = name => { try { return require(name).version; } catch (e) { return e.code; } }; + const inner = new Promise(resolve => queueMicrotask(() => resolve(attempt("flaky-inner")))); + const first = [attempt("flaky-outer"), await inner]; + await fetch(registry + "__next"); + console.log(JSON.stringify([first, [attempt("flaky-outer"), attempt("flaky-inner")]])); + `, + ); + + const notFound = ["MODULE_NOT_FOUND", "MODULE_NOT_FOUND"]; + expect(JSON.parse(stdout)).toEqual([notFound, notFound]); + expect(registry.lookups).toEqual([ + ["/flaky-outer", "/flaky-inner"], + ["/flaky-outer", "/flaky-inner"], + ]); + expect(exitCode).toBe(0); + }); + + test("through every way to resolve", async () => { + using registry = flakyRegistry(() => new Response("{}", { status: 404 })); + const { stdout, exitCode } = await run( + registry, + ` + const ways = { + "require": name => require(name).version, + "require.resolve": name => require.resolve(name).includes(name), + "Bun.resolveSync": name => Bun.resolveSync(name, import.meta.dir).includes(name), + "import.meta.resolve": name => import.meta.resolve(name).includes(name), + "import()": async name => (await import(name)).default.version, + "Bun.resolve": async name => (await Bun.resolve(name, import.meta.dir)).includes(name), + }; + const results = {}; + let i = 0; + for (const [way, resolve] of Object.entries(ways)) { + const name = "flaky-pkg-" + i++; + const attempt = () => Promise.resolve(name).then(resolve).catch(e => e.code); + await fetch(registry + "__break"); + const broken = await attempt(); + await fetch(registry + "__next"); + await fetch(registry + "__heal"); + results[way] = [broken, await attempt()]; + await fetch(registry + "__next"); + } + console.log(JSON.stringify(results)); + `, + ); + + expect(JSON.parse(stdout)).toEqual({ + "require": ["MODULE_NOT_FOUND", "1.0.0"], + "require.resolve": ["MODULE_NOT_FOUND", true], + "Bun.resolveSync": ["ERR_MODULE_NOT_FOUND", true], + "import.meta.resolve": ["ERR_MODULE_NOT_FOUND", true], + "import()": ["ERR_MODULE_NOT_FOUND", "1.0.0"], + "Bun.resolve": ["ERR_MODULE_NOT_FOUND", true], + }); + expect(registry.lookups).toEqual([ + ...Array.from({ length: 6 }, (_, i) => [ + [`/flaky-pkg-${i}`], + [`/flaky-pkg-${i}`, `/flaky-pkg-${i}/-/flaky-pkg-${i}-1.0.0.tgz`], + ]).flat(), + [], + ]); + expect(exitCode).toBe(0); + }); + + // A manifest request that failed before its body was parsed never gave its `NetworkTask` + // back. The pool has 128 inline slots and allocates after that, so LSAN sees the lookups + // after the first 128. Timeout: the leak check at exit takes about 3 s on a debug build, + // and a detected leak makes llvm-symbolizer load the DWARF of the whole binary. + describe.each(["404", "a dropped connection"] as const)("%s does not leak the network task", name => { + test.skipIf(!isASAN)( + "LSAN", + async () => { + using registry = flakyRegistry(failures[name].respond); + const { stdout, stderr, exitCode } = await run( + registry, + ` + for (let i = 0; i < 160; i++) { + try { require("flaky-pkg-" + i); } catch {} + } + console.log("done"); + `, + { + ASAN_OPTIONS: [bunEnv.ASAN_OPTIONS, "detect_leaks=1"].filter(Boolean).join(":"), + LSAN_OPTIONS: `use_registers=0:print_suppressions=0:suppressions=${join(import.meta.dirname, "../../leaksan.supp")}`, + }, + ); + expect({ stdout, stderr, exitCode }).toEqual({ + stdout: "done", + stderr: expect.not.stringContaining("LeakSanitizer: detected memory leaks"), + exitCode: 0, + }); + }, + 30_000, + ); + }); +});