From 1ec1e90772ea782df5438705fbd0921b7b2e2e36 Mon Sep 17 00:00:00 2001 From: James Maes Date: Mon, 10 Aug 2026 15:34:24 -0500 Subject: [PATCH 1/2] feat(store): gate the identity schema and tag sync tokens with their scheme Two silent-failure paths, both cheap to close and both load-bearing for the sync work that follows. The schema is built entirely from CREATE TABLE IF NOT EXISTS, so an older binary opens a newer database successfully and writes the old model into it. Nothing detects that; the file ends up holding two mutually invisible truths. A user_version generation now gates opening: newer than this build is a fail-closed error, and older rebuilds the cached message store. Rebuilding rather than migrating is the pre-release decision recorded in ADR 0002. Message identity cannot be recomputed for rows already stored, so the alternative is keeping them under the old model forever -- a store where half the mail is invisible to every guarantee the new model provides. Accounts, credentials and rules are kept; cached mail, checkpoints and the execution ledger are rebuilt from the server, which is authoritative regardless. The ledger goes with them because its hash chain is keyed on message ids that would no longer resolve, and a chain pointing at absent rows proves nothing. Separately, folder checkpoints now carry a scheme tag. The stored value was "{uidvalidity}:{uidnext}" and the QRESYNC checkpoint that replaces it is "{uidvalidity}:{highestmodseq}" -- identical shape, different meaning. An upgraded engine 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 since" and every message below that point is never enumerated again. That is silent, permanent mail loss with no error anywhere. Any token whose scheme this build does not recognise, including an untagged one, now forces a full re-fetch: over-fetching is recoverable, misreading is not. Refs #387 --- crates/nuncio-mail/src/imap.rs | 98 ++++++++++++--- crates/nuncio-store/src/db.rs | 214 +++++++++++++++++++++++++++++++++ 2 files changed, 296 insertions(+), 16 deletions(-) diff --git a/crates/nuncio-mail/src/imap.rs b/crates/nuncio-mail/src/imap.rs index 2de83f7..ca91725 100644 --- a/crates/nuncio-mail/src/imap.rs +++ b/crates/nuncio-mail/src/imap.rs @@ -86,6 +86,22 @@ fn parse_uid_validity(uid_validity: &str) -> Option { /// first turns that into a safe full re-fetch instead. const CHECKPOINT_DELIM: char = ':'; +/// 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"; + /// 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 @@ -98,6 +114,11 @@ enum FetchRange { /// 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. + /// + /// Also covers a checkpoint whose scheme tag this build does not recognise + /// (see [`CHECKPOINT_SCHEME_UIDNEXT`]) -- including an untagged one, which + /// cannot be distinguished from a future scheme that happens to share its + /// shape. CorruptCheckpoint, /// The stored checkpoint's UIDVALIDITY no longer matches the mailbox's /// current UIDVALIDITY (the mailbox was renumbered) or the server did not @@ -140,7 +161,17 @@ fn resolve_fetch_range(since_state: Option<&str>, current_uid_validity: Option 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), + (Some(validity), 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()), @@ -1534,15 +1568,15 @@ mod tests { // Valid checkpoint whose UIDVALIDITY still matches: narrowed fetch. assert_eq!( - resolve_fetch_range(Some("1:105"), Some(1)), + resolve_fetch_range(Some("v1:uidnext:1:105"), Some(1)), FetchRange::Incremental(105) ); assert_eq!( - resolve_fetch_range(Some("1:105"), Some(1)).sequence_set(), + resolve_fetch_range(Some("v1:uidnext:1:105"), Some(1)).sequence_set(), "105:*" ); assert_eq!( - resolve_fetch_range(Some(" 7 : 42 "), Some(7)), + resolve_fetch_range(Some("v1:uidnext: 7 : 42 "), Some(7)), FetchRange::Incremental(42) ); @@ -1550,18 +1584,18 @@ mod tests { // 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)), + resolve_fetch_range(Some("v1:uidnext:1:105"), Some(2)), FetchRange::UidValidityChanged { stored: 1, current: Some(2) } ); assert_eq!( - resolve_fetch_range(Some("1:105"), Some(2)).sequence_set(), + resolve_fetch_range(Some("v1:uidnext:1:105"), Some(2)).sequence_set(), "1:*" ); assert_eq!( - resolve_fetch_range(Some("1:105"), None), + resolve_fetch_range(Some("v1:uidnext:1:105"), None), FetchRange::UidValidityChanged { stored: 1, current: None @@ -1580,7 +1614,7 @@ mod tests { "a legacy bare-number checkpoint must not be read as a UID boundary" ); assert_eq!( - resolve_fetch_range(Some("1:0"), Some(1)), + resolve_fetch_range(Some("v1:uidnext:1:0"), Some(1)), FetchRange::CorruptCheckpoint ); assert_eq!( @@ -1679,7 +1713,7 @@ mod tests { .await .expect("first sync succeeds"); 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. @@ -1687,7 +1721,7 @@ mod tests { .sync_folder_messages_with_session("INBOX", Some(&checkpoint), &mut session) .await .expect("second sync succeeds"); - assert_eq!(checkpoint2, "1:105"); + assert_eq!(checkpoint2, "v1:uidnext:1:105"); drop(session); let _ = server.await; @@ -1868,7 +1902,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 +1991,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,7 +2005,7 @@ 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" ); assert!( @@ -2046,7 +2080,7 @@ mod tests { .sync_folder_messages_with_session("INBOX", None, &mut session) .await .expect("first sync succeeds"); - assert_eq!(checkpoint, "1:105"); + 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 @@ -2056,7 +2090,7 @@ mod tests { .await .expect("second sync succeeds"); assert_eq!( - checkpoint2, "2:105", + checkpoint2, "v1:uidnext:2:105", "checkpoint must be rewritten under the new UIDVALIDITY" ); @@ -2146,6 +2180,38 @@ mod tests { assert_eq!(parse_imap_uid("jmap-object-id"), None); } + #[test] + fn an_untagged_or_unknown_scheme_checkpoint_forces_a_full_refetch() { + // The hazard this tag exists for. A QRESYNC checkpoint is + // `(uidvalidity, highestmodseq)` -- the same shape as this scheme's + // `(uidvalidity, uidnext)` and a different meaning. Read one as the + // other and the server is handed a UIDNEXT where it expects a MODSEQ; + // since UIDNEXT is normally the larger number it answers "nothing + // changed" and everything below is never enumerated again. Refusing + // to interpret an unrecognised token is the only safe reading. + for token in [ + // Pre-tagging, written by an older build. + "1:105", + // A scheme from a future build. + "v2:qresync:1:900", + // Tag-shaped but not ours. + "v1:modseq:1:105", + // Truncated tag. + "v1:1:105", + ] { + assert_eq!( + resolve_fetch_range(Some(token), Some(1)), + FetchRange::CorruptCheckpoint, + "token {token:?} must not be interpreted as a UIDNEXT boundary" + ); + assert_eq!( + resolve_fetch_range(Some(token), Some(1)).sequence_set(), + "1:*", + "an uninterpretable token must re-fetch everything" + ); + } + } + #[test] fn parse_uid_validity_accepts_only_a_decimal_value() { assert_eq!(parse_uid_validity("7"), Some(7)); diff --git a/crates/nuncio-store/src/db.rs b/crates/nuncio-store/src/db.rs index c22ba55..c34829a 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,10 @@ impl DatabaseEngine { self.ensure_accounts_smtp_columns().await?; self.ensure_accounts_tls_mode_columns().await?; self.ensure_accounts_dav_columns().await?; + // Runs after the tables exist (the reset deletes from them) and after + // the additive column migrations, so a rebuild lands on the current + // shape rather than an intermediate one. + self.reconcile_schema_version().await?; self.ensure_messages_identity_columns().await?; self.backfill_message_fts().await?; @@ -4314,6 +4430,104 @@ mod tests { /// the empty-string default -- with the UNIQUE index not yet built. The /// migration must complete that backfill on the next open and build the /// index without a collision, rather than wedging the daemon. + #[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"); + } + #[tokio::test] async fn messages_identity_migration_completes_from_a_partial_run() { let (engine, _dir) = DatabaseEngine::connect_ephemeral().await.unwrap(); -- 2.54.0 From 1f75527563afff7965b0be949d6cd232c6ae9be1 Mon Sep 17 00:00:00 2001 From: James Maes Date: Mon, 10 Aug 2026 16:50:48 -0500 Subject: [PATCH 2/2] feat(sync): enumerate IMAP changes over a four-rung capability ladder Sync could only ever add. It resolved a UID boundary, fetched forward from it, and stored what came back; nothing ever asked what stopped being there. 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. This is the defect, and it affects a single engine just as much as ten. A folder pass now reports what changed, what vanished, and honestly whether it was able to observe absence at all. The last part is why `present` is an Option: None means the pass was incremental and cannot speak to absence, while Some(empty) means the folder is genuinely empty. Conflating those deletes a mailbox. Four rungs, chosen per mailbox rather than per account, because NOMODSEQ is a mailbox property (RFC 7162 3.1.2.2) and one folder can be unable to answer mod-sequence queries on a server that otherwise can. QRESYNC has the server state removals outright. CONDSTORE narrows the fetch but never volunteers them, so absence comes from enumerating what remains -- this is Gmail's rung, which advertises CONDSTORE and not QRESYNC, and a two-rung ladder would have pushed the largest provider onto full sweeps against a published daily byte cap. Neither extension leaves a UID boundary for arrivals plus an enumeration for removals; that is Exchange. No usable checkpoint fetches everything. The enumeration passes ask for UID only -- no bodies, no flags -- so detecting removals costs one short line per message rather than a download. A mailbox that can supply a mod-sequence gets a mod-sequence checkpoint even after a full pass, so the next sync starts on a higher rung than this one did. A UIDVALIDITY change is not a downgrade to a lesser rung, it is a full pass: UIDs restart, so every stored boundary names a different message. The dangerous reading would be to enumerate the new UID space, find none of the stored ids in it, and conclude the folder was emptied. The raw collector also gains a bounded read. A server that accepts a command and then stops answering must surface as a stall rather than hanging the sync, which is the same failure the typed fetch path already bounds. Refs #390 --- crates/nuncio-mail/src/backend.rs | 50 +- crates/nuncio-mail/src/imap.rs | 841 +++++++++++++----- crates/nuncio-mail/src/imap_raw.rs | 19 +- crates/nuncio-mail/src/jmap.rs | 15 +- crates/nuncio-mail/src/mock.rs | 21 +- .../nuncio-mail/tests/imap_raw_wire_test.rs | 179 ++++ crates/nuncio-store/src/db.rs | 45 + crates/nunciod/src/sync.rs | 37 +- 8 files changed, 966 insertions(+), 241 deletions(-) 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 ca91725..6e883bc 100644 --- a/crates/nuncio-mail/src/imap.rs +++ b/crates/nuncio-mail/src/imap.rs @@ -102,95 +102,147 @@ const CHECKPOINT_DELIM: char = ':'; /// full re-fetch. Over-fetching is recoverable; misreading a token is not. const CHECKPOINT_SCHEME_UIDNEXT: &str = "v1:uidnext"; -/// 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 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. - /// - /// Also covers a checkpoint whose scheme tag this build does not recognise - /// (see [`CHECKPOINT_SCHEME_UIDNEXT`]) -- including an untagged one, which - /// cannot be distinguished from a future scheme that happens to share its - /// shape. - 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; - }; - // Require the scheme tag. A value without one is either pre-tagging or from - // a scheme this build cannot interpret; both must re-fetch rather than be - // guessed at. - let Some(body) = raw - .trim() - .strip_prefix(CHECKPOINT_SCHEME_UIDNEXT) - .and_then(|rest| rest.strip_prefix(CHECKPOINT_DELIM)) - else { - return FetchRange::CorruptCheckpoint; - }; - let parsed = body - .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, } } @@ -719,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, { @@ -758,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. @@ -888,8 +923,15 @@ 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!( + // 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 ), @@ -898,16 +940,222 @@ impl ImapEngine { .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 @@ -1049,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()), _ => { @@ -1457,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 } @@ -1526,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; @@ -1560,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("v1:uidnext:1:105"), Some(1)), - FetchRange::Incremental(105) - ); - assert_eq!( - resolve_fetch_range(Some("v1:uidnext:1:105"), Some(1)).sequence_set(), - "105:*" - ); - assert_eq!( - resolve_fetch_range(Some("v1:uidnext: 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("v1:uidnext:1:105"), Some(2)), - FetchRange::UidValidityChanged { - stored: 1, - current: Some(2) - } - ); - assert_eq!( - resolve_fetch_range(Some("v1:uidnext:1:105"), Some(2)).sequence_set(), - "1:*" - ); - assert_eq!( - resolve_fetch_range(Some("v1:uidnext: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("v1:uidnext: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]; @@ -1655,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 @@ -1677,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()) @@ -1708,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, "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"); + 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: {}", @@ -1842,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()) @@ -1869,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; @@ -2008,13 +2380,23 @@ mod tests { 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:?}" ); } @@ -2042,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()) @@ -2076,19 +2469,21 @@ 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"); + 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, "v1:uidnext:2:105", "checkpoint must be rewritten under the new UIDVALIDITY" @@ -2097,8 +2492,13 @@ mod tests { 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: {}", @@ -2127,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()) @@ -2180,38 +2591,6 @@ mod tests { assert_eq!(parse_imap_uid("jmap-object-id"), None); } - #[test] - fn an_untagged_or_unknown_scheme_checkpoint_forces_a_full_refetch() { - // The hazard this tag exists for. A QRESYNC checkpoint is - // `(uidvalidity, highestmodseq)` -- the same shape as this scheme's - // `(uidvalidity, uidnext)` and a different meaning. Read one as the - // other and the server is handed a UIDNEXT where it expects a MODSEQ; - // since UIDNEXT is normally the larger number it answers "nothing - // changed" and everything below is never enumerated again. Refusing - // to interpret an unrecognised token is the only safe reading. - for token in [ - // Pre-tagging, written by an older build. - "1:105", - // A scheme from a future build. - "v2:qresync:1:900", - // Tag-shaped but not ours. - "v1:modseq:1:105", - // Truncated tag. - "v1:1:105", - ] { - assert_eq!( - resolve_fetch_range(Some(token), Some(1)), - FetchRange::CorruptCheckpoint, - "token {token:?} must not be interpreted as a UIDNEXT boundary" - ); - assert_eq!( - resolve_fetch_range(Some(token), Some(1)).sequence_set(), - "1:*", - "an uninterpretable token must re-fetch everything" - ); - } - } - #[test] fn parse_uid_validity_accepts_only_a_decimal_value() { assert_eq!(parse_uid_validity("7"), Some(7)); @@ -2698,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; @@ -2730,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 c34829a..7bd02f9 100644 --- a/crates/nuncio-store/src/db.rs +++ b/crates/nuncio-store/src/db.rs @@ -1707,6 +1707,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], 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. -- 2.54.0