diff --git a/.changeset/quiet-schedules-write-worker.md b/.changeset/quiet-schedules-write-worker.md new file mode 100644 index 0000000000..10bcc785e4 --- /dev/null +++ b/.changeset/quiet-schedules-write-worker.md @@ -0,0 +1,5 @@ +--- +"@schlessera/brain-ui-server": patch +--- + +Commit schedule proposals, approvals, publication, cancellation and reconciliation off the server's event loop while retaining FULL SQLite durability. diff --git a/docs/decisions/sqlite-write-worker.md b/docs/decisions/sqlite-write-worker.md index d63631ed7e..a770f75642 100644 --- a/docs/decisions/sqlite-write-worker.md +++ b/docs/decisions/sqlite-write-worker.md @@ -245,6 +245,22 @@ are in [the governed-workers decision](governed-writable-workers.md#admitted-obj again on the write connection. Standalone callers without a registered writer run the same procedures inline. Dispatcher-fixture-only claims, boot/restore helpers and the cron runner retain their synchronous connections. +- **Schedule route writes followed** (#3003, within #2850). Proposals, + approvals, pending publication, activation or quarantine, cancellation and + reconciliation commit on the worker + (`packages/ui-server/src/schedules/service-writes.ts`). Each transaction + keeps its reads and `BEGIN IMMEDIATE`; cancellation still invalidates + unstarted occurrences and drops their Queue items and change records in the + same transaction. Publication awaits its pending receipt, then checks Git + and publishes the approved file, then awaits activation or quarantine. + Cancellation awaits its ledger change before retiring the file, and marks + retirement complete only after that file operation. A late publisher of a + cancelled task journals and completes retirement again. Reopening awaits + restored-root adoption and journal reconciliation before its own transaction. + Every route reads its result after the relevant commits; schedule refusals + keep their class, code and HTTP status across the worker. `rootInfo` still + creates the root identity on the app connection during boot, before serving; + the unwired admission service and cron execution retain their own writes. - **Other app-connection writers stay on the loop for now**: retained retries, drafts, policy proposals, storage notices, push subscriptions and VAPID keys, principal revocation and administration, diff --git a/packages/ui-server/src/app.ts b/packages/ui-server/src/app.ts index 99fee9ff59..63667515be 100644 --- a/packages/ui-server/src/app.ts +++ b/packages/ui-server/src/app.ts @@ -755,7 +755,7 @@ export async function createApp(options: CreateAppOptions = {}): Promise = { + invalid_request: 400, invalid_cursor: 400, unauthorized: 403, not_found: 404, key_conflict: 409, + approval_required: 409, approval_expired: 409, definition_conflict: 409, server_unavailable: 503, + unsupported_capability: 503, +}; + +export class ScheduleError extends Error { + readonly status: number; + constructor(readonly code: ScheduleErrorCode, message: string, status?: number) { + super(message); + this.status = status ?? STATUS[code]; + } +} + diff --git a/packages/ui-server/src/schedules/service-writes.ts b/packages/ui-server/src/schedules/service-writes.ts new file mode 100644 index 0000000000..f67805fb1f --- /dev/null +++ b/packages/ui-server/src/schedules/service-writes.ts @@ -0,0 +1,239 @@ +/** Schedule ledger steps; SQL stays ordered around the app's file/Git work (#3003). */ +import type { Database } from "bun:sqlite"; +import { randomBytes } from "node:crypto"; +import { isUsablePrincipal, resolvePrincipal, type Principal } from "../db/principals.js"; +import { procedureCaller, type SqliteWriter, type WireError, type WriteProcedures } from "../db/write-worker.js"; +import { canonicalJson, scheduleFingerprint, sha256Hex } from "./canonical.js"; +import type { StoredDefinition } from "./definition.js"; +import { ScheduleError, type ScheduleErrorCode } from "./errors.js"; +import { dropUnstartedItems } from "./queue-items.js"; +import { parseInstant, SCHEDULE_FRESHNESS_MS } from "./time.js"; +import type { ApprovalRow, ExecutionPolicy, ProposalRow, TaskRow } from "./service.js"; + +export const url = import.meta.url; +interface Root { path: string; identity: string } + +function revive(error: WireError): Error | undefined { + if (error.name === "ScheduleError") + return new ScheduleError(error.fields.code as ScheduleErrorCode, error.message, error.fields.status as number); + return undefined; +} + +export function setup(db: Database, root: Root): WriteProcedures { + // Production reads its clock inside the worker's transaction. Tests can + // supply the service's injected clock without sending a closure to a worker. + let suppliedNow: number | undefined; + const now = () => suppliedNow ?? Date.now(); + const rootInfo = () => root; + const newId = (prefix: string) => `${prefix}_${randomBytes(16).toString("hex")}`; + const isOperator = (principal: Principal) => principal.kind === "owner" || principal.kind === "ambient"; + function usable(id: string): Principal | null { + const current = resolvePrincipal(db, id); + return current && isUsablePrincipal(current, now()) ? current : null; + } + function caller(principal: Principal): Principal { + const current = usable(principal.id); + if (!current) throw new ScheduleError("unauthorized", "This credential is no longer usable.", 401); + return current; + } + function restorePending(): boolean { + const bound = db.query("SELECT root_path FROM schedule_root WHERE id = 1").get() as { root_path: string } | null; + return bound?.root_path !== root.path || Boolean(db.query("SELECT 1 FROM inbox_recovery_state WHERE status = 'pending'").get()); + } + function assertWritable(): void { + if (restorePending()) throw new ScheduleError("unsupported_capability", "Schedule storage awaits restore reconciliation."); + } + const taskRow = (id: string) => db.query("SELECT * FROM schedule_tasks WHERE id = ?").get(id) as TaskRow | null; + function authorized(row: TaskRow): boolean { + if (!usable(row.creator_principal_id)) return false; + const approval = db.query("SELECT approver_principal_id, approver_kind FROM schedule_approvals WHERE id = ?") + .get(row.approval_id) as { approver_principal_id: string; approver_kind: string } | null; + const approver = approval ? usable(approval.approver_principal_id) : null; + return Boolean(approver && isOperator(approver) && approver.kind === approval!.approver_kind); + } + function finished(row: TaskRow, at: number): boolean { + const { when } = JSON.parse(row.definition_json) as StoredDefinition; + if (when.kind === "at") return parseInstant(when.at)! + SCHEDULE_FRESHNESS_MS <= at || + Boolean(db.query("SELECT 1 FROM schedule_occurrences WHERE task_id = ? LIMIT 1").get(row.id)); + return when.endAt !== null && parseInstant(when.endAt)! <= at; + } + const procedures = { + propose(actor: Principal, row: ProposalRow) { + const key = row.request_key; + return db.transaction(() => { + caller(actor); + const raced = db.query("SELECT * FROM schedule_proposals WHERE principal_id = ? AND request_key = ?").get(actor.id, key) as ProposalRow | null; + if (raced) return raced; + db.query(`INSERT INTO schedule_proposals (id, task_id, principal_id, request_key, input_hash, definition_json, + zone_source, execution_policy_json, root_identity, fingerprint, created_at, expires_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`).run(row.id, row.task_id, actor.id, key, row.input_hash, + row.definition_json, row.zone_source, row.execution_policy_json, row.root_identity, row.fingerprint, row.created_at, row.expires_at); + return null; + }).immediate(); + }, + approve(actor: Principal, proposal: ProposalRow, fingerprint: string, policy: ExecutionPolicy) { + return db.transaction(() => { + caller(actor); + const existing = db.query("SELECT * FROM schedule_approvals WHERE proposal_id = ?").get(proposal.id) as ApprovalRow | null; + if (existing) return { approvalId: existing.id }; + if (proposal.expires_at <= now()) throw new ScheduleError("approval_expired", "This proposal expired; propose again with a new key."); + assertWritable(); + if (proposal.root_identity !== rootInfo().identity) throw new ScheduleError("definition_conflict", "The proposal belongs to another brain root."); + if (!usable(proposal.principal_id)) throw new ScheduleError("unauthorized", "The schedule's creator is no longer usable."); + if (canonicalJson(policy) !== proposal.execution_policy_json) + throw new ScheduleError("definition_conflict", "The host execution policy changed since this proposal."); + const id = newId("approval"); + db.query(`INSERT INTO schedule_approvals (id, proposal_id, approver_principal_id, approver_kind, channel, + fingerprint, nonce, approved_at) VALUES (?, ?, ?, ?, 'http', ?, ?, ?)`) + .run(id, proposal.id, actor.id, actor.kind, fingerprint, randomBytes(16).toString("hex"), now()); + return { approvalId: id }; + }).immediate(); + }, + publish(actor: Principal, proposal: ProposalRow, approval: ApprovalRow, policy: ExecutionPolicy, definition: StoredDefinition, text: string) { + return db.transaction(() => { + const at = now(); + caller(actor); + const raced = db.query("SELECT id FROM schedule_tasks WHERE proposal_id = ?").get(proposal.id) as { id: string } | null; + if (raced) return null; + if (proposal.expires_at <= at) throw new ScheduleError("approval_expired", "This approval expired before publication; propose again with a new key."); + assertWritable(); + const approver = usable(approval.approver_principal_id); + if (!approver || !isOperator(approver) || approver.kind !== approval.approver_kind) + throw new ScheduleError("unauthorized", "The approving operator is no longer usable."); + if (approval.fingerprint !== proposal.fingerprint || proposal.root_identity !== rootInfo().identity) + throw new ScheduleError("definition_conflict", "The approval no longer matches this proposal."); + if (canonicalJson(policy) !== proposal.execution_policy_json) + throw new ScheduleError("definition_conflict", "The host execution policy changed since approval."); + const recomputed = scheduleFingerprint({ definition, rootIdentity: proposal.root_identity, + creatorPrincipalId: proposal.principal_id, executionPolicy: JSON.parse(proposal.execution_policy_json) }); + if (recomputed !== proposal.fingerprint) throw new ScheduleError("definition_conflict", "The stored proposal does not match its fingerprint."); + // First-publication timing is rechecked; replay above never reaches here. + if (definition.when.kind === "at" && parseInstant(definition.when.at)! <= at) + throw new ScheduleError("invalid_request", "The one-off time has passed; propose a new future time."); + if (definition.when.kind === "cron" && definition.when.endAt !== null && parseInstant(definition.when.endAt)! <= at) + throw new ScheduleError("invalid_request", "The recurrence end has passed; propose a new schedule."); + db.query("UPDATE schedule_approvals SET consumed_at = ? WHERE id = ? AND consumed_at IS NULL").run(at, approval.id); + db.query(`INSERT INTO schedule_tasks (id, proposal_id, approval_id, creator_principal_id, root_identity, + definition_json, execution_policy_json, fingerprint, file_sha256, zone_source, created_at, state, + blocked_reason, publication, retirement, cancelled_at, evaluated_through, updated_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'publishing', NULL, 'pending', 'none', NULL, ?, ?)`) + .run(proposal.task_id, proposal.id, approval.id, proposal.principal_id, proposal.root_identity, + proposal.definition_json, proposal.execution_policy_json, proposal.fingerprint, sha256Hex(text), + proposal.zone_source, at, at, at); + return proposal.task_id; + }).immediate(); + }, + quarantine(id: string) { + return db.transaction(() => { + db.query(`UPDATE schedule_tasks SET publication = 'quarantined', updated_at = ?, + state = CASE WHEN state = 'publishing' THEN 'paused' ELSE state END, + blocked_reason = CASE WHEN state = 'publishing' THEN 'definition_drift' ELSE blocked_reason END + WHERE id = ? AND publication = 'pending'`).run(now(), id); + }).immediate(); + }, + activate(id: string) { + return db.transaction(() => { + const result = db.query(`UPDATE schedule_tasks SET publication = 'published', updated_at = ?, + state = CASE WHEN state = 'publishing' THEN 'active' ELSE state END + WHERE id = ? AND publication = 'pending'`).run(now(), id); + return result.changes > 0 && taskRow(id)!.state === "active"; + }).immediate(); + }, + retireAgain(id: string) { + db.query("UPDATE schedule_tasks SET retirement = 'pending', updated_at = ? WHERE id = ? AND state = 'cancelled'").run(now(), id); + }, + retired(id: string) { + db.query("UPDATE schedule_tasks SET retirement = 'retired', updated_at = ? WHERE id = ? AND retirement = 'pending'").run(now(), id); + }, + pauseDrifted(id: string) { + db.query("UPDATE schedule_tasks SET state = 'paused', blocked_reason = 'definition_drift', updated_at = ? WHERE id = ? AND state = 'active'").run(now(), id); + }, + cancel(actor: Principal, row: TaskRow, key: string) { + return db.transaction(() => { + caller(actor); + const receipt = db.query("SELECT task_id FROM schedule_cancel_receipts WHERE principal_id = ? AND request_key = ?") + .get(actor.id, key) as { task_id: string } | null; + if (receipt) { + if (receipt.task_id !== row.id) throw new ScheduleError("key_conflict", "This key was already used to cancel a different task."); + return false; + } + const current = taskRow(row.id)!; + const at = now(); + // Future/unstarted only: terminal one-offs and earlier cancellations stay as they are. + const cancellable = ["publishing", "active", "paused"].includes(current.state); + if (cancellable) { + db.query(`UPDATE schedule_tasks SET state = 'cancelled', blocked_reason = NULL, cancelled_at = ?, + retirement = 'pending', updated_at = ? WHERE id = ?`).run(at, at, row.id); + // Unstarted and stale-Action work is invalidated in the same transaction. + // A started attempt keeps its bounded snapshot; this is not a live stop. + db.query(`UPDATE schedule_occurrences SET state = 'cancelled', updated_at = ? + WHERE task_id = ? AND state IN ('queued', 'retrying', 'waiting_for_action')`).run(at, row.id); + // Their unclaimed Queue items go with them. A claim that has not + // started yet is refused by admission's start check instead. + dropUnstartedItems(db, row.id, at); + } + db.query("INSERT INTO schedule_cancel_receipts (principal_id, request_key, task_id, changed, created_at) VALUES (?, ?, ?, ?, ?)") + .run(actor.id, key, row.id, cancellable ? 1 : 0, at); + return cancellable; + }).immediate(); + }, + adoptRestoredRoot() { + const info = rootInfo(); + return db.transaction(() => { + const bound = db.query("SELECT root_path FROM schedule_root WHERE id = 1").get() as { root_path: string } | null; + if (bound?.root_path === info.path) return; + const restored = db.query("SELECT status, brain_root FROM inbox_recovery_state WHERE id = 1").get() as + { status: string; brain_root: string } | null; + if (restored?.status !== "ready" || restored.brain_root !== info.path) + throw new ScheduleError("unsupported_capability", "Schedule storage belongs to another brain root; no restore named this one."); + db.query("UPDATE schedule_root SET root_path = ? WHERE id = 1").run(info.path); + }).immediate(); + }, + reopen(actor: Principal, row: TaskRow, key: string, before: TaskRow, drifted: boolean, + policyResult: { value: ExecutionPolicy } | { error: WireError }) { + const policy = "error" in policyResult ? revive(policyResult.error) ?? new Error(policyResult.error.message) : policyResult.value; + const replay = () => { + const receipt = db.query("SELECT task_id FROM schedule_reconciliations WHERE principal_id = ? AND request_key = ?") + .get(actor.id, key) as { task_id: string } | null; + if (receipt && receipt.task_id !== row.id) throw new ScheduleError("key_conflict", "This key was already used to reconcile a different task."); + return receipt !== null; + }; + return db.transaction(() => { + const at = now(); + caller(actor); + if (replay()) return false; + const current = taskRow(row.id)!; + const reason = current.state === "paused" ? current.blocked_reason : null; + const resolvable = reason === "restore_pending" || reason === "unknown_effect"; + if (resolvable) { + if (drifted || current.updated_at !== before.updated_at || current.publication !== "published") + throw new ScheduleError("definition_conflict", "The definition file does not match the approved snapshot; cancel and create a new schedule."); + if (current.root_identity !== rootInfo().identity) + throw new ScheduleError("definition_conflict", "The task belongs to another brain root."); + if (!authorized(current)) throw new ScheduleError("unauthorized", "The schedule's creator or approving operator is no longer usable."); + if (policy instanceof Error) throw policy; + if (canonicalJson(policy) !== current.execution_policy_json) + throw new ScheduleError("definition_conflict", "The host execution policy changed since approval; cancel and create a new schedule."); + if (db.query(`SELECT 1 FROM schedule_occurrences WHERE task_id = ? AND state IN + ('queued', 'running', 'unwinding', 'waiting_for_action', 'retrying')`).get(current.id)) + throw new ScheduleError("definition_conflict", "The task still has outstanding work."); + db.query("UPDATE schedule_tasks SET state = ?, blocked_reason = NULL, updated_at = ? WHERE id = ?") + .run(finished(current, at) ? "expired" : "active", at, current.id); + } + db.query(`INSERT INTO schedule_reconciliations (principal_id, request_key, task_id, approver_kind, resolved_reason, changed, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?)`).run(actor.id, key, row.id, actor.kind, resolvable ? reason : null, resolvable ? 1 : 0, at); + return resolvable; + }).immediate(); + }, + }; + return Object.fromEntries(Object.entries(procedures).map(([name, procedure]) => [name, + (clock: number | undefined, ...args: unknown[]) => { + suppliedNow = clock; + return Reflect.apply(procedure, procedures, args); + }])); +} + +/** Each call posts at once and settles only after this ledger step commits. */ +export function scheduleWrites(db: Database, root: Root, writer?: SqliteWriter) { + return procedureCaller(db, { url, setup }, root, writer, revive); +} diff --git a/packages/ui-server/src/schedules/service.ts b/packages/ui-server/src/schedules/service.ts index 0451eacf82..b8ccad8faf 100644 --- a/packages/ui-server/src/schedules/service.ts +++ b/packages/ui-server/src/schedules/service.ts @@ -13,7 +13,10 @@ import { createHmac, randomBytes, timingSafeEqual } from "node:crypto"; import { realpathSync } from "node:fs"; import { isUsablePrincipal, resolvePrincipal, type Principal } from "../db/principals.js"; -import { dropUnstartedItems } from "./queue-items.js"; +import { encodeError, type SqliteWriter } from "../db/write-worker.js"; +import { scheduleWrites } from "./service-writes.js"; +import { ScheduleError } from "./errors.js"; +export { ScheduleError, type ScheduleErrorCode } from "./errors.js"; import { canonicalJson, scheduleFingerprint, sha256Hex } from "./canonical.js"; import { DEFINITIONS_DIR, @@ -44,23 +47,6 @@ const ENVELOPE_RESERVE = 256 + MAX_CURSOR_BYTES; const DEFAULT_PAGE = 25; const MAX_PAGE = 100; -export type ScheduleErrorCode = - | "invalid_request" | "invalid_cursor" | "unauthorized" | "not_found" | "key_conflict" - | "approval_required" | "approval_expired" | "definition_conflict" | "server_unavailable" - | "unsupported_capability"; -const STATUS: Record = { - invalid_request: 400, invalid_cursor: 400, unauthorized: 403, not_found: 404, key_conflict: 409, - approval_required: 409, approval_expired: 409, definition_conflict: 409, server_unavailable: 503, - unsupported_capability: 503, -}; - -export class ScheduleError extends Error { - readonly status: number; - constructor(readonly code: ScheduleErrorCode, message: string, status?: number) { - super(message); - this.status = status ?? STATUS[code]; - } -} export interface ExecutionPolicy { backendId: string; @@ -94,12 +80,12 @@ export interface DueCandidate { admittable: boolean; blockedReason: BlockedReason | null; } -interface ProposalRow { +export interface ProposalRow { id: string; task_id: string; principal_id: string; request_key: string; input_hash: string; definition_json: string; zone_source: ZoneSource; execution_policy_json: string; root_identity: string; fingerprint: string; created_at: number; expires_at: number; } -interface ApprovalRow { +export interface ApprovalRow { id: string; proposal_id: string; approver_principal_id: string; approver_kind: string; fingerprint: string; approved_at: number; consumed_at: number | null; } @@ -161,7 +147,7 @@ export function gitCheckIgnored(root: string, boundary: HostBoundary = createHos }; } -export function createScheduleService(db: Database, options: ScheduleServiceOptions) { +export function createScheduleService(db: Database, options: ScheduleServiceOptions, writer?: SqliteWriter) { const now = options.now ?? Date.now; const dispatchAvailable = options.dispatchAvailable ?? (() => false); let root: { path: string; identity: string; secret: Buffer; matches: boolean } | null = null; @@ -169,6 +155,15 @@ export function createScheduleService(db: Database, options: ScheduleServiceOpti let gitCheck: ((relativePath: string) => boolean | Promise) | undefined; const gitIgnored = () => options.gitIgnored ?? (gitCheck ??= gitCheckIgnored(rootInfo().path)); + let writes: ReturnType | undefined; + function write(name: string, ...args: unknown[]): Promise { + if (!writes) { + const { path, identity } = rootInfo(); + writes = scheduleWrites(db, { path, identity }, writer); + } + return writes(name, options.now?.(), ...args); + } + function rootInfo() { if (root) return root; let path: string; @@ -426,26 +421,16 @@ export function createScheduleService(db: Database, options: ScheduleServiceOpti } catch (error) { if (error instanceof ScheduleFileConflictError || error instanceof ScheduleContainmentError) { // Never overwrite: quarantine the task and expose the drift. - db.transaction(() => { - db.query(`UPDATE schedule_tasks SET publication = 'quarantined', updated_at = ?, - state = CASE WHEN state = 'publishing' THEN 'paused' ELSE state END, - blocked_reason = CASE WHEN state = 'publishing' THEN 'definition_drift' ELSE blocked_reason END - WHERE id = ? AND publication = 'pending'`).run(now(), id); - }).immediate(); + await write("quarantine", id); throw new ScheduleError("definition_conflict", "A different definition file occupies this schedule's path."); } throw new ScheduleError("server_unavailable", "Schedule storage is unavailable; retry with the same key."); } - const activated = db.transaction(() => { - const result = db.query(`UPDATE schedule_tasks SET publication = 'published', updated_at = ?, - state = CASE WHEN state = 'publishing' THEN 'active' ELSE state END - WHERE id = ? AND publication = 'pending'`).run(now(), id); - return result.changes > 0 && taskRow(id)!.state === "active"; - }).immediate(); + const activated = await write("activate", id); // A cancellation that won the race retires the file this call just wrote, // even when another publisher's retirement already completed before it. if (taskRow(id)?.state === "cancelled") { - db.query("UPDATE schedule_tasks SET retirement = 'pending', updated_at = ? WHERE id = ? AND state = 'cancelled'").run(now(), id); + await write("retireAgain", id); await completeRetirement(id); } return activated; @@ -459,7 +444,7 @@ export function createScheduleService(db: Database, options: ScheduleServiceOpti let done = false; try { done = await fileStore().retire(id, serializeDefinition(definitionOf(row))); } catch { return; } // The journal stays pending; status reports compensationPending. - if (done) db.query("UPDATE schedule_tasks SET retirement = 'retired', updated_at = ? WHERE id = ? AND retirement = 'pending'").run(now(), id); + if (done) await write("retired", id); } /** Boot/retry reconciliation of every journal. Never dispatches work. */ @@ -475,8 +460,7 @@ export function createScheduleService(db: Database, options: ScheduleServiceOpti const active = db.query("SELECT * FROM schedule_tasks WHERE state = 'active' ORDER BY created_at").all() as TaskRow[]; for (const row of active) { if (await fileDrifted(row)) { - db.query("UPDATE schedule_tasks SET state = 'paused', blocked_reason = 'definition_drift', updated_at = ? WHERE id = ? AND state = 'active'") - .run(now(), row.id); + await write("pauseDrifted", row.id); } } } @@ -529,16 +513,12 @@ export function createScheduleService(db: Database, options: ScheduleServiceOpti const fingerprint = scheduleFingerprint({ definition: materialized.definition, rootIdentity: identity, creatorPrincipalId: actor.id, executionPolicy: policy }); const id = newId("proposal"), at = now(); - const created = db.transaction(() => { - caller(actor); - const raced = db.query("SELECT * FROM schedule_proposals WHERE principal_id = ? AND request_key = ?").get(actor.id, key) as ProposalRow | null; - if (raced) return raced; - db.query(`INSERT INTO schedule_proposals (id, task_id, principal_id, request_key, input_hash, definition_json, - zone_source, execution_policy_json, root_identity, fingerprint, created_at, expires_at) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`).run(id, taskId, actor.id, key, inputHash, - canonicalJson(materialized.definition), materialized.zoneSource, canonicalJson(policy), identity, fingerprint, at, at + PROPOSAL_TTL_MS); - return null; - }).immediate(); + const created = await write("propose", actor, { + id, task_id: taskId, principal_id: actor.id, request_key: key, input_hash: inputHash, + definition_json: canonicalJson(materialized.definition), zone_source: materialized.zoneSource, + execution_policy_json: canonicalJson(policy), root_identity: identity, fingerprint, + created_at: at, expires_at: at + PROPOSAL_TTL_MS, + } satisfies ProposalRow); if (created) { if (created.input_hash !== inputHash) throw new ScheduleError("key_conflict", "This key was already used for a different schedule."); return { status: 200, proposal: proposalView(created) }; @@ -560,22 +540,7 @@ export function createScheduleService(db: Database, options: ScheduleServiceOpti if (!isOperator(actor)) throw new ScheduleError("unauthorized", "Only the operator can approve a schedule."); if (proposal.fingerprint !== fingerprint) throw new ScheduleError("definition_conflict", "The fingerprint does not match this proposal."); const policy = await currentPolicy(); - return db.transaction(() => { - caller(actor); - const existing = db.query("SELECT * FROM schedule_approvals WHERE proposal_id = ?").get(proposal.id) as ApprovalRow | null; - if (existing) return { approvalId: existing.id }; - if (proposal.expires_at <= now()) throw new ScheduleError("approval_expired", "This proposal expired; propose again with a new key."); - assertWritable(); - if (proposal.root_identity !== rootInfo().identity) throw new ScheduleError("definition_conflict", "The proposal belongs to another brain root."); - if (!usable(proposal.principal_id)) throw new ScheduleError("unauthorized", "The schedule's creator is no longer usable."); - if (canonicalJson(policy) !== proposal.execution_policy_json) - throw new ScheduleError("definition_conflict", "The host execution policy changed since this proposal."); - const id = newId("approval"); - db.query(`INSERT INTO schedule_approvals (id, proposal_id, approver_principal_id, approver_kind, channel, - fingerprint, nonce, approved_at) VALUES (?, ?, ?, ?, 'http', ?, ?, ?)`) - .run(id, proposal.id, actor.id, actor.kind, fingerprint, randomBytes(16).toString("hex"), now()); - return { approvalId: id }; - }).immediate(); + return write<{ approvalId: string }>("approve", actor, proposal, fingerprint, policy); } async function publish(principal: Principal, body: unknown): Promise<{ status: 200 | 201; created: boolean; task: ScheduleTask }> { @@ -612,38 +577,7 @@ export function createScheduleService(db: Database, options: ScheduleServiceOpti const policy = await currentPolicy(); assertWritable(); await assertNotIgnored(proposal.task_id); - const taskId = db.transaction(() => { - const at = now(); - caller(actor); - const raced = db.query("SELECT id FROM schedule_tasks WHERE proposal_id = ?").get(proposal.id) as { id: string } | null; - if (raced) return null; - if (proposal.expires_at <= at) throw new ScheduleError("approval_expired", "This approval expired before publication; propose again with a new key."); - assertWritable(); - const approver = usable(approval.approver_principal_id); - if (!approver || !isOperator(approver) || approver.kind !== approval.approver_kind) - throw new ScheduleError("unauthorized", "The approving operator is no longer usable."); - if (approval.fingerprint !== proposal.fingerprint || proposal.root_identity !== rootInfo().identity) - throw new ScheduleError("definition_conflict", "The approval no longer matches this proposal."); - if (canonicalJson(policy) !== proposal.execution_policy_json) - throw new ScheduleError("definition_conflict", "The host execution policy changed since approval."); - const recomputed = scheduleFingerprint({ definition, rootIdentity: proposal.root_identity, - creatorPrincipalId: proposal.principal_id, executionPolicy: JSON.parse(proposal.execution_policy_json) }); - if (recomputed !== proposal.fingerprint) throw new ScheduleError("definition_conflict", "The stored proposal does not match its fingerprint."); - // First-publication timing is rechecked; replay above never reaches here. - if (definition.when.kind === "at" && parseInstant(definition.when.at)! <= at) - throw new ScheduleError("invalid_request", "The one-off time has passed; propose a new future time."); - if (definition.when.kind === "cron" && definition.when.endAt !== null && parseInstant(definition.when.endAt)! <= at) - throw new ScheduleError("invalid_request", "The recurrence end has passed; propose a new schedule."); - db.query("UPDATE schedule_approvals SET consumed_at = ? WHERE id = ? AND consumed_at IS NULL").run(at, approval.id); - db.query(`INSERT INTO schedule_tasks (id, proposal_id, approval_id, creator_principal_id, root_identity, - definition_json, execution_policy_json, fingerprint, file_sha256, zone_source, created_at, state, - blocked_reason, publication, retirement, cancelled_at, evaluated_through, updated_at) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'publishing', NULL, 'pending', 'none', NULL, ?, ?)`) - .run(proposal.task_id, proposal.id, approval.id, proposal.principal_id, proposal.root_identity, - proposal.definition_json, proposal.execution_policy_json, proposal.fingerprint, sha256Hex(text), - proposal.zone_source, at, at, at); - return proposal.task_id; - }).immediate(); + const taskId = await write("publish", actor, proposal, approval, policy, definition, text); if (taskId === null) return publish(principal, body); await completePublication(taskId); return { status: 201, created: true, task: await taskView(taskRow(taskId)!) }; @@ -730,33 +664,7 @@ export function createScheduleService(db: Database, options: ScheduleServiceOpti throw new ScheduleError("invalid_request", "Cancellation needs {key}."); const row = SCHEDULE_ID.test(taskId) ? taskRow(taskId) : null; if (!row || !canSee(actor, row.creator_principal_id)) throw new ScheduleError("not_found", "No such scheduled task."); - const changed = db.transaction(() => { - caller(actor); - const receipt = db.query("SELECT task_id FROM schedule_cancel_receipts WHERE principal_id = ? AND request_key = ?") - .get(actor.id, key) as { task_id: string } | null; - if (receipt) { - if (receipt.task_id !== row.id) throw new ScheduleError("key_conflict", "This key was already used to cancel a different task."); - return false; - } - const current = taskRow(row.id)!; - const at = now(); - // Future/unstarted only: terminal one-offs and earlier cancellations stay as they are. - const cancellable = ["publishing", "active", "paused"].includes(current.state); - if (cancellable) { - db.query(`UPDATE schedule_tasks SET state = 'cancelled', blocked_reason = NULL, cancelled_at = ?, - retirement = 'pending', updated_at = ? WHERE id = ?`).run(at, at, row.id); - // Unstarted and stale-Action work is invalidated in the same transaction. - // A started attempt keeps its bounded snapshot; this is not a live stop. - db.query(`UPDATE schedule_occurrences SET state = 'cancelled', updated_at = ? - WHERE task_id = ? AND state IN ('queued', 'retrying', 'waiting_for_action')`).run(at, row.id); - // Their unclaimed Queue items go with them. A claim that has not - // started yet is refused by admission's start check instead. - dropUnstartedItems(db, row.id, at); - } - db.query("INSERT INTO schedule_cancel_receipts (principal_id, request_key, task_id, changed, created_at) VALUES (?, ?, ?, ?, ?)") - .run(actor.id, key, row.id, cancellable ? 1 : 0, at); - return cancellable; - }).immediate(); + const changed = await write("cancel", actor, row, key); await completeRetirement(row.id); const running = (db.query(`SELECT * FROM schedule_occurrences WHERE task_id = ? AND state IN ('running', 'unwinding') ORDER BY due_at`).all(row.id) as OccurrenceRow[]).map(occurrenceView); @@ -769,17 +677,9 @@ export function createScheduleService(db: Database, options: ScheduleServiceOpti * operator's own statement rather than a claimed matching fingerprint. The * root identity, and with it every approval fingerprint, is unchanged. */ - function adoptRestoredRoot(): void { + async function adoptRestoredRoot(): Promise { const info = rootInfo(); - db.transaction(() => { - const bound = db.query("SELECT root_path FROM schedule_root WHERE id = 1").get() as { root_path: string } | null; - if (bound?.root_path === info.path) return; - const restored = db.query("SELECT status, brain_root FROM inbox_recovery_state WHERE id = 1").get() as - { status: string; brain_root: string } | null; - if (restored?.status !== "ready" || restored.brain_root !== info.path) - throw new ScheduleError("unsupported_capability", "Schedule storage belongs to another brain root; no restore named this one."); - db.query("UPDATE schedule_root SET root_path = ? WHERE id = 1").run(info.path); - }).immediate(); + await write("adoptRestoredRoot"); info.matches = true; } @@ -820,39 +720,15 @@ export function createScheduleService(db: Database, options: ScheduleServiceOpti if (replay()) return { ok: true as const, changed: false, task: await taskView(taskRow(row.id)!) }; if (db.query("SELECT 1 FROM inbox_recovery_state WHERE status = 'pending'").get()) throw new ScheduleError("unsupported_capability", "The operational restore has not finished."); - adoptRestoredRoot(); + await adoptRestoredRoot(); // Journals frozen by the restore can finish now; this never dispatches. await reconcile(); // Only a reopening needs the host policy; a no-op still records its receipt. const policy = await currentPolicy().catch((error: unknown) => error as Error); const before = taskRow(row.id)!; const drifted = await fileDrifted(before); - const changed = db.transaction(() => { - const at = now(); - caller(actor); - if (replay()) return false; - const current = taskRow(row.id)!; - const reason = current.state === "paused" ? current.blocked_reason : null; - const resolvable = reason === "restore_pending" || reason === "unknown_effect"; - if (resolvable) { - if (drifted || current.updated_at !== before.updated_at || current.publication !== "published") - throw new ScheduleError("definition_conflict", "The definition file does not match the approved snapshot; cancel and create a new schedule."); - if (current.root_identity !== rootInfo().identity) - throw new ScheduleError("definition_conflict", "The task belongs to another brain root."); - if (!authorized(current)) throw new ScheduleError("unauthorized", "The schedule's creator or approving operator is no longer usable."); - if (policy instanceof Error) throw policy; - if (canonicalJson(policy) !== current.execution_policy_json) - throw new ScheduleError("definition_conflict", "The host execution policy changed since approval; cancel and create a new schedule."); - if (db.query(`SELECT 1 FROM schedule_occurrences WHERE task_id = ? AND state IN - ('queued', 'running', 'unwinding', 'waiting_for_action', 'retrying')`).get(current.id)) - throw new ScheduleError("definition_conflict", "The task still has outstanding work."); - db.query("UPDATE schedule_tasks SET state = ?, blocked_reason = NULL, updated_at = ? WHERE id = ?") - .run(finished(current, at) ? "expired" : "active", at, current.id); - } - db.query(`INSERT INTO schedule_reconciliations (principal_id, request_key, task_id, approver_kind, resolved_reason, changed, created_at) - VALUES (?, ?, ?, ?, ?, ?, ?)`).run(actor.id, key, row.id, actor.kind, resolvable ? reason : null, resolvable ? 1 : 0, at); - return resolvable; - }).immediate(); + const changed = await write("reopen", actor, row, key, before, drifted, + policy instanceof Error ? { error: encodeError(policy) } : { value: policy }); return { ok: true as const, changed, task: await taskView(taskRow(row.id)!) }; } diff --git a/packages/ui-server/tests/schedule-writes-offloop.test.ts b/packages/ui-server/tests/schedule-writes-offloop.test.ts new file mode 100644 index 0000000000..61719e37a7 --- /dev/null +++ b/packages/ui-server/tests/schedule-writes-offloop.test.ts @@ -0,0 +1,180 @@ +/** Schedule route writes keep their SQL/file order on the shared FULL writer. */ +import { expect, test } from "bun:test"; +import { existsSync, mkdirSync, readFileSync, writeFileSync } from "node:fs"; +import { join } from "node:path"; +import { Hono } from "hono"; +import type { AppEnv } from "../src/app-env.js"; +import { openSqliteWriter } from "../src/db/write-worker.js"; +import { createInboxStore } from "../src/inbox/store.js"; +import { createScheduleRoutes } from "../src/routes/schedules.js"; +import { createScheduleService, ScheduleError } from "../src/schedules/service.js"; +import { cronDefinition, POLICY, scheduleFixture } from "./helpers/schedule-fixture.js"; +import { watchWriteSites } from "./helpers/write-sites.js"; +import * as probe from "./fixtures/write-worker-probe.js"; + +const TABLES = ["schedule_proposals", "schedule_approvals", "schedule_tasks", "schedule_occurrences", + "schedule_cancel_receipts", "schedule_reconciliations", "schedule_root", "inbox_items", "inbox_changes"]; + +async function fixture() { + const f = scheduleFixture(), writer = openSqliteWriter(f.db); + const service = createScheduleService(f.db, { brainRoot: f.root, now: () => f.clock.now, + executionPolicy: () => POLICY, gitIgnored: () => false }, writer); + await service.ready; // rootInfo's boot-only write is deliberately outside the guard. + const propose = async (key = "ithaca") => (await service.propose(f.owner, { key, definition: cronDefinition() })).proposal; + const approved = async (key = "ithaca") => { + const proposal = await propose(key); + const { approvalId } = await service.approve(f.owner, proposal.id, { fingerprint: proposal.fingerprint, decision: "approve" }); + return { proposalId: proposal.id, approvalId, taskId: proposal.taskId }; + }; + const created = async (key = "ithaca") => { + const input = await approved(key); await service.publish(f.owner, { proposalId: input.proposalId, approvalId: input.approvalId }); return input; + }; + return { ...f, writer, service, propose, approved, created, + async close() { await writer.close(); f.close(); } }; +} + +for (const operation of ["propose", "approve", "publish", "cancel", "reopen", "quarantine", "reconcile", "adopt"] as const) { + test(`schedule ${operation} commits on the write worker at FULL, never on the app connection`, async () => { + const f = await fixture(); + try { + expect(f.writer.offloop).toBe(true); + const input = operation === "propose" ? null : operation === "approve" + ? { proposal: await f.propose() } : operation === "publish" || operation === "quarantine" + ? await f.approved() : await f.created(); + const taskId = input && "taskId" in input ? input.taskId : null; + const file = taskId ? join(f.root, "context/scheduled-tasks/definitions", `${taskId}.md`) : null; + if (operation === "quarantine") { + mkdirSync(join(f.root, "context/scheduled-tasks/definitions"), { recursive: true }); + writeFileSync(file!, "An unapproved definition occupies this path.\n"); + } + if (operation === "reopen" || operation === "adopt") + f.db.query("UPDATE schedule_tasks SET state = 'paused', blocked_reason = 'restore_pending' WHERE id = ?").run(taskId); + if (operation === "adopt") { + f.db.query("UPDATE schedule_root SET root_path = ? WHERE id = 1").run(join(f.root, "old")); + f.db.query("INSERT INTO inbox_recovery_state (id, status, brain_root, snapshot_checksum, restored_at) VALUES (1, 'ready', ?, 'odysseus', ?)").run(f.root, f.clock.now); + } + if (operation === "reconcile") writeFileSync(file!, "Drifted definition.\n"); + const sites = await watchWriteSites(f.db, f.writer, TABLES); + if (operation === "propose") { + const proposal = await f.propose(); + expect(f.db.query("SELECT id FROM schedule_proposals WHERE id = ?").get(proposal.id), "SCHEDULE_PROPOSAL_COMMITTED").not.toBeNull(); + } else if (operation === "approve") { + const proposal = (input as { proposal: Awaited> }).proposal; + const { approvalId } = await f.service.approve(f.owner, proposal.id, { fingerprint: proposal.fingerprint, decision: "approve" }); + expect(f.db.query("SELECT id FROM schedule_approvals WHERE id = ?").get(approvalId), "SCHEDULE_APPROVAL_COMMITTED").not.toBeNull(); + } else if (operation === "publish") { + expect((await f.service.publish(f.owner, { proposalId: (input as { proposalId: string }).proposalId, approvalId: (input as { approvalId: string }).approvalId })).task.state, "SCHEDULE_PUBLICATION_COMMITTED").toBe("active"); + expect(readFileSync(file!, "utf8"), "SCHEDULE_APPROVED_FILE_PUBLISHED").toContain("Ithaca"); + } else if (operation === "cancel") { + expect((await f.service.cancel(f.owner, taskId!, { key: "cancel" })).task, "SCHEDULE_CANCEL_AND_RETIRE_COMMITTED") + .toMatchObject({ state: "cancelled", compensationPending: false }); + expect(existsSync(file!), "SCHEDULE_FILE_RETIRED").toBe(false); + } else if (operation === "reopen" || operation === "adopt") { + expect((await f.service.reopen(f.owner, taskId!, { key: "reopen", decision: "reopen" })).task.state, "SCHEDULE_REOPEN_COMMITTED").toBe("active"); + } else if (operation === "quarantine") { + let failure: unknown; + try { await f.service.publish(f.owner, { proposalId: (input as { proposalId: string }).proposalId, approvalId: (input as { approvalId: string }).approvalId }); } catch (error) { failure = error; } + expect(failure, "SCHEDULE_FILE_CONFLICT_CLASS_KEPT").toBeInstanceOf(ScheduleError); + expect(failure, "SCHEDULE_FILE_CONFLICT_CODE_KEPT").toMatchObject({ code: "definition_conflict", status: 409 }); + expect(f.db.query("SELECT publication FROM schedule_tasks WHERE id = ?").get(taskId), "SCHEDULE_QUARANTINE_COMMITTED") + .toEqual({ publication: "quarantined" }); + } else { + await f.service.reconcile(); + expect(f.db.query("SELECT state FROM schedule_tasks WHERE id = ?").get(taskId), "SCHEDULE_DRIFT_PAUSE_COMMITTED").toEqual({ state: "paused" }); + } + expect((await sites.log()).length, `SCHEDULE_${operation.toUpperCase()}_WRITES_ON_WORKER`).toBeGreaterThan(0); + } finally { await f.close(); } + }); +} + +test("schedule publication observes Git between the pending receipt and activation, and replay sees committed writes", async () => { + const f = await fixture(); + try { + const input = await f.approved(); + const observed: string[] = []; + const service = createScheduleService(f.db, { brainRoot: f.root, now: () => f.clock.now, executionPolicy: () => POLICY, + gitIgnored: () => { + const row = f.db.query("SELECT publication FROM schedule_tasks WHERE id = ?").get(input.taskId) as { publication: string } | null; + observed.push(row?.publication ?? "absent"); + return false; + } }, f.writer); + await service.ready; + const sites = await watchWriteSites(f.db, f.writer, TABLES); + const result = await service.publish(f.owner, { proposalId: input.proposalId, approvalId: input.approvalId }); + expect(observed, "SCHEDULE_GIT_CHECK_BETWEEN_COMMITS").toEqual(["absent", "absent", "pending", "pending"]); + expect(result.task.state).toBe("active"); + expect((await service.publish(f.owner, { proposalId: input.proposalId, approvalId: input.approvalId })).created, "SCHEDULE_PUBLICATION_REPLAY_SEES_COMMIT").toBe(false); + expect(await sites.log(), "SCHEDULE_APPROVAL_CONSUMPTION_AND_PUBLICATION_ORDER") + .toEqual(["schedule_approvals", "schedule_tasks", "schedule_tasks"]); + } finally { await f.close(); } +}); + +test("schedule worker refusals keep their class, code and status through the HTTP route", async () => { + const f = await fixture(); + try { + const proposal = await f.propose(); + const held = f.writer.channel(probe, { table: "odysseus_log" }); await held.ready; + const sites = await watchWriteSites(f.db, f.writer, TABLES); + const slow = held.call("slow", "Calypso", 150); + // The policy callback runs before submission; expiry is checked inside the worker transaction. + const service = createScheduleService(f.db, { brainRoot: f.root, now: () => f.clock.now, + executionPolicy: () => { f.clock.now += 16 * 60_000; return POLICY; }, gitIgnored: () => false }, f.writer); + await service.ready; + const app = new Hono().use("*", async (c, next) => { c.set("principal", f.owner); await next(); }); + app.route("/api", createScheduleRoutes(service)); + const response = await app.request(`/api/schedules/proposals/${proposal.id}/approve`, { method: "POST", + headers: { "content-type": "application/json" }, body: JSON.stringify({ fingerprint: proposal.fingerprint, decision: "approve" }) }); + expect(response.status, "SCHEDULE_WORKER_HTTP_REFUSAL_STATUS").toBe(409); + expect(await response.json(), "SCHEDULE_WORKER_HTTP_REFUSAL_CLASS").toMatchObject({ error: { code: "approval_expired" } }); + await slow; + expect(await sites.log(), "SCHEDULE_REFUSED_APPROVAL_ROLLED_BACK").toEqual([]); + } finally { await f.close(); } +}); + +test("a schedule cancelled during publication retires the late file after its activation step commits", async () => { + const f = await fixture(); + try { + const input = await f.approved(); + let cancelled = false; + const service = createScheduleService(f.db, { brainRoot: f.root, now: () => f.clock.now, executionPolicy: () => POLICY, + gitIgnored: async () => { + const row = f.db.query("SELECT publication FROM schedule_tasks WHERE id = ?").get(input.taskId) as { publication: string } | null; + if (row?.publication === "pending" && !cancelled) { + cancelled = true; + await f.service.cancel(f.owner, input.taskId, { key: "cancel-during-publication" }); + } + return false; + } }, f.writer); + await service.ready; + const sites = await watchWriteSites(f.db, f.writer, TABLES); + const result = await service.publish(f.owner, { proposalId: input.proposalId, approvalId: input.approvalId }); + expect(result.task, "SCHEDULE_LATE_PUBLICATION_RETIRED").toMatchObject({ state: "cancelled", compensationPending: false }); + expect(existsSync(join(f.root, "context/scheduled-tasks/definitions", `${input.taskId}.md`)), "SCHEDULE_LATE_ACTIVE_FILE_REMOVED").toBe(false); + expect(await sites.log(), "SCHEDULE_CANCELLATION_PUBLICATION_RETIREMENT_ORDER").toEqual([ + "schedule_approvals", "schedule_tasks", "schedule_tasks", "schedule_cancel_receipts", + "schedule_tasks", "schedule_tasks", "schedule_tasks", + ]); + } finally { await f.close(); } +}); + +test("schedule cancellation drops unstarted Queue work and its change records in the same worker transaction", async () => { + const f = await fixture(); + try { + const input = await f.created(); + const store = createInboxStore(f.db, { now: () => f.clock.now }); + store.ingest({ threadId: "ithaca-thread", itemId: "ithaca-item", dedupKey: "ithaca-key", stagingId: "ithaca-stage", + source: "cli", stakes: 2, expiresAt: f.clock.now + 86_400_000 }); + f.db.query(`INSERT INTO schedule_occurrences (id, task_id, due_at, expires_at, state, operations_used, max_operations, created_at, updated_at) + VALUES ('occ_ithaca', ?, ?, ?, 'queued', 0, 3, ?, ?)`) + .run(input.taskId, f.clock.now, f.clock.now + 86_400_000, f.clock.now, f.clock.now); + f.db.query("INSERT INTO schedule_occurrence_items (occurrence_id, item_id, sequence, created_at) VALUES ('occ_ithaca', 'ithaca-item', 1, ?)").run(f.clock.now); + const sites = await watchWriteSites(f.db, f.writer, [...TABLES, "inbox_threads"]); + await f.service.cancel(f.owner, input.taskId, { key: "cancel-queue" }); + expect(store.getItem("ithaca-item")?.status, "SCHEDULE_QUEUE_DROP_COMMITTED").toBe("dropped"); + expect(f.db.query("SELECT state FROM schedule_occurrences WHERE id = 'occ_ithaca'").get(), "SCHEDULE_OCCURRENCE_CANCEL_COMMITTED") + .toEqual({ state: "cancelled" }); + const log = await sites.log(); + expect(log, "SCHEDULE_QUEUE_CHANGE_RECORDS_ON_WORKER").toContain("inbox_changes"); + expect(log.indexOf("schedule_cancel_receipts"), "SCHEDULE_CANCELLATION_RECEIPT_AFTER_QUEUE_DROP").toBeGreaterThan(log.indexOf("inbox_items")); + } finally { await f.close(); } +});