CRDT Socket Sync

A comprehensive Dart package for synchronizing Conflict-free Replicated Data Types (CRDTs) between multiple clients and a server.

Overview

CRDT Socket Sync provides a robust, real-time synchronization system that allows multiple clients to collaborate on shared documents without conflicts. Built on top of crdt_lf, this package enables seamless data synchronization with automatic conflict resolution.

Features

• 🔄 Real-time Synchronization: Instant propagation of changes across all connected clients
• 🌐 WebSocket Support: Built-in WebSocket client and server implementations
• 🔧 Conflict Resolution: Automatic conflict-free merge of concurrent operations
• 📦 Compression: Optional data compression for efficient network usage
• 🔌 Modular Architecture: Separate client and server components with clean abstractions
• 📡 Automatic Reconnection: Robust connection handling with automatic retry logic
• 💓 Liveness Detection: Ping/pong tracking detects half-open connections and reconnects
• 🚰 Backpressure: Bounded per-connection send queue drops peers that cannot keep up (they re-sync on reconnect)
• 🗜️ History Pruning: Server takes a snapshot and prunes confirmed history once every client aligns on a common frontier
• 🎯 Type Safety: Full Dart type safety with generic document handlers
• 📊 Event Monitoring: Comprehensive event streams for connection and synchronization monitoring
• 🔌 Plugins: Extendable plugin system for custom functionality
• 📮 Relay Mode: An alternative sync model where the server is a CRDT-agnostic relay — it persists and rebroadcasts opaque change blobs while merging happens entirely on the clients (details)

Built-in Plugins

• 📡 Awareness Plugin: Track the awareness of the clients.

Installation

Add this package to your pubspec.yaml:

dependencies:
  crdt_socket_sync: 
  crdt_lf:

Then run:

dart pub get

Communication modes

The package ships two independent synchronization models. Pick one per deployment — the wire format, type-code convention and the plugin system are shared, but the server responsibilities are very different:

• Server–Client mode (CRDT-aware) — the server holds the documents, applies incoming changes, computes deltas and prunes confirmed history. Use it when you want a single authoritative backend that understands crdt_lf.
• Relay mode — the server is dumb: it persists and rebroadcasts opaque change blobs per room and never parses CRDT data; merging happens entirely on the clients. Use it for cheap horizontal rooms, easy porting to other runtimes (including serverless), and to keep the backend free of crdt_lf.

Cross-cutting concerns (plugins, compression, wire format, connection status) are documented once under Shared topics.

Server–Client mode (CRDT-aware)

In this mode the server owns the documents through a CRDTServerRegistry: it validates handshakes with version vectors, applies incoming changes, broadcasts them to the other clients, and takes snapshots to prune confirmed history.

Quick Start

Server Setup

import 'dart:io';
import 'package:crdt_socket_sync/web_socket_server.dart';

void main() async {
  // Create a server registry to manage documents
  final registry = InMemoryCRDTServerRegistry();

  // Create and start the WebSocket server
  final server = WebSocketServer(
    serverFactory: () => HttpServer.bind('localhost', 8080),
    serverRegistry: registry,
  );

  await server.start();
  print('Server started on localhost:8080');

  // Listen to server events
  server.serverEvents.listen((event) {
    print('Server event: ${event.type} - ${event.message}');
  });
}

Client Setup

import 'package:crdt_socket_sync/web_socket_client.dart';
import 'package:crdt_lf/crdt_lf.dart';

void main() async {
  // Create a CRDT document
  final document = CRDTDocument(peerId: PeerId.generate());

  // Register handlers for different data types
  final listHandler = CRDTListHandler<String>(document, 'shared_list');

  // Create the client
  final client = WebSocketClient(
    url: 'ws://localhost:8080',
    document: document,
    author: document.peerId,
  );

  // Monitor connection status
  client.connectionStatus.listen((status) {
    print('Connection status: $status');
  });

  // Connect to server
  final connected = await client.connect();
  if (connected) {
    print('Connected successfully!');

    // Make changes to the document
    listHandler.insert(0, 'Hello, World!');
  }
}

