Place photos: one copy at a time, and Commons' 429 pauses downloads (#112)

ddiff --git a/server/places/photo-cache.js b/server/places/photo-cache.js
iindex ebee4b8..b2b3d07 100644
---- a/server/places/photo-cache.js
++++ b/server/places/photo-cache.js
@@@ -2,22 +2,56 @@
  
  // A checked Commons image is copied once into Media as a public object. The file's Commons page,
  // author and licence stay on the place row; Media serves only the smaller image bytes.
--// Media takes at most 30 uploads a minute and 600 an hour from one caller (media.object.upload, 429 rate_limited).
--// One every UPLOAD_GAP_MS stays under both (about 550 an hour); a 429 pauses every copy for PAUSE_MS.
++//
++// Every copy (download from Commons or Openverse, then upload to Media) runs one at a time for the whole process, one
++// every UPLOAD_GAP_MS:
++// - Media takes at most 30 uploads a minute and 600 an hour from one caller (media.object.upload, 429 rate_limited);
++//   one copy every 6.5 s is about 550 an hour, under both.
++// - Commons refuses bursts too: on 11-09 the places job asked it for every place of a batch at once (371 downloads in a
++//   minute) and every one came back 429. Only the upload was paced then, not the download.
++// A 429 from Media pauses every copy; a 429 from a picture host pauses downloads from that host (Retry-After, at least
++// PAUSE_MS). Paused or refused copies return null and the place is deferred to a later run, never marked "no photo".
  const UPLOAD_GAP_MS = 6_500;
  const PAUSE_MS = 10 * 60 * 1000;
