a55d55e494
hive-c0re reads its cache facts from the store every five minutes and writes /run/hyperhive/attic/{netrc,substituter,public-key}, each 0600. it accepts only an https url, so the pull token never travels in clear. hives get a read grant on the agent-closures bucket, and swarm-queue-client a reader for it. part 2 of 3: nothing uses the files or the reader until the probe lands.
100 lines
3.5 KiB
Rust
100 lines
3.5 KiB
Rust
//! The connection side of the per-agent closure KV bucket. The bucket's names
|
|
//! and stored type are `swarm_wire::agent_closures`.
|
|
|
|
/// Open the closures bucket for writing, creating it if nothing has yet.
|
|
///
|
|
/// **Controller-side only.** `history: 1`: a reader wants the latest build.
|
|
#[cfg(feature = "kv")]
|
|
pub async fn open_or_create(
|
|
client: &async_nats::Client,
|
|
) -> Result<async_nats::jetstream::kv::Store, crate::Error> {
|
|
let js = async_nats::jetstream::new(client.clone());
|
|
crate::kv::open_or_create(
|
|
&js,
|
|
swarm_wire::agent_closures::BUCKET,
|
|
"agent-closures",
|
|
"The latest system closure the swarm builder built for each agent".to_owned(),
|
|
)
|
|
.await
|
|
}
|
|
|
|
/// What reading one agent's key found.
|
|
#[cfg(feature = "agent-closures")]
|
|
#[derive(Debug)]
|
|
pub enum Read {
|
|
/// The bucket answered: the agent's latest build, or `None` for an absent
|
|
/// or deleted key.
|
|
Current(Option<swarm_wire::agent_closures::AgentClosure>),
|
|
/// Nothing answers for the bucket's stream: no build has been recorded
|
|
/// yet, or `JetStream` is not serving it right now.
|
|
NoBucket,
|
|
}
|
|
|
|
/// Read `agent`'s latest closure record with one direct get.
|
|
///
|
|
/// # Errors
|
|
///
|
|
/// A failed request or an undecodable value. A request the queue refused for
|
|
/// permissions reaches the client as a timeout, not as a refusal.
|
|
#[cfg(feature = "agent-closures")]
|
|
pub async fn read(
|
|
client: &async_nats::Client,
|
|
agent: &str,
|
|
) -> Result<Read, Box<dyn std::error::Error + Send + Sync>> {
|
|
use async_nats::jetstream::stream::DirectGetErrorKind;
|
|
use swarm_wire::agent_closures::{STREAM, key_subject};
|
|
|
|
let js = async_nats::jetstream::new(client.clone());
|
|
let stream = js.get_stream_no_info(STREAM).await?;
|
|
let message = match stream.direct_get_last_for_subject(key_subject(agent)).await {
|
|
Ok(message) => message,
|
|
Err(e) if matches!(e.kind(), DirectGetErrorKind::NotFound) => {
|
|
return Ok(Read::Current(None));
|
|
}
|
|
Err(e)
|
|
if matches!(
|
|
e.kind(),
|
|
DirectGetErrorKind::ErrorResponse(status, _)
|
|
if status == async_nats::StatusCode::NO_RESPONDERS
|
|
) =>
|
|
{
|
|
return Ok(Read::NoBucket);
|
|
}
|
|
Err(e) => return Err(e.into()),
|
|
};
|
|
Ok(Read::Current(decode(
|
|
message
|
|
.headers
|
|
.get("KV-Operation")
|
|
.map(async_nats::HeaderValue::as_str),
|
|
&message.payload,
|
|
)?))
|
|
}
|
|
|
|
/// One stored message as a record: a KV delete or purge is stored as an empty
|
|
/// marker carrying the `KV-Operation` header, and reads as no record.
|
|
#[cfg(feature = "agent-closures")]
|
|
fn decode(
|
|
operation: Option<&str>,
|
|
payload: &[u8],
|
|
) -> serde_json::Result<Option<swarm_wire::agent_closures::AgentClosure>> {
|
|
if matches!(operation, Some("DEL" | "PURGE")) {
|
|
return Ok(None);
|
|
}
|
|
serde_json::from_slice(payload).map(Some)
|
|
}
|
|
|
|
#[cfg(all(test, feature = "agent-closures"))]
|
|
mod tests {
|
|
use super::decode;
|
|
|
|
#[test]
|
|
fn a_stored_record_reads_and_a_delete_is_no_record() {
|
|
let payload = br#"{"meta_rev":"abc","store_path":"/nix/store/x","system":"x86_64-linux","built_unix":7,"cached":true}"#;
|
|
let record = decode(None, payload).expect("decodes").expect("a record");
|
|
assert_eq!(record.meta_rev, "abc");
|
|
assert!(record.cached);
|
|
assert!(decode(Some("DEL"), b"").expect("decodes").is_none());
|
|
assert!(decode(Some("PURGE"), b"").expect("decodes").is_none());
|
|
}
|
|
}
|