From f9c10053fb3d9b960e78cfeae1f3ffa21e0dc00a Mon Sep 17 00:00:00 2001 From: Benjamin Carlsson Date: Wed, 28 Oct 2026 23:16:33 +0000 Subject: [PATCH] Keep resume picker pagination moving across backends (#57862) ## Why A slow session backend could block further pagination from responsive backends. Sessions arriving later also need to appear in the correct sort order without moving the selection to a different session. ## What changed - Let each backend continue pagination independently while accepting delayed results from the current listing. - Distinguish pending backend replies from available pages so search waits for results without issuing redundant requests. - Ignore rows and errors from previous listings, and update search progress when a backend fails. - Sort merged sessions globally by the selected timestamp and preserve the selected session and pending page-down distance. ## Testing Add regression coverage for independent backend pagination, delayed failures, fallback errors, stale listing events, search query changes during loading, and selection preservation when rows are reordered. Update the mixed-backend table snapshot test to ingest sessions in separate batches. GitOrigin-RevId: 15773717567c634d8b9bf7639c3bfc78369fee03 --- codex-rs/tui/src/resume_picker.rs | 243 ++++++++++++++-- codex-rs/tui/src/resume_picker/catalog.rs | 15 +- .../tui/src/resume_picker/catalog_tests.rs | 2 +- .../tui/src/resume_picker/page_loading.rs | 7 +- codex-rs/tui/src/resume_picker/sources.rs | 169 +++++++---- .../tui/src/resume_picker/sources_tests.rs | 269 +++++++++++++++--- 6 files changed, 583 insertions(+), 122 deletions(-) diff --git a/codex-rs/tui/src/resume_picker.rs b/codex-rs/tui/src/resume_picker.rs index c054f23d1d..19c953f643 100644 --- a/codex-rs/tui/src/resume_picker.rs +++ b/codex-rs/tui/src/resume_picker.rs @@ -186,6 +186,7 @@ impl SessionPickerAction { struct PageLoadRequest { cursor: Option, request_token: usize, + listing_token: usize, search_token: Option, mode: PageLoadMode, filter_mode: SessionFilterMode, @@ -335,11 +336,12 @@ struct PickerBackground { enum BackgroundEvent { PartialPage { - request_token: usize, + listing_token: usize, page: PickerPage, }, SourceError { - request_token: usize, + listing_token: usize, + next_cursor: Option, }, Page { request_token: usize, @@ -369,6 +371,7 @@ enum BackgroundEvent { enum PageCursor { AppServer(String), Catalog, + SourcesPending, } struct PickerPage { @@ -806,7 +809,7 @@ fn spawn_app_server_page_loader( ); let cursor = request.cursor.map(|cursor| match cursor { PageCursor::AppServer(cursor) => cursor, - PageCursor::Catalog => { + PageCursor::Catalog | PageCursor::SourcesPending => { unreachable!("catalog cursors never reach a source") } }); @@ -963,6 +966,7 @@ struct PickerState { frozen_footer_percent: Option, query: String, search_state: SearchState, + listing_token: usize, next_request_token: usize, next_search_token: usize, picker_loader: PickerLoader, @@ -1167,6 +1171,7 @@ impl PickerState { frozen_footer_percent: None, query: String::new(), search_state: SearchState::Idle, + listing_token: 0, next_request_token: 0, next_search_token: 0, picker_loader, @@ -1590,6 +1595,7 @@ impl PickerState { }; let request_token = self.allocate_request_token(); + self.listing_token = request_token; let mode = self.initial_page_mode; self.pagination .start_load(request_token, search_token, mode); @@ -1598,6 +1604,7 @@ impl PickerState { (self.picker_loader)(PickerLoadRequest::Page(PageLoadRequest { cursor: None, request_token, + listing_token: self.listing_token, search_token, mode, filter_mode: self.filter_mode, @@ -1614,19 +1621,27 @@ impl PickerState { ) -> Result> { match event { BackgroundEvent::PartialPage { - request_token, + listing_token, page, } => { - if self.pagination.is_pending(request_token) { - self.ingest_page(page); + if listing_token != self.listing_token { + return Ok(None); } + self.ingest_page(page); + self.continue_search_if_needed(); } - BackgroundEvent::SourceError { request_token } => { - if self.pagination.is_pending(request_token) { - self.inline_error = - Some("Could not load one backend. Reopen the picker to retry.".into()); - self.request_frame(); + BackgroundEvent::SourceError { + listing_token, + next_cursor, + } => { + if listing_token != self.listing_token { + return Ok(None); } + self.pagination.next_cursor = next_cursor; + self.continue_search_if_needed(); + self.inline_error = + Some("Could not load one backend. Reopen the picker to retry.".into()); + self.request_frame(); } BackgroundEvent::Page { request_token, @@ -1655,6 +1670,7 @@ impl PickerState { (self.picker_loader)(PickerLoadRequest::Page(PageLoadRequest { cursor: None, request_token, + listing_token: self.listing_token, search_token, mode: PageLoadMode::StoreDefault, filter_mode: self.filter_mode, @@ -1680,7 +1696,7 @@ impl PickerState { }; self.ingest_page(page); self.complete_pending_page_down(); - let completed_token = pending.search_token.or(search_token); + let completed_token = self.search_state.active_token(); self.continue_search_if_token_matches(completed_token); } BackgroundEvent::Preview { thread_id, preview } => { @@ -1736,6 +1752,13 @@ impl PickerState { } fn ingest_page(&mut self, page: PickerPage) { + let selected_key = self + .filtered_rows + .get(self.selected) + .and_then(Row::seen_key); + let pending_page_down_distance = self + .pending_page_down_target + .map(|target| target.saturating_sub(self.selected)); let PickerPage { rows, history_modes, @@ -1771,7 +1794,29 @@ impl PickerState { } } + let sort_key = self.sort_key; + self.all_rows.sort_by_key(|row| { + std::cmp::Reverse(match sort_key { + ThreadSortKey::CreatedAt => row.created_at, + ThreadSortKey::UpdatedAt + | ThreadSortKey::RecencyAt + | ThreadSortKey::SectionPosition => row.updated_at.or(row.created_at), + }) + }); self.apply_filter(); + if let Some(selected_key) = selected_key + && let Some(index) = self + .filtered_rows + .iter() + .position(|row| row.seen_key().as_ref() == Some(&selected_key)) + { + self.selected = index; + if let Some(distance) = pending_page_down_distance { + self.pending_page_down_target = Some(index.saturating_add(distance)); + } + self.ensure_selected_visible(); + self.request_frame(); + } } fn complete_pending_page_down(&mut self) { @@ -1991,6 +2036,7 @@ impl PickerState { (self.picker_loader)(PickerLoadRequest::Page(PageLoadRequest { cursor: Some(cursor), request_token, + listing_token: self.listing_token, search_token, mode, filter_mode: self.filter_mode, @@ -4700,7 +4746,7 @@ mod tests { } #[test] - fn resume_table_snapshot() { + fn mixed_backend_resume_table_snapshot() { let loader = page_only_loader(|_| {}); let mut state = PickerState::new( crate::backend_id::BackendId::new(), @@ -4713,7 +4759,7 @@ mod tests { ); let now = parse_timestamp_str("2026-04-28T16:30:00Z").expect("timestamp"); - let rows = vec![ + let mut rows = [ Row { backend_id: None, history_mode: None, @@ -4751,13 +4797,43 @@ mod tests { git_branch: None, }, ]; - state.all_rows = rows.clone(); - state.filtered_rows = rows; + let local_backend = state.backend_id; + let cloud_backend = crate::backend_id::BackendId::new(); + rows[0].backend_id = Some(cloud_backend); + for row in &mut rows[1..] { + row.backend_id = Some(local_backend); + } + state.ingest_page(page( + rows[1..].to_vec(), + /*next_cursor*/ None, + /*num_scanned_files*/ 2, + /*reached_scan_cap*/ false, + )); + state.selected = 0; + state.ingest_page(page( + vec![rows[0].clone()], + /*next_cursor*/ None, + /*num_scanned_files*/ 1, + /*reached_scan_cap*/ false, + )); state.relative_time_reference = Some(now); - state.selected = 1; state.scroll_top = 0; state.update_viewport(/*rows*/ 12, /*width*/ 80); + assert_eq!( + state + .filtered_rows + .iter() + .map(|row| row.backend_id) + .collect::>(), + vec![ + Some(cloud_backend), + Some(local_backend), + Some(local_backend) + ] + ); + assert_eq!(state.selected, 1); + let snapshot = render_picker_list(&state, /*width*/ 80, /*height*/ 12); assert_snapshot!("resume_picker_table", snapshot); } @@ -6432,6 +6508,79 @@ session_picker_view = "dense" assert_eq!(unique_paths.len(), 4); } + #[test] + fn page_ingestion_globally_sorts_rows_and_preserves_selection() { + let loader = page_only_loader(|_| {}); + let mut state = PickerState::new( + crate::backend_id::BackendId::new(), + FrameRequester::test_dummy(), + loader, + ProviderFilter::MatchDefault(String::from("openai")), + /*show_all*/ true, + /*filter_cwd*/ None, + SessionPickerAction::Resume, + ); + state.sort_key = ThreadSortKey::CreatedAt; + state.reset_pagination(); + state.ingest_page(page( + vec![make_row( + "/tmp/selected.jsonl", + "2025-01-02T00:00:00Z", + "selected", + )], + Some("2025-01-02T00:00:00Z"), + /*num_scanned_files*/ 1, + /*reached_scan_cap*/ false, + )); + + let mut newest_update = make_row( + "/tmp/newest-update.jsonl", + "2025-01-01T00:00:00Z", + "newest update", + ); + newest_update.updated_at = parse_timestamp_str("2025-01-04T00:00:00Z"); + state.ingest_page(page( + vec![ + make_row( + "/tmp/newest-created.jsonl", + "2025-01-03T00:00:00Z", + "newest created", + ), + newest_update, + ], + /*next_cursor*/ None, + /*num_scanned_files*/ 2, + /*reached_scan_cap*/ false, + )); + + assert_eq!( + state + .filtered_rows + .iter() + .map(|row| row.preview.as_str()) + .collect::>(), + vec!["newest created", "selected", "newest update"] + ); + assert_eq!(state.filtered_rows[state.selected].preview, "selected"); + + state.sort_key = ThreadSortKey::UpdatedAt; + state.ingest_page(page( + Vec::new(), + /*next_cursor*/ None, + /*num_scanned_files*/ 0, + /*reached_scan_cap*/ false, + )); + assert_eq!( + state + .filtered_rows + .iter() + .map(|row| row.preview.as_str()) + .collect::>(), + vec!["newest update", "newest created", "selected"] + ); + assert_eq!(state.filtered_rows[state.selected].preview, "selected"); + } + #[test] fn ensure_minimum_rows_prefetches_when_underfilled() { let recorded_requests: Arc>> = Arc::new(Mutex::new(Vec::new())); @@ -6783,6 +6932,27 @@ session_picker_view = "dense" .await .unwrap(); assert_eq!(state.selected, 0); + + state.selected = 9; + state.pending_page_down_target = Some(19); + let newer_items = (20..50) + .map(|idx| { + let timestamp = format!("2026-01-{:02}T00:00:00Z", idx - 19); + make_row( + &format!("/tmp/item-{idx}.jsonl"), + ×tamp, + &format!("item-{idx}"), + ) + }) + .collect(); + state.ingest_page(page( + newer_items, + /*next_cursor*/ None, + /*num_scanned_files*/ 30, + /*reached_scan_cap*/ false, + )); + assert_eq!(state.selected, 39); + assert_eq!(state.pending_page_down_target, Some(49)); } #[tokio::test] @@ -7484,6 +7654,8 @@ session_picker_view = "dense" assert_eq!(guard.len(), 1); guard[0].clone() }; + state.set_query("later".to_string()); + assert_eq!(recorded_requests.lock().unwrap().len(), 1); state .handle_background_event(BackgroundEvent::Page { @@ -7517,7 +7689,7 @@ session_picker_view = "dense" vec![make_row( "/tmp/match.jsonl", "2025-01-03T00:00:00Z", - "target log", + "later log", )], Some("2025-01-04T00:00:00Z"), /*num_scanned_files*/ 7, @@ -7554,20 +7726,43 @@ session_picker_view = "dense" .unwrap(); assert_eq!(recorded_requests.lock().unwrap().len(), 1); + let mut pending_page = page( + Vec::new(), + /*next_cursor*/ None, + /*num_scanned_files*/ 3, + /*reached_scan_cap*/ false, + ); + pending_page.next_cursor = Some(PageCursor::SourcesPending); state .handle_background_event(BackgroundEvent::Page { request_token: active_request.request_token, search_token: active_request.search_token, mode: active_request.mode, - page: Ok(page( - Vec::new(), - /*next_cursor*/ None, - /*num_scanned_files*/ 3, - /*reached_scan_cap*/ true, - )), + page: Ok(pending_page), }) .await .unwrap(); + assert!(state.search_state.is_active()); + assert_eq!(recorded_requests.lock().unwrap().len(), 1); + state.set_query("updated".to_string()); + assert!(state.search_state.is_active()); + assert_eq!(recorded_requests.lock().unwrap().len(), 1); + + state + .handle_background_event(BackgroundEvent::PartialPage { + listing_token: state.listing_token, + page: page( + vec![make_row("/s", "2025-01-04T00:00:00Z", "slow")], + Some("2025-01-05T00:00:00Z"), + /*num_scanned_files*/ 1, + /*reached_scan_cap*/ false, + ), + }) + .await + .unwrap(); + assert_eq!(recorded_requests.lock().unwrap().len(), 2); + state.pagination.reached_scan_cap = true; + state.continue_search_if_needed(); assert!(state.filtered_rows.is_empty()); assert!(!state.search_state.is_active()); diff --git a/codex-rs/tui/src/resume_picker/catalog.rs b/codex-rs/tui/src/resume_picker/catalog.rs index cfb5939c65..e207e2a05f 100644 --- a/codex-rs/tui/src/resume_picker/catalog.rs +++ b/codex-rs/tui/src/resume_picker/catalog.rs @@ -30,10 +30,17 @@ pub(super) struct PageRequest { } impl ThreadCatalog { - pub(super) fn has_more(&self, excluded: &HashSet) -> bool { - self.sources - .iter() - .any(|(id, source)| !source.exhausted && !excluded.contains(id)) + pub(super) fn next_cursor(&self, excluded: &HashSet) -> Option { + if self.sources.iter().any(|(id, source)| { + !source.exhausted && source.pending.is_none() && !excluded.contains(id) + }) { + Some(PageCursor::Catalog) + } else { + self.sources + .values() + .any(|source| source.pending.is_some()) + .then_some(PageCursor::SourcesPending) + } } pub(super) fn reset(&mut self) { diff --git a/codex-rs/tui/src/resume_picker/catalog_tests.rs b/codex-rs/tui/src/resume_picker/catalog_tests.rs index 8c6543f410..f73a9ae31f 100644 --- a/codex-rs/tui/src/resume_picker/catalog_tests.rs +++ b/codex-rs/tui/src/resume_picker/catalog_tests.rs @@ -25,7 +25,7 @@ fn listed(thread: ThreadId, cursor: Option<&str>) -> PickerPage { fn cursor(request: &PageRequest) -> Option<&str> { request.cursor.as_ref().map(|cursor| match cursor { PageCursor::AppServer(value) => value.as_str(), - PageCursor::Catalog => panic!("unexpected catalog cursor"), + PageCursor::Catalog | PageCursor::SourcesPending => panic!("unexpected catalog cursor"), }) } diff --git a/codex-rs/tui/src/resume_picker/page_loading.rs b/codex-rs/tui/src/resume_picker/page_loading.rs index 5795c61d06..054372aa1c 100644 --- a/codex-rs/tui/src/resume_picker/page_loading.rs +++ b/codex-rs/tui/src/resume_picker/page_loading.rs @@ -129,15 +129,14 @@ impl PaginationState { if self.is_loading() { return None; } + if matches!(self.next_cursor, Some(PageCursor::SourcesPending)) { + return None; + } self.next_cursor .clone() .map(|cursor| (cursor, self.next_page_mode)) } - pub(super) fn is_pending(&self, request_token: usize) -> bool { - matches!(self.loading, LoadingState::Pending(pending) if pending.request_token == request_token) - } - pub(super) fn is_loading(&self) -> bool { matches!(self.loading, LoadingState::Pending(_)) } diff --git a/codex-rs/tui/src/resume_picker/sources.rs b/codex-rs/tui/src/resume_picker/sources.rs index 5a9efb7901..243d96af68 100644 --- a/codex-rs/tui/src/resume_picker/sources.rs +++ b/codex-rs/tui/src/resume_picker/sources.rs @@ -8,6 +8,21 @@ use futures::StreamExt; use futures::stream::SelectAll; use std::collections::HashSet; +struct PendingPage { + request: PageLoadRequest, + page: catalog::PageRequest, +} + +fn completion_page(catalog: &catalog::ThreadCatalog, failed: &HashSet) -> PickerPage { + PickerPage { + rows: Vec::new(), + history_modes: HashMap::new(), + next_cursor: catalog.next_cursor(failed), + num_scanned_files: 0, + reached_scan_cap: false, + } +} + pub(super) struct Source { pub(super) backend_id: BackendId, pub(super) loader: PickerLoader, @@ -42,11 +57,9 @@ pub(super) fn spawn( } let task = tokio::spawn(async move { let mut catalog = catalog::ThreadCatalog::default(); - let mut pending = HashMap::new(); + let mut pending: HashMap = HashMap::new(); + let mut awaiting_completion = None; let mut failed = HashSet::new(); - let mut current: Option = None; - let mut succeeded = false; - let mut error = None; loop { tokio::select! { biased; @@ -55,24 +68,43 @@ pub(super) fn spawn( let Some(request) = request else { break }; match request { PickerLoadRequest::Page(request) => { - succeeded = false; if request.cursor.is_none() { catalog.reset(); pending.clear(); + awaiting_completion = None; failed.retain(|id| !workers.contains_key(id)); } - error = workers.is_empty().then(|| std::io::Error::other("No session backends are available.")); + let request_token = request.request_token; + let mut started = false; for (id, worker) in &workers { if failed.contains(id) { continue; } if let Some(page) = catalog.begin_page(*id, request.mode) { let mut source_request = request.clone(); source_request.cursor = page.cursor.clone(); source_request.mode = page.mode; - pending.insert(*id, page); + pending.insert(*id, PendingPage { + request: source_request.clone(), + page, + }); worker(PickerLoadRequest::Page(source_request)); + started = true; } } - current = Some(request); + if started { + awaiting_completion = Some(request); + } else { + let page = if workers.is_empty() { + Err(std::io::Error::other("No session backends are available.")) + } else { + Ok(completion_page(&catalog, &failed)) + }; + let _ = output.send(BackgroundEvent::Page { + request_token, + search_token: request.search_token, + mode: request.mode, + page, + }); + } } request => { let backend = match &request { @@ -104,74 +136,105 @@ pub(super) fn spawn( None => { workers.remove(&id); failed.insert(id); - let Some(request) = current.as_ref().filter(|_| pending.contains_key(&id)) else { continue }; - BackgroundEvent::Page { request_token: request.request_token, search_token: request.search_token, mode: request.mode, - page: Err(std::io::Error::other("The session backend disconnected.")) } + let Some(pending_page) = pending.get(&id) else { continue }; + BackgroundEvent::Page { + request_token: pending_page.request.request_token, + search_token: pending_page.request.search_token, + mode: pending_page.request.mode, + page: Err(std::io::Error::other("The session backend disconnected.")), + } } }; match event { BackgroundEvent::Page { request_token, page, .. } => { - if current.as_ref().is_none_or(|request| request.request_token != request_token) { continue; } - let Some(mut page_request) = pending.remove(&id) else { continue }; - let should_fallback = page_request.mode == PageLoadMode::StateDbOnly - && page_request.cursor.is_none() + let Some(pending_page) = pending.get(&id) else { continue }; + if pending_page.request.request_token != request_token { continue; } + let Some(mut pending_page) = pending.remove(&id) else { continue }; + let is_initial_page = pending_page.page.cursor.is_none(); + let should_fallback = pending_page.page.mode == PageLoadMode::StateDbOnly + && is_initial_page && workers.contains_key(&id) && (page.is_err() || page.as_ref().is_ok_and(|page| page.rows.is_empty())); if should_fallback { - let Some(request) = current.as_mut() else { - continue; - }; - request.mode = PageLoadMode::StoreDefault; - let mut fallback = request.clone(); - fallback.cursor = page_request.cursor.clone(); - page_request.mode = PageLoadMode::StoreDefault; - pending.insert(id, page_request); + if let Some(request) = awaiting_completion.as_mut() + && request.request_token == request_token + { + request.mode = PageLoadMode::StoreDefault; + } + let mut fallback = pending_page.request; + fallback.mode = PageLoadMode::StoreDefault; + fallback.cursor = pending_page.page.cursor.clone(); + pending_page.page.mode = PageLoadMode::StoreDefault; + pending_page.request = fallback.clone(); + pending.insert(id, pending_page); workers[&id](PickerLoadRequest::Page(fallback)); continue; } + let mut wait_for_other_source = false; + let mut completion_error = None; match page { Ok(page) => { - let Some(Ok(mut page)) = catalog.finish_page(page_request, Ok(page)) else { continue }; - succeeded = true; - // Only the catalog owns source cursors. The UI receives a - // continuation marker once this round's workers have replied. - page.next_cursor = None; - let _ = output.send(BackgroundEvent::PartialPage { request_token, page }); + let Some(Ok(mut page)) = catalog.finish_page(pending_page.page, Ok(page)) else { continue }; + page.next_cursor = catalog.next_cursor(&failed); + let _ = output.send(BackgroundEvent::PartialPage { + listing_token: pending_page.request.listing_token, + page, + }); } Err(err) => { - let Some(Err(err)) = catalog.finish_page(page_request, Err(err)) else { continue }; + let Some(Err(err)) = catalog.finish_page(pending_page.page, Err(err)) else { continue }; failed.insert(id); - error = Some(err); - let _ = output.send(BackgroundEvent::SourceError { request_token }); + if let Some(request) = awaiting_completion.as_mut() + && request.request_token == request_token + { + request.mode = PageLoadMode::StoreDefault; + } + let _ = output.send(BackgroundEvent::SourceError { + listing_token: pending_page.request.listing_token, + next_cursor: catalog.next_cursor(&failed), + }); + if is_initial_page { + wait_for_other_source = pending.values().any(|pending| { + pending.request.request_token == request_token + }); + completion_error = Some(err); + } } } + let can_continue = matches!( + catalog.next_cursor(&failed), + Some(PageCursor::Catalog) + ); + let completes_source_request = awaiting_completion + .as_ref() + .is_some_and(|request| { + request.request_token == request_token + && !wait_for_other_source + }); + if (completes_source_request || can_continue) + && let Some(request) = awaiting_completion.take() + { + let page = if completes_source_request { + completion_error.map_or_else( + || Ok(completion_page(&catalog, &failed)), + Err, + ) + } else { + Ok(completion_page(&catalog, &failed)) + }; + let _ = output.send(BackgroundEvent::Page { + request_token: request.request_token, + search_token: request.search_token, + mode: request.mode, + page, + }); + } } event => { let _ = output.send(event); } } } } - if pending.is_empty() - && let Some(request) = current.take() - { - let page = if !succeeded && let Some(error) = error.take() { - Err(error) - } else { - Ok(PickerPage { - rows: Vec::new(), - history_modes: HashMap::new(), - next_cursor: catalog.has_more(&failed).then_some(PageCursor::Catalog), - num_scanned_files: 0, - reached_scan_cap: false, - }) - }; - let _ = output.send(BackgroundEvent::Page { - request_token: request.request_token, - search_token: request.search_token, - mode: request.mode, - page, - }); - } } drop(workers); drop(events); diff --git a/codex-rs/tui/src/resume_picker/sources_tests.rs b/codex-rs/tui/src/resume_picker/sources_tests.rs index c37b87fd44..ac3dd836a9 100644 --- a/codex-rs/tui/src/resume_picker/sources_tests.rs +++ b/codex-rs/tui/src/resume_picker/sources_tests.rs @@ -28,6 +28,7 @@ fn source() -> ( fn request(token: usize, cursor: Option) -> PickerLoadRequest { PickerLoadRequest::Page(PageLoadRequest { request_token: token, + listing_token: token, cursor, search_token: None, mode: PageLoadMode::StoreDefault, @@ -100,12 +101,28 @@ async fn sources_stream_rows_and_route_actions_to_each_source() { }; assert_eq!(fallback_request.mode, PageLoadMode::StoreDefault); second_events - .send(reply(/*token*/ 1, thread, /*cursor*/ None)) + .send(reply(/*token*/ 1, thread, Some("second-next"))) .unwrap(); let BackgroundEvent::PartialPage { page, .. } = next(&mut events).await else { panic!("expected responsive source page") }; assert_eq!(page.rows[0].backend_id, Some(second_id)); + assert!(matches!( + next(&mut events).await, + BackgroundEvent::Page { + mode: PageLoadMode::StoreDefault, + page: Ok(_), + .. + } + )); + loader(request(/*token*/ 2, Some(PageCursor::Catalog))); + let Some(PickerLoadRequest::Page(second_page)) = second_requests.recv().await else { + panic!("expected the responsive source to continue") + }; + assert!( + matches!(second_page.cursor, Some(PageCursor::AppServer(cursor)) if cursor == "second-next") + ); + assert!(first_requests.try_recv().is_err()); first_events .send(reply(/*token*/ 1, thread, Some("first-next"))) .unwrap(); @@ -113,6 +130,23 @@ async fn sources_stream_rows_and_route_actions_to_each_source() { panic!("expected other source page") }; assert_eq!(page.rows[0].backend_id, Some(first_id)); + assert!(matches!( + next(&mut events).await, + BackgroundEvent::Page { + request_token: 2, + page: Ok(_), + .. + } + )); + loader(request(/*token*/ 3, Some(PageCursor::Catalog))); + let Some(PickerLoadRequest::Page(next_page)) = first_requests.recv().await else { + panic!("expected the newly ready source to continue") + }; + assert_eq!(next_page.mode, PageLoadMode::StateDbOnly); + assert!( + matches!(next_page.cursor, Some(PageCursor::AppServer(cursor)) if cursor == "first-next") + ); + assert!(second_requests.try_recv().is_err()); loader(PickerLoadRequest::Archive { backend_id: first_id, thread_id: thread, @@ -134,27 +168,21 @@ async fn sources_stream_rows_and_route_actions_to_each_source() { next(&mut events).await, BackgroundEvent::SourceError { .. } )); - assert!(matches!( - next(&mut events).await, - BackgroundEvent::Page { - mode: PageLoadMode::StoreDefault, - page: Ok(_), - .. - } - )); - loader(request(/*token*/ 2, Some(PageCursor::Catalog))); - let Some(PickerLoadRequest::Page(next_page)) = first_requests.recv().await else { - panic!("expected next page") - }; - assert_eq!(next_page.mode, PageLoadMode::StateDbOnly); - assert!( - matches!(next_page.cursor, Some(PageCursor::AppServer(cursor)) if cursor == "first-next") - ); - assert!(second_requests.try_recv().is_err()); - first_events + second_events .send(reply( /*token*/ 2, ThreadId::new(), + /*cursor*/ None, + )) + .unwrap(); + assert!(matches!( + next(&mut events).await, + BackgroundEvent::PartialPage { .. } + )); + first_events + .send(reply( + /*token*/ 3, + ThreadId::new(), /*cursor*/ Some("last-page"), )) .unwrap(); @@ -166,7 +194,7 @@ async fn sources_stream_rows_and_route_actions_to_each_source() { next(&mut events).await, BackgroundEvent::Page { page: Ok(_), .. } )); - loader(request(/*token*/ 3, Some(PageCursor::Catalog))); + loader(request(/*token*/ 4, Some(PageCursor::Catalog))); let Some(PickerLoadRequest::Page(last_page)) = first_requests.recv().await else { panic!("expected last page") }; @@ -175,7 +203,7 @@ async fn sources_stream_rows_and_route_actions_to_each_source() { ); first_events .send(BackgroundEvent::Page { - request_token: 3, + request_token: 4, search_token: None, mode: PageLoadMode::StateDbOnly, page: Err(std::io::Error::other("last page failed")), @@ -183,13 +211,13 @@ async fn sources_stream_rows_and_route_actions_to_each_source() { .unwrap(); assert!(matches!( next(&mut events).await, - BackgroundEvent::SourceError { request_token: 3 } + BackgroundEvent::SourceError { .. } )); assert!(matches!( next(&mut events).await, BackgroundEvent::Page { - request_token: 3, - page: Err(_), + request_token: 4, + page: Ok(_), .. } )); @@ -198,6 +226,114 @@ async fn sources_stream_rows_and_route_actions_to_each_source() { assert!(first_requests.recv().await.is_none()); assert!(second_requests.recv().await.is_none()); } + +#[tokio::test] +async fn late_source_failure_finishes_catalog_progress() { + let (first, mut first_requests, first_events) = source(); + let (second, mut second_requests, second_events) = source(); + let (output, mut events) = mpsc::unbounded_channel(); + let PickerBackground { loader, task } = spawn(vec![first, second], output); + loader(request(/*token*/ 1, /*cursor*/ None)); + first_requests.recv().await.unwrap(); + second_requests.recv().await.unwrap(); + first_events + .send(BackgroundEvent::Page { + request_token: 1, + search_token: None, + mode: PageLoadMode::StoreDefault, + page: Ok(page( + Vec::new(), + /*next_cursor*/ None, + /*num_scanned_files*/ 0, + /*reached_scan_cap*/ false, + )), + }) + .unwrap(); + assert!(matches!( + next(&mut events).await, + BackgroundEvent::PartialPage { .. } + )); + assert!(matches!( + next(&mut events).await, + BackgroundEvent::Page { + page: Ok(PickerPage { + next_cursor: Some(PageCursor::SourcesPending), + .. + }), + .. + } + )); + second_events + .send(BackgroundEvent::Page { + request_token: 1, + search_token: None, + mode: PageLoadMode::StoreDefault, + page: Err(std::io::Error::other("listing failed")), + }) + .unwrap(); + assert!(matches!( + next(&mut events).await, + BackgroundEvent::SourceError { + next_cursor: None, + .. + } + )); + drop(loader); + task.await.unwrap(); +} + +#[tokio::test] +async fn single_source_initial_failure_preserves_fallback_error() { + let (source, mut requests, replies) = source(); + let (output, mut events) = mpsc::unbounded_channel(); + let PickerBackground { loader, task } = spawn(vec![source], output); + let PickerLoadRequest::Page(mut initial) = request(/*token*/ 1, /*cursor*/ None) else { + unreachable!() + }; + initial.mode = PageLoadMode::StateDbOnly; + loader(PickerLoadRequest::Page(initial)); + let Some(PickerLoadRequest::Page(initial)) = requests.recv().await else { + panic!("expected initial listing") + }; + assert_eq!(initial.mode, PageLoadMode::StateDbOnly); + replies + .send(BackgroundEvent::Page { + request_token: 1, + search_token: None, + mode: PageLoadMode::StateDbOnly, + page: Err(std::io::Error::other("indexed listing failed")), + }) + .unwrap(); + let Some(PickerLoadRequest::Page(fallback)) = requests.recv().await else { + panic!("expected store fallback") + }; + assert_eq!(fallback.mode, PageLoadMode::StoreDefault); + replies + .send(BackgroundEvent::Page { + request_token: 1, + search_token: None, + mode: PageLoadMode::StoreDefault, + page: Err(std::io::Error::other("store listing failed")), + }) + .unwrap(); + assert!(matches!( + next(&mut events).await, + BackgroundEvent::SourceError { .. } + )); + let BackgroundEvent::Page { + mode, + page: Err(err), + .. + } = next(&mut events).await + else { + panic!("expected the fallback error") + }; + assert_eq!(mode, PageLoadMode::StoreDefault); + assert_eq!(err.to_string(), "store listing failed"); + drop(loader); + task.await.unwrap(); +} + #[tokio::test] async fn reset_rejects_old_replies_and_missing_sources_never_fall_back() { let (source, mut requests, replies) = source(); @@ -215,14 +351,10 @@ async fn reset_rejects_old_replies_and_missing_sources_never_fall_back() { replies .send(reply(/*token*/ 2, current, /*cursor*/ None)) .unwrap(); - let BackgroundEvent::PartialPage { - request_token, - page, - } = next(&mut events).await - else { + let BackgroundEvent::PartialPage { page, .. } = next(&mut events).await else { panic!("expected current page") }; - assert_eq!((request_token, page.rows[0].backend_id), (2, Some(owner))); + assert_eq!(page.rows[0].backend_id, Some(owner)); assert!(matches!( next(&mut events).await, BackgroundEvent::Page { @@ -256,13 +388,17 @@ async fn reset_does_not_restore_a_disconnected_source() { let owner = fallback.backend_id; let (output, mut events) = mpsc::unbounded_channel(); let PickerBackground { loader, task } = spawn(vec![preferred, fallback], output); - loader(request(/*token*/ 1, /*cursor*/ None)); + let PickerLoadRequest::Page(mut initial) = request(/*token*/ 1, /*cursor*/ None) else { + unreachable!() + }; + initial.mode = PageLoadMode::StateDbOnly; + loader(PickerLoadRequest::Page(initial)); preferred_requests.recv().await.unwrap(); fallback_requests.recv().await.unwrap(); drop(preferred_events); assert!(matches!( next(&mut events).await, - BackgroundEvent::SourceError { request_token: 1 } + BackgroundEvent::SourceError { .. } )); fallback_events .send(reply( @@ -275,7 +411,15 @@ async fn reset_does_not_restore_a_disconnected_source() { next(&mut events).await, BackgroundEvent::PartialPage { .. } )); - next(&mut events).await; + assert!(matches!( + next(&mut events).await, + BackgroundEvent::Page { + request_token: 1, + mode: PageLoadMode::StoreDefault, + page: Ok(_), + .. + } + )); loader(request(/*token*/ 2, /*cursor*/ None)); let thread = ThreadId::new(); fallback_requests.recv().await.unwrap(); @@ -335,6 +479,21 @@ async fn partial_failure_keeps_healthy_rows_visible() { /*search_token*/ None, PageLoadMode::StoreDefault, ); + picker + .handle_background_event(BackgroundEvent::Page { + request_token: 42, + search_token: None, + mode: PageLoadMode::StoreDefault, + page: Ok(page( + Vec::new(), + /*next_cursor*/ None, + /*num_scanned_files*/ 0, + /*reached_scan_cap*/ false, + )), + }) + .await + .unwrap(); + assert!(!picker.pagination.is_loading()); let thread = ThreadId::new(); let BackgroundEvent::Page { page: Ok(page), .. } = reply(/*token*/ 42, thread, /*cursor*/ None) @@ -343,17 +502,25 @@ async fn partial_failure_keeps_healthy_rows_visible() { }; picker .handle_background_event(BackgroundEvent::PartialPage { - request_token: 42, + listing_token: picker.listing_token, page, }) .await .unwrap(); + picker.query = "missing".into(); + picker.apply_filter(); + picker.search_state = SearchState::Active { token: 1 }; + picker.pagination.next_cursor = Some(PageCursor::SourcesPending); picker - .handle_background_event(BackgroundEvent::SourceError { request_token: 42 }) + .handle_background_event(BackgroundEvent::SourceError { + listing_token: picker.listing_token, + next_cursor: None, + }) .await .unwrap(); assert_eq!(picker.all_rows.len(), 1); - assert!(picker.pagination.is_loading()); + assert!(!picker.pagination.is_loading()); + assert!(matches!(picker.search_state, SearchState::Idle)); let backend = crate::test_backend::VT100Backend::new(/*width*/ 80, /*height*/ 1); let mut terminal = crate::custom_terminal::Terminal::with_options(backend).unwrap(); terminal.set_viewport_area(Rect::new(0, 0, 80, 1)); @@ -365,3 +532,33 @@ async fn partial_failure_keeps_healthy_rows_visible() { terminal.flush().unwrap(); insta::assert_snapshot!("partial_backend_failure", terminal.backend().to_string()); } + +#[tokio::test] +async fn queued_partial_page_from_previous_listing_is_ignored() { + let (mut picker, requests) = crate::resume_picker::tests::local_db_first_state(); + picker.start_initial_load(); + let stale_listing_token = requests.lock().unwrap()[0].listing_token; + picker.start_initial_load(); + assert_ne!(stale_listing_token, picker.listing_token); + picker + .handle_background_event(BackgroundEvent::PartialPage { + listing_token: stale_listing_token, + page: page( + vec![make_row("/repo/stale", "2025-01-01T00:00:00Z", "stale row")], + /*next_cursor*/ None, + /*num_scanned_files*/ 1, + /*reached_scan_cap*/ false, + ), + }) + .await + .unwrap(); + picker + .handle_background_event(BackgroundEvent::SourceError { + listing_token: stale_listing_token, + next_cursor: None, + }) + .await + .unwrap(); + assert!(picker.all_rows.is_empty()); + assert!(picker.inline_error.is_none()); +}