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