1use crate::channel::oneshot;
100use crate::combinator::{Either, Select};
101use crate::cx::{Cx, cap};
102use crate::record::AdmissionError;
103use crate::record::task::TaskState;
104use crate::runtime::resource_monitor::RegionPriority;
105use crate::runtime::task_handle::{JoinError, TaskHandle};
106use crate::runtime::{RegionCreateError, RuntimeState, SpawnError, StoredTask};
107use crate::tracing_compat::{debug, debug_span};
108use crate::types::{
109 Budget, CancelReason, CapabilityBudget, CapabilityBudgetRequirements, Outcome, PanicPayload,
110 Policy, RegionId, TaskId,
111};
112use std::future::Future;
113use std::marker::PhantomData;
114use std::pin::Pin;
115use std::sync::Arc;
116use std::task::{Context, Poll};
117
118pub struct Scope<'r, P: Policy = crate::types::policy::FailFast> {
126 pub(crate) region: RegionId,
128 pub(crate) budget: Budget,
130 pub(crate) capability_budget: CapabilityBudget,
132 pub(crate) _policy: PhantomData<&'r P>,
134}
135
136#[derive(Clone, Copy)]
137struct ChildRegionAdmission {
138 budget: Budget,
139 capability_budget: CapabilityBudget,
140 requirements: CapabilityBudgetRequirements,
141 priority: RegionPriority,
142}
143
144#[pin_project::pin_project]
145pub(crate) struct CatchUnwind<F> {
146 #[pin]
147 pub(crate) inner: F,
148}
149
150impl<F: Future> Future for CatchUnwind<F> {
151 type Output = std::thread::Result<F::Output>;
152
153 fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
154 let mut this = self.project();
155 let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
156 this.inner.as_mut().poll(cx)
157 }));
158 match result {
159 Ok(Poll::Pending) => Poll::Pending,
160 Ok(Poll::Ready(v)) => Poll::Ready(Ok(v)),
161 Err(payload) => Poll::Ready(Err(payload)),
162 }
163 }
164}
165
166pub(crate) fn payload_to_string(payload: &Box<dyn std::any::Any + Send>) -> String {
167 payload
168 .downcast_ref::<&str>()
169 .map(ToString::to_string)
170 .or_else(|| payload.downcast_ref::<String>().cloned())
171 .unwrap_or_else(|| "unknown panic".to_string())
172}
173
174struct RegionRunner<'a, Fut> {
175 fut: Pin<&'a mut CatchUnwind<Fut>>,
176 state: Option<&'a mut RuntimeState>,
177 child_region: RegionId,
178}
179
180impl<'a, Fut: Future> Future for RegionRunner<'a, Fut> {
181 type Output = (std::thread::Result<Fut::Output>, &'a mut RuntimeState);
182
183 fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
184 let this = self.get_mut();
185 match this.fut.as_mut().poll(cx) {
186 Poll::Ready(res) => {
187 let state = this.state.take().expect("polled after ready");
188 Poll::Ready((res, state))
189 }
190 Poll::Pending => Poll::Pending,
191 }
192 }
193}
194
195impl<Fut> Drop for RegionRunner<'_, Fut> {
196 fn drop(&mut self) {
197 if let Some(state) = self.state.take() {
198 let reason = CancelReason::fail_fast().with_region(self.child_region);
199 let _ = state.cancel_request(self.child_region, &reason, None);
200 if let Some(region) = state.region(self.child_region) {
201 region.begin_close(None);
202 }
203 state.advance_region_state(self.child_region);
204 }
205 }
206}
207
208struct RegionCloseFuture {
209 state: Arc<parking_lot::Mutex<crate::record::region::RegionCloseState>>,
210}
211
212impl Future for RegionCloseFuture {
213 type Output = ();
214
215 fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
216 let mut state = self.state.lock();
217 if state.closed {
218 Poll::Ready(())
219 } else {
220 if !state.waiters.iter().any(|w| w.will_wake(cx.waker())) {
221 state.waiters.push(cx.waker().clone());
222 }
223 Poll::Pending
224 }
225 }
226}
227
228impl<P: Policy> Scope<'_, P> {
229 #[must_use]
231 #[allow(dead_code)]
232 #[cfg_attr(feature = "test-internals", visibility::make(pub))]
233 pub(crate) fn new(region: RegionId, budget: Budget) -> Self {
234 Self::new_with_capability_budget(region, budget, CapabilityBudget::UNSPECIFIED)
235 }
236
237 #[must_use]
239 #[allow(dead_code)]
240 #[cfg_attr(feature = "test-internals", visibility::make(pub))]
241 pub(crate) fn new_with_capability_budget(
242 region: RegionId,
243 budget: Budget,
244 capability_budget: CapabilityBudget,
245 ) -> Self {
246 Self {
247 region,
248 budget,
249 capability_budget,
250 _policy: PhantomData,
251 }
252 }
253
254 #[must_use]
256 pub fn region_id(&self) -> RegionId {
257 self.region
258 }
259
260 #[must_use]
262 pub fn budget(&self) -> Budget {
263 self.budget
264 }
265
266 #[must_use]
268 pub fn capability_budget(&self) -> CapabilityBudget {
269 self.capability_budget
270 }
271
272 pub fn spawn<F, Fut, Caps>(
382 &self,
383 state: &mut RuntimeState,
384 cx: &Cx<Caps>,
385 f: F,
386 ) -> Result<(TaskHandle<Fut::Output>, StoredTask), SpawnError>
387 where
388 Caps: cap::HasSpawn + Send + Sync + 'static,
389 F: FnOnce(Cx<Caps>) -> Fut + Send + 'static,
390 Fut: Future + Send + 'static,
391 Fut::Output: Send + 'static,
392 {
393 let (tx, rx) = oneshot::channel::<Result<Fut::Output, JoinError>>();
395
396 let task_id = self.create_task_record(state)?;
398
399 let _span = debug_span!(
401 "task_spawn",
402 task_id = ?task_id,
403 region_id = ?self.region,
404 initial_state = "Created",
405 budget_deadline = ?self.budget.deadline,
406 budget_poll_quota = self.budget.poll_quota,
407 budget_cost_quota = ?self.budget.cost_quota,
408 budget_priority = self.budget.priority,
409 budget_source = "scope"
410 )
411 .entered();
412 debug!(
413 task_id = ?task_id,
414 region_id = ?self.region,
415 initial_state = "Created",
416 budget_deadline = ?self.budget.deadline,
417 budget_poll_quota = self.budget.poll_quota,
418 budget_cost_quota = ?self.budget.cost_quota,
419 budget_priority = self.budget.priority,
420 budget_source = "scope",
421 "task spawned"
422 );
423
424 let (child_cx, child_cx_full) = self.build_child_task_cx(state, cx, task_id);
425
426 let handle = TaskHandle::new(task_id, rx, Arc::downgrade(&child_cx.inner));
428
429 if let Some(record) = state.task_mut(task_id) {
432 record.set_cx_inner(child_cx.inner.clone());
433 record.set_cx(child_cx_full.clone());
434 }
435
436 let future = {
452 struct TaskCreationGuard<'a> {
453 state: &'a mut RuntimeState,
454 task_id: TaskId,
455 region_id: RegionId,
456 committed: bool,
457 }
458
459 impl Drop for TaskCreationGuard<'_> {
460 fn drop(&mut self) {
461 if !self.committed {
462 if let Some(region) = self.state.region_mut(self.region_id) {
464 region.remove_task(self.task_id);
465 }
466 self.state.recycle_task(self.task_id);
467 }
468 }
469 }
470
471 let mut guard = TaskCreationGuard {
472 state,
473 task_id,
474 region_id: self.region,
475 committed: false,
476 };
477
478 let fut = f(child_cx);
479 guard.committed = true;
480 fut
481 };
482
483 let wrapped = async move {
487 let result_result = CatchUnwind { inner: future }.await;
488 match result_result {
489 Ok(result) => {
490 let _ = tx.send_blocking(Ok(result));
491 crate::types::Outcome::Ok(())
492 }
493 Err(payload) => {
494 let msg = payload_to_string(&payload);
495 let panic_payload = PanicPayload::new(msg);
496 let _ = tx.send_blocking(Err(JoinError::Panicked(panic_payload.clone())));
497 crate::types::Outcome::Panicked(panic_payload)
498 }
499 }
500 };
501
502 let stored = StoredTask::new_with_id(wrapped, task_id);
504
505 Ok((handle, stored))
506 }
507
508 #[inline]
531 pub fn spawn_task<F, Fut, Caps>(
532 &self,
533 state: &mut RuntimeState,
534 cx: &Cx<Caps>,
535 f: F,
536 ) -> Result<(TaskHandle<Fut::Output>, StoredTask), SpawnError>
537 where
538 Caps: cap::HasSpawn + Send + Sync + 'static,
539 F: FnOnce(Cx<Caps>) -> Fut + Send + 'static,
540 Fut: Future + Send + 'static,
541 Fut::Output: Send + 'static,
542 {
543 self.spawn(state, cx, f)
544 }
545
546 pub fn spawn_registered<F, Fut, Caps>(
573 &self,
574 state: &mut RuntimeState,
575 cx: &Cx<Caps>,
576 f: F,
577 ) -> Result<TaskHandle<Fut::Output>, SpawnError>
578 where
579 Caps: cap::HasSpawn + Send + Sync + 'static,
580 F: FnOnce(Cx<Caps>) -> Fut + Send + 'static,
581 Fut: Future + Send + 'static,
582 Fut::Output: Send + 'static,
583 {
584 let (handle, stored) = self.spawn(state, cx, f)?;
585 state.store_spawned_task(handle.task_id(), stored);
586 Ok(handle)
587 }
588
589 #[allow(clippy::too_many_lines)]
629 pub fn spawn_local<F, Fut, Caps>(
630 &self,
631 state: &mut RuntimeState,
632 cx: &Cx<Caps>,
633 f: F,
634 ) -> Result<TaskHandle<Fut::Output>, SpawnError>
635 where
636 Caps: cap::HasSpawn + Send + Sync + 'static,
637 F: FnOnce(Cx<Caps>) -> Fut + 'static,
638 Fut: Future + 'static,
639 Fut::Output: Send + 'static,
640 {
641 use crate::runtime::stored_task::LocalStoredTask;
642 use crate::runtime::task_handle::JoinError;
643
644 let (result_tx, rx) = oneshot::channel::<Result<Fut::Output, JoinError>>();
646
647 let task_id = self.create_task_record(state)?;
649
650 let _span = debug_span!(
652 "task_spawn",
653 task_id = ?task_id,
654 region_id = ?self.region,
655 initial_state = "Created",
656 budget_deadline = ?self.budget.deadline,
657 budget_poll_quota = self.budget.poll_quota,
658 budget_cost_quota = ?self.budget.cost_quota,
659 budget_priority = self.budget.priority,
660 budget_source = "scope_local"
661 )
662 .entered();
663 debug!(
664 task_id = ?task_id,
665 region_id = ?self.region,
666 initial_state = "Created",
667 budget_deadline = ?self.budget.deadline,
668 budget_poll_quota = self.budget.poll_quota,
669 budget_cost_quota = ?self.budget.cost_quota,
670 budget_priority = self.budget.priority,
671 budget_source = "scope_local",
672 "local task spawned"
673 );
674
675 let (child_cx, child_cx_full) = self.build_child_task_cx(state, cx, task_id);
676
677 let handle = TaskHandle::new(task_id, rx, Arc::downgrade(&child_cx.inner));
679
680 if let Some(record) = state.task_mut(task_id) {
682 record.set_cx_inner(child_cx.inner.clone());
683 record.set_cx(child_cx_full.clone());
684 }
685
686 let future = {
693 struct TaskCreationGuard<'a> {
694 state: &'a mut RuntimeState,
695 task_id: TaskId,
696 region_id: RegionId,
697 committed: bool,
698 }
699
700 impl Drop for TaskCreationGuard<'_> {
701 fn drop(&mut self) {
702 if !self.committed {
703 if let Some(region) = self.state.region_mut(self.region_id) {
705 region.remove_task(self.task_id);
706 }
707 self.state.recycle_task(self.task_id);
708 }
709 }
710 }
711
712 let mut guard = TaskCreationGuard {
713 state,
714 task_id,
715 region_id: self.region,
716 committed: false,
717 };
718
719 let fut = f(child_cx);
720 guard.committed = true;
721 fut
722 };
723
724 let wrapped = async move {
726 let result_result = CatchUnwind { inner: future }.await;
727 match result_result {
728 Ok(result) => {
729 let _ = result_tx.send_blocking(Ok(result));
730 crate::types::Outcome::Ok(())
731 }
732 Err(payload) => {
733 let msg = payload_to_string(&payload);
734 let panic_payload = PanicPayload::new(msg);
735 let _ =
736 result_tx.send_blocking(Err(JoinError::Panicked(panic_payload.clone())));
737 crate::types::Outcome::Panicked(panic_payload)
738 }
739 }
740 };
741
742 let stored = LocalStoredTask::new_with_id(wrapped, task_id);
744
745 crate::runtime::local::store_local_task(task_id, stored);
747
748 if let Some(record) = state.task_mut(task_id) {
752 if let Some(worker_id) = crate::runtime::scheduler::three_lane::current_worker_id() {
753 record.pin_to_worker(worker_id);
754 } else {
755 record.mark_local();
756 }
757 record.wake_state.notify();
758 }
759
760 let scheduled = crate::runtime::scheduler::three_lane::schedule_local_task(task_id);
763
764 if scheduled {
765 return Ok(handle);
766 }
767
768 let _ = crate::runtime::local::remove_local_task(task_id);
770 if let Some(region) = state.region(self.region) {
771 region.remove_task(task_id);
772 }
773 state.recycle_task(task_id);
774 Err(SpawnError::LocalSchedulerUnavailable)
775 }
776
777 pub fn spawn_blocking<F, R, Caps>(
811 &self,
812 state: &mut RuntimeState,
813 cx: &Cx<Caps>, f: F,
815 ) -> Result<(TaskHandle<R>, StoredTask), SpawnError>
816 where
817 Caps: cap::HasSpawn + Send + Sync + 'static,
818 F: FnOnce(Cx<Caps>) -> R + Send + 'static,
819 R: Send + 'static,
820 {
821 let (tx, rx) = oneshot::channel::<Result<R, JoinError>>();
823
824 let task_id = self.create_task_record(state)?;
826
827 debug!(
829 task_id = ?task_id,
830 region_id = ?self.region,
831 initial_state = "Created",
832 poll_quota = self.budget.poll_quota,
833 spawn_kind = "blocking",
834 "blocking task spawned"
835 );
836
837 let (child_cx, child_cx_full) = self.build_child_task_cx(state, cx, task_id);
838
839 let handle = TaskHandle::new(task_id, rx, Arc::downgrade(&child_cx.inner));
841
842 if let Some(record) = state.task_mut(task_id) {
844 record.set_cx_inner(child_cx.inner.clone());
845 record.set_cx(child_cx_full.clone());
846 }
847
848 let wrapped = async move {
855 let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| f(child_cx)));
858 match result {
859 Ok(res) => {
860 let _ = tx.send_blocking(Ok(res));
861 crate::types::Outcome::Ok(())
862 }
863 Err(payload) => {
864 let msg = payload_to_string(&payload);
865 let panic_payload = PanicPayload::new(msg);
866 let _ = tx.send_blocking(Err(JoinError::Panicked(panic_payload.clone())));
867 crate::types::Outcome::Panicked(panic_payload)
868 }
869 }
870 };
871
872 let stored = StoredTask::new_with_id(wrapped, task_id);
873
874 Ok((handle, stored))
875 }
876
877 pub async fn region<P2, F, Fut, T, Caps>(
894 &self,
895 state: &mut RuntimeState,
896 cx: &Cx<Caps>,
897 policy: P2,
898 f: F,
899 ) -> Result<Outcome<T, P2::Error>, RegionCreateError>
900 where
901 P2: Policy,
902 F: FnOnce(Scope<'_, P2>, &mut RuntimeState) -> Fut,
903 Fut: Future<Output = Outcome<T, P2::Error>>,
904 {
905 self.region_with_budget(state, cx, self.budget, policy, f)
906 .await
907 }
908
909 pub async fn region_with_budget<P2, F, Fut, T, Caps>(
914 &self,
915 state: &mut RuntimeState,
916 _cx: &Cx<Caps>,
917 budget: Budget,
918 _policy: P2,
919 f: F,
920 ) -> Result<Outcome<T, P2::Error>, RegionCreateError>
921 where
922 P2: Policy,
923 F: FnOnce(Scope<'_, P2>, &mut RuntimeState) -> Fut,
924 Fut: Future<Output = Outcome<T, P2::Error>>,
925 {
926 self.region_with_budget_and_priority(state, _cx, budget, RegionPriority::Normal, _policy, f)
927 .await
928 }
929
930 pub async fn region_with_priority<P2, F, Fut, T, Caps>(
935 &self,
936 state: &mut RuntimeState,
937 _cx: &Cx<Caps>,
938 priority: RegionPriority,
939 _policy: P2,
940 f: F,
941 ) -> Result<Outcome<T, P2::Error>, RegionCreateError>
942 where
943 P2: Policy,
944 F: FnOnce(Scope<'_, P2>, &mut RuntimeState) -> Fut,
945 Fut: Future<Output = Outcome<T, P2::Error>>,
946 {
947 self.region_with_budget_and_priority(state, _cx, self.budget, priority, _policy, f)
948 .await
949 }
950
951 pub async fn region_with_budget_and_priority<P2, F, Fut, T, Caps>(
953 &self,
954 state: &mut RuntimeState,
955 _cx: &Cx<Caps>,
956 budget: Budget,
957 priority: RegionPriority,
958 _policy: P2,
959 f: F,
960 ) -> Result<Outcome<T, P2::Error>, RegionCreateError>
961 where
962 P2: Policy,
963 F: FnOnce(Scope<'_, P2>, &mut RuntimeState) -> Fut,
964 Fut: Future<Output = Outcome<T, P2::Error>>,
965 {
966 self.region_with_child_admission(
967 state,
968 _cx,
969 ChildRegionAdmission {
970 budget,
971 capability_budget: CapabilityBudget::UNSPECIFIED,
972 requirements: CapabilityBudgetRequirements::NONE,
973 priority,
974 },
975 _policy,
976 f,
977 )
978 .await
979 }
980
981 pub async fn region_with_budget_and_capability_budget<P2, F, Fut, T, Caps>(
987 &self,
988 state: &mut RuntimeState,
989 _cx: &Cx<Caps>,
990 budget: Budget,
991 capability_budget: CapabilityBudget,
992 requirements: CapabilityBudgetRequirements,
993 _policy: P2,
994 f: F,
995 ) -> Result<Outcome<T, P2::Error>, RegionCreateError>
996 where
997 P2: Policy,
998 F: FnOnce(Scope<'_, P2>, &mut RuntimeState) -> Fut,
999 Fut: Future<Output = Outcome<T, P2::Error>>,
1000 {
1001 self.region_with_child_admission(
1002 state,
1003 _cx,
1004 ChildRegionAdmission {
1005 budget,
1006 capability_budget,
1007 requirements,
1008 priority: RegionPriority::Normal,
1009 },
1010 _policy,
1011 f,
1012 )
1013 .await
1014 }
1015
1016 async fn region_with_child_admission<P2, F, Fut, T, Caps>(
1017 &self,
1018 state: &mut RuntimeState,
1019 _cx: &Cx<Caps>,
1020 admission: ChildRegionAdmission,
1021 _policy: P2,
1022 f: F,
1023 ) -> Result<Outcome<T, P2::Error>, RegionCreateError>
1024 where
1025 P2: Policy,
1026 F: FnOnce(Scope<'_, P2>, &mut RuntimeState) -> Fut,
1027 Fut: Future<Output = Outcome<T, P2::Error>>,
1028 {
1029 let child_region = state.create_child_region_with_capability_budget_and_priority(
1030 self.region,
1031 admission.budget,
1032 admission.capability_budget,
1033 admission.requirements,
1034 admission.priority,
1035 )?;
1036 let child_budget = state
1037 .region(child_region)
1038 .map_or(self.budget, crate::record::RegionRecord::budget);
1039 let child_capability_budget = state.region(child_region).map_or(
1040 self.capability_budget,
1041 crate::record::RegionRecord::capability_budget,
1042 );
1043 let child_scope = Scope::<P2>::new_with_capability_budget(
1044 child_region,
1045 child_budget,
1046 child_capability_budget,
1047 );
1048
1049 let fut_result =
1050 std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| f(child_scope, &mut *state)));
1051
1052 let fut = match fut_result {
1053 Ok(fut) => fut,
1054 Err(payload) => {
1055 let reason = CancelReason::fail_fast().with_region(child_region);
1056 let _ = state.cancel_request(child_region, &reason, None);
1057
1058 let orphan_task_ids: Vec<TaskId> = state
1080 .tasks_iter()
1081 .filter_map(|(_, t)| {
1082 if t.owner == child_region
1083 && matches!(
1084 t.state,
1085 TaskState::Created | TaskState::CancelRequested { .. }
1086 )
1087 {
1088 Some(t.id)
1089 } else {
1090 None
1091 }
1092 })
1093 .collect();
1094 for task_id in orphan_task_ids {
1095 if let Some(region) = state.region(child_region) {
1096 region.remove_task(task_id);
1097 }
1098 state.recycle_task(task_id);
1099 }
1100
1101 if let Some(region) = state.region(child_region) {
1102 region.begin_close(None);
1103 }
1104 state.advance_region_state(child_region);
1105 std::panic::resume_unwind(payload);
1106 }
1107 };
1108
1109 let pinned_fut = std::pin::pin!(CatchUnwind { inner: fut });
1110
1111 let runner = RegionRunner {
1112 fut: pinned_fut,
1113 state: Some(state),
1114 child_region,
1115 };
1116
1117 let (result, state) = runner.await;
1118 let outcome = match result {
1119 Ok(outcome) => outcome,
1120 Err(payload) => {
1121 let msg = payload_to_string(&payload);
1122 Outcome::Panicked(PanicPayload::new(msg))
1123 }
1124 };
1125
1126 match &outcome {
1127 Outcome::Ok(_) => {
1128 if let Some(region) = state.region(child_region) {
1129 region.begin_close(None);
1130 }
1131 }
1132 Outcome::Cancelled(reason) => {
1133 let _ = state.cancel_request(child_region, reason, None);
1134 if let Some(region) = state.region(child_region) {
1135 region.begin_close(None);
1136 }
1137 }
1138 Outcome::Err(_) | Outcome::Panicked(_) => {
1139 let reason = CancelReason::fail_fast().with_region(child_region);
1140 let _ = state.cancel_request(child_region, &reason, None);
1141 if let Some(region) = state.region(child_region) {
1142 region.begin_close(None);
1143 }
1144 }
1145 }
1146
1147 let close_notify = state.region(child_region).map(|r| r.close_notify.clone());
1148 state.advance_region_state(child_region);
1149
1150 if let Some(notify) = close_notify {
1151 RegionCloseFuture { state: notify }.await;
1152 }
1153
1154 Ok(outcome)
1155 }
1156
1157 pub async fn join<T1, T2>(
1173 &self,
1174 cx: &Cx,
1175 mut h1: TaskHandle<T1>,
1176 mut h2: TaskHandle<T2>,
1177 ) -> (Result<T1, JoinError>, Result<T2, JoinError>) {
1178 let mut f1 = h1.join(cx);
1179 let mut f2 = h2.join(cx);
1180 let r1 = std::pin::Pin::new(&mut f1).await;
1181 let r2 = std::pin::Pin::new(&mut f2).await;
1182 (r1, r2)
1183 }
1184
1185 fn record_loser_drain_start(&self, cx: &Cx, participants: Vec<TaskId>) -> Option<u64> {
1199 let time = cx.now_for_observability();
1200 cx.loser_drain_history_handle()
1201 .map(|history| history.record_race_start(self.region, participants, time))
1202 }
1203
1204 fn record_loser_drain_task_complete(cx: &Cx, task: TaskId) {
1205 if let Some(history) = cx.loser_drain_history_handle() {
1206 history.record_task_complete(task, cx.now_for_observability());
1207 }
1208 }
1209
1210 fn record_loser_drain_complete(cx: &Cx, race_id: Option<u64>, winner: TaskId) {
1211 if let (Some(race_id), Some(history)) = (race_id, cx.loser_drain_history_handle()) {
1212 history.record_race_complete(race_id, winner, cx.now_for_observability());
1213 }
1214 }
1215
1216 fn best_effort_poll_loser_join<T>(cx: &Cx, handle: &mut TaskHandle<T>) -> bool {
1217 let mut drain = std::pin::pin!(handle.join(cx));
1218 let waker = std::task::Waker::noop();
1219 let mut poll_cx = std::task::Context::from_waker(waker);
1220 matches!(drain.as_mut().poll(&mut poll_cx), Poll::Ready(_))
1221 }
1222
1223 pub async fn race<T>(
1225 &self,
1226 cx: &Cx,
1227 mut h1: TaskHandle<T>,
1228 mut h2: TaskHandle<T>,
1229 ) -> Result<T, JoinError> {
1230 let race_id = self.record_loser_drain_start(cx, vec![h1.task_id(), h2.task_id()]);
1231 let winner = {
1232 let f1 = h1.join_with_drop_reason(cx, CancelReason::race_loser());
1233 let mut f1 = std::pin::pin!(f1);
1234 let f2 = h2.join_with_drop_reason(cx, CancelReason::race_loser());
1235 let mut f2 = std::pin::pin!(f2);
1236 Select::new(f1.as_mut(), f2.as_mut())
1237 .await
1238 .map_err(|_| JoinError::PolledAfterCompletion)?
1239 };
1240
1241 match winner {
1242 Either::Left(res) => {
1243 Self::record_loser_drain_task_complete(cx, h1.task_id());
1244 if matches!(&res, Err(JoinError::Panicked(_)))
1245 && crate::runtime::scheduler::three_lane::current_worker_id().is_none()
1246 {
1247 if Self::best_effort_poll_loser_join(cx, &mut h2) {
1252 Self::record_loser_drain_task_complete(cx, h2.task_id());
1253 }
1254 Self::record_loser_drain_complete(cx, race_id, h1.task_id());
1255 return res;
1256 }
1257 let loser_res = h2.join(cx).await;
1258 Self::record_loser_drain_task_complete(cx, h2.task_id());
1259 Self::record_loser_drain_complete(cx, race_id, h1.task_id());
1260 if let Err(JoinError::Panicked(p)) = res {
1261 Err(JoinError::Panicked(p))
1262 } else if let Err(JoinError::Panicked(p)) = loser_res {
1263 Err(JoinError::Panicked(p))
1264 } else {
1265 res
1266 }
1267 }
1268 Either::Right(res) => {
1269 Self::record_loser_drain_task_complete(cx, h2.task_id());
1270 if matches!(&res, Err(JoinError::Panicked(_)))
1271 && crate::runtime::scheduler::three_lane::current_worker_id().is_none()
1272 {
1273 if Self::best_effort_poll_loser_join(cx, &mut h1) {
1275 Self::record_loser_drain_task_complete(cx, h1.task_id());
1276 }
1277 Self::record_loser_drain_complete(cx, race_id, h2.task_id());
1278 return res;
1279 }
1280 let loser_res = h1.join(cx).await;
1281 Self::record_loser_drain_task_complete(cx, h1.task_id());
1282 Self::record_loser_drain_complete(cx, race_id, h2.task_id());
1283 if let Err(JoinError::Panicked(p)) = res {
1284 Err(JoinError::Panicked(p))
1285 } else if let Err(JoinError::Panicked(p)) = loser_res {
1286 Err(JoinError::Panicked(p))
1287 } else {
1288 res
1289 }
1290 }
1291 }
1292 }
1293
1294 pub async fn hedge<F1, Fut1, F2, Fut2, T>(
1313 &self,
1314 state: &mut RuntimeState,
1315 cx: &Cx,
1316 delay: std::time::Duration,
1317 primary: F1,
1318 backup: F2,
1319 ) -> Result<T, JoinError>
1320 where
1321 F1: FnOnce(Cx) -> Fut1 + Send + 'static,
1322 Fut1: Future<Output = T> + Send + 'static,
1323 F2: FnOnce(Cx) -> Fut2 + Send + 'static,
1324 Fut2: Future<Output = T> + Send + 'static,
1325 T: Send + 'static,
1326 {
1327 use crate::combinator::Either;
1328 use crate::combinator::select::Select;
1329 let mut h1 = self
1331 .spawn_registered(state, cx, primary)
1332 .map_err(|_| JoinError::Cancelled(CancelReason::resource_unavailable()))?;
1333
1334 let primary_or_delay = {
1337 let f1_primary = h1.join(cx);
1338 let mut f1_primary = std::pin::pin!(f1_primary);
1339
1340 let now = cx
1341 .timer_driver()
1342 .map_or_else(crate::time::wall_now, |d| d.now());
1343 let sleep_fut = crate::time::sleep(now, delay);
1344 let mut sleep_pinned = std::pin::pin!(sleep_fut);
1345
1346 let res = Select::new(f1_primary.as_mut(), sleep_pinned.as_mut())
1347 .await
1348 .map_err(|_| JoinError::PolledAfterCompletion)?;
1349 if matches!(res, Either::Right(())) {
1350 f1_primary.defuse_drop_abort();
1351 }
1352 res
1353 };
1354
1355 match primary_or_delay {
1356 Either::Left(res) => {
1357 res
1359 }
1360 Either::Right(()) => {
1361 let Ok(mut h2) = self.spawn_registered(state, cx, backup) else {
1363 h1.abort_with_reason(CancelReason::resource_unavailable());
1366
1367 if crate::runtime::scheduler::three_lane::current_worker_id().is_some() {
1368 match h1.join(cx).await {
1371 Ok(res) => return Ok(res),
1372 Err(JoinError::Panicked(p)) => return Err(JoinError::Panicked(p)),
1373 Err(JoinError::Cancelled(_) | JoinError::PolledAfterCompletion) => {}
1374 }
1375 } else {
1376 let mut drain = std::pin::pin!(h1.join(cx));
1380 let waker = std::task::Waker::noop();
1381 let mut poll_cx = Context::from_waker(waker);
1382 match drain.as_mut().poll(&mut poll_cx) {
1383 std::task::Poll::Ready(Ok(res)) => return Ok(res),
1384 std::task::Poll::Ready(Err(JoinError::Panicked(p))) => {
1385 return Err(JoinError::Panicked(p));
1386 }
1387 _ => {}
1388 }
1389 }
1390
1391 return Err(JoinError::Cancelled(CancelReason::resource_unavailable()));
1392 };
1393
1394 let race_outcome = {
1396 let f1_race = h1.join_with_drop_reason(cx, CancelReason::race_loser());
1397 let mut f1_race = std::pin::pin!(f1_race);
1398 let f2_race = h2.join_with_drop_reason(cx, CancelReason::race_loser());
1399 let mut f2_race = std::pin::pin!(f2_race);
1400 Select::new(f1_race.as_mut(), f2_race.as_mut())
1401 .await
1402 .map_err(|_| JoinError::PolledAfterCompletion)?
1403 };
1404
1405 match race_outcome {
1406 Either::Left(res) => {
1407 if matches!(&res, Err(JoinError::Panicked(_)))
1408 && crate::runtime::scheduler::three_lane::current_worker_id().is_none()
1409 {
1410 Self::best_effort_poll_loser_join(cx, &mut h2);
1411 return res;
1412 }
1413 let loser_res = h2.join(cx).await;
1414 if let Err(JoinError::Panicked(p)) = res {
1415 Err(JoinError::Panicked(p))
1416 } else if let Err(JoinError::Panicked(p)) = loser_res {
1417 Err(JoinError::Panicked(p))
1418 } else {
1419 res
1420 }
1421 }
1422 Either::Right(res) => {
1423 if matches!(&res, Err(JoinError::Panicked(_)))
1424 && crate::runtime::scheduler::three_lane::current_worker_id().is_none()
1425 {
1426 Self::best_effort_poll_loser_join(cx, &mut h1);
1427 return res;
1428 }
1429 let loser_res = h1.join(cx).await;
1430 if let Err(JoinError::Panicked(p)) = res {
1431 Err(JoinError::Panicked(p))
1432 } else if let Err(JoinError::Panicked(p)) = loser_res {
1433 Err(JoinError::Panicked(p))
1434 } else {
1435 res
1436 }
1437 }
1438 }
1439 }
1440 }
1441 }
1442
1443 pub async fn race_all<T>(
1455 &self,
1456 cx: &Cx,
1457 handles: Vec<TaskHandle<T>>,
1458 ) -> Result<(T, usize), JoinError> {
1459 let mut handles = handles;
1460 if handles.is_empty() {
1461 return std::future::poll_fn(|_poll_cx| {
1462 if cx.checkpoint().is_err() {
1463 let reason = cx
1464 .cancel_reason()
1465 .unwrap_or_else(|| CancelReason::user("race_all cancelled"));
1466 std::task::Poll::Ready(Err(JoinError::Cancelled(reason)))
1467 } else {
1468 std::task::Poll::Pending
1469 }
1470 })
1471 .await;
1472 }
1473 let participant_tasks: Vec<_> = handles.iter().map(TaskHandle::task_id).collect();
1474 let race_id = self.record_loser_drain_start(cx, participant_tasks.clone());
1475
1476 let mut futures: Vec<_> = handles
1477 .iter_mut()
1478 .map(|h| h.join_with_drop_reason(cx, CancelReason::race_loser()))
1479 .collect();
1480 let mut ready_results: Vec<Option<Result<T, JoinError>>> = std::iter::repeat_with(|| None)
1481 .take(futures.len())
1482 .collect();
1483
1484 let winner_idx = std::future::poll_fn(|poll_cx| {
1488 let mut newly_ready = Vec::new();
1489
1490 for (i, future) in futures.iter_mut().enumerate() {
1491 if ready_results[i].is_some() {
1492 continue;
1493 }
1494 if let std::task::Poll::Ready(res) = std::pin::Pin::new(future).poll(poll_cx) {
1495 ready_results[i] = Some(res);
1496 newly_ready.push(i);
1497 }
1498 }
1499
1500 if newly_ready.is_empty() {
1501 std::task::Poll::Pending
1502 } else {
1503 let chosen = newly_ready[cx.random_usize(newly_ready.len())];
1505 std::task::Poll::Ready(chosen)
1506 }
1507 })
1508 .await;
1509
1510 let winner_result = ready_results[winner_idx]
1511 .take()
1512 .expect("winner index must have a ready result");
1513 let winner_task = participant_tasks[winner_idx];
1514 Self::record_loser_drain_task_complete(cx, winner_task);
1515
1516 drop(futures);
1519
1520 let mut loser_panic = None;
1523 let mut pending_loser_indices = Vec::new();
1524 for (i, handle) in handles.iter_mut().enumerate() {
1525 if i == winner_idx {
1526 continue;
1527 }
1528 if let Some(res) = ready_results[i].take() {
1529 Self::record_loser_drain_task_complete(cx, handle.task_id());
1530 if let Err(JoinError::Panicked(p)) = res {
1531 if loser_panic.is_none() {
1532 loser_panic = Some(p);
1533 }
1534 }
1535 } else if handle.is_finished() {
1536 let res = handle.join(cx).await;
1537 Self::record_loser_drain_task_complete(cx, handle.task_id());
1538 if let Err(JoinError::Panicked(p)) = res {
1539 if loser_panic.is_none() {
1540 loser_panic = Some(p);
1541 }
1542 }
1543 } else {
1544 pending_loser_indices.push(i);
1545 }
1546 }
1547
1548 for &idx in &pending_loser_indices {
1552 handles[idx].abort_with_reason(CancelReason::race_loser());
1553 }
1554 if matches!(&winner_result, Err(JoinError::Panicked(_)))
1555 && crate::runtime::scheduler::three_lane::current_worker_id().is_none()
1556 {
1557 for idx in pending_loser_indices {
1562 if Self::best_effort_poll_loser_join(cx, &mut handles[idx]) {
1563 Self::record_loser_drain_task_complete(cx, handles[idx].task_id());
1564 }
1565 }
1566 Self::record_loser_drain_complete(cx, race_id, winner_task);
1567 return winner_result.map(|val| (val, winner_idx));
1568 }
1569 for idx in pending_loser_indices {
1570 let res = handles[idx].join(cx).await;
1571 Self::record_loser_drain_task_complete(cx, handles[idx].task_id());
1572 if let Err(JoinError::Panicked(p)) = res {
1573 if loser_panic.is_none() {
1574 loser_panic = Some(p);
1575 }
1576 }
1577 }
1578
1579 let winner_result = winner_result.map(|val| (val, winner_idx));
1580 Self::record_loser_drain_complete(cx, race_id, winner_task);
1581 if matches!(&winner_result, Err(JoinError::Panicked(_))) {
1582 return winner_result;
1583 }
1584
1585 loser_panic.map_or(winner_result, |panic_payload| {
1586 Err(JoinError::Panicked(panic_payload))
1587 })
1588 }
1589
1590 pub async fn join_all<T>(
1594 &self,
1595 cx: &Cx,
1596 mut handles: Vec<TaskHandle<T>>,
1597 ) -> Vec<Result<T, JoinError>> {
1598 let mut futures: Vec<_> = handles.iter_mut().map(|h| h.join(cx)).collect();
1599 let mut results = Vec::with_capacity(futures.len());
1600 for fut in &mut futures {
1601 results.push(std::pin::Pin::new(fut).await);
1602 }
1603 results
1604 }
1605
1606 pub(crate) fn build_child_task_cx<Caps>(
1607 &self,
1608 state: &RuntimeState,
1609 parent_cx: &Cx<Caps>,
1610 task_id: TaskId,
1611 ) -> (Cx<Caps>, Cx<cap::All>) {
1612 let child_observability = parent_cx.child_observability(self.region, task_id);
1615 let child_entropy = parent_cx.child_entropy(task_id);
1616
1617 let io_driver = state.io_driver_handle();
1619 let timer_driver = state.timer_driver_handle();
1620 let logical_clock = state
1622 .logical_clock_mode()
1623 .build_handle(timer_driver.clone());
1624
1625 let io_cap = parent_cx.io_cap_handle();
1627 let registry = parent_cx.registry_handle();
1628 let remote_cap = parent_cx.remote_cap_handle();
1629 let blocking_pool = parent_cx.blocking_pool_handle();
1630 let evidence_sink = parent_cx.evidence_sink_handle();
1631 let macaroon = parent_cx.macaroon_handle();
1632 let pressure_opt = parent_cx.pressure_handle();
1633
1634 let mut child_cx = Cx::<Caps>::new_with_drivers(
1636 self.region,
1637 task_id,
1638 self.budget,
1639 Some(child_observability),
1640 io_driver,
1641 io_cap,
1642 timer_driver,
1643 Some(child_entropy),
1644 )
1645 .with_logical_clock(logical_clock)
1646 .with_registry_handle(registry)
1647 .with_remote_cap_handle(remote_cap)
1648 .with_blocking_pool_handle(blocking_pool)
1649 .with_evidence_sink(evidence_sink)
1650 .with_macaroon_handle(macaroon);
1651 let _ = child_cx.apply_child_capability_budget(
1652 self.capability_budget,
1653 CapabilityBudgetRequirements::NONE,
1654 );
1655
1656 if let Some(pressure) = pressure_opt {
1658 child_cx = child_cx.with_pressure(pressure);
1659 }
1660
1661 child_cx.set_trace_buffer(state.trace_handle());
1663 child_cx.set_loser_drain_history_handle(state.loser_drain_history_handle());
1664 let child_cx_full = child_cx.retype::<cap::All>();
1665
1666 (child_cx, child_cx_full)
1667 }
1668
1669 pub(crate) fn create_task_record(
1673 &self,
1674 state: &mut RuntimeState,
1675 ) -> Result<TaskId, SpawnError> {
1676 let now = state
1677 .timer_driver()
1678 .map_or(state.now, crate::time::TimerDriverHandle::now);
1679
1680 let region = self.region;
1681 let budget = self.budget;
1682 let idx = state.insert_pooled_task_with(|idx, record| {
1683 record.id = TaskId::from_arena(idx);
1692 record.owner = region;
1693 record.created_at = now;
1694 record.deadline = budget.deadline;
1695 record.polls_remaining = budget.poll_quota;
1696 #[cfg(feature = "tracing-integration")]
1697 {
1698 record.created_instant = crate::time::wall_now();
1699 }
1700 });
1701 let task_id = TaskId::from_arena(idx);
1702
1703 if let Some(region) = state.region(self.region) {
1705 if let Err(err) = region.add_task(task_id) {
1706 state.recycle_task(task_id);
1708 return Err(match err {
1709 AdmissionError::Closed => SpawnError::RegionClosed(self.region),
1710 AdmissionError::LimitReached { limit, live, .. } => {
1711 SpawnError::RegionAtCapacity {
1712 region: self.region,
1713 limit,
1714 live,
1715 }
1716 }
1717 });
1718 }
1719 } else {
1720 state.recycle_task(task_id);
1722 return Err(SpawnError::RegionNotFound(self.region));
1723 }
1724
1725 state.record_task_spawn(task_id, self.region);
1726
1727 Ok(task_id)
1728 }
1729
1730 pub fn defer_sync<F>(&self, state: &mut RuntimeState, f: F) -> bool
1754 where
1755 F: FnOnce() + Send + 'static,
1756 {
1757 state.register_sync_finalizer(self.region, f)
1758 }
1759
1760 pub fn defer_async<F>(&self, state: &mut RuntimeState, future: F) -> bool
1781 where
1782 F: Future<Output = ()> + Send + 'static,
1783 {
1784 state.register_async_finalizer(self.region, future)
1785 }
1786}
1787
1788impl<P: Policy> std::fmt::Debug for Scope<'_, P> {
1789 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1790 f.debug_struct("Scope")
1791 .field("region", &self.region)
1792 .field("budget", &self.budget)
1793 .field("capability_budget", &self.capability_budget)
1794 .finish()
1795 }
1796}
1797
1798#[cfg(test)]
1799mod tests {
1800 #![allow(
1801 clippy::pedantic,
1802 clippy::nursery,
1803 clippy::expect_fun_call,
1804 clippy::map_unwrap_or,
1805 clippy::cast_possible_wrap,
1806 clippy::future_not_send
1807 )]
1808 use super::*;
1809 use crate::record::RegionLimits;
1810 use crate::runtime::RuntimeState;
1811 use crate::types::{
1812 CancelKind, CapabilityBudgetDimension, CapabilityBudgetRefusal, Outcome, Time,
1813 };
1814 use futures_lite::future::block_on;
1815 use std::sync::Arc;
1816
1817 fn test_cx() -> Cx<cap::All> {
1818 Cx::for_testing()
1819 }
1820
1821 fn test_scope(region: RegionId, budget: Budget) -> Scope<'static> {
1822 Scope::new(region, budget)
1823 }
1824
1825 fn test_scope_with_capability_budget(
1826 region: RegionId,
1827 budget: Budget,
1828 capability_budget: CapabilityBudget,
1829 ) -> Scope<'static> {
1830 Scope::new_with_capability_budget(region, budget, capability_budget)
1831 }
1832
1833 #[test]
1834 fn spawn_creates_task_record() {
1835 let mut state = RuntimeState::new();
1836 let cx = test_cx();
1837 let region = state.create_root_region(Budget::INFINITE);
1838 let scope = test_scope(region, Budget::INFINITE);
1839
1840 let (handle, _stored) = scope
1841 .spawn(&mut state, &cx, |_| async { 42_i32 })
1842 .expect("should spawn async task");
1843
1844 let task = state.task(handle.task_id());
1846 assert!(task.is_some());
1847
1848 let task = task.expect("task should exist in state");
1850 assert_eq!(task.owner, region);
1851 }
1852
1853 #[test]
1854 fn spawn_inherits_registry_and_remote_capabilities() {
1855 use crate::cx::registry::RegistryHandle;
1856 use crate::remote::{NodeId, RemoteCap};
1857 use std::task::Context;
1858
1859 let mut state = RuntimeState::new();
1860
1861 let registry = crate::cx::NameRegistry::new();
1862 let registry_handle = RegistryHandle::new(Arc::new(registry));
1863 let parent_registry_arc = registry_handle.as_arc();
1864
1865 let cx = test_cx()
1866 .with_registry_handle(Some(registry_handle))
1867 .with_remote_cap(RemoteCap::new().with_local_node(NodeId::new("origin-test")));
1868
1869 let region = state.create_root_region(Budget::INFINITE);
1870 let scope = test_scope(region, Budget::INFINITE);
1871
1872 let mut handle = scope
1873 .spawn_registered(&mut state, &cx, move |cx| async move {
1874 let child_registry = cx.registry_handle().expect("child must inherit registry");
1875 let child_registry_arc = child_registry.as_arc();
1876 let same_registry = Arc::ptr_eq(&child_registry_arc, &parent_registry_arc);
1877
1878 let child_remote = cx.remote().expect("child must inherit remote cap");
1879 let origin = child_remote.local_node().as_str().to_owned();
1880
1881 (same_registry, origin)
1882 })
1883 .expect("should spawn registered task for capability inheritance test");
1884
1885 let waker = std::task::Waker::noop().clone();
1886 let mut poll_cx = Context::from_waker(&waker);
1887
1888 let stored = state
1889 .get_stored_future(handle.task_id())
1890 .expect("spawn_registered must store the task");
1891 assert!(stored.poll(&mut poll_cx).is_ready());
1892
1893 let mut join_fut = std::pin::pin!(handle.join(&cx));
1894 match join_fut.as_mut().poll(&mut poll_cx) {
1895 Poll::Ready(Ok((same_registry, origin))) => {
1896 assert!(
1897 same_registry,
1898 "child should observe the same RegistryCap instance"
1899 );
1900 assert_eq!(origin, "origin-test");
1901 }
1902 other => unreachable!("Expected Ready(Ok(_)), got {other:?}"),
1903 }
1904 }
1905
1906 #[test]
1907 fn spawn_inherits_runtime_timer_driver() {
1908 use std::task::Context;
1909
1910 let mut state = RuntimeState::new();
1911 let clock = Arc::new(crate::time::VirtualClock::new());
1912 state.set_timer_driver(crate::time::TimerDriverHandle::with_virtual_clock(clock));
1913
1914 let cx = test_cx();
1915 let region = state.create_root_region(Budget::INFINITE);
1916 let scope = test_scope(region, Budget::INFINITE);
1917
1918 let (mut handle, mut stored) = scope
1919 .spawn(&mut state, &cx, |cx| async move { cx.has_timer() })
1920 .expect("spawn should succeed");
1921
1922 let waker = std::task::Waker::noop().clone();
1923 let mut poll_cx = Context::from_waker(&waker);
1924 assert!(stored.poll(&mut poll_cx).is_ready());
1925
1926 let mut join_fut = std::pin::pin!(handle.join(&cx));
1927 match join_fut.as_mut().poll(&mut poll_cx) {
1928 Poll::Ready(Ok(has_timer)) => assert!(has_timer),
1929 other => unreachable!("Expected Ready(Ok(_)), got {other:?}"),
1930 }
1931 }
1932
1933 #[test]
1934 fn create_task_record_uses_runtime_timer_driver_time() {
1935 let mut state = RuntimeState::new();
1936 let clock = Arc::new(crate::time::VirtualClock::starting_at(Time::from_millis(
1937 11,
1938 )));
1939 state.set_timer_driver(crate::time::TimerDriverHandle::with_virtual_clock(
1940 clock.clone(),
1941 ));
1942
1943 let region = state.create_root_region(Budget::INFINITE);
1944 let scope = test_scope(region, Budget::INFINITE);
1945
1946 clock.advance(Time::from_millis(7).as_nanos());
1947 let task_id = scope
1948 .create_task_record(&mut state)
1949 .expect("task record should be created");
1950
1951 let task = state.task(task_id).expect("task record");
1952 assert_eq!(task.created_at, Time::from_millis(18));
1953 assert_eq!(task.id, task_id);
1954 }
1955
1956 #[test]
1957 fn spawn_blocking_inherits_runtime_timer_driver() {
1958 use std::task::Context;
1959
1960 let mut state = RuntimeState::new();
1961 let clock = Arc::new(crate::time::VirtualClock::new());
1962 state.set_timer_driver(crate::time::TimerDriverHandle::with_virtual_clock(clock));
1963
1964 let cx = test_cx();
1965 let region = state.create_root_region(Budget::INFINITE);
1966 let scope = test_scope(region, Budget::INFINITE);
1967
1968 let (mut handle, mut stored) = scope
1969 .spawn_blocking(&mut state, &cx, |cx| cx.has_timer())
1970 .expect("spawn_blocking should succeed");
1971
1972 let waker = std::task::Waker::noop().clone();
1973 let mut poll_cx = Context::from_waker(&waker);
1974 assert!(stored.poll(&mut poll_cx).is_ready());
1975
1976 let mut join_fut = std::pin::pin!(handle.join(&cx));
1977 match join_fut.as_mut().poll(&mut poll_cx) {
1978 Poll::Ready(Ok(has_timer)) => assert!(has_timer),
1979 other => unreachable!("Expected Ready(Ok(_)), got {other:?}"),
1980 }
1981 }
1982
1983 #[test]
1984 fn spawn_registered_stores_task() {
1985 let mut state = RuntimeState::new();
1986 let cx = test_cx();
1987 let region = state.create_root_region(Budget::INFINITE);
1988 let scope = test_scope(region, Budget::INFINITE);
1989
1990 let handle = scope
1992 .spawn_registered(&mut state, &cx, |_| async { 42_i32 })
1993 .expect("should spawn and register task");
1994
1995 let task = state.task(handle.task_id());
1997 assert!(task.is_some());
1998 assert_eq!(task.expect("task should exist in state").owner, region);
1999
2000 let stored = state.get_stored_future(handle.task_id());
2002 assert!(stored.is_some(), "spawn_registered should store the task");
2003 }
2004
2005 #[test]
2006 fn spawn_registered_task_can_be_polled() {
2007 use std::task::Context;
2008
2009 let mut state = RuntimeState::new();
2010 let cx = test_cx();
2011 let region = state.create_root_region(Budget::INFINITE);
2012 let scope = test_scope(region, Budget::INFINITE);
2013
2014 let mut handle = scope
2015 .spawn_registered(&mut state, &cx, |_| async { 42_i32 })
2016 .expect("should spawn registered task for polling test");
2017
2018 let waker = std::task::Waker::noop().clone();
2020 let mut poll_cx = Context::from_waker(&waker);
2021
2022 let stored = state
2023 .get_stored_future(handle.task_id())
2024 .expect("should retrieve stored future for polling");
2025 let poll_result = stored.poll(&mut poll_cx);
2026 assert!(
2027 poll_result.is_ready(),
2028 "Simple async should complete in one poll"
2029 );
2030
2031 let mut join_fut = std::pin::pin!(handle.join(&cx));
2033 match join_fut.as_mut().poll(&mut poll_cx) {
2034 Poll::Ready(Ok(val)) => assert_eq!(val, 42),
2035 other => unreachable!("Expected Ready(Ok(42)), got {other:?}"),
2036 }
2037 }
2038
2039 #[test]
2040 fn spawn_blocking_creates_task_record() {
2041 let mut state = RuntimeState::new();
2042 let cx = test_cx();
2043 let region = state.create_root_region(Budget::INFINITE);
2044 let scope = test_scope(region, Budget::INFINITE);
2045
2046 let (handle, _stored) = scope
2047 .spawn_blocking(&mut state, &cx, |_| 42_i32)
2048 .expect("should spawn blocking task");
2049
2050 let task = state.task(handle.task_id());
2052 assert!(task.is_some());
2053 assert_eq!(
2054 task.expect("blocking task should exist in state").owner,
2055 region
2056 );
2057 }
2058
2059 #[test]
2060 fn spawn_local_creates_task_record() {
2061 let mut state = RuntimeState::new();
2062 let cx = test_cx();
2063 let region = state.create_root_region(Budget::INFINITE);
2064 let scope = test_scope(region, Budget::INFINITE);
2065
2066 let local_ready = Arc::new(parking_lot::Mutex::new(std::collections::VecDeque::new()));
2067 let _local_ready_guard =
2068 crate::runtime::scheduler::three_lane::ScopedLocalReady::new(Arc::clone(&local_ready));
2069 let _worker_guard = crate::runtime::scheduler::three_lane::ScopedWorkerId::new(1);
2070
2071 let handle = scope
2074 .spawn_local(&mut state, &cx, |_| async move { 42_i32 })
2075 .unwrap();
2076
2077 let task = state.task(handle.task_id());
2079 assert!(task.is_some());
2080 assert_eq!(task.unwrap().owner, region);
2081 }
2082
2083 #[test]
2084 fn spawn_local_without_scheduler_fails_and_rolls_back() {
2085 let mut state = RuntimeState::new();
2086 let cx = test_cx();
2087 let region = state.create_root_region(Budget::INFINITE);
2088 let scope = test_scope(region, Budget::INFINITE);
2089
2090 let result = scope.spawn_local(&mut state, &cx, |_| async move { 5_i32 });
2091 assert!(matches!(result, Err(SpawnError::LocalSchedulerUnavailable)));
2092
2093 assert!(state.tasks_is_empty());
2095 let region_record = state.region(region).unwrap();
2096 assert!(region_record.task_ids().is_empty());
2097 }
2098
2099 #[test]
2100 fn spawn_local_makes_progress_via_local_ready() {
2101 use std::sync::Arc;
2102 use std::task::Context;
2103
2104 let mut state = RuntimeState::new();
2105 let cx = test_cx();
2106 let region = state.create_root_region(Budget::INFINITE);
2107 let scope = test_scope(region, Budget::INFINITE);
2108
2109 let local_ready = Arc::new(parking_lot::Mutex::new(std::collections::VecDeque::new()));
2110 let _local_ready_guard =
2111 crate::runtime::scheduler::three_lane::ScopedLocalReady::new(Arc::clone(&local_ready));
2112 let _worker_guard = crate::runtime::scheduler::three_lane::ScopedWorkerId::new(1);
2113
2114 let mut handle = scope
2115 .spawn_local(&mut state, &cx, |_| async move { 7_i32 })
2116 .unwrap();
2117
2118 let queued = {
2119 let queue = local_ready.lock();
2120 queue.contains(&handle.task_id())
2121 };
2122 assert!(queued, "spawn_local should enqueue into local_ready");
2123
2124 let task_id = {
2125 let mut queue = local_ready.lock();
2126 queue
2127 .pop_front()
2128 .expect("local_ready should contain spawned task")
2129 };
2130
2131 let mut join_fut = std::pin::pin!(handle.join(&cx));
2132 let waker = std::task::Waker::noop().clone();
2133 let mut ctx = Context::from_waker(&waker);
2134
2135 assert!(join_fut.as_mut().poll(&mut ctx).is_pending());
2136
2137 let mut local_task =
2138 crate::runtime::local::remove_local_task(task_id).expect("local task missing");
2139 assert!(local_task.poll(&mut ctx).is_ready());
2140
2141 match join_fut.as_mut().poll(&mut ctx) {
2142 Poll::Ready(Ok(val)) => assert_eq!(val, 7),
2143 res => unreachable!("Expected Ready(Ok(7)), got {res:?}"),
2144 }
2145 }
2146
2147 #[test]
2148 fn task_added_to_region() {
2149 let mut state = RuntimeState::new();
2150 let cx = test_cx();
2151 let region = state.create_root_region(Budget::INFINITE);
2152 let scope = test_scope(region, Budget::INFINITE);
2153
2154 let (handle, _stored) = scope.spawn(&mut state, &cx, |_| async { 42_i32 }).unwrap();
2155
2156 let region_record = state.region(region).unwrap();
2158 assert!(region_record.task_ids().contains(&handle.task_id()));
2159 }
2160
2161 #[test]
2162 fn multiple_spawns_create_distinct_tasks() {
2163 let mut state = RuntimeState::new();
2164 let cx = test_cx();
2165 let region = state.create_root_region(Budget::INFINITE);
2166 let scope = test_scope(region, Budget::INFINITE);
2167
2168 let (handle1, _) = scope.spawn(&mut state, &cx, |_| async { 1_i32 }).unwrap();
2169 let (handle2, _) = scope.spawn(&mut state, &cx, |_| async { 2_i32 }).unwrap();
2170 let (handle3, _) = scope.spawn(&mut state, &cx, |_| async { 3_i32 }).unwrap();
2171
2172 assert_ne!(handle1.task_id(), handle2.task_id());
2174 assert_ne!(handle2.task_id(), handle3.task_id());
2175 assert_ne!(handle1.task_id(), handle3.task_id());
2176
2177 let region_record = state.region(region).unwrap();
2179 assert!(region_record.task_ids().contains(&handle1.task_id()));
2180 assert!(region_record.task_ids().contains(&handle2.task_id()));
2181 assert!(region_record.task_ids().contains(&handle3.task_id()));
2182 }
2183
2184 #[test]
2185 fn spawn_into_closing_region_should_fail() {
2186 let mut state = RuntimeState::new();
2187 let cx = test_cx();
2188 let region = state.create_root_region(Budget::INFINITE);
2189 let scope = test_scope(region, Budget::INFINITE);
2190
2191 let region_record = state.region_mut(region).expect("region");
2193 region_record.begin_close(None);
2194
2195 let result = scope.spawn(&mut state, &cx, |_| async { 42 });
2197 assert!(matches!(result, Err(SpawnError::RegionClosed(_))));
2198 }
2199
2200 #[test]
2201 fn test_join_manual_poll() {
2202 use std::task::Context;
2203
2204 let mut state = RuntimeState::new();
2205 let cx = test_cx();
2206 let region = state.create_root_region(Budget::INFINITE);
2207 let scope = test_scope(region, Budget::INFINITE);
2208
2209 let (mut handle, mut stored_task) =
2211 scope.spawn(&mut state, &cx, |_| async { 42_i32 }).unwrap();
2212 let mut join_fut = std::pin::pin!(handle.join(&cx));
2216
2217 let waker = std::task::Waker::noop().clone();
2219 let mut ctx = Context::from_waker(&waker);
2220
2221 assert!(join_fut.as_mut().poll(&mut ctx).is_pending());
2223
2224 assert!(stored_task.poll(&mut ctx).is_ready());
2226
2227 match join_fut.as_mut().poll(&mut ctx) {
2229 Poll::Ready(Ok(val)) => assert_eq!(val, 42),
2230 other => unreachable!("Expected Ready(Ok(42)), got {other:?}"),
2231 }
2232 }
2233
2234 #[test]
2235 fn spawn_abort_cancels_task() {
2236 use std::task::{Context, Poll};
2237
2238 let mut state = RuntimeState::new();
2239 let cx = test_cx();
2240 let region = state.create_root_region(Budget::INFINITE);
2241 let scope = test_scope(region, Budget::INFINITE);
2242
2243 let (mut handle, mut stored_task) = scope
2245 .spawn(&mut state, &cx, |cx| async move {
2246 if cx.checkpoint().is_err() {
2248 return "cancelled";
2249 }
2250 "finished"
2251 })
2252 .unwrap();
2253
2254 handle.abort();
2256
2257 let waker = std::task::Waker::noop().clone();
2259 let mut ctx = Context::from_waker(&waker);
2260
2261 match stored_task.poll(&mut ctx) {
2263 Poll::Ready(crate::types::Outcome::Ok(())) => {}
2264 res => unreachable!("Task should have completed with Ok(()), got {res:?}"),
2265 }
2266
2267 let mut join_fut = std::pin::pin!(handle.join(&cx));
2269 match join_fut.as_mut().poll(&mut ctx) {
2270 Poll::Ready(Ok(val)) => assert_eq!(val, "cancelled"),
2271 Poll::Ready(Err(e)) => unreachable!("Task failed unexpectedly: {e}"),
2272 Poll::Pending => unreachable!("Join should be ready"),
2273 }
2274 }
2275
2276 #[test]
2277 fn hedge_backup_spawn_failure_aborts_primary() {
2278 let mut state = RuntimeState::new();
2279 let cx = test_cx();
2280 let region = state.create_root_region(Budget::INFINITE);
2281 let scope = test_scope(region, Budget::INFINITE);
2282
2283 let limits = RegionLimits {
2284 max_tasks: Some(1),
2285 ..RegionLimits::unlimited()
2286 };
2287 assert!(state.set_region_limits(region, limits));
2288
2289 let result = block_on(scope.hedge(
2290 &mut state,
2291 &cx,
2292 std::time::Duration::ZERO,
2293 |_| async { 1_u8 },
2294 |_| async { 2_u8 },
2295 ));
2296
2297 assert!(matches!(
2298 result,
2299 Err(JoinError::Cancelled(reason))
2300 if reason.kind == CancelKind::ResourceUnavailable
2301 ));
2302
2303 let task_id = *state
2304 .region(region)
2305 .expect("region missing")
2306 .task_ids()
2307 .first()
2308 .expect("primary task should remain tracked");
2309
2310 let task = state.task(task_id).expect("primary task record missing");
2311 let (cancel_requested, cancel_reason_kind) = {
2312 let inner = task
2313 .cx_inner
2314 .as_ref()
2315 .expect("primary task must have shared Cx inner")
2316 .read();
2317 (
2318 inner.cancel_requested,
2319 inner.cancel_reason.as_ref().map(|r| r.kind),
2320 )
2321 };
2322
2323 assert!(
2324 cancel_requested,
2325 "primary task must be cancellation-requested when backup spawn fails"
2326 );
2327 assert_eq!(cancel_reason_kind, Some(CancelKind::ResourceUnavailable));
2328 }
2329
2330 #[test]
2331 fn region_closes_empty_child() {
2332 let mut state = RuntimeState::new();
2333 let cx = test_cx();
2334 let parent = state.create_root_region(Budget::INFINITE);
2335 let scope = test_scope(parent, Budget::INFINITE);
2336
2337 let outcome = block_on(scope.region(
2338 &mut state,
2339 &cx,
2340 crate::types::policy::FailFast,
2341 |child, _state| {
2342 let child_id = child.region_id();
2343 async move { Outcome::Ok(child_id) }
2344 },
2345 ))
2346 .expect("child region created");
2347
2348 let child_id = match outcome {
2349 Outcome::Ok(id) => id,
2350 other => unreachable!("expected Outcome::Ok(child_id), got {other:?}"),
2351 };
2352
2353 assert!(
2354 state.region(child_id).is_none(),
2355 "closed child region should be reclaimed from arena"
2356 );
2357
2358 let parent_record = state.region(parent).expect("parent record missing");
2359 assert!(
2360 !parent_record.child_ids().contains(&child_id),
2361 "closed child should be removed from parent"
2362 );
2363 }
2364
2365 #[test]
2366 fn region_budget_is_met_with_parent() {
2367 let mut state = RuntimeState::new();
2368 let cx = test_cx();
2369 let parent = state.create_root_region(Budget::with_deadline_secs(10));
2370 let scope = test_scope(parent, Budget::with_deadline_secs(10));
2371
2372 let outcome = block_on(scope.region_with_budget(
2373 &mut state,
2374 &cx,
2375 Budget::with_deadline_secs(30),
2376 crate::types::policy::FailFast,
2377 |child, _state| {
2378 let child_id = child.region_id();
2379 let child_budget = child.budget();
2380 async move { Outcome::Ok((child_id, child_budget)) }
2381 },
2382 ))
2383 .expect("child region created");
2384
2385 let (child_id, child_budget) = match outcome {
2386 Outcome::Ok(tuple) => tuple,
2387 other => unreachable!("expected Outcome::Ok(child_id), got {other:?}"),
2388 };
2389
2390 assert_eq!(
2391 child_budget.deadline,
2392 Some(crate::types::Time::from_secs(10))
2393 );
2394 assert!(
2395 state.region(child_id).is_none(),
2396 "closed child region should be reclaimed from arena"
2397 );
2398 }
2399
2400 #[test]
2401 fn region_with_priority_uses_requested_admission_priority() {
2402 let mut state = RuntimeState::new();
2403 let cx = test_cx();
2404 let parent = state.create_root_region(Budget::INFINITE);
2405 let scope = test_scope(parent, Budget::INFINITE);
2406
2407 state
2408 .resource_monitor()
2409 .pressure()
2410 .update_degradation_level(
2411 crate::runtime::resource_monitor::ResourceType::Memory,
2412 crate::runtime::resource_monitor::DegradationLevel::Moderate,
2413 );
2414
2415 block_on(scope.region(
2416 &mut state,
2417 &cx,
2418 crate::types::policy::FailFast,
2419 |child, _state| {
2420 let child_id = child.region_id();
2421 async move { Outcome::Ok(child_id) }
2422 },
2423 ))
2424 .expect("normal child scope should be admitted at moderate pressure");
2425
2426 let err = block_on(scope.region_with_priority(
2427 &mut state,
2428 &cx,
2429 RegionPriority::Low,
2430 crate::types::policy::FailFast,
2431 |_child, _state| async move { Outcome::Ok(()) },
2432 ))
2433 .expect_err("low-priority child scope should be rejected at moderate pressure");
2434
2435 assert!(matches!(
2436 err,
2437 RegionCreateError::ResourcePressure {
2438 requested_priority: RegionPriority::Low,
2439 ..
2440 }
2441 ));
2442 }
2443
2444 #[test]
2445 fn region_with_priority_registers_child_shedding_priority() {
2446 let mut state = RuntimeState::new();
2447 let cx = test_cx();
2448 let parent = state.create_root_region(Budget::INFINITE);
2449 let scope = test_scope(parent, Budget::INFINITE);
2450
2451 let outcome = block_on(scope.region_with_priority(
2452 &mut state,
2453 &cx,
2454 RegionPriority::Low,
2455 crate::types::policy::FailFast,
2456 |child, state| {
2457 let child_id = child.region_id();
2458 state
2459 .resource_monitor()
2460 .pressure()
2461 .update_degradation_level(
2462 crate::runtime::resource_monitor::ResourceType::Memory,
2463 crate::runtime::resource_monitor::DegradationLevel::Heavy,
2464 );
2465 let decision = state
2466 .resource_monitor()
2467 .engine()
2468 .should_shed_region(child_id);
2469 async move { Outcome::Ok(decision) }
2470 },
2471 ))
2472 .expect("low-priority child scope should be admitted before pressure rises");
2473
2474 assert!(matches!(
2475 outcome,
2476 Outcome::Ok(crate::runtime::resource_monitor::SheddingDecision::Pause)
2477 ));
2478 }
2479
2480 #[test]
2481 fn region_capability_budget_is_met_with_parent() {
2482 let mut state = RuntimeState::new();
2483 let cx = test_cx();
2484 let parent_budget = CapabilityBudget::new()
2485 .with_memory_bytes(1_024)
2486 .with_io_bytes(8_192);
2487 let parent =
2488 state.create_root_region_with_capability_budget(Budget::INFINITE, parent_budget);
2489 let scope = test_scope_with_capability_budget(parent, Budget::INFINITE, parent_budget);
2490
2491 let outcome = block_on(
2492 scope.region_with_budget_and_capability_budget(
2493 &mut state,
2494 &cx,
2495 Budget::INFINITE,
2496 CapabilityBudget::new()
2497 .with_memory_bytes(4_096)
2498 .with_io_bytes(512),
2499 CapabilityBudgetRequirements::new()
2500 .require_memory_bytes()
2501 .require_io_bytes(),
2502 crate::types::policy::FailFast,
2503 |child, _state| {
2504 let child_budget = child.capability_budget();
2505 async move { Outcome::Ok(child_budget) }
2506 },
2507 ),
2508 )
2509 .expect("child region created");
2510
2511 let child_budget = match outcome {
2512 Outcome::Ok(budget) => budget,
2513 other => unreachable!("expected Outcome::Ok(capability_budget), got {other:?}"),
2514 };
2515
2516 assert_eq!(child_budget.memory_bytes, Some(1_024));
2517 assert_eq!(child_budget.io_bytes, Some(512));
2518 }
2519
2520 #[test]
2521 fn region_capability_budget_required_absent_dimension_rejects() {
2522 let mut state = RuntimeState::new();
2523 let cx = test_cx();
2524 let parent = state.create_root_region(Budget::INFINITE);
2525 let scope = test_scope(parent, Budget::INFINITE);
2526
2527 let err = block_on(scope.region_with_budget_and_capability_budget(
2528 &mut state,
2529 &cx,
2530 Budget::INFINITE,
2531 CapabilityBudget::new(),
2532 CapabilityBudgetRequirements::new().require_artifact_bytes(),
2533 crate::types::policy::FailFast,
2534 |_child, _state| async move { Outcome::Ok(()) },
2535 ))
2536 .expect_err("missing required artifact budget must fail closed");
2537
2538 assert!(matches!(
2539 err,
2540 RegionCreateError::CapabilityBudgetRefused {
2541 parent: refused_parent,
2542 reason: CapabilityBudgetRefusal::MissingRequired(
2543 CapabilityBudgetDimension::ArtifactBytes
2544 ),
2545 } if refused_parent == parent
2546 ));
2547 }
2548
2549 #[test]
2550 fn spawned_task_cx_inherits_scope_capability_budget() {
2551 let mut state = RuntimeState::new();
2552 let cx = test_cx();
2553 let capability_budget = CapabilityBudget::new()
2554 .with_memory_bytes(4_096)
2555 .with_cpu_units(16);
2556 let region =
2557 state.create_root_region_with_capability_budget(Budget::INFINITE, capability_budget);
2558 let scope = test_scope_with_capability_budget(region, Budget::INFINITE, capability_budget);
2559
2560 let (handle, _stored) = scope
2561 .spawn(&mut state, &cx, |child_cx| async move {
2562 child_cx.capability_budget()
2563 })
2564 .expect("spawn should admit task");
2565
2566 let task = state.task(handle.task_id()).expect("task record missing");
2567 let actual = task
2568 .cx_inner
2569 .as_ref()
2570 .expect("task cx inner missing")
2571 .read()
2572 .capability_budget;
2573
2574 assert_eq!(actual, capability_budget);
2575 }
2576
2577 #[test]
2578 fn region_spawns_tasks_in_child() {
2579 use std::task::{Context, Poll};
2580
2581 let mut state = RuntimeState::new();
2582 let cx = test_cx();
2583 let parent = state.create_root_region(Budget::INFINITE);
2584 let scope = test_scope(parent, Budget::INFINITE);
2585
2586 let outcome = block_on(scope.region(
2587 &mut state,
2588 &cx,
2589 crate::types::policy::FailFast,
2590 |child, state| {
2591 let child_id = child.region_id();
2592 let (handle, mut stored) = child
2593 .spawn(state, &cx, |_| async { 7_i32 })
2594 .expect("spawn in child");
2595
2596 let parent_has = state
2597 .region(parent)
2598 .expect("parent record missing")
2599 .task_ids()
2600 .contains(&handle.task_id());
2601 let child_has = state
2602 .region(child_id)
2603 .expect("child record missing")
2604 .task_ids()
2605 .contains(&handle.task_id());
2606
2607 let waker = std::task::Waker::noop().clone();
2608 let mut poll_cx = Context::from_waker(&waker);
2609 let poll_result = stored.poll(&mut poll_cx);
2610 if let Poll::Ready(outcome) = poll_result {
2611 let task_outcome = match outcome {
2612 Outcome::Ok(()) => Outcome::Ok(()),
2613 Outcome::Panicked(payload) => Outcome::Panicked(payload),
2614 other => unreachable!("unexpected task outcome: {other:?}"),
2615 };
2616 if let Some(task_record) = state.task_mut(handle.task_id()) {
2617 task_record.complete(task_outcome);
2618 }
2619 let _ = state.task_completed(handle.task_id());
2620 }
2621
2622 std::future::ready(Outcome::Ok((child_id, parent_has, child_has)))
2623 },
2624 ))
2625 .expect("child region created");
2626
2627 let (child_id, parent_has, child_has) = match outcome {
2628 Outcome::Ok(tuple) => tuple,
2629 other => unreachable!("expected Outcome::Ok(tuple), got {other:?}"),
2630 };
2631
2632 assert!(!parent_has, "task should not be owned by parent region");
2633 assert!(child_has, "task should be owned by child region");
2634
2635 let parent_record = state.region(parent).expect("parent record missing");
2636 assert!(
2637 !parent_record.child_ids().contains(&child_id),
2638 "closed child should be removed from parent"
2639 );
2640 }
2641
2642 #[test]
2643 fn spawn_panic_propagates_as_panicked_error() {
2644 use std::task::{Context, Poll};
2645
2646 let mut state = RuntimeState::new();
2647 let cx = test_cx();
2648 let region = state.create_root_region(Budget::INFINITE);
2649 let scope = test_scope(region, Budget::INFINITE);
2650
2651 let (mut handle, mut stored_task) = scope
2652 .spawn(&mut state, &cx, |_| async {
2653 std::panic::panic_any("oops");
2654 })
2655 .unwrap();
2656
2657 let waker = std::task::Waker::noop().clone();
2659 let mut ctx = Context::from_waker(&waker);
2660
2661 match stored_task.poll(&mut ctx) {
2663 Poll::Ready(crate::types::Outcome::Panicked(_)) => {}
2664 res => unreachable!("Task should have completed with Panicked, got {res:?}"),
2665 }
2666
2667 let mut join_fut = std::pin::pin!(handle.join(&cx));
2669 match join_fut.as_mut().poll(&mut ctx) {
2670 Poll::Ready(Err(JoinError::Panicked(p))) => {
2671 assert_eq!(p.message(), "oops");
2672 }
2673 res => unreachable!("Expected Panicked, got {res:?}"),
2674 }
2675 }
2676
2677 #[test]
2678 fn join_all_success() {
2679 use std::task::{Context, Poll};
2680
2681 let mut state = RuntimeState::new();
2682 let cx = test_cx();
2683 let region = state.create_root_region(Budget::INFINITE);
2684 let scope = test_scope(region, Budget::INFINITE);
2685
2686 let (h1, mut t1) = scope.spawn(&mut state, &cx, |_| async { 1 }).unwrap();
2687 let (h2, mut t2) = scope.spawn(&mut state, &cx, |_| async { 2 }).unwrap();
2688
2689 let waker = std::task::Waker::noop().clone();
2691 let mut ctx = Context::from_waker(&waker);
2692 assert!(t1.poll(&mut ctx).is_ready());
2693 assert!(t2.poll(&mut ctx).is_ready());
2694
2695 let handles = vec![h1, h2];
2696 let mut fut = Box::pin(scope.join_all(&cx, handles));
2697
2698 match fut.as_mut().poll(&mut ctx) {
2699 Poll::Ready(results) => {
2700 assert_eq!(results.len(), 2);
2701 assert_eq!(results[0].as_ref().unwrap(), &1);
2702 assert_eq!(results[1].as_ref().unwrap(), &2);
2703 }
2704 Poll::Pending => unreachable!("join_all should be ready"),
2705 }
2706 }
2707
2708 #[test]
2709 fn race_all_aborted_task_is_drained() {
2710 use std::task::{Context, Poll};
2711
2712 let mut state = RuntimeState::new();
2713 let cx = test_cx();
2714 let region = state.create_root_region(Budget::INFINITE);
2715 let scope = test_scope(region, Budget::INFINITE);
2716
2717 let (h1, mut t1) = scope.spawn(&mut state, &cx, |_| async { 1 }).unwrap();
2719
2720 let (h2, mut t2) = scope
2722 .spawn(&mut state, &cx, |cx| async move {
2723 struct YieldOnce(bool);
2725 impl std::future::Future for YieldOnce {
2726 type Output = ();
2727 fn poll(
2728 mut self: std::pin::Pin<&mut Self>,
2729 cx: &mut std::task::Context<'_>,
2730 ) -> std::task::Poll<()> {
2731 if self.0 {
2732 std::task::Poll::Ready(())
2733 } else {
2734 self.0 = true;
2735 cx.waker().wake_by_ref();
2736 std::task::Poll::Pending
2737 }
2738 }
2739 }
2740 YieldOnce(false).await;
2741
2742 if cx.checkpoint().is_err() {
2744 return 0; }
2746 2
2747 })
2748 .unwrap();
2749
2750 let waker = std::task::Waker::noop().clone();
2751 let mut ctx = Context::from_waker(&waker);
2752
2753 assert!(t1.poll(&mut ctx).is_ready());
2755
2756 let handles = vec![h1, h2];
2758 let mut race_fut = Box::pin(scope.race_all(&cx, handles));
2759
2760 assert!(race_fut.as_mut().poll(&mut ctx).is_pending());
2767
2768 assert!(t2.poll(&mut ctx).is_pending());
2773
2774 assert!(race_fut.as_mut().poll(&mut ctx).is_pending());
2776
2777 assert!(t2.poll(&mut ctx).is_ready());
2786
2787 match race_fut.as_mut().poll(&mut ctx) {
2790 Poll::Ready(Ok((val, idx))) => {
2791 assert_eq!(val, 1);
2792 assert_eq!(idx, 0);
2793 }
2794 res => unreachable!("Expected Ready(Ok((1, 0))), got {res:?}"),
2795 }
2796 }
2797
2798 #[test]
2799 fn race_surfaces_loser_panic_even_if_winner_succeeds() {
2800 use std::task::Context;
2801
2802 let mut state = RuntimeState::new();
2803 let cx = test_cx();
2804 let region = state.create_root_region(Budget::INFINITE);
2805 let scope = test_scope(region, Budget::INFINITE);
2806
2807 let (h1, mut t1) = scope.spawn(&mut state, &cx, |_| async { 1_i32 }).unwrap();
2808 let (h2, mut t2) = scope
2809 .spawn(&mut state, &cx, |_| async {
2810 std::panic::panic_any("loser panic");
2811 })
2812 .unwrap();
2813
2814 let waker = std::task::Waker::noop().clone();
2815 let mut poll_cx = Context::from_waker(&waker);
2816 assert!(t1.poll(&mut poll_cx).is_ready());
2817 assert!(t2.poll(&mut poll_cx).is_ready());
2818
2819 let result = block_on(scope.race(&cx, h1, h2));
2820 assert!(
2821 matches!(result, Err(JoinError::Panicked(_))),
2822 "loser panic must dominate race result, got {result:?}"
2823 );
2824 }
2825
2826 #[test]
2827 fn race_preserves_winner_panic_over_loser_panic() {
2828 use std::task::{Context, Poll};
2829
2830 let mut state = RuntimeState::new();
2831 let cx = test_cx();
2832 let region = state.create_root_region(Budget::INFINITE);
2833 let scope = test_scope(region, Budget::INFINITE);
2834
2835 let (h1, mut t1) = scope
2836 .spawn(&mut state, &cx, |_| async {
2837 std::panic::panic_any("winner panic");
2838 })
2839 .unwrap();
2840 let (h2, mut t2) = scope
2841 .spawn(&mut state, &cx, |_| {
2842 let mut first_poll = true;
2843 std::future::poll_fn(move |poll_cx| {
2844 if first_poll {
2845 first_poll = false;
2846 poll_cx.waker().wake_by_ref();
2847 Poll::Pending
2848 } else {
2849 std::panic::panic_any("loser panic");
2850 }
2851 })
2852 })
2853 .unwrap();
2854
2855 let waker = std::task::Waker::noop().clone();
2856 let mut poll_cx = Context::from_waker(&waker);
2857 assert!(t1.poll(&mut poll_cx).is_ready());
2858 assert!(t2.poll(&mut poll_cx).is_pending());
2859
2860 let result = block_on(scope.race(&cx, h1, h2));
2861 match result {
2862 Err(JoinError::Panicked(payload)) => {
2863 assert_eq!(payload.message(), "winner panic");
2864 }
2865 other => unreachable!("winner panic must dominate race result, got {other:?}"),
2866 }
2867 }
2868
2869 #[test]
2870 fn race_all_surfaces_simultaneous_loser_panic() {
2871 use std::task::Context;
2872
2873 let mut state = RuntimeState::new();
2874 let cx = test_cx();
2875 let region = state.create_root_region(Budget::INFINITE);
2876 let scope = test_scope(region, Budget::INFINITE);
2877
2878 let (h1, mut t1) = scope.spawn(&mut state, &cx, |_| async { 1_i32 }).unwrap();
2879 let (h2, mut t2) = scope
2880 .spawn(&mut state, &cx, |_| async {
2881 std::panic::panic_any("simultaneous loser panic");
2882 })
2883 .unwrap();
2884 let (h3, mut t3) = scope.spawn(&mut state, &cx, |_| async { 3_i32 }).unwrap();
2885
2886 let waker = std::task::Waker::noop().clone();
2887 let mut poll_cx = Context::from_waker(&waker);
2888 assert!(t1.poll(&mut poll_cx).is_ready());
2889 assert!(t2.poll(&mut poll_cx).is_ready());
2890 assert!(t3.poll(&mut poll_cx).is_ready());
2891
2892 let result = block_on(scope.race_all(&cx, vec![h1, h2, h3]));
2893 assert!(
2894 matches!(result, Err(JoinError::Panicked(_))),
2895 "simultaneous loser panic must dominate race_all result, got {result:?}"
2896 );
2897 }
2898
2899 #[test]
2900 fn race_all_preserves_winner_panic_over_loser_panic() {
2901 use std::task::{Context, Poll};
2902
2903 let mut state = RuntimeState::new();
2904 let cx = test_cx();
2905 let region = state.create_root_region(Budget::INFINITE);
2906 let scope = test_scope(region, Budget::INFINITE);
2907
2908 let (h1, mut t1) = scope
2909 .spawn(&mut state, &cx, |_| async {
2910 std::panic::panic_any("winner panic");
2911 })
2912 .unwrap();
2913 let (h2, mut t2) = scope
2914 .spawn(&mut state, &cx, |_| {
2915 let mut first_poll = true;
2916 std::future::poll_fn(move |poll_cx| {
2917 if first_poll {
2918 first_poll = false;
2919 poll_cx.waker().wake_by_ref();
2920 Poll::Pending
2921 } else {
2922 std::panic::panic_any("loser panic");
2923 }
2924 })
2925 })
2926 .unwrap();
2927
2928 let waker = std::task::Waker::noop().clone();
2929 let mut poll_cx = Context::from_waker(&waker);
2930 assert!(t1.poll(&mut poll_cx).is_ready());
2931 assert!(t2.poll(&mut poll_cx).is_pending());
2932
2933 let result = block_on(scope.race_all(&cx, vec![h1, h2]));
2934 match result {
2935 Err(JoinError::Panicked(payload)) => {
2936 assert_eq!(payload.message(), "winner panic");
2937 }
2938 other => unreachable!("winner panic must dominate race_all result, got {other:?}"),
2939 }
2940 }
2941
2942 #[test]
2943 fn race_all_empty_is_pending() {
2944 let mut state = RuntimeState::new();
2945 let cx = test_cx();
2946 let region = state.create_root_region(Budget::INFINITE);
2947 let scope = test_scope(region, Budget::INFINITE);
2948
2949 let fut = scope.race_all::<i32>(&cx, vec![]);
2950 let waker = std::task::Waker::noop();
2951 let mut poll_cx = std::task::Context::from_waker(waker);
2952 let pinned = std::pin::pin!(fut);
2953 let status = std::future::Future::poll(pinned, &mut poll_cx);
2954 assert!(status.is_pending());
2955 }
2956
2957 #[test]
2965 fn conformance_spawn_creates_trackable_task() {
2966 let mut state = RuntimeState::new();
2968 let cx = test_cx();
2969 let region = state.create_root_region(Budget::INFINITE);
2970 let scope = test_scope(region, Budget::INFINITE);
2971
2972 let (handle, _stored) = scope.spawn(&mut state, &cx, |_| async { 42_i32 }).unwrap();
2973
2974 let task_record = state
2976 .task(handle.task_id())
2977 .expect("spawned task must have a record");
2978
2979 assert_eq!(
2981 task_record.owner, region,
2982 "spawned task must be owned by the spawning region"
2983 );
2984
2985 let region_record = state.region(region).expect("spawning region must exist");
2987 assert!(
2988 region_record.task_ids().contains(&handle.task_id()),
2989 "spawning region must track the spawned task"
2990 );
2991 }
2992
2993 #[test]
2994 fn conformance_spawn_enforces_send_bounds() {
2995 let mut state = RuntimeState::new();
2997 let cx = test_cx();
2998 let region = state.create_root_region(Budget::INFINITE);
2999 let scope = test_scope(region, Budget::INFINITE);
3000
3001 let send_data = String::from("test");
3003 let (handle, _stored) = scope
3004 .spawn(&mut state, &cx, move |_| async move {
3005 send_data.len() })
3007 .unwrap();
3008
3009 let task_record = state
3011 .task(handle.task_id())
3012 .expect("Send task must have a record");
3013 assert_eq!(task_record.owner, region);
3014
3015 }
3018
3019 #[test]
3020 fn conformance_join_awaits_task_completion() {
3021 use std::task::Context;
3024
3025 let mut state = RuntimeState::new();
3026 let cx = test_cx();
3027 let region = state.create_root_region(Budget::INFINITE);
3028 let scope = test_scope(region, Budget::INFINITE);
3029
3030 let (mut handle, mut stored) = scope.spawn(&mut state, &cx, |_| async { 123_i32 }).unwrap();
3031
3032 let waker = std::task::Waker::noop().clone();
3033 let mut poll_cx = Context::from_waker(&waker);
3034
3035 let mut join_fut = std::pin::pin!(handle.join(&cx));
3037 assert!(
3038 join_fut.as_mut().poll(&mut poll_cx).is_pending(),
3039 "join must be pending before task completion"
3040 );
3041
3042 assert!(
3044 stored.poll(&mut poll_cx).is_ready(),
3045 "test task must complete in one poll"
3046 );
3047
3048 match join_fut.as_mut().poll(&mut poll_cx) {
3050 std::task::Poll::Ready(Ok(result)) => {
3051 assert_eq!(result, 123, "join must return the task's result");
3052 }
3053 other => panic!("join must be Ready(Ok(123)) after task completion, got {other:?}"),
3054 }
3055 }
3056
3057 #[test]
3058 fn conformance_child_region_task_isolation() {
3059 use std::task::Context;
3062
3063 let mut state = RuntimeState::new();
3064 let cx = test_cx();
3065 let parent_region = state.create_root_region(Budget::INFINITE);
3066 let scope = test_scope(parent_region, Budget::INFINITE);
3067
3068 let outcome = block_on(scope.region(
3069 &mut state,
3070 &cx,
3071 crate::types::policy::FailFast,
3072 |child_scope, state| {
3073 let child_region = child_scope.region_id();
3074
3075 let (handle, mut stored) = child_scope
3077 .spawn(state, &cx, |_| async { 456_i32 })
3078 .expect("spawn in child region must succeed");
3079
3080 let task_record = state
3082 .task(handle.task_id())
3083 .expect("child task must have a record");
3084 let child_owns = task_record.owner == child_region;
3085 let parent_owns = task_record.owner == parent_region;
3086
3087 let parent_tracks = state
3089 .region(parent_region)
3090 .is_some_and(|r| r.task_ids().contains(&handle.task_id()));
3091 let child_tracks = state
3092 .region(child_region)
3093 .is_some_and(|r| r.task_ids().contains(&handle.task_id()));
3094
3095 let waker = std::task::Waker::noop().clone();
3097 let mut poll_cx = Context::from_waker(&waker);
3098 if let std::task::Poll::Ready(outcome) = stored.poll(&mut poll_cx) {
3099 if let Some(task) = state.task_mut(handle.task_id()) {
3100 task.complete(outcome.map_err(|_| {
3101 crate::error::Error::new(crate::error::ErrorKind::Internal)
3102 }));
3103 }
3104 let _ = state.task_completed(handle.task_id());
3105 }
3106
3107 std::future::ready(Outcome::Ok((
3108 child_owns,
3109 parent_owns,
3110 child_tracks,
3111 parent_tracks,
3112 )))
3113 },
3114 ))
3115 .expect("child region must complete");
3116
3117 let (child_owns, parent_owns, child_tracks, parent_tracks) = match outcome {
3118 Outcome::Ok(tuple) => tuple,
3119 other => panic!("expected Ok(ownership_data), got {other:?}"),
3120 };
3121
3122 assert!(
3123 child_owns,
3124 "task spawned in child region must be owned by child"
3125 );
3126 assert!(
3127 !parent_owns,
3128 "task spawned in child region must NOT be owned by parent"
3129 );
3130 assert!(child_tracks, "child region must track its spawned tasks");
3131 assert!(!parent_tracks, "parent region must NOT track child's tasks");
3132 }
3133
3134 #[test]
3135 fn conformance_capability_inheritance() {
3136 use crate::cx::macaroon::MacaroonToken;
3138 use crate::cx::registry::RegistryHandle;
3139 use crate::remote::{NodeId, RemoteCap};
3140 use crate::security::key::AuthKey;
3141 use crate::types::SystemPressure;
3142 use std::sync::Arc;
3143 use std::task::Context;
3144
3145 let mut state = RuntimeState::new();
3146 let region = state.create_root_region(Budget::INFINITE);
3147 let scope = test_scope(region, Budget::INFINITE);
3148
3149 let registry = crate::cx::NameRegistry::new();
3151 let registry_handle = RegistryHandle::new(Arc::new(registry));
3152 let parent_registry_arc = registry_handle.as_arc();
3153 let parent_io_cap: Arc<dyn crate::io::IoCap> =
3154 Arc::new(crate::io::LabIoCap::new_for_tests());
3155 let parent_pressure = Arc::new(SystemPressure::new());
3156 parent_pressure.set_headroom(0.25);
3157 let auth_key = AuthKey::from_seed(7);
3158 let token = MacaroonToken::mint(&auth_key, "scope:spawn", "cx/scope");
3159
3160 let parent_cx = Cx::new_with_io(
3161 crate::types::RegionId::new_for_test(0, 0),
3162 crate::types::TaskId::new_for_test(0, 0),
3163 Budget::INFINITE,
3164 None,
3165 None,
3166 Some(Arc::clone(&parent_io_cap)),
3167 None,
3168 )
3169 .with_registry_handle(Some(registry_handle))
3170 .with_remote_cap(RemoteCap::new().with_local_node(NodeId::new("test-node")))
3171 .with_pressure(Arc::clone(&parent_pressure))
3172 .with_macaroon(token);
3173 let parent_macaroon = parent_cx
3174 .macaroon_handle()
3175 .expect("parent must retain macaroon capability");
3176
3177 let mut handle = scope
3178 .spawn_registered(&mut state, &parent_cx, move |child_cx| async move {
3179 let child_registry = child_cx
3181 .registry_handle()
3182 .expect("child must inherit registry capability");
3183 let same_registry = Arc::ptr_eq(&child_registry.as_arc(), &parent_registry_arc);
3184
3185 let child_remote = child_cx
3187 .remote()
3188 .expect("child must inherit remote capability");
3189 let node_name = child_remote.local_node().as_str().to_owned();
3190
3191 let child_io_cap = child_cx
3193 .io_cap_handle()
3194 .expect("child must inherit I/O capability");
3195 let same_io_cap = Arc::ptr_eq(&child_io_cap, &parent_io_cap);
3196
3197 let child_pressure = child_cx
3199 .pressure_handle()
3200 .expect("child must inherit system pressure");
3201 let same_pressure = Arc::ptr_eq(&child_pressure, &parent_pressure);
3202
3203 let child_macaroon = child_cx
3205 .macaroon_handle()
3206 .expect("child must inherit macaroon capability");
3207 let same_macaroon = Arc::ptr_eq(&child_macaroon, &parent_macaroon);
3208
3209 let has_timer = child_cx.has_timer();
3211
3212 (
3213 same_registry,
3214 node_name,
3215 same_io_cap,
3216 same_pressure,
3217 same_macaroon,
3218 has_timer,
3219 )
3220 })
3221 .unwrap();
3222
3223 let waker = std::task::Waker::noop().clone();
3225 let mut poll_cx = Context::from_waker(&waker);
3226
3227 let stored = state
3228 .get_stored_future(handle.task_id())
3229 .expect("spawn_registered must store the task");
3230 assert!(stored.poll(&mut poll_cx).is_ready());
3231
3232 let mut join_fut = std::pin::pin!(handle.join(&parent_cx));
3233 match join_fut.as_mut().poll(&mut poll_cx) {
3234 std::task::Poll::Ready(Ok((
3235 same_registry,
3236 node_name,
3237 same_io_cap,
3238 same_pressure,
3239 same_macaroon,
3240 has_timer,
3241 ))) => {
3242 assert!(
3243 same_registry,
3244 "child must inherit exact same registry instance"
3245 );
3246 assert_eq!(
3247 node_name, "test-node",
3248 "child must inherit remote capability"
3249 );
3250 assert!(same_io_cap, "child must inherit exact same I/O capability");
3251 assert!(
3252 same_pressure,
3253 "child must inherit exact same system pressure handle"
3254 );
3255 assert!(
3256 same_macaroon,
3257 "child must inherit exact same macaroon capability"
3258 );
3259 assert_eq!(
3260 has_timer,
3261 parent_cx.has_timer(),
3262 "child timer capability should stay consistent with the runtime-backed parent"
3263 );
3264 }
3265 other => panic!("capability inheritance test failed: {other:?}"),
3266 }
3267 }
3268
3269 #[test]
3270 fn conformance_task_cancellation_propagation() {
3271 use std::task::Context;
3274
3275 let mut state = RuntimeState::new();
3276 let cx = test_cx();
3277 let region = state.create_root_region(Budget::INFINITE);
3278 let scope = test_scope(region, Budget::INFINITE);
3279
3280 let (mut handle, mut stored) = scope
3281 .spawn(&mut state, &cx, |cx| async move {
3282 if cx.checkpoint().is_err() {
3284 "cancelled"
3285 } else {
3286 "completed"
3287 }
3288 })
3289 .unwrap();
3290
3291 handle.abort();
3293
3294 let waker = std::task::Waker::noop().clone();
3295 let mut poll_cx = Context::from_waker(&waker);
3296
3297 assert!(
3299 stored.poll(&mut poll_cx).is_ready(),
3300 "cancelled task must still complete"
3301 );
3302
3303 let mut join_fut = std::pin::pin!(handle.join(&cx));
3305 match join_fut.as_mut().poll(&mut poll_cx) {
3306 std::task::Poll::Ready(Ok(result)) => {
3307 assert_eq!(
3308 result, "cancelled",
3309 "cancelled task must observe cancellation via checkpoint()"
3310 );
3311 }
3312 other => panic!("cancelled task join failed: {other:?}"),
3313 }
3314 }
3315
3316 #[test]
3317 fn metamorphic_nested_scope_cancellation_closes_descendants_without_spawn_leaks() {
3318 use std::task::{Context, Poll};
3319
3320 struct YieldOnce(bool);
3321
3322 impl std::future::Future for YieldOnce {
3323 type Output = ();
3324
3325 fn poll(
3326 mut self: std::pin::Pin<&mut Self>,
3327 cx: &mut std::task::Context<'_>,
3328 ) -> Poll<()> {
3329 if self.0 {
3330 Poll::Ready(())
3331 } else {
3332 self.0 = true;
3333 cx.waker().wake_by_ref();
3334 Poll::Pending
3335 }
3336 }
3337 }
3338
3339 let mut state = RuntimeState::new();
3340 let cx = test_cx();
3341 let root = state.create_root_region(Budget::INFINITE);
3342 let child = state
3343 .create_child_region(root, Budget::INFINITE)
3344 .expect("child region");
3345 let grandchild = state
3346 .create_child_region(child, Budget::INFINITE)
3347 .expect("grandchild region");
3348 let child_scope = test_scope(child, Budget::INFINITE);
3349 let grandchild_scope = test_scope(grandchild, Budget::INFINITE);
3350 let finalizer_log = Arc::new(std::sync::Mutex::new(Vec::new()));
3351
3352 let child_log = Arc::clone(&finalizer_log);
3353 assert!(
3354 child_scope.defer_sync(&mut state, move || {
3355 child_log
3356 .lock()
3357 .expect("child finalizer log poisoned")
3358 .push("child");
3359 }),
3360 "child finalizer should register before cancellation"
3361 );
3362
3363 let grandchild_log = Arc::clone(&finalizer_log);
3364 assert!(
3365 grandchild_scope.defer_sync(&mut state, move || {
3366 grandchild_log
3367 .lock()
3368 .expect("grandchild finalizer log poisoned")
3369 .push("grandchild");
3370 }),
3371 "grandchild finalizer should register before cancellation"
3372 );
3373
3374 let mut child_handle = child_scope
3375 .spawn_registered(&mut state, &cx, |task_cx| async move {
3376 YieldOnce(false).await;
3377 if task_cx.checkpoint().is_err() {
3378 "child_cancelled"
3379 } else {
3380 "child_completed"
3381 }
3382 })
3383 .expect("spawn child task");
3384 let child_task_id = child_handle.task_id();
3385
3386 let mut grandchild_handle = grandchild_scope
3387 .spawn_registered(&mut state, &cx, |task_cx| async move {
3388 YieldOnce(false).await;
3389 if task_cx.checkpoint().is_err() {
3390 "grandchild_cancelled"
3391 } else {
3392 "grandchild_completed"
3393 }
3394 })
3395 .expect("spawn grandchild task");
3396 let grandchild_task_id = grandchild_handle.task_id();
3397
3398 let cancel_reason = CancelReason::shutdown().with_region(root);
3399 let cancelled = state.cancel_request(root, &cancel_reason, None);
3400 assert!(
3401 cancelled
3402 .iter()
3403 .any(|(task_id, _)| *task_id == child_task_id),
3404 "parent cancellation must reach child task"
3405 );
3406 assert!(
3407 cancelled
3408 .iter()
3409 .any(|(task_id, _)| *task_id == grandchild_task_id),
3410 "parent cancellation must reach grandchild task"
3411 );
3412
3413 let grandchild_tasks_before_failed_spawn = state
3414 .region(grandchild)
3415 .expect("grandchild region missing")
3416 .task_count();
3417 let live_tasks_before_failed_spawn = state.live_task_count();
3418 let failed_spawn = grandchild_scope.spawn(&mut state, &cx, |_| async { 99_u8 });
3419 let grandchild_tasks_after_failed_spawn = state
3420 .region(grandchild)
3421 .expect("grandchild region missing after failed spawn")
3422 .task_count();
3423 let live_tasks_after_failed_spawn = state.live_task_count();
3424
3425 let waker = std::task::Waker::noop().clone();
3426 let mut poll_cx = Context::from_waker(&waker);
3427 {
3428 let stored = state
3429 .get_stored_future(grandchild_task_id)
3430 .expect("grandchild stored task");
3431 let poll_result = stored.poll(&mut poll_cx);
3432 assert!(
3433 poll_result.is_pending(),
3434 "grandchild task should yield once before observing cancellation"
3435 );
3436 }
3437 {
3438 let stored = state
3439 .get_stored_future(grandchild_task_id)
3440 .expect("grandchild stored task");
3441 let poll_result = stored.poll(&mut poll_cx);
3442 let task_outcome = match poll_result {
3443 Poll::Ready(Outcome::Ok(())) => Outcome::Ok(()),
3444 Poll::Ready(Outcome::Panicked(payload)) => Outcome::Panicked(payload),
3445 other => panic!(
3446 "grandchild task should complete once cancellation is observed: {other:?}"
3447 ),
3448 };
3449 if let Some(task_record) = state.task_mut(grandchild_task_id) {
3450 task_record.complete(task_outcome);
3451 }
3452 }
3453 let _ = state.task_completed(grandchild_task_id);
3454 state.advance_region_state(grandchild);
3455
3456 let mut grandchild_join_fut = std::pin::pin!(grandchild_handle.join(&cx));
3457 let grandchild_result = match grandchild_join_fut.as_mut().poll(&mut poll_cx) {
3458 Poll::Ready(Ok(result)) => result,
3459 other => panic!("grandchild cancellation join should succeed: {other:?}"),
3460 };
3461
3462 {
3463 let stored = state
3464 .get_stored_future(child_task_id)
3465 .expect("child stored task");
3466 let poll_result = stored.poll(&mut poll_cx);
3467 assert!(
3468 poll_result.is_pending(),
3469 "child task should yield once before observing cancellation"
3470 );
3471 }
3472 {
3473 let stored = state
3474 .get_stored_future(child_task_id)
3475 .expect("child stored task");
3476 let poll_result = stored.poll(&mut poll_cx);
3477 let task_outcome = match poll_result {
3478 Poll::Ready(Outcome::Ok(())) => Outcome::Ok(()),
3479 Poll::Ready(Outcome::Panicked(payload)) => Outcome::Panicked(payload),
3480 other => {
3481 panic!("child task should complete once cancellation is observed: {other:?}")
3482 }
3483 };
3484 if let Some(task_record) = state.task_mut(child_task_id) {
3485 task_record.complete(task_outcome);
3486 }
3487 }
3488 let _ = state.task_completed(child_task_id);
3489 state.advance_region_state(child);
3490
3491 let mut child_join_fut = std::pin::pin!(child_handle.join(&cx));
3492 let child_result = match child_join_fut.as_mut().poll(&mut poll_cx) {
3493 Poll::Ready(Ok(result)) => result,
3494 other => panic!("child cancellation join should succeed: {other:?}"),
3495 };
3496
3497 assert_eq!(child_result, "child_cancelled");
3498 assert_eq!(grandchild_result, "grandchild_cancelled");
3499 assert!(
3500 matches!(failed_spawn, Err(SpawnError::RegionClosed(id)) if id == grandchild),
3501 "nested spawn after parent cancellation must fail against the closing grandchild region"
3502 );
3503 assert_eq!(
3504 grandchild_tasks_before_failed_spawn, grandchild_tasks_after_failed_spawn,
3505 "failed spawn after cancellation must not leak task membership into the grandchild region"
3506 );
3507 assert_eq!(
3508 live_tasks_before_failed_spawn, live_tasks_after_failed_spawn,
3509 "failed spawn after cancellation must not inflate runtime task count"
3510 );
3511 assert_eq!(
3512 *finalizer_log.lock().expect("finalizer log poisoned"), vec!["grandchild", "child"],
3514 "nested scope finalizers must run in reverse scope creation order"
3515 );
3516 assert!(
3517 state.region(grandchild).is_none(),
3518 "grandchild region should be reclaimed after close"
3519 );
3520 assert!(
3521 state.region(child).is_none(),
3522 "child region should be reclaimed after close"
3523 );
3524 }
3525
3526 #[test]
3527 fn conformance_race_loser_drain_invariant() {
3528 use std::task::{Context, Poll};
3531
3532 let mut state = RuntimeState::new();
3533 let cx = test_cx();
3534 let region = state.create_root_region(Budget::INFINITE);
3535 let scope = test_scope(region, Budget::INFINITE);
3536
3537 let (winner_handle, mut winner_stored) = scope
3539 .spawn(&mut state, &cx, |_| async { "winner" })
3540 .unwrap();
3541
3542 let (loser_handle, mut loser_stored) = scope
3544 .spawn(&mut state, &cx, |cx| async move {
3545 struct YieldOnce(bool);
3547 impl std::future::Future for YieldOnce {
3548 type Output = ();
3549 fn poll(
3550 mut self: std::pin::Pin<&mut Self>,
3551 cx: &mut std::task::Context<'_>,
3552 ) -> std::task::Poll<()> {
3553 if self.0 {
3554 std::task::Poll::Ready(())
3555 } else {
3556 self.0 = true;
3557 cx.waker().wake_by_ref();
3558 std::task::Poll::Pending
3559 }
3560 }
3561 }
3562 YieldOnce(false).await;
3563
3564 if cx.checkpoint().is_err() {
3566 "loser_cancelled"
3567 } else {
3568 "loser_completed"
3569 }
3570 })
3571 .unwrap();
3572
3573 let waker = std::task::Waker::noop().clone();
3574 let mut poll_cx = Context::from_waker(&waker);
3575
3576 assert!(winner_stored.poll(&mut poll_cx).is_ready());
3578
3579 let handles = vec![winner_handle, loser_handle];
3581 let mut race_fut = std::pin::pin!(scope.race_all(&cx, handles));
3582
3583 assert!(
3585 race_fut.as_mut().poll(&mut poll_cx).is_pending(),
3586 "race must wait for loser to be drained"
3587 );
3588
3589 assert!(loser_stored.poll(&mut poll_cx).is_pending());
3591
3592 assert!(race_fut.as_mut().poll(&mut poll_cx).is_pending());
3594
3595 assert!(loser_stored.poll(&mut poll_cx).is_ready());
3597
3598 match race_fut.as_mut().poll(&mut poll_cx) {
3600 Poll::Ready(Ok((result, winner_index))) => {
3601 assert_eq!(result, "winner", "race must return winner result");
3602 assert_eq!(winner_index, 0, "winner index must be correct");
3603 }
3604 other => panic!("race must complete after loser drain: {other:?}"),
3605 }
3606 }
3607
3608 #[test]
3609 fn conformance_region_quiescence_on_empty() {
3610 let mut state = RuntimeState::new();
3612 let cx = test_cx();
3613 let parent_region = state.create_root_region(Budget::INFINITE);
3614 let scope = test_scope(parent_region, Budget::INFINITE);
3615
3616 let outcome = block_on(scope.region(
3618 &mut state,
3619 &cx,
3620 crate::types::policy::FailFast,
3621 |_child_scope, _state| {
3622 std::future::ready(Outcome::Ok("empty_region_completed"))
3624 },
3625 ))
3626 .expect("empty child region must complete");
3627
3628 match outcome {
3629 Outcome::Ok(result) => {
3630 assert_eq!(
3631 result, "empty_region_completed",
3632 "empty region must reach quiescence immediately"
3633 );
3634 }
3635 other => panic!("empty region must complete successfully: {other:?}"),
3636 }
3637
3638 let parent_record = state
3640 .region(parent_region)
3641 .expect("parent region must exist");
3642 assert!(
3643 parent_record.child_ids().is_empty(),
3644 "completed child region must be removed from parent"
3645 );
3646 }
3647
3648 #[test]
3649 fn region_close_future_wakes_all_registered_waiters() {
3650 use std::sync::atomic::{AtomicUsize, Ordering};
3651 use std::task::{Context, Waker};
3652
3653 struct CountWaker(Arc<AtomicUsize>);
3654
3655 impl std::task::Wake for CountWaker {
3656 fn wake(self: Arc<Self>) {
3657 self.0.fetch_add(1, Ordering::SeqCst);
3658 }
3659
3660 fn wake_by_ref(self: &Arc<Self>) {
3661 self.0.fetch_add(1, Ordering::SeqCst);
3662 }
3663 }
3664
3665 let notify = Arc::new(parking_lot::Mutex::new(
3666 crate::record::region::RegionCloseState {
3667 closed: false,
3668 waiters: Default::default(),
3669 },
3670 ));
3671 let wake_count_a = Arc::new(AtomicUsize::new(0));
3672 let wake_count_b = Arc::new(AtomicUsize::new(0));
3673 let waker_a = Waker::from(Arc::new(CountWaker(Arc::clone(&wake_count_a))));
3674 let waker_b = Waker::from(Arc::new(CountWaker(Arc::clone(&wake_count_b))));
3675 let mut cx_a = Context::from_waker(&waker_a);
3676 let mut cx_b = Context::from_waker(&waker_b);
3677 let mut future_a = RegionCloseFuture {
3678 state: Arc::clone(¬ify),
3679 };
3680 let mut future_b = RegionCloseFuture { state: notify };
3681
3682 assert!(Pin::new(&mut future_a).poll(&mut cx_a).is_pending());
3683 assert!(Pin::new(&mut future_b).poll(&mut cx_b).is_pending());
3684
3685 let waiters = {
3686 let mut state = future_a.state.lock();
3687 state.closed = true;
3688 std::mem::take(&mut state.waiters)
3689 };
3690 for waker in waiters {
3691 waker.wake();
3692 }
3693
3694 assert_eq!(wake_count_a.load(Ordering::SeqCst), 1);
3695 assert_eq!(wake_count_b.load(Ordering::SeqCst), 1);
3696 assert!(Pin::new(&mut future_a).poll(&mut cx_a).is_ready());
3697 assert!(Pin::new(&mut future_b).poll(&mut cx_b).is_ready());
3698 }
3699
3700 #[test]
3701 fn conformance_spawn_into_closed_region_fails() {
3702 let mut state = RuntimeState::new();
3704 let cx = test_cx();
3705 let region = state.create_root_region(Budget::INFINITE);
3706 let scope = test_scope(region, Budget::INFINITE);
3707
3708 let region_record = state.region_mut(region).expect("region must exist");
3710 region_record.begin_close(None);
3711
3712 let spawn_result = scope.spawn(&mut state, &cx, |_| async { 42 });
3714
3715 assert!(
3716 matches!(spawn_result, Err(SpawnError::RegionClosed(_))),
3717 "spawning into closed region must fail with RegionClosed error"
3718 );
3719
3720 assert!(
3722 state.tasks_is_empty() || state.region(region).is_none_or(|r| r.task_ids().is_empty()),
3723 "failed spawn must not create orphaned tasks"
3724 );
3725 }
3726
3727 #[test]
3728 fn conformance_join_multiple_tasks_preserves_results() {
3729 use std::task::{Context, Poll};
3732
3733 let mut state = RuntimeState::new();
3734 let cx = test_cx();
3735 let region = state.create_root_region(Budget::INFINITE);
3736 let scope = test_scope(region, Budget::INFINITE);
3737
3738 let (h1, mut t1) = scope.spawn(&mut state, &cx, |_| async { 100_i32 }).unwrap();
3740 let (h2, mut t2) = scope.spawn(&mut state, &cx, |_| async { 200_i32 }).unwrap();
3741 let (h3, mut t3) = scope.spawn(&mut state, &cx, |_| async { 300_i32 }).unwrap();
3742
3743 let waker = std::task::Waker::noop().clone();
3744 let mut poll_cx = Context::from_waker(&waker);
3745
3746 assert!(t1.poll(&mut poll_cx).is_ready());
3748 assert!(t2.poll(&mut poll_cx).is_ready());
3749 assert!(t3.poll(&mut poll_cx).is_ready());
3750
3751 let handles = vec![h1, h2, h3];
3753 let mut join_all_fut = std::pin::pin!(scope.join_all(&cx, handles));
3754
3755 match join_all_fut.as_mut().poll(&mut poll_cx) {
3756 Poll::Ready(results) => {
3757 assert_eq!(results.len(), 3, "join_all must return all results");
3758
3759 assert_eq!(
3761 results[0].as_ref().unwrap(),
3762 &100,
3763 "first task result must be preserved"
3764 );
3765 assert_eq!(
3766 results[1].as_ref().unwrap(),
3767 &200,
3768 "second task result must be preserved"
3769 );
3770 assert_eq!(
3771 results[2].as_ref().unwrap(),
3772 &300,
3773 "third task result must be preserved"
3774 );
3775 }
3776 other @ Poll::Pending => panic!("join_all must complete with all results: {other:?}"),
3777 }
3778 }
3779
3780 #[test]
3781 fn conformance_panic_propagation_through_join() {
3782 use std::task::{Context, Poll};
3785
3786 let mut state = RuntimeState::new();
3787 let cx = test_cx();
3788 let region = state.create_root_region(Budget::INFINITE);
3789 let scope = test_scope(region, Budget::INFINITE);
3790
3791 let (mut handle, mut stored) = scope
3793 .spawn(&mut state, &cx, |_| async {
3794 std::panic::panic_any("test_panic_message");
3795 })
3796 .unwrap();
3797
3798 let waker = std::task::Waker::noop().clone();
3799 let mut poll_cx = Context::from_waker(&waker);
3800
3801 match stored.poll(&mut poll_cx) {
3803 Poll::Ready(crate::types::Outcome::Panicked(_)) => {
3804 }
3806 other => panic!("panicking task must complete with Panicked outcome: {other:?}"),
3807 }
3808
3809 let mut join_fut = std::pin::pin!(handle.join(&cx));
3811 match join_fut.as_mut().poll(&mut poll_cx) {
3812 Poll::Ready(Err(JoinError::Panicked(payload))) => {
3813 assert_eq!(
3814 payload.message(),
3815 "test_panic_message",
3816 "join must preserve panic payload message"
3817 );
3818 }
3819 other => panic!("join of panicked task must return JoinError::Panicked: {other:?}"),
3820 }
3821 }
3822
3823 #[test]
3831 fn region_factory_panic_after_spawn_cleans_up_orphan_task_records() {
3832 let mut state = RuntimeState::new();
3833 let cx = test_cx();
3834 let parent = state.create_root_region(Budget::INFINITE);
3835 let scope = test_scope(parent, Budget::INFINITE);
3836
3837 let tasks_before = state.tasks_iter().count();
3838
3839 let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
3842 block_on(scope.region_with_budget(
3843 &mut state,
3844 &cx,
3845 Budget::INFINITE,
3846 crate::types::policy::FailFast,
3847 |child, state| {
3848 let (_handle, _stored) = child
3851 .spawn(state, &cx, |_| async { 42_i32 })
3852 .expect("spawn must succeed before the panic");
3853 std::panic::panic_any("factory panic after spawn");
3855 #[allow(unreachable_code)]
3856 std::future::ready(Outcome::Ok(0_i32))
3857 },
3858 ))
3859 }));
3860
3861 assert!(
3862 result.is_err(),
3863 "region_with_budget must re-raise the factory panic",
3864 );
3865
3866 let tasks_after = state.tasks_iter().count();
3870 assert_eq!(
3871 tasks_after, tasks_before,
3872 "post-panic tasks_iter count must match pre-call count \
3873 (orphan TaskRecord not cleaned up: before={tasks_before} after={tasks_after})",
3874 );
3875 }
3876
3877 #[test]
3878 fn spawn_local_wake_state_single_notify() {
3879 use std::sync::Arc;
3882
3883 let mut state = RuntimeState::new();
3884 let cx = test_cx();
3885 let region = state.create_root_region(Budget::INFINITE);
3886 let scope = test_scope(region, Budget::INFINITE);
3887
3888 let local_ready = Arc::new(parking_lot::Mutex::new(std::collections::VecDeque::new()));
3889 let _local_ready_guard =
3890 crate::runtime::scheduler::three_lane::ScopedLocalReady::new(Arc::clone(&local_ready));
3891 let _worker_guard = crate::runtime::scheduler::three_lane::ScopedWorkerId::new(1);
3892
3893 let handle = scope
3894 .spawn_local(&mut state, &cx, |_| async move { "test_result" })
3895 .unwrap();
3896
3897 let task_record = state.task(handle.task_id()).unwrap();
3899
3900 let second_notify_result = task_record.wake_state.notify();
3903 assert!(
3904 !second_notify_result,
3905 "Second notify() should return false - task already notified during spawn_local"
3906 );
3907
3908 let queued = {
3910 let queue = local_ready.lock();
3911 queue.contains(&handle.task_id())
3912 };
3913 assert!(queued, "spawn_local should enqueue task into local_ready");
3914 }
3915}