platform-node: Worker shutsdown when client side interrupts a response stream

[schickling]

What version of Effect is running?

3.13.8

What steps can reproduce the bug?

Add serialized worker test case to:

  it('Serialized 2', () =>
    Effect.gen(function* (_) {
      const pool = yield* _(EffectWorker.makePoolSerialized({ size: 1 }))
      const people = yield* _(
        pool.execute(new GetPersonById({ id: 123 })),
        Stream.runCollect,
        Effect.timeout(2000),
        Effect.exit,
      )
      console.log(people)
      const people2 = yield* _(pool.execute(new GetPersonById({ id: 123 })), Stream.take(1), Stream.runCollect)
      console.log(people2)
      // assert.deepStrictEqual(Chunk.toReadonlyArray(people), [
      //   new Person({ id: 123, name: "test", data: new Uint8Array([1, 2, 3]) }),
      //   new Person({ id: 123, name: "ing", data: new Uint8Array([4, 5, 6]) })
      // ])
    }).pipe(Effect.scoped, Effect.provide(WorkerLive), Effect.runPromise))

And adjust serialized worker:

const WorkerLive = Runner.layerSerialized(WorkerMessage, {
  GetPersonById: (req) => {
    return Stream.make(
      new Person({ id: req.id, name: 'test', data: new Uint8Array([1, 2, 3]) }),
      new Person({ id: req.id, name: 'ing', data: new Uint8Array([4, 5, 6]) }),
    ).pipe(Stream.schedule(Schedule.fixed('1 second')))
  },

And run 'Serialized 2 test.

What is the expected behavior?

This work properly for platform-browser:

What do you see instead?

But fails when running the same tests via the platform-node worker setup:

My suspicion is that the worker runner doesn't just interrupt the handler but also shutsdown the worker itself.

Additional information

(Please ignore that the test output contains ChildProcessRunnerTest)

No response

[schickling]

Update: Seems like the current platform-node worker behaviour is "flaky" as sometimes I seem to get the expected behaviour. e.g. in this run:

But then it breaks again in the next run:

[IMax153]

Closing this issue as a fix has been implemented 👍

Feel free to re-open or open a new issue if any additional problems persist.