hive-c0re pulls closures through the nix daemon; hive-priv verifies the swarm signature before applying
This commit is contained in:
atlas
mara
parent
448b703995
commit
ecd4445680
4 changed files with 1257 additions and 584 deletions
484
hive-c0re/src/cache_pull.rs
Normal file
484
hive-c0re/src/cache_pull.rs
Normal file
|
|
@ -0,0 +1,484 @@
|
|||
//! Pull the swarm's closure into the store through the host's nix daemon.
|
||||
//!
|
||||
//! hive-c0re is not root and does not need to be: `nix-store --realise` of a
|
||||
//! store path is a request to the daemon, which substitutes it. What the daemon
|
||||
//! accepts is root's: the root unit `hive-priv-cache-facts` writes a nix.conf
|
||||
//! fragment ([`CONF_FRAGMENT`]) that lists the swarm cache as a
|
||||
//! `trusted-substituter`, adds its key to `trusted-public-keys` and names the
|
||||
//! netrc. An unprivileged client may name a substituter only from that list,
|
||||
//! cannot lower `require-sigs` or replace the keys, and a path without a
|
||||
//! signature by a trusted key is refused. hive-priv then applies the toplevel
|
||||
//! only if the swarm's key signed it.
|
||||
//!
|
||||
//! The cache's address is read from the fragment, so the address named is
|
||||
//! byte for byte the one the daemon trusts. A missing fragment means this host
|
||||
//! has not been given its cache facts; [`pull`] first asks hive-priv to run the
|
||||
//! facts unit, which is how the swarm's copy of the facts reaches the host
|
||||
//! without waiting for the unit's timer.
|
||||
|
||||
use std::time::Duration;
|
||||
|
||||
use anyhow::Result;
|
||||
use hive_priv_sock::valid_closure_path;
|
||||
|
||||
/// The nix.conf fragment `hive-priv-cache-facts` writes (root-owned, readable).
|
||||
pub const CONF_FRAGMENT: &str = "/run/hyperhive-nix/cache.conf";
|
||||
|
||||
/// The fragment's line that lists the cache the daemon lets a client name.
|
||||
const URL_SETTING: &str = "extra-trusted-substituters";
|
||||
|
||||
/// Wall-clock bound on one pull. A stalled transfer is bounded by nix's own
|
||||
/// `stalled-download-timeout`; this covers the rest.
|
||||
const PULL_TIMEOUT: Duration = Duration::from_mins(30);
|
||||
|
||||
/// Why a closure was not pulled.
|
||||
#[derive(Debug, PartialEq, Eq)]
|
||||
pub enum PullError {
|
||||
/// This host has no cache facts: no fragment, no address in it, or a
|
||||
/// daemon that has not loaded it. Waiting for the facts unit fixes it.
|
||||
FactsMissing(String),
|
||||
/// Nix refused the path: no signature by a key this host trusts.
|
||||
SignatureRefused(String),
|
||||
/// The cache does not have the path.
|
||||
NotFound(String),
|
||||
/// Anything else nix said, or could not do.
|
||||
Failed(String),
|
||||
/// Nix was still running at [`PULL_TIMEOUT`].
|
||||
TimedOut,
|
||||
}
|
||||
|
||||
impl std::fmt::Display for PullError {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
match self {
|
||||
Self::FactsMissing(why) => write!(
|
||||
f,
|
||||
"this hive has no cache facts yet ({why}); a later deploy retries"
|
||||
),
|
||||
Self::SignatureRefused(last) => write!(
|
||||
f,
|
||||
"nix refused the closure, which carries no signature by a key this host trusts: {last}"
|
||||
),
|
||||
Self::NotFound(last) => write!(f, "the swarm cache does not have the closure: {last}"),
|
||||
Self::Failed(last) => f.write_str(last),
|
||||
Self::TimedOut => write!(f, "nix-store timed out after {}s", PULL_TIMEOUT.as_secs()),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl std::error::Error for PullError {}
|
||||
|
||||
/// What a `nix-store` run came to.
|
||||
pub enum Run {
|
||||
Exited { success: bool, stderr: String },
|
||||
TimedOut,
|
||||
}
|
||||
|
||||
/// What [`pull`] does on the host, so a test can stand in for it.
|
||||
pub trait Effects: Sync {
|
||||
/// The fragment's text; `None` when it cannot be read.
|
||||
async fn fragment(&self) -> Option<String>;
|
||||
/// Run the facts unit through hive-priv and wait for it.
|
||||
async fn refresh(&self) -> Result<()>;
|
||||
/// Run `nix-store` with `args`, bounded by [`PULL_TIMEOUT`].
|
||||
async fn nix_store(&self, args: Vec<String>) -> std::io::Result<Run>;
|
||||
}
|
||||
|
||||
/// The cache's address in `fragment`: the value of its
|
||||
/// `extra-trusted-substituters` line, when that is one https address.
|
||||
#[must_use]
|
||||
pub fn cache_url(fragment: &str) -> Option<&str> {
|
||||
fragment.lines().find_map(|line| {
|
||||
let (name, value) = line.split_once('=')?;
|
||||
let value = value.trim();
|
||||
(name.trim() == URL_SETTING
|
||||
&& value.starts_with("https://")
|
||||
&& !value.contains(char::is_whitespace))
|
||||
.then_some(value)
|
||||
})
|
||||
}
|
||||
|
||||
/// The `nix-store` arguments that fetch `store_path` and nothing else.
|
||||
///
|
||||
/// - `--option substituters <url>` names the one cache the daemon lists as
|
||||
/// trusted. An unprivileged client's `substituters` that is not on that list
|
||||
/// empties the list and fails the pull, so the address must be the fragment's.
|
||||
/// - `--max-jobs 0` forbids a local build, so a path the cache lacks fails.
|
||||
/// - No `trusted-public-keys`, `require-sigs` or `netrc-file`: the daemon
|
||||
/// ignores them from an unprivileged client and applies its own.
|
||||
#[must_use]
|
||||
pub fn pull_args(store_path: &str, url: &str) -> Vec<String> {
|
||||
[
|
||||
"--realise",
|
||||
store_path,
|
||||
"--max-jobs",
|
||||
"0",
|
||||
"--option",
|
||||
"substituters",
|
||||
url,
|
||||
]
|
||||
.map(str::to_owned)
|
||||
.into()
|
||||
}
|
||||
|
||||
/// Name nix's refusal from its stderr. A refusal for want of a signature
|
||||
/// or a daemon that does not trust the address is told apart from a path the
|
||||
/// cache lacks and from a cache that is down.
|
||||
fn classify(stderr: &str) -> PullError {
|
||||
let last = stderr
|
||||
.lines()
|
||||
.rev()
|
||||
.find(|l| !l.trim().is_empty())
|
||||
.unwrap_or("nix-store failed with no message")
|
||||
.trim()
|
||||
.to_owned();
|
||||
let lower = stderr.to_lowercase();
|
||||
if lower.contains("untrusted substituter") {
|
||||
PullError::FactsMissing(
|
||||
"the nix daemon does not list the swarm cache as a trusted substituter".to_owned(),
|
||||
)
|
||||
} else if lower.contains("not signed by any of the keys") || lower.contains("signature") {
|
||||
PullError::SignatureRefused(last)
|
||||
} else if lower.contains("don't know how to build")
|
||||
|| lower.contains("no substituter that can build")
|
||||
{
|
||||
PullError::NotFound(last)
|
||||
} else {
|
||||
PullError::Failed(last)
|
||||
}
|
||||
}
|
||||
|
||||
/// Pull `store_path` into the store.
|
||||
///
|
||||
/// # Errors
|
||||
/// [`PullError`]: no cache facts on this host (after one refresh), nix
|
||||
/// refusing the signature, the cache lacking the path, nix failing otherwise,
|
||||
/// or a timeout.
|
||||
pub async fn pull(store_path: &str) -> Result<(), PullError> {
|
||||
pull_with(&Real, store_path).await
|
||||
}
|
||||
|
||||
async fn pull_with(fx: &impl Effects, store_path: &str) -> Result<(), PullError> {
|
||||
if !valid_closure_path(store_path) {
|
||||
return Err(PullError::Failed(format!(
|
||||
"{store_path:?} is not a store path"
|
||||
)));
|
||||
}
|
||||
let url = address_after_refresh(fx).await?;
|
||||
match fx
|
||||
.nix_store(pull_args(store_path, &url))
|
||||
.await
|
||||
.map_err(|e| PullError::Failed(format!("running nix-store: {e}")))?
|
||||
{
|
||||
Run::Exited { success: true, .. } => Ok(()),
|
||||
Run::Exited { stderr, .. } => Err(classify(&stderr)),
|
||||
Run::TimedOut => Err(PullError::TimedOut),
|
||||
}
|
||||
}
|
||||
|
||||
/// The cache's address, after one facts refresh when the host has none.
|
||||
async fn address_after_refresh(fx: &impl Effects) -> Result<String, PullError> {
|
||||
if let Some(url) = address(fx).await {
|
||||
return Ok(url);
|
||||
}
|
||||
let refreshed = fx.refresh().await;
|
||||
address(fx).await.ok_or_else(|| {
|
||||
PullError::FactsMissing(match refreshed {
|
||||
Ok(()) => "the facts unit ran and the swarm has given this hive none".to_owned(),
|
||||
Err(e) => format!("refreshing them failed: {e:#}"),
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
async fn address(fx: &impl Effects) -> Option<String> {
|
||||
fx.fragment()
|
||||
.await
|
||||
.as_deref()
|
||||
.and_then(cache_url)
|
||||
.map(str::to_owned)
|
||||
}
|
||||
|
||||
/// The host: the fragment on disk, hive-priv for the refresh, `nix-store` from
|
||||
/// the system profile through the daemon.
|
||||
struct Real;
|
||||
|
||||
impl Effects for Real {
|
||||
async fn fragment(&self) -> Option<String> {
|
||||
tokio::fs::read_to_string(CONF_FRAGMENT).await.ok()
|
||||
}
|
||||
|
||||
async fn refresh(&self) -> Result<()> {
|
||||
crate::priv_client::refresh_cache_facts().await
|
||||
}
|
||||
|
||||
async fn nix_store(&self, args: Vec<String>) -> std::io::Result<Run> {
|
||||
let mut cmd = tokio::process::Command::new("nix-store");
|
||||
// Explicit, so the store is the daemon's whatever the environment.
|
||||
cmd.args(args)
|
||||
.env("NIX_REMOTE", "daemon")
|
||||
.stdin(std::process::Stdio::null())
|
||||
.stdout(std::process::Stdio::null())
|
||||
.kill_on_drop(true);
|
||||
match tokio::time::timeout(PULL_TIMEOUT, cmd.output()).await {
|
||||
Ok(out) => {
|
||||
let out = out?;
|
||||
Ok(Run::Exited {
|
||||
success: out.status.success(),
|
||||
stderr: String::from_utf8_lossy(&out.stderr).into_owned(),
|
||||
})
|
||||
}
|
||||
Err(_) => Ok(Run::TimedOut),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use std::sync::Mutex;
|
||||
|
||||
use super::*;
|
||||
|
||||
const PATH: &str = "/nix/store/0123456789abcdfghijklmnpqrsvwxyz-nixos-system-atlas";
|
||||
const URL: &str = "https://cache.example/swarm";
|
||||
|
||||
fn fragment(url: &str) -> String {
|
||||
format!(
|
||||
"extra-trusted-substituters = {url}\nextra-trusted-public-keys = swarm:AAAA\nnetrc-file = /run/hyperhive-priv/attic/netrc"
|
||||
)
|
||||
}
|
||||
|
||||
/// A host that answers from a script: the fragment as it is now (and the
|
||||
/// one a refresh leaves), what `nix-store` comes to, and a log of calls.
|
||||
struct Fake {
|
||||
fragment: Mutex<Option<String>>,
|
||||
after_refresh: Option<String>,
|
||||
refresh_fails: bool,
|
||||
nix: Mutex<Option<Run>>,
|
||||
calls: Mutex<Vec<String>>,
|
||||
}
|
||||
|
||||
impl Fake {
|
||||
fn new(fragment: Option<String>, nix: Run) -> Self {
|
||||
Self {
|
||||
fragment: Mutex::new(fragment),
|
||||
after_refresh: None,
|
||||
refresh_fails: false,
|
||||
nix: Mutex::new(Some(nix)),
|
||||
calls: Mutex::new(Vec::new()),
|
||||
}
|
||||
}
|
||||
|
||||
fn calls(&self) -> Vec<String> {
|
||||
self.calls.lock().unwrap().clone()
|
||||
}
|
||||
}
|
||||
|
||||
impl Effects for Fake {
|
||||
async fn fragment(&self) -> Option<String> {
|
||||
self.calls.lock().unwrap().push("fragment".to_owned());
|
||||
self.fragment.lock().unwrap().clone()
|
||||
}
|
||||
|
||||
async fn refresh(&self) -> Result<()> {
|
||||
self.calls.lock().unwrap().push("refresh".to_owned());
|
||||
if self.refresh_fails {
|
||||
anyhow::bail!("hive-priv unreachable");
|
||||
}
|
||||
*self.fragment.lock().unwrap() = self.after_refresh.clone();
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn nix_store(&self, args: Vec<String>) -> std::io::Result<Run> {
|
||||
self.calls
|
||||
.lock()
|
||||
.unwrap()
|
||||
.push(format!("nix-store {}", args.join(" ")));
|
||||
Ok(self.nix.lock().unwrap().take().expect("one run"))
|
||||
}
|
||||
}
|
||||
|
||||
fn exited(success: bool, stderr: &str) -> Run {
|
||||
Run::Exited {
|
||||
success,
|
||||
stderr: stderr.to_owned(),
|
||||
}
|
||||
}
|
||||
|
||||
fn option<'a>(args: &'a [String], name: &str) -> Option<&'a str> {
|
||||
args.windows(3)
|
||||
.find(|w| w[0] == "--option" && w[1] == name)
|
||||
.map(|w| w[2].as_str())
|
||||
}
|
||||
|
||||
/// An unprivileged client can name only the trusted address and cannot set
|
||||
/// keys, signatures or the netrc, so the command carries exactly that, and
|
||||
/// never lets nix build.
|
||||
#[test]
|
||||
fn the_command_names_only_the_swarm_cache_and_never_builds() {
|
||||
let args = pull_args(PATH, URL);
|
||||
assert_eq!(args[..2], ["--realise", PATH]);
|
||||
let jobs = args.iter().position(|a| a == "--max-jobs").expect("flag");
|
||||
assert_eq!(args[jobs + 1], "0");
|
||||
assert_eq!(option(&args, "substituters"), Some(URL));
|
||||
let options: Vec<_> = args.iter().filter(|a| *a == "--option").collect();
|
||||
assert_eq!(options.len(), 1, "{args:?}");
|
||||
for forbidden in ["trusted-public-keys", "require-sigs", "netrc-file"] {
|
||||
assert_eq!(option(&args, forbidden), None, "{forbidden}");
|
||||
}
|
||||
assert!(!args.iter().any(|a| a.starts_with("extra-")), "{args:?}");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn the_address_is_the_fragments_https_line() {
|
||||
assert_eq!(cache_url(&fragment(URL)), Some(URL));
|
||||
for bad in [
|
||||
String::new(),
|
||||
"netrc-file = /x".to_owned(),
|
||||
fragment("http://cache.example/swarm"),
|
||||
fragment("https://cache.example/a b"),
|
||||
fragment(""),
|
||||
// The key's line is not the address's.
|
||||
"extra-trusted-public-keys = https://cache.example/swarm".to_owned(),
|
||||
] {
|
||||
assert_eq!(cache_url(&bad), None, "{bad:?}");
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn with_the_fragment_in_place_it_pulls_without_a_refresh() {
|
||||
let fx = Fake::new(Some(fragment(URL)), exited(true, ""));
|
||||
pull_with(&fx, PATH).await.expect("pulled");
|
||||
assert_eq!(
|
||||
fx.calls(),
|
||||
[
|
||||
"fragment".to_owned(),
|
||||
format!("nix-store {}", pull_args(PATH, URL).join(" "))
|
||||
]
|
||||
);
|
||||
}
|
||||
|
||||
/// Missing facts are refreshed once, and the pull then goes ahead with the
|
||||
/// address the refresh wrote.
|
||||
#[tokio::test]
|
||||
async fn a_missing_fragment_is_refreshed_before_the_pull() {
|
||||
let mut fx = Fake::new(None, exited(true, ""));
|
||||
fx.after_refresh = Some(fragment(URL));
|
||||
pull_with(&fx, PATH).await.expect("pulled");
|
||||
assert_eq!(
|
||||
fx.calls(),
|
||||
[
|
||||
"fragment".to_owned(),
|
||||
"refresh".to_owned(),
|
||||
"fragment".to_owned(),
|
||||
format!("nix-store {}", pull_args(PATH, URL).join(" "))
|
||||
]
|
||||
);
|
||||
}
|
||||
|
||||
/// Facts that are still absent after the refresh, whether the unit ran or
|
||||
/// hive-priv could not start it, stop the pull before nix is asked.
|
||||
#[tokio::test]
|
||||
async fn facts_still_missing_after_the_refresh_are_reported_and_nix_is_not_run() {
|
||||
for (case, refresh_fails, want) in [
|
||||
("unit ran", false, "none"),
|
||||
("refresh failed", true, "hive-priv unreachable"),
|
||||
] {
|
||||
let mut fx = Fake::new(None, exited(true, ""));
|
||||
fx.refresh_fails = refresh_fails;
|
||||
let err = pull_with(&fx, PATH).await.expect_err(case);
|
||||
let PullError::FactsMissing(why) = &err else {
|
||||
panic!("{case}: {err:?}");
|
||||
};
|
||||
assert!(why.contains(want), "{case}: {why}");
|
||||
assert!(
|
||||
format!("{err}").contains("a later deploy retries"),
|
||||
"{case}"
|
||||
);
|
||||
assert!(
|
||||
!fx.calls().iter().any(|c| c.starts_with("nix-store")),
|
||||
"{case}"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_fragment_without_an_address_is_refreshed_like_a_missing_one() {
|
||||
let mut fx = Fake::new(Some("netrc-file = /x".to_owned()), exited(true, ""));
|
||||
fx.after_refresh = Some(fragment(URL));
|
||||
pull_with(&fx, PATH).await.expect("pulled");
|
||||
assert!(fx.calls().contains(&"refresh".to_owned()));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn nixs_refusals_are_told_apart() {
|
||||
for (stderr, want) in [
|
||||
(
|
||||
"warning: ignoring substitute for '/nix/store/x' from 'https://c', as it's not signed by any of the keys in 'trusted-public-keys'\nerror: build of '/nix/store/x' failed",
|
||||
"signature",
|
||||
),
|
||||
(
|
||||
"error: cannot add path 'x' because it lacks a signature by a trusted key",
|
||||
"signature",
|
||||
),
|
||||
(
|
||||
"warning: ignoring untrusted substituter 'https://c', you are not a trusted user.\ndon't know how to build these paths:\nerror: build of 'x' failed",
|
||||
"facts",
|
||||
),
|
||||
(
|
||||
"don't know how to build these paths:\n /nix/store/x\nerror: build of '/nix/store/x' failed",
|
||||
"not found",
|
||||
),
|
||||
(
|
||||
"error: path '/nix/store/x' is required, but there is no substituter that can build it",
|
||||
"not found",
|
||||
),
|
||||
(
|
||||
"error: unable to download 'https://c/x.narinfo': Couldn't connect to server",
|
||||
"other",
|
||||
),
|
||||
] {
|
||||
let kind = match classify(stderr) {
|
||||
PullError::SignatureRefused(_) => "signature",
|
||||
PullError::FactsMissing(_) => "facts",
|
||||
PullError::NotFound(_) => "not found",
|
||||
PullError::Failed(_) => "other",
|
||||
PullError::TimedOut => "timeout",
|
||||
};
|
||||
assert_eq!(kind, want, "{stderr}");
|
||||
}
|
||||
// The last line is what the operator reads.
|
||||
let PullError::Failed(last) = classify("warning: a\n\nerror: Couldn't connect\n\n") else {
|
||||
panic!("not a plain failure");
|
||||
};
|
||||
assert_eq!(last, "error: Couldn't connect");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_failed_run_is_mapped_and_a_timeout_is_a_timeout() {
|
||||
let fx = Fake::new(
|
||||
Some(fragment(URL)),
|
||||
exited(
|
||||
false,
|
||||
"warning: ignoring substitute from 'https://c', as it's not signed by any of the keys",
|
||||
),
|
||||
);
|
||||
assert!(matches!(
|
||||
pull_with(&fx, PATH).await,
|
||||
Err(PullError::SignatureRefused(_))
|
||||
));
|
||||
let fx = Fake::new(Some(fragment(URL)), Run::TimedOut);
|
||||
assert_eq!(pull_with(&fx, PATH).await, Err(PullError::TimedOut));
|
||||
}
|
||||
|
||||
/// Nothing but a store path is ever handed to nix.
|
||||
#[tokio::test]
|
||||
async fn only_a_store_path_is_pulled() {
|
||||
for bad in ["", "main", "/etc/passwd", "--help", "/nix/store/x"] {
|
||||
let fx = Fake::new(Some(fragment(URL)), exited(true, ""));
|
||||
assert!(
|
||||
matches!(pull_with(&fx, bad).await, Err(PullError::Failed(_))),
|
||||
"{bad}"
|
||||
);
|
||||
assert!(fx.calls().is_empty(), "{bad}");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -51,8 +51,8 @@ pub(super) trait Host: Sync {
|
|||
fn pinned_closure(&self, name: &str) -> Option<String>;
|
||||
/// Drop the agent's pin.
|
||||
fn unpin_closure(&self, name: &str);
|
||||
/// Download `store_path` from the swarm's cache into the store, never
|
||||
/// building it.
|
||||
/// Download `store_path` from the swarm's cache into the store through the
|
||||
/// host's nix daemon, never building it.
|
||||
async fn realise_closure(&self, name: &str, store_path: &str) -> Result<()>;
|
||||
/// Re-apply drop-ins, then point the container at the closure `toplevel`.
|
||||
async fn swap_update(
|
||||
|
|
@ -128,7 +128,9 @@ impl Host for RealHost {
|
|||
}
|
||||
|
||||
async fn realise_closure(&self, name: &str, store_path: &str) -> Result<()> {
|
||||
crate::priv_client::realise_closure(name, store_path).await
|
||||
crate::cache_pull::pull(store_path).await?;
|
||||
tracing::debug!(%name, %store_path, "pulled the swarm's closure through the nix daemon");
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn swap_update(
|
||||
|
|
|
|||
|
|
@ -17,6 +17,7 @@ mod actions;
|
|||
mod agent_config;
|
||||
mod agent_set;
|
||||
mod attic;
|
||||
mod cache_pull;
|
||||
mod container_view;
|
||||
mod coordinator;
|
||||
mod dashboard;
|
||||
|
|
|
|||
|
|
@ -267,9 +267,15 @@ pub async fn kill_container(name: &str) -> Result<()> {
|
|||
pub async fn update_container_streaming(
|
||||
name: &str,
|
||||
toplevel: &str,
|
||||
on_line: impl FnMut(PrivStream, &str),
|
||||
mut on_line: impl FnMut(PrivStream, &str),
|
||||
) -> Result<()> {
|
||||
ok(call_streaming(&update_container_request(name, toplevel), on_line).await?)
|
||||
let req = update_container_request(name, toplevel);
|
||||
let first = ok(call_streaming(&req, &mut on_line).await?);
|
||||
if !older_hive_priv_refused(&first) {
|
||||
return first;
|
||||
}
|
||||
ask_older_hive_priv_to_download(name, toplevel).await?;
|
||||
ok(call_streaming(&req, &mut on_line).await?)
|
||||
}
|
||||
|
||||
fn update_container_request(name: &str, toplevel: &str) -> PrivRequest {
|
||||
|
|
@ -285,9 +291,15 @@ fn update_container_request(name: &str, toplevel: &str) -> PrivRequest {
|
|||
pub async fn create_container_streaming(
|
||||
name: &str,
|
||||
toplevel: &str,
|
||||
on_line: impl FnMut(PrivStream, &str),
|
||||
mut on_line: impl FnMut(PrivStream, &str),
|
||||
) -> Result<()> {
|
||||
ok(call_streaming(&create_container_request(name, toplevel), on_line).await?)
|
||||
let req = create_container_request(name, toplevel);
|
||||
let first = ok(call_streaming(&req, &mut on_line).await?);
|
||||
if !older_hive_priv_refused(&first) {
|
||||
return first;
|
||||
}
|
||||
ask_older_hive_priv_to_download(name, toplevel).await?;
|
||||
ok(call_streaming(&req, &mut on_line).await?)
|
||||
}
|
||||
|
||||
fn create_container_request(name: &str, toplevel: &str) -> PrivRequest {
|
||||
|
|
@ -298,17 +310,49 @@ fn create_container_request(name: &str, toplevel: &str) -> PrivRequest {
|
|||
}
|
||||
}
|
||||
|
||||
/// See [`PrivRequest::RealiseClosure`]. A path hive-priv could not download
|
||||
/// reads as an error whose message starts `not available: `.
|
||||
///
|
||||
/// # Errors
|
||||
/// The call failed, or hive-priv answered `not available`.
|
||||
pub async fn realise_closure(agent: &str, store_path: &str) -> Result<()> {
|
||||
/// What an older hive-priv, which applies only a path it downloaded itself,
|
||||
/// says to an apply of any other.
|
||||
const OLDER_PIN_REFUSAL: &str = "is not the closure realised for this agent";
|
||||
|
||||
/// Whether `applied` is an older hive-priv refusing an apply because it did not
|
||||
/// download the path itself (hive-c0re pulled it through the nix daemon). A
|
||||
/// hive-priv of the same release never refuses that way, so a retry through
|
||||
/// [`ask_older_hive_priv_to_download`] happens only while the two restart out of
|
||||
/// step.
|
||||
fn older_hive_priv_refused(applied: &Result<()>) -> bool {
|
||||
applied
|
||||
.as_ref()
|
||||
.is_err_and(|e| format!("{e:#}").contains(OLDER_PIN_REFUSAL))
|
||||
}
|
||||
|
||||
/// Ask an older hive-priv to download `toplevel`, as it did before, which also
|
||||
/// pins the path for the apply that follows. The path is already in the store,
|
||||
/// so this is a lookup.
|
||||
async fn ask_older_hive_priv_to_download(name: &str, toplevel: &str) -> Result<()> {
|
||||
tracing::warn!(%name, %toplevel, "an older hive-priv refused a closure it did not download; asking it to");
|
||||
ok(call(&PrivRequest::RealiseClosure {
|
||||
agent: agent.to_owned(),
|
||||
store_path: store_path.to_owned(),
|
||||
agent: name.to_owned(),
|
||||
store_path: toplevel.to_owned(),
|
||||
})
|
||||
.await?)
|
||||
.context("asking the older hive-priv to download the closure")
|
||||
}
|
||||
|
||||
/// How long hive-c0re waits for hive-priv's facts refresh: hive-priv bounds
|
||||
/// the unit at 60 s.
|
||||
const REFRESH_WAIT: std::time::Duration = std::time::Duration::from_secs(90);
|
||||
|
||||
/// Run the root unit that reads this hive's cache facts from the store and
|
||||
/// writes the nix.conf fragment, and wait for it. An older hive-priv does not
|
||||
/// know the op and answers an error.
|
||||
///
|
||||
/// # Errors
|
||||
/// The call failed or did not answer in time, or the unit failed.
|
||||
pub async fn refresh_cache_facts() -> Result<()> {
|
||||
within(REFRESH_WAIT, async {
|
||||
ok(call(&PrivRequest::RefreshCacheFacts).await?)
|
||||
})
|
||||
.await
|
||||
}
|
||||
|
||||
pub async fn destroy_container(name: &str) -> Result<()> {
|
||||
|
|
@ -684,9 +728,11 @@ mod tests {
|
|||
use super::{
|
||||
BindMount, CredentialMount, HiveFacts, NetworkIsolation, NixBuilds, PrivRequest,
|
||||
PrivResponse, PrivStream, call_at, call_streaming_at, call_with_fd_at, check,
|
||||
create_container_request, parse_qgroup_usage, resource_limits_request,
|
||||
send_snapshot_request, update_container_request, within, write_nspawn_flags_request,
|
||||
create_container_request, older_hive_priv_refused, parse_qgroup_usage,
|
||||
resource_limits_request, send_snapshot_request, update_container_request, within,
|
||||
write_nspawn_flags_request,
|
||||
};
|
||||
use anyhow::Context as _;
|
||||
use serde_json::{Value, json};
|
||||
use std::io::{Read as _, Write as _};
|
||||
use std::os::fd::{AsRawFd as _, FromRawFd as _, OwnedFd, RawFd};
|
||||
|
|
@ -697,6 +743,38 @@ mod tests {
|
|||
serde_json::to_value(req).expect("a request serialises")
|
||||
}
|
||||
|
||||
const REFUSAL: &str = "atlas: /nix/store/x is not the closure realised for this agent (none)";
|
||||
|
||||
/// An older hive-priv's refusal of a path it did not download is the
|
||||
/// signal to ask it to download; a success and any other failure are not.
|
||||
#[test]
|
||||
fn only_an_older_hive_privs_pin_refusal_asks_for_a_download() {
|
||||
assert!(older_hive_priv_refused(&Err(anyhow::anyhow!(REFUSAL))));
|
||||
// As the streaming call words it: the error wraps hive-priv's text.
|
||||
assert!(older_hive_priv_refused(
|
||||
&Err(anyhow::anyhow!(REFUSAL)).context("nixos-container create")
|
||||
));
|
||||
assert!(!older_hive_priv_refused(&Ok(())));
|
||||
for other in [
|
||||
"nixos-container create failed",
|
||||
"atlas: refusing /nix/store/x: has no signature by the swarm cache's key",
|
||||
"connect to hive-priv socket",
|
||||
] {
|
||||
assert!(
|
||||
!older_hive_priv_refused(&Err(anyhow::anyhow!(other))),
|
||||
"{other}"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn the_cache_facts_refresh_names_no_unit() {
|
||||
assert_eq!(
|
||||
wire(&PrivRequest::RefreshCacheFacts),
|
||||
json!({"op": "refresh_cache_facts"})
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn create_and_update_keep_the_name_and_toplevel_apart_and_stream() {
|
||||
assert_eq!(
|
||||
|
|
|
|||
Loading…
Reference in a new issue