diff --git a/src/errors.rs b/src/errors.rs index ed5e905..1b24165 100644 --- a/src/errors.rs +++ b/src/errors.rs @@ -95,6 +95,18 @@ impl HumanizableError for reqwest::header::InvalidHeaderValue { } } +impl HumanizableError for tokio::task::JoinError { + fn to_human_error(self) -> human_errors::Error { + human_errors::wrap_system( + self, + "A backup task terminated unexpectedly before it could complete.", + &[ + "This is likely a bug in github-backup, please report it to us on GitHub along with the error details above.", + ], + ) + } +} + #[derive(Debug)] pub struct ResponseError { pub status_code: StatusCode, diff --git a/src/main.rs b/src/main.rs index 2766a97..9ffbcc0 100644 --- a/src/main.rs +++ b/src/main.rs @@ -82,7 +82,7 @@ async fn run(args: Args, session: &Session) -> Result<(), Error> { .as_ref() .and_then(|s| s.find_next_occurrence(&chrono::Utc::now(), false).ok()); - let handler = LoggingPairingHandler::new(&session); + let handler = LoggingPairingHandler::new(session); pinger.on_start().await; @@ -178,7 +178,7 @@ impl PairingHandler for LoggingPairingHandler<'_> { if error.is(human_errors::Kind::System) { let info = tracing_batteries::ErrorInfo::new(&error) - .with_metadata("policy.kind", format!("{}", policy.kind)); + .with_metadata("policy.kind", policy.kind.to_string()); self.session.record_custom_error(info); } else { @@ -197,7 +197,7 @@ impl PairingHandler for LoggingPairingHandler<'_> { self.session.record_event( "policy::run", [ - ("policy.kind", format!("{}", policy.kind)), + ("policy.kind", policy.kind.to_string()), ("stats.new", summary.new.to_string()), ("stats.unchanged", summary.unchanged.to_string()), ("stats.updated", summary.updated.to_string()), diff --git a/src/pairing.rs b/src/pairing.rs index de36080..144ef44 100644 --- a/src/pairing.rs +++ b/src/pairing.rs @@ -8,6 +8,7 @@ use tracing_batteries::prelude::*; use crate::{ BackupEntity, BackupPolicy, BackupSource, engines::{BackupEngine, BackupState}, + errors::HumanizableError as _, }; pub struct Pairing, T: BackupEngine> { @@ -132,7 +133,11 @@ impl< for to in policy.to.iter() { while join_set.len() >= self.concurrency_limit { debug!("Reached concurrency limit of {}, waiting for a task to complete", self.concurrency_limit); - yield join_set.join_next().await.unwrap().unwrap(); + match join_set.join_next().await { + Some(Ok(result)) => yield result, + Some(Err(e)) => yield Err(e.to_human_error()), + None => break, + } } let span = tracing_batteries::prelude::info_span!(parent: &span, "backup.step", item=%entity, target=%to); @@ -147,7 +152,10 @@ impl< } while let Some(fut) = join_set.join_next().await { - yield fut.unwrap(); + match fut { + Ok(result) => yield result, + Err(e) => yield Err(e.to_human_error()), + } } } } diff --git a/src/sources/github_gist.rs b/src/sources/github_gist.rs index b7a60d7..a57802e 100644 --- a/src/sources/github_gist.rs +++ b/src/sources/github_gist.rs @@ -38,35 +38,34 @@ impl BackupSource for GitHubGistSource { policy: &'a BackupPolicy, cancel: &'a AtomicBool, ) -> impl Stream> + 'a { - let target: GitHubRepoSourceKind = policy.from.as_str().parse().unwrap(); - - let url = format!( - "{}/{}?{}", - policy - .properties - .get("api_url") - .unwrap_or(&"https://api.github.com".to_string()) - .trim_end_matches('/'), - target.api_endpoint(GitHubArtifactKind::Gist), - policy.properties.get("query").unwrap_or(&"".to_string()) - ) - .trim_end_matches('?') - .to_string(); - - tracing_batteries::prelude::debug!("Calling {} to fetch gists", &url); - - let refspecs = policy - .properties - .get("refspecs") - .map(|r| r.split(',').map(|r| r.to_string()).collect::>()); - - let recovery_mode: RecoveryMode = policy - .properties - .get("recovery") - .map(|mode| mode.parse().unwrap()) - .unwrap_or_default(); - async_stream::try_stream! { + let target: GitHubRepoSourceKind = policy.from.as_str().parse()?; + + let url = format!( + "{}/{}?{}", + policy + .properties + .get("api_url") + .unwrap_or(&"https://api.github.com".to_string()) + .trim_end_matches('/'), + target.api_endpoint(GitHubArtifactKind::Gist), + policy.properties.get("query").unwrap_or(&"".to_string()) + ) + .trim_end_matches('?') + .to_string(); + + tracing_batteries::prelude::debug!("Calling {} to fetch gists", &url); + + let refspecs = policy + .properties + .get("refspecs") + .map(|r| r.split(',').map(|r| r.to_string()).collect::>()); + + let recovery_mode: RecoveryMode = match policy.properties.get("recovery") { + Some(mode) => mode.parse()?, + None => RecoveryMode::default(), + }; + if matches!(target, GitHubRepoSourceKind::Gist(_)) { let gist: GitHubGist = self.client.get(&url, &policy.credentials, cancel).await?; yield GitRepo::new( diff --git a/src/sources/github_releases.rs b/src/sources/github_releases.rs index 59129f7..a17e12f 100644 --- a/src/sources/github_releases.rs +++ b/src/sources/github_releases.rs @@ -39,12 +39,13 @@ impl GitHubReleasesSource { let releases_url = format!("{}/releases", repo.url); for await release in self.client.get_paginated(&releases_url, &policy.credentials, cancel) { - if let Err(e) = release { - yield Err(e); - continue; - } - - let release: GitHubRelease = release.unwrap(); + let release: GitHubRelease = match release { + Ok(release) => release, + Err(e) => { + yield Err(e); + continue; + } + }; let mut entity = Release::new( format!("{}/{}", &repo.full_name, &release.tag_name), @@ -170,21 +171,27 @@ impl BackupSource for GitHubReleasesSource { policy: &'a BackupPolicy, cancel: &'a AtomicBool, ) -> impl Stream> + 'a { - let target: GitHubRepoSourceKind = policy.from.as_str().parse().unwrap(); - let url = format!( - "{}/{}?{}", - policy - .properties - .get("api_url") - .unwrap_or(&"https://api.github.com".to_string()) - .trim_end_matches('/'), - target.api_endpoint(GitHubArtifactKind::Release), - policy.properties.get("query").unwrap_or(&"".to_string()) - ) - .trim_end_matches('?') - .to_string(); - async_stream::stream! { + let target: GitHubRepoSourceKind = match policy.from.as_str().parse() { + Ok(target) => target, + Err(e) => { + yield Err(e); + return; + } + }; + let url = format!( + "{}/{}?{}", + policy + .properties + .get("api_url") + .unwrap_or(&"https://api.github.com".to_string()) + .trim_end_matches('/'), + target.api_endpoint(GitHubArtifactKind::Release), + policy.properties.get("query").unwrap_or(&"".to_string()) + ) + .trim_end_matches('?') + .to_string(); + if matches!(target, GitHubRepoSourceKind::Repo(_)) { let repo: GitHubRepo = self.client.get(&url, &policy.credentials, cancel).await?; @@ -193,12 +200,13 @@ impl BackupSource for GitHubReleasesSource { } } else { for await repo in self.client.get_paginated(&url, &policy.credentials, cancel) { - if let Err(e) = repo { - yield Err(e); - continue; - } - - let repo: GitHubRepo = repo.unwrap(); + let repo: GitHubRepo = match repo { + Ok(repo) => repo, + Err(e) => { + yield Err(e); + continue; + } + }; for await file in self.load_releases(policy, &repo, cancel) { yield file; diff --git a/src/sources/github_repo.rs b/src/sources/github_repo.rs index e060b76..083084a 100644 --- a/src/sources/github_repo.rs +++ b/src/sources/github_repo.rs @@ -38,34 +38,34 @@ impl BackupSource for GitHubRepoSource { policy: &'a BackupPolicy, cancel: &'a AtomicBool, ) -> impl Stream> + 'a { - let target: GitHubRepoSourceKind = policy.from.as_str().parse().unwrap(); - let url = format!( - "{}/{}?{}", - policy - .properties - .get("api_url") - .unwrap_or(&"https://api.github.com".to_string()) - .trim_end_matches('/'), - target.api_endpoint(GitHubArtifactKind::Repo), - policy.properties.get("query").unwrap_or(&"".to_string()) - ) - .trim_end_matches('?') - .to_string(); - - tracing_batteries::prelude::debug!("Calling {} to fetch repos", &url); - - let refspecs = policy - .properties - .get("refspecs") - .map(|r| r.split(',').map(|r| r.to_string()).collect::>()); - - let recovery_mode: RecoveryMode = policy - .properties - .get("recovery") - .map(|mode| mode.parse().unwrap()) - .unwrap_or_default(); - async_stream::try_stream! { + let target: GitHubRepoSourceKind = policy.from.as_str().parse()?; + + let url = format!( + "{}/{}?{}", + policy + .properties + .get("api_url") + .unwrap_or(&"https://api.github.com".to_string()) + .trim_end_matches('/'), + target.api_endpoint(GitHubArtifactKind::Repo), + policy.properties.get("query").unwrap_or(&"".to_string()) + ) + .trim_end_matches('?') + .to_string(); + + tracing_batteries::prelude::debug!("Calling {} to fetch repos", &url); + + let refspecs = policy + .properties + .get("refspecs") + .map(|r| r.split(',').map(|r| r.to_string()).collect::>()); + + let recovery_mode: RecoveryMode = match policy.properties.get("recovery") { + Some(mode) => mode.parse()?, + None => RecoveryMode::default(), + }; + if matches!(target, GitHubRepoSourceKind::Repo(_)) { let repo: GitHubRepo = self.client.get(&url, &policy.credentials, cancel).await?; yield GitRepo::new(