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