Skip to content
cryptopeercryptopeer.net

dart-libp2p-custom-protocol

Skill dart-libp2p-custom-protocol. Use when writing your own libp2p protocol in Dart on a dart_libp2p host: protocol IDs, setStreamHandler, newStream, varint length-prefixed framing, P2PStream read/write, deadlines and timeouts, closeWrite vs close vs reset, versioning, request/response and long-lived streams, go-libp2p interop.

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

What it is

Section titled “What it is”

A libp2p protocol is a protocol ID (a string such as /myapp/echo/1.0.0) plus the bytes two peers exchange on a stream opened for it. The host negotiates the ID with multistream-select, then hands you a P2PStream: an ordered, reliable, flow-controlled byte stream (a Yamux stream inside a Noise session). You define the framing and the message flow.

Use something else for: broadcast to many peers (dart-libp2p-gossipsub), key/value lookup (dart-libp2p-kad-dht), block exchange (dart-libp2p-bitswap). Building the host: dart-libp2p-host.

Install

Section titled “Install”
dependencies:
  dart_libp2p: ^4.12.0
import 'dart:async';
import 'dart:typed_data';
import 'package:dart_libp2p/dart_libp2p.dart' hide Option; // Host, P2PStream, PeerId, Context
import 'package:dart_libp2p/utils/varint.dart';              // encodeVarint, decodeVarint

Core workflow

Section titled “Core workflow”

Echo: the minimal pair

Section titled “Echo: the minimal pair”
const echoId = '/myapp/echo/1.0.0';


// Server: register before host.start() so the first Identify lists it.
host.setStreamHandler(echoId, (P2PStream stream, PeerId remote) async {
  final data = await readAll(stream, maxBytes: 64 * 1024);
  await stream.write(data);
  await stream.closeWrite();      // FIN: tells the client the reply is complete
});


// Client
final s = await host.newStream(peer, [echoId], Context());
await s.write(Uint8List.fromList([1, 2, 3]));
await s.closeWrite();             // half-close: "request done", keep reading
final reply = await readAll(s, maxBytes: 64 * 1024);
await s.close();

read() returns whatever chunk arrived, of any size, and an empty list at end of stream (the peer’s FIN or a reset). Never assume one read() is one message.

/// Reads until EOF, refusing more than [maxBytes].
Future<Uint8List> readAll(P2PStream s, {required int maxBytes}) async {
  final out = BytesBuilder(copy: false);
  while (true) {
    final chunk = await s.read();
    if (chunk.isEmpty) return out.takeBytes();
    if (out.length + chunk.length > maxBytes) {
      await s.reset();
      throw StateError('message too large');
    }
    out.add(chunk);
  }
}

Length-prefixed framing (go-libp2p compatible)

Section titled “Length-prefixed framing (go-libp2p compatible)”

For more than one message per stream, prefix each message with its length as an unsigned varint. This is what go-libp2p’s msgio varint reader and pbio.NewDelimitedReader expect, and what dart_libp2p itself uses for DCUtR, Circuit Relay v2 and AutoNAT v2.

Future<void> writeFrame(P2PStream s, Uint8List msg) async {
  final b = BytesBuilder(copy: false)..add(encodeVarint(msg.length))..add(msg);
  await s.write(b.takeBytes());   // one write: prefix and body together
}


class FrameReader {
  FrameReader(this._s, {this.maxFrame = 1 << 20});
  final P2PStream _s;
  final int maxFrame;
  final _buf = BytesBuilder(copy: false);
  Uint8List _pending = Uint8List(0);
  int _pos = 0;


  int get _available => _pending.length - _pos;


  Future<bool> _fill() async {
    final chunk = await _s.read();
    if (chunk.isEmpty) return false;              // EOF
    _buf..add(Uint8List.sublistView(_pending, _pos))..add(chunk);
    _pending = _buf.takeBytes();
    _pos = 0;
    return true;
  }


  /// Next message, or null at a clean EOF between messages.
  Future<Uint8List?> next() async {
    var len = 0, shift = 0;
    for (var i = 0;; i++) {
      if (_available == 0 && !await _fill()) {
        if (i == 0) return null;
        throw const FormatException('EOF inside length prefix');
      }
      final byte = _pending[_pos++];
      len |= (byte & 0x7f) << shift;
      if (byte & 0x80 == 0) break;
      shift += 7;
      if (i >= 8) throw const FormatException('varint too long');
    }
    if (len > maxFrame) throw FormatException('frame of $len bytes');
    while (_available < len) {
      if (!await _fill()) throw const FormatException('EOF inside frame');
    }
    final msg = Uint8List.sublistView(_pending, _pos, _pos + len);
    _pos += len;
    return Uint8List.fromList(msg);
  }
}

