From 3687bc3f0816ca46304c7de8bbc02d4b55e417c1 Mon Sep 17 00:00:00 2001 From: James Maes Date: Fri, 31 Jul 2026 16:31:35 -0500 Subject: [PATCH 1/2] feat(nunciod): bound outbox per-item execution with a timeout (#349) Wrap each outbox mutation's execution in a per-item timeout raced against the shutdown signal, so one hung remote op (a stuck IMAP/JMAP call or webhook) can no longer freeze delivery for every other pending mutation across every account. A timed-out item is dispositioned exactly like any other transient failure: retryable, bumping retry_count, never completed. Refs #290 Co-authored-by: James Maes --- crates/nunciod/Cargo.toml | 1 + crates/nunciod/src/main.rs | 11 +- crates/nunciod/src/outbox.rs | 42 ++- .../nunciod/tests/outbox_executor_e2e_test.rs | 297 +++++++++++++++++- 4 files changed, 338 insertions(+), 13 deletions(-) diff --git a/crates/nunciod/Cargo.toml b/crates/nunciod/Cargo.toml index d1ee17f..060404f 100644 --- a/crates/nunciod/Cargo.toml +++ b/crates/nunciod/Cargo.toml @@ -40,3 +40,4 @@ fs4.workspace = true nuncio-cli = { path = "../nuncio-cli" } sqlx.workspace = true wiremock.workspace = true +tokio = { workspace = true, features = ["test-util"] } diff --git a/crates/nunciod/src/main.rs b/crates/nunciod/src/main.rs index 29bb348..fcdc090 100644 --- a/crates/nunciod/src/main.rs +++ b/crates/nunciod/src/main.rs @@ -153,6 +153,7 @@ async fn main() -> Result<(), Box> { account_secrets.clone(), )); let mut outbox_shutdown = shutdown_signal.clone(); + let mut outbox_drain_shutdown = shutdown_signal.clone(); let outbox_task = tokio::spawn(async move { let mut interval = tokio::time::interval(std::time::Duration::from_secs(5)); loop { @@ -162,13 +163,17 @@ async fn main() -> Result<(), Box> { break; } _ = interval.tick() => { - // Runs to completion once selected: a drain pass already - // in progress is never aborted mid-write by a shutdown - // signal that arrives after this branch was chosen. + // A drain pass already in progress is never aborted + // mid-write by a shutdown signal that arrives after this + // branch was chosen -- but `execute_pending_mutations` + // itself races each item's execution against + // `outbox_drain_shutdown`, so a hung remote op cannot + // block this pass (and therefore shutdown) indefinitely. let summary = nunciod::outbox::execute_pending_mutations( &db_outbox, outbox_env.as_ref(), 50, + &mut outbox_drain_shutdown, ) .await; if summary.completed > 0 || summary.failed > 0 { diff --git a/crates/nunciod/src/outbox.rs b/crates/nunciod/src/outbox.rs index 062c38a..0d5bb8b 100644 --- a/crates/nunciod/src/outbox.rs +++ b/crates/nunciod/src/outbox.rs @@ -16,7 +16,9 @@ //! exceeded, at which point it flips to `failed`. A permanent failure (the //! message no longer exists, a malformed payload, an unknown action) flips to //! `failed` immediately -- there is nothing a retry could fix. Success is -//! never fabricated. +//! never fabricated. A mutation whose execution exceeds +//! [`PER_ITEM_EXECUTION_TIMEOUT`] is dispositioned the same way as any other +//! transient failure -- it is RETRYABLE, never `completed`. //! //! # Testability //! @@ -26,6 +28,7 @@ //! without any live network. Production wires [`ProductionExecutionEnv`], //! which resolves real per-account engines from the store and the OS keyring. +use crate::lifecycle::ShutdownSignal; use async_trait::async_trait; use nuncio_core::AccountConfig; use nuncio_filter::{ @@ -39,8 +42,18 @@ use nuncio_mail::{ use nuncio_store::db::{DatabaseEngine, DatabaseError}; use nuncio_store::vault::{SecretManager, VaultError, WEBHOOK_SIGNING_KEY_ACCOUNT}; use std::sync::Arc; +use std::time::Duration; use thiserror::Error; +/// Bound on how long a single outbox mutation may run before it is treated +/// as a retryable timeout. A healthy IMAP/JMAP/SMTP round trip or webhook +/// POST comfortably finishes well inside this window; a value this generous +/// still keeps one hung remote op from stalling every OTHER pending +/// mutation -- across every account -- behind it in the same drain pass, and +/// from blocking graceful shutdown for longer than the shutdown grace +/// period tolerates. +const PER_ITEM_EXECUTION_TIMEOUT: Duration = Duration::from_secs(30); + /// Errors that can occur while resolving the resources needed to execute a /// single outbox mutation. These are surfaced from a [`RemoteExecutionEnv`] /// and treated as transient (retryable) by the executor unless it has already @@ -119,11 +132,18 @@ pub struct ExecutionSummary { /// Drain up to `limit` pending mutations, executing each against `env` and /// persisting the honest outcome. Never marks a mutation `completed` unless the -/// real operation succeeded. Returns a tally of the pass. +/// real operation succeeded. Each item's execution is bounded by +/// [`PER_ITEM_EXECUTION_TIMEOUT`] so one hung remote op cannot freeze delivery +/// for every other item behind it; `shutdown` is raced against that same +/// timeout so a hung item cannot delay graceful shutdown either -- on +/// shutdown the drain pass stops immediately, leaving the in-flight item (and +/// anything still queued behind it) untouched and `pending` for the next +/// start. Returns a tally of the pass. pub async fn execute_pending_mutations( db: &DatabaseEngine, env: &dyn RemoteExecutionEnv, limit: usize, + shutdown: &mut ShutdownSignal, ) -> ExecutionSummary { let pending = match db.list_pending_mutations(limit).await { Ok(items) => items, @@ -135,7 +155,23 @@ pub async fn execute_pending_mutations( let mut summary = ExecutionSummary::default(); for item in pending { - let disposition = execute_one(db, env, &item).await; + let outcome = tokio::select! { + biased; + _ = shutdown.wait() => { + tracing::info!( + mutation_id = %item.id, + "outbox: drain pass interrupted by shutdown signal; item left pending" + ); + break; + } + result = tokio::time::timeout(PER_ITEM_EXECUTION_TIMEOUT, execute_one(db, env, &item)) => result, + }; + let disposition = match outcome { + Ok(disposition) => disposition, + Err(_elapsed) => Disposition::Retry(format!( + "mutation execution exceeded the {PER_ITEM_EXECUTION_TIMEOUT:?} timeout" + )), + }; let next_retry = item.retry_count + 1; let (status, retry, tally): (&str, i32, fn(&mut ExecutionSummary)) = match disposition { Disposition::Completed => ("completed", item.retry_count, |s| s.completed += 1), diff --git a/crates/nunciod/tests/outbox_executor_e2e_test.rs b/crates/nunciod/tests/outbox_executor_e2e_test.rs index d88509a..d6fd97f 100644 --- a/crates/nunciod/tests/outbox_executor_e2e_test.rs +++ b/crates/nunciod/tests/outbox_executor_e2e_test.rs @@ -13,16 +13,28 @@ use async_trait::async_trait; use nuncio_core::model::Email; +use nuncio_core::EventBus; use nuncio_filter::{FilterEngine, NsqlParser, ValidationOptions, WebhookDispatcher, WebhookError}; use nuncio_mail::{ MailBackend, MessageSender, MockMailBackend, MockMessageSender, RemoteMutationKind, }; use nuncio_store::db::DatabaseEngine; +use nunciod::lifecycle::{ShutdownController, ShutdownSignal}; use nunciod::outbox::{execute_pending_mutations, RemoteExecutionEnv}; use nunciod::sync::apply_filter_actions; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::Arc; use wiremock::matchers::{method, path}; use wiremock::{Mock, MockServer, ResponseTemplate}; +/// A shutdown handle for tests that never trigger it. Holding the returned +/// `ShutdownController` alive for the duration of the test matters: dropping +/// it closes the underlying `watch` channel, which would make `wait()` +/// resolve immediately and be mistaken for a real shutdown request. +fn no_shutdown() -> (ShutdownController, ShutdownSignal) { + ShutdownController::new(Arc::new(EventBus::new())) +} + const ACCOUNT_ID: &str = "acct-outbox-e2e"; const FOLDER_ID: &str = "INBOX"; const CHECKPOINT: &str = "1:105"; @@ -145,6 +157,54 @@ async fn seed_and_enqueue(db: &DatabaseEngine, rule_nsql: &str, subject: &str) { ); } +/// Like [`seed_and_enqueue`], but for a message with its own surrogate id so +/// several calls in the same test enqueue mutations against DISTINCT +/// messages instead of colliding on the shared fixed id `sample_email` +/// otherwise defaults to. +async fn seed_and_enqueue_distinct( + db: &DatabaseEngine, + message_id: &str, + rule_nsql: &str, + subject: &str, +) { + db.save_account(&sample_account()) + .await + .expect("save account"); + db.save_folder_sync_state(ACCOUNT_ID, FOLDER_ID, CHECKPOINT) + .await + .expect("save folder checkpoint"); + + // `remote_id`/`uid_validity` (not just `id`) are part of the store's + // `messages` identity UNIQUE index -- leaving them at `sample_email`'s + // shared default would make this INSERT OR REPLACE the OTHER seeded + // message sharing that same remote identity out from under it. + let mut email = sample_email(subject); + email.id = message_id.to_string(); + email.remote_id = message_id.to_string(); + db.save_email(&email).await.expect("save email"); + + let before = db + .list_pending_mutations(100) + .await + .expect("list pending") + .len(); + + let rule = NsqlParser::parse_rule("Outbox Rule", 1, rule_nsql).expect("parse rule"); + let engine = FilterEngine::new(vec![rule]).expect("compile rule"); + apply_filter_actions(db, &engine, &email).await; + + let after = db + .list_pending_mutations(100) + .await + .expect("list pending") + .len(); + assert_eq!( + after, + before + 1, + "the rule must enqueue exactly one new mutation" + ); +} + #[tokio::test] async fn move_rule_executes_against_backend_and_completes_with_uidvalidity_checkpoint() { let (db, _dir) = DatabaseEngine::connect_ephemeral().await.expect("db"); @@ -156,7 +216,8 @@ async fn move_rule_executes_against_backend_and_completes_with_uidvalidity_check .await; let env = MockEnv::new(); - let summary = execute_pending_mutations(&db, &env, 50).await; + let (_shutdown_ctrl, mut shutdown) = no_shutdown(); + let summary = execute_pending_mutations(&db, &env, 50, &mut shutdown).await; assert_eq!(summary.completed, 1); assert_eq!(summary.failed, 0); @@ -192,7 +253,8 @@ async fn flag_rule_executes_as_set_flagged() { .await; let env = MockEnv::new(); - let summary = execute_pending_mutations(&db, &env, 50).await; + let (_shutdown_ctrl, mut shutdown) = no_shutdown(); + let summary = execute_pending_mutations(&db, &env, 50, &mut shutdown).await; assert_eq!(summary.completed, 1); let applied = env.backend.applied_mutations(); @@ -214,7 +276,8 @@ async fn delete_rule_executes_as_delete() { .await; let env = MockEnv::new(); - let summary = execute_pending_mutations(&db, &env, 50).await; + let (_shutdown_ctrl, mut shutdown) = no_shutdown(); + let summary = execute_pending_mutations(&db, &env, 50, &mut shutdown).await; assert_eq!(summary.completed, 1); let applied = env.backend.applied_mutations(); @@ -234,7 +297,8 @@ async fn a_failing_backend_op_is_never_marked_completed() { let env = MockEnv::new(); env.backend.set_should_fail(true); - let summary = execute_pending_mutations(&db, &env, 50).await; + let (_shutdown_ctrl, mut shutdown) = no_shutdown(); + let summary = execute_pending_mutations(&db, &env, 50, &mut shutdown).await; // The op failed: nothing recorded, nothing completed, and the item is kept // for retry (still pending, with a bumped retry count) -- never faked. @@ -258,7 +322,8 @@ async fn forward_rule_sends_via_the_message_sender() { .await; let env = MockEnv::new(); - let summary = execute_pending_mutations(&db, &env, 50).await; + let (_shutdown_ctrl, mut shutdown) = no_shutdown(); + let summary = execute_pending_mutations(&db, &env, 50, &mut shutdown).await; assert_eq!(summary.completed, 1); let sent = env.sender.sent_messages(); @@ -293,7 +358,8 @@ async fn webhook_rule_dispatches_a_signed_post_to_the_receiver() { .await; let env = MockEnv::new(); - let summary = execute_pending_mutations(&db, &env, 50).await; + let (_shutdown_ctrl, mut shutdown) = no_shutdown(); + let summary = execute_pending_mutations(&db, &env, 50, &mut shutdown).await; // Completion is genuine: wiremock's `.expect(1)` verifies the POST actually // arrived when the server drops. @@ -320,7 +386,8 @@ async fn a_mutation_for_a_vanished_message_fails_permanently() { .expect("save mutation"); let env = MockEnv::new(); - let summary = execute_pending_mutations(&db, &env, 50).await; + let (_shutdown_ctrl, mut shutdown) = no_shutdown(); + let summary = execute_pending_mutations(&db, &env, 50, &mut shutdown).await; // Cannot act on a message that no longer exists -- fail honestly, never // fabricate success, and never touch the backend. @@ -332,3 +399,219 @@ async fn a_mutation_for_a_vanished_message_fails_permanently() { "a permanently-failed mutation is no longer pending" ); } + +/// A [`RemoteExecutionEnv`] whose `mail_backend` call hangs (a `tokio::time` +/// sleep far longer than any timeout under test, standing in for a stuck +/// IMAP/JMAP op) for the first `hang_calls` invocations, then delegates to a +/// normal [`MockEnv`]. Used to prove the per-item execution timeout fires. +struct HangEnv { + inner: MockEnv, + calls: AtomicUsize, + hang_calls: usize, +} + +impl HangEnv { + fn new(hang_calls: usize) -> Self { + Self { + inner: MockEnv::new(), + calls: AtomicUsize::new(0), + hang_calls, + } + } +} + +#[async_trait] +impl RemoteExecutionEnv for HangEnv { + async fn mail_backend( + &self, + account_id: &str, + ) -> Result, nunciod::outbox::OutboxExecuteError> { + let call = self.calls.fetch_add(1, Ordering::SeqCst); + if call < self.hang_calls { + // A `tokio::time`-tracked sleep far longer than any per-item + // timeout under test -- NOT `std::future::pending`, which never + // registers with the time driver and so behaves unpredictably + // once `tokio::time::pause`'s auto-advance-on-idle interacts with + // real (non-timer) I/O elsewhere in the same test, like the + // SQLite pool's own connection bookkeeping. A real (virtual) + // sleep is what the timeout under test is meant to race against. + tokio::time::sleep(std::time::Duration::from_secs(3600)).await; + } + self.inner.mail_backend(account_id).await + } + + async fn message_sender( + &self, + account_id: &str, + ) -> Result, nunciod::outbox::OutboxExecuteError> { + self.inner.message_sender(account_id).await + } + + async fn dispatch_webhook( + &self, + url: &str, + rule_id: &str, + message_id: &str, + subject: &str, + sender: &str, + ) -> Result { + self.inner + .dispatch_webhook(url, rule_id, message_id, subject, sender) + .await + } +} + +/// Seed two mutations so the second proves the drain pass actually moves on +/// after the first times out, rather than the whole pass having stalled. +async fn seed_two_move_mutations(db: &DatabaseEngine) { + seed_and_enqueue_distinct( + db, + "surrogate-hang-1", + "WHERE subject CONTAINS 'Archive' ACTION MOVE TO 'Archive'", + "Please Archive this one", + ) + .await; + seed_and_enqueue_distinct( + db, + "surrogate-hang-2", + "WHERE subject CONTAINS 'Trash' ACTION DELETE", + "Please Trash this one", + ) + .await; +} + +#[tokio::test] +async fn a_hung_mutation_times_out_and_is_retried_never_completed() { + let (db, _dir) = DatabaseEngine::connect_ephemeral().await.expect("db"); + seed_and_enqueue( + &db, + "WHERE subject CONTAINS 'Archive' ACTION MOVE TO 'Archive'", + "Please Archive", + ) + .await; + + // Paused only AFTER setup, and this pass touches the database exactly + // ONCE (a single `get_message`) before the hang -- mixing + // `tokio::time::pause`'s auto-advance-on-idle with real (non-timer) + // background I/O is only safe when there is a single, already-in-flight + // real operation for the auto-advance to race past; see the sibling + // two-pass test below for why a SECOND real DB operation in the same + // paused pass is deliberately avoided. + tokio::time::pause(); + let env = HangEnv::new(usize::MAX); + let (_shutdown_ctrl, mut shutdown) = no_shutdown(); + let summary = execute_pending_mutations(&db, &env, 10, &mut shutdown).await; + // Resume real time before any further DB access: `tokio::time::pause`'s + // auto-advance-on-idle only plays safely with the ONE real DB operation + // this pass performs (the `get_message` before the hang); a second real + // query made while still paused is what made the sibling two-pass test + // flaky before it was split across a resume point. + tokio::time::resume(); + + // The hung item is NEVER marked completed -- it is a retryable timeout, + // exactly like any other transient remote failure. + assert_eq!(summary.completed, 0); + assert_eq!( + summary.retried, 1, + "the hung item must be retried, not lost" + ); + assert_eq!(summary.failed, 0); + + let pending = db.list_pending_mutations(10).await.unwrap(); + assert_eq!(pending.len(), 1, "the hung item stays pending for a retry"); + assert_eq!( + pending[0].retry_count, 1, + "the timeout bumps the retry count exactly like any transient failure" + ); + assert!( + env.inner.backend.applied_mutations().is_empty(), + "the hung item never genuinely reached the backend" + ); +} + +#[tokio::test] +async fn a_timed_out_item_does_not_block_a_later_item_in_a_subsequent_pass() { + let (db, _dir) = DatabaseEngine::connect_ephemeral().await.expect("db"); + seed_two_move_mutations(&db).await; + + // First pass: bound to the FIRST (created-earliest) mutation only, so + // this pass's real DB work is a single `get_message` call before the + // hang, same as the sibling single-item timeout test. Real threaded + // SQLite I/O racing `tokio::time::pause`'s auto-advance is only + // exercised once per paused pass here -- see that test's comment for + // why a second real DB call in the SAME paused pass is unreliable. + tokio::time::pause(); + let env = HangEnv::new(usize::MAX); + let (_shutdown_ctrl, mut shutdown) = no_shutdown(); + let first_pass = execute_pending_mutations(&db, &env, 1, &mut shutdown).await; + assert_eq!(first_pass.completed, 0); + assert_eq!(first_pass.retried, 1, "the hung item must be retried"); + + // Second pass runs under ordinary real time against a normal + // (non-hanging) env, standing in for the outbox worker's next 5-second + // tick: the item that timed out in the first pass did not jam the queue + // -- both it (now against a healthy backend) and the item queued behind + // it are processed and genuinely complete. + tokio::time::resume(); + let normal_env = MockEnv::new(); + let (_shutdown_ctrl2, mut shutdown2) = no_shutdown(); + let second_pass = execute_pending_mutations(&db, &normal_env, 10, &mut shutdown2).await; + assert_eq!( + second_pass.completed, 2, + "both the previously-timed-out item and the one behind it complete" + ); + assert_eq!(second_pass.failed, 0); + + let applied = normal_env.backend.applied_mutations(); + assert_eq!( + applied.len(), + 2, + "both mutations genuinely reached the backend" + ); + + // Neither pass ever fabricated a completion: the queue is now empty + // because both items genuinely succeeded, not because either was + // dropped or silently marked done while still hung. + assert!(db.list_pending_mutations(10).await.unwrap().is_empty()); +} + +#[tokio::test] +async fn shutdown_signal_interrupts_a_hung_item_instead_of_waiting_out_the_timeout() { + let (db, _dir) = DatabaseEngine::connect_ephemeral().await.expect("db"); + seed_and_enqueue( + &db, + "WHERE subject CONTAINS 'Archive' ACTION MOVE TO 'Archive'", + "Please Archive", + ) + .await; + + // Paused only after setup -- see the comment on the sibling timeout test + // for why pausing from the start would spuriously fail the DB pool. + tokio::time::pause(); + + // Hangs on every call -- the shutdown signal, not the timeout, must be + // what unblocks this drain pass. + let env = HangEnv::new(usize::MAX); + let (controller, mut shutdown) = ShutdownController::new(Arc::new(EventBus::new())); + tokio::spawn(async move { + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + controller.trigger(); + }); + + let summary = execute_pending_mutations(&db, &env, 10, &mut shutdown).await; + tokio::time::resume(); + + // The drain pass stopped on shutdown, well before the per-item timeout + // would have elapsed -- the hung item is untouched: not completed, not + // even retried, since its true disposition is still unknown. + assert_eq!(summary.completed, 0); + assert_eq!(summary.retried, 0); + assert_eq!(summary.failed, 0); + + let pending = db.list_pending_mutations(10).await.unwrap(); + assert_eq!(pending.len(), 1); + assert_eq!( + pending[0].retry_count, 0, + "shutdown must leave the interrupted item's disposition untouched" + ); +} -- 2.54.0 From ad33b639c44a881f5a0c6315b1d132bb9d48f612 Mon Sep 17 00:00:00 2001 From: James Maes Date: Fri, 31 Jul 2026 16:34:19 -0500 Subject: [PATCH 2/2] feat(proto): keyset pagination for all list/search RPCs 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 --- 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