various: fix closure narinfo cache

Cargo.toml

@@ -138,6 +138,7 @@ harmonia-store-nar-info        = { git = "https://github.com/nix-community/harmo
 harmonia-store-path            = { git = "https://github.com/nix-community/harmonia.git", rev = "12c6742560dd1ba1e66f5cc2f04cabd4e99ae754" }
 harmonia-store-path-info       = { git = "https://github.com/nix-community/harmonia.git", rev = "12c6742560dd1ba1e66f5cc2f04cabd4e99ae754" }
 harmonia-utils-hash            = { git = "https://github.com/nix-community/harmonia.git", rev = "12c6742560dd1ba1e66f5cc2f04cabd4e99ae754" }
+harmonia-utils-signature       = { git = "https://github.com/nix-community/harmonia.git", rev = "12c6742560dd1ba1e66f5cc2f04cabd4e99ae754" }
 
 # evix, in-process Nix evaluator (nix-eval-jobs replacement) driven as a library.
 evix = { version = "0.3.3", features = [ "flake" ] }

crates/common/Cargo.toml

@@ -6,40 +6,41 @@ license.workspace    = true
 repository.workspace = true
 
 [dependencies]
-argon2.workspace            = true
-base64.workspace            = true
-chrono.workspace            = true
-circus-config.workspace     = true
-circus-logs.workspace       = true
-circus-migrations.workspace = true
-circus-nix.workspace        = true
-circus-types                = { workspace = true, features = [ "sqlx" ] }
-clap.workspace              = true
-color-eyre.workspace        = true
-git2.workspace              = true
-hex.workspace               = true
-lettre.workspace            = true
-libc.workspace              = true
-nix.workspace               = true
-num_enum.workspace          = true
-password-hash.workspace     = true
-regex.workspace             = true
-reqwest.workspace           = true
-ring.workspace              = true
-rustls.workspace            = true
-serde.workspace             = true
-serde_json.workspace        = true
-sha2.workspace              = true
-shellexpand.workspace       = true
-sqlx.workspace              = true
-tempfile.workspace          = true
-thiserror.workspace         = true
-tokio.workspace             = true
-tracing.workspace           = true
-url.workspace               = true
-urlencoding.workspace       = true
-uuid.workspace              = true
-whoami.workspace            = true
+argon2.workspace                   = true
+base64.workspace                   = true
+chrono.workspace                   = true
+circus-config.workspace            = true
+circus-logs.workspace              = true
+circus-migrations.workspace        = true
+circus-nix.workspace               = true
+circus-types                       = { workspace = true, features = [ "sqlx" ] }
+clap.workspace                     = true
+color-eyre.workspace               = true
+git2.workspace                     = true
+harmonia-utils-signature.workspace = true
+hex.workspace                      = true
+lettre.workspace                   = true
+libc.workspace                     = true
+nix.workspace                      = true
+num_enum.workspace                 = true
+password-hash.workspace            = true
+regex.workspace                    = true
+reqwest.workspace                  = true
+ring.workspace                     = true
+rustls.workspace                   = true
+serde.workspace                    = true
+serde_json.workspace               = true
+sha2.workspace                     = true
+shellexpand.workspace              = true
+sqlx.workspace                     = true
+tempfile.workspace                 = true
+thiserror.workspace                = true
+tokio.workspace                    = true
+tracing.workspace                  = true
+url.workspace                      = true
+urlencoding.workspace              = true
+uuid.workspace                     = true
+whoami.workspace                   = true
 
 [dev-dependencies]
 toml.workspace = true

crates/common/src/lib.rs

@@ -10,6 +10,7 @@ pub mod log_storage;
 pub mod migrate;
 pub mod migrate_cli;
 pub mod models;
+pub mod narinfo_signing;
 pub mod pg_notify;
 pub mod psi;
 pub mod repo;

crates/common/src/narinfo_signing.rs

@@ -0,0 +1,42 @@
+//! Shared Nix narinfo fingerprint signing.
+
+use std::path::Path;
+
+use color_eyre::eyre::{Context as _, Result, eyre};
+use harmonia_utils_signature::SecretKey;
+
+/// Read a Nix-format `<name>:<base64 secret>` signing key from disk.
+///
+/// # Errors
+///
+/// Returns an error when the file cannot be read, or the key is malformed or
+/// its public half does not match its seed.
+pub async fn read_signing_key(path: &Path) -> Result<SecretKey> {
+  let raw = tokio::fs::read_to_string(path)
+    .await
+    .with_context(|| format!("read signing key {}", path.display()))?;
+  raw
+    .trim()
+    .parse::<SecretKey>()
+    .map_err(|e| eyre!("invalid signing key: {e:?}"))
+}
+
+/// Sign the canonical Nix narinfo fingerprint, returning
+/// `<name>:<base64 signature>`. References are signed as a sorted set to match
+/// Nix's verification.
+#[must_use]
+pub fn sign_narinfo(
+  key: &SecretKey,
+  store_path: &str,
+  nar_hash: &str,
+  nar_size: i64,
+  references: &[String],
+) -> String {
+  let mut sorted_refs = references.to_vec();
+  sorted_refs.sort();
+  let fingerprint = format!(
+    "1;{store_path};{nar_hash};{nar_size};{}",
+    sorted_refs.join(",")
+  );
+  key.sign(fingerprint.as_bytes()).to_string()
+}

crates/queue-runner/src/rpc/server.rs

@@ -25,7 +25,7 @@ use circus_proto::{
   result_sink,
   runner,
 };
-use color_eyre::eyre::{Context as _, bail, eyre};
+use color_eyre::eyre::{Context as _, bail};
 use sha2::{Digest as _, Sha256};
 use sqlx::PgPool;
 use subtle::ConstantTimeEq as _;
@@ -897,11 +897,7 @@ impl runner::Server for RunnerImpl {
   }
 }
 
