A gRPC network sync layer for crdt, mirrors & inspired by crdt_sync.

  • gRPC-Powered: Synchronizes Crdt nodes in real-time via persistent, bidirectional gRPC streams.
  • BYO Crdt Backend: Bring your own DB, existing support for sqlite_crdt, hive_crdt, postgres_crdt.
  • Multiple Topology Support: Use relayChanges to toggle between hub/chain or mesh mode.
  • Resilient: Auto-reconnect with configurable backoff (uses defaultBackoffStrategy from grpc package), catch-up sync for late joiners and message chunking.
  • Exntensible Pipeline: Custom handshake data, changeset builders, record validators/transformers, per-table event hooks.

Pair with implementations below to have P2P, local-first sync for your app.

Tested and works with:

  • sqlite_crdt
  • hive_crdt
  • postgres_crdt
  • Or create your own custom implementation via CrdtSyncConfig:
    • recordFormat: CrdtRecordFormat.custom
    • mapOutgoingChangeset
    • mapIncomingChangeset

Huge thanks & credits go to cachapa as the author of the crdt engine that powers this.

Warning

  • Co-authored with AI. I created architecture, AI drafted implementation, then artisanally hand refactored.
  • I'm no Dart expert, so there may be language-specific intricacies on handling async processes that were missed.
  • Everything in example/ and test/ was created by AI.

Usage

Minimal server

import 'package:crdt/map_crdt.dart';
import 'package:crdt_grpc_sync/crdt_grpc_sync.dart';

Future<void> main() async {
  final crdt = MapCrdt(['chat']);
  await listen(crdt, 8080, onConnect: (crdtSync, data) {
    print('Peer connected: ${crdtSync.peerId}');
  });
}

Minimal client

import 'package:crdt/map_crdt.dart';
import 'package:crdt_grpc_sync/crdt_grpc_sync.dart';

void main() {
  final crdt = MapCrdt(['chat']);
  final client = CrdtSyncClient(crdt, Uri.parse('grpc://localhost:8080'));
  client.connect();
}

See example/ for working examples.

API

Tip

listen(Crdt crdt, int port, {CrdtSyncServiceConfig config}) — Start gRPC server, returns CrdtSyncServer.

CrdtSyncServer class — Bundles gRPC Server + CrdtSyncService with properties: server, port & method: shutdown().

Customization

All customization via config objects. Pass CrdtSyncConfig / CrdtSyncServiceConfig to listen() or CrdtSyncClient().

// Defaults are as below, null = passthrough
// Shared config (applied to both server and client)
final syncConfig = CrdtSyncConfig(
  changesetBuilder: null,                                // Custom changeset construction
  validateRecord: null,                                  // Reject records before merge
  mapIncomingChangeset: null,                            // Transform after decode
  mapOutgoingChangeset: null,                            // Transform before encode
  recordFormat: null,                                    // Force format (auto-detected if null)
  onChangesetReceived: null,                             // Per-table record count received
  onChangesetSent: null,                                 // Per-table record count sent
  chunkSize: 1000,                                       // Records per gRPC message
  relayChanges: true,                                    // true = hub/chain, false = mesh
  verbose: false,                                        // Log payloads to console
);

  // Server (example: port 8080)
await listen(crdt, 8080, config: CrdtSyncServiceConfig(
  handshakeDataBuilder: null,                            // Build handshake response metadata
  onClientConnecting: null,                              // New handshake request
  onClientConnected: null,                               // Client handshake succeeded
  onClientDisconnected: null,                            // Client disconnected
  syncConfig: CrdtSyncConfig(),                          // Shared config (defaults above)
));

// Client (example: grpc://host:8080)
final client = CrdtSyncClient(
  crdt,
  Uri.parse('grpc://host:8080'),
  handshakeDataBuilder: null,                            // Build handshake metadata
  onConnecting: null,                                    // Connection attempt started
  onConnected: null,                                     // Handshake complete, sync ready
  onDisconnected: null,                                  // Peer disconnected
  onError: null,                                         // Non-fatal sync error
  retryDelayBuilder: null,                               // Custom backoff (null = stop retries)
  maxRetries: null,                                      // Max reconnect (null = infinite)
  channelOptions: null,                                  // Custom gRPC channel config
  syncConfig: CrdtSyncConfig(),                          // Shared config (defaults above)
);

Feature Req and Bugs

File an issue or start a discussion — I'll get around to it when I can!

Libraries

crdt_grpc_sync