mirror of
https://github.com/flectar/mail.git
synced 2026-11-12 02:05:59 +00:00
fix: sync imap moves
This commit is contained in:
1 parent
98720a5b59
commit
ff8af495e1
6 files changed
+629
-70
No files matched your search
@@ -1097,9 +1097,22 @@ pub async fn store_flag(session: &mut Session, uid: u32, flag: &str, add: bool)
|
||||
}
|
||||
|
||||
async fn store_flag_inner(session: &mut Session, uid: u32, flag: &str, add: bool) -> Result<()> {
|
||||
let op = if add { "+FLAGS" } else { "-FLAGS" };
|
||||
store_flag_set_inner(session, &uid.to_string(), flag, add).await
|
||||
}
|
||||
|
||||
async fn store_flag_set_inner(
|
||||
session: &mut Session,
|
||||
uid_set: &str,
|
||||
flag: &str,
|
||||
add: bool,
|
||||
) -> Result<()> {
|
||||
let op = if add {
|
||||
"+FLAGS.SILENT"
|
||||
} else {
|
||||
"-FLAGS.SILENT"
|
||||
};
|
||||
let mut stream = session
|
||||
.uid_store(uid.to_string(), format!("{op} ({flag})"))
|
||||
.uid_store(uid_set, format!("{op} ({flag})"))
|
||||
.await
|
||||
.map_err(|e| CoreError::Imap(e.to_string()))?;
|
||||
while let Some(item) = stream.next().await {
|
||||
@@ -1108,7 +1121,12 @@ async fn store_flag_inner(session: &mut Session, uid: u32, flag: &str, add: bool
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// MOVE if the server supports it, else COPY + \Deleted + EXPUNGE.
|
||||
/// Move one UID without guessing at unsupported server extensions.
|
||||
///
|
||||
/// RFC 6851 only permits `UID MOVE` when `MOVE` is advertised. Legacy IMAP
|
||||
/// servers use COPY + \Deleted instead. Prefer UID EXPUNGE for that fallback;
|
||||
/// without UIDPLUS, leave the source marked \Deleted because mailbox-wide
|
||||
/// EXPUNGE could remove another client's deleted mail.
|
||||
pub async fn uid_move(session: &mut Session, uid: u32, target: &str) -> Result<()> {
|
||||
with_deadline(
|
||||
"UID MOVE",
|
||||
@@ -1119,25 +1137,50 @@ pub async fn uid_move(session: &mut Session, uid: u32, target: &str) -> Result<(
|
||||
}
|
||||
|
||||
async fn uid_move_inner(session: &mut Session, uid: u32, target: &str) -> Result<()> {
|
||||
match session.uid_mv(uid.to_string(), target).await {
|
||||
Ok(()) => Ok(()),
|
||||
Err(_) => {
|
||||
session
|
||||
.uid_copy(uid.to_string(), target)
|
||||
.await
|
||||
.map_err(|e| CoreError::Imap(e.to_string()))?;
|
||||
store_flag(session, uid, "\\Deleted", true).await?;
|
||||
let stream = session
|
||||
.expunge()
|
||||
.await
|
||||
.map_err(|e| CoreError::Imap(e.to_string()))?;
|
||||
futures::pin_mut!(stream);
|
||||
while let Some(item) = stream.next().await {
|
||||
item.map_err(|e| CoreError::Imap(e.to_string()))?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
if uid == 0 {
|
||||
return Err(CoreError::Imap("IMAP UID must be greater than zero".into()));
|
||||
}
|
||||
|
||||
let capabilities = session
|
||||
.capabilities()
|
||||
.await
|
||||
.map_err(|e| CoreError::Imap(e.to_string()))?;
|
||||
if capabilities.has_str("MOVE") {
|
||||
// A failed RFC 6851 MOVE may already have moved this UID. Falling back
|
||||
// after such a failure can duplicate the message, so propagate it and
|
||||
// let the durable action reconcile before retrying.
|
||||
return session
|
||||
.uid_mv(uid.to_string(), target)
|
||||
.await
|
||||
.map_err(|e| CoreError::Imap(e.to_string()));
|
||||
}
|
||||
|
||||
session
|
||||
.uid_copy(uid.to_string(), target)
|
||||
.await
|
||||
.map_err(|e| CoreError::Imap(e.to_string()))?;
|
||||
store_flag_set_inner(session, &uid.to_string(), "\\Deleted", true).await?;
|
||||
|
||||
if capabilities.has_str("UIDPLUS") {
|
||||
let stream = session
|
||||
.uid_expunge(uid.to_string())
|
||||
.await
|
||||
.map_err(|e| CoreError::Imap(e.to_string()))?;
|
||||
futures::pin_mut!(stream);
|
||||
while let Some(item) = stream.next().await {
|
||||
item.map_err(|e| CoreError::Imap(e.to_string()))?;
|
||||
}
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
// There is no race-free way to expunge one UID in base IMAP. Even a
|
||||
// SEARCH immediately before EXPUNGE cannot prevent another client from
|
||||
// marking a different message \Deleted between those commands.
|
||||
tracing::debug!(
|
||||
uid,
|
||||
"legacy IMAP move left source marked deleted because UIDPLUS is unavailable"
|
||||
);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn append(session: &mut Session, folder: &str, raw: &[u8], seen: bool) -> Result<()> {
|
||||
|
||||
@@ -7062,6 +7062,13 @@ fn apply_thread_action(
|
||||
let target = params
|
||||
.and_then(|p| p.target_folder_id)
|
||||
.ok_or_else(|| CoreError::Other("move requires targetFolderId".into()))?;
|
||||
let target_folder = repo::folders::get(&tx, target)?
|
||||
.ok_or_else(|| CoreError::NotFound(format!("folder {target}")))?;
|
||||
if target_folder.account_id != account_id {
|
||||
return Err(CoreError::Other(
|
||||
"messages cannot be moved between accounts".into(),
|
||||
));
|
||||
}
|
||||
for (id, f, u, _r, _s, _role) in &msgs {
|
||||
if f.is_some() && *f != Some(target) {
|
||||
let aid = enqueue_move(
|
||||
@@ -7150,6 +7157,56 @@ fn apply_thread_action(
|
||||
Ok(out)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod move_action_validation_tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn move_rejects_a_folder_owned_by_another_account_without_local_mutation() {
|
||||
let mut conn = db::testutil::conn();
|
||||
db::testutil::seed_account(&conn);
|
||||
conn.execute(
|
||||
"INSERT INTO accounts (id, email, provider, auth_kind, username,
|
||||
imap_host, imap_port, smtp_host, smtp_port, created_at)
|
||||
VALUES (2,'other@test.dev','imap','password','other','h',993,'h',587,0)",
|
||||
[],
|
||||
)
|
||||
.unwrap();
|
||||
conn.execute(
|
||||
"INSERT INTO folders (id, account_id, imap_name, role)
|
||||
VALUES (2, 2, 'Archive', 'archive')",
|
||||
[],
|
||||
)
|
||||
.unwrap();
|
||||
let (thread_id, message_id) =
|
||||
db::testutil::seed_message(&conn, "sender@test.dev", "cross account", false);
|
||||
let before = repo::messages::get_row(&conn, message_id).unwrap().unwrap();
|
||||
|
||||
let error = apply_thread_action(
|
||||
&mut conn,
|
||||
thread_id,
|
||||
ActionKind::Move,
|
||||
Some(&ActionParams {
|
||||
wake_at: None,
|
||||
target_folder_id: Some(2),
|
||||
label_id: None,
|
||||
}),
|
||||
)
|
||||
.unwrap_err();
|
||||
|
||||
assert!(error.to_string().contains("between accounts"));
|
||||
let after = repo::messages::get_row(&conn, message_id).unwrap().unwrap();
|
||||
assert_eq!(after.folder_id, before.folder_id);
|
||||
assert_eq!(after.uid, before.uid);
|
||||
assert_eq!(
|
||||
conn.query_row("SELECT COUNT(*) FROM pending_actions", [], |row| row
|
||||
.get::<_, i64>(0))
|
||||
.unwrap(),
|
||||
0
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/// Inverse of an action: cancel if pending, revert the local mutation, and
|
||||
/// enqueue a compensating remote action when the original already ran.
|
||||
/// Returns the affected thread id.
|
||||
|
||||
@@ -78,19 +78,14 @@ pub async fn execute_due(
|
||||
account_id, action_id, kind = %action.kind, error = %e,
|
||||
"action needs auth; pausing (check mail-host auth / SMTP AUTH enabled)",
|
||||
);
|
||||
// A queued send that can't authenticate would otherwise sit in the
|
||||
// outbox indefinitely with the composer already closed: the message
|
||||
// just vanishes with no warning (only a needs-reauth dot buried in
|
||||
// Settings). Tell the UI so it can surface that the mail wasn't sent
|
||||
// and point the user at re-authentication. Other action kinds stay
|
||||
// quiet; the account's needs_reauth state already covers them.
|
||||
if action.kind == "send" {
|
||||
ctx.bus.emit(CoreEvent::ActionState {
|
||||
action_id,
|
||||
state: "paused".into(),
|
||||
error: Some(e.to_string()),
|
||||
});
|
||||
}
|
||||
// The optimistic local mutation remains visible while auth is
|
||||
// paused, so every action kind must tell the UI it has not yet
|
||||
// reached the server.
|
||||
ctx.bus.emit(CoreEvent::ActionState {
|
||||
action_id,
|
||||
state: "paused".into(),
|
||||
error: Some(e.to_string()),
|
||||
});
|
||||
let msg = "authentication required".to_string();
|
||||
ctx.db
|
||||
.write(move |conn| {
|
||||
@@ -106,18 +101,8 @@ pub async fn execute_due(
|
||||
account_id, action_id, kind = %action.kind, attempts, error = %msg,
|
||||
"action attempt failed",
|
||||
);
|
||||
if attempts >= MAX_ATTEMPTS || is_permanent(&msg) {
|
||||
let m = msg.clone();
|
||||
ctx.db
|
||||
.write(move |conn| {
|
||||
repo::actions::set_state(conn, action_id, "failed", Some(&m))
|
||||
})
|
||||
.await?;
|
||||
ctx.bus.emit(CoreEvent::ActionState {
|
||||
action_id,
|
||||
state: "failed".into(),
|
||||
error: Some(msg),
|
||||
});
|
||||
if attempts >= MAX_ATTEMPTS || is_permanent(&e) {
|
||||
fail_action(ctx, &action, msg).await?;
|
||||
} else {
|
||||
// Exponential backoff with jitter.
|
||||
let delay = (1 << attempts.min(8)) * 1000 + (action_id % 997);
|
||||
@@ -127,6 +112,11 @@ pub async fn execute_due(
|
||||
repo::actions::bump_attempt(conn, action_id, now_ms() + delay, &m)
|
||||
})
|
||||
.await?;
|
||||
ctx.bus.emit(CoreEvent::ActionState {
|
||||
action_id,
|
||||
state: "retrying".into(),
|
||||
error: Some(msg.clone()),
|
||||
});
|
||||
// Connection-level errors: bail out, actor reconnects.
|
||||
if is_connection_failure(&msg) {
|
||||
return Err(e);
|
||||
@@ -146,9 +136,28 @@ pub async fn execute_due(
|
||||
})
|
||||
}
|
||||
|
||||
fn is_permanent(msg: &str) -> bool {
|
||||
// IMAP tagged NO responses and SMTP 5xx are not going to succeed on retry.
|
||||
msg.contains("NO ") || msg.contains("550") || msg.contains("553") || msg.contains("bad ")
|
||||
fn is_permanent(error: &CoreError) -> bool {
|
||||
match error {
|
||||
CoreError::NotFound(_) => true,
|
||||
// Tagged NO/BAD responses reject the command itself. Async IMAP has
|
||||
// used both `NO ...` and `no: ...` display forms across versions.
|
||||
CoreError::Imap(message) => {
|
||||
let message = message.trim().to_ascii_lowercase();
|
||||
["no", "bad"].iter().any(|status| {
|
||||
message == *status
|
||||
|| message
|
||||
.strip_prefix(status)
|
||||
.is_some_and(|tail| matches!(tail.chars().next(), Some(' ' | ':' | '[')))
|
||||
})
|
||||
}
|
||||
// All SMTP 5xx replies are permanent failures of the current request.
|
||||
CoreError::Smtp(message) => message.split(|c: char| !c.is_ascii_digit()).any(|token| {
|
||||
token.len() == 3
|
||||
&& token.starts_with('5')
|
||||
&& token.bytes().all(|byte| byte.is_ascii_digit())
|
||||
}),
|
||||
_ => false,
|
||||
}
|
||||
}
|
||||
|
||||
fn is_connection_failure(msg: &str) -> bool {
|
||||
@@ -160,6 +169,107 @@ fn is_connection_failure(msg: &str) -> bool {
|
||||
|| msg.contains("unexpected eof")
|
||||
}
|
||||
|
||||
fn is_move_kind(kind: &str) -> bool {
|
||||
matches!(
|
||||
kind,
|
||||
"archive" | "unarchive" | "trash" | "spam" | "not_spam" | "move"
|
||||
)
|
||||
}
|
||||
|
||||
/// Mark an exhausted action failed and undo its optimistic move when it is
|
||||
/// still the newest intent for that message. A later queued move owns the
|
||||
/// visible location and must not be overwritten by an older failure.
|
||||
async fn fail_action(
|
||||
ctx: &SyncCtx,
|
||||
action: &repo::actions::PendingAction,
|
||||
message: String,
|
||||
) -> Result<()> {
|
||||
let failed = action.clone();
|
||||
let saved_message = message.clone();
|
||||
let changed_thread = ctx
|
||||
.db
|
||||
.write(move |conn| mark_failed_and_rollback(conn, &failed, &saved_message))
|
||||
.await?;
|
||||
|
||||
if let Some(thread_id) = changed_thread {
|
||||
ctx.bus.emit(CoreEvent::MailUpdated {
|
||||
thread_ids: vec![thread_id],
|
||||
});
|
||||
}
|
||||
ctx.bus.emit(CoreEvent::ActionState {
|
||||
action_id: action.id,
|
||||
state: "failed".into(),
|
||||
error: Some(message),
|
||||
});
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn mark_failed_and_rollback(
|
||||
conn: &mut rusqlite::Connection,
|
||||
failed: &repo::actions::PendingAction,
|
||||
message: &str,
|
||||
) -> Result<Option<i64>> {
|
||||
let tx = conn.transaction()?;
|
||||
repo::actions::set_state(&tx, failed.id, "failed", Some(message))?;
|
||||
if !is_move_kind(&failed.kind) {
|
||||
tx.commit()?;
|
||||
return Ok(None);
|
||||
}
|
||||
|
||||
let (Some(message_id), Some(source_folder), target_folder) = (
|
||||
failed.message_id,
|
||||
failed.payload["srcFolderId"].as_i64(),
|
||||
failed.payload["targetFolderId"].as_i64(),
|
||||
) else {
|
||||
tx.commit()?;
|
||||
return Ok(None);
|
||||
};
|
||||
let has_newer_move: bool = tx.query_row(
|
||||
"SELECT EXISTS(
|
||||
SELECT 1 FROM pending_actions
|
||||
WHERE message_id=?1 AND id>?2
|
||||
AND kind IN ('archive','unarchive','trash','spam','not_spam','move')
|
||||
AND state IN ('pending','inflight','done')
|
||||
)",
|
||||
rusqlite::params![message_id, failed.id],
|
||||
|row| row.get(0),
|
||||
)?;
|
||||
let row = repo::messages::get_row(&tx, message_id)?;
|
||||
if has_newer_move
|
||||
|| row
|
||||
.as_ref()
|
||||
.is_none_or(|row| row.folder_id != target_folder)
|
||||
{
|
||||
tx.commit()?;
|
||||
return Ok(None);
|
||||
}
|
||||
|
||||
let source_uid = failed.payload["srcUid"].as_i64();
|
||||
repo::messages::set_uid_and_folder(&tx, message_id, source_folder, source_uid)?;
|
||||
let multi_mailbox: bool = tx.query_row(
|
||||
"SELECT provider='gmail' OR mail_protocol='jmap' FROM accounts WHERE id=?1",
|
||||
rusqlite::params![failed.account_id],
|
||||
|row| row.get(0),
|
||||
)?;
|
||||
if multi_mailbox {
|
||||
if let Some(target_folder) = target_folder {
|
||||
tx.execute(
|
||||
"DELETE FROM message_folders WHERE message_id=?1 AND folder_id=?2",
|
||||
rusqlite::params![message_id, target_folder],
|
||||
)?;
|
||||
}
|
||||
tx.execute(
|
||||
"INSERT OR IGNORE INTO message_folders (message_id, folder_id) VALUES (?1, ?2)",
|
||||
rusqlite::params![message_id, source_folder],
|
||||
)?;
|
||||
}
|
||||
if let Some(thread_id) = failed.thread_id {
|
||||
repo::threads::recompute(&tx, thread_id)?;
|
||||
}
|
||||
tx.commit()?;
|
||||
Ok(failed.thread_id)
|
||||
}
|
||||
|
||||
async fn execute_one(
|
||||
ctx: &SyncCtx,
|
||||
config: &AccountConfig,
|
||||
@@ -220,21 +330,27 @@ async fn resolve_source(
|
||||
ctx: &SyncCtx,
|
||||
session: &mut Session,
|
||||
action: &repo::actions::PendingAction,
|
||||
) -> Result<Option<(repo::folders::Folder, u32)>> {
|
||||
let src_folder_id = action.payload["srcFolderId"].as_i64();
|
||||
let src_uid = action.payload["srcUid"].as_i64();
|
||||
let (Some(fid), Some(uid)) = (src_folder_id, src_uid) else {
|
||||
return Ok(None);
|
||||
};
|
||||
) -> Result<(repo::folders::Folder, u32)> {
|
||||
let fid = action.payload["srcFolderId"]
|
||||
.as_i64()
|
||||
.ok_or_else(|| CoreError::NotFound("source mailbox for queued move".into()))?;
|
||||
let uid = action.payload["srcUid"]
|
||||
.as_i64()
|
||||
.and_then(|uid| u32::try_from(uid).ok())
|
||||
.filter(|uid| *uid > 0)
|
||||
.ok_or_else(|| CoreError::NotFound("valid source UID for queued move".into()))?;
|
||||
let folder = ctx
|
||||
.db
|
||||
.read(move |conn| repo::folders::get(conn, fid))
|
||||
.await?;
|
||||
let Some(folder) = folder else {
|
||||
return Ok(None);
|
||||
};
|
||||
let folder = folder.ok_or_else(|| CoreError::NotFound("source folder".into()))?;
|
||||
if folder.account_id != action.account_id {
|
||||
return Err(CoreError::Other(
|
||||
"queued move source belongs to another account".into(),
|
||||
));
|
||||
}
|
||||
imap::select(session, &folder.imap_name).await?;
|
||||
Ok(Some((folder, uid as u32)))
|
||||
Ok((folder, uid))
|
||||
}
|
||||
|
||||
async fn flag_action(
|
||||
@@ -278,27 +394,28 @@ async fn move_action(
|
||||
session: &mut Session,
|
||||
action: &repo::actions::PendingAction,
|
||||
) -> Result<()> {
|
||||
let Some(message_id) = action.message_id else {
|
||||
return Ok(());
|
||||
};
|
||||
let Some((_src, uid)) = resolve_source(ctx, session, action).await? else {
|
||||
return Ok(());
|
||||
};
|
||||
let target_folder_id = action.payload["targetFolderId"].as_i64();
|
||||
let Some(tfid) = target_folder_id else {
|
||||
return Ok(());
|
||||
};
|
||||
let message_id = action
|
||||
.message_id
|
||||
.ok_or_else(|| CoreError::NotFound("message for queued move".into()))?;
|
||||
let (_src, uid) = resolve_source(ctx, session, action).await?;
|
||||
let tfid = action.payload["targetFolderId"]
|
||||
.as_i64()
|
||||
.ok_or_else(|| CoreError::NotFound("target mailbox for queued move".into()))?;
|
||||
let target = ctx
|
||||
.db
|
||||
.read(move |conn| repo::folders::get(conn, tfid))
|
||||
.await?
|
||||
.ok_or_else(|| CoreError::NotFound("target folder".into()))?;
|
||||
if target.account_id != action.account_id || target.account_id != config.id {
|
||||
return Err(CoreError::Other(
|
||||
"queued move target belongs to another account".into(),
|
||||
));
|
||||
}
|
||||
|
||||
imap::uid_move(session, uid, &target.imap_name).await?;
|
||||
|
||||
// The message's new UID in the target is unknown (COPYUID not parsed in
|
||||
// v1); clear it so the next target-folder sync re-links by Message-ID.
|
||||
let _ = config;
|
||||
ctx.db
|
||||
.write(move |conn| repo::messages::set_uid_and_folder(conn, message_id, tfid, None))
|
||||
.await?;
|
||||
@@ -629,4 +746,141 @@ mod tests {
|
||||
assert!(is_connection_failure("connection closed"));
|
||||
assert!(!is_connection_failure("temporary mailbox quota"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn command_rejections_are_permanent_but_transport_errors_retry() {
|
||||
assert!(is_permanent(&CoreError::Imap(
|
||||
"NO [NOPERM] permission denied".into()
|
||||
)));
|
||||
assert!(is_permanent(&CoreError::Imap(
|
||||
"bad: invalid command".into()
|
||||
)));
|
||||
assert!(is_permanent(&CoreError::Smtp(
|
||||
"554 transaction failed".into()
|
||||
)));
|
||||
assert!(is_permanent(&CoreError::NotFound("source mailbox".into())));
|
||||
assert!(!is_permanent(&CoreError::Imap(
|
||||
"connection timed out".into()
|
||||
)));
|
||||
assert!(!is_permanent(&CoreError::Smtp(
|
||||
"451 temporary failure".into()
|
||||
)));
|
||||
}
|
||||
|
||||
fn optimistic_move(
|
||||
conn: &rusqlite::Connection,
|
||||
message_id: i64,
|
||||
thread_id: i64,
|
||||
target_folder: i64,
|
||||
) -> repo::actions::PendingAction {
|
||||
let source_uid = repo::messages::get_row(conn, message_id)
|
||||
.unwrap()
|
||||
.unwrap()
|
||||
.uid;
|
||||
let action_id = repo::actions::enqueue(
|
||||
conn,
|
||||
1,
|
||||
"move",
|
||||
Some(message_id),
|
||||
Some(thread_id),
|
||||
&serde_json::json!({
|
||||
"srcFolderId": 1,
|
||||
"srcUid": source_uid,
|
||||
"targetFolderId": target_folder,
|
||||
}),
|
||||
None,
|
||||
)
|
||||
.unwrap();
|
||||
repo::messages::set_uid_and_folder(conn, message_id, target_folder, None).unwrap();
|
||||
repo::actions::get(conn, action_id).unwrap().unwrap()
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn terminal_move_failure_restores_the_remote_source_mapping() {
|
||||
let mut conn = crate::db::testutil::conn();
|
||||
crate::db::testutil::seed_account(&conn);
|
||||
conn.execute(
|
||||
"INSERT INTO folders (id, account_id, imap_name, role)
|
||||
VALUES (2, 1, 'Archive', 'archive')",
|
||||
[],
|
||||
)
|
||||
.unwrap();
|
||||
let (thread_id, message_id) =
|
||||
crate::db::testutil::seed_message(&conn, "sender@test.dev", "move", false);
|
||||
let source_uid = repo::messages::get_row(&conn, message_id)
|
||||
.unwrap()
|
||||
.unwrap()
|
||||
.uid;
|
||||
let action = optimistic_move(&conn, message_id, thread_id, 2);
|
||||
|
||||
assert_eq!(
|
||||
mark_failed_and_rollback(&mut conn, &action, "server rejected move").unwrap(),
|
||||
Some(thread_id)
|
||||
);
|
||||
let restored = repo::messages::get_row(&conn, message_id).unwrap().unwrap();
|
||||
assert_eq!(restored.folder_id, Some(1));
|
||||
assert_eq!(restored.uid, source_uid);
|
||||
assert_eq!(
|
||||
repo::actions::get(&conn, action.id).unwrap().unwrap().state,
|
||||
"failed"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn older_failure_does_not_overwrite_a_completed_newer_move() {
|
||||
let mut conn = crate::db::testutil::conn();
|
||||
crate::db::testutil::seed_account(&conn);
|
||||
conn.execute_batch(
|
||||
"INSERT INTO folders (id, account_id, imap_name, role)
|
||||
VALUES (2, 1, 'Archive', 'archive');
|
||||
INSERT INTO folders (id, account_id, imap_name, role)
|
||||
VALUES (3, 1, 'Later', '');",
|
||||
)
|
||||
.unwrap();
|
||||
let (thread_id, message_id) =
|
||||
crate::db::testutil::seed_message(&conn, "sender@test.dev", "move race", false);
|
||||
let first = optimistic_move(&conn, message_id, thread_id, 2);
|
||||
let newer = repo::actions::enqueue(
|
||||
&conn,
|
||||
1,
|
||||
"move",
|
||||
Some(message_id),
|
||||
Some(thread_id),
|
||||
&serde_json::json!({
|
||||
"srcFolderId": 2,
|
||||
"srcUid": 84,
|
||||
"targetFolderId": 3,
|
||||
}),
|
||||
None,
|
||||
)
|
||||
.unwrap();
|
||||
repo::actions::set_state(&conn, newer, "done", None).unwrap();
|
||||
let newest = repo::actions::enqueue(
|
||||
&conn,
|
||||
1,
|
||||
"move",
|
||||
Some(message_id),
|
||||
Some(thread_id),
|
||||
&serde_json::json!({
|
||||
"srcFolderId": 3,
|
||||
"srcUid": 126,
|
||||
"targetFolderId": 2,
|
||||
}),
|
||||
None,
|
||||
)
|
||||
.unwrap();
|
||||
repo::actions::set_state(&conn, newest, "done", None).unwrap();
|
||||
|
||||
assert_eq!(
|
||||
mark_failed_and_rollback(&mut conn, &first, "server rejected move").unwrap(),
|
||||
None
|
||||
);
|
||||
assert_eq!(
|
||||
repo::messages::get_row(&conn, message_id)
|
||||
.unwrap()
|
||||
.unwrap()
|
||||
.folder_id,
|
||||
Some(2)
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -181,6 +181,192 @@ async fn untrusted_certificate_is_rejected() {
|
||||
task.await.unwrap();
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy)]
|
||||
enum ImapMoveFixture {
|
||||
LegacyUidPlus,
|
||||
AdvertisedMoveRejected,
|
||||
LegacyWithoutUidPlus,
|
||||
}
|
||||
|
||||
async fn imap_move_server(fixture: ImapMoveFixture) -> (u16, tokio::task::JoinHandle<()>) {
|
||||
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
|
||||
let port = listener.local_addr().unwrap().port();
|
||||
let task = tokio::spawn(async move {
|
||||
let (tcp, _) = listener.accept().await.unwrap();
|
||||
let tls = acceptor_for(CERT).accept(tcp).await.unwrap();
|
||||
let mut stream = BufReader::new(tls);
|
||||
stream.write_all(b"* OK test IMAP ready\r\n").await.unwrap();
|
||||
|
||||
let login = line(&mut stream).await;
|
||||
assert!(login.contains(" LOGIN "), "unexpected command: {login:?}");
|
||||
let tag = login.split_whitespace().next().unwrap();
|
||||
stream
|
||||
.write_all(format!("{tag} OK logged in\r\n").as_bytes())
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let select = line(&mut stream).await;
|
||||
assert!(
|
||||
select.contains(" SELECT "),
|
||||
"unexpected command: {select:?}"
|
||||
);
|
||||
let tag = select.split_whitespace().next().unwrap();
|
||||
stream
|
||||
.write_all(
|
||||
format!(
|
||||
"* FLAGS (\\Seen \\Deleted)\r\n\
|
||||
* 2 EXISTS\r\n\
|
||||
* OK [UIDVALIDITY 1] valid\r\n\
|
||||
* OK [UIDNEXT 43] next\r\n\
|
||||
{tag} OK [READ-WRITE] selected\r\n"
|
||||
)
|
||||
.as_bytes(),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let capability = line(&mut stream).await;
|
||||
assert!(
|
||||
capability.contains(" CAPABILITY"),
|
||||
"unexpected command: {capability:?}"
|
||||
);
|
||||
let tag = capability.split_whitespace().next().unwrap();
|
||||
let capabilities = match fixture {
|
||||
ImapMoveFixture::LegacyUidPlus => "IMAP4rev1 UIDPLUS",
|
||||
ImapMoveFixture::AdvertisedMoveRejected => "IMAP4rev1 UIDPLUS MOVE",
|
||||
ImapMoveFixture::LegacyWithoutUidPlus => "IMAP4rev1",
|
||||
};
|
||||
stream
|
||||
.write_all(format!("* CAPABILITY {capabilities}\r\n{tag} OK capability\r\n").as_bytes())
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
match fixture {
|
||||
ImapMoveFixture::LegacyUidPlus => {
|
||||
let copy = line(&mut stream).await;
|
||||
assert!(
|
||||
copy.contains(" UID COPY 42 "),
|
||||
"unexpected command: {copy:?}"
|
||||
);
|
||||
let tag = copy.split_whitespace().next().unwrap();
|
||||
stream
|
||||
.write_all(format!("{tag} OK copied\r\n").as_bytes())
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let store = line(&mut stream).await;
|
||||
assert!(
|
||||
store.contains(" UID STORE 42 +FLAGS.SILENT (\\Deleted)"),
|
||||
"unexpected command: {store:?}"
|
||||
);
|
||||
let tag = store.split_whitespace().next().unwrap();
|
||||
stream
|
||||
.write_all(format!("{tag} OK stored\r\n").as_bytes())
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let expunge = line(&mut stream).await;
|
||||
assert!(
|
||||
expunge.contains(" UID EXPUNGE 42"),
|
||||
"unexpected command: {expunge:?}"
|
||||
);
|
||||
let tag = expunge.split_whitespace().next().unwrap();
|
||||
stream
|
||||
.write_all(format!("* 1 EXPUNGE\r\n{tag} OK expunged\r\n").as_bytes())
|
||||
.await
|
||||
.unwrap();
|
||||
}
|
||||
ImapMoveFixture::AdvertisedMoveRejected => {
|
||||
let move_command = line(&mut stream).await;
|
||||
assert!(
|
||||
move_command.contains(" UID MOVE 42 "),
|
||||
"unexpected command: {move_command:?}"
|
||||
);
|
||||
let tag = move_command.split_whitespace().next().unwrap();
|
||||
stream
|
||||
.write_all(format!("{tag} NO [NOPERM] move denied\r\n").as_bytes())
|
||||
.await
|
||||
.unwrap();
|
||||
}
|
||||
ImapMoveFixture::LegacyWithoutUidPlus => {
|
||||
let copy = line(&mut stream).await;
|
||||
assert!(
|
||||
copy.contains(" UID COPY 42 "),
|
||||
"unexpected command: {copy:?}"
|
||||
);
|
||||
let tag = copy.split_whitespace().next().unwrap();
|
||||
stream
|
||||
.write_all(format!("{tag} OK copied\r\n").as_bytes())
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let store = line(&mut stream).await;
|
||||
assert!(
|
||||
store.contains(" UID STORE 42 +FLAGS.SILENT (\\Deleted)"),
|
||||
"unexpected command: {store:?}"
|
||||
);
|
||||
let tag = store.split_whitespace().next().unwrap();
|
||||
stream
|
||||
.write_all(format!("{tag} OK stored\r\n").as_bytes())
|
||||
.await
|
||||
.unwrap();
|
||||
}
|
||||
}
|
||||
|
||||
// In the rejection case this proves the client did not issue a second
|
||||
// COPY. Without UIDPLUS it proves the client did not use the unsafe,
|
||||
// mailbox-wide EXPUNGE command.
|
||||
let logout = line(&mut stream).await;
|
||||
assert!(
|
||||
logout.contains(" LOGOUT"),
|
||||
"unsafe command after move result: {logout:?}"
|
||||
);
|
||||
let tag = logout.split_whitespace().next().unwrap();
|
||||
stream
|
||||
.write_all(format!("* BYE closing\r\n{tag} OK logout\r\n").as_bytes())
|
||||
.await
|
||||
.unwrap();
|
||||
});
|
||||
(port, task)
|
||||
}
|
||||
|
||||
async fn run_imap_move(fixture: ImapMoveFixture) -> flectar_mail_core::error::Result<()> {
|
||||
let (port, task) = imap_move_server(fixture).await;
|
||||
let mut session = imap::connect_with_settings(
|
||||
"127.0.0.1",
|
||||
port,
|
||||
credentials(),
|
||||
&settings(ConnectionSecurity::Tls),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
imap::select(&mut session, "INBOX").await.unwrap();
|
||||
let result = imap::uid_move(&mut session, 42, "Archive").await;
|
||||
imap::logout(session).await;
|
||||
task.await.unwrap();
|
||||
result
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn imap_move_uses_uidplus_fallback_when_move_is_not_advertised() {
|
||||
run_imap_move(ImapMoveFixture::LegacyUidPlus).await.unwrap();
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn rejected_advertised_move_does_not_fall_back_to_copy() {
|
||||
let error = run_imap_move(ImapMoveFixture::AdvertisedMoveRejected)
|
||||
.await
|
||||
.unwrap_err();
|
||||
assert!(error.to_string().contains("move denied"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn legacy_move_without_uidplus_never_uses_mailbox_wide_expunge() {
|
||||
run_imap_move(ImapMoveFixture::LegacyWithoutUidPlus)
|
||||
.await
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
async fn smtp_connection(mode: ConnectionSecurity) {
|
||||
smtp_connection_with_certificate(mode, CERT).await;
|
||||
}
|
||||
|
||||
@@ -2882,6 +2882,13 @@ pub fn run(platform: PlatformContext) -> Result<(), Box<dyn std::error::Error>>
|
||||
{
|
||||
apply_account_states(&app, &mail_update_state, &pending.account_states);
|
||||
}
|
||||
if let Some(error) = pending.action_error
|
||||
&& let Some(app) = mail_update_app.upgrade()
|
||||
{
|
||||
let message = UiMessage::detail("Message action failed: {}", error);
|
||||
app.set_render_status(message.clone());
|
||||
app.set_sync_status(message);
|
||||
}
|
||||
if pending.calendar_changed
|
||||
&& let Some(app) = mail_update_app.upgrade()
|
||||
&& app.get_active_view() == "calendar"
|
||||
|
||||
@@ -409,6 +409,7 @@ pub(crate) struct PendingCoreUpdates {
|
||||
pub(crate) changed_threads: HashSet<i64>,
|
||||
pub(crate) all_mail_changed: bool,
|
||||
pub(crate) account_states: HashMap<i64, (String, Option<String>)>,
|
||||
pub(crate) action_error: Option<String>,
|
||||
pub(crate) calendar_changed: bool,
|
||||
pub(crate) contacts_changed: bool,
|
||||
}
|
||||
@@ -541,6 +542,17 @@ pub(crate) fn spawn_core_event_listener(
|
||||
.insert(account_id, (sync_state, sync_error));
|
||||
true
|
||||
}
|
||||
Ok(CoreEvent::ActionState {
|
||||
state,
|
||||
error: Some(error),
|
||||
..
|
||||
}) if state == "failed" => {
|
||||
pending
|
||||
.lock()
|
||||
.unwrap_or_else(|poisoned| poisoned.into_inner())
|
||||
.action_error = Some(error);
|
||||
true
|
||||
}
|
||||
Ok(CoreEvent::CalendarUpdated { .. })
|
||||
| Ok(CoreEvent::CalendarEventsAdded { .. }) => {
|
||||
pending
|
||||
|
||||
Reference in new issue
Block a user