diff --git a/crates/nuncio-mail/src/jmap.rs b/crates/nuncio-mail/src/jmap.rs index 9f307de..a29bf62 100644 --- a/crates/nuncio-mail/src/jmap.rs +++ b/crates/nuncio-mail/src/jmap.rs @@ -451,18 +451,14 @@ impl JmapEngine { } } - /// Parse a raw JMAP `Email/get` JSON response payload into a list of - /// [`PlacedMessage`] and the new state string. - pub fn parse_email_get_response( - &self, - raw_json: &str, - ) -> Result<(Vec, String), MailError> { - self.parse_email_get_response_in_folder(raw_json, "inbox") - } - /// Parse a raw JMAP `Email/get` JSON response payload into the occupancies /// it establishes for `folder_id`, plus the account's Email state string. /// + /// The caller supplies the mailbox being parsed for because the placements + /// this produces are keyed by it: a placement labelled with a folder the + /// account does not sync can never be matched again, so a fixed label would + /// silently strand every message it produced. + /// /// A returned message whose `mailboxIds` is present and does NOT name /// `folder_id` is dropped rather than placed: `Email/changes` is /// account-wide (RFC 8621 section 4.2 offers no mailbox filter), so a @@ -470,7 +466,7 @@ impl JmapEngine { /// here would invent an occupancy the server never reported. Membership /// that the server did not return at all is left alone -- unknown is not /// absent. - fn parse_email_get_response_in_folder( + pub fn parse_email_get_response( &self, raw_json: &str, folder_id: &str, @@ -765,7 +761,7 @@ impl JmapEngine { let get_request = Self::build_email_get_request(jmap_account_id, Some(ids)); let get_raw = self.post_jmap(api_url, &get_request).await?; - let (emails, state) = self.parse_email_get_response_in_folder(&get_raw, folder_id)?; + let (emails, state) = self.parse_email_get_response(&get_raw, folder_id)?; let present: Vec = emails.iter().map(|m| m.placement.key()).collect(); @@ -810,7 +806,7 @@ impl JmapEngine { let get_request = Self::build_email_get_request(jmap_account_id, Some(changes.changed.clone())); let get_raw = self.post_jmap(api_url, &get_request).await?; - let (emails, _state) = self.parse_email_get_response_in_folder(&get_raw, folder_id)?; + let (emails, _state) = self.parse_email_get_response(&get_raw, folder_id)?; emails }; @@ -1118,7 +1114,7 @@ mod tests { }"#; let engine = JmapEngine::new("acct-1"); let (emails, _state) = engine - .parse_email_get_response_in_folder(raw, "mb-inbox") + .parse_email_get_response(raw, "mb-inbox") .expect("parse get"); let ids: Vec<&str> = emails .iter() @@ -1142,7 +1138,9 @@ mod tests { ] }"#; let engine = JmapEngine::new("acct-1"); - let (emails, _state) = engine.parse_email_get_response(raw).expect("parse get"); + let (emails, _state) = engine + .parse_email_get_response(raw, "mb-inbox") + .expect("parse get"); assert_eq!( emails[0].email.message_id.as_deref(), @@ -1160,7 +1158,7 @@ mod tests { #[test] fn parse_invalid_jmap_json_fails() { let engine = JmapEngine::new("acct-1"); - let res = engine.parse_email_get_response("{ invalid json }"); + let res = engine.parse_email_get_response("{ invalid json }", "mb-inbox"); assert!(res.is_err()); } diff --git a/crates/nuncio-mail/tests/imap_identity_test.rs b/crates/nuncio-mail/tests/imap_identity_test.rs index 06a2f5a..38ec07a 100644 --- a/crates/nuncio-mail/tests/imap_identity_test.rs +++ b/crates/nuncio-mail/tests/imap_identity_test.rs @@ -193,7 +193,7 @@ async fn an_imap_emailid_and_a_jmap_object_id_produce_the_same_key() { "receivedAt":1700001000,"isUnread":true}}]}},"c1"]]}}"# ); let (jmap, _state) = JmapEngine::new(ACCOUNT) - .parse_email_get_response(&jmap_response) + .parse_email_get_response(&jmap_response, "mb-inbox") .expect("the JMAP response parses"); assert_eq!(imap.upserts.len(), 1); diff --git a/crates/nuncio-mail/tests/wiremock_jmap_test.rs b/crates/nuncio-mail/tests/wiremock_jmap_test.rs index d330f76..7033966 100644 --- a/crates/nuncio-mail/tests/wiremock_jmap_test.rs +++ b/crates/nuncio-mail/tests/wiremock_jmap_test.rs @@ -81,7 +81,7 @@ async fn wiremock_jmap_session_discovery_and_email_get_sync() { let engine = JmapEngine::new("acct-100"); let (emails, new_state) = engine - .parse_email_get_response(&resp.text().await.unwrap()) + .parse_email_get_response(&resp.text().await.unwrap(), "mb-inbox") .expect("parse email get"); assert_eq!(emails.len(), 1); diff --git a/crates/nuncio-store/src/db.rs b/crates/nuncio-store/src/db.rs index 919a2ab..12ef57e 100644 --- a/crates/nuncio-store/src/db.rs +++ b/crates/nuncio-store/src/db.rs @@ -1777,6 +1777,71 @@ impl DatabaseEngine { .collect()) } + /// Every mailbox each of `message_keys` currently occupies, grouped by key. + /// + /// The batched form of [`Self::placements_of`], for callers holding a whole + /// page of messages: one query per chunk instead of one per message. Keys + /// absent from the result occupy no mailbox; callers that need an entry for + /// every requested key should treat a missing entry as an empty set. + /// + /// Each key's placements are ordered by folder id ascending, exactly as + /// [`Self::placements_of`] orders them -- callers take the first placement + /// as the message's canonical mailbox, so the order is part of the + /// contract, not an incidental property of the query plan. + /// + /// Chunked at [`MAX_KEYS_PER_IN_LIST`] for the same reason as + /// [`Self::existing_placements`]: the bind-parameter ceiling belongs to the + /// statement, so it can only be enforced where the statement is built. + /// Duplicate keys are collapsed first, so no key's placements can be + /// gathered from two chunks and appended twice. + pub async fn placements_of_batch( + &self, + message_keys: &[String], + ) -> Result>, DatabaseError> + { + let mut seen = std::collections::HashSet::new(); + let unique: Vec<&String> = message_keys + .iter() + .filter(|key| seen.insert(key.as_str())) + .collect(); + + let mut grouped: std::collections::HashMap> = + std::collections::HashMap::new(); + + for chunk in unique.chunks(MAX_KEYS_PER_IN_LIST) { + let mut builder: sqlx::QueryBuilder<'_, sqlx::Sqlite> = sqlx::QueryBuilder::new( + "SELECT message_key, account_id, folder_id, uidvalidity, uid, read_flag + FROM placements WHERE message_key IN (", + ); + let mut separated = builder.separated(", "); + for key in chunk { + separated.push_bind(*key); + } + builder.push(") ORDER BY message_key ASC, folder_id ASC"); + + let rows: Vec<(String, String, String, String, String, i64)> = builder + .build_query_as() + .fetch_all(&self.pool) + .await + .map_err(DatabaseError::Query)?; + + for (message_key, account_id, folder_id, uid_validity, remote_id, read_flag) in rows { + grouped + .entry(message_key) + .or_default() + .push(nuncio_core::model::Placement { + account_id, + folder_id, + uid_validity, + remote_id, + read: read_flag != 0, + }); + } + } + + Ok(grouped) + } + /// Messages occupying one folder of one account, newest first, paired with /// the placement they were found through. /// @@ -5494,6 +5559,83 @@ mod tests { assert_eq!(known.len(), 1, "and the query must not echo its own input"); } + #[tokio::test] + async fn placements_of_batch_matches_per_message_placements_of_including_order() { + let (engine, _dir) = DatabaseEngine::connect_ephemeral().await.unwrap(); + + // Folders are saved in an order that is neither insertion-sorted nor + // reverse-sorted, so a batch query that forgot to order by folder id + // would come back visibly wrong rather than accidentally right. + for (key, folder, uid) in [ + ("key-a", "Zeta", "1"), + ("key-a", "Alpha", "2"), + ("key-a", "Mid", "3"), + ("key-b", "Sent", "4"), + ("key-b", "Archive", "5"), + // Placed but never asked for: proves the `IN (...)` filter is real. + ("key-unrequested", "INBOX", "6"), + ] { + let (email, placement) = sample_message_and_placement(key, folder, uid); + engine + .save_email_at( + &email, + nuncio_core::model::IdentitySource::EmailId, + &placement, + ) + .await + .unwrap(); + } + + // "key-a" appears twice: a caller may hand over a page with repeats, + // and a key gathered from two chunks must not be appended twice. + let requested = [ + "key-a".to_string(), + "key-b".to_string(), + "key-a".to_string(), + "key-never-saved".to_string(), + ]; + let grouped = engine.placements_of_batch(&requested).await.unwrap(); + + for key in ["key-a", "key-b"] { + let one_at_a_time = engine.placements_of(key).await.unwrap(); + assert!( + one_at_a_time.len() > 1 || key == "key-b", + "the fixture must exercise more than one placement per message" + ); + assert_eq!( + grouped.get(key), + Some(&one_at_a_time), + "the batched read of '{key}' must equal the per-message read, element for element" + ); + + let folders: Vec<&str> = one_at_a_time + .iter() + .map(|placement| placement.folder_id.as_str()) + .collect(); + let mut sorted = folders.clone(); + sorted.sort_unstable(); + assert_eq!( + folders, sorted, + "the first placement of '{key}' is taken as its canonical mailbox, so folder id order is load-bearing" + ); + } + + assert_eq!( + grouped.get("key-a").map(Vec::len), + Some(3), + "a repeated key must not have its placements gathered twice" + ); + assert!( + !grouped.contains_key("key-never-saved"), + "a key that occupies no mailbox must be absent, not invented" + ); + assert!( + !grouped.contains_key("key-unrequested"), + "the batch must answer only for the keys it was asked about" + ); + assert_eq!(grouped.len(), 2, "and for nothing else"); + } + /// A first sync of a real mailbox hands `existing_placements` the entire /// folder in one call. Each key spends four bind parameters, and SQLite /// caps a statement at 32766 of them, so an unchunked `IN (...)` list dies diff --git a/crates/nunciod/src/grpc.rs b/crates/nunciod/src/grpc.rs index 317df20..481c2a8 100644 --- a/crates/nunciod/src/grpc.rs +++ b/crates/nunciod/src/grpc.rs @@ -1025,25 +1025,25 @@ fn map_placement_ref_from_proto(placement: PlacementRefProto) -> nuncio_core::mo /// Fills each message's complete placement set from the store. /// -/// This costs one query per message rather than one per page. That is the -/// honest price of the field: occupancy is per-message state and a page can mix -/// messages that sit in one mailbox with messages that sit in ten. A failure to -/// read one message's placements leaves that message's set empty rather than -/// failing the whole listing, and is logged -- a partial page is more useful -/// than none, and the field is additive so a client that ignores it is -/// unaffected. +/// One batched query for the whole page rather than one per message: occupancy +/// is per-message state, but reading it per message makes a page of fifty cost +/// fifty-one round trips, which shows up directly as page latency. A failure to +/// read the batch leaves every set empty rather than failing the whole listing, +/// and is logged -- a page without placements is more useful than none, and the +/// field is additive so a client that ignores it is unaffected. async fn fill_placements(db: &DatabaseEngine, messages: &mut [MessageProto]) { + let keys: Vec = messages.iter().map(|message| message.id.clone()).collect(); + let mut grouped = match db.placements_of_batch(&keys).await { + Ok(grouped) => grouped, + Err(e) => { + tracing::warn!("Mail: failed to read the placement sets for this page: {e}"); + return; + } + }; + for message in messages.iter_mut() { - match db.placements_of(&message.id).await { - Ok(placements) => { - message.placements = placements.into_iter().map(map_placement_to_proto).collect(); - } - Err(e) => { - tracing::warn!( - message_id = %message.id, - "Mail: failed to read the placement set: {e}" - ); - } + if let Some(placements) = grouped.remove(&message.id) { + message.placements = placements.into_iter().map(map_placement_to_proto).collect(); } } }