apalis_core/
error.rs

1use std::{
2    error::Error as StdError,
3    marker::PhantomData,
4    pin::Pin,
5    task::{Context, Poll},
6    time::Duration,
7};
8use thiserror::Error;
9use tower_service::Service;
10
11/// Convenience type alias
12pub type BoxDynError = Box<dyn StdError + 'static + Send + Sync>;
13/// Execution should be aborted
14/// This signifies that the task should not be retried
15#[derive(Error, Debug)]
16#[error("AbortError: {source}")]
17pub struct AbortError {
18    #[source]
19    source: BoxDynError,
20}
21impl AbortError {
22    /// Create a new abort error
23    pub fn new<E: Into<BoxDynError>>(err: E) -> Self {
24        Self { source: err.into() }
25    }
26}
27
28/// Execution should be retried after a specific duration
29/// This increases the attempts
30#[derive(Error, Debug)]
31#[error("RetryError: {source}")]
32pub struct RetryAfterError {
33    #[source]
34    source: BoxDynError,
35    duration: Duration,
36}
37
38impl RetryAfterError {
39    /// Create a new retry after error
40    pub fn new<E: Into<BoxDynError>>(err: E, duration: Duration) -> Self {
41        Self {
42            source: err.into(),
43            duration,
44        }
45    }
46
47    /// Get the duration after which the task should be retried
48    #[must_use]
49    pub fn get_duration(&self) -> Duration {
50        self.duration
51    }
52}
53
54/// Execution should be deferred, will be retried instantly
55#[derive(Error, Debug)]
56#[error("DeferredError: {source}")]
57pub struct DeferredError {
58    #[source]
59    source: BoxDynError,
60}
61
62impl DeferredError {
63    /// Compose a new Deferred error
64    pub fn new<E: Into<BoxDynError>>(source: E) -> Self {
65        Self {
66            source: source.into(),
67        }
68    }
69}
70
71/// Possible errors that can occur when running a worker.
72#[non_exhaustive]
73#[derive(Error, Debug)]
74pub enum WorkerError {
75    /// An error occurred while polling.
76    #[error("Failed to poll backend: {0}")]
77    BackendError(BoxDynError),
78    /// An error occurred while trying to change the state of the worker.
79    #[error("Failed to handle the new state: {0}")]
80    StateError(#[from] WorkerStateError),
81    /// A worker panicked and the panic was caught.
82    #[error("Worker panicked: {0}")]
83    PanicError(String),
84    /// An error occurred while handling io
85    #[error("IO error: {0}")]
86    IoError(#[from] std::io::Error),
87}
88
89/// Errors related to worker state transitions
90#[derive(Error, Debug)]
91#[non_exhaustive]
92pub enum WorkerStateError {
93    /// Worker not started
94    #[error("Worker not started, did you forget to call worker.start()")]
95    NotStarted,
96    /// Worker already started
97    #[error("Worker already started")]
98    AlreadyStarted,
99    /// Worker is not running
100    #[error("Worker is not running")]
101    NotRunning,
102    /// Worker is not paused
103    #[error("Worker is not paused")]
104    NotPaused,
105    /// Worker is shutting down
106    #[error("Worker is shutting down")]
107    ShuttingDown,
108    /// Invalid state provided
109    #[error("Worker provided with invalid state {0}")]
110    InvalidState(String),
111}
112
113/// A Tower layer for handling and converting service errors into a `BoxDynError`.
114///
115/// The service's error type must implement `std::error::Error`, allowing for flexible
116/// error handling, especially when dealing with trait objects or complex error chains.
117#[derive(Clone, Debug)]
118pub struct ErrorHandlingLayer {
119    _p: PhantomData<()>,
120}
121
122impl ErrorHandlingLayer {
123    /// Create a new ErrorHandlingLayer
124    #[must_use]
125    pub fn new() -> Self {
126        Self { _p: PhantomData }
127    }
128}
129
130impl Default for ErrorHandlingLayer {
131    fn default() -> Self {
132        Self::new()
133    }
134}
135
136impl<S> tower_layer::Layer<S> for ErrorHandlingLayer {
137    type Service = ErrorHandlingService<S>;
138
139    fn layer(&self, service: S) -> Self::Service {
140        ErrorHandlingService { service }
141    }
142}
143
144/// The underlying service
145#[derive(Clone, Debug)]
146pub struct ErrorHandlingService<S> {
147    service: S,
148}
149
150impl<S, Request> Service<Request> for ErrorHandlingService<S>
151where
152    S: Service<Request>,
153    S::Error: std::error::Error + Send + Sync + 'static,
154    S::Future: Send + 'static,
155{
156    type Response = S::Response;
157    type Error = BoxDynError;
158    type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;
159
160    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
161        self.service.poll_ready(cx).map_err(|e| {
162            let boxed_error: BoxDynError = e.into();
163            boxed_error
164        })
165    }
166
167    fn call(&mut self, req: Request) -> Self::Future {
168        let fut = self.service.call(req);
169
170        Box::pin(async move {
171            fut.await.map_err(|e| {
172                let boxed_error: BoxDynError = e.into();
173                boxed_error
174            })
175        })
176    }
177}
apalis_core/error.rs

apalis_core/
error.rs
