Refactor locks and more

Author: flcl42

src/libp2p/Libp2p.Core/Exceptions/Libp2pException.cs

@@ -1,6 +1,10 @@
 // SPDX-FileCopyrightText: 2024 Demerzel Solutions Limited
 // SPDX-License-Identifier: MIT
 
+
+using System.Diagnostics.CodeAnalysis;
+using System.Runtime.CompilerServices;
+
 namespace Nethermind.Libp2p.Core.Exceptions;
 
 public class Libp2pException : Exception
@@ -18,7 +22,20 @@ public class ChannelClosedException() : Libp2pException("Channel closed");
 /// Appears when libp2p is not set up properly in part of protocol tack, IoC, etc.
 /// </summary>
 /// <param name="message"></param>
-public class Libp2pSetupException(string? message = null) : Libp2pException(message);
+public class Libp2pSetupException(string? message = null) : Libp2pException(message)
+{
+    public static void ThrowIfNull([NotNull] object? argument, [CallerArgumentExpression(nameof(argument))] string? paramName = null)
+    {
+        if (argument is null)
+        {
+            Throw(paramName);
+        }
+    }
+
+    [DoesNotReturn]
+    internal static void Throw(string? paramName) =>
+            throw new Libp2pSetupException($"{paramName} is not set during libp2p initialization");
+}
 
 /// <summary>
 /// Appears when there is already active session for the given peer

src/libp2p/Libp2p.Core/MultiaddressExtensions.cs

@@ -1,12 +1,18 @@
-// SPDX-FileCopyrightText: 2023 Demerzel Solutions Limited
+// SPDX-FileCopyrightText: 2025 Demerzel Solutions Limited
 // SPDX-License-Identifier: MIT
 
 using Multiformats.Address;
+using Multiformats.Address.Net;
 using Multiformats.Address.Protocols;
+using System.Net.Sockets;
 
 namespace Nethermind.Libp2p.Core;
 
 public static class MultiaddressExtensions
 {
-    public static PeerId? GetPeerId(this Multiaddress? addr) => addr is not null && addr.Has<P2P>() ? new PeerId(addr.Get<P2P>().ToString()) : default;
+    public static PeerId? GetPeerId(this Multiaddress? addr)
+        => addr is not null && addr.Has<P2P>() ? new PeerId(addr.Get<P2P>().ToString()) : default;
+
+    public static Multiaddress GetEndpointPart(this Multiaddress multiaddress)
+        => multiaddress.ToEndPoint(out ProtocolType proto).ToMultiaddress(proto);
 }

src/libp2p/Libp2p.Protocols.Identify/IdentifyNotifier.cs

@@ -1,4 +1,4 @@
-// SPDX-FileCopyrightText: 2024 Demerzel Solutions Limited
+// SPDX-FileCopyrightText: 2025 Demerzel Solutions Limited
 // SPDX-License-Identifier: MIT
 
 using Nethermind.Libp2p.Core;

src/libp2p/Libp2p.Protocols.Identify/IdentifyProtocol.cs

