diff --git a/DECISIONS.md b/DECISIONS.md index 258d8237..e66aa1ab 100644 --- a/DECISIONS.md +++ b/DECISIONS.md @@ -273,3 +273,4 @@ One line per decision: `[YYYY-MM-DD] : `. Appended by agents and [2026-07-21] Card-descriptor normalization keys on the pre-star merchant segment (post-star for processor prefixes) plus a token_subset match tier, instead of the deferred AI descriptor normalization (data_quality_master Appendix B): deterministic, mirrors into normalize_counterparty_key() so ledger-context template joins stay exact, and fixes the reported Anthropic no-signal case with no new infrastructure. Merchant history now falls back to description because card purchases never carry merchant_name. [2026-07-22] Issue #313 fix limited to the meals warning; left "Representationsgåvor max 180 kr" on the gåvor line untouched: scope rule (only the inverted-VAT claim and repealed ML 8:9 reference), even though the swedish-vat skill lists 300 SEK as the representationsgåvor base; flagged as follow-up in the PR. [2026-07-22] LEGACY_DISCOVERY_HOSTS drift guard (#1093) is an exported validateLegacyDiscoveryHosts() returning a violations list, exercised only by a unit test that pins the registered prod config (app.accounted.se canonical + app.gnubok.se SKV pin), not a startup assertion: CI does not set the prod env vars, so a runtime assertion would either no-op in CI or crash self-hosted deploys with different domains; the test-pinned constants make any allowlist or pin change a deliberate, reviewed edit. +[2026-07-22] Stuck-committing recovery sweep (#843) rejects rows without positive evidence instead of reverting to pending, and only three op types (categorize_transaction, link_transaction_journal_entry, match_transaction_invoice) can recover to committed: no generic side-effect -> pending_op linkage exists yet (that is #842's posted-ids work), so evidence is limited to types whose params identify a target row with an unambiguous posted state; reverting to pending risks re-executing side-effects that posted without a trace (duplicate entries/emails). diff --git a/app/api/pending-operations/expire/cron/__tests__/route.test.ts b/app/api/pending-operations/expire/cron/__tests__/route.test.ts index 4987f63f..5701ade7 100644 --- a/app/api/pending-operations/expire/cron/__tests__/route.test.ts +++ b/app/api/pending-operations/expire/cron/__tests__/route.test.ts @@ -3,6 +3,10 @@ * staged operations are auto-rejected with the commit dispatcher's * result_data shape ({ auto_rejected: true, reason: 'expired' }) so the * /pending UI can render them as "Utgick automatiskt". + * + * The same run also invokes the stuck-'committing' recovery sweep (#843, + * lib/pending-operations/recover-stuck-committing.ts), mocked here: its + * decision logic has its own unit + pg-real tests. */ import { describe, it, expect, vi, beforeEach } from 'vitest' import { NextResponse } from 'next/server' @@ -11,6 +15,15 @@ vi.mock('@/lib/auth/cron', () => ({ verifyCronSecret: vi.fn(() => null), })) +vi.mock('@/lib/pending-operations/recover-stuck-committing', () => ({ + recoverStuckCommittingOperations: vi.fn(async () => ({ + scanned: 0, + committed: 0, + rejected: 0, + skipped: 0, + })), +})) + interface FilterCall { method: string args: unknown[] @@ -57,6 +70,7 @@ vi.mock('@/lib/supabase/server', () => ({ import { GET } from '../route' import { verifyCronSecret } from '@/lib/auth/cron' import { createServiceClient } from '@/lib/supabase/server' +import { recoverStuckCommittingOperations } from '@/lib/pending-operations/recover-stuck-committing' function cronRequest(): Request { return new Request('http://localhost:3000/api/pending-operations/expire/cron') @@ -117,7 +131,45 @@ describe('GET /api/pending-operations/expire/cron', () => { const response = await GET(cronRequest()) const json = await response.json() - expect(json).toEqual({ success: true, expired: 0, cutoff: expect.any(String) }) + expect(json).toEqual({ + success: true, + expired: 0, + cutoff: expect.any(String), + recovery: { scanned: 0, committed: 0, rejected: 0, skipped: 0 }, + }) + }) + + it('invokes the stuck-committing recovery sweep with the service client', async () => { + updateResults = [{ data: [], error: null }] + vi.mocked(recoverStuckCommittingOperations).mockResolvedValueOnce({ + scanned: 3, + committed: 1, + rejected: 2, + skipped: 0, + }) + + const response = await GET(cronRequest()) + const json = await response.json() + + expect(vi.mocked(recoverStuckCommittingOperations)).toHaveBeenCalledTimes(1) + const [client, opts] = vi.mocked(recoverStuckCommittingOperations).mock.calls[0] + expect(client).toBe(vi.mocked(createServiceClient).mock.results[0].value) + expect(opts?.log).toBeDefined() + expect(json.recovery).toEqual({ scanned: 3, committed: 1, rejected: 2, skipped: 0 }) + }) + + it('a recovery sweep failure never masks a successful expiry pass', async () => { + updateResults = [{ data: [{ id: 'op-1', company_id: 'c-1' }], error: null }] + vi.mocked(recoverStuckCommittingOperations).mockRejectedValueOnce(new Error('sweep boom')) + + const response = await GET(cronRequest()) + const json = await response.json() + + expect(response.status).toBe(200) + expect(json.success).toBe(true) + expect(json.expired).toBe(1) + // Static marker only: the raw error stays in the structured log. + expect(json.recovery).toEqual({ error: 'recovery_sweep_failed' }) }) it('returns 401 without touching the database when cron auth fails', async () => { @@ -129,6 +181,7 @@ describe('GET /api/pending-operations/expire/cron', () => { expect(response.status).toBe(401) expect(vi.mocked(createServiceClient)).not.toHaveBeenCalled() + expect(vi.mocked(recoverStuckCommittingOperations)).not.toHaveBeenCalled() }) it('returns an error envelope when the update fails', async () => { diff --git a/app/api/pending-operations/expire/cron/route.ts b/app/api/pending-operations/expire/cron/route.ts index a7e1a3ab..b7184440 100644 --- a/app/api/pending-operations/expire/cron/route.ts +++ b/app/api/pending-operations/expire/cron/route.ts @@ -2,12 +2,20 @@ import { createServiceClient } from '@/lib/supabase/server' import { NextResponse } from 'next/server' import { withCronContext } from '@/lib/api/with-cron-context' import { errorResponse } from '@/lib/errors/get-structured-error' +import { recoverStuckCommittingOperations } from '@/lib/pending-operations/recover-stuck-committing' /** * GET /api/pending-operations/expire/cron, daily 02:30 UTC. * - * Auto-rejects staged operations that have sat at status='pending' for more - * than 30 days. AI agents stage operations for human review; when the chat + * Two sweeps per run: + * + * 1. Expiry: auto-rejects staged operations that have sat at status='pending' + * for more than 30 days. + * 2. Recovery (#843): drives rows stuck in status='committing' beyond the + * safe threshold to a terminal status; see + * lib/pending-operations/recover-stuck-committing.ts for the semantics. + * + * AI agents stage operations for human review; when the chat * session is abandoned the proposal would otherwise linger in the worklist * forever, asking the user to Godkänn/Avvisa something whose context they no * longer remember. A 30-day-old proposal has lost its context regardless of @@ -58,5 +66,18 @@ export const GET = withCronContext('cron.pending_operations_expire', async (_req cutoff: cutoff.toISOString(), }) - return NextResponse.json({ success: true, expired, cutoff: cutoff.toISOString() }) + // Recovery sweep for rows stuck in 'committing' (#843). Isolated so a + // recovery failure never masks a successful expiry pass: the expiry update + // above has already been applied at this point. + let recovery: Record + try { + recovery = { ...(await recoverStuckCommittingOperations(supabase, { log: ctx.log })) } + } catch (err) { + // Detail goes to the structured log only; the response carries a static + // marker so an operator sees the run partially failed. + ctx.log.error('pending_op_recovery sweep failed', err as Error) + recovery = { error: 'recovery_sweep_failed' } + } + + return NextResponse.json({ success: true, expired, cutoff: cutoff.toISOString(), recovery }) }) diff --git a/lib/pending-operations/__tests__/recover-stuck-committing.test.ts b/lib/pending-operations/__tests__/recover-stuck-committing.test.ts new file mode 100644 index 00000000..fd3c634c --- /dev/null +++ b/lib/pending-operations/__tests__/recover-stuck-committing.test.ts @@ -0,0 +1,389 @@ +/** + * Unit tests for the stuck-'committing' recovery sweep (issue #843). + * + * The transition legality against real triggers is covered by + * tests/pg/pending-operations-committing-recovery.pg.test.ts; these tests pin + * the decision logic: row selection filters, per-type evidence probes, the + * conservative rejected-by-default outcome, CAS guards, and the metric-style + * pending_op_recovery log line. + */ +import { describe, it, expect, vi, beforeEach } from 'vitest' +import type { SupabaseClient } from '@supabase/supabase-js' +import { + STUCK_COMMITTING_THRESHOLD_MINUTES, + buildRecoveryUpdate, + findPostedEvidence, + recoverStuckCommittingOperations, + type StuckCommittingRow, +} from '../recover-stuck-committing' +import type { Logger } from '@/lib/logger' + +interface FilterCall { + method: string + args: unknown[] +} + +interface QueryCapture { + kind: 'from' | 'rpc' + target: string + rpcArgs?: unknown + payload?: Record + filters: FilterCall[] +} + +function createCapturingSupabase() { + const captures: QueryCapture[] = [] + const results: Array<{ data: unknown; error: unknown }> = [] + + const buildChain = (capture: QueryCapture) => { + const result = results.shift() ?? { data: null, error: null } + const chain: Record = {} + const record = + (method: string) => + (...args: unknown[]) => { + if (method === 'update') { + capture.payload = args[0] as Record + } else { + capture.filters.push({ method, args }) + } + return chain + } + for (const method of ['select', 'update', 'eq', 'lt', 'order', 'range', 'maybeSingle']) { + chain[method] = vi.fn(record(method)) + } + chain.then = (resolve: (v: unknown) => unknown) => Promise.resolve(result).then(resolve) + return chain + } + + const supabase = { + from: vi.fn((table: string) => { + const capture: QueryCapture = { kind: 'from', target: table, filters: [] } + captures.push(capture) + return buildChain(capture) + }), + rpc: vi.fn((name: string, args: unknown) => { + const capture: QueryCapture = { kind: 'rpc', target: name, rpcArgs: args, filters: [] } + captures.push(capture) + return buildChain(capture) + }), + } as unknown as SupabaseClient + + return { + supabase, + captures, + enqueue(result: { data?: unknown; error?: unknown }) { + results.push({ data: result.data ?? null, error: result.error ?? null }) + }, + } +} + +function createLogSpy(): { log: Logger; calls: Array<{ level: string; args: unknown[] }> } { + const calls: Array<{ level: string; args: unknown[] }> = [] + const log = { + info: vi.fn((...args: unknown[]) => calls.push({ level: 'info', args })), + warn: vi.fn((...args: unknown[]) => calls.push({ level: 'warn', args })), + error: vi.fn((...args: unknown[]) => calls.push({ level: 'error', args })), + child: vi.fn(), + } as unknown as Logger + return { log, calls } +} + +function makeRow(overrides: Partial = {}): StuckCommittingRow { + return { + id: 'op-1', + company_id: 'company-1', + operation_type: 'create_customer', + params: {}, + updated_at: '2026-07-22T00:00:00.000Z', + ...overrides, + } +} + +beforeEach(() => { + vi.clearAllMocks() +}) + +describe('buildRecoveryUpdate', () => { + it('finalizes to committed with the recovered marker when evidence exists', () => { + const row = makeRow({ operation_type: 'categorize_transaction' }) + const update = buildRecoveryUpdate(row, 'transaction_booked', '2026-07-22T02:30:00.000Z') + + expect(update.status).toBe('committed') + expect(update.resolved_at).toBe('2026-07-22T02:30:00.000Z') + expect(update.result_data.recovered).toBe(true) + expect(update.result_data.recovery).toMatchObject({ + reason: 'stuck_committing', + evidence: 'transaction_booked', + stuck_since: row.updated_at, + swept_at: '2026-07-22T02:30:00.000Z', + }) + }) + + it('rejects with an explanation when no evidence exists, never back to pending', () => { + const row = makeRow() + const update = buildRecoveryUpdate(row, null, '2026-07-22T02:30:00.000Z') + + expect(update.status).toBe('rejected') + expect(update.result_data.auto_rejected).toBe(true) + // Distinct from 'expired' so the UI's "Utgick automatiskt" badge (strict + // on reason === 'expired') never claims recovery rows. + expect(update.result_data.reason).toBe('stuck_committing') + expect(update.result_data.recovery).toMatchObject({ + evidence: null, + stuck_since: row.updated_at, + }) + expect((update.result_data.recovery as { note: string }).note).toMatch(/re-staging/) + }) +}) + +describe('findPostedEvidence', () => { + it('categorize_transaction: booked target transaction is positive evidence', async () => { + const { supabase, captures, enqueue } = createCapturingSupabase() + enqueue({ data: true }) + + const evidence = await findPostedEvidence( + supabase, + makeRow({ + operation_type: 'categorize_transaction', + params: { transaction_id: 'tx-1' }, + }), + ) + + expect(evidence).toBe('transaction_booked') + expect(captures).toHaveLength(1) + expect(captures[0]).toMatchObject({ + kind: 'rpc', + target: 'is_transaction_booked', + rpcArgs: { p_transaction_id: 'tx-1' }, + }) + }) + + it('categorize_transaction: unbooked target means no evidence', async () => { + const { supabase, enqueue } = createCapturingSupabase() + enqueue({ data: false }) + + const evidence = await findPostedEvidence( + supabase, + makeRow({ + operation_type: 'categorize_transaction', + params: { transaction_id: 'tx-1' }, + }), + ) + + expect(evidence).toBeNull() + }) + + it('categorize_transaction with allow_duplicate skips the probe: booked proves nothing', async () => { + const { supabase, captures } = createCapturingSupabase() + + const evidence = await findPostedEvidence( + supabase, + makeRow({ + operation_type: 'categorize_transaction', + params: { transaction_id: 'tx-1', allow_duplicate: true }, + }), + ) + + expect(evidence).toBeNull() + expect(captures).toHaveLength(0) + }) + + it('link_transaction_journal_entry: the exact tx+entry pair must be linked', async () => { + const { supabase, captures, enqueue } = createCapturingSupabase() + enqueue({ data: null }) // transactions.journal_entry_id miss + enqueue({ data: { id: 'link-1' } }) // transaction_voucher_links hit + + const evidence = await findPostedEvidence( + supabase, + makeRow({ + operation_type: 'link_transaction_journal_entry', + params: { transaction_id: 'tx-1', journal_entry_id: 'je-1' }, + }), + ) + + expect(evidence).toBe('transaction_linked_to_target_entry') + expect(captures[0].target).toBe('transactions') + expect(captures[1].target).toBe('transaction_voucher_links') + expect(captures[1].filters).toEqual( + expect.arrayContaining([ + { method: 'eq', args: ['company_id', 'company-1'] }, + { method: 'eq', args: ['transaction_id', 'tx-1'] }, + { method: 'eq', args: ['journal_entry_id', 'je-1'] }, + ]), + ) + }) + + it('match_transaction_invoice: an invoice_payments row for the exact pair is evidence', async () => { + const { supabase, captures, enqueue } = createCapturingSupabase() + enqueue({ data: { id: 'payment-1' } }) + + const evidence = await findPostedEvidence( + supabase, + makeRow({ + operation_type: 'match_transaction_invoice', + params: { transaction_id: 'tx-1', invoice_id: 'inv-1' }, + }), + ) + + expect(evidence).toBe('invoice_payment_recorded') + expect(captures[0].target).toBe('invoice_payments') + }) + + it('types without a reliable probe return null without touching the database', async () => { + const { supabase, captures } = createCapturingSupabase() + + for (const operationType of ['create_customer', 'send_invoice', 'lock_period', 'create_voucher']) { + const evidence = await findPostedEvidence( + supabase, + makeRow({ operation_type: operationType, params: { transaction_id: 'tx-1' } }), + ) + expect(evidence).toBeNull() + } + expect(captures).toHaveLength(0) + }) + + it('throws on probe errors so the caller skips instead of rejecting', async () => { + const { supabase, enqueue } = createCapturingSupabase() + enqueue({ error: { message: 'connection reset' } }) + + await expect( + findPostedEvidence( + supabase, + makeRow({ + operation_type: 'categorize_transaction', + params: { transaction_id: 'tx-1' }, + }), + ), + ).rejects.toThrow(/connection reset/) + }) +}) + +describe('recoverStuckCommittingOperations', () => { + it('lists committing rows older than the threshold and drives them terminal with a CAS', async () => { + const { supabase, captures, enqueue } = createCapturingSupabase() + const { log, calls } = createLogSpy() + const now = new Date('2026-07-22T02:30:00.000Z') + + const bookedOp = makeRow({ + id: 'op-booked', + operation_type: 'categorize_transaction', + params: { transaction_id: 'tx-1' }, + }) + const unknownOp = makeRow({ id: 'op-unknown', operation_type: 'create_customer' }) + + enqueue({ data: [bookedOp, unknownOp] }) // listing page + enqueue({ data: true }) // is_transaction_booked probe + enqueue({ data: { id: 'op-booked' } }) // CAS finalize committed + enqueue({ data: { id: 'op-unknown' } }) // CAS finalize rejected + + const summary = await recoverStuckCommittingOperations(supabase, { log, now }) + + expect(summary).toEqual({ scanned: 2, committed: 1, rejected: 1, skipped: 0 }) + + // Listing: status CAS filter + threshold on updated_at (the claim + // timestamp, bumped by the updated_at trigger on the pending->committing + // CAS) + stable order for fetchAllRows pagination. + const listing = captures[0] + expect(listing.target).toBe('pending_operations') + expect(listing.filters).toEqual( + expect.arrayContaining([ + { method: 'eq', args: ['status', 'committing'] }, + { + method: 'lt', + args: [ + 'updated_at', + new Date( + now.getTime() - STUCK_COMMITTING_THRESHOLD_MINUTES * 60_000, + ).toISOString(), + ], + }, + ]), + ) + expect(listing.filters.some((f) => f.method === 'order')).toBe(true) + + // Committed finalize: CAS-guarded on status='committing'. + const committedWrite = captures[2] + expect(committedWrite.payload).toMatchObject({ status: 'committed' }) + expect((committedWrite.payload!.result_data as Record).recovered).toBe(true) + expect(committedWrite.filters).toEqual( + expect.arrayContaining([ + { method: 'eq', args: ['id', 'op-booked'] }, + { method: 'eq', args: ['status', 'committing'] }, + ]), + ) + + // No-evidence finalize: terminal rejected, never back to pending. + const rejectedWrite = captures[3] + expect(rejectedWrite.payload).toMatchObject({ status: 'rejected' }) + expect((rejectedWrite.payload!.result_data as Record).auto_rejected).toBe(true) + expect(rejectedWrite.filters).toEqual( + expect.arrayContaining([ + { method: 'eq', args: ['id', 'op-unknown'] }, + { method: 'eq', args: ['status', 'committing'] }, + ]), + ) + + // Metric-style log line per recovered row. + const recoveryLines = calls.filter((c) => c.args[0] === 'pending_op_recovery') + expect(recoveryLines).toHaveLength(2) + const outcomes = recoveryLines.map( + (c) => (c.args[1] as Record).outcome, + ) + expect(outcomes).toEqual(['committed', 'rejected']) + expect(recoveryLines[0].args[1]).toMatchObject({ + pendingOperationId: 'op-booked', + companyId: 'company-1', + operationType: 'categorize_transaction', + evidence: 'transaction_booked', + }) + }) + + it('skips a row when the evidence probe fails, leaving it for the next run', async () => { + const { supabase, captures, enqueue } = createCapturingSupabase() + const { log, calls } = createLogSpy() + + enqueue({ + data: [ + makeRow({ + id: 'op-probe-fail', + operation_type: 'categorize_transaction', + params: { transaction_id: 'tx-1' }, + }), + ], + }) + enqueue({ error: { message: 'probe down' } }) + + const summary = await recoverStuckCommittingOperations(supabase, { log }) + + expect(summary).toEqual({ scanned: 1, committed: 0, rejected: 0, skipped: 1 }) + // No terminal write was attempted: only the listing + the failed probe. + expect(captures.filter((c) => c.payload !== undefined)).toHaveLength(0) + const line = calls.find((c) => c.args[0] === 'pending_op_recovery') + expect(line?.level).toBe('error') + }) + + it('counts a lost CAS as skipped when a concurrent finalize resolved the row first', async () => { + const { supabase, enqueue } = createCapturingSupabase() + const { log, calls } = createLogSpy() + + enqueue({ data: [makeRow({ id: 'op-raced' })] }) + enqueue({ data: null }) // CAS matched zero rows + + const summary = await recoverStuckCommittingOperations(supabase, { log }) + + expect(summary).toEqual({ scanned: 1, committed: 0, rejected: 0, skipped: 1 }) + const line = calls.find((c) => c.args[0] === 'pending_op_recovery') + expect((line?.args[1] as Record).outcome).toBe('lost_cas') + }) + + it('returns an empty summary when nothing is stuck', async () => { + const { supabase, enqueue } = createCapturingSupabase() + const { log } = createLogSpy() + + enqueue({ data: [] }) + + const summary = await recoverStuckCommittingOperations(supabase, { log }) + + expect(summary).toEqual({ scanned: 0, committed: 0, rejected: 0, skipped: 0 }) + }) +}) diff --git a/lib/pending-operations/commit.ts b/lib/pending-operations/commit.ts index 83854755..0ef0d244 100644 --- a/lib/pending-operations/commit.ts +++ b/lib/pending-operations/commit.ts @@ -4708,11 +4708,17 @@ async function commitPendingOperationInner( if (finalizeError) { // The executor's side-effects already committed (and are immutable); only - // the terminal status write failed. Without surfacing this, the row would - // sit in 'committing' indefinitely: the expire sweep only targets - // 'pending' ops, so nothing would ever reconcile it. Log loudly with the - // ids needed to finalize manually; the response still reports success - // because the actual work is done. + // the terminal status write failed. Log loudly with the ids needed to + // finalize manually; the response still reports success because the + // actual work is done. + // + // Runbook (#843): rows left in 'committing' by this failure are picked up + // by the daily recovery sweep in + // lib/pending-operations/recover-stuck-committing.ts (runs from the + // expire cron, threshold 15 min). Grep logs for 'pending_op_recovery' to + // see per-row outcomes: 'committed' when posted side-effects were + // verified, 'rejected' when no trace was detectable (nothing re-executes + // either way). log.error('failed to finalize pending_operation to committed (left in committing)', finalizeError, { pendingOperationId: pendingOp.id, operationType: pendingOp.operation_type, diff --git a/lib/pending-operations/recover-stuck-committing.ts b/lib/pending-operations/recover-stuck-committing.ts new file mode 100644 index 00000000..c1f4f5ee --- /dev/null +++ b/lib/pending-operations/recover-stuck-committing.ts @@ -0,0 +1,318 @@ +/** + * Recovery sweep for pending_operations stuck in 'committing' (issue #843). + * + * The commit dispatcher (lib/pending-operations/commit.ts) claims an op by + * flipping pending -> committing in an atomic CAS, runs the executor's + * side-effects, then writes the terminal 'committed' status. If the process + * dies between those steps, or the terminal write itself fails (the + * "failed to finalize pending_operation to committed" log line from PR #841), + * the row stays status='committing' forever: the expire cron only sweeps + * status='pending', so stuck rows were invisible until this sweep. + * + * Semantics (deliberately conservative, no new status values; see #842 for + * the failed_partial half): + * + * - Only rows whose updated_at is older than STUCK_COMMITTING_THRESHOLD_MINUTES + * are touched. The claim CAS is an UPDATE, and pending_operations has the + * standard update_updated_at_column() BEFORE UPDATE trigger, so updated_at + * IS the claim timestamp on a stuck row (nothing else updates a + * 'committing' row without also changing its status). 15 minutes is well + * past the Vercel function ceiling (300s), so no in-flight executor can + * still be running when a row qualifies. + * + * - Rows with positive evidence that the side-effects posted are finalized + * to 'committed' with result_data.recovered=true. Evidence is per + * operation_type and only exists where the op's params identify a target + * row whose posted state is observable (see findPostedEvidence). There is + * no generic side-effect -> pending_op linkage today (that is #842's + * "record posted ids" work), so most types have no probe. + * + * - Rows with no detectable evidence go to terminal 'rejected' with a + * result_data explanation. NEVER back to 'pending': re-executing could + * duplicate side-effects that posted without leaving a detectable trace + * (sent emails, journal entries not referenced by the params). A rejected + * proposal is safe: nothing re-runs, and the user can re-stage after + * verifying manually. + * + * - Every terminal write is CAS-guarded on status='committing' so a + * concurrent finalize (or a second sweep run) can never clobber a row that + * just resolved. The DB immutability trigger + * (enforce_pending_operations_immutability) additionally blocks UPDATEs on + * terminal rows; the CAS means we never even hit it. + * + * Observability: one structured warn per recovered row with the constant + * message 'pending_op_recovery' (metric-style; count by `outcome`). + * Runbook: grep Vercel logs for pending_op_recovery. outcome='committed' + * needs no action (side-effects verified present). outcome='rejected' means + * side-effects could not be verified: check the entities named in params and, + * if something did post, leave the rejected row as the audit record and do + * not re-stage the operation. + */ +import type { SupabaseClient } from '@supabase/supabase-js' +import { createLogger, type Logger } from '@/lib/logger' +import { fetchAllRows } from '@/lib/supabase/fetch-all' + +/** + * Minutes a row may sit in 'committing' before the sweep considers it stuck. + * Must stay comfortably above the max request duration (Vercel: 300s) so the + * sweep can never race an in-flight executor. + */ +export const STUCK_COMMITTING_THRESHOLD_MINUTES = 15 + +export interface StuckCommittingRow { + id: string + company_id: string + operation_type: string + params: Record + updated_at: string +} + +/** + * Evidence signals, in the "what did we actually observe" sense. Serialized + * into result_data.recovery.evidence so an auditor can see the basis for a + * recovered 'committed'. + */ +export type PostedEvidence = + | 'transaction_booked' + | 'transaction_linked_to_target_entry' + | 'invoice_payment_recorded' + +export interface RecoverySummary { + scanned: number + committed: number + rejected: number + /** Probe errors + lost CAS races: rows left for the next run. */ + skipped: number +} + +/** + * Probe for positive evidence that a stuck op's side-effects posted. + * + * Returns an evidence label when the target state is observably present, + * null when nothing detectable posted (or the type has no reliable probe). + * Throws on probe/database errors: the caller must then SKIP the row (leave + * it 'committing' for the next run) rather than reject on a transient error. + * + * Probes exist only where params identify the target and the posted state is + * unambiguous: + * + * - categorize_transaction: the staged transaction is anchored to a + * verifikat per the canonical is_transaction_booked(uuid) predicate + * (transactions.journal_entry_id, payment rows, or voucher links). + * Skipped when params.allow_duplicate=true: those ops intentionally post + * on an already-booked transaction, so "booked" proves nothing. + * - link_transaction_journal_entry: the exact (transaction, journal entry) + * pair from params is linked, via transactions.journal_entry_id or a + * transaction_voucher_links row. + * - match_transaction_invoice: an invoice_payments row exists for the exact + * (transaction, invoice) pair from params (unique index guarantees the + * pair is only ever written by a match). + * + * Everything else returns null by design: create_* ops don't know the id of + * the row they would have created, and state-flag types (period locks, + * invoice statuses) can't be distinguished from a user doing the same thing + * manually while the row sat stuck. Be conservative: no evidence, no + * 'committed'. + */ +export async function findPostedEvidence( + supabase: SupabaseClient, + row: StuckCommittingRow, +): Promise { + const params = row.params ?? {} + + switch (row.operation_type) { + case 'categorize_transaction': { + const transactionId = params.transaction_id + if (typeof transactionId !== 'string' || transactionId.length === 0) return null + // allow_duplicate ops post a second entry on an already-booked tx: + // "booked" would be true before the executor ever ran. + if (params.allow_duplicate === true) return null + const { data, error } = await supabase.rpc('is_transaction_booked', { + p_transaction_id: transactionId, + }) + if (error) throw new Error(`is_transaction_booked probe failed: ${error.message}`) + return data === true ? 'transaction_booked' : null + } + + case 'link_transaction_journal_entry': { + const transactionId = params.transaction_id + const journalEntryId = params.journal_entry_id + if (typeof transactionId !== 'string' || typeof journalEntryId !== 'string') return null + + const { data: tx, error: txError } = await supabase + .from('transactions') + .select('id') + .eq('id', transactionId) + .eq('company_id', row.company_id) + .eq('journal_entry_id', journalEntryId) + .maybeSingle() + if (txError) throw new Error(`transactions probe failed: ${txError.message}`) + if (tx) return 'transaction_linked_to_target_entry' + + const { data: link, error: linkError } = await supabase + .from('transaction_voucher_links') + .select('id') + .eq('company_id', row.company_id) + .eq('transaction_id', transactionId) + .eq('journal_entry_id', journalEntryId) + .maybeSingle() + if (linkError) throw new Error(`transaction_voucher_links probe failed: ${linkError.message}`) + return link ? 'transaction_linked_to_target_entry' : null + } + + case 'match_transaction_invoice': { + const transactionId = params.transaction_id + const invoiceId = params.invoice_id + if (typeof transactionId !== 'string' || typeof invoiceId !== 'string') return null + const { data, error } = await supabase + .from('invoice_payments') + .select('id') + .eq('company_id', row.company_id) + .eq('transaction_id', transactionId) + .eq('invoice_id', invoiceId) + .maybeSingle() + if (error) throw new Error(`invoice_payments probe failed: ${error.message}`) + return data ? 'invoice_payment_recorded' : null + } + + default: + return null + } +} + +/** + * Pure builder for the terminal update payload. Exported so unit tests (and + * the pg-real test, which mirrors the exact payload through real triggers) + * pin the shape. + * + * The rejected shape reuses the { auto_rejected, reason } marker family the + * dispatcher and expire cron already write; reason='stuck_committing' is + * distinct from 'expired' so the /pending UI's "Utgick automatiskt" badge + * (strict on reason === 'expired') never claims these rows. + */ +export function buildRecoveryUpdate( + row: StuckCommittingRow, + evidence: PostedEvidence | null, + sweptAtIso: string, +): { + status: 'committed' | 'rejected' + resolved_at: string + result_data: Record +} { + const recovery = { + reason: 'stuck_committing', + evidence, + stuck_since: row.updated_at, + swept_at: sweptAtIso, + } + if (evidence) { + return { + status: 'committed', + resolved_at: sweptAtIso, + result_data: { recovered: true, recovery }, + } + } + return { + status: 'rejected', + resolved_at: sweptAtIso, + result_data: { + auto_rejected: true, + reason: 'stuck_committing', + recovery: { + ...recovery, + note: + 'Operation was stuck in committing and no trace of posted side effects was found. ' + + 'Closed without re-execution; verify the target entities manually before re-staging.', + }, + }, + } +} + +/** + * Sweep rows stuck in 'committing' beyond the threshold and drive each to a + * terminal status. Runs with the service-role client from the expire cron + * (app/api/pending-operations/expire/cron/route.ts); one failing row never + * aborts the rest. + */ +export async function recoverStuckCommittingOperations( + supabase: SupabaseClient, + opts: { log?: Logger; now?: Date } = {}, +): Promise { + const log = opts.log ?? createLogger('pending-operations/recover-stuck-committing') + const now = opts.now ?? new Date() + const cutoff = new Date(now.getTime() - STUCK_COMMITTING_THRESHOLD_MINUTES * 60_000) + + const stuck = await fetchAllRows(({ from, to }) => + supabase + .from('pending_operations') + .select('id, company_id, operation_type, params, updated_at') + .eq('status', 'committing') + .lt('updated_at', cutoff.toISOString()) + .order('id', { ascending: true }) + .range(from, to), + ) + + const summary: RecoverySummary = { + scanned: stuck.length, + committed: 0, + rejected: 0, + skipped: 0, + } + + for (const row of stuck) { + const baseCtx = { + pendingOperationId: row.id, + companyId: row.company_id, + operationType: row.operation_type, + stuckSince: row.updated_at, + ageMinutes: Math.round((now.getTime() - new Date(row.updated_at).getTime()) / 60_000), + } + + let evidence: PostedEvidence | null + try { + evidence = await findPostedEvidence(supabase, row) + } catch (err) { + // Transient probe failure: leave the row 'committing' for the next + // run rather than rejecting on incomplete information. + summary.skipped++ + log.error('pending_op_recovery', err as Error, { ...baseCtx, outcome: 'skipped_probe_error' }) + continue + } + + const update = buildRecoveryUpdate(row, evidence, new Date().toISOString()) + + // CAS on status='committing': if a concurrent finalize resolved the row + // between the listing and this write, zero rows match and we skip. The + // immutability trigger never fires because OLD.status is not terminal on + // any row we actually update. + const { data: updated, error: updateError } = await supabase + .from('pending_operations') + .update(update) + .eq('id', row.id) + .eq('status', 'committing') + .select('id') + .maybeSingle() + + if (updateError) { + summary.skipped++ + log.error('pending_op_recovery', updateError, { ...baseCtx, outcome: 'skipped_write_error' }) + continue + } + if (!updated) { + summary.skipped++ + log.warn('pending_op_recovery', { ...baseCtx, outcome: 'lost_cas' }) + continue + } + + if (update.status === 'committed') summary.committed++ + else summary.rejected++ + + log.warn('pending_op_recovery', { + ...baseCtx, + outcome: update.status, + evidence, + }) + } + + return summary +} diff --git a/tests/pg/pending-operations-committing-recovery.pg.test.ts b/tests/pg/pending-operations-committing-recovery.pg.test.ts new file mode 100644 index 00000000..9dd31d66 --- /dev/null +++ b/tests/pg/pending-operations-committing-recovery.pg.test.ts @@ -0,0 +1,358 @@ +import { randomUUID } from 'crypto' +import type { PoolClient } from 'pg' +import { beforeAll, describe, expect, it } from 'vitest' +import { getPool } from './setup' +import { seedCompany, insertDraftJournalEntry, insertTransaction } from './fixtures' + +/** + * pg-real coverage for the stuck-'committing' recovery sweep (issue #843, + * lib/pending-operations/recover-stuck-committing.ts). + * + * The sweep itself runs through the app-layer Supabase client, so what must + * be proven against real Postgres is the substrate the module relies on: + * + * 1. Row selection: status='committing' AND updated_at older than the + * threshold picks exactly the stuck rows, never fresh claims or + * pending/terminal rows. + * 2. The threshold anchor is trustworthy: the claim CAS + * (pending -> committing) bumps updated_at via the + * update_updated_at_column() BEFORE UPDATE trigger. + * 3. Transition legality: the CAS-guarded committing -> committed and + * committing -> rejected writes (the exact payload shapes + * buildRecoveryUpdate emits) pass the immutability + input-frozen + * triggers; terminal rows are untouchable both via the CAS (0 rows) and + * outright (trigger exception). + * 4. The evidence probe substrate: is_transaction_booked(uuid) exists and + * distinguishes a booked transaction from an unbooked one. + */ + +const THRESHOLD_SQL = `now() - make_interval(mins => 15)` + +async function insertPendingOp( + client: PoolClient, + params: { + userId: string + companyId: string + status: string + updatedAtSql?: string + operationType?: string + opParams?: Record + resolvedAtSql?: string + }, +): Promise { + const id = randomUUID() + await client.query( + `INSERT INTO public.pending_operations + (id, user_id, company_id, operation_type, status, title, params, resolved_at, updated_at) + VALUES ($1, $2, $3, $4, $5, 'committing-recovery test', $6, + ${params.resolvedAtSql ?? 'NULL'}, ${params.updatedAtSql ?? 'now()'})`, + [ + id, + params.userId, + params.companyId, + params.operationType ?? 'categorize_transaction', + params.status, + JSON.stringify(params.opParams ?? {}), + ], + ) + return id +} + +describe('pending_operations stuck-committing recovery (pg-real)', () => { + let userId: string + let companyId: string + let fiscalPeriodId: string + + beforeAll(async () => { + const seeded = await seedCompany() + userId = seeded.userId + companyId = seeded.companyId + fiscalPeriodId = seeded.fiscalPeriodId + }) + + it('selects only committing rows older than the threshold', async () => { + const client = await getPool().connect() + try { + await client.query('BEGIN') + + const stale = await insertPendingOp(client, { + userId, + companyId, + status: 'committing', + updatedAtSql: `now() - interval '20 minutes'`, + }) + // Fresh claim: an in-flight executor may still be running. + await insertPendingOp(client, { + userId, + companyId, + status: 'committing', + updatedAtSql: `now() - interval '5 minutes'`, + }) + // Old but not committing: expiry's problem or already terminal. + await insertPendingOp(client, { + userId, + companyId, + status: 'pending', + updatedAtSql: `now() - interval '20 minutes'`, + }) + await insertPendingOp(client, { + userId, + companyId, + status: 'committed', + updatedAtSql: `now() - interval '20 minutes'`, + resolvedAtSql: 'now()', + }) + await insertPendingOp(client, { + userId, + companyId, + status: 'rejected', + updatedAtSql: `now() - interval '20 minutes'`, + resolvedAtSql: 'now()', + }) + + // Mirrors the module's listing query (status + updated_at cutoff). + const { rows } = await client.query<{ id: string }>( + `SELECT id FROM public.pending_operations + WHERE company_id = $1 + AND status = 'committing' + AND updated_at < ${THRESHOLD_SQL} + ORDER BY id`, + [companyId], + ) + + expect(rows.map((r) => r.id)).toEqual([stale]) + await client.query('ROLLBACK') + } catch (err) { + await client.query('ROLLBACK').catch(() => {}) + throw err + } finally { + client.release() + } + }) + + it('claim CAS (pending -> committing) bumps updated_at, so the threshold anchor is trustworthy', async () => { + const client = await getPool().connect() + try { + await client.query('BEGIN') + + const id = await insertPendingOp(client, { + userId, + companyId, + status: 'pending', + updatedAtSql: `now() - interval '1 hour'`, + }) + + const before = await client.query<{ updated_at: Date }>( + `SELECT updated_at FROM public.pending_operations WHERE id = $1`, + [id], + ) + + // The dispatcher's atomic claim (commit.ts). + const claim = await client.query( + `UPDATE public.pending_operations + SET status = 'committing' + WHERE id = $1 AND status = 'pending'`, + [id], + ) + expect(claim.rowCount).toBe(1) + + const after = await client.query<{ updated_at: Date }>( + `SELECT updated_at FROM public.pending_operations WHERE id = $1`, + [id], + ) + + // update_updated_at_column() sets updated_at = now() on the claim, so a + // 'committing' row's updated_at IS its claim timestamp. + expect(after.rows[0].updated_at.getTime()).toBeGreaterThan( + before.rows[0].updated_at.getTime(), + ) + await client.query('ROLLBACK') + } catch (err) { + await client.query('ROLLBACK').catch(() => {}) + throw err + } finally { + client.release() + } + }) + + it('finalizes a stuck committing row to committed through the real triggers', async () => { + const client = await getPool().connect() + try { + await client.query('BEGIN') + + const id = await insertPendingOp(client, { + userId, + companyId, + status: 'committing', + updatedAtSql: `now() - interval '20 minutes'`, + }) + + // Exact payload shape from buildRecoveryUpdate(evidence != null). + const resultData = { + recovered: true, + recovery: { + reason: 'stuck_committing', + evidence: 'transaction_booked', + stuck_since: '2026-07-22T00:00:00.000Z', + swept_at: '2026-07-22T02:30:00.000Z', + }, + } + const update = await client.query( + `UPDATE public.pending_operations + SET status = 'committed', resolved_at = now(), result_data = $2 + WHERE id = $1 AND status = 'committing'`, + [id, JSON.stringify(resultData)], + ) + expect(update.rowCount).toBe(1) + + const { rows } = await client.query( + `SELECT status, resolved_at, result_data FROM public.pending_operations WHERE id = $1`, + [id], + ) + expect(rows[0].status).toBe('committed') + expect(rows[0].resolved_at).not.toBeNull() + expect(rows[0].result_data).toEqual(resultData) + await client.query('ROLLBACK') + } catch (err) { + await client.query('ROLLBACK').catch(() => {}) + throw err + } finally { + client.release() + } + }) + + it('finalizes a stuck committing row to rejected through the real triggers', async () => { + const client = await getPool().connect() + try { + await client.query('BEGIN') + + const id = await insertPendingOp(client, { + userId, + companyId, + status: 'committing', + updatedAtSql: `now() - interval '20 minutes'`, + operationType: 'create_customer', + }) + + // Exact payload shape from buildRecoveryUpdate(evidence == null). + const resultData = { + auto_rejected: true, + reason: 'stuck_committing', + recovery: { + reason: 'stuck_committing', + evidence: null, + stuck_since: '2026-07-22T00:00:00.000Z', + swept_at: '2026-07-22T02:30:00.000Z', + note: 'Operation was stuck in committing and no trace of posted side effects was found.', + }, + } + const update = await client.query( + `UPDATE public.pending_operations + SET status = 'rejected', resolved_at = now(), result_data = $2 + WHERE id = $1 AND status = 'committing'`, + [id, JSON.stringify(resultData)], + ) + expect(update.rowCount).toBe(1) + + const { rows } = await client.query( + `SELECT status, resolved_at, result_data FROM public.pending_operations WHERE id = $1`, + [id], + ) + expect(rows[0].status).toBe('rejected') + expect(rows[0].resolved_at).not.toBeNull() + expect(rows[0].result_data).toEqual(resultData) + await client.query('ROLLBACK') + } catch (err) { + await client.query('ROLLBACK').catch(() => {}) + throw err + } finally { + client.release() + } + }) + + it('never touches terminal rows: CAS matches zero, unconditional update raises', async () => { + const client = await getPool().connect() + try { + await client.query('BEGIN') + + const committedId = await insertPendingOp(client, { + userId, + companyId, + status: 'committed', + updatedAtSql: `now() - interval '20 minutes'`, + resolvedAtSql: 'now()', + }) + + // The sweep's CAS guard: a terminal row matches zero rows, so the + // immutability trigger never even fires. + const cas = await client.query( + `UPDATE public.pending_operations + SET status = 'rejected', resolved_at = now(), result_data = '{}' + WHERE id = $1 AND status = 'committing'`, + [committedId], + ) + expect(cas.rowCount).toBe(0) + + // And if anything DID try to update a terminal row, the DB blocks it. + await client.query('SAVEPOINT terminal_update') + await expect( + client.query( + `UPDATE public.pending_operations SET result_data = '{"x":1}' WHERE id = $1`, + [committedId], + ), + ).rejects.toThrow(/terminal state/) + await client.query('ROLLBACK TO SAVEPOINT terminal_update') + + const { rows } = await client.query( + `SELECT status FROM public.pending_operations WHERE id = $1`, + [committedId], + ) + expect(rows[0].status).toBe('committed') + await client.query('ROLLBACK') + } catch (err) { + await client.query('ROLLBACK').catch(() => {}) + throw err + } finally { + client.release() + } + }) + + it('is_transaction_booked(uuid) distinguishes booked from unbooked (evidence probe substrate)', async () => { + const journalEntryId = await insertDraftJournalEntry({ + userId, + companyId, + fiscalPeriodId, + status: 'posted', + voucherNumber: 9001, + committedAt: new Date().toISOString(), + }) + const bookedTx = await insertTransaction({ + companyId, + userId, + journalEntryId, + externalId: `recovery-booked-${randomUUID()}`, + }) + const unbookedTx = await insertTransaction({ + companyId, + userId, + externalId: `recovery-unbooked-${randomUUID()}`, + }) + + const client = await getPool().connect() + try { + const booked = await client.query<{ b: boolean }>( + `SELECT public.is_transaction_booked($1) AS b`, + [bookedTx], + ) + expect(booked.rows[0].b).toBe(true) + + const unbooked = await client.query<{ b: boolean }>( + `SELECT public.is_transaction_booked($1) AS b`, + [unbookedTx], + ) + expect(unbooked.rows[0].b).toBe(false) + } finally { + client.release() + } + }) +})