#![cfg(all(feature = "embedded", feature = "ipc", unix))]
use std::io::{BufRead, BufReader};
use std::path::{Path, PathBuf};
use std::process::{Child, Command, Stdio};
use std::sync::{mpsc, Arc};
use std::time::Duration;
use ikigai_core::{
Capability, EndpointSpace, Exact, FnEndpoint, Iri, Kernel, ReprType, Representation, Request,
SystemClock, Verb,
};
use ikigai_resolve::{Resolver, RunContext};
const PATIENCE: Duration = Duration::from_secs(90);
struct Door(Child);
impl Drop for Door {
fn drop(&mut self) {
let _ = self.0.kill();
let _ = self.0.wait();
}
}
fn scratch(case: &str) -> PathBuf {
let dir = std::env::temp_dir().join(format!("iprc-{}-{case}", std::process::id()));
let _ = std::fs::remove_dir_all(&dir);
std::fs::create_dir_all(dir.join("config").join("ikigai")).expect("scratch config home");
dir
}
fn peer(socket: &Path) {
let probe = FnEndpoint::new("probe", |inv: &ikigai_core::Invocation<'_>| {
let retries = inv
.retry_budget()
.map(|pool| pool.remaining().to_string())
.unwrap_or_else(|| "-".to_string());
let left = inv
.time_left()
.map(|left| left.as_millis().to_string())
.unwrap_or_else(|| "-".to_string());
Ok(Representation::new(
ReprType::new("text/plain"),
format!("{retries},{left}").into_bytes(),
))
});
let kernel = Kernel::new(Arc::new(
EndpointSpace::new().bind(Exact::new("urn:rec:probe"), probe),
))
.with_retry_budget(0, 3)
.with_clock(Arc::new(SystemClock));
let listener = ikigai_ipc::bind(socket).expect("the peer binds");
std::thread::spawn(move || {
let _ = ikigai_ipc::serve_shared_on(Arc::new(kernel), listener);
});
}
fn door(home: &Path, door: &Path, peer: &Path) -> Door {
let mount = format!("urn:rec:={}", peer.display());
let mut child = Command::new(env!("CARGO_BIN_EXE_ikigai"))
.args(["serve", door.to_str().unwrap(), "--prefer", &mount])
.env("HOME", home)
.env("XDG_CONFIG_HOME", home.join("config"))
.env("IKIGAI_FILES", home.join("files"))
.env_remove("IKIGAI_CLIENTS")
.env_remove("IKIGAI_GRANTS")
.stdin(Stdio::null())
.stdout(Stdio::null())
.stderr(Stdio::piped())
.spawn()
.expect("the door starts");
let stderr = child.stderr.take().expect("piped stderr");
let (tx, rx) = mpsc::channel();
std::thread::spawn(move || {
let mut seen = String::new();
let mut lines = BufReader::new(stderr).lines().map_while(Result::ok);
for line in lines.by_ref() {
seen.push_str(&line);
seen.push('\n');
if line.contains("serving on") {
let _ = tx.send(Ok(seen.clone()));
lines.for_each(drop);
return;
}
}
let _ = tx.send(Err(seen));
});
let door = Door(child);
match rx.recv_timeout(PATIENCE) {
Ok(Ok(_)) => door,
Ok(Err(seen)) => panic!("the door exited before its banner:\n{seen}"),
Err(e) => panic!("the door never printed its banner: {e}"),
}
}
fn probe() -> Request {
Request::new(Verb::Source, Iri::parse("urn:rec:probe").unwrap())
}
fn seen(answer: &Representation) -> (String, Option<u64>) {
let text = String::from_utf8_lossy(&answer.bytes);
let (retries, left) = text.split_once(',').expect("retries,time_left");
(retries.to_string(), left.parse().ok())
}
#[test]
fn a_prefer_mount_carries_the_runs_deadline_to_its_peer() {
let home = scratch("run");
let peer_socket = home.join("p.sock");
let door_socket = home.join("d.sock");
peer(&peer_socket);
let _door = door(&home, &door_socket, &peer_socket);
let client = ikigai_ipc::connect(&door_socket).expect("the client reaches the door");
let run = RunContext {
retries: Some(2),
time_left: Some(Duration::from_secs(30)),
};
let (answer, _) = client
.issue_as_in_run(probe(), &Capability::root(), &run)
.expect("the peer answers through the door");
let (retries, left) = seen(&answer);
let left = left.unwrap_or_else(|| {
panic!("the deadline stopped at the door: the peer saw no time left ({retries},-)")
});
assert!(
(20_000..=30_000).contains(&left),
"about the carried 30 s, on the peer's clock: {left} ms"
);
assert_eq!(
retries, "3",
"the count stops at the door (serve has no retry budget); the peer mints its own"
);
let (answer, _) = client
.issue_as(probe(), &Capability::root())
.expect("a plain read");
assert_eq!(seen(&answer), ("3".to_string(), None));
let _ = std::fs::remove_dir_all(&home);
}
#[test]
fn a_spent_deadline_is_answered_and_reaches_the_peer_as_zero_left() {
let home = scratch("spent");
let peer_socket = home.join("p.sock");
let door_socket = home.join("d.sock");
peer(&peer_socket);
let _door = door(&home, &door_socket, &peer_socket);
let client = ikigai_ipc::connect(&door_socket).expect("the client reaches the door");
let spent = RunContext {
retries: Some(0),
time_left: Some(Duration::ZERO),
};
let (answer, _) = client
.issue_as_in_run(probe(), &Capability::root(), &spent)
.expect("a spent deadline is answered, not refused at the door");
assert_eq!(seen(&answer), ("3".to_string(), Some(0)));
let own = Request::new(Verb::Source, Iri::parse("urn:host:info").unwrap());
client
.issue_as_in_run(own, &Capability::root(), &spent)
.expect("the door's own resource answers a spent run");
let _ = std::fs::remove_dir_all(&home);
}