chadtmiller.com/pds.js
An atproto PDS in JavaScript, for Cloudflare Workers and Node
0
0
A space write answers before its notices go out
.changeset/space-notices-after-answer.mdnew+15−0
| @@ -0,0 +1,15 @@ | ||
--- | ||
'@pdsjs/core': patch | ||
'@pdsjs/cloudflare': patch | ||
'@pdsjs/spaces': patch | ||
--- | ||
| ||
A space write answers before its notices go out. The notice to each syncer of | ||
the account's own space, and the first try of the notice to another space's | ||
authority, run after the response. The notice to an authority is still stored | ||
first and retried until it lands. Each notice waits at most 3 seconds for an | ||
answer. A slow syncer made every write into its space as slow as itself. | ||
| ||
An extension's host has `background(work)`, which runs work after the response. | ||
Cloudflare keeps the Durable Object running until that work ends. A platform | ||
passes `waitUntil` to the server for that. | ||
packages/cloudflare/src/index.js+1−0
| @@ -1273,6 +1273,7 @@ export class PDSDurableObject { | ||
const current = await state.storage.getAlarm(); | ||
if (!current || at < current) await state.storage.setAlarm(at); | ||
}, | ||
waitUntil: (work) => state.waitUntil(work), | ||
actorStorage, | ||
sharedStorage, | ||
blobs, | ||
packages/core/src/pds.js+9−0
| @@ -197,6 +197,7 @@ export class PersonalDataServer { | ||
* @param {string} [config.updateUrl] - Where a person updates this deployment to a newer release, shown by the account page when one is published. A wizard-deployed worker points at the wizard's update page. Absent, the page names the release without a link. | ||
* @param {string} [config.scriptName] - The worker script name this deployment runs as. The update link carries it, so the update page can find the worker when the PDS answers on a custom domain rather than its workers.dev hostname. | ||
* @param {(at: number) => Promise<void>|void} [config.scheduleWakeup] - Asks the platform to call runExtensionTicks no later than `at` (epoch ms). A platform that already calls it on a short interval leaves this out. | ||
* @param {(work: Promise<unknown>) => void} [config.waitUntil] - Keeps the platform running until `work` settles, after the response is sent. A platform that never cancels work after a response leaves this out. | ||
*/ | ||
constructor({ | ||
actorStorage, | ||
| @@ -227,6 +228,7 @@ export class PersonalDataServer { | ||
updateUrl, | ||
scriptName, | ||
scheduleWakeup, | ||
waitUntil, | ||
}) { | ||
this.actorStorage = actorStorage; | ||
this.sharedStorage = sharedStorage; | ||
| @@ -261,6 +263,7 @@ export class PersonalDataServer { | ||
/** @type {Promise<void>} */ | ||
this._repoLock = Promise.resolve(); | ||
this.scheduleWakeup = scheduleWakeup || null; | ||
this.waitUntil = waitUntil || null; | ||
| ||
// Every host function late-binds into this server, so a factory runs | ||
// before the handlers below exist and its extension reaches them at | ||
| @@ -307,6 +310,12 @@ export class PersonalDataServer { | ||
scheduleWakeup: async (at) => { | ||
await server.scheduleWakeup?.(at); | ||
}, | ||
background: (work) => { | ||
const settled = work.catch((e) => | ||
console.error('Background work failed:', e), | ||
); | ||
server.waitUntil?.(settled); | ||
}, | ||
}; | ||
// Each extension gets the host with its storage behind a scope over the | ||
// collections the extension declares. The scope reads the declaration | ||
packages/core/src/ports.js+1−0
| @@ -552,6 +552,7 @@ | ||
* @property {(request: Request) => Promise<string|null>} readAccountSession - The DID behind the request's account-page session cookie, or null. For a route that answers in its own shape and takes the owner's session alongside a bearer token. | ||
* @property {(request: Request) => Promise<unknown|Response>} readJsonBounded - The request's JSON body, or a 413 Response when the body exceeds the server's JSON limit. | ||
* @property {(at: number) => Promise<void>} scheduleWakeup - Asks the platform to run the extensions' `tick` no later than `at` (epoch ms). On a platform that runs ticks on a short interval this does nothing. | ||
* @property {(work: Promise<unknown>) => void} background - Runs `work` after the response is sent. The platform keeps the request alive until `work` settles. An error from `work` is logged. | ||
*/ | ||
| ||
/** | ||
packages/core/test/extension-tick.test.js+47−1
| @@ -4,8 +4,9 @@ import { describe, expect, test, vi } from 'vitest'; | ||
/** | ||
* @param {Array<import('../src/ports.js').ExtensionFactory>} extensions | ||
* @param {(at: number) => void} [scheduleWakeup] | ||
* @param {(work: Promise<unknown>) => void} [waitUntil] | ||
*/ | ||
function server(extensions, scheduleWakeup) { | ||
function server(extensions, scheduleWakeup, waitUntil) { | ||
return new PersonalDataServer({ | ||
actorStorage: /** @type {any} */ ({}), | ||
sharedStorage: /** @type {any} */ ({}), | ||
| @@ -15,6 +16,7 @@ function server(extensions, scheduleWakeup) { | ||
password: 'pw', | ||
extensions, | ||
scheduleWakeup, | ||
waitUntil, | ||
}); | ||
} | ||
| ||
| @@ -82,4 +84,48 @@ describe('extension ticks', () => { | ||
]); | ||
await expect(host?.scheduleWakeup(1234)).resolves.toBeUndefined(); | ||
}); | ||
| ||
test('hands background work to the platform, settled', async () => { | ||
const error = vi.spyOn(console, 'error').mockImplementation(() => {}); | ||
/** @type {Promise<unknown>[]} */ | ||
const kept = []; | ||
/** @type {import('../src/ports.js').ExtensionHost|undefined} */ | ||
let host; | ||
server( | ||
[ | ||
(h) => { | ||
host = h; | ||
return { name: 'a' }; | ||
}, | ||
], | ||
undefined, | ||
(work) => { | ||
kept.push(work); | ||
}, | ||
); | ||
host?.background(Promise.reject(new Error('boom'))); | ||
expect(kept).toHaveLength(1); | ||
await expect(kept[0]).resolves.toBeUndefined(); | ||
expect(error).toHaveBeenCalledWith( | ||
'Background work failed:', | ||
expect.any(Error), | ||
); | ||
error.mockRestore(); | ||
}); | ||
| ||
test('runs background work without a platform hook', async () => { | ||
/** @type {import('../src/ports.js').ExtensionHost|undefined} */ | ||
let host; | ||
server([ | ||
(h) => { | ||
host = h; | ||
return { name: 'a' }; | ||
}, | ||
]); | ||
const ran = vi.fn(); | ||
host?.background(Promise.resolve().then(ran)); | ||
await Promise.resolve(); | ||
await Promise.resolve(); | ||
expect(ran).toHaveBeenCalled(); | ||
}); | ||
}); | ||
packages/spaces/src/handlers/write.js+5−1
| @@ -179,6 +179,8 @@ function writeAction(write) { | ||
* @param {import('@pdsjs/core/ports').DidResolverPort} ctx.resolveDid | ||
* @param {{collections: string[], check: (write: {space: string|null, collection: string, rkey: string, prevValue: unknown|null, nextValue: unknown|null}) => Promise<void>}} [ctx.recordGuard] | ||
* @param {typeof fetch} [ctx.fetch] | ||
* @param {(work: Promise<unknown>) => void} [ctx.background] - runs the | ||
* syncers' notices after the response; absent, they run before it | ||
* @param {ReturnType<typeof createNotifyOutbox>} [ctx.outbox] - built here | ||
* from the rest of ctx when absent | ||
* @returns {import('@pdsjs/core/pds').Routes} | ||
| @@ -277,7 +279,7 @@ export function createWriteRoutes(ctx) { | ||
data, | ||
}); | ||
if (!sequenced) return; | ||
await forwardToSyncers(ctx, { | ||
const forwarded = forwardToSyncers(ctx, { | ||
authorityDid: did, | ||
space, | ||
repo: did, | ||
| @@ -285,6 +287,8 @@ export function createWriteRoutes(ctx) { | ||
data, | ||
...sequenced, | ||
}); | ||
if (ctx.background) ctx.background(forwarded); | ||
else await forwarded; | ||
return; | ||
} | ||
await outbox.deliver(space, { rev: commit.rev, data }); | ||
packages/spaces/src/extension.js+2−0
| @@ -34,6 +34,7 @@ export function createSpacesExtension({ fetch: fetchImpl } = {}) { | ||
resolveDid: didResolver, | ||
fetch: fetchImpl ?? ((url, init) => fetch(url, init)), | ||
scheduleWakeup: host.scheduleWakeup, | ||
background: host.background, | ||
}); | ||
return { | ||
name: 'spaces', | ||
| @@ -63,6 +64,7 @@ export function createSpacesExtension({ fetch: fetchImpl } = {}) { | ||
getKemKey: host.getKemKey, | ||
fetch: fetchImpl ?? ((url, init) => fetch(url, init)), | ||
outbox, | ||
background: host.background, | ||
}), | ||
...createSpaceAccountRoutes({ | ||
host, | ||
packages/spaces/src/notify.js+4−4
| @@ -7,6 +7,7 @@ | ||
// carries no records, only the repo rev reached and the space revision the | ||
// authority gave the update. | ||
| ||
import { NOTICE_TIMEOUT_MS } from './outbox.js'; | ||
import { createServiceAuth } from './service-auth.js'; | ||
| ||
/** | ||
| @@ -17,10 +18,8 @@ import { createServiceAuth } from './service-auth.js'; | ||
* own next walk. One unreachable recipient must not fail the others, and none of | ||
* them may fail the write that produced the notice. | ||
* | ||
* The caller awaits this, which puts one round trip per recipient in front of | ||
* the response. A Worker cancels a promise still running when it answers, and | ||
* this package holds no execution context to hand the work to, so backgrounding | ||
* it here would mean dropping notifications on Cloudflare. | ||
* The write handler runs this after its response through the host's | ||
* `background`, so no recipient delays the write. | ||
* | ||
* @param {Object} ctx | ||
* @param {import('@pdsjs/core/ports').SpaceStoragePort} ctx.spaceStorage | ||
| @@ -77,6 +76,7 @@ export async function forwardToSyncers( | ||
authorization: `Bearer ${token}`, | ||
}, | ||
body, | ||
signal: AbortSignal.timeout(NOTICE_TIMEOUT_MS), | ||
}, | ||
); | ||
} catch { | ||
packages/spaces/src/outbox.js+10−1
| @@ -21,6 +21,8 @@ export const DELIVERY_DEADLINE_MS = 24 * 60 * 60 * 1000; | ||
const FIRST_RETRY_MS = 10 * 1000; | ||
const MAX_RETRY_MS = 60 * 60 * 1000; | ||
const BATCH = 50; | ||
/** How long one delivery attempt waits for an answer. */ | ||
export const NOTICE_TIMEOUT_MS = 3000; | ||
| ||
/** | ||
* The wait before retry number `attempts`, counted from 1. | ||
| @@ -48,6 +50,8 @@ function isRetryable(status) { | ||
* @param {import('@pdsjs/core/ports').DidResolverPort} ctx.resolveDid | ||
* @param {typeof fetch} [ctx.fetch] | ||
* @param {(at: number) => Promise<void>} [ctx.scheduleWakeup] | ||
* @param {(work: Promise<unknown>) => void} [ctx.background] - runs the first | ||
* delivery attempt after the response; absent, the attempt runs before it | ||
* @param {() => number} [ctx.now] - epoch ms | ||
*/ | ||
export function createNotifyOutbox(ctx) { | ||
| @@ -75,6 +79,7 @@ export function createNotifyOutbox(ctx) { | ||
authorization: `Bearer ${token}`, | ||
}, | ||
body: JSON.stringify(body), | ||
signal: AbortSignal.timeout(NOTICE_TIMEOUT_MS), | ||
}); | ||
if (res.ok) return 'delivered'; | ||
return isRetryable(res.status) ? 'retry' : 'failed'; | ||
| @@ -241,7 +246,11 @@ export function createNotifyOutbox(ctx) { | ||
deadlineAt: new Date(at + DELIVERY_DEADLINE_MS).toISOString(), | ||
}; | ||
await ctx.spaceStorage.enqueueSpaceNotify(row); | ||
await settleWrite(row, await sendWrite(space, rev, data)); | ||
const attempt = sendWrite(space, rev, data).then((outcome) => | ||
settleWrite(row, outcome), | ||
); | ||
if (ctx.background) ctx.background(attempt); | ||
else await attempt; | ||
}, | ||
| ||
/** | ||
packages/spaces/src/routes.js+2−0
| @@ -31,6 +31,8 @@ import { createNotifyOutbox } from './outbox.js'; | ||
* - a veto over writes in its named collections, e.g. git branch protection | ||
* @param {typeof fetch} [ctx.fetch] - injected for the outbound notification | ||
* calls, so tests can observe them without a network | ||
* @param {(work: Promise<unknown>) => void} [ctx.background] - runs write | ||
* notices after the response; absent, they run before it | ||
* @param {ReturnType<typeof import('./outbox.js').createNotifyOutbox>} [ctx.outbox] | ||
* - notices owed to other services; built from ctx when absent | ||
* @returns {import('@pdsjs/core/pds').Routes} | ||