@@ -22,7 +22,7 @@ public IdentifyProtocol(IProtocolStackSettings protocolStackSettings, IdentifyPr
     public async Task DialAsync(IChannel channel, ISessionContext context)
     {
         _logger?.LogInformation("Dial");
-        await ReadAndVerifyIndentity(channel, context);
+        await ReadAndVerifyIdentity(channel, context);
     }
 
     public async Task ListenAsync(IChannel channel, ISessionContext context)

src/libp2p/Libp2p.Protocols.Identify/IdentifyProtocolBase.cs

@@ -20,7 +20,7 @@ public abstract class IdentifyProtocolBase(IProtocolStackSettings protocolStackS
     private readonly IProtocolStackSettings _protocolStackSettings = protocolStackSettings;
     private readonly IdentifyProtocolSettings _settings = settings ?? new IdentifyProtocolSettings();
 
-    protected async Task ReadAndVerifyIndentity(IChannel channel, ISessionContext context)
+    protected async Task ReadAndVerifyIdentity(IChannel channel, ISessionContext context)
     {
         ArgumentNullException.ThrowIfNull(context.State.RemotePublicKey);
         ArgumentNullException.ThrowIfNull(context.State.RemotePeerId);
@@ -34,11 +34,28 @@ protected async Task ReadAndVerifyIndentity(IChannel channel, ISessionContext co
             throw new PeerConnectionException("Malformed peer identity: the remote public key corresponds to a different peer id");
         }
 
-        ulong seq = 0;
+        VerifySignedPeerRecordOrThrow(identify.SignedPeerRecord, context.State.RemotePublicKey, context.State.RemotePeerId, out ulong seq);
 
-        if (identify.SignedPeerRecord is not null)
+        if (_peerStore is not null)
+        {
+            PeerStore.PeerInfo peerInfo = _peerStore.GetPeerInfo(context.State.RemotePeerId);
+
+            if (identify.SignedPeerRecord is not null && peerInfo.Seq >= seq)
+            {
+                // do nothing if seq is for an older record
+                return;
+            }
+            peerInfo.SupportedProtocols = identify.Protocols.ToArray();
+            peerInfo.SignedPeerRecord = identify.SignedPeerRecord;
+            peerInfo.Seq = seq;
+        }
+    }
+
+    private void VerifySignedPeerRecordOrThrow(ByteString? signedPeerRecordBytes, PublicKey remotePublicKey, PeerId remotePeerId, out ulong seq)
+    {
+        if (signedPeerRecordBytes is not null)
         {
-            if (!SigningHelper.VerifyPeerRecord(identify.SignedPeerRecord, context.State.RemotePublicKey, out seq))
+            if (!SigningHelper.VerifyPeerRecord(signedPeerRecordBytes, remotePublicKey, out seq))
             {
                 if (_settings?.PeerRecordsVerificationPolicy == PeerRecordsVerificationPolicy.RequireCorrect)
                 {
@@ -51,43 +68,38 @@ protected async Task ReadAndVerifyIndentity(IChannel channel, ISessionContext co
             }
             else
             {
-                _logger?.LogDebug("Confirmed peer record: {peerId}", context.State.RemotePeerId);
+                _logger?.LogDebug("Confirmed peer record: {peerId}", remotePeerId);
             }
         }
         else if (_settings.PeerRecordsVerificationPolicy != PeerRecordsVerificationPolicy.DoesNotRequire)
         {
             throw new PeerConnectionException("Malformed peer identity: there is no peer record which is required");
         }
 
-        if (_peerStore is not null)
-        {
-            PeerStore.PeerInfo peerInfo = _peerStore.GetPeerInfo(context.State.RemotePeerId);
-
-            if (identify.SignedPeerRecord is not null && peerInfo.Seq >= seq)
-            {
-                // do nothing if seq is for an older record
-                return;
-            }
-            peerInfo.SupportedProtocols = identify.Protocols.ToArray();
-            peerInfo.SignedPeerRecord = identify.SignedPeerRecord;
-            peerInfo.Seq = seq;
-        }
+        seq = 0;
     }
 
     protected async Task SendIdentity(IChannel channel, ISessionContext context, ulong idVersion = 1)
     {
+        ArgumentNullException.ThrowIfNull(context.State.RemoteAddress);
+        Libp2pSetupException.ThrowIfNull(_protocolStackSettings.Protocols);
+
         Identify.Dto.Identify identify = new()
         {
             ProtocolVersion = _settings.ProtocolVersion,
             AgentVersion = _settings.AgentVersion,
             PublicKey = context.Peer.Identity.PublicKey.ToByteString(),
             ListenAddrs = { },
-            ObservedAddr = ByteString.CopyFrom(context.State.RemoteAddress!.ToEndPoint(out ProtocolType proto).ToMultiaddress(proto).ToBytes()),
-            Protocols = { _protocolStackSettings.Protocols!.Select(r => r.Key.Protocol).OfType<ISessionListenerProtocol>().Select(p => p.Id) },
+            ObservedAddr = ByteString.CopyFrom(context.State.RemoteAddress.GetEndpointPart().ToBytes()),
+            Protocols = { _protocolStackSettings.Protocols.Select(r => r.Key.Protocol).OfType<ISessionListenerProtocol>().Select(p => p.Id) },
             SignedPeerRecord = SigningHelper.CreateSignedEnvelope(context.Peer.Identity, [.. context.Peer.ListenAddresses], idVersion),
         };
 
-        ByteString[] endpoints = context.Peer.ListenAddresses.Where(a => !a.ToEndPoint().Address.IsPrivate()).Select(a => a.ToEndPoint(out ProtocolType proto).ToMultiaddress(proto)).Select(a => ByteString.CopyFrom(a.ToBytes())).ToArray();
+        ByteString[] endpoints = context.Peer.ListenAddresses
+            .Where(a => !a.ToEndPoint().Address.IsPrivate())
+            .Select(a => a.ToEndPoint(out ProtocolType proto).ToMultiaddress(proto))
+            .Select(a => ByteString.CopyFrom(a.ToBytes())).ToArray();
+
         identify.ListenAddrs.AddRange(endpoints);
 
         await channel.WriteSizeAndProtobufAsync(identify);

src/libp2p/Libp2p.Protocols.Identify/IdentifyPushProtocol.cs

@@ -1,4 +1,4 @@
-// SPDX-FileCopyrightText: 2023 Demerzel Solutions Limited
+// SPDX-FileCopyrightText: 2025 Demerzel Solutions Limited
 // SPDX-License-Identifier: MIT
 
 using Microsoft.Extensions.Logging;
@@ -25,6 +25,6 @@ public async Task<ulong> DialAsync(IChannel channel, ISessionContext context, ul
     public async Task ListenAsync(IChannel channel, ISessionContext context)
     {
         _logger?.LogDebug("Receiving identity update");
-        await ReadAndVerifyIndentity(channel, context);
+        await ReadAndVerifyIdentity(channel, context);
     }
 }

src/libp2p/Libp2p.Protocols.IpTcp/IpTcpProtocol.cs

@@ -190,8 +190,10 @@ public async Task DialAsync(ITransportContext context, Multiaddress remoteAddr,
                 _ = upChannel.CloseAsync();
                 _logger?.LogDebug("Ctx{0}: end sending, socket exception {1}", connectionCtx.Id, e.Message);
             }
-
-            client.Close();
+            finally
+            {
+                client.Close();
+            }
         });
 
         await Task.WhenAll(receiveTask, sendTask).ContinueWith(t => connectionCtx.Dispose());

src/libp2p/Libp2p.Protocols.MDns/MDnsDiscoveryProtocol.cs

@@ -64,7 +64,7 @@ public Task StartDiscoveryAsync(IReadOnlyList<Multiaddress> localPeerAddrs, Canc
                 }
             }
 
-            _logger?.LogTrace("Records {0}", string.Join(",", service.Resources));
+            _logger?.LogTrace("DNS records to share: {0}", string.Join(",", service.Resources));
             _logger?.LogInformation("Started as {0} {1}", PeerName, ServiceNameOverride ?? ServiceName);
 
             sd.ServiceDiscovered += (s, serviceName) =>

src/libp2p/Libp2p.Protocols.Pubsub/PubsubRouter.cs

@@ -492,7 +492,7 @@ internal enum ConnectionInitiation
     Remote,
 }
 
-readonly struct MessageWithId(MessageId id, Message message)
+internal readonly struct MessageWithId(MessageId id, Message message)
 {
     public MessageId Id { get; } = id;
     public Message Message { get; } = message;

src/libp2p/Libp2p.Protocols.Yamux.Tests/DataWindowTests.cs

@@ -9,21 +9,32 @@ public class DataWindowTests
     [Test]
     public async Task Test_WindowExtendsProperly_WhenChannelWaitsForUpdate()
     {
-        int windowSize = 100;
-        var w = new DataWindow(windowSize);
-        Task spendingTask = Task.Run(async () =>
+        const int WindowSize = 100;
+
+        for (int bytesToSendCase = 0; bytesToSendCase < 500; bytesToSendCase++)
         {
-            int bytesToSend = windowSize + 1;
-            while (bytesToSend != 0)
+            int bytesToSend = bytesToSendCase;
+            int windowUdatesNeeded = (bytesToSend - 1) / WindowSize;
+            int finalAvailable = WindowSize - bytesToSend + WindowSize * windowUdatesNeeded;
+
+            var w = new RemoteDataWindow(WindowSize);
+            Task spendingTask = Task.Run(async () =>
+            {
+                await Task.Delay(bytesToSend % 3);
+                while (bytesToSend != 0)
+                {
+                    bytesToSend -= await w.SpendOrWait(bytesToSend);
+                }
+            });
+
+            await Task.Delay(bytesToSend % 5);
+            for (int i = 0; i < windowUdatesNeeded; i++)
             {
-                bytesToSend -= await w.SpendWindowOrWait(bytesToSend);
+                int val = w.Extend(WindowSize);
             }
-        });
-        await Task.Delay(10);
-        int val = w.ExtendWindowIfNeeded();
 
-        await spendingTask;
-        Assert.That(val, Is.EqualTo(windowSize));
-        Assert.That(w.Available, Is.EqualTo(windowSize - 1));
+            await spendingTask;
+            Assert.That(w.Available, Is.EqualTo(finalAvailable));
+        }
     }
 }