How it works

Connection & Handshake Phase

📖 Diagrams render best in the live documentation.

Real-time Updates

📖 Diagrams render best in the live documentation.

Out-of-sync Recovery

📖 Diagrams render best in the live documentation.

Server Registry

The server stores documents through a CRDTServerRegistry. The bundled InMemoryCRDTServerRegistry keeps everything in memory (documents are lost on restart); implement the interface to plug in your own persistence backend.

The interface is fully asynchronous:

class CustomServerRegistry implements CRDTServerRegistry {
  final Map<String, CRDTDocument> _documents = {};

  @override
  Future<void> addDocument(String documentId, {PeerId? author}) async {
    _documents[documentId] = CRDTDocument(
      peerId: author ?? PeerId.generate(),
      documentId: documentId,
    );
  }

  @override
  Future<CRDTDocument?> getDocument(String documentId) async =>
      _documents[documentId];

  @override
  Future<bool> hasDocument(String documentId) async =>
      _documents.containsKey(documentId);

  @override
  Future<Set<String>> get documentIds async => _documents.keys.toSet();

  // ... removeDocument, documentCount, createSnapshot, getLatestSnapshot,
  //     applyChange (see the CRDTServerRegistry interface for the full list).
}

applyChange MUST let CausallyNotReadyException propagate: the server relies on it to detect an out-of-sync client and trigger a re-sync. Swallowing it would silently drop the change.

Persisting changes & snapshots

InMemoryCRDTServerRegistry is not durable — documents are lost on restart. To persist a registry, back it with one of the crdt_lf storage adapters. Each exposes a CRDTDocumentStorage with changes and snapshots stores you read/write from inside your CRDTServerRegistry (saveChanges, getChanges, saveSnapshot, getSnapshots, …):

• crdt_lf_hive — Hive-backed storage
• crdt_lf_drift — Drift (SQL) storage
• crdt_lf_sqlite — sqlite3 storage

For a complete server-side implementation, see the example HiveServerRegistry: it lazy-loads documents from Hive on first access, appends each change to storage, and periodically snapshots to compact the change history.

Server Events

Monitor detailed synchronization events on the server:

server.serverEvents.listen((event) {
  switch (event.type) {
    case ServerEventType.clientConnected:
      print('Client ${event.data?['clientId']} connected');
      break;
    case ServerEventType.clientDisconnected:
      print('Client ${event.data?['clientId']} disconnected');
      break;
    case ServerEventType.clientChangeApplied:
      print('Change applied from client');
      break;
    // clientHandshake, clientDocumentStatusCreated, clientOutOfSync,
    // messageBroadcasted, messageSent, snapshotCreated, error, ...
  }
});

Imports

// Basic client interfaces
import 'package:crdt_socket_sync/client.dart';

// WebSocket client implementation
import 'package:crdt_socket_sync/web_socket_client.dart';

// Basic server interfaces
import 'package:crdt_socket_sync/server.dart';

// WebSocket server implementation
import 'package:crdt_socket_sync/web_socket_server.dart';

Relay Mode

The package ships a second, independent sync model next to the CRDT-aware server described above: relay mode. Here the server is a dumb relay — it persists and rebroadcasts opaque change blobs per room and never interprets CRDT data; merging happens entirely on the clients. The room is identified by the document id: one relay server hosts many rooms.

When to use it

The only fundamental difference between the two modes is where the CRDT logic lives:

• Server–Client (CRDT-aware) — the server understands crdt_lf: it applies and merges changes and computes the deltas each client needs.
• Relay — all the CRDT logic runs on the clients; the server only stores the changes opaquely and rebroadcasts them, never parsing CRDT data.

Everything else (delivery guarantees, reconnect policy, initial sync) follows from that choice. Pick relay when you want a CRDT-agnostic backend (no crdt_lf on the server, easy to port to other runtimes); pick server–client when a single authoritative backend should understand and merge the documents.

Relay Quick Start

// Server
import 'dart:io';
import 'package:crdt_socket_sync/web_socket_relay_server.dart';

