chadtmiller.com/pds.js
An atproto PDS in JavaScript, for Cloudflare Workers and Node
0
0
// Commit events must carry a diff since the previous repo state, not the
// entire MST. See com.atproto.sync.subscribeRepos #commit `blocks`.
import { PersonalDataServer } from '@pdsjs/core';
import { parseCarFile } from '@pdsjs/core/car';
import { mstInvert } from '@pdsjs/core/mst';
import {
cborDecode,
cborDecodeFirst,
cidToString,
getKeyDepth,
} from '@pdsjs/core/repo';
import { beforeEach, describe, expect, test } from 'vitest';
import {
createMemorySharedStorage,
createMemoryStorage,
} from './helpers/memory-storage.js';
const DID = 'did:plc:committestaaaaaaaaaaaaaa';
const PASSWORD = 'test-password';
/** Count the blocks carried in a stored commit event's CAR. */
function commitBlockCount(
/** @type {any} */ storage,
/** @type {number} */ index,
) {
const evt = cborDecode(storage._events[index].evt);
return parseCarFile(evt.blocks).blocks.size;
}
describe('commit event blocks', () => {
/** @type {ReturnType<typeof createMemoryStorage>} */
let storage;
/** @type {PersonalDataServer} */
let pds;
/** @type {string} */
let token;
beforeEach(async () => {
storage = createMemoryStorage();
pds = new PersonalDataServer({
actorStorage: /** @type {any} */ (storage),
sharedStorage: /** @type {any} */ (createMemorySharedStorage()),
blobs: {
async get() {
return null;
},
async put() {},
async delete() {},
},
jwtSecret: 'test-secret',
hostname: 'pds.example.com',
password: PASSWORD,
});
await pds.fetch(
new Request(`https://pds.example.com/init?did=${DID}`, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({
did: DID,
privateKey: Buffer.from(
crypto.getRandomValues(new Uint8Array(32)),
).toString('hex'),
handle: 'commit.test',
password: PASSWORD,
}),
}),
);
const sessionRes = await pds.fetch(
new Request(
'https://pds.example.com/xrpc/com.atproto.server.createSession',
{
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ identifier: DID, password: PASSWORD }),
},
),
);
const sessionBody = await sessionRes.json();
expect(sessionBody.accessJwt, 'session setup failed').toBeTruthy();
token = sessionBody.accessJwt;
});
/** @param {number} i */
async function createRecord(i) {
const res = await pds.fetch(
new Request(
'https://pds.example.com/xrpc/com.atproto.repo.createRecord',
{
method: 'POST',
headers: {
'Content-Type': 'application/json',
Authorization: `Bearer ${token}`,
},
body: JSON.stringify({
repo: DID,
collection: 'app.bsky.feed.post',
record: { text: `post ${i}`, createdAt: '2026-01-01T00:00:00Z' },
}),
},
),
);
expect(res.status).toBe(200);
}
test('carries only blocks the commit created, not the whole MST', async () => {
for (let i = 0; i < 120; i++) await createRecord(i);
const commitEvents = storage._events.filter((e) => e.commit_cid !== '');
expect(commitEvents.length).toBe(120);
// A single-record write carries the commit, the new record, the MST
// nodes along one root-to-key path, and the nodes beside that path that
// undoing the write reads. So every commit stays bounded by tree depth
// no matter how many records exist. Shipping the whole tree instead
// would grow with the record count (~38 blocks by the 120th commit here).
// Record keys are TIDs with a random component, so the largest commit
// varies between runs: 8 to 16 blocks over 100 runs.
const counts = commitEvents.map((evt) =>
commitBlockCount(storage, storage._events.indexOf(evt)),
);
expect(Math.max(...counts)).toBeLessThan(24);
});
test('includes the commit block and the newly written record', async () => {
await createRecord(0);
await createRecord(1);
const commitEvents = storage._events.filter((e) => e.commit_cid !== '');
const lastIdx = storage._events.indexOf(
commitEvents[commitEvents.length - 1],
);
const evt = cborDecode(storage._events[lastIdx].evt);
const { roots, blocks } = parseCarFile(evt.blocks);
// Commit block is the CAR root and is present
expect(roots.length).toBe(1);
expect(blocks.has(roots[0])).toBe(true);
// The record written by this commit travels with it
const recordCid = evt.ops[0].cid.$link;
expect(blocks.has(recordCid)).toBe(true);
});
// Sync 1.1 consumers invert each commit against prevData, which needs the
// CID an update or delete replaced. Without it jetstream's verifier fails
// the commit and never delivers the op.
describe('op prev', () => {
/**
* @param {string} method
* @param {object} body
*/
async function xrpc(method, body) {
const res = await pds.fetch(
new Request(`https://pds.example.com/xrpc/com.atproto.repo.${method}`, {
method: 'POST',
headers: {
'Content-Type': 'application/json',
Authorization: `Bearer ${token}`,
},
body: JSON.stringify({ repo: DID, ...body }),
}),
);
expect(res.status).toBe(200);
return res.json();
}
/** @param {string} title */
function putGallery(title) {
return xrpc('putRecord', {
collection: 'social.grain.gallery',
rkey: 'g1',
record: { title, createdAt: '2026-01-01T00:00:00Z' },
});
}
/** The ops of the most recent commit event, as stored. */
function lastOps() {
const commits = storage._events.filter((e) => e.commit_cid !== '');
return cborDecode(commits[commits.length - 1].evt).ops;
}
test('create carries no prev', async () => {
await putGallery('original');
expect(lastOps()[0]).not.toHaveProperty('prev');
});
test('update carries the CID it replaced', async () => {
const original = await putGallery('original');
const edited = await putGallery('edited');
const [op] = lastOps();
expect(op.action).toBe('update');
expect(op.cid.$link).toBe(edited.cid);
expect(op.prev.$link).toBe(original.cid);
});
test('delete carries the CID it removed', async () => {
const original = await putGallery('original');
await xrpc('deleteRecord', {
collection: 'social.grain.gallery',
rkey: 'g1',
});
const [op] = lastOps();
expect(op.action).toBe('delete');
expect(op.prev.$link).toBe(original.cid);
});
test('the firehose frame encodes prev as a CID link', async () => {
const original = await putGallery('original');
await putGallery('edited');
const commits = storage._events.filter((e) => e.commit_cid !== '');
const frame = pds._firehose.formatFirehoseEvent(
commits[commits.length - 1],
);
const [, headerLength] = cborDecodeFirst(frame);
const body = cborDecode(frame.subarray(headerLength));
expect(cidToString(body.ops[0].prev)).toBe(original.cid);
});
});
// Sync 1.1 consumers (relays, tap, jetstream) check each commit by undoing
// its ops on the new tree with only the blocks the commit carries, back to
// prevData. One they cannot check makes each of them fetch the whole repo
// again, all at once. Undoing a delete reads the unchanged nodes beside the
// key it put back, so a delete's commit carries those too.
describe('a consumer can check every commit from its own blocks', () => {
/**
* @param {string} method
* @param {object} body
*/
async function xrpc(method, body) {
const res = await pds.fetch(
new Request(`https://pds.example.com/xrpc/com.atproto.repo.${method}`, {
method: 'POST',
headers: {
'Content-Type': 'application/json',
Authorization: `Bearer ${token}`,
},
body: JSON.stringify({ repo: DID, ...body }),
}),
);
expect(res.status, await res.clone().text()).toBe(200);
return res.json();
}
const COLLECTION = 'dev.knotless.check';
/** Creates `n` records keyed `<prefix>0`..., in batches. */
async function seed(/** @type {string} */ prefix, /** @type {number} */ n) {
for (let at = 0; at < n; at += 100) {
const writes = [];
for (let i = at; i < Math.min(n, at + 100); i++) {
writes.push({
$type: 'com.atproto.repo.applyWrites#create',
collection: COLLECTION,
rkey: `${prefix}${String(i).padStart(5, '0')}`,
value: { status: 'running', n: i },
});
}
await xrpc('applyWrites', { writes });
}
}
/** The commit events since event `from`, decoded. */
function commitsSince(/** @type {number} */ from) {
return storage._events
.slice(from)
.filter((e) => e.commit_cid !== '')
.map((e) => cborDecode(e.evt));
}
/**
* Undoes `evt`'s ops with only its own blocks, as a consumer does: throws
* on a node the commit does not carry, and returns the root reached.
*/
async function undo(/** @type {any} */ evt) {
const { roots, blocks } = parseCarFile(evt.blocks);
const commit = cborDecode(
/** @type {Uint8Array} */ (blocks.get(roots[0])),
);
const ops = evt.ops.map((/** @type {any} */ op) => ({
action: op.action,
key: op.path,
prev: op.prev?.$link,
}));
const { root } = await mstInvert(
cidToString(commit.data),
ops,
async (cid) => blocks.get(cid) ?? null,
);
return root;
}
/** Asserts each of `evts` undoes to its prevData from its own blocks. */
async function checkable(/** @type {any[]} */ evts) {
expect(evts.length).toBeGreaterThan(0);
for (const evt of evts) {
const ops = evt.ops.map((/** @type {any} */ o) => o.action).join(',');
await expect(undo(evt), `${evt.rev}: ${ops}`).resolves.toBe(
evt.prevData,
);
}
}
/** The keys of `prefix`, with the deepest MST layer among them. */
async function deepest(/** @type {string[]} */ keys) {
let max = 0;
for (const key of keys) max = Math.max(max, await getKeyDepth(key));
return max;
}
test('a delete that merges the nodes beside its key', async () => {
await seed('a', 300);
const from = storage._events.length;
// Every key in turn, so the deletes include keys above the leaves,
// whose removal merges the subtrees on either side of them.
const keys = [];
for (let i = 0; i < 300; i += 7) {
const rkey = `a${String(i).padStart(5, '0')}`;
keys.push(`${COLLECTION}/${rkey}`);
await xrpc('deleteRecord', { collection: COLLECTION, rkey });
}
expect(await deepest(keys)).toBeGreaterThan(0);
await checkable(commitsSince(from));
});
test('many deletes in one write, as a compaction makes', async () => {
await seed('p', 400);
const from = storage._events.length;
/** @type {Array<{$type: string, collection: string, rkey: string, value?: object}>} */
const writes = [
{
$type: 'com.atproto.repo.applyWrites#create',
collection: COLLECTION,
rkey: 'p99999',
value: { status: 'success', n: -1 },
},
];
for (let i = 0; i < 47 * 8; i += 8) {
writes.push({
$type: 'com.atproto.repo.applyWrites#delete',
collection: COLLECTION,
rkey: `p${String(i).padStart(5, '0')}`,
});
}
await xrpc('applyWrites', { writes });
const [evt] = commitsSince(from);
expect(evt.ops.length).toBe(48);
await checkable([evt]);
});
test('a delete deep in a tree of several layers', async () => {
await seed('d', 2000);
const from = storage._events.length;
// The keys at the deepest layer there is, and one beside each.
const keys = [];
for (let i = 0; i < 2000; i++) {
keys.push(`${COLLECTION}/d${String(i).padStart(5, '0')}`);
}
const top = await deepest(keys);
expect(top).toBeGreaterThanOrEqual(2);
const chosen = [];
for (let i = 0; i < keys.length; i++) {
if ((await getKeyDepth(keys[i])) === top) chosen.push(i, i + 1);
}
for (const i of chosen.filter((i) => i < keys.length)) {
const rkey = keys[i].split('/')[1];
await xrpc('deleteRecord', { collection: COLLECTION, rkey });
}
await checkable(commitsSince(from));
});
test('creates and updates as well', async () => {
await seed('c', 200);
const from = storage._events.length;
for (let i = 0; i < 200; i += 9) {
await xrpc('putRecord', {
collection: COLLECTION,
rkey: `c${String(i).padStart(5, '0')}`,
record: { status: 'success', n: i },
});
await xrpc('putRecord', {
collection: COLLECTION,
rkey: `c${String(i).padStart(5, '0')}x`,
record: { status: 'running', n: i },
});
}
await checkable(commitsSince(from));
});
});
});