Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
23 changes: 20 additions & 3 deletions p2p/sender.go
Original file line number Diff line number Diff line change
Expand Up @@ -344,7 +344,17 @@ func SendReceive(ctx context.Context, p2pNode host.Host, peerID peer.ID,
}

if err := s.CloseWrite(); err != nil {
return errors.Wrap(err, "close write", z.Any("protocol", s.Protocol()))
// A canceled-stream error here is benign: the request was already written and
// delivered above, and the peer resetting our send-direction (STOP_SENDING) does
// not affect the receive-direction, so the response is still readable below.
// This happens much more frequently when QUIC is enabled.
//
//nolint:revive // Prioritise readability.
if isCanceledStreamErr(err) {
log.Debug(ctx, "Closing write of canceled stream", z.Err(err), z.Any("protocol", s.Protocol()))
} else {
return errors.Wrap(err, "close write", z.Any("protocol", s.Protocol()))
}
}

if err = reader.ReadMsg(resp); err != nil {
Expand All @@ -355,7 +365,7 @@ func SendReceive(ctx context.Context, p2pNode host.Host, peerID peer.ID,
// This close error doesn't require a retry or warning mechanism since the message was
// correctly sent to the destination. However, it is still unknown what/who is causing the
// stream to be canceled. This error appears much more frequently when QUIC is enabled.
if strings.Contains(err.Error(), "close called for canceled stream") {
if isCanceledStreamErr(err) {
log.Debug(ctx, "Closing canceled stream", z.Err(err), z.Any("protocol", s.Protocol()))
return nil
}
Expand Down Expand Up @@ -412,7 +422,7 @@ func Send(ctx context.Context, p2pNode host.Host, protoID protocol.ID, peerID pe
// This close error doesn't require a retry or warning mechanism since the message was
// correctly sent to the destination. However, it is still unknown what/who is causing the
// stream to be canceled. This error appears much more frequently when QUIC is enabled.
if strings.Contains(err.Error(), "close called for canceled stream") {
if isCanceledStreamErr(err) {
log.Debug(ctx, "Closing canceled stream", z.Err(err), z.Any("protocol", s.Protocol()))
return nil
}
Expand All @@ -423,6 +433,13 @@ func Send(ctx context.Context, p2pNode host.Host, protoID protocol.ID, peerID pe
return nil
}

// isCanceledStreamErr returns true if the error is the benign QUIC stream error
// returned when closing a stream whose send-direction the peer has already reset
// (via STOP_SENDING). It occurs much more frequently when QUIC is enabled.
func isCanceledStreamErr(err error) bool {
return err != nil && strings.Contains(err.Error(), "close called for canceled stream")
}

// protocolPrefix returns the common prefix of the provided protocol IDs.
func protocolPrefix(pIDs ...protocol.ID) protocol.ID {
if len(pIDs) == 0 {
Expand Down
7 changes: 7 additions & 0 deletions p2p/sender_internal_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -160,3 +160,10 @@ func TestNonZeroResponse(t *testing.T) {
err := SendReceive(ctx, nil, "", nil, &pbv1.Duty{Slot: 1}, "")
require.ErrorContains(t, err, "bug: response proto must be zero value")
}

func TestIsCanceledStreamErr(t *testing.T) {
require.False(t, isCanceledStreamErr(nil))
require.False(t, isCanceledStreamErr(errors.New("some other error")))
require.True(t, isCanceledStreamErr(errors.New("close called for canceled stream 4")))
require.True(t, isCanceledStreamErr(errors.Wrap(errors.New("close called for canceled stream 4"), "close write")))
}
Loading