chadtmiller.com/pds.js
An atproto PDS in JavaScript, for Cloudflare Workers and Node
0
0
// The write guard's rule logic. Everything decidable from the records alone
// runs against synthetic records; the ancestry check runs against a real
// pack, both through the object index and through a whole-chain read. The
// push path around it is exercised end to end in space-remote.test.js.
import { execFile } from 'node:child_process';
import { randomBytes } from 'node:crypto';
import { mkdtempSync, readFileSync, rmSync, writeFileSync } from 'node:fs';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { promisify } from 'node:util';
import { afterAll, beforeAll, describe, expect, it } from 'vitest';
import { buildBundleIndex, encodePackIndex } from '../src/pack-index.js';
import { splitIntoChunks } from '../src/record.js';
import {
collaboratorsOf,
createGitWriteGuard,
GIT_CONFIG_COLLECTION,
linkedPackagesOf,
reviewersOf,
runnerOf,
runnersOf,
} from '../src/rules.js';
const SHA_A = 'a'.repeat(40);
const SHA_B = 'b'.repeat(40);
/** @param {Array<{name: string, sha: string}>} refs */
function repoRecord(refs) {
return {
$type: 'dev.pdsjs.git.repo',
name: 'repo',
refs,
bundles: [],
createdAt: '2026-08-13T00:00:00.000Z',
};
}
/** @param {unknown} config */
function guardWith(config) {
return createGitWriteGuard({
getRecord: async (_space, collection) =>
collection === GIT_CONFIG_COLLECTION ? config : null,
getBlob: async () => {
throw new Error('the test should not need pack data');
},
});
}
/** @param {unknown} prevValue @param {unknown} nextValue @param {unknown} [config] */
function check(prevValue, nextValue, config) {
return guardWith(config).check({
space: null,
collection: 'dev.pdsjs.git.repo',
rkey: 'repo',
prevValue,
nextValue,
});
}
describe('git write guard', () => {
const main = { name: 'refs/heads/main', sha: SHA_A };
const config = { protectedBranches: ['main'] };
it('passes with no config, and on record creation', async () => {
await expect(check(repoRecord([main]), repoRecord([]))).resolves.toBe(
undefined,
);
await expect(check(null, repoRecord([main]), config)).resolves.toBe(
undefined,
);
});
it('passes an unchanged protected branch without reading packs', async () => {
await expect(
check(
repoRecord([main]),
repoRecord([main, { name: 'refs/heads/dev', sha: SHA_B }]),
config,
),
).resolves.toBe(undefined);
});
it('refuses deleting a protected branch', async () => {
await expect(
check(repoRecord([main]), repoRecord([]), config),
).rejects.toThrow(/protected and cannot be deleted/);
});
it('refuses deleting a record holding a protected branch', async () => {
await expect(check(repoRecord([main]), null, config)).rejects.toThrow(
/remove the protection before deleting/,
);
await expect(
check(repoRecord([{ name: 'refs/heads/dev', sha: SHA_B }]), null, config),
).resolves.toBe(undefined);
});
it('ignores a protected name with no branch, and malformed prev records', async () => {
await expect(
check(repoRecord([]), repoRecord([main]), config),
).resolves.toBe(undefined);
await expect(
check({ junk: true }, repoRecord([main]), config),
).resolves.toBe(undefined);
});
it('validates config writes and lets a config deletion through', async () => {
const write = (/** @type {unknown} */ nextValue) =>
guardWith(null).check({
space: null,
collection: GIT_CONFIG_COLLECTION,
rkey: 'repo',
prevValue: null,
nextValue,
});
await expect(write({ protectedBranches: ['main'] })).resolves.toBe(
undefined,
);
await expect(write({})).resolves.toBe(undefined);
await expect(write(null)).resolves.toBe(undefined);
await expect(write({ protectedBranches: 'main' })).rejects.toThrow(
/list of branch names/,
);
await expect(write({ protectedBranches: ['bad name'] })).rejects.toThrow(
/list of branch names/,
);
await expect(write({ protectedBranches: ['*'] })).rejects.toThrow(
/list of branch names/,
);
});
});
describe('runnerOf', () => {
it('reads the DID a config names', () => {
expect(runnerOf({ runner: 'did:plc:runner' })).toBe('did:plc:runner');
});
it('reports no runner for a config without one', () => {
expect(runnerOf({ protectedBranches: ['main'] })).toBe(null);
expect(runnerOf(null)).toBe(null);
expect(runnerOf(undefined)).toBe(null);
});
it('refuses a value that is not a DID', () => {
expect(runnerOf({ runner: 'ci.example.com' })).toBe(null);
expect(runnerOf({ runner: 42 })).toBe(null);
expect(runnerOf({ runner: '' })).toBe(null);
});
});
describe('runnersOf', () => {
it('reads the list a config names', () => {
expect(runnersOf({ runners: ['did:plc:one', 'did:plc:two'] })).toEqual([
'did:plc:one',
'did:plc:two',
]);
});
it('reads the single runner where there is no list', () => {
expect(runnersOf({ runner: 'did:plc:runner' })).toEqual(['did:plc:runner']);
expect(runnersOf({})).toEqual([]);
expect(runnersOf(null)).toEqual([]);
});
it('prefers the list over the single field', () => {
expect(
runnersOf({ runner: 'did:plc:old', runners: ['did:plc:new'] }),
).toEqual(['did:plc:new']);
});
});
describe('collaboratorsOf and reviewersOf', () => {
it('reads the DIDs a config names', () => {
expect(
collaboratorsOf({ collaborators: ['did:plc:bob', 'did:plc:carol'] }),
).toEqual(['did:plc:bob', 'did:plc:carol']);
expect(reviewersOf({ reviewers: ['did:plc:nova'] })).toEqual([
'did:plc:nova',
]);
});
it('reports none for a config that names none', () => {
expect(collaboratorsOf({ runner: 'did:plc:runner' })).toEqual([]);
expect(collaboratorsOf(null)).toEqual([]);
expect(reviewersOf(undefined)).toEqual([]);
});
it('drops anything that is not a DID, and repeats', () => {
expect(
collaboratorsOf({
collaborators: [
'bob.example.com',
42,
'',
'did:plc:bob',
'did:plc:bob',
],
}),
).toEqual(['did:plc:bob']);
expect(collaboratorsOf({ collaborators: 'did:plc:bob' })).toEqual([]);
});
});
describe('linkedPackagesOf', () => {
it('reads the packages a config names', () => {
expect(
linkedPackagesOf({
packages: [
{ registry: 'npm', name: '@scope/lib' },
{ registry: 'oci', name: 'app/server' },
],
}),
).toEqual([
{ registry: 'npm', name: '@scope/lib' },
{ registry: 'oci', name: 'app/server' },
]);
});
it('reports no packages for a config without any', () => {
expect(linkedPackagesOf({ runner: 'did:plc:runner' })).toEqual([]);
expect(linkedPackagesOf(null)).toEqual([]);
expect(linkedPackagesOf({ packages: 'lib' })).toEqual([]);
});
it('drops an entry a reader has no words for', () => {
expect(
linkedPackagesOf({
packages: [
{ registry: 'cargo', name: 'lib' },
{ registry: 'npm', name: '' },
{ registry: 'npm', name: 42 },
{ name: 'lib' },
null,
{ registry: 'npm', name: 'lib' },
],
}),
).toEqual([{ registry: 'npm', name: 'lib' }]);
});
});
describe('the write guard on a config record', () => {
/** @param {unknown} nextValue */
const write = (nextValue) =>
guardWith(null).check({
space: null,
collection: GIT_CONFIG_COLLECTION,
rkey: 'repo',
prevValue: null,
nextValue,
});
it('accepts a package list', async () => {
await expect(
write({ packages: [{ registry: 'npm', name: 'lib' }] }),
).resolves.toBe(undefined);
});
// A registry added after this server was deployed stays writable.
it('accepts a registry this build has no words for', async () => {
await expect(
write({ packages: [{ registry: 'cargo', name: 'lib' }] }),
).resolves.toBe(undefined);
});
it('refuses a malformed package list', async () => {
await expect(write({ packages: 'lib' })).rejects.toThrow(
/{registry, name}/,
);
await expect(write({ packages: [{ name: 'lib' }] })).rejects.toThrow(
/{registry, name}/,
);
await expect(
write({ packages: [{ registry: 'npm', name: 'two words' }] }),
).rejects.toThrow(/{registry, name}/);
});
});
describe('ancestry over the object index', () => {
/** @type {string} */
let dir = '';
/** @type {string[]} */
const commits = [];
/** @type {Record<string, unknown>} */
let nextValue = {};
/** @type {Map<string, Uint8Array>} */
const blobs = new Map();
/** How many bytes each read handed back, by kind. */
let read = { whole: 0, ranged: 0 };
let bundleSize = 0;
const env = {
.../** @type {Record<string, string>} */ (process.env),
GIT_AUTHOR_NAME: 'Test',
GIT_AUTHOR_EMAIL: 'test@example.com',
GIT_COMMITTER_NAME: 'Test',
GIT_COMMITTER_EMAIL: 'test@example.com',
};
/** @param {string} cwd @param {string[]} args */
const runGit = async (cwd, args) =>
(await promisify(execFile)('git', args, { cwd, env })).stdout.trim();
beforeAll(async () => {
dir = mkdtempSync(join(tmpdir(), 'pdsjs-git-rules-'));
const repo = join(dir, 'repo');
await runGit(dir, ['init', '-b', 'main', repo]);
// A file body far larger than the commits, so a walk that reads the
// whole chain is plain to see in the byte counts.
writeFileSync(join(repo, 'big.bin'), randomBytes(512 * 1024));
for (const message of ['one', 'two', 'three']) {
writeFileSync(join(repo, 'log.txt'), `${message}\n`);
await runGit(repo, ['add', '.']);
await runGit(repo, ['commit', '-m', message]);
commits.push(await runGit(repo, ['rev-parse', 'HEAD']));
}
await runGit(repo, ['bundle', 'create', join(dir, 'b.bundle'), 'main']);
const bytes = new Uint8Array(readFileSync(join(dir, 'b.bundle')));
bundleSize = bytes.length;
/** @param {Uint8Array} data @param {string} cid */
const put = (data, cid) => {
blobs.set(cid, data);
return {
$type: 'blob',
ref: { $link: cid },
mimeType: 'application/octet-stream',
size: data.length,
};
};
const parts = splitIntoChunks(bytes, 64 * 1024).map((chunk, i) =>
put(chunk, `bafkreipart${i}`),
);
const encoded = encodePackIndex(await buildBundleIndex(bytes));
const indexParts = splitIntoChunks(encoded, 4096).map((chunk, i) =>
put(chunk, `bafkreiindex${i}`),
);
nextValue = {
$type: 'dev.pdsjs.git.repo',
name: 'repo',
refs: [{ name: 'refs/heads/main', sha: commits[2] }],
bundles: [{ heads: [commits[2]], parts, indexParts }],
createdAt: '2026-08-24T00:00:00.000Z',
};
}, 60_000);
afterAll(() => rmSync(dir, { recursive: true, force: true }));
/** @param {{ranges: boolean}} opts */
function guard(opts) {
read = { whole: 0, ranged: 0 };
return createGitWriteGuard({
getRecord: async (_space, collection) =>
collection === GIT_CONFIG_COLLECTION
? { protectedBranches: ['main'] }
: null,
getBlob: async (cid) => {
const blob = blobs.get(cid) ?? null;
if (blob) read.whole += blob.length;
return blob;
},
getBlobRange: opts.ranges
? async (cid, start, end) => {
const blob = blobs.get(cid);
if (!blob) return null;
const slice = blob.subarray(start, end);
read.ranged += slice.length;
return slice;
}
: undefined,
});
}
/** @param {string} prevSha @param {{ranges: boolean}} opts */
const check = (prevSha, opts) =>
guard(opts).check({
space: null,
collection: 'dev.pdsjs.git.repo',
rkey: 'repo',
prevValue: {
$type: 'dev.pdsjs.git.repo',
name: 'repo',
refs: [{ name: 'refs/heads/main', sha: prevSha }],
bundles: [],
createdAt: '2026-08-24T00:00:00.000Z',
},
nextValue,
});
it('takes a fast-forward without reading the file bodies', async () => {
await expect(check(commits[0], { ranges: true })).resolves.toBe(undefined);
// The index blobs are read whole; the pack is read by range, and the
// commits are a small part of it.
expect(read.ranged).toBeGreaterThan(0);
expect(read.ranged).toBeLessThan(bundleSize / 8);
expect(read.whole + read.ranged).toBeLessThan(bundleSize / 2);
});
it('refuses a tip the protected branch does not reach', async () => {
const orphan = 'f'.repeat(40);
await expect(check(orphan, { ranges: true })).rejects.toThrow(
/only fast-forward/,
);
});
it('agrees with the whole-chain read when no store reads ranges', async () => {
await expect(check(commits[0], { ranges: false })).resolves.toBe(undefined);
// Without ranges every chunk is downloaded, which is the cost the index
// path avoids.
expect(read.whole).toBeGreaterThanOrEqual(bundleSize);
const orphan = 'f'.repeat(40);
await expect(check(orphan, { ranges: false })).rejects.toThrow(
/only fast-forward/,
);
});
});