Join effect worker threads before teardown frees the channel (#124)

* Join effect worker threads before teardown frees the channel

- Store spawn/fetch/file worker handles on their slots instead of detaching, join them in reclaim and unconditionally in deinit: the old ~5s give-up abandoned workers still holding slot/queue/io pointers into memory the owner frees right after, which is how a torn-down harness segfaulted the next test inside a stale slot's child_mutex.
- Spawn each child into its own POSIX process group and kill the group on cancel/teardown, so shell-wrapped commands' grandchildren cannot hold the stdout pipe open and stall the worker (and now the join) past the cancel.
- Regression tests: teardown mid-stream joins and returns promptly with no running slots, and an 8-round teardown storm (immediate and cancel-racing) over recycled harness memory.

* Make the ts-core e2e battery tolerant of congested CI runners

- Scale the host battery's real-child wait budget 10x (20s -> 200s): the waits prove correctness and poll, so a healthy run still returns in milliseconds while a loaded runner gets the slack it needs to schedule and reap /bin/sh children.
- A blown wait budget now tears the effects channel down (kill children, join workers) before surfacing TestTimedOut, so one slow test can never cascade a straggling child into the next test's harness.
- The soundboard dispatch-latency test asserts the best of three attempts with unchanged budgets: scheduler contention only adds time, so real regressions still fail while parallel-suite noise no longer does.

* Bound file-worker teardown: interrupt, then abandon-and-leak, never hang

- Effects.deinit joined every worker unconditionally, but a file worker blocked in I/O that nothing converges (a write to a FIFO with no reader, a stalled network filesystem) made teardown hang forever behind it; file workers now get an injectable budget (file_join_deadline_ms, 15s default) with a best-effort cancel of the blocked task at the halfway mark, and past it teardown detaches the thread, warns once naming the stuck op and path, and deliberately leaks everything the worker can still reach (its context, its data buffer, the executor io) so the owner can free the channel safely. Spawn and fetch joins stay unconditional.
- The blocking phase moved out of the channel: each file worker supervises its op as a cancelable Io task (mirroring fetchWorkerMain, which also supplies the platform interruption: SIG.IO signaling on POSIX, NtCancelSynchronousIoFile on Windows, via the threaded io) against a heap FileWorkerContext holding the path copy, buffer, and a commit/abandon handshake - the worker only touches the slot and queue after committing under the context mutex, so an abandoned worker that wakes later walks only leaked memory.
- Regression tests: a write against a reader-less FIFO is abandoned within a tiny injected deadline (loud counter, healthy process after, the woken worker exercised under the leak invariant), the default interruption path joins the same posture with no leak, and the happy-path teardown pins the abandon counter at zero.

* Move the abandonable leak into process-lifetime storage and reject uncancelable fetches

- Allocate FileWorkerContext, its private data buffer, and the shared IoThreaded executor from a process-lifetime allocator (page_allocator) so an abandoned worker never walks memory that dies with the owner's arena or GPA; the happy path frees all three through the same seam (joinWorker, deinit), and a committed read publishes its bytes into the slot's channel-owned delivery buffer.
- Refuse a fetch whose exchange cannot start as a cancelable task instead of running it inline: an inline exchange observes neither cancel nor the timeout and would hang deinit's unconditional fetch join, so the honest terminal is one journaled .rejected (replay reproduces it like any transport failure), with an injectable fetch_concurrent_start seam and a warn-once counter.
- Pin both with tests: an arena-backed channel abandons a FIFO-stuck worker, dies, and the woken worker walks only process-lived memory (leak-checked happy path alongside); the fetch rejection seam delivers exactly one .rejected and teardown returns promptly.

* Bound spawn-worker teardown: interrupt, then abandon-and-leak, never hang

- Group-kill cannot guarantee spawn convergence: a descendant that leaves the child's process group (setsid, a shell's set -m background job) keeps the inherited stdout write end open, so the worker's read never sees EOF and deinit's unconditional join hung forever; spawn workers now get the file workers' full discipline — an injectable budget (spawn_join_deadline_ms, same 15s default) with a best-effort cancel of the blocked task at the halfway mark, then a detach-warn-and-leak abandon with its own abandoned_spawn_workers counter, so spawn, fetch, and file teardown all share one terminal guarantee: bounded return, and every byte a live thread can still touch stays valid forever (Windows, where the direct-handle terminate never reached descendants, is bounded by the same net; job objects remain the future strengthening).
- The blocking phase's world moved out of the channel into a process-lived SpawnWorkerContext (argv/stdin copies, the published-child kill handshake, private framing/collect buffers, the stderr tail ring, drop accounting), supervised as a cancelable Io task exactly like file ops; unlike a file op a spawn DELIVERS while it blocks, so streaming lines enqueue under the context's abandon fence (produceSpawnLine re-checks the abandon with every channel touch) and the committed epilogue publishes collect payloads into the slot's channel-owned delivery buffer, preserving cancel semantics, the stale-event window, and record/replay byte-identity.
- Regression tests pin the escaped-descendant shape (/bin/bash -c 'set -m; sleep 300 & echo $!' — portable setsid): the default interruption path joins it with no leak, the disabled-interruption path abandons it within a tiny injected deadline (loud counter, healthy process after, the woken worker proven to reap its zombie child through only leaked memory), the arena-lifetime test wakes an abandoned worker after the owner's allocator died, and the existing mid-stream teardown pins abandon-count zero.

