From bc30a038ab8049bc8f3283a0b6ec17d48c7de5fb Mon Sep 17 00:00:00 2001 From: Tim Disney Date: Thu, 20 Aug 2026 22:27:53 -0700 Subject: [PATCH] cleanup --- backend/test/feed-timeline.spec.ts | 13 +-- docs/RUNBOOK.md | 10 ++ feed-proxy/fly.staging.toml | 24 +---- feed-proxy/fly.toml | 104 +++++++----------- feed-proxy/src/app.ts | 8 +- feed-proxy/src/index.ts | 68 ++++-------- feed-proxy/src/ingest-push.test.ts | 164 +++++++++++++++++++++++++++++ feed-proxy/src/ingest-push.ts | 68 ++++++++++++ 8 files changed, 318 insertions(+), 141 deletions(-) diff --git a/backend/test/feed-timeline.spec.ts b/backend/test/feed-timeline.spec.ts index 7ba5600..ad6b6a1 100644 --- a/backend/test/feed-timeline.spec.ts +++ b/backend/test/feed-timeline.spec.ts @@ -925,16 +925,17 @@ describe('feed timeline (D1 ingest + serve)', () => { it('leaves the revision alone when only the starved flag moves', async () => { // The reader payload holds erroring feeds only, so a crawl-capacity change - // must not make every client re-download it. + // must not make every client re-download it. One shared report object: the + // rev hashes last_error_at/next_retry_at, so calling brokenFeed() twice + // flakes whenever the wall clock crosses a second between the two reports + // (CI run 32431095138). await addSubscription(TEST_DID, FEED_A); - await reportHealth([brokenFeed(FEED_A)]); + const report = brokenFeed(FEED_A); + await reportHealth([report]); const before = (await timeline()).healthRev; await addSubscription(TEST_DID, FEED_B); - await reportHealth([ - brokenFeed(FEED_A), - { feedUrl: FEED_B, errorCount: 0, crawlStale: true }, - ]); + await reportHealth([report, { feedUrl: FEED_B, errorCount: 0, crawlStale: true }]); expect((await timeline()).healthRev).toBe(before); }); diff --git a/docs/RUNBOOK.md b/docs/RUNBOOK.md index bbf7863..30d94a0 100644 --- a/docs/RUNBOOK.md +++ b/docs/RUNBOOK.md @@ -420,6 +420,16 @@ first hop of two; this section is the second. **None of it is wired to an alert yet** — the signals are on the admin and on the proxy, and this is the list to walk when the reader looks stale but every tile above is green. +> **Running any `wrangler d1 execute` in this runbook by hand:** run it from the +> **repo root** (or any directory with no `wrangler.toml` in scope), not from +> `backend/`. The checked-in `backend/wrangler.toml` carries a +> `YOUR_D1_DATABASE_ID` placeholder that CI substitutes at deploy time, so from +> `backend/` every command fails with `Invalid uuid`. With no config in scope, +> wrangler resolves `skyreader` / `skyreader-staging` by **name** against the +> account, which is what these commands want. This applies to every D1 command +> below — including the rollback commands, which is the worst moment to +> discover it. + | Signal | Where | Healthy | How to check | | --------------------------- | ----------------------------------------------- | ---------------------------------- | -------------------------------------------------------------------------------------------------------------------- | | Crawler talking to us | `sync_state.crawler_heartbeat_at` (D1) | stamped within ~5 min | `npx wrangler d1 execute skyreader --remote --command "SELECT * FROM sync_state WHERE key = 'crawler_heartbeat_at'"` | diff --git a/feed-proxy/fly.staging.toml b/feed-proxy/fly.staging.toml index 61a3c66..3fc2f17 100644 --- a/feed-proxy/fly.staging.toml +++ b/feed-proxy/fly.staging.toml @@ -51,28 +51,14 @@ primary_region = "sjc" # (the 5-minutely crawl-set pull is the stamper now, so this window only # governs descope lag and pull-outage tolerance). WARM_ACTIVE_WINDOW_SECONDS = "86400" - # Lowered 500 → 200 alongside prod; see fly.toml for the full rationale - # (crawl set is ~5,586 feeds, so the cap now bounds per-tick CPU rather - # than chasing a fresh-everything target). + # 200/tick over prod's ~1,630-feed scoped crawl set = ~8-minute cycle, + # accepted 2026-08-21; see fly.toml for the freshness/CPU math. WARM_BATCH_CAP = "200" WARM_CONCURRENCY = "16" WARM_MENTIONS = "false" - # Governor on demand-driven feed fetches (batch inline misses + the background - # refresh a STALE read fires). Turned ON 2026-08-20 on evidence: /stats showed - # inFlight=87 with the warmer paused, i.e. 87 concurrent upstream fetches driven - # purely by reader polling, on 2 vCPUs, while the backfill was trying to drain. - # - # Set deliberately LOW during the backfill. With `fresh: 0` in the cache every - # read already falls through to the past-stale path and gets served prior - # content, so throttling fetches costs almost nothing user-visible right now — - # and the CPU it frees goes to the ingest pusher, which is the thing actually - # blocking progress. Excess fetches queue (up to FEED_FETCH_QUEUE_MAX, default - # 500) and are shed as "no fresh content", never as a feed error. - # - # Raise toward WARM_CONCURRENCY (16) or above once the backfill has drained and - # freshness matters again. Watch `feedFetch.queued` on /stats: persistently deep - # means this is too tight for real demand. - FEED_FETCH_CONCURRENCY = "8" + # Governor on demand-driven feed fetches; 16 matches WARM_CONCURRENCY + # (raised from the backfill-era 8 on 2026-08-21). See fly.toml. + FEED_FETCH_CONCURRENCY = "16" EXTRACT_CONCURRENCY = "4" EXTRACT_QUEUE_MAX = "20" diff --git a/feed-proxy/fly.toml b/feed-proxy/fly.toml index 5c691c6..005e990 100644 --- a/feed-proxy/fly.toml +++ b/feed-proxy/fly.toml @@ -2,14 +2,10 @@ app = "skyreader-feed-proxy" primary_region = "sjc" # KEEP IN SYNC WITH fly.staging.toml. The two files must differ only in `app`, the -# `INGEST_URL` in [env], and `[vm]`; everything else is deliberately identical so -# staging soaks what prod will run. -# -# `[vm]` diverges permanently: prod crawls ~5,586 feeds and needs dedicated CPU -# (see the block above [vm]); staging crawls ~100 and shared is ample. Note the -# consequence — staging CANNOT reproduce prod's CPU behaviour, so a change that -# affects per-tick crawl cost has to be reasoned about, not soaked. -# +# `INGEST_URL` in [env], and (if ever needed) `[vm]`; everything else is +# deliberately identical so staging soaks what prod will run. As of 2026-08-21 +# the [vm] blocks are identical too (both shared-cpu-2x/1024, see below), so +# staging genuinely reproduces prod's CPU behaviour again. # # SINGLETON INVARIANT — run exactly ONE machine. This app is not horizontally # scalable as written: the cache is a SQLite DB on the per-machine `proxy_data` @@ -38,35 +34,28 @@ primary_region = "sjc" # would stop warming whenever traffic is idle, defeating the purpose. min_machines_running = 1 -# performance, NOT shared. Two attempts on shared/2/1024 (2026-08-20) both starved -# the event loop within minutes: /health timing out at 12s+, and the 5-minutely -# crawl-set pull taking 93s instead of 2s. The second attempt was AFTER the -# allocation fix (resident memory 862 MB -> 123 MB) and AFTER the D1 timeline moved -# reads off this box entirely, so neither memory nor read traffic explains it. -# Shared vCPUs run on burst credits and this workload is sustained, so it simply -# cannot hold them. +# shared-2x/1024, settled 2026-08-21 after the capacity incident. Steady-state +# demand is ~0.05 cores against shared-2x's ~0.125-core sustained budget +# (2 vCPUs x 1/16 burst baseline). Three fixes made this size honest, each +# verified by measurement before the next: +# 1. Crawl-set scoping (backend handleCrawlSet): ~5,586 feeds -> ~1,630, only +# feeds with a recently-active subscriber are crawled. +# 2. No-change fast path (cache.body_hash + wantParsed=false on the warm +# path): per-poll CPU ~105ms -> ~15ms. +# 3. Covering indexes (idx_cache_warm, idx_feed_items_push, ...): the +# periodic scans stopped re-reading ~1.3 GB/min of blob pages, so the DB +# no longer needs to fit in page cache and 1 GB is plenty (rss ~160 MB). +# src/query-plans.test.ts pins every hot query to its index. # -# Memory is oversized as a side effect: performance presets carry a 2 GB/core -# minimum, and actual usage is ~123 MB. Do not read the headroom as needed. +# Before resizing on a bad-looking graph, know the instruments: loadavg on +# shared VMs counts throttle-queueing, and everywhere counts D-state disk +# waits — high load + idle CPU means iowait, not demand. Judge this box by +# /health latency (~0.12s good), the crawl-set pull duration in logs (2s good, +# 90s starved), and diskstats deltas. Changing cpu_kind migrates the machine to +# a different host and forks the volume. One variable per deploy. # -# The underlying driver was a crawl set of ~5,586 feeds — roughly 4x the ~1,330 -# this box was originally sized for. It grew because the crawl-set pull stamped -# `last_requested_at` on EVERY subscribed feed, where read traffic used to stamp -# only feeds someone actually opened. The backend now scopes the crawl set to -# feeds with a recently-active subscriber (handleCrawlSet in -# backend/src/routes/ingest.ts); once the registered set settles back near that -# original size, re-testing shared is on the table. Check the proxy's -# "Crawl set: N feed(s) registered" log line and `warm.cycleSeconds` on /stats -# before revisiting. -# performance-1x/4096, matching the live machine — one variable changes per -# deploy. CPU demand measured ~0.05 cores after the no-change fast path -# (2026-08-21). Memory is deliberately ABOVE the 1x preset: with the DB bigger -# than a 2 GB machine's page cache, the periodic scans went disk-bound (~1.3 -# GB/min reads, event loop in D-state on sync SQLite — the iowait incident). -# The covering indexes (idx_cache_warm etc.) remove that dependence; once -# diskstats confirms reads collapsed, step memory_mb to 2048, verify, and only -# then try shared-cpu-2x/1024 — judged by /health latency and crawl-set pull -# duration over 30+ minutes, never by loadavg (throttling inflates it). +# Rollback ladder if starvation returns: performance-1x/2048, then +# performance-2x/4096 (both held this workload on 2026-08-21). [vm] cpu_kind = "shared" cpus = 2 @@ -97,23 +86,16 @@ primary_region = "sjc" # resumes, but the drained-window restart is a full re-crawl burst). One day # keeps (a) responsive and (b) comfortable. WARM_ACTIVE_WINDOW_SECONDS = "86400" - # Lowered 500 → 200 on 2026-08-20, during the initial prod ingest backfill. - # - # The old 500 was sized for "~1330 active feeds need ~440 refreshes/tick to - # never go stale". That premise is gone: the crawl-set pull now stamps - # last_requested_at on EVERY subscribed feed, so the active set is ~5,586 and - # keeping it all fresh would need ~1,860/tick — unreachable at any cap this box - # can afford. Freshness is already best-effort, so the cap's real job is now to - # bound per-tick CPU rather than to chase a fresh-everything target. + # Deliberate: 200/tick over the ~1,630-feed scoped crawl set gives a full + # cycle of ~8 minutes — new items reach readers up to ~8 minutes after + # publish, slightly over the 5-minute CACHE_TTL and accepted (2026-08-21). + # The warmer is therefore permanently "saturated" by design; that log line is + # only news if cycleSeconds drifts well past ~500s. # - # 200/tick cycles all 5,586 feeds in ~28 minutes (5586/200 ticks × 60s), against - # ~11 minutes at 500. That is the accepted trade: the box was event-loop starved - # at 500 (health checks timing out for minutes at a stretch, post-deploy smoke - # failing) on 2 shared vCPUs while also draining the backfill. - # - # Raise it back once the backfill has drained and CPU headroom is measurable — - # `/stats` reports rssMb now, and the warmer logs a saturation line every tick - # it fills the cap, so both sides of the trade are visible. + # Raising it buys freshness at a linear CPU price: ~330/tick restores true + # 5-minute freshness but lifts warm-loop load from ~0.05 to ~0.08 cores + # against shared-2x's ~0.125 sustained budget — affordable, but do it as its + # own watched deploy, not as a rider. WARM_BATCH_CAP = "200" # Left at 16 deliberately: this bounds PEAK parallelism, not total work, and one # variable at a time makes the effect of the cap change readable. @@ -133,20 +115,14 @@ primary_region = "sjc" # memory if extractions start shedding under normal load. # Governor on demand-driven feed fetches (batch inline misses + the background # refresh a STALE read fires). Turned ON 2026-08-20 on evidence: /stats showed - # inFlight=87 with the warmer paused, i.e. 87 concurrent upstream fetches driven - # purely by reader polling, on 2 vCPUs, while the backfill was trying to drain. - # - # Set deliberately LOW during the backfill. With `fresh: 0` in the cache every - # read already falls through to the past-stale path and gets served prior - # content, so throttling fetches costs almost nothing user-visible right now — - # and the CPU it frees goes to the ingest pusher, which is the thing actually - # blocking progress. Excess fetches queue (up to FEED_FETCH_QUEUE_MAX, default - # 500) and are shed as "no fresh content", never as a feed error. + # inFlight=87 with the warmer paused — 87 concurrent upstream fetches driven + # purely by reader polling. Excess fetches queue (up to FEED_FETCH_QUEUE_MAX, + # default 500) and are shed as "no fresh content", never as a feed error. # - # Raise toward WARM_CONCURRENCY (16) or above once the backfill has drained and - # freshness matters again. Watch `feedFetch.queued` on /stats: persistently deep - # means this is too tight for real demand. - FEED_FETCH_CONCURRENCY = "8" + # 16 matches WARM_CONCURRENCY (raised from the backfill-era 8 on 2026-08-21; + # the backfill is long drained and fetches are cheap now). Watch + # `feedFetch.queued` on /stats: persistently deep means this is too tight. + FEED_FETCH_CONCURRENCY = "16" EXTRACT_CONCURRENCY = "4" EXTRACT_QUEUE_MAX = "20" diff --git a/feed-proxy/src/app.ts b/feed-proxy/src/app.ts index 6026791..7e58ff9 100644 --- a/feed-proxy/src/app.ts +++ b/feed-proxy/src/app.ts @@ -1987,9 +1987,13 @@ export function createApp(db: Database, config: AppConfig) { .get(now - warmRefreshThresholdMs, now, now - warmActiveWindowMs); const total = eligible?.count ?? rows.length; if (total > warmBatchCap) { + const ticks = Math.ceil(total / warmBatchCap); console.warn( - `[Proxy] Warmer saturated: ${total} author document sets due for refresh but cap is ${warmBatchCap}; ` + - `${total - warmBatchCap} wait this tick and may go stale. Raise WARM_BATCH_CAP/WARM_CONCURRENCY.` + `[Proxy] Document warm cycle: ${total} author set(s) due at ${warmBatchCap}/tick ` + + `every ${warmIntervalMs / 1000}s — a full pass takes ~${ticks} tick(s), so ` + + `documents missed by the firehose surface up to that late. This lane only ` + + `carries the load while the firehose is down; raise WARM_BATCH_CAP only if ` + + `that staleness is actually hurting readers.` ); } } diff --git a/feed-proxy/src/index.ts b/feed-proxy/src/index.ts index f151beb..d820a59 100644 --- a/feed-proxy/src/index.ts +++ b/feed-proxy/src/index.ts @@ -6,7 +6,13 @@ import { mkdirSync } from 'fs'; import { createApp, initDatabase, cleanupCache } from './app'; import { DocumentFirehose } from './jetstream'; import { pingHeartbeat } from './heartbeat'; -import { pushDirtyItems, pullCrawlSet, reportFeedHealth, type IngestConfig } from './ingest-push'; +import { + pushDirtyItems, + pullCrawlSet, + reportFeedHealth, + createPushLoop, + type IngestConfig, +} from './ingest-push'; // Config const PROXY_SECRET = process.env.PROXY_SECRET; @@ -206,55 +212,17 @@ if (INGEST_ENABLED) { secret: PROXY_SECRET, batchSize: INGEST_BATCH_SIZE, }; - let pushRunning = false; - let pushFailures = 0; - let pushBlockedUntil = 0; - - /** - * One push, which re-schedules itself while a backlog remains. - * - * INGEST_INTERVAL_MS is tuned for steady state — a trickle of freshly crawled - * items — and at 100 items per tick it drains ~400/min. That is the wrong shape - * for a backlog: the first prod backfill queued 175k items, where the interval - * (not the work) was the bottleneck and the pusher sat idle ~90% of the time. - * - * So when a push comes back with `hasMore`, go again after a short delay - * instead of waiting out the interval. This self-limits: the moment the backlog - * clears, `hasMore` is false and the loop reverts to the plain interval, with - * no configuration to remember to change back. - * - * Two guards keep it from spinning: - * - Only chain on `pushed > 0`. A push that moved nothing cannot have made - * progress, so chaining on it would busy-loop against D1 forever if - * anything ever left rows permanently dirty. - * - Only chain on success. A failure sets `pushBlockedUntil`, and the - * existing exponential backoff must own the retry timing. - */ - const runPush = (): void => { - if (pushRunning || Date.now() < pushBlockedUntil) return; - pushRunning = true; - let drainMore = false; - pushDirtyItems(db, ingestConfig) - .then((result) => { - if (result.error) { - pushFailures++; - pushBlockedUntil = Date.now() + pushBackoff(pushFailures); - console.error(`[Proxy] Ingest push failed (${pushFailures}): ${result.error}`); - return; - } - pushFailures = 0; - if (result.pushed > 0) console.log(`[Proxy] Ingest pushed ${result.pushed} item(s)`); - drainMore = result.hasMore && result.pushed > 0; - }) - .catch((error) => { - console.error('[Proxy] Ingest push error:', error); - reportError(error, { tags: { source: 'ingest-push' } }); - }) - .finally(() => { - pushRunning = false; - if (drainMore) setTimeout(runPush, INGEST_CHAIN_DELAY_MS); - }); - }; + // Drain chaining, failure backoff, and the re-entrancy guard live in + // createPushLoop (ingest-push.ts), where they are unit-tested; this wires in + // the real clock, timer, pusher, and Sentry. + const runPush = createPushLoop({ + push: () => pushDirtyItems(db, ingestConfig), + chainDelayMs: INGEST_CHAIN_DELAY_MS, + backoff: pushBackoff, + schedule: (fn, delayMs) => setTimeout(fn, delayMs), + now: Date.now, + onError: (error) => reportError(error, { tags: { source: 'ingest-push' } }), + }); setInterval(runPush, INGEST_INTERVAL_MS); let crawlSetRunning = false; diff --git a/feed-proxy/src/ingest-push.test.ts b/feed-proxy/src/ingest-push.test.ts index d079262..0437489 100644 --- a/feed-proxy/src/ingest-push.test.ts +++ b/feed-proxy/src/ingest-push.test.ts @@ -18,7 +18,9 @@ import { selectFeedHealth, MAX_HEALTH_REPORT_FEEDS, countDirtyRows, + createPushLoop, type IngestConfig, + type PushResult, } from './ingest-push'; import type { FeedItem } from './types'; @@ -504,3 +506,165 @@ describe('feed health reporting', () => { expect(result.error).toContain('503'); }); }); + +describe('push loop (drain chaining, backoff, re-entrancy)', () => { + interface Deferred { + promise: Promise; + resolve: (r: PushResult) => void; + reject: (e: unknown) => void; + } + + function deferred(): Deferred { + let resolve!: (r: PushResult) => void; + let reject!: (e: unknown) => void; + const promise = new Promise((res, rej) => { + resolve = res; + reject = rej; + }); + return { promise, resolve, reject }; + } + + // Let the loop's .then/.finally microtasks run. + const settle = () => new Promise((r) => setTimeout(r, 0)); + + interface Harness { + runPush: () => void; + pushes: Deferred[]; + scheduled: Array<{ fn: () => void; delayMs: number }>; + backoffCalls: number[]; + clock: { now: number }; + } + + function harness(): Harness { + const pushes: Deferred[] = []; + const scheduled: Array<{ fn: () => void; delayMs: number }> = []; + const backoffCalls: number[] = []; + const clock = { now: 1_000_000 }; + const runPush = createPushLoop({ + push: () => { + const d = deferred(); + pushes.push(d); + return d.promise; + }, + chainDelayMs: 1000, + backoff: (failures) => { + backoffCalls.push(failures); + return 30_000; + }, + schedule: (fn, delayMs) => scheduled.push({ fn, delayMs }), + now: () => clock.now, + }); + return { runPush, pushes, scheduled, backoffCalls, clock }; + } + + it('chains with the configured delay while a backlog remains, then stops', async () => { + const h = harness(); + + h.runPush(); + expect(h.pushes.length).toBe(1); + h.pushes[0].resolve({ pushed: 100, hasMore: true }); + await settle(); + + expect(h.scheduled.length).toBe(1); + expect(h.scheduled[0].delayMs).toBe(1000); + + // The chained run drains the rest; no further chaining once hasMore=false. + h.scheduled[0].fn(); + expect(h.pushes.length).toBe(2); + h.pushes[1].resolve({ pushed: 50, hasMore: false }); + await settle(); + expect(h.scheduled.length).toBe(1); + }); + + it('never chains on a push that moved nothing, even with hasMore set', async () => { + const h = harness(); + + h.runPush(); + h.pushes[0].resolve({ pushed: 0, hasMore: true }); + await settle(); + + expect(h.scheduled.length).toBe(0); + }); + + it('is a no-op while a push is already in flight', async () => { + const h = harness(); + + h.runPush(); + h.runPush(); + h.runPush(); + expect(h.pushes.length).toBe(1); + + h.pushes[0].resolve({ pushed: 1, hasMore: false }); + await settle(); + h.runPush(); + expect(h.pushes.length).toBe(2); + }); + + it('a failed push blocks the loop until the backoff elapses, without chaining', async () => { + const h = harness(); + + h.runPush(); + h.pushes[0].resolve({ pushed: 0, hasMore: false, error: 'HTTP 503' }); + await settle(); + expect(h.scheduled.length).toBe(0); + + // Inside the backoff window: refused. + h.clock.now += 29_999; + h.runPush(); + expect(h.pushes.length).toBe(1); + + // Past it: allowed again. + h.clock.now += 2; + h.runPush(); + expect(h.pushes.length).toBe(2); + }); + + it('the failure streak feeds the backoff and resets on success', async () => { + const h = harness(); + + h.runPush(); + h.pushes[0].resolve({ pushed: 0, hasMore: false, error: 'boom' }); + await settle(); + h.clock.now += 60_000; + + h.runPush(); + h.pushes[1].resolve({ pushed: 0, hasMore: false, error: 'boom' }); + await settle(); + h.clock.now += 60_000; + + h.runPush(); + h.pushes[2].resolve({ pushed: 5, hasMore: false }); + await settle(); + + h.runPush(); + h.pushes[3].resolve({ pushed: 0, hasMore: false, error: 'boom' }); + await settle(); + + expect(h.backoffCalls).toEqual([1, 2, 1]); + }); + + it('a rejected push reports the error and leaves the loop runnable', async () => { + const errors: unknown[] = []; + const pushes: Deferred[] = []; + const runPush = createPushLoop({ + push: () => { + const d = deferred(); + pushes.push(d); + return d.promise; + }, + chainDelayMs: 1000, + backoff: () => 30_000, + schedule: () => {}, + now: () => 0, + onError: (e) => errors.push(e), + }); + + runPush(); + pushes[0].reject(new Error('network down')); + await settle(); + expect(errors.length).toBe(1); + + runPush(); + expect(pushes.length).toBe(2); + }); +}); diff --git a/feed-proxy/src/ingest-push.ts b/feed-proxy/src/ingest-push.ts index 9f0c990..b1d1b0a 100644 --- a/feed-proxy/src/ingest-push.ts +++ b/feed-proxy/src/ingest-push.ts @@ -80,6 +80,74 @@ export const DIRTY_COUNT_SQL = `SELECT COUNT(*) AS count LEFT JOIN push_state ps ON ps.seq = fi.seq WHERE ps.seq IS NULL OR ps.pushed_hash <> COALESCE(fi.content_hash, '')`; +export interface PushLoopDeps { + /** One push attempt — pushDirtyItems bound to this proxy's db/config. */ + push: () => Promise; + /** Gap between back-to-back pushes while draining a backlog. */ + chainDelayMs: number; + /** Backoff schedule after `failures` consecutive failed pushes. */ + backoff: (failures: number) => number; + schedule: (fn: () => void, delayMs: number) => void; + now: () => number; + /** Unexpected-rejection hook (Sentry in prod). */ + onError?: (error: unknown) => void; +} + +/** + * The push loop: one push, which re-schedules itself while a backlog remains. + * The caller drives the steady-state cadence (setInterval in index.ts); this + * owns the drain chaining, the failure backoff, and the re-entrancy guard. + * + * The steady-state interval is tuned for a trickle of freshly crawled items — + * at 100 items per tick it drains ~400/min. That is the wrong shape for a + * backlog: the first prod backfill queued 175k items, where the interval (not + * the work) was the bottleneck and the pusher sat idle ~90% of the time. So + * when a push comes back with `hasMore`, go again after `chainDelayMs` instead + * of waiting out the interval. This self-limits: the moment the backlog clears, + * `hasMore` is false and the loop reverts to the plain interval, with no + * configuration to remember to change back. + * + * Two guards keep it from spinning: + * - Only chain on `pushed > 0`. A push that moved nothing cannot have made + * progress, so chaining on it would busy-loop against D1 forever if + * anything ever left rows permanently dirty. + * - Only chain on success. A failure blocks the loop until `backoff` elapses, + * and the backoff owns the retry timing. + */ +export function createPushLoop(deps: PushLoopDeps): () => void { + let running = false; + let failures = 0; + let blockedUntil = 0; + + const runPush = (): void => { + if (running || deps.now() < blockedUntil) return; + running = true; + let drainMore = false; + deps + .push() + .then((result) => { + if (result.error) { + failures++; + blockedUntil = deps.now() + deps.backoff(failures); + console.error(`[Proxy] Ingest push failed (${failures}): ${result.error}`); + return; + } + failures = 0; + if (result.pushed > 0) console.log(`[Proxy] Ingest pushed ${result.pushed} item(s)`); + drainMore = result.hasMore && result.pushed > 0; + }) + .catch((error) => { + console.error('[Proxy] Ingest push error:', error); + deps.onError?.(error); + }) + .finally(() => { + running = false; + if (drainMore) deps.schedule(runPush, deps.chainDelayMs); + }); + }; + return runPush; +} + export function selectDirtyRows(db: Database, limit: number): DirtyRow[] { return db.query(DIRTY_ROWS_SQL).all(limit); }