The package also ships BufferedP2PStreamReader (package:dart_libp2p/p2p/protocol/circuitv2/util/buffered_reader.dart) with readVarint(), readExact(n) and isEOF; it throws on EOF mid-read.

For protobuf messages, write msg.writeToBuffer() as the frame body and parse with MyMsg.fromBuffer(frame).

Request/response with a timeout

Section titled “Request/response with a timeout”

One stream per request is the libp2p norm: it is cheap (Yamux) and the stream end delimits the reply.

Future<Uint8List> request(Host host, PeerId peer, Uint8List req) async {
  final s = await host.newStream(peer, ['/myapp/kv/1.0.0'], Context())
      .timeout(const Duration(seconds: 15));
  try {
    await s.setDeadline(DateTime.now().add(const Duration(seconds: 10)));
    await writeFrame(s, req).timeout(const Duration(seconds: 10));
    await s.closeWrite();
    final reply = await FrameReader(s).next().timeout(const Duration(seconds: 10));
    if (reply == null) throw StateError('peer closed without a reply');
    await s.close();
    return reply;
  } catch (_) {
    await s.reset();     // abort: the peer sees a reset, resources are freed
    rethrow;
  }
}


host.setStreamHandler('/myapp/kv/1.0.0', (s, remote) async {
  await s.setDeadline(DateTime.now().add(const Duration(seconds: 10)));
  final req = await FrameReader(s, maxFrame: 64 * 1024).next();
  if (req == null) { await s.close(); return; }
  await writeFrame(s, handle(remote, req));
  await s.closeWrite();
  await s.close();
});

Long-lived stream

Section titled “Long-lived stream”
host.setStreamHandler('/myapp/feed/1.0.0', (s, remote) async {
  final reader = FrameReader(s);
  try {
    while (true) {
      final msg = await reader.next();
      if (msg == null) break;          // peer closed its side
      onMessage(remote, msg);
    }
  } finally {
    await s.close();
  }
});

Without a deadline a Yamux read waits at most 5 minutes, then throws YamuxStreamTimeoutException. On an idle long-lived stream send an application heartbeat frame more often than that, or catch the timeout and keep reading. Watch host.eventBus for EvtPeerConnectednessChanged (notConnected) and reopen the stream when the peer reconnects.

Versioning

Section titled “Versioning”
// Server speaks both; the negotiated ID tells you which.
for (final id in ['/myapp/kv/2.0.0', '/myapp/kv/1.0.0']) {
  host.setStreamHandler(id, (s, remote) async {
    if (s.protocol() == '/myapp/kv/2.0.0') return serveV2(s, remote);
    return serveV1(s, remote);
  });
}


// Client lists preferences in order; multistream picks the first the peer has.
final s = await host.newStream(peer, ['/myapp/kv/2.0.0', '/myapp/kv/1.0.0'], Context());
print(s.protocol());


// Or match a family with one handler:
host.setStreamHandlerMatch('/myapp/kv/', (id) => id.startsWith('/myapp/kv/'), handler);


// Ask the peerstore (filled by Identify) before opening:
final best = await host.peerStore.protoBook
    .firstSupportedProtocol(peer, ['/myapp/kv/2.0.0', '/myapp/kv/1.0.0']);

Bump the ID on any incompatible wire change. Add fields compatibly (protobuf) within a version. host.removeStreamHandler(id) withdraws a protocol and pushes the change to connected peers.

Key concepts / API map

Section titled “Key concepts / API map”
  • Host.setStreamHandler(ProtocolID, StreamHandler); StreamHandler is Future<void> Function(P2PStream stream, PeerId remotePeer).
  • Host.setStreamHandlerMatch(pid, bool Function(ProtocolID) match, handler), Host.removeStreamHandler(pid).
  • Host.newStream(PeerId, List<ProtocolID>, Context): dials if needed, waits for Identify, negotiates (bounded by the negotiation timeout, 10 s default).
  • P2PStream.read([maxLength]): next chunk, at most maxLength bytes; empty at EOF.
  • P2PStream.write(Uint8List): waits for flow-control window; throws StateError after closeWrite().
  • closeWrite(): send FIN, keep reading. closeRead(): stop reading, pending read gets EOF.
  • close(): FIN plus local teardown of both directions; no reads afterwards.
  • reset(): abort both directions (Yamux RST); the peer’s read ends with an error or EOF.
  • setDeadline(DateTime?) (null clears), setReadDeadline, setWriteDeadline.
  • protocol(), id(), conn (remotePeer, remoteMultiaddr), stat() (direction, limited), scope(), isClosed, isWritable.
  • stream.scope().setService('my.service'), reserveMemory(bytes, ReservationPriority.high): attribute resources to your service for the resource manager.
  • encodeVarint / decodeVarint (utils/varint.dart).
  • Exceptions: YamuxStreamTimeoutException, YamuxStreamStateException (p2p/transport/multiplexing/yamux/yamux_exceptions.dart), ResourceLimitExceededException.

