merkle-crdt

Skill merkle-crdt. Use when writing Dart code against the merkledag package (import ‘package:merkledag/merkledag.dart’): building a MerkleCRDT or MerkleClock, writing a CRDTPayload or using GSet, implementing DAGSyncer or Broadcaster, encoding or verifying MerkleNode bytes, or testing replicas locally. For libp2p replication use merkle-crdt-sync.

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

What it is

merkledag is a Dart implementation of Merkle-CRDTs (Sanjuán et al., arXiv 2004.00107): a content-addressed Merkle-DAG whose nodes carry state-based CRDT payloads, so replicas that see the same heads converge without coordination. It is transport-agnostic: you supply a DAGSyncer (fetch and store nodes by CID) and a Broadcaster (announce and receive head CIDs). It has no networking of its own.

Use something else when:

• you want replication between dart_libp2p peers (heads over GossipSub, DAG blocks over streams): load the merkle-crdt-sync skill (dart_libp2p_merkle_crdt), which implements both interfaces for you;
• you need generic IPFS blocks or Bitswap: load dart-libp2p-bitswap. The nodes here are not dag-pb (see Gotchas).

Install

dependencies:
  merkledag: ^2.0.0
  dcid: ^1.0.0   # CID is not re-exported; you need it to name or parse CIDs

import 'dart:typed_data';
import 'package:dcid/dcid.dart';            // CID, CID.fromString, CID.fromBytes
import 'package:merkledag/merkledag.dart';  // MerkleCRDT, MerkleClock, MerkleNode, GSet, ...

SDK constraint: ^3.5.0. Runtime deps: crypto, dcid, collection, logging, synchronized.

Core workflow

Minimal in-memory replica

The package does not export test doubles (the MockDAGSyncer and MockBroadcaster used in the README live in test/). Write your own:

import 'dart:async';
import 'package:dcid/dcid.dart';
import 'package:merkledag/merkledag.dart';

class MemorySyncer<T> implements DAGSyncer<T> {
  final Map<CID, MerkleNode<T>> nodes = {};
  @override
  Future<MerkleNode<T>?> get(CID cid) async => nodes[cid];
  @override
  Future<void> put(MerkleNode<T> node) async => nodes[node.cid] = node;
}

class LocalBus implements Broadcaster {
  final _controller = StreamController<String>.broadcast();
  @override
  Future<void> broadcast(String cidString) async => _controller.add(cidString);
  @override
  Stream<String> subscribe() => _controller.stream;
}

Future<void> main() async {
  final crdt = MerkleCRDT<Set<String>, GSet<String>>(
    dagSyncer: MemorySyncer<GSet<String>>(),
    broadcaster: LocalBus(),
  );

  final head = await crdt.add(GSet<String>()..add('apple')..add('banana'));
  await crdt.add(GSet<String>.fromList(['cherry']));

  final state = await crdt.getState();       // GSet<String>? (null when empty)
  print(state?.value);                        // {apple, banana, cherry}
  print(crdt.roots);                          // Set<CID>, normally one head
  print(head);                                // bafy... (CIDv1, base32)

  await crdt.close();                         // cancels the broadcast subscription
}

add(payload) merges payload into the current state, creates one node whose children are all current roots, puts it, makes it the single root, and then broadcasts its CID (outside the lock). If the merged state is byte-identical to the single current head, nothing is created and the existing head is returned without a broadcast.

Two replicas converging

Replicas share nothing but the DAGSyncer (so they can fetch each other’s nodes) and the broadcast channel:

final store = MemorySyncer<GSet<String>>();
final bus = LocalBus();
final a = MerkleCRDT<Set<String>, GSet<String>>(dagSyncer: store, broadcaster: bus);
final b = MerkleCRDT<Set<String>, GSet<String>>(dagSyncer: store, broadcaster: bus);

await a.add(GSet<String>.fromList(['x']));
await b.add(GSet<String>.fromList(['y']));
await Future<void>.delayed(Duration.zero); // let broadcasts be handled

