fix(ski-routing): keep one job's timeout to itself and bound the area refresh

The router thread now takes one job at a time from a queue kept on the
main thread, so a job's clock starts when the thread takes it up. A job
that times out ends only its own thread, and only while that thread is
still the current one; the jobs in line go to a fresh thread instead of
waiting out their own timers and killing the new thread in turn. A
thread that dies fails the job it held, and a thread that cannot start
fails each job on its own.

- the daily refresh fetches changed areas two at a time and stops once
  the TREK Places breaker opens; the rest keep their older copies
- the winter index cap is on the unpacked JSON fetch hands over, so it
  follows from the host's 8 MB gzip cap (64 MB) instead of matching it
- a build name is read to the rule a ski tour stores it by, so a build
  TREK could not save is refused when the winter layer is read
- an unused import in the piste router contract is gone
This commit is contained in:
Maurice
2026-10-20 21:06:48 +02:00
parent 2d0c0597b3
commit 1d72fbd25f
12 changed files with 245 additions and 47 deletions
@@ -3,6 +3,7 @@
* registrar only, sweeping unused area files, fetching the index when the
* host serves a new build, and fetching the kept areas whose graph changed.
*/
import { logInfo } from '../../../../src/nest/audit/audit-log.logger';
import type { CronRegistrarService } from '../../../../src/nest/scheduling/cron-registrar.service';
import { SkiAreasJob } from '../../../../src/nest/ski-routing/ski-areas.job';
import type { SkiAreasService, SkiIndex } from '../../../../src/nest/ski-routing/ski-areas.service';
@@ -37,6 +38,7 @@ function setup(refreshed: 'updated' | 'unchanged' | 'refused' = 'updated') {
{ id: 'w:2', sha8: '00000000' },
]),
ensureArea: vi.fn().mockResolvedValue({ file: 'x', stale: false }),
placesHeldBack: vi.fn(() => false),
};
const registrar = { isEnabled: vi.fn(() => true), register: vi.fn() };
const job = new SkiAreasJob(areas as unknown as SkiAreasService, registrar as unknown as CronRegistrarService);
@@ -82,4 +84,37 @@ describe('SkiAreasJob', () => {
failing.areas.ensureArea.mockRejectedValue(new Error('no'));
await expect(failing.job.tick()).resolves.toBeUndefined();
});
it('SKIJOB-004: fetches the changed areas two at a time, and leaves the rest once the breaker opens', async () => {
const { job, areas } = setup('updated');
const ids = Array.from({ length: 6 }, (_, i) => `w:${10 + i}`);
areas.index.mockResolvedValue({ areas: ids.map((id) => row(id, 'd')) } as unknown as SkiIndex);
areas.keptAreas.mockResolvedValue(ids.map((id) => ({ id, sha8: '00000000' })));
let open = 0;
let most = 0;
let calls = 0;
areas.ensureArea.mockImplementation(async () => {
const call = ++calls;
open += 1;
most = Math.max(most, open);
await new Promise((resolve) => setImmediate(resolve));
open -= 1;
if (call === 2) throw new Error('TREK Places API 503');
return { file: 'x', stale: call === 3 };
});
// The breaker opens once four areas were asked for.
areas.placesHeldBack.mockImplementation(() => calls >= 4);
vi.mocked(logInfo).mockClear();
await job.tick();
expect(most).toBe(2);
expect(areas.ensureArea).toHaveBeenCalledTimes(4);
expect(vi.mocked(logInfo)).toHaveBeenCalledWith(
'[Ski] 2 of 6 kept ski areas fetched anew, 2 left for later while TREK Places is unreachable',
);
areas.placesHeldBack.mockReturnValue(false);
areas.ensureArea.mockResolvedValue({ file: 'x', stale: false });
vi.mocked(logInfo).mockClear();
await job.tick();
expect(vi.mocked(logInfo)).toHaveBeenCalledWith('[Ski] 6 of 6 kept ski areas fetched anew');
});
});
@@ -74,6 +74,7 @@ let dir: string;
let host: ReturnType<typeof hosted>;
const winter = {
enabled: vi.fn(() => true),
heldBack: vi.fn(() => false),
meta: vi.fn(),
index: vi.fn(),
area: vi.fn(),
@@ -209,6 +210,9 @@ describe('an area graph', () => {
winter.enabled.mockReturnValue(false);
expect(areas.enabled()).toBe(false);
expect(areas.stateOf(row('w:400', [9, 46, 9.01, 46.01]))).toBe('unavailable');
expect(areas.placesHeldBack()).toBe(false);
winter.heldBack.mockReturnValueOnce(true);
expect(areas.placesHeldBack()).toBe(true);
});
it('SKIAREA-014: lists the areas kept, and sweeps half writes and files unused for 90 days', async () => {
@@ -221,7 +221,8 @@ describe('WorkerSkiRouterRunner', () => {
runner.run(job(['hang'])).catch((err: Error) => err.message),
);
await expect(runner.run(job(['hang']))).rejects.toThrow('busy');
await vi.advanceTimersByTimeAsync(answerTimeoutMs(job(['hang'])));
// One job at a time is with the thread, each on its own clock.
for (let i = 0; i < MAX_WAITING_JOBS; i++) await vi.advanceTimersByTimeAsync(answerTimeoutMs(job(['hang'])));
expect(await Promise.all(waiting)).toEqual(Array.from({ length: MAX_WAITING_JOBS }, () => 'timeout'));
await runner.close();
} finally {
@@ -229,6 +230,50 @@ describe('WorkerSkiRouterRunner', () => {
}
});
it('SKIRUN-025: a job times out on its own clock from when the thread takes it, and takes no other job down', async () => {
const entry = await script(
'slow',
`parentPort.on('message', (m) => {
if (m.forget || m.job.files[0] === 'hang') return;
parentPort.postMessage({ id: m.id, result: [] });
});`,
);
vi.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout'] });
try {
const runner = new WorkerSkiRouterRunner(entry);
const ms = answerTimeoutMs(job(['hang']));
const first = runner.run(job(['hang'])).catch((err: Error) => err.message);
const second = runner.run(job(['hang'])).catch((err: Error) => err.message);
const third = runner.run(job(['fine']));
await vi.advanceTimersByTimeAsync(ms);
expect(await first).toBe('timeout');
// The second job waited behind the first; its clock started only when the fresh thread took it.
let secondDone = false;
void second.then(() => {
secondDone = true;
});
await vi.advanceTimersByTimeAsync(ms - 1);
expect(secondDone).toBe(false);
await vi.advanceTimersByTimeAsync(1);
expect(await second).toBe('timeout');
// The third job runs on the thread after it and is answered; no earlier clock ends it or its thread.
await expect(third).resolves.toEqual([]);
await vi.advanceTimersByTimeAsync(ms);
await expect(runner.run(job(['fine']))).resolves.toEqual([]);
await runner.close();
} finally {
vi.useRealTimers();
}
});
it('SKIRUN-026: a thread that cannot start fails each job in line on its own', async () => {
// A bare relative path is one node refuses to start a thread from.
const runner = new WorkerSkiRouterRunner('no-such-router.js');
await expect(runner.run(job(['x']))).rejects.toThrow();
await expect(runner.run(job(['y']))).rejects.toThrow();
await runner.close();
});
it('SKIRUN-024: a job of many legs waits longer for its answer, never beyond a minute', () => {
const one = job(['x']);
expect(answerTimeoutMs(one)).toBe(8000);
@@ -134,6 +134,18 @@ describe('the winter index', () => {
stub(answer(200, JSON.stringify({ build: '', format: 1, areas: [] })));
await expect(trekPlacesWinterIndex(null)).rejects.toThrow();
});
it('PLACESWINTER-012: reads an unpacked index past 8 MB, and refuses a build a tour could not keep', async () => {
const filler = Array.from({ length: 32_000 }, (_, i) => ({ ...ROW, id: `w:${i + 1}` }));
const json = JSON.stringify({ build: '2026-11-02', format: 1, areas: filler });
expect(Buffer.byteLength(json)).toBeGreaterThan(8 * 1024 * 1024);
stub(answer(200, json));
const read = await trekPlacesWinterIndex(null);
if (read.status !== 'ok') throw new Error('not read');
expect(read.index.areas).toHaveLength(32_000);
stub(answer(200, JSON.stringify({ build: 'nightly build', format: 1, areas: [] })));
await expect(trekPlacesWinterIndex(null)).rejects.toThrow();
});
});
describe('an area graph', () => {
@@ -180,6 +192,9 @@ describe('the opt-out and the breaker', () => {
await expect(trekPlacesWinterMeta()).rejects.toThrow('TREK Places API 500');
await expect(trekPlacesWinterMeta()).rejects.toThrow('not retrying yet');
expect(calls).toHaveLength(5);
expect(new TrekPlacesWinterService().heldBack()).toBe(true);
resetTrekPlacesBreaker();
expect(new TrekPlacesWinterService().heldBack()).toBe(false);
});
it('PLACESWINTER-032: the provider hands each call to the client', async () => {