Mirrors/chatmail-core
1
0
Fork 0
mirror of https://github.com/chatmail/core.git synced 2026-04-29 11:26:29 +03:00
Code Issues Packages Projects Releases Wiki Activity

Refactor fetch_many_msgs and add more logging

Browse Source
This commit is contained in:
link2xt
2022-12-03 23:12:32 +00:00
parent 22b4640b31
commit 3743aaa16e
2 changed files with 113 additions and 72 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGELOG.md
View File

@@ -4,6 +4,7 @@
### Changes ### Changes
- Refactor: Remove the remaining AsRef<str> #3669 - Refactor: Remove the remaining AsRef<str> #3669
- Add more logging to `fetch_many_msgs` and refactor it #3811
- Small speedup #3780 - Small speedup #3780
- Log the reason when the message cannot be sent to the chat #3810 - Log the reason when the message cannot be sent to the chat #3810

196
src/imap.rs
View File

@@ -1374,22 +1374,27 @@ impl Imap {
&mut self, &mut self,
context: &Context, context: &Context,
folder: &str, folder: &str,
server_uids: Vec<u32>, request_uids: Vec<u32>,
uid_message_ids: &BTreeMap<u32, String>, uid_message_ids: &BTreeMap<u32, String>,
fetch_partially: bool, fetch_partially: bool,
fetching_existing_messages: bool, fetching_existing_messages: bool,
) -> Result<(Option<u32>, Vec<ReceivedMsg>)> { ) -> Result<(Option<u32>, Vec<ReceivedMsg>)> {
let mut last_uid = None;
let mut received_msgs = Vec::new(); let mut received_msgs = Vec::new();
if server_uids.is_empty() {
return Ok((None, Vec::new())); if request_uids.is_empty() {
return Ok((last_uid, received_msgs));
} }
let session = self.session.as_mut().context("no IMAP session")?; let session = self.session.as_mut().context("no IMAP session")?;
let sets = build_sequence_sets(&server_uids)?; for (request_uids, set) in build_sequence_sets(&request_uids)? {
let mut last_uid = None; info!(
context,
for (server_uids, set) in sets.iter() { "Starting a {} FETCH of message set \"{}\".",
let mut msgs = match session if fetch_partially { "partial" } else { "full" },
set
);
let mut fetch_responses = match session
.uid_fetch( .uid_fetch(
&set, &set,
if fetch_partially { if fetch_partially {
@@ -1399,94 +1404,122 @@ impl Imap {
}, },
) )
.await .await
.with_context(|| format!("fetching messages {} from folder \"{}\"", &set, folder))
{ {
Ok(msgs) => msgs, Ok(fetch_responses) => fetch_responses,
Err(err) => { Err(err) => {
// TODO: maybe differentiate between IO and input/parsing problems // We want to reconnect regardless of whether it's an I/O error or parsing
// so we don't reconnect if we have a (rare) input/output parsing problem? // error. If the protocol parser ends up in incorrect state because of some
// incompatiblity with a server, reset may help.
self.should_reconnect = true; self.should_reconnect = true;
bail!( return Err(err);
"Error on fetching messages #{} from folder \"{}\"; error={}.",
&set,
folder,
err
);
} }
}; };
let mut uid_msgs = server_uids // Map from UIDs to unprocessed FETCH results. We put unprocessed FETCH results here
.iter() // when we want to process other messages first.
.map(|&uid| (uid, None)) let mut uid_msgs = HashMap::with_capacity(request_uids.len());
.collect::<HashMap<_, _>>();
let mut server_uids_it = server_uids.iter().peekable();
let mut count = 0; let mut count = 0;
while let Some(&&server_uid) = server_uids_it.peek() { for &request_uid in request_uids.iter() {
let mut msg = uid_msgs.insert(server_uid, None).flatten(); // Check if FETCH response is already in `uid_msgs`.
while msg.is_none() { let mut fetch_response = uid_msgs.remove(&request_uid);
let msg_unwrapped = match msgs.next().await {
Some(Ok(msg)) => msg, // Try to find a requsted UID in returned FETCH responses.
Some(Err(_)) => continue, while fetch_response.is_none() {
None => break, let next_fetch_response =
if let Some(next_fetch_response) = fetch_responses.next().await {
next_fetch_response
} else {
// No more FETCH responses received from the server.
break;
}; };
let msg_uid = msg_unwrapped.uid.unwrap_or_default();
if !uid_msgs.contains_key(&msg_uid) { let next_fetch_response = match next_fetch_response {
Ok(next_fetch_response) => next_fetch_response,
Err(err) => {
warn!(context, "Failed to process IMAP FETCH result: {}.", err);
continue;
}
};
if let Some(next_uid) = next_fetch_response.uid {
if next_uid == request_uid {
fetch_response = Some(next_fetch_response);
} else if !request_uids.contains(&next_uid) {
// (size of `request_uids` is bounded by IMAP command length limit,
// search in this vector is always fast)
// Unwanted UIDs are possible because of unsolicited responses, e.g. if // Unwanted UIDs are possible because of unsolicited responses, e.g. if
// another client changes \Seen flag on a message after we do a prefetch but // another client changes \Seen flag on a message after we do a prefetch but
// before fetch. It's not an error if we receive such unsolicited response. // before fetch. It's not an error if we receive such unsolicited response.
continue;
}
msg = Some(msg_unwrapped);
if msg_uid != server_uid && uid_msgs.insert(msg_uid, msg.take()).is_some() {
warn!(context, "Got duplicated UID {}", msg_uid);
}
}
let msg = match msg {
Some(msg) => msg,
None => {
warn!(context, "Missed UID {} in the server response", server_uid);
server_uids_it.next();
continue;
}
};
server_uids_it.next();
count += 1;
let is_deleted = msg.flags().any(|flag| flag == Flag::Deleted);
let (body, partial) = if fetch_partially {
(msg.header(), msg.size) // `BODY.PEEK[HEADER]` goes to header() ...
} else {
(msg.body(), None) // ... while `BODY.PEEK[]` goes to body() - and includes header()
};
if is_deleted || body.is_none() {
info!( info!(
context, context,
"Not processing deleted or empty msg {}", server_uid "Skipping not requested FETCH response for UID {}.", next_uid
); );
last_uid = Some(server_uid); } else if uid_msgs.insert(next_uid, next_fetch_response).is_some() {
warn!(context, "Got duplicated UID {}.", next_uid);
}
} else {
info!(context, "Skipping FETCH response without UID.");
}
}
let fetch_response = match fetch_response {
Some(fetch) => fetch,
None => {
warn!(
context,
"Missed UID {} in the server response.", request_uid
);
continue;
}
};
count += 1;
let is_deleted = fetch_response.flags().any(|flag| flag == Flag::Deleted);
let (body, partial) = if fetch_partially {
(fetch_response.header(), fetch_response.size) // `BODY.PEEK[HEADER]` goes to header() ...
} else {
(fetch_response.body(), None) // ... while `BODY.PEEK[]` goes to body() - and includes header()
};
if is_deleted {
info!(context, "Not processing deleted msg {}.", request_uid);
last_uid = Some(request_uid);
continue; continue;
} }
// XXX put flags into a set and pass them to receive_imf let body = if let Some(body) = body {
let context = context.clone(); body
} else {
info!(
context,
"Not processing message {} without a BODY.", request_uid
);
last_uid = Some(request_uid);
continue;
};
// safe, as we checked above that there is a body. let is_seen = fetch_response.flags().any(|flag| flag == Flag::Seen);
let body = body
.context("we checked that message has body right above, but it has vanished")?;
let is_seen = msg.flags().any(|flag| flag == Flag::Seen);
let rfc724_mid = if let Some(rfc724_mid) = uid_message_ids.get(&server_uid) { let rfc724_mid = if let Some(rfc724_mid) = uid_message_ids.get(&request_uid) {
rfc724_mid rfc724_mid
} else { } else {
warn!( error!(
context, context,
"No Message-ID corresponding to UID {} passed in uid_messsage_ids", "No Message-ID corresponding to UID {} passed in uid_messsage_ids.",
server_uid request_uid
); );
continue; continue;
}; };
info!(
context,
"Passing message UID {} to receive_imf().", request_uid
);
match receive_imf_inner( match receive_imf_inner(
&context, context,
rfc724_mid, rfc724_mid,
body, body,
is_seen, is_seen,
@@ -1501,22 +1534,29 @@ impl Imap {
} }
} }
Err(err) => { Err(err) => {
warn!(context, "receive_imf error: {:#}", err); warn!(context, "receive_imf error: {:#}.", err);
} }
}; };
last_uid = Some(server_uid) last_uid = Some(request_uid)
} }
// If we don't process the whole response, IMAP client is left in a broken state where // If we don't process the whole response, IMAP client is left in a broken state where
// it will try to process the rest of response as the next response. // it will try to process the rest of response as the next response.
while msgs.next().await.is_some() {} while fetch_responses.next().await.is_some() {}
if count != server_uids.len() {
if count != request_uids.len() {
warn!( warn!(
context, context,
"failed to fetch all uids: got {}, requested {}, we requested the UIDs {:?} using {:?}", "Failed to fetch all UIDs: got {}, requested {}, we requested the UIDs {:?}.",
count, count,
server_uids.len(), request_uids.len(),
server_uids, request_uids,
sets, );
} else {
info!(
context,
"Successfully received {} UIDs.",
request_uids.len()
); );
} }
} }