macp_storage/
log_store.rs

1use std::collections::HashMap;
2use tokio::sync::RwLock;
3
4#[derive(Clone, Debug, PartialEq, serde::Serialize, serde::Deserialize)]
5pub enum EntryKind {
6    Incoming,
7    Internal,
8    Checkpoint,
9}
10
11#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
12pub struct LogEntry {
13    pub message_id: String,
14    pub received_at_ms: i64,
15    pub sender: String,
16    pub message_type: String,
17    pub raw_payload: Vec<u8>,
18    pub entry_kind: EntryKind,
19    #[serde(default)]
20    pub session_id: String,
21    #[serde(default)]
22    pub mode: String,
23    #[serde(default)]
24    pub macp_version: String,
25    /// Original envelope timestamp for replay determinism.
26    #[serde(default)]
27    pub timestamp_unix_ms: i64,
28    /// Mode version bound at SessionStart acceptance when the payload's own
29    /// `mode_version` was empty (non-strict extension modes take the registered
30    /// descriptor's version). Replay uses this recorded binding and never
31    /// re-derives it from the live registry. `None` on legacy entries and on
32    /// entries whose payload carried the version explicitly — legacy histories
33    /// keep their original (empty-version) binding semantics.
34    #[serde(default)]
35    pub bound_mode_version: Option<String>,
36    /// Session-semantics revision recorded on the SessionStart entry (see
37    /// `macp_core::session::CURRENT_SEMANTICS_REV`). Legacy entries
38    /// deserialize as 0; replay applies the recorded revision so old
39    /// histories keep the acceptance-time behavior they were written under.
40    #[serde(default)]
41    pub semantics_rev: u32,
42    /// Suspension cap resolved and bound at SessionStart acceptance
43    /// (payload value, or the runtime default when the payload carried 0).
44    /// Replay uses this recorded value, never live configuration
45    /// (RFC-MACP-0003 §2). `None` on legacy entries — those sessions keep
46    /// default-cap semantics.
47    #[serde(default)]
48    pub bound_max_suspend_ms: Option<i64>,
49    /// On `Checkpoint` entries produced by log compaction: how many accepted
50    /// (Incoming) envelope ordinals the compaction discarded. Ordinals of
51    /// entries after the checkpoint continue from this base, keeping the
52    /// passive-subscribe sequence stable across compaction and restart.
53    /// `0` on all other entries and on legacy checkpoints.
54    #[serde(default)]
55    pub compacted_incoming_ordinals: u64,
56}
57
58pub struct LogStore {
59    logs: RwLock<HashMap<String, Vec<LogEntry>>>,
60}
61
62impl Default for LogStore {
63    fn default() -> Self {
64        Self::new()
65    }
66}
67
68impl LogStore {
69    pub fn new() -> Self {
70        Self {
71            logs: RwLock::new(HashMap::new()),
72        }
73    }
74
75    pub async fn create_session_log(&self, session_id: &str) {
76        let mut guard = self.logs.write().await;
77        guard.entry(session_id.to_string()).or_default();
78    }
79
80    pub async fn append(&self, session_id: &str, entry: LogEntry) {
81        let mut guard = self.logs.write().await;
82        guard.entry(session_id.to_string()).or_default().push(entry);
83    }
84
85    pub async fn get_log(&self, session_id: &str) -> Option<Vec<LogEntry>> {
86        let guard = self.logs.read().await;
87        guard.get(session_id).cloned()
88    }
89
90    /// Returns accepted (Incoming) log entries strictly after `after_sequence`,
91    /// paired with their **1-based accepted-envelope ordinal**.
92    ///
93    /// Sequence contract (RFC-MACP-0006 §3.2): the per-session sequence is the
94    /// ordinal of accepted session envelopes — the first accepted envelope
95    /// (SessionStart) is 1. `after_sequence` is EXCLUSIVE: `0` replays from
96    /// the start, `n` resumes after the n-th accepted envelope. Internal and
97    /// Checkpoint entries never consume ordinals, so client-visible sequences
98    /// are contiguous and stable regardless of interleaved internal records —
99    /// concretely, `SessionSuspend`/`SessionResume`/`SessionCancel`/`TtlExpired`
100    /// entries (`EntryKind::Internal`, per §3.2:117) are exactly the kind of
101    /// interleaved record this filter excludes from the sequence.
102    ///
103    /// (The previous implementation compared against the raw combined log
104    /// index inclusively — non-contiguous, shifting with internal entries,
105    /// and off by one against the RFC's `after_sequence + 1` replay rule.)
106    pub async fn get_incoming_after(
107        &self,
108        session_id: &str,
109        after_sequence: u64,
110    ) -> Result<Vec<(u64, LogEntry)>, u64> {
111        let guard = self.logs.read().await;
112        let Some(entries) = guard.get(session_id) else {
113            return Ok(Vec::new());
114        };
115        // Compaction may have replaced older entries with a checkpoint that
116        // records how many accepted ordinals it discarded; remaining Incoming
117        // entries continue from that base. A resume below the base asks for
118        // history that no longer exists — surfaced as Err(base) so the
119        // transport can return a clear error instead of silently skipping.
120        let base: u64 = entries
121            .iter()
122            .filter(|e| e.entry_kind == EntryKind::Checkpoint)
123            .map(|e| e.compacted_incoming_ordinals)
124            .max()
125            .unwrap_or(0);
126        if after_sequence < base {
127            return Err(base);
128        }
129        Ok(entries
130            .iter()
131            .filter(|e| e.entry_kind == EntryKind::Incoming)
132            .enumerate()
133            .map(|(i, e)| (base + (i + 1) as u64, e))
134            .filter(|(ordinal, _)| *ordinal > after_sequence)
135            .map(|(ordinal, e)| (ordinal, e.clone()))
136            .collect())
137    }
138
139    /// Drop a session's in-memory log (eviction). The durable log remains in
140    /// storage; a later restart replays it if needed.
141    pub async fn remove_session_log(&self, session_id: &str) {
142        let mut guard = self.logs.write().await;
143        guard.remove(session_id);
144    }
145
146    /// Replace a session's in-memory log wholesale — used by compaction so
147    /// memory and storage stay in step (previously only disk was rewritten,
148    /// leaving divergent in-memory history until restart).
149    pub async fn replace_session_log(&self, session_id: &str, entries: Vec<LogEntry>) {
150        let mut guard = self.logs.write().await;
151        guard.insert(session_id.to_string(), entries);
152    }
153}
154
155#[cfg(test)]
156mod tests {
157    use super::*;
158
159    fn entry(id: &str, kind: EntryKind) -> LogEntry {
160        LogEntry {
161            message_id: id.into(),
162            received_at_ms: 1_700_000_000_000,
163            sender: "test".into(),
164            message_type: "Message".into(),
165            raw_payload: vec![],
166            entry_kind: kind,
167            session_id: String::new(),
168            mode: String::new(),
169            macp_version: String::new(),
170            timestamp_unix_ms: 1_700_000_000_000,
171            bound_mode_version: None,
172            semantics_rev: 0,
173            bound_max_suspend_ms: None,
174            compacted_incoming_ordinals: 0,
175        }
176    }
177
178    #[tokio::test]
179    async fn create_append_get_round_trip() {
180        let store = LogStore::new();
181        store.create_session_log("s1").await;
182        store.append("s1", entry("m1", EntryKind::Incoming)).await;
183        store.append("s1", entry("m2", EntryKind::Incoming)).await;
184
185        let log = store.get_log("s1").await.unwrap();
186        assert_eq!(log.len(), 2);
187        assert_eq!(log[0].message_id, "m1");
188        assert_eq!(log[1].message_id, "m2");
189    }
190
191    #[tokio::test]
192    async fn get_incoming_after_uses_accepted_ordinals_exclusive() {
193        let store = LogStore::new();
194        store.create_session_log("s1").await;
195        store.append("s1", entry("m0", EntryKind::Incoming)).await; // ordinal 1
196        store.append("s1", entry("m1", EntryKind::Internal)).await; // no ordinal
197        store.append("s1", entry("m2", EntryKind::Incoming)).await; // ordinal 2
198        store.append("s1", entry("m3", EntryKind::Incoming)).await; // ordinal 3
199        store.append("s1", entry("m4", EntryKind::Checkpoint)).await; // no ordinal
200
201        // after_sequence=0: from the start (RFC-0006 §3.2), all Incoming,
202        // 1-based contiguous ordinals unaffected by interleaved internal
203        // entries.
204        let all = store.get_incoming_after("s1", 0).await.unwrap();
205        assert_eq!(all.len(), 3);
206        assert_eq!((all[0].0, all[0].1.message_id.as_str()), (1, "m0"));
207        assert_eq!((all[1].0, all[1].1.message_id.as_str()), (2, "m2"));
208        assert_eq!((all[2].0, all[2].1.message_id.as_str()), (3, "m3"));
209
210        // after_sequence is EXCLUSIVE: a client that saw ordinal 2 resumes
211        // with after=2 and receives only ordinal 3 (no re-delivery).
212        let after2 = store.get_incoming_after("s1", 2).await.unwrap();
213        assert_eq!(after2.len(), 1);
214        assert_eq!(after2[0].0, 3);
215        assert_eq!(after2[0].1.message_id, "m3");
216
217        // nonexistent session returns empty
218        let empty = store.get_incoming_after("nope", 0).await.unwrap();
219        assert!(empty.is_empty());
220    }
221
222    #[tokio::test]
223    async fn get_incoming_after_ordinals_survive_compaction() {
224        let store = LogStore::new();
225        store.create_session_log("s1").await;
226        // A compaction checkpoint that discarded 5 accepted ordinals, then
227        // two post-compaction accepted entries: their ordinals continue at 6.
228        let mut cp = entry("cp", EntryKind::Checkpoint);
229        cp.compacted_incoming_ordinals = 5;
230        store.append("s1", cp).await;
231        store.append("s1", entry("m6", EntryKind::Incoming)).await;
232        store.append("s1", entry("m7", EntryKind::Incoming)).await;
233
234        let after5 = store.get_incoming_after("s1", 5).await.unwrap();
235        assert_eq!(after5.len(), 2);
236        assert_eq!(after5[0].0, 6);
237        assert_eq!(after5[1].0, 7);
238
239        let after6 = store.get_incoming_after("s1", 6).await.unwrap();
240        assert_eq!(after6.len(), 1);
241        assert_eq!(after6[0].1.message_id, "m7");
242
243        // Resuming below the compaction base is an error (history gone),
244        // never a silent skip.
245        assert!(matches!(store.get_incoming_after("s1", 3).await, Err(5)));
246    }
247}
macp_storage/log_store.rs

macp_storage/
log_store.rs
