1use std::path::{Path, PathBuf};
33
34use serde::{Deserialize, Serialize};
35use serde_json::Value;
36
37use crate::errors::RuntimeError;
38use crate::http_client::http_json;
39use crate::origin::{stamp_for_completion, TrustClass, PEER_ORIGIN_HEADER, PEER_TRUST_HEADER};
40
41pub const SPAWNER_ENV: &str = "MURMUR_SPAWNER";
47
48pub const DELEGATION_ID_HEADER: &str = "x-murmur-delegation-id";
51
52pub const COMPLETION_SESSION_HEADER: &str = "x-murmur-completion-session";
58
59pub const DELEGATION_ID_PREFIX: &str = "dlg_";
62
63pub const COMPLETION_FILE: &str = "completion.json";
65
66pub const MAX_DETAIL_BYTES: usize = 2_000;
71
72pub fn new_delegation_id() -> String {
77 format!("{DELEGATION_ID_PREFIX}{}", uuid::Uuid::now_v7().simple())
78}
79
80#[derive(Debug, Clone, PartialEq, Eq)]
86pub struct CompletionAddress {
87 pub url: String,
89 pub trust: TrustClass,
92}
93
94#[derive(Debug, Clone)]
99pub struct Spawner {
100 pub session_id: String,
104 pub context_id: String,
107 pub report_to: Option<CompletionAddress>,
109}
110
111#[derive(Debug, Clone, PartialEq, Eq)]
113pub struct SpawnerHandle {
114 pub session_id: String,
115 pub context_id: String,
116 pub delegation_id: String,
117 pub report_to: Option<CompletionAddress>,
118}
119
120impl SpawnerHandle {
121 pub fn for_delegation(spawner: &Spawner, delegation_id: String) -> Self {
123 Self {
124 session_id: spawner.session_id.clone(),
125 context_id: spawner.context_id.clone(),
126 delegation_id,
127 report_to: spawner.report_to.clone(),
128 }
129 }
130
131 pub fn to_env_value(&self) -> String {
136 let mut value = serde_json::json!({
137 "session_id": self.session_id,
138 "context_id": self.context_id,
139 "delegation_id": self.delegation_id,
140 });
141 if let Some(address) = &self.report_to {
142 value["url"] = Value::String(address.url.clone());
143 value["trust"] = Value::String(address.trust.as_str().to_string());
144 }
145 value.to_string()
146 }
147
148 pub fn from_env() -> Result<Option<Self>, RuntimeError> {
155 let Some(raw) = std::env::var_os(SPAWNER_ENV) else {
156 return Ok(None);
157 };
158 let raw = raw.to_string_lossy().to_string();
159 if raw.trim().is_empty() {
160 return Ok(None);
161 }
162 Self::parse(raw.trim()).map(Some)
163 }
164
165 pub fn parse(value: &str) -> Result<Self, RuntimeError> {
167 let unreadable = |reason: String| RuntimeError::SpawnerHandleUnreadable { reason };
168 let parsed: Value = serde_json::from_str(value)
169 .map_err(|error| unreadable(format!("it is not JSON: {error}")))?;
170 let field = |name: &str| -> Result<String, RuntimeError> {
171 parsed
172 .get(name)
173 .and_then(Value::as_str)
174 .filter(|value| !value.is_empty())
175 .map(str::to_string)
176 .ok_or_else(|| unreadable(format!("it carries no '{name}'")))
177 };
178 let report_to =
182 match (parsed.get("url").is_some(), parsed.get("trust").is_some()) {
183 (false, false) => None,
184 (true, true) => {
185 let trust_value = field("trust")?;
186 let trust = TrustClass::parse(&trust_value).ok_or_else(|| {
190 unreadable(format!("'{trust_value}' is not a trust class"))
191 })?;
192 Some(CompletionAddress {
193 url: field("url")?,
194 trust,
195 })
196 }
197 (true, false) => return Err(unreadable(
198 "it carries a 'url' with no 'trust'; a completion address is both or neither"
199 .to_string(),
200 )),
201 (false, true) => return Err(unreadable(
202 "it carries a 'trust' with no 'url'; a completion address is both or neither"
203 .to_string(),
204 )),
205 };
206 Ok(Self {
207 session_id: field("session_id")?,
208 context_id: field("context_id")?,
209 delegation_id: field("delegation_id")?,
210 report_to,
211 })
212 }
213}
214
215#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
217#[serde(rename_all = "lowercase")]
218pub enum DelegationStatus {
219 Ok,
221 Error,
223 Crashed,
225 Terminated,
228}
229
230impl DelegationStatus {
231 pub fn as_str(&self) -> &'static str {
232 match self {
233 Self::Ok => "ok",
234 Self::Error => "error",
235 Self::Crashed => "crashed",
236 Self::Terminated => "terminated",
237 }
238 }
239}
240
241#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
243#[serde(rename_all = "lowercase")]
244pub enum Reporter {
245 Child,
247 Launcher,
249}
250
251impl Reporter {
252 pub fn as_str(&self) -> &'static str {
253 match self {
254 Self::Child => "child",
255 Self::Launcher => "launcher",
256 }
257 }
258}
259
260#[derive(Debug, Clone, Serialize, Deserialize)]
266pub struct DelegationOutcome {
267 pub delegation_id: String,
269 pub capsule_name: String,
270 pub capsule_version: String,
271 pub session_id: String,
273 pub status: DelegationStatus,
274 #[serde(skip_serializing_if = "Option::is_none")]
278 pub result_path: Option<String>,
279 pub workdir: String,
282 pub duration_ms: u64,
283 #[serde(skip_serializing_if = "Option::is_none")]
286 pub detail: Option<String>,
287 pub reported_by: Reporter,
288 pub delivered: bool,
291 #[serde(skip_serializing_if = "Option::is_none")]
293 pub delivery_error: Option<String>,
294}
295
296impl DelegationOutcome {
297 pub fn message_text(&self) -> String {
300 let mut text = format!(
301 "Delegated capsule finished.\n\
302 delegation_id: {}\n\
303 capsule: {}@{}\n\
304 session_id: {}\n\
305 status: {}\n\
306 duration_ms: {}\n\
307 workdir: {}\n",
308 self.delegation_id,
309 self.capsule_name,
310 self.capsule_version,
311 self.session_id,
312 self.status.as_str(),
313 self.duration_ms,
314 self.workdir,
315 );
316 match &self.result_path {
317 Some(path) => text.push_str(&format!("result: {path} (in that workdir)")),
318 None => text.push_str("result: none (the child wrote no result file)"),
319 }
320 if let Some(detail) = &self.detail {
321 text.push_str(&format!("\ndetail: {detail}"));
322 }
323 text.push_str("\n\nThe child's own output is in that file and is not reproduced here.");
324 text
325 }
326
327 fn with_bounded_detail(mut self) -> Self {
329 if let Some(detail) = self.detail.take() {
330 self.detail = Some(bound_detail(detail));
331 }
332 self
333 }
334}
335
336pub(crate) fn bound_detail(detail: String) -> String {
338 if detail.len() <= MAX_DETAIL_BYTES {
339 return detail;
340 }
341 let mut end = MAX_DETAIL_BYTES;
342 while end > 0 && !detail.is_char_boundary(end) {
343 end -= 1;
344 }
345 format!("{} […]", &detail[..end])
346}
347
348pub fn child_result_candidates(
356 child_workdir: &Path,
357 session_id: &str,
358 task_id: &str,
359) -> [PathBuf; 4] {
360 let session_out = child_workdir.join(".murmur").join(session_id).join("out");
361 let workdir_out = child_workdir.join("out");
362 let per_task = format!("result_{task_id}.txt");
363 [
364 session_out.join(&per_task),
365 session_out.join("result.txt"),
366 workdir_out.join(&per_task),
367 workdir_out.join("result.txt"),
368 ]
369}
370
371pub fn child_result_path(child_workdir: &Path, session_id: &str, task_id: &str) -> Option<PathBuf> {
373 child_result_candidates(child_workdir, session_id, task_id)
374 .into_iter()
375 .find(|path| path.is_file())
376}
377
378pub fn completion_path(workdir: &Path) -> PathBuf {
382 workdir.join(COMPLETION_FILE)
383}
384
385pub fn read_completion(workdir: &Path) -> Option<DelegationOutcome> {
390 let raw = std::fs::read_to_string(completion_path(workdir)).ok()?;
391 serde_json::from_str(&raw).ok()
392}
393
394pub fn write_completion(workdir: &Path, outcome: &DelegationOutcome) -> Result<(), String> {
404 let body = serde_json::to_string_pretty(outcome)
405 .map_err(|error| format!("failed to serialize the completion: {error}"))?;
406 let target = completion_path(workdir);
407 let temp = workdir.join(format!(
408 ".{COMPLETION_FILE}.{}.tmp",
409 uuid::Uuid::now_v7().simple()
410 ));
411 std::fs::write(&temp, format!("{body}\n"))
412 .map_err(|error| format!("failed to write {COMPLETION_FILE}: {error}"))?;
413 std::fs::rename(&temp, &target).map_err(|error| {
414 let _ = std::fs::remove_file(&temp);
415 format!("failed to write {COMPLETION_FILE}: {error}")
416 })
417}
418
419pub fn deliver_completion(
427 handle: &SpawnerHandle,
428 address: &CompletionAddress,
429 outcome: &DelegationOutcome,
430) -> Result<(), String> {
431 let stamped = stamp_for_completion(Some(address.trust));
432 let body = serde_json::json!({
433 "jsonrpc": "2.0",
434 "id": format!("req_{}", uuid::Uuid::now_v7().simple()),
435 "method": "message/send",
436 "params": {
437 "message": {
438 "messageId": format!("msg_{}", uuid::Uuid::now_v7().simple()),
439 "contextId": handle.context_id,
440 "role": "user",
441 "parts": [{"text": outcome.message_text()}]
442 }
443 }
444 })
445 .to_string();
446
447 let response = http_json(
448 "POST",
449 &address.url,
450 Some(&body),
451 &[
452 (PEER_ORIGIN_HEADER, stamped.origin().as_str()),
453 (PEER_TRUST_HEADER, stamped.trust().as_str()),
454 (DELEGATION_ID_HEADER, handle.delegation_id.as_str()),
455 (COMPLETION_SESSION_HEADER, handle.session_id.as_str()),
456 ],
457 )?;
458
459 if let Some(error) = response.get("error") {
463 let message = error
464 .get("message")
465 .and_then(Value::as_str)
466 .unwrap_or("the parent refused the completion");
467 return Err(message.to_string());
468 }
469 if response
470 .pointer("/result/status/state")
471 .and_then(Value::as_str)
472 == Some("rejected")
473 {
474 return Err(
475 "the parent rejected the completion: its queue is full or its session has stopped taking work"
476 .to_string(),
477 );
478 }
479 Ok(())
480}
481
482pub fn report_completion(
491 handle: &SpawnerHandle,
492 address: &CompletionAddress,
493 outcome: DelegationOutcome,
494 workdir: &Path,
495) -> DelegationOutcome {
496 let mut outcome = outcome.with_bounded_detail();
497 outcome.delivered = false;
498 outcome.delivery_error = None;
499 if let Err(reason) = write_completion(workdir, &outcome) {
500 crate::runtime_err!(
501 "[capsule-runtime] delegation {}: {reason}",
502 outcome.delegation_id
503 );
504 }
505
506 match deliver_completion(handle, address, &outcome) {
507 Ok(()) => outcome.delivered = true,
508 Err(reason) => {
509 crate::runtime_err!(
513 "[capsule-runtime] delegation {}: the completion could not be delivered to {}: {reason}; recorded in {}",
514 outcome.delegation_id,
515 address.url,
516 completion_path(workdir).display(),
517 );
518 outcome.delivery_error = Some(reason);
519 }
520 }
521 if let Err(reason) = write_completion(workdir, &outcome) {
522 crate::runtime_err!(
523 "[capsule-runtime] delegation {}: {reason}",
524 outcome.delegation_id
525 );
526 }
527 outcome
528}
529
530pub fn record_terminated(workdir: &Path, outcome: DelegationOutcome) -> DelegationOutcome {
535 let mut outcome = outcome.with_bounded_detail();
536 outcome.delivered = false;
537 outcome.delivery_error = None;
538 if let Err(reason) = write_completion(workdir, &outcome) {
539 crate::runtime_err!(
540 "[capsule-runtime] delegation {}: {reason}",
541 outcome.delegation_id
542 );
543 }
544 outcome
545}
546
547#[cfg(test)]
548mod tests {
549 use super::*;
550
551 fn handle() -> SpawnerHandle {
552 SpawnerHandle {
553 session_id: "ses_parent".to_string(),
554 context_id: "ctx_parent".to_string(),
555 delegation_id: "dlg_0001".to_string(),
556 report_to: Some(address()),
557 }
558 }
559
560 fn address() -> CompletionAddress {
561 CompletionAddress {
562 url: "http://127.0.0.1:7777".to_string(),
563 trust: TrustClass::Untrusted,
564 }
565 }
566
567 fn outcome() -> DelegationOutcome {
568 DelegationOutcome {
569 delegation_id: "dlg_0001".to_string(),
570 capsule_name: "worker".to_string(),
571 capsule_version: "0.1.0".to_string(),
572 session_id: "ses_child".to_string(),
573 status: DelegationStatus::Ok,
574 result_path: Some("out/result.txt".to_string()),
575 workdir: "/tmp/parent/.murmur/children/worker-abc".to_string(),
576 duration_ms: 42,
577 detail: None,
578 reported_by: Reporter::Child,
579 delivered: false,
580 delivery_error: None,
581 }
582 }
583
584 #[test]
585 fn a_handle_round_trips_through_its_env_value() {
586 for trust in [TrustClass::Trusted, TrustClass::Untrusted] {
587 let mut original = handle();
588 original.report_to = Some(CompletionAddress {
589 url: address().url,
590 trust,
591 });
592 let parsed = SpawnerHandle::parse(&original.to_env_value()).expect("a written handle");
593 assert_eq!(parsed, original);
594 }
595 }
596
597 #[test]
600 fn a_lineage_only_handle_carries_no_address() {
601 let mut original = handle();
602 original.report_to = None;
603 let written = original.to_env_value();
604 assert!(!written.contains("url"), "{written}");
605 assert!(!written.contains("trust"), "{written}");
606
607 let parsed = SpawnerHandle::parse(&written).expect("a written handle");
608 assert_eq!(parsed, original);
609 assert_eq!(parsed.session_id, "ses_parent");
610 assert_eq!(parsed.delegation_id, "dlg_0001");
611 }
612
613 #[test]
614 fn an_unreadable_handle_is_an_error_and_not_an_absence() {
615 let refused = [
616 "".to_string(),
617 "not json".to_string(),
618 "{}".to_string(),
619 serde_json::json!({"url": "http://x", "session_id": "s", "context_id": "c",
620 "delegation_id": "dlg_1"})
621 .to_string(),
622 serde_json::json!({"url": "http://x", "session_id": "s", "context_id": "c",
623 "trust": "maybe", "delegation_id": "dlg_1"})
624 .to_string(),
625 serde_json::json!({"session_id": "s", "context_id": "c", "delegation_id": "dlg_1",
628 "trust": "trusted"})
629 .to_string(),
630 ];
631 for value in refused {
632 let error = SpawnerHandle::parse(&value).unwrap_err();
633 assert!(
634 matches!(error, RuntimeError::SpawnerHandleUnreadable { .. }),
635 "{value:?} produced {error}"
636 );
637 assert!(
638 error
639 .to_string()
640 .contains("tell its spawner that it finished"),
641 "the refusal must say why it is fatal: {error}"
642 );
643 }
644 }
645
646 #[test]
648 fn delegation_ids_are_prefixed_and_unique() {
649 let first = new_delegation_id();
650 let second = new_delegation_id();
651 assert_ne!(first, second);
652 for id in [&first, &second] {
653 assert!(id.starts_with(DELEGATION_ID_PREFIX), "{id}");
654 assert!(
655 id[DELEGATION_ID_PREFIX.len()..]
656 .chars()
657 .all(|c| c.is_ascii_hexdigit() && !c.is_ascii_uppercase()),
658 "{id}"
659 );
660 }
661 }
662
663 #[test]
665 fn the_message_names_the_result_and_never_carries_it() {
666 let text = outcome().message_text();
667 assert!(text.contains("delegation_id: dlg_0001"), "{text}");
668 assert!(text.contains("capsule: worker@0.1.0"), "{text}");
669 assert!(text.contains("session_id: ses_child"), "{text}");
670 assert!(text.contains("status: ok"), "{text}");
671 assert!(text.contains("result: out/result.txt"), "{text}");
672 assert!(text.contains("not reproduced here"), "{text}");
673 }
674
675 #[test]
677 fn a_crash_detail_is_capped() {
678 let mut crashed = outcome();
679 crashed.status = DelegationStatus::Crashed;
680 crashed.detail = Some("x".repeat(MAX_DETAIL_BYTES * 4));
681 let bounded = crashed.with_bounded_detail();
682 let detail = bounded.detail.expect("the detail survives, bounded");
683 assert!(
684 detail.len() <= MAX_DETAIL_BYTES + " […]".len(),
685 "detail is {} bytes",
686 detail.len()
687 );
688 assert!(detail.ends_with("[…]"), "the cut is marked: {detail}");
689 }
690
691 #[test]
692 fn a_completion_round_trips_through_the_file() {
693 let dir = tempfile::tempdir().unwrap();
694 assert!(read_completion(dir.path()).is_none());
695
696 let mut written = outcome();
697 written.delivered = true;
698 write_completion(dir.path(), &written).unwrap();
699
700 let read = read_completion(dir.path()).expect("the written completion reads back");
701 assert_eq!(read.delegation_id, written.delegation_id);
702 assert_eq!(read.status, DelegationStatus::Ok);
703 assert!(read.delivered);
704 assert_eq!(read.result_path.as_deref(), Some("out/result.txt"));
705
706 let raw = std::fs::read_to_string(completion_path(dir.path())).unwrap();
708 assert!(!raw.contains("detail"), "{raw}");
709 assert!(!raw.contains("delivery_error"), "{raw}");
710 }
711
712 #[test]
715 fn a_terminated_delegation_is_recorded_and_not_posted() {
716 let dir = tempfile::tempdir().unwrap();
717 let mut ended = outcome();
718 ended.status = DelegationStatus::Terminated;
719 ended.reported_by = Reporter::Launcher;
720 ended.detail = Some("the parent ended this delegation".to_string());
721
722 let recorded = record_terminated(dir.path(), ended);
723 assert!(!recorded.delivered);
724 assert!(recorded.delivery_error.is_none());
725 let read = read_completion(dir.path()).expect("the record exists");
726 assert_eq!(read.status, DelegationStatus::Terminated);
727 assert_eq!(read.reported_by, Reporter::Launcher);
728 }
729
730 #[test]
732 fn an_undeliverable_completion_is_recorded_with_its_reason() {
733 let dir = tempfile::tempdir().unwrap();
734 let nowhere = CompletionAddress {
736 url: "http://127.0.0.1:1".to_string(),
737 trust: TrustClass::Untrusted,
738 };
739
740 let reported = report_completion(&handle(), &nowhere, outcome(), dir.path());
741 assert!(!reported.delivered);
742 let reason = reported.delivery_error.expect("the refusal is recorded");
743 assert!(reason.contains("failed to connect"), "{reason}");
744
745 let read = read_completion(dir.path()).expect("the record exists");
746 assert!(!read.delivered);
747 assert!(read.delivery_error.is_some());
748 }
749
750 #[test]
756 fn a_rewrite_is_never_visible_half_written() {
757 let dir = tempfile::tempdir().unwrap();
758 write_completion(dir.path(), &outcome()).unwrap();
759
760 let path = dir.path().to_path_buf();
761 let writer = std::thread::spawn(move || {
762 for turn in 0..400 {
763 let mut record = outcome();
764 record.delivered = turn % 2 == 0;
765 record.detail = Some("x".repeat(64 * 1024));
768 write_completion(&path, &record).unwrap();
769 }
770 });
771
772 let mut reads = 0;
773 while !writer.is_finished() {
774 assert!(
775 read_completion(dir.path()).is_some(),
776 "a concurrent reader saw no readable completion after {reads} good reads"
777 );
778 reads += 1;
779 }
780 writer.join().unwrap();
781 assert!(reads > 0, "the reader never got to run");
782
783 let strays: Vec<_> = std::fs::read_dir(dir.path())
785 .unwrap()
786 .filter_map(|entry| entry.ok().map(|entry| entry.file_name()))
787 .filter(|name| name != COMPLETION_FILE)
788 .collect();
789 assert!(strays.is_empty(), "temporary files left behind: {strays:?}");
790 }
791}