taskvisor 0.7.0

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

//! Commands and completion messages exchanged by the controller engine.

use std::sync::Arc;

use tokio::sync::oneshot;

use crate::{
    RuntimeError,
    controller::spec::ControllerSpec,
    core::{OutcomeTx, RemovalCompletion},
    identity::TaskId,
};

pub(super) enum ControllerCommand {
    /// Apply admission policy for one new submission.
    Submit(Submission),
    /// Start one identity operation after earlier controller commands are handled.
    ///
    /// Queued-work lookup happens inline in command order.
    /// Registry fallback for every other id runs in a bounded worker and may finish concurrently with later operations.
    ManageIdentity {
        id: TaskId,
        operation: IdentityOperation,
        reply: oneshot::Sender<Result<bool, RuntimeError>>,
    },
}

/// Identity operation owned by one accepted controller command.
#[derive(Clone, Copy, Debug)]
pub(super) enum IdentityOperation {
    Remove,
    TryRemove,
    Cancel,
    TryCancel,
    CancelWithTimeout(std::time::Duration),
    TryCancelWithTimeout(std::time::Duration),
}

impl IdentityOperation {
    /// Optional reason carried by `TaskRemoveRequested` when queued work is removed.
    pub(super) fn request_reason(self) -> Option<&'static str> {
        match self {
            Self::Remove | Self::TryRemove => None,
            Self::Cancel
            | Self::TryCancel
            | Self::CancelWithTimeout(_)
            | Self::TryCancelWithTimeout(_) => Some("manual_cancel"),
        }
    }
}

/// Ensures an accepted identity caller receives an explicit shutdown result if its worker aborts.
pub(super) struct IdentityReply {
    sender: Option<oneshot::Sender<Result<bool, RuntimeError>>>,
}

impl IdentityReply {
    pub(super) fn new(sender: oneshot::Sender<Result<bool, RuntimeError>>) -> Self {
        Self {
            sender: Some(sender),
        }
    }

    pub(super) fn send(mut self, result: Result<bool, RuntimeError>) {
        if let Some(sender) = self.sender.take() {
            let _ = sender.send(result);
        }
    }
}

impl Drop for IdentityReply {
    fn drop(&mut self) {
        if let Some(sender) = self.sender.take() {
            let _ = sender.send(Err(RuntimeError::ShuttingDown));
        }
    }
}

/// Submission accepted by the controller command channel.
pub(super) struct Submission {
    /// Pre-minted identity used for events, slot state, and final outcome correlation.
    pub(super) id: TaskId,
    /// Admission policy, task spec, and optional slot key.
    pub(super) spec: ControllerSpec,
    /// Optional watched-outcome sender for `submit_and_watch`.
    pub(super) done: Option<OutcomeTx>,
}

/// Authoritative registry decision for one in-flight slot admission.
pub(super) struct AdmissionResult {
    /// Pre-minted identity used to reject stale results safely.
    pub(super) id: TaskId,
    /// Slot that owned `id` when the Add command was committed.
    pub(super) slot_name: Arc<str>,
    /// Direct registry decision and the terminal signal for an accepted task.
    pub(super) decision: Result<RemovalCompletion, RuntimeError>,
}

/// Reliable registry cleanup for one admitted slot owner.
pub(super) struct CompletionResult {
    /// Runtime identity that reached terminal registry cleanup.
    pub(super) id: TaskId,
    /// Slot that owned `id` when completion tracking started.
    pub(super) slot_name: Arc<str>,
}

/// Result of ordering removal for one controller-owned runtime task.
pub(super) struct RemovalResult {
    /// Runtime identity whose removal was requested.
    pub(super) id: TaskId,
    /// Slot that owned `id` when removal tracking started.
    pub(super) slot_name: Arc<str>,
    /// Direct registry claim decision or management-plane failure.
    pub(super) decision: Result<bool, RuntimeError>,
}