The commit dispatcher claims an op with an atomic pending -> committing CAS; if the process dies after side-effects post but before the terminal committed write (or that write fails, the PR #841 log line), the row sat in status='committing' forever: the expire cron only sweeps 'pending'. Add lib/pending-operations/recover-stuck-committing.ts, invoked from the existing daily expire cron (no new vercel.json entry): - Only rows whose updated_at (the claim timestamp: the CAS bumps it via the update_updated_at_column trigger) is older than 15 minutes, well past the 300s Vercel function ceiling, so in-flight executors are never raced. - Positive evidence that side-effects posted finalizes the row to committed with result_data.recovered=true. Evidence exists only where params identify a target with an unambiguous posted state: categorize_transaction (is_transaction_booked RPC, skipped for allow_duplicate), link_transaction_journal_entry (exact tx+entry link), match_transaction_invoice (invoice_payments pair row). - No evidence: terminal rejected with an explanatory result_data, never back to pending (re-execution could duplicate side-effects that posted without a trace). Reason 'stuck_committing' is distinct from 'expired' so the UI badge never claims these rows. - Every terminal write is CAS-guarded on status='committing'; probe errors skip the row for the next run. - One structured 'pending_op_recovery' warn per row (count by outcome); runbook comment added next to the #841 finalize-failure log line. Tests: unit coverage for the decision logic and cron wiring (401, sweep invoked, failure isolation), plus a pg-real test proving row selection, the trustworthy updated_at anchor, committing -> terminal transitions through the real immutability/input-frozen triggers, and the is_transaction_booked evidence substrate. Fixes #843 Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Jakob Wennberg
co-authored by
Claude Fable 5
Claude Fable 5
parent
8294899543
commit
25a7261eda
@@ -273,3 +273,4 @@ One line per decision: `[YYYY-MM-DD] <decision>: <why>`. 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).
|
||||
|
||||
@@ -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 () => {
|
||||
|
||||
@@ -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<string, unknown>
|
||||
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 })
|
||||
})
|
||||
|
||||
@@ -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<string, unknown>
|
||||
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<string, unknown> = {}
|
||||
const record =
|
||||
(method: string) =>
|
||||
(...args: unknown[]) => {
|
||||
if (method === 'update') {
|
||||
capture.payload = args[0] as Record<string, unknown>
|
||||
} 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> = {}): 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<string, unknown>).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<string, unknown>).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<string, unknown>).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<string, unknown>).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 })
|
||||
})
|
||||
})
|
||||
@@ -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,
|
||||
|
||||
@@ -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<string, unknown>
|
||||
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<PostedEvidence | null> {
|
||||
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<string, unknown>
|
||||
} {
|
||||
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<RecoverySummary> {
|
||||
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<StuckCommittingRow>(({ 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
|
||||
}
|
||||
@@ -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<string, unknown>
|
||||
resolvedAtSql?: string
|
||||
},
|
||||
): Promise<string> {
|
||||
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()
|
||||
}
|
||||
})
|
||||
})
|
||||
Reference in New Issue
Block a user