diff --git a/.github/workflows/verify.yml b/.github/workflows/verify.yml index cbec652280..22711433e8 100644 --- a/.github/workflows/verify.yml +++ b/.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 diff --git a/benchmarks/duckshop/edge_cases/orchestration/busy_control_store.json b/benchmarks/duckshop/edge_cases/orchestration/busy_control_store.json new file mode 100644 index 0000000000..ffe74d65fa --- /dev/null +++ b/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 +} diff --git a/benchmarks/duckshop/edge_cases/orchestration/lost_worker_restart.json b/benchmarks/duckshop/edge_cases/orchestration/lost_worker_restart.json new file mode 100644 index 0000000000..79875b1e24 --- /dev/null +++ b/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" +} 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 0000000000..0d8f9d076e --- /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-cli/src/workers/_helpers/recovery.rs b/crates/sqlbuild-cli/src/workers/_helpers/recovery.rs index 6d2e706260..ccb5e7582b 100644 --- a/crates/sqlbuild-cli/src/workers/_helpers/recovery.rs +++ b/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] diff --git a/crates/sqlbuild-cli/src/workers/_helpers/supervisor.rs b/crates/sqlbuild-cli/src/workers/_helpers/supervisor.rs index a08f9e04fd..ecfff4859a 100644 --- a/crates/sqlbuild-cli/src/workers/_helpers/supervisor.rs +++ b/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; diff --git a/crates/sqlbuild-cli/src/workers/models.rs b/crates/sqlbuild-cli/src/workers/models.rs index e64217bd12..852f344fde 100644 --- a/crates/sqlbuild-cli/src/workers/models.rs +++ b/crates/sqlbuild-cli/src/workers/models.rs @@ -96,6 +96,8 @@ pub(super) struct ClaimedWorker { pub(super) stopping: Option, pub(super) exit: Option, 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. diff --git a/crates/sqlbuild-orchestration/src/api/errors.rs b/crates/sqlbuild-orchestration/src/api/errors.rs index ded2c84b0e..460182f7d1 100644 --- a/crates/sqlbuild-orchestration/src/api/errors.rs +++ b/crates/sqlbuild-orchestration/src/api/errors.rs @@ -27,6 +27,13 @@ pub struct ApiError { impl From 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, diff --git a/crates/sqlbuild-orchestration/src/platform/main/run_maintenance.rs b/crates/sqlbuild-orchestration/src/platform/main/run_maintenance.rs index 422b2974df..d977873af7 100644 --- a/crates/sqlbuild-orchestration/src/platform/main/run_maintenance.rs +++ b/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 = 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()); diff --git a/crates/sqlbuild-orchestration/src/scheduler/main/run_scheduler.rs b/crates/sqlbuild-orchestration/src/scheduler/main/run_scheduler.rs index 4233635a0d..2917ec1447 100644 --- a/crates/sqlbuild-orchestration/src/scheduler/main/run_scheduler.rs +++ b/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; } }, } } diff --git a/crates/sqlbuild-orchestration/src/scheduler/types.rs b/crates/sqlbuild-orchestration/src/scheduler/types.rs index c32501956f..18c32c7ac9 100644 --- a/crates/sqlbuild-orchestration/src/scheduler/types.rs +++ b/crates/sqlbuild-orchestration/src/scheduler/types.rs @@ -17,4 +17,7 @@ pub trait RunLauncher { store: &ControlStore, lease: Option<&SchedulerLease>, ) -> impl Future> + Send; + + /// Hears each transient store failure the scheduler retries instead of stopping. + fn retrying(&mut self, _error: &ControlError) {} } diff --git a/crates/sqlbuild-orchestration/src/store/errors.rs b/crates/sqlbuild-orchestration/src/store/errors.rs index 8d552fd243..f58f626b6b 100644 --- a/crates/sqlbuild-orchestration/src/store/errors.rs +++ b/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(_) diff --git a/crates/sqlbuild-python/src/bindings/_helpers/compiled_project/constants.rs b/crates/sqlbuild-python/src/bindings/_helpers/compiled_project/constants.rs index a472d4d696..8b4a8dedbf 100644 --- a/crates/sqlbuild-python/src/bindings/_helpers/compiled_project/constants.rs +++ b/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; 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 e117942e2d..45463f934b 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/crates/sqlbuild-python/src/bindings/_helpers/compiled_project/snapshot_worker.rs b/crates/sqlbuild-python/src/bindings/_helpers/compiled_project/snapshot_worker.rs index dc0f759a1f..aa2342912c 100644 --- a/crates/sqlbuild-python/src/bindings/_helpers/compiled_project/snapshot_worker.rs +++ b/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; } diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/compile/COMPILE_BENCHMARKS.md b/tests/e2e/src/sqlbuild/cli/commands/main/compile/COMPILE_BENCHMARKS.md index ba7adcfdeb..90332a2565 100644 --- a/tests/e2e/src/sqlbuild/cli/commands/main/compile/COMPILE_BENCHMARKS.md +++ b/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; diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_dense_compile_performance.py b/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_dense_compile_performance.py index e4526c4b7f..9d997e32e5 100644 --- a/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_dense_compile_performance.py +++ b/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 diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_fresh_process_compile_cache_performance.py b/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_fresh_process_compile_cache_performance.py index 4fef919698..529eac3886 100644 --- a/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_fresh_process_compile_cache_performance.py +++ b/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_fresh_process_compile_cache_performance.py @@ -20,6 +20,7 @@ assert_complete_compile_cache_hit, assert_successful_compile_cache_payload, fresh_process_compile_cache_metrics, + measure_runner_calibration, run_fresh_process_compile_cache_benchmark, ) @@ -200,6 +201,21 @@ def test_given_semantic_project_when_compiling_across_processes_then_cache_is_in test_case: FreshProcessCompileCachePerformanceGuardTestCase, ) -> None: assert_required_cgroup_memory_limit() + # Fixed work that does not depend on SQLBuild scales the wall budgets on a slower runner + # class; the in-job ratios below and the same-runner ratio jobs catch a slower compiler. + factor: float = measure_runner_calibration().factor + cold_budget: float = test_case.expected_cold_max_wall_seconds * factor + warm_budget: float = test_case.expected_warm_max_wall_seconds * factor + edit_budget: float = test_case.expected_edit_max_wall_seconds * factor + project_config_edit_budget: float = ( + test_case.expected_project_config_edit_max_wall_seconds * factor + ) + print( + f"fresh-process cache models={test_case.model_count} cold_budget={cold_budget:.2f}s " + f"warm_budget={warm_budget:.2f}s edit_budget={edit_budget:.2f}s " + f"project_config_edit_budget={project_config_edit_budget:.2f}s", + flush=True, + ) result: FreshProcessCompileCacheBenchmarkResult = run_fresh_process_compile_cache_benchmark( project_dir=tmp_path / f"semantic_cache_{test_case.model_count}", model_count=test_case.model_count, @@ -209,12 +225,10 @@ def test_given_semantic_project_when_compiling_across_processes_then_cache_is_in macro_count=test_case.macro_count, test_count=test_case.test_count, audit_count=test_case.audit_count, - expected_cold_max_wall_seconds=test_case.expected_cold_max_wall_seconds, - expected_warm_max_wall_seconds=test_case.expected_warm_max_wall_seconds, - expected_edit_max_wall_seconds=test_case.expected_edit_max_wall_seconds, - expected_project_config_edit_max_wall_seconds=( - test_case.expected_project_config_edit_max_wall_seconds - ), + expected_cold_max_wall_seconds=cold_budget, + expected_warm_max_wall_seconds=warm_budget, + expected_edit_max_wall_seconds=edit_budget, + expected_project_config_edit_max_wall_seconds=project_config_edit_budget, macro_call_interval=test_case.macro_call_interval, scoped_macros=test_case.scoped_macros, compiler_only=test_case.compiler_only, @@ -246,7 +260,7 @@ def test_given_semantic_project_when_compiling_across_processes_then_cache_is_in assert_successful_compile_cache_payload(measurement=measurement, test_case=test_case) assert measurement.peak_rss_bytes < test_case.expected_max_rss_bytes - assert result.cold.elapsed_seconds < test_case.expected_cold_max_wall_seconds + assert result.cold.elapsed_seconds < cold_budget cache_write_wall_ratio: float = ( result.cold.elapsed_seconds / result.cache_disabled.elapsed_seconds ) @@ -265,7 +279,7 @@ def test_given_semantic_project_when_compiling_across_processes_then_cache_is_in result.after_macro_edit, result.after_project_config_edit, ): - assert measurement.elapsed_seconds < test_case.expected_warm_max_wall_seconds + assert measurement.elapsed_seconds < warm_budget for measurement in (result.warm, result.after_leaf_edit, result.after_macro_edit): assert ( measurement.elapsed_seconds / result.cold.elapsed_seconds @@ -276,15 +290,12 @@ def test_given_semantic_project_when_compiling_across_processes_then_cache_is_in <= _MAX_AFTER_PROJECT_REWRITE_TO_COLD_RATIO ) for measurement in (result.leaf_edit, result.macro_edit): - assert measurement.elapsed_seconds < test_case.expected_edit_max_wall_seconds + assert measurement.elapsed_seconds < edit_budget assert ( measurement.elapsed_seconds / result.cold.elapsed_seconds <= test_case.expected_max_edit_to_cold_ratio ) - assert ( - result.project_config_edit.elapsed_seconds - < test_case.expected_project_config_edit_max_wall_seconds - ) + assert result.project_config_edit.elapsed_seconds < project_config_edit_budget assert result.cache_bytes < test_case.expected_max_cache_bytes assert result.cache_disabled.semantic_fingerprint == test_case.expected_cold_fingerprint diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_fresh_process_compile_performance.py b/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_fresh_process_compile_performance.py index cab25e452e..199387d04e 100644 --- a/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_fresh_process_compile_performance.py +++ b/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_fresh_process_compile_performance.py @@ -23,8 +23,8 @@ _MIB: int = 1024 * 1024 _GIB: int = 1024 * _MIB # Temporary allowance for the native parser switch: each wall budget below is its original value -# times the approved 1.42 fresh cold wall ratio ceiling. Drop the multiplier, back to the original -# value, when the 1.10x same-runner ratio guards return. +# times 1.42, the approved fresh 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.42 diff --git a/tests/integration/src/sqlbuild/compiler/compiled_project/publication/test_local_recovery.py b/tests/integration/src/sqlbuild/compiler/compiled_project/publication/test_local_recovery.py index c4c773829f..7aeb27703f 100644 --- a/tests/integration/src/sqlbuild/compiler/compiled_project/publication/test_local_recovery.py +++ b/tests/integration/src/sqlbuild/compiler/compiled_project/publication/test_local_recovery.py @@ -1,5 +1,6 @@ """A replacement Linux dev scheduler retains physical ownership of surviving local workers.""" +import json import os import signal import sqlite3 @@ -252,3 +253,51 @@ def test_given_an_unstarted_orphan_when_recovery_publication_fails_then_claims_w with suppress(ProcessLookupError): signal.pidfd_send_signal(handle, signal.SIGKILL) os.close(handle) + + +@pytest.mark.skipif(sys.platform != "linux", reason="Linux inherited-lock and pidfd recovery") +@pytest.mark.parametrize( + "test_case", + [ + WorkerExecutionTestCase( + "a launched worker lost with its scheduler", + "SELECT 23 AS amount", + "", + "failed", + [], + launcher="uncooperative", + ) + ], + ids=lambda case: case.description, +) +def test_given_a_worker_lost_with_its_scheduler_when_restarted_then_the_run_fails_naming_the_restart( + test_case: WorkerExecutionTestCase, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + edge_case: dict[str, Any] = json.loads( + Path("benchmarks/duckshop/edge_cases/orchestration/lost_worker_restart.json").read_text() + ) + with scheduled_worker(tmp_path, monkeypatch, test_case) as worker: + pid: int = wait_for_interpreter(worker.process, worker.project / "worker.pid", worker.log) + worker.process.kill() + assert worker.process.wait(timeout=10) == -signal.SIGKILL + os.kill(pid, signal.SIGKILL) + log: Path = tmp_path / "replacement-scheduler.log" + with location_process( + [ + str(Path(sys.executable).with_name("sqb")), + "scheduler", + "--database-env", + "SQLBUILD_SCHEDULER_TEST_STORE", + "--launchers", + str(tmp_path / "launchers.json"), + ], + log, + tmp_path / "replacement-scheduler.stdout", + ) as process: + wait_for_log(process, log, "Worker supervision recovered for run", timeout=45) + run: dict[str, Any] = wait_for_run(worker.api, process, log) + assert (run["status"], run["status_reason"]) == ( + test_case.expected_status, + edge_case["expected_reason"], + ), log.read_text() + assert "Worker launched for run" not in log.read_text() 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 fc0a001287..96029d4518 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 @@ -38,6 +38,7 @@ from tests.integration.src.sqlbuild.compiler.compiled_project.publication.helpers import ( cancel_running_statement, claimed_worker, + running_statement, scheduled_worker, snapshot_api, wait_for_log, @@ -193,6 +194,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", [ @@ -517,6 +583,44 @@ def test_given_a_scheduled_job_when_published_then_an_isolated_worker_reports_re assert run["status"] == test_case.expected_status, worker.log.read_text() +@pytest.mark.parametrize( + "test_case", + [ + WorkerExecutionTestCase( + "a write lock outlasting the busy timeout during a running statement", + Path( + "benchmarks/duckshop/edge_cases/orchestration/local_worker_recovery.sql" + ).read_text(), + "SELECT count(*) FROM duckdb_tables() WHERE table_name = 'orders'", + "cancelled", + [(0,)], + ) + ], + ids=lambda case: case.description, +) +def test_given_a_busy_control_store_when_a_build_runs_then_the_scheduler_retries_and_keeps_its_outcome( + test_case: WorkerExecutionTestCase, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + edge_case: dict[str, Any] = json.loads( + Path("benchmarks/duckshop/edge_cases/orchestration/busy_control_store.json").read_text() + ) + with scheduled_worker(tmp_path, monkeypatch, test_case) as worker: + active: dict[str, Any] = running_statement(worker) + holder: sqlite3.Connection = sqlite3.connect(worker.control_store, isolation_level=None) + try: + _ = holder.execute("BEGIN IMMEDIATE") + time.sleep(edge_case["lock_seconds"]) + assert worker.process.poll() is None, worker.log.read_text() + finally: + _ = holder.execute("ROLLBACK") + holder.close() + cancel_running_statement(worker, run_id=active["run_id"]) + run: dict[str, Any] = worker_result(worker, test_case) + assert run["status"] == test_case.expected_status, worker.log.read_text() + assert "the scheduler retries: control store: database is locked" in worker.log.read_text() + assert worker.process.poll() is None, worker.log.read_text() + + @pytest.mark.parametrize( "test_case", [