Author: ctate

changelog.d/spawn-e2e-stability.md

@@ -0,0 +1,4 @@
+fix: **Spawn teardown crash window closed**: the effects channel now joins every spawn, fetch, and file worker thread that converges before its teardown returns (previously it gave up after ~5s and abandoned them), so cancelling or quitting while a real child is still streaming can no longer leave a stale worker writing into freed memory.
+- **Spawn cancel reaches the whole process tree**: each spawned child runs in its own process group and cancel/teardown signals the group (POSIX), so shell-wrapped commands (`sh -c "a; b"`) no longer leave orphaned grandchildren holding the stream open past the cancel.
+- **Every worker class tears down bounded**: spawn, fetch, and file workers now share one terminal guarantee — teardown returns within a budget and never frees memory a live thread can still touch. A file worker stuck in blocking I/O that nothing can converge (a write to a FIFO with no reader, a stalled network filesystem), or a spawn worker held hostage by a descendant that escaped the kill's process group (`setsid` daemonization, a shell's `set -m` background job) while holding the stdout pipe open, no longer hangs teardown: teardown interrupts the blocked syscall best-effort at half its budget and, past the full 15s, abandons the worker with one warning and a small deliberate leak. Everything an abandoned worker can still reach lives in process-lifetime storage, so the leak stays safe even when the app tears down the allocator behind the channel right after.
+- **A fetch that cannot start cancellably is rejected, never run inline**: when the executor cannot start the exchange as a cancelable task, the fetch now delivers one honest `.rejected` terminal instead of silently running an exchange that would evade `cancel`, the timeout, and teardown's join.

src/runtime/effects.zig

@@ -9,6 +9,7 @@ const app_manifest = @import("app_manifest");
 const core = @import("core.zig");
 const ui_app_model = @import("ui_app.zig");
 const effects_mod = @import("effects.zig");
+const clock_mod = @import("clock.zig");
 
 const canvas_label = "fetch-canvas";
 
@@ -980,6 +981,47 @@ test "real fetch reports connection refused as connect_failed" {
     try std.testing.expectEqual(@as(usize, 0), h.app_state.model.body_len);
 }
 
