fix: don't publish a partial extract when the producer aborts

[anaselmhamdi]

The bug

A run that dies mid-stream still swaps its partial extract into the production table and records the job as succeeded.

The producer sends the same clean QUEUE_TERMINATION signal whether it exhausted the source or aborted on an error — the break out of the error path falls through to the very same queue.terminate() call a successful run makes. The consumer can't tell them apart, so it treats the abort as the last iteration:

1. producer.py — SOURCE_ERROR breaks, then self.queue.terminate(iteration=...), identical to the success path
2. consumer.py — sees QUEUE_TERMINATION, calls write_records_and_update_cursor(..., last_iteration=True), returns SUCCESS
3. destination.py — last_iteration=True sets JobStatus.SUCCEEDED and calls finalize() (on BigQuery, the WRITE_TRUNCATE copy publishing {table}_temp over the production table), with no check of producer status at all

Production impact

Observed on a full_refresh HubSpot stream that lost its OAuth token at iteration 3517 of 4400:

05:30:21  Successfully wrote 26300 records to destination contacts_merged_audit
05:30:24  Replacing <project>.bizon_hubspot.contacts_merged_audit with temp table ... via copy job

Confirmed in BigQuery INFORMATION_SCHEMA.JOBS (COPY DONE OK) and in stream_jobs, which recorded that run as succeeded. The production table was republished at ~80% of its rows (~88k records missing) on three separate days. Nothing downstream could detect it — only the process exit code indicated a failure at all.

The fix

• New QUEUE_TERMINATION_ERROR signal (QueueMessage.signal is already Optional[str], so this is backward-compatible).
• terminate() takes signal, defaulting to QUEUE_TERMINATION; threaded through all three queue adapters.
• The producer sends QUEUE_TERMINATION_ERROR when return_value != SUCCESS.
• The consumer aborts on that signal without writing a last iteration, without finalizing, and without marking the job succeeded. The check sits before the transform so a transform error can't mask the producer's failure.

The job is deliberately left running — the state get_or_create_job() already recovers from, since 0.5.2 cancels and restarts a leftover running full refresh from page 1.

Also fixed: thread.py truthiness bug

if result_producer.SUCCESS: is an attribute access on the enum class, which resolves to the PipelineReturnStatus.SUCCESS member and is therefore always truthy. The else branch setting consumer_stop_event was unreachable, and the runner logged "Producer thread has finished successfully" for every failure. (The consumer branch three lines below always compared correctly with == SUCCESS.)

Fixing this alone would not be sufficient — by the time the runner observes the producer result, the consumer has usually already drained the termination message and finalized. The producer/consumer signalling is the load-bearing fix; this is the belt to its braces.

Tests

tests/engine/test_consumer_error_termination.py — two tests mirroring the production shape (records land, then the stream closes):

• test_error_termination_does_not_finalize_or_mark_succeeded — asserts SOURCE_ERROR, finalize() never called, job not SUCCEEDED
• test_normal_termination_still_finalizes_and_marks_succeeded — the happy path is unchanged

Verified the test actually catches the bug: neutering only the new consumer guard makes the first test fail with assert success == source_error while the happy path still passes.

Full suite: 31 failed, 276 passed, 12 skipped, 3 errors on this branch vs 31 failed, 274 passed, 12 skipped, 3 errors on pristine main — identical pre-existing failures (they require live BigQuery/Postgres credentials), plus exactly the 2 new tests. ruff check and ruff format --check clean.

Reviewer note — behaviour change

This changes shared behaviour for every pipeline. Pipelines that today quietly "succeed" on a partial extract will start failing loudly and leaving running jobs. That's the intended outcome, but it's worth watching the first few nights after upgrading.

🤖 Generated with Claude Code