chadtmiller.com/pds.js
An atproto PDS in JavaScript, for Cloudflare Workers and Node
0
0
/**
* Writing check records into the runner's own repo.
*
* A run publishes twice: `running` when the steps start, then the outcome
* under the same record key with a compare-and-swap. A reader that sees only
* the first write knows the runner stopped mid-run.
*/
import { createTid } from '@pdsjs/core/repo';
import { XrpcError } from '@pdsjs/git';
import {
GIT_CHECK_COLLECTION,
GIT_LATEST_CHECK_COLLECTION,
} from './lexicon.js';
import { latestCheckKey } from './spec.js';
/** Log bytes kept on a check record. */
const DEFAULT_MAX_LOG_BYTES = 1024 * 1024;
/** Attempts at one PDS write before the failure reaches the runner. */
const WRITE_ATTEMPTS = 3;
/** Pause before a second attempt; the next pause doubles it. */
const DEFAULT_RETRY_DELAY_MS = 1000;
const TRUNCATION_NOTICE = '[earlier output dropped, log too large]\n';
/**
* Keep the tail of a log, which is where a failing step reports. The notice
* replaces what was dropped, so the blob never exceeds the cap.
* @param {string} text
* @param {number} maxBytes
* @returns {Uint8Array}
*/
export function truncateLogs(text, maxBytes) {
const bytes = new TextEncoder().encode(text);
if (bytes.length <= maxBytes) return bytes;
const notice = new TextEncoder().encode(TRUNCATION_NOTICE);
const tail = bytes.subarray(bytes.length - (maxBytes - notice.length));
const out = new Uint8Array(notice.length + tail.length);
out.set(notice, 0);
out.set(tail, notice.length);
return out;
}
/**
* The outcome a set of step results describes. Fewer results than steps means
* the executor stopped early, which it only does after a failure.
* @param {import('./executor.js').StepResult[]} results
* @returns {'success'|'failure'}
*/
export function checkStatus(results) {
return results.every((result) => result.exitCode === 0)
? 'success'
: 'failure';
}
/**
* @typedef {Object} CheckSubject
* @property {string} uri
* @property {string} cid
*/
/**
* @typedef {Object} LatestCheckEntry
* @property {string} workflow
* @property {CheckSubject} subject
* @property {CheckSubject} check
* @property {string} sha
* @property {string} status
* @property {string} startedAt
* @property {string} [finishedAt]
*/
/**
* Which of a repository's refs the runner last left mid-run.
*
* The check records are the runner's own memory of what it was doing, and
* they outlive its state file. A daemon that starts with no state adopts the
* refs it finds so it does not rerun the whole record; a ref whose latest
* check is still `running` is the one exception, because the run it belongs
* to was cut off and nothing else will finish it.
* @param {Object} ctx
* @param {import('@pdsjs/git').XrpcClient} ctx.client - the runner's PDS
* @param {string} ctx.did - the runner's own DID
* @returns {(repoDid: string, repoRkey: string, refs: string[]) => Promise<string[]>}
*/
export function createUnfinishedRefsReader(ctx) {
const { client, did } = ctx;
return async (repoDid, repoRkey, refs) => {
/** @type {string[]} */
const unfinished = [];
for (const ref of refs) {
const rkey = latestCheckKey(repoDid, repoRkey, ref);
// A repository the runner has never published for reads as absent,
// which is not a cut-off run.
const current = await client.getRecord(
did,
GIT_LATEST_CHECK_COLLECTION,
rkey,
);
const entries = latestCheckEntries(current?.value);
if (entries.some((entry) => entry.status === 'running')) {
unfinished.push(ref);
}
}
return unfinished;
};
}
/**
* The per-workflow entries of a latest-check record. A record written
* before entries existed carries one workflow's fields at the top level;
* it reads as one entry.
* @param {unknown} value
* @returns {LatestCheckEntry[]}
*/
export function latestCheckEntries(value) {
if (!value || typeof value !== 'object') return [];
const record = /** @type {Record<string, unknown>} */ (value);
if (Array.isArray(record.checks)) {
return /** @type {LatestCheckEntry[]} */ (record.checks);
}
if (typeof record.status !== 'string') return [];
const single = /** @type {LatestCheckEntry} */ ({
workflow: typeof record.workflow === 'string' ? record.workflow : 'ci',
subject: record.subject,
check: record.check,
sha: record.sha,
status: record.status,
startedAt: record.startedAt,
...(record.finishedAt ? { finishedAt: record.finishedAt } : {}),
});
return [single];
}
/**
* @typedef {Object} PublisherContext
* @property {import('@pdsjs/git').XrpcClient} client - authenticated as the runner
* @property {string} did - the runner's DID
* @property {number} [maxLogBytes]
* @property {number} [retryDelayMs]
*/
/** @typedef {ReturnType<typeof createPublisher>} Publisher */
/**
* @param {PublisherContext} ctx
*/
export function createPublisher(ctx) {
const { client, did } = ctx;
const maxLogBytes = ctx.maxLogBytes ?? DEFAULT_MAX_LOG_BYTES;
const retryDelayMs = ctx.retryDelayMs ?? DEFAULT_RETRY_DELAY_MS;
/**
* Run one PDS write, retrying a server error or a dropped connection.
* A client error names a permanent condition (a failed swap, a rejected
* record) and throws at once. Without the retries, one transient failure
* on the outcome write leaves the check `running` forever, and a reader
* cannot tell it from a runner that died mid-run.
* @template T
* @param {() => Promise<T>} write
* @returns {Promise<T>}
*/
async function withRetries(write) {
for (let attempt = 1; ; attempt++) {
try {
return await write();
} catch (err) {
const permanent =
err instanceof XrpcError && err.status > 0 && err.status < 500;
if (permanent || attempt === WRITE_ATTEMPTS) throw err;
await new Promise((resolve) =>
setTimeout(resolve, retryDelayMs * attempt),
);
}
}
}
/**
* Update this workflow's entry in the repository's latest-check record.
*
* The record holds one entry per workflow, so a run reads the record,
* replaces its own entry and writes the rest back unchanged. The swap
* condition guards the read-modify-write; a lost swap re-reads whoever
* won. One daemon runs serially, so the loop settles at once in practice.
* @param {Object} run
* @param {string} run.repoDid
* @param {string} run.repoName - what the repository is called
* @param {string} run.repoRkey - the repository's record key
* @param {CheckSubject} run.subject
* @param {string} run.ref
* @param {string} run.sha
* @param {string} run.workflow
* @param {string} run.startedAt
* @param {CheckSubject} check
* @param {'running'|'success'|'failure'|'error'} status
* @param {string} [finishedAt]
* @returns {Promise<void>}
*/
async function writeLatest(run, check, status, finishedAt) {
const rkey = latestCheckKey(run.repoDid, run.repoRkey, run.ref);
const entry = {
workflow: run.workflow,
subject: run.subject,
check,
sha: run.sha,
status,
startedAt: run.startedAt,
...(finishedAt ? { finishedAt } : {}),
};
for (let attempt = 1; ; attempt++) {
const current = await withRetries(() =>
client.getRecord(did, GIT_LATEST_CHECK_COLLECTION, rkey),
);
const others = latestCheckEntries(current?.value).filter(
(existing) => existing.workflow !== run.workflow,
);
try {
await withRetries(() =>
client.putRecord(
did,
GIT_LATEST_CHECK_COLLECTION,
rkey,
{
$type: GIT_LATEST_CHECK_COLLECTION,
ref: run.ref,
checks: [...others, entry],
},
current ? current.cid : null,
),
);
return;
} catch (err) {
const lostSwap =
err instanceof XrpcError && err.error === 'InvalidSwap';
if (!lostSwap || attempt === WRITE_ATTEMPTS) throw err;
}
}
}
return {
/**
* @param {Object} run
* @param {string} run.repoDid
* @param {string} run.repoName - what the repository is called
* @param {string} run.repoRkey - the repository's record key
* @param {CheckSubject} run.subject
* @param {string} run.ref
* @param {string} run.sha
* @param {string} run.workflow
* @param {string} run.startedAt
* @returns {Promise<{rkey: string, cid: string}>}
*/
async start(run) {
const rkey = createTid();
const { uri, cid } = await withRetries(() =>
client.putRecord(
did,
GIT_CHECK_COLLECTION,
rkey,
{
$type: GIT_CHECK_COLLECTION,
subject: run.subject,
ref: run.ref,
sha: run.sha,
workflow: run.workflow,
status: 'running',
startedAt: run.startedAt,
},
null,
),
);
await writeLatest(run, { uri, cid }, 'running');
return { rkey, cid };
},
/**
* @param {Object} run
* @param {string} run.rkey
* @param {string} run.cid - the record version this replaces
* @param {string} run.repoDid
* @param {string} run.repoName - what the repository is called
* @param {string} run.repoRkey - the repository's record key
* @param {CheckSubject} run.subject
* @param {string} run.ref
* @param {string} run.sha
* @param {string} run.workflow
* @param {'success'|'failure'|'error'} run.status
* @param {import('./executor.js').StepResult[]} run.steps
* @param {string} run.startedAt
* @param {string} run.logs
* @returns {Promise<string>} the record uri
*/
async finish(run) {
const bytes = truncateLogs(run.logs, maxLogBytes);
const logs = await withRetries(() =>
client.uploadBlob(bytes, 'text/plain'),
);
const finishedAt = new Date().toISOString();
const { uri, cid } = await withRetries(() =>
client.putRecord(
did,
GIT_CHECK_COLLECTION,
run.rkey,
{
$type: GIT_CHECK_COLLECTION,
subject: run.subject,
ref: run.ref,
sha: run.sha,
workflow: run.workflow,
status: run.status,
steps: run.steps,
logs,
startedAt: run.startedAt,
finishedAt,
},
run.cid,
),
);
await writeLatest(run, { uri, cid }, run.status, finishedAt);
return uri;
},
};
}