From 754a476146ac68679fda8ef9c34fabc881c0edc5 Mon Sep 17 00:00:00 2001 From: Kevin Longe <273747162+kvlonge@users.noreply.github.com> Date: Wed, 11 Nov 2026 09:56:28 +0000 Subject: [PATCH] fix: keep workers from restarting nodes the planner already skipped A scheduled build reported a source-freshness skip as finished and then reported the same node as started when the executor reached it. The control API rejects that change, so the worker stopped and every node it had not reached yet ended skipped, including branches independent of the stale source. Workers now never report a start for a node whose outcome is already final. --- .../orchestration/worker_freshness_skip.yml | 16 +++++ .../compiled_project/snapshot_execution.rs | 10 ++- .../publication/test_worker.py | 65 +++++++++++++++++++ 3 files changed, 89 insertions(+), 2 deletions(-) create mode 100644 benchmarks/duckshop/edge_cases/orchestration/worker_freshness_skip.yml diff --git a/benchmarks/duckshop/edge_cases/orchestration/worker_freshness_skip.yml b/benchmarks/duckshop/edge_cases/orchestration/worker_freshness_skip.yml new file mode 100644 index 000000000..0d8f9d076 --- /dev/null +++ b/benchmarks/duckshop/edge_cases/orchestration/worker_freshness_skip.yml @@ -0,0 +1,16 @@ +# A scheduled build where raw_orders fails its freshness check and raw_returns has none. +# The planner skips the models that read raw_orders before execution; the worker must never +# report those skips as started, and models that read raw_returns must still build. +sources: + - name: raw_orders + description: Raw orders. + expression: SELECT 1 AS order_id + freshness: + strategy: sql + type: timestamp + query: SELECT CAST('2000-01-01 00:00:00' AS TIMESTAMP) AS data_version + age_policy: + error_after: 1h + - name: raw_returns + description: Raw returns. + expression: SELECT 2 AS return_id diff --git a/crates/sqlbuild-python/src/bindings/_helpers/compiled_project/snapshot_execution.rs b/crates/sqlbuild-python/src/bindings/_helpers/compiled_project/snapshot_execution.rs index a5fe394f0..f5fc5b8bb 100644 --- a/crates/sqlbuild-python/src/bindings/_helpers/compiled_project/snapshot_execution.rs +++ b/crates/sqlbuild-python/src/bindings/_helpers/compiled_project/snapshot_execution.rs @@ -213,10 +213,16 @@ impl WorkerBuild { self.start_key(py, &format!("{kind}:{name}")) } - /// Report a selected node as started; the deterministic event ID makes a repeat a no-op. + /// Report a selected node as started unless its outcome is final, such as a planner skip. pub(crate) fn start_key(&self, py: Python<'_>, key: &str) -> PyResult<()> { self.check_cancelled(py)?; - if self.execution.claim.request.selection.contains(key) { + if self.execution.claim.request.selection.contains(key) + && !self + .completed + .lock() + .map_err(|_| poisoned())? + .contains_key(key) + { self.worker.bind(py).get().change( py, &format!("node-started/{key}"), diff --git a/tests/integration/src/sqlbuild/compiler/compiled_project/publication/test_worker.py b/tests/integration/src/sqlbuild/compiler/compiled_project/publication/test_worker.py index 0fb2cbabe..f6b52e8b3 100644 --- a/tests/integration/src/sqlbuild/compiler/compiled_project/publication/test_worker.py +++ b/tests/integration/src/sqlbuild/compiler/compiled_project/publication/test_worker.py @@ -193,6 +193,71 @@ def test_given_a_scheduled_group_when_workers_execute_then_dependencies_and_fail assert warehouse.sql(test_case.warehouse_query).fetchall() == test_case.expected_rows +@pytest.mark.parametrize( + "test_case", + [ + WorkerExecutionTestCase( + "a stale source blocks only its own branch", + "SELECT 11 AS amount", + "SELECT amount FROM orders UNION ALL SELECT return_id FROM returns ORDER BY 1", + "partial", + [(2,), (11,)], + files={ + "sources/raw.yml": Path( + "benchmarks/duckshop/edge_cases/orchestration/worker_freshness_skip.yml" + ).read_text(), + "models/stale_orders.sql": ( + "MODEL (description 'Stale orders.', materialized table);\n" + 'SELECT order_id FROM __source("raw_orders")\n' + ), + "models/late_orders.sql": ( + "MODEL (description 'Late orders.', materialized table);\n" + 'SELECT order_id FROM __source("raw_orders")\n' + ), + "models/returns.sql": ( + "MODEL (description 'Returns.', materialized table);\n" + 'SELECT return_id FROM __source("raw_returns")\n' + ), + "orchestration/orders.py": ( + "from sqlbuild.orchestration import job, schedule\n" + 'annual_orders = job("annual_orders", ' + 'select="orders stale_orders late_orders returns", ' + 'description="Build orders.", ' + 'schedule=schedule(cron="0 0 * * *", timezone="UTC", ' + 'starts_at="2000-01-01T00:00:00Z"))\n' + ), + }, + ) + ], + ids=lambda case: case.description, +) +def test_given_a_stale_source_when_a_worker_builds_then_blocked_nodes_finish_skipped_unstarted( + test_case: WorkerExecutionTestCase, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + with scheduled_worker(tmp_path, monkeypatch, test_case) as worker: + run: dict[str, Any] = wait_for_run( + worker.api, worker.process, worker.log, job_key="annual_orders" + ) + assert run["status"] == test_case.expected_status, worker.log.read_text() + with sqlite3.connect(worker.control_store) as control: + nodes: list[tuple[str, str, bool]] = [ + (key, status, started is not None) + for key, status, started in control.execute( + "SELECT node_key, status, started_at FROM sqb_run_nodes " + "WHERE run_id=? AND kind='node' ORDER BY node_key", + (run["run_id"],), + ) + ] + assert nodes == [ + ("model:late_orders", "skipped", False), + ("model:orders", "success", True), + ("model:returns", "success", True), + ("model:stale_orders", "skipped", False), + ], worker.log.read_text() + with duckdb.connect(str(worker.project / "warehouse.duckdb"), read_only=True) as warehouse: + assert warehouse.sql(test_case.warehouse_query).fetchall() == test_case.expected_rows + + @pytest.mark.parametrize( "test_case", [