rivet/state/
run_aggregate.rs

1//! Aggregate run summary store — one row per `rivet run` invocation.
2//!
3//! Per-export rows stay in `export_metrics`.  This table answers:
4//! - "what happened in the last N scheduled runs as a whole?"
5//! - "which run produced the most data / the most failures?"
6//!
7//! `details_json` carries the per-export breakdown so callers do not have to
8//! join on `run_at` ranges to reconstruct the run.  This is intentional: the
9//! aggregate row is observational, not a source of truth — `export_metrics`
10//! remains the canonical per-export record.
11use crate::error::Result;
12
13use super::StateStore;
14
15/// One aggregated `rivet run`.
16#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
17pub struct RunAggregate {
18    /// Unique id assigned by the pipeline (`agg_<utc_ts>`).
19    pub run_aggregate_id: String,
20    pub started_at: String,
21    pub finished_at: String,
22    pub duration_ms: i64,
23    pub config_path: Option<String>,
24    /// `sequential` | `parallel-threads` | `parallel-processes`.
25    pub parallel_mode: String,
26    pub total_exports: usize,
27    pub success_count: usize,
28    pub failed_count: usize,
29    pub skipped_count: usize,
30    pub total_rows: i64,
31    pub total_files: i64,
32    pub total_bytes: u64,
33    pub per_export: Vec<RunAggregateEntry>,
34}
35
36/// Per-export row inside an aggregate.
37#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
38pub struct RunAggregateEntry {
39    pub export_name: String,
40    pub status: String,
41    pub run_id: String,
42    pub rows: i64,
43    pub files: i64,
44    pub bytes: u64,
45    pub duration_ms: i64,
46    pub mode: String,
47    pub error_message: Option<String>,
48}
49
50impl StateStore {
51    /// Persist an aggregate.  `per_export` is serialized as a JSON array into
52    /// `details_json`.
53    pub fn record_run_aggregate(&self, agg: &RunAggregate) -> Result<()> {
54        let details = serde_json::to_string(&agg.per_export)
55            .map_err(|e| anyhow::anyhow!("run_aggregate: serialize details_json: {:#}", e))?;
56        let sql = "INSERT INTO run_aggregate (
57                run_aggregate_id, started_at, finished_at, duration_ms,
58                config_path, parallel_mode,
59                total_exports, success_count, failed_count, skipped_count,
60                total_rows, total_files, total_bytes, details_json
61            ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14)";
62        self.execute(
63            sql,
64            &[
65                agg.run_aggregate_id.as_str().into(),
66                agg.started_at.as_str().into(),
67                agg.finished_at.as_str().into(),
68                agg.duration_ms.into(),
69                agg.config_path.clone().into(),
70                agg.parallel_mode.as_str().into(),
71                (agg.total_exports as i64).into(),
72                (agg.success_count as i64).into(),
73                (agg.failed_count as i64).into(),
74                (agg.skipped_count as i64).into(),
75                agg.total_rows.into(),
76                agg.total_files.into(),
77                (agg.total_bytes as i64).into(),
78                details.into(),
79            ],
80        )?;
81        Ok(())
82    }
83
84    /// Most-recent aggregates first.
85    #[allow(dead_code)]
86    pub fn get_recent_run_aggregates(&self, limit: usize) -> Result<Vec<RunAggregate>> {
87        let sql = "SELECT run_aggregate_id, started_at, finished_at, duration_ms,
88                    config_path, parallel_mode,
89                    total_exports, success_count, failed_count, skipped_count,
90                    total_rows, total_files, total_bytes, details_json
91             FROM run_aggregate
92             ORDER BY finished_at DESC
93             LIMIT ?1";
94        self.query(sql, &[(limit as i64).into()], |r| {
95            let per_export: Vec<RunAggregateEntry> =
96                serde_json::from_str(&r.text(13)).unwrap_or_default();
97            RunAggregate {
98                run_aggregate_id: r.text(0),
99                started_at: r.text(1),
100                finished_at: r.text(2),
101                duration_ms: r.i64(3),
102                config_path: r.opt_text(4),
103                parallel_mode: r.text(5),
104                total_exports: r.i64(6) as usize,
105                success_count: r.i64(7) as usize,
106                failed_count: r.i64(8) as usize,
107                skipped_count: r.i64(9) as usize,
108                total_rows: r.i64(10),
109                total_files: r.i64(11),
110                total_bytes: r.i64(12) as u64,
111                per_export,
112            }
113        })
114    }
115}
116
117#[cfg(test)]
118mod tests {
119    use super::*;
120
121    fn sample(id: &str) -> RunAggregate {
122        RunAggregate {
123            run_aggregate_id: id.into(),
124            started_at: "2026-04-27T10:00:00Z".into(),
125            finished_at: "2026-04-27T10:11:30Z".into(),
126            duration_ms: 690_000,
127            config_path: Some("pilot.yaml".into()),
128            parallel_mode: "sequential".into(),
129            total_exports: 2,
130            success_count: 1,
131            failed_count: 1,
132            skipped_count: 0,
133            total_rows: 1_500_000,
134            total_files: 12,
135            total_bytes: 750 * 1024 * 1024,
136            per_export: vec![
137                RunAggregateEntry {
138                    export_name: "orders".into(),
139                    status: "success".into(),
140                    run_id: "orders_20260427T100000".into(),
141                    rows: 1_000_000,
142                    files: 10,
143                    bytes: 600 * 1024 * 1024,
144                    duration_ms: 600_000,
145                    mode: "chunked".into(),
146                    error_message: None,
147                },
148                RunAggregateEntry {
149                    export_name: "users".into(),
150                    status: "failed".into(),
151                    run_id: "users_20260427T100000".into(),
152                    rows: 500_000,
153                    files: 2,
154                    bytes: 150 * 1024 * 1024,
155                    duration_ms: 80_000,
156                    mode: "full".into(),
157                    error_message: Some("connection reset".into()),
158                },
159            ],
160        }
161    }
162
163    #[test]
164    fn record_and_query_round_trip() {
165        let s = StateStore::open_in_memory().unwrap();
166        s.record_run_aggregate(&sample("agg_001")).unwrap();
167        s.record_run_aggregate(&sample("agg_002")).unwrap();
168
169        let rows = s.get_recent_run_aggregates(10).unwrap();
170        assert_eq!(rows.len(), 2);
171        let ids: Vec<_> = rows.iter().map(|r| r.run_aggregate_id.as_str()).collect();
172        assert!(ids.contains(&"agg_001"));
173        assert!(ids.contains(&"agg_002"));
174
175        let r = rows
176            .iter()
177            .find(|r| r.run_aggregate_id == "agg_001")
178            .unwrap();
179        assert_eq!(r.total_exports, 2);
180        assert_eq!(r.success_count, 1);
181        assert_eq!(r.failed_count, 1);
182        assert_eq!(r.total_rows, 1_500_000);
183        assert_eq!(r.per_export.len(), 2);
184        assert_eq!(r.per_export[0].export_name, "orders");
185        assert_eq!(
186            r.per_export[1].error_message.as_deref(),
187            Some("connection reset")
188        );
189    }
190
191    #[test]
192    fn limit_is_respected() {
193        let s = StateStore::open_in_memory().unwrap();
194        for i in 0..5 {
195            let mut a = sample(&format!("agg_{i:03}"));
196            a.finished_at = format!("2026-04-27T10:{:02}:00Z", i);
197            s.record_run_aggregate(&a).unwrap();
198        }
199        let rows = s.get_recent_run_aggregates(3).unwrap();
200        assert_eq!(rows.len(), 3);
201        assert_eq!(rows[0].run_aggregate_id, "agg_004");
202        assert_eq!(rows[1].run_aggregate_id, "agg_003");
203        assert_eq!(rows[2].run_aggregate_id, "agg_002");
204    }
205
206    #[test]
207    fn empty_per_export_is_allowed() {
208        let s = StateStore::open_in_memory().unwrap();
209        let mut a = sample("agg_empty");
210        a.per_export.clear();
211        a.total_exports = 0;
212        a.success_count = 0;
213        a.failed_count = 0;
214        s.record_run_aggregate(&a).unwrap();
215
216        let rows = s.get_recent_run_aggregates(10).unwrap();
217        assert_eq!(rows.len(), 1);
218        assert!(rows[0].per_export.is_empty());
219    }
220}
rivet/state/run_aggregate.rs

rivet/state/
run_aggregate.rs
