Skip to content
cryptopeercryptopeer.net

dart-libp2p-gossipsub

Skill dart-libp2p-gossipsub. Use when writing Dart code that publishes or subscribes over libp2p pubsub with package dart_libp2p_pubsub (PubSub, GossipSubRouter, FloodSubRouter, RandomSubRouter): topics, message validation, peer scoring, GossipSubParams tuning, topic discovery, tracing, testing, go-libp2p-pubsub interop.

SKILL.md (Claude Code, Codex, Cursor) · dart-libp2p-gossipsub.md (Boubou) · install: curl -fsSL https://cryptopeer.net/skills/install.sh | sh -s -- dart-libp2p-gossipsub

What it is

Section titled “What it is”

dart_libp2p_pubsub is topic-based publish/subscribe on top of a dart_libp2p host. GossipSubRouter follows go-libp2p-pubsub v0.15.0: it speaks /meshsub/1.3.0, 1.2.0, 1.1.0, 1.0.0 and /floodsub/1.0.0 and negotiates one per peer, with validation, opt-in peer scoring, IHAVE/IWANT/IDONTWANT, Peer Exchange and topic discovery. FloodSubRouter and RandomSubRouter exist for testing and research.

Use something else when you need request/response or bulk transfer (open a libp2p stream, see dart-libp2p-host), content addressing (dart-libp2p-bitswap) or replicated state (merkle-crdt-sync, which uses this package for heads). Building the host itself is covered by the dart-libp2p-host skill.

Install

Section titled “Install”
dependencies:
  dart_libp2p_pubsub: ^3.1.0
  dart_libp2p: '>=4.0.1 <5.0.0'   # what dart_libp2p_pubsub 3.1.0 requires
import 'dart:async';   // Completer, TimeoutException
import 'dart:convert';
import 'dart:typed_data';
import 'package:dart_libp2p_pubsub/dart_libp2p_pubsub.dart';
// If you also import the dart_libp2p barrel, hide its event-bus Subscription:
import 'package:dart_libp2p/dart_libp2p.dart' hide Subscription;

Core workflow

Section titled “Core workflow”

Minimal node

Section titled “Minimal node”
// host: a started dart_libp2p Host whose peerstore holds its private key
// (see dart-libp2p-host).
final router = GossipSubRouter();          // default GossipSubParams
final pubsub = PubSub(host, router);       // attaches the router
await pubsub.start();                       // starts heartbeat, greets peers


const topic = '/my-app/1.0.0/chat';
final sub = pubsub.subscribe(topic);        // first subscribe joins the mesh
sub.stream.listen((dynamic m) {
  if (m is! PubSubMessage) return;
  if (m.from == host.id) return;            // own publishes are delivered too
  print('${m.from.toBase58()}: ${utf8.decode(m.data)}');
}, onError: (e, s) => print('sub error: $e'));


// The host must be connected to at least one peer on the topic.
await host.connect(AddrInfo(otherPeerId, otherAddrs));
await pubsub.publish(topic, Uint8List.fromList(utf8.encode('hello')));


// Shutdown
await sub.cancel();        // closes the stream, PRUNEs mesh peers on last sub
await pubsub.stop();       // stops router, closes pubsub streams

Wait for peers before publishing

Section titled “Wait for peers before publishing”
await pubsub.publish(topic, data,
    ready: minTopicSize(3), readyTimeout: const Duration(seconds: 30));
// Throws TimeoutException if not ready in time (message is not sent),
// StateError if PubSub stops first.

Validate messages (per topic)

Section titled “Validate messages (per topic)”
pubsub.registerTopicValidator(topic, (PeerId receivedFrom, PubSubMessage msg) async {
  if (msg.data.length > 4096) return ValidationResult.reject; // invalid: penalise
  if (!await isRelevant(msg)) return ValidationResult.ignore; // drop, no penalty
  return ValidationResult.accept;                              // forward + deliver
}, timeout: const Duration(seconds: 2), concurrency: 64);


pubsub.unregisterTopicValidator(topic); // returns false if none was registered

Legacy, all topics, synchronous: pubsub.registerMessageValidator((topic, m) => true).

Tune the mesh

