fix: guard processSendQueue against non-writable stream status by lodekeeper · Pull Request #3416 · libp2p/js-libp2p

lodekeeper · 2026-03-21T16:24:48Z

What

Add a writeStatus guard at the top of processSendQueue() in abstract-message-stream.ts to bail out when the stream is no longer writable.

Why

A race condition causes an uncaught StreamStateError when a drain event fires on a stream whose underlying transport is already closing:

TCP connection starts closing → writeStatus transitions to 'closing'
A drain event fires on the underlying TCP stream
The noise layer relays this as safeDispatchEvent('drain') on the encrypted stream
continueSendingOnDrain → processSendQueue() → sendData() → send() on the now-closing stream
send() throws StreamStateError — inside an event listener with no try/catch → uncaught exception

Observed on Lodestar mainnet (v1.41.0, @libp2p/utils@7.0.13): 6 uncaught exceptions in 7 days.

Fix

Single guard at the top of processSendQueue(), consistent with the existing bail-out pattern:

if (this.writeStatus !== 'writable') {
  this.log.trace('not processing send queue as stream write status is %s', this.writeStatus)
  return false
}

Test

Added a regression test that:

Creates a stream pair with capacity: 1 to trigger backpressure
Fills the send buffer until send() returns false
Sets writeStatus to 'closing'
Dispatches a drain event
Asserts sendData is NOT called (the guard catches it) and no error is thrown

Closes #3415

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 libp2p#3415

…ests

dozyio · 2026-04-06T12:12:08Z

@lodekeeper - can you review this commit 17d9da8 please

lodekeeper · 2026-04-06T12:19:11Z

@dozyio Reviewed commit 17d9da8 — looks good overall.

abstract-message-stream.ts: The willSend defensive copy + comment makes the intent clear — guard against sendData mutating the slice before throwing. The rename is cleaner than the original toSend requeue. No issues here.

Tests: Both cases are solid:

preserve on error: verifies buffer is intact after a StreamStateError on drain ✓
no duplication on transient error: the throwOnce pattern correctly proves data isn't counted twice across two drain events ✓

One minor note: the second test sets writeStatus = 'closing' before the first drain but the continueSendingOnDrain guard only bails on statuses that are neither 'writable' nor 'closing'. So 'closing' still reaches the try/catch — which is the right flow for the test to exercise. Just worth a comment in the test body if you want it to be self-documenting, but it's not required.

Happy to address any remaining feedback before merge.

dozyio · 2026-04-22T04:19:24Z

@lodekeeper it seems test are failing after merging latest main branch - could you investigate?

The drain handler was changed in libp2p#3435 to wrap processSendQueue() in queueMicrotask() to avoid a Maximum call stack size exceeded error. That merge of main into this branch causes the StreamStateError drain tests added in 17d9da8 to fail: dispatchEvent(new Event('drain')) now returns before sendData is invoked, so synchronous assertions on stub call counts and writeBufferLength see stale state. Await the microtask queue between dispatching drain and asserting so the tests observe the post-drain state the scenarios were designed to verify. 🤖 Generated with AI assistance

lodekeeper · 2026-04-22T04:28:34Z

@dozyio Investigated — pushed cbc1f2e2d.

Root cause: the merge of main brought in #3435 (commit 5917715d0c), which wraps processSendQueue() in queueMicrotask(...) inside the drain handler to prevent a Maximum call stack size exceeded error. After that change, the two drain-race tests added in 17d9da8ab (should preserve queued bytes when sendData throws StreamStateError during drain and should not duplicate queued bytes after transient StreamStateError during drain) dispatch a synchronous drain event and then immediately assert on stub.calledOnce / writeBufferLength, but sendData is now deferred to a microtask — so the stub has been called 0 times when the assertion fires (expected false to be true at line 495).

Fix: await a microtask flush (await new Promise<void>((resolve) => { queueMicrotask(resolve) })) between dispatchEvent('drain') and the post-drain assertions. Preserves test intent (sendData is still invoked, bytes are still preserved / requeued); just adapts to the deferred scheduling. No source change needed.

tabcat

Thanks for the work @lodekeeper
I'm going to get this over the finish line for you. Feel free to leave comments but edits are not requested from you at this time.

tabcat · 2026-05-14T09:04:00Z

+            this.log('send failed during queue processing, stream is %s - %e', this.writeStatus, err)
+            // Requeue the defensive copy in case sendData mutated `toSend`
+            // before throwing.
+            this.writeBuffer.prepend(willSend)


i dont like this. imo the data should be dropped. a lot could happen inside of this.sendData before it throws

tabcat · 2026-05-14T09:04:36Z

+            // Requeue the defensive copy in case sendData mutated `toSend`
+            // before throwing.
+            this.writeBuffer.prepend(willSend)
+            return false


send should error if we were not able to send the data

tabcat · 2026-05-14T10:03:23Z

wrapping this with try/catch and logging the error will fix the original uncaught runtime error symptom on its own.

tabcat · 2026-05-14T10:18:50Z

      this.log.trace('not processing send queue as already sending data')
      return true
    }