void main() async {
  final server = WebSocketRelayServer(
    serverFactory: () => HttpServer.bind('localhost', 8080),
    // store: defaults to InMemoryRelayStore()
  );
  await server.start();
}

// Client
import 'package:crdt_lf/crdt_lf.dart';
import 'package:crdt_socket_sync/web_socket_relay_client.dart';

void main() async {
  final document = CRDTDocument(
    peerId: PeerId.generate(),
    documentId: 'my-room',
  );
  final text = CRDTFugueTextHandler(document, 'content');

  final client = WebSocketRelayClient(
    url: 'ws://localhost:8080',
    document: document,
    author: document.peerId,
  );

  await client.connect();
  text.insert(0, 'Hello, relay!'); // queued, pushed, acked, rebroadcast
}

Local edits are delivered at-least-once: a change leaves the client queue only when the relay acknowledges it, and unacked changes survive reconnects (re-delivery is harmless because peers de-duplicate imported changes). Persistence is pluggable through the RelayStore interface (InMemoryRelayStore by default).

Join & Welcome

📖 Diagrams render best in the live documentation.

Push, Ack & Rebroadcast

📖 Diagrams render best in the live documentation.

Log Compaction

The room log grows with every push. Past a threshold (RelayProtocol.logCompactThreshold), the relay sets compact: true on an ack or welcome, asking that one client (rate-limited per room) to upload a snapshot. The relay stores the snapshot and deletes the covered log entries; late joiners then bootstrap from snapshot plus residual log.

📖 Diagrams render best in the live documentation.

The client caps upToSeq at the highest contiguous sequence it has imported, so a snapshot can never silently drop a concurrent change that the relay logged but this client has not received yet.

Client seq window

The relay stamps every change blob with a monotonic seq. Each client keeps a window over the room log: maxContiguous — the highest seq S such that it has imported every entry in [1, S] — plus any detached ranges above it, the holes left when concurrent clients' rebroadcasts arrive out of order. Three inbound frames feed the window — a welcome (the whole prefix), an ack of your own push, and rebroadcast changes — and whenever a hole fills, the contiguous frontier advances.

maxContiguous is the only thing that bounds a snapshot upload: a compaction request is capped at it, so the client can never claim to cover a seq it has not imported (which would let the relay delete a change no snapshot holds). The seq window is pure relay bookkeeping — it does not decide what the document shows; that is the CRDT's own causal job (importChanges applies the causally-ready subset and skips the rest).

📖 Diagrams render best in the live documentation.

Relay Plugins & Awareness

The relay client and server are regular CRDTSocketClient / CRDTSocketServer implementations, so the plugin system works unchanged — including the awareness plugin, since presence only needs sessions and broadcasting, never CRDT parsing:

final server = WebSocketRelayServer(
  serverFactory: () => HttpServer.bind('localhost', 8080),
  plugins: [ServerAwarenessPlugin()],
);

final client = WebSocketRelayClient(
  url: 'ws://localhost:8080',
  document: document,
  author: document.peerId,
  plugins: [ClientAwarenessPlugin()],
);

Awareness state is ephemeral: it is rebroadcast to the room but never persisted in the RelayStore.

Implementing a relay server

WebSocketRelayServer is only one implementation of the relay contract. Because the server never parses CRDT data, the relay side is just a JSON message contract over a WebSocket and can be implemented on any runtime — including serverless hosts such as Cloudflare Workers + Durable Objects.

The greyhound_markdown app is a concrete example: its client uses WebSocketRelayClient from this library, while its server is currently a Cloudflare Worker in TypeScript (server/, run with wrangler) rather than the Dart WebSocketRelayServer — a TypeScript reference implementation of the relay contract below.

Frames are JSON envelopes typed by an integer type code, with documentId identifying the room. CRDT change/snapshot payloads travel as opaque base64 strings. Your server must handle:

