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/2] 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 cd908d1d9..a74b538f4 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/2] 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 a74b538f4..68a9765fe 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 462a1de8f..959dcdaa0 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 adb985a36..f81489ca9 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 7f3820572..522cf3fbf 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 1f24eb892..25fb89dd2 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 c8c2643d4..52fcb1016 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 82a1c4fba..000000000 --- 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 a0ead3308..b182827d2 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 f439a58dd..fb08f51c1 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 7c7dd5635..9411afe73 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 06a889dbf..881c48bea 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 35639fe18..e6eb0e5ab 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 000000000..ffd57c826 --- /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 000000000..7facf2034 --- /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 000000000..04ba18619 --- /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 000000000..d2bdadbba --- /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 000000000..1615c7506 --- /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 000000000..c4132976a --- /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 000000000..b65e4319a --- /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 000000000..b56fbfd5c --- /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 000000000..211f45258 --- /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 000000000..eca8b366e --- /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 000000000..ad5637bf5 --- /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 000000000..e74973061 --- /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 000000000..40c81d6d6 --- /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 000000000..911b6fc15 --- /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 000000000..691cfdff9 --- /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 000000000..3fd82c8a7 --- /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 000000000..167d752f7 --- /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 000000000..382881ac2 --- /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 000000000..5eccb3e14 --- /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 000000000..8ba089838 --- /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 000000000..6979318a9 --- /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 ecc2f19c8..f62124e4f 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",