Salastil/homefeed
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Give direct-publish items their own tick so a slow AI backlog can't block them

Browse Source 
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.
This commit is contained in:
Claude
2026-07-27 14:48:38 +00:00
parent afdb5ad036
commit 2bb8463c09
2 changed files with 77 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/queue/priorityQueue.ts
+52 -23
View File
@@ -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<number> {
const activeEvents = eventsDb.listActiveEvents();
@@ -107,41 +110,31 @@ export async function runPassthroughCycle(settings: GlobalSettings): Promise<num
}
/**
* One pass of the synthesis queue: cluster whatever's unclustered, 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.
* Publishes every item that never needs the AI at all: YouTube/Nitter/Telegram items
* (always their own article, merge or no merge) and items whose category has "No AI"
* set (Category priority admin pane). Deliberately its own guarded tick in scheduler.ts,
* separate from runSynthesisCycle — these items have no reason to wait behind a slow AI
* merge backlog (a generate() call can run minutes on CPU-only inference; see
* ollama-provider.ts), and runSynthesisCycle's own reentrancy guard used to make them
* do exactly that: stuck for however long the current cycle's clustering/synthesis
* portion took, since both used to run under one guarded function. Runs regardless of
* Ollama's reachability — nothing here calls the AI.
*/
export async function runSynthesisCycle(provider: InferenceProvider, settings: GlobalSettings): Promise<number> {
export async function runDirectPublishCycle(settings: GlobalSettings): Promise<number> {
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<number> {
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;
}
backend/src/queue/scheduler.ts
+25 -2
View File
@@ -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<void>) {
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'
);
}