taskvisor 0.9.0

In-process Tokio task supervisor: one job per key with queue/replace/reject admission, retries, graceful shutdown, and reliable final outcomes
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

//! Defines task removal without a terminal-cleanup wait.

use crate::RuntimeError;

use super::{FailFast, TaskTarget, Waiting};
use crate::core::SupervisorHandle;

/// A task-removal operation with typed command-queue policy.
///
/// Success waits for the removal claim decision but not for terminal task cleanup.
/// Use a cancellation operation when terminal confirmation is required.
#[must_use = "a remove operation starts no work until `.execute()` is awaited"]
pub struct RemoveOperation<'a, Admission = Waiting, Target = TaskTarget> {
    handle: &'a SupervisorHandle,
    target: Target,
    _admission: Admission,
}

impl<Admission, Target> std::fmt::Debug for RemoveOperation<'_, Admission, Target>
where
    Admission: std::fmt::Debug,
{
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("RemoveOperation")
            .field("target_type", &std::any::type_name::<Target>())
            .field("admission", &self._admission)
            .finish_non_exhaustive()
    }
}

impl<'a, Target> RemoveOperation<'a, Waiting, Target> {
    #[inline]
    pub(crate) fn new(handle: &'a SupervisorHandle, target: Target) -> Self {
        Self {
            handle,
            target,
            _admission: Waiting,
        }
    }

    /// Immediate admission to the required management queue.
    #[inline]
    pub fn fail_fast(self) -> RemoveOperation<'a, FailFast, Target> {
        RemoveOperation {
            handle: self.handle,
            target: self.target,
            _admission: FailFast,
        }
    }
}

impl<Target> RemoveOperation<'_, Waiting, Target>
where
    Target: Into<TaskTarget>,
{
    /// Removal-claim result after waiting command-queue admission.
    ///
    /// # Errors
    ///
    /// - Returns [`RuntimeError::ShuttingDown`] when runtime intake is closed;
    /// - [`RuntimeError::ResourceLimitReached`] when a [`TaskTarget::Id`] routed through a configured controller exhausts the identity-operation budget.
    #[inline]
    pub async fn execute(self) -> Result<bool, RuntimeError> {
        match self.target.into() {
            TaskTarget::Id(id) => {
                #[cfg(feature = "controller")]
                if let Some(controller) = self.handle.controller() {
                    return controller.handle().remove(id).await;
                }
                self.handle.core().remove(id).await
            }
            TaskTarget::Name(name) => self.handle.core().remove_by_name(name).await,
        }
    }
}

impl<Target> RemoveOperation<'_, FailFast, Target>
where
    Target: Into<TaskTarget>,
{
    /// Removal-claim result after immediate command-queue admission.
    ///
    /// # Errors
    ///
    /// - Returns [`RuntimeError::ShuttingDown`] when runtime intake is closed;
    /// - [`RuntimeError::CommandQueueFull`] when the required management queue has no capacity;
    /// - [`RuntimeError::ResourceLimitReached`] when a [`TaskTarget::Id`] routed through a configured controller exhausts the identity-operation budget.
    #[inline]
    pub async fn execute(self) -> Result<bool, RuntimeError> {
        match self.target.into() {
            TaskTarget::Id(id) => {
                #[cfg(feature = "controller")]
                if let Some(controller) = self.handle.controller() {
                    return controller.handle().try_remove(id).await;
                }
                self.handle.core().try_remove(id).await
            }
            TaskTarget::Name(name) => self.handle.core().try_remove_by_name(name).await,
        }
    }
}