diff --git a/crates/nuncio-core/src/export.rs b/crates/nuncio-core/src/export.rs index 79a71e4..83fd163 100644 --- a/crates/nuncio-core/src/export.rs +++ b/crates/nuncio-core/src/export.rs @@ -87,10 +87,15 @@ impl ExportEngine { writer.write_all(from_line.as_bytes())?; total_bytes += from_line.len() as u64; - // Headers. No X-Nuncio-Folder header: the folder is a property of - // a placement now, not of the message, so `Email` has none to - // name here. Restoring it would mean threading the relevant - // `Placement` into this function's signature. + // Headers. `X-Nuncio-Folder` is permanently gone, not pending: a + // message's folder is a property of a placement, and a message may + // hold several at once (the same message filed in INBOX and in + // Archive is one message, not two). A single header could therefore + // only ever name one of them, which would be a wrong answer dressed + // as an authoritative one -- and MBOX has no way to spell "these + // several". Callers that need occupancy read it from the store, + // where it is complete. `X-Nuncio-Account` stays because an account + // *is* single-valued on the message. let headers = format!( "Message-ID: <{}>\r\nFrom: {}\r\nTo: {}\r\nSubject: {}\r\nDate: {}\r\nX-Nuncio-Account: {}\r\nContent-Type: text/plain; charset=utf-8\r\n\r\n", email.id, email.sender, email.recipient, email.subject, date_str, email.account_id diff --git a/crates/nuncio-mail/src/imap.rs b/crates/nuncio-mail/src/imap.rs index fe93558..57c5922 100644 --- a/crates/nuncio-mail/src/imap.rs +++ b/crates/nuncio-mail/src/imap.rs @@ -798,7 +798,7 @@ impl ImapEngine { /// returned checkpoint is derived from real server state -- the mailbox's /// UIDVALIDITY plus its `UIDNEXT` (or one past the highest UID actually /// seen) after selecting the folder -- never from a synthetic counter. - pub async fn sync_folder_messages_with_session( + pub async fn sync_folder_changes_with_session( &self, folder_id: &str, since_state: Option<&str>, @@ -807,7 +807,7 @@ impl ImapEngine { where S: AsyncRead + AsyncWrite + Unpin + Send + Debug + 'static, { - self.sync_folder_messages_batched( + self.sync_folder_changes_batched( folder_id, since_state, session, @@ -842,7 +842,7 @@ impl ImapEngine { fields(account_id = %self.account_id, folder_id = %folder_id), err )] - async fn sync_folder_messages_batched( + async fn sync_folder_changes_batched( &self, folder_id: &str, since_state: Option<&str>, @@ -923,6 +923,14 @@ impl ImapEngine { drive_fetch_stream(&mut fetch_stream, folder_id, emails.len()).await? }; for item in &items { + // The enumeration pass rejected UID-less items; a response to a + // `UID FETCH` must carry one (RFC 3501 section 6.4.8), so this + // only fires for a non-conforming server. Recording it anyway + // would write a placement at the meaningless UID 0. + if item.uid.filter(|u| *u >= 1).is_none() { + skipped_no_uid += 1; + continue; + } emails.push(self.build_email_from_fetch(folder_id, server_uid_validity, item)?); } } @@ -945,6 +953,12 @@ impl ImapEngine { drive_fetch_stream(&mut fetch_stream, folder_id, emails.len()).await? }; for item in &items { + // Same guard as the body pass: an item with no UID cannot be + // placed, and UID 0 is not an address. + if item.uid.filter(|u| *u >= 1).is_none() { + skipped_no_uid += 1; + continue; + } emails.push(self.build_headers_only_email(folder_id, server_uid_validity, item)); } } @@ -1497,8 +1511,8 @@ impl ImapEngine { /// Fetch a folder's messages over a freshly authenticated session, /// honoring `since_state` for an incremental fetch and returning the new /// sync checkpoint alongside the messages. See - /// [`Self::sync_folder_messages_with_session`] for the checkpoint model. - pub async fn sync_folder_messages( + /// [`Self::sync_folder_changes_with_session`] for the checkpoint model. + pub async fn sync_folder_changes( &self, folder_id: &str, since_state: Option<&str>, @@ -1517,7 +1531,7 @@ impl ImapEngine { .connect_session(username, password) .await?; let result = self - .sync_folder_messages_with_session(folder_id, since_state, &mut session) + .sync_folder_changes_with_session(folder_id, since_state, &mut session) .await; let _ = session.logout().await; result @@ -1971,7 +1985,7 @@ impl MailBackend for ImapEngine { folder_id: &str, since_state: Option<&str>, ) -> Result { - self.sync_folder_messages(folder_id, since_state).await + self.sync_folder_changes(folder_id, since_state).await } async fn apply_mutation( @@ -2327,7 +2341,7 @@ mod tests { // First sync: no checkpoint -> full history; returns // "{uidvalidity}:{uidnext}" from real server state. let first_pass = engine - .sync_folder_messages_with_session("INBOX", None, &mut session) + .sync_folder_changes_with_session("INBOX", None, &mut session) .await .expect("first sync succeeds"); let (first, checkpoint) = (first_pass.upserts, first_pass.next_state); @@ -2337,7 +2351,7 @@ mod tests { // Second sync: feed back the checkpoint; UIDVALIDITY still matches // (1), so the fetch is narrowed. let __changes = engine - .sync_folder_messages_with_session("INBOX", Some(&checkpoint), &mut session) + .sync_folder_changes_with_session("INBOX", Some(&checkpoint), &mut session) .await .expect("second sync succeeds"); let (_second, checkpoint2) = (__changes.upserts, __changes.next_state); @@ -2416,7 +2430,7 @@ mod tests { } } - /// Drive `sync_folder_messages_batched` over a scripted, in-memory IMAP + /// Drive `sync_folder_changes_batched` over a scripted, in-memory IMAP /// server that serves a fixed set of `(uid, size)` messages. The server /// answers the enumeration pass (UID + RFC822.SIZE), full-body batches /// (`BODY[]`), and headers-only batches (`BODY[HEADER]`) from the fixture, @@ -2522,7 +2536,7 @@ mod tests { .expect("scripted login succeeds"); let engine = ImapEngine::new("acct-1", "example.test", 993); let changes = engine - .sync_folder_messages_batched("INBOX", since_state.as_deref(), &mut session, batch_size) + .sync_folder_changes_batched("INBOX", since_state.as_deref(), &mut session, batch_size) .await .expect("batched sync succeeds"); drop(session); @@ -2779,7 +2793,7 @@ mod tests { // First sync under UIDVALIDITY 1 stores "1:105". let __changes = engine - .sync_folder_messages_with_session("INBOX", None, &mut session) + .sync_folder_changes_with_session("INBOX", None, &mut session) .await .expect("first sync succeeds"); let (_first, checkpoint) = (__changes.upserts, __changes.next_state); @@ -2789,7 +2803,7 @@ mod tests { // is stale, so the fetch MUST be a full `1:*`, and the checkpoint is // rewritten under the new UIDVALIDITY. let __changes = engine - .sync_folder_messages_with_session("INBOX", Some(&checkpoint), &mut session) + .sync_folder_changes_with_session("INBOX", Some(&checkpoint), &mut session) .await .expect("second sync succeeds"); let (_second, checkpoint2) = (__changes.upserts, __changes.next_state); @@ -2873,7 +2887,7 @@ mod tests { // elapsed time despite the 60s virtual timeout expiring. let started = std::time::Instant::now(); let err = engine - .sync_folder_messages_with_session("INBOX", None, &mut session) + .sync_folder_changes_with_session("INBOX", None, &mut session) .await .expect_err("a never-responding FETCH must surface an error"); assert!( @@ -3428,7 +3442,7 @@ mod tests { let engine = ImapEngine::new("acct-completion", "example.test", 993); let __changes = engine - .sync_folder_messages_with_session("INBOX", None, &mut session) + .sync_folder_changes_with_session("INBOX", None, &mut session) .await .expect("sync succeeds"); let (emails, _checkpoint) = (__changes.upserts, __changes.next_state); @@ -3576,7 +3590,7 @@ mod tests { .expect("scripted login succeeds"); let engine = ImapEngine::new("acct-1", "example.test", 993); let changes = engine - .sync_folder_messages_batched(folder_id, None, &mut session, 8) + .sync_folder_changes_batched(folder_id, None, &mut session, 8) .await .expect("scripted sync succeeds"); drop(session); diff --git a/crates/nuncio-store/src/db.rs b/crates/nuncio-store/src/db.rs index 9aa7253..24ced58 100644 --- a/crates/nuncio-store/src/db.rs +++ b/crates/nuncio-store/src/db.rs @@ -727,6 +727,21 @@ impl DatabaseEngine { /// fired the rule again. `INSERT ... ON CONFLICT DO NOTHING` makes winning /// the claim a single atomic statement, so two passes racing over the same /// message cannot both act. + /// + /// A claim outlives the message it was made against. Nothing reaps + /// `filter_fired` when the last placement goes, so a message deleted and + /// later re-delivered byte for byte will not fire its rules a second time. + /// That is deliberate: reaping alongside the message would hand a remote + /// party a way to re-trigger third-party side effects at will -- a + /// mailing-list message deleted and re-sent would `FORWARD` again -- and + /// under content-addressed identity an identical redelivery genuinely *is* + /// the same message, so declining to re-fire is the honest reading. The + /// failure this leaves open ("a rule that should have fired again did not") + /// is the safe direction for actions that reach outside the mailbox. + /// + /// The cost is a table that only grows, bounded by rules x messages ever + /// seen at roughly 150 bytes a row, and cleared only when an identity + /// schema bump resets `user_version` and rebuilds the derived tables. pub async fn claim_filter_fire( &self, rule_id: &str, @@ -1446,18 +1461,22 @@ impl DatabaseEngine { // an upsert, so after the fact there is no way to tell an inserted // placement from an updated one -- and the caller needs that distinction // to decide whether this is a first arrival worth filtering. - let (already_placed,): (i64,) = sqlx::query_as( - "SELECT COUNT(*) FROM placements + // The occupant's key is read, not just its existence, because the upsert below may + // repoint this slot at a different message and the previous key is unrecoverable + // afterwards. + let occupant: Option<(String,)> = sqlx::query_as( + "SELECT message_key FROM placements WHERE account_id = ? AND folder_id = ? AND uidvalidity = ? AND uid = ?", ) .bind(&placement.account_id) .bind(&placement.folder_id) .bind(&placement.uid_validity) .bind(&placement.remote_id) - .fetch_one(&mut *tx) + .fetch_optional(&mut *tx) .await .map_err(DatabaseError::Query)?; - let placement_is_new = already_placed == 0; + let placement_is_new = occupant.is_none(); + let displaced_key = occupant.map(|(key,)| key).filter(|key| *key != email.id); let inserted = sqlx::query( r#" @@ -1521,6 +1540,24 @@ impl DatabaseEngine { .await .map_err(DatabaseError::Query)?; + // Repointing a slot at a different message drops the displaced message's placement + // without it ever passing through [`Self::delete_placements`]. If that was its last + // one, nothing else will ever delete it -- and its decrypted body would stay readable + // in `messages_fts` forever, since only the `messages_ad` trigger clears that index. + // Reap it here on exactly the terms `delete_placements` uses: gone only when nothing + // points at it any more. + if let Some(displaced_key) = displaced_key { + sqlx::query( + "DELETE FROM messages WHERE message_key = ? + AND NOT EXISTS (SELECT 1 FROM placements WHERE message_key = ?)", + ) + .bind(&displaced_key) + .bind(&displaced_key) + .execute(&mut *tx) + .await + .map_err(DatabaseError::Query)?; + } + tx.commit().await.map_err(DatabaseError::Query)?; Ok(SaveOutcome { @@ -5055,6 +5092,90 @@ mod tests { ); } + #[tokio::test] + async fn repointing_a_slot_reaps_the_displaced_message_and_its_plaintext_index() { + let (engine, _dir) = DatabaseEngine::connect_ephemeral().await.unwrap(); + let (first, inbox) = sample_message_and_placement("key-1", "INBOX", "5"); + engine + .save_email_at(&first, nuncio_core::model::IdentitySource::EmailId, &inbox) + .await + .unwrap(); + + // The same mailbox slot re-saved under a different identity. The displaced + // message never passes through `delete_placements`, so this write is its only + // chance to be reaped. + let (second, _) = sample_message_and_placement("key-2", "INBOX", "5"); + engine + .save_email_at(&second, nuncio_core::model::IdentitySource::EmailId, &inbox) + .await + .unwrap(); + + engine + .get_message("key-1") + .await + .expect_err("the displaced message has no placements left"); + let (fts,): (i64,) = + sqlx::query_as("SELECT COUNT(*) FROM messages_fts WHERE message_key = 'key-1'") + .fetch_one(engine.pool()) + .await + .unwrap(); + assert_eq!( + fts, 0, + "a repoint that strands a message must not leave its decrypted body \ + readable in the FTS index" + ); + + engine + .get_message("key-2") + .await + .expect("the message that took the slot must survive"); + let placements = engine.placements_of("key-2").await.unwrap(); + assert_eq!(placements.len(), 1, "the slot holds exactly one occupant"); + } + + #[tokio::test] + async fn repointing_a_slot_keeps_a_displaced_message_that_still_lives_elsewhere() { + let (engine, _dir) = DatabaseEngine::connect_ephemeral().await.unwrap(); + let (first, inbox) = sample_message_and_placement("key-1", "INBOX", "5"); + let archive = nuncio_core::model::Placement { + folder_id: "Archive".into(), + remote_id: "9".into(), + ..inbox.clone() + }; + engine + .save_email_at(&first, nuncio_core::model::IdentitySource::EmailId, &inbox) + .await + .unwrap(); + engine + .save_email_at( + &first, + nuncio_core::model::IdentitySource::EmailId, + &archive, + ) + .await + .unwrap(); + + let (second, _) = sample_message_and_placement("key-2", "INBOX", "5"); + engine + .save_email_at(&second, nuncio_core::model::IdentitySource::EmailId, &inbox) + .await + .unwrap(); + + engine + .get_message("key-1") + .await + .expect("the displaced message still lives in Archive"); + let (fts,): (i64,) = + sqlx::query_as("SELECT COUNT(*) FROM messages_fts WHERE message_key = 'key-1'") + .fetch_one(engine.pool()) + .await + .unwrap(); + assert_eq!(fts, 1, "search must still find a message that still exists"); + let placements = engine.placements_of("key-1").await.unwrap(); + assert_eq!(placements.len(), 1); + assert_eq!(placements[0].folder_id, "Archive"); + } + #[tokio::test] async fn folders_are_derived_from_placements_and_scoped_per_account() { let (engine, _dir) = DatabaseEngine::connect_ephemeral().await.unwrap();