operon 0.7.0

A workflow engine for parallel, incremental scheduling of DAG-defined multiplex tasks.
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
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139

use std::collections::VecDeque;
use std::error::Error;

use async_trait::async_trait;
use tokio_util::task::AbortOnDropHandle;

use crate::scheduler::SchedulerError;
use crate::scheduler::events::ControlEvent;

mod clean;
mod fresh;
mod init;
mod rebuild;
mod running;
mod stale;
mod start;

pub(super) use init::InitTransition;

pub(super) enum NextState<E> {
    Next(Box<dyn SchedulerState<Error = E>>),
    Exit { exit_ui: bool },
}

#[async_trait]
pub(super) trait SchedulerState: Send + Sync {
    type Error: Error + Send + Sync + 'static;

    /// Resolves once the state has progressed on its own, without a control event.
    ///
    /// States that only ever move on a control event should never resolve.
    /// The scheduler loop races this against incoming control events, so implementations must be
    /// cancel-safe.
    async fn wait_progress(&mut self) -> Result<(), Self::Error> {
        std::future::pending().await
    }

    /// Advances the state after [`SchedulerState::wait_progress`] has resolved.
    async fn handle_progress(self: Box<Self>) -> Result<NextState<Self::Error>, Self::Error>;

    async fn handle_control_event(
        self: Box<Self>,
        evt: ControlEvent,
    ) -> Result<NextState<Self::Error>, Self::Error>;
}

#[async_trait]
pub(super) trait SchedulerTransition: Send + Sync + 'static {
    type Error: Error + Send + Sync + 'static;
    fn warn_msg(&self) -> Option<&'static str>;
    async fn execute(self) -> Result<NextState<Self::Error>, Self::Error>;
}

pub(super) struct TransitionState<E> {
    warn_msg: Option<&'static str>,
    handle: AbortOnDropHandle<Result<NextState<E>, E>>,
    next: Option<NextState<E>>,
    events: VecDeque<ControlEvent>,
}

impl<E: Error + Send + Sync + 'static> TransitionState<E> {
    pub(super) fn new<T: SchedulerTransition<Error = E>>(transition: T) -> Self {
        let warn_msg = transition.warn_msg();
        let handle =
            AbortOnDropHandle::new(tokio::task::spawn(
                async move { transition.execute().await },
            ));
        Self {
            warn_msg,
            handle,
            next: None,
            events: VecDeque::new(),
        }
    }
}

#[async_trait]
impl<UErr, SErr, MErr> SchedulerState for TransitionState<SchedulerError<UErr, SErr, MErr>>
where
    UErr: std::fmt::Debug + std::fmt::Display + Send + Sync + 'static,
    SErr: Error + Send + Sync + 'static,
    MErr: Error + Send + Sync + 'static,
{
    type Error = SchedulerError<UErr, SErr, MErr>;

    async fn wait_progress(&mut self) -> Result<(), Self::Error> {
        if self.next.is_none() {
            let next = (&mut self.handle)
                .await
                .map_err(SchedulerError::from)
                .flatten()?;
            self.next = Some(next);
        }
        Ok(())
    }

    async fn handle_progress(mut self: Box<Self>) -> Result<NextState<Self::Error>, Self::Error> {
        let Some(mut next) = self.next.take() else {
            return Ok(NextState::Next(self));
        };

        while let Some(evt) = self.events.pop_front() {
            match next {
                NextState::Next(state) => {
                    next = state.handle_control_event(evt).await?;
                }
                NextState::Exit { exit_ui } => return Ok(NextState::Exit { exit_ui }),
            }
        }

        Ok(next)
    }

    async fn handle_control_event(
        mut self: Box<Self>,
        evt: ControlEvent,
    ) -> Result<NextState<Self::Error>, Self::Error> {
        if let Some(msg) = self.warn_msg {
            tracing::warn!("{msg}")
        }
        self.events.push_back(evt);

        Ok(NextState::Next(self))
    }
}

impl<UErr, SErr, MErr> NextState<SchedulerError<UErr, SErr, MErr>>
where
    UErr: std::fmt::Debug + std::fmt::Display + Send + Sync + 'static,
    SErr: Error + Send + Sync + 'static,
    MErr: Error + Send + Sync + 'static,
{
    /// Boxes a concrete state into the next state.
    pub(super) fn next(
        state: impl SchedulerState<Error = SchedulerError<UErr, SErr, MErr>> + 'static,
    ) -> Self {
        NextState::Next(Box::new(state))
    }
}