diff --git a/server/src/nest/maps/trek-places-winter.service.ts b/server/src/nest/maps/trek-places-winter.service.ts index 52b47ae8c..52b0d7d9c 100644 --- a/server/src/nest/maps/trek-places-winter.service.ts +++ b/server/src/nest/maps/trek-places-winter.service.ts @@ -1,5 +1,6 @@ import { readEnv } from '../../app-config'; import { + trekPlacesHeldBack, trekPlacesWinterArea, trekPlacesWinterIndex, trekPlacesWinterMeta, @@ -24,6 +25,11 @@ export class TrekPlacesWinterService { return readEnv().maps.trekPlacesEnabled; } + /** Whether the circuit breaker holds requests back after the host failed several times. */ + heldBack(): boolean { + return trekPlacesHeldBack(); + } + meta(): Promise { return trekPlacesWinterMeta(); } diff --git a/server/src/nest/maps/trek-places.client.ts b/server/src/nest/maps/trek-places.client.ts index 88741eb64..4bf4ba26e 100644 --- a/server/src/nest/maps/trek-places.client.ts +++ b/server/src/nest/maps/trek-places.client.ts @@ -15,6 +15,7 @@ import { readEnv, getAppUrl } from '../../app-config'; import { toApiLang } from './maps.helpers'; import { normalizePlaceWebsite, + skiRouteBuildSchema, winterIndexRowSchema, winterMetaSchema, type WinterIndex, @@ -887,7 +888,13 @@ export async function trekPlacesRoutes( export type TrekPlacesWinterBody = { status: 'not_modified' } | { status: 'ok'; bytes: Buffer; etag: string | null }; const WINTER_META = { timeoutMs: 5_000, maxBytes: 64 * 1024 }; -const WINTER_INDEX = { timeoutMs: 15_000, maxBytes: 8 * 1024 * 1024 }; +/** + * The host caps the index it stores at 8 MB of gzip and sends it packed; + * fetch unpacks it before it is read, so the cap here is on the unpacked + * JSON, which is several times larger. 64 MB holds whatever the host's cap + * lets through and still bounds a body that is not the index at all. + */ +const WINTER_INDEX = { timeoutMs: 15_000, maxBytes: 64 * 1024 * 1024 }; const WINTER_AREA = { timeoutMs: 30_000, maxBytes: 32 * 1024 * 1024 }; const WINTER_AREA_ID = /^[a-z]:[0-9]+(-[0-9a-z]+)?$/; @@ -901,7 +908,7 @@ function bareTag(value: string | null): string | null { /** The head of the index; its rows are read one by one. */ const winterIndexHeadSchema = z.object({ - build: z.string().min(1).max(40), + build: skiRouteBuildSchema, format: z.number().int(), areas: z.array(z.unknown()), }); diff --git a/server/src/nest/ski-routing/ski-areas.job.ts b/server/src/nest/ski-routing/ski-areas.job.ts index 9e883e6b3..ec5708602 100644 --- a/server/src/nest/ski-routing/ski-areas.job.ts +++ b/server/src/nest/ski-routing/ski-areas.job.ts @@ -10,6 +10,9 @@ import { Injectable, type OnApplicationBootstrap } from '@nestjs/common'; * wait for it. Area files unused for 90 days go. Nothing runs while TREK * Places is switched off. */ +/** How many changed areas are downloaded at once. */ +const REFRESH_CONCURRENCY = 2; + @Injectable() export class SkiAreasJob implements OnApplicationBootstrap { constructor( @@ -34,7 +37,12 @@ export class SkiAreasJob implements OnApplicationBootstrap { } } - /** The areas kept here whose graph the new build changed, fetched again. */ + /** + * The areas kept here whose graph the new build changed, fetched again two + * at a time so a build that changed many does not open dozens of downloads + * against the host at once. Once the host's breaker opens the rest wait: + * their kept copies still plan, and a planner that needs one fetches it. + */ private async refreshKept(): Promise { const index = await this.areas.index(); if (!index) return; @@ -42,9 +50,25 @@ export class SkiAreasJob implements OnApplicationBootstrap { const changed = index.areas.filter((area) => kept.some((copy) => copy.id === area.row.id && copy.sha8 !== area.row.sha256.slice(0, 8)), ); - const outcomes = await Promise.allSettled(changed.map((area) => this.areas.ensureArea(area.row))); - const failed = outcomes.filter((outcome) => outcome.status === 'rejected').length; if (changed.length === 0) return; - logInfo(`[Ski] ${changed.length - failed} of ${changed.length} kept ski areas fetched anew`); + const pending = [...changed]; + let fetched = 0; + const fetchOne = async (): Promise => { + while (!this.areas.placesHeldBack()) { + const area = pending.shift(); + if (!area) return; + try { + if (!(await this.areas.ensureArea(area.row)).stale) fetched += 1; + } catch { + // The kept copy stays; the next planner to need the area asks again. + } + } + }; + await Promise.all(Array.from({ length: REFRESH_CONCURRENCY }, fetchOne)); + const left = pending.length; + logInfo( + `[Ski] ${fetched} of ${changed.length} kept ski areas fetched anew` + + (left > 0 ? `, ${left} left for later while TREK Places is unreachable` : ''), + ); } } diff --git a/server/src/nest/ski-routing/ski-areas.service.ts b/server/src/nest/ski-routing/ski-areas.service.ts index 9ac868286..2f7ef2f2a 100644 --- a/server/src/nest/ski-routing/ski-areas.service.ts +++ b/server/src/nest/ski-routing/ski-areas.service.ts @@ -283,6 +283,11 @@ export class SkiAreasService implements OnApplicationBootstrap { await rename(temp, file); } + /** Whether the Places host failed often enough that requests to it are held back for now. */ + placesHeldBack(): boolean { + return this.winter.heldBack(); + } + /** Whether an area's graph is here, on its way, or cannot be had right now. */ stateOf(row: WinterIndexRow): 'ready' | 'loading' | 'unavailable' { if (existsSync(this.fileOf(row))) return 'ready'; diff --git a/server/src/nest/ski-routing/worker-router-runner.ts b/server/src/nest/ski-routing/worker-router-runner.ts index d68b5ea80..ad3c19d5b 100644 --- a/server/src/nest/ski-routing/worker-router-runner.ts +++ b/server/src/nest/ski-routing/worker-router-runner.ts @@ -12,8 +12,16 @@ export const MAX_WAITING_JOBS = 8; /** A thread with nothing to do ends after half an hour; the next job starts one again. */ const IDLE_MS = 30 * 60 * 1000; -interface Waiting { +/** A job in line for the thread, not yet handed to it. */ +interface Queued { + id: number; + job: SkiJob; settle: (outcome: { result: unknown } | { error: Error }) => void; +} + +/** The job the thread is working on, the thread it went to, and its clock. */ +interface Running extends Queued { + worker: Worker; timer: NodeJS.Timeout; } @@ -26,33 +34,31 @@ export function answerTimeoutMs(job: SkiJob): number { /** * The piste router's own thread, so a search never holds up the server's - * requests. Started on the first job and ended when idle; a job not answered - * in time fails as `timeout` and the thread is replaced, a thread that dies - * fails what it held, and the next job finds a fresh one. + * requests. Started on the first job and ended when idle. The thread is + * handed one job at a time, so a job's clock starts when the thread takes it + * up, not while it waits behind a slow one. A job not answered in time fails + * as `timeout` and its thread is replaced; a thread that dies fails the job + * it held. Either way the jobs still in line go to a fresh thread. */ export class WorkerSkiRouterRunner implements SkiRouterRunner { private thread: Worker | null = null; private lastId = 0; - private readonly waiting = new Map(); + private readonly queue: Queued[] = []; + private running: Running | null = null; private idleTimer: NodeJS.Timeout | null = null; constructor(private readonly entry = join(__dirname, 'ski-router.worker.js')) {} run(job: J): Promise> { - if (this.waiting.size >= MAX_WAITING_JOBS) return Promise.reject(new Error('busy')); - const worker = this.ensureThread(); + if (this.queue.length + (this.running ? 1 : 0) >= MAX_WAITING_JOBS) return Promise.reject(new Error('busy')); const id = ++this.lastId; return new Promise>((resolve, reject) => { - const timer = setTimeout(() => { - this.waiting.delete(id); - reject(new Error('timeout')); - this.replaceThread(); - }, answerTimeoutMs(job)); - this.waiting.set(id, { - timer, + this.queue.push({ + id, + job, settle: (outcome) => ('error' in outcome ? reject(outcome.error) : resolve(outcome.result as SkiJobResult)), }); - worker.postMessage({ id, job }); + this.next(); }); } @@ -62,12 +68,31 @@ export class WorkerSkiRouterRunner implements SkiRouterRunner { async close(): Promise { if (this.idleTimer) clearTimeout(this.idleTimer); - this.failWaiting(new Error('closed')); + this.idleTimer = null; + this.failAll(new Error('closed')); const worker = this.thread; this.thread = null; await worker?.terminate(); } + /** Hands the next job in line to the thread, once the one before it is done. */ + private next(): void { + if (this.running) return; + const queued = this.queue.shift(); + if (!queued) return; + let worker: Worker; + try { + worker = this.ensureThread(); + } catch (err: unknown) { + queued.settle({ error: err instanceof Error ? err : new Error(String(err)) }); + this.next(); + return; + } + const timer = setTimeout(() => this.timedOut(queued.id), answerTimeoutMs(queued.job)); + this.running = { ...queued, worker, timer }; + worker.postMessage({ id: queued.id, job: queued.job }); + } + private ensureThread(): Worker { if (this.idleTimer) clearTimeout(this.idleTimer); this.idleTimer = setTimeout(() => void this.close(), IDLE_MS); @@ -79,7 +104,7 @@ export class WorkerSkiRouterRunner implements SkiRouterRunner { arg === '--require' || arg === '-r' || all[index - 1] === '--require' || all[index - 1] === '-r', ); const worker = new Worker(this.entry, { execArgv }); - worker.on('message', (reply: Reply) => this.receive(reply)); + worker.on('message', (reply: Reply) => this.receive(worker, reply)); worker.on('error', (err: Error) => this.lost(worker, err)); worker.on('exit', () => this.lost(worker, new Error('router thread ended'))); worker.unref(); @@ -87,33 +112,49 @@ export class WorkerSkiRouterRunner implements SkiRouterRunner { return worker; } - private receive(reply: Reply): void { - const waiting = this.waiting.get(reply.id); - if (!waiting) return; - this.waiting.delete(reply.id); - clearTimeout(waiting.timer); - if (reply.result !== undefined) waiting.settle({ result: reply.result }); - else waiting.settle({ error: new Error(reply.error ?? 'router failed') }); + private receive(worker: Worker, reply: Reply): void { + const running = this.running; + if (running?.worker !== worker || running.id !== reply.id) return; + this.finish( + running, + reply.result !== undefined ? { result: reply.result } : { error: new Error(reply.error ?? 'router failed') }, + ); } - /** A thread that ended or failed takes its jobs with it; one already replaced leaves the new thread's alone. */ + /** Only the job still running on the thread it went to times out; its thread goes with it. */ + private timedOut(id: number): void { + const running = this.running; + if (running?.id !== id) return; + this.dropThread(running.worker); + this.finish(running, { error: new Error('timeout') }); + } + + /** A thread that ended or failed takes its job with it; a thread already replaced touches no other job. */ private lost(worker: Worker, err: Error): void { - if (this.thread !== worker) return; - this.thread = null; - this.failWaiting(err); + if (this.thread === worker) this.thread = null; + const running = this.running; + if (running?.worker === worker) this.finish(running, { error: err }); } - private replaceThread(): void { - const worker = this.thread; - this.thread = null; - void worker?.terminate(); + private finish(running: Running, outcome: { result: unknown } | { error: Error }): void { + clearTimeout(running.timer); + this.running = null; + running.settle(outcome); + this.next(); } - private failWaiting(err: Error): void { - for (const [id, waiting] of this.waiting) { - clearTimeout(waiting.timer); - this.waiting.delete(id); - waiting.settle({ error: err }); + private dropThread(worker: Worker): void { + if (this.thread === worker) this.thread = null; + void worker.terminate(); + } + + private failAll(err: Error): void { + const running = this.running; + this.running = null; + if (running) { + clearTimeout(running.timer); + running.settle({ error: err }); } + for (const queued of this.queue.splice(0)) queued.settle({ error: err }); } } diff --git a/server/tests/unit/nest/ski-routing/ski-areas.job.test.ts b/server/tests/unit/nest/ski-routing/ski-areas.job.test.ts index 43728d0de..379bba11b 100644 --- a/server/tests/unit/nest/ski-routing/ski-areas.job.test.ts +++ b/server/tests/unit/nest/ski-routing/ski-areas.job.test.ts @@ -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'); + }); }); diff --git a/server/tests/unit/nest/ski-routing/ski-areas.service.test.ts b/server/tests/unit/nest/ski-routing/ski-areas.service.test.ts index 1af861a20..8363e338b 100644 --- a/server/tests/unit/nest/ski-routing/ski-areas.service.test.ts +++ b/server/tests/unit/nest/ski-routing/ski-areas.service.test.ts @@ -74,6 +74,7 @@ let dir: string; let host: ReturnType; 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 () => { diff --git a/server/tests/unit/nest/ski-routing/ski-router-runner.test.ts b/server/tests/unit/nest/ski-routing/ski-router-runner.test.ts index cd7f0afdb..a2d8f1333 100644 --- a/server/tests/unit/nest/ski-routing/ski-router-runner.test.ts +++ b/server/tests/unit/nest/ski-routing/ski-router-runner.test.ts @@ -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); diff --git a/server/tests/unit/nest/trek-places.winter.test.ts b/server/tests/unit/nest/trek-places.winter.test.ts index 157c5d872..b5309acbe 100644 --- a/server/tests/unit/nest/trek-places.winter.test.ts +++ b/server/tests/unit/nest/trek-places.winter.test.ts @@ -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 () => { diff --git a/shared/src/ski/ski-route.schema.spec.ts b/shared/src/ski/ski-route.schema.spec.ts index 6fdfffabd..c975c46d3 100644 --- a/shared/src/ski/ski-route.schema.spec.ts +++ b/shared/src/ski/ski-route.schema.spec.ts @@ -266,4 +266,12 @@ describe('the winter layer of the Places host (SKI-ROUTE)', () => { expect(winterIndexSchema.parse(index).areas[0]).toEqual(row); expect(winterIndexSchema.safeParse({ ...index, areas: [{ ...row, id: 'x' }] }).success).toBe(false); }); + + it('SKI-ROUTE-SCHEMA-011: refuses a build name a ski tour could not be saved with', () => { + const index = { build: '2026-11-02', format: 1, areas: [] }; + for (const build of ['nightly build', '2026+1', 'a:b']) { + expect(winterIndexSchema.safeParse({ ...index, build }).success).toBe(false); + } + expect(winterIndexSchema.safeParse({ ...index, build: '2026-11-02_redo.2' }).success).toBe(true); + }); }); diff --git a/shared/src/ski/ski-route.schema.ts b/shared/src/ski/ski-route.schema.ts index dc4efb6fe..e4dddad28 100644 --- a/shared/src/ski/ski-route.schema.ts +++ b/shared/src/ski/ski-route.schema.ts @@ -1,7 +1,6 @@ import { SKI_DIFFICULTIES, skiDisciplineSchema, - skiRouteBuildSchema, skiRouteOptionsSchema, skiTechniqueSchema, } from '../tours/ski.schema'; diff --git a/shared/src/ski/winter-index.schema.ts b/shared/src/ski/winter-index.schema.ts index c7d4c8155..4cc1d79a1 100644 --- a/shared/src/ski/winter-index.schema.ts +++ b/shared/src/ski/winter-index.schema.ts @@ -1,3 +1,5 @@ +import { skiRouteBuildSchema } from '../tours/ski.schema'; + import { z } from 'zod'; /** @@ -10,6 +12,13 @@ import { z } from 'zod'; const sha256Schema = z.string().regex(/^[0-9a-f]{64}$/); +/** + * A build is read to the same rule a ski tour stores it by, so a build TREK + * could not save with a tour is refused when the layer is read, not on the + * tour's next save. + */ +const winterBuildSchema = skiRouteBuildSchema; + /** The attribution block the winter layer carries; only the two lines TREK prints are required. */ export const winterAttributionSchema = z.object({ data: z.string(), @@ -18,7 +27,7 @@ export const winterAttributionSchema = z.object({ export type WinterAttribution = z.infer; export const winterMetaSchema = z.object({ - build: z.string().min(1).max(40), + build: winterBuildSchema, format: z.number().int(), areas: z.number().int().nonnegative(), indexSha: sha256Schema, @@ -58,7 +67,7 @@ export const winterIndexRowSchema = z.object({ export type WinterIndexRow = z.infer; export const winterIndexSchema = z.object({ - build: z.string().min(1).max(40), + build: winterBuildSchema, format: z.number().int(), areas: z.array(winterIndexRowSchema), });