From 3a7a49690de4d9ebbdc07b7cb20557728d88808e Mon Sep 17 00:00:00 2001 From: "[._.]/ Adam Eivy" Date: Sun, 30 Aug 2026 22:56:50 +0000 Subject: [PATCH 1/2] fix: bound and cancel Nominatim requests (#160) --- .changelog/NEXT.md | 1 + server/src/routes/map.routes.ts | 10 +- server/src/services/geocode.service.ts | 111 ++++++++++++++------ tests/unit/routes/mapGeocodeStream.spec.ts | 47 +++++++++ tests/unit/services/geocode.service.spec.ts | 83 +++++++++++++++ 5 files changed, 219 insertions(+), 33 deletions(-) create mode 100644 tests/unit/routes/mapGeocodeStream.spec.ts create mode 100644 tests/unit/services/geocode.service.spec.ts diff --git a/.changelog/NEXT.md b/.changelog/NEXT.md index 64db6f7b..6dcc8b9f 100644 --- a/.changelog/NEXT.md +++ b/.changelog/NEXT.md @@ -50,6 +50,7 @@ ## Fixed +- Nominatim geocoding requests now time out after 15 seconds and are cancelled when their map-stream client disconnects, preventing stalled upstream sockets from blocking the shared geocoding queue. - Search results now keep their alphabetical ordering. The batch person-loader (`getPersonsBatch`) re-orders rows back to the requested order, fixing a regression where SQLite's `WHERE person_id IN (...)` returned rows in table order and silently discarded the search query's `ORDER BY display_name` (so the default, unsorted search view appeared randomly ordered). - Platform comparison now treats equivalent place spellings as matches: "Dallas, Texas, USA" vs "Dallas, Texas, United States" (and U.S.A. / United States of America / state abbreviations like TX vs Texas, UK vs United Kingdom, etc.) β€” no longer flagged as `different`. Place containment is now suffix-based, so "Texas" no longer falsely matches "Texarkana" - Platform comparison now treats equivalent date formats as matches (e.g., "1979-07-31" vs "31 JUL 1979") diff --git a/server/src/routes/map.routes.ts b/server/src/routes/map.routes.ts index 1b73b6e7..853155d6 100644 --- a/server/src/routes/map.routes.ts +++ b/server/src/routes/map.routes.ts @@ -72,11 +72,16 @@ mapRouter.get('/geocode/stream', async (req: Request, res: Response) => { logger.api('map', `πŸ—ΊοΈ Starting batch geocode of ${placesToGeocode.length} places`); + const controller = new AbortController(); let cancelled = false; - req.on('close', () => { cancelled = true; }); + const cancel = () => { + cancelled = true; + controller.abort(); + }; + req.on('close', cancel); const streamResult = await (async () => { - for await (const progress of geocodeService.batchGeocode(placesToGeocode)) { + for await (const progress of geocodeService.batchGeocode(placesToGeocode, controller.signal)) { if (cancelled) return 'cancelled'; sendEvent(progress); } @@ -93,6 +98,7 @@ mapRouter.get('/geocode/stream', async (req: Request, res: Response) => { sendEvent({ type: 'complete', current: placesToGeocode.length, total: placesToGeocode.length }); } + req.off('close', cancel); res.end(); }); diff --git a/server/src/services/geocode.service.ts b/server/src/services/geocode.service.ts index 43021193..73d96221 100644 --- a/server/src/services/geocode.service.ts +++ b/server/src/services/geocode.service.ts @@ -14,6 +14,8 @@ const APP_VERSION = process.env.npm_package_version ?? 'unknown'; const USER_AGENT = `SparseTree/${APP_VERSION} (genealogy toolkit; https://github.com/atomantic/SparseTree)`; const REQUEST_DELAY_MS = 1100; // Nominatim requires 1 req/sec max const RATE_LIMIT_PAUSE_MS = 60_000; +// Bound each upstream attempt so a stalled socket cannot block the shared queue forever. +const REQUEST_TIMEOUT_MS = 15_000; // Serialized rate limiter: chains promises so only one request runs at a time let requestChain = Promise.resolve(); @@ -36,8 +38,19 @@ function normalizePlaceText(text: string): string { return text.toLowerCase().trim(); } -function delay(ms: number): Promise { - return new Promise(resolve => setTimeout(resolve, ms)); +function delay(ms: number, signal?: AbortSignal): Promise { + if (signal?.aborted) return Promise.resolve(false); + + return new Promise(resolve => { + const timeout = setTimeout(finish, ms, true); + const onAbort = () => finish(false); + function finish(completed: boolean): void { + clearTimeout(timeout); + signal?.removeEventListener('abort', onAbort); + resolve(completed); + } + signal?.addEventListener('abort', onAbort, { once: true }); + }); } /** @@ -83,38 +96,69 @@ function ensurePending(placeText: string): void { ); } -type FetchResult = { status: 'found'; result: NominatimResult } | { status: 'not_found' } | { status: 'error' }; +type FetchResult = { status: 'found'; result: NominatimResult } | { status: 'not_found' } | { status: 'error' } | { status: 'cancelled' }; + +async function fetchWithDeadline(url: string, query: string, callerSignal?: AbortSignal): Promise { + if (callerSignal?.aborted) return 'cancelled'; + + const controller = new AbortController(); + let timedOut = false; + const onCallerAbort = () => controller.abort(callerSignal?.reason); + callerSignal?.addEventListener('abort', onCallerAbort, { once: true }); + const timeout = setTimeout(() => { + timedOut = true; + controller.abort(); + }, REQUEST_TIMEOUT_MS); + + try { + return await fetch(url, { headers: { 'User-Agent': USER_AGENT }, signal: controller.signal }); + } catch (error) { + if (callerSignal?.aborted) { + logger.warn('geocode', `Geocode request status=cancelled place=${JSON.stringify(query)} reason=caller_cancelled`); + return 'cancelled'; + } + const reason = timedOut ? 'timeout' : 'network_error'; + logger.warn('geocode', `Geocode request status=error place=${JSON.stringify(query)} reason=${reason}`); + return null; + } finally { + clearTimeout(timeout); + callerSignal?.removeEventListener('abort', onCallerAbort); + } +} /** * Single Nominatim request, serialized through a promise queue to guarantee rate limiting. * Returns a tri-state: found (with result), not_found (empty results), or error (network/server failure). */ -function fetchNominatim(query: string): Promise { +export function fetchNominatim(query: string, signal?: AbortSignal): Promise { const work = async (): Promise => { - await delay(REQUEST_DELAY_MS); - - const url = `${NOMINATIM_URL}?${new URLSearchParams({ q: query, format: 'json', limit: '1' })}`; - - const response = await fetch(url, { headers: { 'User-Agent': USER_AGENT } }).catch((err: Error) => { - logger.error('geocode', `🌐 Network error for "${query}": ${err.message}`); - return null; - }); - - if (!response) return { status: 'error' }; + try { + if (!await delay(REQUEST_DELAY_MS, signal)) return { status: 'cancelled' }; + + const url = `${NOMINATIM_URL}?${new URLSearchParams({ q: query, format: 'json', limit: '1' })}`; + const response = await fetchWithDeadline(url, query, signal); + + if (response === 'cancelled') return { status: 'cancelled' }; + if (!response) return { status: 'error' }; + + if (response.status === 429) { + logger.warn('geocode', `⏳ Rate limited by Nominatim, pausing ${RATE_LIMIT_PAUSE_MS / 1000}s`); + if (!await delay(RATE_LIMIT_PAUSE_MS, signal)) return { status: 'cancelled' }; + const retry = await fetchWithDeadline(url, query, signal); + if (retry === 'cancelled') return { status: 'cancelled' }; + if (!retry?.ok) return { status: 'error' }; + const retryData: NominatimResult[] = await retry.json(); + return retryData[0] ? { status: 'found', result: retryData[0] } : { status: 'not_found' }; + } - if (response.status === 429) { - logger.warn('geocode', `⏳ Rate limited by Nominatim, pausing ${RATE_LIMIT_PAUSE_MS / 1000}s`); - await delay(RATE_LIMIT_PAUSE_MS); - const retry = await fetch(url, { headers: { 'User-Agent': USER_AGENT } }).catch(() => null); - if (!retry?.ok) return { status: 'error' }; - const retryData: NominatimResult[] = await retry.json(); - return retryData[0] ? { status: 'found', result: retryData[0] } : { status: 'not_found' }; + if (!response.ok) return { status: 'error' }; + const data: NominatimResult[] = await response.json(); + return data[0] ? { status: 'found', result: data[0] } : { status: 'not_found' }; + } catch (error) { + if (signal?.aborted) return { status: 'cancelled' }; + logger.warn('geocode', `Geocode response status=error place=${JSON.stringify(query)} reason=response_error`); + return { status: 'error' }; } - - if (!response.ok) return { status: 'error' }; - - const data: NominatimResult[] = await response.json(); - return data[0] ? { status: 'found', result: data[0] } : { status: 'not_found' }; }; // Chain onto the request queue so only one request runs at a time @@ -129,9 +173,9 @@ function fetchNominatim(query: string): Promise { * if the full query returns no results, progressively strip the leftmost (most specific) * segment and retry with broader locations. Stops at 2 remaining segments minimum. */ -type QueryResult = { lat: number; lng: number; displayName: string; status: 'resolved' } | { status: 'not_found' } | { status: 'error' }; +type QueryResult = { lat: number; lng: number; displayName: string; status: 'resolved' } | { status: 'not_found' } | { status: 'error' } | { status: 'cancelled' }; -async function queryNominatim(placeText: string): Promise { +export async function queryNominatim(placeText: string, signal?: AbortSignal): Promise { const parts = placeText.split(',').map(s => s.trim()).filter(Boolean); let hadError = false; @@ -140,7 +184,9 @@ async function queryNominatim(placeText: string): Promise { const maxSkip = Math.max(0, parts.length - 2); for (let skip = 0; skip <= maxSkip; skip++) { const query = parts.slice(skip).join(', '); - const result = await fetchNominatim(query); + const result = await fetchNominatim(query, signal); + + if (result.status === 'cancelled' || signal?.aborted) return { status: 'cancelled' }; if (result.status === 'found') { if (skip > 0) { @@ -168,10 +214,11 @@ export interface GeocodeProgress { * Batch geocode a list of places, yielding progress events. * Skips places already resolved or marked not_found. */ -async function* batchGeocode(places: string[]): AsyncGenerator { +async function* batchGeocode(places: string[], signal?: AbortSignal): AsyncGenerator { const total = places.length; for (let i = 0; i < places.length; i++) { + if (signal?.aborted) return; const place = places[i]; const normalized = normalizePlaceText(place); @@ -186,7 +233,9 @@ async function* batchGeocode(places: string[]): AsyncGenerator ensurePending(normalized); // Query Nominatim - const result = await queryNominatim(normalized); + const result = await queryNominatim(normalized, signal); + + if (result.status === 'cancelled' || signal?.aborted) return; if (result.status === 'resolved') { upsertPlace(normalized, result.lat, result.lng, result.displayName, 'resolved'); diff --git a/tests/unit/routes/mapGeocodeStream.spec.ts b/tests/unit/routes/mapGeocodeStream.spec.ts new file mode 100644 index 00000000..46d6e7db --- /dev/null +++ b/tests/unit/routes/mapGeocodeStream.spec.ts @@ -0,0 +1,47 @@ +import { EventEmitter } from 'node:events'; +import { describe, expect, it, vi } from 'vitest'; + +const batchGeocode = vi.fn(); +const sendEvent = vi.fn(); + +vi.mock('../../../server/src/services/geocode.service.js', () => ({ + geocodeService: { batchGeocode, getGeocodeStats: vi.fn(), resetNotFound: vi.fn() }, +})); +vi.mock('../../../server/src/services/map.service.js', () => ({ + mapService: { getUngeocodedPlaces: vi.fn(() => ['First place', 'Second place']) }, +})); +vi.mock('../../../server/src/db/sqlite.service.js', () => ({ + sqliteService: { queryOne: vi.fn(() => ({ db_id: 'db-1' })) }, +})); +vi.mock('../../../server/src/lib/logger.js', () => ({ logger: { api: vi.fn() } })); +vi.mock('../../../server/src/utils/sseHelpers.js', () => ({ initSSEData: vi.fn(() => sendEvent) })); + +const { mapRouter } = await import('../../../server/src/routes/map.routes.js'); + +describe('map geocode stream cancellation', () => { + it('aborts the batch when the SSE client disconnects', async () => { + let receivedSignal: AbortSignal | undefined; + batchGeocode.mockImplementation(async function* (_places: string[], signal?: AbortSignal) { + receivedSignal = signal; + await new Promise(resolve => signal?.addEventListener('abort', () => resolve(), { once: true })); + yield { type: 'progress', current: 1, total: 2, place: 'First place', status: 'resolved' }; + }); + + const layer = (mapRouter as unknown as { stack: Array<{ route?: { path: string; stack: Array<{ handle: (req: unknown, res: unknown) => Promise }> } }> }).stack + .find(item => item.route?.path === '/geocode/stream'); + const handler = layer?.route?.stack[0]?.handle; + expect(handler).toBeDefined(); + + const req = Object.assign(new EventEmitter(), { query: { dbId: 'db-1' } }); + const res = { end: vi.fn() }; + const completion = handler!(req, res); + await vi.waitFor(() => expect(batchGeocode).toHaveBeenCalledTimes(1)); + + req.emit('close'); + await completion; + + expect(receivedSignal?.aborted).toBe(true); + expect(sendEvent).not.toHaveBeenCalled(); + expect(res.end).toHaveBeenCalledTimes(1); + }); +}); diff --git a/tests/unit/services/geocode.service.spec.ts b/tests/unit/services/geocode.service.spec.ts new file mode 100644 index 00000000..9504c732 --- /dev/null +++ b/tests/unit/services/geocode.service.spec.ts @@ -0,0 +1,83 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; + +const queryOne = vi.fn(); +const run = vi.fn(() => ({ changes: 1 })); + +vi.mock('../../../server/src/db/sqlite.service.js', () => ({ + sqliteService: { queryOne, queryAll: vi.fn(() => []), run }, +})); + +vi.mock('../../../server/src/lib/logger.js', () => ({ + logger: { + ok: vi.fn(), + warn: vi.fn(), + error: vi.fn(), + }, +})); + +const { fetchNominatim, geocodeService } = await import('../../../server/src/services/geocode.service.js'); + +const response = (status: number, body: unknown): Response => new Response(JSON.stringify(body), { status }); + +describe('geocodeService Nominatim lifecycle', () => { + beforeEach(() => { + queryOne.mockReset(); + run.mockClear(); + vi.useFakeTimers(); + }); + + afterEach(() => { + vi.useRealTimers(); + vi.unstubAllGlobals(); + }); + + it('times out a stalled request and lets the next queued request proceed', async () => { + const fetchMock = vi.fn() + .mockImplementationOnce((_url: string, init: RequestInit) => new Promise((_resolve, reject) => { + init.signal?.addEventListener('abort', () => reject(new DOMException('Timed out', 'AbortError'))); + })) + .mockResolvedValueOnce(response(200, [{ lat: '1', lon: '2', display_name: 'Recovered' }])); + vi.stubGlobal('fetch', fetchMock); + + const stalled = fetchNominatim('Stalled place'); + await vi.advanceTimersByTimeAsync(1_100 + 15_000); + await expect(stalled).resolves.toEqual({ status: 'error' }); + + const recovered = fetchNominatim('Recovered place'); + await vi.advanceTimersByTimeAsync(1_100); + await expect(recovered).resolves.toMatchObject({ status: 'found' }); + expect(fetchMock).toHaveBeenCalledTimes(2); + }); + + it('does not make another upstream request after a batch is cancelled', async () => { + const fetchMock = vi.fn((_url: string, init: RequestInit) => new Promise((_resolve, reject) => { + init.signal?.addEventListener('abort', () => reject(new DOMException('Cancelled', 'AbortError'))); + })); + vi.stubGlobal('fetch', fetchMock); + queryOne.mockReturnValue(undefined); + + const controller = new AbortController(); + const iterator = geocodeService.batchGeocode(['First place', 'Second place'], controller.signal); + const progress = iterator.next(); + await vi.advanceTimersByTimeAsync(1_100); + expect(fetchMock).toHaveBeenCalledTimes(1); + + controller.abort(); + await expect(progress).resolves.toMatchObject({ done: true }); + await vi.advanceTimersByTimeAsync(20_000); + expect(fetchMock).toHaveBeenCalledTimes(1); + expect(run).toHaveBeenCalledTimes(1); // pending record only; cancellation is never persisted as not_found/error + }); + + it('preserves the 429 pause and retry policy', async () => { + const fetchMock = vi.fn() + .mockResolvedValueOnce(response(429, [])) + .mockResolvedValueOnce(response(200, [{ lat: '3', lon: '4', display_name: 'Retried' }])); + vi.stubGlobal('fetch', fetchMock); + + const request = fetchNominatim('Rate limited place'); + await vi.advanceTimersByTimeAsync(1_100 + 60_000); + await expect(request).resolves.toMatchObject({ status: 'found' }); + expect(fetchMock).toHaveBeenCalledTimes(2); + }); +}); From 0bb36369deba8abb75179d8df43383fbd884e402 Mon Sep 17 00:00:00 2001 From: "[._.]/ Adam Eivy" Date: Sun, 30 Aug 2026 22:59:55 +0000 Subject: [PATCH 2/2] fix: cover geocode response bodies with deadline (#160) --- server/src/routes/map.routes.ts | 2 +- server/src/services/geocode.service.ts | 17 +++++++++++------ tests/unit/routes/mapGeocodeStream.spec.ts | 1 - tests/unit/services/geocode.service.spec.ts | 15 +++++++++++++++ 4 files changed, 27 insertions(+), 8 deletions(-) diff --git a/server/src/routes/map.routes.ts b/server/src/routes/map.routes.ts index 853155d6..d7e8f575 100644 --- a/server/src/routes/map.routes.ts +++ b/server/src/routes/map.routes.ts @@ -94,7 +94,7 @@ mapRouter.get('/geocode/stream', async (req: Request, res: Response) => { return 'error'; }); - if (streamResult === 'ok') { + if (streamResult === 'ok' && !cancelled) { sendEvent({ type: 'complete', current: placesToGeocode.length, total: placesToGeocode.length }); } diff --git a/server/src/services/geocode.service.ts b/server/src/services/geocode.service.ts index 73d96221..09cd3fb0 100644 --- a/server/src/services/geocode.service.ts +++ b/server/src/services/geocode.service.ts @@ -97,8 +97,9 @@ function ensurePending(placeText: string): void { } type FetchResult = { status: 'found'; result: NominatimResult } | { status: 'not_found' } | { status: 'error' } | { status: 'cancelled' }; +type NominatimResponse = { status: number; ok: boolean; data: NominatimResult[] }; -async function fetchWithDeadline(url: string, query: string, callerSignal?: AbortSignal): Promise { +async function fetchWithDeadline(url: string, query: string, callerSignal?: AbortSignal): Promise { if (callerSignal?.aborted) return 'cancelled'; const controller = new AbortController(); @@ -111,7 +112,13 @@ async function fetchWithDeadline(url: string, query: string, callerSignal?: Abor }, REQUEST_TIMEOUT_MS); try { - return await fetch(url, { headers: { 'User-Agent': USER_AGENT }, signal: controller.signal }); + const response = await fetch(url, { headers: { 'User-Agent': USER_AGENT }, signal: controller.signal }); + if (response.status === 429) { + void response.body?.cancel().catch(() => {}); + return { status: response.status, ok: response.ok, data: [] }; + } + const data = response.ok ? await response.json() as NominatimResult[] : []; + return { status: response.status, ok: response.ok, data }; } catch (error) { if (callerSignal?.aborted) { logger.warn('geocode', `Geocode request status=cancelled place=${JSON.stringify(query)} reason=caller_cancelled`); @@ -147,13 +154,11 @@ export function fetchNominatim(query: string, signal?: AbortSignal): Promise { batchGeocode.mockImplementation(async function* (_places: string[], signal?: AbortSignal) { receivedSignal = signal; await new Promise(resolve => signal?.addEventListener('abort', () => resolve(), { once: true })); - yield { type: 'progress', current: 1, total: 2, place: 'First place', status: 'resolved' }; }); const layer = (mapRouter as unknown as { stack: Array<{ route?: { path: string; stack: Array<{ handle: (req: unknown, res: unknown) => Promise }> } }> }).stack diff --git a/tests/unit/services/geocode.service.spec.ts b/tests/unit/services/geocode.service.spec.ts index 9504c732..86d1a865 100644 --- a/tests/unit/services/geocode.service.spec.ts +++ b/tests/unit/services/geocode.service.spec.ts @@ -49,6 +49,21 @@ describe('geocodeService Nominatim lifecycle', () => { expect(fetchMock).toHaveBeenCalledTimes(2); }); + it('keeps the deadline active while reading a stalled response body', async () => { + const fetchMock = vi.fn((_url: string, init: RequestInit) => ({ + status: 200, + ok: true, + json: () => new Promise((_resolve, reject) => { + init.signal?.addEventListener('abort', () => reject(new DOMException('Timed out', 'AbortError'))); + }), + })); + vi.stubGlobal('fetch', fetchMock); + + const request = fetchNominatim('Slow body place'); + await vi.advanceTimersByTimeAsync(1_100 + 15_000); + await expect(request).resolves.toEqual({ status: 'error' }); + }); + it('does not make another upstream request after a batch is cancelled', async () => { const fetchMock = vi.fn((_url: string, init: RequestInit) => new Promise((_resolve, reject) => { init.signal?.addEventListener('abort', () => reject(new DOMException('Cancelled', 'AbortError')));