From 5e9cd6f761eb4a0c039dc9380075965f75f1f561 Mon Sep 17 00:00:00 2001 From: Mattsson <111893710+mattssonn@users.noreply.github.com> Date: Wed, 26 Aug 2026 23:39:29 +0200 Subject: [PATCH] fix(enable-banking): unstarve the daily bank sync cron (#1969) * fix(enable-banking): unstarve the daily bank sync cron The sync cron self-limited to 50s (no maxDuration export, so the route ran under the 60s platform default) and processed ~17 connections per day against 123 entitled active connections: any given connection only got an automatic sync every 4-7 days, and users bridged the gap by clicking 'Synka' manually, which pushed them to the back of the queue. - export maxDuration = 300 (Vercel Pro ceiling the code always assumed) - sync loop budget 50s -> 230s; health probe gets the 280s leftover - connections sync in concurrent waves of 4 with per-connection error isolation preserved - MAX_CONNECTIONS_PER_RUN 50 -> 300 (safety cap only; one run now covers the whole entitled queue) Cadence stays once daily at 05:00 UTC by design; users who want more can sync manually. Co-Authored-By: Claude Fable 5 * fix(enable-banking): serialize same-company connections within sync waves Skeptic refutation: the post-sync unattended reconciliation sweep is company-scoped, so two connections of one company syncing concurrently run two identical whole-company sweeps whose unlinked-GL-line snapshots race; both can claim the same journal entry for different bank transactions, leaving the GL short while every surface shows reconciled. Waves now fan out over company groups instead of raw connections: one company's connections sync sequentially inside a single wave slot, unrelated companies still run 4-wide. Co-Authored-By: Claude Fable 5 * fix(enable-banking): re-check the sync time budget inside company groups Review finding (PR Reviewer Guide): the budget was only checked between waves, so one company with many connections could run past 230s inside a single wave and eat the health-probe and teardown margin. Co-Authored-By: Claude Fable 5 --------- Co-authored-by: Claude Fable 5 --- .../sync/cron/__tests__/route.test.ts | 86 ++++++++++++++++++- .../enable-banking/sync/cron/route.ts | 77 +++++++++++++---- 2 files changed, 145 insertions(+), 18 deletions(-) diff --git a/app/api/extensions/enable-banking/sync/cron/__tests__/route.test.ts b/app/api/extensions/enable-banking/sync/cron/__tests__/route.test.ts index 7ab95b4e..a1b7a88c 100644 --- a/app/api/extensions/enable-banking/sync/cron/__tests__/route.test.ts +++ b/app/api/extensions/enable-banking/sync/cron/__tests__/route.test.ts @@ -288,15 +288,95 @@ describe('GET /api/extensions/enable-banking/sync/cron: session health probe', ( await expect(response.json()).resolves.toMatchObject({ processed: 1 }) }) - it('applies the fifty-connection cap after entitlement filtering', async () => { - state.active = Array.from({ length: 51 }, (_, index) => connection({ + it('applies the connection cap after entitlement filtering', async () => { + state.active = Array.from({ length: 301 }, (_, index) => connection({ id: `paid-${index}`, company_id: `11111111-1111-4111-8111-${String(index).padStart(12, '0')}`, })) await GET(cronRequest()) - expect(mocks.syncAccountTransactions).toHaveBeenCalledTimes(50) + expect(mocks.syncAccountTransactions).toHaveBeenCalledTimes(300) + }) + + it('syncs connections in concurrent waves of four', async () => { + const started: string[] = [] + const resolvers: (() => void)[] = [] + mocks.syncAccountTransactions.mockImplementation((...args: unknown[]) => { + started.push(args[3] as string) + return new Promise(resolve => { + resolvers.push(() => resolve({ imported: 0, duplicates: 0, errors: 0 })) + }) + }) + state.active = Array.from({ length: 6 }, (_, index) => connection({ + id: `conn-${index}`, + company_id: `11111111-1111-4111-8111-${String(index).padStart(12, '0')}`, + })) + + const responsePromise = GET(cronRequest()) + + // The first wave fans out to exactly SYNC_CONCURRENCY connections; the + // second wave must not start until every sync in the first has settled. + await vi.waitFor(() => expect(started).toHaveLength(4)) + resolvers.splice(0).forEach(resolve => resolve()) + await vi.waitFor(() => expect(started).toHaveLength(6)) + resolvers.splice(0).forEach(resolve => resolve()) + + const response = await responsePromise + await expect(response.json()).resolves.toMatchObject({ processed: 6 }) + }) + + it('never syncs two connections of the same company concurrently', async () => { + // The post-sync unattended sweep is company-scoped: two concurrent sweeps + // for one company can both read a journal entry as unlinked and claim it + // for different bank transactions. Same-company connections must serialize. + const started: string[] = [] + const resolvers: (() => void)[] = [] + mocks.syncAccountTransactions.mockImplementation((...args: unknown[]) => { + started.push(args[3] as string) + return new Promise(resolve => { + resolvers.push(() => resolve({ imported: 0, duplicates: 0, errors: 0 })) + }) + }) + state.active = [ + connection({ id: 'same-a', company_id: 'company-shared' }), + connection({ id: 'same-b', company_id: 'company-shared' }), + connection({ id: 'other', company_id: 'company-other' }), + ] + + const responsePromise = GET(cronRequest()) + + // First wave: one slot per company, so same-b must wait for same-a. + await vi.waitFor(() => expect(started).toContain('other')) + expect(started).toEqual(expect.arrayContaining(['same-a', 'other'])) + expect(started).not.toContain('same-b') + resolvers.splice(0).forEach(resolve => resolve()) + await vi.waitFor(() => expect(started).toContain('same-b')) + resolvers.splice(0).forEach(resolve => resolve()) + + const response = await responsePromise + await expect(response.json()).resolves.toMatchObject({ processed: 3 }) + }) + + it('isolates one failing connection inside a wave', async () => { + mocks.syncAccountTransactions.mockImplementation((...args: unknown[]) => { + if (args[3] === 'conn-1') return Promise.reject(new Error('ASPSP 500')) + return Promise.resolve({ imported: 0, duplicates: 0, errors: 0 }) + }) + state.active = Array.from({ length: 4 }, (_, index) => connection({ + id: `conn-${index}`, + company_id: `11111111-1111-4111-8111-${String(index).padStart(12, '0')}`, + })) + + const response = await GET(cronRequest()) + + const body = await response.json() + expect(body.processed).toBe(4) + expect(body.totalFailed).toBe(1) + const failed = body.results.find( + (r: { connectionId: string }) => r.connectionId === 'conn-1', + ) + expect(failed).toMatchObject({ status: 'error', errors: 1 }) }) it('probes a connection whose accounts are all deselected', async () => { diff --git a/app/api/extensions/enable-banking/sync/cron/route.ts b/app/api/extensions/enable-banking/sync/cron/route.ts index 16a2e52b..5e9bc711 100644 --- a/app/api/extensions/enable-banking/sync/cron/route.ts +++ b/app/api/extensions/enable-banking/sync/cron/route.ts @@ -31,15 +31,26 @@ import type { StoredAccount } from '@/extensions/general/enable-banking/types' ensureInitialized() -const MAX_CONNECTIONS_PER_RUN = 50 +// Without this export the route runs under the platform default (60s), which +// is why the sync loop used to self-limit to 50s and starve the queue: ~17 +// connections per day against 100+ entitled active connections, so any given +// connection only got an automatic sync every 4-7 days. +export const maxDuration = 300 + +const MAX_CONNECTIONS_PER_RUN = 300 +// Connections synced concurrently within one wave. Enable Banking calls are +// I/O-bound, so a small fan-out multiplies throughput without hammering the +// ASPSPs; per-connection error isolation is preserved inside each wave. +const SYNC_CONCURRENCY = 4 /** * GET /api/extensions/enable-banking/sync/cron * Automatic daily bank transaction sync * Runs at 05:00 UTC (07:00 Swedish time) * - * Processes up to 50 connections per run (Vercel Pro 300s timeout). - * Prioritizes connections not synced for the longest time. + * Sized so one run covers every entitled active connection (Vercel Pro 300s + * timeout, 4-way concurrency). Prioritizes connections not synced for the + * longest time, so anything cut off by the time budget is first tomorrow. * Deduplication via external_id makes repeated runs safe. */ export const GET = withCronContext('cron.bank_sync', async (_request, ctx) => { @@ -92,8 +103,8 @@ export const GET = withCronContext('cron.bank_sync', async (_request, ctx) => { } // Apply the batch limit only after entitlement filtering. Otherwise old - // free-tier rows can permanently occupy the first 50 queue positions and - // prevent every paying connection behind them from syncing. + // free-tier rows can permanently occupy the head of the queue and prevent + // every paying connection behind them from syncing. const connections = candidateConnections .filter(connection => entitledCompanyIds.has(connection.company_id)) .slice(0, MAX_CONNECTIONS_PER_RUN) @@ -108,7 +119,9 @@ export const GET = withCronContext('cron.bank_sync', async (_request, ctx) => { // do (a company whose only connection is parked in 'pending_selection' has // nothing to sync but can absolutely have a dead session). const startTime = Date.now() - const TIME_BUDGET_MS = 50_000 // 50s: leave 10s margin for Vercel timeout + // 230s of the 300s maxDuration for the sync loop; the rest is reserved for + // the health probe pass and response teardown below. + const TIME_BUDGET_MS = 230_000 const baseUrl = process.env.NEXT_PUBLIC_APP_URL || 'http://localhost:3000' const results: { @@ -132,12 +145,7 @@ export const GET = withCronContext('cron.bank_sync', async (_request, ctx) => { const notifyKey = (c: { user_id: string; session_id: string | null }) => `${c.user_id}:${c.session_id ?? 'none'}` - for (const connection of connections) { - if (Date.now() - startTime > TIME_BUDGET_MS) { - ctx.log.info('time budget reached', { processedSoFar: results.length }) - break - } - + const syncConnection = async (connection: (typeof connections)[number]) => { try { const daysLeft = getDaysUntilExpiry(connection.consent_expires) const isExpired = daysLeft !== null && daysLeft <= 0 @@ -166,7 +174,7 @@ export const GET = withCronContext('cron.bank_sync', async (_request, ctx) => { status: 'expired', daysUntilExpiry: 0, }) - continue + return } const expiringSoon = isConsentExpiringSoon(connection.consent_expires) @@ -222,7 +230,7 @@ export const GET = withCronContext('cron.bank_sync', async (_request, ctx) => { status: 'skipped', daysUntilExpiry: daysLeft, }) - continue + return } // Detect SIE overlap: skip auto-categorization if the sync range @@ -374,6 +382,42 @@ export const GET = withCronContext('cron.bank_sync', async (_request, ctx) => { } } + // Concurrency is per COMPANY, not per connection: the unattended sweep after + // an SIE-overlap sync is company-scoped (it reconciles every cash account of + // the company), so two connections of one company syncing concurrently would + // run two identical whole-company sweeps whose read-time "unlinked GL lines" + // snapshots race, and both can claim the same journal entry for different + // bank transactions. Grouping keeps one company's connections sequential + // while unrelated companies still fan out. + const companyGroups = new Map() + for (const connection of connections) { + const group = companyGroups.get(connection.company_id) + if (group) group.push(connection) + else companyGroups.set(connection.company_id, [connection]) + } + const groups = [...companyGroups.values()] + + const syncCompanyGroup = async (group: typeof connections) => { + for (const connection of group) { + // Re-check inside the group too: a company with many connections would + // otherwise run to completion past the budget and eat the health-probe + // and teardown margin before the between-waves check fires. + if (Date.now() - startTime > TIME_BUDGET_MS) return + await syncConnection(connection) + } + } + + // Waves of SYNC_CONCURRENCY company groups: the budget check sits between + // waves, and each connection keeps its own try/catch above, so one slow or + // failing bank affects at most its own wave slot. + for (let offset = 0; offset < groups.length; offset += SYNC_CONCURRENCY) { + if (Date.now() - startTime > TIME_BUDGET_MS) { + ctx.log.info('time budget reached', { processedSoFar: results.length }) + break + } + await Promise.all(groups.slice(offset, offset + SYNC_CONCURRENCY).map(syncCompanyGroup)) + } + // Health probe for connections this run did NOT prove alive by syncing them. // // A sync failure is the only thing that used to move a connection off @@ -385,7 +429,10 @@ export const GET = withCronContext('cron.bank_sync', async (_request, ctx) => { // user read old balances as current. Probing costs one cheap session call // per connection and only ever acts on a definite 'dead'. const probeResults: { connectionId: string; bankName: string }[] = [] - const PROBE_BUDGET_MS = 100_000 + // Total-elapsed ceiling (measured from startTime, like TIME_BUDGET_MS): the + // probe pass gets whatever the sync loop left of it, with 20s of maxDuration + // spare for teardown. + const PROBE_BUDGET_MS = 280_000 const provenAlive = new Set( results.filter(r => r.status === 'synced' || r.status === 'expiring_soon').map(r => r.connectionId) )