Allow stream sinks to keep streams open

Add StreamSinkOptions.closeStream with a default of true so callers can
preserve caller-owned writable streams during sink disposal. Disposal
still waits for pending and buffered writes and releases the writer lock.

Document the option and cover blocking, non-blocking, and active-flush
disposal behavior.

Closes #203

Assisted-by: Codex:gpt-5.6-sol

Author: dahlia

CHANGES.md

@@ -8,6 +8,13 @@ Version 2.4.0
 
 To be released.
 
+### @logtape/logtape
+
+ -  Added the `StreamSinkOptions.closeStream` option to dispose stream sinks
+    without closing caller-owned streams. [[#203]]
+
+[#203]: https://github.com/dahlia/logtape/issues/203
+
 
 Version 2.3.1
 -------------

changes.d/logtape/stream-lifecycle.md

@@ -0,0 +1,2 @@
+ -  Added the `StreamSinkOptions.closeStream` option to dispose stream sinks
+    without closing caller-owned streams. [[#203]]

docs/manual/sinks.md

@@ -184,6 +184,26 @@ await configure({
 > Node.js stream.  You can use [`Writable.toWeb()`] method to convert a Node.js
 > stream to a `WritableStream`.
 
+By default, disposing a stream sink closes its `WritableStream`.  For a
+caller-owned stream that needs to remain open, such as a Node.js standard
+stream, set `closeStream` to `false`:
+
+~~~~ typescript twoslash
+// @noErrors: 2345
+import "@types/node";
+import { getStreamSink } from "@logtape/logtape";
+import stream from "node:stream";
+
+const stderrSink = getStreamSink(
+  stream.Writable.toWeb(process.stderr),
+  { closeStream: false },
+);
+~~~~
+
+Disposing the sink still waits for pending writes, flushes buffered records
+when non-blocking mode is enabled, and releases the writer lock without closing
+the stream.
+
 See also `getStreamSink()` function and `StreamSinkOptions` interface
 in the API reference for more details.
 

packages/logtape/src/sink.test.ts

@@ -109,13 +109,17 @@ interface ConsoleMock extends Console {
 
 test("getStreamSink()", async () => {
   let buffer: string = "";
+  let closed = false;
   const decoder = new TextDecoder();
   const sink = getStreamSink(
     new WritableStream({
       write(chunk: Uint8Array) {
         buffer += decoder.decode(chunk);
         return Promise.resolve();
       },
+      close() {
+        closed = true;
+      },
     }),
   );
   sink(trace);
@@ -136,6 +140,40 @@ test("getStreamSink()", async () => {
 2023-11-14 22:13:20.000 +00:00 [FTL] my-app·junk: Hello, 123 & 456!
 `,
   );
+  assert.strictEqual(closed, true);
+});
+
+test("getStreamSink() with closeStream: false", async () => {
+  // Arrange
+  let buffer = "";
+  let closed = false;
+  const decoder = new TextDecoder();
+  const stream = new WritableStream({
+    write(chunk: Uint8Array) {
+      buffer += decoder.decode(chunk);
+    },
+    close() {
+      closed = true;
+    },
+  });
+  const sink = getStreamSink(stream, { closeStream: false });
+
+  // Act
+  sink(info);
+  await sink[Symbol.asyncDispose]();
+
+  // Assert
+  assert.strictEqual(
+    buffer,
+    "2023-11-14 22:13:20.000 +00:00 [INF] my-app·junk: " +
+      "Hello, 123 & 456!\n",
+  );
+  assert.strictEqual(closed, false);
+  const writer = stream.getWriter();
+  await writer.write(new TextEncoder().encode("after disposal"));
+  await writer.close();
+  writer.releaseLock();
+  assert.ok(buffer.endsWith("after disposal"));
 });
 
 test("getStreamSink() with nonBlocking - simple boolean", async () => {
@@ -300,6 +338,90 @@ test("getStreamSink() with nonBlocking - flush on dispose", async () => {
   );
 });
 
+test(
+  "getStreamSink() with nonBlocking and closeStream: false",
+  async () => {
+    // Arrange
+    let buffer = "";
+    let closed = false;
+    const decoder = new TextDecoder();
+    const stream = new WritableStream({
+      write(chunk: Uint8Array) {
+        buffer += decoder.decode(chunk);
+      },
+      close() {
+        closed = true;
+      },
+    });
+    const sink = getStreamSink(stream, {
+      closeStream: false,
+      nonBlocking: {
+        bufferSize: 100,
+        flushInterval: 5000,
+      },
+    });
+
+    // Act
+    sink(info);
+    await sink[Symbol.asyncDispose]();
+
+    // Assert
+    assert.strictEqual(
+      buffer,
+      "2023-11-14 22:13:20.000 +00:00 [INF] my-app·junk: " +
+        "Hello, 123 & 456!\n",
+    );
+    assert.strictEqual(closed, false);
+    const writer = stream.getWriter();
+    await writer.close();
+    writer.releaseLock();
+  },
+);
+
+test(
+  "getStreamSink() with nonBlocking waits for an active flush on dispose",
+  async () => {
+    // Arrange
+    let markWriteStarted: () => void = () => {};
+    let finishWrite: () => void = () => {};
+    const writeStarted = new Promise<void>((resolve) => {
+      markWriteStarted = resolve;
+    });
+    const writeCanFinish = new Promise<void>((resolve) => {
+      finishWrite = resolve;
+    });
+    const stream = new WritableStream({
+      async write() {
+        markWriteStarted();
+        await writeCanFinish;
+      },
+    });
+    const sink = getStreamSink(stream, {
+      closeStream: false,
+      nonBlocking: { bufferSize: 1 },
+    });
+
+    // Act
+    sink(info);
+    await beforeDeadline(writeStarted, "The buffered write did not start");
+    let disposed = false;
+    const disposePromise = Promise.resolve(sink[Symbol.asyncDispose]()).then(
+      () => {
+        disposed = true;
+      },
+    );
+    await delay(0);
+
+    // Assert
+    assert.strictEqual(disposed, false);
+    finishWrite();
+    await beforeDeadline(disposePromise, "The stream sink was not disposed");
+    const writer = stream.getWriter();
+    await writer.close();
+    writer.releaseLock();
+  },
+);
+
 test("getStreamSink() with nonBlocking - buffer overflow protection", async () => {
   let buffer: string = "";
   const decoder = new TextDecoder();
@@ -3292,3 +3414,22 @@ function recordWithLevel(level: LogLevel): LogRecord {
     properties: {},
   };
 }
+
+async function beforeDeadline<T>(
+  promise: Promise<T>,
+  message: string,
+): Promise<T> {
+  let timeoutId: ReturnType<typeof globalThis.setTimeout> | undefined;
+  try {
+    return await Promise.race([
+      promise,
+      new Promise<never>((_, reject) => {
+        timeoutId = globalThis.setTimeout(() => {
+          reject(new Error(message));
+        }, 1000);
+      }),
+    ]);
+  } finally {
+    if (timeoutId !== undefined) globalThis.clearTimeout(timeoutId);
+  }
+}

packages/logtape/src/sink.ts

@@ -89,6 +89,16 @@ export interface StreamSinkOptions {
    */
   encoder?: { encode(text: string): Uint8Array };
 
+  /**
+   * Whether to close the stream when the sink is disposed.  Set this to
+   * `false` for caller-owned streams that need to remain open after disposal.
+   * The sink still waits for pending writes and releases its writer lock.
+   *
+   * @default `true`
+   * @since 2.4.0
+   */
+  closeStream?: boolean;
+
   /**
    * Enable non-blocking mode with optional buffer configuration.
    * When enabled, log records are buffered and flushed in the background.
@@ -156,6 +166,7 @@ export function getStreamSink(
 ): Sink & AsyncDisposable {
   const formatter = options.formatter ?? defaultTextFormatter;
   const encoder = options.encoder ?? new TextEncoder();
+  const closeStream = options.closeStream ?? true;
   const writer = stream.getWriter();
 
   if (!options.nonBlocking) {
@@ -167,8 +178,12 @@ export function getStreamSink(
         .then(() => writer.write(bytes));
     };
     sink[Symbol.asyncDispose] = async () => {
-      await lastPromise;
-      await writer.close();
+      try {
+        await lastPromise;
+        if (closeStream) await writer.close();
+      } finally {
+        writer.releaseLock();
+      }
     };
     return markSinkAsImmediate(sink);
   }
@@ -240,11 +255,18 @@ export function getStreamSink(
       clearInterval(flushTimer);
       flushTimer = null;
     }
-    await flush();
     try {
-      await writer.close();
-    } catch {
-      // Writer might already be closed or errored
+      await activeFlush;
+      await flush();
+      if (closeStream) {
+        try {
+          await writer.close();
+        } catch {
+          // Writer might already be closed or errored
+        }
+      }
+    } finally {
+      writer.releaseLock();
     }
   };