diff --git a/crates/nuncio-mail/src/backend.rs b/crates/nuncio-mail/src/backend.rs index c776170..f25e9bf 100644 --- a/crates/nuncio-mail/src/backend.rs +++ b/crates/nuncio-mail/src/backend.rs @@ -91,6 +91,50 @@ pub struct FolderChanges { pub next_state: String, } +/// What a mutation attempt actually established. +/// +/// Two states are not enough. A protocol that answers `OK` for a command that +/// did nothing -- which IMAP does, by design, for a UID that no longer exists +/// (RFC 3501 section 6.4.8) -- makes "succeeded" and "failed" an incomplete +/// partition. The missing third state is *the server accepted the command and +/// its response does not establish what happened*, and collapsing that into +/// either neighbour is how a lost mutation gets recorded as done. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum MutationOutcome { + /// The server's response proves the change was made. + /// + /// Requires positive evidence: a `COPYUID` naming the destination UIDs, a + /// removal report for a delete, or an explicit acknowledgement for a flag + /// change. `token` carries whatever the server returned to address the + /// result afterwards (the destination UID for a move), when it returned + /// one at all -- a DAV server that rewrites the object MUST NOT return an + /// ETag (RFC 4791 section 5.3.4), so absence here is ordinary. + Applied { + /// Server-assigned handle for the result, when the response carried one. + token: Option, + }, + /// Another client changed the message first, and the server said so. + /// + /// Requires positive evidence too: `[MODIFIED …]` (RFC 7162 section 3.1.3) + /// or an HTTP 412. A mutation that merely failed to find its target is + /// **not** a conflict -- `MODIFIED` reports "changed by someone else", + /// never "gone". + Conflict { + /// What the server reported about the conflicting state. + observed: String, + }, + /// The command was accepted and its response settles nothing. + /// + /// The honest reading of an absent `COPYUID` (only `SHOULD` for `MOVE` per + /// RFC 6851 section 4.3, and legitimately omitted for a `UIDNOTSTICKY` + /// destination), or of a UID set that matched nothing. The caller must + /// re-enumerate to find out, not guess. + Unknown { + /// Why the response was inconclusive. + reason: String, + }, +} + /// Protocol-agnostic mail backend engine trait implemented by JMAP and IMAP engines. #[async_trait] pub trait MailBackend: Send + Sync { @@ -124,11 +168,15 @@ pub trait MailBackend: Send + Sync { } /// Apply a remote mutation (move/copy/flag/delete) to a single message - /// against the real server. MUST return `Ok(())` only when the server - /// genuinely applied the change -- implementations must never fabricate - /// success, and must refuse to act (returning an error) when the target - /// message cannot be safely identified (e.g. an IMAP UIDVALIDITY mismatch). - async fn apply_mutation(&self, spec: &RemoteMutationSpec) -> Result<(), MailError>; + /// against the real server. + /// + /// Returns what the server's response actually proves. Implementations + /// must never report [`MutationOutcome::Applied`] on the strength of a + /// tagged `OK` alone, and must refuse to act (returning `Err`) when the + /// target cannot be safely identified -- e.g. an IMAP UIDVALIDITY + /// mismatch, where the stored UID now names a different message. + async fn apply_mutation(&self, spec: &RemoteMutationSpec) + -> Result; } /// A composed outbound email message ready to send over SMTP. Deliberately diff --git a/crates/nuncio-mail/src/imap.rs b/crates/nuncio-mail/src/imap.rs index 6e883bc..ac273d7 100644 --- a/crates/nuncio-mail/src/imap.rs +++ b/crates/nuncio-mail/src/imap.rs @@ -52,7 +52,7 @@ const MAX_MESSAGES_PER_FETCH_BATCH: usize = 200; /// would ever accept is pointless to transfer only to reject. const MAX_MESSAGE_BODY_BYTES: u32 = 25 * 1024 * 1024; -use crate::backend::{MailBackend, RemoteMutationKind, RemoteMutationSpec}; +use crate::backend::{MailBackend, MutationOutcome, RemoteMutationKind, RemoteMutationSpec}; use crate::parser::{MailError, MimeParserAdapter}; /// Recover a raw IMAP UID from a [`RemoteMutationSpec::remote_id`] (a decimal @@ -102,6 +102,34 @@ const CHECKPOINT_DELIM: char = ':'; /// full re-fetch. Over-fetching is recoverable; misreading a token is not. const CHECKPOINT_SCHEME_UIDNEXT: &str = "v1:uidnext"; +/// Read a `COPY`/`MOVE` outcome from whether the server proved it. +/// +/// `COPYUID` present is proof and names the destination UIDs. Absent is +/// **unknown, never failure**: RFC 6851 section 4.3 makes it only `SHOULD` for +/// `MOVE`, RFC 4315 section 3 permits omitting it for a `UIDNOTSTICKY` +/// destination or one the client cannot `SELECT`, and a UID set matching +/// nothing succeeds having done nothing (RFC 3501 section 6.4.8). Those are +/// indistinguishable from here, and the caller has to re-enumerate to tell. +fn copy_outcome(copy_uid: Option, op: &str) -> MutationOutcome { + match copy_uid { + Some(proof) if !proof.destination_uids.is_empty() => MutationOutcome::Applied { + token: Some( + proof + .destination_uids + .iter() + .map(u32::to_string) + .collect::>() + .join(","), + ), + }, + _ => MutationOutcome::Unknown { + reason: format!( + "the server accepted the UID {op} without a COPYUID, so nothing proves it moved" + ), + }, + } +} + /// Scheme for a mod-sequence checkpoint: `"v2:modseq:{uidvalidity}:{modseq}"`. /// /// Written by both mod-sequence rungs. QRESYNC and CONDSTORE differ in how @@ -1343,7 +1371,7 @@ impl ImapEngine { &self, spec: &RemoteMutationSpec, session: &mut async_imap::Session, - ) -> Result<(), MailError> + ) -> Result where S: AsyncRead + AsyncWrite + Unpin + Send + Debug + 'static, { @@ -1411,54 +1439,82 @@ impl ImapEngine { match &spec.kind { RemoteMutationKind::SetFlagged { value } => { let op = if *value { "+FLAGS" } else { "-FLAGS" }; - self.uid_store_flags(session, &uid_set, op, "\\Flagged") - .await + let exchange = crate::imap_raw::run_collected( + session, + &format!("UID STORE {uid_set} {op} (\\Flagged)"), + ) + .await?; + if let Some(conflicting) = exchange.modified_uids() { + return Ok(MutationOutcome::Conflict { + observed: format!("server reported MODIFIED for uid(s) {conflicting:?}"), + }); + } + if !exchange.ok { + return Err(MailError::ImapError(format!( + "UID STORE {op} for uid {uid_set} failed: {}", + exchange.information.as_deref().unwrap_or("no detail") + ))); + } + // An untagged FETCH echoing the new flags is proof. Its absence + // is not failure: RFC 3501 section 6.4.6 lets a server suppress + // it, and a store that changes nothing legitimately produces + // none -- so the honest answer is that nothing was established. + if exchange.fetched_uids.is_empty() { + Ok(MutationOutcome::Unknown { + reason: "the server acknowledged the store without echoing the message" + .to_string(), + }) + } else { + Ok(MutationOutcome::Applied { token: None }) + } } RemoteMutationKind::Copy { to_folder } => { - session.uid_copy(&uid_set, to_folder).await.map_err(|e| { - MailError::ImapError(format!( + let exchange = crate::imap_raw::run_collected( + session, + &format!("UID COPY {uid_set} {to_folder}"), + ) + .await?; + if !exchange.ok { + return Err(MailError::ImapError(format!( "UID COPY of '{}' to '{}' failed: {}", - spec.message_id, to_folder, e - )) - }) + spec.message_id, + to_folder, + exchange.information.as_deref().unwrap_or("no detail") + ))); + } + Ok(copy_outcome(exchange.copy_uid, "COPY")) } RemoteMutationKind::Move { to_folder } => { self.uid_move(session, &uid_set, to_folder, &spec.message_id) .await } RemoteMutationKind::Delete => { - self.uid_store_flags(session, &uid_set, "+FLAGS", "\\Deleted") - .await?; - self.uid_expunge_scoped(session, &uid_set).await + let store = crate::imap_raw::run_collected( + session, + &format!("UID STORE {uid_set} +FLAGS (\\Deleted)"), + ) + .await?; + if !store.ok { + return Err(MailError::ImapError(format!( + "UID STORE +FLAGS (\\Deleted) for uid {uid_set} failed: {}", + store.information.as_deref().unwrap_or("no detail") + ))); + } + let expunged = self.uid_expunge_scoped(session, &uid_set).await?; + // Under QRESYNC the removal is reported as VANISHED rather than + // EXPUNGE (RFC 6851 section 4.4), so counting only EXPUNGE + // would report failure on every successful delete. + if expunged { + Ok(MutationOutcome::Applied { token: None }) + } else { + Ok(MutationOutcome::Unknown { + reason: "the server reported no removal for the expunged uid".to_string(), + }) + } } } } - /// Issue a `UID STORE {uid} {op} ({flag})` and drain its untagged `FETCH` - /// responses, surfacing any protocol error. `op` is `+FLAGS`/`-FLAGS`. - async fn uid_store_flags( - &self, - session: &mut async_imap::Session, - uid_set: &str, - op: &str, - flag: &str, - ) -> Result<(), MailError> - where - S: AsyncRead + AsyncWrite + Unpin + Send + Debug + 'static, - { - let query = format!("{op} ({flag})"); - let updates = session.uid_store(uid_set, &query).await.map_err(|e| { - MailError::ImapError(format!("UID STORE {query} for uid {uid_set} failed: {e}")) - })?; - tokio::pin!(updates); - while let Some(item) = updates.next().await { - item.map_err(|e| { - MailError::ImapError(format!("reading UID STORE response failed: {e}")) - })?; - } - Ok(()) - } - /// Permanently remove ONLY the target `uid_set` from the selected folder /// via RFC 4315 `UID EXPUNGE`, draining its untagged `EXPUNGE` responses. /// @@ -1473,7 +1529,7 @@ impl ImapEngine { &self, session: &mut async_imap::Session, uid_set: &str, - ) -> Result<(), MailError> + ) -> Result where S: AsyncRead + AsyncWrite + Unpin + Send + Debug + 'static, { @@ -1491,17 +1547,19 @@ impl ImapEngine { ))); } - let expunged = session - .uid_expunge(uid_set) - .await - .map_err(|e| MailError::ImapError(format!("UID EXPUNGE {uid_set} failed: {e}")))?; - tokio::pin!(expunged); - while let Some(item) = expunged.next().await { - item.map_err(|e| { - MailError::ImapError(format!("reading UID EXPUNGE response failed: {e}")) - })?; + // Collected rather than streamed, because a QRESYNC-enabled server + // answers with VANISHED instead of EXPUNGE (RFC 6851 section 4.4) and + // the typed expunge stream only yields the latter. Reading just that + // stream reports "nothing removed" on every successful delete. + let exchange = + crate::imap_raw::run_collected(session, &format!("UID EXPUNGE {uid_set}")).await?; + if !exchange.ok { + return Err(MailError::ImapError(format!( + "UID EXPUNGE {uid_set} failed: {}", + exchange.information.as_deref().unwrap_or("no detail") + ))); } - Ok(()) + Ok(!exchange.vanished_uids.is_empty() || !exchange.expunged_seqs.is_empty()) } /// Move a message by UID, preferring RFC 6851 `UID MOVE` and falling back @@ -1515,7 +1573,7 @@ impl ImapEngine { uid_set: &str, to_folder: &str, message_id: &str, - ) -> Result<(), MailError> + ) -> Result where S: AsyncRead + AsyncWrite + Unpin + Send + Debug + 'static, { @@ -1527,27 +1585,53 @@ impl ImapEngine { drop(caps); if has_move { - return session.uid_mv(uid_set, to_folder).await.map_err(|e| { - MailError::ImapError(format!( - "UID MOVE of '{message_id}' to '{to_folder}' failed: {e}" - )) - }); + let exchange = + crate::imap_raw::run_collected(session, &format!("UID MOVE {uid_set} {to_folder}")) + .await?; + if !exchange.ok { + return Err(MailError::ImapError(format!( + "UID MOVE of '{message_id}' to '{to_folder}' failed: {}", + exchange.information.as_deref().unwrap_or("no detail") + ))); + } + return Ok(copy_outcome(exchange.copy_uid, "MOVE")); } - session.uid_copy(uid_set, to_folder).await.map_err(|e| { - MailError::ImapError(format!( - "UID COPY (move fallback) of '{message_id}' to '{to_folder}' failed: {e}" - )) - })?; - self.uid_store_flags(session, uid_set, "+FLAGS", "\\Deleted") - .await?; - self.uid_expunge_scoped(session, uid_set).await + let copied = + crate::imap_raw::run_collected(session, &format!("UID COPY {uid_set} {to_folder}")) + .await?; + if !copied.ok { + return Err(MailError::ImapError(format!( + "UID COPY (move fallback) of '{message_id}' to '{to_folder}' failed: {}", + copied.information.as_deref().unwrap_or("no detail") + ))); + } + // The copy half is what the destination proof refers to; the source + // removal below is separately verified, and a failure there leaves a + // duplicate rather than a lost message. + let outcome = copy_outcome(copied.copy_uid, "COPY"); + let stored = crate::imap_raw::run_collected( + session, + &format!("UID STORE {uid_set} +FLAGS (\\Deleted)"), + ) + .await?; + if !stored.ok { + return Err(MailError::ImapError(format!( + "UID STORE +FLAGS (\\Deleted) during move fallback failed: {}", + stored.information.as_deref().unwrap_or("no detail") + ))); + } + self.uid_expunge_scoped(session, uid_set).await?; + Ok(outcome) } /// Apply a remote mutation over a freshly authenticated session. See /// [`Self::apply_mutation_with_session`] for the UIDVALIDITY guard and the /// per-action protocol commands. - pub async fn apply_remote_mutation(&self, spec: &RemoteMutationSpec) -> Result<(), MailError> { + pub async fn apply_remote_mutation( + &self, + spec: &RemoteMutationSpec, + ) -> Result { let (username, password) = match (&self.username, &self.password) { (Some(u), Some(p)) => (u.as_str(), p.as_str()), _ => { @@ -1713,7 +1797,10 @@ impl MailBackend for ImapEngine { self.sync_folder_messages(folder_id, since_state).await } - async fn apply_mutation(&self, spec: &RemoteMutationSpec) -> Result<(), MailError> { + async fn apply_mutation( + &self, + spec: &RemoteMutationSpec, + ) -> Result { self.apply_remote_mutation(spec).await } } @@ -2612,7 +2699,7 @@ mod tests { advertise_move: bool, advertise_uidplus: bool, select_uid_validity: u32, - ) -> (Result<(), MailError>, Vec) { + ) -> (Result, Vec) { let (client_io, server_io) = tokio::io::duplex(8192); let commands = Arc::new(std::sync::Mutex::new(Vec::::new())); let server_commands = commands.clone(); diff --git a/crates/nuncio-mail/src/jmap.rs b/crates/nuncio-mail/src/jmap.rs index 7d63652..d09d12f 100644 --- a/crates/nuncio-mail/src/jmap.rs +++ b/crates/nuncio-mail/src/jmap.rs @@ -5,7 +5,7 @@ use nuncio_core::model::{Email, Folder}; use serde::{Deserialize, Serialize}; use serde_json::{json, Value}; -use crate::backend::{MailBackend, RemoteMutationKind, RemoteMutationSpec}; +use crate::backend::{MailBackend, MutationOutcome, RemoteMutationKind, RemoteMutationSpec}; use crate::parser::MailError; /// Stand-in for the UIDVALIDITY scope on JMAP, which has no such concept: a @@ -602,7 +602,10 @@ impl MailBackend for JmapEngine { } #[tracing::instrument(skip(self), fields(account_id = %self.account_id), err)] - async fn apply_mutation(&self, spec: &RemoteMutationSpec) -> Result<(), MailError> { + async fn apply_mutation( + &self, + spec: &RemoteMutationSpec, + ) -> Result { if !self.has_credentials() { return Err(MailError::AuthError( "JMAP remote mutation requires credentials".to_string(), @@ -639,7 +642,12 @@ impl MailBackend for JmapEngine { let request = Self::build_email_set_request(&account_id, spec); let raw = self.post_jmap(&session.api_url, &request).await?; let expect_destroy = matches!(spec.kind, RemoteMutationKind::Delete); - Self::confirm_email_set_applied(&raw, &spec.remote_id, expect_destroy) + // `Email/set` names the ids it updated or destroyed, so a confirmed + // response is genuine proof rather than a bare acknowledgement. JMAP + // has no COPYUID-style ambiguity here; the id is stable and the server + // either lists it or does not. + Self::confirm_email_set_applied(&raw, &spec.remote_id, expect_destroy)?; + Ok(MutationOutcome::Applied { token: None }) } } diff --git a/crates/nuncio-mail/src/lib.rs b/crates/nuncio-mail/src/lib.rs index fa913fe..9fc19ad 100644 --- a/crates/nuncio-mail/src/lib.rs +++ b/crates/nuncio-mail/src/lib.rs @@ -18,7 +18,8 @@ pub mod smtp; pub mod test_server; pub use backend::{ - MailBackend, MessageSender, OutboundMessage, RemoteMutationKind, RemoteMutationSpec, + FolderChanges, MailBackend, MessageSender, MutationOutcome, OutboundMessage, + RemoteMutationKind, RemoteMutationSpec, }; pub use imap::{IdleSocketState, ImapDualSocketManager, ImapEngine}; pub use jmap::JmapEngine; diff --git a/crates/nuncio-mail/src/mock.rs b/crates/nuncio-mail/src/mock.rs index adbf97b..f592192 100644 --- a/crates/nuncio-mail/src/mock.rs +++ b/crates/nuncio-mail/src/mock.rs @@ -4,7 +4,9 @@ use async_trait::async_trait; use nuncio_core::model::{Email, Folder}; use std::sync::{Arc, Mutex}; -use crate::backend::{MailBackend, MessageSender, OutboundMessage, RemoteMutationSpec}; +use crate::backend::{ + MailBackend, MessageSender, MutationOutcome, OutboundMessage, RemoteMutationSpec, +}; use crate::parser::MailError; /// Thread-safe mock mail backend for offline testing. @@ -162,7 +164,10 @@ impl MailBackend for MockMailBackend { }) } - async fn apply_mutation(&self, spec: &RemoteMutationSpec) -> Result<(), MailError> { + async fn apply_mutation( + &self, + spec: &RemoteMutationSpec, + ) -> Result { let should_fail = self .should_fail .lock() @@ -179,7 +184,7 @@ impl MailBackend for MockMailBackend { .lock() .map_err(|e| MailError::ParseFailed(e.to_string()))?; applied.push(spec.clone()); - Ok(()) + Ok(MutationOutcome::Applied { token: None }) } } diff --git a/crates/nuncio-mail/tests/imap_raw_wire_test.rs b/crates/nuncio-mail/tests/imap_raw_wire_test.rs index 9d03e60..42a67a7 100644 --- a/crates/nuncio-mail/tests/imap_raw_wire_test.rs +++ b/crates/nuncio-mail/tests/imap_raw_wire_test.rs @@ -478,3 +478,137 @@ async fn a_uidvalidity_change_refetches_rather_than_deleting_the_folder() { // part of identity -- so the caller replaces rather than merges. assert_ne!(second.upserts[0].id, first.upserts[0].id); } + +/// Build an engine pointed at the mock and apply one mutation through the +/// real backend path. +async fn apply( + server: &MockImapServer, + kind: nuncio_mail::RemoteMutationKind, + uid: u32, + uid_validity: &str, +) -> nuncio_mail::MutationOutcome { + use nuncio_mail::MailBackend; + let engine = nuncio_mail::ImapEngine::with_credentials( + "acct-1", + server.host(), + server.port(), + nuncio_core::TlsMode::Plain, + "user", + "pass", + ); + engine + .apply_mutation(&nuncio_mail::RemoteMutationSpec { + message_id: "m-1".to_string(), + remote_id: uid.to_string(), + folder_id: "INBOX".to_string(), + uid_validity: uid_validity.to_string(), + kind, + }) + .await + .expect("the mutation attempt itself succeeds") +} + +/// UIDVALIDITY the mock assigns to the first mailbox created. +const FIRST_UID_VALIDITY: &str = "1000"; + +#[tokio::test] +async fn a_move_that_lost_a_race_is_unknown_rather_than_applied() { + // The silent-loss case. The server answers OK for a UID that no longer + // exists (RFC 3501 6.4.8), so only the absence of COPYUID distinguishes a + // real move from a no-op -- and absence cannot mean "applied". + let server = MockImapServer::start(ServerProfile::Qresync) + .await + .expect("server starts"); + server.create_mailbox("INBOX"); + server.create_mailbox("Archive"); + server.create_mailbox("Trash"); + let uid = server.append_message("INBOX", "Subject: contested\r\n\r\nx", &[]); + + let applied = apply( + &server, + nuncio_mail::RemoteMutationKind::Move { + to_folder: "Archive".to_string(), + }, + uid, + FIRST_UID_VALIDITY, + ) + .await; + assert!( + matches!( + applied, + nuncio_mail::MutationOutcome::Applied { token: Some(_) } + ), + "a real move must be Applied and name its destination UID, got {applied:?}" + ); + + let lost = apply( + &server, + nuncio_mail::RemoteMutationKind::Move { + to_folder: "Trash".to_string(), + }, + uid, + FIRST_UID_VALIDITY, + ) + .await; + assert!( + matches!(lost, nuncio_mail::MutationOutcome::Unknown { .. }), + "a move of an already-moved UID must be Unknown, never Applied, got {lost:?}" + ); + assert!( + server.uids("Trash").is_empty(), + "and nothing actually reached the destination" + ); +} + +#[tokio::test] +async fn a_delete_is_applied_only_when_the_server_reports_the_removal() { + let server = MockImapServer::start(ServerProfile::Qresync) + .await + .expect("server starts"); + server.create_mailbox("INBOX"); + let uid = server.append_message("INBOX", "Subject: gone\r\n\r\nx", &[]); + + let outcome = apply( + &server, + nuncio_mail::RemoteMutationKind::Delete, + uid, + FIRST_UID_VALIDITY, + ) + .await; + + // Under QRESYNC the proof arrives as VANISHED, not EXPUNGE. Reading only + // the typed expunge stream would report nothing removed here. + assert!( + matches!(outcome, nuncio_mail::MutationOutcome::Applied { .. }), + "a delete the server confirmed must be Applied, got {outcome:?}" + ); + assert!(server.uids("INBOX").is_empty()); +} + +#[tokio::test] +async fn a_delete_of_an_absent_uid_is_unknown() { + let server = MockImapServer::start(ServerProfile::Qresync) + .await + .expect("server starts"); + server.create_mailbox("INBOX"); + server.append_message("INBOX", "Subject: present\r\n\r\nx", &[]); + + // UID 999 does not exist; the server ignores it and answers OK. + let outcome = apply( + &server, + nuncio_mail::RemoteMutationKind::Delete, + 999, + FIRST_UID_VALIDITY, + ) + .await; + + assert!( + matches!(outcome, nuncio_mail::MutationOutcome::Unknown { .. }), + "a delete the server never confirmed must not be Applied, got {outcome:?}" + ); + assert_eq!( + server.uids("INBOX").len(), + 1, + "the real message must be untouched" + ); +} diff --git a/crates/nunciod/src/outbox.rs b/crates/nunciod/src/outbox.rs index 5c00417..01a5b7e 100644 --- a/crates/nunciod/src/outbox.rs +++ b/crates/nunciod/src/outbox.rs @@ -36,8 +36,8 @@ use nuncio_filter::{ WebhookError, }; use nuncio_mail::{ - MailBackend, MailError, MessageSender, OutboundMessage, RemoteMutationKind, RemoteMutationSpec, - SmtpTransportEngine, + MailBackend, MailError, MessageSender, MutationOutcome, OutboundMessage, RemoteMutationKind, + RemoteMutationSpec, SmtpTransportEngine, }; use nuncio_store::db::{DatabaseEngine, DatabaseError}; use nuncio_store::vault::{SecretManager, VaultError, WEBHOOK_SIGNING_KEY_ACCOUNT}; @@ -129,6 +129,12 @@ pub trait RemoteExecutionEnv: Send + Sync { enum Disposition { /// The real operation genuinely succeeded; mark `completed`. Completed, + /// Another client changed the message first and the server said so. + /// + /// Terminal, and deliberately not a retry: the user's intent was formed + /// against state that no longer exists, so re-issuing it would either do + /// nothing or overwrite someone else's change. + Conflict(String), /// A transient failure; keep retrying until `MAX_RETRIES` is exceeded. Retry(String), /// A permanent failure nothing could retry away; mark `failed` now. @@ -209,6 +215,15 @@ pub async fn execute_pending_mutations( ); ("completed", item.retry_count, |s| s.completed += 1) } + Disposition::Conflict(observed) => { + tracing::warn!( + mutation_id = %item.id, + op = %item.mutation_type, + message_id = %item.message_id, + "outbox: mutation conflicted with another client: {observed}" + ); + ("conflicted", item.retry_count, |s| s.failed += 1) + } Disposition::Permanent(reason) => { tracing::warn!( mutation_id = %item.id, @@ -341,7 +356,19 @@ async fn dispatch_mailbox_mutation( ); match backend.apply_mutation(&spec).await { - Ok(()) => Disposition::Completed, + Ok(MutationOutcome::Applied { .. }) => Disposition::Completed, + // Another client won. Retrying would either do nothing or overwrite a + // change the user did not make, so this stops here and stays visible + // rather than being quietly re-attempted. + Ok(MutationOutcome::Conflict { observed }) => Disposition::Conflict(observed), + // The server accepted the command and proved nothing. Treated as + // retryable, because the alternative -- recording it as done -- is the + // exact silent-loss this outcome exists to prevent. A genuinely + // applied mutation re-attempted is a no-op; a lost one recorded as + // complete is unrecoverable. + Ok(MutationOutcome::Unknown { reason }) => { + Disposition::Retry(format!("mutation outcome unverified: {reason}")) + } Err(e) => Disposition::Retry(format!("remote mutation failed: {e}")), } }