Add type Error to Job

Cargo.toml

@@ -21,6 +21,7 @@ members = [
   "jobs-sled",
   "jobs-tokio",
   "examples/basic-example",
+  "examples/error-example",
   "examples/long-example",
   "examples/managed-example",
   "examples/metrics-example",

examples/basic-example/src/main.rs

@@ -84,6 +84,7 @@ impl MyJob {
 
 impl Job for MyJob {
     type State = MyState;
+    type Error = Error;
     type Future = Ready<Result<(), Error>>;
     type Spawner = Spawner;
 

examples/error-example/.gitignore

@@ -0,0 +1 @@
+/my-sled-db

examples/error-example/Cargo.toml

@@ -0,0 +1,20 @@
+[package]
+name = "error-example"
+version = "0.1.0"
+authors = ["asonix <asonix@asonix.dog>"]
+edition = "2021"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+actix-rt = "2.0.0"
+anyhow = "1.0"
+background-jobs = { version = "0.17.0", path = "../..", features = [
+  "error-logging",
+  "sled",
+] }
+time = "0.3"
+tracing = "0.1"
+tracing-subscriber = { version = "0.3", features = ["env-filter", "fmt"] }
+serde = { version = "1.0", features = ["derive"] }
+sled = "0.34"

examples/error-example/src/main.rs

@@ -0,0 +1,132 @@
+use actix_rt::Arbiter;
+use anyhow::Error;
+use background_jobs::{actix::WorkerConfig, sled::Storage, Job, MaxRetries};
+use std::{
+    future::{ready, Ready},
+    time::{Duration, SystemTime},
+};
+use tracing::info;
+use tracing_subscriber::EnvFilter;
+
+const DEFAULT_QUEUE: &str = "default";
+
+#[derive(Clone, Debug)]
+pub struct MyState {
+    pub app_name: String,
+}
+
+#[derive(Clone, Debug, serde::Deserialize, serde::Serialize)]
+pub struct MyJob {
+    some_usize: usize,
+    other_usize: usize,
+}
+
+#[derive(Clone, Debug, serde::Deserialize, serde::Serialize)]
+pub struct ErroringJob;
+
+#[actix_rt::main]
+async fn main() -> Result<(), Error> {
+    let env_filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info"));
+
+    tracing_subscriber::fmt::fmt()
+        .with_env_filter(env_filter)
+        .init();
+
+    // Set up our Storage
+    let db = sled::Config::new().temporary(true).open()?;
+    let storage = Storage::new(db)?;
+
+    let arbiter = Arbiter::new();
+
+    // Configure and start our workers
+    let queue_handle =
+        WorkerConfig::new_in_arbiter(arbiter.handle(), storage, |_| MyState::new("My App"))
+            .register::<ErroringJob>()
+            .register::<MyJob>()
+            .set_worker_count(DEFAULT_QUEUE, 16)
+            .start();
+
+    // Queue some panicking job
+    for _ in 0..32 {
+        queue_handle.queue(ErroringJob).await?;
+    }
+
+    // Queue our jobs
+    queue_handle.queue(MyJob::new(1, 2)).await?;
+    queue_handle.queue(MyJob::new(3, 4)).await?;
+    queue_handle.queue(MyJob::new(5, 6)).await?;
+    queue_handle
+        .schedule(MyJob::new(7, 8), SystemTime::now() + Duration::from_secs(2))
+        .await?;
+
+    // Block on Actix
+    actix_rt::signal::ctrl_c().await?;
+
+    arbiter.stop();
+    let _ = arbiter.join();
+
+    Ok(())
+}
+
+impl MyState {
+    pub fn new(app_name: &str) -> Self {
+        MyState {
+            app_name: app_name.to_owned(),
+        }
+    }
+}
+
+impl MyJob {
+    pub fn new(some_usize: usize, other_usize: usize) -> Self {
+        MyJob {
+            some_usize,
+            other_usize,
+        }
+    }
+}
+
+impl Job for MyJob {
+    type State = MyState;
+    type Error = Error;
+    type Future = Ready<Result<(), Error>>;
+
+    // The name of the job. It is super important that each job has a unique name,
+    // because otherwise one job will overwrite another job when they're being
+    // registered.
+    const NAME: &'static str = "MyJob";
+
+    // The queue that this processor belongs to
+    //
+    // Workers have the option to subscribe to specific queues, so this is important to
+    // determine which worker will call the processor
+    //
+    // Jobs can optionally override the queue they're spawned on
+    const QUEUE: &'static str = DEFAULT_QUEUE;
+
+    // The number of times background-jobs should try to retry a job before giving up
+    //
+    // Jobs can optionally override this value
+    const MAX_RETRIES: MaxRetries = MaxRetries::Count(1);
+
+    fn run(self, state: MyState) -> Self::Future {
+        info!("{}: args, {:?}", state.app_name, self);
+
+        ready(Ok(()))
+    }
+}
+
+impl Job for ErroringJob {
+    type State = MyState;
+    type Error = Error;
+    type Future = Ready<Result<(), Error>>;
+
+    const NAME: &'static str = "ErroringJob";
+
+    const QUEUE: &'static str = DEFAULT_QUEUE;
+
+    const MAX_RETRIES: MaxRetries = MaxRetries::Count(0);
+
+    fn run(self, _: MyState) -> Self::Future {
+        ready(Err(anyhow::anyhow!("boom")))
+    }
+}

examples/long-example/src/main.rs

@@ -90,6 +90,7 @@ impl MyJob {
 
 impl Job for MyJob {
     type State = MyState;
+    type Error = Error;
     type Future = Ready<Result<(), Error>>;
     type Spawner = Spawner;
 
@@ -120,6 +121,7 @@ impl Job for MyJob {
 
 impl Job for LongJob {
     type State = MyState;
+    type Error = Error;
     type Future = Pin<Box<dyn Future<Output = Result<(), Error>>>>;
     type Spawner = Spawner;
 

examples/managed-example/src/main.rs

@@ -102,6 +102,7 @@ impl MyJob {
 
 impl Job for MyJob {
     type State = MyState;
+    type Error = Error;
     type Future = Ready<Result<(), Error>>;
     type Spawner = Spawner;
 
@@ -132,6 +133,7 @@ impl Job for MyJob {
 
 impl Job for StopJob {
     type State = MyState;
+    type Error = Error;
     type Future = Ready<Result<(), Error>>;
     type Spawner = Spawner;
 

examples/metrics-example/src/main.rs

@@ -89,6 +89,7 @@ impl MyJob {
 
 impl Job for MyJob {
     type State = MyState;
+    type Error = Error;
     type Future = Pin<Box<dyn Future<Output = Result<(), Error>> + 'static>>;
     type Spawner = Spawner;
 

examples/panic-example/src/main.rs

@@ -83,6 +83,7 @@ impl MyJob {
 
 impl Job for MyJob {
     type State = MyState;
+    type Error = Error;
     type Future = Ready<Result<(), Error>>;
 
     // The name of the job. It is super important that each job has a unique name,
@@ -112,6 +113,7 @@ impl Job for MyJob {
 
 impl Job for PanickingJob {
     type State = MyState;
+    type Error = Error;
     type Future = Ready<Result<(), Error>>;
 
     const NAME: &'static str = "PanickingJob";

examples/postgres-example/src/main.rs

@@ -90,6 +90,7 @@ impl MyJob {
 
 impl Job for MyJob {
     type State = MyState;
+    type Error = anyhow::Error;
     type Future = Ready<anyhow::Result<()>>;
     type Spawner = Spawner;
 

examples/tokio-example/src/main.rs

@@ -79,6 +79,7 @@ impl MyJob {
 
 impl Job for MyJob {
     type State = MyState;
+    type Error = Error;
     type Future = Ready<Result<(), Error>>;
 
     // The name of the job. It is super important that each job has a unique name,

jobs-core/Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 name = "background-jobs-core"
 description = "Core types for implementing an asynchronous jobs processor"
-version = "0.17.0"
+version = "0.17.1"
 license = "AGPL-3.0"
 authors = ["asonix <asonix@asonix.dog>"]
 repository = "https://git.asonix.dog/asonix/background-jobs"

jobs-core/src/job.rs

@@ -48,8 +48,11 @@ pub trait Job: Serialize + DeserializeOwned + 'static {
     /// The application state provided to this job at runtime.
     type State: Clone + 'static;
 
+    /// The error type this job returns
+    type Error: Into<Box<dyn std::error::Error>>;
+
     /// The future returned by this job
-    type Future: Future<Output = Result<(), Error>> + Send;
+    type Future: Future<Output = Result<(), Self::Error>> + Send;
 
     /// The name of the job
     ///
@@ -176,9 +179,9 @@ where
         let (fut, span) = res?;
 
         if let Some(span) = span {
-            fut.instrument(span).await?;
+            fut.instrument(span).await.map_err(Into::into)?;
         } else {
-            fut.await?;
+            fut.await.map_err(Into::into)?;
         }
 
         Ok(())

jobs-core/src/lib.rs

@@ -6,8 +6,6 @@
 //! This crate shouldn't be depended on directly, except in the case of implementing a custom jobs
 //! processor. For a default solution based on Actix and Sled, look at the `background-jobs` crate.
 
-use anyhow::Error;
-
 mod catch_unwind;
 mod job;
 mod job_info;
@@ -28,8 +26,8 @@ pub use unsend_job::{JoinError, UnsendJob, UnsendSpawner};
 /// The error type returned by the `process` method
 pub enum JobError {
     /// Some error occurred while processing the job
-    #[error("Error performing job: {0}")]
-    Processing(#[from] Error),
+    #[error("{0}")]
+    Processing(#[from] Box<dyn std::error::Error>),
 
     /// Creating a `Job` type from the provided `serde_json::Value` failed
     #[error("Could not make JSON value from arguments")]

jobs-core/src/unsend_job.rs

@@ -1,5 +1,4 @@
 use crate::{Backoff, Job, MaxRetries};
-use anyhow::Error;
 use serde::{de::DeserializeOwned, ser::Serialize};
 use std::{
     fmt::Debug,
@@ -45,10 +44,13 @@ pub trait UnsendJob: Serialize + DeserializeOwned + 'static {
     /// The application state provided to this job at runtime.
     type State: Clone + 'static;
 
+    /// The error type this job returns
+    type Error: Into<Box<dyn std::error::Error>> + Send;
+
     /// The future returned by this job
     ///
     /// Importantly, this Future does not require Send
-    type Future: Future<Output = Result<(), Error>>;
+    type Future: Future<Output = Result<(), Self::Error>>;
 
     /// The spawner type that will be used to spawn the unsend future
     type Spawner: UnsendSpawner;
@@ -148,7 +150,8 @@ where
     T: UnsendJob,
 {
     type State = T::State;
-    type Future = UnwrapFuture<<T::Spawner as UnsendSpawner>::Handle<Result<(), Error>>>;
+    type Error = T::Error;
+    type Future = UnwrapFuture<<T::Spawner as UnsendSpawner>::Handle<Result<(), Self::Error>>>;
 
     const NAME: &'static str = <Self as UnsendJob>::NAME;
     const QUEUE: &'static str = <Self as UnsendJob>::QUEUE;