fix: guard processSendQueue against non-writable stream status

Add a try-catch around sendData() in processSendQueue() to catch
StreamStateError when the underlying transport closes between a drain
event and the send attempt. This prevents the error from propagating
as an uncaught exception.

Also add a writeStatus guard in continueSendingOnDrain to skip
processing when the stream has fully closed.

Fixes #3415

Author: lodekeeper

TASK.md

@@ -0,0 +1,46 @@
+# Task: Fix uncaught StreamStateError on drain event during stream close
+
+## Issue
+https://github.com/libp2p/js-libp2p/issues/3415
+
+## Problem
+A race condition in `packages/utils/src/abstract-message-stream.ts` causes an uncaught exception when a `drain` event fires on a stream whose underlying transport is already closing.
+
+The `continueSendingOnDrain` listener (line ~97) calls `processSendQueue()` when `writableNeedsDrain` is true. But `processSendQueue()` (line ~428) has no guard for `writeStatus` — it proceeds to call `sendData()` which calls `send()` on the underlying stream. `send()` at line ~170 throws `StreamStateError` because `writeStatus === 'closing'`. The throw happens in an event listener context → uncaught exception.
+
+## Fix Required
+
+### 1. Source fix in `packages/utils/src/abstract-message-stream.ts`
+
+Add a `writeStatus` guard at the top of `processSendQueue()`, consistent with the existing bail-out pattern:
+
+```typescript
+protected processSendQueue (): boolean {
+    // NEW: bail if the stream is no longer writable - a drain event can
+    // arrive after writeStatus transitions to 'closing'/'closed' during
+    // connection teardown
+    if (this.writeStatus !== 'writable') {
+      this.log.trace('not processing send queue as stream write status is %s', this.writeStatus)
+      return false
+    }
+
+    // ... existing bail-out checks (writableNeedsDrain, empty buffer, sendingData) unchanged
+```
+
+### 2. Test in `packages/utils/test/stream-utils-test.spec.ts`
+
+Add a test that verifies `processSendQueue` does NOT throw when called on a closing stream. The test should:
+
+1. Create a stream pair using `streamPair()` from `../src/stream-pair.ts`
+2. Send data to fill the buffer (use `outgoing.send(data)` until it returns `false`, indicating drain is needed)
+3. Start closing the stream (call `outgoing.close()` without await — or set writeStatus to closing)
+4. Dispatch a drain event on the stream: `outgoing.dispatchEvent(new Event('drain'))`
+5. Verify no error is thrown (the drain should be silently ignored)
+
+Look at the existing test patterns in the file for style guidance. The test should go in the `describe('messageStreamToDuplex', ...)` block or a new `describe('processSendQueue', ...)` block.
+
+## Constraints
+- Only modify files in `packages/utils/`
+- Keep the fix minimal — one guard check
+- Follow existing code style (2-space indent, single quotes, etc.)
+- The test must actually exercise the race condition path (drain on closing stream)

packages/utils/src/abstract-message-stream.ts

@@ -96,6 +96,15 @@ export abstract class AbstractMessageStream<Timeline extends MessageStreamTimeli
 
     const continueSendingOnDrain = (): void => {
       if (this.writableNeedsDrain) {
+        // a drain event can arrive after writeStatus transitions to
+        // 'closing'/'closed' during connection teardown - bail out to
+        // avoid calling send() on a non-writable stream which would
+        // throw an uncaught StreamStateError
+        if (this.writeStatus !== 'writable' && this.writeStatus !== 'closing') {
+          this.log.trace('not processing send queue on drain as stream write status is %s', this.writeStatus)
+          return
+        }
+
         this.log.trace('drain event received, continue sending data')
         this.writableNeedsDrain = false
         this.processSendQueue()
@@ -476,7 +485,22 @@ export abstract class AbstractMessageStream<Timeline extends MessageStreamTimeli
 
         // sending data can cause buffers to fill up, events to be emitted and
         // this method to be invoked again
-        const sendResult = this.sendData(toSend)
+        let sendResult: SendResult
+        try {
+          sendResult = this.sendData(toSend)
+        } catch (err: any) {
+          // the underlying transport may have closed between the drain event
+          // and this send attempt - treat as a failed send rather than letting
+          // the error propagate as an uncaught exception
+          if (err.name === 'StreamStateError') {
+            this.log('send failed during queue processing, stream is %s - %e', this.writeStatus, err)
+            this.writeBuffer.prepend(toSend)
+            return false
+          }
+
+          throw err
+        }
+
         canSendMore = sendResult.canSendMore
         sentBytes += sendResult.sentBytes
 

packages/utils/test/stream-utils-test.spec.ts

@@ -412,4 +412,58 @@ describe('stream-pair', () => {
     await expect(outgoing.onDrain()).to.eventually.be.rejected
       .with.property('name', 'StreamResetError')
   })
+
+  it('should not throw uncaught error on drain event when stream is closed', async () => {
+    const [outgoing] = await streamPair({
+      capacity: 1,
+      delay: 100
+    })
+
+    // fill the write buffer to trigger backpressure
+    while (outgoing.send(Uint8Array.from([0, 1, 2, 3]))) {
+      // drain the send buffer
+    }
+    expect(outgoing.send(Uint8Array.from([4, 5, 6, 7]))).to.be.false()
+
+    // set the stream to closed - drain events after this should be ignored
+    outgoing.writeStatus = 'closed'
+
+    // dispatching drain on a closed stream should not throw
+    expect(() => {
+      outgoing.dispatchEvent(new Event('drain'))
+    }).to.not.throw()
+
+    outgoing.abort(new Error('cleanup'))
+  })
+
+  it('should catch StreamStateError from sendData during closing', async () => {
+    const [outgoing] = await streamPair({
+      capacity: 1,
+      delay: 100
+    })
+    const outgoingStream = outgoing as typeof outgoing & {
+      sendData(data: Uint8ArrayList): { sentBytes: number, canSendMore: boolean }
+    }
+
+    // fill the write buffer to trigger backpressure
+    while (outgoing.send(Uint8Array.from([0, 1, 2, 3]))) {
+      // drain the send buffer
+    }
+    expect(outgoing.send(Uint8Array.from([4, 5, 6, 7]))).to.be.false()
+
+    // stub sendData to throw StreamStateError (simulates underlying
+    // transport closing while the outer stream is still flushing)
+    const err = new Error('Cannot write to a stream that is closing')
+    err.name = 'StreamStateError'
+    Sinon.stub(outgoingStream, 'sendData').throws(err)
+
+    outgoing.writeStatus = 'closing'
+
+    // the drain event should not throw - the error should be caught internally
+    expect(() => {
+      outgoing.dispatchEvent(new Event('drain'))
+    }).to.not.throw()
+
+    outgoing.abort(new Error('cleanup'))
+  })
 })