1use std::collections::{BTreeSet, HashMap, HashSet};
27use std::path::{Path, PathBuf};
28
29use murmur_artifact::{
30 load_roster, resolve_roster_path, roster_warning_link, sha256_hex, ArtifactName, LockedSha256,
31 MurmurLock, Registry, RegistryError, Roster, RosterError, 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 {
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 by a peer: it takes the formation's tasks \
138 from the launch or from a client at its door"
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 mut members = Vec::with_capacity(roster.members.len());
426 for member in &roster.members {
427 let coordinate = format!("{}@{}", member.capsule, member.version);
428 let unresolvable = |reason: String| RosterRefusal::MemberUnresolvable {
429 member: member.name.clone(),
430 coordinate: coordinate.clone(),
431 reason,
432 };
433
434 let capsule = ArtifactName::parse(&member.capsule)
435 .map_err(|error| unresolvable(error.to_string()))?;
436 let resolved = registry
437 .resolve(&capsule, &member.version)
438 .map_err(|error| {
439 unresolvable(match error {
440 RegistryError::NotFound { .. } => {
441 "not installed in the project or global store".to_string()
442 }
443 other => other.to_string(),
444 })
445 })?;
446 let actual = sha256_hex(&resolved.bytes);
449 if actual != resolved.sha256 {
450 return Err(unresolvable(format!(
451 "the stored bytes hash to {actual}, but the store records {}",
452 resolved.sha256
453 )));
454 }
455 let manifest_yaml =
456 extract_manifest_yaml(&member.capsule, &member.version, &resolved.bytes)
457 .map_err(|error| unresolvable(error.to_string()))?;
458 let manifest = RuntimeManifest::from_yaml_str(&manifest_yaml).map_err(|error| {
459 unresolvable(format!("its packed murmur.yaml does not parse: {error}"))
460 })?;
461
462 if let Some(pinned) = lock.and_then(|lock| lock.artifact_for(&member.capsule)) {
463 let incoming = LockedSha256::any(resolved.sha256.clone());
464 if let Some(conflict) = pinned.conflict_with(&member.version, &incoming) {
465 return Err(RosterRefusal::MemberLockConflict {
466 member: member.name.clone(),
467 capsule: member.capsule.clone(),
468 conflict,
469 });
470 }
471 }
472
473 members.push(AdmittedMember {
474 name: member.name.clone(),
475 capsule: member.capsule.clone(),
476 version: member.version.clone(),
477 entry: member.entry,
478 sha256: resolved.sha256,
479 manifest,
480 });
481 }
482
483 let pairs: BTreeSet<(usize, usize)> = match &roster.reachability {
484 RosterReachability::Closed => BTreeSet::new(),
485 RosterReachability::All => {
487 let serving: Vec<usize> = members
488 .iter()
489 .enumerate()
490 .filter(|(_, member)| member.manifest.accepts_peer_tasks())
491 .map(|(index, _)| index)
492 .collect();
493 serving
494 .iter()
495 .flat_map(|from| {
496 serving
497 .iter()
498 .filter(move |to| *to != from && **to != entry)
499 .map(move |to| (*from, *to))
500 })
501 .collect()
502 }
503 RosterReachability::Rules(rules) => {
504 let index_of = &index_of;
505 let pairs: BTreeSet<(usize, usize)> = rules
506 .iter()
507 .flat_map(|rule| {
508 let from = index_of[rule.from.as_str()];
509 rule.to.iter().map(move |to| (from, index_of[to.as_str()]))
510 })
511 .collect();
512 let callees: BTreeSet<usize> = pairs.iter().map(|(_, to)| *to).collect();
513 if let Some(refused) = callees
514 .into_iter()
515 .find(|to| !members[*to].manifest.accepts_peer_tasks())
516 {
517 return Err(RosterRefusal::CalleeDoesNotServePeers {
518 member: members[refused].name.clone(),
519 });
520 }
521 pairs
522 }
523 };
524
525 if !pairs.is_empty() {
526 if let Some(public) = members
527 .iter()
528 .find(|member| !member.requires_authentication())
529 {
530 return Err(RosterRefusal::MemberNotAuthenticated {
531 member: public.name.clone(),
532 });
533 }
534 }
535
536 let edges = pairs
537 .into_iter()
538 .map(|(from, to)| RosterEdge {
539 from: members[from].name.clone(),
540 to: members[to].name.clone(),
541 })
542 .collect();
543
544 Ok(AdmittedRoster {
545 source: None,
546 members,
547 entry,
548 reachability: roster.reachability.clone(),
549 edges,
550 })
551}
552
553#[cfg(test)]
554mod tests {
555 use std::collections::BTreeMap;
556 use std::fs;
557 use std::io::Write;
558
559 use murmur_artifact::{
560 ArtifactMeta, LocalRegistry, LockOrigin, LockedArtifact, ReachabilityRule, RosterMember,
561 RuntimeType, LOCK_VERSION,
562 };
563
564 use super::*;
565
566 const SERVES: &str = "exports:\n peer_tasks:\n accept: true\n";
567 const REFUSES: &str = "exports:\n peer_tasks:\n accept: false\n";
568 const AUTHENTICATED: &str = "network:\n authentication:\n scheme: bearer\n";
569
570 fn manifest(name: &str, version: &str, serves: bool, authenticated: bool) -> String {
571 let mut yaml = format!("name: {name}\nversion: {version}\nartifacts: []\n");
572 if serves {
573 yaml.push_str(SERVES);
574 }
575 if authenticated {
576 yaml.push_str(AUTHENTICATED);
577 }
578 yaml
579 }
580
581 fn pack(manifest_yaml: &str) -> Vec<u8> {
582 let mut cursor = std::io::Cursor::new(Vec::<u8>::new());
583 {
584 let mut zip = zip::ZipWriter::new(&mut cursor);
585 zip.start_file("murmur.yaml", zip::write::SimpleFileOptions::default())
586 .unwrap();
587 zip.write_all(manifest_yaml.as_bytes()).unwrap();
588 zip.finish().unwrap();
589 }
590 cursor.into_inner()
591 }
592
593 fn install(store: &LocalRegistry, name: &str, version: &str, manifest_yaml: &str) -> String {
596 let bytes = pack(manifest_yaml);
597 let sha256 = sha256_hex(&bytes);
598 store
599 .store_installed_overwrite(
600 ArtifactMeta {
601 name: ArtifactName::from_parts("local", name).unwrap(),
602 version: version.to_string(),
603 runtime: RuntimeType::Wasm,
604 artifact_runtime: "capsule".to_string(),
605 description: None,
606 tags: Vec::new(),
607 wit_contracts: None,
608 owner_id: None,
609 built_at_level: None,
610 },
611 &bytes,
612 &sha256,
613 )
614 .unwrap();
615 sha256
616 }
617
618 fn member(name: &str, version: &str, entry: bool) -> RosterMember {
620 RosterMember {
621 name: name.to_string(),
622 capsule: format!("local/{name}"),
623 version: version.to_string(),
624 entry,
625 }
626 }
627
628 fn rule(from: &str, to: &[&str]) -> ReachabilityRule {
629 ReachabilityRule {
630 from: from.to_string(),
631 to: to.iter().map(ToString::to_string).collect(),
632 }
633 }
634
635 fn edges(admitted: &AdmittedRoster) -> Vec<(&str, &str)> {
636 admitted
637 .edges()
638 .iter()
639 .map(|edge| (edge.from.as_str(), edge.to.as_str()))
640 .collect()
641 }
642
643 fn lock_pinning(name: &str, version: &str, sha256: &str) -> MurmurLock {
644 MurmurLock {
645 lock_version: LOCK_VERSION,
646 artifacts: vec![LockedArtifact {
647 name: format!("local/{name}"),
648 resolved_version: version.to_string(),
649 sha256: LockedSha256::any(sha256.to_string()),
650 origin: LockOrigin::Operator,
651 owner_id: None,
652 }],
653 }
654 }
655
656 struct Formation {
658 _dir: tempfile::TempDir,
659 store: LocalRegistry,
660 hashes: BTreeMap<&'static str, String>,
661 roster: Roster,
662 }
663
664 fn formation() -> Formation {
665 let dir = tempfile::tempdir().unwrap();
666 let store = LocalRegistry::new(dir.path().join("artifacts"));
667 let mut hashes = BTreeMap::new();
668 for (name, version) in [
669 ("planner", "0.3.0"),
670 ("coder", "1.2.0"),
671 ("reviewer", "0.9.0"),
672 ] {
673 hashes.insert(
674 name,
675 install(&store, name, version, &manifest(name, version, true, true)),
676 );
677 }
678 let roster = Roster {
679 members: vec![
680 member("planner", "0.3.0", true),
681 member("coder", "1.2.0", false),
682 member("reviewer", "0.9.0", false),
683 ],
684 reachability: RosterReachability::Rules(vec![
685 rule("planner", &["coder", "reviewer"]),
686 rule("reviewer", &["coder"]),
687 ]),
688 };
689 Formation {
690 _dir: dir,
691 store,
692 hashes,
693 roster,
694 }
695 }
696
697 fn refusal(roster: &Roster, store: &LocalRegistry, lock: Option<&MurmurLock>) -> RosterRefusal {
698 admit_roster(roster, store, lock).expect_err("the roster should be refused")
699 }
700
701 #[test]
702 fn s1_admits_members_in_order_bound_to_their_hashes_with_explicit_edges() {
703 let f = formation();
704 let admitted = admit_roster(&f.roster, &f.store, None).unwrap();
705 let names: Vec<_> = admitted.members().iter().map(|m| m.name.as_str()).collect();
706 assert_eq!(names, ["planner", "coder", "reviewer"]);
707 for member in admitted.members() {
708 assert_eq!(member.sha256, f.hashes[member.name.as_str()]);
709 assert_eq!(
710 member.manifest.name,
711 murmur_artifact::tool_name_of(&member.capsule)
712 );
713 assert!(member.manifest.accepts_peer_tasks());
714 }
715 assert_eq!(admitted.entry().name, "planner");
716 assert_eq!(
717 edges(&admitted),
718 [
719 ("planner", "coder"),
720 ("planner", "reviewer"),
721 ("reviewer", "coder")
722 ]
723 );
724 assert!(admitted.may_call("planner", "coder"));
725 assert!(!admitted.may_call("coder", "planner"));
726 assert!(admitted.declares_peer_traffic());
727 let callees: Vec<_> = admitted
728 .callees("planner")
729 .map(|m| m.name.as_str())
730 .collect();
731 assert_eq!(callees, ["coder", "reviewer"]);
732 assert_eq!(admitted.source(), None);
733 }
734
735 #[test]
736 fn repeated_edges_collapse_and_edges_sort_by_roster_index() {
737 let mut f = formation();
738 f.roster.reachability = RosterReachability::Rules(vec![
739 rule("reviewer", &["coder", "coder"]),
740 rule("planner", &["reviewer", "coder"]),
741 rule("planner", &["coder"]),
742 ]);
743 let admitted = admit_roster(&f.roster, &f.store, None).unwrap();
744 assert_eq!(
745 edges(&admitted),
746 [
747 ("planner", "coder"),
748 ("planner", "reviewer"),
749 ("reviewer", "coder")
750 ]
751 );
752 }
753
754 #[test]
755 fn s2_a_single_public_member_declares_no_traffic() {
756 let dir = tempfile::tempdir().unwrap();
757 let store = LocalRegistry::new(dir.path());
758 install(
759 &store,
760 "solo",
761 "1.0.0",
762 &manifest("solo", "1.0.0", false, false),
763 );
764 let roster = Roster {
765 members: vec![member("solo", "1.0.0", true)],
766 reachability: RosterReachability::Closed,
767 };
768 let admitted = admit_roster(&roster, &store, None).unwrap();
769 assert!(admitted.edges().is_empty());
770 assert!(!admitted.declares_peer_traffic());
771 assert_eq!(admitted.entry().name, "solo");
772 }
773
774 #[test]
775 fn s3_all_pairs_only_members_that_both_serve_peers() {
776 let dir = tempfile::tempdir().unwrap();
777 let store = LocalRegistry::new(dir.path());
778 for (name, serves) in [("a", true), ("b", true), ("c", false), ("d", false)] {
779 install(
780 &store,
781 name,
782 "1.0.0",
783 &manifest(name, "1.0.0", serves, true),
784 );
785 }
786 let roster = Roster {
787 members: ["a", "b", "c", "d"]
788 .iter()
789 .map(|name| member(name, "1.0.0", *name == "a"))
790 .collect(),
791 reachability: RosterReachability::All,
792 };
793 let admitted = admit_roster(&roster, &store, None).unwrap();
794 assert_eq!(edges(&admitted), [("a", "b")]);
796 assert!(matches!(admitted.reachability(), RosterReachability::All));
797
798 for (name, serves) in [("a", false), ("b", true), ("c", false), ("d", false)] {
800 install(
801 &store,
802 name,
803 "1.0.0",
804 &manifest(name, "1.0.0", serves, false),
805 );
806 }
807 let admitted = admit_roster(&roster, &store, None).unwrap();
808 assert!(admitted.edges().is_empty());
809 }
810
811 #[test]
812 fn s4_entry_must_be_designated_exactly_once() {
813 let mut f = formation();
814 f.roster.members[0].entry = false;
815 assert_eq!(
816 refusal(&f.roster, &f.store, None),
817 RosterRefusal::EntryNotDesignated
818 );
819
820 f.roster.members[0].entry = true;
821 f.roster.members[1].entry = true;
822 assert_eq!(
823 refusal(&f.roster, &f.store, None),
824 RosterRefusal::EntryDesignatedMoreThanOnce {
825 members: vec!["planner".to_string(), "coder".to_string()]
826 }
827 );
828 }
829
830 #[test]
831 fn s5_a_duplicate_name_is_refused_before_entry_and_before_any_lookup() {
832 let empty = tempfile::tempdir().unwrap();
833 let empty_store = LocalRegistry::new(empty.path());
834 let f = formation();
835 let roster = Roster {
836 members: vec![
837 member("coder", "1.2.0", true),
838 member("coder", "1.2.0", true),
839 ],
840 reachability: RosterReachability::Closed,
841 };
842 for store in [&f.store, &empty_store] {
843 assert_eq!(
844 refusal(&roster, store, None),
845 RosterRefusal::DuplicateMember {
846 member: "coder".to_string()
847 }
848 );
849 }
850 }
851
852 #[test]
853 fn s6_a_rule_naming_no_member_is_refused_naming_it() {
854 let mut f = formation();
855 f.roster.reachability =
856 RosterReachability::Rules(vec![rule("planner", &["coder", "tester"])]);
857 assert_eq!(
858 refusal(&f.roster, &f.store, None),
859 RosterRefusal::UnknownMemberInRule {
860 from: "planner".to_string(),
861 unknown: "tester".to_string()
862 }
863 );
864 f.roster.reachability = RosterReachability::Rules(vec![rule("ghost", &["coder"])]);
865 assert_eq!(
866 refusal(&f.roster, &f.store, None),
867 RosterRefusal::UnknownMemberInRule {
868 from: "ghost".to_string(),
869 unknown: "ghost".to_string()
870 }
871 );
872 }
873
874 #[test]
875 fn a_rule_calling_the_entry_member_is_refused_naming_both() {
876 let mut f = formation();
877 f.roster.reachability = RosterReachability::Rules(vec![
878 rule("planner", &["coder", "reviewer"]),
879 rule("reviewer", &["coder", "planner"]),
880 ]);
881 assert_eq!(
882 refusal(&f.roster, &f.store, None),
883 RosterRefusal::RuleCallsEntryMember {
884 from: "reviewer".to_string(),
885 entry: "planner".to_string()
886 }
887 );
888 }
889
890 #[test]
891 fn the_first_rule_calling_the_entry_member_in_file_order_is_named() {
892 let mut f = formation();
893 f.roster.reachability = RosterReachability::Rules(vec![
894 rule("planner", &["coder"]),
895 rule("coder", &["reviewer", "planner"]),
896 rule("reviewer", &["planner"]),
897 ]);
898 assert_eq!(
899 refusal(&f.roster, &f.store, None),
900 RosterRefusal::RuleCallsEntryMember {
901 from: "coder".to_string(),
902 entry: "planner".to_string()
903 }
904 );
905 }
906
907 #[test]
908 fn a_rule_calling_the_entry_member_is_refused_before_any_lookup() {
909 let empty = tempfile::tempdir().unwrap();
910 let empty_store = LocalRegistry::new(empty.path());
911 let mut f = formation();
912 f.roster.reachability =
913 RosterReachability::Rules(vec![rule("reviewer", &["coder", "planner"])]);
914 assert_eq!(
915 refusal(&f.roster, &empty_store, None),
916 RosterRefusal::RuleCallsEntryMember {
917 from: "reviewer".to_string(),
918 entry: "planner".to_string()
919 }
920 );
921
922 install(
925 &f.store,
926 "planner",
927 "0.3.0",
928 &manifest("planner", "0.3.0", false, true),
929 );
930 assert_eq!(
931 refusal(&f.roster, &f.store, None),
932 RosterRefusal::RuleCallsEntryMember {
933 from: "reviewer".to_string(),
934 entry: "planner".to_string()
935 }
936 );
937 }
938
939 #[test]
940 fn an_unknown_rule_name_is_refused_before_a_rule_calling_the_entry_member() {
941 let mut f = formation();
942 for rules in [
943 vec![rule("ghost", &["coder"]), rule("reviewer", &["planner"])],
944 vec![rule("reviewer", &["planner"]), rule("coder", &["ghost"])],
945 ] {
946 f.roster.reachability = RosterReachability::Rules(rules);
947 assert!(matches!(
948 refusal(&f.roster, &f.store, None),
949 RosterRefusal::UnknownMemberInRule { unknown, .. } if unknown == "ghost"
950 ));
951 }
952 }
953
954 #[test]
955 fn no_admitted_edge_ever_leads_into_the_entry_member() {
956 let mut f = formation();
957 let reachabilities = [
958 RosterReachability::All,
959 f.roster.reachability.clone(),
960 RosterReachability::Rules(vec![
961 rule("reviewer", &["coder", "coder"]),
962 rule("planner", &["reviewer", "coder"]),
963 rule("planner", &["coder"]),
964 ]),
965 RosterReachability::Rules(vec![
966 rule("planner", &["coder"]),
967 rule("reviewer", &["coder"]),
968 ]),
969 ];
970 for reachability in reachabilities {
971 f.roster.reachability = reachability;
972 let admitted = admit_roster(&f.roster, &f.store, None).unwrap();
973 let entry = admitted.entry().name.clone();
974 assert!(admitted.declares_peer_traffic());
975 assert!(
976 admitted.edges().iter().all(|edge| edge.to != entry),
977 "{:?}",
978 admitted.edges()
979 );
980 for member in admitted.members() {
981 assert!(admitted.callees(&member.name).all(|callee| !callee.entry));
982 }
983 }
984
985 f.roster.reachability = RosterReachability::All;
987 for entry in 0..f.roster.members.len() {
988 for (index, member) in f.roster.members.iter_mut().enumerate() {
989 member.entry = index == entry;
990 }
991 let admitted = admit_roster(&f.roster, &f.store, None).unwrap();
992 let entry = admitted.entry().name.as_str();
993 let pairs = edges(&admitted);
994 assert_eq!(pairs.len(), 4, "{pairs:?}");
995 assert!(pairs.iter().all(|(_, to)| *to != entry), "{pairs:?}");
996 assert_eq!(pairs.iter().filter(|(from, _)| *from == entry).count(), 2);
997 }
998 }
999
1000 #[test]
1001 fn s7_a_callee_must_serve_peers_but_a_caller_need_not() {
1002 for reviewer_manifest in [
1003 "name: reviewer\nversion: 0.9.0\nartifacts: []\nnetwork:\n authentication:\n scheme: bearer\n"
1004 .to_string(),
1005 format!("name: reviewer\nversion: 0.9.0\nartifacts: []\n{REFUSES}{AUTHENTICATED}"),
1006 ] {
1007 let mut f = formation();
1008 install(&f.store, "reviewer", "0.9.0", &reviewer_manifest);
1009 assert_eq!(
1010 refusal(&f.roster, &f.store, None),
1011 RosterRefusal::CalleeDoesNotServePeers {
1012 member: "reviewer".to_string()
1013 }
1014 );
1015 f.roster.reachability = RosterReachability::Rules(vec![
1016 rule("planner", &["coder"]),
1017 rule("reviewer", &["coder"]),
1018 ]);
1019 admit_roster(&f.roster, &f.store, None).unwrap();
1020 }
1021 }
1022
1023 #[test]
1024 fn s8_with_traffic_every_member_needs_authentication_even_off_every_edge() {
1025 let f = formation();
1026 install(
1027 &f.store,
1028 "coder",
1029 "1.2.0",
1030 &manifest("coder", "1.2.0", true, false),
1031 );
1032 assert_eq!(
1033 refusal(&f.roster, &f.store, None),
1034 RosterRefusal::MemberNotAuthenticated {
1035 member: "coder".to_string()
1036 }
1037 );
1038
1039 let mut f = formation();
1040 install(
1041 &f.store,
1042 "notes",
1043 "1.0.0",
1044 &manifest("notes", "1.0.0", false, false),
1045 );
1046 f.roster.members.push(member("notes", "1.0.0", false));
1047 assert_eq!(
1048 refusal(&f.roster, &f.store, None),
1049 RosterRefusal::MemberNotAuthenticated {
1050 member: "notes".to_string()
1051 }
1052 );
1053 }
1054
1055 #[test]
1056 fn s9_an_uninstalled_or_unreadable_member_is_refused_naming_its_coordinate() {
1057 let mut f = formation();
1058 f.roster.members[1].version = "9.9.9".to_string();
1059 match refusal(&f.roster, &f.store, None) {
1060 RosterRefusal::MemberUnresolvable {
1061 member, coordinate, ..
1062 } => {
1063 assert_eq!(member, "coder");
1064 assert_eq!(coordinate, "local/coder@9.9.9");
1065 }
1066 other => panic!("unexpected refusal {other:?}"),
1067 }
1068
1069 let f = formation();
1070 install(&f.store, "coder", "1.2.0", "name: [unclosed\n");
1071 match refusal(&f.roster, &f.store, None) {
1072 RosterRefusal::MemberUnresolvable {
1073 member, coordinate, ..
1074 } => {
1075 assert_eq!(member, "coder");
1076 assert_eq!(coordinate, "local/coder@1.2.0");
1077 }
1078 other => panic!("unexpected refusal {other:?}"),
1079 }
1080 }
1081
1082 #[test]
1083 fn bytes_that_do_not_match_the_store_hash_are_refused() {
1084 let f = formation();
1085 let artifact = f
1086 .store
1087 .root()
1088 .join("local")
1089 .join("coder")
1090 .join("1.2.0")
1091 .join("coder-1.2.0.mur.zip");
1092 fs::write(&artifact, pack(&manifest("coder", "1.2.0", true, false))).unwrap();
1093 assert!(matches!(
1094 refusal(&f.roster, &f.store, None),
1095 RosterRefusal::MemberUnresolvable { member, .. } if member == "coder"
1096 ));
1097 }
1098
1099 #[test]
1100 fn s10_the_lock_verifies_a_member_it_pins() {
1101 let f = formation();
1102 let agreeing = lock_pinning("coder", "1.2.0", &f.hashes["coder"]);
1103 admit_roster(&f.roster, &f.store, Some(&agreeing)).unwrap();
1104
1105 for lock in [
1106 lock_pinning("coder", "1.2.0", &"0".repeat(64)),
1107 lock_pinning("coder", "1.1.0", &f.hashes["coder"]),
1108 ] {
1109 match refusal(&f.roster, &f.store, Some(&lock)) {
1110 RosterRefusal::MemberLockConflict {
1111 member,
1112 capsule,
1113 conflict,
1114 } => {
1115 assert_eq!(member, "coder");
1116 assert_eq!(capsule, "local/coder");
1117 assert!(!conflict.is_empty());
1118 }
1119 other => panic!("unexpected refusal {other:?}"),
1120 }
1121 }
1122 }
1123
1124 #[test]
1125 fn s13_a_duplicate_and_an_unknown_rule_name_is_always_the_duplicate() {
1126 let f = formation();
1127 let roster = Roster {
1128 members: vec![
1129 member("coder", "1.2.0", true),
1130 member("coder", "1.2.0", false),
1131 ],
1132 reachability: RosterReachability::Rules(vec![rule("coder", &["ghost"])]),
1133 };
1134 for _ in 0..5 {
1135 assert!(matches!(
1136 refusal(&roster, &f.store, None),
1137 RosterRefusal::DuplicateMember { .. }
1138 ));
1139 }
1140 }
1141
1142 #[test]
1143 fn admit_roster_file_maps_every_load_error_to_malformed_and_records_the_source() {
1144 let f = formation();
1145 let project = tempfile::tempdir().unwrap();
1146 assert!(matches!(
1147 admit_roster_file(project.path(), &f.store, None),
1148 Err(RosterRefusal::Malformed { message, .. }) if message.contains("roster.yaml not found")
1149 ));
1150
1151 fs::write(
1152 resolve_roster_path(project.path()),
1153 "roster_version: 2\nmembers: []\n",
1154 )
1155 .unwrap();
1156 assert!(matches!(
1157 admit_roster_file(project.path(), &f.store, None),
1158 Err(RosterRefusal::Malformed { message, .. }) if message.contains("roster_version")
1159 ));
1160
1161 fs::write(
1162 resolve_roster_path(project.path()),
1163 "roster_version: 1\nmembers:\n - {name: planner, capsule: local/planner, version: 0.3.0, entry: true}\n",
1164 )
1165 .unwrap();
1166 let admitted = admit_roster_file(project.path(), &f.store, None).unwrap();
1167 assert_eq!(
1168 admitted.source(),
1169 Some(resolve_roster_path(project.path()).as_path())
1170 );
1171 }
1172
1173 #[test]
1174 fn every_refusal_names_roster_yaml() {
1175 for refusal in [
1176 RosterRefusal::DuplicateMember { member: "a".into() },
1177 RosterRefusal::EntryNotDesignated,
1178 RosterRefusal::EntryDesignatedMoreThanOnce {
1179 members: vec!["a".into(), "b".into()],
1180 },
1181 RosterRefusal::UnknownMemberInRule {
1182 from: "a".into(),
1183 unknown: "b".into(),
1184 },
1185 RosterRefusal::RuleCallsEntryMember {
1186 from: "a".into(),
1187 entry: "b".into(),
1188 },
1189 RosterRefusal::MemberUnresolvable {
1190 member: "a".into(),
1191 coordinate: "a@1".into(),
1192 reason: "r".into(),
1193 },
1194 RosterRefusal::MemberLockConflict {
1195 member: "a".into(),
1196 capsule: "a".into(),
1197 conflict: "c".into(),
1198 },
1199 RosterRefusal::CalleeDoesNotServePeers { member: "a".into() },
1200 RosterRefusal::MemberNotAuthenticated { member: "a".into() },
1201 ] {
1202 assert!(refusal.to_string().starts_with("roster.yaml"), "{refusal}");
1203 }
1204 }
1205
1206 const TEN_CALLERS_WARNING: &str =
1207 "warning[W-ROS-001]: roster.yaml lets 10 members call 'reviewer' \
1208 (w01, w02, w03, w04, w05, w06, w07, w08, w09, w10), but it holds 2 tasks at once — one \
1209 running and lifecycle.queue_depth: 1 waiting — so a call that arrives while it is full \
1210 waits and is offered again, and is rejected only if it stays full until the call's \
1211 deadline; lifecycle.queue_depth: 9 would hold them all \
1212 (https://docs.murmur.nexus/reference/diagnostics/#w-ros-001)";
1213
1214 fn manifest_with_lifecycle(name: &str, lifecycle_yaml: &str) -> String {
1217 let mut yaml = manifest(name, "1.0.0", true, true);
1218 if !lifecycle_yaml.is_empty() {
1219 yaml.push_str(&format!("lifecycle:\n {lifecycle_yaml}\n"));
1220 }
1221 yaml
1222 }
1223
1224 const QUEUE_DEPTH_1: &str = "task_acceptance: queue\n after_task: sleep";
1225
1226 fn queue_at(depth: usize) -> String {
1227 format!("task_acceptance: queue\n after_task: sleep\n queue_depth: {depth}")
1228 }
1229
1230 fn admit_with(
1232 members: &[(&str, &str)],
1233 reachability: RosterReachability,
1234 ) -> (tempfile::TempDir, AdmittedRoster) {
1235 let dir = tempfile::tempdir().unwrap();
1236 let store = LocalRegistry::new(dir.path());
1237 for (name, lifecycle) in members {
1238 install(
1239 &store,
1240 name,
1241 "1.0.0",
1242 &manifest_with_lifecycle(name, lifecycle),
1243 );
1244 }
1245 let roster = Roster {
1246 members: members
1247 .iter()
1248 .enumerate()
1249 .map(|(index, (name, _))| member(name, "1.0.0", index == 0))
1250 .collect(),
1251 reachability,
1252 };
1253 let admitted = admit_roster(&roster, &store, None).unwrap();
1254 (dir, admitted)
1255 }
1256
1257 const WORKERS: [&str; 10] = [
1258 "w01", "w02", "w03", "w04", "w05", "w06", "w07", "w08", "w09", "w10",
1259 ];
1260
1261 fn ten_workers_one_reviewer(reviewer_lifecycle: &str) -> (tempfile::TempDir, AdmittedRoster) {
1263 let mut members = vec![("lead", "")];
1264 members.extend(WORKERS.iter().map(|worker| (*worker, QUEUE_DEPTH_1)));
1265 members.push(("reviewer", reviewer_lifecycle));
1266 let mut rules = vec![rule("lead", &WORKERS)];
1267 rules.extend(WORKERS.iter().map(|worker| rule(worker, &["reviewer"])));
1268 admit_with(&members, RosterReachability::Rules(rules))
1269 }
1270
1271 #[test]
1272 fn ten_callers_overflow_a_reviewer_at_depth_one() {
1273 let (_dir, admitted) = ten_workers_one_reviewer(QUEUE_DEPTH_1);
1274 assert_eq!(
1275 admitted.caller_overflows(),
1276 [CallerOverflow {
1277 member: "reviewer".to_string(),
1278 capsule: "local/reviewer".to_string(),
1279 version: "1.0.0".to_string(),
1280 callers: WORKERS.iter().map(ToString::to_string).collect(),
1281 task_acceptance: TaskAcceptance::Queue,
1282 queue_depth: 1,
1283 holds: 2,
1284 sufficient_queue_depth: 9,
1285 }]
1286 );
1287 }
1288
1289 #[test]
1290 fn the_ten_callers_warning_line_is_exact() {
1291 let (_dir, admitted) = ten_workers_one_reviewer(QUEUE_DEPTH_1);
1292 let overflows = admitted.caller_overflows();
1293 assert_eq!(caller_overflow_warning(&overflows[0]), TEN_CALLERS_WARNING);
1294 assert_eq!(
1295 caller_overflow_fix(&overflows[0]),
1296 "reviewer (local/reviewer@1.0.0): set lifecycle.queue_depth: 9 in its murmur.yaml so \
1297 all 10 members that may call it are held at once, or narrow its callers in \
1298 roster.yaml (warning[W-ROS-001])"
1299 );
1300 }
1301
1302 #[test]
1303 fn a_reviewer_deep_enough_for_every_caller_does_not_overflow() {
1304 for depth in [9, 10] {
1305 let (_dir, admitted) = ten_workers_one_reviewer(&queue_at(depth));
1306 assert!(
1307 admitted.caller_overflows().is_empty(),
1308 "depth {depth} holds all ten callers"
1309 );
1310 }
1311 }
1312
1313 #[test]
1314 fn all_counts_every_expanded_caller_and_never_the_entry_member() {
1315 for (lifecycle, overflows) in [
1316 (String::new(), true),
1317 (QUEUE_DEPTH_1.to_string(), true),
1318 (queue_at(2), false),
1319 ] {
1320 let members: Vec<(&str, &str)> = ["a", "b", "c", "d"]
1321 .iter()
1322 .map(|name| (*name, lifecycle.as_str()))
1323 .collect();
1324 let (_dir, admitted) = admit_with(&members, RosterReachability::All);
1325 let found = admitted.caller_overflows();
1326 if !overflows {
1327 assert!(found.is_empty(), "'{lifecycle}' holds three callers");
1328 continue;
1329 }
1330 let named: Vec<_> = found.iter().map(|o| o.member.as_str()).collect();
1331 assert_eq!(named, ["b", "c", "d"], "'{lifecycle}'");
1332 for overflow in &found {
1333 assert_eq!(overflow.callers.len(), 3, "{overflow:?}");
1334 assert!(!overflow.callers.contains(&overflow.member));
1335 assert_eq!(overflow.sufficient_queue_depth, 2);
1336 }
1337 assert_eq!(found[0].callers, ["a", "c", "d"]);
1338 }
1339 }
1340
1341 #[test]
1342 fn a_single_member_with_two_callers_is_told_to_switch_to_queue() {
1343 let (_dir, admitted) = admit_with(
1344 &[("lead", ""), ("helper", ""), ("coder", "")],
1345 RosterReachability::Rules(vec![
1346 rule("lead", &["coder", "helper"]),
1347 rule("helper", &["coder"]),
1348 ]),
1349 );
1350 let overflows = admitted.caller_overflows();
1351 assert_eq!(overflows.len(), 1);
1352 let overflow = &overflows[0];
1353 assert_eq!(overflow.member, "coder");
1354 assert_eq!(overflow.callers, ["lead", "helper"]);
1355 assert_eq!(overflow.task_acceptance, TaskAcceptance::Single);
1356 assert_eq!(overflow.holds, 1);
1357 assert_eq!(overflow.sufficient_queue_depth, 1);
1358 assert_eq!(
1359 caller_overflow_warning(overflow),
1360 "warning[W-ROS-001]: roster.yaml lets 2 members call 'coder' (lead, helper), but it \
1361 holds 1 task at once — lifecycle.task_acceptance: single — so a call that arrives \
1362 while it is busy waits and is offered again, and is rejected only if it stays busy \
1363 until the call's deadline; lifecycle.task_acceptance: queue with \
1364 lifecycle.queue_depth: 1 would hold them all \
1365 (https://docs.murmur.nexus/reference/diagnostics/#w-ros-001)"
1366 );
1367 assert_eq!(
1368 caller_overflow_fix(overflow),
1369 "coder (local/coder@1.0.0): set lifecycle.task_acceptance: queue and \
1370 lifecycle.queue_depth: 1 in its murmur.yaml so all 2 members that may call it are \
1371 held at once, or narrow its callers in roster.yaml (warning[W-ROS-001])"
1372 );
1373 }
1374
1375 #[test]
1376 fn a_member_that_accepts_no_task_overflows_with_one_caller() {
1377 let (_dir, admitted) = admit_with(
1378 &[("lead", ""), ("idle", "task_acceptance: none")],
1379 RosterReachability::Rules(vec![rule("lead", &["idle"])]),
1380 );
1381 let overflows = admitted.caller_overflows();
1382 assert_eq!(overflows.len(), 1);
1383 assert_eq!(overflows[0].holds, 0);
1384 assert_eq!(
1385 caller_overflow_warning(&overflows[0]),
1386 "warning[W-ROS-001]: roster.yaml lets 1 member call 'idle' (lead), but it holds no \
1387 task — lifecycle.task_acceptance: none — so no call is ever taken: each is offered \
1388 again until its deadline and then rejected; \
1389 lifecycle.task_acceptance: queue with lifecycle.queue_depth: 1 would hold them all \
1390 (https://docs.murmur.nexus/reference/diagnostics/#w-ros-001)"
1391 );
1392 assert_eq!(
1393 caller_overflow_fix(&overflows[0]),
1394 "idle (local/idle@1.0.0): set lifecycle.task_acceptance: queue and \
1395 lifecycle.queue_depth: 1 in its murmur.yaml so the 1 member that may call it is held \
1396 at once, or narrow its callers in roster.yaml (warning[W-ROS-001])"
1397 );
1398 }
1399
1400 #[test]
1403 fn a_queue_member_at_depth_zero_is_refused_at_admission() {
1404 let dir = tempfile::tempdir().unwrap();
1405 let store = LocalRegistry::new(dir.path());
1406 for (name, lifecycle) in [("lead", String::new()), ("shut", queue_at(0))] {
1407 install(
1408 &store,
1409 name,
1410 "1.0.0",
1411 &manifest_with_lifecycle(name, &lifecycle),
1412 );
1413 }
1414 let roster = Roster {
1415 members: vec![
1416 member("lead", "1.0.0", true),
1417 member("shut", "1.0.0", false),
1418 ],
1419 reachability: RosterReachability::Rules(vec![rule("lead", &["shut"])]),
1420 };
1421 match admit_roster(&roster, &store, None) {
1422 Err(RosterRefusal::MemberUnresolvable { member, reason, .. }) => {
1423 assert_eq!(member, "shut");
1424 assert!(reason.contains("lifecycle.queue_depth"), "{reason}");
1425 }
1426 other => panic!("expected MemberUnresolvable for shut, got {other:?}"),
1427 }
1428 }
1429
1430 #[test]
1433 fn the_entry_member_is_never_named() {
1434 let (_dir, admitted) = admit_with(
1435 &[
1436 ("lead", "task_acceptance: none"),
1437 ("a", ""),
1438 ("b", &queue_at(5)),
1439 ],
1440 RosterReachability::All,
1441 );
1442 assert!(!admitted.edges().is_empty());
1443 assert!(admitted
1444 .caller_overflows()
1445 .iter()
1446 .all(|overflow| overflow.member != "lead"));
1447 }
1448
1449 #[test]
1450 fn a_roster_with_no_edges_has_no_overflow() {
1451 let (_dir, admitted) = admit_with(
1452 &[
1453 ("lead", "task_acceptance: none"),
1454 ("idle", "task_acceptance: none"),
1455 ],
1456 RosterReachability::Closed,
1457 );
1458 assert!(admitted.edges().is_empty());
1459 assert!(admitted.caller_overflows().is_empty());
1460 }
1461}