diff --git a/crdt-dart/test/conformance/round_trip_test.dart b/crdt-dart/test/conformance/round_trip_test.dart index 29e0415..7cfed2a 100644 --- a/crdt-dart/test/conformance/round_trip_test.dart +++ b/crdt-dart/test/conformance/round_trip_test.dart @@ -131,7 +131,9 @@ void main() { final done = Completer(); 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(); } diff --git a/crdt-js/src/__tests__/batch.test.ts b/crdt-js/src/__tests__/batch.test.ts index 80aa07d..ebb5d00 100644 --- a/crdt-js/src/__tests__/batch.test.ts +++ b/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", () => { diff --git a/crdt-js/src/__tests__/cow-store.test.ts b/crdt-js/src/__tests__/cow-store.test.ts index 10588fa..8e50064 100644 --- a/crdt-js/src/__tests__/cow-store.test.ts +++ b/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", () => { diff --git a/crdt-js/src/__tests__/merge.test.ts b/crdt-js/src/__tests__/merge.test.ts index 4f3c7b5..2835898 100644 --- a/crdt-js/src/__tests__/merge.test.ts +++ b/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 => ({ + 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 { + 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 ({ + entries: { '"a { + const fresh: ORSetState = { entries: { '"a\\u003cb"': [{ node: "go", hlc: { ts: 150, c: 0, node: "go" } }] }, removed: {} }; + expect(setElements(mergeSet(legacy(false), fresh))).toEqual(["a { + 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 { 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" diff --git a/crdt-js/src/__tests__/persist-debounce.test.ts b/crdt-js/src/__tests__/persist-debounce.test.ts index d51abc0..49f51f7 100644 --- a/crdt-js/src/__tests__/persist-debounce.test.ts +++ 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(); }); diff --git a/crdt-js/src/__tests__/sync.test.ts b/crdt-js/src/__tests__/sync.test.ts index 80206a3..052fa2a 100644 --- a/crdt-js/src/__tests__/sync.test.ts +++ b/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); + }); +}); diff --git a/crdt-js/src/__tests__/transport.test.ts b/crdt-js/src/__tests__/transport.test.ts index c7feef8..51e9b9d 100644 --- a/crdt-js/src/__tests__/transport.test.ts +++ b/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); + }); +}); diff --git a/crdt-js/src/__tests__/undo-sync.test.ts b/crdt-js/src/__tests__/undo-sync.test.ts new file mode 100644 index 0000000..2388154 --- /dev/null +++ b/crdt-js/src/__tests__/undo-sync.test.ts @@ -0,0 +1,197 @@ +import { describe, it, expect } from "vitest"; +import { SyncEngine } from "../sync.js"; +import { CRDTClient } from "../client.js"; +import { CRDTStore } from "../store.js"; +import type { Transport, ChangeRecord } from "../types.js"; + +// Undo and redo emit compensating changes, so an undo survives a sync and +// every replica ends up showing the same thing. (Go parity: the server +// only knows about changes it is sent.) + +const HLC0 = { ts: "0", c: 0, node: "" }; + +function relay(): Transport { + const log: ChangeRecord[] = []; + return { + async pull() { return { changes: [...log], latest_hlc: HLC0 }; }, + async push(req) { + log.push(...req.changes); + return { merged: req.changes.length, latest_hlc: HLC0 }; + }, + }; +} + +function pair() { + const t = relay(); + const mk = (node: string) => { + const client = new CRDTClient({ nodeID: node, transport: t }); + const store = new CRDTStore(node, client.clock, undefined, { persistDebounceMs: 0 }); + return { store, engine: new SyncEngine(client, store) }; + }; + const a = mk("a"); + const b = mk("b"); + const syncAll = async () => { + await a.engine.sync(); + await b.engine.sync(); + await a.engine.sync(); + }; + return { a: a.store, b: b.store, syncAll }; +} + +const doc = (s: CRDTStore) => s.getDocument>("t", "1"); + +describe("undo survives sync", () => { + it("lww", async () => { + const { a, b, syncAll } = pair(); + a.setField("t", "1", "title", "v1"); + await syncAll(); + a.setField("t", "1", "title", "v2"); + await syncAll(); + + expect(a.undo()).toBe(true); + expect(a.pendingCount).toBe(1); + await syncAll(); + expect(doc(a)?.title).toBe("v1"); + expect(doc(b)?.title).toBe("v1"); + }); + + it("counter", async () => { + const { a, b, syncAll } = pair(); + a.incrementCounter("t", "1", "n", 5); + await syncAll(); + b.incrementCounter("t", "1", "n", 2); + a.incrementCounter("t", "1", "n", 3); + await syncAll(); + + expect(a.undo()).toBe(true); + await syncAll(); + expect(doc(a)?.n).toBe(7); + expect(doc(b)?.n).toBe(7); + }); + + it("set add and remove", async () => { + const { a, b, syncAll } = pair(); + a.addToSet("t", "1", "tags", ["x"]); + await syncAll(); + a.addToSet("t", "1", "tags", ["x", "y"]); + await syncAll(); + a.undo(); // the second add: y goes, x stays because the first add still holds it + await syncAll(); + expect(doc(a)?.tags).toEqual(["x"]); + expect(doc(b)?.tags).toEqual(["x"]); + + a.removeFromSet("t", "1", "tags", ["x"]); + await syncAll(); + a.undo(); + await syncAll(); + expect(doc(a)?.tags).toEqual(["x"]); + expect(doc(b)?.tags).toEqual(["x"]); + }); + + it("list insert and delete", async () => { + const { a, b, syncAll } = pair(); + const first = a.insertIntoList("t", "1", "items", "a"); + a.insertIntoList("t", "1", "items", "c", first.hlc); + a.insertIntoList("t", "1", "items", "b", first.hlc); + await syncAll(); + expect(doc(b)?.items).toEqual(["a", "b", "c"]); + + a.undo(); // the insert of "b" + await syncAll(); + expect(doc(a)?.items).toEqual(["a", "c"]); + expect(doc(b)?.items).toEqual(["a", "c"]); + + a.deleteFromList("t", "1", "items", first.hlc); + await syncAll(); + a.undo(); // "a" comes back where it was + await syncAll(); + expect(doc(a)?.items).toEqual(["a", "c"]); + expect(doc(b)?.items).toEqual(["a", "c"]); + }); + + it("text insert, delete and format", async () => { + const { a, b, syncAll } = pair(); + a.insertText("t", "1", "body", 0, "hello world"); + await syncAll(); + a.deleteText("t", "1", "body", 5, 6); + a.formatText("t", "1", "body", 0, 5, { bold: true }); + await syncAll(); + + a.undo(); // format + a.undo(); // delete + await syncAll(); + for (const s of [a, b]) { + expect(s.getText("t", "1", "body")).toBe("hello world"); + expect(s.getTextDelta("t", "1", "body")).toEqual([{ insert: "hello world" }]); + } + }); + + it("text insert undo removes only what it inserted, beside a concurrent edit", async () => { + const { a, b, syncAll } = pair(); + a.insertText("t", "1", "body", 0, "abc"); + await syncAll(); + b.insertText("t", "1", "body", 0, "X"); + await syncAll(); + + a.undo(); // the "abc" insert + await syncAll(); + expect(a.getText("t", "1", "body")).toBe("X"); + expect(b.getText("t", "1", "body")).toBe("X"); + }); + + it("document path", async () => { + const { a, b, syncAll } = pair(); + a.setDocumentField("t", "1", "meta", "author.name", "ann"); + await syncAll(); + a.setDocumentField("t", "1", "meta", "author.name", "bob"); + a.setDocumentField("t", "1", "meta", "author.age", 30); + await syncAll(); + + a.undo(); // age set: the path did not exist before + a.undo(); // name back to ann + await syncAll(); + expect(doc(a)?.meta).toEqual(doc(b)?.meta); + expect(JSON.stringify(doc(a)?.meta)).toContain("ann"); + expect(JSON.stringify(doc(a)?.meta)).not.toContain("30"); + }); + + it("redo re-applies after a sync", async () => { + const { a, b, syncAll } = pair(); + a.setField("t", "1", "title", "v1"); + a.setField("t", "1", "title", "v2"); + await syncAll(); + a.undo(); + await syncAll(); + expect(a.redo()).toBe(true); + await syncAll(); + expect(doc(a)?.title).toBe("v2"); + expect(doc(b)?.title).toBe("v2"); + expect(a.undo()).toBe(true); + await syncAll(); + expect(doc(b)?.title).toBe("v1"); + }); +}); + +describe("undo of a record delete", () => { + it("drops the tombstone while it is still pending", async () => { + const { a, b, syncAll } = pair(); + a.setField("t", "1", "title", "kept"); + await syncAll(); + a.deleteDocument("t", "1"); + expect(a.undo()).toBe(true); + expect(a.pendingCount).toBe(0); + await syncAll(); + expect(doc(a)?.title).toBe("kept"); + expect(doc(b)?.title).toBe("kept"); + }); + + it("returns false once the tombstone was pushed, because tombstones are sticky", async () => { + const { a, b, syncAll } = pair(); + a.setField("t", "1", "title", "gone"); + a.deleteDocument("t", "1"); + await syncAll(); + expect(a.undo()).toBe(false); + expect(doc(a)).toBeNull(); + expect(doc(b)).toBeNull(); + }); +}); diff --git a/crdt-js/src/errors.ts b/crdt-js/src/errors.ts index 25b0e52..e606266 100644 --- a/crdt-js/src/errors.ts +++ b/crdt-js/src/errors.ts @@ -56,9 +56,10 @@ export class CRDTError extends Error { * network-level failure, e.g. DNS or a dropped connection), which is * typically transient too. * - * This is the single source of truth for HTTP retryability — both - * `TransportError`'s constructor (below) and `HttpTransport`'s own retry - * loop (`src/transport.ts`) call this instead of encoding the rule twice. + * This is the single source of truth for HTTP retryability. `TransportError`'s + * constructor (below) applies it, and `HttpTransport`'s retry loop reads the + * resulting `retryable` flag, so a push rejection that arrives as a 500 is + * not retried either. */ export function isRetryableStatus(status?: number): boolean { return ( @@ -69,9 +70,32 @@ export function isRetryableStatus(status?: number): boolean { ); } +/** + * The Go server's error text for a push it refused deterministically: + * validation (`crdt: change[i]: ...`, `crdt: push exceeds max changes`, + * `crdt: push node_id is required`) or a BeforeInboundChange hook + * (`crdt: inbound change hook: ...`). + */ +const PUSH_REJECTION_TEXT = + /crdt: (inbound change hook:|change\[\d+\]:|push exceeds max changes|push node_id is required)/; + +/** + * Whether a failed push was refused deterministically by the server. + * Nothing from that push was merged, and resending the same changes gets + * the same answer, so it must not be retried as is. Current Go servers + * answer with 422. Older ones answer 500 and WebSocket error frames carry + * no status, so for those the Go error text decides. + */ +export function isPushRejection(err: unknown): boolean { + if (!(err instanceof CRDTError)) return false; + if (err.statusCode === 422) return true; + return (err.statusCode === undefined || err.statusCode === 500) && + PUSH_REJECTION_TEXT.test(err.message); +} + /** * Error thrown by transport operations. - * Extends CRDTError for backward compatibility — existing + * Extends CRDTError for backward compatibility. Existing * `catch (e) { if (e instanceof CRDTError) }` patterns still work. */ export class TransportError extends CRDTError { @@ -83,6 +107,7 @@ export class TransportError extends CRDTError { isRetryableStatus(statusCode) ); this.name = "TransportError"; + if (isPushRejection(this)) this.retryable = false; } } diff --git a/crdt-js/src/index.ts b/crdt-js/src/index.ts index 7de8c47..c74f622 100644 --- a/crdt-js/src/index.ts +++ b/crdt-js/src/index.ts @@ -149,6 +149,7 @@ export { SyncError, PluginError, isRetryableStatus, + isPushRejection, } from "./errors.js"; // Client @@ -168,6 +169,7 @@ export type { StateSnapshot } from "./store.js"; // Sync orchestration export { SyncEngine } from "./sync.js"; +export type { PushRejection } from "./sync.js"; // Undo/Redo export type { UndoEntry } from "./undo.js"; diff --git a/crdt-js/src/merge.ts b/crdt-js/src/merge.ts index bcd81f5..2ea2a35 100644 --- a/crdt-js/src/merge.ts +++ b/crdt-js/src/merge.ts @@ -104,6 +104,117 @@ export function counterValue(state: PNCounterState): number { // --- OR-Set --- +/** + * The key an OR-Set element is stored under: the bytes Go's json.Marshal + * produces for the same value, so a remove matches an element whichever + * side added it. Object fields are sorted, and <, >, &, U+2028 and U+2029 + * are escaped, which JSON.stringify does not do. + */ +export function setElementKey(elem: unknown): string { + return goMarshal(elem); +} + +function goMarshal(v: unknown): string { + if (v === null || v === undefined || typeof v === "function" || typeof v === "symbol") { + return "null"; + } + if (typeof (v as { toJSON?: unknown }).toJSON === "function") { + return goMarshal((v as { toJSON: () => unknown }).toJSON()); + } + if (Array.isArray(v)) { + return "[" + v.map(goMarshal).join(",") + "]"; + } + if (typeof v === "object") { + const obj = v as Record; + const keys = Object.keys(obj) + .filter((k) => obj[k] !== undefined && typeof obj[k] !== "function") + .sort(compareUTF8); + return "{" + keys.map((k) => goEscape(JSON.stringify(k)) + ":" + goMarshal(obj[k])).join(",") + "}"; + } + return goEscape(JSON.stringify(v)); +} + +function goEscape(json: string): string { + return json.replace(/[<>&\u2028\u2029]/g, (ch) => + "\\u" + ch.charCodeAt(0).toString(16).padStart(4, "0")); +} + +/** + * Compare strings by their UTF-8 bytes, the order Go's sort.Strings uses. + * Plain JS comparison orders UTF-16 code units, which puts characters + * above U+FFFF before U+E000 to U+FFFF. + */ +export function compareUTF8(a: string, b: string): number { + const n = Math.min(a.length, b.length); + for (let i = 0; i < n; i++) { + const x = a.charCodeAt(i); + const y = b.charCodeAt(i); + if (x !== y) return utf8Rank(x) - utf8Rank(y); + } + return a.length - b.length; +} + +// Shift surrogates above U+E000 to U+FFFF so code units compare in code +// point order, which is UTF-8 byte order. +function utf8Rank(unit: number): number { + if (unit >= 0xd800 && unit <= 0xdfff) return unit + 0x2000; + if (unit >= 0xe000) return unit - 0x800; + return unit; +} + +const canonicalCache = new WeakMap(); + +/** + * The state with every element key in setElementKey form. States written + * before keys were canonical can hold an element under its JSON.stringify + * key (such as "a | null = null; + for (const key of Object.keys(state.entries)) { + if (!MAY_NEED_REKEY.test(key)) continue; + let canon: string; + try { + canon = setElementKey(JSON.parse(key)); + } catch { + continue; + } + if (canon !== key) (rekey ??= new Map()).set(key, canon); + } + if (!rekey) { + canonicalCache.set(state, state); + return state; + } + + const entries: Record = {}; + for (const [key, tags] of Object.entries(state.entries)) { + const k = rekey.get(key) ?? key; + entries[k] = [...(entries[k] ?? []), ...tags]; + } + for (const k of Object.keys(entries)) entries[k] = deduplicateTags(entries[k]); + const removed: Record = { ...state.removed }; + for (const [old, canon] of rekey) { + for (const t of state.entries[old]) { + if (state.removed[removedKey(old, t)]) removed[removedKey(canon, t)] = true; + } + } + const out = { entries, removed }; + canonicalCache.set(state, out); + canonicalCache.set(out, out); + return out; +} + +// setElementKey output never holds a raw <, > or &, U+2028 or U+2029, and +// only objects can have their fields out of order. +const MAY_NEED_REKEY = new RegExp("[<>&{" + String.fromCharCode(0x2028, 0x2029) + "]"); + /** * Compute the deterministic tag key matching Go's tagKey() from set.go. * Format: "nodeID:HLC{ts: c: node:}" @@ -143,8 +254,10 @@ export function mergeSet( local: ORSetState | null, remote: ORSetState | null ): ORSetState { - if (local == null) return remote!; - if (remote == null) return local; + if (local == null) return canonicalSetState(remote!); + if (remote == null) return canonicalSetState(local); + local = canonicalSetState(local); + remote = canonicalSetState(remote); const merged = newORSetState(); @@ -178,8 +291,9 @@ export function mergeSet( * Port of ORSetState.Elements() from set.go. */ export function setElements(state: ORSetState): unknown[] { + state = canonicalSetState(state); const result: unknown[] = []; - const keys = Object.keys(state.entries).sort(); + const keys = Object.keys(state.entries).sort(compareUTF8); for (const key of keys) { const tags = state.entries[key]; @@ -480,6 +594,21 @@ export function mergeFullFieldState( } } +/** + * The HLC/node pair of whichever of local state or the incoming change is + * newer, used to stamp a field after applying an op. Mirrors Go pickNewer: + * a redelivered older op must not regress the field clock. + */ +function pickNewer( + local: FieldState | null, + change: ChangeRecord +): { hlc: HLC; node_id: string } { + if (local && hlcAfter(local.hlc, change.hlc)) { + return { hlc: local.hlc, node_id: local.node_id }; + } + return { hlc: change.hlc, node_id: change.node_id }; +} + /** * Merge a ChangeRecord into existing FieldState. * Dispatches to the correct merge function based on CRDTType. @@ -526,14 +655,13 @@ export function mergeFieldState( } return { type: "counter", - hlc: change.hlc, - node_id: change.node_id, + ...pickNewer(local, change), counter_state: localCounter, }; } case "set": { - const localSet = local?.set_state ?? newORSetState(); + const localSet = canonicalSetState(local?.set_state ?? newORSetState()); let entries = localSet.entries; let removed = localSet.removed; @@ -542,7 +670,7 @@ export function mergeFieldState( if (op.op === "add") { const newTag: ORSetTag = { node: change.node_id, hlc: change.hlc }; for (const elem of op.elements) { - const key = JSON.stringify(elem); + const key = setElementKey(elem); entries = withEntry(entries, key, withAppended(entries[key], newTag)); } } else if (op.op === "remove") { @@ -551,14 +679,14 @@ export function mergeFieldState( // Exact observed-remove: the op names the tags it saw, scoped // to the elements it removes. for (const elem of op.elements) { - const key = JSON.stringify(elem); + const key = setElementKey(elem); for (const t of op.tags) keys.push(removedKey(key, t)); } } else { // Legacy remove: only tags older than the remove's HLC — // concurrent-or-newer adds survive (add-wins). Matches Go. for (const elem of op.elements) { - const key = JSON.stringify(elem); + const key = setElementKey(elem); for (const t of entries[key] ?? []) { if (hlcAfter(change.hlc, t.hlc)) keys.push(removedKey(key, t)); } @@ -570,8 +698,7 @@ export function mergeFieldState( return { type: "set", - hlc: change.hlc, - node_id: change.node_id, + ...pickNewer(local, change), set_state: entries === localSet.entries && removed === localSet.removed ? localSet : { entries, removed }, @@ -624,8 +751,7 @@ export function mergeFieldState( return { type: "list", - hlc: change.hlc, - node_id: change.node_id, + ...pickNewer(local, change), list_state: nodes === localList.nodes ? localList : { nodes }, }; } @@ -644,8 +770,7 @@ export function mergeFieldState( const prefix = path + "."; return { type: "document", - hlc: change.hlc, - node_id: change.node_id, + ...pickNewer(local, change), doc_state: { fields: withoutKeys( localDoc.fields, @@ -664,16 +789,14 @@ export function mergeFieldState( }; return { type: "document", - hlc: change.hlc, - node_id: change.node_id, + ...pickNewer(local, change), doc_state: mergeDocumentState(localDoc, remoteDoc), }; } } return { type: "document", - hlc: change.hlc, - node_id: change.node_id, + ...pickNewer(local, change), doc_state: localDoc, }; } @@ -683,13 +806,9 @@ export function mergeFieldState( const nextText = change.text_op ? applyTextOpTo(localText, change.text_op, change.node_id, change.hlc) : localText; - // Stamp with whichever of local/change is newer (Go pickNewer - // parity) — a redelivered older op must not regress the field clock. - const keepLocal = local && hlcAfter(local.hlc, change.hlc); return { type: "text", - hlc: keepLocal ? local.hlc : change.hlc, - node_id: keepLocal ? local.node_id : change.node_id, + ...pickNewer(local, change), text_state: nextText, }; } diff --git a/crdt-js/src/store.ts b/crdt-js/src/store.ts index 580eafc..f4e0b1c 100644 --- a/crdt-js/src/store.ts +++ b/crdt-js/src/store.ts @@ -21,6 +21,7 @@ import type { TextState, TextOperation, TextRef, + TextSpan, TextDeltaSegment, } from "./types.js"; import { HybridClock, hlcString, hlcIsZero } from "./hlc.js"; @@ -33,10 +34,14 @@ import { listNodeIds, documentResolve, tagKey, + setElementKey, + canonicalSetState, } from "./merge.js"; import { newTextState, cloneTextState, + textInsertSpan, + textSpanSegments, textInsert, textDeleteOp, textFormat, @@ -48,6 +53,7 @@ import { import { withEntry, withoutKeys } from "./immutable.js"; import { MemoryStorage } from "./storage.js"; import { UndoManager } from "./undo.js"; +import type { UndoEntry } from "./undo.js"; import { PluginManager } from "./plugin.js"; import type { StorePlugin, WriteEvent, MergeEvent } from "./plugin.js"; import { CRDTError, CRDTErrorCode } from "./errors.js"; @@ -102,6 +108,15 @@ function normalizeHLCKeys(doc: DocumentState): void { } } +/** + * A record delete, as opposed to a document path delete (a tombstoned + * document change that carries a path). Mirrors applyChangeInternal. + */ +function isRecordDelete(change: ChangeRecord): boolean { + return change.tombstone === true && + !(change.crdt_type === "document" && change.value !== undefined); +} + export class CRDTStore { private nodeID: string; private clock: HybridClock; @@ -464,9 +479,10 @@ export class CRDTStore { const hlc = this.clock.now(); // Name the observed tags so the remove is exact everywhere (true // observed-remove semantics; receivers don't guess by HLC). - const setState = this.getDoc(table, pk)?.fields[field]?.set_state; + const stored = this.getDoc(table, pk)?.fields[field]?.set_state; + const setState = stored ? canonicalSetState(stored) : undefined; const tags = elements.flatMap( - (elem) => setState?.entries[JSON.stringify(elem)] ?? [] + (elem) => setState?.entries[setElementKey(elem)] ?? [] ); const change: ChangeRecord = { table, @@ -804,6 +820,16 @@ export class CRDTStore { * Delete a document (tombstone). Returns the ChangeRecord for push. */ deleteDocument(table: string, pk: string): ChangeRecord { + const { change, previousState } = this.writeTombstone(table, pk); + this.undoManager.record(change, previousState); + return change; + } + + /** Tombstone a record and queue it, without recording undo history. */ + private writeTombstone( + table: string, + pk: string + ): { change: ChangeRecord; previousState: FieldState | null } { this.assertPendingCapacity(); const hlc = this.clock.now(); const change: ChangeRecord = { @@ -830,12 +856,11 @@ export class CRDTStore { tombstone_hlc: hlc, }); - this.undoManager.record(change, previousState); this.enqueuePending(change); this.persistDocument(table, pk); this.persistPending(); this.notifyListeners(table, pk); - return change; + return { change, previousState }; } // --- Undo / Redo --- @@ -851,66 +876,258 @@ export class CRDTStore { } /** - * Undo the last local mutation by restoring the previous field state. - * Returns true if an undo was performed, false if the stack was empty. + * Undo the last local mutation. The undo is written as new changes with + * fresh clocks and queued for push, so it reaches the server and every + * other replica. Returns true if an undo was performed, false if the + * stack was empty or the entry can no longer be undone (a record delete + * the server has already seen). */ undo(): boolean { - const entry = this.undoManager.undo(); - if (!entry) return false; + return this.step(() => this.undoManager.popUndo(), (e) => this.undoManager.pushUndo(e), + (e) => this.undoManager.pushRedo(e)); + } - const { change, previousState } = entry; + /** + * Redo the last undone mutation. Like undo, it writes new changes with + * fresh clocks, so the redo reaches the server too. Returns true if a + * redo was performed, false if the stack was empty. + */ + redo(): boolean { + return this.step(() => this.undoManager.popRedo(), (e) => this.undoManager.pushRedo(e), + (e) => this.undoManager.pushUndo(e)); + } - const doc = this.docOrEmpty(change.table, change.pk); - if (change.tombstone) { - // previousState === null means the document was NOT tombstoned before. - this.setDocument(change.table, change.pk, { - ...doc, - tombstone: previousState !== null, - tombstone_hlc: previousState?.hlc, - }); - } else if (previousState === null) { - // Field didn't exist before; remove it. - this.setDocument(change.table, change.pk, { - ...doc, - fields: withoutKeys(doc.fields, (k) => k === change.field), - }); - } else { - this.setDocument( - change.table, - change.pk, - this.withField(doc, change.field, previousState) - ); + /** + * Pop an entry, apply its inverse, and push the inverse onto the other + * stack. When the inverse cannot be written (the pending queue is full) + * the entry goes back where it came from. + */ + private step( + pop: () => UndoEntry | null, + putBack: (entry: UndoEntry) => void, + pushOther: (entry: UndoEntry) => void + ): boolean { + const entry = pop(); + if (!entry) return false; + let inverse: UndoEntry | null; + try { + inverse = this.invertEntry(entry); + } catch (err) { + putBack(entry); + throw err; } - - this.persistDocument(change.table, change.pk); - this.notifyListeners(change.table, change.pk); + if (!inverse) return false; + pushOther(inverse); return true; } /** - * Redo the last undone mutation by re-applying the change. - * Returns true if a redo was performed, false if the stack was empty. + * Apply the inverse of an undo entry as new local changes, queued for + * push like any other write, and return the entry that inverts them in + * turn. Returns null when the entry cannot be inverted: a record delete + * the server has already seen stays deleted, because tombstones are + * sticky on the server. */ - redo(): boolean { - const entry = this.undoManager.redo(); - if (!entry) return false; - - const { change } = entry; + private invertEntry(entry: UndoEntry): UndoEntry | null { + const { table, pk } = entry.change; + if (!entry.group && isRecordDelete(entry.change)) { + return entry.undelete ? this.redeleteRecord(entry) : this.undeleteRecord(entry); + } - if (change.tombstone) { - const doc = this.docOrEmpty(change.table, change.pk); - this.setDocument(change.table, change.pk, { - ...doc, - tombstone: true, - tombstone_hlc: change.hlc, - }); - } else { + const changes = this.inverseChanges(entry); + this.assertPendingCapacity(changes.length); + const applied: UndoEntry[] = changes.map((change) => { + const previousState = this.captureFieldState(change.table, change.pk, change.field); this.applyChangeInternal(change); + this.enqueuePending(change); + return { change, previousState, timestamp: Date.now() }; + }); + if (applied.length > 0) { + this.persistDocument(table, pk); + this.persistPending(); + this.notifyListeners(table, pk); } + if (applied.length === 1) return applied[0]; + return { + change: applied[applied.length - 1]?.change ?? entry.change, + previousState: null, + timestamp: Date.now(), + group: applied, + }; + } + + /** + * The changes that take a field from the state after `entry.change` + * back to `entry.previousState`, each with a fresh clock. They are ops + * against what the change itself did, not a reset to the old state, so + * edits other replicas made in the meantime survive the undo. + */ + private inverseChanges(entry: UndoEntry): ChangeRecord[] { + if (entry.group) { + return [...entry.group].reverse().flatMap((e) => this.inverseChanges(e)); + } + const { change, previousState: prev } = entry; + const at = (hlc: HLC, rest: Partial & Pick): ChangeRecord => ({ + table: change.table, + pk: change.pk, + field: change.field, + hlc, + node_id: this.nodeID, + ...rest, + }); + const next = (rest: Partial & Pick): ChangeRecord => + at(this.clock.now(), rest); + + switch (change.crdt_type) { + case "lww": + // There is no field delete on the wire, so undoing a field's first + // write leaves it null. + return [next({ crdt_type: "lww", value: prev?.value ?? null })]; + + case "counter": { + const dInc = (change.counter_delta?.inc ?? 0) - (prev?.counter_state?.inc[change.node_id] ?? 0); + const dDec = (change.counter_delta?.dec ?? 0) - (prev?.counter_state?.dec[change.node_id] ?? 0); + if (dInc === 0 && dDec === 0) return []; + // Totals only grow, so take the change back by adding its delta to + // the other side. + const cur = this.counterTotals(change.table, change.pk, change.field); + return [next({ crdt_type: "counter", counter_delta: { inc: cur.inc + dDec, dec: cur.dec + dInc } })]; + } + + case "set": { + const op = change.set_op; + if (!op) return []; + if (op.op === "add") { + // Remove only the tag this add created: an element that was in the + // set before the add keeps its older tags. + return [next({ + crdt_type: "set", + set_op: { op: "remove", elements: op.elements, tags: [{ node: change.node_id, hlc: change.hlc }] }, + })]; + } + const before = new Set((prev?.set_state ? setElements(prev.set_state) : []).map(setElementKey)); + const back = op.elements.filter((e) => before.has(setElementKey(e))); + return back.length > 0 ? [next({ crdt_type: "set", set_op: { op: "add", elements: back } })] : []; + } + case "list": { + const op = change.list_op; + if (!op) return []; + if (op.op === "insert") { + const id = op.node_id && !hlcIsZero(op.node_id) ? op.node_id : change.hlc; + return [next({ crdt_type: "list", list_op: { op: "delete", node_id: id } })]; + } + if (op.op === "delete" && op.node_id) { + const node = prev?.list_state?.nodes[hlcString(op.node_id)]; + if (!node || node.tombstone) return []; + // A list delete can't be revived, so put the value back as a new + // node right after the deleted one, where it used to sit. + const hlc = this.clock.now(); + return [at(hlc, { + crdt_type: "list", + list_op: { op: "insert", node_id: hlc, parent_id: op.node_id, value: node.value }, + })]; + } + return []; + } + + case "text": { + const op = change.text_op; + if (!op) return []; + if (op.op === "insert") { + const span = textInsertSpan(op, change.hlc); + return span ? [next({ crdt_type: "text", text_op: { op: "delete", spans: [span] } })] : []; + } + const before = prev?.text_state ?? newTextState(); + if (op.op === "delete") { + // Deleted characters stay tombstoned, so the content goes back in + // as new characters chained after the last one deleted. + const segs = textSpanSegments(before, op.spans ?? [], true); + if (segs.length === 0) return []; + const last = segs[segs.length - 1].span; + let ref: TextRef = { origin: last.origin, offset: last.start + last.length - 1 }; + return segs.map((seg) => { + const hlc = this.clock.now(); + const attrs = Object.fromEntries( + Object.entries(seg.attrs).filter(([, v]) => v !== null && v !== undefined)); + const restored = at(hlc, { + crdt_type: "text", + text_op: { + op: "insert", ref, origin: hlc, content: seg.content, + ...(Object.keys(attrs).length > 0 ? { attrs } : {}), + }, + }); + ref = { origin: hlc, offset: Array.from(seg.content).length - 1 }; + return restored; + }); + } + if (op.op === "format") { + // Set each formatted run back to the attribute values it had, + // null where the attribute was not set. + const names = Object.keys(op.attrs ?? {}); + const runs = new Map; spans: TextSpan[] }>(); + for (const seg of textSpanSegments(before, op.spans ?? [], false)) { + const attrs = Object.fromEntries(names.map((k) => [k, seg.attrs[k] ?? null])); + const key = JSON.stringify(attrs); + const run = runs.get(key) ?? { attrs, spans: [] }; + run.spans.push(seg.span); + runs.set(key, run); + } + return [...runs.values()].map((run) => + next({ crdt_type: "text", text_op: { op: "format", spans: run.spans, attrs: run.attrs } })); + } + return []; + } + + case "document": { + const path = (change.value as { path?: string } | undefined)?.path; + if (!path) return []; + const before = prev?.doc_state?.fields ?? {}; + const prefix = path + "."; + const leaves = Object.entries(before).filter(([k]) => k === path || k.startsWith(prefix)); + const setPath = (p: string, v: unknown) => next({ crdt_type: "document", value: { path: p, value: v } }); + if (change.tombstone) return leaves.map(([k, fs]) => setPath(k, fs.value)); + if (before[path]) return [setPath(path, before[path].value)]; + return [ + next({ crdt_type: "document", tombstone: true, value: { path } }), + ...leaves.map(([k, fs]) => setPath(k, fs.value)), + ]; + } + + default: + return []; + } + } + + /** + * Undo a record delete. Only possible while the tombstone is still + * pending: the server keeps a tombstone for good once it has one. + */ + private undeleteRecord(entry: UndoEntry): UndoEntry | null { + const { change, previousState } = entry; + const at = this.pending.indexOf(change); + if (at < 0) return null; + const doc = this.docOrEmpty(change.table, change.pk); + // A newer tombstone from elsewhere is in effect: nothing to restore. + if (previousState === null && doc.tombstone_hlc && hlcString(doc.tombstone_hlc) !== hlcString(change.hlc)) { + return null; + } + this.pending.splice(at, 1); + this.setDocument(change.table, change.pk, { + ...doc, + tombstone: previousState !== null, + tombstone_hlc: previousState?.hlc, + }); this.persistDocument(change.table, change.pk); + this.persistPending(); this.notifyListeners(change.table, change.pk); - return true; + return { change, previousState, timestamp: Date.now(), undelete: true }; + } + + /** Redo a record delete that undo took back. */ + private redeleteRecord(entry: UndoEntry): UndoEntry { + const { change, previousState } = this.writeTombstone(entry.change.table, entry.change.pk); + return { change, previousState, timestamp: Date.now() }; } // --- Sync --- @@ -1014,11 +1231,11 @@ export class CRDTStore { * push call, applyChangeInternal() and undoManager.record() have * already run against live store state. */ - private assertPendingCapacity(): void { + private assertPendingCapacity(count = 1): void { if ( this.throwOnOverflow && this.maxPendingChanges > 0 && - this.pending.length >= this.maxPendingChanges + this.pending.length + count > this.maxPendingChanges ) { throw new CRDTError( `crdt: pending queue full (${this.maxPendingChanges} changes)`, diff --git a/crdt-js/src/sync.ts b/crdt-js/src/sync.ts index c2fb0de..6aadf17 100644 --- a/crdt-js/src/sync.ts +++ b/crdt-js/src/sync.ts @@ -8,6 +8,13 @@ import type { ChangeRecord, HLC, SyncReport } from "./types.js"; import type { CRDTClient } from "./client.js"; import type { CRDTStore } from "./store.js"; +import { isPushRejection } from "./errors.js"; + +/** A pending change the server refused, with the error it answered. */ +export interface PushRejection { + change: ChangeRecord; + error: Error; +} /** * Drives one pull -> apply -> push -> clear cycle, on a timer or on demand. @@ -27,6 +34,7 @@ export class SyncEngine { private inFlight: Promise | null = null; private timer: ReturnType | null = null; private onlineHandler: (() => void) | null = null; + private rejectionHandlers = new Set<(rejections: PushRejection[]) => void>(); constructor( private client: CRDTClient, @@ -38,6 +46,18 @@ export class SyncEngine { return this._lastSyncTime; } + /** + * Notified when the server refuses pending changes (a validation failure + * or a hook rejection). Those changes are dropped from the pending queue, + * because resending them would only be refused again and would hold up + * every change queued behind them. They stay applied locally, so surface + * them to the user. Returns an unsubscribe function. + */ + onPushRejected(handler: (rejections: PushRejection[]) => void): () => void { + this.rejectionHandlers.add(handler); + return () => { this.rejectionHandlers.delete(handler); }; + } + /** Server HLC watermark from the last successful pull. */ get lastPulledHLC(): HLC | null { return this._lastPulledHLC; @@ -92,11 +112,18 @@ export class SyncEngine { // beforePush hook itself. const toPush = plugins.dispatchBeforePush(snapshot); if (toPush && toPush.length > 0) { - const resp = await this.client.push(toPush); - report.pushed = toPush.length; - report.merged = resp.merged; - this.store.clearPendingChanges(snapshot); - plugins.dispatchAfterPush({ pushed: toPush.length, changes: toPush }); + try { + const resp = await this.client.push(toPush); + report.pushed = toPush.length; + report.merged = resp.merged; + this.store.clearPendingChanges(snapshot); + plugins.dispatchAfterPush({ pushed: toPush.length, changes: toPush }); + } catch (err) { + if (!isPushRejection(err)) throw err; + // The server merged nothing from the batch. Find out which + // changes it refuses by pushing them one at a time. + await this.isolateRejected(toPush, snapshot, report); + } } } @@ -104,6 +131,51 @@ export class SyncEngine { return report; } + /** + * Push a refused batch one change at a time. Accepted changes are kept, + * changes refused on their own are dropped and reported. A transient + * failure stops the walk and leaves the rest pending for the next sync. + */ + private async isolateRejected( + toPush: ChangeRecord[], + snapshot: ChangeRecord[], + report: SyncReport + ): Promise { + const accepted: ChangeRecord[] = []; + const rejections: PushRejection[] = []; + let failure: unknown = null; + + for (const change of toPush) { + try { + const resp = await this.client.push([change]); + accepted.push(change); + report.merged += resp.merged; + } catch (err) { + if (!isPushRejection(err)) { + failure = err; + break; + } + rejections.push({ change, error: err as Error }); + } + } + + report.pushed = accepted.length; + report.rejected = rejections.length; + // A clean walk settles the whole snapshot, matching the success path + // (see the beforePush note on this class). After a transient failure + // only what was settled leaves the queue. + this.store.clearPendingChanges( + failure === null ? snapshot : [...accepted, ...rejections.map((r) => r.change)] + ); + if (accepted.length > 0) { + this.store.pluginManager.dispatchAfterPush({ pushed: accepted.length, changes: accepted }); + } + if (rejections.length > 0) { + for (const handler of this.rejectionHandlers) handler(rejections); + } + if (failure !== null) throw failure; + } + /** * Sync periodically, and immediately whenever the environment reports it * is back online. Returns a stop function — call it on unmount. diff --git a/crdt-js/src/text.ts b/crdt-js/src/text.ts index e772fe6..702671c 100644 --- a/crdt-js/src/text.ts +++ b/crdt-js/src/text.ts @@ -519,6 +519,67 @@ export function textFormat( return op; } +// --- Addressing for undo --- + +/** + * The address range an insert op created: where applyTextOp put its + * characters. Undoing the insert tombstones exactly this range, so text + * other replicas typed around it is left alone. + */ +export function textInsertSpan(op: TextOperation, clock: HLC): TextSpan | null { + if (op.op !== "insert" || !op.content) return null; + const origin = isZeroHLC(op.origin) ? clock : op.origin!; + let start = 0; + if (op.ref && !isHeadRef(op.ref) && textOriginKey(op.ref.origin) === textOriginKey(origin)) { + start = op.ref.offset + 1; + } + return { origin, start, length: runes(op.content).length }; +} + +/** One document-order run of characters inside some spans. */ +export interface TextSpanSegment { + span: TextSpan; + content: string; + /** Attribute values on the run, null where an attribute is cleared. */ + attrs: Record; +} + +/** + * The runs of `state` that fall inside `spans`, in document order. With + * `visibleOnly`, tombstoned characters are skipped. Undo reads these from + * the state before an op to put back what a delete removed or what a + * format overwrote. + */ +export function textSpanSegments( + state: TextState, + spans: TextSpan[], + visibleOnly: boolean +): TextSpanSegment[] { + const out: TextSpanSegment[] = []; + for (const seg of textWalk(state)) { + if (visibleOnly && !segVisible(seg)) continue; + const key = textOriginKey(seg.frag.origin); + const segStart = seg.frag.start + seg.from; + const segEnd = seg.frag.start + seg.to; + for (const sp of spans) { + if (textOriginKey(sp.origin) !== key) continue; + const from = Math.max(segStart, sp.start); + const to = Math.min(segEnd, sp.start + sp.length); + if (from >= to) continue; + const attrs: Record = {}; + for (const [k, a] of Object.entries(seg.frag.attrs ?? {})) { + attrs[k] = a.value ?? null; + } + out.push({ + span: { origin: seg.frag.origin, start: from, length: to - from }, + content: substringRunes(seg.frag.content, from - seg.frag.start, to - seg.frag.start), + attrs, + }); + } + } + return out; +} + // --- State-based merge (mirrors Go MergeText) --- function subFragment(f: TextFragment, from: number, to: number): TextFragment { diff --git a/crdt-js/src/transport.ts b/crdt-js/src/transport.ts index 0e7c47a..f674d32 100644 --- a/crdt-js/src/transport.ts +++ b/crdt-js/src/transport.ts @@ -20,7 +20,7 @@ import type { PresenceState, PresenceSnapshot, } from "./types.js"; -import { TransportError, isRetryableStatus } from "./errors.js"; +import { TransportError } from "./errors.js"; import { CRDTStream } from "./stream.js"; import { Backoff } from "./backoff.js"; import type { BackoffOptions } from "./backoff.js"; @@ -150,7 +150,7 @@ export class HttpTransport implements Transport { `CRDT ${path} returned ${response.status}: ${text}`, response.status ); - if (!isRetryableStatus(response.status)) throw err; + if (!err.retryable) throw err; lastError = err; continue; } @@ -158,7 +158,7 @@ export class HttpTransport implements Transport { return (await parseJsonOrEmpty(response)); } catch (err) { // A non-retryable TransportError was thrown deliberately above. - if (err instanceof TransportError && !isRetryableStatus(err.statusCode)) { + if (err instanceof TransportError && !err.retryable) { throw err; } lastError = err; diff --git a/crdt-js/src/types.ts b/crdt-js/src/types.ts index 60fce33..d5bf4d1 100644 --- a/crdt-js/src/types.ts +++ b/crdt-js/src/types.ts @@ -277,6 +277,8 @@ export interface SyncReport { pushed: number; merged: number; conflicts: number; + /** Changes the server refused and the engine dropped (see SyncEngine.onPushRejected). */ + rejected?: number; } // --- Presence Types --- diff --git a/crdt-js/src/undo.ts b/crdt-js/src/undo.ts index 8138a0b..5ee9480 100644 --- a/crdt-js/src/undo.ts +++ b/crdt-js/src/undo.ts @@ -17,6 +17,18 @@ export interface UndoEntry { previousState: FieldState | null; /** Timestamp of the operation (Date.now()). */ timestamp: number; + /** + * Set when the entry stands for several changes applied in order (an + * undo that needed more than one compensating change). Inverting the + * entry inverts each of them, last first. Empty when the inversion had + * nothing to do. + */ + group?: UndoEntry[]; + /** + * Set on the entry an undone record delete leaves on the redo stack: + * redoing it deletes the record again. + */ + undelete?: boolean; } /** @@ -105,6 +117,32 @@ export class UndoManager { return entry; } + /** + * Pop the newest undo entry without touching the redo stack. The store + * uses this with pushRedo() to swap in the compensating entry it built. + */ + popUndo(): UndoEntry | null { + return this.undoStack.pop() ?? null; + } + + /** Pop the newest redo entry without touching the undo stack. */ + popRedo(): UndoEntry | null { + return this.redoStack.pop() ?? null; + } + + /** Push onto the undo stack without clearing redo history. */ + pushUndo(entry: UndoEntry): void { + this.undoStack.push(entry); + if (this.undoStack.length > this.maxHistory) { + this.undoStack.splice(0, this.undoStack.length - this.maxHistory); + } + } + + /** Push onto the redo stack. */ + pushRedo(entry: UndoEntry): void { + this.redoStack.push(entry); + } + /** Clear all undo and redo history. */ clear(): void { this.undoStack = []; diff --git a/crdt/apply.go b/crdt/apply.go index 74b207a..5757a81 100644 --- a/crdt/apply.go +++ b/crdt/apply.go @@ -1,6 +1,7 @@ package crdt import ( + "bytes" "encoding/json" "fmt" ) @@ -105,19 +106,27 @@ func ApplyChange(engine *MergeEngine, local *FieldState, c *ChangeRecord) (*Fiel // setOpState builds the one-op remote OR-Set state for a set change. func setOpState(local *FieldState, c *ChangeRecord) (*FieldState, error) { - var elements []json.RawMessage + var raw []json.RawMessage if len(c.SetOp.Elements) > 0 { - if err := json.Unmarshal(c.SetOp.Elements, &elements); err != nil { + if err := json.Unmarshal(c.SetOp.Elements, &raw); err != nil { return nil, fmt.Errorf("crdt: set_op elements: %w", err) } } + elements := make([]string, len(raw)) + for i, el := range raw { + key, err := canonicalElementKey(el) + if err != nil { + return nil, fmt.Errorf("crdt: set_op element %d: %w", i, err) + } + elements[i] = key + } remote := NewORSetState() switch c.SetOp.Op { case SetOpAdd: tag := Tag{NodeID: c.NodeID, HLC: c.HLC} for _, el := range elements { - remote.Entries[string(el)] = []Tag{tag} + remote.Entries[el] = []Tag{tag} } case SetOpRemove: @@ -127,7 +136,7 @@ func setOpState(local *FieldState, c *ChangeRecord) (*FieldState, error) { // tag; scoping keeps siblings alive). for _, el := range elements { for _, tag := range c.SetOp.Tags { - remote.Removed[removedKey(string(el), tag)] = true + remote.Removed[removedKey(el, tag)] = true } } } else { @@ -137,9 +146,9 @@ func setOpState(local *FieldState, c *ChangeRecord) (*FieldState, error) { localSet := SetFromFieldState(local) if localSet != nil { for _, el := range elements { - for _, tag := range localSet.Entries[string(el)] { + for _, tag := range localSet.Entries[el] { if c.HLC.After(tag.HLC) { - remote.Removed[removedKey(string(el), tag)] = true + remote.Removed[removedKey(el, tag)] = true } } } @@ -152,6 +161,21 @@ func setOpState(local *FieldState, c *ChangeRecord) (*FieldState, error) { return remote.ToFieldState(c.HLC, c.NodeID), nil } +// canonicalElementKey re-encodes an element as it arrived on the wire into +// the key ORSetState.Add would use for the same value (json.Marshal: sorted +// object fields, <, > and & escaped), so one value never splits across two +// keys depending on who serialized it. Numbers keep their literal text, so +// large integers are not rounded through float64. +func canonicalElementKey(raw json.RawMessage) (string, error) { + dec := json.NewDecoder(bytes.NewReader(raw)) + dec.UseNumber() + var v any + if err := dec.Decode(&v); err != nil { + return "", err + } + return marshalElement(v) +} + // listOpState builds the one-op remote RGA state for a list change. func listOpState(c *ChangeRecord) (*FieldState, error) { remote := NewRGAListState() diff --git a/crdt/apply_test.go b/crdt/apply_test.go index 65325cd..2ab4a0d 100644 --- a/crdt/apply_test.go +++ b/crdt/apply_test.go @@ -355,3 +355,91 @@ func TestSyncer_InboundSetOp_Persisted(t *testing.T) { t.Fatalf("persisted state missing removed tag; writes: %v", written) } } + +func TestApplyChange_SetKeysMatchJSONMarshal(t *testing.T) { + // Elements arrive as the sender serialized them. Keying by those raw + // bytes would split one element from the same value added through + // ORSetState.Add, which keys by json.Marshal. + for _, tc := range []struct { + name string + raw string + want any + }{ + {"escaped string", `["a= limit + if !page.full { + return page, nil + } + + edge, err := ms.executor.QueryContext(ctx, fmt.Sprintf( + `SELECT pk_hash, field_name, hlc_ts, hlc_counter, node_id, tombstone, crdt_state + FROM %s + WHERE hlc_ts = $1 AND hlc_counter = $2`, + shadowTable, + ), page.last.Timestamp, page.last.Counter) + if err != nil { + return changePage{}, fmt.Errorf("crdt: read changes at page edge: %w", err) + } + defer edge.Close() + + have := make(map[string]bool) + for i := len(page.cursors) - 1; i >= 0 && !cursorAfter(page.last, page.cursors[i]); i-- { + have[shadowRowKey(page.changes[i].PK, page.changes[i].Field, page.cursors[i].NodeID)] = true + } + if scanErr := scanPageRows(edge, table, &page, have); scanErr != nil { + return changePage{}, scanErr + } + return page, nil +} + +// shadowRowKey identifies a shadow row: its primary key. +func shadowRowKey(pk, field, node string) string { + return pk + "\x00" + field + "\x00" + node +} + +// scanPageRows appends rows to page in order, skipping any row whose key +// (see shadowRowKey) is in skip. +func scanPageRows(rows Rows, table string, page *changePage, skip map[string]bool) error { for rows.Next() { var row MetadataRow if err := rows.Scan( &row.PKHash, &row.FieldName, &row.HLCTS, &row.HLCCount, &row.NodeID, &row.Tombstone, &row.CRDTState, ); err != nil { - return changePage{}, fmt.Errorf("crdt: scan change: %w", err) + return fmt.Errorf("crdt: scan change: %w", err) + } + if skip[shadowRowKey(row.PKHash, row.FieldName, row.NodeID)] { + continue } cursor := rowCursor(&row) page.last = cursor page.changes = append(page.changes, rowChange(table, &row)) page.cursors = append(page.cursors, cursor) } - if err := rows.Err(); err != nil { - return changePage{}, err - } - page.full = len(page.changes) >= limit - return page, nil + return rows.Err() } // rowChange turns one shadow row into the change record a pull delivers. diff --git a/crdt/server.go b/crdt/server.go index 8aff308..25fc787 100644 --- a/crdt/server.go +++ b/crdt/server.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "encoding/json" + "errors" "fmt" "math" "net/http" @@ -32,7 +33,9 @@ type SyncController struct { presenceTTL time.Duration presenceBufferSize int presence *PresenceManager - presenceCh chan PresenceEvent // buffered channel for broadcasting to SSE streams + presenceMu sync.Mutex + presenceSubs map[chan PresenceEvent]struct{} // one buffered channel per stream + presenceLegacy chan PresenceEvent // shared channel behind PresenceChannel // Time-travel configuration (nil when disabled). timeTravel *TimeTravelConfig @@ -77,18 +80,10 @@ func NewSyncController(plugin *Plugin, opts ...SyncControllerOption) *SyncContro if bufSize <= 0 { bufSize = 256 } - c.presenceCh = make(chan PresenceEvent, bufSize) - c.presence = NewPresenceManager(c.presenceTTL, func(event PresenceEvent) { - // Non-blocking send to the broadcast channel. - select { - case c.presenceCh <- event: - default: - c.logger.Warn("crdt: presence event dropped (channel full)", - log.String("topic", event.Topic), - log.String("node_id", event.NodeID), - ) - } - }, c.logger) + c.presenceBufferSize = bufSize + c.presenceSubs = make(map[chan PresenceEvent]struct{}) + c.presenceLegacy = make(chan PresenceEvent, bufSize) + c.presence = NewPresenceManager(c.presenceTTL, c.broadcastPresence, c.logger) } // Initialize room manager if enabled (requires presence). @@ -347,6 +342,29 @@ func applySyncFilter(changes []ChangeRecord, filter *SyncFilter) []ChangeRecord return result } +// ErrPushRejected marks a push the server refused deterministically: it +// failed validation or a BeforeInboundChange hook rejected one of its +// changes. Nothing from the push was merged, and sending the same changes +// again gets the same answer, so clients should not retry it. Test for it +// with errors.Is; the HTTP handlers answer it with 422. +var ErrPushRejected = errors.New("crdt: push rejected") + +// pushRejectedError keeps the original error text and makes errors.Is match +// ErrPushRejected as well as the underlying cause. +type pushRejectedError struct{ err error } + +func (r *pushRejectedError) Error() string { return r.err.Error() } +func (r *pushRejectedError) Unwrap() []error { return []error{r.err, ErrPushRejected} } + +// PushErrorStatus is the HTTP status for an error from HandlePush: 422 for +// a deterministic rejection, 500 for anything else. +func PushErrorStatus(err error) int { + if errors.Is(err, ErrPushRejected) { + return http.StatusUnprocessableEntity + } + return http.StatusInternalServerError +} + // HandlePush processes a push request, merging remote changes locally. // This is the core logic used by both Forge and HTTP handlers. // @@ -365,7 +383,7 @@ func (c *SyncController) HandlePush(ctx context.Context, req *PushRequest) (*Pus if c.metrics != nil { c.metrics.ValidationErrors.Add(1) } - return nil, err + return nil, &pushRejectedError{err} } } @@ -385,7 +403,7 @@ func (c *SyncController) HandlePush(ctx context.Context, req *PushRequest) (*Pus processedChange, err := c.hooks.BeforeInboundChange(ctx, &change) if err != nil { - return nil, fmt.Errorf("crdt: inbound change hook: %w", err) + return nil, &pushRejectedError{fmt.Errorf("crdt: inbound change hook: %w", err)} } if processedChange != nil { processed = append(processed, processedChange) @@ -798,10 +816,60 @@ func (c *SyncController) Logger() log.Logger { return c.logger } -// PresenceChannel returns the channel for receiving presence events to -// broadcast over SSE streams. Returns nil if presence is disabled. +// SubscribePresence returns a channel that receives every presence event +// until ctx is done. Each call gets its own channel, so every stream sees +// every event. Returns nil if presence is disabled. The channel is never +// closed; stop reading it once ctx is done. +func (c *SyncController) SubscribePresence(ctx context.Context) <-chan PresenceEvent { + if c.presence == nil { + return nil + } + ch := c.addPresenceSub() + go func() { + <-ctx.Done() + c.presenceMu.Lock() + delete(c.presenceSubs, ch) + c.presenceMu.Unlock() + }() + return ch +} + +// PresenceChannel returns one channel shared by every caller, so concurrent +// readers split the events between them. Returns nil if presence is disabled. +// +// Deprecated: use SubscribePresence, which gives each stream every event. func (c *SyncController) PresenceChannel() <-chan PresenceEvent { - return c.presenceCh + return c.presenceLegacy +} + +func (c *SyncController) addPresenceSub() chan PresenceEvent { + ch := make(chan PresenceEvent, c.presenceBufferSize) + c.presenceMu.Lock() + c.presenceSubs[ch] = struct{}{} + c.presenceMu.Unlock() + return ch +} + +// broadcastPresence hands event to every subscriber without blocking; a +// subscriber whose buffer is full misses the event. The legacy shared +// channel drops silently once full, since it may have no reader at all. +func (c *SyncController) broadcastPresence(event PresenceEvent) { + select { + case c.presenceLegacy <- event: + default: + } + c.presenceMu.Lock() + defer c.presenceMu.Unlock() + for ch := range c.presenceSubs { + select { + case ch <- event: + default: + c.logger.Warn("crdt: presence event dropped (channel full)", + log.String("topic", event.Topic), + log.String("node_id", event.NodeID), + ) + } + } } // Close cleans up the controller's resources (presence manager, etc.). @@ -868,7 +936,7 @@ func (c *SyncController) httpHandlePush(w http.ResponseWriter, r *http.Request) resp, err := c.HandlePush(r.Context(), &req) if err != nil { - writeError(w, http.StatusInternalServerError, err.Error()) + writeError(w, PushErrorStatus(err), err.Error()) return } diff --git a/crdt/server_test.go b/crdt/server_test.go index 7db27c2..5d3439d 100644 --- a/crdt/server_test.go +++ b/crdt/server_test.go @@ -555,6 +555,17 @@ func (f *shadowFake) query(_ context.Context, query string, args ...any) (Rows, f.mu.Lock() defer f.mu.Unlock() rows := f.tables[f.tableOf(query)] + if strings.Contains(query, "hlc_counter = $2") { + // Every row at exactly one HLC (timestamp, counter), unlimited. + ts, counter := args[0].(int64), args[1].(uint32) + var out []mockRow + for _, r := range rows { + if r.hlcTS == ts && r.hlcCount == counter { + out = append(out, r) + } + } + return &mockRows{rows: out}, nil + } if !strings.Contains(query, "hlc_ts > $1") { return &mockRows{}, nil // ReadState: start every push from empty state } @@ -650,3 +661,116 @@ func (h *pkRejectHook) BeforeInboundChange(_ context.Context, c *ChangeRecord) ( } return c, nil } + +func TestSyncController_SubscribePresence_FansOutToEveryConsumer(t *testing.T) { + ctrl := NewSyncController(newTestPlugin(), WithPresenceEnabled(true)) + defer ctrl.Close() + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + a := ctrl.SubscribePresence(ctx) + b := ctrl.SubscribePresence(ctx) + + _, err := ctrl.HandlePresenceUpdate(ctx, &PresenceUpdate{ + NodeID: "n1", Topic: "room", Data: json.RawMessage(`{"x":1}`), + }) + require.NoError(t, err) + + for name, ch := range map[string]<-chan PresenceEvent{"a": a, "b": b} { + select { + case ev := <-ch: + assert.Equal(t, "n1", ev.NodeID, name) + case <-time.After(time.Second): + t.Fatalf("consumer %s never received the presence event", name) + } + } +} + +func TestNewHTTPHandler_Push_DeterministicRejectionIs422(t *testing.T) { + ok := ChangeRecord{ + Table: "docs", PK: "1", Field: "title", CRDTType: TypeLWW, NodeID: "remote", + HLC: HLC{Timestamp: time.Now().UnixNano(), NodeID: "remote"}, Value: json.RawMessage(`"v"`), + } + unknownType := ok + unknownType.CRDTType = "nope" + + for _, tc := range []struct { + name string + opts []SyncControllerOption + changes []ChangeRecord + wantErr string + }{ + {"hook rejection", []SyncControllerOption{WithControllerSyncHook(&errorInboundHook{})}, []ChangeRecord{ok}, "crdt: inbound change hook: inbound rejected"}, + {"validation", []SyncControllerOption{WithValidation(DefaultValidationConfig())}, []ChangeRecord{unknownType}, "crdt: change[0]: crdt: unknown crdt type: nope"}, + } { + t.Run(tc.name, func(t *testing.T) { + server := httptest.NewServer(NewHTTPHandler(newTestPluginWithChanges(nil), tc.opts...)) + defer server.Close() + + body, _ := json.Marshal(PushRequest{Changes: tc.changes, NodeID: "remote"}) + resp, err := http.Post(server.URL+"/push", "application/json", bytes.NewReader(body)) //nolint:noctx // test + require.NoError(t, err) + defer resp.Body.Close() + + assert.Equal(t, http.StatusUnprocessableEntity, resp.StatusCode) + var errBody map[string]string + require.NoError(t, json.NewDecoder(resp.Body).Decode(&errBody)) + assert.Equal(t, tc.wantErr, errBody["error"]) + }) + } +} + +func TestSyncController_HandlePull_SharedHLCAtPageEdgeNotSkipped(t *testing.T) { + // One write stamps every field (and every row of a bulk statement) with + // the same HLC, so the page can end in the middle of a run of rows that + // share it. The cursor is (timestamp, counter), so resuming after the + // last row returned would skip the rest of the run. + fake := newShadowFake() + rows := lwwRows(DefaultChangesLimit+100, 1_000) + for i := DefaultChangesLimit - 50; i < len(rows); i++ { + rows[i].hlcTS = 999_999 + } + fake.tables["docs"] = rows + ctrl := NewSyncController(fake.plugin()) + + seen := map[string]bool{} + since := HLC{} + for range 5 { + resp, err := ctrl.HandlePull(context.Background(), &PullRequest{ + Tables: []string{"docs"}, Since: since, NodeID: "client", + }) + require.NoError(t, err) + for _, c := range resp.Changes { + seen[c.PK] = true + } + if len(resp.Changes) == 0 { + break + } + since = resp.LatestHLC + } + + assert.Len(t, seen, DefaultChangesLimit+100, "every row sharing the edge HLC must arrive") +} + +func TestSyncController_HandlePull_PageEdgeKeepsEveryNodesRow(t *testing.T) { + // The shadow key is (pk, field, node): two nodes' rows for one field + // can share the edge position, one inside the page and one past it. + fake := newShadowFake() + rows := lwwRows(DefaultChangesLimit+1, 1_000) + inside, past := &rows[DefaultChangesLimit-1], &rows[DefaultChangesLimit] + past.hlcTS = inside.hlcTS + past.pkHash, past.fieldName = inside.pkHash, inside.fieldName + past.nodeID = "b" + fake.tables["docs"] = rows + ctrl := NewSyncController(fake.plugin()) + + resp, err := ctrl.HandlePull(context.Background(), &PullRequest{Tables: []string{"docs"}, NodeID: "client"}) + require.NoError(t, err) + nodes := map[string]bool{} + for _, c := range resp.Changes { + if c.PK == inside.pkHash { + nodes[c.NodeID] = true + } + } + assert.Equal(t, map[string]bool{"a": true, "b": true}, nodes, "both nodes' rows at the edge position") +} diff --git a/crdt/set.go b/crdt/set.go index 37a2dac..6ad18be 100644 --- a/crdt/set.go +++ b/crdt/set.go @@ -3,6 +3,7 @@ package crdt import ( "encoding/json" "sort" + "strings" ) // ORSetState holds the state for an Observed-Remove Set with add-wins @@ -120,11 +121,12 @@ func (s *ORSetState) hasActiveTags(elem string, tags []Tag) bool { // idempotent. func MergeSet(local, remote *ORSetState) *ORSetState { if local == nil { - return remote + return canonicalORSet(remote) } if remote == nil { - return local + return canonicalORSet(local) } + local, remote = canonicalORSet(local), canonicalORSet(remote) merged := NewORSetState() @@ -201,5 +203,72 @@ func SetFromFieldState(fs *FieldState) *ORSetState { if fs.SetState == nil { return NewORSetState() } - return fs.SetState + return canonicalORSet(fs.SetState) +} + +// canonicalORSet returns s with every element key in the form +// ORSetState.Add uses (json.Marshal bytes). States written before keys were +// canonical can hold an element under the raw bytes a client sent, such as +// "a or &, U+2028 or U+2029, and only objects can have +// their fields out of order. +func mayNeedRekey(key string) bool { + return strings.ContainsFunc(key, func(r rune) bool { + switch r { + case '<', '>', '&', '{', 0x2028, 0x2029: + return true + } + return false + }) } diff --git a/crdt/transport_ws.go b/crdt/transport_ws.go index b83bc69..4483078 100644 --- a/crdt/transport_ws.go +++ b/crdt/transport_ws.go @@ -463,6 +463,8 @@ type WebSocketHandler struct { mu sync.Mutex subscribing bool tables []string + since *HLC // cursor from the latest subscribe, consumed by streamLoop + generation uint64 // bumped by subscribe/unsubscribe; frames from older streams are dropped closed atomic.Bool resubscribeCh chan struct{} // signals stream restart on new subscription } @@ -536,8 +538,11 @@ func (h *WebSocketHandler) handleMessage(ctx context.Context, msg WebSocketMessa h.sendResponse(msg.RequestID, WSPushResponse, resp) case WSSubscribe: + // since is optional: without it the first subscribe streams from + // HLC zero and a resubscribe continues from the last change sent. var sub struct { Tables []string `json:"tables"` + Since *HLC `json:"since,omitempty"` } if err := json.Unmarshal(msg.Payload, &sub); err != nil { h.sendError(msg.RequestID, "invalid subscribe request") @@ -545,18 +550,41 @@ func (h *WebSocketHandler) handleMessage(ctx context.Context, msg WebSocketMessa } h.mu.Lock() h.tables = sub.Tables + h.since = sub.Since + h.generation++ if !h.subscribing { h.subscribing = true go h.streamLoop(ctx) } else { - // Signal the existing stream loop to restart with new tables. - select { - case h.resubscribeCh <- struct{}{}: - default: - } + h.signalResubscribe() } h.mu.Unlock() + case WSUnsubscribe: + h.mu.Lock() + h.tables = nil + h.since = nil + h.generation++ + h.signalResubscribe() + h.mu.Unlock() + + case WSPresenceGet: + var req struct { + Topic string `json:"topic"` + } + if len(msg.Payload) > 0 { + if err := json.Unmarshal(msg.Payload, &req); err != nil { + h.sendError(msg.RequestID, "invalid presence get request") + return + } + } + snapshot, err := h.ctrl.HandleGetPresence(ctx, req.Topic) + if err != nil { + h.sendError(msg.RequestID, err.Error()) + return + } + h.sendResponse(msg.RequestID, WSPresenceSnap, snapshot) + case WSPresenceUpdate: if h.ctrl.Presence() == nil { h.sendError(msg.RequestID, "presence not enabled") @@ -583,15 +611,29 @@ func (h *WebSocketHandler) handleMessage(ctx context.Context, msg WebSocketMessa } } +// signalResubscribe wakes streamLoop to pick up a new subscription. +// The caller holds h.mu. +func (h *WebSocketHandler) signalResubscribe() { + select { + case h.resubscribeCh <- struct{}{}: + default: + } +} + func (h *WebSocketHandler) streamLoop(ctx context.Context) { - presenceCh := h.ctrl.PresenceChannel() + presenceCh := h.ctrl.SubscribePresence(ctx) lastHLC := HLC{} for { - // Read current table list. + // Read the current subscription. h.mu.Lock() tables := make([]string, len(h.tables)) copy(tables, h.tables) + if h.since != nil { + lastHLC = *h.since + h.since = nil + } + gen := h.generation h.mu.Unlock() if len(tables) == 0 { @@ -604,8 +646,12 @@ func (h *WebSocketHandler) streamLoop(ctx context.Context) { } } - ch, err := h.ctrl.StreamChangesSince(ctx, tables, lastHLC) + // Each subscription gets its own poller, stopped when the + // subscription is replaced. + streamCtx, cancel := context.WithCancel(ctx) + ch, err := h.ctrl.StreamChangesSince(streamCtx, tables, lastHLC) if err != nil { + cancel() h.logger.Error("crdt: ws stream error", log.String("error", err.Error())) return } @@ -615,20 +661,25 @@ func (h *WebSocketHandler) streamLoop(ctx context.Context) { for !restarted { select { case <-ctx.Done(): + cancel() return case <-h.resubscribeCh: - // Tables changed — restart the stream. restarted = true case changes, ok := <-ch: if !ok { + cancel() return } - for _, change := range changes { - h.sendResponse("", WSChange, change) - if change.HLC.After(lastHLC) { - lastHLC = change.HLC + h.mu.Lock() + if h.generation == gen { + h.sendResponse("", WSChanges, changes) + for _, change := range changes { + if change.HLC.After(lastHLC) { + lastHLC = change.HLC + } } } + h.mu.Unlock() case event, ok := <-presenceCh: if !ok { continue @@ -636,6 +687,7 @@ func (h *WebSocketHandler) streamLoop(ctx context.Context) { h.sendResponse("", WSPresenceEvent, event) } } + cancel() } } diff --git a/crdt/transport_ws_test.go b/crdt/transport_ws_test.go index 80b82ba..cddde51 100644 --- a/crdt/transport_ws_test.go +++ b/crdt/transport_ws_test.go @@ -478,3 +478,104 @@ func TestWebSocketTransport_Options(t *testing.T) { assert.Equal(t, 15*time.Second, transport.pingInterval) assert.Equal(t, []string{"t1", "t2"}, transport.tables) } + +// readWS waits for the next frame the handler wrote, skipping types in skip. +func readWS(t *testing.T, conn *mockWSConn, skip ...WSMessageType) WebSocketMessage { + t.Helper() + deadline := time.After(2 * time.Second) + for { + select { + case raw := <-conn.writeCh: + var msg WebSocketMessage + require.NoError(t, json.Unmarshal(raw, &msg)) + skipped := false + for _, s := range skip { + skipped = skipped || msg.Type == s + } + if !skipped { + return msg + } + case <-deadline: + t.Fatal("timeout waiting for a websocket frame") + } + } +} + +func TestWebSocketHandler_Subscribe_HonoursSinceAndBatches(t *testing.T) { + fake := newShadowFake() + fake.tables["docs"] = lwwRows(5, 100) // hlc_ts 100..104 + ctrl := NewSyncController(fake.plugin(), WithStreamPollInterval(10*time.Millisecond)) + conn := newMockWSConn() + handler := NewWebSocketHandler(ctrl, conn, nil) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + payload, _ := json.Marshal(map[string]any{ + "tables": []string{"docs"}, + "since": HLC{Timestamp: 102, NodeID: "a"}, + }) + handler.handleMessage(ctx, WebSocketMessage{Type: WSSubscribe, Payload: payload}) + + msg := readWS(t, conn) + require.Equal(t, WSChanges, msg.Type, "a poll batch should travel as one changes frame") + var changes []ChangeRecord + require.NoError(t, json.Unmarshal(msg.Payload, &changes)) + require.Len(t, changes, 2, "only rows after the client's cursor") + assert.Equal(t, int64(103), changes[0].HLC.Timestamp) + assert.Equal(t, int64(104), changes[1].HLC.Timestamp) +} + +func TestWebSocketHandler_PresenceGet_AnswersSnapshot(t *testing.T) { + ctrl := NewSyncController(newTestPlugin(), WithPresenceEnabled(true)) + defer ctrl.Close() + conn := newMockWSConn() + handler := NewWebSocketHandler(ctrl, conn, nil) + ctx := context.Background() + + _, err := ctrl.HandlePresenceUpdate(ctx, &PresenceUpdate{ + NodeID: "n1", Topic: "room", Data: json.RawMessage(`{"x":1}`), + }) + require.NoError(t, err) + + payload, _ := json.Marshal(map[string]string{"topic": "room"}) + handler.handleMessage(ctx, WebSocketMessage{Type: WSPresenceGet, Payload: payload, RequestID: "req-p"}) + + msg := readWS(t, conn) + require.Equal(t, WSPresenceSnap, msg.Type) + assert.Equal(t, "req-p", msg.RequestID) + var snap PresenceSnapshot + require.NoError(t, json.Unmarshal(msg.Payload, &snap)) + assert.Equal(t, "room", snap.Topic) + require.Len(t, snap.States, 1) + assert.Equal(t, "n1", snap.States[0].NodeID) +} + +func TestWebSocketHandler_Unsubscribe_StopsStreamWithoutError(t *testing.T) { + fake := newShadowFake() + fake.tables["docs"] = lwwRows(1, 100) + ctrl := NewSyncController(fake.plugin(), WithStreamPollInterval(10*time.Millisecond)) + conn := newMockWSConn() + handler := NewWebSocketHandler(ctrl, conn, nil) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + sub, _ := json.Marshal(map[string]any{"tables": []string{"docs"}}) + handler.handleMessage(ctx, WebSocketMessage{Type: WSSubscribe, Payload: sub}) + first := readWS(t, conn) + require.NotEqual(t, WSError, first.Type) + + handler.handleMessage(ctx, WebSocketMessage{Type: WSUnsubscribe, RequestID: "req-u"}) + // New rows after unsubscribing must not be streamed, and the + // unsubscribe itself is not an error. + fake.mu.Lock() + fake.tables["docs"] = append(fake.tables["docs"], lwwRows(1, 500)...) + fake.mu.Unlock() + + select { + case raw := <-conn.writeCh: + var msg WebSocketMessage + require.NoError(t, json.Unmarshal(raw, &msg)) + t.Fatalf("unexpected frame after unsubscribe: %s %s", msg.Type, msg.Payload) + case <-time.After(150 * time.Millisecond): + } +} diff --git a/docs/content/docs/crdt/api-reference.mdx b/docs/content/docs/crdt/api-reference.mdx index db069d2..43c2df5 100644 --- a/docs/content/docs/crdt/api-reference.mdx +++ b/docs/content/docs/crdt/api-reference.mdx @@ -237,7 +237,8 @@ type Transport interface { | `HandlePresenceUpdate` | `(ctx, *PresenceUpdate) (*PresenceEvent, error)` | Process a presence update | | `HandleGetPresence` | `(ctx, topic string) (*PresenceSnapshot, error)` | Get presence snapshot for a topic | | `Presence` | `() *PresenceManager` | Access the presence manager (nil if disabled) | -| `PresenceChannel` | `() <-chan PresenceEvent` | Broadcast channel for stream integration | +| `SubscribePresence` | `(ctx) <-chan PresenceEvent` | Per-stream channel that gets every presence event until `ctx` is done | +| `PresenceChannel` | `() <-chan PresenceEvent` | Deprecated: one channel shared by every caller, so readers split the events | | `Close` | `()` | Clean up controller resources | ### Controller Options diff --git a/docs/content/docs/crdt/presence.mdx b/docs/content/docs/crdt/presence.mdx index 53c4a5b..9374787 100644 --- a/docs/content/docs/crdt/presence.mdx +++ b/docs/content/docs/crdt/presence.mdx @@ -399,7 +399,8 @@ type PresenceSnapshot struct { | `HandlePresenceUpdate` | `(ctx, *PresenceUpdate) (*PresenceEvent, error)` | Process a presence update (validates, delegates to manager) | | `HandleGetPresence` | `(ctx, topic string) (*PresenceSnapshot, error)` | Get presence snapshot for a topic | | `Presence` | `() *PresenceManager` | Access the underlying manager (nil if disabled) | -| `PresenceChannel` | `() <-chan PresenceEvent` | Broadcast channel for stream integration | +| `SubscribePresence` | `(ctx) <-chan PresenceEvent` | Per-stream channel that gets every presence event until `ctx` is done | +| `PresenceChannel` | `() <-chan PresenceEvent` | Deprecated: one channel shared by every caller, so readers split the events | | `Close` | `()` | Clean up controller resources | ### Helper diff --git a/docs/content/docs/crdt/sync.mdx b/docs/content/docs/crdt/sync.mdx index 38d9cb9..457cc1a 100644 --- a/docs/content/docs/crdt/sync.mdx +++ b/docs/content/docs/crdt/sync.mdx @@ -218,6 +218,8 @@ type SyncHook interface { | `BeforeOutboundChange` | Before pushing a local change to peers | Modified change, `nil` to skip, error to abort | | `BeforeOutboundRead` | Before returning changes in a pull response | Filtered/modified slice | +On the server, `BeforeInboundChange` runs for every change in a push before any of them merges. Returning an error rejects the whole push. Nothing is applied, `HandlePush` returns an error that matches `crdt.ErrPushRejected`, and the HTTP handlers answer 422 with the error text. Validation failures are handled the same way. If you serve pushes from your own handler, `crdt.PushErrorStatus(err)` gives you the status. Returning `nil` is different: only that change is skipped, the rest merge, and the response's `merged` count comes out lower. + ### BaseSyncHook Embed `BaseSyncHook` and override only the methods you need: diff --git a/docs/content/docs/crdt/typescript-client.mdx b/docs/content/docs/crdt/typescript-client.mdx index ed84da2..d185801 100644 --- a/docs/content/docs/crdt/typescript-client.mdx +++ b/docs/content/docs/crdt/typescript-client.mdx @@ -290,6 +290,59 @@ stream.on((event) => { stream.connect(); ``` +## SyncEngine + +`SyncEngine` runs the sync cycle. It pulls, applies what came back, pushes your pending changes, and clears the ones the server took. `CRDTProvider` and `useCRDT` build one for you. Outside React you make it yourself: + +```typescript +import { CRDTClient, CRDTStore, SyncEngine } from "@grove-js/crdt"; + +const client = new CRDTClient({ + baseURL: "https://api.example.com/sync", + nodeID: "browser-1", +}); +const store = new CRDTStore("browser-1", client.clock); +const engine = new SyncEngine(client, store); + +// One cycle. The report counts what moved. +const { pulled, pushed, merged, rejected } = await engine.sync(); + +// On a timer, and whenever the browser comes back online. +const stop = engine.start({ intervalMs: 30_000 }); +``` + +Calling `sync()` while a run is in flight hands you that same run, so two callers never push the same batch twice. `start()` swallows failed runs. The pending queue is still there, so the next tick tries again. + +### Rejected changes + +The server can refuse a push outright, because a `BeforeInboundChange` hook said no or the push failed validation. It answers 422 and merges nothing from it. Sending the same batch again would only get the same answer and hold up everything queued behind it, so the engine doesn't. It pushes that batch again one change at a time, keeps every change the server accepts, and drops the ones it refuses on their own. Subscribe to find out which: + +```typescript +const unsubscribe = engine.onPushRejected((rejections) => { + for (const { change, error } of rejections) { + console.warn(`refused ${change.table}/${change.pk}.${change.field}: ${error.message}`); + } +}); +``` + +A dropped change stays applied in your local store, so tell the user. Older servers answer the same rejections with a 500. The engine recognizes the error text the Go server sends, so those aren't retried either. Anything else (a 500 for some other reason, a 503, a timeout, a dropped connection) still counts as transient: the batch stays pending and goes out on the next sync. + +In React, take the engine from the context: + +```tsx +import { useEffect } from "react"; +import { useCRDTContext } from "@grove-js/crdt/react"; + +function RejectionNotice() { + const { engine } = useCRDTContext(); + useEffect( + () => engine.onPushRejected((rs) => alert(`${rs.length} change(s) were refused`)), + [engine], + ); + return null; +} +``` + ## React Hooks Import from the `./react` sub-export: @@ -662,6 +715,12 @@ console.log(store.canUndo); // boolean console.log(store.canRedo); // boolean ``` +Undo and redo are ordinary writes. Each one goes out as new changes with fresh clocks and sits in the pending queue like any other edit, so the next sync carries it to the server and every other replica sees the same result. A few things follow from that: + +- Undoing a field's first write sets it to `null`, because the wire has no field delete. +- Undoing a record delete only works while the tombstone is still pending. Once the server has it, `undo()` returns `false`, since the server never brings a deleted record back. +- Undo needs room in the pending queue. With `throwOnOverflow` set and the queue full, `undo()` throws `OfflineQueueFull` and keeps the entry for later. + ### React Hook ```tsx diff --git a/docs/content/docs/crdt/websocket.mdx b/docs/content/docs/crdt/websocket.mdx index 74dd512..3037f3d 100644 --- a/docs/content/docs/crdt/websocket.mdx +++ b/docs/content/docs/crdt/websocket.mdx @@ -281,6 +281,20 @@ Client Server Requests and responses are matched by `id`. Events (changes, presence) are pushed without an `id` and dispatched to the appropriate subscriber. +## Subscribing to changes + +Send `subscribe` with the tables you want. If you already hold a cursor, pass it as `since` and the server streams only the changes after it: + +```json +{"type": "subscribe", "payload": {"tables": ["documents"], "since": {"ts": "1706000000000000000", "c": 0, "node": "node-1"}}} +``` + +Leave `since` out and the first subscribe starts from the beginning of each table. A second `subscribe` on the same connection replaces the first. Without `since` it carries on from the last change you were sent. Each poll arrives as one `changes` frame holding an array of change records. + +Send `unsubscribe` to stop the stream. The server doesn't reply, and you can subscribe again on the same connection. + +To see who's in a topic right now, send `presence_get` with `{"topic": "documents:doc-1"}` and a `request_id`. You get back a `presence_snapshot` with the same `request_id`, or an `error` frame if presence is disabled. + ## Error Handling WebSocket errors are delivered as error frames: diff --git a/extension/extension.go b/extension/extension.go index 815519e..6d7af3f 100644 --- a/extension/extension.go +++ b/extension/extension.go @@ -911,7 +911,7 @@ func (c *crdtForgeController) handlePush(ctx forge.Context) error { resp, err := c.ctrl.HandlePush(ctx.Request().Context(), &req) if err != nil { - return ctx.JSON(500, map[string]string{"error": err.Error()}) + return ctx.JSON(crdt.PushErrorStatus(err), map[string]string{"error": err.Error()}) } return ctx.JSON(200, resp) @@ -999,7 +999,7 @@ func (c *crdtForgeController) handleStream(ctx forge.Context, stream forge.Strea } // Get presence channel (nil if presence is disabled). - presenceCh := c.ctrl.PresenceChannel() + presenceCh := c.ctrl.SubscribePresence(stream.Context()) // Send a comment frame whenever the stream has been idle for a whole // interval. The timer restarts after every event, so a busy stream sends