fix: bound and cancel Nominatim requests

.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.
 - AI discovery now rejects unsafe batch settings, limits background runs to one per family database, and lets an active run be cancelled without leaving provider work behind.
 - **[issue-158] Unknown API routes now return JSON errors** — Requests to unrecognized `/api` paths receive a stable 404 error envelope instead of the browser app's HTML, while client-side navigation continues to use the SPA fallback.
 - 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).

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);
     }
@@ -89,10 +94,11 @@ 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 });
   }
 
+  req.off('close', cancel);
   res.end();
 });
 

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<void> {
-  return new Promise(resolve => setTimeout(resolve, ms));
+function delay(ms: number, signal?: AbortSignal): Promise<boolean> {
+  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,74 @@ 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' };
+type NominatimResponse = { status: number; ok: boolean; data: NominatimResult[] };
+
+async function fetchWithDeadline(url: string, query: string, callerSignal?: AbortSignal): Promise<NominatimResponse | null | 'cancelled'> {
+  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 {
+    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`);
+      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<FetchResult> {
+export function fetchNominatim(query: string, signal?: AbortSignal): Promise<FetchResult> {
   const work = async (): Promise<FetchResult> => {
-    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' };
+        return retry.data[0] ? { status: 'found', result: retry.data[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' };
+      return response.data[0] ? { status: 'found', result: response.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 +178,9 @@ function fetchNominatim(query: string): Promise<FetchResult> {
  * 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<QueryResult> {
+export async function queryNominatim(placeText: string, signal?: AbortSignal): Promise<QueryResult> {
   const parts = placeText.split(',').map(s => s.trim()).filter(Boolean);
   let hadError = false;
 
@@ -140,7 +189,9 @@ async function queryNominatim(placeText: string): Promise<QueryResult> {
   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 +219,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<GeocodeProgress> {
+async function* batchGeocode(places: string[], signal?: AbortSignal): AsyncGenerator<GeocodeProgress> {
   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 +238,9 @@ async function* batchGeocode(places: string[]): AsyncGenerator<GeocodeProgress>
     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');

tests/unit/routes/mapGeocodeStream.spec.ts

@@ -0,0 +1,46 @@
+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<void>(resolve => signal?.addEventListener('abort', () => resolve(), { once: true }));
+    });
+
+    const layer = (mapRouter as unknown as { stack: Array<{ route?: { path: string; stack: Array<{ handle: (req: unknown, res: unknown) => Promise<void> }> } }> }).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);
+  });
+});

tests/unit/services/geocode.service.spec.ts

@@ -0,0 +1,98 @@
+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('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')));
+    }));
+    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);
+  });
+});