From cd2d0681342f822cc487443585ca1c96dd7c21f5 Mon Sep 17 00:00:00 2001 From: Travis Rich Date: Mon, 5 Oct 2026 15:03:10 -0400 Subject: [PATCH] feat: push to Underlay with the v2 delta protocol Underlay v2 removes negotiate. A push now reads the head's manifest, diffs our records against it by (type, id) and hash, and uploads only upserts and deletes through a push session (POST /push, /records, /deletes, async /commit). - hash: JCS (RFC 8785) written as a string, matching the server; the old sorted-object JSON.stringify put integer-like keys first. canonicalize stays for PubPub's own signatures. - client: readHead (paged manifest, pinned to the head semver), diffAgainstHead, batches sized by the session's line and byte limits, session abandoned on failure, 409 conflict / "No changes detected" told apart, missing files uploaded then a new session, presigned upload for files over 32 MB, metadata sent as a patch of changed fields only, default base URL on www (the apex 301 turns POSTs into GETs). - incremental: resolve cache-hit records by (type, id) and self-heal cache entries whose stored hashes no longer match. - worker: cache persisted on noop too; commit change counts in the push log. - /accounts/me lists orgs as `orgs`, not `accounts`: the org picker and the connection test's org check now work. --- server/underlay/__tests__/client.test.ts | 728 +++++++++++---- server/underlay/__tests__/hash.test.ts | 21 +- server/underlay/__tests__/incremental.test.ts | 59 +- server/underlay/client.ts | 826 ++++++++++++------ server/underlay/hash.ts | 84 +- server/underlay/incremental.ts | 78 +- server/underlay/mapping.ts | 13 +- workers/tasks/pushToUnderlay.ts | 48 +- 8 files changed, 1344 insertions(+), 513 deletions(-) diff --git a/server/underlay/__tests__/client.test.ts b/server/underlay/__tests__/client.test.ts index f7cc59648..93038b6f7 100644 --- a/server/underlay/__tests__/client.test.ts +++ b/server/underlay/__tests__/client.test.ts @@ -1,13 +1,14 @@ -import type { UnderlayPushPayload } from '../mapping'; +import type { UnderlayPushPayload, UnderlayRecord } from '../mapping'; import { afterEach, describe, expect, it, vi } from 'vitest'; -import { UnderlayClient } from '../client'; +import { diffAgainstHead, metadataChanges, UnderlayClient } from '../client'; +import { hashRecord, hashSchema } from '../hash'; /** - * Async commit. Underlay finalizes a large commit in the background: it answers 202 and records the - * outcome on the session, which the client polls. These tests pin that handshake — including the - * fallback to a synchronous 201, so the two services can be deployed in either order. + * Delta push. The client reads the head's manifest, diffs its own records against it, and uploads + * only upserts and deletes through a push session. These tests drive the client against a small + * in-memory fake of Underlay's push API. */ const BASE = 'https://underlay.test/api'; @@ -26,287 +27,638 @@ const makeClient = () => const json = (body: unknown, status = 200) => new Response(JSON.stringify(body), { status, headers: { 'Content-Type': 'application/json' } }); -/** A minimal push: one record, no files. */ -const payload = (): UnderlayPushPayload => ({ - records: [{ id: 'r1', type: 'Pub', data: { slug: 'p1' } }], +const PubSchema = { type: 'object', properties: { slug: { type: 'string' } } }; +const NoteSchema = { type: 'object', properties: { text: { type: 'string' } } }; + +const rec = (id: string, slug: string, type = 'Pub'): UnderlayRecord => ({ + id, + type, + data: { slug }, +}); + +const payloadOf = ( + records: UnderlayRecord[], + extra: Partial = {}, +): UnderlayPushPayload => ({ + records, files: [], fileHashes: [], - schemas: {}, - manifest: [{ id: 'r1', type: 'Pub', hash: 'h1' }], + schemas: { Pub: PubSchema }, + ...extra, }); +type HeadRecord = { id: string; type: string; hash: string; private?: boolean }; + +const headOf = (records: UnderlayRecord[], extra: HeadRecord[] = []): HeadRecord[] => [ + ...records.map((r) => ({ + id: r.id, + type: r.type, + hash: hashRecord(r).hash, + ...(r.private ? { private: true } : {}), + })), + ...extra, +]; + +type Fake = { + calls: string[]; + /** Bodies of every `POST /push` (session open). */ + opens: Record[]; + /** NDJSON lines uploaded per kind, across all sessions. */ + lines: { records: Record[]; deletes: Record[] }; + /** Request bodies (line counts) of each records/deletes POST, in order. */ + batches: { kind: string; lines: number }[]; + headers: { url: string; auth: boolean }[]; +}; + /** - * Routes fetches by URL so a test only has to describe the handful of responses it cares about. - * `sessionStates` is consumed one poll at a time, so a test can script `committing → committed`. + * A fake Underlay. `head` is the current version (null = no versions); `commit` scripts each + * commit's response; `sessionStates` is consumed one poll at a time. */ -const stubFetch = (opts: { - commit: Response | (() => Response); +const fakeUnderlay = (opts: { + head?: { + semver: string; + records: HeadRecord[]; + schemas?: Record; + metadata?: Record | null; + } | null; + manifestPageSize?: number; + open?: () => Response | null; + commit?: () => Response; sessionStates?: unknown[]; + recordsResponse?: () => Response; + limits?: Record; onPut?: (url: string) => void; -}) => { - const calls: string[] = []; + upload?: { statuses: string[] }; +}): Fake => { + const fake: Fake = { + calls: [], + opens: [], + lines: { records: [], deletes: [] }, + batches: [], + headers: [], + }; const states = [...(opts.sessionStates ?? [])]; - const fetchMock = vi.fn(async (input: any, init?: any) => { - const url = String(input); - calls.push(`${init?.method ?? 'GET'} ${url}`); - - // Re-read of the current head, done when push() recovers by re-negotiating. - if (url.endsWith('/versions/latest')) { - return json({ semver: '1.0.0' }); - } - if (url.endsWith('/versions/negotiate')) { - // The server already has the records; these tests are about the commit handshake. - return json({ session_id: 's1', needed_records: [], needed_files: [] }); - } - if (url.includes('/versions/negotiate/s1/records')) { - return json({ ok: true }); - } - if (url.includes('/versions/negotiate/s1/commit')) { - return typeof opts.commit === 'function' ? opts.commit() : opts.commit; - } - if (url.includes('/files/sha256:')) { - opts.onPut?.(url); - return json({ status: 'exists' }); - } - // Session poll. - if (url.endsWith('/versions/negotiate/s1')) { - return json(states.shift() ?? { status: 'committing' }); - } - throw new Error(`unexpected fetch: ${url}`); - }); - vi.stubGlobal('fetch', fetchMock); - return { calls, fetchMock }; + const uploadStatuses = [...(opts.upload?.statuses ?? [])]; + let sessionCount = 0; + const head = opts.head ?? null; + const pageSize = opts.manifestPageSize ?? 25_000; + + vi.stubGlobal( + 'fetch', + vi.fn(async (input: any, init?: any) => { + const url = String(input); + const method = init?.method ?? 'GET'; + fake.calls.push(`${method} ${url}`); + fake.headers.push({ url, auth: new Headers(init?.headers).has('Authorization') }); + const path = url.split('?')[0]; + const query = new URL(url).searchParams; + + if (path === `${COLLECTION}/versions/latest`) { + return head + ? json({ semver: head.semver, metadata: head.metadata ?? null }) + : json({ error: 'No versions' }, 404); + } + if (head && path === `${COLLECTION}/versions/${head.semver}/manifest`) { + const start = Number(query.get('cursor') ?? 0); + const page = head.records.slice(start, start + pageSize); + const next = start + pageSize; + return json({ + semver: head.semver, + schemas: Object.fromEntries( + Object.entries(head.schemas ?? { Pub: PubSchema }).map(([t, s]) => [ + t, + hashSchema(s), + ]), + ), + records: page, + pagination: { + hasMore: next < head.records.length, + nextCursor: next < head.records.length ? String(next) : null, + }, + }); + } + if (method === 'POST' && path === `${COLLECTION}/push`) { + fake.opens.push(JSON.parse(init.body)); + const custom = opts.open?.(); + if (custom) return custom; + sessionCount += 1; + return json({ + session_id: `s${sessionCount}`, + base: head?.semver ?? null, + needed_files: [], + limits: opts.limits ?? {}, + }); + } + const sessionMatch = path.match(/\/push\/(s\d+)(?:\/(records|deletes|commit))?$/); + if (sessionMatch) { + const [, , action] = sessionMatch; + if (method === 'POST' && (action === 'records' || action === 'deletes')) { + const lines = String(init.body) + .split('\n') + .filter(Boolean) + .map((l) => JSON.parse(l)); + fake.batches.push({ kind: action, lines: lines.length }); + fake.lines[action].push(...lines); + if (action === 'records' && opts.recordsResponse) return opts.recordsResponse(); + return json({ received: lines.length }); + } + if (method === 'POST' && action === 'commit') { + return opts.commit + ? opts.commit() + : json( + { semver: 'v1.1.0', hash: 'ulv2:x', recordCount: 1, fileCount: 0 }, + 201, + ); + } + if (method === 'DELETE') return json({ ok: true }); + if (method === 'GET') return json(states.shift() ?? { status: 'committing' }); + } + if (method === 'POST' && path === `${COLLECTION}/files/uploads`) { + return json({ id: 'u1', url: 'https://storage.test/put/u1', expiresIn: 3600 }, 201); + } + if (url === 'https://storage.test/put/u1') return new Response(null, { status: 200 }); + if (method === 'POST' && path === `${COLLECTION}/files/uploads/u1/complete`) { + return json({ id: 'u1', status: 'verifying' }, 202); + } + if (method === 'GET' && path === `${COLLECTION}/files/uploads/u1`) { + return json({ id: 'u1', status: uploadStatuses.shift() ?? 'verified' }); + } + if (method === 'PUT' && path.startsWith(`${COLLECTION}/files/`)) { + opts.onPut?.(url); + return json({ status: 'stored' }, 201); + } + throw new Error(`unexpected fetch: ${method} ${url}`); + }), + ); + return fake; }; afterEach(() => { vi.unstubAllGlobals(); }); +describe('underlay/client — diff', () => { + it('upserts new and changed records, deletes vanished ones, and leaves the rest alone', () => { + const same = rec('a', 'same'); + const changed = rec('b', 'new'); + const added = rec('c', 'added'); + const head = { + semver: 'v1.0.0', + metadata: null, + schemaHashes: { Pub: hashSchema(PubSchema), Note: hashSchema(NoteSchema) }, + records: new Map( + [ + ...headOf([same, rec('b', 'old'), rec('gone', 'x')]), + // A type we no longer push: dropping the type removes it, so no delete line. + { id: 'n1', type: 'Note', hash: 'h' }, + ].map((r) => [`${r.type}\u0000${r.id}`, { ...r, private: false }]), + ), + }; + const manifest = headOf([same, changed, added]); + const plan = diffAgainstHead(manifest, { Pub: PubSchema }, hashSchema, head); + expect(plan.upserts.map((u) => u.id).sort()).toEqual(['b', 'c']); + expect(plan.deletes).toEqual([{ type: 'Pub', id: 'gone' }]); + // The Note type is gone from our schema set. + expect(plan.schemasChanged).toBe(true); + }); + + it('upserts a record whose hash is unchanged but whose access set moved', () => { + const r = rec('a', 'x'); + const head = { + semver: 'v1.0.0', + metadata: null, + schemaHashes: { Pub: hashSchema(PubSchema) }, + records: new Map([ + [`Pub\u0000a`, { id: 'a', type: 'Pub', hash: hashRecord(r).hash, private: true }], + ]), + }; + const plan = diffAgainstHead(headOf([r]), { Pub: PubSchema }, hashSchema, head); + expect(plan.upserts.map((u) => u.id)).toEqual(['a']); + expect(plan.schemasChanged).toBe(false); + }); + + it('sends only the metadata fields that differ, treating null as absent', () => { + expect(metadataChanges({ readme: 'x' }, { readme: 'x', description: 'd' })).toBeNull(); + expect(metadataChanges({ readme: null }, null)).toBeNull(); + expect(metadataChanges({ readme: null }, { readme: 'old' })).toEqual({ readme: null }); + expect(metadataChanges({ readme: 'new' }, { readme: 'old' })).toEqual({ readme: 'new' }); + }); +}); + +describe('underlay/client — push', () => { + it('first push: opens against base null and uploads every record', async () => { + const fake = fakeUnderlay({ head: null }); + const records = [rec('a', '1'), rec('b', '2')]; + const result = await makeClient().push(payloadOf(records), 'msg', { readme: 'hi' }); + + expect(result).toMatchObject({ status: 'committed', semver: 'v1.1.0' }); + expect(fake.opens[0]).toMatchObject({ + base: null, + message: 'msg', + app_id: 'pubpub', + schemas: { Pub: PubSchema }, + metadata_patch: { readme: 'hi' }, + }); + // Negotiate-era fields must not leak into the open: `files` would make referenced files + // members-only, and `manifest` no longer exists. + expect(fake.opens[0]).not.toHaveProperty('files'); + expect(fake.opens[0]).not.toHaveProperty('manifest'); + expect(fake.lines.records.map((l) => l.id)).toEqual(['a', 'b']); + expect(fake.batches.some((b) => b.kind === 'deletes')).toBe(false); + }); + + it('uploads only the diff against the head', async () => { + const fake = fakeUnderlay({ + head: { + semver: 'v2.0.0', + records: headOf([rec('a', '1'), rec('b', 'old'), rec('gone', 'x')]), + }, + }); + await makeClient().push(payloadOf([rec('a', '1'), rec('b', 'new'), rec('c', '3')]), 'msg'); + + expect(fake.opens[0].base).toBe('v2.0.0'); + expect(fake.lines.records.map((l) => l.id).sort()).toEqual(['b', 'c']); + expect(fake.lines.deletes).toEqual([{ type: 'Pub', id: 'gone' }]); + }); + + it('does not open a session when the head already matches', async () => { + const records = [rec('a', '1')]; + const fake = fakeUnderlay({ + head: { semver: 'v2.0.0', records: headOf(records), metadata: { readme: 'same' } }, + }); + const result = await makeClient().push(payloadOf(records), 'msg', { readme: 'same' }); + + expect(result.status).toBe('noop'); + expect(fake.opens).toHaveLength(0); + }); + + it('pushes a readme-only change as a metadata patch', async () => { + const records = [rec('a', '1')]; + const fake = fakeUnderlay({ + head: { + semver: 'v2.0.0', + records: headOf(records), + metadata: { readme: 'old', license: 'CC-BY' }, + }, + }); + await makeClient().push(payloadOf(records), 'msg', { readme: 'new' }); + + // Only the readme: the license set in Underlay is not ours to overwrite. + expect(fake.opens[0].metadata_patch).toEqual({ readme: 'new' }); + expect(fake.lines.records).toHaveLength(0); + }); + + it('reads every manifest page, pinned to the head semver', async () => { + const records = [rec('a', '1'), rec('b', '2'), rec('c', '3')]; + const fake = fakeUnderlay({ + head: { semver: 'v2.0.0', records: headOf(records) }, + manifestPageSize: 2, + }); + const result = await makeClient().push(payloadOf(records), 'msg'); + + // Records on page two were seen too, so nothing looked new. + expect(result.status).toBe('noop'); + const manifestCalls = fake.calls.filter((c) => c.includes('/manifest')); + expect(manifestCalls).toHaveLength(2); + expect(manifestCalls.every((c) => c.includes('/versions/v2.0.0/manifest'))).toBe(true); + }); + + it('splits uploads by the session limits on lines and bytes', async () => { + const records = Array.from({ length: 5 }, (_, i) => rec(`r${i}`, 'x'.repeat(50))); + const fake = fakeUnderlay({ + head: null, + limits: { batch_lines: 2, batch_bytes: 1_000_000 }, + }); + await makeClient().push(payloadOf(records), 'msg'); + expect(fake.batches.map((b) => b.lines)).toEqual([2, 2, 1]); + + vi.unstubAllGlobals(); + // Each line is ~85 bytes, so a 200-byte budget fits two. + const byBytes = fakeUnderlay({ + head: null, + limits: { batch_lines: 100, batch_bytes: 200 }, + }); + await makeClient().push(payloadOf(records), 'msg'); + expect(byBytes.batches.map((b) => b.lines)).toEqual([2, 2, 1]); + }); + + it('sends the private flag on records that carry it', async () => { + const fake = fakeUnderlay({ head: null }); + await makeClient().push( + payloadOf([{ ...rec('a', '1'), private: true }, rec('b', '2')]), + 'msg', + ); + expect(fake.lines.records).toEqual([ + { id: 'a', type: 'Pub', data: { slug: '1' }, private: true }, + { id: 'b', type: 'Pub', data: { slug: '2' } }, + ]); + }); + + it('produces a cache-hit record the head lacks through resolveRecord', async () => { + const cached = rec('cached', 'from-cache'); + const fake = fakeUnderlay({ head: { semver: 'v1.0.0', records: [] } }); + const resolveRecord = vi.fn(async () => cached); + await makeClient().push( + payloadOf([], { manifest: headOf([cached]), resolveRecord }), + 'msg', + ); + expect(resolveRecord).toHaveBeenCalledWith( + expect.objectContaining({ id: 'cached', type: 'Pub' }), + ); + expect(fake.lines.records.map((l) => l.id)).toEqual(['cached']); + }); + + it('fails clearly, and abandons the session, when a record cannot be produced', async () => { + fakeUnderlay({ head: null }); + await expect( + makeClient().push(payloadOf([], { manifest: headOf([rec('lost', 'x')]) }), 'msg'), + ).rejects.toThrow(/can no longer produce/); + }); + + it('abandons the session when the server refuses an upload', async () => { + const fake = fakeUnderlay({ + head: null, + recordsResponse: () => + json( + { error: 'Invalid records', validationErrors: [{ line: 1 }], totalErrors: 1 }, + 422, + ), + }); + await expect(makeClient().push(payloadOf([rec('a', '1')]), 'msg')).rejects.toThrow( + /refused the records upload/, + ); + expect(fake.calls).toContain(`DELETE ${COLLECTION}/push/s1`); + }); + + it('starts over against the new head after a version conflict at open', async () => { + let opens = 0; + const fake = fakeUnderlay({ + head: { semver: 'v1.0.0', records: [] }, + open: () => { + opens += 1; + return opens === 1 + ? json({ error: 'Version conflict', currentVersion: 'v1.1.0' }, 409) + : null; + }, + }); + const result = await makeClient().push(payloadOf([rec('a', '1')]), 'msg'); + expect(result.status).toBe('committed'); + expect(fake.opens).toHaveLength(2); + // The head was re-read before the second open. + expect(fake.calls.filter((c) => c.endsWith('/versions/latest'))).toHaveLength(2); + }); + + it('treats "No changes detected" at commit as a noop, not a conflict', async () => { + const fake = fakeUnderlay({ + head: null, + commit: () => json({ error: 'No changes detected', hash: 'ulv2:x' }, 409), + }); + const result = await makeClient().push(payloadOf([rec('a', '1')]), 'msg'); + expect(result.status).toBe('noop'); + expect(fake.opens).toHaveLength(1); + }); + + it('sends credentials when reading the head', async () => { + const fake = fakeUnderlay({ head: { semver: 'v3.0.0', records: [] } }); + const head = await makeClient().readHead(); + expect(head?.semver).toBe('v3.0.0'); + // Anonymous, these 404 on a private collection and every record looks new. + expect(fake.headers.filter((h) => h.url.includes('/versions/')).every((h) => h.auth)).toBe( + true, + ); + }); +}); + describe('underlay/client — async commit', () => { it('requests an async commit and polls the session until it is committed', async () => { - const { calls } = stubFetch({ - commit: json({ session_id: 's1', status: 'committing' }, 202), + const fake = fakeUnderlay({ + head: null, + commit: () => json({ session_id: 's1', status: 'committing' }, 202), sessionStates: [ { status: 'committing', result: null, error: null }, { status: 'committed', - result: { semver: '1.1.0', hash: 'abc', recordCount: 1, fileCount: 0 }, + result: { + semver: 'v1.1.0', + hash: 'abc', + recordCount: 1, + fileCount: 0, + changes: { added: 1, updated: 0, removed: 0 }, + }, }, ], }); - const result = await makeClient().push(payload(), '1.0.0', 'msg'); + const result = await makeClient().push(payloadOf([rec('a', '1')]), 'msg'); expect(result).toEqual({ status: 'committed', - semver: '1.1.0', + semver: 'v1.1.0', hash: 'abc', recordCount: 1, fileCount: 0, + changes: { added: 1, updated: 0, removed: 0 }, }); // The commit must opt in to async, or the request hangs open past the 60s client timeout. - expect(calls.some((c) => c.includes('/commit?async=true'))).toBe(true); + expect(fake.calls.some((c) => c.includes('/push/s1/commit?async=true'))).toBe(true); // It kept polling through the non-terminal state rather than giving up on the first read. - expect(calls.filter((c) => c === `GET ${COLLECTION}/versions/negotiate/s1`)).toHaveLength( - 2, - ); + expect(fake.calls.filter((c) => c === `GET ${COLLECTION}/push/s1`)).toHaveLength(2); }); - it('still accepts a synchronous 201 from an Underlay that predates async commit', async () => { - const { calls } = stubFetch({ - commit: json({ semver: '2.0.0', hash: 'def', recordCount: 1, fileCount: 0 }, 201), - }); - - const result = await makeClient().push(payload(), '1.0.0', 'msg'); - - expect(result).toEqual({ - status: 'committed', - semver: '2.0.0', - hash: 'def', - recordCount: 1, - fileCount: 0, + it('accepts a synchronous 201', async () => { + const fake = fakeUnderlay({ + head: null, + commit: () => + json({ semver: 'v2.0.0', hash: 'def', recordCount: 1, fileCount: 0 }, 201), }); - // No polling — the version came back inline. - expect(calls.some((c) => c === `GET ${COLLECTION}/versions/negotiate/s1`)).toBe(false); + const result = await makeClient().push(payloadOf([rec('a', '1')]), 'msg'); + expect(result).toMatchObject({ status: 'committed', semver: 'v2.0.0' }); + expect(fake.calls.some((c) => c === `GET ${COLLECTION}/push/s1`)).toBe(false); }); - it('surfaces the server error when the async finalize fails', async () => { - stubFetch({ - commit: json({ session_id: 's1', status: 'committing' }, 202), + it('surfaces the server error when the async commit fails', async () => { + fakeUnderlay({ + head: null, + commit: () => json({ session_id: 's1', status: 'committing' }, 202), sessionStates: [ - { status: 'failed', error: { statusCode: 400, error: 'Manifest incomplete' } }, + { status: 'failed', error: { statusCode: 422, error: 'Schema validation failed' } }, ], }); - - await expect(makeClient().push(payload(), '1.0.0', 'msg')).rejects.toThrow( - /Manifest incomplete/, + await expect(makeClient().push(payloadOf([rec('a', '1')]), 'msg')).rejects.toThrow( + /Schema validation failed/, ); }); it('fails with a clear message when the session expires mid-commit', async () => { - stubFetch({ - commit: json({ session_id: 's1', status: 'committing' }, 202), + fakeUnderlay({ + head: null, + commit: () => json({ session_id: 's1', status: 'committing' }, 202), sessionStates: [{ status: 'expired' }], }); - - await expect(makeClient().push(payload(), '1.0.0', 'msg')).rejects.toThrow(/expired/i); + await expect(makeClient().push(payloadOf([rec('a', '1')]), 'msg')).rejects.toThrow( + /expired/i, + ); }); it('gives up with an actionable message if the commit never reaches a terminal state', async () => { - stubFetch({ - commit: json({ session_id: 's1', status: 'committing' }, 202), - sessionStates: [], // always 'committing' + fakeUnderlay({ + head: null, + commit: () => json({ session_id: 's1', status: 'committing' }, 202), + sessionStates: [], }); - - await expect(makeClient().push(payload(), '1.0.0', 'msg')).rejects.toThrow( + await expect(makeClient().push(payloadOf([rec('a', '1')]), 'msg')).rejects.toThrow( /did not finish within/, ); }); - it('uploads files the failed commit asked for, then re-negotiates and succeeds', async () => { + it('uploads files a failed commit asked for, then starts a new session and succeeds', async () => { const uploaded: string[] = []; - let commitCount = 0; - // First commit fails wanting a file; after we upload it, the retried push commits. - const { calls } = stubFetch({ + let commits = 0; + const fake = fakeUnderlay({ + head: null, commit: () => { - commitCount += 1; - return json({ session_id: 's1', status: 'committing' }, 202); + commits += 1; + return json({ session_id: `s${commits}`, status: 'committing' }, 202); }, sessionStates: [ - { status: 'failed', error: { filesNeeded: ['sha256:f1'] } }, + { + status: 'failed', + error: { statusCode: 422, error: 'Missing files', filesNeeded: ['f1'] }, + }, { status: 'committed', - result: { semver: '1.2.0', hash: 'ghi', recordCount: 1, fileCount: 1 }, + result: { semver: 'v1.2.0', hash: 'ghi', recordCount: 1, fileCount: 1 }, }, ], onPut: (url) => uploaded.push(url), }); - const withFile: UnderlayPushPayload = { - ...payload(), - files: [{ hash: 'f1', contentType: 'text/html', bytes: Buffer.from('x') }], - fileHashes: ['f1'], - }; - - const result = await makeClient().push(withFile, '1.0.0', 'msg'); + const result = await makeClient().push( + payloadOf([rec('a', '1')], { + files: [{ hash: 'f1', contentType: 'text/html', bytes: Buffer.from('x') }], + fileHashes: ['f1'], + }), + 'msg', + ); expect(result.status).toBe('committed'); - expect(uploaded.some((u) => u.includes('sha256:f1'))).toBe(true); - // The failed session can't be re-committed, so recovery means a second negotiate. - expect(calls.filter((c) => c === `POST ${COLLECTION}/versions/negotiate`)).toHaveLength(2); - expect(commitCount).toBe(2); + expect(uploaded.some((u) => u.endsWith('/files/f1'))).toBe(true); + // The failed session can't be committed again, so recovery is a second session. + expect(fake.opens).toHaveLength(2); + expect(commits).toBe(2); }); - it('does not retry forever when the needed file cannot be produced', async () => { - stubFetch({ - commit: json({ session_id: 's1', status: 'committing' }, 202), - sessionStates: [{ status: 'failed', error: { filesNeeded: ['sha256:missing'] } }], + it('does not retry when a missing file cannot be produced', async () => { + const fake = fakeUnderlay({ + head: null, + commit: () => json({ error: 'Missing files', filesNeeded: ['missing'] }, 422), }); - - await expect(makeClient().push(payload(), '1.0.0', 'msg')).rejects.toThrow( + await expect(makeClient().push(payloadOf([rec('a', '1')]), 'msg')).rejects.toThrow( /can no longer produce/, ); + expect(fake.opens).toHaveLength(1); }); -}); -describe('underlay/client — authenticated reads', () => { - it('sends credentials when reading the latest version', async () => { - const seen: { url: string; auth: boolean }[] = []; - vi.stubGlobal( - 'fetch', - vi.fn(async (input: any, init?: any) => { - const headers = new Headers(init?.headers); - seen.push({ url: String(input), auth: headers.has('Authorization') }); - if (String(input).endsWith('/versions/latest')) { - return json({ semver: '3.0.0' }); + it('keeps polling when a session read fails, instead of failing a commit that is still running', async () => { + let polls = 0; + fakeUnderlay({ head: null, commit: () => json({ session_id: 's1' }, 202) }); + const fakeFetch = globalThis.fetch as unknown as ReturnType; + const inner = fakeFetch.getMockImplementation()! as ( + input: any, + init?: any, + ) => Promise; + fakeFetch.mockImplementation(async (input: any, init?: any) => { + if (String(input) === `${COLLECTION}/push/s1` && (init?.method ?? 'GET') === 'GET') { + polls += 1; + // A 200 whose body is not JSON: `json()` throws, which reaches the poll loop's catch + // directly. (Throwing from fetch would be absorbed by `request`'s own retry.) + if (polls <= 3) { + return new Response('502 upstream', { status: 200 }); } - return json({}, 404); - }), + return json({ + status: 'committed', + result: { semver: 'v1.1.0', hash: 'h', recordCount: 1, fileCount: 0 }, + }); + } + return inner(input, init); + }); + + const result = await makeClient().push(payloadOf([rec('a', '1')]), 'msg'); + expect(result).toMatchObject({ status: 'committed', semver: 'v1.1.0' }); + expect(polls).toBeGreaterThan(3); + }); +}); + +describe('underlay/client — files', () => { + it('uploads a file over 32 MB through a presigned upload, without sending the API key to storage', async () => { + const fake = fakeUnderlay({ head: null, upload: { statuses: ['verifying', 'verified'] } }); + const bytes = Buffer.alloc(32 * 1024 * 1024 + 1); + await makeClient().putFile({ hash: 'big', contentType: 'application/pdf', bytes }); + + expect(fake.calls).toEqual([ + `POST ${COLLECTION}/files/uploads`, + 'PUT https://storage.test/put/u1', + `POST ${COLLECTION}/files/uploads/u1/complete`, + `GET ${COLLECTION}/files/uploads/u1`, + `GET ${COLLECTION}/files/uploads/u1`, + ]); + expect(fake.headers.find((h) => h.url.startsWith('https://storage.test'))?.auth).toBe( + false, ); + }); - const base = await makeClient().getBaseVersion(); - expect(base).toBe('3.0.0'); + it('fails when the server cannot verify a presigned upload', async () => { + fakeUnderlay({ head: null, upload: { statuses: ['failed'] } }); + await expect( + makeClient().putFile({ + hash: 'big', + contentType: 'application/pdf', + bytes: Buffer.alloc(32 * 1024 * 1024 + 1), + }), + ).rejects.toThrow(/could not verify/); + }); - // Anonymous, this 404s on a private collection and is misread as "no versions yet" — which - // then pushes base_version: null and dies on a 409 conflict. - const call = seen.find((c) => c.url.endsWith('/versions/latest')); - expect(call?.auth).toBe(true); + it('uploads small files with a plain PUT', async () => { + const fake = fakeUnderlay({ head: null }); + await makeClient().putFile({ + hash: 'small', + contentType: 'text/html', + bytes: Buffer.from('x'), + }); + expect(fake.calls).toEqual([`PUT ${COLLECTION}/files/small`]); }); +}); +describe('underlay/client — connection', () => { it('sends credentials when checking whether the collection exists', async () => { const seen: { url: string; auth: boolean }[] = []; vi.stubGlobal( 'fetch', vi.fn(async (input: any, init?: any) => { - const headers = new Headers(init?.headers); - seen.push({ url: String(input), auth: headers.has('Authorization') }); + seen.push({ + url: String(input), + auth: new Headers(init?.headers).has('Authorization'), + }); return json({ slug: 'coll' }); }), ); - await makeClient().ensureCollection(); expect(seen[0]?.auth).toBe(true); // It existed, so nothing was created. expect(seen.some((c) => c.url.endsWith('/collections'))).toBe(false); }); -}); -describe('underlay/client — poll resilience and missing-file diagnosis', () => { - it('keeps polling when a session read fails, instead of failing a commit that is still running', async () => { - let polls = 0; + it("checks the organization against /accounts/me's orgs", async () => { vi.stubGlobal( 'fetch', - vi.fn(async (input: any) => { - const url = String(input); - if (url.endsWith('/versions/latest')) return json({ semver: '1.0.0' }); - if (url.endsWith('/versions/negotiate')) { - return json({ session_id: 's1', needed_records: [], needed_files: [] }); - } - if (url.includes('/commit')) return json({ session_id: 's1' }, 202); - if (url.endsWith('/versions/negotiate/s1')) { - polls += 1; - // A 200 whose body is not JSON: `json()` throws, which reaches the poll loop's - // catch directly. (Throwing from fetch instead would be absorbed by `request`'s - // own retry, so it would never exercise this path.) - if (polls <= 3) { - return new Response('502 upstream', { - status: 200, - headers: { 'Content-Type': 'text/html' }, - }); - } - return json({ - status: 'committed', - result: { semver: '1.1.0', hash: 'h', recordCount: 1, fileCount: 0 }, - }); - } - throw new Error(`unexpected ${url}`); - }), - ); - - const result = await makeClient().push(payload(), '1.0.0', 'msg'); - expect(result).toMatchObject({ status: 'committed', semver: '1.1.0' }); - expect(polls).toBeGreaterThan(3); - }); - - it('reports which files could not be produced rather than a generic commit failure', async () => { - vi.stubGlobal( - 'fetch', - vi.fn(async (input: any) => { - const url = String(input); - if (url.endsWith('/versions/latest')) return json({ semver: '1.0.0' }); - if (url.endsWith('/versions/negotiate')) { - return json({ session_id: 's1', needed_records: [], needed_files: [] }); - } - // Synchronous 422 naming a file the push cannot regenerate. - if (url.includes('/commit')) { - return json({ error: 'missing', filesNeeded: ['sha256:gone'] }, 422); - } - throw new Error(`unexpected ${url}`); - }), - ); - - await expect(makeClient().push(payload(), '1.0.0', 'msg')).rejects.toThrow( - /can no longer produce/, + vi.fn(async (input: any) => + String(input).endsWith('/accounts/me') + ? json({ id: 'u', orgs: [{ slug: 'someone-else', name: 'Else' }] }) + : json({ slug: 'coll' }), + ), ); + const check = await makeClient().verifyConnection(); + expect(check.ok).toBe(false); + expect(check.message).toMatch(/does not have access to the organization "org"/); }); }); diff --git a/server/underlay/__tests__/hash.test.ts b/server/underlay/__tests__/hash.test.ts index a612d5b91..17de9cbe0 100644 --- a/server/underlay/__tests__/hash.test.ts +++ b/server/underlay/__tests__/hash.test.ts @@ -1,6 +1,6 @@ import { describe, expect, it } from 'vitest'; -import { canonicalize, hashBytes, hashRecord, hashSchema } from '../hash'; +import { canonicalize, hashBytes, hashRecord, hashSchema, jcs } from '../hash'; describe('underlay/hash', () => { it('canonicalizes the spec example to sorted-key JSON', () => { @@ -60,6 +60,25 @@ describe('underlay/hash', () => { expect(withExtra.hash).toBe(without.hash); }); + it('sorts integer-like keys as strings, as JCS does (not in JavaScript enumeration order)', () => { + const { canonical } = hashRecord({ + id: 'x', + type: 'T', + data: { b: 1, '10': 'ten', '9': 'nine', a: { '2': true, '1': false } }, + }); + // "10" < "9" < "a" < "b" by UTF-16 code units. A sorted object passed to JSON.stringify would + // put "9" before "10". + expect(canonical).toBe( + '{"id":"x","type":"T","data":{"10":"ten","9":"nine","a":{"1":false,"2":true},"b":1}}', + ); + }); + + it('matches JSON.stringify for strings and numbers', () => { + expect(jcs({ s: 'é\n"', n: [1, 1.5, -0, 1e21] })).toBe( + '{"n":[1,1.5,0,1e+21],"s":"é\\n\\""}', + ); + }); + it('hashes schemas by canonicalized content', () => { expect(hashSchema({ type: 'object', properties: { title: { type: 'string' } } })).toBe( '75a323804435b510db3f8cc7fda9a750763b0ee894a8e28a00ffafc4823ee03a', diff --git a/server/underlay/__tests__/incremental.test.ts b/server/underlay/__tests__/incremental.test.ts index 2db97bb9f..3b5a0e120 100644 --- a/server/underlay/__tests__/incremental.test.ts +++ b/server/underlay/__tests__/incremental.test.ts @@ -715,8 +715,8 @@ describe('underlay/incremental — resumed push re-supplies records without hoar const pubUpdatedAt = { p1: '2026-01-05T00:00:00.000Z', p2: '2026-01-06T00:00:00.000Z' }; /** - * A checkpointed pub has had its FILES uploaded, but records are only sent after negotiate — so a - * push that died mid-mapping leaves cache entries whose records the server has never seen. The + * A checkpointed pub has had its FILES uploaded, but records are only sent in the push session — + * so a push that died mid-mapping leaves cache entries whose records the server has never seen. The * next push must be able to produce them on demand, without retaining the file bytes that * re-mapping also yields. */ @@ -786,3 +786,58 @@ describe('underlay/incremental — resumed push re-supplies records without hoar expect(calls.p1).toBe(2); }); }); + +describe('underlay/incremental — cached hashes from an older hash function', () => { + const pubs = [makePub('p1', 1)]; + const pubUpdatedAt = { p1: '2026-01-05T00:00:00.000Z' }; + + /** A real cache entry whose stored record hashes no longer match what the pub maps to. */ + const staleEntries = async () => { + const first = await buildIncrementalPush({ + community, + collections: [], + pubs, + pubUpdatedAt, + options: OPTIONS, + cacheEntries: [], + mapPub: makeMapPub({}), + }); + return first.cacheUpserts.map((entry) => ({ + ...entry, + recordHashes: Object.fromEntries( + Object.entries(entry.recordHashes).map(([id, r]) => [ + id, + { ...r, hash: `old-${r.hash}` }, + ]), + ), + })); + }; + + it('resolves a record by (type, id) and queues a corrected cache entry', async () => { + const entries = await staleEntries(); + const resumed = await buildIncrementalPush({ + community, + collections: [], + pubs, + pubUpdatedAt, + options: OPTIONS, + cacheEntries: entries, + mapPub: makeMapPub({}), + }); + expect(resumed.stats.cacheHits).toBe(1); + expect(resumed.cacheUpserts).toEqual([]); + + // The manifest carries the stale hash, so a lookup by hash alone finds nothing… + const entry = resumed.payload.manifest!.find((m) => m.type === 'Pub' && m.id === 'p1')!; + expect(entry.hash.startsWith('old-')).toBe(true); + expect(await resumed.payload.resolveRecordByHash?.(entry.hash)).toBeNull(); + + // …but (type, id) still produces the record. + const record = await resumed.payload.resolveRecord?.(entry); + expect(record).toMatchObject({ id: 'p1', type: 'Pub' }); + + // And the entry is corrected, so the next push's manifest matches the head. + expect(resumed.cacheUpserts).toHaveLength(1); + expect(resumed.cacheUpserts[0].recordHashes.p1.hash).toBe(hashRecord(record!).hash); + }); +}); diff --git a/server/underlay/client.ts b/server/underlay/client.ts index 92e7540a4..2c75298c2 100644 --- a/server/underlay/client.ts +++ b/server/underlay/client.ts @@ -1,29 +1,53 @@ -import { hashRecord } from './hash'; +import { hashSchema, jcs } from './hash'; import { buildManifest, + type JsonSchema, + type ManifestEntry, type UnderlayFile, type UnderlayPushPayload, type UnderlayRecord, } from './mapping'; /** - * Minimal client for the Underlay push (negotiate) protocol. - * @see https://underlay.org — public/llms.txt §"Writing Data: The Push Flow" + * Minimal client for the Underlay delta push protocol. + * @see https://www.underlay.org/llms.txt §"Writing Data: The Push Flow" + * + * A push reads the head version's manifest, diffs our full record set against it, and uploads only + * the records that changed plus the (type, id) pairs that went away. The server's manifest is the + * source of truth for what it holds; PubPub's own push cache only decides what to re-render. * * Framework-free so it can be driven from a worker. Uses global fetch + AbortController; no * external HTTP dependency. */ -const DEFAULT_BASE_URL = 'https://underlay.org/api'; +// The apex domain 301s to www. fetch follows that by turning a POST into a GET and dropping the +// Authorization header across origins, so the default must be the www host itself. +const DEFAULT_BASE_URL = 'https://www.underlay.org/api'; const REQUEST_TIMEOUT_MS = 60_000; -const RECORD_BATCH_SIZE = 10_000; const MAX_RETRIES = 4; +/** Manifest page size: the server's maximum (each entry is ~120 bytes). */ +const MANIFEST_PAGE_LIMIT = 25_000; + +/** Fallbacks for a session whose `limits` omit a batch limit; the server's values win. */ +const DEFAULT_BATCH_LINES = 10_000; +const DEFAULT_BATCH_BYTES = 16 * 1024 * 1024; + /** - * Async-commit polling. Underlay validates every record, folds the version digests and writes - * version_records during a commit — minutes of work on a large collection, far past - * REQUEST_TIMEOUT_MS. Holding the request open would abort and then *retry* that work, so we ask - * for an async commit and poll the session instead. + * Largest file the plain `PUT …/files/:hash` accepts. Larger files go through a presigned upload. + * Files are streamed during mapping, before any session reports `limits.file_bytes`, so this is + * fixed here; a 413 from the plain PUT also falls back to the presigned upload. + */ +const SMALL_UPLOAD_BYTES = 32 * 1024 * 1024; +/** A presigned PUT of a large export is one long request; give it far more than the API timeout. */ +const LARGE_UPLOAD_TIMEOUT_MS = 30 * 60 * 1000; +/** How long to wait for the server to verify a presigned upload's hash. */ +const UPLOAD_VERIFY_TIMEOUT_MS = 10 * 60 * 1000; + +/** + * Async-commit polling. Underlay commits a large push in the background — minutes of work, far past + * REQUEST_TIMEOUT_MS. Holding the request open would abort and then *retry* that work, so we ask for + * an async commit and poll the session instead. */ const COMMIT_POLL_INITIAL_MS = 2_000; const COMMIT_POLL_MAX_MS = 15_000; @@ -40,14 +64,23 @@ export type PushClientOptions = { /** Identifies the pushing app + actor in the commit metadata. */ appId?: string; actorId?: string; - /** Async-commit poll timing. Overridable so tests don't wait real seconds. */ + /** Async-commit (and upload-verify) poll timing. Overridable so tests don't wait real seconds. */ pollIntervalMs?: number; pollTimeoutMs?: number; }; +export type PushChanges = { added: number; updated: number; removed: number }; + export type PushResult = | { status: 'noop'; reason: string } - | { status: 'committed'; semver: string; hash: string; recordCount: number; fileCount: number }; + | { + status: 'committed'; + semver: string; + hash: string; + recordCount: number; + fileCount: number; + changes?: PushChanges; + }; const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)); @@ -57,10 +90,10 @@ export class UnderlayPushError extends Error { public readonly statusCode?: number, public readonly detail?: unknown, /** - * The push can recover by re-negotiating from scratch. Set when a commit failed for a cause - * we've since fixed (missing files, now uploaded) but whose session can no longer be - * re-committed — Underlay only accepts a commit on an `open` session, and a failed async - * finalize leaves it `failed`. + * The push can recover by starting over against the current head: a version conflict + * (someone published after we read the manifest), or a commit refused for a cause we've + * since fixed (missing files, now uploaded). A refused commit leaves its session `failed`, + * and only an `open` session accepts a commit, so recovery always means a new session. */ public readonly retriable = false, ) { @@ -93,6 +126,9 @@ export const formatUnderlayError = (error: unknown): string => { return error instanceof Error ? error.message : String(error); }; +/** `/accounts/me` lists the caller's organizations as `orgs`. */ +type MeBody = { orgs?: { slug: string; name?: string | null }[] }; + /** One authenticated fetch, parsed as JSON, with every failure mode reported instead of thrown. */ const fetchJson = async ( url: string, @@ -164,7 +200,7 @@ export async function probeUnderlay( ); } const meText = await meResp.text(); - let me: { accounts?: { slug: string; name: string }[] }; + let me: MeBody; try { me = JSON.parse(meText); } catch { @@ -186,7 +222,7 @@ export async function probeUnderlay( meText.slice(0, 200), ); } - const accounts: UnderlayAccount[] = (me.accounts ?? []).map((a) => ({ + const accounts: UnderlayAccount[] = (me.orgs ?? []).map((a) => ({ slug: a.slug, name: a.name || a.slug, })); @@ -221,6 +257,92 @@ export async function probeUnderlay( } } +/** The head version as a delta push needs it: what the server holds, by (type, id). */ +export type UnderlayHead = { + semver: string; + metadata: Record | null; + /** type → schema hash (bare hex). */ + schemaHashes: Record; + records: Map; +}; + +export type PushPlan = { + /** Our entries the head lacks, holds with a different hash, or holds in the other access set. */ + upserts: ManifestEntry[]; + /** Pairs the head holds that we no longer have, limited to types we still push. */ + deletes: { type: string; id: string }[]; + schemasChanged: boolean; +}; + +const recordKey = (type: string, id: string) => `${type}\u0000${id}`; +const bareHash = (hash: string) => hash.replace(/^sha256:/, ''); + +/** + * Diff our full record set against the head. A delete is only sent for a type that stays in our + * schema set: dropping a type removes its records with it, and the server refuses a delete line + * whose type isn't in the session's type set. + */ +export const diffAgainstHead = ( + manifest: ManifestEntry[], + schemas: Record, + schemaHash: (schema: JsonSchema) => string, + head: UnderlayHead | null, +): PushPlan => { + const upserts: ManifestEntry[] = []; + const ours = new Set(); + for (const entry of manifest) { + const key = recordKey(entry.type, entry.id); + ours.add(key); + const theirs = head?.records.get(key); + if (!theirs || theirs.hash !== entry.hash || theirs.private !== Boolean(entry.private)) { + upserts.push(entry); + } + } + const deletes: PushPlan['deletes'] = []; + for (const [key, theirs] of head?.records ?? []) { + if (!ours.has(key) && theirs.type in schemas) { + deletes.push({ type: theirs.type, id: theirs.id }); + } + } + const theirSchemas = head?.schemaHashes ?? {}; + const schemasChanged = + !head || + Object.keys(theirSchemas).length !== Object.keys(schemas).length || + Object.entries(schemas).some(([type, schema]) => theirSchemas[type] !== schemaHash(schema)); + return { upserts, deletes, schemasChanged }; +}; + +/** + * The fields of `patch` whose value differs from the head's metadata, or null if none do. A null + * value means "absent" on either side, so clearing a readme that was never set is not a change. + */ +export const metadataChanges = ( + patch: Record | undefined, + headMetadata: Record | null, +): Record | null => { + if (!patch) { + return null; + } + const changed: Record = {}; + for (const [key, value] of Object.entries(patch)) { + if (jcs(value ?? null) !== jcs(headMetadata?.[key] ?? null)) { + changed[key] = value ?? null; + } + } + return Object.keys(changed).length > 0 ? changed : null; +}; + +type SessionLimits = { batch_lines?: number; batch_bytes?: number }; + +/** A commit refusal, as the synchronous response body or a failed session's `error`. */ +type CommitErrorBody = { + error?: string; + statusCode?: number; + currentVersion?: string | null; + filesNeeded?: string[]; + hash?: string; +}; + export class UnderlayClient { private readonly baseUrl: string; private readonly apiKey: string; @@ -250,13 +372,16 @@ export class UnderlayClient { private async request( url: string, init: RequestInit & { rawBody?: Buffer | string } = {}, - { auth = true }: { auth?: boolean } = {}, + { + auth = true, + timeoutMs = REQUEST_TIMEOUT_MS, + }: { auth?: boolean; timeoutMs?: number } = {}, ): Promise { const { rawBody, ...rest } = init; let lastError: unknown; for (let attempt = 0; attempt < MAX_RETRIES; attempt += 1) { const controller = new AbortController(); - const timeout = setTimeout(() => controller.abort(), REQUEST_TIMEOUT_MS); + const timeout = setTimeout(() => controller.abort(), timeoutMs); try { const headers = new Headers(rest.headers); if (auth) { @@ -308,27 +433,100 @@ export class UnderlayClient { } } + /** A response body as JSON when it is JSON, else its text (for error detail). */ + private async body(response: Response): Promise { + const text = await response.text(); + try { + return JSON.parse(text); + } catch { + return text.slice(0, 500); + } + } + /** - * Returns the latest version semver, or null if the collection has no versions yet. + * Read the head version and its full manifest (ids, types and hashes; no bodies). Returns null + * when the collection has no versions yet. + * + * Sent authenticated: Underlay resolves these routes through the caller's access, so an anonymous + * request against a PRIVATE collection 404s — indistinguishable from "no versions yet", which + * would make every record look new. Authenticated as a member, the manifest also lists the + * private set, marked `"private": true`. * - * Sent authenticated: Underlay resolves this route through the caller's access, so an anonymous - * request against a PRIVATE collection 404s — which is indistinguishable here from "no versions - * yet". That would push `base_version: null`, the server would reject the mismatch with a 409, - * and the conflict retry would re-read the same 404 and fail again. Every push after the first - * would die on a misleading version conflict the moment a collection was made private. + * Pages after the first are read at the head's semver rather than `latest`, so a version + * published mid-read can't splice two versions into one manifest. */ - async getBaseVersion(): Promise { - const response = await this.request(`${this.collectionPath()}/versions/latest`, { + async readHead(): Promise { + const latest = await this.request(`${this.collectionPath()}/versions/latest`, { method: 'GET', }); - if (response.status === 404) { + if (latest.status === 404) { return null; } - if (!response.ok) { - throw new UnderlayPushError('Failed to fetch latest version', response.status); + if (!latest.ok) { + throw new UnderlayPushError( + 'Failed to fetch the latest version', + latest.status, + await this.body(latest), + ); } - const body = await this.json<{ semver?: string }>(response); - return body.semver ?? null; + const version = await this.json<{ + semver: string; + metadata?: Record | null; + }>(latest); + + const records: UnderlayHead['records'] = new Map(); + let schemaHashes: Record = {}; + let cursor: string | null = null; + let firstPage = true; + do { + const params = new URLSearchParams({ limit: String(MANIFEST_PAGE_LIMIT) }); + if (cursor) { + params.set('cursor', cursor); + } + // biome-ignore lint/performance/noAwaitInLoops: manifest pages are cursor-chained + const response = await this.request( + `${this.collectionPath()}/versions/${encodeURIComponent(version.semver)}/manifest?${params}`, + { method: 'GET' }, + ); + if (!response.ok) { + throw new UnderlayPushError( + `Failed to read the manifest of ${version.semver}`, + response.status, + await this.body(response), + ); + } + // biome-ignore lint/performance/noAwaitInLoops: manifest pages are cursor-chained + const page = await this.json<{ + schemas?: Record; + records?: { id: string; type: string; hash: string; private?: boolean }[]; + pagination?: { hasMore?: boolean; nextCursor?: string | null }; + }>(response); + if (firstPage) { + schemaHashes = Object.fromEntries( + Object.entries(page.schemas ?? {}).map(([type, hash]) => [ + type, + bareHash(hash), + ]), + ); + firstPage = false; + } + for (const r of page.records ?? []) { + records.set(recordKey(r.type, r.id), { + id: r.id, + type: r.type, + hash: bareHash(r.hash), + private: r.private === true, + }); + } + cursor = page.pagination?.hasMore ? (page.pagination.nextCursor ?? null) : null; + } while (cursor); + + return { + semver: version.semver, + metadata: version.metadata ?? null, + schemaHashes, + records, + }; } /** @@ -384,9 +582,7 @@ export class UnderlayClient { }); // 2. Organization access. - const accounts = ((me.body as { accounts?: { slug: string }[] })?.accounts ?? []).map( - (a) => a.slug, - ); + const accounts = ((me.body as MeBody)?.orgs ?? []).map((a) => a.slug); if (accounts.length > 0 && !accounts.includes(this.owner)) { steps.push({ name: 'organization', @@ -413,7 +609,6 @@ export class UnderlayClient { }); return fail(`Could not check the collection: ${col.error}`); } - console.log('col', col); if (col.kind === 'json') { steps.push({ name: 'collection', @@ -450,8 +645,8 @@ export class UnderlayClient { /** Ensure the collection exists, creating it under the org if missing. */ async ensureCollection(): Promise { - // Authenticated for the same reason as getBaseVersion: an anonymous probe of a private - // collection 404s, sending us down the create path for something that already exists. + // Authenticated for the same reason as readHead: an anonymous probe of a private collection + // 404s, sending us down the create path for something that already exists. const response = await this.request(this.collectionPath(), { method: 'GET' }); if (response.ok) { return; @@ -475,177 +670,316 @@ export class UnderlayClient { /** * Upload one file's bytes, content-addressed by hash. Public so a push can stream files up as it - * maps them instead of accumulating every byte in memory until commit; the endpoint is idempotent - * (an already-present hash is a cheap no-op), so re-uploading across attempts is harmless. + * maps them instead of accumulating every byte in memory until commit. Uploading a file the + * collection already holds is harmless, so re-uploading across attempts is too. A file uploaded + * and verified for the collection counts as held even if no version references it yet. */ async putFile(file: UnderlayFile): Promise { - return this.uploadFile(file); - } - - private async uploadFile(file: UnderlayFile): Promise { + if (file.bytes.length > SMALL_UPLOAD_BYTES) { + return this.uploadLargeFile(file); + } // Send the file's real content type — Underlay reports back whatever it received, so this is // what downstream consumers see. Falls back to octet-stream only when we truly don't know. - const response = await this.request(`${this.collectionPath()}/files/sha256:${file.hash}`, { + const response = await this.request(`${this.collectionPath()}/files/${file.hash}`, { method: 'PUT', headers: { 'Content-Type': file.contentType || 'application/octet-stream' }, rawBody: file.bytes, }); + if (response.status === 413) { + // A server whose small-upload limit is below ours. + return this.uploadLargeFile(file); + } if (!response.ok) { - const detail = await response.text(); throw new UnderlayPushError( `Failed to upload file ${file.hash}`, response.status, - detail, + await this.body(response), ); } } - private async sendRecords( - sessionId: string, - neededHashes: string[], - recordByHash: Map, - resolveRecordByHash?: (hash: string) => Promise, - ) { - // Resolve every needed hash to a record — from the in-memory set, or (incremental path) by - // lazily re-hydrating the owning pub when the server asks for a record we didn't materialize. - const toSend: UnderlayRecord[] = []; - for (const hash of neededHashes) { - let rec = recordByHash.get(hash); - if (!rec && resolveRecordByHash) { - // biome-ignore lint/performance/noAwaitInLoops: lazy hydration is bounded by needed records - rec = (await resolveRecordByHash(hash)) ?? undefined; + /** + * Presigned upload, for files over the plain PUT's limit (large PDF/EPUB exports): start an + * upload, PUT the bytes straight to storage, complete it, and wait until the server has hashed + * the bytes and marked the file verified. Up to 5 GiB is one PUT; beyond that the server asks for + * a multipart upload, which no PubPub export comes near, so that is refused clearly. + */ + private async uploadLargeFile(file: UnderlayFile): Promise { + const uploads = `${this.collectionPath()}/files/uploads`; + const start = await this.request(uploads, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ + hash: file.hash, + size: file.bytes.length, + mimeType: file.contentType || 'application/octet-stream', + }), + }); + if (!start.ok) { + throw new UnderlayPushError( + `Failed to start the upload of file ${file.hash}`, + start.status, + await this.body(start), + ); + } + const ticket = await this.json<{ id: string; url?: string }>(start); + if (!ticket.url) { + throw new UnderlayPushError( + `File ${file.hash} (${file.bytes.length} bytes) needs a multipart upload, which PubPub does not support.`, + ); + } + + // The presigned URL carries its own authorization; our API key must not go to storage. + const put = await this.request( + ticket.url, + { method: 'PUT', rawBody: file.bytes }, + { auth: false, timeoutMs: LARGE_UPLOAD_TIMEOUT_MS }, + ); + if (!put.ok) { + throw new UnderlayPushError( + `Failed to upload file ${file.hash} to storage`, + put.status, + await this.body(put), + ); + } + + const complete = await this.request(`${uploads}/${ticket.id}/complete`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: '{}', + }); + if (!complete.ok) { + throw new UnderlayPushError( + `Failed to complete the upload of file ${file.hash}`, + complete.status, + await this.body(complete), + ); + } + + const deadline = Date.now() + Math.min(this.pollTimeoutMs, UPLOAD_VERIFY_TIMEOUT_MS); + let delay = this.pollIntervalMs; + while (Date.now() < deadline) { + // biome-ignore lint/performance/noAwaitInLoops: polling is inherently sequential + await sleep(delay); + delay = Math.min(COMMIT_POLL_MAX_MS, Math.round(delay * 1.5)); + let status: { status?: string; error?: unknown }; + try { + const response = await this.request(`${uploads}/${ticket.id}`, { method: 'GET' }); + if (!response.ok) { + continue; + } + status = await this.json(response); + } catch { + // A failed read isn't a failed upload; the server verifies regardless. + continue; + } + if (status.status === 'verified') { + return; + } + if (status.status === 'failed') { + throw new UnderlayPushError( + `Underlay could not verify the upload of file ${file.hash}`, + undefined, + status.error, + ); + } + } + throw new UnderlayPushError(`The upload of file ${file.hash} was not verified in time.`); + } + + /** + * Produce the body of every record to upsert. Fresh records are in memory; a record of a pub that + * came from the push cache is re-mapped on demand. Done BEFORE the session opens, so a slow + * re-render can't let the session's idle timer run out between uploads. + */ + private async resolveUpserts( + upserts: ManifestEntry[], + payload: UnderlayPushPayload, + ): Promise { + const freshByKey = new Map(); + for (const record of payload.records) { + freshByKey.set(recordKey(record.type, record.id), record); + } + const out: UnderlayRecord[] = []; + for (const entry of upserts) { + let record = freshByKey.get(recordKey(entry.type, entry.id)); + if (!record && payload.resolveRecord) { + // biome-ignore lint/performance/noAwaitInLoops: lazy hydration is bounded by the upserts + record = (await payload.resolveRecord(entry)) ?? undefined; } - if (!rec) { + if (!record && payload.resolveRecordByHash) { + // biome-ignore lint/performance/noAwaitInLoops: lazy hydration is bounded by the upserts + record = (await payload.resolveRecordByHash(entry.hash)) ?? undefined; + } + if (!record) { throw new UnderlayPushError( - `The Underlay server requested a record we can no longer produce (hash ${hash}). ` + + `Underlay is missing record ${entry.type}/${entry.id}, and we can no longer produce it. ` + 'This usually means content cached from a previous push (e.g. an asset inside a release) ' + 'can no longer be regenerated — check the [underlay] warnings in the worker logs for skipped assets.', ); } - toSend.push(rec); - } - for (let i = 0; i < toSend.length; i += RECORD_BATCH_SIZE) { - const batch = toSend.slice(i, i + RECORD_BATCH_SIZE); - const ndjson = `${batch - .map((r) => JSON.stringify({ id: r.id, type: r.type, data: r.data })) - .join('\n')}\n`; - // biome-ignore lint/performance/noAwaitInLoops: batches must be sent sequentially + out.push(record); + } + return out; + } + + /** Upload NDJSON lines to a session, in batches within the server's line and byte limits. */ + private async sendLines( + sessionId: string, + kind: 'records' | 'deletes', + lines: string[], + limits: SessionLimits, + ): Promise { + const maxLines = limits.batch_lines ?? DEFAULT_BATCH_LINES; + const maxBytes = limits.batch_bytes ?? DEFAULT_BATCH_BYTES; + let batch: string[] = []; + let batchBytes = 0; + const flush = async () => { const response = await this.request( - `${this.collectionPath()}/versions/negotiate/${sessionId}/records`, + `${this.collectionPath()}/push/${sessionId}/${kind}`, { method: 'POST', headers: { 'Content-Type': 'application/x-ndjson' }, - rawBody: ndjson, + rawBody: `${batch.join('\n')}\n`, }, ); if (!response.ok) { - const detail = await response.text(); - throw new UnderlayPushError('Failed to send records', response.status, detail); + throw new UnderlayPushError( + `Underlay refused the ${kind} upload`, + response.status, + await this.body(response), + ); } + batch = []; + batchBytes = 0; + }; + for (const line of lines) { + const size = Buffer.byteLength(line) + 1; + if (batch.length > 0 && (batch.length >= maxLines || batchBytes + size > maxBytes)) { + // biome-ignore lint/performance/noAwaitInLoops: batches must be sent sequentially + await flush(); + } + batch.push(line); + batchBytes += size; + } + if (batch.length > 0) { + await flush(); + } + } + + /** Best effort: free the session slot (20 per user) after a push fails partway. */ + private async abandon(sessionId: string): Promise { + try { + await this.request(`${this.collectionPath()}/push/${sessionId}`, { method: 'DELETE' }); + } catch { + // The server expires idle sessions on its own. } } /** - * Run the full push. `payload` is the mapped records/schemas/files; `baseVersion` is the semver - * the caller diffed against (null on first push). Retries once on a 409 version conflict by - * re-fetching the latest version and re-negotiating. + * Run the full push: diff `payload` against the head, then open a session, upload the upserts and + * deletes, and commit. `metadataPatch` holds version metadata fields to keep in sync (e.g. + * `{readme}`); only the fields that differ from the head are sent, and any metadata set in + * Underlay itself is left alone. Starts over once against the new head on a version conflict, or + * after uploading files a commit said were missing. */ async push( payload: UnderlayPushPayload, - baseVersion: string | null, message: string, - metadata?: Record, + metadataPatch?: Record, ): Promise { // Incremental pushes supply a precomputed manifest spanning cache-reused + fresh records; // otherwise derive it from the in-memory records. const manifest = payload.manifest ?? buildManifest(payload.records); - const recordByHash = new Map(); - for (const record of payload.records) { - recordByHash.set(hashRecord(record).hash, record); - } + const filesByHash = new Map(payload.files.map((f) => [f.hash, f])); - const negotiateOnce = async (base: string | null): Promise => { - const negotiateResponse = await this.request( - `${this.collectionPath()}/versions/negotiate`, - { - method: 'POST', - headers: { 'Content-Type': 'application/json' }, - body: JSON.stringify({ - base_version: base, - message, - app_id: this.appId, - actor_id: this.actorId, - schemas: payload.schemas, - manifest, - // Every referenced hash — which on a streaming push is a superset of the - // bytes still held in `files` (those were uploaded and dropped during mapping). - files: payload.fileHashes ?? payload.files.map((f) => f.hash), - ...(metadata ? { metadata } : {}), - }), - }, + const attempt = async (): Promise => { + const head = await this.readHead(); + const plan = diffAgainstHead(manifest, payload.schemas, hashSchema, head); + const patch = metadataChanges(metadataPatch, head?.metadata ?? null); + if ( + head && + plan.upserts.length === 0 && + plan.deletes.length === 0 && + !plan.schemasChanged && + !patch + ) { + return { status: 'noop', reason: `Underlay ${head.semver} already matches` }; + } + console.info( + `[underlay] Diff against ${head?.semver ?? 'an empty collection'}: ${plan.upserts.length} upsert(s), ${plan.deletes.length} delete(s)${plan.schemasChanged ? ', schemas changed' : ''}${patch ? ', metadata changed' : ''}.`, ); + const upserts = await this.resolveUpserts(plan.upserts, payload); - if (negotiateResponse.status === 409) { - throw new UnderlayPushError('Version conflict', 409); - } - if (!negotiateResponse.ok) { - const detail = await negotiateResponse.text(); - throw new UnderlayPushError('Negotiate failed', negotiateResponse.status, detail); + const open = await this.request(`${this.collectionPath()}/push`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ + base: head?.semver ?? null, + message, + app_id: this.appId, + actor_id: this.actorId, + // The full type set: it replaces the base's. + schemas: payload.schemas, + ...(patch ? { metadata_patch: patch } : {}), + }), + }); + if (open.status === 409) { + throw new UnderlayPushError('Version conflict', 409, await this.body(open), true); } - - const session = await this.json<{ - session_id: string; - needed_records: string[]; - needed_files: string[]; - }>(negotiateResponse); - - // Upload needed files. - const filesByHash = new Map(payload.files.map((f) => [f.hash, f])); - for (const hash of session.needed_files) { - let file = filesByHash.get(hash); - if (!file && payload.resolveFileByHash) { - // biome-ignore lint/performance/noAwaitInLoops: lazy hydration is bounded by needed files - file = (await payload.resolveFileByHash(hash)) ?? undefined; - } - if (!file) { - throw new UnderlayPushError( - `The Underlay server requested a file we can no longer produce (sha256:${hash}). ` + - 'This usually means an asset from a previous push failed to download this time — ' + - 'check the [underlay] warnings in the worker logs for the skipped asset URLs.', - ); - } - // biome-ignore lint/performance/noAwaitInLoops: bounded by needed files - await this.uploadFile(file); + if (!open.ok) { + throw new UnderlayPushError( + 'Could not open a push session', + open.status, + await this.body(open), + ); } + const session = await this.json<{ session_id: string; limits?: SessionLimits }>(open); + const limits = session.limits ?? {}; - // Send needed records. - if (session.needed_records.length > 0) { - await this.sendRecords( + try { + await this.sendLines( session.session_id, - session.needed_records, - recordByHash, - payload.resolveRecordByHash, + 'records', + upserts.map((r) => + JSON.stringify({ + id: r.id, + type: r.type, + data: r.data, + ...(r.private ? { private: true } : {}), + }), + ), + limits, ); + await this.sendLines( + session.session_id, + 'deletes', + plan.deletes.map((d) => JSON.stringify({ type: d.type, id: d.id })), + limits, + ); + const result = await this.commit( + session.session_id, + filesByHash, + payload.resolveFileByHash, + ); + if (result.status === 'committed' && result.changes) { + console.info( + `[underlay] ${result.semver}: ${result.changes.added} added, ${result.changes.updated} updated, ${result.changes.removed} removed.`, + ); + } + return result; + } catch (err) { + await this.abandon(session.session_id); + throw err; } - - // Commit, retrying once for late-uploaded files. - const commit = await this.commit( - session.session_id, - filesByHash, - payload.resolveFileByHash, - ); - return commit; }; try { - return await negotiateOnce(baseVersion); + return await attempt(); } catch (err) { - // 409: someone pushed while we were diffing. `retriable`: the commit was rejected for a - // cause we've since fixed (missing files, now uploaded) on a session that can no longer - // be re-committed. Both are resolved by re-negotiating against the current head, once. - if (err instanceof UnderlayPushError && (err.statusCode === 409 || err.retriable)) { - const freshBase = await this.getBaseVersion(); - return negotiateOnce(freshBase); + if (err instanceof UnderlayPushError && err.retriable) { + console.info(`[underlay] ${err.message}; starting over against the current head.`); + return attempt(); } throw err; } @@ -661,7 +995,7 @@ export class UnderlayClient { ): Promise { const unresolved: string[] = []; for (const ref of filesNeeded) { - const hash = ref.replace(/^sha256:/, ''); + const hash = bareHash(ref); let file = filesByHash.get(hash); if (!file && resolveFileByHash) { // biome-ignore lint/performance/noAwaitInLoops: bounded retry @@ -671,15 +1005,68 @@ export class UnderlayClient { unresolved.push(hash); continue; } - await this.uploadFile(file); + await this.putFile(file); } return unresolved; } + /** + * Turn a refused commit into a result or an error. The body is the synchronous response, or a + * failed session's `error`; either way the session is now `failed` and can't be committed again. + */ + private async commitRefused( + status: number | undefined, + body: unknown, + filesByHash: Map, + resolveFileByHash?: (hash: string) => Promise, + ): Promise { + const detail = (body && typeof body === 'object' ? body : {}) as CommitErrorBody; + const code = status ?? detail.statusCode; + if (detail.error === 'No changes detected') { + return { status: 'noop', reason: 'Underlay reported no changes' }; + } + if (detail.error === 'Version conflict') { + throw new UnderlayPushError('Version conflict', 409, body, true); + } + if (detail.filesNeeded && detail.filesNeeded.length > 0) { + const unresolved = await this.uploadMissingFiles( + detail.filesNeeded, + filesByHash, + resolveFileByHash, + ); + // Starting over when we know we could not supply everything just trades a specific + // diagnosis for a generic failure. + if (unresolved.length > 0) { + throw new UnderlayPushError( + `Commit rejected: the Underlay server needs ${unresolved.length} file(s) we can no longer produce. ` + + 'Check the [underlay] warnings in the worker logs for skipped assets.', + 422, + body, + ); + } + throw new UnderlayPushError( + 'Commit rejected for missing files; they have been uploaded, retrying', + 422, + body, + true, + ); + } + if (code === 503) { + // Storage cleanup ran during the commit; the server says to push again. + throw new UnderlayPushError( + detail.error ?? 'Underlay storage was busy', + 503, + body, + true, + ); + } + throw new UnderlayPushError(detail.error ?? 'Commit failed', code, body); + } + /** * Poll a session whose commit was accepted asynchronously, until it reports a terminal status. * - * A transient failure to *read* the session is not a failed commit — the finalize is running + * A transient failure to *read* the session is not a failed commit — the commit is running * server-side regardless — so read errors are swallowed and retried until the deadline. */ private async awaitAsyncCommit( @@ -697,7 +1084,7 @@ export class UnderlayClient { await sleep(delay); delay = Math.min(COMMIT_POLL_MAX_MS, Math.round(delay * 1.5)); - // The server keeps finalizing regardless of whether we can read the session, so a failed + // The server keeps committing regardless of whether we can read the session, so a failed // poll must not fail the push. `request` throws once its own retries are exhausted, and // a malformed body throws from `json` — both are transient from here, so both keep // polling. If reads never recover, the deadline below produces the actionable timeout. @@ -708,14 +1095,14 @@ export class UnderlayClient { hash: string; recordCount: number; fileCount: number; + changes?: PushChanges; } | null; error?: unknown; }; try { - const response = await this.request( - `${this.collectionPath()}/versions/negotiate/${sessionId}`, - { method: 'GET' }, - ); + const response = await this.request(`${this.collectionPath()}/push/${sessionId}`, { + method: 'GET', + }); if (!response.ok) { continue; } @@ -736,42 +1123,17 @@ export class UnderlayClient { } if (session.status === 'failed') { - const detail = session.error as - | { error?: string; filesNeeded?: string[] } - | undefined; - // Same recovery the synchronous 422 path gets: upload what the server is missing and - // push again. The failed session can't be re-committed, so this re-negotiates. - if (detail?.filesNeeded && detail.filesNeeded.length > 0) { - const unresolved = await this.uploadMissingFiles( - detail.filesNeeded, - filesByHash, - resolveFileByHash, - ); - if (unresolved.length === 0) { - throw new UnderlayPushError( - 'Commit rejected for missing files; they have been uploaded, retrying', - 422, - detail, - true, - ); - } - throw new UnderlayPushError( - `Commit rejected: the Underlay server needs ${unresolved.length} file(s) we can no longer produce. ` + - 'Check the [underlay] warnings in the worker logs for skipped assets.', - 422, - detail, - ); - } - throw new UnderlayPushError( - detail?.error ?? 'Underlay reported the commit as failed', + return this.commitRefused( undefined, - session.error, + session.error ?? { error: 'Underlay reported the commit as failed' }, + filesByHash, + resolveFileByHash, ); } if (session.status === 'expired') { throw new UnderlayPushError( - 'The negotiate session expired before the commit finished', + 'The push session expired before the commit finished', undefined, session, ); @@ -781,7 +1143,7 @@ export class UnderlayClient { if (now - lastLoggedAt >= COMMIT_LOG_INTERVAL_MS) { lastLoggedAt = now; console.info( - `[underlay] Commit still finalizing (${Math.round((now - startedAt) / 1000)}s elapsed)…`, + `[underlay] Commit still running (${Math.round((now - startedAt) / 1000)}s elapsed)…`, ); } } @@ -797,66 +1159,36 @@ export class UnderlayClient { filesByHash: Map, resolveFileByHash?: (hash: string) => Promise, ): Promise { - // Ask for an async finalize. An Underlay that predates async commit ignores the query param - // and answers 201 with the version inline, which the 2xx path below handles unchanged — so - // this is safe against either side being deployed first. - const doCommit = () => - this.request( - `${this.collectionPath()}/versions/negotiate/${sessionId}/commit?async=true`, - { method: 'POST' }, - ); - - let response = await doCommit(); + const response = await this.request( + `${this.collectionPath()}/push/${sessionId}/commit?async=true`, + { method: 'POST' }, + ); - // 422 with missing files → upload them and retry once. - if (response.status === 422) { - const body = await this.json<{ - error?: string; - filesNeeded?: string[]; - extraFields?: string[]; - }>(response); - if (body.filesNeeded && body.filesNeeded.length > 0) { - const unresolved = await this.uploadMissingFiles( - body.filesNeeded, - filesByHash, - resolveFileByHash, - ); - // Retrying the commit when we know we could not supply everything just trades a - // specific diagnosis for a generic "Commit failed". Same message the async path gives. - if (unresolved.length > 0) { - throw new UnderlayPushError( - `Commit rejected: the Underlay server needs ${unresolved.length} file(s) we can no longer produce. ` + - 'Check the [underlay] warnings in the worker logs for skipped assets.', - 422, - body, - ); - } - response = await doCommit(); - } else { - throw new UnderlayPushError( - body.error ?? 'Commit rejected (422)', - 422, - body.extraFields ?? body, - ); - } + // 202: the server is committing in the background; the outcome lands on the session. + if (response.status === 202) { + return this.awaitAsyncCommit(sessionId, filesByHash, resolveFileByHash); } - if (!response.ok) { - const detail = await response.text(); - throw new UnderlayPushError('Commit failed', response.status, detail); + if (response.ok) { + const committed = await this.json<{ + semver: string; + hash: string; + recordCount: number; + fileCount: number; + changes?: PushChanges; + }>(response); + return { status: 'committed', ...committed }; } - // 202: the server is finalizing in the background; the outcome lands on the session. - if (response.status === 202) { + const body = await this.body(response); + // A commit retried after a lost response finds its session already committing: keep polling + // rather than reporting the first attempt's own progress as a failure. + if ( + response.status === 409 && + (body as CommitErrorBody | null)?.error === 'Session is committing' + ) { return this.awaitAsyncCommit(sessionId, filesByHash, resolveFileByHash); } - - const committed = await this.json<{ - semver: string; - hash: string; - recordCount: number; - fileCount: number; - }>(response); - return { status: 'committed', ...committed }; + return this.commitRefused(response.status, body, filesByHash, resolveFileByHash); } } diff --git a/server/underlay/hash.ts b/server/underlay/hash.ts index ac5ed1223..b2c65a2cb 100644 --- a/server/underlay/hash.ts +++ b/server/underlay/hash.ts @@ -1,17 +1,21 @@ import { createHash } from 'crypto'; /** - * Content-addressed hashing for Underlay push, ported verbatim from the Underlay server - * (`src/lib/core/hash.ts`). Any record or schema pushed to Underlay MUST hash identically to the - * way the server hashes it, or the push is rejected with "Unexpected record hash". Keep this in - * lockstep with the server implementation. + * Content-addressed hashing for Underlay push, ported from the Underlay reference implementation + * (`packages/protocol/src/{jcs,hash}.ts`). A delta push diffs our record hashes against the head's + * manifest, so any record that hashes differently from the server looks changed on every push. + * Keep this in lockstep with the server implementation. * - * @see https://underlay.org — public/llms.txt §"Record Hashing" + * @see https://www.underlay.org/llms.txt §"Record Hashing" */ /** * Recursively sort object keys alphabetically (by Unicode code point), preserving array order. * Ensures `{"b":1,"a":2}` and `{"a":2,"b":1}` produce the same serialization (and thus hash). + * + * Used only for PubPub's own local signatures (no-op guard, facet cascade). It is NOT the Underlay + * hash: `JSON.stringify` of the result still lists integer-like keys ("9", "10") first. Kept as is + * so existing stored signatures stay valid. */ export const canonicalize = (value: unknown): unknown => { if (value === null || typeof value !== 'object') { @@ -28,34 +32,66 @@ export const canonicalize = (value: unknown): unknown => { }; /** - * Hash a record's canonical JSON. The top-level keys MUST appear in the order id, type, data, and - * `data` MUST be canonicalized. The `private` flag is deliberately not part of the hash. + * RFC 8785 JSON Canonicalization Scheme: `JSON.stringify` for every primitive, object keys sorted by + * UTF-16 code units. Written out as a string because a JavaScript object enumerates integer-like + * keys first whatever order they were inserted in, so "sort into a new object, then stringify" + * cannot produce sorted output for them. + */ +export const jcs = (value: unknown): string => { + switch (typeof value) { + case 'string': + return JSON.stringify(value); + case 'boolean': + return value ? 'true' : 'false'; + case 'number': + if (!Number.isFinite(value)) { + throw new TypeError('JCS: non-finite number'); + } + // JSON.stringify renders -0 as "0", which is what JCS requires. + return JSON.stringify(value); + case 'object': { + if (value === null) { + return 'null'; + } + if (Array.isArray(value)) { + return `[${value.map(jcs).join(',')}]`; + } + const object = value as Record; + // Array.prototype.sort() with no comparator compares UTF-16 code units — JCS's key order. + const parts: string[] = []; + for (const key of Object.keys(object).sort()) { + if (object[key] === undefined) { + continue; + } + parts.push(`${JSON.stringify(key)}:${jcs(object[key])}`); + } + return `{${parts.join(',')}}`; + } + default: + throw new TypeError(`JCS: cannot serialize ${typeof value}`); + } +}; + +const sha256Hex = (input: string | Buffer | Uint8Array): string => + createHash('sha256').update(input).digest('hex'); + +/** + * Hash a record. The canonical form is a fixed `{"id","type","data"}` envelope with only `data` + * canonicalized by JCS. The `private` flag is deliberately not part of the hash. */ export const hashRecord = (record: { id: string; type: string; data: unknown; }): { hash: string; canonical: string } => { - const canonical = JSON.stringify({ - id: record.id, - type: record.type, - data: canonicalize(record.data), - }); - const hash = createHash('sha256').update(canonical).digest('hex'); - return { hash, canonical }; + const canonical = `{"id":${JSON.stringify(record.id)},"type":${JSON.stringify(record.type)},"data":${jcs(record.data)}}`; + return { hash: sha256Hex(canonical), canonical }; }; -/** - * Hash a JSON Schema document. Schemas are content-addressed by the SHA-256 of their canonicalized - * JSON, using the same rules as records. - */ -export const hashSchema = (schemaBody: unknown): string => - createHash('sha256') - .update(JSON.stringify(canonicalize(schemaBody))) - .digest('hex'); +/** Hash a JSON Schema document: SHA-256 of its JCS form. */ +export const hashSchema = (schemaBody: unknown): string => sha256Hex(jcs(schemaBody)); /** * Hash an arbitrary buffer of bytes (used for file references `{"$file":"sha256:"}`). */ -export const hashBytes = (bytes: Buffer | Uint8Array): string => - createHash('sha256').update(bytes).digest('hex'); +export const hashBytes = (bytes: Buffer | Uint8Array): string => sha256Hex(bytes); diff --git a/server/underlay/incremental.ts b/server/underlay/incremental.ts index e06635520..c4801c836 100644 --- a/server/underlay/incremental.ts +++ b/server/underlay/incremental.ts @@ -23,10 +23,10 @@ import { * re-rendering) only the pubs whose content changed since the last successful push; unchanged pubs * contribute their cached record + file hashes to the manifest without any rendering or asset fetch. * - * Correctness guarantee: even for a cache-hit pub, if the server unexpectedly asks for one of its - * records or files (e.g. it was GC'd, or a prior push never fully landed), the returned payload's - * `resolveRecordByHash` / `resolveFileByHash` lazily re-map that pub on demand and produce it. So a - * cache hit never means "we can't fulfil a needed_records request". + * Correctness guarantee: the client diffs this manifest against the head's, so a cache-hit pub's + * record can still turn out to be one the server lacks (a prior push never fully landed) or holds + * differently. The returned payload's `resolveRecord` / `resolveFileByHash` lazily re-map that pub on + * demand and produce it. So a cache hit never means "we can't upload a record the diff needs". * * See planning/pubpub-underlay-integration.md §3c. */ @@ -342,9 +342,9 @@ export const buildIncrementalPush = async ( // held for the rest of the push. // // Bytes are dropped only when they can be produced again. A scope file has no owning pub for - // `hydratePubRecords` to re-map, so its ONLY recovery path is re-fetching the source URL via + // `hydratePub` to re-map, so its ONLY recovery path is re-fetching the source URL via // `assetCache.byHash` — which the localizer populates for assets.pubpub.org URLs alone. Dropping - // an externally-hosted branding image would make a later `needed_files` request for it + // an externally-hosted branding image would make a later commit's missing-files refusal for it // unrecoverable, so those stay in memory. That set is small (branding images not on // assets.pubpub.org are the exception), so this costs little and removes the sharp edge. if (streaming) { @@ -415,8 +415,8 @@ export const buildIncrementalPush = async ( } } - // Everything still queued must land before the manifest is declared: negotiate announces these - // hashes as present, so an unfinished upload would make the server ask for a file mid-commit. + // Everything still queued must land before the push commits: the records reference these files, + // and a commit refuses any referenced file the collection doesn't hold yet. await pool?.drain(); if (flushCacheEntries && pendingCheckpoint.length > 0) { await flushCacheEntries(pendingCheckpoint.splice(0, pendingCheckpoint.length)); @@ -436,7 +436,7 @@ export const buildIncrementalPush = async ( } } - // ── Lazy re-hydration: produce a needed record/file for a cache-hit pub on demand. ────────── + // ── Lazy re-hydration: produce a record/file for a cache-hit pub on demand. ───────────────── // // Only RECORDS are memoized. Re-mapping a pub also produces its file bytes, and holding those // would defeat the streaming this module exists to do: a resumed push can legitimately need to @@ -444,34 +444,67 @@ export const buildIncrementalPush = async ( // would reaccumulate the entire collection in memory. // // Why a resumed push needs this at all: a checkpointed pub has had its files uploaded, but its - // records are only ever sent AFTER negotiate. A push that dies mid-mapping therefore leaves pubs - // marked cached whose records the server has never seen, and the next push gets them all back in - // `needed_records`. Records are metadata-sized (the HTML and exports live in files), so memoizing - // them is bounded; bytes are not. - const hydratedRecordsByPubId = new Map>(); - const hydratePubRecords = async (pubId: string): Promise> => { - const cached = hydratedRecordsByPubId.get(pubId); + // records are only ever sent at the end, in the push session. A push that dies mid-mapping + // therefore leaves pubs marked cached whose records the server has never seen, and the next + // push's diff lists them all as upserts. Records are metadata-sized (the HTML and exports live in + // files), so memoizing them is bounded; bytes are not. + type HydratedPub = { byHash: Map; byKey: Map }; + const recordKey = (type: string, id: string) => `${type}\u0000${id}`; + const hydratedByPubId = new Map(); + const hydratePub = async (pubId: string): Promise => { + const cached = hydratedByPubId.get(pubId); if (cached) { return cached; } - const recordMap = new Map(); + const hydrated: HydratedPub = { byHash: new Map(), byKey: new Map() }; const pub = pubById.get(pubId); if (pub) { - const { records } = await mapPub(pub); + const { records, files } = await mapPub(pub); for (const record of records) { - recordMap.set(hashRecord(record).hash, record); + hydrated.byHash.set(hashRecord(record).hash, record); + hydrated.byKey.set(recordKey(record.type, record.id), record); + } + // Self-heal a cache entry whose hashes no longer match what the pub maps to (e.g. hashes + // stored before the hash function changed). Without this the stale hash would differ from + // the head's on every push and the record would be re-sent each time. `cacheUpserts` is + // the array returned below, so the correction is persisted with the rest after the push. + const entry = entryByPubId.get(pubId); + const fresh = recordHashesFrom(records); + if ( + entry && + !cacheUpserts.some((upsert) => upsert.pubId === pubId) && + JSON.stringify(canonicalize(fresh)) !== + JSON.stringify(canonicalize(entry.recordHashes)) + ) { + cacheUpserts.push({ + ...entry, + recordHashes: fresh, + fileHashes: files.map((f) => f.hash), + }); } } - hydratedRecordsByPubId.set(pubId, recordMap); - return recordMap; + hydratedByPubId.set(pubId, hydrated); + return hydrated; }; + const resolveRecord = async (entry: ManifestEntry): Promise => { + const pubId = pubIdByHash.get(entry.hash); + if (!pubId) { + return null; + } + const hydrated = await hydratePub(pubId); + return ( + hydrated.byKey.get(recordKey(entry.type, entry.id)) ?? + hydrated.byHash.get(entry.hash) ?? + null + ); + }; const resolveRecordByHash = async (hash: string): Promise => { const pubId = pubIdByHash.get(hash); if (!pubId) { return null; } - return (await hydratePubRecords(pubId)).get(hash) ?? null; + return (await hydratePub(pubId)).byHash.get(hash) ?? null; }; const resolveFileByHash = async (hash: string): Promise => { // A cache-resolved file (e.g. a scope image) has no bytes in memory — fetch them from the @@ -516,6 +549,7 @@ export const buildIncrementalPush = async ( fileHashes: [...allFileHashes].sort(), schemas, manifest, + resolveRecord, resolveRecordByHash, resolveFileByHash, }; diff --git a/server/underlay/mapping.ts b/server/underlay/mapping.ts index ed78b79e1..57ce878c0 100644 --- a/server/underlay/mapping.ts +++ b/server/underlay/mapping.ts @@ -67,8 +67,8 @@ export type UnderlayPushPayload = { files: UnderlayFile[]; /** * Every file hash this version references, whether or not its bytes are still in `files`. The - * streaming path uploads bytes during mapping and keeps only hashes, so this — not `files` — is - * what the negotiate call must declare. Falls back to the hashes of `files` when omitted. + * streaming path uploads bytes during mapping and keeps only hashes. Feeds the no-op signature; + * Underlay itself learns referenced files from the records' `$file` references. */ fileHashes?: string[]; /** @@ -78,10 +78,13 @@ export type UnderlayPushPayload = { */ manifest?: ManifestEntry[]; /** - * Lazy fallback for the incremental path: produce a record whose hash the server asked for but - * which isn't in `records` (e.g. a cache-hit pub the server unexpectedly needs re-sent). Returns - * null if the hash can't be resolved. + * Lazy fallback for the incremental path: produce the record behind a manifest entry that isn't + * in `records` (a cache-hit pub whose record the head lacks or holds differently). Looks the + * record up by (type, id), so it still works when the cached hash was computed differently from + * the freshly-mapped one. Returns null if the entry can't be resolved. */ + resolveRecord?: (entry: ManifestEntry) => Promise; + /** As `resolveRecord`, by hash alone. */ resolveRecordByHash?: (hash: string) => Promise; /** Lazy fallback: produce file bytes the server asked for but which aren't in `files`. */ resolveFileByHash?: (hash: string) => Promise; diff --git a/workers/tasks/pushToUnderlay.ts b/workers/tasks/pushToUnderlay.ts index a164a08b9..600e3fd27 100644 --- a/workers/tasks/pushToUnderlay.ts +++ b/workers/tasks/pushToUnderlay.ts @@ -524,10 +524,13 @@ export const pushToUnderlayTask = async (input: PushToUnderlayInput) => { // Version metadata pushed to Underlay (produces a patch version when it changes). Folded into // the no-op signature below so a readme-only edit isn't skipped as "no changes". const pushMetadata = integration.readme ? { readme: integration.readme } : undefined; + // Sent as a patch, so metadata set in Underlay itself (description, license, …) survives. A + // cleared readme is sent as null; the client only sends fields that differ from the head. + const metadataPatch = { readme: integration.readme || null }; // The collection must exist before any file can be uploaded into it, and the streaming // uploader below runs during mapping — so this moves ahead of the build (it used to sit just - // before negotiate). Creating it early is harmless: it was going to be created either way, + // before the push). Creating it early is harmless: it was going to be created either way, // and a push that fails afterwards just leaves an empty collection with no versions. await client.ensureCollection(); @@ -589,15 +592,13 @@ export const pushToUnderlayTask = async (input: PushToUnderlayInput) => { } console.info( - `[underlay] Mapped ${incremental.stats.totalPubs} pub(s): ${incremental.stats.cacheHits} cache hit(s), ${incremental.stats.cacheMisses} re-mapped, ${uploadedFileCount} file(s) streamed. Negotiating…`, + `[underlay] Mapped ${incremental.stats.totalPubs} pub(s): ${incremental.stats.cacheHits} cache hit(s), ${incremental.stats.cacheMisses} re-mapped, ${uploadedFileCount} file(s) streamed. Diffing against Underlay…`, ); - const baseVersion = await client.getBaseVersion(); const result = await client.push( incremental.payload, - baseVersion, `PubPub sync for ${community.subdomain}`, - pushMetadata, + metadataPatch, ); const warnings: AssetWarning[] = [...assetWarnings.values()]; @@ -633,36 +634,35 @@ export const pushToUnderlayTask = async (input: PushToUnderlayInput) => { semver: result.semver, recordCount: result.recordCount, fileCount: result.fileCount, - message: `Pushed version ${result.semver}`, + message: result.changes + ? `Pushed version ${result.semver}: ${result.changes.added} added, ${result.changes.updated} updated, ${result.changes.removed} removed` + : `Pushed version ${result.semver}`, warnings, }); } - // Persist the cache only after a successful commit. - await applyPushCache( - integration.id, - incremental.cacheUpserts, - incremental.presentPubIds, - ); - // Record any newly-fetched immutable asset URLs so future pushes skip the download. - if (assetCache.learned.size > 0) { - await saveCachedAssetHashes( - [...assetCache.learned].map(([url, asset]) => ({ - url, - hash: asset.hash, - mimeType: asset.mimeType, - fileName: asset.fileName, - })), - ); - } } else { await recordPushResult(communityId, { status: 'noop', manifestHash: incremental.signature, }); if (logId) { - await finishPushLog(logId, { status: 'noop', warnings }); + await finishPushLog(logId, { status: 'noop', message: result.reason, warnings }); } } + // Persist the cache only once Underlay holds exactly this content: after a commit, or when + // the head already matched (a noop) — which is also when stale cache entries self-heal. + await applyPushCache(integration.id, incremental.cacheUpserts, incremental.presentPubIds); + // Record any newly-fetched immutable asset URLs so future pushes skip the download. + if (assetCache.learned.size > 0) { + await saveCachedAssetHashes( + [...assetCache.learned].map(([url, asset]) => ({ + url, + hash: asset.hash, + mimeType: asset.mimeType, + fileName: asset.fileName, + })), + ); + } return { ...result, stats: incremental.stats, warnings }; } catch (error) { const detail = formatUnderlayError(error);