feat: stream reset for incremental syncs

[anaselmhamdi]

Context

Incremental streams derive their watermark from the last SUCCEEDED StreamJob (last_run = last_successful_job.created_at). Once that chain exists there is no way to say "forget the watermark, re-pull the whole stream, and replace the table" — the only escape hatch today is manually deleting backend rows.

This adds a stream reset: one run that re-fetches the stream from scratch and replaces the destination table, after which incremental resumes normally from that run.

Design

Everything hangs off a single config field, source.reset. That choice is what kept the change small: init_job() runs in the parent before the producer and consumer are submitted, and both are handed the same bizon_config / config objects — so the flag is resolved once and reaches both sides with no signature changes.

Three triggers, all converging on that flag:

bizon run config.yml --reset                    # one-shot, manual
bizon stream reset config.yml                   # queued in the backend, consumed by the next run
bizon stream reset config.yml --cancel          # withdraw it
bizon stream reset config.yml --stream deals    # pick the stream, for templated configs

source: { sync_mode: incremental, reset: true }

The bizon stream reset form is the important one. A --reset flag only helps someone typing the command by hand — if your pipeline is a cron/Airflow job whose command is hardcoded as bizon run myconfig.yml, you'd have to edit the job spec, run once, then edit it back. The marker form writes a row to a new stream_resets table, and an unchanged bizon run myconfig.yml picks it up.

Granularity is per stream, not per pipeline. The marker is keyed on (name, source_name, stream_name) — the same triple as get_last_successful_stream_job — so a reset is exactly as scoped as the watermark it overrides, and resetting one stream never affects another under the same pipeline name. Multi-stream configs (the streams: block) can't be reset at all: they require sync_mode: stream, while reset is incremental-only.

What the run does differently:

• Producer skips the watermark lookup and calls get() instead of get_records_after().
• Destination needs no reset-specific code. SyncMetadata.from_bizon_config() maps a reset onto sync_mode: full_refresh, so destinations replace their table through their existing full-refresh path. This is the sync mode of the materialization, not of the job — which is why the job row can stay incremental while the destination still replaces.
• Job row stays incremental, so get_last_successful_stream_job picks the reset run up as the next watermark automatically.

Crash safety — the part that made the flag alone insufficient: every reset job has a consumed stream_resets row bound to it, so a retry recognises the in-flight job as a reset instead of silently degrading into an append. _ensure_clean_temp_table() drops a stale temp table once per reset, but only when the job has written no destination cursor yet — otherwise it would discard iterations the resuming producer will not re-fetch.

Nothing validates the stream name in bizon stream reset (the source is never instantiated), so a typo — most likely via --stream — would queue a reset that silently never fires. The command warns when the target stream has no previous successful job. It's a warning, not an error, since a stream can legitimately have run without ever succeeding.

An earlier revision of this PR exposed a separate destination_sync_mode property and gated it behind an allowlist, which limited reset to bigquery + logger. Collapsing it into the factory removed the property, four destination call sites and the allowlist, and made reset work everywhere. Verified end-to-end that the file destination — previously rejected outright — now replaces its output on a reset (5 rows, not 10).

Verification

Ran the real CLI end-to-end across four runs (dummy → logger, sqlite backend):

1. First run full-fetches (5 records), job succeeds
2. Second run uses the watermark — fetching records after 12:28:23
3. bizon stream reset → third run logs the reset, cancels the stale job, re-fetches all 5
4. Fourth run resumes incremental from the reset job's timestamp (12:28:42)

Inspected the backend afterwards: marker consumed exactly once and bound to the reset job; every job row still incremental. Also confirmed the command itself only does one INSERT — job and cursor rows are untouched, nothing is deleted, and no pipeline runs.

Per-stream isolation verified both directions: resetting plants left creatures on the normal incremental path, then --stream creatures fired correctly.

47 new tests across backend CRUD, reset resolution, producer dispatch, BigQuery staging/finalize, config validation, and the CLI. Full suite: 266 passed. The 28 failures are all pre-existing on a clean tree (verified by stashing) — they need real Postgres/GCP credentials.

Postgres-backed tests were run against the repo's docker-compose.yml (postgres:14, same as CI): all 14 my_pg_backend backend tests pass, including the 3 new stream_resets ones. With Postgres up the full suite is 283 passed; the 14 remaining failures all require real GCP credentials (live BigQuery) and fail identically on a clean tree.

Notes for review

Two decisions worth a second opinion:

• bigquery_streaming (v1) is excluded. It has no finalize() and no staging table, so it appends even on a plain full refresh — its full-refresh support is already broken independently of this PR. A reset there is a config validation error rather than silent duplication. Every other destination works.
• A reset not already in flight force-creates a fresh job, cancelling any running one. Correct for reset semantics, but a reset of a huge stream that keeps crashing before writing any cursor would restart from zero each time.

Backwards compatibility. Checked out origin/main's tests/ and ran them against this branch: all 233 pass, with no failures outside the GCP-credential set. Only 9 lines are removed across bizon/, and each reduces to the original when reset is off (X or (False and …) → X, etc.); the BigQuery destination diff is purely additive, with temp_table_id and finalize() byte-identical to main. source.reset defaults to False, and stream_resets is a new table rather than a column on stream_jobs precisely so create_all_tables() handles it with no migration.

One deliberate exception: AbstractBackend gains 5 abstract methods, so an out-of-tree backend subclass would fail at instantiation. BackendFactory is a closed enum dispatch with no plugin path, so there is no supported way to have written one — and failing loudly at construction beats a backend that silently can't reset. Flagging it rather than burying it.

Also, two pre-existing bugs found while working here, deliberately not touched:

• StreamJob.created_at uses default=datetime.now(tz=UTC) — a value bound at import time, not a callable — so every job inserted by a process shares the module-load timestamp. Since that value is the incremental watermark, this is load-bearing for correctness.
• destination.py:146-148 returns early when the buffer is empty, skipping finalize() — so a full refresh (and therefore a reset) over an empty source never replaces the main table.

🤖 Generated with Claude Code