From b409c09e25cf2f5ddf875a642da9b3689b13adcd Mon Sep 17 00:00:00 2001 From: Lucas Berger Date: Fri, 5 Jun 2026 21:01:57 -0400 Subject: [PATCH] =?UTF-8?q?feat(03-11):=20GREEN=20=E2=80=94=20durable=20cr?= =?UTF-8?q?eate-before-delete=20gating=20+=20drain=20concurrency=20guard?= =?UTF-8?q?=20(CR-04,=20CR-05)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - CR-04: delete rows with groupId query DB for sibling create status before dispatch - sibling 'pending': defer delete to later cycle (leave row pending) - sibling 'failed'/'dead': mark delete failed permanently (original event preserved, D-04) - sibling 'done': dispatch delete normally - CR-05: module-level isDraining guard; overlapping 15s cycles are no-ops - SINGLE-PROCESS ONLY — documented limitation for multi-replica deployments - Fix mockFromFn to use Symbol.for('drizzle:Name') instead of JSON.stringify (circular) - Update D-04 ordering test to queue sibling-status mock response --- apps/api/src/broker/outboxWorker.ts | 244 ++++++++++++++------- apps/api/tests/broker/outboxWorker.test.ts | 24 +- 2 files changed, 174 insertions(+), 94 deletions(-) diff --git a/apps/api/src/broker/outboxWorker.ts b/apps/api/src/broker/outboxWorker.ts index afbeaa1..085fe07 100644 --- a/apps/api/src/broker/outboxWorker.ts +++ b/apps/api/src/broker/outboxWorker.ts @@ -26,7 +26,7 @@ import { schedule } from 'node-cron' import { and, eq, lte } from 'drizzle-orm' import { db } from '../db/client.js' -import { calendarOutbox, memberCredentials } from '../db/schema.js' +import { calendarEvents, calendarOutbox, memberCredentials } from '../db/schema.js' import { createFastmailClient } from './client.js' import { decryptPassword } from './crypto.js' import { syncCalendar } from './sync.js' @@ -53,6 +53,24 @@ const HARD_FAIL_STATUSES = new Set([400, 401, 403]) /** HTTP status code for CalDAV If-Match conflict — D-08 conflict flow. */ const CONFLICT_STATUS = 412 +// ── Drain concurrency guard (CR-05) ────────────────────────────────────────── + +/** + * Module-level drain guard — prevents overlapping 15s drain cycles from + * double-dispatching the same still-pending outbox row. + * + * SINGLE-PROCESS LIMITATION: This guard is valid ONLY for the single-process + * Unraid deployment of this two-user app where all drain cycles share the same + * Node.js module instance. A multi-process or multi-replica deployment (e.g. + * running multiple API containers behind a load balancer) would require a + * durable DB row-claim instead: + * UPDATE calendar_outbox SET status='processing' + * WHERE id=? AND status='pending' + * and only the process that wins the affected-rows check would dispatch the row. + * Do not remove this comment if deploying to multi-process infrastructure. + */ +let isDraining = false + // ── Credential + client loading ────────────────────────────────────────────── /** @@ -269,113 +287,169 @@ async function dispatchRow(row: OutboxRow): Promise { * are sorted before operation='delete'. If the create fails, the linked delete is * skipped (duplicate is recoverable; lost event is not — D-04 / T-03-14). * + * Concurrency guard (CR-05): the module-level `isDraining` flag ensures overlapping + * 15s scheduler invocations are no-ops for the single-process deployment. + * + * Durable create-before-delete gate (CR-04): for delete rows with a groupId, the + * worker queries the DB for the sibling create row's status. It does NOT rely on + * both rows co-occurring in the same in-memory batch. + * * Per-row errors are caught and logged so one bad row cannot crash the loop. */ export async function runOutboxDrain(): Promise { - // Fetch pending rows: WHERE status='pending' AND next_attempt_at <= NOW() - const pending = (await db - .select() - .from(calendarOutbox) - .where( - and( - eq(calendarOutbox.status, 'pending'), - lte(calendarOutbox.nextAttemptAt, new Date()), - ), - )) as OutboxRow[] + // CR-05: single-process concurrency guard (see isDraining declaration for limitations) + if (isDraining) return + isDraining = true - if (pending.length === 0) return + try { + // Fetch pending rows: WHERE status='pending' AND next_attempt_at <= NOW() + const pending = (await db + .select() + .from(calendarOutbox) + .where( + and( + eq(calendarOutbox.status, 'pending'), + lte(calendarOutbox.nextAttemptAt, new Date()), + ), + )) as OutboxRow[] - // D-04 edit-as-move ordering: sort so create rows come before delete rows within the same groupId. - // Rows without a groupId are unaffected (stable relative order preserved). - const sorted = [...pending].sort((a, b) => { - if (a.groupId && b.groupId && a.groupId === b.groupId) { - if (a.operation === 'create' && b.operation === 'delete') return -1 - if (a.operation === 'delete' && b.operation === 'create') return 1 - } - return 0 - }) + if (pending.length === 0) return - // Track groupIds where the create failed so the linked delete is skipped (D-04 / T-03-14) - const failedCreateGroups = new Set() + // D-04 edit-as-move ordering: sort so create rows come before delete rows within the same groupId. + // Rows without a groupId are unaffected (stable relative order preserved). + // This is a fast path; the authoritative gate is the durable DB sibling-status check below. + const sorted = [...pending].sort((a, b) => { + if (a.groupId && b.groupId && a.groupId === b.groupId) { + if (a.operation === 'create' && b.operation === 'delete') return -1 + if (a.operation === 'delete' && b.operation === 'create') return 1 + } + return 0 + }) - for (const row of sorted) { - // D-04: if the create for this group failed, skip the paired delete - if (row.operation === 'delete' && row.groupId && failedCreateGroups.has(row.groupId)) { - console.warn( - `[outboxWorker] Skipping delete row.id=${row.id} — create for groupId=${row.groupId} failed (D-04)`, - ) - continue - } + // Track groupIds where the create failed within this batch (fast path for same-batch pairs). + // Cross-batch ordering is enforced durably by the DB sibling-status check inside the loop. + const failedCreateGroups = new Set() - try { - const result = await dispatchRow(row) + for (const row of sorted) { + // D-04 fast path: if the create for this group already failed in this batch, skip the delete + if (row.operation === 'delete' && row.groupId && failedCreateGroups.has(row.groupId)) { + console.warn( + `[outboxWorker] Skipping delete row.id=${row.id} — create for groupId=${row.groupId} failed this batch (D-04)`, + ) + continue + } - if (result.conflict) { - // 412 — mark failed (no retry), re-sync calendar so UI sees authoritative state (D-08) - await db - .update(calendarOutbox) - .set({ status: 'failed', lastError: result.error ?? '412 conflict' }) - .where(eq(calendarOutbox.id, row.id)) - await triggerTargetedResync(row.calendarUrl, row.userId) + // CR-04: Durable create-before-delete gate — query DB for sibling create status. + // This prevents the delete from running when the create pair straddles drain batches. + if (row.operation === 'delete' && row.groupId) { + const siblingRows = (await db + .select({ status: calendarOutbox.status }) + .from(calendarOutbox) + .where( + and( + eq(calendarOutbox.groupId, row.groupId), + eq(calendarOutbox.operation, 'create'), + ), + )) as Array<{ status: string }> - if (row.groupId && row.operation === 'create') { - failedCreateGroups.add(row.groupId) + const siblingStatus = siblingRows[0]?.status + + if (siblingStatus !== 'done') { + if (siblingStatus === 'failed' || siblingStatus === 'dead') { + // Sibling create failed permanently — skip this delete forever (D-04: original preserved) + console.warn( + `[outboxWorker] Paired create for groupId=${row.groupId} is ${siblingStatus} — marking delete row.id=${row.id} failed (original event preserved, D-04)`, + ) + await db + .update(calendarOutbox) + .set({ + status: 'failed', + lastError: 'paired create did not succeed — original preserved', + }) + .where(eq(calendarOutbox.id, row.id)) + } else { + // Sibling create is still pending/processing — defer this delete to a later cycle + console.warn( + `[outboxWorker] Deferring delete row.id=${row.id} — sibling create (groupId=${row.groupId}) is not yet done (status=${siblingStatus ?? 'not found'})`, + ) + // Leave the delete row pending; do NOT update its status + } + continue } - } else if (result.success) { - // Success — mark done, trigger targeted re-sync to refresh the cache (D-06) - await db - .update(calendarOutbox) - .set({ status: 'done' }) - .where(eq(calendarOutbox.id, row.id)) - await triggerTargetedResync(row.calendarUrl, row.userId) - } else if (result.hardFail) { - // Hard fail — mark failed immediately, no retry (D-07) - await db - .update(calendarOutbox) - .set({ status: 'failed', lastError: result.error ?? 'Hard fail' }) - .where(eq(calendarOutbox.id, row.id)) + } - if (row.groupId && row.operation === 'create') { - failedCreateGroups.add(row.groupId) - } - } else { - // Transient — exponential backoff or dead-letter (D-07 / T-03-12) - const nextAttemptCount = row.attemptCount + 1 - if (nextAttemptCount >= MAX_ATTEMPTS) { - // Dead-letter: max attempts reached (T-03-12) + try { + const result = await dispatchRow(row) + + if (result.conflict) { + // 412 — mark failed (no retry), re-sync calendar so UI sees authoritative state (D-08) await db .update(calendarOutbox) - .set({ - status: 'dead', - attemptCount: nextAttemptCount, - lastError: result.error ?? 'Max attempts exceeded', - }) + .set({ status: 'failed', lastError: result.error ?? '412 conflict' }) + .where(eq(calendarOutbox.id, row.id)) + await triggerTargetedResync(row.calendarUrl, row.userId) + + if (row.groupId && row.operation === 'create') { + failedCreateGroups.add(row.groupId) + } + } else if (result.success) { + // Success — mark done, trigger targeted re-sync to refresh the cache (D-06) + await db + .update(calendarOutbox) + .set({ status: 'done' }) + .where(eq(calendarOutbox.id, row.id)) + await triggerTargetedResync(row.calendarUrl, row.userId) + } else if (result.hardFail) { + // Hard fail — mark failed immediately, no retry (D-07) + await db + .update(calendarOutbox) + .set({ status: 'failed', lastError: result.error ?? 'Hard fail' }) .where(eq(calendarOutbox.id, row.id)) if (row.groupId && row.operation === 'create') { failedCreateGroups.add(row.groupId) } } else { - // WR-01: use row.attemptCount (the attempt that just failed, 0-based) as the backoff index. - // This makes the first retry wait BACKOFF_SECONDS[0]=15s, not BACKOFF_SECONDS[1]=60s. - const backoffMs = (BACKOFF_SECONDS[row.attemptCount] ?? 1800) * 1000 - await db - .update(calendarOutbox) - .set({ - attemptCount: nextAttemptCount, - nextAttemptAt: new Date(Date.now() + backoffMs), - lastError: result.error, - }) - .where(eq(calendarOutbox.id, row.id)) + // Transient — exponential backoff or dead-letter (D-07 / T-03-12) + const nextAttemptCount = row.attemptCount + 1 + if (nextAttemptCount >= MAX_ATTEMPTS) { + // Dead-letter: max attempts reached (T-03-12) + await db + .update(calendarOutbox) + .set({ + status: 'dead', + attemptCount: nextAttemptCount, + lastError: result.error ?? 'Max attempts exceeded', + }) + .where(eq(calendarOutbox.id, row.id)) + + if (row.groupId && row.operation === 'create') { + failedCreateGroups.add(row.groupId) + } + } else { + // WR-01: use row.attemptCount (the attempt that just failed, 0-based) as the backoff index. + // This makes the first retry wait BACKOFF_SECONDS[0]=15s, not BACKOFF_SECONDS[1]=60s. + const backoffMs = (BACKOFF_SECONDS[row.attemptCount] ?? 1800) * 1000 + await db + .update(calendarOutbox) + .set({ + attemptCount: nextAttemptCount, + nextAttemptAt: new Date(Date.now() + backoffMs), + lastError: result.error, + }) + .where(eq(calendarOutbox.id, row.id)) + } } + } catch (err) { + // Per-row error isolation: log but never crash the loop (T-03-13) + console.error( + `[outboxWorker] Error dispatching row.id=${row.id} uid=${row.uid}:`, + err instanceof Error ? err.message : String(err), + ) } - } catch (err) { - // Per-row error isolation: log but never crash the loop (T-03-13) - console.error( - `[outboxWorker] Error dispatching row.id=${row.id} uid=${row.uid}:`, - err instanceof Error ? err.message : String(err), - ) } + } finally { + isDraining = false } } diff --git a/apps/api/tests/broker/outboxWorker.test.ts b/apps/api/tests/broker/outboxWorker.test.ts index d31bcf3..e998103 100644 --- a/apps/api/tests/broker/outboxWorker.test.ts +++ b/apps/api/tests/broker/outboxWorker.test.ts @@ -126,16 +126,12 @@ function wireMockChain() { mockUpdate.mockReturnValue({ set: mockUpdateSet }) // mockFromFn differentiates by table argument: // - memberCredentials table → returns FAKE_CRED_ROW (so loadClientForUser succeeds by default) - // - anything else → returns mockPendingRows (outbox query) + // - anything else (calendarOutbox, calendarEvents) → returns mockWherePending + // Use Symbol.for('drizzle:Name') to identify the table — JSON.stringify throws on circular Drizzle + // table structures so it cannot be used for table identification. mockFromFn.mockImplementation((table: unknown) => { - // Drizzle table objects have a Symbol.for('drizzle:Name') property and a [Table.Symbol.Name]. - // The safest approach: JSON.stringify often includes the table config name. - let isCred = false - try { - isCred = JSON.stringify(table).includes('member_credentials') - } catch { - // table not serializable — not a credential table - } + const tableName = (table as Record)[Symbol.for('drizzle:Name')] ?? '' + const isCred = tableName === 'member_credentials' return { where: isCred ? vi.fn().mockResolvedValue([FAKE_CRED_ROW]) @@ -307,6 +303,16 @@ describe('runOutboxDrain — edit-as-move ordering (D-04)', () => { // Both rows in the pending list mockPendingRows = [deleteRow, createRow] + // The durable sibling-status check (CR-04) runs for the delete row with groupId. + // It queries calendarOutbox for the sibling create's status. By the time the delete + // is processed (create was sorted and dispatched first), we simulate the sibling as 'done'. + // The base mockWherePending returns mockPendingRows for all calendarOutbox selects; we + // override just the sibling-status call with mockImplementationOnce queued after the + // pending-rows select call. + mockWherePending + .mockImplementationOnce(() => Promise.resolve([deleteRow, createRow])) // pending-rows select + .mockImplementationOnce(() => Promise.resolve([{ status: 'done' }])) // sibling-status select + await runOutboxDrain() // CREATE must be called before DELETE