Code	Name	Dir	Fields (beyond type, documentId)
20	hello	C→S	author
21	welcome	S→C	sessionId, snapshot: string|null, changes: string[], seq, logLength, compact
22	push	C→S	changes: string[]
23	ack	S→C	seq, count, logLength, compact
24	changes	S→others	changes: string[], seq, from: string|null
25	snapshotUpload	C→S	snapshot, upToSeq
26	stateRequest	C→S	— (reply with a welcome, same sessionId)
5	ping	C→S	timestamp → reply pong (6) {originalTimestamp, responseTimestamp}
7	error	S→C	code, message
100–102	awareness	C↔S	presence passthrough (store + rebroadcast)

Server responsibilities: assign a sessionId per connection and return it in the welcome; append pushed blobs to a per-room log with monotonic sequence numbers; put the last assigned seq on both the ack and the rebroadcast changes (clients rely on it to bound their snapshot uploads); serve snapshot + log on hello/stateRequest; on snapshotUpload, store the snapshot and truncate the log up to upToSeq; drive compaction by setting compact: true (rate-limited, one client per room) once the log passes your threshold.

Two hard requirements:

• Reply to every ping with a pong. The client treats a missing pong within its ping timeout (default 30s) as a dead connection and reconnects.
• Send the welcome within the client's handshake timeout (default 5s; injectable via WebSocketRelayClient(handshakeTimeout: ...) for hosts with cold starts).

Also accept binary inbound frames: the Dart client sends UTF-8 JSON as binary WebSocket frames (decode them as text). Replies may be text frames. With the default NoCompression, no compression handling is needed. To keep serverless costs down you can lengthen pingInterval/pingTimeout on the client so idle rooms are probed less often.

Relay Imports

// Relay client interfaces
import 'package:crdt_socket_sync/relay_client.dart';

// WebSocket relay client implementation
import 'package:crdt_socket_sync/web_socket_relay_client.dart';

// Relay server interfaces (RelayStore, compaction, session)
import 'package:crdt_socket_sync/relay_server.dart';

// WebSocket relay server implementation
import 'package:crdt_socket_sync/web_socket_relay_server.dart';

Shared topics

The following concerns work the same way in both communication modes.

Plugins

The package provides a plugin system that allows you to extend the functionality of the client and the server. A plugin can be only on the client or only on the server or both.

// it's important that `MyClientPlugin` extends `ClientSyncPlugin` (not implements).
// `ClientSyncPlugin` makes some "magic" to make the plugin work.
class MyClientPlugin extends ClientSyncPlugin {
  @override
  void onMessage(Message message) {
    print('message: $message');
  }
}

// it's important that `MyServerPlugin` extends `ServerSyncPlugin` (not implements).
// `ServerSyncPlugin` makes some "magic" to make the plugin work.
class MyServerPlugin extends ServerSyncPlugin {
  @override
  void onMessage(Message message) {
    print('message: $message');
  }
}

final client = WebSocketClient(
  url: 'ws://localhost:8080',
  document: document,
  author: document.peerId,
  plugins: [MyClientPlugin()],
);

final server = WebSocketServer(
  serverFactory: () => HttpServer.bind('localhost', 8080),
  serverRegistry: registry,
  plugins: [MyServerPlugin()],
);

A plugin can send new message types to clients and the server. To do so, it must extend the Message class and implement the fromJson method to decode the messages.

The same plugins work on the relay pair (WebSocketRelayClient / WebSocketRelayServer) unchanged — see Relay Plugins & Awareness.

Awareness Plugin

The awareness plugin is a plugin that allows you to track the awareness of the clients. It is a plugin that is both on the client and the server. It is used to track the awareness of the clients and to send the awareness to the server and to the clients.

The example provided uses the awareness plugin to track the active users (name, surname, random color) and their relative position in the document.

Compression

Compressor is an injectable interface (compress/decompress over List<int>) available on both clients and both servers. The default is NoCompression. The package intentionally ships no third-party compressor, so you can plug in whatever fits your platform and bandwidth needs.

⚠️ Compression is symmetric. If a server injects a compressor, every client connecting to it must inject the same compressor — otherwise the peer cannot decode incoming messages.

