streams: reader-drop race loses an owned ReadableStream element without disposal #342

Summary

batches()'s reader/gone race silently drops an already-enqueued own
element instead of releasing it, when the consumer drops its handle in the
same turn a ReadableStream controller enqueues a value.

Location

const r = await Promise.race([reader.read(), gone]);
if (r === READER_GONE) return;

Authority

contracts/embedder-api.md, "Streams of resources" (stream<own<R>>):

Untaken elements are destroyed. Failed chunk conversion releases its
converted prefix; short writes, cancellation, and peer faults release the
undelivered tail. Cleanup attempts every element and preserves an existing
operation error; a standalone cleanup error is reported.

An element the pump never delivers to the guest is "untaken" by this
definition regardless of which side loses the race — the contract does not
carve out an exception for "the read had already resolved when reader-gone
also resolved."

Expected

When the consumer's handle drops (out.drop()) in the same turn a pending
reader.read() settles with a real value, one of two outcomes satisfies the
contract:

• the pump takes the element and then tears it down through the untaken-tail
  path (disposed=1), or
• the pump cancels the reader before the read can resolve, so the value never
  reaches the pump and the source cancel() hook fires (cancelled=1,
  disposed deferred to the source's own cleanup).

Either way, the Ticket element must not simply vanish unreleased.

Actual

Promise.race([reader.read(), gone]) resolves to gone because it was
already-settled/faster, and the branch returns immediately without ever
awaiting or inspecting the reader.read() result. The already-resolved read
promise (and the Ticket instance it carries) is discarded with no
[Symbol.dispose]() call and no reader.cancel() — the finally block's
if (!completed) await reader.cancel() still runs, but by then the value was
already pulled off the stream via read(), so cancel() on an empty reader
does not recover it and the source's cancel() hook never fires either
(observed cancelled=1 below is from a separate prior scenario in the
repro; see actual output).

Verified twice with a fresh Translator instance (deterministic, not a
transient scheduling fluke):

{"disposed":0,"cancelled":1,"locked":false}

disposed=0 — the enqueued Ticket was never disposed.
cancelled=1 — misleading: this counts the source's cancel() invocation,
which does fire (the finally cancels the reader), but arrives too late to
stop the value that was already resolved via read() and thrown away by the
race. The element itself is unreachable and unreleased.

Repro

Adapting the verified scratch harness to repo-relative paths (run from the
repository root):

// tools/repro-stream-reader-drop.ts
import { instantiate } from "./runtime/src/embedder/mod.ts";
import { Translator } from "./runtime/src/shim/mod.ts";

const translator = await Translator.create(
  await Deno.readFile("./translator/translator_shim.wasm"),
);
const turn = () => new Promise<void>((r) => setTimeout(r, 0));

async function fixture(name: string, imports: Record<string, unknown> = {}) {
  const componentBytes = await Deno.readFile(
    `./runtime/tests/embedder/${name}.wasm`,
  );
  const { plan, adapters } = translator.translate(componentBytes);
  return await instantiate({ plan, adapters, componentBytes }, imports, {
    jspi: false,
  });
}

let disposed = 0, cancelled = 0;
class Ticket {
  [Symbol.dispose]() {
    disposed++;
  }
}
const c = await fixture("stream-host", { "host:streams/api": { ticket: Ticket } });
let ctl!: ReadableStreamDefaultController<Ticket>;
const src = new ReadableStream<Ticket>({
  start(c) {
    ctl = c;
  },
  cancel() {
    cancelled++;
  },
});
const out = await c.exports.passTickets(src);
await turn();
out.drop();
ctl.enqueue(new Ticket());
await turn();
console.log(
  "Web Stream reader-drop/late-enqueue:",
  JSON.stringify({ disposed, cancelled, locked: src.locked }),
);

$ deno run --allow-read tools/repro-stream-reader-drop.ts
Web Stream reader-drop/late-enqueue: {"disposed":0,"cancelled":1,"locked":false}

Expected disposed:1 (or an equivalent guaranteed release); got disposed:0.

Suggested fix direction

Don't let gone short-circuit past an already-resolved reader.read(). One
shape: race first only to detect reader-loss, then drain any read that had
already settled concurrently before returning, e.g.

const raced = await Promise.race([
  reader.read().then((r) => ({ kind: "read" as const, r })),
  gone.then(() => ({ kind: "gone" as const })),
]);
if (raced.kind === "gone") {
  // A read may have already resolved in the same microtask turn; drain it
  // so its element reaches the untaken-tail release path instead of being
  // silently discarded.
  ...
  return;
}

The exact mechanism is an implementation choice; the observable requirement
is that no stream<own<R>> element that a producer successfully handed to
read() can vanish without either reaching the guest or being released
through the untaken-element path.

Environment

• Baseline commit: 1a4f5e1c6f3fc906ce576acdc4499ef2745d6e41
• Verified twice with a fresh Translator instance per run (ruled out
  cross-run state contamination)