NDJSON stream decoder leaks source on parse failure and ignores reader backpressure #4649

Failure

In Eve 0.73.0 (Node 26.8.2, Linux, Local World), parseNdjsonStream in dist/src/execution/ndjson-stream.js releases its source reader when JSON parsing or the event normalizer throws, but never cancels the source. Cancelling the errored output afterwards rejects without invoking the underlying cancel hook. A Local World source therefore retains its chunk:<stream> and close:<stream> listeners and 100 ms directory poller.

An isolated reproduction using the bundled createWorld and parseNdjsonStream left 12 listeners after 12 failed decodes. Twelve successful reads followed by cancellation left zero. This is an error-path cleanup defect; it does not require a model or HTTP server.

The same parser also reads eagerly in start() for the lifetime of the stream. A downstream reader that stops consuming does not apply backpressure to decoding: the parser keeps parsing and enqueuing events. On a long session that can retain the entire decoded replay in addition to the underlying stream buffers.

Minimal reproduction

Run this from a project with eve@0.73.0 installed:

import { createRequire } from 'node:module';
import { dirname, join } from 'node:path';
import { pathToFileURL } from 'node:url';
const require = createRequire(import.meta.url);
const root = dirname(require.resolve('eve/package.json'));
const { parseNdjsonStream } = await import(pathToFileURL(join(root, 'dist/src/execution/ndjson-stream.js')));
let cancelled = 0;
for (let i = 0; i < 12; i++) {
  const source = new ReadableStream({
    start(c) { c.enqueue(new TextEncoder().encode('invalid json\n')); },
    cancel() { cancelled++; }
  });
  const reader = parseNdjsonStream(() => source).getReader();
  await reader.read().catch(() => {});
  await reader.cancel().catch(() => {});
  reader.releaseLock();
}
console.log({ cancelled }); // actual 0; expected 12

The event normalizer throwing with syntactically valid JSON produces the same result.

Expected

Cancel and release the source on decode/normalization failure, preserving the original error. Decode from pull() so the output consumer controls read-ahead. Keep ordinary cancellation and final-line/UTF-8 handling intact.

Related: #199 (closed; its cancellation hook is present in this release), #2880 (similar listener warnings, different demonstrated trigger). The existing fixes do not cover this reproduction. This report does not claim that decoding failure explains every observed listener warning or all memory growth.