From 2bb8463c09b56c637a2d883474dd969bde7ed7d5 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 27 Jul 2026 14:48:38 +0000 Subject: [PATCH] Give direct-publish items their own tick so a slow AI backlog can't block them MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Reported: after clearing all articles/media and rescanning every source, items in "No AI" categories weren't publishing instantly like they should. Root cause: runSynthesisCycle bundled three unrelated jobs into one function, all guarded by a single reentrancy lock (added earlier this session to stop the AI-merge path from racing itself into duplicate articles): (1) YouTube/Nitter/Telegram direct-publish, (2) "No AI" category direct-publish, (3) embed/cluster/AI-merge. A mass rescan produces a big backlog of slow generate() calls for (3) — each one can run minutes on this CPU-only hardware — and since the whole function shared one guard, a newly-ingested "No AI" item had to wait for that entire backlog to drain before its own (fast, no-AI-needed) publish step even got a turn. Split into two independently-scheduled, independently-guarded ticks: runDirectPublishCycle (source-type-driven + "No AI"-category items, regardless of Ollama's reachability) and runSynthesisCycle (now only the embed/cluster/merge path). They operate on disjoint item sets, so running them "concurrently" is safe — no risk of the duplicate-publish race the shared guard was originally added to prevent. Verified directly: with a mock provider whose generate() call takes 3 seconds (standing in for a multi-minute real one), a "No AI" category item published in 68ms — before the slow merge was even close to finishing — while the merge itself still completed correctly on its own schedule. --- backend/src/queue/priorityQueue.ts | 75 +++++++++++++++++++++--------- backend/src/queue/scheduler.ts | 27 ++++++++++- 2 files changed, 77 insertions(+), 25 deletions(-) diff --git a/backend/src/queue/priorityQueue.ts b/backend/src/queue/priorityQueue.ts index 15323e3..679f594 100644 --- a/backend/src/queue/priorityQueue.ts +++ b/backend/src/queue/priorityQueue.ts @@ -88,7 +88,10 @@ async function publishItemsDirect( * pipeline, this doesn't wait out the hold-before-publish window: that window exists to * give corroborating sources time to arrive before an AI merge locks in, which doesn't * apply here since there's no merging happening at all — each item is just itself. - * Still respects category priority. + * Still respects category priority. Harmless overlap with runDirectPublishCycle (which + * runs regardless of reachability) — an item already published by one is simply gone + * from the other's next "unclustered" query, since assignCluster lands before either + * moves on to its next item. */ export async function runPassthroughCycle(settings: GlobalSettings): Promise { const activeEvents = eventsDb.listActiveEvents(); @@ -107,41 +110,31 @@ export async function runPassthroughCycle(settings: GlobalSettings): Promise { +export async function runDirectPublishCycle(settings: GlobalSettings): Promise { const activeEvents = eventsDb.listActiveEvents(); const items = contentItemsDb.unclusteredItemsExcludingSources([]); if (items.length === 0) return 0; - // One fetch of the full source list per cycle, reused below for both the - // direct-publish partition and each item's category/type lookups — avoids a - // separate sourcesDb.getSource() round-trip per item. const sourcesById = new Map(sourcesDb.listSources().map((s) => [s.id, s])); const categories = categoriesDb.listCategories(); - const rankByName = new Map(categories.map((c) => [c.name.toLowerCase(), c.priorityRank])); - // YouTube videos, Nitter tweets, and Telegram messages never get LLM-merged with - // anything else — each is always its own article, same shape whether the AI service - // is up or not. Route them straight to publishDirect, same as the no-AI passthrough path. const directPublishSourceIds = new Set( [...sourcesById.values()].filter((s) => s.type === 'youtube' || s.type === 'nitter' || s.type === 'telegram').map((s) => s.id) ); const [typeDirectItems, remaining] = partition(items, (item) => directPublishSourceIds.has(item.sourceId)); - // A category with disableAi set (see the Category priority admin pane) opts its - // items out of clustering/synthesis entirely — each publishes on its own, using its - // own source's text, same as the source-type-driven direct items above. const aiDisabledCategoryNames = new Set(categories.filter((c) => c.disableAi).map((c) => c.name.toLowerCase())); - const [categoryDirectItems, mergeableItems] = partition(remaining, (item) => - inAiDisabledCategory(item, aiDisabledCategoryNames, sourcesById) - ); + const [categoryDirectItems] = partition(remaining, (item) => inAiDisabledCategory(item, aiDisabledCategoryNames, sourcesById)); const publishedTypeDirect = await publishItemsDirect( typeDirectItems, @@ -159,6 +152,42 @@ export async function runSynthesisCycle(provider: InferenceProvider, settings: G 'Direct publish failed' ); + return publishedTypeDirect + publishedCategoryDirect; +} + +/** + * One pass of the synthesis queue: cluster whatever's unclustered (excluding items + * runDirectPublishCycle already owns — see there), ordered by admin-defined category + * priority, and publish clusters that have cleared the hold-before-publish window. + * Items claimed by an active tracked event (belonging to one of its sources and + * matching its keyword filter, if any) publish exactly like everything else — + * individually or merged with same-story coverage — just tagged with the event's id so + * they're browsable under it and eligible for eventsRecap.ts's periodic AI wrap-up. + */ +export async function runSynthesisCycle(provider: InferenceProvider, settings: GlobalSettings): Promise { + const activeEvents = eventsDb.listActiveEvents(); + const items = contentItemsDb.unclusteredItemsExcludingSources([]); + if (items.length === 0) return 0; + + // One fetch of the full source list per cycle, reused below for both the + // direct-publish exclusion and each item's category/rank lookups — avoids a + // separate sourcesDb.getSource() round-trip per item. + const sourcesById = new Map(sourcesDb.listSources().map((s) => [s.id, s])); + const categories = categoriesDb.listCategories(); + const rankByName = new Map(categories.map((c) => [c.name.toLowerCase(), c.priorityRank])); + + // YouTube/Nitter/Telegram items and AI-disabled-category items are runDirectPublishCycle's + // job (its own guarded tick, so a slow merge backlog here never blocks them) — excluded + // here too since a batch just ingested this instant may still be unclustered when this + // runs before that cycle's own pass gets to it. + const directPublishSourceIds = new Set( + [...sourcesById.values()].filter((s) => s.type === 'youtube' || s.type === 'nitter' || s.type === 'telegram').map((s) => s.id) + ); + const aiDisabledCategoryNames = new Set(categories.filter((c) => c.disableAi).map((c) => c.name.toLowerCase())); + const mergeableItems = items.filter( + (item) => !directPublishSourceIds.has(item.sourceId) && !inAiDisabledCategory(item, aiDisabledCategoryNames, sourcesById) + ); + const ranked = mergeableItems .map((item) => ({ item, rank: primaryCategoryRank(item, rankByName, sourcesById) })) .sort((a, b) => a.rank - b.rank) @@ -216,5 +245,5 @@ export async function runSynthesisCycle(provider: InferenceProvider, settings: G ); } - return published + publishedTypeDirect + publishedCategoryDirect; + return published; } diff --git a/backend/src/queue/scheduler.ts b/backend/src/queue/scheduler.ts index f75670d..28074a2 100644 --- a/backend/src/queue/scheduler.ts +++ b/backend/src/queue/scheduler.ts @@ -1,5 +1,5 @@ import { pollDueSources } from '../ingestion/poller.js'; -import { runSynthesisCycle, runPassthroughCycle } from './priorityQueue.js'; +import { runSynthesisCycle, runPassthroughCycle, runDirectPublishCycle } from './priorityQueue.js'; import { runEventRecaps } from './eventsRecap.js'; import { runRetentionSweep } from './retention.js'; import { OllamaProvider } from '../inference/ollama-provider.js'; @@ -10,6 +10,7 @@ import { pollStocksNow } from '../stocks/poller.js'; import { pollPoe2Now } from '../poe2/poller.js'; const POLL_TICK_MS = 60_000; // checks which sources are due every minute; each source's own interval governs actual fetch frequency +const DIRECT_PUBLISH_TICK_MS = 60_000; const SYNTHESIS_TICK_MS = 60_000; const RETENTION_TICK_MS = 60 * 60_000; // hourly const WEATHER_TICK_MS = 45 * 60_000; @@ -26,6 +27,13 @@ const POE2_TICK_MS = 60 * 60_000; // poe.ninja's own overview data doesn't refre * fresh, differently-worded article by an overlapping cycle, repeatedly, until the * first cycle's assignCluster() finally landed. Node is single-threaded, so the only * source of "concurrent" runs here is exactly this interval overlap. + * + * Each call gets its own independent `running` flag/timer — the direct-publish and + * synthesis ticks are deliberately two separate calls to this (not one shared guard) + * precisely so a slow AI-merge backlog on one never blocks the other's fast, + * no-AI-needed items from publishing on schedule. They operate on disjoint item sets + * (see priorityQueue.ts), so there's no risk of the two racing each other into a + * duplicate publish the way an overlapping call to the *same* fn would. */ function everyTickSkippingOverlap(ms: number, fn: () => Promise) { let running = false; @@ -53,6 +61,18 @@ export function startScheduler() { } }); + everyTickSkippingOverlap(DIRECT_PUBLISH_TICK_MS, async () => { + try { + const settings = settingsDb.getSettings(); + const published = await runDirectPublishCycle(settings); + if (published > 0) { + logger.info('scheduler', `Direct-publish tick: published ${published} article(s)`); + } + } catch (err) { + logger.error('scheduler', `Direct-publish tick failed: ${(err as Error).message}`); + } + }); + everyTickSkippingOverlap(SYNTHESIS_TICK_MS, async () => { try { const settings = settingsDb.getSettings(); @@ -120,5 +140,8 @@ export function startScheduler() { pollPoe2Now().catch((err) => logger.error('poe2', `Poll tick failed: ${err.message}`)); }, POE2_TICK_MS); - logger.info('scheduler', 'Started: poll every 1m, synthesis every 1m, retention every 1h, weather every 45m, stocks every 15m, poe2 every 1h'); + logger.info( + 'scheduler', + 'Started: poll every 1m, direct-publish every 1m, synthesis every 1m, retention every 1h, weather every 45m, stocks every 15m, poe2 every 1h' + ); }