Nethermind.Libp2p.Protocols.Pubsub 1.1.1

Pubsub protocol set

This package implements the libp2p pubsub protocol family. It includes floodsub and gossipsub protocol handlers and provides a shared PubsubRouter for topic operations.

See the libp2p pubsub spec for protocol details.

Setup

Enable pubsub on the standard stack with WithPubsub():

using Microsoft.Extensions.DependencyInjection;
using System.Text;
using Nethermind.Libp2p;
using Nethermind.Libp2p.Core;
using Nethermind.Libp2p.Protocols.Pubsub;

ServiceProvider provider = new ServiceCollection()
    .AddLibp2p(builder => builder.WithPubsub())
    .BuildServiceProvider();

ILocalPeer peer = provider.GetRequiredService<IPeerFactory>().Create();
await peer.StartListenAsync();

PubsubRouter router = provider.GetRequiredService<PubsubRouter>();
await router.StartAsync(peer);

StartAsync connects the router to the local peer and starts heartbeat and reconnect loops. Call it once for the peer that owns the router.

Subscribing to a topic

GetTopic returns an ITopic and subscribes by default:

ITopic chat = router.GetTopic("chat");

chat.OnMessage += (peerId, bytes) =>
{
    string text = Encoding.UTF8.GetString(bytes);
    Console.WriteLine($"{peerId}: {text}");
};

The peerId argument is the connected peer that delivered the RPC message to the local router. It is not necessarily the original pubsub message author when messages are forwarded through the mesh. If your application needs author identity, include it in the published payload and validate it at the application layer.

To create a topic handle without subscribing immediately, pass subscribe: false and call Subscribe() later:

ITopic alerts = router.GetTopic("alerts", subscribe: false);
alerts.Subscribe();

Only a current local subscription admits received full messages. A remote announcement, score configuration, publishing fanout, or an unsubscribed topic handle does not establish delivery interest. Unwanted payloads are ignored before message-ID calculation, signature or application validation, caches, delivery scoring, events, and forwarding. Subscriptions and control messages in the same RPC still run. Subscribed topics retain their default delivery rewards and invalid-message penalties even without explicit score parameters.

Publishing

Publish raw bytes or protobuf messages:

chat.Publish(Encoding.UTF8.GetBytes("hello from dotnet-libp2p"));

// Any Google.Protobuf.IMessage can also be published.
// chat.Publish(myProtobufMessage);

Publishing does not create peers by itself. The router sends messages to connected pubsub peers and uses PeerStore.OnNewPeer to dial newly discovered peers that advertise pubsub support.

Validating received messages

VerifyMessage runs after libp2p signature verification and before the message is delivered, forwarded, or placed in the message cache. Its callback takes the connected peer that sent the RPC and the message. Applications using the older message-only callback must add the peer argument. Return Accepted to deliver and forward, Rejected to record an invalid delivery, Ignored to suppress without a penalty, or Throttled to drop without caching the message ID. Do not mutate the message in the callback.

For asynchronous checks, assign one OnDeferredMessage callback before configuring VerifyMessage to return Deferred, especially when the router is already running. The callback receives the same message instance outside the router lock. Complete it with CompleteValidation(message, verdict) before the timeout; a false return means the message is no longer pending, expired, or changed after authentication. A deferred message is not delivered or forwarded while pending.

router.OnDeferredMessage = async (sourcePeer, message) =>
{
    bool valid = await ValidateAsync(sourcePeer, message);
    router.CompleteValidation(message, valid ? MessageValidity.Accepted : MessageValidity.Rejected);
};
router.VerifyMessage = (sourcePeer, message) => NeedsAsyncCheck(message)
    ? MessageValidity.Deferred
    : MessageValidity.Accepted;