-/// Sign a narinfo fingerprint with the Nix-format signing key on disk.
-///
-/// Nix key files are one line: `<key-name>:<base64 secret>`. The secret
-/// is a 64-byte concatenation of the Ed25519 seed and public key (the
-/// canonical libsodium "secret key" layout). Output is
+/// Sign a narinfo fingerprint with the on-disk Nix signing key, returning
 /// `<key-name>:<base64 signature>`.
 async fn sign_fingerprint(
   key_file: &std::path::Path,
@@ -910,35 +906,10 @@ async fn sign_fingerprint(
   nar_size: i64,
   references: &[String],
 ) -> color_eyre::Result<String> {
-  use base64::{Engine as _, engine::general_purpose::STANDARD as B64};
-  use ring::signature::Ed25519KeyPair;
-  let raw = tokio::fs::read_to_string(key_file)
-    .await
-    .with_context(|| format!("read signing key {}", key_file.display()))?;
-  let raw = raw.trim();
-  let (name, secret_b64) = raw
-    .split_once(':')
-    .ok_or_else(|| eyre!("signing key not in `name:base64` form"))?;
-  let secret = B64
-    .decode(secret_b64)
-    .with_context(|| "signing key base64 decode")?;
-  if secret.len() != 64 {
-    bail!("signing key has {} bytes, expected 64", secret.len());
-  }
-  // libsodium layout is `seed (32) || public key (32)`. ring wants the
-  // seed alone.
-  let key = Ed25519KeyPair::from_seed_unchecked(&secret[..32])
-    .map_err(|e| eyre!("ring rejected key seed: {e}"))?;
-  // Nix builds the fingerprint from a sorted StorePathSet and re-sorts the
-  // references when verifying so the signed order must match.
-  let mut sorted_refs = references.to_vec();
-  sorted_refs.sort();
-  let fingerprint = format!(
-    "1;{store_path};{nar_hash};{nar_size};{}",
-    sorted_refs.join(",")
-  );
-  let sig = key.sign(fingerprint.as_bytes());
-  Ok(format!("{name}:{}", B64.encode(sig.as_ref())))
+  let key = circus_common::narinfo_signing::read_signing_key(key_file).await?;
+  Ok(circus_common::narinfo_signing::sign_narinfo(
+    &key, store_path, nar_hash, nar_size, references,
+  ))
 }
 
 /// Map a compression algorithm name to the conventional NAR file extension.

crates/queue-runner/src/runner_loop.rs

@@ -2,7 +2,7 @@ use std::{sync::Arc, time::Duration};
 
 use circus_common::{
   error::Result as CiResult,
-  models::{BuildStatus, JobsetState},
+  models::{Build, BuildStatus, JobsetState},
   repo,
 };
 use circus_config::HotConfig;
@@ -96,6 +96,87 @@ async fn mark_build_done(
   .map(|_| ())
 }
 
+async fn publish_succeeded_output_closure(
+  pool: &PgPool,
+  worker_pool: &WorkerPool,
+  build_id: Uuid,
+  output_paths: &[String],
+) {
+  if output_paths.is_empty() {
+    return;
+  }
+  let project_id = match repo::builds::get(pool, build_id).await {
+    Ok(build) => {
+      if let Some((project, _)) = get_project_for_build(pool, &build).await {
+        Some(project.id)
+      } else {
+        tracing::warn!(
+          build_id = %build_id,
+          "could not resolve project for closure narinfo publication"
+        );
+        None
+      }
+    },
+    Err(e) => {
+      tracing::warn!(
+        build_id = %build_id,
+        "failed to reload build for closure narinfo publication: {e}"
+      );
+      None
+    },
+  };
+  worker_pool
+    .persist_closure_narinfos(build_id, output_paths, project_id)
+    .await;
+}
+
+async fn completed_build_output_paths(
+  pool: &PgPool,
+  build: &Build,
+) -> Vec<String> {
+  let mut output_paths = match repo::build_outputs::list_for_build(
+    pool, build.id,
+  )
+  .await
+  {
+    Ok(outputs) => {
+      outputs
+        .into_iter()
+        .filter_map(|output| output.path)
+        .collect::<Vec<_>>()
+    },
+    Err(e) => {
+      tracing::warn!(
+        build_id = %build.id,
+        "failed to load completed build outputs for closure narinfo publication: {e}"
+      );
+      Vec::new()
+    },
+  };
+
+  if output_paths.is_empty()
+    && let Some(outputs) =
+      build.outputs.as_ref().and_then(|value| value.as_object())
+  {
+    output_paths.extend(
+      outputs
+        .values()
+        .filter_map(serde_json::Value::as_str)
+        .map(ToOwned::to_owned),
+    );
+  }
+
+  if output_paths.is_empty()
+    && let Some(output_path) = &build.build_output_path
+  {
+    output_paths.push(output_path.clone());
+  }
+
+  output_paths.sort();
+  output_paths.dedup();
+  output_paths
+}
+
 /// Main queue runner loop. Polls for pending builds and dispatches them to
 /// workers.
 ///
@@ -256,6 +337,16 @@ pub async fn run(
                 .await
                 {
                   tracing::warn!(build_id = %build.id, "Failed to complete dedup build: {e}");
+                } else {
+                  let output_paths =
+                    completed_build_output_paths(&pool, &existing).await;
+                  publish_succeeded_output_closure(
+                    &pool,
+                    &worker_pool,
+                    build.id,
+                    &output_paths,
+                  )
+                  .await;
                 }
                 continue;
               },
@@ -303,6 +394,14 @@ pub async fn run(
               .await
               {
                 tracing::warn!(build_id = %build.id, "Failed to complete FOD build: {e}");
+              } else {
+                publish_succeeded_output_closure(
+                  &pool,
+                  &worker_pool,
+                  build.id,
+                  std::slice::from_ref(&output_path),
+                )
+                .await;
               }
               continue;
             }