filip/Stalwart
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

Automatic retry for import/export blob downloads (#14)

Browse Source
This commit is contained in:
mdecimus
2023-08-01 19:11:59 +02:00
parent fa2b101931
commit c2e909a09f
11 changed files with 107 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
crates/cli/src/modules/export.rs
View File

@@ -38,6 +38,8 @@ use jmap_client::{
use serde::Serialize;
use tokio::io::AsyncWriteExt;
use crate::modules::RETRY_ATTEMPTS;
use super::{cli::ExportCommands, name_to_id, UnwrapResult};
pub async fn cmd_export(mut client: Client, command: ExportCommands) {
@@ -94,10 +96,22 @@ pub async fn cmd_export(mut client: Client, command: ExportCommands) {
blob_path.push(&blob_id);
futures.push(async move {
let bytes = client
.download(&blob_id)
.await
.unwrap_result("download blob");
let mut retry_count = 0;
let bytes = loop {
match client.download(&blob_id).await {
Ok(bytes) => break bytes,
Err(_) if retry_count < RETRY_ATTEMPTS => {
tokio::time::sleep(std::time::Duration::from_secs(1)).await;
retry_count += 1;
}
result => {
result.unwrap_result("download blob");
return;
}
}
};
tokio::fs::OpenOptions::new()
.create(true)
.write(true)

130
crates/cli/src/modules/import.rs
View File

@@ -46,7 +46,7 @@ use mail_parser::mailbox::{
use serde::de::DeserializeOwned;
use tokio::{fs::File, io::AsyncReadExt};
use crate::modules::{name_to_id, UnwrapResult};
use crate::modules::{name_to_id, UnwrapResult, RETRY_ATTEMPTS};
use super::{
cli::{ImportCommands, MailboxFormat},
@@ -331,43 +331,54 @@ pub async fn cmd_import(mut client: Client, command: ImportCommands) {
pbs.1 += 1;
}
if let Err(err) = client
.email_import(
message.contents,
[mailbox_id.as_ref()],
if !message.flags.is_empty() {
message
.flags
.into_iter()
.map(|f| match f {
maildir::Flag::Passed => "$passed",
maildir::Flag::Replied => "$answered",
maildir::Flag::Seen => "$seen",
maildir::Flag::Trashed => "$deleted",
maildir::Flag::Draft => "$draft",
maildir::Flag::Flagged => "$flagged",
})
.into()
} else {
None
},
if message.internal_date > 0 {
(message.internal_date as i64).into()
} else {
None
},
)
.await
{
failures.lock().unwrap().push(format!(
concat!(
"Failed to import message {} ",
"with identifier '{}': {}"
),
message_num, message.identifier, err
));
} else {
total_imported.fetch_add(1, Ordering::Relaxed);
let mut retry_count = 0;
loop {
match client
.email_import(
message.contents.clone(),
[mailbox_id.as_ref()],
if !message.flags.is_empty() {
message
.flags
.iter()
.map(|f| match f {
maildir::Flag::Passed => "$passed",
maildir::Flag::Replied => "$answered",
maildir::Flag::Seen => "$seen",
maildir::Flag::Trashed => "$deleted",
maildir::Flag::Draft => "$draft",
maildir::Flag::Flagged => "$flagged",
})
.into()
} else {
None
},
if message.internal_date > 0 {
(message.internal_date as i64).into()
} else {
None
},
)
.await
{
Ok(_) => {
total_imported.fetch_add(1, Ordering::Relaxed);
}
Err(_) if retry_count < RETRY_ATTEMPTS => {
retry_count += 1;
continue;
}
Err(err) => {
failures.lock().unwrap().push(format!(
concat!(
"Failed to import message {} ",
"with identifier '{}': {}"
),
message_num, message.identifier, err
));
}
}
break;
}
});
@@ -648,22 +659,33 @@ async fn import_emails(
}
}
if let Err(err) = client
.email_import(
contents,
mailboxes,
if !keywords.is_empty() {
Some(keywords)
} else {
None
},
email.received_at(),
)
.await
{
eprintln!("Failed to import emailId {id}: {err}");
} else {
total_imported.fetch_add(1, Ordering::Relaxed);
let mut retry_count = 0;
loop {
match client
.email_import(
contents.clone(),
mailboxes.clone(),
if !keywords.is_empty() {
Some(keywords.clone())
} else {
None
},
email.received_at(),
)
.await
{
Ok(_) => {
total_imported.fetch_add(1, Ordering::Relaxed);
}
Err(_) if retry_count < RETRY_ATTEMPTS => {
retry_count += 1;
continue;
}
Err(err) => {
eprintln!("Failed to import emailId {id}: {err}");
}
}
break;
}
});

2
crates/cli/src/modules/mod.rs
View File

@@ -38,6 +38,8 @@ pub mod import;
pub mod queue;
pub mod report;
const RETRY_ATTEMPTS: usize = 5;
pub trait UnwrapResult<T> {
fn unwrap_result(self, action: &str) -> T;
}

2
crates/imap/src/op/append.rs
View File

@@ -142,7 +142,7 @@ impl SessionData {
keywords: message.flags.into_iter().map(Keyword::from).collect(),
received_at: message.received_at.map(|d| d as u64),
skip_duplicates: false,
encrypt: true,
encrypt: self.jmap.config.encrypt && self.jmap.config.encrypt_append,
})
.await
{

2
crates/jmap/src/api/config.rs
View File

@@ -138,6 +138,8 @@ impl crate::Config {
principal_allow_lookups: settings
.property("jmap.principal.allow-lookups")?
.unwrap_or(true),
encrypt: settings.property_or_static("jmap.encryption.enable", "true")?,
encrypt_append: settings.property_or_static("jmap.encryption.append", "false")?,
};
config.add_capabilites(settings);
Ok(config)

2
crates/jmap/src/api/http.rs
View File

@@ -237,7 +237,7 @@ pub async fn parse_jmap_request(
_ => (),
}
}
"crypto" => match *req.method() {
"crypto" if jmap.config.encrypt => match *req.method() {
Method::GET => {
return jmap.handle_crypto_update(&mut req).await;
}

2
crates/jmap/src/email/import.rs
View File

@@ -141,7 +141,7 @@ impl JMAP {
keywords: email.keywords,
received_at: email.received_at.map(|r| r.into()),
skip_duplicates: false,
encrypt: true,
encrypt: self.config.encrypt && self.config.encrypt_append,
})
.await
{

2
crates/jmap/src/email/set.rs
View File

@@ -737,7 +737,7 @@ impl JMAP {
keywords,
received_at,
skip_duplicates: false,
encrypt: false,
encrypt: self.config.encrypt && self.config.encrypt_append,
})
.await
{

3
crates/jmap/src/lib.rs
View File

@@ -150,6 +150,9 @@ pub struct Config {
pub oauth_expiry_refresh_token_renew: u64,
pub oauth_max_auth_attempts: u32,
pub encrypt: bool,
pub encrypt_append: bool,
pub principal_allow_lookups: bool,
pub capabilities: BaseCapabilities,

2
crates/jmap/src/services/ingest.rs
View File

@@ -104,7 +104,7 @@ impl JMAP {
keywords: vec![],
received_at: None,
skip_duplicates: true,
encrypt: true,
encrypt: self.config.encrypt,
})
.await
}

2
crates/jmap/src/sieve/ingest.rs
View File

@@ -450,7 +450,7 @@ impl JMAP {
keywords: sieve_message.flags,
received_at: None,
skip_duplicates: true,
encrypt: true,
encrypt: self.config.encrypt,
})
.await
{