A background job processor.
The work finished.
Did the job?

/** A deterministic report processor. Time and worker scheduling are explicit inputs. */
export type Fault = 'none' | 'slow' | 'transient' | 'invalid' | 'crash-after-output';
export type OutputPolicy = 'per-job' | 'per-attempt';
export type JobStatus = 'queued' | 'running' | 'retry' | 'succeeded' | 'failed' | 'cancelled';
export interface ReportInput {
	workspace: string;
	month: string;
}
export interface Job {
	id: string;
	requestId: string;
	input: ReportInput;
	fingerprint: string;
	fault: Fault;
	policy: OutputPolicy;
	status: JobStatus;
	attempts: number;
	token: number;
	leaseUntil: number;
	due: number;
	createdAt: number;
	cancelRequested: boolean;
	lastError: string;
}
export interface Artifact {
	key: string;
	jobId: string;
	fingerprint: string;
	attempt: number;
	committedAt: number;
	csv: string;
}
export interface ProcessorEvent {
	tick: number;
	jobId: string;
	requestId: string;
	attempt: number;
	kind: string;
	message: string;
}
export interface Database {
	version: 1;
	clock: number;
	jobs: Job[];
	artifacts: Artifact[];
	events: ProcessorEvent[];
}
export const emptyDatabase = (): Database => ({
	version: 1,
	clock: 0,
	jobs: [],
	artifacts: [],
	events: []
});

export interface Store {
	read(): Database;
	// A failed callback must commit nothing. Concurrent writers must serialize.
	transaction<T>(change: (draft: Database) => T): T;
}

export class MemoryStore implements Store {
	private state = emptyDatabase();
	read() {
		return structuredClone(this.state);
	}
	transaction<T>(change: (draft: Database) => T): T {
		const draft = this.read();
		const result = change(draft);
		this.state = structuredClone(draft);
		return result;
	}
}

export const limits = { jobs: 6, attempts: 3, lease: 4, events: 120 } as const;
export const isTerminal = (job: Job) => ['succeeded', 'failed', 'cancelled'].includes(job.status);
export function record(db: Database, job: Job, kind: string, message: string) {
	db.events.push({
		tick: db.clock,
		jobId: job.id,
		requestId: job.requestId,
		attempt: job.attempts,
		kind,
		message
	});
	db.events = db.events.slice(-limits.events);
}
export const fingerprint = (input: ReportInput) =>
	JSON.stringify([input.workspace, input.month, 'fixture-v1']);

/** The report reads a fixed, synthetic data revision so a retry has identical meaning. */
export function renderReport(input: ReportInput): string {
	if (!/^[a-z0-9-]{1,32}$/.test(input.workspace) || !/^2026-(0[1-9]|1[0-2])$/.test(input.month)) {
		throw new Error('Invalid synthetic report input');
	}
	return `workspace,month,team,resolved\n${input.workspace},${input.month},support,42\n${input.workspace},${input.month},operations,17\n`;
}

export function enqueue(
	store: Store,
	requestId: string,
	input: ReportInput,
	fault: Fault = 'none',
	policy: OutputPolicy = 'per-job'
): string {
	if (!/^[a-z0-9-]{1,48}$/.test(requestId)) throw new Error('Invalid request ID');
	renderReport(input); // Validate before accepting; the invalid fault simulates a later permanent failure.
	return store.transaction((db) => {
		const existing = db.jobs.find(
			(job) => job.requestId === requestId && job.input.workspace === input.workspace
		);
		if (existing) {
			if (existing.fingerprint !== fingerprint(input))
				throw new Error('Request ID reused with different input');
			record(db, existing, 'request-reused', 'Repeated submission returned the original job.');
			return existing.id;
		}
		if (db.jobs.length >= limits.jobs)
			throw new Error('Demo storage limit reached: reset to start another run');
		const job: Job = {
			id: `export-${db.jobs.length + 1}`,
			requestId,
			input: { ...input },
			fingerprint: fingerprint(input),
			fault,
			policy,
			status: 'queued',
			attempts: 0,
			token: 0,
			leaseUntil: 0,
			due: db.clock,
			createdAt: db.clock,
			cancelRequested: false,
			lastError: ''
		};
		db.jobs.push(job);
		record(db, job, 'accepted', 'Job committed before acceptance was returned.');
		return job.id;
	});
}

