diff --git a/docs/artifact-sync.md b/docs/artifact-sync.md new file mode 100644 index 0000000000..8a8e0709c7 --- /dev/null +++ b/docs/artifact-sync.md @@ -0,0 +1,17 @@ +# Local artifact sync + +The host saves relay records through `ArtifactStorageService`. Each plugin defines one collection and one fixed query. A Web Lock covers `{viewer,relay,owner,collectionName}`. Its owner cancels reads and waits for commits before it releases the lock. + +The native transport exposes `session.artifactIncremental` when the relay advertises the exact `artifacts.incremental` v1 contract. It covers permanent channels, 100-row pages, 100-ID current reads, an 8 MiB response bound, and a 30-day checkpoint lifetime. The relay prerequisite is [PR 8523](https://github.com/block/buzz/pull/8523). + +The client sends `artifact: "incremental"` requests. WebSocket sync uses `artifacts.sync` and `session.artifacts.sync()` for a complete initial set followed by live changes. Each capability works independently. The saved store uses incremental reads so a restart can resume from its last checkpoint. + +The first read discovers authorized records. Later reads use a checkpoint with a 120-second overlap from the last successful server start time. Only a full paged pass saves progress. Live pushes request current-ID reads through the same collection queue. They do not advance progress or directly write saved rows. The existing outbox acceptance observer also requests a current read. + +The client applies signed upserts and explicit removals. A removal uses the locally observed revision for a conditional write. It retains the body and preferences, and marks the record unavailable. A failed commit prevents progress. Empty replies never establish absence. + +A policy change or an expired checkpoint requires a fresh query plus bounded revalidation of saved IDs. Missing, unmatched, and unauthorized IDs return the same unavailable result. Recovery saves progress only after both steps succeed. It does not return a hidden roster. + +Refresh generations, schema checks, and owner cancellation fence each commit. Failed reads keep saved progress and data. Transient failures get three short retries, then a delay of 60–90 seconds. Invalid replies and capacity failures use that delay directly. Each long retry adds a fresh random delay of less than 30 seconds to the 60-second minimum. A persistent failure exposes `refreshError()`. Successful delta completion clears it. A 60-second repair runs independently of live traffic and respects a pending retry delay. A bounded set holds up to 1,000 live IDs; overflow requests another delta. + +The 120-second overlap assumes bounded database delay. It does not prove recovery from every delayed commit. TTL retirement is outside this contract. Native end-to-end verification needs the compatible relay and client together. diff --git a/src/app/services.ts b/src/app/services.ts index 61540b16bb..6f3855badd 100644 --- a/src/app/services.ts +++ b/src/app/services.ts @@ -1,4 +1,5 @@ import { ArtifactStorageService } from "../features/artifact-storage/service"; +import { provideArtifactStoreSync } from "../features/relay/artifact-store-sync"; import { createAdminAccess } from "../bundled/admin-access"; import { nativeRelayStaff } from "../features/relay-staff/native"; import { createLocalAgentActivity } from "../features/agents/local-activity"; @@ -122,6 +123,7 @@ export function createServices() { }); const relay = communities.relay; const artifactStorage = new ArtifactStorageService(ctx); + provideArtifactStoreSync(ctx, artifactStorage); // Expose v2 only with the complete label/projection/frame/Destination/guard slice. new SoloNavigationService(ctx, relay, plugins); // The one recovery point for federated sign-in; see onSignedIn. diff --git a/src/features/relay/artifact-incremental-reader.test.ts b/src/features/relay/artifact-incremental-reader.test.ts new file mode 100644 index 0000000000..0a92761557 --- /dev/null +++ b/src/features/relay/artifact-incremental-reader.test.ts @@ -0,0 +1,432 @@ +import { expect, it, vi } from "vitest"; +import { + ARTIFACT_INCREMENTAL_LIMITS, + ArtifactIncrementalError, + artifactIncrementalCapability, + artifactIncrementalFailure, + createArtifactIncrementalReader, + type ArtifactIncrementalQuery, + type ArtifactIncrementalRequest, +} from "./artifact-incremental-reader"; +import { hasEventProof } from "./events"; +import { keypair, signed } from "./testing"; + +const artifactId = "11111111-1111-4111-8111-111111111111"; +const channelId = "22222222-2222-4222-8222-222222222222"; +const startedAt = "2026-10-21T22:00:01.123456Z"; +const query = { + artifact: "current", + kinds: [45010], + "#type": ["test.task"], +} as const; +const author = keypair(); +const signal = () => new AbortController().signal; +const tags = [ + ["ar", "1"], + ["d", artifactId], + ["h", channelId], + ["type", "test.task"], + ["op", "create"], +]; +const revision = (eventTags = tags, kind = 45010) => + signed(author, { kind, content: "{}", tags: eventTags }); +const page = (changes: unknown[] = []) => ({ + version: 1, + startedAt, + overlapSeconds: 120, + changes, + nextCursor: null, + checkpoint: "checkpoint-final", +}); + +it("requires each advertised sync limit", () => { + expect( + artifactIncrementalCapability({ + artifacts: { incremental: ARTIFACT_INCREMENTAL_LIMITS }, + }), + ).toBe(true); + for (const key of Object.keys(ARTIFACT_INCREMENTAL_LIMITS)) { + const limits: Record = { ...ARTIFACT_INCREMENTAL_LIMITS }; + delete limits[key]; + expect( + artifactIncrementalCapability({ artifacts: { incremental: limits } }), + ).toBe(false); + } + expect( + artifactIncrementalCapability({ + artifacts: { + incremental: { ...ARTIFACT_INCREMENTAL_LIMITS, overlapSeconds: 60 }, + }, + }), + ).toBe(false); + expect( + artifactIncrementalCapability({ supported_extensions: ["nip-ar"] }), + ).toBe(false); +}); +it("reads signed changes and gives a checkpoint only on the final page", async () => { + const event = revision(); + const request = vi.fn(async (_filter: ArtifactIncrementalRequest) => + Response.json({ + ...page([ + { type: "upsert", artifactId, event }, + { type: "remove", artifactId: channelId }, + { type: "removeChannel", channelId }, + ]), + nextCursor: "page-two", + checkpoint: null, + }), + ); + const reader = createArtifactIncrementalReader(request); + const first = await reader.delta( + query, + { checkpoint: "prior-checkpoint" }, + signal(), + ); + expect(first).toMatchObject({ + startedAt, + overlapSeconds: 120, + nextCursor: "page-two", + checkpoint: null, + }); + expect(first.changes).toHaveLength(3); + const change = first.changes[0]; + expect(change?.type).toBe("upsert"); + if (change?.type === "upsert") expect(hasEventProof(change.event)).toBe(true); + request.mockImplementationOnce(async () => Response.json(page())); + const last = await reader.delta(query, { cursor: "page-two" }, signal()); + expect(last.checkpoint).toBe("checkpoint-final"); + expect(request.mock.calls.map(([filter]) => filter)).toEqual([ + { artifact: "incremental", query, checkpoint: "prior-checkpoint" }, + { + artifact: "incremental", + query, + cursor: "page-two", + }, + ]); +}); +it.each([ + { nextCursor: "more", checkpoint: "early-checkpoint" }, + { nextCursor: null, checkpoint: null }, + { nextCursor: "", checkpoint: null }, + { nextCursor: "bad cursor", checkpoint: null }, + { nextCursor: null, checkpoint: "\n" }, + { nextCursor: null, checkpoint: "x".repeat(4097) }, + { overlapSeconds: 119 }, + { version: 2 }, + { + changes: Array.from({ length: 101 }, () => ({ + type: "remove", + artifactId, + })), + }, +])("rejects invalid page fields %j", async (fields) => { + const reader = createArtifactIncrementalReader(async () => + Response.json({ ...page(), ...fields }), + ); + await expect(reader.delta(query, {}, signal())).rejects.toMatchObject({ + code: "incremental_invalid", + }); +}); +it.each([ + "2026-02-30T22:00:01.123456Z", + "2026-10-21T24:00:01.123456Z", + "2026-10-21T22:60:01.123456Z", + "2026-10-21T22:00:01.Z", + "2026-10-21T22:00:01.1234567890Z", + "2026-10-21T22:00:01.123456", + "2026-10-21T22:00:01.123456+24:00", +])("rejects invalid sync time %s", async (value) => { + const reader = createArtifactIncrementalReader(async () => + Response.json({ ...page(), startedAt: value }), + ); + await expect(reader.delta(query, {}, signal())).rejects.toMatchObject({ + code: "incremental_invalid", + }); +}); +it.each([ + "2026-10-21T22:00:01Z", + "2026-10-21T22:00:01.123Z", + "2026-10-21T22:00:01.123456Z", + "2026-10-21T22:00:01.123456789Z", + "2024-02-29T22:00:01.123456+05:30", +])("accepts relay sync time %s", async (time) => { + const reader = createArtifactIncrementalReader(async () => + Response.json({ ...page(), startedAt: time }), + ); + expect((await reader.delta(query, {}, signal())).startedAt).toBe(time); + const current = createArtifactIncrementalReader(async () => + Response.json({ + version: 1, + startedAt: time, + changes: [{ type: "remove", artifactId }], + }), + ); + expect((await current.current(query, [artifactId], signal())).startedAt).toBe( + time, + ); +}); +it.each( + [ + tags.filter(([name]) => name !== "ar"), + [...tags, ["d", artifactId]], + tags.map((tag) => (tag[0] === "d" ? ["d", channelId] : tag)), + tags.map((tag) => (tag[0] === "h" ? ["h", "invalid"] : tag)), + tags.map((tag) => (tag[0] === "type" ? ["type", "other.type"] : tag)), + tags.map((tag) => (tag[0] === "ar" ? ["ar", "2"] : tag)), + tags.map((tag) => (tag[0] === "op" ? ["op", "delete"] : tag)), + tags.map((tag) => (tag[0] === "op" ? ["op", "unknown"] : tag)), + tags.map((tag) => + tag[0] === "type" ? ["type", "test.task", "extra"] : tag, + ), + ].map((eventTags) => ({ eventTags })), +)( + "rejects signed events with invalid artifact tags %j", + async ({ eventTags }) => { + const reader = createArtifactIncrementalReader(async () => + Response.json( + page([{ type: "upsert", artifactId, event: revision(eventTags) }]), + ), + ); + await expect(reader.delta(query, {}, signal())).rejects.toMatchObject({ + code: "incremental_invalid", + }); + }, +); +it("rejects a wrong event kind or a modified signed body", async () => { + for (const event of [ + revision(tags, 9), + { ...revision(), content: "modified" }, + ]) { + const reader = createArtifactIncrementalReader(async () => + Response.json(page([{ type: "upsert", artifactId, event }])), + ); + await expect(reader.delta(query, {}, signal())).rejects.toMatchObject({ + code: "incremental_invalid", + }); + } +}); +it.each([ + { type: "remove", artifactId: "local-id" }, + { type: "remove", artifactId: "00000000-0000-0000-0000-000000000000" }, + { type: "removeChannel", channelId: "local-channel" }, + { type: "unknown", artifactId }, +])("rejects invalid delta identities %j", async (change) => { + const reader = createArtifactIncrementalReader(async () => + Response.json(page([change])), + ); + await expect(reader.delta(query, {}, signal())).rejects.toMatchObject({ + code: "incremental_invalid", + }); +}); +it("requires one current result for each requested ID and accepts a uniform remove for a local ID", async () => { + const event = revision(); + const request = vi.fn(async (_filter: ArtifactIncrementalRequest) => + Response.json({ + version: 1, + startedAt, + changes: [ + { type: "remove", artifactId: "local-id" }, + { type: "upsert", artifactId, event }, + ], + }), + ); + const rows = await createArtifactIncrementalReader(request).current( + query, + [artifactId, "local-id"], + signal(), + ); + expect(rows.changes).toHaveLength(2); + expect(request.mock.calls[0]?.[0]).toEqual({ + artifact: "incremental", + action: "current", + query, + artifactIds: [artifactId, "local-id"], + }); +}); +it.each( + [ + [], + [ + { type: "remove", artifactId }, + { type: "remove", artifactId }, + ], + [{ type: "remove", artifactId: channelId }], + [{ type: "removeChannel", channelId }], + ].map((rows) => ({ rows })), +)( + "rejects missing, duplicate or unrelated current results %j", + async ({ rows }) => { + const reader = createArtifactIncrementalReader(async () => + Response.json({ version: 1, startedAt, changes: rows }), + ); + await expect( + reader.current(query, [artifactId], signal()), + ).rejects.toMatchObject({ code: "incremental_invalid" }); + }, +); +it.each( + [ + [], + [""], + [artifactId, artifactId], + ["é".repeat(129)], + Array.from({ length: 101 }, (_, i) => `local-${i}`), + ].map((ids) => ({ ids })), +)("rejects invalid current input before a request %j", async ({ ids }) => { + const request = vi.fn(); + await expect( + createArtifactIncrementalReader(request).current(query, ids, signal()), + ).rejects.toMatchObject({ code: "incremental_invalid" }); + expect(request).not.toHaveBeenCalled(); +}); +it("admits 100 bounded current IDs", async () => { + const ids = Array.from({ length: 100 }, (_, i) => `${i}-${"é".repeat(126)}`); + const reader = createArtifactIncrementalReader(async () => + Response.json({ + version: 1, + startedAt, + changes: ids.map((artifactId) => ({ type: "remove", artifactId })), + }), + ); + expect((await reader.current(query, ids, signal())).changes).toHaveLength( + 100, + ); +}); +it.each([ + { ...query, "#type": [] }, + { ...query, "#type": ["one", "two"] }, + { ...query, "#type": [""] }, + { ...query, kinds: [9] }, + { ...query, "#h": [channelId] }, +])("rejects unsupported queries before a request %j", async (value) => { + const request = vi.fn(); + await expect( + createArtifactIncrementalReader(request).delta( + value as ArtifactIncrementalQuery, + {}, + signal(), + ), + ).rejects.toMatchObject({ code: "incremental_invalid" }); + expect(request).not.toHaveBeenCalled(); +}); +it.each([ + { checkpoint: "prior-checkpoint", cursor: "page-two" }, + { checkpoint: "" }, + { cursor: "has space" }, + { checkpoint: "x".repeat(4097) }, +])("rejects invalid input tokens %j", async (position) => { + const request = vi.fn(); + await expect( + createArtifactIncrementalReader(request).delta(query, position, signal()), + ).rejects.toMatchObject({ code: "incremental_invalid" }); + expect(request).not.toHaveBeenCalled(); +}); +it.each([ + "incremental_invalid", + "incremental_limit", + "incremental_reset", + "incremental_unavailable", +] as const)("preserves error code %s without relay text", async (code) => { + const reader = createArtifactIncrementalReader(async () => + Response.json( + { code, reason: "policy_changed", error: "private relay detail" }, + { status: 409 }, + ), + ); + await expect(reader.delta(query, {}, signal())).rejects.toMatchObject({ + code, + reason: "policy_changed", + }); + await expect(reader.delta(query, {}, signal())).rejects.not.toThrow( + "private relay detail", + ); + expect( + artifactIncrementalFailure({ code, reason: "checkpoint_expired" }), + ).toEqual({ + error: "Artifact sync request fails", + code, + reason: "checkpoint_expired", + }); +}); +it("rejects unknown error metadata and hides request failures", async () => { + expect(artifactIncrementalFailure({ code: "unknown" })).toBeUndefined(); + expect( + artifactIncrementalFailure({ + code: "incremental_reset", + reason: "unknown", + }), + ).toBeUndefined(); + const reader = createArtifactIncrementalReader(async () => { + throw new Error("private transport detail"); + }); + await expect(reader.delta(query, {}, signal())).rejects.toBeInstanceOf( + ArtifactIncrementalError, + ); + await expect(reader.delta(query, {}, signal())).rejects.toMatchObject({ + code: "incremental_unavailable", + message: "Artifact sync request fails", + }); +}); +it.each(["not json", new Uint8Array([0xff])])( + "rejects invalid response encoding", + async (body) => { + const reader = createArtifactIncrementalReader( + async () => new Response(body), + ); + await expect(reader.delta(query, {}, signal())).rejects.toMatchObject({ + code: "incremental_invalid", + }); + }, +); +it("bounds response bytes and cancels the stream", async () => { + const cancel = vi.fn(); + const reader = createArtifactIncrementalReader( + async () => + new Response( + new ReadableStream({ + start(controller) { + controller.enqueue( + new Uint8Array(ARTIFACT_INCREMENTAL_LIMITS.responseBytes / 2), + ); + controller.enqueue( + new Uint8Array(ARTIFACT_INCREMENTAL_LIMITS.responseBytes / 2 + 1), + ); + }, + cancel, + }), + ), + ); + await expect(reader.delta(query, {}, signal())).rejects.toMatchObject({ + code: "incremental_limit", + }); + expect(cancel).toHaveBeenCalledOnce(); +}); +it("fences cancellation before requests and during response consumption", async () => { + const stopped = new AbortController(); + stopped.abort(); + const request = vi.fn(); + await expect( + createArtifactIncrementalReader(request).delta(query, {}, stopped.signal), + ).rejects.toMatchObject({ name: "AbortError" }); + expect(request).not.toHaveBeenCalled(); + const controller = new AbortController(); + const cancel = vi.fn(); + const reader = createArtifactIncrementalReader( + async () => + new Response( + new ReadableStream( + { + pull() { + controller.abort(); + }, + cancel, + }, + { highWaterMark: 0 }, + ), + ), + ); + await expect( + reader.delta(query, {}, controller.signal), + ).rejects.toMatchObject({ name: "AbortError" }); + expect(cancel).toHaveBeenCalledOnce(); +}); diff --git a/src/features/relay/artifact-incremental-reader.ts b/src/features/relay/artifact-incremental-reader.ts new file mode 100644 index 0000000000..386f90040c --- /dev/null +++ b/src/features/relay/artifact-incremental-reader.ts @@ -0,0 +1,393 @@ +import { eventDto, type RelayEvent } from "./events"; +import { yieldToHost } from "./yield"; + +export const ARTIFACT_INCREMENTAL_LIMITS = Object.freeze({ + version: 1, + overlapSeconds: 120, + pageRows: 100, + currentRows: 100, + responseBytes: 8_388_608, + permanentChannelsOnly: true, + checkpointLifetimeSeconds: 2_592_000, +}); +export type ArtifactIncrementalQuery = Readonly<{ + artifact: "current"; + kinds: readonly [45010]; + "#type": readonly string[]; +}>; +export type ArtifactIncrementalChange = + | Readonly<{ type: "upsert"; artifactId: string; event: RelayEvent }> + | Readonly<{ type: "remove"; artifactId: string }> + | Readonly<{ type: "removeChannel"; channelId: string }>; +export type ArtifactIncrementalPage = Readonly<{ + startedAt: string; + overlapSeconds: 120; + changes: readonly ArtifactIncrementalChange[]; + nextCursor: string | null; + checkpoint: string | null; +}>; +export type ArtifactCurrentPage = Readonly<{ + startedAt: string; + changes: readonly Exclude< + ArtifactIncrementalChange, + { type: "removeChannel" } + >[]; +}>; +export type ArtifactIncrementalRequest = + | Readonly<{ + artifact: "incremental"; + query: ArtifactIncrementalQuery; + checkpoint?: string; + cursor?: string; + }> + | Readonly<{ + artifact: "incremental"; + action: "current"; + query: ArtifactIncrementalQuery; + artifactIds: readonly string[]; + }>; +export type ArtifactIncrementalReader = { + delta( + query: ArtifactIncrementalQuery, + position: { checkpoint?: string; cursor?: string }, + signal: AbortSignal, + ): Promise; + current( + query: ArtifactIncrementalQuery, + artifactIds: readonly string[], + signal: AbortSignal, + ): Promise; +}; +type ErrorCode = + | "incremental_invalid" + | "incremental_limit" + | "incremental_reset" + | "incremental_unsupported_scope" + | "incremental_unavailable"; +type ResetReason = "policy_changed" | "checkpoint_expired"; +export class ArtifactIncrementalError extends Error { + constructor( + readonly code: ErrorCode, + message: string, + readonly reason?: ResetReason, + readonly status?: number, + ) { + super(message); + } +} +const object = (value: unknown): value is Record => + value !== null && typeof value === "object" && !Array.isArray(value); +const uuid = (value: unknown): value is string => + typeof value === "string" && + /^[0-9a-f]{8}-(?:[0-9a-f]{4}-){3}[0-9a-f]{12}$/.test(value) && + value !== "00000000-0000-0000-0000-000000000000"; +const token = (value: unknown): value is string => + typeof value === "string" && /^[\x21-\x7e]{1,4096}$/.test(value); +const invalid = () => + new ArtifactIncrementalError( + "incremental_invalid", + "Invalid artifact sync response", + ); + +export function artifactIncrementalFailure(value: unknown) { + if ( + !object(value) || + ![ + "incremental_invalid", + "incremental_limit", + "incremental_reset", + "incremental_unsupported_scope", + "incremental_unavailable", + ].includes(String(value.code)) || + (value.reason !== undefined && + value.reason !== "policy_changed" && + value.reason !== "checkpoint_expired") + ) + return; + return { + error: "Artifact sync request fails", + code: value.code as ErrorCode, + ...(value.reason ? { reason: value.reason as ResetReason } : {}), + }; +} +export function artifactIncrementalCapability(info: unknown): boolean { + if ( + !object(info) || + !object(info.artifacts) || + !object(info.artifacts.incremental) + ) + return false; + const sync = info.artifacts.incremental; + return Object.entries(ARTIFACT_INCREMENTAL_LIMITS).every( + ([key, value]) => sync[key] === value, + ); +} +function queryType(query: ArtifactIncrementalQuery): string { + if ( + !object(query) || + Object.keys(query).length !== 3 || + query.artifact !== "current" || + !Array.isArray(query.kinds) || + query.kinds.length !== 1 || + query.kinds[0] !== 45010 || + !Array.isArray(query["#type"]) || + query["#type"].length !== 1 || + typeof query["#type"][0] !== "string" || + !query["#type"][0].length + ) + throw invalid(); + return query["#type"][0]; +} +function timestamp(value: unknown): value is string { + if (typeof value !== "string") return false; + const parts = + /^(\d{4})-(\d{2})-(\d{2})T(\d{2}):(\d{2}):(\d{2})(?:\.\d{1,9})?(?:Z|[+-](\d{2}):(\d{2}))$/.exec( + value, + ); + if (!parts) return false; + const year = Number(parts[1]); + const month = Number(parts[2]); + const day = Number(parts[3]); + const days = [ + 31, + year % 4 === 0 && (year % 100 !== 0 || year % 400 === 0) ? 29 : 28, + 31, + 30, + 31, + 30, + 31, + 31, + 30, + 31, + 30, + 31, + ]; + return ( + month >= 1 && + month <= 12 && + day >= 1 && + day <= (days[month - 1] ?? 0) && + Number(parts[4]) < 24 && + Number(parts[5]) < 60 && + Number(parts[6]) < 60 && + Number(parts[7] ?? 0) < 24 && + Number(parts[8] ?? 0) < 60 + ); +} +async function responseValue(response: Response, signal: AbortSignal) { + if (!response.body) { + signal.throwIfAborted(); + throw invalid(); + } + const reader = response.body.getReader(); + const abort = () => { + void reader.cancel(signal.reason).catch(() => {}); + }; + signal.addEventListener("abort", abort, { once: true }); + const decoder = new TextDecoder("utf-8", { fatal: true }); + let bytes = 0; + let text = ""; + try { + while (true) { + signal.throwIfAborted(); + const { value, done } = await reader.read(); + if (done) { + text += decoder.decode(); + break; + } + bytes += value.byteLength; + if (bytes > ARTIFACT_INCREMENTAL_LIMITS.responseBytes) + throw new ArtifactIncrementalError( + "incremental_limit", + "Artifact sync response exceeds capacity", + ); + text += decoder.decode(value, { stream: true }); + } + signal.throwIfAborted(); + const raw: unknown = JSON.parse(text); + if (!response.ok) { + const failure = artifactIncrementalFailure(raw); + throw failure + ? new ArtifactIncrementalError( + failure.code, + failure.error, + failure.reason, + response.status, + ) + : new ArtifactIncrementalError( + "incremental_unavailable", + "Artifact sync request fails", + ); + } + return raw; + } catch (error) { + signal.throwIfAborted(); + if (error instanceof ArtifactIncrementalError) throw error; + throw invalid(); + } finally { + signal.removeEventListener("abort", abort); + await reader.cancel().catch(() => {}); + reader.releaseLock(); + } +} +async function changes( + raw: unknown, + type: string, + signal: AbortSignal, + currentIds?: ReadonlySet, +): Promise { + if (!Array.isArray(raw) || raw.length > ARTIFACT_INCREMENTAL_LIMITS.pageRows) + throw invalid(); + const result: ArtifactIncrementalChange[] = []; + for (const change of raw) { + signal.throwIfAborted(); + if (!object(change)) throw invalid(); + if (change.type === "removeChannel" && uuid(change.channelId)) { + result.push( + Object.freeze({ type: change.type, channelId: change.channelId }), + ); + } else if ( + change.type === "remove" && + typeof change.artifactId === "string" && + (currentIds ? currentIds.has(change.artifactId) : uuid(change.artifactId)) + ) { + result.push( + Object.freeze({ type: change.type, artifactId: change.artifactId }), + ); + } else if (change.type === "upsert" && uuid(change.artifactId)) { + let event: RelayEvent; + try { + event = eventDto(change.event); + } catch { + throw invalid(); + } + const one = (name: string) => { + const tags = event.tags.filter(([key]) => key === name); + return tags.length === 1 && tags[0]?.length === 2 + ? tags[0][1] + : undefined; + }; + if ( + event.kind !== 45010 || + one("d") !== change.artifactId || + !uuid(one("h")) || + one("type") !== type || + one("ar") !== "1" || + !["create", "update", "move", "restore"].includes(one("op") ?? "") + ) + throw invalid(); + result.push( + Object.freeze({ + type: change.type, + artifactId: change.artifactId, + event, + }), + ); + } else throw invalid(); + if (result.length % 12 === 0) await yieldToHost(); + } + signal.throwIfAborted(); + return Object.freeze(result); +} + +/** Sync covers permanent channels only. The authenticated transport owns POST requests. */ +export function createArtifactIncrementalReader( + request: ( + filter: ArtifactIncrementalRequest, + signal: AbortSignal, + ) => Promise, +): ArtifactIncrementalReader { + const send = async ( + filter: ArtifactIncrementalRequest, + signal: AbortSignal, + ) => { + signal.throwIfAborted(); + let response: Response; + try { + response = await request(filter, signal); + } catch { + signal.throwIfAborted(); + throw new ArtifactIncrementalError( + "incremental_unavailable", + "Artifact sync request fails", + ); + } + const raw = await responseValue(response, signal); + if (!object(raw) || raw.version !== 1 || !timestamp(raw.startedAt)) + throw invalid(); + return raw; + }; + return { + async delta(query, position, signal) { + const type = queryType(query); + if ( + (position.checkpoint !== undefined && position.cursor !== undefined) || + (position.checkpoint !== undefined && !token(position.checkpoint)) || + (position.cursor !== undefined && !token(position.cursor)) + ) + throw invalid(); + const raw = await send( + { + artifact: "incremental", + query, + ...(position.checkpoint !== undefined + ? { checkpoint: position.checkpoint } + : {}), + ...(position.cursor !== undefined ? { cursor: position.cursor } : {}), + }, + signal, + ); + if ( + raw.overlapSeconds !== 120 || + (raw.nextCursor !== null && !token(raw.nextCursor)) || + (raw.nextCursor === null + ? !token(raw.checkpoint) + : raw.checkpoint !== null) + ) + throw invalid(); + return Object.freeze({ + startedAt: raw.startedAt as string, + overlapSeconds: 120, + changes: await changes(raw.changes, type, signal), + nextCursor: raw.nextCursor as string | null, + checkpoint: raw.checkpoint as string | null, + }); + }, + async current(query, artifactIds, signal) { + const type = queryType(query); + if ( + !Array.isArray(artifactIds) || + !artifactIds.length || + artifactIds.length > ARTIFACT_INCREMENTAL_LIMITS.currentRows || + artifactIds.some( + (id) => + typeof id !== "string" || + !id.length || + new TextEncoder().encode(id).byteLength > 256, + ) || + new Set(artifactIds).size !== artifactIds.length + ) + throw invalid(); + const raw = await send( + { artifact: "incremental", action: "current", query, artifactIds }, + signal, + ); + const rows = await changes( + raw.changes, + type, + signal, + new Set(artifactIds), + ); + const wanted = new Set(artifactIds); + for (const row of rows) { + if (row.type === "removeChannel" || !wanted.delete(row.artifactId)) + throw invalid(); + } + if (wanted.size) throw invalid(); + return Object.freeze({ + startedAt: raw.startedAt as string, + changes: rows as ArtifactCurrentPage["changes"], + }); + }, + }; +} diff --git a/src/features/relay/artifact-reconcile.test.ts b/src/features/relay/artifact-reconcile.test.ts new file mode 100644 index 0000000000..3994afd52a --- /dev/null +++ b/src/features/relay/artifact-reconcile.test.ts @@ -0,0 +1,583 @@ +import { expect, it, vi } from "vitest"; +import type { + ArtifactEntry, + HostArtifactCollection, +} from "../artifact-storage/types"; +import { + reconcileArtifactDelta, + reconcileArtifactCurrent, +} from "./artifact-reconcile"; +import { + ArtifactIncrementalError, + createArtifactIncrementalReader, + type ArtifactIncrementalReader, +} from "./artifact-incremental-reader"; +import { keypair, signed } from "./testing"; +import { eventDto } from "./events"; +const channel = "11111111-1111-4111-8111-111111111111"; +const id = "22222222-2222-4222-8222-222222222222"; +const event = eventDto( + signed(keypair(), { + kind: 45010, + content: "{}", + tags: [ + ["ar", "1"], + ["d", id], + ["h", channel], + ["type", "test.task"], + ["op", "create"], + ], + }), +); +const old = JSON.stringify({ protocol: "incremental-v1", checkpoint: "old" }); +function setup(checkpoint: string | null = old) { + const rows = new Map(); + const controller = new AbortController(); + const collection = { + checkpoint: vi.fn(async () => checkpoint), + beginRefresh: vi.fn(async (source: string) => ({ + source, + generation: "generation", + })), + setRefreshError: vi.fn(async () => {}), + get: vi.fn(async (id: string) => rows.get(id)), + knownIds: vi.fn( + async ({ channelId, after, limit }) => { + const ids = [...rows] + .filter(([, row]) => !channelId || row.record.channelId === channelId) + .map(([key]) => key) + .sort() + .filter((key) => after === undefined || key > after); + const page = ids.slice(0, limit); + const next = ids.length > page.length ? page.at(-1) : undefined; + return { + ids: page, + ...(next !== undefined ? { next } : {}), + }; + }, + ), + applyBatch: vi.fn(async (batch) => { + for (const item of batch.records ?? []) + rows.set(item.id, { + record: { + id: item.id, + channelId: item.channelId, + current: item.revision, + availability: { state: "unknown" }, + }, + }); + for (const item of batch.availability ?? []) { + const row = rows.get(item.id); + if ((row?.record.current?.id ?? null) !== item.expectedRevision) + throw new Error("Artifact revision changed"); + if (row) + rows.set(item.id, { + ...row, + record: { ...row.record, availability: item.value }, + }); + } + }), + completeRefresh: vi.fn(async () => {}), + }; + const page = { + version: 1 as const, + startedAt: "2026-10-21T12:00:00.000001Z", + overlapSeconds: 120 as const, + changes: [], + nextCursor: null, + checkpoint: "next", + }; + const reader = { + delta: vi.fn(async () => page), + current: vi.fn( + async (_query, ids) => ({ + version: 1, + startedAt: page.startedAt, + changes: ids.map((artifactId) => ({ type: "remove", artifactId })), + }), + ), + }; + const host = collection as unknown as HostArtifactCollection; + return { + rows, + collection, + reader, + page, + controller, + delta: () => + reconcileArtifactDelta(reader, host, "test.task", controller.signal), + current: (ids: string[]) => + reconcileArtifactCurrent( + reader, + host, + "test.task", + ids, + controller.signal, + ), + }; +} +it("saves delta progress only after all pages commit", async () => { + const h = setup(); + h.reader.delta.mockResolvedValueOnce({ + ...h.page, + changes: [{ type: "upsert", artifactId: id, event }], + nextCursor: "second", + checkpoint: null, + }); + await h.delta(); + expect(h.reader.delta.mock.calls[1]?.[1]).toEqual({ + cursor: "second", + }); + expect(h.collection.applyBatch.mock.invocationCallOrder[0]).toBeLessThan( + h.collection.completeRefresh.mock.invocationCallOrder[0] as number, + ); + expect(h.collection.completeRefresh).toHaveBeenCalledWith( + expect.objectContaining({ generation: "generation" }), + { + expected: old, + next: JSON.stringify({ protocol: "incremental-v1", checkpoint: "next" }), + }, + h.controller.signal, + ); +}); +it("continues an incremental pass through the relay request validator", async () => { + const h = setup(); + const request = vi.fn( + async ( + filter: import("./artifact-incremental-reader").ArtifactIncrementalRequest, + ) => { + if ("checkpoint" in filter && "cursor" in filter) + return Response.json({ code: "incremental_invalid" }, { status: 400 }); + if ("cursor" in filter) + return Response.json({ + ...h.page, + changes: [{ type: "remove", artifactId: id }], + }); + return Response.json({ + ...h.page, + changes: [{ type: "upsert", artifactId: id, event }], + nextCursor: "second", + checkpoint: null, + }); + }, + ); + await reconcileArtifactDelta( + createArtifactIncrementalReader(request), + h.collection as unknown as HostArtifactCollection, + "test.task", + h.controller.signal, + ); + expect(request.mock.calls.map(([filter]) => filter)).toEqual([ + { + artifact: "incremental", + query: { artifact: "current", kinds: [45010], "#type": ["test.task"] }, + checkpoint: "old", + }, + { + artifact: "incremental", + query: { artifact: "current", kinds: [45010], "#type": ["test.task"] }, + cursor: "second", + }, + ]); + expect(h.rows.get(id)?.record.availability.state).toBe("unavailable"); + expect(h.collection.completeRefresh).toHaveBeenCalledOnce(); +}); +it("does not infer absence from an empty delta", async () => { + const h = setup(); + h.rows.set(id, { + record: { + id, + current: { id: "older", content: "saved" }, + availability: { state: "available" }, + }, + }); + await h.delta(); + expect(h.rows.get(id)?.record.availability.state).toBe("available"); + expect(h.reader.current).not.toHaveBeenCalled(); + expect(h.collection.knownIds).not.toHaveBeenCalled(); +}); +it("removes a locally observed revision after missed intermediate changes", async () => { + const h = setup(); + h.rows.set(id, { + record: { + id, + current: { id: "locally-observed", content: "saved" }, + availability: { state: "available" }, + }, + preference: { pinned: false }, + }); + h.reader.delta.mockResolvedValue({ + ...h.page, + changes: [{ type: "remove", artifactId: id }], + }); + await h.delta(); + expect(h.collection.applyBatch).toHaveBeenCalledWith( + expect.objectContaining({ + availability: [ + { + id, + expectedRevision: "locally-observed", + value: { state: "unavailable" }, + }, + ], + }), + h.controller.signal, + ); + expect(h.rows.get(id)).toMatchObject({ + preference: { pinned: false }, + record: { + current: { content: "saved" }, + availability: { state: "unavailable" }, + }, + }); +}); +it.each(["page", "commit"])( + "keeps progress after a failed %s", + async (failure) => { + const h = setup(); + h.reader.delta.mockResolvedValueOnce({ + ...h.page, + changes: [{ type: "upsert", artifactId: id, event }], + nextCursor: "second", + checkpoint: null, + }); + if (failure === "page") + h.reader.delta.mockRejectedValueOnce(new Error("Disconnected")); + else + h.collection.applyBatch.mockRejectedValueOnce( + new Error("Artifact revision changed"), + ); + await expect(h.delta()).rejects.toThrow(); + expect(h.collection.completeRefresh).not.toHaveBeenCalled(); + }, +); +it.each(["policy_changed", "checkpoint_expired"])( + "recovers %s with bootstrap and known-ID reads before progress", + async (reason) => { + const h = setup(); + h.rows.set("imported", { + record: { + id: "imported", + local: { title: "Local" }, + availability: { state: "unknown" }, + }, + }); + h.reader.delta.mockRejectedValueOnce( + new ArtifactIncrementalError( + "incremental_reset", + "Reset", + reason as "policy_changed" | "checkpoint_expired", + ), + ); + await h.delta(); + expect(h.reader.delta.mock.calls[1]?.[1]).toEqual({ + checkpoint: undefined, + cursor: undefined, + }); + expect(h.reader.current.mock.calls[0]?.[1]).toEqual(["imported"]); + expect(h.reader.current.mock.invocationCallOrder[0]).toBeLessThan( + h.collection.completeRefresh.mock.invocationCallOrder[0] as number, + ); + expect(h.rows.get("imported")?.record.availability.state).toBe( + "unavailable", + ); + }, +); +it.each(["success", "bootstrap failure", "revalidation failure"])( + "recovers a rejected saved checkpoint with %s", + async (outcome) => { + const h = setup(); + h.rows.set(id, { record: { id, availability: { state: "available" } } }); + const requests: import("./artifact-incremental-reader").ArtifactIncrementalRequest[] = + []; + const reader = createArtifactIncrementalReader(async (filter) => { + requests.push(filter); + if ("checkpoint" in filter || outcome === "bootstrap failure") + return Response.json({ code: "incremental_invalid" }, { status: 400 }); + if ("action" in filter) { + if (outcome === "revalidation failure") + return Response.json( + { code: "incremental_unavailable" }, + { status: 503 }, + ); + return Response.json({ + ...h.page, + changes: [{ type: "remove", artifactId: id }], + }); + } + return Response.json(h.page); + }); + const pass = reconcileArtifactDelta( + reader, + h.collection as unknown as HostArtifactCollection, + "test.task", + h.controller.signal, + ); + if (outcome === "success") { + await pass; + expect(h.rows.get(id)?.record.availability.state).toBe("unavailable"); + expect(h.collection.completeRefresh).toHaveBeenCalledWith( + expect.anything(), + { + expected: old, + next: JSON.stringify({ + protocol: "incremental-v1", + checkpoint: "next", + }), + }, + h.controller.signal, + ); + } else { + await expect(pass).rejects.toBeInstanceOf(ArtifactIncrementalError); + expect(h.collection.completeRefresh).not.toHaveBeenCalled(); + expect(h.rows.get(id)?.record.availability.state).toBe("available"); + } + expect( + requests.map((request) => + "checkpoint" in request + ? "saved" + : "action" in request + ? "current" + : "fresh", + ), + ).toEqual( + outcome === "bootstrap failure" + ? ["saved", "fresh"] + : ["saved", "fresh", "current"], + ); + }, +); +it.each(["cursor", "response", "request"])( + "does not reset an invalid %s", + async (failure) => { + const h = setup(); + const request = vi.fn( + async ( + filter: import("./artifact-incremental-reader").ArtifactIncrementalRequest, + ) => { + if (failure === "response") return Response.json({ version: 1 }); + if ("checkpoint" in filter) + return Response.json({ + ...h.page, + nextCursor: "second", + checkpoint: null, + }); + return Response.json({ code: "incremental_invalid" }, { status: 400 }); + }, + ); + const reader = + failure === "request" + ? h.reader + : createArtifactIncrementalReader(request); + if (failure === "request") + h.reader.delta.mockRejectedValue( + new ArtifactIncrementalError("incremental_invalid", "Invalid request"), + ); + await expect( + reconcileArtifactDelta( + reader, + h.collection as unknown as HostArtifactCollection, + "test.task", + h.controller.signal, + ), + ).rejects.toMatchObject({ code: "incremental_invalid" }); + expect( + failure === "request" ? h.reader.delta : request, + ).toHaveBeenCalledTimes(failure === "cursor" ? 2 : 1); + expect(h.collection.completeRefresh).not.toHaveBeenCalled(); + }, +); +it("keeps recovery incomplete after failed revalidation", async () => { + const h = setup(null); + h.rows.set(id, { record: { id, availability: { state: "available" } } }); + h.reader.current.mockRejectedValue(new Error("Disconnected")); + await expect(h.delta()).rejects.toThrow("Disconnected"); + expect(h.collection.completeRefresh).not.toHaveBeenCalled(); + expect(h.rows.get(id)?.record.availability.state).toBe("available"); +}); +it("revalidates saved IDs on the first pass and migrates an old checkpoint", async () => { + const h = setup("old-snapshot-token"); + h.rows.set(id, { record: { id, availability: { state: "unknown" } } }); + await h.delta(); + expect(h.reader.delta.mock.calls[0]?.[1].checkpoint).toBeUndefined(); + expect(h.reader.current).toHaveBeenCalledOnce(); +}); +it("keeps a moved artifact when the old channel is removed later", async () => { + const h = setup(); + const oldChannel = "33333333-3333-4333-8333-333333333333"; + h.rows.set(id, { + record: { id, channelId: oldChannel, availability: { state: "available" } }, + }); + h.reader.delta.mockResolvedValue({ + ...h.page, + changes: [ + { type: "upsert", artifactId: id, event }, + { type: "removeChannel", channelId: oldChannel }, + ], + }); + await h.delta(); + expect(h.rows.get(id)?.record).toMatchObject({ + channelId: channel, + availability: { state: "available" }, + }); +}); +it.each([100, 101, 201])( + "removes %s saved channel records in batches of at most 100", + async (count) => { + const h = setup(); + for (let index = 0; index < count; index++) { + const id = `saved-${String(index).padStart(3, "0")}`; + h.rows.set(id, { + record: { + id, + channelId: channel, + current: { id: `revision-${id}`, content: "saved" }, + availability: { state: "available" }, + }, + preference: { pinned: true }, + }); + } + h.reader.delta.mockResolvedValue({ + ...h.page, + changes: [{ type: "removeChannel", channelId: channel }], + }); + await h.delta(); + const batches = h.collection.applyBatch.mock.calls.map(([batch]) => batch); + expect(batches.map((batch) => batch.availability?.length)).toEqual( + count === 100 ? [100] : count === 101 ? [100, 1] : [100, 100, 1], + ); + expect(batches.flatMap((batch) => batch.availability ?? [])).toEqual( + [...h.rows.keys()].map((id) => ({ + id, + expectedRevision: `revision-${id}`, + value: { state: "unavailable" }, + })), + ); + for (const row of h.rows.values()) { + expect(row.record.availability.state).toBe("unavailable"); + expect(row.record.current?.content).toBe("saved"); + expect(row.preference).toEqual({ pinned: true }); + } + expect( + h.collection.applyBatch.mock.invocationCallOrder.at(-1), + ).toBeLessThan( + h.collection.completeRefresh.mock.invocationCallOrder[0] as number, + ); + }, +); +it("keeps progress when a later channel removal batch fails", async () => { + const h = setup(); + for (let index = 0; index < 101; index++) { + const id = `saved-${String(index).padStart(3, "0")}`; + h.rows.set(id, { + record: { id, channelId: channel, availability: { state: "available" } }, + }); + } + h.reader.delta.mockResolvedValue({ + ...h.page, + changes: [{ type: "removeChannel", channelId: channel }], + }); + const apply = h.collection.applyBatch; + h.collection.applyBatch = vi + .fn() + .mockImplementationOnce(apply) + .mockRejectedValueOnce(new Error("Artifact revision changed")); + await expect(h.delta()).rejects.toThrow("Artifact revision changed"); + expect(h.rows.get("saved-099")?.record.availability.state).toBe( + "unavailable", + ); + expect(h.rows.get("saved-100")?.record.availability.state).toBe("available"); + expect(h.collection.completeRefresh).not.toHaveBeenCalled(); + expect(await h.collection.checkpoint()).toBe(old); +}); +it("restores an upsert after removal of its old channel", async () => { + const h = setup(); + h.rows.set(id, { + record: { id, channelId: channel, availability: { state: "available" } }, + }); + h.reader.delta.mockResolvedValue({ + ...h.page, + changes: [ + { type: "removeChannel", channelId: channel }, + { type: "upsert", artifactId: id, event }, + ], + }); + await h.delta(); + expect(h.rows.get(id)?.record.availability.state).toBe("available"); +}); +it("does not advance delta progress or clear its error on a current read", async () => { + const h = setup(); + await h.current([id]); + expect(h.collection.completeRefresh).not.toHaveBeenCalled(); + expect(h.collection.checkpoint).not.toHaveBeenCalled(); +}); +it.each(["delta", "current"] as const)( + "reports unsupported scope on the first %s failure and preserves progress", + async (operation) => { + const h = setup(); + const error = new ArtifactIncrementalError( + "incremental_unsupported_scope", + "Unsupported scope", + ); + h.reader[operation].mockRejectedValue(error); + await expect( + operation === "delta" ? h.delta() : h.current([id]), + ).rejects.toBe(error); + expect(h.reader[operation]).toHaveBeenCalledOnce(); + expect(h.collection.setRefreshError).toHaveBeenCalledExactlyOnceWith( + expect.objectContaining({ generation: "generation" }), + "Could not refresh saved data", + h.controller.signal, + ); + expect(h.collection.completeRefresh).not.toHaveBeenCalled(); + expect(await h.collection.checkpoint()).toBe(old); + }, +); +it("splits a current batch on capacity and rejects an oversized single result", async () => { + const h = setup(); + h.reader.current.mockImplementation(async (_query, ids) => { + if (ids.length > 1) + throw new ArtifactIncrementalError("incremental_limit", "Large"); + return { + version: 1, + startedAt: h.page.startedAt, + changes: [{ type: "remove", artifactId: ids[0] as string }], + }; + }); + await h.current([id, "imported"]); + expect(h.reader.current).toHaveBeenCalledTimes(3); + h.reader.current.mockRejectedValue( + new ArtifactIncrementalError("incremental_limit", "Large"), + ); + await expect(h.current([id])).rejects.toMatchObject({ + code: "incremental_limit", + }); +}); +it("rejects repeated cursors and changed capture times", async () => { + const h = setup(); + h.reader.delta.mockResolvedValue({ + ...h.page, + nextCursor: "same", + checkpoint: null, + }); + await expect(h.delta()).rejects.toThrow("cursor repeats"); + h.reader.delta + .mockReset() + .mockResolvedValueOnce({ + ...h.page, + nextCursor: "second", + checkpoint: null, + }) + .mockResolvedValue({ ...h.page, startedAt: "2026-10-21T12:00:01.000001Z" }); + await expect(h.delta()).rejects.toThrow("capture changes"); + expect(h.collection.completeRefresh).not.toHaveBeenCalled(); +}); +it("cancels before a late local completion", async () => { + const h = setup(); + h.reader.delta.mockImplementation(async () => { + h.controller.abort(); + return h.page; + }); + await expect(h.delta()).rejects.toMatchObject({ name: "AbortError" }); + expect(h.collection.completeRefresh).not.toHaveBeenCalled(); +}); diff --git a/src/features/relay/artifact-reconcile.ts b/src/features/relay/artifact-reconcile.ts new file mode 100644 index 0000000000..b0f6f7c74c --- /dev/null +++ b/src/features/relay/artifact-reconcile.ts @@ -0,0 +1,339 @@ +import type { + HostArtifactCollection, + ReceivedArtifact, +} from "../artifact-storage/types"; +import type { RelayEvent } from "./events"; +import { + ARTIFACT_INCREMENTAL_LIMITS, + ArtifactIncrementalError, + type ArtifactIncrementalChange, + type ArtifactIncrementalReader, +} from "./artifact-incremental-reader"; +import type { ArtifactRefresh } from "../artifact-storage/types"; + +export function receivedArtifact( + event: RelayEvent, + type: string, +): ReceivedArtifact { + const one = (name: string) => { + const tags = event.tags.filter(([key]) => key === name); + if (tags.length !== 1 || tags[0]?.length !== 2) + throw new Error("Invalid artifact envelope"); + return tags[0][1] as string; + }; + const id = one("d"); + const channelId = one("h"); + const op = one("op"); + const previous = event.tags.filter(([key]) => key === "prev"); + if ( + event.kind !== 45010 || + one("ar") !== "1" || + one("type") !== type || + !["create", "update", "move", "delete", "restore"].includes(op) || + (op === "create" + ? previous.length !== 0 + : previous.length !== 1 || + previous[0]?.length !== 2 || + !/^[0-9a-f]{64}$/.test(previous[0]?.[1] ?? "")) + ) + throw new Error("Invalid artifact envelope"); + return { + id, + channelId, + type, + revision: { + id: event.id, + content: event.content, + event, + }, + }; +} + +const queryFor = (type: string) => ({ + artifact: "current" as const, + kinds: [45010] as const, + "#type": [type], +}); +const sourceFor = (type: string) => JSON.stringify(queryFor(type)); +function savedCheckpoint(value: string | null): string | undefined { + if (!value) return; + try { + const saved = JSON.parse(value); + if ( + saved?.protocol === "incremental-v1" && + typeof saved.checkpoint === "string" + ) + return saved.checkpoint; + } catch {} +} +async function applyChanges( + collection: HostArtifactCollection, + refresh: ArtifactRefresh, + changes: readonly ArtifactIncrementalChange[], + type: string, + signal: AbortSignal, +) { + let records: ReceivedArtifact[] = []; + let availability: NonNullable< + Parameters[0]["availability"] + >[number][] = []; + async function flush() { + if (!records.length && !availability.length) return; + await collection.applyBatch( + { source: refresh.source, refresh, records, availability }, + signal, + ); + records = []; + availability = []; + } + async function remove(id: string, channelId?: string) { + const saved = await collection.get(id); + signal.throwIfAborted(); + if ( + !saved || + (channelId !== undefined && saved.record.channelId !== channelId) + ) + return; + availability.push({ + id, + expectedRevision: saved.record.current?.id ?? null, + value: { state: "unavailable" }, + }); + } + const pending = new Set(); + for (const change of changes) { + signal.throwIfAborted(); + if (change.type !== "removeChannel" && pending.has(change.artifactId)) { + await flush(); + pending.clear(); + } + if (change.type !== "removeChannel") pending.add(change.artifactId); + if (change.type === "upsert") { + const record = receivedArtifact(change.event, type); + records.push(record); + availability.push({ + id: record.id, + expectedRevision: record.revision.id, + value: { state: "available" }, + }); + } else if (change.type === "remove") await remove(change.artifactId); + else { + await flush(); + pending.clear(); + let after: string | undefined; + do { + const page = await collection.knownIds({ + channelId: change.channelId, + ...(after !== undefined ? { after } : {}), + limit: 100, + }); + signal.throwIfAborted(); + for (const id of page.ids) await remove(id, change.channelId); + await flush(); + after = page.next; + } while (after !== undefined); + } + } + await flush(); +} +async function currentBatch( + reader: ArtifactIncrementalReader, + collection: HostArtifactCollection, + refresh: ArtifactRefresh, + type: string, + ids: readonly string[], + signal: AbortSignal, +): Promise { + try { + const result = await reader.current(queryFor(type), ids, signal); + signal.throwIfAborted(); + await applyChanges(collection, refresh, result.changes, type, signal); + } catch (error) { + if ( + !(error instanceof ArtifactIncrementalError) || + error.code !== "incremental_limit" || + ids.length === 1 + ) + throw error; + const middle = Math.ceil(ids.length / 2); + await currentBatch( + reader, + collection, + refresh, + type, + ids.slice(0, middle), + signal, + ); + await currentBatch( + reader, + collection, + refresh, + type, + ids.slice(middle), + signal, + ); + } +} +async function revalidateKnown( + reader: ArtifactIncrementalReader, + collection: HostArtifactCollection, + refresh: ArtifactRefresh, + type: string, + signal: AbortSignal, +) { + let after: string | undefined; + do { + const page = await collection.knownIds({ + ...(after !== undefined ? { after } : {}), + limit: ARTIFACT_INCREMENTAL_LIMITS.currentRows, + }); + signal.throwIfAborted(); + if (page.ids.length) + await currentBatch(reader, collection, refresh, type, page.ids, signal); + after = page.next; + } while (after !== undefined); +} +async function reportFailure( + collection: HostArtifactCollection, + refresh: ArtifactRefresh, + error: unknown, + signal: AbortSignal, + requested: boolean, +) { + const permanent = + error instanceof ArtifactIncrementalError && + [ + "incremental_limit", + "incremental_invalid", + "incremental_unsupported_scope", + ].includes(error.code); + if (!signal.aborted && (requested || permanent)) + await collection.setRefreshError( + refresh, + error instanceof ArtifactIncrementalError && + error.code === "incremental_limit" + ? "The relay limit prevents a refresh of saved data" + : "Could not refresh saved data", + signal, + ); +} +export async function reconcileArtifactCurrent( + reader: ArtifactIncrementalReader, + collection: HostArtifactCollection, + type: string, + ids: readonly string[], + signal: AbortSignal, + options: { reportFailure?: boolean } = {}, +) { + const refresh = await collection.beginRefresh(sourceFor(type)); + try { + for ( + let offset = 0; + offset < ids.length; + offset += ARTIFACT_INCREMENTAL_LIMITS.currentRows + ) + await currentBatch( + reader, + collection, + refresh, + type, + ids.slice(offset, offset + ARTIFACT_INCREMENTAL_LIMITS.currentRows), + signal, + ); + signal.throwIfAborted(); + } catch (error) { + await reportFailure( + collection, + refresh, + error, + signal, + options.reportFailure === true, + ); + throw error; + } +} +export async function reconcileArtifactDelta( + reader: ArtifactIncrementalReader, + collection: HostArtifactCollection, + type: string, + signal: AbortSignal, + options: { reportFailure?: boolean } = {}, +) { + const expected = await collection.checkpoint(); + signal.throwIfAborted(); + const refresh = await collection.beginRefresh(sourceFor(type)); + let checkpoint = savedCheckpoint(expected); + let recovery = checkpoint === undefined; + try { + let next: string; + while (true) { + let cursor: string | undefined; + try { + let startedAt: string | undefined; + const cursors = new Set(); + while (true) { + const page = await reader.delta( + queryFor(type), + cursor !== undefined + ? { cursor } + : checkpoint !== undefined + ? { checkpoint } + : {}, + signal, + ); + signal.throwIfAborted(); + if (startedAt !== undefined && page.startedAt !== startedAt) + throw new Error("Artifact sync capture changes between pages"); + startedAt = page.startedAt; + await applyChanges(collection, refresh, page.changes, type, signal); + if (page.nextCursor === null) { + if (!page.checkpoint) + throw new Error("Artifact sync completion has no checkpoint"); + next = page.checkpoint; + break; + } + if (cursors.has(page.nextCursor)) + throw new Error("Artifact sync page cursor repeats"); + cursors.add(page.nextCursor); + cursor = page.nextCursor; + } + break; + } catch (error) { + if ( + !(error instanceof ArtifactIncrementalError) || + (error.code !== "incremental_reset" && + !( + error.code === "incremental_invalid" && + error.status === 400 && + checkpoint !== undefined && + cursor === undefined + )) || + recovery + ) + throw error; + checkpoint = undefined; + recovery = true; + } + } + if (recovery) + await revalidateKnown(reader, collection, refresh, type, signal); + signal.throwIfAborted(); + await collection.completeRefresh( + refresh, + { + expected, + next: JSON.stringify({ protocol: "incremental-v1", checkpoint: next }), + }, + signal, + ); + } catch (error) { + await reportFailure( + collection, + refresh, + error, + signal, + options.reportFailure === true, + ); + throw error; + } +} diff --git a/src/features/relay/artifact-store-sync.test.ts b/src/features/relay/artifact-store-sync.test.ts new file mode 100644 index 0000000000..2460b8fc66 --- /dev/null +++ b/src/features/relay/artifact-store-sync.test.ts @@ -0,0 +1,571 @@ +import { afterEach, expect, it, vi } from "vitest"; +import type { Context } from "@deepseek-ai/cordis"; +import type { + ArtifactStorageService, + OpenArtifactCollection, +} from "../artifact-storage/service"; +import { provideArtifactStoreSync } from "./artifact-store-sync"; +import { ArtifactIncrementalError } from "./artifact-incremental-reader"; +import { + reconcileArtifactDelta, + reconcileArtifactCurrent, +} from "./artifact-reconcile"; +import { keypair, signed } from "./testing"; +import { eventDto, type EventData, type RelayEvent } from "./events"; +vi.mock("./artifact-reconcile", async (original) => ({ + ...(await original()), + reconcileArtifactDelta: vi.fn(async () => {}), + reconcileArtifactCurrent: vi.fn(async () => {}), +})); +const stops: (() => void)[] = []; +afterEach(async () => { + for (const stop of stops.splice(0)) stop(); + await vi.advanceTimersByTimeAsync(0); + vi.mocked(reconcileArtifactDelta).mockReset().mockResolvedValue(undefined); + vi.mocked(reconcileArtifactCurrent).mockReset().mockResolvedValue(undefined); + vi.unstubAllGlobals(); + vi.restoreAllMocks(); + vi.useRealTimers(); +}); +function setup() { + vi.useFakeTimers(); + vi.spyOn(Math, "random").mockReturnValue(0); + vi.stubGlobal("navigator", { + locks: { request: vi.fn(async (_key, _options, callback) => callback()) }, + }); + const listeners = new Set<() => void>(); + const liveListeners = new Set<() => void>(); + let receive: (value: { events: readonly RelayEvent[] }) => void = () => {}; + let accepted: (event: EventData) => void = () => {}; + let opened: (entry: OpenArtifactCollection) => void = () => {}; + let online = true; + let scoped = true; + const session = { + artifactIncremental: {}, + live: { + snapshot: () => ({ status: online ? "connected" : "retrying" }), + subscribe: (listener: () => void) => { + liveListeners.add(listener); + return () => { + liveListeners.delete(listener); + }; + }, + }, + subscribeLive: (listener: typeof receive) => { + receive = listener; + return () => { + receive = () => {}; + }; + }, + outbox: { + observeSend: ( + listener: ( + event: EventData, + ) => ((pending: readonly EventData[]) => void) | undefined, + ) => { + accepted = (event) => listener(event)?.([]); + return () => { + accepted = () => {}; + }; + }, + }, + }; + const ctx = { + relay: { + snapshot: () => ({ + ...(scoped ? { viewer: "viewer", origin: "https://relay.test" } : {}), + status: scoped ? "ready" : "connecting", + cached: false, + session, + }), + subscribe: (listener: () => void) => { + listeners.add(listener); + return () => { + listeners.delete(listener); + }; + }, + }, + effect: (effect: () => () => void) => { + stops.push(effect()); + }, + }; + provideArtifactStoreSync( + ctx as unknown as Context, + { + onOpen: (listener: typeof opened) => { + opened = listener; + return () => {}; + }, + } as unknown as ArtifactStorageService, + ); + const add = (version = 1) => { + const controller = new AbortController(); + const collection = { applyBatch: vi.fn(async () => {}) }; + opened({ + viewer: "viewer", + relay: "https://relay.test", + owner: "test.plugin", + definition: { name: "tasks", type: "test.task", version }, + collection, + signal: controller.signal, + } as unknown as OpenArtifactCollection); + return { controller, collection }; + }; + return { + add, + scope: (value: boolean) => { + scoped = value; + for (const listener of listeners) listener(); + }, + confirmed: (event: RelayEvent) => receive({ events: [event] }), + accepted: (event: EventData) => accepted(event), + connection: (value: boolean) => { + online = value; + for (const listener of liveListeners) listener(); + }, + }; +} +it("coalesces collection repair and retires closed collections", async () => { + const h = setup(); + const entry = h.add(); + await vi.advanceTimersByTimeAsync(0); + expect(navigator.locks.request).toHaveBeenCalledTimes(1); + expect(reconcileArtifactDelta).toHaveBeenCalledOnce(); + await vi.advanceTimersByTimeAsync(60_000); + expect(reconcileArtifactDelta).toHaveBeenCalledTimes(2); + entry.controller.abort(); + await vi.advanceTimersByTimeAsync(60_000); + expect(reconcileArtifactDelta).toHaveBeenCalledTimes(2); +}); +it("aborts reconciliation on disconnect and starts a new capture on reconnect", async () => { + const h = setup(); + vi.mocked(reconcileArtifactDelta).mockImplementationOnce( + async (_reader, _collection, _type, signal) => { + await new Promise((resolve) => + signal.addEventListener("abort", () => resolve(), { once: true }), + ); + }, + ); + h.add(); + await vi.advanceTimersByTimeAsync(0); + const signal = vi.mocked(reconcileArtifactDelta).mock.calls[0]?.[3]; + h.connection(false); + expect(signal?.aborted).toBe(true); + h.connection(true); + await vi.advanceTimersByTimeAsync(0); + expect(reconcileArtifactDelta).toHaveBeenCalledTimes(2); +}); +it("keeps an open collection through a reconnect without relay scope", async () => { + const h = setup(); + vi.mocked(reconcileArtifactDelta).mockImplementationOnce( + async (_reader, _collection, _type, signal) => { + await new Promise((resolve) => + signal.addEventListener("abort", () => resolve(), { once: true }), + ); + }, + ); + const entry = h.add(); + await vi.advanceTimersByTimeAsync(0); + const signal = vi.mocked(reconcileArtifactDelta).mock.calls[0]?.[3]; + h.scope(false); + expect(signal?.aborted).toBe(true); + expect(entry.controller.signal.aborted).toBe(false); + await vi.advanceTimersByTimeAsync(60_000); + expect(reconcileArtifactDelta).toHaveBeenCalledOnce(); + h.scope(true); + await vi.advanceTimersByTimeAsync(0); + expect(reconcileArtifactDelta).toHaveBeenCalledTimes(2); + const event = eventDto( + signed(keypair(), { + kind: 45010, + content: "{}", + tags: [ + ["d", "task"], + ["type", "test.task"], + ], + }), + ); + h.confirmed(event); + await vi.advanceTimersByTimeAsync(0); + expect(reconcileArtifactCurrent).toHaveBeenCalledOnce(); + h.accepted(event); + await vi.advanceTimersByTimeAsync(0); + expect(reconcileArtifactCurrent).toHaveBeenCalledTimes(2); + await vi.advanceTimersByTimeAsync(60_000); + expect(reconcileArtifactDelta).toHaveBeenCalledTimes(3); + expect(navigator.locks.request).toHaveBeenCalledOnce(); +}); +it("refreshes after proven acceptance without writing raw revisions", async () => { + const h = setup(); + const entry = h.add(); + await vi.advanceTimersByTimeAsync(0); + const event = signed(keypair(), { + kind: 45010, + content: "{}", + tags: [ + ["d", "task"], + ["h", "channel"], + ["ar", "1"], + ["type", "test.task"], + ["op", "create"], + ], + }); + h.confirmed(eventDto(event)); + await vi.advanceTimersByTimeAsync(0); + expect(entry.collection.applyBatch).not.toHaveBeenCalled(); + expect(reconcileArtifactDelta).toHaveBeenCalledOnce(); + expect(reconcileArtifactCurrent).toHaveBeenCalledOnce(); +}); + +it("releases an expired version so a current collection can own sync", async () => { + const h = setup(); + vi.mocked(reconcileArtifactDelta).mockRejectedValueOnce( + new Error("Artifact collection version changed"), + ); + const old = h.add(); + await vi.advanceTimersByTimeAsync(60_000); + expect(reconcileArtifactDelta).toHaveBeenCalledOnce(); + h.add(); + await vi.advanceTimersByTimeAsync(0); + old.controller.abort(); + h.add(); + await vi.advanceTimersByTimeAsync(0); + expect(navigator.locks.request).toHaveBeenCalledTimes(2); +}); + +it("coalesces live invalidations during a capture into one later refresh", async () => { + const h = setup(); + let release = () => {}; + vi.mocked(reconcileArtifactDelta).mockImplementationOnce(async () => { + await new Promise((resolve) => { + release = resolve; + }); + }); + const entry = h.add(); + await vi.advanceTimersByTimeAsync(0); + const event = eventDto( + signed(keypair(), { + kind: 45010, + content: "{}", + tags: [ + ["type", "test.task"], + ["d", "task"], + ], + }), + ); + for (let i = 0; i < 50; i++) h.confirmed(event); + expect(reconcileArtifactDelta).toHaveBeenCalledOnce(); + release(); + await vi.advanceTimersByTimeAsync(0); + expect(reconcileArtifactDelta).toHaveBeenCalledOnce(); + expect(reconcileArtifactCurrent).toHaveBeenCalledOnce(); + expect(entry.collection.applyBatch).not.toHaveBeenCalled(); +}); + +it("keeps capture admission failures on the repair interval despite live traffic", async () => { + const h = setup(); + vi.mocked(reconcileArtifactDelta).mockRejectedValue( + new ArtifactIncrementalError("incremental_limit", "Too many memberships"), + ); + h.add(); + await vi.advanceTimersByTimeAsync(0); + const event = eventDto( + signed(keypair(), { + kind: 45010, + content: "{}", + tags: [ + ["type", "test.task"], + ["d", "task"], + ], + }), + ); + for (let i = 0; i < 20; i++) { + h.confirmed(event); + await vi.advanceTimersByTimeAsync(1); + } + expect(reconcileArtifactDelta).toHaveBeenCalledOnce(); + await vi.advanceTimersByTimeAsync(60_000); + expect(reconcileArtifactDelta).toHaveBeenCalledTimes(2); +}); +it.each([ + [0, 60_000], + [0.5, 75_000], + [0.99999, 89_999], +])( + "keeps random draw %s behind its %s ms retry deadline", + async (draw, delay) => { + const h = setup(); + vi.spyOn(Math, "random").mockReturnValueOnce(draw).mockReturnValue(0); + vi.mocked(reconcileArtifactDelta).mockRejectedValue( + new ArtifactIncrementalError("incremental_limit", "Capture limit"), + ); + h.add(); + await vi.advanceTimersByTimeAsync(0); + const event = eventDto( + signed(keypair(), { + kind: 45010, + content: "{}", + tags: [ + ["type", "test.task"], + ["d", "task"], + ], + }), + ); + await vi.advanceTimersByTimeAsync(delay - 1); + h.confirmed(event); + await vi.advanceTimersByTimeAsync(0); + expect(reconcileArtifactDelta).toHaveBeenCalledOnce(); + await vi.advanceTimersByTimeAsync(1); + expect(reconcileArtifactDelta).toHaveBeenCalledTimes(2); + await vi.advanceTimersByTimeAsync(59_999); + expect(reconcileArtifactDelta).toHaveBeenCalledTimes(2); + await vi.advanceTimersByTimeAsync(1); + expect(reconcileArtifactDelta).toHaveBeenCalledTimes(3); + }, +); +it("spreads retries across owners after simultaneous capture refusals", async () => { + const owners = Array.from({ length: 32 }, () => setup()); + const random = vi.spyOn(Math, "random"); + for (let index = 0; index < owners.length; index++) + random.mockReturnValueOnce(index / owners.length); + vi.mocked(reconcileArtifactDelta).mockRejectedValue( + new ArtifactIncrementalError("incremental_unavailable", "Capture busy"), + ); + for (const owner of owners) owner.add(); + await vi.advanceTimersByTimeAsync(7000); + expect(reconcileArtifactDelta).toHaveBeenCalledTimes(128); + vi.mocked(reconcileArtifactDelta).mockResolvedValue(undefined); + await vi.advanceTimersByTimeAsync(59_999); + expect(reconcileArtifactDelta).toHaveBeenCalledTimes(128); + let elapsed = 59_999; + for (let index = 0; index < owners.length; index++) { + const deadline = 60_000 + Math.floor((index / owners.length) * 30_000); + await vi.advanceTimersByTimeAsync(deadline - elapsed); + expect(reconcileArtifactDelta).toHaveBeenCalledTimes(129 + index); + elapsed = deadline; + } +}); +it("keeps a current handle when an older in-flight version fails", async () => { + const h = setup(); + let fail = () => {}; + vi.mocked(reconcileArtifactDelta).mockImplementationOnce(async () => { + await new Promise((_resolve, reject) => { + fail = () => reject(new Error("Artifact collection version changed")); + }); + }); + const old = h.add(1); + await vi.advanceTimersByTimeAsync(0); + const current = h.add(2); + fail(); + await vi.advanceTimersByTimeAsync(0); + expect(reconcileArtifactDelta).toHaveBeenCalledTimes(2); + expect(vi.mocked(reconcileArtifactDelta).mock.calls[1]?.[1]).toBe( + current.collection, + ); + old.controller.abort(); + await vi.advanceTimersByTimeAsync(60_000); + expect(reconcileArtifactDelta).toHaveBeenCalledTimes(3); + expect(navigator.locks.request).toHaveBeenCalledOnce(); +}); +it("keeps duplicate delayed hints behind an in-flight current read", async () => { + const h = setup(); + h.add(); + await vi.advanceTimersByTimeAsync(0); + let finish = () => {}; + vi.mocked(reconcileArtifactCurrent).mockImplementationOnce(async () => { + await new Promise((resolve) => { + finish = resolve; + }); + }); + const event = eventDto( + signed(keypair(), { + kind: 45010, + content: "{}", + tags: [ + ["type", "test.task"], + ["d", "task"], + ], + }), + ); + h.confirmed(event); + await vi.advanceTimersByTimeAsync(0); + for (let i = 0; i < 50; i++) h.confirmed(event); + expect(reconcileArtifactCurrent).toHaveBeenCalledOnce(); + finish(); + await vi.advanceTimersByTimeAsync(0); + expect(reconcileArtifactCurrent).toHaveBeenCalledTimes(2); + expect(reconcileArtifactDelta).toHaveBeenCalledOnce(); +}); +it("uses bounded current reads and requests a delta after live overflow", async () => { + const h = setup(); + let finish = () => {}; + vi.mocked(reconcileArtifactDelta).mockImplementationOnce(async () => { + await new Promise((resolve) => { + finish = resolve; + }); + }); + h.add(); + await vi.advanceTimersByTimeAsync(0); + const key = keypair(); + for (let i = 0; i < 1002; i++) + h.confirmed( + eventDto( + signed(key, { + kind: 45010, + content: "{}", + tags: [ + ["type", "test.task"], + ["d", `task-${i}`], + ], + }), + ), + ); + finish(); + await vi.advanceTimersByTimeAsync(0); + expect(reconcileArtifactCurrent).toHaveBeenCalledTimes(10); + expect( + vi + .mocked(reconcileArtifactCurrent) + .mock.calls.every((call) => call[3].length <= 100), + ).toBe(true); + expect(reconcileArtifactDelta).toHaveBeenCalledTimes(2); +}, 15_000); // This case signs and verifies 1,002 distinct live events. + +it("keeps capacity backoff when a live read starts before the repair timer", async () => { + const h = setup(); + h.add(); + await vi.advanceTimersByTimeAsync(30_000); + vi.mocked(reconcileArtifactCurrent).mockRejectedValueOnce( + new ArtifactIncrementalError("incremental_limit", "Large"), + ); + const event = eventDto( + signed(keypair(), { + kind: 45010, + content: "{}", + tags: [ + ["type", "test.task"], + ["d", "task"], + ], + }), + ); + h.confirmed(event); + await vi.advanceTimersByTimeAsync(0); + expect(reconcileArtifactCurrent).toHaveBeenCalledOnce(); + await vi.advanceTimersByTimeAsync(30_000); + h.confirmed(event); + await vi.advanceTimersByTimeAsync(0); + expect(reconcileArtifactCurrent).toHaveBeenCalledOnce(); + await vi.advanceTimersByTimeAsync(30_000); + expect(reconcileArtifactCurrent).toHaveBeenCalledTimes(2); +}); + +it("keeps transient retry times under live hints and reports a persistent current failure", async () => { + const h = setup(); + h.add(); + await vi.advanceTimersByTimeAsync(0); + vi.mocked(reconcileArtifactCurrent).mockRejectedValue( + new Error("Disconnected"), + ); + const event = eventDto( + signed(keypair(), { + kind: 45010, + content: "{}", + tags: [ + ["type", "test.task"], + ["d", "task"], + ], + }), + ); + h.confirmed(event); + await vi.advanceTimersByTimeAsync(0); + for (let i = 0; i < 10; i++) { + h.confirmed(event); + await vi.advanceTimersByTimeAsync(50); + } + expect(reconcileArtifactCurrent).toHaveBeenCalledOnce(); + await vi.advanceTimersByTimeAsync(500); + expect(reconcileArtifactCurrent).toHaveBeenCalledTimes(2); + await vi.advanceTimersByTimeAsync(2000); + expect(reconcileArtifactCurrent).toHaveBeenCalledTimes(3); + await vi.advanceTimersByTimeAsync(4000); + expect(reconcileArtifactCurrent).toHaveBeenCalledTimes(4); + expect(vi.mocked(reconcileArtifactCurrent).mock.calls[3]?.[5]).toEqual({ + reportFailure: true, + }); + expect(reconcileArtifactDelta).toHaveBeenCalledOnce(); +}); + +it("keeps the periodic delta deadline under continuing live hints", async () => { + const h = setup(); + h.add(); + await vi.advanceTimersByTimeAsync(0); + const event = eventDto( + signed(keypair(), { + kind: 45010, + content: "{}", + tags: [ + ["type", "test.task"], + ["d", "task"], + ], + }), + ); + for (let i = 0; i < 6; i++) { + await vi.advanceTimersByTimeAsync(9000); + h.confirmed(event); + await vi.advanceTimersByTimeAsync(0); + } + expect(reconcileArtifactDelta).toHaveBeenCalledOnce(); + await vi.advanceTimersByTimeAsync(6000); + expect(reconcileArtifactDelta).toHaveBeenCalledTimes(2); +}); + +it("keeps a failed delta retry round ahead of queued live reads", async () => { + const h = setup(); + vi.mocked(reconcileArtifactDelta).mockRejectedValue( + new Error("Disconnected"), + ); + h.add(); + await vi.advanceTimersByTimeAsync(0); + const event = eventDto( + signed(keypair(), { + kind: 45010, + content: "{}", + tags: [ + ["type", "test.task"], + ["d", "task"], + ], + }), + ); + h.confirmed(event); + await vi.advanceTimersByTimeAsync(7000); + expect(reconcileArtifactDelta).toHaveBeenCalledTimes(4); + expect(vi.mocked(reconcileArtifactDelta).mock.calls[3]?.[4]).toEqual({ + reportFailure: true, + }); + expect(reconcileArtifactCurrent).not.toHaveBeenCalled(); +}); + +it("requests a current read from the existing outbox acceptance observer", async () => { + const h = setup(); + h.add(); + await vi.advanceTimersByTimeAsync(0); + h.accepted( + signed(keypair(), { + kind: 45010, + content: "{}", + tags: [ + ["type", "test.task"], + ["d", "task"], + ], + }), + ); + await vi.advanceTimersByTimeAsync(0); + expect(reconcileArtifactCurrent).toHaveBeenCalledWith( + expect.anything(), + expect.anything(), + "test.task", + ["task"], + expect.any(AbortSignal), + expect.anything(), + ); +}); diff --git a/src/features/relay/artifact-store-sync.ts b/src/features/relay/artifact-store-sync.ts new file mode 100644 index 0000000000..d748b44b5c --- /dev/null +++ b/src/features/relay/artifact-store-sync.ts @@ -0,0 +1,319 @@ +import type { Context } from "@deepseek-ai/cordis"; +import type { + ArtifactStorageService, + OpenArtifactCollection, +} from "../artifact-storage/service"; +import { hasEventProof } from "./events"; +import type { RelaySession } from "./session"; +import { + reconcileArtifactDelta, + reconcileArtifactCurrent, +} from "./artifact-reconcile"; +import { + ArtifactIncrementalError, + ARTIFACT_INCREMENTAL_LIMITS, +} from "./artifact-incremental-reader"; +import { transportPartition } from "./partition"; +import { RECOVERY_DELAYS_MS } from "./recovery"; + +const REPAIR_MS = 60_000; +type Group = { + key: string; + scope: string; + entries: Set; + lifetime: AbortController; + request(): void; +}; + +/** The existing session supplies live traffic. Web Locks transfer sync work between windows. */ +export function provideArtifactStoreSync( + ctx: Context, + storage: ArtifactStorageService, +) { + const groups = new Map(); + function ensureGroup( + viewer: string, + relay: string, + owner: string, + name: string, + ) { + const scope = transportPartition({ viewer, scope: relay, relayAuthor: "" }); + const key = JSON.stringify([viewer, relay, owner, name]); + let group = groups.get(key); + if (!group) { + const lifetime = new AbortController(); + group = { key, scope, entries: new Set(), lifetime, request: () => {} }; + groups.set(key, group); + const owned = group; + if (typeof navigator !== "undefined" && navigator.locks) { + void navigator.locks + .request( + `buzz-artifact-sync:${key}`, + { mode: "exclusive", signal: lifetime.signal }, + () => own(owned, viewer, relay), + ) + .catch((error) => { + if (!lifetime.signal.aborted) + console.error("Artifact sync ownership fails", error); + }); + } + } + return group; + } + const stopOpen = storage.onOpen((entry) => { + const group = ensureGroup( + entry.viewer, + entry.relay, + entry.owner, + entry.definition.name, + ); + group.entries.add(entry); + group.request(); + entry.signal.addEventListener( + "abort", + () => { + group.entries.delete(entry); + if (!group.entries.size) { + group.lifetime.abort(); + if (groups.get(group.key) === group) groups.delete(group.key); + } + }, + { once: true }, + ); + }); + const refreshScope = () => { + const { viewer, origin } = ctx.relay.snapshot(); + const scope = + viewer && origin + ? transportPartition({ viewer, scope: origin, relayAuthor: "" }) + : undefined; + if (scope === undefined) return; + for (const [key, group] of groups) + if (group.scope !== scope) { + group.lifetime.abort(); + groups.delete(key); + } + }; + const stopScope = ctx.relay.subscribe(refreshScope); + refreshScope(); + async function own(group: Group, viewer: string, relay: string) { + let session: RelaySession | undefined; + let stopSession = () => {}; + let stopLive = () => {}; + let stopWrites = () => {}; + let connected = false; + let running: Promise | undefined; + let requested = true; + const dirty = new Set(); + let retry = 0; + let notBefore = 0; + let timer: ReturnType | undefined; + let batch: AbortController | undefined; + const active = () => { + const now = ctx.relay.snapshot(); + return ( + now.viewer === viewer && + now.origin === relay && + now.status === "ready" && + !now.cached && + now.session.live.snapshot().status === "connected" + ); + }; + const request = () => { + requested = true; + pump(); + }; + const hint = (event: import("./events").EventData) => { + if ( + event.kind === 45010 && + [...group.entries].some((entry) => + event.tags.some( + ([key, value]) => key === "type" && value === entry.definition.type, + ), + ) + ) { + const id = event.tags.find(([key]) => key === "d")?.[1]; + if (id && dirty.size < 1000) dirty.add(id); + else requested = true; + pump(); + } else if ([45011, 9005, 44100, 44101].includes(event.kind)) request(); + }; + const schedule = (delay: number) => { + clearTimeout(timer); + timer = setTimeout(() => { + timer = undefined; + pump(); + }, delay); + }; + function pump() { + if ( + running || + !group.entries.size || + group.lifetime.signal.aborted || + !active() || + !session?.artifactIncremental || + (!requested && !dirty.size) + ) + return; + if (Date.now() < notBefore) { + schedule(notBefore - Date.now()); + return; + } + clearTimeout(timer); + timer = undefined; + const current = session; + const reader = current.artifactIncremental; + if (!reader) return; + const entry = [...group.entries][0]; + if (!entry || entry.signal.aborted) return; + batch = new AbortController(); + const signal = AbortSignal.any([ + group.lifetime.signal, + entry.signal, + batch.signal, + ]); + running = (async () => { + try { + if (requested) { + requested = false; + try { + await reconcileArtifactDelta( + reader, + entry.collection, + entry.definition.type, + signal, + { reportFailure: retry >= 3 }, + ); + } catch (error) { + requested = true; + throw error; + } + } + const ids = [...dirty].slice( + 0, + ARTIFACT_INCREMENTAL_LIMITS.currentRows, + ); + if (ids.length) { + for (const id of ids) dirty.delete(id); + try { + await reconcileArtifactCurrent( + reader, + entry.collection, + entry.definition.type, + ids, + signal, + { reportFailure: retry >= 3 }, + ); + } catch (error) { + for (const id of ids) { + if (dirty.size < 1000) dirty.add(id); + else requested = true; + } + throw error; + } + } + retry = 0; + notBefore = 0; + } catch (error) { + if ( + error instanceof Error && + error.message === "Artifact collection version changed" + ) { + group.entries.delete(entry); + if (!group.entries.size) { + group.lifetime.abort(); + if (groups.get(group.key) === group) groups.delete(group.key); + } else { + retry = 0; + requested = true; + } + } else if (!signal.aborted) { + retry++; + const permanent = + error instanceof ArtifactIncrementalError && + [ + "incremental_limit", + "incremental_invalid", + "incremental_unsupported_scope", + ].includes(error.code); + const delay = + !permanent && retry <= 3 + ? (RECOVERY_DELAYS_MS[retry - 1] ?? REPAIR_MS) + : REPAIR_MS + Math.floor(Math.random() * 30_000); + notBefore = Date.now() + delay; + schedule(delay); + } + } + })().finally(() => { + running = undefined; + batch = undefined; + if (!group.lifetime.signal.aborted && (requested || dirty.size)) pump(); + }); + } + group.request = request; + const repair = setInterval(request, REPAIR_MS); + const update = () => { + const now = ctx.relay.snapshot(); + const next = + now.viewer === viewer && now.origin === relay ? now.session : undefined; + const changed = next !== session; + if (changed) { + batch?.abort(); + stopSession(); + stopLive(); + stopWrites(); + session = next; + if (next) { + stopSession = next.live.subscribe(update); + stopLive = next.subscribeLive( + ({ events }) => { + for (const event of events) if (hasEventProof(event)) hint(event); + }, + { replay: true }, + ); + stopWrites = + next.outbox?.observeSend((event) => + event.kind === 45010 ? () => hint(event) : undefined, + ) ?? (() => {}); + } + } + const ready = active(); + if (!ready) { + batch?.abort(); + requested = true; + } else if (!connected || changed) { + retry = 0; + notBefore = 0; + request(); + } + connected = ready; + }; + const stopRelay = ctx.relay.subscribe(update); + update(); + try { + await new Promise((resolve) => { + if (group.lifetime.signal.aborted) resolve(); + else + group.lifetime.signal.addEventListener("abort", () => resolve(), { + once: true, + }); + }); + } finally { + batch?.abort(); + clearTimeout(timer); + clearInterval(repair); + stopRelay(); + stopSession(); + stopLive(); + stopWrites(); + group.request = () => {}; + await running; + } + } + ctx.effect(() => () => { + stopOpen(); + stopScope(); + for (const group of groups.values()) group.lifetime.abort(); + groups.clear(); + }); +} diff --git a/src/features/relay/http-admission.ts b/src/features/relay/http-admission.ts index 97c12c5eda..980331df3e 100644 --- a/src/features/relay/http-admission.ts +++ b/src/features/relay/http-admission.ts @@ -23,6 +23,7 @@ export class ApiCapacity extends ApiNotSent { } export type ApiFailure = Readonly<{ error: string; + code?: string; quota?: "api"; paused?: true; retryAfterMs?: number; @@ -84,7 +85,10 @@ export function apiFailure(status: number, value: unknown): ApiFailure { * `reason` may keep an exact, caller-allowed upstream refusal from that body. */ export async function readApiFailure( response: Response, - reason?: (body: unknown, status: number) => string | undefined, + reason?: ( + body: unknown, + status: number, + ) => string | Pick | undefined, ): Promise { if (!response.body) return apiFailure(response.status, undefined); const reader = response.body.getReader(); @@ -109,7 +113,12 @@ export async function readApiFailure( } const failure = apiFailure(response.status, body); const allowed = failure.quota ? undefined : reason?.(body, response.status); - return allowed ? { ...failure, error: allowed } : failure; + return allowed + ? { + ...failure, + ...(typeof allowed === "string" ? { error: allowed } : allowed), + } + : failure; } finally { await reader.cancel().catch(() => {}); reader.releaseLock(); @@ -248,7 +257,10 @@ export async function admittedApiRequest( request: () => Promise, signal?: AbortSignal, priority: Ticket["priority"] = "foreground", - reason?: (body: unknown, status: number) => string | undefined, + reason?: ( + body: unknown, + status: number, + ) => string | Pick | undefined, ): Promise { return lane.run( async () => { diff --git a/src/features/relay/native.test.ts b/src/features/relay/native.test.ts index 9cb3ea8b6c..d3bf18069f 100644 --- a/src/features/relay/native.test.ts +++ b/src/features/relay/native.test.ts @@ -1,3 +1,7 @@ +import { + ARTIFACT_INCREMENTAL_LIMITS, + ArtifactIncrementalError, +} from "./artifact-incremental-reader"; import { ReadStateTimestampRejected, READ_STATE_TIMESTAMP_REFUSAL, @@ -2083,3 +2087,46 @@ it("cancellation during ingress prevents further chunks and final upload", async ).toHaveLength(1); expect(uploads).toHaveLength(0); }); + +it.each([ + { live: false, incremental: false }, + { live: true, incremental: false }, + { live: false, incremental: true }, + { live: true, incremental: true }, +])( + "keeps live and incremental artifact capabilities separate: %j", + async ({ live, incremental }) => { + discovery = { + artifacts: { + ...(live ? { sync: { version: 1 } } : {}), + ...(incremental ? { incremental: ARTIFACT_INCREMENTAL_LIMITS } : {}), + }, + }; + const transport = await connectNativeTransport(community); + expect(transport.artifactSync).toBe(live ? true : undefined); + expect(transport.artifactIncremental !== undefined).toBe(incremental); + }, +); + +it.each([ + "incremental_limit", + "incremental_reset", + "incremental_invalid", + "incremental_unsupported_scope", + "incremental_unavailable", +])("preserves %s through native admission", async (code) => { + discovery = { artifacts: { incremental: ARTIFACT_INCREMENTAL_LIMITS } }; + const transport = await connectNativeTransport(community); + respond = () => ({ + status: 422, + body: { error: "scope or capacity", code }, + }); + assert.exists(transport.artifactIncremental); + const work = transport.artifactIncremental.delta( + { artifact: "current", kinds: [45010], "#type": ["task"] }, + {}, + new AbortController().signal, + ); + await expect(work).rejects.toBeInstanceOf(ArtifactIncrementalError); + await expect(work).rejects.toMatchObject({ code }); +}); diff --git a/src/features/relay/native.ts b/src/features/relay/native.ts index 42072d2b4c..a04512d238 100644 --- a/src/features/relay/native.ts +++ b/src/features/relay/native.ts @@ -1,3 +1,8 @@ +import { + artifactIncrementalCapability, + artifactIncrementalFailure, + createArtifactIncrementalReader, +} from "./artifact-incremental-reader"; import { verifiedProfileAuthority, verifiedProfileDiscovery, @@ -439,6 +444,21 @@ export async function connectNativeTransport( // Capabilities describe implemented host operations, not everything this key can sign. return { ...transport, + ...(artifactIncrementalCapability(info) + ? { + artifactIncremental: createArtifactIncrementalReader( + (filter, signal) => + admitSignedRequest( + origin, + transport.viewer, + () => nativeRelayRequest(origin, "/query", [filter]), + signal, + "background", + artifactIncrementalFailure, + ), + ), + } + : {}), workflows: workflowHost(async (route, body, signal) => { if (route !== "workflow-runs") throw new Error("Invalid workflow read"); workflowRunsPath(body); diff --git a/src/features/relay/session.ts b/src/features/relay/session.ts index 8ef96810b2..7d4f1e2bc2 100644 --- a/src/features/relay/session.ts +++ b/src/features/relay/session.ts @@ -2154,6 +2154,7 @@ export function createRelaySession( liveWasConnected = connected; }); const session = Object.freeze({ + artifactIncremental: transport?.artifactIncremental, memberAdditions, presence, viewer: transport?.viewer, diff --git a/src/features/relay/transport.ts b/src/features/relay/transport.ts index 0ae047f12b..55d514a803 100644 --- a/src/features/relay/transport.ts +++ b/src/features/relay/transport.ts @@ -1,3 +1,4 @@ +import type { ArtifactIncrementalReader } from "./artifact-incremental-reader.ts"; import type { VerifiedProfileAuthority } from "./verified-profile"; import type { ReminderHost } from "./reminders"; import { @@ -88,6 +89,7 @@ export interface RelayWriter { ): Promise | Promise; } export interface ReadTransport { + readonly artifactIncremental?: ArtifactIncrementalReader; readonly projectGit?: RepoGit; readonly readAgentMemories?: MemoryReader; /** Session-scoped owner proof, not an arbitrary signing capability. */ @@ -1336,7 +1338,7 @@ export const admitSignedRequest = ( request: () => Promise, signal?: AbortSignal, priority: "foreground" | "background" = "foreground", - reason?: (body: unknown) => string | undefined, + reason?: Parameters[4], ) => admittedApiRequest( signedAdmissions(relayOrigin(origin), viewer).api, diff --git a/tests/browser/artifact-sync.spec.mjs b/tests/browser/artifact-sync.spec.mjs new file mode 100644 index 0000000000..dc50831ba0 --- /dev/null +++ b/tests/browser/artifact-sync.spec.mjs @@ -0,0 +1,87 @@ +import { fileURLToPath } from "node:url"; +import { test as base, expect } from "./playwright-test.mjs"; +import { createServer } from "./vite-server.mjs"; + +const test = base.extend({ + syncUrl: [ + async ({ browserName }, use) => { + if (!["chromium", "webkit"].includes(browserName)) + throw new Error("Unsupported sync test engine"); + const server = await createServer({ + root: fileURLToPath(new URL("../../", import.meta.url)), + configFile: false, + envDir: false, + logLevel: "error", + server: { host: "127.0.0.1", port: 0 }, + }); + await server.listen(); + try { + await use( + `${server.resolvedUrls.local[0]}tests/fixtures/artifact-sync.html`, + ); + } finally { + await server.close(); + } + }, + { scope: "worker" }, + ], +}); +async function open(page, url) { + await page.goto(url); + await page.waitForFunction(() => !!window.artifactSyncTest); + await page.evaluate(() => window.artifactSyncTest.open("collection-owner")); +} + +// Real Web Locks and window teardown establish ownership. IndexedDB keeps shared progress. +test("syncs a follower-only collection and transfers a shared collection on window close", async ({ + page, + context, + syncUrl, +}) => { + await open(page, syncUrl); + await page.evaluate(() => window.artifactSyncTest.openCollection("main")); + await expect + .poll(() => page.evaluate(() => window.artifactSyncTest.checkpoint("main"))) + .not.toBeNull(); + const follower = await context.newPage(); + await open(follower, syncUrl); + await follower.evaluate(() => + window.artifactSyncTest.openCollection("archive"), + ); + await expect + .poll(() => + follower.evaluate(() => window.artifactSyncTest.checkpoint("archive")), + ) + .not.toBeNull(); + await follower.evaluate(() => window.artifactSyncTest.openCollection("main")); + await expect + .poll(() => + follower.evaluate(async () => { + const locks = await navigator.locks.query(); + return { held: locks.held.length, pending: locks.pending.length }; + }), + ) + .toEqual({ held: 2, pending: 1 }); + expect(await page.evaluate(() => window.artifactSyncTest.captures)).toEqual([ + "main", + ]); + expect( + await follower.evaluate(() => window.artifactSyncTest.captures), + ).toEqual(["archive"]); + const before = await follower.evaluate(() => + window.artifactSyncTest.checkpoint("main"), + ); + await page.close(); + await expect + .poll(() => follower.evaluate(() => window.artifactSyncTest.captures)) + .toEqual(["archive", "main"]); + await expect + .poll(() => + follower.evaluate(() => window.artifactSyncTest.checkpoint("main")), + ) + .not.toBe(before); + expect( + (await follower.evaluate(() => window.artifactSyncTest.read("main"))) + .preference, + ).toEqual({ pinned: false }); +}); diff --git a/tests/fixtures/artifact-sync.html b/tests/fixtures/artifact-sync.html new file mode 100644 index 0000000000..dbf0ed5792 --- /dev/null +++ b/tests/fixtures/artifact-sync.html @@ -0,0 +1 @@ +Artifact sync checks

