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

use std::sync::Arc;

use uuid::Uuid;

use crate::meta_storage::mem::error::MemResult;
use crate::meta_storage::mem::store::MemStore;
use crate::meta_storage::mem::{MemMetaStorage, MemResolutionQueryBuilder, MemTicketQueryBuilder};
use crate::meta_storage::{MetaClientApi, MetaConnApi, MetaTxApi};
use crate::schema::{DimensionMetadata, RunFootprint, TableShape, TaskMetadata};

/// A handle on the in-memory store.
///
/// The store is process-local, so a "connection" is just a shared handle to it. It owns its handle,
/// so it can be handed out for `'static`.
#[derive(Debug)]
pub struct MemConn {
    store: Arc<MemStore>,
}

impl MemConn {
    pub(super) fn new(store: Arc<MemStore>) -> Self {
        Self { store }
    }
}

impl MetaConnApi<MemMetaStorage> for MemConn {
    /// Opens a transaction.
    ///
    /// Writes through it land as they are issued.
    async fn transaction(&mut self) -> MemResult<MemTx<'_>> {
        Ok(MemTx { store: &self.store })
    }

    fn as_client(&self) -> MemClient<'_> {
        MemClient(&self.store)
    }
}

/// A transaction over the in-memory store.
#[derive(Debug)]
pub struct MemTx<'a> {
    store: &'a MemStore,
}

impl MetaTxApi<MemMetaStorage> for MemTx<'_> {
    /// Commits the transaction: a no-op, as its writes already landed.
    async fn commit(self) -> MemResult<()> {
        Ok(())
    }

    /// Rolls the transaction back: a no-op, as its writes already landed.
    async fn rollback(self) -> MemResult<()> {
        Ok(())
    }

    fn as_client(&self) -> MemClient<'_> {
        MemClient(self.store)
    }
}

/// A borrowed handle for issuing queries against the in-memory store.
#[derive(Debug, Clone, Copy)]
pub struct MemClient<'a>(&'a MemStore);

impl MetaClientApi<MemMetaStorage> for MemClient<'_> {
    fn ticket<const N: usize>(&self, task_meta: TaskMetadata<N>) -> MemTicketQueryBuilder<'_, N> {
        self.0.ticket(task_meta)
    }

    fn resolution<const N: usize>(
        &self,
        dim_meta: DimensionMetadata<N>,
    ) -> MemResolutionQueryBuilder<'_, N> {
        self.0.resolution(dim_meta)
    }

    // --- Schema initialization ---
    //
    // Tables are registered on first use by the ticket and resolution builders.

    async fn init_schema(&self) -> MemResult<()> {
        Ok(())
    }

    async fn init_ticket_hash(&self) -> MemResult<()> {
        Ok(())
    }

    async fn init_dimension_hash(&self) -> MemResult<()> {
        Ok(())
    }

    async fn init_ticket_status_type(&self) -> MemResult<()> {
        Ok(())
    }

    async fn init_ticket_summary(&self) -> MemResult<()> {
        Ok(())
    }

    // --- Footprint ---

    async fn init_footprint(&self) -> MemResult<TableShape> {
        Ok(TableShape::CURRENT)
    }

    async fn clear_footprint(&self) -> MemResult<()> {
        self.0.clear_footprint()
    }

    async fn get_footprint(&self) -> MemResult<Option<RunFootprint>> {
        self.0.get_footprint()
    }

    async fn upsert_run(&self, footprint: &RunFootprint) -> MemResult<()> {
        self.0.upsert_run(footprint)
    }

    async fn put_execution(&self, run_id: Uuid, execution_id: Uuid) -> MemResult<()> {
        self.0.put_execution(run_id, execution_id)
    }

    async fn update_execution_on_finish(
        &self,
        footprint: &RunFootprint,
        execution_id: Uuid,
    ) -> MemResult<()> {
        self.0.update_execution_on_finish(footprint, execution_id)
    }
}