Section titled “Tune the mesh”
final router = GossipSubRouter(
  params: GossipSubParams(
    D: 8, DLow: 6, DHigh: 12,
    heartbeatInterval: const Duration(seconds: 1),
    seenMessagesTTL: const Duration(minutes: 2),
  ),
);

Turn on peer scoring

Section titled “Turn on peer scoring”
final router = GossipSubRouter(
  scoreParams: PeerScoreParams(
    topics: {
      topic: TopicScoreParams(
        topicWeight: 1,
        invalidMessageDeliveriesWeight: -10,                 // P4
        invalidMessageDeliveriesDecay: scoreParameterDecay(const Duration(hours: 1)),
      ),
    },
    behaviourPenaltyWeight: -10,                             // P7
    behaviourPenaltyDecay: scoreParameterDecay(const Duration(hours: 1)),
  ),
  scoreThresholds: const PeerScoreThresholds(
    gossipThreshold: -10, publishThreshold: -50, graylistThreshold: -80,
    acceptPXThreshold: 10, opportunisticGraftThreshold: 5,
  ),
);
// A topic created later:
router.setTopicScoreParams('/my-app/1.0.0/room/42', TopicScoreParams(topicWeight: 1));
// Inspect:
final s = router.score?.score(peerId);
final snap = router.score?.snapshot(peerId); // score, topics, behaviourPenalty...

Topic discovery through the DHT

Section titled “Topic discovery through the DHT”

IpfsDHT (dart_libp2p_kad_dht) implements dart_libp2p’s Discovery, so it can be passed directly (see dart-libp2p-kad-dht):

final pubsub = PubSub(host, GossipSubRouter(), discovery: dht);
// Advertises each subscribed topic under 'floodsub:<topic>' (same as Go),
// polls every second for topics without enough peers and dials what it finds.

Other options on PubSub

Section titled “Other options on PubSub”
PubSub(host, router,
  privateKey: key,                 // default: host's key from its peerstore
  signaturePolicy: MessageSignaturePolicy.strictSign, // strictNoSign, laxSign, laxNoSign
  noAuthor: false,                 // true omits from/seqno (forces a NoSign policy)
  messageIdFn: defaultMessageIdFn, // from bytes + seqno, as Go
  maxMessageSize: 1024 * 1024,     // defaultMaxMessageSize
  peerOutboundQueueSize: 32,       // defaultPeerOutboundQueueSize
  validatorTimeout: const Duration(seconds: 5),
  validateThrottle: 8192,
  blacklist: Blacklist(),
  tracer: JsonEventTracer(...),    // or PbEventTracer; default NoOpEventTracer
);
pubsub.blacklistPeer(badPeer);
pubsub.getTopics();

Other routers: FloodSubRouter() and RandomSubRouter(networkSize: n). Bootstrapper/PX node: GossipSubRouter(doPX: true).

Key concepts / API map

Section titled “Key concepts / API map”
  • PubSub(host, router, {...}): the service; owns subscriptions, validation, signing, streams.
  • PubSub.start() / stop(): stop() keeps subscriptions and validators; start() rejoins.
  • PubSub.subscribe(topic) -> Subscription; unsubscribe(topic) cancels all of a topic’s subscriptions.
  • Subscription.stream: Stream<dynamic> (broadcast) of PubSubMessage; cancel(), isCancelled.
  • PubSub.publish(topic, Uint8List, {ready, readyTimeout}): signs, validates locally, routes, delivers locally.
  • PubSubMessage: topic, data (List<int>), from (author PeerId), seqno, receivedFrom (null if local), rpcMessage.
  • TopicValidator: FutureOr<ValidationResult> Function(PeerId receivedFrom, PubSubMessage).
  • ValidationResult.accept / reject / ignore.
  • PubSub.announceSubscriptionsTo(peer): resend our subscriptions to one peer (normally automatic).
  • GossipSubRouter({params, scoreParams, scoreThresholds, doPX, peerGaterParams, testExtension}).
  • GossipSubRouter.mesh / fanout: Map<String, Set<PeerId>>, handy for test assertions.
  • GossipSubRouter.score (PeerScore?), setTopicScoreParams, certifiedAddrBook, isStarted.
  • GossipSubParams: D 6, DLow 5, DHigh 12, DScore 4, DOut 2, DLazy 6, fanoutTTL 1 min, heartbeatInterval 1 s, pruneBackoff 1 min, unsubscribeBackoff 10 s, floodPublish true, prunePeers 16, seenMessagesTTL 2 min, seenMessagesStrategy.
  • PeerScoreParams, TopicScoreParams, PeerScoreThresholds, scoreParameterDecay(Duration): Go’s scoring model (P1 to P7).
  • PeerGaterParams: Go’s peer gater, off by default.
  • minTopicSize(n), RouterReady, Router.enoughPeers(topic, n).
  • Protocol IDs: gossipSubIDv10..v13, floodSubID, randomSubID.
  • discoveryNamespacePrefix (floodsub:), discoveryPollInterval, defaultDiscoveryConnector.
  • MessageIdFn, defaultMessageIdFn, messageIdFromBytes, messageIdToBytes.
  • EventTracer, NoOpEventTracer, JsonEventTracer, PbEventTracer (Go’s trace formats).
  • Router: interface for custom routers (protocols, acceptFrom, enoughPeers, handleRpc, …).

