diff --git a/crates/irtt-client/src/managed/model.rs b/crates/irtt-client/src/managed/model.rs index 4056717..4215e3f 100644 --- a/crates/irtt-client/src/managed/model.rs +++ b/crates/irtt-client/src/managed/model.rs @@ -155,6 +155,10 @@ pub struct ManagedTargetStatus { #[derive(Clone, Debug, Eq, PartialEq)] pub struct ManagedStatus { pub lifecycle: ManagedLifecycle, + /// Whether the task durably observed an explicit stop request. + /// + /// This remains false until the task observes the request latch. Natural + /// stopping and terminality alone do not imply a stop request was observed. pub stop_requested: bool, pub applied_command_sequence: u64, pub desired_target_count: usize, @@ -171,9 +175,21 @@ pub struct ManagedStatus { pub discarded_target_outcomes: u64, pub targets: Arc<[ManagedTargetStatus]>, pub recent_target_outcomes: Arc<[ManagedTargetOutcome]>, + /// Authoritative final accounting, present if and only if `lifecycle` is + /// [`ManagedLifecycle::Completed`] or [`ManagedLifecycle::Failed`]. + /// + /// Shares the terminal task phase's outcome with [`ManagedEvent::Completed`] + /// or [`ManagedEvent::Failed`]. Abandoned deliberately has no outcome. pub final_outcome: Option>, } +/// Receiving half of authoritative durable latest-state observation. +/// +/// The current snapshot is immediately readable. Updates may coalesce, and the +/// final snapshot remains available after the sender closes at task termination. +/// Use [`ManagedEventSubscription`] for lossy presentation events. +pub type ManagedStatusSubscription = tokio::sync::watch::Receiver>; + /// Lossy presentation event emitted by the managed task. #[allow(clippy::large_enum_variant)] #[derive(Clone, Debug, Eq, PartialEq)] diff --git a/crates/irtt-client/src/managed/task.rs b/crates/irtt-client/src/managed/task.rs index 98e40e9..28841ab 100644 --- a/crates/irtt-client/src/managed/task.rs +++ b/crates/irtt-client/src/managed/task.rs @@ -29,10 +29,10 @@ use super::{ classify_client_error, ManagedClientConfig, ManagedCommandAcknowledgement, ManagedCommandApplyError, ManagedCommandError, ManagedCompletionPolicy, ManagedConfigError, ManagedDriverFailure, ManagedEndReason, ManagedEvent, ManagedEventSubscription, - ManagedLifecycle, ManagedOutcome, ManagedPacing, ManagedStatus, ManagedSubscribeError, - ManagedTargetConfig, ManagedTargetEndReason, ManagedTargetFailure, ManagedTargetFailureKind, - ManagedTargetFailurePhase, ManagedTargetLifecycle, ManagedTargetOutcome, ManagedTargetStatus, - TargetInstance, + ManagedLifecycle, ManagedOutcome, ManagedPacing, ManagedStatus, ManagedStatusSubscription, + ManagedSubscribeError, ManagedTargetConfig, ManagedTargetEndReason, ManagedTargetFailure, + ManagedTargetFailureKind, ManagedTargetFailurePhase, ManagedTargetLifecycle, + ManagedTargetOutcome, ManagedTargetStatus, TargetInstance, }; const TARGET_WORK_BUDGET: usize = 128; @@ -101,8 +101,6 @@ struct StopSignal { requested: AtomicBool, update_admission: AtomicU8, wake: watch::Sender<()>, - acknowledged: AtomicBool, - acknowledgement: watch::Sender<()>, } #[derive(Clone, Copy, Debug, Eq, PartialEq)] @@ -162,12 +160,6 @@ impl StopSignal { fn is_requested(&self) -> bool { self.requested.load(Ordering::Acquire) } - - fn acknowledge(&self) { - if !self.acknowledged.swap(true, Ordering::AcqRel) { - self.acknowledgement.send_replace(()); - } - } } /// Cloneable control and observation capability for a managed task. @@ -194,6 +186,16 @@ impl ManagedClientHandle { Arc::clone(&self.status.borrow()) } + /// Subscribe to authoritative durable latest-state observation. + /// + /// The current snapshot is immediately available through `borrow()`. Updates + /// may coalesce; this is not a lossless event log. After the task terminates, + /// the sender closes but the final snapshot remains readable. Use + /// [`Self::subscribe`] for the lossy presentation-event stream. + pub fn subscribe_status(&self) -> ManagedStatusSubscription { + self.status.clone() + } + /// Subscribe to future lossy presentation events. pub fn subscribe(&self) -> Result { self.events @@ -205,7 +207,7 @@ impl ManagedClientHandle { /// Request idempotent graceful stop and return a durable-status receipt. pub fn stop(&self) -> ManagedStopReceipt { self.stop.request(); - ManagedStopReceipt::new(Arc::clone(&self.stop)) + ManagedStopReceipt::new(self.subscribe_status()) } /// Submit the complete desired target set, replacing any previous one. @@ -303,17 +305,33 @@ enum ManagedCommand { } /// Receipt resolving once stop is durably observed or the task is terminal. +/// +/// Resolves from authoritative status as soon as `stop_requested` is true or +/// the lifecycle is Completed, Failed, or Abandoned. It need not wait for +/// graceful cleanup after stop observation. #[must_use = "await the receipt to observe durable stop acknowledgement"] pub struct ManagedStopReceipt { future: Pin + Send>>, } impl ManagedStopReceipt { - fn new(stop: Arc) -> Self { - let mut acknowledgement = stop.acknowledgement.subscribe(); + fn new(mut status: ManagedStatusSubscription) -> Self { let future = Box::pin(async move { - while !stop.acknowledged.load(Ordering::Acquire) { - let _ = acknowledgement.changed().await; + while !{ + let snapshot = status.borrow(); + snapshot.stop_requested + || matches!( + snapshot.lifecycle, + ManagedLifecycle::Completed + | ManagedLifecycle::Failed + | ManagedLifecycle::Abandoned + ) + } { + // A closed sender retains the terminal snapshot. Never wait + // indefinitely if the channel closes. + if status.changed().await.is_err() { + break; + } } }); Self { future } @@ -688,12 +706,48 @@ struct TaskResources { stop: Arc, } -#[derive(Clone, Copy, Eq, PartialEq)] -enum DriverState { +enum TaskPhase { NotStarted, Running, Stopping, - Finished, + Completed { outcome: Arc }, + Failed { outcome: Arc }, + Abandoned, +} + +impl TaskPhase { + fn lifecycle(&self) -> ManagedLifecycle { + match self { + Self::NotStarted => ManagedLifecycle::NotStarted, + Self::Running => ManagedLifecycle::Running, + Self::Stopping => ManagedLifecycle::Stopping, + Self::Completed { .. } => ManagedLifecycle::Completed, + Self::Failed { .. } => ManagedLifecycle::Failed, + Self::Abandoned => ManagedLifecycle::Abandoned, + } + } + + fn final_outcome(&self) -> Option> { + match self { + Self::Completed { outcome } | Self::Failed { outcome } => Some(Arc::clone(outcome)), + _ => None, + } + } + + fn is_terminal(&self) -> bool { + matches!( + self, + Self::Completed { .. } | Self::Failed { .. } | Self::Abandoned + ) + } + + fn is_running(&self) -> bool { + matches!(self, Self::Running) + } + + fn is_stopping(&self) -> bool { + matches!(self, Self::Stopping) + } } #[derive(Clone, Copy)] @@ -707,8 +761,7 @@ enum OpenSessionFailureCleanup { pub struct ManagedClientTask { #[cfg(test)] timeout_inspections: std::cell::RefCell>, - state: DriverState, - lifecycle: ManagedLifecycle, + phase: TaskPhase, config: ManagedClientConfig, targets: Vec, commands: mpsc::Receiver, @@ -725,8 +778,8 @@ pub struct ManagedClientTask { stagger_remaining: usize, send_gate: Option, last_stagger_send: Option, + // Explicit stop observation is independent of natural stopping or terminality. stop_observed: bool, - final_outcome: Option>, next_generation: u64, applied_command_sequence: u64, } @@ -779,7 +832,7 @@ impl ManagedClientTask { }) .collect::>(); Arc::new(ManagedStatus { - lifecycle: self.lifecycle, + lifecycle: self.phase.lifecycle(), stop_requested: self.stop_observed, applied_command_sequence: self.applied_command_sequence, desired_target_count: self @@ -800,7 +853,7 @@ impl ManagedClientTask { discarded_target_outcomes: self.history.discarded, targets: Arc::from(targets.into_boxed_slice()), recent_target_outcomes: self.history.recent(), - final_outcome: self.final_outcome.clone(), + final_outcome: self.phase.final_outcome(), }) } @@ -890,8 +943,7 @@ impl ManagedClientTask { fn begin_running(&mut self) -> Result<(), ManagedDriverFailure> { tokio::runtime::Handle::try_current().map_err(|_| ManagedDriverFailure::NoTokioRuntime)?; - self.state = DriverState::Running; - self.lifecycle = ManagedLifecycle::Running; + self.phase = TaskPhase::Running; self.replace_status(); self.publish_event(ManagedEvent::Started); Ok(()) @@ -903,15 +955,12 @@ impl ManagedClientTask { } self.stop_observed = true; self.resources().stop.begin_stopping(); - let lifecycle_transition = - self.state != DriverState::Stopping || self.lifecycle != ManagedLifecycle::Stopping; - self.state = DriverState::Stopping; - self.lifecycle = ManagedLifecycle::Stopping; + let lifecycle_transition = !self.phase.is_stopping(); + self.phase = TaskPhase::Stopping; self.replace_status(); if lifecycle_transition { self.publish_event(ManagedEvent::Stopping); } - self.resources().stop.acknowledge(); self.scan_remaining = self.targets.len(); self.burst_remaining = 0; } @@ -934,7 +983,7 @@ impl ManagedClientTask { // This check is the driver-side application linearization point. // `apply_targets` cannot suspend, so a stop observed here wins over // this command; once it passes, this command may complete atomically. - let result = if self.state != DriverState::Running + let result = if !self.phase.is_running() || self.resources().stop.update_admission() != UpdateAdmission::Open { Err(ManagedCommandApplyError::Stopping) @@ -1015,8 +1064,7 @@ impl ManagedClientTask { && self.config.completion == ManagedCompletionPolicy::FinishWhenQuiescent; if stopping { self.resources().stop.begin_stopping(); - self.state = DriverState::Stopping; - self.lifecycle = ManagedLifecycle::Stopping; + self.phase = TaskPhase::Stopping; } // A transaction has one externally visible durable point: all runtime and @@ -1180,7 +1228,7 @@ impl ManagedClientTask { fn poll_target(&mut self, index: usize, cx: &mut Context<'_>, now: Instant) -> bool { let retirement = self.effective_retirement(index); - let stopping = self.state == DriverState::Stopping || retirement.is_some(); + let stopping = self.phase.is_stopping() || retirement.is_some(); match self.targets[index].phase_mut() { TargetPhase::Pending { .. } => { if stopping { @@ -1956,7 +2004,7 @@ impl ManagedClientTask { self.timeout_inspections.borrow_mut().push(index); match self.targets[index].phase() { TargetPhase::Active { client, .. } => { - self.state != DriverState::Stopping + !self.phase.is_stopping() && self.effective_retirement(index).is_none() && client .next_probe_timeout_deadline() @@ -2090,7 +2138,7 @@ impl ManagedClientTask { _ => {} } } - if self.state == DriverState::Running + if self.phase.is_running() && self.config.pacing == ManagedPacing::Staggered && self.active_count() > 0 { @@ -2160,11 +2208,14 @@ impl ManagedClientTask { self.history .outcome(end_reason, self.applied_command_sequence), ); - self.final_outcome = Some(Arc::clone(&outcome)); - self.lifecycle = if failed { - ManagedLifecycle::Failed + self.phase = if failed { + TaskPhase::Failed { + outcome: Arc::clone(&outcome), + } } else { - ManagedLifecycle::Completed + TaskPhase::Completed { + outcome: Arc::clone(&outcome), + } }; self.replace_status(); self.publish_event(if failed { @@ -2176,11 +2227,9 @@ impl ManagedClientTask { outcome: Arc::clone(&outcome), } }); - self.resources().stop.acknowledge(); if let Some(resources) = self.resources.as_mut() { resources.events.take(); } - self.state = DriverState::Finished; self.wake = None; self.timer = None; self.resources.take(); @@ -2228,11 +2277,11 @@ impl Future for ManagedClientTask { fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { let this = self.as_mut().get_mut(); - if this.state == DriverState::Finished { + if this.phase.is_terminal() { panic!("ManagedClientTask polled after completion"); } - if this.state == DriverState::NotStarted { + if matches!(this.phase, TaskPhase::NotStarted) { if this.resources().stop.is_requested() { this.observe_stop(); } else if let Err(failure) = this.begin_running() { @@ -2251,7 +2300,7 @@ impl Future for ManagedClientTask { immediate |= this.poll_timeout_pass(now); immediate |= this.poll_target_pass(cx, now); this.prune_undesired_terminal(); - if this.state == DriverState::Running { + if this.phase.is_running() { immediate |= match this.config.pacing { ManagedPacing::Staggered => this.poll_staggered_send(cx, now), ManagedPacing::Burst => this.poll_burst_sends(cx, now), @@ -2263,8 +2312,8 @@ impl Future for ManagedClientTask { } if this.all_targets_terminal() { - match this.state { - DriverState::Stopping => { + match this.phase { + TaskPhase::Stopping => { return this.seal( if this.stop_observed { ManagedEndReason::StopRequested @@ -2274,11 +2323,10 @@ impl Future for ManagedClientTask { false, ); } - DriverState::Running + TaskPhase::Running if this.config.completion == ManagedCompletionPolicy::FinishWhenQuiescent => { - this.state = DriverState::Stopping; - this.lifecycle = ManagedLifecycle::Stopping; + this.phase = TaskPhase::Stopping; this.resources().stop.begin_stopping(); this.replace_status(); this.publish_event(ManagedEvent::Stopping); @@ -2301,16 +2349,14 @@ impl Future for ManagedClientTask { impl Drop for ManagedClientTask { fn drop(&mut self) { - if self.state == DriverState::Finished || self.resources.is_none() { + if self.phase.is_terminal() || self.resources.is_none() { return; } - self.lifecycle = ManagedLifecycle::Abandoned; - self.final_outcome = None; + self.phase = TaskPhase::Abandoned; self.resources().stop.close_updates(); self.commands.close(); self.replace_status(); self.publish_event(ManagedEvent::Abandoned); - self.resources().stop.acknowledge(); if let Some(resources) = self.resources.as_mut() { resources.events.take(); } @@ -2394,13 +2440,10 @@ fn build_task( let (event_sender, _) = broadcast::channel(config.event_capacity); let weak_events = event_sender.downgrade(); let (wake_sender, wake_receiver) = watch::channel(()); - let (acknowledgement_sender, _) = watch::channel(()); let stop = Arc::new(StopSignal { requested: AtomicBool::new(false), update_admission: AtomicU8::new(UpdateAdmission::Open as u8), wake: wake_sender, - acknowledged: AtomicBool::new(false), - acknowledgement: acknowledgement_sender, }); let history = OutcomeHistory::new(config.outcome_history_limit); let initial_targets = runtimes @@ -2414,8 +2457,9 @@ fn build_task( remote: None, }) .collect::>(); + let phase = TaskPhase::NotStarted; let initial = Arc::new(ManagedStatus { - lifecycle: ManagedLifecycle::NotStarted, + lifecycle: phase.lifecycle(), stop_requested: false, applied_command_sequence: 0, desired_target_count: runtimes.len(), @@ -2432,7 +2476,7 @@ fn build_task( discarded_target_outcomes: 0, targets: Arc::from(initial_targets.into_boxed_slice()), recent_target_outcomes: history.recent(), - final_outcome: None, + final_outcome: phase.final_outcome(), }); let (status_sender, status_receiver) = watch::channel(initial); let (command_sender, command_receiver) = mpsc::channel(config.command_capacity); @@ -2445,8 +2489,7 @@ fn build_task( let task = ManagedClientTask { #[cfg(test)] timeout_inspections: std::cell::RefCell::new(Vec::new()), - state: DriverState::NotStarted, - lifecycle: ManagedLifecycle::NotStarted, + phase, config, targets: runtimes, commands: command_receiver, @@ -2464,7 +2507,6 @@ fn build_task( send_gate: None, last_stagger_send: None, stop_observed: false, - final_outcome: None, next_generation, applied_command_sequence: 0, }; @@ -2547,7 +2589,7 @@ mod tests { let client = AsyncClient::connect("127.0.0.1:9", ClientConfig::default()) .await .unwrap(); - task.state = DriverState::Running; + task.phase = TaskPhase::Running; task.install_target_phase( 0, TargetPhase::Active { diff --git a/crates/irtt-client/tests/managed_control.rs b/crates/irtt-client/tests/managed_control.rs index 1e7b80d..908ea3b 100644 --- a/crates/irtt-client/tests/managed_control.rs +++ b/crates/irtt-client/tests/managed_control.rs @@ -20,6 +20,13 @@ fn config() -> ManagedClientConfig { #[test] fn stop_resolves_every_accepted_update_and_closes_admission() { let (task, handle) = ManagedClient::task(config(), vec![]).unwrap(); + let mut subscription: ManagedStatusSubscription = handle.subscribe_status(); + assert_eq!( + subscription.borrow().lifecycle, + ManagedLifecycle::NotStarted + ); + assert!(!subscription.borrow().stop_requested); + assert!(subscription.borrow().final_outcome.is_none()); let receipts: Vec<_> = (0..3) .map(|index| { handle @@ -35,7 +42,12 @@ fn stop_resolves_every_accepted_update_and_closes_admission() { Err(ManagedCommandError::QueueFull) )); - let stop = handle.stop(); + let mut stop = pin!(handle.stop()); + assert!(stop + .as_mut() + .poll(&mut Context::from_waker(Waker::noop())) + .is_pending()); + assert!(!handle.status().stop_requested); assert!(matches!( handle.update_targets(vec![]), Err(ManagedCommandError::Stopping) @@ -61,6 +73,16 @@ fn stop_resolves_every_accepted_update_and_closes_admission() { assert_eq!(status.lifecycle, ManagedLifecycle::Completed); assert!(status.stop_requested); assert_eq!(status.final_outcome.as_deref(), Some(&outcome)); + subscription.changed().await.unwrap(); + assert!(std::sync::Arc::ptr_eq( + &subscription.borrow_and_update(), + &status + )); + assert!(subscription.changed().await.is_err()); + assert_eq!( + subscription.borrow().final_outcome.as_deref(), + Some(&outcome) + ); assert!(matches!( handle.update_targets(vec![]), Err(ManagedCommandError::DriverClosed) @@ -103,12 +125,14 @@ fn failed_driver_resolves_every_accepted_update_with_its_terminal_failure() { } let status = handle.status(); assert_eq!(status.lifecycle, ManagedLifecycle::Failed); + assert!(!status.stop_requested); assert_eq!(status.final_outcome.as_deref(), Some(&outcome)); assert!(matches!( handle.update_targets(vec![]), Err(ManagedCommandError::DriverClosed) )); assert!(pin!(handle.stop()).as_mut().poll(&mut context).is_ready()); + assert!(!handle.status().stop_requested); } #[test] diff --git a/crates/irtt-integration-tests/tests/managed_behavior.rs b/crates/irtt-integration-tests/tests/managed_behavior.rs index e58be0d..8f04bd2 100644 --- a/crates/irtt-integration-tests/tests/managed_behavior.rs +++ b/crates/irtt-integration-tests/tests/managed_behavior.rs @@ -238,6 +238,10 @@ async fn target_updates_preserve_identical_sessions_and_retire_removed_generatio let status = handle.status(); assert_eq!(status.lifecycle, ManagedLifecycle::Completed); assert_eq!(status.final_outcome.as_ref(), Some(&outcome)); + assert!(Arc::ptr_eq( + status.final_outcome.as_ref().unwrap(), + &outcome + )); break; } _ => {} diff --git a/crates/irtt-integration-tests/tests/managed_lifecycle.rs b/crates/irtt-integration-tests/tests/managed_lifecycle.rs index 061db3e..e7a84b7 100644 --- a/crates/irtt-integration-tests/tests/managed_lifecycle.rs +++ b/crates/irtt-integration-tests/tests/managed_lifecycle.rs @@ -238,23 +238,50 @@ fn quiescent_rejects_empty_initial_targets() { #[test] fn explicit_empty_waits_for_stop() { + let server = start_server(ServerBehavior::Echo, None); let mut config = config(ManagedPacing::Staggered); config.completion = ManagedCompletionPolicy::ExplicitStop; + config.client.request.duration = None; let (task, handle) = ManagedClient::task(config, vec![]).unwrap(); - let outcome = runtime().block_on(async move { - let mut task = Box::pin(task); - assert!( - tokio::time::timeout(Duration::from_millis(20), task.as_mut()) - .await - .is_err() - ); - assert_eq!(handle.status().lifecycle, ManagedLifecycle::Running); - let receipt = handle.stop(); - let outcome = task.await; - receipt.await; - outcome + let mut events = handle.subscribe().unwrap(); + let outcome = runtime().block_on(async { + tokio::time::timeout(Duration::from_secs(2), async { + let mut task = Box::pin(task); + assert!(poll_once(task.as_mut()).is_pending()); + assert_eq!(handle.status().lifecycle, ManagedLifecycle::Running); + let update = handle + .update_targets(vec![target("one", server.addr)]) + .unwrap(); + loop { + tokio::select! { + _ = task.as_mut() => panic!("explicit-stop task completed before stop"), + event = events.recv() => { + if matches!(event.unwrap(), ManagedEvent::Client { + event: irtt_client::ClientEvent::EchoReply { .. }, .. + }) { + break; + } + } + } + } + update.await.unwrap(); + let mut receipt = Box::pin(handle.stop()); + assert!(poll_once(receipt.as_mut()).is_pending()); + assert!(poll_once(task.as_mut()).is_pending()); + let status = handle.status(); + assert_eq!(status.lifecycle, ManagedLifecycle::Stopping); + assert!(status.stop_requested); + assert!(status.final_outcome.is_none()); + // The driver is paused during graceful cleanup: receipt readiness must + // come from durable stop observation, not eventual task completion. + assert!(poll_once(receipt.as_mut()).is_ready()); + task.await + }) + .await + .expect("stop observation and cleanup must complete") }); assert_eq!(outcome.end_reason, ManagedEndReason::StopRequested); + assert!(has_close(&server.finish())); } #[test] @@ -311,6 +338,8 @@ fn pre_poll_stop() { ) .unwrap(); let mut receipt = Box::pin(handle.stop()); + assert!(poll_once(receipt.as_mut()).is_pending()); + assert!(!handle.status().stop_requested); let mut task = Box::pin(task); let Poll::Ready(outcome) = poll_once(task.as_mut()) else { panic!("pre-stopped task did not complete") @@ -469,6 +498,10 @@ fn event_loss_independence() { assert_eq!(outcome.successful_target_outcomes, 1); let status = handle.status(); assert_eq!(status.lifecycle, ManagedLifecycle::Completed); + assert!(!status.stop_requested); + let mut receipt = Box::pin(handle.stop()); + assert!(poll_once(receipt.as_mut()).is_ready()); + assert!(Arc::ptr_eq(&status, &handle.status())); assert_eq!(status.targets.len(), 1); let target = &status.targets[0]; assert!(target.desired); @@ -511,16 +544,32 @@ fn dropping_all_handles_does_not_stop() { #[test] fn task_drop_abandonment() { - let (task, handle) = ManagedClient::task( - config(ManagedPacing::Staggered), - vec![target("one", "127.0.0.1:9".parse().unwrap())], - ) - .unwrap(); - let mut events = handle.subscribe().unwrap(); - drop(task); - assert_eq!(handle.status().lifecycle, ManagedLifecycle::Abandoned); - assert_eq!(handle.status().total_target_outcomes, 0); - assert!(matches!(events.try_recv(), Ok(ManagedEvent::Abandoned))); + // A latched but unobserved stop must not turn abandonment into completion. + for request_stop in [false, true] { + let (task, handle) = ManagedClient::task( + config(ManagedPacing::Staggered), + vec![target("one", "127.0.0.1:9".parse().unwrap())], + ) + .unwrap(); + let mut events = handle.subscribe().unwrap(); + let mut receipt = request_stop.then(|| Box::pin(handle.stop())); + if let Some(receipt) = &mut receipt { + assert!(poll_once(receipt.as_mut()).is_pending()); + } + drop(task); + assert_eq!(handle.status().lifecycle, ManagedLifecycle::Abandoned); + assert_eq!(handle.status().total_target_outcomes, 0); + assert!(!handle.status().stop_requested); + assert!(handle.status().final_outcome.is_none()); + assert!(matches!(events.try_recv(), Ok(ManagedEvent::Abandoned))); + if let Some(receipt) = &mut receipt { + assert!(poll_once(receipt.as_mut()).is_ready()); + } + let mut late_stop = Box::pin(handle.stop()); + assert!(poll_once(late_stop.as_mut()).is_ready()); + assert_eq!(handle.status().lifecycle, ManagedLifecycle::Abandoned); + assert!(!handle.status().stop_requested); + } } #[test]