Support Gossipsub partial messages

src/libp2p/Libp2p.Protocols.Pubsub/Dto/Rpc.cs

@@ -3166,8 +3166,9 @@ public void MergeFrom(pb::CodedInputStream input) {
   }
 
   /// <summary>
-  /// The Partial Messages registry identifies topics as opaque bytes. Go's
-  /// current string declaration is wire compatible with this canonical form.
+  /// The Partial Messages registry identifies topics as opaque bytes. The base
+  /// pubsub API uses UTF-8 topic strings, so the router validates and decodes this
+  /// field at its boundary. Go's current string declaration is wire compatible.
   /// </summary>
   [global::System.Diagnostics.DebuggerDisplayAttribute("{ToString(),nq}")]
   public sealed partial class PartialMessagesExtension : pb::IMessage<PartialMessagesExtension>

src/libp2p/Libp2p.Protocols.Pubsub/Dto/Rpc.proto

@@ -71,8 +71,9 @@ message ControlExtensions {
     optional bool partialMessages = 10;
 }
 
-// The Partial Messages registry identifies topics as opaque bytes. Go's
-// current string declaration is wire compatible with this canonical form.
+// The Partial Messages registry identifies topics as opaque bytes. The base
+// pubsub API uses UTF-8 topic strings, so the router validates and decodes this
+// field at its boundary. Go's current string declaration is wire compatible.
 message PartialMessagesExtension {
     optional bytes topicID = 1;
     optional bytes groupID = 2;

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

@@ -0,0 +1,31 @@
+// SPDX-FileCopyrightText: 2026 Demerzel Solutions Limited
+// SPDX-License-Identifier: MIT
+
+using Nethermind.Libp2p.Core;
+
+namespace Nethermind.Libp2p.Protocols.Pubsub;
+
+/// <summary>
+/// A topic configured for the Gossipsub v1.3 Partial Messages extension.
+/// </summary>
+public interface IPartialMessagesTopic : ITopic
+{
+    /// <summary>
+    /// Raised for application-defined partial message payloads received for this topic.
+    /// </summary>
+    event Action<PeerId, PartialMessage>? OnPartialMessage;
+
+    bool RequestsPartialMessages { get; }
+    bool SupportsSendingPartialMessages { get; }
+
+    /// <summary>
+    /// Sends an application-defined partial message to the topic mesh or fanout peers.
+    /// </summary>
+    void PublishPartial(byte[] groupId, byte[]? partialMessage = null, byte[]? partsMetadata = null);
+
+    /// <summary>
+    /// Sends an application-defined partial message to a connected peer, including
+    /// a non-mesh peer selected by application gossip logic.
+    /// </summary>
+    void SendPartial(PeerId peerId, byte[] groupId, byte[]? partialMessage = null, byte[]? partsMetadata = null);
+}

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

@@ -0,0 +1,43 @@
+// SPDX-FileCopyrightText: 2026 Demerzel Solutions Limited
+// SPDX-License-Identifier: MIT
+
+namespace Nethermind.Libp2p.Protocols.Pubsub;
+
+/// <summary>
+/// Application-defined data carried by the Gossipsub v1.3 Partial Messages extension.
+/// </summary>
+public sealed class PartialMessage
+{
+    public PartialMessage(string topicId, byte[] groupId, byte[]? partialData, byte[]? partsMetadata)
+    {
+        ArgumentNullException.ThrowIfNull(topicId);
+        ArgumentNullException.ThrowIfNull(groupId);
+
+        TopicId = topicId;
+        GroupId = groupId;
+        PartialData = partialData;
+        PartsMetadata = partsMetadata;
+    }
+
+    public string TopicId { get; }
+    public byte[] GroupId { get; }
+    public byte[]? PartialData { get; }
+    public byte[]? PartsMetadata { get; }
+}
+
+/// <summary>
+/// Per-topic capabilities advertised through Gossipsub subscription options.
+/// </summary>
+public sealed class PartialMessagesTopicOptions
+{
+    /// <summary>
+    /// Requests partial data from peers. This also requires support for sending
+    /// partial messages.
+    /// </summary>
+    public bool RequestPartialMessages { get; init; }
+
+    /// <summary>
+    /// Signals that this topic can send partial data and receive parts metadata.
+    /// </summary>
+    public bool SupportsSendingPartialMessages { get; init; }
+}

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

@@ -0,0 +1,104 @@
+// SPDX-FileCopyrightText: 2026 Demerzel Solutions Limited
+// SPDX-License-Identifier: MIT
+
+namespace Nethermind.Libp2p.Protocols.Pubsub;
+
+/// <summary>
+/// Bounded local partial-message group state used to give applications concrete
+/// group identifiers when gossiping to non-mesh peers.
+/// </summary>
+internal sealed class PartialMessageGossipCache
+{
+    private sealed class Group(byte[] id, int remainingHeartbeats)
+    {
+        public byte[] Id { get; } = id;
+        public int RemainingHeartbeats { get; set; } = remainingHeartbeats;
+        public LinkedListNode<Group>? AgeNode { get; set; }
+    }
+
+    private sealed class TopicGroups
+    {
+        public Dictionary<string, Group> ById { get; } = [];
+        public LinkedList<Group> ByAge { get; } = [];
+    }
+
+    private readonly Dictionary<string, TopicGroups> topics = [];
+    private readonly int maxGroupsPerTopic;
+    private readonly int groupTtlHeartbeats;
+
+    public PartialMessageGossipCache(int maxGroupsPerTopic, int groupTtlHeartbeats)
+    {
+        ArgumentOutOfRangeException.ThrowIfNegativeOrZero(maxGroupsPerTopic);
+        ArgumentOutOfRangeException.ThrowIfNegativeOrZero(groupTtlHeartbeats);
+        this.maxGroupsPerTopic = maxGroupsPerTopic;
+        this.groupTtlHeartbeats = groupTtlHeartbeats;
+    }
+
+    public void Track(string topic, ReadOnlySpan<byte> groupId)
+    {
+        TopicGroups groups = topics.GetValueOrDefault(topic) ?? CreateTopic(topic);
+        string key = Convert.ToHexString(groupId);
+        if (groups.ById.TryGetValue(key, out Group? existing))
+        {
+            existing.RemainingHeartbeats = groupTtlHeartbeats;
+            groups.ByAge.Remove(existing.AgeNode!);
+            existing.AgeNode = groups.ByAge.AddLast(existing);
+            return;
+        }
+
+        if (groups.ById.Count == maxGroupsPerTopic)
+        {
+            Remove(groups, groups.ByAge.First!.Value);
+        }
+
+        Group group = new(groupId.ToArray(), groupTtlHeartbeats);
+        group.AgeNode = groups.ByAge.AddLast(group);
+        groups.ById.Add(key, group);
+    }
+
+    public IReadOnlyList<byte[]> GetGroupIds(string topic)
+    {
+        return topics.TryGetValue(topic, out TopicGroups? groups)
+            ? groups.ByAge.Select(group => group.Id.ToArray()).ToArray()
+            : [];
+    }
+
+    public void Heartbeat()
+    {
+        foreach ((string topic, TopicGroups groups) in topics.ToArray())
+        {
+            LinkedListNode<Group>? node = groups.ByAge.First;
+            while (node is not null)
+            {
+                LinkedListNode<Group>? next = node.Next;
+                Group group = node.Value;
+                group.RemainingHeartbeats--;
+                if (group.RemainingHeartbeats == 0)
+                {
+                    Remove(groups, group);
+                }
+
+                node = next;
+            }
+
+            if (groups.ById.Count == 0)
+            {
+                topics.Remove(topic);
+            }
+        }
+    }
+
+    private TopicGroups CreateTopic(string topic)
+    {
+        TopicGroups groups = new();
+        topics.Add(topic, groups);
+        return groups;
+    }
+
+    private static void Remove(TopicGroups groups, Group group)
+    {
+        groups.ById.Remove(Convert.ToHexString(group.Id));
+        groups.ByAge.Remove(group.AgeNode!);
+        group.AgeNode = null;
+    }
+}

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

@@ -40,6 +40,18 @@ public int MaxRpcBytes
     public SignaturePolicy DefaultSignaturePolicy { get; set; } = SignaturePolicy.StrictSign;
     public int MaxIdontwantMessages { get; set; } = 50;
 
+    /// <summary>
+    /// Enables the opt-in Gossipsub v1.3 Partial Messages extension. The router
+    /// advertises it only to v1.3 peers.
+    /// </summary>
+    public bool EnablePartialMessages { get; set; }
+
+    /// <summary>Number of heartbeats to retain a locally published partial-message group for gossip.</summary>
+    public int PartialMessageGossipTtlHeartbeats { get; set; } = 3;
+
+    /// <summary>Maximum locally published partial-message groups retained for one topic.</summary>
+    public int MaxPartialMessageGroupsPerTopic { get; set; } = 255;
+
     public Func<Message, MessageId> GetMessageId { get; set; } = ConcatFromAndSeqno;
 
     public enum SignaturePolicy

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

@@ -6,17 +6,21 @@
 using Nethermind.Libp2p.Core;
 using Nethermind.Libp2p.Protocols.Pubsub.Dto;
 using System.Collections.Concurrent;
+using System.Text;
 
 namespace Nethermind.Libp2p.Protocols.Pubsub;
 
 public partial class PubsubRouter : IRoutingStateContainer, IDisposable
 {
+    private static readonly UTF8Encoding StrictUtf8 = new(encoderShouldEmitUTF8Identifier: false, throwOnInvalidBytes: true);
+
     internal void OnRpc(PeerId peerId, Rpc rpc, string? protocolId = null, bool isFirstRpc = true)
     {
         try
         {
             ConcurrentDictionary<PeerId, Rpc> peerMessages = new();
             List<(string Topic, PeerId PeerId, byte[] Data)> receivedMessages = [];
+            List<(string Topic, PeerId PeerId, PartialMessage Message)> receivedPartialMessages = [];
             lock (this)
             {
                 HandleExtensions(peerId, rpc, protocolId, isFirstRpc);
@@ -31,6 +35,11 @@ internal void OnRpc(PeerId peerId, Rpc rpc, string? protocolId = null, bool isFi
                     HandleSubscriptions(peerId, rpc.Subscriptions);
                 }
 
+                if (rpc.Partial is not null)
+                {
+                    HandlePartialMessage(peerId, rpc.Partial, protocolId, receivedPartialMessages);
+                }
+
                 if (rpc.Control is not null)
                 {
                     if (rpc.Control.Graft.Count != 0)
@@ -64,6 +73,11 @@ internal void OnRpc(PeerId peerId, Rpc rpc, string? protocolId = null, bool isFi
                 OnMessage?.Invoke(topic, receivedFrom, data);
             }
 
+            foreach ((string topic, PeerId receivedFrom, PartialMessage message) in receivedPartialMessages)
+            {
+                OnPartialMessage?.Invoke(topic, receivedFrom, message);
+            }
+
             foreach (KeyValuePair<PeerId, Rpc> peerMessage in peerMessages)
             {
                 peerState.GetValueOrDefault(peerMessage.Key)?.Send(peerMessage.Value);
@@ -98,6 +112,57 @@ private void HandleExtensions(PeerId peerId, Rpc rpc, string? protocolId, bool i
             ApplyBehaviorPenalty(peerId, 1.0);
             logger?.LogDebug("Ignoring repeated Gossipsub v1.3 extensions from {peerId}", peerId);
         }
+        if (isFirstRpc && extensions?.PartialMessages == true)
+        {
+            peer.SupportsPartialMessagesExtension = true;
+        }
+    }
+
+    private void HandlePartialMessage(PeerId peerId, PartialMessagesExtension partialMessage, string? protocolId, List<(string Topic, PeerId PeerId, PartialMessage Message)> receivedPartialMessages)
+    {
+        if (!_settings.EnablePartialMessages ||
+            protocolId is not null && protocolId != GossipsubProtocolVersionV13 ||
+            !peerState.TryGetValue(peerId, out PubsubPeer? peer) ||
+            !peer.SupportsPartialMessagesExtension)
+        {
+            return;
+        }
+
+        if (!partialMessage.HasTopicID ||
+            !partialMessage.HasGroupID ||
+            (!partialMessage.HasPartialMessage && !partialMessage.HasPartsMetadata))
+        {
+            logger?.LogDebug("Ignoring an incomplete Partial Messages extension payload from {peerId}", peerId);
+            return;
+        }
+
+        if (!TryDecodeTopicId(partialMessage.TopicID, out string topicId))
+        {
+            logger?.LogDebug("Ignoring a Partial Messages extension payload with a non-UTF-8 topic from {peerId}", peerId);
+            return;
+        }
+
+        topicState.TryGetValue(topicId, out Topic? topic);
+        if (partialMessage.HasPartialMessage && topic?.RequestsPartialMessages is not true)
+        {
+            ApplyBehaviorPenalty(peerId, 1.0);
+            logger?.LogDebug("Ignoring unsolicited partial data from {peerId} for topic {topicId}", peerId, topicId);
+            return;
+        }
+
+        if (topic?.IsSubscribed is not true || !topic.SupportsSendingPartialMessages)
+        {
+            return;
+        }
+
+        receivedPartialMessages.Add((
+            topicId,
+            peerId,
+            new PartialMessage(
+                topicId,
+                partialMessage.GroupID.ToByteArray(),
+                partialMessage.HasPartialMessage ? partialMessage.PartialMessage.ToByteArray() : null,
+                partialMessage.HasPartsMetadata ? partialMessage.PartsMetadata.ToByteArray() : null)));
     }
 
     private void HandleNewMessages(PeerId peerId, IEnumerable<Message> messages, ConcurrentDictionary<PeerId, Rpc> peerMessages, List<(string Topic, PeerId PeerId, byte[] Data)> receivedMessages)
@@ -166,7 +231,10 @@ private void HandleNewMessages(PeerId peerId, IEnumerable<Message> messages, Con
                     {
                         continue;
                     }
-                    peerMessages.GetOrAdd(peer, _ => new Rpc()).Publish.Add(message);
+                    if (ShouldSendFullMessage(peer, message.Topic))
+                    {
+                        peerMessages.GetOrAdd(peer, _ => new Rpc()).Publish.Add(message);
+                    }
                 }
             }
             if (mesh.TryGetValue(message.Topic, out topicPeers))
@@ -179,7 +247,7 @@ private void HandleNewMessages(PeerId peerId, IEnumerable<Message> messages, Con
                     }
 
                     // Only forward to peers above publish threshold (Gossipsub v1.1)
-                    if (GetPeerScore(peer) >= _settings.PublishThreshold)
+                    if (GetPeerScore(peer) >= _settings.PublishThreshold && ShouldSendFullMessage(peer, message.Topic))
                     {
                         peerMessages.GetOrAdd(peer, _ => new Rpc()).Publish.Add(message);
                     }
@@ -188,6 +256,20 @@ private void HandleNewMessages(PeerId peerId, IEnumerable<Message> messages, Con
         }
     }
 
+    private static bool TryDecodeTopicId(ByteString topicIdBytes, out string topicId)
+    {
+        try
+        {
+            topicId = StrictUtf8.GetString(topicIdBytes.Span);
+            return true;
+        }
+        catch (DecoderFallbackException)
+        {
+            topicId = string.Empty;
+            return false;
+        }
+    }
+
     private void HandleSubscriptions(PeerId peerId, IEnumerable<Rpc.Types.SubOpts> subscriptions)
     {
         foreach (Rpc.Types.SubOpts? sub in subscriptions)
@@ -207,9 +289,19 @@ private void HandleSubscriptions(PeerId peerId, IEnumerable<Rpc.Types.SubOpts> s
                 {
                     fPeers.GetOrAdd(sub.Topicid, _ => []).Add(peerId);
                 }
+
+                if (_settings.EnablePartialMessages && state.SupportsPartialMessagesExtension)
+                {
+                    bool requestsPartialMessages = sub.RequestsPartial;
+                    state.UpdatePartialMessagesSubscription(
+                        sub.Topicid,
+                        requestsPartialMessages,
+                        sub.SupportsSendingPartial);
+                }
             }
             else
             {
+                state.RemovePartialMessagesSubscription(sub.Topicid);
                 if (state.IsGossipSub)
                 {
                     gPeers.GetOrAdd(sub.Topicid, _ => []).Remove(peerId);