boxology_runtime/
composition.rs

1//! Composition registration, validation, and transport-aware start.
2
3use std::{collections::BTreeMap, future::Future, sync::Arc, task::Poll, time::Duration};
4
5use boxology_contract::{
6    BoxHandle, BoxId, CallContext, CapabilityDescriptor, CapabilityId, Detail, ErasedCallError,
7    ErasedTarget, ExposureLevel, ImplementationDescriptor, SlotValue,
8};
9use tokio_util::sync::CancellationToken;
10
11use crate::{
12    AssemblyError, AssemblyErrors, ImportHandle, Imports, LocalBinding, RemoteImportTarget,
13    TransportBinding, TransportExposure, TransportHandle, TransportRuntime, TransportTaskTracker,
14};
15
16/// A selected target for one declared import slot.
17#[derive(Clone)]
18pub struct ImportTarget(ImportTargetKind);
19#[derive(Clone)]
20enum ImportTargetKind {
21    Local(BoxId),
22    Remote(Arc<dyn RemoteImportTarget>),
23}
24impl ImportTarget {
25    /// Selects a registered in-process provider box.
26    pub fn local(provider: BoxId) -> Self {
27        Self(ImportTargetKind::Local(provider))
28    }
29
30    /// Selects an already-configured caller-side target.
31    pub fn remote(target: Arc<dyn RemoteImportTarget>) -> Self {
32        Self(ImportTargetKind::Remote(target))
33    }
34}
35impl std::fmt::Debug for ImportTarget {
36    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
37        match &self.0 {
38            ImportTargetKind::Local(provider) => {
39                write!(formatter, "ImportTarget(Local({provider:?}))")
40            }
41            ImportTargetKind::Remote(_) => formatter.write_str("ImportTarget(Remote(<redacted>))"),
42        }
43    }
44}
45impl PartialEq for ImportTarget {
46    fn eq(&self, other: &Self) -> bool {
47        match (&self.0, &other.0) {
48            (ImportTargetKind::Local(left), ImportTargetKind::Local(right)) => left == right,
49            (ImportTargetKind::Remote(left), ImportTargetKind::Remote(right)) => {
50                Arc::ptr_eq(left, right)
51            }
52            _ => false,
53        }
54    }
55}
56impl Eq for ImportTarget {}
57
58struct CallerTargetAdapter(Arc<dyn RemoteImportTarget>);
59impl ErasedTarget for CallerTargetAdapter {
60    fn call<'a>(
61        &'a self,
62        capability: &'a CapabilityId,
63        context: CallContext,
64        input: SlotValue,
65    ) -> std::pin::Pin<Box<dyn Future<Output = Result<SlotValue, ErasedCallError>> + Send + 'a>>
66    {
67        self.0.call(capability, context, input)
68    }
69}
70struct BoxRegistration {
71    descriptor: ImplementationDescriptor,
72    handles: BTreeMap<BoxId, ImportHandle>,
73    target: Arc<dyn ErasedTarget>,
74}
75struct ImportResolution {
76    consumer: BoxId,
77    slot: BoxId,
78    target: ImportTarget,
79}
80type ExposureRegistration = (BoxId, CapabilityId, ExposureLevel, usize);
81type BindingGroup = (usize, Box<dyn ErasedTransportBinding>);
82trait ErasedTransportBinding: Send + Sync {
83    fn conform(
84        &self,
85        descriptor: &CapabilityDescriptor,
86        level: ExposureLevel,
87    ) -> Result<(), Detail>;
88    fn prepare(&self, descriptors: &[&'static CapabilityDescriptor]) -> Result<(), Detail>;
89    fn start(
90        &self,
91        exposures: Arc<[TransportExposure]>,
92        tracker: TransportTaskTracker,
93        activation: CancellationToken,
94    ) -> Result<Box<dyn TransportHandle>, Detail>;
95}
96struct OwnedTransport<B: TransportBinding> {
97    binding: Arc<B>,
98    config: Arc<B::Config>,
99}
100impl<B: TransportBinding> ErasedTransportBinding for OwnedTransport<B> {
101    fn conform(
102        &self,
103        descriptor: &CapabilityDescriptor,
104        level: ExposureLevel,
105    ) -> Result<(), Detail> {
106        self.binding.conform(descriptor, level)
107    }
108    fn prepare(&self, descriptors: &[&'static CapabilityDescriptor]) -> Result<(), Detail> {
109        self.binding.prepare(descriptors)
110    }
111    fn start(
112        &self,
113        exposures: Arc<[TransportExposure]>,
114        tracker: TransportTaskTracker,
115        activation: CancellationToken,
116    ) -> Result<Box<dyn TransportHandle>, Detail> {
117        let runtime = TransportRuntime::new(exposures, tracker, self.config.clone(), activation);
118        self.binding
119            .start(runtime)
120            .map(|handle| Box::new(handle) as Box<dyn TransportHandle>)
121    }
122}
123/// Builds and validates one composition without starting traffic.
124#[derive(Default)]
125pub struct CompositionBuilder {
126    boxes: Vec<BoxRegistration>,
127    resolutions: Vec<ImportResolution>,
128    exposures: Vec<ExposureRegistration>,
129    bindings: Vec<BindingGroup>,
130}
131
132/// One box registered in a composition draft.
133///
134/// This typed token lets application code connect and expose a box without repeating its string
135/// identity or walking its implementation descriptor.
136#[derive(Clone)]
137pub struct RegisteredBox {
138    id: BoxId,
139    capabilities: Arc<[CapabilityId]>,
140}
141
142impl RegisteredBox {
143    /// Returns the registered box identity.
144    pub fn id(&self) -> &BoxId {
145        &self.id
146    }
147}
148
149impl CompositionBuilder {
150    /// Constructs an empty composition builder.
151    pub fn new() -> Self {
152        Self::default()
153    }
154    /// Registers a box and immediately constructs its target from lazy imports.
155    pub fn add_box<T, F>(&mut self, descriptor: ImplementationDescriptor, factory: F) -> &mut Self
156    where
157        T: ErasedTarget + 'static,
158        F: FnOnce(Imports) -> T,
159    {
160        let imports = Imports::new(
161            descriptor
162                .imports()
163                .iter()
164                .map(|import| (import.slot_id().clone(), import.capabilities().to_vec())),
165        );
166        let handles = imports.cloned_handles();
167        let target: Arc<dyn ErasedTarget> = Arc::new(factory(imports));
168        self.boxes.push(BoxRegistration {
169            descriptor,
170            handles,
171            target,
172        });
173        self
174    }
175    /// Registers a box and returns a token for concise typed wiring.
176    pub fn register<T, F>(
177        &mut self,
178        descriptor: ImplementationDescriptor,
179        factory: F,
180    ) -> RegisteredBox
181    where
182        T: ErasedTarget + 'static,
183        F: FnOnce(Imports) -> T,
184    {
185        let registered = RegisteredBox {
186            id: descriptor.contract().box_id().clone(),
187            capabilities: descriptor
188                .contract()
189                .capabilities()
190                .iter()
191                .map(|capability| capability.id().clone())
192                .collect(),
193        };
194        self.add_box(descriptor, factory);
195        registered
196    }
197    /// Connects the consumer import slot named by the provider box to that local provider.
198    pub fn connect(&mut self, consumer: &RegisteredBox, provider: &RegisteredBox) -> &mut Self {
199        self.resolve_import(
200            consumer.id.clone(),
201            provider.id.clone(),
202            ImportTarget::local(provider.id.clone()),
203        )
204    }
205    /// Exposes every capability of one registered box through a transport allocation.
206    pub fn expose_all<B>(
207        &mut self,
208        provider: &RegisteredBox,
209        transport: Arc<B>,
210        level: ExposureLevel,
211    ) -> &mut Self
212    where
213        B: TransportBinding,
214    {
215        for capability in provider.capabilities.iter().cloned() {
216            self.expose(provider.id.clone(), capability, transport.clone(), level);
217        }
218        self
219    }
220    /// Creates a generated typed handle and exposes its box in-process.
221    pub fn handle<H>(&mut self, provider: &RegisteredBox) -> H
222    where
223        H: BoxHandle,
224    {
225        let local = Arc::new(LocalBinding::new());
226        self.expose_all(provider, local.clone(), ExposureLevel::CodeOnly);
227        H::from_erased(local)
228    }
229    /// Records a target selection for one consumer import slot.
230    pub fn resolve_import(
231        &mut self,
232        consumer: BoxId,
233        slot: BoxId,
234        target: ImportTarget,
235    ) -> &mut Self {
236        self.resolutions.push(ImportResolution {
237            consumer,
238            slot,
239            target,
240        });
241        self
242    }
243    /// Exposes one exact provider capability through a configured transport allocation.
244    pub fn expose<B>(
245        &mut self,
246        provider: BoxId,
247        capability: CapabilityId,
248        transport: Arc<B>,
249        level: ExposureLevel,
250    ) -> &mut Self
251    where
252        B: TransportBinding,
253    {
254        let allocation_key = Arc::as_ptr(&transport).cast::<()>().addr();
255        let binding_group = self
256            .bindings
257            .iter()
258            .position(|group| group.0 == allocation_key)
259            .unwrap_or_else(|| {
260                let index = self.bindings.len();
261                let config = transport.config();
262                self.bindings.push((
263                    allocation_key,
264                    Box::new(OwnedTransport {
265                        binding: transport,
266                        config,
267                    }),
268                ));
269                index
270            });
271        self.exposures
272            .push((provider, capability, level, binding_group));
273        self
274    }
275    /// Reports every assembly failure without sealing imports or starting traffic.
276    pub fn validate(&self) -> Result<(), AssemblyErrors> {
277        let mut errors = Vec::new();
278        let mut registrations = BTreeMap::new();
279        for (index, registration) in self.boxes.iter().enumerate() {
280            let box_id = registration.descriptor.contract().box_id();
281            if registrations.contains_key(box_id) {
282                errors.push(AssemblyError::DuplicateBox {
283                    box_id: box_id.clone(),
284                });
285            } else {
286                registrations.insert(box_id, index);
287            }
288        }
289        let mut states = BTreeMap::new();
290        for resolution in &self.resolutions {
291            let Some(&consumer_index) = registrations.get(&resolution.consumer) else {
292                let consumer = resolution.consumer.clone();
293                errors.push(AssemblyError::UnknownImportConsumer { consumer });
294                continue;
295            };
296            let consumer = &self.boxes[consumer_index];
297            if !consumer
298                .descriptor
299                .imports()
300                .iter()
301                .any(|import| import.slot_id() == &resolution.slot)
302            {
303                let consumer = resolution.consumer.clone();
304                let slot = resolution.slot.clone();
305                errors.push(AssemblyError::UnknownImportSlot { consumer, slot });
306                continue;
307            }
308            let key = (&resolution.consumer, &resolution.slot);
309            if states.insert(key, None).is_some() {
310                let consumer = resolution.consumer.clone();
311                let slot = resolution.slot.clone();
312                errors.push(AssemblyError::DuplicateImportResolution { consumer, slot });
313                continue;
314            }
315            match &resolution.target.0 {
316                ImportTargetKind::Local(target) => {
317                    if let Some(&provider_index) = registrations.get(target) {
318                        states.insert(key, Some(provider_index));
319                    } else {
320                        errors.push(AssemblyError::UnknownImportTarget {
321                            consumer: resolution.consumer.clone(),
322                            slot: resolution.slot.clone(),
323                            target: target.clone(),
324                        });
325                    }
326                }
327                ImportTargetKind::Remote(target) => {
328                    let import = consumer
329                        .descriptor
330                        .imports()
331                        .iter()
332                        .find(|import| import.slot_id() == &resolution.slot)
333                        .unwrap();
334                    for capability in import.capabilities() {
335                        if !target.supports_capability(capability) {
336                            errors.push(AssemblyError::MissingImportedCapability {
337                                consumer: resolution.consumer.clone(),
338                                slot: resolution.slot.clone(),
339                                capability: capability.clone(),
340                            });
341                        }
342                    }
343                }
344            }
345        }
346        for (index, registration) in self.boxes.iter().enumerate() {
347            let consumer = registration.descriptor.contract().box_id();
348            if registrations.get(consumer) != Some(&index) {
349                continue;
350            }
351            for import in registration.descriptor.imports() {
352                match states.get(&(consumer, import.slot_id())) {
353                    None => {
354                        let consumer = consumer.clone();
355                        let slot = import.slot_id().clone();
356                        errors.push(AssemblyError::MissingImportResolution { consumer, slot });
357                    }
358                    Some(Some(provider_index)) => {
359                        let contract = self.boxes[*provider_index].descriptor.contract();
360                        let provided = contract.capabilities();
361                        for capability in import.capabilities() {
362                            if !provided.iter().any(|known| known.id() == capability) {
363                                errors.push(AssemblyError::MissingImportedCapability {
364                                    consumer: consumer.clone(),
365                                    slot: import.slot_id().clone(),
366                                    capability: capability.clone(),
367                                });
368                            }
369                        }
370                    }
371                    Some(None) => {}
372                }
373            }
374        }
375        for (provider, capability, level, binding_group) in &self.exposures {
376            let Some(&provider_index) = registrations.get(provider) else {
377                errors.push(AssemblyError::UnknownExposureProvider {
378                    provider: provider.clone(),
379                });
380                continue;
381            };
382            let Some(descriptor) = self.boxes[provider_index]
383                .descriptor
384                .contract()
385                .capabilities()
386                .iter()
387                .find(|descriptor| descriptor.id() == capability)
388            else {
389                errors.push(AssemblyError::UnknownExposedCapability {
390                    provider: provider.clone(),
391                    capability: capability.clone(),
392                });
393                continue;
394            };
395            if *level > descriptor.max_exposure() {
396                errors.push(AssemblyError::ExposureExceedsMaximum {
397                    capability: capability.clone(),
398                    requested: *level,
399                    maximum: descriptor.max_exposure(),
400                });
401            } else if let Err(detail) = self.bindings[*binding_group].1.conform(descriptor, *level)
402            {
403                errors.push(AssemblyError::TransportConformanceFailed {
404                    capability: capability.clone(),
405                    detail,
406                });
407            }
408        }
409        AssemblyErrors::from_errors(errors).map_or(Ok(()), Err)
410    }
411
412    /// Validates, starts every transport closed, then atomically commits traffic.
413    pub fn start(self) -> Result<Composition, AssemblyErrors> {
414        self.validate()?;
415        let mut grouped = vec![Vec::new(); self.bindings.len()];
416        for (provider, capability, level, binding_group) in &self.exposures {
417            let registration = self
418                .boxes
419                .iter()
420                .find(|registration| registration.descriptor.contract().box_id() == provider)
421                .unwrap();
422            let descriptor = registration
423                .descriptor
424                .contract()
425                .capabilities()
426                .iter()
427                .find(|descriptor| descriptor.id() == capability)
428                .unwrap();
429            grouped[*binding_group].push(TransportExposure::new(
430                descriptor,
431                *level,
432                registration.target.clone(),
433            ));
434        }
435        let grouped: Vec<Arc<[TransportExposure]>> = grouped.into_iter().map(Arc::from).collect();
436        for (binding, exposures) in self.bindings.iter().zip(&grouped) {
437            let descriptors: Vec<_> = exposures
438                .iter()
439                .map(TransportExposure::descriptor)
440                .collect();
441            if let Err(detail) = binding.1.prepare(&descriptors) {
442                return Err(single_error(AssemblyError::TransportPrepareFailed {
443                    detail,
444                }));
445            }
446        }
447        let tracker = TransportTaskTracker::new();
448        let activation = CancellationToken::new();
449        let mut handles = Vec::with_capacity(self.bindings.len());
450        for (binding, exposures) in self.bindings.iter().zip(&grouped) {
451            match binding
452                .1
453                .start(exposures.clone(), tracker.clone(), activation.clone())
454            {
455                Ok(handle) => handles.push(handle),
456                Err(detail) => {
457                    for handle in handles.iter().rev() {
458                        handle.stop_intake();
459                    }
460                    for handle in handles.iter().rev() {
461                        handle.cancel_tasks();
462                    }
463                    for handle in handles.iter().rev() {
464                        handle.abort_tasks();
465                    }
466                    return Err(single_error(AssemblyError::TransportStartFailed { detail }));
467                }
468            }
469        }
470        for registration in &self.boxes {
471            let consumer = registration.descriptor.contract().box_id();
472            for import in registration.descriptor.imports() {
473                let resolution = self
474                    .resolutions
475                    .iter()
476                    .find(|resolution| {
477                        &resolution.consumer == consumer && resolution.slot == *import.slot_id()
478                    })
479                    .unwrap();
480                let target: Arc<dyn ErasedTarget> = match &resolution.target.0 {
481                    ImportTargetKind::Local(provider) => self
482                        .boxes
483                        .iter()
484                        .find(|registration| {
485                            registration.descriptor.contract().box_id() == provider
486                        })
487                        .unwrap()
488                        .target
489                        .clone(),
490                    ImportTargetKind::Remote(target) => {
491                        Arc::new(CallerTargetAdapter(target.clone()))
492                    }
493                };
494                let handle = registration.handles.get(import.slot_id()).unwrap();
495                assert!(
496                    handle.seal(target).is_ok(),
497                    "import handle was already sealed"
498                );
499            }
500        }
501        activation.cancel();
502        Ok(Composition {
503            _boxes: self.boxes,
504            _bindings: self.bindings.into_iter().map(|group| group.1).collect(),
505            _exposures: grouped,
506            _handles: handles,
507            _tracker: tracker,
508            _activation: activation,
509        })
510    }
511}
512fn single_error(error: AssemblyError) -> AssemblyErrors {
513    AssemblyErrors::from_errors(vec![error]).unwrap()
514}
515/// A successfully validated and activated composition.
516pub struct Composition {
517    _boxes: Vec<BoxRegistration>,
518    _bindings: Vec<Box<dyn ErasedTransportBinding>>,
519    _exposures: Vec<Arc<[TransportExposure]>>,
520    _handles: Vec<Box<dyn TransportHandle>>,
521    _tracker: TransportTaskTracker,
522    _activation: CancellationToken,
523}
524
525impl Composition {
526    /// Stops transport intake and drains, cancels, or aborts all tracked work.
527    pub async fn shutdown(mut self, drain_timeout: Duration) -> Result<(), ErasedCallError> {
528        for handle in self._handles.iter().rev() {
529            handle.stop_intake();
530        }
531        self._tracker.close();
532        if completes_within(&self._tracker, drain_timeout).await {
533            return Ok(());
534        }
535        for handle in self._handles.iter().rev() {
536            handle.cancel_tasks();
537        }
538        if completes_within(&self._tracker, drain_timeout).await {
539            return Ok(());
540        }
541        for handle in self._handles.iter().rev() {
542            handle.abort_tasks();
543        }
544        let handles = std::mem::take(&mut self._handles);
545        let mut first_failure = None;
546        for handle in handles.into_iter().rev() {
547            if let Err(detail) = handle.join_tasks().await
548                && first_failure.is_none()
549            {
550                first_failure = Some(detail);
551            }
552        }
553        let result = first_failure.map_or(Ok(()), |detail| Err(ErasedCallError::Internal(detail)));
554        drop(self);
555        result
556    }
557}
558
559async fn completes_within(tracker: &TransportTaskTracker, duration: Duration) -> bool {
560    let mut completion = Box::pin(tracker.wait());
561    let mut timeout = Box::pin(tokio::time::sleep(duration));
562    std::future::poll_fn(|context| {
563        if completion.as_mut().poll(context).is_ready() {
564            return Poll::Ready(true);
565        }
566        timeout.as_mut().poll(context).map(|()| false)
567    })
568    .await
569}
570
571#[cfg(test)]
572mod shutdown_tests {
573    use super::*;
574    use crate::TransportJoinFuture;
575    use std::{future::pending, sync::Mutex};
576    use tokio::task::JoinHandle;
577    use tokio_util::task::task_tracker::TaskTrackerToken;
578
579    #[derive(Clone, Copy)]
580    enum Exit {
581        CancelAfter(Duration),
582        Never,
583    }
584
585    struct LifecycleHandle {
586        id: u8,
587        trace: Arc<Mutex<Vec<String>>>,
588        cancel: CancellationToken,
589        tokens: Mutex<Vec<TaskTrackerToken>>,
590        tasks: Vec<JoinHandle<Result<(), Detail>>>,
591    }
592
593    impl LifecycleHandle {
594        fn record(&self, phase: &str) {
595            self.trace
596                .lock()
597                .unwrap()
598                .push(format!("{phase}{}", self.id));
599        }
600    }
601
602    impl TransportHandle for LifecycleHandle {
603        fn stop_intake(&self) {
604            self.record("stop");
605        }
606
607        fn cancel_tasks(&self) {
608            self.record("cancel");
609            self.cancel.cancel();
610            self.tokens.lock().unwrap().clear();
611        }
612
613        fn abort_tasks(&self) {
614            self.record("abort");
615            for task in &self.tasks {
616                task.abort();
617            }
618        }
619
620        fn join_tasks(self: Box<Self>) -> TransportJoinFuture {
621            Box::pin(async move {
622                let mut first_failure = None;
623                for (index, task) in self.tasks.into_iter().enumerate() {
624                    let event = format!("join{}.{index}", self.id);
625                    self.trace.lock().unwrap().push(event.clone());
626                    let result = match task.await {
627                        Ok(result) => result,
628                        Err(_) => Err(Detail::new(event)),
629                    };
630                    if let Err(detail) = result
631                        && first_failure.is_none()
632                    {
633                        first_failure = Some(detail);
634                    }
635                }
636                first_failure.map_or(Ok(()), Err)
637            })
638        }
639    }
640
641    fn handle(
642        id: u8,
643        tracker: &TransportTaskTracker,
644        trace: &Arc<Mutex<Vec<String>>>,
645        exit: Option<Exit>,
646        task_count: usize,
647    ) -> LifecycleHandle {
648        let cancel = CancellationToken::new();
649        let tasks = (0..task_count)
650            .map(|_| {
651                let cancel = cancel.clone();
652                tracker.spawn(async move {
653                    match exit.expect("task requires an exit mode") {
654                        Exit::CancelAfter(delay) => {
655                            cancel.cancelled().await;
656                            tokio::time::sleep(delay).await;
657                            Ok(())
658                        }
659                        Exit::Never => pending().await,
660                    }
661                })
662            })
663            .collect();
664        LifecycleHandle {
665            id,
666            trace: trace.clone(),
667            cancel,
668            tokens: Mutex::new(Vec::new()),
669            tasks,
670        }
671    }
672
673    fn token_handle(
674        id: u8,
675        tracker: &TransportTaskTracker,
676        trace: &Arc<Mutex<Vec<String>>>,
677    ) -> LifecycleHandle {
678        let mut handle = handle(id, tracker, trace, None, 0);
679        handle.tokens = Mutex::new(vec![tracker.token()]);
680        handle
681    }
682
683    fn composition(tracker: &TransportTaskTracker, handles: Vec<LifecycleHandle>) -> Composition {
684        Composition {
685            _boxes: Vec::new(),
686            _bindings: Vec::new(),
687            _exposures: Vec::new(),
688            _handles: handles
689                .into_iter()
690                .map(|handle| Box::new(handle) as Box<dyn TransportHandle>)
691                .collect(),
692            _tracker: tracker.clone(),
693            _activation: CancellationToken::new(),
694        }
695    }
696
697    fn events(trace: &Arc<Mutex<Vec<String>>>) -> Vec<String> {
698        trace.lock().unwrap().clone()
699    }
700
701    fn run_paused(future: impl Future<Output = ()>) {
702        let runtime = tokio::runtime::Builder::new_current_thread()
703            .enable_time()
704            .build()
705            .unwrap();
706        runtime.block_on(async {
707            tokio::time::pause();
708            future.await;
709        });
710    }
711
712    #[test]
713    fn immediate_drain_closes_tracker_stops_in_reverse_and_wins_zero_tie() {
714        run_paused(async {
715            let tracker = TransportTaskTracker::new();
716            let trace = Arc::new(Mutex::new(Vec::new()));
717            let handles = (1..=2)
718                .map(|id| handle(id, &tracker, &trace, None, 0))
719                .collect();
720
721            assert_eq!(
722                composition(&tracker, handles)
723                    .shutdown(Duration::ZERO)
724                    .await,
725                Ok(())
726            );
727            assert!(tracker.is_closed() && tracker.is_empty());
728            assert_eq!(events(&trace), ["stop2", "stop1"]);
729        });
730    }
731
732    #[test]
733    fn drain_timeout_cancels_in_reverse_then_uses_a_fresh_grace_window() {
734        run_paused(async {
735            let tracker = TransportTaskTracker::new();
736            let trace = Arc::new(Mutex::new(Vec::new()));
737            let handles = (1..=2)
738                .map(|id| {
739                    handle(
740                        id,
741                        &tracker,
742                        &trace,
743                        Some(Exit::CancelAfter(Duration::from_secs(4))),
744                        1,
745                    )
746                })
747                .collect();
748
749            assert_eq!(
750                composition(&tracker, handles)
751                    .shutdown(Duration::from_secs(5))
752                    .await,
753                Ok(())
754            );
755            assert!(tracker.is_empty());
756            assert_eq!(events(&trace), ["stop2", "stop1", "cancel2", "cancel1"]);
757        });
758    }
759
760    #[test]
761    fn grace_completion_wins_a_same_poll_zero_timeout_tie() {
762        run_paused(async {
763            let tracker = TransportTaskTracker::new();
764            let trace = Arc::new(Mutex::new(Vec::new()));
765            let handles = (1..=2)
766                .map(|id| token_handle(id, &tracker, &trace))
767                .collect();
768
769            assert_eq!(
770                composition(&tracker, handles)
771                    .shutdown(Duration::ZERO)
772                    .await,
773                Ok(())
774            );
775            assert!(tracker.is_empty());
776            assert_eq!(events(&trace), ["stop2", "stop1", "cancel2", "cancel1"]);
777        });
778    }
779
780    #[test]
781    fn forced_cleanup_aborts_globally_then_joins_every_task_in_reverse() {
782        run_paused(async {
783            let tracker = TransportTaskTracker::new();
784            let trace = Arc::new(Mutex::new(Vec::new()));
785            let handles = [(1, 1), (2, 1), (3, 2)]
786                .into_iter()
787                .map(|(id, count)| handle(id, &tracker, &trace, Some(Exit::Never), count))
788                .collect();
789
790            let error = composition(&tracker, handles)
791                .shutdown(Duration::ZERO)
792                .await
793                .unwrap_err();
794            assert_eq!(error, ErasedCallError::Internal(Detail::new("join3.0")));
795            assert!(tracker.is_closed() && tracker.is_empty());
796            assert_eq!(
797                events(&trace),
798                [
799                    "stop3", "stop2", "stop1", "cancel3", "cancel2", "cancel1", "abort3", "abort2",
800                    "abort1", "join3.0", "join3.1", "join2.0", "join1.0",
801                ]
802            );
803        });
804    }
805}
boxology_runtime/composition.rs

boxology_runtime/
composition.rs
