From 4b66703280cffb0e757e9b7aa87cc0ab77050a16 Mon Sep 17 00:00:00 2001 From: Krzysztof Rudnicki Date: Sun, 5 Jul 2026 20:48:07 +0200 Subject: [PATCH 1/2] Initial implementation: unified CRDT merge scheme + GitHub sync transport Pure-Dart mirror of the Python crdt-sync package, module-for-module and test-for-test, so the merge algorithm canonically agrees on both sides of a sync between a Dart client and a Python one. No Flutter SDK dependency (package:http only) per Workstream D's scoping decision (2026-07-04). 59 tests, clean dart analyze. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_017McuA1nAi6jBT1EaS5BrVf --- crdt_sync_dart/.gitignore | 6 + crdt_sync_dart/LICENSE | 21 +++ crdt_sync_dart/README.md | 75 ++++++++ crdt_sync_dart/analysis_options.yaml | 11 ++ crdt_sync_dart/lib/crdt_sync.dart | 8 + crdt_sync_dart/lib/src/github_client.dart | 171 +++++++++++++++++ crdt_sync_dart/lib/src/hlc.dart | 102 ++++++++++ crdt_sync_dart/lib/src/log.dart | 20 ++ crdt_sync_dart/lib/src/record.dart | 142 ++++++++++++++ crdt_sync_dart/lib/src/sync.dart | 48 +++++ crdt_sync_dart/pubspec.yaml | 16 ++ crdt_sync_dart/test/github_client_test.dart | 195 ++++++++++++++++++++ crdt_sync_dart/test/hlc_test.dart | 117 ++++++++++++ crdt_sync_dart/test/log_test.dart | 85 +++++++++ crdt_sync_dart/test/record_test.dart | 141 ++++++++++++++ crdt_sync_dart/test/sync_test.dart | 192 +++++++++++++++++++ 16 files changed, 1350 insertions(+) create mode 100644 crdt_sync_dart/.gitignore create mode 100644 crdt_sync_dart/LICENSE create mode 100644 crdt_sync_dart/README.md create mode 100644 crdt_sync_dart/analysis_options.yaml create mode 100644 crdt_sync_dart/lib/crdt_sync.dart create mode 100644 crdt_sync_dart/lib/src/github_client.dart create mode 100644 crdt_sync_dart/lib/src/hlc.dart create mode 100644 crdt_sync_dart/lib/src/log.dart create mode 100644 crdt_sync_dart/lib/src/record.dart create mode 100644 crdt_sync_dart/lib/src/sync.dart create mode 100644 crdt_sync_dart/pubspec.yaml create mode 100644 crdt_sync_dart/test/github_client_test.dart create mode 100644 crdt_sync_dart/test/hlc_test.dart create mode 100644 crdt_sync_dart/test/log_test.dart create mode 100644 crdt_sync_dart/test/record_test.dart create mode 100644 crdt_sync_dart/test/sync_test.dart diff --git a/crdt_sync_dart/.gitignore b/crdt_sync_dart/.gitignore new file mode 100644 index 0000000..08e2419 --- /dev/null +++ b/crdt_sync_dart/.gitignore @@ -0,0 +1,6 @@ +.dart_tool/ +.packages +build/ +pubspec.lock +coverage/ +*.iml diff --git a/crdt_sync_dart/LICENSE b/crdt_sync_dart/LICENSE new file mode 100644 index 0000000..d1aa637 --- /dev/null +++ b/crdt_sync_dart/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2026 Krzysztof Rudnicki + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/crdt_sync_dart/README.md b/crdt_sync_dart/README.md new file mode 100644 index 0000000..a466940 --- /dev/null +++ b/crdt_sync_dart/README.md @@ -0,0 +1,75 @@ +# crdt_sync + +Shared CRDT merge scheme + GitHub-Contents-API sync transport, extracted for +reuse across four personal apps that each need cross-device sync: `todo`, +`diet_guard`'s Flutter app, `wake_alarm`'s `phone_app`, and screen-locker's +`workout_app`. Pure Dart -- no Flutter SDK dependency (`package:http` only) +-- so it works in both Flutter apps and plain Dart tooling. Mirrors the +[Python `crdt-sync`](https://github.com/kuhyx/crdt-sync) package +module-for-module and test-for-test, so the merge algorithm canonically +agrees on both sides of any sync between a Dart client and a Python one. + +## The merge scheme + +See the Python package's README for the full rationale; in short, a generic +**LWW-map-with-sticky-remove**: + +- `Hlc` -- a Hybrid Logical Clock `(wallTimeMs, counter, nodeId)`, totally + ordered, monotonic even across clock skew. +- `Record` -- an id, a per-field last-writer-wins map (`fieldName -> + (value, Hlc)`), and a sticky `deleted` flag (`deleted = a.deleted || + b.deleted`, never the other way -- a delete can never be silently undone). +- `Log` -- `Map`, merged by taking the union of ids and + merging shared ones. + +All three merges are commutative and idempotent -- see `test/log_test.dart` +for the concrete convergence-property tests. + +## Install + +Pin to a tag once this is pushed: + +```yaml +dependencies: + crdt_sync: + git: + url: https://github.com/kuhyx/crdt_sync_dart + ref: v0.1.0 +``` + +## Usage + +```dart +import 'package:crdt_sync/crdt_sync.dart'; + +final nodeId = 'phone'; +final clock = Hlc.newTick(nodeId); +final record = Record( + id: 'abc123', + fields: {'text': ('buy milk', clock)}, +); + +final client = GitHubClient(owner: 'kuhyx', repo: 'my-app-sync', token: token); +final merged = await syncLog( + client: client, + deviceId: nodeId, + pathPrefix: 'devices', + localLog: {record.id: record}, + encode: (log) => jsonEncode(log.map((k, r) => MapEntry(k, r.toJson()))), + decode: (text) => (jsonDecode(text) as Map).map( + (k, v) => MapEntry(k, Record.fromJson(v as Map)), + ), +); +``` + +`syncLog` pulls every other device's last-pushed log from +`//...`, merges each into the local log with +`mergeLogs`, then pushes this device's own merged result back up. + +## Development + +```bash +dart pub get +dart analyze +dart test +``` diff --git a/crdt_sync_dart/analysis_options.yaml b/crdt_sync_dart/analysis_options.yaml new file mode 100644 index 0000000..e9907b0 --- /dev/null +++ b/crdt_sync_dart/analysis_options.yaml @@ -0,0 +1,11 @@ +# This file configures the analyzer for this pure-Dart package (no Flutter +# dependency, so `package:lints` rather than `package:flutter_lints`). + +include: package:lints/recommended.yaml + +linter: + rules: + - prefer_single_quotes + - always_declare_return_types + - avoid_dynamic_calls + - unawaited_futures diff --git a/crdt_sync_dart/lib/crdt_sync.dart b/crdt_sync_dart/lib/crdt_sync.dart new file mode 100644 index 0000000..ed94c6d --- /dev/null +++ b/crdt_sync_dart/lib/crdt_sync.dart @@ -0,0 +1,8 @@ +/// Shared CRDT merge scheme + GitHub-Contents-API sync transport. +library; + +export 'src/github_client.dart'; +export 'src/hlc.dart'; +export 'src/log.dart'; +export 'src/record.dart'; +export 'src/sync.dart'; diff --git a/crdt_sync_dart/lib/src/github_client.dart b/crdt_sync_dart/lib/src/github_client.dart new file mode 100644 index 0000000..c30df5b --- /dev/null +++ b/crdt_sync_dart/lib/src/github_client.dart @@ -0,0 +1,171 @@ +import 'dart:convert'; + +import 'package:http/http.dart' as http; + +const _apiBase = 'https://api.github.com'; +const _httpNotFound = 404; + +/// Raised for a GitHub API failure the caller must not silently ignore. +class GitHubSyncError implements Exception { + GitHubSyncError(this.message); + + final String message; + + @override + String toString() => 'GitHubSyncError: $message'; +} + +/// Raised when the configured repo itself is unreachable. +/// +/// Distinguished from a path-404 (nothing pushed to that path yet, which is +/// benign -- it just means no other device has synced before) so the caller +/// can tell "the repo name is wrong or the token isn't scoped to it" apart +/// from "no other device has synced yet". +class RepoNotFoundError extends GitHubSyncError { + RepoNotFoundError(super.message); +} + +/// Minimal GitHub Contents API client used as dumb file storage. +/// +/// GitHub is used purely as file storage via the REST Contents API, not a +/// git clone -- there is no working tree and no git-level merge; the only +/// merge is the domain-level one in `mergeLogs`. Mirrors +/// `crdt_sync._github.GitHubSyncClient` on the Python side: unlike `todo`'s +/// original `GitHubClient`, [listDirectory] returns every entry name +/// regardless of type (not just files), because `crdt_sync`'s +/// `//` layout needs to discover device +/// *subdirectories*; and [putFileText] resolves its own current SHA rather +/// than requiring the caller to track it. +class GitHubClient { + GitHubClient({ + required this.owner, + required this.repo, + required String token, + http.Client? httpClient, + }) // Dart forbids private named params, so this can't be an initializing + // formal; assign it explicitly. + // ignore: prefer_initializing_formals + : _token = token, + _http = httpClient ?? http.Client(); + + final String owner; + final String repo; + final String _token; + final http.Client _http; + + Map get _headers => { + 'Authorization': 'Bearer $_token', + 'Accept': 'application/vnd.github+json', + }; + + Uri _contentsUri(String path) => + Uri.parse('$_apiBase/repos/$owner/$repo/contents/$path'); + + Future _get(String path) async { + try { + return await _http.get(_contentsUri(path), headers: _headers); + } on http.ClientException catch (exc) { + throw GitHubSyncError('network error reading $path: $exc'); + } + } + + Future _repoExists() async { + try { + final res = await _http.get( + Uri.parse('$_apiBase/repos/$owner/$repo'), + headers: _headers, + ); + return res.statusCode >= 200 && res.statusCode < 300; + } on http.ClientException { + return false; + } + } + + Future _raiseForMissingPath(String path) async { + if (!await _repoExists()) { + throw RepoNotFoundError( + '$owner/$repo not found, private without access, or the token ' + 'lacks contents permission (while reading $path)', + ); + } + } + + /// Returns the entry names directly under [path] (empty if unused). + /// + /// Includes both files and subdirectories -- see the class doc for why + /// this differs from a plain "list files" helper. + Future> listDirectory(String path) async { + final res = await _get(path); + if (res.statusCode == _httpNotFound) { + await _raiseForMissingPath(path); + return []; + } + if (res.statusCode < 200 || res.statusCode >= 300) { + throw GitHubSyncError('GET $path (list) failed: ${res.statusCode}'); + } + final decoded = jsonDecode(res.body); + if (decoded is! List) return []; + return decoded + .whereType>() + .map((e) => e['name']) + .whereType() + .toList(); + } + + /// Returns the decoded text content at [path], or null if unused. + Future getFileText(String path) async { + final res = await _get(path); + if (res.statusCode == _httpNotFound) { + await _raiseForMissingPath(path); + return null; + } + if (res.statusCode < 200 || res.statusCode >= 300) { + throw GitHubSyncError('GET $path failed: ${res.statusCode}'); + } + final decoded = jsonDecode(res.body) as Map; + final content = (decoded['content'] as String? ?? '').replaceAll('\n', ''); + return utf8.decode(base64.decode(content)); + } + + Future _existingSha(String path) async { + final res = await _get(path); + if (res.statusCode == _httpNotFound) { + await _raiseForMissingPath(path); + return null; + } + if (res.statusCode < 200 || res.statusCode >= 300) { + throw GitHubSyncError('GET $path (for sha) failed: ${res.statusCode}'); + } + final decoded = jsonDecode(res.body) as Map; + return decoded['sha'] as String?; + } + + /// Creates or updates the file at [path] with [text]. + Future putFileText( + String path, + String text, { + required String message, + }) async { + final sha = await _existingSha(path); + final body = { + 'message': message, + 'content': base64.encode(utf8.encode(text)), + 'sha': ?sha, + }; + late final http.Response res; + try { + res = await _http.put( + _contentsUri(path), + headers: _headers, + body: jsonEncode(body), + ); + } on http.ClientException catch (exc) { + throw GitHubSyncError('network error pushing $path: $exc'); + } + if (res.statusCode < 200 || res.statusCode >= 300) { + throw GitHubSyncError('PUT $path failed: ${res.statusCode}'); + } + } + + void close() => _http.close(); +} diff --git a/crdt_sync_dart/lib/src/hlc.dart b/crdt_sync_dart/lib/src/hlc.dart new file mode 100644 index 0000000..aef2fc4 --- /dev/null +++ b/crdt_sync_dart/lib/src/hlc.dart @@ -0,0 +1,102 @@ +/// A Hybrid Logical Clock: a totally-ordered, monotonic per-node timestamp. +/// +/// Combines wall-clock time with a logical counter so two ticks issued by +/// the same node in the same millisecond still get a strict order, and two +/// ticks from different nodes are always comparable (the node id breaks +/// ties), without requiring synchronized clocks across devices. Mirrors +/// `crdt_sync._hlc.Hlc` on the Python side field-for-field so the two +/// serialize to and parse the same wire format. +class Hlc implements Comparable { + const Hlc({ + required this.wallTimeMs, + required this.counter, + required this.nodeId, + }); + + final int wallTimeMs; + final int counter; + final String nodeId; + + static const _isoPrefixLen = 23; // "YYYY-MM-DDTHH:MM:SS.mmm" + + /// Returns the next clock value for [nodeId]. + /// + /// Passing [previous] (this node's last-issued clock) is what makes the + /// clock monotonic even when the wall clock hasn't advanced (or has gone + /// backwards). [wallTimeMsOverride] is exposed for deterministic tests. + static Hlc newTick(String nodeId, {Hlc? previous, int? wallTimeMsOverride}) { + final now = + wallTimeMsOverride ?? DateTime.now().toUtc().millisecondsSinceEpoch; + if (previous == null) { + return Hlc(wallTimeMs: now, counter: 0, nodeId: nodeId); + } + final newWall = now > previous.wallTimeMs ? now : previous.wallTimeMs; + final counter = newWall == previous.wallTimeMs ? previous.counter + 1 : 0; + return Hlc(wallTimeMs: newWall, counter: counter, nodeId: nodeId); + } + + /// Serializes to a lexicographically sortable string: + /// `Z--`. + String toStr() { + final dt = DateTime.fromMillisecondsSinceEpoch(wallTimeMs, isUtc: true); + String two(int n) => n.toString().padLeft(2, '0'); + final iso = + '${dt.year.toString().padLeft(4, '0')}-${two(dt.month)}-${two(dt.day)}' + 'T${two(dt.hour)}:${two(dt.minute)}:${two(dt.second)}' + '.${(wallTimeMs % 1000).toString().padLeft(3, '0')}'; + final counterHex = counter.toRadixString(16).padLeft(4, '0'); + return '${iso}Z-$counterHex-$nodeId'; + } + + /// Parses the format produced by [toStr]. Throws [FormatException]. + static Hlc fromStr(String text) { + final zIndex = text.indexOf('Z-'); + if (zIndex == -1 || zIndex != _isoPrefixLen) { + throw FormatException('not a valid Hlc string', text); + } + final isoPart = text.substring(0, zIndex); + final rest = text.substring(zIndex + 2); + final dashIndex = rest.indexOf('-'); + if (dashIndex == -1) { + throw FormatException('not a valid Hlc string', text); + } + final counterHex = rest.substring(0, dashIndex); + final nodeId = rest.substring(dashIndex + 1); + final dt = DateTime.parse('${isoPart}Z'); + return Hlc( + wallTimeMs: dt.millisecondsSinceEpoch, + counter: int.parse(counterHex, radix: 16), + nodeId: nodeId, + ); + } + + @override + int compareTo(Hlc other) { + if (wallTimeMs != other.wallTimeMs) { + return wallTimeMs.compareTo(other.wallTimeMs); + } + if (counter != other.counter) { + return counter.compareTo(other.counter); + } + return nodeId.compareTo(other.nodeId); + } + + bool operator >(Hlc other) => compareTo(other) > 0; + bool operator >=(Hlc other) => compareTo(other) >= 0; + bool operator <(Hlc other) => compareTo(other) < 0; + bool operator <=(Hlc other) => compareTo(other) <= 0; + + @override + bool operator ==(Object other) => + other is Hlc && + wallTimeMs == other.wallTimeMs && + counter == other.counter && + nodeId == other.nodeId; + + @override + int get hashCode => Object.hash(wallTimeMs, counter, nodeId); + + @override + String toString() => + 'Hlc(wallTimeMs: $wallTimeMs, counter: $counter, nodeId: $nodeId)'; +} diff --git a/crdt_sync_dart/lib/src/log.dart b/crdt_sync_dart/lib/src/log.dart new file mode 100644 index 0000000..993e416 --- /dev/null +++ b/crdt_sync_dart/lib/src/log.dart @@ -0,0 +1,20 @@ +import 'record.dart'; + +/// A collection of [Record]s keyed by id. +typedef Log = Map; + +/// Returns the union of [local] and [remote], merging shared ids. +/// +/// Commutative and idempotent: `mergeLogs(a, b) == mergeLogs(b, a)` and +/// `mergeLogs(x, x) == x`, so pull order between devices never matters and a +/// repeated sync tick is a no-op. +Log mergeLogs(Log local, Log remote) { + final merged = Map.from(local); + for (final entry in remote.entries) { + final existing = merged[entry.key]; + merged[entry.key] = existing == null + ? entry.value + : mergeRecord(existing, entry.value); + } + return merged; +} diff --git a/crdt_sync_dart/lib/src/record.dart b/crdt_sync_dart/lib/src/record.dart new file mode 100644 index 0000000..5b31756 --- /dev/null +++ b/crdt_sync_dart/lib/src/record.dart @@ -0,0 +1,142 @@ +import 'hlc.dart'; + +/// A field value paired with the clock it was written at. +typedef Field = (Object?, Hlc); + +/// One CRDT record: an id, per-field LWW values, and a sticky tombstone. +/// +/// Mirrors `crdt_sync.Record` on the Python side. Deliberately +/// domain-agnostic -- translating an app's own model (a note, a food-log +/// entry, a workout entry, a wake-time value) to and from [Record] is each +/// app's job, not this library's. +class Record { + const Record({ + required this.id, + required this.fields, + this.deleted = false, + this.deletedHlc, + }); + + /// Stable identifier, unique within one `Log`. + final String id; + + /// Field name -> `(value, Hlc)`. Concurrent edits to the same field + /// converge to whichever side has the greater [Hlc]. + final Map fields; + + /// Whether this record is tombstoned. Deletion is monotonic: once true on + /// either side of a merge, it stays true. + final bool deleted; + + /// The clock value of the delete, if any -- informational/ordering only; + /// it does not participate in the merge decision (see [mergeRecord]). + final Hlc? deletedHlc; + + Map toJson() => { + 'id': id, + 'fields': fields.map( + (name, field) => MapEntry(name, [field.$1, field.$2.toStr()]), + ), + 'deleted': deleted, + 'deletedHlc': deletedHlc?.toStr(), + }; + + factory Record.fromJson(Map json) { + final rawFields = json['fields'] as Map; + final fields = rawFields.map((name, value) { + final pair = value as List; + return MapEntry(name, (pair[0], Hlc.fromStr(pair[1] as String))); + }); + final deletedHlcStr = json['deletedHlc'] as String?; + return Record( + id: json['id'] as String, + fields: fields, + deleted: json['deleted'] as bool, + deletedHlc: deletedHlcStr == null ? null : Hlc.fromStr(deletedHlcStr), + ); + } + + @override + bool operator ==(Object other) { + if (other is! Record) return false; + return id == other.id && + deleted == other.deleted && + deletedHlc == other.deletedHlc && + _fieldsEqual(fields, other.fields); + } + + static bool _fieldsEqual(Map a, Map b) { + if (a.length != b.length) return false; + for (final entry in a.entries) { + if (b[entry.key] != entry.value) return false; + } + return true; + } + + @override + int get hashCode => Object.hash( + id, + deleted, + deletedHlc, + Object.hashAllUnordered( + fields.entries.map((e) => Object.hash(e.key, e.value)), + ), + ); + + @override + String toString() => + 'Record(id: $id, fields: $fields, deleted: $deleted, ' + 'deletedHlc: $deletedHlc)'; +} + +/// Returns whichever of [a], [b] has the greater [Hlc]. +/// +/// Ties (equal [Hlc]) keep [a]; two ticks never compare equal across +/// different nodes, so a tie only happens when both sides are literally the +/// same tick re-merged, in which case the value is expected to be identical +/// anyway. +Field mergeField(Field a, Field b) => a.$2 >= b.$2 ? a : b; + +Hlc? _mergeDeletedHlc(Record a, Record b) { + if (a.deleted && b.deleted) { + if (a.deletedHlc == null) return b.deletedHlc; + if (b.deletedHlc == null) return a.deletedHlc; + return a.deletedHlc! >= b.deletedHlc! ? a.deletedHlc : b.deletedHlc; + } + if (a.deleted) return a.deletedHlc; + if (b.deleted) return b.deletedHlc; + return null; +} + +/// Merges two versions of the same record. +/// +/// Per-field last-writer-wins over the union of field names, plus a sticky +/// delete: `deleted = a.deleted || b.deleted`, never the other way around. A +/// tombstone can never be resurrected by merging in an older, non-deleted +/// copy pulled from a device that hasn't seen the delete yet. +/// +/// Commutative and idempotent: `mergeRecord(a, b) == mergeRecord(b, a)` and +/// `mergeRecord(a, a) == a`. +/// +/// Throws [ArgumentError] if `a.id != b.id` -- merging two different +/// records is a caller bug, not a case to silently paper over. +Record mergeRecord(Record a, Record b) { + if (a.id != b.id) { + throw ArgumentError( + 'cannot merge records with different ids: ${a.id} != ${b.id}', + ); + } + final mergedFields = Map.from(a.fields); + for (final entry in b.fields.entries) { + final existing = mergedFields[entry.key]; + mergedFields[entry.key] = existing == null + ? entry.value + : mergeField(existing, entry.value); + } + return Record( + id: a.id, + fields: mergedFields, + deleted: a.deleted || b.deleted, + deletedHlc: _mergeDeletedHlc(a, b), + ); +} diff --git a/crdt_sync_dart/lib/src/sync.dart b/crdt_sync_dart/lib/src/sync.dart new file mode 100644 index 0000000..946729b --- /dev/null +++ b/crdt_sync_dart/lib/src/sync.dart @@ -0,0 +1,48 @@ +import 'github_client.dart'; +import 'log.dart'; +import 'record.dart'; + +const _defaultFilename = 'log.json'; + +/// Runs one full sync tick: pull every other device's log, merge, push. +/// +/// Pulls from `//` for every device +/// directory GitHub reports under [pathPrefix], merges each into +/// [localLog] with [mergeLogs], then pushes this device's own merged result +/// to `//`. +/// +/// [encode] serializes a merged log for pushing. [decode] parses a remote +/// device's pushed text back into a log; throwing [FormatException] is +/// treated as a corrupt/unparsable push, and that device is skipped for +/// this tick rather than aborting the whole sync. +Future syncLog({ + required GitHubClient client, + required String deviceId, + required String pathPrefix, + required Log localLog, + required String Function(Log log) encode, + required Log Function(String text) decode, + String filename = _defaultFilename, + String commitMessage = 'crdt_sync: update log', +}) async { + var mergedLog = Map.from(localLog); + for (final otherDeviceId in await client.listDirectory(pathPrefix)) { + if (otherDeviceId == deviceId) continue; + final text = await client.getFileText( + '$pathPrefix/$otherDeviceId/$filename', + ); + if (text == null) continue; + try { + mergedLog = mergeLogs(mergedLog, decode(text)); + } on FormatException { + continue; + } + } + + await client.putFileText( + '$pathPrefix/$deviceId/$filename', + encode(mergedLog), + message: commitMessage, + ); + return mergedLog; +} diff --git a/crdt_sync_dart/pubspec.yaml b/crdt_sync_dart/pubspec.yaml new file mode 100644 index 0000000..f1d6505 --- /dev/null +++ b/crdt_sync_dart/pubspec.yaml @@ -0,0 +1,16 @@ +name: crdt_sync +description: >- + Shared CRDT merge scheme + GitHub-Contents-API sync transport for + cross-device apps. Pure Dart, no Flutter dependency. +version: 0.1.0 +publish_to: 'none' + +environment: + sdk: ^3.12.2 + +dependencies: + http: ^1.6.0 + +dev_dependencies: + lints: ^5.0.0 + test: ^1.25.0 diff --git a/crdt_sync_dart/test/github_client_test.dart b/crdt_sync_dart/test/github_client_test.dart new file mode 100644 index 0000000..6f70145 --- /dev/null +++ b/crdt_sync_dart/test/github_client_test.dart @@ -0,0 +1,195 @@ +import 'dart:convert'; + +import 'package:crdt_sync/crdt_sync.dart'; +import 'package:http/http.dart' as http; +import 'package:http/testing.dart' as http_testing; +import 'package:test/test.dart'; + +http.Response _response(int statusCode, [Object? jsonBody]) => + http.Response(jsonEncode(jsonBody ?? {}), statusCode); + +/// Sentinel meaning "this GET throws a network exception", since a real +/// [http.Response] can't carry an invalid status code. +final _networkError = Object(); + +/// Builds a client whose GETs return [getResponses] in call order (each +/// either an [http.Response] or [_networkError]), and whose PUT returns +/// [putResponse] (or throws [putError] if network fails). +GitHubClient _client( + List getResponses, { + http.Response? putResponse, + Object? putError, +}) { + var getIndex = 0; + final mock = http_testing.MockClient((request) async { + if (request.method == 'GET') { + final response = getResponses[getIndex]; + getIndex++; + if (identical(response, _networkError)) { + throw http.ClientException('offline'); + } + return response as http.Response; + } + if (request.method == 'PUT') { + if (putError != null) throw putError; + return putResponse!; + } + throw UnsupportedError('unexpected method ${request.method}'); + }); + return GitHubClient( + owner: 'kuhyx', + repo: 'crdt-sync-demo', + token: 'fake-token', + httpClient: mock, + ); +} + +void main() { + group('getFileText', () { + test('returns decoded content on success', () async { + final encoded = base64.encode(utf8.encode('hello world')); + final client = _client([ + _response(200, {'content': encoded}), + ]); + expect(await client.getFileText('devices/pc/log.json'), 'hello world'); + }); + + test('returns null for an unused path on a real repo', () async { + final client = _client([_response(404), _response(200)]); + expect(await client.getFileText('devices/phone/log.json'), isNull); + }); + + test('throws RepoNotFoundError when the repo itself is missing', () async { + final client = _client([_response(404), _response(404)]); + expect( + () => client.getFileText('devices/pc/log.json'), + throwsA(isA()), + ); + }); + + test('throws GitHubSyncError on a non-2xx non-404', () async { + final client = _client([_response(500)]); + expect( + () => client.getFileText('devices/pc/log.json'), + throwsA(isA()), + ); + }); + + test('throws GitHubSyncError on a network exception', () async { + final client = _client([_networkError]); + expect( + () => client.getFileText('devices/pc/log.json'), + throwsA(isA()), + ); + }); + + test( + 'treats a network error during the repo check as repo missing', + () async { + final client = _client([_response(404), _networkError]); + expect( + () => client.getFileText('devices/pc/log.json'), + throwsA(isA()), + ); + }, + ); + }); + + group('listDirectory', () { + test('returns entry names regardless of type', () async { + final payload = [ + {'name': 'pc', 'type': 'dir'}, + {'name': 'phone', 'type': 'dir'}, + {'not_a_name': 'x'}, + ]; + final client = _client([_response(200, payload)]); + expect(await client.listDirectory('devices'), ['pc', 'phone']); + }); + + test('returns an empty list when the response is not a list', () async { + final client = _client([ + _response(200, {'unexpected': 'shape'}), + ]); + expect(await client.listDirectory('devices'), isEmpty); + }); + + test('returns an empty list for an unused path on a real repo', () async { + final client = _client([_response(404), _response(200)]); + expect(await client.listDirectory('devices'), isEmpty); + }); + + test('throws RepoNotFoundError when the repo itself is missing', () async { + final client = _client([_response(404), _response(404)]); + expect( + () => client.listDirectory('devices'), + throwsA(isA()), + ); + }); + + test('throws GitHubSyncError on a non-2xx non-404', () async { + final client = _client([_response(500)]); + expect( + () => client.listDirectory('devices'), + throwsA(isA()), + ); + }); + }); + + group('putFileText', () { + test('creates a new file with no sha when none existed', () async { + final client = _client([ + _response(404), + _response(200), + ], putResponse: _response(201)); + await client.putFileText('devices/pc/log.json', '{}', message: 'm'); + }); + + test('updates an existing file by including its sha', () async { + final client = _client([ + _response(200, {'sha': 'abc123'}), + ], putResponse: _response(200)); + await client.putFileText('devices/pc/log.json', '{}', message: 'm'); + }); + + test( + 'throws RepoNotFoundError when checking sha on a missing repo', + () async { + final client = _client([_response(404), _response(404)]); + expect( + () => client.putFileText('devices/pc/log.json', '{}', message: 'm'), + throwsA(isA()), + ); + }, + ); + + test('throws GitHubSyncError when the sha check itself fails', () async { + final client = _client([_response(500)]); + expect( + () => client.putFileText('devices/pc/log.json', '{}', message: 'm'), + throwsA(isA()), + ); + }); + + test('throws GitHubSyncError on a put network exception', () async { + final client = _client([ + _response(404), + _response(200), + ], putError: http.ClientException('offline')); + expect( + () => client.putFileText('devices/pc/log.json', '{}', message: 'm'), + throwsA(isA()), + ); + }); + + test('throws GitHubSyncError on a put non-2xx response', () async { + final client = _client([ + _response(404), + _response(200), + ], putResponse: _response(422)); + expect( + () => client.putFileText('devices/pc/log.json', '{}', message: 'm'), + throwsA(isA()), + ); + }); + }); +} diff --git a/crdt_sync_dart/test/hlc_test.dart b/crdt_sync_dart/test/hlc_test.dart new file mode 100644 index 0000000..234e86c --- /dev/null +++ b/crdt_sync_dart/test/hlc_test.dart @@ -0,0 +1,117 @@ +import 'package:crdt_sync/crdt_sync.dart'; +import 'package:test/test.dart'; + +Hlc _make(int wallTimeMs, {int counter = 0, String nodeId = 'node-a'}) => + Hlc(wallTimeMs: wallTimeMs, counter: counter, nodeId: nodeId); + +void main() { + group('ordering', () { + test('greater wall time wins', () { + expect(_make(200) > _make(100), isTrue); + }); + + test('equal wall time, greater counter wins', () { + expect(_make(100, counter: 2) > _make(100, counter: 1), isTrue); + }); + + test('equal wall time and counter breaks tie on nodeId', () { + expect( + _make(100, counter: 1, nodeId: 'b') > + _make(100, counter: 1, nodeId: 'a'), + isTrue, + ); + }); + + test('identical clocks compare equal', () { + expect( + _make(100, counter: 1, nodeId: 'a'), + equals(_make(100, counter: 1, nodeId: 'a')), + ); + }); + }); + + group('newTick', () { + test('first tick has counter zero', () { + final tick = Hlc.newTick('node-a', wallTimeMsOverride: 1000); + expect(tick, equals(_make(1000))); + }); + + test('advancing wall clock resets counter', () { + final previous = _make(1000, counter: 5); + final tick = Hlc.newTick( + 'node-a', + previous: previous, + wallTimeMsOverride: 2000, + ); + expect(tick, equals(_make(2000))); + }); + + test('stalled wall clock increments counter', () { + final previous = _make(1000, counter: 5); + final tick = Hlc.newTick( + 'node-a', + previous: previous, + wallTimeMsOverride: 1000, + ); + expect(tick, equals(_make(1000, counter: 6))); + }); + + test('regressed wall clock still advances monotonically', () { + final previous = _make(1000, counter: 5); + final tick = Hlc.newTick( + 'node-a', + previous: previous, + wallTimeMsOverride: 500, + ); + expect(tick, equals(_make(1000, counter: 6))); + }); + + test('defaults to the real clock when unset', () { + final tick = Hlc.newTick('node-a'); + expect(tick.nodeId, 'node-a'); + expect(tick.wallTimeMs, greaterThan(0)); + }); + }); + + group('string round trip', () { + test('round trips through toStr and fromStr', () { + final original = _make(1751000123456, counter: 7, nodeId: 'phone'); + expect(Hlc.fromStr(original.toStr()), equals(original)); + }); + + test('toStr is lexicographically sortable by wall time', () { + expect(_make(1000).toStr().compareTo(_make(2000).toStr()), lessThan(0)); + }); + + test('toStr is lexicographically sortable by counter', () { + expect( + _make( + 1000, + counter: 1, + ).toStr().compareTo(_make(1000, counter: 200).toStr()), + lessThan(0), + ); + }); + + test('fromStr rejects a missing Z separator', () { + expect( + () => Hlc.fromStr('not-a-valid-clock-string-at-all'), + throwsFormatException, + ); + }); + + test('fromStr rejects a wrong-length iso prefix', () { + expect( + () => Hlc.fromStr('2026-07-05T12:00:00Z-0000-node-a'), + throwsFormatException, + ); + }); + + test('fromStr rejects a missing node id separator', () { + expect( + () => Hlc.fromStr('2026-07-05T12:00:00.000Z-0000'), + throwsFormatException, + ); + }); + }); +} diff --git a/crdt_sync_dart/test/log_test.dart b/crdt_sync_dart/test/log_test.dart new file mode 100644 index 0000000..7be4532 --- /dev/null +++ b/crdt_sync_dart/test/log_test.dart @@ -0,0 +1,85 @@ +import 'package:crdt_sync/crdt_sync.dart'; +import 'package:test/test.dart'; + +Hlc _make(int wallTimeMs, {int counter = 0, String nodeId = 'node-a'}) => + Hlc(wallTimeMs: wallTimeMs, counter: counter, nodeId: nodeId); + +(Log, Log) _sampleLogs() { + final a = { + 'shared': Record(id: 'shared', fields: {'text': ('a-version', _make(100))}), + 'only-in-a': Record(id: 'only-in-a', fields: {'text': ('a', _make(50))}), + }; + final b = { + 'shared': Record(id: 'shared', fields: {'text': ('b-version', _make(200))}), + 'only-in-b': Record(id: 'only-in-b', fields: {'text': ('b', _make(50))}), + }; + return (a, b); +} + +void main() { + group('mergeLogs', () { + test('returns the union of disjoint ids', () { + final local = { + 'a': Record(id: 'a', fields: {'text': ('a', _make(100))}), + }; + final remote = { + 'b': Record(id: 'b', fields: {'text': ('b', _make(100))}), + }; + expect(mergeLogs(local, remote), {...local, ...remote}); + }); + + test('merges a shared id field by field', () { + final local = { + 'a': Record(id: 'a', fields: {'text': ('old', _make(100))}), + }; + final remote = { + 'a': Record(id: 'a', fields: {'text': ('new', _make(200))}), + }; + final merged = mergeLogs(local, remote); + expect(merged['a']!.fields['text'], equals(('new', _make(200)))); + }); + + test('a delete on one side survives the merge', () { + final local = { + 'a': Record(id: 'a', fields: {}, deleted: true, deletedHlc: _make(200)), + }; + final remote = { + 'a': Record(id: 'a', fields: {'text': ('still here', _make(100))}), + }; + expect(mergeLogs(local, remote)['a']!.deleted, isTrue); + }); + + test('does not mutate either input', () { + final local = { + 'a': Record(id: 'a', fields: {'text': ('a', _make(100))}), + }; + final remote = { + 'b': Record(id: 'b', fields: {'text': ('b', _make(100))}), + }; + final localBefore = Map.from(local); + final remoteBefore = Map.from(remote); + mergeLogs(local, remote); + expect(local, equals(localBefore)); + expect(remote, equals(remoteBefore)); + }); + }); + + group('convergence properties', () { + test('is commutative', () { + final (a, b) = _sampleLogs(); + expect(mergeLogs(a, b), equals(mergeLogs(b, a))); + }); + + test('is idempotent', () { + final (a, _) = _sampleLogs(); + expect(mergeLogs(a, a), equals(a)); + }); + + test('repeated merge of the same remote is a no-op', () { + final (a, b) = _sampleLogs(); + final once = mergeLogs(a, b); + final twice = mergeLogs(once, b); + expect(once, equals(twice)); + }); + }); +} diff --git a/crdt_sync_dart/test/record_test.dart b/crdt_sync_dart/test/record_test.dart new file mode 100644 index 0000000..24bebaa --- /dev/null +++ b/crdt_sync_dart/test/record_test.dart @@ -0,0 +1,141 @@ +import 'package:crdt_sync/crdt_sync.dart'; +import 'package:test/test.dart'; + +Hlc _make(int wallTimeMs, {int counter = 0, String nodeId = 'node-a'}) => + Hlc(wallTimeMs: wallTimeMs, counter: counter, nodeId: nodeId); + +void main() { + group('mergeField', () { + test('greater Hlc wins', () { + final older = ('a', _make(100)); + final newer = ('b', _make(200)); + expect(mergeField(older, newer), equals(newer)); + expect(mergeField(newer, older), equals(newer)); + }); + + test('equal Hlc keeps the first argument', () { + final clock = _make(100); + final a = ('a', clock); + final b = ('a', clock); + expect(mergeField(a, b), equals(a)); + }); + }); + + group('mergeRecord', () { + test('throws when ids differ', () { + final a = Record(id: 'a', fields: {}); + final b = Record(id: 'b', fields: {}); + expect(() => mergeRecord(a, b), throwsArgumentError); + }); + + test('merges disjoint fields', () { + final a = Record(id: 'x', fields: {'text': ('hello', _make(100))}); + final b = Record(id: 'x', fields: {'priority': ('high', _make(50))}); + final merged = mergeRecord(a, b); + expect(merged.fields, { + 'text': ('hello', _make(100)), + 'priority': ('high', _make(50)), + }); + }); + + test('shared field keeps the later write', () { + final a = Record(id: 'x', fields: {'text': ('old', _make(100))}); + final b = Record(id: 'x', fields: {'text': ('new', _make(200))}); + final merged = mergeRecord(a, b); + expect(merged.fields['text'], equals(('new', _make(200)))); + }); + + test('delete is sticky against an older non-deleted copy', () { + final deleted = Record( + id: 'x', + fields: {}, + deleted: true, + deletedHlc: _make(200), + ); + final stale = Record( + id: 'x', + fields: {'text': ('resurrected', _make(100))}, + ); + expect(mergeRecord(deleted, stale).deleted, isTrue); + expect(mergeRecord(stale, deleted).deleted, isTrue); + }); + + test('neither side deleted stays not deleted', () { + final a = Record(id: 'x', fields: {}); + final b = Record(id: 'x', fields: {}); + final merged = mergeRecord(a, b); + expect(merged.deleted, isFalse); + expect(merged.deletedHlc, isNull); + }); + + test('both sides deleted keeps the later delete clock', () { + final a = Record( + id: 'x', + fields: {}, + deleted: true, + deletedHlc: _make(100), + ); + final b = Record( + id: 'x', + fields: {}, + deleted: true, + deletedHlc: _make(200), + ); + expect(mergeRecord(a, b).deletedHlc, equals(_make(200))); + }); + + test('both deleted but one side missing a clock keeps the other', () { + final a = Record(id: 'x', fields: {}, deleted: true); + final b = Record( + id: 'x', + fields: {}, + deleted: true, + deletedHlc: _make(200), + ); + expect(mergeRecord(a, b).deletedHlc, equals(_make(200))); + expect(mergeRecord(b, a).deletedHlc, equals(_make(200))); + }); + + test('is commutative', () { + final recordA = Record(id: 'x', fields: {'text': ('a', _make(100))}); + final recordB = Record( + id: 'x', + fields: {'text': ('b', _make(200)), 'extra': ('e', _make(1))}, + ); + expect( + mergeRecord(recordA, recordB), + equals(mergeRecord(recordB, recordA)), + ); + }); + + test('is idempotent', () { + final record = Record(id: 'x', fields: {'text': ('a', _make(100))}); + expect(mergeRecord(record, record), equals(record)); + }); + }); + + group('Record JSON round trip', () { + test('round trips a record with fields', () { + final record = Record( + id: 'x', + fields: {'text': ('hello', _make(100, nodeId: 'pc'))}, + ); + expect(Record.fromJson(record.toJson()), equals(record)); + }); + + test('round trips a deleted record', () { + final record = Record( + id: 'x', + fields: {}, + deleted: true, + deletedHlc: _make(200), + ); + expect(Record.fromJson(record.toJson()), equals(record)); + }); + + test('round trips a record with no delete clock', () { + final record = Record(id: 'x', fields: {}); + expect(Record.fromJson(record.toJson()), equals(record)); + }); + }); +} diff --git a/crdt_sync_dart/test/sync_test.dart b/crdt_sync_dart/test/sync_test.dart new file mode 100644 index 0000000..c906d80 --- /dev/null +++ b/crdt_sync_dart/test/sync_test.dart @@ -0,0 +1,192 @@ +import 'dart:convert'; + +import 'package:crdt_sync/crdt_sync.dart'; +import 'package:http/http.dart' as http; +import 'package:http/testing.dart' as http_testing; +import 'package:test/test.dart'; + +Hlc _make(int wallTimeMs, {int counter = 0, String nodeId = 'node-a'}) => + Hlc(wallTimeMs: wallTimeMs, counter: counter, nodeId: nodeId); + +String _encode(Log log) => + jsonEncode(log.map((id, record) => MapEntry(id, record.toJson()))); + +Log _decode(String text) => (jsonDecode(text) as Map).map( + (id, data) => MapEntry(id, Record.fromJson(data as Map)), +); + +http.Response _response(int statusCode, [Object? jsonBody]) => + http.Response(jsonEncode(jsonBody ?? {}), statusCode); + +class _PutCall { + _PutCall(this.path, this.body); + final String path; + final Map body; +} + +/// Builds a [GitHubClient] backed by an in-memory router: GET +/// `.../contents/` returns `contentResponses[key]` (404 if absent), the +/// bare repo-existence GET always succeeds, and every PUT is recorded into +/// [putCalls] and answered with 200. +({GitHubClient client, List<_PutCall> putCalls}) _client({ + Map contentResponses = const {}, +}) { + final putCalls = <_PutCall>[]; + final mock = http_testing.MockClient((request) async { + final path = request.url.path; + if (!path.contains('/contents/')) { + return _response(200); + } + final key = path.split('/contents/').last; + if (request.method == 'PUT') { + putCalls.add( + _PutCall(key, jsonDecode(request.body) as Map), + ); + return _response(200); + } + return contentResponses[key] ?? _response(404); + }); + final client = GitHubClient( + owner: 'kuhyx', + repo: 'crdt-sync-demo', + token: 'fake-token', + httpClient: mock, + ); + return (client: client, putCalls: putCalls); +} + +http.Response _directoryOf(List deviceIds) => + _response(200, deviceIds.map((id) => {'name': id, 'type': 'dir'}).toList()); + +http.Response _fileContaining(String text) => + _response(200, {'content': base64.encode(utf8.encode(text))}); + +void main() { + group('syncLog', () { + test('pushes the local log when no other devices have synced', () async { + final localLog = { + 'a': Record(id: 'a', fields: {'text': ('hello', _make(100))}), + }; + final (:client, :putCalls) = _client( + contentResponses: {'devices': _directoryOf([])}, + ); + + final merged = await syncLog( + client: client, + deviceId: 'pc', + pathPrefix: 'devices', + localLog: localLog, + encode: _encode, + decode: _decode, + ); + + expect(merged, equals(localLog)); + expect(putCalls, hasLength(1)); + expect(putCalls.single.path, 'devices/pc/log.json'); + }); + + test('skips its own device id when listing', () async { + final (:client, :putCalls) = _client( + contentResponses: { + 'devices': _directoryOf(['pc', 'phone']), + 'devices/phone/log.json': _fileContaining('{}'), + }, + ); + + await syncLog( + client: client, + deviceId: 'pc', + pathPrefix: 'devices', + localLog: {}, + encode: _encode, + decode: _decode, + ); + + expect(putCalls, hasLength(1)); + }); + + test('skips a device with no pushed file yet', () async { + final (:client, :putCalls) = _client( + contentResponses: { + 'devices': _directoryOf(['phone']), + }, + ); + + final merged = await syncLog( + client: client, + deviceId: 'pc', + pathPrefix: 'devices', + localLog: {}, + encode: _encode, + decode: _decode, + ); + + expect(merged, isEmpty); + expect(putCalls.single.body['content'], base64.encode(utf8.encode('{}'))); + }); + + test('skips a device whose pushed file is corrupt', () async { + final (:client, :putCalls) = _client( + contentResponses: { + 'devices': _directoryOf(['phone']), + 'devices/phone/log.json': _fileContaining('{not valid json'), + }, + ); + + final merged = await syncLog( + client: client, + deviceId: 'pc', + pathPrefix: 'devices', + localLog: {}, + encode: _encode, + decode: _decode, + ); + + expect(merged, isEmpty); + expect(putCalls, hasLength(1)); + }); + + test('merges in a remote device\'s entries', () async { + final remoteLog = { + 'b': Record(id: 'b', fields: {'text': ('from phone', _make(100))}), + }; + final (:client, :putCalls) = _client( + contentResponses: { + 'devices': _directoryOf(['phone']), + 'devices/phone/log.json': _fileContaining(_encode(remoteLog)), + }, + ); + + final merged = await syncLog( + client: client, + deviceId: 'pc', + pathPrefix: 'devices', + localLog: {}, + encode: _encode, + decode: _decode, + ); + + expect(merged, equals(remoteLog)); + }); + + test('uses a custom filename and commit message', () async { + final (:client, :putCalls) = _client( + contentResponses: {'devices': _directoryOf([])}, + ); + + await syncLog( + client: client, + deviceId: 'pc', + pathPrefix: 'devices', + localLog: {}, + encode: _encode, + decode: _decode, + filename: 'notes.json', + commitMessage: 'custom message', + ); + + expect(putCalls.single.path, 'devices/pc/notes.json'); + expect(putCalls.single.body['message'], 'custom message'); + }); + }); +} From 6b14e6d36d137d90851f00d1e282a2069c13d3a6 Mon Sep 17 00:00:00 2001 From: Krzysztof Rudnicki Date: Sun, 5 Jul 2026 20:55:22 +0200 Subject: [PATCH 2/2] Fix wire-format key mismatch with Python (deletedHlc -> deleted_hlc) Record.toJson() was writing the delete-clock field as deletedHlc while the Python package's Record.to_dict() writes deleted_hlc -- an undetected cross-language wire divergence: a tombstoned record synced Python<->Dart would silently lose its delete clock on the receiving side. Also widen syncLog's decode error handling from FormatException alone to also catch TypeError, matching the Python side's (ValueError, KeyError, TypeError) so a valid-JSON-but-wrong-shape push from an incompatible writer is skipped per-device instead of aborting the whole sync. Adds a fixture test shared verbatim with the Python package's test suite so this class of divergence fails loudly going forward. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_017McuA1nAi6jBT1EaS5BrVf --- crdt_sync_dart/lib/src/record.dart | 6 ++++-- crdt_sync_dart/lib/src/sync.dart | 11 +++++++--- crdt_sync_dart/test/record_test.dart | 31 ++++++++++++++++++++++++++++ crdt_sync_dart/test/sync_test.dart | 25 ++++++++++++++++++++++ 4 files changed, 68 insertions(+), 5 deletions(-) diff --git a/crdt_sync_dart/lib/src/record.dart b/crdt_sync_dart/lib/src/record.dart index 5b31756..acc1719 100644 --- a/crdt_sync_dart/lib/src/record.dart +++ b/crdt_sync_dart/lib/src/record.dart @@ -38,7 +38,9 @@ class Record { (name, field) => MapEntry(name, [field.$1, field.$2.toStr()]), ), 'deleted': deleted, - 'deletedHlc': deletedHlc?.toStr(), + // Wire key is snake_case to match crdt_sync's Python `Record.to_dict` -- + // this is a cross-language wire format, not a Dart-idiomatic API. + 'deleted_hlc': deletedHlc?.toStr(), }; factory Record.fromJson(Map json) { @@ -47,7 +49,7 @@ class Record { final pair = value as List; return MapEntry(name, (pair[0], Hlc.fromStr(pair[1] as String))); }); - final deletedHlcStr = json['deletedHlc'] as String?; + final deletedHlcStr = json['deleted_hlc'] as String?; return Record( id: json['id'] as String, fields: fields, diff --git a/crdt_sync_dart/lib/src/sync.dart b/crdt_sync_dart/lib/src/sync.dart index 946729b..ab4cb5c 100644 --- a/crdt_sync_dart/lib/src/sync.dart +++ b/crdt_sync_dart/lib/src/sync.dart @@ -12,9 +12,12 @@ const _defaultFilename = 'log.json'; /// to `//`. /// /// [encode] serializes a merged log for pushing. [decode] parses a remote -/// device's pushed text back into a log; throwing [FormatException] is -/// treated as a corrupt/unparsable push, and that device is skipped for -/// this tick rather than aborting the whole sync. +/// device's pushed text back into a log; throwing [FormatException] (bad +/// JSON syntax) or [TypeError] (valid JSON, wrong shape -- e.g. a missing +/// or mistyped field during [Record.fromJson]) is treated as a +/// corrupt/unparsable push, and that device is skipped for this tick +/// rather than aborting the whole sync. Mirrors the Python `sync_log`'s +/// `(ValueError, KeyError, TypeError)` catch for the same reason. Future syncLog({ required GitHubClient client, required String deviceId, @@ -36,6 +39,8 @@ Future syncLog({ mergedLog = mergeLogs(mergedLog, decode(text)); } on FormatException { continue; + } on TypeError { + continue; } } diff --git a/crdt_sync_dart/test/record_test.dart b/crdt_sync_dart/test/record_test.dart index 24bebaa..cfc45a7 100644 --- a/crdt_sync_dart/test/record_test.dart +++ b/crdt_sync_dart/test/record_test.dart @@ -138,4 +138,35 @@ void main() { expect(Record.fromJson(record.toJson()), equals(record)); }); }); + + group('cross-language wire format', () { + // Pins the exact wire shape shared with crdt-sync's Python Record. + // These values are duplicated verbatim in the Python package's + // `test_record.py` (`TestCrossLanguageWireFormat`). If this test and + // that one both pass, the two languages agree on the wire format; if + // only one changes, the two suites diverge and at least one of them + // fails -- catching exactly the kind of key-naming mismatch + // (`deleted_hlc` vs `deletedHlc`) that neither language's own + // round-trip test can see on its own. + test('matches the fixture shared with the Python package', () { + final record = Record( + id: 'abc123', + fields: { + 'text': ('hello', Hlc(wallTimeMs: 100, counter: 0, nodeId: 'pc')), + }, + deleted: true, + deletedHlc: Hlc(wallTimeMs: 1000, counter: 0, nodeId: 'node-a'), + ); + final expected = { + 'id': 'abc123', + 'fields': { + 'text': ['hello', '1970-01-01T00:00:00.100Z-0000-pc'], + }, + 'deleted': true, + 'deleted_hlc': '1970-01-01T00:00:01.000Z-0000-node-a', + }; + expect(record.toJson(), equals(expected)); + expect(Record.fromJson(expected), equals(record)); + }); + }); } diff --git a/crdt_sync_dart/test/sync_test.dart b/crdt_sync_dart/test/sync_test.dart index c906d80..84204d1 100644 --- a/crdt_sync_dart/test/sync_test.dart +++ b/crdt_sync_dart/test/sync_test.dart @@ -146,6 +146,31 @@ void main() { expect(putCalls, hasLength(1)); }); + test('skips a device whose pushed JSON has the wrong shape', () async { + // Valid JSON that isn't a record map (e.g. from an incompatible + // writer) must be skipped like corrupt JSON, not crash the whole + // sync -- this is what the `on TypeError` catch in `syncLog` is + // for, not just JSON syntax errors caught by `FormatException`. + final (:client, :putCalls) = _client( + contentResponses: { + 'devices': _directoryOf(['phone']), + 'devices/phone/log.json': _fileContaining('{"a": 5}'), + }, + ); + + final merged = await syncLog( + client: client, + deviceId: 'pc', + pathPrefix: 'devices', + localLog: {}, + encode: _encode, + decode: _decode, + ); + + expect(merged, isEmpty); + expect(putCalls, hasLength(1)); + }); + test('merges in a remote device\'s entries', () async { final remoteLog = { 'b': Record(id: 'b', fields: {'text': ('from phone', _make(100))}),