fix: keep scheduled builds running through a busy store and planner skips

.github/workflows/verify.yml

@@ -845,20 +845,10 @@ jobs:
         include:
           - kind: dense
             max_ratio: "1.10"
-            # Temporary native-analysis allowance on the cold compile only; restore 1.10 when the
-            # native analysis cold path is optimised. Measured 1.326/1.347 wall and 1.642/1.677 CPU
-            # over main, plus 10%.
-            cold_max_ratio: "1.49"
-            cold_max_cpu_ratio: "1.85"
             edit_max_ratio: "1.20"
             edit_max_cpu_ratio: "1.75"
           - kind: fresh
             max_ratio: "1.10"
-            # Temporary native-analysis allowance on the cold compile only; restore 1.10 when the
-            # native analysis cold path is optimised. Measured 1.285/1.153 wall and 1.551/1.404 CPU
-            # over main, plus 10%.
-            cold_max_ratio: "1.42"
-            cold_max_cpu_ratio: "1.71"
           - kind: dense
             label: contracts CPU vs base
             gate_phase: contracts_cpu_ms
@@ -867,9 +857,7 @@ jobs:
             label: model analysis CPU vs base
             gate_phase: model_analysis_cpu_ms
             compile_args: --compile-arg=--select=*
-            # Temporary native-analysis allowance; restore 1.10 when the native analysis path is
-            # optimised. Measured 2.660/2.554 over main, plus 10%.
-            max_ratio: "2.93"
+            max_ratio: "1.10"
           - kind: dense
             label: custom rules CPU vs base
             gate_phase: custom_rules_cpu_ms

benchmarks/duckshop/edge_cases/orchestration/busy_control_store.json

@@ -0,0 +1,4 @@
+{
+  "description": "Another connection holds the SQLite control store's write lock for longer than the store's busy timeout while a scheduled build's warehouse statement is running. The scheduler retries its cycle instead of exiting and terminating its worker, the statement keeps running, and the run ends with its real outcome: here an operator cancellation after the lock is released.",
+  "lock_seconds": 35
+}

benchmarks/duckshop/edge_cases/orchestration/lost_worker_restart.json

@@ -0,0 +1,4 @@
+{
+  "description": "A scheduler and its launched local worker are both killed before the worker reports anything. The replacement scheduler adopts the claim, finds the worker gone and fails the run with a reason that names the restart.",
+  "expected_reason": "the control service restarted during this run, and its worker stopped without reporting a result"
+}

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-cli/src/workers/_helpers/recovery.rs

