macp_storage/storage/
compaction.rs

1use crate::log_store::{EntryKind, LogEntry};
2use crate::registry::PersistedSession;
3use macp_core::session::Session;
4use std::io;
5
6use super::StorageBackend;
7
8/// Compact a session's log into a single checkpoint entry.
9///
10/// This replaces all existing log entries with a single `Checkpoint` entry
11/// containing the serialized session state. Should only be called on sessions
12/// in terminal state (Resolved/Expired/Cancelled).
13pub async fn compact_session_log(
14    storage: &dyn StorageBackend,
15    session_id: &str,
16    session: &Session,
17    discarded_incoming_ordinals: u64,
18) -> io::Result<LogEntry> {
19    let persisted = PersistedSession::from(session);
20    let raw_payload = serde_json::to_vec(&persisted)
21        .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?;
22
23    let now = chrono::Utc::now().timestamp_millis();
24    let checkpoint = LogEntry {
25        message_id: String::new(),
26        received_at_ms: now,
27        sender: "_runtime".into(),
28        message_type: "Checkpoint".into(),
29        raw_payload,
30        entry_kind: EntryKind::Checkpoint,
31        session_id: session_id.into(),
32        mode: session.mode.clone(),
33        macp_version: String::new(),
34        timestamp_unix_ms: now,
35        // Checkpoints carry the full serialized session (including its bound
36        // mode_version), so no separate binding record is needed here.
37        bound_mode_version: None,
38        semantics_rev: 0,
39        bound_max_suspend_ms: None,
40        // Preserve the passive-subscribe sequence across compaction: the
41        // checkpoint records how many accepted ordinals it replaced, so
42        // post-compaction entries keep contiguous client-visible ordinals
43        // and resumes below the base get an explicit error (RFC-0006 §3.2).
44        compacted_incoming_ordinals: discarded_incoming_ordinals,
45    };
46
47    storage
48        .replace_log(session_id, std::slice::from_ref(&checkpoint))
49        .await?;
50    Ok(checkpoint)
51}
52
53#[cfg(test)]
54mod tests {
55    use super::*;
56    use crate::storage::FileBackend;
57
58    use std::collections::HashSet;
59
60    fn sample_session(id: &str) -> Session {
61        Session::builder(id, "macp.mode.decision.v1", "alice")
62            .ttl_expiry(61_000)
63            .ttl_ms(60_000)
64            .started_at_unix_ms(1_000)
65            .mode_state(vec![1, 2, 3])
66            .participants(vec!["alice".into(), "bob".into()])
67            .seen_message_ids(HashSet::from(["m1".into()]))
68            .intent("test intent")
69            .mode_version("1.0.0")
70            .configuration_version("cfg-1")
71            .policy_version("pol-1")
72            .context_id("test-ctx")
73            .roots(vec![macp_pb::pb::Root {
74                uri: "root://1".into(),
75                name: "r1".into(),
76            }])
77            .build()
78    }
79
80    fn sample_entry(id: &str) -> LogEntry {
81        LogEntry {
82            message_id: id.into(),
83            received_at_ms: 1_700_000_000_000,
84            sender: "alice".into(),
85            message_type: "Message".into(),
86            raw_payload: vec![],
87            entry_kind: EntryKind::Incoming,
88            session_id: String::new(),
89            mode: String::new(),
90            macp_version: String::new(),
91            timestamp_unix_ms: 1_700_000_000_000,
92            bound_mode_version: None,
93            semantics_rev: 0,
94            bound_max_suspend_ms: None,
95            compacted_incoming_ordinals: 0,
96        }
97    }
98
99    #[tokio::test]
100    async fn compaction_replaces_log_with_single_checkpoint() {
101        let dir = tempfile::tempdir().unwrap();
102        let backend = FileBackend::new(dir.path().to_path_buf()).unwrap();
103        backend.create_session_storage("s1").await.unwrap();
104
105        for i in 0..5 {
106            backend
107                .append_log_entry("s1", &sample_entry(&format!("m{i}")))
108                .await
109                .unwrap();
110        }
111        assert_eq!(backend.load_log("s1").await.unwrap().len(), 5);
112
113        let session = sample_session("s1");
114        let checkpoint = compact_session_log(&backend, "s1", &session, 5)
115            .await
116            .unwrap();
117
118        assert_eq!(checkpoint.entry_kind, EntryKind::Checkpoint);
119        assert_eq!(checkpoint.compacted_incoming_ordinals, 5);
120        assert_eq!(checkpoint.session_id, "s1");
121        assert_eq!(checkpoint.mode, "macp.mode.decision.v1");
122
123        // The checkpoint payload must round-trip to the compacted session.
124        let persisted: PersistedSession = serde_json::from_slice(&checkpoint.raw_payload).unwrap();
125        assert_eq!(persisted.session_id, "s1");
126        assert_eq!(persisted.mode, "macp.mode.decision.v1");
127        assert_eq!(persisted.participants, vec!["alice", "bob"]);
128        assert_eq!(persisted.ttl_ms, 60_000);
129
130        // The durable log now contains exactly the checkpoint entry.
131        let log = backend.load_log("s1").await.unwrap();
132        assert_eq!(log.len(), 1);
133        assert_eq!(log[0].entry_kind, EntryKind::Checkpoint);
134        assert_eq!(log[0].compacted_incoming_ordinals, 5);
135        let reloaded: PersistedSession = serde_json::from_slice(&log[0].raw_payload).unwrap();
136        assert_eq!(reloaded.session_id, "s1");
137    }
138
139    #[tokio::test]
140    async fn append_after_compaction_keeps_checkpoint_first() {
141        let dir = tempfile::tempdir().unwrap();
142        let backend = FileBackend::new(dir.path().to_path_buf()).unwrap();
143        backend.create_session_storage("s1").await.unwrap();
144
145        for i in 0..3 {
146            backend
147                .append_log_entry("s1", &sample_entry(&format!("m{i}")))
148                .await
149                .unwrap();
150        }
151
152        let session = sample_session("s1");
153        compact_session_log(&backend, "s1", &session, 3)
154            .await
155            .unwrap();
156
157        backend
158            .append_log_entry("s1", &sample_entry("post-compaction"))
159            .await
160            .unwrap();
161
162        let log = backend.load_log("s1").await.unwrap();
163        assert_eq!(log.len(), 2);
164        assert_eq!(log[0].entry_kind, EntryKind::Checkpoint);
165        assert_eq!(log[1].entry_kind, EntryKind::Incoming);
166        assert_eq!(log[1].message_id, "post-compaction");
167    }
168}
macp_storage/storage/compaction.rs

macp_storage/storage/
compaction.rs