// Each may now hold two concurrent roots; state is already merged.
print((await a.getState())!.value); // {x, y}
await a.add(GSet<String>());        // folds concurrent roots into one head

Feeding heads in yourself

Pass subscribe: false and call merge when you want to authorise, deduplicate or rate-limit heads before they are applied:

final crdt = MerkleCRDT<Set<String>, GSet<String>>(
  dagSyncer: syncer,
  broadcaster: broadcaster,   // still used by add() to announce
  subscribe: false,
);

Future<void> onRemoteHead(String cidString) async {
  final cid = CID.fromString(cidString);       // throws on garbage: catch it
  final changed = await crdt.merge(cid);       // false: already included or unfetchable
  if (changed) refreshUi();
}

onRemoteUpdateProcessed (constructor argument) fires only when a broadcast received through the subscription changed the roots; it does not fire for your own merge calls.

A custom payload

A payload must be a state-based CRDT: merge commutative, associative and idempotent, returning the concrete type, with canonical bytes.

import 'dart:convert';

/// Grow-only max register.
class MaxInt implements CRDTPayload<int> {
  @override
  final int value;
  MaxInt(this.value);

  factory MaxInt.fromBytes(Uint8List b) => MaxInt(int.parse(utf8.decode(b)));

  @override
  MaxInt merge(CRDTPayload<int> other) =>
      MaxInt(value >= other.value ? value : other.value);

  @override
  Uint8List toCanonicalBytes() => Uint8List.fromList(utf8.encode('$value'));

  @override
  String toString() => 'MaxInt($value)';
}

final crdt = MerkleCRDT<int, MaxInt>(dagSyncer: MemorySyncer<MaxInt>(), broadcaster: LocalBus());

Serialising nodes for a real DAGSyncer

final bytes = node.toBytes(); // varint(len) CID bytes, then the hashed body

// On receipt, always decode with verification and the CID you asked for.
final MerkleNode<GSet<String>> received =
    MerkleNode.fromBytes<GSet<String>, Set<String>>(
  bytes,
  GSet<String>.fromCanonicalBytes,
  expectedCid: requestedCid,
); // throws InvalidNodeException on tamper, truncation, trailing or non-canonical bytes

A disk or network syncer sketch:

class BytesSyncer implements DAGSyncer<GSet<String>> {
  BytesSyncer(this.load, this.save);
  final Future<Uint8List?> Function(CID) load;
  final Future<void> Function(CID, Uint8List) save;

  @override
  Future<MerkleNode<GSet<String>>?> get(CID cid) async {
    final bytes = await load(cid);
    if (bytes == null) return null;
    try {
      return MerkleNode.fromBytes<GSet<String>, Set<String>>(
          bytes, GSet<String>.fromCanonicalBytes, expectedCid: cid);
    } on InvalidNodeException {
      return null; // treat a bad block as missing
    }
  }

  @override
  Future<void> put(MerkleNode<GSet<String>> node) => save(node.cid, node.toBytes());
}

Persisting and restoring heads

Roots live only in memory. Store crdt.roots (as strings) yourself and replay them after a restart; merge on an empty replica adopts the head:

await prefs.setStringList('heads', crdt.roots.map((c) => '$c').toList());
// ... after restart, with a DAGSyncer over the same store:
for (final s in prefs.getStringList('heads') ?? const []) {
  await crdt.merge(CID.fromString(s));
}

Merkle-Clock only

MerkleClock<T> is the DAG without CRDT state: an event log whose heads you merge.

final clock = MerkleClock<GSet<String>>(dagSyncer: MemorySyncer<GSet<String>>());
final e1 = await clock.addNode(GSet<String>.fromList(['event-1']));
final changed = await clock.merge(remoteHeadCid);  // bool
final seen = await clock.isIncluded(e1);

Key concepts / API map