Configure MaxPendingValidationMessages, MaxPendingValidationBytes, and PendingValidationTimeout in PubsubSettings to bound deferred work. PendingValidationCount reports the current unexpired count. If the store is full, the callback is absent, or the timeout is invalid, the router drops the deferred message without scoring its source because these are local capacity or configuration conditions. Expired entries are removed when the router next processes a message or queries the count; application validation work is not cancelled on expiry or disposal. An invalid signature is rejected before VerifyMessage and does not reserve its message ID, so replaying the same invalid message repeats signature verification.

Connecting pubsub peers

Use one of the discovery mechanisms to populate PeerStore, or add known peer addresses manually:

using Multiformats.Address;
using Nethermind.Libp2p.Core.Discovery;

PeerStore peerStore = provider.GetRequiredService<PeerStore>();
peerStore.Discover([
    Multiaddress.Decode("/ip4/127.0.0.1/tcp/4001/p2p/12D3KooW...")
]);

When a new peer is discovered, PubsubRouter dials it and negotiates the best supported pubsub protocol from the remote peer's identified protocols. Gossipsub v1.3 is preferred, then gossipsub v1.2, gossipsub v1.1, gossipsub v1.0, and floodsub.

Application-managed connections

If the application owns connection scheduling, peer limits or compatibility checks, register settings before resolving the router:

services.AddSingleton(new PubsubSettings { AutoConnect = false });

With AutoConnect disabled, the router does not dial discovered peers (including previously stored addresses and PX), connect configured direct peers, or retry closed pubsub streams. Heartbeats and gossip on existing sessions continue. Identify, Identify Push and discovery still verify and store metadata; storing an address does not grant permission to connect. The default is true, preserving automatic connection management.

The application dials the transport, completes its checks, then opens a supported pubsub protocol on the admitted session. For example, if the peer advertises Gossipsub v1.3:

Task gossipStream = admittedSession.DialAsync<GossipsubProtocolV13>(token);

The task lasts for the stream's lifetime: keep and observe it with the application's session tasks. The router opens a reply stream on the same session when the remote opens pubsub; this does not dial another transport connection. Disabling automatic connections is not an inbound protocol admission gate. Applications that require checks before serving inbound pubsub must enforce those checks in their protocol handlers. The application also owns reconnecting direct peers when automatic connections are disabled, including the explicit peering agreement's connectivity requirement.

This setting controls Pubsub only. Other enabled protocols that dial transports, such as Kad-DHT, need their own integration with the application's connection manager. No beacon-specific admission policy is part of Identify, the peer store or the session API.

Unsubscribing

chat.Unsubscribe();
router.UnsubscribeAll();

Unsubscribing announces the topic leave to connected pubsub peers and prunes the local mesh for that topic. It also removes pending validations for the topic and releases their capacity; their later CompleteValidation calls return false, even after resubscribing. Already accepted callbacks queued for dispatch may finish concurrently with unsubscribe. Application validation work is not cancelled.

Outstanding IWANT promises survive local unsubscribe. Replies on the unwanted topic are ignored and do not fulfill those promises, so expiry can still incur a behavioral penalty for the peer that announced the message.

Mesh admission and prune backoff

GRAFT admission requires a nonnegative peer score. Rejected negative-score and premature GRAFTs receive PRUNE without peer exchange (PX); both refresh the local backoff. Heartbeat pruning also records backoff, including pruning for excess mesh degree. Gossipsub v1.0 receives topic-only PRUNE; v1.1–v1.3 also receive the backoff duration.

Received nonzero PRUNE backoff is capped at 24 hours, before converting its wire uint64 seconds or calculating a deadline. This is a local defensive policy, not a protocol maximum. An omitted or zero duration uses configured PruneBackoff. Configured local PruneBackoff and UnsubscribeBackoff (milliseconds) remain authoritative, including durations longer than 24 hours. Negative local durations are treated as zero; zero unsubscribe backoff permits immediate re-entry. Wire durations round up to whole seconds.

Backoff uses the router's injected UTC TimeProvider. Refresh preserves a later existing deadline within a retention horizon of the maximum of 24 hours and the two configured local durations. Deadlines saturate at the maximum representable date. Receipt of PRUNE removes the peer from the mesh before processing backoff or PX, so extreme durations cannot leave membership intact or abort later controls in that RPC.

