From 1ec1e90772ea782df5438705fbd0921b7b2e2e36 Mon Sep 17 00:00:00 2001 From: James Maes Date: Mon, 10 Aug 2026 15:34:24 -0500 Subject: [PATCH] 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