chadtmiller.com/pds.js
An atproto PDS in JavaScript, for Cloudflare Workers and Node
0
0
getRepo lets other requests run while it reads the repo
.changeset/get-repo-yields.mdnew+9−0
| @@ -0,0 +1,9 @@ | ||
--- | ||
'@pdsjs/core': patch | ||
--- | ||
| ||
`com.atproto.sync.getRepo` lets other requests run while it reads the repo. It | ||
waits on a timer after every 500 blocks. On Cloudflare, one Durable Object | ||
answers every request, and a whole-repo read held all of them. An indexer | ||
fetching a large repo after a commit made the account's own requests wait 3 to | ||
7 seconds. | ||
packages/core/src/handlers/xrpc-sync.js+10−0
| @@ -16,6 +16,9 @@ import { buildCarFile, cborDecode, cidToString } from '../repo.js'; | ||
* @property {() => Promise<boolean>} isDeactivated | ||
*/ | ||
| ||
/** Blocks buildFullRepoCar reads between two waits on a timer. */ | ||
const BLOCKS_PER_WAIT = 500; | ||
| ||
/** @returns {Response} */ | ||
function repoNotFound() { | ||
return Response.json( | ||
| @@ -132,6 +135,10 @@ export function createSyncHandlers(ctx) { | ||
* | ||
* Throws if any of those blocks is absent. A caller must not answer with a | ||
* part of a repo, because a relay accepts it as the whole one. | ||
* | ||
* The walk waits on a timer after every BLOCKS_PER_WAIT blocks. A Durable | ||
* Object delivers no other request until its code waits on something other | ||
* than a promise already settled, and a whole repo is thousands of blocks. | ||
* @param {string} commitCid | ||
* @returns {Promise<Uint8Array>} | ||
*/ | ||
| @@ -152,6 +159,9 @@ export function createSyncHandlers(ctx) { | ||
if (!taken.has(cidStr)) { | ||
taken.add(cidStr); | ||
blocks.push({ cid: cidStr, data }); | ||
if (blocks.length % BLOCKS_PER_WAIT === 0) { | ||
await new Promise((resolve) => setTimeout(resolve, 0)); | ||
} | ||
} | ||
return data; | ||
}; | ||
test/xrpc-sync.test.js+27−3
| @@ -75,10 +75,14 @@ function createMockStorage() { | ||
} | ||
| ||
/** | ||
* Sync handlers over a real one-record repository. | ||
* @param {{did?: string|null, deactivated?: boolean}} [options] | ||
* Sync handlers over a real repository of one record, and `extraRecords` more. | ||
* @param {{did?: string|null, deactivated?: boolean, extraRecords?: number}} [options] | ||
*/ | ||
async function createSync({ did = DID, deactivated = false } = {}) { | ||
async function createSync({ | ||
did = DID, | ||
deactivated = false, | ||
extraRecords = 0, | ||
} = {}) { | ||
const { privateKey } = await generateKeyPair(); | ||
const { carBytes } = await buildRepositoryCar({ | ||
did: DID, | ||
| @@ -89,6 +93,11 @@ async function createSync({ did = DID, deactivated = false } = {}) { | ||
rkey: RKEY, | ||
record: { $type: COLLECTION, text: 'synced' }, | ||
}, | ||
...Array.from({ length: extraRecords }, (_, i) => ({ | ||
collection: COLLECTION, | ||
rkey: `3extra${String(i).padStart(6, '0')}`, | ||
record: { $type: COLLECTION, text: `extra ${i}` }, | ||
})), | ||
], | ||
}); | ||
| ||
| @@ -234,6 +243,21 @@ describe('getRepo', () => { | ||
expect(response.status).toBe(404); | ||
}); | ||
| ||
it('lets a request that arrives meanwhile run before it ends', async () => { | ||
// More than one BLOCKS_PER_WAIT of blocks, so the walk waits at least once. | ||
const { sync } = await createSync({ extraRecords: 600 }); | ||
/** @type {string[]} */ | ||
const order = []; | ||
const served = call(sync, '/xrpc/com.atproto.sync.getRepo').then(() => { | ||
order.push('getRepo'); | ||
}); | ||
// A timer is a task, as an arriving request is. | ||
setTimeout(() => order.push('arrived'), 0); | ||
await served; | ||
| ||
expect(order).toEqual(['arrived', 'getRepo']); | ||
}); | ||
| ||
it('is a 500 naming the block when a descendant is missing', async () => { | ||
const { sync, actorStorage } = await createSync(); | ||
const commitCid = actorStorage.commits[0].cid; | ||