Add to the queue, and let a person's submission cut in line

apps/web/src/app/api/submit/route.js

@@ -1,4 +1,4 @@
-import { submitCatalogue, hashIp } from '@rssamplifier/ingest';
+import { submitCatalogue, hashIp, EXPRESS_MAX } from '@rssamplifier/ingest';
 import { parseOpml } from '@rssamplifier/feed';
 import { q, newId } from '@rssamplifier/db';
 
@@ -41,12 +41,12 @@ const SNIFF_CHARS = 4096;
 /**
  * Entries above which a submission is handed over rather than imported here.
  *
- * Importing means `importFeeds`: one read of every feed URL and slug in the
- * directory, then a round trip per five hundred rows. That is seconds for a
- * paste and minutes for a subscription export — a hundred and ten thousand
- * entries is over two hundred sequential round trips inside a single request
- * with a five-minute ceiling on it, which is a coin toss at best and loses the
- * whole upload when it comes up wrong.
+ * Below this a submission is queued entry by entry: a slug is claimed for each
+ * one and the rows are written in chunks. That is milliseconds for a paste and
+ * minutes for a subscription export — a hundred and ten thousand entries is
+ * hundreds of sequential round trips inside a single request with a five-minute
+ * ceiling on it, which is a coin toss at best and loses the whole upload when
+ * it comes up wrong.
  *
  * Past this the entries are staged instead, exactly as the batched uploader
  * stages them: one bulk insert per couple of thousand, no lookups, no slugs,
@@ -70,6 +70,25 @@ const STAGE_ABOVE = 5_000;
  */
 const STAGE_CHUNK = 2_000;
 
+/**
+ * How long a single-URL submission is resolved for before the submitter is sent
+ * to the status page instead.
+ *
+ * One URL is still resolved while its submitter waits, because landing on the
+ * blog you just added is the nicest thing this page does. What it must not do
+ * is wait without a bound: a site that publishes no feed costs up to eleven
+ * sequential candidate fetches at a fifteen-second timeout, and the submitter
+ * has no way to tell that from a page that has simply hung.
+ *
+ * Past this the request answers with the status page and the resolve carries on
+ * in the background — the same promise, so the feed is still inserted exactly
+ * once and there is no queued duplicate racing it.
+ */
+const INLINE_WAIT_MS = Number(process.env['SUBMIT_INLINE_WAIT_MS'] ?? 8_000) || 8_000;
+
+/** Returned by the race below when the inline resolve outlived its budget. */
+const TOO_SLOW = Symbol('too-slow');
+
 /**
  * Split a paste into candidate URLs.
  *
@@ -268,7 +287,14 @@ export async function POST(req) {
     resolveQueued = resolve;
   });
 
-  const opts = { submissionId, onQueued: (n) => resolveQueued(n) };
+  const opts = {
+    submissionId,
+    // Small enough to have been typed rather than exported, so it goes in the
+    // express lane and is crawled within a tick or two instead of behind the
+    // backlog. See EXPRESS_MAX.
+    priority: catalogue.length <= EXPRESS_MAX ? 1 : 0,
+    onQueued: (n) => resolveQueued(n),
+  };
   const work = submitCatalogue(client, catalogue, opts).then(async (result) => {
     await q.completeSubmission(client, submissionId, {
       accepted_count: result.accepted.length,
@@ -285,23 +311,42 @@ export async function POST(req) {
   const statusUrl = `${siteUrl()}/submissions/${submissionId}`;
 
   if (browser) {
-    // An upload with a queue behind it is answered the moment that queue is
-    // durable: the status page streams the rest, so waiting for a hundred
-    // sequential fetches would buy the submitter nothing but a blank tab.
+    // Anything with a queue behind it is answered the moment that queue is
+    // durable, which since submitCatalogue stopped resolving lists inline is
+    // every submission of more than one URL. The status page streams the rest.
     const settled = work.catch(() => null);
     const queued = await Promise.race([queuedCount, settled.then(() => 0)]);
 
     if (queued > 0) {
       return new Response(null, { status: 303, headers: { location: `/submissions/${submissionId}` } });
     }
 
-    // A handful of URLs resolves in seconds and has somewhere better to land:
-    // the blog itself. Nothing is gained by bouncing that through a status page.
-    const result = await settled;
-    const first = result?.accepted?.[0];
-    const location = first ? `/${first.slug}` : '/submit?error=1';
+    // One URL, and it has somewhere better to land than a status page: the blog
+    // itself. Bounded, because the resolve behind it is not — see INLINE_WAIT_MS.
+    let timer;
+    const deadline = new Promise((resolve) => {
+      timer = setTimeout(() => resolve(TOO_SLOW), INLINE_WAIT_MS);
+    });
+
+    try {
+      const result = await Promise.race([settled, deadline]);
+
+      // Still resolving. It carries on in the background and completes the
+      // submission when it lands, so the status page is the honest answer now.
+      if (result === TOO_SLOW) {
+        return new Response(null, {
+          status: 303,
+          headers: { location: `/submissions/${submissionId}` },
+        });
+      }
+
+      const first = result?.accepted?.[0];
+      const location = first ? `/${first.slug}` : '/submit?error=1';
 
-    return new Response(null, { status: 303, headers: { location } });
+      return new Response(null, { status: 303, headers: { location } });
+    } finally {
+      clearTimeout(timer);
+    }
   }
 
   const result = await work;

apps/web/src/lib/mcp/tools.js

@@ -1,6 +1,6 @@
 import { q, newId, authors as people } from '@rssamplifier/db';
 import { topicSlug } from '@rssamplifier/feed';
-import { submitCatalogue, hashIp } from '@rssamplifier/ingest';
+import { submitCatalogue, hashIp, EXPRESS_MAX } from '@rssamplifier/ingest';
 
 import { db, siteUrl } from '../db.js';
 import { readerView } from '../reader.js';
@@ -24,8 +24,17 @@ import { clip, plainText } from './text.js';
 /** How much extracted article text one `read_post` call may return. */
 const ARTICLE_LIMIT = 24_000;
 
