merkle-crdt-sync

Skill merkle-crdt-sync. 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.

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

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

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

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).

typedef Replica = MerkleCrdtGossipManager<Set<String>, GSet<String>>; // <V, P>

Future<Replica> startReplica(Host host, WriterAuthorizer writers) async {
  final pubsub = PubSub(host, GossipSubRouter()); // StrictSign is the default
  await pubsub.start();

  final fetcher = P2PStreamDagFetcher<GSet<String>, Set<String>>( // <P, V>
    host,
    InMemoryBlockStore(),
    defaultFetchProtocolId,           // '/merkle-crdt/fetch/2.0.0'
    GSet<String>.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<String>.fromList(['apple']));

Read state

final Set<String>? value = await replica.getState();   // merged logical value
final GSet<String>? payload = await replica.getPayload(); // merged payload
final Set<CID> 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)

final docs = MerkleCrdtGossipManager<Map<String, dynamic>, DocumentCRDT>(
  dagFetcher: P2PStreamDagFetcher<DocumentCRDT, Map<String, dynamic>>(
      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

// 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

final fetcher = P2PStreamDagFetcher<GSet<String>, Set<String>>(
  host, store, defaultFetchProtocolId, GSet<String>.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.

class SqliteBlockStore implements BlockStore<GSet<String>, Set<String>> {
  @override
  Future<MerkleNode<GSet<String>>?> get(CID cid) async { /* load bytes, then
    MerkleNode.fromBytes<GSet<String>, Set<String>>(bytes,
        GSet<String>.fromCanonicalBytes, expectedCid: cid) */ }
  @override
  Future<bool> has(CID cid) async { /* ... */ throw UnimplementedError(); }
  @override
  Future<void> put(MerkleNode<GSet<String>> node) async {
    /* insert node.toBytes() keyed by node.cid unless already present */ }
}

Shut down

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<V, P>: 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<V?>), onCidChanged (Stream<String>), heads, latestKnownCid, currentState.
• GossipSubAnnouncer(host, pubsub, crdtInstanceId, {topic, required writers}): implements merkledag Broadcaster; topic, announcements (Stream<HeadAnnouncement>), broadcast(cidString), close().
• HeadAnnouncement: cid and authenticated author from.
• WriterAuthorizer: .only(peerIds) (a PeerIdAllowlist), .allowAll(), or subclass and override isAllowed(PeerId).
• P2PStreamDagFetcher<P, V>(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<P, V>: the fetch protocol client and server; fetch(cid, peer), fetchNodeFromProvider(cid, addrInfo), close().
• defaultFetchProtocolId: '/merkle-crdt/fetch/2.0.0'.
• BlockStore<P, V> (get, has, put) and InMemoryBlockStore<P, V> (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/<instance id>; 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<V, P> (value, payload), the fetcher and stores are P2PStreamDagFetcher<P, V> / InMemoryBlockStore<P, V> (payload, value). <Set<String>, GSet<String>> for the manager, <GSet<String>, Set<String>> 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:

import 'dart:async';                  // FutureOr
import 'package:test/test.dart';       // fail

Future<void> until(FutureOr<bool> 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<void>.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<String>.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