• MerkleCRDT<V, P extends CRDTPayload<V>>: replica. add(P) -> Future<CID>, getState() -> Future<P?>, merge(CID) -> Future<bool>, close(), roots, plus the inherited DAG queries below. Constructor: dagSyncer, broadcaster, onRemoteUpdateProcessed, subscribe = true, maxCacheSize = 1000, maxTraversalNodes = 10000, maxMissingNodes = 8.
• MerkleClock<T>: addNode(T) -> Future<CID>, merge(CID) -> Future<bool>; same limits as above, no broadcaster.
• Inherited from the (unexported) AbstractMerkleDAG<T>: roots (copy), getNode(CID) (cache then syncer), isIncluded(CID), isDescendantOf(desc, ancestor) (async, fetches), isDescendant(desc, ancestor) (sync, cache only), isCurrentDAGIncludedIn(CID), mergeHead(CID), setRootsForTesting(Set<CID>).
• MerkleNode<T>: cid, payload, children; MerkleNode.create(payload, children); MerkleNode.computeCid(payload, children); hasValidCid; toBytes(); MerkleNode.fromBytes<P, V>(bytes, factory, {expectedCid}). The plain constructor does not check the CID.
• InvalidNodeException (extends FormatException): thrown by fromBytes.
• CRDTPayload<T>: value, merge(other), toCanonicalBytes(), toString().
• GSet<T>: grow-only set. GSet(), GSet.fromList, GSet.fromJson, GSet.fromCanonicalBytes (String elements only), add, contains, length, isEmpty, isNotEmpty, elements, value, toList, toJson. It is the only payload shipped.
• DAGSyncer<T>: get(CID) -> Future<MerkleNode<T>?>, put(MerkleNode<T>).
• Broadcaster: broadcast(String cidString), subscribe() -> Stream<String>.
• Constants: codecDagPb (0x70), codecRaw (0x55), mhSha2_256 (0x12).

Gotchas

1. The README and example use mocks you cannot import. MockDAGSyncer and MockBroadcaster are in test/ (and example/ imports ../test/merkledag_test.mocks.dart). Write your own in-memory implementations as above.
2. CID comes from dcid, not merkledag. Add dcid to your pubspec and import package:dcid/dcid.dart to name, parse or compare CIDs.
3. State-based only. Every node carries the full merged state, and getState() merges only the current roots’ payloads, never the history. Op-based or delta payloads that need replaying from genesis give wrong results. add() takes a delta and merges it into the state for you.
4. merge must be a real CRDT join. Commutative, associative, idempotent, and must return the concrete P (add and getState cast with as P; a wrong return type is a runtime TypeError). “Last writer wins” by wall clock is not convergent unless ties break deterministically.
5. toCanonicalBytes decides the CID. Equal states must give identical bytes: sort collections, fix encodings, no maps in insertion order, no timestamps unless they are part of the state. fromBytes re-encodes the decoded payload and rejects it if the bytes differ, so the factory must be an exact inverse.
6. GSet.fromCanonicalBytes only supports String elements; other T throws UnsupportedError. GSet sorts and encodes elements by toString(), so elements whose strings collide (1 and '1') are not distinguishable on the wire. GSet.hashCode is the inner set’s identity hash and disagrees with ==; do not use GSets as map keys or set members.
7. 2.0.0 changed every CID. The hashed body is now length-prefixed payload plus sorted, length-prefixed child CIDs. Nodes or stores written by 1.x do not load and must be rebuilt; replicas on 1.x and 2.x cannot sync. Old docs that say merkledag: ^1.0.0 are stale.
8. Not IPFS dag-pb, despite the codec. CIDs are CIDv1, sha2-256, codec 0x70 (dag-pb), but the body is this package’s own varint framing, not a dag-pb protobuf. Do not hand these blocks to IPFS or Bitswap peers expecting to decode them as dag-pb; they can be stored and moved as opaque bytes only.
9. A head that cannot be fetched is dropped, not retried. merge returns false when the remote node is unavailable, and the subscribed handler silently swallows every error (bad CID string, syncer exception). Make your DAGSyncer.get wait or retry for remote nodes, or use subscribe: false and retry yourself; a later broadcast of a newer head also recovers it.
10. Your DAGSyncer must persist what it fetched. merge only calls get; nodes fetched from peers are never put locally. They sit in an LRU cache of maxCacheSize nodes. If get later cannot return a root’s node, getState() skips it (or returns null if it is the first root), and the next add() folds that partial state into the new single head, losing the missing data from view. Back get with a store that keeps every node it has served.
11. Bounded traversal leaves extra roots. Merges walk at most maxTraversalNodes nodes and stop after maxMissingNodes (default 8) unavailable ones. When cut short, the remote head is kept beside the local roots. State stays correct; the next add() folds them into one. Several roots after a merge is normal, not an error.
12. Roots are not persisted. A new MerkleCRDT starts empty (getState() returns null). Save roots and replay them with merge on start-up. Do not use setRootsForTesting in production: it skips all checks.
13. getNode rejects mismatched nodes. A node the syncer returns under a different CID is treated as missing. The plain MerkleNode(...) constructor does not compute the CID; use MerkleNode.create, or check hasValidCid.
14. No-op adds do not broadcast. Adding an element already in the state returns the current head and announces nothing. Peers that missed the earlier broadcast will not learn it from a repeat; re-broadcast crdt.roots periodically from your own code if delivery is lossy.
15. add throws if put throws, and roots stay unchanged. Broadcast failures also surface from add, after the node is already the local head; retry the broadcast, not the add.
16. MerkleClock with non-CRDTPayload payloads hashes toString(), and fromBytes only decodes CRDTPayload types. Use a CRDTPayload if the clock’s nodes must travel as bytes.
17. Call close() when a replica is discarded, or its broadcast subscription keeps it alive and merging.

