Skip to content

Ingestion Handoff Orchestration (Issue #26)

Ownership split (scraper PR #183)

rag is the sole owner of the ingestion_handoffs and ingestion_jobs lifecycle. The scraper only inserts queued handoff rows (one per (reference_id, payload_hash)) and never writes ingestion_jobs directly. Scraper PR #183 removes its direct ingestion_jobs writes and must not merge before rag's scheduled consumer (below) ships.

Scheduled consumer

An APScheduler job (handoff_consumer, every 60s, app/services/background_jobs.py) drains queued handoffs:

  1. Select up to 10 oldest status='queued' rows (rows with exhausted attempts are skipped).
  2. Claim each with a conditional queued -> running UPDATE — the same pattern as IngestionService.claim_job(). Postgres row locking guarantees one winner; a lost race means another instance owns it, so it is skipped.
  3. For each claimed row, IngestionHandoffService.orchestrate() converts it into an ingestion_jobs row (+ ingestion_audit entry), dedupes against already active/ready jobs, and marks the handoff completed or failed.

The queued ingestion job is then picked up by the existing ingestion worker (drain_queued_ingestion_jobs, every 30s) and executed to ready.

Crash recovery: a handoff claimed queued -> running only leaves running via orchestration's terminal update. If the process restarts in between, a stale handoff reaper (stale_handoff_reaper, every 5m) resets rows stuck in running longer than STALE_HANDOFF_TIMEOUT_MINUTES (default 15) back to queued with reason_code=requeued_stale_running — not failed, since a crash is no evidence about the payload. attempt_count is left untouched (the attempt was already counted when orchestration started).

Lifecycle

ingestion_handoffs.status uses:

  • queued
  • running
  • completed
  • failed

reason_code captures why the transition happened (examples: queued_from_scrape, ingestion_job_queued, ingestion_job_already_active, retry_exhausted).

Failed handoffs are re-queued by the scraper (intentional)

When rag marks a handoff failed (retries exhausted or a non-retryable error), that row stays failed — rag never resurrects it. Instead, the scraper intentionally re-queues the work on its next scrape of the same payload: it re-queues the handoff for the same (reference_id, payload_hash), giving the reference another full attempt budget. This is the designed retry-across-runs loop, not a bug: transient environment failures (broker/db/network outages) heal on the next scrape cycle without rag needing its own failed-handoff reaper. Permanent failures keep failing with the same payload_hash and stay visible as failed rows with a reason_code + error_message until the source document (and therefore its hash) changes.

Status: the scraper-side re-queue (failed -> queued with a fresh attempt budget) ships in scraper PR #183 — until #183 merges, a failed handoff stays failed and is not retried by anyone.

Retry strategy

  • Max attempts: 3 (configurable at service init).
  • Retry only on transient errors (timeouts, connection/rate-limit/503 patterns).
  • Backoff is exponential (base * 2^(attempt-1)); default base is 0 in current implementation to keep sync calls fast.
  • On max-attempt exhaustion, handoff is marked failed with reason_code=retry_exhausted.

Idempotency / dedupe

  • DB-level dedupe: UNIQUE(reference_id, payload_hash) on ingestion_handoffs.
  • Runtime dedupe: if latest ingestion job for a reference is already active/ready, no duplicate job payload is inserted.

Queryability

  • GET /scraping/{source}/handoffs lists handoff records.
  • Supports filters: status, scrape_run_id, pagination (limit, offset).

Sync response additions

POST /scraping/{source}/sync now includes:

  • handoff_queued_count
  • handoff_completed_count
  • handoff_failed_count
  • handoff_skipped_count

Rollback plan (migration 029)

If rollback is required, treat this migration as data-destructive for handoff history only (does not delete references or ingestion_jobs rows).

DROP TRIGGER IF EXISTS trg_update_ingestion_handoff_timestamp ON ingestion_handoffs;
DROP TABLE IF EXISTS ingestion_handoffs;
DROP FUNCTION IF EXISTS update_ingestion_handoff_timestamp();

Post-rollback expectation: - /scraping/{source}/sync still persists references, but handoff orchestration/query endpoint (/handoffs) is unavailable until migration 029 is re-applied.