fix(shipwrecks): a failed or empty source refresh keeps its last good wrecks, and waits 15 minutes before the next try

A refresh with one source down (or Wikidata answering 200 with no results)
was cached as the day's snapshot, replacing the full set; and once the
snapshot was a day old, every request after a failure fetched both sources
again. Each source now keeps its own last good records, served as stale for
up to seven days, and a source that failed is not asked again for 15 minutes,
even when there is no data yet. A Wikidata answer without results, or with no
wreck in it, counts as a failure.

docs/CURRENT-STATE.md

@@ -159,9 +159,10 @@ off by default. `/api/shipwrecks` downloads EMODnet Human Activities'
 compiled from SHOM, Ireland's National Monuments Service, Historic England and
 the Oxford Roman Economy Project; CC BY 4.0) and a Wikidata SPARQL query for
 shipwreck items with a coordinate and at least one Wikipedia article (about
-1,450 worldwide; CC0), at most once a day, and serves the last good snapshot as
+1,450 worldwide; CC0), each at most once a day. Each source keeps its own last
-stale for up to seven days when a refresh fails. One source failing still serves
+good records: a source that fails, or answers with nothing usable, keeps serving
-the other. EMODnet rows the portal took from wrecksite.eu are skipped (that site
+its earlier records as stale for up to seven days and is not asked again for
+15 minutes, and never removes the other source's wrecks. EMODnet rows the portal took from wrecksite.eu are skipped (that site
 is not openly licensed). A Wikidata wreck with the same name within 3 km of an
 EMODnet wreck is merged into one record that keeps the article link and takes
 EMODnet's year, sinking text, depth and position accuracy. Only ship columns are

server/providers/shipwrecks.js

@@ -11,8 +11,10 @@ import { clientKey, makeRateLimiter } from './common/rate-limit.js';
 const MIB = 1024 * 1024;
 /** EMODnet republishes the wreck layer about once a year; Wikidata daily at most. */
 const REFRESH_MS = 24 * 60 * 60 * 1000;
-/** A failed refresh keeps serving the last snapshot this long, marked stale. */
+/** A source's last good records keep being served this long, marked stale. */
 const STALE_MAX_MS = 7 * REFRESH_MS;
