Drain an upload even while the crawler is catching up

apps/poller/src/index.js

@@ -283,6 +283,58 @@ async function tick() {
       }
     }
 
+    // An upload hands over its list and leaves; this is where the list becomes
+    // feeds. One slice a tick rather than a whole submission, so a very large
+    // catalogue cannot hold the crawl above hostage while it queues — the two
+    // share the process and the import is the one that can wait.
+    //
+    // Above the catch-up return, and deliberately the only thing that is.
+    // Everything below is work nobody asked for by name: discovery finds feeds
+    // on its own schedule, cards and clusters decorate what is already indexed,
+    // and none of it has somebody watching a page. An import is the opposite —
+    // a person handed us a list, was given a URL to follow, and that page says
+    // "working". In catch-up mode it said "working" for ever: a 109,474-entry
+    // upload on 2026-08-19 staged all its entries in 55 seconds and then sat at
+    // 0 queued, because this block was eleven lines below a `return`. Nothing
+    // errored, nothing logged, and the only way to find out was to read the
+    // poller source.
+    //
+    // Catch-up mode exists to stop unrelated writes competing with a deep
+    // first-crawl backlog, and that reasoning holds for everything else here.
+    // It does not hold for the one queue a user is actively waiting on, and a
+    // slice a tick is a small enough price that it never had to.
+    try {
+      const drained = await drainImport(db);
+      if (drained.ran) {
+        log('import-drain', {
+          submission: drained.submissionId,
+          queued: drained.queued,
+          skipped: drained.skipped,
+          remaining: drained.remaining,
+          finished: drained.finished,
+        });
+      }
+    } catch (err) {
+      // One bad slice must not take the crawl down with it; the entries are
+      // still staged, so the next tick tries again.
+      log('import-drain-error', { message: String(err?.message ?? err) });
+    }
+
+    // Above the return for the same reason, and because the two are a pair: the
+    // daemon that drains the queue is the one that tells the submitter it
+    // drained. Draining in catch-up mode while leaving this below would finish
+    // somebody's upload and never say so, which is a stranger failure than not
+    // draining at all. A no-op when no mail provider is configured.
+    //
+    // Guarded, unlike where it used to sit. Down there a throw only cost the
+    // housekeeping that followed it; up here it would cost the crawl.
+    try {
+      const notified = await notifyFinishedSubmissions(db);
+      if (notified.sent || notified.failed) log('notified', notified);
+    } catch (err) {
+      log('notify-error', { message: String(err?.message ?? err) });
+    }
+
     // All work below is resumable enrichment or housekeeping. In recovery mode
     // the deep first-crawl queue is the job, and returning here lets the next
     // minute tick begin immediately instead of waiting behind unrelated writes.
@@ -333,32 +385,6 @@ async function tick() {
       }
     }
 
-    // An upload hands over its list and leaves; this is where the list becomes
-    // feeds. One slice a tick rather than a whole submission, so a very large
-    // catalogue cannot hold the crawl above hostage while it queues — the two
-    // share the process and the import is the one that can wait.
-    try {
-      const drained = await drainImport(db);
-      if (drained.ran) {
-        log('import-drain', {
-          submission: drained.submissionId,
-          queued: drained.queued,
-          skipped: drained.skipped,
-          remaining: drained.remaining,
-          finished: drained.finished,
-        });
-      }
-    } catch (err) {
-      // One bad slice must not take the crawl down with it; the entries are
-      // still staged, so the next tick tries again.
-      log('import-drain-error', { message: String(err?.message ?? err) });
-    }
-
-    // Queued submissions finish long after the upload, so the daemon that
-    // drains the queue is also what tells the submitter it drained. A no-op
-    // when no mail provider is configured.
-    const notified = await notifyFinishedSubmissions(db);
-    if (notified.sent || notified.failed) log('notified', notified);
 
     const toldSearchers = await notifyFinishedDiscoveries(db);
     if (toldSearchers.sent || toldSearchers.failed) log('notified-discovery', toldSearchers);

apps/poller/test/catchup-order.test.js

@@ -0,0 +1,78 @@
+import assert from 'node:assert/strict';
+import { test } from 'node:test';
+import { readFile } from 'node:fs/promises';
+import { dirname, join } from 'node:path';
+import { fileURLToPath } from 'node:url';
+
+/**
+ * That catch-up mode cannot strand somebody's upload again.
+ *
+ * On 2026-08-19 a 109,474-entry list was submitted. Every entry staged in 55
+ * seconds and then nothing happened at all: `queued_count` stayed 0, the
+ * submitter's page said "working", and it would have said so for ever. The
+ * cause was position — `drainImport` sat eleven lines *below*
+ * `if (catchupOnly) return`, and `CRAWL_CATCHUP=1` is set on the poller.
+ *
+ * Nothing errored and nothing logged, because from the tick's point of view
+ * nothing went wrong; it simply never reached the queue a person was waiting
+ * on. The only way to find it was to read this file.
+ *
+ * This is a source-order assertion rather than a real test of `tick()`, and
+ * that is a compromise worth naming: `apps/poller/src/index.js` connects to a
+ * database and starts timers at import, so it cannot be loaded by a test
+ * without a refactor far larger than the bug. Checking the order of three
+ * lines in the source is crude, and it does hold the one invariant whose
+ * violation is invisible from the outside.
+ */
+
+const POLLER = join(dirname(fileURLToPath(import.meta.url)), '..', 'src', 'index.js');
+const source = await readFile(POLLER, 'utf8');
+
+/**
+ * Where a snippet appears, asserted to be present exactly once so a rename
+ * fails loudly here rather than silently passing on the wrong line.
+ *
+ * @param {string} needle
+ * @returns {number}
+ */
+function positionOf(needle) {
+  const first = source.indexOf(needle);
+  assert.notEqual(first, -1, `expected to find ${JSON.stringify(needle)} in the poller`);
+  assert.equal(
+    source.indexOf(needle, first + 1),
+    -1,
+    `${JSON.stringify(needle)} appears more than once; this check needs a better anchor`,
+  );
+  return first;
+}
+
+test('the catch-up early return still exists to be checked against', () => {
+  // If this ever goes away the rest of the file is asserting nothing, and a
+  // silently vacuous test is worse than no test.
+  positionOf('if (catchupOnly) return;');
+});
+
+test('an upload is drained even while the crawler is catching up', () => {
+  assert.ok(
+    positionOf('await drainImport(db)') < positionOf('if (catchupOnly) return;'),
+    'drainImport must run before the catch-up return, or a submission sits at 0% for ever',
+  );
+});
+
+test('and the submitter is told, since draining without telling is stranger still', () => {
+  assert.ok(
+    positionOf('await notifyFinishedSubmissions(db)') < positionOf('if (catchupOnly) return;'),
+    'notifyFinishedSubmissions must run before the catch-up return: the daemon that drains ' +
+      'the queue is the one that says it drained',
+  );
+});
+
+test('discovery stays below the return, because nobody is waiting on it', () => {
+  // The other half of the rule. Everything catch-up mode skips is work the
+  // crawler gave itself; an import is work a person handed over and was given a
+  // URL to watch. If this ever moves up, the mode has stopped meaning anything.
+  assert.ok(
+    positionOf('await notifyFinishedDiscoveries(db)') > positionOf('if (catchupOnly) return;'),
+    'discovery notifications are housekeeping and belong after the catch-up return',
+  );
+});