fix: restore terminal hourly lifecycle after persistence failure (#56)

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<void>; resolve: () => void; workId: string }
+    { attempt: TbankSyncAttempt; done: Promise<void>; resolve: () => void; workId: string }
   >();
 
   async ensureDetail(
@@ -208,7 +208,7 @@ export class CatalogEnrichmentService {
         const done = new Promise<void>((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();

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

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<void>;
+type Accounts = Awaited<ReturnType<TbankAccessService['local']>>['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<string>();
   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<ReturnType<TbankAccessService['local']>>['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<string, number>,
+    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<string, SettledWork>();
-    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()];
   }
 }
 

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<string, string>;
+  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 },