These rules follow gossipsub v1.1 scoring and backoff; the 24-hour cap is this implementation's policy.

Remote subscription limits

Remote membership is retained independently of local subscriptions so a later local subscription can discover peers. Configure these positive settings before constructing the router:

Each membership charges 2L * topic.Length bytes for UTF-16 characters, including when multiple peers share a topic. Count limits also bound collection overhead; byte limits do not measure total managed-heap usage. Application-created local topics and historical scoring keep their existing lifetimes.

Announcements are processed in wire order. Per-peer limits always apply. At a global count or byte limit, a membership for a currently subscribed local topic can reclaim capacity by evicting the oldest unwanted memberships. Existing memberships on subscribed topics are protected. Local subscribe and unsubscribe update that priority without discarding remote announcements. The router checks that the entire new membership can fit before evicting anything; all count and byte limits remain hard bounds.

Other additions exceeding a budget are ignored without a penalty or disconnect; later entries, including unsubscribe, still run. Duplicate announcements consume no additional capacity and can update partial-message options. Rejected or evicted memberships require a new announcement to become known again, so discovery of topics without a local subscription is best effort at saturation. GRAFT's implicit membership uses the same priority and limits, and responds with PRUNE if it cannot fit. Saturation by subscribed memberships can still reject further wanted memberships; applications must size the limits for their active topics and peers.

Remote unsubscribe and disconnect release membership and partial-message metadata. Empty remote topic keys are removed from IRoutingStateContainer.FloodsubPeers and GossipsubPeers; local subscription alone no longer creates these keys. An active local subscription keeps its empty mesh so heartbeat can refill it. Local unsubscribe retains remote announcements, and publishing without subscribing continues to use fanout.

Outbound work and incoming dispatch

Configure these positive limits before constructing the router:

Every outbound RPC uses the same budgets, including before a reply stream is attached. Admission never waits for space. Each peer has one asynchronous writer and owns immutable serialized frames; mutating an RPC after admission cannot change queued data. Reservations remain charged until the underlying write finishes, including a partly consumed write during shutdown. Byte limits measure encoded payloads, rather than total managed memory; entry limits also bound queue overhead. Required control queued behind a busy writer, or before a reply stream attaches, is coalesced: a subscribe, unsubscribe, GRAFT or PRUNE for a topic the pending frame does not yet mention merges into that frame, so subscribing to many topics in a burst costs one entry rather than one per topic. A repeated GRAFT is redundant and a repeated PRUNE is superseded; control for a topic already pending starts a new frame so receivers see the same per-topic order. Merging charges only the byte delta, keeps the frame's original admission time, and never reads the caller's RPC again.

At capacity, payloads, partial messages, IHAVE, IDONTWANT and IWANT requests drop newest without a score penalty. Losing a subscribe, unsubscribe, GRAFT, PRUNE or first-frame extension advertisement instead closes that peer generation and disconnects its owning session. A failed write also closes the generation, because continuing a possibly incomplete framed stream is unsafe. This session closure can affect other protocols sharing the connection. Local subscriptions survive teardown; a fresh connection advertises the current snapshot. Insufficient capacity for that snapshot closes the new connection again. Automatic connections retain the existing retry policy; with AutoConnect = false, the application owns recovery.

Required control has reserved headroom within both router-wide limits: the smaller of one peer's limit and half the router-wide limit, rounded up, is unavailable to optional traffic. With the defaults, optional traffic can occupy at most 4,064 entries and 60 MiB; required control can use the full 4,096 entries and 64 MiB. This keeps stalled peers' optional traffic from consuming the space for healthy peers' subscription, mesh and initial extension announcements. An RPC that combines required control with payloads, partial messages, IHAVE, IWANT or IDONTWANT is queued as two frames: a control-only frame checked against the full limits, followed by the optional frame checked against the optional limits. Optional content therefore never occupies the reserved headroom, including when a remote peer provokes a combined response such as a PRUNE with IWANT replies. Dropping the optional frame leaves the control frame queued and the generation open. Control-only frames remain bounded by the per-peer limits and OutboundRpcTimeout. A one-entry router-wide limit leaves no optional entry capacity. If required traffic also exhausts the total budget, or an individual required frame cannot fit, its recipient still closes rather than retaining inconsistent subscription or mesh state. No capacity waiters, retry backlog or separate control queue are created.

