apalis_core/backend/
finalize.rs

1use std::{
2    fmt::Debug,
3    marker::PhantomData,
4    pin::Pin,
5    task::{Context, Poll},
6};
7
8use futures_sink::Sink;
9use futures_util::SinkExt;
10
11use crate::{
12    backend::{codec::Codec, *},
13    delegate_sink,
14    task::Task,
15    worker::context::WorkerContext,
16};
17
18/// Error type that combines backend errors and service errors
19#[derive(Debug, thiserror::Error)]
20#[non_exhaustive]
21pub enum DurableError<PollError, CodecError> {
22    /// Error originating from the the poll
23    #[error("PollNextError: {0}")]
24    PollNextError(PollError),
25    /// Error originating from the decoding of the task
26    #[error("Task decoding error: {0}")]
27    CodecError(CodecError),
28}
29
30/// Finalizes an intermediate backend configuration into a concrete backend.
31///
32/// `B` represents the backend being built, while `Args` is the type of
33/// arguments the finalized backend exposes to workers.
34///
35/// Implementations of this trait determine how the final backend is composed.
36/// For example, [`Ephemeral`] returns the backend unchanged, while
37/// [`Durable`] wraps it in [`DurableBackend`] to decode persisted task
38/// arguments.
39pub trait FinalizeBackend<B, Args> {
40    /// The concrete backend produced by finalization.
41    type Backend;
42
43    /// Finalizes the current backend.
44    fn finalize(current: B) -> Self::Backend;
45}
46
47/// Finalization mode for an ephemeral backend.
48///
49/// An ephemeral backend is used as-is and does not perform any additional
50/// persistence-related wrapping.
51#[derive(Debug, Clone, Copy, Default)]
52pub struct Ephemeral;
53
54impl<B, Args> FinalizeBackend<B, Args> for Ephemeral {
55    type Backend = B;
56
57    fn finalize(current: B) -> Self::Backend {
58        current
59    }
60}
61
62/// Finalization mode for a durable backend.
63///
64/// Durable backends store task arguments in their wire format and decode them
65/// when tasks are retrieved.
66#[derive(Debug, Clone, Copy, Default)]
67pub struct Durable;
68
69impl<B, Args> FinalizeBackend<B, Args> for Durable
70where
71    B: BackendConfig + WireFormatBackend,
72{
73    type Backend = DurableBackend<B, Args>;
74
75    fn finalize(current: B) -> Self::Backend {
76        DurableBackend {
77            backend: current,
78            _marker: PhantomData,
79        }
80    }
81}
82
83/// A backend wrapper that decodes persisted task arguments.
84///
85/// `DurableBackend` wraps a [`WireFormatBackend`] whose task arguments are
86/// stored in a compact wire representation. The configured [`Codec`] is used
87/// to decode those arguments back into `Args` as tasks are polled.
88#[derive(Debug)]
89pub struct DurableBackend<B, Args> {
90    backend: B,
91    _marker: PhantomData<Args>,
92}
93
94impl<B: Clone, Args> Clone for DurableBackend<B, Args> {
95    fn clone(&self) -> Self {
96        Self {
97            backend: self.backend.clone(),
98            _marker: PhantomData,
99        }
100    }
101}
102
103impl<B, Compact, Args, Err> Backend for DurableBackend<B, Args>
104where
105    B: Backend<Task = Task<Compact>> + WireFormatBackend,
106    B::Codec: Codec<Args, Compact = Compact, Error = Err>,
107    Err: std::error::Error + Sync + Send + 'static,
108{
109    type Task = Task<Args>;
110    type Error = DurableError<B::Error, Err>;
111
112    fn poll_ready(
113        &mut self,
114        cx: &mut Context<'_>,
115        worker: &WorkerContext,
116    ) -> Poll<Result<(), Self::Error>> {
117        self.backend
118            .poll_ready(cx, worker)
119            .map_err(|e| DurableError::PollNextError(e))
120    }
121
122    fn poll_next(
123        &mut self,
124        cx: &mut Context<'_>,
125        worker: &WorkerContext,
126    ) -> Poll<Option<Result<Self::Task, Self::Error>>> {
127        self.backend.poll_next(cx, worker).map(|s| match s {
128            Some(Ok(task)) => {
129                let codec = self.codec();
130
131                let task = task
132                    .try_map_args(|s| codec.decode(&s))
133                    .map_err(DurableError::CodecError);
134
135                Some(task)
136            }
137            Some(Err(e)) => Some(Err(DurableError::PollNextError(e))),
138            None => None,
139        })
140    }
141
142    fn poll_close(
143        &mut self,
144        cx: &mut Context<'_>,
145        worker: &WorkerContext,
146    ) -> Poll<Result<(), Self::Error>> {
147        self.backend
148            .poll_close(cx, worker)
149            .map_err(|e| DurableError::PollNextError(e))
150    }
151}
152
153impl<B, Args> WireFormatBackend for DurableBackend<B, Args>
154where
155    B: WireFormatBackend + Backend,
156{
157    type Compact = B::Compact;
158    type Codec = B::Codec;
159
160    fn codec(&self) -> &Self::Codec {
161        self.backend.codec()
162    }
163}
164
165impl<B, Args> BackendConfig for DurableBackend<B, Args>
166where
167    B: BackendConfig,
168{
169    type Id = B::Id;
170    type Args = Args;
171    type Kind = Durable;
172    type Config = B::Config;
173    type Layer = B::Layer;
174
175    fn config(&self) -> &Self::Config {
176        self.backend.config()
177    }
178
179    fn middleware(&mut self, worker: &mut WorkerContext) -> Self::Layer {
180        self.backend.middleware(worker)
181    }
182}
183
184delegate_sink!(DurableBackend<B, Args>, backend);
185
186/// Finalization mode for a durable backend.
187///
188/// Durable backends store task arguments in their wire format and decode them
189/// when tasks are retrieved.
190#[derive(Debug, Clone, Copy, Default)]
191pub struct DurableCompact;
192
193impl<B, Args> FinalizeBackend<B, Args> for DurableCompact
194where
195    B: BackendConfig + WireFormatBackend,
196{
197    type Backend = DurableBackend<B, Args>;
198
199    fn finalize(current: B) -> Self::Backend {
200        DurableBackend {
201            backend: current,
202            _marker: PhantomData,
203        }
204    }
205}
apalis_core/backend/finalize.rs

apalis_core/backend/
finalize.rs
