openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-30 19:18:49 +00:00
Code Issues Packages Projects Releases Wiki Activity

Keep resume picker pagination moving across backends (#57862)

Browse Source 
## 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
This commit is contained in:
Benjamin Carlsson
2026-10-28 23:27:31 +00:00
committed by copyberry
parent bd2e9bf0f4
commit f9c10053fb
6 changed files with 583 additions and 122 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/tui/src/resume_picker.rs
+219 -24
View File
@@ -186,6 +186,7 @@ impl SessionPickerAction {
struct PageLoadRequest {
cursor: Option<PageCursor>,
request_token: usize,
listing_token: usize,
search_token: Option<usize>,
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<PageCursor>,
},
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<u8>,
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<Option<SessionSelection>> {
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<_>>(),
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<_>>(),
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<_>>(),
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<Mutex<Vec<PageLoadRequest>>> = 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"),
&timestamp,
&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());
codex-rs/tui/src/resume_picker/catalog.rs
+11 -4
View File
@@ -30,10 +30,17 @@ pub(super) struct PageRequest {
}
impl ThreadCatalog {
pub(super) fn has_more(&self, excluded: &HashSet<BackendId>) -> bool {
self.sources
.iter()
.any(|(id, source)| !source.exhausted && !excluded.contains(id))
pub(super) fn next_cursor(&self, excluded: &HashSet<BackendId>) -> Option<PageCursor> {
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) {
codex-rs/tui/src/resume_picker/catalog_tests.rs
+1 -1
View File
@@ -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"),
})
}
codex-rs/tui/src/resume_picker/page_loading.rs
+3 -4
View File
@@ -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(_))
}
codex-rs/tui/src/resume_picker/sources.rs
+116 -53
View File
@@ -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<BackendId>) -> 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<BackendId, PendingPage> = HashMap::new();
let mut awaiting_completion = None;
let mut failed = HashSet::new();
let mut current: Option<PageLoadRequest> = 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);
codex-rs/tui/src/resume_picker/sources_tests.rs
+233 -36
View File
@@ -28,6 +28,7 @@ fn source() -> (
fn request(token: usize, cursor: Option<PageCursor>) -> 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());
}