Job Failure

CHANGELOG.md

@@ -10,6 +10,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Added
 
 - Support Faktory's `MUTATE` API ([#87])
+- Make `Failure` struct public ([#89])
 
 ### Changed
 
@@ -22,6 +23,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Security
 
 [#87]: https://github.com/jonhoo/faktory-rs/pull/87
+[#89]: https://github.com/jonhoo/faktory-rs/pull/89
 
 ## [0.13.0] - 2024-10-27
 

src/lib.rs

@@ -106,8 +106,8 @@ mod worker;
 pub use crate::error::Error;
 
 pub use crate::proto::{
-    Client, Connection, DataSnapshot, FaktoryState, Job, JobBuilder, JobId, MutationFilter,
-    MutationFilterBuilder, MutationTarget, Reconnect, ServerSnapshot, WorkerId,
+    Client, Connection, DataSnapshot, Failure, FaktoryState, Job, JobBuilder, JobId,
+    MutationFilter, MutationFilterBuilder, MutationTarget, Reconnect, ServerSnapshot, WorkerId,
 };
 
 pub use crate::worker::{JobRunner, StopDetails, StopReason, Worker, WorkerBuilder};

src/proto/mod.rs

@@ -12,8 +12,8 @@ pub use client::{Client, Connection};
 mod single;
 
 pub use single::{
-    DataSnapshot, FaktoryState, Job, JobBuilder, JobId, MutationFilter, MutationFilterBuilder,
-    MutationTarget, ServerSnapshot, WorkerId,
+    DataSnapshot, Failure, FaktoryState, Job, JobBuilder, JobId, MutationFilter,
+    MutationFilterBuilder, MutationTarget, ServerSnapshot, WorkerId,
 };
 
 pub(crate) use single::{Ack, Fail, Info, Push, PushBulk, QueueAction, QueueControl};

src/proto/single/mod.rs

@@ -124,6 +124,8 @@ pub struct Job {
     /// Defaults to 25.
     #[serde(skip_serializing_if = "Option::is_none")]
     #[builder(default = "Some(JOB_DEFAULT_RETRY_COUNT)")]
+    // TODO: this should probably be a usize, see Failure::retry_count
+    // TODO: and Failure::retry_remaining. This can go to 0.14 release
     pub retry: Option<isize>,
 
     /// The priority of this job from 1-9 (9 is highest).
@@ -206,19 +208,45 @@ impl JobBuilder {
     }
 }
 
+/// Details on a job's failure.
 #[derive(Serialize, Deserialize, Debug, Clone)]
+#[non_exhaustive]
 pub struct Failure {
-    retry_count: usize,
-    failed_at: String,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    next_at: Option<String>,
+    /// Number of times this job has been retried.
+    pub retry_count: usize,
+
+    /// Number of remaining retry attempts.
+    ///
+    /// This is the difference between how many times this job
+    /// _can_ be retried (see [`Job::retry`]) and the number of retry
+    /// attempts that have already been made (see [`Failure::retry_count`]).
+    #[serde(rename = "remaining")]
+    pub retry_remaining: usize,
+
+    /// Last time this job failed.
+    pub failed_at: DateTime<Utc>,
+
+    /// When this job will be retried.
+    ///
+    /// This will be `None` if there are no retry
+    /// attempts (see [`Failure::retry_remaining`]) left.
     #[serde(skip_serializing_if = "Option::is_none")]
-    message: Option<String>,
+    pub next_at: Option<DateTime<Utc>>,
+
+    /// Error message, if any.
     #[serde(skip_serializing_if = "Option::is_none")]
+    pub message: Option<String>,
+
+    // This is Some("unknown") most of the time, and we are not making
+    // it public for now, see discussion:
+    // https://github.com/jonhoo/faktory-rs/pull/89#discussion_r1899423130
+    /// Error kind, if known.
     #[serde(rename = "errtype")]
-    kind: Option<String>,
+    pub(crate) kind: Option<String>,
+
+    /// Stack trace from last failure, if any.
     #[serde(skip_serializing_if = "Option::is_none")]
-    backtrace: Option<Vec<String>>,
+    pub backtrace: Option<Vec<String>>,
 }
 
 impl Job {
@@ -263,8 +291,8 @@ impl Job {
     }
 
     /// Data about this job's most recent failure.
-    pub fn failure(&self) -> &Option<Failure> {
-        &self.failure
+    pub fn failure(&self) -> Option<&Failure> {
+        self.failure.as_ref()
     }
 }
 

src/worker/mod.rs

@@ -360,7 +360,22 @@
                 let fail = match e {
                     Failed::BadJobType(jt) => Fail::generic(jid, format!("No handler for {}", jt)),
                     Failed::Application(e) => Fail::generic_with_backtrace(jid, e),
-                    Failed::HandlerPanic(e) => Fail::generic_with_backtrace(jid, e),
+                    Failed::HandlerPanic(e) => {
+                        if e.is_cancelled() {
+                            Fail::generic(jid, "job processing was cancelled")
+                        } else if e.is_panic() {
+                            let panic_obj = e.into_panic();
+                            if panic_obj.is::<String>() {
+                                Fail::generic(jid, *panic_obj.downcast::<String>().unwrap())
+                            } else if panic_obj.is::<&'static str>() {
+                                Fail::generic(jid, *panic_obj.downcast::<&'static str>().unwrap())
+                            } else {
+                                Fail::generic(jid, "job processing panicked")
+                            }
+                        } else {
+                            Fail::generic_with_backtrace(jid, e)
+                        }
+                    }
                 };
                 self.worker_states.register_failure(worker, fail.clone());
                 self.c.issue(&fail).await?.read_ok().await?;