--- name: merkle-crdt-sync description: Use when replicating merkledag Merkle-CRDTs between dart_libp2p peers with the dart_libp2p_merkle_crdt package (import package:dart_libp2p_merkle_crdt/dart_libp2p_merkle_crdt.dart): MerkleCrdtGossipManager, GossipSubAnnouncer, P2PStreamDagFetcher, WriterAuthorizer, BlockStore, DocumentCRDT, optional DHT block lookup. triggers: dart_libp2p_merkle_crdt, MerkleCrdtGossipManager, GossipSubAnnouncer, P2PStreamDagFetcher, WriterAuthorizer, InMemoryBlockStore, DocumentCRDT, defaultFetchProtocolId, merkle-crdt/heads, /merkle-crdt/fetch --- # merkle-crdt-sync (dart_libp2p_merkle_crdt) ## What it is `dart_libp2p_merkle_crdt` replicates a `MerkleCRDT` from the `merkledag` package between `dart_libp2p` peers. Each replica announces its heads (root CIDs) on a GossipSub topic; a peer that hears an unknown head fetches the missing blocks over a libp2p stream protocol, verifies each one against its CID, and merges. The announcing peer is asked first, so sync works without a DHT (for example over a BLE mesh); a Kademlia DHT is an optional fallback. Use something else for: - The CRDT data types, `MerkleNode`, `MerkleCRDT` and local use: skill `merkle-crdt` (package `merkledag`). - Host construction (transports, Noise, Yamux): skill `dart-libp2p-host`. - GossipSub tuning, scoring, validators in general: skill `dart-libp2p-gossipsub`. - DHT setup and bootstrap: skill `dart-libp2p-kad-dht`. - Arbitrary IPFS block exchange: skill `dart-libp2p-bitswap` (this package does not use Bitswap). - Social payloads (profile, following, outbox, ActivityPub-style activities): the separate `dart_libp2p_merkle_crdt_social` package. ## Install ```yaml dependencies: dart_libp2p_merkle_crdt: ^2.0.0 merkledag: ^2.0.0 dart_libp2p: '>=4.0.1 <5.0.0' dart_libp2p_pubsub: '>=1.6.0 <4.0.0' dart_libp2p_kad_dht: '>=1.2.0 <2.0.0' # only if you pass a DHT dcid: ^1.0.0 # if you handle CID values directly ``` ```dart import 'package:dart_libp2p/dart_libp2p.dart' hide Subscription; // Host, PeerId, AddrInfo import 'package:dart_libp2p_merkle_crdt/dart_libp2p_merkle_crdt.dart'; import 'package:dart_libp2p_pubsub/dart_libp2p_pubsub.dart'; // PubSub, GossipSubRouter import 'package:merkledag/merkledag.dart'; // GSet, CRDTPayload, MerkleNode import 'package:dcid/dcid.dart' show CID; ``` The single library export brings in the manager, announcer, fetcher, block stores, framing helpers, `DHTProvider`, `DocumentCRDT` and `VectorClock`. ## Core workflow ### Minimal replica (two peers, grow-only set) Adapted from `example/dart_libp2p_merkle_crdt_example.dart`. `host` is a started `Host` (see `dart-libp2p-host`). ```dart typedef Replica = MerkleCrdtGossipManager, GSet>; // Future startReplica(Host host, WriterAuthorizer writers) async { final pubsub = PubSub(host, GossipSubRouter()); // StrictSign is the default await pubsub.start(); final fetcher = P2PStreamDagFetcher, Set>( // host, InMemoryBlockStore(), defaultFetchProtocolId, // '/merkle-crdt/fetch/2.0.0' GSet.fromCanonicalBytes, // payload decoder // dht: dht, // optional IpfsDHTv2 ); final announcer = GossipSubAnnouncer(host, pubsub, 'example-set', writers: writers); return Replica(dagFetcher: fetcher, announcer: announcer); } // Both peers use the same instance id and the same writer set. final writers = WriterAuthorizer.only([hostA.id, hostB.id]); final a = await startReplica(hostA, writers); final b = await startReplica(hostB, writers); await hostB.connect(AddrInfo(hostA.id, hostA.addrs)); b.onStateChanged.listen((state) => print('B sees: $state')); await a.applyLocalUpdate(GSet.fromList(['apple'])); ``` ### Read state ```dart final Set? value = await replica.getState(); // merged logical value final GSet? payload = await replica.getPayload(); // merged payload final Set heads = replica.heads; // >1 = concurrent writes final cached = replica.currentState; // value at last notification replica.onCidChanged.listen((cid) => print('head now $cid')); ``` ### Update a document (build from the current payload) ```dart final docs = MerkleCrdtGossipManager, DocumentCRDT>( dagFetcher: P2PStreamDagFetcher>( host, InMemoryBlockStore(), defaultFetchProtocolId, DocumentCRDT.fromBytes), announcer: GossipSubAnnouncer(host, pubsub, 'docs', writers: writers), ); const me = 'actor-a'; // stable and unique per writer final current = await docs.getPayload() ?? DocumentCRDT.empty(); await docs.applyLocalUpdate(current.put('title', 'Hello', actorId: me)); // Deletes are tombstones, again built from the current payload. await docs.applyLocalUpdate((await docs.getPayload())!.delete('title', actorId: me)); ``` `DocumentCRDT` also has `patch(map, actorId:)`, `deepMerge(map, actorId:)`, `fromMap(map, actorId:)`, `withStrategy(...)` and `withConflictResolver(...)`. ### Recover after a partition or late join ```dart // GossipSub does not replay missed messages. Re-announce every head (and // re-provide them to the DHT if one is configured). await replica.forceAnnounceCurrentHead(); ``` A peer that only hears the latest head still catches up: the fetcher walks the DAG backwards block by block from the announcer (tested with 30 updates). ### Optional DHT fallback ```dart final fetcher = P2PStreamDagFetcher, Set>( host, store, defaultFetchProtocolId, GSet.fromCanonicalBytes, dht: dht, // IpfsDHTv2 from dart_libp2p_kad_dht maxProviders: 3, // providers asked per block ); ``` With a DHT, every stored block is provided (fire and forget), and a block no recent announcer has is looked up with `findProvidersAsync`. ### Persistent storage Implement the three-method `BlockStore`; store only verified nodes. ```dart class SqliteBlockStore implements BlockStore, Set> { @override Future>?> get(CID cid) async { /* load bytes, then MerkleNode.fromBytes, Set>(bytes, GSet.fromCanonicalBytes, expectedCid: cid) */ } @override Future has(CID cid) async { /* ... */ throw UnimplementedError(); } @override Future put(MerkleNode> node) async { /* insert node.toBytes() keyed by node.cid unless already present */ } } ``` ### Shut down ```dart await replica.close(); // cancels retries, leaves the topic, removes the // stream handler, closes fetcher and announcer await pubsub.stop(); // PubSub, DHT and Host are yours to close await host.close(); ``` ## Key concepts / API map - `MerkleCrdtGossipManager`: the replica. Constructor takes `dagFetcher`, `announcer`, `loggerName`, `maxTraversalNodes` (10000), `maxMissingNodes` (8), `maxRetries` (3), `retryDelay` (2 s, doubling). - `applyLocalUpdate(P)` returns the new head `CID`; `getState()`, `getPayload()`, `refreshState()`, `forceAnnounceCurrentHead()`, `close()`. - `onStateChanged` (`Stream`), `onCidChanged` (`Stream`), `heads`, `latestKnownCid`, `currentState`. - `GossipSubAnnouncer(host, pubsub, crdtInstanceId, {topic, required writers})`: implements merkledag `Broadcaster`; `topic`, `announcements` (`Stream`), `broadcast(cidString)`, `close()`. - `HeadAnnouncement`: `cid` and authenticated author `from`. - `WriterAuthorizer`: `.only(peerIds)` (a `PeerIdAllowlist`), `.allowAll()`, or subclass and override `isAllowed(PeerId)`. - `P2PStreamDagFetcher(host, blockStore, protocolId, payloadFactory, {dht, maxBlockSize = 4 MiB, fetchTimeout = 10 s, maxHintPeers = 8, maxProviders = 3})`: implements `DAGSyncer`; `get`, `put`, `notePeer`, `blockStore`, `nodeProtocol`, `dhtProvider`, `isDHTAvailable`, `getStats()`. - `P2PNodeProtocol`: the fetch protocol client and server; `fetch(cid, peer)`, `fetchNodeFromProvider(cid, addrInfo)`, `close()`. - `defaultFetchProtocolId`: `'/merkle-crdt/fetch/2.0.0'`. - `BlockStore` (`get`, `has`, `put`) and `InMemoryBlockStore` (`length`). - `DHTProvider`: wraps an optional `IpfsDHTv2`; every call is a no-op without one. - `encodeFrame(bytes)`, `FrameReader(stream).readFrame(maxLength:)`, `FrameException`: unsigned-varint length-prefixed frames. - `DocumentCRDT`, `DocumentEntry`, `ConflictResolutionStrategy` (`lastWriterWins`, `firstWriterWins`, `custom`; `merge` deprecated), `ConflictResolver`, `VectorClock`, `CausalRelation`. Wire protocol (2.0.0): - Heads topic `merkle-crdt/heads/2.0.0/`; each message is the raw bytes of one CID (at most 128 bytes). - Fetch: one request and one response per stream, each one frame. Request `0x01 ++ cidBytes`. Response `0x00 ++ MerkleNode.toBytes()`, `0x01` (not held) or `0x02 ++ utf8 error`. ## Gotchas - **Type parameters are in opposite orders.** The manager is `MerkleCrdtGossipManager` (value, payload), the fetcher and stores are `P2PStreamDagFetcher` / `InMemoryBlockStore` (payload, value). `, GSet>` for the manager, `, Set>` for the fetcher. - **`writers` is required, and both sides must agree.** Announcements from a peer the local `WriterAuthorizer` does not allow are dropped before any fetch and are not relayed (validator returns `ignore`). If B's update never reaches A, check A's allowlist contains B's `PeerId`. - **Writer checks need signed messages.** The check uses `message.from`, authenticated only under StrictSign (the `PubSub` default). Constructing `PubSub(..., noAuthor: true)` or a no-sign policy removes the author, so an allowlist can no longer work. Keep the default. - **One validator per topic.** The announcer registers a topic validator on first use. If something else already validates that topic it logs a warning and only filters what is delivered locally, so relaying is then decided by the other validator. Do not register your own validator on `merkle-crdt/heads/...`. - **Nothing happens until the topic is joined.** The announcer joins (subscribes) lazily; the manager subscribes in its constructor. Create the manager before expecting to receive heads. - **GossipSub needs a mesh.** A publish made before the peers have grafted each other is lost, and GossipSub never replays it. Connect the hosts, wait for the mesh (tests poll `router.mesh[topic]`), and call `forceAnnounceCurrentHead()` after reconnects, on a timer, or when a peer joins. Announce failures are logged, not thrown. - **Same instance id, same topic.** Peers sync only if they pass the same `crdtInstanceId` (or the same explicit `topic`) and the same fetch protocol ID. Use one instance id per CRDT document. - **State-based payloads only.** Each node carries the full merged state; the value is the merge of the current heads. `merge` must be commutative, associative and idempotent, and `toCanonicalBytes` deterministic. A payload whose decoded form does not re-encode to the same bytes is rejected on receipt (`InvalidNodeException`), so a lossy `payloadFactory` breaks sync. - **`GSet.fromCanonicalBytes` supports `String` elements only**; other element types throw `UnsupportedError`. Write your own payload for other types. - **No-op updates produce no head.** `applyLocalUpdate` with a payload that does not change the state returns the existing head and broadcasts nothing. - **Build `DocumentCRDT` updates from `getPayload()`.** Conflicts are resolved by vector-clock sum, then actor ID, then tombstone-over-value, then canonical JSON. An update built from `DocumentCRDT.empty()` has a small clock and can lose to older writes. Each writer needs a stable, unique `actorId`. - **`ConflictResolutionStrategy.merge` is deprecated** and behaves as `lastWriterWins`. Model nested fields as separate keys. If replicas use different strategies, the one listed first in the enum wins the merge. - **Custom resolvers are not serialised.** Re-attach in the payload factory: `(b) => DocumentCRDT.fromBytes(b).withConflictResolver(resolver)`. The resolver must be commutative, associative, idempotent and must not depend on which argument is local, or replicas diverge. - **Deleting is a CRDT operation.** Use `DocumentCRDT.delete` (tombstone). Never remove blocks from a `BlockStore`: peers still walk them, and the next announcement fetches them again. `put` of an existing CID is a no-op by design. - **Several heads is normal.** Concurrent writes leave more than one head; the next local update folds them into one node. `latestKnownCid` returns only one of them; use `heads` to compare replicas. - **Merging is bounded.** A merge visits at most `maxTraversalNodes` nodes and gives up after `maxMissingNodes` unavailable ones. Replicas diverged further than that will not converge in one merge; raise the bounds rather than removing them. An unfetchable head is retried `maxRetries` times with backoff, then logged and dropped until announced again. - **Block size limit is 4 MiB per node** (both sides). Large states replicate (tests send more than 800 KiB) but a state above `maxBlockSize` cannot be served; split it across several CRDT instances. - **Without a DHT only recent announcers are asked** (the last `maxHintPeers`). A peer that holds blocks but never announced is not found unless a DHT is configured. DHT calls use `cid.toV1()`. - **2.0.0 does not interoperate with 1.x.** Topic, fetch protocol, CID derivation (merkledag 2.0.0) and the announcement format all changed. 1.x sent JSON announcements; 2.x sends raw CID bytes. - **dart_libp2p below 4.0.1** fails with "Stream is closed" in the UDX/yamux write path; keep the pinned range. `dart_libp2p_pubsub` below 1.6.0 lacks topic validators. - **The manager owns the fetcher and announcer.** `close()` closes both; do not share one fetcher or announcer between managers. It does not close `PubSub`, the DHT or the `Host`. - **Not interoperable with IPFS/go-libp2p sync.** The heads topic and fetch protocol are specific to this package; Bitswap peers cannot serve these blocks. ## Testing The package tests use real hosts over UDX loopback (`test/real_net_stack.dart` builds a `BasicHost` with Noise and Yamux) and wait on conditions rather than fixed sleeps: ```dart import 'dart:async'; // FutureOr import 'package:test/test.dart'; // fail Future until(FutureOr Function() condition, String what, {Duration timeout = const Duration(seconds: 20)}) async { final deadline = DateTime.now().add(timeout); while (!await condition()) { if (DateTime.now().isAfter(deadline)) fail('Timed out waiting for $what'); await Future.delayed(const Duration(milliseconds: 50)); } } // Wait for the GossipSub mesh before writing. final router = GossipSubRouter(); final pubsub = PubSub(host, router); // ... await until(() => router.mesh[announcer.topic]?.contains(otherId) ?? false, 'mesh'); await a.applyLocalUpdate(GSet.fromList(['x'])); await until(() async => (await b.getState() ?? {}).contains('x'), 'B to sync'); ``` Tips: - Shorten timings: `fetchTimeout: Duration(seconds: 3)` on the fetcher and `retryDelay: Duration(milliseconds: 200)` on the manager. - Replace `0.0.0.0` with `127.0.0.1` in listen addresses before dialling. - Compare convergence with `heads` sets and `getState()`, not with `latestKnownCid`. - Test a misbehaving peer by registering your own handler for `defaultFetchProtocolId` on a second host and replying with `encodeFrame`; forged blocks, oversized frames and silent peers must all return `null`. - For payload logic alone, test `merge` in every order without a network (see `test/unit/p2p/crdt/document_crdt_convergence_test.dart`). - In `tearDown`, close managers, then DHTs, then `pubsub.stop()`, then `host.close()`. Run: `dart test` (integration tests open UDP sockets on loopback). ## Red lines - Never use `WriterAuthorizer.allowAll()` for a CRDT that is not meant to be open to every peer on the topic. - Never disable message signing (`noAuthor`, no-sign policies) on a `PubSub` that carries heads topics. - Never store a node in a `BlockStore` that was not made with `MerkleNode.create` or decoded with `MerkleNode.fromBytes(..., expectedCid:)`. Never build `MerkleNode(cid: ...)` from untrusted data. - Never delete or overwrite blocks to "delete" data; use tombstones. - Never remove the traversal, block-size or timeout limits: they are what stops a hostile peer from stalling or exhausting the replica. - Never change the topic format, frame format or protocol ID while claiming compatibility with 2.0.0 peers; version them instead. - Always `close()` managers you create; otherwise stream handlers, retry timers and subscriptions leak. ## Reference - Docs: https://cryptopeer.net/libp2p/merkle-crdt-sync/ - Data types and local use: https://cryptopeer.net/libp2p/merkle-crdt/ - GossipSub: https://cryptopeer.net/libp2p/pubsub/gossipsub-usage/ - DHT: https://cryptopeer.net/libp2p/dht/ - Repository: https://github.com/stephanfeb/dart-libp2p-merkle-crdt - Example: `example/dart_libp2p_merkle_crdt_example.dart`; integration tests: `test/integration/sync_test.dart`