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

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<S>(
+    pub async fn sync_folder_changes_with_session<S>(
         &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<S>(
+    async fn sync_folder_changes_batched<S>(
         &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<FolderChanges, MailError> {
-        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);

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();