A real, cross-platform gzip compressor is only a few lines on top of the pure-Dart, synchronous package:archive (works on the Dart VM and Flutter/web, unlike dart:io's GZipCodec):

import 'package:archive/archive.dart';

class GzipCompression implements Compressor {
  const GzipCompression();

  @override
  List<int> compress(List<int> data) => GZipEncoder().encode(data)!;

  @override
  List<int> decompress(List<int> data) => GZipDecoder().decodeBytes(data);
}

Inject it on both sides:

// Server with compression
final server = WebSocketServer(
  serverFactory: () => HttpServer.bind('localhost', 8080),
  serverRegistry: registry,
  compressor: const GzipCompression(),
);

// Client with the matching compressor
final client = WebSocketClient(
  url: 'ws://localhost:8080',
  document: document,
  author: author,
  compressor: const GzipCompression(),
);

See the full working implementation and how it's wired into the demo in example/lib/src/gzip_compression.dart (mirrored in the Flutter client at client_example/lib/gzip_compression.dart). To try it end-to-end, start the server with --compress and build the client with --dart-define=USE_COMPRESSION=true:

# server (from the example/ directory)
dart run lib/main.dart --compress

# Flutter client (from the client_example/ directory)
flutter run --dart-define=USE_COMPRESSION=true

Connection status & error handling

Both clients expose a connectionStatus stream with automatic recovery:

client.connectionStatus.listen((status) {
  switch (status) {
    case ConnectionStatus.connected:
      // Normal operation
      break;
    case ConnectionStatus.reconnecting:
      // Automatic reconnection in progress
      break;
    case ConnectionStatus.error:
      // Handle connection errors
      break;
    case ConnectionStatus.disconnected:
      // Clean disconnection
      break;
  }
});

Liveness is tracked with ping/pong: a missing pong within the ping timeout is treated as a dead (half-open) connection and triggers a reconnect. The CRDT-aware client reconnects on a fixed interval with capped attempts; the relay client uses exponential backoff with jitter and retries forever by default.

Wire format & type codes

Messages are exchanged as JSON envelopes (typed by an integer type code) where the heavy CRDT payloads are carried as base64-encoded binary blobs produced by crdt_lf's native binary methods:

Field	Encoding
Change	base64(change.toBytes())
Snapshot	base64(snapshot.toBytes())
VersionVector	base64(versionVector.toBytes())

On the receiver side, the corresponding fromBytes factories rebuild the objects. Operation payloads inside a Change are already compact binary blobs produced by the handler's ValueCodec<T>, so the wire payload is independent from the in-memory representation of your custom value types.

Message type codes follow a convention: 0-19 core protocol, 20-39 relay protocol, 100+ plugins (the awareness plugin uses 100-102). The base Message carries the frames shared by every mode (ping/pong/error); the CRDT-aware sync frames live under SyncMessage and the relay frames under RelayMessage. In relay messages the blobs stay opaque strings end-to-end: only relay clients decode them.

Requires crdt_lf ^3.0.0, which provides the schema-versioned Change / Snapshot binary format.

Examples

This package provides two examples:

• example/ — a persistent (Hive-backed) WebSocket server for the CRDT-aware server–client mode.
• client_example/ — the Flutter client counterpart, which connects to the server and shows the same examples as crdt_lf, live over the real backend.

Run the server and a client. The workspace contains a .vscode folder with launch settings to run both server and client.

The server example and the flutter example already use the awareness plugin.

For a full relay mode application (relay client from this library + a Cloudflare Worker relay server), see the greyhound_markdown app.

Apps

• greyhound_markdown — Real-time collaborative markdown editor built on crdt_lf, using relay mode (relay client from this library; Cloudflare Worker server).

Roadmap

A roadmap is available in the project page. The roadmap provides a high-level overview of the project's goals and the current status of the project.

Contributing

Contributions are welcome! Please read the contributing guidelines and submit pull requests to the main repository.

Packages

Other bricks of the crdt "system" are:

• crdt_lf
• crdt_lf_flutter
• hlc_dart
• crdt_lf_hive
• crdt_lf_drift
• crdt_lf_sqlite