import type { SqliteDatabase } from '../../infrastructure/database/database.js'; /** * Reconciles jobs that the console still believes are in flight. * * Every job is executed inside the request that created it, so a row left in `running` after a * crash can never finish on its own. The official Hub exposes the same capability on its command * ledger; here it closes the loop against our own job table instead of a remote queue. * * `queued` rows never reached a transport, so they are cancelled. `running` and `cancelling` rows * may or may not have reached the device, so they become `unknown-result`, which is the only * honest terminal state for an interrupted write. * * The service also owns operator cancellation, which the published contract promises on * `POST /api/v1/jobs/{jobId}/cancel`. Both paths share one rule: a job only ever moves from an * active state to a terminal one, and every write is guarded by the active state, so a cancel * racing an in-flight executor loses cleanly instead of overwriting a real result. */ const DISPATCHED_STATUSES = new Set(['running', 'cancelling']); const QUEUED_STATUSES = new Set(['queued']); const ACTIVE_STATUSES = new Set([...DISPATCHED_STATUSES, ...QUEUED_STATUSES]); const ACTIVE_PLACEHOLDERS = '?, ?, ?'; /** On-demand sweeps leave a margin so a slow batch in another request is never clobbered. */ export const DEFAULT_RECONCILE_GRACE_MS = 15 * 60 * 1000; export interface JobReconcileResult { /** Jobs closed by this sweep. */ readonly interrupted: number; /** Jobs still considered in flight once the sweep finished. */ readonly pending: number; } export interface JobReconcileSummary { readonly pending: number; readonly dispatched: number; readonly queued: number; readonly oldestCreatedAt: string | null; } export interface JobReconcileOptions { readonly db: SqliteDatabase; readonly now?: () => Date; } export type JobCancelErrorCode = 'NOT_FOUND' | 'NOT_CANCELLABLE'; export class JobCancelError extends Error { constructor( readonly code: JobCancelErrorCode, message: string, ) { super(message); this.name = 'JobCancelError'; } } const CANCEL_RESULT_CODE = 'CANCELLED_BY_OPERATOR'; const isActiveStatus = (status: unknown): status is string => typeof status === 'string' && ACTIVE_STATUSES.has(status); export class JobReconcileService { readonly #db: SqliteDatabase; readonly #now: () => Date; constructor(options: JobReconcileOptions) { this.#db = options.db; this.#now = options.now ?? (() => new Date()); } summary(olderThanMs = 0): JobReconcileSummary { const cutoff = this.#cutoff(olderThanMs); const rows = this.#db .prepare( 'SELECT status, created_at FROM jobs WHERE status IN (?,?,?) AND created_at <= ? ORDER BY created_at ASC, id ASC', ) .all('queued', 'running', 'cancelling', cutoff) as Array<{ status: unknown; created_at: unknown; }>; let dispatched = 0; let queued = 0; for (const row of rows) { if (typeof row.status === 'string' && DISPATCHED_STATUSES.has(row.status)) dispatched += 1; else if (typeof row.status === 'string' && QUEUED_STATUSES.has(row.status)) queued += 1; } const oldest = rows[0]?.created_at; return { pending: rows.length, dispatched, queued, oldestCreatedAt: typeof oldest === 'string' ? oldest : null, }; } reconcile(olderThanMs = DEFAULT_RECONCILE_GRACE_MS): JobReconcileResult { const cutoff = this.#cutoff(olderThanMs); const closedAt = this.#now().toISOString(); const interrupted = this.#db.transaction((): number => { const rows = this.#db .prepare( 'SELECT id, status FROM jobs WHERE status IN (?,?,?) AND created_at <= ? ORDER BY created_at ASC, id ASC', ) .all('queued', 'running', 'cancelling', cutoff) as Array<{ id: unknown; status: unknown; }>; const closeItems = this.#db.prepare( `UPDATE job_items SET status = ?, result_code = ?, finished_at = ?, updated_at = ? WHERE job_id = ? AND status IN ('queued','running','cancelling')`, ); const closeAttempts = this.#db.prepare( "UPDATE job_attempts SET status = ?, finished_at = ? WHERE job_id = ? AND status = 'running'", ); const closeJob = this.#db.prepare( 'UPDATE jobs SET status = ?, finished_at = ?, updated_at = ? WHERE id = ? AND status IN (?,?,?)', ); let count = 0; for (const row of rows) { if (typeof row.id !== 'string' || !isActiveStatus(row.status)) continue; const dispatched = DISPATCHED_STATUSES.has(row.status); const finalStatus = dispatched ? 'unknown-result' : 'cancelled'; const resultCode = dispatched ? 'INTERRUPTED' : 'NEVER_DISPATCHED'; closeItems.run(finalStatus, resultCode, closedAt, closedAt, row.id); closeAttempts.run(finalStatus, closedAt, row.id); closeJob.run(finalStatus, closedAt, closedAt, row.id, 'queued', 'running', 'cancelling'); count += 1; } return count; })(); return { interrupted, pending: this.summary().pending }; } #cutoff(olderThanMs: number): string { if (!Number.isFinite(olderThanMs) || olderThanMs < 0) { throw new RangeError('olderThanMs must be a non-negative finite number'); } return new Date(this.#now().getTime() - olderThanMs).toISOString(); } /** Requests cancellation of a job that has not reached a terminal state yet. */ cancel(jobId: string): void { if (typeof jobId !== 'string' || jobId.length === 0 || jobId.length > 128) { throw new JobCancelError('NOT_FOUND', 'Job was not found'); } const closedAt = this.#now().toISOString(); const changed = this.#db.transaction((): boolean => { const row = this.#db.prepare('SELECT status FROM jobs WHERE id = ?').get(jobId) as | { status: unknown } | undefined; if (!row) throw new JobCancelError('NOT_FOUND', 'Job was not found'); if (!isActiveStatus(row.status)) throw new JobCancelError('NOT_CANCELLABLE', 'Job already reached a terminal state'); this.#db .prepare( `UPDATE job_items SET status = 'cancelled', result_code = ?, finished_at = ?, updated_at = ? WHERE job_id = ? AND status IN ('queued', 'running', 'cancelling')`, ) .run(CANCEL_RESULT_CODE, closedAt, closedAt, jobId); this.#db .prepare( "UPDATE job_attempts SET status = 'cancelled', finished_at = ? WHERE job_id = ? AND status = 'running'", ) .run(closedAt, jobId); const updated = this.#db .prepare( `UPDATE jobs SET status = 'cancelled', finished_at = ?, updated_at = ? WHERE id = ? AND status IN (${ACTIVE_PLACEHOLDERS})`, ) .run(closedAt, closedAt, jobId, 'queued', 'running', 'cancelling'); return updated.changes === 1; })(); if (!changed) throw new JobCancelError('NOT_CANCELLABLE', 'Job already reached a terminal state'); } }