From 834f6a8ddc40ca371103834354df6d14d19c0c82 Mon Sep 17 00:00:00 2001 From: James Maes Date: Fri, 31 Jul 2026 16:42:25 -0500 Subject: [PATCH 1/2] feat(proto): keyset pagination for all list/search RPCs (#350) Apply one AIP-158 keyset pagination convention (page_size/page_token/ next_page_token, opaque cursor, not offset) to every list/search RPC: ListMessages, ListFolders, ListRules, ListEvents, ListContacts, SearchMessages, and reconcile Audit.ListRecords off its unstable limit/offset. A malformed page_token returns a typed VALIDATION_FAILED ErrorInfo (InvalidArgument). Regenerate the nuncio.v1 golden descriptor. Refs #275 Co-authored-by: James Maes --- crates/nuncio-cli/src/args.rs | 28 +- crates/nuncio-cli/src/runner.rs | 600 +++++++++++------- .../proto/nuncio/v1/descriptor.bin | Bin 14437 -> 15069 bytes .../nuncio-proto/proto/nuncio/v1/nuncio.proto | 139 +++- crates/nuncio-store/src/db.rs | 327 ++++++++++ crates/nuncio-store/src/search.rs | 111 ++++ crates/nunciod/src/grpc.rs | 521 ++++++++++++--- crates/nunciod/src/lib.rs | 1 + crates/nunciod/src/pagination.rs | 168 +++++ .../tests/caldav_production_e2e_test.rs | 4 + crates/nunciod/tests/calendar_e2e_test.rs | 4 + crates/nunciod/tests/contacts_e2e_test.rs | 6 + crates/nunciod/tests/spine_e2e_test.rs | 3 +- 13 files changed, 1536 insertions(+), 376 deletions(-) create mode 100644 crates/nunciod/src/pagination.rs diff --git a/crates/nuncio-cli/src/args.rs b/crates/nuncio-cli/src/args.rs index 554f691..4994093 100644 --- a/crates/nuncio-cli/src/args.rs +++ b/crates/nuncio-cli/src/args.rs @@ -449,14 +449,13 @@ pub enum SystemSubcommand { #[derive(Subcommand, Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "kebab-case")] pub enum AuditSubcommand { - /// List persisted WORM audit ledger records, sequence ascending. + /// List persisted WORM audit ledger records, sequence ascending. Follows + /// keyset pagination to the end, fetching `page_size` records per request. List { - /// Max records to fetch (0 = server default). - #[arg(short, long, default_value_t = 50, help = "Max records to fetch")] - limit: u32, - /// Pagination offset. - #[arg(short, long, default_value_t = 0, help = "Pagination offset")] - offset: u32, + /// Records fetched per page (0 = server default). All pages are + /// followed via the keyset cursor and printed together. + #[arg(short, long, default_value_t = 50, help = "Records fetched per page")] + page_size: u32, }, /// Verify the WORM audit ledger's cryptographic HMAC hash-chain /// integrity. @@ -716,25 +715,18 @@ mod tests { cli_list.command, Commands::System { action: SystemSubcommand::Audit { - action: AuditSubcommand::List { - limit: 50, - offset: 0, - } + action: AuditSubcommand::List { page_size: 50 } } } ); - let cli_list_paged = Cli::parse_from([ - "nuncio", "system", "audit", "list", "--limit", "10", "--offset", "5", - ]); + let cli_list_paged = + Cli::parse_from(["nuncio", "system", "audit", "list", "--page-size", "10"]); assert_eq!( cli_list_paged.command, Commands::System { action: SystemSubcommand::Audit { - action: AuditSubcommand::List { - limit: 10, - offset: 5, - } + action: AuditSubcommand::List { page_size: 10 } } } ); diff --git a/crates/nuncio-cli/src/runner.rs b/crates/nuncio-cli/src/runner.rs index 9bf5dbe..33b51e1 100644 --- a/crates/nuncio-cli/src/runner.rs +++ b/crates/nuncio-cli/src/runner.rs @@ -498,8 +498,8 @@ impl HeadlessRunner { Commands::System { action } => match action { SystemSubcommand::Status => self.handle_system_status(json_mode).await, SystemSubcommand::Audit { action } => match action { - AuditSubcommand::List { limit, offset } => { - self.handle_audit_list(*limit, *offset, json_mode).await + AuditSubcommand::List { page_size } => { + self.handle_audit_list(*page_size, json_mode).await } AuditSubcommand::Verify => self.handle_audit_verify(json_mode).await, }, @@ -726,33 +726,48 @@ impl HeadlessRunner { Err(e) => return Self::render_error(&e, json_mode), }; - match client - .list_messages(nuncio_proto::v1::ListMessagesRequest { - folder_id: folder.to_string(), - limit: 0, - }) - .await - { - Ok(response) => { - let messages = response.into_inner().messages; - if json_mode { - let messages_json: Vec = - messages.iter().map(message_proto_to_json).collect(); - format_json(&json!({ - "folder": folder, - "messages": messages_json - })) - } else { - format!("Folder '{}': {} message(s) found", folder, messages.len()) + // Follow `next_page_token` to the end so the command shows the whole + // folder, transparently exercising the keyset pagination convention. + let mut messages = Vec::new(); + let mut page_token = String::new(); + loop { + match client + .list_messages(nuncio_proto::v1::ListMessagesRequest { + folder_id: folder.to_string(), + page_size: 0, + page_token: page_token.clone(), + }) + .await + { + Ok(response) => { + let page = response.into_inner(); + messages.extend(page.messages); + if page.next_page_token.is_empty() { + break; + } + page_token = page.next_page_token; + } + Err(status) => { + return Self::render_error( + &format!( + "nunciod daemon rejected list_messages: {}", + Self::clean_status_message(&status) + ), + json_mode, + ); } } - Err(status) => Self::render_error( - &format!( - "nunciod daemon rejected list_messages: {}", - Self::clean_status_message(&status) - ), - json_mode, - ), + } + + if json_mode { + let messages_json: Vec = + messages.iter().map(message_proto_to_json).collect(); + format_json(&json!({ + "folder": folder, + "messages": messages_json + })) + } else { + format!("Folder '{}': {} message(s) found", folder, messages.len()) } } @@ -824,40 +839,55 @@ impl HeadlessRunner { Err(e) => return Self::render_error(&e, json_mode), }; - match client - .search_messages(nuncio_proto::v1::SearchMessagesRequest { - query: query.to_string(), - }) - .await - { - Ok(response) => { - let hits = response.into_inner().hits; - if json_mode { - let hits_json: Vec = hits - .iter() - .map(|h| { - json!({ - "id": h.id, - "title": h.title, - "snippet": h.snippet, - }) - }) - .collect(); - format_json(&json!({ - "query": query, - "results": hits_json - })) - } else { - format!("Search complete for '{}' ({} matches)", query, hits.len()) + // Follow `next_page_token` to the end so search returns every hit. + let mut hits = Vec::new(); + let mut page_token = String::new(); + loop { + match client + .search_messages(nuncio_proto::v1::SearchMessagesRequest { + query: query.to_string(), + page_size: 0, + page_token: page_token.clone(), + }) + .await + { + Ok(response) => { + let page = response.into_inner(); + hits.extend(page.hits); + if page.next_page_token.is_empty() { + break; + } + page_token = page.next_page_token; + } + Err(status) => { + return Self::render_error( + &format!( + "nunciod daemon rejected search_messages: {}", + Self::clean_status_message(&status) + ), + json_mode, + ); } } - Err(status) => Self::render_error( - &format!( - "nunciod daemon rejected search_messages: {}", - Self::clean_status_message(&status) - ), - json_mode, - ), + } + + if json_mode { + let hits_json: Vec = hits + .iter() + .map(|h| { + json!({ + "id": h.id, + "title": h.title, + "snippet": h.snippet, + }) + }) + .collect(); + format_json(&json!({ + "query": query, + "results": hits_json + })) + } else { + format!("Search complete for '{}' ({} matches)", query, hits.len()) } } @@ -869,36 +899,52 @@ impl HeadlessRunner { Err(e) => return Self::render_error(&e, json_mode), }; - match client - .list_folders(nuncio_proto::v1::ListFoldersRequest {}) - .await - { - Ok(response) => { - let folders = response.into_inner().folders; - if json_mode { - let folders_json: Vec = folders - .iter() - .map(|f| { - json!({ - "id": f.id, - "name": f.name, - "total_messages": f.total_messages, - "unread_messages": f.unread_messages, - }) - }) - .collect(); - format_json(&json!({ "folders": folders_json })) - } else { - format!("Available Mailbox Folders: {} folders found", folders.len()) + // Follow `next_page_token` to the end so every folder is listed. + let mut folders = Vec::new(); + let mut page_token = String::new(); + loop { + match client + .list_folders(nuncio_proto::v1::ListFoldersRequest { + page_size: 0, + page_token: page_token.clone(), + }) + .await + { + Ok(response) => { + let page = response.into_inner(); + folders.extend(page.folders); + if page.next_page_token.is_empty() { + break; + } + page_token = page.next_page_token; + } + Err(status) => { + return Self::render_error( + &format!( + "nunciod daemon rejected list_folders: {}", + Self::clean_status_message(&status) + ), + json_mode, + ); } } - Err(status) => Self::render_error( - &format!( - "nunciod daemon rejected list_folders: {}", - Self::clean_status_message(&status) - ), - json_mode, - ), + } + + if json_mode { + let folders_json: Vec = folders + .iter() + .map(|f| { + json!({ + "id": f.id, + "name": f.name, + "total_messages": f.total_messages, + "unread_messages": f.unread_messages, + }) + }) + .collect(); + format_json(&json!({ "folders": folders_json })) + } else { + format!("Available Mailbox Folders: {} folders found", folders.len()) } } @@ -1093,46 +1139,63 @@ impl HeadlessRunner { /// `system audit list`: a real thin gRPC client of the running /// `nunciod` daemon's `nuncio.v1.Audit` API. Lists a page of the /// daemon's real, persisted WORM audit ledger, sequence ascending. - async fn handle_audit_list(&self, limit: u32, offset: u32, json_mode: bool) -> String { + async fn handle_audit_list(&self, page_size: u32, json_mode: bool) -> String { let mut client = match self.connect_audit_client().await { Ok(client) => client, Err(e) => return Self::render_error(&e, json_mode), }; - match client - .list_records(nuncio_proto::v1::ListRecordsRequest { limit, offset }) - .await - { - Ok(response) => { - let records = response.into_inner().records; - if json_mode { - let records_json: Vec = - records.iter().map(audit_record_proto_to_json).collect(); - format_json(&json!({ "records": records_json })) - } else if records.is_empty() { - "No audit records recorded.".to_string() - } else { - let mut out = - String::from("SEQ ACTOR ACTION TIMESTAMP\n"); - for r in records { - out.push_str(&format!( - "{:<4} {:<20} {:<20} {}\n", - r.sequence, - r.actor, - r.action, - format_timestamp(&r.timestamp) - )); + // Follow `next_page_token` to the end so the whole ledger is listed, + // fetching `page_size` records per keyset page. + let mut records = Vec::new(); + let mut page_token = String::new(); + loop { + match client + .list_records(nuncio_proto::v1::ListRecordsRequest { + page_size, + page_token: page_token.clone(), + }) + .await + { + Ok(response) => { + let page = response.into_inner(); + records.extend(page.records); + if page.next_page_token.is_empty() { + break; } - out + page_token = page.next_page_token; + } + Err(status) => { + return Self::render_error( + &format!( + "nunciod daemon rejected list_records: {}", + Self::clean_status_message(&status) + ), + json_mode, + ); } } - Err(status) => Self::render_error( - &format!( - "nunciod daemon rejected list_records: {}", - Self::clean_status_message(&status) - ), - json_mode, - ), + } + + if json_mode { + let records_json: Vec = + records.iter().map(audit_record_proto_to_json).collect(); + format_json(&json!({ "records": records_json })) + } else if records.is_empty() { + "No audit records recorded.".to_string() + } else { + let mut out = + String::from("SEQ ACTOR ACTION TIMESTAMP\n"); + for r in records { + out.push_str(&format!( + "{:<4} {:<20} {:<20} {}\n", + r.sequence, + r.actor, + r.action, + format_timestamp(&r.timestamp) + )); + } + out } } @@ -1208,37 +1271,52 @@ impl HeadlessRunner { Err(e) => return Self::render_error(&e, json_mode), }; - match client - .list_rules(nuncio_proto::v1::ListRulesRequest {}) - .await - { - Ok(response) => { - let rules = response.into_inner().rules; - if json_mode { - let rules_json: Vec = - rules.iter().map(filter_rule_proto_to_json).collect(); - format_json(&json!({ "rules": rules_json })) - } else if rules.is_empty() { - "No filter rules configured.".to_string() - } else { - let mut out = - String::from("ID PRIORITY ENABLED NAME NSQL\n"); - for r in rules { - out.push_str(&format!( - "{:<10} {:<8} {:<7} {:<20} {}\n", - r.id, r.priority, r.enabled, r.name, r.nsql_text - )); + // Follow `next_page_token` to the end so every rule is listed. + let mut rules = Vec::new(); + let mut page_token = String::new(); + loop { + match client + .list_rules(nuncio_proto::v1::ListRulesRequest { + page_size: 0, + page_token: page_token.clone(), + }) + .await + { + Ok(response) => { + let page = response.into_inner(); + rules.extend(page.rules); + if page.next_page_token.is_empty() { + break; } - out + page_token = page.next_page_token; + } + Err(status) => { + return Self::render_error( + &format!( + "nunciod daemon rejected list_rules: {}", + Self::clean_status_message(&status) + ), + json_mode, + ); } } - Err(status) => Self::render_error( - &format!( - "nunciod daemon rejected list_rules: {}", - Self::clean_status_message(&status) - ), - json_mode, - ), + } + + if json_mode { + let rules_json: Vec = + rules.iter().map(filter_rule_proto_to_json).collect(); + format_json(&json!({ "rules": rules_json })) + } else if rules.is_empty() { + "No filter rules configured.".to_string() + } else { + let mut out = String::from("ID PRIORITY ENABLED NAME NSQL\n"); + for r in rules { + out.push_str(&format!( + "{:<10} {:<8} {:<7} {:<20} {}\n", + r.id, r.priority, r.enabled, r.name, r.nsql_text + )); + } + out } } @@ -2240,36 +2318,52 @@ impl HeadlessRunner { Err(e) => return Self::render_error(&e, json_mode), }; - match client - .list_events(nuncio_proto::v1::ListEventsRequest { - account_id: account.to_string(), - calendar_id: calendar.to_string(), - start_window: Some(nuncio_proto::time::timestamp_from_unix_secs(start)), - end_window: Some(nuncio_proto::time::timestamp_from_unix_secs(end)), - }) - .await - { - Ok(response) => { - let events = response.into_inner().events; - if json_mode { - let events_json: Vec = - events.iter().map(calendar_event_proto_to_json).collect(); - format_json(&json!({ - "account": account, - "calendar": calendar, - "events": events_json - })) - } else { - format!("Calendar Events: {} event(s) found", events.len()) + // Follow `next_page_token` to the end so every event in the window is + // listed, transparently exercising keyset pagination. + let mut events = Vec::new(); + let mut page_token = String::new(); + loop { + match client + .list_events(nuncio_proto::v1::ListEventsRequest { + account_id: account.to_string(), + calendar_id: calendar.to_string(), + start_window: Some(nuncio_proto::time::timestamp_from_unix_secs(start)), + end_window: Some(nuncio_proto::time::timestamp_from_unix_secs(end)), + page_size: 0, + page_token: page_token.clone(), + }) + .await + { + Ok(response) => { + let page = response.into_inner(); + events.extend(page.events); + if page.next_page_token.is_empty() { + break; + } + page_token = page.next_page_token; + } + Err(status) => { + return Self::render_error( + &format!( + "nunciod daemon rejected list_events: {}", + Self::clean_status_message(&status) + ), + json_mode, + ); } } - Err(status) => Self::render_error( - &format!( - "nunciod daemon rejected list_events: {}", - Self::clean_status_message(&status) - ), - json_mode, - ), + } + + if json_mode { + let events_json: Vec = + events.iter().map(calendar_event_proto_to_json).collect(); + format_json(&json!({ + "account": account, + "calendar": calendar, + "events": events_json + })) + } else { + format!("Calendar Events: {} event(s) found", events.len()) } } @@ -2348,35 +2442,59 @@ impl HeadlessRunner { Err(e) => return Self::render_error(&e, json_mode), }; - match client - .list_contacts(nuncio_proto::v1::ListContactsRequest { - account_id: account.to_string(), - }) - .await - { - Ok(response) => { - let contacts = response.into_inner().contacts; - if json_mode { - let contacts_json: Vec = - contacts.iter().map(contact_proto_to_json).collect(); - format_json(&json!({ "account": account, "contacts": contacts_json })) - } else { - format!( - "Contacts: {} contact(s) found in address book.", - contacts.len() - ) - } + let contacts = match self.fetch_all_contacts(&mut client, account).await { + Ok(contacts) => contacts, + Err(status) => { + return Self::render_error( + &format!( + "nunciod daemon rejected list_contacts: {}", + Self::clean_status_message(&status) + ), + json_mode, + ); } - Err(status) => Self::render_error( - &format!( - "nunciod daemon rejected list_contacts: {}", - Self::clean_status_message(&status) - ), - json_mode, - ), + }; + + if json_mode { + let contacts_json: Vec = + contacts.iter().map(contact_proto_to_json).collect(); + format_json(&json!({ "account": account, "contacts": contacts_json })) + } else { + format!( + "Contacts: {} contact(s) found in address book.", + contacts.len() + ) } } + /// Follows `next_page_token` to fetch every contact in `account`'s address + /// book over the keyset-paginated `ListContacts` RPC, shared by + /// [`Self::handle_contact_list`] and [`Self::handle_contact_search`]. + async fn fetch_all_contacts( + &self, + client: &mut nuncio_proto::client::AuthenticatedContactsClient, + account: &str, + ) -> Result, tonic::Status> { + let mut contacts = Vec::new(); + let mut page_token = String::new(); + loop { + let page = client + .list_contacts(nuncio_proto::v1::ListContactsRequest { + account_id: account.to_string(), + page_size: 0, + page_token: page_token.clone(), + }) + .await? + .into_inner(); + contacts.extend(page.contacts); + if page.next_page_token.is_empty() { + break; + } + page_token = page.next_page_token; + } + Ok(contacts) + } + /// `contact search`: fetches `account`'s full `ListContacts` result over /// the real gRPC API, then filters it client-side by `query` against /// display name, organization, and email address (case-insensitive) -- @@ -2387,42 +2505,37 @@ impl HeadlessRunner { Err(e) => return Self::render_error(&e, json_mode), }; - match client - .list_contacts(nuncio_proto::v1::ListContactsRequest { - account_id: account.to_string(), - }) - .await - { - Ok(response) => { - let matches: Vec<_> = response - .into_inner() - .contacts - .into_iter() - .filter(|c| contact_matches_query(c, query)) - .collect(); - if json_mode { - let contacts_json: Vec = - matches.iter().map(contact_proto_to_json).collect(); - format_json(&json!({ - "account": account, - "query": query, - "contacts": contacts_json - })) - } else { - format!( - "Search ('{}'): {} contact(s) matched.", - query, - matches.len() - ) - } + let contacts = match self.fetch_all_contacts(&mut client, account).await { + Ok(contacts) => contacts, + Err(status) => { + return Self::render_error( + &format!( + "nunciod daemon rejected list_contacts: {}", + Self::clean_status_message(&status) + ), + json_mode, + ); } - Err(status) => Self::render_error( - &format!( - "nunciod daemon rejected list_contacts: {}", - Self::clean_status_message(&status) - ), - json_mode, - ), + }; + + let matches: Vec<_> = contacts + .into_iter() + .filter(|c| contact_matches_query(c, query)) + .collect(); + if json_mode { + let contacts_json: Vec = + matches.iter().map(contact_proto_to_json).collect(); + format_json(&json!({ + "account": account, + "query": query, + "contacts": contacts_json + })) + } else { + format!( + "Search ('{}'): {} contact(s) matched.", + query, + matches.len() + ) } } @@ -3231,6 +3344,7 @@ mod tests { total_messages: 3, unread_messages: 1, }], + next_page_token: String::new(), })) } @@ -3243,6 +3357,7 @@ mod tests { message.folder_id = req.folder_id; Ok(tonic::Response::new(ListMessagesResponse { messages: vec![message], + next_page_token: String::new(), })) } @@ -3286,6 +3401,7 @@ mod tests { title: "Stub Subject".to_string(), snippet: format!("...{}...", req.query), }], + next_page_token: String::new(), })) } @@ -3693,6 +3809,7 @@ mod tests { .unwrap_or_else(|e| e.into_inner()) = Some(req); Ok(tonic::Response::new(ListEventsResponse { events: vec![stub_event()], + next_page_token: String::new(), })) } @@ -3875,6 +3992,7 @@ mod tests { stub_contact("ct-stub-1", "Alice Stub", "Kof22", "alice@nuncio.mx"), stub_contact("ct-stub-2", "Bob Other", "OtherCo", "bob@nuncio.mx"), ], + next_page_token: String::new(), })) } @@ -4114,6 +4232,7 @@ mod tests { 1_700_000_000, )), }], + next_page_token: String::new(), })) } @@ -4723,6 +4842,7 @@ mod tests { previous_block_hash: "GENESIS".to_string(), record_hmac: "cafebabe".to_string(), }], + next_page_token: String::new(), })) } @@ -4758,10 +4878,7 @@ mod tests { .execute_command( &Commands::System { action: SystemSubcommand::Audit { - action: AuditSubcommand::List { - limit: 0, - offset: 0, - }, + action: AuditSubcommand::List { page_size: 0 }, }, }, true, @@ -4824,10 +4941,7 @@ mod tests { .execute_command( &Commands::System { action: SystemSubcommand::Audit { - action: AuditSubcommand::List { - limit: 0, - offset: 0, - }, + action: AuditSubcommand::List { page_size: 0 }, }, }, true, diff --git a/crates/nuncio-proto/proto/nuncio/v1/descriptor.bin b/crates/nuncio-proto/proto/nuncio/v1/descriptor.bin index 94ba2c20c5f91ea86c5d255886932d3c91912fcc..828d294a8ece515d22673a031fc3e9e511ea69c8 100644 GIT binary patch delta 874 zcmaD_aJO^=BNNlD!p+P~K75P;lLG`q?1ZGbI13WfQ{#&>t5PKx6&N*mgE&9}!9c!{ zEEgA8vLrt{HBW*GD9afHRvrQpP|BK|Cg7wcq{hXcms(L0kD-(oBpd)V45B=5@+tvi zTNDdHM&h=R8QsFX$zKHAG>EluvX$Uv#>~llLb^7n-XzV|$<{)ljD?dY3V9N3^X6MZ zGnoVZxCDGMi%Wt^b5e_gQVUB{i%U@5L1J)cOzseKAl?lp#N4=;I+-R5D4s=&W>&nR z&W0K4_Z5i`_06hE&zTtuCiAM9pjk^=uq)+Fc2*<7+D^5dOu+$MLf{}zP0lY$p-Q0V QPj1(ACqB^6X-?+@04Q$=y8r+H delta 210 zcmcax`m|sJBNNl3oXyNkK75QKlLG`qn3S9*FA{KKQu3J0Cul4#!o`}CnVVT6!KA>b z!5akOD7j6p6Lg#WM({GD?&JePx|219Lm53M*9&`Y{vbS)SyYHiz$de~B&akewKyoX zur#%}M9E?DEpdma*=;kk%5!E$lgY2uO+X^C8g83UYwTp= sb>b2Nn~|EFUz9SrS4K`sNQ{duKP{~|73LTyPsx4qb#3>}ZaUNX0Hu#XlmGw# diff --git a/crates/nuncio-proto/proto/nuncio/v1/nuncio.proto b/crates/nuncio-proto/proto/nuncio/v1/nuncio.proto index 368af82..2024e64 100644 --- a/crates/nuncio-proto/proto/nuncio/v1/nuncio.proto +++ b/crates/nuncio-proto/proto/nuncio/v1/nuncio.proto @@ -375,11 +375,11 @@ message TestAccountConnectionResponse { // carved out for "just one more" unauthenticated service. See // `nunciod::grpc::serve_on_listener`. service Mail { - // ListFolders returns every mailbox folder known to the store, with - // total/unread message counts. + // ListFolders returns a keyset-paginated page of mailbox folders known to + // the store, folder id ascending, each with total/unread message counts. rpc ListFolders(ListFoldersRequest) returns (ListFoldersResponse); - // ListMessages returns messages in a single folder, newest first, capped - // at `limit`. + // ListMessages returns a keyset-paginated page of messages in a single + // folder, newest first. rpc ListMessages(ListMessagesRequest) returns (ListMessagesResponse); // GetMessage returns a single message by ID, including its (decrypted) // body. @@ -387,7 +387,8 @@ service Mail { // MarkRead updates a message's read/unread flag. rpc MarkRead(MarkReadRequest) returns (MarkReadResponse); // SearchMessages runs a full-text (FTS5) search over cached message - // subjects, senders, and bodies. + // subjects, senders, and bodies, returning a keyset-paginated page of hits + // ordered by relevance rank. rpc SearchMessages(SearchMessagesRequest) returns (SearchMessagesResponse); // SendMessage composes and sends a real outbound email over SMTP, built // from the configured account's SMTP endpoint @@ -462,26 +463,53 @@ message Folder { uint64 unread_messages = 4; } -// ListFoldersRequest currently carries no parameters. -message ListFoldersRequest {} +// ListFoldersRequest pages the folder listing (folder id ascending) using +// the shared keyset convention: `page_size` bounds a page and `page_token` +// is the opaque cursor returned as the previous page's `next_page_token`. +// See `ListMessagesRequest` for the full semantics. +message ListFoldersRequest { + // Maximum folders to return in this page; 0 means "server default". The + // server clamps this to its documented maximum. + uint32 page_size = 1; + // Opaque keyset cursor from a prior response's `next_page_token`; empty + // starts from the first page. Not an offset -- resumes strictly after the + // last folder of the previous page. + string page_token = 2; +} // ListFoldersResponse. message ListFoldersResponse { repeated Folder folders = 1; + // Opaque cursor for the next page, or empty when this is the last page. + string next_page_token = 2; } -// ListMessagesRequest. `limit` caps the number of messages returned -// (newest first); a value of 0 means "use the server's default limit". +// ListMessagesRequest pages a folder's messages, newest first, using the +// shared keyset pagination convention used by every list/search RPC: +// - `page_size` bounds one page; 0 selects the server default and the +// server clamps any value to its documented maximum. +// - `page_token` is an OPAQUE cursor, NOT a numeric offset. It encodes the +// stable sort key (here `(received_at, id)`) of the last row of the +// previous page, so paging resumes strictly AFTER that row and stays +// correct under concurrent inserts (no skipped or duplicated rows). +// Leave empty to fetch the first page; otherwise pass a prior response's +// `next_page_token` verbatim. A token the server cannot decode is +// rejected with a `VALIDATION_FAILED` error, never silently ignored. message ListMessagesRequest { // Mailbox folder identifier (e.g. "inbox"). string folder_id = 1; // Maximum number of messages to return; 0 means "server default". - uint32 limit = 2; + uint32 page_size = 2; + // Opaque keyset cursor from a prior response's `next_page_token`; empty + // starts from the newest message. + string page_token = 3; } // ListMessagesResponse. message ListMessagesResponse { repeated Message messages = 1; + // Opaque cursor for the next page, or empty when this is the last page. + string next_page_token = 2; } // GetMessageRequest. @@ -507,10 +535,19 @@ message MarkReadRequest { // response (as opposed to a `Status` error) is confirmation enough. message MarkReadResponse {} -// SearchMessagesRequest. +// SearchMessagesRequest. Pages hits using the shared keyset convention (see +// `ListMessagesRequest`); the cursor here encodes `(rank, id)` so paging +// preserves the relevance ordering while resuming strictly after the last +// hit of the previous page. message SearchMessagesRequest { // Full-text search query string. string query = 1; + // Maximum hits to return in this page; 0 means "server default". Clamped + // to the server's documented maximum. + uint32 page_size = 2; + // Opaque keyset cursor from a prior response's `next_page_token`; empty + // starts from the highest-ranked hit. + string page_token = 3; } // Mirrors `nuncio_store::search::SearchHit`. @@ -526,6 +563,8 @@ message MessageSearchHit { // SearchMessagesResponse. message SearchMessagesResponse { repeated MessageSearchHit hits = 1; + // Opaque cursor for the next page, or empty when this is the last page. + string next_page_token = 2; } // SendMessageRequest carries a composed outbound email. The sending @@ -595,7 +634,8 @@ service Filters { // Rejects a syntactically or semantically invalid rule with // `Status::invalid_argument` -- nothing is ever persisted on failure. rpc CreateRule(CreateRuleRequest) returns (CreateRuleResponse); - // ListRules returns every persisted filter rule, priority ascending. + // ListRules returns a keyset-paginated page of persisted filter rules, + // priority ascending. rpc ListRules(ListRulesRequest) returns (ListRulesResponse); // DeleteRule removes a persisted filter rule by ID and reloads the // daemon's live `FilterEngine` rule set so the removal takes effect @@ -690,12 +730,23 @@ message CreateRuleResponse { FilterRule rule = 1; } -// ListRulesRequest currently carries no parameters. -message ListRulesRequest {} +// ListRulesRequest pages rules (priority ascending, id as tiebreaker) using +// the shared keyset convention; the cursor encodes `(priority, id)`. See +// `ListMessagesRequest` for the full semantics. +message ListRulesRequest { + // Maximum rules to return in this page; 0 means "server default". Clamped + // to the server's documented maximum. + uint32 page_size = 1; + // Opaque keyset cursor from a prior response's `next_page_token`; empty + // starts from the highest-priority rule. + string page_token = 2; +} // ListRulesResponse. message ListRulesResponse { repeated FilterRule rules = 1; + // Opaque cursor for the next page, or empty when this is the last page. + string next_page_token = 2; } // DeleteRuleRequest. @@ -956,9 +1007,9 @@ service Calendar { // returns `Status::internal` describing that gap unless a test double has // been injected -- never a fabricated `synced_count`. rpc Sync(CalendarSyncRequest) returns (CalendarSyncResponse); - // ListEvents returns persisted events for `account_id` whose window - // overlaps `[start_window, end_window]`, ordered by start time -- always - // genuinely persisted data, never fabricated. + // ListEvents returns a keyset-paginated page of persisted events for + // `account_id` whose window overlaps `[start_window, end_window]`, ordered + // by start time -- always genuinely persisted data, never fabricated. rpc ListEvents(ListEventsRequest) returns (ListEventsResponse); // GetEvent returns a single persisted event by ID. rpc GetEvent(GetEventRequest) returns (GetEventResponse); @@ -1001,17 +1052,29 @@ message CalendarSyncResponse { uint64 synced_count = 1; } -// ListEventsRequest. +// ListEventsRequest pages events (start time ascending, id as tiebreaker) +// using the shared keyset convention; the cursor encodes `(start_time, id)`. +// Recurrence expansion happens server-side before paging, so an expanded +// occurrence's `(start_time, id)` is what the cursor tracks. See +// `ListMessagesRequest` for the full semantics. message ListEventsRequest { string account_id = 1; string calendar_id = 2; google.protobuf.Timestamp start_window = 3; google.protobuf.Timestamp end_window = 4; + // Maximum events to return in this page; 0 means "server default". Clamped + // to the server's documented maximum. + uint32 page_size = 5; + // Opaque keyset cursor from a prior response's `next_page_token`; empty + // starts from the earliest event in the window. + string page_token = 6; } // ListEventsResponse. message ListEventsResponse { repeated CalendarEvent events = 1; + // Opaque cursor for the next page, or empty when this is the last page. + string next_page_token = 2; } // GetEventRequest. @@ -1052,9 +1115,9 @@ service Contacts { // describing that gap unless a test double has been injected -- never a // fabricated `synced_count`. rpc Sync(ContactsSyncRequest) returns (ContactsSyncResponse); - // ListContacts returns persisted contacts for `account_id`, ordered by - // interaction count then display name -- always genuinely persisted data, - // never fabricated. + // ListContacts returns a keyset-paginated page of persisted contacts for + // `account_id`, ordered by interaction count (descending) then display + // name -- always genuinely persisted data, never fabricated. rpc ListContacts(ListContactsRequest) returns (ListContactsResponse); // GetContact returns a single persisted contact by ID. rpc GetContact(GetContactRequest) returns (GetContactResponse); @@ -1132,14 +1195,24 @@ message ContactsSyncResponse { // ListContactsRequest. Contacts are always scoped to the account whose // address book they belong to, mirroring the store's `list_contacts` -// signature. +// signature. Pages using the shared keyset convention; the cursor encodes +// `(interaction_count, display_name, id)` to match the sort order. See +// `ListMessagesRequest` for the full semantics. message ListContactsRequest { string account_id = 1; + // Maximum contacts to return in this page; 0 means "server default". + // Clamped to the server's documented maximum. + uint32 page_size = 2; + // Opaque keyset cursor from a prior response's `next_page_token`; empty + // starts from the most-interacted contact. + string page_token = 3; } // ListContactsResponse. message ListContactsResponse { repeated Contact contacts = 1; + // Opaque cursor for the next page, or empty when this is the last page. + string next_page_token = 2; } // GetContactRequest. @@ -1183,8 +1256,8 @@ message CreateContactResponse { // `BearerAuthInterceptor` that guards every other service `nunciod` // exposes -- see `nunciod::grpc::serve_on_listener`. service Audit { - // ListRecords returns a page of the real, persisted WORM audit ledger, - // sequence ascending. + // ListRecords returns a keyset-paginated page of the real, persisted WORM + // audit ledger, sequence ascending. rpc ListRecords(ListRecordsRequest) returns (ListRecordsResponse); // VerifyChain re-verifies the ENTIRE persisted WORM audit ledger's // HMAC hash-chain integrity, using the ledger's real WORM HMAC key @@ -1215,17 +1288,25 @@ message AuditRecord { string record_hmac = 7; } -// ListRecordsRequest. `limit` caps the number of records returned -// (sequence ascending); a value of 0 means "use the server's default -// limit". +// ListRecordsRequest pages the WORM audit ledger, sequence ascending, using +// the shared keyset convention that every list/search RPC uses -- replacing +// the previous limit/offset paging, whose offset was unstable under +// concurrent appends. The cursor encodes the last record's `sequence`. See +// `ListMessagesRequest` for the full semantics. message ListRecordsRequest { - uint32 limit = 1; - uint32 offset = 2; + // Maximum records to return in this page; 0 means "server default". + // Clamped to the server's documented maximum. + uint32 page_size = 1; + // Opaque keyset cursor from a prior response's `next_page_token`; empty + // starts from the lowest sequence. + string page_token = 2; } // ListRecordsResponse. message ListRecordsResponse { repeated AuditRecord records = 1; + // Opaque cursor for the next page, or empty when this is the last page. + string next_page_token = 2; } // VerifyChainRequest currently carries no parameters. diff --git a/crates/nuncio-store/src/db.rs b/crates/nuncio-store/src/db.rs index 7cba21a..c58912c 100644 --- a/crates/nuncio-store/src/db.rs +++ b/crates/nuncio-store/src/db.rs @@ -1291,6 +1291,102 @@ impl DatabaseEngine { .collect::, DatabaseError>>() } + /// Keyset-paginated listing of a folder's messages, newest first + /// (`received_at DESC, id DESC`). `after` is the `(received_at, id)` of + /// the last message of the previous page; `None` starts from the newest. + /// Fetches `page_size + 1` rows to detect a following page: returns at + /// most `page_size` messages plus the `(received_at, id)` cursor of the + /// last returned message when more remain (else `None`). + #[allow(clippy::type_complexity)] + pub async fn list_messages_page( + &self, + folder_id: &str, + after: Option<(i64, String)>, + page_size: usize, + ) -> Result<(Vec, Option<(i64, String)>), DatabaseError> { + let fetch = page_size.saturating_add(1); + let mut builder: sqlx::QueryBuilder<'_, sqlx::Sqlite> = sqlx::QueryBuilder::new( + "SELECT id, account_id, folder_id, subject, sender, recipient, received_at, read_flag, body_plain, body_html, remote_id, uid_validity FROM messages WHERE folder_id = ", + ); + builder.push_bind(folder_id.to_string()); + if let Some((ts, id)) = &after { + builder.push(" AND (received_at < "); + builder.push_bind(*ts); + builder.push(" OR (received_at = "); + builder.push_bind(*ts); + builder.push(" AND id < "); + builder.push_bind(id.clone()); + builder.push("))"); + } + builder.push(" ORDER BY received_at DESC, id DESC LIMIT "); + builder.push_bind(fetch as i64); + + let rows = builder + .build_query_as::<( + String, + String, + String, + String, + String, + String, + i64, + i64, + Option, + Option, + String, + String, + )>() + .fetch_all(&self.pool) + .await + .map_err(DatabaseError::Query)?; + + let has_more = rows.len() > page_size; + let mut emails = rows + .into_iter() + .take(page_size) + .map(|r| { + let dec_plain = + r.8.map(|p| { + crate::cipher::PayloadCipher::decrypt_text_at_rest(&self.storage_key, &p) + }) + .transpose() + .map_err(|e| DatabaseError::Decryption(e.to_string()))?; + let dec_html = + r.9.map(|h| { + crate::cipher::PayloadCipher::decrypt_text_at_rest(&self.storage_key, &h) + }) + .transpose() + .map_err(|e| DatabaseError::Decryption(e.to_string()))?; + Ok(nuncio_core::model::Email { + id: r.0, + account_id: r.1, + folder_id: r.2, + remote_id: r.10, + uid_validity: r.11, + subject: r.3, + sender: r.4, + recipient: r.5, + received_at: r.6, + read: r.7 != 0, + body_plain: dec_plain, + body_html: dec_html, + attachments: Vec::new(), + }) + }) + .collect::, DatabaseError>>()?; + + let next = if has_more { + emails + .last() + .map(|e| (e.received_at, e.id.clone())) + .filter(|_| !emails.is_empty()) + } else { + None + }; + emails.shrink_to_fit(); + Ok((emails, next)) + } + /// Retrieve a single message by ID. #[allow(clippy::type_complexity)] pub async fn get_message( @@ -1632,6 +1728,69 @@ impl DatabaseEngine { Ok(rows.iter().map(contact_from_row).collect()) } + /// Keyset-paginated listing of `account_id`'s contacts, in the same order + /// as [`Self::list_contacts`] (`interaction_count DESC, display_name ASC, + /// id ASC`). `after` is the `(interaction_count, display_name, id)` of the + /// last contact of the previous page. Fetches `page_size + 1` rows to + /// detect a following page and returns the `(interaction_count, + /// display_name, id)` cursor of the last returned contact when more remain. + /// + /// See the confidentiality/dependency note on [`Self::save_contact`]. + #[allow(clippy::type_complexity)] + pub async fn list_contacts_page( + &self, + account_id: &str, + after: Option<(i64, String, String)>, + page_size: usize, + ) -> Result<(Vec, Option<(i64, String, String)>), DatabaseError> { + let fetch = page_size.saturating_add(1); + let mut builder: sqlx::QueryBuilder<'_, sqlx::Sqlite> = sqlx::QueryBuilder::new( + "SELECT id, account_id, display_name, given_name, family_name, organization, \ + job_title, notes, avatar_url, emails_json, phones_json, is_favorite, \ + interaction_count, last_interacted_at, created_at, updated_at \ + FROM contacts WHERE account_id = ", + ); + builder.push_bind(account_id.to_string()); + if let Some((ic, dn, id)) = &after { + builder.push(" AND (interaction_count < "); + builder.push_bind(*ic); + builder.push(" OR (interaction_count = "); + builder.push_bind(*ic); + builder.push(" AND (display_name > "); + builder.push_bind(dn.clone()); + builder.push(" OR (display_name = "); + builder.push_bind(dn.clone()); + builder.push(" AND id > "); + builder.push_bind(id.clone()); + builder.push("))))"); + } + builder.push(" ORDER BY interaction_count DESC, display_name ASC, id ASC LIMIT "); + builder.push_bind(fetch as i64); + + let rows = builder + .build() + .fetch_all(&self.pool) + .await + .map_err(DatabaseError::Query)?; + + let has_more = rows.len() > page_size; + let contacts: Vec = + rows.iter().take(page_size).map(contact_from_row).collect(); + + let next = if has_more { + contacts.last().map(|c| { + ( + c.interaction_count as i64, + c.display_name.clone(), + c.id.clone(), + ) + }) + } else { + None + }; + Ok((contacts, next)) + } + /// Retrieve a single contact by ID. /// /// See the confidentiality/dependency note on [`Self::save_contact`]. @@ -1704,6 +1863,52 @@ impl DatabaseEngine { .collect()) } + /// Keyset-paginated listing of folders with message counts, folder id + /// ascending. `after` is the id of the last folder of the previous page. + /// Fetches `page_size + 1` groups to detect a following page and returns + /// the id of the last returned folder as the cursor when more remain. + pub async fn list_folders_page( + &self, + after: Option, + page_size: usize, + ) -> Result<(Vec, Option), DatabaseError> { + let fetch = page_size.saturating_add(1); + let mut builder: sqlx::QueryBuilder<'_, sqlx::Sqlite> = sqlx::QueryBuilder::new( + "SELECT folder_id, COUNT(*) as total, SUM(CASE WHEN read_flag = 0 THEN 1 ELSE 0 END) as unread FROM messages ", + ); + if let Some(id) = &after { + builder.push("WHERE folder_id > "); + builder.push_bind(id.clone()); + } + builder.push(" GROUP BY folder_id ORDER BY folder_id ASC LIMIT "); + builder.push_bind(fetch as i64); + + let rows = builder + .build_query_as::<(String, i64, i64)>() + .fetch_all(&self.pool) + .await + .map_err(DatabaseError::Query)?; + + let has_more = rows.len() > page_size; + let folders: Vec = rows + .into_iter() + .take(page_size) + .map(|(folder_id, total, unread)| nuncio_core::model::Folder { + id: folder_id.clone(), + name: folder_id, + total_messages: total as usize, + unread_messages: unread as usize, + }) + .collect(); + + let next = if has_more { + folders.last().map(|f| f.id.clone()) + } else { + None + }; + Ok((folders, next)) + } + /// Save or replace a [`nuncio_filter::FilterRule`]. pub async fn save_filter_rule( &self, @@ -1772,6 +1977,77 @@ impl DatabaseEngine { Ok(rules) } + /// Keyset-paginated listing of persisted filter rules, `priority ASC, id + /// ASC`. `after` is the `(priority, id)` of the last rule of the previous + /// page. Fetches `page_size + 1` rows to detect a following page. + /// + /// The following-page detection and the returned cursor are computed from + /// the RAW rows (their stored `(priority, id)`), before parsing -- a row + /// whose stored NSQL no longer parses is skipped from the returned rules + /// exactly as [`Self::list_filter_rules`] does, but it still counts toward + /// the page boundary, so a page with skipped rows still advances the + /// cursor correctly and never re-emits or gaps around a rule. + #[allow(clippy::type_complexity)] + pub async fn list_filter_rules_page( + &self, + after: Option<(i32, String)>, + page_size: usize, + ) -> Result<(Vec, Option<(i32, String)>), DatabaseError> { + let fetch = page_size.saturating_add(1); + let mut builder: sqlx::QueryBuilder<'_, sqlx::Sqlite> = sqlx::QueryBuilder::new( + "SELECT id, name, priority, enabled, nsql_text, created_at, updated_at FROM filter_rules ", + ); + if let Some((priority, id)) = &after { + builder.push("WHERE (priority > "); + builder.push_bind(*priority as i64); + builder.push(" OR (priority = "); + builder.push_bind(*priority as i64); + builder.push(" AND id > "); + builder.push_bind(id.clone()); + builder.push("))"); + } + builder.push(" ORDER BY priority ASC, id ASC LIMIT "); + builder.push_bind(fetch as i64); + + let rows = builder + .build_query_as::<(String, String, i64, i64, String, i64, i64)>() + .fetch_all(&self.pool) + .await + .map_err(DatabaseError::Query)?; + + let has_more = rows.len() > page_size; + let page: Vec<(String, String, i64, i64, String, i64, i64)> = + rows.into_iter().take(page_size).collect(); + + let next = if has_more { + page.last().map(|r| (r.2 as i32, r.0.clone())) + } else { + None + }; + + let mut rules = Vec::new(); + for (id, name, priority, enabled, nsql_text, created_at, updated_at) in page { + match nuncio_filter::NsqlParser::parse_rule(&name, priority as i32, &nsql_text) { + Ok(mut parsed) => { + parsed.id = id; + parsed.enabled = enabled != 0; + parsed.created_at = created_at; + parsed.updated_at = updated_at; + rules.push(parsed); + } + Err(err) => { + tracing::warn!( + rule_id = %id, + rule_name = %name, + error = %err, + "stored filter rule failed to parse and will not be enforced" + ); + } + } + } + Ok((rules, next)) + } + /// Delete a [`nuncio_filter::FilterRule`] by ID. pub async fn delete_filter_rule(&self, rule_id: &str) -> Result<(), DatabaseError> { sqlx::query("DELETE FROM filter_rules WHERE id = ?") @@ -2261,6 +2537,57 @@ impl DatabaseEngine { .collect()) } + /// Keyset-paginated listing of the WORM audit ledger, `sequence ASC`. + /// `after` is the sequence of the last record of the previous page; + /// `None` starts from the lowest sequence. Fetches `page_size + 1` rows to + /// detect a following page and returns the sequence of the last returned + /// record as the cursor when more remain. Keyset paging on the + /// monotonically increasing sequence is stable under concurrent appends, + /// unlike the previous limit/offset paging. + pub async fn list_worm_audit_records_page( + &self, + after: Option, + page_size: usize, + ) -> Result<(Vec, Option), DatabaseError> { + let fetch = page_size.saturating_add(1); + let mut builder: sqlx::QueryBuilder<'_, sqlx::Sqlite> = sqlx::QueryBuilder::new( + "SELECT sequence, timestamp_ns, actor, action, data_hash, previous_block_hash, record_hmac FROM worm_audit_records ", + ); + if let Some(seq) = after { + builder.push("WHERE sequence > "); + builder.push_bind(seq as i64); + } + builder.push(" ORDER BY sequence ASC LIMIT "); + builder.push_bind(fetch as i64); + + let rows = builder + .build_query_as::<(i64, i64, String, String, String, String, String)>() + .fetch_all(&self.pool) + .await?; + + let has_more = rows.len() > page_size; + let records: Vec = rows + .into_iter() + .take(page_size) + .map(|r| nuncio_core::WormAuditRecord { + sequence: r.0 as u64, + timestamp_ns: r.1, + actor: r.2, + action: r.3, + data_hash: r.4, + previous_block_hash: r.5, + record_hmac: r.6, + }) + .collect(); + + let next = if has_more { + records.last().map(|r| r.sequence) + } else { + None + }; + Ok((records, next)) + } + /// Verify the entire WORM cryptographic audit log chain, using the WORM HMAC key /// provisioned for this engine. pub async fn verify_worm_audit_chain(&self) -> Result<(), DatabaseError> { diff --git a/crates/nuncio-store/src/search.rs b/crates/nuncio-store/src/search.rs index 0befdf5..dc25f64 100644 --- a/crates/nuncio-store/src/search.rs +++ b/crates/nuncio-store/src/search.rs @@ -68,6 +68,71 @@ impl<'a> SearchEngine<'a> { .collect()) } + /// Keyset-paginated full-text search over message subjects, senders, and + /// bodies, ordered by relevance (`rank ASC, id ASC`). `after` is the + /// `(rank, id)` of the last hit of the previous page; `None` starts from + /// the highest-ranked hit. Fetches `page_size + 1` hits to detect a + /// following page and returns the `(rank, id)` cursor of the last returned + /// hit when more remain. + /// + /// The bm25 `rank` is materialized in an inner query (where the FTS5 + /// `MATCH` is in scope) into an ordinary column, so the outer query can + /// both keyset-filter and order by `(rank, id)` -- preserving relevance + /// order across pages while resuming strictly after the previous page's + /// last hit. + pub async fn search_messages_page( + &self, + query: &str, + after: Option<(f64, String)>, + page_size: usize, + ) -> Result<(Vec, Option<(f64, String)>), DatabaseError> { + let clean_query = Self::sanitize_fts5_query(query); + if clean_query.is_empty() { + return Ok((Vec::new(), None)); + } + let fetch = page_size.saturating_add(1); + + let mut builder: sqlx::QueryBuilder<'_, sqlx::Sqlite> = sqlx::QueryBuilder::new( + "SELECT id, title, snippet, rk FROM ( \ + SELECT id, subject AS title, \ + snippet(messages_fts, 3, '', '', '...', 10) AS snippet, rank AS rk \ + FROM messages_fts WHERE messages_fts MATCH ", + ); + builder.push_bind(clean_query); + builder.push(") "); + if let Some((rank, id)) = &after { + builder.push("WHERE (rk > "); + builder.push_bind(*rank); + builder.push(" OR (rk = "); + builder.push_bind(*rank); + builder.push(" AND id > "); + builder.push_bind(id.clone()); + builder.push("))"); + } + builder.push(" ORDER BY rk ASC, id ASC LIMIT "); + builder.push_bind(fetch as i64); + + let rows = builder + .build_query_as::<(String, String, String, f64)>() + .fetch_all(self.db.pool()) + .await + .map_err(DatabaseError::Query)?; + + let has_more = rows.len() > page_size; + let mut last_key: Option<(f64, String)> = None; + let hits: Vec = rows + .into_iter() + .take(page_size) + .map(|(id, title, snippet, rk)| { + last_key = Some((rk, id.clone())); + SearchHit { id, title, snippet } + }) + .collect(); + + let next = if has_more { last_key } else { None }; + Ok((hits, next)) + } + /// Perform a full-text trigram search over calendar event summaries and locations. pub async fn search_events(&self, query: &str) -> Result, DatabaseError> { let clean_query = Self::sanitize_fts5_query(query); @@ -224,6 +289,52 @@ mod tests { assert_eq!(hits[0].title, "Legacy Roadmap Notes"); } + /// Paging FTS search results across multiple pages via the `(rank, id)` + /// keyset must return every matching hit EXACTLY ONCE, no dupes and no + /// gaps -- proving the rank-materializing subquery keyset works end to + /// end over the real encrypt-then-index write path. + #[tokio::test] + async fn search_messages_page_returns_every_hit_once_no_gaps() { + let (db, _dir) = DatabaseEngine::connect_ephemeral().await.unwrap(); + for i in 0..5 { + db.save_email(&sample_email( + &format!("msg-{i}"), + "Budget Planning Meeting", + "annual budget revenue forecast discussion", + )) + .await + .unwrap(); + } + + let search = SearchEngine::new(&db); + let mut seen = Vec::new(); + let mut after: Option<(f64, String)> = None; + let mut pages = 0; + loop { + let (hits, next) = search + .search_messages_page("budget", after.clone(), 2) + .await + .unwrap(); + pages += 1; + assert!(hits.len() <= 2, "page must not exceed page_size"); + seen.extend(hits.iter().map(|h| h.id.clone())); + match next { + Some(cursor) => after = Some(cursor), + None => break, + } + assert!(pages < 100, "pagination must terminate"); + } + + seen.sort(); + seen.dedup(); + assert_eq!( + seen.len(), + 5, + "every matching message returned exactly once" + ); + assert!(pages >= 3, "5 hits at page_size 2 must span multiple pages"); + } + #[tokio::test] async fn fts5_event_search_and_triggers() { let (db, _dir) = DatabaseEngine::connect_ephemeral().await.unwrap(); diff --git a/crates/nunciod/src/grpc.rs b/crates/nunciod/src/grpc.rs index 7ac4718..234b4e1 100644 --- a/crates/nunciod/src/grpc.rs +++ b/crates/nunciod/src/grpc.rs @@ -5,6 +5,7 @@ //! //! This is the daemon's sole client-facing transport. +use crate::pagination::{self, CursorField}; use nuncio_cal::CalendarBackend; use nuncio_contacts::ContactsBackend; use nuncio_core::{CoreCommand, CoreEvent, EventBus}; @@ -745,10 +746,6 @@ impl Accounts for AccountsGrpcService { } } -/// Default cap on messages returned by `ListMessages` when the caller -/// supplies `limit: 0` ("use the server default"). -const DEFAULT_LIST_MESSAGES_LIMIT: usize = 50; - /// Maps a `nuncio_core::model::Attachment` onto its wire-format /// `nuncio.v1.Attachment` representation. fn map_attachment_to_proto(attachment: nuncio_core::model::Attachment) -> AttachmentProto { @@ -1083,6 +1080,11 @@ impl Calendar for CalendarGrpcService { } let start_window = require_window_bound(req.start_window, "start_window")?; let end_window = require_window_bound(req.end_window, "end_window")?; + let page_size = pagination::clamp_page_size(req.page_size); + let after = pagination::parse_token(&req.page_token, |f| match f { + [CursorField::I(ts), CursorField::S(id)] => Some((*ts, id.clone())), + _ => None, + })?; let all_in_window = self .db @@ -1101,12 +1103,45 @@ impl Calendar for CalendarGrpcService { events.retain(|e| req.calendar_id.is_empty() || e.calendar_id == req.calendar_id); + // Recurrence expansion materializes the result set in memory, so keyset + // pagination is applied to the sorted expanded list here rather than at + // the store. Each occurrence has a distinct `(start_time, id)` (an + // occurrence shares its master's id but never its start time), so the + // pair is a stable, unique keyset over the whole set. + events.sort_by(|a, b| { + a.start_time + .cmp(&b.start_time) + .then_with(|| a.id.cmp(&b.id)) + }); + if let Some((ts, id)) = &after { + events.retain(|e| e.start_time > *ts || (e.start_time == *ts && e.id > *id)); + } + + let has_more = events.len() > page_size; + events.truncate(page_size); + let next_page_token = if has_more { + events + .last() + .map(|e| { + pagination::encode_cursor(&[ + CursorField::I(e.start_time), + CursorField::S(e.id.clone()), + ]) + }) + .unwrap_or_default() + } else { + String::new() + }; + let events = events .into_iter() .map(map_calendar_event_to_proto) .collect(); - Ok(Response::new(ListEventsResponse { events })) + Ok(Response::new(ListEventsResponse { + events, + next_page_token, + })) } async fn get_event( @@ -1212,17 +1247,35 @@ impl Contacts for ContactsGrpcService { if req.account_id.is_empty() { return Err(Status::invalid_argument("account_id is required")); } + let page_size = pagination::clamp_page_size(req.page_size); + let after = pagination::parse_token(&req.page_token, |f| match f { + [CursorField::I(ic), CursorField::S(dn), CursorField::S(id)] => { + Some((*ic, dn.clone(), id.clone())) + } + _ => None, + })?; - let contacts = self + let (contacts, next) = self .db - .list_contacts(&req.account_id) + .list_contacts_page(&req.account_id, after, page_size) .await - .map_err(|e| Status::internal(format!("failed to list contacts: {e}")))? - .into_iter() - .map(map_contact_to_proto) - .collect(); + .map_err(|e| Status::internal(format!("failed to list contacts: {e}")))?; - Ok(Response::new(ListContactsResponse { contacts })) + let next_page_token = next + .map(|(ic, dn, id)| { + pagination::encode_cursor(&[ + CursorField::I(ic), + CursorField::S(dn), + CursorField::S(id), + ]) + }) + .unwrap_or_default(); + let contacts = contacts.into_iter().map(map_contact_to_proto).collect(); + + Ok(Response::new(ListContactsResponse { + contacts, + next_page_token, + })) } async fn get_contact( @@ -1364,18 +1417,30 @@ struct MailGrpcService { impl Mail for MailGrpcService { async fn list_folders( &self, - _request: Request, + request: Request, ) -> Result, Status> { - let folders = self - .db - .list_folders() - .await - .map_err(|e| Status::internal(format!("failed to list folders: {e}")))? - .into_iter() - .map(map_folder_to_proto) - .collect(); + let req = request.into_inner(); + let page_size = pagination::clamp_page_size(req.page_size); + let after = pagination::parse_token(&req.page_token, |f| match f { + [CursorField::S(id)] => Some(id.clone()), + _ => None, + })?; - Ok(Response::new(ListFoldersResponse { folders })) + let (folders, next) = self + .db + .list_folders_page(after, page_size) + .await + .map_err(|e| Status::internal(format!("failed to list folders: {e}")))?; + + let next_page_token = next + .map(|id| pagination::encode_cursor(&[CursorField::S(id)])) + .unwrap_or_default(); + let folders = folders.into_iter().map(map_folder_to_proto).collect(); + + Ok(Response::new(ListFoldersResponse { + folders, + next_page_token, + })) } async fn list_messages( @@ -1386,22 +1451,27 @@ impl Mail for MailGrpcService { if req.folder_id.is_empty() { return Err(Status::invalid_argument("folder_id is required")); } - let limit = if req.limit == 0 { - DEFAULT_LIST_MESSAGES_LIMIT - } else { - req.limit as usize - }; + let page_size = pagination::clamp_page_size(req.page_size); + let after = pagination::parse_token(&req.page_token, |f| match f { + [CursorField::I(ts), CursorField::S(id)] => Some((*ts, id.clone())), + _ => None, + })?; - let messages = self + let (messages, next) = self .db - .list_messages(&req.folder_id, limit) + .list_messages_page(&req.folder_id, after, page_size) .await - .map_err(|e| Status::internal(format!("failed to list messages: {e}")))? - .into_iter() - .map(map_email_to_proto) - .collect(); + .map_err(|e| Status::internal(format!("failed to list messages: {e}")))?; - Ok(Response::new(ListMessagesResponse { messages })) + let next_page_token = next + .map(|(ts, id)| pagination::encode_cursor(&[CursorField::I(ts), CursorField::S(id)])) + .unwrap_or_default(); + let messages = messages.into_iter().map(map_email_to_proto).collect(); + + Ok(Response::new(ListMessagesResponse { + messages, + next_page_token, + })) } async fn get_message( @@ -1467,11 +1537,24 @@ impl Mail for MailGrpcService { request: Request, ) -> Result, Status> { let req = request.into_inner(); + let page_size = pagination::clamp_page_size(req.page_size); + let after = pagination::parse_token(&req.page_token, |f| match f { + [CursorField::F(rank), CursorField::S(id)] => Some((*rank, id.clone())), + _ => None, + })?; + let search = SearchEngine::new(&self.db); - let hits = search - .search_messages(&req.query) + let (hits, next) = search + .search_messages_page(&req.query, after, page_size) .await - .map_err(|e| Status::internal(format!("search failed: {e}")))? + .map_err(|e| Status::internal(format!("search failed: {e}")))?; + + let next_page_token = next + .map(|(rank, id)| { + pagination::encode_cursor(&[CursorField::F(rank), CursorField::S(id)]) + }) + .unwrap_or_default(); + let hits = hits .into_iter() .map(|hit| MessageSearchHit { id: hit.id, @@ -1480,7 +1563,10 @@ impl Mail for MailGrpcService { }) .collect(); - Ok(Response::new(SearchMessagesResponse { hits })) + Ok(Response::new(SearchMessagesResponse { + hits, + next_page_token, + })) } /// SendMessage composes and sends a real outbound email over SMTP via @@ -1753,18 +1839,32 @@ impl Filters for FiltersGrpcService { async fn list_rules( &self, - _request: Request, + request: Request, ) -> Result, Status> { - let rules = self - .db - .list_filter_rules() - .await - .map_err(|e| Status::internal(format!("failed to list filter rules: {e}")))? - .into_iter() - .map(map_filter_rule_to_proto) - .collect(); + let req = request.into_inner(); + let page_size = pagination::clamp_page_size(req.page_size); + let after = pagination::parse_token(&req.page_token, |f| match f { + [CursorField::I(priority), CursorField::S(id)] => Some((*priority as i32, id.clone())), + _ => None, + })?; - Ok(Response::new(ListRulesResponse { rules })) + let (rules, next) = self + .db + .list_filter_rules_page(after, page_size) + .await + .map_err(|e| Status::internal(format!("failed to list filter rules: {e}")))?; + + let next_page_token = next + .map(|(priority, id)| { + pagination::encode_cursor(&[CursorField::I(priority as i64), CursorField::S(id)]) + }) + .unwrap_or_default(); + let rules = rules.into_iter().map(map_filter_rule_to_proto).collect(); + + Ok(Response::new(ListRulesResponse { + rules, + next_page_token, + })) } async fn delete_rule( @@ -2178,8 +2278,6 @@ impl Export for ExportGrpcService { /// Default cap on records returned by `Audit/ListRecords` when the caller /// supplies `limit: 0` ("use the server default"). -const DEFAULT_LIST_RECORDS_LIMIT: u32 = 100; - /// Maps a `nuncio_core::WormAuditRecord` onto its wire-format /// `nuncio.v1.AuditRecord` representation. `record_hmac` is a verification /// MAC output, never the WORM HMAC signing key itself -- see the message's @@ -2214,22 +2312,27 @@ impl Audit for AuditGrpcService { request: Request, ) -> Result, Status> { let req = request.into_inner(); - let limit = if req.limit == 0 { - DEFAULT_LIST_RECORDS_LIMIT - } else { - req.limit - }; + let page_size = pagination::clamp_page_size(req.page_size); + let after = pagination::parse_token(&req.page_token, |f| match f { + [CursorField::U(seq)] => Some(*seq), + _ => None, + })?; - let records = self + let (records, next) = self .db - .list_worm_audit_records(limit, req.offset) + .list_worm_audit_records_page(after, page_size) .await - .map_err(|e| Status::internal(format!("failed to list audit records: {e}")))? - .into_iter() - .map(map_audit_record_to_proto) - .collect(); + .map_err(|e| Status::internal(format!("failed to list audit records: {e}")))?; - Ok(Response::new(ListRecordsResponse { records })) + let next_page_token = next + .map(|seq| pagination::encode_cursor(&[CursorField::U(seq)])) + .unwrap_or_default(); + let records = records.into_iter().map(map_audit_record_to_proto).collect(); + + Ok(Response::new(ListRecordsResponse { + records, + next_page_token, + })) } /// Re-verifies the ENTIRE persisted WORM audit ledger via @@ -3721,6 +3824,110 @@ mod tests { } } + /// Paging a folder larger than `page_size` across multiple pages via + /// `next_page_token` must return every message EXACTLY ONCE, with no + /// duplicates and no gaps, and in the stable newest-first keyset order + /// (`received_at DESC, id DESC`). Some messages deliberately share a + /// `received_at` so the `(received_at, id)` tiebreaker is exercised across + /// a page boundary. + #[tokio::test] + async fn list_messages_keyset_pagination_returns_every_message_once_no_gaps() { + let (db, _dir) = DatabaseEngine::connect_ephemeral() + .await + .expect("connect ephemeral test db"); + let seed = [ + ("m1", 1_700_000_000_i64), + ("m2", 1_700_000_000), + ("m3", 1_700_000_010), + ("m4", 1_700_000_010), + ("m5", 1_700_000_020), + ("m6", 1_700_000_030), + ("m7", 1_700_000_030), + ]; + for (id, ts) in seed { + let mut email = sample_email(id, "inbox", "subject", "body"); + email.received_at = ts; + db.save_email(&email).await.expect("seed message"); + } + + let event_bus = Arc::new(EventBus::new()); + let secrets = Arc::new(SecretManager::mock()); + let filter_engine = Arc::new(FilterEngine::new(Vec::new()).expect("empty rule set")); + let (addr, _handle) = spawn_test_server_with( + event_bus, + Arc::new(db), + filter_engine, + secrets, + "correct-token", + ) + .await; + let mut client = MailClient::connect(format!("http://{addr}")) + .await + .expect("client connects"); + + let mut seen = Vec::new(); + let mut page_token = String::new(); + let mut pages = 0; + loop { + let resp = client + .list_messages(authed_bearer_request(ListMessagesRequest { + folder_id: "inbox".to_string(), + page_size: 2, + page_token: page_token.clone(), + })) + .await + .expect("list_messages page succeeds") + .into_inner(); + pages += 1; + assert!( + resp.messages.len() <= 2, + "a page must never exceed the requested page_size" + ); + seen.extend(resp.messages.iter().map(|m| m.id.clone())); + if resp.next_page_token.is_empty() { + break; + } + page_token = resp.next_page_token; + assert!(pages < 100, "pagination must terminate"); + } + + // Every seeded message appears exactly once (no dupes, no gaps), in + // the stable newest-first order including the id tiebreaker. + assert_eq!( + seen, + vec!["m7", "m6", "m5", "m4", "m3", "m2", "m1"], + "keyset paging must yield the full set once, newest-first" + ); + assert!(pages >= 4, "a 7-item set at page_size 2 must span >1 page"); + } + + /// A `page_token` the server cannot decode is rejected with a typed + /// `VALIDATION_FAILED` `ErrorInfo` mapped to `InvalidArgument` -- never a + /// panic and never a silently-empty first page. + #[tokio::test] + async fn list_messages_rejects_malformed_page_token() { + let event_bus = Arc::new(EventBus::new()); + let (addr, _handle, _dir) = spawn_test_server(event_bus, "correct-token").await; + let mut client = MailClient::connect(format!("http://{addr}")) + .await + .expect("client connects"); + + let err = client + .list_messages(authed_bearer_request(ListMessagesRequest { + folder_id: "inbox".to_string(), + page_size: 2, + page_token: "not-a-valid-cursor".to_string(), + })) + .await + .expect_err("a malformed page_token must be rejected"); + assert_eq!(err.code(), Code::InvalidArgument); + assert_eq!( + nuncio_proto::errors::error_reason(&err), + Some(ErrorReason::ValidationFailed), + "malformed page_token must carry a typed VALIDATION_FAILED reason" + ); + } + #[tokio::test] async fn mail_rpcs_reject_missing_bearer_token() { // Confirms `Mail` is mounted behind its own `BearerAuthInterceptor` @@ -3733,7 +3940,10 @@ mod tests { .expect("client connects"); let err = client - .list_folders(ListFoldersRequest {}) + .list_folders(ListFoldersRequest { + page_size: 0, + page_token: String::new(), + }) .await .expect_err("missing bearer token must be rejected"); assert_eq!(err.code(), Code::Unauthenticated); @@ -3741,7 +3951,8 @@ mod tests { let err = client .list_messages(ListMessagesRequest { folder_id: "inbox".to_string(), - limit: 10, + page_size: 10, + page_token: String::new(), }) .await .expect_err("missing bearer token must be rejected"); @@ -3767,6 +3978,8 @@ mod tests { let err = client .search_messages(SearchMessagesRequest { query: "hello".to_string(), + page_size: 0, + page_token: String::new(), }) .await .expect_err("missing bearer token must be rejected"); @@ -3814,6 +4027,8 @@ mod tests { calendar_id: "cal-1".to_string(), start_window: Some(nuncio_proto::time::timestamp_from_unix_secs(0)), end_window: Some(nuncio_proto::time::timestamp_from_unix_secs(i64::MAX)), + page_size: 0, + page_token: String::new(), }) .await .expect_err("missing bearer token must be rejected"); @@ -3885,7 +4100,8 @@ mod tests { let err = client .list_messages(authed_bearer_request(ListMessagesRequest { folder_id: String::new(), - limit: 10, + page_size: 10, + page_token: String::new(), })) .await .expect_err("empty folder_id must be rejected"); @@ -3942,7 +4158,10 @@ mod tests { // ListFolders reports the seeded folder with the right counts. let folders = client - .list_folders(authed_bearer_request(ListFoldersRequest {})) + .list_folders(authed_bearer_request(ListFoldersRequest { + page_size: 0, + page_token: String::new(), + })) .await .expect("list_folders succeeds") .into_inner() @@ -3956,7 +4175,8 @@ mod tests { let messages = client .list_messages(authed_bearer_request(ListMessagesRequest { folder_id: "inbox".to_string(), - limit: 10, + page_size: 10, + page_token: String::new(), })) .await .expect("list_messages succeeds") @@ -3988,6 +4208,8 @@ mod tests { let hits = client .search_messages(authed_bearer_request(SearchMessagesRequest { query: "revenue".to_string(), + page_size: 0, + page_token: String::new(), })) .await .expect("search_messages succeeds") @@ -4513,7 +4735,10 @@ mod tests { assert_eq!(err.code(), Code::Unauthenticated); let err = client - .list_rules(ListRulesRequest {}) + .list_rules(ListRulesRequest { + page_size: 0, + page_token: String::new(), + }) .await .expect_err("missing bearer token must be rejected"); assert_eq!(err.code(), Code::Unauthenticated); @@ -4639,7 +4864,10 @@ mod tests { // ListRules reflects the persisted rule. let list_response = client - .list_rules(authed_bearer_request(ListRulesRequest {})) + .list_rules(authed_bearer_request(ListRulesRequest { + page_size: 0, + page_token: String::new(), + })) .await .expect("list_rules succeeds") .into_inner(); @@ -4655,7 +4883,10 @@ mod tests { .expect("delete_rule succeeds"); let list_after_delete = client - .list_rules(authed_bearer_request(ListRulesRequest {})) + .list_rules(authed_bearer_request(ListRulesRequest { + page_size: 0, + page_token: String::new(), + })) .await .expect("list_rules succeeds") .into_inner(); @@ -4685,7 +4916,10 @@ mod tests { assert_eq!(err.code(), Code::InvalidArgument); let list_response = client - .list_rules(authed_bearer_request(ListRulesRequest {})) + .list_rules(authed_bearer_request(ListRulesRequest { + page_size: 0, + page_token: String::new(), + })) .await .expect("list_rules succeeds") .into_inner(); @@ -4756,7 +4990,10 @@ mod tests { ); let list_response = client - .list_rules(authed_bearer_request(ListRulesRequest {})) + .list_rules(authed_bearer_request(ListRulesRequest { + page_size: 0, + page_token: String::new(), + })) .await .expect("list_rules succeeds") .into_inner(); @@ -4821,7 +5058,10 @@ mod tests { // Nothing was persisted by either validation call. let list_response = client - .list_rules(authed_bearer_request(ListRulesRequest {})) + .list_rules(authed_bearer_request(ListRulesRequest { + page_size: 0, + page_token: String::new(), + })) .await .expect("list_rules succeeds") .into_inner(); @@ -4889,7 +5129,10 @@ mod tests { // Previewing must never persist the rule. let list_response = client - .list_rules(authed_bearer_request(ListRulesRequest {})) + .list_rules(authed_bearer_request(ListRulesRequest { + page_size: 0, + page_token: String::new(), + })) .await .expect("list_rules succeeds") .into_inner(); @@ -4998,7 +5241,10 @@ mod tests { ); let list_response = client - .list_rules(authed_bearer_request(ListRulesRequest {})) + .list_rules(authed_bearer_request(ListRulesRequest { + page_size: 0, + page_token: String::new(), + })) .await .expect("list_rules succeeds") .into_inner(); @@ -5077,7 +5323,10 @@ mod tests { assert_eq!(err.code(), Code::InvalidArgument); let list_response = client - .list_rules(authed_bearer_request(ListRulesRequest {})) + .list_rules(authed_bearer_request(ListRulesRequest { + page_size: 0, + page_token: String::new(), + })) .await .expect("list_rules succeeds") .into_inner(); @@ -5154,7 +5403,10 @@ mod tests { assert_eq!(response.errors.len(), 1); let list_response = client - .list_rules(authed_bearer_request(ListRulesRequest {})) + .list_rules(authed_bearer_request(ListRulesRequest { + page_size: 0, + page_token: String::new(), + })) .await .expect("list_rules succeeds") .into_inner(); @@ -5406,8 +5658,8 @@ mod tests { let err = client .list_records(ListRecordsRequest { - limit: 10, - offset: 0, + page_size: 10, + page_token: String::new(), }) .await .expect_err("missing bearer token must be rejected"); @@ -5455,8 +5707,8 @@ mod tests { let response = client .list_records(authed_bearer_request(ListRecordsRequest { - limit: 0, - offset: 0, + page_size: 0, + page_token: String::new(), })) .await .expect("list_records succeeds") @@ -5471,6 +5723,81 @@ mod tests { assert_eq!(response.records[1].action, "test.action.two"); } + /// Keyset paging the WORM ledger across multiple pages must return every + /// record EXACTLY ONCE, in ascending sequence, with no dupes or gaps. + #[tokio::test] + async fn list_records_keyset_pagination_returns_every_record_once_no_gaps() { + let (db, _dir) = DatabaseEngine::connect_ephemeral() + .await + .expect("connect ephemeral test db"); + for i in 0..7 { + db.append_worm_audit_record( + "system.test", + &format!("test.action.{i}"), + format!("payload-{i}").as_bytes(), + ) + .await + .expect("append audit record"); + } + + let event_bus = Arc::new(EventBus::new()); + let secrets = Arc::new(SecretManager::mock()); + let filter_engine = Arc::new(FilterEngine::new(Vec::new()).expect("empty rule set")); + let (addr, _handle) = spawn_test_server_with( + event_bus, + Arc::new(db), + filter_engine, + secrets, + "correct-token", + ) + .await; + let mut client = AuditClient::connect(format!("http://{addr}")) + .await + .expect("client connects"); + + let mut seen = Vec::new(); + let mut page_token = String::new(); + let mut pages = 0; + loop { + let resp = client + .list_records(authed_bearer_request(ListRecordsRequest { + page_size: 2, + page_token: page_token.clone(), + })) + .await + .expect("list_records page succeeds") + .into_inner(); + pages += 1; + assert!(resp.records.len() <= 2, "page must not exceed page_size"); + seen.extend(resp.records.iter().map(|r| r.sequence)); + if resp.next_page_token.is_empty() { + break; + } + page_token = resp.next_page_token; + assert!(pages < 100, "pagination must terminate"); + } + + assert_eq!( + seen, + vec![1, 2, 3, 4, 5, 6, 7], + "keyset paging must yield every sequence once, ascending" + ); + + // A malformed page_token is a typed VALIDATION_FAILED error. + let err = client + .list_records(authed_bearer_request(ListRecordsRequest { + page_size: 2, + page_token: "garbage".to_string(), + })) + .await + .expect_err("malformed page_token must be rejected"); + assert_eq!(err.code(), Code::InvalidArgument); + assert_eq!( + nuncio_proto::errors::error_reason(&err), + Some(ErrorReason::ValidationFailed) + ); + } + /// Proves `Audit/VerifyChain` reports `valid: true` and the real /// record count on an intact, honestly-signed chain -- never a /// fabricated `valid`. @@ -5645,6 +5972,8 @@ mod tests { calendar_id: "cal-work".to_string(), start_window: Some(nuncio_proto::time::timestamp_from_unix_secs(1_699_999_000)), end_window: Some(nuncio_proto::time::timestamp_from_unix_secs(1_700_004_000)), + page_size: 0, + page_token: String::new(), })) .await .expect("list_events succeeds") @@ -5704,6 +6033,8 @@ mod tests { calendar_id: "cal-work".to_string(), start_window: Some(nuncio_proto::time::timestamp_from_unix_secs(1_699_999_000)), end_window: Some(nuncio_proto::time::timestamp_from_unix_secs(1_700_004_000)), + page_size: 0, + page_token: String::new(), })) .await .expect("list_events succeeds") @@ -5730,6 +6061,8 @@ mod tests { calendar_id: "cal-work".to_string(), start_window: Some(nuncio_proto::time::timestamp_from_unix_secs(0)), end_window: Some(nuncio_proto::time::timestamp_from_unix_secs(i64::MAX)), + page_size: 0, + page_token: String::new(), })) .await .expect_err("empty account_id must be rejected"); @@ -5894,6 +6227,8 @@ mod tests { calendar_id: "cal-work".to_string(), start_window: Some(nuncio_proto::time::timestamp_from_unix_secs(1_699_999_000)), end_window: Some(nuncio_proto::time::timestamp_from_unix_secs(1_700_004_000)), + page_size: 0, + page_token: String::new(), })) .await .expect("list_events succeeds") @@ -6068,6 +6403,8 @@ mod tests { let err = client .list_contacts(ListContactsRequest { account_id: "acct-1".to_string(), + page_size: 0, + page_token: String::new(), }) .await .expect_err("missing bearer token must be rejected"); @@ -6126,6 +6463,8 @@ mod tests { let response = client .list_contacts(authed_bearer_request(ListContactsRequest { account_id: "acct-contacts-list".to_string(), + page_size: 0, + page_token: String::new(), })) .await .expect("list_contacts succeeds") @@ -6239,6 +6578,8 @@ mod tests { let list_response = client .list_contacts(authed_bearer_request(ListContactsRequest { account_id: "acct-contacts-injected".to_string(), + page_size: 0, + page_token: String::new(), })) .await .expect("list_contacts succeeds") @@ -6350,7 +6691,10 @@ mod tests { .await .expect("Mail client connects"); let err = client - .list_folders(ListFoldersRequest {}) + .list_folders(ListFoldersRequest { + page_size: 0, + page_token: String::new(), + }) .await .expect_err("Mail must reject an unauthenticated call"); assert_eq!(err.code(), Code::Unauthenticated); @@ -6360,7 +6704,10 @@ mod tests { .await .expect("Filters client connects"); let err = client - .list_rules(ListRulesRequest {}) + .list_rules(ListRulesRequest { + page_size: 0, + page_token: String::new(), + }) .await .expect_err("Filters must reject an unauthenticated call"); assert_eq!(err.code(), Code::Unauthenticated); @@ -6385,8 +6732,8 @@ mod tests { .expect("Audit client connects"); let err = client .list_records(ListRecordsRequest { - limit: 10, - offset: 0, + page_size: 10, + page_token: String::new(), }) .await .expect_err("Audit must reject an unauthenticated call"); @@ -6402,6 +6749,8 @@ mod tests { calendar_id: "cal-1".to_string(), start_window: Some(nuncio_proto::time::timestamp_from_unix_secs(0)), end_window: Some(nuncio_proto::time::timestamp_from_unix_secs(i64::MAX)), + page_size: 0, + page_token: String::new(), }) .await .expect_err("Calendar must reject an unauthenticated call"); @@ -6414,6 +6763,8 @@ mod tests { let err = client .list_contacts(ListContactsRequest { account_id: "acct-1".to_string(), + page_size: 0, + page_token: String::new(), }) .await .expect_err("Contacts must reject an unauthenticated call"); diff --git a/crates/nunciod/src/lib.rs b/crates/nunciod/src/lib.rs index 0f3ef2d..93fcc13 100644 --- a/crates/nunciod/src/lib.rs +++ b/crates/nunciod/src/lib.rs @@ -9,6 +9,7 @@ pub mod lifecycle; pub mod lock; pub mod orchestrator; pub mod outbox; +pub mod pagination; pub mod send; pub mod sync; pub use orchestrator::SelfHealingSyncOrchestrator; diff --git a/crates/nunciod/src/pagination.rs b/crates/nunciod/src/pagination.rs new file mode 100644 index 0000000..958341a --- /dev/null +++ b/crates/nunciod/src/pagination.rs @@ -0,0 +1,168 @@ +//! Shared keyset ("cursor") pagination for the daemon's list/search RPCs. +//! +//! Every `nuncio.v1` list/search RPC pages results with the same convention +//! (AIP-158): a request carries `page_size` + `page_token` and a response +//! carries `next_page_token`. The `page_token` is an OPAQUE cursor, never a +//! numeric offset: it encodes the stable sort key (and id tiebreaker) of the +//! last row of the previous page so the next page resumes strictly AFTER that +//! row. Keyset paging is stable under concurrent inserts -- rows are never +//! skipped or returned twice as the underlying table changes between page +//! fetches, which offset paging cannot guarantee. +//! +//! The wire encoding is deliberately private to the daemon: it is a +//! hex-encoded JSON array of typed [`CursorField`]s. Clients must treat the +//! token as opaque and echo it back verbatim; a token this module cannot +//! decode is rejected with a typed [`ErrorReason::ValidationFailed`] error +//! (mapped to `InvalidArgument`), never silently ignored or treated as the +//! first page. + +use nuncio_proto::errors; +use nuncio_proto::v1::ErrorReason; +use serde::{Deserialize, Serialize}; +use tonic::Status; + +/// Page size used when a request leaves `page_size` at 0 ("server default"). +pub const DEFAULT_PAGE_SIZE: u32 = 50; + +/// Hard upper bound on a single page, so a client cannot force the daemon to +/// materialize an unbounded result set in one call. +pub const MAX_PAGE_SIZE: u32 = 500; + +/// Resolves a requested `page_size` into a concrete row count: 0 selects +/// [`DEFAULT_PAGE_SIZE`], and any value is clamped to [`MAX_PAGE_SIZE`]. +#[must_use] +pub fn clamp_page_size(requested: u32) -> usize { + let n = if requested == 0 { + DEFAULT_PAGE_SIZE + } else { + requested.min(MAX_PAGE_SIZE) + }; + n as usize +} + +/// One typed component of an opaque keyset cursor. A cursor is an ordered +/// list of these, matching a list's stable sort key followed by its id +/// tiebreaker (e.g. `[I(received_at), S(id)]` for messages). +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub enum CursorField { + /// A signed integer sort key (e.g. a unix timestamp or a priority). + I(i64), + /// An unsigned integer sort key (e.g. an audit sequence). + U(u64), + /// A floating-point sort key (e.g. an FTS5 relevance rank). + F(f64), + /// A textual sort key or id (e.g. a message id or display name). + S(String), +} + +/// Encodes a cursor's component list into an opaque `page_token`. +#[must_use] +pub fn encode_cursor(fields: &[CursorField]) -> String { + // The component list is a fixed, small set of primitives, so + // serialization cannot realistically fail; fall back to an empty payload + // rather than panicking, keeping this crate free of `unwrap`/`expect`. + let json = serde_json::to_vec(fields).unwrap_or_default(); + hex::encode(json) +} + +fn malformed_token() -> Status { + errors::status( + ErrorReason::ValidationFailed, + "malformed page_token: not a valid pagination cursor", + ) +} + +/// Decodes an opaque `page_token` back into its component list, rejecting a +/// token that is not a hex-encoded cursor JSON payload with a typed +/// `VALIDATION_FAILED` error. +pub fn decode_cursor(token: &str) -> Result, Status> { + let bytes = hex::decode(token).map_err(|_| malformed_token())?; + serde_json::from_slice(&bytes).map_err(|_| malformed_token()) +} + +/// Decodes an optional `page_token` into a strongly-typed cursor value. +/// +/// An empty token means "first page" and yields `Ok(None)`. A non-empty +/// token is decoded and handed to `shape`, which pattern-matches the expected +/// component layout for a specific list and returns `Some(value)`; a token +/// that decodes but does not match the expected layout (e.g. a cursor minted +/// for a different list) is rejected with the same typed `VALIDATION_FAILED` +/// error as an undecodable token, so a caller can never resume from a cursor +/// shaped for the wrong query. +pub fn parse_token( + token: &str, + shape: impl FnOnce(&[CursorField]) -> Option, +) -> Result, Status> { + if token.is_empty() { + return Ok(None); + } + let fields = decode_cursor(token)?; + shape(&fields).map(Some).ok_or_else(malformed_token) +} + +#[cfg(test)] +mod tests { + use super::*; + use tonic::Code; + + #[test] + fn clamp_page_size_applies_default_and_max() { + assert_eq!(clamp_page_size(0), DEFAULT_PAGE_SIZE as usize); + assert_eq!(clamp_page_size(10), 10); + assert_eq!(clamp_page_size(MAX_PAGE_SIZE + 1), MAX_PAGE_SIZE as usize); + assert_eq!(clamp_page_size(u32::MAX), MAX_PAGE_SIZE as usize); + } + + #[test] + fn cursor_round_trips_through_encode_decode() { + let fields = vec![ + CursorField::I(1_700_000_000), + CursorField::S("msg-9".into()), + ]; + let token = encode_cursor(&fields); + assert!(!token.is_empty()); + assert_eq!(decode_cursor(&token).unwrap(), fields); + } + + #[test] + fn empty_token_parses_as_first_page() { + let parsed = parse_token("", |f| match f { + [CursorField::I(n)] => Some(*n), + _ => None, + }) + .unwrap(); + assert_eq!(parsed, None); + } + + #[test] + fn malformed_token_is_validation_failed() { + // Not valid hex. + let err = decode_cursor("zzz not-hex").unwrap_err(); + assert_eq!(err.code(), Code::InvalidArgument); + assert_eq!( + errors::error_reason(&err), + Some(ErrorReason::ValidationFailed) + ); + + // Valid hex, but not a cursor JSON payload. + let err = decode_cursor(&hex::encode(b"not json")).unwrap_err(); + assert_eq!(err.code(), Code::InvalidArgument); + } + + #[test] + fn wrong_shape_token_is_rejected() { + // A cursor with a single int component handed to a shape expecting + // (int, string) must be rejected, not silently accepted. + let token = encode_cursor(&[CursorField::I(1)]); + let err = parse_token(&token, |f| match f { + [CursorField::I(n), CursorField::S(s)] => Some((*n, s.clone())), + _ => None, + }) + .unwrap_err(); + assert_eq!(err.code(), Code::InvalidArgument); + assert_eq!( + errors::error_reason(&err), + Some(ErrorReason::ValidationFailed) + ); + } +} diff --git a/crates/nunciod/tests/caldav_production_e2e_test.rs b/crates/nunciod/tests/caldav_production_e2e_test.rs index 51a9d1f..6a210f3 100644 --- a/crates/nunciod/tests/caldav_production_e2e_test.rs +++ b/crates/nunciod/tests/caldav_production_e2e_test.rs @@ -198,6 +198,8 @@ async fn caldav_production_sync_builds_real_client_and_persists_events() { calendar_id: CALENDAR_ID.to_string(), start_window: Some(nuncio_proto::time::timestamp_from_unix_secs(1_700_000_000)), end_window: Some(nuncio_proto::time::timestamp_from_unix_secs(1_800_000_000)), + page_size: 0, + page_token: String::new(), }) .await .expect("list_events succeeds") @@ -234,6 +236,8 @@ async fn caldav_production_sync_builds_real_client_and_persists_events() { calendar_id: CALENDAR_ID.to_string(), start_window: Some(nuncio_proto::time::timestamp_from_unix_secs(0)), end_window: Some(nuncio_proto::time::timestamp_from_unix_secs(i64::MAX)), + page_size: 0, + page_token: String::new(), }) .await .expect_err("missing bearer token must be rejected"); diff --git a/crates/nunciod/tests/calendar_e2e_test.rs b/crates/nunciod/tests/calendar_e2e_test.rs index 3c3ea63..f9cf5c9 100644 --- a/crates/nunciod/tests/calendar_e2e_test.rs +++ b/crates/nunciod/tests/calendar_e2e_test.rs @@ -154,6 +154,8 @@ async fn calendar_sync_then_list_and_get_round_trip_over_authenticated_grpc() { calendar_id: CALENDAR_ID.to_string(), start_window: Some(nuncio_proto::time::timestamp_from_unix_secs(1_699_999_000)), end_window: Some(nuncio_proto::time::timestamp_from_unix_secs(1_700_020_000)), + page_size: 0, + page_token: String::new(), }) .await .expect("list_events succeeds") @@ -189,6 +191,8 @@ async fn calendar_sync_then_list_and_get_round_trip_over_authenticated_grpc() { calendar_id: CALENDAR_ID.to_string(), start_window: Some(nuncio_proto::time::timestamp_from_unix_secs(0)), end_window: Some(nuncio_proto::time::timestamp_from_unix_secs(i64::MAX)), + page_size: 0, + page_token: String::new(), }) .await .expect_err("missing bearer token must be rejected"); diff --git a/crates/nunciod/tests/contacts_e2e_test.rs b/crates/nunciod/tests/contacts_e2e_test.rs index f231f39..8cc1cf5 100644 --- a/crates/nunciod/tests/contacts_e2e_test.rs +++ b/crates/nunciod/tests/contacts_e2e_test.rs @@ -146,6 +146,8 @@ async fn contacts_sync_create_and_list_round_trip_over_authenticated_grpc() { let list_response = contacts_client .list_contacts(ListContactsRequest { account_id: ACCOUNT_ID.to_string(), + page_size: 0, + page_token: String::new(), }) .await .expect("list_contacts succeeds") @@ -201,6 +203,8 @@ async fn contacts_sync_create_and_list_round_trip_over_authenticated_grpc() { let second_list = second_client .list_contacts(ListContactsRequest { account_id: ACCOUNT_ID.to_string(), + page_size: 0, + page_token: String::new(), }) .await .expect("list_contacts succeeds on second connection") @@ -231,6 +235,8 @@ async fn contacts_sync_create_and_list_round_trip_over_authenticated_grpc() { let err = unauthenticated_client .list_contacts(ListContactsRequest { account_id: ACCOUNT_ID.to_string(), + page_size: 0, + page_token: String::new(), }) .await .expect_err("missing bearer token must be rejected"); diff --git a/crates/nunciod/tests/spine_e2e_test.rs b/crates/nunciod/tests/spine_e2e_test.rs index 42f786c..d7ad21f 100644 --- a/crates/nunciod/tests/spine_e2e_test.rs +++ b/crates/nunciod/tests/spine_e2e_test.rs @@ -226,7 +226,8 @@ async fn full_mail_spine_round_trips_over_authenticated_grpc_with_mocked_protoco let list_messages_response = mail_client .list_messages(ListMessagesRequest { folder_id: "inbox".to_string(), - limit: 0, + page_size: 0, + page_token: String::new(), }) .await .expect("list_messages succeeds") -- 2.54.0 From df33f391ac56ee2c8bf86c5ee52cf730cd69349a Mon Sep 17 00:00:00 2001 From: James Maes Date: Fri, 31 Jul 2026 16:51:58 -0500 Subject: [PATCH 2/2] feat(mail): bound IMAP fetch by batch count and per-message size Enumerate a folder's UIDs and sizes cheaply (no bodies), then pull full bodies in explicit UID-set batches of at most MAX_MESSAGES_PER_FETCH_BATCH so the transient working set stays bounded regardless of mailbox size. Messages whose RFC822.SIZE exceeds MAX_MESSAGE_BODY_BYTES are fetched headers-only and recorded with an empty body plus a WARN naming the uid and size, rather than buffered whole or silently dropped. Refs #301 --- crates/nuncio-mail/src/imap.rs | 742 +++++++++++++++++++++++++++------ 1 file changed, 623 insertions(+), 119 deletions(-) diff --git a/crates/nuncio-mail/src/imap.rs b/crates/nuncio-mail/src/imap.rs index c5288ac..e022380 100644 --- a/crates/nuncio-mail/src/imap.rs +++ b/crates/nuncio-mail/src/imap.rs @@ -31,6 +31,27 @@ const CONNECT_TIMEOUT: Duration = Duration::from_secs(15); /// honest timeout rather than hanging the sync indefinitely. const FETCH_ITEM_TIMEOUT: Duration = Duration::from_secs(60); +/// Upper bound on how many messages a single `UID FETCH` may pull full bodies +/// for. A folder sync never issues one unbounded `UID FETCH ` that +/// buffers every body at once; instead the messages in range are enumerated +/// cheaply (UIDs + sizes, no bodies) and their bodies fetched in explicit +/// UID-set batches of at most this many. This bounds the transient in-memory +/// working set of full message bodies to a fixed window, so a mailbox with an +/// enormous number of messages cannot drive the daemon to exhaust memory in a +/// single fetch regardless of how many messages the server holds. +const MAX_MESSAGES_PER_FETCH_BATCH: usize = 200; + +/// Upper bound on the RFC822 size (bytes) of a message whose full body the sync +/// will pull into memory. The server reports each message's size via +/// `RFC822.SIZE` during enumeration; a message above this bound is NOT fetched +/// with its full body (which could be arbitrarily large and OOM the daemon). +/// Instead only its headers are fetched and it is recorded with an empty body +/// plus a WARN naming the UID and size, rather than being silently dropped or +/// buffered whole. The bound matches the MIME parser's own maximum payload +/// ([`MimeParserAdapter::MAX_PAYLOAD_BYTES`]): a body larger than the parser +/// would ever accept is pointless to transfer only to reject. +const MAX_MESSAGE_BODY_BYTES: u32 = 25 * 1024 * 1024; + use crate::backend::{MailBackend, RemoteMutationKind, RemoteMutationSpec}; use crate::parser::{MailError, MimeParserAdapter}; @@ -191,6 +212,91 @@ pub fn build_fetch_command_query() -> &'static str { "(FLAGS INTERNALDATE RFC822.SIZE ENVELOPE BODY.PEEK[])" } +/// The `UID FETCH` query for the cheap enumeration pass: UID plus `RFC822.SIZE` +/// only, deliberately WITHOUT a `BODY.PEEK[]`. This lets the sync discover every +/// message's UID and size (to decide per-message whether the body is safe to +/// pull) without transferring any body, so the enumeration stays small even for +/// a huge mailbox. +fn build_enumerate_command_query() -> &'static str { + "(UID RFC822.SIZE)" +} + +/// The `UID FETCH` query for a message whose body exceeds +/// [`MAX_MESSAGE_BODY_BYTES`]: headers only (`BODY.PEEK[HEADER]`), never the +/// full `BODY.PEEK[]`. The oversized body is intentionally left unfetched; the +/// headers are enough to record honest metadata (subject, sender, date). +fn build_headers_only_command_query() -> &'static str { + "(FLAGS INTERNALDATE RFC822.SIZE BODY.PEEK[HEADER])" +} + +/// Render an IMAP ENVELOPE address as `mailbox@host`, or bare `mailbox` when no +/// host is present. Both components are decoded from their raw bytes lossily. +/// Used only when a FETCH item carries an ENVELOPE but no body, so a minimal +/// record can still name the sender/recipient. +fn format_envelope_address(addr: &async_imap::imap_proto::Address) -> String { + let mailbox = addr + .mailbox + .as_ref() + .map_or("", |m| std::str::from_utf8(m).unwrap_or("")); + let host = addr + .host + .as_ref() + .map_or("", |h| std::str::from_utf8(h).unwrap_or("")); + if host.is_empty() { + mailbox.to_string() + } else { + format!("{}@{}", mailbox, host) + } +} + +/// Render an explicit IMAP UID sequence-set from a batch of UIDs, e.g. +/// `[12, 15, 18]` -> `"12,15,18"`. Used to fetch a bounded batch of specific +/// messages rather than an open-ended `n:*` range. +fn join_uid_batch(uids: &[u32]) -> String { + uids.iter() + .map(|u| u.to_string()) + .collect::>() + .join(",") +} + +/// Drive one already-issued `UID FETCH` response stream to completion, +/// collecting its items. Every individual item read is bounded by +/// [`FETCH_ITEM_TIMEOUT`]: a slow-but-progressing fetch is fine, but a single +/// item that never arrives (dead socket, server-side hang) surfaces as an +/// honest [`MailError::FetchStalled`] instead of hanging the sync. `read_so_far` +/// is the count of messages already read across earlier batches, so the stall +/// message reports the true progress. +async fn drive_fetch_stream( + stream: &mut St, + folder_id: &str, + read_so_far: usize, +) -> Result, MailError> +where + St: tokio_stream::Stream> + Unpin, + E: std::fmt::Display, +{ + let mut items = Vec::new(); + loop { + let next_item = tokio::time::timeout(FETCH_ITEM_TIMEOUT, stream.next()) + .await + .map_err(|_| { + MailError::FetchStalled(format!( + "no FETCH response for folder '{}' within {}s ({} message(s) read so far)", + folder_id, + FETCH_ITEM_TIMEOUT.as_secs(), + read_so_far + items.len() + )) + })?; + let Some(fetch_res) = next_item else { + break; + }; + let fetch_data = fetch_res + .map_err(|e| MailError::ImapError(format!("failed reading fetch item: {}", e)))?; + items.push(fetch_data); + } + Ok(items) +} + /// Establish an encrypted TLS stream to an IMAP server. pub async fn connect_tls_stream(host: &str, port: u16) -> Result, MailError> { let connector = build_tls_connector()?; @@ -583,6 +689,40 @@ impl ImapEngine { since_state: Option<&str>, session: &mut async_imap::Session, ) -> Result<(Vec, String), MailError> + where + S: AsyncRead + AsyncWrite + Unpin + Send + Debug + 'static, + { + self.sync_folder_messages_batched( + folder_id, + since_state, + session, + MAX_MESSAGES_PER_FETCH_BATCH, + ) + .await + } + + /// Core folder sync, parameterised by the maximum number of messages whose + /// full bodies are pulled per `UID FETCH`. Production callers pass + /// [`MAX_MESSAGES_PER_FETCH_BATCH`]; the parameter is a seam so tests can + /// exercise the multi-batch path with a small window instead of a huge + /// fixture. + /// + /// The sync runs in two phases so the transient working set stays bounded + /// regardless of mailbox size: + /// 1. A cheap enumeration ([`build_enumerate_command_query`]) reads every + /// in-range message's UID and `RFC822.SIZE` with NO body. Messages at or + /// below [`MAX_MESSAGE_BODY_BYTES`] are queued for a full-body fetch; + /// oversized ones are queued for a headers-only fetch and WARN-logged. + /// 2. Each queue is drained in explicit UID-set batches of at most + /// `batch_size`, so the full bodies held in memory at once never exceed + /// one batch. + async fn sync_folder_messages_batched( + &self, + folder_id: &str, + since_state: Option<&str>, + session: &mut async_imap::Session, + batch_size: usize, + ) -> Result<(Vec, String), MailError> where S: AsyncRead + AsyncWrite + Unpin + Send + Debug + 'static, { @@ -609,133 +749,93 @@ impl ImapEngine { FetchRange::Full | FetchRange::Incremental(_) => {} } - let query = build_fetch_command_query(); + let batch_size = batch_size.max(1); let sequence_set = range.sequence_set(); - let mut fetch_stream = session.uid_fetch(&sequence_set, query).await.map_err(|e| { - MailError::ImapError(format!( - "UID FETCH failed for folder '{}': {}", - folder_id, e - )) - })?; - let mut emails = Vec::new(); - let mut max_uid_seen: u32 = 0; - loop { - // Bound EVERY individual item read: a slow-but-progressing large - // fetch is fine, but a single item that never arrives must surface - // as an honest `FetchStalled` instead of hanging the whole sync. - let next_item = tokio::time::timeout(FETCH_ITEM_TIMEOUT, fetch_stream.next()) + // Phase 1: enumerate UIDs and sizes only -- no bodies transferred, so + // this stays small even for an enormous mailbox. + let enumeration = { + let mut enumerate_stream = session + .uid_fetch(&sequence_set, build_enumerate_command_query()) .await - .map_err(|_| { - MailError::FetchStalled(format!( - "no FETCH response for folder '{}' within {}s ({} message(s) read so far)", - folder_id, - FETCH_ITEM_TIMEOUT.as_secs(), - emails.len() + .map_err(|e| { + MailError::ImapError(format!( + "UID FETCH (enumerate) failed for folder '{}': {}", + folder_id, e )) })?; - let Some(fetch_res) = next_item else { - break; + drive_fetch_stream(&mut enumerate_stream, folder_id, 0).await? + }; + + let mut max_uid_seen: u32 = 0; + let mut body_uids: Vec = Vec::new(); + let mut oversized_uids: Vec = Vec::new(); + for item in &enumeration { + // A message with no UID cannot be addressed by a follow-up fetch, + // so it is skipped rather than issued a UID-less request. + let Some(uid) = item.uid.filter(|u| *u >= 1) else { + continue; }; - let fetch_data = fetch_res - .map_err(|e| MailError::ImapError(format!("failed reading fetch item: {}", e)))?; - - let uid_num = fetch_data.uid.unwrap_or(0); - max_uid_seen = max_uid_seen.max(uid_num); - // The message is addressed on the wire by its UID within the - // folder's UIDVALIDITY scope; the persisted id is an opaque - // surrogate hashed over both plus the account and folder, so the - // same UID in another folder/account can never collide. - let remote_id = uid_num.to_string(); - let uid_validity = server_uid_validity - .map(|v| v.to_string()) - .unwrap_or_default(); - let email_id = - Email::surrogate_id(&self.account_id, folder_id, &uid_validity, &remote_id); - - let is_read = fetch_data - .flags() - .any(|flag| matches!(flag, async_imap::types::Flag::Seen)); - - let email = if let Some(raw_bytes) = fetch_data.body() { - let mut email = MimeParserAdapter::parse_mime( - &email_id, - &self.account_id, - folder_id, - raw_bytes, - )?; - email.remote_id = remote_id; - email.uid_validity = uid_validity; - email.read = is_read; - email - } else { - let subject = fetch_data - .envelope() - .and_then(|env| env.subject.as_ref()) - .map_or("No Subject".to_string(), |s| { - String::from_utf8_lossy(s).to_string() - }); - - let sender = fetch_data - .envelope() - .and_then(|env| env.from.as_ref()) - .and_then(|froms| froms.first()) - .map_or("unknown@nuncio.mx".to_string(), |addr| { - let mailbox = addr - .mailbox - .as_ref() - .map_or("", |m| std::str::from_utf8(m).unwrap_or("")); - let host = addr - .host - .as_ref() - .map_or("", |h| std::str::from_utf8(h).unwrap_or("")); - if host.is_empty() { - mailbox.to_string() - } else { - format!("{}@{}", mailbox, host) - } - }); - - let recipient = fetch_data - .envelope() - .and_then(|env| env.to.as_ref()) - .and_then(|tos| tos.first()) - .map_or("me@nuncio.mx".to_string(), |addr| { - let mailbox = addr - .mailbox - .as_ref() - .map_or("", |m| std::str::from_utf8(m).unwrap_or("")); - let host = addr - .host - .as_ref() - .map_or("", |h| std::str::from_utf8(h).unwrap_or("")); - if host.is_empty() { - mailbox.to_string() - } else { - format!("{}@{}", mailbox, host) - } - }); - - let received_at = fetch_data.internal_date().map_or(0, |dt| dt.timestamp()); - - Email { - id: email_id, - account_id: self.account_id.clone(), - folder_id: folder_id.to_string(), - remote_id, - uid_validity, - subject, - sender, - recipient, - received_at, - read: is_read, - body_plain: None, - body_html: None, - attachments: Vec::new(), + max_uid_seen = max_uid_seen.max(uid); + match item.size { + Some(size) if size > MAX_MESSAGE_BODY_BYTES => { + tracing::warn!( + folder_id, + uid, + size, + max_body_bytes = MAX_MESSAGE_BODY_BYTES, + "message body exceeds the per-message size cap; fetching headers only \ + and recording it with an empty body instead of buffering the full body" + ); + oversized_uids.push(uid); } - }; + _ => body_uids.push(uid), + } + } + drop(enumeration); - emails.push(email); + let mut emails = Vec::new(); + + // Phase 2a: normal messages -- full bodies, in bounded UID-set batches. + for chunk in body_uids.chunks(batch_size) { + let set = join_uid_batch(chunk); + let items = { + let mut fetch_stream = session + .uid_fetch(&set, build_fetch_command_query()) + .await + .map_err(|e| { + MailError::ImapError(format!( + "UID FETCH failed for folder '{}': {}", + folder_id, e + )) + })?; + drive_fetch_stream(&mut fetch_stream, folder_id, emails.len()).await? + }; + for item in &items { + emails.push(self.build_email_from_fetch(folder_id, server_uid_validity, item)?); + } + } + + // Phase 2b: oversized messages -- headers only, recorded with an empty + // body. Also batched, so a run of oversized messages cannot itself + // buffer an unbounded number of header fetches at once. + for chunk in oversized_uids.chunks(batch_size) { + let set = join_uid_batch(chunk); + let items = { + let mut fetch_stream = session + .uid_fetch(&set, build_headers_only_command_query()) + .await + .map_err(|e| { + MailError::ImapError(format!( + "UID FETCH (headers) failed for folder '{}': {}", + folder_id, e + )) + })?; + drive_fetch_stream(&mut fetch_stream, folder_id, emails.len()).await? + }; + for item in &items { + emails.push(self.build_headers_only_email(folder_id, server_uid_validity, item)); + } } // Derive the next checkpoint from real server state as @@ -760,6 +860,137 @@ impl ImapEngine { Ok((emails, new_checkpoint)) } + /// Build an [`Email`] from a full-body (or envelope-only) FETCH item. When + /// the item carries a `BODY[]` it is parsed as a full MIME message; when it + /// does not (a server that returned only the ENVELOPE), a minimal record is + /// synthesised from the envelope so the message is still surfaced honestly. + fn build_email_from_fetch( + &self, + folder_id: &str, + server_uid_validity: Option, + fetch_data: &async_imap::types::Fetch, + ) -> Result { + let uid_num = fetch_data.uid.unwrap_or(0); + // The message is addressed on the wire by its UID within the folder's + // UIDVALIDITY scope; the persisted id is an opaque surrogate hashed over + // both plus the account and folder, so the same UID in another + // folder/account can never collide. + let remote_id = uid_num.to_string(); + let uid_validity = server_uid_validity + .map(|v| v.to_string()) + .unwrap_or_default(); + let email_id = Email::surrogate_id(&self.account_id, folder_id, &uid_validity, &remote_id); + + let is_read = fetch_data + .flags() + .any(|flag| matches!(flag, async_imap::types::Flag::Seen)); + + if let Some(raw_bytes) = fetch_data.body() { + let mut email = + MimeParserAdapter::parse_mime(&email_id, &self.account_id, folder_id, raw_bytes)?; + email.remote_id = remote_id; + email.uid_validity = uid_validity; + email.read = is_read; + return Ok(email); + } + + let subject = fetch_data + .envelope() + .and_then(|env| env.subject.as_ref()) + .map_or("No Subject".to_string(), |s| { + String::from_utf8_lossy(s).to_string() + }); + + let sender = fetch_data + .envelope() + .and_then(|env| env.from.as_ref()) + .and_then(|froms| froms.first()) + .map_or("unknown@nuncio.mx".to_string(), format_envelope_address); + + let recipient = fetch_data + .envelope() + .and_then(|env| env.to.as_ref()) + .and_then(|tos| tos.first()) + .map_or("me@nuncio.mx".to_string(), format_envelope_address); + + let received_at = fetch_data.internal_date().map_or(0, |dt| dt.timestamp()); + + Ok(Email { + id: email_id, + account_id: self.account_id.clone(), + folder_id: folder_id.to_string(), + remote_id, + uid_validity, + subject, + sender, + recipient, + received_at, + read: is_read, + body_plain: None, + body_html: None, + attachments: Vec::new(), + }) + } + + /// Build an [`Email`] for a message whose body exceeded + /// [`MAX_MESSAGE_BODY_BYTES`]: its headers (`BODY[HEADER]`) are parsed for + /// honest metadata (subject/sender/date) but the oversized body is left + /// unfetched, so the record carries an EMPTY body rather than a fabricated + /// one. If the server returned no header section, this falls back to the + /// envelope-only record; either way the message is surfaced, never dropped. + fn build_headers_only_email( + &self, + folder_id: &str, + server_uid_validity: Option, + fetch_data: &async_imap::types::Fetch, + ) -> Email { + let uid_num = fetch_data.uid.unwrap_or(0); + let remote_id = uid_num.to_string(); + let uid_validity = server_uid_validity + .map(|v| v.to_string()) + .unwrap_or_default(); + let email_id = Email::surrogate_id(&self.account_id, folder_id, &uid_validity, &remote_id); + let is_read = fetch_data + .flags() + .any(|flag| matches!(flag, async_imap::types::Flag::Seen)); + + if let Some(header_bytes) = fetch_data.header() { + if let Ok(mut email) = + MimeParserAdapter::parse_mime(&email_id, &self.account_id, folder_id, header_bytes) + { + // A headers-only parse yields no body, but be explicit that the + // oversized body was intentionally not fetched. + email.remote_id = remote_id; + email.uid_validity = uid_validity; + email.read = is_read; + email.body_plain = None; + email.body_html = None; + email.attachments = Vec::new(); + return email; + } + } + + // No usable header section: fall back to whatever the envelope offers. + // `build_email_from_fetch` cannot error on this path (no body to parse), + // but if it ever did, surface a minimal honest record rather than panic. + self.build_email_from_fetch(folder_id, server_uid_validity, fetch_data) + .unwrap_or_else(|_| Email { + id: email_id, + account_id: self.account_id.clone(), + folder_id: folder_id.to_string(), + remote_id, + uid_validity, + subject: "No Subject".to_string(), + sender: "unknown@nuncio.mx".to_string(), + recipient: "me@nuncio.mx".to_string(), + received_at: 0, + read: is_read, + body_plain: None, + body_html: None, + attachments: Vec::new(), + }) + } + /// Fetch a folder's messages over a freshly authenticated session, /// honoring `since_state` for an incremental fetch and returning the new /// sync checkpoint alongside the messages. See @@ -1426,6 +1657,279 @@ mod tests { ); } + /// The RFC822 body of a scripted fixture message. + fn fixture_body(uid: u32) -> String { + format!( + "Subject: Message {uid}\r\nFrom: sender{uid}@example.test\r\n\ + To: me@example.test\r\n\r\nBody of message {uid}.\r\n" + ) + } + + /// The header block (no body) of a scripted fixture message. + fn fixture_header(uid: u32) -> String { + format!( + "Subject: Message {uid}\r\nFrom: sender{uid}@example.test\r\n\ + To: me@example.test\r\n\r\n" + ) + } + + /// Resolve which fixture UIDs a `UID FETCH` sequence-set token addresses. A + /// range `lo:*` selects every fixture UID `>= lo`; an explicit comma list + /// selects exactly those UIDs. Mirrors just enough server-side sequence-set + /// handling for the batched-fetch tests. + fn requested_uids(set: &str, all: &[(u32, u32)]) -> Vec { + if set.contains(':') { + let lo: u32 = set + .split(':') + .next() + .unwrap_or("1") + .trim() + .parse() + .unwrap_or(1); + all.iter().map(|(u, _)| *u).filter(|u| *u >= lo).collect() + } else { + set.split(',') + .filter_map(|s| s.trim().parse::().ok()) + .filter(|u| all.iter().any(|(fu, _)| fu == u)) + .collect() + } + } + + /// Drive `sync_folder_messages_batched` over a scripted, in-memory IMAP + /// server that serves a fixed set of `(uid, size)` messages. The server + /// answers the enumeration pass (UID + RFC822.SIZE), full-body batches + /// (`BODY[]`), and headers-only batches (`BODY[HEADER]`) from the fixture, + /// and records every `UID FETCH` command line the client issued so a test + /// can assert on the exact batching. Returns the synced emails, the new + /// checkpoint, and the captured `UID FETCH` command lines. + async fn run_batched_sync( + messages: Vec<(u32, u32)>, + uidvalidity: u32, + uidnext: u32, + since_state: Option, + batch_size: usize, + ) -> (Vec, String, Vec) { + let (client_io, server_io) = tokio::io::duplex(65536); + let fetch_cmds = Arc::new(std::sync::Mutex::new(Vec::::new())); + let server_cmds = fetch_cmds.clone(); + let exists = messages.len() as u32; + + let server = tokio::spawn(async move { + let mut io = server_io; + while let Some(line) = read_scripted_line(&mut io).await { + let tag = line.split_whitespace().next().unwrap_or("").to_string(); + let upper = line.to_ascii_uppercase(); + if upper.contains("UID FETCH") { + if let Ok(mut g) = server_cmds.lock() { + g.push(line.clone()); + } + let set = line.split_whitespace().nth(3).unwrap_or("").to_string(); + let uids = requested_uids(&set, &messages); + let mut resp = String::new(); + for uid in uids { + let size = messages + .iter() + .find(|(u, _)| *u == uid) + .map_or(0, |(_, s)| *s); + if upper.contains("BODY.PEEK[HEADER]") { + let hdr = fixture_header(uid); + resp.push_str(&format!( + "* {uid} FETCH (UID {uid} RFC822.SIZE {size} FLAGS (\\Seen) \ + BODY[HEADER] {{{}}}\r\n{hdr})\r\n", + hdr.len() + )); + } else if upper.contains("BODY.PEEK[]") { + let body = fixture_body(uid); + resp.push_str(&format!( + "* {uid} FETCH (UID {uid} RFC822.SIZE {size} FLAGS (\\Seen) \ + BODY[] {{{}}}\r\n{body})\r\n", + body.len() + )); + } else { + // Enumeration pass: UID + size only, no body. + resp.push_str(&format!( + "* {uid} FETCH (UID {uid} RFC822.SIZE {size})\r\n" + )); + } + } + resp.push_str(&format!("{tag} OK FETCH completed\r\n")); + let _ = io.write_all(resp.as_bytes()).await; + } else if upper.contains("LOGIN") { + let _ = io + .write_all(format!("{tag} OK Logged in\r\n").as_bytes()) + .await; + } else if upper.contains("SELECT") { + let resp = format!( + "* FLAGS (\\Seen)\r\n* {exists} EXISTS\r\n* 0 RECENT\r\n\ + * OK [UIDVALIDITY {uidvalidity}] ok\r\n* OK [UIDNEXT {uidnext}] ok\r\n\ + {tag} OK [READ-WRITE] SELECT done\r\n" + ); + let _ = io.write_all(resp.as_bytes()).await; + } else if upper.contains("LOGOUT") { + let _ = io + .write_all(format!("* BYE\r\n{tag} OK LOGOUT\r\n").as_bytes()) + .await; + break; + } + } + }); + + let client = async_imap::Client::new(client_io); + let mut session = client + .login("user", "pass") + .await + .map_err(|(e, _)| e) + .expect("scripted login succeeds"); + let engine = ImapEngine::new("acct-1", "example.test", 993); + let (emails, checkpoint) = engine + .sync_folder_messages_batched("INBOX", since_state.as_deref(), &mut session, batch_size) + .await + .expect("batched sync succeeds"); + drop(session); + let _ = server.await; + + let cmds = fetch_cmds.lock().expect("lock captured commands").clone(); + (emails, checkpoint, cmds) + } + + /// The captured `UID FETCH` command lines that pulled FULL bodies (a + /// `BODY.PEEK[]`, not the headers-only `BODY.PEEK[HEADER]` nor the + /// bodyless enumeration). + fn body_fetch_commands(cmds: &[String]) -> Vec<&String> { + cmds.iter() + .filter(|c| { + let u = c.to_ascii_uppercase(); + u.contains("BODY.PEEK[]") && !u.contains("BODY.PEEK[HEADER]") + }) + .collect() + } + + #[tokio::test] + async fn a_large_folder_is_fetched_across_multiple_bounded_batches() { + // Five messages with a batch window of two must be pulled across + // multiple bounded `UID FETCH` batches -- never a single unbounded + // `1:*` body fetch. The enumeration discovers the UIDs; the bodies are + // then fetched in explicit UID-set chunks of at most the batch size. + let messages: Vec<(u32, u32)> = (1..=5).map(|u| (u, 100)).collect(); + let (emails, checkpoint, cmds) = run_batched_sync(messages, 1, 6, None, 2).await; + + assert_eq!(emails.len(), 5, "every message must be synced"); + assert_eq!(checkpoint, "1:6"); + + let body_cmds = body_fetch_commands(&cmds); + assert_eq!( + body_cmds.len(), + 3, + "5 messages / batch of 2 => 3 bounded body fetches, got: {cmds:?}" + ); + for cmd in &body_cmds { + assert!( + !cmd.contains("1:*") && !cmd.contains(":*"), + "a body fetch must be a bounded UID set, never an open range: {cmd}" + ); + let set = cmd.split_whitespace().nth(3).unwrap_or(""); + let count = set.split(',').count(); + assert!( + count <= 2, + "each body fetch must carry at most the batch size (2) UIDs, got: {cmd}" + ); + } + } + + #[tokio::test] + async fn an_oversized_message_is_fetched_headers_only_not_buffered_whole() { + // Three messages where the middle one reports an RFC822.SIZE above the + // per-message cap. Its full body MUST NOT be requested; instead it is + // fetched headers-only and recorded with an empty body, while the two + // normal messages get full bodies. + let huge = MAX_MESSAGE_BODY_BYTES + 1; + let messages: Vec<(u32, u32)> = vec![(1, 100), (2, huge), (3, 100)]; + let (emails, _checkpoint, cmds) = run_batched_sync(messages, 1, 4, None, 200).await; + + assert_eq!(emails.len(), 3, "no message may be silently dropped"); + + // The oversized UID (2) must NEVER appear in a full-body fetch. + for cmd in body_fetch_commands(&cmds) { + let set = cmd.split_whitespace().nth(3).unwrap_or(""); + assert!( + !set.split(',').any(|u| u.trim() == "2"), + "the oversized message's full body must never be fetched, got: {cmd}" + ); + } + // It MUST be fetched headers-only. + assert!( + cmds.iter().any(|c| { + let u = c.to_ascii_uppercase(); + u.contains("BODY.PEEK[HEADER]") && c.split_whitespace().nth(3).unwrap_or("") == "2" + }), + "the oversized message must be fetched headers-only, got: {cmds:?}" + ); + + // The oversized message is recorded with honest metadata and an EMPTY + // body -- never a fabricated one. + let oversized = emails + .iter() + .find(|e| e.remote_id == "2") + .expect("oversized message must be present"); + assert!( + oversized.body_plain.is_none() && oversized.body_html.is_none(), + "oversized message must carry no body, got: {oversized:?}" + ); + assert!( + oversized.subject.contains('2'), + "oversized message must keep honest header metadata, got: {oversized:?}" + ); + + // The normal messages DID get full bodies. + for uid in ["1", "3"] { + let email = emails + .iter() + .find(|e| e.remote_id == uid) + .expect("normal message present"); + assert!( + email.body_plain.is_some(), + "normal message {uid} must have a body, got: {email:?}" + ); + } + } + + #[tokio::test] + async fn batched_sync_preserves_incremental_checkpointing() { + // Batching must not break the incremental checkpoint: a first sync + // records "{uidvalidity}:{uidnext}", and a second sync fed that + // checkpoint enumerates only the narrowed `{uidnext}:*` range (so a + // fire-once-per-new-message consumer never re-sees old mail). + let messages: Vec<(u32, u32)> = (1..=3).map(|u| (u, 100)).collect(); + let (first, checkpoint, first_cmds) = + run_batched_sync(messages.clone(), 1, 4, None, 2).await; + assert_eq!(first.len(), 3); + assert_eq!(checkpoint, "1:4"); + assert!( + first_cmds.iter().any(|c| c.contains("1:*")), + "first sync enumerates the full range, got: {first_cmds:?}" + ); + + let (second, checkpoint2, second_cmds) = + run_batched_sync(messages, 1, 4, Some(checkpoint.clone()), 2).await; + assert_eq!( + second.len(), + 0, + "no new mail arrived, so an incremental sync returns nothing" + ); + assert_eq!( + checkpoint2, "1:4", + "checkpoint stays stable across an empty pass" + ); + assert!( + second_cmds.iter().any(|c| c.contains("4:*")), + "second sync must enumerate the narrowed range, got: {second_cmds:?}" + ); + assert!( + !second_cmds.iter().any(|c| c.contains("1:*")), + "second sync must NOT re-enumerate the full range, got: {second_cmds:?}" + ); + } + #[tokio::test] async fn a_uidvalidity_change_forces_a_full_resync_instead_of_skipping_mail() { // A mailbox renumber changes UIDVALIDITY and restarts UIDs from low -- 2.54.0