export function claim(store: Store): { jobId: string; token: number } | undefined {
	return store.transaction((db) => {
		const job = db.jobs.find(
			(candidate) =>
				(candidate.status === 'queued' || candidate.status === 'retry') && candidate.due <= db.clock
		);
		if (!job) return undefined;
		job.status = 'running';
		job.attempts += 1;
		job.token += 1;
		job.leaseUntil = db.clock + limits.lease;
		record(
			db,
			job,
			'claimed',
			`Attempt ${job.attempts} owns token ${job.token}; lease ends at tick ${job.leaseUntil}.`
		);
		return { jobId: job.id, token: job.token };
	});
}

function owned(db: Database, jobId: string, token: number): Job | undefined {
	return db.jobs.find(
		(job) =>
			job.id === jobId &&
			job.status === 'running' &&
			job.token === token &&
			job.leaseUntil > db.clock
	);
}

export function publish(store: Store, jobId: string, token: number): boolean {
	return store.transaction((db) => {
		const job = owned(db, jobId, token);
		if (!job || job.cancelRequested) return false;
		const key =
			job.policy === 'per-job'
				? `${job.input.workspace}/${job.id}`
				: `${job.input.workspace}/${job.id}/attempt-${job.attempts}`;
		const existing = db.artifacts.find((artifact) => artifact.key === key);
		if (existing) {
			if (existing.fingerprint !== job.fingerprint) throw new Error('Output key conflict');
			record(
				db,
				job,
				'output-reused',
				'The original CSV receipt was reused; no second artifact was created.'
			);
		} else {
			db.artifacts.push({
				key,
				jobId,
				fingerprint: job.fingerprint,
				attempt: job.attempts,
				committedAt: db.clock,
				csv: renderReport(job.input)
			});
			record(db, job, 'output-committed', `CSV bytes and receipt committed together as ${key}.`);
		}
		return true;
	});
}

export function acknowledge(store: Store, jobId: string, token: number): boolean {
	return store.transaction((db) => {
		const job = owned(db, jobId, token);
		if (!job) return false;
		if (!db.artifacts.some((artifact) => artifact.jobId === jobId))
			throw new Error('Cannot acknowledge missing output');
		job.status = 'succeeded';
		job.leaseUntil = 0;
		job.lastError = '';
		record(db, job, 'acknowledged', 'Job marked succeeded after its output was committed.');
		return true;
	});
}

function retryOrFail(db: Database, job: Job, reason: string, permanent = false) {
	job.lastError = reason;
	job.leaseUntil = 0;
	job.token += 1; // Any old worker loses permission to mutate this job.
	if (permanent || job.attempts >= limits.attempts) {
		// A final lost acknowledgement must not hide a committed result.
		const resultExists = db.artifacts.some((artifact) => artifact.jobId === job.id);
		job.status = resultExists ? 'succeeded' : 'failed';
		record(
			db,
			job,
			resultExists ? 'reconciled' : 'failed',
			resultExists ? 'Attempt budget ended; reconciled the committed CSV receipt.' : reason
		);
		return;
	}
	// Reproducible jitter for this fixture; production uses independent jitter and a wall-clock budget.
	const jitter = Number(job.id.split('-')[1]) % 2;
	job.due = db.clock + Math.min(8, 2 ** job.attempts) + jitter;
	job.status = 'retry';
	record(db, job, 'retry-scheduled', `${reason} Next attempt is eligible at tick ${job.due}.`);
}

