magi_code/sessions/
manager.rs

1use super::metadata::{SessionMetadataRecord, SessionMetadataSummary};
2use super::read::validate_session_id;
3use super::store::{prepare_session_root, primary_path};
4use crate::persistence::CrossProcessFileLock;
5use std::{
6    collections::HashMap,
7    fmt,
8    path::{Path, PathBuf},
9    sync::{Arc, Mutex, OnceLock, Weak},
10};
11use uuid::Uuid;
12
13#[derive(Debug, Clone, PartialEq, Eq)]
14pub(crate) struct SessionInternalDiagnostic {
15    pub(crate) session_id: Option<String>,
16    pub(crate) message: String,
17}
18
19#[derive(Debug, Clone, PartialEq, Eq)]
20pub(crate) struct SessionListReport {
21    pub(crate) summaries: Vec<SessionMetadataSummary>,
22    pub(crate) diagnostics: Vec<SessionInternalDiagnostic>,
23}
24
25#[derive(Debug, Clone)]
26pub struct SessionManager {
27    pub(in crate::sessions) root: PathBuf,
28}
29
30impl SessionManager {
31    #[must_use]
32    pub fn new(root: PathBuf) -> Self {
33        Self { root }
34    }
35
36    pub(crate) fn create(&self) -> anyhow::Result<Session> {
37        prepare_session_root(&self.root)?;
38        let id = Uuid::new_v4().to_string();
39        let path = self.path_for_valid_id(&id)?;
40        // Do not create JSONL until first event append; avoids orphan session files.
41        Ok(Session::new(id, path))
42    }
43
44    pub fn open(&self, id: impl Into<String>) -> anyhow::Result<Session> {
45        let id = validate_session_id(id.into())?;
46        Ok(Session::new(id.clone(), self.path_for_valid_id(&id)?))
47    }
48
49    pub(crate) fn open_existing(&self, id: impl Into<String>) -> anyhow::Result<Session> {
50        let session = self.open(id)?;
51        super::store::open_existing_primary(&self.root, &session.id)?
52            .ok_or_else(|| anyhow::anyhow!("session JSONL is missing"))?;
53        Ok(session)
54    }
55
56    #[cfg(test)]
57    pub(crate) fn list(&self) -> anyhow::Result<Vec<Session>> {
58        Ok(self
59            .list_metadata_summaries()?
60            .into_iter()
61            .map(|summary| summary.session)
62            .collect())
63    }
64
65    pub(crate) fn most_recent(&self) -> anyhow::Result<Option<Session>> {
66        let report = self.list_metadata_report()?;
67        if !report.diagnostics.is_empty() {
68            anyhow::bail!("session discovery found unreadable or unsafe history");
69        }
70        Ok(report
71            .summaries
72            .into_iter()
73            .last()
74            .map(|summary| summary.session))
75    }
76
77    pub(crate) fn path_for_valid_id(&self, id: &str) -> anyhow::Result<PathBuf> {
78        primary_path(&self.root, id)
79    }
80}
81
82pub struct Session {
83    pub(in crate::sessions) id: String,
84    pub(in crate::sessions) path: PathBuf,
85    pub(in crate::sessions) metadata_cache: Arc<Mutex<Option<SessionMetadataRecord>>>,
86    pub(crate) maintenance: crate::progressive_session_maintenance::ObserverSlot,
87    pub(crate) compaction_request: Arc<Mutex<Option<crate::compaction::CachedCompactionRequest>>>,
88    replay_state: Arc<SessionReplayState>,
89    active_lease: Arc<Mutex<Option<Arc<ActiveLeaseSlot>>>>,
90    // Shared only by clones of this admitted standalone owner, never by independent opens.
91    standalone_writer: Option<Arc<CrossProcessFileLock>>,
92}
93
94struct SessionReplayState {
95    generation: std::sync::atomic::AtomicU64,
96}
97
98type ActiveLeaseSlot = Mutex<Option<Arc<CrossProcessFileLock>>>;
99
100static ACTIVE_SESSION_LEASES: OnceLock<Mutex<HashMap<PathBuf, Weak<ActiveLeaseSlot>>>> =
101    OnceLock::new();
102static SESSION_REPLAY_GENERATIONS: OnceLock<Mutex<HashMap<PathBuf, Weak<SessionReplayState>>>> =
103    OnceLock::new();
104
105impl fmt::Debug for Session {
106    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
107        formatter
108            .debug_struct("Session")
109            .field("id", &self.id)
110            .field("path", &self.path)
111            .finish()
112    }
113}
114
115impl Clone for Session {
116    fn clone(&self) -> Self {
117        Self {
118            id: self.id.clone(),
119            path: self.path.clone(),
120            metadata_cache: Arc::clone(&self.metadata_cache),
121            maintenance: Arc::clone(&self.maintenance),
122            compaction_request: Arc::clone(&self.compaction_request),
123            replay_state: Arc::clone(&self.replay_state),
124            active_lease: Arc::clone(&self.active_lease),
125            standalone_writer: self.standalone_writer.clone(),
126        }
127    }
128}
129
130impl PartialEq for Session {
131    fn eq(&self, other: &Self) -> bool {
132        self.id == other.id && self.path == other.path
133    }
134}
135
136impl Eq for Session {}
137
138#[derive(Debug)]
139pub(crate) struct SessionWriterBusy;
140impl std::fmt::Display for SessionWriterBusy {
141    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
142        f.write_str("session is busy: another process owns the writer")
143    }
144}
145impl std::error::Error for SessionWriterBusy {}
146impl Session {
147    pub(in crate::sessions) fn new(id: String, path: PathBuf) -> Self {
148        let replay_state = replay_state_for_path(&path);
149        Self {
150            id,
151            path,
152            metadata_cache: Arc::new(Mutex::new(None)),
153            maintenance: Arc::new(Mutex::new(None)),
154            compaction_request: Arc::new(Mutex::new(None)),
155            replay_state,
156            active_lease: Arc::new(Mutex::new(None)),
157            standalone_writer: None,
158        }
159    }
160
161    pub(crate) fn admit_standalone_writer(mut self) -> anyhow::Result<Self> {
162        if self.standalone_writer.is_none() {
163            let writer = self
164                .try_frontend_writer()?
165                .ok_or_else(|| anyhow::Error::new(SessionWriterBusy))?;
166            self.standalone_writer = Some(Arc::new(writer));
167        }
168        self.activate()
169    }
170
171    pub(crate) fn activate(self) -> anyhow::Result<Self> {
172        self.ensure_active_lease()?;
173        Ok(self)
174    }
175
176    pub(in crate::sessions) fn ensure_active_lease(&self) -> anyhow::Result<()> {
177        let slot = {
178            let mut local_slot = self
179                .active_lease
180                .lock()
181                .map_err(|_| anyhow::anyhow!("active session lease slot was poisoned"))?;
182            if let Some(slot) = local_slot.as_ref() {
183                Arc::clone(slot)
184            } else {
185                let slot = active_lease_slot(&self.path)?;
186                *local_slot = Some(Arc::clone(&slot));
187                slot
188            }
189        };
190        let mut lease = slot
191            .lock()
192            .map_err(|_| anyhow::anyhow!("active session lease was poisoned"))?;
193        if lease.is_none() {
194            let target = active_lease_target(&self.path);
195            *lease = Some(Arc::new(
196                CrossProcessFileLock::try_acquire(&target)?
197                    .ok_or_else(|| anyhow::Error::new(SessionWriterBusy))?,
198            ));
199        }
200        Ok(())
201    }
202
203    pub(in crate::sessions) fn metadata_cache(&self) -> &Mutex<Option<SessionMetadataRecord>> {
204        &self.metadata_cache
205    }
206}
207
208fn replay_state_for_path(path: &Path) -> Arc<SessionReplayState> {
209    let key = normalize_active_lease_path(path);
210    let registry = SESSION_REPLAY_GENERATIONS.get_or_init(|| Mutex::new(HashMap::new()));
211    let Ok(mut states) = registry.lock() else {
212        // A poisoned optimization registry must not make durable sessions unavailable.
213        return Arc::new(SessionReplayState {
214            generation: std::sync::atomic::AtomicU64::new(0),
215        });
216    };
217    if let Some(state) = states.get(&key).and_then(Weak::upgrade) {
218        return state;
219    }
220    states.retain(|_, state| state.strong_count() > 0);
221    let state = Arc::new(SessionReplayState {
222        generation: std::sync::atomic::AtomicU64::new(0),
223    });
224    states.insert(key, Arc::downgrade(&state));
225    state
226}
227
228fn active_lease_slot(path: &Path) -> anyhow::Result<Arc<ActiveLeaseSlot>> {
229    let key = normalize_active_lease_path(path);
230    let registry = ACTIVE_SESSION_LEASES.get_or_init(|| Mutex::new(HashMap::new()));
231    let mut slots = registry
232        .lock()
233        .map_err(|_| anyhow::anyhow!("active session lease registry was poisoned"))?;
234    if let Some(slot) = slots.get(&key).and_then(Weak::upgrade) {
235        return Ok(slot);
236    }
237    slots.retain(|_, slot| slot.strong_count() > 0);
238    let slot = Arc::new(Mutex::new(None));
239    slots.insert(key, Arc::downgrade(&slot));
240    Ok(slot)
241}
242
243fn normalize_active_lease_path(path: &Path) -> PathBuf {
244    if let Ok(canonical) = path.canonicalize() {
245        return canonical;
246    }
247    if let (Some(parent), Some(file_name)) = (path.parent(), path.file_name())
248        && let Ok(parent) = parent.canonicalize()
249    {
250        return parent.join(file_name);
251    }
252    path.to_path_buf()
253}
254
255pub(in crate::sessions) fn active_lease_target(path: &Path) -> PathBuf {
256    path.with_extension("active")
257}
258
259impl Session {
260    pub fn id(&self) -> &str {
261        &self.id
262    }
263
264    pub(crate) fn path(&self) -> &Path {
265        &self.path
266    }
267
268    pub(crate) fn replay_generation(&self) -> u64 {
269        self.replay_state
270            .generation
271            .load(std::sync::atomic::Ordering::Acquire)
272    }
273
274    pub(in crate::sessions) fn mark_replay_changed(&self) {
275        self.replay_state
276            .generation
277            .fetch_add(1, std::sync::atomic::Ordering::AcqRel);
278    }
279}
magi_code/sessions/manager.rs

magi_code/sessions/
manager.rs
