fetch: don't double-release the request-stream ref when a native body sink ends inline

src/runtime/webcore/fetch/FetchTasklet.rs

@@ -668,6 +668,16 @@ impl FetchTasklet {
         }) {
             crate::webcore::readable_stream::NativeWireResult::Wired => return,
             crate::webcore::readable_stream::NativeWireResult::EndedInline(err) => {
+                // The source finished inside the wire attempt, so leave the
+                // sink in the state `end_from_stream` leaves it: ended, with
+                // the source and task detached. `write_end_request` below is
+                // the single balancing release of the `+1` taken above; a
+                // sink left `ended == false` here would make the terminal
+                // `cancel_request_body_sink` treat it as a live native sink
+                // and release that ref a second time, freeing the tasklet
+                // while it is still in use.
+                sink.ended = true;
+                sink.source.clear();
                 sink.task = None;
                 let err_js = err.map(|err| {
                     let err_js = err.to_js(&global_this);

test/js/web/fetch/fetch-abort-stream-body.test.ts

@@ -55,6 +55,30 @@ test
     expect(exitCode).toBe(0);
   });
 
+// A native ByteStream request body (an upstream response body piped into
+// fetch) that errors or finishes between fetch() and the can_stream tick is
+// ended inline by wire_native_sink. That path released the request-stream ref
+// but left the sink installed as live, so the terminal
+// cancel_request_body_sink released the same ref again and freed the
+// FetchTasklet while the completion path was still using it. ASAN-only: the
+// release build corrupts silently. Details in the fixture.
+test
+  .skipIf(!isASAN)
+  .concurrent("piping an erroring upstream body into fetch does not double-release the tasklet", async () => {
+    await using proc = Bun.spawn({
+      cmd: [bunExe(), join(import.meta.dir, "fetch-stream-body-ended-inline-fixture.ts")],
+      env: { ...bunEnv, ITER: "100" },
+      stdout: "pipe",
+      stderr: "pipe",
+    });
+
+    const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
+
+    expect(stderr).not.toContain("AddressSanitizer");
+    expect(stdout).toBe("done 100\n");
+    expect(exitCode).toBe(0);
+  });
+
 test("aborting fetch with a ReadableStream request body does not double-cancel the sink", async () => {
   await using proc = Bun.spawn({
     cmd: [bunExe(), join(import.meta.dir, "fetch-abort-stream-body-fixture.ts")],