onlyne_client/reconcile/
ledger.rs

1use std::collections::HashMap;
2use std::sync::Mutex;
3
4use onlyne_store::session::{FaultRecord, SessionLedger, SessionRecord, VersionedSession};
5
6use crate::backend::SessionRef;
7
8/// In-memory [`SessionLedger`] for tests and for callers that hold the live
9/// [`SessionRef`] map next to the bridge.
10#[derive(Debug, Default)]
11pub struct MemoryLedger {
12    sessions: Mutex<HashMap<String, (SessionRecord, VersionedSession)>>,
13    known: Mutex<HashMap<String, i64>>,
14    faults: Mutex<Vec<FaultRecord>>,
15    events: Mutex<Vec<(String, serde_json::Value)>>,
16    alerts: Mutex<Vec<String>>,
17    next_fault_id: Mutex<i64>,
18}
19
20impl MemoryLedger {
21    pub fn new() -> Self {
22        Self::default()
23    }
24
25    /// Register a task the bridge may seed a row for, with its attempt count.
26    pub fn track_task(&self, task_id: &str, attempt: i64) {
27        self.known
28            .lock()
29            .unwrap()
30            .insert(task_id.to_string(), attempt);
31    }
32
33    /// Insert a live session ref the bridge prefers over the stored reference.
34    pub fn track_session(&self, session: SessionRef) {
35        let mut sessions = self.sessions.lock().unwrap();
36        let entry = sessions.entry(session.task_id.clone()).or_insert_with(|| {
37            let record = SessionRecord {
38                session_id: session.task_id.clone(),
39                task_id: session.task_id.clone(),
40                agent_state: String::new(),
41                delivery_state: String::new(),
42                resource_state: String::new(),
43                recovery_substate: String::new(),
44                desired_json: String::new(),
45                observed_json: String::new(),
46                generation: 0,
47                seq: -1,
48                backend_ref: String::new(),
49                mismatch_count: 0,
50                updated_at: 0,
51            };
52            let stored = VersionedSession {
53                agent_state: String::new(),
54                delivery_state: String::new(),
55                resource_state: String::new(),
56                recovery_substate: String::new(),
57                desired_json: String::new(),
58                observed_json: String::new(),
59                generation: 0,
60                seq: -1,
61                backend_ref: String::new(),
62                mismatch_count: 0,
63                updated_at: 0,
64            };
65            (record, stored)
66        });
67        entry.0.backend_ref = serde_json::to_string(&session).unwrap_or_else(|_| "{}".into());
68    }
69
70    pub fn events(&self) -> Vec<(String, serde_json::Value)> {
71        self.events.lock().unwrap().clone()
72    }
73
74    pub fn alerts(&self) -> Vec<String> {
75        self.alerts.lock().unwrap().clone()
76    }
77}
78
79impl SessionLedger for MemoryLedger {
80    fn get_session(&self, task_id: &str) -> anyhow::Result<Option<SessionRecord>> {
81        Ok(self
82            .sessions
83            .lock()
84            .unwrap()
85            .get(task_id)
86            .map(|(record, _)| record.clone()))
87    }
88
89    fn upsert_session(&self, task_id: &str, version: &VersionedSession) -> anyhow::Result<bool> {
90        let mut sessions = self.sessions.lock().unwrap();
91        let entry = sessions.entry(task_id.to_string()).or_insert_with(|| {
92            let record = SessionRecord {
93                session_id: task_id.to_string(),
94                task_id: task_id.to_string(),
95                agent_state: String::new(),
96                delivery_state: String::new(),
97                resource_state: String::new(),
98                recovery_substate: String::new(),
99                desired_json: String::new(),
100                observed_json: String::new(),
101                generation: 0,
102                seq: -1,
103                backend_ref: String::new(),
104                mismatch_count: 0,
105                updated_at: 0,
106            };
107            let stored = VersionedSession {
108                agent_state: String::new(),
109                delivery_state: String::new(),
110                resource_state: String::new(),
111                recovery_substate: String::new(),
112                desired_json: String::new(),
113                observed_json: String::new(),
114                generation: 0,
115                seq: -1,
116                backend_ref: String::new(),
117                mismatch_count: 0,
118                updated_at: 0,
119            };
120            (record, stored)
121        });
122        let (record, _) = &*entry;
123        let newer = version.generation > record.generation
124            || (version.generation == record.generation && version.seq > record.seq);
125        // An empty row (generation 0, seq -1) accepts the first write
126        // unconditionally so a seed can land.
127        let is_seed = record.generation == 0 && record.seq == -1;
128        if !newer && !is_seed {
129            return Ok(false);
130        }
131        entry.0 = SessionRecord {
132            session_id: task_id.to_string(),
133            task_id: task_id.to_string(),
134            agent_state: version.agent_state.clone(),
135            delivery_state: version.delivery_state.clone(),
136            resource_state: version.resource_state.clone(),
137            recovery_substate: version.recovery_substate.clone(),
138            desired_json: version.desired_json.clone(),
139            observed_json: version.observed_json.clone(),
140            generation: version.generation,
141            seq: version.seq,
142            backend_ref: version.backend_ref.clone(),
143            mismatch_count: version.mismatch_count,
144            updated_at: version.updated_at,
145        };
146        entry.1 = version.clone();
147        Ok(true)
148    }
149
150    fn task_is_known(&self, task_id: &str) -> anyhow::Result<bool> {
151        let known = self.known.lock().unwrap();
152        if known.contains_key(task_id) {
153            return Ok(true);
154        }
155        Ok(self.sessions.lock().unwrap().contains_key(task_id))
156    }
157
158    fn task_attempt(&self, task_id: &str) -> anyhow::Result<i64> {
159        Ok(self
160            .known
161            .lock()
162            .unwrap()
163            .get(task_id)
164            .copied()
165            .unwrap_or(0))
166    }
167
168    fn list_faults(&self, task_id: &str) -> anyhow::Result<Vec<FaultRecord>> {
169        Ok(self
170            .faults
171            .lock()
172            .unwrap()
173            .iter()
174            .filter(|f| f.task_id == task_id)
175            .cloned()
176            .collect())
177    }
178
179    fn insert_fault(&self, fault: &FaultRecord) -> anyhow::Result<i64> {
180        let mut next = self.next_fault_id.lock().unwrap();
181        *next += 1;
182        let id = *next;
183        let mut faults = self.faults.lock().unwrap();
184        faults.push(FaultRecord {
185            id,
186            ..fault.clone()
187        });
188        Ok(id)
189    }
190
191    fn emit(&self, kind: &str, data: serde_json::Value) {
192        self.events.lock().unwrap().push((kind.to_string(), data));
193    }
194
195    fn note_alert(&self, line: String) {
196        self.alerts.lock().unwrap().push(line);
197    }
198}
onlyne_client/reconcile/ledger.rs

onlyne_client/reconcile/
ledger.rs
