a3s_flow/store/
local_file.rs

1use async_trait::async_trait;
2use chrono::{DateTime, Utc};
3use std::collections::{BTreeMap, BTreeSet};
4use std::path::{Path, PathBuf};
5use std::sync::Arc;
6use tokio::fs::{File, OpenOptions};
7use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
8use tokio::sync::Mutex;
9use uuid::Uuid;
10
11use crate::error::{FlowError, Result};
12use crate::model::{project_run, FlowEvent, FlowEventEnvelope};
13
14use super::{
15    retention::{linked_flow_run_id, plan_history_retention, FlowHistoryRetentionPolicy},
16    FlowEventStore,
17};
18
19/// JSONL-backed event store for local durable runs.
20///
21/// Each workflow run is stored as `<root>/<run_id>.jsonl`; every line is a full
22/// [`FlowEventEnvelope`]. The store serializes appends inside this process, but
23/// it does not provide cross-process locking. Use it for local development,
24/// embedded Rust hosts, and crash/restart durability. An unterminated malformed
25/// tail is treated as a torn append and truncated before the next write;
26/// terminated or interior corruption remains an error. Use a database-backed
27/// store for multi-writer deployments.
28#[derive(Debug, Clone)]
29pub struct LocalFileEventStore {
30    root: PathBuf,
31    lock: Arc<Mutex<()>>,
32}
33
34#[derive(Debug, Clone, Copy, Eq, PartialEq)]
35enum TailRepair {
36    None,
37    AppendDelimiter,
38    Truncate(u64),
39}
40
41#[derive(Debug)]
42struct LoadedEventLog {
43    events: Vec<FlowEventEnvelope>,
44    tail_repair: TailRepair,
45}
46
47impl LocalFileEventStore {
48    pub fn new(root: impl Into<PathBuf>) -> Self {
49        Self {
50            root: root.into(),
51            lock: Arc::new(Mutex::new(())),
52        }
53    }
54
55    pub fn root(&self) -> &Path {
56        &self.root
57    }
58
59    fn run_path(&self, run_id: &str) -> Result<PathBuf> {
60        if !is_safe_run_id(run_id) {
61            return Err(FlowError::Store(format!(
62                "run id {run_id:?} is not safe for local file storage"
63            )));
64        }
65        Ok(self.root.join(format!("{run_id}.jsonl")))
66    }
67
68    async fn load_inner(&self, run_id: &str, missing_is_empty: bool) -> Result<LoadedEventLog> {
69        let path = self.run_path(run_id)?;
70        let file = match File::open(&path).await {
71            Ok(file) => file,
72            Err(err) if err.kind() == std::io::ErrorKind::NotFound && missing_is_empty => {
73                return Ok(LoadedEventLog {
74                    events: Vec::new(),
75                    tail_repair: TailRepair::None,
76                });
77            }
78            Err(err) if err.kind() == std::io::ErrorKind::NotFound => {
79                return Err(FlowError::RunNotFound(run_id.to_string()));
80            }
81            Err(err) => return Err(FlowError::Io(err)),
82        };
83
84        let mut reader = BufReader::new(file);
85        let mut events = Vec::new();
86        let mut line_no = 0usize;
87        let mut valid_prefix_len = 0u64;
88        let mut buffer = Vec::new();
89        loop {
90            buffer.clear();
91            let bytes_read = reader.read_until(b'\n', &mut buffer).await?;
92            if bytes_read == 0 {
93                break;
94            }
95            line_no += 1;
96            let terminated = buffer.last() == Some(&b'\n');
97            let line = if terminated {
98                &buffer[..buffer.len() - 1]
99            } else {
100                buffer.as_slice()
101            };
102            if line.iter().all(u8::is_ascii_whitespace) {
103                if !terminated {
104                    return Ok(LoadedEventLog {
105                        events,
106                        tail_repair: TailRepair::Truncate(valid_prefix_len),
107                    });
108                }
109                valid_prefix_len = checked_file_offset(valid_prefix_len, bytes_read, &path)?;
110                continue;
111            }
112            let envelope: FlowEventEnvelope = match serde_json::from_slice(line) {
113                Ok(envelope) => envelope,
114                Err(_) if !terminated => {
115                    return Ok(LoadedEventLog {
116                        events,
117                        tail_repair: TailRepair::Truncate(valid_prefix_len),
118                    });
119                }
120                Err(err) => {
121                    return Err(FlowError::Store(format!(
122                        "failed to decode event line {line_no} from {}: {err}",
123                        path.display()
124                    )));
125                }
126            };
127            if envelope.run_id != run_id {
128                return Err(FlowError::Store(format!(
129                    "event line {line_no} in {} belongs to run {}, not {run_id}",
130                    path.display(),
131                    envelope.run_id
132                )));
133            }
134            events.push(envelope);
135            valid_prefix_len = checked_file_offset(valid_prefix_len, bytes_read, &path)?;
136            if !terminated {
137                return Ok(LoadedEventLog {
138                    events,
139                    tail_repair: TailRepair::AppendDelimiter,
140                });
141            }
142        }
143
144        Ok(LoadedEventLog {
145            events,
146            tail_repair: TailRepair::None,
147        })
148    }
149
150    async fn list_inner(
151        &self,
152        run_id: &str,
153        missing_is_empty: bool,
154    ) -> Result<Vec<FlowEventEnvelope>> {
155        Ok(self.load_inner(run_id, missing_is_empty).await?.events)
156    }
157
158    fn validate_existing_log(&self, run_id: &str, events: &[FlowEventEnvelope]) -> Result<()> {
159        if events.is_empty() {
160            return Ok(());
161        }
162        project_run(run_id, events)?;
163        Ok(())
164    }
165
166    async fn append_inner(&self, run_id: &str, event: FlowEvent) -> Result<FlowEventEnvelope> {
167        tokio::fs::create_dir_all(&self.root).await?;
168        self.ensure_linked_flow_run_exists(&event).await?;
169
170        let LoadedEventLog {
171            events,
172            tail_repair,
173        } = self.load_inner(run_id, true).await?;
174        self.validate_existing_log(run_id, &events)?;
175        let envelope = FlowEventEnvelope {
176            run_id: run_id.to_string(),
177            sequence: events.last().map_or(1, |event| event.sequence + 1),
178            event_id: Uuid::new_v4(),
179            timestamp: Utc::now(),
180            event,
181        };
182
183        self.repair_tail(run_id, tail_repair).await?;
184        self.write_envelope(&envelope).await?;
185        Ok(envelope)
186    }
187
188    async fn append_if_sequence_inner(
189        &self,
190        run_id: &str,
191        expected_sequence: u64,
192        event: FlowEvent,
193    ) -> Result<FlowEventEnvelope> {
194        tokio::fs::create_dir_all(&self.root).await?;
195        self.ensure_linked_flow_run_exists(&event).await?;
196
197        let LoadedEventLog {
198            events,
199            tail_repair,
200        } = self.load_inner(run_id, true).await?;
201        self.validate_existing_log(run_id, &events)?;
202        let actual_sequence = events.last().map_or(0, |event| event.sequence);
203        if actual_sequence != expected_sequence {
204            return Err(FlowError::EventConflict {
205                run_id: run_id.to_string(),
206                expected_sequence,
207                actual_sequence,
208            });
209        }
210
211        let envelope = FlowEventEnvelope {
212            run_id: run_id.to_string(),
213            sequence: actual_sequence + 1,
214            event_id: Uuid::new_v4(),
215            timestamp: Utc::now(),
216            event,
217        };
218
219        self.repair_tail(run_id, tail_repair).await?;
220        self.write_envelope(&envelope).await?;
221        Ok(envelope)
222    }
223
224    async fn ensure_linked_flow_run_exists(&self, event: &FlowEvent) -> Result<()> {
225        let Some(linked_run_id) = linked_flow_run_id(event) else {
226            return Ok(());
227        };
228        let events = self.list_inner(linked_run_id, false).await?;
229        if events.is_empty() {
230            return Err(FlowError::RunNotFound(linked_run_id.to_string()));
231        }
232        self.validate_existing_log(linked_run_id, &events)
233    }
234
235    async fn repair_tail(&self, run_id: &str, repair: TailRepair) -> Result<()> {
236        let path = self.run_path(run_id)?;
237        match repair {
238            TailRepair::None => Ok(()),
239            TailRepair::AppendDelimiter => {
240                let mut file = OpenOptions::new().append(true).open(path).await?;
241                file.write_all(b"\n").await?;
242                file.flush().await?;
243                file.sync_data().await?;
244                Ok(())
245            }
246            TailRepair::Truncate(valid_prefix_len) => {
247                let file = OpenOptions::new().write(true).open(path).await?;
248                file.set_len(valid_prefix_len).await?;
249                file.sync_data().await?;
250                Ok(())
251            }
252        }
253    }
254
255    async fn write_envelope(&self, envelope: &FlowEventEnvelope) -> Result<()> {
256        let path = self.run_path(&envelope.run_id)?;
257        let mut file = OpenOptions::new()
258            .create(true)
259            .append(true)
260            .open(path)
261            .await?;
262        let mut line = serde_json::to_vec(envelope)?;
263        line.push(b'\n');
264        file.write_all(&line).await?;
265        file.flush().await?;
266        file.sync_data().await?;
267        Ok(())
268    }
269
270    async fn list_run_ids_inner(&self) -> Result<Vec<String>> {
271        let mut ids = Vec::new();
272
273        let mut dir = match tokio::fs::read_dir(&self.root).await {
274            Ok(dir) => dir,
275            Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(ids),
276            Err(err) => return Err(FlowError::Io(err)),
277        };
278
279        while let Some(entry) = dir.next_entry().await? {
280            let path = entry.path();
281            if path.extension().and_then(|ext| ext.to_str()) != Some("jsonl") {
282                continue;
283            }
284            let Some(stem) = path.file_stem().and_then(|stem| stem.to_str()) else {
285                continue;
286            };
287            if is_safe_run_id(stem) {
288                ids.push(stem.to_string());
289            }
290        }
291
292        ids.sort();
293        Ok(ids)
294    }
295
296    /// Remove complete linked components of terminal local run histories whose
297    /// terminal event timestamps are strictly before `terminal_before`.
298    ///
299    /// A running, suspended, or recent parent or child protects every history
300    /// linked to it. Corrupt histories and dangling child references are
301    /// returned as errors or retained rather than deleted, so operators can
302    /// inspect them before cleanup.
303    pub async fn prune_terminal_runs_older_than(
304        &self,
305        terminal_before: DateTime<Utc>,
306    ) -> Result<Vec<String>> {
307        let _guard = self.lock.lock().await;
308        let mut histories = BTreeMap::new();
309        for run_id in self.list_run_ids_inner().await? {
310            let events = self.list_inner(&run_id, false).await?;
311            self.validate_existing_log(&run_id, &events)?;
312            histories.insert(run_id, events);
313        }
314        let mut plan = plan_history_retention(
315            &histories,
316            &BTreeSet::new(),
317            &FlowHistoryRetentionPolicy::new(terminal_before),
318            "local file",
319        )?;
320
321        let mut removed = Vec::new();
322        for run_id in &plan.deletable_run_ids {
323            let path = self.run_path(run_id)?;
324            match tokio::fs::remove_file(&path).await {
325                Ok(()) => removed.push(run_id.clone()),
326                Err(err) if err.kind() == std::io::ErrorKind::NotFound => {}
327                Err(err) => return Err(FlowError::Io(err)),
328            }
329        }
330
331        plan.report.deleted_run_ids = removed;
332        Ok(plan.report.deleted_run_ids)
333    }
334}
335
336#[async_trait]
337impl FlowEventStore for LocalFileEventStore {
338    async fn append(&self, run_id: &str, event: FlowEvent) -> Result<FlowEventEnvelope> {
339        let _guard = self.lock.lock().await;
340        self.append_inner(run_id, event).await
341    }
342
343    async fn append_if_sequence(
344        &self,
345        run_id: &str,
346        expected_sequence: u64,
347        event: FlowEvent,
348    ) -> Result<FlowEventEnvelope> {
349        let _guard = self.lock.lock().await;
350        self.append_if_sequence_inner(run_id, expected_sequence, event)
351            .await
352    }
353
354    async fn list(&self, run_id: &str) -> Result<Vec<FlowEventEnvelope>> {
355        let _guard = self.lock.lock().await;
356        self.list_inner(run_id, false).await
357    }
358
359    async fn list_run_ids(&self) -> Result<Vec<String>> {
360        let _guard = self.lock.lock().await;
361        self.list_run_ids_inner().await
362    }
363}
364
365fn checked_file_offset(current: u64, bytes_read: usize, path: &Path) -> Result<u64> {
366    let bytes_read = u64::try_from(bytes_read).map_err(|_| {
367        FlowError::Store(format!(
368            "event line length from {} exceeds the supported file offset",
369            path.display()
370        ))
371    })?;
372    current.checked_add(bytes_read).ok_or_else(|| {
373        FlowError::Store(format!(
374            "event log {} exceeds the supported file offset",
375            path.display()
376        ))
377    })
378}
379
380fn is_safe_run_id(run_id: &str) -> bool {
381    !run_id.is_empty()
382        && run_id
383            .chars()
384            .all(|ch| ch.is_ascii_alphanumeric() || ch == '-' || ch == '_')
385}
a3s_flow/store/local_file.rs

a3s_flow/store/
local_file.rs