Artifact sync checks

diff --git a/tests/fixtures/artifact-sync.ts b/tests/fixtures/artifact-sync.ts new file mode 100644 index 0000000000..769acda3b3 --- /dev/null +++ b/tests/fixtures/artifact-sync.ts @@ -0,0 +1,126 @@ +import type { Context } from "@deepseek-ai/cordis"; +import type { + ArtifactStorageService, + OpenArtifactCollection, +} from "../../src/features/artifact-storage/service"; +import { createArtifactStore } from "../../src/features/artifact-storage/store"; +import type { HostArtifactCollection } from "../../src/features/artifact-storage/types"; +import type { ArtifactIncrementalReader } from "../../src/features/relay/artifact-incremental-reader"; +import { provideArtifactStoreSync } from "../../src/features/relay/artifact-store-sync"; + +const viewer = "a".repeat(64); +const captures: string[] = []; +const collections = new Map< + string, + { collection: HostArtifactCollection; controller: AbortController } +>(); +let store: ReturnType; +let relay: string; +let opened: (entry: OpenArtifactCollection) => void; +let stop = () => {}; +const api = { + captures, + open(scope: string) { + relay = `https://${scope}.test`; + store = createArtifactStore(viewer, relay); + const reader: ArtifactIncrementalReader = { + async delta(query) { + captures.push(query["#type"][0]?.slice("test.sync.".length) ?? ""); + return { + startedAt: "2026-10-21T12:00:00.000001Z", + overlapSeconds: 120, + changes: [], + nextCursor: null, + checkpoint: crypto.randomUUID(), + }; + }, + async current(_query, artifactIds) { + return { + startedAt: "2026-10-21T12:00:00.000002Z", + changes: artifactIds.map((artifactId) => ({ + type: "remove" as const, + artifactId, + })), + }; + }, + }; + const session = { + artifactIncremental: reader, + live: { + snapshot: () => ({ status: "connected" }), + subscribe: () => () => {}, + }, + subscribeLive: () => () => {}, + }; + const ctx = { + relay: { + snapshot: () => ({ + viewer, + origin: relay, + status: "ready", + cached: false, + session, + }), + subscribe: () => () => {}, + }, + effect: (effect: () => () => void) => { + stop = effect(); + }, + }; + provideArtifactStoreSync( + ctx as unknown as Context, + { + onOpen: (listener: typeof opened) => { + opened = listener; + return () => {}; + }, + } as unknown as ArtifactStorageService, + ); + }, + async openCollection(name: string) { + const definition = { + name, + type: `test.sync.${name}`, + version: 1, + project: () => [], + }; + const collection = store.collection("test.plugin", definition); + const controller = new AbortController(); + collections.set(name, { collection, controller }); + await collection.importBatch({ + records: [{ id: "task", local: { title: "Saved" } }], + preferences: [{ id: "task", value: { pinned: false } }], + marker: "ready", + }); + opened({ + viewer, + relay, + owner: "test.plugin", + definition, + collection, + signal: controller.signal, + }); + }, + checkpoint(name: string) { + const entry = collections.get(name); + if (!entry) throw new Error("Collection is not open"); + return entry.collection.checkpoint(); + }, + read(name: string) { + const entry = collections.get(name); + if (!entry) throw new Error("Collection is not open"); + return entry.collection.get("task"); + }, + close() { + stop(); + for (const entry of collections.values()) entry.controller.abort(); + store?.close(); + }, +}; +declare global { + interface Window { + artifactSyncTest: typeof api; + } +} +window.artifactSyncTest = api; +window.addEventListener("pagehide", () => api.close());