queue-runner and builder: fail fast on infrastructure errors

The queue-runner should not try to recover from a down database or
invalid store state — only builder failures should be totally
recoverable since builders are numerous and expected to come and go.

The builder should die if it cannot communicate with the queue-runner —
the queue-runner will notice and retry the step elsewhere.

Also there are `nix_utils::BaseStore` trait changes:

- `is_valid_path` now returns `Result<bool, Error>` instead of `bool`

- `query_path_info` now returns `Result<Option<...>, Error>` instead of
  `Option<...>`

- `query_path_infos` now returns `Result<HashMap<...>, Error>` instead
  of `HashMap<...>`

- `compute_closure_size` now returns `Result<u64, Error>` instead of
  `u64`

Previously the trait implementations silently swallowed FFI errors as
"path not found" / empty / zero. This was the root cause of silent
failures: a daemon connection error looked like "path is present" to
callers, causing paths to be skipped from import and builds to fail with
cryptic errors.

The C++ `query_path_info` now checks `isValidPath` first and returns a
`found: false` sentinel instead of throwing `InvalidPath` — eliminating
brittle error-message string matching on the Rust side.

All callers updated to propagate errors with `?`.

**Note: Manual review of the queue runner error comments is not yet
complete!**

Queue-runner changes:

- `create_build` returns `Result`; an invalid drv path is a hard error
  (indicates a GC rooting bug) rather than silently aborting the build

- `handle_previous_failure` and `handle_cached_build` errors propagate
  (DB unreachable = stop processing)

- `process_new_builds` returns `Result` and propagates child build
  creation errors

- Queue monitor loop propagates `get_queued_builds`,
  `process_queue_change`, and `handle_jobset_change` errors instead of
  logging and continuing

- Log directory creation at startup is fatal on failure

- mTLS misconfiguration uses `anyhow::ensure!`

- Non-fatal `tracing::error!` sites documented with comments
  explaining why they are correct to log-and-continue

Builder changes:

- `filter_missing` returns `Result` — daemon connection failure is
  a hard error, not "path is present"

- `handle_request` errors propagate (kills the builder — queue-runner
  will retry)

- Ping message construction failure breaks the ping stream (builder
  exits), with the error captured and returned from the gRPC function

- `submit_build_result` helper deduplicates the retry-then-report
  pattern for `complete_build` calls

- Build task returns `Result<()>` instead of `()` — failure to submit
  results after retries propagates rather than calling
  `process::exit(1)`

- mTLS misconfiguration uses `anyhow::ensure!`

- Unreachable error paths in log stream (`utils.rs`) now panic with
  explanatory comments

- `substitute_output` S3 replication uses async-block-as-try-block
  pattern for clean early exit on daemon error

Author: Ericson2314

subprojects/crates/binary-cache/src/lib.rs

