faucet_cli/serve/history/
mod.rs

1//! Run-history storage. The trait is defined in full now; the in-memory backend
2//! lives in `memory.rs`, and the feature-gated SQL backends (`postgres.rs` /
3//! `sqlite.rs`, sharing `sql.rs`) wrap themselves in `fallback.rs` so an
4//! unreachable backend degrades to in-memory rather than refusing to start.
5//! See spec §11 + §20.
6
7#[cfg(any(feature = "serve-history-postgres", feature = "serve-history-sqlite"))]
8pub mod fallback;
9pub mod memory;
10#[cfg(feature = "serve-history-postgres")]
11pub mod postgres;
12#[cfg(any(feature = "serve-history-postgres", feature = "serve-history-sqlite"))]
13pub mod sql;
14#[cfg(feature = "serve-history-sqlite")]
15pub mod sqlite;
16
17use crate::error::CliResult;
18use crate::executor::InvocationOutcome;
19use crate::serve::config::HistoryBackendSpec;
20use async_trait::async_trait;
21use chrono::{DateTime, Utc};
22use serde::{Deserialize, Serialize};
23use std::collections::BTreeMap;
24use std::sync::Arc;
25use std::time::Duration;
26
27/// Lifecycle state of a submitted run.
28#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
29#[serde(rename_all = "snake_case")]
30pub enum RunStatus {
31    Queued,
32    Running,
33    Completed,
34    Failed,
35    Cancelled,
36}
37
38impl RunStatus {
39    pub fn is_terminal(self) -> bool {
40        matches!(self, Self::Completed | Self::Failed | Self::Cancelled)
41    }
42    pub fn as_str(self) -> &'static str {
43        match self {
44            Self::Queued => "queued",
45            Self::Running => "running",
46            Self::Completed => "completed",
47            Self::Failed => "failed",
48            Self::Cancelled => "cancelled",
49        }
50    }
51}
52
53/// Serializable mirror of one pipeline invocation's outcome.
54#[derive(Debug, Clone, Serialize, Deserialize)]
55pub struct InvocationRecord {
56    pub row_id: String,
57    pub parent_record_key: Option<String>,
58    pub records_written: usize,
59    pub error: Option<String>,
60}
61
62impl From<&InvocationOutcome> for InvocationRecord {
63    fn from(o: &InvocationOutcome) -> Self {
64        Self {
65            row_id: o.row_id.clone(),
66            parent_record_key: o.parent_record_key.clone(),
67            records_written: o.records_written,
68            error: o.error.clone(),
69        }
70    }
71}
72
73/// One run's full record — the GET / list element (spec §6).
74#[derive(Debug, Clone, Serialize, Deserialize)]
75pub struct RunRecord {
76    pub run_id: String,
77    pub name: Option<String>,
78    pub labels: BTreeMap<String, String>,
79    pub status: RunStatus,
80    pub submitted_at: DateTime<Utc>,
81    pub started_at: Option<DateTime<Utc>>,
82    pub finished_at: Option<DateTime<Utc>>,
83    pub elapsed_secs: Option<f64>,
84    pub records_written: u64,
85    pub invocations: Vec<InvocationRecord>,
86    pub error: Option<String>,
87    pub idempotency_key: Option<String>,
88    #[serde(skip_serializing_if = "Option::is_none")]
89    pub doctor_report: Option<serde_json::Value>,
90}
91
92impl RunRecord {
93    /// A freshly-submitted run, before it acquires an execution slot.
94    pub fn queued(
95        run_id: String,
96        name: Option<String>,
97        labels: BTreeMap<String, String>,
98        idempotency_key: Option<String>,
99        submitted_at: DateTime<Utc>,
100    ) -> Self {
101        Self {
102            run_id,
103            name,
104            labels,
105            status: RunStatus::Queued,
106            submitted_at,
107            started_at: None,
108            finished_at: None,
109            elapsed_secs: None,
110            records_written: 0,
111            invocations: Vec::new(),
112            error: None,
113            idempotency_key,
114            doctor_report: None,
115        }
116    }
117}
118
119/// Result of an atomic idempotency-key claim.
120#[derive(Debug, Clone, PartialEq, Eq)]
121pub enum Claim {
122    /// Key is new (or its prior claim expired) — caller owns it for `run_id`.
123    Fresh,
124    /// Key was already claimed with a matching payload — replay this run id.
125    Replay(String),
126    /// Key was claimed with a *different* payload — 409.
127    Conflict,
128}
129
130/// Result of a delete attempt.
131#[derive(Debug, Clone, Copy, PartialEq, Eq)]
132pub enum DeleteOutcome {
133    Deleted,
134    NotFound,
135    StillRunning,
136}
137
138/// Filter + pagination for `list`. `limit`/`cursor` are resolved by the handler.
139#[derive(Debug, Default, Clone)]
140pub struct ListFilter {
141    pub status: Option<RunStatus>,
142    pub name: Option<String>,
143    pub since: Option<DateTime<Utc>>,
144    pub until: Option<DateTime<Utc>>,
145    pub limit: usize,
146    pub cursor: Option<String>,
147}
148
149/// One page of `list` results, ordered `(submitted_at DESC, run_id DESC)`.
150#[derive(Debug)]
151pub struct ListPage {
152    pub runs: Vec<RunRecord>,
153    pub next_cursor: Option<String>,
154}
155
156/// Backend failure. The memory backend never returns one; the variant exists so
157/// the async trait stays fallible for the Phase 5 SQL backends.
158#[derive(Debug, thiserror::Error)]
159pub enum HistoryError {
160    #[error("run-history backend error: {0}")]
161    Backend(String),
162    /// The backend is degraded and the operation can't be honored safely
163    /// (e.g. an idempotency claim that would risk a duplicate run). Maps to a
164    /// `503` so the caller can retry once the backend recovers (#146 M5).
165    #[error("{0}")]
166    Degraded(String),
167}
168
169#[async_trait]
170pub trait RunHistory: Send + Sync {
171    /// Atomically claim `key` for `run_id` (or report a replay/conflict). A prior
172    /// claim older than `window` is treated as expired and re-claimable.
173    async fn claim_idempotency(
174        &self,
175        key: &str,
176        fingerprint: &str,
177        run_id: &str,
178        window: Duration,
179    ) -> Result<Claim, HistoryError>;
180
181    /// Insert or replace a run record.
182    async fn upsert(&self, rec: &RunRecord) -> Result<(), HistoryError>;
183
184    async fn get(&self, id: &str) -> Result<Option<RunRecord>, HistoryError>;
185
186    async fn list(&self, filter: &ListFilter) -> Result<ListPage, HistoryError>;
187
188    /// Delete a terminal run. Non-terminal → `StillRunning` (caller maps to 409).
189    async fn delete(&self, id: &str) -> Result<DeleteOutcome, HistoryError>;
190
191    /// Drop terminal records finished longer than `retain_for` ago. Returns the
192    /// number removed.
193    async fn purge_expired(&self, retain_for: Duration) -> Result<usize, HistoryError>;
194
195    /// Mark non-terminal records whose owning instance's lease has expired as
196    /// failed (instance-fenced orphan recovery — never touches a live peer's
197    /// heartbeated runs, #146 H7). Returns the number recovered. The memory
198    /// backend has nothing to recover (returns 0).
199    async fn recover_orphans(&self) -> Result<usize, HistoryError>;
200
201    /// Heartbeat: extend the lease of *this* instance's own non-terminal runs so
202    /// a peer's [`recover_orphans`](Self::recover_orphans) won't reclaim them.
203    /// Returns the number of leases renewed. The memory backend (single-process,
204    /// unshared) is a no-op returning 0.
205    async fn renew_leases(&self) -> Result<usize, HistoryError> {
206        Ok(0)
207    }
208
209    /// True when the backend is in fallback mode (drives `/readyz`). Always false
210    /// for memory.
211    fn degraded(&self) -> bool;
212}
213
214/// Build the configured run-history backend. `Memory` is always available; the
215/// SQL backends require their respective `serve-history-*` build features (a
216/// clear error otherwise). A SQL backend that fails to connect at startup
217/// degrades to in-memory (via `FallbackHistory`) rather than aborting boot.
218pub async fn connect(
219    spec: &HistoryBackendSpec,
220    idem_retention: Duration,
221    lease_ttl: Duration,
222    instance_id: &str,
223) -> CliResult<Arc<dyn RunHistory>> {
224    match spec {
225        HistoryBackendSpec::Memory => {
226            Ok(Arc::new(memory::MemoryHistory::new(idem_retention)) as Arc<dyn RunHistory>)
227        }
228        HistoryBackendSpec::Postgres(url) => {
229            connect_postgres(url, idem_retention, lease_ttl, instance_id).await
230        }
231        HistoryBackendSpec::Sqlite(url) => {
232            connect_sqlite(url, idem_retention, lease_ttl, instance_id).await
233        }
234    }
235}
236
237#[cfg(feature = "serve-history-postgres")]
238async fn connect_postgres(
239    url: &str,
240    idem: Duration,
241    lease_ttl: Duration,
242    instance_id: &str,
243) -> CliResult<Arc<dyn RunHistory>> {
244    Ok(into_history(
245        postgres::PostgresHistory::connect(url, idem, lease_ttl, instance_id.to_string()).await,
246        idem,
247        "postgres",
248    ))
249}
250
251#[cfg(not(feature = "serve-history-postgres"))]
252async fn connect_postgres(
253    _url: &str,
254    _idem: Duration,
255    _lease_ttl: Duration,
256    _instance_id: &str,
257) -> CliResult<Arc<dyn RunHistory>> {
258    Err(crate::error::CliError::Serve(
259        "persistent Postgres run history requires building faucet with the \
260         `serve-history-postgres` feature"
261            .into(),
262    ))
263}
264
265#[cfg(feature = "serve-history-sqlite")]
266async fn connect_sqlite(
267    url: &str,
268    idem: Duration,
269    lease_ttl: Duration,
270    instance_id: &str,
271) -> CliResult<Arc<dyn RunHistory>> {
272    Ok(into_history(
273        sqlite::SqliteHistory::connect(url, idem, lease_ttl, instance_id.to_string()).await,
274        idem,
275        "sqlite",
276    ))
277}
278
279#[cfg(not(feature = "serve-history-sqlite"))]
280async fn connect_sqlite(
281    _url: &str,
282    _idem: Duration,
283    _lease_ttl: Duration,
284    _instance_id: &str,
285) -> CliResult<Arc<dyn RunHistory>> {
286    Err(crate::error::CliError::Serve(
287        "persistent SQLite run history requires building faucet with the \
288         `serve-history-sqlite` feature"
289            .into(),
290    ))
291}
292
293/// Wrap a SQL backend in `FallbackHistory`: healthy on success; degraded-on-
294/// in-memory (server stays up, `/readyz` reports 503) on a connect failure.
295#[cfg(any(feature = "serve-history-postgres", feature = "serve-history-sqlite"))]
296fn into_history<H: RunHistory + 'static>(
297    result: Result<H, HistoryError>,
298    idem: Duration,
299    label: &'static str,
300) -> Arc<dyn RunHistory> {
301    match result {
302        Ok(backend) => Arc::new(fallback::FallbackHistory::healthy(
303            Box::new(backend),
304            idem,
305            label,
306        )),
307        Err(e) => {
308            tracing::error!(
309                backend = label, error = %e,
310                "run-history backend unavailable at startup; starting DEGRADED on in-memory store"
311            );
312            Arc::new(fallback::FallbackHistory::degraded_at_startup(idem, label))
313        }
314    }
315}
316
317#[cfg(test)]
318mod tests {
319    use super::*;
320
321    #[test]
322    fn terminal_classification() {
323        assert!(!RunStatus::Queued.is_terminal());
324        assert!(!RunStatus::Running.is_terminal());
325        assert!(RunStatus::Completed.is_terminal());
326        assert!(RunStatus::Failed.is_terminal());
327        assert!(RunStatus::Cancelled.is_terminal());
328    }
329
330    #[test]
331    fn run_record_serializes_status_snake_case() {
332        let rec = RunRecord::queued(
333            "r1".into(),
334            Some("n".into()),
335            Default::default(),
336            None,
337            Utc::now(),
338        );
339        let v = serde_json::to_value(&rec).unwrap();
340        assert_eq!(v["status"], "queued");
341        assert_eq!(v["run_id"], "r1");
342        // doctor_report is skipped when None.
343        assert!(v.get("doctor_report").is_none());
344    }
345}
faucet_cli/serve/history/mod.rs

faucet_cli/serve/history/
mod.rs
