diff --git a/crates/nuncio-core/src/lib.rs b/crates/nuncio-core/src/lib.rs index 0bc6b68..5921251 100644 --- a/crates/nuncio-core/src/lib.rs +++ b/crates/nuncio-core/src/lib.rs @@ -93,6 +93,19 @@ pub enum CoreEvent { /// Action description. action_taken: String, }, + /// Progress notification emitted during an inbound mail sync, so a + /// long-running sync of a large mailbox is observably progressing rather + /// than indistinguishable from a hang. Emitted per folder as it completes. + SyncProgress { + /// Account being synced, if the sync is scoped to one. + account_id: Option, + /// Folder whose messages were just fetched and persisted. + folder_id: String, + /// Messages fetched from this folder on this pass. + fetched: usize, + /// Cumulative messages fetched across the sync so far. + total: usize, + }, /// Progress notification during bulk filter execution over a batch of emails. BatchFilterProgress { /// Messages processed so far in the batch. diff --git a/crates/nuncio-mail/Cargo.toml b/crates/nuncio-mail/Cargo.toml index 7099f8e..644a401 100644 --- a/crates/nuncio-mail/Cargo.toml +++ b/crates/nuncio-mail/Cargo.toml @@ -31,3 +31,6 @@ reqwest.workspace = true [dev-dependencies] tempfile.workspace = true wiremock.workspace = true +# `test-util` (not in tokio's `full`) enables paused-clock tests so the +# bounded per-item FETCH timeout can be asserted without a real-time wait. +tokio = { workspace = true, features = ["test-util"] } diff --git a/crates/nuncio-mail/src/imap.rs b/crates/nuncio-mail/src/imap.rs index 0366e6e..36e8a3c 100644 --- a/crates/nuncio-mail/src/imap.rs +++ b/crates/nuncio-mail/src/imap.rs @@ -23,9 +23,108 @@ use tokio_stream::StreamExt; /// hang a connect -- and therefore a `TestAccountConnection` RPC -- forever. const CONNECT_TIMEOUT: Duration = Duration::from_secs(15); +/// Upper bound on how long a single `FETCH` response item may take to arrive +/// once the `UID FETCH` command has been acknowledged. Fetching full bodies +/// for a large mailbox in one command can be legitimately slow overall, so the +/// bound is per item rather than for the whole stream; but a single item that +/// never arrives (dead socket, server-side hang) must still surface as an +/// honest timeout rather than hanging the sync indefinitely. +const FETCH_ITEM_TIMEOUT: Duration = Duration::from_secs(60); + use crate::backend::MailBackend; use crate::parser::{MailError, MimeParserAdapter}; +/// Delimiter between the two components of a folder sync checkpoint. The +/// checkpoint state string is `"{uidvalidity}{DELIM}{uidnext}"` (e.g. +/// `"42:105"`): the mailbox's UIDVALIDITY at the time it was written, then the +/// UID boundary to resume from. Both components are decimal `u32`. +/// +/// UIDVALIDITY MUST be carried because IMAP UIDs are only meaningful within a +/// stable UIDVALIDITY (RFC 3501 s2.3.1.1). If a mailbox is recreated and its +/// UIDVALIDITY changes, UIDs restart from low numbers, so resuming from a UID +/// boundary captured under the old UIDVALIDITY would issue `{old_uidnext}:*` +/// against renumbered low UIDs -- which per `n:*` semantics returns only the +/// single highest message and silently skips the rest. Comparing UIDVALIDITY +/// first turns that into a safe full re-fetch instead. +const CHECKPOINT_DELIM: char = ':'; + +/// The `UID FETCH` sequence-set to issue for a folder sync, resolved from the +/// stored checkpoint. Kept as a small typed value (rather than collapsing to a +/// bare range string) so the distinct fall-back reasons stay observable: all +/// of `Full`/`CorruptCheckpoint`/`UidValidityChanged` fetch `1:*`, but only the +/// latter two are worth flagging, and tests can assert on which branch fired. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum FetchRange { + /// No prior checkpoint (first-ever sync of this folder): fetch everything. + Full, + /// A checkpoint was stored but could not be parsed (malformed, or a legacy + /// bare-UID checkpoint predating UIDVALIDITY tracking). Falls back to a + /// full fetch, surfaced (logged) rather than silently ignored. + CorruptCheckpoint, + /// The stored checkpoint's UIDVALIDITY no longer matches the mailbox's + /// current UIDVALIDITY (the mailbox was renumbered) or the server did not + /// report a current UIDVALIDITY to compare against. Stored UIDs are no + /// longer meaningful, so this falls back to a safe full re-fetch. + UidValidityChanged { + /// UIDVALIDITY recorded in the stored checkpoint. + stored: u32, + /// UIDVALIDITY the server reports now, if any. + current: Option, + }, + /// A valid checkpoint whose UIDVALIDITY still matches: fetch only UIDs at + /// or above the stored boundary, i.e. messages that arrived since. + Incremental(u32), +} + +impl FetchRange { + /// Render the IMAP `UID FETCH` sequence-set. Every non-incremental variant + /// fetches all UIDs (`1:*`); an incremental checkpoint `n` fetches `n:*`, + /// which the server bounds to UIDs `>= n` -- a genuinely narrower fetch. + fn sequence_set(self) -> String { + match self { + FetchRange::Full + | FetchRange::CorruptCheckpoint + | FetchRange::UidValidityChanged { .. } => "1:*".to_string(), + FetchRange::Incremental(uid) => format!("{}:*", uid), + } + } +} + +/// Resolve the fetch range from a stored `since_state` checkpoint and the +/// mailbox's CURRENT UIDVALIDITY (as just reported by `SELECT`). `None` means +/// no prior sync (full fetch). A checkpoint in the `"{uidvalidity}:{uidnext}"` +/// format proceeds incrementally only when its UIDVALIDITY still matches the +/// server's current one; a mismatch (mailbox renumbered) or a missing current +/// UIDVALIDITY forces a safe full re-fetch. A malformed or legacy bare-number +/// checkpoint is treated as corrupt (full fetch, flagged), never misread as a +/// live UID boundary. +fn resolve_fetch_range(since_state: Option<&str>, current_uid_validity: Option) -> FetchRange { + let Some(raw) = since_state else { + return FetchRange::Full; + }; + let parsed = raw + .split_once(CHECKPOINT_DELIM) + .and_then(|(validity, uid)| { + Some(( + validity.trim().parse::().ok()?, + uid.trim().parse::().ok()?, + )) + }); + 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, + }, + } +} + /// State of the dedicated IMAP IDLE socket listener. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum IdleSocketState { @@ -449,21 +548,53 @@ impl ImapEngine { &self.socket_manager } - /// Fetch and parse all messages in a folder over an active IMAP session. + /// Fetch and parse the messages in a folder over an active IMAP session, + /// returning the fetched messages and the new sync checkpoint. + /// + /// `since_state` is the checkpoint returned by a previous sync of this + /// folder, encoded as `"{uidvalidity}:{uidnext}"` (see [`CHECKPOINT_DELIM`]). + /// The `UID FETCH` is narrowed to `{uidnext}:*` only when the checkpoint's + /// UIDVALIDITY still matches the mailbox's current UIDVALIDITY; a mismatch + /// (the mailbox was renumbered) or a malformed/legacy checkpoint forces a + /// safe full re-fetch instead of silently skipping renumbered mail. The + /// returned checkpoint is derived from real server state -- the mailbox's + /// UIDVALIDITY plus its `UIDNEXT` (or one past the highest UID actually + /// seen) after selecting the folder -- never from a synthetic counter. pub async fn sync_folder_messages_with_session( &self, folder_id: &str, + since_state: Option<&str>, session: &mut async_imap::Session, - ) -> Result, MailError> + ) -> Result<(Vec, String), MailError> where S: AsyncRead + AsyncWrite + Unpin + Send + Debug + 'static, { - session.select(folder_id).await.map_err(|e| { + 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 query = build_fetch_command_query(); - let mut fetch_stream = session.uid_fetch("1:*", query).await.map_err(|e| { + let sequence_set = range.sequence_set(); + let mut fetch_stream = session.uid_fetch(&sequence_set, query).await.map_err(|e| { MailError::ImapError(format!( "UID FETCH failed for folder '{}': {}", folder_id, e @@ -471,11 +602,29 @@ impl ImapEngine { })?; let mut emails = Vec::new(); - while let Some(fetch_res) = fetch_stream.next().await { + let mut max_uid_seen: u32 = 0; + loop { + // Bound EVERY individual item read: a slow-but-progressing large + // fetch is fine, but a single item that never arrives must surface + // as an honest `FetchStalled` instead of hanging the whole sync. + let next_item = tokio::time::timeout(FETCH_ITEM_TIMEOUT, fetch_stream.next()) + .await + .map_err(|_| { + MailError::FetchStalled(format!( + "no FETCH response for folder '{}' within {}s ({} message(s) read so far)", + folder_id, + FETCH_ITEM_TIMEOUT.as_secs(), + emails.len() + )) + })?; + let Some(fetch_res) = next_item else { + break; + }; let fetch_data = fetch_res .map_err(|e| MailError::ImapError(format!("failed reading fetch item: {}", e)))?; let uid_num = fetch_data.uid.unwrap_or(0); + max_uid_seen = max_uid_seen.max(uid_num); let email_id = format!("imap-uid-{}", uid_num); let is_read = fetch_data @@ -559,14 +708,44 @@ impl ImapEngine { emails.push(email); } - Ok(emails) + // Derive the next checkpoint from real server state as + // `"{uidvalidity}:{uidnext}"`. The UID boundary is the mailbox's + // authoritative UIDNEXT, else one past the highest UID actually seen. + // A checkpoint is only meaningful when both a UIDVALIDITY and a UID + // boundary are known; if either is missing (a server that omitted + // UIDVALIDITY, or an empty fetch that revealed no UIDNEXT/UID), keep + // the prior checkpoint rather than writing a validity-less one that the + // next sync could misread. As a last resort (no prior checkpoint + // either) emit an intentionally unparseable marker so the next sync + // safely falls back to a full fetch. + let boundary_uid = server_uid_next + .or_else(|| (max_uid_seen > 0).then_some(max_uid_seen.saturating_add(1))); + let new_checkpoint = match (server_uid_validity, boundary_uid) { + (Some(validity), Some(uid)) => format!("{}{}{}", validity, CHECKPOINT_DELIM, uid), + _ => since_state + .map(str::to_string) + .unwrap_or_else(|| "full-resync-required".to_string()), + }; + + Ok((emails, new_checkpoint)) } - /// Execute `UID FETCH 1:* (FLAGS INTERNALDATE RFC822.SIZE ENVELOPE BODY.PEEK[])` and parse returned messages into domain `Email` entities. - pub async fn sync_folder_messages(&self, folder_id: &str) -> Result, MailError> { + /// Fetch a folder's messages over a freshly authenticated session, + /// honoring `since_state` for an incremental fetch and returning the new + /// sync checkpoint alongside the messages. See + /// [`Self::sync_folder_messages_with_session`] for the checkpoint model. + pub async fn sync_folder_messages( + &self, + folder_id: &str, + since_state: Option<&str>, + ) -> Result<(Vec, String), MailError> { let (username, password) = match (&self.username, &self.password) { (Some(u), Some(p)) => (u.as_str(), p.as_str()), _ => { + // Credential-less fixture path used by unit tests that never + // reach a real server. The checkpoint (one past the fixture's + // UID) is internally consistent with the fixture rather than + // derived from `emails.len()`. let mock_emails = vec![Email { id: "imap-uid-100".to_string(), account_id: self.account_id.clone(), @@ -580,7 +759,7 @@ impl ImapEngine { body_html: None, attachments: Vec::new(), }]; - return Ok(mock_emails); + return Ok((mock_emails, "101".to_string())); } }; @@ -588,11 +767,11 @@ impl ImapEngine { .socket_manager .connect_session(username, password) .await?; - let emails = self - .sync_folder_messages_with_session(folder_id, &mut session) - .await?; + let result = self + .sync_folder_messages_with_session(folder_id, since_state, &mut session) + .await; let _ = session.logout().await; - Ok(emails) + result } /// Listen for real-time IMAP IDLE notification events on an active IMAP session. @@ -742,11 +921,9 @@ impl MailBackend for ImapEngine { async fn sync_messages( &self, folder_id: &str, - _since_state: Option<&str>, + since_state: Option<&str>, ) -> Result<(Vec, String), MailError> { - let emails = self.sync_folder_messages(folder_id).await?; - let modseq = format!("imap-modseq-{}", emails.len()); - Ok((emails, modseq)) + self.sync_folder_messages(folder_id, since_state).await } async fn send_email(&self, _email: &Email) -> Result<(), MailError> { @@ -781,8 +958,9 @@ mod tests { assert_eq!(folders.len(), 2); assert_eq!(folders[0].id, "INBOX"); - let (emails, modseq) = engine.sync_messages("INBOX", None).await?; - assert_eq!(modseq, "imap-modseq-1"); + let (emails, checkpoint) = engine.sync_messages("INBOX", None).await?; + // Credential-less fixture: checkpoint is one past the fixture UID. + assert_eq!(checkpoint, "101"); assert_eq!(emails.len(), 1); assert_eq!(emails[0].id, "imap-uid-100"); @@ -790,6 +968,342 @@ mod tests { Ok(()) } + #[test] + fn resolve_fetch_range_distinguishes_none_valid_corrupt_and_renumbered_checkpoints() { + // First-ever sync: full history (current UIDVALIDITY irrelevant). + assert_eq!(resolve_fetch_range(None, Some(1)), FetchRange::Full); + assert_eq!(resolve_fetch_range(None, None).sequence_set(), "1:*"); + + // Valid checkpoint whose UIDVALIDITY still matches: narrowed fetch. + assert_eq!( + resolve_fetch_range(Some("1:105"), Some(1)), + FetchRange::Incremental(105) + ); + assert_eq!( + resolve_fetch_range(Some("1:105"), Some(1)).sequence_set(), + "105:*" + ); + assert_eq!( + resolve_fetch_range(Some(" 7 : 42 "), Some(7)), + FetchRange::Incremental(42) + ); + + // UIDVALIDITY changed (mailbox renumbered) or the server reported none + // to compare against: MUST fall back to a full fetch, distinctly, so a + // renumber can never silently skip the low new UIDs. + assert_eq!( + resolve_fetch_range(Some("1:105"), Some(2)), + FetchRange::UidValidityChanged { + stored: 1, + current: Some(2) + } + ); + assert_eq!( + resolve_fetch_range(Some("1:105"), Some(2)).sequence_set(), + "1:*" + ); + assert_eq!( + resolve_fetch_range(Some("1:105"), None), + FetchRange::UidValidityChanged { + stored: 1, + current: None + } + ); + + // Malformed or legacy bare-number (pre-UIDVALIDITY) checkpoints are + // corrupt -> full fetch, never misread as a live UID boundary. + assert_eq!( + resolve_fetch_range(Some(""), Some(1)), + FetchRange::CorruptCheckpoint + ); + assert_eq!( + resolve_fetch_range(Some("105"), Some(1)), + FetchRange::CorruptCheckpoint, + "a legacy bare-number checkpoint must not be read as a UID boundary" + ); + assert_eq!( + resolve_fetch_range(Some("1:0"), Some(1)), + FetchRange::CorruptCheckpoint + ); + assert_eq!( + resolve_fetch_range(Some("not:auid"), Some(1)), + FetchRange::CorruptCheckpoint + ); + assert_eq!( + resolve_fetch_range(Some("not-a-uid"), Some(1)).sequence_set(), + "1:*" + ); + } + + /// Read a single CRLF-terminated line from a scripted-server stream, + /// returning `None` at EOF (client hung up). Used only by the in-memory + /// duplex tests below; the real fetch path never parses raw lines. + async fn read_scripted_line(reader: &mut R) -> Option { + let mut buf = Vec::new(); + let mut byte = [0u8; 1]; + loop { + match reader.read(&mut byte).await { + Ok(0) => { + return if buf.is_empty() { + None + } else { + Some(String::from_utf8_lossy(&buf).into_owned()) + } + } + Ok(_) => { + if byte[0] == b'\n' { + return Some(String::from_utf8_lossy(&buf).into_owned()); + } + if byte[0] != b'\r' { + buf.push(byte[0]); + } + } + Err(_) => return None, + } + } + } + + #[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 + // whose server side is a scripted IMAP responder. Proves the SECOND + // sync issues a GENUINELY narrower `UID FETCH` derived from the first + // sync's returned checkpoint -- not a full re-fetch. + let (client_io, server_io) = tokio::io::duplex(8192); + let fetch_ranges = Arc::new(std::sync::Mutex::new(Vec::::new())); + let server_ranges = fetch_ranges.clone(); + + let server = tokio::spawn(async move { + let mut io = server_io; + 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("UID FETCH") { + if let Ok(mut g) = server_ranges.lock() { + g.push(line.clone()); + } + let _ = io + .write_all(format!("{tag} OK FETCH completed\r\n").as_bytes()) + .await; + } else if upper.contains("LOGIN") { + let _ = io + .write_all(format!("{tag} OK Logged in\r\n").as_bytes()) + .await; + } else if upper.contains("SELECT") { + let resp = format!( + "* FLAGS (\\Seen)\r\n* 3 EXISTS\r\n* 0 RECENT\r\n\ + * OK [UIDVALIDITY 1] ok\r\n* OK [UIDNEXT 105] ok\r\n\ + {tag} OK [READ-WRITE] SELECT done\r\n" + ); + let _ = io.write_all(resp.as_bytes()).await; + } else if upper.contains("LOGOUT") { + let _ = io + .write_all(format!("* BYE\r\n{tag} OK LOGOUT\r\n").as_bytes()) + .await; + break; + } + } + }); + + let client = async_imap::Client::new(client_io); + let mut session = client + .login("user", "pass") + .await + .map_err(|(e, _)| e) + .expect("scripted login succeeds"); + + let engine = ImapEngine::new("acct-1", "example.test", 993); + + // First sync: no checkpoint -> full history; returns + // "{uidvalidity}:{uidnext}" from real server state. + let (first, checkpoint) = engine + .sync_folder_messages_with_session("INBOX", None, &mut session) + .await + .expect("first sync succeeds"); + assert!(first.is_empty()); + assert_eq!(checkpoint, "1:105"); + + // Second sync: feed back the checkpoint; UIDVALIDITY still matches + // (1), so the fetch is narrowed. + let (_second, checkpoint2) = engine + .sync_folder_messages_with_session("INBOX", Some(&checkpoint), &mut session) + .await + .expect("second sync succeeds"); + assert_eq!(checkpoint2, "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"); + assert!( + ranges[0].contains("1:*"), + "first sync must fetch full history, got: {}", + ranges[0] + ); + assert!( + ranges[1].contains("105:*"), + "second sync must be narrowed to the checkpoint, got: {}", + ranges[1] + ); + assert!( + !ranges[1].contains("1:*"), + "second sync must NOT re-fetch full history, got: {}", + ranges[1] + ); + } + + #[tokio::test] + async fn a_uidvalidity_change_forces_a_full_resync_instead_of_skipping_mail() { + // A mailbox renumber changes UIDVALIDITY and restarts UIDs from low + // numbers. Resuming from the old UID boundary (`105:*`) would return + // only the single highest message and silently drop the rest, so a + // UIDVALIDITY mismatch MUST force a full `1:*` re-fetch. The scripted + // server reports UIDVALIDITY 1 on the first SELECT and 2 on the second. + let (client_io, server_io) = tokio::io::duplex(8192); + let fetch_ranges = Arc::new(std::sync::Mutex::new(Vec::::new())); + let server_ranges = fetch_ranges.clone(); + + let server = tokio::spawn(async move { + let mut io = server_io; + let mut select_count = 0u32; + 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("UID FETCH") { + if let Ok(mut g) = server_ranges.lock() { + g.push(line.clone()); + } + let _ = io + .write_all(format!("{tag} OK FETCH completed\r\n").as_bytes()) + .await; + } else if upper.contains("LOGIN") { + let _ = io + .write_all(format!("{tag} OK Logged in\r\n").as_bytes()) + .await; + } else if upper.contains("SELECT") { + select_count += 1; + // First SELECT: UIDVALIDITY 1. Second: a DIFFERENT + // UIDVALIDITY (2), i.e. the mailbox was renumbered. + let uidvalidity = if select_count == 1 { 1 } else { 2 }; + let resp = format!( + "* FLAGS (\\Seen)\r\n* 3 EXISTS\r\n* 0 RECENT\r\n\ + * OK [UIDVALIDITY {uidvalidity}] ok\r\n* OK [UIDNEXT 105] ok\r\n\ + {tag} OK [READ-WRITE] SELECT done\r\n" + ); + let _ = io.write_all(resp.as_bytes()).await; + } else if upper.contains("LOGOUT") { + let _ = io + .write_all(format!("* BYE\r\n{tag} OK LOGOUT\r\n").as_bytes()) + .await; + break; + } + } + }); + + let client = async_imap::Client::new(client_io); + let mut session = client + .login("user", "pass") + .await + .map_err(|(e, _)| e) + .expect("scripted login succeeds"); + + let engine = ImapEngine::new("acct-1", "example.test", 993); + + // First sync under UIDVALIDITY 1 stores "1:105". + let (_first, checkpoint) = engine + .sync_folder_messages_with_session("INBOX", None, &mut session) + .await + .expect("first sync succeeds"); + assert_eq!(checkpoint, "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 + .sync_folder_messages_with_session("INBOX", Some(&checkpoint), &mut session) + .await + .expect("second sync succeeds"); + assert_eq!( + checkpoint2, "2:105", + "checkpoint must be rewritten under the new UIDVALIDITY" + ); + + drop(session); + let _ = server.await; + + let ranges = fetch_ranges.lock().expect("lock captured ranges"); + assert_eq!(ranges.len(), 2, "expected exactly two UID FETCH commands"); + assert!( + ranges[0].contains("1:*"), + "first sync fetches full history, got: {}", + ranges[0] + ); + assert!( + ranges[1].contains("1:*") && !ranges[1].contains("105:*"), + "a UIDVALIDITY change MUST force a full re-fetch (1:*), never 105:*, got: {}", + ranges[1] + ); + } + + #[tokio::test(start_paused = true)] + async fn a_never_responding_fetch_surfaces_fetchstalled_without_hanging() { + // The scripted server answers LOGIN and SELECT but then goes silent on + // the FETCH. With the tokio clock paused, the bounded per-item timeout + // fast-forwards, so the stall is turned into an honest `FetchStalled` + // error in negligible WALL-CLOCK time -- proving it never hangs. + let (client_io, server_io) = tokio::io::duplex(8192); + + let server = tokio::spawn(async move { + let mut io = server_io; + 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("UID FETCH") { + // Deliberately never respond: a mid-stream server stall. + std::future::pending::<()>().await; + } else if upper.contains("LOGIN") { + let _ = io + .write_all(format!("{tag} OK Logged in\r\n").as_bytes()) + .await; + } else if upper.contains("SELECT") { + let resp = format!( + "* 1 EXISTS\r\n* OK [UIDVALIDITY 1] ok\r\n* OK [UIDNEXT 42] ok\r\n\ + {tag} OK [READ-WRITE] SELECT done\r\n" + ); + let _ = io.write_all(resp.as_bytes()).await; + } + } + }); + + let client = async_imap::Client::new(client_io); + let mut session = client + .login("user", "pass") + .await + .map_err(|(e, _)| e) + .expect("scripted login succeeds"); + let engine = ImapEngine::new("acct-1", "example.test", 993); + + // Real wall clock, not the (paused) tokio clock: proves negligible + // elapsed time despite the 60s virtual timeout expiring. + let started = std::time::Instant::now(); + let err = engine + .sync_folder_messages_with_session("INBOX", None, &mut session) + .await + .expect_err("a never-responding FETCH must surface an error"); + assert!( + matches!(err, MailError::FetchStalled(_)), + "expected FetchStalled, got: {err:?}" + ); + assert!( + started.elapsed() < Duration::from_secs(5), + "stalled fetch must not hang in wall-clock time, took {:?}", + started.elapsed() + ); + + server.abort(); + } + #[tokio::test] async fn with_tls_mode_records_mode_and_resolves_default_ports() { // A `0` port resolves to the conventional default for each mode. diff --git a/crates/nuncio-mail/src/mock.rs b/crates/nuncio-mail/src/mock.rs index 8c2ec27..d2352fe 100644 --- a/crates/nuncio-mail/src/mock.rs +++ b/crates/nuncio-mail/src/mock.rs @@ -14,12 +14,41 @@ pub struct MockMailBackend { messages: Arc>>, sent_messages: Arc>>, should_fail: Arc>, + /// Every `since_state` argument this mock's `sync_messages` was called + /// with, in call order, so tests can prove a caller threads the returned + /// checkpoint back into the next sync (real incrementality). + since_state_calls: Arc>>>, + /// The checkpoint token this mock returns from `sync_messages`. + returned_state: Arc>, } impl MockMailBackend { /// Create a new `MockMailBackend` with empty storage. pub fn new() -> Self { - Self::default() + let backend = Self::default(); + if let Ok(mut state) = backend.returned_state.lock() { + *state = "mock-state-token-100".to_string(); + } + backend + } + + /// Override the checkpoint token this mock returns from `sync_messages`, + /// so a test can control the exact state a caller must persist and thread + /// back on the next sync. + pub fn set_returned_state(&self, state: &str) { + if let Ok(mut guard) = self.returned_state.lock() { + *guard = state.to_string(); + } + } + + /// Every `since_state` argument passed to `sync_messages`, in call order. + /// A caller that genuinely resumes from a checkpoint will show `None` on + /// the first call and the previously-returned state on the next. + pub fn since_state_calls(&self) -> Vec> { + self.since_state_calls + .lock() + .map(|guard| guard.clone()) + .unwrap_or_default() } /// Configure the mock to simulate network failure errors. @@ -74,8 +103,14 @@ impl MailBackend for MockMailBackend { async fn sync_messages( &self, folder_id: &str, - _since_state: Option<&str>, + since_state: Option<&str>, ) -> Result<(Vec, String), MailError> { + // 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() { + calls.push(since_state.map(str::to_string)); + } + let should_fail = self .should_fail .lock() @@ -85,6 +120,19 @@ impl MailBackend for MockMailBackend { "simulated network failure".to_string(), )); } + let returned_state = self + .returned_state + .lock() + .map_err(|e| MailError::ParseFailed(e.to_string()))? + .clone(); + + // Simulate a since_state-aware backend: with a prior checkpoint there + // 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)); + } + let messages = self .messages .lock() @@ -94,7 +142,7 @@ impl MailBackend for MockMailBackend { .filter(|m| m.folder_id == folder_id) .cloned() .collect(); - Ok((matches, "mock-state-token-100".to_string())) + Ok((matches, returned_state)) } async fn send_email(&self, email: &Email) -> Result<(), MailError> { diff --git a/crates/nuncio-mail/src/parser.rs b/crates/nuncio-mail/src/parser.rs index 3f7c896..28976b9 100644 --- a/crates/nuncio-mail/src/parser.rs +++ b/crates/nuncio-mail/src/parser.rs @@ -32,6 +32,14 @@ pub enum MailError { #[error("IMAP error: {0}")] ImapError(String), + /// A single FETCH response item exceeded its bounded read timeout without + /// the server responding. Distinct from a generic [`MailError::ImapError`] + /// so callers can tell an honest mid-stream stall (dead socket, server-side + /// hang) apart from a protocol/parse failure instead of the sync hanging + /// forever. + #[error("IMAP FETCH stalled: {0}")] + FetchStalled(String), + /// Network or IO error. #[error("network I/O error: {0}")] NetworkError(String), diff --git a/crates/nuncio-store/src/db.rs b/crates/nuncio-store/src/db.rs index 30cc2f8..5f43173 100644 --- a/crates/nuncio-store/src/db.rs +++ b/crates/nuncio-store/src/db.rs @@ -649,6 +649,19 @@ impl DatabaseEngine { INSERT INTO contacts_fts(id, display_name, organization, emails_json) VALUES (new.id, new.display_name, COALESCE(new.organization, ''), new.emails_json); END; + + -- Per-folder inbound-sync checkpoint. `state` is an opaque, + -- protocol-defined resumption token (an IMAP UID boundary such as + -- UIDNEXT, a JMAP state string) that a backend hands back after a + -- sync and expects to receive on the next sync so it can fetch only + -- what changed since. Keyed by (account, folder) because the token + -- is meaningful only within a single mailbox on a single account. + CREATE TABLE IF NOT EXISTS folder_sync_state ( + account_id TEXT NOT NULL, + folder_id TEXT NOT NULL, + state TEXT NOT NULL, + PRIMARY KEY (account_id, folder_id) + ); "#, ) .execute(&self.pool) @@ -915,6 +928,51 @@ impl DatabaseEngine { .collect()) } + /// Fetch the persisted inbound-sync checkpoint for a folder, or `None` if + /// this account/folder pair has never completed a sync. A `None` result is + /// the signal to the backend that it must perform a full first-time fetch. + pub async fn get_folder_sync_state( + &self, + account_id: &str, + folder_id: &str, + ) -> Result, DatabaseError> { + let row: Option<(String,)> = sqlx::query_as( + "SELECT state FROM folder_sync_state WHERE account_id = ? AND folder_id = ?", + ) + .bind(account_id) + .bind(folder_id) + .fetch_optional(&self.pool) + .await + .map_err(DatabaseError::Query)?; + Ok(row.map(|(state,)| state)) + } + + /// Persist the inbound-sync checkpoint a backend returned for a folder, + /// overwriting any prior value (a checkpoint is a running high-water mark, + /// not history). Written only after a folder's messages have been fetched + /// and persisted, so the stored checkpoint can never advance past work that + /// actually landed in the store. + pub async fn save_folder_sync_state( + &self, + account_id: &str, + folder_id: &str, + state: &str, + ) -> Result<(), DatabaseError> { + sqlx::query( + r#" + INSERT OR REPLACE INTO folder_sync_state (account_id, folder_id, state) + VALUES (?, ?, ?) + "#, + ) + .bind(account_id) + .bind(folder_id) + .bind(state) + .execute(&self.pool) + .await + .map_err(DatabaseError::Query)?; + Ok(()) + } + /// Save an [`nuncio_core::model::Email`] to SQLite (INSERT OR REPLACE). /// /// The message body is encrypted (AES-256-GCM) before being written to the `messages` @@ -2696,6 +2754,61 @@ mod tests { .expect("second delete is a no-op"); } + #[tokio::test] + async fn folder_sync_state_round_trip_is_scoped_and_overwrites() { + let (engine, _dir) = DatabaseEngine::connect_ephemeral().await.unwrap(); + + // Absent before any sync completes. + assert_eq!( + engine + .get_folder_sync_state("acct-1", "INBOX") + .await + .expect("get succeeds"), + None + ); + + engine + .save_folder_sync_state("acct-1", "INBOX", "105") + .await + .expect("save succeeds"); + assert_eq!( + engine + .get_folder_sync_state("acct-1", "INBOX") + .await + .expect("get succeeds"), + Some("105".to_string()) + ); + + // A later checkpoint overwrites (high-water mark, not history). + engine + .save_folder_sync_state("acct-1", "INBOX", "220") + .await + .expect("save succeeds"); + assert_eq!( + engine + .get_folder_sync_state("acct-1", "INBOX") + .await + .expect("get succeeds"), + Some("220".to_string()) + ); + + // Keyed by (account, folder): a different folder/account is unaffected. + assert_eq!( + engine + .get_folder_sync_state("acct-1", "Sent") + .await + .expect("get succeeds"), + None + ); + assert_eq!( + engine + .get_folder_sync_state("acct-2", "INBOX") + .await + .expect("get succeeds"), + None + ); + } + /// Proves the additive `smtp_host`/`smtp_port` /// migration is backfill-safe. Simulates a database created BEFORE this /// feature existed (an `accounts` table with no `smtp_host`/`smtp_port` diff --git a/crates/nunciod/src/grpc.rs b/crates/nunciod/src/grpc.rs index e2e189e..97083d4 100644 --- a/crates/nunciod/src/grpc.rs +++ b/crates/nunciod/src/grpc.rs @@ -87,15 +87,17 @@ pub enum GrpcServeError { } /// Maps a `nuncio_core::CoreEvent` domain event onto its wire-format -/// `nuncio.v1.Event` representation, faithfully carrying every variant's +/// `nuncio.v1.Event` representation, carrying every wire-visible variant's /// fields into the matching `oneof` case (see `proto/nuncio/v1/nuncio.proto` -/// for the wire contract). +/// for the wire contract). Returns `None` for events that are internal-only +/// and have no case in the versioned contract, so they are dropped from the +/// gRPC subscription stream rather than forced onto the wire. /// /// `usize` fields (`processed`, `total`, `matched`, `salvaged_rules_count`) /// are narrowed to `u64` for the wire; these are in-process counters that /// cannot realistically approach `u64::MAX`, and protobuf has no native /// `usize` type. -fn map_core_event(event: CoreEvent) -> Event { +fn map_core_event(event: CoreEvent) -> Option { let kind = match event { CoreEvent::SyncStarted { account_id } => Kind::SyncStarted(SyncStarted { account_id }), CoreEvent::SyncCompleted { account_id } => { @@ -140,8 +142,12 @@ fn map_core_event(event: CoreEvent) -> Event { }), CoreEvent::Error { message } => Kind::Error(EventError { message }), CoreEvent::ShuttingDown => Kind::ShuttingDown(ShuttingDown {}), + // Internal sync-progress signal: intentionally not forwarded to gRPC + // subscribers, as it has no case in the versioned wire contract. It is + // observed in-process (e.g. by the daemon's own subscribers) only. + CoreEvent::SyncProgress { .. } => return None, }; - Event { kind: Some(kind) } + Some(Event { kind: Some(kind) }) } /// `nuncio.v1.System` gRPC service implementation backed by the daemon's live @@ -191,7 +197,7 @@ impl System for SystemGrpcService { ) -> Result, Status> { let receiver = self.event_bus.subscribe_events(); let mapped = BroadcastStream::new(receiver).filter_map(|item| match item { - Ok(core_event) => Some(Ok(map_core_event(core_event))), + Ok(core_event) => map_core_event(core_event).map(Ok), Err(BroadcastStreamRecvError::Lagged(skipped)) => { tracing::warn!( "gRPC Subscribe stream lagged behind the daemon event bus; skipped {} \ diff --git a/crates/nunciod/src/sync.rs b/crates/nunciod/src/sync.rs index d81c0aa..a3c4522 100644 --- a/crates/nunciod/src/sync.rs +++ b/crates/nunciod/src/sync.rs @@ -18,7 +18,7 @@ //! only places that resolve a real account's credentials and construct a //! real backend; they then delegate the actual fetch/persist/event-emission //! work to [`sync_with_backend`]. -use nuncio_core::{AccountConfig, AccountProtocol, CoreCommand, EventBus}; +use nuncio_core::{AccountConfig, AccountProtocol, CoreCommand, CoreEvent, EventBus}; use nuncio_filter::{FilterEngine, OutboxManager, RuleAction}; use nuncio_mail::{ImapEngine, JmapEngine, MailBackend, MailError}; use nuncio_store::db::DatabaseError; @@ -235,13 +235,27 @@ async fn is_new_message(db: &nuncio_store::db::DatabaseEngine, message_id: &str) /// [`MockMailBackend`]. async fn fetch_and_persist( db: &nuncio_store::db::DatabaseEngine, + event_bus: &EventBus, backend: &dyn MailBackend, filter_engine: &FilterEngine, + account_id: Option<&str>, ) -> Result { let folders = backend.sync_folders().await?; let mut synced = 0usize; for folder in folders { - let (emails, _state) = backend.sync_messages(&folder.id, None).await?; + // Resume from the folder's stored checkpoint so the backend can fetch + // only what changed since the last sync. A checkpoint is keyed by + // (account, folder), so a run with no account context (a mock-driven + // `SyncAll` with `account_id == None`) has nowhere to key it and always + // performs a full fetch. + let last_state = match account_id { + 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()) + .await?; + let fetched = emails.len(); for email in emails { let is_new = is_new_message(db, &email.id).await; db.save_email(&email).await?; @@ -250,6 +264,19 @@ async fn fetch_and_persist( } synced += 1; } + // 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. + if let Some(acct) = account_id { + db.save_folder_sync_state(acct, &folder.id, &new_state) + .await?; + } + event_bus.publish_event(CoreEvent::SyncProgress { + account_id: account_id.map(str::to_string), + folder_id: folder.id.clone(), + fetched, + total: synced, + }); } Ok(synced) } @@ -282,7 +309,8 @@ pub async fn sync_with_backend( None => event_bus.process_command(CoreCommand::SyncAll), } - let result = fetch_and_persist(db, backend, filter_engine).await; + let result = + fetch_and_persist(db, event_bus, backend, filter_engine, account_id.as_deref()).await; if let Err(e) = &result { event_bus.process_command(CoreCommand::ReportError { @@ -394,7 +422,15 @@ pub async fn run_all_accounts_sync( match secrets.get_secret(&config.keyring_secret_key) { Ok(password) => { let backend = build_mail_backend(&config, &password); - match fetch_and_persist(db, backend.as_ref(), filter_engine).await { + match fetch_and_persist( + db, + event_bus, + backend.as_ref(), + filter_engine, + Some(&config.id), + ) + .await + { Ok(count) => total_synced += count, Err(e) => { event_bus.process_command(CoreCommand::ReportError { @@ -619,6 +655,16 @@ mod tests { account_id: Some("acct-mock-1".to_string()) } ); + // A per-folder progress event is published between start and complete. + assert_eq!( + events.recv().await.expect("progress event"), + CoreEvent::SyncProgress { + account_id: Some("acct-mock-1".to_string()), + folder_id: "inbox".to_string(), + fetched: 2, + total: 2, + } + ); assert_eq!( events.recv().await.expect("complete event"), CoreEvent::SyncCompleted { @@ -654,6 +700,16 @@ mod tests { events.recv().await.expect("start event"), CoreEvent::SyncStarted { account_id: None } ); + // The folder is empty, but a progress event still marks it done. + assert_eq!( + events.recv().await.expect("progress event"), + CoreEvent::SyncProgress { + account_id: None, + folder_id: "inbox".to_string(), + fetched: 0, + total: 0, + } + ); assert_eq!( events.recv().await.expect("complete event"), CoreEvent::SyncCompleted { account_id: None } @@ -791,6 +847,15 @@ mod tests { account_id: Some("acct-jmap-real-1".to_string()) } ); + assert_eq!( + events.recv().await.expect("progress event"), + CoreEvent::SyncProgress { + account_id: Some("acct-jmap-real-1".to_string()), + folder_id: "jmap-inbox".to_string(), + fetched: 1, + total: 1, + } + ); assert_eq!( events.recv().await.expect("complete event"), CoreEvent::SyncCompleted { @@ -1202,4 +1267,69 @@ mod tests { "no new outbox mutation should be enqueued on a repeat sync" ); } + + #[tokio::test] + async fn sync_with_backend_persists_and_threads_the_folder_checkpoint_across_syncs() { + // The regression this story exists to kill: the returned checkpoint + // must be persisted and fed back into the NEXT sync, so the second + // sync is genuinely incremental rather than a full re-fetch. Proven by + // capturing the exact `since_state` the backend received on each call. + let (db, _dir) = DatabaseEngine::connect_ephemeral() + .await + .expect("ephemeral db"); + let event_bus = EventBus::new(); + + let mock = MockMailBackend::new(); + mock.set_returned_state("uidnext-4242"); + mock.add_folder(Folder { + id: "inbox".to_string(), + name: "Inbox".to_string(), + total_messages: 1, + unread_messages: 1, + }); + mock.add_message(mock_email("m-inc-1", "inbox", "First")); + + let filter_engine = empty_filter_engine(); + + // First sync: no checkpoint exists yet, so the backend is called with + // `None` and does a full fetch (one message). + let first = sync_with_backend( + &db, + &event_bus, + &mock, + &filter_engine, + Some("acct-inc-1".to_string()), + ) + .await + .expect("first sync succeeds"); + assert_eq!(first, 1); + + // The returned checkpoint must now be persisted for (account, folder). + assert_eq!( + db.get_folder_sync_state("acct-inc-1", "inbox") + .await + .expect("read checkpoint"), + Some("uidnext-4242".to_string()) + ); + + // Second sync: the daemon must feed the stored checkpoint back in. The + // mock returns nothing new for a non-None checkpoint -> narrower fetch. + let second = sync_with_backend( + &db, + &event_bus, + &mock, + &filter_engine, + Some("acct-inc-1".to_string()), + ) + .await + .expect("second sync succeeds"); + assert_eq!(second, 0, "an incremental second sync fetches nothing new"); + + // The proof of genuine incrementality: the backend saw `None` first, + // then the exact checkpoint the first sync returned. + assert_eq!( + mock.since_state_calls(), + vec![None, Some("uidnext-4242".to_string())], + ); + } }