Incremental ETL + Idempotent Pinecone Upserts (FRED → Mongo → Pinecone)

Summary

Make the data pipeline incremental and idempotent:

• Fetch only new/updated FRED observations.
• Store a watermark in Mongo to avoid reprocessing.
• Upsert vectors to Pinecone in deterministic, duplicate-safe batches.
• Add concurrency limits, retries, and lightweight observability.

Why

• Current full re-ingest is slow and expensive for ~10k+ points.
• Duplicate vectors and long serverless runs create instability.
• This enables faster refreshes, smaller bills, and more reliable RAG.

Scope

• Backend only (backend/src/**).
• ETL, Pinecone adapter, and supporting configs.
• No frontend UI changes.

Implementation Plan

1. Mongo watermark collection

   • Create ingestion_watermarks:

     { _id: "fred", lastIngestedAt: ISODate, etlVersion: string }
     

   • Read lastIngestedAt at start; default far past date when absent.

   • After successful commit, update atomically.

2. Incremental FRED fetch

   • Use lastIngestedAt to request only newer data (date filter).
   • Add ETag / If-Modified-Since headers; handle 304 Not Modified.
   • Exponential backoff + jitter (3–5 retries) for 429/5xx.

3. Deterministic vector IDs + idempotency

   • ID = sha1(${seriesId}:${observationDate}).
   • Skip upsert if Mongo already has this obs and vector hash matches.

4. Batched Pinecone upserts

   • Batch size: 200 (configurable).
   • Concurrency limit via p-limit (QUEUE_CONCURRENCY, default 5).
   • Attach metadata: { seriesId, date, frequency, seasonality, units }.

5. Config & env

   • New envs:

     QUEUE_CONCURRENCY=5
     PINECONE_BATCH_SIZE=200
     ETL_VERSION=v1
     

   • Respect existing PINECONE_* and DB vars.

6. CLI entrypoints

   • npm run etl:incremental → incremental ingest (FRED→Mongo).
   • npm run vectors:upsert → vectorize + Pinecone upsert (incremental).
   • Keep runAll.ts but call these two sequentially.

7. Light observability

   • Log per-run stats: fetched, inserted, updated, skipped, upserted, duration.
   • Warn on cache misses and Pinecone rate limits.

Acceptance Criteria

• ✅ Running npm run etl:incremental twice in a row processes 0 new rows the second time (no duplicates).
• ✅ Pinecone contains no duplicate vectors for the same (seriesId, date).
• ✅ End-to-end incremental run completes ≥60% faster than a full re-ingest on the same dataset.
• ✅ Concurrency/env flags are respected and documented.
• ✅ Unit tests cover: watermark logic, id generation, batcher behavior, retry/backoff.
• ✅ Dry run mode (DRY_RUN=1) logs actions without writing.

Test Plan

• Unit

  • Watermark read/write with atomic update.
  • ID generation is stable for fixed input.
  • Retry logic triggers on mocked 429/5xx, respects max attempts.
  • Batch splitter yields correct sizes and final remainder.

• Integration (mocked)

  • Mock FRED: 304 path and late-arriving data.
  • Mock Pinecone: upsert called with deduped IDs & metadata.

• Manual

  • Seed DB with N records; run incremental twice; verify no change on run 2.
  • Increase FRED data by M entries; run incremental; verify exactly M new vectors.

Risks & Mitigations

• Clock skew / late data: Always allow a small overlap window (e.g., re-query last 3 days) and dedupe by ID.
• Pinecone rate limits: Backoff + concurrency cap; surface warnings.
• Corrupt watermark: Fallback to overlap window + dedupe; alert in logs.

Rollout

1. Land code behind default-safe configs.
2. Run in staging with DRY_RUN; verify logs.
3. Enable in prod; monitor duration and upsert counts for first 3 runs.

Developer Notes (snippets)

Deterministic ID

import crypto from "node:crypto";
export const vecId = (seriesId: string, date: string) =>
  crypto.createHash("sha1").update(`${seriesId}:${date}`).digest("hex");

Batch upsert skeleton

import pLimit from "p-limit";
const limit = pLimit(Number(process.env.QUEUE_CONCURRENCY || 5));
const size = Number(process.env.PINECONE_BATCH_SIZE || 200);

export async function upsertBatches(points, index) {
  for (let i = 0; i < points.length; i += size) {
    const batch = points.slice(i, i + size);
    await limit(() => index.upsert(batch));
  }
}

Overlap window

const OVERLAP_DAYS = 3;
const start = lastIngestedAt ? subDays(lastIngestedAt, OVERLAP_DAYS) : null;

Documentation

• Update README: new scripts, env vars, and incremental behavior.
• Add a brief “Pipeline Overview” diagram in /backend/docs.

---