rivet/state/
file_log.rs

1use crate::error::Result;
2
3use super::StateStore;
4
5/// One row from `file_log` (formerly `file_manifest`; renamed in schema v8).
6#[derive(Debug, serde::Serialize)]
7#[allow(dead_code)]
8pub struct FileRecord {
9    pub run_id: String,
10    pub export_name: String,
11    pub file_name: String,
12    pub row_count: i64,
13    pub bytes: i64,
14    pub format: String,
15    pub compression: Option<String>,
16    pub created_at: String,
17}
18
19/// File log store — reads and writes `file_log`.
20///
21/// Historical note: this table was named `file_manifest` prior to schema v8.
22/// The name was reclaimed for the 0.7.0 cloud-output JSON manifest contract;
23/// the internal SQLite log was renamed to `file_log` to remove the overload.
24///
25/// Invariant I2 (Write Before Log) governs when `record_file` is called:
26/// only after a destination write succeeds.  Failed writes produce no log entry.
27/// Invariant I7 (File-Log Failure Is Non-Fatal) means callers use `let _ = record_file(...)`.
28impl StateStore {
29    #[allow(clippy::too_many_arguments)]
30    pub fn record_file(
31        &self,
32        run_id: &str,
33        export_name: &str,
34        file_name: &str,
35        row_count: i64,
36        bytes: i64,
37        format: &str,
38        compression: Option<&str>,
39    ) -> Result<()> {
40        self.execute(
41            "INSERT INTO file_log (run_id, export_name, file_name, row_count, bytes, format, compression, created_at) \
42             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
43            &[
44                run_id.into(),
45                export_name.into(),
46                file_name.into(),
47                row_count.into(),
48                bytes.into(),
49                format.into(),
50                compression.into(),
51                chrono::Utc::now().to_rfc3339().into(),
52            ],
53        )?;
54        Ok(())
55    }
56
57    pub fn get_files(&self, export_name: Option<&str>, limit: usize) -> Result<Vec<FileRecord>> {
58        let cols =
59            "run_id, export_name, file_name, row_count, bytes, format, compression, created_at";
60        match export_name {
61            Some(name) => self.query(
62                &format!(
63                    "SELECT {cols} FROM file_log WHERE export_name = ?1 ORDER BY id DESC LIMIT ?2"
64                ),
65                &[name.into(), (limit as i64).into()],
66                file_record,
67            ),
68            None => self.query(
69                &format!("SELECT {cols} FROM file_log ORDER BY id DESC LIMIT ?1"),
70                &[(limit as i64).into()],
71                file_record,
72            ),
73        }
74    }
75
76    /// Every `file_log` row for one `run_id`, in write order. Unlike `chunk_task`
77    /// (one `file_name` per chunk, the FIRST rotation sibling only), file_log
78    /// records EVERY committed part — including all `max_file_size` rotation
79    /// siblings — with its real `bytes`, written per-part before `complete_chunk_
80    /// task`. This is the crash-consistent source the chunked resume reconstructs a
81    /// COMPLETE destination manifest from when none exists (round-5 fix).
82    pub fn list_files_for_run(&self, run_id: &str) -> Result<Vec<FileRecord>> {
83        let cols =
84            "run_id, export_name, file_name, row_count, bytes, format, compression, created_at";
85        self.query(
86            &format!("SELECT {cols} FROM file_log WHERE run_id = ?1 ORDER BY id ASC"),
87            &[run_id.into()],
88            file_record,
89        )
90    }
91}
92
93/// The `FileRecord` projection, written once for both backends.
94fn file_record(r: &dyn super::row::StateRow) -> FileRecord {
95    FileRecord {
96        run_id: r.text(0),
97        export_name: r.text(1),
98        file_name: r.text(2),
99        row_count: r.i64(3),
100        bytes: r.i64(4),
101        format: r.text(5),
102        compression: r.opt_text(6),
103        created_at: r.text(7),
104    }
105}
106
107#[cfg(test)]
108mod tests {
109    use super::*;
110
111    fn store() -> StateStore {
112        StateStore::open_in_memory().expect("in-memory store")
113    }
114
115    #[test]
116    fn record_and_query_files() {
117        let s = store();
118        s.record_file(
119            "run_001",
120            "orders",
121            "orders_20260329.parquet",
122            50000,
123            4096,
124            "parquet",
125            Some("zstd"),
126        )
127        .unwrap();
128        s.record_file(
129            "run_001",
130            "orders",
131            "orders_20260329_chunk1.parquet",
132            25000,
133            2048,
134            "parquet",
135            Some("zstd"),
136        )
137        .unwrap();
138        s.record_file(
139            "run_002",
140            "users",
141            "users_20260329.csv",
142            1000,
143            500,
144            "csv",
145            None,
146        )
147        .unwrap();
148
149        let files = s.get_files(Some("orders"), 10).unwrap();
150        assert_eq!(files.len(), 2);
151        assert_eq!(files[0].run_id, "run_001");
152        assert_eq!(files[0].row_count, 25000);
153
154        let all = s.get_files(None, 10).unwrap();
155        assert_eq!(all.len(), 3);
156    }
157
158    #[test]
159    fn files_limit_works() {
160        let s = store();
161        for i in 0..10 {
162            s.record_file(
163                &format!("r{}", i),
164                "t",
165                &format!("f{}.parquet", i),
166                i,
167                i * 100,
168                "parquet",
169                None,
170            )
171            .unwrap();
172        }
173        let files = s.get_files(Some("t"), 3).unwrap();
174        assert_eq!(files.len(), 3);
175    }
176}
rivet/state/file_log.rs

rivet/state/
file_log.rs
