feat(chord): carry a writer epoch on sync uploads

uploadSession takes an optional epoch and sends it with every slot
(putBatch's options). A remote that issues epochs refuses a stale one
with stale_epoch, which uploadSession reports as
{ type: "stale_epoch", at, epoch }; the host forks the copy, as for
diverged (S7). Remotes that issue no epochs ignore it, and the slot rule
stays as the backstop.

Who issues epochs, and when one device may take over from another, is
left to the remote and the host (RFC 0074, open question 6).

The sync memory bench's table owner gains its live queries, which a
merge of main made required.

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<UploadResult> {
-	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;

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<PutBatchResult>;
 	/** A batch's body; `undefined` when its slot is folded or absent. */
 	getBatch(sessionId: string, after: Seq, context: Context): Promise<SyncBody | undefined>;

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;
 

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<SessionSyncMaintenance["putBatch"]>) => {
+				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();