fix: keep workers from restarting nodes the planner already skipped

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

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}"),

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",
     [