--let pausedUntil = 0;
--// One pace for the whole process: the places job and on-demand discovery each make a cache, and together they must
--// stay under Media's limit too.
--let nextUploadAt = 0;
++const MAX_PAUSE_MS = 60 * 60 * 1000;
++// One queue, one clock and one set of pauses for the process: the places job, the cities job and on-demand discovery
++// each make a cache, and together they must stay under the limits.
++const pacing = { tail: Promise.resolve(), nextCopyAt: 0, pausedUntil: new Map() };
++
++function resetPhotoPacing() {
++    pacing.tail = Promise.resolve();
++    pacing.nextCopyAt = 0;
++    pacing.pausedUntil.clear();
++}
++
++function isPaused(key) {
++    return Date.now() < (pacing.pausedUntil.get(key) || 0);
++}
++
++function pause(key, ms, log, why) {
++    if (!isPaused(key)) log.warn(`[Tours photos] ${why}: copies pause for ${Math.round(ms / 60000)} minutes`);
++    pacing.pausedUntil.set(key, Date.now() + ms);
++}
++
++function retryAfterMs(res) {
++    const raw = res?.headers?.get ? res.headers.get('retry-after') : null;
++    let ms = PAUSE_MS;
++    if (raw) {
++        const secs = Number(raw);
++        ms = Number.isFinite(secs) ? secs * 1000 : Date.parse(raw) - Date.now();
++        if (!Number.isFinite(ms)) ms = PAUSE_MS;
++    }
++    return Math.min(MAX_PAUSE_MS, Math.max(PAUSE_MS, ms));
++}
++
++function oneAtATime(fn) {
++    const run = pacing.tail.then(fn, fn);
++    pacing.tail = run.then(() => {}, () => {});
++    return run;
++}
++
  function createPhotoCache({ config, s, fetchImpl = globalThis.fetch, log = console, objectsClient = null, uploadGapMs = UPLOAD_GAP_MS }) {
      let objects = objectsClient;
--    async function uploadSlot() {
--        const now = Date.now();
--        const at = Math.max(now, nextUploadAt);
--        nextUploadAt = at + uploadGapMs;
--        if (at > now) await new Promise((ok) => setTimeout(ok, at - now));
--    }
      function client() {
          if (!objects) {
              const { createServiceTokenClient } = require('openvibe-sdk/auth');
@@@ -30,38 +64,41 @@ function createPhotoCache({ config, s, fetchImpl = globalThis.fetch, log = conso
          }
          return objects;
      }
--    async function cache(image) {
--        if (!image?.sourceUrl || !image?.url) return image;
--        // Media said "too many" a moment ago: skip (the place is deferred and tried at the next run).
--        if (Date.now() < pausedUntil) return null;
--        if (config.drill) return image.provider === 'openverse' ? null : image;
--        if (!config.oauth.clientSecret) return null;
++    async function copied(image) {
          const old = await s.db.maybe('SELECT media_url FROM tours_photo_assets WHERE source_url=$1', [image.sourceUrl]);
--        if (old) return { ...image, url: old.media_url };
++        return old ? { ...image, url: old.media_url } : null;
++    }
++    async function copy(image, src, host) {
++        // Waited in the queue: Media or this host may have said "too many" since, and another caller may have copied it.
++        if (isPaused('media') || isPaused(host)) return null;
++        const done = await copied(image);
++        if (done) return done;
++        const wait = pacing.nextCopyAt - Date.now();
++        if (wait > 0) await new Promise((ok) => setTimeout(ok, wait));
++        pacing.nextCopyAt = Date.now() + uploadGapMs;
          try {
--            const src = new URL(image.url);
--            const commons = /^(?:thumb|upload)\.wikimedia\.org$/.test(src.hostname);
--            const openverse = src.hostname === 'api.openverse.org' && /^\/v1\/images\/[0-9a-f-]{36}\/thumb\/$/i.test(src.pathname);
--            if (src.protocol !== 'https:' || (!commons && !openverse)) return null;
              // Commons renders only its standard thumbnail widths (250, 330, 500, 960, 1280 …): 640 answered 400 for every
              // file, so no photo reached Media (11-08). Ask for 960; if that is refused (a small original cannot be enlarged),
              // take the exact URL Commons' imageinfo returned.
              const tries = [];
--            if (commons && /\/\d+px-[^/]+$/.test(src.pathname)) { const t = new URL(src); t.pathname = t.pathname.replace(/\/\d+px-([^/]+)$/, '/960px-$1'); tries.push(t.toString()); }
++            if (host === 'commons' && /\/\d+px-[^/]+$/.test(src.pathname)) { const t = new URL(src); t.pathname = t.pathname.replace(/\/\d+px-([^/]+)$/, '/960px-$1'); tries.push(t.toString()); }
              tries.push(src.toString());
              let res = null;
              for (const u of tries) {
                  res = await fetchImpl(u, { headers: { 'User-Agent': config.openData.userAgent }, signal: AbortSignal.timeout(12_000) });
--                if (res.ok) break;
++                if (res.ok || res.status === 429) break;
++            }
++            if (res.status === 429) {
++                pause(host, retryAfterMs(res), log, `${src.hostname} is rate limiting picture downloads (429)`);
++                return null;
              }
              if (!res.ok) { log.warn(`[Tours photos] ${src.hostname} answered ${res.status} for ${image.sourceUrl}`); return null; }
              const type = String(res.headers.get('content-type') || '').split(';')[0].toLowerCase();
              const length = Number(res.headers.get('content-length') || 0);
--            if (!res.ok || !/^image\/(?:jpeg|png|webp)$/.test(type) || length > 4_000_000) return null;
++            if (!/^image\/(?:jpeg|png|webp)$/.test(type) || length > 4_000_000) return null;
              const bytes = Buffer.from(await res.arrayBuffer());
              if (bytes.length > 4_000_000) return null;
              const file = image.sourceUrl.split('/').pop().slice(0, 100) || 'place-photo.jpg';
--            await uploadSlot();
              const obj = await client().upload(bytes, { kind: 'file', visibility: 'public', mimeType: type, filename: file });
              const url = obj.public_url || obj.url;
              if (!url || !/^https:\/\/openvibe\.media\//.test(url)) { log.warn(`[Tours photos] Media answered no public openvibe.media URL for ${image.sourceUrl}`); return null; }
@@@ -70,15 +107,30 @@ function createPhotoCache({ config, s, fetchImpl = globalThis.fetch, log = conso
              return { ...image, url };
          } catch (err) {
              if (err && (err.status === 429 || /\b429\b|rate_limited/.test(String(err.message)))) {
--                if (Date.now() >= pausedUntil) log.warn(`[Tours photos] Media is rate limiting uploads: copies pause for ${PAUSE_MS / 60000} minutes (${err.message})`);
--                pausedUntil = Date.now() + PAUSE_MS;
++                pause('media', PAUSE_MS, log, `Media is rate limiting uploads (${err.message})`);
                  return null;
              }
--            log.warn(`[Tours photos] Media copy unavailable: ${err.message}`);
++            log.warn(`[Tours photos] copy failed for ${image.sourceUrl}: ${err.message}`);
              return null;
          }
      }
++    async function cache(image) {
++        if (!image?.sourceUrl || !image?.url) return image;
++        if (config.drill) return image.provider === 'openverse' ? null : image;
++        if (!config.oauth.clientSecret) return null;
++        let src;
++        try { src = new URL(image.url); } catch { return null; }
++        const commons = /^(?:thumb|upload)\.wikimedia\.org$/.test(src.hostname);
++        const openverse = src.hostname === 'api.openverse.org' && /^\/v1\/images\/[0-9a-f-]{36}\/thumb\/$/i.test(src.pathname);
++        if (src.protocol !== 'https:' || (!commons && !openverse)) return null;
++        // Already in Media: no download, even while copies are paused.
++        const done = await copied(image);
++        if (done) return done;
++        const host = commons ? 'commons' : 'openverse';
++        if (isPaused('media') || isPaused(host)) return null;
++        return oneAtATime(() => copy(image, src, host));
++    }
      return { cache };
  }
  
--module.exports = { createPhotoCache };
++module.exports = { createPhotoCache, resetPhotoPacing };
ddiff --git a/test/photo-cache-widths.test.js b/test/photo-cache-widths.test.js
iindex 3514349..eac73c7 100644
---- a/test/photo-cache-widths.test.js
++++ b/test/photo-cache-widths.test.js
@@@ -2,12 +2,14 @@
  /**
   * Tours' photo cache asks Commons for a thumbnail width Commons renders (960), falls back to the exact URL imageinfo
   * returned when that is refused, and says so in the log when nothing can be copied. On 11-08 every place photo failed
-- * silently: the cache asked for 640px, which Commons answers with 400.
++ * silently: the cache asked for 640px, which Commons answers with 400. On 11-09 Commons answered 429 to every download:
++ * the places job asked for a whole batch at once (371 in a minute). Copies now run one at a time, and a host's 429
++ * pauses downloads from it.
   *
   *   node test/photo-cache-widths.test.js
   */
  const assert = require('assert');
--const { createPhotoCache } = require('../server/places/photo-cache');
++const { createPhotoCache, resetPhotoPacing } = require('../server/places/photo-cache');
  
  const config = { oauth: { clientSecret: 'x' }, openData: { userAgent: 'test' }, drill: false };
  const db = { maybe: async () => null, exec: async () => {} };
@@@ -54,5 +56,36 @@ const objectsClient = { upload: async (bytes) => { uploaded.push(bytes.length);
      assert.strictEqual(await limited.cache({ ...image, sourceUrl: image.sourceUrl + "429b" }), null);
      assert.strictEqual(fetched, before, "paused: nothing is downloaded");
      assert.strictEqual(warned429.filter((m) => /rate limiting uploads/.test(m)).length, 1, "the pause is logged once");
--    console.log('photo cache: 960px first, the imageinfo URL as fallback, refusals logged');
++
++    // 6. Copies run one at a time for the process, a gap apart: a batch never downloads in parallel.
++    resetPhotoPacing();
++    let inFlight = 0, most = 0;
++    const starts = [];
++    const queued = createPhotoCache({ config, s: { db }, objectsClient, uploadGapMs: 60, log: { warn() {} },
++        fetchImpl: async () => { inFlight += 1; most = Math.max(most, inFlight); starts.push(Date.now()); await new Promise((r) => setTimeout(r, 5)); inFlight -= 1; return ok(); } });
++    await Promise.all([1, 2, 3, 4].map((n) => queued.cache({ ...image, sourceUrl: `${image.sourceUrl}q${n}` })));
++    assert.strictEqual(most, 1, 'one download at a time');
++    assert.ok(starts.every((t, i) => i === 0 || t - starts[i - 1] >= 55), 'downloads are a gap apart');
++
++    // 7. Commons says 429: no second try for that file, Commons downloads pause (logged once), Openverse still copies.
++    resetPhotoPacing();
++    const warnedHost = [];
++    let commonsAsked = 0, openverseAsked = 0;
++    const tooMany = { ok: false, status: 429, headers: { get: (h) => (h === 'retry-after' ? '5' : 'text/html') }, arrayBuffer: async () => new ArrayBuffer(0) };
++    const hosts = createPhotoCache({ config, s: { db }, objectsClient, uploadGapMs: 0, log: { warn: (m) => warnedHost.push(m) },
++        fetchImpl: async (u) => { if (/wikimedia/.test(u)) { commonsAsked += 1; return tooMany; } openverseAsked += 1; return ok(); } });
++    const batch = await Promise.all([1, 2, 3].map((n) => hosts.cache({ ...image, sourceUrl: `${image.sourceUrl}c${n}` })));
++    assert.deepStrictEqual(batch, [null, null, null], 'refused copies are deferred');
++    assert.strictEqual(commonsAsked, 1, 'after a 429 nothing more is asked of Commons (not even the fallback URL)');
++    assert.strictEqual(warnedHost.filter((m) => /rate limiting picture downloads/.test(m)).length, 1, 'the pause is logged once');
++    assert.ok(warnedHost.some((m) => /pause for 10 minutes/.test(m)), 'a short Retry-After still pauses at least 10 minutes');
++    const verse = await hosts.cache({ url: 'https://api.openverse.org/v1/images/0b8a4c62-5f3e-4c1e-9a1b-2a7f4e1d9c33/thumb/', sourceUrl: 'https://www.flickr.com/photos/x/1', provider: 'openverse' });
++    assert.ok(verse && openverseAsked === 1, 'another host is not paused');
++
++    // 8. While paused, a photo already in Media is still returned.
++    const known = createPhotoCache({ config, s: { db: { maybe: async () => ({ media_url: 'https://openvibe.media/o/med_known1' }), exec: async () => {} } },
++        objectsClient, uploadGapMs: 0, log: { warn() {} }, fetchImpl: async () => { throw new Error('no download expected'); } });
++    assert.strictEqual((await known.cache(image)).url, 'https://openvibe.media/o/med_known1');
++    resetPhotoPacing();
++    console.log('photo cache: 960px first, the imageinfo URL as fallback, refusals logged, one copy at a time, 429s pause');
  })().catch((e) => { console.error(e); process.exitCode = 1; });