diff --git a/src/jsc/event_loop.zig b/src/jsc/event_loop.zig index f43114231e56..c8076ee67c32 100644 --- a/src/jsc/event_loop.zig +++ b/src/jsc/event_loop.zig @@ -643,6 +643,16 @@ pub fn enqueueTaskConcurrent(this: *EventLoop, task: *ConcurrentTask) void { if (this.virtual_machine.has_terminated) { @panic("EventLoop.enqueueTaskConcurrent: VM has terminated"); } + // A freshly-constructed ConcurrentTask must have `.next` set to `.none` or + // `.auto_delete` (pointer bits zero) before being pushed. `pushBatch` calls + // `PackedNextPtr.setPtr` which *preserves* the low `auto_delete` bit, so if + // `.next` was left undefined the preserved bit is garbage and the event loop + // may call `bun.destroy` on an embedded (interior-pointer) ConcurrentTask. + // Test the raw bits directly rather than calling `getPtr()` — `@ptrFromInt` + // on a misaligned address (e.g. 0xAA..AA) would trip Zig's alignment safety + // check and panic before this diagnostic ever prints. + const next_bits = @intFromEnum(task.next); + bun.assertf((next_bits & ~@as(usize, 1)) == 0, "ConcurrentTask.next must be initialized (use ConcurrentTask.from or struct init) before enqueueTaskConcurrent; got 0x{x:0>16}", .{next_bits}); } if (comptime Environment.isDebug) { diff --git a/src/runtime/node/node_fs_watcher.zig b/src/runtime/node/node_fs_watcher.zig index d62a412625d1..d15d6625ba3d 100644 --- a/src/runtime/node/node_fs_watcher.zig +++ b/src/runtime/node/node_fs_watcher.zig @@ -119,8 +119,7 @@ pub const FSWatcher = struct { if (this.ctx.refTask()) { var that = FSWatchTask.new(this.*); this.count = 0; - that.concurrent_task.task = jsc.Task.init(that); - this.ctx.enqueueTaskConcurrent(&that.concurrent_task); + this.ctx.enqueueTaskConcurrent(that.concurrent_task.from(that, .manual_deinit)); return; } // closed or detached so just cleanEntries diff --git a/test/js/node/watch/fs.watch.concurrent-task.test.ts b/test/js/node/watch/fs.watch.concurrent-task.test.ts new file mode 100644 index 000000000000..178becc2d85a --- /dev/null +++ b/test/js/node/watch/fs.watch.concurrent-task.test.ts @@ -0,0 +1,89 @@ +// Regression test for FSWatchTaskPosix.enqueue() leaving `concurrent_task.next` +// undefined. `PackedNextPtr.setPtr` preserves the low `auto_delete` bit, so when +// `.next` is undefined the preserved bit is garbage; if it reads as 1 the event +// loop calls `bun.destroy` on the *embedded* ConcurrentTask (an interior pointer) +// and corrupts the heap. +// +// The fix uses `ConcurrentTask.from(that, .manual_deinit)` which fully initializes +// both `.task` and `.next`. A debug assertion in `enqueueTaskConcurrent` verifies +// the pointer bits of `.next` are zero before push, which fires on any regression +// of this pattern (Zig's debug `undefined` = 0xAA..AA → non-zero pointer bits). +import { expect, test } from "bun:test"; +import { bunEnv, bunExe, isWindows, tempDir } from "harness"; + +// FSWatchTaskPosix is the POSIX-only code path. +test.skipIf(isWindows)( + "fs.watch: FSWatchTask enqueue fully initializes ConcurrentTask", + async () => { + using dir = tempDir("fswatch-concurrent-task", { + "a.txt": "a", + "b.txt": "b", + "c.txt": "c", + "d.txt": "d", + }); + + // The regression signal here is the debug assertion / no heap corruption in + // FSWatchTask.enqueue(), not event-delivery count. On macOS, directory watches + // route through FSEvents which has ~50ms coalescing latency and async stream + // registration, so we wait for the first event before counting stress rounds + // rather than assuming a fixed number of setImmediate turns is "enough time". + const fixture = /* js */ ` + const fs = require("fs"); + const path = require("path"); + const dir = ${JSON.stringify(String(dir))}; + + let received = 0; + const watchers = []; + // Many watchers → many FSWatchTask.enqueue() calls per batch of fs events. + for (let i = 0; i < 64; i++) { + watchers.push(fs.watch(dir, () => { received++; })); + } + + function done() { + for (const w of watchers) w.close(); + // Allow any in-flight watcher-thread tasks to drain. + setImmediate(() => { + console.log("OK " + received); + process.exit(0); + }); + } + + const files = ["a.txt", "b.txt", "c.txt", "d.txt"]; + function write() { + for (const f of files) fs.writeFileSync(path.join(dir, f), "v" + received); + } + + // Phase 1: write until the first event arrives (condition, not time). + const started = Date.now(); + (async () => { + while (received === 0) { + write(); + await new Promise(r => setImmediate(r)); + // Give up after a generous bound; the regression signal is the + // assertion / no heap corruption, not delivery count, so still pass. + if (Date.now() - started > 10_000) return done(); + } + // Phase 2: now that enqueue() is known to be firing, stress it. + for (let round = 0; round < 50; round++) { + write(); + await new Promise(r => setImmediate(r)); + } + done(); + })(); + `; + + await using proc = Bun.spawn({ + cmd: [bunExe(), "-e", fixture], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + + expect(stderr).toBe(""); + expect(stdout).toStartWith("OK "); + expect(exitCode).toBe(0); + }, + 30_000, +);