---
name: merkle-crdt
description: 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.
triggers: merkledag, MerkleCRDT, MerkleClock, MerkleNode, CRDTPayload, GSet, DAGSyncer, Broadcaster, InvalidNodeException, Merkle-CRDT, merkle clock
---

# merkle-crdt (package `merkledag`)

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

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

```dart
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:

```dart
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, `put`s 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:

```dart
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:

```dart
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.

```dart
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

```dart
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:

```dart
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:

```dart
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.

```dart
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 `GSet`s 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