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

//! Advances an idle slot to its next queued submission.
//!
//! Registry rejection, physical completion, or removal of a capacity waiter can clear a slot owner.
//! Their handlers call this module after the slot returns to `Idle`.
//! It pops work in queue order and sends each candidate to `handoff`.
//!
//! A handoff failure before registry commit rejects that candidate and moves to the next one.
//! Recovered values return for cleanup after the slot unlocks.

use std::sync::Arc;

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

use super::{
    super::{Controller, TrackedOperations, state::SlotState},
    cleanup::StartFailure,
};

impl Controller {
    /// Starts the first queued submission that can enter registry admission.
    ///
    /// The slot must be idle. Failed handoffs are returned for deferred cleanup.
    pub(in crate::controller::engine) fn start_next_from_queue(
        &self,
        sup: &Arc<SupervisorCore>,
        slot: &mut SlotState,
        slot_name: &Arc<str>,
        operations: &mut TrackedOperations,
    ) -> Vec<(StartFailure, Option<TaskOutcome>)> {
        let mut deferred_drops = Vec::new();
        debug_assert!(slot.is_idle());
        if !slot.is_idle() {
            return deferred_drops;
        }
        while let Some(next) = self.pop_queued_front(slot) {
            let next_id = next.id;
            match self.start_in_slot(sup, slot, slot_name, next, operations) {
                Ok(()) => {
                    self.bus.publish_lazy(|| {
                        Event::new(EventKind::ControllerSubmitted)
                            .with_task(Arc::clone(slot_name))
                            .with_id(next_id)
                            .with_reason(format!("started_from_queue depth={}", slot.queue.len()))
                    });
                    return deferred_drops;
                }
                Err(uncommitted) => {
                    let kind = Self::rejection_kind_for_runtime_error(&uncommitted.error);
                    let reason = format!("queue_start_failed: {}", uncommitted.error);
                    self.bus.publish_lazy(|| {
                        Event::new(EventKind::ControllerRejected)
                            .with_task(Arc::clone(slot_name))
                            .with_id(next_id)
                            .with_rejection_kind(kind)
                            .with_reason(reason.clone())
                    });
                    let terminal = self.finalize_rejected(next_id, kind, &reason);
                    deferred_drops.push((uncommitted, terminal));
                }
            }
        }
        deferred_drops
    }
}