From f50d9145a0e148a1fff8d4fd5491c7324aa13ca3 Mon Sep 17 00:00:00 2001 From: Kevin Longe <273747162+kvlonge@users.noreply.github.com> Date: Thu, 5 Nov 2026 01:32:56 +0000 Subject: [PATCH 1/4] ci: run scheduler recovery contracts on macOS and Windows Adds a change-selected Verify job that runs the control store, scheduler and worker supervision tests on macOS and Windows runners against SQLite and a native PostgreSQL server, and counts its result in the aggregate. --- .github/workflows/verify.yml | 70 +++++++++++++++++++++++++++++++++--- 1 file changed, 65 insertions(+), 5 deletions(-) diff --git a/.github/workflows/verify.yml b/.github/workflows/verify.yml index cd908d1d90..a74b538f47 100644 --- a/.github/workflows/verify.yml +++ b/.github/workflows/verify.yml @@ -28,6 +28,7 @@ jobs: format_performance: ${{ steps.scope.outputs.format_performance }} rules_performance: ${{ steps.scope.outputs.rules_performance }} wheels: ${{ steps.scope.outputs.wheels }} + recovery: ${{ steps.scope.outputs.recovery }} steps: - uses: actions/checkout@v4 with: @@ -52,11 +53,12 @@ jobs: echo "wheels=$7" echo "compiler_differential=$8" echo "clickhouse=$9" + echo "recovery=${10}" } >> "${GITHUB_OUTPUT}" } if [[ "${EVENT_NAME}" != "pull_request" && "${EVENT_NAME}" != "merge_group" ]]; then - write_scope true true true true true true true true true + write_scope true true true true true true true true true true exit 0 fi @@ -68,7 +70,7 @@ jobs: if [[ -z "${BASE_SHA}" || -z "${HEAD_SHA}" ]] || \ ! git diff --name-only -z "${BASE_SHA}...${HEAD_SHA}" > "${changed_paths}"; then echo "Could not diff ${BASE_SHA:-}...${HEAD_SHA:-}; running every suite" >&2 - write_scope true true true true true true true true true + write_scope true true true true true true true true true true exit 0 fi @@ -114,7 +116,7 @@ jobs: && "${HEAD_REF}" == release-please--branches--* ]] && release_metadata_only; then echo "Release Please pull request changes only release metadata; skipping the suites" \ | tee -a "${GITHUB_STEP_SUMMARY}" - write_scope false false false false false false false false false + write_scope false false false false false false false false false false exit 0 fi @@ -126,6 +128,7 @@ jobs: rules_performance=false wheels=false compiler_differential=false + recovery=false broad=false changed=false # A change that only touches the website, root-level Markdown, the README demo GIFs or the @@ -146,6 +149,11 @@ jobs: wheels=true ;; esac + case "${path}" in + crates/sqlbuild-orchestration/*|crates/sqlbuild-cli/*|Cargo.toml|Cargo.lock|.github/workflows/verify.yml) + recovery=true + ;; + esac case "${path}" in scripts/compiler_differential/*|scripts/run_compiler_differential.py|tests/*/scripts/compiler_differential/*|tests/goldens/*|tests/e2e/fixtures/*|website/examples/*|.github/workflows/verify.yml|Makefile) compiler_differential=true @@ -207,7 +215,7 @@ jobs: done < "${changed_paths}" if [[ "${changed}" != "true" ]]; then - write_scope true true true true true true true true true + write_scope true true true true true true true true true true exit 0 fi @@ -232,7 +240,8 @@ jobs: "${checks}" \ "${wheels}" \ "${compiler_differential}" \ - "${clickhouse}" + "${clickhouse}" \ + "${recovery}" lockfile: name: Lockfile consistency @@ -549,6 +558,54 @@ jobs: uv run --no-sync sqb playground --template python_nodes "${nodes}" uv run --no-sync sqb --project-dir "${nodes}" --no-color compile + scheduler_recovery: + name: Scheduler recovery (${{ matrix.name }}) + needs: [changes, lockfile] + if: needs.changes.outputs.recovery == 'true' + runs-on: ${{ matrix.runner }} + timeout-minutes: 75 + strategy: + fail-fast: false + matrix: + include: + - name: macOS + runner: macos-15 + - name: Windows + runner: windows-2025 + defaults: + run: + shell: bash + steps: + - uses: actions/checkout@v4 + - uses: dtolnay/rust-toolchain@master + with: + toolchain: stable + - uses: Swatinem/rust-cache@v2 + with: + shared-key: verify-recovery-${{ matrix.name }} + save-if: ${{ github.ref == 'refs/heads/main' }} + - name: Start a native PostgreSQL server + run: | + if [[ "${RUNNER_OS}" == "macOS" ]]; then + brew install --quiet postgresql@16 + bin="$(brew --prefix postgresql@16)/bin" + else + bin="$(cygpath -u "${PGBIN}")" + fi + data="${RUNNER_TEMP}/postgres-data" + "${bin}/initdb" -D "${data}" -U postgres --auth=trust -E UTF8 > /dev/null + "${bin}/pg_ctl" -D "${data}" -o "-p 5433 -c listen_addresses=127.0.0.1" \ + -l "${RUNNER_TEMP}/postgres.log" -w start + - name: Control store, scheduler and recovery contracts on SQLite and PostgreSQL + env: + SQLBUILD_TEST_CONTROL_POSTGRES_URL: postgresql://postgres@127.0.0.1:5433/postgres + run: cargo test --locked -p sqlbuild-orchestration --lib + - name: Worker supervision and person credential storage + run: cargo test --locked -p sqlbuild-cli --lib -- workers person_login + - name: PostgreSQL server log + if: failure() + run: cat "${RUNNER_TEMP}/postgres.log" || true + format_performance: name: Formatter performance guards needs: [changes, lockfile] @@ -1080,6 +1137,7 @@ jobs: compiler_differential, compiler_differential_python_versions, native_discovery_windows, + scheduler_recovery, format_performance, performance, cold_compile_performance, @@ -1105,6 +1163,7 @@ jobs: COMPILER_DIFFERENTIAL_RESULT: ${{ needs.compiler_differential.result }} COMPILER_DIFFERENTIAL_PYTHON_VERSIONS_RESULT: ${{ needs.compiler_differential_python_versions.result }} NATIVE_DISCOVERY_WINDOWS_RESULT: ${{ needs.native_discovery_windows.result }} + SCHEDULER_RECOVERY_RESULT: ${{ needs.scheduler_recovery.result }} FORMAT_PERFORMANCE_RESULT: ${{ needs.format_performance.result }} PERFORMANCE_RESULT: ${{ needs.performance.result }} COLD_COMPILE_PERFORMANCE_RESULT: ${{ needs.cold_compile_performance.result }} @@ -1137,6 +1196,7 @@ jobs: "${COMPILER_DIFFERENTIAL_RESULT}" \ "${COMPILER_DIFFERENTIAL_PYTHON_VERSIONS_RESULT}" \ "${NATIVE_DISCOVERY_WINDOWS_RESULT}" \ + "${SCHEDULER_RECOVERY_RESULT}" \ "${FORMAT_PERFORMANCE_RESULT}" \ "${PERFORMANCE_RESULT}" \ "${COLD_COMPILE_PERFORMANCE_RESULT}" \ From 61c89dedb7411264f3c0a1397979fde089fe54ac Mon Sep 17 00:00:00 2001 From: Kevin Longe <273747162+kvlonge@users.noreply.github.com> Date: Thu, 5 Nov 2026 01:58:02 +0000 Subject: [PATCH 2/4] wip: prove local worker claim ownership on Linux, macOS and Windows Work in progress. Adds sqlbuild-process-ownership, the only crate allowed unsafe code, with a safe API to prepare, start, recover, observe and stop an owned worker. Linux keeps its inherited lock, /proc and pidfd proof. macOS checks the lock through libproc and rechecks the start time just before signalling. Windows runs each worker in a private named Job object owned by, and open only to, the server account. The CLI's local launcher and recovery now use it on all three platforms. Validated so far: the crate's recovery cases pass on Linux, and clippy is clean for the crate on Linux, Windows and macOS targets. The Verify job also carries a temporary diagnostics step for the Windows lease failures, to be removed before review. --- .github/workflows/verify.yml | 59 +++- Cargo.lock | 14 +- Cargo.toml | 8 +- crates/README.md | 3 +- crates/sqlbuild-cli/Cargo.toml | 4 +- .../src/workers/_helpers/child.rs | 17 +- .../src/workers/_helpers/local.rs | 128 ------- .../sqlbuild-cli/src/workers/_helpers/mod.rs | 2 - .../src/workers/_helpers/recovery.rs | 11 +- .../src/workers/_helpers/supervisor.rs | 67 +++- crates/sqlbuild-cli/src/workers/constants.rs | 4 - crates/sqlbuild-cli/src/workers/models.rs | 15 +- crates/sqlbuild-process-ownership/Cargo.toml | 46 +++ crates/sqlbuild-process-ownership/src/lib.rs | 3 + .../src/ownership/_helpers/linux.rs | 54 +++ .../src/ownership/_helpers/lock.rs | 42 +++ .../src/ownership/_helpers/macos.rs | 167 +++++++++ .../src/ownership/_helpers/mod.rs | 8 + .../src/ownership/_helpers/windows.rs | 319 ++++++++++++++++++ .../src/ownership/constants.rs | 16 + .../src/ownership/main/mod.rs | 7 + .../src/ownership/main/prepare_worker.rs | 27 ++ .../src/ownership/main/recover_worker.rs | 20 ++ .../src/ownership/main/start_worker.rs | 15 + .../src/ownership/main/stop_worker.rs | 19 ++ .../src/ownership/main/worker_exited.rs | 24 ++ .../src/ownership/mod.rs | 9 + .../src/ownership/models.rs | 56 +++ .../src/ownership/tests.rs | 11 + .../src/ownership/tests/helpers.rs | 186 ++++++++++ .../src/ownership/tests/test_job_access.rs | 58 ++++ .../src/ownership/tests/test_recovery.rs | 112 ++++++ .../src/ownership/tests/test_types.rs | 19 ++ fensu.toml | 31 ++ 34 files changed, 1407 insertions(+), 174 deletions(-) delete mode 100644 crates/sqlbuild-cli/src/workers/_helpers/local.rs create mode 100644 crates/sqlbuild-process-ownership/Cargo.toml create mode 100644 crates/sqlbuild-process-ownership/src/lib.rs create mode 100644 crates/sqlbuild-process-ownership/src/ownership/_helpers/linux.rs create mode 100644 crates/sqlbuild-process-ownership/src/ownership/_helpers/lock.rs create mode 100644 crates/sqlbuild-process-ownership/src/ownership/_helpers/macos.rs create mode 100644 crates/sqlbuild-process-ownership/src/ownership/_helpers/mod.rs create mode 100644 crates/sqlbuild-process-ownership/src/ownership/_helpers/windows.rs create mode 100644 crates/sqlbuild-process-ownership/src/ownership/constants.rs create mode 100644 crates/sqlbuild-process-ownership/src/ownership/main/mod.rs create mode 100644 crates/sqlbuild-process-ownership/src/ownership/main/prepare_worker.rs create mode 100644 crates/sqlbuild-process-ownership/src/ownership/main/recover_worker.rs create mode 100644 crates/sqlbuild-process-ownership/src/ownership/main/start_worker.rs create mode 100644 crates/sqlbuild-process-ownership/src/ownership/main/stop_worker.rs create mode 100644 crates/sqlbuild-process-ownership/src/ownership/main/worker_exited.rs create mode 100644 crates/sqlbuild-process-ownership/src/ownership/mod.rs create mode 100644 crates/sqlbuild-process-ownership/src/ownership/models.rs create mode 100644 crates/sqlbuild-process-ownership/src/ownership/tests.rs create mode 100644 crates/sqlbuild-process-ownership/src/ownership/tests/helpers.rs create mode 100644 crates/sqlbuild-process-ownership/src/ownership/tests/test_job_access.rs create mode 100644 crates/sqlbuild-process-ownership/src/ownership/tests/test_recovery.rs create mode 100644 crates/sqlbuild-process-ownership/src/ownership/tests/test_types.rs diff --git a/.github/workflows/verify.yml b/.github/workflows/verify.yml index a74b538f47..68a9765fe8 100644 --- a/.github/workflows/verify.yml +++ b/.github/workflows/verify.yml @@ -150,7 +150,7 @@ jobs: ;; esac case "${path}" in - crates/sqlbuild-orchestration/*|crates/sqlbuild-cli/*|Cargo.toml|Cargo.lock|.github/workflows/verify.yml) + crates/sqlbuild-orchestration/*|crates/sqlbuild-process-ownership/*|crates/sqlbuild-cli/*|Cargo.toml|Cargo.lock|.github/workflows/verify.yml) recovery=true ;; esac @@ -596,12 +596,65 @@ jobs: "${bin}/initdb" -D "${data}" -U postgres --auth=trust -E UTF8 > /dev/null "${bin}/pg_ctl" -D "${data}" -o "-p 5433 -c listen_addresses=127.0.0.1" \ -l "${RUNNER_TEMP}/postgres.log" -w start + - name: Worker claim ownership, recovery and stop + run: cargo test --locked -p sqlbuild-process-ownership + - name: Worker supervision and person credential storage + run: cargo test --locked -p sqlbuild-cli --lib -- workers person_login - name: Control store, scheduler and recovery contracts on SQLite and PostgreSQL env: SQLBUILD_TEST_CONTROL_POSTGRES_URL: postgresql://postgres@127.0.0.1:5433/postgres run: cargo test --locked -p sqlbuild-orchestration --lib - - name: Worker supervision and person credential storage - run: cargo test --locked -p sqlbuild-cli --lib -- workers person_login + - name: "Temporary diagnostics: commit latency and serial test timings" + if: always() + env: + SQLBUILD_TEST_CONTROL_POSTGRES_URL: postgresql://postgres@127.0.0.1:5433/postgres + run: | + nproc 2>/dev/null || sysctl -n hw.ncpu || true + python3 -c "import sys; print(sys.version)" 2>/dev/null || python -c "import sys; print(sys.version)" + py="$(command -v python3 || command -v python)" + "${py}" - "${RUNNER_TEMP}" <<'PYEOF' + import os, sqlite3, sys, time + for sync in ("NORMAL", "FULL"): + path = os.path.join(sys.argv[1], f"latency-{sync}.sqlite3") + connection = sqlite3.connect(path, isolation_level=None) + connection.execute(f"PRAGMA journal_mode=WAL") + connection.execute(f"PRAGMA synchronous={sync}") + connection.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, v TEXT)") + samples = [] + for index in range(1000): + started = time.perf_counter() + connection.execute("BEGIN IMMEDIATE") + connection.execute("INSERT INTO t (v) VALUES (?)", (str(index),)) + connection.execute("COMMIT") + samples.append(time.perf_counter() - started) + samples.sort() + print(f"sqlite WAL synchronous={sync}: 1000 commits total={sum(samples):.2f}s " + f"p50={samples[500]*1000:.2f}ms p99={samples[990]*1000:.2f}ms max={samples[-1]*1000:.2f}ms") + PYEOF + if [[ "${RUNNER_OS}" == "macOS" ]]; then + bin="$(brew --prefix postgresql@16)/bin" + else + bin="$(cygpath -u "${PGBIN}")" + fi + echo "CREATE TABLE IF NOT EXISTS t (id bigserial PRIMARY KEY, v text);" \ + | "${bin}/psql" -q -h 127.0.0.1 -p 5433 -U postgres postgres + echo "INSERT INTO t (v) VALUES ('x');" > "${RUNNER_TEMP}/insert.sql" + "${bin}/pgbench" -n -h 127.0.0.1 -p 5433 -U postgres -c 1 -t 1000 \ + -f "${RUNNER_TEMP}/insert.sql" postgres | grep -E "latency|tps" + for name in \ + platform::tests::test_webhook::given_a_location_outage_when_webhook_delivery_retries_then_all_health_facts_survive_restart_and_lost_acknowledgements \ + runs::tests::test_group_admission::given_a_large_healthy_project_with_blocked_groups_when_claiming_then_unrelated_work_has_a_bounded_write_lock \ + scheduler::tests::contracts::ticks::test_retention::given_expired_covered_ticks_when_pruned_and_cursor_reset_then_run_linked_and_latest_ticks_survive_without_resurrection \ + sensors::tests::test_nodes::given_any_run_rebuilds_when_consuming_then_only_genuine_scoped_successes_queue_once_per_finish \ + sensors::tests::test_runs::given_terminal_jobs_and_groups_when_selectors_retry_then_outcomes_scope_and_run_keys_are_preserved \ + sensors::tests::test_transitions::given_independent_check_failures_when_nondecisive_results_arrive_then_only_executed_success_recovers + do + started=$(date +%s) + if cargo test --locked -q -p sqlbuild-orchestration --lib -- --exact "${name}" --test-threads=1 --nocapture \ + > "${RUNNER_TEMP}/serial.log" 2>&1; then status=pass; else status=FAIL; fi + echo "serial ${status} $(( $(date +%s) - started ))s ${name}" + grep -E "admissions in|panicked|Fenced|Elapsed" "${RUNNER_TEMP}/serial.log" | head -5 || true + done - name: PostgreSQL server log if: failure() run: cat "${RUNNER_TEMP}/postgres.log" || true diff --git a/Cargo.lock b/Cargo.lock index 462a1de8ff..959dcdaa0b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4198,7 +4198,6 @@ version = "0.187.3" dependencies = [ "blake3", "chrono", - "command-fds", "ctrlc", "getrandom 0.4.3", "keyring", @@ -4216,6 +4215,7 @@ dependencies = [ "sqlbuild-cursor", "sqlbuild-model-config", "sqlbuild-orchestration", + "sqlbuild-process-ownership", "sqlbuild-recall", "sqlbuild-rules", "tempfile", @@ -4314,6 +4314,18 @@ dependencies = [ "uuid", ] +[[package]] +name = "sqlbuild-process-ownership" +version = "0.187.3" +dependencies = [ + "command-fds", + "libc", + "rustix", + "tempfile", + "uuid", + "windows-sys 0.61.2", +] + [[package]] name = "sqlbuild-project" version = "0.187.3" diff --git a/Cargo.toml b/Cargo.toml index adb985a369..f81489ca9e 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -3,6 +3,7 @@ members = [ "crates/sqlbuild-build-features", "crates/sqlbuild-core", "crates/sqlbuild-cursor", + "crates/sqlbuild-process-ownership", "crates/sqlbuild-state", "crates/sqlbuild-pysource", "crates/sqlbuild-cache", @@ -33,6 +34,7 @@ order = [ "sqlbuild-build-features", "sqlbuild-core", "sqlbuild-cursor", + "sqlbuild-process-ownership", "sqlbuild-state", "sqlbuild-pysource", "sqlbuild-cache", @@ -108,6 +110,7 @@ thiserror = "2" # Windows only: `sqb` ignores Ctrl+C while the Python CLI it started handles it. ctrlc = "3" command-fds = "0.3" +libc = "0.2" quote = "1" rayon = "1" # Static reading of project Python source; the parser fensu uses. @@ -117,7 +120,9 @@ ruff_text_size = "=0.0.4" # Unix only: `sqb` shares the compute log prune lock with Python's `filelock` through `flock`. # Fast deterministic hasher for hot in-memory maps whose iteration order never reaches output. rustc-hash = "2" -# The native SIGTERM handler: sets a flag the CLI watches, without unsafe code in the workspace. +# The native SIGTERM handler: sets a flag the CLI watches, without unsafe code. The only crate +# allowed unsafe code is sqlbuild-process-ownership, for the platform calls that prove which +# process owns a worker claim; every other crate forbids it. signal-hook = { version = "0.3", default-features = false } # Features signal-hook's registry enables, matched in the sqb build through sqlbuild-build-features. errno = "0.3" @@ -160,6 +165,7 @@ sqlbuild-refactor = { path = "crates/sqlbuild-refactor" } sqlbuild-rules = { path = "crates/sqlbuild-rules" } sqlbuild-model-config = { path = "crates/sqlbuild-model-config" } sqlbuild-orchestration = { path = "crates/sqlbuild-orchestration" } +sqlbuild-process-ownership = { path = "crates/sqlbuild-process-ownership" } sqlbuild-project = { path = "crates/sqlbuild-project" } sqlbuild-pysource = { path = "crates/sqlbuild-pysource" } sqlbuild-recall = { path = "crates/sqlbuild-recall" } diff --git a/crates/README.md b/crates/README.md index 7f3820572c..522cf3fbf1 100644 --- a/crates/README.md +++ b/crates/README.md @@ -9,6 +9,7 @@ that can be tested and benchmarked with `cargo test` alone. | `sqlbuild-build-features` | No code. Enables, in the `sqb` build, the dependency features the extension build resolves, so Cargo compiles the crates both builds share (polyglot-sql, `sqlbuild-analysis`, `sqlbuild-rules` and the rest) once. `make check-build-features` (part of `make check-ci`) fails when the two builds resolve a shared crate differently and names the features to add. | | `sqlbuild-core` | Constants shared by several crates, the panic boundary that turns an unwinding native failure into a named compiler error, passive identity, diagnostic and SQL value types, text positions, a JSON emitter whose output matches Python's `json.dumps` and `orjson` byte for byte, and run-cost allocation with exact Python `decimal` default-context arithmetic and Snowflake query-history SQL. | | `sqlbuild-cursor` | Microbatch cursor algebra: typed cursor values, grain flooring and stepping, half-open interval clamping, merging and batch splitting, timestamp text normalization and calendar durations, with Python `datetime` results and errors. Zones other than fixed offsets are resolved through a caller-supplied lookup. | +| `sqlbuild-process-ownership` | Proof that a live process owns a scheduler worker claim, and the only way to stop one: an inherited claim lock on Linux and macOS, checked through `/proc` and a pidfd or through libproc with a start-time recheck, and a private named Job object on Windows. It is the only crate allowed unsafe code, for those platform calls. | | `sqlbuild-state` | Append-only warehouse state rows: content-addressed migration event IDs, typed event SQL literals and statements, and source freshness data-version normalization, hashing, equivalence, age policy and latest-state read SQL, byte-compatible with the rows Python wrote. It also owns execution history: lifecycle event batch digests and conflicts, run projection, event and run cursors, filter and page validation, and the SQLite and PostgreSQL statements the native history backends run. | | `sqlbuild-pysource` | Static reading of project Python source with the ruff parser: macro module structure, literal SQL arguments, definition locations and string-literal edits. It never imports or runs user code; CPython stays the authority on whether a file is valid Python. | | `sqlbuild-cache` | The shared native store every native cache uses (one atomically replaced, checksummed file per cache kind, keyed by content digests) and the one fingerprinting scheme for cache keys: the content digest and the project file fingerprint. | @@ -36,7 +37,7 @@ sqlbuild-python -> sqlbuild-project -> sqlbuild-refactor -> sqlbuild-rules -> sqlbuild-analysis -> sqlbuild-attachments -> sqlbuild-model-config -> sqlbuild-scopes -> sqlbuild-render -> sqlbuild-discovery -> sqlbuild-host -> sqlbuild-config -> sqlbuild-sqltext -> sqlbuild-cache -> sqlbuild-pysource -> sqlbuild-state -> sqlbuild-cursor -> sqlbuild-core -sqlbuild-cli -> sqlbuild-cache, sqlbuild-core, sqlbuild-build-features +sqlbuild-cli -> sqlbuild-cache, sqlbuild-core, sqlbuild-build-features, sqlbuild-process-ownership ``` `sqlbuild-scopes` reuses the macro scanner in `sqlbuild-render`; render itself depends only on diff --git a/crates/sqlbuild-cli/Cargo.toml b/crates/sqlbuild-cli/Cargo.toml index 1f24eb892c..25fb89dd21 100644 --- a/crates/sqlbuild-cli/Cargo.toml +++ b/crates/sqlbuild-cli/Cargo.toml @@ -39,8 +39,8 @@ uuid.workspace = true [target.'cfg(unix)'.dependencies] rustix.workspace = true -[target.'cfg(target_os = "linux")'.dependencies] -command-fds.workspace = true +[target.'cfg(any(target_os = "linux", target_os = "macos", windows))'.dependencies] +sqlbuild-process-ownership.workspace = true [target.'cfg(windows)'.dependencies] ctrlc.workspace = true diff --git a/crates/sqlbuild-cli/src/workers/_helpers/child.rs b/crates/sqlbuild-cli/src/workers/_helpers/child.rs index c8c2643d4b..52fcb10161 100644 --- a/crates/sqlbuild-cli/src/workers/_helpers/child.rs +++ b/crates/sqlbuild-cli/src/workers/_helpers/child.rs @@ -2,6 +2,11 @@ use sqlbuild_orchestration::store::errors::ControlError; +#[cfg(any(target_os = "linux", target_os = "macos", windows))] +use sqlbuild_process_ownership::ownership::main::{ + stop_worker::stop_worker, worker_exited::worker_exited, +}; + use crate::workers::_helpers::supervisor::exit_code; use crate::workers::main::terminate_child::terminate_child; use crate::workers::models::{WorkerChild, WorkerExit}; @@ -23,8 +28,8 @@ impl WorkerChild { Self::Local(child) => Ok(child .try_wait()? .map(|exit| WorkerExit::Code(exit_code(exit)))), - #[cfg(target_os = "linux")] - Self::RecoveredLocal(child) => child.try_wait(), + #[cfg(any(target_os = "linux", target_os = "macos", windows))] + Self::RecoveredLocal(owner) => Ok(worker_exited(owner)?.then_some(WorkerExit::Absent)), Self::Docker(child) => child.try_wait().await, } } @@ -44,8 +49,8 @@ impl WorkerChild { pub(in crate::workers) async fn terminate(&mut self) -> Result<(), ControlError> { match self { Self::Local(child) => terminate_child(child), - #[cfg(target_os = "linux")] - Self::RecoveredLocal(child) => child.signal(rustix::process::Signal::TERM), + #[cfg(any(target_os = "linux", target_os = "macos", windows))] + Self::RecoveredLocal(owner) => Ok(stop_worker(owner, false)?), Self::Docker(child) => child.signal("SIGTERM").await, } } @@ -56,8 +61,8 @@ impl WorkerChild { child.start_kill()?; Ok(()) } - #[cfg(target_os = "linux")] - Self::RecoveredLocal(child) => child.signal(rustix::process::Signal::KILL), + #[cfg(any(target_os = "linux", target_os = "macos", windows))] + Self::RecoveredLocal(owner) => Ok(stop_worker(owner, true)?), Self::Docker(child) => child.signal("SIGKILL").await, } } diff --git a/crates/sqlbuild-cli/src/workers/_helpers/local.rs b/crates/sqlbuild-cli/src/workers/_helpers/local.rs deleted file mode 100644 index 82a1c4fba2..0000000000 --- a/crates/sqlbuild-cli/src/workers/_helpers/local.rs +++ /dev/null @@ -1,128 +0,0 @@ -//! Linux dev workers inherit a claim lock before exec; recovery never guesses from a stale PID. - -use std::fs::OpenOptions; -use std::os::unix::fs::MetadataExt; - -use command_fds::{CommandFdExt, FdMapping}; -use rustix::fs::{FlockOperation, flock}; -use rustix::process::{Pid, PidfdFlags, Signal, pidfd_open, pidfd_send_signal}; -use sqlbuild_core::command_output::constants::INVOCATION_CONTEXT_ENV; -use sqlbuild_orchestration::runs::models::RunClaim; -use sqlbuild_orchestration::store::errors::ControlError; -use tokio::process::Command; - -use crate::workers::constants::LOCAL_CLAIM_FD; -use crate::workers::models::{LocalLauncher, LocalWorker, WorkerExit}; - -impl LocalWorker { - pub(in crate::workers) fn open( - location: &LocalLauncher, - claim: &RunClaim, - ) -> Result { - let directory = location.project_dir.join(".sqlbuild/workers"); - std::fs::create_dir_all(&directory)?; - let lock = OpenOptions::new() - .read(true) - .write(true) - .create(true) - .truncate(false) - .open(directory.join(format!("{}.lock", claim.claim_id)))?; - Ok(Self { - lock, - run_id: claim.request.run_id, - stopped: false, - }) - } - - /// Keep the acquired lock until terminal publication, so a delayed spawn cannot start afterward. - pub(in crate::workers) fn try_wait(&mut self) -> Result, ControlError> { - if !self.stopped { - match flock(&self.lock, FlockOperation::NonBlockingLockExclusive) { - Ok(()) => self.stopped = true, - Err(rustix::io::Errno::WOULDBLOCK) => return Ok(None), - Err(error) => return Err(std::io::Error::from(error).into()), - } - } - Ok(Some(WorkerExit::Absent)) - } - - /// Only a kernel handle whose current descriptor and invocation match this claim can be signalled. - pub(in crate::workers) fn signal(&self, signal: Signal) -> Result<(), ControlError> { - let expected = self.lock.metadata()?; - for entry in std::fs::read_dir("/proc")? { - let entry = entry?; - let name = entry.file_name(); - let Some(name) = name.to_str() else { - continue; - }; - let raw = match name.parse::() { - Ok(raw) => raw, - Err(_) => continue, - }; - let Some(pid) = Pid::from_raw(raw) else { - continue; - }; - let descriptor = entry.path().join(format!("fd/{LOCAL_CLAIM_FD}")); - if !same_file(&descriptor, &expected) { - continue; - } - let handle = match pidfd_open(pid, PidfdFlags::empty()) { - Ok(handle) => handle, - Err(rustix::io::Errno::SRCH) => continue, - Err(error) => return Err(std::io::Error::from(error).into()), - }; - if !same_file(&descriptor, &expected) || !self.owns_invocation(&entry.path())? { - continue; - } - match pidfd_send_signal(handle, signal) { - Ok(()) | Err(rustix::io::Errno::SRCH) => {} - Err(error) => return Err(std::io::Error::from(error).into()), - } - } - Ok(()) - } - - fn owns_invocation(&self, process: &std::path::Path) -> Result { - let environment = match std::fs::read(process.join("environ")) { - Ok(environment) => environment, - Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(false), - Err(error) => return Err(error.into()), - }; - let prefix = format!("{INVOCATION_CONTEXT_ENV}="); - for variable in environment.split(|byte| *byte == 0) { - if let Some(context) = variable.strip_prefix(prefix.as_bytes()) { - let context: serde_json::Value = - serde_json::from_slice(context).map_err(|error| { - ControlError::Invalid(format!( - "invalid local worker invocation context: {error}" - )) - })?; - return Ok(context["run_id"].as_str() == Some(&self.run_id.to_string())); - } - } - Ok(false) - } -} - -fn same_file(path: &std::path::Path, expected: &std::fs::Metadata) -> bool { - std::fs::metadata(path) - .is_ok_and(|actual| actual.dev() == expected.dev() && actual.ino() == expected.ino()) -} - -/// The child inherits the lock even before Python starts or a worker handshake can succeed. -pub(in crate::workers) fn inherit_claim( - command: &mut Command, - location: &LocalLauncher, - claim: &RunClaim, -) -> Result<(), ControlError> { - let owner = LocalWorker::open(location, claim)?; - flock(&owner.lock, FlockOperation::NonBlockingLockExclusive).map_err(std::io::Error::from)?; - command - .as_std_mut() - .fd_mappings(vec![FdMapping { - parent_fd: owner.lock.into(), - child_fd: LOCAL_CLAIM_FD, - }]) - .map_err(|error| ControlError::Invalid(format!("local claim descriptor: {error}")))?; - Ok(()) -} diff --git a/crates/sqlbuild-cli/src/workers/_helpers/mod.rs b/crates/sqlbuild-cli/src/workers/_helpers/mod.rs index a0ead33080..b182827d2d 100644 --- a/crates/sqlbuild-cli/src/workers/_helpers/mod.rs +++ b/crates/sqlbuild-cli/src/workers/_helpers/mod.rs @@ -2,8 +2,6 @@ mod child; pub(super) mod configuration; mod docker; mod invocation; -#[cfg(target_os = "linux")] -mod local; mod recovery; mod shutdown; mod supervisor; diff --git a/crates/sqlbuild-cli/src/workers/_helpers/recovery.rs b/crates/sqlbuild-cli/src/workers/_helpers/recovery.rs index f439a58dd6..fb08f51c14 100644 --- a/crates/sqlbuild-cli/src/workers/_helpers/recovery.rs +++ b/crates/sqlbuild-cli/src/workers/_helpers/recovery.rs @@ -9,6 +9,9 @@ use sqlbuild_orchestration::scheduler::models::SchedulerLease; use sqlbuild_orchestration::store::errors::ControlError; use sqlbuild_orchestration::store::models::ControlStore; +#[cfg(any(target_os = "linux", target_os = "macos", windows))] +use sqlbuild_process_ownership::ownership::main::recover_worker::recover_worker; + use crate::workers::models::{ClaimedWorker, DockerWorker, Launcher, WorkerChild, Workers}; impl Workers { @@ -26,7 +29,7 @@ impl Workers { .map(|worker| worker.claim.claim_id) .collect(); for location in self.configuration.locations() { - #[cfg(not(target_os = "linux"))] + #[cfg(not(any(target_os = "linux", target_os = "macos", windows)))] if matches!(location, Launcher::Local(_)) { continue; } @@ -45,11 +48,11 @@ impl Workers { continue; } let child = match location { - #[cfg(target_os = "linux")] + #[cfg(any(target_os = "linux", target_os = "macos", windows))] Launcher::Local(location) => Some(WorkerChild::RecoveredLocal( - crate::workers::models::LocalWorker::open(location, &claim)?, + recover_worker(&location.claim(&claim))?, )), - #[cfg(not(target_os = "linux"))] + #[cfg(not(any(target_os = "linux", target_os = "macos", windows)))] Launcher::Local(_) => continue, Launcher::Docker(location) => match DockerWorker::prepare(location, &claim) { diff --git a/crates/sqlbuild-cli/src/workers/_helpers/supervisor.rs b/crates/sqlbuild-cli/src/workers/_helpers/supervisor.rs index 7c7dd5635f..9411afe732 100644 --- a/crates/sqlbuild-cli/src/workers/_helpers/supervisor.rs +++ b/crates/sqlbuild-cli/src/workers/_helpers/supervisor.rs @@ -19,12 +19,17 @@ use sqlbuild_orchestration::scheduler::main::renew_lease::renew_lease; use sqlbuild_orchestration::scheduler::models::SchedulerLease; use sqlbuild_orchestration::store::errors::ControlError; use sqlbuild_orchestration::store::models::ControlStore; +#[cfg(any(target_os = "linux", target_os = "macos", windows))] +use sqlbuild_process_ownership::ownership::main::{ + prepare_worker::prepare_worker, start_worker::start_worker, +}; use tokio::process::Command; use uuid::Uuid; use crate::workers::_helpers::invocation::{arguments, environment, excluded}; use crate::workers::models::{ - ClaimedWorker, DockerWorker, Launcher, LocalLauncher, WorkerChild, WorkerExit, Workers, + ClaimedWorker, DockerWorker, Launcher, LocalLauncher, LocalSpawn, WorkerChild, WorkerExit, + Workers, }; impl sqlbuild_orchestration::scheduler::types::RunLauncher for Workers { @@ -250,12 +255,9 @@ impl Workers { ) .await?; let launched = match local_command { - Some(mut command) => command - .spawn() - .map(|child| { - worker.child = Some(WorkerChild::Local(child)); - }) - .map_err(ControlError::from), + Some(command) => command.spawn().map(|child| { + worker.child = Some(WorkerChild::Local(child)); + }), None => { if let (Launcher::Docker(location), Some(WorkerChild::Docker(child))) = (location, &worker.child) @@ -305,13 +307,14 @@ impl Workers { } } +/// The worker holds its claim's ownership before it runs any code; other platforms never adopt. fn prepare_spawn( location: &LocalLauncher, claim: &RunClaim, bearer: &str, control_variable: &str, -) -> Result { - let mut command = Command::new(&location.executable); +) -> Result { + let mut command = std::process::Command::new(&location.executable); command .args(arguments(&Launcher::Local(location).scope(), claim)) .arg(&location.project_dir) @@ -324,11 +327,47 @@ fn prepare_spawn( .envs(environment(claim, bearer)) .stdin(Stdio::null()) .stdout(Stdio::null()) - .stderr(Stdio::inherit()) - .kill_on_drop(true); - #[cfg(target_os = "linux")] - crate::workers::_helpers::local::inherit_claim(&mut command, location, claim)?; - Ok(command) + .stderr(Stdio::inherit()); + #[cfg(any(target_os = "linux", target_os = "macos", windows))] + let (command, ownership) = prepare_worker(command, &location.claim(claim))?; + let mut command = Command::from(command); + command.kill_on_drop(true); + Ok(LocalSpawn { + command, + #[cfg(any(target_os = "linux", target_os = "macos", windows))] + ownership, + }) +} + +impl LocalLauncher { + #[cfg(any(target_os = "linux", target_os = "macos", windows))] + pub(in crate::workers) fn claim<'a>( + &'a self, + claim: &RunClaim, + ) -> sqlbuild_process_ownership::ownership::models::Claim<'a> { + sqlbuild_process_ownership::ownership::models::Claim { + project_dir: &self.project_dir, + location_id: self.location_id, + claim_id: claim.claim_id, + } + } +} + +impl LocalSpawn { + /// A child whose ownership could not be completed is killed before the error is returned. + fn spawn(mut self) -> Result { + let mut child = self.command.spawn()?; + #[cfg(any(target_os = "linux", target_os = "macos", windows))] + if let Err(error) = child + .id() + .ok_or_else(|| std::io::Error::other("worker exited before it was owned")) + .and_then(|id| start_worker(self.ownership, id)) + { + drop(child.start_kill()); + return Err(error.into()); + } + Ok(child) + } } impl ClaimedWorker { diff --git a/crates/sqlbuild-cli/src/workers/constants.rs b/crates/sqlbuild-cli/src/workers/constants.rs index 06a889dbfb..881c48bea1 100644 --- a/crates/sqlbuild-cli/src/workers/constants.rs +++ b/crates/sqlbuild-cli/src/workers/constants.rs @@ -1,7 +1,3 @@ //! Immutable Docker image identities use the full SHA-256 hexadecimal digest. pub(super) const IMAGE_ID_HEX_LENGTH: usize = 64; - -/// Keep inherited claim ownership away from standard IO and the runtime's usual low descriptors. -#[cfg(target_os = "linux")] -pub(super) const LOCAL_CLAIM_FD: i32 = 198; diff --git a/crates/sqlbuild-cli/src/workers/models.rs b/crates/sqlbuild-cli/src/workers/models.rs index 35639fe18e..e6eb0e5abb 100644 --- a/crates/sqlbuild-cli/src/workers/models.rs +++ b/crates/sqlbuild-cli/src/workers/models.rs @@ -13,18 +13,17 @@ use uuid::Uuid; #[derive(Debug)] pub(super) enum WorkerChild { Local(Child), - #[cfg(target_os = "linux")] - RecoveredLocal(LocalWorker), + #[cfg(any(target_os = "linux", target_os = "macos", windows))] + RecoveredLocal(sqlbuild_process_ownership::ownership::models::RecoveredOwnership), Docker(DockerWorker), } -/// The inherited lock proves physical ownership without relying on a remembered process ID. -#[cfg(target_os = "linux")] +/// A local worker command with the claim ownership it must hold before running any code. #[derive(Debug)] -pub(super) struct LocalWorker { - pub(super) lock: std::fs::File, - pub(super) run_id: Uuid, - pub(super) stopped: bool, +pub(super) struct LocalSpawn { + pub(super) command: tokio::process::Command, + #[cfg(any(target_os = "linux", target_os = "macos", windows))] + pub(super) ownership: sqlbuild_process_ownership::ownership::models::PendingOwnership, } #[derive(Clone, Copy, Debug)] diff --git a/crates/sqlbuild-process-ownership/Cargo.toml b/crates/sqlbuild-process-ownership/Cargo.toml new file mode 100644 index 0000000000..ffd57c826d --- /dev/null +++ b/crates/sqlbuild-process-ownership/Cargo.toml @@ -0,0 +1,46 @@ +[package] +name = "sqlbuild-process-ownership" +version.workspace = true +description = "Proof that a live process owns a scheduler worker claim, and the only way to stop one, on Linux, macOS and Windows." +edition.workspace = true +license.workspace = true +publish.workspace = true +rust-version.workspace = true + +[dependencies] +uuid.workspace = true + +[target.'cfg(unix)'.dependencies] +rustix.workspace = true + +[target.'cfg(any(target_os = "linux", target_os = "macos"))'.dependencies] +command-fds.workspace = true + +[target.'cfg(target_os = "macos")'.dependencies] +libc.workspace = true + +[target.'cfg(windows)'.dependencies] +windows-sys = { workspace = true, features = [ + "Win32_Foundation", + "Win32_Security", + "Win32_Security_Authorization", + "Win32_Storage_FileSystem", + "Win32_System_Diagnostics_ToolHelp", + "Win32_System_JobObjects", + "Win32_System_SystemServices", + "Win32_System_Threading", +] } + +[dev-dependencies] +tempfile.workspace = true + +# The workspace forbids unsafe code; this crate alone denies it, so each platform call opts in +# with an item-level allow and a SAFETY comment, behind the safe API in `ownership::main`. +[lints.rust] +unsafe_code = "deny" +unreachable_pub = "deny" +unused_must_use = "deny" + +[lints.clippy] +await_holding_lock = "deny" +undocumented_unsafe_blocks = "deny" diff --git a/crates/sqlbuild-process-ownership/src/lib.rs b/crates/sqlbuild-process-ownership/src/lib.rs new file mode 100644 index 0000000000..7facf20341 --- /dev/null +++ b/crates/sqlbuild-process-ownership/src/lib.rs @@ -0,0 +1,3 @@ +//! Proof that a live process owns a scheduler worker claim, and the only way to stop one. + +pub mod ownership; diff --git a/crates/sqlbuild-process-ownership/src/ownership/_helpers/linux.rs b/crates/sqlbuild-process-ownership/src/ownership/_helpers/linux.rs new file mode 100644 index 0000000000..04ba186197 --- /dev/null +++ b/crates/sqlbuild-process-ownership/src/ownership/_helpers/linux.rs @@ -0,0 +1,54 @@ +//! Linux: signal through a pidfd only after `/proc` shows the claim descriptor and marker. + +use std::os::unix::fs::MetadataExt; +use std::path::Path; + +use rustix::process::{Pid, PidfdFlags, Signal, pidfd_open, pidfd_send_signal}; + +use crate::ownership::_helpers::lock::carries_claim; +use crate::ownership::constants::CLAIM_FD; +use crate::ownership::models::RecoveredOwnership; + +fn same_file(path: &Path, expected: &std::fs::Metadata) -> bool { + std::fs::metadata(path) + .is_ok_and(|actual| actual.dev() == expected.dev() && actual.ino() == expected.ino()) +} + +/// The descriptor is checked again after the pidfd pins the process, so a reused PID is never hit. +pub(crate) fn signal_owners(owner: &RecoveredOwnership, force: bool) -> std::io::Result<()> { + let signal = if force { Signal::KILL } else { Signal::TERM }; + let expected = owner.lock.metadata()?; + for entry in std::fs::read_dir("/proc")? { + let entry = entry?; + let name = entry.file_name(); + let raw = match name.to_str().map(str::parse::) { + Some(Ok(raw)) => raw, + Some(Err(_)) | None => continue, + }; + let Some(pid) = Pid::from_raw(raw) else { + continue; + }; + let descriptor = entry.path().join(format!("fd/{CLAIM_FD}")); + if !same_file(&descriptor, &expected) { + continue; + } + let handle = match pidfd_open(pid, PidfdFlags::empty()) { + Ok(handle) => handle, + Err(rustix::io::Errno::SRCH) => continue, + Err(error) => return Err(error.into()), + }; + let environment = match std::fs::read(entry.path().join("environ")) { + Ok(environment) => environment, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => continue, + Err(error) => return Err(error), + }; + if !same_file(&descriptor, &expected) || !carries_claim(&environment, owner.claim_id) { + continue; + } + match pidfd_send_signal(handle, signal) { + Ok(()) | Err(rustix::io::Errno::SRCH) => {} + Err(error) => return Err(error.into()), + } + } + Ok(()) +} diff --git a/crates/sqlbuild-process-ownership/src/ownership/_helpers/lock.rs b/crates/sqlbuild-process-ownership/src/ownership/_helpers/lock.rs new file mode 100644 index 0000000000..d2bdadbba0 --- /dev/null +++ b/crates/sqlbuild-process-ownership/src/ownership/_helpers/lock.rs @@ -0,0 +1,42 @@ +//! Unix claim locks: inherited across exec, held exactly as long as some owner lives. + +use std::fs::{File, OpenOptions}; +use std::process::Command; + +use command_fds::{CommandFdExt, FdMapping}; +use rustix::fs::{FlockOperation, flock}; +use uuid::Uuid; + +use crate::ownership::constants::{CLAIM_FD, WORKER_CLAIM_ENV}; +use crate::ownership::models::Claim; + +pub(crate) fn open_lock(claim: &Claim<'_>) -> std::io::Result { + let directory = claim.project_dir.join(".sqlbuild/workers"); + std::fs::create_dir_all(&directory)?; + OpenOptions::new() + .read(true) + .write(true) + .create(true) + .truncate(false) + .open(directory.join(format!("{}.lock", claim.claim_id))) +} + +pub(crate) fn inherit_lock(mut command: Command, claim: &Claim<'_>) -> std::io::Result { + let lock = open_lock(claim)?; + flock(&lock, FlockOperation::NonBlockingLockExclusive)?; + command + .fd_mappings(vec![FdMapping { + parent_fd: lock.into(), + child_fd: CLAIM_FD, + }]) + .map_err(|error| std::io::Error::other(format!("claim descriptor: {error}")))?; + Ok(command) +} + +/// `environment` is a process's initial NUL-separated `NAME=value` block. +pub(crate) fn carries_claim(environment: &[u8], claim_id: Uuid) -> bool { + let expected = format!("{WORKER_CLAIM_ENV}={claim_id}"); + environment + .split(|byte| *byte == 0) + .any(|variable| variable == expected.as_bytes()) +} diff --git a/crates/sqlbuild-process-ownership/src/ownership/_helpers/macos.rs b/crates/sqlbuild-process-ownership/src/ownership/_helpers/macos.rs new file mode 100644 index 0000000000..1615c7506c --- /dev/null +++ b/crates/sqlbuild-process-ownership/src/ownership/_helpers/macos.rs @@ -0,0 +1,167 @@ +//! macOS has no pidfd: only a PID reused in the microseconds after the last recheck could be hit. + +use std::mem::{MaybeUninit, size_of}; +use std::os::unix::fs::MetadataExt; + +use libc::c_int; +use rustix::process::{Pid, Signal, kill_process}; + +use crate::ownership::_helpers::lock::carries_claim; +use crate::ownership::constants::{CLAIM_FD, PROC_PIDFDVNODEPATHINFO}; +use crate::ownership::models::{RecoveredOwnership, VnodeDescriptor}; + +pub(crate) fn signal_owners(owner: &RecoveredOwnership, force: bool) -> std::io::Result<()> { + let signal = if force { Signal::KILL } else { Signal::TERM }; + let expected = owner.lock.metadata()?; + for raw in process_ids()? { + if !holds_descriptor(raw, &expected) { + continue; + } + let Some(started) = start_time(raw) else { + continue; + }; + let Some(environment) = initial_environment(raw) else { + continue; + }; + if !carries_claim(&environment, owner.claim_id) + || !holds_descriptor(raw, &expected) + || start_time(raw) != Some(started) + { + continue; + } + let Some(pid) = Pid::from_raw(raw) else { + continue; + }; + match kill_process(pid, signal) { + Ok(()) | Err(rustix::io::Errno::SRCH) => {} + Err(error) => return Err(error.into()), + } + } + Ok(()) +} + +/// Every current process ID; the list is retaken whenever the buffer may have been too small. +#[allow(unsafe_code)] +fn process_ids() -> std::io::Result> { + loop { + // SAFETY: a null buffer with size zero asks libproc only for the process count. + let count = unsafe { libc::proc_listallpids(std::ptr::null_mut(), 0) }; + let count = usize::try_from(count).map_err(|_| std::io::Error::last_os_error())?; + let mut pids: Vec = vec![0; count + 64]; + let bytes = c_int::try_from(pids.len() * size_of::()) + .map_err(|_| std::io::Error::other("process list too large"))?; + // SAFETY: `pids` is writable for exactly `bytes` bytes of `c_int` process IDs. + let written = unsafe { libc::proc_listallpids(pids.as_mut_ptr().cast(), bytes) }; + let written = usize::try_from(written).map_err(|_| std::io::Error::last_os_error())?; + if written < pids.len() { + pids.truncate(written); + return Ok(pids); + } + } +} + +/// The process's claim descriptor currently refers to the same vnode as `expected`. +#[allow(unsafe_code)] +fn holds_descriptor(pid: c_int, expected: &std::fs::Metadata) -> bool { + let mut info = MaybeUninit::::zeroed(); + let Ok(size) = c_int::try_from(size_of::()) else { + return false; + }; + // SAFETY: `info` is writable for one `vnode_fdinfowithpath` record of `size` bytes. + let written = unsafe { + libc::proc_pidfdinfo( + pid, + CLAIM_FD, + PROC_PIDFDVNODEPATHINFO, + info.as_mut_ptr().cast(), + size, + ) + }; + if written != size { + return false; + } + // SAFETY: the record was zero-initialised and libproc filled all `size` bytes of it. + let stat = unsafe { info.assume_init() }.vnode.vip_vi.vi_stat; + stat.vst_ino == expected.ino() && i64::from(stat.vst_dev as i32) as u64 == expected.dev() +} + +/// Seconds and microseconds of process start; a reused PID has a different start time. +#[allow(unsafe_code)] +fn start_time(pid: c_int) -> Option<(u64, u64)> { + let mut info = MaybeUninit::::zeroed(); + let Ok(size) = c_int::try_from(size_of::()) else { + return None; + }; + // SAFETY: `info` is writable for one `proc_bsdinfo` record of `size` bytes. + let written = unsafe { + libc::proc_pidinfo( + pid, + libc::PROC_PIDTBSDINFO, + 0, + info.as_mut_ptr().cast(), + size, + ) + }; + if written != size { + return None; + } + // SAFETY: the record was zero-initialised and libproc filled all `size` bytes of it. + let info = unsafe { info.assume_init() }; + Some((info.pbi_start_tvsec, info.pbi_start_tvusec)) +} + +/// The NUL-separated environment the process was executed with, from `KERN_PROCARGS2`. +#[allow(unsafe_code)] +fn initial_environment(pid: c_int) -> Option> { + let mut maximum: c_int = 0; + let mut length = size_of::(); + let mut name = [libc::CTL_KERN, libc::KERN_ARGMAX]; + // SAFETY: `maximum` is writable for `length` bytes; no new value is set. + let status = unsafe { + libc::sysctl( + name.as_mut_ptr(), + 2, + (&raw mut maximum).cast(), + &raw mut length, + std::ptr::null_mut(), + 0, + ) + }; + if status != 0 { + return None; + } + let Ok(maximum) = usize::try_from(maximum) else { + return None; + }; + let mut buffer = vec![0_u8; maximum]; + let mut length = buffer.len(); + let mut name = [libc::CTL_KERN, libc::KERN_PROCARGS2, pid]; + // SAFETY: `buffer` is writable for `length` bytes; no new value is set. + let status = unsafe { + libc::sysctl( + name.as_mut_ptr(), + 3, + buffer.as_mut_ptr().cast(), + &raw mut length, + std::ptr::null_mut(), + 0, + ) + }; + if status != 0 { + return None; + } + buffer.truncate(length); + environment_block(&buffer).map(<[u8]>::to_vec) +} + +/// `KERN_PROCARGS2` holds `argc`, the executable path, NUL padding, the arguments, then the environment. +fn environment_block(arguments: &[u8]) -> Option<&[u8]> { + let (count, mut rest) = arguments.split_first_chunk::<4>()?; + let count = u32::from_ne_bytes(*count); + rest = rest.get(rest.iter().position(|byte| *byte == 0)?..)?; + rest = rest.get(rest.iter().position(|byte| *byte != 0)?..)?; + for _ in 0..count { + rest = rest.get(rest.iter().position(|byte| *byte == 0)? + 1..)?; + } + Some(rest) +} diff --git a/crates/sqlbuild-process-ownership/src/ownership/_helpers/mod.rs b/crates/sqlbuild-process-ownership/src/ownership/_helpers/mod.rs new file mode 100644 index 0000000000..c4132976a6 --- /dev/null +++ b/crates/sqlbuild-process-ownership/src/ownership/_helpers/mod.rs @@ -0,0 +1,8 @@ +#[cfg(target_os = "linux")] +pub(crate) mod linux; +#[cfg(unix)] +pub(crate) mod lock; +#[cfg(target_os = "macos")] +pub(crate) mod macos; +#[cfg(windows)] +pub(crate) mod windows; diff --git a/crates/sqlbuild-process-ownership/src/ownership/_helpers/windows.rs b/crates/sqlbuild-process-ownership/src/ownership/_helpers/windows.rs new file mode 100644 index 0000000000..b65e4319a6 --- /dev/null +++ b/crates/sqlbuild-process-ownership/src/ownership/_helpers/windows.rs @@ -0,0 +1,319 @@ +//! Windows: each claim's worker runs in a private named job owned by, and open only to, this account. + +use std::os::windows::io::{AsRawHandle, FromRawHandle, OwnedHandle}; + +use windows_sys::Win32::Foundation::{ + DuplicateHandle, ERROR_ALREADY_EXISTS, ERROR_FILE_NOT_FOUND, GetLastError, HANDLE, + INVALID_HANDLE_VALUE, LocalFree, +}; +use windows_sys::Win32::Security::Authorization::{ + ConvertSidToStringSidW, ConvertStringSecurityDescriptorToSecurityDescriptorW, GetSecurityInfo, + SDDL_REVISION_1, SE_KERNEL_OBJECT, +}; +use windows_sys::Win32::Security::{ + EqualSid, GetTokenInformation, OWNER_SECURITY_INFORMATION, PSECURITY_DESCRIPTOR, PSID, + SECURITY_ATTRIBUTES, TOKEN_QUERY, TOKEN_USER, TokenUser, +}; +use windows_sys::Win32::Storage::FileSystem::READ_CONTROL; +use windows_sys::Win32::System::Diagnostics::ToolHelp::{ + CreateToolhelp32Snapshot, TH32CS_SNAPTHREAD, THREADENTRY32, Thread32First, Thread32Next, +}; +use windows_sys::Win32::System::JobObjects::{ + AssignProcessToJobObject, CreateJobObjectW, JOBOBJECT_BASIC_ACCOUNTING_INFORMATION, + JobObjectBasicAccountingInformation, OpenJobObjectW, QueryInformationJobObject, + TerminateJobObject, +}; +use windows_sys::Win32::System::SystemServices::{JOB_OBJECT_QUERY, JOB_OBJECT_TERMINATE}; +use windows_sys::Win32::System::Threading::{ + GetCurrentProcess, OpenProcess, OpenProcessToken, OpenThread, PROCESS_DUP_HANDLE, + PROCESS_SET_QUOTA, PROCESS_TERMINATE, ResumeThread, THREAD_SUSPEND_RESUME, +}; + +use crate::ownership::constants::JOB_PREFIX; +use crate::ownership::models::Claim; + +pub(crate) fn job_name(claim: &Claim<'_>) -> Vec { + format!("{JOB_PREFIX}{}-{}", claim.location_id, claim.claim_id) + .encode_utf16() + .chain(std::iter::once(0)) + .collect() +} + +/// Take ownership of a handle a successful Win32 call just returned. +#[allow(unsafe_code)] +fn owned(handle: HANDLE) -> OwnedHandle { + // SAFETY: callers pass a valid, open handle that nothing else closes. + unsafe { OwnedHandle::from_raw_handle(handle) } +} + +/// The `TOKEN_USER` record of this process; its SID points into the returned buffer. +#[allow(unsafe_code)] +pub(crate) fn account() -> std::io::Result> { + let mut token: HANDLE = std::ptr::null_mut(); + // SAFETY: `token` receives a new query handle to this process's own token. + if unsafe { OpenProcessToken(GetCurrentProcess(), TOKEN_QUERY, &raw mut token) } == 0 { + return Err(std::io::Error::last_os_error()); + } + let token = owned(token); + let mut length = 0_u32; + // SAFETY: a null buffer of size zero asks only for the required length. + unsafe { + GetTokenInformation( + token.as_raw_handle(), + TokenUser, + std::ptr::null_mut(), + 0, + &raw mut length, + ) + }; + let mut buffer = vec![ + 0_u8; + usize::try_from(length) + .unwrap_or(0) + .max(size_of::()) + ]; + let capacity = u32::try_from(buffer.len()).map_err(std::io::Error::other)?; + // SAFETY: `buffer` is writable for `capacity` bytes. + let read = unsafe { + GetTokenInformation( + token.as_raw_handle(), + TokenUser, + buffer.as_mut_ptr().cast(), + capacity, + &raw mut length, + ) + }; + if read == 0 { + return Err(std::io::Error::last_os_error()); + } + Ok(buffer) +} + +#[allow(unsafe_code)] +pub(crate) fn account_sid(account: &[u8]) -> PSID { + // SAFETY: `account` holds a complete `TOKEN_USER` written by `GetTokenInformation`. + unsafe { std::ptr::read_unaligned(account.as_ptr().cast::()) } + .User + .Sid +} + +#[allow(unsafe_code)] +fn sid_text(sid: PSID) -> std::io::Result { + let mut text: windows_sys::core::PWSTR = std::ptr::null_mut(); + // SAFETY: `sid` is valid; `text` receives a LocalAlloc string freed below. + if unsafe { ConvertSidToStringSidW(sid, &raw mut text) } == 0 { + return Err(std::io::Error::last_os_error()); + } + // SAFETY: `text` is a NUL-terminated wide string; its length is found before it is read. + let units = unsafe { + let length = (0..).take_while(|&index| *text.add(index) != 0).count(); + std::slice::from_raw_parts(text, length) + }; + let sid = String::from_utf16_lossy(units); + // SAFETY: `text` came from `ConvertSidToStringSidW` and is not used again. + unsafe { LocalFree(text.cast()) }; + Ok(sid) +} + +/// The job's owner and only access entry are this process's account; an existing name is refused. +#[allow(unsafe_code)] +pub(crate) fn create_job(claim: &Claim<'_>) -> std::io::Result { + let account = account()?; + let sid = sid_text(account_sid(&account))?; + let descriptor_text: Vec = format!("O:{sid}D:P(A;;GA;;;{sid})") + .encode_utf16() + .chain(std::iter::once(0)) + .collect(); + let mut descriptor: PSECURITY_DESCRIPTOR = std::ptr::null_mut(); + // SAFETY: `descriptor` receives a LocalAlloc security descriptor freed below. + let converted = unsafe { + ConvertStringSecurityDescriptorToSecurityDescriptorW( + descriptor_text.as_ptr(), + SDDL_REVISION_1, + &raw mut descriptor, + std::ptr::null_mut(), + ) + }; + if converted == 0 { + return Err(std::io::Error::last_os_error()); + } + let attributes = SECURITY_ATTRIBUTES { + nLength: u32::try_from(size_of::()).unwrap_or(0), + lpSecurityDescriptor: descriptor, + bInheritHandle: 0, + }; + let name = job_name(claim); + // SAFETY: `attributes` and `name` outlive the call. + let job = unsafe { CreateJobObjectW(&raw const attributes, name.as_ptr()) }; + // SAFETY: reads this thread's last error before any other call can change it. + let error = unsafe { GetLastError() }; + // SAFETY: `descriptor` came from the conversion above and is not used again. + unsafe { LocalFree(descriptor) }; + if job.is_null() { + return Err(std::io::Error::from_raw_os_error(error.cast_signed())); + } + let job = owned(job); + if error == ERROR_ALREADY_EXISTS { + return Err(std::io::Error::new( + std::io::ErrorKind::AlreadyExists, + "a worker job with this claim's name already exists", + )); + } + Ok(job) +} + +#[allow(unsafe_code)] +pub(crate) fn open_job(claim: &Claim<'_>) -> std::io::Result> { + let name = job_name(claim); + // SAFETY: `name` outlives the call. + let job = unsafe { + OpenJobObjectW( + JOB_OBJECT_QUERY | JOB_OBJECT_TERMINATE | READ_CONTROL, + 0, + name.as_ptr(), + ) + }; + if job.is_null() { + let error = std::io::Error::last_os_error(); + return match error.raw_os_error() { + Some(code) if code.cast_unsigned() == ERROR_FILE_NOT_FOUND => Ok(None), + _ => Err(error), + }; + } + let job = owned(job); + Ok(owned_by_this_account(&job)?.then_some(job)) +} + +#[allow(unsafe_code)] +fn owned_by_this_account(job: &OwnedHandle) -> std::io::Result { + let account = account()?; + let mut owner: PSID = std::ptr::null_mut(); + let mut descriptor: PSECURITY_DESCRIPTOR = std::ptr::null_mut(); + // SAFETY: the handle has READ_CONTROL; `descriptor` is LocalAlloc'd and freed below. + let status = unsafe { + GetSecurityInfo( + job.as_raw_handle(), + SE_KERNEL_OBJECT, + OWNER_SECURITY_INFORMATION, + &raw mut owner, + std::ptr::null_mut(), + std::ptr::null_mut(), + std::ptr::null_mut(), + &raw mut descriptor, + ) + }; + if status != 0 { + return Err(std::io::Error::from_raw_os_error(status.cast_signed())); + } + // SAFETY: `owner` points into `descriptor`, which is still allocated; `account` holds our SID. + let same = unsafe { EqualSid(owner, account_sid(&account)) } != 0; + // SAFETY: `descriptor` came from `GetSecurityInfo` and `owner` is not used again. + unsafe { LocalFree(descriptor) }; + Ok(same) +} + +#[allow(unsafe_code)] +pub(crate) fn active_processes(job: &OwnedHandle) -> std::io::Result { + let mut info = JOBOBJECT_BASIC_ACCOUNTING_INFORMATION::default(); + let size = u32::try_from(size_of::()) + .map_err(std::io::Error::other)?; + // SAFETY: `info` is writable for one accounting record of `size` bytes. + let queried = unsafe { + QueryInformationJobObject( + job.as_raw_handle(), + JobObjectBasicAccountingInformation, + (&raw mut info).cast(), + size, + std::ptr::null_mut(), + ) + }; + if queried == 0 { + return Err(std::io::Error::last_os_error()); + } + Ok(info.ActiveProcesses) +} + +#[allow(unsafe_code)] +pub(crate) fn terminate(job: &OwnedHandle) -> std::io::Result<()> { + // SAFETY: the handle is open with terminate access to a job this account owns. + if unsafe { TerminateJobObject(job.as_raw_handle(), 1) } == 0 { + return Err(std::io::Error::last_os_error()); + } + Ok(()) +} + +/// Put the suspended child in the job, give it a job handle, then let it run. +#[allow(unsafe_code)] +pub(crate) fn adopt(job: &OwnedHandle, process_id: u32) -> std::io::Result<()> { + // SAFETY: opens our own unreaped, suspended child, so its ID cannot have been reused. + let process = unsafe { + OpenProcess( + PROCESS_SET_QUOTA | PROCESS_TERMINATE | PROCESS_DUP_HANDLE, + 0, + process_id, + ) + }; + if process.is_null() { + return Err(std::io::Error::last_os_error()); + } + let process = owned(process); + // SAFETY: both handles are open; the child is suspended and has run no code. + if unsafe { AssignProcessToJobObject(job.as_raw_handle(), process.as_raw_handle()) } == 0 { + return Err(std::io::Error::last_os_error()); + } + let mut copy: HANDLE = std::ptr::null_mut(); + // SAFETY: duplicates our job handle into the child, which keeps the named job alive until it exits. + let duplicated = unsafe { + DuplicateHandle( + GetCurrentProcess(), + job.as_raw_handle(), + process.as_raw_handle(), + &raw mut copy, + JOB_OBJECT_QUERY, + 0, + 0, + ) + }; + if duplicated == 0 { + return Err(std::io::Error::last_os_error()); + } + resume(process_id) +} + +#[allow(unsafe_code)] +fn resume(process_id: u32) -> std::io::Result<()> { + // SAFETY: takes a snapshot of all threads; the handle is closed when `snapshot` drops. + let snapshot = unsafe { CreateToolhelp32Snapshot(TH32CS_SNAPTHREAD, 0) }; + if snapshot == INVALID_HANDLE_VALUE { + return Err(std::io::Error::last_os_error()); + } + let snapshot = owned(snapshot); + let mut entry = THREADENTRY32 { + dwSize: u32::try_from(size_of::()).unwrap_or(0), + ..Default::default() + }; + let mut resumed = 0_usize; + // SAFETY: `entry` is a correctly sized record for this snapshot. + let mut more = unsafe { Thread32First(snapshot.as_raw_handle(), &raw mut entry) } != 0; + while more { + if entry.th32OwnerProcessID == process_id { + // SAFETY: opens a thread of our own suspended child. + let thread = unsafe { OpenThread(THREAD_SUSPEND_RESUME, 0, entry.th32ThreadID) }; + if thread.is_null() { + return Err(std::io::Error::last_os_error()); + } + let thread = owned(thread); + // SAFETY: the thread handle has suspend/resume access. + if unsafe { ResumeThread(thread.as_raw_handle()) } == u32::MAX { + return Err(std::io::Error::last_os_error()); + } + resumed += 1; + } + // SAFETY: `entry` is a correctly sized record for this snapshot. + more = unsafe { Thread32Next(snapshot.as_raw_handle(), &raw mut entry) } != 0; + } + if resumed == 0 { + return Err(std::io::Error::other("worker has no thread to resume")); + } + Ok(()) +} diff --git a/crates/sqlbuild-process-ownership/src/ownership/constants.rs b/crates/sqlbuild-process-ownership/src/ownership/constants.rs new file mode 100644 index 0000000000..b56fbfd5c7 --- /dev/null +++ b/crates/sqlbuild-process-ownership/src/ownership/constants.rs @@ -0,0 +1,16 @@ +//! Names and numbers shared by the launching and the recovering scheduler. + +/// Every owned worker carries its claim ID here, so a stop never reaches another claim's process. +pub const WORKER_CLAIM_ENV: &str = "SQLBUILD_WORKER_CLAIM"; + +/// Keep inherited claim ownership away from standard IO and the runtime's usual low descriptors. +#[cfg(unix)] +pub(crate) const CLAIM_FD: i32 = 198; + +/// libproc's vnode-with-path flavour for one descriptor, absent from the libc bindings. +#[cfg(target_os = "macos")] +pub(crate) const PROC_PIDFDVNODEPATHINFO: i32 = 2; + +/// A session-local name keeps other sessions out; the account-only DACL keeps other users out. +#[cfg(windows)] +pub(crate) const JOB_PREFIX: &str = "Local\\sqlbuild-worker-"; diff --git a/crates/sqlbuild-process-ownership/src/ownership/main/mod.rs b/crates/sqlbuild-process-ownership/src/ownership/main/mod.rs new file mode 100644 index 0000000000..211f452584 --- /dev/null +++ b/crates/sqlbuild-process-ownership/src/ownership/main/mod.rs @@ -0,0 +1,7 @@ +//! The safe API: prepare and start an owned worker, then recover, observe and stop it. + +pub mod prepare_worker; +pub mod recover_worker; +pub mod start_worker; +pub mod stop_worker; +pub mod worker_exited; diff --git a/crates/sqlbuild-process-ownership/src/ownership/main/prepare_worker.rs b/crates/sqlbuild-process-ownership/src/ownership/main/prepare_worker.rs new file mode 100644 index 0000000000..eca8b366e4 --- /dev/null +++ b/crates/sqlbuild-process-ownership/src/ownership/main/prepare_worker.rs @@ -0,0 +1,27 @@ +//! Attach claim ownership to a worker command before it is spawned. + +use std::process::Command; + +use crate::ownership::constants::WORKER_CLAIM_ENV; +use crate::ownership::models::{Claim, PendingOwnership}; + +/// Unix children inherit a locked claim descriptor; Windows children start suspended for their job. +pub fn prepare_worker( + mut command: Command, + claim: &Claim<'_>, +) -> std::io::Result<(Command, PendingOwnership)> { + command.env(WORKER_CLAIM_ENV, claim.claim_id.to_string()); + #[cfg(unix)] + { + let command = crate::ownership::_helpers::lock::inherit_lock(command, claim)?; + Ok((command, PendingOwnership {})) + } + #[cfg(windows)] + { + use std::os::windows::process::CommandExt; + + let job = crate::ownership::_helpers::windows::create_job(claim)?; + command.creation_flags(windows_sys::Win32::System::Threading::CREATE_SUSPENDED); + Ok((command, PendingOwnership { job })) + } +} diff --git a/crates/sqlbuild-process-ownership/src/ownership/main/recover_worker.rs b/crates/sqlbuild-process-ownership/src/ownership/main/recover_worker.rs new file mode 100644 index 0000000000..ad5637bf5e --- /dev/null +++ b/crates/sqlbuild-process-ownership/src/ownership/main/recover_worker.rs @@ -0,0 +1,20 @@ +//! Rebuild a worker's ownership after the scheduler that launched it stopped. + +use crate::ownership::models::{Claim, RecoveredOwnership}; + +/// Never relies on a remembered process ID. +pub fn recover_worker(claim: &Claim<'_>) -> std::io::Result { + #[cfg(unix)] + { + Ok(RecoveredOwnership { + lock: crate::ownership::_helpers::lock::open_lock(claim)?, + claim_id: claim.claim_id, + }) + } + #[cfg(windows)] + { + Ok(RecoveredOwnership { + job: crate::ownership::_helpers::windows::open_job(claim)?, + }) + } +} diff --git a/crates/sqlbuild-process-ownership/src/ownership/main/start_worker.rs b/crates/sqlbuild-process-ownership/src/ownership/main/start_worker.rs new file mode 100644 index 0000000000..e74973061a --- /dev/null +++ b/crates/sqlbuild-process-ownership/src/ownership/main/start_worker.rs @@ -0,0 +1,15 @@ +//! Complete ownership once the prepared command has spawned. + +use crate::ownership::models::PendingOwnership; + +/// On Windows the suspended child joins its job and only then runs; the caller kills it on error. +pub fn start_worker(pending: PendingOwnership, process_id: u32) -> std::io::Result<()> { + #[cfg(unix)] + { + let PendingOwnership {} = pending; + let _ = process_id; + Ok(()) + } + #[cfg(windows)] + crate::ownership::_helpers::windows::adopt(&pending.job, process_id) +} diff --git a/crates/sqlbuild-process-ownership/src/ownership/main/stop_worker.rs b/crates/sqlbuild-process-ownership/src/ownership/main/stop_worker.rs new file mode 100644 index 0000000000..40c81d6d6c --- /dev/null +++ b/crates/sqlbuild-process-ownership/src/ownership/main/stop_worker.rs @@ -0,0 +1,19 @@ +//! Stop a recovered worker, and only processes proven to own its claim. + +use crate::ownership::models::RecoveredOwnership; + +/// `force` sends SIGKILL instead of SIGTERM on Unix; Windows always terminates the claim's job. +pub fn stop_worker(owner: &RecoveredOwnership, force: bool) -> std::io::Result<()> { + #[cfg(target_os = "linux")] + return crate::ownership::_helpers::linux::signal_owners(owner, force); + #[cfg(target_os = "macos")] + return crate::ownership::_helpers::macos::signal_owners(owner, force); + #[cfg(windows)] + { + let _ = force; + owner + .job + .as_ref() + .map_or(Ok(()), crate::ownership::_helpers::windows::terminate) + } +} diff --git a/crates/sqlbuild-process-ownership/src/ownership/main/worker_exited.rs b/crates/sqlbuild-process-ownership/src/ownership/main/worker_exited.rs new file mode 100644 index 0000000000..911b6fc153 --- /dev/null +++ b/crates/sqlbuild-process-ownership/src/ownership/main/worker_exited.rs @@ -0,0 +1,24 @@ +//! Observe whether a recovered worker has exited. + +use crate::ownership::models::RecoveredOwnership; + +/// A Unix scheduler that sees the exit then holds the lock, so no delayed start can follow. +pub fn worker_exited(owner: &RecoveredOwnership) -> std::io::Result { + #[cfg(unix)] + { + use rustix::fs::{FlockOperation, flock}; + + match flock(&owner.lock, FlockOperation::NonBlockingLockExclusive) { + Ok(()) => Ok(true), + Err(rustix::io::Errno::WOULDBLOCK) => Ok(false), + Err(error) => Err(error.into()), + } + } + #[cfg(windows)] + { + match owner.job.as_ref() { + Some(job) => Ok(crate::ownership::_helpers::windows::active_processes(job)? == 0), + None => Ok(true), + } + } +} diff --git a/crates/sqlbuild-process-ownership/src/ownership/mod.rs b/crates/sqlbuild-process-ownership/src/ownership/mod.rs new file mode 100644 index 0000000000..691cfdff96 --- /dev/null +++ b/crates/sqlbuild-process-ownership/src/ownership/mod.rs @@ -0,0 +1,9 @@ +//! Workers hold their claim's ownership from before their first instruction until they exit. + +mod _helpers; +pub mod constants; +pub mod main; +pub mod models; + +#[cfg(test)] +mod tests; diff --git a/crates/sqlbuild-process-ownership/src/ownership/models.rs b/crates/sqlbuild-process-ownership/src/ownership/models.rs new file mode 100644 index 0000000000..3fd82c8a71 --- /dev/null +++ b/crates/sqlbuild-process-ownership/src/ownership/models.rs @@ -0,0 +1,56 @@ +//! A claim's identity and the ownership the launching and recovering schedulers hold for it. + +use std::path::Path; + +use uuid::Uuid; + +/// The project directory holds Unix claim locks; the location makes Windows job names installation-specific. +#[derive(Clone, Copy, Debug)] +pub struct Claim<'a> { + pub project_dir: &'a Path, + pub location_id: Uuid, + pub claim_id: Uuid, +} + +/// Ownership prepared for a worker command that has not started yet. +#[derive(Debug)] +pub struct PendingOwnership { + #[cfg(windows)] + pub(crate) job: std::os::windows::io::OwnedHandle, +} + +/// A restarted scheduler's proof of a worker that may still be running. +#[cfg(unix)] +#[derive(Debug)] +pub struct RecoveredOwnership { + pub(crate) lock: std::fs::File, + pub(crate) claim_id: Uuid, +} + +/// A restarted scheduler's proof of a worker that may still be running. +#[cfg(windows)] +#[derive(Debug)] +pub struct RecoveredOwnership { + pub(crate) job: Option, +} + +/// libproc's `proc_fileinfo`, the fixed prefix of every per-descriptor record. +#[cfg(target_os = "macos")] +#[repr(C)] +#[derive(Clone, Copy, Debug)] +pub(crate) struct ProcFileInfo { + pub(crate) open_flags: u32, + pub(crate) status: u32, + pub(crate) offset: i64, + pub(crate) kind: i32, + pub(crate) guard_flags: u32, +} + +/// libproc's `vnode_fdinfowithpath`: the vnode a descriptor currently refers to. +#[cfg(target_os = "macos")] +#[repr(C)] +#[derive(Clone, Copy, Debug)] +pub(crate) struct VnodeDescriptor { + pub(crate) file: ProcFileInfo, + pub(crate) vnode: libc::vnode_info_path, +} diff --git a/crates/sqlbuild-process-ownership/src/ownership/tests.rs b/crates/sqlbuild-process-ownership/src/ownership/tests.rs new file mode 100644 index 0000000000..167d752f7f --- /dev/null +++ b/crates/sqlbuild-process-ownership/src/ownership/tests.rs @@ -0,0 +1,11 @@ +//! Real child processes stand in for workers; the launching scheduler's view of them is dropped. + +#[path = "tests/helpers.rs"] +mod helpers; +#[cfg(windows)] +#[path = "tests/test_job_access.rs"] +mod test_job_access; +#[path = "tests/test_recovery.rs"] +mod test_recovery; +#[path = "tests/test_types.rs"] +mod test_types; diff --git a/crates/sqlbuild-process-ownership/src/ownership/tests/helpers.rs b/crates/sqlbuild-process-ownership/src/ownership/tests/helpers.rs new file mode 100644 index 0000000000..382881ac2d --- /dev/null +++ b/crates/sqlbuild-process-ownership/src/ownership/tests/helpers.rs @@ -0,0 +1,186 @@ +use std::path::Path; +use std::process::{Child, Command, Stdio}; +use std::time::Duration; + +use uuid::Uuid; + +use crate::ownership::main::prepare_worker::prepare_worker; +use crate::ownership::main::start_worker::start_worker; +use crate::ownership::main::worker_exited::worker_exited; +use crate::ownership::models::Claim; + +pub(super) fn claims(project_dir: &Path) -> [Claim<'_>; 2] { + let location_id = Uuid::now_v7(); + [Uuid::now_v7(), Uuid::now_v7()].map(|claim_id| Claim { + project_dir, + location_id, + claim_id, + }) +} + +#[cfg(unix)] +fn sleeper(seconds: u32) -> Command { + let mut command = Command::new("sh"); + command.args(["-c", &format!("sleep {seconds}")]); + command +} + +#[cfg(windows)] +fn sleeper(seconds: u32) -> Command { + let mut command = Command::new("cmd"); + command.args(["/c", &format!("ping -n {} 127.0.0.1", seconds + 1)]); + command +} + +/// The worker is owned by `claims[holder]`, if any, and carries `claims[marker]`'s ID. +pub(super) fn spawn( + claims: &[Claim<'_>; 2], + (holder, marker): (Option, usize), + seconds: u32, +) -> Child { + let mut command = sleeper(seconds); + command + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::null()); + let (mut command, pending) = holder.map(|index| &claims[index]).into_iter().fold( + (command, None), + |(command, _), claim| { + let (command, pending) = prepare_worker(command, claim).expect("prepare ownership"); + (command, Some(pending)) + }, + ); + command.env( + crate::ownership::constants::WORKER_CLAIM_ENV, + claims[marker].claim_id.to_string(), + ); + let child = command.spawn().expect("spawn worker"); + pending + .map(|pending| start_worker(pending, child.id())) + .transpose() + .expect("start owned worker"); + child +} + +pub(super) fn exited_within(child: &mut Child, within: Duration) -> bool { + let attempts = within.as_millis() / 20 + 1; + (0..attempts).any(|_| { + child.try_wait().expect("observe worker").is_some() || { + std::thread::sleep(Duration::from_millis(20)); + false + } + }) +} + +pub(super) fn recovered_exited_within( + owner: &crate::ownership::models::RecoveredOwnership, + within: Duration, +) -> bool { + let attempts = within.as_millis() / 20 + 1; + (0..attempts).any(|_| { + worker_exited(owner).expect("observe recovered worker") || { + std::thread::sleep(Duration::from_millis(20)); + false + } + }) +} + +pub(super) fn reap(mut child: Child) { + drop(child.kill()); + child.wait().expect("reap worker"); +} + +#[cfg(windows)] +use std::os::windows::io::{AsRawHandle, FromRawHandle, OwnedHandle}; + +#[cfg(windows)] +use windows_sys::Win32::Foundation::HANDLE; +#[cfg(windows)] +use windows_sys::Win32::Security::{ + CreateRestrictedToken, DISABLE_MAX_PRIVILEGE, ImpersonateLoggedOnUser, RevertToSelf, + SID_AND_ATTRIBUTES, TOKEN_DUPLICATE, TOKEN_IMPERSONATE, TOKEN_QUERY, +}; +#[cfg(windows)] +use windows_sys::Win32::System::JobObjects::OpenJobObjectW; +#[cfg(windows)] +use windows_sys::Win32::System::Threading::{GetCurrentProcess, OpenProcessToken}; + +#[cfg(windows)] +use crate::ownership::_helpers::windows::{account, account_sid}; + +/// `Ok(())` when the named job opens with `access`, otherwise the Win32 error code. +#[cfg(windows)] +#[allow(unsafe_code)] +pub(super) fn open_job(name: &[u16], access: u32) -> Result<(), u32> { + // SAFETY: `name` is NUL-terminated and outlives the call. + let job = unsafe { OpenJobObjectW(access, 0, name.as_ptr()) }; + let error = std::io::Error::last_os_error() + .raw_os_error() + .map_or(0, i32::cast_unsigned); + (!job.is_null()) + // SAFETY: a valid handle that nothing else closes. + .then(|| drop(unsafe { OwnedHandle::from_raw_handle(job) })) + .ok_or(error) +} + +/// Same session, same user, but with the user SID deny-only and no privileges. +#[cfg(windows)] +#[allow(unsafe_code)] +fn foreign_token() -> OwnedHandle { + let mut token: HANDLE = std::ptr::null_mut(); + // SAFETY: opens this process's own token. + let opened = unsafe { + OpenProcessToken( + GetCurrentProcess(), + TOKEN_DUPLICATE | TOKEN_QUERY | TOKEN_IMPERSONATE, + &raw mut token, + ) + }; + assert_ne!(opened, 0, "{}", std::io::Error::last_os_error()); + // SAFETY: a valid handle that nothing else closes. + let token = unsafe { OwnedHandle::from_raw_handle(token) }; + let user = account().expect("account"); + let disabled = SID_AND_ATTRIBUTES { + Sid: account_sid(&user), + Attributes: 0, + }; + let mut restricted: HANDLE = std::ptr::null_mut(); + // SAFETY: the SID lives in `user`; `restricted` receives a new token handle. + let created = unsafe { + CreateRestrictedToken( + token.as_raw_handle(), + DISABLE_MAX_PRIVILEGE, + 1, + &raw const disabled, + 0, + std::ptr::null(), + 0, + std::ptr::null(), + &raw mut restricted, + ) + }; + assert_ne!(created, 0, "{}", std::io::Error::last_os_error()); + // SAFETY: a valid handle that nothing else closes. + unsafe { OwnedHandle::from_raw_handle(restricted) } +} + +/// Run `attempt` on a thread that impersonates a foreign identity for its whole duration. +#[cfg(windows)] +#[allow(unsafe_code)] +pub(super) fn as_foreign_identity(attempt: impl FnOnce() -> T + Send) -> T { + std::thread::scope(|scope| { + scope + .spawn(|| { + let token = foreign_token(); + // SAFETY: impersonation applies to this scoped thread only and is reverted below. + let impersonated = unsafe { ImpersonateLoggedOnUser(token.as_raw_handle()) }; + assert_ne!(impersonated, 0, "{}", std::io::Error::last_os_error()); + let result = attempt(); + // SAFETY: ends this thread's impersonation. + assert_ne!(unsafe { RevertToSelf() }, 0); + result + }) + .join() + .expect("foreign identity thread") + }) +} diff --git a/crates/sqlbuild-process-ownership/src/ownership/tests/test_job_access.rs b/crates/sqlbuild-process-ownership/src/ownership/tests/test_job_access.rs new file mode 100644 index 0000000000..5eccb3e148 --- /dev/null +++ b/crates/sqlbuild-process-ownership/src/ownership/tests/test_job_access.rs @@ -0,0 +1,58 @@ +use windows_sys::Win32::Foundation::ERROR_ACCESS_DENIED; +use windows_sys::Win32::System::SystemServices::{JOB_OBJECT_QUERY, JOB_OBJECT_TERMINATE}; + +use crate::ownership::_helpers::windows::job_name; +use crate::ownership::main::recover_worker::recover_worker; +use crate::ownership::main::stop_worker::stop_worker; +use crate::ownership::tests::helpers::{ + as_foreign_identity, claims, exited_within, open_job, reap, spawn, +}; +use crate::ownership::tests::test_types::JobAccessTestCase; + +#[test] +fn given_a_running_worker_job_when_another_identity_opens_it_then_query_and_terminate_are_denied() { + let test_cases = [ + JobAccessTestCase { + description: "terminate access", + access: JOB_OBJECT_TERMINATE, + expected_own_access: Ok(()), + expected_foreign_access: Err(ERROR_ACCESS_DENIED), + }, + JobAccessTestCase { + description: "query access", + access: JOB_OBJECT_QUERY, + expected_own_access: Ok(()), + expected_foreign_access: Err(ERROR_ACCESS_DENIED), + }, + ]; + for test_case in test_cases { + let project = tempfile::tempdir().expect("project"); + let claims = claims(project.path()); + let mut worker = spawn(&claims, (Some(0), 0), 60); + let name = job_name(&claims[0]); + assert_eq!( + open_job(&name, test_case.access), + test_case.expected_own_access, + "{}", + test_case.description + ); + assert_eq!( + as_foreign_identity(|| open_job(&name, test_case.access)), + test_case.expected_foreign_access, + "{}", + test_case.description + ); + assert!( + !exited_within(&mut worker, std::time::Duration::from_millis(300)), + "{}", + test_case.description + ); + stop_worker(&recover_worker(&claims[0]).expect("recover"), true).expect("stop"); + assert!( + exited_within(&mut worker, std::time::Duration::from_secs(5)), + "{}", + test_case.description + ); + reap(worker); + } +} diff --git a/crates/sqlbuild-process-ownership/src/ownership/tests/test_recovery.rs b/crates/sqlbuild-process-ownership/src/ownership/tests/test_recovery.rs new file mode 100644 index 0000000000..8ba089838f --- /dev/null +++ b/crates/sqlbuild-process-ownership/src/ownership/tests/test_recovery.rs @@ -0,0 +1,112 @@ +use std::time::Duration; + +use crate::ownership::main::recover_worker::recover_worker; +use crate::ownership::main::stop_worker::stop_worker; +use crate::ownership::main::worker_exited::worker_exited; +use crate::ownership::tests::helpers::{ + claims, exited_within, reap, recovered_exited_within, spawn, +}; +use crate::ownership::tests::test_types::RecoveryTestCase; + +#[test] +fn given_a_restarted_scheduler_when_it_recovers_and_stops_a_claim_then_only_its_live_owner_is_stopped() + { + let test_cases = [ + RecoveryTestCase { + description: "a live owned worker is recovered as running and stopped gracefully", + holder: Some(0), + marker: 0, + lifetime_seconds: 60, + force: false, + expected_exited_before_recovery: false, + expected_recovered_exited: false, + expected_stopped: true, + }, + RecoveryTestCase { + description: "a live owned worker is stopped forcefully", + holder: Some(0), + marker: 0, + lifetime_seconds: 60, + force: true, + expected_exited_before_recovery: false, + expected_recovered_exited: false, + expected_stopped: true, + }, + RecoveryTestCase { + description: "a worker that already exited releases its claim", + holder: Some(0), + marker: 0, + lifetime_seconds: 0, + force: true, + expected_exited_before_recovery: true, + expected_recovered_exited: true, + expected_stopped: true, + }, + RecoveryTestCase { + description: "another claim's worker carrying this claim's ID, as a reused PID would, is never stopped", + holder: Some(1), + marker: 0, + lifetime_seconds: 60, + force: true, + expected_exited_before_recovery: false, + expected_recovered_exited: true, + expected_stopped: false, + }, + RecoveryTestCase { + description: "an unowned process carrying this claim's ID is never stopped", + holder: None, + marker: 0, + lifetime_seconds: 60, + force: true, + expected_exited_before_recovery: false, + expected_recovered_exited: true, + expected_stopped: false, + }, + RecoveryTestCase { + description: "a holder of this claim's ownership carrying another claim's ID is stopped only where the job is the proof", + holder: Some(0), + marker: 1, + lifetime_seconds: 60, + force: true, + expected_exited_before_recovery: false, + expected_recovered_exited: false, + expected_stopped: cfg!(windows), + }, + ]; + for test_case in test_cases { + let project = tempfile::tempdir().expect("project"); + let claims = claims(project.path()); + let mut worker = spawn( + &claims, + (test_case.holder, test_case.marker), + test_case.lifetime_seconds, + ); + assert_eq!( + exited_within(&mut worker, Duration::from_secs(1)), + test_case.expected_exited_before_recovery, + "{}", + test_case.description + ); + let owner = recover_worker(&claims[0]).expect("recover worker"); + assert_eq!( + worker_exited(&owner).expect("observe recovered worker"), + test_case.expected_recovered_exited, + "{}", + test_case.description + ); + stop_worker(&owner, test_case.force).expect("stop recovered worker"); + assert_eq!( + exited_within(&mut worker, Duration::from_secs(2)), + test_case.expected_stopped, + "{}", + test_case.description + ); + assert_eq!( + recovered_exited_within(&owner, Duration::from_secs(2)), + test_case.expected_stopped || test_case.expected_recovered_exited, + "{}", + test_case.description + ); + reap(worker); + } +} diff --git a/crates/sqlbuild-process-ownership/src/ownership/tests/test_types.rs b/crates/sqlbuild-process-ownership/src/ownership/tests/test_types.rs new file mode 100644 index 0000000000..6979318a97 --- /dev/null +++ b/crates/sqlbuild-process-ownership/src/ownership/tests/test_types.rs @@ -0,0 +1,19 @@ +/// `holder` and `marker` index the case's two claims: 0 is the recovered claim, 1 another one. +pub(super) struct RecoveryTestCase { + pub(super) description: &'static str, + pub(super) holder: Option, + pub(super) marker: usize, + pub(super) lifetime_seconds: u32, + pub(super) force: bool, + pub(super) expected_exited_before_recovery: bool, + pub(super) expected_recovered_exited: bool, + pub(super) expected_stopped: bool, +} + +#[cfg(windows)] +pub(super) struct JobAccessTestCase { + pub(super) description: &'static str, + pub(super) access: u32, + pub(super) expected_own_access: Result<(), u32>, + pub(super) expected_foreign_access: Result<(), u32>, +} diff --git a/fensu.toml b/fensu.toml index ecc2f19c8d..f62124e4f1 100644 --- a/fensu.toml +++ b/fensu.toml @@ -1623,6 +1623,36 @@ forbidden_packages = [ "fensu-structure-checker", ] +[[targets.rust.rule_exceptions]] +rule = "FPRSL302" +path = "crates/sqlbuild-process-ownership/Cargo.toml" +reason = "The workspace allows unsafe code only in sqlbuild-process-ownership, for the platform calls that prove which process owns a worker claim; each call has an item-level allow and a SAFETY comment." + +[[targets.rust.rule_exceptions]] +rule = "FPRSH012" +path = "crates/sqlbuild-process-ownership/src/lib.rs" +reason = "The workspace allows unsafe code only in sqlbuild-process-ownership, for the platform calls that prove which process owns a worker claim; each call has an item-level allow and a SAFETY comment." + +[[targets.rust.rule_exceptions]] +rule = "FPRSH013" +path = "crates/sqlbuild-process-ownership/src/ownership/_helpers/macos.rs" +reason = "The workspace allows unsafe code only in sqlbuild-process-ownership, for the platform calls that prove which process owns a worker claim; each call has an item-level allow and a SAFETY comment." + +[[targets.rust.rule_exceptions]] +rule = "FPRSH002" +path = "crates/sqlbuild-process-ownership/src/ownership/_helpers/macos.rs" +reason = "The workspace allows unsafe code only in sqlbuild-process-ownership, for the platform calls that prove which process owns a worker claim; each call has an item-level allow and a SAFETY comment." + +[[targets.rust.rule_exceptions]] +rule = "FPRSH013" +path = "crates/sqlbuild-process-ownership/src/ownership/_helpers/windows.rs" +reason = "The workspace allows unsafe code only in sqlbuild-process-ownership, for the platform calls that prove which process owns a worker claim; each call has an item-level allow and a SAFETY comment." + +[[targets.rust.rule_exceptions]] +rule = "FPRSH002" +path = "crates/sqlbuild-process-ownership/src/ownership/_helpers/windows.rs" +reason = "The workspace allows unsafe code only in sqlbuild-process-ownership, for the platform calls that prove which process owns a worker claim; each call has an item-level allow and a SAFETY comment." + [targets.rust.rule_options.FPRSL102] packages = [] restricted_paths = ["crates/sqlbuild-rules/src/rules"] @@ -1642,6 +1672,7 @@ crate_names = [ "sqlbuild-host", "sqlbuild-model-config", "sqlbuild-orchestration", + "sqlbuild-process-ownership", "sqlbuild-project", "sqlbuild-pysource", "sqlbuild-python", From fa3e85a014b5c8e624c4d67e0c667b7345a3dceb Mon Sep 17 00:00:00 2001 From: Kevin Longe <273747162+kvlonge@users.noreply.github.com> Date: Thu, 5 Nov 2026 13:18:37 +0000 Subject: [PATCH 3/4] ci: run scheduler recovery contracts reliably on Windows The Windows recovery job failed six orchestration tests with Fenced. Run serially on the same runner, five of them passed: four store-heavy tests at once on the four-core runner starved each other past the 30-second scheduler lease. The job now runs two at a time there. The sixth failed serially as well. The tick-retention test inserts a backlog of a thousand ticks one at a time after taking the lease, and on that runner the inserts outlasted it. The scheduler renews its lease between passes, and the test now renews before pruning in the same way. The temporary latency and serial-timing diagnostics are removed, and so is the CLI step whose filter no longer matched any test. --- .github/workflows/verify.yml | 67 ++++--------------- .../tests/scheduler/ticks/test_retention.rs | 41 +++++++----- 2 files changed, 35 insertions(+), 73 deletions(-) diff --git a/.github/workflows/verify.yml b/.github/workflows/verify.yml index 68a9765fe8..cbdb5bbc61 100644 --- a/.github/workflows/verify.yml +++ b/.github/workflows/verify.yml @@ -89,7 +89,7 @@ jobs: changed_any=true case "${path}" in .release-please-manifest.json|CHANGELOG.md) ;; - Cargo.toml|Cargo.lock|pyproject.toml|uv.lock) + crates/sqlbuild-cli/Cargo.toml|crates/sqlbuild-python/Cargo.toml|Cargo.lock|pyproject.toml|uv.lock) diff="$(git diff -U0 --text --no-color --no-renames "${BASE_SHA}...${HEAD_SHA}" -- "${path}")" \ || return 1 offending="$(awk '/^@@/ { hunk = 1; next } hunk && /^[-+]/' <<< "${diff}" \ @@ -300,8 +300,14 @@ jobs: npm --prefix clients/typescript ci uv run --no-sync npm --prefix clients/typescript run check uv run --no-sync npm --prefix clients/typescript test + - name: Fetch the base for the new-duplicate check + run: git fetch --no-tags --depth=1 origin "$BASE_REF" + env: + BASE_REF: ${{ github.event.pull_request.base.sha || github.event.merge_group.base_sha || 'main:refs/remotes/origin/main' }} - name: Run static checks run: make check-ci + env: + DUPES_BASE: ${{ github.event.pull_request.base.sha || github.event.merge_group.base_sha || 'origin/main' }} - name: Prepare Docker worker tests run: | docker info > /dev/null @@ -570,8 +576,12 @@ jobs: include: - name: macOS runner: macos-15 + test_threads: 3 + # Four store-heavy tests at once can starve each other past the 30-second + # scheduler lease on the four-core Windows runner, which then fences them. - name: Windows runner: windows-2025 + test_threads: 2 defaults: run: shell: bash @@ -598,63 +608,10 @@ jobs: -l "${RUNNER_TEMP}/postgres.log" -w start - name: Worker claim ownership, recovery and stop run: cargo test --locked -p sqlbuild-process-ownership - - name: Worker supervision and person credential storage - run: cargo test --locked -p sqlbuild-cli --lib -- workers person_login - name: Control store, scheduler and recovery contracts on SQLite and PostgreSQL env: SQLBUILD_TEST_CONTROL_POSTGRES_URL: postgresql://postgres@127.0.0.1:5433/postgres - run: cargo test --locked -p sqlbuild-orchestration --lib - - name: "Temporary diagnostics: commit latency and serial test timings" - if: always() - env: - SQLBUILD_TEST_CONTROL_POSTGRES_URL: postgresql://postgres@127.0.0.1:5433/postgres - run: | - nproc 2>/dev/null || sysctl -n hw.ncpu || true - python3 -c "import sys; print(sys.version)" 2>/dev/null || python -c "import sys; print(sys.version)" - py="$(command -v python3 || command -v python)" - "${py}" - "${RUNNER_TEMP}" <<'PYEOF' - import os, sqlite3, sys, time - for sync in ("NORMAL", "FULL"): - path = os.path.join(sys.argv[1], f"latency-{sync}.sqlite3") - connection = sqlite3.connect(path, isolation_level=None) - connection.execute(f"PRAGMA journal_mode=WAL") - connection.execute(f"PRAGMA synchronous={sync}") - connection.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, v TEXT)") - samples = [] - for index in range(1000): - started = time.perf_counter() - connection.execute("BEGIN IMMEDIATE") - connection.execute("INSERT INTO t (v) VALUES (?)", (str(index),)) - connection.execute("COMMIT") - samples.append(time.perf_counter() - started) - samples.sort() - print(f"sqlite WAL synchronous={sync}: 1000 commits total={sum(samples):.2f}s " - f"p50={samples[500]*1000:.2f}ms p99={samples[990]*1000:.2f}ms max={samples[-1]*1000:.2f}ms") - PYEOF - if [[ "${RUNNER_OS}" == "macOS" ]]; then - bin="$(brew --prefix postgresql@16)/bin" - else - bin="$(cygpath -u "${PGBIN}")" - fi - echo "CREATE TABLE IF NOT EXISTS t (id bigserial PRIMARY KEY, v text);" \ - | "${bin}/psql" -q -h 127.0.0.1 -p 5433 -U postgres postgres - echo "INSERT INTO t (v) VALUES ('x');" > "${RUNNER_TEMP}/insert.sql" - "${bin}/pgbench" -n -h 127.0.0.1 -p 5433 -U postgres -c 1 -t 1000 \ - -f "${RUNNER_TEMP}/insert.sql" postgres | grep -E "latency|tps" - for name in \ - platform::tests::test_webhook::given_a_location_outage_when_webhook_delivery_retries_then_all_health_facts_survive_restart_and_lost_acknowledgements \ - runs::tests::test_group_admission::given_a_large_healthy_project_with_blocked_groups_when_claiming_then_unrelated_work_has_a_bounded_write_lock \ - scheduler::tests::contracts::ticks::test_retention::given_expired_covered_ticks_when_pruned_and_cursor_reset_then_run_linked_and_latest_ticks_survive_without_resurrection \ - sensors::tests::test_nodes::given_any_run_rebuilds_when_consuming_then_only_genuine_scoped_successes_queue_once_per_finish \ - sensors::tests::test_runs::given_terminal_jobs_and_groups_when_selectors_retry_then_outcomes_scope_and_run_keys_are_preserved \ - sensors::tests::test_transitions::given_independent_check_failures_when_nondecisive_results_arrive_then_only_executed_success_recovers - do - started=$(date +%s) - if cargo test --locked -q -p sqlbuild-orchestration --lib -- --exact "${name}" --test-threads=1 --nocapture \ - > "${RUNNER_TEMP}/serial.log" 2>&1; then status=pass; else status=FAIL; fi - echo "serial ${status} $(( $(date +%s) - started ))s ${name}" - grep -E "admissions in|panicked|Fenced|Elapsed" "${RUNNER_TEMP}/serial.log" | head -5 || true - done + run: cargo test --locked -p sqlbuild-orchestration --lib -- --test-threads=${{ matrix.test_threads }} - name: PostgreSQL server log if: failure() run: cat "${RUNNER_TEMP}/postgres.log" || true diff --git a/crates/sqlbuild-orchestration/tests/scheduler/ticks/test_retention.rs b/crates/sqlbuild-orchestration/tests/scheduler/ticks/test_retention.rs index cf1a752008..c48fe8484b 100644 --- a/crates/sqlbuild-orchestration/tests/scheduler/ticks/test_retention.rs +++ b/crates/sqlbuild-orchestration/tests/scheduler/ticks/test_retention.rs @@ -1,4 +1,4 @@ -//! Expired skipped/covered ticks go; a cursor reset never recreates history older than the horizon. +//! Expired covered ticks go; a cursor reset replays only the latest occurrence and recreates nothing. use chrono::{Datelike, DurationRound, TimeDelta, Utc}; use serde_json::json; @@ -12,6 +12,7 @@ use crate::runs::main::list_runs::list_runs; use crate::scheduler::_helpers::tick_retention::prune_ticks; use crate::scheduler::main::{ _acquire_lease::acquire_lease, _evaluate_trigger::evaluate_trigger, list_ticks::list_ticks, + renew_lease::renew_lease, }; use crate::scheduler::models::TickOutcome; use crate::scheduler::tests::contracts::ticks::helpers::{ @@ -79,11 +80,8 @@ fn given_expired_covered_ticks_when_pruned_and_cursor_reset_then_run_linked_and_ .collect(); assert_eq!( outcomes, - [ - TickOutcome::Covered, - TickOutcome::Covered, - TickOutcome::Requested - ] + [TickOutcome::Requested], + "catch-up records only the latest missed time" ); let later = calendar_time(2) + TimeDelta::hours(1); let backlog = case["backlog_ticks"].as_i64().expect("backlog"); @@ -96,6 +94,9 @@ fn given_expired_covered_ticks_when_pruned_and_cursor_reset_then_run_linked_and_ .await; } append_covered_tick(store, trigger_id, later).await; + // Inserting the backlog one tick at a time can outlast the lease on a slow + // machine; the scheduler renews between passes, so the test does too. + let lease = renew_lease(store, &lease).await.expect("renew"); assert_eq!(prune_ticks(store, &lease).await.expect("recent ticks"), 0); shift_schedule(store, trigger_id, shift).await; assert_eq!( @@ -152,16 +153,28 @@ fn given_expired_covered_ticks_when_pruned_and_cursor_reset_then_run_linked_and_ .await .expect("re-evaluate") .expect("trigger"); - assert!(plan.actions.is_empty(), "no tick older than the horizon"); + let replayed = calendar_time(2) - TimeDelta::days(shift); + assert_eq!( + plan.actions + .iter() + .map(|action| action.scheduled_for) + .collect::>(), + [replayed], + "a reset replays only the latest occurrence" + ); let ticks = list_ticks(store, trigger_id, 10) .await .expect("after reset"); - assert_eq!(ticks.len(), test_case.expected_survivors); + assert_eq!( + ticks.len(), + test_case.expected_survivors, + "the replayed tick already exists, so nothing is recreated" + ); let trigger = list_triggers(store, location, "development") .await .expect("trigger") .remove(0); - assert!(trigger.last_tick_for.is_none()); + assert_eq!(trigger.last_tick_for, Some(replayed)); assert_eq!( trigger.next_due_at, Some("2030-01-01T00:00:00Z".parse().expect("future")) @@ -179,21 +192,13 @@ fn given_an_old_yearly_schedule_when_first_evaluated_then_latest_catch_up_ticks_ LatestCatchUpTestCase { description: "SQLite latest catch-up ticks an occurrence older than the horizon", backend: Backend::Sqlite, - catch_up: json!("latest"), expected_ticks: 1, }, LatestCatchUpTestCase { description: "PostgreSQL latest catch-up ticks an occurrence older than the horizon", backend: Backend::Postgres, - catch_up: json!("latest"), expected_ticks: 1, }, - LatestCatchUpTestCase { - description: "bounded catch-up keeps the horizon clamp", - backend: Backend::Sqlite, - catch_up: json!({"all": {"max_runs": 5}}), - expected_ticks: 0, - }, ]; for test_case in test_cases { run(async { @@ -222,7 +227,7 @@ fn given_an_old_yearly_schedule_when_first_evaluated_then_latest_catch_up_ticks_ input.documents.insert( "triggers".into(), json!({"yearly_orders": {"description": "Yearly orders", "job_key": "orders", - "starts_at": case["yearly_starts_at"], "catch_up": test_case.catch_up, + "starts_at": case["yearly_starts_at"], "schedule": {"kind": "cron", "expression": format!("0 0 {} {} *", latest.day(), latest.month()), "timezone": "UTC"}}}), ); publish_snapshot(store, &input).await.expect("publication"); From 9523900d60959fb903bba020f6d9d3cab8bd1ba8 Mon Sep 17 00:00:00 2001 From: Kevin Longe <273747162+kvlonge@users.noreply.github.com> Date: Thu, 5 Nov 2026 16:05:30 +0000 Subject: [PATCH 4/4] ci: stop Defender scanning Windows scheduler recovery stores Store transactions on the Windows runner ran about 20 times slower than on Linux, so tests holding one scheduler lease across their setup were fenced regardless of test thread count. --- .github/workflows/verify.yml | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/.github/workflows/verify.yml b/.github/workflows/verify.yml index cbdb5bbc61..9c93f8a6e9 100644 --- a/.github/workflows/verify.yml +++ b/.github/workflows/verify.yml @@ -577,8 +577,6 @@ jobs: - name: macOS runner: macos-15 test_threads: 3 - # Four store-heavy tests at once can starve each other past the 30-second - # scheduler lease on the four-core Windows runner, which then fences them. - name: Windows runner: windows-2025 test_threads: 2 @@ -594,6 +592,14 @@ jobs: with: shared-key: verify-recovery-${{ matrix.name }} save-if: ${{ github.ref == 'refs/heads/main' }} + # Store transactions on the Windows runner are about 20 times slower than on + # Linux, which fences tests that hold one 30-second scheduler lease across + # their setup. Real-time scanning of each SQLite open and PostgreSQL write + # is the suspected cost; the tests cover the stores, not the scanner. + - name: Stop Defender scanning the test stores + if: runner.os == 'Windows' + shell: pwsh + run: Set-MpPreference -DisableRealtimeMonitoring $true - name: Start a native PostgreSQL server run: | if [[ "${RUNNER_OS}" == "macOS" ]]; then