--- name: dart-libp2p-gossipsub description: 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. triggers: dart_libp2p_pubsub, GossipSubRouter, GossipSubParams, FloodSubRouter, RandomSubRouter, PubSubMessage, registerTopicValidator, PeerScoreParams, PeerScoreThresholds, meshsub, gossipsub --- # dart_libp2p_pubsub (GossipSub, FloodSub, RandomSub) ## 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 ```yaml dependencies: dart_libp2p_pubsub: ^3.1.0 dart_libp2p: '>=4.0.1 <5.0.0' # what dart_libp2p_pubsub 3.1.0 requires ``` ```dart 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 ### Minimal node ```dart // 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 ```dart 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) ```dart 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 ```dart 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 ```dart 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 `IpfsDHT` (dart_libp2p_kad_dht) implements dart_libp2p's `Discovery`, so it can be passed directly (see `dart-libp2p-kad-dht`): ```dart final pubsub = PubSub(host, GossipSubRouter(), discovery: dht); // Advertises each subscribed topic under 'floodsub:' (same as Go), // polls every second for topics without enough peers and dials what it finds. ``` ### Other options on PubSub ```dart 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 - `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` (broadcast) of `PubSubMessage`; `cancel()`, `isCancelled`. - `PubSub.publish(topic, Uint8List, {ready, readyTimeout})`: signs, validates locally, routes, delivers locally. - `PubSubMessage`: `topic`, `data` (`List`), `from` (author `PeerId`), `seqno`, `receivedFrom` (null if local), `rpcMessage`. - `TopicValidator`: `FutureOr 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>`, 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 - **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`.** 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/`), 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 `Router`s 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 - **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: ```dart Future 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(); 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 - 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 - Docs: https://cryptopeer.net/libp2p/pubsub/ (pages: network-setup, gossipsub-usage, gossipsub-deep-dive, testing, configuration, best-practices), e.g. https://cryptopeer.net/libp2p/pubsub/configuration/ - Repo: https://github.com/stephanfeb/dart_libp2p_pubsub (CHANGELOG.md has the go-libp2p-pubsub alignment history) - Package: https://pub.dev/packages/dart_libp2p_pubsub - Real usage: `dart-libp2p-demos/internet_chat` (DHT rendezvous + GossipSub rooms) - Related skills: `dart-libp2p-host`, `dart-libp2p-kad-dht`, `merkle-crdt-sync`