NodeSink.fromWritable leaves the Writable open when the upstream fails or the sink is interrupted #9352

What version of Effect is running?

[email protected], @effect/[email protected], on Node 24.21.0 and 26.11.1. main still has it after #9333.

What steps can reproduce the bug?

NodeSink.fromWritable ends the Writable it gets from evaluate when the upstream completes. If the upstream fails, or the sink is interrupted (Effect.timeout, a lost race, a closing scope), it neither ends nor destroys it. With evaluate: () => fs.createWriteStream(path) the caller holds no reference to close it, so every failed or interrupted run leaks a file descriptor.

import * as NodeSink from "@effect/platform-node-shared/NodeSink"
import { Effect, Stream } from "effect"
import * as fs from "node:fs"
import * as os from "node:os"
import * as path from "node:path"

const run = (name, upstream) =>
  Effect.gen(function*() {
    let file
    const sink = NodeSink.fromWritable({
      evaluate: () => (file = fs.createWriteStream(path.join(os.tmpdir(), `${name}.txt`))),
      onError: (e) => e
    })
    yield* upstream.pipe(Stream.run(sink), Effect.timeout("50 millis"), Effect.exit)
    yield* Effect.sleep("200 millis")
    console.log(name, { ended: file.writableEnded, destroyed: file.destroyed, fdOpen: file.fd !== null })
  })

await Effect.runPromise(Effect.gen(function*() {
  yield* run("complete", Stream.make("a"))
  yield* run("fail", Stream.make("a").pipe(Stream.concat(Stream.fail(new Error("boom")))))
  yield* run("interrupt", Stream.make("a").pipe(Stream.concat(Stream.never)))
}))

complete { ended: true, destroyed: true, fdOpen: false }
fail { ended: false, destroyed: false, fdOpen: true }
interrupt { ended: false, destroyed: false, fdOpen: true }

What is the expected behavior?

When the sink fails or is interrupted, it ends or destroys the writable, as it ends it on completion. With endOnDone: false, which NodeStdio uses for stdout and stderr, the writable should stay open.

The siblings already do this:

• Sink.fromWritableStream aborts its WritableStream on upstream failure and on interruption (Channel.ts#L1774-L1780). The tests "aborts instead of closing on upstream failure" and "… on interruption" pin it (Sink.test.ts#L23-L79).
• NodeStream.fromReadable destroys its readable on any exit unless closeOnDone: false (NodeStream.ts#L395-L405).

In v3 (@effect/[email protected]), the fail case above ended the writable (internal/stream.ts#L210-L224).

What do you see instead?

The writable stays open: writableEnded and destroyed are both false, and a file stream's descriptor stays open for the life of the process.

Additional information

writable.end() is reached only through Pull.catchDone, that is, only when the upstream ends (NodeSink.ts#L117-L131). A failure propagates past it, and on interruption the cleanup only removes the drain listener (#L110-L113). On main, #9333 keeps that shape: end() is still only in Pull.catchDone, and the new Effect.ensuring only removes listeners. fromDuplex and pipeThroughDuplex are not affected, because their readable side destroys the duplex on exit.

  it.effect.each(["done", "upstream_failure", "interrupt"] as const)(
    "ends or destroys the writable when the sink exits: %s",
    (mode) =>
      Effect.gen(function*() {
        const error = new Error("upstream failed")
        const written = yield* Deferred.make<void>()
        const items: Array<string> = []
        const writable = new Writable({
          write(chunk, _encoding, callback) {
            items.push(chunk.toString())
            Deferred.doneUnsafe(written, Effect.void)
            callback()
          }
        })
        yield* Effect.addFinalizer(() => Effect.sync(() => writable.destroy()))
        const upstream = mode === "done"
          ? Stream.make("a")
          : Stream.make("a").pipe(Stream.concat(mode === "upstream_failure" ? Stream.fail(error) : Stream.never))
        const fiber = yield* upstream.pipe(
          Stream.run(NodeSink.fromWritable({ evaluate: () => writable, onError: (error) => error })),
          Effect.forkScoped
        )
        yield* Deferred.await(written)
        if (mode === "interrupt") {
          yield* Fiber.interrupt(fiber)
        }
        const exit = yield* Fiber.await(fiber)
        assert.deepEqual(items, ["a"])
        if (mode === "done") {
          assert.deepEqual(exit, Exit.void)
        } else if (mode === "upstream_failure") {
          assert.deepEqual(exit, Exit.fail(error))
        } else {
          assert.isTrue(Exit.hasInterrupts(exit))
        }
        assert.deepEqual({ endedOrDestroyed: writable.writableEnded || writable.destroyed }, {
          endedOrDestroyed: true
        })
      })
  )