diff --git a/docs/CURRENT-STATE.md b/docs/CURRENT-STATE.md index 4b29de4f5..64e0aac63 100644 --- a/docs/CURRENT-STATE.md +++ b/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 -stale for up to seven days when a refresh fails. One source failing still serves -the other. EMODnet rows the portal took from wrecksite.eu are skipped (that site +1,450 worldwide; CC0), each at most once a day. Each source keeps its own last +good records: a source that fails, or answers with nothing usable, keeps serving +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 diff --git a/server/providers/shipwrecks.js b/server/providers/shipwrecks.js index 44491209d..464dd67de 100644 --- a/server/providers/shipwrecks.js +++ b/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. */ - async function load() { - const [emodnet, wikidata] = await Promise.allSettled([ - fetchEmodnet(), - fetchWikidata(), - ]); - if (emodnet.status === 'rejected' && wikidata.status === 'rejected') - throw new Error('shipwrecks_unavailable'); - const records = mergeShipwreckRecords( - emodnet.status === 'fulfilled' ? emodnet.value : [], - wikidata.status === 'fulfilled' ? wikidata.value : [], + const fresh = (state, at) => + state.good !== null && at - state.good.fetchedAt < REFRESH_MS; + const usable = (state, at) => + state.good !== null && at - state.good.fetchedAt <= STALE_MAX_MS; + const due = (state, at) => !fresh(state, at) && at >= state.retryAt; + + /** Refresh the sources that are due; failures keep their last good records. */ + async function refresh() { + const at = now(); + const names = Object.keys(sources).filter((name) => due(sources[name], at)); + 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(), - sources: { - emodnet: emodnet.status === 'fulfilled', - wikidata: wikidata.status === 'fulfilled', - }, - records, + value: { fetchedAt, sources: sourceFlags, records: snapshot.records }, + stale, }; } async function acquire() { - if (snapshot && now() - snapshot.fetchedAt < REFRESH_MS) - return { value: snapshot, stale: false }; - try { - const { promise } = coalesceProxyRequest(inFlight, 'snapshot', load); - snapshot = await promise; - return { value: snapshot, stale: false }; - } catch (error) { - if (snapshot && now() - snapshot.fetchedAt <= STALE_MAX_MS) - return { value: snapshot, stale: true }; - throw error; - } + const at = now(); + if (Object.values(sources).some((state) => due(state, at))) + await coalesceProxyRequest(inFlight, 'refresh', refresh).promise; + const composed = compose(now()); + if (!composed) throw new Error('shipwrecks_unavailable'); + return composed; } async function handler(req, res) { diff --git a/src/data/shipwrecksProxy.test.mjs b/src/data/shipwrecksProxy.test.mjs index 45e6b6973..17818a133 100644 --- a/src/data/shipwrecksProxy.test.mjs +++ b/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')) { - if (wikidataFails) return new Response('', { status: 503 }); - return Response.json({ results: { bindings: [] } }); - } - if (emodnetFails) return new Response('', { status: 503 }); - if (href.includes('ows.emodnet-humanactivities.eu')) - return Response.json(EMODNET); - return new Response('', { status: 404 }); + const name = href.includes('query.wikidata.org') + ? 'wikidata' + : href.includes('ows.emodnet-humanactivities.eu') + ? 'emodnet' + : null; + if (!name) return new Response('', { status: 404 }); + if (state[name] === 'down') return new Response('', { status: 503 }); + 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[0].name, 'HMS/m A1'); - assert.equal(first.body.records[0].year, 1911); + assert.equal(first.body.records.length, 2); + const a1 = first.body.records.find((record) => record.name === 'HMS/m A1'); + 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'); +});