taskvisor 0.4.2

Task supervisor for Tokio: restarts background tasks on failure with exponential backoff and jitter, graceful shutdown, and lifecycle events
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

//! Controller shutdown drain.
//!
//! When the controller loop exits, some submissions may still be waiting for a terminal decision.
//!
//! They can be in two places:
//! - `watchers`: submissions already seen by the controller, with a parked waiter,
//! - `rx`: submissions accepted by the intake channel but not processed yet.
//!
//! This module resolves both groups as `TaskOutcome::Rejected` with `controller_shutting_down`.

use std::sync::Arc;

use tokio::sync::mpsc;

use crate::core::TaskOutcome;
use crate::events::{Event, EventKind};
use crate::identity::TaskId;

use super::{Controller, Submission};

impl Controller {
    /// Resolves every still-pending watched submission during controller shutdown.
    ///
    /// This preserves the `submit_and_watch` contract:
    /// a submission that never reaches the runtime must resolve as [`TaskOutcome::Rejected`], not as a dropped oneshot.
    ///
    /// The drain is split into two parts:
    /// - already parked watchers are rejected through [`finalize_rejected`](Self::finalize_rejected),
    /// - not-yet-processed intake submissions are drained from `rx` and rejected directly.
    ///
    /// `rx.close()` prevents new messages from being accepted while the remaining buffered submissions are drained.
    pub(super) fn finalize_pending_on_shutdown(&self, rx: &mut mpsc::Receiver<Submission>) {
        rx.close();

        let pending: Vec<TaskId> = self.watchers.iter().map(|e| *e.key()).collect();
        for id in pending {
            self.bus.publish(
                Event::new(EventKind::ControllerRejected)
                    .with_id(id)
                    .with_reason(crate::reasons::CONTROLLER_SHUTTING_DOWN),
            );
            self.finalize_rejected(id, crate::reasons::CONTROLLER_SHUTTING_DOWN);
        }

        while let Ok(sub) = rx.try_recv() {
            self.bus.publish(
                Event::new(EventKind::ControllerRejected)
                    .with_task(sub.spec.slot_name().to_owned())
                    .with_id(sub.id)
                    .with_reason(crate::reasons::CONTROLLER_SHUTTING_DOWN),
            );

            if let Some(done) = sub.done {
                let _ = done.send(TaskOutcome::Rejected {
                    reason: Arc::from(crate::reasons::CONTROLLER_SHUTTING_DOWN),
                });
            }
        }
    }
}