eoka_server/operations/
mod.rs

1pub mod commands;
2pub mod sidecar;
3pub(crate) mod store;
4
5use anyhow::{anyhow, Context, Result};
6use eoka::{Action, ActionOptions, ActionOutcome, DispatchObserver, Error, Page};
7use eoka_protocol::{OperationListArgs, OperationRecord, OperationState};
8use std::{
9    collections::HashMap,
10    future::Future,
11    path::PathBuf,
12    pin::Pin,
13    sync::{Arc, Mutex},
14    time::Duration,
15};
16use store::{timestamp, Store};
17use tokio::sync::{watch, OnceCell};
18
19#[derive(Clone, Default)]
20pub struct OperationManager {
21    inner: Arc<Manager>,
22}
23#[derive(Default)]
24struct Manager {
25    store: OnceCell<Arc<Store>>,
26    directory: Option<PathBuf>,
27    jobs: Mutex<HashMap<String, watch::Sender<bool>>>,
28}
29
30#[derive(Default)]
31pub(crate) struct Execution {
32    pub canonical: Option<Vec<u8>>,
33    pub exploratory: bool,
34}
35
36pub(crate) struct Intent {
37    pub store: Arc<Store>,
38    pub id: String,
39}
40impl DispatchObserver for Intent {
41    fn before_dispatch(
42        &self,
43        preparation: bool,
44    ) -> Pin<Box<dyn Future<Output = eoka::Result<()>> + Send + '_>> {
45        Box::pin(async move {
46            self.store
47                .update(self.id.clone(), move |record| {
48                    if preparation {
49                        record.preparation_intent = true;
50                    } else {
51                        record.dispatch_intent = true;
52                        record.outcome = ActionOutcome::OutcomeUnknown;
53                    }
54                })
55                .await
56                .and_then(|record| {
57                    if matches!(record.state, OperationState::Running)
58                        && record.runtime_id == self.store.runtime
59                        && (if preparation {
60                            record.preparation_intent
61                        } else {
62                            record.dispatch_intent
63                        })
64                    {
65                        Ok(())
66                    } else {
67                        Err(anyhow!("Operation is no longer owned and running"))
68                    }
69                })
70                .map_err(|error| {
71                    Error::transport_io(
72                        "Could not durably record dispatch intent",
73                        std::io::Error::other(error),
74                    )
75                })
76        })
77    }
78}
79
80impl OperationManager {
81    pub fn in_directory(directory: PathBuf) -> Self {
82        Self {
83            inner: Arc::new(Manager {
84                directory: Some(directory),
85                ..Default::default()
86            }),
87        }
88    }
89    pub(crate) async fn store(&self) -> Result<&Arc<Store>> {
90        self.inner
91            .store
92            .get_or_try_init(|| async {
93                Store::open(match &self.inner.directory {
94                    Some(path) => path.clone(),
95                    None => store::state_directory()?,
96                })
97                .await
98            })
99            .await
100    }
101    pub async fn submit(
102        &self,
103        page: Page,
104        id: String,
105        action: Action,
106        options: ActionOptions,
107    ) -> Result<OperationRecord> {
108        self.submit_with(page, id, action, options, Execution::default())
109            .await
110    }
111
112    pub(crate) async fn submit_with(
113        &self,
114        page: Page,
115        id: String,
116        action: Action,
117        options: ActionOptions,
118        execution: Execution,
119    ) -> Result<OperationRecord> {
120        let manager = self.clone();
121        tokio::spawn(async move {
122            manager
123                .submit_owned(page, id, action, options, execution)
124                .await
125        })
126        .await
127        .context("Operation admission task failed; inspect its ID before retrying")?
128    }
129
130    async fn submit_owned(
131        &self,
132        page: Page,
133        id: String,
134        action: Action,
135        mut options: ActionOptions,
136        execution: Execution,
137    ) -> Result<OperationRecord> {
138        let parsed = uuid::Uuid::parse_str(&id)
139            .context("operation_id must be a UUID generated before sending the request")?;
140        if parsed.to_string() != id {
141            return Err(anyhow!("operation_id must be a canonical lowercase UUID"));
142        }
143        let store = self.store().await?.clone();
144        let canonical = match execution.canonical {
145            Some(canonical) => canonical,
146            None => serde_json::to_vec(&serde_json::json!({"action":action,"options":options}))?,
147        };
148        let initial = OperationRecord {
149            id: id.clone(),
150            runtime_id: store.runtime.clone(),
151            target_id: page.target_id().into(),
152            action: action.kind().into(),
153            state: OperationState::Running,
154            outcome: ActionOutcome::NotDispatched,
155            dispatch_intent: false,
156            preparation_intent: false,
157            started_at_ms: timestamp(),
158            finished_at_ms: None,
159            receipt: None,
160            error_code: None,
161        };
162        let (record, created) = store.admit(initial, canonical).await?;
163        if !created {
164            return Ok(record);
165        }
166        let (cancel, mut canceled) = watch::channel(false);
167        self.inner
168            .jobs
169            .lock()
170            .map_err(|_| anyhow!("Operation registry unavailable"))?
171            .insert(id.clone(), cancel);
172        options.observer = Some(Arc::new(Intent {
173            store: store.clone(),
174            id: id.clone(),
175        }));
176        let manager = self.clone();
177        tokio::spawn(async move {
178            let result = tokio::select! {
179                biased;
180                _ = canceled.changed() => None,
181                result = async {
182                    if execution.exploratory { page.explore(action, options).await }
183                    else { page.act(action, options).await }
184                } => Some(result),
185            };
186            let finish = store
187                .update(id.clone(), move |record| {
188                    record.state = OperationState::Finished;
189                    record.finished_at_ms = Some(timestamp());
190                    match result {
191                        Some(Ok(receipt)) => {
192                            record.outcome = receipt.outcome;
193                            record.receipt = Some(receipt);
194                        }
195                        Some(Err(Error::Action { receipt, code, .. })) => {
196                            record.outcome = receipt.outcome;
197                            record.receipt = Some(*receipt);
198                            record.error_code = Some(code);
199                        }
200                        other => {
201                            record.outcome = if record.dispatch_intent {
202                                ActionOutcome::OutcomeUnknown
203                            } else {
204                                ActionOutcome::NotDispatched
205                            };
206                            record.error_code = Some(
207                                if other.is_none() {
208                                    "operation_canceled"
209                                } else {
210                                    "action_failed"
211                                }
212                                .into(),
213                            );
214                        }
215                    }
216                })
217                .await;
218            if let Err(error) = finish {
219                tracing::error!(operation_id = %id, %error, "Could not persist operation completion; outcome remains uncertain");
220            }
221            if let Ok(mut jobs) = manager.inner.jobs.lock() {
222                jobs.remove(&id);
223            }
224        });
225        Ok(record)
226    }
227    pub async fn get(&self, id: &str) -> Result<OperationRecord> {
228        let store = self.store().await?;
229        store.recover().await?;
230        store.get(id.into()).await
231    }
232    pub async fn list(&self, args: OperationListArgs) -> Result<Vec<OperationRecord>> {
233        let limit = args.limit.unwrap_or(100).clamp(1, 500);
234        let before = args.before_ms.unwrap_or(i64::MAX as u64);
235        self.store()
236            .await?
237            .run(move |connection| {
238                let mut query = connection.prepare(
239                    "SELECT record FROM operations WHERE started < ?
240                     ORDER BY started DESC,id LIMIT ?",
241                )?;
242                let rows = query.query_map(rusqlite::params![before, limit], |row| {
243                    row.get::<_, String>(0)
244                })?;
245                rows.map(|row| Ok(serde_json::from_str(&row?)?)).collect()
246            })
247            .await
248    }
249    pub async fn wait(&self, id: &str, timeout_ms: u64) -> Result<OperationRecord> {
250        if timeout_ms > 86_400_000 {
251            return Err(anyhow!("Wait timeout exceeds 24 hours"));
252        }
253        let deadline = tokio::time::Instant::now() + Duration::from_millis(timeout_ms);
254        loop {
255            let record = self.get(id).await?;
256            if !matches!(record.state, OperationState::Running)
257                || tokio::time::Instant::now() >= deadline
258            {
259                return Ok(record);
260            }
261            tokio::time::sleep_until(
262                deadline.min(tokio::time::Instant::now() + Duration::from_millis(25)),
263            )
264            .await;
265        }
266    }
267    pub async fn cancel(&self, id: &str) -> Result<OperationRecord> {
268        let record = self.get(id).await?;
269        self.cancel_admitted(id, record)
270    }
271
272    pub(crate) fn cancel_admitted(
273        &self,
274        id: &str,
275        record: OperationRecord,
276    ) -> Result<OperationRecord> {
277        if matches!(record.state, OperationState::Running) {
278            let jobs = self
279                .inner
280                .jobs
281                .lock()
282                .map_err(|_| anyhow!("Operation registry unavailable"))?;
283            let sender = jobs.get(id).ok_or_else(|| {
284                anyhow!("operation_owned_elsewhere: request cancellation from the owning runtime")
285            })?;
286            sender.send_replace(true);
287        }
288        Ok(record)
289    }
290    pub async fn stop(&self) {
291        let ids = if let Ok(jobs) = self.inner.jobs.lock() {
292            jobs.iter()
293                .map(|(id, cancel)| {
294                    cancel.send_replace(true);
295                    id.clone()
296                })
297                .collect::<Vec<_>>()
298        } else {
299            Vec::new()
300        };
301        for id in ids {
302            if let Err(error) = self.wait(&id, 5000).await {
303                tracing::warn!(operation_id = %id, %error, "Could not confirm operation shutdown");
304            }
305        }
306    }
307    pub fn is_active(&self) -> bool {
308        self.inner
309            .jobs
310            .lock()
311            .map(|j| !j.is_empty())
312            .unwrap_or(true)
313    }
314}
315
316#[cfg(test)]
317mod tests;
eoka_server/operations/mod.rs

eoka_server/operations/
mod.rs
