supervisor.rs (25783B)
1 //! Join-owned task supervision with typed completion policy. 2 3 use core::{fmt, future::Future}; 4 use std::{collections::HashMap, error::Error}; 5 6 use tokio::task::{Id, JoinError, JoinSet}; 7 8 use crate::HostError; 9 10 use super::{CancellationToken, ShutdownPhase, TaskClassification, TaskMetadata, UnfinishedWork}; 11 12 /// Owns every spawned service task until its join result is observed. 13 #[must_use = "a task supervisor must be run or drained so authoritative tasks are joined"] 14 pub struct TaskSupervisor { 15 cancellation: CancellationToken, 16 tasks: JoinSet<TaskCompletion>, 17 controls: HashMap<Id, TaskControl>, 18 } 19 20 struct TaskControl { 21 metadata: TaskMetadata, 22 cancellation: CancellationToken, 23 } 24 25 impl TaskSupervisor { 26 pub fn new() -> Self { 27 Self { 28 cancellation: CancellationToken::new(), 29 tasks: JoinSet::new(), 30 controls: HashMap::new(), 31 } 32 } 33 34 /// Returns a cloneable observer and cancellation authority for composition boundaries. 35 #[must_use] 36 pub fn cancellation_token(&self) -> CancellationToken { 37 self.cancellation.clone() 38 } 39 40 /// Requests cooperative cancellation of every task child token. 41 pub fn request_cancellation(&self) { 42 self.cancellation.cancel(); 43 } 44 45 #[must_use] 46 pub fn task_count(&self) -> usize { 47 self.tasks.len() 48 } 49 50 #[must_use] 51 pub fn is_empty(&self) -> bool { 52 self.tasks.is_empty() 53 } 54 55 /// Registers and spawns one task on the current runtime without exposing a detachable handle. 56 pub fn spawn<F, Fut>( 57 &mut self, 58 metadata: TaskMetadata, 59 task: F, 60 ) -> Result<(), TaskRegistrationError> 61 where 62 F: FnOnce(CancellationToken) -> Fut + Send + 'static, 63 Fut: Future<Output = Result<(), HostError>> + Send + 'static, 64 { 65 if self 66 .controls 67 .values() 68 .any(|active| active.metadata.name() == metadata.name()) 69 { 70 return Err(TaskRegistrationError::DuplicateName); 71 } 72 let runtime = 73 tokio::runtime::Handle::try_current().map_err(|_| TaskRegistrationError::NoRuntime)?; 74 let task_metadata = metadata.clone(); 75 let child = self.cancellation.child_token(); 76 let completion_observer = child.clone(); 77 let phase_cancellation = child.clone(); 78 let abort = self.tasks.spawn_on( 79 async move { 80 let result = task(child).await; 81 TaskCompletion { 82 metadata: task_metadata, 83 cancelled_at_return: completion_observer.is_cancelled(), 84 result, 85 } 86 }, 87 &runtime, 88 ); 89 self.controls.insert( 90 abort.id(), 91 TaskControl { 92 metadata, 93 cancellation: phase_cancellation, 94 }, 95 ); 96 Ok(()) 97 } 98 99 pub(crate) fn request_phase_cancellation(&self, phase: ShutdownPhase) { 100 for control in self.controls.values() { 101 if control.metadata.shutdown_phase() == Some(phase) { 102 control.cancellation.cancel(); 103 } 104 } 105 } 106 107 pub(crate) async fn supervise_phase( 108 &mut self, 109 phase: ShutdownPhase, 110 ) -> Result<Vec<SupervisedTaskExit>, SupervisionFailure> { 111 let mut exits = Vec::new(); 112 while self.has_phase_work(phase) { 113 let outcome = self 114 .join_next() 115 .await 116 .expect("phase work must retain a join-owned task"); 117 match outcome { 118 Ok(exit) => exits.push(exit), 119 Err(failure) => return Err(failure), 120 } 121 } 122 Ok(exits) 123 } 124 125 /// Observes all task exits, cancels peers on the first fatal outcome, and drains every join. 126 pub async fn supervise(&mut self) -> Result<Vec<SupervisedTaskExit>, SupervisionFailure> { 127 let mut exits = Vec::with_capacity(self.tasks.len()); 128 let mut first_failure = None; 129 while let Some(outcome) = self.join_next().await { 130 match outcome { 131 Ok(exit) => exits.push(exit), 132 Err(failure) => { 133 if first_failure.is_none() { 134 first_failure = Some(failure); 135 } 136 } 137 } 138 } 139 if let Some(failure) = first_failure { 140 Err(failure) 141 } else { 142 Ok(exits) 143 } 144 } 145 146 /// Returns the next classified task outcome as soon as it is joined. 147 /// 148 /// Optional failures are returned without cancelling peers. A fatal 149 /// outcome requests cancellation before it is returned. Cancelling this 150 /// wait does not remove a task or lose its later outcome. 151 pub async fn join_next(&mut self) -> Option<Result<SupervisedTaskExit, SupervisionFailure>> { 152 let joined = self.tasks.join_next_with_id().await?; 153 let outcome = self.classify_join(joined); 154 if outcome.is_err() { 155 self.cancellation.cancel(); 156 } 157 Some(outcome) 158 } 159 160 fn classify_join( 161 &mut self, 162 joined: Result<(Id, TaskCompletion), JoinError>, 163 ) -> Result<SupervisedTaskExit, SupervisionFailure> { 164 match joined { 165 Ok((id, completion)) => { 166 self.controls.remove(&id); 167 classify_completion(completion) 168 } 169 Err(error) => { 170 let metadata = self 171 .controls 172 .remove(&error.id()) 173 .map(|control| control.metadata); 174 let cancelled_during_shutdown = 175 error.is_cancelled() && self.cancellation.is_cancelled(); 176 if cancelled_during_shutdown && let Some(metadata) = metadata { 177 return Ok(SupervisedTaskExit::expected(metadata)); 178 } 179 let kind = if error.is_panic() { 180 SupervisionFailureKind::TaskPanicked 181 } else if error.is_cancelled() { 182 SupervisionFailureKind::UnexpectedCancellation 183 } else { 184 SupervisionFailureKind::JoinFailed 185 }; 186 match metadata { 187 Some(metadata) if !metadata.classification().failure_is_fatal() => Ok( 188 SupervisedTaskExit::optional_failure(metadata, Box::new(error)), 189 ), 190 metadata => Err(SupervisionFailure::new( 191 metadata, 192 kind, 193 Some(Box::new(error)), 194 )), 195 } 196 } 197 } 198 } 199 200 pub(crate) fn unfinished_work(&self) -> UnfinishedWork { 201 if self.controls.is_empty() { 202 UnfinishedWork::None 203 } else if self 204 .controls 205 .values() 206 .any(|control| control.metadata.classification().failure_is_fatal()) 207 { 208 UnfinishedWork::FatalAuthoritative 209 } else { 210 UnfinishedWork::RecoverableOptional 211 } 212 } 213 214 pub(crate) async fn abort_and_drain(&mut self) { 215 self.cancellation.cancel(); 216 self.tasks.abort_all(); 217 while self.join_next().await.is_some() {} 218 } 219 220 fn has_phase_work(&self, phase: ShutdownPhase) -> bool { 221 self.controls.values().any(|control| { 222 control.metadata.shutdown_phase() == Some(phase) 223 || (phase == ShutdownPhase::DrainOperations 224 && control.metadata.classification() == TaskClassification::OneShot) 225 }) 226 } 227 } 228 229 impl Default for TaskSupervisor { 230 fn default() -> Self { 231 Self::new() 232 } 233 } 234 235 fn classify_completion( 236 completion: TaskCompletion, 237 ) -> Result<SupervisedTaskExit, SupervisionFailure> { 238 match completion.result { 239 Ok(()) 240 if completion.metadata.classification() == TaskClassification::Critical 241 && !completion.cancelled_at_return => 242 { 243 Err(SupervisionFailure::new( 244 Some(completion.metadata), 245 SupervisionFailureKind::UnexpectedCompletion, 246 None, 247 )) 248 } 249 Ok(()) => Ok(SupervisedTaskExit::expected(completion.metadata)), 250 Err(error) if completion.metadata.classification().failure_is_fatal() => { 251 Err(SupervisionFailure::new( 252 Some(completion.metadata), 253 SupervisionFailureKind::TaskReturnedError, 254 Some(Box::new(error)), 255 )) 256 } 257 Err(error) => Ok(SupervisedTaskExit::optional_failure( 258 completion.metadata, 259 Box::new(error), 260 )), 261 } 262 } 263 264 struct TaskCompletion { 265 metadata: TaskMetadata, 266 cancelled_at_return: bool, 267 result: Result<(), HostError>, 268 } 269 270 /// Nonfatal observed task completion retained for metrics and status consumers. 271 pub struct SupervisedTaskExit { 272 metadata: TaskMetadata, 273 status: SupervisedTaskExitStatus, 274 source: Option<Box<dyn Error + Send + Sync + 'static>>, 275 } 276 277 impl SupervisedTaskExit { 278 fn expected(metadata: TaskMetadata) -> Self { 279 Self { 280 metadata, 281 status: SupervisedTaskExitStatus::ExpectedCompletion, 282 source: None, 283 } 284 } 285 286 fn optional_failure( 287 metadata: TaskMetadata, 288 source: Box<dyn Error + Send + Sync + 'static>, 289 ) -> Self { 290 Self { 291 metadata, 292 status: SupervisedTaskExitStatus::OptionalFailure, 293 source: Some(source), 294 } 295 } 296 297 #[must_use] 298 pub const fn metadata(&self) -> &TaskMetadata { 299 &self.metadata 300 } 301 302 #[must_use] 303 pub const fn status(&self) -> SupervisedTaskExitStatus { 304 self.status 305 } 306 307 /// Returns the trusted internal cause for an optional failure. 308 #[must_use] 309 pub fn source(&self) -> Option<&(dyn Error + Send + Sync + 'static)> { 310 self.source.as_deref() 311 } 312 } 313 314 impl fmt::Debug for SupervisedTaskExit { 315 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { 316 formatter 317 .debug_struct("SupervisedTaskExit") 318 .field("metadata", &self.metadata) 319 .field("status", &self.status) 320 .field("source", &self.source.as_ref().map(|_| "<redacted>")) 321 .finish() 322 } 323 } 324 325 #[derive(Clone, Copy, Debug, PartialEq, Eq)] 326 pub enum SupervisedTaskExitStatus { 327 ExpectedCompletion, 328 OptionalFailure, 329 } 330 331 /// Fatal supervisor outcome after every remaining task has been joined. 332 pub struct SupervisionFailure { 333 metadata: Option<TaskMetadata>, 334 kind: SupervisionFailureKind, 335 source: Option<Box<dyn Error + Send + Sync + 'static>>, 336 } 337 338 impl SupervisionFailure { 339 fn new( 340 metadata: Option<TaskMetadata>, 341 kind: SupervisionFailureKind, 342 source: Option<Box<dyn Error + Send + Sync + 'static>>, 343 ) -> Self { 344 Self { 345 metadata, 346 kind, 347 source, 348 } 349 } 350 351 #[must_use] 352 pub const fn metadata(&self) -> Option<&TaskMetadata> { 353 self.metadata.as_ref() 354 } 355 356 #[must_use] 357 pub const fn kind(&self) -> SupervisionFailureKind { 358 self.kind 359 } 360 } 361 362 impl fmt::Debug for SupervisionFailure { 363 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { 364 formatter 365 .debug_struct("SupervisionFailure") 366 .field("metadata", &self.metadata) 367 .field("kind", &self.kind) 368 .field("source", &self.source.as_ref().map(|_| "<redacted>")) 369 .finish() 370 } 371 } 372 373 impl fmt::Display for SupervisionFailure { 374 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { 375 formatter.write_str("authoritative service task supervision failed") 376 } 377 } 378 379 impl Error for SupervisionFailure { 380 fn source(&self) -> Option<&(dyn Error + 'static)> { 381 self.source 382 .as_deref() 383 .map(|source| source as &(dyn Error + 'static)) 384 } 385 } 386 387 #[derive(Clone, Copy, Debug, PartialEq, Eq)] 388 pub enum SupervisionFailureKind { 389 TaskReturnedError, 390 TaskPanicked, 391 UnexpectedCompletion, 392 UnexpectedCancellation, 393 JoinFailed, 394 } 395 396 /// Failure to register a task without spawning it. 397 #[derive(Clone, Copy, Debug, PartialEq, Eq)] 398 pub enum TaskRegistrationError { 399 DuplicateName, 400 NoRuntime, 401 } 402 403 impl fmt::Display for TaskRegistrationError { 404 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { 405 formatter.write_str(match self { 406 Self::DuplicateName => "supervised task name is already active", 407 Self::NoRuntime => "supervised task registration requires an active runtime", 408 }) 409 } 410 } 411 412 impl Error for TaskRegistrationError {} 413 414 #[cfg(test)] 415 mod tests { 416 use std::sync::{ 417 Arc, 418 atomic::{AtomicUsize, Ordering}, 419 }; 420 421 use crate::{HostErrorKind, ShutdownPhase, TaskClassification, TaskName}; 422 423 use super::*; 424 425 fn metadata(name: &str, classification: TaskClassification) -> TaskMetadata { 426 let shutdown_phase = classification 427 .requires_shutdown_phase() 428 .then_some(ShutdownPhase::CancelIngress); 429 TaskMetadata::new(TaskName::new(name).unwrap(), classification, shutdown_phase).unwrap() 430 } 431 432 fn metadata_at( 433 name: &str, 434 classification: TaskClassification, 435 shutdown_phase: Option<ShutdownPhase>, 436 ) -> TaskMetadata { 437 TaskMetadata::new(TaskName::new(name).unwrap(), classification, shutdown_phase).unwrap() 438 } 439 440 #[test] 441 fn registration_without_a_runtime_fails_before_spawning() { 442 let mut supervisor = TaskSupervisor::new(); 443 assert_eq!( 444 supervisor.spawn( 445 metadata("critical_worker", TaskClassification::Critical), 446 |_| async { Ok(()) }, 447 ), 448 Err(TaskRegistrationError::NoRuntime) 449 ); 450 assert!(supervisor.is_empty()); 451 } 452 453 #[tokio::test] 454 async fn critical_error_cancels_peers_propagates_and_drains_all_joins() { 455 let drained = Arc::new(AtomicUsize::new(0)); 456 let mut supervisor = TaskSupervisor::new(); 457 let peer_drained = Arc::clone(&drained); 458 supervisor 459 .spawn( 460 metadata("peer_worker", TaskClassification::Critical), 461 move |token| async move { 462 token.cancelled().await; 463 peer_drained.fetch_add(1, Ordering::SeqCst); 464 Ok(()) 465 }, 466 ) 467 .unwrap(); 468 supervisor 469 .spawn( 470 metadata("failing_worker", TaskClassification::Critical), 471 |_| async { Err(HostError::new(HostErrorKind::TaskFailure)) }, 472 ) 473 .unwrap(); 474 475 let error = supervisor.supervise().await.unwrap_err(); 476 assert_eq!(error.kind(), SupervisionFailureKind::TaskReturnedError); 477 assert_eq!(error.metadata().unwrap().name().as_str(), "failing_worker"); 478 assert!(error.source().is_some()); 479 assert_eq!(drained.load(Ordering::SeqCst), 1); 480 assert!(supervisor.is_empty()); 481 assert!(supervisor.cancellation_token().is_cancelled()); 482 } 483 484 #[tokio::test] 485 async fn phase_cancellation_stops_and_joins_only_the_assigned_tasks() { 486 let ingress_stopped = Arc::new(AtomicUsize::new(0)); 487 let network_stopped = Arc::new(AtomicUsize::new(0)); 488 let mut supervisor = TaskSupervisor::new(); 489 let ingress = Arc::clone(&ingress_stopped); 490 supervisor 491 .spawn( 492 metadata_at( 493 "ingress_worker", 494 TaskClassification::Critical, 495 Some(ShutdownPhase::CancelIngress), 496 ), 497 move |token| async move { 498 token.cancelled().await; 499 ingress.fetch_add(1, Ordering::SeqCst); 500 Ok(()) 501 }, 502 ) 503 .unwrap(); 504 let network = Arc::clone(&network_stopped); 505 supervisor 506 .spawn( 507 metadata_at( 508 "network_worker", 509 TaskClassification::Critical, 510 Some(ShutdownPhase::CloseNetwork), 511 ), 512 move |token| async move { 513 token.cancelled().await; 514 network.fetch_add(1, Ordering::SeqCst); 515 Ok(()) 516 }, 517 ) 518 .unwrap(); 519 520 supervisor.request_phase_cancellation(ShutdownPhase::CancelIngress); 521 let ingress_exits = supervisor 522 .supervise_phase(ShutdownPhase::CancelIngress) 523 .await 524 .unwrap(); 525 assert_eq!(ingress_exits.len(), 1); 526 assert_eq!(ingress_stopped.load(Ordering::SeqCst), 1); 527 assert_eq!(network_stopped.load(Ordering::SeqCst), 0); 528 assert_eq!(supervisor.task_count(), 1); 529 530 supervisor.request_phase_cancellation(ShutdownPhase::CloseNetwork); 531 let network_exits = supervisor 532 .supervise_phase(ShutdownPhase::CloseNetwork) 533 .await 534 .unwrap(); 535 assert_eq!(network_exits.len(), 1); 536 assert_eq!(network_stopped.load(Ordering::SeqCst), 1); 537 assert!(supervisor.is_empty()); 538 } 539 540 #[tokio::test] 541 async fn drain_operations_joins_one_shot_work_without_cancelling_it() { 542 let mut supervisor = TaskSupervisor::new(); 543 supervisor 544 .spawn( 545 metadata_at("bounded_operation", TaskClassification::OneShot, None), 546 |token| async move { 547 assert!(!token.is_cancelled()); 548 Ok(()) 549 }, 550 ) 551 .unwrap(); 552 553 supervisor.request_phase_cancellation(ShutdownPhase::DrainOperations); 554 let exits = supervisor 555 .supervise_phase(ShutdownPhase::DrainOperations) 556 .await 557 .unwrap(); 558 assert_eq!(exits.len(), 1); 559 assert_eq!( 560 exits[0].status(), 561 SupervisedTaskExitStatus::ExpectedCompletion 562 ); 563 assert!(supervisor.is_empty()); 564 } 565 566 #[tokio::test] 567 async fn critical_panic_and_early_success_are_fatal() { 568 for (name, expected, panic_task) in [ 569 ("panic_worker", SupervisionFailureKind::TaskPanicked, true), 570 ( 571 "early_worker", 572 SupervisionFailureKind::UnexpectedCompletion, 573 false, 574 ), 575 ] { 576 let mut supervisor = TaskSupervisor::new(); 577 supervisor 578 .spawn( 579 metadata(name, TaskClassification::Critical), 580 move |_| async move { 581 assert!(!panic_task, "sensitive panic payload"); 582 Ok(()) 583 }, 584 ) 585 .unwrap(); 586 let error = supervisor.supervise().await.unwrap_err(); 587 assert_eq!(error.kind(), expected); 588 assert!(!error.to_string().contains("sensitive")); 589 assert!(supervisor.is_empty()); 590 } 591 } 592 593 #[tokio::test] 594 async fn optional_error_and_one_shot_success_are_observed_without_failure() { 595 let mut supervisor = TaskSupervisor::new(); 596 supervisor 597 .spawn( 598 metadata("optional_worker", TaskClassification::Optional), 599 |_| async { Err(HostError::new(HostErrorKind::TaskFailure)) }, 600 ) 601 .unwrap(); 602 supervisor 603 .spawn( 604 metadata("startup_once", TaskClassification::OneShot), 605 |_| async { Ok(()) }, 606 ) 607 .unwrap(); 608 609 let mut exits = supervisor.supervise().await.unwrap(); 610 exits.sort_by(|left, right| left.metadata().name().cmp(right.metadata().name())); 611 assert_eq!(exits.len(), 2); 612 assert_eq!(exits[0].status(), SupervisedTaskExitStatus::OptionalFailure); 613 assert!(exits[0].source().is_some()); 614 assert_eq!( 615 exits[1].status(), 616 SupervisedTaskExitStatus::ExpectedCompletion 617 ); 618 assert!(supervisor.is_empty()); 619 } 620 621 #[tokio::test] 622 async fn optional_failure_is_observable_while_a_critical_peer_is_alive() { 623 let mut supervisor = TaskSupervisor::new(); 624 supervisor 625 .spawn( 626 metadata("critical_worker", TaskClassification::Critical), 627 |token| async move { 628 token.cancelled().await; 629 Ok(()) 630 }, 631 ) 632 .unwrap(); 633 supervisor 634 .spawn( 635 metadata("optional_worker", TaskClassification::Optional), 636 |_| async { Err(HostError::new(HostErrorKind::TaskFailure)) }, 637 ) 638 .unwrap(); 639 640 let exit = supervisor.join_next().await.unwrap().unwrap(); 641 assert_eq!(exit.metadata().name().as_str(), "optional_worker"); 642 assert_eq!(exit.status(), SupervisedTaskExitStatus::OptionalFailure); 643 assert_eq!(supervisor.task_count(), 1); 644 assert!(!supervisor.cancellation_token().is_cancelled()); 645 646 supervisor.request_cancellation(); 647 assert!(supervisor.join_next().await.unwrap().is_ok()); 648 assert!(supervisor.is_empty()); 649 } 650 651 #[tokio::test] 652 async fn optional_failure_remains_observed_before_a_later_fatal_exit() { 653 let (release_fatal, wait_for_release) = tokio::sync::oneshot::channel(); 654 let mut supervisor = TaskSupervisor::new(); 655 supervisor 656 .spawn( 657 metadata("fatal_worker", TaskClassification::Critical), 658 |_| async move { 659 let _ = wait_for_release.await; 660 Err(HostError::new(HostErrorKind::TaskFailure)) 661 }, 662 ) 663 .unwrap(); 664 supervisor 665 .spawn( 666 metadata("optional_worker", TaskClassification::Optional), 667 |_| async { Err(HostError::new(HostErrorKind::TaskFailure)) }, 668 ) 669 .unwrap(); 670 671 let optional = supervisor.join_next().await.unwrap().unwrap(); 672 assert_eq!(optional.metadata().name().as_str(), "optional_worker"); 673 assert_eq!(optional.status(), SupervisedTaskExitStatus::OptionalFailure); 674 release_fatal.send(()).unwrap(); 675 676 let fatal = supervisor.join_next().await.unwrap().unwrap_err(); 677 assert_eq!(fatal.kind(), SupervisionFailureKind::TaskReturnedError); 678 assert_eq!(fatal.metadata().unwrap().name().as_str(), "fatal_worker"); 679 assert!(supervisor.cancellation_token().is_cancelled()); 680 assert!(supervisor.is_empty()); 681 } 682 683 #[tokio::test] 684 async fn supervision_retains_only_the_first_of_multiple_fatal_failures() { 685 let mut supervisor = TaskSupervisor::new(); 686 for name in ["first_failure", "second_failure"] { 687 supervisor 688 .spawn(metadata(name, TaskClassification::Critical), |_| async { 689 Err(HostError::new(HostErrorKind::TaskFailure)) 690 }) 691 .unwrap(); 692 } 693 694 let error = supervisor.supervise().await.unwrap_err(); 695 assert_eq!(error.kind(), SupervisionFailureKind::TaskReturnedError); 696 assert!(matches!( 697 error.metadata().unwrap().name().as_str(), 698 "first_failure" | "second_failure" 699 )); 700 assert!(supervisor.is_empty()); 701 } 702 703 #[tokio::test] 704 async fn optional_panic_is_an_observable_nonfatal_exit() { 705 let mut supervisor = TaskSupervisor::new(); 706 supervisor 707 .spawn( 708 metadata("optional_panic", TaskClassification::Optional), 709 |_| async { 710 panic!("redacted optional panic"); 711 #[allow(unreachable_code)] 712 Ok(()) 713 }, 714 ) 715 .unwrap(); 716 717 let exit = supervisor.join_next().await.unwrap().unwrap(); 718 assert_eq!(exit.status(), SupervisedTaskExitStatus::OptionalFailure); 719 assert!(exit.source().is_some()); 720 assert!(supervisor.is_empty()); 721 } 722 723 #[tokio::test] 724 async fn externally_cancelled_critical_task_may_complete_successfully() { 725 let mut supervisor = TaskSupervisor::new(); 726 supervisor 727 .spawn( 728 metadata("critical_worker", TaskClassification::Critical), 729 |token| async move { 730 token.cancelled().await; 731 Ok(()) 732 }, 733 ) 734 .unwrap(); 735 supervisor.request_cancellation(); 736 737 let exits = supervisor.supervise().await.unwrap(); 738 assert_eq!(exits.len(), 1); 739 assert_eq!( 740 exits[0].status(), 741 SupervisedTaskExitStatus::ExpectedCompletion 742 ); 743 assert!(supervisor.is_empty()); 744 } 745 746 #[tokio::test] 747 async fn duplicate_active_names_are_rejected_without_detaching_work() { 748 let mut supervisor = TaskSupervisor::new(); 749 supervisor 750 .spawn( 751 metadata("same_worker", TaskClassification::Critical), 752 |token| async move { 753 token.cancelled().await; 754 Ok(()) 755 }, 756 ) 757 .unwrap(); 758 assert_eq!( 759 supervisor.spawn( 760 metadata("same_worker", TaskClassification::Critical), 761 |_| async { Ok(()) }, 762 ), 763 Err(TaskRegistrationError::DuplicateName) 764 ); 765 assert_eq!(supervisor.task_count(), 1); 766 supervisor.request_cancellation(); 767 assert!(supervisor.supervise().await.is_ok()); 768 } 769 }