diff --git a/src/libp2p/Libp2p.Protocols.Pubsub.Tests/DirectPeersTests.cs b/src/libp2p/Libp2p.Protocols.Pubsub.Tests/DirectPeersTests.cs new file mode 100644 index 00000000..691b01e2 --- /dev/null +++ b/src/libp2p/Libp2p.Protocols.Pubsub.Tests/DirectPeersTests.cs @@ -0,0 +1,322 @@ +// SPDX-FileCopyrightText: 2026 Demerzel Solutions Limited +// SPDX-License-Identifier: MIT + +using Multiformats.Address; +using Nethermind.Libp2p.Core.Discovery; +using Nethermind.Libp2p.Protocols; +using Nethermind.Libp2p.Protocols.Pubsub.Dto; +using System.Collections.ObjectModel; + +namespace Nethermind.Libp2p.Protocols.Pubsub.Tests; + +[TestFixture] +public class DirectPeersTests +{ + [Test] + public void DirectPeers_ForwardValidMessagesDespitePeerScores() + { + const string topic = "topic"; + Multiaddress senderAddress = TestPeers.Multiaddr(1); + Multiaddress receiverAddress = TestPeers.Multiaddr(2); + PeerId senderPeerId = senderAddress.GetPeerId()!; + PeerId receiverPeerId = receiverAddress.GetPeerId()!; + PubsubRouter router = new( + new PeerStore(), + new PubsubSettings { DirectPeers = [senderAddress, receiverAddress] }); + ITopic localTopic = router.GetTopic(topic); + List senderRpcs = []; + List receiverRpcs = []; + TaskCompletionSource senderConnection = new(); + TaskCompletionSource receiverConnection = new(); + router.OutboundConnection(senderAddress, PubsubRouter.GossipsubProtocolVersionV11, senderConnection.Task, senderRpcs.Add); + router.OutboundConnection(receiverAddress, PubsubRouter.GossipsubProtocolVersionV11, receiverConnection.Task, receiverRpcs.Add); + router.OnRpc(senderPeerId, new Rpc().WithTopics([topic], [])); + router.OnRpc(receiverPeerId, new Rpc().WithTopics([topic], [])); + + router.SetAppSpecificScore(senderPeerId, -20); + router.SetAppSpecificScore(receiverPeerId, -20); + senderRpcs.Clear(); + receiverRpcs.Clear(); + + PeerId? receivedFrom = null; + localTopic.OnMessage += (peerId, _) => receivedFrom = peerId; + Identity author = TestPeers.Identity(1); + router.OnRpc(senderPeerId, new Rpc().WithMessages(topic, 1, author.PeerId.Bytes, [1, 2, 3], author)); + + Assert.Multiple(() => + { + Assert.That(receivedFrom, Is.EqualTo(senderPeerId)); + Assert.That(receiverRpcs.Single().Publish.Single().Data.ToByteArray(), Is.EqualTo(new byte[] { 1, 2, 3 })); + Assert.That(senderRpcs, Is.Empty); + }); + + senderConnection.SetResult(); + receiverConnection.SetResult(); + } + + [Test] + public void DirectPeers_AreNeverAddedToTheMesh() + { + const string topic = "topic"; + Multiaddress directAddress = TestPeers.Multiaddr(1); + Multiaddress firstMeshAddress = TestPeers.Multiaddr(2); + Multiaddress secondMeshAddress = TestPeers.Multiaddr(3); + PubsubRouter router = new( + new PeerStore(), + new PubsubSettings { DirectPeers = [directAddress] }); + IRoutingStateContainer state = router; + _ = router.GetTopic(topic); + TaskCompletionSource connection = new(); + + foreach (Multiaddress address in new[] { directAddress, firstMeshAddress, secondMeshAddress }) + { + router.OutboundConnection(address, PubsubRouter.GossipsubProtocolVersionV11, connection.Task, _ => { }); + router.OnRpc(address.GetPeerId()!, new Rpc().WithTopics([topic], [])); + } + + router.Heartbeat().GetAwaiter().GetResult(); + + Assert.Multiple(() => + { + Assert.That(state.GossipsubPeers[topic], Has.Member(directAddress.GetPeerId())); + Assert.That(state.Mesh[topic], Does.Not.Contain(directAddress.GetPeerId())); + Assert.That(state.Mesh[topic], Has.Count.EqualTo(2)); + }); + + connection.SetResult(); + } + + [Test] + public void DirectPeerGrafts_AreRejectedWithPrune() + { + const string topic = "topic"; + Multiaddress directAddress = TestPeers.Multiaddr(1); + PeerId directPeerId = directAddress.GetPeerId()!; + PubsubRouter router = new( + new PeerStore(), + new PubsubSettings { DirectPeers = [directAddress], PruneBackoff = 2_000 }); + _ = router.GetTopic(topic); + List sentRpcs = []; + TaskCompletionSource connection = new(); + router.OutboundConnection(directAddress, PubsubRouter.GossipsubProtocolVersionV11, connection.Task, sentRpcs.Add); + sentRpcs.Clear(); + + Rpc graft = new() { Control = new ControlMessage() }; + graft.Control.Graft.Add(new ControlGraft { TopicID = topic }); + router.OnRpc(directPeerId, graft); + + ControlPrune prune = sentRpcs.Single().Control.Prune.Single(); + Assert.Multiple(() => + { + Assert.That(prune.TopicID, Is.EqualTo(topic)); + Assert.That(prune.Backoff, Is.EqualTo(2)); + }); + connection.SetResult(); + } + + [TestCase(false)] + [TestCase(true)] + public async Task Router_ConnectsConfiguredDirectPeersAtStartup(bool addressesAlreadyKnown) + { + PeerStore peerStore = new(); + Multiaddress directAddress = TestPeers.Multiaddr(1); + PeerId directPeerId = directAddress.GetPeerId()!; + peerStore.GetPeerInfo(directPeerId).SupportedProtocols = [PubsubRouter.GossipsubProtocolVersionV12]; + if (addressesAlreadyKnown) + { + peerStore.Discover([directAddress]); + } + PubsubRouter router = new(peerStore, new PubsubSettings { DirectPeers = [directAddress] }); + + TaskCompletionSource protocolDialed = new(TaskCreationOptions.RunContinuationsAsynchronously); + ISession session = Substitute.For(); + session.RemoteAddress.Returns(directAddress); + session.DialAsync(Arg.Any()).Returns(_ => + { + protocolDialed.TrySetResult(); + return Task.CompletedTask; + }); + + ILocalPeer localPeer = Substitute.For(); + localPeer.Identity.Returns(TestPeers.Identity(2)); + localPeer.ListenAddresses.Returns(new ObservableCollection()); + localPeer.DialAsync(Arg.Any(), Arg.Any()).Returns(session); + + using CancellationTokenSource cancellation = new(); + await router.StartAsync(localPeer, cancellation.Token); + + await protocolDialed.Task.WaitAsync(TimeSpan.FromSeconds(2)); + _ = localPeer.Received(1).DialAsync(Arg.Any(), Arg.Any()); + _ = session.Received(1).DialAsync(Arg.Any()); + + cancellation.Cancel(); + } + + [Test] + public async Task PublishPartial_ExcludesDirectPeersFromFanoutAndMesh() + { + const string topic = "topic"; + Multiaddress directAddress = TestPeers.Multiaddr(1); + Multiaddress otherAddress = TestPeers.Multiaddr(2); + using PubsubRouter router = new(new PeerStore(), new PubsubSettings + { + DirectPeers = [directAddress], + EnablePartialMessages = true, + }); + IRoutingStateContainer state = router; + router.GetPartialMessagesTopic(topic, + new PartialMessagesTopicOptions { SupportsSendingPartialMessages = true }, subscribe: false); + ILocalPeer localPeer = Substitute.For(); + localPeer.Identity.Returns(TestPeers.Identity(3)); + localPeer.ListenAddresses.Returns(new ObservableCollection()); + using CancellationTokenSource cancellation = new(); + TaskCompletionSource connection = new(); + try + { + await router.StartAsync(localPeer, cancellation.Token); + foreach (Multiaddress address in new[] { directAddress, otherAddress }) + { + router.OutboundConnection(address, PubsubRouter.GossipsubProtocolVersionV13, connection.Task, _ => { }); + router.OnRpc(address.GetPeerId()!, new Rpc().WithTopics([topic], [])); + } + + router.PublishPartial(topic, [1], partialMessage: [2]); + Assert.That(state.Fanout[topic], Is.EquivalentTo(new[] { otherAddress.GetPeerId() })); + + router.Subscribe(topic); + Assert.That(state.Mesh[topic], Is.EquivalentTo(new[] { otherAddress.GetPeerId() })); + } + finally + { + cancellation.Cancel(); + connection.TrySetResult(); + } + } + + [Test] + public void Subscribe_ExcludesDirectPeersWhenMovingFanoutToMesh() + { + const string topic = "topic"; + Multiaddress directAddress = TestPeers.Multiaddr(1); + Multiaddress otherAddress = TestPeers.Multiaddr(2); + PeerId otherPeerId = otherAddress.GetPeerId()!; + using PubsubRouter router = new(new PeerStore(), new PubsubSettings { DirectPeers = [directAddress] }); + IRoutingStateContainer state = router; + TaskCompletionSource connection = new(); + try + { + router.OutboundConnection(otherAddress, PubsubRouter.GossipsubProtocolVersionV12, connection.Task, _ => { }); + router.OnRpc(otherPeerId, new Rpc().WithTopics([topic], [])); + state.Fanout[topic] = [directAddress.GetPeerId()!, otherPeerId]; + + router.Subscribe(topic); + + Assert.That(state.Mesh[topic], Is.EquivalentTo(new[] { otherPeerId })); + Assert.That(state.Fanout.ContainsKey(topic), Is.False); + } + finally + { + connection.SetResult(); + } + } + + [Test] + public async Task Router_ReconnectsDisconnectedDirectPeersIndependentlyOfOtherIntervals() + { + Multiaddress directAddress = TestPeers.Multiaddr(1); + PeerStore peerStore = new(); + peerStore.GetPeerInfo(directAddress.GetPeerId()!).SupportedProtocols = [PubsubRouter.GossipsubProtocolVersionV12]; + using PubsubRouter router = new(peerStore, new PubsubSettings + { + DirectPeers = [directAddress], + DirectConnectPeriod = 50, + ReconnectionPeriod = 60_000, + HeartbeatInterval = 60_000, + }); + TaskCompletionSource firstConnection = new(); + TaskCompletionSource secondConnection = new(); + TaskCompletionSource redialed = new(TaskCreationOptions.RunContinuationsAsynchronously); + int protocolDials = 0; + ISession session = Substitute.For(); + session.RemoteAddress.Returns(directAddress); + session.DialAsync(Arg.Any()).Returns(_ => + { + int attempt = Interlocked.Increment(ref protocolDials); + router.OutboundConnection(directAddress, PubsubRouter.GossipsubProtocolVersionV12, + attempt == 1 ? firstConnection.Task : secondConnection.Task, _ => { }); + if (attempt > 1) + { + redialed.TrySetResult(); + } + return Task.CompletedTask; + }); + ILocalPeer localPeer = Substitute.For(); + localPeer.Identity.Returns(TestPeers.Identity(2)); + localPeer.ListenAddresses.Returns(new ObservableCollection()); + localPeer.DialAsync(Arg.Any(), Arg.Any()).Returns(session); + using CancellationTokenSource cancellation = new(); + try + { + await router.StartAsync(localPeer, cancellation.Token); + await Task.Delay(200); + _ = localPeer.Received(1).DialAsync(Arg.Any(), Arg.Any()); + + firstConnection.SetResult(); + await redialed.Task.WaitAsync(TimeSpan.FromSeconds(3)); + + await Task.Delay(200); + _ = localPeer.Received(2).DialAsync(Arg.Any(), Arg.Any()); + _ = session.Received(2).DialAsync(Arg.Any()); + } + finally + { + cancellation.Cancel(); + firstConnection.TrySetResult(); + secondConnection.TrySetResult(); + } + } + + [Test] + public async Task Router_DoesNotRepeatInFlightDirectPeerDials() + { + Multiaddress directAddress = TestPeers.Multiaddr(1); + PeerStore peerStore = new(); + peerStore.Discover([directAddress]); + await using PubsubRouter router = new(peerStore, new PubsubSettings + { + DirectPeers = [directAddress], + DirectConnectPeriod = 20, + ReconnectionPeriod = 60_000, + HeartbeatInterval = 60_000, + }); + TaskCompletionSource slowDial = new(TaskCreationOptions.RunContinuationsAsynchronously); + ILocalPeer localPeer = Substitute.For(); + localPeer.Identity.Returns(TestPeers.Identity(2)); + localPeer.ListenAddresses.Returns(new ObservableCollection()); + localPeer.DialAsync(Arg.Any(), Arg.Any()).Returns(slowDial.Task); + + try + { + await router.StartAsync(localPeer); + await Task.Delay(200); + + _ = localPeer.Received(1).DialAsync(Arg.Any(), Arg.Any()); + } + finally + { + slowDial.TrySetCanceled(); + } + } + + [Test] + public void DirectPeers_RequirePeerIdsInTheirAddresses() + { + PubsubSettings settings = new() + { + DirectPeers = new[] { Multiaddress.Decode("/ip4/127.0.0.1/tcp/4001") }, + }; + + ArgumentException exception = Assert.Throws(() => new PubsubRouter(new PeerStore(), settings))!; + Assert.That(exception.ParamName, Is.EqualTo(nameof(PubsubSettings.DirectPeers))); + } +} diff --git a/src/libp2p/Libp2p.Protocols.Pubsub.Tests/PubsubProtocolTests.cs b/src/libp2p/Libp2p.Protocols.Pubsub.Tests/PubsubProtocolTests.cs index e534377d..652f7d95 100644 --- a/src/libp2p/Libp2p.Protocols.Pubsub.Tests/PubsubProtocolTests.cs +++ b/src/libp2p/Libp2p.Protocols.Pubsub.Tests/PubsubProtocolTests.cs @@ -39,6 +39,18 @@ public void Publish_WithNullMessage_ThrowsArgumentNullException() Assert.Throws(() => router.Publish("test-topic", null!)); } + [Test] + public async Task Publish_WithoutSubscription_DoesNotThrow() + { + PubsubRouter router = new(new PeerStore()); + using CancellationTokenSource cancellation = new(); + await router.StartAsync(new LocalPeerStub(), cancellation.Token); + + Assert.DoesNotThrow(() => router.Publish("test-topic", [1, 2, 3])); + + cancellation.Cancel(); + } + [Test] public void Topic_OnMessage_IncludesReceivedFromPeerId() { diff --git a/src/libp2p/Libp2p.Protocols.Pubsub/PubSubSettings.cs b/src/libp2p/Libp2p.Protocols.Pubsub/PubSubSettings.cs index 67e2df1e..3a486a6d 100644 --- a/src/libp2p/Libp2p.Protocols.Pubsub/PubSubSettings.cs +++ b/src/libp2p/Libp2p.Protocols.Pubsub/PubSubSettings.cs @@ -3,6 +3,7 @@ using Nethermind.Libp2p.Core; using Nethermind.Libp2p.Protocols.Pubsub.Dto; +using Multiformats.Address; namespace Nethermind.Libp2p.Protocols.Pubsub; @@ -22,6 +23,18 @@ public class PubsubSettings public int MaxConnections { get; set; } + /// + /// Peers with reciprocal explicit peering agreements. Each address must + /// contain a peer ID and is configured before the router starts. + /// + public Multiaddress[] DirectPeers { get; set; } = []; + + /// + /// Interval in milliseconds for reconnecting disconnected direct peers. Gossipsub recommends + /// five minutes. + /// + public int DirectConnectPeriod { get; set; } = 5 * 60 * 1000; + public int HeartbeatInterval { get; set; } = 1_000; // Time between heartbeats 1 second public int FanoutTtl { get; set; } = 60 * 1000; // Time-to-live for each topic's fanout state 60 seconds public int mcache_len { get; set; } = 5; // Number of history windows in message cache 5 diff --git a/src/libp2p/Libp2p.Protocols.Pubsub/PubsubRouter.Rpc.cs b/src/libp2p/Libp2p.Protocols.Pubsub/PubsubRouter.Rpc.cs index 06972e63..5291cab8 100644 --- a/src/libp2p/Libp2p.Protocols.Pubsub/PubsubRouter.Rpc.cs +++ b/src/libp2p/Libp2p.Protocols.Pubsub/PubsubRouter.Rpc.cs @@ -168,7 +168,7 @@ private void HandlePartialMessage(PeerId peerId, PartialMessagesExtension partia private void HandleNewMessages(PeerId peerId, IEnumerable messages, ConcurrentDictionary peerMessages, List<(string Topic, PeerId PeerId, byte[] Data)> receivedMessages) { // Check if peer is graylisted (Gossipsub v1.1) - if (ShouldGraylistPeer(peerId)) + if (!IsDirectPeer(peerId) && ShouldGraylistPeer(peerId)) { logger?.LogDebug("Ignoring messages from graylisted peer {peerId}", peerId); return; @@ -223,11 +223,19 @@ private void HandleNewMessages(PeerId peerId, IEnumerable messages, Con PeerId author = new(message.From.ToArray()); receivedMessages.Add((message.Topic, peerId, message.Data.ToByteArray())); + foreach (PeerId directPeerId in GetDirectPeersForTopic(message.Topic)) + { + if (directPeerId != author && directPeerId != peerId && ShouldSendFullMessage(directPeerId, message.Topic)) + { + peerMessages.GetOrAdd(directPeerId, _ => new Rpc()).Publish.Add(message); + } + } + if (fPeers.TryGetValue(message.Topic, out HashSet? topicPeers)) { foreach (PeerId peer in topicPeers) { - if (peer == author || peer == peerId) + if (peer == author || peer == peerId || IsDirectPeer(peer)) { continue; } @@ -241,7 +249,7 @@ private void HandleNewMessages(PeerId peerId, IEnumerable messages, Con { foreach (PeerId peer in topicPeers) { - if (peer == author || peer == peerId) + if (peer == author || peer == peerId || IsDirectPeer(peer)) { continue; } @@ -329,6 +337,15 @@ private void HandleGraft(PeerId peerId, IEnumerable grafts, Concur { foreach (ControlGraft? graft in grafts) { + if (IsDirectPeer(peerId)) + { + logger?.LogWarning("Rejecting GRAFT from direct peer {peerId} for topic {topic}", peerId, graft.TopicID); + peerMessages.GetOrAdd(peerId, _ => new Rpc()) + .Ensure(r => r.Control.Prune) + .Add(new ControlPrune { TopicID = graft.TopicID, Backoff = (ulong)Math.Max(1, _settings.PruneBackoff / 1_000) }); + continue; + } + if (topicState.GetValueOrDefault(graft.TopicID)?.IsSubscribed is not true || !mesh.TryGetValue(graft.TopicID, out HashSet? topicMesh)) { diff --git a/src/libp2p/Libp2p.Protocols.Pubsub/PubsubRouter.Topics.cs b/src/libp2p/Libp2p.Protocols.Pubsub/PubsubRouter.Topics.cs index c00eb054..15e1e8da 100644 --- a/src/libp2p/Libp2p.Protocols.Pubsub/PubsubRouter.Topics.cs +++ b/src/libp2p/Libp2p.Protocols.Pubsub/PubsubRouter.Topics.cs @@ -112,7 +112,8 @@ public void Subscribe(string topicId) { foreach (PeerId peerId in fanoutPeers) { - if (gPeers[topicId].Contains(peerId) && + if (!IsDirectPeer(peerId) && + gPeers[topicId].Contains(peerId) && peerState.TryGetValue(peerId, out PubsubPeer? peer) && GetPeerScore(peerId) >= 0 && (!peer.Backoff.TryGetValue(topicId, out DateTime backoff) || backoff <= DateTime.Now) && @@ -238,10 +239,19 @@ public void Publish(string topicId, byte[] message) rpc.WithMessages(topicId, seqNo++, localPeer.Identity.PeerId.Bytes, message, localPeer.Identity); } + HashSet directRecipients = GetDirectPeersForTopic(topicId).ToHashSet(); + foreach (PeerId peerId in directRecipients) + { + if (ShouldSendFullMessage(peerId, topicId)) + { + peerState.GetValueOrDefault(peerId)?.Send(rpc); + } + } + // Floodsub peers always get the message. foreach (PeerId peerId in fPeers.GetValueOrDefault(topicId) ?? []) { - if (ShouldSendFullMessage(peerId, topicId)) + if (!directRecipients.Contains(peerId) && ShouldSendFullMessage(peerId, topicId)) { peerState.GetValueOrDefault(peerId)?.Send(rpc); } @@ -253,7 +263,9 @@ public void Publish(string topicId, byte[] message) // Send to all gossipsub peers above publish threshold foreach (PeerId peerId in allGossipsubPeers) { - if (GetPeerScore(peerId) >= _settings.PublishThreshold && ShouldSendFullMessage(peerId, topicId)) + if (!directRecipients.Contains(peerId) && + GetPeerScore(peerId) >= _settings.PublishThreshold && + ShouldSendFullMessage(peerId, topicId)) { peerState.GetValueOrDefault(peerId)?.Send(rpc); } @@ -264,7 +276,9 @@ public void Publish(string topicId, byte[] message) // Standard gossipsub v1.0 behavior: send to mesh or fanout foreach (PeerId peerId in meshPeers) { - if (GetPeerScore(peerId) >= _settings.PublishThreshold && ShouldSendFullMessage(peerId, topicId)) + if (!directRecipients.Contains(peerId) && + GetPeerScore(peerId) >= _settings.PublishThreshold && + ShouldSendFullMessage(peerId, topicId)) { peerState.GetValueOrDefault(peerId)?.Send(rpc); } @@ -281,7 +295,7 @@ public void Publish(string topicId, byte[] message) if (topicPeers is { Count: > 0 }) { // Select peers with non-negative scores - var eligiblePeers = topicPeers.Where(p => GetPeerScore(p) >= 0).ToList(); + var eligiblePeers = topicPeers.Where(p => !IsDirectPeer(p) && GetPeerScore(p) >= 0).ToList(); foreach (PeerId peer in eligiblePeers.Take(_settings.Degree)) { topicFanout.Add(peer); @@ -291,7 +305,9 @@ public void Publish(string topicId, byte[] message) foreach (PeerId peerId in topicFanout) { - if (GetPeerScore(peerId) >= _settings.PublishThreshold && ShouldSendFullMessage(peerId, topicId)) + if (!directRecipients.Contains(peerId) && + GetPeerScore(peerId) >= _settings.PublishThreshold && + ShouldSendFullMessage(peerId, topicId)) { peerState.GetValueOrDefault(peerId)?.Send(rpc); } @@ -328,7 +344,7 @@ public void PublishPartial(string topicId, byte[] groupId, byte[]? partialMessag HashSet fanoutPeers = fanout.GetOrAdd(topicId, _ => []); if (fanoutPeers.Count == 0 && gPeers.TryGetValue(topicId, out HashSet? topicPeers)) { - foreach (PeerId peerId in topicPeers.Where(peerId => GetPeerScore(peerId) >= 0).Take(_settings.Degree)) + foreach (PeerId peerId in topicPeers.Where(peerId => !IsDirectPeer(peerId) && GetPeerScore(peerId) >= 0).Take(_settings.Degree)) { fanoutPeers.Add(peerId); } diff --git a/src/libp2p/Libp2p.Protocols.Pubsub/PubsubRouter.cs b/src/libp2p/Libp2p.Protocols.Pubsub/PubsubRouter.cs index d777b2ac..853ab244 100644 --- a/src/libp2p/Libp2p.Protocols.Pubsub/PubsubRouter.cs +++ b/src/libp2p/Libp2p.Protocols.Pubsub/PubsubRouter.cs @@ -278,6 +278,8 @@ public Action? SendRpc private readonly ConcurrentBag reconnections = []; private readonly PeerStore _peerStore; + private readonly IReadOnlyDictionary directPeers; + private readonly ConcurrentDictionary pendingDirectDials = new(); private ulong seqNo = 1; // Lifetime of work owned by the router: background loops and dials it starts itself. @@ -335,6 +337,12 @@ public PubsubRouter(PeerStore store, PubsubSettings? settings = null, ILoggerFac throw new InvalidOperationException("StrictNoSign requires a custom GetMessageId function."); } + if (_settings.DirectConnectPeriod <= 0) + { + throw new ArgumentOutOfRangeException(nameof(PubsubSettings.DirectConnectPeriod), "DirectConnectPeriod must be positive."); + } + + directPeers = CreateDirectPeers(_settings.DirectPeers); _messageCache = new(_settings.MessageCacheTtl); _limboMessageCache = new(_settings.MessageCacheTtl); _idontwantMessages = new(_settings.MessageCacheTtl); @@ -364,13 +372,26 @@ public Task StartAsync(ILocalPeer localPeer, CancellationToken token = default) _lifetime = CancellationTokenSource.CreateLinkedTokenSource(token, _stopped.Token); CancellationToken lifetime = _lifetime.Token; - _loops = Task.WhenAll( + List loops = [ Task.Run(() => LoopHeartbeat(lifetime), lifetime), - Task.Run(() => LoopReconnect(lifetime), lifetime)); + Task.Run(() => LoopReconnect(lifetime), lifetime), + ]; + if (directPeers.Count > 0) + { + loops.Add(Task.Run(() => LoopReconnectDirectPeers(lifetime), lifetime)); + } + _loops = Task.WhenAll(loops); + + foreach (Multiaddress[] directPeerAddresses in directPeers.Values) + { + _peerStore.Discover(directPeerAddresses); + } + ConnectDirectPeers(lifetime); _onNewPeer = (addrs) => { - if (addrs.Any(a => a.GetPeerId()! == localPeer.Identity.PeerId)) + // Direct peers are dialed by ConnectDirectPeers. + if (addrs.Any(a => a.GetPeerId() is PeerId peerId && (peerId == localPeer.Identity.PeerId || IsDirectPeer(peerId)))) { return; } @@ -403,6 +424,15 @@ private async Task LoopReconnect(CancellationToken token) } } + private async Task LoopReconnectDirectPeers(CancellationToken token) + { + while (!token.IsCancellationRequested) + { + await Task.Delay(_settings.DirectConnectPeriod, token); + ConnectDirectPeers(token); + } + } + private void Track(Task connect) { _connects.TryAdd(connect, 0); @@ -546,6 +576,43 @@ private void Reconnect(CancellationToken token) } } + private static IReadOnlyDictionary CreateDirectPeers(IEnumerable? configuredPeers) + { + return (configuredPeers ?? []) + .Select(address => (PeerId: address.GetPeerId() ?? throw new ArgumentException("A direct peer address must include a peer ID.", nameof(PubsubSettings.DirectPeers)), Address: address)) + .GroupBy(entry => entry.PeerId) + .ToDictionary(group => group.Key, group => group.Select(entry => entry.Address).ToArray()); + } + + private void ConnectDirectPeers(CancellationToken token) + { + foreach ((PeerId peerId, Multiaddress[] addresses) in directPeers) + { + // The direct-peer loop owns retries, so a dial that is still in flight is not repeated. + if (peerState.ContainsKey(peerId) || !pendingDirectDials.TryAdd(peerId, 0)) + { + continue; + } + + Task connect = Connect(addresses, token); + Track(connect); + _ = connect.ContinueWith(t => pendingDirectDials.TryRemove(peerId, out byte _), TaskScheduler.Default); + } + } + + private bool IsDirectPeer(PeerId peerId) => directPeers.ContainsKey(peerId); + + private bool IsDirectPeerSubscribedTo(PeerId peerId, string topic) + { + return (fPeers.TryGetValue(topic, out HashSet? floodsubPeers) && floodsubPeers.Contains(peerId)) || + (gPeers.TryGetValue(topic, out HashSet? gossipsubPeers) && gossipsubPeers.Contains(peerId)); + } + + private IEnumerable GetDirectPeersForTopic(string topic) + { + return directPeers.Keys.Where(peerId => IsDirectPeerSubscribedTo(peerId, topic)); + } + public Task Heartbeat() { // Apply score decay @@ -584,6 +651,7 @@ public Task Heartbeat() // Need to graft more peers - exclude peers with negative scores PeerId[] peersToGraft = (gPeers.GetValueOrDefault(topic) ?? []) .Where(p => !meshPeers.Contains(p) + && !IsDirectPeer(p) && GetPeerScore(p) >= 0 // Only graft non-negative scoring peers && (peerState.GetValueOrDefault(p)?.Backoff.TryGetValue(topic, out DateTime backoff) != true || backoff <= DateTime.Now)) .Take(_settings.Degree - meshPeers.Count).ToArray(); @@ -669,7 +737,9 @@ public Task Heartbeat() int peerCountToAdd = _settings.Degree - fanout[fanoutTopic].Count; if (peerCountToAdd > 0) { - foreach (PeerId? peerId in (gPeers.GetValueOrDefault(fanoutTopic) ?? []).Where(p => !fanout[fanoutTopic].Contains(p)).Take(peerCountToAdd)) + foreach (PeerId? peerId in (gPeers.GetValueOrDefault(fanoutTopic) ?? []) + .Where(p => !fanout[fanoutTopic].Contains(p) && !IsDirectPeer(p)) + .Take(peerCountToAdd)) { fanout[fanoutTopic].Add(peerId); } @@ -704,6 +774,7 @@ onPartialGossip is not null && PeerId[] eligiblePeers = topicGossipsubPeers .Where(p => !topicMesh.Contains(p) && !fanoutPeers.Contains(p) + && !IsDirectPeer(p) && GetPeerScore(p) >= _settings.GossipThreshold) .ToArray(); @@ -779,7 +850,7 @@ private void RemovePeer(PeerId peerId, Multiaddress addr, PubsubPeer peer) topicPeers.Remove(peerId); } } - if (!peer.ReconnectionPolicy.Suppressed && !_stopped.IsCancellationRequested) + if (!peer.ReconnectionPolicy.Suppressed && !_stopped.IsCancellationRequested && !IsDirectPeer(peerId)) { reconnections.Add(new Reconnection([addr], _settings.ReconnectionAttempts, peer.ReconnectionPolicy)); }