use std::path::{Path, PathBuf};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use crate::errors::RuntimeError;
use crate::http_client::http_json;
use crate::origin::{stamp_for_completion, TrustClass, PEER_ORIGIN_HEADER, PEER_TRUST_HEADER};
pub const SPAWNER_ENV: &str = "MURMUR_SPAWNER";
pub const DELEGATION_ID_HEADER: &str = "x-murmur-delegation-id";
pub const COMPLETION_SESSION_HEADER: &str = "x-murmur-completion-session";
pub const DELEGATION_ID_PREFIX: &str = "dlg_";
pub const COMPLETION_FILE: &str = "completion.json";
pub const MAX_DETAIL_BYTES: usize = 2_000;
pub fn new_delegation_id() -> String {
format!("{DELEGATION_ID_PREFIX}{}", uuid::Uuid::now_v7().simple())
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CompletionAddress {
pub url: String,
pub trust: TrustClass,
}
#[derive(Debug, Clone)]
pub struct Spawner {
pub session_id: String,
pub context_id: String,
pub report_to: Option<CompletionAddress>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SpawnerHandle {
pub session_id: String,
pub context_id: String,
pub delegation_id: String,
pub report_to: Option<CompletionAddress>,
}
impl SpawnerHandle {
pub fn for_delegation(spawner: &Spawner, delegation_id: String) -> Self {
Self {
session_id: spawner.session_id.clone(),
context_id: spawner.context_id.clone(),
delegation_id,
report_to: spawner.report_to.clone(),
}
}
pub fn to_env_value(&self) -> String {
let mut value = serde_json::json!({
"session_id": self.session_id,
"context_id": self.context_id,
"delegation_id": self.delegation_id,
});
if let Some(address) = &self.report_to {
value["url"] = Value::String(address.url.clone());
value["trust"] = Value::String(address.trust.as_str().to_string());
}
value.to_string()
}
pub fn from_env() -> Result<Option<Self>, RuntimeError> {
let Some(raw) = std::env::var_os(SPAWNER_ENV) else {
return Ok(None);
};
let raw = raw.to_string_lossy().to_string();
if raw.trim().is_empty() {
return Ok(None);
}
Self::parse(raw.trim()).map(Some)
}
pub fn parse(value: &str) -> Result<Self, RuntimeError> {
let unreadable = |reason: String| RuntimeError::SpawnerHandleUnreadable { reason };
let parsed: Value = serde_json::from_str(value)
.map_err(|error| unreadable(format!("it is not JSON: {error}")))?;
let field = |name: &str| -> Result<String, RuntimeError> {
parsed
.get(name)
.and_then(Value::as_str)
.filter(|value| !value.is_empty())
.map(str::to_string)
.ok_or_else(|| unreadable(format!("it carries no '{name}'")))
};
let report_to =
match (parsed.get("url").is_some(), parsed.get("trust").is_some()) {
(false, false) => None,
(true, true) => {
let trust_value = field("trust")?;
let trust = TrustClass::parse(&trust_value).ok_or_else(|| {
unreadable(format!("'{trust_value}' is not a trust class"))
})?;
Some(CompletionAddress {
url: field("url")?,
trust,
})
}
(true, false) => return Err(unreadable(
"it carries a 'url' with no 'trust'; a completion address is both or neither"
.to_string(),
)),
(false, true) => return Err(unreadable(
"it carries a 'trust' with no 'url'; a completion address is both or neither"
.to_string(),
)),
};
Ok(Self {
session_id: field("session_id")?,
context_id: field("context_id")?,
delegation_id: field("delegation_id")?,
report_to,
})
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum DelegationStatus {
Ok,
Error,
Crashed,
Terminated,
}
impl DelegationStatus {
pub fn as_str(&self) -> &'static str {
match self {
Self::Ok => "ok",
Self::Error => "error",
Self::Crashed => "crashed",
Self::Terminated => "terminated",
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum Reporter {
Child,
Launcher,
}
impl Reporter {
pub fn as_str(&self) -> &'static str {
match self {
Self::Child => "child",
Self::Launcher => "launcher",
}
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DelegationOutcome {
pub delegation_id: String,
pub capsule_name: String,
pub capsule_version: String,
pub session_id: String,
pub status: DelegationStatus,
#[serde(skip_serializing_if = "Option::is_none")]
pub result_path: Option<String>,
pub workdir: String,
pub duration_ms: u64,
#[serde(skip_serializing_if = "Option::is_none")]
pub detail: Option<String>,
pub reported_by: Reporter,
pub delivered: bool,
#[serde(skip_serializing_if = "Option::is_none")]
pub delivery_error: Option<String>,
}
impl DelegationOutcome {
pub fn message_text(&self) -> String {
let mut text = format!(
"Delegated capsule finished.\n\
delegation_id: {}\n\
capsule: {}@{}\n\
session_id: {}\n\
status: {}\n\
duration_ms: {}\n\
workdir: {}\n",
self.delegation_id,
self.capsule_name,
self.capsule_version,
self.session_id,
self.status.as_str(),
self.duration_ms,
self.workdir,
);
match &self.result_path {
Some(path) => text.push_str(&format!("result: {path} (in that workdir)")),
None => text.push_str("result: none (the child wrote no result file)"),
}
if let Some(detail) = &self.detail {
text.push_str(&format!("\ndetail: {detail}"));
}
text.push_str("\n\nThe child's own output is in that file and is not reproduced here.");
text
}
fn with_bounded_detail(mut self) -> Self {
if let Some(detail) = self.detail.take() {
self.detail = Some(bound_detail(detail));
}
self
}
}
pub(crate) fn bound_detail(detail: String) -> String {
if detail.len() <= MAX_DETAIL_BYTES {
return detail;
}
let mut end = MAX_DETAIL_BYTES;
while end > 0 && !detail.is_char_boundary(end) {
end -= 1;
}
format!("{} […]", &detail[..end])
}
pub fn child_result_candidates(
child_workdir: &Path,
session_id: &str,
task_id: &str,
) -> [PathBuf; 4] {
let session_out = child_workdir.join(".murmur").join(session_id).join("out");
let workdir_out = child_workdir.join("out");
let per_task = format!("result_{task_id}.txt");
[
session_out.join(&per_task),
session_out.join("result.txt"),
workdir_out.join(&per_task),
workdir_out.join("result.txt"),
]
}
pub fn child_result_path(child_workdir: &Path, session_id: &str, task_id: &str) -> Option<PathBuf> {
child_result_candidates(child_workdir, session_id, task_id)
.into_iter()
.find(|path| path.is_file())
}
pub fn completion_path(workdir: &Path) -> PathBuf {
workdir.join(COMPLETION_FILE)
}
pub fn read_completion(workdir: &Path) -> Option<DelegationOutcome> {
let raw = std::fs::read_to_string(completion_path(workdir)).ok()?;
serde_json::from_str(&raw).ok()
}
pub fn write_completion(workdir: &Path, outcome: &DelegationOutcome) -> Result<(), String> {
let body = serde_json::to_string_pretty(outcome)
.map_err(|error| format!("failed to serialize the completion: {error}"))?;
let target = completion_path(workdir);
let temp = workdir.join(format!(
".{COMPLETION_FILE}.{}.tmp",
uuid::Uuid::now_v7().simple()
));
std::fs::write(&temp, format!("{body}\n"))
.map_err(|error| format!("failed to write {COMPLETION_FILE}: {error}"))?;
std::fs::rename(&temp, &target).map_err(|error| {
let _ = std::fs::remove_file(&temp);
format!("failed to write {COMPLETION_FILE}: {error}")
})
}
pub fn deliver_completion(
handle: &SpawnerHandle,
address: &CompletionAddress,
outcome: &DelegationOutcome,
) -> Result<(), String> {
let stamped = stamp_for_completion(Some(address.trust));
let body = serde_json::json!({
"jsonrpc": "2.0",
"id": format!("req_{}", uuid::Uuid::now_v7().simple()),
"method": "message/send",
"params": {
"message": {
"messageId": format!("msg_{}", uuid::Uuid::now_v7().simple()),
"contextId": handle.context_id,
"role": "user",
"parts": [{"text": outcome.message_text()}]
}
}
})
.to_string();
let response = http_json(
"POST",
&address.url,
Some(&body),
&[
(PEER_ORIGIN_HEADER, stamped.origin().as_str()),
(PEER_TRUST_HEADER, stamped.trust().as_str()),
(DELEGATION_ID_HEADER, handle.delegation_id.as_str()),
(COMPLETION_SESSION_HEADER, handle.session_id.as_str()),
],
)?;
if let Some(error) = response.get("error") {
let message = error
.get("message")
.and_then(Value::as_str)
.unwrap_or("the parent refused the completion");
return Err(message.to_string());
}
if response
.pointer("/result/status/state")
.and_then(Value::as_str)
== Some("rejected")
{
return Err("the parent rejected the completion: its queue is full".to_string());
}
Ok(())
}
pub fn report_completion(
handle: &SpawnerHandle,
address: &CompletionAddress,
outcome: DelegationOutcome,
workdir: &Path,
) -> DelegationOutcome {
let mut outcome = outcome.with_bounded_detail();
outcome.delivered = false;
outcome.delivery_error = None;
if let Err(reason) = write_completion(workdir, &outcome) {
crate::runtime_err!(
"[capsule-runtime] delegation {}: {reason}",
outcome.delegation_id
);
}
match deliver_completion(handle, address, &outcome) {
Ok(()) => outcome.delivered = true,
Err(reason) => {
crate::runtime_err!(
"[capsule-runtime] delegation {}: the completion could not be delivered to {}: {reason}; recorded in {}",
outcome.delegation_id,
address.url,
completion_path(workdir).display(),
);
outcome.delivery_error = Some(reason);
}
}
if let Err(reason) = write_completion(workdir, &outcome) {
crate::runtime_err!(
"[capsule-runtime] delegation {}: {reason}",
outcome.delegation_id
);
}
outcome
}
pub fn record_terminated(workdir: &Path, outcome: DelegationOutcome) -> DelegationOutcome {
let mut outcome = outcome.with_bounded_detail();
outcome.delivered = false;
outcome.delivery_error = None;
if let Err(reason) = write_completion(workdir, &outcome) {
crate::runtime_err!(
"[capsule-runtime] delegation {}: {reason}",
outcome.delegation_id
);
}
outcome
}
#[cfg(test)]
mod tests {
use super::*;
fn handle() -> SpawnerHandle {
SpawnerHandle {
session_id: "ses_parent".to_string(),
context_id: "ctx_parent".to_string(),
delegation_id: "dlg_0001".to_string(),
report_to: Some(address()),
}
}
fn address() -> CompletionAddress {
CompletionAddress {
url: "http://127.0.0.1:7777".to_string(),
trust: TrustClass::Untrusted,
}
}
fn outcome() -> DelegationOutcome {
DelegationOutcome {
delegation_id: "dlg_0001".to_string(),
capsule_name: "worker".to_string(),
capsule_version: "0.1.0".to_string(),
session_id: "ses_child".to_string(),
status: DelegationStatus::Ok,
result_path: Some("out/result.txt".to_string()),
workdir: "/tmp/parent/.murmur/children/worker-abc".to_string(),
duration_ms: 42,
detail: None,
reported_by: Reporter::Child,
delivered: false,
delivery_error: None,
}
}
#[test]
fn a_handle_round_trips_through_its_env_value() {
for trust in [TrustClass::Trusted, TrustClass::Untrusted] {
let mut original = handle();
original.report_to = Some(CompletionAddress {
url: address().url,
trust,
});
let parsed = SpawnerHandle::parse(&original.to_env_value()).expect("a written handle");
assert_eq!(parsed, original);
}
}
#[test]
fn a_lineage_only_handle_carries_no_address() {
let mut original = handle();
original.report_to = None;
let written = original.to_env_value();
assert!(!written.contains("url"), "{written}");
assert!(!written.contains("trust"), "{written}");
let parsed = SpawnerHandle::parse(&written).expect("a written handle");
assert_eq!(parsed, original);
assert_eq!(parsed.session_id, "ses_parent");
assert_eq!(parsed.delegation_id, "dlg_0001");
}
#[test]
fn an_unreadable_handle_is_an_error_and_not_an_absence() {
let refused = [
"".to_string(),
"not json".to_string(),
"{}".to_string(),
serde_json::json!({"url": "http://x", "session_id": "s", "context_id": "c",
"delegation_id": "dlg_1"})
.to_string(),
serde_json::json!({"url": "http://x", "session_id": "s", "context_id": "c",
"trust": "maybe", "delegation_id": "dlg_1"})
.to_string(),
serde_json::json!({"session_id": "s", "context_id": "c", "delegation_id": "dlg_1",
"trust": "trusted"})
.to_string(),
];
for value in refused {
let error = SpawnerHandle::parse(&value).unwrap_err();
assert!(
matches!(error, RuntimeError::SpawnerHandleUnreadable { .. }),
"{value:?} produced {error}"
);
assert!(
error
.to_string()
.contains("tell its spawner that it finished"),
"the refusal must say why it is fatal: {error}"
);
}
}
#[test]
fn delegation_ids_are_prefixed_and_unique() {
let first = new_delegation_id();
let second = new_delegation_id();
assert_ne!(first, second);
for id in [&first, &second] {
assert!(id.starts_with(DELEGATION_ID_PREFIX), "{id}");
assert!(
id[DELEGATION_ID_PREFIX.len()..]
.chars()
.all(|c| c.is_ascii_hexdigit() && !c.is_ascii_uppercase()),
"{id}"
);
}
}
#[test]
fn the_message_names_the_result_and_never_carries_it() {
let text = outcome().message_text();
assert!(text.contains("delegation_id: dlg_0001"), "{text}");
assert!(text.contains("capsule: worker@0.1.0"), "{text}");
assert!(text.contains("session_id: ses_child"), "{text}");
assert!(text.contains("status: ok"), "{text}");
assert!(text.contains("result: out/result.txt"), "{text}");
assert!(text.contains("not reproduced here"), "{text}");
}
#[test]
fn a_crash_detail_is_capped() {
let mut crashed = outcome();
crashed.status = DelegationStatus::Crashed;
crashed.detail = Some("x".repeat(MAX_DETAIL_BYTES * 4));
let bounded = crashed.with_bounded_detail();
let detail = bounded.detail.expect("the detail survives, bounded");
assert!(
detail.len() <= MAX_DETAIL_BYTES + " […]".len(),
"detail is {} bytes",
detail.len()
);
assert!(detail.ends_with("[…]"), "the cut is marked: {detail}");
}
#[test]
fn a_completion_round_trips_through_the_file() {
let dir = tempfile::tempdir().unwrap();
assert!(read_completion(dir.path()).is_none());
let mut written = outcome();
written.delivered = true;
write_completion(dir.path(), &written).unwrap();
let read = read_completion(dir.path()).expect("the written completion reads back");
assert_eq!(read.delegation_id, written.delegation_id);
assert_eq!(read.status, DelegationStatus::Ok);
assert!(read.delivered);
assert_eq!(read.result_path.as_deref(), Some("out/result.txt"));
let raw = std::fs::read_to_string(completion_path(dir.path())).unwrap();
assert!(!raw.contains("detail"), "{raw}");
assert!(!raw.contains("delivery_error"), "{raw}");
}
#[test]
fn a_terminated_delegation_is_recorded_and_not_posted() {
let dir = tempfile::tempdir().unwrap();
let mut ended = outcome();
ended.status = DelegationStatus::Terminated;
ended.reported_by = Reporter::Launcher;
ended.detail = Some("the parent ended this delegation".to_string());
let recorded = record_terminated(dir.path(), ended);
assert!(!recorded.delivered);
assert!(recorded.delivery_error.is_none());
let read = read_completion(dir.path()).expect("the record exists");
assert_eq!(read.status, DelegationStatus::Terminated);
assert_eq!(read.reported_by, Reporter::Launcher);
}
#[test]
fn an_undeliverable_completion_is_recorded_with_its_reason() {
let dir = tempfile::tempdir().unwrap();
let nowhere = CompletionAddress {
url: "http://127.0.0.1:1".to_string(),
trust: TrustClass::Untrusted,
};
let reported = report_completion(&handle(), &nowhere, outcome(), dir.path());
assert!(!reported.delivered);
let reason = reported.delivery_error.expect("the refusal is recorded");
assert!(reason.contains("failed to connect"), "{reason}");
let read = read_completion(dir.path()).expect("the record exists");
assert!(!read.delivered);
assert!(read.delivery_error.is_some());
}
#[test]
fn a_rewrite_is_never_visible_half_written() {
let dir = tempfile::tempdir().unwrap();
write_completion(dir.path(), &outcome()).unwrap();
let path = dir.path().to_path_buf();
let writer = std::thread::spawn(move || {
for turn in 0..400 {
let mut record = outcome();
record.delivered = turn % 2 == 0;
record.detail = Some("x".repeat(64 * 1024));
write_completion(&path, &record).unwrap();
}
});
let mut reads = 0;
while !writer.is_finished() {
assert!(
read_completion(dir.path()).is_some(),
"a concurrent reader saw no readable completion after {reads} good reads"
);
reads += 1;
}
writer.join().unwrap();
assert!(reads > 0, "the reader never got to run");
let strays: Vec<_> = std::fs::read_dir(dir.path())
.unwrap()
.filter_map(|entry| entry.ok().map(|entry| entry.file_name()))
.filter(|name| name != COMPLETION_FILE)
.collect();
assert!(strays.is_empty(), "temporary files left behind: {strays:?}");
}
}