From a1cddd985744aeab20488f8e9b140f58bd2efe96 Mon Sep 17 00:00:00 2001 From: Sergey Krylov Date: Wed, 28 Oct 2026 20:28:55 +0300 Subject: [PATCH] test: replace operations sync delays with explicit barriers (WealthLedger/backend#104) --- test/tbank-operations-sync.e2e-spec.ts | 121 ++++++++++++++++--------- 1 file changed, 80 insertions(+), 41 deletions(-) diff --git a/test/tbank-operations-sync.e2e-spec.ts b/test/tbank-operations-sync.e2e-spec.ts index aad2971..11af629 100644 --- a/test/tbank-operations-sync.e2e-spec.ts +++ b/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(); + const started = deferred(); 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(); + const started = deferred(); 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(); + const started = deferred(); 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(); + const started = deferred(); const remainder = deferred(); + const remainderStarted = deferred(); const early = { ...january, to: new Date('2026-01-20Z') }; const late = { ...january, from: new Date('2026-01-10Z') }; upstream.getOperationsByCursor - .mockReturnValueOnce(first.promise) - .mockReturnValueOnce(remainder.promise); + .mockImplementationOnce(() => { + 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 ( - let index = 0; - index < 500 && upstream.getOperationsByCursor.mock.calls.length < 2; - index += 1 - ) { - await delay(2); - } + await remainderStarted.promise; 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(); 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(); + const blockedStarted = deferred(); upstream.getOperationsByCursor .mockResolvedValueOnce(complete) - .mockReturnValueOnce(blocked.promise); + .mockImplementationOnce(() => { + blockedStarted.resolve(); + return blocked.promise; + }); const handle = requiredHandle( await service.request('owner', [january, february], false), ); - for ( - let index = 0; - index < 500 && upstream.getOperationsByCursor.mock.calls.length < 2; - index += 1 - ) { - await delay(2); - } + await blockedStarted.promise; 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>(); + 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(); - upstream.getOperationsByCursor.mockReturnValueOnce(blocked.promise); + const started = deferred(); + 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(); + const blockedStarted = deferred(); 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 ( - let index = 0; - index < 500 && upstream.getOperationsByCursor.mock.calls.length < 2; - index += 1 - ) { - await delay(2); - } - expect(upstream.getOperationsByCursor).toHaveBeenCalledTimes(2); + await blockedStarted.promise; 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>(); + 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'), ); -- 2.54.0