Merge pull request 'fix: bound hourly step awaits by the per-step wait limit (#56)' (#78) from fix/56-hourly-step-await-bound into develop

Reviewed-on: #78

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);
+    const member = await this.withBound(
-    if (member) {
+      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);
+    const current = await this.withBound(this.access.local(userId));
-    if (current.revision !== revision) {
+    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<T>(step: Promise<T>): Promise<T | typeof BOUND_EXPIRED> {
+    step.catch(() => undefined);
+    return new Promise<T | typeof BOUND_EXPIRED>((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);
+      const work = await this.withBound(
-      if (work) {
+        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(
+      const id = await this.withBound(
-        userId,
+        this.operations.request(
-        [
+          userId,
-          {
+          [
-            accountId,
+            {
-            from: monthStartOf(new Date()),
+              accountId,
-            to: new Date(),
+              from: monthStartOf(new Date()),
-          },
+              to: new Date(),
-        ],
+            },
-        true,
+          ],
-        'screen',
+          true,
-        revision,
+          '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({
+      // A failed dispatch has no durable step row; record the problem.
-        accountId,
-        errorCode: TbankErrorCode.Unavailable,
-        section: 'operations',
-      });
     }
+    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);
+      const ids = await this.withBound(this.events.refresh(userId, accountId));
-      for (const id of ids) {
+      if (ids !== BOUND_EXPIRED) {
-        await dispatch({ accountId, id, section: 'events' });
+        for (const id of ids) {
+          await dispatch({ accountId, id, section: 'events' });
+        }
+        return;
       }
     } catch {
-      dispatchProblems.push({
+      // A failed dispatch has no durable step row; record the problem.
-        accountId,
-        errorCode: TbankErrorCode.Unavailable,
-        section: 'events',
-      });
     }
+    dispatchProblems.push({
+      accountId,
+      errorCode: TbankErrorCode.Unavailable,
+      section: 'events',
+    });
   }
 
   private async settle(

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