Stream.timeoutFail doesn't work for hanging Node Readable

[stebl]

What version of Effect is running?

3.13.2

What steps can reproduce the bug?

I can reproduce this in jest,

it('given a hanging stream, should timeout', async () => {
  jest.useFakeTimers();

  const readable = new Readable({
    read() {
      /**
       * Override _read.  Keep the stream open by never pushing null to the buffer
       */
    },
  });
  readable.push(Buffer.from(`1234567890`));

  const stream = () =>
    pipe(
      Stream.fromAsyncIterable(readable, (e) => e),
      Stream.timeoutFail(() => new Error('Timeout'), '2 seconds'),
      Stream.runForEachChunk((chunk) =>
        Effect.sync(() => {
          console.log(chunk);
        }),
      ),
    );

  const program = stream();

  const promise = Effect.runPromise(program)
    .then(console.log)
    .catch((e) => expect(e.message).toBe('Timeout'));

  await jest.advanceTimersByTimeAsync(10 * 60 * 1000); // 10 minutes
  expect.assertions(1);
  await promise;  // test times out, never resolves
});

What is the expected behavior?

Expect the effect to result in a timeout error

What do you see instead?

Instead the test times out after 45 seconds (our test config)

Additional information

Any workarounds here?

Stream.fromReadableStream isn't applicable here b/c the Web API stream is not the same as the node stream.

The timeout works when a generator is provided, example below. Unfortunately the code is receiving a Readable from an HTTP request, so the difference in behavior between Readable and Iterator is likely to result in the stream hanging in production.

it.only('given a generator that stalls, will timeout', async () => {
  jest.useFakeTimers();

  async function* generate() {
    yield 1;
    yield 2;
    await new Promise((resolve) => setTimeout(resolve, 5000));
    yield 3;
    yield 4;
  }

  const stream = () =>
    pipe(
      Stream.fromAsyncIterable(generate(), (e) => e),
      Stream.timeoutFail(() => new Error('Timeout'), '2 seconds'),
      Stream.runForEachChunk((chunk) =>
        Effect.sync(() => {
          console.log(chunk);
        }),
      ),
    );

  const program = stream();

  const promise = Effect.runPromise(program)
    .then(console.log)
    .catch((e) => expect(e.message).toBe('Timeout'));

  await jest.advanceTimersByTimeAsync(10 * 60 * 1000 + 1500);
  expect.assertions(1);
  await promise;  // resolves with caught error
});