@@ -66,6 +66,7 @@ impl Workers {
                         stopping: None,
                         exit: None,
                         launch_failure: rejected.then_some("docker_configuration_rejected"),
+                        recovered: true,
                     });
                     if rejected {
                         self.active[index]

crates/sqlbuild-cli/src/workers/_helpers/supervisor.rs

@@ -105,6 +105,17 @@ impl sqlbuild_orchestration::scheduler::types::RunLauncher for Workers {
                     } else {
                         RunOutcome::Failed
                     };
+                    worker.publish_launch_failure(store, lease).await?;
+                    let reason = if worker.child.is_none() {
+                        "worker launch preparation failed before a child started"
+                    } else if worker.recovered
+                        && worker.stopping.is_none()
+                        && state.cancel_requested_at.is_none()
+                    {
+                        "the control service restarted during this run, and its worker stopped without reporting a result"
+                    } else {
+                        "worker stopped; physical exit or absence confirmed"
+                    };
                     publish(
                         store,
                         lease,
@@ -113,7 +124,7 @@ impl sqlbuild_orchestration::scheduler::types::RunLauncher for Workers {
                             "local-exit",
                             RunChange::Finished {
                                 outcome,
-                                reason: "worker stopped; physical exit or absence confirmed".into(),
+                                reason: reason.into(),
                                 exit_code: worker.exit.and_then(WorkerExit::code),
                             },
                         ),
@@ -137,6 +148,15 @@ impl sqlbuild_orchestration::scheduler::types::RunLauncher for Workers {
     ) -> Result<(), ControlError> {
         self.shutdown_workers(store, lease).await
     }
+
+    /// A claim whose commit outcome is unknown is adopted by the next recovery pass.
+    fn retrying(&mut self, error: &ControlError) {
+        self.recovered_epoch = None;
+        drop(writeln!(
+            std::io::stderr().lock(),
+            "Control store unavailable; the scheduler retries: {error}"
+        ));
+    }
 }
 
 impl Workers {
@@ -182,6 +202,7 @@ impl Workers {
                     stopping: None,
                     exit: None,
                     launch_failure: None,
+                    recovered: false,
                 });
                 let worker = &mut self.active[index];
                 let claim = &worker.claim;

crates/sqlbuild-cli/src/workers/models.rs

@@ -96,6 +96,8 @@ pub(super) struct ClaimedWorker {
     pub(super) stopping: Option<Instant>,
     pub(super) exit: Option<WorkerExit>,
     pub(super) launch_failure: Option<&'static str>,
+    /// Supervision resumed after a scheduler restart rather than from this scheduler's launch.
+    pub(super) recovered: bool,
 }
 
 /// Physical ownership survives leadership loss; only the current fence changes durable lifecycle.

crates/sqlbuild-orchestration/src/api/errors.rs

@@ -27,6 +27,13 @@ pub struct ApiError {
 
 impl From<ControlError> for ApiError {
     fn from(error: ControlError) -> Self {
+        if error.transient() {
+            return Self {
+                status: StatusCode::SERVICE_UNAVAILABLE,
+                message: "control store is busy; retry the request".into(),
+                decision_id: None,
+            };
+        }
         match error {
             ControlError::Invalid(message) => Self {
                 status: StatusCode::BAD_REQUEST,

crates/sqlbuild-orchestration/src/platform/main/run_maintenance.rs

@@ -31,7 +31,11 @@ pub async fn run_maintenance(
     let mut reported = 0;
     let mut retained_at: Option<Instant> = None;
     while !*shutdown.borrow() {
-        scan_health(store, replica_id, &shutdown).await?;
+        match scan_health(store, replica_id, &shutdown).await {
+            Ok(()) => {}
+            Err(error) if error.transient() => {}
+            Err(error) => return Err(error),
+        }
         if retained_at.is_none_or(|at| at.elapsed() >= Duration::from_secs(60)) {
             match retain(store, retention, &sinks).await {
                 Ok(retained) => {
@@ -40,7 +44,7 @@ pub async fn run_maintenance(
                     }
                     reported = retained;
                 }
-                Err(ControlError::Database(error)) if ControlError::transient_database(&error) => {}
+                Err(error) if error.transient() => {}
                 Err(error) => return Err(error),
             }
             retained_at = Some(Instant::now());

crates/sqlbuild-orchestration/src/scheduler/main/run_scheduler.rs

@@ -1,5 +1,7 @@
 //! Embedded and standalone roles share the same interruptible, durable scheduling loop.
 
+use std::time::Duration;
+
 use tokio::sync::watch;
 
 use crate::scheduler::main::_scheduler_cycle::scheduler_cycle;
@@ -8,7 +10,7 @@ use crate::scheduler::types::RunLauncher;
 use crate::store::errors::ControlError;
 use crate::store::models::ControlStore;
 
-/// Durable database state drives work; shutdown interrupts the wait.
+/// Durable database state drives work; a transient store failure retries the next cycle.
 pub async fn run_scheduler(
     store: &ControlStore,
     mut state: SchedulerState,
@@ -18,12 +20,27 @@ pub async fn run_scheduler(
     let mut last_lease = None;
     let result = async {
         while !*shutdown.borrow() {
-            let (next_state, cycle) = scheduler_cycle(store, state).await?;
-            state = next_state;
-            last_lease = state.lease.clone();
-            launcher.poll(store, state.lease.as_ref()).await?;
+            let polled = match scheduler_cycle(store, state.clone()).await {
+                Ok((next_state, cycle)) => {
+                    state = next_state;
+                    last_lease = state.lease.clone();
+                    launcher
+                        .poll(store, state.lease.as_ref())
+                        .await
+                        .map(|()| cycle.wait_for)
+                }
+                Err(error) => Err(error),
+            };
+            let wait_for = match polled {
+                Ok(wait_for) => wait_for,
+                Err(error) if error.transient() => {
+                    launcher.retrying(&error);
+                    Duration::from_secs(1)
+                }
+                Err(error) => return Err(error),
+            };
             tokio::select! {
-                _ = tokio::time::sleep(cycle.wait_for.min(std::time::Duration::from_secs(1))) => {},
+                _ = tokio::time::sleep(wait_for.min(Duration::from_secs(1))) => {},
                 result = shutdown.changed() => { if result.is_err() { break; } },
             }
         }

crates/sqlbuild-orchestration/src/scheduler/types.rs

@@ -17,4 +17,7 @@ pub trait RunLauncher {
         store: &ControlStore,
         lease: Option<&SchedulerLease>,
     ) -> impl Future<Output = Result<(), ControlError>> + Send;
+
+    /// Hears each transient store failure the scheduler retries instead of stopping.
+    fn retrying(&mut self, _error: &ControlError) {}
 }

crates/sqlbuild-orchestration/src/store/errors.rs

@@ -27,6 +27,19 @@ pub enum ControlError {
 }
 
 impl ControlError {
+    /// A store that is briefly busy or unreachable; the same operation may succeed when retried.
+    #[must_use]
+    pub fn transient(&self) -> bool {
+        match self {
+            Self::Database(error) => Self::transient_database(error),
+            Self::Sqlite(rusqlite::Error::SqliteFailure(error, _)) => matches!(
+                error.code,
+                rusqlite::ErrorCode::DatabaseBusy | rusqlite::ErrorCode::DatabaseLocked
+            ),
+            _ => false,
+        }
+    }
+
     pub(crate) fn transient_database(error: &sqlx_core::Error) -> bool {
         match error {
             sqlx_core::Error::Io(_)

crates/sqlbuild-python/src/bindings/_helpers/compiled_project/constants.rs

@@ -3,3 +3,5 @@
 pub(crate) const SKIP_ACTION: &str = "skip";
 pub(crate) const NO_CHANGE_REASON: &str = "no_change";
 pub(crate) const REPORT_ATTEMPTS: usize = 3;
+/// The HTTP status a control server answers while its store is busy; reports retry it.
+pub(crate) const STORE_BUSY_STATUS: u16 = 503;

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

crates/sqlbuild-python/src/bindings/_helpers/compiled_project/snapshot_worker.rs

@@ -6,12 +6,13 @@ use pyo3::exceptions::{PyRuntimeError, PyValueError};
 use pyo3::prelude::{PyRef, PyResult, Python};
 use pyo3::{pyclass, pymethods};
 use sqlbuild_orchestration::runs::models::RunChange;
+use sqlbuild_orchestration::worker::errors::WorkerError;
 use sqlbuild_orchestration::worker::main::connect_worker::connect_worker;
 use sqlbuild_orchestration::worker::main::start_worker::start_worker;
 use sqlbuild_orchestration::worker::models::{WorkerClient, WorkerSession};
 use uuid::Uuid;
 
-use crate::bindings::_helpers::compiled_project::constants::REPORT_ATTEMPTS;
+use crate::bindings::_helpers::compiled_project::constants::{REPORT_ATTEMPTS, STORE_BUSY_STATUS};
 use crate::bindings::_helpers::compiled_project::snapshot::NativeOrchestrationSnapshot;
 
 struct WorkerOwner {
@@ -60,7 +61,7 @@ impl NativeOrchestrationWorker {
         self.with_owner(py, |owner| Ok(owner.session.execution.clone()))
     }
 
-    /// A bounded retry keeps the exact event identity across an uncertain acknowledgement.
+    /// A bounded retry keeps the event identity across an uncertain acknowledgement or busy store.
     pub(crate) fn change(&self, py: Python<'_>, key: &str, change: RunChange) -> PyResult<()> {
         self.with_owner(py, |owner| {
             let event_id = Uuid::new_v5(&owner.session.execution.claim.claim_id, key.as_bytes());
@@ -70,8 +71,15 @@ impl NativeOrchestrationWorker {
                     attempts += 1;
                     match owner.client.report(event_id, change.clone()).await {
                         Ok(_) => return Ok(()),
-                        Err(sqlbuild_orchestration::worker::errors::WorkerError::Transport(_))
-                            if attempts < REPORT_ATTEMPTS =>
+                        Err(error)
+                            if attempts < REPORT_ATTEMPTS
+                                && match &error {
+                                    WorkerError::Transport(_) => true,
+                                    WorkerError::Rejected(status) => {
+                                        status.as_u16() == STORE_BUSY_STATUS
+                                    }
+                                    _ => false,
+                                } =>
                         {
                             tokio::time::sleep(std::time::Duration::from_millis(200)).await;
                         }

tests/e2e/src/sqlbuild/cli/commands/main/compile/COMPILE_BENCHMARKS.md

@@ -88,11 +88,13 @@ All cases carry `models_3000` identifiers so they run only in the 3,000-model fr
 
 ## Same-runner comparison with the base branch
 
-Absolute budgets must absorb the spread between hosted runners, so every pull request that touches
-the compiler also runs `scripts/compare_compile_performance.py` on the dense and the fresh
-3,000-model benchmark. Base and head each compile a project written by their own generator, on the
-same runner, three alternating runs per mode, and the medians of wall and CPU time must stay within
-1.10x of base for each mode:
+Absolute budgets must absorb the spread between hosted runner classes, up to 1.9x on the same
+compile: the dense and fresh-process cache guards scale their wall budgets by a fixed calibration
+workload on slower classes. An absolute budget therefore cannot catch a slower compiler, so every
+pull request that touches the compiler also runs `scripts/compare_compile_performance.py` on the
+dense and the fresh 3,000-model benchmark. Base and head each compile a project written by their
+own generator, on the same runner, three alternating runs per mode, and the medians of wall and CPU
+time must stay within 1.10x of base for each mode:
 
 - cold: `sqb compile --json --no-cache` after removing `target`, after one untimed warm-up;
 - warm: an unchanged `sqb compile --json` after one untimed run that writes the cache;

tests/e2e/src/sqlbuild/cli/commands/main/compile/test_dense_compile_performance.py

@@ -45,8 +45,8 @@
 _MIB: int = 1024 * 1024
 _GIB: int = 1024 * _MIB
 # Temporary allowance for the native parser switch: each cold wall budget below is its original
-# value times the approved 1.49 dense cold wall ratio ceiling. Drop the multiplier, back to the
-# original value, when the 1.10x same-runner ratio guards return.
+# value times 1.49, the approved dense cold slowdown of native analysis. Drop the multiplier, back
+# to the original value, when the native analysis cold path is optimised.
 _NATIVE_COLD_WALL_ALLOWANCE: float = 1.49
 # A delta reruns the edited model or the edited SQL test's few models; a full compile all of them.
 _MAX_DELTA_ANALYSED_MODELS: int = 10