1use std::{
2 num::NonZeroUsize,
3 ops::Deref,
4 time::{Duration, Instant},
5};
6
7use super::{
8 CachedPocketIcBaseline, ControllerSnapshotError, SnapshotRestoreFunding,
9 bounded_pool::{BoundedSlotLease, BoundedSlotPool, Slot},
10};
11
12#[non_exhaustive]
14#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
15pub enum ResetDomainKind {
16 PocketIcTime,
18 ExtraCanisters,
20 PendingMessages,
22 SubnetState,
24 ExternalResources,
26}
27
28#[non_exhaustive]
30#[derive(Clone, Copy, Debug, Eq, PartialEq)]
31pub enum TimeResetPolicy {
32 PreserveCurrent,
34 RestoreBaseline,
36 RebuildOnMutation,
38}
39
40#[non_exhaustive]
42#[derive(Clone, Copy, Debug, Eq, PartialEq)]
43pub enum ExtraCanisterPolicy {
44 RequireBaselineSet,
46 RemoveTracked,
48 RebuildOnChange,
50}
51
52#[non_exhaustive]
54#[derive(Clone, Copy, Debug, Eq, PartialEq)]
55pub enum StateResetPolicy {
56 ResetByRecipe,
58 ValidateUnchanged,
60 IrrelevantByRecipeContract,
62 RebuildOnChange,
64}
65
66#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
94pub struct ResetPolicies {
95 pub pocket_ic_time: Option<TimeResetPolicy>,
97 pub extra_canisters: Option<ExtraCanisterPolicy>,
99 pub pending_messages: Option<StateResetPolicy>,
101 pub subnet_state: Option<StateResetPolicy>,
103 pub external_resources: Option<StateResetPolicy>,
105}
106
107#[derive(Clone, Copy, Debug, Eq, PartialEq)]
109pub struct ResetRequirements {
110 pub funding: SnapshotRestoreFunding,
112 pub policies: ResetPolicies,
114}
115
116#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
121pub struct ResetReceipt {
122 pub policies: ResetPolicies,
124}
125
126#[non_exhaustive]
128#[derive(Clone, Debug, Eq, PartialEq, thiserror::Error)]
129pub enum BaselinePoolContractError {
130 #[error("required reset domain {domain:?} was not achieved")]
132 MissingResetDomain {
133 domain: ResetDomainKind,
135 },
136 #[error("reset achievement {achievement:?} does not satisfy {requirement:?} for {domain:?}")]
138 ResetPolicyMismatch {
139 domain: ResetDomainKind,
141 requirement: ResetPolicies,
143 achievement: ResetPolicies,
145 },
146}
147
148#[non_exhaustive]
150#[derive(Clone, Debug, Eq, PartialEq)]
151pub enum PreparedBaseline {
152 Built,
154 Restored {
156 reset: ResetReceipt,
158 },
159}
160
161#[non_exhaustive]
163#[derive(Clone, Copy, Debug, Eq, PartialEq)]
164pub enum BaselinePreparationStage {
165 Build,
167 RestoreCanisters,
169 ResetNonSnapshotState,
171 DriveToReadiness,
173 ValidateBuilt,
175 ValidateRestored,
177}
178
179#[non_exhaustive]
181#[derive(Clone, Debug, Eq, PartialEq)]
182pub enum RebuildReason {
183 DeadPocketIcTransport,
185 SnapshotRestoreFailure,
187 ResetFailure,
189 ReadinessFailure,
191 ResetCoverageMismatch,
193 InvariantValidationFailure,
195 ExplicitLeaseInvalidation,
197 UnwindWhileLeased,
199 PreparationCancelled,
201 RecipeClassified {
203 code: String,
205 },
206}
207
208#[non_exhaustive]
210#[derive(Clone, Debug, Eq, PartialEq)]
211pub enum FailureDisposition {
212 Fatal,
214 Rebuild(RebuildReason),
216}
217
218#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
220pub struct BaselinePoolTimings {
221 wait: Duration,
222 build: Option<Duration>,
223 restore: Option<Duration>,
224 reset: Option<Duration>,
225 readiness: Option<Duration>,
226 validation: Option<Duration>,
227 stale_teardown: Option<Duration>,
228 total: Duration,
229}
230
231#[derive(Clone, Debug, Eq, PartialEq)]
233pub struct BaselinePoolOutcome {
234 pub slot: usize,
236 pub timings: BaselinePoolTimings,
238 pub disposition: BaselinePoolDisposition,
240}
241
242#[non_exhaustive]
244#[derive(Clone, Debug, Eq, PartialEq)]
245pub enum BaselinePoolDisposition {
246 Built,
248 Restored,
250 Rebuilt(RebuildReason),
252}
253
254#[non_exhaustive]
256#[derive(Debug, thiserror::Error)]
257pub enum BaselinePoolPreparationError<E> {
258 #[error("baseline {stage} failed: {source}")]
260 Recipe {
261 stage: BaselinePreparationStage,
263 source: E,
265 },
266 #[error("baseline canister restore failed: {0}")]
268 Restore(#[source] ControllerSnapshotError),
269 #[error("baseline pool contract failed: {0}")]
271 Contract(#[source] BaselinePoolContractError),
272}
273
274#[non_exhaustive]
276#[derive(Debug)]
277pub enum BaselinePoolError<E> {
278 Preparation {
280 error: BaselinePoolPreparationError<E>,
282 timings: Box<BaselinePoolTimings>,
284 },
285 RecoveryFailed {
287 original: Box<BaselinePoolPreparationError<E>>,
289 rebuild: Box<BaselinePoolPreparationError<E>>,
291 timings: Box<BaselinePoolTimings>,
293 },
294}
295
296enum PreparationFailure<E> {
298 Single(BaselinePoolPreparationError<E>),
299 Recovery {
300 original: Box<BaselinePoolPreparationError<E>>,
301 rebuild: Box<BaselinePoolPreparationError<E>>,
302 },
303}
304
305impl<E> PreparationFailure<E> {
306 fn with_timings(self, timings: BaselinePoolTimings) -> BaselinePoolError<E> {
307 let timings = Box::new(timings);
308 match self {
309 Self::Single(error) => BaselinePoolError::Preparation { error, timings },
310 Self::Recovery { original, rebuild } => BaselinePoolError::RecoveryFailed {
311 original,
312 rebuild,
313 timings,
314 },
315 }
316 }
317}
318
319pub trait PocketIcBaselineRecipe: Send + Sync + 'static {
321 type Metadata: Send + Sync + 'static;
323 type Error: std::error::Error + Send + Sync + 'static;
325
326 fn reset_requirements(&self) -> &ResetRequirements;
328
329 fn build(
336 &self,
337 ) -> impl std::future::Future<
338 Output = Result<CachedPocketIcBaseline<Self::Metadata>, Self::Error>,
339 > + Send;
340
341 fn prepare_restore(
351 &self,
352 _baseline: &CachedPocketIcBaseline<Self::Metadata>,
353 ) -> impl std::future::Future<Output = Result<(), Self::Error>> + Send {
354 async { Ok(()) }
355 }
356
357 fn reset_non_snapshot_state(
362 &self,
363 baseline: &CachedPocketIcBaseline<Self::Metadata>,
364 ) -> impl std::future::Future<Output = Result<ResetReceipt, Self::Error>> + Send;
365
366 fn drive_to_readiness(
374 &self,
375 baseline: &CachedPocketIcBaseline<Self::Metadata>,
376 ) -> impl std::future::Future<Output = Result<(), Self::Error>> + Send;
377
378 fn validate(
384 &self,
385 baseline: &CachedPocketIcBaseline<Self::Metadata>,
386 preparation: &PreparedBaseline,
387 ) -> impl std::future::Future<Output = Result<(), Self::Error>> + Send;
388
389 fn classify_failure(
391 &self,
392 stage: BaselinePreparationStage,
393 _error: &(dyn std::error::Error + 'static),
394 ) -> FailureDisposition {
395 FailureDisposition::Rebuild(stage.default_rebuild_reason())
396 }
397}
398
399pub struct CachedPocketIcBaselinePool<R>
411where
412 R: PocketIcBaselineRecipe,
413{
414 recipe: R,
415 slots: BoundedSlotPool<CachedPocketIcBaseline<R::Metadata>, RebuildReason>,
416}
417
418pub struct CachedPocketIcBaselinePoolGuard<'a, R>
420where
421 R: PocketIcBaselineRecipe,
422{
423 slot: BoundedSlotLease<'a, CachedPocketIcBaseline<R::Metadata>, RebuildReason>,
424}
425
426impl ResetRequirements {
427 fn verify(&self, receipt: ResetReceipt) -> Result<(), BaselinePoolContractError> {
428 self.verify_policy(
429 receipt,
430 ResetDomainKind::PocketIcTime,
431 self.policies.pocket_ic_time,
432 receipt.policies.pocket_ic_time,
433 )?;
434 self.verify_policy(
435 receipt,
436 ResetDomainKind::ExtraCanisters,
437 self.policies.extra_canisters,
438 receipt.policies.extra_canisters,
439 )?;
440 self.verify_policy(
441 receipt,
442 ResetDomainKind::PendingMessages,
443 self.policies.pending_messages,
444 receipt.policies.pending_messages,
445 )?;
446 self.verify_policy(
447 receipt,
448 ResetDomainKind::SubnetState,
449 self.policies.subnet_state,
450 receipt.policies.subnet_state,
451 )?;
452 self.verify_policy(
453 receipt,
454 ResetDomainKind::ExternalResources,
455 self.policies.external_resources,
456 receipt.policies.external_resources,
457 )
458 }
459
460 fn verify_policy<P: PartialEq>(
461 &self,
462 receipt: ResetReceipt,
463 domain: ResetDomainKind,
464 requirement: Option<P>,
465 achievement: Option<P>,
466 ) -> Result<(), BaselinePoolContractError> {
467 let Some(requirement) = requirement else {
468 return Ok(());
469 };
470 let Some(achievement) = achievement else {
471 return Err(BaselinePoolContractError::MissingResetDomain { domain });
472 };
473 if requirement != achievement {
474 return Err(BaselinePoolContractError::ResetPolicyMismatch {
475 domain,
476 requirement: self.policies,
477 achievement: receipt.policies,
478 });
479 }
480 Ok(())
481 }
482}
483
484impl BaselinePreparationStage {
485 #[must_use]
490 pub fn default_rebuild_reason(self) -> RebuildReason {
491 match self {
492 Self::RestoreCanisters => RebuildReason::SnapshotRestoreFailure,
493 Self::ResetNonSnapshotState => RebuildReason::ResetFailure,
494 Self::DriveToReadiness => RebuildReason::ReadinessFailure,
495 Self::ValidateRestored | Self::ValidateBuilt => {
496 RebuildReason::InvariantValidationFailure
497 }
498 Self::Build => RebuildReason::RecipeClassified {
499 code: "build".to_owned(),
500 },
501 }
502 }
503}
504
505impl BaselinePoolTimings {
506 #[must_use]
508 pub const fn wait(self) -> Duration {
509 self.wait
510 }
511
512 #[must_use]
514 pub const fn build(self) -> Option<Duration> {
515 self.build
516 }
517
518 #[must_use]
520 pub const fn restore(self) -> Option<Duration> {
521 self.restore
522 }
523
524 #[must_use]
526 pub const fn reset(self) -> Option<Duration> {
527 self.reset
528 }
529
530 #[must_use]
532 pub const fn readiness(self) -> Option<Duration> {
533 self.readiness
534 }
535
536 #[must_use]
538 pub const fn validation(self) -> Option<Duration> {
539 self.validation
540 }
541
542 #[must_use]
544 pub const fn stale_teardown(self) -> Option<Duration> {
545 self.stale_teardown
546 }
547
548 #[must_use]
550 pub const fn total(self) -> Duration {
551 self.total
552 }
553}
554
555impl std::fmt::Display for BaselinePoolTimings {
556 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
557 write!(
558 formatter,
559 "total={:?} wait={:?} build={:?} restore={:?} reset={:?} readiness={:?} validation={:?} stale_teardown={:?}",
560 self.total,
561 self.wait,
562 self.build,
563 self.restore,
564 self.reset,
565 self.readiness,
566 self.validation,
567 self.stale_teardown,
568 )
569 }
570}
571
572impl std::fmt::Display for BaselinePoolOutcome {
573 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
574 match &self.disposition {
575 BaselinePoolDisposition::Built => formatter.write_str("built")?,
576 BaselinePoolDisposition::Restored => formatter.write_str("restored")?,
577 BaselinePoolDisposition::Rebuilt(reason) => {
578 write!(
579 formatter,
580 "rebuilt slot={} reason={reason:?} {}",
581 self.slot, self.timings
582 )?;
583 return Ok(());
584 }
585 }
586 write!(formatter, " slot={} {}", self.slot, self.timings)
587 }
588}
589
590impl<E> BaselinePoolError<E> {
591 #[must_use]
593 pub const fn timings(&self) -> BaselinePoolTimings {
594 match self {
595 Self::Preparation { timings, .. } | Self::RecoveryFailed { timings, .. } => **timings,
596 }
597 }
598}
599
600impl<R> CachedPocketIcBaselinePool<R>
601where
602 R: PocketIcBaselineRecipe,
603{
604 #[must_use]
609 pub const fn new(capacity: NonZeroUsize, recipe: R) -> Self {
610 Self {
611 recipe,
612 slots: BoundedSlotPool::new(capacity),
613 }
614 }
615
616 pub async fn close(self) -> Vec<crate::pocket_ic::ClientError> {
619 let mut failures = Vec::new();
620 for baseline in self.slots.into_values() {
621 if let Err(error) = baseline.close().await {
622 failures.push(error);
623 }
624 }
625 failures
626 }
627
628 #[must_use]
630 pub const fn capacity(&self) -> NonZeroUsize {
631 self.slots.capacity()
632 }
633
634 #[expect(
646 clippy::significant_drop_tightening,
647 reason = "lease and coordinator scope deliberately cover the complete operation"
648 )]
649 pub async fn acquire(
650 &self,
651 ) -> Result<
652 (CachedPocketIcBaselinePoolGuard<'_, R>, BaselinePoolOutcome),
653 BaselinePoolError<R::Error>,
654 > {
655 let total_started = Instant::now();
656 let mut slot = self.slots.acquire().await;
657 slot.begin_preparation();
658 let mut timings = BaselinePoolTimings {
659 wait: slot.wait(),
660 ..BaselinePoolTimings::default()
661 };
662 let slot_index = slot.slot_index();
663
664 let result = Box::pin(self.prepare_slot(&mut slot, &mut timings)).await;
665 slot.finish_preparation();
666 timings.total = total_started.elapsed();
667 let disposition = result.map_err(|error| error.with_timings(timings))?;
668 Ok((
669 CachedPocketIcBaselinePoolGuard { slot },
670 BaselinePoolOutcome {
671 slot: slot_index,
672 timings,
673 disposition,
674 },
675 ))
676 }
677
678 async fn prepare_slot(
679 &self,
680 slot: &mut BoundedSlotLease<'_, CachedPocketIcBaseline<R::Metadata>, RebuildReason>,
681 timings: &mut BaselinePoolTimings,
682 ) -> Result<BaselinePoolDisposition, PreparationFailure<R::Error>> {
683 let (disposition, original) = match slot.state() {
684 Slot::Reusable(_) => {
685 let Err(original) = self.prepare_reused(slot, timings).await else {
686 return Ok(BaselinePoolDisposition::Restored);
687 };
688 let FailureDisposition::Rebuild(reason) = self.failure_disposition(&original)
689 else {
690 Self::discard_stale_slot(slot, timings).await;
693 return Err(PreparationFailure::Single(original));
694 };
695 (BaselinePoolDisposition::Rebuilt(reason), Some(original))
696 }
697 Slot::Empty => (BaselinePoolDisposition::Built, None),
698 Slot::Invalidated { reason, .. } => {
699 (BaselinePoolDisposition::Rebuilt(reason.clone()), None)
700 }
701 Slot::Interrupted(_) => (
702 BaselinePoolDisposition::Rebuilt(RebuildReason::PreparationCancelled),
703 None,
704 ),
705 Slot::Unwound(_) => (
706 BaselinePoolDisposition::Rebuilt(RebuildReason::UnwindWhileLeased),
707 None,
708 ),
709 };
710
711 if slot.get().is_some() {
712 Self::discard_stale_slot(slot, timings).await;
713 }
714 let build = self.build_slot(slot, timings).await;
715 if let Err(error) = build {
716 return Err(match original {
717 Some(original) => PreparationFailure::Recovery {
718 original: Box::new(original),
719 rebuild: Box::new(error),
720 },
721 None => PreparationFailure::Single(error),
722 });
723 }
724
725 Ok(disposition)
726 }
727
728 async fn prepare_reused(
729 &self,
730 slot: &BoundedSlotLease<'_, CachedPocketIcBaseline<R::Metadata>, RebuildReason>,
731 timings: &mut BaselinePoolTimings,
732 ) -> Result<(), BaselinePoolPreparationError<R::Error>> {
733 let baseline = slot
734 .get()
735 .expect("reusable baseline slot must be populated");
736
737 let started = Instant::now();
738 let restore = async {
739 self.recipe
740 .prepare_restore(baseline)
741 .await
742 .map_err(|source| BaselinePoolPreparationError::Recipe {
743 stage: BaselinePreparationStage::RestoreCanisters,
744 source,
745 })?;
746 baseline
747 .restore(self.recipe.reset_requirements().funding)
748 .await
749 .map_err(BaselinePoolPreparationError::Restore)
750 }
751 .await;
752 add_timing(&mut timings.restore, started.elapsed());
753 restore?;
754
755 let reset = timed_recipe_stage(
756 &mut timings.reset,
757 BaselinePreparationStage::ResetNonSnapshotState,
758 self.recipe.reset_non_snapshot_state(baseline),
759 )
760 .await?;
761 timed_recipe_stage(
762 &mut timings.readiness,
763 BaselinePreparationStage::DriveToReadiness,
764 self.recipe.drive_to_readiness(baseline),
765 )
766 .await?;
767 self.recipe
768 .reset_requirements()
769 .verify(reset)
770 .map_err(BaselinePoolPreparationError::Contract)?;
771
772 let preparation = PreparedBaseline::Restored { reset };
773 self.validate_baseline(baseline, &preparation, timings)
774 .await?;
775 Ok(())
776 }
777
778 async fn build_slot(
779 &self,
780 slot: &mut BoundedSlotLease<'_, CachedPocketIcBaseline<R::Metadata>, RebuildReason>,
781 timings: &mut BaselinePoolTimings,
782 ) -> Result<(), BaselinePoolPreparationError<R::Error>> {
783 let baseline = timed_recipe_stage(
784 &mut timings.build,
785 BaselinePreparationStage::Build,
786 self.recipe.build(),
787 )
788 .await?;
789
790 if let Err(error) = self
791 .validate_baseline(&baseline, &PreparedBaseline::Built, timings)
792 .await
793 {
794 drop_baseline_safely(baseline).await;
795 return Err(error);
796 }
797 let replaced = slot.replace(baseline);
798 debug_assert!(replaced.is_none());
799 Ok(())
800 }
801
802 async fn validate_baseline(
803 &self,
804 baseline: &CachedPocketIcBaseline<R::Metadata>,
805 preparation: &PreparedBaseline,
806 timings: &mut BaselinePoolTimings,
807 ) -> Result<(), BaselinePoolPreparationError<R::Error>> {
808 let stage = match preparation {
809 PreparedBaseline::Built => BaselinePreparationStage::ValidateBuilt,
810 PreparedBaseline::Restored { .. } => BaselinePreparationStage::ValidateRestored,
811 };
812 timed_recipe_stage(
813 &mut timings.validation,
814 stage,
815 self.recipe.validate(baseline, preparation),
816 )
817 .await
818 }
819
820 fn failure_disposition(
821 &self,
822 error: &BaselinePoolPreparationError<R::Error>,
823 ) -> FailureDisposition {
824 match error {
825 BaselinePoolPreparationError::Recipe { stage, source } => {
826 self.recipe.classify_failure(*stage, source)
827 }
828 BaselinePoolPreparationError::Restore(source) => self
829 .recipe
830 .classify_failure(BaselinePreparationStage::RestoreCanisters, source),
831 BaselinePoolPreparationError::Contract(_) => {
832 FailureDisposition::Rebuild(RebuildReason::ResetCoverageMismatch)
833 }
834 }
835 }
836
837 async fn discard_stale_slot(
838 slot: &mut BoundedSlotLease<'_, CachedPocketIcBaseline<R::Metadata>, RebuildReason>,
839 timings: &mut BaselinePoolTimings,
840 ) {
841 let started = Instant::now();
842 if let Some(stale) = slot.take() {
843 drop_baseline_safely(stale).await;
844 }
845 timings.stale_teardown = Some(started.elapsed());
846 }
847}
848
849impl<R> CachedPocketIcBaselinePoolGuard<'_, R>
850where
851 R: PocketIcBaselineRecipe,
852{
853 #[must_use]
855 pub const fn slot(&self) -> usize {
856 self.slot.slot_index()
857 }
858
859 pub fn invalidate(&mut self, reason: RebuildReason) {
861 self.slot.invalidate(reason);
862 }
863}
864
865impl<R> Deref for CachedPocketIcBaselinePoolGuard<'_, R>
866where
867 R: PocketIcBaselineRecipe,
868{
869 type Target = CachedPocketIcBaseline<R::Metadata>;
870
871 fn deref(&self) -> &Self::Target {
872 self.slot
873 .get()
874 .expect("leased baseline pool slot must be populated")
875 }
876}
877
878async fn timed_recipe_stage<T, E>(
879 timing: &mut Option<Duration>,
880 stage: BaselinePreparationStage,
881 operation: impl std::future::Future<Output = Result<T, E>>,
882) -> Result<T, BaselinePoolPreparationError<E>> {
883 let started = Instant::now();
884 let result = operation.await;
885 add_timing(timing, started.elapsed());
886 result.map_err(|source| BaselinePoolPreparationError::Recipe { stage, source })
887}
888
889const fn add_timing(total: &mut Option<Duration>, elapsed: Duration) {
890 *total = Some(match *total {
891 None => elapsed,
892 Some(previous) => previous.saturating_add(elapsed),
893 });
894}
895
896async fn drop_baseline_safely<M>(baseline: CachedPocketIcBaseline<M>) {
897 if let Err(error) = baseline.close().await {
898 tracing::warn!("PocketIC baseline deletion failed: {error}");
899 }
900}
901
902impl std::fmt::Display for BaselinePreparationStage {
903 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
904 formatter.write_str(match self {
905 Self::Build => "build",
906 Self::RestoreCanisters => "canister restore",
907 Self::ResetNonSnapshotState => "non-snapshot reset",
908 Self::DriveToReadiness => "readiness",
909 Self::ValidateBuilt => "built-baseline validation",
910 Self::ValidateRestored => "restored-baseline validation",
911 })
912 }
913}
914
915impl<E> std::fmt::Display for BaselinePoolError<E>
916where
917 E: std::fmt::Display,
918{
919 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
920 match self {
921 Self::Preparation { error, .. } => error.fmt(formatter),
922 Self::RecoveryFailed {
923 original, rebuild, ..
924 } => write!(
925 formatter,
926 "baseline preparation failed ({original}); rebuilding the slot also failed: {rebuild}",
927 ),
928 }
929 }
930}
931
932impl<E> std::error::Error for BaselinePoolError<E>
933where
934 E: std::error::Error + 'static,
935{
936 fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
937 match self {
938 Self::Preparation { error, .. } => Some(error),
939 Self::RecoveryFailed { original, .. } => Some(original.as_ref()),
940 }
941 }
942}
943
944#[cfg(test)]
945mod tests;