From 368bd4bc0d7d908b5033604bbc3a1f43da70c785 Mon Sep 17 00:00:00 2001 From: flavius Date: Sat, 1 Aug 2026 16:20:46 +0300 Subject: [PATCH 1/2] perf(convex): cut prod bill from ~$41/mo to ~$1-3/mo MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Prod was disabled for exceeding Convex free-plan limits. Action compute was 81% of the bill, and summarizationNode.processSummaryJob alone was 92% of that (~75% of the total). Convex bills action compute by WALL-CLOCK time including time spent awaiting the network, so the cost was waiting, not computing. Summarization (the dominant cost): - Default event_summary_body_fetch_enabled to false. The key had no row in prod, so it silently ran on the `true` code default, fetching an article body per selected article on every job. - Treat provider 429s as backpressure, not failure: defer without consuming an attempt. Previously a 429 burned 1 of 3 attempts, which is why ~395 jobs sat permanently dead and only ~31% of events were ever summarized. Bounded at ~24h so a permanently rate-limited job still surfaces as failed. - Stop sleeping inside the action on 429 (maxRetries: 1 on the summary path). - Body-fetch deadline 60s -> 12s; parallelize pre-work round trips. Per job: ~16s -> ~2-4s of billed wall clock. Pipeline cadence -> 4 staggered windows/day (:00 ingest, :15 enrich, :30 cluster, :45 summarize). Worst case ~6h to feed, a deliberate trade. Claim divergence and daily-quiz crons disabled; both were already off at the config layer and were waking only to find nothing to do. Self-chaining (required by the cadence change): neither enrichment nor clustering drained its backlog — each did one batch and stopped, keeping up only because crons fired every 30/40min. At 4 windows/day that is 160+128 against ~1,300 articles/day. Both now chain until drained with a bounded chainDepth, and hand off on cumulative totals so a terminal empty batch cannot strand the drain. Clustering now fires follow-on passes once per drain rather than once per batch — that per-batch kick, not the cron, was what produced ~1,680 summary jobs/day. Database I/O: - events.getPublishedEvents (top consumer): cursor-anchored ranked pagination. Deep pages previously rescanned the top 250 rows, and because recency dominates trendingScore, every publish invalidated every open page-2+ subscription. The cursor bounds the range above so new events fall outside. - pipeline.countProcessingEventsOlderThan: saturating count, 10k docs -> 400. - Bound assorted unbounded collect()/scan paths. Storage (previously unbounded, so the bill compounded monthly): retention for articleEmbeddings (45d) and archived detached articles (90d). Alerting retuned for the batched cadence, including two rules whose windows contained no pipeline run at all and so could never fire. Fixes a pre-existing ranked-pagination bug surfaced by the new tie-run tests: compareRankedPayload tie-broke on ASCENDING eventId against a descending index traversal, so a tie run longer than the page buffer left the window holding a biased sample and the feed silently dead-ended mid-list. Every comparator key now descends, and the tiebreak uses the preview row's own _id (the index is on publicEventPreviews, whose implicit last key is that table's _id). Migrations (run after deploy; prod config rows override code defaults): migrations:applyCostReductionConfig migrations:requeueRateLimitedSummaryJobs migrations:purgeOrphanedStorageFiles (reclaims ~992MB of orphaned blobs; eventShareAssets is the only _storage reference and it is empty) Co-Authored-By: Claude Opus 5 --- packages/backend/convex/clustering.ts | 128 ++++-- packages/backend/convex/config.ts | 40 +- packages/backend/convex/crons.ts | 252 ++++++++---- packages/backend/convex/enrichmentNode.ts | 73 +++- packages/backend/convex/events.ts | 384 +++++++++++------ .../backend/convex/feedPagination.test.ts | 264 ++++++++++++ packages/backend/convex/lib/aiCall.ts | 26 ++ .../backend/convex/lib/feedSerialization.ts | 30 +- packages/backend/convex/migrations.ts | 253 ++++++++++++ packages/backend/convex/pipeline.ts | 293 +++++++++++-- packages/backend/convex/retention.test.ts | 346 +++++++++++++++- packages/backend/convex/retention.ts | 385 +++++++++++++++++- packages/backend/convex/summarization.ts | 34 +- packages/backend/convex/summarizationNode.ts | 79 +++- 14 files changed, 2292 insertions(+), 295 deletions(-) create mode 100644 packages/backend/convex/feedPagination.test.ts diff --git a/packages/backend/convex/clustering.ts b/packages/backend/convex/clustering.ts index 7debba1..29f41e3 100644 --- a/packages/backend/convex/clustering.ts +++ b/packages/backend/convex/clustering.ts @@ -50,6 +50,13 @@ const MERGE_LOCK_KEY = "mergeNearDuplicateEvents"; const RECLUSTER_SINGLETONS_LOCK_KEY = "reclusterRecentSingletonEvents"; const MERGE_LOCK_TTL_MS = 20 * 60 * 1000; const CLUSTER_BATCH_SIZE = 32; +// Self-chaining bounds for clusterEnrichedArticles. 60 x CLUSTER_BATCH_SIZE(32) +// = 1,920 articles per triggered drain, which covers a full day of intake +// (~1,300/day) with headroom while guaranteeing the chain always terminates. +// The delay is long enough that the pipeline lock from the previous batch has +// been released before the next one tries to acquire it. +const MAX_CLUSTER_CHAIN_DEPTH = 60; +const CLUSTER_CHAIN_DELAY_MS = 10_000; const RECENT_EVENT_WINDOW_MS = 48 * 60 * 60 * 1000; const MAX_CANDIDATE_EVENTS = 220; // Each vector-search neighbor is hydrated (candidacy + ~10KB embedding doc), @@ -6230,10 +6237,62 @@ export const reclusterRecentSingletonEvents = internalAction({ }, }); +/** + * Follow-on passes scheduled once a clustering DRAIN completes. + * + * These are deliberately deferred to the end of the self-chain rather than run + * per batch: previously each clustering batch kicked summarization directly, + * which is what actually drove summary volume (~1,680 processSummaryJob + * invocations/day) and made it the largest line item on the Convex bill. + * + * The totals passed in are CUMULATIVE across the whole chain. Using per-batch + * counters here would strand a drain whose final batch happened to cluster + * nothing — the events created by earlier links would never be summarized, and + * since summaries gate publishing, they would never go public. + */ +async function scheduleClusteringFollowUps( + ctx: ActionCtx, + totals: { clusteredIntoExisting: number; createdEvents: number }, +): Promise { + const touchedEvents = totals.clusteredIntoExisting + totals.createdEvents; + if (touchedEvents > 0) { + await ctx.scheduler.runAfter( + MERGE_NEAR_DUPLICATES_DELAY_MS, + internal.clustering.mergeNearDuplicateEvents, + {}, + ); + } + if (totals.createdEvents > 0) { + await ctx.scheduler.runAfter( + RECLUSTER_RECENT_SINGLETONS_DELAY_MS, + internal.clustering.reclusterRecentSingletonEvents, + {}, + ); + } + // Events publish only after a summary, so kick summarization once the drain + // finishes rather than waiting up to 6h for the next cron window. + if (touchedEvents > 0) { + await ctx.scheduler.runAfter( + 0, + internal.summarizationNode.summarizeQueuedEvents, + {}, + ); + } +} + export const clusterEnrichedArticles = internalAction({ - args: {}, + // These are set only by this action rescheduling itself; the cron, + // enrichment's hand-off, and manual invocations all start at 0. The + // `*SoFar` counters accumulate across the chain so the terminal link can + // decide the follow-on passes from the drain's total work, not its own batch. + args: { + chainDepth: v.optional(v.number()), + clusteredSoFar: v.optional(v.number()), + createdSoFar: v.optional(v.number()), + }, handler: async ( ctx, + { chainDepth = 0, clusteredSoFar = 0, createdSoFar = 0 }, ): Promise<{ clusteredIntoExisting: number; createdEvents: number; @@ -6260,6 +6319,13 @@ export const clusterEnrichedArticles = internalAction({ console.log( `[clustering] clusterEnrichedArticles already running (owner=${lock.owner}, expiresAt=${new Date(lock.expiresAt).toISOString()})`, ); + // A chained link that loses the lock still has to hand off whatever the + // earlier links accomplished, or that work is stranded unsummarized (and + // therefore unpublished) until a later cron window happens to pick it up. + await scheduleClusteringFollowUps(ctx, { + clusteredIntoExisting: clusteredSoFar, + createdEvents: createdSoFar, + }); return { clusteredIntoExisting: 0, createdEvents: 0, @@ -6286,6 +6352,12 @@ export const clusterEnrichedArticles = internalAction({ ); if (!hasEnriched) { console.log("[clustering] No enriched articles to cluster"); + // Chain terminator: the previous link may have taken a full batch (and + // so chained) only for the backlog to be empty by the time this ran. + await scheduleClusteringFollowUps(ctx, { + clusteredIntoExisting: clusteredSoFar, + createdEvents: createdSoFar, + }); return { clusteredIntoExisting: 0, createdEvents: 0, @@ -6302,6 +6374,11 @@ export const clusterEnrichedArticles = internalAction({ if (articles.length === 0) { console.log("[clustering] No enriched articles to cluster"); + // Chain terminator — see the note on the !hasEnriched branch above. + await scheduleClusteringFollowUps(ctx, { + clusteredIntoExisting: clusteredSoFar, + createdEvents: createdSoFar, + }); return { clusteredIntoExisting: 0, createdEvents: 0, @@ -6862,30 +6939,35 @@ export const clusterEnrichedArticles = internalAction({ clusterRunReservationSettled = true; } await flushJobMetrics(ctx, metrics, startedAt); - if (clusteredIntoExisting + createdEvents > 0) { - await ctx.scheduler.runAfter( - MERGE_NEAR_DUPLICATES_DELAY_MS, - internal.clustering.mergeNearDuplicateEvents, - {}, - ); - } - if (createdEvents > 0) { - await ctx.scheduler.runAfter( - RECLUSTER_RECENT_SINGLETONS_DELAY_MS, - internal.clustering.reclusterRecentSingletonEvents, - {}, - ); - } - // Events publish only after a summary, so kick summarization immediately - // after a clustering batch instead of waiting up to 45 min for the cron — - // any event that just crossed the summary/publish bar gets its perspective - // summaries + globalImpact (and goes public) as soon as possible. - if (clusteredIntoExisting + createdEvents > 0) { + + // Self-chain until the enriched-article backlog is drained. + // + // This action clusters at most CLUSTER_BATCH_SIZE articles per run. That + // kept up when the cron fired every 40 minutes, but the pipeline now runs + // in 4 batched windows per day for cost reasons, and 4 x 32 = 128 + // articles/day is far below the ~1,300/day intake. Without chaining the + // backlog would grow forever and the feed would stall. + const likelyMoreWaiting = articles.length === CLUSTER_BATCH_SIZE; + const shouldChain = + likelyMoreWaiting && chainDepth + 1 < MAX_CLUSTER_CHAIN_DEPTH; + const clusteredTotal = clusteredSoFar + clusteredIntoExisting; + const createdTotal = createdSoFar + createdEvents; + + if (shouldChain) { await ctx.scheduler.runAfter( - 0, - internal.summarizationNode.summarizeQueuedEvents, - {}, + CLUSTER_CHAIN_DELAY_MS, + internal.clustering.clusterEnrichedArticles, + { + chainDepth: chainDepth + 1, + clusteredSoFar: clusteredTotal, + createdSoFar: createdTotal, + }, ); + } else { + await scheduleClusteringFollowUps(ctx, { + clusteredIntoExisting: clusteredTotal, + createdEvents: createdTotal, + }); } return { diff --git a/packages/backend/convex/config.ts b/packages/backend/convex/config.ts index 00f2e34..2e23a40 100644 --- a/packages/backend/convex/config.ts +++ b/packages/backend/convex/config.ts @@ -697,13 +697,13 @@ export const seedDefaults = internalMutation({ key: "event_summary_enqueue_limit", value: 40, description: - "Maximum number of recent published events inspected for summary eligibility per summarization run.", + "Maximum number of recent published events inspected for summary eligibility per summarization run. This scan (summarization.enqueueEligibleEventSummaries) was the 2nd largest database-I/O consumer in the app at 2.15 GB, but the fix was cadence, not depth: dropping from 32 runs/day to 4 already cuts that I/O ~8x. Lowering this further would starve the queue of eligible events instead, since summaries gate publishing.", }, { key: "event_summary_batch_size", - value: 8, + value: 12, description: - "Maximum number of queued event summary jobs processed per summarization run. Summaries gate publishing, so this is sized to keep pace with clustering rather than trickle.", + "Maximum number of queued event summary jobs processed per summarization run. Summaries gate publishing, so this must keep pace with the eligible-event rate or the feed stalls: at 4 runs/day this allows ~48 summaries/day. Jobs are staggered JOB_STAGGER_MS apart (~7.5 requests/min), which stays inside Gemini's free-tier RPM; the 429 backpressure path defers rather than failing, so overshoot self-corrects instead of burning attempts.", }, { key: "event_summary_max_attempts", @@ -725,15 +725,15 @@ export const seedDefaults = internalMutation({ }, { key: "event_summary_max_input_articles", - value: 8, + value: 6, description: - "Maximum number of recent articles included in one event summarization prompt. Lower values reduce data egress (fewer bodies fetched and sent to the model) and token cost.", + "Maximum number of recent articles included in one event summarization prompt. Reduced in cost mode (12 -> 8 in #60, now 6): prompt size drives both data egress (billed per GB leaving Convex) and model latency, which is itself billed as action compute. 6 articles are still enough for a multi-perspective summary.", }, { key: "event_summary_body_fetch_enabled", - value: true, + value: false, description: - "When true, the summarizer fetches each selected article's body transiently at summarization time (used in memory for the prompt, never stored) instead of relying only on the short extracted summary + RSS snippet.", + "When true, the summarizer fetches each selected article's body transiently at summarization time (used in memory for the prompt, never stored) instead of relying only on the short extracted summary + RSS snippet. DEFAULT OFF (cost mode): Convex bills action compute by wall-clock time including network waits, so fetching one body per selected article (up to event_summary_max_input_articles per job) made this the single most expensive operation in the app. Turning it on again is the biggest cost regression available — measure before doing so.", }, { key: "event_summary_body_chars", @@ -1169,9 +1169,21 @@ export const seedDefaults = internalMutation({ }, { key: "pipeline_alert_check_interval_minutes", - value: 20, + value: 720, + description: + "Nominal interval for pipeline alert checks. Must track the check-pipeline-alerts cron in crons.ts (cost mode: 2x daily), otherwise absent-run and staleness alerts fire spuriously.", + }, + { + key: "archived_article_retention_days", + value: 90, + description: + "Age after which an ARCHIVED, event-detached article row (and its embeddings) is deleted. Only applies to articles archived by singletonCleanup as stale singletons/processing events, which belong to no event — never to articles reachable from the feed. Hand-labeled clusterPairLabels rows are excluded by the purge job.", + }, + { + key: "article_embedding_retention_days", + value: 45, description: - "Nominal interval for pipeline alert checks.", + "Age after which an article's 512-dimension embedding row is deleted. Clustering only ever compares recent articles, so older embeddings are dead weight — and unbounded articleEmbeddings growth (~1,300 articles/day) was the main driver of database storage cost. Article rows themselves are retained; only the vectors are purged.", }, ]; @@ -1216,10 +1228,12 @@ export const seedDefaults = internalMutation({ article_fact_extraction_model: ['"gpt-5-nano"'], article_bias_detection_model: ['"gpt-5-nano"'], claim_analysis_model: ['"gpt-5-nano"'], - // Migrate the prior summary input cap (12) down to the new egress-reduced - // default (8). This key is intentionally not force-managed, so operator - // overrides (any other value) are preserved. - event_summary_max_input_articles: ["12"], + // Migrate the prior summary input caps down to the current default. + // "12" was the original; "8" was the intermediate egress-reduced value + // from #60, which never reached prod before the deployment was disabled — + // both are listed so either state converges. This key is intentionally + // not force-managed, so operator overrides (any other value) are kept. + event_summary_max_input_articles: ["12", "8"], }; let created = 0; diff --git a/packages/backend/convex/crons.ts b/packages/backend/convex/crons.ts index c160406..a33b7c3 100644 --- a/packages/backend/convex/crons.ts +++ b/packages/backend/convex/crons.ts @@ -3,14 +3,31 @@ import { internal } from "./_generated/api"; const crons = cronJobs(); +// =========================================================================== +// COST MODE — batched pipeline windows +// =========================================================================== +// Convex bills action compute by WALL-CLOCK time (including time spent waiting +// on the network), and database I/O by bytes read. Running the pipeline +// continuously was costing ~$41/mo against a <$10/mo budget, so the chain now +// runs in 4 batch windows per day instead of on independent short intervals. +// +// The stages are phase-staggered inside each window (:00 ingest → :15 enrich → +// :30 cluster → :45 summarize) so a story still flows end-to-end within ~45 +// minutes of being ingested. `crons.interval` is epoch-phase-aligned, which +// would have fired every stage simultaneously and added a full window of +// latency per stage, so these use explicit cron expressions instead. +// +// Freshness cost: worst-case ~6h from publication to appearing in the feed. +// This is a deliberate, authorised trade to keep the app online. // --------------------------------------------------------------------------- -// RSS Ingestion — Every 60 minutes + +// --------------------------------------------------------------------------- +// RSS Ingestion — 4x daily (00:00, 06:00, 12:00, 18:00 UTC) // --------------------------------------------------------------------------- // Fetches all curated RSS feeds, deduplicates articles, and inserts new ones. -// Frequency can be tuned down to 30min once the pipeline is proven stable. -crons.interval( +crons.cron( "ingest-rss-feeds", - { minutes: 60 }, + "0 0,6,12,18 * * *", internal.ingestion.ingestAllFeeds, ); @@ -21,123 +38,151 @@ crons.interval( // automated refresh is ever wanted again. // --------------------------------------------------------------------------- -// Article Enrichment (Embeddings) — Every 30 minutes +// Article Enrichment (Embeddings) — 4x daily (:15 past each window) // --------------------------------------------------------------------------- -// Generates embeddings for unprocessed articles. -// Runs more frequently than ingestion to keep the pipeline flowing and to work -// down the unprocessed-article backlog faster. -crons.interval( +// Generates embeddings for unprocessed articles. Runs 15 minutes after each +// ingest window so the freshly inserted articles are picked up in the same pass. +// The action self-chains until the article backlog is drained, so this trigger +// only needs to start each window (chainDepth defaults to 0). +crons.cron( "enrich-articles", - { minutes: 30 }, + "15 0,6,12,18 * * *", internal.enrichmentNode.enrichUnprocessedArticles, + {}, ); // --------------------------------------------------------------------------- -// Article Clustering — Every 40 minutes +// Article Clustering — 4x daily (:30 past each window) // --------------------------------------------------------------------------- // Clusters enriched articles into published events so the feed can render // real ingested data even before AI summarization exists. -crons.interval( +// Self-chains until the enriched backlog is drained, then hands off to merge, +// recluster and summarization once (not once per batch). +crons.cron( "cluster-enriched-articles", - { minutes: 40 }, + "30 0,6,12,18 * * *", internal.clustering.clusterEnrichedArticles, + {}, ); // --------------------------------------------------------------------------- -// Event Merge Pass — Every 20 minutes +// Event Merge Pass — 2x daily // --------------------------------------------------------------------------- // Collapses near-duplicate recently published events created across separate -// clustering runs. The action uses a DB-backed lease plus seed/top-K caps, so -// the tighter cadence should skip overlapping work instead of piling up load. -crons.interval( +// clustering runs. This was the single largest vector-search consumer after +// clustering itself (527 query-GB), and at a 20-minute cadence it mostly +// re-scanned events it had already compared. Twice daily, offset from the +// clustering windows, still catches duplicates created across separate runs. +crons.cron( "merge-near-duplicate-events", - { minutes: 20 }, + "45 1,13 * * *", internal.clustering.mergeNearDuplicateEvents, ); // --------------------------------------------------------------------------- -// Singleton Recluster Pass — Every 30 minutes +// Singleton Recluster Pass — Daily (03:45 UTC) // --------------------------------------------------------------------------- // Re-examines recent singleton / tiny events after more articles have landed, // improving recall for stories that were under-clustered during the online pass. // This cadence is paired with a DB-backed lease, no-candidate short-circuit, // seed caps, and reduced vector top-K so fallback recovery stays bounded. -crons.interval( +crons.cron( "recluster-recent-singletons", - { minutes: 30 }, + "45 3 * * *", internal.clustering.reclusterRecentSingletonEvents, ); // --------------------------------------------------------------------------- -// Stale Singleton Archive — Every 53 minutes (drifting) +// Stale Singleton Archive — Daily (02:20 UTC) // --------------------------------------------------------------------------- // Archives stale processing singletons so they stop inflating the vector index. // The job yields (skips) whenever a clustering job holds a pipeline lock to -// avoid concurrent mutation of hot event/embedding rows. Convex interval crons -// are epoch-phase-aligned, so an *hourly* cadence is an exact multiple of the -// 20-min merge and 30-min recluster cadences and fired in lockstep with them -// every single time — guaranteeing a blocking lock and a 100% skip rate. A -// 53-minute cadence is coprime with 20/30/40/60, so archive drifts across -// phases and regularly lands in quiet windows without ever starving the core -// clustering pipeline (which keeps priority). -crons.interval( +// avoid concurrent mutation of hot event/embedding rows. +// +// This previously ran every 53 minutes: interval crons are epoch-phase-aligned, +// so an hourly cadence was an exact multiple of the old 20/30/40-minute +// clustering cadences and fired in lockstep with them every time, guaranteeing +// a blocking lock and a 100% skip rate. 53 is coprime with 20/30/40/60, so the +// job drifted across phases and regularly landed in quiet windows. +// +// That reasoning is obsolete now that every clustering stage runs at explicit +// fixed times. 02:20 UTC simply sits in a gap between the 00:xx pipeline window +// and the 03:45 recluster pass, so the lock is free without needing to drift. +crons.cron( "archive-stale-singleton-events", - { minutes: 53 }, + "20 2 * * *", internal.singletonCleanup.archiveStaleSingletonEvents, {}, ); // --------------------------------------------------------------------------- -// Event Summarization — Hourly -// --------------------------------------------------------------------------- -// Generates GPT-backed perspective summaries for published events that have -// enough source diversity. Runs independently so clustering is never blocked on -// model latency or budget state. Cadence eased 45min → 1h: the action-compute -// win comes from the per-job body-fetch fix (bounded hold time), not from -// starving throughput, so capacity stays close to the original (batchSize=8 → -// ~192 jobs/day) to keep pace with clustering intake. Queue depth is watched by -// summary-queue-health. -crons.interval( +// Event Summarization — 4x daily (:45 past each window) +// --------------------------------------------------------------------------- +// Generates perspective summaries for published events that have enough source +// diversity. Runs 45 minutes into each window so clustering has already created +// the events this pass will summarize. +// +// This action's downstream job (processSummaryJob) was 92% of all Convex action +// compute. Cadence alone was not the problem — the per-job wall clock was — but +// a lower cadence also keeps us inside Gemini's free-tier rate limit, which is +// what was generating the 429 storm that wasted most of that compute. +// +// This supersedes the hourly cadence from #60, which deliberately preserved +// throughput ("the win comes from the per-job body-fetch fix, not from starving +// throughput"). That was not enough on its own — the deployment still exceeded +// the free plan — so throughput is now cut too. +crons.cron( "summarize-published-events", - { hours: 1 }, + "45 0,6,12,18 * * *", internal.summarizationNode.summarizeQueuedEvents, {}, ); // --------------------------------------------------------------------------- -// Summary Queue Health — Hourly +// Summary Queue Health — Daily (05:30 UTC) // --------------------------------------------------------------------------- // Warns in logs when queued jobs duplicate the same event or queue depth // grows enough to threaten coverage. -crons.interval( +crons.cron( "summary-queue-health", - { hours: 1 }, + "30 5 * * *", internal.summarizationNode.alertOnSummaryQueueHealth, {}, ); // --------------------------------------------------------------------------- -// Claim Divergence Detection — Every 45 minutes +// Claim Divergence Detection — DISABLED (cost mode) // --------------------------------------------------------------------------- // Builds the eventClaims graph from atomic facts so the product can show // agreements, conflicts, framing differences, and lean-specific exclusives. -crons.interval( - "detect-event-claims", - { minutes: 45 }, - internal.claimDivergenceNode.processStaleEventClaims, - {}, -); +// +// Claim analysis is already switched off in prod at the config layer +// (article_fact_extraction_enabled = false), so this cron was waking every 45 +// minutes only to read config, find nothing to do, and exit — pure billed +// compute and database I/O for zero product value. The cron is disabled to stop +// paying for that. Re-enable it together with the config flag, not before. +// crons.interval( +// "detect-event-claims", +// { minutes: 45 }, +// internal.claimDivergenceNode.processStaleEventClaims, +// {}, +// ); // --------------------------------------------------------------------------- -// Daily News Quiz — Daily +// Daily News Quiz — DISABLED (cost mode) // --------------------------------------------------------------------------- // Generates one globally shared UTC-dated quiz from grounded claim/fact data. -crons.daily( - "generate-daily-news-quiz", - { hourUTC: 6, minuteUTC: 0 }, - internal.quizNode.generateDailyQuiz, - {}, -); +// +// The quiz is already switched off in prod, and it is derived from claim/fact +// data that the disabled claim pipeline no longer produces — so this was a +// daily model-backed action producing nothing usable. Re-enable alongside +// claim analysis and the quiz feature flag. +// crons.daily( +// "generate-daily-news-quiz", +// { hourUTC: 6, minuteUTC: 0 }, +// internal.quizNode.generateDailyQuiz, +// {}, +// ); // --------------------------------------------------------------------------- // Article Bias Outlier Detection — Daily @@ -152,24 +197,24 @@ crons.daily( // ); // --------------------------------------------------------------------------- -// AI Budget Reservation Cleanup — Hourly +// AI Budget Reservation Cleanup — 2x daily // --------------------------------------------------------------------------- // Deletes expired budget reservations outside the OpenAI-call hot path. -crons.interval( +crons.cron( "cleanup-ai-budget-reservations", - { hours: 1 }, + "5 1,13 * * *", internal.aiBudget.cleanupExpiredAiBudgetReservations, {}, ); // --------------------------------------------------------------------------- -// Vector Search Reservation Cleanup — Hourly +// Vector Search Reservation Cleanup — 2x daily // --------------------------------------------------------------------------- // Releases expired vector-search reservations outside the reservation hot path // so every semantic lookup no longer scans stale reservations first. -crons.interval( +crons.cron( "cleanup-vector-search-reservations", - { hours: 1 }, + "10 1,13 * * *", internal.vectorSearchBudget.cleanupExpiredVectorSearchReservations, {}, ); @@ -198,14 +243,20 @@ crons.daily( ); // --------------------------------------------------------------------------- -// Pipeline Alert Checks — Every 20 minutes +// Pipeline Alert Checks — 2x daily // --------------------------------------------------------------------------- // Writes pipelineAlerts rows for persistent fallback mode, publish droughts, // stuck processing growth, vector-budget burn rate, job error rates, and absent // archive runs. Alerts stay in Convex and are surfaced in /admin/pipeline. -crons.interval( +// +// At a 20-minute cadence this was the 3rd largest database-I/O consumer in the +// app (1.68 GB, via pipeline.countProcessingEventsOlderThan) — admin-only +// telemetry costing real money 72 times a day. With the pipeline now running in +// 4 daily windows there is nothing to observe between windows anyway, so this +// runs twice daily, shortly after the 00:xx and 12:xx windows complete. +crons.cron( "check-pipeline-alerts", - { minutes: 20 }, + "50 1,13 * * *", internal.pipeline.checkPipelineAlerts, {}, ); @@ -265,39 +316,86 @@ crons.daily( ); // --------------------------------------------------------------------------- -// Pipeline Runtime Config Snapshot — Every 5 minutes +// Storage Retention (cost mode) — see STORAGE_RETENTION_DEFAULTS in retention.ts +// --------------------------------------------------------------------------- +// These are operational/storage-cost purges, distinct from the legal data +// minimization jobs above. Database storage was growing without bound (~1,300 +// articles/day, each with a 512-dimension embedding), which meant the bill +// compounded every month regardless of any other saving. These cap it. +// +// All three self-chain via the scheduler until their backlog is drained, so a +// daily trigger is enough. NOTE: the FIRST drain reads every row it deletes, so +// expect a one-off spike in database I/O the first day this ships. + +// Deletes vectors for articles past article_embedding_retention_days (45). +// Article rows are kept; only the embeddings go. Clustering's widest lookback +// is 48h, so a 45-day floor is ~22x more history than anything actually reads. +crons.daily( + "retention-purge-stale-article-embeddings", + { hourUTC: 4, minuteUTC: 5 }, + internal.retention.purgeStaleArticleEmbeddings, + {}, +); + +// Deletes articles archived as stale singletons past +// archived_article_retention_days (90). These belong to no event. +crons.daily( + "retention-purge-archived-articles", + { hourUTC: 4, minuteUTC: 15 }, + internal.retention.purgeArchivedDetachedArticles, + {}, +); + +// Full-table orphan sweep. WEEKLY ONLY (Sundays): unlike the jobs above this +// cannot use a head-of-index scan — an orphan can have any creation time — so +// it reads every embedding row, which is billed database I/O. The daily stale +// purge already collects every orphan older than 45 days for free, so this +// exists only to catch recent orphans from interrupted writes. +crons.cron( + "retention-purge-orphaned-article-embeddings", + "35 4 * * 0", + internal.retention.purgeOrphanedArticleEmbeddings, + {}, +); + +// --------------------------------------------------------------------------- +// Pipeline Runtime Config Snapshot — 5 min before each pipeline window // --------------------------------------------------------------------------- // Collapses the per-key clustering config reads into one compact document that // pipeline jobs read on every run. Without this the snapshot is never built and // jobs silently fall back to N per-key reads. -crons.interval( +crons.cron( "refresh-pipeline-runtime-config", - { hours: 1 }, + "55 23,5,11,17 * * *", internal.config.refreshPipelineRuntimeConfig, {}, ); // --------------------------------------------------------------------------- -// Anonymous Trending Feed Snapshot — Every 2 minutes +// Anonymous Trending Feed Snapshot — 4x daily, after each pipeline window // --------------------------------------------------------------------------- // Precomputes the trending first page so anonymous/cold loads skip the live // ranked scan. Rebuilt on a cron (not on every preview write) to avoid write // amplification and contention on the single snapshot document. -crons.interval( +// +// Rebuilt 45 minutes after each summarization pass so the snapshot reflects +// that window's freshly summarized events. There is no new content to surface +// between windows, so a tighter cadence would rewrite an identical document. +crons.cron( "rebuild-public-feed-snapshots", - { minutes: 20 }, + "30 1,7,13,19 * * *", internal.events.rebuildPublicFeedSnapshotsJob, {}, ); // --------------------------------------------------------------------------- -// Hot Vector Table Prune — Hourly +// Hot Vector Table Prune — Daily (02:40 UTC) // --------------------------------------------------------------------------- // Deletes eventEmbeddingHot rows for events that have gone quiet so the hot // clustering index stays small. Active events are re-added by the write path. -crons.interval( +crons.cron( "prune-hot-event-embeddings", - { hours: 1 }, + "40 2 * * *", internal.clustering.pruneHotEventEmbeddings, {}, ); diff --git a/packages/backend/convex/enrichmentNode.ts b/packages/backend/convex/enrichmentNode.ts index 746bed7..329e6c4 100644 --- a/packages/backend/convex/enrichmentNode.ts +++ b/packages/backend/convex/enrichmentNode.ts @@ -73,6 +73,13 @@ const EMBEDDING_VERSION = 4; /** How many article pages to fetch/extract in parallel inside one batch. */ const EXTRACTION_CONCURRENCY = 5; +// Self-chaining bounds for enrichUnprocessedArticles. 60 x BATCH_SIZE(40) = +// 2,400 articles per triggered drain, which covers a full day of intake +// (~1,300/day) in a single window with headroom, while still guaranteeing the +// chain terminates if the claim query ever misbehaves. +const MAX_ENRICHMENT_CHAIN_DEPTH = 60; +const ENRICHMENT_CHAIN_DELAY_MS = 5_000; + const ARTICLE_FACTS_JSON_SCHEMA = { name: "ArticleAtomicFacts", strict: true, @@ -1114,9 +1121,18 @@ async function runEnrichmentBatch( * Called by the cron job every 30 minutes. */ export const enrichUnprocessedArticles = internalAction({ - args: {}, + // `chainDepth` and `enrichedSoFar` are set only by this action rescheduling + // itself; the cron and all manual invocations start at 0. `enrichedSoFar` + // accumulates across the chain so the LAST link knows whether the drain as a + // whole produced anything — a terminal batch can legitimately enrich zero + // while earlier links enriched hundreds. See the self-chaining note below. + args: { + chainDepth: v.optional(v.number()), + enrichedSoFar: v.optional(v.number()), + }, handler: async ( ctx, + { chainDepth = 0, enrichedSoFar = 0 }, ): Promise<{ enriched: number; failed: number; @@ -1180,6 +1196,19 @@ export const enrichUnprocessedArticles = internalAction({ if (articles.length === 0) { console.log("[enrichment] No unprocessed articles to enrich"); + // This is a chain terminator, not just an idle run: the previous link may + // have claimed a full batch (and so chained) only for the backlog to be + // empty by the time this link ran. Without handing off here, a drain whose + // article count is an exact multiple of BATCH_SIZE would enrich everything + // and then never schedule clustering. + const shouldScheduleClustering = enrichedSoFar > 0; + if (shouldScheduleClustering) { + await ctx.scheduler.runAfter( + 90_000, + internal.clustering.clusterEnrichedArticles, + {}, + ); + } await ctx.runMutation(internal.pipeline.insertRunLog, { jobName: "enrichUnprocessedArticles", runId, @@ -1188,7 +1217,11 @@ export const enrichUnprocessedArticles = internalAction({ durationMs: Date.now() - startedAt, status: "ok", counters: { claimedArticles: 0, enrichedArticles: 0, failedArticles: 0 }, - gauges: { scheduledClustering: false }, + gauges: { + scheduledClustering: shouldScheduleClustering, + enrichedSoFar, + chainDepth, + }, metadata: {}, }); return { enriched: 0, failed: 0, skipped: false }; @@ -1196,7 +1229,37 @@ export const enrichUnprocessedArticles = internalAction({ try { const result = await runEnrichmentBatch(ctx, articles, runId); - if (result.enriched > 0) { + + // Self-chain until the backlog is drained. + // + // This action processes at most BATCH_SIZE articles and then stops. That + // was fine when the cron fired every 30 minutes (48 runs/day x 40 = ~1,920 + // articles/day, comfortably above the ~1,300/day intake), but the pipeline + // now runs in 4 batched windows per day for cost reasons — 4 x 40 = 160 + // articles/day, which would leave the backlog growing forever and starve + // clustering of input. + // + // Chaining instead of polling is also strictly cheaper: an idle deployment + // schedules nothing at all, while throughput still scales with real intake. + // A full batch means there are probably more articles waiting. + const likelyMoreWaiting = articles.length === BATCH_SIZE; + const shouldChain = + likelyMoreWaiting && chainDepth + 1 < MAX_ENRICHMENT_CHAIN_DEPTH; + const enrichedTotal = enrichedSoFar + result.enriched; + + if (shouldChain) { + await ctx.scheduler.runAfter( + ENRICHMENT_CHAIN_DELAY_MS, + internal.enrichmentNode.enrichUnprocessedArticles, + { chainDepth: chainDepth + 1, enrichedSoFar: enrichedTotal }, + ); + } + + // Only kick clustering once the chain has finished draining, so a long + // drain schedules one clustering pass instead of one per batch. The test + // is on the CHAIN total, not this batch: a terminal batch that enriched + // nothing must still hand off if earlier links in the chain did work. + if (enrichedTotal > 0 && !shouldChain) { await ctx.scheduler.runAfter( 90_000, internal.clustering.clusterEnrichedArticles, @@ -1224,7 +1287,9 @@ export const enrichUnprocessedArticles = internalAction({ tokensUsed: result.tokensUsed ?? 0, }, gauges: { - scheduledClustering: result.enriched > 0, + scheduledClustering: enrichedTotal > 0 && !shouldChain, + chainDepth, + chainedToNext: shouldChain, failureRatio: articles.length > 0 ? result.failed / articles.length : 0, }, diff --git a/packages/backend/convex/events.ts b/packages/backend/convex/events.ts index 9ebb37a..223b805 100644 --- a/packages/backend/convex/events.ts +++ b/packages/backend/convex/events.ts @@ -23,14 +23,39 @@ import { foldDiacriticsToAscii } from "./lib/romanian"; import { getConfig } from "./config"; import { EVENT_SHARE_ASSET_GENERATION_ENABLED_KEY } from "./shareAssets"; -// Ranked-feed candidate scan caps. getPublishedEvents is the single largest -// database-I/O consumer: the live ranked path reads this many full ~4KB -// preview docs per uncached load just to rank and return one page. Trimmed -// (250 -> 150 trending, 500 -> 200 topic) to cut read bytes on every trending -// page-2+ load and every topic feed, at a small cost to how deep the ranking -// pool reaches. -const TRENDING_SCAN_LIMIT = 150; -const TOPIC_SCAN_LIMIT = 200; +// COST: ranked feed pagination is the app's single largest source of database +// I/O, so it is deliberately cursor-anchored rather than "scan the top N and +// slice in JS". +// +// The old shape re-scanned the top TRENDING_SCAN_LIMIT (250) preview rows for +// *every* page — including page 5 — and then threw ~95% of them away. Two +// separate costs came out of that: +// 1. bytes read per execution (250 fat preview docs ≈ 0.6 MB for a 6-item +// page), and +// 2. reactive re-execution: a query that reads the top of `by_trending_score` +// is invalidated by every newly-published event (recency dominates the +// score, so new events always land at the top of that range). Every open +// page-2+ subscription therefore re-ran, and re-read its 250 rows, on +// roughly every publish. +// +// Anchoring each page to an index range that is bounded *above* by the previous +// page's score fixes both: a page reads ~pageSize rows, and new high-scoring +// events fall outside the read range so they no longer invalidate deep pages. +const RANKED_PAGE_BUFFER = 12; +// A run of rows tying on the index key (identical trendingScore, or identical +// lastUpdatedAt) can be longer than RANKED_PAGE_BUFFER. When that happens every +// row in the window sorts at or before the cursor and the page comes back +// empty, which would look identical to "the feed ended". Widen the window +// geometrically instead. 4 attempts covers a tie run of ~1,000 rows; the retry +// only runs in that rare case, so the normal path still reads exactly once. +const RANKED_TIE_RUN_MAX_ATTEMPTS = 4; +const RANKED_TIE_RUN_GROWTH = 4; +const RANKED_MAX_PAGE_SIZE = 50; +// Ranked pagination depth cap. Preserves the old "the ranked feed ends" UX +// (previously an implicit side effect of the 250/500-row scan window) so +// infinite scroll cannot walk the whole table one cheap page at a time. +const RANKED_MAX_DEPTH = 240; +const RANKED_DEPTH_MARKER = "|d"; const FEED_SORT_VALIDATOR = v.union(v.literal("recent"), v.literal("trending")); @@ -40,102 +65,193 @@ function sortEventsForFeed(events: PublicPreviewRow[], sort: FeedSort) { }); } -function paginateRankedEvents( - events: PublicPreviewRow[], - cursor: string | null, - targetSize: number, - sort: FeedSort, -) { - const resumePayload = decodeRankedCursor(cursor); - const resumeIndex = resumePayload - ? events.findIndex((event) => event.eventId === resumePayload.eventId) - : -1; - const startIndex = - resumeIndex >= 0 - ? resumeIndex + 1 - : resumePayload - ? events.findIndex( - (event) => - compareRankedPayload(rankedPayload(event, sort), resumePayload) > - 0, - ) - : 0; - const normalizedStartIndex = startIndex >= 0 ? startIndex : events.length; - const page = events.slice( - normalizedStartIndex, - normalizedStartIndex + targetSize, - ); - const isDone = normalizedStartIndex + page.length >= events.length; - const lastReturned = page[page.length - 1]; - +// `encodeRankedCursor` percent-encodes its JSON payload, so "|" can never occur +// inside a base cursor and is safe as a depth separator. Depth travels on the +// cursor so the server stays stateless; a cursor without a marker (e.g. one +// minted by the snapshot builder) simply starts at depth 0. +function splitRankedCursor(cursor: string | null) { + if (!cursor) return { base: null as string | null, depth: 0 }; + const markerIndex = cursor.lastIndexOf(RANKED_DEPTH_MARKER); + if (markerIndex < 0) return { base: cursor, depth: 0 }; + const parsed = Number(cursor.slice(markerIndex + RANKED_DEPTH_MARKER.length)); return { - page, - isDone, - continueCursor: - isDone || !lastReturned ? "" : encodeRankedCursor(lastReturned, sort), + base: cursor.slice(0, markerIndex), + depth: Number.isFinite(parsed) && parsed > 0 ? Math.floor(parsed) : 0, }; } +function withDepth(cursor: string, depth: number) { + return cursor === "" ? "" : `${cursor}${RANKED_DEPTH_MARKER}${depth}`; +} + function snapshotKey(sort: FeedSort, topicId?: Id<"topics">) { + // Topic-scoped snapshots come from #60; the cursor-anchored reader below + // consumes them the same way it consumes the global one. return topicId ? `anonymous:first-page:${sort}:topic:${topicId}` : `anonymous:first-page:${sort}`; } -async function getTopicFeedCandidates( +type RankedWindow = { + rows: PublicPreviewRow[]; + /** The index range was exhausted, so there is nothing after this window. */ + reachedEnd: boolean; +}; + +/** + * Reads at most `limit` preview rows starting at `fromScore` (inclusive) and + * walking down the ranking. `fromScore === undefined` starts at the top. + */ +async function readRankedWindow( ctx: QueryCtx, - topicId: Id<"topics">, + topicId: Id<"topics"> | undefined, sort: FeedSort, - limit: number = TOPIC_SCAN_LIMIT, -) { + fromScore: number | undefined, + limit: number, +): Promise { + if (topicId) { + const table = ctx.db.query("publicEventPreviewTopics"); + const rows = + sort === "trending" + ? await (fromScore === undefined + ? table.withIndex("by_topic_trending", (q) => + q.eq("topicId", topicId), + ) + : table.withIndex("by_topic_trending", (q) => + q.eq("topicId", topicId).lte("trendingScore", fromScore), + ) + ) + .order("desc") + .take(limit) + : await (fromScore === undefined + ? table.withIndex("by_topic_updated", (q) => + q.eq("topicId", topicId), + ) + : table.withIndex("by_topic_updated", (q) => + q.eq("topicId", topicId).lte("lastUpdatedAt", fromScore), + ) + ) + .order("desc") + .take(limit); + const previews = await Promise.all( + rows.map((row) => ctx.db.get(row.previewId)), + ); + return { + rows: previews.filter( + (preview): preview is Doc<"publicEventPreviews"> => preview !== null, + ), + reachedEnd: rows.length < limit, + }; + } + + const table = ctx.db.query("publicEventPreviews"); const rows = sort === "trending" - ? await ctx.db - .query("publicEventPreviewTopics") - .withIndex("by_topic_trending", (q) => q.eq("topicId", topicId)) + ? await (fromScore === undefined + ? table.withIndex("by_trending_score") + : table.withIndex("by_trending_score", (q) => + q.lte("trendingScore", fromScore), + ) + ) .order("desc") .take(limit) - : await ctx.db - .query("publicEventPreviewTopics") - .withIndex("by_topic_updated", (q) => q.eq("topicId", topicId)) + : await (fromScore === undefined + ? table.withIndex("by_last_updated_at") + : table.withIndex("by_last_updated_at", (q) => + q.lte("lastUpdatedAt", fromScore), + ) + ) .order("desc") .take(limit); - const previews = await Promise.all(rows.map((row) => ctx.db.get(row.previewId))); - return previews.filter( - (preview): preview is Doc<"publicEventPreviews"> => preview !== null, - ); + return { rows, reachedEnd: rows.length < limit }; } -async function getFeedCandidates( +async function paginateRanked( ctx: QueryCtx, + topicId: Id<"topics"> | undefined, sort: FeedSort, - scanLimit: number, + cursor: string | null, + requestedSize: number, ) { - if (sort === "trending") { - return await ctx.db - .query("publicEventPreviews") - .withIndex("by_trending_score") - .order("desc") - .take(scanLimit); + const targetSize = Math.min(Math.max(requestedSize, 1), RANKED_MAX_PAGE_SIZE); + const { base, depth } = splitRankedCursor(cursor); + const resumePayload = decodeRankedCursor(base); + + const remainingDepth = Math.max(0, RANKED_MAX_DEPTH - depth); + if (remainingDepth === 0) { + return { page: [] as PublicPreviewRow[], isDone: true, continueCursor: "" }; } + const pageSize = Math.min(targetSize, remainingDepth); + + // Read only one page plus a small buffer. The buffer absorbs rows that tie on + // the index key with the cursor (they sort before it under the full + // comparator and get dropped below). + const rowsAfterCursor = (windowRows: PublicPreviewRow[]) => { + const sorted = sortEventsForFeed(windowRows, sort); + return resumePayload + ? sorted.filter( + (event) => + compareRankedPayload(rankedPayload(event, sort), resumePayload) > 0, + ) + : sorted; + }; + + let limit = pageSize + RANKED_PAGE_BUFFER; + let window = await readRankedWindow( + ctx, + topicId, + sort, + resumePayload?.score, + limit, + ); + let after = rowsAfterCursor(window.rows); + + // An empty result with more rows still available means the window fell + // entirely inside a tie run, not that the feed ended. Widen and retry. + for ( + let attempt = 1; + after.length === 0 && + !window.reachedEnd && + attempt < RANKED_TIE_RUN_MAX_ATTEMPTS; + attempt++ + ) { + limit *= RANKED_TIE_RUN_GROWTH; + window = await readRankedWindow( + ctx, + topicId, + sort, + resumePayload?.score, + limit, + ); + after = rowsAfterCursor(window.rows); + } + + const reachedEnd = window.reachedEnd; + const page = after.slice(0, pageSize); + const lastReturned = page[page.length - 1]; + const nextDepth = depth + page.length; + const isDone = + (reachedEnd && after.length <= pageSize) || + nextDepth >= RANKED_MAX_DEPTH || + !lastReturned; - return await ctx.db - .query("publicEventPreviews") - .withIndex("by_last_updated_at") - .order("desc") - .take(scanLimit); + return { + page, + isDone, + continueCursor: isDone + ? "" + : withDepth(encodeRankedCursor(lastReturned, sort), nextDepth), + }; } -async function getRankedFeedCandidates( +async function getTopicFeedCandidates( ctx: QueryCtx, - topicId: Id<"topics"> | undefined, + topicId: Id<"topics">, sort: FeedSort, + limit: number, ) { - const scanLimit = topicId ? TOPIC_SCAN_LIMIT : TRENDING_SCAN_LIMIT; - if (topicId) { - return await getTopicFeedCandidates(ctx, topicId, sort); - } - return await getFeedCandidates(ctx, sort, scanLimit); + const { rows } = await readRankedWindow(ctx, topicId, sort, undefined, limit); + return rows; } export const getPublishedEvents = query({ @@ -146,16 +262,27 @@ export const getPublishedEvents = query({ }, handler: async (ctx, args) => { const sort = args.sort ?? "trending"; + const numItems = Math.min( + Math.max(args.paginationOpts.numItems, 1), + RANKED_MAX_PAGE_SIZE, + ); - let events; - - // Anonymous trending first-page acceleration. The trending feed (global and - // per-topic) otherwise requires an expensive ranked scan on every cold load. - // Serve the cached snapshot for the first page, then hand pagination back to - // the live ranked query via the stored `ranked:` cursor so the feed never - // dead-ends at the snapshot size. (Recent is a cheap indexed pagination and - // is not cached.) - if (sort === "trending" && args.paginationOpts.cursor === null) { + // Anonymous trending acceleration. The snapshot document holds the whole + // precomputed first slice of the trending ranking plus the `ranked:` cursor + // that follows each item, so it can serve *any* page whose cursor it still + // recognises — not just page 1. With the default 6-item page size that is + // the first four pages of the feed served from a single document read + // instead of four ranked index scans, and — critically — that document only + // changes when the rebuild cron runs, so the subscription re-runs on that + // cadence instead of on every publish. + // + // Topic-scoped snapshots (#60) are keyed separately and read through this + // same path; when one is absent the live cursor-anchored query below is + // authoritative, so a missing snapshot only costs an index range read. + if (sort === "trending") { + const { base: snapshotCursor, depth: cursorDepth } = splitRankedCursor( + args.paginationOpts.cursor, + ); const snapshot = await ctx.db .query("publicFeedSnapshots") .withIndex("by_key", (q) => @@ -169,19 +296,29 @@ export const getPublishedEvents = query({ cursors: string[]; }; const items = parsed.items ?? []; - if (items.length === 0) { - return { page: [], isDone: true, continueCursor: "" }; + const cursors = parsed.cursors ?? []; + // Resolve where this request resumes inside the snapshot. A cursor the + // snapshot does not recognise (stale generation) or one past the + // snapshot tail yields startIndex < 0 / >= items.length and falls + // through to the live ranked path below, which is authoritative. + const resumeIndex = + snapshotCursor === null ? -1 : cursors.indexOf(snapshotCursor); + const startIndex = + snapshotCursor === null ? 0 : resumeIndex < 0 ? -1 : resumeIndex + 1; + if (startIndex >= 0 && startIndex < items.length) { + const page = items.slice(startIndex, startIndex + numItems); + const boundaryIndex = startIndex + page.length - 1; + // An absent boundary cursor means the snapshot cannot hand off to + // the live ranked path, so the feed ends here. + const rawCursor = cursors[boundaryIndex] ?? ""; + const nextDepth = cursorDepth + page.length; + const isDone = rawCursor === "" || nextDepth >= RANKED_MAX_DEPTH; + return { + page, + isDone, + continueCursor: isDone ? "" : withDepth(rawCursor, nextDepth), + }; } - const page = items.slice(0, args.paginationOpts.numItems); - const boundaryIndex = - Math.min(page.length, parsed.cursors.length) - 1; - const continueCursor = - boundaryIndex >= 0 ? (parsed.cursors[boundaryIndex] ?? "") : ""; - return { - page, - isDone: continueCursor === "", - continueCursor, - }; } catch (error) { console.error( "[events] Failed to parse public feed snapshot:", @@ -193,36 +330,30 @@ export const getPublishedEvents = query({ } if (args.topicId || sort === "trending") { - const targetSize = args.paginationOpts.numItems; - const publishedMatches = await getRankedFeedCandidates( + const events = await paginateRanked( ctx, args.topicId, sort, - ); - const sortedMatches = sortEventsForFeed(publishedMatches, sort); - - events = paginateRankedEvents( - sortedMatches, args.paginationOpts.cursor, - targetSize, - sort, + numItems, ); - } else { - // Defensive reset in case a ranked cursor is reused after ranked mode is - // cleared. - const paginationOpts = args.paginationOpts.cursor?.startsWith( - RANKED_CURSOR_PREFIX, - ) - ? { ...args.paginationOpts, cursor: null } - : args.paginationOpts; - - events = await ctx.db - .query("publicEventPreviews") - .withIndex("by_last_updated_at") - .order("desc") - .paginate(paginationOpts); + return { ...events, page: events.page.map(toFeedEvent) }; } + // Defensive reset in case a ranked cursor is reused after ranked mode is + // cleared. + const paginationOpts = args.paginationOpts.cursor?.startsWith( + RANKED_CURSOR_PREFIX, + ) + ? { ...args.paginationOpts, cursor: null, numItems } + : { ...args.paginationOpts, numItems }; + + const events = await ctx.db + .query("publicEventPreviews") + .withIndex("by_last_updated_at") + .order("desc") + .paginate(paginationOpts); + return { ...events, page: events.page.map(toFeedEvent), @@ -359,9 +490,12 @@ export const getSitemapPublishedEvents = query({ limit: v.optional(v.number()), }, handler: async (ctx, args) => { + // COST: this is a public query with no caller left (the sitemap is served + // from `publicSitemapSnapshots`), so an unbounded default was a free way for + // anyone holding the deployment URL to read tens of MB of preview rows. const safeLimit = Math.min( - Math.max(Math.floor(args.limit ?? 5000), 1), - 10000, + Math.max(Math.floor(args.limit ?? 2000), 1), + 2000, ); const events = await ctx.db .query("publicEventPreviews") @@ -528,6 +662,8 @@ export const rescorePublicPreviews = internalMutation({ }, }); +const EVENT_PAGE_ARTICLE_LIMIT = 60; + export const getEventBySlug = query({ args: { slug: v.string() }, handler: async (ctx, args) => { @@ -549,10 +685,16 @@ export const getEventBySlug = query({ .collect(); const topicIds = eventTopicRows.map((r) => r.topicId); + // COST: article docs are the fattest rows in the schema (summary, atomic + // facts, entities, bias components), and this runs on every event page view + // and every crawler hit. A merged mega-event can accumulate well over a + // hundred articles; the page only ever renders a coverage list, so cap the + // read. Events at the cap show their most-recently-ingested coverage. const articles = await ctx.db .query("articles") .withIndex("by_event", (q) => q.eq("eventId", event._id)) - .collect(); + .order("desc") + .take(EVENT_PAGE_ARTICLE_LIMIT); // Custom social preview images are gated by a single kill switch. While it // is off (dev and prod), never serve a generated share asset — even if the // DB still holds "ready" rows from a previous run — so the OG/share image diff --git a/packages/backend/convex/feedPagination.test.ts b/packages/backend/convex/feedPagination.test.ts new file mode 100644 index 0000000..339bff1 --- /dev/null +++ b/packages/backend/convex/feedPagination.test.ts @@ -0,0 +1,264 @@ +import { convexTest } from "convex-test"; +import { describe, expect, test } from "vitest"; + +import { computeTrendingScore } from "./lib/publicEventPreviews"; +import schema from "./schema"; +import { api, internal } from "./_generated/api"; + +// See interactions.test.ts for the glob rationale (drops convex.config.ts, +// *.test.ts and *.d.ts so the Better Auth component never instantiates). +const modules = ( + import.meta as unknown as { + glob: (pattern: string) => Record Promise>; + } +).glob("./**/!(*.*.*)*.*s"); + +const HOUR_MS = 3_600_000; + +async function seed(count: number) { + const t = convexTest(schema, modules); + const now = Date.now(); + await t.run(async (ctx) => { + for (let i = 0; i < count; i++) { + const title = `event ${String(i).padStart(3, "0")}`; + const lastUpdatedAt = now - i * HOUR_MS; + const eventId = await ctx.db.insert("events", { + title, + slug: title.replace(/\s+/g, "-"), + status: "published", + firstPublishedAt: now - 200 * HOUR_MS, + } as never); + await ctx.db.insert("publicEventPreviews", { + eventId, + slug: title.replace(/\s+/g, "-"), + title, + firstPublishedAt: now - 200 * HOUR_MS, + lastUpdatedAt, + articleCount: 1, + sourceCount: 1, + topicIds: [], + trendingScore: computeTrendingScore({ + sourceCount: 1, + articleCount: 1, + lastUpdatedAt, + firstPublishedAt: now - 200 * HOUR_MS, + }), + sourceBiasCounts: { left: 0, center: 0, right: 0 }, + sources: [], + updatedAt: now, + }); + } + }); + return t; +} + +async function walk( + t: Awaited>, + args: Record, + pageSize: number, + maxPages = 100, +) { + const titles: string[] = []; + let cursor: string | null = null; + let pages = 0; + for (;;) { + const res: { + page: Array<{ title: string }>; + isDone: boolean; + continueCursor: string; + } = await t.query(api.events.getPublishedEvents, { + ...args, + paginationOpts: { numItems: pageSize, cursor }, + }); + titles.push(...res.page.map((e) => e.title)); + pages++; + if (res.isDone) break; + // Fail loudly rather than returning a truncated result: a walk that runs + // away is a bug, and silently capping it would let the completeness + // assertions below pass on partial data. + expect(pages, "pagination did not terminate").toBeLessThan(maxPages); + cursor = res.continueCursor; + expect(cursor).not.toBe(""); + } + return { titles, pages }; +} + +/** + * Seed `count` previews that all share one `lastUpdatedAt`, and therefore one + * `trendingScore`. Ties are the pathological case for cursor-anchored ranking: + * every row sits at the same index key, so the cursor cannot separate them by + * score alone. + */ +async function seedTied(count: number) { + const t = convexTest(schema, modules); + const now = Date.now(); + const lastUpdatedAt = now - HOUR_MS; + const firstPublishedAt = now - 200 * HOUR_MS; + await t.run(async (ctx) => { + for (let i = 0; i < count; i++) { + const title = `tied ${String(i).padStart(3, "0")}`; + const slug = title.replace(/\s+/g, "-"); + const eventId = await ctx.db.insert("events", { + title, + slug, + status: "published", + firstPublishedAt, + } as never); + await ctx.db.insert("publicEventPreviews", { + eventId, + slug, + title, + firstPublishedAt, + lastUpdatedAt, + articleCount: 1, + sourceCount: 1, + topicIds: [], + trendingScore: computeTrendingScore({ + sourceCount: 1, + articleCount: 1, + lastUpdatedAt, + firstPublishedAt, + }), + sourceBiasCounts: { left: 0, center: 0, right: 0 }, + sources: [], + updatedAt: now, + }); + } + }); + return t; +} + +// COST-MODE regression guard. `events.getPublishedEvents` was rewritten from +// "scan the top 250 preview rows on every page and slice in JS" to cursor- +// anchored index ranges, with the anonymous trending snapshot serving several +// pages instead of just the first. Those two changes cut the app's single +// largest source of database I/O, but they move real ordering/termination logic +// onto the cursor — so pin the observable contract: a full walk must return +// every event exactly once, in ranked order, and must terminate. +describe("ranked pagination (cursor-anchored)", () => { + test("trending walk yields every event exactly once, in score order", async () => { + const t = await seed(40); + const { titles } = await walk(t, { sort: "trending" }, 6); + expect(titles.length).toBe(40); + expect(new Set(titles).size).toBe(40); + expect(titles).toEqual([...titles].sort()); + }); + + // Regression: a run of rows tying on the index key that is longer than + // RANKED_PAGE_BUFFER filled an entire window with rows sorting at or before + // the cursor. The page came back empty, which was indistinguishable from "the + // feed ended" — so the feed silently truncated mid-list and the reader simply + // stopped seeing events. 40 tied rows is well past the 12-row buffer. + test("tie run longer than the page buffer does not truncate the feed", async () => { + const t = await seedTied(40); + const { titles } = await walk(t, { sort: "trending" }, 6); + expect(titles.length).toBe(40); + expect(new Set(titles).size).toBe(40); + }); + + test("tie run is walkable on the recent sort too", async () => { + const t = await seedTied(30); + const { titles } = await walk(t, { sort: "recent" }, 6); + expect(titles.length).toBe(30); + expect(new Set(titles).size).toBe(30); + }); + + test("depth cap stops the ranked feed", async () => { + const t = await seed(300); + const { titles } = await walk(t, { sort: "trending" }, 10); + expect(titles.length).toBe(240); + expect(new Set(titles).size).toBe(240); + }); + + test("snapshot serves multiple pages then hands off to the live path", async () => { + const t = await seed(60); + await t.mutation(internal.events.rebuildPublicFeedSnapshotsJob, {}); + const { titles } = await walk(t, { sort: "trending" }, 6); + expect(titles.length).toBe(60); + expect(new Set(titles).size).toBe(60); + expect(titles).toEqual([...titles].sort()); + }); + + test("topic-filtered trending walk is complete and ordered", async () => { + const t = convexTest(schema, modules); + const now = Date.now(); + const topicId = await t.run(async (ctx) => { + const tid = await ctx.db.insert("topics", { + slug: "politica", + displayName: "Politica", + } as never); + for (let i = 0; i < 30; i++) { + const title = `t ${String(i).padStart(3, "0")}`; + const lastUpdatedAt = now - i * HOUR_MS; + const eventId = await ctx.db.insert("events", { + title, + slug: title.replace(/\s+/g, "-"), + status: "published", + firstPublishedAt: now - 200 * HOUR_MS, + } as never); + const trendingScore = computeTrendingScore({ + sourceCount: 1, + articleCount: 1, + lastUpdatedAt, + firstPublishedAt: now - 200 * HOUR_MS, + }); + const previewId = await ctx.db.insert("publicEventPreviews", { + eventId, + slug: title.replace(/\s+/g, "-"), + title, + firstPublishedAt: now - 200 * HOUR_MS, + lastUpdatedAt, + articleCount: 1, + sourceCount: 1, + topicIds: [tid], + trendingScore, + sourceBiasCounts: { left: 0, center: 0, right: 0 }, + sources: [], + updatedAt: now, + }); + await ctx.db.insert("publicEventPreviewTopics", { + topicId: tid, + eventId, + previewId, + lastUpdatedAt, + firstPublishedAt: now - 200 * HOUR_MS, + trendingScore, + updatedAt: now, + }); + } + return tid; + }); + const { titles } = await walk(t, { sort: "trending", topicId }, 7); + expect(titles.length).toBe(30); + expect(new Set(titles).size).toBe(30); + expect(titles).toEqual([...titles].sort()); + }); + + test("stale snapshot cursor falls through to the live path", async () => { + const t = await seed(40); + await t.mutation(internal.events.rebuildPublicFeedSnapshotsJob, {}); + const first: { continueCursor: string } = await t.query( + api.events.getPublishedEvents, + { sort: "trending", paginationOpts: { numItems: 6, cursor: null } }, + ); + // Simulate a client holding a cursor from a snapshot generation the server + // no longer has. + await t.run(async (ctx) => { + const snap = await ctx.db.query("publicFeedSnapshots").first(); + if (snap) { + await ctx.db.patch(snap._id, { + payloadJson: JSON.stringify({ items: [], cursors: [] }), + }); + } + }); + const second: { page: Array<{ title: string }> } = await t.query( + api.events.getPublishedEvents, + { + sort: "trending", + paginationOpts: { numItems: 6, cursor: first.continueCursor }, + }, + ); + expect(second.page.length).toBe(6); + expect(second.page[0].title).toBe("event 006"); + }); +}); diff --git a/packages/backend/convex/lib/aiCall.ts b/packages/backend/convex/lib/aiCall.ts index 7affa8c..5946b3d 100644 --- a/packages/backend/convex/lib/aiCall.ts +++ b/packages/backend/convex/lib/aiCall.ts @@ -103,6 +103,32 @@ function isRetryableError(error: unknown): boolean { ); } +/** + * Provider rate/quota rejection (HTTP 429 / RESOURCE_EXHAUSTED). Matches both + * a live SDK error object (which carries `status`) and the flattened message + * string `callLLM` returns to its callers (e.g. "429 status code (no body)"), + * because by the time an action's catch block sees it the status field is gone. + * + * Callers should treat this as *backpressure* — defer the work — rather than a + * failure that burns a retry attempt: the request never reached the model, so + * nothing about the input is wrong. + * + * Additive helper: `isRetryableError` and `callLLM` behaviour are unchanged. + */ +export function isRateLimitError(error: unknown): boolean { + if (errorStatus(error) === 429) return true; + const message = + error instanceof Error + ? error.message + : typeof error === "string" + ? error + : ""; + return ( + message.length > 0 && + /\b429\b|RESOURCE_EXHAUSTED|rate[ _-]?limit|quota/i.test(message) + ); +} + function isFatalError(error: unknown): boolean { const status = errorStatus(error); return status === 401 || status === 403 || status === 404; diff --git a/packages/backend/convex/lib/feedSerialization.ts b/packages/backend/convex/lib/feedSerialization.ts index 7bac232..d4c5c71 100644 --- a/packages/backend/convex/lib/feedSerialization.ts +++ b/packages/backend/convex/lib/feedSerialization.ts @@ -12,6 +12,7 @@ export const RANKED_CURSOR_PREFIX = "ranked:"; export type PublicPreviewRow = Pick< Doc<"publicEventPreviews">, + | "_id" | "eventId" | "slug" | "title" @@ -36,6 +37,11 @@ export type PublicPreviewRow = Pick< export type RankedCursorPayload = { eventId: Id<"events">; + // The preview row's own id. This is the final tiebreak, and it must be the + // PREVIEW id rather than the event id: ranked reads walk an index on + // publicEventPreviews, whose implicit last key is that table's `_id`. Optional + // so cursors encoded before this field existed still decode. + previewId?: string; score: number; updatedAt: number; firstPublishedAt: number; @@ -47,6 +53,7 @@ export function rankedPayload( ): RankedCursorPayload { return { eventId: event.eventId, + previewId: String(event._id), score: sort === "trending" ? event.trendingScore : event.lastUpdatedAt, updatedAt: event.lastUpdatedAt, firstPublishedAt: event.firstPublishedAt, @@ -57,11 +64,32 @@ export function compareRankedPayload( a: RankedCursorPayload, b: RankedCursorPayload, ): number { + // EVERY key here descends, matching the `.order("desc")` index traversal that + // ranked reads use. That correspondence is load-bearing, not cosmetic: the + // cursor walk assumes each window is a contiguous prefix of this total order. + // + // The old final tiebreak was ASCENDING eventId, which inverted the scan + // direction inside a run of rows sharing a score. The window then held a + // biased sample of the tie run (the rows the descending scan reached first), + // the page returned that sample's tail, and the cursor advanced past every + // remaining tied row — so the feed silently dead-ended mid-list. Widening the + // window could not help, because widening reaches rows that sort BEFORE the + // cursor under an inverted tiebreak. + const previewIdOrder = + a.previewId === undefined || b.previewId === undefined + ? 0 + : a.previewId < b.previewId + ? 1 + : a.previewId > b.previewId + ? -1 + : 0; return ( b.score - a.score || b.updatedAt - a.updatedAt || b.firstPublishedAt - a.firstPublishedAt || - String(a.eventId).localeCompare(String(b.eventId)) + previewIdOrder || + // Legacy cursors carry no previewId; fall back so they still order stably. + String(b.eventId).localeCompare(String(a.eventId)) ); } diff --git a/packages/backend/convex/migrations.ts b/packages/backend/convex/migrations.ts index bd60382..37d104b 100644 --- a/packages/backend/convex/migrations.ts +++ b/packages/backend/convex/migrations.ts @@ -24,6 +24,7 @@ import { } from "./lib/userProfile"; import { deleteByEventIndex, EVENT_CHILD_TABLES } from "./singletonCleanup"; import { truncateThirdPartySnippet } from "./lib/compliance"; +import { isRateLimitError } from "./lib/aiCall"; import { syncPublicEventPreview } from "./lib/publicEventPreviews"; const MAX_FACT_EXTRACTION_ATTEMPTS = 3; @@ -1176,3 +1177,255 @@ export const backfillPreviewSearchText = mutation({ }; }, }); + +/** + * COST MODE — force the stored config rows that override the code defaults. + * + * `config.getBatch` resolves a key from the `config` table and only falls back + * to the seeded default when the row is ABSENT. Prod already has rows for most + * of the expensive knobs, so editing the defaults in config.ts is not enough on + * its own — those rows have to be rewritten. This migration does that. + * + * (`event_summary_body_fetch_enabled` is the exception: it has no row in prod, + * so the new `false` default takes effect on deploy without this migration. + * It is written here anyway so the state is explicit and identical everywhere.) + * + * Idempotent. Run dry first: + * npx convex run --prod migrations:applyCostReductionConfig '{"dryRun":true}' + * npx convex run --prod migrations:applyCostReductionConfig '{"dryRun":false}' + * + * Afterwards refresh the pipeline snapshot so running jobs pick the values up: + * npx convex run --prod config:refreshPipelineRuntimeConfig '{}' + */ +export const applyCostReductionConfig = internalMutation({ + args: { + dryRun: v.optional(v.boolean()), + }, + handler: async (ctx, args) => { + const dryRun = args.dryRun ?? true; + + // Value encoding must match config.setInternal / config.getBatch: the + // `value` column holds a JSON-encoded string, not the raw scalar. + const targets: Array<{ key: string; value: unknown; why: string }> = [ + { + key: "event_summary_body_fetch_enabled", + value: false, + why: "up to 12 billed network fetches per summary job — the single most expensive operation in the app", + }, + { + key: "event_summary_batch_size", + value: 12, + why: "summaries gate publishing, so throughput must keep pace with eligible events (~48/day at 4 runs/day) while staying inside Gemini free-tier RPM", + }, + { + key: "event_summary_enqueue_limit", + value: 40, + why: "unchanged from the previous default — the 2.15 GB I/O was fixed by cadence (32 runs/day to 4), and lowering depth would starve the publish queue", + }, + { + key: "event_summary_max_input_articles", + value: 6, + why: "prompt size drives data egress and model latency, both billed", + }, + { + key: "pipeline_alert_check_interval_minutes", + value: 720, + why: "must track the check-pipeline-alerts cron (now 2x daily) or absent-run alerts fire spuriously", + }, + { + key: "article_embedding_retention_days", + value: 45, + why: "caps unbounded articleEmbeddings storage growth; clustering's widest lookback is 48h", + }, + { + key: "archived_article_retention_days", + value: 90, + why: "deletes articles archived as stale singletons, which belong to no event", + }, + ]; + + const changed: Array<{ key: string; from: string | null; to: string }> = []; + const unchanged: string[] = []; + + for (const target of targets) { + const encoded = JSON.stringify(target.value); + const row = await ctx.db + .query("config") + .withIndex("by_key", (q) => q.eq("key", target.key)) + .unique(); + + if (row && row.value === encoded) { + unchanged.push(target.key); + continue; + } + + changed.push({ + key: target.key, + from: row ? row.value : null, + to: encoded, + }); + + if (dryRun) continue; + + if (row) { + await ctx.db.patch(row._id, { + value: encoded, + updatedAt: Date.now(), + }); + } else { + await ctx.db.insert("config", { + key: target.key, + value: encoded, + description: target.why, + updatedAt: Date.now(), + }); + } + } + + return { dryRun, changedCount: changed.length, changed, unchanged }; + }, +}); + +/** + * COST MODE — revive summary jobs that were killed by Gemini rate limits. + * + * Before the backpressure fix, a 429 called `markSummaryJobFailed`, consuming + * one of only 3 attempts. In prod this left ~395 jobs parked at attempts=3 with + * a far-future `nextAttemptAt`, permanently dead — the direct cause of the + * "only ~31% of events ever get summarized" problem. Rate limiting is + * backpressure, not a failure of the job, so those jobs deserve another run. + * + * Only revives jobs whose recorded error looks like a rate limit. Jobs that + * failed for real reasons (blocked_ungrounded, blocked_verbatim, empty + * response) are LEFT ALONE — re-running those would just burn budget again. + * + * Idempotent, and bounded so a single call cannot blow the transaction limit. + * Run repeatedly until `remaining` is 0: + * npx convex run --prod migrations:requeueRateLimitedSummaryJobs '{"dryRun":true}' + * npx convex run --prod migrations:requeueRateLimitedSummaryJobs '{"dryRun":false}' + */ +export const requeueRateLimitedSummaryJobs = internalMutation({ + args: { + dryRun: v.optional(v.boolean()), + limit: v.optional(v.number()), + }, + handler: async (ctx, args) => { + const dryRun = args.dryRun ?? true; + const limit = Math.min(Math.max(args.limit ?? 200, 1), 500); + + const failed = await ctx.db + .query("eventSummaryJobs") + .withIndex("by_status_updatedAt", (q) => q.eq("status", "failed")) + .order("desc") + .take(1000); + + // Share the runtime's rate-limit detection so this migration classifies + // jobs exactly the way the summarization defer path does — a divergence + // here would revive the wrong jobs, or silently miss the right ones. + const candidates = failed.filter((job) => isRateLimitError(job.lastError)); + + const now = Date.now(); + let requeued = 0; + + for (const job of candidates.slice(0, limit)) { + if (!dryRun) { + await ctx.db.patch(job._id, { + status: "queued", + attempts: 0, + nextAttemptAt: now, + lastError: undefined, + processingRunId: undefined, + leaseExpiresAt: undefined, + updatedAt: now, + }); + } + requeued++; + } + + return { + dryRun, + scannedFailed: failed.length, + rateLimited: candidates.length, + requeued, + remaining: Math.max(candidates.length - requeued, 0), + skippedRealFailures: failed.length - candidates.length, + }; + }, +}); + +/** + * COST MODE — reclaim orphaned Convex file storage. + * + * Prod file storage sits at ~992 MB against a 1 GB free allowance, but + * `eventShareAssets.storageId` is the ONLY `v.id("_storage")` reference in the + * entire schema (see schema.ts), and that table is empty in prod. Every stored + * file is therefore unreachable: they are share-asset images generated before + * the 2026-07-07 prod database wipe, which cleared the rows but left the blobs + * behind. Share-asset generation is intentionally disabled, so nothing will + * ever reference them again. + * + * DESTRUCTIVE AND IRREVERSIBLE. Deleted blobs cannot be recovered. The guard + * below re-derives the live reference set at run time rather than trusting the + * analysis above, so it stays correct if share assets are ever re-enabled. + * + * Paginated. Run dry first and confirm `wouldDelete` matches expectations, then + * run for real, feeding `continueCursor` back in as `cursor` until `isDone`: + * npx convex run --prod migrations:purgeOrphanedStorageFiles '{"dryRun":true}' + * npx convex run --prod migrations:purgeOrphanedStorageFiles '{"dryRun":false}' + * npx convex run --prod migrations:purgeOrphanedStorageFiles '{"dryRun":false,"cursor":""}' + */ +export const purgeOrphanedStorageFiles = internalMutation({ + args: { + dryRun: v.optional(v.boolean()), + limit: v.optional(v.number()), + cursor: v.optional(v.union(v.string(), v.null())), + }, + handler: async (ctx, args) => { + const dryRun = args.dryRun ?? true; + const limit = Math.min(Math.max(args.limit ?? 200, 1), 500); + + // Re-derive the live reference set instead of assuming it is empty. + const referenced = new Set(); + for (const asset of await ctx.db.query("eventShareAssets").collect()) { + if (asset.storageId) referenced.add(asset.storageId); + } + + // Paginate rather than repeatedly `take`-ing from the head of the table. + // Retained (referenced) files stay at the head forever, so a head scan + // would re-examine them on every call and could never reach the rows behind + // them. In a dry run nothing is deleted at all, so a head scan would simply + // return the same page every time. + const page = await ctx.db.system + .query("_storage") + .paginate({ numItems: limit, cursor: args.cursor ?? null }); + + let removed = 0; + let deletedBytes = 0; + let kept = 0; + + for (const file of page.page) { + if (referenced.has(file._id)) { + kept++; + continue; + } + removed++; + deletedBytes += file.size ?? 0; + if (!dryRun) { + await ctx.storage.delete(file._id); + } + } + + return { + dryRun, + referencedCount: referenced.size, + scanned: page.page.length, + kept, + [dryRun ? "wouldDelete" : "deleted"]: removed, + approxMbReclaimed: Number((deletedBytes / 1048576).toFixed(2)), + // Feed this back in as `cursor` to continue; `isDone` means the whole + // table has been walked, not merely that this page was clean. + isDone: page.isDone, + continueCursor: page.isDone ? null : page.continueCursor, + }; + }, +}); diff --git a/packages/backend/convex/pipeline.ts b/packages/backend/convex/pipeline.ts index 09fd0c1..df08e4f 100644 --- a/packages/backend/convex/pipeline.ts +++ b/packages/backend/convex/pipeline.ts @@ -14,14 +14,68 @@ import type { Doc } from "./_generated/dataModel"; const DAY_MS = 24 * 60 * 60 * 1000; const HOUR_MS = 60 * 60 * 1000; -const FRESHNESS_SLO_MS = 60 * 60 * 1000; + +// --------------------------------------------------------------------------- +// Pipeline cadence (cost mode) — the basis for every time-based alert below +// --------------------------------------------------------------------------- +// The pipeline no longer runs continuously. It runs in four batched windows per +// day (ingest 00/06/12/18 UTC, then enrich :15, cluster :30, summarize :45), +// with maintenance jobs demoted to once or twice daily and the alert check +// itself down to 2x daily (01:50 / 13:50 UTC). +// +// Thresholds tuned for the old continuous cadence are all *unsatisfiable* under +// this one — a 1-hour freshness SLO against a 6-hour publish cadence, a 30-min +// enrichment window against a job that runs every 6 hours, a 4-hour +// archive-run window against a job that now runs daily. Left alone they would +// fire on every check, turning /admin/pipeline into permanent red and burning +// writes to `pipelineAlerts`. +// +// So: every threshold below is derived from these periods plus a tolerance. +// Retuning the crons means retuning these constants and nothing else. +const PIPELINE_WINDOW_MS = 6 * HOUR_MS; // ingest → … → summarize batch cadence +const DAILY_JOB_PERIOD_MS = DAY_MS; // archive / recluster / prune cadence +const WINDOWS_PER_DAY = Math.round(DAY_MS / PIPELINE_WINDOW_MS); // 4 +// Offset from a window's start to the end of its summarize step (crons.ts runs +// ingest at :00 and summarize at :45). Spend for a window has not landed until +// this much of it has elapsed. +const WINDOW_COMPLETION_OFFSET_MS = 45 * 60 * 1000; +// How much slack a healthy pipeline gets: one fully missed batch window. +const ALERT_TOLERANCE = 2; +// Fallback when `pipeline_alert_check_interval_minutes` cannot be read. Must +// track the check-pipeline-alerts cron. +const DEFAULT_ALERT_CHECK_PERIOD_MS = 12 * HOUR_MS; +// How far a burn rate may run ahead of the day's expected progress before it is +// treated as a projected exhaustion. +const VECTOR_BURN_RATE_MARGIN = 0.25; + +// Feed freshness. Content can only become visible once per batch window, so the +// previous 1-hour SLO was unsatisfiable by construction (worst case is ~6h *by +// design*). Allow one fully missed window before calling the feed stale. +const FRESHNESS_SLO_MS = PIPELINE_WINDOW_MS * ALERT_TOLERANCE; + +// A job running every `periodMs` is only "absent" once a full period *plus* one +// alert-check period has elapsed without an ok run. Without the second term, a +// check that lands just before the job's next run alerts on a perfectly healthy +// pipeline — e.g. archive runs daily at 02:20 and the check runs at 01:50, so +// the legitimately observed gap is ~23.5h. +function absentRunWindowMs(periodMs: number, alertCheckPeriodMs: number) { + return periodMs + alertCheckPeriodMs; +} + const ARTICLE_QUEUE_STATUSES = [ "unprocessed", "enriched", "processing", "archived", ] as const; -const DIAGNOSTIC_COUNT_LIMIT = 5000; +// COST: every one of these "counts" is really a scan that materialises full +// documents (Convex has no count aggregate), and the tables being counted are +// the two fattest in the schema (articles, events). These run inside reactive +// admin `query`s, so an open /admin/pipeline tab re-executes them on every +// article write. 2000 still comfortably exceeds a normal day's volume +// (~1300 articles/day), so day-scoped counters stay exact; only whole-table +// queue depths saturate, and those are read as "deep" either way. +const DIAGNOSTIC_COUNT_LIMIT = 2000; const pipelineMetricValue = v.union( v.string(), @@ -405,13 +459,16 @@ export const getStuckProcessingEvents = query({ handler: async (ctx) => { await requireAdminUser(ctx); const now = Date.now(); + // Bounded: the response is five age buckets plus the 20 oldest rows, so a + // 5000-row scan of full event docs bought nothing but I/O. Ascending order + // means the oldest (the ones that matter) are always the rows we keep. const events = await ctx.db .query("events") .withIndex("by_status_last_article_at", (q) => q.eq("status", "processing"), ) .order("asc") - .take(5000); + .take(1000); const buckets: Record = { "<1h": 0, "1-6h": 0, @@ -620,13 +677,15 @@ export const getPipelineDoctor = query({ .lt("enrichmentLeaseExpiresAt", staleLeaseCutoff), ), ); + // Only the first 20 survivors of the filter below are ever returned, so a + // 500-row scan of full event docs was ~2.5x more I/O than the answer needs. const recentProcessingEvents = await ctx.db .query("events") .withIndex("by_status_last_article_at", (q) => q.eq("status", "processing"), ) .order("desc") - .take(500); + .take(200); const oneShortOfPublish = recentProcessingEvents .filter((event) => { const articleCount = event.articleCount ?? 1; @@ -831,30 +890,48 @@ export const checkPipelineAlerts = internalAction({ const now = Date.now(); const startedAt = now; const runId = `checkPipelineAlerts-${startedAt}`; - const hourAgo = now - HOUR_MS; - const dayAgo = now - DAY_MS; let evaluatedRules = 0; let stuckProcessingOver72h = 0; let status: "ok" | "error" = "ok"; let errorMessage: string | undefined; try { + const { alertCheckPeriodMs } = await ctx.runQuery( + internal.pipeline.getAlertCadence, + {}, + ); + // Every rule below reasons over "what happened since the previous check", + // so this is the natural evaluation window for anything that runs at + // least once per batch window. + const sinceLastCheck = now - alertCheckPeriodMs; + // Daily jobs need a wider log history than the check period: the archive + // job legitimately last ran ~23.5h before the 01:50 check, so a 24h log + // window would miss it by minutes and alert on a healthy pipeline. + const logLookbackMs = absentRunWindowMs( + DAILY_JOB_PERIOD_MS, + alertCheckPeriodMs, + ); + const logs = await ctx.runQuery(internal.pipeline.getRecentPipelineLogs, { - since: dayAgo, + since: now - logLookbackMs, limit: 1000, }); + // COST: only fetch the vector runs the fallback rule actually looks at + // (since the previous check) instead of a full day's worth. const vectorRuns = await ctx.runQuery( internal.pipeline.getRecentVectorRunsForAlerts, - { since: dayAgo }, + { since: sinceLastCheck }, ); + // Clustering runs once per batch window, so an hour-wide window would + // almost never contain a clustering run at all and this rule could never + // fire. Evaluate every run since the previous check instead; 3 fallbacks + // in that span still means "persistently degraded", not "one bad batch". const recentFallbackRuns = ( vectorRuns as Array> ).filter( (run: Doc<"vectorSearchRuns">) => - run.jobName === "clusterEnrichedArticles" && - run.usedFallbackMode && - run.createdAt >= hourAgo, + run.jobName === "clusterEnrichedArticles" && run.usedFallbackMode, ); evaluatedRules++; if (recentFallbackRuns.length >= 3) { @@ -862,8 +939,11 @@ export const checkPipelineAlerts = internalAction({ severity: "warning", code: "fallback_persistent", message: - "clusterEnrichedArticles has used fallback mode at least 3 times in the last hour.", - details: { count: recentFallbackRuns.length }, + "clusterEnrichedArticles has used fallback mode at least 3 times since the previous alert check.", + details: { + count: recentFallbackRuns.length, + windowMs: alertCheckPeriodMs, + }, }); } @@ -872,12 +952,17 @@ export const checkPipelineAlerts = internalAction({ { since: now - FRESHNESS_SLO_MS }, ); evaluatedRules++; + // FRESHNESS_SLO_MS is now one batch window x tolerance, so this window + // always spans at least one completed summarize step. Zero visible + // previews across two whole windows means the pipeline really has stopped + // producing, not that we looked between batches. if (visible === 0) { await upsertAlert(ctx, { severity: "warning", code: "feed_visibility_drought", - message: - "No public feed previews became visible or refreshed within the 60-minute freshness SLO.", + message: `No public feed previews became visible or refreshed within the ${Math.round( + FRESHNESS_SLO_MS / HOUR_MS, + )}-hour freshness SLO (${ALERT_TOLERANCE} batch windows).`, details: { since: now - FRESHNESS_SLO_MS }, }); } @@ -886,41 +971,75 @@ export const checkPipelineAlerts = internalAction({ internal.pipeline.getVectorBudgetForAlerts, {}, ); + // Vector spend no longer accrues smoothly across the day — it arrives in + // WINDOWS_PER_DAY roughly equal steps. Expected progress must therefore be + // measured in completed batch windows, not elapsed hours; the old + // hour-based model treated a perfectly on-plan post-window reading as an + // overrun (at 13:50 it expected 58% used while 3 of 4 windows had run). + // + // A window counts as completed only once its summarize step has run, not + // the moment it starts — otherwise the whole of 00:00-00:45 is credited + // with spend that has not happened yet, and a manual alert check in that + // gap reads as an overrun. + const nowDate = new Date(now); + const startOfUtcDay = Date.UTC( + nowDate.getUTCFullYear(), + nowDate.getUTCMonth(), + nowDate.getUTCDate(), + ); + const windowsCompleted = Math.max( + 0, + Math.min( + WINDOWS_PER_DAY, + Math.floor( + (now - startOfUtcDay - WINDOW_COMPLETION_OFFSET_MS) / + PIPELINE_WINDOW_MS, + ) + 1, + ), + ); + const expectedDailyProgress = windowsCompleted / WINDOWS_PER_DAY; evaluatedRules++; - if (budget.ratio >= 0.75 && new Date(now).getUTCHours() < 18) { + // "Most of the budget gone with most of the day still to run." Only + // meaningful while at least half the day's windows are still ahead: at + // 3-of-4 windows completed, 75% used is exactly on plan, not an alarm. + if (budget.ratio >= 0.75 && windowsCompleted <= WINDOWS_PER_DAY / 2) { await upsertAlert(ctx, { severity: "warning", code: "vector_budget_burn_rate", - message: "Vector search qGB usage exceeded 75% before 18:00 UTC.", + message: `Vector search qGB usage exceeded 75% with only ${windowsCompleted} of ${WINDOWS_PER_DAY} daily pipeline windows completed.`, details: { usedQgb: budget.usedQgb, limitQgb: budget.limitQgb, ratio: budget.ratio, + windowsCompleted, }, }); } - const utcHour = new Date(now).getUTCHours(); - const expectedDailyProgress = (utcHour + 1) / 24; evaluatedRules++; - if (budget.ratio >= Math.min(0.9, expectedDailyProgress + 0.25)) { + if ( + budget.ratio >= + Math.min(0.9, expectedDailyProgress + VECTOR_BURN_RATE_MARGIN) + ) { await upsertAlert(ctx, { severity: "warning", code: "p0_budget_projected_exhaustion", message: - "Vector budget burn rate is ahead of UTC-day progress; throttle non-core pipeline work before feed creation is affected.", + "Vector budget burn rate is ahead of the day's completed pipeline windows; throttle non-core pipeline work before feed creation is affected.", details: { usedQgb: budget.usedQgb, limitQgb: budget.limitQgb, ratio: budget.ratio, expectedDailyProgress, + windowsCompleted, }, }); } - stuckProcessingOver72h = await ctx.runQuery( + const stuckProcessing = await ctx.runQuery( internal.pipeline.countProcessingEventsOlderThan, { ageMs: 72 * HOUR_MS }, ); + stuckProcessingOver72h = stuckProcessing.count; const recentAlertGauge = (logs as Array>) .filter( (log) => @@ -933,7 +1052,21 @@ export const checkPipelineAlerts = internalAction({ ? undefined : recentAlertGauge.gauges.stuckProcessingOver72h; evaluatedRules++; - if ( + // The count saturates at STUCK_PROCESSING_CAP so it stays cheap to + // compute. Once saturated, growth is no longer observable — but a backlog + // that large is itself the condition worth alerting on, so saturation + // raises the alert directly instead of silently going quiet. + if (stuckProcessing.saturated) { + await upsertAlert(ctx, { + severity: "warning", + code: "stuck_processing_growth", + message: `Processing events older than 72 hours reached the ${STUCK_PROCESSING_CAP}+ alerting cap.`, + details: { + current: stuckProcessingOver72h, + saturated: true, + }, + }); + } else if ( typeof previousStuckProcessingOver72h === "number" && stuckProcessingOver72h > previousStuckProcessingOver72h ) { @@ -965,20 +1098,29 @@ export const checkPipelineAlerts = internalAction({ byJob.set(log.jobName, row); } evaluatedRules++; + const logLookbackHours = Math.round(logLookbackMs / HOUR_MS); for (const [jobName, row] of byJob.entries()) { + // The 3-run minimum is what keeps this rule honest under the batched + // cadence: per-window jobs accumulate ~6 runs across the log lookback, + // so a single bad batch cannot trip it, while once-daily jobs never + // reach the minimum and are covered by the absent-ok-run checks below + // instead of by a 1-of-1 "0% success" false alarm. if (row.total >= 3 && row.ok / row.total < 0.8) { await upsertAlert(ctx, { severity: "error", code: `job_error_rate:${jobName}`, - message: `${jobName} success ratio is below 80% over the last 24 hours.`, + message: `${jobName} success ratio is below 80% over the last ${logLookbackHours} hours.`, details: { total: row.total, ok: row.ok }, }); } } + // Enrichment runs once per batch window (:15 past), so a 30-minute window + // never contained a run and this rule was dead. Evaluate every enrichment + // run since the previous check instead. const enrichmentWindow = (logs as Array>).filter( (log) => log.jobName === "enrichUnprocessedArticles" && - log.startedAt >= now - 30 * 60 * 1000, + log.startedAt >= sinceLastCheck, ); const enrichmentAttempts = enrichmentWindow.reduce( (sum, log) => sum + (log.counters.claimedArticles ?? 0), @@ -997,29 +1139,40 @@ export const checkPipelineAlerts = internalAction({ severity: "warning", code: "enrichment_failure_rate", message: - "More than 20% of claimed enrichment articles failed in the last 30 minutes.", + "More than 20% of claimed enrichment articles failed since the previous alert check.", details: { attempts: enrichmentAttempts, failures: enrichmentFailures, ratio: enrichmentFailures / enrichmentAttempts, + windowMs: alertCheckPeriodMs, }, }); } + // archiveStaleSingletonEvents now runs once daily (02:20 UTC) rather than + // continuously, so the old 4-hour window guaranteed a false alarm on every + // check. One full job period plus one alert period absorbs the legitimate + // ~23.5h gap seen by the 01:50 check while still catching a job that has + // genuinely stopped. + const archiveAbsentWindowMs = absentRunWindowMs( + DAILY_JOB_PERIOD_MS, + alertCheckPeriodMs, + ); const archiveOk = (logs as Array>).some( (log: Doc<"pipelineRunLogs">) => log.jobName === "archiveStaleSingletonEvents" && log.status === "ok" && - log.startedAt >= now - 4 * HOUR_MS, + log.startedAt >= now - archiveAbsentWindowMs, ); evaluatedRules++; if (!archiveOk) { await upsertAlert(ctx, { severity: "warning", code: "archive_run_absent", - message: - "archiveStaleSingletonEvents has not produced an ok log in the last 4 hours.", - details: {}, + message: `archiveStaleSingletonEvents has not produced an ok log in the last ${Math.round( + archiveAbsentWindowMs / HOUR_MS, + )} hours.`, + details: { windowMs: archiveAbsentWindowMs }, }); } } catch (error) { @@ -1116,28 +1269,82 @@ export const countVisiblePreviewsSince = internalQuery({ }, }); +/** + * Saturating count of "stuck processing" events, used only by the 20-minute + * alert cron. + * + * COST: this was the third-largest database-I/O consumer in the whole app + * (~1.7 GB / 19 days). It ran 72x a day and read up to 5000 *full* event + * documents per scan — twice, because the second scan re-reads essentially the + * same rows just to catch legacy rows that predate `lastArticleAt`. + * + * The alert rule only asks "is the backlog growing?", never "exactly how big is + * it?", so an exact count was never needed. We now stop at STUCK_PROCESSING_CAP + * and report saturation explicitly: below the cap the growth comparison works + * exactly as before, and at the cap `saturated` is itself the alarm (a backlog + * of 300+ stuck events is already a page-worthy state, and growth beyond it + * tells the operator nothing new). + */ +const STUCK_PROCESSING_CAP = 300; +// Legacy rows missing `lastArticleAt` are a fixed, shrinking set that the +// primary index cannot see. Sample a small window rather than walking every +// old processing event to count what is almost always zero. +const STUCK_PROCESSING_LEGACY_CAP = 100; + export const countProcessingEventsOlderThan = internalQuery({ args: { ageMs: v.number() }, handler: async (ctx, args) => { const cutoff = Date.now() - Math.max(0, args.ageMs); - const withLastArticleAt = await limitedCount(() => - ctx.db - .query("events") - .withIndex("by_status_last_article_at", (q) => - q.eq("status", "processing").lt("lastArticleAt", cutoff), - ), - ); - const withoutLastArticleAt = await limitedReduce( + const withLastArticleAt = await limitedCount( () => ctx.db .query("events") - .withIndex("by_status_recency", (q) => - q.eq("status", "processing").lt("firstPublishedAt", cutoff), + .withIndex("by_status_last_article_at", (q) => + q.eq("status", "processing").lt("lastArticleAt", cutoff), ), - 0, - (acc, row) => (row.lastArticleAt === undefined ? acc + 1 : acc), + STUCK_PROCESSING_CAP, + ); + // Short-circuit: already saturated, so the legacy scan cannot change the + // reported signal. + const withoutLastArticleAt = + withLastArticleAt >= STUCK_PROCESSING_CAP + ? 0 + : await limitedReduce( + () => + ctx.db + .query("events") + .withIndex("by_status_recency", (q) => + q.eq("status", "processing").lt("firstPublishedAt", cutoff), + ), + 0, + (acc, row) => (row.lastArticleAt === undefined ? acc + 1 : acc), + STUCK_PROCESSING_LEGACY_CAP, + ); + const count = Math.min( + withLastArticleAt + withoutLastArticleAt, + STUCK_PROCESSING_CAP, + ); + return { count, saturated: count >= STUCK_PROCESSING_CAP }; + }, +}); + +/** + * The alert-check cadence, read from config so the thresholds track the + * check-pipeline-alerts cron without a code deploy. `checkPipelineAlerts` is an + * action and has no `ctx.db`, hence the wrapper. + */ +export const getAlertCadence = internalQuery({ + args: {}, + handler: async (ctx) => { + const minutes = await readConfigNumber( + ctx, + "pipeline_alert_check_interval_minutes", + DEFAULT_ALERT_CHECK_PERIOD_MS / 60_000, ); - return withLastArticleAt + withoutLastArticleAt; + // Clamp so a mistyped config value can neither disable alerting (absurdly + // wide windows) nor make it hair-trigger (windows narrower than a run). + const clamped = Math.min(Math.max(minutes, 5), 24 * 60); + return { alertCheckPeriodMs: clamped * 60_000 }; }, }); diff --git a/packages/backend/convex/retention.test.ts b/packages/backend/convex/retention.test.ts index 445518e..7f8436e 100644 --- a/packages/backend/convex/retention.test.ts +++ b/packages/backend/convex/retention.test.ts @@ -4,11 +4,15 @@ import { readFileSync } from "node:fs"; import { join } from "node:path"; import { convexTest } from "convex-test"; -import { describe, expect, test } from "vitest"; +import { afterEach, describe, expect, test, vi } from "vitest"; import schema from "./schema"; import { internal } from "./_generated/api"; -import { RETENTION_POLICY } from "./retention"; +import { + RETENTION_POLICY, + STORAGE_RETENTION_CONFIG_KEYS, + STORAGE_RETENTION_DEFAULTS, +} from "./retention"; const modules = ( import.meta as unknown as { @@ -149,3 +153,341 @@ describe("retention purges (L11)", () => { expect(cronSource).toContain("retention-purge-expired-insights"); }); }); + +// --------------------------------------------------------------------------- +// Storage-cost retention (Convex bills per GB-month) +// --------------------------------------------------------------------------- +// convex-test stamps `_creationTime` at insert, so these tests seed fixtures at +// the real clock and then jump `Date.now()` forward past the retention window. +// Only `Date` is faked so convex-test's own async plumbing keeps working. + +const CLOCK_SKIP_DAYS = 400; + +function jumpClockPastRetention(base: number): number { + const clockNow = base + CLOCK_SKIP_DAYS * DAY_MS; + vi.useFakeTimers({ toFake: ["Date"] }); + vi.setSystemTime(clockNow); + return clockNow; +} + +describe("storage-cost retention purges", () => { + afterEach(() => { + vi.useRealTimers(); + }); + + test("embeddings outside the clustering window are purged; the article row survives", async () => { + const t = convexTest(schema, modules); + const clockNow = Date.now() + CLOCK_SKIP_DAYS * DAY_MS; + + const ids = await t.run(async (ctx) => { + const sourceId = await ctx.db.insert("sources", { + domain: "storage.ro", + name: "Storage", + baseBias: 0, + reliabilityScore: 5, + }); + const staleArticleId = await ctx.db.insert("articles", { + sourceId, + title: "Stale", + url: "https://storage.ro/stale", + canonicalUrl: "https://storage.ro/stale", + status: "clustered", + publishedAt: clockNow - 120 * DAY_MS, + }); + const staleEmbeddingId = await ctx.db.insert("articleEmbeddings", { + articleId: staleArticleId, + embedding: [0.1, 0.2, 0.3], + version: 1, + }); + const freshArticleId = await ctx.db.insert("articles", { + sourceId, + title: "Fresh", + url: "https://storage.ro/fresh", + canonicalUrl: "https://storage.ro/fresh", + status: "clustered", + publishedAt: clockNow - 1 * DAY_MS, + }); + const freshEmbeddingId = await ctx.db.insert("articleEmbeddings", { + articleId: freshArticleId, + embedding: [0.4, 0.5, 0.6], + version: 1, + }); + return { + staleArticleId, + staleEmbeddingId, + freshArticleId, + freshEmbeddingId, + }; + }); + + jumpClockPastRetention(Date.now()); + + const result = await t.mutation( + internal.retention.purgeStaleArticleEmbeddings, + {}, + ); + expect(result.deleted).toBe(1); + expect(result.skipped).toBe(1); + + await t.run(async (ctx) => { + // The vector is gone... + expect(await ctx.db.get(ids.staleEmbeddingId)).toBeNull(); + // ...but the article it belonged to is untouched. + expect(await ctx.db.get(ids.staleArticleId)).not.toBeNull(); + // Articles still inside the window keep their vector. + expect(await ctx.db.get(ids.freshEmbeddingId)).not.toBeNull(); + expect(await ctx.db.get(ids.freshArticleId)).not.toBeNull(); + }); + }); + + test("the embedding window is tunable via config and defaults to 45 days", async () => { + const t = convexTest(schema, modules); + const clockNow = Date.now() + CLOCK_SKIP_DAYS * DAY_MS; + + const embeddingId = await t.run(async (ctx) => { + const sourceId = await ctx.db.insert("sources", { + domain: "cfg.ro", + name: "Cfg", + baseBias: 0, + reliabilityScore: 5, + }); + const articleId = await ctx.db.insert("articles", { + sourceId, + title: "Sixty days old", + url: "https://cfg.ro/a", + canonicalUrl: "https://cfg.ro/a", + status: "clustered", + publishedAt: clockNow - 60 * DAY_MS, + }); + // A longer window than the 60-day-old article -> must be kept. + await ctx.db.insert("config", { + key: STORAGE_RETENTION_CONFIG_KEYS.articleEmbeddingDays, + value: JSON.stringify(180), + description: "test override", + updatedAt: Date.now(), + }); + return ctx.db.insert("articleEmbeddings", { + articleId, + embedding: [0.1], + version: 1, + }); + }); + + jumpClockPastRetention(Date.now()); + + const held = await t.mutation( + internal.retention.purgeStaleArticleEmbeddings, + {}, + ); + expect(held.deleted).toBe(0); + await t.run(async (ctx) => { + expect(await ctx.db.get(embeddingId)).not.toBeNull(); + }); + + // With the default 45-day window the same row is beyond retention. + const purged = await t.mutation( + internal.retention.purgeStaleArticleEmbeddings, + { retentionDays: STORAGE_RETENTION_DEFAULTS.articleEmbeddingDays }, + ); + expect(purged.deleted).toBe(1); + await t.run(async (ctx) => { + expect(await ctx.db.get(embeddingId)).toBeNull(); + }); + }); + + test("orphaned embeddings (article already deleted) are purged", async () => { + const t = convexTest(schema, modules); + + const { orphanId, liveEmbeddingId } = await t.run(async (ctx) => { + const sourceId = await ctx.db.insert("sources", { + domain: "orphan.ro", + name: "Orphan", + baseBias: 0, + reliabilityScore: 5, + }); + const doomedArticleId = await ctx.db.insert("articles", { + sourceId, + title: "Doomed", + url: "https://orphan.ro/x", + canonicalUrl: "https://orphan.ro/x", + status: "clustered", + publishedAt: Date.now(), + }); + const orphanId = await ctx.db.insert("articleEmbeddings", { + articleId: doomedArticleId, + embedding: [0.1], + version: 1, + }); + await ctx.db.delete(doomedArticleId); + + const liveArticleId = await ctx.db.insert("articles", { + sourceId, + title: "Live", + url: "https://orphan.ro/y", + canonicalUrl: "https://orphan.ro/y", + status: "clustered", + publishedAt: Date.now(), + }); + const liveEmbeddingId = await ctx.db.insert("articleEmbeddings", { + articleId: liveArticleId, + embedding: [0.2], + version: 1, + }); + return { orphanId, liveEmbeddingId }; + }); + + const result = await t.mutation( + internal.retention.purgeOrphanedArticleEmbeddings, + {}, + ); + expect(result.deleted).toBe(1); + expect(result.done).toBe(true); + + await t.run(async (ctx) => { + expect(await ctx.db.get(orphanId)).toBeNull(); + expect(await ctx.db.get(liveEmbeddingId)).not.toBeNull(); + }); + }); + + test("archived+detached articles are purged; attached, recent and gold-set articles are not", async () => { + const t = convexTest(schema, modules); + const clockNow = Date.now() + CLOCK_SKIP_DAYS * DAY_MS; + + const ids = await t.run(async (ctx) => { + const sourceId = await ctx.db.insert("sources", { + domain: "arch.ro", + name: "Arch", + baseBias: 0, + reliabilityScore: 5, + }); + const eventId = await ctx.db.insert("events", { + title: "Live event", + slug: "live-event", + status: "published", + firstPublishedAt: clockNow, + articleCount: 1, + sourceCount: 1, + sourceIds: [sourceId], + }); + + const purgeableId = await ctx.db.insert("articles", { + sourceId, + title: "Archived + detached", + url: "https://arch.ro/1", + canonicalUrl: "https://arch.ro/1", + status: "archived", + archivedAt: clockNow - 200 * DAY_MS, + archivedReason: "stale_singleton", + publishedAt: clockNow - 200 * DAY_MS, + }); + const purgeableEmbeddingId = await ctx.db.insert("articleEmbeddings", { + articleId: purgeableId, + embedding: [0.1], + version: 1, + }); + + // Recently archived — still inside the window. + const recentlyArchivedId = await ctx.db.insert("articles", { + sourceId, + title: "Recently archived", + url: "https://arch.ro/2", + canonicalUrl: "https://arch.ro/2", + status: "archived", + archivedAt: clockNow - 3 * DAY_MS, + archivedReason: "stale_singleton", + publishedAt: clockNow - 3 * DAY_MS, + }); + + // Carries an archivedReason but was requeued into a live event: the + // per-row guards must refuse to delete it. + const requeuedId = await ctx.db.insert("articles", { + sourceId, + eventId, + title: "Requeued", + url: "https://arch.ro/3", + canonicalUrl: "https://arch.ro/3", + status: "enriched", + archivedAt: clockNow - 200 * DAY_MS, + archivedReason: "stale_processing", + publishedAt: clockNow - 200 * DAY_MS, + }); + + // Old + archived + detached, but cited by the clustering gold set. + const labeledId = await ctx.db.insert("articles", { + sourceId, + title: "Gold set member", + url: "https://arch.ro/4", + canonicalUrl: "https://arch.ro/4", + status: "archived", + archivedAt: clockNow - 200 * DAY_MS, + archivedReason: "stale_singleton", + publishedAt: clockNow - 200 * DAY_MS, + }); + await ctx.db.insert("clusterPairLabels", { + pairKey: `${labeledId}:${purgeableId}`, + leftArticleId: labeledId, + rightArticleId: purgeableId, + sameEvent: false, + labeledAt: clockNow - 210 * DAY_MS, + }); + + return { + purgeableId, + purgeableEmbeddingId, + recentlyArchivedId, + requeuedId, + labeledId, + }; + }); + + jumpClockPastRetention(Date.now()); + + const result = await t.mutation( + internal.retention.purgeArchivedDetachedArticles, + {}, + ); + // purgeableId is itself cited by the gold-set label, so nothing is deleted. + expect(result.deleted).toBe(0); + + await t.run(async (ctx) => { + await ctx.db.delete( + (await ctx.db.query("clusterPairLabels").first())!._id, + ); + }); + + const second = await t.mutation( + internal.retention.purgeArchivedDetachedArticles, + {}, + ); + expect(second.deleted).toBe(2); // purgeable + previously-labeled + expect(second.deletedEmbeddings).toBe(1); + + await t.run(async (ctx) => { + expect(await ctx.db.get(ids.purgeableId)).toBeNull(); + expect(await ctx.db.get(ids.purgeableEmbeddingId)).toBeNull(); + expect(await ctx.db.get(ids.labeledId)).toBeNull(); + // Guarded rows survive. + expect(await ctx.db.get(ids.recentlyArchivedId)).not.toBeNull(); + expect(await ctx.db.get(ids.requeuedId)).not.toBeNull(); + }); + }); + + test("storage purges log their data class and deleted count", async () => { + const t = convexTest(schema, modules); + await t.mutation(internal.retention.purgeStaleArticleEmbeddings, {}); + await t.mutation(internal.retention.purgeOrphanedArticleEmbeddings, {}); + await t.mutation(internal.retention.purgeArchivedDetachedArticles, {}); + + const logs = await t.run(async (ctx) => + ctx.db.query("pipelineRunLogs").collect(), + ); + const jobNames = logs.map((log) => log.jobName); + expect(jobNames).toContain("retention:article_embeddings_stale"); + expect(jobNames).toContain("retention:article_embeddings_orphaned"); + expect(jobNames).toContain("retention:articles_archived_detached"); + for (const log of logs) { + expect(log.counters.deleted).toBeDefined(); + expect(log.metadata.dataClass).toBeDefined(); + } + }); +}); diff --git a/packages/backend/convex/retention.ts b/packages/backend/convex/retention.ts index 21f0ac7..96c7af7 100644 --- a/packages/backend/convex/retention.ts +++ b/packages/backend/convex/retention.ts @@ -11,12 +11,22 @@ * - unverified accounts: authMaintenance.cleanupExpiredUnverifiedAccounts * (7 days), already scheduled in crons.ts. * - opted-out domain content: purged immediately on state change (L5). + * + * --------------------------------------------------------------------------- + * Storage-cost retention (see STORAGE_RETENTION_DEFAULTS below) + * --------------------------------------------------------------------------- + * RETENTION_POLICY covers the *legal* minimization classes. The classes below + * are *operational* — they exist to stop unbounded Convex database growth + * (billed per GB-month). They are runtime-tunable via the `config` table so + * the window can be shortened without a deploy; the inline defaults here are + * authoritative until the key is seeded. */ import { v } from "convex/values"; import { internalMutation } from "./_generated/server"; import type { MutationCtx } from "./_generated/server"; import { internal } from "./_generated/api"; +import { getConfig } from "./config"; // Moved to lib/retentionPolicy.ts (pure) so the web privacy policy renders // from the exact object the purge crons enforce. @@ -25,11 +35,80 @@ import { RETENTION_POLICY } from "./lib/retentionPolicy"; const PURGE_BATCH = 200; +const DAY_MS = 24 * 60 * 60 * 1000; + +/** + * Operational (storage-cost) retention. Each entry documents a data class, + * the config key that tunes it and the default used before that key exists. + * + * - articleEmbeddings (stale): 512-float vectors (~4 KB+ per row) that exist + * ONLY so clustering can vector-match *recent* articles. Clustering never + * looks further back than `DEFAULT_RECLUSTER_WINDOW_HOURS` (48h, see + * clustering.ts), so a vector whose article is 45 days old can no longer + * influence any clustering decision — it is pure paid storage. The article + * row itself is KEPT (it is rendered in the UI and referenced by events); + * only the vector is dropped. + * - articleEmbeddings (orphaned): vectors whose `articleId` no longer + * resolves. Unreachable by every read path in the codebase (all of them go + * article -> by_article -> embedding), so they are pure garbage. + * - articles (archived + detached): articles that singletonCleanup archived + * (`archivedReason` set, `eventId` cleared, their event deleted). Excluded + * from enrichment and clustering by status, never rendered. See the + * safety notes on purgeArchivedDetachedArticles. + */ +export const STORAGE_RETENTION_DEFAULTS = { + /** Article embedding vectors, keyed on the article's publishedAt. */ + articleEmbeddingDays: 45, + /** Articles archived by singletonCleanup and detached from every event. */ + archivedArticleDays: 90, +} as const; + +/** Config keys that tune the storage-cost purges (registered in config.ts). */ +export const STORAGE_RETENTION_CONFIG_KEYS = { + articleEmbeddingDays: "article_embedding_retention_days", + archivedArticleDays: "archived_article_retention_days", +} as const; + +/** + * The only archivedReason values singletonCleanup ever writes. Both paths + * delete the owning event in the same mutation, so an article carrying either + * reason is guaranteed to be detached from every event. + */ +const ARCHIVED_REASONS = ["stale_singleton", "stale_processing"] as const; + +/** + * clusterPairLabels is a hand-curated set (admin-labeled), so it is small. + * If it ever exceeds this we refuse to delete articles rather than risk + * deleting one the gold set references. + */ +const CLUSTER_LABEL_GUARD_LIMIT = 5000; + +function clampBatch(value: number | undefined, fallback: number): number { + if (typeof value !== "number" || !Number.isFinite(value)) return fallback; + return Math.max(1, Math.min(500, Math.floor(value))); +} + +function clampRetentionDays( + value: number | undefined, + fallback: number, + floor: number, +): number { + if (typeof value !== "number" || !Number.isFinite(value)) return fallback; + // Never let a bad config value shrink the window below the floor — that is + // the guard that keeps a typo from deleting vectors clustering still needs. + return Math.max(floor, Math.floor(value)); +} + async function logPurgeRun( ctx: MutationCtx, dataClass: string, deleted: number, done: boolean, + extra?: { + counters?: Record; + gauges?: Record; + metadata?: Record; + }, ) { const now = Date.now(); await ctx.db.insert("pipelineRunLogs", { @@ -39,9 +118,9 @@ async function logPurgeRun( finishedAt: now, durationMs: 0, status: "ok", - counters: { deleted }, - gauges: { done }, - metadata: { dataClass }, + counters: { deleted, ...(extra?.counters ?? {}) }, + gauges: { done, ...(extra?.gauges ?? {}) }, + metadata: { dataClass, ...(extra?.metadata ?? {}) }, createdAt: now, }); } @@ -137,3 +216,303 @@ export const purgeExpiredUserInsights = internalMutation({ return { deleted: candidates.length, done }; }, }); + +// --------------------------------------------------------------------------- +// Storage-cost purges (Convex bills per GB-month of database storage) +// --------------------------------------------------------------------------- + +/** + * Delete `articleEmbeddings` rows belonging to articles outside the clustering + * window. THE ARTICLE ROWS ARE LEFT INTACT — only the vector is dropped. + * + * Why this is safe: every read path for an embedding starts from an article + * (`by_article` / `by_article_version`) and every one of those callers is + * scoped to recent articles — clustering's recluster window defaults to 48h + * (clustering.ts DEFAULT_RECLUSTER_WINDOW_HOURS) and enrichment only embeds + * articles it is currently processing. A 45-day window is ~22x the widest + * window anything actually reads, so no live clustering decision can change. + * Consumers that miss an embedding degrade gracefully (they `return null` / + * skip the article) rather than throwing. + * + * Scan strategy: walk the table on the implicit `by_creation_time` index + * oldest-first with `_creationTime < cutoff`, so each invocation only touches + * the head of the index and deleted rows never get re-read. Every candidate is + * then double-checked against its article's `publishedAt` before deletion, so + * a vector is only dropped when BOTH its creation time and its article are + * beyond the window. Rows whose article no longer exists are deleted too (they + * are orphans by definition and would otherwise stall the head of the scan). + * + * Re-runnable and self-chaining: it reschedules itself while a full batch is + * still being deleted, and stops as soon as a batch makes no progress. + */ +export const purgeStaleArticleEmbeddings = internalMutation({ + args: { + retentionDays: v.optional(v.number()), + batchSize: v.optional(v.number()), + }, + handler: async (ctx, args) => { + const configured = await getConfig( + ctx, + STORAGE_RETENTION_CONFIG_KEYS.articleEmbeddingDays, + STORAGE_RETENTION_DEFAULTS.articleEmbeddingDays, + ); + // Floor of 7 days: even a fat-fingered config value can never cut into + // the 48h clustering window (plus a wide safety margin). + const retentionDays = clampRetentionDays( + args.retentionDays ?? configured, + STORAGE_RETENTION_DEFAULTS.articleEmbeddingDays, + 7, + ); + const batchSize = clampBatch(args.batchSize, PURGE_BATCH); + const cutoff = Date.now() - retentionDays * DAY_MS; + + const candidates = await ctx.db + .query("articleEmbeddings") + .withIndex("by_creation_time", (q) => q.lt("_creationTime", cutoff)) + .take(batchSize); + + let deleted = 0; + let orphaned = 0; + let skipped = 0; + for (const row of candidates) { + const article = await ctx.db.get(row.articleId); + if (article === null) { + // Unreachable garbage — no read path can reach it. + await ctx.db.delete(row._id); + deleted++; + orphaned++; + continue; + } + if (article.publishedAt >= cutoff) { + // Old vector, recently published article (backfill edge case). + // Keep it: the article could still be inside a clustering window. + skipped++; + continue; + } + await ctx.db.delete(row._id); + deleted++; + } + + // Stop when the head of the index is exhausted, or when a full batch + // produced no deletions (otherwise skipped rows would loop forever). + const done = candidates.length < batchSize || deleted === 0; + await logPurgeRun(ctx, "article_embeddings_stale", deleted, done, { + counters: { scanned: candidates.length, orphaned, skipped }, + metadata: { retentionDays, cutoff }, + }); + + if (!done) { + await ctx.scheduler.runAfter( + 0, + internal.retention.purgeStaleArticleEmbeddings, + { retentionDays, batchSize }, + ); + } + return { deleted, scanned: candidates.length, skipped, orphaned, done }; + }, +}); + +/** + * Delete `articleEmbeddings` rows whose `articleId` no longer resolves to an + * article. These are unreachable by construction — every consumer looks the + * embedding up *from* an article via `by_article` / `by_article_version`, so a + * vector with no article can never be read again. + * + * Unlike the stale purge this must sweep the whole table (an orphan can have + * any creation time), so it walks it once with a persisted pagination cursor + * chained through the scheduler instead of restarting from the front. Because + * a sweep reads every vector row, this is bandwidth-expensive: schedule it + * infrequently (weekly is plenty — the stale purge already collects every + * orphan older than the embedding window for free). + */ +export const purgeOrphanedArticleEmbeddings = internalMutation({ + args: { + cursor: v.optional(v.union(v.string(), v.null())), + batchSize: v.optional(v.number()), + scannedSoFar: v.optional(v.number()), + }, + handler: async (ctx, args) => { + const batchSize = clampBatch(args.batchSize, PURGE_BATCH); + + const page = await ctx.db.query("articleEmbeddings").paginate({ + cursor: args.cursor ?? null, + numItems: batchSize, + }); + + let deleted = 0; + for (const row of page.page) { + const article = await ctx.db.get(row.articleId); + if (article !== null) continue; + await ctx.db.delete(row._id); + deleted++; + } + + const scanned = (args.scannedSoFar ?? 0) + page.page.length; + const done = page.isDone; + await logPurgeRun(ctx, "article_embeddings_orphaned", deleted, done, { + counters: { scanned: page.page.length, scannedTotal: scanned }, + }); + + if (!done) { + await ctx.scheduler.runAfter( + 0, + internal.retention.purgeOrphanedArticleEmbeddings, + { + cursor: page.continueCursor, + batchSize, + scannedSoFar: scanned, + }, + ); + } + return { deleted, scanned, done }; + }, +}); + +/** + * Delete articles that singletonCleanup archived AND detached from every + * event, once they are past the archived-article window. Their embeddings go + * with them. + * + * Safety reasoning — every one of these must hold, and each is re-checked on + * the row itself before the delete, not just assumed from the index: + * 1. `archivedReason` is only ever set by singletonCleanup + * (`stale_singleton` / `stale_processing`), and in BOTH of those paths the + * owning event is `ctx.db.delete(args.eventId)`-ed in the same mutation + * and the article is patched with `eventId: undefined`. So the article + * belongs to no event — live, published, or otherwise. + * 2. We still assert `eventId === undefined` and `status === "archived"` per + * row, so an article that got requeued into the pipeline (which clears + * `archivedAt`/`archivedReason` and sets status back to `enriched`) can + * never be picked up mid-flight. + * 3. Nothing renders archived articles: enrichment and clustering filter them + * out by status (enrichment.ts shouldEnrich/shouldReembed, clustering's + * `by_status_published` queries only take `enriched`/`clustered`), and the + * only reader of the archived class is an admin count query + * (pipeline.getArchivedArticleStats). + * 4. `clusterPairLabels` is the one table that can reference an article + * outside an event (a hand-labeled clustering gold set). We load it and + * refuse to delete anything it references. If that table is ever larger + * than the guard limit we bail out entirely rather than risk it. + */ +export const purgeArchivedDetachedArticles = internalMutation({ + args: { + retentionDays: v.optional(v.number()), + batchSize: v.optional(v.number()), + // Set only when this mutation reschedules itself; see the watermark note in + // the handler. Positionally aligned with ARCHIVED_REASONS. + cursors: v.optional(v.array(v.number())), + }, + handler: async (ctx, args) => { + const configured = await getConfig( + ctx, + STORAGE_RETENTION_CONFIG_KEYS.archivedArticleDays, + STORAGE_RETENTION_DEFAULTS.archivedArticleDays, + ); + const retentionDays = clampRetentionDays( + args.retentionDays ?? configured, + STORAGE_RETENTION_DEFAULTS.archivedArticleDays, + 14, + ); + const batchSize = clampBatch(args.batchSize, PURGE_BATCH); + const cutoff = Date.now() - retentionDays * DAY_MS; + + // Guard 4: never delete an article that the clustering gold set cites. + const labels = await ctx.db + .query("clusterPairLabels") + .take(CLUSTER_LABEL_GUARD_LIMIT); + if (labels.length >= CLUSTER_LABEL_GUARD_LIMIT) { + // Can't prove the guard set is complete — do nothing rather than guess. + await logPurgeRun(ctx, "articles_archived_detached", 0, true, { + metadata: { skippedReason: "cluster_label_guard_overflow" }, + }); + return { deleted: 0, done: true, skippedReason: "guard_overflow" }; + } + const labeledArticleIds = new Set(); + for (const label of labels) { + labeledArticleIds.add(label.leftArticleId); + labeledArticleIds.add(label.rightArticleId); + } + + // One `archivedAt` watermark per entry in ARCHIVED_REASONS, same order. + // + // Deleting a row removes it from the index, so the head advances by itself + // whenever a batch deletes something. The hazard is a batch that deletes + // NOTHING: rows held back by the guards below (requeued articles, or + // articles cited by the clusterPairLabels gold set) stay at the head and get + // re-read on every run. Once enough of them accumulate to fill a batch the + // purge reports `deleted === 0`, concludes it is done, and can never reach + // the rows behind them. Advancing the watermark past a stuck batch is what + // guarantees forward progress. + const cursors = ARCHIVED_REASONS.map((_, index) => args.cursors?.[index] ?? 0); + + let deleted = 0; + let deletedEmbeddings = 0; + let scanned = 0; + let protectedByLabel = 0; + let anyCandidates = false; + + for (const [index, reason] of ARCHIVED_REASONS.entries()) { + if (deleted >= batchSize) break; + const from = cursors[index] ?? 0; + const candidates = await ctx.db + .query("articles") + .withIndex("by_archived_reason", (q) => + q + .eq("archivedReason", reason) + .gte("archivedAt", from) + .lt("archivedAt", cutoff), + ) + .take(batchSize - deleted); + scanned += candidates.length; + if (candidates.length > 0) anyCandidates = true; + const deletedBefore = deleted; + + for (const article of candidates) { + // Guards 1-2, re-asserted on the row itself. + if (article.status !== "archived") continue; + if (article.eventId !== undefined) continue; + if (labeledArticleIds.has(article._id)) { + protectedByLabel++; + continue; + } + + const embeddings = await ctx.db + .query("articleEmbeddings") + .withIndex("by_article", (q) => q.eq("articleId", article._id)) + .collect(); + for (const row of embeddings) { + await ctx.db.delete(row._id); + deletedEmbeddings++; + } + await ctx.db.delete(article._id); + deleted++; + } + + // Only step over a batch that made no progress. Skipping is safe here + // precisely because these rows are permanently protected — and confining + // the +1ms step to the stuck case keeps it from stepping over rows that + // merely share a millisecond with the last deleted row. + if (deleted === deletedBefore && candidates.length > 0) { + const last = candidates[candidates.length - 1]!; + cursors[index] = (last.archivedAt ?? from) + 1; + } + } + + // Finished only when no reason had anything left in range — not merely when + // this particular batch deleted nothing. + const done = !anyCandidates; + await logPurgeRun(ctx, "articles_archived_detached", deleted, done, { + counters: { scanned, deletedEmbeddings, protectedByLabel }, + metadata: { retentionDays, cutoff, cursors: JSON.stringify(cursors) }, + }); + + if (!done) { + await ctx.scheduler.runAfter( + 0, + internal.retention.purgeArchivedDetachedArticles, + { retentionDays, batchSize, cursors }, + ); + } + return { deleted, deletedEmbeddings, scanned, done }; + }, +}); diff --git a/packages/backend/convex/summarization.ts b/packages/backend/convex/summarization.ts index 19548d8..1e1b291 100644 --- a/packages/backend/convex/summarization.ts +++ b/packages/backend/convex/summarization.ts @@ -656,6 +656,10 @@ export const startSummaryJob = internalMutation({ _id: job._id, eventId: job.eventId, attempts: job.attempts + 1, + // Needed by the rate-limit deferral ceiling: because deferring refunds + // the attempt, `attempts` cannot measure how long a job has been stuck, + // so the caller measures elapsed time since it was first queued. + requestedAt: job.requestedAt, }, }; }, @@ -1395,20 +1399,46 @@ export const markSummaryJobBlockedVerbatim = internalMutation({ }, }); +/** + * Push a job back onto the queue without failing it — used for backpressure + * (AI budget exhausted, provider 429). The work never ran, so nothing about + * the job is wrong and it should not count against `maxAttempts`. + * + * `runId` + `refundAttempt` are optional and opt-in: when the caller already + * leased the job via `startSummaryJob` (which increments `attempts`), passing + * both rolls that increment back so a rate-limited job is not permanently + * given up after three provider 429s. Callers that defer *before* leasing + * (budget check) omit them and keep the previous behaviour exactly. + */ export const deferSummaryJob = internalMutation({ args: { jobId: v.id("eventSummaryJobs"), reason: v.string(), retryAfterMs: v.number(), + runId: v.optional(v.string()), + refundAttempt: v.optional(v.boolean()), }, - handler: async (ctx, { jobId, reason, retryAfterMs }) => { + handler: async ( + ctx, + { jobId, reason, retryAfterMs, runId, refundAttempt }, + ) => { const job = await ctx.db.get(jobId); if (!job || job.status === "succeeded" || job.status === "skipped") { return { updated: false as const }; } + // Only refund the attempt we actually own — if another worker has since + // re-leased the job, leave its attempt counter alone. + const ownsLease = + job.status === "processing" && + runId !== undefined && + job.processingRunId === runId; + const attempts = + refundAttempt && ownsLease ? Math.max(0, job.attempts - 1) : job.attempts; + await ctx.db.patch(jobId, { status: job.status === "failed" ? "failed" : "queued", + attempts, processingRunId: undefined, leaseExpiresAt: undefined, nextAttemptAt: Date.now() + Math.max(60_000, retryAfterMs), @@ -1416,7 +1446,7 @@ export const deferSummaryJob = internalMutation({ updatedAt: Date.now(), }); - return { updated: true as const }; + return { updated: true as const, attempts }; }, }); diff --git a/packages/backend/convex/summarizationNode.ts b/packages/backend/convex/summarizationNode.ts index 6819596..80248cf 100644 --- a/packages/backend/convex/summarizationNode.ts +++ b/packages/backend/convex/summarizationNode.ts @@ -7,7 +7,7 @@ import type { ActionCtx } from "./_generated/server"; import type { Id } from "./_generated/dataModel"; import { internal } from "./_generated/api"; import { shutdownPostHog } from "./lib/openai"; -import { callLLM } from "./lib/aiCall"; +import { callLLM, isRateLimitError } from "./lib/aiCall"; import { fetchArticleBodyText } from "./lib/articleExtraction"; import { buildEventSummaryPrompt, @@ -68,6 +68,17 @@ const DEFAULT_BODY_FETCH_CONCURRENCY = 8; // fraction of the held time. Both are overridable via config. const DEFAULT_BODY_FETCH_TIMEOUT_MS = 12_000; const JOB_LEASE_TTL_MS = 10 * 60 * 1000; +// How long a rate-limited (429) job waits before it is eligible again. Free-tier +// quota windows are per-minute *and* per-day, so retrying in seconds just burns +// billed action time on another 429; wait out the window instead. +const RATE_LIMIT_DEFER_MS = 45 * 60 * 1000; +// Ceiling on how long a job may keep deferring on rate limits before it is +// allowed to fail normally. Deferral refunds the attempt, so without this a +// permanently rate-limited job retries forever and never shows up in any +// failure metric. ~24h is well past any daily quota reset: still rate limited +// after a full day means the quota is genuinely too small, which is exactly the +// thing an operator needs told. +const RATE_LIMIT_DEFER_CEILING_MS = 24 * 60 * 60 * 1000; const BASE_RETRY_DELAY_MS = 5 * 60 * 1000; const JOB_STAGGER_MS = 8000; const SUMMARY_WORD_LIMITS = { @@ -324,8 +335,7 @@ function retryDelayMs(attempts: number): number { * the errors worth retrying on the fallback model rather than failing. */ function isQuotaError(error: unknown): boolean { - const message = error instanceof Error ? error.message : String(error); - return /\b429\b|RESOURCE_EXHAUSTED|quota/i.test(message); + return isRateLimitError(error); } /** @@ -384,6 +394,13 @@ async function generateSummaryWithModel( eventId, }, runtime: ctx, + // COST: callLLM's default retry loop sleeps in-process between attempts, + // and Convex bills that sleep as action compute. A 429 here is not a + // transient blip we can wait out inside the job — the free-tier quota is + // gone for minutes to hours. Let it bubble out on the first attempt so + // processSummaryJob can defer the job (backpressure) instead of paying + // to sit in a sleep. Other callLLM callers keep the default maxRetries. + maxRetries: 1, }); inputTokens += response.usage.inputTokens; @@ -772,6 +789,9 @@ async function verifySummaryGrounding( ], context: { callType: "event_summary", eventId }, runtime: ctx, + // Advisory pass — on failure we fall through to full-candidate + // entailment, so an in-process retry sleep buys nothing but billed time. + maxRetries: 1, }); const vectors = embeddingResponse.result; if (vectors && vectors.length === sentences.length + articles.length) { @@ -825,6 +845,9 @@ async function verifySummaryGrounding( ], context: { callType: "event_summary", eventId }, runtime: ctx, + // Same reasoning as the summary call: a 429 should defer the job, not be + // slept through on Convex's billed clock. + maxRetries: 1, }); if (!entailment.result) { throw new Error( @@ -1196,7 +1219,17 @@ export const processSummaryJob = internalAction({ skipped: boolean; budgetExhausted: boolean; }> => { - const paused = await ctx.runQuery(internal.config.isPipelinePaused, {}); + // COST: every runQuery is a billed round trip that extends this action's + // wall clock. These three are side-effect-free reads whose results are all + // needed before any work starts, so they run concurrently — one round trip + // of latency instead of three. Short-circuit precedence below is unchanged + // (paused > budget > lease). + const [paused, settings, budget] = await Promise.all([ + ctx.runQuery(internal.config.isPipelinePaused, {}), + loadSummarySettings(ctx, {}), + ctx.runQuery(internal.aiBudget.checkBudget, {}), + ]); + if (paused) { console.log("[summarization] Pipeline paused — skipping job"); return { @@ -1208,11 +1241,9 @@ export const processSummaryJob = internalAction({ }; } - const settings = await loadSummarySettings(ctx, {}); const runId = randomUUID(); let budgetExhausted = false; - const budget = await ctx.runQuery(internal.aiBudget.checkBudget, {}); if (!budget.allowed) { budgetExhausted = true; await ctx.runMutation(internal.summarization.deferSummaryJob, { @@ -1591,6 +1622,42 @@ export const processSummaryJob = internalAction({ } catch (error) { const message = error instanceof Error ? error.message : "Unknown summarization error"; + + // Backpressure, not failure: a provider 429 means the request never + // reached the model, so nothing about this event is wrong. Failing it + // burned one of only `maxAttempts` tries and was the main reason most + // events never got summarized at all. Defer well past the quota window + // and refund the attempt this run consumed. + // + // Bounded, though: because the deferral refunds the attempt, a job that is + // rate limited forever would retry forever and never appear in any failure + // metric — an invisible stall rather than a visible problem. Past the + // ceiling, stop refunding and let it fail through the normal path so queue + // health and error-rate alerting can see it. + const rateLimitedForMs = Date.now() - job.requestedAt; + if ( + isRateLimitError(error) && + rateLimitedForMs < RATE_LIMIT_DEFER_CEILING_MS + ) { + console.warn( + `[summarization] Rate limited on event ${job.eventId} — deferring (attempt refunded): ${message}`, + ); + await ctx.runMutation(internal.summarization.deferSummaryJob, { + jobId: job._id, + runId, + refundAttempt: true, + reason: `rate_limited: ${message}`, + retryAfterMs: RATE_LIMIT_DEFER_MS, + }); + return { + processed: true, + succeeded: false, + failed: false, + skipped: true, + budgetExhausted, + }; + } + console.error( `[summarization] Failed to summarize event ${job.eventId}: ${message}`, ); From 7150ca2af8f3697f6b54f56de748569914751669 Mon Sep 17 00:00:00 2001 From: flavius Date: Sat, 1 Aug 2026 16:26:46 +0300 Subject: [PATCH 2/2] fix(convex): keep rate-limit helper out of the "use node" boundary MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit migrations.ts imported isRateLimitError from lib/aiCall.ts, which is a "use node" module. Convex bundles every module without that directive for the V8 runtime, so the import dragged posthog-node's node built-ins into the V8 bundle and `convex codegen` failed with "Could not resolve node:async_hooks". Type checking does not catch this — it only surfaces at codegen, which is why it reached CI. Extract the predicate into a dependency-free lib/rateLimitError module that either runtime can import; aiCall re-exports it so existing callers are unchanged. Also commits the regenerated _generated/api.d.ts for the new retention and migration functions. Co-Authored-By: Claude Opus 5 --- packages/backend/convex/_generated/api.d.ts | 2 + packages/backend/convex/lib/aiCall.ts | 25 +++-------- packages/backend/convex/lib/rateLimitError.ts | 45 +++++++++++++++++++ packages/backend/convex/migrations.ts | 4 +- 4 files changed, 56 insertions(+), 20 deletions(-) create mode 100644 packages/backend/convex/lib/rateLimitError.ts diff --git a/packages/backend/convex/_generated/api.d.ts b/packages/backend/convex/_generated/api.d.ts index fe19f95..9ee0c16 100644 --- a/packages/backend/convex/_generated/api.d.ts +++ b/packages/backend/convex/_generated/api.d.ts @@ -55,6 +55,7 @@ import type * as lib_politeFetch from "../lib/politeFetch.js"; import type * as lib_publicEventPreviews from "../lib/publicEventPreviews.js"; import type * as lib_quizHelpers from "../lib/quizHelpers.js"; import type * as lib_rateLimit from "../lib/rateLimit.js"; +import type * as lib_rateLimitError from "../lib/rateLimitError.js"; import type * as lib_retentionPolicy from "../lib/retentionPolicy.js"; import type * as lib_romanian from "../lib/romanian.js"; import type * as lib_sourceBias from "../lib/sourceBias.js"; @@ -144,6 +145,7 @@ declare const fullApi: ApiFromModules<{ "lib/publicEventPreviews": typeof lib_publicEventPreviews; "lib/quizHelpers": typeof lib_quizHelpers; "lib/rateLimit": typeof lib_rateLimit; + "lib/rateLimitError": typeof lib_rateLimitError; "lib/retentionPolicy": typeof lib_retentionPolicy; "lib/romanian": typeof lib_romanian; "lib/sourceBias": typeof lib_sourceBias; diff --git a/packages/backend/convex/lib/aiCall.ts b/packages/backend/convex/lib/aiCall.ts index 5946b3d..75d3739 100644 --- a/packages/backend/convex/lib/aiCall.ts +++ b/packages/backend/convex/lib/aiCall.ts @@ -6,6 +6,12 @@ import type { ActionCtx } from "../_generated/server"; import { calculateCost, calculateCostWithCachedInput } from "../aiBudget"; import { getLLMClient, isPostHogInstrumented } from "./openai"; import { buildChatTuningParams, providerForModel } from "./modelRouting"; +// Rate-limit detection lives in a dependency-free module so non-"use node" +// modules (e.g. migrations.ts) can share it without dragging the Node-only +// SDK imports above into the V8 bundle. +import { errorStatus, isRateLimitError } from "./rateLimitError"; + +export { isRateLimitError }; export type AICallType = | "fact_extraction" @@ -85,12 +91,6 @@ function sleep(ms: number): Promise { return new Promise((resolve) => setTimeout(resolve, ms)); } -function errorStatus(error: unknown): number | undefined { - const candidate = error as { status?: unknown; code?: unknown }; - if (typeof candidate.status === "number") return candidate.status; - if (typeof candidate.code === "number") return candidate.code; - return undefined; -} function isRetryableError(error: unknown): boolean { const status = errorStatus(error); @@ -115,19 +115,6 @@ function isRetryableError(error: unknown): boolean { * * Additive helper: `isRetryableError` and `callLLM` behaviour are unchanged. */ -export function isRateLimitError(error: unknown): boolean { - if (errorStatus(error) === 429) return true; - const message = - error instanceof Error - ? error.message - : typeof error === "string" - ? error - : ""; - return ( - message.length > 0 && - /\b429\b|RESOURCE_EXHAUSTED|rate[ _-]?limit|quota/i.test(message) - ); -} function isFatalError(error: unknown): boolean { const status = errorStatus(error); diff --git a/packages/backend/convex/lib/rateLimitError.ts b/packages/backend/convex/lib/rateLimitError.ts new file mode 100644 index 0000000..2f56962 --- /dev/null +++ b/packages/backend/convex/lib/rateLimitError.ts @@ -0,0 +1,45 @@ +/** + * Provider rate-limit detection, kept deliberately dependency-free. + * + * This lives apart from `lib/aiCall.ts` because that module is `"use node"` + * (it pulls in the OpenAI/PostHog SDKs, which import node built-ins). Convex + * bundles every module WITHOUT a `"use node"` directive for the V8 runtime, so + * a plain query/mutation module that imports from aiCall fails the bundle with + * "Could not resolve node:async_hooks" — at `convex codegen` time, which type + * checking alone does not catch. + * + * Both runtimes need the same answer to "was this a rate limit?", so the + * predicate lives here with no imports at all and is safe from either side. + */ + +export function errorStatus(error: unknown): number | undefined { + if (typeof error !== "object" || error === null) return undefined; + const candidate = error as { status?: unknown; code?: unknown }; + if (typeof candidate.status === "number") return candidate.status; + if (typeof candidate.code === "number") return candidate.code; + return undefined; +} + +/** + * True for provider 429s / quota exhaustion. + * + * Matches both a live SDK error object and the flattened message string that + * `callLLM` hands back to its callers — by the time `processSummaryJob` sees a + * failure, the error object is gone and only text like + * `"429 status code (no body)"` survives. Stored `lastError` strings on + * `eventSummaryJobs` are the same shape, which is why the requeue migration can + * reuse this. + */ +export function isRateLimitError(error: unknown): boolean { + if (errorStatus(error) === 429) return true; + const message = + error instanceof Error + ? error.message + : typeof error === "string" + ? error + : ""; + return ( + message.length > 0 && + /\b429\b|RESOURCE_EXHAUSTED|rate[ _-]?limit|quota/i.test(message) + ); +} diff --git a/packages/backend/convex/migrations.ts b/packages/backend/convex/migrations.ts index 37d104b..79344f8 100644 --- a/packages/backend/convex/migrations.ts +++ b/packages/backend/convex/migrations.ts @@ -24,7 +24,9 @@ import { } from "./lib/userProfile"; import { deleteByEventIndex, EVENT_CHILD_TABLES } from "./singletonCleanup"; import { truncateThirdPartySnippet } from "./lib/compliance"; -import { isRateLimitError } from "./lib/aiCall"; +// Deliberately NOT from "./lib/aiCall": that module is "use node", and this one +// is not, so importing across the boundary breaks the V8 bundle at codegen time. +import { isRateLimitError } from "./lib/rateLimitError"; import { syncPublicEventPreview } from "./lib/publicEventPreviews"; const MAX_FACT_EXTRACTION_ATTEMPTS = 3;