From 5671204b889865fd886913ae28fd09cd9e900370 Mon Sep 17 00:00:00 2001 From: Tryanks Date: Mon, 5 Oct 2026 06:13:01 +0800 Subject: [PATCH] perf: read a cold thread's log off the host mailbox Opening a cold thread, the first event of an uncached thread, and history pages or full outputs of a thread nobody has open each parsed and folded the whole log on the host mailbox, stalling every other session for as long as that took. A session now has at most one hydration. The store writer opens the log in its queue order and records its length, so the read holds exactly the appends queued before it; the parse and fold run on the blocking pool. Records accepted meanwhile are held in order and take the positions after the snapshot's end. When the log is resident the waiting subscription windows go out first, each ending where the snapshot ends, then the held records, so a subscriber sees every record once. A subscription's acknowledgement waits for its window. The log is kept only if the session is still resident. Part of #524 --- crates/runtime/src/app/events.rs | 48 +- crates/runtime/src/app/history.rs | 436 ++++++++++---- crates/runtime/src/app/mod.rs | 22 +- crates/runtime/src/app/sessions.rs | 103 ++-- crates/runtime/src/app/snapshots.rs | 3 +- crates/runtime/src/app/store_write.rs | 11 +- crates/runtime/src/app/tests.rs | 798 +++++++++++++++++++++----- crates/runtime/src/pipe.rs | 21 +- crates/services/src/store.rs | 48 +- 9 files changed, 1119 insertions(+), 371 deletions(-) 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