taskvisor 0.8.0

In-process Tokio task supervisor with retries, graceful shutdown, reliable final outcomes, and per-key admission control
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130

//! Reserves supervisor-owned cleanup capacity for submissions.
//!
//! Submission methods wrap each [`ControllerSpec`] in an `OwnedTask` before command intake.
//! Waiting stops if the controller channel closes.
//! Fail-fast reservation converts capacity and worker-start failures into controller errors.

use std::future::Future;

use crate::{
    controller::{ControllerError, ControllerSpec},
    core::deferred_drop::{
        DropAdmissionError, DropCapacityError, DropReservation, DropStartError, OWNERSHIP_RESOURCE,
        OwnedTask,
    },
    events::Event,
};

use super::ControllerHandle;

impl ControllerHandle {
    /// Waits for ownership or reports that controller intake has closed.
    async fn reserve_or_closed(
        &self,
        reservation: impl Future<Output = Result<DropReservation, DropAdmissionError>>,
    ) -> Result<DropReservation, ControllerError> {
        tokio::pin!(reservation);
        tokio::select! {
            biased;
            _ = self.tx.closed() => Err(ControllerError::Closed),
            result = &mut reservation => result.map_err(Self::admission_error),
        }
    }

    /// Waits for deterministic test capacity or controller closure.
    #[cfg(test)]
    async fn reserve_capacity_or_closed(
        &self,
        reservation: impl Future<Output = Result<DropReservation, DropCapacityError>>,
    ) -> Result<DropReservation, ControllerError> {
        tokio::pin!(reservation);
        tokio::select! {
            biased;
            _ = self.tx.closed() => Err(ControllerError::Closed),
            result = &mut reservation => result.map_err(Self::capacity_error),
        }
    }

    /// Converts bounded exhaustion or closed cleanup admission into a controller error.
    fn capacity_error(error: DropCapacityError) -> ControllerError {
        match error.limit() {
            Some(limit) => ControllerError::ResourceLimit {
                resource: OWNERSHIP_RESOURCE,
                limit: limit.get(),
            },
            None => ControllerError::Closed,
        }
    }

    /// Converts cleanup worker startup failure into a controller error.
    fn start_error(error: DropStartError) -> ControllerError {
        ControllerError::ThreadStartFailed {
            component: "destructor_isolation",
            worker: error.worker(),
            kind: error.source_kind(),
            raw_os_error: error.raw_os_error(),
        }
    }

    /// Converts ownership admission failure into a controller error.
    fn admission_error(error: DropAdmissionError) -> ControllerError {
        match error {
            DropAdmissionError::Start(error) => Self::start_error(error),
            DropAdmissionError::Capacity(error) => Self::capacity_error(error),
        }
    }

    /// Reserves destructor ownership before command intake.
    pub(super) async fn own(
        &self,
        spec: ControllerSpec,
    ) -> Result<OwnedTask<ControllerSpec>, ControllerError> {
        #[cfg(test)]
        let reservation = match &self.reservation_source {
            Some(source) => self.reserve_capacity_or_closed(source.reserve()).await?,
            None => self.reserve_or_closed(self.drop_domain.reserve()).await?,
        };
        #[cfg(not(test))]
        let reservation = self.reserve_or_closed(self.drop_domain.reserve()).await?;

        let retained = spec.task_spec().task().clone();
        let mut owned = OwnedTask::new(spec, retained, reservation);
        let bus = self.bus.clone();
        owned.cleanup.set_panic_reporter(move |message| {
            bus.publish_lazy(|| {
                Event::runtime_failure("controller", format!("task_drop_panicked: {message}"))
            });
        });
        Ok(owned)
    }

    /// Reserves destructor ownership without waiting.
    pub(super) fn try_own(
        &self,
        spec: ControllerSpec,
    ) -> Result<OwnedTask<ControllerSpec>, ControllerError> {
        #[cfg(test)]
        let reservation = match &self.reservation_source {
            Some(source) => source.try_reserve().map_err(Self::capacity_error),
            None => self
                .drop_domain
                .try_reserve()
                .map_err(Self::admission_error),
        };
        #[cfg(not(test))]
        let reservation = self
            .drop_domain
            .try_reserve()
            .map_err(Self::admission_error);

        let retained = spec.task_spec().task().clone();
        let mut owned = OwnedTask::new(spec, retained, reservation?);
        let bus = self.bus.clone();
        owned.cleanup.set_panic_reporter(move |message| {
            bus.publish_lazy(|| {
                Event::runtime_failure("controller", format!("task_drop_panicked: {message}"))
            });
        });
        Ok(owned)
    }
}