fix(crdt): finish the sync defects from the Dart port, server and crdt-js

crdt-dart/test/conformance/round_trip_test.dart

@@ -131,7 +131,9 @@ void main() {
       final done = Completer<void>();
       final sub = ws.subscribe(const StreamConfig(tables: ['notes']));
       sub.on((e) {
+        // The server sends each poll's batch as one `changes` frame.
         if (e is StreamChange) streamed.add(e.change.pk);
+        if (e is StreamChanges) streamed.addAll(e.changes.map((c) => c.pk));
         if (streamed.containsAll(['n1', 'n2']) && !done.isCompleted) {
           done.complete();
         }

crdt-js/src/__tests__/batch.test.ts

@@ -26,7 +26,10 @@ describe("BatchWriter parity", () => {
     store.batch("t", "p").setField("a", 1).commit();
     expect(store.canUndo).toBe(true);
     store.undo();
-    expect(store.getDocument<{ a?: number }>("t", "p")?.a).toBeUndefined();
+    // Undo writes a compensating change, and there is no field delete on
+    // the wire, so a field's first write undoes to null.
+    expect(store.getDocument<{ a?: number | null }>("t", "p")?.a).toBeNull();
+    expect(store.pendingCount).toBe(2);
   });
 
   it("a batch notifies subscribers exactly once", () => {

crdt-js/src/__tests__/cow-store.test.ts

@@ -108,8 +108,10 @@ describe("copy-on-write store", () => {
     expect(store.getText("notes", "n1", "body")).toBe("hello");
     expect(store.undo()).toBe(true);
     expect(store.getText("notes", "n1", "body")).toBe("");
-    // The field is gone entirely, not left behind as an empty text state.
-    expect(store.exportTable("notes")["n1"].fields).not.toHaveProperty("body");
+    // The undo is a delete of the inserted range, queued for push, so the
+    // field stays behind with its characters tombstoned.
+    const queued = store.getPendingChanges();
+    expect(queued[queued.length - 1].text_op?.op).toBe("delete");
   });
 
   it("gives afterMerge the post-merge field state as the result", () => {

crdt-js/src/__tests__/merge.test.ts

@@ -811,3 +811,108 @@ describe("Document CRDT merge", () => {
     expect(resolved.address).toEqual({ street: "123 Main St", city: "Springfield" });
   });
 });
+
+// ---------------------------------------------------------------------------
+// Field clock under redelivery (Go ApplyChange keeps the newer HLC)
+// ---------------------------------------------------------------------------
+
+describe("mergeFieldState field clock", () => {
+  const older: HLC = { ts: 100, c: 0, node: "a" };
+  const newer: HLC = { ts: 200, c: 0, node: "b" };
+
+  const changeAt = (
+    crdt_type: ChangeRecord["crdt_type"],
+    hlc: HLC,
+    extra: Partial<ChangeRecord>,
+  ): ChangeRecord => ({
+    table: "t", pk: "1", field: "f", crdt_type, hlc, node_id: hlc.node, ...extra,
+  });
+
+  const cases: Array<[string, (hlc: HLC) => ChangeRecord]> = [
+    ["counter", (hlc) => changeAt("counter", hlc, { counter_delta: { inc: 1, dec: 0 } })],
+    ["set", (hlc) => changeAt("set", hlc, { set_op: { op: "add", elements: [hlc.node] } })],
+    ["list", (hlc) => changeAt("list", hlc, {
+      list_op: { op: "insert", node_id: hlc, parent_id: { ts: 0, c: 0, node: "" }, value: hlc.node },
+    })],
+    ["document", (hlc) => changeAt("document", hlc, { value: { path: `p.${hlc.node}`, value: 1 } })],
+  ];
+
+  it.each(cases)("%s: a redelivered older change does not regress the field HLC", (_type, make) => {
+    const afterNewer = mergeFieldState(null, make(newer));
+    const afterOlder = mergeFieldState(afterNewer, make(older));
+    expect(afterOlder.hlc).toEqual(newer);
+    expect(afterOlder.node_id).toBe("b");
+  });
+});
+
+// ---------------------------------------------------------------------------
+// Set element keys follow Go json.Marshal bytes
+// ---------------------------------------------------------------------------
+
+describe("set element keys (Go parity)", () => {
+  const goHLC: HLC = { ts: 100, c: 0, node: "go" };
+  const jsHLC: HLC = { ts: 200, c: 0, node: "js" };
+
+  it("removes an element a Go writer added under its json.Marshal key", () => {
+    // Go ORSetState.Add keys "a<b" as json.Marshal does: "a\u003cb".
+    const goState: ORSetState = { entries: { '"a\\u003cb"': [{ node: "go", hlc: goHLC }] }, removed: {} };
+    const local = { type: "set" as const, hlc: goHLC, node_id: "go", set_state: goState };
+    const removed = mergeFieldState(local, {
+      table: "t", pk: "1", field: "tags", crdt_type: "set", hlc: jsHLC, node_id: "js",
+      set_op: { op: "remove", elements: ["a<b"] },
+    });
+    expect(setElements(removed.set_state!)).toEqual([]);
+  });
+
+  it("keys objects by sorted field names, so key order does not split an element", () => {
+    let state = mergeFieldState(null, {
+      table: "t", pk: "1", field: "tags", crdt_type: "set", hlc: goHLC, node_id: "go",
+      set_op: { op: "add", elements: [{ b: 1, a: 2 }] },
+    });
+    state = mergeFieldState(state, {
+      table: "t", pk: "1", field: "tags", crdt_type: "set", hlc: jsHLC, node_id: "js",
+      set_op: { op: "add", elements: [{ a: 2, b: 1 }] },
+    });
+    expect(Object.keys(state.set_state!.entries)).toEqual(['{"a":2,"b":1}']);
+  });
+
+  it("orders elements by UTF-8 bytes like Go sort.Strings", () => {
+    const tag = [{ node: "go", hlc: goHLC }];
+    // U+FF01 encodes as EF BC 81, U+1F600 as F0 9F 98 80: Go puts U+FF01 first.
+    // In UTF-16 the emoji's high surrogate (D83D) sorts before FF01.
+    const state: ORSetState = { entries: { '"\u{1F600}"': tag, '"！"': tag }, removed: {} };
+    expect(setElements(state)).toEqual(["！", "\u{1F600}"]);
+  });
+});
+
+// ---------------------------------------------------------------------------
+// Set states written before element keys were canonical
+// ---------------------------------------------------------------------------
+
+describe("set states with legacy element keys", () => {
+  const t1: ORSetTag = { node: "js", hlc: { ts: 100, c: 0, node: "js" } };
+  // "a<b" keyed by JSON.stringify, as crdt-js stored it before.
+  const legacy = (removed: boolean): ORSetState => ({
+    entries: { '"a<b"': [t1] },
+    removed: removed ? { ['"a<b"|' + tagKey(t1)]: true } : {},
+  });
+
+  it("does not show an element twice next to its canonical twin", () => {
+    const fresh: ORSetState = { entries: { '"a\\u003cb"': [{ node: "go", hlc: { ts: 150, c: 0, node: "go" } }] }, removed: {} };
+    expect(setElements(mergeSet(legacy(false), fresh))).toEqual(["a<b"]);
+  });
+
+  it("keeps a removed element removed when it reappears under the canonical key", () => {
+    const readd: ORSetState = { entries: { '"a\\u003cb"': [t1] }, removed: {} };
+    expect(setElements(mergeSet(legacy(true), readd))).toEqual([]);
+  });
+
+  it("lets a remove match an element stored under its legacy key", () => {
+    const local = { type: "set" as const, hlc: t1.hlc, node_id: "js", set_state: legacy(false) };
+    const out = mergeFieldState(local, {
+      table: "t", pk: "1", field: "tags", crdt_type: "set", hlc: { ts: 200, c: 0, node: "js" }, node_id: "js",
+      set_op: { op: "remove", elements: ["a<b"], tags: [t1] },
+    });
+    expect(setElements(out.set_state!)).toEqual([]);
+  });
+});

crdt-js/src/__tests__/offline.test.ts

@@ -62,6 +62,11 @@ describe("offline queue", () => {
     expect(doc?.c).toBeUndefined();
     expect(doc?.a).toBe(1);
     expect(doc?.b).toBe(2);
+    // Undo writes a compensating change, so it needs room in the queue too,
+    // and a refused undo keeps its entry.
+    expect(() => store.undo()).toThrow(CRDTError);
+    expect(store.canUndo).toBe(true);
+    store.clearPendingChanges(); // as a push would
     // Nor in undo history: only "a" and "b" were ever recorded, so exactly
     // two undos succeed and a third finds nothing.
     expect(store.undo()).toBe(true); // undoes "b"

crdt-js/src/__tests__/persist-debounce.test.ts

@@ -55,10 +55,11 @@ describe("persistence debounce", () => {
     expect(spy).not.toHaveBeenCalled();
   });
 
-  it("does not re-persist pending changes on a document-only flush (undo)", async () => {
-    // undo() also only touches document state. Mirrors the drain test but
-    // through the explicit flushPersistence() escape hatch instead of the
-    // timer, since flushPersistence gates on the same dirty flag.
+  it("does not re-persist pending changes on a document-only flush (applyChanges)", async () => {
+    // Mirrors the drain test but through the explicit flushPersistence()
+    // escape hatch instead of the timer, since flushPersistence gates on
+    // the same dirty flag. (undo() used to serve here; it now queues a
+    // compensating change, so it is no longer document-only.)
     const storage = new MemoryStorage();
     const store = new CRDTStore("n1", new HybridClock("n1"), storage, {
       persistDebounceMs: 10,
@@ -68,7 +69,10 @@ describe("persistence debounce", () => {
     await store.flushPersistence(); // clears the dirty flag from setField
 
     const spy = vi.spyOn(storage, "savePendingChanges");
-    store.undo();
+    store.applyChanges([{
+      table: "t", pk: "p", field: "f", crdt_type: "lww",
+      hlc: { ts: 1, c: 0, node: "remote" }, node_id: "remote", value: "hi",
+    }]);
     await store.flushPersistence();
     expect(spy).not.toHaveBeenCalled();
   });

crdt-js/src/__tests__/sync.test.ts

@@ -4,6 +4,7 @@ import { CRDTClient } from "../client.js";
 import { CRDTStore } from "../store.js";
 import type { Transport, PushRequest, ChangeRecord } from "../types.js";
 import type { StorePlugin, SyncHook, PresenceHook } from "../plugin.js";
+import { TransportError } from "../errors.js";
 
 const HLC0 = { ts: "0", c: 0, node: "" };
 
@@ -151,3 +152,81 @@ describe("presence hooks", () => {
     expect(seen).toEqual(["join"]);
   });
 });
+
+describe("SyncEngine push rejection", () => {
+  // A relay that rejects any batch containing pk "bad" the way the Go
+  // server rejects a BeforeInboundChange hook error: nothing merged, and
+  // the same answer however often the batch is resent.
+  function rejecting(status: number) {
+    const accepted: string[] = [];
+    const transport: Transport = {
+      async pull() { return { changes: [], latest_hlc: HLC0 }; },
+      async push(req) {
+        if (req.changes.some((c) => c.pk === "bad")) {
+          throw new TransportError(
+            `CRDT /push returned ${status}: crdt: inbound change hook: pk not writable`, status);
+        }
+        accepted.push(...req.changes.map((c) => c.pk));
+        return { merged: req.changes.length, latest_hlc: HLC0 };
+      },
+    };
+    const client = new CRDTClient({ nodeID: "n1", transport });
+    const store = new CRDTStore("n1", client.clock, undefined, { persistDebounceMs: 0 });
+    return { accepted, store, engine: new SyncEngine(client, store) };
+  }
+
+  it("a deterministically rejected change does not block later writes forever", async () => {
+    const { accepted, store, engine } = rejecting(422);
+    store.setField("t", "bad", "f", 1);
+    await engine.sync().catch(() => {});
+    store.setField("t", "good", "f", 2);
+    await engine.sync().catch(() => {});
+
+    expect(accepted).toContain("good");
+    expect(store.pendingCount).toBe(0);
+  });
+
+  it("isolates the rejected change and pushes the rest of its batch", async () => {
+    const { accepted, store, engine } = rejecting(422);
+    const rejected: string[] = [];
+    engine.onPushRejected((rs) => rejected.push(...rs.map((r) => r.change.pk)));
+
+    store.setField("t", "good1", "f", 1);
+    store.setField("t", "bad", "f", 2);
+    store.setField("t", "good2", "f", 3);
+    const report = await engine.sync();
+
+    expect(accepted.sort()).toEqual(["good1", "good2"]);
+    expect(rejected).toEqual(["bad"]);
+    expect(report.rejected).toBe(1);
+    expect(report.pushed).toBe(2);
+    expect(store.pendingCount).toBe(0);
+  });
+
+  it("recognises a rejection from a server that still answers 500", async () => {
+    const { accepted, store, engine } = rejecting(500);
+    const rejected: string[] = [];
+    engine.onPushRejected((rs) => rejected.push(...rs.map((r) => r.change.pk)));
+
+    store.setField("t", "bad", "f", 1);
+    store.setField("t", "good", "f", 2);
+    await engine.sync();
+
+    expect(accepted).toEqual(["good"]);
+    expect(rejected).toEqual(["bad"]);
+  });
+
+  it("keeps the batch pending on a transient failure", async () => {
+    const transport: Transport = {
+      async pull() { return { changes: [], latest_hlc: HLC0 }; },
+      async push() { throw new TransportError("CRDT /push returned 503: unavailable", 503); },
+    };
+    const client = new CRDTClient({ nodeID: "n1", transport });
+    const store = new CRDTStore("n1", client.clock, undefined, { persistDebounceMs: 0 });
+    const engine = new SyncEngine(client, store);
+
+    store.setField("t", "p", "f", 1);
+    await expect(engine.sync()).rejects.toThrow("503");
+    expect(store.pendingCount).toBe(1);
+  });
+});

crdt-js/src/__tests__/transport.test.ts

@@ -353,3 +353,25 @@ describe("timeout, retry, empty body", () => {
     await expect(t.pull({ tables: ["a"], node_id: "n1" })).rejects.toThrow();
   });
 });
+
+describe("HttpTransport push rejection", () => {
+  it("does not retry a 500 carrying the Go server's hook rejection", async () => {
+    const fetchFn = mockFetch({ error: "crdt: inbound change hook: pk not writable" }, 500);
+    const transport = new HttpTransport({ baseURL: "https://api.example.com/sync", fetch: fetchFn });
+
+    const err = await transport.push({ changes: [], node_id: "n1" }).catch((e: unknown) => e);
+    expect(err).toBeInstanceOf(TransportError);
+    expect((err as TransportError).retryable).toBe(false);
+    expect((fetchFn as any).mock.calls).toHaveLength(1);
+  });
+
+  it("still retries a plain 500", async () => {
+    const fetchFn = mockFetch({ error: "crdt: write state: disk full" }, 500);
+    const transport = new HttpTransport({
+      baseURL: "https://api.example.com/sync", fetch: fetchFn, retries: 2,
+    });
+
+    await transport.push({ changes: [], node_id: "n1" }).catch(() => {});
+    expect((fetchFn as any).mock.calls).toHaveLength(3);
+  });
+});