fix: skip hourly terms overlapped by a long walk (WealthLedger/backend#54)
This commit was merged in pull request #93.
This commit is contained in:
@@ -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
|
||||||
|
|||||||
@@ -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');
|
||||||
|
|||||||
Reference in New Issue
Block a user