Data Pipeline Correctness

Degree of freedom: MIXED. Layering and window design [HIGH freedom]; idempotency, atomic writes, overlap lock, and the DoD [LOW freedom — run exactly].

How to reason

1. Observe — job, retry path, write targets, schedule
2. Interpret — at-least-once vs atomic vs contract vs overlap
3. Classify — upsert / window-recompute / quarantine / lock / watermark
4. Severity — retry double-count outranks a missing metric

Worked example

Observe: nightly refreshorderstats does count = count + 1; cron overlapped twice; dashboard totals jumped.
Interpret: at-least-once delivery + non-idempotent delta.
Classify: recompute-and-replace the day window; pgtryadvisory_lock; persist a watermark.
Verify: re-run the same window → identical rows; overlap skipped; pipeline_runs recorded.

Self-critique before reporting

• Idempotent — same-window re-run proven identical, not assumed
• Atomic — mid-fail leaves no half-written batch
• Locked — the scheduled job has an overlap guard
• Right owner — schema/constraints → audit-db-schema; post-hoc corruption hunt → plan-data-integrity

Pipelines fail silently: a retry double-counts, a partial write corrupts a table, a schema drift poisons a dashboard, and nobody notices until the numbers are wrong. This skill bakes correctness in at build time. It complements post-hoc data-integrity audit skills (which detect these after the fact) and the Supabase plugin (DB/Edge Functions/RLS).

When this fires

Any job that moves, transforms, or aggregates data: ingestion/ETL/ELT, scheduled aggregations, edge-function workers, pg_cron jobs, queue consumers, webhook processors, backfills, materialized-view refreshes.

Non-negotiables (the 5 that prevent silent corruption) [LOW freedom — run exactly]

1. Idempotency — running the same job twice must not change the result. Retries, at-least-once queues, and overlapping cron fires are guaranteed, not hypothetical.

- Use INSERT ... ON CONFLICT (natural_key) DO UPDATE (upsert), not blind INSERT. - Derive a deterministic dedup key from the source event, not now() or a random id. - For aggregates: recompute-and-replace a window, or use idempotent deltas — never count = count + 1 on a path that can retry.

1. Atomicity — a job either fully applies or not at all. No half-written batches.

- Wrap multi-row writes in a transaction; stage to a temp/raw table then swap. - A function that writes to 3 tables must not leave 1 of them updated on failure.

1. Data contracts — validate shape at the boundary before trusting input.

- Parse/validate (zod / pydantic / JSON schema) at ingestion; reject or quarantine bad rows, don't any-cast them downstream. - Pin expected columns/types; fail loudly on schema drift instead of silently coercing.

1. Explicit delivery semantics — know and document whether each stage is at-least-once, at-most-once, or exactly-once, and make the consumer match. Most queues/cron are at-least-once → consumers MUST be idempotent (see #1).

1. Observability — a pipeline you can't see is a pipeline that's already broken.

- Emit per-run: rows in / out / rejected, duration, watermark, status. Persist it (a pipeline_runs table or logs), don't just console.log. - Alert on: zero rows when rows expected, reject-rate spike, run overran, run skipped.

Staging architecture (default to 4 layers) [HIGH freedom]

Raw → land source data unchanged, append-only, with ingested_at + source id
Staged → cleaned, typed, validated, deduped (1 row per natural key)
Curated → business entities, joined/enriched, the query surface
Aggregated→ rollups / metrics / materialized views for dashboards

Each layer is rebuildable from the one before it. Never transform-in-place on raw; never let dashboards read raw.

Backfills [HIGH freedom]

• Make jobs parameterized by window (--from, --to / date partition), not "everything since forever". The same code runs the nightly slice and the historical backfill.
• Backfills must be idempotent and chunked (partition by day/range) so a failure resumes, not restarts.
• Use a watermark (last-processed timestamp/id, persisted) for incremental runs; never re-scan the whole source each run.

Failure handling [HIGH freedom]

• Dead-letter bad/failed records to a quarantine table or DLQ with the error + payload; keep the main run moving. Silent try/catch {} that swallows errors is banned.
• Retries: bounded, with backoff; only retry transient errors (network/timeout), never validation failures (they'll just fail again).
• Make partial progress resumable via the watermark, not a full redo.

Anti-patterns (reject on sight) [HIGH freedom]

• Monolithic DAG / mega-function doing fetch+transform+load+notify in one untestable blob → split into testable stages.
• count = count + 1 / balance = balance + x on a retryable path → not idempotent.
• SELECT * into a typed model without a contract → schema drift time bomb.
• N+1 writes in a loop instead of a batched upsert → slow + non-atomic.
• Cron with no overlap guard (job B starts before job A finishes) → double processing. Add a lock / pgtryadvisory_lock / "skip if running".
• Reading dashboards straight off raw ingestion tables.

Supabase / edge-function specifics [LOW freedom — run exactly]

• pg_cron is at-least-once and can overlap under load → make the SQL/function idempotent and guard with an advisory lock.
• Edge-function workers triggered by table inserts: dedupe on the row's natural key; the trigger can fire more than once.
• Heavy aggregation belongs in SQL / materialized views (refresh on a schedule), not in a function looping row-by-row.
• Deploy + verify the function, cron, and any new table/policy on the remote in the same turn — see full-stack-ship-discipline.

Definition of done [LOW freedom — do not skip]

• Re-running the job produces the identical result (idempotency proven, not assumed).
• A mid-run failure leaves no half-applied state (atomicity).
• Bad input is rejected/quarantined with a contract, not silently coerced.
• Incremental runs use a persisted watermark; backfill is windowed + chunked.
• Per-run metrics are emitted and an alert exists for zero-rows / reject-spike / overrun.
• No overlap hazard on scheduled jobs (lock or skip-if-running).

Composes with

• audit-db-schema — the schema/constraints the pipeline writes into.
• supabase-postgres-best-practices — Postgres-level query/index tuning (official Supabase plugin).
• full-stack-ship-discipline — deploy + verify functions/cron/policies on the remote.
• workflow-spec-tdd — spec the contract + test idempotency/edge cases before coding.
• Project-local data-integrity audit skills — post-hoc detection of the failures this prevents.