Gotchas

Section titled “Gotchas”
  • No connection, no mesh. PubSub never dials on its own (except discovery and PX). Subscribing does not find peers: connect with host.connect, a bootstrap list, or pass discovery:. A mesh only forms between peers that are connected and both subscribed to the same topic.
  • Mesh formation takes heartbeats. Subscribing GRAFTs up to D peers at once, but peers that connect later are grafted by the heartbeat (1 s). Publishing in the same tick as connecting can reach nobody. Use publish(..., ready: minTopicSize(n)) or wait for router.mesh[topic] to be non-empty.
  • Listen before messages arrive. Subscription.stream is a broadcast stream and does not buffer. Attach the listener right after subscribe.
  • The stream is Stream<dynamic>. Check m is PubSubMessage before use.
  • Your own messages come back. publish delivers to local subscribers too. Filter on m.from == host.id (or m.receivedFrom == null) if you do not want them.
  • Signing needs a private key. With no privateKey:, PubSub signs with the host’s key from host.peerStore.keyBook. If the peerstore has none, publish throws StateError. Before 1.4.1 such messages were dropped silently.
  • Messages for topics you do not subscribe to are dropped before validation (3.0.0, as Go). A relay node must subscribe to every topic it should carry.
  • GossipSubParams is validated in the constructor (ArgumentError): DLow <= D <= DHigh, DScore <= DHigh, DOut < DLow and DOut < D ~/ 2, historyGossip <= historyLength. With the default DOut: 2 any D <= 4 throws; small meshes must set DOut (and DLow, default 5) too, e.g. GossipSubParams(D: 3, DLow: 2, DHigh: 4, DScore: 2, DOut: 0). All zero is the bootstrapper setting (no mesh).
  • Scoring is all or nothing. scoreParams and scoreThresholds must be given together, or the constructor throws. Thresholds must satisfy graylistThreshold <= publishThreshold <= gossipThreshold <= 0, acceptPXThreshold >= 0, opportunisticGraftThreshold >= 0. With scoring on and all thresholds at 0, any negative score graylists a peer (its RPCs are ignored).
  • reject costs the delivering peer, not the author. Return reject only for messages that are invalid for everyone; honest forwarders relay whatever their own validator accepts. Use ignore for stale or unwanted-but-valid messages.
  • Validator limits. One validator per topic: registering a second throws StateError (unregister first). A validator that throws or exceeds its timeout gives ignore. Validators also run on your own publishes, and a rejected local message is dropped with only a log warning.
  • Duplicates are not re-validated. A message ID is remembered for seenMessagesTTL (2 min) once its signature checks out. Do not rely on a validator seeing a resent copy.
  • Message IDs changed in 2.0.0 to Go’s from + seqno. Nodes on 1.x and

    =2.0.0 compute different IDs: upgrade a network together. A custom messageIdFn must be identical on every node.

  • Size limit. RPC frames above maxMessageSize (1 MiB) reset the stream. A publish whose RPC would exceed it throws ArgumentError. Large payloads belong in Bitswap or a stream; publish a CID instead.
  • Topic names are global strings. Use versioned, app-specific paths (/my-app/1.0.0/room/<id>), not chat or test. Go peers must use the exact same string.
  • Every node in a network should use the same heartbeatInterval.
  • Peer Exchange is off by default (doPX: true for bootstrappers), but the router accepts PX from any peer scoring at least acceptPXThreshold.
  • Outbound queue drops. Each peer has a 32-RPC queue; bursts beyond it are dropped (traced as DROP_RPC). Raise peerOutboundQueueSize for bursty publishers.
  • Stale workarounds in older code. Code written for 1.x (including the internet_chat demo, pinned to ^1.4.2) sets fanoutTTL short because the heartbeat used to run every fanoutTTL, and calls announceSubscriptionsTo after every connect. Since 1.6.0 the heartbeat uses heartbeatInterval and subscriptions are exchanged on connect; do not copy these workarounds.
  • Upgrading 2.x to 3.x: custom Routers must implement enoughPeers; trace files switched to Go’s formats; GossipSub now negotiates /meshsub/1.3.0. Upgrading 1.x to 2.x: scoring moved from PubSub to GossipSubRouter (router.score), PX became opt-in, stop() no longer disposes the tracer.
  • stop() does not end subscription streams or dispose the tracer. Cancel subscriptions and call tracer.dispose() yourself on final shutdown.
  • Do not share a router. PubSub attaches the router in its constructor; create one router per PubSub.
  • Go interop (verified by test/interop/go_pubsub_interop_test.dart): Go publishes and Dart receives, Dart publishes and Go receives, and IHAVE/IWANT repair, over TCP + Noise + Yamux. Defaults (strict signing, message IDs, floodsub: discovery namespace) match Go; keep them unless every peer changes.

