mirror of
https://github.com/whyour/qinglong.git
synced 2026-09-20 16:07:11 +08:00
535 lines
16 KiB
TypeScript
535 lines
16 KiB
TypeScript
import { constants } from 'fs';
|
|
import fs from 'fs/promises';
|
|
import path from 'path';
|
|
import { randomBytes } from 'crypto';
|
|
import { lock } from 'proper-lockfile';
|
|
import {
|
|
MAX_WORKER_EXECUTION_OFFER_JOURNAL_ENTRIES,
|
|
MAX_WORKER_EXECUTION_OFFER_JOURNAL_PAGE_SIZE,
|
|
MAX_WORKER_EXECUTION_OFFER_RECORD_BYTES,
|
|
cloneWorkerExecutionOfferJournalRecord,
|
|
parseWorkerExecutionOfferJournalRecord,
|
|
serializeWorkerExecutionOfferJournalRecord,
|
|
type WorkerExecutionOfferJournalRecord,
|
|
} from '../../domain/workerExecutionOffer';
|
|
import { assertRunDispatchOfferId } from '../../domain/runDispatchOffer';
|
|
import type {
|
|
WorkerExecutionOfferJournal,
|
|
WorkerExecutionOfferJournalCreateResult,
|
|
WorkerExecutionOfferJournalPage,
|
|
} from '../../ports/workerExecutionOfferJournal';
|
|
import type {
|
|
WorkerExecutionOfferJournalOwnership,
|
|
WorkerExecutionOfferJournalOwnershipState,
|
|
} from '../../ports/workerExecutionOfferJournalOwnership';
|
|
|
|
const JOURNAL_FILE_PATTERN = /^([0-9a-f]{64})\.json$/;
|
|
|
|
export const MIN_WORKER_OFFER_JOURNAL_LOCK_STALE_MS = 5_000;
|
|
export const MAX_WORKER_OFFER_JOURNAL_LOCK_STALE_MS = 5 * 60_000;
|
|
|
|
export interface WorkerExecutionOfferLockProvider {
|
|
acquire(options: {
|
|
root: string;
|
|
lockfilePath: string;
|
|
staleMs: number;
|
|
updateMs: number;
|
|
onCompromised(error: Error): void;
|
|
}): Promise<() => Promise<void>>;
|
|
}
|
|
|
|
const properLockProvider: WorkerExecutionOfferLockProvider = {
|
|
acquire(options) {
|
|
return lock(options.root, {
|
|
stale: options.staleMs,
|
|
update: options.updateMs,
|
|
retries: 0,
|
|
realpath: true,
|
|
lockfilePath: options.lockfilePath,
|
|
onCompromised: options.onCompromised,
|
|
});
|
|
},
|
|
};
|
|
|
|
export class WorkerExecutionOfferJournalCapacityError extends Error {
|
|
constructor(readonly maximumEntries: number) {
|
|
super(`Worker execution offer journal reached ${maximumEntries} entries`);
|
|
this.name = 'WorkerExecutionOfferJournalCapacityError';
|
|
}
|
|
}
|
|
|
|
export class WorkerExecutionOfferJournalRevisionError extends Error {
|
|
constructor(readonly offerId: string) {
|
|
super(`Worker execution offer journal revision changed for ${offerId}`);
|
|
this.name = 'WorkerExecutionOfferJournalRevisionError';
|
|
}
|
|
}
|
|
|
|
export class WorkerExecutionOfferJournalNotFoundError extends Error {
|
|
constructor(readonly offerId: string) {
|
|
super(`Worker execution offer journal entry ${offerId} was not found`);
|
|
this.name = 'WorkerExecutionOfferJournalNotFoundError';
|
|
}
|
|
}
|
|
|
|
export class WorkerExecutionOfferJournalOwnershipError extends Error {
|
|
constructor(
|
|
readonly reason: 'already_owned' | 'not_owned' | 'compromised',
|
|
readonly cause?: unknown,
|
|
) {
|
|
super(`Worker execution offer journal ownership failed: ${reason}`);
|
|
this.name = 'WorkerExecutionOfferJournalOwnershipError';
|
|
}
|
|
}
|
|
|
|
function isCode(error: unknown, code: string): boolean {
|
|
return (
|
|
typeof error === 'object' &&
|
|
error !== null &&
|
|
'code' in error &&
|
|
(error as { code?: string }).code === code
|
|
);
|
|
}
|
|
|
|
function assertIntegerBetween(
|
|
name: string,
|
|
value: number,
|
|
minimum: number,
|
|
maximum: number,
|
|
): void {
|
|
if (!Number.isSafeInteger(value) || value < minimum || value > maximum) {
|
|
throw new RangeError(`${name} must be between ${minimum} and ${maximum}`);
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Worker-local restart journal. One bounded file per offer avoids a database or
|
|
* sidecar on edge devices. The Worker runtime must exclusively own this root.
|
|
*/
|
|
export class WorkerExecutionOfferFileJournal
|
|
implements WorkerExecutionOfferJournal, WorkerExecutionOfferJournalOwnership
|
|
{
|
|
private readonly maximumEntries: number;
|
|
private readonly ownershipStaleMs: number;
|
|
private readonly lockProvider: WorkerExecutionOfferLockProvider;
|
|
private readonly onOwnershipCompromised?: (error: Error) => void;
|
|
private ownerState: WorkerExecutionOfferJournalOwnershipState = 'unowned';
|
|
private releaseOwner?: () => Promise<void>;
|
|
private ownershipError?: Error;
|
|
private mutationTail: Promise<void> = Promise.resolve();
|
|
|
|
constructor(
|
|
private readonly root: string,
|
|
options: {
|
|
maximumEntries?: number;
|
|
ownershipStaleMs?: number;
|
|
lockProvider?: WorkerExecutionOfferLockProvider;
|
|
onOwnershipCompromised?: (error: Error) => void;
|
|
} = {},
|
|
) {
|
|
if (!path.isAbsolute(root)) {
|
|
throw new RangeError(
|
|
'Worker execution offer journal root must be absolute',
|
|
);
|
|
}
|
|
this.maximumEntries = options.maximumEntries ?? 64;
|
|
assertIntegerBetween(
|
|
'maximumEntries',
|
|
this.maximumEntries,
|
|
1,
|
|
MAX_WORKER_EXECUTION_OFFER_JOURNAL_ENTRIES,
|
|
);
|
|
this.ownershipStaleMs = options.ownershipStaleMs ?? 30_000;
|
|
assertIntegerBetween(
|
|
'ownershipStaleMs',
|
|
this.ownershipStaleMs,
|
|
MIN_WORKER_OFFER_JOURNAL_LOCK_STALE_MS,
|
|
MAX_WORKER_OFFER_JOURNAL_LOCK_STALE_MS,
|
|
);
|
|
this.lockProvider = options.lockProvider ?? properLockProvider;
|
|
this.onOwnershipCompromised = options.onOwnershipCompromised;
|
|
}
|
|
|
|
ownershipState(): WorkerExecutionOfferJournalOwnershipState {
|
|
return this.ownerState;
|
|
}
|
|
|
|
async acquireOwnership(): Promise<'acquired' | 'already_owned'> {
|
|
if (this.ownerState === 'owned') return 'already_owned';
|
|
if (this.ownerState === 'releasing') {
|
|
throw new WorkerExecutionOfferJournalOwnershipError('not_owned');
|
|
}
|
|
if (this.ownerState === 'compromised') {
|
|
throw new WorkerExecutionOfferJournalOwnershipError(
|
|
'compromised',
|
|
this.ownershipError,
|
|
);
|
|
}
|
|
await this.ensureRoot();
|
|
try {
|
|
const release = await this.lockProvider.acquire({
|
|
root: this.root,
|
|
lockfilePath: path.join(this.root, '.owner.lock'),
|
|
staleMs: this.ownershipStaleMs,
|
|
updateMs: Math.max(1_000, Math.floor(this.ownershipStaleMs / 2)),
|
|
onCompromised: (error) => this.compromiseOwnership(error),
|
|
});
|
|
this.releaseOwner = release;
|
|
this.ownerState = 'owned';
|
|
return 'acquired';
|
|
} catch (error) {
|
|
if (isCode(error, 'ELOCKED')) {
|
|
throw new WorkerExecutionOfferJournalOwnershipError(
|
|
'already_owned',
|
|
error,
|
|
);
|
|
}
|
|
throw error;
|
|
}
|
|
}
|
|
|
|
async releaseOwnership(): Promise<'released' | 'not_owned' | 'compromised'> {
|
|
if (this.ownerState === 'unowned') return 'not_owned';
|
|
if (this.ownerState === 'compromised') return 'compromised';
|
|
if (this.ownerState === 'releasing') return 'not_owned';
|
|
while (true) {
|
|
const pending = this.mutationTail;
|
|
await pending;
|
|
if (pending === this.mutationTail) break;
|
|
}
|
|
const ownerStateAfterMutations = this.ownershipState();
|
|
if (ownerStateAfterMutations === 'compromised') return 'compromised';
|
|
if (ownerStateAfterMutations !== 'owned') return 'not_owned';
|
|
const release = this.releaseOwner;
|
|
if (!release) {
|
|
this.compromiseOwnership(
|
|
new Error('Worker offer journal owner release capability is missing'),
|
|
);
|
|
return 'compromised';
|
|
}
|
|
this.ownerState = 'releasing';
|
|
try {
|
|
await release();
|
|
this.releaseOwner = undefined;
|
|
this.ownerState = 'unowned';
|
|
return 'released';
|
|
} catch (error) {
|
|
const compromised =
|
|
error instanceof Error
|
|
? error
|
|
: new Error('Worker offer journal owner release failed');
|
|
this.compromiseOwnership(compromised);
|
|
throw new WorkerExecutionOfferJournalOwnershipError('compromised', error);
|
|
}
|
|
}
|
|
|
|
async create(
|
|
record: WorkerExecutionOfferJournalRecord,
|
|
): Promise<WorkerExecutionOfferJournalCreateResult> {
|
|
this.assertOwned();
|
|
const candidate = cloneWorkerExecutionOfferJournalRecord(record);
|
|
if (candidate.revision !== 0 || candidate.state !== 'accepted') {
|
|
throw new TypeError(
|
|
'A new Worker offer journal entry must be accepted at revision zero',
|
|
);
|
|
}
|
|
return this.serializeMutation(async () => {
|
|
const target = this.target(candidate.offer.offerId);
|
|
if (await this.exists(target)) return 'exists';
|
|
const names = await this.entryNames();
|
|
if (names.length >= this.maximumEntries) {
|
|
throw new WorkerExecutionOfferJournalCapacityError(this.maximumEntries);
|
|
}
|
|
const temporary = this.temporary(candidate.offer.offerId);
|
|
try {
|
|
await this.writeTemporary(temporary, candidate);
|
|
this.assertOwned();
|
|
await fs.link(temporary, target);
|
|
await this.bestEffortSyncDirectory();
|
|
return 'created';
|
|
} catch (error) {
|
|
if (isCode(error, 'EEXIST')) return 'exists';
|
|
throw error;
|
|
} finally {
|
|
await this.bestEffortUnlink(temporary);
|
|
}
|
|
});
|
|
}
|
|
|
|
async read(
|
|
offerId: string,
|
|
): Promise<WorkerExecutionOfferJournalRecord | undefined> {
|
|
this.assertOwned();
|
|
assertRunDispatchOfferId(offerId);
|
|
let handle: fs.FileHandle;
|
|
try {
|
|
handle = await fs.open(
|
|
this.target(offerId),
|
|
constants.O_RDONLY | (constants.O_NOFOLLOW ?? 0),
|
|
);
|
|
} catch (error) {
|
|
if (isCode(error, 'ENOENT')) return undefined;
|
|
throw error;
|
|
}
|
|
try {
|
|
const stat = await handle.stat();
|
|
if (!stat.isFile()) {
|
|
throw new TypeError(
|
|
'Worker execution offer journal entry must be a regular file',
|
|
);
|
|
}
|
|
const bytes = Buffer.allocUnsafe(
|
|
MAX_WORKER_EXECUTION_OFFER_RECORD_BYTES + 1,
|
|
);
|
|
const { bytesRead } = await handle.read(bytes, 0, bytes.length, 0);
|
|
if (bytesRead > MAX_WORKER_EXECUTION_OFFER_RECORD_BYTES) {
|
|
throw new TypeError(
|
|
'Worker execution offer journal entry exceeds the byte limit',
|
|
);
|
|
}
|
|
const record = parseWorkerExecutionOfferJournalRecord(
|
|
bytes.subarray(0, bytesRead),
|
|
);
|
|
if (record.offer.offerId !== offerId) {
|
|
throw new TypeError(
|
|
'Worker offer journal path and payload do not match',
|
|
);
|
|
}
|
|
return record;
|
|
} finally {
|
|
await handle.close();
|
|
}
|
|
}
|
|
|
|
async replace(
|
|
record: WorkerExecutionOfferJournalRecord,
|
|
expectedRevision: number,
|
|
): Promise<void> {
|
|
this.assertOwned();
|
|
const candidate = cloneWorkerExecutionOfferJournalRecord(record);
|
|
assertIntegerBetween(
|
|
'expectedRevision',
|
|
expectedRevision,
|
|
0,
|
|
Number.MAX_SAFE_INTEGER - 1,
|
|
);
|
|
if (candidate.revision !== expectedRevision + 1) {
|
|
throw new TypeError('Replacement journal revision must increment by one');
|
|
}
|
|
await this.serializeMutation(async () => {
|
|
const current = await this.read(candidate.offer.offerId);
|
|
if (!current) {
|
|
throw new WorkerExecutionOfferJournalNotFoundError(
|
|
candidate.offer.offerId,
|
|
);
|
|
}
|
|
if (current.revision !== expectedRevision) {
|
|
throw new WorkerExecutionOfferJournalRevisionError(
|
|
candidate.offer.offerId,
|
|
);
|
|
}
|
|
const temporary = this.temporary(candidate.offer.offerId);
|
|
try {
|
|
await this.writeTemporary(temporary, candidate);
|
|
this.assertOwned();
|
|
await fs.rename(temporary, this.target(candidate.offer.offerId));
|
|
await this.bestEffortSyncDirectory();
|
|
} finally {
|
|
await this.bestEffortUnlink(temporary);
|
|
}
|
|
});
|
|
}
|
|
|
|
async remove(offerId: string, expectedRevision?: number): Promise<boolean> {
|
|
this.assertOwned();
|
|
assertRunDispatchOfferId(offerId);
|
|
if (expectedRevision !== undefined) {
|
|
assertIntegerBetween(
|
|
'expectedRevision',
|
|
expectedRevision,
|
|
0,
|
|
Number.MAX_SAFE_INTEGER,
|
|
);
|
|
}
|
|
return this.serializeMutation(async () => {
|
|
if (expectedRevision !== undefined) {
|
|
const current = await this.read(offerId);
|
|
if (!current) return false;
|
|
if (current.revision !== expectedRevision) {
|
|
throw new WorkerExecutionOfferJournalRevisionError(offerId);
|
|
}
|
|
}
|
|
try {
|
|
this.assertOwned();
|
|
await fs.unlink(this.target(offerId));
|
|
await this.bestEffortSyncDirectory();
|
|
return true;
|
|
} catch (error) {
|
|
if (isCode(error, 'ENOENT')) return false;
|
|
throw error;
|
|
}
|
|
});
|
|
}
|
|
|
|
async list(
|
|
options: { afterOfferId?: string; limit?: number } = {},
|
|
): Promise<WorkerExecutionOfferJournalPage> {
|
|
this.assertOwned();
|
|
if (options.afterOfferId !== undefined) {
|
|
assertRunDispatchOfferId(options.afterOfferId);
|
|
}
|
|
const limit = options.limit ?? 32;
|
|
assertIntegerBetween(
|
|
'limit',
|
|
limit,
|
|
1,
|
|
MAX_WORKER_EXECUTION_OFFER_JOURNAL_PAGE_SIZE,
|
|
);
|
|
const names = await this.entryNames();
|
|
const offerIds = names
|
|
.map((name) => JOURNAL_FILE_PATTERN.exec(name)?.[1])
|
|
.filter((value): value is string => value !== undefined)
|
|
.filter(
|
|
(offerId) =>
|
|
options.afterOfferId === undefined || offerId > options.afterOfferId,
|
|
)
|
|
.sort();
|
|
const selected = offerIds.slice(0, limit + 1);
|
|
const hasMore = selected.length > limit;
|
|
const pageIds = selected.slice(0, limit);
|
|
const records: WorkerExecutionOfferJournalRecord[] = [];
|
|
for (const offerId of pageIds) {
|
|
const record = await this.read(offerId);
|
|
if (!record) {
|
|
throw new WorkerExecutionOfferJournalRevisionError(offerId);
|
|
}
|
|
records.push(record);
|
|
}
|
|
return {
|
|
records,
|
|
...(hasMore && pageIds.length
|
|
? { nextAfterOfferId: pageIds[pageIds.length - 1] }
|
|
: {}),
|
|
};
|
|
}
|
|
|
|
private target(offerId: string): string {
|
|
assertRunDispatchOfferId(offerId);
|
|
return path.join(this.root, `${offerId}.json`);
|
|
}
|
|
|
|
private temporary(offerId: string): string {
|
|
return path.join(
|
|
this.root,
|
|
`.${offerId}.${randomBytes(16).toString('hex')}.tmp`,
|
|
);
|
|
}
|
|
|
|
private async ensureRoot(): Promise<void> {
|
|
await fs.mkdir(this.root, { recursive: true, mode: 0o700 });
|
|
const stat = await fs.lstat(this.root);
|
|
if (!stat.isDirectory() || stat.isSymbolicLink()) {
|
|
throw new TypeError(
|
|
'Worker execution offer journal root must be a real directory',
|
|
);
|
|
}
|
|
await fs.chmod(this.root, 0o700);
|
|
}
|
|
|
|
private assertOwned(): void {
|
|
if (this.ownerState === 'owned') return;
|
|
throw new WorkerExecutionOfferJournalOwnershipError(
|
|
this.ownerState === 'compromised' ? 'compromised' : 'not_owned',
|
|
this.ownershipError,
|
|
);
|
|
}
|
|
|
|
private compromiseOwnership(error: Error): void {
|
|
this.ownershipError = error;
|
|
this.releaseOwner = undefined;
|
|
this.ownerState = 'compromised';
|
|
try {
|
|
this.onOwnershipCompromised?.(error);
|
|
} catch {
|
|
// Ownership loss must remain visible even if diagnostics fail.
|
|
}
|
|
}
|
|
|
|
private async entryNames(): Promise<string[]> {
|
|
let names: string[];
|
|
try {
|
|
names = await fs.readdir(this.root);
|
|
} catch (error) {
|
|
if (isCode(error, 'ENOENT')) return [];
|
|
throw error;
|
|
}
|
|
const entries = names.filter((name) => JOURNAL_FILE_PATTERN.test(name));
|
|
if (entries.length > this.maximumEntries) {
|
|
throw new WorkerExecutionOfferJournalCapacityError(this.maximumEntries);
|
|
}
|
|
return entries;
|
|
}
|
|
|
|
private async writeTemporary(
|
|
temporary: string,
|
|
record: WorkerExecutionOfferJournalRecord,
|
|
): Promise<void> {
|
|
const serialized = serializeWorkerExecutionOfferJournalRecord(record);
|
|
let handle: fs.FileHandle | undefined;
|
|
try {
|
|
handle = await fs.open(temporary, 'wx', 0o600);
|
|
await handle.writeFile(serialized, 'utf8');
|
|
await handle.sync();
|
|
} finally {
|
|
await handle?.close().catch(() => undefined);
|
|
}
|
|
}
|
|
|
|
private async exists(target: string): Promise<boolean> {
|
|
try {
|
|
const stat = await fs.lstat(target);
|
|
if (!stat.isFile()) {
|
|
throw new TypeError(
|
|
'Worker execution offer journal target must be a regular file',
|
|
);
|
|
}
|
|
return true;
|
|
} catch (error) {
|
|
if (isCode(error, 'ENOENT')) return false;
|
|
throw error;
|
|
}
|
|
}
|
|
|
|
private async bestEffortUnlink(target: string): Promise<void> {
|
|
try {
|
|
await fs.unlink(target);
|
|
} catch (error) {
|
|
if (!isCode(error, 'ENOENT')) {
|
|
// Temp cleanup is diagnostic-only; the atomically published record wins.
|
|
}
|
|
}
|
|
}
|
|
|
|
private async bestEffortSyncDirectory(): Promise<void> {
|
|
try {
|
|
const handle = await fs.open(this.root, constants.O_RDONLY);
|
|
try {
|
|
await handle.sync();
|
|
} finally {
|
|
await handle.close();
|
|
}
|
|
} catch {
|
|
// Some supported filesystems cannot fsync directories.
|
|
}
|
|
}
|
|
|
|
private serializeMutation<T>(operation: () => Promise<T>): Promise<T> {
|
|
const result = this.mutationTail.then(operation, operation);
|
|
this.mutationTail = result.then(
|
|
() => undefined,
|
|
() => undefined,
|
|
);
|
|
return result;
|
|
}
|
|
}
|