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 signer: None,
653 }],
654 }
655 }
656
657 struct Formation {
659 _dir: tempfile::TempDir,
660 store: LocalRegistry,
661 hashes: BTreeMap<&'static str, String>,
662 roster: Roster,
663 }
664
665 fn formation() -> Formation {
666 let dir = tempfile::tempdir().unwrap();
667 let store = LocalRegistry::new(dir.path().join("artifacts"));
668 let mut hashes = BTreeMap::new();
669 for (name, version) in [
670 ("planner", "0.3.0"),
671 ("coder", "1.2.0"),
672 ("reviewer", "0.9.0"),
673 ] {
674 hashes.insert(
675 name,
676 install(&store, name, version, &manifest(name, version, true, true)),
677 );
678 }
679 let roster = Roster {
680 members: vec![
681 member("planner", "0.3.0", true),
682 member("coder", "1.2.0", false),
683 member("reviewer", "0.9.0", false),
684 ],
685 reachability: RosterReachability::Rules(vec![
686 rule("planner", &["coder", "reviewer"]),
687 rule("reviewer", &["coder"]),
688 ]),
689 };
690 Formation {
691 _dir: dir,
692 store,
693 hashes,
694 roster,
695 }
696 }
697
698 fn refusal(roster: &Roster, store: &LocalRegistry, lock: Option<&MurmurLock>) -> RosterRefusal {
699 admit_roster(roster, store, lock).expect_err("the roster should be refused")
700 }
701
702 #[test]
703 fn s1_admits_members_in_order_bound_to_their_hashes_with_explicit_edges() {
704 let f = formation();
705 let admitted = admit_roster(&f.roster, &f.store, None).unwrap();
706 let names: Vec<_> = admitted.members().iter().map(|m| m.name.as_str()).collect();
707 assert_eq!(names, ["planner", "coder", "reviewer"]);
708 for member in admitted.members() {
709 assert_eq!(member.sha256, f.hashes[member.name.as_str()]);
710 assert_eq!(
711 member.manifest.name,
712 murmur_artifact::tool_name_of(&member.capsule)
713 );
714 assert!(member.manifest.accepts_peer_tasks());
715 }
716 assert_eq!(admitted.entry().name, "planner");
717 assert_eq!(
718 edges(&admitted),
719 [
720 ("planner", "coder"),
721 ("planner", "reviewer"),
722 ("reviewer", "coder")
723 ]
724 );
725 assert!(admitted.may_call("planner", "coder"));
726 assert!(!admitted.may_call("coder", "planner"));
727 assert!(admitted.declares_peer_traffic());
728 let callees: Vec<_> = admitted
729 .callees("planner")
730 .map(|m| m.name.as_str())
731 .collect();
732 assert_eq!(callees, ["coder", "reviewer"]);
733 assert_eq!(admitted.source(), None);
734 }
735
736 #[test]
737 fn repeated_edges_collapse_and_edges_sort_by_roster_index() {
738 let mut f = formation();
739 f.roster.reachability = RosterReachability::Rules(vec![
740 rule("reviewer", &["coder", "coder"]),
741 rule("planner", &["reviewer", "coder"]),
742 rule("planner", &["coder"]),
743 ]);
744 let admitted = admit_roster(&f.roster, &f.store, None).unwrap();
745 assert_eq!(
746 edges(&admitted),
747 [
748 ("planner", "coder"),
749 ("planner", "reviewer"),
750 ("reviewer", "coder")
751 ]
752 );
753 }
754
755 #[test]
756 fn s2_a_single_public_member_declares_no_traffic() {
757 let dir = tempfile::tempdir().unwrap();
758 let store = LocalRegistry::new(dir.path());
759 install(
760 &store,
761 "solo",
762 "1.0.0",
763 &manifest("solo", "1.0.0", false, false),
764 );
765 let roster = Roster {
766 members: vec![member("solo", "1.0.0", true)],
767 reachability: RosterReachability::Closed,
768 };
769 let admitted = admit_roster(&roster, &store, None).unwrap();
770 assert!(admitted.edges().is_empty());
771 assert!(!admitted.declares_peer_traffic());
772 assert_eq!(admitted.entry().name, "solo");
773 }
774
775 #[test]
776 fn s3_all_pairs_only_members_that_both_serve_peers() {
777 let dir = tempfile::tempdir().unwrap();
778 let store = LocalRegistry::new(dir.path());
779 for (name, serves) in [("a", true), ("b", true), ("c", false), ("d", false)] {
780 install(
781 &store,
782 name,
783 "1.0.0",
784 &manifest(name, "1.0.0", serves, true),
785 );
786 }
787 let roster = Roster {
788 members: ["a", "b", "c", "d"]
789 .iter()
790 .map(|name| member(name, "1.0.0", *name == "a"))
791 .collect(),
792 reachability: RosterReachability::All,
793 };
794 let admitted = admit_roster(&roster, &store, None).unwrap();
795 assert_eq!(edges(&admitted), [("a", "b")]);
797 assert!(matches!(admitted.reachability(), RosterReachability::All));
798
799 for (name, serves) in [("a", false), ("b", true), ("c", false), ("d", false)] {
801 install(
802 &store,
803 name,
804 "1.0.0",
805 &manifest(name, "1.0.0", serves, false),
806 );
807 }
808 let admitted = admit_roster(&roster, &store, None).unwrap();
809 assert!(admitted.edges().is_empty());
810 }
811
812 #[test]
813 fn s4_entry_must_be_designated_exactly_once() {
814 let mut f = formation();
815 f.roster.members[0].entry = false;
816 assert_eq!(
817 refusal(&f.roster, &f.store, None),
818 RosterRefusal::EntryNotDesignated
819 );
820
821 f.roster.members[0].entry = true;
822 f.roster.members[1].entry = true;
823 assert_eq!(
824 refusal(&f.roster, &f.store, None),
825 RosterRefusal::EntryDesignatedMoreThanOnce {
826 members: vec!["planner".to_string(), "coder".to_string()]
827 }
828 );
829 }
830
831 #[test]
832 fn s5_a_duplicate_name_is_refused_before_entry_and_before_any_lookup() {
833 let empty = tempfile::tempdir().unwrap();
834 let empty_store = LocalRegistry::new(empty.path());
835 let f = formation();
836 let roster = Roster {
837 members: vec![
838 member("coder", "1.2.0", true),
839 member("coder", "1.2.0", true),
840 ],
841 reachability: RosterReachability::Closed,
842 };
843 for store in [&f.store, &empty_store] {
844 assert_eq!(
845 refusal(&roster, store, None),
846 RosterRefusal::DuplicateMember {
847 member: "coder".to_string()
848 }
849 );
850 }
851 }
852
853 #[test]
854 fn s6_a_rule_naming_no_member_is_refused_naming_it() {
855 let mut f = formation();
856 f.roster.reachability =
857 RosterReachability::Rules(vec![rule("planner", &["coder", "tester"])]);
858 assert_eq!(
859 refusal(&f.roster, &f.store, None),
860 RosterRefusal::UnknownMemberInRule {
861 from: "planner".to_string(),
862 unknown: "tester".to_string()
863 }
864 );
865 f.roster.reachability = RosterReachability::Rules(vec![rule("ghost", &["coder"])]);
866 assert_eq!(
867 refusal(&f.roster, &f.store, None),
868 RosterRefusal::UnknownMemberInRule {
869 from: "ghost".to_string(),
870 unknown: "ghost".to_string()
871 }
872 );
873 }
874
875 #[test]
876 fn a_rule_calling_the_entry_member_is_refused_naming_both() {
877 let mut f = formation();
878 f.roster.reachability = RosterReachability::Rules(vec![
879 rule("planner", &["coder", "reviewer"]),
880 rule("reviewer", &["coder", "planner"]),
881 ]);
882 assert_eq!(
883 refusal(&f.roster, &f.store, None),
884 RosterRefusal::RuleCallsEntryMember {
885 from: "reviewer".to_string(),
886 entry: "planner".to_string()
887 }
888 );
889 }
890
891 #[test]
892 fn the_first_rule_calling_the_entry_member_in_file_order_is_named() {
893 let mut f = formation();
894 f.roster.reachability = RosterReachability::Rules(vec![
895 rule("planner", &["coder"]),
896 rule("coder", &["reviewer", "planner"]),
897 rule("reviewer", &["planner"]),
898 ]);
899 assert_eq!(
900 refusal(&f.roster, &f.store, None),
901 RosterRefusal::RuleCallsEntryMember {
902 from: "coder".to_string(),
903 entry: "planner".to_string()
904 }
905 );
906 }
907
908 #[test]
909 fn a_rule_calling_the_entry_member_is_refused_before_any_lookup() {
910 let empty = tempfile::tempdir().unwrap();
911 let empty_store = LocalRegistry::new(empty.path());
912 let mut f = formation();
913 f.roster.reachability =
914 RosterReachability::Rules(vec![rule("reviewer", &["coder", "planner"])]);
915 assert_eq!(
916 refusal(&f.roster, &empty_store, None),
917 RosterRefusal::RuleCallsEntryMember {
918 from: "reviewer".to_string(),
919 entry: "planner".to_string()
920 }
921 );
922
923 install(
926 &f.store,
927 "planner",
928 "0.3.0",
929 &manifest("planner", "0.3.0", false, true),
930 );
931 assert_eq!(
932 refusal(&f.roster, &f.store, None),
933 RosterRefusal::RuleCallsEntryMember {
934 from: "reviewer".to_string(),
935 entry: "planner".to_string()
936 }
937 );
938 }
939
940 #[test]
941 fn an_unknown_rule_name_is_refused_before_a_rule_calling_the_entry_member() {
942 let mut f = formation();
943 for rules in [
944 vec![rule("ghost", &["coder"]), rule("reviewer", &["planner"])],
945 vec![rule("reviewer", &["planner"]), rule("coder", &["ghost"])],
946 ] {
947 f.roster.reachability = RosterReachability::Rules(rules);
948 assert!(matches!(
949 refusal(&f.roster, &f.store, None),
950 RosterRefusal::UnknownMemberInRule { unknown, .. } if unknown == "ghost"
951 ));
952 }
953 }
954
955 #[test]
956 fn no_admitted_edge_ever_leads_into_the_entry_member() {
957 let mut f = formation();
958 let reachabilities = [
959 RosterReachability::All,
960 f.roster.reachability.clone(),
961 RosterReachability::Rules(vec![
962 rule("reviewer", &["coder", "coder"]),
963 rule("planner", &["reviewer", "coder"]),
964 rule("planner", &["coder"]),
965 ]),
966 RosterReachability::Rules(vec![
967 rule("planner", &["coder"]),
968 rule("reviewer", &["coder"]),
969 ]),
970 ];
971 for reachability in reachabilities {
972 f.roster.reachability = reachability;
973 let admitted = admit_roster(&f.roster, &f.store, None).unwrap();
974 let entry = admitted.entry().name.clone();
975 assert!(admitted.declares_peer_traffic());
976 assert!(
977 admitted.edges().iter().all(|edge| edge.to != entry),
978 "{:?}",
979 admitted.edges()
980 );
981 for member in admitted.members() {
982 assert!(admitted.callees(&member.name).all(|callee| !callee.entry));
983 }
984 }
985
986 f.roster.reachability = RosterReachability::All;
988 for entry in 0..f.roster.members.len() {
989 for (index, member) in f.roster.members.iter_mut().enumerate() {
990 member.entry = index == entry;
991 }
992 let admitted = admit_roster(&f.roster, &f.store, None).unwrap();
993 let entry = admitted.entry().name.as_str();
994 let pairs = edges(&admitted);
995 assert_eq!(pairs.len(), 4, "{pairs:?}");
996 assert!(pairs.iter().all(|(_, to)| *to != entry), "{pairs:?}");
997 assert_eq!(pairs.iter().filter(|(from, _)| *from == entry).count(), 2);
998 }
999 }
1000
1001 #[test]
1002 fn s7_a_callee_must_serve_peers_but_a_caller_need_not() {
1003 for reviewer_manifest in [
1004 "name: reviewer\nversion: 0.9.0\nartifacts: []\nnetwork:\n authentication:\n scheme: bearer\n"
1005 .to_string(),
1006 format!("name: reviewer\nversion: 0.9.0\nartifacts: []\n{REFUSES}{AUTHENTICATED}"),
1007 ] {
1008 let mut f = formation();
1009 install(&f.store, "reviewer", "0.9.0", &reviewer_manifest);
1010 assert_eq!(
1011 refusal(&f.roster, &f.store, None),
1012 RosterRefusal::CalleeDoesNotServePeers {
1013 member: "reviewer".to_string()
1014 }
1015 );
1016 f.roster.reachability = RosterReachability::Rules(vec![
1017 rule("planner", &["coder"]),
1018 rule("reviewer", &["coder"]),
1019 ]);
1020 admit_roster(&f.roster, &f.store, None).unwrap();
1021 }
1022 }
1023
1024 #[test]
1025 fn s8_with_traffic_every_member_needs_authentication_even_off_every_edge() {
1026 let f = formation();
1027 install(
1028 &f.store,
1029 "coder",
1030 "1.2.0",
1031 &manifest("coder", "1.2.0", true, false),
1032 );
1033 assert_eq!(
1034 refusal(&f.roster, &f.store, None),
1035 RosterRefusal::MemberNotAuthenticated {
1036 member: "coder".to_string()
1037 }
1038 );
1039
1040 let mut f = formation();
1041 install(
1042 &f.store,
1043 "notes",
1044 "1.0.0",
1045 &manifest("notes", "1.0.0", false, false),
1046 );
1047 f.roster.members.push(member("notes", "1.0.0", false));
1048 assert_eq!(
1049 refusal(&f.roster, &f.store, None),
1050 RosterRefusal::MemberNotAuthenticated {
1051 member: "notes".to_string()
1052 }
1053 );
1054 }
1055
1056 #[test]
1057 fn s9_an_uninstalled_or_unreadable_member_is_refused_naming_its_coordinate() {
1058 let mut f = formation();
1059 f.roster.members[1].version = "9.9.9".to_string();
1060 match refusal(&f.roster, &f.store, None) {
1061 RosterRefusal::MemberUnresolvable {
1062 member, coordinate, ..
1063 } => {
1064 assert_eq!(member, "coder");
1065 assert_eq!(coordinate, "local/coder@9.9.9");
1066 }
1067 other => panic!("unexpected refusal {other:?}"),
1068 }
1069
1070 let f = formation();
1071 install(&f.store, "coder", "1.2.0", "name: [unclosed\n");
1072 match refusal(&f.roster, &f.store, None) {
1073 RosterRefusal::MemberUnresolvable {
1074 member, coordinate, ..
1075 } => {
1076 assert_eq!(member, "coder");
1077 assert_eq!(coordinate, "local/coder@1.2.0");
1078 }
1079 other => panic!("unexpected refusal {other:?}"),
1080 }
1081 }
1082
1083 #[test]
1084 fn bytes_that_do_not_match_the_store_hash_are_refused() {
1085 let f = formation();
1086 let artifact = f
1087 .store
1088 .root()
1089 .join("local")
1090 .join("coder")
1091 .join("1.2.0")
1092 .join("coder-1.2.0.mur.zip");
1093 fs::write(&artifact, pack(&manifest("coder", "1.2.0", true, false))).unwrap();
1094 assert!(matches!(
1095 refusal(&f.roster, &f.store, None),
1096 RosterRefusal::MemberUnresolvable { member, .. } if member == "coder"
1097 ));
1098 }
1099
1100 #[test]
1101 fn s10_the_lock_verifies_a_member_it_pins() {
1102 let f = formation();
1103 let agreeing = lock_pinning("coder", "1.2.0", &f.hashes["coder"]);
1104 admit_roster(&f.roster, &f.store, Some(&agreeing)).unwrap();
1105
1106 for lock in [
1107 lock_pinning("coder", "1.2.0", &"0".repeat(64)),
1108 lock_pinning("coder", "1.1.0", &f.hashes["coder"]),
1109 ] {
1110 match refusal(&f.roster, &f.store, Some(&lock)) {
1111 RosterRefusal::MemberLockConflict {
1112 member,
1113 capsule,
1114 conflict,
1115 } => {
1116 assert_eq!(member, "coder");
1117 assert_eq!(capsule, "local/coder");
1118 assert!(!conflict.is_empty());
1119 }
1120 other => panic!("unexpected refusal {other:?}"),
1121 }
1122 }
1123 }
1124
1125 #[test]
1126 fn s13_a_duplicate_and_an_unknown_rule_name_is_always_the_duplicate() {
1127 let f = formation();
1128 let roster = Roster {
1129 members: vec![
1130 member("coder", "1.2.0", true),
1131 member("coder", "1.2.0", false),
1132 ],
1133 reachability: RosterReachability::Rules(vec![rule("coder", &["ghost"])]),
1134 };
1135 for _ in 0..5 {
1136 assert!(matches!(
1137 refusal(&roster, &f.store, None),
1138 RosterRefusal::DuplicateMember { .. }
1139 ));
1140 }
1141 }
1142
1143 #[test]
1144 fn admit_roster_file_maps_every_load_error_to_malformed_and_records_the_source() {
1145 let f = formation();
1146 let project = tempfile::tempdir().unwrap();
1147 assert!(matches!(
1148 admit_roster_file(project.path(), &f.store, None),
1149 Err(RosterRefusal::Malformed { message, .. }) if message.contains("roster.yaml not found")
1150 ));
1151
1152 fs::write(
1153 resolve_roster_path(project.path()),
1154 "roster_version: 2\nmembers: []\n",
1155 )
1156 .unwrap();
1157 assert!(matches!(
1158 admit_roster_file(project.path(), &f.store, None),
1159 Err(RosterRefusal::Malformed { message, .. }) if message.contains("roster_version")
1160 ));
1161
1162 fs::write(
1163 resolve_roster_path(project.path()),
1164 "roster_version: 1\nmembers:\n - {name: planner, capsule: local/planner, version: 0.3.0, entry: true}\n",
1165 )
1166 .unwrap();
1167 let admitted = admit_roster_file(project.path(), &f.store, None).unwrap();
1168 assert_eq!(
1169 admitted.source(),
1170 Some(resolve_roster_path(project.path()).as_path())
1171 );
1172 }
1173
1174 #[test]
1175 fn every_refusal_names_roster_yaml() {
1176 for refusal in [
1177 RosterRefusal::DuplicateMember { member: "a".into() },
1178 RosterRefusal::EntryNotDesignated,
1179 RosterRefusal::EntryDesignatedMoreThanOnce {
1180 members: vec!["a".into(), "b".into()],
1181 },
1182 RosterRefusal::UnknownMemberInRule {
1183 from: "a".into(),
1184 unknown: "b".into(),
1185 },
1186 RosterRefusal::RuleCallsEntryMember {
1187 from: "a".into(),
1188 entry: "b".into(),
1189 },
1190 RosterRefusal::MemberUnresolvable {
1191 member: "a".into(),
1192 coordinate: "a@1".into(),
1193 reason: "r".into(),
1194 },
1195 RosterRefusal::MemberLockConflict {
1196 member: "a".into(),
1197 capsule: "a".into(),
1198 conflict: "c".into(),
1199 },
1200 RosterRefusal::CalleeDoesNotServePeers { member: "a".into() },
1201 RosterRefusal::MemberNotAuthenticated { member: "a".into() },
1202 ] {
1203 assert!(refusal.to_string().starts_with("roster.yaml"), "{refusal}");
1204 }
1205 }
1206
1207 const TEN_CALLERS_WARNING: &str =
1208 "warning[W-ROS-001]: roster.yaml lets 10 members call 'reviewer' \
1209 (w01, w02, w03, w04, w05, w06, w07, w08, w09, w10), but it holds 2 tasks at once — one \
1210 running and lifecycle.queue_depth: 1 waiting — so a call that arrives while it is full \
1211 waits and is offered again, and is rejected only if it stays full until the call's \
1212 deadline; lifecycle.queue_depth: 9 would hold them all \
1213 (https://docs.murmur.nexus/reference/diagnostics/#w-ros-001)";
1214
1215 fn manifest_with_lifecycle(name: &str, lifecycle_yaml: &str) -> String {
1218 let mut yaml = manifest(name, "1.0.0", true, true);
1219 if !lifecycle_yaml.is_empty() {
1220 yaml.push_str(&format!("lifecycle:\n {lifecycle_yaml}\n"));
1221 }
1222 yaml
1223 }
1224
1225 const QUEUE_DEPTH_1: &str = "task_acceptance: queue\n after_task: sleep";
1226
1227 fn queue_at(depth: usize) -> String {
1228 format!("task_acceptance: queue\n after_task: sleep\n queue_depth: {depth}")
1229 }
1230
1231 fn admit_with(
1233 members: &[(&str, &str)],
1234 reachability: RosterReachability,
1235 ) -> (tempfile::TempDir, AdmittedRoster) {
1236 let dir = tempfile::tempdir().unwrap();
1237 let store = LocalRegistry::new(dir.path());
1238 for (name, lifecycle) in members {
1239 install(
1240 &store,
1241 name,
1242 "1.0.0",
1243 &manifest_with_lifecycle(name, lifecycle),
1244 );
1245 }
1246 let roster = Roster {
1247 members: members
1248 .iter()
1249 .enumerate()
1250 .map(|(index, (name, _))| member(name, "1.0.0", index == 0))
1251 .collect(),
1252 reachability,
1253 };
1254 let admitted = admit_roster(&roster, &store, None).unwrap();
1255 (dir, admitted)
1256 }
1257
1258 const WORKERS: [&str; 10] = [
1259 "w01", "w02", "w03", "w04", "w05", "w06", "w07", "w08", "w09", "w10",
1260 ];
1261
1262 fn ten_workers_one_reviewer(reviewer_lifecycle: &str) -> (tempfile::TempDir, AdmittedRoster) {
1264 let mut members = vec![("lead", "")];
1265 members.extend(WORKERS.iter().map(|worker| (*worker, QUEUE_DEPTH_1)));
1266 members.push(("reviewer", reviewer_lifecycle));
1267 let mut rules = vec![rule("lead", &WORKERS)];
1268 rules.extend(WORKERS.iter().map(|worker| rule(worker, &["reviewer"])));
1269 admit_with(&members, RosterReachability::Rules(rules))
1270 }
1271
1272 #[test]
1273 fn ten_callers_overflow_a_reviewer_at_depth_one() {
1274 let (_dir, admitted) = ten_workers_one_reviewer(QUEUE_DEPTH_1);
1275 assert_eq!(
1276 admitted.caller_overflows(),
1277 [CallerOverflow {
1278 member: "reviewer".to_string(),
1279 capsule: "local/reviewer".to_string(),
1280 version: "1.0.0".to_string(),
1281 callers: WORKERS.iter().map(ToString::to_string).collect(),
1282 task_acceptance: TaskAcceptance::Queue,
1283 queue_depth: 1,
1284 holds: 2,
1285 sufficient_queue_depth: 9,
1286 }]
1287 );
1288 }
1289
1290 #[test]
1291 fn the_ten_callers_warning_line_is_exact() {
1292 let (_dir, admitted) = ten_workers_one_reviewer(QUEUE_DEPTH_1);
1293 let overflows = admitted.caller_overflows();
1294 assert_eq!(caller_overflow_warning(&overflows[0]), TEN_CALLERS_WARNING);
1295 assert_eq!(
1296 caller_overflow_fix(&overflows[0]),
1297 "reviewer (local/reviewer@1.0.0): set lifecycle.queue_depth: 9 in its murmur.yaml so \
1298 all 10 members that may call it are held at once, or narrow its callers in \
1299 roster.yaml (warning[W-ROS-001])"
1300 );
1301 }
1302
1303 #[test]
1304 fn a_reviewer_deep_enough_for_every_caller_does_not_overflow() {
1305 for depth in [9, 10] {
1306 let (_dir, admitted) = ten_workers_one_reviewer(&queue_at(depth));
1307 assert!(
1308 admitted.caller_overflows().is_empty(),
1309 "depth {depth} holds all ten callers"
1310 );
1311 }
1312 }
1313
1314 #[test]
1315 fn all_counts_every_expanded_caller_and_never_the_entry_member() {
1316 for (lifecycle, overflows) in [
1317 (String::new(), true),
1318 (QUEUE_DEPTH_1.to_string(), true),
1319 (queue_at(2), false),
1320 ] {
1321 let members: Vec<(&str, &str)> = ["a", "b", "c", "d"]
1322 .iter()
1323 .map(|name| (*name, lifecycle.as_str()))
1324 .collect();
1325 let (_dir, admitted) = admit_with(&members, RosterReachability::All);
1326 let found = admitted.caller_overflows();
1327 if !overflows {
1328 assert!(found.is_empty(), "'{lifecycle}' holds three callers");
1329 continue;
1330 }
1331 let named: Vec<_> = found.iter().map(|o| o.member.as_str()).collect();
1332 assert_eq!(named, ["b", "c", "d"], "'{lifecycle}'");
1333 for overflow in &found {
1334 assert_eq!(overflow.callers.len(), 3, "{overflow:?}");
1335 assert!(!overflow.callers.contains(&overflow.member));
1336 assert_eq!(overflow.sufficient_queue_depth, 2);
1337 }
1338 assert_eq!(found[0].callers, ["a", "c", "d"]);
1339 }
1340 }
1341
1342 #[test]
1343 fn a_single_member_with_two_callers_is_told_to_switch_to_queue() {
1344 let (_dir, admitted) = admit_with(
1345 &[("lead", ""), ("helper", ""), ("coder", "")],
1346 RosterReachability::Rules(vec![
1347 rule("lead", &["coder", "helper"]),
1348 rule("helper", &["coder"]),
1349 ]),
1350 );
1351 let overflows = admitted.caller_overflows();
1352 assert_eq!(overflows.len(), 1);
1353 let overflow = &overflows[0];
1354 assert_eq!(overflow.member, "coder");
1355 assert_eq!(overflow.callers, ["lead", "helper"]);
1356 assert_eq!(overflow.task_acceptance, TaskAcceptance::Single);
1357 assert_eq!(overflow.holds, 1);
1358 assert_eq!(overflow.sufficient_queue_depth, 1);
1359 assert_eq!(
1360 caller_overflow_warning(overflow),
1361 "warning[W-ROS-001]: roster.yaml lets 2 members call 'coder' (lead, helper), but it \
1362 holds 1 task at once — lifecycle.task_acceptance: single — so a call that arrives \
1363 while it is busy waits and is offered again, and is rejected only if it stays busy \
1364 until the call's deadline; lifecycle.task_acceptance: queue with \
1365 lifecycle.queue_depth: 1 would hold them all \
1366 (https://docs.murmur.nexus/reference/diagnostics/#w-ros-001)"
1367 );
1368 assert_eq!(
1369 caller_overflow_fix(overflow),
1370 "coder (local/coder@1.0.0): set lifecycle.task_acceptance: queue and \
1371 lifecycle.queue_depth: 1 in its murmur.yaml so all 2 members that may call it are \
1372 held at once, or narrow its callers in roster.yaml (warning[W-ROS-001])"
1373 );
1374 }
1375
1376 #[test]
1377 fn a_member_that_accepts_no_task_overflows_with_one_caller() {
1378 let (_dir, admitted) = admit_with(
1379 &[("lead", ""), ("idle", "task_acceptance: none")],
1380 RosterReachability::Rules(vec![rule("lead", &["idle"])]),
1381 );
1382 let overflows = admitted.caller_overflows();
1383 assert_eq!(overflows.len(), 1);
1384 assert_eq!(overflows[0].holds, 0);
1385 assert_eq!(
1386 caller_overflow_warning(&overflows[0]),
1387 "warning[W-ROS-001]: roster.yaml lets 1 member call 'idle' (lead), but it holds no \
1388 task — lifecycle.task_acceptance: none — so no call is ever taken: each is offered \
1389 again until its deadline and then rejected; \
1390 lifecycle.task_acceptance: queue with lifecycle.queue_depth: 1 would hold them all \
1391 (https://docs.murmur.nexus/reference/diagnostics/#w-ros-001)"
1392 );
1393 assert_eq!(
1394 caller_overflow_fix(&overflows[0]),
1395 "idle (local/idle@1.0.0): set lifecycle.task_acceptance: queue and \
1396 lifecycle.queue_depth: 1 in its murmur.yaml so the 1 member that may call it is held \
1397 at once, or narrow its callers in roster.yaml (warning[W-ROS-001])"
1398 );
1399 }
1400
1401 #[test]
1404 fn a_queue_member_at_depth_zero_is_refused_at_admission() {
1405 let dir = tempfile::tempdir().unwrap();
1406 let store = LocalRegistry::new(dir.path());
1407 for (name, lifecycle) in [("lead", String::new()), ("shut", queue_at(0))] {
1408 install(
1409 &store,
1410 name,
1411 "1.0.0",
1412 &manifest_with_lifecycle(name, &lifecycle),
1413 );
1414 }
1415 let roster = Roster {
1416 members: vec![
1417 member("lead", "1.0.0", true),
1418 member("shut", "1.0.0", false),
1419 ],
1420 reachability: RosterReachability::Rules(vec![rule("lead", &["shut"])]),
1421 };
1422 match admit_roster(&roster, &store, None) {
1423 Err(RosterRefusal::MemberUnresolvable { member, reason, .. }) => {
1424 assert_eq!(member, "shut");
1425 assert!(reason.contains("lifecycle.queue_depth"), "{reason}");
1426 }
1427 other => panic!("expected MemberUnresolvable for shut, got {other:?}"),
1428 }
1429 }
1430
1431 #[test]
1434 fn the_entry_member_is_never_named() {
1435 let (_dir, admitted) = admit_with(
1436 &[
1437 ("lead", "task_acceptance: none"),
1438 ("a", ""),
1439 ("b", &queue_at(5)),
1440 ],
1441 RosterReachability::All,
1442 );
1443 assert!(!admitted.edges().is_empty());
1444 assert!(admitted
1445 .caller_overflows()
1446 .iter()
1447 .all(|overflow| overflow.member != "lead"));
1448 }
1449
1450 #[test]
1451 fn a_roster_with_no_edges_has_no_overflow() {
1452 let (_dir, admitted) = admit_with(
1453 &[
1454 ("lead", "task_acceptance: none"),
1455 ("idle", "task_acceptance: none"),
1456 ],
1457 RosterReachability::Closed,
1458 );
1459 assert!(admitted.edges().is_empty());
1460 assert!(admitted.caller_overflows().is_empty());
1461 }
1462}