diff --git a/packages/chord/src/sync/client.ts b/packages/chord/src/sync/client.ts index 532c44b40..b20626bd9 100644 --- a/packages/chord/src/sync/client.ts +++ b/packages/chord/src/sync/client.ts @@ -106,7 +106,12 @@ export type UploadResult = * commits would reuse their sequences, so nothing is uploaded. Pull what it lost before the copy commits again; a pull * applies from the local head. */ - | { readonly type: "behind"; readonly head: Seq; readonly syncedHead: Seq }; + | { readonly type: "behind"; readonly head: Seq; readonly syncedHead: Seq } + /** + * The remote refused this device's epoch: another device holds the right to write the session now. Like + * `diverged`, the copy's commits after `at` are a different history, so the host forks the copy (S7). + */ + | { readonly type: "stale_epoch"; readonly at: Seq; readonly epoch: number }; /** * Upload what this copy committed since it last shared the remote's history, from the store's outbox, and trim the @@ -125,9 +130,11 @@ export async function uploadSession(options: { readonly source: Store; readonly store: SessionSyncStore; readonly state: SyncStateStore; + /** The writer's epoch, from whatever grants it (the remote, Radius); sent with every slot. */ + readonly epoch?: number; readonly context: Context; }): Promise { - const { sessionId, device, source, store, state, context } = options; + const { sessionId, device, source, store, state, epoch, context } = options; const outbox = outboxOf(source); let synced = sharedHead(await state.get(sessionId, context), storeHead(source)); let uploaded: Seq | undefined; @@ -155,12 +162,16 @@ export async function uploadSession(options: { const after = synced; let result: PutBatchResult; try { - result = await store.putBatch(sessionId, after, to, device, body, context, archive ? { archive } : {}); + result = await store.putBatch(sessionId, after, to, device, body, context, { + ...(archive ? { archive } : {}), + ...(epoch === undefined ? {} : { epoch }), + }); } finally { // A remote that answered without reading the archive leaves the export holding the commit queue. await close?.(); } // A history that does not end at `after` is another one too: the remote never held this copy's past `after`. + if (result.type === "stale_epoch") return { type: "stale_epoch", at: after, epoch: result.epoch }; if (result.type === "not_head") return { type: "diverged", at: after }; if (result.type === "exists" && result.device !== device) return { type: "diverged", at: after }; synced = result.type === "created" ? to : result.to; diff --git a/packages/chord/src/sync/remote.ts b/packages/chord/src/sync/remote.ts index 773f48dc8..a674d234d 100644 --- a/packages/chord/src/sync/remote.ts +++ b/packages/chord/src/sync/remote.ts @@ -42,7 +42,12 @@ export type PutBatchResult = * No slot starts at `after`, and the history does not end there: a slot there would branch the chain inside a batch * or leave a gap. The uploader's history is not the remote's from `after` on. */ - | { readonly type: "not_head"; readonly head: Seq }; + | { readonly type: "not_head"; readonly head: Seq } + /** + * The upload's `epoch` is not the session's current one: another device was given the right to write the session + * since this one was. Who grants it, and when, is the remote's (RFC 0074's single writer); Chord only reports it. + */ + | { readonly type: "stale_epoch"; readonly epoch: number }; /** A session as the remote's listing describes it. */ export type SessionSummary = { @@ -74,7 +79,11 @@ export interface SessionSyncStore { device: string, body: SyncBody, context: Context, - options?: { readonly archive?: boolean }, + options?: { + readonly archive?: boolean; + /** The writer's epoch: a remote that issues epochs refuses a stale one (`stale_epoch`); others ignore it. */ + readonly epoch?: number; + }, ): Promise; /** A batch's body; `undefined` when its slot is folded or absent. */ getBatch(sessionId: string, after: Seq, context: Context): Promise; diff --git a/packages/chord/test/bench/sync-worker.ts b/packages/chord/test/bench/sync-worker.ts index 3c9eb13dd..5392cb70f 100644 --- a/packages/chord/test/bench/sync-worker.ts +++ b/packages/chord/test/bench/sync-worker.ts @@ -11,6 +11,7 @@ import { BACKGROUND_CONTEXT as context } from "../../src/context/index.ts"; import { int, str } from "../../src/contracts/index.ts"; import { DurableObjectSqliteDatabase } from "../../src/sqlite/cloudflare.ts"; import { sqliteBackend } from "../../src/store/backend.ts"; +import { LiveQueries } from "../../src/store/live-query.ts"; import type { StagedWrite } from "../../src/store/step.ts"; import { type ChordStore, openStore } from "../../src/store/store.ts"; import { defineTable } from "../../src/store/table.ts"; @@ -63,7 +64,7 @@ type Env = { const Rows = defineTable("rows", { version: 1, columns: { id: str(), n: int(), text: str() }, key: ["id"] }); const TABLE = "bench.sync:rows"; const OPEN = { admit: () => true, adopt: () => {} }; -const owner = { label: "bench", scope: {}, assertLive: () => {}, fail: () => {} }; +const owner = { label: "bench", scope: {}, queries: new LiveQueries(), assertLive: () => {}, fail: () => {} }; /** Where a stored body is split into rows: below the object's 2 MB value limit. */ const PART = 512 * 1024; diff --git a/packages/chord/test/sync/sync.test.ts b/packages/chord/test/sync/sync.test.ts index c8c416049..c4058a35c 100644 --- a/packages/chord/test/sync/sync.test.ts +++ b/packages/chord/test/sync/sync.test.ts @@ -396,6 +396,41 @@ describe("syncing a session between devices", () => { expect(await reads(await replica.store())).toEqual(await reads(await laptop.store())); }); + it("passes the writer's epoch to the remote and reports a refused one as stale_epoch", async () => { + const store = await remote(); + const laptop = device("laptop", store); + await laptop.commit(seedBatches().slice(0, 2)); + const shared = (await laptop.store()).lastSeq; + let current = 1; + const epochs: (number | undefined)[] = []; + // A remote that issues epochs (Radius): refuses an upload whose epoch is not the current one. + const issuing = overriding(store, { + putBatch: async (...args: Parameters) => { + const epoch = args[6]?.epoch; + epochs.push(epoch); + if (epoch !== current) return { type: "stale_epoch", epoch: current }; + return store.putBatch(...args); + }, + }); + const source = await laptop.store(); + const upload = (epoch: number) => + uploadSession({ + sessionId: SESSION, + device: "laptop", + source, + store: issuing, + state: laptop.state, + epoch, + context, + }); + expect((await upload(1)).type).toBe("uploaded"); + // The remote gave the session to another writer: this device's epoch is stale. + current = 2; + await laptop.commit([entry(100)]); + expect(await upload(1)).toEqual({ type: "stale_epoch", at: shared, epoch: 2 }); + expect(epochs).toEqual([1, 1]); + }); + it("refuses to upload a copy that lost uploaded commits, and pulls them back", async () => { const store = await remote(); const batches = seedBatches();