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 <noreply@anthropic.com>
This commit is contained in:
1 parent
c782a1264b
commit
446d3b9c21
2 files changed
+176
-32
No files matched your search
@@ -974,6 +974,18 @@ async fn transcript(
|
|||||||
query.limit,
|
query.limit,
|
||||||
)
|
)
|
||||||
.map_err(bad_request)?;
|
.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))
|
Ok(axum::Json(events))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -7,7 +7,8 @@
|
|||||||
//! a backend restart invisible to a phone holding a cursor.
|
//! a backend restart invisible to a phone holding a cursor.
|
||||||
|
|
||||||
use std::fs::{File, OpenOptions};
|
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::os::unix::fs::OpenOptionsExt;
|
||||||
use std::path::Path;
|
use std::path::Path;
|
||||||
|
|
||||||
@@ -144,19 +145,18 @@ impl Transcript {
|
|||||||
/// was several seconds of messages arriving oldest-first, which reads as
|
/// was several seconds of messages arriving oldest-first, which reads as
|
||||||
/// the app loading top-down because that is exactly what it was doing.
|
/// the app loading top-down because that is exactly what it was doing.
|
||||||
///
|
///
|
||||||
/// `before` pages backwards for history somebody actually scrolls to; the
|
/// `before` pages backwards for history somebody actually scrolls to. Only
|
||||||
/// file is read whole each time because a transcript is small and a
|
/// the window is parsed; see [`Indexed`] for why that is the whole cost of
|
||||||
/// seek-backwards reader would be a lot of machinery for a list that fits
|
/// this call.
|
||||||
/// in memory anyway.
|
|
||||||
pub fn read_window(path: &Path, before: Option<u64>, limit: usize) -> Result<Vec<SeqEvent>> {
|
pub fn read_window(path: &Path, before: Option<u64>, limit: usize) -> Result<Vec<SeqEvent>> {
|
||||||
let mut all = read_after(path, 0)?;
|
let Some(indexed) = Indexed::read(path)? else {
|
||||||
if let Some(before) = before {
|
return Ok(Vec::new());
|
||||||
all.retain(|entry| entry.seq < before);
|
};
|
||||||
}
|
let end = match before {
|
||||||
if all.len() > limit {
|
Some(before) => indexed.first_at_or_after(before)?,
|
||||||
all.drain(..all.len() - limit);
|
None => indexed.lines.len(),
|
||||||
}
|
};
|
||||||
Ok(all)
|
indexed.parse(end.saturating_sub(limit)..end)
|
||||||
}
|
}
|
||||||
|
|
||||||
/// How far behind a reconnecting subscriber can be and still be handed the
|
/// 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
|
/// Everything after `after`, or the newest `limit` when that is more than
|
||||||
/// `limit` events.
|
/// `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<CatchUp> {
|
pub fn catch_up(path: &Path, after: u64, limit: usize) -> Result<CatchUp> {
|
||||||
let mut events = read_after(path, after)?;
|
let Some(indexed) = Indexed::read(path)? else {
|
||||||
if events.len() > limit {
|
return Ok(CatchUp::Continue(Vec::new()));
|
||||||
events.drain(..events.len() - limit);
|
};
|
||||||
return Ok(CatchUp::Restart(events));
|
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
|
/// 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
|
/// an empty transcript, not an error -- the session just hasn't produced an
|
||||||
/// event yet.
|
/// event yet.
|
||||||
pub fn read_after(path: &Path, after: u64) -> Result<Vec<SeqEvent>> {
|
pub fn read_after(path: &Path, after: u64) -> Result<Vec<SeqEvent>> {
|
||||||
let file = match File::open(path) {
|
let Some(indexed) = Indexed::read(path)? else {
|
||||||
Ok(file) => file,
|
return Ok(Vec::new());
|
||||||
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 mut events = Vec::new();
|
let start = indexed.first_at_or_after(after.saturating_add(1))?;
|
||||||
for line in BufReader::new(file).lines() {
|
indexed.parse(start..indexed.lines.len())
|
||||||
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()))?;
|
/// The transcript's lines located but not read, so that a reader can find
|
||||||
if entry.seq > after {
|
/// the range it wants and parse only that.
|
||||||
events.push(entry);
|
///
|
||||||
|
/// 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<Range<usize>>,
|
||||||
|
}
|
||||||
|
|
||||||
|
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<Option<Self>> {
|
||||||
|
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<usize> {
|
||||||
|
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(events)
|
Ok(low)
|
||||||
|
}
|
||||||
|
|
||||||
|
/// One line's sequence number, without building the event on it.
|
||||||
|
fn seq_at(&self, index: usize) -> Result<u64> {
|
||||||
|
#[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<usize>) -> Result<Vec<SeqEvent>> {
|
||||||
|
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()
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
@@ -341,6 +434,45 @@ mod tests {
|
|||||||
assert_eq!(reopened.next_seq, 4);
|
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::<Vec<_>>(),
|
||||||
|
[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::<Vec<_>>(),
|
||||||
|
[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]
|
#[test]
|
||||||
fn a_missing_file_reads_as_empty() {
|
fn a_missing_file_reads_as_empty() {
|
||||||
let dir = tempfile::tempdir().expect("tempdir");
|
let dir = tempfile::tempdir().expect("tempdir");
|
||||||
|
|||||||
Reference in new issue
Block a user