export function cancel(store: Store, jobId: string): string {
	return store.transaction((db) => {
		const job = db.jobs.find((candidate) => candidate.id === jobId);
		if (!job) return 'Job not found.';
		if (db.artifacts.some((artifact) => artifact.jobId === job.id))
			return 'Too late to cancel: the CSV is already committed. Cancellation cannot undo it.';
		if (isTerminal(job)) return `Job is already ${job.status}.`;
		if (job.status === 'running') {
			job.cancelRequested = true;
			record(
				db,
				job,
				'cancel-requested',
				'Worker will observe cancellation at its next cooperative checkpoint.'
			);
			return 'Cancellation requested; work has not stopped yet.';
		}
		job.status = 'cancelled';
		job.token += 1;
		record(db, job, 'cancelled', 'Cancelled before the next attempt started.');
		return 'Pending work cancelled.';
	});
}

interface Work {
	jobId: string;
	token: number;
	remaining: number;
	phase: 'render' | 'ack';
}
export interface Worker {
	id: string;
	online: boolean;
	work?: Work;
}

/** Worker slots are volatile. The Store owns job and output state across worker restarts. */
export class Processor {
	private store: Store;
	private slots: Worker[];
	constructor(store: Store, concurrency = 2) {
		if (!Number.isInteger(concurrency) || concurrency < 1 || concurrency > 3)
			throw new Error('Choose 1–3 workers');
		this.store = store;
		this.slots = Array.from({ length: concurrency }, (_, index) => ({
			id: `worker-${index + 1}`,
			online: true
		}));
	}
	workers(): Worker[] {
		return structuredClone(this.slots);
	}
	recover(): void {
		for (const slot of this.slots) slot.online = true;
	}
	crash(workerId: string): boolean {
		const slot = this.slots.find((worker) => worker.id === workerId);
		if (!slot?.online) return false;
		if (slot.work) {
			const jobId = slot.work.jobId;
			this.store.transaction((db) => {
				const job = db.jobs.find((item) => item.id === jobId)!;
				record(
					db,
					job,
					'worker-crashed',
					`${slot.id} disappeared; its lease remains until expiry.`
				);
			});
		}
		slot.online = false;
		slot.work = undefined;
		return true;
	}
	advance(): void {
		this.store.transaction((db) => {
			db.clock += 1;
			for (const job of db.jobs) {
				if (job.status !== 'running' || job.leaseUntil > db.clock) continue;
				if (job.cancelRequested && !db.artifacts.some((artifact) => artifact.jobId === job.id)) {
					job.status = 'cancelled';
					job.token += 1;
					job.leaseUntil = 0;
					record(
						db,
						job,
						'cancelled',
						'Expired attempt was cancelled before any output committed.'
					);
				} else retryOrFail(db, job, 'Worker lease expired.');
			}
		});
		for (const slot of this.slots) {
			if (!slot.online) continue;
			const work = slot.work;
			if (work) {
				const job = this.store.transaction((db) => {
					const active = owned(db, work.jobId, work.token);
					if (!active) return undefined;
					if (active.cancelRequested) {
						active.status = 'cancelled';
						active.token += 1;
						active.leaseUntil = 0;
						record(db, active, 'cancelled', 'Worker stopped at a checkpoint before publishing.');
						return undefined;
					}
					active.leaseUntil = db.clock + limits.lease; // Heartbeat while this worker is making progress.
					return structuredClone(active);
				});
				if (!job) {
					slot.work = undefined;
					continue;
				}
				if (work.phase === 'ack') {
					acknowledge(this.store, work.jobId, work.token);
					slot.work = undefined;
				} else if (--work.remaining <= 0) {
					if (job.fault === 'invalid' || (job.fault === 'transient' && job.attempts < 3)) {
						this.store.transaction((db) => {
							const active = owned(db, work.jobId, work.token);
							if (active)
								retryOrFail(
									db,
									active,
									job.fault === 'invalid'
										? 'Unsupported source schema.'
										: 'Report source temporarily unavailable.',
									job.fault === 'invalid'
								);
						});
						slot.work = undefined;
					} else if (publish(this.store, work.jobId, work.token)) {
						work.phase = 'ack';
						if (job.fault === 'crash-after-output' && job.attempts === 1) this.crash(slot.id);
					} else slot.work = undefined;
				}
			}
			if (!slot.work && slot.online) {
				const next = claim(this.store);
				if (next) {
					const job = this.store.read().jobs.find((item) => item.id === next.jobId)!;
					slot.work = { ...next, phase: 'render', remaining: job.fault === 'slow' ? 6 : 2 };
				}
			}
		}
	}
}

