lash_core/runtime/observation/
replay.rs

1use std::collections::{HashMap, VecDeque};
2use std::fmt;
3use std::sync::Mutex as StdMutex;
4use std::time::{Duration, Instant};
5
6use tokio::sync::broadcast;
7
8use crate::runtime::{LashRuntime, RuntimeSessionState};
9
10const SESSION_CURSOR_PREFIX: &str = "lashsc1:";
11const DEFAULT_LIVE_REPLAY_CAPACITY: usize = 2048;
12const DEFAULT_LIVE_REPLAY_TTL: Duration = Duration::from_secs(120);
13
14#[derive(
15    Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash, serde::Serialize, serde::Deserialize,
16)]
17#[serde(transparent)]
18pub struct SessionRevision(pub u64);
19
20impl SessionRevision {
21    pub fn new(revision: u64) -> Self {
22        Self(revision)
23    }
24
25    pub fn as_u64(self) -> u64 {
26        self.0
27    }
28
29    pub(super) fn from_runtime(runtime: &LashRuntime) -> Self {
30        Self::from_state(&runtime.export_persisted_state())
31    }
32
33    pub(super) fn from_state(state: &RuntimeSessionState) -> Self {
34        Self(state.head_revision.unwrap_or(state.turn_index as u64))
35    }
36}
37
38#[derive(Clone, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
39#[serde(transparent)]
40pub struct SessionCursor(String);
41
42impl SessionCursor {
43    pub(crate) fn new(
44        session_id: impl AsRef<str>,
45        revision: SessionRevision,
46        live_position: u64,
47    ) -> Self {
48        Self(format!(
49            "{SESSION_CURSOR_PREFIX}{}:{live_position}:{}",
50            revision.0,
51            session_id.as_ref()
52        ))
53    }
54
55    #[cfg(test)]
56    pub(super) fn from_raw_for_testing(raw: impl Into<String>) -> Self {
57        Self(raw.into())
58    }
59
60    pub fn as_str(&self) -> &str {
61        &self.0
62    }
63
64    pub(crate) fn parse_for_session(
65        &self,
66        expected_session_id: &str,
67    ) -> Result<ParsedSessionCursor, SessionCursorError> {
68        let parsed = self.parse()?;
69        if parsed.session_id != expected_session_id {
70            return Err(SessionCursorError::WrongSession {
71                expected_session_id: expected_session_id.to_string(),
72                actual_session_id: parsed.session_id,
73            });
74        }
75        Ok(parsed)
76    }
77
78    fn parse(&self) -> Result<ParsedSessionCursor, SessionCursorError> {
79        let payload = self.0.strip_prefix(SESSION_CURSOR_PREFIX).ok_or_else(|| {
80            SessionCursorError::Malformed {
81                message: "missing cursor prefix".to_string(),
82            }
83        })?;
84        let mut parts = payload.splitn(3, ':');
85        let revision = parts
86            .next()
87            .ok_or_else(|| SessionCursorError::Malformed {
88                message: "missing session revision".to_string(),
89            })?
90            .parse::<u64>()
91            .map_err(|err| SessionCursorError::Malformed {
92                message: format!("invalid session revision: {err}"),
93            })?;
94        let live_position = parts
95            .next()
96            .ok_or_else(|| SessionCursorError::Malformed {
97                message: "missing live replay position".to_string(),
98            })?
99            .parse::<u64>()
100            .map_err(|err| SessionCursorError::Malformed {
101                message: format!("invalid live replay position: {err}"),
102            })?;
103        let session_id = parts
104            .next()
105            .filter(|value| !value.is_empty())
106            .ok_or_else(|| SessionCursorError::Malformed {
107                message: "missing session id".to_string(),
108            })?
109            .to_string();
110        Ok(ParsedSessionCursor {
111            session_id,
112            revision: SessionRevision(revision),
113            live_position,
114        })
115    }
116}
117
118impl fmt::Debug for SessionCursor {
119    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
120        f.write_str("SessionCursor(<opaque>)")
121    }
122}
123
124impl fmt::Display for SessionCursor {
125    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
126        f.write_str(&self.0)
127    }
128}
129
130#[derive(Clone, Debug)]
131pub(crate) struct ParsedSessionCursor {
132    pub session_id: String,
133    pub revision: SessionRevision,
134    pub live_position: u64,
135}
136
137#[derive(Clone, Debug, thiserror::Error)]
138pub enum SessionCursorError {
139    #[error("malformed session cursor: {message}")]
140    Malformed { message: String },
141    #[error("session cursor belongs to `{actual_session_id}`, not `{expected_session_id}`")]
142    WrongSession {
143        expected_session_id: String,
144        actual_session_id: String,
145    },
146}
147
148#[derive(Clone, Debug)]
149pub struct SessionObservation {
150    pub read_view: crate::SessionReadView,
151    pub cursor: SessionCursor,
152}
153
154#[derive(Clone, Debug)]
155pub struct SessionObservationEvent {
156    pub session_id: String,
157    pub revision: SessionRevision,
158    pub cursor: SessionCursor,
159    pub payload: SessionObservationEventPayload,
160}
161
162#[derive(Clone, Copy, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
163#[serde(rename_all = "snake_case")]
164pub enum SessionQueueEventKind {
165    Enqueued,
166    Cancelled,
167}
168
169#[derive(Clone, Copy, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
170#[serde(rename_all = "snake_case")]
171pub enum SessionProcessEventKind {
172    Started,
173    Cancelled,
174}
175
176#[derive(Clone, Debug)]
177#[allow(clippy::large_enum_variant)]
178pub enum SessionObservationEventPayload {
179    TurnActivity(crate::TurnActivity),
180    Committed {
181        read_view: crate::SessionReadView,
182    },
183    AgentFrameSwitched {
184        frame_id: String,
185    },
186    QueueChanged {
187        kind: SessionQueueEventKind,
188        batch_ids: Vec<String>,
189    },
190    ProcessChanged {
191        kind: SessionProcessEventKind,
192        process_ids: Vec<String>,
193    },
194}
195
196#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
197pub struct LiveReplayGap {
198    pub session_id: String,
199    pub requested_cursor: SessionCursor,
200    pub latest_cursor: SessionCursor,
201    pub latest_revision: SessionRevision,
202    pub reason: LiveReplayGapReason,
203}
204
205#[derive(Clone, Copy, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
206#[serde(rename_all = "snake_case")]
207pub enum LiveReplayGapReason {
208    Trimmed,
209    Unavailable,
210}
211
212#[derive(Clone, Debug, thiserror::Error)]
213pub enum LiveReplayStoreError {
214    #[error("{0}")]
215    Cursor(#[from] SessionCursorError),
216    #[error("live replay store error: {0}")]
217    Store(String),
218    #[error("live replay subscriber lagged by {0} events")]
219    SubscriberLagged(u64),
220    #[error("live replay channel closed")]
221    Closed,
222}
223
224#[derive(Clone, Debug)]
225pub enum LiveReplayResult {
226    Replayed(Vec<SessionObservationEvent>),
227    Gap(LiveReplayGapReason),
228}
229
230pub enum LiveReplaySubscribeResult {
231    Subscribed(LiveReplaySubscription),
232    Gap(LiveReplayGapReason),
233}
234
235pub struct LiveReplaySubscription {
236    replay: VecDeque<SessionObservationEvent>,
237    receiver: broadcast::Receiver<SessionObservationEvent>,
238}
239
240impl LiveReplaySubscription {
241    fn new(
242        replay: Vec<SessionObservationEvent>,
243        receiver: broadcast::Receiver<SessionObservationEvent>,
244    ) -> Self {
245        Self {
246            replay: replay.into(),
247            receiver,
248        }
249    }
250
251    pub async fn next_event(&mut self) -> Result<SessionObservationEvent, LiveReplayStoreError> {
252        if let Some(event) = self.replay.pop_front() {
253            return Ok(event);
254        }
255        match self.receiver.recv().await {
256            Ok(event) => Ok(event),
257            Err(broadcast::error::RecvError::Lagged(count)) => {
258                Err(LiveReplayStoreError::SubscriberLagged(count))
259            }
260            Err(broadcast::error::RecvError::Closed) => Err(LiveReplayStoreError::Closed),
261        }
262    }
263}
264
265#[derive(Clone, Debug)]
266pub enum SessionResume {
267    Replayed {
268        events: Vec<SessionObservationEvent>,
269    },
270    Gap {
271        observation: SessionObservation,
272        gap: LiveReplayGap,
273    },
274}
275
276pub enum SessionObservationSubscription {
277    Subscribed(LiveReplaySubscription),
278    Gap {
279        observation: SessionObservation,
280        gap: LiveReplayGap,
281    },
282}
283
284/// Bounded, best-effort live replay for host reconnects.
285///
286/// Runtime turn execution calls this trait from synchronous boundary code. All
287/// methods must therefore be fast and nonblocking from the runtime's point of
288/// view. A custom external store should expose local or buffered behavior here,
289/// or offload blocking transport and durability work internally. Runtime turn
290/// execution must not wait for slow network or storage durability in this path.
291pub trait LiveReplayStore: Send + Sync {
292    /// Append one observation event and return its assigned cursor.
293    ///
294    /// This must be fast and nonblocking from the runtime's point of view.
295    fn append(
296        &self,
297        session_id: &str,
298        revision: SessionRevision,
299        payload: SessionObservationEventPayload,
300    ) -> Result<SessionObservationEvent, LiveReplayStoreError>;
301
302    /// Return buffered events after `cursor`, or report a recoverable gap.
303    ///
304    /// This must be fast and nonblocking from the runtime's point of view.
305    fn replay_after_cursor(
306        &self,
307        cursor: &SessionCursor,
308    ) -> Result<LiveReplayResult, LiveReplayStoreError>;
309
310    /// Subscribe after `cursor`, replaying buffered events before live events.
311    ///
312    /// This must be fast and nonblocking from the runtime's point of view.
313    fn subscribe_after_cursor(
314        &self,
315        cursor: &SessionCursor,
316    ) -> Result<LiveReplaySubscribeResult, LiveReplayStoreError>;
317
318    /// Return the latest cursor known locally for a session.
319    ///
320    /// This must be fast and nonblocking from the runtime's point of view.
321    fn current_cursor(&self, session_id: &str, revision: SessionRevision) -> SessionCursor;
322
323    /// Apply best-effort retention trimming for a session.
324    ///
325    /// This must be fast and nonblocking from the runtime's point of view.
326    fn trim_session(&self, session_id: &str) -> Result<(), LiveReplayStoreError>;
327}
328
329#[derive(Clone, Debug)]
330pub struct InMemoryLiveReplayStoreConfig {
331    pub max_events_per_session: usize,
332    pub max_age: Duration,
333}
334
335impl Default for InMemoryLiveReplayStoreConfig {
336    fn default() -> Self {
337        Self {
338            max_events_per_session: DEFAULT_LIVE_REPLAY_CAPACITY,
339            max_age: DEFAULT_LIVE_REPLAY_TTL,
340        }
341    }
342}
343
344#[derive(Debug)]
345pub struct InMemoryLiveReplayStore {
346    config: InMemoryLiveReplayStoreConfig,
347    sessions: StdMutex<HashMap<String, LiveReplaySessionBuffer>>,
348}
349
350impl InMemoryLiveReplayStore {
351    pub fn new(config: InMemoryLiveReplayStoreConfig) -> Self {
352        Self {
353            config,
354            sessions: StdMutex::new(HashMap::new()),
355        }
356    }
357
358    pub fn with_bounds(max_events_per_session: usize, max_age: Duration) -> Self {
359        Self::new(InMemoryLiveReplayStoreConfig {
360            max_events_per_session,
361            max_age,
362        })
363    }
364}
365
366impl Default for InMemoryLiveReplayStore {
367    fn default() -> Self {
368        Self::new(InMemoryLiveReplayStoreConfig::default())
369    }
370}
371
372#[derive(Debug)]
373struct LiveReplaySessionBuffer {
374    events: VecDeque<StoredObservationEvent>,
375    tail_position: u64,
376    sender: Option<broadcast::Sender<SessionObservationEvent>>,
377}
378
379impl LiveReplaySessionBuffer {
380    fn new() -> Self {
381        Self {
382            events: VecDeque::new(),
383            tail_position: 0,
384            sender: None,
385        }
386    }
387
388    fn subscribe(
389        &mut self,
390        channel_capacity: usize,
391    ) -> broadcast::Receiver<SessionObservationEvent> {
392        match self.sender.as_ref() {
393            Some(sender) => sender.subscribe(),
394            None => {
395                let (sender, receiver) = broadcast::channel(channel_capacity.max(1));
396                self.sender = Some(sender);
397                receiver
398            }
399        }
400    }
401
402    fn publish(&mut self, event: SessionObservationEvent) {
403        let Some(sender) = self.sender.as_ref() else {
404            return;
405        };
406        if sender.send(event).is_err() {
407            self.sender = None;
408        }
409    }
410}
411
412#[derive(Clone, Debug)]
413struct StoredObservationEvent {
414    position: u64,
415    appended_at: Instant,
416    event: SessionObservationEvent,
417}
418
419impl InMemoryLiveReplayStore {
420    fn trim_locked(config: &InMemoryLiveReplayStoreConfig, buffer: &mut LiveReplaySessionBuffer) {
421        let now = Instant::now();
422        while buffer.events.len() > config.max_events_per_session {
423            buffer.events.pop_front();
424        }
425        while buffer
426            .events
427            .front()
428            .is_some_and(|event| now.duration_since(event.appended_at) > config.max_age)
429        {
430            buffer.events.pop_front();
431        }
432    }
433
434    fn gap_reason_for_cursor(
435        buffer: Option<&LiveReplaySessionBuffer>,
436        cursor_position: u64,
437    ) -> Option<LiveReplayGapReason> {
438        let Some(buffer) = buffer else {
439            return (cursor_position > 0).then_some(LiveReplayGapReason::Unavailable);
440        };
441        if cursor_position > buffer.tail_position {
442            return Some(LiveReplayGapReason::Unavailable);
443        }
444        let Some(first) = buffer.events.front() else {
445            return (cursor_position < buffer.tail_position)
446                .then_some(LiveReplayGapReason::Trimmed);
447        };
448        if cursor_position + 1 < first.position {
449            Some(LiveReplayGapReason::Trimmed)
450        } else {
451            None
452        }
453    }
454}
455
456impl LiveReplayStore for InMemoryLiveReplayStore {
457    fn append(
458        &self,
459        session_id: &str,
460        revision: SessionRevision,
461        payload: SessionObservationEventPayload,
462    ) -> Result<SessionObservationEvent, LiveReplayStoreError> {
463        let mut sessions = self
464            .sessions
465            .lock()
466            .map_err(|_| LiveReplayStoreError::Store("live replay mutex poisoned".to_string()))?;
467        let buffer = sessions
468            .entry(session_id.to_string())
469            .or_insert_with(LiveReplaySessionBuffer::new);
470        buffer.tail_position = buffer.tail_position.saturating_add(1);
471        let cursor = SessionCursor::new(session_id, revision, buffer.tail_position);
472        let event = SessionObservationEvent {
473            session_id: session_id.to_string(),
474            revision,
475            cursor,
476            payload,
477        };
478        buffer.events.push_back(StoredObservationEvent {
479            position: buffer.tail_position,
480            appended_at: Instant::now(),
481            event: event.clone(),
482        });
483        Self::trim_locked(&self.config, buffer);
484        buffer.publish(event.clone());
485        Ok(event)
486    }
487
488    fn replay_after_cursor(
489        &self,
490        cursor: &SessionCursor,
491    ) -> Result<LiveReplayResult, LiveReplayStoreError> {
492        let parsed = cursor.parse()?;
493        let _cursor_revision = parsed.revision;
494        let mut sessions = self
495            .sessions
496            .lock()
497            .map_err(|_| LiveReplayStoreError::Store("live replay mutex poisoned".to_string()))?;
498        if let Some(buffer) = sessions.get_mut(&parsed.session_id) {
499            Self::trim_locked(&self.config, buffer);
500        }
501        let buffer = sessions.get(&parsed.session_id);
502        if let Some(reason) = Self::gap_reason_for_cursor(buffer, parsed.live_position) {
503            return Ok(LiveReplayResult::Gap(reason));
504        }
505        let events = buffer
506            .map(|buffer| {
507                buffer
508                    .events
509                    .iter()
510                    .filter(|event| event.position > parsed.live_position)
511                    .map(|event| event.event.clone())
512                    .collect()
513            })
514            .unwrap_or_default();
515        Ok(LiveReplayResult::Replayed(events))
516    }
517
518    fn subscribe_after_cursor(
519        &self,
520        cursor: &SessionCursor,
521    ) -> Result<LiveReplaySubscribeResult, LiveReplayStoreError> {
522        let parsed = cursor.parse()?;
523        let _cursor_revision = parsed.revision;
524        let mut sessions = self
525            .sessions
526            .lock()
527            .map_err(|_| LiveReplayStoreError::Store("live replay mutex poisoned".to_string()))?;
528        let buffer = sessions
529            .entry(parsed.session_id.clone())
530            .or_insert_with(LiveReplaySessionBuffer::new);
531        Self::trim_locked(&self.config, buffer);
532        if let Some(reason) = Self::gap_reason_for_cursor(Some(buffer), parsed.live_position) {
533            return Ok(LiveReplaySubscribeResult::Gap(reason));
534        }
535        let replay = buffer
536            .events
537            .iter()
538            .filter(|event| event.position > parsed.live_position)
539            .map(|event| event.event.clone())
540            .collect();
541        let receiver = buffer.subscribe(self.config.max_events_per_session);
542        Ok(LiveReplaySubscribeResult::Subscribed(
543            LiveReplaySubscription::new(replay, receiver),
544        ))
545    }
546
547    fn current_cursor(&self, session_id: &str, revision: SessionRevision) -> SessionCursor {
548        let tail_position = self
549            .sessions
550            .lock()
551            .ok()
552            .and_then(|sessions| sessions.get(session_id).map(|buffer| buffer.tail_position))
553            .unwrap_or(0);
554        SessionCursor::new(session_id, revision, tail_position)
555    }
556
557    fn trim_session(&self, session_id: &str) -> Result<(), LiveReplayStoreError> {
558        let mut sessions = self
559            .sessions
560            .lock()
561            .map_err(|_| LiveReplayStoreError::Store("live replay mutex poisoned".to_string()))?;
562        if let Some(buffer) = sessions.get_mut(session_id) {
563            Self::trim_locked(&self.config, buffer);
564        }
565        Ok(())
566    }
567}
568
569#[cfg(test)]
570mod tests {
571    use super::*;
572
573    fn activity(text: &str) -> SessionObservationEventPayload {
574        SessionObservationEventPayload::TurnActivity(crate::TurnActivity::independent(
575            crate::TurnEvent::AssistantProseDelta {
576                text: text.to_string(),
577            },
578        ))
579    }
580
581    #[test]
582    fn session_cursor_round_trips_and_debug_is_opaque() {
583        let cursor = SessionCursor::new("session:with:colon", SessionRevision(3), 9);
584        let encoded = serde_json::to_string(&cursor).expect("serialize");
585        let decoded: SessionCursor = serde_json::from_str(&encoded).expect("deserialize");
586        assert_eq!(decoded, cursor);
587        assert_eq!(format!("{cursor:?}"), "SessionCursor(<opaque>)");
588        let parsed = cursor
589            .parse_for_session("session:with:colon")
590            .expect("parse");
591        assert_eq!(parsed.revision, SessionRevision(3));
592        assert_eq!(parsed.live_position, 9);
593    }
594
595    #[test]
596    fn session_cursor_rejects_malformed_and_wrong_session() {
597        let malformed = SessionCursor::from_raw_for_testing("bad");
598        assert!(matches!(
599            malformed.parse_for_session("s"),
600            Err(SessionCursorError::Malformed { .. })
601        ));
602        let cursor = SessionCursor::new("actual", SessionRevision(0), 0);
603        assert!(matches!(
604            cursor.parse_for_session("expected"),
605            Err(SessionCursorError::WrongSession { .. })
606        ));
607    }
608
609    #[test]
610    fn in_memory_replay_store_replays_after_cursor_in_order() {
611        let store = InMemoryLiveReplayStore::default();
612        let start = store.current_cursor("s", SessionRevision(0));
613        store
614            .append("s", SessionRevision(0), activity("a"))
615            .expect("append a");
616        store
617            .append("s", SessionRevision(0), activity("b"))
618            .expect("append b");
619        let LiveReplayResult::Replayed(events) = store.replay_after_cursor(&start).expect("replay")
620        else {
621            panic!("expected replay");
622        };
623        assert_eq!(events.len(), 2);
624        match &events[0].payload {
625            SessionObservationEventPayload::TurnActivity(activity) => match &activity.event {
626                crate::TurnEvent::AssistantProseDelta { text } => assert_eq!(text, "a"),
627                _ => panic!("wrong event"),
628            },
629            _ => panic!("wrong payload"),
630        }
631    }
632
633    #[test]
634    fn in_memory_replay_store_reports_gap_after_capacity_trim() {
635        let store = InMemoryLiveReplayStore::with_bounds(1, Duration::from_secs(120));
636        let start = store.current_cursor("s", SessionRevision(0));
637        store
638            .append("s", SessionRevision(0), activity("a"))
639            .expect("append a");
640        store
641            .append("s", SessionRevision(0), activity("b"))
642            .expect("append b");
643        assert!(matches!(
644            store.replay_after_cursor(&start).expect("gap"),
645            LiveReplayResult::Gap(LiveReplayGapReason::Trimmed)
646        ));
647    }
648
649    #[test]
650    fn in_memory_replay_store_reports_gap_after_ttl_trim() {
651        let store = InMemoryLiveReplayStore::with_bounds(16, Duration::from_millis(1));
652        let start = store.current_cursor("s", SessionRevision(0));
653        store
654            .append("s", SessionRevision(0), activity("a"))
655            .expect("append a");
656        std::thread::sleep(Duration::from_millis(5));
657        assert!(matches!(
658            store.replay_after_cursor(&start).expect("gap"),
659            LiveReplayResult::Gap(LiveReplayGapReason::Trimmed)
660        ));
661    }
662
663    #[test]
664    fn in_memory_replay_store_reports_unavailable_for_cursor_ahead_of_tail() {
665        let store = InMemoryLiveReplayStore::default();
666        let ahead = SessionCursor::new("s", SessionRevision(0), 99);
667        assert!(matches!(
668            store.replay_after_cursor(&ahead).expect("gap"),
669            LiveReplayResult::Gap(LiveReplayGapReason::Unavailable)
670        ));
671    }
672
673    #[tokio::test]
674    async fn in_memory_replay_subscription_yields_replay_then_live() {
675        let store = InMemoryLiveReplayStore::default();
676        let start = store.current_cursor("s", SessionRevision(0));
677        store
678            .append("s", SessionRevision(0), activity("a"))
679            .expect("append a");
680        let LiveReplaySubscribeResult::Subscribed(mut subscription) =
681            store.subscribe_after_cursor(&start).expect("subscribe")
682        else {
683            panic!("expected subscription");
684        };
685        let first = subscription.next_event().await.expect("replay");
686        assert_eq!(first.session_id, "s");
687        store
688            .append("s", SessionRevision(0), activity("b"))
689            .expect("append b");
690        let second = subscription.next_event().await.expect("live");
691        match second.payload {
692            SessionObservationEventPayload::TurnActivity(activity) => match activity.event {
693                crate::TurnEvent::AssistantProseDelta { text } => assert_eq!(text, "b"),
694                _ => panic!("wrong event"),
695            },
696            _ => panic!("wrong payload"),
697        }
698    }
699
700    #[test]
701    fn in_memory_replay_store_allocates_live_channel_lazily() {
702        let store = InMemoryLiveReplayStore::default();
703        let start = store.current_cursor("s", SessionRevision(0));
704        store
705            .append("s", SessionRevision(0), activity("a"))
706            .expect("append a");
707        {
708            let sessions = store.sessions.lock().expect("sessions");
709            assert!(sessions.get("s").expect("buffer").sender.is_none());
710        }
711        let LiveReplaySubscribeResult::Subscribed(subscription) =
712            store.subscribe_after_cursor(&start).expect("subscribe")
713        else {
714            panic!("expected subscription");
715        };
716        {
717            let sessions = store.sessions.lock().expect("sessions");
718            assert!(sessions.get("s").expect("buffer").sender.is_some());
719        }
720        drop(subscription);
721        store
722            .append("s", SessionRevision(0), activity("b"))
723            .expect("append b");
724        let sessions = store.sessions.lock().expect("sessions");
725        assert!(sessions.get("s").expect("buffer").sender.is_none());
726    }
727
728    #[test]
729    fn in_memory_replay_subscription_reports_gap_after_capacity_trim() {
730        let store = InMemoryLiveReplayStore::with_bounds(1, Duration::from_secs(120));
731        let start = store.current_cursor("s", SessionRevision(0));
732        store
733            .append("s", SessionRevision(0), activity("a"))
734            .expect("append a");
735        store
736            .append("s", SessionRevision(0), activity("b"))
737            .expect("append b");
738        assert!(matches!(
739            store.subscribe_after_cursor(&start).expect("subscribe"),
740            LiveReplaySubscribeResult::Gap(LiveReplayGapReason::Trimmed)
741        ));
742    }
743
744    #[test]
745    fn in_memory_replay_subscription_reports_gap_after_ttl_trim() {
746        let store = InMemoryLiveReplayStore::with_bounds(16, Duration::from_millis(1));
747        let start = store.current_cursor("s", SessionRevision(0));
748        store
749            .append("s", SessionRevision(0), activity("a"))
750            .expect("append a");
751        std::thread::sleep(Duration::from_millis(5));
752        assert!(matches!(
753            store.subscribe_after_cursor(&start).expect("subscribe"),
754            LiveReplaySubscribeResult::Gap(LiveReplayGapReason::Trimmed)
755        ));
756    }
757}
lash_core/runtime/observation/replay.rs

lash_core/runtime/observation/
replay.rs
