chadtmiller.com/pds.js
An atproto PDS in JavaScript, for Cloudflare Workers and Node
0
0
import { XrpcError } from '@pdsjs/git';
import { describe, expect, it } from 'vitest';
import {
GIT_CHECK_COLLECTION,
GIT_LATEST_CHECK_COLLECTION,
} from '../src/lexicon.js';
import {
checkStatus,
createPublisher,
createUnfinishedRefsReader,
truncateLogs,
} from '../src/publish.js';
const SUBJECT = {
uri: 'at://did:plc:abc/dev.pdsjs.git.repo/proj',
cid: 'bafyreiaaa',
};
function fakeClient() {
/** @type {Array<Record<string, *>>} */
const writes = [];
/** @type {Array<{bytes: Uint8Array, mimeType: string}>} */
const blobs = [];
/** @type {Map<string, string>} */
const stored = new Map();
/** @type {Map<string, unknown>} */
const records = new Map();
const client = {
/**
* @param {string} repo
* @param {string} collection
* @param {string} rkey
*/
async getRecord(repo, collection, rkey) {
const uri = `at://${repo}/${collection}/${rkey}`;
if (!stored.has(uri)) return null;
return { uri, cid: stored.get(uri), value: records.get(uri) };
},
/**
* Enforces the server's swap rule: a cid must match what is stored, null
* requires that nothing is stored, and undefined states no condition.
* @param {string} repo
* @param {string} collection
* @param {string} rkey
* @param {Record<string, *>} record
* @param {string|null} [swapRecord]
*/
async putRecord(repo, collection, rkey, record, swapRecord) {
const uri = `at://${repo}/${collection}/${rkey}`;
if (
swapRecord !== undefined &&
swapRecord !== (stored.get(uri) ?? null)
) {
throw new XrpcError(400, 'InvalidSwap', 'cid mismatch');
}
writes.push({ repo, collection, rkey, record, swapRecord });
const cid = `cid-${writes.length}`;
stored.set(uri, cid);
records.set(uri, record);
return { uri, cid };
},
/**
* @param {Uint8Array} bytes
* @param {string} mimeType
*/
async uploadBlob(bytes, mimeType) {
blobs.push({ bytes, mimeType });
return { $type: 'blob', ref: { $link: 'bafylogs' }, size: bytes.length };
},
};
/** Seed a stored record, as a runner from before this build left it. */
const seed = (
/** @type {string} */ uri,
/** @type {unknown} */ value,
/** @type {string} */ cid,
) => {
stored.set(uri, cid);
records.set(uri, value);
};
return { client, writes, blobs, seed };
}
describe('truncateLogs', () => {
it('leaves a log under the cap alone', () => {
const bytes = truncateLogs('short', 1000);
expect(new TextDecoder().decode(bytes)).toBe('short');
});
it('keeps the tail and stays within the cap', () => {
const bytes = truncateLogs(`${'a'.repeat(500)}TAIL`, 100);
const text = new TextDecoder().decode(bytes);
expect(bytes.length).toBeLessThanOrEqual(100);
expect(text).toContain('TAIL');
expect(text.startsWith('[earlier output dropped')).toBe(true);
});
});
describe('checkStatus', () => {
it('succeeds when every step exited zero', () => {
expect(
checkStatus([
{ name: 'a', exitCode: 0, durationMs: 1 },
{ name: 'b', exitCode: 0, durationMs: 1 },
]),
).toBe('success');
});
it('fails when a step exited non-zero', () => {
expect(checkStatus([{ name: 'a', exitCode: 2, durationMs: 1 }])).toBe(
'failure',
);
});
});
const RUN = {
repoDid: 'did:plc:abc',
repoName: 'proj',
repoRkey: '3lprojkey',
subject: SUBJECT,
ref: 'refs/heads/main',
sha: 'a'.repeat(40),
workflow: 'ci',
startedAt: '2026-08-18T00:00:00.000Z',
};
/** @param {ReturnType<typeof fakeClient>} fake */
function publisherFor(fake) {
return createPublisher({
client: /** @type {any} */ (fake.client),
did: 'did:plc:runner',
retryDelayMs: 0,
});
}
/**
* Make the client's putRecord fail before each success.
* @param {ReturnType<typeof fakeClient>} fake
* @param {() => Error} makeError
* @param {number} failures
* @returns {() => number} how many calls happened
*/
function failPutRecord(fake, makeError, failures) {
const succeed = fake.client.putRecord.bind(fake.client);
let calls = 0;
fake.client.putRecord = async (...args) => {
calls += 1;
if (calls <= failures) throw makeError();
return succeed(...args);
};
return () => calls;
}
/** @param {Array<Record<string, *>>} writes @param {string} collection */
const writesTo = (writes, collection) =>
writes.filter((write) => write.collection === collection);
describe('createPublisher', () => {
it('writes a running record that no record may replace', async () => {
const fake = fakeClient();
const { rkey, cid } = await publisherFor(fake).start(RUN);
const check = writesTo(fake.writes, GIT_CHECK_COLLECTION)[0];
expect(check.repo).toBe('did:plc:runner');
expect(check.swapRecord).toBe(null);
expect(check.record.status).toBe('running');
expect(check.record.subject).toEqual(SUBJECT);
expect(rkey).toHaveLength(13);
expect(cid).toBe('cid-1');
});
it('swaps the running record for the outcome, with logs as text', async () => {
const fake = fakeClient();
const publisher = publisherFor(fake);
const { rkey, cid } = await publisher.start(RUN);
const uri = await publisher.finish({
...RUN,
rkey,
cid,
status: 'failure',
steps: [{ name: 'test', exitCode: 1, durationMs: 5 }],
logs: 'output here',
});
expect(fake.blobs[0].mimeType).toBe('text/plain');
expect(new TextDecoder().decode(fake.blobs[0].bytes)).toBe('output here');
const outcome = writesTo(fake.writes, GIT_CHECK_COLLECTION)[1];
expect(outcome.swapRecord).toBe(cid);
expect(outcome.rkey).toBe(rkey);
expect(outcome.record.status).toBe('failure');
expect(outcome.record.steps).toHaveLength(1);
expect(outcome.record.finishedAt).toBeTypeOf('string');
expect(uri).toBe(`at://did:plc:runner/${GIT_CHECK_COLLECTION}/${rkey}`);
});
it('points the latest-check record at the running check', async () => {
const fake = fakeClient();
const { rkey } = await publisherFor(fake).start(RUN);
const latest = writesTo(fake.writes, GIT_LATEST_CHECK_COLLECTION)[0];
expect(latest.rkey).toBe('did:plc:abc:3lprojkey:refs_heads_main');
// null: this repository has no pointer yet, and the swap says so.
expect(latest.swapRecord).toBe(null);
const entry = latest.record.checks[0];
expect(entry.workflow).toBe('ci');
expect(entry.status).toBe('running');
expect(entry.finishedAt).toBeUndefined();
expect(entry.check).toEqual({
uri: `at://did:plc:runner/${GIT_CHECK_COLLECTION}/${rkey}`,
cid: 'cid-1',
});
expect(entry.subject).toEqual(SUBJECT);
});
it('replaces its own running pointer with the outcome', async () => {
const fake = fakeClient();
const publisher = publisherFor(fake);
const { rkey, cid } = await publisher.start(RUN);
await publisher.finish({
...RUN,
rkey,
cid,
status: 'success',
steps: [],
logs: '',
});
const pointers = writesTo(fake.writes, GIT_LATEST_CHECK_COLLECTION);
expect(pointers).toHaveLength(2);
expect(pointers[1].record.checks).toHaveLength(1);
expect(pointers[1].record.checks[0].status).toBe('success');
expect(pointers[1].record.checks[0].finishedAt).toBeTypeOf('string');
// The swap names the version start() wrote, so the read-modify-write
// loses to a concurrent writer instead of overwriting it.
expect(pointers[1].swapRecord).toBe('cid-2');
});
it('publishes a second check over the first', async () => {
const fake = fakeClient();
const publisher = publisherFor(fake);
for (const status of ['success', 'failure']) {
const { rkey, cid } = await publisher.start(RUN);
await publisher.finish({
...RUN,
rkey,
cid,
status: /** @type {'success'|'failure'} */ (status),
steps: [],
logs: '',
});
}
const pointers = writesTo(fake.writes, GIT_LATEST_CHECK_COLLECTION);
expect(pointers).toHaveLength(4);
expect(pointers[3].record.checks).toHaveLength(1);
expect(pointers[3].record.checks[0].status).toBe('failure');
});
it('keeps one pointer entry per workflow', async () => {
const fake = fakeClient();
const publisher = publisherFor(fake);
const first = await publisher.start(RUN);
await publisher.finish({
...RUN,
...first,
status: 'success',
steps: [],
logs: '',
});
await publisher.start({ ...RUN, workflow: 'mirror' });
const pointers = writesTo(fake.writes, GIT_LATEST_CHECK_COLLECTION);
const latest = pointers[pointers.length - 1].record;
expect(latest.checks.map((/** @type {*} */ c) => c.workflow)).toEqual([
'ci',
'mirror',
]);
expect(latest.checks[0].status).toBe('success');
expect(latest.checks[1].status).toBe('running');
});
it('reads a pointer from before entries existed as one entry', async () => {
const fake = fakeClient();
fake.seed(
`at://did:plc:runner/${GIT_LATEST_CHECK_COLLECTION}/did:plc:abc:3lprojkey:refs_heads_main`,
{
$type: GIT_LATEST_CHECK_COLLECTION,
subject: SUBJECT,
check: { uri: 'at://old', cid: 'old-cid' },
ref: RUN.ref,
sha: RUN.sha,
workflow: 'ci',
status: 'success',
startedAt: RUN.startedAt,
finishedAt: RUN.startedAt,
},
'cid-old',
);
await publisherFor(fake).start({ ...RUN, workflow: 'mirror' });
const latest = writesTo(fake.writes, GIT_LATEST_CHECK_COLLECTION)[0];
expect(latest.swapRecord).toBe('cid-old');
expect(
latest.record.checks.map((/** @type {*} */ c) => c.workflow),
).toEqual(['ci', 'mirror']);
expect(latest.record.checks[0].check.uri).toBe('at://old');
});
it('retries a write past a server error', async () => {
const fake = fakeClient();
const calls = failPutRecord(
fake,
() => new XrpcError(500, 'InternalServerError', 'boom'),
2,
);
const { cid } = await publisherFor(fake).start(RUN);
expect(cid).toBe('cid-1');
// Two failures, one success, then the untouched latest-check write.
expect(calls()).toBe(4);
});
it('gives the error to the caller when every attempt fails', async () => {
const fake = fakeClient();
const calls = failPutRecord(
fake,
() => new XrpcError(502, 'BadGateway', 'down'),
Infinity,
);
await expect(publisherFor(fake).start(RUN)).rejects.toThrow('down');
expect(calls()).toBe(3);
});
it('does not retry a client error', async () => {
const fake = fakeClient();
const calls = failPutRecord(
fake,
() => new XrpcError(400, 'InvalidSwap', 'cid mismatch'),
Infinity,
);
await expect(publisherFor(fake).start(RUN)).rejects.toThrow('cid mismatch');
expect(calls()).toBe(1);
});
it('retries a dropped connection, which throws no XrpcError', async () => {
const fake = fakeClient();
const calls = failPutRecord(fake, () => new TypeError('fetch failed'), 1);
const { cid } = await publisherFor(fake).start(RUN);
expect(cid).toBe('cid-1');
expect(calls()).toBe(3);
});
});
describe('createUnfinishedRefsReader', () => {
const RUNNER = 'did:plc:runner';
const OWNER = 'did:plc:abc';
/**
* @param {ReturnType<typeof fakeClient>} fake
* @param {string} ref
* @param {Array<{workflow: string, status: string}>} checks
*/
async function seed(fake, ref, checks) {
await fake.client.putRecord(
RUNNER,
GIT_LATEST_CHECK_COLLECTION,
`${OWNER}:proj:${ref.replace(/[^a-zA-Z0-9.:_~-]/g, '_')}`,
{ $type: GIT_LATEST_CHECK_COLLECTION, ref, checks },
null,
);
}
it('names only the refs an entry is still running for', async () => {
const fake = fakeClient();
await seed(fake, 'refs/heads/main', [
{ workflow: 'ci', status: 'success' },
{ workflow: 'publish-wizard', status: 'running' },
]);
await seed(fake, 'refs/tags/v1.0.0', [
{ workflow: 'ci', status: 'success' },
]);
const unfinished = createUnfinishedRefsReader({
client: /** @type {*} */ (fake.client),
did: RUNNER,
});
expect(
await unfinished(OWNER, 'proj', [
'refs/heads/main',
'refs/tags/v1.0.0',
// Never published for, so the record is absent rather than running.
'refs/heads/topic',
]),
).toEqual(['refs/heads/main']);
});
});