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

use std::collections::HashMap;
use std::sync::{Arc, RwLock};

use uuid::Uuid;

use crate::meta_storage::mem::error::MemResult;
use crate::meta_storage::mem::resolution::ResolutionTable;
use crate::meta_storage::mem::ticket::TicketTable;
use crate::schema::RunFootprint;

/// The in-memory metadata store.
///
/// Ticket and resolution tables are keyed by task/dimension id.
/// Their coordinates carry their arity at runtime, so one store holds tables spanning tasks of
/// differing arity without a type parameter.
///
/// # Concurrency
///
/// A table takes one lock over all of its rows, so writes to the same task serialize.
#[derive(Debug, Default)]
pub(super) struct MemStore {
    tickets: RwLock<HashMap<&'static str, Arc<TicketTable>>>,
    resolutions: RwLock<HashMap<&'static str, Arc<ResolutionTable>>>,
    footprint: RwLock<Option<RunFootprint>>,
    executions: RwLock<HashMap<Uuid, Execution>>,
}

/// One recorded execution of a run.
#[derive(Debug)]
struct Execution {
    run_id: Uuid,
    ended_at: Option<chrono::DateTime<chrono::Utc>>,
    end_reason: Option<String>,
}

impl MemStore {
    /// Registers a task's ticket table, keeping an existing one.
    pub(super) fn init_ticket_table(&self, task_id: &'static str) -> MemResult<()> {
        let _ = self
            .tickets
            .write()?
            .entry(task_id)
            .or_insert_with(|| Arc::new(TicketTable::default()));
        Ok(())
    }

    /// Returns a task's ticket table, if it has been registered.
    pub(super) fn ticket_table(
        &self,
        task_id: &'static str,
    ) -> MemResult<Option<Arc<TicketTable>>> {
        Ok(self.tickets.read()?.get(task_id).cloned())
    }

    /// Registers a dimension's resolution table, keeping an existing one.
    pub(super) fn init_resolution_table(&self, dim_id: &'static str) -> MemResult<()> {
        let _ = self
            .resolutions
            .write()?
            .entry(dim_id)
            .or_insert_with(|| Arc::new(ResolutionTable::default()));
        Ok(())
    }

    /// Returns a dimension's resolution table, if it has been registered.
    pub(super) fn resolution_table(
        &self,
        dim_id: &'static str,
    ) -> MemResult<Option<Arc<ResolutionTable>>> {
        Ok(self.resolutions.read()?.get(dim_id).cloned())
    }

    pub(super) fn clear_footprint(&self) -> MemResult<()> {
        *self.footprint.write()? = None;
        self.executions.write()?.clear();
        Ok(())
    }

    pub(super) fn get_footprint(&self) -> MemResult<Option<RunFootprint>> {
        Ok(self.footprint.read()?.clone())
    }

    pub(super) fn upsert_run(&self, footprint: &RunFootprint) -> MemResult<()> {
        *self.footprint.write()? = Some(footprint.clone());
        Ok(())
    }

    pub(super) fn put_execution(&self, run_id: Uuid, execution_id: Uuid) -> MemResult<()> {
        let _old_value = self.executions.write()?.insert(
            execution_id,
            Execution {
                run_id,
                ended_at: None,
                end_reason: None,
            },
        );
        Ok(())
    }

    pub(super) fn update_execution_on_finish(
        &self,
        footprint: &RunFootprint,
        execution_id: Uuid,
    ) -> MemResult<()> {
        let mut executions = self.executions.write()?;
        let Some(execution) = executions.get_mut(&execution_id) else {
            return Ok(());
        };
        if execution.run_id != footprint.metadata.run_id {
            return Ok(());
        }
        execution.ended_at = Some(footprint.at);
        execution.end_reason = Some(footprint.metadata.state.to_string());
        Ok(())
    }
}