diff --git a/apps/web/README.md b/apps/web/README.md index 6426e6a07..7d2574911 100644 --- a/apps/web/README.md +++ b/apps/web/README.md @@ -4,7 +4,7 @@ Promoters find local DJs who fit their shows. DJs (called _projects_: a solo DJ, All data lives on the AT Protocol. Public profiles are ordinary atproto records; everything else (requests, staff roles, the audit log, booking threads, personal planners) lives in [atproto spaces](https://atproto.com/blog/atproto-spaces-alpha), the protocol's permissioned-data extension, where each space's read policy decides who can read it. Published projects, the DJ stats, sets, press kits, shows and follows their managers and fans publish, and the few Bluesky and userinput.app records the site reads are indexed by [HappyView](https://happyview.dev), an AppView whose Lua scripts answer directory search and the show matcher and serve those records to the site. The app keeps its own SQLite index for space data, which it can rebuild from the network, alongside what exists only there (sessions, notifications, press kit view receipts and the geocode cache). Every public record is read through HappyView, never straight from a PDS or Bluesky's AppView (see [Directory](#directory) for the few exceptions); writes go to the PDS. -> Atproto spaces is an alpha (released 2026). This app is built against `@atproto/*@0.0.0-spaces-alpha-20261029171913` and the [0016 spaces proposal](https://github.com/bluesky-social/proposals/tree/main/0016-permissioned-data). Expect breaking changes upstream; all versions are pinned. +> Atproto spaces is an alpha (released 2026). This app is built against `@atproto/*@0.0.0-spaces-alpha-20261105160032` and the [0016 spaces proposal](https://github.com/bluesky-social/proposals/tree/main/0016-permissioned-data). Expect breaking changes upstream; all versions are pinned. ## Features @@ -207,7 +207,7 @@ A mix whose link is a plyr.fm track (`https://plyr.fm/track/`) also gets a ` ### Sync -The app is a space syncer. For each watched space it asks the space host (the authority's PDS) for a space credential, sending a delegation token from the root's PDS and a fresh P-256 key the credential is bound to, and signs every request with that key (RFC 9421). It enumerates writers with `listRepos`, pulls each repo's oplog with `listRepoOps`, and checks the result against the repo's LtHash commit, falling back to a full download (`getRepo`, a STAR archive) when needed. Commits and records aren't signed: the repo host MACs each response with a secret shared for that one request, derived from the author's `#atproto_kem` key. When a MAC fails, the app refetches the author's DID document once, in case the key rotated. Accounts whose DID document has no KEM key can't be synced, and the system page says so. A deleted space is dropped with everything derived from it, however often the authority says so, and a space recreated under the same URI syncs from scratch. A repo that fails doesn't hold back the others: the pass carries on through the space's later pages, and the repo is retried on its own, after a minute at first and at most an hour once it keeps failing. A record whose projection into the app's tables fails, such as an inquiry read while the directory is down, waits on the same schedule and shows on the system page until it goes through ([ADR 0009](../../docs/adr/0009-synced-records-queue-for-projection.md)). It registers for `notifyWrite` (verified with inter-service auth) and also polls. Records are validated against the lexicons, and authority rules are enforced: only root-authored decisions, roles, and details count, a press kit's grants count only from their space's authority (in a project's request space, only for a press kit of that project) and drafts only from the project's current managers, a booking thread's project comes from the root's opening record (or the root's audit copy of its creation) rather than the inquiry, booking records from non-participants are ignored, and a booking status record counts only when the thread's rules allow its author that move ([ADR 0008](../../docs/adr/0008-booking-status-follows-the-rules-not-record-dates.md)). +The app is a space syncer. For each watched space it asks the space host (the authority's PDS) for a space credential, sending a delegation token from the root's PDS and a fresh P-256 key the credential is bound to, and signs every request with that key (RFC 9421). It enumerates writers with `listRepos`, pulls each repo's oplog with `listRepoOps`, and checks the result against the repo's LtHash commit, falling back to a full download (`getRepo`, a STAR archive) when needed. Commits and records aren't signed: the repo host MACs each response with a secret shared for that one request, derived from the author's `#atproto_kem` key. When a MAC fails, the app refetches the author's DID document once, in case the key rotated. Accounts whose DID document has no KEM key can't be synced, and the system page says so. A deleted space is dropped with everything derived from it, however often the authority says so, and a space recreated under the same URI syncs from scratch. A repo that fails doesn't hold back the others: the pass carries on through the space's later pages, and the repo is retried on its own, after a minute at first and at most an hour once it keeps failing. A writer whose account is taken down, suspended, or deactivated is skipped the same way without failing the pass, and the system page shows why. A space whose authority is taken down or deactivated is only suspended, so its local copy stays. A record whose projection into the app's tables fails, such as an inquiry read while the directory is down, waits on the same schedule and shows on the system page until it goes through ([ADR 0009](../../docs/adr/0009-synced-records-queue-for-projection.md)). It registers for `notifyWrite` (verified with inter-service auth) and also polls. Records are validated against the lexicons, and authority rules are enforced: only root-authored decisions, roles, and details count, a press kit's grants count only from their space's authority (in a project's request space, only for a press kit of that project) and drafts only from the project's current managers, a booking thread's project comes from the root's opening record (or the root's audit copy of its creation) rather than the inquiry, booking records from non-participants are ignored, and a booking status record counts only when the thread's rules allow its author that move ([ADR 0008](../../docs/adr/0008-booking-status-follows-the-rules-not-record-dates.md)). ### Directory diff --git a/apps/web/package.json b/apps/web/package.json index 3aea38c65..1b763386b 100644 --- a/apps/web/package.json +++ b/apps/web/package.json @@ -45,21 +45,21 @@ "test:storybook": "vitest run --project storybook" }, "dependencies": { - "@atproto-labs/did-resolver": "0.0.0-spaces-alpha-20261029171913", - "@atproto-labs/fetch-node": "0.0.0-spaces-alpha-20261029171913", - "@atproto-labs/handle-resolver-node": "0.0.0-spaces-alpha-20261029171913", - "@atproto/api": "0.0.0-spaces-alpha-20261029171913", - "@atproto/common-web": "0.0.0-spaces-alpha-20261029171913", - "@atproto/crypto": "0.0.0-spaces-alpha-20261029171913", - "@atproto/identity": "0.0.0-spaces-alpha-20261029171913", - "@atproto/lex-cbor": "0.0.0-spaces-alpha-20261029171913", - "@atproto/lex-data": "0.0.0-spaces-alpha-20261029171913", - "@atproto/lex-json": "0.0.0-spaces-alpha-20261029171913", - "@atproto/lexicon": "0.0.0-spaces-alpha-20261029171913", - "@atproto/oauth-client-node": "0.0.0-spaces-alpha-20261029171913", - "@atproto/oauth-scopes": "0.0.0-spaces-alpha-20261029171913", - "@atproto/space": "0.0.0-spaces-alpha-20261029171913", - "@atproto/syntax": "0.0.0-spaces-alpha-20261029171913", + "@atproto-labs/did-resolver": "0.0.0-spaces-alpha-20261105160032", + "@atproto-labs/fetch-node": "0.0.0-spaces-alpha-20261105160032", + "@atproto-labs/handle-resolver-node": "0.0.0-spaces-alpha-20261105160032", + "@atproto/api": "0.0.0-spaces-alpha-20261105160032", + "@atproto/common-web": "0.0.0-spaces-alpha-20261105160032", + "@atproto/crypto": "0.0.0-spaces-alpha-20261105160032", + "@atproto/identity": "0.0.0-spaces-alpha-20261105160032", + "@atproto/lex-cbor": "0.0.0-spaces-alpha-20261105160032", + "@atproto/lex-data": "0.0.0-spaces-alpha-20261105160032", + "@atproto/lex-json": "0.0.0-spaces-alpha-20261105160032", + "@atproto/lexicon": "0.0.0-spaces-alpha-20261105160032", + "@atproto/oauth-client-node": "0.0.0-spaces-alpha-20261105160032", + "@atproto/oauth-scopes": "0.0.0-spaces-alpha-20261105160032", + "@atproto/space": "0.0.0-spaces-alpha-20261105160032", + "@atproto/syntax": "0.0.0-spaces-alpha-20261105160032", "@aws-sdk/client-s3": "^3.1159.0", "@fontsource-variable/instrument-sans": "^5.3.0", "@fontsource-variable/jetbrains-mono": "^5.3.0", @@ -80,8 +80,8 @@ "zod": "^4.6.5" }, "devDependencies": { - "@atproto/dev-env": "0.0.0-spaces-alpha-20261029171913", - "@atproto/sync": "0.0.0-spaces-alpha-20261029171913", + "@atproto/dev-env": "0.0.0-spaces-alpha-20261105160032", + "@atproto/sync": "0.0.0-spaces-alpha-20261105160032", "@axe-core/playwright": "^4.13.0", "@lhci/cli": "^0.15.1", "@playwright/test": "^1.63.0", diff --git a/apps/web/src/lib/server/atproto/credentials.ts b/apps/web/src/lib/server/atproto/credentials.ts index 2ff3eb5e4..dc99ec667 100644 --- a/apps/web/src/lib/server/atproto/credentials.ts +++ b/apps/web/src/lib/server/atproto/credentials.ts @@ -11,7 +11,6 @@ import { parseMacHeader, parseSpaceToken, RECORD_MAC_TYPE, - RepoVerificationError, SpaceMacError, verifyRecord, WANT_MAC_HEADER @@ -49,9 +48,12 @@ export function isCredentialRejected(cause: unknown): boolean { return cause instanceof XrpcResponseError && (cause.status === 401 || CREDENTIAL_REJECTED.has(cause.error)); } -/** Whether a response's MAC didn't verify. `verifyRepoStar` reports a commit that fails its MAC as a RepoVerificationError. */ +/** + * Whether a response's MAC didn't verify. Not a RepoVerificationError: that also covers an archive that doesn't + * parse, which a fresh KEM key can't fix. + */ export function isMacFailure(cause: unknown): boolean { - return cause instanceof SpaceMacError || cause instanceof RepoVerificationError; + return cause instanceof SpaceMacError; } type RequestInput = { diff --git a/apps/web/src/lib/server/sync/engine.test.ts b/apps/web/src/lib/server/sync/engine.test.ts index cdcfd658c..2c792b8e8 100644 --- a/apps/web/src/lib/server/sync/engine.test.ts +++ b/apps/web/src/lib/server/sync/engine.test.ts @@ -325,6 +325,15 @@ describe('SpaceSyncer against a repo host', () => { ]); }); + it('reports it with the space, though it never synced', async () => { + const s = await failOnce(); + + expect((await s.status())[0].repos).toEqual([ + { did: DJ, rev: expect.any(String), lastSyncedAt: expect.any(String), lastError: null }, + { did: BAD, rev: null, lastSyncedAt: null, lastError: 'UpstreamFailure' } + ]); + }); + it('retries it after its backoff, though listRepos no longer lists it', async () => { const s = await failOnce(); host.repoErrors.clear(); @@ -621,6 +630,111 @@ describe('SpaceSyncer against a repo host', () => { ); }); + /** Has the host's getRepo archive rewritten by `edit`, keeping its MAC header. */ + const editArchive = (edit: (archive: Buffer) => Buffer) => + tamper(['getRepo'], async (response) => { + const archive = edit(Buffer.from(await response.arrayBuffer())); + + return new Response(new Uint8Array(archive), { headers: response.headers }); + }); + + it('reads a repo archive framed with control bytes, as the alpha host has sent since Nov 5', async () => { + const s = await syncer(); + await host.put(DJ, NOTE, 'a', note('one')); + const served: Buffer[] = []; + editArchive((archive) => (served.push(archive), archive)); + + await s.sync(SPACE); + expect(await records()).toEqual([['a', 'one']]); + // Each record follows a 0x01 byte, and a 0x00 byte ends the archive. + const path = Buffer.from(`${NOTE}/a`); + expect(served[0].includes(Buffer.concat([Buffer.from([1, path.length]), path]))).toBe(true); + expect(served[0].at(-1)).toBe(0); + }); + + describe('a repo archive that does not read', () => { + const PATH = `${NOTE}/a`; + + it.each([ + ['is cut off before its end byte', (archive: Buffer) => archive.subarray(0, -1), 'unexpected end of archive'], + [ + 'has a record path that is not one', + (archive: Buffer) => archive.fill('&', archive.indexOf(PATH), archive.indexOf(PATH) + 1), + `invalid NSID in record path: &${PATH.slice(1)}` + ] + ])("fails the repo when it %s, with the reader's error, and keeps the KEM key", async (_, edit, message) => { + const s = await syncer(); + await host.put(DJ, NOTE, 'a', note('one')); + editArchive(edit); + + await expect(s.sync(SPACE)).rejects.toThrow(message); + expect(host.identity.refreshed).toEqual([]); + expect(calls('getRepo')).toHaveLength(1); + expect(await db.selectFrom('syncRepoRetry').select('lastError').execute()).toEqual([{ lastError: message }]); + }); + }); + + it('refreshes the KEM key once when a full download fails its MAC', async () => { + const s = await syncer(); + await host.put(DJ, NOTE, 'a', note('one')); + host.forgeMacs = true; + + await expect(s.sync(SPACE)).rejects.toThrow('Commit does not match its MAC'); + expect(host.identity.refreshed).toEqual([DJ]); + expect(calls('getRepo')).toHaveLength(2); + expect(await records()).toEqual([]); + }); + + it.each(['RepoTakendown', 'RepoSuspended', 'RepoDeactivated'])( + 'skips a writer whose repo host answers %s, without failing the pass or downloading the repo', + async (error) => { + const WRITER = 'did:plc:writer'; + const s = await syncer(); + await host.put(DJ, NOTE, 'a', note('one')); + await host.put(WRITER, NOTE, 'b', note('two')); + await s.sync(SPACE); + const downloads = calls('getRepo').length; + + host.repoErrors.set(WRITER, error); + await host.put(WRITER, NOTE, 'c', note('three')); + await host.put(DJ, NOTE, 'd', note('four')); + await s.sync(SPACE); + + expect(calls('getRepo')).toHaveLength(downloads); + expect(await records()).toEqual([ + ['a', 'one'], + ['b', 'two'], + ['d', 'four'] + ]); + expect(await db.selectFrom('syncRepoRetry').select(['repoDid', 'lastError']).execute()).toEqual([ + { repoDid: WRITER, lastError: error } + ]); + const status = (await s.status())[0]; + expect(status.lastError).toBe(error); + expect(status.repos.find((repo) => repo.did === WRITER)?.lastError).toBe(error); + } + ); + + it.each(['RepoTakendown', 'RepoDeactivated'])( + 'keeps a space whose authority answers %s, since that only suspends it', + async (error) => { + await host.put(DJ, NOTE, 'a', note('one')); + await (await syncer()).sync(SPACE); + + host.overrides.set('com.atproto.space.listRepos', () => Response.json({ error }, { status: 400 })); + await expect((await syncer()).sync(SPACE)).rejects.toThrow(error); + host.overrides.clear(); + host.credentialError = error; + // A new syncer has no credential cached, so it asks the authority for one. + await expect((await syncer()).sync(SPACE)).rejects.toThrow(error); + + expect(dropped).toEqual([]); + expect(await records()).toEqual([['a', 'one']]); + expect(await db.selectFrom('syncRepoRetry').selectAll().execute()).toEqual([]); + expect(await db.selectFrom('spaceWatch').select('lastError').execute()).toEqual([{ lastError: error }]); + } + ); + it('keeps a failure that is not an Error as its text on the repo', async () => { const s = await syncer(); await host.put(DJ, NOTE, 'a', note('one')); diff --git a/apps/web/src/lib/server/sync/engine.ts b/apps/web/src/lib/server/sync/engine.ts index 1c90519ab..b1d0d409c 100644 --- a/apps/web/src/lib/server/sync/engine.ts +++ b/apps/web/src/lib/server/sync/engine.ts @@ -6,10 +6,11 @@ import { COMMIT_MAC_TYPE, type Commit, type RecordPathParts, + readRepoStar, RepoCommit, SpaceMacError, - verifyCommit, - verifyRepoStarFull + type VerifiedRecord, + verifyCommit } from '@atproto/space'; import PQueue from 'p-queue'; import { z } from 'zod'; @@ -72,7 +73,8 @@ export type SyncStatus = { lastSyncedAt: string | null; lastError: string | null; registrationExpiresAt: string | null; - repos: { did: string; rev: string; lastSyncedAt: string; lastError: string | null }[]; + /** A repo that has failed since before it ever synced has no rev or sync time. */ + repos: { did: string; rev: string | null; lastSyncedAt: string | null; lastError: string | null }[]; }; /** @@ -114,6 +116,17 @@ function authoritySaysGone(error: XrpcResponseError): boolean { const failsWholeSpace = (cause: unknown) => cause instanceof XrpcResponseError && cause.nsid === 'com.atproto.space.getSpaceCredential'; +const REPO_UNAVAILABLE = new Set(['RepoTakendown', 'RepoSuspended', 'RepoDeactivated']); + +/** + * Whether a writer's repo host won't serve the repo because the account is taken down, suspended, or + * deactivated. A full download fails the same way, and the pass carries on without it: the repo waits out its + * backoff, so it's picked up again once the account is back. The same errors from getSpaceCredential mean the + * authority's account is, which suspends the whole space. + */ +const isRepoUnavailable = (cause: unknown) => + cause instanceof XrpcResponseError && REPO_UNAVAILABLE.has(cause.error) && !failsWholeSpace(cause); + /** How long a repo waits after its `attempts`th failure in a row: 1 minute, doubling up to an hour. */ const retryDelayMs = (attempts: number) => Math.min(60_000 * 2 ** (attempts - 1), 3_600_000); @@ -253,7 +266,7 @@ export class SpaceSyncer { for (const result of results) { if (result.status === 'fulfilled') changed += Number(result.value); else if (failsWholeSpace(result.reason)) throw result.reason; - else failures.push(result.reason); + else if (!isRepoUnavailable(result.reason)) failures.push(result.reason); } }; @@ -409,7 +422,7 @@ export class SpaceSyncer { return; } catch (error) { // A full download would fail the same way: readVerified retries a MAC failure with a fresh KEM key. - if (isCredentialRejected(error) || isMacFailure(error)) throw error; + if (isCredentialRejected(error) || isMacFailure(error) || isRepoUnavailable(error)) throw error; logger.info('incremental sync failed; recovering full repo', { err: error, space, repoDid }); } } @@ -423,6 +436,9 @@ export class SpaceSyncer { if (!failsWholeSpace(error)) await this.recordFailure(space, repoDid, message); + if (isRepoUnavailable(error)) + logger.info('skipping a repo its host will not serve', { err: error, space, repoDid }); + if (local) { await db .updateTable('syncedRepo') @@ -566,8 +582,21 @@ export class SpaceSyncer { if (!response.body) throw new Error('Empty repo response'); - // A STAR archive: the commit (checked against the MAC), then every record, which must fold into its data. - const verified = await verifyRepoStarFull(response.body, { space, repo: repoDid, ...macParams() }); + // A STAR archive: the commit, then every record, which must fold into its data. verifyRepoStar would check the + // commit too, but reports a MAC that doesn't verify as the same RepoVerificationError as an archive that + // doesn't parse, and only the first is worth refreshing the KEM key for. + const archive = await readRepoStar(response.body); + const records: VerifiedRecord[] = []; + + try { + if (!verifyCommit(archive.commit, { space, repo: repoDid, ...macParams() })) { + throw new SpaceMacError('Commit does not match its MAC'); + } + + for await (const record of archive.records) records.push(record); + } finally { + await archive[Symbol.asyncDispose](); + } const existing = await this.deps.db .selectFrom('spaceRecord') @@ -580,7 +609,7 @@ export class SpaceSyncer { const changes: SyncedChange[] = []; const known = new Map(existing.map((row) => [`${row.collection}/${row.rkey}`, row.cid])); - for (const record of verified.records) { + for (const record of records) { const path = `${record.collection}/${record.rkey}`; present.add(path); const cid = record.cid.toString(); @@ -611,7 +640,7 @@ export class SpaceSyncer { } } - await this.apply(space, repoDid, pds, verified.commit, RepoCommit.fromRecords(verified.records), changes); + await this.apply(space, repoDid, pds, archive.commit, RepoCommit.fromRecords(records), changes); } private async apply( @@ -811,11 +840,23 @@ export class SpaceSyncer { async status(): Promise { const watches = await this.deps.db.selectFrom('spaceWatch').selectAll().orderBy('createdAt').execute(); - const repos = await this.deps.db + const synced = await this.deps.db .selectFrom('syncedRepo') .select(['spaceUri', 'repoDid', 'rev', 'lastSyncedAt', 'lastError']) .execute(); + // A repo that never synced has no syncedRepo row, only its retry, which is all that shows why. + const unsynced = await this.deps.db + .selectFrom('syncRepoRetry as retry') + .leftJoin('syncedRepo as synced', (join) => + join.onRef('synced.spaceUri', '=', 'retry.spaceUri').onRef('synced.repoDid', '=', 'retry.repoDid') + ) + .where('synced.repoDid', 'is', null) + .select(['retry.spaceUri', 'retry.repoDid', 'retry.lastError']) + .execute(); + + const repos = [...synced, ...unsynced.map((repo) => ({ ...repo, rev: null, lastSyncedAt: null }))]; + return watches.map((watch) => ({ space: watch.spaceUri, type: watch.spaceType, @@ -825,6 +866,7 @@ export class SpaceSyncer { lastError: watch.lastError, registrationExpiresAt: watch.registrationExpiresAt, repos: repos + .values() .filter((repo) => repo.spaceUri === watch.spaceUri) .map((repo) => ({ did: repo.repoDid, @@ -832,6 +874,7 @@ export class SpaceSyncer { lastSyncedAt: repo.lastSyncedAt, lastError: repo.lastError })) + .toArray() })); } } diff --git a/apps/web/src/routes/admin/system/+page.svelte b/apps/web/src/routes/admin/system/+page.svelte index 8ad1cbfb9..2414fd9d6 100644 --- a/apps/web/src/routes/admin/system/+page.svelte +++ b/apps/web/src/routes/admin/system/+page.svelte @@ -100,7 +100,7 @@ {#if space.lastError}

{space.lastError}

{/if}