export function seed(
	store: Store,
	fault: Fault = 'crash-after-output',
	policy: OutputPolicy = 'per-job'
) {
	enqueue(store, 'request-1', { workspace: 'acme', month: '2026-08' }, fault, policy);
	enqueue(store, 'request-2', { workspace: 'acme', month: '2026-07' });
	enqueue(store, 'request-3', { workspace: 'acme', month: '2026-06' });
}

import { DatabaseSync } from 'node:sqlite';
import { emptyDatabase, type Database, type Store } from './processor.ts';

/** A tiny local adapter. One JSON row keeps the teaching transaction boundary inspectable. */
export class SqliteStore implements Store {
	private db: DatabaseSync;
	constructor(filename: string) {
		this.db = new DatabaseSync(filename);
		this.db.exec(`
			PRAGMA journal_mode = WAL;
			PRAGMA synchronous = FULL;
			PRAGMA busy_timeout = 3000;
			CREATE TABLE IF NOT EXISTS processor_state (
				id INTEGER PRIMARY KEY CHECK (id = 1),
				body TEXT NOT NULL
			);
		`);
		this.db
			.prepare('INSERT OR IGNORE INTO processor_state (id, body) VALUES (1, ?)')
			.run(JSON.stringify(emptyDatabase()));
	}
	read(): Database {
		const row = this.db.prepare('SELECT body FROM processor_state WHERE id = 1').get()!;
		const state = JSON.parse(String(row.body)) as Database;
		if (state.version !== 1) throw new Error('Unsupported processor state version');
		return state;
	}
	transaction<T>(change: (draft: Database) => T): T {
		this.db.exec('BEGIN IMMEDIATE');
		try {
			const draft = this.read();
			const result = change(draft);
			this.db
				.prepare('UPDATE processor_state SET body = ? WHERE id = 1')
				.run(JSON.stringify(draft));
			this.db.exec('COMMIT');
			return result;
		} catch (error) {
			this.db.exec('ROLLBACK');
			throw error;
		}
	}
	close() {
		this.db.close();
	}
}

import { existsSync } from 'node:fs';
import { enqueue, isTerminal, Processor } from './processor.ts';
import { SqliteStore } from './sqlite-store.ts';

// Run with Node 22.21+; node:sqlite is experimental in the verified Node 22 runtime.
const [phase, filename] = process.argv.slice(2);
if (!filename || !['crash', 'recover', 'inspect'].includes(phase)) {
	throw new Error(
		'Usage: node --experimental-strip-types run.ts crash|recover|inspect /tmp/new-report-run.sqlite'
	);
}
if (phase === 'crash' && existsSync(filename))
	throw new Error('Choose a fresh database filename; existing runs are preserved');
if (phase !== 'crash' && !existsSync(filename)) throw new Error('Run the crash phase first');

const store = new SqliteStore(filename);
if (phase === 'crash') {
	enqueue(store, 'request-1', { workspace: 'acme', month: '2026-08' }, 'crash-after-output');
}
const processor = new Processor(store, 1);
if (phase !== 'inspect') {
	for (let step = 0; step < 80 && !store.read().jobs.every(isTerminal); step++) {
		processor.advance();
		if (phase === 'crash' && processor.workers().some((worker) => !worker.online)) {
			console.log(JSON.stringify({ phase, state: store.read() }, null, 2));
			// Deliberately exit without acknowledging the job or gracefully closing its database.
			process.exit(17);
		}
	}
}
console.log(JSON.stringify({ phase, state: store.read() }, null, 2));
store.close();