1use std::collections::{BTreeSet, HashMap, HashSet};
27use std::path::{Path, PathBuf};
28
29use murmur_artifact::{
30 current_platform, load_roster, resolve_roster_path, roster_warning_link, sha256_hex,
31 LockedSha256, MurmurLock, Registry, RegistryError, Roster, RosterReachability, RuntimeManifest,
32 TaskAcceptance, ROSTER_FILENAME, W_ROS_001,
33};
34use thiserror::Error;
35
36use crate::artifact::extract_manifest_yaml;
37
38#[derive(Debug, Clone)]
40pub struct AdmittedRoster {
41 source: Option<PathBuf>,
42 members: Vec<AdmittedMember>,
43 entry: usize,
44 reachability: RosterReachability,
45 edges: Vec<RosterEdge>,
46}
47
48#[derive(Debug, Clone)]
50pub struct AdmittedMember {
51 pub name: String,
53 pub capsule: String,
55 pub version: String,
57 pub entry: bool,
59 pub sha256: String,
62 pub manifest: RuntimeManifest,
64}
65
66impl AdmittedMember {
67 #[must_use]
70 pub fn requires_authentication(&self) -> bool {
71 self.manifest
72 .network
73 .as_ref()
74 .is_some_and(|network| network.authentication.is_some())
75 }
76}
77
78#[derive(Debug, Clone, PartialEq, Eq)]
80pub struct RosterEdge {
81 pub from: String,
82 pub to: String,
83}
84
85#[derive(Debug, Clone, PartialEq, Eq)]
87pub struct CallerOverflow {
88 pub member: String,
89 pub capsule: String,
90 pub version: String,
91 pub callers: Vec<String>,
93 pub task_acceptance: TaskAcceptance,
94 pub queue_depth: usize,
95 pub holds: usize,
98 pub sufficient_queue_depth: usize,
101}
102
103#[derive(Debug, Clone, PartialEq, Eq, Error)]
106pub enum RosterRefusal {
107 #[error("{message}")]
110 Malformed { message: String },
111 #[error("{ROSTER_FILENAME}: member name '{member}' is declared more than once")]
112 DuplicateMember { member: String },
113 #[error(
114 "{ROSTER_FILENAME}: no member is marked `entry: true`; a roster designates exactly one \
115 entry member"
116 )]
117 EntryNotDesignated,
118 #[error(
119 "{ROSTER_FILENAME}: members {} are each marked `entry: true`; a roster designates exactly \
120 one entry member",
121 quoted_list(members)
122 )]
123 EntryDesignatedMoreThanOnce { members: Vec<String> },
124 #[error(
127 "{ROSTER_FILENAME}: the reachability rule from '{from}' names '{unknown}', which is not a \
128 member of this roster"
129 )]
130 UnknownMemberInRule { from: String, unknown: String },
131 #[error(
133 "{ROSTER_FILENAME}: the reachability rule from '{from}' lists the entry member '{entry}' \
134 in `to`, but the entry member is never called: it runs the formation's own task from \
135 launch until the formation ends"
136 )]
137 RuleCallsEntryMember { from: String, entry: String },
138 #[error("{ROSTER_FILENAME}: member '{member}' ({coordinate}) cannot be admitted: {reason}")]
140 MemberUnresolvable {
141 member: String,
142 coordinate: String,
143 reason: String,
144 },
145 #[error(
147 "{ROSTER_FILENAME}: member '{member}': murmur.lock conflict for '{capsule}': {conflict}"
148 )]
149 MemberLockConflict {
150 member: String,
151 capsule: String,
152 conflict: String,
153 },
154 #[error(
155 "{ROSTER_FILENAME}: member '{member}' is listed in a reachability rule's `to`, but its \
156 manifest does not declare `exports.peer_tasks.accept: true`"
157 )]
158 CalleeDoesNotServePeers { member: String },
159 #[error(
160 "{ROSTER_FILENAME}: member '{member}' declares no `network.authentication`, and this \
161 roster declares peer traffic, so every member's door must require authentication"
162 )]
163 MemberNotAuthenticated { member: String },
164}
165
166fn quoted_list(names: &[String]) -> String {
167 names
168 .iter()
169 .map(|name| format!("'{name}'"))
170 .collect::<Vec<_>>()
171 .join(", ")
172}
173
174impl AdmittedRoster {
175 #[must_use]
178 pub fn source(&self) -> Option<&Path> {
179 self.source.as_deref()
180 }
181
182 #[must_use]
184 pub fn members(&self) -> &[AdmittedMember] {
185 &self.members
186 }
187
188 #[must_use]
189 pub fn member(&self, name: &str) -> Option<&AdmittedMember> {
190 self.members.iter().find(|member| member.name == name)
191 }
192
193 #[must_use]
197 pub fn entry(&self) -> &AdmittedMember {
198 &self.members[self.entry]
199 }
200
201 #[must_use]
203 pub fn reachability(&self) -> &RosterReachability {
204 &self.reachability
205 }
206
207 #[must_use]
210 pub fn edges(&self) -> &[RosterEdge] {
211 &self.edges
212 }
213
214 #[must_use]
216 pub fn may_call(&self, from: &str, to: &str) -> bool {
217 self.edges
218 .iter()
219 .any(|edge| edge.from == from && edge.to == to)
220 }
221
222 pub fn callees<'a>(&'a self, from: &'a str) -> impl Iterator<Item = &'a AdmittedMember> + 'a {
224 self.edges
225 .iter()
226 .filter(move |edge| edge.from == from)
227 .filter_map(|edge| self.member(&edge.to))
228 }
229
230 #[must_use]
232 pub fn declares_peer_traffic(&self) -> bool {
233 !self.edges.is_empty()
234 }
235
236 #[must_use]
242 pub fn caller_overflows(&self) -> Vec<CallerOverflow> {
243 self.members
244 .iter()
245 .filter_map(|member| {
246 let callers: Vec<String> = self
247 .edges
248 .iter()
249 .filter(|edge| edge.to == member.name)
250 .map(|edge| edge.from.clone())
251 .collect();
252 let lifecycle = member.manifest.effective_lifecycle();
253 let holds = lifecycle.tasks_held_at_once();
254 (callers.len() > holds).then(|| CallerOverflow {
255 member: member.name.clone(),
256 capsule: member.capsule.clone(),
257 version: member.version.clone(),
258 sufficient_queue_depth: callers.len().saturating_sub(1).max(1),
259 callers,
260 task_acceptance: lifecycle.task_acceptance,
261 queue_depth: lifecycle.queue_depth,
262 holds,
263 })
264 })
265 .collect()
266 }
267}
268
269const NEVER_TAKEN: &str =
272 "no call is ever taken: each is offered again until its deadline and then rejected";
273
274#[must_use]
277pub fn caller_overflow_warning(overflow: &CallerOverflow) -> String {
278 let count = overflow.callers.len();
279 let noun = if count == 1 { "member" } else { "members" };
280 let depth = overflow.sufficient_queue_depth;
281 let (holding, rejection, remedy) = match overflow.task_acceptance {
282 TaskAcceptance::Queue if overflow.queue_depth > 0 => (
283 format!(
284 "{} tasks at once — one running and lifecycle.queue_depth: {} waiting —",
285 overflow.holds, overflow.queue_depth
286 ),
287 "a call that arrives while it is full waits and is offered again, and is rejected only \
288 if it stays full until the call's deadline",
289 format!("lifecycle.queue_depth: {depth}"),
290 ),
291 TaskAcceptance::Queue => (
292 "no task — lifecycle.queue_depth: 0 —".to_string(),
293 NEVER_TAKEN,
294 format!("lifecycle.queue_depth: {depth}"),
295 ),
296 TaskAcceptance::Single => (
297 "1 task at once — lifecycle.task_acceptance: single —".to_string(),
298 "a call that arrives while it is busy waits and is offered again, and is rejected only \
299 if it stays busy until the call's deadline",
300 format!("lifecycle.task_acceptance: queue with lifecycle.queue_depth: {depth}"),
301 ),
302 TaskAcceptance::None => (
303 "no task — lifecycle.task_acceptance: none —".to_string(),
304 NEVER_TAKEN,
305 format!("lifecycle.task_acceptance: queue with lifecycle.queue_depth: {depth}"),
306 ),
307 };
308 format!(
309 "warning[{W_ROS_001}]: {ROSTER_FILENAME} lets {count} {noun} call '{}' ({}), but it holds \
310 {holding} so {rejection}; {remedy} would hold them all ({})",
311 overflow.member,
312 overflow.callers.join(", "),
313 roster_warning_link(W_ROS_001),
314 )
315}
316
317#[must_use]
319pub fn caller_overflow_fix(overflow: &CallerOverflow) -> String {
320 let depth = overflow.sufficient_queue_depth;
321 let setting = match overflow.task_acceptance {
322 TaskAcceptance::Queue => format!("lifecycle.queue_depth: {depth}"),
323 TaskAcceptance::Single | TaskAcceptance::None => {
324 format!("lifecycle.task_acceptance: queue and lifecycle.queue_depth: {depth}")
325 }
326 };
327 let held = match overflow.callers.len() {
328 1 => "the 1 member that may call it is held".to_string(),
329 count => format!("all {count} members that may call it are held"),
330 };
331 format!(
332 "{} ({}@{}): set {setting} in its murmur.yaml so {held} at once, or narrow its callers \
333 in {ROSTER_FILENAME} (warning[{W_ROS_001}])",
334 overflow.member, overflow.capsule, overflow.version,
335 )
336}
337
338pub fn admit_roster_file(
345 project_dir: &Path,
346 registry: &dyn Registry,
347 lock: Option<&MurmurLock>,
348) -> Result<AdmittedRoster, RosterRefusal> {
349 let path = resolve_roster_path(project_dir);
350 let roster = load_roster(&path).map_err(|error| RosterRefusal::Malformed {
351 message: error.to_string(),
352 })?;
353 let mut admitted = admit_roster(&roster, registry, lock)?;
354 admitted.source = Some(path);
355 Ok(admitted)
356}
357
358pub fn admit_roster(
361 roster: &Roster,
362 registry: &dyn Registry,
363 lock: Option<&MurmurLock>,
364) -> Result<AdmittedRoster, RosterRefusal> {
365 let mut seen = HashSet::new();
366 for member in &roster.members {
367 if !seen.insert(member.name.as_str()) {
368 return Err(RosterRefusal::DuplicateMember {
369 member: member.name.clone(),
370 });
371 }
372 }
373
374 let claimants: Vec<usize> = roster
375 .members
376 .iter()
377 .enumerate()
378 .filter(|(_, member)| member.entry)
379 .map(|(index, _)| index)
380 .collect();
381 let entry = match claimants.as_slice() {
382 [] => return Err(RosterRefusal::EntryNotDesignated),
383 [only] => *only,
384 several => {
385 return Err(RosterRefusal::EntryDesignatedMoreThanOnce {
386 members: several
387 .iter()
388 .map(|index| roster.members[*index].name.clone())
389 .collect(),
390 })
391 }
392 };
393
394 let index_of: HashMap<&str, usize> = roster
395 .members
396 .iter()
397 .enumerate()
398 .map(|(index, member)| (member.name.as_str(), index))
399 .collect();
400 if let RosterReachability::Rules(rules) = &roster.reachability {
401 for rule in rules {
402 for name in std::iter::once(&rule.from).chain(&rule.to) {
403 if !index_of.contains_key(name.as_str()) {
404 return Err(RosterRefusal::UnknownMemberInRule {
405 from: rule.from.clone(),
406 unknown: name.clone(),
407 });
408 }
409 }
410 }
411 let entry_name = &roster.members[entry].name;
412 if let Some(rule) = rules.iter().find(|rule| rule.to.contains(entry_name)) {
413 return Err(RosterRefusal::RuleCallsEntryMember {
414 from: rule.from.clone(),
415 entry: entry_name.clone(),
416 });
417 }
418 }
419
420 let platform = current_platform();
421 let mut members = Vec::with_capacity(roster.members.len());
422 for member in &roster.members {
423 let coordinate = format!("{}@{}", member.capsule, member.version);
424 let unresolvable = |reason: String| RosterRefusal::MemberUnresolvable {
425 member: member.name.clone(),
426 coordinate: coordinate.clone(),
427 reason,
428 };
429
430 let resolved = registry
431 .resolve_with_platform(&member.capsule, &member.version, Some(platform))
432 .map_err(|error| {
433 unresolvable(match error {
434 RegistryError::NotFound { .. } => {
435 "not installed in the project or global store".to_string()
436 }
437 other => other.to_string(),
438 })
439 })?;
440 let actual = sha256_hex(&resolved.bytes);
443 if actual != resolved.sha256 {
444 return Err(unresolvable(format!(
445 "the stored bytes hash to {actual}, but the store records {}",
446 resolved.sha256
447 )));
448 }
449 let manifest_yaml =
450 extract_manifest_yaml(&member.capsule, &member.version, &resolved.bytes)
451 .map_err(|error| unresolvable(error.to_string()))?;
452 let manifest = RuntimeManifest::from_yaml_str(&manifest_yaml).map_err(|error| {
453 unresolvable(format!("its packed murmur.yaml does not parse: {error}"))
454 })?;
455
456 if let Some(pinned) = lock.and_then(|lock| lock.artifact_for(&member.capsule)) {
457 let incoming = LockedSha256::for_resolved(&resolved, platform);
458 if let Some(conflict) = pinned.conflict_with(&member.version, &incoming) {
459 return Err(RosterRefusal::MemberLockConflict {
460 member: member.name.clone(),
461 capsule: member.capsule.clone(),
462 conflict,
463 });
464 }
465 }
466
467 members.push(AdmittedMember {
468 name: member.name.clone(),
469 capsule: member.capsule.clone(),
470 version: member.version.clone(),
471 entry: member.entry,
472 sha256: resolved.sha256,
473 manifest,
474 });
475 }
476
477 let pairs: BTreeSet<(usize, usize)> = match &roster.reachability {
478 RosterReachability::Closed => BTreeSet::new(),
479 RosterReachability::All => {
481 let serving: Vec<usize> = members
482 .iter()
483 .enumerate()
484 .filter(|(_, member)| member.manifest.accepts_peer_tasks())
485 .map(|(index, _)| index)
486 .collect();
487 serving
488 .iter()
489 .flat_map(|from| {
490 serving
491 .iter()
492 .filter(move |to| *to != from && **to != entry)
493 .map(move |to| (*from, *to))
494 })
495 .collect()
496 }
497 RosterReachability::Rules(rules) => {
498 let index_of = &index_of;
499 let pairs: BTreeSet<(usize, usize)> = rules
500 .iter()
501 .flat_map(|rule| {
502 let from = index_of[rule.from.as_str()];
503 rule.to.iter().map(move |to| (from, index_of[to.as_str()]))
504 })
505 .collect();
506 let callees: BTreeSet<usize> = pairs.iter().map(|(_, to)| *to).collect();
507 if let Some(refused) = callees
508 .into_iter()
509 .find(|to| !members[*to].manifest.accepts_peer_tasks())
510 {
511 return Err(RosterRefusal::CalleeDoesNotServePeers {
512 member: members[refused].name.clone(),
513 });
514 }
515 pairs
516 }
517 };
518
519 if !pairs.is_empty() {
520 if let Some(public) = members
521 .iter()
522 .find(|member| !member.requires_authentication())
523 {
524 return Err(RosterRefusal::MemberNotAuthenticated {
525 member: public.name.clone(),
526 });
527 }
528 }
529
530 let edges = pairs
531 .into_iter()
532 .map(|(from, to)| RosterEdge {
533 from: members[from].name.clone(),
534 to: members[to].name.clone(),
535 })
536 .collect();
537
538 Ok(AdmittedRoster {
539 source: None,
540 members,
541 entry,
542 reachability: roster.reachability.clone(),
543 edges,
544 })
545}
546
547#[cfg(test)]
548mod tests {
549 use std::collections::BTreeMap;
550 use std::fs;
551 use std::io::Write;
552
553 use murmur_artifact::{
554 ArtifactMeta, LocalRegistry, LockOrigin, LockedArtifact, ReachabilityRule, RosterMember,
555 RuntimeType, LOCK_VERSION,
556 };
557
558 use super::*;
559
560 const SERVES: &str = "exports:\n peer_tasks:\n accept: true\n";
561 const REFUSES: &str = "exports:\n peer_tasks:\n accept: false\n";
562 const AUTHENTICATED: &str = "network:\n authentication:\n scheme: bearer\n";
563
564 fn manifest(name: &str, version: &str, serves: bool, authenticated: bool) -> String {
565 let mut yaml = format!("name: {name}\nversion: {version}\nartifacts: []\n");
566 if serves {
567 yaml.push_str(SERVES);
568 }
569 if authenticated {
570 yaml.push_str(AUTHENTICATED);
571 }
572 yaml
573 }
574
575 fn pack(manifest_yaml: &str) -> Vec<u8> {
576 let mut cursor = std::io::Cursor::new(Vec::<u8>::new());
577 {
578 let mut zip = zip::ZipWriter::new(&mut cursor);
579 zip.start_file("murmur.yaml", zip::write::SimpleFileOptions::default())
580 .unwrap();
581 zip.write_all(manifest_yaml.as_bytes()).unwrap();
582 zip.finish().unwrap();
583 }
584 cursor.into_inner()
585 }
586
587 fn install(store: &LocalRegistry, name: &str, version: &str, manifest_yaml: &str) -> String {
589 let bytes = pack(manifest_yaml);
590 let sha256 = sha256_hex(&bytes);
591 store
592 .store_installed_overwrite(
593 ArtifactMeta {
594 name: name.to_string(),
595 version: version.to_string(),
596 runtime: RuntimeType::Wasm,
597 artifact_runtime: "capsule".to_string(),
598 platforms: Vec::new(),
599 description: None,
600 tags: Vec::new(),
601 wit_contracts: None,
602 },
603 &bytes,
604 &sha256,
605 )
606 .unwrap();
607 sha256
608 }
609
610 fn member(name: &str, version: &str, entry: bool) -> RosterMember {
611 RosterMember {
612 name: name.to_string(),
613 capsule: name.to_string(),
614 version: version.to_string(),
615 entry,
616 }
617 }
618
619 fn rule(from: &str, to: &[&str]) -> ReachabilityRule {
620 ReachabilityRule {
621 from: from.to_string(),
622 to: to.iter().map(ToString::to_string).collect(),
623 }
624 }
625
626 fn edges(admitted: &AdmittedRoster) -> Vec<(&str, &str)> {
627 admitted
628 .edges()
629 .iter()
630 .map(|edge| (edge.from.as_str(), edge.to.as_str()))
631 .collect()
632 }
633
634 fn lock_pinning(name: &str, version: &str, sha256: &str) -> MurmurLock {
635 MurmurLock {
636 lock_version: LOCK_VERSION,
637 artifacts: vec![LockedArtifact {
638 name: name.to_string(),
639 resolved_version: version.to_string(),
640 sha256: LockedSha256::any(sha256.to_string()),
641 origin: LockOrigin::Operator,
642 }],
643 }
644 }
645
646 struct Formation {
648 _dir: tempfile::TempDir,
649 store: LocalRegistry,
650 hashes: BTreeMap<&'static str, String>,
651 roster: Roster,
652 }
653
654 fn formation() -> Formation {
655 let dir = tempfile::tempdir().unwrap();
656 let store = LocalRegistry::new(dir.path().join("artifacts"));
657 let mut hashes = BTreeMap::new();
658 for (name, version) in [
659 ("planner", "0.3.0"),
660 ("coder", "1.2.0"),
661 ("reviewer", "0.9.0"),
662 ] {
663 hashes.insert(
664 name,
665 install(&store, name, version, &manifest(name, version, true, true)),
666 );
667 }
668 let roster = Roster {
669 members: vec![
670 member("planner", "0.3.0", true),
671 member("coder", "1.2.0", false),
672 member("reviewer", "0.9.0", false),
673 ],
674 reachability: RosterReachability::Rules(vec![
675 rule("planner", &["coder", "reviewer"]),
676 rule("reviewer", &["coder"]),
677 ]),
678 };
679 Formation {
680 _dir: dir,
681 store,
682 hashes,
683 roster,
684 }
685 }
686
687 fn refusal(roster: &Roster, store: &LocalRegistry, lock: Option<&MurmurLock>) -> RosterRefusal {
688 admit_roster(roster, store, lock).expect_err("the roster should be refused")
689 }
690
691 #[test]
692 fn s1_admits_members_in_order_bound_to_their_hashes_with_explicit_edges() {
693 let f = formation();
694 let admitted = admit_roster(&f.roster, &f.store, None).unwrap();
695 let names: Vec<_> = admitted.members().iter().map(|m| m.name.as_str()).collect();
696 assert_eq!(names, ["planner", "coder", "reviewer"]);
697 for member in admitted.members() {
698 assert_eq!(member.sha256, f.hashes[member.name.as_str()]);
699 assert_eq!(member.manifest.name, member.capsule);
700 assert!(member.manifest.accepts_peer_tasks());
701 }
702 assert_eq!(admitted.entry().name, "planner");
703 assert_eq!(
704 edges(&admitted),
705 [
706 ("planner", "coder"),
707 ("planner", "reviewer"),
708 ("reviewer", "coder")
709 ]
710 );
711 assert!(admitted.may_call("planner", "coder"));
712 assert!(!admitted.may_call("coder", "planner"));
713 assert!(admitted.declares_peer_traffic());
714 let callees: Vec<_> = admitted
715 .callees("planner")
716 .map(|m| m.name.as_str())
717 .collect();
718 assert_eq!(callees, ["coder", "reviewer"]);
719 assert_eq!(admitted.source(), None);
720 }
721
722 #[test]
723 fn repeated_edges_collapse_and_edges_sort_by_roster_index() {
724 let mut f = formation();
725 f.roster.reachability = RosterReachability::Rules(vec![
726 rule("reviewer", &["coder", "coder"]),
727 rule("planner", &["reviewer", "coder"]),
728 rule("planner", &["coder"]),
729 ]);
730 let admitted = admit_roster(&f.roster, &f.store, None).unwrap();
731 assert_eq!(
732 edges(&admitted),
733 [
734 ("planner", "coder"),
735 ("planner", "reviewer"),
736 ("reviewer", "coder")
737 ]
738 );
739 }
740
741 #[test]
742 fn s2_a_single_public_member_declares_no_traffic() {
743 let dir = tempfile::tempdir().unwrap();
744 let store = LocalRegistry::new(dir.path());
745 install(
746 &store,
747 "solo",
748 "1.0.0",
749 &manifest("solo", "1.0.0", false, false),
750 );
751 let roster = Roster {
752 members: vec![member("solo", "1.0.0", true)],
753 reachability: RosterReachability::Closed,
754 };
755 let admitted = admit_roster(&roster, &store, None).unwrap();
756 assert!(admitted.edges().is_empty());
757 assert!(!admitted.declares_peer_traffic());
758 assert_eq!(admitted.entry().name, "solo");
759 }
760
761 #[test]
762 fn s3_all_pairs_only_members_that_both_serve_peers() {
763 let dir = tempfile::tempdir().unwrap();
764 let store = LocalRegistry::new(dir.path());
765 for (name, serves) in [("a", true), ("b", true), ("c", false), ("d", false)] {
766 install(
767 &store,
768 name,
769 "1.0.0",
770 &manifest(name, "1.0.0", serves, true),
771 );
772 }
773 let roster = Roster {
774 members: ["a", "b", "c", "d"]
775 .iter()
776 .map(|name| member(name, "1.0.0", *name == "a"))
777 .collect(),
778 reachability: RosterReachability::All,
779 };
780 let admitted = admit_roster(&roster, &store, None).unwrap();
781 assert_eq!(edges(&admitted), [("a", "b")]);
783 assert!(matches!(admitted.reachability(), RosterReachability::All));
784
785 for (name, serves) in [("a", false), ("b", true), ("c", false), ("d", false)] {
787 install(
788 &store,
789 name,
790 "1.0.0",
791 &manifest(name, "1.0.0", serves, false),
792 );
793 }
794 let admitted = admit_roster(&roster, &store, None).unwrap();
795 assert!(admitted.edges().is_empty());
796 }
797
798 #[test]
799 fn s4_entry_must_be_designated_exactly_once() {
800 let mut f = formation();
801 f.roster.members[0].entry = false;
802 assert_eq!(
803 refusal(&f.roster, &f.store, None),
804 RosterRefusal::EntryNotDesignated
805 );
806
807 f.roster.members[0].entry = true;
808 f.roster.members[1].entry = true;
809 assert_eq!(
810 refusal(&f.roster, &f.store, None),
811 RosterRefusal::EntryDesignatedMoreThanOnce {
812 members: vec!["planner".to_string(), "coder".to_string()]
813 }
814 );
815 }
816
817 #[test]
818 fn s5_a_duplicate_name_is_refused_before_entry_and_before_any_lookup() {
819 let empty = tempfile::tempdir().unwrap();
820 let empty_store = LocalRegistry::new(empty.path());
821 let f = formation();
822 let roster = Roster {
823 members: vec![
824 member("coder", "1.2.0", true),
825 member("coder", "1.2.0", true),
826 ],
827 reachability: RosterReachability::Closed,
828 };
829 for store in [&f.store, &empty_store] {
830 assert_eq!(
831 refusal(&roster, store, None),
832 RosterRefusal::DuplicateMember {
833 member: "coder".to_string()
834 }
835 );
836 }
837 }
838
839 #[test]
840 fn s6_a_rule_naming_no_member_is_refused_naming_it() {
841 let mut f = formation();
842 f.roster.reachability =
843 RosterReachability::Rules(vec![rule("planner", &["coder", "tester"])]);
844 assert_eq!(
845 refusal(&f.roster, &f.store, None),
846 RosterRefusal::UnknownMemberInRule {
847 from: "planner".to_string(),
848 unknown: "tester".to_string()
849 }
850 );
851 f.roster.reachability = RosterReachability::Rules(vec![rule("ghost", &["coder"])]);
852 assert_eq!(
853 refusal(&f.roster, &f.store, None),
854 RosterRefusal::UnknownMemberInRule {
855 from: "ghost".to_string(),
856 unknown: "ghost".to_string()
857 }
858 );
859 }
860
861 #[test]
862 fn a_rule_calling_the_entry_member_is_refused_naming_both() {
863 let mut f = formation();
864 f.roster.reachability = RosterReachability::Rules(vec![
865 rule("planner", &["coder", "reviewer"]),
866 rule("reviewer", &["coder", "planner"]),
867 ]);
868 assert_eq!(
869 refusal(&f.roster, &f.store, None),
870 RosterRefusal::RuleCallsEntryMember {
871 from: "reviewer".to_string(),
872 entry: "planner".to_string()
873 }
874 );
875 }
876
877 #[test]
878 fn the_first_rule_calling_the_entry_member_in_file_order_is_named() {
879 let mut f = formation();
880 f.roster.reachability = RosterReachability::Rules(vec![
881 rule("planner", &["coder"]),
882 rule("coder", &["reviewer", "planner"]),
883 rule("reviewer", &["planner"]),
884 ]);
885 assert_eq!(
886 refusal(&f.roster, &f.store, None),
887 RosterRefusal::RuleCallsEntryMember {
888 from: "coder".to_string(),
889 entry: "planner".to_string()
890 }
891 );
892 }
893
894 #[test]
895 fn a_rule_calling_the_entry_member_is_refused_before_any_lookup() {
896 let empty = tempfile::tempdir().unwrap();
897 let empty_store = LocalRegistry::new(empty.path());
898 let mut f = formation();
899 f.roster.reachability =
900 RosterReachability::Rules(vec![rule("reviewer", &["coder", "planner"])]);
901 assert_eq!(
902 refusal(&f.roster, &empty_store, None),
903 RosterRefusal::RuleCallsEntryMember {
904 from: "reviewer".to_string(),
905 entry: "planner".to_string()
906 }
907 );
908
909 install(
912 &f.store,
913 "planner",
914 "0.3.0",
915 &manifest("planner", "0.3.0", false, true),
916 );
917 assert_eq!(
918 refusal(&f.roster, &f.store, None),
919 RosterRefusal::RuleCallsEntryMember {
920 from: "reviewer".to_string(),
921 entry: "planner".to_string()
922 }
923 );
924 }
925
926 #[test]
927 fn an_unknown_rule_name_is_refused_before_a_rule_calling_the_entry_member() {
928 let mut f = formation();
929 for rules in [
930 vec![rule("ghost", &["coder"]), rule("reviewer", &["planner"])],
931 vec![rule("reviewer", &["planner"]), rule("coder", &["ghost"])],
932 ] {
933 f.roster.reachability = RosterReachability::Rules(rules);
934 assert!(matches!(
935 refusal(&f.roster, &f.store, None),
936 RosterRefusal::UnknownMemberInRule { unknown, .. } if unknown == "ghost"
937 ));
938 }
939 }
940
941 #[test]
942 fn no_admitted_edge_ever_leads_into_the_entry_member() {
943 let mut f = formation();
944 let reachabilities = [
945 RosterReachability::All,
946 f.roster.reachability.clone(),
947 RosterReachability::Rules(vec![
948 rule("reviewer", &["coder", "coder"]),
949 rule("planner", &["reviewer", "coder"]),
950 rule("planner", &["coder"]),
951 ]),
952 RosterReachability::Rules(vec![
953 rule("planner", &["coder"]),
954 rule("reviewer", &["coder"]),
955 ]),
956 ];
957 for reachability in reachabilities {
958 f.roster.reachability = reachability;
959 let admitted = admit_roster(&f.roster, &f.store, None).unwrap();
960 let entry = admitted.entry().name.clone();
961 assert!(admitted.declares_peer_traffic());
962 assert!(
963 admitted.edges().iter().all(|edge| edge.to != entry),
964 "{:?}",
965 admitted.edges()
966 );
967 for member in admitted.members() {
968 assert!(admitted.callees(&member.name).all(|callee| !callee.entry));
969 }
970 }
971
972 f.roster.reachability = RosterReachability::All;
974 for entry in 0..f.roster.members.len() {
975 for (index, member) in f.roster.members.iter_mut().enumerate() {
976 member.entry = index == entry;
977 }
978 let admitted = admit_roster(&f.roster, &f.store, None).unwrap();
979 let entry = admitted.entry().name.as_str();
980 let pairs = edges(&admitted);
981 assert_eq!(pairs.len(), 4, "{pairs:?}");
982 assert!(pairs.iter().all(|(_, to)| *to != entry), "{pairs:?}");
983 assert_eq!(pairs.iter().filter(|(from, _)| *from == entry).count(), 2);
984 }
985 }
986
987 #[test]
988 fn s7_a_callee_must_serve_peers_but_a_caller_need_not() {
989 for reviewer_manifest in [
990 "name: reviewer\nversion: 0.9.0\nartifacts: []\nnetwork:\n authentication:\n scheme: bearer\n"
991 .to_string(),
992 format!("name: reviewer\nversion: 0.9.0\nartifacts: []\n{REFUSES}{AUTHENTICATED}"),
993 ] {
994 let mut f = formation();
995 install(&f.store, "reviewer", "0.9.0", &reviewer_manifest);
996 assert_eq!(
997 refusal(&f.roster, &f.store, None),
998 RosterRefusal::CalleeDoesNotServePeers {
999 member: "reviewer".to_string()
1000 }
1001 );
1002 f.roster.reachability = RosterReachability::Rules(vec![
1003 rule("planner", &["coder"]),
1004 rule("reviewer", &["coder"]),
1005 ]);
1006 admit_roster(&f.roster, &f.store, None).unwrap();
1007 }
1008 }
1009
1010 #[test]
1011 fn s8_with_traffic_every_member_needs_authentication_even_off_every_edge() {
1012 let f = formation();
1013 install(
1014 &f.store,
1015 "coder",
1016 "1.2.0",
1017 &manifest("coder", "1.2.0", true, false),
1018 );
1019 assert_eq!(
1020 refusal(&f.roster, &f.store, None),
1021 RosterRefusal::MemberNotAuthenticated {
1022 member: "coder".to_string()
1023 }
1024 );
1025
1026 let mut f = formation();
1027 install(
1028 &f.store,
1029 "notes",
1030 "1.0.0",
1031 &manifest("notes", "1.0.0", false, false),
1032 );
1033 f.roster.members.push(member("notes", "1.0.0", false));
1034 assert_eq!(
1035 refusal(&f.roster, &f.store, None),
1036 RosterRefusal::MemberNotAuthenticated {
1037 member: "notes".to_string()
1038 }
1039 );
1040 }
1041
1042 #[test]
1043 fn s9_an_uninstalled_or_unreadable_member_is_refused_naming_its_coordinate() {
1044 let mut f = formation();
1045 f.roster.members[1].version = "9.9.9".to_string();
1046 match refusal(&f.roster, &f.store, None) {
1047 RosterRefusal::MemberUnresolvable {
1048 member, coordinate, ..
1049 } => {
1050 assert_eq!(member, "coder");
1051 assert_eq!(coordinate, "coder@9.9.9");
1052 }
1053 other => panic!("unexpected refusal {other:?}"),
1054 }
1055
1056 let f = formation();
1057 install(&f.store, "coder", "1.2.0", "name: [unclosed\n");
1058 match refusal(&f.roster, &f.store, None) {
1059 RosterRefusal::MemberUnresolvable {
1060 member, coordinate, ..
1061 } => {
1062 assert_eq!(member, "coder");
1063 assert_eq!(coordinate, "coder@1.2.0");
1064 }
1065 other => panic!("unexpected refusal {other:?}"),
1066 }
1067 }
1068
1069 #[test]
1070 fn bytes_that_do_not_match_the_store_hash_are_refused() {
1071 let f = formation();
1072 let artifact = f
1073 .store
1074 .root()
1075 .join("coder")
1076 .join("1.2.0")
1077 .join("coder-1.2.0.mur.zip");
1078 fs::write(&artifact, pack(&manifest("coder", "1.2.0", true, false))).unwrap();
1079 assert!(matches!(
1080 refusal(&f.roster, &f.store, None),
1081 RosterRefusal::MemberUnresolvable { member, .. } if member == "coder"
1082 ));
1083 }
1084
1085 #[test]
1086 fn s10_the_lock_verifies_a_member_it_pins() {
1087 let f = formation();
1088 let agreeing = lock_pinning("coder", "1.2.0", &f.hashes["coder"]);
1089 admit_roster(&f.roster, &f.store, Some(&agreeing)).unwrap();
1090
1091 for lock in [
1092 lock_pinning("coder", "1.2.0", &"0".repeat(64)),
1093 lock_pinning("coder", "1.1.0", &f.hashes["coder"]),
1094 ] {
1095 match refusal(&f.roster, &f.store, Some(&lock)) {
1096 RosterRefusal::MemberLockConflict {
1097 member,
1098 capsule,
1099 conflict,
1100 } => {
1101 assert_eq!(member, "coder");
1102 assert_eq!(capsule, "coder");
1103 assert!(!conflict.is_empty());
1104 }
1105 other => panic!("unexpected refusal {other:?}"),
1106 }
1107 }
1108 }
1109
1110 #[test]
1111 fn s13_a_duplicate_and_an_unknown_rule_name_is_always_the_duplicate() {
1112 let f = formation();
1113 let roster = Roster {
1114 members: vec![
1115 member("coder", "1.2.0", true),
1116 member("coder", "1.2.0", false),
1117 ],
1118 reachability: RosterReachability::Rules(vec![rule("coder", &["ghost"])]),
1119 };
1120 for _ in 0..5 {
1121 assert!(matches!(
1122 refusal(&roster, &f.store, None),
1123 RosterRefusal::DuplicateMember { .. }
1124 ));
1125 }
1126 }
1127
1128 #[test]
1129 fn admit_roster_file_maps_every_load_error_to_malformed_and_records_the_source() {
1130 let f = formation();
1131 let project = tempfile::tempdir().unwrap();
1132 assert!(matches!(
1133 admit_roster_file(project.path(), &f.store, None),
1134 Err(RosterRefusal::Malformed { message }) if message.contains("roster.yaml not found")
1135 ));
1136
1137 fs::write(
1138 resolve_roster_path(project.path()),
1139 "roster_version: 2\nmembers: []\n",
1140 )
1141 .unwrap();
1142 assert!(matches!(
1143 admit_roster_file(project.path(), &f.store, None),
1144 Err(RosterRefusal::Malformed { message }) if message.contains("roster_version")
1145 ));
1146
1147 fs::write(
1148 resolve_roster_path(project.path()),
1149 "roster_version: 1\nmembers:\n - {name: planner, capsule: planner, version: 0.3.0, entry: true}\n",
1150 )
1151 .unwrap();
1152 let admitted = admit_roster_file(project.path(), &f.store, None).unwrap();
1153 assert_eq!(
1154 admitted.source(),
1155 Some(resolve_roster_path(project.path()).as_path())
1156 );
1157 }
1158
1159 #[test]
1160 fn every_refusal_names_roster_yaml() {
1161 for refusal in [
1162 RosterRefusal::DuplicateMember { member: "a".into() },
1163 RosterRefusal::EntryNotDesignated,
1164 RosterRefusal::EntryDesignatedMoreThanOnce {
1165 members: vec!["a".into(), "b".into()],
1166 },
1167 RosterRefusal::UnknownMemberInRule {
1168 from: "a".into(),
1169 unknown: "b".into(),
1170 },
1171 RosterRefusal::RuleCallsEntryMember {
1172 from: "a".into(),
1173 entry: "b".into(),
1174 },
1175 RosterRefusal::MemberUnresolvable {
1176 member: "a".into(),
1177 coordinate: "a@1".into(),
1178 reason: "r".into(),
1179 },
1180 RosterRefusal::MemberLockConflict {
1181 member: "a".into(),
1182 capsule: "a".into(),
1183 conflict: "c".into(),
1184 },
1185 RosterRefusal::CalleeDoesNotServePeers { member: "a".into() },
1186 RosterRefusal::MemberNotAuthenticated { member: "a".into() },
1187 ] {
1188 assert!(refusal.to_string().starts_with("roster.yaml"), "{refusal}");
1189 }
1190 }
1191
1192 const TEN_CALLERS_WARNING: &str =
1193 "warning[W-ROS-001]: roster.yaml lets 10 members call 'reviewer' \
1194 (w01, w02, w03, w04, w05, w06, w07, w08, w09, w10), but it holds 2 tasks at once — one \
1195 running and lifecycle.queue_depth: 1 waiting — so a call that arrives while it is full \
1196 waits and is offered again, and is rejected only if it stays full until the call's \
1197 deadline; lifecycle.queue_depth: 9 would hold them all \
1198 (https://docs.murmur.nexus/reference/diagnostics/#w-ros-001)";
1199
1200 fn manifest_with_lifecycle(name: &str, lifecycle_yaml: &str) -> String {
1203 let mut yaml = manifest(name, "1.0.0", true, true);
1204 if !lifecycle_yaml.is_empty() {
1205 yaml.push_str(&format!("lifecycle:\n {lifecycle_yaml}\n"));
1206 }
1207 yaml
1208 }
1209
1210 const QUEUE_DEPTH_1: &str = "task_acceptance: queue\n after_task: sleep";
1211
1212 fn queue_at(depth: usize) -> String {
1213 format!("task_acceptance: queue\n after_task: sleep\n queue_depth: {depth}")
1214 }
1215
1216 fn admit_with(
1218 members: &[(&str, &str)],
1219 reachability: RosterReachability,
1220 ) -> (tempfile::TempDir, AdmittedRoster) {
1221 let dir = tempfile::tempdir().unwrap();
1222 let store = LocalRegistry::new(dir.path());
1223 for (name, lifecycle) in members {
1224 install(
1225 &store,
1226 name,
1227 "1.0.0",
1228 &manifest_with_lifecycle(name, lifecycle),
1229 );
1230 }
1231 let roster = Roster {
1232 members: members
1233 .iter()
1234 .enumerate()
1235 .map(|(index, (name, _))| member(name, "1.0.0", index == 0))
1236 .collect(),
1237 reachability,
1238 };
1239 let admitted = admit_roster(&roster, &store, None).unwrap();
1240 (dir, admitted)
1241 }
1242
1243 const WORKERS: [&str; 10] = [
1244 "w01", "w02", "w03", "w04", "w05", "w06", "w07", "w08", "w09", "w10",
1245 ];
1246
1247 fn ten_workers_one_reviewer(reviewer_lifecycle: &str) -> (tempfile::TempDir, AdmittedRoster) {
1249 let mut members = vec![("lead", "")];
1250 members.extend(WORKERS.iter().map(|worker| (*worker, QUEUE_DEPTH_1)));
1251 members.push(("reviewer", reviewer_lifecycle));
1252 let mut rules = vec![rule("lead", &WORKERS)];
1253 rules.extend(WORKERS.iter().map(|worker| rule(worker, &["reviewer"])));
1254 admit_with(&members, RosterReachability::Rules(rules))
1255 }
1256
1257 #[test]
1258 fn ten_callers_overflow_a_reviewer_at_depth_one() {
1259 let (_dir, admitted) = ten_workers_one_reviewer(QUEUE_DEPTH_1);
1260 assert_eq!(
1261 admitted.caller_overflows(),
1262 [CallerOverflow {
1263 member: "reviewer".to_string(),
1264 capsule: "reviewer".to_string(),
1265 version: "1.0.0".to_string(),
1266 callers: WORKERS.iter().map(ToString::to_string).collect(),
1267 task_acceptance: TaskAcceptance::Queue,
1268 queue_depth: 1,
1269 holds: 2,
1270 sufficient_queue_depth: 9,
1271 }]
1272 );
1273 }
1274
1275 #[test]
1276 fn the_ten_callers_warning_line_is_exact() {
1277 let (_dir, admitted) = ten_workers_one_reviewer(QUEUE_DEPTH_1);
1278 let overflows = admitted.caller_overflows();
1279 assert_eq!(caller_overflow_warning(&overflows[0]), TEN_CALLERS_WARNING);
1280 assert_eq!(
1281 caller_overflow_fix(&overflows[0]),
1282 "reviewer (reviewer@1.0.0): set lifecycle.queue_depth: 9 in its murmur.yaml so all 10 \
1283 members that may call it are held at once, or narrow its callers in roster.yaml \
1284 (warning[W-ROS-001])"
1285 );
1286 }
1287
1288 #[test]
1289 fn a_reviewer_deep_enough_for_every_caller_does_not_overflow() {
1290 for depth in [9, 10] {
1291 let (_dir, admitted) = ten_workers_one_reviewer(&queue_at(depth));
1292 assert!(
1293 admitted.caller_overflows().is_empty(),
1294 "depth {depth} holds all ten callers"
1295 );
1296 }
1297 }
1298
1299 #[test]
1300 fn all_counts_every_expanded_caller_and_never_the_entry_member() {
1301 for (lifecycle, overflows) in [
1302 (String::new(), true),
1303 (QUEUE_DEPTH_1.to_string(), true),
1304 (queue_at(2), false),
1305 ] {
1306 let members: Vec<(&str, &str)> = ["a", "b", "c", "d"]
1307 .iter()
1308 .map(|name| (*name, lifecycle.as_str()))
1309 .collect();
1310 let (_dir, admitted) = admit_with(&members, RosterReachability::All);
1311 let found = admitted.caller_overflows();
1312 if !overflows {
1313 assert!(found.is_empty(), "'{lifecycle}' holds three callers");
1314 continue;
1315 }
1316 let named: Vec<_> = found.iter().map(|o| o.member.as_str()).collect();
1317 assert_eq!(named, ["b", "c", "d"], "'{lifecycle}'");
1318 for overflow in &found {
1319 assert_eq!(overflow.callers.len(), 3, "{overflow:?}");
1320 assert!(!overflow.callers.contains(&overflow.member));
1321 assert_eq!(overflow.sufficient_queue_depth, 2);
1322 }
1323 assert_eq!(found[0].callers, ["a", "c", "d"]);
1324 }
1325 }
1326
1327 #[test]
1328 fn a_single_member_with_two_callers_is_told_to_switch_to_queue() {
1329 let (_dir, admitted) = admit_with(
1330 &[("lead", ""), ("helper", ""), ("coder", "")],
1331 RosterReachability::Rules(vec![
1332 rule("lead", &["coder", "helper"]),
1333 rule("helper", &["coder"]),
1334 ]),
1335 );
1336 let overflows = admitted.caller_overflows();
1337 assert_eq!(overflows.len(), 1);
1338 let overflow = &overflows[0];
1339 assert_eq!(overflow.member, "coder");
1340 assert_eq!(overflow.callers, ["lead", "helper"]);
1341 assert_eq!(overflow.task_acceptance, TaskAcceptance::Single);
1342 assert_eq!(overflow.holds, 1);
1343 assert_eq!(overflow.sufficient_queue_depth, 1);
1344 assert_eq!(
1345 caller_overflow_warning(overflow),
1346 "warning[W-ROS-001]: roster.yaml lets 2 members call 'coder' (lead, helper), but it \
1347 holds 1 task at once — lifecycle.task_acceptance: single — so a call that arrives \
1348 while it is busy waits and is offered again, and is rejected only if it stays busy \
1349 until the call's deadline; lifecycle.task_acceptance: queue with \
1350 lifecycle.queue_depth: 1 would hold them all \
1351 (https://docs.murmur.nexus/reference/diagnostics/#w-ros-001)"
1352 );
1353 assert_eq!(
1354 caller_overflow_fix(overflow),
1355 "coder (coder@1.0.0): set lifecycle.task_acceptance: queue and \
1356 lifecycle.queue_depth: 1 in its murmur.yaml so all 2 members that may call it are \
1357 held at once, or narrow its callers in roster.yaml (warning[W-ROS-001])"
1358 );
1359 }
1360
1361 #[test]
1362 fn a_member_that_accepts_no_task_overflows_with_one_caller() {
1363 let (_dir, admitted) = admit_with(
1364 &[("lead", ""), ("idle", "task_acceptance: none")],
1365 RosterReachability::Rules(vec![rule("lead", &["idle"])]),
1366 );
1367 let overflows = admitted.caller_overflows();
1368 assert_eq!(overflows.len(), 1);
1369 assert_eq!(overflows[0].holds, 0);
1370 assert_eq!(
1371 caller_overflow_warning(&overflows[0]),
1372 "warning[W-ROS-001]: roster.yaml lets 1 member call 'idle' (lead), but it holds no \
1373 task — lifecycle.task_acceptance: none — so no call is ever taken: each is offered \
1374 again until its deadline and then rejected; \
1375 lifecycle.task_acceptance: queue with lifecycle.queue_depth: 1 would hold them all \
1376 (https://docs.murmur.nexus/reference/diagnostics/#w-ros-001)"
1377 );
1378 assert_eq!(
1379 caller_overflow_fix(&overflows[0]),
1380 "idle (idle@1.0.0): set lifecycle.task_acceptance: queue and lifecycle.queue_depth: 1 \
1381 in its murmur.yaml so the 1 member that may call it is held at once, or narrow its \
1382 callers in roster.yaml (warning[W-ROS-001])"
1383 );
1384 }
1385
1386 #[test]
1387 fn a_queue_member_at_depth_zero_holds_no_task() {
1388 let (_dir, admitted) = admit_with(
1389 &[("lead", ""), ("shut", &queue_at(0))],
1390 RosterReachability::Rules(vec![rule("lead", &["shut"])]),
1391 );
1392 let overflows = admitted.caller_overflows();
1393 assert_eq!(overflows.len(), 1);
1394 assert_eq!(
1395 caller_overflow_warning(&overflows[0]),
1396 "warning[W-ROS-001]: roster.yaml lets 1 member call 'shut' (lead), but it holds no \
1397 task — lifecycle.queue_depth: 0 — so no call is ever taken: each is offered again \
1398 until its deadline and then rejected; \
1399 lifecycle.queue_depth: 1 would hold them all \
1400 (https://docs.murmur.nexus/reference/diagnostics/#w-ros-001)"
1401 );
1402 }
1403
1404 #[test]
1407 fn the_entry_member_is_never_named() {
1408 let (_dir, admitted) = admit_with(
1409 &[
1410 ("lead", "task_acceptance: none"),
1411 ("a", ""),
1412 ("b", &queue_at(5)),
1413 ],
1414 RosterReachability::All,
1415 );
1416 assert!(!admitted.edges().is_empty());
1417 assert!(admitted
1418 .caller_overflows()
1419 .iter()
1420 .all(|overflow| overflow.member != "lead"));
1421 }
1422
1423 #[test]
1424 fn a_roster_with_no_edges_has_no_overflow() {
1425 let (_dir, admitted) = admit_with(
1426 &[
1427 ("lead", "task_acceptance: none"),
1428 ("idle", "task_acceptance: none"),
1429 ],
1430 RosterReachability::Closed,
1431 );
1432 assert!(admitted.edges().is_empty());
1433 assert!(admitted.caller_overflows().is_empty());
1434 }
1435}