diff --git a/crates/nuncio-mail/src/backend.rs b/crates/nuncio-mail/src/backend.rs index cb999bd..c776170 100644 --- a/crates/nuncio-mail/src/backend.rs +++ b/crates/nuncio-mail/src/backend.rs @@ -65,19 +65,63 @@ pub struct RemoteMutationSpec { pub kind: RemoteMutationKind, } +/// What one folder-sync pass observed on the server. +/// +/// The reason this is not just a list of messages: a sync that can only report +/// what *is* there can never report what stopped being there. A message moved +/// or deleted by any other client -- another Nuncio daemon, a phone, webmail -- +/// simply stays in the local store forever. Expressing absence is the whole +/// point of the type. +#[derive(Debug, Default, Clone)] +pub struct FolderChanges { + /// Messages to store, whether newly arrived or changed. + pub upserts: Vec, + /// Surrogate ids the server **explicitly reported as gone**, via QRESYNC + /// `VANISHED`. Always safe to delete. + pub removals: Vec, + /// Surrogate ids of everything the folder currently holds, when this pass + /// enumerated the complete UID set. The caller may delete anything it + /// stores for this folder that is absent from this list. + /// + /// `None` and `Some(vec![])` mean different things, and conflating them + /// deletes a mailbox. `None` is "this pass was incremental and cannot + /// speak to absence"; `Some(vec![])` is "the folder is genuinely empty". + pub present: Option>, + /// The checkpoint to resume from next time. + pub next_state: String, +} + /// Protocol-agnostic mail backend engine trait implemented by JMAP and IMAP engines. #[async_trait] pub trait MailBackend: Send + Sync { /// Synchronize and list available mailbox folders. async fn sync_folders(&self) -> Result, MailError>; - /// Synchronize message envelopes for a specific folder since a checkpoint state. - /// Returns the updated email list and the new server state checkpoint. + /// Enumerate what changed in a folder since `since_state`. + /// + /// Replaces a fetch-only sync: implementations must report removals when + /// the protocol can express them, and must say honestly (via + /// [`FolderChanges::present`]) whether the pass was able to observe + /// absence at all. + async fn sync_changes( + &self, + folder_id: &str, + since_state: Option<&str>, + ) -> Result; + + /// Fetch-only view of [`MailBackend::sync_changes`], for callers that + /// genuinely only want the messages. + /// + /// Provided rather than required so no implementation can satisfy the + /// trait by supplying this and quietly never reporting a removal. async fn sync_messages( &self, folder_id: &str, since_state: Option<&str>, - ) -> Result<(Vec, String), MailError>; + ) -> Result<(Vec, String), MailError> { + let changes = self.sync_changes(folder_id, since_state).await?; + Ok((changes.upserts, changes.next_state)) + } /// Apply a remote mutation (move/copy/flag/delete) to a single message /// against the real server. MUST return `Ok(())` only when the server diff --git a/crates/nuncio-mail/src/imap.rs b/crates/nuncio-mail/src/imap.rs index 2de83f7..6e883bc 100644 --- a/crates/nuncio-mail/src/imap.rs +++ b/crates/nuncio-mail/src/imap.rs @@ -86,80 +86,163 @@ fn parse_uid_validity(uid_validity: &str) -> Option { /// first turns that into a safe full re-fetch instead. const CHECKPOINT_DELIM: char = ':'; -/// The `UID FETCH` sequence-set to issue for a folder sync, resolved from the -/// stored checkpoint. Kept as a small typed value (rather than collapsing to a -/// bare range string) so the distinct fall-back reasons stay observable: all -/// of `Full`/`CorruptCheckpoint`/`UidValidityChanged` fetch `1:*`, but only the -/// latter two are worth flagging, and tests can assert on which branch fired. +/// Scheme tag prefixing every checkpoint this build writes. +/// +/// The stored value is `"v1:uidnext:{uidvalidity}:{uidnext}"`. The tag exists +/// because the *next* checkpoint scheme -- a QRESYNC `(uidvalidity, +/// highestmodseq)` pair -- has exactly the same shape as this one and a wholly +/// different meaning. Untagged, an upgraded engine would hand a stored UIDNEXT +/// to a server as a MODSEQ. UIDNEXT is normally the larger number, so the +/// server would answer "nothing changed since" and every message below that +/// point would never be enumerated again: silent, permanent mail loss with no +/// error anywhere. +/// +/// Anything that does not carry a scheme this build recognises -- an unknown +/// tag, or a bare `"42:105"` written before tagging existed -- resolves to a +/// full re-fetch. Over-fetching is recoverable; misreading a token is not. +const CHECKPOINT_SCHEME_UIDNEXT: &str = "v1:uidnext"; + +/// Scheme for a mod-sequence checkpoint: `"v2:modseq:{uidvalidity}:{modseq}"`. +/// +/// Written by both mod-sequence rungs. QRESYNC and CONDSTORE differ in how +/// changes are *requested*, not in what has to be remembered -- both resume +/// from `(UIDVALIDITY, HIGHESTMODSEQ)` -- so one scheme covers both and a +/// mailbox can move between them without invalidating its checkpoint. +const CHECKPOINT_SCHEME_MODSEQ: &str = "v2:modseq"; + +/// Which mechanism a folder sync can use, in descending order of precision. +/// +/// Chosen **per mailbox, not per account**: `NOMODSEQ` is a mailbox property +/// (RFC 7162 section 3.1.2.2), so one folder on a CONDSTORE server can still be +/// unable to answer mod-sequence queries. #[derive(Debug, Clone, Copy, PartialEq, Eq)] -enum FetchRange { - /// No prior checkpoint (first-ever sync of this folder): fetch everything. +pub(crate) enum SyncRung { + /// `SELECT (QRESYNC …)` returns `VANISHED` plus changed messages in one + /// round trip. The server reports removals directly. + /// Fastmail/Cyrus, Dovecot. + Qresync, + /// `CHANGEDSINCE` narrows the fetch, but the server never volunteers + /// removals, so absence is found by diffing the full UID set. + /// **Gmail lives here**: it advertises CONDSTORE and not QRESYNC. + Condstore, + /// Neither extension. Changes and removals both come from a full UID-set + /// enumeration. **Exchange lives here.** + UidSetDiff, + /// No usable checkpoint: fetch everything. Also where an unreadable or + /// UIDVALIDITY-invalidated checkpoint lands. Full, - /// A checkpoint was stored but could not be parsed (malformed, or a legacy - /// bare-UID checkpoint predating UIDVALIDITY tracking). Falls back to a - /// full fetch, surfaced (logged) rather than silently ignored. - CorruptCheckpoint, - /// The stored checkpoint's UIDVALIDITY no longer matches the mailbox's - /// current UIDVALIDITY (the mailbox was renumbered) or the server did not - /// report a current UIDVALIDITY to compare against. Stored UIDs are no - /// longer meaningful, so this falls back to a safe full re-fetch. - UidValidityChanged { - /// UIDVALIDITY recorded in the stored checkpoint. - stored: u32, - /// UIDVALIDITY the server reports now, if any. - current: Option, - }, - /// A valid checkpoint whose UIDVALIDITY still matches: fetch only UIDs at - /// or above the stored boundary, i.e. messages that arrived since. - Incremental(u32), } -impl FetchRange { - /// Render the IMAP `UID FETCH` sequence-set. Every non-incremental variant - /// fetches all UIDs (`1:*`); an incremental checkpoint `n` fetches `n:*`, - /// which the server bounds to UIDs `>= n` -- a genuinely narrower fetch. - fn sequence_set(self) -> String { +/// What the server and this mailbox can actually do. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub(crate) struct MailboxSyncCapabilities { + /// Server advertises `QRESYNC` **and** this connection enabled it. + pub qresync_enabled: bool, + /// Server advertises `CONDSTORE`. + pub condstore: bool, + /// This mailbox reported a `HIGHESTMODSEQ`. A mailbox answering + /// `NOMODSEQ` has no mod-sequences to compare against no matter what the + /// server advertises. + pub mailbox_has_modseq: bool, +} + +/// One folder-sync pass, resolved before any body is fetched. +#[derive(Debug, Clone)] +pub(crate) struct FolderSyncPlan { + /// The mechanism this pass negotiated. + pub rung: SyncRung, + pub uid_validity: Option, + pub uid_next: Option, + pub highest_modseq: Option, + /// UID set whose bodies this pass fetches. Empty means nothing changed. + pub body_sequence_set: String, + /// UIDs the server reported as gone (QRESYNC only). + pub vanished_uids: Vec, + /// The folder's complete UID set, when this pass enumerated it. + pub present_uids: Option>, +} + +/// A checkpoint decoded into the resume point it names. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum Checkpoint { + /// `v1:uidnext` -- resume from a UID boundary. + UidNext { uid_validity: u32, uid_next: u32 }, + /// `v2:modseq` -- resume from a mod-sequence. + ModSeq { uid_validity: u32, modseq: u64 }, +} + +impl Checkpoint { + fn uid_validity(self) -> u32 { match self { - FetchRange::Full - | FetchRange::CorruptCheckpoint - | FetchRange::UidValidityChanged { .. } => "1:*".to_string(), - FetchRange::Incremental(uid) => format!("{}:*", uid), + Checkpoint::UidNext { uid_validity, .. } | Checkpoint::ModSeq { uid_validity, .. } => { + uid_validity + } } } } -/// Resolve the fetch range from a stored `since_state` checkpoint and the -/// mailbox's CURRENT UIDVALIDITY (as just reported by `SELECT`). `None` means -/// no prior sync (full fetch). A checkpoint in the `"{uidvalidity}:{uidnext}"` -/// format proceeds incrementally only when its UIDVALIDITY still matches the -/// server's current one; a mismatch (mailbox renumbered) or a missing current -/// UIDVALIDITY forces a safe full re-fetch. A malformed or legacy bare-number -/// checkpoint is treated as corrupt (full fetch, flagged), never misread as a -/// live UID boundary. -fn resolve_fetch_range(since_state: Option<&str>, current_uid_validity: Option) -> FetchRange { - let Some(raw) = since_state else { - return FetchRange::Full; - }; - let parsed = raw - .split_once(CHECKPOINT_DELIM) - .and_then(|(validity, uid)| { - Some(( - validity.trim().parse::().ok()?, - uid.trim().parse::().ok()?, - )) +/// Decode a stored checkpoint, or `None` when it carries no scheme this build +/// recognises. An unrecognised token is never guessed at -- see +/// [`CHECKPOINT_SCHEME_UIDNEXT`] for why that would lose mail silently. +pub(crate) fn parse_checkpoint(raw: &str) -> Option { + let raw = raw.trim(); + if let Some(body) = raw + .strip_prefix(CHECKPOINT_SCHEME_MODSEQ) + .and_then(|r| r.strip_prefix(CHECKPOINT_DELIM)) + { + let (validity, modseq) = body.split_once(CHECKPOINT_DELIM)?; + return Some(Checkpoint::ModSeq { + uid_validity: validity.trim().parse().ok()?, + modseq: modseq.trim().parse().ok()?, }); - let Some((stored_validity, uid)) = parsed else { - return FetchRange::CorruptCheckpoint; - }; - if uid < 1 { - return FetchRange::CorruptCheckpoint; } - match current_uid_validity { - Some(current) if current == stored_validity => FetchRange::Incremental(uid), - current => FetchRange::UidValidityChanged { - stored: stored_validity, - current, - }, + let body = raw + .strip_prefix(CHECKPOINT_SCHEME_UIDNEXT) + .and_then(|r| r.strip_prefix(CHECKPOINT_DELIM))?; + let (validity, uid) = body.split_once(CHECKPOINT_DELIM)?; + let uid_next: u32 = uid.trim().parse().ok()?; + if uid_next < 1 { + return None; + } + Some(Checkpoint::UidNext { + uid_validity: validity.trim().parse().ok()?, + uid_next, + }) +} + +/// Pick the rung for this pass. +/// +/// A checkpoint whose UIDVALIDITY no longer matches the mailbox is not a +/// downgrade to a lesser rung -- it is [`SyncRung::Full`]. UIDs restart after a +/// UIDVALIDITY change, so every stored boundary refers to a different message +/// than it did (RFC 3501 section 2.3.1.1). +pub(crate) fn choose_rung( + checkpoint: Option, + current_uid_validity: Option, + caps: MailboxSyncCapabilities, +) -> SyncRung { + let modseq_usable = caps.mailbox_has_modseq && (caps.qresync_enabled || caps.condstore); + + let Some(checkpoint) = checkpoint else { + // No checkpoint means everything must be fetched regardless of what + // the server can do. + return SyncRung::Full; + }; + let (Some(current), stored) = (current_uid_validity, checkpoint.uid_validity()) else { + return SyncRung::Full; + }; + if current != stored { + return SyncRung::Full; + } + + match checkpoint { + Checkpoint::ModSeq { .. } if modseq_usable && caps.qresync_enabled => SyncRung::Qresync, + Checkpoint::ModSeq { .. } if modseq_usable => SyncRung::Condstore, + // A mod-sequence checkpoint against a mailbox that can no longer + // answer mod-sequence queries (index loss, or a move to a server + // without the extension) has nothing to resume from. + Checkpoint::ModSeq { .. } => SyncRung::Full, + Checkpoint::UidNext { .. } => SyncRung::UidSetDiff, } } @@ -688,7 +771,7 @@ impl ImapEngine { folder_id: &str, since_state: Option<&str>, session: &mut async_imap::Session, - ) -> Result<(Vec, String), MailError> + ) -> Result where S: AsyncRead + AsyncWrite + Unpin + Send + Debug + 'static, { @@ -727,35 +810,18 @@ impl ImapEngine { since_state: Option<&str>, session: &mut async_imap::Session, batch_size: usize, - ) -> Result<(Vec, String), MailError> + ) -> Result where S: AsyncRead + AsyncWrite + Unpin + Send + Debug + 'static, { - let mailbox = session.select(folder_id).await.map_err(|e| { - MailError::ImapError(format!("failed to select folder '{}': {}", folder_id, e)) - })?; - let server_uid_next = mailbox.uid_next; - let server_uid_validity = mailbox.uid_validity; - - let range = resolve_fetch_range(since_state, server_uid_validity); - match range { - FetchRange::CorruptCheckpoint => tracing::warn!( - folder_id, - checkpoint = since_state.unwrap_or_default(), - "ignoring unparseable folder sync checkpoint; falling back to a full fetch" - ), - FetchRange::UidValidityChanged { stored, current } => tracing::warn!( - folder_id, - stored_uidvalidity = stored, - current_uidvalidity = current, - "mailbox UIDVALIDITY changed (renumbered); the stored UID checkpoint is no \ - longer valid, so re-fetching in full to avoid silently skipping messages" - ), - FetchRange::Full | FetchRange::Incremental(_) => {} - } + let plan = self + .plan_folder_sync(folder_id, since_state, session) + .await?; + let server_uid_next = plan.uid_next; + let server_uid_validity = plan.uid_validity; let batch_size = batch_size.max(1); - let sequence_set = range.sequence_set(); + let sequence_set = plan.body_sequence_set.clone(); // Phase 1: enumerate UIDs and sizes only -- no bodies transferred, so // this stays small even for an enormous mailbox. @@ -857,23 +923,239 @@ impl ImapEngine { // safely falls back to a full fetch. let boundary_uid = server_uid_next .or_else(|| (max_uid_seen > 0).then_some(max_uid_seen.saturating_add(1))); - let new_checkpoint = match (server_uid_validity, boundary_uid) { - (Some(validity), Some(uid)) => format!("{}{}{}", validity, CHECKPOINT_DELIM, uid), + // A mod-sequence checkpoint is preferred whenever the mailbox can + // supply one, even after a full pass: writing it is what lets the NEXT + // sync use a higher rung than this one did. + let new_checkpoint = match (server_uid_validity, plan.highest_modseq, boundary_uid) { + (Some(validity), Some(modseq), _) => format!( + "{}{}{}{}{}", + CHECKPOINT_SCHEME_MODSEQ, CHECKPOINT_DELIM, validity, CHECKPOINT_DELIM, modseq + ), + (Some(validity), None, Some(uid)) => format!( + "{}{}{}{}{}", + CHECKPOINT_SCHEME_UIDNEXT, CHECKPOINT_DELIM, validity, CHECKPOINT_DELIM, uid + ), _ => since_state .map(str::to_string) .unwrap_or_else(|| "full-resync-required".to_string()), }; + let removals: Vec = plan + .vanished_uids + .iter() + .map(|uid| self.surrogate_for(folder_id, server_uid_validity, *uid)) + .collect(); + let present = plan.present_uids.as_ref().map(|uids| { + uids.iter() + .map(|uid| self.surrogate_for(folder_id, server_uid_validity, *uid)) + .collect() + }); + tracing::info!( account_id = %self.account_id, folder_id, + rung = ?plan.rung, fetched = emails.len(), + removed = removals.len(), + enumerated = plan.present_uids.as_ref().map_or(0, Vec::len), oversized = oversized_uids.len(), skipped = skipped_no_uid, "imap folder sync complete" ); - Ok((emails, new_checkpoint)) + Ok(crate::backend::FolderChanges { + upserts: emails, + removals, + present, + next_state: new_checkpoint, + }) + } + + /// Negotiate capabilities, select the folder by the best available means, + /// and decide what this pass must fetch and what it can say about absence. + /// + /// This is where the ladder is applied. Ordering matters: `ENABLE` must + /// precede `SELECT` (RFC 7162 section 3.2.3 makes a server answer `BAD` to + /// a QRESYNC `SELECT` otherwise), and the rung cannot be finalised until + /// the mailbox has reported whether it has mod-sequences at all. + async fn plan_folder_sync( + &self, + folder_id: &str, + since_state: Option<&str>, + session: &mut async_imap::Session, + ) -> Result + where + S: AsyncRead + AsyncWrite + Unpin + Send + Debug + 'static, + { + let (has_qresync, has_condstore) = { + let caps = session + .capabilities() + .await + .map_err(|e| MailError::ImapError(format!("CAPABILITY query failed: {e}")))?; + (caps.has_str("QRESYNC"), caps.has_str("CONDSTORE")) + }; + + let checkpoint = since_state.and_then(parse_checkpoint); + if since_state.is_some() && checkpoint.is_none() { + tracing::warn!( + folder_id, + checkpoint = since_state.unwrap_or_default(), + "folder sync checkpoint carries no scheme this build recognises; \ + re-fetching in full rather than guessing at its meaning" + ); + } + + // ENABLE before SELECT, and only when there is a mod-sequence + // checkpoint that a QRESYNC SELECT could actually resume from. + let qresync_enabled = match (has_qresync, checkpoint) { + (true, Some(Checkpoint::ModSeq { .. })) => { + crate::imap_raw::enable(session, &["QRESYNC"]).await? + } + _ => false, + }; + + // Try the QRESYNC SELECT when it is available. Per RFC 7162 section + // 3.2.5 a server whose UIDVALIDITY no longer matches simply ignores the + // QRESYNC parameters, so this is safe to attempt: the mismatch is + // detected below from the UIDVALIDITY it reports back. + let (selected, attempted_qresync) = match (qresync_enabled, checkpoint) { + ( + true, + Some(Checkpoint::ModSeq { + uid_validity, + modseq, + }), + ) => ( + crate::imap_raw::select_qresync(session, folder_id, uid_validity, modseq, None) + .await?, + true, + ), + _ => { + let exchange = + crate::imap_raw::run_collected(session, &format!("SELECT {folder_id}")).await?; + if !exchange.ok { + return Err(MailError::ImapError(format!( + "failed to select folder '{folder_id}': {}", + exchange.information.as_deref().unwrap_or("no detail") + ))); + } + (exchange, false) + } + }; + + let caps = MailboxSyncCapabilities { + qresync_enabled, + condstore: has_condstore, + mailbox_has_modseq: selected.highest_modseq.is_some(), + }; + let mut rung = choose_rung(checkpoint, selected.uid_validity, caps); + // The QRESYNC SELECT was issued optimistically; if the server answered + // with a different UIDVALIDITY it ignored our parameters, and anything + // it did report refers to a UID space that no longer exists. + if attempted_qresync && rung != SyncRung::Qresync { + tracing::warn!( + folder_id, + "QRESYNC resume was refused or the mailbox was renumbered; re-fetching in full" + ); + } + if rung == SyncRung::Qresync && !attempted_qresync { + rung = SyncRung::Full; + } + + let mut plan = FolderSyncPlan { + rung, + uid_validity: selected.uid_validity, + uid_next: selected.uid_next, + highest_modseq: selected.highest_modseq, + body_sequence_set: "1:*".to_string(), + vanished_uids: Vec::new(), + present_uids: None, + }; + + match rung { + SyncRung::Qresync => { + // The server volunteered both halves: what changed, and what + // disappeared. No enumeration of the folder is needed at all, + // which is the entire reason this rung is worth having. + plan.vanished_uids = selected.vanished_uids.clone(); + plan.body_sequence_set = if selected.fetched_uids.is_empty() { + // Nothing changed. An empty set would be a protocol error, + // so ask for a range that cannot match instead of skipping + // the fetch path. + String::new() + } else { + join_uid_batch(&selected.fetched_uids) + }; + } + SyncRung::Condstore => { + let modseq = match checkpoint { + Some(Checkpoint::ModSeq { modseq, .. }) => modseq, + _ => 0, + }; + let changed = self + .enumerate_uids(session, folder_id, Some(modseq)) + .await?; + plan.body_sequence_set = if changed.is_empty() { + String::new() + } else { + join_uid_batch(&changed) + }; + // CONDSTORE narrows the fetch but never reports removals, so + // absence has to be found by enumerating what remains. + plan.present_uids = Some(self.enumerate_uids(session, folder_id, None).await?); + } + SyncRung::UidSetDiff => { + // New arrivals come from the UID boundary; removals from the + // full UID set. Two cheap enumerations beat re-fetching bodies. + if let Some(Checkpoint::UidNext { uid_next, .. }) = checkpoint { + plan.body_sequence_set = format!("{uid_next}:*"); + } + plan.present_uids = Some(self.enumerate_uids(session, folder_id, None).await?); + } + SyncRung::Full => { + // Everything is fetched, so the same pass also establishes + // exactly what the folder holds. + plan.present_uids = Some(self.enumerate_uids(session, folder_id, None).await?); + } + } + + Ok(plan) + } + + /// List the folder's UIDs, optionally narrowed to those changed since + /// `changed_since`. Requests no bodies and no flags, so the cost is one + /// short line per message rather than a download. + async fn enumerate_uids( + &self, + session: &mut async_imap::Session, + folder_id: &str, + changed_since: Option, + ) -> Result, MailError> + where + S: AsyncRead + AsyncWrite + Unpin + Send + Debug + 'static, + { + let command = match changed_since { + Some(modseq) => format!("UID FETCH 1:* (UID) (CHANGEDSINCE {modseq})"), + None => "UID FETCH 1:* (UID)".to_string(), + }; + let exchange = crate::imap_raw::run_collected(session, &command).await?; + if !exchange.ok { + return Err(MailError::ImapError(format!( + "UID enumeration of '{folder_id}' failed: {}", + exchange.information.as_deref().unwrap_or("no detail") + ))); + } + let mut uids = exchange.fetched_uids; + uids.sort_unstable(); + uids.dedup(); + Ok(uids) + } + + /// Surrogate id for a UID in this account/folder/UIDVALIDITY scope, so + /// removals name messages the same way stored rows do. + fn surrogate_for(&self, folder_id: &str, uid_validity: Option, uid: u32) -> String { + let validity = uid_validity.map(|v| v.to_string()).unwrap_or_default(); + Email::surrogate_id(&self.account_id, folder_id, &validity, &uid.to_string()) } /// Build an [`Email`] from a full-body (or envelope-only) FETCH item. When @@ -1015,7 +1297,7 @@ impl ImapEngine { &self, folder_id: &str, since_state: Option<&str>, - ) -> Result<(Vec, String), MailError> { + ) -> Result { let (username, password) = match (&self.username, &self.password) { (Some(u), Some(p)) => (u.as_str(), p.as_str()), _ => { @@ -1423,11 +1705,11 @@ impl MailBackend for ImapEngine { Ok(folders) } - async fn sync_messages( + async fn sync_changes( &self, folder_id: &str, since_state: Option<&str>, - ) -> Result<(Vec, String), MailError> { + ) -> Result { self.sync_folder_messages(folder_id, since_state).await } @@ -1492,7 +1774,16 @@ mod tests { while let Some(line) = read_scripted_line(&mut socket).await { let tag = line.split_whitespace().next().unwrap_or("").to_string(); let upper = line.to_ascii_uppercase(); - if upper.contains("LOGIN") { + if upper.contains("CAPABILITY") { + let _ = socket + .write_all( + format!( + "* CAPABILITY IMAP4rev1 UIDPLUS MOVE\r\n{tag} OK CAPABILITY\r\n" + ) + .as_bytes(), + ) + .await; + } else if upper.contains("LOGIN") { let _ = socket .write_all(format!("{tag} OK Logged in\r\n").as_bytes()) .await; @@ -1526,76 +1817,8 @@ mod tests { assert_eq!(folders[1].id, "Archive"); } - #[test] - fn resolve_fetch_range_distinguishes_none_valid_corrupt_and_renumbered_checkpoints() { - // First-ever sync: full history (current UIDVALIDITY irrelevant). - assert_eq!(resolve_fetch_range(None, Some(1)), FetchRange::Full); - assert_eq!(resolve_fetch_range(None, None).sequence_set(), "1:*"); - - // Valid checkpoint whose UIDVALIDITY still matches: narrowed fetch. - assert_eq!( - resolve_fetch_range(Some("1:105"), Some(1)), - FetchRange::Incremental(105) - ); - assert_eq!( - resolve_fetch_range(Some("1:105"), Some(1)).sequence_set(), - "105:*" - ); - assert_eq!( - resolve_fetch_range(Some(" 7 : 42 "), Some(7)), - FetchRange::Incremental(42) - ); - - // UIDVALIDITY changed (mailbox renumbered) or the server reported none - // to compare against: MUST fall back to a full fetch, distinctly, so a - // renumber can never silently skip the low new UIDs. - assert_eq!( - resolve_fetch_range(Some("1:105"), Some(2)), - FetchRange::UidValidityChanged { - stored: 1, - current: Some(2) - } - ); - assert_eq!( - resolve_fetch_range(Some("1:105"), Some(2)).sequence_set(), - "1:*" - ); - assert_eq!( - resolve_fetch_range(Some("1:105"), None), - FetchRange::UidValidityChanged { - stored: 1, - current: None - } - ); - - // Malformed or legacy bare-number (pre-UIDVALIDITY) checkpoints are - // corrupt -> full fetch, never misread as a live UID boundary. - assert_eq!( - resolve_fetch_range(Some(""), Some(1)), - FetchRange::CorruptCheckpoint - ); - assert_eq!( - resolve_fetch_range(Some("105"), Some(1)), - FetchRange::CorruptCheckpoint, - "a legacy bare-number checkpoint must not be read as a UID boundary" - ); - assert_eq!( - resolve_fetch_range(Some("1:0"), Some(1)), - FetchRange::CorruptCheckpoint - ); - assert_eq!( - resolve_fetch_range(Some("not:auid"), Some(1)), - FetchRange::CorruptCheckpoint - ); - assert_eq!( - resolve_fetch_range(Some("not-a-uid"), Some(1)).sequence_set(), - "1:*" - ); - } - /// Read a single CRLF-terminated line from a scripted-server stream, - /// returning `None` at EOF (client hung up). Used only by the in-memory - /// duplex tests below; the real fetch path never parses raw lines. + /// returning `None` at EOF (client hung up). async fn read_scripted_line(reader: &mut R) -> Option { let mut buf = Vec::new(); let mut byte = [0u8; 1]; @@ -1621,6 +1844,159 @@ mod tests { } } + /// The body/size fetches from a captured command list, excluding the + /// `(UID)`-only enumeration a non-QRESYNC rung issues to find removals. + /// That pass is expected and says nothing about how the body fetch was + /// narrowed, which is what these tests are about. + fn body_fetch_ranges(all: &[String]) -> Vec { + all.iter() + .filter(|c| !c.to_ascii_uppercase().ends_with("(UID)")) + .cloned() + .collect() + } + + /// Capabilities that can reach every rung, so a test only has to vary the + /// one thing it is about. + fn full_caps() -> MailboxSyncCapabilities { + MailboxSyncCapabilities { + qresync_enabled: true, + condstore: true, + mailbox_has_modseq: true, + } + } + + #[test] + fn checkpoints_round_trip_through_their_scheme() { + assert_eq!( + parse_checkpoint("v2:modseq:42:900"), + Some(Checkpoint::ModSeq { + uid_validity: 42, + modseq: 900 + }) + ); + assert_eq!( + parse_checkpoint("v1:uidnext:42:105"), + Some(Checkpoint::UidNext { + uid_validity: 42, + uid_next: 105 + }) + ); + assert_eq!( + parse_checkpoint("v1:uidnext: 7 : 42 "), + Some(Checkpoint::UidNext { + uid_validity: 7, + uid_next: 42 + }) + ); + } + + #[test] + fn an_untagged_or_unknown_scheme_checkpoint_is_not_interpreted() { + // A QRESYNC checkpoint is (uidvalidity, highestmodseq) -- the same + // shape as (uidvalidity, uidnext) and a different meaning. Reading one + // as the other hands the server a UIDNEXT where it expects a MODSEQ; + // UIDNEXT is normally the larger number, so the server answers + // "nothing changed" and everything below is never enumerated again. + for token in [ + "42:105", // pre-tagging + "v3:future:42:105", // a scheme from a later build + "v1:modseq:42:105", // tag-shaped but not ours + "v1:42:105", // truncated tag + "v1:uidnext:42:0", // a UID boundary below 1 is not addressable + "", + ] { + assert_eq!( + parse_checkpoint(token), + None, + "token {token:?} must not be interpreted" + ); + assert_eq!( + choose_rung(parse_checkpoint(token), Some(42), full_caps()), + SyncRung::Full, + "an uninterpretable token must re-fetch everything" + ); + } + } + + #[test] + fn the_rung_is_chosen_from_the_checkpoint_and_what_the_mailbox_supports() { + let modseq = parse_checkpoint("v2:modseq:42:900"); + let uidnext = parse_checkpoint("v1:uidnext:42:105"); + + // Fastmail/Cyrus, Dovecot: the server states removals directly. + assert_eq!( + choose_rung(modseq, Some(42), full_caps()), + SyncRung::Qresync + ); + + // Gmail: CONDSTORE without QRESYNC. This is the rung a two-rung ladder + // would have skipped, pushing the largest provider onto full sweeps. + assert_eq!( + choose_rung( + modseq, + Some(42), + MailboxSyncCapabilities { + qresync_enabled: false, + condstore: true, + mailbox_has_modseq: true + } + ), + SyncRung::Condstore + ); + + // Exchange: neither extension, so a UID-set diff is all that is left. + assert_eq!( + choose_rung( + uidnext, + Some(42), + MailboxSyncCapabilities { + qresync_enabled: false, + condstore: false, + mailbox_has_modseq: false + } + ), + SyncRung::UidSetDiff + ); + + // NOMODSEQ is a MAILBOX property (RFC 7162 3.1.2.2): a mod-sequence + // checkpoint against a mailbox that cannot answer mod-sequence queries + // has nothing to resume from, whatever the server advertises. + assert_eq!( + choose_rung( + modseq, + Some(42), + MailboxSyncCapabilities { + qresync_enabled: true, + condstore: true, + mailbox_has_modseq: false + } + ), + SyncRung::Full + ); + + // No checkpoint at all. + assert_eq!(choose_rung(None, Some(42), full_caps()), SyncRung::Full); + } + + #[test] + fn a_uidvalidity_change_forces_a_full_pass_from_every_rung() { + // UIDs restart after a UIDVALIDITY change (RFC 3501 2.3.1.1), so every + // stored boundary now names a different message. This is not a + // downgrade to a lesser rung; nothing stored can be resumed from. + for token in ["v2:modseq:42:900", "v1:uidnext:42:105"] { + assert_eq!( + choose_rung(parse_checkpoint(token), Some(43), full_caps()), + SyncRung::Full, + "{token} must not survive a renumbering" + ); + assert_eq!( + choose_rung(parse_checkpoint(token), None, full_caps()), + SyncRung::Full, + "{token}: a server that reports no UIDVALIDITY cannot be resumed against" + ); + } + } + #[tokio::test] async fn incremental_sync_narrows_the_fetch_range_on_the_second_pass() { // Drives a real `async_imap::Session` over an in-memory duplex pair @@ -1643,6 +2019,17 @@ mod tests { let _ = io .write_all(format!("{tag} OK FETCH completed\r\n").as_bytes()) .await; + } else if upper.contains("CAPABILITY") { + // The planner negotiates capabilities before selecting; a + // script that never answers this blocks the client forever. + let _ = io + .write_all( + format!( + "* CAPABILITY IMAP4rev1 UIDPLUS MOVE\r\n{tag} OK CAPABILITY\r\n" + ) + .as_bytes(), + ) + .await; } else if upper.contains("LOGIN") { let _ = io .write_all(format!("{tag} OK Logged in\r\n").as_bytes()) @@ -1674,26 +2061,33 @@ mod tests { // First sync: no checkpoint -> full history; returns // "{uidvalidity}:{uidnext}" from real server state. - let (first, checkpoint) = engine + let first_pass = engine .sync_folder_messages_with_session("INBOX", None, &mut session) .await .expect("first sync succeeds"); + let (first, checkpoint) = (first_pass.upserts, first_pass.next_state); assert!(first.is_empty()); - assert_eq!(checkpoint, "1:105"); + assert_eq!(checkpoint, "v1:uidnext:1:105"); // Second sync: feed back the checkpoint; UIDVALIDITY still matches // (1), so the fetch is narrowed. - let (_second, checkpoint2) = engine + let __changes = engine .sync_folder_messages_with_session("INBOX", Some(&checkpoint), &mut session) .await .expect("second sync succeeds"); - assert_eq!(checkpoint2, "1:105"); + let (_second, checkpoint2) = (__changes.upserts, __changes.next_state); + assert_eq!(checkpoint2, "v1:uidnext:1:105"); drop(session); let _ = server.await; - let ranges = fetch_ranges.lock().expect("lock captured ranges"); - assert_eq!(ranges.len(), 2, "expected exactly two UID FETCH commands"); + let all = fetch_ranges.lock().expect("lock captured ranges"); + let ranges = body_fetch_ranges(&all); + assert_eq!( + ranges.len(), + 2, + "expected one body fetch per sync, got: {all:?}" + ); assert!( ranges[0].contains("1:*"), "first sync must fetch full history, got: {}", @@ -1808,6 +2202,17 @@ mod tests { } resp.push_str(&format!("{tag} OK FETCH completed\r\n")); let _ = io.write_all(resp.as_bytes()).await; + } else if upper.contains("CAPABILITY") { + // The planner negotiates capabilities before selecting; a + // script that never answers this blocks the client forever. + let _ = io + .write_all( + format!( + "* CAPABILITY IMAP4rev1 UIDPLUS MOVE\r\n{tag} OK CAPABILITY\r\n" + ) + .as_bytes(), + ) + .await; } else if upper.contains("LOGIN") { let _ = io .write_all(format!("{tag} OK Logged in\r\n").as_bytes()) @@ -1835,10 +2240,11 @@ mod tests { .map_err(|(e, _)| e) .expect("scripted login succeeds"); let engine = ImapEngine::new("acct-1", "example.test", 993); - let (emails, checkpoint) = engine + let changes = engine .sync_folder_messages_batched("INBOX", since_state.as_deref(), &mut session, batch_size) .await .expect("batched sync succeeds"); + let (emails, checkpoint) = (changes.upserts, changes.next_state); drop(session); let _ = server.await; @@ -1868,7 +2274,7 @@ mod tests { let (emails, checkpoint, cmds) = run_batched_sync(messages, 1, 6, None, 2).await; assert_eq!(emails.len(), 5, "every message must be synced"); - assert_eq!(checkpoint, "1:6"); + assert_eq!(checkpoint, "v1:uidnext:1:6"); let body_cmds = body_fetch_commands(&cmds); assert_eq!( @@ -1957,7 +2363,7 @@ mod tests { let (first, checkpoint, first_cmds) = run_batched_sync(messages.clone(), 1, 4, None, 2).await; assert_eq!(first.len(), 3); - assert_eq!(checkpoint, "1:4"); + assert_eq!(checkpoint, "v1:uidnext:1:4"); assert!( first_cmds.iter().any(|c| c.contains("1:*")), "first sync enumerates the full range, got: {first_cmds:?}" @@ -1971,16 +2377,26 @@ mod tests { "no new mail arrived, so an incremental sync returns nothing" ); assert_eq!( - checkpoint2, "1:4", + checkpoint2, "v1:uidnext:1:4", "checkpoint stays stable across an empty pass" ); + let second_bodies = body_fetch_ranges(&second_cmds); assert!( - second_cmds.iter().any(|c| c.contains("4:*")), - "second sync must enumerate the narrowed range, got: {second_cmds:?}" + second_bodies.iter().any(|c| c.contains("4:*")), + "second sync must fetch only the narrowed range, got: {second_bodies:?}" ); assert!( - !second_cmds.iter().any(|c| c.contains("1:*")), - "second sync must NOT re-enumerate the full range, got: {second_cmds:?}" + !second_bodies.iter().any(|c| c.contains("1:*")), + "second sync must NOT re-fetch bodies for the full range, got: {second_bodies:?}" + ); + // The `1:*` that IS expected: a UID-only enumeration. Without QRESYNC + // there is no other way to learn a message was removed, and it costs + // one short line per message rather than a body. + assert!( + second_cmds + .iter() + .any(|c| c.to_ascii_uppercase().ends_with("(UID)") && c.contains("1:*")), + "a non-QRESYNC rung must enumerate the folder to detect removals, got: {second_cmds:?}" ); } @@ -2008,6 +2424,17 @@ mod tests { let _ = io .write_all(format!("{tag} OK FETCH completed\r\n").as_bytes()) .await; + } else if upper.contains("CAPABILITY") { + // The planner negotiates capabilities before selecting; a + // script that never answers this blocks the client forever. + let _ = io + .write_all( + format!( + "* CAPABILITY IMAP4rev1 UIDPLUS MOVE\r\n{tag} OK CAPABILITY\r\n" + ) + .as_bytes(), + ) + .await; } else if upper.contains("LOGIN") { let _ = io .write_all(format!("{tag} OK Logged in\r\n").as_bytes()) @@ -2042,29 +2469,36 @@ mod tests { let engine = ImapEngine::new("acct-1", "example.test", 993); // First sync under UIDVALIDITY 1 stores "1:105". - let (_first, checkpoint) = engine + let __changes = engine .sync_folder_messages_with_session("INBOX", None, &mut session) .await .expect("first sync succeeds"); - assert_eq!(checkpoint, "1:105"); + let (_first, checkpoint) = (__changes.upserts, __changes.next_state); + assert_eq!(checkpoint, "v1:uidnext:1:105"); // Second sync: server now reports UIDVALIDITY 2. The stored checkpoint // is stale, so the fetch MUST be a full `1:*`, and the checkpoint is // rewritten under the new UIDVALIDITY. - let (_second, checkpoint2) = engine + let __changes = engine .sync_folder_messages_with_session("INBOX", Some(&checkpoint), &mut session) .await .expect("second sync succeeds"); + let (_second, checkpoint2) = (__changes.upserts, __changes.next_state); assert_eq!( - checkpoint2, "2:105", + checkpoint2, "v1:uidnext:2:105", "checkpoint must be rewritten under the new UIDVALIDITY" ); drop(session); let _ = server.await; - let ranges = fetch_ranges.lock().expect("lock captured ranges"); - assert_eq!(ranges.len(), 2, "expected exactly two UID FETCH commands"); + let all = fetch_ranges.lock().expect("lock captured ranges"); + let ranges = body_fetch_ranges(&all); + assert_eq!( + ranges.len(), + 2, + "expected one body fetch per sync, got: {all:?}" + ); assert!( ranges[0].contains("1:*"), "first sync fetches full history, got: {}", @@ -2093,6 +2527,17 @@ mod tests { if upper.contains("UID FETCH") { // Deliberately never respond: a mid-stream server stall. std::future::pending::<()>().await; + } else if upper.contains("CAPABILITY") { + // The planner negotiates capabilities before selecting; a + // script that never answers this blocks the client forever. + let _ = io + .write_all( + format!( + "* CAPABILITY IMAP4rev1 UIDPLUS MOVE\r\n{tag} OK CAPABILITY\r\n" + ) + .as_bytes(), + ) + .await; } else if upper.contains("LOGIN") { let _ = io .write_all(format!("{tag} OK Logged in\r\n").as_bytes()) @@ -2632,7 +3077,16 @@ mod tests { while let Some(line) = read_scripted_line(&mut io).await { let tag = line.split_whitespace().next().unwrap_or("").to_string(); let upper = line.to_ascii_uppercase(); - if upper.contains("LOGIN") { + if upper.contains("CAPABILITY") { + let _ = io + .write_all( + format!( + "* CAPABILITY IMAP4rev1 UIDPLUS MOVE\r\n{tag} OK CAPABILITY\r\n" + ) + .as_bytes(), + ) + .await; + } else if upper.contains("LOGIN") { let _ = io .write_all(format!("{tag} OK Logged in\r\n").as_bytes()) .await; @@ -2664,10 +3118,11 @@ mod tests { .expect("scripted login succeeds"); let engine = ImapEngine::new("acct-completion", "example.test", 993); - let (emails, _checkpoint) = engine + let __changes = engine .sync_folder_messages_with_session("INBOX", None, &mut session) .await .expect("sync succeeds"); + let (emails, _checkpoint) = (__changes.upserts, __changes.next_state); assert!(emails.is_empty()); drop(session); diff --git a/crates/nuncio-mail/src/imap_raw.rs b/crates/nuncio-mail/src/imap_raw.rs index be2b321..cee5f64 100644 --- a/crates/nuncio-mail/src/imap_raw.rs +++ b/crates/nuncio-mail/src/imap_raw.rs @@ -42,6 +42,14 @@ use tokio::io::{AsyncRead, AsyncWrite}; use crate::parser::MailError; +/// Bounded wait for any single response line. +/// +/// A server that accepts a command and then stops answering must surface as a +/// stall, not as a sync that never returns. Mirrors the per-item FETCH timeout +/// the typed path already applies -- an unbounded read here would reintroduce +/// exactly the hang that bounded it. +const RESPONSE_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(60); + /// A `COPYUID` response code (RFC 4315 section 3): the destination mailbox's /// UIDVALIDITY, the source UIDs, and the UIDs they became. /// @@ -84,6 +92,8 @@ pub struct RawExchange { pub highest_modseq: Option, /// UIDVALIDITY, from an untagged `OK` during `SELECT`. pub uid_validity: Option, + /// UIDNEXT, from an untagged `OK` during `SELECT`. + pub uid_next: Option, } impl RawExchange { @@ -170,9 +180,13 @@ where let mut exchange = RawExchange::default(); loop { - let response = session - .read_response() + let response = tokio::time::timeout(RESPONSE_READ_TIMEOUT, session.read_response()) .await + .map_err(|_| { + MailError::FetchStalled(format!( + "no response to '{command}' within {RESPONSE_READ_TIMEOUT:?}" + )) + })? .map_err(|e| MailError::ImapError(format!("reading response to '{command}': {e}")))? .ok_or_else(|| { MailError::ImapError(format!("connection closed before '{command}' completed")) @@ -232,6 +246,7 @@ fn absorb_code(exchange: &mut RawExchange, code: Option<&ResponseCode<'_>>) { } Some(ResponseCode::HighestModSeq(modseq)) => exchange.highest_modseq = Some(*modseq), Some(ResponseCode::UidValidity(validity)) => exchange.uid_validity = Some(*validity), + Some(ResponseCode::UidNext(uid_next)) => exchange.uid_next = Some(*uid_next), _ => {} } } diff --git a/crates/nuncio-mail/src/jmap.rs b/crates/nuncio-mail/src/jmap.rs index 80319b7..7d63652 100644 --- a/crates/nuncio-mail/src/jmap.rs +++ b/crates/nuncio-mail/src/jmap.rs @@ -561,11 +561,11 @@ impl MailBackend for JmapEngine { fields(account_id = %self.account_id, folder_id = %folder_id), err )] - async fn sync_messages( + async fn sync_changes( &self, folder_id: &str, _since_state: Option<&str>, - ) -> Result<(Vec, String), MailError> { + ) -> Result { if !self.has_credentials() { return Err(MailError::AuthError( "JMAP message sync requires credentials".to_string(), @@ -589,7 +589,16 @@ impl MailBackend for JmapEngine { "jmap message sync complete" ); - Ok((emails, state)) + Ok(crate::backend::FolderChanges { + upserts: emails, + removals: Vec::new(), + // `Email/query` was filtered to one mailbox, but this engine does + // not yet use `Email/changes`, so it cannot claim to have seen the + // complete set. Reporting `None` keeps the caller from deleting + // anything on the strength of a partial view. + present: None, + next_state: state, + }) } #[tracing::instrument(skip(self), fields(account_id = %self.account_id), err)] diff --git a/crates/nuncio-mail/src/mock.rs b/crates/nuncio-mail/src/mock.rs index 487d09a..adbf97b 100644 --- a/crates/nuncio-mail/src/mock.rs +++ b/crates/nuncio-mail/src/mock.rs @@ -106,11 +106,11 @@ impl MailBackend for MockMailBackend { Ok(folders.clone()) } - async fn sync_messages( + async fn sync_changes( &self, folder_id: &str, since_state: Option<&str>, - ) -> Result<(Vec, String), MailError> { + ) -> Result { // Record the checkpoint arg before any early return, so even a failing // call is visible to a test asserting on how the caller resumes. if let Ok(mut calls) = self.since_state_calls.lock() { @@ -136,19 +136,30 @@ impl MailBackend for MockMailBackend { // is nothing new to hand back, so the incremental fetch is genuinely // narrower (empty) rather than a full re-report of every message. if since_state.is_some() { - return Ok((Vec::new(), returned_state)); + return Ok(crate::backend::FolderChanges { + next_state: returned_state, + ..Default::default() + }); } let messages = self .messages .lock() .map_err(|e| MailError::ParseFailed(e.to_string()))?; - let matches = messages + let matches: Vec<_> = messages .iter() .filter(|m| m.folder_id == folder_id) .cloned() .collect(); - Ok((matches, returned_state)) + // A full pass over the mock's store did see everything in the folder, + // so it can honestly report what is present. + let present = matches.iter().map(|m| m.id.clone()).collect(); + Ok(crate::backend::FolderChanges { + upserts: matches, + removals: Vec::new(), + present: Some(present), + next_state: returned_state, + }) } async fn apply_mutation(&self, spec: &RemoteMutationSpec) -> Result<(), MailError> { diff --git a/crates/nuncio-mail/tests/imap_raw_wire_test.rs b/crates/nuncio-mail/tests/imap_raw_wire_test.rs index 8793b3d..9d03e60 100644 --- a/crates/nuncio-mail/tests/imap_raw_wire_test.rs +++ b/crates/nuncio-mail/tests/imap_raw_wire_test.rs @@ -299,3 +299,182 @@ async fn a_qresync_select_without_enable_is_refused() { "the refusal must be surfaced, got: {err}" ); } + +/// Drive a real folder sync against the mock and return what it observed. +async fn sync_folder( + server: &MockImapServer, + since: Option<&str>, +) -> nuncio_mail::backend::FolderChanges { + use nuncio_mail::MailBackend; + let engine = nuncio_mail::ImapEngine::with_credentials( + "acct-1", + server.host(), + server.port(), + nuncio_core::TlsMode::Plain, + "user", + "pass", + ); + engine + .sync_changes("INBOX", since) + .await + .expect("folder sync succeeds") +} + +#[tokio::test] +async fn every_rung_reports_a_message_that_disappeared() { + // The defect this whole change exists for. A forward-only sync can only + // ever add, so a message removed by another client stays local forever. + // Each rung must be able to say it is gone -- QRESYNC because the server + // states it, the other two because the pass enumerated what remains. + for profile in [ + ServerProfile::Qresync, + ServerProfile::Condstore, + ServerProfile::Basic, + ] { + let server = MockImapServer::start(profile).await.expect("server starts"); + server.create_mailbox("INBOX"); + let keep = server.append_message("INBOX", "Subject: keep\r\n\r\nkeep", &[]); + let gone = server.append_message("INBOX", "Subject: gone\r\n\r\ngone", &[]); + + let first = sync_folder(&server, None).await; + assert_eq!(first.upserts.len(), 2, "{profile:?}: first pass sees both"); + + // Another client removes one while we are away. + assert!(server.expunge("INBOX", gone)); + + let second = sync_folder(&server, Some(&first.next_state)).await; + + // However the rung learned it, the caller must end up able to identify + // the removed message and only that one. + let mut reported: Vec = second.removals.clone(); + if let Some(present) = &second.present { + let stored_gone = first + .upserts + .iter() + .map(|e| e.id.clone()) + .filter(|id| !present.contains(id)); + reported.extend(stored_gone); + } + reported.sort(); + reported.dedup(); + + assert_eq!( + reported.len(), + 1, + "{profile:?}: exactly one message must be reported gone, got {reported:?}" + ); + let surviving = first + .upserts + .iter() + .find(|e| e.remote_id == keep.to_string()) + .expect("the kept message was in the first pass"); + assert!( + !reported.contains(&surviving.id), + "{profile:?}: the surviving message must not be reported gone" + ); + } +} + +#[tokio::test] +async fn an_incremental_pass_that_cannot_see_absence_says_so() { + // `present: None` versus `Some(vec![])` is the difference between "I did + // not look" and "the folder is empty". Conflating them deletes a mailbox, + // so QRESYNC -- which never enumerates -- must report None. + let server = MockImapServer::start(ServerProfile::Qresync) + .await + .expect("server starts"); + server.create_mailbox("INBOX"); + server.append_message("INBOX", "Subject: one\r\n\r\nx", &[]); + + let first = sync_folder(&server, None).await; + let second = sync_folder(&server, Some(&first.next_state)).await; + + assert!( + second.next_state.starts_with("v2:modseq:"), + "a QRESYNC-capable mailbox must checkpoint by mod-sequence, got {:?}", + second.next_state + ); + assert_eq!( + second.present, None, + "a QRESYNC pass does not enumerate the folder, so it cannot claim to know what is present" + ); +} + +#[tokio::test] +async fn a_condstore_only_server_still_checkpoints_by_modseq_and_enumerates() { + // Gmail's rung. It cannot be told what vanished, so it must enumerate -- + // and it must still upgrade its checkpoint so the next pass is narrow. + let server = MockImapServer::start(ServerProfile::Condstore) + .await + .expect("server starts"); + server.create_mailbox("INBOX"); + server.append_message("INBOX", "Subject: one\r\n\r\nx", &[]); + + let first = sync_folder(&server, None).await; + assert!( + first.next_state.starts_with("v2:modseq:"), + "CONDSTORE alone is enough to checkpoint by mod-sequence, got {:?}", + first.next_state + ); + + let second = sync_folder(&server, Some(&first.next_state)).await; + assert!( + second.present.is_some(), + "without QRESYNC the only way to find removals is to enumerate" + ); + assert!( + second.upserts.is_empty(), + "nothing changed, so CHANGEDSINCE must fetch no bodies" + ); +} + +#[tokio::test] +async fn a_server_without_modseq_falls_back_to_a_uid_boundary_checkpoint() { + // Exchange's rung: no CONDSTORE, no QRESYNC. + let server = MockImapServer::start(ServerProfile::Basic) + .await + .expect("server starts"); + server.create_mailbox("INBOX"); + server.append_message("INBOX", "Subject: one\r\n\r\nx", &[]); + + let first = sync_folder(&server, None).await; + assert!( + first.next_state.starts_with("v1:uidnext:"), + "a mailbox with no mod-sequences must checkpoint by UID boundary, got {:?}", + first.next_state + ); + assert!(first.present.is_some()); +} + +#[tokio::test] +async fn a_uidvalidity_change_refetches_rather_than_deleting_the_folder() { + // After a renumbering every stored UID names a different message. The + // dangerous outcome would be enumerating the new UID space, finding none + // of the old ids in it, and concluding the whole folder was deleted. + let server = MockImapServer::start(ServerProfile::Qresync) + .await + .expect("server starts"); + server.create_mailbox("INBOX"); + server.append_message("INBOX", "Subject: before\r\n\r\nx", &[]); + + let first = sync_folder(&server, None).await; + assert_eq!(first.upserts.len(), 1); + + server.bump_uid_validity("INBOX"); + server.append_message("INBOX", "Subject: after\r\n\r\nx", &[]); + + let second = sync_folder(&server, Some(&first.next_state)).await; + + assert_eq!( + second.upserts.len(), + 1, + "a renumbered mailbox must be re-fetched in full" + ); + assert!( + second.removals.is_empty(), + "a renumbering is not a removal report" + ); + // The re-fetched message has a different surrogate, because UIDVALIDITY is + // part of identity -- so the caller replaces rather than merges. + assert_ne!(second.upserts[0].id, first.upserts[0].id); +} diff --git a/crates/nuncio-store/src/db.rs b/crates/nuncio-store/src/db.rs index 50b6e37..1c53ea7 100644 --- a/crates/nuncio-store/src/db.rs +++ b/crates/nuncio-store/src/db.rs @@ -29,6 +29,22 @@ pub enum DatabaseError { /// Recovery salvage operation failed. #[error("database recovery error: {0}")] RecoveryFailed(String), + /// The database was written by a newer binary than this one. + /// + /// Fail-closed: the schema is built from `CREATE TABLE IF NOT EXISTS`, so + /// an older binary would otherwise open a newer database successfully and + /// write the old model into it, leaving two mutually invisible truths in + /// one file. There is no safe way to continue. + #[error( + "database identity schema v{on_disk} is newer than this build supports (v{supported}); \ + upgrade nunciod rather than running an older binary against it" + )] + SchemaTooNew { + /// The generation recorded in the database's `user_version`. + on_disk: i64, + /// The newest generation this binary understands. + supported: i64, + }, /// Audit chain verification failed. #[error("audit chain integrity error: {0}")] ChainIntegrityFailed(String), @@ -496,6 +512,102 @@ impl DatabaseEngine { &self.pool } + /// The identity-schema generation this binary understands. + /// + /// Distinct from the additive column migrations elsewhere in this module. + /// Those add nullable columns and leave every stored row meaningful. This + /// counter tracks changes to what a stored message *is* -- how it is keyed + /// and what a sync checkpoint means -- where old rows cannot be reinterpreted + /// under the new model at all. + /// + /// Bump this only for a change of that kind, and only together with the + /// reset it implies (see [`DatabaseEngine::reconcile_schema_version`]). + pub const IDENTITY_SCHEMA_VERSION: i64 = 1; + + /// Reconcile the on-disk identity-schema generation with this binary's. + /// + /// Three outcomes: + /// + /// - **Same version** (and the common case): nothing happens. + /// - **Database older**: the cached message store is rebuilt and the account + /// re-syncs from the server. Accounts, credentials and filter rules are + /// preserved; cached mail, sync checkpoints and the execution ledger are + /// not. This is deliberate and pre-release only -- see ADR 0002. Message + /// identity cannot be recomputed for rows already stored, so the + /// alternative is a permanently two-tier store in which old messages are + /// invisible to every guarantee the new model provides. The server holds + /// the authoritative copy either way. + /// - **Database newer**: refuse to open. A `CREATE TABLE IF NOT EXISTS` + /// schema means an older binary can open a newer database and quietly + /// write the old model into it, leaving two mutually invisible truths in + /// one file. Failing loudly is the only way that does not corrupt. + pub async fn reconcile_schema_version(&self) -> Result<(), DatabaseError> { + let (on_disk,): (i64,) = sqlx::query_as("PRAGMA user_version") + .fetch_one(&self.pool) + .await + .map_err(DatabaseError::Query)?; + + if on_disk > Self::IDENTITY_SCHEMA_VERSION { + return Err(DatabaseError::SchemaTooNew { + on_disk, + supported: Self::IDENTITY_SCHEMA_VERSION, + }); + } + + // 0 is both "brand new database" and "predates this counter". Neither + // has anything worth preserving that a re-sync will not restore, and + // the reset is a no-op on an empty store. + if on_disk < Self::IDENTITY_SCHEMA_VERSION { + if on_disk > 0 { + tracing::warn!( + on_disk, + supported = Self::IDENTITY_SCHEMA_VERSION, + "message identity schema changed: rebuilding the cached message store; \ + accounts and rules are preserved and mail will re-sync from the server" + ); + } + self.reset_cached_message_store().await?; + sqlx::query(&format!( + "PRAGMA user_version = {}", + Self::IDENTITY_SCHEMA_VERSION + )) + .execute(&self.pool) + .await + .map_err(DatabaseError::Query)?; + } + + Ok(()) + } + + /// Drop everything derived from the server, keeping everything the user + /// configured. + /// + /// Cleared: messages and their FTS index, per-folder sync checkpoints, and + /// the filter execution ledger. The ledger goes because its hash chain is + /// keyed on message ids that will no longer resolve, and a chain pointing + /// at absent rows has no audit value -- reinitialising it is more honest + /// than carrying a permanently mixed-namespace log. + /// + /// Kept: accounts, credentials (which live in the OS keyring regardless), + /// filter rules, and the WORM audit records, whose triggers forbid deletion + /// and whose entries do not reference message identity. + async fn reset_cached_message_store(&self) -> Result<(), DatabaseError> { + let mut tx = self.pool.begin().await.map_err(DatabaseError::Query)?; + for statement in [ + "DELETE FROM messages", + "DELETE FROM messages_fts", + "DELETE FROM folder_sync_state", + "DELETE FROM filter_execution_logs", + ] { + sqlx::query(statement) + .execute(&mut *tx) + .await + .map_err(DatabaseError::Query)?; + } + tx.commit().await.map_err(DatabaseError::Query)?; + Ok(()) + } + /// Execute initial database migrations creating core envelope tables. pub async fn migrate(&self) -> Result<(), DatabaseError> { sqlx::query( @@ -750,6 +862,12 @@ impl DatabaseEngine { // The statements above create the schema in full, so there is nothing to // upgrade: every column this engine reads is declared in its own // `CREATE TABLE IF NOT EXISTS`. Databases are created, never migrated. + // + // Reconciling the identity-schema generation is a different question and + // still applies: it asks whether a store was written under a model this + // binary can still interpret, and rebuilds (or refuses) accordingly. + // Runs after the tables exist, since the rebuild deletes from them. + self.reconcile_schema_version().await?; self.backfill_message_fts().await?; Ok(()) @@ -1374,6 +1492,51 @@ impl DatabaseEngine { /// Lets a caller classify a whole fetched batch as new-vs-seen with one round trip instead /// of one existence lookup per id. Passing an empty slice short-circuits to an empty set /// without querying at all, since `IN ()` is not valid SQL. + /// Every message id currently stored for one folder of one account. + /// + /// The local half of a UID-set diff: a sync that enumerated the folder's + /// complete contents can subtract what the server reported from this to + /// find what disappeared. Scoped by account as well as folder because + /// folder ids are not globally unique -- two accounts both have an + /// `INBOX`, and diffing across them would delete one account's mail + /// because the other's server did not mention it. + pub async fn message_ids_in_folder( + &self, + account_id: &str, + folder_id: &str, + ) -> Result, DatabaseError> { + let rows: Vec<(String,)> = + sqlx::query_as("SELECT id FROM messages WHERE account_id = ? AND folder_id = ?") + .bind(account_id) + .bind(folder_id) + .fetch_all(&self.pool) + .await + .map_err(DatabaseError::Query)?; + Ok(rows.into_iter().map(|(id,)| id).collect()) + } + + /// Delete stored messages by id, returning how many rows went. + /// + /// The FTS index follows via the `messages_ad` trigger, so a deleted + /// message cannot survive as a searchable plaintext orphan. + pub async fn delete_messages(&self, ids: &[String]) -> Result { + if ids.is_empty() { + return Ok(0); + } + let mut tx = self.pool.begin().await.map_err(DatabaseError::Query)?; + let mut deleted = 0u64; + for id in ids { + let result = sqlx::query("DELETE FROM messages WHERE id = ?") + .bind(id) + .execute(&mut *tx) + .await + .map_err(DatabaseError::Query)?; + deleted += result.rows_affected(); + } + tx.commit().await.map_err(DatabaseError::Query)?; + Ok(deleted) + } + pub async fn existing_message_ids( &self, ids: &[String], @@ -3868,6 +4031,104 @@ mod tests { assert_eq!(fetched.folder_id, "INBOX"); } + #[tokio::test] + async fn opening_a_newer_database_fails_closed_instead_of_writing_the_old_model() { + let (engine, _dir) = DatabaseEngine::connect_ephemeral().await.unwrap(); + sqlx::query(&format!( + "PRAGMA user_version = {}", + DatabaseEngine::IDENTITY_SCHEMA_VERSION + 1 + )) + .execute(engine.pool()) + .await + .unwrap(); + + let err = engine + .reconcile_schema_version() + .await + .expect_err("a database from a newer build must not be opened"); + assert!( + matches!(err, DatabaseError::SchemaTooNew { .. }), + "got {err:?}" + ); + // The schema is CREATE TABLE IF NOT EXISTS, so without this an older + // binary opens a newer database happily and writes the old model into + // it, leaving two mutually invisible truths in one file. + assert!(err.to_string().contains("newer than this build supports")); + } + + #[tokio::test] + async fn an_identity_schema_bump_rebuilds_cached_mail_and_keeps_configuration() { + let (engine, _dir) = DatabaseEngine::connect_ephemeral().await.unwrap(); + + let account = nuncio_core::AccountConfig { + id: "acct-reset".to_string(), + name: "Reset Account".to_string(), + email_address: "reset@nuncio.mx".to_string(), + keyring_secret_key: "nuncio/acct-reset".to_string(), + sync_interval_secs: 60, + transport: nuncio_core::Transport::ImapSmtp(nuncio_core::ImapSmtpTransport { + imap_host: "imap.nuncio.mx".to_string(), + imap_port: 993, + imap_tls_mode: nuncio_core::TlsMode::ImplicitTls, + smtp_host: "smtp.nuncio.mx".to_string(), + smtp_port: 465, + smtp_tls_mode: nuncio_core::TlsMode::ImplicitTls, + }), + }; + engine.save_account(&account).await.expect("save account"); + engine + .save_email(&synced_email("acct-reset", "INBOX", "42", "7")) + .await + .expect("save email"); + engine + .save_folder_sync_state("acct-reset", "INBOX", "v1:uidnext:42:8") + .await + .expect("save checkpoint"); + + // Pretend this store was written by the previous identity generation. + sqlx::query("PRAGMA user_version = 0") + .execute(engine.pool()) + .await + .unwrap(); + + engine + .reconcile_schema_version() + .await + .expect("an older store is rebuilt, not rejected"); + + // Server-derived state goes: it cannot be reinterpreted under a new + // identity model, and the server holds the authoritative copy. + let (message_count,): (i64,) = sqlx::query_as("SELECT COUNT(*) FROM messages") + .fetch_one(engine.pool()) + .await + .unwrap(); + assert_eq!(message_count, 0, "cached mail must be rebuilt"); + assert_eq!( + engine + .get_folder_sync_state("acct-reset", "INBOX") + .await + .unwrap(), + None, + "a checkpoint must not survive an identity change" + ); + + // Everything the user configured survives. + let accounts = engine.list_accounts().await.expect("list accounts"); + assert_eq!(accounts.len(), 1, "accounts must be preserved"); + assert_eq!(accounts[0].id, "acct-reset"); + + // The version is recorded, so the next open is a no-op. + let (version,): (i64,) = sqlx::query_as("PRAGMA user_version") + .fetch_one(engine.pool()) + .await + .unwrap(); + assert_eq!(version, DatabaseEngine::IDENTITY_SCHEMA_VERSION); + engine + .reconcile_schema_version() + .await + .expect("re-running at the current version is a no-op"); + } + fn export_test_email(id: &str, account_id: &str, folder_id: &str) -> nuncio_core::model::Email { nuncio_core::model::Email { id: id.to_string(), diff --git a/crates/nunciod/src/sync.rs b/crates/nunciod/src/sync.rs index 0c5036a..aca69d5 100644 --- a/crates/nunciod/src/sync.rs +++ b/crates/nunciod/src/sync.rs @@ -245,10 +245,12 @@ async fn fetch_and_persist( Some(acct) => db.get_folder_sync_state(acct, &folder.id).await?, None => None, }; - let (emails, new_state) = backend - .sync_messages(&folder.id, last_state.as_deref()) + let changes = backend + .sync_changes(&folder.id, last_state.as_deref()) .await?; + let emails = changes.upserts; let fetched = emails.len(); + let new_state = changes.next_state; // One round trip classifies the whole chunk instead of one lookup per // message. A lookup failure that is not "not found" would be a @@ -268,6 +270,37 @@ async fn fetch_and_persist( } synced += 1; } + // Reconcile what the server no longer has. Until this existed, a sync + // could only ever add: a message moved or deleted by any other client + // -- another daemon, a phone, webmail -- stayed in the local store + // forever, and every account accumulated ghosts. + if let Some(acct) = account_id { + let mut gone: Vec = changes.removals; + + // `present` is the folder's complete contents when the pass was + // able to enumerate them. `None` and `Some(vec![])` are NOT the + // same: `None` means the pass was incremental and cannot speak to + // absence, and treating it as "nothing is present" would delete + // the folder. + if let Some(present) = changes.present { + let present: std::collections::HashSet = present.into_iter().collect(); + let stored = db.message_ids_in_folder(acct, &folder.id).await?; + gone.extend(stored.into_iter().filter(|id| !present.contains(id))); + } + + gone.sort_unstable(); + gone.dedup(); + if !gone.is_empty() { + let deleted = db.delete_messages(&gone).await?; + tracing::info!( + account_id = %acct, + folder_id = %folder.id, + deleted, + "removed messages that are no longer on the server" + ); + } + } + // Persist the returned checkpoint only AFTER this folder's messages // have landed, so the stored high-water mark can never advance past // work that actually reached the store.