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    ///
100    /// (The previous implementation compared against the raw combined log
101    /// index inclusively — non-contiguous, shifting with internal entries,
102    /// and off by one against the RFC's `after_sequence + 1` replay rule.)
103    pub async fn get_incoming_after(
104        &self,
105        session_id: &str,
106        after_sequence: u64,
107    ) -> Result<Vec<(u64, LogEntry)>, u64> {
108        let guard = self.logs.read().await;
109        let Some(entries) = guard.get(session_id) else {
110            return Ok(Vec::new());
111        };
112        // Compaction may have replaced older entries with a checkpoint that
113        // records how many accepted ordinals it discarded; remaining Incoming
114        // entries continue from that base. A resume below the base asks for
115        // history that no longer exists — surfaced as Err(base) so the
116        // transport can return a clear error instead of silently skipping.
117        let base: u64 = entries
118            .iter()
119            .filter(|e| e.entry_kind == EntryKind::Checkpoint)
120            .map(|e| e.compacted_incoming_ordinals)
121            .max()
122            .unwrap_or(0);
123        if after_sequence < base {
124            return Err(base);
125        }
126        Ok(entries
127            .iter()
128            .filter(|e| e.entry_kind == EntryKind::Incoming)
129            .enumerate()
130            .map(|(i, e)| (base + (i + 1) as u64, e))
131            .filter(|(ordinal, _)| *ordinal > after_sequence)
132            .map(|(ordinal, e)| (ordinal, e.clone()))
133            .collect())
134    }
135
136    /// Drop a session's in-memory log (eviction). The durable log remains in
137    /// storage; a later restart replays it if needed.
138    pub async fn remove_session_log(&self, session_id: &str) {
139        let mut guard = self.logs.write().await;
140        guard.remove(session_id);
141    }
142
143    /// Replace a session's in-memory log wholesale — used by compaction so
144    /// memory and storage stay in step (previously only disk was rewritten,
145    /// leaving divergent in-memory history until restart).
146    pub async fn replace_session_log(&self, session_id: &str, entries: Vec<LogEntry>) {
147        let mut guard = self.logs.write().await;
148        guard.insert(session_id.to_string(), entries);
149    }
150}
151
152#[cfg(test)]
153mod tests {
154    use super::*;
155
156    fn entry(id: &str, kind: EntryKind) -> LogEntry {
157        LogEntry {
158            message_id: id.into(),
159            received_at_ms: 1_700_000_000_000,
160            sender: "test".into(),
161            message_type: "Message".into(),
162            raw_payload: vec![],
163            entry_kind: kind,
164            session_id: String::new(),
165            mode: String::new(),
166            macp_version: String::new(),
167            timestamp_unix_ms: 1_700_000_000_000,
168            bound_mode_version: None,
169            semantics_rev: 0,
170            bound_max_suspend_ms: None,
171            compacted_incoming_ordinals: 0,
172        }
173    }
174
175    #[tokio::test]
176    async fn create_append_get_round_trip() {
177        let store = LogStore::new();
178        store.create_session_log("s1").await;
179        store.append("s1", entry("m1", EntryKind::Incoming)).await;
180        store.append("s1", entry("m2", EntryKind::Incoming)).await;
181
182        let log = store.get_log("s1").await.unwrap();
183        assert_eq!(log.len(), 2);
184        assert_eq!(log[0].message_id, "m1");
185        assert_eq!(log[1].message_id, "m2");
186    }
187
188    #[tokio::test]
189    async fn get_incoming_after_uses_accepted_ordinals_exclusive() {
190        let store = LogStore::new();
191        store.create_session_log("s1").await;
192        store.append("s1", entry("m0", EntryKind::Incoming)).await; // ordinal 1
193        store.append("s1", entry("m1", EntryKind::Internal)).await; // no ordinal
194        store.append("s1", entry("m2", EntryKind::Incoming)).await; // ordinal 2
195        store.append("s1", entry("m3", EntryKind::Incoming)).await; // ordinal 3
196        store.append("s1", entry("m4", EntryKind::Checkpoint)).await; // no ordinal
197
198        // after_sequence=0: from the start (RFC-0006 §3.2), all Incoming,
199        // 1-based contiguous ordinals unaffected by interleaved internal
200        // entries.
201        let all = store.get_incoming_after("s1", 0).await.unwrap();
202        assert_eq!(all.len(), 3);
203        assert_eq!((all[0].0, all[0].1.message_id.as_str()), (1, "m0"));
204        assert_eq!((all[1].0, all[1].1.message_id.as_str()), (2, "m2"));
205        assert_eq!((all[2].0, all[2].1.message_id.as_str()), (3, "m3"));
206
207        // after_sequence is EXCLUSIVE: a client that saw ordinal 2 resumes
208        // with after=2 and receives only ordinal 3 (no re-delivery).
209        let after2 = store.get_incoming_after("s1", 2).await.unwrap();
210        assert_eq!(after2.len(), 1);
211        assert_eq!(after2[0].0, 3);
212        assert_eq!(after2[0].1.message_id, "m3");
213
214        // nonexistent session returns empty
215        let empty = store.get_incoming_after("nope", 0).await.unwrap();
216        assert!(empty.is_empty());
217    }
218
219    #[tokio::test]
220    async fn get_incoming_after_ordinals_survive_compaction() {
221        let store = LogStore::new();
222        store.create_session_log("s1").await;
223        // A compaction checkpoint that discarded 5 accepted ordinals, then
224        // two post-compaction accepted entries: their ordinals continue at 6.
225        let mut cp = entry("cp", EntryKind::Checkpoint);
226        cp.compacted_incoming_ordinals = 5;
227        store.append("s1", cp).await;
228        store.append("s1", entry("m6", EntryKind::Incoming)).await;
229        store.append("s1", entry("m7", EntryKind::Incoming)).await;
230
231        let after5 = store.get_incoming_after("s1", 5).await.unwrap();
232        assert_eq!(after5.len(), 2);
233        assert_eq!(after5[0].0, 6);
234        assert_eq!(after5[1].0, 7);
235
236        let after6 = store.get_incoming_after("s1", 6).await.unwrap();
237        assert_eq!(after6.len(), 1);
238        assert_eq!(after6[0].1.message_id, "m7");
239
240        // Resuming below the compaction base is an error (history gone),
241        // never a silent skip.
242        assert!(matches!(store.get_incoming_after("s1", 3).await, Err(5)));
243    }
244}
macp_storage/log_store.rs

macp_storage/
log_store.rs
