a3s_flow/store/sqlite/
retention.rs

1use std::collections::{BTreeMap, BTreeSet};
2
3use a3s_orm::{sql_query, SqliteTransaction};
4use chrono::Utc;
5
6use crate::error::{FlowError, Result};
7use crate::model::FlowEventEnvelope;
8use crate::store::retention::{history_checksum, plan_history_retention, validate_history_hold};
9use crate::store::{
10    FlowHistoryHold, FlowHistoryRetentionPolicy, FlowHistoryRetentionReport, FlowHistoryTombstone,
11};
12
13use super::{
14    execute_sqlite, fetch_all_sqlite, fetch_optional_sqlite, latest_sqlite_sequence,
15    map_sqlite_transaction, row_to_envelope, SqliteEventStore,
16};
17
18pub(super) use crate::store::retention::required_linked_flow_run_id;
19
20impl SqliteEventStore {
21    /// Persist an audit hold that prevents a run history from being pruned.
22    ///
23    /// Repeating the same `(run_id, hold_id, reason)` is idempotent. Reusing a
24    /// hold ID with another reason returns a conflict so audit intent cannot be
25    /// silently replaced.
26    pub async fn hold_history(&self, run_id: &str, hold_id: &str, reason: &str) -> Result<()> {
27        validate_history_hold(run_id, hold_id, reason)?;
28        let run_id = run_id.to_string();
29        let hold_id = hold_id.to_string();
30        let reason = reason.to_string();
31        let result = self
32            .executor
33            .transaction(|transaction| {
34                Box::pin(async move {
35                    ensure_sqlite_history_not_tombstoned(transaction, &run_id).await?;
36                    if latest_sqlite_sequence(transaction, &run_id).await? == 0 {
37                        return Err(FlowError::RunNotFound(run_id));
38                    }
39                    let existing = fetch_optional_sqlite(
40                        transaction,
41                        sql_query::<String>(
42                            "SELECT reason FROM flow_history_holds WHERE run_id = ",
43                        )
44                        .bind(run_id.clone())
45                        .append(" AND hold_id = ")
46                        .bind(hold_id.clone()),
47                    )
48                    .await?;
49                    match existing {
50                        Some(existing) if existing == reason => return Ok(()),
51                        Some(_) => {
52                            return Err(FlowError::RunConflict {
53                                run_id,
54                                reason: format!(
55                                    "history hold {hold_id:?} differs from the durable hold"
56                                ),
57                            });
58                        }
59                        None => {}
60                    }
61                    execute_sqlite(
62                        transaction,
63                        sql_query::<()>(
64                            "INSERT INTO flow_history_holds (run_id, hold_id, reason, created_at) VALUES (",
65                        )
66                        .bind(run_id)
67                        .append(", ")
68                        .bind(hold_id)
69                        .append(", ")
70                        .bind(reason)
71                        .append(", ")
72                        .bind(Utc::now().to_rfc3339())
73                        .append(")"),
74                    )
75                    .await?;
76                    Ok(())
77                })
78            })
79            .await;
80        map_sqlite_transaction(result)
81    }
82
83    /// Release one audit hold. Returns false when the hold did not exist.
84    pub async fn release_history_hold(&self, run_id: &str, hold_id: &str) -> Result<bool> {
85        if run_id.trim().is_empty() || hold_id.trim().is_empty() {
86            return Err(FlowError::InvalidTransition(
87                "history hold run id and hold id must not be empty".to_string(),
88            ));
89        }
90        let run_id = run_id.to_string();
91        let hold_id = hold_id.to_string();
92        let result = self
93            .executor
94            .transaction(|transaction| {
95                Box::pin(async move {
96                    let rows = execute_sqlite(
97                        transaction,
98                        sql_query::<()>("DELETE FROM flow_history_holds WHERE run_id = ")
99                            .bind(run_id)
100                            .append(" AND hold_id = ")
101                            .bind(hold_id),
102                    )
103                    .await?;
104                    Ok(rows > 0)
105                })
106            })
107            .await;
108        map_sqlite_transaction(result)
109    }
110
111    /// List durable audit holds for one run in stable hold-ID order.
112    pub async fn history_holds(&self, run_id: &str) -> Result<Vec<FlowHistoryHold>> {
113        let rows = fetch_all_sqlite(
114            &self.executor,
115            sql_query::<(String, String, String, String)>(
116                "SELECT run_id, hold_id, reason, created_at FROM flow_history_holds WHERE run_id = ",
117            )
118            .bind(run_id)
119            .append(" ORDER BY hold_id ASC"),
120        )
121        .await?;
122        rows.into_iter().map(history_hold_row).collect()
123    }
124
125    /// Read the minimal audit tombstone retained after history deletion.
126    pub async fn history_tombstone(&self, run_id: &str) -> Result<Option<FlowHistoryTombstone>> {
127        fetch_optional_sqlite(
128            &self.executor,
129            sql_query::<(String, String, i64, String, String, String)>(
130                "SELECT run_id, deleted_at, terminal_sequence, terminal_event_id, terminal_event_key, history_sha256 FROM flow_history_tombstones WHERE run_id = ",
131            )
132            .bind(run_id),
133        )
134        .await?
135        .map(history_tombstone_row)
136        .transpose()
137    }
138
139    /// Delete complete eligible terminal histories in one immediate transaction.
140    ///
141    /// The scan preserves durable holds and linked components, writes a checksum
142    /// tombstone, and only then deletes event rows. It never performs partial
143    /// stream compaction.
144    pub async fn prune_terminal_history(
145        &self,
146        policy: FlowHistoryRetentionPolicy,
147    ) -> Result<FlowHistoryRetentionReport> {
148        let result = self
149            .executor
150            .transaction(|transaction| {
151                Box::pin(async move { prune_sqlite_history(transaction, &policy).await })
152            })
153            .await;
154        map_sqlite_transaction(result)
155    }
156}
157
158async fn prune_sqlite_history(
159    transaction: &SqliteTransaction,
160    policy: &FlowHistoryRetentionPolicy,
161) -> Result<FlowHistoryRetentionReport> {
162    let rows = fetch_all_sqlite(
163        transaction,
164        sql_query::<(String, i64, String, String, String)>(
165            "SELECT run_id, sequence, event_id, timestamp, event_json FROM flow_events ORDER BY run_id ASC, sequence ASC",
166        ),
167    )
168    .await?;
169    let mut histories = BTreeMap::<String, Vec<FlowEventEnvelope>>::new();
170    for row in rows {
171        let envelope = row_to_envelope(row)?;
172        histories
173            .entry(envelope.run_id.clone())
174            .or_default()
175            .push(envelope);
176    }
177
178    let hold_run_ids = fetch_all_sqlite(
179        transaction,
180        sql_query::<String>("SELECT DISTINCT run_id FROM flow_history_holds"),
181    )
182    .await?
183    .into_iter()
184    .collect::<BTreeSet<_>>();
185    let mut plan = plan_history_retention(&histories, &hold_run_ids, policy, "SQLite")?;
186
187    for run_id in &plan.deletable_run_ids {
188        let history = histories.get(run_id).ok_or_else(|| {
189            FlowError::Store(format!("retention lost SQLite history for {run_id}"))
190        })?;
191        let terminal = history.last().ok_or_else(|| {
192            FlowError::Store(format!("retention found empty SQLite history for {run_id}"))
193        })?;
194        let terminal_sequence = i64::try_from(terminal.sequence).map_err(|error| {
195            FlowError::Store(format!(
196                "terminal sequence {} for {run_id} exceeds SQLite integer range: {error}",
197                terminal.sequence
198            ))
199        })?;
200        execute_sqlite(
201            transaction,
202            sql_query::<()>(
203                "INSERT INTO flow_history_tombstones (run_id, deleted_at, terminal_sequence, terminal_event_id, terminal_event_key, history_sha256) VALUES (",
204            )
205            .bind(run_id.clone())
206            .append(", ")
207            .bind(Utc::now().to_rfc3339())
208            .append(", ")
209            .bind(terminal_sequence)
210            .append(", ")
211            .bind(terminal.event_id.to_string())
212            .append(", ")
213            .bind(terminal.event.event_key())
214            .append(", ")
215            .bind(history_checksum(history)?)
216            .append(")"),
217        )
218        .await?;
219        execute_sqlite(
220            transaction,
221            sql_query::<()>("DELETE FROM flow_events WHERE run_id = ").bind(run_id.clone()),
222        )
223        .await?;
224    }
225
226    plan.report.deleted_run_ids = plan.deletable_run_ids.into_iter().collect();
227    Ok(plan.report)
228}
229
230fn history_hold_row(
231    (run_id, hold_id, reason, created_at): (String, String, String, String),
232) -> Result<FlowHistoryHold> {
233    Ok(FlowHistoryHold {
234        run_id,
235        hold_id,
236        reason,
237        created_at: created_at.parse().map_err(|error| {
238            FlowError::Store(format!(
239                "invalid SQLite history hold timestamp {created_at}: {error}"
240            ))
241        })?,
242    })
243}
244
245fn history_tombstone_row(
246    (
247        run_id,
248        deleted_at,
249        terminal_sequence,
250        terminal_event_id,
251        terminal_event_key,
252        history_sha256,
253    ): (String, String, i64, String, String, String),
254) -> Result<FlowHistoryTombstone> {
255    Ok(FlowHistoryTombstone {
256        run_id,
257        deleted_at: deleted_at.parse().map_err(|error| {
258            FlowError::Store(format!(
259                "invalid SQLite history tombstone timestamp {deleted_at}: {error}"
260            ))
261        })?,
262        terminal_sequence: u64::try_from(terminal_sequence).map_err(|error| {
263            FlowError::Store(format!(
264                "invalid SQLite tombstone terminal sequence {terminal_sequence}: {error}"
265            ))
266        })?,
267        terminal_event_id: terminal_event_id.parse().map_err(|error| {
268            FlowError::Store(format!(
269                "invalid SQLite tombstone event id {terminal_event_id}: {error}"
270            ))
271        })?,
272        terminal_event_key,
273        history_sha256,
274    })
275}
276
277pub(super) async fn ensure_sqlite_history_not_tombstoned(
278    transaction: &SqliteTransaction,
279    run_id: &str,
280) -> Result<()> {
281    let tombstoned = fetch_optional_sqlite(
282        transaction,
283        sql_query::<String>("SELECT run_id FROM flow_history_tombstones WHERE run_id = ")
284            .bind(run_id),
285    )
286    .await?
287    .is_some();
288    if tombstoned {
289        return Err(FlowError::RunConflict {
290            run_id: run_id.to_string(),
291            reason: "history was pruned and its run ID is tombstoned".to_string(),
292        });
293    }
294    Ok(())
295}
a3s_flow/store/sqlite/retention.rs

a3s_flow/store/sqlite/
retention.rs