@@ -56,15 +56,16 @@ pub async fn path_to_narinfo(
     store: &nix_utils::LocalStore,
     path: &StorePath,
 ) -> Result<NarInfo, CacheError> {
-    let Some(path_info) = store.query_path_info(path).await else {
-        return Err(CacheError::PathNotFound {
+    let path_info = store
+        .query_path_info(path)
+        .await?
+        .ok_or_else(|| CacheError::PathNotFound {
             path: path.to_string(),
-        });
-    };
+        })?;
     let narinfo = narinfo_simple(path, path_info, Compression::None);
     let queried_references = store
         .query_path_infos(&narinfo.info.info.references.iter().collect::<Vec<_>>())
-        .await;
+        .await?;
     for r in &narinfo.info.info.references {
         if !queried_references.contains_key(r) {
             return Err(CacheError::ReferenceVerifyError(narinfo.path, r.to_owned()));
@@ -549,11 +550,13 @@ impl S3BinaryCacheClient {
         store: &nix_utils::LocalStore,
         path: &StorePath,
     ) -> Result<NarInfo, CacheError> {
-        let Some(path_info) = store.query_path_info(path).await else {
-            return Err(CacheError::PathNotFound {
-                path: path.to_string(),
-            });
-        };
+        let path_info =
+            store
+                .query_path_info(path)
+                .await?
+                .ok_or_else(|| CacheError::PathNotFound {
+                    path: path.to_string(),
+                })?;
         let narinfo = narinfo_from_path_info(
             path,
             path_info,
@@ -563,7 +566,7 @@ impl S3BinaryCacheClient {
         );
         let queried_references = store
             .query_path_infos(&narinfo.info.info.references.iter().collect::<Vec<_>>())
-            .await;
+            .await?;
         for r in &narinfo.info.info.references {
             if !queried_references.contains_key(r) {
                 return Err(CacheError::ReferenceVerifyError(narinfo.path, r.to_owned()));

subprojects/crates/nix-utils/examples/is_valid_path.rs

@@ -1,13 +1,14 @@
 use nix_utils::{self, BaseStore as _};
 
 #[tokio::main]
-async fn main() {
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let store = nix_utils::LocalStore::init();
     let store_dir = nix_utils::get_store_dir();
     println!(
         "storepath={store_dir} valid={}",
         store
             .is_valid_path(&AsRef::<str>::as_ref(&store_dir).parse().unwrap())
-            .await
+            .await?
     );
+    Ok(())
 }

subprojects/crates/nix-utils/examples/path_infos.rs

@@ -1,7 +1,7 @@
 use nix_utils::BaseStore as _;
 
 #[tokio::main]
-async fn main() {
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let local = nix_utils::LocalStore::init();
 
     let p1 = "ihl4ya67glh9815v1lanyqph0p7hdzfb-hdf5-cpp-1.14.6-bin"
@@ -19,16 +19,17 @@ async fn main() {
     println!("{infos:?}");
     println!(
         "closure_size {p1}: {}",
-        local.compute_closure_size(&p1).await
+        local.compute_closure_size(&p1).await?
     );
     println!(
         "closure_size {p2}: {}",
-        local.compute_closure_size(&p2).await
+        local.compute_closure_size(&p2).await?
     );
     println!(
         "closure_size {p3}: {}",
-        local.compute_closure_size(&p3).await
+        local.compute_closure_size(&p3).await?
     );
 
     println!("stats: {:?}", local.get_store_stats());
+    Ok(())
 }

subprojects/crates/nix-utils/src/drv.rs

@@ -34,7 +34,7 @@ pub async fn query_drv(
         return Ok(None);
     }
 
-    if !store.is_valid_path(drv).await {
+    if !store.is_valid_path(drv).await? {
         return Ok(None);
     }
 

subprojects/crates/nix-utils/src/lib.rs

@@ -79,6 +79,7 @@ mod ffi {
 
     #[derive(Debug, Clone)]
     struct InternalPathInfo {
+        found: bool,
         store_dir: String,
         deriver: String,
         nar_hash: Vec<u8>,
@@ -368,6 +369,7 @@ fn to_internal_path_info(
     let nar_hash_bytes: Vec<u8> = AsRef::<[u8]>::as_ref(&info.nar_hash).to_vec();
 
     ffi::InternalPathInfo {
+        found: true,
         store_dir: store_dir.to_str().to_owned(),
         deriver: info
             .deriver
@@ -386,19 +388,18 @@ fn to_internal_path_info(
 }
 
 pub trait BaseStore {
-    #[must_use]
     /// Check whether a path is valid.
-    fn is_valid_path(&self, path: &StorePath) -> impl Future<Output = bool>;
+    fn is_valid_path(&self, path: &StorePath) -> impl Future<Output = Result<bool, Error>>;
 
     fn query_path_info(
         &self,
         path: &StorePath,
-    ) -> impl Future<Output = Option<UnkeyedValidPathInfo>>;
+    ) -> impl Future<Output = Result<Option<UnkeyedValidPathInfo>, Error>>;
     fn query_path_infos(
         &self,
         paths: &[&StorePath],
-    ) -> impl Future<Output = HashMap<StorePath, UnkeyedValidPathInfo>>;
-    fn compute_closure_size(&self, path: &StorePath) -> impl Future<Output = u64>;
+    ) -> impl Future<Output = Result<HashMap<StorePath, UnkeyedValidPathInfo>, Error>>;
+    fn compute_closure_size(&self, path: &StorePath) -> impl Future<Output = Result<u64, Error>>;
 
     fn clear_path_info_cache(&self);
 
@@ -529,33 +530,35 @@ where
 
 impl BaseStore for BaseStoreImpl {
     #[inline]
-    async fn is_valid_path(&self, path: &StorePath) -> bool {
+    async fn is_valid_path(&self, path: &StorePath) -> Result<bool, Error> {
         let store = self.wrapper.clone();
         let path = self.print_store_path(path);
-        asyncify(move || ffi::is_valid_path(store.as_raw(), &path))
-            .await
-            .unwrap_or(false)
+        asyncify(move || ffi::is_valid_path(store.as_raw(), &path)).await
     }
 
     #[inline]
-    async fn query_path_info(&self, path: &StorePath) -> Option<UnkeyedValidPathInfo> {
+    async fn query_path_info(
+        &self,
+        path: &StorePath,
+    ) -> Result<Option<UnkeyedValidPathInfo>, Error> {
         let store = self.wrapper.clone();
         let path = self.print_store_path(path);
         asyncify(move || {
-            Ok(ffi::query_path_info(store.as_raw(), &path)
-                .ok()
-                .map(Into::into))
+            let info = ffi::query_path_info(store.as_raw(), &path)?;
+            if info.found {
+                Ok(Some(info.into()))
+            } else {
+                Ok(None)
+            }
         })
         .await
-        .ok()
-        .flatten()
     }
 
     #[inline]
     async fn query_path_infos(
         &self,
         paths: &[&StorePath],
-    ) -> HashMap<StorePath, UnkeyedValidPathInfo> {
+    ) -> Result<HashMap<StorePath, UnkeyedValidPathInfo>, Error> {
         let paths = paths.iter().map(|v| (*v).to_owned()).collect::<Vec<_>>();
 
         asyncify({
@@ -564,27 +567,22 @@ impl BaseStore for BaseStoreImpl {
                 let mut res = HashMap::with_capacity(paths.len());
                 for p in paths {
                     let full_path = self_.print_store_path(&p);
-                    if let Some(info) = ffi::query_path_info(self_.wrapper.as_raw(), &full_path)
-                        .ok()
-                        .map(Into::into)
-                    {
-                        res.insert(p, info);
+                    let info = ffi::query_path_info(self_.wrapper.as_raw(), &full_path)?;
+                    if info.found {
+                        res.insert(p, info.into());
                     }
                 }
                 Ok(res)
             }
         })
         .await
-        .unwrap_or_default()
     }
 
     #[inline]
-    async fn compute_closure_size(&self, path: &StorePath) -> u64 {
+    async fn compute_closure_size(&self, path: &StorePath) -> Result<u64, Error> {
         let store = self.wrapper.clone();
         let path = self.print_store_path(path);
-        asyncify(move || ffi::compute_closure_size(store.as_raw(), &path))
-            .await
-            .unwrap_or_default()
+        asyncify(move || ffi::compute_closure_size(store.as_raw(), &path)).await
     }
 
     #[inline]
@@ -726,7 +724,7 @@ impl LocalStore {
         tokio_stream::iter(outputs)
             .map(|(name, path)| async move {
                 match path {
-                    Some(p) if self.is_valid_path(&p).await => None,
+                    Some(p) if self.is_valid_path(&p).await.unwrap_or(false) => None,
                     other => Some((name, other)),
                 }
             })
@@ -778,25 +776,28 @@ impl LocalStore {
 
 impl BaseStore for LocalStore {
     #[inline]
-    async fn is_valid_path(&self, path: &StorePath) -> bool {
+    async fn is_valid_path(&self, path: &StorePath) -> Result<bool, Error> {
         self.base.is_valid_path(path).await
     }
 
     #[inline]
-    async fn query_path_info(&self, path: &StorePath) -> Option<UnkeyedValidPathInfo> {
+    async fn query_path_info(
+        &self,
+        path: &StorePath,
+    ) -> Result<Option<UnkeyedValidPathInfo>, Error> {
         self.base.query_path_info(path).await
     }
 
     #[inline]
     async fn query_path_infos(
         &self,
         paths: &[&StorePath],
-    ) -> HashMap<StorePath, UnkeyedValidPathInfo> {
+    ) -> Result<HashMap<StorePath, UnkeyedValidPathInfo>, Error> {
         self.base.query_path_infos(paths).await
     }
 
     #[inline]
-    async fn compute_closure_size(&self, path: &StorePath) -> u64 {
+    async fn compute_closure_size(&self, path: &StorePath) -> Result<u64, Error> {
         self.base.compute_closure_size(path).await
     }
 
@@ -907,7 +908,7 @@ impl RemoteStore {
 
         tokio_stream::iter(paths)
             .map(|p| async move {
-                if self.is_valid_path(&p).await {
+                if self.is_valid_path(&p).await.unwrap_or(false) {
                     None
                 } else {
                     Some(p)
@@ -929,7 +930,7 @@ impl RemoteStore {
         tokio_stream::iter(outputs)
             .map(|(name, path)| async move {
                 match path {
-                    Some(p) if self.is_valid_path(&p).await => None,
+                    Some(p) if self.is_valid_path(&p).await.unwrap_or(false) => None,
                     other => Some((name, other)),
                 }
             })
@@ -942,25 +943,28 @@ impl RemoteStore {
 
 impl BaseStore for RemoteStore {
     #[inline]
-    async fn is_valid_path(&self, path: &StorePath) -> bool {
+    async fn is_valid_path(&self, path: &StorePath) -> Result<bool, Error> {
         self.base.is_valid_path(path).await
     }
 
     #[inline]
-    async fn query_path_info(&self, path: &StorePath) -> Option<UnkeyedValidPathInfo> {
+    async fn query_path_info(
+        &self,
+        path: &StorePath,
+    ) -> Result<Option<UnkeyedValidPathInfo>, Error> {
         self.base.query_path_info(path).await
     }
 
     #[inline]
     async fn query_path_infos(
         &self,
         paths: &[&StorePath],
-    ) -> HashMap<StorePath, UnkeyedValidPathInfo> {
+    ) -> Result<HashMap<StorePath, UnkeyedValidPathInfo>, Error> {
         self.base.query_path_infos(paths).await
     }
 
     #[inline]
-    async fn compute_closure_size(&self, path: &StorePath) -> u64 {
+    async fn compute_closure_size(&self, path: &StorePath) -> Result<u64, Error> {
         self.base.compute_closure_size(path).await
     }
 

subprojects/crates/nix-utils/src/nix.cpp

@@ -113,7 +113,11 @@ bool is_valid_path(const StoreWrapper &wrapper, rust::Str path) {
 
 InternalPathInfo query_path_info(const StoreWrapper &wrapper, rust::Str path) {
   auto store = wrapper._store;
-  auto info = store->queryPathInfo(store->parseStorePath(AS_VIEW(path)));
+  auto storePath = store->parseStorePath(AS_VIEW(path));
+  if (!store->isValidPath(storePath)) {
+    return InternalPathInfo{false, "", "", {}, 0, 0, {}, {}, ""};
+  }
+  auto info = store->queryPathInfo(storePath);
 
   // Return raw SHA256 bytes (32 bytes) — Rust side converts to NarHash directly.
   rust::Vec<uint8_t> narhash_bytes;
@@ -133,6 +137,7 @@ InternalPathInfo query_path_info(const StoreWrapper &wrapper, rust::Str path) {
 
   // TODO(conni2461): Replace "" with option
   return InternalPathInfo{
+      true,
       rust::String(store->storeDir.data(), store->storeDir.size()),
       extract_opt_path(info->deriver),
       narhash_bytes,