-/** Feeds one `submit_feed` call will resolve before queueing the rest. */
-const SUBMIT_INLINE_LIMIT = 20;
+/**
+ * Feeds one `submit_feed` call will resolve before queueing the rest.
+ *
+ * One, matching the form, and for the same reason: a resolve is an outbound
+ * fetch — up to eleven of them sequentially at a fifteen-second timeout — so
+ * twenty of them was a tool call that could run for minutes and time out with
+ * nothing to show. A list is queued instead and the caller is handed the status
+ * URL it already returns. A single URL still resolves, so an agent submitting
+ * one blog gets its slug back in the same call.
+ */
+const SUBMIT_INLINE_LIMIT = 1;
 
 /** Submissions allowed per IP per hour, matching /api/submit. */
 const SUBMIT_RATE_LIMIT = 20;
@@ -561,7 +570,15 @@ export const TOOLS = [
       const result = await submitCatalogue(
         client,
         urls.map((url) => ({ url })),
-        { submissionId, inlineLimit: SUBMIT_INLINE_LIMIT },
+        {
+          submissionId,
+          inlineLimit: urls.length === 1 ? SUBMIT_INLINE_LIMIT : 0,
+          // A hand-written tool call is the same population as a hand-typed
+          // paste, so it gets the same express lane and the same bound. The
+          // tool accepts up to 200 URLs, which is past EXPRESS_MAX, so this
+          // is a real test rather than a formality.
+          priority: urls.length <= EXPRESS_MAX ? 1 : 0,
+        },
       );
 
       await q.completeSubmission(client, submissionId, {

packages/db/migrations/20260819145808_feed_priority.sql

@@ -0,0 +1,25 @@
+-- An express lane for feeds a person submitted by hand.
+--
+-- `dueFeeds` orders by `next_fetch_at asc`, which is right for a directory that
+-- is keeping up and exactly wrong for one that is not. A new submission is
+-- stamped with `now`, and there are ~307,000 feeds from the bulk uploads whose
+-- next_fetch_at is already in the past -- so a blog somebody submits today
+-- sorts behind every one of them and is not crawled for days. The submitter
+-- watches a status page that says "pending" and concludes the site is broken.
+--
+-- Priority is set only by the submit route, and only for a submission small
+-- enough to have come from a person rather than an export. It is not a general
+-- scheduling knob: nothing else writes it.
+alter table feeds add column priority integer not null default 0;
+
+-- Partial on purpose, and both halves of the predicate matter.
+--
+-- `priority > 0` keeps the index to the handful of hand-submitted feeds instead
+-- of all 416,000. `last_fetched_at is null` is what makes the lane self-
+-- clearing: the crawler sets that column on success *and* on failure, so a feed
+-- leaves this index after exactly one attempt and cannot camp in the fast lane
+-- for ever. Together they mean the express query reads an index that is
+-- normally empty and never larger than one afternoon's submissions.
+create index if not exists idx_feeds_express
+  on feeds (next_fetch_at)
+  where priority > 0 and last_fetched_at is null;

packages/db/src/queries.js

@@ -2179,14 +2179,69 @@ export async function recentlyCrawled(db, limit = 20) {
   return rows;
 }
 
+/** The columns a crawl needs off a feed row. Shared by both due queries. */
+const DUE_COLUMNS = `id, slug, title, feed_url, error_count, fetch_interval_minutes, source_kind,
+                 item_count, last_published_at,
+                 http_etag, http_last_modified, content_hash, change_log`;
+
+/**
+ * The share of a tick reserved for hand-submitted feeds.
+ *
+ * Half, so the express lane cannot starve the backlog no matter how many
+ * submissions arrive: a tick always spends at least half of itself on the
+ * ordinary queue. In practice the reservation is never taken up — real people
+ * submit a handful of blogs an hour against a batch of twenty-five — so this
+ * is a ceiling on the pathological case rather than a division of normal work.
+ */
+const EXPRESS_SHARE = 0.5;
+
 /**
- * Feeds whose next_fetch_at has passed.
+ * Feeds a person submitted by hand and that have never been crawled.
+ *
+ * The express lane. Read before the ordinary queue because ordering the two
+ * together cannot work: they are ordered by `next_fetch_at asc` and the backlog
+ * is *older*, so a submission stamped `now` sorts last behind ~307,000 feeds
+ * that were overdue before it arrived. Sorting by priority instead would put a
+ * 416,000-row sort in front of every tick. Two reads, the first against a
+ * partial index that is normally empty, is the cheap way to say "these first".
+ *
+ * `last_fetched_at is null` is in the predicate rather than being cleared after
+ * the fact, so this expedites the *first* crawl only — which is the whole of
+ * what a submitter is waiting for. Afterwards the feed is scheduled on its own
+ * publishing rhythm like everything else.
+ *
+ * @param {Client} db
+ * @param {number} [limit]
+ * @returns {Promise<object[]>}
+ */
+export async function expressFeeds(db, limit = 25) {
+  if (limit <= 0) return [];
+
+  const { rows } = await db.execute({
+    sql: `select ${DUE_COLUMNS}
+          from feeds
+          where priority > 0 and last_fetched_at is null
+            and status <> 'dead' and next_fetch_at <= ?
+          order by next_fetch_at asc limit ?`,
+    args: [nowIso(), limit],
+  });
+  return rows;
+}
+
+/**
+ * Feeds whose next_fetch_at has passed, hand-submitted ones first.
  *
  * @param {Client} db
  * @param {number} [limit]
  * @returns {Promise<object[]>}
  */
 export async function dueFeeds(db, limit = 25) {
+  // Bounded rather than unbounded even though the express table is tiny: the
+  // point of a reserved share is that it is reserved in both directions.
+  const express = await expressFeeds(db, Math.floor(limit * EXPRESS_SHARE));
+  const remaining = limit - express.length;
+  if (remaining <= 0) return express;
+
   const { rows } = await db.execute({
     // slug and title are along for the log: a crawler log line that names the
     // blog and links to its page is worth two columns the crawl itself is
@@ -2206,15 +2261,19 @@ export async function dueFeeds(db, limit = 25) {
     // added in 0032: what the server's validators were, what the feed contained,
     // and when we last saw that change. They are what lets a feed stating no
     // dates be scheduled on evidence rather than on the doubling ladder.
-    sql: `select id, slug, title, feed_url, error_count, fetch_interval_minutes, source_kind,
-                 item_count, last_published_at,
-                 http_etag, http_last_modified, content_hash, change_log
+    //
+    // The express rows are excluded rather than deduplicated afterwards: they
+    // have just been read and are about to be crawled, and handing the same
+    // feed to two workers in one tick is two crawls of it.
+    sql: `select ${DUE_COLUMNS}
           from feeds
           where status <> 'dead' and next_fetch_at <= ?
+            and not (priority > 0 and last_fetched_at is null)
           order by next_fetch_at asc limit ?`,
-    args: [nowIso(), limit],
+    args: [nowIso(), remaining],
   });
-  return rows;
+
+  return express.concat(rows);
 }
 
 /**
@@ -2336,13 +2395,14 @@ export async function insertFeedsBulk(db, feeds) {
 
   const now = nowIso();
   const row = `(?, ?, ?, ?, ?, null, null, null, null, '[]', 'pending', null, null, null, 0,
-                60, ?, 0, ?, ?, ?)`;
+                60, ?, 0, ?, ?, ?, ?)`;
 
   const result = await db.execute({
     sql: `insert into feeds
       (id, slug, feed_url, site_url, title, description, language, image_url, author,
        categories, status, last_fetched_at, last_success_at, last_error, error_count,
-       fetch_interval_minutes, next_fetch_at, item_count, submission_id, created_at, updated_at)
+       fetch_interval_minutes, next_fetch_at, item_count, submission_id, created_at, updated_at,
+       priority)
       values ${feeds.map(() => row).join(', ')}
       on conflict do nothing`,
     args: feeds.flatMap((f) => [
@@ -2355,6 +2415,10 @@ export async function insertFeedsBulk(db, feeds) {
       f.submission_id ?? null,
       now,
       now,
+      // Zero unless the caller is putting a hand-submitted feed in the express
+      // lane. Written here rather than defaulted by the column so that a row
+      // inserted by an import is explicit about not being expedited.
+      Number(f.priority) > 0 ? 1 : 0,
     ]),
   });