Testing

• Use in-memory DAGSyncer/Broadcaster classes like those above. Sharing one syncer between replicas models a perfect network; give each replica its own map and copy nodes across to model partitions.
• Broadcast handling is asynchronous. After broadcast, await pumpEventQueue() (from package:test) or Future.delayed(Duration.zero) before asserting on roots or state.
• Prefer subscribe: false plus explicit await crdt.merge(cid) for deterministic tests; merge’s bool tells you whether roots changed.
• To simulate failures, write a syncer that throws, returns null, or hangs on unknown CIDs (see CountingSyncer in test/hardening_test.dart, which also counts get calls to assert traversal cost). Put .timeout(...) on awaits that touch a hanging syncer.
• Assert convergence on (await crdt.getState())!.value, not on roots: equal states can sit under different root sets until the next add.
• Round-trip toBytes/fromBytes with expectedCid for every custom payload, and check that a tampered byte throws InvalidNodeException.
• Run the package’s own suite with dart test in the repo.

Red lines

• Never deserialise untrusted node bytes without MerkleNode.fromBytes (and expectedCid when you asked for a specific CID). Never build nodes from received parts with the unchecked MerkleNode(...) constructor.
• Never merge a CID from the network without catching parse and syncer errors; never let a remote peer drive unbounded traversal by raising maxTraversalNodes or maxMissingNodes without a reason.
• Never change a payload’s toCanonicalBytes encoding in a deployed system: every CID changes and existing replicas and stores stop matching.
• Never mix 1.x and 2.x replicas or stores.
• Never implement merge with randomness, wall-clock reads or I/O.
• Never leave a MerkleCRDT running without close() when you are done.

Reference

• Docs: https://cryptopeer.net/libp2p/merkle-crdt/
• Replication over libp2p: https://cryptopeer.net/libp2p/merkle-crdt-sync/ (skill merkle-crdt-sync)
• Repo: https://github.com/stephanfeb/merkle-crdt (package merkledag, see DEVELOPER-GUIDE.md, CHANGELOG.md, test/hardening_test.dart)
• Paper: https://arxiv.org/abs/2004.00107