Stream.interruptWhenDeferred doesn't instantly interrupt the stream

[dilame]

What version of Effect is running?

3.4.6

What steps can reproduce the bug?

When using Stream.interruptWhenDeferred in conjunction with Deferred.succeed, the stream is not interrupted as expected. Instead of the stream being interrupted after the first element, it continues to process all elements. This behavior is contrary to the expected functionality of Stream.interruptWhenDeferred.

import { Deferred, Effect, Stream } from 'effect';
import * as console from 'node:console';

const s1 = Stream.make(1, 2, 3)

const program = Effect.gen(function* () {
  const interrupt = yield* Deferred.make<void>();
  return yield* s1.pipe(
    Stream.tap(() => Deferred.succeed(interrupt, void 0)),
    Stream.interruptWhenDeferred(interrupt),
    Stream.runCollect,
  );
});

Effect.runPromise(program).then(
  (r) => console.log(r),
  (e) => console.error(e),
);

What is the expected behavior?

{ _id: 'Chunk', values: [ 1 ] }

What do you see instead?

{ _id: 'Chunk', values: [ 1, 2, 3 ] }

Additional information

This issue affects the ability to properly interrupt streams using Deferred, leading to potential performance and logical errors in applications relying on this behavior.

[mikearnaldi]

The stream is fully synchronous and there is no chance of interrupting a sync op, if you add a delay it works as expected:

import { Deferred, Effect, Stream } from "effect"
import * as console from "node:console"

const s1 = Stream.make(1, 2, 3).pipe(
  Stream.tap(() => Effect.sleep("100 millis"))
)

const program = Effect.gen(function*() {
  const interrupt = yield* Deferred.make<void>()
  return yield* s1.pipe(
    Stream.tap(() => Deferred.succeed(interrupt, void 0)),
    Stream.interruptWhenDeferred(interrupt),
    Stream.runCollect
  )
})

Effect.runPromise(program).then(
  (r) => console.log(r),
  (e) => console.error(e)
)

[mikearnaldi]

Similarly if you add a yield with a lower priority than 0:

import { Deferred, Effect, Stream } from "effect"
import * as console from "node:console"

const s1 = Stream.make(1, 2, 3).pipe(
  Stream.tap(() => Effect.yieldNow({ priority: 1 }))
)

const program = Effect.gen(function*() {
  const interrupt = yield* Deferred.make<void>()
  return yield* s1.pipe(
    Stream.tap(() => Deferred.succeed(interrupt, void 0)),
    Stream.interruptWhenDeferred(interrupt),
    Stream.runCollect
  )
})

Effect.runPromise(program).then(
  (r) => console.log(r),
  (e) => console.error(e)
)

[IMax153]

Going to close this as it's working as intended.