Gotchas

Section titled “Gotchas”
  • close() is not half-close. After close() the stream is gone locally, and read() throws. A client that writes a request and calls close() never gets the reply. Use closeWrite(), read to EOF, then close().
  • A server that reads to EOF needs the client’s FIN. If the handler loops until read() is empty, the client must closeWrite(), or both sides wait. go-libp2p and js-libp2p send FIN on a WINDOW_UPDATE frame; this is handled since 3.0.0.
  • One read() is not one message. Chunks split and merge at any byte. Use a length prefix or read to EOF.
  • Always end every stream with close() or reset(), on both sides and on every error path. Default limits allow 68 inbound streams per peer per protocol; leaked streams make the next ones fail with ResourceLimitExceededException. (Before 4.7.1 a stream closed by both sides but not read to EOF kept its scope; stay on a current version.)
  • Write deadlines are weak. setWriteDeadline is stored but not enforced by Yamux streams, and setDeadline is checked only when a write starts. A write blocked on a peer that stopped reading waits. Wrap writes in .timeout(...) and reset() the stream on timeout.
  • Read deadlines: with setDeadline/setReadDeadline, a read past the deadline throws YamuxStreamTimeoutException (or YamuxStreamStateException if the stream died meanwhile). Without one, a read gives up after 5 minutes. MultiplexerConfig.streamReadTimeout is not used by the stream code.
  • Context(timeout: ...) does not bound newStream. Use Future.timeout around the call, or Context().withDialPeerTimeout(d) for the dial part only.
  • A handler that throws resets only its own stream and is logged at WARNING (4.0.1). Still catch your own errors, and reset deliberately.
  • Handlers are not awaited by the muxer: each runs concurrently. Guard shared state, and do not rely on handler order across streams.
  • Unsupported protocol: newStream throws when the peer supports none of the IDs; check the peerstore first if you want a fallback without a round trip. Protocol IDs longer than about 1 KB fail multistream (MessageTooLargeException).
  • Bound every allocation by a size you chose. A varint length comes from the peer; check it against maxFrame before buffering.
  • Relayed connections are limited. A stream over /p2p-circuit shares the relay’s per-connection budget (1 MiB and 1 h on a dart_libp2p relay, 128 KiB and 2 min on a go-libp2p relay). stream.stat().limited tells you. Keep bulk transfers for direct connections (see dart-libp2p-nat-relay).
  • New streams prefer a direct connection to the peer over a relayed one (4.0.0). Do not cache a Conn; call newStream each time.
  • Prefer the documented newStream(peer, pids, Context()). Older docs omit the Context, use stream.stream, or read host.services; none exist.

Testing

Section titled “Testing”

Use two real hosts on loopback (see dart-libp2p-host Testing) and exercise the protocol end to end:

test('kv round trip', () async {
  b.setStreamHandler('/myapp/kv/1.0.0', kvHandler);   // a, b: Hosts from setUp
  final reply = await request(a, b.id, Uint8List.fromList([1]))
      .timeout(const Duration(seconds: 10));
  expect(reply, isNotEmpty);
}, timeout: const Timeout(Duration(seconds: 30)));
  • Test the edges: a peer that closes before replying (next() returns null), a frame over maxFrame, a handler that throws (the client should see EOF or an error, and the next stream must work), many sequential requests (catches leaked streams against the resource limits).
  • Unit-test the framing without a network by feeding FrameReader a fake P2PStream whose read() returns scripted chunks, including splits inside the varint.
  • For go-libp2p interop, use a TCP listen address; the repo’s interop/go-peer shows a Go echo server (/echo/1.0.0) and the Dart side in test/interop/go_interop_host_test.dart.

Red lines

Section titled “Red lines”
  • Never trust a length, count or ID from the peer without a limit.
  • Never leave a stream open on an error path; reset() it.
  • Never change the wire format under an existing protocol ID; add a new version and keep serving the old one while peers migrate.
  • Never use reserved IDs (/ipfs/..., /libp2p/..., /meshsub/...) for your own protocol; namespace it (/myapp/...).
  • Never block a handler on unbounded work while holding the stream; apply a deadline.

Reference

Section titled “Reference”

Built by Stephan Februarygithub.com/stephanfebLinkedIn