chadtmiller.com/pds.js
An atproto PDS in JavaScript, for Cloudflare Workers and Node
0
0
// @pdsjs/spaces/outbox - notices this server owes other services, persisted
// before they are sent and retried until they land.
//
// Two kinds. As a repo host, this server owes the authority of each space it
// writes into a notifyWrite for the repo's latest rev: the authority's writer
// set, and with it listRepos, is built from these notices. As an authority, it
// owes every syncer registered for a space it deleted a notifySpaceDeleted.
//
// A notice that fails for a reason that may pass is retried with backoff. A
// write notice stops when it lands, when a newer write to the same space
// replaces it, or a day after its write. A deletion notice stops when it
// lands, when the syncer's registration expires, or when the space exists
// again.
import { spaceHostAudience, spaceHostEndpoint } from './authority.js';
import { createServiceAuth } from './service-auth.js';
import { parseSpaceUri } from './uri.js';
/** How long one repo rev is retried before delivery stops. */
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.
* @param {number} attempts
* @returns {number}
*/
export function retryDelay(attempts) {
return Math.min(FIRST_RETRY_MS * 2 ** (attempts - 1), MAX_RETRY_MS);
}
/**
* Whether a response says the same request may succeed later.
* @param {number} status
* @returns {boolean}
*/
function isRetryable(status) {
return status === 408 || status === 425 || status === 429 || status >= 500;
}
/**
* @param {Object} ctx
* @param {import('@pdsjs/core/ports').SpaceStoragePort} ctx.spaceStorage
* @param {() => Promise<string|null>} ctx.getDid
* @param {() => Promise<import('./token.js').SpaceSigner>} ctx.getSigner
* @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) {
const now = ctx.now ?? Date.now;
/**
* POST a notice, signed as `iss` for `aud`.
* @param {string} url
* @param {{iss: string, aud: string, lxm: string}} auth
* @param {Record<string, unknown>} body
* @returns {Promise<'delivered'|'retry'|'failed'>}
*/
async function post(url, { iss, aud, lxm }, body) {
try {
const token = await createServiceAuth({
iss,
aud,
lxm,
signer: await ctx.getSigner(),
});
const res = await (ctx.fetch ?? fetch)(url, {
method: 'POST',
headers: {
'content-type': 'application/json',
authorization: `Bearer ${token}`,
},
body: JSON.stringify(body),
signal: AbortSignal.timeout(NOTICE_TIMEOUT_MS),
});
if (res.ok) return 'delivered';
return isRetryable(res.status) ? 'retry' : 'failed';
} catch {
return 'retry';
}
}
/**
* One delivery attempt of a write notice.
* @param {string} space
* @param {string} repoRev
* @param {Uint8Array} data
* @returns {Promise<'delivered'|'retry'|'failed'>}
*/
async function sendWrite(space, repoRev, data) {
const did = await ctx.getDid();
if (!did) return 'failed';
/** @type {string} */
let spaceDid;
try {
({ spaceDid } = parseSpaceUri(space));
} catch {
return 'failed';
}
/** @type {string} */
let endpoint;
try {
const didDoc = await ctx.resolveDid(spaceDid);
if (!didDoc) return 'retry';
endpoint = spaceHostEndpoint(didDoc, spaceDid);
} catch {
// A DID document naming no PDS yet.
return 'retry';
}
return post(
`${endpoint}/xrpc/com.atproto.space.notifyWrite`,
{
iss: did,
aud: spaceHostAudience(spaceDid),
lxm: 'com.atproto.space.notifyWrite',
},
{ space, repo: did, repoRev, data: { $bytes: toBase64(data) } },
);
}
/**
* One delivery attempt of a deletion notice.
* @param {string} space
* @param {string} service - the syncer's service identifier
* @param {string} serviceEndpoint
* @returns {Promise<'delivered'|'retry'|'failed'>}
*/
async function sendDeleted(space, service, serviceEndpoint) {
const did = await ctx.getDid();
if (!did) return 'failed';
return post(
`${serviceEndpoint}/xrpc/com.atproto.space.notifySpaceDeleted`,
{ iss: did, aud: service, lxm: 'com.atproto.space.notifySpaceDeleted' },
{ space },
);
}
/**
* When the next retry of a notice that has failed `attempts` times falls,
* or null once that is past its deadline.
* @param {number} attempts
* @param {string} deadlineAt
* @returns {number|null}
*/
function nextAttempt(attempts, deadlineAt) {
const at = now() + retryDelay(attempts);
return at <= Date.parse(deadlineAt) ? at : null;
}
/**
* Act on a write notice's outcome. `attempts` counts the failures before it.
* @param {import('@pdsjs/core/ports').SpaceNotifyRow} row
* @param {'delivered'|'retry'|'failed'} outcome
* @returns {Promise<void>}
*/
async function settleWrite(row, outcome) {
if (outcome === 'retry') {
const attempts = row.attempts + 1;
const at = nextAttempt(attempts, row.deadlineAt);
if (at !== null) {
await ctx.spaceStorage.rescheduleSpaceNotify(
row.space,
row.repoRev,
attempts,
new Date(at).toISOString(),
);
await ctx.scheduleWakeup?.(at);
return;
}
}
await ctx.spaceStorage.deleteSpaceNotify(row.space, row.repoRev);
}
/**
* Act on a deletion notice's outcome.
* @param {import('@pdsjs/core/ports').SpaceDeletedNotifyRow} row
* @param {'delivered'|'retry'|'failed'} outcome
* @returns {Promise<void>}
*/
async function settleDeleted(row, outcome) {
if (outcome === 'retry') {
const attempts = row.attempts + 1;
// The last retry falls on the deadline itself.
const at = Math.min(
now() + retryDelay(attempts),
Date.parse(row.deadlineAt),
);
if (at > now()) {
await ctx.spaceStorage.rescheduleSpaceDeletedNotify(
row.space,
row.service,
attempts,
new Date(at).toISOString(),
);
await ctx.scheduleWakeup?.(at);
return;
}
}
await ctx.spaceStorage.deleteSpaceDeletedNotify(row.space, row.service);
}
/**
* Retry a deletion notice that is due. A space that exists again is not
* deleted any more, and the notice would purge the syncer's new copy.
* @param {import('@pdsjs/core/ports').SpaceDeletedNotifyRow} row
* @returns {Promise<void>}
*/
async function retryDeleted(row) {
const current = await ctx.spaceStorage.getSpace(row.space);
if (!current?.deletedAt || Date.parse(row.deadlineAt) < now()) {
await ctx.spaceStorage.deleteSpaceDeletedNotify(row.space, row.service);
return;
}
await settleDeleted(
row,
await sendDeleted(row.space, row.service, row.serviceEndpoint),
);
}
return {
/**
* Owe a space's authority this write, and try to deliver it now. A newer
* write to the same space replaces an older undelivered one and starts its
* retries and deadline afresh.
* @param {string} space
* @param {{rev: string, data: Uint8Array}} commit
* @returns {Promise<void>}
*/
async deliver(space, { rev, data }) {
const at = now();
/** @type {import('@pdsjs/core/ports').SpaceNotifyRow} */
const row = {
space,
repoRev: rev,
data,
attempts: 0,
nextAttemptAt: new Date(at).toISOString(),
deadlineAt: new Date(at + DELIVERY_DEADLINE_MS).toISOString(),
};
await ctx.spaceStorage.enqueueSpaceNotify(row);
const attempt = sendWrite(space, rev, data).then((outcome) =>
settleWrite(row, outcome),
);
if (ctx.background) ctx.background(attempt);
else await attempt;
},
/**
* Owe every registered syncer a notice that this space is deleted, and
* try each now. A syncer is owed it until its registration expires: after
* that it must renew its credential, and is answered SpaceDeleted.
* @param {string} space
* @param {import('@pdsjs/core/ports').SpaceRecipientRow[]} recipients
* @returns {Promise<void>}
*/
async deliverSpaceDeleted(space, recipients) {
const at = new Date(now()).toISOString();
await Promise.all(
recipients.map(async (recipient) => {
/** @type {import('@pdsjs/core/ports').SpaceDeletedNotifyRow} */
const row = {
space,
service: recipient.serviceDid,
serviceEndpoint: recipient.serviceEndpoint,
attempts: 0,
nextAttemptAt: at,
deadlineAt: recipient.expiresAt,
};
await ctx.spaceStorage.enqueueSpaceDeletedNotify(row);
await settleDeleted(
row,
await sendDeleted(space, row.service, row.serviceEndpoint),
);
}),
);
},
/**
* Retry every notice that is due.
* @returns {Promise<number|null>} when the next one falls due, or null
*/
async tick() {
const nowIso = new Date(now()).toISOString();
const due = await ctx.spaceStorage.listDueSpaceNotify(nowIso, BATCH);
for (const row of due) {
await settleWrite(
row,
await sendWrite(row.space, row.repoRev, row.data),
);
}
const deleted = await ctx.spaceStorage.listDueSpaceDeletedNotify(
nowIso,
BATCH,
);
for (const row of deleted) await retryDeleted(row);
const next = await ctx.spaceStorage.nextSpaceNotifyAt();
return next === null ? null : Date.parse(next);
},
};
}
/**
* @param {Uint8Array} bytes
* @returns {string}
*/
function toBase64(bytes) {
let binary = '';
for (const b of bytes) binary += String.fromCharCode(b);
return btoa(binary).replace(/=+$/, '');
}