From 01cf669b9f759671a291d4f89946ba9c83fb350e Mon Sep 17 00:00:00 2001 From: Junyi Ou Date: Tue, 20 Oct 2026 23:15:52 -0400 Subject: [PATCH 1/2] feat(dgw): page recording search with a cursor Searching every recording doesn't fit in one response, so `POST /jet/jrec/search` now returns pages. A page ends after reading about 16 MiB of logs or after `maxHits` hits, from 1 to 1000 and 1000 by default; 0 or more than 1000 is a 400. `nextCursor` replaces `limitReached`: sending the same `query`, `recordings` and `startTime` with it as `cursor` gets the next page, until a page has no `nextCursor`. Gateway keeps no state between pages. The cursor holds the recording, file, byte offset and line number to resume at, plus a hash of the search, so a cursor sent with another search is a 400. A cursor whose position cannot exist is a 400 too, and its file name is only matched against the recording's manifest, never joined into a path. Pages never go back, so a recording that starts behind the cursor while paging isn't searched. The paging rules are in `search.intent.md`. Issue: DGW-463 Co-Authored-By: Claude Opus 5.5 --- Cargo.lock | 1 + .../src/api/jrec/search.intent.md | 15 +- devolutions-gateway/src/api/jrec/search.rs | 455 ++++++++++++++---- testsuite/Cargo.toml | 1 + testsuite/tests/cli/dgw/recording_search.rs | 235 ++++++++- 5 files changed, 611 insertions(+), 96 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 7c9a28980..e83edf1fb 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -7724,6 +7724,7 @@ dependencies = [ "fastrand", "futures-util", "gateway-db", + "hex", "ipnetwork", "jmux-proto", "libsql", diff --git a/devolutions-gateway/src/api/jrec/search.intent.md b/devolutions-gateway/src/api/jrec/search.intent.md index 24bda9e24..91f5d4fb1 100644 --- a/devolutions-gateway/src/api/jrec/search.intent.md +++ b/devolutions-gateway/src/api/jrec/search.intent.md @@ -32,16 +32,25 @@ Search has its own scope because `gateway.recordings.read` only lists recording ## Request `recordings` is `"all"` or a list of recording IDs, and stays open to new forms added without breaking existing callers. -`"all"` orders recordings by start time, latest first, then by ID, the largest UUID as a 128-bit number first, so that the order is the same on every request. +`"all"` orders recordings by start time, latest first, then by ID, the largest UUID as a 128-bit number first, so that the order is the same on every request and every page. A list keeps its order and searches a repeated ID once. A request with a field Gateway does not know is rejected, as `rg` and `grep` reject an unknown flag, so that an option added later is never silently ignored by an older Gateway. The optional `startTime` range takes RFC 3339 date-times and keeps a recording when `from <= startTime < to`, using the start time in `recording.json`, never anything inside a log. A missing end leaves that side of the range open. -## Limits +## Pages -One search reads about 16 MiB of logs at most and returns at most 1,000 hits, so that searching every recording takes a bounded time. +Results come in pages, so that searching every recording takes a bounded time per request. +A page ends after reading about 16 MiB of logs or after `maxHits` hits, whichever comes first. +`maxHits` is from 1 to 1,000 and defaults to 1,000; a value outside that range is rejected, never clamped. +A page carries a `nextCursor` until every log is read, and a page with fewer hits than `maxHits` may still have one, since the read limit can end it first. + +Gateway keeps no state between pages; the cursor holds where the next page starts. +A cursor is accepted only with the same `query`, `recordings` and `startTime` as the page that gave it. +The hit that does not fit in a page opens the next one, so that every page moves the search forward and returns each hit once. +Pages never go back: a recording that starts, or a log that grows, behind the cursor while paging is not searched. +A log replaced while paging, as an AI analysis rerun does, is read from the cursor's byte offset on, so lines can be missed or misnumbered; this is accepted for now. ## Later diff --git a/devolutions-gateway/src/api/jrec/search.rs b/devolutions-gateway/src/api/jrec/search.rs index 914723674..afb7c7dca 100644 --- a/devolutions-gateway/src/api/jrec/search.rs +++ b/devolutions-gateway/src/api/jrec/search.rs @@ -1,12 +1,15 @@ +use std::cmp::Reverse; use std::collections::HashSet; -use std::io; +use std::io::{self, SeekFrom}; +use std::mem; use axum::Json; use axum::extract::State; use camino::{Utf8Path, Utf8PathBuf}; +use sha2::{Digest as _, Sha256}; use time::OffsetDateTime; use tokio::fs::File; -use tokio::io::{AsyncBufReadExt as _, BufReader}; +use tokio::io::{AsyncBufReadExt as _, AsyncSeekExt as _, BufReader}; use uuid::Uuid; use super::is_safe_recording_file_name; @@ -16,12 +19,12 @@ use crate::extract::RecordingsSearchScope; use crate::http::HttpError; use crate::token::RecordingFileType; -/// Most hits returned by one search. +/// Most hits returned by one page, and the page size when the request does not set one. const MAX_HITS: usize = 1_000; -/// Log bytes one search reads before it stops, so that searching every recording takes a bounded time. +/// Log bytes one page reads before it ends, so that every page takes a bounded time. /// -/// A search reads past it by about [`MAX_LINE_BYTES`] at most, to finish reading the start of a line. +/// A page reads past it by about [`MAX_LINE_BYTES`] at most, to finish reading the start of a line. const MAX_SCANNED_BYTES: u64 = 16 * 1024 * 1024; /// Bytes of a line that are matched and returned; the rest of a longer line is skipped. @@ -42,6 +45,14 @@ pub(crate) struct RecordingSearchRequest { /// Only search the recordings that started in this range. #[serde(default)] start_time: Option, + /// Most hits in the page, from 1 to 1000; 1000 when missing. + #[serde(default)] + #[cfg_attr(feature = "openapi", schema(minimum = 1, maximum = 1000))] + max_hits: Option, + /// Where the page starts: the `nextCursor` of the previous page, sent unchanged with the same `query`, + /// `recordings` and `startTime`. + #[serde(default)] + cursor: Option, } /// `"all"`, or a list of recording IDs. @@ -122,8 +133,12 @@ impl StartTimeRange { pub(crate) struct RecordingSearchResponse { /// Matching lines, newest recording first when searching all recordings, otherwise in the requested order. hits: Vec, - /// The search stopped before reading every log, so more lines may match. - limit_reached: bool, + /// Where the next page starts; absent when every log has been read. + /// + /// The cursor is opaque, and its format may change between versions. + /// A page with fewer hits than asked for may still be followed by more pages. + #[serde(skip_serializing_if = "Option::is_none")] + next_cursor: Option, } /// A line that matches the search text. @@ -156,6 +171,12 @@ pub(crate) struct RecordingSearchHit { /// matched as raw text. /// Only the files listed under `artifacts` in `recording.json` with the `slog` type are searched. /// Recordings without a readable `recording.json` and IDs with no recording on this instance give no hits. +/// +/// Results come in pages. +/// To get the next page, send the same `query`, `recordings` and `startTime` again, +/// with the `nextCursor` of the previous page as `cursor`. +/// Pages never go back, so a recording that starts, or a log that grows, behind the cursor while paging is not +/// searched. #[cfg_attr(feature = "openapi", utoipa::path( post, operation_id = "SearchRecordings", @@ -176,23 +197,82 @@ pub(crate) async fn search_recordings( Json(request): Json, ) -> Result, HttpError> { let text = SearchText::new(&request.query); + let max_hits = page_size(request.max_hits).ok_or_else(|| HttpError::bad_request().msg("invalid maxHits"))?; + let invalid_cursor = || HttpError::bad_request().msg("invalid cursor for this search"); + + // A cursor is only followed in the search it came from, which these fields identify. + let search_id = { + let listed_ids = match &request.recordings { + RecordingSelection::All(AllRecordings::All) => None, + RecordingSelection::Listed(recording_ids) => Some(recording_ids), + }; + let start_time = request.start_time.as_ref().map(|range| { + ( + range.from.map(OffsetDateTime::unix_timestamp_nanos), + range.to.map(OffsetDateTime::unix_timestamp_nanos), + ) + }); - let recording_root = conf_handle.get_conf().recording_path.clone(); + let search = + serde_json::to_vec(&(&request.query, listed_ids, start_time)).expect("a search serializes to JSON"); + hex::encode(&Sha256::digest(search)[..16]) + }; - let mut recordings = match request.recordings { - RecordingSelection::All(AllRecordings::All) => all_recording_logs_newest_first(&recording_root) - .await - .map_err(HttpError::internal().with_msg("failed recording listing").err())?, + let mut cursor = request + .cursor + .map(|cursor| { + Cursor::decode(&cursor) + .filter(|cursor| cursor.search_id == search_id) + .ok_or_else(invalid_cursor) + }) + .transpose()?; + + let listed_ids = match request.recordings { + RecordingSelection::All(AllRecordings::All) => None, RecordingSelection::Listed(recording_ids) => { let mut seen = HashSet::new(); - let mut recordings = Vec::with_capacity(recording_ids.len()); - for recording_id in recording_ids - .into_iter() - .filter(|recording_id| seen.insert(*recording_id)) - { + Some( + recording_ids + .into_iter() + .filter(|recording_id| seen.insert(*recording_id)) + .collect::>(), + ) + } + }; + + let recording_root = conf_handle.get_conf().recording_path.clone(); + + let mut recordings = match listed_ids { + None => { + let mut recordings = all_recording_logs_newest_first(&recording_root) + .await + .map_err(HttpError::internal().with_msg("failed recording listing").err())?; + + if let Some(cursor) = &cursor { + let cursor_key = newest_first(cursor.start_time, cursor.recording_id); + let resumed = recordings.partition_point(|recording| { + newest_first(recording.start_time, recording.recording_id) < cursor_key + }); + recordings.drain(..resumed); + } + + recordings + } + Some(recording_ids) => { + let resumed = match &cursor { + Some(cursor) => recording_ids + .iter() + .position(|recording_id| *recording_id == cursor.recording_id) + .ok_or_else(invalid_cursor)?, + None => 0, + }; + + let mut recordings = Vec::with_capacity(recording_ids.len() - resumed); + for recording_id in &recording_ids[resumed..] { let recording_dir = recording_root.join(recording_id.to_string()); - recordings.extend(read_recording_logs(recording_dir, recording_id).await); + recordings.extend(read_recording_logs(recording_dir, *recording_id).await); } + recordings } }; @@ -201,28 +281,110 @@ pub(crate) async fn search_recordings( recordings.retain(|recording| start_time.contains(recording.start_time)); } - let mut search = Search::new(text); + let mut search = Search::new(text, max_hits); for recording in recordings { - for file_name in &recording.file_names { + let (first_file, mut position) = match cursor.take() { + Some(cursor) if cursor.recording_id == recording.recording_id => { + // Searching the recording again from its first log would return hits twice. + let Some(index) = recording + .file_names + .iter() + .position(|file_name| *file_name == cursor.file_name) + else { + continue; + }; + + (index, cursor.position) + } + _ => (0, FilePosition::default()), + }; + + for file_name in &recording.file_names[first_file..] { let path = recording.dir.join(file_name); - match search.search_file(recording.recording_id, file_name, &path).await { - Ok(()) => {} + match search + .search_file(recording.recording_id, file_name, &path, mem::take(&mut position)) + .await + { + Ok(None) => {} + Ok(Some(position)) => { + let next_cursor = Cursor { + search_id, + recording_id: recording.recording_id, + start_time: recording.start_time, + file_name: file_name.clone(), + position, + }; + + return Ok(Json(search.into_response(Some(next_cursor.encode())))); + } // A recording can be deleted while it is searched. Err(error) if error.kind() == io::ErrorKind::NotFound => { trace!(%path, "Recording log listed in the manifest is missing"); } Err(error) => warn!(%path, %error, "Failed to search a recording log"), } - - if search.limit_reached { - return Ok(Json(search.into_response())); - } } } - Ok(Json(search.into_response())) + Ok(Json(search.into_response(None))) +} + +/// The number of hits a page holds at most, or `None` when `max_hits` is out of range. +fn page_size(max_hits: Option) -> Option { + let Some(max_hits) = max_hits else { + return Some(MAX_HITS); + }; + + let max_hits = usize::try_from(max_hits).ok()?; + (1..=MAX_HITS).contains(&max_hits).then_some(max_hits) +} + +/// Where the next page of a search starts. +/// +/// The caller gets it as an opaque string: hex-encoded JSON. +/// A cursor cannot reach a file the search would not read, because only its recording ID and file name are matched +/// against the searched recordings and their manifests, and neither is ever joined into a path. +#[derive(Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +struct Cursor { + search_id: String, + recording_id: Uuid, + start_time: i64, + file_name: String, + position: FilePosition, +} + +impl Cursor { + fn encode(&self) -> String { + let json = serde_json::to_vec(self).expect("a cursor serializes to JSON"); + hex::encode(json) + } + + fn decode(cursor: &str) -> Option { + let json = hex::decode(cursor).ok()?; + let cursor: Self = serde_json::from_slice(&json).ok()?; + + // A caller can edit a cursor, and reading on from a position past these bounds would overflow. + // Every line takes at least one byte, so a cursor Gateway made never counts more lines than bytes. + let FilePosition { + offset, line_number, .. + } = cursor.position; + (line_number <= offset && i64::try_from(offset).is_ok()).then_some(cursor) + } +} + +/// Where reading continues in one log. +#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +struct FilePosition { + /// Bytes from the start of the file. + offset: u64, + /// Lines whose start was read before `offset`, so that line numbers go on counting from there. + line_number: u64, + /// The position is inside a line longer than [`MAX_LINE_BYTES`], whose rest is skipped up to its line break. + in_long_line: bool, } /// The query, lowercased the same way as the text it is compared with. @@ -522,78 +684,91 @@ impl<'a> JsonCursor<'a> { } } +/// One page of a search. struct Search { text: SearchText, + max_hits: usize, hits: Vec, scanned_bytes: u64, - limit_reached: bool, } impl Search { - fn new(text: SearchText) -> Self { + fn new(text: SearchText, max_hits: usize) -> Self { Self { text, + max_hits, hits: Vec::new(), scanned_bytes: 0, - limit_reached: false, } } - fn into_response(self) -> RecordingSearchResponse { + fn into_response(self, next_cursor: Option) -> RecordingSearchResponse { RecordingSearchResponse { hits: self.hits, - limit_reached: self.limit_reached, + next_cursor, } } - /// Adds the lines of one log that match the search text. - async fn search_file(&mut self, recording_id: Uuid, file_name: &str, path: &Utf8Path) -> io::Result<()> { - let mut reader = BufReader::new(File::open(path).await?); + /// Adds the lines of one log that match the search text, reading from `position` on. + /// + /// Returns where the next page continues when the page ends in this log, and `None` when the log is done. + async fn search_file( + &mut self, + recording_id: Uuid, + file_name: &str, + path: &Utf8Path, + mut position: FilePosition, + ) -> io::Result> { + let mut file = File::open(path).await?; + file.seek(SeekFrom::Start(position.offset)).await?; + + let mut reader = BufReader::new(file); let mut line = Vec::new(); - let mut line_number = 0; - let mut offset = 0; loop { + if position.in_long_line { + let budget = MAX_SCANNED_BYTES.saturating_sub(self.scanned_bytes); + let skipped = skip_rest_of_line(&mut reader, budget).await?; + self.scanned_bytes += skipped.bytes; + position.offset += skipped.bytes; + + if !skipped.finished { + return Ok(Some(position)); + } + + position.in_long_line = false; + } + + let here = position; + if MAX_SCANNED_BYTES <= self.scanned_bytes { - self.limit_reached = !reader.fill_buf().await?.is_empty(); - return Ok(()); + let at_end = reader.fill_buf().await?.is_empty(); + return Ok((!at_end).then_some(here)); } let Some(read) = read_line(&mut reader, &mut line).await? else { - return Ok(()); + return Ok(None); }; self.scanned_bytes += read.bytes; - line_number += 1; - let line_offset = offset; - offset += read.bytes; + position.line_number += 1; + position.offset += read.bytes; + position.in_long_line = read.rest_unread; if self.text.is_in(&line) { - if MAX_HITS <= self.hits.len() { - self.limit_reached = true; - return Ok(()); + // The hit that does not fit opens the next page, which always keeps its first hit. + if self.max_hits <= self.hits.len() { + return Ok(Some(here)); } self.hits.push(RecordingSearchHit { recording_id, file_name: file_name.to_owned(), - line_number, - offset: line_offset, + line_number: position.line_number, + offset: here.offset, line: String::from_utf8_lossy(&line).into_owned(), truncated: read.truncated, }); } - - if read.rest_unread { - let budget = MAX_SCANNED_BYTES.saturating_sub(self.scanned_bytes); - let skipped = skip_rest_of_line(&mut reader, budget).await?; - self.scanned_bytes += skipped.bytes; - offset += skipped.bytes; - - if !skipped.finished { - self.limit_reached = true; - return Ok(()); - } - } } } } @@ -686,12 +861,19 @@ async fn all_recording_logs_newest_first(recording_root: &Utf8Path) -> anyhow::R recordings.extend(read_recording_logs(dir, recording_id).await); } - // Ordering by ID too keeps the order of recordings that start together the same on every request. - recordings.sort_unstable_by(|a, b| (b.start_time, b.recording_id).cmp(&(a.start_time, a.recording_id))); + recordings.sort_unstable_by_key(|recording| newest_first(recording.start_time, recording.recording_id)); Ok(recordings) } +/// Sort key of the order `"all"` searches recordings in: newest first by start time, then by ID. +/// +/// Ordering by ID too keeps the order of recordings that start together the same on every page, +/// so that a cursor finds its place in the next page's listing. +fn newest_first(start_time: i64, recording_id: Uuid) -> Reverse<(i64, Uuid)> { + Reverse((start_time, recording_id)) +} + struct LineRead { /// Bytes consumed, line break included. bytes: u64, @@ -775,7 +957,7 @@ struct Skipped { } /// Consumes the rest of a line up to its line break, reading at most `budget` bytes, -/// so that a log without line breaks cannot make one search read all of it. +/// so that a log without line breaks cannot make one page read all of it. async fn skip_rest_of_line(reader: &mut BufReader, budget: u64) -> io::Result { let mut bytes = 0; @@ -809,18 +991,32 @@ mod tests { use super::*; - async fn search_log(query: &str, content: &[u8]) -> Search { - let dir = tempfile::tempdir().expect("temp dir"); + fn write_log(dir: &tempfile::TempDir, content: &[u8]) -> Utf8PathBuf { let path = Utf8PathBuf::from_path_buf(dir.path().join("ai-analysis-0.slog")).expect("UTF-8 path"); std::fs::write(&path, content).expect("write log"); - - let mut search = Search::new(SearchText::new(query)); - search - .search_file(Uuid::nil(), "ai-analysis-0.slog", &path) + path + } + + async fn search_page( + query: &str, + max_hits: usize, + path: &Utf8Path, + position: FilePosition, + ) -> (Search, Option) { + let mut search = Search::new(SearchText::new(query), max_hits); + let next = search + .search_file(Uuid::nil(), "ai-analysis-0.slog", path, position) .await .expect("search log"); - search + (search, next) + } + + async fn search_log(query: &str, content: &[u8]) -> Search { + let dir = tempfile::tempdir().expect("temp dir"); + let path = write_log(&dir, content); + + search_page(query, MAX_HITS, &path, FilePosition::default()).await.0 } fn hit_lines(search: &Search) -> Vec<(u64, &str)> { @@ -1038,7 +1234,7 @@ mod tests { } // oracle: intent devolutions-gateway/src/api/jrec/search.intent.md §Hits - // oracle: intent devolutions-gateway/src/api/jrec/search.intent.md §Limits + // oracle: intent devolutions-gateway/src/api/jrec/search.intent.md §Pages #[test] fn limits_match_the_intent() { assert_eq!(MAX_LINE_BYTES, 256 * 1024); @@ -1046,47 +1242,122 @@ mod tests { assert_eq!(MAX_SCANNED_BYTES, 16 * 1024 * 1024); } - // oracle: intent devolutions-gateway/src/api/jrec/search.intent.md §Limits + // oracle: intent devolutions-gateway/src/api/jrec/search.intent.md §Pages #[tokio::test] - async fn stops_at_the_hit_limit() { - let content = "hit\n".repeat(MAX_HITS + 1); - - let search = search_log("hit", content.as_bytes()).await; - - assert_eq!(search.hits.len(), MAX_HITS); - assert!(search.limit_reached); - } - - // oracle: intent devolutions-gateway/src/api/jrec/search.intent.md §Limits - #[tokio::test] - async fn stops_inside_a_line_without_breaks_at_the_read_budget() { + async fn continues_a_line_without_breaks_on_the_next_page() { let budget = usize::try_from(MAX_SCANNED_BYTES).expect("budget fits in usize"); let mut content = b"hit\n".to_vec(); content.resize(content.len() + budget + 1024 * 1024, b'x'); content.extend_from_slice(b"\nhit\n"); + let dir = tempfile::tempdir().expect("temp dir"); + let path = write_log(&dir, &content); - let search = search_log("hit", &content).await; + let (first, next) = search_page("hit", MAX_HITS, &path, FilePosition::default()).await; + let (second, last) = search_page("hit", MAX_HITS, &path, next.expect("a second page")).await; - assert_eq!(hit_lines(&search), [(1, "hit")]); - assert!(search.limit_reached); - assert!(search.scanned_bytes <= MAX_SCANNED_BYTES); + assert!(first.scanned_bytes <= MAX_SCANNED_BYTES); + assert_eq!(hit_lines(&first), [(1, "hit")]); + assert_eq!(hit_lines(&second), [(3, "hit")]); + assert_eq!(last, None); } - // oracle: mutant limitReached set when the read budget runs out exactly where the log ends + // oracle: intent devolutions-gateway/src/api/jrec/search.intent.md §Pages + #[rstest] + #[case::missing(None, Some(MAX_HITS))] + #[case::one(Some(1), Some(1))] + #[case::most(Some(1_000), Some(1_000))] + #[case::zero(Some(0), None)] + #[case::over_the_most(Some(1_001), None)] + fn accepts_a_page_size_from_1_to_1000(#[case] max_hits: Option, #[case] expected: Option) { + assert_eq!(page_size(max_hits), expected); + } + + // oracle: intent devolutions-gateway/src/api/jrec/search.intent.md §Pages #[rstest] #[case::with_a_final_line_break(b"\n".as_slice())] #[case::without_one(b"".as_slice())] #[tokio::test] - async fn reaches_no_limit_when_the_read_budget_ends_with_the_log(#[case] end: &[u8]) { + async fn gives_no_next_page_when_the_read_budget_ends_with_the_log(#[case] end: &[u8]) { let budget = usize::try_from(MAX_SCANNED_BYTES).expect("budget fits in usize"); let mut content = b"hit\n".to_vec(); content.resize(budget - end.len(), b'x'); content.extend_from_slice(end); + let dir = tempfile::tempdir().expect("temp dir"); + let path = write_log(&dir, &content); - let search = search_log("hit", &content).await; + let (search, next) = search_page("hit", MAX_HITS, &path, FilePosition::default()).await; assert_eq!(hit_lines(&search), [(1, "hit")]); - assert!(!search.limit_reached); + assert_eq!(next, None); + } + + proptest::proptest! { + // oracle: intent devolutions-gateway/src/api/jrec/search.intent.md §Pages + #[test] + fn returns_each_hit_once_across_pages( + lines in proptest::collection::vec("[a-c ]{0,4}(hit)?[a-c ]{0,4}", 0..30), + max_hits in 1..=5usize, + ) { + let content = lines.join("\n"); + let dir = tempfile::tempdir().expect("temp dir"); + let path = write_log(&dir, content.as_bytes()); + + let mut expected = Vec::new(); + let mut offset = 0; + for (line, line_number) in lines.iter().zip(1..) { + if line.contains("hit") { + expected.push((line_number, offset, line.clone())); + } + offset += u64::try_from(line.len() + 1).expect("line length fits in u64"); + } + + let runtime = tokio::runtime::Builder::new_current_thread().build().expect("runtime"); + let mut hits = Vec::new(); + let mut pages = 0; + let mut position = Some(FilePosition::default()); + while let Some(start) = position { + pages += 1; + proptest::prop_assert!(pages <= expected.len() + 1, "paging does not end"); + + let (page, next) = runtime.block_on(search_page("hit", max_hits, &path, start)); + proptest::prop_assert!(page.hits.len() <= max_hits); + hits.extend(page.hits.into_iter().map(|hit| (hit.line_number, hit.offset, hit.line))); + position = next; + } + + proptest::prop_assert_eq!(hits, expected); + } + } + + fn cursor_at(offset: u64, line_number: u64) -> Cursor { + Cursor { + search_id: "0".repeat(32), + recording_id: Uuid::nil(), + start_time: 100, + file_name: "ai-analysis-0.slog".to_owned(), + position: FilePosition { + offset, + line_number, + in_long_line: false, + }, + } + } + + // oracle: intent devolutions-gateway/src/api/jrec/search.intent.md §Pages + #[rstest] + #[case::start(0, 0, true)] + #[case::one_byte_per_line(10, 10, true)] + #[case::more_lines_than_bytes(10, 11, false)] + #[case::last_seek_position(i64::MAX.unsigned_abs(), 0, true)] + #[case::past_the_last_seek_position(i64::MAX.unsigned_abs() + 1, 0, false)] + fn accepts_only_cursors_whose_position_can_exist( + #[case] offset: u64, + #[case] line_number: u64, + #[case] accepted: bool, + ) { + let cursor = cursor_at(offset, line_number).encode(); + + assert_eq!(Cursor::decode(&cursor).is_some(), accepted); } // oracle: intent devolutions-gateway/src/api/jrec/search.intent.md §What is searched diff --git a/testsuite/Cargo.toml b/testsuite/Cargo.toml index aa8fa4fc7..2204d5d18 100644 --- a/testsuite/Cargo.toml +++ b/testsuite/Cargo.toml @@ -41,6 +41,7 @@ devolutions-gateway = { path = "../devolutions-gateway" } futures-util = "0.3" agent-tunnel-libsql.path = "../crates/agent-tunnel-libsql" gateway-db.path = "../crates/gateway-db" +hex = "0.4" ipnetwork = "0.20" jmux-proto = { path = "../crates/jmux-proto" } libsql = { version = "0.10", default-features = false, features = ["core"] } diff --git a/testsuite/tests/cli/dgw/recording_search.rs b/testsuite/tests/cli/dgw/recording_search.rs index 7355e692c..f624e503c 100644 --- a/testsuite/tests/cli/dgw/recording_search.rs +++ b/testsuite/tests/cli/dgw/recording_search.rs @@ -4,6 +4,7 @@ use std::path::Path; use anyhow::Context as _; use base64::Engine as _; +use rstest::rstest; use testsuite::cli::start_dgw; use testsuite::dgw_config::{DgwConfig, DgwConfigHandle}; use uuid::Uuid; @@ -147,6 +148,39 @@ async fn search(http_port: u16, scope: &str, request: serde_json::Value) -> anyh .context("send search request") } +/// Follows `nextCursor` from the first page to the last, and returns every page. +async fn search_every_page(http_port: u16, mut request: serde_json::Value) -> anyhow::Result> { + let mut pages = Vec::new(); + + loop { + let response = search(http_port, "gateway.recordings.search", request.clone()).await?; + anyhow::ensure!( + response.status() == reqwest::StatusCode::OK, + "status {}", + response.status() + ); + + let page: serde_json::Value = response.json().await?; + let next_cursor = page.get("nextCursor").cloned(); + pages.push(page); + + let Some(next_cursor) = next_cursor else { + return Ok(pages); + }; + anyhow::ensure!(pages.len() < 100, "paging does not end"); + request["cursor"] = next_cursor; + } +} + +/// `(recordingId, fileName, lineNumber)` of each hit of every page, in order. +fn hit_summaries_of(pages: &[serde_json::Value]) -> anyhow::Result> { + Ok(pages + .iter() + .map(hit_summaries) + .collect::>>()? + .concat()) +} + /// `(recordingId, fileName, lineNumber)` of each hit. fn hit_summaries(body: &serde_json::Value) -> anyhow::Result> { body["hits"] @@ -185,7 +219,7 @@ async fn searches_the_slog_artifacts_of_every_recording_newest_first() -> anyhow (OLDER_RECORDING.to_string(), "ai-analysis-0.slog".to_owned(), 2), ] ); - assert_eq!(body["limitReached"], false); + assert!(body.get("nextCursor").is_none()); Ok(()) } @@ -246,6 +280,205 @@ async fn searches_only_the_recordings_started_in_the_time_range() -> anyhow::Res Ok(()) } +// oracle: intent devolutions-gateway/src/api/jrec/search.intent.md §Pages +// oracle: intent devolutions-gateway/src/api/jrec/search.intent.md §Request +#[tokio::test] +async fn pages_through_every_hit_once() -> anyhow::Result<()> { + let recordings = tempfile::tempdir()?; + let log = "Created User\n".repeat(600); + write_analysis_recording(recordings.path(), OLDER_RECORDING, 100, &log)?; + write_analysis_recording(recordings.path(), NEWER_RECORDING, 200, &log)?; + let (_recordings, config_handle, _process) = start_on(recordings).await?; + + let pages = search_every_page( + config_handle.http_port(), + serde_json::json!({ "query": "created user", "recordings": "all" }), + ) + .await?; + + let page_sizes: Vec<_> = pages + .iter() + .map(|page| page["hits"].as_array().map_or(0, Vec::len)) + .collect(); + assert_eq!(page_sizes, [1_000, 200]); + + let expected: Vec<_> = [NEWER_RECORDING, OLDER_RECORDING] + .into_iter() + .flat_map(|id| (1..=600).map(move |line_number| (id.to_string(), "ai-analysis-0.slog".to_owned(), line_number))) + .collect(); + assert_eq!(hit_summaries_of(&pages)?, expected); + + Ok(()) +} + +// oracle: intent devolutions-gateway/src/api/jrec/search.intent.md §Request +// oracle: intent devolutions-gateway/src/api/jrec/search.intent.md §Pages +#[tokio::test] +async fn pages_through_recordings_that_start_together_highest_id_first() -> anyhow::Result<()> { + let recordings = tempfile::tempdir()?; + for id in 1..=3 { + write_analysis_recording(recordings.path(), Uuid::from_u128(id), 100, "Created User")?; + } + let (_recordings, config_handle, _process) = start_on(recordings).await?; + + let pages = search_every_page( + config_handle.http_port(), + serde_json::json!({ "query": "created user", "recordings": "all", "maxHits": 1 }), + ) + .await?; + + assert_eq!(pages.len(), 3); + assert_eq!( + hit_summaries_of(&pages)?, + [3u128, 2, 1].map(|id| (Uuid::from_u128(id).to_string(), "ai-analysis-0.slog".to_owned(), 1)) + ); + + Ok(()) +} + +// oracle: intent devolutions-gateway/src/api/jrec/search.intent.md §Request +// oracle: intent devolutions-gateway/src/api/jrec/search.intent.md §Pages +#[tokio::test] +async fn pages_through_listed_recordings_in_their_order() -> anyhow::Result<()> { + let (_recordings, config_handle, _process) = start_with_recordings().await?; + + let pages = search_every_page( + config_handle.http_port(), + serde_json::json!({ + "query": "created user", + "recordings": [OLDER_RECORDING.to_string(), NEWER_RECORDING.to_string()], + "maxHits": 1, + }), + ) + .await?; + + assert_eq!(pages.len(), 2); + assert_eq!( + hit_summaries_of(&pages)?, + [ + (OLDER_RECORDING.to_string(), "ai-analysis-0.slog".to_owned(), 2), + (NEWER_RECORDING.to_string(), "ai-analysis-0.slog".to_owned(), 1), + ] + ); + + Ok(()) +} + +#[rstest] +#[case::other_query(serde_json::json!({ "query": "opened users", "recordings": "all" }))] +#[case::query_in_another_case(serde_json::json!({ "query": "Created User", "recordings": "all" }))] +#[case::other_recordings(serde_json::json!({ + "query": "created user", + "recordings": [NEWER_RECORDING.to_string(), OLDER_RECORDING.to_string()], +}))] +#[case::other_start_time(serde_json::json!({ +// oracle: intent devolutions-gateway/src/api/jrec/search.intent.md §Pages + "query": "created user", + "recordings": "all", + "startTime": { "from": "1970-01-01T00:00:00Z" }, +}))] +#[tokio::test] +async fn rejects_a_cursor_from_another_search(#[case] mut other_search: serde_json::Value) -> anyhow::Result<()> { + let (_recordings, config_handle, _process) = start_with_recordings().await?; + + let response = search( + config_handle.http_port(), + "gateway.recordings.search", + serde_json::json!({ "query": "created user", "recordings": "all", "maxHits": 1 }), + ) + .await?; + let first_page: serde_json::Value = response.json().await?; + other_search["cursor"] = first_page["nextCursor"].clone(); + assert!(other_search["cursor"].is_string()); + + let response = search(config_handle.http_port(), "gateway.recordings.search", other_search).await?; + assert_eq!(response.status(), reqwest::StatusCode::BAD_REQUEST); + + Ok(()) +} + +// oracle: intent devolutions-gateway/src/api/jrec/search.intent.md §Pages +// oracle: intent devolutions-gateway/src/api/jrec/search.intent.md §What is searched +#[tokio::test] +async fn resumes_after_the_cursor_recording_is_deleted() -> anyhow::Result<()> { + let recordings = tempfile::tempdir()?; + for (id, start_time) in [(1, 100), (2, 200), (3, 300)] { + write_analysis_recording(recordings.path(), Uuid::from_u128(id), start_time, "Created User")?; + } + let (recordings, config_handle, _process) = start_on(recordings).await?; + let mut request = serde_json::json!({ "query": "created user", "recordings": "all", "maxHits": 1 }); + + let response = search(config_handle.http_port(), "gateway.recordings.search", request.clone()).await?; + let first_page: serde_json::Value = response.json().await?; + std::fs::remove_dir_all(recordings.path().join(Uuid::from_u128(2).to_string()))?; + request["cursor"] = first_page["nextCursor"].clone(); + let next_pages = search_every_page(config_handle.http_port(), request).await?; + + let ai_analysis = |id| (Uuid::from_u128(id).to_string(), "ai-analysis-0.slog".to_owned(), 1); + assert_eq!(hit_summaries(&first_page)?, [ai_analysis(3)]); + assert_eq!(hit_summaries_of(&next_pages)?, [ai_analysis(1)]); + + Ok(()) +} + +/// Rewrites fields of a cursor, which is hex-encoded JSON, as a caller could. +fn edit_cursor(cursor: &serde_json::Value, edits: serde_json::Value) -> anyhow::Result { + let mut fields: serde_json::Value = serde_json::from_slice(&hex::decode(cursor.as_str().context("cursor")?)?)?; + for (field, value) in edits.as_object().context("edits")? { + fields[field] = value.clone(); + } + + Ok(serde_json::Value::String(hex::encode(serde_json::to_vec(&fields)?))) +} + +// oracle: intent devolutions-gateway/src/api/jrec/search.intent.md §What is searched +#[tokio::test] +async fn does_not_follow_a_cursor_out_of_its_recording() -> anyhow::Result<()> { + let (_recordings, config_handle, _process) = start_with_recordings().await?; + let mut request = serde_json::json!({ "query": "created user", "recordings": "all", "maxHits": 1 }); + + let response = search(config_handle.http_port(), "gateway.recordings.search", request.clone()).await?; + let first_page: serde_json::Value = response.json().await?; + // From the start of the file, so that reading `outside.slog` would find its line. + request["cursor"] = edit_cursor( + &first_page["nextCursor"], + serde_json::json!({ + "fileName": "../outside.slog", + "position": { "offset": 0, "lineNumber": 0, "inLongLine": false }, + }), + )?; + let response = search(config_handle.http_port(), "gateway.recordings.search", request).await?; + + assert_eq!(response.status(), reqwest::StatusCode::OK); + let body: serde_json::Value = response.json().await?; + assert!(hit_summaries(&body)?.is_empty()); + + Ok(()) +} + +// oracle: intent devolutions-gateway/src/api/jrec/search.intent.md §Pages +#[tokio::test] +async fn rejects_a_cursor_for_a_recording_not_in_the_list() -> anyhow::Result<()> { + let (_recordings, config_handle, _process) = start_with_recordings().await?; + let mut request = serde_json::json!({ + "query": "created user", + "recordings": [OLDER_RECORDING.to_string(), NEWER_RECORDING.to_string()], + "maxHits": 1, + }); + + let response = search(config_handle.http_port(), "gateway.recordings.search", request.clone()).await?; + let first_page: serde_json::Value = response.json().await?; + request["cursor"] = edit_cursor( + &first_page["nextCursor"], + serde_json::json!({ "recordingId": Uuid::from_u128(3).to_string() }), + )?; + let response = search(config_handle.http_port(), "gateway.recordings.search", request).await?; + + assert_eq!(response.status(), reqwest::StatusCode::BAD_REQUEST); + + Ok(()) +} + // oracle: intent devolutions-gateway/src/api/jrec/search.intent.md §Access #[tokio::test] async fn rejects_a_token_without_the_search_scope() -> anyhow::Result<()> { From 704ed091e9dc4455703f8f62874352ddf023bfc4 Mon Sep 17 00:00:00 2001 From: Junyi Ou Date: Fri, 23 Oct 2026 14:12:39 -0400 Subject: [PATCH 2/2] docs(dgw): say a search page ends between hits Co-Authored-By: Claude Opus 5.5 --- devolutions-gateway/src/api/jrec/search.intent.md | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/devolutions-gateway/src/api/jrec/search.intent.md b/devolutions-gateway/src/api/jrec/search.intent.md index 91f5d4fb1..3bca91179 100644 --- a/devolutions-gateway/src/api/jrec/search.intent.md +++ b/devolutions-gateway/src/api/jrec/search.intent.md @@ -48,7 +48,9 @@ A page carries a `nextCursor` until every log is read, and a page with fewer hit Gateway keeps no state between pages; the cursor holds where the next page starts. A cursor is accepted only with the same `query`, `recordings` and `startTime` as the page that gave it. -The hit that does not fit in a page opens the next one, so that every page moves the search forward and returns each hit once. +A page ends only between hits, so no hit is split across pages or cut short to fit one. +A hit that would exceed `maxHits` opens the next page, so that every page moves the search forward and returns each hit once. +A hit on the line that crosses the read limit stays in that page rather than moving to the next, since the next page would read that line again. Pages never go back: a recording that starts, or a log that grows, behind the cursor while paging is not searched. A log replaced while paging, as an AI analysis rerun does, is read from the cursor's byte offset on, so lines can be missed or misnumbered; this is accepted for now.