diff --git a/src/modules/investments/instruments/application/catalog-enrichment.service.ts b/src/modules/investments/instruments/application/catalog-enrichment.service.ts index 87b5f2f..0c40573 100644 --- a/src/modules/investments/instruments/application/catalog-enrichment.service.ts +++ b/src/modules/investments/instruments/application/catalog-enrichment.service.ts @@ -58,7 +58,7 @@ export class CatalogEnrichmentService { private running = false; private readonly completions = new Map< string, - { done: Promise; resolve: () => void; workId: string } + { attempt: TbankSyncAttempt; done: Promise; resolve: () => void; workId: string } >(); async ensureDetail( @@ -208,7 +208,7 @@ export class CatalogEnrichmentService { const done = new Promise((finish) => { resolve = finish; }); - this.completions.set(attemptId, { workId: id, done, resolve }); + this.completions.set(attemptId, { attempt, done, resolve, workId: id }); const key = producerKey(input); const producer = this.producers.get(key); if (producer) producer.observers.push({ attempt, input }); @@ -580,6 +580,14 @@ export class CatalogEnrichmentService { } } } catch { + // Persistence failed mid-batch: finish every still-open observer + // attempt durably so no source row is left running. Best effort — a + // row that stays open is re-reserved by the next reader. + for (const completion of this.completions.values()) { + await this.repository + .finish(completion.attempt, false) + .catch(() => undefined); + } for (const completion of this.completions.values()) completion.resolve(); this.completions.clear(); this.active.clear(); diff --git a/src/modules/investments/operations/application/brokerage-operations-sync.service.ts b/src/modules/investments/operations/application/brokerage-operations-sync.service.ts index 4207384..ca3dbf3 100644 --- a/src/modules/investments/operations/application/brokerage-operations-sync.service.ts +++ b/src/modules/investments/operations/application/brokerage-operations-sync.service.ts @@ -209,6 +209,12 @@ export class BrokerageOperationsSyncService { 'tbank.operations.sync.persistence_failed', { userId: attempt.userId, code: TbankErrorCode.Unavailable }, ); + // The step could not even record its failure; close the attempt + // durably so it does not stay running. Best effort — an open row + // is re-reserved by the next request. + await this.repository + .fail(attempt, false) + .catch(() => undefined); } const next = pending ? await this.repository.pending(attempt) : null; if (next) { diff --git a/src/modules/investments/tbank-sync/tbank-hourly.scheduler.ts b/src/modules/investments/tbank-sync/tbank-hourly.scheduler.ts index 7fc8e11..17d0aa6 100644 --- a/src/modules/investments/tbank-sync/tbank-hourly.scheduler.ts +++ b/src/modules/investments/tbank-sync/tbank-hourly.scheduler.ts @@ -1,6 +1,7 @@ import { Injectable, OnModuleDestroy, OnModuleInit } from '@nestjs/common'; import { randomUUID } from 'node:crypto'; +import { StructuredLogger } from '../../../common/logging/structured-logger.service'; import { TbankErrorCode } from '../../../integrations/tbank/tbank-error-code'; import { TbankAccessService, @@ -12,6 +13,9 @@ import { BrokerageOperationsSyncService } from '../operations/operations.api'; import { PortfolioService } from '../portfolio/portfolio.api'; import { UpcomingEventsService } from '../upcoming-events/upcoming-events.service'; import { + HOURLY_CAUSE_INTERRUPTED, + type HourlyCause, + type HourlyChild, type HourlySection, TbankSyncRepository, } from './tbank-sync.repository'; @@ -19,11 +23,16 @@ import { const HOUR_MS = 3_600_000; const TICK_MS = 60_000; const WORK_POLL_MS = 100; +const WORK_POLL_BACKOFF_FACTOR = 2; +const WORK_POLL_MAX_MS = 5_000; +// ponytail: fixed 15-minute step wait bound (900_000 ms), no config knob; +// make it configurable if slow steps become regular. +const WORK_WAIT_BOUND_MS = 900_000; const FIRST_DAY_OF_MONTH = 1; -const ONE_ATTEMPT = 1; -const NO_ATTEMPTS = 0; +const NO_MILLISECONDS = 0; -type SettledWork = { errorCode: null | string; id: string; state: string }; +type DispatchStep = (child: HourlyChild) => Promise; +type Accounts = Awaited>['accounts']; // One native timer per process. The timer only proposes work; the durable // reservation in TbankSyncRepository is the fence that keeps a single @@ -32,6 +41,9 @@ type SettledWork = { errorCode: null | string; id: string; state: string }; @Injectable() export class TbankHourlyScheduler implements OnModuleDestroy, OnModuleInit { private readonly processRunId = randomUUID(); + // Attempt ids of walks currently executing in this process; a running row + // outside this set lost its settle and is closed by recovery. + private readonly active = new Set(); private timer: NodeJS.Timeout | null = null; private ticking = false; @@ -43,6 +55,7 @@ export class TbankHourlyScheduler implements OnModuleDestroy, OnModuleInit { private readonly portfolio: PortfolioService, private readonly operations: BrokerageOperationsSyncService, private readonly events: UpcomingEventsService, + private readonly logger: StructuredLogger, ) {} onModuleInit() { @@ -74,6 +87,7 @@ export class TbankHourlyScheduler implements OnModuleDestroy, OnModuleInit { // A fresh connection starts its first walk immediately, regardless of the // previous term of the replaced connection. private async launch(userId: string) { + await this.recoverAbandoned(userId); const reserved = await this.repository.reserveHourly( userId, this.processRunId, @@ -87,135 +101,263 @@ export class TbankHourlyScheduler implements OnModuleDestroy, OnModuleInit { return reserved.work; } - // One membership pass, then the owners refresh their own datasets. The - // walk publishes only safe section results and holds no transaction or - // upstream call inside one. - private async walk(attempt: TbankSyncAttempt) { - const failures: HourlySection[] = []; - const attempted = { count: NO_ATTEMPTS }; - try { - const { userId, revision } = attempt; - await this.refreshMembership(userId, revision, failures, attempted); - if (failures.some((item) => item.errorCode === TbankErrorCode.InvalidToken)) { - return void this.settle(attempt, failures, attempted.count); - } - const current = await this.access.local(userId); - if (current.revision !== revision) { - return void this.settle(attempt, failures, attempted.count); - } - await this.refreshAccounts(userId, revision, current.accounts, failures, attempted); - await this.settle(attempt, failures, attempted.count); - } catch { - await this.settle(attempt, failures, attempted.count); + // A running hourly row of this process that has no active walk lost its + // settle (suppressed persistence error). Close it from its durable step + // results; rows of other processes are closed by the reservation restart + // branch. Rides the existing tick: no new timers or sanitors. + private async recoverAbandoned(userId: string) { + const rows = await this.repository.runningHourly(userId); + for (const row of rows) { + if (row.processRunId !== this.processRunId) continue; + if (this.active.has(row.attemptId)) continue; + await this.repository + .settleHourly( + { + userId, + revision: row.revision, + id: row.id, + attemptId: row.attemptId, + }, + HOURLY_CAUSE_INTERRUPTED, + ) + .catch(() => undefined); } } + // One membership pass, then the owners refresh their own datasets. The + // walk publishes only safe section results and holds no transaction or + // upstream call inside one. The attempt outcome is derived from durable + // step rows by the repository, never from walk-local state. + private async walk(attempt: TbankSyncAttempt) { + this.active.add(attempt.attemptId); + try { + const { userId, revision } = attempt; + const children: HourlyChild[] = []; + // Required sections whose dispatch failed never get a durable step + // row; they are recorded as problems so the attempt cannot succeed. + const dispatchProblems: HourlySection[] = []; + const dispatch: DispatchStep = async (child) => { + children.push(child); + await this.repository.recordHourlyChild(attempt, child); + }; + const accounts = await this.refreshMembership( + attempt, + dispatch, + dispatchProblems, + ); + if (!accounts) return; + await this.refreshAccounts( + userId, + revision, + accounts, + dispatch, + dispatchProblems, + ); + await this.waitWorks(userId, children.map((child) => child.id)); + for (const account of accounts) { + await this.events.get(userId, account.accountId).catch(() => undefined); + } + await this.settle(attempt, null, dispatchProblems); + } catch { + await this.settle(attempt, HOURLY_CAUSE_INTERRUPTED); + } finally { + this.active.delete(attempt.attemptId); + } + } + + // Returns the current accounts, or null when the attempt is already + // settled (invalid token or a replaced connection revision). private async refreshMembership( - userId: string, - revision: string, - failures: HourlySection[], - attempted: { count: number }, + attempt: TbankSyncAttempt, + dispatch: DispatchStep, + dispatchProblems: HourlySection[], ) { + const { userId, revision } = attempt; const member = await this.membership.request(userId, revision, true); - attempted.count += ONE_ATTEMPT; - if (!member) return; - await member.done; - const errorCode = await this.repository.workError(userId, member.id); - if (errorCode) failures.push({ section: 'membership', errorCode }); + if (member) { + await dispatch({ id: member.id, section: 'membership' }); + await member.done; + if ( + (await this.repository.workError(userId, member.id)) === + TbankErrorCode.InvalidToken + ) { + await this.settle(attempt); + return null; + } + } else { + dispatchProblems.push({ + section: 'membership', + errorCode: TbankErrorCode.Unavailable, + }); + } + const current = await this.access.local(userId); + if (current.revision !== revision) { + await this.settle(attempt, HOURLY_CAUSE_INTERRUPTED); + return null; + } + return current.accounts; } private async refreshAccounts( userId: string, revision: string, - accounts: Awaited>['accounts'], - failures: HourlySection[], - attempted: { count: number }, + accounts: Accounts, + dispatch: DispatchStep, + dispatchProblems: HourlySection[], ) { for (const account of accounts) { - await this.refreshPortfolio(userId, revision, account.accountId, failures, attempted); + await this.dispatchPortfolio( + userId, + revision, + account.accountId, + dispatch, + dispatchProblems, + ); } for (const account of accounts) { - await this.refreshOperations(userId, revision, account.accountId, failures, attempted); + await this.dispatchOperations( + userId, + revision, + account.accountId, + dispatch, + dispatchProblems, + ); } for (const account of accounts) { - await this.refreshEvents(userId, account.accountId, failures, attempted); + await this.dispatchEvents( + userId, + account.accountId, + dispatch, + dispatchProblems, + ); } } - private async refreshPortfolio(userId: string, revision: string, accountId: string, failures: HourlySection[], attempted: { count: number }) { - const work = await this.portfolio.request(userId, revision, accountId, true); - attempted.count += ONE_ATTEMPT; - if (!work) return; - await work.done; - const errorCode = await this.repository.workError(userId, work.id); - if (errorCode) failures.push({ section: 'portfolio', errorCode, accountId }); - } - - private async refreshOperations(userId: string, revision: string, accountId: string, failures: HourlySection[], attempted: { count: number }) { - attempted.count += ONE_ATTEMPT; + private async dispatchPortfolio( + userId: string, + revision: string, + accountId: string, + dispatch: DispatchStep, + dispatchProblems: HourlySection[], + ) { try { - const id = await this.operations.request(userId, [{ accountId, from: monthStartOf(new Date()), to: new Date() }], true, 'screen', revision); - if (!id) return; - const errorCode = await this.settledWorkError(userId, [id]); - if (errorCode) failures.push({ section: 'operations', errorCode, accountId }); + const work = await this.portfolio.request(userId, revision, accountId, true); + if (work) { + await dispatch({ accountId, id: work.id, section: 'portfolio' }); + return; + } } catch { - failures.push({ section: 'operations', errorCode: TbankErrorCode.Unavailable, accountId }); + // A failed dispatch has no durable step row; record the problem. + } + dispatchProblems.push({ + accountId, + errorCode: TbankErrorCode.Unavailable, + section: 'portfolio', + }); + } + + private async dispatchOperations( + userId: string, + revision: string, + accountId: string, + dispatch: DispatchStep, + dispatchProblems: HourlySection[], + ) { + try { + const id = await this.operations.request( + userId, + [ + { + accountId, + from: monthStartOf(new Date()), + to: new Date(), + }, + ], + true, + 'screen', + revision, + ); + if (id) await dispatch({ accountId, id, section: 'operations' }); + } catch { + dispatchProblems.push({ + accountId, + errorCode: TbankErrorCode.Unavailable, + section: 'operations', + }); } } - private async refreshEvents(userId: string, accountId: string, failures: HourlySection[], attempted: { count: number }) { - attempted.count += ONE_ATTEMPT; + private async dispatchEvents( + userId: string, + accountId: string, + dispatch: DispatchStep, + dispatchProblems: HourlySection[], + ) { try { const ids = await this.events.refresh(userId, accountId); - const failed = ids.length ? (await this.settledWorks(userId, ids)).find((row) => row.state === 'failed') : undefined; - if (failed?.errorCode) failures.push({ section: 'events', errorCode: failed.errorCode, accountId }); - else await this.events.get(userId, accountId); + for (const id of ids) { + await dispatch({ accountId, id, section: 'events' }); + } } catch { - failures.push({ section: 'events', errorCode: TbankErrorCode.Unavailable, accountId }); + dispatchProblems.push({ + accountId, + errorCode: TbankErrorCode.Unavailable, + section: 'events', + }); } } private async settle( attempt: TbankSyncAttempt, - failures: HourlySection[], - attempted: number, + cause: HourlyCause | null = null, + dispatchProblems: HourlySection[] = [], ) { - let result: 'failed' | 'partial' | 'succeeded' = 'succeeded'; - if (failures.length) { - result = attempted > NO_ATTEMPTS && failures.length >= attempted ? 'failed' : 'partial'; - } + // A suppressed settle error is not lost: the walk leaves the active set + // and the next tick closes the attempt from its durable step results. await this.repository - .settleHourly(attempt, result, failures) - .catch(() => undefined); + .settleHourly(attempt, cause, dispatchProblems) + .catch(() => { + this.logger.event('error', 'tbank.hourly.settle_failed', { + userId: attempt.userId, + }); + return undefined; + }); } - private async settledWorkError(userId: string, ids: string[]) { - const rows = await this.settledWorks(userId, ids); - return rows.find((row) => row.state === 'failed')?.errorCode ?? null; + // Drops settled and missing rows from the wait set and expires waits that + // passed the per-step bound. + private advanceWaiting( + waiting: Map, + rows: { id: string; state: string }[], + ) { + for (const row of rows) { + if (row.state !== 'running') waiting.delete(row.id); + } + for (const id of [...waiting.keys()]) { + if (!rows.some((row) => row.id === id)) { + waiting.delete(id); + continue; + } + const startedAt = waiting.get(id) ?? NO_MILLISECONDS; + if (Date.now() - startedAt >= WORK_WAIT_BOUND_MS) waiting.delete(id); + } } - // Polls the durable work rows until they leave the running state; a row - // deleted with its connection counts as interrupted, not failed. - private async settledWorks(userId: string, ids: string[]) { - const settled = new Map(); - const pending = new Set(ids); - while (pending.size) { - const rows = await this.repository.works(userId, [...pending]); - for (const row of rows) { - if (row.state !== 'running') { - settled.set(row.id, row); - pending.delete(row.id); - } - } - for (const id of [...pending]) { - if (!rows.some((row) => row.id === id)) { - settled.set(id, { id, state: 'cancelled', errorCode: null }); - pending.delete(id); - } + // Waits for every dispatched step to leave the running state. Each wait is + // bounded so a stuck step cannot hold the attempt running forever; the + // bound does not cancel the upstream call, and late results stay in the + // step row where the attempt fence decides whether they count. + private async waitWorks(userId: string, ids: string[]) { + const waiting = new Map(ids.map((id) => [id, Date.now()])); + let pollMs = WORK_POLL_MS; + while (waiting.size) { + const rows = await this.repository.works(userId, [...waiting.keys()]); + this.advanceWaiting(waiting, rows); + if (waiting.size) { + await new Promise((resolve) => setTimeout(resolve, pollMs)); + pollMs = Math.min(pollMs * WORK_POLL_BACKOFF_FACTOR, WORK_POLL_MAX_MS); } - if (pending.size) await new Promise((resolve) => setTimeout(resolve, WORK_POLL_MS)); } - return [...settled.values()]; } } diff --git a/src/modules/investments/tbank-sync/tbank-sync.repository.ts b/src/modules/investments/tbank-sync/tbank-sync.repository.ts index ae82586..cebd86f 100644 --- a/src/modules/investments/tbank-sync/tbank-sync.repository.ts +++ b/src/modules/investments/tbank-sync/tbank-sync.repository.ts @@ -18,7 +18,19 @@ export type HourlySection = { errorCode: null | string; section: 'events' | 'membership' | 'operations' | 'portfolio'; }; +export type HourlyChild = { + accountId?: string; + id: string; + section: HourlySection['section']; +}; +export const HOURLY_CAUSE_DEADLINE = 'DEADLINE'; +export const HOURLY_CAUSE_INTERRUPTED = 'INTERRUPTED'; +export type HourlyCause = typeof HOURLY_CAUSE_INTERRUPTED; +type StepOutcome = { problem?: HourlySection; success: boolean }; +type StepRow = { attemptId: string; errorCode: null | string; state: string }; export type HourlyProgress = { + attempts?: Record; + children?: HourlyChild[]; result?: 'failed' | 'partial' | 'succeeded'; sections: HourlySection[]; }; @@ -191,12 +203,10 @@ export class TbankSyncRepository { const due = now.valueOf() - previous.createdAt.valueOf() >= hourMs; if (previous.state === 'running') { if (previous.processRunId === runId) return null; - // A dead process cannot finish its attempt; interrupt it, but its - // start stays the basis of the next term. - await tx.tbankSyncWork.update({ - where: { userId_revision_scopeKey: key }, - data: { state: 'cancelled', errorCode: null }, - }); + // A dead process cannot finish its attempt; close it from its + // actual step results, but its start stays the basis of the next + // term. + await this.closeHourly(tx, previous, 'INTERRUPTED', []); if (!due) return null; } else if (!due) return null; } @@ -220,28 +230,146 @@ export class TbankSyncRepository { }); } - settleHourly( - attempt: TbankSyncAttempt, - result: 'failed' | 'partial' | 'succeeded', + private classifyStep( + child: HourlyChild, + row: StepRow | undefined, + recorded: string | undefined, + cause: HourlyCause | null, + ): StepOutcome { + if (row && row.attemptId === recorded && row.state === 'complete') { + return { success: true }; + } + const stale = !row || (recorded !== undefined && row.attemptId !== recorded); + let errorCode: null | string = null; + if (!stale && row.state === 'failed') errorCode = row.errorCode; + else if (!stale && cause === HOURLY_CAUSE_INTERRUPTED) { + errorCode = HOURLY_CAUSE_INTERRUPTED; + } else if (!stale) errorCode = HOURLY_CAUSE_DEADLINE; + return { + problem: { accountId: child.accountId, errorCode, section: child.section }, + success: false, + }; + } + + private deriveResult( + cause: HourlyCause | null, sections: HourlySection[], + successes: number, + ): 'failed' | 'partial' | 'succeeded' { + if (!cause && !sections.length) return 'succeeded'; + return successes ? 'partial' : 'failed'; + } + + // The outcome is derived from the durable state of the dispatched steps, + // never from an in-memory error list that an exception path can empty. + // `cause` marks a close before normal completion and caps the result at + // `partial`; `dispatchProblems` records required sections that never got a + // durable step row because their dispatch failed. + private async closeHourly( + tx: Prisma.TransactionClient, + work: { + attemptId: string; + id: string; + progress: Prisma.JsonValue; + revision: string; + userId: string; + }, + cause: HourlyCause | null, + dispatchProblems: HourlySection[], ) { - return this.prisma.tbankSyncWork.updateMany({ - where: { - id: attempt.id, - attemptId: attempt.attemptId, - userId: attempt.userId, - revision: attempt.revision, - state: 'running', - }, + const progress = work.progress as HourlyProgress; + const children = progress.children ?? []; + const rows = children.length + ? await tx.tbankSyncWork.findMany({ + where: { + userId: work.userId, + id: { in: children.map((child) => child.id) }, + }, + select: { attemptId: true, errorCode: true, id: true, state: true }, + }) + : []; + const byId = new Map(rows.map((row) => [row.id, row])); + const outcomes = children.map((child) => + this.classifyStep( + child, + byId.get(child.id), + progress.attempts?.[child.id], + cause, + ), + ); + const sections = outcomes + .flatMap((outcome) => (outcome.problem ? [outcome.problem] : [])) + .concat(dispatchProblems); + const successes = outcomes.filter((outcome) => outcome.success).length; + const result = this.deriveResult(cause, sections, successes); + return tx.tbankSyncWork.updateMany({ + where: { id: work.id, attemptId: work.attemptId, state: 'running' }, data: { state: result === 'failed' ? 'failed' : 'complete', errorCode: - sections.find((section) => section.errorCode)?.errorCode ?? null, - progress: { result, sections }, + cause ?? + sections.find((section) => section.errorCode)?.errorCode ?? + null, + progress: { + attempts: progress.attempts ?? {}, + children, + result, + sections, + }, }, }); } + async settleHourly( + attempt: TbankSyncAttempt, + cause: HourlyCause | null = null, + dispatchProblems: HourlySection[] = [], + ) { + return this.prisma.$transaction(async (tx) => { + const work = await tx.tbankSyncWork.findFirst({ where: attempt }); + if (!work || work.state !== 'running') return { count: 0 }; + return this.closeHourly(tx, work, cause, dispatchProblems); + }); + } + + // Records a dispatched step on the attempt before it is awaited, so a lost + // walk can still derive its outcome from durable rows. + async recordHourlyChild(attempt: TbankSyncAttempt, child: HourlyChild) { + return this.prisma.$transaction(async (tx) => { + const work = await tx.tbankSyncWork.findFirst({ + where: { ...attempt, state: 'running' }, + select: { progress: true }, + }); + if (!work) return false; + const progress = work.progress as HourlyProgress; + const row = await tx.tbankSyncWork.findFirst({ + where: { userId: attempt.userId, id: child.id }, + select: { attemptId: true }, + }); + const updated = await tx.tbankSyncWork.updateMany({ + where: { ...attempt, state: 'running' }, + data: { + progress: { + attempts: { + ...progress.attempts, + ...(row ? { [child.id]: row.attemptId } : {}), + }, + children: [...(progress.children ?? []), child], + sections: progress.sections ?? [], + }, + }, + }); + return Boolean(updated.count); + }); + } + + runningHourly(userId: string) { + return this.prisma.tbankSyncWork.findMany({ + where: { userId, scopeKey: 'hourly', state: 'running' }, + select: { attemptId: true, id: true, processRunId: true, revision: true }, + }); + } + async hourly(userId: string) { const connection = await this.prisma.tbankConnection.findFirst({ where: { userId }, diff --git a/test/tbank-hourly-lifecycle.e2e-spec.ts b/test/tbank-hourly-lifecycle.e2e-spec.ts new file mode 100644 index 0000000..a564679 --- /dev/null +++ b/test/tbank-hourly-lifecycle.e2e-spec.ts @@ -0,0 +1,525 @@ +import { ConfigService } from '@nestjs/config'; +import assert from 'node:assert/strict'; +import { randomUUID } from 'node:crypto'; +import { setImmediate as immediate } from 'node:timers/promises'; + +import { TbankAccessService } from '../src/integrations/tbank/tbank-access.service'; +import { TbankTokenService } from '../src/integrations/tbank/tbank-token.service'; +import { CatalogEnrichmentService } from '../src/modules/investments/instruments/application/catalog-enrichment.service'; +import { CatalogDiscoveryRepository } from '../src/modules/investments/instruments/infrastructure/catalog-discovery.repository'; +import { BrokerageOperationsSyncService } from '../src/modules/investments/operations/application/brokerage-operations-sync.service'; +import { BrokerageOperationsRepository } from '../src/modules/investments/operations/infrastructure/brokerage-operations.repository'; +import { TbankHourlyScheduler } from '../src/modules/investments/tbank-sync/tbank-hourly.scheduler'; +import { TbankSyncRepository } from '../src/modules/investments/tbank-sync/tbank-sync.repository'; +import { PrismaService } from '../src/prisma/prisma.service'; + +type HourlyProgressView = { + result?: string; + sections: { accountId?: string; errorCode: null | string; section: string }[]; +}; + +// Real SQLite proves reservation, outcome derivation, recovery and fences; +// the step owners are controlled so each failure mode can be injected. +describe('hourly attempt lifecycle (SQLite)', () => { + const prisma = new PrismaService( + new ConfigService({ DATABASE_URL: process.env.DATABASE_URL }), + ); + const tokens = new TbankTokenService( + new ConfigService({ + TOKEN_ENCRYPTION_KEY: process.env.TOKEN_ENCRYPTION_KEY, + }) as never, + ); + const repository = new TbankSyncRepository(prisma); + const access = { local: jest.fn() }; + const membership = { request: jest.fn() }; + const portfolio = { request: jest.fn() }; + const operations = { request: jest.fn() }; + const events = { refresh: jest.fn(), get: jest.fn() }; + const logger = { event: jest.fn() }; + const createScheduler = () => + new TbankHourlyScheduler( + prisma, + repository, + access as never, + membership as never, + portfolio as never, + operations as never, + events as never, + logger as never, + ); + const USER = 'hourly-owner'; + const REVISION = 'hourly-revision'; + + function child( + state: 'complete' | 'failed' | 'running', + errorCode: null | string, + scopeKey: string, + ) { + return prisma.tbankSyncWork.create({ + data: { + id: randomUUID(), + userId: USER, + revision: REVISION, + scopeKey: `test:${scopeKey}:${randomUUID()}`, + attemptId: randomUUID(), + processRunId: 'child-process', + priority: 0, + intent: {}, + progress: {}, + state, + errorCode, + }, + }); + } + + function hourlyRow() { + return prisma.tbankSyncWork.findFirst({ + where: { userId: USER, scopeKey: 'hourly' }, + }); + } + + async function pump(turns = 20) { + for (let i = 0; i < turns; i++) { + await immediate(); + await jest.advanceTimersByTimeAsync(50); + } + } + + async function runWalk(scheduler: TbankHourlyScheduler, advanceMs = 50) { + await scheduler.tick(); + for (let i = 0; i < 400; i++) { + await immediate(); + await jest.advanceTimersByTimeAsync(advanceMs); + const row = await hourlyRow(); + if (row && row.state !== 'running') return row; + } + throw new Error('hourly attempt did not settle'); + } + + beforeAll(async () => { + await prisma.$connect(); + await prisma.user.create({ + data: { id: USER, email: 'hourly@example.invalid', passwordHash: 'x' }, + }); + }); + afterAll(async () => prisma.$disconnect()); + beforeEach(async () => { + jest.resetAllMocks(); + await prisma.tbankSyncWork.deleteMany(); + await prisma.tbankConnection.deleteMany(); + await prisma.tbankConnection.create({ + data: { + userId: USER, + revision: REVISION, + encryptedToken: tokens.encrypt('synthetic'), + connectedAt: new Date(), + }, + }); + jest.useFakeTimers({ + doNotFake: ['queueMicrotask', 'setImmediate', 'nextTick'], + }); + access.local.mockResolvedValue({ + revision: REVISION, + accounts: [{ accountId: 'a' }, { accountId: 'b' }], + }); + membership.request.mockImplementation(async () => { + const row = await child('complete', null, 'membership'); + return { id: row.id, done: Promise.resolve() }; + }); + portfolio.request.mockImplementation( + async (_userId: string, _revision: string, accountId: string) => { + const row = await child('complete', null, `portfolio:${accountId}`); + return { id: row.id, done: Promise.resolve() }; + }, + ); + operations.request.mockImplementation(async () => { + const row = await child('complete', null, 'operations'); + return row.id; + }); + events.refresh.mockResolvedValue([]); + events.get.mockResolvedValue(undefined); + }); + afterEach(() => { + jest.useRealTimers(); + jest.restoreAllMocks(); + }); + + it('settles the first walk succeeded only when every step row completed', async () => { + const row = await runWalk(createScheduler()); + expect(row.state).toBe('complete'); + const progress = row.progress as HourlyProgressView; + expect(progress.result).toBe('succeeded'); + expect(progress.sections).toEqual([]); + const read = await repository.hourly(USER); + expect(read?.state).toBe('succeeded'); + }); + + it('settles partial and keeps successful steps when one account fails', async () => { + portfolio.request.mockImplementation( + async (_userId: string, _revision: string, accountId: string) => { + const failed = accountId === 'b'; + const row = await child( + failed ? 'failed' : 'complete', + failed ? 'TBANK_UNAVAILABLE' : null, + `portfolio:${accountId}`, + ); + return { id: row.id, done: Promise.resolve() }; + }, + ); + const row = await runWalk(createScheduler()); + const progress = row.progress as HourlyProgressView; + expect(progress.result).toBe('partial'); + expect(progress.sections).toHaveLength(1); + expect(progress.sections[0].section).toBe('portfolio'); + expect(progress.sections[0].accountId).toBe('b'); + expect(progress.sections[0].errorCode).toBe('TBANK_UNAVAILABLE'); + }); + + it('settles failed when the membership step fails before any update', async () => { + membership.request.mockImplementation(async () => { + const row = await child('failed', 'INVALID_TBANK_TOKEN', 'membership'); + return { id: row.id, done: Promise.resolve() }; + }); + const row = await runWalk(createScheduler()); + const progress = row.progress as HourlyProgressView; + expect(progress.result).toBe('failed'); + expect(row.errorCode).toBe('INVALID_TBANK_TOKEN'); + expect(progress.sections[0].section).toBe('membership'); + }); + + it('never succeeds when no required step succeeded', async () => { + membership.request.mockImplementation(async () => { + const row = await child('failed', 'TBANK_UNAVAILABLE', 'membership'); + return { id: row.id, done: Promise.resolve() }; + }); + portfolio.request.mockImplementation( + async (_userId: string, _revision: string, accountId: string) => { + const row = await child( + 'failed', + 'TBANK_UNAVAILABLE', + `portfolio:${accountId}`, + ); + return { id: row.id, done: Promise.resolve() }; + }, + ); + operations.request.mockRejectedValue(new Error('db')); + events.refresh.mockRejectedValue(new Error('db')); + const row = await runWalk(createScheduler()); + const progress = row.progress as HourlyProgressView; + expect(progress.result).toBe('failed'); + expect(progress.result).not.toBe('succeeded'); + assert.ok(progress.sections.length > 0); + }); + + it('caps the outcome at partial when the coordinator itself breaks', async () => { + access.local.mockRejectedValueOnce(new Error('db')); + const row = await runWalk(createScheduler()); + const progress = row.progress as HourlyProgressView; + expect(progress.result).toBe('partial'); + expect(row.errorCode).toBe('INTERRUPTED'); + }); + + it('records a section whose dispatch failed as a problem', async () => { + events.refresh.mockRejectedValue(new Error('db')); + const row = await runWalk(createScheduler()); + const progress = row.progress as HourlyProgressView; + expect(progress.result).toBe('partial'); + const eventsSection = progress.sections.find( + (entry) => entry.section === 'events', + ); + assert.ok(eventsSection); + expect(eventsSection.errorCode).toBe('TBANK_UNAVAILABLE'); + expect(row.errorCode).toBe('TBANK_UNAVAILABLE'); + }); + + it('closes a stuck step as DEADLINE and keeps its row running', async () => { + operations.request.mockImplementation(async () => { + const row = await child('running', null, 'operations'); + return row.id; + }); + const scheduler = createScheduler(); + await scheduler.tick(); + await pump(); + await jest.advanceTimersByTimeAsync(16 * 60 * 1000); + await pump(20); + const row = await hourlyRow(); + assert.ok(row); + expect(row.state).not.toBe('running'); + const progress = row.progress as HourlyProgressView; + expect(progress.result).toBe('partial'); + assert.ok( + progress.sections.some( + (entry) => + entry.section === 'operations' && entry.errorCode === 'DEADLINE', + ), + ); + const stuck = await prisma.tbankSyncWork.findFirst({ + where: { userId: USER, state: 'running' }, + }); + assert.ok(stuck, 'the wait bound does not cancel the upstream call'); + }); + + it('recovers a lost settle and unblocks the next attempt at its term', async () => { + const settleSpy = jest + .spyOn(repository, 'settleHourly') + .mockRejectedValueOnce(new Error('db')); + const scheduler = createScheduler(); + await scheduler.tick(); + for (let i = 0; i < 40 && settleSpy.mock.calls.length === 0; i++) { + await immediate(); + await jest.advanceTimersByTimeAsync(50); + } + expect(settleSpy.mock.calls).toHaveLength(1); + const open = await hourlyRow(); + expect(open?.state).toBe('running'); + await pump(20); + await scheduler.tick(); + await pump(20); + const closed = await hourlyRow(); + expect(closed?.state).toBe('complete'); + expect((closed?.progress as HourlyProgressView).result).toBe('partial'); + expect(closed?.errorCode).toBe('INTERRUPTED'); + await jest.advanceTimersByTimeAsync(61 * 60 * 1000); + await scheduler.tick(); + await pump(30); + const next = await hourlyRow(); + expect(next?.attemptId).not.toBe(closed?.attemptId); + expect(next?.state).not.toBe('running'); + expect((next?.progress as HourlyProgressView).result).toBe('succeeded'); + }); + + it('closes an abandoned foreign-process attempt and keeps the normal term', async () => { + const stuck = await child('complete', null, 'membership'); + const deadAttemptId = randomUUID(); + await prisma.tbankSyncWork.create({ + data: { + id: randomUUID(), + userId: USER, + revision: REVISION, + scopeKey: 'hourly', + attemptId: deadAttemptId, + processRunId: 'dead-process', + priority: 0, + intent: {}, + state: 'running', + progress: { + children: [{ id: stuck.id, section: 'membership' }], + attempts: { [stuck.id]: stuck.attemptId }, + sections: [], + }, + }, + }); + const scheduler = createScheduler(); + await runWalk(scheduler); + const closed = await hourlyRow(); + expect(closed?.attemptId).toBe(deadAttemptId); + expect(closed?.state).not.toBe('running'); + expect((closed?.progress as HourlyProgressView).result).toBe('partial'); + expect(closed?.errorCode).toBe('INTERRUPTED'); + const hourlyCount = await prisma.tbankSyncWork.count({ + where: { userId: USER, scopeKey: 'hourly' }, + }); + expect(hourlyCount).toBe(1); + await jest.advanceTimersByTimeAsync(61 * 60 * 1000); + await scheduler.tick(); + await pump(30); + const next = await hourlyRow(); + expect(next?.attemptId).not.toBe(deadAttemptId); + expect(next?.state).not.toBe('running'); + }); + + it('does not overlap a running attempt with a second walk', async () => { + operations.request.mockImplementation(async () => { + const row = await child('running', null, 'operations'); + return row.id; + }); + const scheduler = createScheduler(); + await scheduler.tick(); + await pump(); + expect((await hourlyRow())?.state).toBe('running'); + await scheduler.tick(); + await pump(); + const hourlyCount = await prisma.tbankSyncWork.count({ + where: { userId: USER, scopeKey: 'hourly' }, + }); + expect(hourlyCount).toBe(1); + expect((await hourlyRow())?.state).toBe('running'); + await jest.advanceTimersByTimeAsync(16 * 60 * 1000); + await pump(20); + expect((await hourlyRow())?.state).not.toBe('running'); + }); + + it('does not start a new attempt before the next hourly term', async () => { + const scheduler = createScheduler(); + const row = await runWalk(scheduler); + const firstAttemptId = row.attemptId; + expect(row.state).toBe('complete'); + await scheduler.tick(); + expect((await hourlyRow())?.attemptId).toBe(firstAttemptId); + await jest.advanceTimersByTimeAsync(59 * 60 * 1000); + await scheduler.tick(); + expect((await hourlyRow())?.attemptId).toBe(firstAttemptId); + await jest.advanceTimersByTimeAsync(2 * 60 * 1000); + await scheduler.tick(); + await pump(30); + expect((await hourlyRow())?.attemptId).not.toBe(firstAttemptId); + }); + + it('counts a late step result before the attempt outcome is recorded', async () => { + const late: { id: string }[] = []; + operations.request.mockImplementation(async () => { + const row = await child('running', null, 'operations'); + late.push(row); + return row.id; + }); + const scheduler = createScheduler(); + await scheduler.tick(); + await pump(); + assert.ok(late.length > 0); + for (const row of late) { + await prisma.tbankSyncWork.update({ + where: { id: row.id }, + data: { state: 'complete' }, + }); + } + let settled = false; + for (let i = 0; i < 100 && !settled; i++) { + await immediate(); + await jest.advanceTimersByTimeAsync(200); + const current = await hourlyRow(); + settled = Boolean(current && current.state !== 'running'); + } + const row = await hourlyRow(); + expect(row).not.toBeNull(); + expect(row?.state).not.toBe('running'); + expect((row?.progress as HourlyProgressView).result).toBe('succeeded'); + }); + + it('treats a step superseded by a successor as stale and keeps CAS fencing', async () => { + const stale = await child('complete', null, 'membership'); + const attemptId = randomUUID(); + const id = randomUUID(); + await prisma.tbankSyncWork.create({ + data: { + id, + userId: USER, + revision: REVISION, + scopeKey: 'hourly', + attemptId, + processRunId: 'dead-process', + priority: 0, + intent: {}, + state: 'running', + progress: { + children: [{ id: stale.id, section: 'membership' }], + attempts: { [stale.id]: randomUUID() }, + sections: [], + }, + }, + }); + const foreign = await repository.settleHourly( + { userId: USER, revision: REVISION, id, attemptId: randomUUID() }, + 'INTERRUPTED', + ); + expect(foreign.count).toBe(0); + expect((await hourlyRow())?.state).toBe('running'); + const closed = await repository.settleHourly( + { userId: USER, revision: REVISION, id, attemptId }, + 'INTERRUPTED', + ); + expect(closed.count).toBe(1); + const progress = (await hourlyRow())?.progress as HourlyProgressView; + expect(progress.result).toBe('failed'); + expect(progress.sections[0].errorCode).toBeNull(); + const late = await repository.settleHourly( + { userId: USER, revision: REVISION, id, attemptId }, + 'INTERRUPTED', + ); + expect(late.count).toBe(0); + }); + + it('creates no reservations after a controlled shutdown', async () => { + const scheduler = createScheduler(); + scheduler.onModuleInit(); + scheduler.onModuleDestroy(); + await jest.advanceTimersByTimeAsync(2 * 60 * 60 * 1000); + await pump(5); + expect(await prisma.tbankSyncWork.count({ where: { userId: USER } })).toBe(0); + }); + + it('closes the operations step durably when its finalizer cannot record', async () => { + await prisma.tbankConnection.update({ + where: { userId: USER }, + data: { membershipAt: new Date() }, + }); + await prisma.tbankAccount.create({ + data: { + userId: USER, + revision: REVISION, + accountId: 'a', + type: 'BROKER', + name: 'Test', + upstreamOrder: 0, + }, + }); + const accessReal = new TbankAccessService(prisma, tokens); + const operationsRepository = new BrokerageOperationsRepository(prisma); + const sync = new BrokerageOperationsSyncService( + operationsRepository, + accessReal, + { getOperationsByCursor: jest.fn() } as never, + logger as never, + ); + jest + .spyOn(operationsRepository, 'pending') + .mockRejectedValueOnce(new Error('db')); + jest + .spyOn(operationsRepository, 'fail') + .mockRejectedValueOnce(new Error('db')); + const id = await sync.request( + USER, + [ + { + accountId: 'a', + from: new Date(Date.UTC(2026, 0, 1)), + to: new Date(), + }, + ] as never, + true, + 'screen', + REVISION, + ); + assert.ok(id); + await pump(40); + const row = await prisma.tbankSyncWork.findFirst({ + where: { userId: USER, id }, + }); + expect(row?.state).toBe('failed'); + }); + + it('finishes open catalog steps durably when publication fails mid-batch', async () => { + const accessReal = new TbankAccessService(prisma, tokens); + const catalogRepository = new CatalogDiscoveryRepository(prisma); + const enrichment = new CatalogEnrichmentService( + catalogRepository, + accessReal, + { getLastPrices: jest.fn().mockResolvedValue([]) } as never, + logger as never, + {} as never, + {} as never, + {} as never, + ); + jest + .spyOn(catalogRepository, 'publishSource') + .mockRejectedValue(new Error('db')); + await enrichment.requestMany(USER, REVISION, 'share', ['uid'], false, [ + 'last-price', + ]); + await pump(40); + const rows = await prisma.tbankSyncWork.findMany({ + where: { userId: USER, scopeKey: { not: 'hourly' } }, + }); + assert.ok(rows.length > 0); + for (const row of rows) expect(row.state).toBe('failed'); + }); +});