Make document lifecycle fan-out durable and atomic

Pre-v47 `createDocumentWithLifecycle` wrapped each hook in its own
try/catch and always returned success, so a failure in any of the
downstream hooks (thread event, checkpoint approval, deliverable
review, automation dispatch, reviewer wake) left a live document row
behind with a silently incomplete lifecycle.

Two invariants now hold:

- DB-writing hooks (thread integration, checkpoint approval,
  deliverable review) run inside the same `BEGIN IMMEDIATE`
  transaction as `saveDocument`. A throw rolls back the document row
  along with every partial write, so an operator never sees a
  persisted document with a missing thread or skipped approval.

- Cross-process side effects (automation dispatch, reviewer wake) are
  enqueued as `document_lifecycle_events` rows inside the same
  transaction and processed post-commit. Success stamps
  `status = 'succeeded'`; failure stamps `status = 'failed'` with
  `last_error`, so partial lifecycle state is observable via
  `listFailedForCompany` instead of disappearing into the log tail.

Migration v47 installs the ledger; `buildDocumentLifecycleEventMethods`
exposes the enqueue/running/succeeded/failed transitions; the lifecycle
service drives both.
This commit is contained in:
ZacharyZhang-NY committed 2026-04-19 15:53:55 -04:00
1 parent a36884360a
commit edbebe6109
5 files changed
+326 -117

No files matched your search

