fix: skip hourly terms overlapped by a long walk (WealthLedger/backend#54) #93

Merged
codex-bot merged 1 commits from fix/54-hourly-overlap into develop 2026-10-14 22:43:08 +03:00
2 changed files with 89 additions and 2 deletions
@@ -183,7 +183,8 @@ export class TbankSyncRepository {
// The timer only proposes work; this reservation is the durable fence. The // The timer only proposes work; this reservation is the durable fence. The
// attempt start (createdAt) is the basis of the next hourly term, including // 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) { reserveHourly(userId: string, runId: string, hourMs: number) {
return this.prisma.$transaction(async (tx) => { return this.prisma.$transaction(async (tx) => {
const connection = await tx.tbankConnection.findFirst({ const connection = await tx.tbankConnection.findFirst({
@@ -201,7 +202,16 @@ export class TbankSyncRepository {
}); });
const now = new Date(); const now = new Date();
if (previous) { 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.state === 'running') {
if (previous.processRunId === runId) return null; if (previous.processRunId === runId) return null;
// A dead process cannot finish its attempt; close it from its // A dead process cannot finish its attempt; close it from its
+77
View File
@@ -385,6 +385,83 @@ describe('hourly attempt lifecycle (SQLite)', () => {
expect(next?.state).not.toBe('running'); 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 () => { it('does not overlap a running attempt with a second walk', async () => {
operations.request.mockImplementation(async () => { operations.request.mockImplementation(async () => {
const row = await child('running', null, 'operations'); const row = await child('running', null, 'operations');