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) )