From 52fa3a36baa5ab236a7ebd64ef1b4f447c91e129 Mon Sep 17 00:00:00 2001 From: Sergey Krylov Date: Mon, 12 Oct 2026 22:17:53 +0300 Subject: [PATCH] fix: bound hourly step awaits by the per-step wait limit (#56) --- .../tbank-sync/tbank-hourly.scheduler.ts | 115 ++++++++++++------ test/tbank-hourly-lifecycle.e2e-spec.ts | 57 +++++++++ 2 files changed, 138 insertions(+), 34 deletions(-) diff --git a/src/modules/investments/tbank-sync/tbank-hourly.scheduler.ts b/src/modules/investments/tbank-sync/tbank-hourly.scheduler.ts index 17d0aa6..8f89f01 100644 --- a/src/modules/investments/tbank-sync/tbank-hourly.scheduler.ts +++ b/src/modules/investments/tbank-sync/tbank-hourly.scheduler.ts @@ -28,6 +28,9 @@ 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; +// An await that outlived WORK_WAIT_BOUND_MS. The underlying call keeps +// running; its late durable result is fenced like any other late publication. +const BOUND_EXPIRED = Symbol('hourly-bound-expired'); const FIRST_DAY_OF_MONTH = 1; const NO_MILLISECONDS = 0; @@ -155,7 +158,10 @@ export class TbankHourlyScheduler implements OnModuleDestroy, OnModuleInit { ); await this.waitWorks(userId, children.map((child) => child.id)); for (const account of accounts) { - await this.events.get(userId, account.accountId).catch(() => undefined); + const pending = this.events + .get(userId, account.accountId) + .catch(() => undefined); + await this.withBound(pending); } await this.settle(attempt, null, dispatchProblems); } catch { @@ -173,11 +179,14 @@ export class TbankHourlyScheduler implements OnModuleDestroy, OnModuleInit { dispatchProblems: HourlySection[], ) { const { userId, revision } = attempt; - const member = await this.membership.request(userId, revision, true); - if (member) { + const member = await this.withBound( + this.membership.request(userId, revision, true), + ); + if (member && member !== BOUND_EXPIRED) { await dispatch({ id: member.id, section: 'membership' }); - await member.done; + const done = await this.withBound(member.done); if ( + done !== BOUND_EXPIRED && (await this.repository.workError(userId, member.id)) === TbankErrorCode.InvalidToken ) { @@ -185,19 +194,45 @@ export class TbankHourlyScheduler implements OnModuleDestroy, OnModuleInit { return null; } } else { + // No durable step id (rejected request or an expired wait): record the + // observable problem; a late row stays fenced by its durable attempt id. dispatchProblems.push({ section: 'membership', errorCode: TbankErrorCode.Unavailable, }); } - const current = await this.access.local(userId); - if (current.revision !== revision) { + const current = await this.withBound(this.access.local(userId)); + if (current === BOUND_EXPIRED || current.revision !== revision) { await this.settle(attempt, HOURLY_CAUSE_INTERRUPTED); return null; } return current.accounts; } + // Races a step await against the same per-step bound the row wait uses, so + // a dispatch that never resolves cannot hold the attempt running forever. + // The call itself is not cancelled; a rejection landing after the expiry is + // swallowed here because the expiry already accounts the step as a problem. + private withBound(step: Promise): Promise { + step.catch(() => undefined); + return new Promise((resolve, reject) => { + const timer = setTimeout( + () => resolve(BOUND_EXPIRED), + WORK_WAIT_BOUND_MS, + ); + step.then( + (value) => { + clearTimeout(timer); + resolve(value); + }, + (error: unknown) => { + clearTimeout(timer); + reject(error instanceof Error ? error : new Error(String(error))); + }, + ); + }); + } + private async refreshAccounts( userId: string, revision: string, @@ -241,8 +276,10 @@ export class TbankHourlyScheduler implements OnModuleDestroy, OnModuleInit { dispatchProblems: HourlySection[], ) { try { - const work = await this.portfolio.request(userId, revision, accountId, true); - if (work) { + const work = await this.withBound( + this.portfolio.request(userId, revision, accountId, true), + ); + if (work && work !== BOUND_EXPIRED) { await dispatch({ accountId, id: work.id, section: 'portfolio' }); return; } @@ -264,27 +301,33 @@ export class TbankHourlyScheduler implements OnModuleDestroy, OnModuleInit { dispatchProblems: HourlySection[], ) { try { - const id = await this.operations.request( - userId, - [ - { - accountId, - from: monthStartOf(new Date()), - to: new Date(), - }, - ], - true, - 'screen', - revision, + const id = await this.withBound( + this.operations.request( + userId, + [ + { + accountId, + from: monthStartOf(new Date()), + to: new Date(), + }, + ], + true, + 'screen', + revision, + ), ); - if (id) await dispatch({ accountId, id, section: 'operations' }); + if (id && id !== BOUND_EXPIRED) { + await dispatch({ accountId, id, section: 'operations' }); + return; + } } catch { - dispatchProblems.push({ - accountId, - errorCode: TbankErrorCode.Unavailable, - section: 'operations', - }); + // A failed dispatch has no durable step row; record the problem. } + dispatchProblems.push({ + accountId, + errorCode: TbankErrorCode.Unavailable, + section: 'operations', + }); } private async dispatchEvents( @@ -294,17 +337,21 @@ export class TbankHourlyScheduler implements OnModuleDestroy, OnModuleInit { dispatchProblems: HourlySection[], ) { try { - const ids = await this.events.refresh(userId, accountId); - for (const id of ids) { - await dispatch({ accountId, id, section: 'events' }); + const ids = await this.withBound(this.events.refresh(userId, accountId)); + if (ids !== BOUND_EXPIRED) { + for (const id of ids) { + await dispatch({ accountId, id, section: 'events' }); + } + return; } } catch { - dispatchProblems.push({ - accountId, - errorCode: TbankErrorCode.Unavailable, - section: 'events', - }); + // A failed dispatch has no durable step row; record the problem. } + dispatchProblems.push({ + accountId, + errorCode: TbankErrorCode.Unavailable, + section: 'events', + }); } private async settle( diff --git a/test/tbank-hourly-lifecycle.e2e-spec.ts b/test/tbank-hourly-lifecycle.e2e-spec.ts index a564679..08a1509 100644 --- a/test/tbank-hourly-lifecycle.e2e-spec.ts +++ b/test/tbank-hourly-lifecycle.e2e-spec.ts @@ -96,6 +96,19 @@ describe('hourly attempt lifecycle (SQLite)', () => { throw new Error('hourly attempt did not settle'); } + // Sequential per-step bounds: advance past each bound until the attempt + // closes. Covers walks whose awaits hang before any durable step row exists. + async function runBoundedWalk(scheduler: TbankHourlyScheduler) { + await scheduler.tick(); + for (let i = 0; i < 12; i++) { + await pump(5); + await jest.advanceTimersByTimeAsync(16 * 60 * 1000); + 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({ @@ -259,6 +272,50 @@ describe('hourly attempt lifecycle (SQLite)', () => { assert.ok(stuck, 'the wait bound does not cancel the upstream call'); }); + it('settles the attempt when a portfolio dispatch never resolves', async () => { + portfolio.request.mockImplementation(() => new Promise(() => undefined)); + const row = await runBoundedWalk(createScheduler()); + const progress = row.progress as HourlyProgressView; + expect(progress.result).toBe('partial'); + const problems = progress.sections.filter( + (entry) => entry.section === 'portfolio', + ); + expect(problems).toHaveLength(2); + expect(problems[0].errorCode).toBe('TBANK_UNAVAILABLE'); + }); + + it('settles the attempt when the membership dispatch never resolves', async () => { + membership.request.mockImplementation(() => new Promise(() => undefined)); + const row = await runBoundedWalk(createScheduler()); + const progress = row.progress as HourlyProgressView; + expect(progress.result).toBe('partial'); + expect(progress.sections).toHaveLength(1); + expect(progress.sections[0].section).toBe('membership'); + expect(progress.sections[0].errorCode).toBe('TBANK_UNAVAILABLE'); + }); + + it('bounds a membership step promise that never resolves', async () => { + membership.request.mockImplementation(async () => { + const row = await child('running', null, 'membership'); + return { id: row.id, done: new Promise(() => undefined) }; + }); + const row = await runBoundedWalk(createScheduler()); + const progress = row.progress as HourlyProgressView; + expect(progress.result).toBe('partial'); + const membershipSection = progress.sections.find( + (entry) => entry.section === 'membership', + ); + expect(membershipSection?.errorCode).toBe('DEADLINE'); + }); + + it('settles the attempt when a post-wait refresh never resolves', async () => { + events.get.mockImplementation(() => new Promise(() => undefined)); + const row = await runBoundedWalk(createScheduler()); + const progress = row.progress as HourlyProgressView; + expect(progress.result).toBe('succeeded'); + expect(progress.sections).toEqual([]); + }); + it('recovers a lost settle and unblocks the next attempt at its term', async () => { const settleSpy = jest .spyOn(repository, 'settleHourly') -- 2.54.0