chadtmiller.com/pds.js
An atproto PDS in JavaScript, for Cloudflare Workers and Node
0
0
// @pdsjs/core/handlers/account-backup - The snapshot engine and its account API routes.
//
// A snapshot is `snapshots/<stamp>/{repo.car, manifest.json, key.enc?}` in the
// backup target, with blobs deduplicated across snapshots at `blobs/<cid>` —
// content-addressed, so a run only uploads what the target does not already
// hold. The manifest is written last: a snapshot without one is incomplete and
// never counts for restore or blob retention.
import {
BACKUP_CHECKPOINT_MS,
BACKUP_RUNS_KEPT,
defaultBackupState,
isStaleRun,
keyFingerprint,
sealKeyBackup,
snapshotStamp,
} from '../backup.js';
import { base64UrlDecode } from '../crypto.js';
/** Heads a snapshot tries before giving up, should commits keep landing. */
const SNAPSHOT_TRIES = 3;
/**
* Count the bytes of a stream as they pass, and give the reader a place to
* checkpoint between them. One blob is more than one request to the target,
* and the run's heartbeat has to keep beating across them.
* @param {ReadableStream<Uint8Array>} body
* @param {(bytes: number) => Promise<void>} onBytes
* @returns {ReadableStream<Uint8Array>}
*/
function meterStream(body, onBytes) {
/** @type {TransformStream<Uint8Array, Uint8Array>} */
const meter = new TransformStream({
async transform(chunk, controller) {
controller.enqueue(chunk);
await onBytes(chunk.byteLength);
},
});
return body.pipeThrough(meter);
}
/**
* These are structural types. Thus a member does not name the module that owns
* the function you give it.
*
* @typedef {Object} BackupContext
* @property {import('../ports.js').ActorStoragePort} actorStorage
* @property {import('../ports.js').BlobPort} blobs
* @property {import('../ports.js').BackupTargetPort|null} backupTarget - Absent, backups report unavailable and the scheduler never runs
* @property {string} [hostname] - Recorded in the manifest
* @property {() => Promise<string|null>} getDid
* @property {(commitCid: string) => Promise<Uint8Array>} buildFullRepoCar
* @property {(request: Request, url: URL, handler: (did: string, body: any) => Promise<unknown>, options?: {write?: boolean}) => Promise<Response>} accountApi
*/
/**
* @param {BackupContext} ctx
* @returns {{
* backupsAvailable: () => boolean,
* readBackupState: () => Promise<import('../backup.js').BackupState>,
* writeBackupState: (state: import('../backup.js').BackupState) => Promise<void>,
* maybeRunScheduledBackup: (now?: number) => Promise<import('../backup.js').BackupRun|null>,
* runBackup: (trigger?: 'manual'|'scheduled') => Promise<import('../backup.js').BackupRun>,
* pruneBackups: (retain: number) => Promise<void>,
* publicBackupView: (state: import('../backup.js').BackupState) => Promise<{available: boolean, settings: {enabled: boolean, intervalHours: number, retain: number, keyBackup: {createdAt: string, stale: boolean}|null}, nextRunAt: number|null, runs: import('../backup.js').BackupRun[]}>,
* routes: import('../pds.js').Routes,
* }}
*/
export function createBackupHandlers(ctx) {
const {
actorStorage,
blobs,
backupTarget,
hostname,
getDid,
buildFullRepoCar,
accountApi,
} = ctx;
// This flag applies to one isolate. The heartbeat in the stored run is the
// claim that continues after a restart.
let backupRunning = false;
/**
* Whether this deployment can take backups: a target to write to, and an
* adapter that can hold the settings document.
* @returns {boolean}
*/
function backupsAvailable() {
return Boolean(
backupTarget &&
actorStorage.getBackupState &&
actorStorage.setBackupState,
);
}
/**
* The stored backup document, over defaults so a document written before a
* field existed still reads whole.
* @returns {Promise<import('../backup.js').BackupState>}
*/
async function readBackupState() {
const fallback = defaultBackupState();
const stored = actorStorage.getBackupState
? await actorStorage.getBackupState()
: null;
if (!stored) return fallback;
return {
config: { ...fallback.config, ...stored.config },
runs: stored.runs || [],
nextRunAt: stored.nextRunAt ?? null,
};
}
/**
* @param {import('../backup.js').BackupState} state
* @returns {Promise<void>}
*/
async function writeBackupState(state) {
if (actorStorage.setBackupState) {
await actorStorage.setBackupState(state);
}
}
/**
* Run a backup if one is enabled and due. Called from each platform's
* hourly tick (interval on Node and Deno, alarm on Cloudflare), so
* schedules land within an hour of their mark.
* @param {number} [now] - Epoch ms
* @returns {Promise<import('../backup.js').BackupRun|null>} The finished
* run, or null when nothing was due
*/
async function maybeRunScheduledBackup(now = Date.now()) {
if (!backupsAvailable() || backupRunning) return null;
const state = await readBackupState();
if (!state.config.enabled || !state.nextRunAt || now < state.nextRunAt) {
return null;
}
// A run in another isolate is still proving itself alive; don't stack.
const running = state.runs.find((r) => r.status === 'running');
if (running && !isStaleRun(running, now)) return null;
return runBackup('scheduled');
}
/**
* Take one snapshot now. Execution failures land in the returned run as
* status 'error' rather than throwing; only being unable to start —
* no target, no account, a run already going — throws.
* @param {'manual'|'scheduled'} [trigger]
* @returns {Promise<import('../backup.js').BackupRun>}
*/
async function runBackup(trigger = 'manual') {
const target = backupTarget;
if (!target || !backupsAvailable()) {
throw new Error('No backup target is configured');
}
if (backupRunning) throw new Error('A backup is already running');
backupRunning = true;
try {
const did = await getDid();
if (!did) throw new Error('No account on this server yet');
let commit = await actorStorage.getLatestCommit();
if (!commit) throw new Error('No repository to back up');
const state = await readBackupState();
const startedAt = new Date();
// The lock is the persisted heartbeat, not just the flag above: a run
// whose isolate died mid-copy stops beating and loses its claim, so an
// interruption can never wedge backups until a restart.
const running = state.runs.find((r) => r.status === 'running');
if (running && !isStaleRun(running, startedAt.getTime())) {
throw new Error('A backup is already running');
}
if (running) {
running.status = 'error';
running.error = 'Interrupted — the server stopped mid-run';
running.finishedAt = running.heartbeatAt || running.startedAt;
}
// Two runs can share a millisecond; nudge the stamp until the name is
// free so the second run cannot overwrite the first.
let stampTime = startedAt.getTime();
let stamp = snapshotStamp(startedAt);
while (state.runs.some((r) => r.id === stamp)) {
stampTime += 1;
stamp = snapshotStamp(new Date(stampTime));
}
const snapshot = `snapshots/${stamp}`;
/** @type {import('../backup.js').BackupRun} */
const run = {
id: stamp,
trigger,
status: 'running',
startedAt: startedAt.toISOString(),
finishedAt: null,
heartbeatAt: startedAt.toISOString(),
blobsCopied: 0,
blobsTotal: 0,
blobsFailed: 0,
carBytes: 0,
bytesUploaded: 0,
snapshot,
keyIncluded: Boolean(state.config.keyBackup),
keyStale: false,
};
// Record the run and advance the schedule before doing the work, so a
// failed run waits out the interval instead of retrying every tick.
state.runs = [run, ...state.runs].slice(0, BACKUP_RUNS_KEPT);
if (state.config.enabled) {
state.nextRunAt =
startedAt.getTime() + state.config.intervalHours * 3600_000;
}
await writeBackupState(state);
// Persist the run's counters every few seconds: the page polls them as
// live progress, and the fresh heartbeat is what keeps this run's claim
// on the lock. Re-read before writing so a settings save that lands
// mid-run is never clobbered.
let lastCheckpoint = Date.now();
const checkpoint = async (force = false) => {
if (!force && Date.now() - lastCheckpoint < BACKUP_CHECKPOINT_MS) {
return;
}
lastCheckpoint = Date.now();
run.heartbeatAt = new Date().toISOString();
const fresh = await readBackupState();
fresh.runs = [run, ...fresh.runs.filter((r) => r.id !== run.id)].slice(
0,
BACKUP_RUNS_KEPT,
);
await writeBackupState(fresh);
};
try {
/** @type {Uint8Array|null} */
let car = null;
for (let tries = 1; !car; tries++) {
try {
car = await buildFullRepoCar(commit.cid);
} catch (err) {
// A commit landing during the walk drops nodes of the tree it has
// not reached yet: the snapshot is of the new head instead.
const head = await actorStorage.getLatestCommit();
if (tries === SNAPSHOT_TRIES || !head || head.cid === commit.cid) {
throw err;
}
commit = head;
}
}
run.carBytes = car.byteLength;
await checkpoint(true);
// Blobs the target already holds are skipped, which is what makes
// every run after the first cheap.
const existing = new Set(await target.list('blobs/'));
/** @type {string[]} */
const blobCids = [];
/** @type {string[]} */
const missing = [];
/** @type {Array<{cid: string, error: string}>} */
const blobErrors = [];
/** @type {string|null} */
let cursor = null;
do {
const page = await actorStorage.listBlobs(cursor, 500);
for (const cid of page.cids) {
blobCids.push(cid);
if (existing.has(`blobs/${cid}`)) continue;
try {
// A blob can be larger than the memory this process has, and
// larger than one request to the target. Stream it when both
// sides can, and read it whole only when one of them cannot.
if (target.putStream && blobs.getStream) {
const streamed = await blobs.getStream(did, cid);
if (!streamed) {
missing.push(cid);
continue;
}
await target.putStream(
`blobs/${cid}`,
meterStream(streamed.body, async (bytes) => {
run.bytesUploaded += bytes;
await checkpoint();
}),
streamed.mimeType,
);
} else {
const blob = await blobs.get(did, cid);
if (!blob) {
missing.push(cid);
continue;
}
await target.put(`blobs/${cid}`, blob.data, blob.mimeType);
run.bytesUploaded += blob.data.byteLength;
}
} catch (err) {
// One blob the target refuses costs the snapshot that blob, not
// the repo and not the other blobs. The next run tries it again:
// a blob the target does not hold is never skipped. A target
// that is failing for every blob fails the CAR too, a few lines
// down, and that does end the run.
const message = err instanceof Error ? err.message : String(err);
missing.push(cid);
blobErrors.push({ cid, error: message });
run.blobsFailed += 1;
if (!run.blobError) run.blobError = message;
await checkpoint();
continue;
}
run.blobsCopied += 1;
await checkpoint();
}
cursor = page.cursor;
} while (cursor);
run.blobsTotal = blobCids.length;
await checkpoint(true);
const keyBackup = state.config.keyBackup;
if (keyBackup) {
const stored = await actorStorage.getPrivateKey();
run.keyStale =
!stored || (await keyFingerprint(stored)) !== keyBackup.keyHash;
}
await target.put(
`${snapshot}/repo.car`,
car,
'application/vnd.ipld.car',
);
run.bytesUploaded += car.byteLength;
if (keyBackup) {
const keyBytes = new TextEncoder().encode(
JSON.stringify(keyBackup, null, 2),
);
await target.put(`${snapshot}/key.enc`, keyBytes, 'application/json');
run.bytesUploaded += keyBytes.byteLength;
}
const manifest = {
version: 1,
did,
handle: await actorStorage.getHandle(),
hostname: hostname ?? null,
createdAt: startedAt.toISOString(),
commit: { cid: commit.cid, rev: commit.rev, seq: commit.seq },
carBytes: car.byteLength,
blobs: blobCids,
// Everything this snapshot does not hold, whatever the cause, so a
// restore reads one list. `blobErrors` says why for the ones the
// target refused.
missingBlobs: missing,
blobErrors,
keyIncluded: Boolean(keyBackup),
keyStale: run.keyStale,
};
const manifestBytes = new TextEncoder().encode(
JSON.stringify(manifest, null, 2),
);
await target.put(
`${snapshot}/manifest.json`,
manifestBytes,
'application/json',
);
run.bytesUploaded += manifestBytes.byteLength;
await checkpoint(true);
await pruneBackups(state.config.retain);
run.status = 'ok';
} catch (err) {
run.status = 'error';
run.error = err instanceof Error ? err.message : String(err);
}
run.finishedAt = new Date().toISOString();
// Re-read before persisting the finished run: settings may have been
// saved while the copy ran, and this write must not undo them.
const fresh = await readBackupState();
fresh.runs = [run, ...fresh.runs.filter((r) => r.id !== run.id)].slice(
0,
BACKUP_RUNS_KEPT,
);
await writeBackupState(fresh);
return run;
} finally {
backupRunning = false;
}
}
/**
* Drop snapshots beyond the newest `retain`, then delete blobs no kept
* manifest references. Blob GC is skipped entirely if no kept manifest is
* readable — better to hold garbage than to reap blobs still needed.
* @param {number} retain
* @returns {Promise<void>}
*/
async function pruneBackups(retain) {
const target = backupTarget;
if (!target) return;
/** @type {Map<string, string[]>} */
const byStamp = new Map();
for (const key of await target.list('snapshots/')) {
const stamp = key.slice('snapshots/'.length).split('/')[0];
if (!stamp) continue;
const keys = byStamp.get(stamp) ?? [];
keys.push(key);
byStamp.set(stamp, keys);
}
const stamps = [...byStamp.keys()].sort().reverse();
const keep = stamps.slice(0, Math.max(1, retain));
for (const stamp of stamps.slice(Math.max(1, retain))) {
for (const key of byStamp.get(stamp) ?? []) {
await target.delete(key);
}
}
const referenced = new Set();
let readableManifests = 0;
for (const stamp of keep) {
const bytes = await target.get(`snapshots/${stamp}/manifest.json`);
if (!bytes) continue;
try {
const manifest = JSON.parse(new TextDecoder().decode(bytes));
readableManifests += 1;
for (const cid of manifest.blobs || []) referenced.add(`blobs/${cid}`);
} catch {
// An unreadable manifest names blobs we cannot know; the guard
// below keeps GC from running on partial knowledge.
readableManifests = 0;
break;
}
}
if (readableManifests === 0) return;
for (const key of await target.list('blobs/')) {
if (!referenced.has(key)) await target.delete(key);
}
}
/**
* The backup document as the account API tells it: everything but the
* sealed key ciphertext, which never leaves the server this way.
* @param {import('../backup.js').BackupState} state
* @returns {Promise<{available: boolean, settings: {enabled: boolean, intervalHours: number, retain: number, keyBackup: {createdAt: string, stale: boolean}|null}, nextRunAt: number|null, runs: import('../backup.js').BackupRun[]}>}
*/
async function publicBackupView(state) {
const keyBackup = state.config.keyBackup;
/** @type {{createdAt: string, stale: boolean}|null} */
let keyView = null;
if (keyBackup) {
const stored = await actorStorage.getPrivateKey();
keyView = {
createdAt: keyBackup.createdAt,
stale: !stored || (await keyFingerprint(stored)) !== keyBackup.keyHash,
};
}
return {
available: backupsAvailable(),
settings: {
enabled: state.config.enabled,
intervalHours: state.config.intervalHours,
retain: state.config.retain,
keyBackup: keyView,
},
nextRunAt: state.config.enabled ? state.nextRunAt : null,
// A run that stopped beating is shown as what it is. The stored row is
// corrected for real when the next run claims the lock.
runs: state.runs.map((r) =>
isStaleRun(r, Date.now())
? {
...r,
status: /** @type {const} */ ('error'),
error: 'Interrupted — the server stopped mid-run',
finishedAt: r.heartbeatAt || r.startedAt,
}
: r,
),
};
}
/**
* GET /account/api/backups - Settings, schedule, and run history
* @param {Request} request
* @param {URL} url
* @returns {Promise<Response>}
*/
async function handleApiBackups(request, url) {
return accountApi(request, url, async () =>
publicBackupView(await readBackupState()),
);
}
/**
* POST /account/api/backups/settings - Change any subset of the settings;
* a passphrase here seals the signing key into future snapshots, and
* removeKeyBackup takes it back out.
* @param {Request} request
* @param {URL} url
* @returns {Promise<Response>}
*/
async function handleApiBackupsSettings(request, url) {
return accountApi(
request,
url,
async (_did, body) => {
if (!backupsAvailable()) {
throw new Error('No backup target is configured');
}
const state = await readBackupState();
const config = state.config;
if (typeof body.enabled === 'boolean') config.enabled = body.enabled;
if (body.intervalHours !== undefined) {
const hours = Number(body.intervalHours);
if (!Number.isFinite(hours) || hours < 1 || hours > 720) {
throw new Error('intervalHours must be between 1 and 720');
}
config.intervalHours = hours;
}
if (body.retain !== undefined) {
const retain = Number(body.retain);
if (!Number.isInteger(retain) || retain < 1 || retain > 50) {
throw new Error('retain must be between 1 and 50');
}
config.retain = retain;
}
if (body.removeKeyBackup === true) config.keyBackup = null;
// The browser derives the AES key from the passphrase (PBKDF2 at a
// count Workers' WebCrypto refuses to run) and sends key, salt and
// count. The passphrase itself never reaches the server.
if (body.keyDerived) {
const { key, salt, iterations } = body.keyDerived;
if (typeof key !== 'string' || typeof salt !== 'string') {
throw new Error('keyDerived needs base64url key and salt');
}
if (
!Number.isInteger(iterations) ||
iterations < 100_000 ||
iterations > 10_000_000
) {
throw new Error('keyDerived.iterations out of range');
}
const derivedKey = base64UrlDecode(key);
const saltBytes = base64UrlDecode(salt);
if (derivedKey.byteLength !== 32 || saltBytes.byteLength < 8) {
throw new Error('keyDerived key must be 32 bytes, salt at least 8');
}
const stored = await actorStorage.getPrivateKey();
if (!stored) throw new Error('Server has no signing key');
config.keyBackup = await sealKeyBackup(
{ derivedKey, salt: saltBytes, iterations },
stored,
);
}
if (config.enabled) {
// Due one interval after the last run, whatever its outcome — or
// right away if there has never been one. The hourly tick picks it
// up from there.
const lastRun = state.runs[0];
state.nextRunAt = lastRun
? Date.parse(lastRun.startedAt) + config.intervalHours * 3600_000
: Date.now();
} else {
state.nextRunAt = null;
}
await writeBackupState(state);
return publicBackupView(state);
},
{ write: true },
);
}
/**
* POST /account/api/backups/run - Take a snapshot now. Answers with the
* finished run; a run that failed partway is an 'error' run, not an HTTP
* error.
* @param {Request} request
* @param {URL} url
* @returns {Promise<Response>}
*/
async function handleApiBackupsRun(request, url) {
return accountApi(request, url, async () => runBackup('manual'), {
write: true,
});
}
return {
backupsAvailable,
readBackupState,
writeBackupState,
maybeRunScheduledBackup,
runBackup,
pruneBackups,
publicBackupView,
routes: {
'/account/api/backups': {
method: 'GET',
handler: handleApiBackups,
},
'/account/api/backups/settings': {
method: 'POST',
handler: handleApiBackupsSettings,
},
'/account/api/backups/run': {
method: 'POST',
handler: handleApiBackupsRun,
},
},
};
}