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

//! Signals registry cleanup and physical actor release for one removal.
//!
//! Natural actor exit, remove, cancel, and shutdown all enter the same removal path.
//! [`RemovalCompletion`] lets other components observe that path without owning the actor join.
//! Public cancellation waits for logical completion. The controller also waits for physical completion before it reuses a slot.
//!
//! ```text
//! registered ───────────► removing ──────────────────────────► membership and reporting ──► logical
//! force-aborted actor ──► physical exit and terminal match ──► physical
//! ```
//!
//! Logical completion means membership was removed and terminal reporting was attempted.
//! Physical completion means the actor has exited and force-abort tracking has ended.
//! These latches do not request cancellation.

use tokio::sync::oneshot;
use tokio_util::sync::CancellationToken;

use crate::core::outcome::TaskOutcome;

/// Sender used to resolve a watched task with its final [`TaskOutcome`].
pub(crate) type OutcomeTx = oneshot::Sender<TaskOutcome>;

/// Shared two-phase completion for one registry removal.
#[derive(Clone, Debug)]
pub(crate) struct RemovalCompletion {
    /// Wakes public cancellation waiters after terminal commit.
    logical: CancellationToken,
    /// Wakes controller replacement after the physical attempt is fully released.
    physical: CancellationToken,
}

impl RemovalCompletion {
    /// Creates a new incomplete terminal-cleanup signal.
    pub(crate) fn new() -> Self {
        Self {
            logical: CancellationToken::new(),
            physical: CancellationToken::new(),
        }
    }

    /// Waits for logical completion without owning the removal operation.
    pub(crate) async fn wait(&self) {
        self.logical.cancelled().await;
    }

    /// Waits for both logical and physical completion.
    #[cfg(feature = "controller")]
    pub(crate) async fn wait_physical(&self) {
        self.logical.cancelled().await;
        self.physical.cancelled().await;
    }

    /// Returns `true` when terminal registry cleanup has been committed.
    pub(super) fn is_complete(&self) -> bool {
        self.logical.is_cancelled()
    }

    #[cfg(test)]
    /// Returns whether the physical attempt has been fully released.
    pub(super) fn is_physical_complete(&self) -> bool {
        self.physical.is_cancelled()
    }

    /// Marks the bounded logical terminal transition complete.
    pub(super) fn complete_logical(&self) {
        self.logical.cancel();
    }

    /// Marks the physical attempt as fully released.
    pub(super) fn complete_physical(&self) {
        self.physical.cancel();
    }

    /// Returns whether both values observe the same physical-release latch.
    pub(super) fn shares_physical_latch(&self, other: &Self) -> bool {
        self.physical == other.physical
    }
}