Repository Public
1/** Executes one claimed run: replays the logic from the top, then settles the run's fate. */2import { ResultEnvelope } from '@aos/shared';3import { type Kernel } from './db.js';4import {5 BudgetExhaustedError, isControlError, KernelStorageError, LeaseLostError, NonDeterminismError, ReconcileAborted, RunCancelled, Suspend, WorkerStopping,6} from './errors.js';7import { RunCtx } from './context.js';8import { fencedTx, loadJournal } from './journal.js';9import { Heartbeat, releaseLease, type ClaimedRun } from './lease.js';10import { envelope, finalizeRun } from './runs.js';11import type { KernelHooks, RunLogic } from './types.js';1213export interface ExecuteOptions {14 logic: RunLogic;15 leaseMs: number;16 heartbeatMs: number;17 inlineSleepMs: number;18 dailyCapUsd: (missionId: string) => Promise<number | null>;19 hooks?: KernelHooks;20}2122/** What became of the claim (for logs and tests). */23export type Outcome = 'finished' | 'suspended' | 'cancelled' | 'released' | 'lease_lost' | 'storage_error';2425export interface Execution {26 readonly runId: string;27 /** Graceful shutdown: abort ctx.signal; the run is left 'running' with a released lease. */28 stop(): void;29 /** A cancel doorbell rang for this run: check now instead of at the next heartbeat. */30 poke(): void;31 done: Promise<Outcome>;32}3334export function executeRun(k: Kernel, claimed: ClaimedRun, opts: ExecuteOptions): Execution {35 const { run, token } = claimed;36 const log = k.log.child({ runId: run.id, missionId: run.mission_id });37 let ctx: RunCtx | null = null;38 let stopRequested = false;39 const hb = new Heartbeat(k, run.id, token, opts.leaseMs, opts.heartbeatMs, {40 lost: (why) => {41 log.warn({ why }, 'lease lost; abandoning the run');42 ctx?.abort(new LeaseLostError(run.id, why));43 },44 cancel: () => ctx?.abort(new RunCancelled()),45 });4647 const finalize = async (env: ResultEnvelope): Promise<Outcome> => {48 hb.stop(); // the run is about to leave 'running': a heartbeat racing with it would report a lost lease49 await fencedTx(k, run.id, token, (tx, r) => finalizeRun(tx, r, r.cancel_requested && env.status !== 'cancelled' ? envelope('cancelled', 'Cancelled by the owner.') : env));50 log.info({ status: env.status }, 'run finished');51 return env.status === 'cancelled' ? 'cancelled' : 'finished';52 };5354 const main = async (): Promise<Outcome> => {55 hb.start();56 if (run.cancel_requested) return finalize(envelope('cancelled', 'Cancelled by the owner.'));57 const journal = await loadJournal(k, run.id);58 ctx = new RunCtx(k, claimed, journal, {59 hooks: opts.hooks,60 dailyCapUsd: () => opts.dailyCapUsd(run.mission_id),61 inlineSleepMs: opts.inlineSleepMs,62 onSuspend: () => hb.stop(),63 });64 if (stopRequested) ctx.abort(new WorkerStopping());65 log.debug({ recordedSteps: journal.size, prevStatus: claimed.prevStatus }, 'run claimed');6667 let result: ResultEnvelope | undefined;68 let error: unknown;69 try {70 result = await opts.logic(ctx);71 } catch (err) {72 error = err;73 }74 await ctx.settle(); // parallel branches: let in-flight steps commit (or be refused) first7576 // 1. The context decides before the logic's own outcome: a suspended run must not be finalized,77 // and a logic that swallowed an abort must not turn a shutdown, a lost lease or a storage error78 // into a result. `cause` is the sticky abort reason, else a control error thrown by the logic.79 if (ctx.state === 'suspended') return 'suspended';80 const cause = ctx.abortReason ?? (isControlError(error) || error instanceof NonDeterminismError ? (error as Error) : null);81 if (cause instanceof LeaseLostError) return 'lease_lost';82 if (cause instanceof KernelStorageError) {83 // Infrastructure failure: never a verdict on the run. Leave it resumable.84 log.error({ err: cause.message }, 'storage error; run left resumable');85 await releaseLease(k, run.id, token).catch(() => undefined);86 return 'storage_error';87 }88 if (cause instanceof RunCancelled) {89 return finalize(result?.status === 'cancelled' && ResultEnvelope.safeParse(result).success ? ResultEnvelope.parse(result) : envelope('cancelled', 'Cancelled by the owner.'));90 }91 if (cause instanceof WorkerStopping) {92 await releaseLease(k, run.id, token);93 return 'released';94 }95 if (cause instanceof Suspend) return 'suspended'; // unreachable in practice (state is set first)96 // 2. Verdicts.97 if (cause instanceof ReconcileAborted) return finalize(envelope('cancelled', cause.message));98 if (cause instanceof NonDeterminismError) {99 log.error({ err: cause.message }, 'non-deterministic run logic');100 return finalize(envelope('failed', cause.message));101 }102 if (error instanceof BudgetExhaustedError) return finalize(envelope('budget_exhausted', error.message));103 if (error !== undefined) {104 const msg = error instanceof Error ? error.message : String(error);105 log.warn({ errName: error instanceof Error ? error.name : 'unknown' }, 'run logic threw');106 return finalize(envelope('failed', `Run failed: ${msg}`));107 }108 const parsed = ResultEnvelope.safeParse(result);109 if (!parsed.success) return finalize(envelope('failed', 'Run logic returned an invalid result envelope.'));110 const extra = [...journal.keys()].filter((s) => s >= (ctx as RunCtx).allocated);111 if (extra.length > 0) {112 const e = new NonDeterminismError(run.id, Math.min(...extra), `logic returned after ${(ctx as RunCtx).allocated} steps but the journal has more`);113 return finalize(envelope('failed', e.message));114 }115 return finalize(parsed.data);116 };117118 const done = main()119 .catch((err: unknown): Outcome => {120 if (err instanceof LeaseLostError) return 'lease_lost';121 // Could not load the journal / finalize: leave the run to lease expiry.122 log.error({ err: (err as Error).message }, 'run execution aborted; run left resumable');123 return 'storage_error';124 })125 .finally(() => {126 hb.stop();127 ctx?.abort(new LeaseLostError(run.id, 'execution closed')); // stop stragglers (abandoned branches)128 });129130 return {131 runId: run.id,132 stop: () => {133 stopRequested = true;134 ctx?.abort(new WorkerStopping());135 },136 poke: () => hb.poke(),137 done,138 };139}140