refactor: improve file storage adding csv functionality

apalis-core/src/backend/impls/memory.rs

@@ -39,6 +39,7 @@
 //! - [`WorkerContext`]
 use crate::backend::BackendExt;
 use crate::backend::codec::IdentityCodec;
+use crate::error::BoxDynError;
 use crate::features_table;
 use crate::task::metadata::{MetadataExt, MetadataStore};
 use crate::{
@@ -136,14 +137,25 @@ impl MetadataExt for MemoryContext {
     }
 }
 
+/// Error type for MemoryStorage operations
+#[derive(Debug, thiserror::Error)]
+pub enum MemoryStorageError {
+    /// Error occurred while sending a task to the in-memory channel
+    #[error("Failed to send task: {0}")]
+    SendError(#[from] SendError),
+    /// Error occurred while flushing the in-memory channel
+    #[error("Failed to add task to storage: {0}")]
+    Other(BoxDynError),
+}
+
 impl<Args: Send + 'static> MemoryStorage<Args, MemoryContext> {
     /// Create a new in-memory storage
     #[must_use]
     pub fn new() -> Self {
         let (sender, receiver) = unbounded();
-        let sender = Box::new(sender)
+        let sender = Box::new(sender.sink_map_err(|e| e.into()))
             as Box<
-                dyn Sink<Task<Args, MemoryContext, RandomId>, Error = SendError>
+                dyn Sink<Task<Args, MemoryContext, RandomId>, Error = MemoryStorageError>
                     + Send
                     + Sync
                     + Unpin,
@@ -167,7 +179,7 @@ impl<Args: Send + 'static, Ctx> MemoryStorage<Args, Ctx> {
 }
 
 impl<Args, Ctx> Sink<Task<Args, Ctx, RandomId>> for MemoryStorage<Args, Ctx> {
-    type Error = SendError;
+    type Error = MemoryStorageError;
 
     fn poll_ready(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
         self.as_mut().sender.poll_ready_unpin(cx)
@@ -191,7 +203,13 @@ impl<Args, Ctx> Sink<Task<Args, Ctx, RandomId>> for MemoryStorage<Args, Ctx> {
 
 type ArcMemorySink<Args, Ctx = MemoryContext> = Arc<
     Mutex<
-        Box<dyn Sink<Task<Args, Ctx, RandomId>, Error = SendError> + Send + Sync + Unpin + 'static>,
+        Box<
+            dyn Sink<Task<Args, Ctx, RandomId>, Error = MemoryStorageError>
+                + Send
+                + Sync
+                + Unpin
+                + 'static,
+        >,
     >,
 >;
 
@@ -232,7 +250,7 @@ impl<Args, Ctx> Clone for MemorySink<Args, Ctx> {
 }
 
 impl<Args, Ctx> Sink<Task<Args, Ctx, RandomId>> for MemorySink<Args, Ctx> {
-    type Error = SendError;
+    type Error = MemoryStorageError;
 
     fn poll_ready(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
         let mut lock = ready!(self.inner.lock().poll_unpin(cx));
@@ -301,8 +319,8 @@ impl<Args: 'static + Clone + Send, Ctx: 'static + Default> Backend for MemorySto
 
     type Context = Ctx;
 
-    type Error = SendError;
-    type Stream = TaskStream<Task<Args, Ctx, RandomId>, SendError>;
+    type Error = MemoryStorageError;
+    type Stream = TaskStream<Task<Args, Ctx, RandomId>, MemoryStorageError>;
     type Layer = Identity;
     type Beat = BoxStream<'static, Result<(), Self::Error>>;
 

apalis-core/src/task/metadata.rs

@@ -14,6 +14,7 @@ use crate::task::Task;
 use crate::task_fn::FromRequest;
 use std::collections::HashMap;
 use std::convert::Infallible;
+#[cfg(feature = "tracing")]
 use std::fmt;
 use std::ops::Deref;
 #[cfg(feature = "tracing")]
@@ -452,9 +453,16 @@ impl MetadataStore {
     }
 
     /// Get a typed metadata entry.
+    #[must_use = "Extracted metadata should be used or handled"]
     pub fn extract_as<M: Metadata>(&self) -> Result<M, M::Error> {
         M::extract(self)
     }
+
+    /// Create a `MetadataStore` from a `HashMap<String, String>`.
+    #[must_use]
+    pub fn from_map(map: HashMap<String, String>) -> Self {
+        Self(map)
+    }
 }
 
 /// Implemented by types that can be stored as metadata.

apalis-core/src/task/task_id.rs

@@ -140,7 +140,7 @@ mod random_id {
         let count = COUNTER.fetch_add(1, Ordering::Relaxed);
         let rand_part = encode_base64(xorshift64(timestamp ^ count), RANDOM_LEN);
 
-        format!("{time_str}{rand_part}")
+        format!("{time_str}{rand_part}{count}")
     }
 
     /// Returns current time in milliseconds since UNIX epoch.

apalis-core/src/worker/builder.rs

@@ -196,12 +196,10 @@ where
     B: Backend<Args = Args, Context = Ctx>,
 {
     /// Consumes the builder and a service to construct the final worker
-    pub fn build<W: IntoWorkerServiceExt<Args, Ctx, Svc, B, M>, Svc>(
-        self,
-        service: W,
-    ) -> Worker<Args, Ctx, W::Backend, Svc, M>
+    pub fn build<W, Svc>(self, service: W) -> Worker<Args, Ctx, W::Backend, Svc, M>
     where
         Svc: Service<Task<Args, Ctx, B::IdType>>,
+        W: IntoWorkerServiceExt<Args, Ctx, Svc, B, M>,
     {
         service.build_with(self)
     }

examples/dag-workflow/src/main.rs

@@ -1,5 +1,5 @@
 use apalis::prelude::*;
-use apalis_file_storage::JsonStorage;
+use apalis_file_storage::CsvStorage;
 use apalis_workflow::{DagFlow, WorkflowSink};
 use tracing::info;
 
@@ -30,7 +30,8 @@ async fn main() -> Result<(), BoxDynError> {
         std::env::set_var("RUST_LOG", "debug");
     };
     tracing_subscriber::fmt::init();
-    let mut backend = JsonStorage::new_temp().unwrap();
+    let mut backend = CsvStorage::new_temp().unwrap();
+
     backend.push_start(vec![42, 43, 44]).await.unwrap();
 
     let dag_flow = DagFlow::new("user-info-workflow");