From 5bc3980139cd64f50be9ffdfcd7975421d4bceb3 Mon Sep 17 00:00:00 2001
From: Sergey Krylov <ksv741@mail.ru>
Date: Wed, 14 Oct 2026 22:16:13 +0300
Subject: [PATCH] fix: skip hourly terms overlapped by a long walk
 (WealthLedger/backend#54)

---
 .../tbank-sync/tbank-sync.repository.ts       | 14 +++-
 test/tbank-hourly-lifecycle.e2e-spec.ts       | 77 +++++++++++++++++++
 2 files changed, 89 insertions(+), 2 deletions(-)

diff --git a/src/modules/investments/tbank-sync/tbank-sync.repository.ts b/src/modules/investments/tbank-sync/tbank-sync.repository.ts
index b10fbcc..dbeab81 100644
--- a/src/modules/investments/tbank-sync/tbank-sync.repository.ts
+++ b/src/modules/investments/tbank-sync/tbank-sync.repository.ts
@@ -183,7 +183,8 @@ export class TbankSyncRepository {
 
   // The timer only proposes work; this reservation is the durable fence. The
   // attempt start (createdAt) is the basis of the next hourly term, including
-  // failed and interrupted attempts.
+  // failed and interrupted attempts. After downtime the next tick is due at
+  // once, so no catch-up queue forms.
   reserveHourly(userId: string, runId: string, hourMs: number) {
     return this.prisma.$transaction(async (tx) => {
       const connection = await tx.tbankConnection.findFirst({
@@ -201,7 +202,16 @@ export class TbankSyncRepository {
       });
       const now = new Date();
       if (previous) {
-        const due = now.valueOf() - previous.createdAt.valueOf() >= hourMs;
+        // Terms are hourMs steps from the attempt start; a walk still running
+        // at a term consumes it. A dead process's attempt has no known end, so
+        // it consumes only its own term.
+        const started = previous.createdAt.valueOf();
+        const ended =
+          previous.state === 'running' ? started : previous.updatedAt.valueOf();
+        const dueAt =
+          started +
+          Math.max(hourMs, Math.ceil((ended - started) / hourMs) * hourMs);
+        const due = now.valueOf() >= dueAt;
         if (previous.state === 'running') {
           if (previous.processRunId === runId) return null;
           // A dead process cannot finish its attempt; close it from its
diff --git a/test/tbank-hourly-lifecycle.e2e-spec.ts b/test/tbank-hourly-lifecycle.e2e-spec.ts
index 08a1509..b9bdae5 100644
--- a/test/tbank-hourly-lifecycle.e2e-spec.ts
+++ b/test/tbank-hourly-lifecycle.e2e-spec.ts
@@ -385,6 +385,83 @@ describe('hourly attempt lifecycle (SQLite)', () => {
     expect(next?.state).not.toBe('running');
   });
 
+  describe('hourly term eligibility', () => {
+    const at = (h: number, m = 0) => new Date(Date.UTC(2026, 0, 1, h, m));
+    const seed = (
+      state: 'complete' | 'failed' | 'running',
+      start: Date,
+      end: Date,
+      processRunId = 'other-process',
+    ) =>
+      prisma.tbankSyncWork.create({
+        data: {
+          id: randomUUID(),
+          userId: USER,
+          revision: REVISION,
+          scopeKey: 'hourly',
+          attemptId: randomUUID(),
+          processRunId,
+          priority: 0,
+          intent: {},
+          progress: {},
+          state,
+          createdAt: start,
+          updatedAt: end,
+        },
+      });
+    const reserve = async (when: Date) => {
+      jest.setSystemTime(when);
+      return repository.reserveHourly(USER, 'this-process', 3_600_000);
+    };
+
+    it.each([
+      ['short attempt', 'complete', at(10), at(10, 5), at(10, 59), at(11)],
+      ['failed attempt', 'failed', at(10), at(10, 5), at(10, 59), at(11)],
+      ['walk past one term', 'complete', at(10), at(11, 10), at(11, 59), at(12)],
+      ['walk past two terms', 'complete', at(10), at(12, 10), at(12, 59), at(13)],
+      ['walk ending on a term', 'complete', at(10), at(11), at(10, 59), at(11)],
+    ] as const)('%s', async (_name, state, start, end, early, due) => {
+      await seed(state, start, end);
+      expect(await reserve(early)).toBeNull();
+      const reserved = await reserve(due);
+      expect(reserved?.work.state).toBe('running');
+      expect(reserved?.work.createdAt).toEqual(due);
+    });
+
+    it('starts a new revision at once', async () => {
+      await seed('complete', at(10), at(10, 5));
+      await prisma.tbankConnection.deleteMany({ where: { userId: USER } });
+      await prisma.tbankConnection.create({
+        data: {
+          userId: USER,
+          revision: 'next-revision',
+          encryptedToken: tokens.encrypt('synthetic'),
+          connectedAt: at(10, 10),
+        },
+      });
+      expect((await reserve(at(10, 10)))?.work.revision).toBe('next-revision');
+    });
+
+    it('closes a dead process attempt without early retry and runs one catch-up after downtime', async () => {
+      await seed('running', at(10), at(10, 1), 'dead-process');
+      expect(await reserve(at(10, 30))).toBeNull();
+      expect((await hourlyRow())?.errorCode).toBe('INTERRUPTED');
+      expect(await reserve(at(10, 59))).toBeNull();
+      expect((await reserve(at(14, 30)))?.work.createdAt).toEqual(at(14, 30));
+      expect(await reserve(at(14, 31))).toBeNull();
+      expect(
+        await prisma.tbankSyncWork.count({
+          where: { userId: USER, scopeKey: 'hourly' },
+        }),
+      ).toBe(1);
+    });
+
+    it('does not reserve while this process still runs the attempt', async () => {
+      await seed('running', at(10), at(10, 1), 'this-process');
+      expect(await reserve(at(12, 30))).toBeNull();
+    });
+  });
+
   it('does not overlap a running attempt with a second walk', async () => {
     operations.request.mockImplementation(async () => {
       const row = await child('running', null, 'operations');