`6b46cf2`

perf: cap disk reads at 2000 messages + run async off sync pool

Authored by

espadonne 1 month ago

SHA: 6b46cf28b1d371dc4e6fdd50a4003f9e34aa0ce6
Parents: 6c20a65
Tree: 5deb3b2

2 changed files

Status	File	+	-
M	`src-tauri/src/commands.rs`	34	8
M	`src-tauri/src/core/reader.rs`	119	21

src-tauri/src/commands.rsmodified

      resize_pty as core_resize_pty, spawn_pty as core_spawn_pty, write_pty as core_write_pty,
      PtyHandle, PtyRequest, RingBuffer,
  };
 -use crate::core::reader::read_session as core_read_session;
 +use crate::core::reader::read_session_limited as core_read_session_limited;
  use crate::core::schema::{
      Message, PermissionMode, Project, SessionDetail, SessionSource, SessionSummary,
  };
      Ok(sessions)
+ }
 +/// Default tail cap for disk session reads. Bounded so a 100 MB+
 +/// session file can't hang the main thread during IPC serialize /
 +/// deserialize. The frontend can opt in to the full transcript by
 +/// passing an explicit `messageLimit` (future API) once we have
 +/// lazy-scroll in the viewer.
 +const DEFAULT_SESSION_MESSAGE_LIMIT: usize = 2000;