Testing

Section titled “Testing”
  • Real stack, in process. Build two or more hosts on loopback (see dart-libp2p-host; the repo’s test/real_net_stack.dart has a createLibp2pNode helper), start a PubSub on each, connect them, subscribe on both, then wait for the mesh before publishing:
Future<void> waitForMesh(GossipSubRouter r, String topic,
    {Duration timeout = const Duration(seconds: 10)}) async {
  final end = DateTime.now().add(timeout);
  while ((r.mesh[topic]?.isEmpty ?? true)) {
    if (DateTime.now().isAfter(end)) throw TimeoutException('no mesh on $topic');
    await Future.delayed(const Duration(milliseconds: 100));
  }
}


final got = Completer<PubSubMessage>();
subA.stream.listen((m) { if (m is PubSubMessage && !got.isCompleted) got.complete(m); });
await waitForMesh(routerA, topic);
await psB.publish(topic, Uint8List.fromList(utf8.encode('hi')));
final msg = await got.future.timeout(const Duration(seconds: 10));
expect(msg.from, hostB.id);
  • Prefer polling router.mesh or publish(ready: minTopicSize(1)) over fixed Future.delayed sleeps (the repo’s own tests use 5 to 8 s sleeps; they are slow and flaky under load).
  • Mocks, no network. test/integration/message_propagation_test.dart shows a MockHost/MockNetwork/TestNetworkManager that pipes streams in memory, then routerA.addPeer(hostB.id, gossipSubIDv11) to link routers directly. These mocks are in test/, not exported; copy the pattern. Pass privateKey: when the mock peerstore has no key.
  • Discovery can be tested with an in-memory Discovery implementation (test/integration/discovery_test.dart, _MemoryDiscovery).
  • Always await pubsub.stop() and close hosts in tearDown, or timers keep the test isolate alive. Give tests an explicit timeout:.
  • Turn on logs with Logger.root.level = Level.FINE (loggers PubSub, GossipSubRouter, PubSubComm, …); the library never prints.

Red lines

Section titled “Red lines”
  • Never turn off signature verification (laxSign, laxNoSign) or use noAuthor on an open network unless every node agrees and messages are authenticated another way.
  • Never trust PubSubMessage.data without a topic validator on networks with untrusted peers; the validator is what keeps your node from relaying junk.
  • Never return reject for messages that are merely unwanted; it penalises honest relays and can split the mesh.
  • Never change message ID function, signature policy or topic names on one node only; that breaks interop with the rest of the network (Dart or Go).
  • Never publish large blobs over pubsub; publish a reference.
  • Never leave PubSub running after the host closes; call stop() first.
  • Never import from package:dart_libp2p_pubsub/src/... in application code; use the public barrel.

Reference

Section titled “Reference”

Built by Stephan Februarygithub.com/stephanfebLinkedIn