fix(crawl): stop one host starving a tier; show header follows the feed (0.6.1)

.github/workflows/ci.yml

@@ -10,6 +10,22 @@ jobs:
   test:
     name: install, typecheck, test, build (Bun)
     runs-on: ubuntu-latest
+    # The crawler's SQL is tested against a real Postgres: those tests skip without
+    # TEST_DATABASE_URL, and a scheduling query cannot be proved any other way.
+    services:
+      postgres:
+        image: postgres:16-alpine
+        env:
+          POSTGRES_PASSWORD: test
+        ports:
+          - 5432:5432
+        options: >-
+          --health-cmd "pg_isready -U postgres"
+          --health-interval 5s
+          --health-timeout 5s
+          --health-retries 10
+    env:
+      TEST_DATABASE_URL: postgres://postgres:test@127.0.0.1:5432/postgres
     steps:
       - uses: actions/checkout@v4
       # The version comes from packageManager in package.json, the same Bun the

package.json

@@ -1,6 +1,6 @@
 {
   "name": "p0dcasters",
-  "version": "0.6.0",
+  "version": "0.6.1",
   "private": true,
   "type": "module",
   "scripts": {

scripts/daemon-lib.mjs

@@ -357,6 +357,10 @@ export function buildDaemon(deps) {
     // digest back. Each task still runs one at a time on its own. The show
     // crawler's reads come on top, so it never waits behind either.
     concurrency: 2 + (crawl ? crawl.concurrency + 1 : 0),
+    // A tick starts at most `concurrency` reads, so this is the crawler's ceiling:
+    // 24 slots every 2 s is 43k reads an hour. The default 5 s (17k) only just
+    // covers the 15k an hour the catalogue needs, and can never clear a backlog.
+    ...(crawl ? { tickMs: SHOWS_TICK_MS } : {}),
     onRun,
     onStall: async (s) => {
       out.error(`[directory] stalled: no success since ${s.lastOk ? new Date(s.lastOk).toISOString() : "start"}${s.lastError ? `: ${s.lastError}` : ""}`);
@@ -387,8 +391,9 @@ export function buildDaemon(deps) {
   return daemon;
 }
 
-/** Feeds read per tier at once from one host: anchor.fm alone is 116k shows. */
+/** Feeds read at once from one host, across all tiers: anchor.fm alone is 116k shows. */
 export const SHOWS_PER_HOST = 4;
+export const SHOWS_TICK_MS = 2_000;
 export const SHOWS_JOB = "shows";
 
 /**

src/app/podcast/[slug]/page.tsx

@@ -9,7 +9,7 @@ import Art from "@/components/Art";
 import Ad from "@/components/Ad";
 import { AD_TEXT } from "@/lib/ads";
 import FollowButton from "@/components/FollowButton";
-import { LatestButton, ShowEpisodes } from "@/components/ShowEpisodes";
+import { EpisodeTotal, LatestButton, LatestDate, ShowEpisodes } from "@/components/ShowEpisodes";
 import PlaylistButtons from "@/components/PlaylistButtons";
 import ShareButton from "@/components/ShareButton";
 import RefreshButton from "@/components/RefreshButton";
@@ -165,11 +165,18 @@ export default async function Show({ params }: { params: Promise<{ slug: string
 
           <div className="facts">
             <div>
-              <b>{Number(p.episode_count).toLocaleString()}</b> episodes
+              <b>
+                <Suspense fallback={Number(p.episode_count).toLocaleString()}>
+                  <EpisodeTotal feedUrl={p.feed_url} stored={Number(p.episode_count)} />
+                </Suspense>
+              </b>{" "}
+              episodes
             </div>
             <div>
               <b>
-                <TimeAgo unix={p.newest_pubdate} />
+                <Suspense fallback={<TimeAgo unix={p.newest_pubdate} />}>
+                  <LatestDate feedUrl={p.feed_url} stored={p.newest_pubdate} />
+                </Suspense>
               </b>{" "}
               latest
             </div>

src/components/ShowEpisodes.tsx

@@ -1,4 +1,5 @@
-import { fetchEpisodes } from "@/lib/feed";
+import { fetchEpisodes, publishedNewest } from "@/lib/feed";
+import TimeAgo from "@/components/TimeAgo";
 import { EpisodeList, PlayLatest, type Show } from "@/components/Episodes";
 
 /**
@@ -13,6 +14,23 @@ export async function LatestButton({ feedUrl, show }: { feedUrl: string; show: S
   return <PlayLatest episodes={episodes} show={show} />;
 }
 
+/**
+ * The header's "N episodes" and "3d ago latest". The directory row is only as fresh
+ * as the last crawl, and this page has just read the feed, so it must not state an
+ * older date than the list under it. The stored value is the Suspense fallback.
+ * Only ever forward, like the crawler: a feed that trims its archive to its last
+ * hundred episodes has not lost the rest.
+ */
+export async function EpisodeTotal({ feedUrl, stored }: { feedUrl: string; stored: number }) {
+  const episodes = await fetchEpisodes(feedUrl);
+  return <>{Math.max(stored, episodes.length).toLocaleString()}</>;
+}
+
+export async function LatestDate({ feedUrl, stored }: { feedUrl: string; stored: number | null }) {
+  const episodes = await fetchEpisodes(feedUrl);
+  return <TimeAgo unix={Math.max(stored ?? 0, publishedNewest(episodes)) || null} />;
+}
+
 export async function ShowEpisodes({ feedUrl, show }: { feedUrl: string; show: Show }) {
   const episodes = await fetchEpisodes(feedUrl);
   return (

src/lib/feed-crawl.ts

@@ -34,6 +34,8 @@ const UA = "p0dcasters/1.0 (+https://p0dcasters.com)";
 const TIER = tierSql("minutes");
 /** The last this-many publish dates are kept, so a 304 can still be scheduled on rhythm. */
 const KEEP_DATES = 30;
+/** How many of a tier's oldest due rows dueFeeds shares out between their hosts. */
+const DUE_WINDOW = 1000;
 
 export const FEED_STATE_DDL = [
   `CREATE TABLE IF NOT EXISTS feed_state(
@@ -88,13 +90,29 @@ export async function seedFeedState(q: Query): Promise<{ added: number; dropped:
   return { added: added.length, dropped: dropped.length };
 }
 
-/** Due shows of one tier (0 hot .. 4 archived), oldest due first. */
+/**
+ * Due shows of one tier (0 hot .. 4 archived): the oldest due of each host, then
+ * each host's second oldest, and so on.
+ *
+ * Not plain oldest-first. The daemon over-reads a few dozen rows and skips any on a
+ * host already at its cap or on hold, so a head of the queue that is all one host
+ * yields nothing: in November 2026 58 of the first 60 rows of a tier were one
+ * platform, that tier read no feed at all for days, and 320k shows went overdue.
+ * Taking turns inside a window of the oldest rows keeps one host from blocking the
+ * others, and stays one short index scan.
+ */
 export async function dueFeeds(q: Query, tierIndex: number, limit: number): Promise<FeedRow[]> {
   const rows = await q(
-    `SELECT feed_url, minutes, fails, etag, last_modified, dates
-     FROM feed_state WHERE ${TIER} = $1 AND next_at <= now()
-     ORDER BY next_at LIMIT $2`,
-    [tierIndex, limit],
+    `SELECT feed_url, minutes, fails, etag, last_modified, dates FROM (
+       SELECT *, row_number() OVER (PARTITION BY lower(substring(feed_url from '^[a-z]+://([^/:?#]+)')) ORDER BY next_at) AS turn
+       FROM (
+         SELECT feed_url, minutes, fails, etag, last_modified, dates, next_at
+         FROM feed_state WHERE ${TIER} = $1 AND next_at <= now()
+         ORDER BY next_at LIMIT $3
+       ) oldest
+     ) ranked
+     ORDER BY turn, next_at LIMIT $2`,
+    [tierIndex, limit, Math.max(DUE_WINDOW, limit)],
   );
   return rows.map((r) => ({
     key: String(r.feed_url),

src/lib/feed.ts

@@ -243,6 +243,17 @@ async function loadEpisodes(feedUrl: string): Promise<Episode[]> {
   }
 }
 
+/**
+ * The newest episode date (unix seconds) a feed states, 0 if none. A date more than
+ * a day ahead is a feed's scheduling mistake, not a release, so it does not count.
+ */
+export function publishedNewest(episodes: Episode[], nowMs = Date.now()): number {
+  const latest = Math.floor(nowMs / 1000) + 86_400;
+  let best = 0;
+  for (const e of episodes) if (e.pubdate > best && e.pubdate <= latest) best = e.pubdate;
+  return best;
+}
+
 export async function fetchEpisodes(feedUrl: string): Promise<Episode[]> {
   const hit = episodeCache.get(feedUrl);
   if (hit) return hit;

test/feed-crawl.test.mjs

@@ -2,6 +2,7 @@ import { describe, expect, test, afterAll, beforeAll } from "bun:test";
 import { SQL } from "bun";
 import { createDaemon } from "@profullstack/scraper-daemon";
 import * as lib from "../src/lib/feed-crawl.ts";
+import { publishedNewest } from "../src/lib/feed.ts";
 import { addShowCrawler } from "../scripts/daemon-lib.mjs";
 
 const url = process.env.TEST_DATABASE_URL;
@@ -42,6 +43,19 @@ describe.skipIf(!url)("feed_state against Postgres", () => {
     expect(await lib.seedFeedState(q)).toEqual({ added: 0, dropped: 0 });
   });
 
+  test("one host's backlog at the head of a tier does not hide every other host", async () => {
+    await q(`DELETE FROM feed_state`);
+    // Sixty overdue shows from one platform, then the show that used to starve behind them.
+    const rows = Array.from({ length: 60 }, (_, i) => `('https://feeds.busy.example/${i}.xml', 1679, now() - interval '9 days' + ${i} * interval '1 second')`);
+    rows.push(`('https://feed.other.example/show.xml', 1679, now() - interval '8 days')`);
+    await q(`INSERT INTO feed_state(feed_url, minutes, next_at) VALUES ${rows.join(",")}`);
+    const due = await lib.dueFeeds(q, 2, 30);
+    expect(due).toHaveLength(30);
+    expect(due.map((f) => f.url).slice(0, 2)).toEqual(["https://feeds.busy.example/0.xml", "https://feed.other.example/show.xml"]);
+    await q(`DELETE FROM feed_state`);
+    await lib.seedFeedState(q);
+  });
+
   test("due shows come back per tier, and a crawl moves the directory row forward", async () => {
     await q(`UPDATE feed_state SET next_at = now() - interval '1 minute'`);
     const slow = await lib.dueFeeds(q, 2, 10);
@@ -107,3 +121,12 @@ describe("addShowCrawler", () => {
     expect(saved[0].plan.tier).toBe("hot");
   });
 });
+
+test("the show header's date is the newest real release, not a scheduling mistake", () => {
+  const at = (pubdate) => ({ pubdate });
+  const now = NOW * 1000;
+  expect(publishedNewest([], now)).toBe(0);
+  expect(publishedNewest([at(0)], now)).toBe(0);
+  expect(publishedNewest([at(NOW - DAY), at(NOW - 3 * DAY)], now)).toBe(NOW - DAY);
+  expect(publishedNewest([at(NOW + 30 * DAY), at(NOW - DAY)], now)).toBe(NOW - DAY);
+});