From 8de9ba01db1cea1b4f370bddf3fd909d74c55b76 Mon Sep 17 00:00:00 2001 From: Florian <45694132+flo-bit@users.noreply.github.com> Date: Thu, 27 Aug 2026 08:18:55 +0200 Subject: [PATCH] integrate feedback --- packages/contrail-spaces-alpha/README.md | 14 + .../contrail-spaces-alpha/src/protocol.ts | 11 +- packages/contrail-spaces-alpha/src/sync.ts | 197 ++++++++---- packages/contrail-spaces-alpha/src/worker.ts | 13 +- .../contrail-spaces-alpha/tests/sync.test.ts | 300 ++++++++++++++++++ todo/spaces.md | 5 + 6 files changed, 478 insertions(+), 62 deletions(-) create mode 100644 packages/contrail-spaces-alpha/tests/sync.test.ts create mode 100644 todo/spaces.md diff --git a/packages/contrail-spaces-alpha/README.md b/packages/contrail-spaces-alpha/README.md index 05788e7..bc9fb9a 100644 --- a/packages/contrail-spaces-alpha/README.md +++ b/packages/contrail-spaces-alpha/README.md @@ -54,6 +54,20 @@ A Cloudflare deployment needs: Credentials and DPoP private keys are AES-256-GCM encrypted with Space-generation-bound associated data. The D1 projection and backups still contain plaintext private records; this is access control, not end-to-end encryption. +Incremental and recovery work are tuned independently: + +```ts +createSpacesWorker({ + // ... + syncBudget: { + maxIncrementalOperations: 10, // whole-repo operations, including excluded collections + recoveryBatchSize: 50, // in-scope records per staged ingest batch + }, +}); +``` + +The operation endpoint has no collection filter, because metadata from every collection is required to verify the whole-repo commit. Contrail therefore requests only enough operations to enforce the incremental bound, but a PDS response can still include unused values from excluded collections. + ## Deliberately deferred - blobs; diff --git a/packages/contrail-spaces-alpha/src/protocol.ts b/packages/contrail-spaces-alpha/src/protocol.ts index 2d8df01..6e58719 100644 --- a/packages/contrail-spaces-alpha/src/protocol.ts +++ b/packages/contrail-spaces-alpha/src/protocol.ts @@ -279,9 +279,11 @@ async function credentialQuery( service: string, method: string, params: Record, + options: { signal?: AbortSignal } = {}, ): Promise { return assertResponse(await transport.fetch(xrpcUrl(service, method, params), { headers: { accept: "application/json" }, + signal: options.signal, })); } @@ -367,8 +369,15 @@ export function listRepoOps( limit?: number; cursor?: string; }, + options: { signal?: AbortSignal } = {}, ): Promise { - return credentialQuery(transport, writerPds, "com.atproto.space.listRepoOps", input); + return credentialQuery( + transport, + writerPds, + "com.atproto.space.listRepoOps", + input, + options, + ); } export async function getRepoCar( diff --git a/packages/contrail-spaces-alpha/src/sync.ts b/packages/contrail-spaces-alpha/src/sync.ts index 6bfe45c..0873b8f 100644 --- a/packages/contrail-spaces-alpha/src/sync.ts +++ b/packages/contrail-spaces-alpha/src/sync.ts @@ -28,6 +28,7 @@ import { SpaceIdentityResolver, SpaceProtocolError, type ProtocolOptions, + type RepoOp, type SignedCommitInput, } from "./protocol"; import { @@ -59,6 +60,44 @@ export interface SpaceTypeConfig { policy: SpaceUserPolicy; } +export interface SpacesSyncBudget { + /** Maximum whole-repo operations verified from one incremental checkpoint. + * Operations from excluded collections still consume this bound. */ + maxIncrementalOperations: number; + /** Maximum in-scope records projected by one staged recovery batch. */ + recoveryBatchSize: number; +} + +export type SpacesSyncBudgetOptions = Partial; + +export const DEFAULT_SPACES_SYNC_BUDGET = Object.freeze({ + maxIncrementalOperations: 10, + recoveryBatchSize: 50, +}) satisfies SpacesSyncBudget; + +function boundedSyncInteger(value: number, label: string): number { + if (!Number.isSafeInteger(value) || value < 1 || value > 50) { + throw new TypeError(`${label} must be an integer from 1 through 50`); + } + return value; +} + +export function resolveSpacesSyncBudget( + value: SpacesSyncBudgetOptions = {}, +): SpacesSyncBudget { + return { + maxIncrementalOperations: boundedSyncInteger( + value.maxIncrementalOperations ?? + DEFAULT_SPACES_SYNC_BUDGET.maxIncrementalOperations, + "syncBudget.maxIncrementalOperations", + ), + recoveryBatchSize: boundedSyncInteger( + value.recoveryBatchSize ?? DEFAULT_SPACES_SYNC_BUDGET.recoveryBatchSize, + "syncBudget.recoveryBatchSize", + ), + }; +} + export interface SpacesSyncConfig { projection: ContrailConfig; spaceTypes: Record; @@ -67,9 +106,7 @@ export interface SpacesSyncConfig { reconcileIntervalMs?: number; notificationRegistration?: "required" | "best-effort" | "disabled"; registrationRenewalWindowMs?: number; - /** Maximum mutations admitted with one verified incremental checkpoint. - * Larger commits recover through a staged full CAR instead (default 10). */ - maxAtomicMutations?: number; + syncBudget?: SpacesSyncBudgetOptions; /** Maximum complete repo CAR buffered by the Worker (default 16 MiB). */ maxRepoCarBytes?: number; protocol?: ProtocolOptions; @@ -91,6 +128,13 @@ class SyncLeaseLostError extends Error { } } +class SyncDeadlineExceededError extends Error { + constructor() { + super("Space reconciliation deadline reached during incremental sync"); + this.name = "SyncDeadlineExceededError"; + } +} + function bytesEqual(left: Uint8Array, right: Uint8Array): boolean { if (left.byteLength !== right.byteLength) return false; for (let index = 0; index < left.byteLength; index++) { @@ -115,6 +159,7 @@ function collectionAllowed( export class SpacesSyncEngine { readonly identities: SpaceIdentityResolver; private readonly logger: Pick; + private readonly budget: SpacesSyncBudget; constructor( readonly db: Database, @@ -122,6 +167,7 @@ export class SpacesSyncEngine { ) { this.identities = new SpaceIdentityResolver(config.protocol); this.logger = config.logger ?? console; + this.budget = resolveSpacesSyncBudget(config.syncBudget); } private scope(watch: SpaceWatch) { @@ -272,7 +318,7 @@ export class SpacesSyncEngine { if (Date.now() >= deadline) break; const state = local.get(repoDid); if (!state || state.rev !== remote.rev || !bytesEqual(state.commitHash, remote.hash)) { - await this.syncRepo(watch.spaceUri, repoDid); + await this.syncRepo(watch.spaceUri, repoDid, undefined, { deadline }); } } @@ -310,6 +356,7 @@ export class SpacesSyncEngine { spaceUri: string, repoDid: string, advertised?: { rev?: string; hash?: Uint8Array }, + options: { deadline?: number } = {}, ): Promise { const watch = await getSpaceWatch(this.db, spaceUri); if (!watch || watch.status === "hidden") return; @@ -346,9 +393,16 @@ export class SpacesSyncEngine { await this.recoverRepo(watch, repoDid, transport, assertLease); } else { try { - await this.incrementalRepo(watch, state, transport, assertLease); + await this.incrementalRepo( + watch, + state, + transport, + assertLease, + options.deadline, + ); } catch (error) { - if (error instanceof SyncLeaseLostError) throw error; + if (error instanceof SyncLeaseLostError || + error instanceof SyncDeadlineExceededError) throw error; this.logger.warn( `[spaces] incremental sync fell back to recovery for ${spaceUri} ${repoDid}`, error, @@ -386,21 +440,43 @@ export class SpacesSyncEngine { local: SpaceRepoState, transport: SpaceCredentialTransport, assertLease: () => Promise, + deadline?: number, ): Promise { const writerPds = await this.identities.resolvePds(local.repoDid); const state = RepoCommit.fromState(local.ltHash); - const events: IngestEvent[] = []; + const finalProjectionOps = new Map(); + let operationCount = 0; let cursor: string | undefined; let commit: SignedCommit | undefined; do { - const page = await listRepoOps(transport, writerPds, { - space: watch.spaceUri, - repo: local.repoDid, - since: local.rev, - cursor, - limit: 1000, - }); + const remainingMs = deadline === undefined ? undefined : deadline - Date.now(); + if (remainingMs !== undefined && remainingMs <= 0) { + throw new SyncDeadlineExceededError(); + } + const signal = remainingMs === undefined + ? undefined + : AbortSignal.timeout(Math.max(1, remainingMs)); + let page; + try { + page = await listRepoOps(transport, writerPds, { + space: watch.spaceUri, + repo: local.repoDid, + since: local.rev, + cursor, + limit: Math.min( + 1000, + this.budget.maxIncrementalOperations - operationCount + 1, + ), + }, { signal }); + } catch (error) { + if (signal?.aborted) throw new SyncDeadlineExceededError(); + throw error; + } for (const operation of page.ops) { + operationCount++; + if (operationCount > this.budget.maxIncrementalOperations) { + throw new Error("Incremental sync exceeds the bounded operation limit"); + } state.applyOp({ collection: operation.collection, rkey: operation.rkey, @@ -408,46 +484,17 @@ export class SpacesSyncEngine { prev: operation.prev ? parseCid(operation.prev) : null, }); if (!collectionAllowed(this.config, watch, operation.collection)) continue; - if (operation.cid && operation.value === undefined) { - throw new Error("Incremental operation omitted its record value"); - } - const observedAt = Date.now() * 1000; - events.push(createIngestEvent({ - uri: formatSpaceRecordUri({ - spaceUri: watch.spaceUri, - writerDid: local.repoDid, - collection: operation.collection, - rkey: operation.rkey, - }), - did: local.repoDid, - collection: operation.collection, - rkey: operation.rkey, - operation: operation.cid - ? operation.prev - ? "update" - : "create" - : "delete", - cid: operation.cid, - value: operation.value, - timeUs: recordTimeUs( - operation.value, - operation.collection, - this.config.projection, - observedAt, - ), - indexedAt: observedAt, - source: { - id: `space:${watch.spaceUri}:${watch.generation}`, - revision: operation.rev, - time_us: observedAt, - }, - })); - if (events.length > (this.config.maxAtomicMutations ?? 10)) { - throw new Error("Incremental commit exceeds the bounded atomic mutation limit"); - } + finalProjectionOps.set( + `${operation.collection}\0${operation.rkey}`, + { operation, order: operationCount }, + ); } cursor = page.cursor; if (page.commit) commit = asSignedCommit(page.commit); + // A cursor proves that at least one more whole-repo operation remains. + if (cursor && operationCount >= this.budget.maxIncrementalOperations) { + throw new Error("Incremental sync exceeds the bounded operation limit"); + } } while (cursor); if (!commit) throw new Error("Incremental operation log did not reach a commit"); const key = await this.identities.resolveSigningKey(local.repoDid); @@ -468,8 +515,50 @@ export class SpacesSyncEngine { throw new Error("Incremental commit signature or LtHash verification failed"); } } - if (events.length > (this.config.maxAtomicMutations ?? 10)) { - throw new Error("Incremental commit exceeds the bounded atomic mutation limit"); + + // A host can inline only values that are still current when it serves a + // page. Keep applying every operation to RepoCommit above, but project only + // the last operation for each path. Earlier value-less creates/updates are + // thereby omitted only after a later operation confirms they were stale. + const events: IngestEvent[] = []; + const operations = [...finalProjectionOps.values()] + .sort((left, right) => left.order - right.order) + .map(({ operation }) => operation); + for (const operation of operations) { + if (operation.cid && operation.value === undefined) { + throw new Error("Current incremental operation omitted its record value"); + } + const observedAt = Date.now() * 1000; + events.push(createIngestEvent({ + uri: formatSpaceRecordUri({ + spaceUri: watch.spaceUri, + writerDid: local.repoDid, + collection: operation.collection, + rkey: operation.rkey, + }), + did: local.repoDid, + collection: operation.collection, + rkey: operation.rkey, + operation: operation.cid + ? operation.prev + ? "update" + : "create" + : "delete", + cid: operation.cid, + value: operation.value, + timeUs: recordTimeUs( + operation.value, + operation.collection, + this.config.projection, + observedAt, + ), + indexedAt: observedAt, + source: { + id: `space:${watch.spaceUri}:${watch.generation}`, + revision: operation.rev, + time_us: observedAt, + }, + })); } await assertLease(); const checkpoint = saveRepoStateStatement(this.db, { @@ -562,7 +651,7 @@ export class SpacesSyncEngine { }; await assertLease(); await deleteIsolatedPartitionGeneration(this.db, this.config.projection, target); - const chunkSize = this.config.maxAtomicMutations ?? 10; + const chunkSize = this.budget.recoveryBatchSize; for (let offset = 0; offset < events.length; offset += chunkSize) { await assertLease(); const chunk = events.slice(offset, offset + chunkSize); diff --git a/packages/contrail-spaces-alpha/src/worker.ts b/packages/contrail-spaces-alpha/src/worker.ts index 7688200..2304d27 100644 --- a/packages/contrail-spaces-alpha/src/worker.ts +++ b/packages/contrail-spaces-alpha/src/worker.ts @@ -28,7 +28,9 @@ import { type SpaceWatch, } from "./storage"; import { + resolveSpacesSyncBudget, SpacesSyncEngine, + type SpacesSyncBudgetOptions, type SpaceTypeConfig, } from "./sync"; import { parseSpaceUri, spaceProjectionKey } from "./uri"; @@ -151,7 +153,8 @@ export interface SpacesWorkerOptions "required" | "best-effort" | "disabled"); - maxAtomicMutations?: number; + /** Independent whole-repo incremental and staged-recovery work bounds. */ + syncBudget?: SpacesSyncBudgetOptions; maxRepoCarBytes?: number; protocol?: { plcUrl?: string; @@ -593,11 +596,7 @@ export function createSpacesWorker 50)) { - throw new TypeError("maxAtomicMutations must be an integer from 1 through 50"); - } + const syncBudget = resolveSpacesSyncBudget(options.syncBudget); if (options.maxRepoCarBytes !== undefined && (!Number.isSafeInteger(options.maxRepoCarBytes) || options.maxRepoCarBytes < 1)) { throw new TypeError("maxRepoCarBytes must be a positive integer"); @@ -725,7 +724,7 @@ export function createSpacesWorker { + const bytes = (value: Uint8Array) => ({ + $bytes: Buffer.from(value).toString("base64url"), + }); + return { + ver: commit.ver, + rev: commit.rev, + hash: bytes(commit.hash), + ikm: bytes(commit.ikm), + sig: bytes(commit.sig), + mac: bytes(commit.mac), + }; +} + +function transportFor( + handler: (url: URL, init?: RequestInit) => Response | Promise, +): SpaceCredentialTransport { + return { + fetch: (input: string | URL | Request, init?: RequestInit) => + handler(new URL(input instanceof Request ? input.url : input), init), + } as SpaceCredentialTransport; +} + +type IncrementalRepo = ( + watch: SpaceWatch, + local: SpaceRepoState, + transport: SpaceCredentialTransport, + assertLease: () => Promise, + deadline?: number, +) => Promise; + +function incremental(engine: SpacesSyncEngine): IncrementalRepo { + return (engine as unknown as { incrementalRepo: IncrementalRepo }) + .incrementalRepo.bind(engine); +} + +async function engineFor( + syncBudget?: { maxIncrementalOperations?: number; recoveryBatchSize?: number }, +) { + const db = createSqliteDatabase(":memory:"); + await initSpacesStorage(db, projection); + const engine = new SpacesSyncEngine(db, { + projection, + spaceTypes: { + "garden.atmo.circle": { + collections: [COLLECTION], + policy: "member-list", + }, + }, + serviceAudience: "did:web:provider.test#spaces", + credentialEncryptionKey: "unused-in-direct-test", + syncBudget, + }); + vi.spyOn(engine.identities, "resolvePds").mockResolvedValue("https://writer.test"); + return { db, engine }; +} + +async function signingKey() { + const keypair = await Secp256k1PrivateKeyExportable.createKeypair(); + const didKey = await keypair.exportPublicKey("did"); + return { + didKey, + signer: { + jwtAlg: "ES256K", + did: () => didKey, + sign: (value: Uint8Array) => keypair.sign(value), + }, + }; +} + +describe("Spaces incremental synchronization", () => { + it("projects an update without recovering when its superseded create has no value", async () => { + const { db, engine } = await engineFor(); + const { didKey, signer } = await signingKey(); + vi.spyOn(engine.identities, "resolveSigningKey").mockResolvedValue(didKey); + const oldRecord = { text: "old" }; + const currentRecord = { text: "current" }; + const [oldSerialized, currentSerialized] = await Promise.all([ + serializeRecord(COLLECTION, "one", oldRecord), + serializeRecord(COLLECTION, "one", currentRecord), + ]); + const commit = await RepoCommit.fromRecords([currentSerialized]).sign( + { space: SPACE, author: WRITER, rev: "3" }, + signer, + ); + const operations: RepoOp[] = [{ + rev: "2", + collection: COLLECTION, + rkey: "one", + cid: oldSerialized.cid.toString(), + prev: null, + }, { + rev: "3", + collection: COLLECTION, + rkey: "one", + cid: currentSerialized.cid.toString(), + prev: oldSerialized.cid.toString(), + value: currentRecord, + }]; + const transport = transportFor(() => new Response(JSON.stringify({ + ops: operations, + commit: encodedCommit(commit), + }), { headers: { "content-type": "application/json" } })); + + await db.batch([activateIsolatedPartitionStatement(db, { + scope: { kind: "isolated", key: spaceProjectionKey(SPACE, 1) }, + partition: WRITER, + generation: 1, + })]); + await incremental(engine)(watch(), localState(), transport, async () => {}); + + const result = await queryIsolatedRecords(db, projection, { + scope: { kind: "isolated", key: spaceProjectionKey(SPACE, 1) }, + collection: COLLECTION, + }); + expect(result.records).toHaveLength(1); + expect(JSON.parse(result.records[0].record!)).toEqual(currentRecord); + expect((await getRepoState(db, watch(), WRITER))?.rev).toBe("3"); + }); + + it("projects a delete without recovering when its superseded create has no value", async () => { + const { db, engine } = await engineFor(); + const { didKey, signer } = await signingKey(); + vi.spyOn(engine.identities, "resolveSigningKey").mockResolvedValue(didKey); + const created = await serializeRecord(COLLECTION, "one", { text: "temporary" }); + const commit = await new RepoCommit().sign( + { space: SPACE, author: WRITER, rev: "3" }, + signer, + ); + const operations: RepoOp[] = [{ + rev: "2", + collection: COLLECTION, + rkey: "one", + cid: created.cid.toString(), + prev: null, + }, { + rev: "3", + collection: COLLECTION, + rkey: "one", + cid: null, + prev: created.cid.toString(), + }]; + const transport = transportFor(() => new Response(JSON.stringify({ + ops: operations, + commit: encodedCommit(commit), + }), { headers: { "content-type": "application/json" } })); + + await incremental(engine)(watch(), localState(), transport, async () => {}); + + const result = await queryIsolatedRecords(db, projection, { + scope: { kind: "isolated", key: spaceProjectionKey(SPACE, 1) }, + collection: COLLECTION, + }); + expect(result.records).toEqual([]); + expect(await db.prepare( + `SELECT operation FROM isolated_record_versions + WHERE scope_key = ? AND partition_key = ? AND partition_generation = ?`, + ).bind(spaceProjectionKey(SPACE, 1), WRITER, 1).first()).toEqual({ + operation: "delete", + }); + expect((await getRepoState(db, watch(), WRITER))?.rev).toBe("3"); + }); + + it("counts excluded-collection operations against the whole-repo limit", async () => { + const { engine } = await engineFor({ maxIncrementalOperations: 2 }); + let requestedLimit: string | null = null; + const transport = transportFor((url) => { + requestedLimit = url.searchParams.get("limit"); + return new Response(JSON.stringify({ + ops: ["one", "two", "three"].map((rkey, index): RepoOp => ({ + rev: String(index + 2), + collection: EXCLUDED_COLLECTION, + rkey, + cid: null, + prev: null, + })), + }), { headers: { "content-type": "application/json" } }); + }); + + await expect(incremental(engine)( + watch(), + localState(), + transport, + async () => {}, + )).rejects.toThrow(/bounded operation limit/); + expect(requestedLimit).toBe("3"); + }); + + it("stops between operation pages at the reconciliation deadline", async () => { + const { engine } = await engineFor(); + let now = 100; + const nowSpy = vi.spyOn(Date, "now").mockImplementation(() => now); + let requests = 0; + const transport = transportFor(() => { + requests++; + now = 200; + return new Response(JSON.stringify({ ops: [], cursor: "next" }), { + headers: { "content-type": "application/json" }, + }); + }); + + try { + await expect(incremental(engine)( + watch(), + localState(), + transport, + async () => {}, + 150, + )).rejects.toThrow(/deadline reached/); + expect(requests).toBe(1); + } finally { + nowSpy.mockRestore(); + } + }); +}); + +describe("Spaces sync budget", () => { + it("keeps incremental and recovery bounds independent", () => { + expect(resolveSpacesSyncBudget()).toEqual(DEFAULT_SPACES_SYNC_BUDGET); + expect(resolveSpacesSyncBudget({ maxIncrementalOperations: 4 })).toEqual({ + maxIncrementalOperations: 4, + recoveryBatchSize: 50, + }); + expect(resolveSpacesSyncBudget({ recoveryBatchSize: 20 })).toEqual({ + maxIncrementalOperations: 10, + recoveryBatchSize: 20, + }); + expect(() => resolveSpacesSyncBudget({ recoveryBatchSize: 0 })).toThrow( + "syncBudget.recoveryBatchSize must be an integer from 1 through 50", + ); + }); +}); diff --git a/todo/spaces.md b/todo/spaces.md new file mode 100644 index 0000000..a8a0fb0 --- /dev/null +++ b/todo/spaces.md @@ -0,0 +1,5 @@ +- lexicon publishing +- local dev +- different ways to allow access +- [x] client side updates (check for non-cloudflare deployment) +- public and private data mix \ No newline at end of file -- 2.51.2