macp_storage/storage/
file.rs

1use crate::log_store::LogEntry;
2use crate::registry::PersistedSession;
3use macp_core::session::Session;
4use std::fs;
5use std::io;
6use std::path::{Path, PathBuf};
7use tokio::fs as tfs;
8use tokio::io::AsyncWriteExt;
9
10use super::StorageBackend;
11
12pub struct FileBackend {
13    base_dir: PathBuf,
14}
15
16impl FileBackend {
17    pub fn new(base_dir: PathBuf) -> io::Result<Self> {
18        fs::create_dir_all(base_dir.join("sessions"))?;
19        Ok(Self { base_dir })
20    }
21
22    fn session_dir(&self, session_id: &str) -> PathBuf {
23        self.base_dir.join("sessions").join(session_id)
24    }
25
26    pub(crate) fn session_file(&self, session_id: &str) -> PathBuf {
27        self.session_dir(session_id).join("session.json")
28    }
29
30    pub(crate) fn log_file(&self, session_id: &str) -> PathBuf {
31        self.session_dir(session_id).join("log.jsonl")
32    }
33
34    async fn atomic_write(path: &Path, data: &[u8]) -> io::Result<()> {
35        let tmp_path = path.with_extension("json.tmp");
36        // Crash-atomic write: fsync the tmp file BEFORE the rename (otherwise
37        // the rename can be durable while the data is not, leaving a
38        // zero-length or partial file after power loss), then fsync the
39        // parent directory so the rename itself is durable.
40        {
41            let mut file = tfs::File::create(&tmp_path).await?;
42            tokio::io::AsyncWriteExt::write_all(&mut file, data).await?;
43            file.sync_data().await?;
44        }
45        tfs::rename(&tmp_path, path).await?;
46        if let Some(parent) = path.parent() {
47            if let Ok(dir) = tfs::File::open(parent).await {
48                let _ = dir.sync_data().await;
49            }
50        }
51        Ok(())
52    }
53}
54
55#[async_trait::async_trait]
56impl StorageBackend for FileBackend {
57    async fn create_session_storage(&self, session_id: &str) -> io::Result<()> {
58        tfs::create_dir_all(self.session_dir(session_id)).await
59    }
60
61    async fn save_session(&self, session: &Session) -> io::Result<()> {
62        let persisted = PersistedSession::from(session);
63        let bytes = serde_json::to_vec_pretty(&persisted)
64            .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?;
65        Self::atomic_write(&self.session_file(&session.session_id), &bytes).await
66    }
67
68    async fn load_session(&self, session_id: &str) -> io::Result<Option<Session>> {
69        let path = self.session_file(session_id);
70        if tfs::metadata(&path).await.is_err() {
71            return Ok(None);
72        }
73        let bytes = tfs::read(&path).await?;
74        let persisted: PersistedSession = serde_json::from_slice(&bytes)
75            .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?;
76        Ok(Some(Session::from(persisted)))
77    }
78
79    async fn load_all_sessions(&self) -> io::Result<Vec<Session>> {
80        let ids = self.list_session_ids().await?;
81        let mut sessions = Vec::new();
82        for id in ids {
83            match self.load_session(&id).await {
84                Ok(Some(s)) => sessions.push(s),
85                Ok(None) => {}
86                Err(e) => {
87                    eprintln!("warning: failed to load session {id}: {e}; skipping");
88                }
89            }
90        }
91        Ok(sessions)
92    }
93
94    async fn delete_session(&self, session_id: &str) -> io::Result<()> {
95        let dir = self.session_dir(session_id);
96        if tfs::metadata(&dir).await.is_ok() {
97            tfs::remove_dir_all(&dir).await?;
98        }
99        Ok(())
100    }
101
102    async fn list_session_ids(&self) -> io::Result<Vec<String>> {
103        let sessions_dir = self.base_dir.join("sessions");
104        if tfs::metadata(&sessions_dir).await.is_err() {
105            return Ok(vec![]);
106        }
107        let mut ids = Vec::new();
108        let mut entries = tfs::read_dir(&sessions_dir).await?;
109        while let Some(entry) = entries.next_entry().await? {
110            if !entry.file_type().await?.is_dir() {
111                continue;
112            }
113            ids.push(entry.file_name().to_string_lossy().to_string());
114        }
115        Ok(ids)
116    }
117
118    async fn append_log_entry(&self, session_id: &str, entry: &LogEntry) -> io::Result<()> {
119        let path = self.log_file(session_id);
120        let mut line = serde_json::to_string(entry)
121            .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?;
122        line.push('\n');
123
124        let mut file = tfs::OpenOptions::new()
125            .create(true)
126            .append(true)
127            .open(&path)
128            .await?;
129        file.write_all(line.as_bytes()).await?;
130        file.sync_data().await?;
131        Ok(())
132    }
133
134    async fn load_log(&self, session_id: &str) -> io::Result<Vec<LogEntry>> {
135        let path = self.log_file(session_id);
136        if tfs::metadata(&path).await.is_err() {
137            return Ok(vec![]);
138        }
139        let content = tfs::read_to_string(&path).await?;
140        let mut entries = Vec::new();
141        for (line_num, line) in content.lines().enumerate() {
142            if line.trim().is_empty() {
143                continue;
144            }
145            match serde_json::from_str::<LogEntry>(line) {
146                Ok(entry) => entries.push(entry),
147                Err(e) => {
148                    eprintln!(
149                        "warning: failed to parse log entry at {}:{}: {e}; skipping",
150                        path.display(),
151                        line_num + 1
152                    );
153                }
154            }
155        }
156        Ok(entries)
157    }
158
159    async fn replace_log(&self, session_id: &str, entries: &[LogEntry]) -> io::Result<()> {
160        let path = self.log_file(session_id);
161        let mut data = String::new();
162        for entry in entries {
163            let line = serde_json::to_string(entry)
164                .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?;
165            data.push_str(&line);
166            data.push('\n');
167        }
168        let tmp_path = path.with_extension("jsonl.tmp");
169        tfs::write(&tmp_path, data.as_bytes()).await?;
170        tfs::rename(&tmp_path, &path).await
171    }
172}
173
174#[cfg(test)]
175mod tests {
176    use super::*;
177    use crate::log_store::EntryKind;
178
179    use std::collections::HashSet;
180
181    fn sample_session(id: &str) -> Session {
182        Session::builder(id, "macp.mode.decision.v1", "alice")
183            .ttl_expiry(61_000)
184            .ttl_ms(60_000)
185            .started_at_unix_ms(1_000)
186            .mode_state(vec![1, 2, 3])
187            .participants(vec!["alice".into(), "bob".into()])
188            .seen_message_ids(HashSet::from(["m1".into()]))
189            .intent("test intent")
190            .mode_version("1.0.0")
191            .configuration_version("cfg-1")
192            .policy_version("pol-1")
193            .context_id("test-ctx")
194            .roots(vec![macp_pb::pb::Root {
195                uri: "root://1".into(),
196                name: "r1".into(),
197            }])
198            .build()
199    }
200
201    fn sample_entry(id: &str) -> LogEntry {
202        LogEntry {
203            message_id: id.into(),
204            received_at_ms: 1_700_000_000_000,
205            sender: "alice".into(),
206            message_type: "Message".into(),
207            raw_payload: vec![],
208            entry_kind: EntryKind::Incoming,
209            session_id: String::new(),
210            mode: String::new(),
211            macp_version: String::new(),
212            timestamp_unix_ms: 1_700_000_000_000,
213            bound_mode_version: None,
214            semantics_rev: 0,
215            bound_max_suspend_ms: None,
216            compacted_incoming_ordinals: 0,
217        }
218    }
219
220    #[tokio::test]
221    async fn file_backend_session_round_trip() {
222        let dir = tempfile::tempdir().unwrap();
223        let backend = FileBackend::new(dir.path().to_path_buf()).unwrap();
224
225        let session = sample_session("s1");
226        backend.create_session_storage("s1").await.unwrap();
227        backend.save_session(&session).await.unwrap();
228
229        let loaded = backend.load_session("s1").await.unwrap().unwrap();
230        assert_eq!(loaded.session_id, "s1");
231        assert_eq!(loaded.ttl_ms, 60_000);
232        assert_eq!(loaded.mode_version, "1.0.0");
233        assert!(loaded.seen_message_ids.contains("m1"));
234        assert_eq!(loaded.participants, vec!["alice", "bob"]);
235    }
236
237    #[tokio::test]
238    async fn file_backend_log_append_and_load() {
239        let dir = tempfile::tempdir().unwrap();
240        let backend = FileBackend::new(dir.path().to_path_buf()).unwrap();
241
242        backend.create_session_storage("s1").await.unwrap();
243        backend
244            .append_log_entry("s1", &sample_entry("m1"))
245            .await
246            .unwrap();
247        backend
248            .append_log_entry("s1", &sample_entry("m2"))
249            .await
250            .unwrap();
251        backend
252            .append_log_entry("s1", &sample_entry("m3"))
253            .await
254            .unwrap();
255
256        let log = backend.load_log("s1").await.unwrap();
257        assert_eq!(log.len(), 3);
258        assert_eq!(log[0].message_id, "m1");
259        assert_eq!(log[1].message_id, "m2");
260        assert_eq!(log[2].message_id, "m3");
261    }
262
263    #[tokio::test]
264    async fn file_backend_load_all_sessions() {
265        let dir = tempfile::tempdir().unwrap();
266        let backend = FileBackend::new(dir.path().to_path_buf()).unwrap();
267
268        for id in ["s1", "s2", "s3"] {
269            backend.create_session_storage(id).await.unwrap();
270            backend.save_session(&sample_session(id)).await.unwrap();
271        }
272
273        let mut sessions = backend.load_all_sessions().await.unwrap();
274        sessions.sort_by(|a, b| a.session_id.cmp(&b.session_id));
275        assert_eq!(sessions.len(), 3);
276        assert_eq!(sessions[0].session_id, "s1");
277        assert_eq!(sessions[1].session_id, "s2");
278        assert_eq!(sessions[2].session_id, "s3");
279    }
280
281    #[tokio::test]
282    async fn append_only_no_full_rewrite() {
283        let dir = tempfile::tempdir().unwrap();
284        let backend = FileBackend::new(dir.path().to_path_buf()).unwrap();
285        backend.create_session_storage("s1").await.unwrap();
286
287        for i in 0..100 {
288            backend
289                .append_log_entry("s1", &sample_entry(&format!("m{}", i)))
290                .await
291                .unwrap();
292        }
293
294        let content = fs::read_to_string(backend.log_file("s1")).unwrap();
295        let line_count = content.lines().count();
296        assert_eq!(line_count, 100);
297
298        let log = backend.load_log("s1").await.unwrap();
299        assert_eq!(log.len(), 100);
300    }
301
302    #[tokio::test]
303    async fn write_ordering_log_before_session() {
304        let dir = tempfile::tempdir().unwrap();
305        let backend = FileBackend::new(dir.path().to_path_buf()).unwrap();
306        backend.create_session_storage("s1").await.unwrap();
307
308        backend
309            .append_log_entry("s1", &sample_entry("m1"))
310            .await
311            .unwrap();
312
313        let log = backend.load_log("s1").await.unwrap();
314        assert_eq!(log.len(), 1);
315        assert_eq!(log[0].message_id, "m1");
316
317        assert!(backend.load_session("s1").await.unwrap().is_none());
318
319        backend.save_session(&sample_session("s1")).await.unwrap();
320        assert!(backend.load_session("s1").await.unwrap().is_some());
321    }
322
323    #[tokio::test]
324    async fn delete_session_removes_directory() {
325        let dir = tempfile::tempdir().unwrap();
326        let backend = FileBackend::new(dir.path().to_path_buf()).unwrap();
327
328        backend.create_session_storage("s1").await.unwrap();
329        backend.save_session(&sample_session("s1")).await.unwrap();
330        backend
331            .append_log_entry("s1", &sample_entry("m1"))
332            .await
333            .unwrap();
334
335        assert!(backend.load_session("s1").await.unwrap().is_some());
336
337        backend.delete_session("s1").await.unwrap();
338        assert!(backend.load_session("s1").await.unwrap().is_none());
339        assert!(backend.load_log("s1").await.unwrap().is_empty());
340
341        // Idempotent
342        backend.delete_session("s1").await.unwrap();
343    }
344
345    #[tokio::test]
346    async fn list_session_ids_returns_directories() {
347        let dir = tempfile::tempdir().unwrap();
348        let backend = FileBackend::new(dir.path().to_path_buf()).unwrap();
349
350        for id in ["s1", "s2", "s3"] {
351            backend.create_session_storage(id).await.unwrap();
352        }
353
354        let mut ids = backend.list_session_ids().await.unwrap();
355        ids.sort();
356        assert_eq!(ids, vec!["s1", "s2", "s3"]);
357    }
358
359    #[tokio::test]
360    async fn replace_log_atomically_overwrites() {
361        let dir = tempfile::tempdir().unwrap();
362        let backend = FileBackend::new(dir.path().to_path_buf()).unwrap();
363        backend.create_session_storage("s1").await.unwrap();
364
365        for i in 0..10 {
366            backend
367                .append_log_entry("s1", &sample_entry(&format!("m{i}")))
368                .await
369                .unwrap();
370        }
371        assert_eq!(backend.load_log("s1").await.unwrap().len(), 10);
372
373        let replacement = vec![sample_entry("checkpoint")];
374        backend.replace_log("s1", &replacement).await.unwrap();
375
376        let log = backend.load_log("s1").await.unwrap();
377        assert_eq!(log.len(), 1);
378        assert_eq!(log[0].message_id, "checkpoint");
379    }
380
381    #[tokio::test]
382    async fn ttl_ms_backward_compat_deserialization() {
383        let dir = tempfile::tempdir().unwrap();
384        let base = dir.path();
385        let backend = FileBackend::new(base.to_path_buf()).unwrap();
386        backend.create_session_storage("s1").await.unwrap();
387
388        let json = serde_json::json!({
389            "session_id": "s1",
390            "state": "Open",
391            "ttl_expiry": 61000,
392            "started_at_unix_ms": 1000,
393            "resolution": null,
394            "mode": "macp.mode.decision.v1",
395            "mode_state": [],
396            "participants": ["alice"],
397            "seen_message_ids": [],
398            "intent": "",
399            "mode_version": "1.0.0",
400            "configuration_version": "cfg",
401            "policy_version": "pol",
402            "context": [],
403            "roots": [],
404            "initiator_sender": "alice"
405        });
406        fs::write(
407            backend.session_file("s1"),
408            serde_json::to_vec_pretty(&json).unwrap(),
409        )
410        .unwrap();
411
412        let loaded = backend.load_session("s1").await.unwrap().unwrap();
413        assert_eq!(loaded.ttl_ms, 60_000);
414    }
415
416    #[tokio::test]
417    async fn load_log_skips_truncated_trailing_line() {
418        let dir = tempfile::tempdir().unwrap();
419        let backend = FileBackend::new(dir.path().to_path_buf()).unwrap();
420        backend.create_session_storage("s1").await.unwrap();
421
422        backend
423            .append_log_entry("s1", &sample_entry("m1"))
424            .await
425            .unwrap();
426        backend
427            .append_log_entry("s1", &sample_entry("m2"))
428            .await
429            .unwrap();
430
431        // Simulate a torn write: the process died mid-append, leaving the
432        // final line truncated in the middle of the JSON object.
433        let full_line = serde_json::to_string(&sample_entry("m3")).unwrap();
434        let torn = &full_line[..full_line.len() / 2];
435        use std::io::Write;
436        let mut f = fs::OpenOptions::new()
437            .append(true)
438            .open(backend.log_file("s1"))
439            .unwrap();
440        f.write_all(torn.as_bytes()).unwrap();
441        drop(f);
442
443        // load_log's documented intent is to warn and skip unparseable
444        // lines, preserving all prior intact entries.
445        let log = backend.load_log("s1").await.unwrap();
446        assert_eq!(log.len(), 2);
447        assert_eq!(log[0].message_id, "m1");
448        assert_eq!(log[1].message_id, "m2");
449    }
450
451    #[tokio::test]
452    async fn load_session_errors_on_corrupt_session_json() {
453        let dir = tempfile::tempdir().unwrap();
454        let backend = FileBackend::new(dir.path().to_path_buf()).unwrap();
455        backend.create_session_storage("s1").await.unwrap();
456
457        fs::write(backend.session_file("s1"), b"{not valid json").unwrap();
458
459        let err = backend.load_session("s1").await.unwrap_err();
460        assert_eq!(err.kind(), io::ErrorKind::InvalidData);
461    }
462}
macp_storage/storage/file.rs

macp_storage/storage/
file.rs