+/** After a source fails, it is not asked again for this long. */
+const RETRY_MS = 15 * 60 * 1000;
 // Only the columns the card uses; ship_char/obj_desc histories can name people
 // and are not read beyond a short vessel type.
 const EMODNET_PROPERTIES = [
@@ -48,6 +50,13 @@ export function shipwrecksProxy({
   fetchImpl = (...args) => globalThis.fetch(...args),
   now = () => Date.now(),
 } = {}) {
+  // Each source keeps its last good records on its own, so one failing (or
+  // answering with nothing usable) never removes the other's wrecks or its
+  // own earlier ones. `retryAt` holds a failed source back for RETRY_MS.
+  const sources = {
+    emodnet: { load: fetchEmodnet, good: null, retryAt: 0 },
+    wikidata: { load: fetchWikidata, good: null, retryAt: 0 },
+  };
   let snapshot = null;
   const inFlight = new Map();
   const allow = makeRateLimiter({ windowMs: 60_000, max: 30, globalMax: 600 });
@@ -75,43 +84,77 @@ export function shipwrecksProxy({
   }
 
   async function fetchWikidata() {
-    return normalizeWikidataResults(await upstreamJson(WIKIDATA_URL, 16 * MIB));
+    const payload = await upstreamJson(WIKIDATA_URL, 16 * MIB);
+    if (!Array.isArray(payload?.results?.bindings))
+      throw new Error('invalid_wikidata_results');
+    const records = normalizeWikidataResults(payload);
+    // Wikidata lists over a thousand wrecks; none at all is a broken answer.
+    if (!records.length) throw new Error('invalid_wikidata_results');
+    return records;
   }
 
-  /** One source failing still serves the other; both failing is an error. */
+  const fresh = (state, at) =>
-  async function load() {
+    state.good !== null && at - state.good.fetchedAt < REFRESH_MS;
-    const [emodnet, wikidata] = await Promise.allSettled([
+  const usable = (state, at) =>
-      fetchEmodnet(),
+    state.good !== null && at - state.good.fetchedAt <= STALE_MAX_MS;
-      fetchWikidata(),
+  const due = (state, at) => !fresh(state, at) && at >= state.retryAt;
-    ]);
+
-    if (emodnet.status === 'rejected' && wikidata.status === 'rejected')
+  /** Refresh the sources that are due; failures keep their last good records. */
-      throw new Error('shipwrecks_unavailable');
+  async function refresh() {
-    const records = mergeShipwreckRecords(
+    const at = now();
-      emodnet.status === 'fulfilled' ? emodnet.value : [],
+    const names = Object.keys(sources).filter((name) => due(sources[name], at));
-      wikidata.status === 'fulfilled' ? wikidata.value : [],
+    const results = await Promise.allSettled(
+      names.map((name) => sources[name].load()),
     );
+    const done = now();
+    results.forEach((result, i) => {
+      const state = sources[names[i]];
+      if (result.status === 'fulfilled')
+        state.good = { records: result.value, fetchedAt: done };
+      else state.retryAt = done + RETRY_MS;
+    });
+  }
+
+  /** The served snapshot: every source with usable records, merged once. */
+  function compose(at) {
+    const sourceFlags = {};
+    let stale = false;
+    let fetchedAt = null;
+    for (const [name, state] of Object.entries(sources)) {
+      sourceFlags[name] = usable(state, at);
+      if (!sourceFlags[name]) continue;
+      if (!fresh(state, at)) stale = true;
+      fetchedAt = Math.min(fetchedAt ?? Infinity, state.good.fetchedAt);
+    }
+    if (!sourceFlags.emodnet && !sourceFlags.wikidata) return null;
+    const key = `${sourceFlags.emodnet && sources.emodnet.good.fetchedAt}:${
+      sourceFlags.wikidata && sources.wikidata.good.fetchedAt
+    }`;
+    if (snapshot?.key !== key) {
+      // The merge writes EMODnet facts into Wikidata records: work on copies
+      // so a source's kept records stay as fetched.
+      const copy = (name) =>
+        sourceFlags[name]
+          ? sources[name].good.records.map((record) => ({ ...record }))
+          : [];
+      snapshot = {
+        key,
+        records: mergeShipwreckRecords(copy('emodnet'), copy('wikidata')),
+      };
+    }
     return {
-      fetchedAt: now(),
+      value: { fetchedAt, sources: sourceFlags, records: snapshot.records },
-      sources: {
+      stale,
-        emodnet: emodnet.status === 'fulfilled',
-        wikidata: wikidata.status === 'fulfilled',
-      },
-      records,
     };
   }
 
   async function acquire() {
-    if (snapshot && now() - snapshot.fetchedAt < REFRESH_MS)
+    const at = now();
-      return { value: snapshot, stale: false };
+    if (Object.values(sources).some((state) => due(state, at)))
-    try {
+      await coalesceProxyRequest(inFlight, 'refresh', refresh).promise;
-      const { promise } = coalesceProxyRequest(inFlight, 'snapshot', load);
+    const composed = compose(now());
-      snapshot = await promise;
+    if (!composed) throw new Error('shipwrecks_unavailable');
-      return { value: snapshot, stale: false };
+    return composed;
-    } catch (error) {
-      if (snapshot && now() - snapshot.fetchedAt <= STALE_MAX_MS)
-        return { value: snapshot, stale: true };
-      throw error;
-    }
   }
 
   async function handler(req, res) {

src/data/shipwrecksProxy.test.mjs

@@ -14,23 +14,55 @@ const EMODNET = {
   ],
 };
 
+const WIKIDATA = {
+  results: {
+    bindings: [
+      {
+        item: { value: 'http://www.wikidata.org/entity/Q25173' },
+        itemLabel: { value: 'Titanic' },
+        coord: { value: 'Point(-49.9469 41.7325)' },
+      },
+    ],
+  },
+};
+
+/**
+ * Upstream fake. Each source answers 'ok', 'down' (HTTP 503) or 'malformed'
+ * (HTTP 200 with a body that is not that source's data); change `state` to
+ * switch between requests.
+ */
 function fakeFetch({ wikidataFails = false, emodnetFails = false } = {}) {
   const calls = [];
+  const state = {
+    emodnet: emodnetFails ? 'down' : 'ok',
+    wikidata: wikidataFails ? 'down' : 'ok',
+  };
   const fetchImpl = async (url, options) => {
     calls.push({ url: String(url), options });
     const href = String(url);
-    if (href.includes('query.wikidata.org')) {
+    const name = href.includes('query.wikidata.org')
-      if (wikidataFails) return new Response('', { status: 503 });
+      ? 'wikidata'
-      return Response.json({ results: { bindings: [] } });
+      : href.includes('ows.emodnet-humanactivities.eu')
-    }
+        ? 'emodnet'
-    if (emodnetFails) return new Response('', { status: 503 });
+        : null;
-    if (href.includes('ows.emodnet-humanactivities.eu'))
+    if (!name) return new Response('', { status: 404 });
-      return Response.json(EMODNET);
+    if (state[name] === 'down') return new Response('', { status: 503 });
-    return new Response('', { status: 404 });
+    if (state[name] === 'malformed') return Response.json({});
+    return Response.json(name === 'wikidata' ? WIKIDATA : EMODNET);
   };
-  return { fetchImpl, calls };
+  const callsTo = (name) =>
+    calls.filter(({ url }) =>
+      url.includes(
+        name === 'wikidata'
+          ? 'query.wikidata.org'
+          : 'ows.emodnet-humanactivities',
+      ),
+    ).length;
+  return { fetchImpl, calls, state, callsTo };
 }
 
+const HOUR = 60 * 60 * 1000;
+
 function handlerOf(plugin) {
   let handler;
   plugin.configureServer({
@@ -67,9 +99,9 @@ test('serves one cached snapshot from fixed upstreams', async () => {
   const handler = handlerOf(shipwrecksProxy({ fetchImpl }));
   const first = await request(handler);
   assert.equal(first.status, 200);
-  assert.equal(first.body.records.length, 1);
+  assert.equal(first.body.records.length, 2);
-  assert.equal(first.body.records[0].name, 'HMS/m A1');
+  const a1 = first.body.records.find((record) => record.name === 'HMS/m A1');
-  assert.equal(first.body.records[0].year, 1911);
+  assert.equal(a1.year, 1911);
   assert.deepEqual(first.body.sources, { emodnet: true, wikidata: true });
   for (const { url } of calls)
     assert.match(
@@ -130,3 +162,101 @@ test('rejects other methods and routes', async () => {
   assert.equal((await request(handler, { method: 'POST' })).status, 405);
   assert.equal((await request(handler, { url: '/x' })).status, 404);
 });
+
+test("a refresh that gets nothing usable keeps every source's last good wrecks", async () => {
+  let clock = 0;
+  const upstream = fakeFetch();
+  const handler = handlerOf(
+    shipwrecksProxy({ fetchImpl: upstream.fetchImpl, now: () => clock }),
+  );
+  const first = await request(handler);
+  assert.equal(first.body.records.length, 2);
+  // A day later EMODnet is down and Wikidata answers 200 with no results.
+  upstream.state.emodnet = 'down';
+  upstream.state.wikidata = 'malformed';
+  clock = 25 * HOUR;
+  const later = await request(handler);
+  assert.equal(later.status, 200);
+  assert.equal(later.body.records.length, 2, 'no wreck is lost');
+  assert.deepEqual(later.body.sources, { emodnet: true, wikidata: true });
+  assert.equal(later.body.stale, true);
+  assert.equal(later.body.fetchedAt, 0, 'the age of the oldest data served');
+});
+
+test('one source failing keeps its own last records while the other refreshes', async () => {
+  let clock = 0;
+  const upstream = fakeFetch();
+  const handler = handlerOf(
+    shipwrecksProxy({ fetchImpl: upstream.fetchImpl, now: () => clock }),
+  );
+  await request(handler);
+  upstream.state.wikidata = 'down';
+  clock = 25 * HOUR;
+  const later = await request(handler);
+  assert.equal(later.body.records.length, 2);
+  assert.deepEqual(later.body.sources, { emodnet: true, wikidata: true });
+  assert.equal(later.body.stale, true);
+  assert.equal(upstream.callsTo('emodnet'), 2, 'EMODnet was refreshed');
+});
+
+test('a failed source waits before it is asked again', async () => {
+  let clock = 0;
+  const upstream = fakeFetch();
+  const handler = handlerOf(
+    shipwrecksProxy({ fetchImpl: upstream.fetchImpl, now: () => clock }),
+  );
+  await request(handler);
+  upstream.state.emodnet = 'down';
+  upstream.state.wikidata = 'down';
+  clock = 25 * HOUR;
+  await request(handler);
+  const asked = upstream.calls.length;
+  for (let i = 0; i < 10; i += 1) {
+    clock += 60 * 1000;
+    const res = await request(handler);
+    assert.equal(res.status, 200);
+    assert.equal(res.body.stale, true);
+  }
+  assert.equal(upstream.calls.length, asked, 'no upstream call while waiting');
+  clock = 25 * HOUR + 16 * 60 * 1000;
+  upstream.state.emodnet = 'ok';
+  upstream.state.wikidata = 'ok';
+  const back = await request(handler);
+  assert.equal(
+    upstream.calls.length,
+    asked + 2,
+    'both are retried after the wait',
+  );
+  assert.equal(back.body.stale, undefined);
+});
+
+test('with no data yet, failures are retried only after the wait', async () => {
+  let clock = 0;
+  const upstream = fakeFetch({ emodnetFails: true, wikidataFails: true });
+  const handler = handlerOf(
+    shipwrecksProxy({ fetchImpl: upstream.fetchImpl, now: () => clock }),
+  );
+  assert.equal((await request(handler)).status, 502);
+  const asked = upstream.calls.length;
+  clock += 5 * 60 * 1000;
+  assert.equal((await request(handler)).status, 502);
+  assert.equal(upstream.calls.length, asked);
+  clock += 11 * 60 * 1000;
+  assert.equal((await request(handler)).status, 502);
+  assert.equal(upstream.calls.length, asked * 2);
+});
+
+test('a source older than seven days is no longer served', async () => {
+  let clock = 0;
+  const upstream = fakeFetch();
+  const handler = handlerOf(
+    shipwrecksProxy({ fetchImpl: upstream.fetchImpl, now: () => clock }),
+  );
+  await request(handler);
+  upstream.state.emodnet = 'down';
+  upstream.state.wikidata = 'down';
+  clock = 7 * 24 * HOUR;
+  assert.equal((await request(handler)).status, 200, 'seven days is served');
+  clock = 7 * 24 * HOUR + 16 * 60 * 1000;
+  assert.equal((await request(handler)).status, 502, 'older is dropped');
+});