OutboundRpcTimeout is a positive construction-time deadline measured with monotonic time. Every frame must complete within this interval after admission, including queue residence and waiting for a reply stream to attach. Expiry closes the generation holding the stalled work, aborts its stream and disconnects its owning session without a score penalty. One owned timer per generation tracks its oldest outstanding frame and is disposed during teardown. Queued frames release capacity on teardown; an active write remains charged until the underlying operation actually terminates, even after cancellation or abort has been requested.

IWANT promises reserve bounded tracking capacity only after admission. Queue residence and stalled writes do not consume the follow-up deadline. A successful channel write starts that deadline; fulfillment, failed writes and teardown cancel pending promises. A write completion from an old connection cannot arm a replacement connection's promise. Incoming request and response-attempt limits remain conservative abuse limits even when outbound work is dropped.

Receive loops await message-ID admission before reading another frame. The incoming budgets bound both semaphore waiters and RPCs retained through dispatch. An admitted RPC stays charged until its synchronous application callbacks return, so inbound publish concurrency is bounded by the entry budget divided by handler latency; slow handlers cost incoming capacity rather than forwarding latency. Each connected peer may hold at most its per-peer share, charged across every inbound stream it has opened, so one peer cannot take the router-wide budget by opening many streams. The share narrows that exposure rather than closing it: the router-wide check runs first and reserves nothing for peers holding nothing, so with the defaults sixteen peers each holding eight entries, or fewer peers holding large RPCs up to the byte budget, exhaust the router-wide budget and every other peer's publishes are skipped with reason capacity until entries release. A peer holding nothing always gets one RPC admitted, up to MaxRpcBytes, so a share smaller than a maximum-size RPC never blocks delivery; the share bounds that peer's additional concurrent RPCs. The router-wide byte budget must be at least MaxRpcBytes, which construction enforces. At either limit, publishes are skipped before hashing, validation, caching, delivery credit or penalties; subscriptions and controls still run, and the skipped message IDs are not cached, so gossip can request them again. A skip also cancels the sender's pending IWANT promises, as a throttled validation does, so local overload never becomes a broken-promise penalty for a peer that did deliver. Control-only, unwanted-topic and graylisted traffic keep their early gates. Calculations share the existing concurrency limit with synchronous application publishing. Cancellation stops pending admission; already running host calculations and application callbacks cannot be forcibly cancelled.

Forwarding and IDONTWANT admission happen before synchronous full-message and partial-message callbacks, including accepted deferred validations. Application events retain their signatures, ordering and exception isolation and run outside the router lock. An admitted frame is queued work, not an acknowledgement of delivery.

The dotnet-libp2p meter reports libp2p.pubsub.outbound.dropped and libp2p.pubsub.overload.disconnects, with bounded reason labels (capacity, serialization or timeout), and libp2p.pubsub.incoming.skipped with reason capacity (router-wide) or peer-capacity. None carry peer or topic labels.

Stopping the router

The router stops when the token passed to StartAsync is cancelled or when the router is disposed. Disposal cancels its heartbeat and reconnect loops and the dials it started, closes the pubsub streams it serves and stops reacting to new PeerStore peers. It does not dispose the PeerStore or the local peer.

await router.DisposeAsync(); // waits for the owned background work to finish

A router resolved from the service provider is disposed together with the provider, so dispose it yourself only when you created it with new. Dispose() stops the same work without waiting for it. A disposed router cannot be started again.