fix: don't trust state left behind by a killed process (v0.5.2)

[anaselmhamdi]

Four fault-tolerance fixes found operating ~63 batch pipelines against BigQuery on Kubernetes (Argo Workflows), where pods are subject to a hard activeDeadlineSeconds kill. A killed process never gets to mark its job failed, and every recovery path trusted the state it left behind.

Ordered by damage done.

1. full_refresh resumed dead jobs — silent data outage

get_or_create_job() resumed any job still in running. Right for incremental; a silent outage for full_refresh: the run continues a stale item list, never reaches the last iteration, so never calls finalize() — and because the job stays running, the next run resumes it too. No attempt ever completes a pass.

Observed in production: a table sat at its week-old contents (142,454 rows) through seven consecutive "successful" runs. Nothing surfaced as failing.

Such a job is now cancelled and started fresh. Resets are structurally unaffected — resolve_reset() returns False for any non-incremental mode, so they never reach this branch and keep their own stream_resets recovery contract.

2. full_refresh never cleared its staging table

Fixing #1 alone would have made things worse, which is why these ship together. Loads always WRITE_APPEND into {table}_temp and finalize() publishes it with a WRITE_TRUNCATE copy, but _ensure_clean_temp_table() only dropped a stale temp table for reset runs. A fresh job would have published its own rows alongside the killed run's — trading stale-but-consistent data for duplicates. (132,602 rows had accumulated in _temp across the seven attempts above, with overlapping and missing cursor ranges.)

The guard's own docstring justified the exclusion with "a reset … is the one incremental case that publishes with WRITE_TRUNCATE" — but plain full_refresh also stages into _temp and also publishes with WRITE_TRUNCATE. It now covers every run that does. Plain incremental stages into {table}_incremental and still appends across runs, unchanged.

3. Concurrent pipelines crashed creating the state tables

create_all_tables() inspects before creating (checkfirst=True), but inspect-then-create is not atomic: pipelines sharing a source — and so a dataset — starting on the same cron all saw a table missing, all issued CREATE TABLE, and every process but one died with 409 Already Exists (DuplicateTable on Postgres). Hit bizon_linear (3 pipelines) and bizon_customerio (6) simultaneously.

Losing that race is not an error. Verified by re-inspecting rather than by catching a dialect-specific exception, since this adapter also serves SQLite and Postgres and must not import the BigQuery client libraries. A genuine failure still raises.

This race was always present for every state table — it is only reachable while a table is missing, which is why it first surfaced when 0.5.0 added stream_resets. It would fire again for anyone bootstrapping a fresh dataset, and again on the next new state table.

4. Unattributable crash resuming a cursor with no pagination

create_destination_cursor() stores a falsy pagination as SQL NULL rather than "{}", while the reader called json.loads() on it unconditionally — so a perfectly successful cursor (get_last_cursor_by_job_id filters on success == True) crashed with the JSON object must be str, bytes or bytearray, not NoneType, naming neither the job, the stream, nor a remedy. That sent one investigation down an API/auth dead end.

Failing is correct: Cursor.update_state() treats an empty pagination as source exhausted, so there is nothing to resume from, and handing it back would restart the source at page one and re-fetch the whole stream. It now fails with the job id, the stream, and how to recover.

Tests

6 new/rewritten, all passing. test_create_job_and_recover encoded bug #1 — it asserted a running job is resumed, on a config that defaults to full_refresh — so it is split into the incremental case (unchanged) and a full_refresh regression test. All five pre-existing _ensure_clean_temp_table reset tests pass untouched.

Verified against a stashed pristine tree: the full suite's FAILED/ERROR sets are byte-identical before and after (34 each), so no regressions. Those 34 are pre-existing and environmental — Postgres fixtures and live-BigQuery tests.

Not verified locally: the Postgres-backed paths. Port 5432 was held by an unrelated container, so those fixtures could not run here and will first be exercised in CI.

Release

Bumped to v0.5.2 (pyproject.toml, uv.lock, CHANGELOG.md) in a separate chore(release): commit. Tag after merge to publish.

Behavior change worth calling out in review: a full_refresh job left in running is no longer resumed.

🤖 Generated with Claude Code