++
  /// Full viewer payload for one session. Dispatches on `source`:
  ///
  /// - `Disk` (the default): `encoded_dir` is the physical source-dir
  ///   `projectId` field on an archive SessionSummary), and we
  ///   synthesize a prompt-only [`SessionDetail`] from the in-memory
  ///   `history.jsonl` log.
 +///
 +/// **Async** so the file read + JSONL parse + serialize-to-IPC
 +/// pipeline runs on the tokio pool rather than Tauri's sync
 +/// command pool; this prevents a slow 170 MB read from blocking
 +/// other commands (`list_ptys`, `write_pty`, etc.) while it runs.
  #[tauri::command]
 -pub fn read_session(
 +pub async fn read_session(
      encoded_dir: String,
      session_id: String,
      #[allow(non_snake_case)] source: Option<SessionSource>,
      state: State<'_, AppState>,
  ) -> IpcResult<SessionDetail> {
      match source.unwrap_or(SessionSource::Disk) {
 -        SessionSource::Disk => read_disk_session(&encoded_dir, &session_id, &state),
 +        SessionSource::Disk => {
 +            let projects_root = state.projects_root.clone();
 +            // Move the blocking file read off the async runtime.
 +            let encoded_dir_owned = encoded_dir;
 +            let session_id_owned = session_id;
 +            tauri::async_runtime::spawn_blocking(move || {
 +                read_disk_session_blocking(
 +                    &projects_root,
 +                    &encoded_dir_owned,
 +                    &session_id_owned,
 +                )
 +            })
 +            .await
 +            .map_err(|e| format!("read_session join error: {e}"))?
 +        }
          SessionSource::Archive => read_archive_session(&encoded_dir, &state),
+     }
+ }
 -fn read_disk_session(
 +fn read_disk_session_blocking(
 +    projects_root: &Path,
      encoded_dir: &str,
      session_id: &str,
 -    state: &AppState,
  ) -> IpcResult<SessionDetail> {
 -    let path = state
 -        .projects_root
 +    let path = projects_root
          .join(encoded_dir)
          .join(format!("{session_id}.jsonl"));
      if !path.exists() {
          return Err(format!("session not found: {encoded_dir}/{session_id}"));
+     }
 -    core_read_session(&path, encoded_dir).map_err(|e| e.to_string())
 +    core_read_session_limited(&path, encoded_dir, Some(DEFAULT_SESSION_MESSAGE_LIMIT))
 +        .map_err(|e| e.to_string())
+ }
  fn read_archive_session(project_path: &str, state: &AppState) -> IpcResult<SessionDetail> {

src-tauri/src/core/reader.rsmodified

  use crate::core::schema::{ContentBlock, Message, RawEvent, SessionDetail, Usage};
  pub fn read_session(path: &Path, project_id: &str) -> CoreResult<SessionDetail> {
 +    read_session_limited(path, project_id, None)
 +}
++
 +/// Read a session, optionally capping the result to the most recent
 +/// `limit` messages. Required for host performance on very large
 +/// sessions — one user's armfortas session is 171 MB of JSONL and
 +/// that's enough to hang the main thread for seconds during IPC
 +/// deserialization. Passing `Some(N)` streams the whole file but
 +/// only retains the tail, so the wire payload stays bounded.
 +///
 +/// The returned [`SessionDetail`] is identical in shape regardless
 +/// of whether `limit` was set; truncation is invisible to the
 +/// frontend except that `messages.len() < summary.message_count`.
 +pub fn read_session_limited(
 +    path: &Path,
 +    project_id: &str,
 +    limit: Option<usize>,
 +) -> CoreResult<SessionDetail> {
      let summary = summarize(path, project_id)?;
 -    let messages = read_messages(path)?;
 +    let messages = read_messages(path, limit)?;
      Ok(SessionDetail { summary, messages })
+ }
 -fn read_messages(path: &Path) -> CoreResult<Vec<Message>> {
 +fn read_messages(path: &Path, limit: Option<usize>) -> CoreResult<Vec<Message>> {
      let file = File::open(path)?;
      let reader = BufReader::new(file);
 -    let mut out = Vec::new();
      let mut fallback_counter: u32 = 0;
 -    for line in reader.lines() {
 -        let line = match line {
 -            Ok(l) => l,
 -            Err(_) => continue,
 -        };
 -        if line.is_empty() {
 -            continue;
 -        }
 -        let ev: RawEvent = match serde_json::from_str(&line) {
 -            Ok(e) => e,
 -            Err(_) => continue,
 -        };
 -        if ev.is_sidechain.unwrap_or(false) {
 -            continue;
 +    // When a cap is set, keep a rolling tail so we never allocate a
 +    // `Vec<Message>` bigger than the cap + 1. This keeps memory
 +    // bounded on multi-hundred-megabyte files.
 +    match limit {
 +        Some(cap) if cap > 0 => {
 +            let mut tail: std::collections::VecDeque<Message> =
 +                std::collections::VecDeque::with_capacity(cap);
 +            for line in reader.lines() {
 +                let line = match line {
 +                    Ok(l) => l,
 +                    Err(_) => continue,
 +                };
 +                if line.is_empty() {
 +                    continue;
 +                }
 +                let ev: RawEvent = match serde_json::from_str(&line) {
 +                    Ok(e) => e,
 +                    Err(_) => continue,
 +                };
 +                if ev.is_sidechain.unwrap_or(false) {
 +                    continue;
 +                }
 +                if let Some(msg) = raw_to_message(ev, &mut fallback_counter) {
 +                    if tail.len() == cap {
 +                        tail.pop_front();
 +                    }
 +                    tail.push_back(msg);
 +                }
 +            }
 +            Ok(tail.into_iter().collect())
+         }
 -        if let Some(msg) = raw_to_message(ev, &mut fallback_counter) {
 -            out.push(msg);
 +        _ => {
 +            let mut out = Vec::new();
 +            for line in reader.lines() {
 +                let line = match line {
 +                    Ok(l) => l,
 +                    Err(_) => continue,
 +                };
 +                if line.is_empty() {
 +                    continue;
 +                }
 +                let ev: RawEvent = match serde_json::from_str(&line) {
 +                    Ok(e) => e,
 +                    Err(_) => continue,
 +                };
 +                if ev.is_sidechain.unwrap_or(false) {
 +                    continue;
 +                }
 +                if let Some(msg) = raw_to_message(ev, &mut fallback_counter) {
 +                    out.push(msg);
 +                }
 +            }
 +            Ok(out)
+         }
+     }
+-
 -    Ok(out)
+ }
  /// Convert one raw event into a timeline message. Returns `None` for
          assert_eq!(text, "main");
+     }
 +    #[test]
 +    fn limit_returns_tail_and_preserves_order() {
 +        let tmp = tempdir().unwrap();
 +        let path = tmp.path().join("s.jsonl");
 +        let mut lines: Vec<String> = Vec::new();
 +        for i in 0..10 {
 +            lines.push(format!(
 +                "{{\"type\":\"user\",\"uuid\":\"u{i}\",\"timestamp\":\"2026-04-11T00:55:{:02}.000Z\",\"sessionId\":\"abc\",\"message\":{{\"role\":\"user\",\"content\":\"msg{i}\"}}}}",
 +                i
 +            ));
 +        }
 +        write_fixture(
 +            &path,
 +            &lines.iter().map(|s| s.as_str()).collect::<Vec<_>>(),
 +        );
++
 +        let detail = read_session_limited(&path, "-Users-me-repo", Some(3)).unwrap();
 +        assert_eq!(detail.messages.len(), 3);
 +        // Summary still reflects total message count.
 +        assert_eq!(detail.summary.message_count, 10);
 +        let texts: Vec<&str> = detail
 +            .messages
 +            .iter()
 +            .filter_map(|m| match m {
 +                Message::User { text, .. } => Some(text.as_str()),
 +                _ => None,
 +            })
 +            .collect();
 +        assert_eq!(texts, vec!["msg7", "msg8", "msg9"]);
 +    }
++
 +    #[test]
 +    fn limit_zero_is_treated_as_unlimited() {
 +        // 0 is a weird edge case; we treat it as "no cap" so a
 +        // caller mishandling its limit arg still gets data.
 +        let tmp = tempdir().unwrap();
 +        let path = tmp.path().join("s.jsonl");
 +        write_fixture(
 +            &path,
 +            &[
 +                r#"{"type":"user","uuid":"u1","timestamp":"2026-04-11T00:55:35.000Z","sessionId":"abc","message":{"role":"user","content":"hi"}}"#,
 +            ],
 +        );
 +        let detail = read_session_limited(&path, "-Users-me-repo", Some(0)).unwrap();
 +        assert_eq!(detail.messages.len(), 1);
 +    }
++
      #[test]
      fn survives_partial_last_line() {
          let tmp = tempdir().unwrap();