+test "a fetch whose exchange cannot start cancellably is rejected, never run inline" {
+    var h = try Harness.create();
+    defer h.destroy();
+    const fx = &h.app_state.effects;
+    // Pin the no-capacity path through the injectable seam (the real
+    // trigger — the executor refusing `std.Io.concurrent` — needs
+    // resource exhaustion). If a regression ran the exchange inline
+    // instead, this loopback connect would fail fast as
+    // `.connect_failed` and the assert below would catch the lie.
+    fx.fetch_concurrent_start = false;
+
+    var url_buffer: [128]u8 = undefined;
+    test_url = std.fmt.bufPrint(&url_buffer, "http://127.0.0.1:9/", .{}) catch unreachable;
+    test_method = .GET;
+    test_headers = &.{};
+    test_payload = null;
+    test_timeout_ms = effects_mod.default_effect_fetch_timeout_ms;
+    try h.app_state.dispatch(&h.harness.runtime, 1, .start);
+    try waitForResponse(&h);
+
+    // Exactly one honest `.rejected` terminal — the fetch never
+    // started — counted through the rejection seam.
+    try std.testing.expectEqual(effects_mod.EffectFetchOutcome.rejected, h.app_state.model.outcome.?);
+    try std.testing.expectEqual(@as(usize, 1), h.app_state.model.rejected_count);
+    try std.testing.expectEqual(@as(u16, 0), h.app_state.model.status);
+    try std.testing.expectEqual(@as(usize, 0), h.app_state.model.body_len);
+    try std.testing.expectEqual(@as(u32, 1), fx.fetch_start_rejections.load(.acquire));
+    try std.testing.expectEqual(@as(usize, 0), h.app_state.effects.activeCount());
+
+    // Teardown returns promptly: no uncancelable inline exchange is
+    // left for the unconditional fetch join to wait on.
+    const start_ns = clock_mod.monotonicNanoseconds();
+    fx.deinit();
+    const elapsed_ms = (clock_mod.monotonicNanoseconds() - start_ns) / std.time.ns_per_ms;
+    try std.testing.expect(elapsed_ms < 10_000);
+    for (&fx.slots) |*slot| {
+        try std.testing.expect(slot.worker_thread == null);
+        try std.testing.expect(slot.state.load(.acquire) != .running);
+    }
+}
+
 test "real fetch times out against a hanging route" {
     var h = try Harness.create();
     defer h.destroy();

src/runtime/effects_fetch_tests.zig

@@ -11,6 +11,7 @@ const app_manifest = @import("app_manifest");
 const core = @import("core.zig");
 const ui_app_model = @import("ui_app.zig");
 const effects_mod = @import("effects.zig");
+const clock_mod = @import("clock.zig");
 
 const canvas_label = "file-canvas";
 
@@ -346,6 +347,16 @@ test "real executor writes a file (creating parent dirs) and reads it back" {
     try h.app_state.dispatch(&h.harness.runtime, 1, .load);
     try waitForRealResult(&h, 4);
     try std.testing.expectEqualStrings("{\"n\":2}", h.app_state.model.bytesPrefix());
+
+    // Happy-path teardown pin: quick local ops join promptly and the
+    // abandon safety net never fires (no leak, no detached thread).
+    const fx = &h.app_state.effects;
+    fx.deinit();
+    try std.testing.expectEqual(@as(u32, 0), fx.abandoned_file_workers);
+    for (&fx.slots) |*slot| {
+        try std.testing.expect(slot.worker_thread == null);
+        try std.testing.expect(slot.state.load(.acquire) != .running);
+    }
 }
 
 test "real executor reports missing files as not_found" {
@@ -386,6 +397,215 @@ test "real executor cuts over-bound reads with outcome truncated" {
     );
 }
 
+// --------------------------------------------------- bounded teardown
+
+/// Create a FIFO at `path` (POSIX-only; callers gate on the platform).
+/// A file write against it blocks forever inside `open(O_WRONLY)`
+/// while no reader exists — the uninterruptible-blocking-I/O posture
+/// the teardown deadline exists for.
+fn makeFifo(path: []const u8) !void {
+    var command_buffer: [512]u8 = undefined;
+    const command = try std.fmt.bufPrint(&command_buffer, "mkfifo '{s}'", .{path});
+    const result = try std.process.run(std.testing.allocator, std.testing.io, .{
+        .argv = &.{ "/bin/sh", "-c", command },
+    });
+    defer std.testing.allocator.free(result.stdout);
+    defer std.testing.allocator.free(result.stderr);
+    try std.testing.expect(result.term == .exited and result.term.exited == 0);
+}
+
+test "teardown abandons a file worker stuck on a reader-less FIFO and leaks its world loudly" {
+    if (builtin.os.tag == .windows) return error.SkipZigTest;
+    const io = std.testing.io;
+    var tmp = std.testing.tmpDir(.{});
+    defer tmp.cleanup();
+
+    var fifo_path_buffer: [256]u8 = undefined;
+    const fifo_path = try std.fmt.bufPrint(&fifo_path_buffer, ".zig-cache/tmp/{s}/stuck.fifo", .{tmp.sub_path[0..]});
+    try makeFifo(fifo_path);
+
+    var h = try Harness.create();
+    defer h.destroy();
+    const fx = &h.app_state.effects;
+    // Tiny injected budget, interruption disabled: this test pins the
+    // SAFETY NET (abandon-and-leak). The interruption path that
+    // normally converges first has its own test below.
+    fx.file_join_deadline_ms = 300;
+    fx.file_join_interrupt = false;
+
+    test_path = fifo_path;
+    test_bytes = "never delivered";
+    try h.app_state.dispatch(&h.harness.runtime, 1, .save);
+    // Let the worker reach the blocking open. Not required for the
+    // abandon to fire — with interruption off, any still-running
+    // posture past the deadline is abandoned — but it exercises the
+    // real blocked shape.
+    try std.Io.sleep(io, std.Io.Duration.fromMilliseconds(50), .awake);
+
+    const start_ns = clock_mod.monotonicNanoseconds();
+    fx.deinit();
+    const elapsed_ms = (clock_mod.monotonicNanoseconds() - start_ns) / std.time.ns_per_ms;
+
+    // Teardown returned on the budget's order of magnitude (generous
+    // bound for congested runners) instead of hanging behind the FIFO
+    // forever...
+    try std.testing.expect(elapsed_ms < 10_000);
+    // ...and abandoned exactly the stuck worker, loudly through the
+    // counter seam, leaving no joinable thread and no running slot —
+    // the owner may free the channel's memory right now.
+    try std.testing.expectEqual(@as(u32, 1), fx.abandoned_file_workers);
+    for (&fx.slots) |*slot| {
+        try std.testing.expect(slot.worker_thread == null);
+        try std.testing.expect(slot.state.load(.acquire) != .running);
+    }
+
+    // The process is healthy after the leak: a fresh channel runs a
+    // real file round trip and its own safety net stays quiet.
+    var h2 = try Harness.create();
+    defer h2.destroy();
+    var path_buffer: [256]u8 = undefined;
+    test_path = try std.fmt.bufPrint(&path_buffer, ".zig-cache/tmp/{s}/healthy.json", .{tmp.sub_path[0..]});
+    test_bytes = "{\"alive\":true}";
+    try h2.app_state.dispatch(&h2.harness.runtime, 1, .save);
+    try waitForRealResult(&h2, 1);
+    try std.testing.expectEqual(effects_mod.EffectFileOutcome.ok, h2.app_state.model.last_outcome.?);
+    try std.testing.expectEqual(@as(u32, 0), h2.app_state.effects.abandoned_file_workers);
+
+    // Wake the abandoned worker under the leak invariant: opening the
+    // FIFO's read end completes its blocked open; it performs its
+    // write against its (leaked, still-valid) context and buffer,
+    // closes the FIFO, finds itself abandoned, and exits without
+    // touching the torn-down channel — if the leak were not honored,
+    // this is where a use-after-free would crash the test. Draining to
+    // EOF keeps the read end open across the worker's write (a write
+    // into a reader-less pipe raises SIGPIPE, and the io's no-op
+    // handler is not installed between tests) and proves the worker
+    // really woke: EOF only arrives once it opened and closed the
+    // write end.
+    var reader = try std.Io.Dir.cwd().openFile(io, fifo_path, .{});
+    defer reader.close(io);
+    var drain_buffer: [128]u8 = undefined;
+    while (true) {
+        const read_slices: [1][]u8 = .{&drain_buffer};
+        const count = reader.readStreaming(io, &read_slices) catch break;
+        if (count == 0) break;
+    }
+}
+
+test "an abandoned file worker survives the owner's allocator dying: its leak is process-lived only" {
+    if (builtin.os.tag == .windows) return error.SkipZigTest;
+    const io = std.testing.io;
+    var tmp = std.testing.tmpDir(.{});
+    defer tmp.cleanup();
+
+    var fifo_path_buffer: [256]u8 = undefined;
+    const fifo_path = try std.fmt.bufPrint(&fifo_path_buffer, ".zig-cache/tmp/{s}/arena.fifo", .{tmp.sub_path[0..]});
+    try makeFifo(fifo_path);
+
+    // The channel — and every caller-side allocation it makes — lives
+    // in an arena backed by the leak-checking testing allocator and is
+    // deinitialized right after teardown: the exact owner-lifetime
+    // posture the abandon leak must survive. The channel struct itself
+    // sits in the arena too, so even the worker's `self` pointer dies
+    // with the owner.
+    const Channel = effects_mod.Effects(FileMsg);
+    var arena = std.heap.ArenaAllocator.init(std.testing.allocator);
+    var arena_live = true;
+    defer if (arena_live) arena.deinit();
+    const fx = try arena.allocator().create(Channel);
+    fx.* = Channel.init(arena.allocator());
+    // Tiny injected budget, interruption disabled: pin the
+    // abandon-and-leak safety net, exactly like the loud-leak test.
+    fx.file_join_deadline_ms = 300;
+    fx.file_join_interrupt = false;
+
+    fx.writeFile(.{ .key = 1, .path = fifo_path, .bytes = "never delivered", .on_result = null });
+    try std.Io.sleep(io, std.Io.Duration.fromMilliseconds(50), .awake);
+
+    fx.deinit();
+    try std.testing.expectEqual(@as(u32, 1), fx.abandoned_file_workers);
+
+    // Kill the owner's allocator: everything the channel ever got from
+    // it — including the channel struct — is gone. The abandoned
+    // worker must not notice: all it can still reach (its context,
+    // that context's buffer, the executor io) is `process_allocator`
+    // storage.
+    arena.deinit();
+    arena_live = false;
+
+    // Wake the abandoned worker under that invariant: opening the
+    // FIFO's read end completes its blocked open; it writes through
+    // its process-lived context and buffer, finds itself abandoned,
+    // and exits without touching the dead arena — if any of its
+    // reachable memory were caller-allocated, this walk is where the
+    // use-after-free would crash the test. Draining to EOF keeps the
+    // read end open across the worker's write and proves the worker
+    // really woke.
+    var reader = try std.Io.Dir.cwd().openFile(io, fifo_path, .{});
+    defer reader.close(io);
+    var drain_buffer: [128]u8 = undefined;
+    while (true) {
+        const read_slices: [1][]u8 = .{&drain_buffer};
+        const count = reader.readStreaming(io, &read_slices) catch break;
+        if (count == 0) break;
+    }
+
+    // And the happy path still frees everything through the same
+    // seams: a fresh channel backed DIRECTLY by the testing allocator
+    // runs a real write to completion and tears down joined — the
+    // leak check at test end guards the caller-side allocations, and
+    // `joinWorker`/`deinit` return the context, worker buffer, and
+    // executor io to `process_allocator`.
+    var healthy = Channel.init(std.testing.allocator);
+    defer healthy.deinit();
+    var path_buffer: [256]u8 = undefined;
+    const healthy_path = try std.fmt.bufPrint(&path_buffer, ".zig-cache/tmp/{s}/healthy.json", .{tmp.sub_path[0..]});
+    healthy.writeFile(.{ .key = 2, .path = healthy_path, .bytes = "{\"alive\":true}", .on_result = null });
+    var waited_ms: usize = 0;
+    while (waited_ms < 20_000) : (waited_ms += 10) {
+        if (healthy.hasPending()) break;
+        try std.Io.sleep(io, std.Io.Duration.fromMilliseconds(10), .awake);
+    }
+    try std.testing.expect(healthy.hasPending());
+    healthy.deinit();
+    try std.testing.expectEqual(@as(u32, 0), healthy.abandoned_file_workers);
+}
+
+test "teardown interrupts a file worker stuck on a reader-less FIFO and joins it with no leak" {
+    if (builtin.os.tag == .windows) return error.SkipZigTest;
+    const io = std.testing.io;
+    var tmp = std.testing.tmpDir(.{});
+    defer tmp.cleanup();
+
+    var fifo_path_buffer: [256]u8 = undefined;
+    const fifo_path = try std.fmt.bufPrint(&fifo_path_buffer, ".zig-cache/tmp/{s}/interrupted.fifo", .{tmp.sub_path[0..]});
+    try makeFifo(fifo_path);
+
+    var h = try Harness.create();
+    defer h.destroy();
+    const fx = &h.app_state.effects;
+    // Interruption stays on (the default): the best-effort cancel at
+    // the halfway mark must interrupt the blocked open and JOIN the
+    // worker — the abandon safety net must never fire here.
+    fx.file_join_deadline_ms = 2_000;
+
+    test_path = fifo_path;
+    test_bytes = "never delivered";
+    try h.app_state.dispatch(&h.harness.runtime, 1, .save);
+    try std.Io.sleep(io, std.Io.Duration.fromMilliseconds(50), .awake);
+
+    fx.deinit();
+
+    // Joined, not leaked: the syscall interruption converged the
+    // worker inside the deadline (deinit is deadline-bounded either
+    // way, so reaching these asserts already proves it returned).
+    try std.testing.expectEqual(@as(u32, 0), fx.abandoned_file_workers);
+    for (&fx.slots) |*slot| {
+        try std.testing.expect(slot.worker_thread == null);
+        try std.testing.expect(slot.state.load(.acquire) != .running);
+    }
+}
+
 test "a cancel racing a finished file effect still reports one cancelled terminal" {
     var tmp = std.testing.tmpDir(.{});
     defer tmp.cleanup();