WealthLedger/backend
5
0
Fork 0
Code Issues 6 Pull Requests Actions Packages Projects Releases Wiki Activity

test: replace operations sync delays with explicit barriers (WealthLedger/backend#104) #145

Merged
codex-bot merged 1 commits from test/104-operations-sync-delays into develop 2026-10-28 20:41:24 +03:00
Conversation 0 Commits 1 Files Changed 1
+80 -41
1 changed files with 80 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit a1cddd9857 - Show all commits
test/tbank-operations-sync.e2e-spec.ts
+80 -41
View File
@@ -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)
.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<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 (
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<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 (
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<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'),
);