test(planning): make the checkpoint cancellation test deterministic
The cancelled stage takes about a millisecond, so under load the worker could finish it before the cancel message arrived and report it as completed. The real-worker harness now holds the worker after reporting that stage until the cancel is delivered, keeping every assertion. Refs #291 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
ac69a41fc8
commit
dfe6d28d2b
1 file changed
+21
-4
@@ -31,6 +31,7 @@ describe('real planning worker cancellation', () => {
|
|||||||
let holdArtifacts = false;
|
let holdArtifacts = false;
|
||||||
const heldArtifactMessages: (() => void)[] = [];
|
const heldArtifactMessages: (() => void)[] = [];
|
||||||
let simulateLongCalculation = false;
|
let simulateLongCalculation = false;
|
||||||
|
let holdForCancelAtStage: string | undefined;
|
||||||
let simulatedElapsed = 0;
|
let simulatedElapsed = 0;
|
||||||
const threads = new Set<NodeWorker>();
|
const threads = new Set<NodeWorker>();
|
||||||
const sentTypes: string[] = [];
|
const sentTypes: string[] = [];
|
||||||
@@ -43,8 +44,9 @@ describe('real planning worker cancellation', () => {
|
|||||||
const entry = join(directory, 'entry.ts');
|
const entry = join(directory, 'entry.ts');
|
||||||
writeFileSync(
|
writeFileSync(
|
||||||
entry,
|
entry,
|
||||||
`import { parentPort, workerData } from 'node:worker_threads';
|
`import { parentPort, receiveMessageOnPort, workerData } from 'node:worker_threads';
|
||||||
let elapsed = 0;
|
let elapsed = 0;
|
||||||
|
const pause = new Int32Array(new SharedArrayBuffer(4));
|
||||||
const originalNow = performance.now.bind(performance);
|
const originalNow = performance.now.bind(performance);
|
||||||
if (workerData.simulateLongCalculation)
|
if (workerData.simulateLongCalculation)
|
||||||
Object.defineProperty(performance, 'now', { value: () => originalNow() + elapsed });
|
Object.defineProperty(performance, 'now', { value: () => originalNow() + elapsed });
|
||||||
@@ -52,6 +54,16 @@ globalThis.self = { postMessage: (message) => {
|
|||||||
if (workerData.simulateLongCalculation && message.type === 'progress' && message.stage === 'optimization:default-unplaced-at-risk')
|
if (workerData.simulateLongCalculation && message.type === 'progress' && message.stage === 'optimization:default-unplaced-at-risk')
|
||||||
setTimeout(() => { elapsed = 20_000; parentPort.postMessage({ type: 'test-clock', elapsed }); }, 0);
|
setTimeout(() => { elapsed = 20_000; parentPort.postMessage({ type: 'test-clock', elapsed }); }, 0);
|
||||||
parentPort.postMessage(message);
|
parentPort.postMessage(message);
|
||||||
|
// The worker reads its queue only between event-loop turns, so a stage that is shorter than the
|
||||||
|
// host round trip finishes first. Hold the stage until the host's cancel arrives, then deliver it.
|
||||||
|
if (message.type === 'progress' && message.stage === workerData.holdForCancelAtStage) {
|
||||||
|
let next;
|
||||||
|
for (let waits = 0; !(next = receiveMessageOnPort(parentPort)); waits++) {
|
||||||
|
if (waits > 5000) throw new Error('The host never sent the cancel request.');
|
||||||
|
Atomics.wait(pause, 0, 0, 1);
|
||||||
|
}
|
||||||
|
globalThis.self.onmessage({ data: next.message });
|
||||||
|
}
|
||||||
} };
|
} };
|
||||||
await import(${JSON.stringify(resolve('src/app/services/leave-planner.worker.ts'))});
|
await import(${JSON.stringify(resolve('src/app/services/leave-planner.worker.ts'))});
|
||||||
parentPort.on('message', (data) => globalThis.self.onmessage({ data }));
|
parentPort.on('message', (data) => globalThis.self.onmessage({ data }));
|
||||||
@@ -73,6 +85,7 @@ parentPort.on('message', (data) => globalThis.self.onmessage({ data }));
|
|||||||
localStorage.clear();
|
localStorage.clear();
|
||||||
heartbeats = 0;
|
heartbeats = 0;
|
||||||
simulateLongCalculation = false;
|
simulateLongCalculation = false;
|
||||||
|
holdForCancelAtStage = undefined;
|
||||||
simulatedElapsed = 0;
|
simulatedElapsed = 0;
|
||||||
sentTypes.length = 0;
|
sentTypes.length = 0;
|
||||||
receivedTypes.length = 0;
|
receivedTypes.length = 0;
|
||||||
@@ -80,7 +93,9 @@ parentPort.on('message', (data) => globalThis.self.onmessage({ data }));
|
|||||||
class RealWorker {
|
class RealWorker {
|
||||||
onmessage: ((event: MessageEvent) => void) | null = null;
|
onmessage: ((event: MessageEvent) => void) | null = null;
|
||||||
onerror: (() => void) | null = null;
|
onerror: (() => void) | null = null;
|
||||||
private thread = new NodeWorker(bundle, { workerData: { simulateLongCalculation } });
|
private thread = new NodeWorker(bundle, {
|
||||||
|
workerData: { simulateLongCalculation, holdForCancelAtStage },
|
||||||
|
});
|
||||||
constructor() {
|
constructor() {
|
||||||
threads.add(this.thread);
|
threads.add(this.thread);
|
||||||
this.thread.on('message', (data) => {
|
this.thread.on('message', (data) => {
|
||||||
@@ -677,6 +692,8 @@ parentPort.on('message', (data) => globalThis.self.onmessage({ data }));
|
|||||||
});
|
});
|
||||||
|
|
||||||
it('preserves balances and provenance after a valid checkpoint', async () => {
|
it('preserves balances and provenance after a valid checkpoint', async () => {
|
||||||
|
const cancelStage = 'optimization:default-non-working-day-gain';
|
||||||
|
holdForCancelAtStage = cancelStage;
|
||||||
const calculation = createPlanningCalculation({
|
const calculation = createPlanningCalculation({
|
||||||
horizon: { start: '2024-01-01', end: '2025-01-01' },
|
horizon: { start: '2024-01-01', end: '2025-01-01' },
|
||||||
balances: [{ id: 'cp', type: 'cp', quantity: 3, expiryDate: '2024-12-31' }],
|
balances: [{ id: 'cp', type: 'cp', quantity: 3, expiryDate: '2024-12-31' }],
|
||||||
@@ -696,7 +713,7 @@ parentPort.on('message', (data) => globalThis.self.onmessage({ data }));
|
|||||||
...presets.slice(1),
|
...presets.slice(1),
|
||||||
],
|
],
|
||||||
progress: (stage) => {
|
progress: (stage) => {
|
||||||
if (stage === 'optimization:default-non-working-day-gain') calculation.cancel();
|
if (stage === cancelStage) calculation.cancel();
|
||||||
},
|
},
|
||||||
});
|
});
|
||||||
const outcome = await calculation.outcome;
|
const outcome = await calculation.outcome;
|
||||||
@@ -704,7 +721,7 @@ parentPort.on('message', (data) => globalThis.self.onmessage({ data }));
|
|||||||
if (outcome.kind !== 'PlanningIncomplete') return;
|
if (outcome.kind !== 'PlanningIncomplete') return;
|
||||||
expect(outcome.reasons[0].code).toBe('user-cancelled');
|
expect(outcome.reasons[0].code).toBe('user-cancelled');
|
||||||
expect(outcome.completedStages).toContain('optimization:default-holiday-week-value');
|
expect(outcome.completedStages).toContain('optimization:default-holiday-week-value');
|
||||||
expect(outcome.completedStages).not.toContain('optimization:default-non-working-day-gain');
|
expect(outcome.completedStages).not.toContain(cancelStage);
|
||||||
const incumbent = outcome.incumbent!;
|
const incumbent = outcome.incumbent!;
|
||||||
expect(incumbent.placements).toHaveLength(3);
|
expect(incumbent.placements).toHaveLength(3);
|
||||||
expect(incumbent.placements).toContainEqual({
|
expect(incumbent.placements).toContainEqual({
|
||||||
|
|||||||
Reference in new issue
Block a user