crates/dwarves-embed/src/mimir_release.rs
//! **The UI's leg of the 🚀 Mímir release plan** — ONE short `dwarf.v1.MimirRelease.Plan`
//! call to the forge. The forge answers at once: the plan it keeps (`done`), or the jera
//! job computing it (`queued` / `running`). [`ask`] is one call; the pane asks again
//! while the answer is on the queue ([`ask_until_planned`]). Nothing here reads a
//! checkout or dials nornir: the forge reads Mímir.
use std::time::Duration;
use anyhow::{Context, Result};
use crate::pb;
/// One answer of the forge, as the pane reads it.
#[derive(Clone, Debug, Default, PartialEq)]
pub struct PlanAnswer {
/// `done` · `queued` · `running` · `failed` · `refused` — or this client's
/// `unreachable` when the question never reached the forge.
pub state: String,
pub job_id: String,
pub note: String,
/// The plan (JSON), when the forge has one — also beside a queued refresh.
pub plan_json: String,
pub waited_ms: u64,
pub age_ms: u64,
/// Where it was asked (`MimirRelease.Plan @ <forge> / <workspace>`).
pub via: String,
}
impl PlanAnswer {
/// Is the answer on the forge's jera queue (ask again)?
pub fn is_waiting(&self) -> bool {
matches!(self.state.as_str(), "queued" | "running")
}
}
/// How long [`ask_until_planned`] waits between two asks. Every ask is one short call.
pub const REASK: Duration = Duration::from_millis(1500);
/// How long [`ask_until_planned`] keeps asking before it hands the waiting answer back.
pub const PATIENCE: Duration = Duration::from_secs(30 * 60);
fn via(endpoint: &str, workspace: &str) -> String {
format!(
"MimirRelease.Plan @ {} / {}",
crate::norm_endpoint(endpoint),
if workspace.is_empty() {
"(no workspace)"
} else {
workspace
}
)
}
fn plan_rpc(
endpoint: &str,
token: &str,
workspace: &str,
fresh: bool,
) -> Result<pb::MimirReleaseAnswer> {
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.context("build tokio runtime for the forge client")?;
let endpoint = crate::norm_endpoint(endpoint);
let (token, workspace) = (token.to_string(), workspace.to_string());
rt.block_on(async move {
let bearer = nornir_service::client::bearer(&token).context("parse bearer token")?;
let channel = tonic::transport::Channel::from_shared(endpoint.clone())
.with_context(|| format!("invalid server url `{endpoint}`"))?
.connect_timeout(Duration::from_secs(3))
.timeout(crate::READ_TIMEOUT)
.connect()
.await
.with_context(|| format!("connect to dwarves-server at {endpoint}"))?;
let mut client = pb::mimir_release_client::MimirReleaseClient::with_interceptor(
channel,
move |mut req: tonic::Request<()>| {
if let Some(b) = &bearer {
req.metadata_mut().insert("authorization", b.clone());
}
Ok(req)
},
)
.max_decoding_message_size(64 << 20);
anyhow::Ok(
client
.plan(pb::MimirReleaseAsk { workspace, fresh })
.await
.context("MimirRelease.Plan RPC")?
.into_inner(),
)
})
}
/// **Ask the forge ONCE.** Total: a question that never reached the forge is an
/// answer with the state `unreachable` and the reason.
pub fn ask(endpoint: &str, token: &str, workspace: &str, fresh: bool) -> PlanAnswer {
let via = via(endpoint, workspace);
match plan_rpc(endpoint, token, workspace, fresh) {
Ok(a) => PlanAnswer {
state: a.state,
job_id: a.job_id,
note: a.note,
plan_json: a.plan_json,
waited_ms: a.waited_ms,
age_ms: a.age_ms,
via,
},
Err(e) => PlanAnswer {
state: "unreachable".into(),
note: format!("{e:#}"),
via,
..Default::default()
},
}
}
/// **Ask until the plan is there.** The first ask carries `fresh`; while the answer is
/// on the forge's jera queue it is asked for again every [`REASK`] (each a short call,
/// `fresh` off — the job is already queued), handing every waiting answer to
/// `on_wait` (the pane shows the job), until it is not waiting or [`PATIENCE`] passed.
/// Call it off the UI thread.
pub fn ask_until_planned(
endpoint: &str,
token: &str,
workspace: &str,
fresh: bool,
on_wait: &dyn Fn(&PlanAnswer),
) -> PlanAnswer {
ask_until(
|f| ask(endpoint, token, workspace, f),
fresh,
REASK,
PATIENCE,
on_wait,
)
}
/// [`ask_until_planned`] over any asker — the loop, testable without a socket.
pub fn ask_until(
asker: impl Fn(bool) -> PlanAnswer,
fresh: bool,
reask: Duration,
patience: Duration,
on_wait: &dyn Fn(&PlanAnswer),
) -> PlanAnswer {
let t0 = std::time::Instant::now();
let mut a = asker(fresh);
while a.is_waiting() && t0.elapsed() < patience {
on_wait(&a);
std::thread::sleep(reask);
a = asker(false);
}
a
}
#[cfg(test)]
mod tests {
use super::*;
use std::sync::Mutex;
/// The forge's file and this client's are ONE contract: byte for byte.
#[test]
fn the_proto_is_the_forges_file_byte_for_byte() {
let here = include_str!("../proto/mimir_release.proto");
let forge = include_str!("../../dwarves/proto/mimir_release.proto");
assert_eq!(here, forge, "crates/dwarves-embed/proto/mimir_release.proto must equal crates/dwarves/proto/mimir_release.proto");
}
/// While the plan is on the queue the client asks again — `fresh` only on the
/// first ask, or a refresh would queue itself for ever — and hands every waiting
/// answer to the pane; it stops at the first answer that is not waiting.
#[test]
fn it_asks_again_while_the_plan_is_on_jera_and_fresh_only_once() {
let asked: Mutex<Vec<bool>> = Mutex::new(Vec::new());
let waits: Mutex<Vec<String>> = Mutex::new(Vec::new());
let script = ["queued", "running", "done"];
let a = ask_until(
|fresh| {
let mut v = asked.lock().unwrap();
v.push(fresh);
PlanAnswer {
state: script[v.len() - 1].into(),
job_id: "j-1".into(),
..Default::default()
}
},
true,
Duration::from_millis(1),
Duration::from_secs(5),
&|w| waits.lock().unwrap().push(w.state.clone()),
);
assert_eq!(a.state, "done");
assert_eq!(*asked.lock().unwrap(), [true, false, false]);
assert_eq!(*waits.lock().unwrap(), ["queued", "running"]);
}
/// A forge that is not there is an answer, not a panic and not an empty plan.
#[test]
fn an_unreachable_forge_is_a_state() {
let a = ask("http://127.0.0.1:1", "", "nordisk", false);
assert_eq!(a.state, "unreachable", "{a:?}");
assert!(a.plan_json.is_empty() && !a.note.is_empty());
assert!(a.via.contains("MimirRelease.Plan") && a.via.contains("nordisk"));
}
}