Added archived article state, pipeline run logs, and pipeline alerts … - #31
Merged
Conversation
…in schema.ts (line 478). Added stale singleton archive pipeline with idempotent per-event cleanup, child-row cleanup, lock checks, self-rescheduling, admin trigger, and structured logs in singletonCleanup.ts (line 315). Recalibrated vector budget defaults and per-search byte estimation in vectorSearchBudget.ts (line 345). Tuned clustering/merge/recluster limits, seed caps, no-candidate fast paths, and run logging in clustering.ts (line 4599). Added admin-only diagnostics, alerts, manual job triggers, and calibration mutation in pipeline.ts (line 93). Added /admin/pipeline with funnel, queues, budget, logs, stuck events, archive stats, alerts, manual triggers, and calibration form in admin.pipeline.tsx (line 21). Added operator migration notes in pipeline-stabilization-migration.md (line 1).
Contributor
|
Warning Rate limit exceeded
You’ve run out of usage credits. Purchase more in the billing tab. ⌛ How to resolve this issue?After the wait time has elapsed, a review can be triggered using the We recommend that you space out your commits to avoid hitting the rate limit. 🚦 How do rate limits work?CodeRabbit enforces hourly rate limits for each developer per organization. Our paid plans have higher rate limits than the trial, open-source and free plans. In all cases, we re-allow further reviews after a brief timeout. Please see our FAQ for further information. ℹ️ Review info⚙️ Run configurationConfiguration used: Path: .coderabbit.yaml Review profile: ASSERTIVE Plan: Pro Run ID: 📒 Files selected for processing (7)
WalkthroughThis PR adds comprehensive pipeline observability and automated stale event archival. It introduces tables for run logs and alerts, implements health rule evaluation, improves vector search budget estimation via calibration, and adds a stale singleton cleanup system that archives events based on age and metadata thresholds. Articles gain archived status support, configuration gains validation and new tuning defaults, and crons accelerate critical jobs to activate the systems. An admin dashboard provides monitoring and control. ChangesPipeline Observability and Stale Event Archival
Sequence Diagram(s)sequenceDiagram
participant Cron as "15-min Cron"
participant CheckAlerts as "checkPipelineAlerts"
participant QueryLogs as "Pipeline Logs/Runs"
participant RuleEval as "Rule Evaluation"
participant InsertLog as "insertRunLog"
participant UpsertAlert as "upsertPipelineAlert"
Cron->>CheckAlerts: trigger periodic check
CheckAlerts->>QueryLogs: fetch last 24h pipeline logs
CheckAlerts->>QueryLogs: fetch recent vector budget runs
CheckAlerts->>RuleEval: evaluate fallback/publish/budget/error rules
RuleEval->>CheckAlerts: list of triggered alerts
CheckAlerts->>InsertLog: log check execution with status
loop for each alert
CheckAlerts->>UpsertAlert: write triggered alert (if no existing)
end
CheckAlerts->>Cron: return check result
sequenceDiagram
participant Admin as "Admin User"
participant Dashboard as "Admin Dashboard"
participant TriggerAPI as "triggerArchive<br/>StaleSingletonEvents"
participant ArchiveAction as "archiveStaleSingleton<br/>Events"
participant LockSvc as "Pipeline Lock Service"
participant CandidateQ as "getStaleSingleton<br/>Candidates"
participant EventMut as "archiveSingletonEvent"
participant DB as "Database"
Admin->>Dashboard: click Archive button
Dashboard->>TriggerAPI: call triggerArchiveStaleSingletonEvents()
TriggerAPI->>ArchiveAction: schedule with autoContinue=true
ArchiveAction->>LockSvc: acquire ARCHIVE_LOCK
LockSvc->>ArchiveAction: lock acquired
ArchiveAction->>CandidateQ: fetch stale events in batches
CandidateQ->>ArchiveAction: candidate list
loop process batch
ArchiveAction->>EventMut: archiveSingletonEvent(eventId)
EventMut->>DB: update articles (archive/requeue)
EventMut->>DB: delete event and related rows
EventMut->>ArchiveAction: return counters
end
ArchiveAction->>DB: insertRunLog(metrics, counters)
ArchiveAction->>LockSvc: release ARCHIVE_LOCK
ArchiveAction->>Admin: show result toast
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Possibly related PRs
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Actionable comments posted: 13
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@apps/web/src/routes/admin.pipeline.tsx`:
- Around line 352-364: The button that calls acknowledgeAlert (the Button with
onClick calling acknowledgeAlert({ alertId: alert._id })) only shows a visual
toast; add an ARIA live region to announce success/failure by introducing a
local state (e.g., ackStatus and setAckStatus) and updating it in the promise
resolution and rejection handlers (where you currently call toast.success): set
ackStatus to a descriptive message on success and an error message on failure;
render a visually-hidden element near the Button with role="status",
aria-live="polite" (and aria-atomic="true") that outputs ackStatus so screen
readers will receive the acknowledgement status. Ensure the state is
reset/cleared as appropriate after a short time if needed.
- Around line 306-322: The calibration form lacks an ARIA live region to
announce success/failure; add a visually-hidden status element (e.g., a <div>
with aria-live="polite" and role="status" and a hidden CSS utility class) inside
the same component near the Input/Button block and wire it to a new state
variable (e.g., calibrationStatus / setCalibrationStatus) so that
saveCalibration sets "Calibration saved" or the error text on success/failure;
ensure the live region content is updated whenever saveCalibration completes and
is updated/cleared when the form changes (e.g., on setObservedQgbInput) so
screen readers receive the status updates.
- Around line 456-458: The UI currently slices JSON.stringify(log.counters)
which can cut JSON mid-token and confuse operators; instead, render a safe,
human-readable summary by extracting Object.entries(log.counters) and showing
the first few key:value pairs (e.g., join the first 3 entries as "key: value"
with an ellipsis if more), and wire the full JSON.stringify(log.counters, null,
2) into a hover/tooltip or title attribute for full inspection; update the cell
rendering that currently uses JSON.stringify(log.counters).slice(0, 160) to use
this key-based summary and tooltip approach so output is never mid-syntax and
remains debuggable.
In `@packages/backend/convex/clustering.ts`:
- Around line 284-291: The extra
ctx.runQuery(internal.vectorSearchBudget.calibratePerSearchBytes, {}) calls
should be removed from hot paths by fetching calibration once at job start and
passing the resulting perSearchBytes into the existing job-level state (e.g.,
add a perSearchBytes field on JobMetrics or a small JobContext), then use that
cached value in reserveVectorSearch, consumeVectorSearchReservation, and
flushJobMetrics (refer to the functions reserveVectorSearch,
consumeVectorSearchReservation, flushJobMetrics and the
internal.vectorSearchBudget.calibratePerSearchBytes query); ensure you call
calibratePerSearchBytes only once at job initialization, populate
JobMetrics/JobContext.perSearchBytes, and replace subsequent ctx.runQuery calls
with reads of that field.
In `@packages/backend/convex/crons.ts`:
- Around line 182-194: The three daily maintenance crons clustered at
04:30–04:55 UTC can cause contention; change scheduling or add runtime
monitoring for the job named "cleanup-pipeline-run-logs" so we can observe
impact: either stagger this crons.daily invocation away from the other two (move
hourUTC/minuteUTC in the crons.daily call) or add instrumentation around
internal.pipeline.cleanupPipelineRunLogs to record start/end timestamps,
execution duration, row counts, and any Convex write latencies/error rates to
logs/metrics so we can monitor load during the window and adjust later.
In `@packages/backend/convex/pipeline.ts`:
- Around line 126-180: The current handlers (in pipeline.ts) derive totals from
.take(5000) snapshots (e.g., the queries assigned to articles, processingEvents,
publishedEvents, previews and the counts inside currentQueues) which underreport
when results exceed 5,000; replace these bounded reads with true counts by
implementing a paginatedCount utility that repeatedly queries with a safe page
size and a cursor/last-key until no more rows (or use a DB-provided
count/aggregate if available), then call that utility instead of relying on
.take(5000) for totals and queue counts; apply the same fix to
getArchivedArticleStats usage referenced in the comment (the by_archived_reason
queries) so last24h/last7d use full counts rather than truncated snapshots.
- Around line 645-653: The current count only includes events with lastArticleAt
older than cutoff; include processing events where lastArticleAt is null but
firstPublishedAt is older than cutoff by adjusting the logic around cutoff:
compute cutoff as shown, then count both sets—(1) events from
ctx.db.query("events").withIndex("by_status_last_article_at", ...) where status
=== "processing" and lastArticleAt < cutoff, and (2) events where status ===
"processing" and lastArticleAt is null and firstPublishedAt < cutoff (e.g., run
a second query filtering firstPublishedAt or use a predicate equivalent of
lastArticleAt ?? firstPublishedAt < cutoff) and sum the two counts instead of
only using the existing .withIndex(...) result length.
- Around line 497-507: The code currently selects the oldest matching alert by
sorting ascending and taking [0]; change it to pick the most recent prior alert
instead: filter logs for jobName === "checkPipelineAlerts" and numeric
gauges.stuckProcessingOver72h, then sort by startedAt descending (e.g. (a,b) =>
b.startedAt - a.startedAt) and take [0] as the previous baseline used to compute
previousStuckProcessingOver72h (update the variable name if helpful from
oldestAlertGauge to recentAlertGauge to reflect the logic).
In `@packages/backend/convex/schema.ts`:
- Around line 898-914: pipelineAlerts lacks an index to find unresolved alerts
by code, causing upsertPipelineAlert to miss a newer unresolved row; add a
compound index on (code, resolvedAt) (e.g. name it "by_code_resolved" or
similar) to the pipelineAlerts table definition and update upsertPipelineAlert
to query that index with resolvedAt === undefined to reliably detect existing
active alerts for a code.
In `@packages/backend/convex/singletonCleanup.ts`:
- Around line 431-437: The current scan always fetches the first
settings.batchSize * 2 rows and can starve eligible rows; change the loop that
calls internal.singletonCleanup.getStaleSingletonCandidates to paginate by
passing/receiving a continuation token or last-seen key so each runQuery fetches
the next window (use symbols candidates and getStaleSingletonCandidates to
locate the call), and update the selection logic to skip ineligible rows while
continuing to fetch additional pages until you either accumulate the desired
number of eligible items or storage returns no more rows; also compute hasMore
from whether the storage response indicated more candidates (e.g., response
length or continuation token) rather than from the eligible array, and respect
settings.batchSize when deciding stop conditions.
- Around line 296-300: The archive path leaves a dangling article.eventId
reference while deleting the event; update the patch call that sets
status/archivedAt/archivedReason (the ctx.db.patch on article._id) to also clear
eventId (e.g., set eventId: null or remove the field) so the article no longer
points to the deleted event, and apply the same fix to the other archive branch
around the event deletion (the other ctx.db.patch/delete pairing) to ensure no
archived article retains an eventId reference.
- Around line 405-510: The try/finally in archiveStaleSingletonEvents misses
logging when an error occurs; wrap the main work in a try/catch (inside the
existing try so the finally still runs) and in the catch call logArchiveRun(ctx,
{ runId, startedAt, status: "error", reason: error.message or stringified error,
counters, gauges: { /* include hasMore/articleAction if available */ },
metadata: archiveSettingsMetadata(settings) }) before re-throwing the error, so
failures are recorded; reference functions/variables:
archiveStaleSingletonEvents, logArchiveRun, runId, startedAt, counters,
archiveSettingsMetadata, and ensure the existing finally still calls
ctx.runMutation(internal.ingestion.releasePipelineLock, { key: ARCHIVE_LOCK_KEY,
owner }).
In `@packages/backend/convex/vectorSearchBudget.ts`:
- Around line 134-141: The current 24h aggregation in vectorSearchBudget.ts uses
ctx.db.query("vectorSearchRuns").withIndex("by_createdAt", ...).take(5000) which
truncates results and undercounts vectorSearches during busy periods; replace
the single .take(5000) call with a batched/paginated fetch over the index (e.g.,
loop using cursor/skip or the DB client's pagination API) to read all runs with
createdAt >= cutoff and accumulate the sum into vectorSearches before computing
perSearchBytes so the calibration uses the full 24h sample.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: ASSERTIVE
Plan: Pro
Run ID: 17642f41-063d-4e6f-8366-885f592c5e47
⛔ Files ignored due to path filters (2)
apps/web/src/routeTree.gen.tsis excluded by!**/routeTree.gen.tspackages/backend/convex/_generated/api.d.tsis excluded by!**/_generated/**,!**/_generated/**
📒 Files selected for processing (12)
apps/web/src/routes/admin.pipeline.tsxdocs/pipeline-stabilization-migration.mdpackages/backend/convex/clustering.tspackages/backend/convex/config.tspackages/backend/convex/crons.tspackages/backend/convex/enrichment.tspackages/backend/convex/enrichmentNode.tspackages/backend/convex/ingestion.tspackages/backend/convex/pipeline.tspackages/backend/convex/schema.tspackages/backend/convex/singletonCleanup.tspackages/backend/convex/vectorSearchBudget.ts
Merged
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
…in schema.ts (line 478).
Added stale singleton archive pipeline with idempotent per-event cleanup, child-row cleanup, lock checks, self-rescheduling, admin trigger, and structured logs in singletonCleanup.ts (line 315). Recalibrated vector budget defaults and per-search byte estimation in vectorSearchBudget.ts (line 345). Tuned clustering/merge/recluster limits, seed caps, no-candidate fast paths, and run logging in clustering.ts (line 4599). Added admin-only diagnostics, alerts, manual job triggers, and calibration mutation in pipeline.ts (line 93). Added /admin/pipeline with funnel, queues, budget, logs, stuck events, archive stats, alerts, manual triggers, and calibration form in admin.pipeline.tsx (line 21). Added operator migration notes in pipeline-stabilization-migration.md (line 1).
Summary by CodeRabbit
Release Notes
New Features
Documentation