chadtmiller.com/pds.js
An atproto PDS in JavaScript, for Cloudflare Workers and Node
0
0
// Backup engine tests - snapshots, incremental blob copy, retention, sealed key
import { describe, expect, it } from 'vitest';
import {
decryptKeyBackup,
defaultBackupState,
encryptKeyBackup,
} from '../packages/core/src/backup.js';
import { parseCarFile } from '../packages/core/src/car.js';
import { generateKeyPair } from '../packages/core/src/crypto.js';
import { loadRepositoryFromCar } from '../packages/core/src/loader.js';
import { PersonalDataServer } from '../packages/core/src/pds.js';
import { buildRepositoryCar } from '../packages/readonly/src/build-car.js';
const DID = 'did:plc:backuptest';
/** An in-memory BackupTargetPort that remembers every put. */
function createMemoryTarget() {
/** @type {Map<string, Uint8Array>} */
const objects = new Map();
return {
objects,
async put(
/** @type {string} */ key,
/** @type {Uint8Array} */ data,
/** @type {string|undefined} */ _contentType,
) {
objects.set(key, data);
},
async putStream(
/** @type {string} */ key,
/** @type {ReadableStream<Uint8Array>} */ body,
/** @type {string|undefined} */ _contentType,
) {
/** @type {Uint8Array[]} */
const chunks = [];
let size = 0;
for await (const chunk of body) {
chunks.push(chunk);
size += chunk.byteLength;
}
const data = new Uint8Array(size);
let at = 0;
for (const chunk of chunks) {
data.set(chunk, at);
at += chunk.byteLength;
}
objects.set(key, data);
},
async get(/** @type {string} */ key) {
return objects.get(key) || null;
},
async list(/** @type {string} */ prefix) {
return [...objects.keys()].filter((key) => key.startsWith(prefix));
},
async delete(/** @type {string} */ key) {
objects.delete(key);
},
};
}
/** Actor storage holding a real loaded repo plus blob metadata. */
function createMockStorage() {
/** @type {Map<string, Uint8Array>} */
const blocks = new Map();
/** @type {Map<string, {cid: string, value: Uint8Array}>} */
const records = new Map();
/** @type {Array<{seq: number, cid: string, rev: string}>} */
const commits = [];
/** @type {Map<string, {mimeType: string, size: number, createdAt: number}>} */
const blobMeta = new Map();
/** @type {{privateKey: Uint8Array|null, kemKey: Uint8Array|null, backupState: any}} */
const metadata = { privateKey: null, kemKey: null, backupState: null };
return {
blobMeta,
metadata,
async getBlock(/** @type {string} */ cid) {
return blocks.get(cid) || null;
},
async putBlock(/** @type {string} */ cid, /** @type {Uint8Array} */ data) {
blocks.set(cid, data);
},
async getRecord(/** @type {string} */ uri) {
return records.get(uri) || null;
},
async putRecord(
/** @type {string} */ uri,
/** @type {string} */ cid,
/** @type {string} */ _collection,
/** @type {string} */ _rkey,
/** @type {Uint8Array} */ value,
) {
records.set(uri, { cid, value });
},
async listAllRecords() {
return [];
},
async getLatestCommit() {
return commits.length > 0 ? commits[commits.length - 1] : null;
},
async putCommit(
/** @type {number} */ seq,
/** @type {string} */ cid,
/** @type {string} */ rev,
) {
commits.push({ seq, cid, rev });
},
async putEvent() {},
async getBlob(/** @type {string} */ cid) {
return blobMeta.get(cid) || null;
},
async listBlobs() {
return { cids: [...blobMeta.keys()].sort(), cursor: null };
},
async getDid() {
return DID;
},
async setDid() {},
async getHandle() {
return 'backup.test';
},
async getPrivateKey() {
return metadata.privateKey;
},
async setPrivateKey(/** @type {Uint8Array} */ key) {
metadata.privateKey = key;
},
async getKemKey() {
return metadata.kemKey ?? null;
},
async setKemKey(/** @type {Uint8Array} */ key) {
metadata.kemKey = key;
},
async getBackupState() {
return metadata.backupState;
},
async setBackupState(/** @type {any} */ state) {
metadata.backupState = state;
},
};
}
/**
* Blob store with a couple of stored blobs. `streaming` adds the optional
* `getStream`, which is what sends a backup down the streaming path.
* @param {{streaming?: boolean}} [options]
*/
function createMockBlobs({ streaming = false } = {}) {
/** @type {Map<string, {data: Uint8Array, mimeType: string}>} */
const store = new Map();
return {
store,
async get(/** @type {string} */ _did, /** @type {string} */ cid) {
return store.get(cid) || null;
},
async put() {},
async delete(/** @type {string} */ _did, /** @type {string} */ cid) {
store.delete(cid);
},
...(streaming
? {
async getStream(
/** @type {string} */ _did,
/** @type {string} */ cid,
) {
const blob = store.get(cid);
if (!blob) return null;
return {
/** @type {ReadableStream<Uint8Array>} */
body: new ReadableStream({
start(controller) {
controller.enqueue(blob.data);
controller.close();
},
}),
mimeType: blob.mimeType,
};
},
}
: {}),
};
}
/**
* A PDS over a real single-record repository, with `blobCount` fake blobs
* registered in both blob metadata and the blob store.
*/
async function createBackupPds({ blobCount = 2, streaming = false } = {}) {
const { privateKey } = await generateKeyPair();
const { carBytes } = await buildRepositoryCar({
did: DID,
privateKey,
records: [
{
collection: 'app.bsky.feed.post',
rkey: '3abc123',
record: { $type: 'app.bsky.feed.post', text: 'backed up' },
},
],
});
const actorStorage = createMockStorage();
await loadRepositoryFromCar(carBytes, /** @type {any} */ (actorStorage));
const blobs = createMockBlobs({ streaming });
for (let i = 0; i < blobCount; i++) {
const cid = `bafkblob${i}`;
const data = new Uint8Array([i, i + 1, i + 2]);
blobs.store.set(cid, { data, mimeType: 'image/png' });
actorStorage.blobMeta.set(cid, {
mimeType: 'image/png',
size: data.byteLength,
createdAt: Date.now(),
});
}
const target = createMemoryTarget();
const pds = new PersonalDataServer({
actorStorage: /** @type {any} */ (actorStorage),
sharedStorage: /** @type {any} */ ({}),
blobs: /** @type {any} */ (blobs),
jwtSecret: 'test-secret',
backupTarget: target,
});
return { pds, backup: pds._backup, actorStorage, blobs, target };
}
describe('runBackup', () => {
it('writes a complete snapshot: repo CAR, blobs, manifest', async () => {
const { backup, target } = await createBackupPds();
const run = await backup.runBackup('manual');
expect(run.status).toBe('ok');
expect(run.blobsTotal).toBe(2);
expect(run.blobsCopied).toBe(2);
expect(run.carBytes).toBeGreaterThan(0);
expect(run.finishedAt).not.toBeNull();
const car = await target.get(`${run.snapshot}/repo.car`);
expect(car).not.toBeNull();
const { roots, blocks } = parseCarFile(/** @type {Uint8Array} */ (car));
expect(roots).toHaveLength(1);
expect(blocks.size).toBeGreaterThan(0);
const manifestBytes = await target.get(`${run.snapshot}/manifest.json`);
const manifest = JSON.parse(
new TextDecoder().decode(/** @type {Uint8Array} */ (manifestBytes)),
);
expect(manifest.did).toBe(DID);
expect(manifest.handle).toBe('backup.test');
expect(manifest.blobs.sort()).toEqual(['bafkblob0', 'bafkblob1']);
expect(manifest.commit.cid).toBe(roots[0]);
expect(manifest.keyIncluded).toBe(false);
expect(await target.get('blobs/bafkblob0')).toEqual(
new Uint8Array([0, 1, 2]),
);
});
it('skips blobs the target already holds', async () => {
const { backup } = await createBackupPds();
const first = await backup.runBackup('manual');
expect(first.blobsCopied).toBe(2);
const second = await backup.runBackup('manual');
expect(second.status).toBe('ok');
expect(second.blobsCopied).toBe(0);
expect(second.blobsTotal).toBe(2);
expect(second.id).not.toBe(first.id);
});
it('streams blobs when the store and the target both can', async () => {
const { backup, target } = await createBackupPds({ streaming: true });
const run = await backup.runBackup('manual');
expect(run.status).toBe('ok');
expect(run.blobsCopied).toBe(2);
expect(await target.get('blobs/bafkblob0')).toEqual(
new Uint8Array([0, 1, 2]),
);
// A streamed blob is counted by the bytes that pass, so the run's total
// is the CAR, the manifest and the two three-byte blobs.
const manifest = /** @type {Uint8Array} */ (
await target.get(`${run.snapshot}/manifest.json`)
);
expect(run.bytesUploaded).toBe(run.carBytes + manifest.byteLength + 6);
});
it('records a streamed blob the store has lost', async () => {
const { backup, blobs, target } = await createBackupPds({
streaming: true,
});
blobs.store.delete('bafkblob1');
const run = await backup.runBackup('manual');
expect(run.status).toBe('ok');
expect(run.blobsCopied).toBe(1);
const manifest = JSON.parse(
new TextDecoder().decode(
/** @type {Uint8Array} */ (
await target.get(`${run.snapshot}/manifest.json`)
),
),
);
expect(manifest.missingBlobs).toEqual(['bafkblob1']);
});
it('records blobs missing from the store without failing the run', async () => {
const { backup, blobs, target } = await createBackupPds();
blobs.store.delete('bafkblob1');
const run = await backup.runBackup('manual');
expect(run.status).toBe('ok');
expect(run.blobsCopied).toBe(1);
const manifest = JSON.parse(
new TextDecoder().decode(
/** @type {Uint8Array} */ (
await target.get(`${run.snapshot}/manifest.json`)
),
),
);
expect(manifest.missingBlobs).toEqual(['bafkblob1']);
});
it('finishes the snapshot when the target refuses one blob', async () => {
const { backup, target } = await createBackupPds();
const put = target.put.bind(target);
target.put = async (key, data, contentType) => {
if (key === 'blobs/bafkblob1') {
throw new Error('Backup target put blobs/bafkblob1 failed: HTTP 413');
}
return put(key, data, contentType);
};
const run = await backup.runBackup('manual');
// The repo, the other blob and the manifest are all in the snapshot.
expect(run.status).toBe('ok');
expect(run.blobsCopied).toBe(1);
expect(run.blobsFailed).toBe(1);
expect(run.blobError).toMatch(/HTTP 413/);
expect(run.error).toBeUndefined();
expect(await target.get(`${run.snapshot}/repo.car`)).not.toBeNull();
expect(await target.get('blobs/bafkblob0')).not.toBeNull();
const manifest = JSON.parse(
new TextDecoder().decode(
/** @type {Uint8Array} */ (
await target.get(`${run.snapshot}/manifest.json`)
),
),
);
expect(manifest.missingBlobs).toEqual(['bafkblob1']);
expect(manifest.blobErrors).toEqual([
{ cid: 'bafkblob1', error: expect.stringMatching(/HTTP 413/) },
]);
});
it('tries a refused blob again on the next run', async () => {
const { backup, target } = await createBackupPds();
const put = target.put.bind(target);
let refuse = true;
target.put = async (key, data, contentType) => {
if (refuse && key === 'blobs/bafkblob1') throw new Error('HTTP 413');
return put(key, data, contentType);
};
const first = await backup.runBackup('manual');
expect(first.blobsFailed).toBe(1);
refuse = false;
const second = await backup.runBackup('manual');
expect(second.status).toBe('ok');
expect(second.blobsFailed).toBe(0);
expect(second.blobsCopied).toBe(1);
expect(await target.get('blobs/bafkblob1')).toEqual(
new Uint8Array([1, 2, 3]),
);
});
it('fails the run when the target stops taking anything', async () => {
const { backup, target } = await createBackupPds();
target.put = async () => {
throw new Error('Backup target put failed: HTTP 403');
};
const run = await backup.runBackup('manual');
// Every blob failed, and so did the repo CAR the manifest depends on.
expect(run.status).toBe('error');
expect(run.error).toMatch(/HTTP 403/);
expect(await target.get(`${run.snapshot}/manifest.json`)).toBeNull();
});
it('keeps run history, newest first', async () => {
const { backup } = await createBackupPds();
await backup.runBackup('manual');
await backup.runBackup('scheduled');
const state = await backup.readBackupState();
expect(state.runs).toHaveLength(2);
expect(state.runs[0].trigger).toBe('scheduled');
expect(state.runs[1].trigger).toBe('manual');
});
});
describe('pruneBackups', () => {
it('drops old snapshots and reaps blobs nothing references', async () => {
const { backup, actorStorage, blobs, target } = await createBackupPds();
const state = defaultBackupState();
state.config.retain = 1;
await actorStorage.setBackupState(state);
const first = await backup.runBackup('manual');
// The account loses a blob; the next manifest no longer references it.
blobs.store.delete('bafkblob1');
actorStorage.blobMeta.delete('bafkblob1');
const second = await backup.runBackup('manual');
expect(await target.get(`${first.snapshot}/repo.car`)).toBeNull();
expect(await target.get(`${first.snapshot}/manifest.json`)).toBeNull();
expect(await target.get(`${second.snapshot}/repo.car`)).not.toBeNull();
// Referenced by the kept manifest: stays. Orphaned by the drop: reaped.
expect(await target.get('blobs/bafkblob0')).not.toBeNull();
expect(await target.get('blobs/bafkblob1')).toBeNull();
});
});
describe('heartbeat lock', () => {
/**
* A stored 'running' run whose heartbeat is `ageMs` in the past.
* @param {number} ageMs
*/
function runningRun(ageMs) {
const beat = new Date(Date.now() - ageMs).toISOString();
return {
id: 'stuck-run',
trigger: /** @type {const} */ ('manual'),
status: /** @type {const} */ ('running'),
startedAt: beat,
finishedAt: null,
heartbeatAt: beat,
blobsCopied: 1,
blobsTotal: 0,
blobsFailed: 0,
carBytes: 0,
bytesUploaded: 3,
snapshot: 'snapshots/stuck-run',
keyIncluded: false,
keyStale: false,
};
}
it('refuses to start beside a run that is still beating', async () => {
const { backup, actorStorage } = await createBackupPds();
const state = defaultBackupState();
state.runs = [runningRun(5_000)];
await actorStorage.setBackupState(state);
await expect(backup.runBackup('manual')).rejects.toThrow(/already running/);
expect(await backup.maybeRunScheduledBackup()).toBeNull();
});
it('reclaims the lock from a dead run and marks it interrupted', async () => {
const { backup, actorStorage } = await createBackupPds();
const state = defaultBackupState();
state.runs = [runningRun(10 * 60_000)];
await actorStorage.setBackupState(state);
const run = await backup.runBackup('manual');
expect(run.status).toBe('ok');
const after = await backup.readBackupState();
const stuck = after.runs.find((r) => r.id === 'stuck-run');
expect(stuck?.status).toBe('error');
expect(stuck?.error).toMatch(/Interrupted/);
expect(stuck?.finishedAt).toBe(stuck?.heartbeatAt);
});
it('presents a silent running run as interrupted without rewriting it', async () => {
const { backup, actorStorage } = await createBackupPds();
const state = defaultBackupState();
state.runs = [runningRun(10 * 60_000)];
await actorStorage.setBackupState(state);
const view = await backup.publicBackupView(await backup.readBackupState());
expect(view.runs[0].status).toBe('error');
expect(view.runs[0].error).toMatch(/Interrupted/);
// The stored row is untouched; only the next run's lock claim fixes it.
const stored = await backup.readBackupState();
expect(stored.runs[0].status).toBe('running');
});
it('heartbeats while copying', async () => {
const { backup } = await createBackupPds();
const run = await backup.runBackup('manual');
expect(run.heartbeatAt).toBeTruthy();
expect(Date.parse(/** @type {string} */ (run.heartbeatAt))).not.toBeNaN();
});
});
describe('scheduling', () => {
it('does nothing while disabled', async () => {
const { backup } = await createBackupPds();
expect(await backup.maybeRunScheduledBackup()).toBeNull();
});
it('runs once due and advances the schedule', async () => {
const { backup, actorStorage } = await createBackupPds();
const state = defaultBackupState();
state.config.enabled = true;
state.nextRunAt = Date.now() - 1000;
await actorStorage.setBackupState(state);
const run = await backup.maybeRunScheduledBackup();
expect(run?.status).toBe('ok');
expect(run?.trigger).toBe('scheduled');
const after = await backup.readBackupState();
expect(after.nextRunAt).toBeGreaterThan(Date.now());
// The advanced schedule means an immediate second tick does nothing.
expect(await backup.maybeRunScheduledBackup()).toBeNull();
});
});
describe('sealed key backup', () => {
it('seals from browser-derived material and opens with the passphrase', async () => {
// What the browser does: PBKDF2 the passphrase into the sealing key.
const passphrase = 'open sesame 123';
const salt = crypto.getRandomValues(new Uint8Array(16));
const iterations = 100_000;
const baseKey = await crypto.subtle.importKey(
'raw',
new TextEncoder().encode(passphrase),
'PBKDF2',
false,
['deriveBits'],
);
const bits = await crypto.subtle.deriveBits(
{ name: 'PBKDF2', hash: 'SHA-256', salt, iterations },
baseKey,
256,
);
const { sealKeyBackup } = await import('../packages/core/src/backup.js');
const keyBytes = crypto.getRandomValues(new Uint8Array(32));
const sealed = await sealKeyBackup(
{ derivedKey: new Uint8Array(bits), salt, iterations },
keyBytes,
);
expect(sealed.iterations).toBe(iterations);
expect(await decryptKeyBackup(passphrase, sealed)).toEqual(keyBytes);
});
it('round-trips the signing key through a snapshot', async () => {
const { backup, actorStorage, target } = await createBackupPds();
const keyBytes = crypto.getRandomValues(new Uint8Array(32));
await actorStorage.setPrivateKey(keyBytes);
const state = defaultBackupState();
state.config.keyBackup = await encryptKeyBackup(
'open sesame 123',
keyBytes,
);
await actorStorage.setBackupState(state);
const run = await backup.runBackup('manual');
expect(run.keyIncluded).toBe(true);
expect(run.keyStale).toBe(false);
const sealed = JSON.parse(
new TextDecoder().decode(
/** @type {Uint8Array} */ (await target.get(`${run.snapshot}/key.enc`)),
),
);
expect(await decryptKeyBackup('open sesame 123', sealed)).toEqual(keyBytes);
await expect(decryptKeyBackup('wrong passphrase', sealed)).rejects.toThrow(
/passphrase/,
);
});
it('flags a key backup made before the current key', async () => {
const { backup, actorStorage } = await createBackupPds();
const oldKey = crypto.getRandomValues(new Uint8Array(32));
await actorStorage.setPrivateKey(oldKey);
const state = defaultBackupState();
state.config.keyBackup = await encryptKeyBackup('open sesame 123', oldKey);
await actorStorage.setBackupState(state);
// The key rotates; the sealed copy is now of a dead key.
await actorStorage.setPrivateKey(
crypto.getRandomValues(new Uint8Array(32)),
);
const run = await backup.runBackup('manual');
expect(run.keyStale).toBe(true);
const view = await backup.publicBackupView(await backup.readBackupState());
expect(view.settings.keyBackup?.stale).toBe(true);
});
});
describe('availability', () => {
it('reports unavailable without a target', async () => {
const { actorStorage, blobs } = await createBackupPds();
const pds = new PersonalDataServer({
actorStorage: /** @type {any} */ (actorStorage),
sharedStorage: /** @type {any} */ ({}),
blobs: /** @type {any} */ (blobs),
jwtSecret: 'test-secret',
});
expect(pds._backup.backupsAvailable()).toBe(false);
await expect(pds._backup.runBackup('manual')).rejects.toThrow(/target/);
});
it('exposes the scheduler on the server for the platform tick', async () => {
const { pds } = await createBackupPds();
expect(await pds.maybeRunScheduledBackup()).toBeNull();
});
it('merges its routes into the server table', async () => {
const { pds } = await createBackupPds();
for (const [path, method] of [
['/account/api/backups', 'GET'],
['/account/api/backups/settings', 'POST'],
['/account/api/backups/run', 'POST'],
]) {
const route = pds.findRoute(path);
expect(route, path).toBeTruthy();
expect(route?.method, path).toBe(method);
expect(typeof route?.handler, path).toBe('function');
}
});
it('keeps space routes alongside them', async () => {
const { actorStorage, blobs, target } = await createBackupPds();
const pds = new PersonalDataServer({
actorStorage: /** @type {any} */ (actorStorage),
sharedStorage: /** @type {any} */ ({}),
blobs: /** @type {any} */ (blobs),
jwtSecret: 'test-secret',
backupTarget: target,
spaceStorage: /** @type {any} */ ({}),
extensions: [
() => ({
name: 'spaces',
routes: {
'/xrpc/com.atproto.space.getRepo': {
handler: async () => new Response('ok'),
},
},
}),
],
});
expect(pds.findRoute('/account/api/backups')).toBeTruthy();
expect(pds.findRoute('/xrpc/com.atproto.space.getRepo')).toBeTruthy();
expect(pds.spacesEnabled).toBe(true);
});
});