Preserve per-source pacing across queue jobs and add source pause

scraper/jobs/mangadex-archive.js

@@ -69,7 +69,7 @@ async function runMangaDexArchive(options) {
   const chapterMinMs = Number(options.chapterMinMs);
   const chapterMaxMs = Number(options.chapterMaxMs);
   const http = new SlowHttp({
-    minMs, maxMs, maxRetries: 2,
+    gateKey: 'mangadex', minMs, maxMs, maxRetries: 2,
     userAgent: 'MangaCircus-Scraper/1.0 (personal archival; sequential MangaDex requests)',
     onEvent: item => event(item.level, item.message, { status: item.status }),
   });

scraper/lib/http.js

@@ -3,10 +3,24 @@
 const { PoliteGate, sleep } = require('./delay');
 
 class StopScrapeError extends Error {}
+const sourceGates = new Map();
+
+function gateFor(options) {
+  const next = new PoliteGate(options);
+  if (!options.gateKey) return next;
+  const key = String(options.gateKey);
+  const existing = sourceGates.get(key);
+  if (!existing) { sourceGates.set(key, next); return next; }
+  // A new job may use updated host settings, but it must not forget when the
+  // previous job last requested this same source.
+  existing.minMs = next.minMs;
+  existing.maxMs = next.maxMs;
+  return existing;
+}
 
 class SlowHttp {
   constructor(options = {}) {
-    this.gate = new PoliteGate(options);
+    this.gate = gateFor(options);
     this.userAgent = options.userAgent || 'MangaCircus-Scraper/1.0 (personal archival; sequential requests)';
     this.maxRetries = options.maxRetries ?? 2;
     this.onEvent = options.onEvent;

scraper/plugins/broker.js

@@ -89,11 +89,11 @@ async function pinnedFetch(value, options = {}) {
 }
 
 function createWeebHttp(options = {}) {
-  return new SlowHttp({ ...options, transport: pinnedFetch, validateUrl: validateWeebRequest });
+  return new SlowHttp({ ...options, gateKey: 'weebcentral', transport: pinnedFetch, validateUrl: validateWeebRequest });
 }
 
 function createHostHttp(sourceHost, options = {}) {
-  return new SlowHttp({ ...options, transport: pinnedFetch,
+  return new SlowHttp({ ...options, gateKey: sourceHost, transport: pinnedFetch,
     validateUrl: (value, requestOptions) => validateHostRequest(value, sourceHost, requestOptions) });
 }
 

scraper/service.js

@@ -634,6 +634,23 @@ class ScraperService {
   }
 
   async control(action, id) {
+    if (action === 'pause-source' || action === 'resume-source') {
+      const source = String(id || '');
+      if (!this.adapterList.some(adapter => adapter.id === source)) throw new Error('Unknown source');
+      if (action === 'resume-source' && this.state.queue.some(item => item.source === source
+        && item.status === 'error' && item.failure?.kind === 'access')) {
+        throw new Error('Review the source access block before resuming');
+      }
+      if (action === 'pause-source') {
+        this.state.sourcePaused[source] = true;
+        const activeId = this.state.currentIds[source];
+        const activeItem = this.state.queue.find(item => item.id === activeId);
+        if (activeItem) activeItem.stopReason = 'paused';
+      } else delete this.state.sourcePaused[source];
+      this.recomputeStatus();
+      await this.event('info', `${source} source ${action === 'pause-source' ? 'paused' : 'resumed'} by operator`, { source });
+      return;
+    }
     if (action === 'inspect' || action === 'confirm') {
       const item = this.state.queue.find(entry => entry.id === id);
       if (!item || Object.values(this.state.currentIds || {}).includes(item.id)) throw new Error('Job not found or already active');

scraper/sources/mangadex.js

@@ -8,7 +8,7 @@ const { controlledHttp } = require('./common');
 module.exports = {
   id: 'mangadex', label: 'MangaDex', match: mangaDexInfo,
   async discover(item, options) {
-    const http = new SlowHttp({ minMs: options.settings.minMs, maxMs: options.settings.maxMs,
+    const http = new SlowHttp({ gateKey: 'mangadex', minMs: options.settings.minMs, maxMs: options.settings.maxMs,
       maxRetries: 2, userAgent: 'MangaCircus-Scraper/1.0 (personal archival; sequential MangaDex requests)',
       onEvent: options.onEvent });
     return discoverMangaDexCatalog(item.url, controlledHttp(http, options.shouldStop));

scraper/test/source-broker.test.js

@@ -54,3 +54,15 @@ test('a guarded 403 stops immediately without redirect or retry', async () => {
   await assert.rejects(http.get('https://weebcentral.com/series/X', { accept: 'text/html' }), StopScrapeError);
   assert.equal(calls, 1);
 });
+
+test('separate jobs on one source retain the configured request gap', async () => {
+  const gateKey = `fixture-${Date.now()}-${Math.random()}`;
+  const transport = async () => ({ status: 200, ok: true });
+  const first = new SlowHttp({ gateKey, minMs: 40, maxMs: 40, transport });
+  await first.get('https://example.org/one');
+  const second = new SlowHttp({ gateKey, minMs: 40, maxMs: 40, transport });
+  assert.equal(second.gate, first.gate);
+  const started = Date.now();
+  await second.get('https://example.org/two');
+  assert.ok(Date.now() - started >= 30, 'a new job must not reset its source delay');
+});

scraper/test/source-pause.test.js

@@ -0,0 +1,26 @@
+'use strict';
+
+const test = require('node:test');
+const assert = require('node:assert/strict');
+const fs = require('node:fs/promises');
+const os = require('node:os');
+const path = require('node:path');
+const { ScraperService } = require('../service');
+
+test('an operator can pause one source while other lanes continue', async t => {
+  const root = await fs.mkdtemp(path.join(os.tmpdir(), 'mc-source-pause-'));
+  const service = await new ScraperService({ libraryRoot: path.join(root, 'library'),
+    dataRoot: path.join(root, 'data') }).init();
+  t.after(async () => { await service.close(); await fs.rm(root, { recursive: true, force: true }); });
+  await assert.rejects(service.control('pause-source', 'unknown'), /Unknown source/);
+  await service.control('pause-source', 'weebcentral');
+  assert.equal(service.state.sourcePaused.weebcentral, true);
+  await service.control('resume');
+  assert.equal(service.state.paused, false);
+  assert.equal(service.state.sourcePaused.weebcentral, true);
+  service.state.queue.push({ source: 'weebcentral', status: 'error', failure: { kind: 'access' } });
+  await assert.rejects(service.control('resume-source', 'weebcentral'), /access block/);
+  service.state.queue.pop();
+  await service.control('resume-source', 'weebcentral');
+  assert.equal(service.state.sourcePaused.weebcentral, undefined);
+});