Skip to content

Fix sends racing with disconnect and handshake cancellation #1374

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
NicolasDorier merged 1 commit into from
Nov 6, 2026
Merged

Fix sends racing with disconnect and handshake cancellation #1374

Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
37 changes: 23 additions & 14 deletions NBitcoin/Protocol/Node.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -185,7 +185,7 @@ public NodeConnection(Node node, Socket socket)
internal bool IsVerbose => Logs.NodeServer.IsEnabled(LogLevel.Trace);


internal BlockingCollection<SentMessage> Messages = new BlockingCollection<SentMessage>(new ConcurrentQueue<SentMessage>());
internal readonly BlockingCollection<SentMessage> Messages = new BlockingCollection<SentMessage>(new ConcurrentQueue<SentMessage>());
public void BeginListen()
{
new Thread(() =>
Expand Down Expand Up @@ -261,18 +261,19 @@ public void BeginListen()
ar?.Dispose();
}

// Close the queue before draining it so a concurrent send cannot be left without a consumer.
Messages.CompleteAdding();
if (processing != null)
Messages.Add(processing);
processing.Completion.TrySetException(new InvalidOperationException("The peer has been disconnected"));

foreach (var pending in Messages)
while (Messages.TryTake(out var pending))
{
if (IsVerbose)
{
Logs.NodeServer.LogTrace("The connection cancelled before the message was sent");
}
pending.Completion.SetException(new InvalidOperationException("The peer has been disconnected"));
pending.Completion.TrySetException(new InvalidOperationException("The peer has been disconnected"));
}
Messages = new BlockingCollection<SentMessage>(new ConcurrentQueue<SentMessage>());
Logs.NodeServer.LogInformation("Stop sending");
Cleanup(unhandledException);
}).Start();
Expand Down Expand Up @@ -947,12 +948,20 @@ public Task SendMessageAsync(Payload payload)

Action final = () =>
{
_Connection.Messages.Add(new SentMessage()
try
{
Payload = payload,
ActivityId = activity,
Completion = completion
});
_Connection.Messages.Add(new SentMessage()
{
Payload = payload,
ActivityId = activity,
Completion = completion
});
}
catch (InvalidOperationException)
{
// The sender may have stopped after the IsConnected check, including while a filter ran.
completion.TrySetException(new InvalidOperationException("The peer has been disconnected"));
}
};
OnSendingMessage(payload, final);
return completion.Task;
Expand Down Expand Up @@ -1104,7 +1113,7 @@ public VersionPayload PeerVersion
p.Message.Payload is VerAckPayload))
{

await SendMessageAsync(MyVersion).ConfigureAwait(false);
await SendMessageAsync(MyVersion).WithCancellation(cancellationToken).ConfigureAwait(false);
#if !NO_CHANNELS
var version = await listener.ReceivePayloadAsync<VersionPayload>(cancellationToken).ConfigureAwait(false);
#else
Expand Down Expand Up @@ -1135,10 +1144,10 @@ public VersionPayload PeerVersion
if (ProtocolCapabilities.SupportAddrv2)
{
// Signal ADDRv2 support (BIP155).
await SendMessageAsync(new SendAddrV2Payload()).ConfigureAwait(false);
await SendMessageAsync(new SendAddrV2Payload()).WithCancellation(cancellationToken).ConfigureAwait(false);
}

await SendMessageAsync(new VerAckPayload()).ConfigureAwait(false);
await SendMessageAsync(new VerAckPayload()).WithCancellation(cancellationToken).ConfigureAwait(false);

#if !NO_CHANNELS
await listener.ReceivePayloadAsync<VerAckPayload>(cancellationToken).ConfigureAwait(false);
Expand All @@ -1156,7 +1165,7 @@ public VersionPayload PeerVersion
await SendMessageAsync(new AddrPayload(new NetworkAddress(MyVersion.AddressFrom)
{
Time = DateTimeOffset.UtcNow
})).ConfigureAwait(false);
})).WithCancellation(cancellationToken).ConfigureAwait(false);
}
}
}
Expand Down
Loading