crates/nuncio-mail/src/imap.rs

@@ -86,6 +86,22 @@ fn parse_uid_validity(uid_validity: &str) -> Option<u32> {
 /// 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<u
     let Some(raw) = since_state else {
         return FetchRange::Full;
     };
-    let parsed = raw
+    // 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((
@@ -858,7 +889,10 @@ impl ImapEngine {
         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),
+            (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));

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