the write guard should go here (line 460), log, return false.

tabcat · 2026-05-14T10:20:27Z

+        // 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
+        }
+


should live in processSendQueue

tabcat · 2026-05-14T10:24:45Z

-        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)
+            // Requeue the defensive copy in case sendData mutated `toSend`
+            // before throwing.
+            this.writeBuffer.prepend(willSend)
+            return false
+          }
+
+          throw err
+        }
+


should not be wrapped and return false on failure from sendData. can catch the uncaught runtime error by wrapping the call in continueSendingOnDrain

lodekeeper

Understood — thanks @tabcat, appreciate you taking it over. Happy to answer any questions about the original implementation if context would help. Will keep an eye on the thread.

Consolidate the guard alongside the existing bail-outs. Drop the inner sendData try/catch so send() throws consistently with the existing closed/closing check; the drain handler's microtask wrapper catches and logs.

tabcat · 2026-05-14T11:00:09Z

@lodekeeper care to review the current diff with main?

lodekeeper · 2026-05-14T11:05:04Z

Reviewed the current diff against main. The implementation matches the design notes from your inline comments:

L453–457 — writeStatus guard moved into processSendQueue (returns false and logs), addresses the should live in processSendQueue / write-guard-on-line-460 notes.
L100–110 — drain-handler microtask wrapped in try/catch with log.error, addresses wrapping this with try/catch and logging the error will fix the original uncaught runtime error symptom.
Inner sendData try/catch removed — send() now propagates StreamStateError consistently with the existing closed/closing check (matches send should error if we were not able to send the data + should not be wrapped).

Tests cover the three resulting paths:

drain on closed stream → no uncaught error
drain mid-closing while sendData throws StreamStateError → caught by microtask wrapper
direct send() when sendData throws → error propagates to caller

build and most Examples checks are green; test-node / browser / interop matrix still running, no failures so far. mergeable: MERGEABLE, status UNSTABLE only because CI is in progress. LGTM from my side — happy to address anything else if reviewers flag something.

lodekeeper requested a review from a team as a code owner March 21, 2026 16:24

lodekeeper changed the title ~~fix(@libp2p/utils): guard processSendQueue against non-writable stream status~~ fix(@libp2p/utils): guard processSendQueue on non-writable stream Mar 21, 2026

lodekeeper changed the title ~~fix(@libp2p/utils): guard processSendQueue on non-writable stream~~ fix: guard processSendQueue against non-writable stream status Mar 21, 2026

lodekeeper force-pushed the fix/stream-drain-race-guard branch 3 times, most recently from 9538027 to c1c6584 Compare March 21, 2026 17:06

lodekeeper force-pushed the fix/stream-drain-race-guard branch from c1c6584 to a8c2cf2 Compare March 21, 2026 17:06

lodekeeper and others added 4 commits March 21, 2026 17:35

ci: retrigger CI for flaky WebRTC and integration test failures

ef4dc8b

Merge branch 'main' into fix/stream-drain-race-guard

3a8e341

chore: use defensive copy instead of potentially modified data, few t…

17d9da8

…ests

Merge branch 'main' into fix/stream-drain-race-guard

14a8c27

dozyio approved these changes Apr 6, 2026

View reviewed changes

dozyio requested a review from tabcat April 6, 2026 12:14

tabcat self-assigned this Apr 10, 2026

Merge branch 'main' into fix/stream-drain-race-guard

cb7fcbd

tabcat reviewed May 14, 2026

View reviewed changes

lodekeeper commented May 14, 2026

View reviewed changes

tabcat added 3 commits May 14, 2026 17:47

refactor: move writeStatus guard into processSendQueue

1dfb825

Consolidate the guard alongside the existing bail-outs. Drop the inner sendData try/catch so send() throws consistently with the existing closed/closing check; the drain handler's microtask wrapper catches and logs.

chore: drop redundant guard comment

03bfe4f

chore: drop redundant test comments

bf0e8d7

tabcat merged commit 8230eb5 into libp2p:main May 14, 2026
34 checks passed

Conversation

lodekeeper commented Mar 21, 2026

What

Why

Fix

Test

Uh oh!

dozyio commented Apr 6, 2026

Uh oh!

lodekeeper commented Apr 6, 2026

Uh oh!

dozyio commented Apr 22, 2026

Uh oh!

lodekeeper commented Apr 22, 2026

Uh oh!

tabcat left a comment

Choose a reason for hiding this comment

Uh oh!

tabcat May 14, 2026

Choose a reason for hiding this comment

Uh oh!

tabcat May 14, 2026

Choose a reason for hiding this comment

Uh oh!

tabcat May 14, 2026

Choose a reason for hiding this comment

Uh oh!

tabcat May 14, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

tabcat May 14, 2026

Choose a reason for hiding this comment

Uh oh!

tabcat May 14, 2026

Choose a reason for hiding this comment

Uh oh!

lodekeeper left a comment

Choose a reason for hiding this comment

Uh oh!

tabcat commented May 14, 2026

Uh oh!

lodekeeper commented May 14, 2026

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

tabcat May 14, 2026 •

edited

Loading