From 446d3b9c211dd13335dc82f2343e8851973225ef Mon Sep 17 00:00:00 2001 From: iris <2+iris@noreply.localhost> Date: Sun, 30 Aug 2026 19:49:39 -0400 Subject: [PATCH] Find a transcript page by bisection instead of parsing the file Every page-back re-read and re-parsed the whole transcript and then threw away all but the window. That is the cost that grows with the conversation rather than with the answer: measured on a 21 MB, 24,000-event transcript, one page took ~500ms of server time to return 620 KB, and it took the same 500ms whichever page was asked for. A phone scrolling up pays that per page, and every stream reconnect pays it again to discover there is nothing new. Sequence numbers only increase, so the boundary of a range is a bisection. `Indexed` locates the lines without reading them, finds the edge by parsing one line per halving, and parses only what is going to be returned. Same page, 232ms including the 120ms `--delay` -- so ~110ms, of which ~20ms is the file scan and the rest is serialising the 620 KB that was always going to be sent. `catch_up` gets it too, and there the case that looks least interesting is the one that mattered: a subscriber with no cursor asks for the whole conversation and is handed the last CATCH_UP_LIMIT events of it. The file is still read whole, which is a deliberate stop -- finding the tail without reading forwards means a chunked backwards reader, and locating a line is not what the half-second was going to. Co-Authored-By: Claude Opus 5 --- server/src/routes.rs | 12 ++ server/src/session/transcript.rs | 198 +++++++++++++++++++++++++------ 2 files changed, 177 insertions(+), 33 deletions(-) diff --git a/server/src/routes.rs b/server/src/routes.rs index e949d90..c761808 100644 --- a/server/src/routes.rs +++ b/server/src/routes.rs @@ -974,6 +974,18 @@ async fn transcript( query.limit, ) .map_err(bad_request)?; + // How far back a phone has paged, and how much each page cost it to get + // there, which is the one question this route raises and nothing else + // can answer: the app asks for events and draws rows, and the ratio + // between them is a property of the conversation. `RUST_LOG=ai_server=debug`. + tracing::debug!( + session = %id, + before = ?query.before, + limit = query.limit, + got = events.len(), + oldest = ?events.first().map(|entry| entry.seq), + "transcript page" + ); Ok(axum::Json(events)) } diff --git a/server/src/session/transcript.rs b/server/src/session/transcript.rs index 249844d..fdd32f8 100644 --- a/server/src/session/transcript.rs +++ b/server/src/session/transcript.rs @@ -7,7 +7,8 @@ //! a backend restart invisible to a phone holding a cursor. use std::fs::{File, OpenOptions}; -use std::io::{BufRead, BufReader, Write}; +use std::io::Write; +use std::ops::Range; use std::os::unix::fs::OpenOptionsExt; use std::path::Path; @@ -144,19 +145,18 @@ impl Transcript { /// was several seconds of messages arriving oldest-first, which reads as /// the app loading top-down because that is exactly what it was doing. /// -/// `before` pages backwards for history somebody actually scrolls to; the -/// file is read whole each time because a transcript is small and a -/// seek-backwards reader would be a lot of machinery for a list that fits -/// in memory anyway. +/// `before` pages backwards for history somebody actually scrolls to. Only +/// the window is parsed; see [`Indexed`] for why that is the whole cost of +/// this call. pub fn read_window(path: &Path, before: Option, limit: usize) -> Result> { - let mut all = read_after(path, 0)?; - if let Some(before) = before { - all.retain(|entry| entry.seq < before); - } - if all.len() > limit { - all.drain(..all.len() - limit); - } - Ok(all) + let Some(indexed) = Indexed::read(path)? else { + return Ok(Vec::new()); + }; + let end = match before { + Some(before) => indexed.first_at_or_after(before)?, + None => indexed.lines.len(), + }; + indexed.parse(end.saturating_sub(limit)..end) } /// How far behind a reconnecting subscriber can be and still be handed the @@ -191,37 +191,130 @@ pub enum CatchUp { /// Everything after `after`, or the newest `limit` when that is more than /// `limit` events. +/// +/// The window is chosen before anything is parsed, which matters most in +/// the case that looks least interesting: a subscriber with no cursor at +/// all asks for the whole conversation and is going to be handed the last +/// [`CATCH_UP_LIMIT`] events of it. Parsing the discarded prefix first is +/// the whole file's worth of work to produce a screenful. pub fn catch_up(path: &Path, after: u64, limit: usize) -> Result { - let mut events = read_after(path, after)?; - if events.len() > limit { - events.drain(..events.len() - limit); - return Ok(CatchUp::Restart(events)); + let Some(indexed) = Indexed::read(path)? else { + return Ok(CatchUp::Continue(Vec::new())); + }; + let end = indexed.lines.len(); + let start = indexed.first_at_or_after(after.saturating_add(1))?; + if end - start > limit { + return Ok(CatchUp::Restart(indexed.parse(end - limit..end)?)); } - Ok(CatchUp::Continue(events)) + Ok(CatchUp::Continue(indexed.parse(start..end)?)) } /// Replays every event with `seq > after`, oldest first. A missing file is /// an empty transcript, not an error -- the session just hasn't produced an /// event yet. pub fn read_after(path: &Path, after: u64) -> Result> { - let file = match File::open(path) { - Ok(file) => file, - Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()), - Err(err) => return Err(err).with_context(|| format!("read transcript {}", path.display())), + let Some(indexed) = Indexed::read(path)? else { + return Ok(Vec::new()); }; - let mut events = Vec::new(); - for line in BufReader::new(file).lines() { - let line = line.context("read transcript line")?; - if line.trim().is_empty() { - continue; - } - let entry: SeqEvent = serde_json::from_str(&line) - .with_context(|| format!("bad transcript line in {}", path.display()))?; - if entry.seq > after { - events.push(entry); + let start = indexed.first_at_or_after(after.saturating_add(1))?; + indexed.parse(start..indexed.lines.len()) +} + +/// The transcript's lines located but not read, so that a reader can find +/// the range it wants and parse only that. +/// +/// Both readers above want a *range* of the file -- everything after a +/// cursor, or the window before one -- and both used to reach it by parsing +/// every line and discarding the ones outside it. That is the cost that +/// grows with the conversation rather than with the answer: measured on a +/// 21 MB, 24,000-event transcript, one page took **500 ms of server time to +/// return 600 KB**, and it took the same 500 ms whichever page was asked +/// for, since the work was the file rather than the window. A phone paging +/// back through history pays it per page, and every stream reconnect pays +/// it again to discover there is nothing new. +/// +/// Sequence numbers only ever increase -- the writer assigns them, one per +/// appended line, continuing from the last on reopen -- so the boundary of +/// a range is a bisection. This parses one line per halving, and the caller +/// parses only what it is going to return. The file is still read whole, +/// which is a deliberate stop: finding the tail without reading forwards +/// means a chunked backwards reader, and locating a line is not what the +/// half-second was going to. +struct Indexed<'a> { + path: &'a Path, + text: String, + /// Byte range of each non-blank line, in the order they were written. + lines: Vec>, +} + +impl<'a> Indexed<'a> { + /// `None` for a file that isn't there, which is a session that has not + /// produced an event yet rather than a failure. + fn read(path: &'a Path) -> Result> { + let text = match std::fs::read_to_string(path) { + Ok(text) => text, + Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(None), + Err(err) => { + return Err(err).with_context(|| format!("read transcript {}", path.display())); + } + }; + let mut lines = Vec::new(); + let mut start = 0; + while start < text.len() { + let end = text[start..] + .find('\n') + .map(|at| start + at) + .unwrap_or(text.len()); + if !text[start..end].trim().is_empty() { + lines.push(start..end); + } + start = end + 1; } + Ok(Some(Self { path, text, lines })) + } + + /// The index of the first line numbered `seq` or higher, or the end + /// when every line is older than that. + /// + /// A bisection, which is only correct because the file is in sequence + /// order; it is append-only and nothing else writes it. A line that + /// cannot be read is reported here rather than silently treated as + /// out of range, because the answer would be a window off by however + /// much of the file the bad line hid. + fn first_at_or_after(&self, seq: u64) -> Result { + let (mut low, mut high) = (0, self.lines.len()); + while low < high { + let middle = (low + high) / 2; + if self.seq_at(middle)? < seq { + low = middle + 1; + } else { + high = middle; + } + } + Ok(low) + } + + /// One line's sequence number, without building the event on it. + fn seq_at(&self, index: usize) -> Result { + #[derive(Deserialize)] + struct JustSeq { + seq: u64, + } + let line = &self.text[self.lines[index].clone()]; + let entry: JustSeq = serde_json::from_str(line) + .with_context(|| format!("bad transcript line in {}", self.path.display()))?; + Ok(entry.seq) + } + + fn parse(&self, range: Range) -> Result> { + self.lines[range] + .iter() + .map(|at| { + serde_json::from_str(&self.text[at.clone()]) + .with_context(|| format!("bad transcript line in {}", self.path.display())) + }) + .collect() } - Ok(events) } #[cfg(test)] @@ -341,6 +434,45 @@ mod tests { assert_eq!(reopened.next_seq, 4); } + #[test] + fn a_window_is_the_events_before_a_cursor_and_nothing_else() { + let dir = tempfile::tempdir().expect("tempdir"); + let path = dir.path().join("transcript.jsonl"); + let mut transcript = Transcript::open(&path).expect("open"); + for n in 1..=10 { + transcript + .append(text(&n.to_string()), 0.0) + .expect("append"); + } + + // No cursor is the newest page, which is what opening a session asks for. + let newest = read_window(&path, None, 3).expect("window"); + assert_eq!( + newest.iter().map(|entry| entry.seq).collect::>(), + [8, 9, 10] + ); + + // Then backwards from the oldest of those, exclusive: the page a phone + // scrolling up asks for must not repeat the row it is scrolling from. + let older = read_window(&path, Some(8), 3).expect("window"); + assert_eq!( + older.iter().map(|entry| entry.seq).collect::>(), + [5, 6, 7] + ); + + // Asking for more than there is gives what there is, rather than failing. + assert_eq!(read_window(&path, None, 100).expect("window").len(), 10); + + // Nothing before the first event, which is how the phone learns to stop + // paging. An empty answer here is the end of the history, not a fault. + assert!(read_window(&path, Some(1), 3).expect("window").is_empty()); + assert!( + read_window(&dir.path().join("nope.jsonl"), None, 3) + .expect("window") + .is_empty() + ); + } + #[test] fn a_missing_file_reads_as_empty() { let dir = tempfile::tempdir().expect("tempdir");