From 47295776219fc72e1c9d7d3d58c40ede19782670 Mon Sep 17 00:00:00 2001 From: Zachery Aaron Shores-Chmielewski Date: Fri, 28 Aug 2026 02:09:51 +0400 Subject: [PATCH] chore(test): enforce audited timeout policy Route cargo lint through repository timeout validation and strict Clippy, reject unregistered timing primitives with compiler-resolved call budgets, bound Nextest concurrency, and replace polling waits in test infrastructure. --- .cargo/config.toml | 2 +- .config/nextest.toml | 12 +- tools/actor-control-flow-lint/driver.rs | 136 ++-- .../actor-control-flow-lint/rustc-wrapper.py | 5 +- .../tests/contracts.rs | 39 +- .../fixtures/fail-owner-timing/Cargo.lock | 25 + .../fixtures/fail-owner-timing/Cargo.toml | 10 + .../fixtures/fail-owner-timing/src/lib.rs | 6 + .../fixtures/fail-timing-overage/Cargo.lock | 7 + .../fixtures/fail-timing-overage/Cargo.toml | 7 + .../fixtures/fail-timing-overage/src/lib.rs | 7 + .../fixtures/pass-execution-owner/src/lib.rs | 3 - .../fixtures/pass-test-wait/tests/wait.rs | 6 +- .../actor-control-flow-lint/timing_policy.rs | 602 ++++++++++++++++++ tools/timeout-policy.py | 198 ++++++ tools/vastai/src/logs.rs | 3 - tools/vastai/src/test_http.rs | 15 +- xtask/src/main.rs | 32 +- 18 files changed, 1041 insertions(+), 74 deletions(-) create mode 100644 tools/actor-control-flow-lint/tests/fixtures/fail-owner-timing/Cargo.lock create mode 100644 tools/actor-control-flow-lint/tests/fixtures/fail-owner-timing/Cargo.toml create mode 100644 tools/actor-control-flow-lint/tests/fixtures/fail-owner-timing/src/lib.rs create mode 100644 tools/actor-control-flow-lint/tests/fixtures/fail-timing-overage/Cargo.lock create mode 100644 tools/actor-control-flow-lint/tests/fixtures/fail-timing-overage/Cargo.toml create mode 100644 tools/actor-control-flow-lint/tests/fixtures/fail-timing-overage/src/lib.rs create mode 100644 tools/actor-control-flow-lint/timing_policy.rs create mode 100755 tools/timeout-policy.py diff --git a/.cargo/config.toml b/.cargo/config.toml index 304616c..af5dc4e 100644 --- a/.cargo/config.toml +++ b/.cargo/config.toml @@ -3,4 +3,4 @@ rustc-workspace-wrapper = "tools/actor-control-flow-lint/rustc-wrapper.py" [alias] xtask = "run --package xtask --" -lint = "clippy --workspace --all-targets --all-features" +lint = "run --package xtask -- lint" diff --git a/.config/nextest.toml b/.config/nextest.toml index f6f2b4e..c0b2ffe 100644 --- a/.config/nextest.toml +++ b/.config/nextest.toml @@ -1,2 +1,12 @@ +# Workspace-wide test-runner configuration for cargo-nextest. +# +# The suite mixes deadline-sensitive process and network tests (real children, +# real QUIC engines) with compute-heavy property tests. At full 12-thread +# parallelism OS scheduling delays blow through those deadlines — different +# timing tests fail on every loaded run — and the namespace reference-model +# guarantee test (~25s unloaded) stretches past the default 60s slow-timeout. +# Bounded threads plus a 180s slow period trades wall time for determinism +# without masking true hangs. [profile.default] -slow-timeout = { period = "60s", terminate-after = 1 } +test-threads = 4 +slow-timeout = { period = "180s" } diff --git a/tools/actor-control-flow-lint/driver.rs b/tools/actor-control-flow-lint/driver.rs index 1f5d9db..30b48d6 100755 --- a/tools/actor-control-flow-lint/driver.rs +++ b/tools/actor-control-flow-lint/driver.rs @@ -26,23 +26,22 @@ const EXECUTION_OWNERS: &[&str] = &[ "swactor", "swactor-engine", "swactor-process", + "swactor-process-context", "swactor-transport", "swactor-vastai", "telemetry", ]; /// Test and benchmark harness packages that must not be workspace dependencies. -const TEST_SUPPORT_OWNERS: &[&str] = - &["actor-control-flow-lint-tests", "swactor-benchmarks"]; +const TEST_SUPPORT_OWNERS: &[&str] = &[ + "actor-control-flow-lint-tests", + "myelin-e2e-fuzz", + "swactor-benchmarks", +]; /// Policy-bearing crates that must never enter an execution owner's dependency /// closure. -const DOMAIN_CONTROL_CRATES: &[&str] = &[ - "myelin", - "provisioning", - "swactor-job-runner", - "xtask", -]; +const DOMAIN_CONTROL_CRATES: &[&str] = &["myelin", "provisioning", "xtask"]; #[derive(Clone, Copy)] struct Capability { @@ -54,8 +53,7 @@ struct Capability { const MOVE_TO_OWNER: &str = "move stream mechanics into an approved execution owner or move the decision into an actor"; -const USE_ACTOR_TIMER: &str = - "schedule a typed actor message through the engine; the receiving actor owns the deadline decision"; +const USE_ACTOR_TIMER: &str = "schedule a typed actor message through the engine; the receiving actor owns the deadline decision"; /// Stable resolved item paths. These are deliberately compiler identities, not /// spellings found in source, so re-exports, renamed imports, and local wrappers @@ -179,6 +177,8 @@ const CAPABILITIES: &[Capability] = &[ }, ]; +include!("timing_policy.rs"); + struct ActorControlFlowCallbacks { package: String, test_build: bool, @@ -186,17 +186,11 @@ struct ActorControlFlowCallbacks { } impl Callbacks for ActorControlFlowCallbacks { - fn after_analysis<'tcx>( - &mut self, - _compiler: &Compiler, - tcx: TyCtxt<'tcx>, - ) -> Compilation { - if TEST_SUPPORT_OWNERS.contains(&self.package.as_str()) { - return Compilation::Continue; - } + fn after_analysis<'tcx>(&mut self, _compiler: &Compiler, tcx: TyCtxt<'tcx>) -> Compilation { + let role_exempt = TEST_SUPPORT_OWNERS.contains(&self.package.as_str()) + || EXECUTION_OWNERS.contains(&self.package.as_str()); if EXECUTION_OWNERS.contains(&self.package.as_str()) { check_owner_dependencies(tcx, &self.package); - return Compilation::Continue; } for owner in tcx.hir_body_owners() { @@ -206,8 +200,11 @@ impl Callbacks for ActorControlFlowCallbacks { tcx, typeck, package: &self.package, + caller: tcx.def_path_str(owner.to_def_id()), + role_exempt, test_build: self.test_build, trace: self.trace, + timing_counts: Vec::new(), }; visitor.visit_body(body); } @@ -232,8 +229,11 @@ struct CapabilityVisitor<'a, 'tcx> { tcx: TyCtxt<'tcx>, typeck: &'tcx TypeckResults<'tcx>, package: &'a str, + caller: String, + role_exempt: bool, test_build: bool, trace: bool, + timing_counts: Vec<(String, usize)>, } impl<'tcx> Visitor<'tcx> for CapabilityVisitor<'_, 'tcx> { @@ -256,11 +256,46 @@ impl<'tcx> Visitor<'tcx> for CapabilityVisitor<'_, 'tcx> { } impl CapabilityVisitor<'_, '_> { - fn check(&self, def_id: rustc_hir::def_id::DefId, span: Span) { + fn check(&mut self, def_id: rustc_hir::def_id::DefId, span: Span) { let path = self.tcx.def_path_str(def_id); let normalized_path = normalize_def_path(&path); - if self.trace && is_candidate_name(self.tcx.item_name(def_id)) { - eprintln!("actor-control-flow trace: {path}"); + let item_name = self.tcx.item_name(def_id); + if self.trace && is_candidate_name(item_name) { + eprintln!( + "actor-control-flow trace: package=`{}` caller=`{}` callee=`{path}`", + self.package, self.caller + ); + } + + if GUARDED_TIMING_ITEMS.contains(&item_name.as_str()) { + let occurrence = self.next_timing_occurrence(&normalized_path); + let allowance = TIMING_ALLOWANCES.iter().find(|allowance| { + allowance.package == self.package + && allowance.caller == self.caller + && allowance.callee == normalized_path + }); + match allowance { + Some(allowance) if occurrence <= allowance.calls => return, + Some(allowance) => { + self.tcx.dcx().span_err( + span, + format!( + "unapproved timing primitive: `{normalized_path}` occurrence {occurrence} exceeds the {} audited call(s) in `{}` (`{}`); remove the timer or update the central timing registry after architectural review", + allowance.calls, self.caller, allowance.purpose + ), + ); + } + None => { + self.tcx.dcx().span_err( + span, + format!( + "unapproved timing primitive: `{normalized_path}` is forbidden in workspace package `{}` at `{}`; use typed engine scheduling or register an exact external-resource boundary", + self.package, self.caller + ), + ); + } + } + return; } let Some(capability) = CAPABILITIES @@ -270,8 +305,7 @@ impl CapabilityVisitor<'_, '_> { return; }; - - if self.test_build && capability.test_wait { + if self.role_exempt || (self.test_build && capability.test_wait) { return; } @@ -283,6 +317,19 @@ impl CapabilityVisitor<'_, '_> { ), ); } + + fn next_timing_occurrence(&mut self, callee: &str) -> usize { + if let Some((_, count)) = self + .timing_counts + .iter_mut() + .find(|(known, _)| known == callee) + { + *count += 1; + return *count; + } + self.timing_counts.push((callee.to_owned(), 1)); + 1 + } } fn normalize_def_path(path: &str) -> String { @@ -314,29 +361,24 @@ fn normalize_def_path(path: &str) -> String { } fn is_candidate_name(name: Symbol) -> bool { - matches!( - name.as_str(), - "block_on" - | "blocking_recv" - | "has_work" - | "interval" - | "interval_at" - | "new_current_thread" - | "new_multi_thread" - | "recv" - | "recv_deadline" - | "recv_timeout" - | "sleep" - | "sleep_until" - | "spawn" - | "spawn_blocking" - | "spawn_local" - | "tick" - | "timeout" - | "timeout_at" - | "timer" - | "try_tick" - ) + GUARDED_TIMING_ITEMS.contains(&name.as_str()) + || matches!( + name.as_str(), + "block_on" + | "blocking_recv" + | "has_work" + | "interval" + | "interval_at" + | "new_current_thread" + | "new_multi_thread" + | "recv" + | "spawn" + | "spawn_blocking" + | "spawn_local" + | "tick" + | "timer" + | "try_tick" + ) } fn main() -> ExitCode { diff --git a/tools/actor-control-flow-lint/rustc-wrapper.py b/tools/actor-control-flow-lint/rustc-wrapper.py index e7e679e..c45a5ee 100755 --- a/tools/actor-control-flow-lint/rustc-wrapper.py +++ b/tools/actor-control-flow-lint/rustc-wrapper.py @@ -24,6 +24,7 @@ def main() -> None: rustc_args = sys.argv[2:] root = Path(__file__).resolve().parents[2] source = Path(__file__).with_name("driver.rs") + policy = Path(__file__).with_name("timing_policy.rs") target_root = Path(os.environ.get("CARGO_TARGET_DIR", root / "target")) if not target_root.is_absolute(): target_root = root / target_root @@ -50,7 +51,9 @@ def main() -> None: except (OSError, subprocess.CalledProcessError) as error: fail(f"cannot inspect pinned rustc: {error}") - digest = hashlib.sha256(source.read_bytes() + version.encode()).hexdigest()[:20] + digest = hashlib.sha256( + source.read_bytes() + b"\0" + policy.read_bytes() + b"\0" + version.encode() + ).hexdigest()[:20] driver = cache / f"driver-{digest}" lock_path = cache / "build.lock" diff --git a/tools/actor-control-flow-lint/tests/contracts.rs b/tools/actor-control-flow-lint/tests/contracts.rs index cb9ae19..cc3a3e2 100644 --- a/tools/actor-control-flow-lint/tests/contracts.rs +++ b/tools/actor-control-flow-lint/tests/contracts.rs @@ -9,6 +9,18 @@ fn repository_root() -> PathBuf { .to_path_buf() } +fn driver_revision(root: &Path) -> String { + let mut hash = 0xcbf29ce484222325_u64; + for path in ["driver.rs", "timing_policy.rs"] { + let source = std::fs::read(root.join("tools/actor-control-flow-lint").join(path)) + .expect("read compiler policy source"); + for byte in source { + hash = (hash ^ u64::from(byte)).wrapping_mul(0x100000001b3); + } + } + format!("--cfg=actor_lint_driver_{hash:016x}") +} + fn cargo_check(fixture: &str, extra_args: &[&str]) -> Output { let root = repository_root(); let fixture_dir = Path::new(env!("CARGO_MANIFEST_DIR")) @@ -28,6 +40,7 @@ fn cargo_check(fixture: &str, extra_args: &[&str]) -> Output { .env("CARGO_TARGET_DIR", target_dir) .env("CARGO_TERM_COLOR", "never") .env("RUSTC_WORKSPACE_WRAPPER", wrapper) + .env("RUSTFLAGS", driver_revision(&root)) .env_remove("CARGO_MAKEFLAGS") .env_remove("MAKEFLAGS"); command.output().expect("run fixture cargo check") @@ -50,7 +63,7 @@ fn compiler_policy_contracts() { "blocking task spawning", "engine task scheduling", "OS thread creation", - "thread sleeping", + "unapproved timing primitive", "direct timer driving", "runtime construction or driving", "blocking receive used as a controller", @@ -74,6 +87,30 @@ fn compiler_policy_contracts() { "missing owner dependency diagnostic:\n{dependency_stderr}" ); + let owner_timing = cargo_check("fail-owner-timing", &[]); + assert!( + !owner_timing.status.success(), + "unapproved execution-owner timer compiled" + ); + let owner_timing_stderr = stderr(&owner_timing); + assert!( + owner_timing_stderr.contains( + "unapproved timing primitive: `tokio::time::sleep` is forbidden in workspace package `iroh-driver`" + ), + "missing resolved owner-timer diagnostic:\n{owner_timing_stderr}" + ); + + let timing_overage = cargo_check("fail-timing-overage", &[]); + assert!( + !timing_overage.status.success(), + "timing allowance overage compiled" + ); + let timing_overage_stderr = stderr(&timing_overage); + assert!( + timing_overage_stderr.contains("occurrence 2 exceeds the 1 audited call(s)"), + "missing timing-overage diagnostic:\n{timing_overage_stderr}" + ); + for fixture in ["pass-actor-domain", "pass-execution-owner"] { let output = cargo_check(fixture, &[]); assert!( diff --git a/tools/actor-control-flow-lint/tests/fixtures/fail-owner-timing/Cargo.lock b/tools/actor-control-flow-lint/tests/fixtures/fail-owner-timing/Cargo.lock new file mode 100644 index 0000000..8d47493 --- /dev/null +++ b/tools/actor-control-flow-lint/tests/fixtures/fail-owner-timing/Cargo.lock @@ -0,0 +1,25 @@ +# This file is automatically @generated by Cargo. +# It is not intended for manual editing. +version = 4 + +[[package]] +name = "iroh-driver" +version = "0.0.0" +dependencies = [ + "tokio", +] + +[[package]] +name = "pin-project-lite" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd" + +[[package]] +name = "tokio" +version = "1.53.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "202caea871b69668250d242070849eb495be178ed697a3e98aebce5bc81a0bed" +dependencies = [ + "pin-project-lite", +] diff --git a/tools/actor-control-flow-lint/tests/fixtures/fail-owner-timing/Cargo.toml b/tools/actor-control-flow-lint/tests/fixtures/fail-owner-timing/Cargo.toml new file mode 100644 index 0000000..8763273 --- /dev/null +++ b/tools/actor-control-flow-lint/tests/fixtures/fail-owner-timing/Cargo.toml @@ -0,0 +1,10 @@ +[package] +name = "iroh-driver" +version = "0.0.0" +edition = "2024" +publish = false + +[dependencies] +tokio = { version = "1", features = ["time"] } + +[workspace] diff --git a/tools/actor-control-flow-lint/tests/fixtures/fail-owner-timing/src/lib.rs b/tools/actor-control-flow-lint/tests/fixtures/fail-owner-timing/src/lib.rs new file mode 100644 index 0000000..aa25157 --- /dev/null +++ b/tools/actor-control-flow-lint/tests/fixtures/fail-owner-timing/src/lib.rs @@ -0,0 +1,6 @@ +use std::time::Duration; +use tokio::time::sleep as pause; + +pub async fn unapproved_owner_wait() { + pause(Duration::from_millis(1)).await; +} diff --git a/tools/actor-control-flow-lint/tests/fixtures/fail-timing-overage/Cargo.lock b/tools/actor-control-flow-lint/tests/fixtures/fail-timing-overage/Cargo.lock new file mode 100644 index 0000000..dc864a3 --- /dev/null +++ b/tools/actor-control-flow-lint/tests/fixtures/fail-timing-overage/Cargo.lock @@ -0,0 +1,7 @@ +# This file is automatically @generated by Cargo. +# It is not intended for manual editing. +version = 4 + +[[package]] +name = "actor-lint-timing-overage" +version = "0.0.0" diff --git a/tools/actor-control-flow-lint/tests/fixtures/fail-timing-overage/Cargo.toml b/tools/actor-control-flow-lint/tests/fixtures/fail-timing-overage/Cargo.toml new file mode 100644 index 0000000..a08ebdf --- /dev/null +++ b/tools/actor-control-flow-lint/tests/fixtures/fail-timing-overage/Cargo.toml @@ -0,0 +1,7 @@ +[package] +name = "actor-lint-timing-overage" +version = "0.0.0" +edition = "2024" +publish = false + +[workspace] diff --git a/tools/actor-control-flow-lint/tests/fixtures/fail-timing-overage/src/lib.rs b/tools/actor-control-flow-lint/tests/fixtures/fail-timing-overage/src/lib.rs new file mode 100644 index 0000000..7791015 --- /dev/null +++ b/tools/actor-control-flow-lint/tests/fixtures/fail-timing-overage/src/lib.rs @@ -0,0 +1,7 @@ +use std::time::Duration; + +pub fn exceeds_registered_timing_budget() { + let (_sender, receiver) = std::sync::mpsc::channel::<()>(); + let _ = receiver.recv_timeout(Duration::from_millis(1)); + let _ = receiver.recv_timeout(Duration::from_millis(1)); +} diff --git a/tools/actor-control-flow-lint/tests/fixtures/pass-execution-owner/src/lib.rs b/tools/actor-control-flow-lint/tests/fixtures/pass-execution-owner/src/lib.rs index 7503fe8..138e091 100644 --- a/tools/actor-control-flow-lint/tests/fixtures/pass-execution-owner/src/lib.rs +++ b/tools/actor-control-flow-lint/tests/fixtures/pass-execution-owner/src/lib.rs @@ -1,8 +1,5 @@ -use std::time::Duration; - pub fn start_transport_pump() { tokio::spawn(async { let _io_type: Option = None; - tokio::time::sleep(Duration::from_millis(1)).await; }); } diff --git a/tools/actor-control-flow-lint/tests/fixtures/pass-test-wait/tests/wait.rs b/tools/actor-control-flow-lint/tests/fixtures/pass-test-wait/tests/wait.rs index 6b42541..07a273f 100644 --- a/tools/actor-control-flow-lint/tests/fixtures/pass-test-wait/tests/wait.rs +++ b/tools/actor-control-flow-lint/tests/fixtures/pass-test-wait/tests/wait.rs @@ -4,9 +4,5 @@ use std::time::Duration; fn bounded_observation_wait_is_allowed() { let (sender, receiver) = std::sync::mpsc::channel(); sender.send(7_u8).expect("send observation"); - assert_eq!( - receiver.recv_timeout(Duration::from_millis(10)), - Ok(7) - ); - std::thread::sleep(Duration::from_millis(1)); + assert_eq!(receiver.recv_timeout(Duration::from_millis(10)), Ok(7)); } diff --git a/tools/actor-control-flow-lint/timing_policy.rs b/tools/actor-control-flow-lint/timing_policy.rs new file mode 100644 index 0000000..81820d9 --- /dev/null +++ b/tools/actor-control-flow-lint/timing_policy.rs @@ -0,0 +1,602 @@ +/// Calls with these resolved item names can create wall-clock waits or +/// failure-producing deadlines. They are denied in every workspace role unless +/// the exact compiler-resolved call site is registered below. +const GUARDED_TIMING_ITEMS: &[&str] = &[ + "park_timeout", + "recv_deadline", + "set_read_timeout", + "recv_timeout", + "sleep", + "sleep_until", + "timeout", + "timeout_at", + "wait_timeout", + "wait_timeout_while", +]; + +#[derive(Clone, Copy)] +struct TimingAllowance { + package: &'static str, + caller: &'static str, + callee: &'static str, + calls: usize, + purpose: &'static str, +} + +const fn timing( + package: &'static str, + caller: &'static str, + callee: &'static str, + calls: usize, + purpose: &'static str, +) -> TimingAllowance { + TimingAllowance { + package, + caller, + callee, + calls, + purpose, + } +} + +/// Exact audited boundaries. A rename, a different API, or one additional call +/// exceeds this registry and fails compilation. Never register a generic +/// timeout wrapper: callers of a timing API must remain visible to this guard. +const TIMING_ALLOWANCES: &[TimingAllowance] = &[ + // Runtime and external-resource boundaries. + timing( + "swactor-engine", + "::poll", + "tokio::time::sleep", + 1, + "Tokio execution-backend adapter", + ), + timing( + "swactor-process", + "::read", + "std::thread::sleep", + 1, + "external process-file polling", + ), + timing( + "swactor-process", + "operations::find_process_identities_with_retry", + "std::thread::sleep", + 1, + "external process discovery retry", + ), + timing( + "swactor-process", + "operations::spawn_identity_exit_wait::{closure#0}", + "std::thread::sleep", + 1, + "external process exit polling", + ), + timing( + "swactor-process", + "operations::spawn_shared_child_wait::{closure#0}", + "std::thread::sleep", + 1, + "external child exit polling", + ), + timing( + "swactor-process", + "operations::terminate_process_group", + "std::thread::sleep", + 1, + "process termination grace period", + ), + timing( + "swactor-process", + "operations::wait_for_path", + "std::thread::sleep", + 1, + "external path publication polling", + ), + timing( + "swactor-process", + "operations::wait_shared_child_or_kill", + "std::thread::sleep", + 1, + "child termination grace period", + ), + timing( + "iroh-driver", + "edge_transport::spawn_edge_send_pump", + "std::sync::mpsc::Receiver::recv_timeout", + 1, + "edge sender startup handoff", + ), + timing( + "iroh-driver", + "edge_transport::spawn_edge_send_pump::{closure#0}::{closure#0}", + "swactor_engine::EngineHandle::timeout", + 2, + "edge stream write and finish deadlines", + ), + timing( + "iroh-driver", + "iroh_driver::AdapterPump::spawn_connect::{closure#0}", + "swactor_engine::EngineHandle::timeout", + 1, + "peer connection deadline", + ), + timing( + "iroh-driver", + "iroh_driver::IrohDriver::send_tagged_gossip::{closure#0}", + "swactor_engine::EngineHandle::timeout", + 1, + "gossip connection deadline", + ), + timing( + "iroh-driver", + "iroh_driver::IrohDriver::connect_peer::{closure#0}", + "swactor_engine::EngineHandle::timeout", + 1, + "peer connection deadline", + ), + timing( + "iroh-driver", + "iroh_driver::IrohDriver::spawn_join_request::{closure#0}", + "swactor_engine::EngineHandle::timeout", + 1, + "seed connection deadline", + ), + timing( + "swactor-vastai", + "blocking::BlockingVastClient::list_by_label_with_retry", + "std::thread::sleep", + 1, + "provider API retry backoff", + ), + timing( + "swactor-vastai", + "client::VastClient::with_base_url", + "reqwest::ClientBuilder::timeout", + 1, + "provider HTTP request deadline", + ), + timing( + "swactor-vastai", + "lease::provision_fleet::{closure#0}", + "tokio::time::sleep", + 1, + "provider lease pacing", + ), + timing( + "swactor-vastai", + "lease::provision_one::{closure#0}", + "tokio::time::sleep", + 1, + "provider retry backoff", + ), + timing( + "swactor-vastai", + "monitor::wait_for_running_with_policy::{closure#0}", + "tokio::time::sleep", + 2, + "provider lifecycle polling", + ), + timing( + "swactor-vastai", + "monitor::wait_for_ssh_endpoint_with_policy::{closure#0}", + "tokio::time::sleep", + 2, + "provider endpoint polling", + ), + timing( + "swactor-vastai", + "teardown::destroy_instance_with_retry_policy::{closure#0}", + "tokio::time::sleep", + 1, + "provider teardown backoff", + ), + timing( + "telemetry", + "endpoint::TelemetrySubscription::recv_timeout", + "crossbeam_channel::Receiver::recv_timeout", + 1, + "bounded external telemetry observation", + ), + // Test and harness safety fuses. These bound real external concurrency; + // virtual-time behavior continues to use the stepping backend. + timing( + "dashboard", + "control::properties::generated_concurrent_bridge_commands_forward_once_and_shutdown::{closure#1}", + "std::sync::mpsc::Receiver::recv_timeout", + 1, + "test completion fuse", + ), + timing( + "dashboard", + "server::tests::send_control_request::{closure#0}", + "tokio::time::timeout", + 1, + "test HTTP completion fuse", + ), + timing( + "data-plane", + "cancellation_and_transport_faults_reclaim_waiters_and_preserve_unrelated_progress::{closure#0}", + "std::thread::sleep", + 1, + "test cleanup observation poll", + ), + timing( + "data-plane", + "cancelled_write_open_releases_queued_grant", + "std::thread::sleep", + 2, + "test cancellation observation poll", + ), + timing( + "data-plane", + "host_read_resolves_file_source_and_seals_final_arena_lease", + "std::thread::sleep", + 4, + "test lease reclamation observation poll", + ), + timing( + "data-plane", + "live_view_prevents_reclaim_until_last_guard_drops", + "std::thread::sleep", + 1, + "test lease reclamation observation poll", + ), + timing( + "distribution", + "snapshot_and_swim_telemetry::swim_telemetry_records_probe_events_and_timeout_state_without_fabricating_rtt", + "std::thread::sleep", + 2, + "wall-clock telemetry contract", + ), + timing( + "iroh-driver", + "common::iroh::IrohTestCluster::pump_until", + "std::thread::sleep", + 1, + "live network convergence fuse", + ), + timing( + "iroh-driver", + "common::iroh::pump_until", + "std::thread::sleep", + 1, + "live network convergence fuse", + ), + timing( + "iroh-driver", + "common::iroh::pump_until_pair", + "std::thread::sleep", + 1, + "live network convergence fuse", + ), + timing( + "iroh-driver", + "endpoint_addr_includes_home_relay", + "std::thread::sleep", + 1, + "live relay convergence fuse", + ), + timing( + "iroh-driver", + "pull_collector_cancellation_interrupts_inflight_io", + "std::sync::mpsc::Receiver::recv_timeout", + 1, + "test setup completion fuse", + ), + timing( + "iroh-driver", + "pull_collector_cancellation_interrupts_inflight_io", + "std::thread::sleep", + 2, + "live network cancellation observation", + ), + timing( + "iroh-driver", + "real_iroh_transfer_delivers_exact_file_bytes", + "std::sync::mpsc::Receiver::recv_timeout", + 2, + "live transfer completion fuse", + ), + timing( + "iroh-driver", + "real_iroh_transfer_delivers_exact_file_bytes", + "std::thread::sleep", + 2, + "live transfer progress poll", + ), + timing( + "iroh-driver", + "same_runtime_transfer_bypasses_iroh_self_connection", + "std::sync::mpsc::Receiver::recv_timeout", + 4, + "local transfer completion fuse", + ), + timing( + "iroh-driver", + "edge_transport::EdgeSendHandle::finish", + "std::sync::mpsc::Receiver::recv_timeout", + 1, + "edge transport completion fuse", + ), + timing( + "iroh-driver", + "recv_until", + "std::sync::mpsc::Receiver::recv_timeout", + 1, + "live transport observation fuse", + ), + timing( + "myelin", + "node::worker_node_runtime::control_flow_properties::agent_cpu_sampler_emits_decodable_host_stats", + "std::thread::sleep", + 1, + "hardware sampler observation", + ), + timing( + "myelin", + "orchestration::app::hardware_telemetry_tests::orchestrator_emits_all_host_hardware_channels", + "std::thread::sleep", + 1, + "hardware sampler observation", + ), + timing( + "myelin", + "provisioning::tests::local_process_stop_releases_all_observer_actors", + "std::thread::sleep", + 1, + "process shutdown observation", + ), + timing( + "myelin", + "provisioning::tests::recv_provider_event", + "std::sync::mpsc::Receiver::recv_timeout", + 1, + "provider observation fuse", + ), + timing( + "myelin", + "tests::data_namespace_guarantees::wait_until", + "std::thread::sleep", + 1, + "live namespace convergence fuse", + ), + timing( + "myelin", + "tests::engine_composition::poll_connect", + "std::thread::sleep", + 1, + "live server connection fuse", + ), + timing( + "myelin", + "tests::engine_composition::recv_within", + "std::thread::sleep", + 1, + "engine observation fuse", + ), + timing( + "myelin-e2e-fuzz", + "harness::control::::kill_node", + "std::thread::sleep", + 1, + "external node termination polling", + ), + timing( + "myelin-e2e-fuzz", + "harness::convergence::::provision", + "std::thread::sleep", + 1, + "external provisioning polling", + ), + timing( + "myelin-e2e-fuzz", + "harness::ClusterHarness::teardown", + "std::thread::sleep", + 1, + "external process teardown polling", + ), + timing( + "myelin-e2e-fuzz", + "harness::convergence::::wait_contextual_control", + "std::thread::sleep", + 1, + "external control-plane polling", + ), + timing( + "myelin-e2e-fuzz", + "harness::control::::wait_execution", + "std::thread::sleep", + 1, + "external execution polling", + ), + timing( + "myelin-e2e-fuzz", + "harness::convergence::::wait_for_dashboard", + "std::thread::sleep", + 1, + "external dashboard readiness polling", + ), + timing( + "myelin-e2e-fuzz", + "harness::control::::stop_orchestrator_injection", + "std::thread::sleep", + 1, + "external orchestrator termination polling", + ), + timing( + "myelin-e2e-fuzz", + "harness::control::::wait_for_no_workload_processes", + "std::thread::sleep", + 1, + "external native-process census polling", + ), + timing( + "myelin-e2e-fuzz", + "harness::ClusterHarness::assert_healthy", + "std::thread::sleep", + 1, + "external resource cleanup polling", + ), + timing( + "myelin-e2e-fuzz", + "resources::http_json", + "ureq::Request::timeout", + 2, + "external HTTP request fuse", + ), + timing( + "swactor-engine", + "common::wait_for", + "std::thread::sleep", + 1, + "engine test completion fuse", + ), + timing( + "swactor-engine", + "engine_interval_recurs", + "std::sync::mpsc::Receiver::recv_timeout", + 1, + "engine test completion fuse", + ), + timing( + "swactor-engine", + "engine_runs_without_an_ambient_tokio_runtime", + "std::sync::mpsc::Receiver::recv_timeout", + 1, + "engine test completion fuse", + ), + timing( + "swactor-engine", + "engine_timer_can_be_created_off_runtime", + "std::sync::mpsc::Receiver::recv_timeout", + 1, + "engine test completion fuse", + ), + timing( + "swactor-engine", + "engine_timer_fires", + "std::sync::mpsc::Receiver::recv_timeout", + 1, + "engine test completion fuse", + ), + timing( + "swactor-engine", + "idle_core_observes_late_external_work_within_idle_interval", + "std::thread::sleep", + 1, + "real idle-driver timing contract", + ), + timing( + "swactor-engine", + "spawned_supporting_work_runs", + "std::sync::mpsc::Receiver::recv_timeout", + 1, + "engine test completion fuse", + ), + timing( + "swactor-engine", + "stepping_blocking_work_runs_isolated", + "std::thread::sleep", + 1, + "blocking worker completion poll", + ), + timing( + "swactor-engine", + "stepping_blocking_work_runs_isolated::{closure#0}", + "std::thread::sleep", + 1, + "blocking worker timing probe", + ), + timing( + "swactor-process", + "drive_once", + "std::thread::sleep", + 1, + "process integration driver pacing", + ), + timing( + "swactor-process", + "supervisor::tests::collect_until_finished", + "std::thread::sleep", + 1, + "process supervisor completion fuse", + ), + timing( + "swactor-process", + "supervisor::tests::join_finished", + "std::thread::sleep", + 1, + "process supervisor completion fuse", + ), + timing( + "swactor-process", + "supervisor::tests::supervisor_stop_escalates_to_kill_after_deadline", + "std::thread::sleep", + 1, + "process escalation timing contract", + ), + timing( + "swactor-vastai", + "monitor::tests::loading_state_remains_slow_progress_before_terminal_evidence::{closure#0}", + "tokio::time::timeout", + 1, + "provider test non-completion fuse", + ), + timing( + "telemetry", + "hardware::tests::blocking_sampler_runs_sequentially_with_monotonic_sequences", + "std::sync::mpsc::Receiver::recv_timeout", + 2, + "sampler test completion fuse", + ), + timing( + "telemetry", + "submit_text_owned_queues_owned_string", + "telemetry::TelemetrySubscription::recv_timeout", + 1, + "telemetry test observation fuse", + ), + timing( + "xtask", + "demo::feed::properties::generated_supervisor_transitions_are_once_only_nonblocking_and_clean::{closure#1}", + "std::sync::mpsc::Receiver::recv_timeout", + 1, + "blocking-provider test completion fuse", + ), + timing( + "actor-lint-test-wait-pass", + "bounded_observation_wait_is_allowed", + "std::sync::mpsc::Receiver::recv_timeout", + 1, + "compiler policy pass fixture", + ), + timing( + "actor-lint-timing-overage", + "exceeds_registered_timing_budget", + "std::sync::mpsc::Receiver::recv_timeout", + 1, + "compiler policy overage fixture", + ), + timing( + "swactor-vastai", + "test_http::serve", + "std::net::TcpStream::set_read_timeout", + 1, + "test HTTP request completion fuse", + ), + timing( + "telemetry", + "carry_over_real_socket", + "std::net::UdpSocket::set_read_timeout", + 1, + "live telemetry socket completion fuse", + ), + timing( + "python", + "context::PyTestDataPlaneHost::run_contextual_process", + "std::thread::sleep", + 1, + "debug test-host contextual output poll interval", + ), +]; diff --git a/tools/timeout-policy.py b/tools/timeout-policy.py new file mode 100755 index 0000000..1f316de --- /dev/null +++ b/tools/timeout-policy.py @@ -0,0 +1,198 @@ +#!/usr/bin/env python3 +"""Reject unregistered Python timeout and sleep call sites.""" + +from __future__ import annotations + +import ast +import sys +from collections import defaultdict +from dataclasses import dataclass +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +SOURCE_ROOTS = ("apps", "crates", "src", "tests", "tools", "xtask") +SKIPPED_PARTS = {".venv", "__pycache__", "target"} + + +@dataclass(frozen=True) +class Allowance: + path: str + caller: str + callee: str + calls: int + purpose: str + + +ALLOWANCES = ( + Allowance( + "apps/myelin/node-image/tinygrad_worker.py", + "fetch_whole", + "urllib.request.urlopen", + 1, + "model download request deadline", + ), + Allowance( + "apps/myelin/node-image/tinygrad_worker.py", + "CpuLineSampler._run", + "self._stop.wait", + 1, + "cpu sampler poll interval", + ), + Allowance( + "crates/bindings/python/tests/test_bootstrap.py", + "test_real_exec_attachment_and_blob_mapping", + "subprocess.run", + 1, + "child-process test completion fuse", + ), +) + + +class TimingVisitor(ast.NodeVisitor): + def __init__(self, path: str) -> None: + self.path = path + self.aliases: dict[str, str] = {} + self.scopes: list[str] = [] + self.occurrences: defaultdict[tuple[str, str], int] = defaultdict(int) + self.violations: list[str] = [] + + def visit_Import(self, node: ast.Import) -> None: + for name in node.names: + if name.asname: + self.aliases[name.asname] = name.name + else: + root = name.name.split(".", 1)[0] + self.aliases[root] = root + + def visit_ImportFrom(self, node: ast.ImportFrom) -> None: + if node.module is not None: + for name in node.names: + self.aliases[name.asname or name.name] = f"{node.module}.{name.name}" + + def visit_ClassDef(self, node: ast.ClassDef) -> None: + self.scopes.append(node.name) + self.generic_visit(node) + self.scopes.pop() + + def visit_FunctionDef(self, node: ast.FunctionDef) -> None: + self.scopes.append(node.name) + self.generic_visit(node) + self.scopes.pop() + + def visit_AsyncFunctionDef(self, node: ast.AsyncFunctionDef) -> None: + self.scopes.append(node.name) + self.generic_visit(node) + self.scopes.pop() + + def visit_Call(self, node: ast.Call) -> None: + callee = self._callee(node.func) + if callee is not None and self._is_guarded(node, callee): + caller = ".".join(self.scopes) or "" + key = (caller, callee) + self.occurrences[key] += 1 + occurrence = self.occurrences[key] + allowance = next( + ( + item + for item in ALLOWANCES + if item.path == self.path + and item.caller == caller + and item.callee == callee + ), + None, + ) + if allowance is None: + self.violations.append( + f"{self.path}:{node.lineno}: unapproved timing primitive " + f"`{callee}` in `{caller}`" + ) + elif occurrence > allowance.calls: + self.violations.append( + f"{self.path}:{node.lineno}: timing occurrence {occurrence} exceeds " + f"the {allowance.calls} audited call(s) in `{caller}` " + f"(`{allowance.purpose}`)" + ) + self.generic_visit(node) + + def _callee(self, node: ast.expr) -> str | None: + parts: list[str] = [] + while isinstance(node, ast.Attribute): + parts.append(node.attr) + node = node.value + if not isinstance(node, ast.Name): + return None + root = self.aliases.get(node.id, node.id) + return ".".join((root, *reversed(parts))) + + @staticmethod + def _is_guarded(node: ast.Call, callee: str) -> bool: + if any(keyword.arg == "timeout" for keyword in node.keywords): + return True + return callee in {"asyncio.sleep", "asyncio.wait_for", "time.sleep"} or callee.endswith( + (".set_read_timeout", ".set_write_timeout", ".settimeout", ".wait") + ) + + +def python_sources() -> list[Path]: + sources: list[Path] = [] + for source_root in SOURCE_ROOTS: + for path in (ROOT / source_root).rglob("*.py"): + if not SKIPPED_PARTS.intersection(path.parts): + sources.append(path) + return sorted(sources) + + +def check_source(path: Path) -> list[str]: + relative = path.relative_to(ROOT).as_posix() + source = path.read_text(encoding="utf-8") + syntax = ast.parse(source, filename=relative) + visitor = TimingVisitor(relative) + visitor.visit(syntax) + return visitor.violations + + +def self_test() -> None: + visitor = TimingVisitor("probe.py") + visitor.visit(ast.parse("import time as clock\ndef work():\n clock.sleep(1)\n")) + assert visitor.violations == [ + "probe.py:3: unapproved timing primitive `time.sleep` in `work`" + ] + + overage = TimingVisitor("crates/bindings/python/tests/test_bootstrap.py") + overage.visit( + ast.parse( + "import subprocess\n" + "def test_real_exec_attachment_and_blob_mapping():\n" + " subprocess.run([], timeout=1)\n" + " subprocess.run([], timeout=1)\n" + ) + ) + assert overage.violations == [ + "crates/bindings/python/tests/test_bootstrap.py:4: timing occurrence 2 " + "exceeds the 1 audited call(s) in " + "`test_real_exec_attachment_and_blob_mapping` " + "(`child-process test completion fuse`)" + ] + + +def main() -> int: + if sys.argv[1:] == ["--self-test"]: + self_test() + return 0 + if sys.argv[1:]: + print("usage: timeout-policy.py [--self-test]", file=sys.stderr) + return 2 + + self_test() + + violations = [violation for path in python_sources() for violation in check_source(path)] + if violations: + print("Python timeout policy violations:", file=sys.stderr) + print("\n".join(violations), file=sys.stderr) + return 1 + print("python-timeout-policy: OK") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tools/vastai/src/logs.rs b/tools/vastai/src/logs.rs index 647681c..5851ed4 100644 --- a/tools/vastai/src/logs.rs +++ b/tools/vastai/src/logs.rs @@ -1,5 +1,3 @@ -use std::time::Duration; - pub async fn request_logs( client: &reqwest::Client, base_url: &str, @@ -24,7 +22,6 @@ pub async fn request_logs( } pub async fn fetch_logs(client: &reqwest::Client, log_url: &str) -> Result { - tokio::time::sleep(Duration::from_secs(5)).await; let resp = client .get(log_url) .send() diff --git a/tools/vastai/src/test_http.rs b/tools/vastai/src/test_http.rs index 58f9f72..861603b 100644 --- a/tools/vastai/src/test_http.rs +++ b/tools/vastai/src/test_http.rs @@ -55,9 +55,6 @@ impl TestHttpServer { pub fn start(routes: Vec) -> Result { let listener = TcpListener::bind(("127.0.0.1", 0)) .map_err(|error| format!("bind test HTTP server: {error}"))?; - listener - .set_nonblocking(true) - .map_err(|error| format!("configure test HTTP server: {error}"))?; let address = listener .local_addr() .map_err(|error| format!("read test HTTP address: {error}"))?; @@ -66,14 +63,11 @@ impl TestHttpServer { let thread_requests = Arc::clone(&requests); let thread_stop = Arc::clone(&stop); let join = std::thread::spawn(move || { - while !thread_stop.load(Ordering::Acquire) { - match listener.accept() { - Ok((stream, _)) => serve(stream, &routes, &thread_requests), - Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => { - std::thread::sleep(Duration::from_millis(5)); - } - Err(_) => break, + while let Ok((stream, _)) = listener.accept() { + if thread_stop.load(Ordering::Acquire) { + break; } + serve(stream, &routes, &thread_requests); } }); Ok(Self { @@ -96,6 +90,7 @@ impl TestHttpServer { impl Drop for TestHttpServer { fn drop(&mut self) { self.stop.store(true, Ordering::Release); + let _ = TcpStream::connect(self.address); if let Some(join) = self.join.take() { let _ = join.join(); } diff --git a/xtask/src/main.rs b/xtask/src/main.rs index 34b1a24..d1e1905 100644 --- a/xtask/src/main.rs +++ b/xtask/src/main.rs @@ -13,7 +13,7 @@ const TEST_STEPS: &[TestStep] = &[ args: &["lint"], }, TestStep { - label: "all Rust tests (60s per-test timeout)", + label: "all Rust tests (180s per-test timeout)", args: &["nextest", "run", "--workspace", "--all-features"], }, TestStep { @@ -82,6 +82,7 @@ COMMANDS: Run the visual provisioning-reconciler demo. check-telemetry-isolation Verify no frame types appear in control-plane modules. + lint Run the Python timeout policy and strict workspace Clippy. test Run strict lint plus every Rust and Python test." ); } @@ -138,6 +139,31 @@ fn run_tests() -> ExitCode { ExitCode::SUCCESS } +fn run_lint() -> ExitCode { + let policy = workspace_root().join("tools/timeout-policy.py"); + let policy_status = swactor_process::command_status(Command::new("python3").arg(policy)); + if !policy_status.is_ok_and(|status| status.success()) { + return ExitCode::from(1); + } + + let mut clippy = cargo_command(); + clippy.args(["clippy", "--workspace", "--all-targets", "--all-features"]); + // Pin pyo3 to the same interpreter the Python test flow uses: a newer + // system python exceeds the bundled pyo3's supported maximum and would + // fail the workspace build here even though tests build fine. + if let Some(tools) = PythonTestTools::discover() { + clippy.env("PYO3_PYTHON", tools.python); + } + match swactor_process::command_status(&mut clippy) { + Ok(status) if status.success() => ExitCode::SUCCESS, + Ok(_) => ExitCode::from(1), + Err(error) => { + eprintln!("Failed to execute cargo clippy: {error}"); + ExitCode::from(1) + } + } +} + fn nextest_available() -> bool { let available = swactor_process::command_status( cargo_command() @@ -185,7 +211,8 @@ impl PythonTestTools { build_command .current_dir(&self.directory) .arg("develop") - .env("PYO3_PYTHON", &self.python), + .arg("--features") + .arg("test-host"), ) .is_ok_and(|status| status.success()); if !built { @@ -298,6 +325,7 @@ fn main() -> ExitCode { ExitCode::from(1) } } + Some("lint") if args.next().is_none() => run_lint(), Some("test") if args.next().is_none() => run_tests(), Some("demo") => demo::run(&args.collect::>()), Some("help" | "--help" | "-h") | None => {