diff --git a/crates/runtime/src/app/events.rs b/crates/runtime/src/app/events.rs index 50e0ccc12..9ac6addb5 100644 --- a/crates/runtime/src/app/events.rs +++ b/crates/runtime/src/app/events.rs @@ -512,19 +512,34 @@ impl AppState { event: event.clone(), elided: None, }; - let end = self.resident_log(session_id).end(); - self.emit_domain( - Topic::SessionEvents { - session_id: session_id.to_string(), - }, - ServerEvent::SessionEvent { - epoch: end.epoch, - position: end.position as u64, - record: history::wire_record(&record).into_owned(), - }, - cx, - ); - self.resident_log(session_id).push(record); + let topic = Topic::SessionEvents { + session_id: session_id.to_string(), + }; + // A session nobody holds needs no position: the append alone keeps + // the record, and any later read of the log is queued behind it. + if let Some(end) = self.event_records.get(session_id).map(SessionLog::end) { + self.emit_domain( + topic, + ServerEvent::SessionEvent { + epoch: end.epoch, + position: end.position as u64, + record: history::wire_record(&record).into_owned(), + }, + cx, + ); + self.event_records + .get_mut(session_id) + .expect("cached above") + .push(record); + } else if self.log_hydrations.contains_key(session_id) + || self.resident(session_id).is_some() + || self.subscriptions.contains(&topic) + { + // Its position comes after whatever the log holds, so it is + // sent once the log is read; the read is queued before the + // append below, which it therefore does not see. + self.hydrate_log(session_id, None, cx).pending.push(record); + } self.enqueue_store_write( StoreWrite::AppendEvent { id: session_id.to_string(), @@ -629,11 +644,13 @@ impl AppState { let previous_title = meta.title; let title_meta = title_session_meta(&self.settings, meta.cwd); let regenerate = first_message.is_none(); - // The cache includes accepted messages whose disk writes are still queued. + // The cache includes accepted messages whose disk writes are still + // queued; without it the JSONL is read once they are written. let records = regenerate .then(|| self.event_records.get(&session_id)) .flatten() .map(|log| log.records().to_vec()); + let written = (regenerate && records.is_none()).then(|| self.store_write_barrier(cx)); let store = self.store.clone(); let settings = self.settings.clone(); let settings_store = self.settings_store.clone(); @@ -642,6 +659,9 @@ impl AppState { let host_cx = cx.clone(); HostCx::spawn_detached(cx, async move { + if let Some(written) = written { + let _ = written.recv().await; + } let read_id = session_id.clone(); let old_title = previous_title.clone(); let input = host_cx diff --git a/crates/runtime/src/app/history.rs b/crates/runtime/src/app/history.rs index 2ddadcf4a..eb714269e 100644 --- a/crates/runtime/src/app/history.rs +++ b/crates/runtime/src/app/history.rs @@ -3,8 +3,8 @@ use agent::DeltaKind; use std::borrow::Cow; use std::ops::Range; use tcode_protocol::{ - HostMessage, MAX_SESSION_HISTORY_BYTES, OUTPUT_PREVIEW_BYTES, SESSION_HISTORY_RECORDS, - SESSION_WINDOW_BYTES, SessionWindow, + CommandResponse, HostMessage, MAX_SESSION_HISTORY_BYTES, OUTPUT_PREVIEW_BYTES, + SESSION_HISTORY_RECORDS, SESSION_WINDOW_BYTES, SessionWindow, Subscription, }; /// Count serialized bytes without allocating a copy of a large record. @@ -221,11 +221,11 @@ pub(super) struct LogCursor { /// /// Memory policy: [`AppState::event_records`] holds a log for every live or /// parked session (bounded by the resident LRU) and for nothing else. A log -/// is loaded when a client opens the session or when the session first -/// appends in this process, and is dropped once the session leaves residency -/// and the store writer has flushed every append queued from it -/// ([`AppState::release_stale_session_logs`]): until then the log, not the -/// JSONL, is the whole conversation. +/// is read off the mailbox ([`Hydration`]) when a client opens the session +/// or when the session first appends in this process, and is dropped once +/// the session leaves residency and the store writer has flushed every +/// append queued from it ([`AppState::release_stale_session_logs`]): until +/// then the log, not the JSONL, is the whole conversation. /// /// A resident log keeps the layout it was loaded in. Compaction rewrites the /// JSONL beneath it without touching it, so its positions, its fold and every @@ -246,11 +246,6 @@ pub(super) struct SessionLog { } impl SessionLog { - pub(super) fn load(store: &SessionStore, session_id: &str) -> Self { - let log = store.read_log(session_id); - Self::new(log.epoch, log.records) - } - pub(super) fn new(epoch: u64, records: impl IntoIterator) -> Self { let mut log = Self { epoch, @@ -307,97 +302,13 @@ impl SessionLog { .checked_sub(1) .map_or(start, |last| self.turn_starts[last]) } -} - -impl AppState { - /// The session's cached log, loaded from the JSONL when nothing is cached - /// yet. From then until it is released it is the whole conversation, in - /// the layout it was loaded in. - pub(super) fn resident_log(&mut self, session_id: &str) -> &mut SessionLog { - if !self.event_records.contains_key(session_id) { - let log = SessionLog::load(&self.store, session_id); - self.cache_log(session_id.to_string(), log); - } - self.event_records - .get_mut(session_id) - .expect("cached above") - } - - /// Cache a log read from the JSONL. A rewrite of a cached log would - /// only renumber it for the next load, so the background pass leaves it - /// to the live triggers. - pub(super) fn cache_log(&mut self, session_id: String, log: SessionLog) { - self.withdraw_pass_compaction(&session_id); - self.event_records.insert(session_id, log); - } - - /// The session's log, cached for the resident session it belongs to. A - /// non-resident session is read cold and not retained, so paging it never - /// grows the cache. - fn history_log(&mut self, session_id: &str) -> std::borrow::Cow<'_, SessionLog> { - if !self.event_records.contains_key(session_id) && self.resident(session_id).is_none() { - return std::borrow::Cow::Owned(SessionLog::load(&self.store, session_id)); - } - std::borrow::Cow::Borrowed(self.resident_log(session_id)) - } - - /// Queue a store-writer barrier for every cached log whose session left - /// residency, and drop the log once the barrier echoes: a cold read after - /// that sees every append the log had accepted. Appends that race the - /// barrier re-arm it. The log is compacted first, so a turn that never - /// completed is compacted too, and the next load serves that layout. - pub(super) fn release_stale_session_logs(&mut self, cx: &mut HostCx) { - let stale: Vec<(String, LogCursor)> = self - .event_records - .iter() - .filter(|(id, log)| log.release_barrier.is_none() && self.resident(id).is_none()) - .map(|(id, log)| (id.clone(), log.end())) - .collect(); - for (session_id, flushed) in stale { - self.event_records - .get_mut(&session_id) - .expect("stale log collected above") - .release_barrier = Some(flushed); - self.schedule_compaction(&session_id, cx); - let (completion, completed) = smol::channel::bounded(1); - self.enqueue_store_write(StoreWrite::Flush(completion), cx); - let host_cx = cx.clone(); - HostCx::spawn_detached(cx, async move { - // A stopped writer has dropped its queue as well; nothing - // later can still land in the JSONL. - let _ = completed.recv().await; - host_cx.enqueue(move |state, cx| { - let Some(log) = state.event_records.get_mut(&session_id) else { - return; - }; - log.release_barrier = None; - let appended_since = log.end() != flushed; - if state.resident(&session_id).is_some() { - return; - } - if appended_since { - state.release_stale_session_logs(cx); - } else { - state.event_records.remove(&session_id); - } - }); - }); - } - } /// The window answering a subscription: the records from its cursor on - /// when the cursor is in the layout this host serves, otherwise a - /// baseline that replaces whatever the client holds. - pub(crate) fn session_events_snapshot( - &mut self, - subscription: &tcode_protocol::Subscription, - ) -> ServerEvent { - let Topic::SessionEvents { session_id } = &subscription.topic else { - unreachable!() - }; - let log = self.history_log(session_id); + /// when the cursor is in this log's layout, otherwise a baseline that + /// replaces whatever the client holds. + pub(super) fn events_window(&self, subscription: &Subscription) -> ServerEvent { let after = subscription.after.filter(|after| { - subscription.epoch == Some(log.epoch) && *after <= log.records.len() as u64 + subscription.epoch == Some(self.epoch) && *after <= self.records.len() as u64 }); let empty = HostMessage::Event(EventEnvelope { request_id: Some(u64::MAX), @@ -405,7 +316,7 @@ impl AppState { event: ServerEvent::SessionSnapshot(empty_window()), }); match session_window( - &log, + self, after.map(|after| after as usize), wire_overhead(&empty), ) { @@ -414,26 +325,24 @@ impl AppState { } } - pub(crate) fn session_history_page( - &mut self, - session_id: &str, + pub(super) fn history_page( + &self, epoch: u64, before: u64, limit: u32, ) -> Result { - let log = self.history_log(session_id); - if epoch != log.epoch { + if epoch != self.epoch { let empty = HostMessage::QueryResult { id: u64::MAX, result: Ok(QueryResponse::SessionHistoryReset(empty_window())), }; - return session_window(&log, None, wire_overhead(&empty)) + return session_window(self, None, wire_overhead(&empty)) .map(QueryResponse::SessionHistoryReset); } - let records = log.records(); + let records = self.records(); let end = before.min(records.len() as u64) as usize; let count = (limit as usize).clamp(1, SESSION_HISTORY_RECORDS); - let requested = log.turn_aligned_start(end.saturating_sub(count))..end; + let requested = self.turn_aligned_start(end.saturating_sub(count))..end; let empty = HostMessage::QueryResult { id: u64::MAX, result: Ok(QueryResponse::SessionHistoryPage { @@ -446,7 +355,7 @@ impl AppState { }; let window = wire_window(records, requested.clone(), true, wire_overhead(&empty))?; Ok(QueryResponse::SessionHistoryPage { - epoch: log.epoch, + epoch: self.epoch, from: window.range.start as u64, end: window.range.end as u64, truncated: window.range.len() < requested.len(), @@ -455,13 +364,12 @@ impl AppState { } /// The whole output of one item, as the full log folds it. - pub(crate) fn item_output( - &mut self, + pub(super) fn item_output( + &self, session_id: &str, item_id: &str, ) -> Result { - let log = self.history_log(session_id); - let output = log + let output = self .fold() .entries .iter() @@ -488,6 +396,306 @@ impl AppState { } } +/// A session's log being read and folded off the mailbox, and what waits for +/// it. A session has at most one, and none while its log is resident. +/// +/// The read is of a [`LogSnapshot`](tcode_services::store::LogSnapshot) the +/// store writer takes in its queue order, so it holds every record whose +/// append was queued before the hydration began and none queued after. Every record accepted for the +/// session from then on is held in `pending`, in order, and takes the +/// positions after the snapshot's end; it reaches subscribers once the log is +/// resident, after the replies that end where the snapshot ends. +pub(super) struct Hydration { + pub(super) pending: Vec, + /// Subscription replies; one without a request id answers no request. + replies: Vec<(Option, Subscription)>, + /// Queries answered from the whole log. + queries: Vec, + /// The timeline load waiting for the log, with how many `pending` records + /// had been accepted when it was requested. + timeline: Option<(TimelineLoad, usize)>, +} + +type LogQuery = Box; + +impl AppState { + /// The session's hydration, begun unless its log is being read already. + /// Callers check first that the log is not resident. + pub(super) fn hydrate_log( + &mut self, + session_id: &str, + timeline: Option, + cx: &mut HostCx, + ) -> &mut Hydration { + if !self.log_hydrations.contains_key(session_id) { + let (snapshot, taken) = smol::channel::bounded(1); + self.enqueue_store_write( + StoreWrite::SnapshotLog { + id: session_id.to_string(), + snapshot, + }, + cx, + ); + let read_id = session_id.to_string(); + let host_cx = cx.clone(); + HostCx::spawn_detached(cx, async move { + // The store writer stops only with the host. + let Ok(snapshot) = taken.recv().await else { + return; + }; + let log = host_cx + .unblock(move || { + let read = snapshot.read(); + SessionLog::new(read.epoch, read.records) + }) + .await; + host_cx.enqueue(move |state, cx| state.finish_hydration(read_id, log, cx)); + }); + self.log_hydrations.insert( + session_id.to_string(), + Hydration { + pending: Vec::new(), + replies: Vec::new(), + queries: Vec::new(), + timeline: None, + }, + ); + } + let hydration = self + .log_hydrations + .get_mut(session_id) + .expect("inserted above"); + if let Some(load) = timeline { + hydration.timeline = Some((load, hydration.pending.len())); + } + hydration + } + + /// Serve everything that waited for the log, in the order of its + /// records, and keep the log only if its session is resident. + fn finish_hydration(&mut self, session_id: String, mut log: SessionLog, cx: &mut HostCx) { + let hydration = self + .log_hydrations + .remove(&session_id) + .expect("a hydration is removed only when it finishes"); + let retained = self.resident(&session_id).is_some(); + for (request_id, subscription) in &hydration.replies { + self.reply_from_log(&log, *request_id, subscription, cx); + } + let topic = Topic::SessionEvents { + session_id: session_id.clone(), + }; + let mut pending = hydration.pending.into_iter(); + let accepted_before = hydration.timeline.map_or(0, |(_, accepted)| accepted); + for record in pending.by_ref().take(accepted_before) { + self.accept_pending(&mut log, record, &topic, cx); + } + // A timeline load folds what had been accepted when it was requested; + // what came after follows any `mark_idle`, as it arrived. + let folded = hydration + .timeline + .filter(|_| retained) + .map(|(load, _)| (load, log.end(), log.fold().clone())); + for record in pending { + self.accept_pending(&mut log, record, &topic, cx); + } + for query in hydration.queries { + query(&log); + } + if !retained { + return; + } + self.withdraw_pass_compaction(&session_id); + self.event_records.insert(session_id.clone(), log); + if let Some((load, cursor, fold)) = folded { + self.fold_timeline(session_id, load, cursor, fold, cx); + } + } + + /// Append a record accepted while the log was being read, sending it as + /// `record_event` would have. + fn accept_pending( + &self, + log: &mut SessionLog, + record: SessionEventRecord, + topic: &Topic, + cx: &mut HostCx, + ) { + let end = log.end(); + self.emit_domain( + topic.clone(), + ServerEvent::SessionEvent { + epoch: end.epoch, + position: end.position as u64, + record: wire_record(&record).into_owned(), + }, + cx, + ); + log.push(record); + } + + /// Answer a subscription with a window of `log` unless nobody is + /// subscribed any more, and acknowledge the request. + fn reply_from_log( + &self, + log: &SessionLog, + request_id: Option, + subscription: &Subscription, + cx: &mut HostCx, + ) { + if self.subscriptions.contains(&subscription.topic) { + cx.emit(HostEvent::Domain(EventEnvelope { + request_id, + topic: subscription.topic.clone(), + event: log.events_window(subscription), + })); + } + if let Some(id) = request_id { + cx.send_message(HostMessage::Ack { + id, + result: Ok(CommandResponse::Unit), + }); + } + } + + /// Answer a subscription with its snapshot and acknowledge the request. + /// A session's events are answered from its log: now when it is + /// resident, otherwise once it has been read, and the acknowledgement + /// waits with it, so the window is the request's first reply. + pub(crate) fn reply_to_subscription( + &mut self, + request_id: Option, + subscription: Subscription, + cx: &mut HostCx, + ) { + if let Topic::SessionEvents { session_id } = &subscription.topic { + match self.event_records.get(session_id) { + Some(log) => self.reply_from_log(log, request_id, &subscription, cx), + None => { + let session_id = session_id.clone(); + self.hydrate_log(&session_id, None, cx) + .replies + .push((request_id, subscription)); + } + } + return; + } + if let Some(mut snapshot) = self.subscription_snapshot(&subscription) { + snapshot.request_id = request_id; + cx.emit(HostEvent::Domain(snapshot)); + } + if let Some(id) = request_id { + cx.send_message(HostMessage::Ack { + id, + result: Ok(CommandResponse::Unit), + }); + } + } + + /// Answer a query from the session's log: now when it is resident, + /// otherwise once it has been read. + fn query_session_log( + &mut self, + session_id: &str, + query: impl FnOnce(&SessionLog) -> Result + + Send + + 'static, + cx: &mut HostCx, + ) -> HostTask> { + if let Some(log) = self.event_records.get(session_id) { + let result = query(log); + return cx.spawn_background(async move { result }); + } + let (answer, answered) = smol::channel::bounded(1); + self.hydrate_log(session_id, None, cx) + .queries + .push(Box::new(move |log| { + let _ = answer.try_send(query(log)); + })); + cx.spawn_background(async move { + answered.recv().await.unwrap_or_else(|_| { + Err(tcode_protocol::ProtocolError { + code: "host_stopped".into(), + message: "the host stopped before the session's log was read".into(), + }) + }) + }) + } + + pub(crate) fn session_history_page( + &mut self, + session_id: &str, + epoch: u64, + before: u64, + limit: u32, + cx: &mut HostCx, + ) -> HostTask> { + self.query_session_log( + session_id, + move |log| log.history_page(epoch, before, limit), + cx, + ) + } + + pub(crate) fn item_output( + &mut self, + session_id: &str, + item_id: String, + cx: &mut HostCx, + ) -> HostTask> { + let read_id = session_id.to_string(); + self.query_session_log( + session_id, + move |log| log.item_output(&read_id, &item_id), + cx, + ) + } + + /// Queue a store-writer barrier for every cached log whose session left + /// residency, and drop the log once the barrier echoes: a cold read after + /// that sees every append the log had accepted. Appends that race the + /// barrier re-arm it. The log is compacted first, so a turn that never + /// completed is compacted too, and the next load serves that layout. + pub(super) fn release_stale_session_logs(&mut self, cx: &mut HostCx) { + let stale: Vec<(String, LogCursor)> = self + .event_records + .iter() + .filter(|(id, log)| log.release_barrier.is_none() && self.resident(id).is_none()) + .map(|(id, log)| (id.clone(), log.end())) + .collect(); + for (session_id, flushed) in stale { + self.event_records + .get_mut(&session_id) + .expect("stale log collected above") + .release_barrier = Some(flushed); + self.schedule_compaction(&session_id, cx); + let (completion, completed) = smol::channel::bounded(1); + self.enqueue_store_write(StoreWrite::Flush(completion), cx); + let host_cx = cx.clone(); + HostCx::spawn_detached(cx, async move { + // A stopped writer has dropped its queue as well; nothing + // later can still land in the JSONL. + let _ = completed.recv().await; + host_cx.enqueue(move |state, cx| { + let Some(log) = state.event_records.get_mut(&session_id) else { + return; + }; + log.release_barrier = None; + let appended_since = log.end() != flushed; + if state.resident(&session_id).is_some() { + return; + } + if appended_since { + state.release_stale_session_logs(cx); + } else { + state.event_records.remove(&session_id); + } + }); + }); + } + } +} + /// The bytes a reply spends besides its records, measured on `empty`, a reply /// without records whose numbers are as long as any reply's. fn wire_overhead(empty: &HostMessage) -> usize { diff --git a/crates/runtime/src/app/mod.rs b/crates/runtime/src/app/mod.rs index 154b729e6..0426ae288 100644 --- a/crates/runtime/src/app/mod.rs +++ b/crates/runtime/src/app/mod.rs @@ -204,6 +204,22 @@ enum TimelineLoadTarget { Background, } +/// One request to rebuild a resident session's timeline from its log. +#[derive(Debug, Clone, Copy)] +struct TimelineLoad { + generation: u64, + target: TimelineLoadTarget, +} + +impl TimelineLoad { + fn mark_idle(&self) -> bool { + match self.target { + TimelineLoadTarget::Active { mark_idle } => mark_idle, + TimelineLoadTarget::Background => true, + } + } +} + #[derive(Debug, Clone, Copy, Serialize, Deserialize)] struct TerminalPreferences { open: bool, @@ -232,7 +248,7 @@ mod compaction; mod events; mod git; mod history; -use history::SessionLog; +use history::{Hydration, SessionLog}; mod lifecycle; mod options; mod orchestrate; @@ -395,6 +411,9 @@ pub struct AppState { /// Resident sessions' event logs; see [`SessionLog`] for what is cached /// and when it is dropped. event_records: HashMap, + /// Logs being read off the mailbox, by session; never one that is in + /// `event_records`. + log_hydrations: HashMap, compactions: compaction::Compactions, /// Composer-draft review notes, keyed by session id (in-memory only). review_comment_drafts: HashMap>, @@ -559,6 +578,7 @@ impl AppState { timeline_load_generations: HashMap::new(), subscriptions: HashSet::new(), event_records: HashMap::new(), + log_hydrations: HashMap::new(), compactions: Default::default(), review_comment_drafts: HashMap::new(), pending_relaunch, diff --git a/crates/runtime/src/app/sessions.rs b/crates/runtime/src/app/sessions.rs index 729231973..d415679f0 100644 --- a/crates/runtime/src/app/sessions.rs +++ b/crates/runtime/src/app/sessions.rs @@ -998,17 +998,17 @@ impl AppState { ); state.upsert_sessions_in_memory([fork.clone()]); state.select_session(&fork.id, cx); - if let Some(snapshot) = - state.subscription_snapshot(&tcode_protocol::Subscription { + state.reply_to_subscription( + None, + tcode_protocol::Subscription { topic: Topic::SessionEvents { session_id: fork.id.clone(), }, after: None, epoch: None, - }) - { - cx.emit(HostEvent::Domain(snapshot)); - } + }, + cx, + ); } Err(error) => { state.report_error(RuntimeError::PersistEvent { error }, cx); @@ -1533,45 +1533,51 @@ impl AppState { let Some(cwd) = intended.map(|session| session.meta.cwd.clone()) else { return; }; + if matches!(target, TimelineLoadTarget::Active { .. }) { + self.refresh_session_git_branch(session_id.clone(), cwd, cx); + } + let load = TimelineLoad { generation, target }; // A cached log is the whole conversation, including appends whose // disk writes are still queued, so the timeline derives from it and - // never from a JSONL that can lag the store writer. A session opened - // for a client loads its log here, on the mailbox, because the - // snapshot that answers the subscription needs it in the same turn; - // a background session parses off the mailbox and caches on completion. - let cached = match target { - TimelineLoadTarget::Active { .. } => Some(&*self.resident_log(&session_id)), - TimelineLoadTarget::Background => self.event_records.get(&session_id), + // never from a JSONL that can lag the store writer. + match self.event_records.get(&session_id) { + Some(log) => { + let cursor = log.end(); + let fold = log.fold().clone(); + self.fold_timeline(session_id, load, cursor, fold, cx); + } + None => { + self.hydrate_log(&session_id, Some(load), cx); + } } - .map(|log| (log.end(), log.fold().clone())); - let (mark_idle, load_branch) = match target { - TimelineLoadTarget::Active { mark_idle } => (mark_idle, true), - TimelineLoadTarget::Background => (true, false), - }; - let store = self.store.clone(); + } + + /// Mark `fold`, the cached log's fold up to `cursor`, idle as `load` + /// asks, then make it the session's unless a later load superseded `load` + /// or the session left the residency it was loaded for. + pub(super) fn fold_timeline( + &mut self, + session_id: String, + load: TimelineLoad, + cursor: history::LogCursor, + mut fold: Timeline, + cx: &mut HostCx, + ) { + let mark_idle = load.mark_idle(); let host_cx = cx.clone(); HostCx::spawn_detached(cx, async move { - let read_id = session_id.clone(); - let (timeline, folded, loaded, git_branch) = host_cx + let timeline = host_cx .unblock(move || { - let (mut timeline, folded, loaded) = match cached { - Some((folded, fold)) => (fold, folded, None), - None => { - let log = SessionLog::load(&store, &read_id); - (log.fold().clone(), log.end(), Some(log)) - } - }; if mark_idle { - timeline.mark_idle(); + fold.mark_idle(); } - let git_branch = load_branch.then(|| read_git_branch(&cwd)); - (timeline, folded, loaded, git_branch) + fold }) .await; host_cx.enqueue(move |state, cx| { - let generation_matches = - state.timeline_load_generations.get(&session_id).copied() == Some(generation); - let target_matches = match target { + let generation_matches = state.timeline_load_generations.get(&session_id).copied() + == Some(load.generation); + let target_matches = match load.target { TimelineLoadTarget::Active { .. } => { state.residents.live.contains_key(&session_id) } @@ -1582,36 +1588,17 @@ impl AppState { if !generation_matches || !target_matches { return; } - // An append during the parse already loaded and extended its - // own copy; that one carries the newer records. - if let Some(loaded) = loaded - && !state.event_records.contains_key(&session_id) - { - state.cache_log(session_id.clone(), loaded); - } let mut timeline = timeline; + // A resident session's log stays cached, in the layout it was + // read in; records it accepted after `cursor` continue it. if let Some(log) = state.event_records.get(&session_id) { - if log.epoch() == folded.epoch { - // Records appended while the fold ran continue the - // same layout. - for record in log.records().iter().skip(folded.position) { - timeline.apply_at(record.ts, &record.event); - } - } else { - // The cached copy was loaded after a rewrite renumbered - // the log, so `folded` says nothing about which of its - // records the fold covered; its own fold covers them all. - timeline = log.fold().clone(); - if mark_idle { - timeline.mark_idle(); - } + debug_assert_eq!(log.epoch(), cursor.epoch); + for record in log.records().iter().skip(cursor.position) { + timeline.apply_at(record.ts, &record.event); } } if let Some(session) = state.resident_mut(&session_id) { session.timeline = timeline; - if let Some(git_branch) = git_branch { - session.git_branch = git_branch; - } } state.repair_orphaned_mirror_turn(&session_id, cx); }); diff --git a/crates/runtime/src/app/snapshots.rs b/crates/runtime/src/app/snapshots.rs index f8b419621..9222b55c2 100644 --- a/crates/runtime/src/app/snapshots.rs +++ b/crates/runtime/src/app/snapshots.rs @@ -280,7 +280,8 @@ impl AppState { Topic::SessionStatus { session_id } => { ServerEvent::SessionStatusReplaced(self.session_status_snapshot(session_id)?) } - Topic::SessionEvents { .. } => self.session_events_snapshot(subscription), + // Answered from the session's log by `reply_to_subscription`. + Topic::SessionEvents { .. } => return None, Topic::RuntimeEvents => return None, Topic::Preview { .. } => return None, // Retained latest-run status, so a client that subscribes after a diff --git a/crates/runtime/src/app/store_write.rs b/crates/runtime/src/app/store_write.rs index 5b2969622..3004d407d 100644 --- a/crates/runtime/src/app/store_write.rs +++ b/crates/runtime/src/app/store_write.rs @@ -2,7 +2,7 @@ use std::sync::atomic::Ordering; use super::compaction::{Compacted, CompactionRequest}; use super::*; -use tcode_services::store::{CompactOutcome, CompactRejection, IndexWrite}; +use tcode_services::store::{CompactOutcome, CompactRejection, IndexWrite, LogSnapshot}; pub(super) enum StoreWrite { AppendEvent { @@ -48,6 +48,11 @@ pub(super) enum StoreWrite { /// Delete the originals and temporary files earlier runs' compactions /// left beside the logs. DiscardCompactionLeftovers, + /// Open one log as every write queued before this one left it. + SnapshotLog { + id: String, + snapshot: smol::channel::Sender, + }, Flush(smol::channel::Sender<()>), } @@ -376,6 +381,10 @@ fn run_store_write( error: err.to_string(), }) }), + StoreWrite::SnapshotLog { id, snapshot } => { + let _ = snapshot.try_send(store.snapshot_log(&id)); + None + } StoreWrite::Flush(completion) => { let _ = completion.try_send(()); None diff --git a/crates/runtime/src/app/tests.rs b/crates/runtime/src/app/tests.rs index 3e7ae2c50..6bdbcb14e 100644 --- a/crates/runtime/src/app/tests.rs +++ b/crates/runtime/src/app/tests.rs @@ -4955,6 +4955,7 @@ fn model_fallback_stops_active_session_when_abort_on_model_fallback_is_enabled() }); assert!(matches!(receiver.try_recv(), Ok(SessionCommand::Shutdown))); + cx.run_until_parked(); let outgoing = cx.drain_outgoing(); assert!(outgoing.iter().any(|message| matches!( message, @@ -7118,6 +7119,14 @@ fn mux_clients_target_independent_drafts_and_receive_only_their_session_tail() { epoch: None, }) .unwrap(); + let window = smol::block_on(smol::future::race( + async { link.events().recv().await.unwrap() }, + async { + smol::Timer::after(Duration::from_secs(5)).await; + panic!("the subscription was not answered") + }, + )); + assert!(matches!(window.event, ServerEvent::SessionSnapshot(_))); id }; let id_one = draft(&one, "one"); @@ -7247,7 +7256,7 @@ fn session_history_snapshot_pages_and_absolute_tail_cursors() { after: None, epoch: None, }; - let snapshot = state.subscription_snapshot(&subscription).unwrap(); + let snapshot = events_reply(state, &subscription); assert!( tcode_protocol::encode_line(&HostMessage::Event(snapshot.clone())) .unwrap() @@ -7275,7 +7284,9 @@ fn session_history_snapshot_pages_and_absolute_tail_cursors() { end, truncated, .. - } = state.session_history_page("large", 0, before, 200).unwrap() + } = state.event_records["large"] + .history_page(0, before, 200) + .unwrap() else { panic!("page") }; @@ -7287,13 +7298,14 @@ fn session_history_snapshot_pages_and_absolute_tail_cursors() { } assert_eq!(loaded, records); for after in [0, 17, 1800, 1999, 2000] { - let snapshot = state - .subscription_snapshot(&tcode_protocol::Subscription { + let snapshot = events_reply( + state, + &tcode_protocol::Subscription { after: Some(after), epoch: Some(0), ..subscription.clone() - }) - .unwrap(); + }, + ); let ServerEvent::SessionSnapshot(tcode_protocol::SessionWindow { from, records: tail, @@ -7369,15 +7381,16 @@ fn history_snapshot_and_pages_start_at_turn_boundaries() { .event_records .insert("streamed".into(), SessionLog::new(0, records.clone())); - let snapshot = state - .subscription_snapshot(&tcode_protocol::Subscription { + let snapshot = events_reply( + state, + &tcode_protocol::Subscription { topic: Topic::SessionEvents { session_id: "streamed".into(), }, after: None, epoch: None, - }) - .unwrap(); + }, + ); let ServerEvent::SessionSnapshot(tcode_protocol::SessionWindow { from, .. }) = snapshot.event else { @@ -7394,8 +7407,8 @@ fn history_snapshot_and_pages_start_at_turn_boundaries() { records: page, truncated, .. - } = state - .session_history_page("streamed", 0, from, 200) + } = state.event_records["streamed"] + .history_page(0, from, 200) .unwrap() else { panic!("page") @@ -7462,68 +7475,50 @@ fn persist_streamed_turns(store: &SessionStore, id: &str, turns: u64) -> Vec 0 { - let QueryResponse::SessionHistoryPage { - records: page, - from, - end, - .. - } = state.session_history_page("paged", 0, before, 200).unwrap() - else { - panic!("page") - }; - assert_eq!(end, before); - loaded.splice(0..0, page); - before = from; - } - assert_eq!( - Timeline::fold_events(loaded), - Timeline::fold_events(records), - "merged deltas fold as the log does" - ); - }); + let mut loaded = window.records; + let mut before = window.from; + let mut id = 2; + while before > 0 { + let QueryResponse::SessionHistoryPage { + records: page, + from, + end, + .. + } = query_reply( + &through_pipe(&state, cx, id, history_page("paged", 0, before)), + id, + ) + else { + panic!("page") + }; + assert_eq!(end, before); + loaded.splice(0..0, page); + before = from; + id += 1; + } + assert_eq!( + Timeline::fold_events(loaded), + Timeline::fold_events(records), + "merged deltas fold as the log does" + ); assert_eq!( store.event_reads() - reads_before_open, 1, @@ -7539,16 +7534,17 @@ fn history_pages_of_an_opened_session_parse_the_log_once() { fn session_log_follows_residency_and_flushes_before_release() { let cx = &mut TestAppContext::default(); let store = TestStore::new("history-log-residency"); - let mut meta = SessionMeta::new(ProviderKind::ClaudeCode, store.root().clone(), None); - meta.id = "resident".into(); - seed_index(&store, &[], &[meta.clone()]); + open_stored_session(&store, "resident"); let persisted = persist_streamed_turns(&store, "resident", 2).len() as u64; let state = cx.new_entity(TestClientState::new((*store).clone())); let reads_before_open = store.event_reads(); let (commands, _actor) = smol::channel::unbounded(); - let released = state.update(cx, |state, cx| { - state.select_session("resident", cx); + window_reply( + &through_pipe(&state, cx, 1, subscribe_events("resident", None, None)), + 1, + ); + state.update(cx, |state, cx| { state.record_event( "resident", &AgentEvent::ItemCompleted(ThreadItem { @@ -7561,45 +7557,56 @@ fn session_log_follows_residency_and_flushes_before_release() { }, }), cx, - ); - let snapshot = state - .subscription_snapshot(&tcode_protocol::Subscription { - topic: Topic::SessionEvents { - session_id: "resident".into(), - }, - after: Some(persisted), - epoch: Some(0), - }) - .unwrap(); - let ServerEvent::SessionSnapshot(tcode_protocol::SessionWindow { - records, - total, - total_turns, - .. - }) = snapshot.event - else { - panic!("snapshot") - }; - assert_eq!((total, total_turns), (persisted + 1, 3)); - assert!( - matches!(&records[..], [record] if matches!(&record.event, AgentEvent::ItemCompleted(item) if item.id == "user-late")) - ); - let QueryResponse::SessionHistoryPage { from, records, .. } = state - .session_history_page("resident", 0, persisted + 1, 1) - .unwrap() - else { - panic!("page") - }; - assert_eq!((from, records.len()), (persisted, 1)); - assert_eq!( - state.store.event_reads() - reads_before_open, - 1, - "the appended record is served from the log, not a re-parse" - ); + ) + }); + let window = window_reply( + &through_pipe( + &state, + cx, + 2, + subscribe_events("resident", Some(persisted), Some(0)), + ), + 2, + ); + assert_eq!((window.total, window.total_turns), (persisted + 1, 3)); + assert!( + matches!(&window.records[..], [record] if matches!(&record.event, AgentEvent::ItemCompleted(item) if item.id == "user-late")) + ); + let page = tcode_protocol::ClientPayload::Query(tcode_protocol::Query::SessionHistoryPage { + session_id: "resident".into(), + epoch: 0, + before: persisted + 1, + limit: 1, + }); + let QueryResponse::SessionHistoryPage { from, records, .. } = + query_reply(&through_pipe(&state, cx, 3, page), 3) + else { + panic!("page") + }; + assert_eq!((from, records.len()), (persisted, 1)); + assert_eq!( + store.event_reads() - reads_before_open, + 1, + "the appended record is served from the log, not a re-parse" + ); - // Parked with a live provider, the session stays resident and cached. - state.selected_session_mut().unwrap().runtime = Runtime::Live(commands); - state.park_active(cx); + // Parked with a live provider, the session stays resident and cached. + state.update(cx, |state, _| { + state.residents.live.get_mut("resident").unwrap().runtime = Runtime::Live(commands); + }); + through_pipe( + &state, + cx, + 4, + tcode_protocol::ClientPayload::Unsubscribe(Subscription { + topic: Topic::SessionEvents { + session_id: "resident".into(), + }, + after: None, + epoch: None, + }), + ); + let released = state.update(cx, |state, cx| { assert!(state.residents.parked.contains_key("resident")); assert!(state.event_records.contains_key("resident")); @@ -7612,26 +7619,36 @@ fn session_log_follows_residency_and_flushes_before_release() { state.event_records["resident"].fold().clone() }); cx.run_until(|state| !state.event_records.contains_key("resident")); - state.update(cx, |state, _| { - assert!(!state.event_records.contains_key("resident")); - // Release compacts the log, so the conversation, not its record - // count, is what the next cold open must find. - assert_eq!( - Timeline::fold_events(state.store.read_events("resident")), - released - ); - }); + // Release compacts the log, so the conversation, not its record count, + // is what the next cold open must find. + assert_eq!( + Timeline::fold_events(store.read_events("resident")), + released + ); assert_eq!(store.event_reads() - reads_before_open, 2); } /// Send `payload` through the host pipe as client message `id`, and return -/// every message the host serialized since the last drain, once it parked. +/// every message the host serialized since the last drain, once the host +/// answered it and parked. fn through_pipe( state: &TestEntity, cx: &mut TestAppContext, id: u64, payload: tcode_protocol::ClientPayload, ) -> Vec { + send_to_host(state, cx, id, payload); + answered(cx, id) +} + +/// Hand `payload` to the host as client message `id`, without running +/// anything it queues. +fn send_to_host( + state: &TestEntity, + cx: &mut TestAppContext, + id: u64, + payload: tcode_protocol::ClientPayload, +) { state.update(cx, |state, host_cx| { crate::pipe::handle_client_message( &mut state.host, @@ -7643,8 +7660,38 @@ fn through_pipe( }, ) }); - cx.run_until_parked(); - cx.drain_outgoing() +} + +/// Run the host until it answered client message `id` and parked, and +/// return every message it serialized since the last drain. +fn answered(cx: &mut TestAppContext, id: u64) -> Vec { + let deadline = Instant::now() + Duration::from_secs(5); + let mut messages = Vec::new(); + loop { + cx.run_until_parked(); + messages.extend(cx.drain_outgoing()); + if messages.iter().any(|message| { + matches!(message, HostMessage::Ack { id: reply, .. } | HostMessage::QueryResult { id: reply, .. } if *reply == id) + }) { + return messages; + } + assert!( + Instant::now() < deadline, + "the host did not answer message {id} within five seconds" + ); + } +} + +/// The reply a subscription to a session whose log is resident gets. +fn events_reply(state: &AppState, subscription: &Subscription) -> EventEnvelope { + let Topic::SessionEvents { session_id } = &subscription.topic else { + panic!("a subscription to a session's events") + }; + EventEnvelope { + request_id: None, + topic: subscription.topic.clone(), + event: state.event_records[session_id].events_window(subscription), + } } fn subscribe_events( @@ -7724,6 +7771,423 @@ fn open_stored_session(store: &SessionStore, id: &str) -> SessionMeta { meta } +/// Subscriptions made while a cold log is read share that one read. Each +/// gets a window ending where the read log ends, and a record accepted +/// meanwhile follows every window, once, at the next position of the served +/// layout. +#[test] +fn subscriptions_made_while_a_cold_log_is_read_share_it_and_get_each_record_once() { + let cx = &mut TestAppContext::default(); + let store = TestStore::new("cold-subscriptions"); + open_stored_session(&store, "cold"); + persist_streamed_turns(&store, "cold", 2); + assert!(matches!( + store.compact_log("cold"), + tcode_services::store::CompactOutcome::Rewritten { .. } + )); + let persisted = store.read_events("cold").len() as u64; + let state = cx.new_entity(TestClientState::new((*store).clone())); + let reads_before_open = store.event_reads(); + + send_to_host(&state, cx, 1, subscribe_events("cold", None, None)); + send_to_host(&state, cx, 2, subscribe_events("cold", None, None)); + let live = AgentEvent::Warning { + message: "first live".into(), + }; + state.update(cx, |state, cx| { + assert!(state.log_hydrations.contains_key("cold")); + state.on_event("cold", live.clone(), cx); + }); + let messages = answered(cx, 2); + + let live_at = messages + .iter() + .position(|message| { + matches!( + message, + HostMessage::Event(EventEnvelope { + event: ServerEvent::SessionEvent { .. }, + .. + }) + ) + }) + .expect("the record accepted during the read is sent"); + for id in [1, 2] { + let window = window_reply(&messages, id); + assert_eq!( + (window.epoch, window.end, window.total), + (1, persisted, persisted) + ); + let window_at = messages + .iter() + .position(|message| { + matches!(message, HostMessage::Event(EventEnvelope { request_id: Some(request), .. }) if *request == id) + }) + .unwrap(); + assert!(window_at < live_at, "window {id} comes before the record"); + } + assert_eq!( + live_records(&messages) + .into_iter() + .map(|(epoch, position, record)| (epoch, position, record.event)) + .collect::>(), + [(1, persisted, live)] + ); + assert_eq!( + store.event_reads() - reads_before_open, + 1, + "both subscriptions were answered from one read" + ); +} + +/// The first event of a resident thread whose log is not loaded starts the +/// read of that log, which therefore does not contain it: a subscription +/// made before the read completes gets a window without it, and the event +/// follows once, at the position the window ends. +#[test] +fn the_event_that_starts_a_read_follows_the_window_once() { + let cx = &mut TestAppContext::default(); + let store = TestStore::new("first-event-read"); + let state = cx.new_entity(TestClientState::new((*store).clone())); + let first = AgentEvent::Warning { + message: "first".into(), + }; + let id = state.update(cx, |state, cx| { + state.start_draft("project".into(), store.root().clone(), cx); + let id = state.active_session_id().unwrap().to_string(); + state.record_event(&id, &first, cx); + assert!(state.log_hydrations.contains_key(&id)); + id + }); + send_to_host(&state, cx, 1, subscribe_events(&id, None, None)); + let messages = answered(cx, 1); + + let window = window_reply(&messages, 1); + assert_eq!((window.epoch, window.end, window.records.len()), (0, 0, 0)); + assert_eq!( + live_records(&messages) + .into_iter() + .map(|(epoch, position, record)| (epoch, position, record.event)) + .collect::>(), + [(0, 0, first)] + ); +} + +/// A subscriber that leaves a cold thread before its log is read gets no +/// window for it, and a log read for a session that is no longer resident is +/// not kept. +#[test] +fn a_cold_thread_left_before_its_log_is_read_sends_no_window_and_keeps_no_log() { + let cx = &mut TestAppContext::default(); + let store = TestStore::new("cold-left"); + open_stored_session(&store, "left"); + persist_streamed_turns(&store, "left", 2); + let state = cx.new_entity(TestClientState::new((*store).clone())); + + send_to_host(&state, cx, 1, subscribe_events("left", None, None)); + send_to_host( + &state, + cx, + 2, + tcode_protocol::ClientPayload::Unsubscribe(Subscription { + topic: Topic::SessionEvents { + session_id: "left".into(), + }, + after: None, + epoch: None, + }), + ); + state.read(|state| assert!(state.log_hydrations.contains_key("left"))); + let messages = answered(cx, 1); + + assert!( + messages + .iter() + .any(|message| matches!(message, HostMessage::Ack { id: 2, .. })) + ); + assert!( + !messages.iter().any(|message| matches!( + message, + HostMessage::Event(EventEnvelope { + topic: Topic::SessionEvents { .. }, + .. + }) + )), + "nothing of the thread reaches the client that left it" + ); + state.read(|state| { + assert!(state.resident("left").is_none()); + assert!(state.log_hydrations.is_empty()); + assert!(state.event_records.is_empty()); + }); +} + +/// History pages and whole outputs of a thread no client holds open are +/// answered from its whole log, read off the mailbox and not kept. +#[test] +fn pages_and_outputs_of_a_thread_nobody_opened_are_read_and_not_kept() { + let cx = &mut TestAppContext::default(); + let store = TestStore::new("cold-queries"); + open_stored_session(&store, "cold"); + let mut records = persist_streamed_turns(&store, "cold", 2); + let output = "o".repeat(2 * tcode_protocol::OUTPUT_PREVIEW_BYTES); + let tool = tool_call("tool", output.clone()); + store.append_event("cold", 5000, &tool.event).unwrap(); + records.push(tool); + let total = records.len() as u64; + let state = cx.new_entity(TestClientState::new((*store).clone())); + + let QueryResponse::SessionHistoryPage { + epoch, + end, + records: page, + .. + } = query_reply( + &through_pipe(&state, cx, 1, history_page("cold", 0, total)), + 1, + ) + else { + panic!("page") + }; + assert_eq!((epoch, end), (0, total)); + assert!( + matches!(&page.last().unwrap().event, AgentEvent::ItemCompleted(item) if item.id == "tool") + ); + let read_output = tcode_protocol::ClientPayload::Query(tcode_protocol::Query::ReadItemOutput { + session_id: "cold".into(), + item_id: "tool".into(), + }); + assert_eq!( + query_reply(&through_pipe(&state, cx, 2, read_output), 2), + QueryResponse::ItemOutput(output) + ); + state.read(|state| { + assert!(state.resident("cold").is_none()); + assert!(state.log_hydrations.is_empty()); + assert!(state.event_records.is_empty()); + }); +} + +/// A child whose turn completes while its log is still being read reports +/// that turn: its status and result, and the callback to its parent, see the +/// completion and the final message accepted just before it. +#[test] +fn a_child_completing_while_its_log_is_read_reports_that_turn() { + let cx = &mut TestAppContext::default(); + let store = TestStore::new("child-completing-while-read"); + let mut child = SessionMeta::new(ProviderKind::Codex, store.root().clone(), None); + child.id = "child".into(); + child.parent_session_id = Some("parent".into()); + child.archive_on_complete = false; + seed_index(&store, &[], std::slice::from_ref(&child)); + persist_streamed_turns(&store, "child", 1); + let state = cx.new_entity(TestClientState::new((*store).clone())); + let (parent_commands, parent_receiver) = smol::channel::unbounded(); + + state.update(cx, |state, cx| { + let mut parent = live_session(ProviderKind::Codex, parent_commands); + parent.meta.id = "parent".into(); + parent.turn_in_flight = true; + state + .residents + .parked + .insert(parent.meta.id.clone(), parent); + state.load_background_session(child, cx); + for event in [ + AgentEvent::TurnStarted { + turn_id: "turn-new".into(), + }, + persisted_assistant_event("the new answer"), + AgentEvent::TurnCompleted { + turn_id: "turn-new".into(), + status: TurnStatus::Completed, + usage: None, + }, + ] { + state.on_event("child", event, cx); + } + assert!(state.log_hydrations.contains_key("child")); + + let (reply, response) = smol::channel::bounded(1); + state.handle_orchestrate_op( + orchestrate_mcp::OrchestrateOp::Status { + parent_id: "parent".into(), + thread_id: Some("child".into()), + }, + reply, + cx, + ); + let status = response.try_recv().unwrap().unwrap(); + assert_eq!(status[0]["state"], "completed"); + assert_eq!(status[0]["last_output_tail"], "the new answer"); + let (reply, response) = smol::channel::bounded(1); + state.handle_orchestrate_op( + orchestrate_mcp::OrchestrateOp::Result { + parent_id: "parent".into(), + thread_id: "child".into(), + }, + reply, + cx, + ); + let result = response.try_recv().unwrap().unwrap(); + assert_eq!(result["state"], "completed"); + assert_eq!(result["final_message"], "the new answer"); + }); + + cx.run_until(|state| state.callback_last_turn.contains_key("child")); + let callback = std::iter::from_fn(|| parent_receiver.try_recv().ok()) + .find_map(|command| match command { + SessionCommand::Steer { text, .. } => Some(text), + _ => None, + }) + .expect("the parent is called back"); + assert!(callback.starts_with("[orchestrate] thread child")); + assert!(callback.ends_with("\nthe new answer")); +} + +/// Deliver `event` as the session's provider would, as soon as a read of the +/// session's log is in flight. Queued on the mailbox before that read began, +/// it runs before the read's completion, which is queued only after. +fn record_while_hydrating( + state: &mut AppState, + cx: &mut HostCx, + session_id: &'static str, + event: AgentEvent, + recorded: smol::channel::Sender<()>, +) { + assert!(!state.event_records.contains_key(session_id)); + if state.log_hydrations.contains_key(session_id) { + state.on_event(session_id, event, cx); + let _ = recorded.try_send(()); + } else { + cx.enqueue(move |state, cx| record_while_hydrating(state, cx, session_id, event, recorded)); + } +} + +/// Two clients of a real host, through the multiplexer, open a cold thread +/// whose first live event arrives while its log is read: each gets its own +/// window, routed to it although it comes long after the request, then the +/// event; every record reaches each client once, at consecutive positions of +/// the served layout, and the log is read once. +#[test] +fn clients_opening_a_cold_thread_through_the_mux_get_each_record_once() { + use crate::pipe::{HostServices, spawn_host}; + use tcode_client::HostLink; + + let root = TestStore::new("cold-open-mux"); + open_stored_session(&root, "cold"); + persist_streamed_turns(&root, "cold", 3); + assert!(matches!( + root.compact_log("cold"), + tcode_services::store::CompactOutcome::Rewritten { .. } + )); + let persisted = root.read_events("cold").len() as u64; + let reads_before_open = root.event_reads(); + let host = spawn_host((*root).clone(), HostServices::default()).unwrap(); + let mux = tcode_traverse::HostMux::new(host.to_host.clone(), host.from_host.clone()); + let clients: Vec = (0..2) + .map(|_| { + let connection = mux.attach(); + let link = HostLink::new(connection.to_host, connection.from_host); + smol::spawn({ + let link = link.clone(); + async move { link.pump().await } + }) + .detach(); + link + }) + .collect(); + + // Hold the host until the event is queued ahead of anything the + // subscriptions start. + let (entered, host_held) = smol::channel::bounded(1); + let (release, released) = smol::channel::bounded::<()>(1); + let (recorded, event_recorded) = smol::channel::bounded(1); + let holder = host.clone(); + let held = smol::spawn(async move { + holder + .update_state_for_test(move |_, cx| { + entered.try_send(()).unwrap(); + released.recv_blocking().unwrap(); + let event = AgentEvent::Warning { + message: "first live".into(), + }; + cx.enqueue(move |state, cx| { + record_while_hydrating(state, cx, "cold", event, recorded) + }); + }) + .await + }); + host_held.recv_blocking().unwrap(); + for link in &clients { + link.subscribe(Subscription { + topic: Topic::SessionEvents { + session_id: "cold".into(), + }, + after: None, + epoch: None, + }) + .unwrap(); + } + release.try_send(()).unwrap(); + smol::block_on(held).unwrap(); + + fn within_five_seconds(future: impl std::future::Future) -> T { + smol::block_on(smol::future::race(future, async { + smol::Timer::after(Duration::from_secs(5)).await; + panic!("the host stopped sending the thread") + })) + } + within_five_seconds(event_recorded.recv()).unwrap(); + let next_event = |link: &HostLink| within_five_seconds(link.events().recv()).unwrap(); + for link in &clients { + let mut window = None; + let mut end = 0; + let mut seen = 0; + while end < persisted + 1 { + match next_event(link).event { + ServerEvent::SessionSnapshot(received) => { + assert!(window.is_none(), "one window per subscription"); + assert_eq!(received.epoch, 1); + end = received.end; + seen += received + .records + .iter() + .filter(|record| matches!(record.event, AgentEvent::Warning { .. })) + .count(); + window = Some(received); + } + ServerEvent::SessionEvent { + epoch, + position, + record, + } => { + // The mux routes a topic to a connection once it forwards + // the subscription, so a record can precede the window of + // a subscription the host answers later; clients drop it. + if window.is_none() { + continue; + } + assert_eq!((epoch, position), (1, end)); + end += 1; + seen += usize::from(matches!(record.event, AgentEvent::Warning { .. })); + } + other => panic!("unexpected {other:?}"), + } + } + assert_eq!(end, persisted + 1); + assert_eq!(seen, 1, "the live record reaches the client once"); + } + assert_eq!( + root.event_reads() - reads_before_open, + 1, + "every subscriber was served by one read" + ); + clients[0].shutdown_blocking().unwrap(); + host.to_host.close(); + host.stopped.recv_blocking().unwrap(); +} + /// A cursor is a position in one layout. Against a log rewritten under a new /// epoch, a position from the old layout, even one the new layout still has, /// or a position without its layout gets a baseline that replaces what the @@ -8099,9 +8563,14 @@ fn startup_recovers_logs_from_every_step_a_rewrite_can_stop_at() { }) .unwrap(); } - link.command_blocking(Command::ClearRelaunchMarker).unwrap(); + let deadline = Instant::now() + Duration::from_secs(5); let mut served = HashMap::new(); - while let Ok(envelope) = link.events().try_recv() { + while served.len() < 3 { + assert!(Instant::now() < deadline, "every subscription is answered"); + let Ok(envelope) = link.events().try_recv() else { + std::thread::sleep(Duration::from_millis(1)); + continue; + }; if let (Topic::SessionEvents { session_id }, ServerEvent::SessionSnapshot(window)) = (envelope.topic, envelope.event) { @@ -8202,7 +8671,7 @@ fn history_byte_budget_preserves_contiguous_records_and_reports_shrinking() { after: None, epoch: None, }; - let mut snapshot = state.subscription_snapshot(&subscription).unwrap(); + let mut snapshot = events_reply(state, &subscription); snapshot.request_id = Some(u64::MAX); assert!( tcode_protocol::encode_line(&HostMessage::Event(snapshot.clone())) @@ -8221,8 +8690,8 @@ fn history_byte_budget_preserves_contiguous_records_and_reports_shrinking() { }; assert!(truncated); assert_eq!(tail, records[from as usize..]); - let response = state - .session_history_page("large", 0, 10, u32::MAX) + let response = state.event_records["large"] + .history_page(0, 10, u32::MAX) .unwrap(); let line = tcode_protocol::encode_line(&HostMessage::QueryResult { id: u64::MAX, @@ -8241,13 +8710,14 @@ fn history_byte_budget_preserves_contiguous_records_and_reports_shrinking() { }; assert!(truncated); assert_eq!(page, records[from as usize..]); - let snapshot = state - .subscription_snapshot(&tcode_protocol::Subscription { + let snapshot = events_reply( + state, + &tcode_protocol::Subscription { after: Some(0), epoch: Some(0), ..subscription - }) - .unwrap(); + }, + ); let ServerEvent::SessionSnapshot(tcode_protocol::SessionWindow { from, records: tail, @@ -8268,8 +8738,8 @@ fn history_byte_budget_preserves_contiguous_records_and_reports_shrinking() { .event_records .insert("large".into(), SessionLog::new(0, records)); assert_eq!( - state - .session_history_page("large", 0, 10, 1) + state.event_records["large"] + .history_page(0, 10, 1) .unwrap_err() .code, "history_record_too_large" @@ -8643,22 +9113,22 @@ fn history_paging_bench() { let state = cx.new_entity(TestClientState::new((*store).clone())); let started = Instant::now(); state.update(cx, |state, cx| state.select_session("bench", cx)); + cx.run_until(|state| state.event_records.contains_key("bench")); let opened = started.elapsed(); let (snapshot_elapsed, pages) = state.update(cx, |state, _| { + let log = &state.event_records["bench"]; let started = Instant::now(); - let snapshot = state - .subscription_snapshot(&tcode_protocol::Subscription { - topic: Topic::SessionEvents { - session_id: "bench".into(), - }, - after: None, - epoch: None, - }) - .unwrap(); + let snapshot = log.events_window(&tcode_protocol::Subscription { + topic: Topic::SessionEvents { + session_id: "bench".into(), + }, + after: None, + epoch: None, + }); let snapshot_elapsed = started.elapsed(); let ServerEvent::SessionSnapshot(tcode_protocol::SessionWindow { epoch, from, total, .. - }) = snapshot.event + }) = snapshot else { panic!("snapshot") }; @@ -8667,9 +9137,8 @@ fn history_paging_bench() { let mut pages = Vec::new(); while before > 0 && pages.len() < 16 { let started = Instant::now(); - let QueryResponse::SessionHistoryPage { from, records, .. } = state - .session_history_page( - "bench", + let QueryResponse::SessionHistoryPage { from, records, .. } = log + .history_page( epoch, before, tcode_protocol::SESSION_HISTORY_RECORDS as u32, @@ -8689,7 +9158,7 @@ fn history_paging_bench() { eprintln!("page {index}: {elapsed:?} ({records} records)"); } eprintln!( - "select_session: {opened:?}; snapshot: {snapshot_elapsed:?}; {} pages: {paged:?}; snapshot + pages: {:?}", + "opened (read off the mailbox, to parked): {opened:?}; snapshot: {snapshot_elapsed:?}; {} pages: {paged:?}; snapshot + pages: {:?}", pages.len(), snapshot_elapsed + paged ); @@ -8844,15 +9313,16 @@ fn history_sends_output_previews_and_reads_whole_outputs_on_request() { state .event_records .insert("outputs".into(), SessionLog::new(0, records.clone())); - let snapshot = state - .subscription_snapshot(&tcode_protocol::Subscription { + let snapshot = events_reply( + state, + &tcode_protocol::Subscription { topic: Topic::SessionEvents { session_id: "outputs".into(), }, after: None, epoch: None, - }) - .unwrap(); + }, + ); let ServerEvent::SessionSnapshot(tcode_protocol::SessionWindow { records: sent, .. }) = snapshot.event else { @@ -8893,15 +9363,22 @@ fn history_sends_output_previews_and_reads_whole_outputs_on_request() { ]) ); assert_eq!( - state.item_output("outputs", "tool").unwrap(), + state.event_records["outputs"] + .item_output("outputs", "tool") + .unwrap(), QueryResponse::ItemOutput(tool_output.clone()) ); assert_eq!( - state.item_output("outputs", "command").unwrap(), + state.event_records["outputs"] + .item_output("outputs", "command") + .unwrap(), QueryResponse::ItemOutput(command_output.clone()) ); assert_eq!( - state.item_output("outputs", "missing").unwrap_err().code, + state.event_records["outputs"] + .item_output("outputs", "missing") + .unwrap_err() + .code, "unknown_item_output" ); }); @@ -8946,7 +9423,9 @@ fn superseded_turn_changes_cross_without_diffs() { .event_records .insert("changes".into(), SessionLog::new(0, records.clone())); let QueryResponse::SessionHistoryPage { records: sent, .. } = - state.session_history_page("changes", 0, 4, 200).unwrap() + state.event_records["changes"] + .history_page(0, 4, 200) + .unwrap() else { panic!("page") }; @@ -8979,15 +9458,16 @@ fn history_windows_are_byte_budgeted() { state .event_records .insert("wide".into(), SessionLog::new(0, records.clone())); - let mut snapshot = state - .subscription_snapshot(&tcode_protocol::Subscription { + let mut snapshot = events_reply( + state, + &tcode_protocol::Subscription { topic: Topic::SessionEvents { session_id: "wide".into(), }, after: None, epoch: None, - }) - .unwrap(); + }, + ); snapshot.request_id = Some(u64::MAX); let line = tcode_protocol::encode_line(&HostMessage::Event(snapshot.clone())).unwrap(); assert!(line.len() <= tcode_protocol::SESSION_WINDOW_BYTES); @@ -9017,7 +9497,9 @@ fn history_windows_are_byte_budgeted() { from, records: sent, .. - } = state.session_history_page("wide", 0, 200, 200).unwrap() + } = state.event_records["wide"] + .history_page(0, 200, 200) + .unwrap() else { panic!("page") }; diff --git a/crates/runtime/src/pipe.rs b/crates/runtime/src/pipe.rs index ba759f275..5bdfbf024 100644 --- a/crates/runtime/src/pipe.rs +++ b/crates/runtime/src/pipe.rs @@ -12,7 +12,7 @@ use tcode_protocol::{EventEnvelope, ServerEvent, Subscription, Topic}; use tcode_services::store::SessionStore; use crate::app::{AppState, DomainDiff}; -use crate::host::{HostCx, HostEvent, HostFn}; +use crate::host::{HostCx, HostFn}; /// Optional process-local services attached before the host starts accepting /// client traffic. @@ -290,14 +290,7 @@ pub(crate) fn handle_client_message(state: &mut AppState, cx: &mut HostCx, messa } ClientPayload::Subscribe(subscription) => { state.subscribe(&subscription, cx); - if let Some(mut snapshot) = state.subscription_snapshot(&subscription) { - snapshot.request_id = Some(id); - cx.emit(HostEvent::Domain(snapshot)); - } - cx.send_message(HostMessage::Ack { - id, - result: Ok(CommandResponse::Unit), - }); + state.reply_to_subscription(Some(id), subscription, cx); } ClientPayload::Unsubscribe(subscription) => { state.unsubscribe(&subscription, cx); @@ -607,10 +600,7 @@ fn dispatch_query( epoch, before, limit, - } => { - let result = app.session_history_page(&session_id, epoch, before, limit); - cx.spawn_background(async move { result }) - } + } => app.session_history_page(&session_id, epoch, before, limit, cx), Query::Hosting { .. } => cx.spawn_background(async { Err(ProtocolError { code: "unsupported".into(), @@ -739,10 +729,7 @@ fn dispatch_query( Query::ReadItemOutput { session_id, item_id, - } => { - let result = app.item_output(&session_id, &item_id); - cx.spawn_background(async move { result }) - } + } => app.item_output(&session_id, item_id, cx), Query::ArchivedSessions => { let sessions = app.archived_sessions(); cx.spawn_background(async move { Ok(QueryResponse::ArchivedSessions(sessions)) }) diff --git a/crates/services/src/store.rs b/crates/services/src/store.rs index a0d2ca520..ec3955179 100644 --- a/crates/services/src/store.rs +++ b/crates/services/src/store.rs @@ -132,8 +132,8 @@ impl SessionStore { }) } - /// How many times [`SessionStore::read_events`] parsed a log through this - /// store or any of its clones. + /// How many times a log was parsed through this store or any of its + /// clones. #[cfg(any(test, feature = "test-support"))] pub fn event_reads(&self) -> usize { self.event_reads.load(std::sync::atomic::Ordering::Relaxed) @@ -312,18 +312,50 @@ impl SessionStore { self.read_log(id).records } - /// [`SessionStore::read_events`] with the layout the records are in. Both - /// come from one open file, so a rewrite renamed over the log meanwhile - /// cannot pair one layout's epoch with the other's records. + /// [`SessionStore::read_events`] with the layout the records are in. pub fn read_log(&self, id: &str) -> EventLog { + self.snapshot_log(id).read() + } + + /// Open a session's log as it is now, without reading it. Cheap enough + /// for the store writer to take between two writes. + pub fn snapshot_log(&self, id: &str) -> LogSnapshot { + let file = File::open(self.events_path(id)).ok().and_then(|file| { + let len = file.metadata().ok()?.len(); + Some((file, len)) + }); + LogSnapshot { + id: id.to_string(), + file, + #[cfg(any(test, feature = "test-support"))] + event_reads: self.event_reads.clone(), + } + } +} + +/// A session's log file opened at one moment, with the length it had then. +pub struct LogSnapshot { + id: String, + file: Option<(File, u64)>, + #[cfg(any(test, feature = "test-support"))] + event_reads: std::sync::Arc, +} + +impl LogSnapshot { + /// Parse every record the log held when it was opened. The epoch and the + /// records come from that one open file, so a rewrite renamed over the + /// log since cannot pair one layout's epoch with the other's records, and + /// what was appended since is not read. + pub fn read(self) -> EventLog { #[cfg(any(test, feature = "test-support"))] self.event_reads .fetch_add(1, std::sync::atomic::Ordering::Relaxed); + let id = self.id; let mut log = EventLog::default(); - let Ok(file) = File::open(self.events_path(id)) else { + let Some((file, len)) = self.file else { return log; }; - for (number, line) in BufReader::new(file).lines().enumerate() { + for (number, line) in BufReader::new(file.take(len)).lines().enumerate() { let Ok(line) = line else { break }; let trimmed = line.trim(); if trimmed.is_empty() { @@ -342,7 +374,9 @@ impl SessionStore { } log } +} +impl SessionStore { /// Atomically clone one session's append-only event log. A missing source /// is an empty transcript and therefore succeeds without creating a file. /// The copy is byte for byte, layout header included: it starts in the