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