@@ -0,0 +1,131 @@
import { randomUUID } from "node:crypto";
import type { DatabaseSync, SQLInputValue } from "node:sqlite";
import type { ReadAll } from "./helpers";
import { nowIso } from "./helpers";
/**
* Document-lifecycle event row shape returned by listers.
* `status` moves `pending -> running -> (succeeded | failed)`. `attempts`
* tracks how many processor passes have tried this hook so a sweeper can
* bound retries without losing the ledger.
*/
export interface DocumentLifecycleEventRow {
id: string;
companyId: string;
documentId: string;
hookType: string;
status: "pending" | "running" | "succeeded" | "failed";
attempts: number;
lastError: string | null;
payloadJson: string;
createdAt: string;
updatedAt: string;
completedAt: string | null;
}
export interface EnqueueInput {
companyId: string;
documentId: string;
hookType: string;
payloadJson?: string;
}
export function buildDocumentLifecycleEventMethods(
getDb: () => DatabaseSync,
readAll: ReadAll,
) {
return {
enqueue(input: EnqueueInput): string {
const id = randomUUID();
const ts = nowIso();
getDb().prepare(
`INSERT INTO document_lifecycle_events
(id, company_id, document_id, hook_type, status, attempts, last_error, payload_json, created_at, updated_at, completed_at)
VALUES (?, ?, ?, ?, 'pending', 0, NULL, ?, ?, ?, NULL)`,
).run(
id,
input.companyId,
input.documentId,
input.hookType,
input.payloadJson ?? "{}",
ts,
ts,
);
return id;
},
markRunning(id: string): void {
const ts = nowIso();
getDb().prepare(
"UPDATE document_lifecycle_events SET status = 'running', attempts = attempts + 1, updated_at = ? WHERE id = ?",
).run(ts, id);
},
markSucceeded(id: string): void {
const ts = nowIso();
getDb().prepare(
"UPDATE document_lifecycle_events SET status = 'succeeded', last_error = NULL, updated_at = ?, completed_at = ? WHERE id = ?",
).run(ts, ts, id);
},
markFailed(id: string, error: string): void {
const ts = nowIso();
getDb().prepare(
"UPDATE document_lifecycle_events SET status = 'failed', last_error = ?, updated_at = ? WHERE id = ?",
).run(error.slice(0, 2000), ts, id);
},
/**
* Claim the batch of pending events for a document so the caller can
* drain them post-commit. Ordered by `created_at` so the fan-out
* reproduces the enqueue order.
*/
listPendingForDocument(documentId: string): DocumentLifecycleEventRow[] {
return readRows(readAll,
"SELECT id, company_id, document_id, hook_type, status, attempts, last_error, payload_json, created_at, updated_at, completed_at FROM document_lifecycle_events WHERE document_id = ? AND status = 'pending' ORDER BY created_at ASC",
documentId,
);
},
listFailedForCompany(companyId: string, limit = 200): DocumentLifecycleEventRow[] {
return readRows(readAll,
"SELECT id, company_id, document_id, hook_type, status, attempts, last_error, payload_json, created_at, updated_at, completed_at FROM document_lifecycle_events WHERE company_id = ? AND status = 'failed' ORDER BY updated_at DESC LIMIT ?",
companyId,
limit,
);
},
};
}
function readRows(
readAll: ReadAll,
sql: string,
...params: SQLInputValue[]
): DocumentLifecycleEventRow[] {
const raw = readAll<{
id: string;
company_id: string;
document_id: string;
hook_type: string;
status: "pending" | "running" | "succeeded" | "failed";
attempts: number;
last_error: string | null;
payload_json: string;
created_at: string;
updated_at: string;
completed_at: string | null;
}>(sql, ...params);
return raw.map((row) => ({
id: row.id,
companyId: row.company_id,
documentId: row.document_id,
hookType: row.hook_type,
status: row.status,
attempts: row.attempts,
lastError: row.last_error,
payloadJson: row.payload_json,
createdAt: row.created_at,
updatedAt: row.updated_at,
completedAt: row.completed_at,
}));
}
+53
View File
@@ -0,0 +1,53 @@
import log from "electron-log/main.js";
import type { Migration, MigrationDeps } from "./types";
export function createMigrationsV47V47(deps: MigrationDeps): Migration[] {
const { db } = deps;
return [
{
// Durable fan-out ledger for document-creation side effects.
//
// Before v47 `createDocumentWithLifecycle` wrapped every downstream
// hook (automation dispatch, thread integration, checkpoint engine,
// reviewer wake) in an individual `try/catch` and always returned
// success. A failure in any single hook left the document row live
// with a silently incomplete lifecycle — thread views, approvals,
// or reviewer wakes disappeared without trace.
//
// The ledger is the tracked-state surface the operator reads to
// reason about lifecycle coverage: one row per hook per document
// is inserted inside the same transaction as the document write.
// The processor advances `status` from `pending` to `succeeded` or
// `failed`, stamps `last_error` on failure, and `attempts` lets the
// sweeper retry within a bounded policy.
version: 47,
name: "document_lifecycle_events_ledger",
fn: () => {
db.exec(`
CREATE TABLE IF NOT EXISTS document_lifecycle_events (
id TEXT PRIMARY KEY,
company_id TEXT NOT NULL,
document_id TEXT NOT NULL,
hook_type TEXT NOT NULL,
status TEXT NOT NULL,
attempts INTEGER NOT NULL DEFAULT 0,
last_error TEXT,
payload_json TEXT NOT NULL DEFAULT '{}',
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
completed_at TEXT
)
`);
db.exec(`
CREATE INDEX IF NOT EXISTS idx_document_lifecycle_events_company
ON document_lifecycle_events(company_id, status, updated_at)
`);
db.exec(`
CREATE INDEX IF NOT EXISTS idx_document_lifecycle_events_document
ON document_lifecycle_events(document_id, hook_type)
`);
log.info("[migration v47] document_lifecycle_events ledger installed");
},
},
];
}
+2
View File
@@ -11,6 +11,7 @@ import { createMigrationsV43V43 } from "./migrations/v43-v43";
import { createMigrationsV44V44 } from "./migrations/v44-v44";
import { createMigrationsV45V45 } from "./migrations/v45-v45";
import { createMigrationsV46V46 } from "./migrations/v46-v46";
import { createMigrationsV47V47 } from "./migrations/v47-v47";
import {
columnExists,
nowIso,
@@ -197,6 +198,7 @@ export function runMigrations(
...createMigrationsV44V44(deps),
...createMigrationsV45V45(deps),
...createMigrationsV46V46(deps),
...createMigrationsV47V47(deps),
];
for (const migration of migrations) {