From 1a95d037d1477797c721fc2798839fbf902f201e Mon Sep 17 00:00:00 2001 From: jj Date: Mon, 7 Sep 2026 22:25:21 +0200 Subject: [PATCH 1/2] worker: add option to mount cached input directories as read-only Chromium builds for macOS use the same Mac SDK across thousands of compile jobs. The directory cache already shares the prepared SDK, but each job still recreates its directories and links, then deletes them afterward. Add experimental_readonly_input_mounts so Linux workers can mount selected cached directories read-only into each job instead. This avoids repeated filesystem work for large, unchanged inputs such as SDKs. The option is disabled by default. Mounted directories stay in the cache until job cleanup finishes, and incompatible jobs use normal input setup. Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_01HBxQtV9S561AzBo7aPCYFj --- .../vocabularies/TraceMachina/accept.txt | 1 + nativelink-config/src/cas_server.rs | 36 ++ nativelink-util/src/metrics.rs | 37 ++ nativelink-worker/BUILD.bazel | 1 + nativelink-worker/src/directory_cache.rs | 176 ++++++-- nativelink-worker/src/input_mounts.rs | 235 +++++++++++ nativelink-worker/src/lib.rs | 1 + nativelink-worker/src/local_worker.rs | 10 + nativelink-worker/src/namespace_utils.rs | 128 +++++- .../src/running_actions_manager.rs | 254 ++++++++++-- .../tests/directory_cache_test.rs | 175 ++++++++ .../tests/namespace_utils_test.rs | 143 +++++++ .../tests/running_actions_manager_test.rs | 382 ++++++++++++++++++ .../docs/explanations/architecture.mdx | 3 + .../content/docs/operate/observability.mdx | 9 + .../docs/reference/nativelink-config/main.mdx | 20 +- 16 files changed, 1549 insertions(+), 62 deletions(-) create mode 100644 nativelink-worker/src/input_mounts.rs diff --git a/.github/styles/config/vocabularies/TraceMachina/accept.txt b/.github/styles/config/vocabularies/TraceMachina/accept.txt index f0201e45f8b..c0832d0ef15 100644 --- a/.github/styles/config/vocabularies/TraceMachina/accept.txt +++ b/.github/styles/config/vocabularies/TraceMachina/accept.txt @@ -317,6 +317,7 @@ deserialization Dockerised Merkle subtree +subtrees hardlink multiplicatively SELinux diff --git a/nativelink-config/src/cas_server.rs b/nativelink-config/src/cas_server.rs index bb55746500d..b4c598158a4 100644 --- a/nativelink-config/src/cas_server.rs +++ b/nativelink-config/src/cas_server.rs @@ -1538,6 +1538,42 @@ pub struct LocalWorkerConfig { #[serde(default)] pub experimental_active_input_leases: bool, + /// Reuse immutable input directories through read-only bind mounts on Linux. + /// The worker prepares each selected subtree once per REAPI `Directory` + /// digest in the directory cache, then mounts it into each action instead of + /// recreating its filesystem entries. + /// + /// Paths are relative to the action input root, for example `["out/x/sdk"]`. + /// They must be nonempty, canonical, and non-overlapping. Absolute paths, + /// empty path components, and `.`/`..` components are rejected. + /// Requires `directory_cache`, `use_namespaces: true`, and + /// `use_mount_namespace: true`. + /// + /// Actions declaring overlapping outputs, working inside a selected subtree, + /// or using a persistent worker fall back to ordinary materialization. + /// Output paths reached through symlink aliases cannot be checked for + /// overlap. Use this only with trusted workloads that do not modify the + /// selected trees; read-only input mounts are not a security boundary. + /// + /// Select trees whose contents (and Directory digest) are reused across + /// many actions. Trees that change per action add cache construction work + /// without reuse. When mounts apply, other inputs use ordinary downloads, + /// bypassing both root and subtree directory caching for that action. + /// Failed mount preparation retries once without mounts; a failed mount + /// syscall still aborts the action before its command runs. + /// + /// An action can remount its inputs writable in its own user namespace + /// and corrupt the shared cache. Later actions can then produce incorrect + /// results from those inputs. Enable only for trusted commands. + /// + /// Cache entries remain pinned until action cleanup and may temporarily + /// exceed the cache's eviction budget. Pinned bytes have no separate limit, + /// so concurrent distinct trees can exhaust the disk. Leave headroom. + /// + /// Default: [] (disabled) + #[serde(default)] + pub experimental_readonly_input_mounts: Vec, + /// Whether to use namespaces to isolate the execution. This is only available /// on Linux. It is highly recommended as it avoids a number of issues with /// zombie processes and also provides additional hermeticity. If explicitly set diff --git a/nativelink-util/src/metrics.rs b/nativelink-util/src/metrics.rs index cd6405b1576..f35b5dfeb81 100644 --- a/nativelink-util/src/metrics.rs +++ b/nativelink-util/src/metrics.rs @@ -888,6 +888,43 @@ pub fn execution_output_bytes(action_result: &ActionResult) -> u64 { + action_result.stderr_digest.size_bytes() } +/// Input mount counters, emitted by workers rather than schedulers. +pub static INPUT_MOUNT_METRICS: LazyLock = LazyLock::new(|| { + let meter = global::meter_with_scope(InstrumentationScope::builder("nativelink").build()); + InputMountMetrics { + cache_hits: meter + .u64_counter("worker.input_mount.cache_hits") + .with_description("Cached directories reused for input mount preparation") + .build(), + cache_misses: meter + .u64_counter("worker.input_mount.cache_misses") + .with_description("Input mount directories requiring cache construction") + .build(), + fallbacks: meter + .u64_counter("worker.input_mount.fallbacks") + .with_description( + "Actions using ordinary input preparation despite configured input mounts", + ) + .build(), + prepare_failures: meter + .u64_counter("worker.input_mount.prepare_failures") + .with_description("Failed input mount preparations retried without mounts") + .build(), + } +}); + +#[derive(Debug)] +pub struct InputMountMetrics { + /// Cached directories reused for input mount preparation. + pub cache_hits: metrics::Counter, + /// Input mount directories requiring cache construction. + pub cache_misses: metrics::Counter, + /// Actions using ordinary input preparation despite configured input mounts. + pub fallbacks: metrics::Counter, + /// Failed input mount preparations retried without mounts. + pub prepare_failures: metrics::Counter, +} + /// Global worker fleet metrics instruments. pub static WORKER_METRICS: LazyLock = LazyLock::new(|| { let meter = global::meter_with_scope(InstrumentationScope::builder("nativelink").build()); diff --git a/nativelink-worker/BUILD.bazel b/nativelink-worker/BUILD.bazel index b32036527eb..fa2c2076830 100644 --- a/nativelink-worker/BUILD.bazel +++ b/nativelink-worker/BUILD.bazel @@ -13,6 +13,7 @@ rust_library( "src/buck2_file_capture.rs", "src/capacity.rs", "src/directory_cache.rs", + "src/input_mounts.rs", "src/lib.rs", "src/local_worker.rs", "src/persistent_worker/live_worker.rs", diff --git a/nativelink-worker/src/directory_cache.rs b/nativelink-worker/src/directory_cache.rs index a376fc5589b..49c6f6a6082 100644 --- a/nativelink-worker/src/directory_cache.rs +++ b/nativelink-worker/src/directory_cache.rs @@ -171,6 +171,22 @@ impl Drop for EntryPin { } } +/// A prepared tree held against eviction for the lifetime of this handle. +/// A cache hit can return this handle without walking or recreating the tree. +/// Consumers must expose the prepared tree read-only. +#[derive(Debug)] +pub struct PreparedDirectory { + path: PathBuf, + size: u64, + pin: EntryPin, +} + +impl PreparedDirectory { + pub fn path(&self) -> &Path { + &self.path + } +} + /// Drop guard for an in-progress temp construction tree: if the owning /// future is dropped before [`Self::disarm`] (cancellation — e.g. /// `buffer_unordered` drops sibling constructions on the first error), the @@ -441,6 +457,44 @@ impl DirectoryCache { Ok(hit) } + /// Acquires a reusable tree without copying it to an action workspace. + /// The returned handle must outlive every mount or other use of its path. + pub async fn prepare_for_mount( + &self, + digest: DigestInfo, + lease: Option<&dyn DigestLease>, + ) -> Result { + self.maybe_log_summary(); + acquire_digest(lease, &digest); + if let Some(prepared) = self.acquire_entry(&digest).await { + nativelink_util::metrics::INPUT_MOUNT_METRICS + .cache_hits + .add(1, &[]); + return Ok(prepared); + } + let lock = { + let mut locks = self.construction_locks.lock().await; + locks + .entry(digest) + .or_insert_with(|| Arc::new(Mutex::new(()))) + .clone() + }; + let _guard = lock.lock().await; + let result = if let Some(prepared) = self.acquire_entry(&digest).await { + nativelink_util::metrics::INPUT_MOUNT_METRICS + .cache_hits + .add(1, &[]); + Ok(prepared) + } else { + nativelink_util::metrics::INPUT_MOUNT_METRICS + .cache_misses + .add(1, &[]); + self.construct_entry(digest, None, true, lease).await + }; + self.forget_construction_lock(&digest).await; + result + } + /// Core get-or-create flow shared by root materializations /// ([`Self::get_or_create`]) and, with `experimental_subtree_caching` /// enabled, subtree materializations (`create_subdirectory`). Returns @@ -466,7 +520,7 @@ impl DirectoryCache { acquire_digest(lease, &digest); // Fast path: serve from an existing entry. - if let Some(size) = self.try_materialize_from_cache(&digest, dest_path).await { + if let Some(size) = self.try_materialize_from_cache(&digest, dest_path).await? { return Ok((true, size)); } @@ -514,33 +568,44 @@ impl DirectoryCache { /// case the dead entry has been invalidated (removed + tombstoned, so it /// cannot fail every future request forever) and the possibly partially /// populated destination cleared, so the caller can construct fresh. + /// If another caller still holds the entry, return the error without + /// invalidating it: that caller may be using its tree through a mount. async fn try_materialize_from_cache( &self, digest: &DigestInfo, dest_path: &Path, - ) -> Option { - let (cache_path, size, pin) = self.acquire_entry(digest).await?; + ) -> Result, Error> { + let Some(PreparedDirectory { + path: cache_path, + size, + pin, + }) = self.acquire_entry(digest).await + else { + return Ok(None); + }; debug!(?digest, ?cache_path, "Directory cache HIT"); match hardlink_directory_tree(&cache_path, dest_path).await { Ok(method) => { drop(pin); self.record_clone_method(method); - Some(size) + Ok(Some(size)) } Err(e) => { + if !self.invalidate_entry(digest, &pin).await { + return Err(e).err_tip(|| "Cannot rebuild a directory while it is in use"); + } warn!( ?digest, error = ?e, "Failed to hardlink from cache, invalidating entry and reconstructing" ); - self.invalidate_entry(digest, &pin).await; drop(pin); // The failed walk may have partially populated the // destination; clear it so the reconstruction starts clean // (`hardlink_directory_tree` refuses an existing destination // and `create_file` fails on leftovers). Self::remove_tree_best_effort(dest_path).await; - None + Ok(None) } } } @@ -562,10 +627,28 @@ impl DirectoryCache { // we waited on the construction lock. If the entry turns out to be // damaged it is invalidated and we fall through to rebuild it fresh // (exactly once — we hold the construction lock). - if let Some(size) = self.try_materialize_from_cache(&digest, dest_path).await { + if let Some(size) = self.try_materialize_from_cache(&digest, dest_path).await? { return Ok((true, size)); } + let prepared = self + .construct_entry(digest, protos, prefetch_on_miss, lease) + .await?; + let method = hardlink_directory_tree(prepared.path(), dest_path) + .await + .err_tip(|| "Failed to hardlink newly cached directory")?; + self.record_clone_method(method); + Ok((false, prepared.size)) + } + + /// Constructs and publishes a pinned entry, without copying it elsewhere. + async fn construct_entry( + &self, + digest: DigestInfo, + protos: Option<&HashMap>, + prefetch_on_miss: bool, + lease: Option<&dyn DigestLease>, + ) -> Result { // Construct the directory into a UNIQUE temp path first, then // atomically publish it to the canonical `cache_root/` path // via rename. Constructing at the canonical path directly would @@ -630,11 +713,9 @@ impl DirectoryCache { // the expensive filesystem deletion is dispatched off the lock so // eviction I/O never serializes other callers. // - // The new entry is inserted pre-pinned. The hardlink-to-destination - // below runs unlocked, and a concurrent caller for an unrelated - // digest could otherwise pick this brand-new entry as an eviction - // victim and delete its tree mid-hardlink. Dropping the pin releases - // it once the hardlink is done. + // Publish already pinned: a concurrent insertion must not evict the + // tree before its caller has finished copying or mounting it. The + // returned handle keeps the pin alive for the entire use. let ref_count = Arc::new(AtomicUsize::new(1)); let pin = EntryPin { ref_count: Arc::clone(&ref_count), @@ -664,14 +745,11 @@ impl DirectoryCache { }; Self::dispatch_evictions(tombstones); - // Hardlink to destination (unlocked). The entry is pinned so it - // cannot be evicted from under this hardlink. - let result = hardlink_directory_tree(&cache_path, dest_path).await; - drop(pin); - let method = result.err_tip(|| "Failed to hardlink newly cached directory")?; - self.record_clone_method(method); - - Ok((false, size)) + Ok(PreparedDirectory { + path: cache_path, + size, + pin, + }) } /// Removes a cache entry whose on-disk tree failed to materialize @@ -684,17 +762,24 @@ impl DirectoryCache { /// attempt: invalidation is skipped if the map now holds a DIFFERENT /// entry for this digest (single-flight already rebuilt it), identified /// by refcount-handle pointer identity. - async fn invalidate_entry(&self, digest: &DigestInfo, pin: &EntryPin) { + /// Returns `false` if another caller still pins the entry. + async fn invalidate_entry(&self, digest: &DigestInfo, pin: &EntryPin) -> bool { let tombstones = { let mut cache = self.cache.write().await; let is_same_entry = cache .get(digest) .is_some_and(|m| Arc::ptr_eq(&m.ref_count, &pin.ref_count)); if !is_same_entry { - return; + return true; + } + // The failed materialization owns one pin. Other pins may belong + // to mounted actions, whose source must not be renamed or deleted. + // Holding the write lock prevents new pins during this check. + if pin.ref_count.load(Ordering::SeqCst) > 1 { + return false; } let Some(metadata) = cache.remove(digest) else { - return; + return true; }; self.map_entries.fetch_sub(1, Ordering::Relaxed); self.map_size_bytes @@ -702,6 +787,7 @@ impl DirectoryCache { self.tombstone_victims(vec![metadata.path]).await }; Self::dispatch_evictions(tombstones); + true } /// Best-effort `remove_dir_all` that treats `NotFound` as success and @@ -760,26 +846,26 @@ impl DirectoryCache { } /// If `digest` is cached, pins it against eviction and returns a - /// snapshot of its on-disk path, its recorded size, and the RAII pin. + /// handle holding its on-disk path, recorded size, and the RAII pin. /// Runs under the cache READ lock — the refcount and LRU timestamp are /// atomics, so hits do not serialize on the write lock. The increment is /// race-free against eviction because eviction requires the write lock, /// which excludes every read-lock holder: an entry with an outstanding /// pin is always observed with a non-zero count by `evict_lru`. - async fn acquire_entry(&self, digest: &DigestInfo) -> Option<(PathBuf, u64, EntryPin)> { + async fn acquire_entry(&self, digest: &DigestInfo) -> Option { let cache = self.cache.read().await; let metadata = cache.get(digest)?; metadata .last_access .store(unix_nanos_now(), Ordering::Relaxed); metadata.ref_count.fetch_add(1, Ordering::SeqCst); - Some(( - metadata.path.clone(), - metadata.size, - EntryPin { + Some(PreparedDirectory { + path: metadata.path.clone(), + size: metadata.size, + pin: EntryPin { ref_count: Arc::clone(&metadata.ref_count), }, - )) + }) } /// Removes an idle construction lock after the last caller finishes. @@ -2359,4 +2445,34 @@ mod tests { assert!(cache.construction_locks.lock().await.is_empty()); Ok(()) } + + #[nativelink_test] + async fn failed_materialization_preserves_mounted_entry() -> Result<(), Error> { + let temp_dir = TempDir::new().unwrap(); + let (store, digest) = setup_test_store(&temp_dir).await; + let cache = DirectoryCache::new( + DirectoryCacheConfig { + cache_root: temp_dir.path().join("cache"), + ..Default::default() + }, + store, + ) + .await?; + let prepared = cache.prepare_for_mount(digest, None).await?; + // An invalid destination must not invalidate another action's mount. + let dest = temp_dir.path().join("dest"); + fs::write(&dest, b"not a directory").await?; + assert!(cache.get_or_create(digest, &dest).await.is_err()); + let cached = cache.acquire_entry(&digest).await.unwrap(); + assert!( + Arc::ptr_eq(&prepared.pin.ref_count, &cached.pin.ref_count), + "failed materialization replaced a pinned mount source" + ); + assert!(prepared.path().join("test.txt").exists()); + drop(cached); + drop(prepared); + fs::remove_file(&dest).await?; + assert!(cache.get_or_create(digest, &dest).await?); + Ok(()) + } } diff --git a/nativelink-worker/src/input_mounts.rs b/nativelink-worker/src/input_mounts.rs new file mode 100644 index 00000000000..59c16c82442 --- /dev/null +++ b/nativelink-worker/src/input_mounts.rs @@ -0,0 +1,235 @@ +// Copyright 2026 The NativeLink Authors. All rights reserved. +// +// Licensed under the Functional Source License, Version 1.1, Apache 2.0 Future License (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// See LICENSE file for details +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use std::collections::HashSet; +use std::path::{Component, Path, PathBuf}; +use std::sync::Arc; + +use nativelink_error::{Code, Error, ResultExt, make_err}; +use nativelink_proto::build::bazel::remote::execution::v2::{Command, Directory}; +use nativelink_store::ac_utils::get_and_decode_digest; +use nativelink_store::fast_slow_store::FastSlowStore; +use nativelink_util::common::DigestInfo; +use parking_lot::Mutex; + +use crate::directory_cache::{DigestLease, DirectoryCache, PreparedDirectory}; + +/// Selected immutable input subtrees and their cache pins for one action. +/// Pins remain held through cleanup, including after cancelled preparation. +#[derive(Debug)] +pub struct InputMounts { + cache: Arc, + targets: HashSet, + #[cfg(target_os = "linux")] + root: PathBuf, + prepared: Mutex>, +} + +/// Validates that mount paths are canonical, nonempty, and non-overlapping. +/// Paths are relative to the action input root. Mounting the whole root would +/// prevent the action from creating outputs. +pub fn validate_paths(paths: &[String]) -> Result<(), Error> { + for (i, path) in paths.iter().enumerate() { + if !is_canonical_relative(path) + || paths[..i] + .iter() + .any(|other| paths_overlap(Path::new(path), Path::new(other))) + { + return Err(make_err!( + Code::InvalidArgument, + "Read-only input mount paths must be nonempty, canonical, non-overlapping relative paths: {path:?}" + )); + } + } + Ok(()) +} + +/// Checks only the ancestors of selected paths, without walking their contents. +/// Actions without a matching directory should keep using the ordinary cache. +/// `paths` must have passed `validate_paths`. +pub async fn has_mountable_inputs( + cas_store: &FastSlowStore, + root_digest: DigestInfo, + paths: &[String], + lease: Option<&dyn DigestLease>, +) -> Result { + 'target: for path in paths { + let mut digest = root_digest; + for component in path.split('/') { + if let Some(lease) = lease { + lease.acquire(&digest); + } + let directory = get_and_decode_digest::(cas_store, digest.into()).await?; + let Some(child) = directory + .directories + .iter() + .find(|child| child.name == component) + else { + continue 'target; + }; + digest = child + .digest + .as_ref() + .err_tip(|| "Missing input directory digest")? + .try_into()?; + } + return Ok(true); + } + Ok(false) +} + +fn is_canonical_relative(path: &str) -> bool { + !path.is_empty() + && !path.contains(['\\', '\0']) + && path.split('/').all(|part| !matches!(part, "" | "." | "..")) +} + +// Outputs are relative to the command's working directory and may use `..` +// (for example Chromium's ../clang-crashreports). Normalize lexically within +// the input root before checking overlap. Symlink aliases remain a workload +// constraint; resolving them would require walking the input tree on every hit. +fn resolve_output(cwd: &str, output: &str) -> Option { + if output.contains(['\\', '\0']) { + return None; + } + let mut path = PathBuf::from(cwd); + for component in Path::new(output).components() { + match component { + Component::Normal(part) => path.push(part), + Component::CurDir => {} + Component::ParentDir => { + if !path.pop() { + return None; + } + } + Component::RootDir | Component::Prefix(_) => return None, + } + } + Some(path) +} + +fn paths_overlap(a: &Path, b: &Path) -> bool { + a.starts_with(b) || b.starts_with(a) +} + +impl InputMounts { + /// Creates mount state for a compatible command, or returns `None` to use + /// ordinary materialization. Declared outputs must not overlap a selected + /// subtree: the output collector cannot see the child's private mounts. + pub fn for_command( + cache: Arc, + paths: &[String], + root: &Path, + command: &Command, + ) -> Option> { + let cwd = &command.working_directory; + if !cwd.is_empty() && !is_canonical_relative(cwd) { + return None; + } + if paths.iter().any(|path| Path::new(cwd).starts_with(path)) { + return None; + } + for output in command + .output_paths + .iter() + .chain(&command.output_files) + .chain(&command.output_directories) + { + // Empty output paths designate the whole working directory. + let output = resolve_output(cwd, output)?; + if paths + .iter() + .any(|path| paths_overlap(&output, Path::new(path))) + { + return None; + } + } + Some(Arc::new(Self { + cache, + targets: paths.iter().map(|path| root.join(path)).collect(), + #[cfg(target_os = "linux")] + root: root.to_owned(), + prepared: Mutex::new(Vec::new()), + })) + } + + /// Prepares and pins a selected subtree before the input walk descends into it. + /// Returns `true` when the caller should leave an empty mount point instead + /// of materializing the subtree, or `false` when the path is not selected. + pub async fn prepare( + &self, + path: &Path, + digest: DigestInfo, + lease: Option<&dyn DigestLease>, + ) -> Result { + if !self.targets.contains(path) { + return Ok(false); + } + let directory = self.cache.prepare_for_mount(digest, lease).await?; + self.prepared.lock().push((path.to_owned(), directory)); + Ok(true) + } + + /// Resolve a program in a prepared mount before the child namespace exists. + /// Validate symlinks against the pinned source, then return its path in the + /// child workspace, not the cache path hidden by namespace setup. + #[cfg(target_os = "linux")] + pub fn executable_path( + &self, + cwd: &str, + program: &std::ffi::OsStr, + ) -> Result, Error> { + let Some(program) = program.to_str() else { + return Ok(None); + }; + let Some(relative) = resolve_output(cwd, program) else { + return Ok(None); + }; + let candidate = self.root.join(relative); + for (target, directory) in self.prepared.lock().iter() { + let Ok(suffix) = candidate.strip_prefix(target) else { + continue; + }; + let source = directory.path().canonicalize()?; + let executable = source.join(suffix).canonicalize().err_tip(|| { + format!( + "Could not resolve executable in read-only input mount: {}", + candidate.display() + ) + })?; + let relative = executable.strip_prefix(&source).map_err(|_| { + make_err!( + Code::InvalidArgument, + "Executable symlink escapes its read-only input mount: {}", + candidate.display() + ) + })?; + return Ok(Some(target.join(relative))); + } + Ok(None) + } + + /// Allocates mount paths and reads mount flags before entering pre-exec. + #[cfg(target_os = "linux")] + pub fn bind_mounts(&self) -> Result, Error> { + self.prepared + .lock() + .iter() + .map(|(target, directory)| { + crate::namespace_utils::ReadOnlyBindMount::new(directory.path(), target) + .map_err(Error::from) + }) + .collect() + } +} diff --git a/nativelink-worker/src/lib.rs b/nativelink-worker/src/lib.rs index 48997ca77a8..06e7c8b7724 100644 --- a/nativelink-worker/src/lib.rs +++ b/nativelink-worker/src/lib.rs @@ -15,6 +15,7 @@ pub mod buck2_file_capture; pub mod capacity; pub mod directory_cache; +pub mod input_mounts; pub mod local_worker; #[cfg(target_os = "linux")] pub mod namespace_utils; diff --git a/nativelink-worker/src/local_worker.rs b/nativelink-worker/src/local_worker.rs index 07f3566a583..0a0d10215f5 100644 --- a/nativelink-worker/src/local_worker.rs +++ b/nativelink-worker/src/local_worker.rs @@ -1234,6 +1234,14 @@ pub async fn new_local_worker( #[cfg(not(target_os = "linux"))] let persistent_workers_namespaced = false; + #[cfg(not(target_os = "linux"))] + if !config.experimental_readonly_input_mounts.is_empty() { + return Err(make_err!( + Code::Unavailable, + "Read-only input mounts not supported on non-Linux OSes" + )); + } + let running_actions_manager = Arc::new(RunningActionsManagerImpl::new(RunningActionsManagerArgs { root_action_directory: config.work_directory.clone(), @@ -1271,6 +1279,8 @@ pub async fn new_local_worker( directory_cache, active_input_leases: config.experimental_active_input_leases, #[cfg(target_os = "linux")] + readonly_input_mounts: config.experimental_readonly_input_mounts.clone(), + #[cfg(target_os = "linux")] use_namespaces, })?); // What actions leave behind is reaped here; see `crate::reaper`. diff --git a/nativelink-worker/src/namespace_utils.rs b/nativelink-worker/src/namespace_utils.rs index 206759a92ee..6723555511b 100644 --- a/nativelink-worker/src/namespace_utils.rs +++ b/nativelink-worker/src/namespace_utils.rs @@ -458,6 +458,7 @@ fn perform_remount( tmp_directory: Option<&core::ffi::CStr>, root_action_directory: &core::ffi::CStr, action_directory: &core::ffi::CStr, + input_mounts: &[ReadOnlyBindMount], ) -> Result<(), Error> { // Make the mount namespace private to avoid changes propagating back to the host. // SAFETY: mount is async-signal-safe. We pass a null pointer for the source and valid @@ -475,6 +476,12 @@ fn perform_remount( return Err(Error::last_os_error()); } + // Install inputs while cache paths are still visible. The recursive action + // bind below preserves these mounts when its parent is masked or hidden by + // the private /tmp. Resolving sources here also keeps them in this mount + // namespace, unlike descriptors opened by the parent before unshare. + mount_readonly_inputs(input_mounts)?; + // Bind mount the action directory to itself to "save" its current contents before // we mask its parent. // SAFETY: mount is async-signal-safe. We pass valid C-string pointers for the paths. @@ -566,6 +573,96 @@ fn perform_remount( Ok(()) } +/// Paths for a bind mount, allocated in the parent before the pre-exec hook. +/// The caller must pin the cache entry until the action exits and its mounts are released. +#[derive(Debug)] +pub struct ReadOnlyBindMount { + source: std::ffi::CString, + target: std::ffi::CString, + remount_flags: libc::c_ulong, +} + +impl ReadOnlyBindMount { + pub fn new(source: &std::path::Path, target: &std::path::Path) -> Result { + use std::os::unix::ffi::OsStrExt; + + // Linux statvfs flag, not exposed by the libc crate on musl. + const ST_RELATIME: libc::c_ulong = 0x1000; + + // The child changes its working directory before running pre-exec. + let source = std::path::absolute(source)?; + let target = std::path::absolute(target)?; + let source = std::ffi::CString::new(source.as_os_str().as_bytes())?; + let mut stats = core::mem::MaybeUninit::::uninit(); + // SAFETY: the source is a valid C string and stats is writable storage. + if unsafe { libc::statvfs(source.as_ptr(), stats.as_mut_ptr()) } != 0 { + return Err(Error::last_os_error()); + } + // SAFETY: statvfs initialized the structure on success. + let flags = unsafe { stats.assume_init() }.f_flag; + let mut remount_flags = + libc::MS_BIND | libc::MS_REMOUNT | libc::MS_RDONLY | libc::MS_NOSUID | libc::MS_NODEV; + // A user namespace cannot clear inherited locked mount flags. Preserve + // execute and atime restrictions while adding read-only protection. + for (stat_flag, mount_flag) in [ + (libc::ST_NOEXEC, libc::MS_NOEXEC), + (libc::ST_NOATIME, libc::MS_NOATIME), + (libc::ST_NODIRATIME, libc::MS_NODIRATIME), + (ST_RELATIME, libc::MS_RELATIME), + ] { + if flags & stat_flag != 0 { + remount_flags |= mount_flag; + } + } + Ok(Self { + source, + target: std::ffi::CString::new(target.as_os_str().as_bytes())?, + remount_flags, + }) + } +} + +/// Installs prepared input trees inside the private mount namespace, before +/// masking the action root. +/// +/// Call only from the child's pre-exec hook: all strings and flags are +/// prepared beforehand, and this function uses only mount syscalls and errno. +/// Failure aborts the spawn rather than executing against empty placeholders. +fn mount_readonly_inputs(mounts: &[ReadOnlyBindMount]) -> Result<(), Error> { + for mount in mounts { + // SAFETY: both paths are valid, preallocated C strings; the source + // cache entry is kept alive by the action through its cleanup. + if unsafe { + libc::mount( + mount.source.as_ptr(), + mount.target.as_ptr(), + core::ptr::null(), + libc::MS_BIND, + core::ptr::null(), + ) + } != 0 + { + return Err(Error::last_os_error()); + } + // MS_RDONLY on the initial bind is ignored by Linux; remount the + // action's bind explicitly. This leaves the worker's cache mount alone. + // SAFETY: this modifies only the bind in the child's private namespace. + if unsafe { + libc::mount( + core::ptr::null(), + mount.target.as_ptr(), + core::ptr::null(), + mount.remount_flags, + core::ptr::null(), + ) + } != 0 + { + return Err(Error::last_os_error()); + } + } + Ok(()) +} + /// A hook for a `Command::spawn` to create the process in a new namespace. /// This creates a stub process that the Command points at which forwards /// SIGKILL to the actual process in the new user, PID, UTS and IPC @@ -585,6 +682,30 @@ pub fn configure_namespace( root_action_directory: &core::ffi::CStr, action_directory: &core::ffi::CStr, ) -> std::io::Result<()> { + configure_namespace_with_input_mounts( + mount, + tmp_directory, + root_action_directory, + action_directory, + &[], + ) +} + +/// Like `configure_namespace`, with prepared immutable input trees. Mounts are +/// installed only after making the namespace private, and before binding the +/// private `/tmp` and masking the action root, so sources under either stay +/// reachable. All paths must be allocated and cache entries pinned by the +/// parent. This function is async-signal-safe. +pub fn configure_namespace_with_input_mounts( + mount: bool, + tmp_directory: Option<&core::ffi::CStr>, + root_action_directory: &core::ffi::CStr, + action_directory: &core::ffi::CStr, + input_mounts: &[ReadOnlyBindMount], +) -> std::io::Result<()> { + if !mount && !input_mounts.is_empty() { + return Err(Error::from_raw_os_error(libc::EINVAL)); + } // SAFETY: It is always safe to call geteuid on Posix. let uid = unsafe { libc::geteuid() }; // SAFETY: It is always safe to call getegid on Posix. @@ -626,7 +747,12 @@ pub fn configure_namespace( // Configure the mount namespace if enabled. if mount { - perform_remount(tmp_directory, root_action_directory, action_directory)?; + perform_remount( + tmp_directory, + root_action_directory, + action_directory, + input_mounts, + )?; } // Set hostname to "nativelink" to ensure reproducibility. diff --git a/nativelink-worker/src/running_actions_manager.rs b/nativelink-worker/src/running_actions_manager.rs index 34bd7dccd5a..b9d773e5c35 100644 --- a/nativelink-worker/src/running_actions_manager.rs +++ b/nativelink-worker/src/running_actions_manager.rs @@ -68,7 +68,7 @@ use nativelink_util::action_messages::{ }; use nativelink_util::common::{DigestInfo, fs}; use nativelink_util::digest_hasher::{DigestHasher, DigestHasherFunc}; -use nativelink_util::metrics::{record_input_relink, record_prepare_retry}; +use nativelink_util::metrics::{INPUT_MOUNT_METRICS, record_input_relink, record_prepare_retry}; use nativelink_util::metrics_utils::{AsyncCounterWrapper, CounterWithTime}; use nativelink_util::store_trait::{Store, StoreLike, UploadSizeInfo}; use nativelink_util::task::JoinHandleDropGuard; @@ -88,6 +88,7 @@ use tracing::{debug, error, info, trace, warn}; use uuid::Uuid; use crate::buck2_file_capture::Buck2FileCapture; +use crate::input_mounts::InputMounts; use crate::persistent_worker::{ Input as PersistentWorkerInput, PersistentWorkerPool, PoolConfig, WireFormat, WorkRequest, WorkerKey, @@ -870,28 +871,36 @@ pub fn download_to_directory_bounded<'a>( current_directory: &'a str, downloads: &'a Semaphore, ) -> BoxFuture<'a, Result<(), Error>> { - download_to_directory_with_lease( + download_to_directory_with_mounts( cas_store, filesystem_store, digest, current_directory, None, + None, downloads, ) } -fn download_to_directory_with_lease<'a>( +fn download_to_directory_with_mounts<'a>( cas_store: &'a FastSlowStore, filesystem_store: Pin<&'a FilesystemStore>, digest: &'a DigestInfo, current_directory: &'a str, input_lease: Option>, + input_mounts: Option>, downloads: &'a Semaphore, ) -> BoxFuture<'a, Result<(), Error>> { async move { - let (dirs, files, inline_nodes) = - collect_download_links(cas_store, digest, current_directory, input_lease, downloads) - .await?; + let (dirs, files, inline_nodes) = collect_download_links( + cas_store, + digest, + current_directory, + input_lease, + input_mounts, + downloads, + ) + .await?; let (exec_files, plain_files): (Vec<_>, Vec<_>) = files.into_iter().partition(|file| file.executable); @@ -1354,12 +1363,24 @@ fn collect_download_links<'a>( digest: &'a DigestInfo, current_directory: &'a str, input_lease: Option>, + input_mounts: Option>, downloads: &'a Semaphore, ) -> BoxFuture<'a, Result> { async move { if let Some(input_lease) = &input_lease { input_lease.lease_digest(digest); } + if let Some(mounts) = &input_mounts { + let lease = input_lease + .as_deref() + .map(|lease| -> &dyn crate::directory_cache::DigestLease { lease }); + if mounts + .prepare(Path::new(current_directory), *digest, lease) + .await? + { + return Ok((Vec::new(), Vec::new(), Vec::new())); + } + } // The permit covers the fetch alone, never the walk below it: a // parent holding one while its children wait for theirs would // deadlock a tree deeper than the pool. @@ -1424,6 +1445,7 @@ fn collect_download_links<'a>( input_lease.lease_digest(&digest); } let input_lease = input_lease.clone(); + let input_mounts = input_mounts.clone(); futures.push( async move { let (mut sub_dirs, files, inline) = collect_download_links( @@ -1431,6 +1453,7 @@ fn collect_download_links<'a>( &digest, &new_directory_path, input_lease, + input_mounts, downloads, ) .await @@ -1723,12 +1746,13 @@ async fn prepare_action_inputs_with_lease( } // Traditional path (cache disabled or failed) - download_to_directory_with_lease( + download_to_directory_with_mounts( cas_store, filesystem_store, digest, work_directory, input_lease, + None, downloads, ) .await @@ -2405,6 +2429,7 @@ struct RunningActionImplState { // that prevented the action from running, upload failures, timeouts, exc... // but we have (or could have) the action results (like stderr/stdout). error: Option, + input_mounts: Option>, } #[derive(Debug)] @@ -2463,6 +2488,7 @@ impl RunningActionImpl { resource_usage: None, execution_metadata, error: None, + input_mounts: None, }), // Set to true only after the action is inserted into the manager. // The constructor can fail the operation-id uniqueness check @@ -2496,35 +2522,28 @@ impl RunningActionImpl { /// against the same struggling store. The two fetches are retried /// together: a Command read that fails during the same storm would /// otherwise fail the action while the inputs were still being retried. + /// A `prefetched_command`, read earlier to choose the read-only input + /// mounts, is used instead of reading the Command again. async fn fetch_with_retry( &self, filesystem_store: Pin<&FilesystemStore>, deadline: Option, + prefetched_command: Option, ) -> Result { let started = Instant::now(); let mut attempt: u32 = 1; let mut retries = PrepareRetries::new(record_prepare_outcome); loop { - let command_fut = self.metrics().get_proto_command_from_store.wrap(async { - get_and_decode_digest::( - self.running_actions_manager.cas_store.as_ref(), - self.action_info.command_digest.into(), - ) - .await - .err_tip(|| "Converting command_digest to Command") - }); - let inputs_fut = - self.metrics() - .download_to_directory - .wrap(prepare_action_inputs_with_lease( - &self.running_actions_manager.directory_cache, - &self.running_actions_manager.cas_store, - filesystem_store, - &self.action_info.input_root_digest, - &self.work_directory, - self.input_lease.clone(), - &self.running_actions_manager.input_downloads, - )); + let command_fut = async { + match &prefetched_command { + Some(command) => Ok(command.clone()), + None => self.fetch_command().await, + } + }; + let inputs_fut = self + .metrics() + .download_to_directory + .wrap(self.download_inputs(filesystem_store)); let err = match try_join(command_fut, inputs_fut).await { Ok((command, ())) => { if attempt > 1 { @@ -2588,6 +2607,147 @@ impl RunningActionImpl { .err_tip(|| format!("Error recreating work directory {}", self.work_directory)) } + /// Fetches and decodes the action's Command. + async fn fetch_command(&self) -> Result { + self.metrics() + .get_proto_command_from_store + .wrap(async { + get_and_decode_digest::( + self.running_actions_manager.cas_store.as_ref(), + self.action_info.command_digest.into(), + ) + .await + .err_tip(|| "Converting command_digest to Command") + }) + .await + } + + /// Chooses the read-only input mounts for this action and keeps them in + /// its state, where their cache pins are held through cleanup. Whether + /// mounts apply depends on the Command's output paths, so with mounts + /// configured the Command is read here and returned for the fetch to + /// reuse; without them nothing is read, and the Command and inputs are + /// fetched concurrently. A failure here only means ordinary input setup: + /// the fetch that follows reads the Command and inputs with its retries. + async fn select_input_mounts(&self) -> Option { + let manager = &self.running_actions_manager; + #[cfg(target_os = "linux")] + let mount_paths = &manager.readonly_input_mounts; + #[cfg(not(target_os = "linux"))] + let mount_paths: &[String] = &[]; + if mount_paths.is_empty() { + return None; + } + let command = if action_supports_persistent_workers(&self.action_info).is_none() { + match self.fetch_command().await { + Ok(command) => Some(command), + Err(error) => { + warn!( + ?error, + "Reading the Command to choose input mounts failed; using ordinary input setup" + ); + None + } + } + } else { + None + }; + let mut input_mounts = match (&command, &manager.directory_cache) { + (Some(command), Some(cache)) => InputMounts::for_command( + cache.clone(), + mount_paths, + Path::new(&self.work_directory), + command, + ), + _ => None, + }; + if input_mounts.is_some() { + match crate::input_mounts::has_mountable_inputs( + &manager.cas_store, + self.action_info.input_root_digest, + mount_paths, + self.input_lease + .as_deref() + .map(|lease| -> &dyn crate::directory_cache::DigestLease { lease }), + ) + .await + { + Ok(true) => {} + Ok(false) => input_mounts = None, + Err(error) => { + warn!( + ?error, + "Checking for mountable inputs failed; using ordinary input setup" + ); + input_mounts = None; + } + } + } + if input_mounts.is_none() { + INPUT_MOUNT_METRICS.fallbacks.add(1, &[]); + } + self.state.lock().input_mounts.clone_from(&input_mounts); + command + } + + /// Materializes the input tree into the work directory: through the + /// read-only input mounts chosen for this action, which leave a mount + /// point for each selected subtree, or else through the directory cache + /// when one is configured. If preparing the mounts fails, they are + /// dropped and the tree is downloaded without them; later attempts of + /// the fetch then take the ordinary path. + async fn download_inputs(&self, filesystem_store: Pin<&FilesystemStore>) -> Result<(), Error> { + let manager = &self.running_actions_manager; + let downloads = &manager.input_downloads; + let input_mounts = self.state.lock().input_mounts.clone(); + if input_mounts.is_none() { + return prepare_action_inputs_with_lease( + &manager.directory_cache, + &manager.cas_store, + filesystem_store, + &self.action_info.input_root_digest, + &self.work_directory, + self.input_lease.clone(), + downloads, + ) + .await; + } + let result = download_to_directory_with_mounts( + &manager.cas_store, + filesystem_store, + &self.action_info.input_root_digest, + &self.work_directory, + self.input_lease.clone(), + input_mounts, + downloads, + ) + .await; + if let Err(error) = result { + INPUT_MOUNT_METRICS.prepare_failures.add(1, &[]); + INPUT_MOUNT_METRICS.fallbacks.add(1, &[]); + warn!( + ?error, + "Input mount preparation failed; retrying without mounts" + ); + // No child has started. Discard partial inputs before + // releasing pins and retrying the ordinary download once. + fs::remove_dir_all(&self.work_directory).await?; + self.state.lock().input_mounts.take(); + fs::create_dir(&self.work_directory).await?; + download_to_directory_with_mounts( + &manager.cas_store, + filesystem_store, + &self.action_info.input_root_digest, + &self.work_directory, + self.input_lease.clone(), + None, + downloads, + ) + .await?; + } + Ok(()) + } + async fn inner_prepare_action( self: Arc, download_deadline: Option, @@ -2638,6 +2798,7 @@ impl RunningActionImpl { let command = { let filesystem_store_pin = Pin::new(self.running_actions_manager.filesystem_store.as_ref()); + let prefetched_command = self.select_input_mounts().await; fs::create_dir(&self.work_directory) .await .err_tip(|| format!("Error creating work directory {}", self.work_directory))?; @@ -2645,7 +2806,7 @@ impl RunningActionImpl { self.did_cleanup.store(false, Ordering::Release); // Fetch and decode the Command and download the input tree into // the work directory, retried together after a store failure. - self.fetch_with_retry(filesystem_store_pin, download_deadline) + self.fetch_with_retry(filesystem_store_pin, download_deadline, prefetched_command) .await .map_err(tag_missing_input)? }; @@ -2701,6 +2862,12 @@ impl RunningActionImpl { arg: &OsStr, working_directory: &String, ) -> Result { + #[cfg(target_os = "linux")] + if let Some(mounts) = self.state.lock().input_mounts.as_ref() + && let Some(path) = mounts.executable_path(working_directory, arg)? + { + return Ok(path); + } // If the program contains a slash, we treat it as a path and resolve it relative to the work directory. Ok(if Path::new(arg).components().count() > 1 { let canonical_path = PathBuf::from(&self.work_directory) @@ -3136,15 +3303,22 @@ impl RunningActionImpl { None }; + let input_mounts = self.state.lock().input_mounts.clone(); + let bind_mounts = input_mounts + .as_ref() + .map(|mounts| mounts.bind_mounts()) + .transpose()? + .unwrap_or_default(); // SAFETY: This function is specifically designed to operate in a async-signal-safe // environment. unsafe { command_builder.pre_exec(move || { - crate::namespace_utils::configure_namespace( + crate::namespace_utils::configure_namespace_with_input_mounts( mount, tmp_directory.as_deref(), &root_action_directory, &action_directory, + &bind_mounts, ) }); } @@ -4038,9 +4212,11 @@ impl Drop for RunningActionImpl { let running_actions_manager = self.running_actions_manager.clone(); let action_directory = self.action_directory.clone(); let input_lease = self.input_lease.clone(); + let input_mounts = self.state.get_mut().input_mounts.take(); background_spawn!("running_action_impl_drop", async move { let cleanup_result = do_cleanup(&running_actions_manager, &operation_id, &action_directory).await; + drop(input_mounts); if let Some(input_lease) = input_lease { input_lease.release().await; } @@ -4227,6 +4403,7 @@ impl RunningAction for RunningActionImpl { .await; self.has_manager_entry.store(false, Ordering::Release); self.did_cleanup.store(true, Ordering::Release); + self.state.lock().input_mounts.take(); // The work directory and manager entry are gone before the // lease is released. If this future is cancelled while the // release task is finishing, Drop still observes a completed @@ -4715,6 +4892,8 @@ pub struct RunningActionsManagerArgs<'a> { /// exceed their configured `max_bytes` / `max_count` eviction limits. pub active_input_leases: bool, #[cfg(target_os = "linux")] + pub readonly_input_mounts: Vec, + #[cfg(target_os = "linux")] pub use_namespaces: UseNamespaces, } @@ -4785,6 +4964,8 @@ pub struct RunningActionsManagerImpl { /// actions this worker stages; see /// `ExecutionConfiguration::max_concurrent_input_downloads`. input_downloads: Semaphore, + #[cfg(target_os = "linux")] + readonly_input_mounts: Vec, persistent_worker_pool: PersistentWorkerPool, /// Retires idle persistent workers; ends with the manager. _persistent_worker_sweeper: JoinHandleDropGuard<()>, @@ -4802,6 +4983,19 @@ impl RunningActionsManagerImpl { args: RunningActionsManagerArgs<'_>, callbacks: Callbacks, ) -> Result { + #[cfg(target_os = "linux")] + { + crate::input_mounts::validate_paths(&args.readonly_input_mounts)?; + if !args.readonly_input_mounts.is_empty() + && (!matches!(args.use_namespaces, UseNamespaces::YesAndMount { .. }) + || args.directory_cache.is_none()) + { + return Err(make_err!( + Code::InvalidArgument, + "Read-only input mounts require Linux mount namespaces and a directory cache" + )); + } + } // Sadly because of some limitations of how Any works we need to clone more times than optimal. let filesystem_store = args .cas_store @@ -4853,6 +5047,8 @@ impl RunningActionsManagerImpl { cleanup_complete_notify: Arc::new(Notify::new()), directory_cache: args.directory_cache, active_input_leases: args.active_input_leases, + #[cfg(target_os = "linux")] + readonly_input_mounts: args.readonly_input_mounts, persistent_worker_pool, _persistent_worker_sweeper: persistent_worker_sweeper, buck2_captures: Mutex::new(HashMap::new()), diff --git a/nativelink-worker/tests/directory_cache_test.rs b/nativelink-worker/tests/directory_cache_test.rs index b26025a558a..feee628ff80 100644 --- a/nativelink-worker/tests/directory_cache_test.rs +++ b/nativelink-worker/tests/directory_cache_test.rs @@ -1645,3 +1645,178 @@ async fn get_tree_prefetch_follows_server_pagination() -> Result<(), Error> { Ok(()) } + +#[nativelink_test] +async fn prepared_mount_tree_is_shared_and_pinned_until_release() -> Result<(), Error> { + let slow = MemoryStore::new(&MemorySpec::default()); + let cache = Arc::new( + DirectoryCache::new( + DirectoryCacheConfig { + max_entries: 1, + cache_root: make_temp_path("mount_cache").into(), + ..Default::default() + }, + make_cas_store(slow.clone()).await, + ) + .await?, + ); + let mut digests = Vec::new(); + for name in ["sdk-a", "sdk-b", "sdk-c"] { + let directory = ProtoDirectory { + directories: vec![DirectoryNode { + name: name.to_owned(), + digest: Some(proto_digest(&ProtoDirectory::default()).into()), + }], + ..Default::default() + }; + let digest = proto_digest(&directory); + slow.update_oneshot(digest, directory.encode_to_vec().into()) + .await?; + digests.push(digest); + } + let empty = ProtoDirectory::default(); + slow.update_oneshot(proto_digest(&empty), empty.encode_to_vec().into()) + .await?; + let mut handles = + futures::future::try_join_all((0..16).map(|_| cache.prepare_for_mount(digests[0], None))) + .await?; + let path = handles[0].path().to_owned(); + assert!(handles.iter().all(|handle| handle.path() == path)); + assert!(path.join("sdk-a").is_dir()); + let other = cache.prepare_for_mount(digests[1], None).await?; + assert!( + path.join("sdk-a").is_dir(), + "active mount source was evicted" + ); + // Even one remaining action must protect the shared source. + let last = handles.pop().unwrap(); + drop(handles); + assert!(path.exists()); + drop(last); + let third = cache.prepare_for_mount(digests[2], None).await?; + // Eviction renames the old source under the cache lock; deletion may follow + // asynchronously. The other two still-pinned trees must remain accessible. + assert!(!path.exists()); + assert!(other.path().join("sdk-b").is_dir()); + assert!(third.path().join("sdk-c").is_dir()); + Ok(()) +} + +#[nativelink_test] +async fn readonly_mount_paths_and_output_fallback() -> Result<(), Error> { + use nativelink_proto::build::bazel::remote::execution::v2::Command; + use nativelink_worker::input_mounts::{InputMounts, validate_paths}; + let paths = vec!["out/x/sdk".to_owned()]; + validate_paths(&paths)?; + for invalid in [ + "", "/sdk", "../sdk", "sdk/../x", "sdk//x", "./sdk", "sdk/", "sdk\\x", "sdk\0", + ] { + assert!( + validate_paths(&[invalid.to_owned()]).is_err(), + "{invalid:?}" + ); + } + assert!(validate_paths(&["sdk".into(), "sdk/headers".into()]).is_err()); + assert!(validate_paths(&["sdk".into(), "sdk".into()]).is_err()); + validate_paths(&["sdk".into(), "sdk2".into()])?; + let cache = Arc::new( + DirectoryCache::new( + DirectoryCacheConfig { + cache_root: make_temp_path("mount_fallback").into(), + ..Default::default() + }, + make_cas_store(MemoryStore::new(&MemorySpec::default())).await, + ) + .await?, + ); + let eligible = |command: Command| { + InputMounts::for_command(cache.clone(), &paths, Path::new("/action"), &command).is_some() + }; + assert!(eligible(Command { + working_directory: "out/x".into(), + output_paths: vec!["obj/a.o".into(), "../clang-crashreports".into()], + ..Default::default() + })); + for output in [ + "sdk", + "sdk/generated.h", + "", + "obj/../sdk/file", + "../../../escape", + "/absolute", + ] { + assert!( + !eligible(Command { + working_directory: "out/x".into(), + output_paths: vec![output.into()], + ..Default::default() + }), + "{output}" + ); + } + assert!(!eligible(Command { + output_directories: vec!["out".into()], + ..Default::default() + })); + assert!(!eligible(Command { + output_files: vec!["out/x/sdk/file".into()], + ..Default::default() + })); + assert!(!eligible(Command { + working_directory: "out/x/sdk/include".into(), + ..Default::default() + })); + Ok(()) +} + +#[nativelink_test] +async fn mount_selection_requires_an_input_directory() -> Result<(), Error> { + use nativelink_worker::input_mounts::has_mountable_inputs; + + let store = MemoryStore::new(&MemorySpec::default()); + // The selected SDK's contents are deliberately absent: selection should + // read its ancestors, not download the SDK or an unrelated subtree. + let sdk = ProtoDirectory { + directories: vec![DirectoryNode { + name: "headers".into(), + digest: Some(proto_digest(&ProtoDirectory::default()).into()), + }], + ..Default::default() + }; + let branch = ProtoDirectory { + directories: vec![DirectoryNode { + name: "sdk".into(), + digest: Some(proto_digest(&sdk).into()), + }], + ..Default::default() + }; + let root = ProtoDirectory { + directories: vec![ + branch.directories[0].clone(), + DirectoryNode { + name: "out".into(), + digest: Some(proto_digest(&branch).into()), + }, + ], + symlinks: vec![SymlinkNode { + name: "alias".into(), + target: "sdk".into(), + ..Default::default() + }], + ..Default::default() + }; + let digest = proto_digest(&root); + store + .update_oneshot(digest, root.encode_to_vec().into()) + .await?; + store + .update_oneshot(proto_digest(&branch), branch.encode_to_vec().into()) + .await?; + let cas = make_cas_store(store).await; + assert!(has_mountable_inputs(&cas, digest, &["out/sdk".into()], None).await?); + assert!(has_mountable_inputs(&cas, digest, &["sdk".into()], None).await?); + assert!(has_mountable_inputs(&cas, digest, &["absent".into(), "sdk".into()], None).await?); + assert!(!has_mountable_inputs(&cas, digest, &["out/x/sdk".into()], None).await?); + assert!(!has_mountable_inputs(&cas, digest, &["alias".into()], None).await?); + Ok(()) +} diff --git a/nativelink-worker/tests/namespace_utils_test.rs b/nativelink-worker/tests/namespace_utils_test.rs index 5648f7f8460..2b37d2d0a52 100644 --- a/nativelink-worker/tests/namespace_utils_test.rs +++ b/nativelink-worker/tests/namespace_utils_test.rs @@ -611,3 +611,146 @@ async fn test_maybe_namespaced_child_try_wait() -> Result<(), Error> { Ok(()) } + +#[nativelink_test] +async fn readonly_input_mount_is_private_and_source_survives_masking() -> Result<(), Error> { + if !namespace_utils::namespaces_supported(true, false) { + assert!( + std::env::var_os("NATIVELINK_REQUIRE_MOUNT_TESTS").is_none(), + "mount namespaces required by NATIVELINK_REQUIRE_MOUNT_TESTS" + ); + eprintln!("SKIP: mount namespaces unavailable"); + return Ok(()); + } + let root = std::path::PathBuf::from(nativelink_util::common::make_temp_path("readonly_mount")); + let action = root.join("action"); + let target = action.join("sdk"); + // Intentionally put the source under the root that configure_namespace + // masks. Inputs must be mounted before the root is masked. + let source = root.join("cache"); + std::fs::create_dir_all(&target)?; + std::fs::create_dir_all(&source)?; + std::fs::write(source.join("header.h"), "sdk-content")?; + std::os::unix::fs::symlink("header.h", source.join("alias.h"))?; + let cwd = std::env::current_dir()?; + let relative_source = pathdiff::diff_paths(&source, &cwd).unwrap(); + let relative_target = pathdiff::diff_paths(&target, &cwd).unwrap(); + let mounts = vec![namespace_utils::ReadOnlyBindMount::new( + &relative_source, + &relative_target, + )?]; + let root_c = CString::new(root.as_os_str().as_encoded_bytes()).unwrap(); + let action_c = CString::new(action.as_os_str().as_encoded_bytes()).unwrap(); + let mut command = Command::new("sh"); + command.current_dir(&action); + command.arg("-c").arg( + "test \"$(cat \"$1/alias.h\")\" = sdk-content && ! touch \"$1/new.h\" && ! rm \"$1/header.h\"" + ).arg("sh").arg(&target); + // SAFETY: namespace setup and prepared bind mounts use only operations + // designed for the pre-exec child; no parent locks are held. + unsafe { + command.pre_exec(move || { + namespace_utils::configure_namespace_with_input_mounts( + true, None, &root_c, &action_c, &mounts, + ) + }); + } + let output = command.output()?; + assert!(output.status.success(), "{output:?}"); + assert!( + std::fs::read_dir(&target)?.next().is_none(), + "mount escaped into parent" + ); + assert_eq!( + std::fs::read_to_string(source.join("header.h"))?, + "sdk-content" + ); + assert!(!source.join("new.h").exists()); + + // A failed input mount must abort spawning, never run the command against + // an empty placeholder and accidentally cache an incorrect action result. + let mounts = vec![namespace_utils::ReadOnlyBindMount::new( + &source, + &target.join("missing"), + )?]; + let root_c = CString::new(root.to_str().unwrap())?; + let action_c = CString::new(action.to_str().unwrap())?; + let mut command = Command::new("sh"); + command.args(["-c", "exit 0"]); + // SAFETY: the pre-exec hook only uses prepared paths and namespace syscalls. + unsafe { + command.pre_exec(move || { + namespace_utils::configure_namespace_with_input_mounts( + true, None, &root_c, &action_c, &mounts, + ) + }); + } + assert!(command.spawn().is_err()); + std::fs::remove_dir_all(root)?; + Ok(()) +} + +#[nativelink_test] +async fn readonly_input_mount_survives_private_tmp() -> Result<(), Error> { + if !namespace_utils::namespaces_supported(true, true) { + assert!( + std::env::var_os("NATIVELINK_REQUIRE_MOUNT_TESTS").is_none(), + "mount namespaces required by NATIVELINK_REQUIRE_MOUNT_TESTS" + ); + eprintln!("SKIP: mount namespaces with a private /tmp unavailable"); + return Ok(()); + } + // A worker whose root, and with it the directory cache, lives under + // /tmp: the private /tmp hides the cache, so inputs must be mounted + // before it is bound and still be reachable afterwards. + let root = unique_path(Path::new("/tmp"), "readonly_mount_tmp"); + let action = root.join("action"); + let work = action.join("work"); + let tmp = action.join("tmp"); + let target = work.join("sdk"); + let source = root.join("cache"); + std::fs::create_dir_all(&target)?; + std::fs::create_dir_all(&tmp)?; + std::fs::create_dir_all(&source)?; + std::fs::write(source.join("header.h"), "sdk-content")?; + let mounts = vec![namespace_utils::ReadOnlyBindMount::new(&source, &target)?]; + let root_c = CString::new(root.to_str().unwrap())?; + let action_c = CString::new(action.to_str().unwrap())?; + let tmp_c = CString::new(tmp.to_str().unwrap())?; + let mut command = Command::new("sh"); + command.current_dir(&work); + command + .arg("-c") + .arg( + "test \"$(cat sdk/header.h)\" = sdk-content \ + && test \"$(cat \"$1/header.h\")\" = sdk-content \ + && ! touch sdk/new.h \ + && test ! -e \"$2\" \ + && echo scratch > /tmp/scratch", + ) + .arg("sh") + .arg(&target) + .arg(&source); + // SAFETY: the pre-exec hook only uses prepared paths and namespace syscalls. + unsafe { + command.pre_exec(move || { + namespace_utils::configure_namespace_with_input_mounts( + true, + Some(&tmp_c), + &root_c, + &action_c, + &mounts, + ) + }); + } + let output = command.output()?; + assert!(output.status.success(), "{output:?}"); + assert_eq!( + std::fs::read_to_string(tmp.join("scratch"))?, + "scratch\n", + "/tmp in the action must be its own tmp directory" + ); + assert!(!source.join("new.h").exists()); + std::fs::remove_dir_all(root)?; + Ok(()) +} diff --git a/nativelink-worker/tests/running_actions_manager_test.rs b/nativelink-worker/tests/running_actions_manager_test.rs index 37e5d017b6f..e1d54373fe6 100644 --- a/nativelink-worker/tests/running_actions_manager_test.rs +++ b/nativelink-worker/tests/running_actions_manager_test.rs @@ -209,6 +209,276 @@ mod tests { .await } + #[cfg(target_os = "linux")] + #[nativelink_test] + async fn readonly_inputs_reuse_subtrees_and_fall_back_for_outputs() -> Result<(), Error> { + if !namespace_utils::namespaces_supported(true, true) { + assert!( + env::var_os("NATIVELINK_REQUIRE_MOUNT_TESTS").is_none(), + "mount namespaces required by NATIVELINK_REQUIRE_MOUNT_TESTS" + ); + eprintln!("SKIP: mount namespaces unavailable"); + return Ok(()); + } + let (_fast, _slow, cas, _ac) = setup_stores().await?; + let root = make_temp_path("readonly_actions"); + fs::create_dir_all(&root).await?; + let cache = Arc::new( + DirectoryCache::new( + DirectoryCacheConfig { + cache_root: PathBuf::from(&root).join("cache"), + ..Default::default() + }, + cas.clone(), + ) + .await?, + ); + let manager = Arc::new(RunningActionsManagerImpl::new(RunningActionsManagerArgs { + root_action_directory: root.clone(), + execution_configuration: ExecutionConfiguration::default(), + cas_store: cas.clone(), + ac_store: None, + historical_store: Store::new(cas.clone()), + upload_action_result_config: &UploadActionResultConfig { + upload_ac_results_strategy: UploadCacheResultsStrategy::Never, + ..Default::default() + }, + max_action_timeout: Duration::from_secs(30), + max_upload_timeout: Duration::from_secs(30), + max_download_timeout: Duration::from_secs(30), + max_cleanup_wait: Duration::from_secs(30), + max_cleanup_backoff: Duration::from_millis(10), + timeout_handled_externally: false, + active_input_leases: true, + readonly_input_mounts: vec!["sdk".into()], + directory_cache: Some(cache.clone()), + // The private /tmp is the default with a mount namespace; with + // the test root under /tmp it also hides the cache sources. + use_namespaces: + nativelink_worker::running_actions_manager::UseNamespaces::YesAndMount { + isolate_tmp: true, + }, + })?); + let tool = Bytes::from_static(b"#!/bin/sh\nif [ $# -gt 0 ]; then printf sdk-content > \"$1\"; else printf sdk-content; fi"); + let tool_digest = compute_buf_digest(&tool, &mut DigestHasherFunc::Sha256.hasher()); + cas.as_ref().update_oneshot(tool_digest, tool).await?; + let sdk_digest = serialize_and_upload_message( + &Directory { + files: vec![FileNode { + name: "tool".into(), + digest: Some(tool_digest.into()), + is_executable: true, + ..Default::default() + }], + symlinks: vec![ + SymlinkNode { + name: "alias".into(), + target: "tool".into(), + ..Default::default() + }, + SymlinkNode { + name: "escape".into(), + target: "/bin/sh".into(), + ..Default::default() + }, + ], + ..Default::default() + }, + cas.as_pin(), + &mut DigestHasherFunc::Sha256.hasher(), + ) + .await?; + let empty_digest = serialize_and_upload_message( + &Directory::default(), + cas.as_pin(), + &mut DigestHasherFunc::Sha256.hasher(), + ) + .await?; + for i in 0..8 { + let sdk_digest = if i == 7 { + // Force a new SDK entry to fail construction while CAS remains usable. + let directory = Directory { + files: vec![FileNode { + name: "tool".into(), + digest: Some(tool_digest.into()), + is_executable: true, + ..Default::default() + }], + symlinks: vec![SymlinkNode { + name: "alias".into(), + target: "./tool".into(), + ..Default::default() + }], + ..Default::default() + }; + fs::remove_dir_all(PathBuf::from(&root).join("cache")).await?; + tokio::fs::write(PathBuf::from(&root).join("cache"), b"blocked").await?; + serialize_and_upload_message( + &directory, + cas.as_pin(), + &mut DigestHasherFunc::Sha256.hasher(), + ) + .await? + } else { + sdk_digest + }; + // Change an unrelated input path, keeping the SDK digest stable. + let input_digest = serialize_and_upload_message( + &Directory { + directories: vec![ + DirectoryNode { + name: format!("other{i}"), + digest: Some(empty_digest.into()), + }, + DirectoryNode { + name: "sdk".into(), + digest: Some(sdk_digest.into()), + }, + ], + ..Default::default() + }, + cas.as_pin(), + &mut DigestHasherFunc::Sha256.hasher(), + ) + .await?; + let output_path = if i == 2 { "sdk/result" } else { "result" }; + let script = if i == 2 { + "./sdk/alias > sdk/result" + } else if i == 7 { + "./sdk/alias > result" + } else { + "./sdk/alias > result && ! touch sdk/forbidden" + }; + let command_digest = serialize_and_upload_message( + &Command { + arguments: if (4..=6).contains(&i) { + // Execute the mounted tool directly, including from a nested cwd. + let program = match i { + 5 => "../sdk/alias", + 6 => "./sdk/escape", + _ => "./sdk/alias", + }; + vec![program.into(), "result".into()] + } else { + vec!["/bin/sh".into(), "-c".into(), script.into()] + }, + working_directory: if i == 5 { + format!("other{i}") + } else { + String::new() + }, + environment_variables: vec![EnvironmentVariable { + name: "PATH".into(), + value: env::var("PATH").unwrap(), + }], + output_paths: vec![output_path.into()], + ..Default::default() + }, + cas.as_pin(), + &mut DigestHasherFunc::Sha256.hasher(), + ) + .await?; + let action_digest = serialize_and_upload_message( + &Action { + command_digest: Some(command_digest.into()), + input_root_digest: Some(input_digest.into()), + ..Default::default() + }, + cas.as_pin(), + &mut DigestHasherFunc::Sha256.hasher(), + ) + .await?; + let action = manager + .create_and_add_action( + "test-worker".into(), + StartExecute { + execute_request: Some(ExecuteRequest { + action_digest: Some(action_digest.into()), + digest_function: ProtoDigestFunction::Sha256.into(), + ..Default::default() + }), + operation_id: OperationId::default().to_string(), + ..Default::default() + }, + ) + .await? + .prepare_action() + .await?; + if i == 7 { + assert!( + PathBuf::from(action.get_work_directory()) + .join("sdk/tool") + .is_file() + ); + assert_eq!(cache.stats().await.in_use_entries, 0); + } + if i < 2 { + assert!( + fs::read_dir(format!("{}/sdk", action.get_work_directory())) + .await? + .as_mut() + .next_entry() + .await? + .is_none(), + "parent workspace must contain only a mount point" + ); + assert_eq!( + cache.stats().await.entries, + 1, + "changed roots must reuse the SDK entry" + ); + assert_eq!(cache.stats().await.in_use_entries, 1); + } + if i == 3 { + // An abandoned prepared action must keep its source pinned + // until background cleanup removes the workspace, then release it. + let work_directory = action.get_work_directory().clone(); + assert_eq!(cache.stats().await.in_use_entries, 1); + drop(action); + tokio::time::timeout(Duration::from_secs(5), async { + while cache.stats().await.in_use_entries != 0 { + tokio::time::sleep(Duration::from_millis(1)).await; + } + }) + .await?; + assert!(!PathBuf::from(&work_directory).exists()); + continue; + } + if i == 6 { + let error = + action.clone().execute().await.expect_err( + "an executable symlink escaping the mounted tree must be rejected", + ); + assert_eq!(error.code, Code::InvalidArgument); + action.cleanup().await?; + continue; + } + let result = action + .clone() + .execute() + .await? + .upload_results() + .await? + .get_finished_result() + .await?; + assert_eq!(result.exit_code, 0); + assert_eq!(result.output_files.len(), 1); + let output = cas + .as_ref() + .get_part_unchunked(result.output_files[0].digest, 0, None) + .await?; + assert_eq!(output.as_ref(), b"sdk-content"); + action.cleanup().await?; + assert_eq!( + cache.stats().await.in_use_entries, + 0, + "cleanup must release the tree pin" + ); + } + fs::remove_dir_all(root).await?; + Ok(()) + } + const NOW_TIME: u64 = 10000; fn make_system_time(add_time: u64) -> SystemTime { @@ -405,6 +675,8 @@ mod tests { max_cleanup_backoff: Duration::from_millis(DEFAULT_MAX_CLEANUP_BACKOFF), timeout_handled_externally: false, active_input_leases: true, + #[cfg(target_os = "linux")] + readonly_input_mounts: Vec::new(), directory_cache: Some(directory_cache), #[cfg(target_os = "linux")] use_namespaces: use_namespaces(), @@ -566,6 +838,8 @@ mod tests { max_cleanup_backoff: Duration::from_millis(DEFAULT_MAX_CLEANUP_BACKOFF), timeout_handled_externally: false, active_input_leases: false, + #[cfg(target_os = "linux")] + readonly_input_mounts: Vec::new(), directory_cache: Some(directory_cache), #[cfg(target_os = "linux")] use_namespaces: use_namespaces(), @@ -1366,6 +1640,8 @@ mod tests { max_cleanup_backoff: Duration::from_millis(DEFAULT_MAX_CLEANUP_BACKOFF), timeout_handled_externally: false, active_input_leases: false, + #[cfg(target_os = "linux")] + readonly_input_mounts: Vec::new(), directory_cache: None, #[cfg(target_os = "linux")] use_namespaces: use_namespaces(), @@ -1496,6 +1772,8 @@ mod tests { max_cleanup_backoff: Duration::from_millis(DEFAULT_MAX_CLEANUP_BACKOFF), timeout_handled_externally: false, active_input_leases: false, + #[cfg(target_os = "linux")] + readonly_input_mounts: Vec::new(), directory_cache: None, #[cfg(target_os = "linux")] use_namespaces: use_namespaces(), @@ -1628,6 +1906,8 @@ mod tests { max_cleanup_backoff: Duration::from_millis(DEFAULT_MAX_CLEANUP_BACKOFF), timeout_handled_externally: false, active_input_leases: false, + #[cfg(target_os = "linux")] + readonly_input_mounts: Vec::new(), directory_cache: None, #[cfg(target_os = "linux")] use_namespaces: use_namespaces(), @@ -1815,6 +2095,8 @@ mod tests { max_cleanup_backoff: Duration::from_millis(DEFAULT_MAX_CLEANUP_BACKOFF), timeout_handled_externally: false, active_input_leases: false, + #[cfg(target_os = "linux")] + readonly_input_mounts: Vec::new(), directory_cache: None, #[cfg(target_os = "linux")] use_namespaces: use_namespaces(), @@ -2004,6 +2286,8 @@ mod tests { max_cleanup_backoff: Duration::from_millis(DEFAULT_MAX_CLEANUP_BACKOFF), timeout_handled_externally: false, active_input_leases: false, + #[cfg(target_os = "linux")] + readonly_input_mounts: Vec::new(), directory_cache: None, #[cfg(target_os = "linux")] use_namespaces: use_namespaces(), @@ -2262,6 +2546,8 @@ mod tests { max_cleanup_backoff: Duration::from_millis(DEFAULT_MAX_CLEANUP_BACKOFF), timeout_handled_externally: false, active_input_leases: false, + #[cfg(target_os = "linux")] + readonly_input_mounts: Vec::new(), directory_cache: None, #[cfg(target_os = "linux")] use_namespaces: use_namespaces(), @@ -2418,6 +2704,8 @@ mod tests { max_cleanup_backoff: Duration::from_millis(DEFAULT_MAX_CLEANUP_BACKOFF), timeout_handled_externally: false, active_input_leases: false, + #[cfg(target_os = "linux")] + readonly_input_mounts: Vec::new(), directory_cache: None, #[cfg(target_os = "linux")] use_namespaces: use_namespaces(), @@ -2565,6 +2853,8 @@ mod tests { max_cleanup_backoff: Duration::from_millis(DEFAULT_MAX_CLEANUP_BACKOFF), timeout_handled_externally: false, active_input_leases: false, + #[cfg(target_os = "linux")] + readonly_input_mounts: Vec::new(), directory_cache: None, #[cfg(target_os = "linux")] use_namespaces: use_namespaces(), @@ -2707,6 +2997,8 @@ mod tests { max_cleanup_backoff: Duration::from_millis(DEFAULT_MAX_CLEANUP_BACKOFF), timeout_handled_externally: false, active_input_leases: false, + #[cfg(target_os = "linux")] + readonly_input_mounts: Vec::new(), directory_cache: None, #[cfg(target_os = "linux")] use_namespaces: use_namespaces(), @@ -2935,6 +3227,8 @@ exit 0 max_cleanup_backoff: Duration::from_millis(DEFAULT_MAX_CLEANUP_BACKOFF), timeout_handled_externally: false, active_input_leases: false, + #[cfg(target_os = "linux")] + readonly_input_mounts: Vec::new(), directory_cache: None, #[cfg(target_os = "linux")] use_namespaces: use_namespaces(), @@ -3125,6 +3419,8 @@ exit 0 max_cleanup_backoff: Duration::from_millis(DEFAULT_MAX_CLEANUP_BACKOFF), timeout_handled_externally: false, active_input_leases: false, + #[cfg(target_os = "linux")] + readonly_input_mounts: Vec::new(), directory_cache: None, #[cfg(target_os = "linux")] use_namespaces: use_namespaces(), @@ -3302,6 +3598,8 @@ exit 0 max_cleanup_backoff: Duration::from_millis(DEFAULT_MAX_CLEANUP_BACKOFF), timeout_handled_externally: false, active_input_leases: false, + #[cfg(target_os = "linux")] + readonly_input_mounts: Vec::new(), directory_cache: None, #[cfg(target_os = "linux")] use_namespaces: use_namespaces(), @@ -3552,6 +3850,8 @@ exit 0 max_cleanup_backoff: Duration::from_millis(DEFAULT_MAX_CLEANUP_BACKOFF), timeout_handled_externally: false, active_input_leases: false, + #[cfg(target_os = "linux")] + readonly_input_mounts: Vec::new(), directory_cache: None, #[cfg(target_os = "linux")] use_namespaces: use_namespaces(), @@ -3649,6 +3949,8 @@ exit 0 max_cleanup_backoff: Duration::from_millis(DEFAULT_MAX_CLEANUP_BACKOFF), timeout_handled_externally: false, active_input_leases: false, + #[cfg(target_os = "linux")] + readonly_input_mounts: Vec::new(), directory_cache: None, // Host pids and a host-visible pid file: the point is what the kill // reaches, not the namespaces. @@ -4043,6 +4345,8 @@ exit 0 max_cleanup_backoff: Duration::from_millis(DEFAULT_MAX_CLEANUP_BACKOFF), timeout_handled_externally: false, active_input_leases: false, + #[cfg(target_os = "linux")] + readonly_input_mounts: Vec::new(), directory_cache: None, #[cfg(target_os = "linux")] use_namespaces: use_namespaces(), @@ -4548,6 +4852,8 @@ exit 0 max_cleanup_backoff: Duration::from_millis(DEFAULT_MAX_CLEANUP_BACKOFF), timeout_handled_externally: false, active_input_leases: false, + #[cfg(target_os = "linux")] + readonly_input_mounts: Vec::new(), directory_cache: None, #[cfg(target_os = "linux")] use_namespaces: use_namespaces(), @@ -5005,6 +5311,8 @@ exit 0 max_cleanup_backoff: Duration::from_millis(DEFAULT_MAX_CLEANUP_BACKOFF), timeout_handled_externally: false, active_input_leases: false, + #[cfg(target_os = "linux")] + readonly_input_mounts: Vec::new(), directory_cache: None, #[cfg(target_os = "linux")] use_namespaces: use_namespaces(), @@ -5144,6 +5452,8 @@ exit 0 max_cleanup_backoff: Duration::from_millis(DEFAULT_MAX_CLEANUP_BACKOFF), timeout_handled_externally: false, active_input_leases: false, + #[cfg(target_os = "linux")] + readonly_input_mounts: Vec::new(), directory_cache: None, #[cfg(target_os = "linux")] use_namespaces: use_namespaces(), @@ -5325,6 +5635,8 @@ exit 0 max_cleanup_backoff: Duration::from_millis(DEFAULT_MAX_CLEANUP_BACKOFF), timeout_handled_externally: false, active_input_leases: false, + #[cfg(target_os = "linux")] + readonly_input_mounts: Vec::new(), directory_cache: None, // The namespaced stub turns SIGTERM into SIGKILL for its child, // so the grace period is only observable without it. @@ -5469,6 +5781,8 @@ exit 0 max_cleanup_backoff: Duration::from_millis(DEFAULT_MAX_CLEANUP_BACKOFF), timeout_handled_externally: false, active_input_leases: false, + #[cfg(target_os = "linux")] + readonly_input_mounts: Vec::new(), directory_cache: None, #[cfg(target_os = "linux")] use_namespaces: use_namespaces(), @@ -5579,6 +5893,8 @@ exit 0 max_cleanup_backoff: Duration::from_millis(DEFAULT_MAX_CLEANUP_BACKOFF), timeout_handled_externally: false, active_input_leases: false, + #[cfg(target_os = "linux")] + readonly_input_mounts: Vec::new(), directory_cache: None, #[cfg(target_os = "linux")] use_namespaces: use_namespaces(), @@ -5737,6 +6053,8 @@ exit 0 max_cleanup_backoff: Duration::from_millis(DEFAULT_MAX_CLEANUP_BACKOFF), timeout_handled_externally: false, active_input_leases: false, + #[cfg(target_os = "linux")] + readonly_input_mounts: Vec::new(), directory_cache: None, #[cfg(target_os = "linux")] use_namespaces: use_namespaces(), @@ -5859,6 +6177,8 @@ exit 0 max_cleanup_backoff: Duration::from_millis(DEFAULT_MAX_CLEANUP_BACKOFF), timeout_handled_externally: false, active_input_leases: false, + #[cfg(target_os = "linux")] + readonly_input_mounts: Vec::new(), directory_cache: None, #[cfg(target_os = "linux")] use_namespaces: use_namespaces(), @@ -5930,6 +6250,8 @@ exit 0 max_cleanup_backoff: Duration::from_millis(DEFAULT_MAX_CLEANUP_BACKOFF), timeout_handled_externally: false, active_input_leases: false, + #[cfg(target_os = "linux")] + readonly_input_mounts: Vec::new(), directory_cache: None, #[cfg(target_os = "linux")] use_namespaces: use_namespaces(), @@ -6094,6 +6416,8 @@ exit 0 max_cleanup_backoff: Duration::from_millis(DEFAULT_MAX_CLEANUP_BACKOFF), timeout_handled_externally: false, active_input_leases: false, + #[cfg(target_os = "linux")] + readonly_input_mounts: Vec::new(), directory_cache: None, #[cfg(target_os = "linux")] use_namespaces: use_namespaces(), @@ -6145,6 +6469,8 @@ exit 0 max_cleanup_backoff: Duration::from_millis(DEFAULT_MAX_CLEANUP_BACKOFF), timeout_handled_externally: false, active_input_leases: false, + #[cfg(target_os = "linux")] + readonly_input_mounts: Vec::new(), directory_cache: None, #[cfg(target_os = "linux")] use_namespaces: use_namespaces(), @@ -6213,6 +6539,8 @@ exit 0 max_cleanup_backoff: Duration::from_millis(DEFAULT_MAX_CLEANUP_BACKOFF), timeout_handled_externally: false, active_input_leases: false, + #[cfg(target_os = "linux")] + readonly_input_mounts: Vec::new(), directory_cache: None, #[cfg(target_os = "linux")] use_namespaces: use_namespaces(), @@ -6400,6 +6728,8 @@ exit 1 max_cleanup_backoff: Duration::from_millis(DEFAULT_MAX_CLEANUP_BACKOFF), timeout_handled_externally: false, active_input_leases: false, + #[cfg(target_os = "linux")] + readonly_input_mounts: Vec::new(), directory_cache: None, #[cfg(target_os = "linux")] use_namespaces: use_namespaces(), @@ -6494,6 +6824,8 @@ exit 1 max_cleanup_backoff: Duration::from_millis(DEFAULT_MAX_CLEANUP_BACKOFF), timeout_handled_externally: false, active_input_leases: false, + #[cfg(target_os = "linux")] + readonly_input_mounts: Vec::new(), directory_cache: None, #[cfg(target_os = "linux")] use_namespaces: use_namespaces(), @@ -6574,6 +6906,8 @@ exit 1 max_cleanup_backoff: Duration::from_millis(DEFAULT_MAX_CLEANUP_BACKOFF), timeout_handled_externally: false, active_input_leases: false, + #[cfg(target_os = "linux")] + readonly_input_mounts: Vec::new(), directory_cache: None, #[cfg(target_os = "linux")] use_namespaces: use_namespaces(), @@ -6661,6 +6995,8 @@ exit 1 max_cleanup_backoff: Duration::from_millis(DEFAULT_MAX_CLEANUP_BACKOFF), timeout_handled_externally: false, active_input_leases: false, + #[cfg(target_os = "linux")] + readonly_input_mounts: Vec::new(), directory_cache: None, #[cfg(target_os = "linux")] use_namespaces: use_namespaces(), @@ -6769,6 +7105,8 @@ exit 1 max_cleanup_backoff: Duration::from_millis(DEFAULT_MAX_CLEANUP_BACKOFF), timeout_handled_externally: false, active_input_leases: false, + #[cfg(target_os = "linux")] + readonly_input_mounts: Vec::new(), directory_cache: None, #[cfg(target_os = "linux")] use_namespaces: use_namespaces(), @@ -6821,6 +7159,8 @@ exit 1 max_cleanup_backoff: Duration::from_millis(DEFAULT_MAX_CLEANUP_BACKOFF), timeout_handled_externally: false, active_input_leases: false, + #[cfg(target_os = "linux")] + readonly_input_mounts: Vec::new(), directory_cache: None, #[cfg(target_os = "linux")] use_namespaces: use_namespaces(), @@ -6894,6 +7234,8 @@ exit 1 max_cleanup_backoff: Duration::from_millis(DEFAULT_MAX_CLEANUP_BACKOFF), timeout_handled_externally: false, active_input_leases: false, + #[cfg(target_os = "linux")] + readonly_input_mounts: Vec::new(), directory_cache: None, #[cfg(target_os = "linux")] use_namespaces: use_namespaces(), @@ -7018,6 +7360,8 @@ exit 1 max_cleanup_backoff: Duration::from_millis(DEFAULT_MAX_CLEANUP_BACKOFF), timeout_handled_externally: false, active_input_leases: false, + #[cfg(target_os = "linux")] + readonly_input_mounts: Vec::new(), directory_cache: None, #[cfg(target_os = "linux")] use_namespaces: use_namespaces(), @@ -7112,6 +7456,8 @@ exit 1 max_cleanup_backoff: Duration::from_millis(DEFAULT_MAX_CLEANUP_BACKOFF), timeout_handled_externally: false, active_input_leases: false, + #[cfg(target_os = "linux")] + readonly_input_mounts: Vec::new(), directory_cache: None, #[cfg(target_os = "linux")] use_namespaces: use_namespaces(), @@ -7206,6 +7552,8 @@ exit 1 max_cleanup_backoff: Duration::from_millis(DEFAULT_MAX_CLEANUP_BACKOFF), timeout_handled_externally: false, active_input_leases: false, + #[cfg(target_os = "linux")] + readonly_input_mounts: Vec::new(), directory_cache: None, #[cfg(target_os = "linux")] use_namespaces: use_namespaces(), @@ -7301,6 +7649,8 @@ exit 1 max_cleanup_backoff: Duration::from_millis(DEFAULT_MAX_CLEANUP_BACKOFF), timeout_handled_externally: false, active_input_leases: false, + #[cfg(target_os = "linux")] + readonly_input_mounts: Vec::new(), directory_cache: None, #[cfg(target_os = "linux")] use_namespaces: use_namespaces(), @@ -7463,6 +7813,8 @@ exit 1 max_cleanup_backoff: Duration::from_millis(DEFAULT_MAX_CLEANUP_BACKOFF), timeout_handled_externally: false, active_input_leases: false, + #[cfg(target_os = "linux")] + readonly_input_mounts: Vec::new(), directory_cache: None, #[cfg(target_os = "linux")] use_namespaces: use_namespaces(), @@ -7640,6 +7992,8 @@ exit 1 max_cleanup_backoff: Duration::from_millis(DEFAULT_MAX_CLEANUP_BACKOFF), timeout_handled_externally: false, active_input_leases: false, + #[cfg(target_os = "linux")] + readonly_input_mounts: Vec::new(), directory_cache: None, #[cfg(target_os = "linux")] use_namespaces: use_namespaces(), @@ -7754,6 +8108,8 @@ exit 1 max_cleanup_backoff: Duration::from_millis(DEFAULT_MAX_CLEANUP_BACKOFF), timeout_handled_externally: false, active_input_leases: false, + #[cfg(target_os = "linux")] + readonly_input_mounts: Vec::new(), directory_cache: None, #[cfg(target_os = "linux")] use_namespaces: use_namespaces(), @@ -7870,6 +8226,8 @@ exit 1 max_cleanup_backoff: Duration::from_millis(DEFAULT_MAX_CLEANUP_BACKOFF), timeout_handled_externally: false, active_input_leases: false, + #[cfg(target_os = "linux")] + readonly_input_mounts: Vec::new(), directory_cache: None, // Pin namespaces off so this exercises the no-pre_exec/posix_spawn // path regardless of what the host kernel supports. @@ -7990,6 +8348,8 @@ exit 1 max_cleanup_backoff: Duration::from_millis(DEFAULT_MAX_CLEANUP_BACKOFF), timeout_handled_externally: false, active_input_leases: false, + #[cfg(target_os = "linux")] + readonly_input_mounts: Vec::new(), directory_cache: None, use_namespaces: nativelink_worker::running_actions_manager::UseNamespaces::YesAndMount { @@ -8117,6 +8477,8 @@ exit 1 max_cleanup_backoff: Duration::from_millis(DEFAULT_MAX_CLEANUP_BACKOFF), timeout_handled_externally: false, active_input_leases: false, + #[cfg(target_os = "linux")] + readonly_input_mounts: Vec::new(), directory_cache: None, #[cfg(target_os = "linux")] use_namespaces: use_namespaces(), @@ -8282,6 +8644,8 @@ while [ ! -f "$0.d/release" ]; do "{sleep}" 0.01; done max_cleanup_backoff: Duration::from_millis(DEFAULT_MAX_CLEANUP_BACKOFF), timeout_handled_externally: false, active_input_leases: false, + #[cfg(target_os = "linux")] + readonly_input_mounts: Vec::new(), directory_cache: None, #[cfg(target_os = "linux")] use_namespaces: nativelink_worker::running_actions_manager::UseNamespaces::No, @@ -8534,6 +8898,8 @@ while [ ! -f "$0.d/release" ]; do "{sleep}" 0.01; done max_cleanup_backoff: Duration::from_millis(DEFAULT_MAX_CLEANUP_BACKOFF), timeout_handled_externally: false, active_input_leases: false, + #[cfg(target_os = "linux")] + readonly_input_mounts: Vec::new(), directory_cache: None, #[cfg(target_os = "linux")] use_namespaces: use_namespaces(), @@ -8635,6 +9001,8 @@ done max_cleanup_backoff: Duration::from_millis(DEFAULT_MAX_CLEANUP_BACKOFF), timeout_handled_externally: false, active_input_leases: false, + #[cfg(target_os = "linux")] + readonly_input_mounts: Vec::new(), directory_cache: None, #[cfg(target_os = "linux")] use_namespaces: use_namespaces(), @@ -8822,6 +9190,8 @@ done max_cleanup_backoff: Duration::from_millis(DEFAULT_MAX_CLEANUP_BACKOFF), timeout_handled_externally: false, active_input_leases: false, + #[cfg(target_os = "linux")] + readonly_input_mounts: Vec::new(), directory_cache: None, #[cfg(target_os = "linux")] use_namespaces: use_namespaces(), @@ -8956,6 +9326,8 @@ done max_cleanup_backoff: Duration::from_millis(DEFAULT_MAX_CLEANUP_BACKOFF), timeout_handled_externally: false, active_input_leases: false, + #[cfg(target_os = "linux")] + readonly_input_mounts: Vec::new(), directory_cache: None, #[cfg(target_os = "linux")] use_namespaces: use_namespaces(), @@ -9111,6 +9483,8 @@ done max_cleanup_backoff: Duration::from_millis(DEFAULT_MAX_CLEANUP_BACKOFF), timeout_handled_externally: false, active_input_leases: false, + #[cfg(target_os = "linux")] + readonly_input_mounts: Vec::new(), directory_cache: None, #[cfg(target_os = "linux")] use_namespaces: use_namespaces(), @@ -9227,6 +9601,8 @@ done max_cleanup_backoff: Duration::from_millis(DEFAULT_MAX_CLEANUP_BACKOFF), timeout_handled_externally: false, active_input_leases: false, + #[cfg(target_os = "linux")] + readonly_input_mounts: Vec::new(), directory_cache: None, #[cfg(target_os = "linux")] use_namespaces: use_namespaces(), @@ -9373,6 +9749,8 @@ done max_cleanup_backoff: Duration::from_millis(DEFAULT_MAX_CLEANUP_BACKOFF), timeout_handled_externally: false, active_input_leases: false, + #[cfg(target_os = "linux")] + readonly_input_mounts: Vec::new(), directory_cache: None, #[cfg(target_os = "linux")] use_namespaces: use_namespaces(), @@ -9550,6 +9928,8 @@ done max_cleanup_backoff: Duration::from_millis(DEFAULT_MAX_CLEANUP_BACKOFF), timeout_handled_externally: false, active_input_leases: false, + #[cfg(target_os = "linux")] + readonly_input_mounts: Vec::new(), directory_cache: None, #[cfg(target_os = "linux")] use_namespaces: use_namespaces(), @@ -9711,6 +10091,8 @@ done max_cleanup_backoff: Duration::from_millis(DEFAULT_MAX_CLEANUP_BACKOFF), timeout_handled_externally: false, active_input_leases: false, + #[cfg(target_os = "linux")] + readonly_input_mounts: Vec::new(), directory_cache: None, #[cfg(target_os = "linux")] use_namespaces: use_namespaces(), diff --git a/web/apps/docs/content/docs/explanations/architecture.mdx b/web/apps/docs/content/docs/explanations/architecture.mdx index f0a23e953b0..44b9afb6b91 100644 --- a/web/apps/docs/content/docs/explanations/architecture.mdx +++ b/web/apps/docs/content/docs/explanations/architecture.mdx @@ -95,6 +95,9 @@ the merge rule, the ordering rule and the retry rule. A worker fetches the action's inputs from the CAS, materialises them into a directory, runs the command, uploads the outputs, and reports back. +Linux workers can optionally share cached input directories between actions +through [read-only mounts](/reference/nativelink-config/main#localworkerconfig), +avoiding repeated directory creation. Workers hold no durable state. A worker that dies mid-action costs you that action, not your cache. What they do hold is a local filesystem that is diff --git a/web/apps/docs/content/docs/operate/observability.mdx b/web/apps/docs/content/docs/operate/observability.mdx index 75f65fb194b..582ca1fe7c2 100644 --- a/web/apps/docs/content/docs/operate/observability.mdx +++ b/web/apps/docs/content/docs/operate/observability.mdx @@ -504,6 +504,15 @@ not your pipeline. +## Experimental input mounts [#experimental-input-mounts] + +Workers emit `worker.input_mount.cache_hits` and `cache_misses` to show +whether selected directories are reused or rebuilt. `worker.input_mount.fallbacks` +counts actions using ordinary input setup despite configured mounts; +`worker.input_mount.prepare_failures` counts failed preparations retried without +mounts. These counters describe preparation, not successful command execution. +These names use OTLP notation; see [Prometheus naming conventions](/reference/metrics#how-an-instrument-becomes-a-prometheus-series) for the conversion. + ## What's next [#whats-next] diff --git a/web/apps/docs/content/docs/reference/nativelink-config/main.mdx b/web/apps/docs/content/docs/reference/nativelink-config/main.mdx index 1e18d69b7a9..b830b795fd9 100644 --- a/web/apps/docs/content/docs/reference/nativelink-config/main.mdx +++ b/web/apps/docs/content/docs/reference/nativelink-config/main.mdx @@ -5,7 +5,7 @@ full: true --- {/* AUTOGENERATED — do not edit by hand. - Source: nativelink-config @ main (6f4a212b) + Source: nativelink-config @ main (23b4472a) Regenerate from web/: bun --filter @nativelink/docs gen:config-reference */} @@ -919,6 +919,7 @@ Configuration for `ExperimentalMongoDB` store. | `max_action_timeout_s` | integer (uint) | — | 20 minutes | The maximum time an action is allowed to run. If a task requests for a timeout longer than this time limit, the task will be rejected. Value in seconds. | | `max_upload_timeout_s` | integer (uint) | — | 10 minutes | Maximum time allowed for uploading action results to CAS after execution completes. If upload takes longer than this, the action fails with `DeadlineExceeded` and may be retried by the scheduler. Value in seconds. | | `max_download_timeout_s` | integer (uint) | — | 10 minutes | Maximum time allowed for fetching an action's inputs into the worker's store and its directory before the command runs. A fetch that hangs (a store that never answers) otherwise holds the action's slot for good while the worker looks healthy; past this the action fails with `DeadlineExceeded` and may be retried by the scheduler, and a warning names the action every minute before that. Value in seconds. | +| `max_concurrent_input_downloads` | integer (uint) | — | 64 | How many input blobs this worker fetches into its store at once, across every action it is staging. Each fetch is a read from the CAS or the object store, so this bounds what one worker asks of them at a time: several cold actions share this many fetches instead of each opening its own. Blobs already in the store are linked without one. | | `max_cleanup_wait_s` | integer (uint) | — | 30 seconds | Maximum time to wait for action directory cleanup before timing out. Value in seconds. | | `max_cleanup_backoff_ms` | integer (uint) | — | 500 milliseconds | Maximum backoff duration for exponential backoff when waiting for cleanup. Value in milliseconds. | | `max_inflight_tasks` | integer (uint64) | — | 0 (infinite tasks) | Maximum number of inflight tasks this worker can cope with. | @@ -943,6 +944,7 @@ Configuration for `ExperimentalMongoDB` store. | `additional_environment` | map of string to [EnvironmentSource](#environmentsource) | — | — | An optional mapping of environment names to set for the execution as well as those specified in the action itself. If set, will set each key as an environment variable before executing the job with the value of the environment variable being the value of the property of the action being executed of that name or the fixed value. | | `directory_cache` | [DirectoryCacheConfig](#directorycacheconfig) | — | — | Optional directory cache configuration for improving performance by caching reconstructed input directories and using hardlinks instead of rebuilding them from CAS for every action. | | `experimental_active_input_leases` | boolean | — | false (eviction behavior is unchanged) | Optional and experimental: lease every digest in an active action's input Merkle closure in the worker's locally eviction-managed CAS tiers (the `cas_fast_slow_store`'s filesystem-backed stores) until the action has finished cleanup. This prevents `Lost inputs no longer available remotely` failures caused by local fast-tier eviction while inputs are being materialized under cache pressure. | +| `experimental_readonly_input_mounts` | array of string | — | [] (disabled) | Reuse immutable input directories through read-only bind mounts on Linux. The worker prepares each selected subtree once per REAPI `Directory` digest in the directory cache, then mounts it into each action instead of recreating its filesystem entries. | | `use_namespaces` | boolean | — | False | Whether to use namespaces to isolate the execution. This is only available on Linux. It is highly recommended as it avoids a number of issues with zombie processes and also provides additional hermeticity. If explicitly set to true and it is not supported the worker will exit with an error. | | `use_mount_namespace` | boolean | — | False | Whether to use a mount namespace to isolate the worker root. This is only available on Linux and when `use_namespaces` is true. It is highly recommended provides additional hermeticity. If explicitly set to true and it is not supported or `use_namespaces` is not set to true the worker will exit with an error. | | `isolate_tmp` | boolean | — | True when `use_mount_namespace` is true and `/tmp` exists, | Whether to give each action a private `/tmp` inside its mount namespace: the action's own `tmp` directory (the one `TMPDIR` points at) is bound over `/tmp`, so tools that hardcode `/tmp` write there too. Concurrent actions no longer collide on predictable paths under `/tmp`, nothing leaks between actions through it, and what an action writes there is removed with the action. | @@ -957,6 +959,7 @@ Configuration for `ExperimentalMongoDB` store. | `enable_active_action_count_metric` | boolean | — | false | Periodically count the actions in each stage and report them as the `execution.active.count` metric. | | `worker_timeout_s` | integer (uint64) | — | 10 seconds (four default keepalive intervals) | Remove workers from pool once the worker has not responded in this amount of time in seconds. Any message from the worker counts, not only keepalives. Eviction requeues everything the worker held, so keep this well above the longest pause a loaded worker can see. | | `max_action_executing_timeout_s` | integer (uint64) | — | 0 (disabled) | Maximum time (seconds) an action can stay in Executing state without any worker update before being timed out and re-queued. This applies regardless of worker keepalive status, catching cases where a worker is alive (sending keepalives) but stuck on a specific action. Set to 0 to disable (relies only on `worker_timeout_s`). | +| `executing_sweep_interval_s` | integer (int64) | — | `30` | How often the scheduler sweeps executing actions for ones to retire: an action past `max_action_executing_timeout_s` (or its own `Action.timeout` plus `worker_timeout_s`) with no word from its worker, and an action whose worker is connected to this scheduler but no longer runs it, which means its result was lost on the way to the store. Without the sweep those checks run only inside a client's `Execute` or `WaitExecution` call, once per `worker_timeout_s`, and never once the client has gone. Defaults to 30s. Zero or any negative value disables it. | | `unacknowledged_kill_timeout_s` | integer (uint64) | — | 60 seconds | Evict a worker that has not reported back on an operation it was told to kill within this many seconds. A healthy worker acknowledges a kill in moments; one that cannot is wedged, and its keepalives would otherwise keep `worker_timeout_s` from ever firing while the dead operation holds its slot. Eviction requeues the worker's other operations. | | `dispatch_ack_timeout_s` | integer (uint64) | — | unset (off) | Requeue a dispatched action the worker has not acknowledged within this many seconds, untried and without counting an attempt, and pause the worker. The clock is the worker's own messages: a dispatch is dated by the worker's last message before it, and the check runs on each later message, so keep this above a few keepalive intervals. Workers from v1.7.3 acknowledge every dispatch; older workers never do, so leave this unset while any are connected or their every action would be requeued. | | `unsatisfiable_action_timeout_s` | integer (uint64) | — | 0 (never fail) | Fail a queued action once no connected worker has been able to run it for this many seconds, instead of leaving it queued forever. An action counts as impossible to run when no connected worker could take it even when idle: an `exact` value no worker has, a required key no worker declares, or a `minimum` larger than any worker's total. An action that is only waiting for a busy worker is never failed by this setting, and neither is any action while no workers are connected at all. | @@ -966,8 +969,9 @@ Configuration for `ExperimentalMongoDB` store. | `allocation_strategy` | [WorkerAllocationStrategy](#workerallocationstrategy) | — | `"least_recently_used"` | The strategy used to assign workers jobs. | | `live_memory_veto` | string | — | unset (off) | The name of a `minimum` platform property, usually `memory_kb`, that the scheduler compares against the free memory each worker reports with its keepalive: a worker reporting less than the action asks for is skipped for that action, whatever the admission ledger says. The ledger only subtracts what actions declare; this catches the ones that declared too little. Workers older than v1.7.3 report nothing and are never vetoed. | | `memory_escalation` | [MemoryEscalationSpec](#memoryescalationspec) | — | unset (a memory kill fails the action) | When a worker reports an action killed for memory (`KILLED_MEMORY`, by its own reservation enforcement or by the kernel), requeue the action with a larger reservation instead of failing it. Escalations have their own budget (`max_steps`) and do not spend `max_job_retries`. The reservation grows to the largest memory any connected worker advertises (or `max_kb`); the last step reserves that worker whole, memory and CPU, so the action runs alone, and only a kill there fails the action: nothing in the fleet could run it. | +| `reservation_hold` | [ReservationHoldSpec](#reservationholdspec) | — | unset (off) | Hold a worker for a queued action that keeps finding no room. An action whose reservation only fits an idle worker starves under load: every pass tries it first and finds each worker partly busy, then fills the room that opens with newer, smaller actions, so no worker ever drains for it. With this set, once such an action has waited `hold_after_s` and fits no busy worker until more than one of that worker's actions finish (one finishing would make room for an action that is only waiting its turn), the scheduler picks a worker that would fit it idle and offers it nothing else until a worker takes the action, it leaves the queue, or `max_hold_s` passes. An action no connected worker would fit even idle gets no hold. Each scheduler replica holds only workers connected to it; replicas sharing a queue take turns per action, by a hash of the action over the replicas whose workers would fit it as the fleet census last read them, so one worker is held for it fleet-wide. While a replica cannot read the census it holds on its own. | | `experimental_backend` | [ExperimentalSimpleSchedulerBackend](#experimentalsimpleschedulerbackend) | — | memory | The storage backend to use for the scheduler. | -| `worker_match_logging_interval_s` | integer (int64) | — | `10` | Every N seconds, do logging of worker matching e.g. "worker busy", "can't find any worker" Defaults to 10s. Can be set to `-1` to disable | +| `worker_match_logging_interval_s` | integer (int64) | — | `-1` | Every N seconds, do logging of worker matching e.g. "worker busy", "can't find any worker", and the oldest queued and executing operations. Each pass lists every queued and executing operation from the store, on every scheduler replica: with a shared Redis that is a full index read per interval per replica, which is why it is off unless asked for. Turn it on for a look, not for good. Defaults to -1 (disabled). Any positive value enables it. | | `fallback_match_interval_s` | integer (int64) | — | `5` | Every N seconds, run a worker matching pass even if no task or worker change notification arrived. This is a safety net for missed notifications and for scheduler backends with eventually consistent searches (for example Redis), where an operation that was re-queued may not be visible to the search triggered by its own notification. Without this, such an operation can stay queued until an unrelated event triggers another matching pass. Defaults to 5s. Zero or any negative value disables it. | ## SchedulerGrpcSpec @@ -983,6 +987,7 @@ build at the main scheduler directly though. | `retry` | [Retry](#retry) | — | — | Retry configuration to use when a network request fails. | | `max_concurrent_requests` | integer (uint) | — | unlimited | Limit the number of simultaneous upstream requests to this many. A value of zero is treated as unlimited. If the limit is reached the request is queued. | | `connections_per_endpoint` | integer (uint) | — | 1 | The number of connections to make to each specified endpoint to balance the load over multiple TCP connections. | +| `forward_headers` | array of string | — | — | Header names to copy from the incoming client request onto every request sent to the upstream scheduler. Use this to pass through dynamic credentials such as a JWT sent by the build client. | ## CacheLookupSpec @@ -1122,6 +1127,9 @@ The total delay is additive, so this example produces 9.525 to 15.875 s of total | `multipart_max_concurrent_uploads` | integer (uint) | — | 10 | Maximum number of concurrent `UploadPart` requests per `MultipartUpload`. | | `insecure_allow_http` | boolean | — | false | Allow unencrypted HTTP connections. Only use this for local testing. | | `disable_http2` | boolean | — | false | Disable HTTP/2 connections and only use HTTP/1.1. Default client configuration will have HTTP/1.1 and HTTP/2 enabled for connection schemes. HTTP/2 should be disabled if environments have poor support or performance related to HTTP/2. Safe to keep default unless underlying network environment, S3, or GCS API servers specify otherwise. | +| `throttle_retry_delay_s` | number (float) | — | 2.0 | Minimum number of seconds to wait after a response that asked the client to slow down (HTTP 503 `ServerBusy` or 429) when it carries no `Retry-After` header.0 | +| `throttle_retry_max_delay_s` | number (float) | — | 60.0 | Longest wait honoured from a `Retry-After` header. A server or proxy asking for more waits this long instead, and the capped wait is logged. Must not be less than `throttle_retry_delay_s`.0 | +| `throttle_retry_budget_s` | number (float) | — | 120.0 | Seconds one operation may spend waiting on throttled responses without spending any of `retry.max_retries`: while the budget lasts each throttle waits what the server asked (at least `throttle_retry_delay_s`, at most `throttle_retry_max_delay_s`) and the attempt is not counted, so a long throttle is ridden out rather than failed on the seventh try. Once it is used up, throttles spend attempts and backoff like any other transient error. 0 turns the budget off.0 | ## CompressionAlgorithm @@ -1368,6 +1376,14 @@ workers are able to run the task. | `cpu_property` | string | — | `cpu_count` | The `minimum` platform property carrying the CPU reservation. The last escalation reserves the largest worker's whole memory and, through this property, its whole CPU, so nothing shares the worker with the action. Empty leaves CPU alone. | | `max_steps` | integer (uint64) | — | 8 | The most escalations one action gets before the scheduler completes it with `FailedPrecondition`; 0 lets the ceiling alone bound them. | +## ReservationHoldSpec + +| Field | Type | Required | Default | Description | +| --- | --- | --- | --- | --- | +| `hold_after_s` | integer (uint64) | — | 30 | Seconds a queued action has to have waited, finding no room on every pass, before a worker is held for it. Must be above zero. | +| `max_hold_s` | integer (uint64) | — | 300 | Seconds a hold lasts at most, capped at a week. A hold that long without the action fitting means the worker never reported room for it even drained; the hold is given up so the worker is not idle for good, and the action is not held for again until as long has passed, twice as long for each such expiry in a row on a worker that was idle (up to sixteen times), since then nothing kept the action off the worker but the action itself. Must be above zero. | +| `max_held_workers` | integer (uint) | — | 0 | Most workers held at once on this scheduler replica. 0 is one tenth of the workers connected to it, and at least one. | + ## ExperimentalSimpleSchedulerBackend One of: From bf40ef497fc0ece4af2a4e56ce3015648cfbc362 Mon Sep 17 00:00:00 2001 From: Marcus Eagan Date: Sat, 17 Oct 2026 01:23:52 +0200 Subject: [PATCH 2/2] Fix clippy and vale failures from the rebase onto main The test file now imports PathBuf, so the fully qualified path is an unnecessary qualification under -D warnings. Regenerating the config reference picked up a scheduler doc comment that vale rejects ("-1" with a hyphen); format it as code at the source and regenerate. Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_01HBxQtV9S561AzBo7aPCYFj --- nativelink-config/src/schedulers.rs | 2 +- nativelink-worker/tests/namespace_utils_test.rs | 2 +- .../docs/content/docs/reference/nativelink-config/main.mdx | 4 ++-- 3 files changed, 4 insertions(+), 4 deletions(-) diff --git a/nativelink-config/src/schedulers.rs b/nativelink-config/src/schedulers.rs index ad00f642c80..e1edd19bfcc 100644 --- a/nativelink-config/src/schedulers.rs +++ b/nativelink-config/src/schedulers.rs @@ -375,7 +375,7 @@ pub struct SimpleSpec { /// operation from the store, on every scheduler replica: with a shared /// Redis that is a full index read per interval per replica, which is /// why it is off unless asked for. Turn it on for a look, not for good. - /// Defaults to -1 (disabled). Any positive value enables it. + /// Defaults to `-1` (disabled). Any positive value enables it. #[serde( default = "default_worker_match_logging_interval_s", deserialize_with = "convert_duration_with_shellexpand_and_negative" diff --git a/nativelink-worker/tests/namespace_utils_test.rs b/nativelink-worker/tests/namespace_utils_test.rs index 2b37d2d0a52..1a6651d5b59 100644 --- a/nativelink-worker/tests/namespace_utils_test.rs +++ b/nativelink-worker/tests/namespace_utils_test.rs @@ -622,7 +622,7 @@ async fn readonly_input_mount_is_private_and_source_survives_masking() -> Result eprintln!("SKIP: mount namespaces unavailable"); return Ok(()); } - let root = std::path::PathBuf::from(nativelink_util::common::make_temp_path("readonly_mount")); + let root = PathBuf::from(nativelink_util::common::make_temp_path("readonly_mount")); let action = root.join("action"); let target = action.join("sdk"); // Intentionally put the source under the root that configure_namespace diff --git a/web/apps/docs/content/docs/reference/nativelink-config/main.mdx b/web/apps/docs/content/docs/reference/nativelink-config/main.mdx index b830b795fd9..339ac4f06e6 100644 --- a/web/apps/docs/content/docs/reference/nativelink-config/main.mdx +++ b/web/apps/docs/content/docs/reference/nativelink-config/main.mdx @@ -5,7 +5,7 @@ full: true --- {/* AUTOGENERATED — do not edit by hand. - Source: nativelink-config @ main (23b4472a) + Source: nativelink-config @ main (1a95d037) Regenerate from web/: bun --filter @nativelink/docs gen:config-reference */} @@ -971,7 +971,7 @@ Configuration for `ExperimentalMongoDB` store. | `memory_escalation` | [MemoryEscalationSpec](#memoryescalationspec) | — | unset (a memory kill fails the action) | When a worker reports an action killed for memory (`KILLED_MEMORY`, by its own reservation enforcement or by the kernel), requeue the action with a larger reservation instead of failing it. Escalations have their own budget (`max_steps`) and do not spend `max_job_retries`. The reservation grows to the largest memory any connected worker advertises (or `max_kb`); the last step reserves that worker whole, memory and CPU, so the action runs alone, and only a kill there fails the action: nothing in the fleet could run it. | | `reservation_hold` | [ReservationHoldSpec](#reservationholdspec) | — | unset (off) | Hold a worker for a queued action that keeps finding no room. An action whose reservation only fits an idle worker starves under load: every pass tries it first and finds each worker partly busy, then fills the room that opens with newer, smaller actions, so no worker ever drains for it. With this set, once such an action has waited `hold_after_s` and fits no busy worker until more than one of that worker's actions finish (one finishing would make room for an action that is only waiting its turn), the scheduler picks a worker that would fit it idle and offers it nothing else until a worker takes the action, it leaves the queue, or `max_hold_s` passes. An action no connected worker would fit even idle gets no hold. Each scheduler replica holds only workers connected to it; replicas sharing a queue take turns per action, by a hash of the action over the replicas whose workers would fit it as the fleet census last read them, so one worker is held for it fleet-wide. While a replica cannot read the census it holds on its own. | | `experimental_backend` | [ExperimentalSimpleSchedulerBackend](#experimentalsimpleschedulerbackend) | — | memory | The storage backend to use for the scheduler. | -| `worker_match_logging_interval_s` | integer (int64) | — | `-1` | Every N seconds, do logging of worker matching e.g. "worker busy", "can't find any worker", and the oldest queued and executing operations. Each pass lists every queued and executing operation from the store, on every scheduler replica: with a shared Redis that is a full index read per interval per replica, which is why it is off unless asked for. Turn it on for a look, not for good. Defaults to -1 (disabled). Any positive value enables it. | +| `worker_match_logging_interval_s` | integer (int64) | — | `-1` | Every N seconds, do logging of worker matching e.g. "worker busy", "can't find any worker", and the oldest queued and executing operations. Each pass lists every queued and executing operation from the store, on every scheduler replica: with a shared Redis that is a full index read per interval per replica, which is why it is off unless asked for. Turn it on for a look, not for good. Defaults to `-1` (disabled). Any positive value enables it. | | `fallback_match_interval_s` | integer (int64) | — | `5` | Every N seconds, run a worker matching pass even if no task or worker change notification arrived. This is a safety net for missed notifications and for scheduler backends with eventually consistent searches (for example Redis), where an operation that was re-queued may not be visible to the search triggered by its own notification. Without this, such an operation can stay queued until an unrelated event triggers another matching pass. Defaults to 5s. Zero or any negative value disables it. | ## SchedulerGrpcSpec