test/tbank-operations-sync.e2e-spec.ts

@@ -76,7 +76,7 @@ describe('Request-driven operations synchronization (SQLite)', () => {
         where: { id },
       });
       if (work.state === expected) return work;
-      await delay(2);
+      await delay(2); // Backoff between exact DB-state polls; not a timing barrier.
     }
     throw new Error(`Sync did not reach ${expected}`);
   }
@@ -113,6 +113,7 @@ describe('Request-driven operations synchronization (SQLite)', () => {
     });
     service = createService();
   });
+  afterEach(() => jest.restoreAllMocks());
   afterAll(async () => prisma.$disconnect());
 
   it.each([[false, 1], [true, 1], [false, 2], [true, 2]] as const)(
@@ -186,13 +187,17 @@ describe('Request-driven operations synchronization (SQLite)', () => {
 
   it('publishes the forced visible period before unrelated pending history and retains it if history fails', async () => {
     const gate = deferred<TbankOperationsByCursorResponse>();
+    const started = deferred<void>();
     const march = {
       ...january,
       from: new Date('2026-03-01Z'),
       to: new Date('2026-04-01Z'),
     };
     upstream.getOperationsByCursor
-      .mockReturnValueOnce(gate.promise)
+      .mockImplementationOnce(() => {
+        started.resolve();
+        return gate.promise;
+      })
       .mockImplementation((_token, request) => {
         if (Number(request.from?.seconds) + 1 === march.from.valueOf() / 1000) {
           return Promise.resolve({
@@ -205,7 +210,9 @@ describe('Request-driven operations synchronization (SQLite)', () => {
     const history = requiredHandle(
       await service.request('owner', [february, march], false, 'history'),
     );
-    await delay(5);
+    // RPC starts after reserve/claim commits and pending reads the history plan.
+    // Waiting for state=running alone would not ensure that pending was read.
+    await started.promise;
     const screen = requiredHandle(
       await service.request('owner', [march], true),
     );
@@ -225,8 +232,12 @@ describe('Request-driven operations synchronization (SQLite)', () => {
 
   it('moves a newly visible interval ahead of pending backfill and records an unknown requested floor', async () => {
     const gate = deferred<TbankOperationsByCursorResponse>();
+    const started = deferred<void>();
     upstream.getOperationsByCursor
-      .mockReturnValueOnce(gate.promise)
+      .mockImplementationOnce(() => {
+        started.resolve();
+        return gate.promise;
+      })
       .mockResolvedValue({ hasNext: false, items: [] });
     const march = {
       ...january,
@@ -241,7 +252,7 @@ describe('Request-driven operations synchronization (SQLite)', () => {
         'history',
       ),
     );
-    await delay(5);
+    await started.promise;
     await service.request('owner', [march], false);
     gate.resolve({ hasNext: false, items: [] });
     await state(history, 'complete');
@@ -471,7 +482,7 @@ describe('Request-driven operations synchronization (SQLite)', () => {
         expect(saved.state).toBe(
           checkpoint === 'after-publish' ? 'complete' : 'running',
         );
-        await delay(10);
+        await delay(10); // Bounded pause; absence of a recovery RPC cannot be awaited.
         expect(upstream.getOperationsByCursor).not.toHaveBeenCalled();
         if (checkpoint === 'after-publish') {
           // eslint-disable-next-line jest/no-conditional-expect -- The parameterized fixture chooses the scenario; each branch asserts its outcome and the test also asserts common invariants.
@@ -504,13 +515,17 @@ describe('Request-driven operations synchronization (SQLite)', () => {
 
   it('joins a request bridging active groups and keeps both original handles observable', async () => {
     const blocked = deferred<typeof complete>();
+    const started = deferred<void>();
     upstream.getOperationsByCursor
-      .mockReturnValueOnce(blocked.promise)
+      .mockImplementationOnce(() => {
+        started.resolve();
+        return blocked.promise;
+      })
       .mockResolvedValue({ hasNext: false, items: [] });
     const early = { ...january, to: new Date('2026-01-10Z') };
     const late = { ...january, from: new Date('2026-01-20Z') };
     const first = requiredHandle(await service.request('owner', [early], true));
-    await delay(5);
+    await started.promise;
     const second = requiredHandle(await service.request('owner', [late], true));
     const merged = requiredHandle(
       await service.request('owner', [january], true),
@@ -526,25 +541,28 @@ describe('Request-driven operations synchronization (SQLite)', () => {
 
   it('extends an overlapping atomic group without rereading its intersection or publishing a partial result', async () => {
     const first = deferred<typeof complete>();
+    const started = deferred<void>();
     const remainder = deferred<typeof complete>();
+    const remainderStarted = deferred<void>();
     const early = { ...january, to: new Date('2026-01-20Z') };
     const late = { ...january, from: new Date('2026-01-10Z') };
     upstream.getOperationsByCursor
-      .mockReturnValueOnce(first.promise)
+      .mockImplementationOnce(() => {
-      .mockReturnValueOnce(remainder.promise);
+        started.resolve();
+        return first.promise;
+      })
+      .mockImplementationOnce(() => {
+        remainderStarted.resolve();
+        return remainder.promise;
+      });
     const handle = requiredHandle(
       await service.request('owner', [early], true),
     );
-    await delay(5);
+    await started.promise;
+    expect(upstream.getOperationsByCursor).toHaveBeenCalledTimes(1);
     expect(await service.request('owner', [late], true)).toBe(handle);
     first.resolve({ hasNext: false, items: [] });
-    for (
+    await remainderStarted.promise;
-      let index = 0;
-      index < 500 && upstream.getOperationsByCursor.mock.calls.length < 2;
-      index += 1
-    ) {
-      await delay(2);
-    }
     expect(upstream.getOperationsByCursor).toHaveBeenCalledTimes(2);
     expect(upstream.getOperationsByCursor.mock.calls[1][1].from).toEqual({
       seconds: String(early.to.valueOf() / 1000 - 1),
@@ -561,13 +579,17 @@ describe('Request-driven operations synchronization (SQLite)', () => {
       items: never[];
       nextCursor: string;
     }>();
+    const started = deferred<void>();
     upstream.getOperationsByCursor
-      .mockReturnValueOnce(first.promise)
+      .mockImplementationOnce(() => {
+        started.resolve();
+        return first.promise;
+      })
       .mockResolvedValue({ hasNext: false, items: [] });
     const history = requiredHandle(
       await service.request('owner', [january], false, 'history'),
     );
-    await delay(5);
+    await started.promise;
     const screen = requiredHandle(
       await service.request('owner', [february], true),
     );
@@ -583,19 +605,17 @@ describe('Request-driven operations synchronization (SQLite)', () => {
 
   it('keeps published initial windows on recovery and restarts only the unfinished month', async () => {
     const blocked = deferred<typeof complete>();
+    const blockedStarted = deferred<void>();
     upstream.getOperationsByCursor
       .mockResolvedValueOnce(complete)
-      .mockReturnValueOnce(blocked.promise);
+      .mockImplementationOnce(() => {
+        blockedStarted.resolve();
+        return blocked.promise;
+      });
     const handle = requiredHandle(
       await service.request('owner', [january, february], false),
     );
-    for (
+    await blockedStarted.promise;
-      let index = 0;
-      index < 500 && upstream.getOperationsByCursor.mock.calls.length < 2;
-      index += 1
-    ) {
-      await delay(2);
-    }
     expect(await prisma.tbankOperation.count()).toBe(1);
     const before = await prisma.tbankOperationCoverage.findMany();
     expect(before).toHaveLength(1);
@@ -616,8 +636,15 @@ describe('Request-driven operations synchronization (SQLite)', () => {
         })
       ).fetchedAt,
     ).toEqual(before[0].fetchedAt);
+    const stalePage = deferred<Promise<boolean>>();
+    const stagePage = repository.stagePage.bind(repository);
+    jest.spyOn(repository, 'stagePage').mockImplementation((...args) => {
+      const result = stagePage(...args);
+      stalePage.resolve(result);
+      return result;
+    });
     blocked.resolve(complete);
-    await delay(10);
+    expect(await stalePage.promise).toBe(false);
     expect(await prisma.tbankOperation.count()).toBe(1);
     expect(await prisma.tbankOperationCoverage.count()).toBe(2);
   });
@@ -651,7 +678,11 @@ describe('Request-driven operations synchronization (SQLite)', () => {
 
   it('coalesces twenty requests, observes without RPC and completes only after the last page', async () => {
     const blocked = deferred<typeof complete>();
-    upstream.getOperationsByCursor.mockReturnValueOnce(blocked.promise);
+    const started = deferred<void>();
+    upstream.getOperationsByCursor.mockImplementationOnce(() => {
+      started.resolve();
+      return blocked.promise;
+    });
     const handles = await Promise.all(
       Array.from({ length: 20 }, () =>
         service.request('owner', [january], true),
@@ -659,7 +690,7 @@ describe('Request-driven operations synchronization (SQLite)', () => {
     );
     expect(new Set(handles).size).toBe(1);
     const handle = requiredHandle(handles[0]);
-    await delay(5);
+    await started.promise;
     expect(upstream.getOperationsByCursor).toHaveBeenCalledTimes(1);
     expect((await service.observe('owner', handle)).state).toBe('running');
     expect(await prisma.tbankOperation.count()).toBe(0);
@@ -699,6 +730,7 @@ describe('Request-driven operations synchronization (SQLite)', () => {
     expect(await prisma.tbankOperationCoverage.findMany()).toEqual(oldCoverage);
     const calls = upstream.getOperationsByCursor.mock.calls.length;
     await service.observe('owner', handle);
+    // Bounded pause; absence of a background retry cannot be awaited.
     await delay(10);
     expect(upstream.getOperationsByCursor).toHaveBeenCalledTimes(calls);
     upstream.getOperationsByCursor
@@ -713,29 +745,29 @@ describe('Request-driven operations synchronization (SQLite)', () => {
 
   it('starts an unfinished interval from page one after a new run request and ignores late old pages', async () => {
     const blocked = deferred<typeof complete>();
+    const blockedStarted = deferred<void>();
     upstream.getOperationsByCursor
       .mockResolvedValueOnce({
         hasNext: true,
         nextCursor: 'provider-cursor',
         items: [operation()],
       })
-      .mockReturnValueOnce(blocked.promise);
+      .mockImplementationOnce(() => {
+        blockedStarted.resolve();
+        return blocked.promise;
+      });
     const handle = requiredHandle(
       await service.request('owner', [january], true),
     );
-    for (
+    await blockedStarted.promise;
-      let index = 0;
-      index < 500 && upstream.getOperationsByCursor.mock.calls.length < 2;
-      index += 1
-    ) {
-      await delay(2);
-    }
-    expect(upstream.getOperationsByCursor).toHaveBeenCalledTimes(2);
     const before = await prisma.tbankSyncWork.findUniqueOrThrow({
       where: { id: handle },
     });
+    expect(await prisma.tbankSyncStage.count()).toBeGreaterThan(0);
+
     service = createService();
     await service.observe('owner', handle);
+    // Bounded pause; absence of work adoption by observe cannot be awaited.
     await delay(5);
     expect(upstream.getOperationsByCursor).toHaveBeenCalledTimes(2);
     upstream.getOperationsByCursor.mockResolvedValueOnce({
@@ -752,8 +784,15 @@ describe('Request-driven operations synchronization (SQLite)', () => {
     expect(
       upstream.getOperationsByCursor.mock.calls[2][1].cursor,
     ).toBeUndefined();
+    const stalePage = deferred<Promise<boolean>>();
+    const stagePage = repository.stagePage.bind(repository);
+    jest.spyOn(repository, 'stagePage').mockImplementation((...args) => {
+      const result = stagePage(...args);
+      stalePage.resolve(result);
+      return result;
+    });
     blocked.resolve(complete);
-    await delay(10);
+    expect(await stalePage.promise).toBe(false);
     expect((await prisma.tbankOperation.findFirstOrThrow()).payload).toEqual(
       operation('30'),
     );