Timeout variant and per-attempt timeout in TaskConfig (#13)

- Added `Timeout` variant to `CanoError` for bounded operation deadline exceedance.
- Introduced `attempt_timeout` in `TaskConfig` to wrap attempts in `tokio::time::timeout`.
- Updated documentation to reflect new timeout features and their integration with retries.
- Improved panic safety in workflows by catching panics and returning `CanoError::TaskExecution`.
- Added tests for timeout behavior and bulkhead concurrency limits.

Author: nassor

Cargo.lock

@@ -3,7 +3,7 @@ resolver = "3"
 members = ["cano", "cano-macros"]
 
 [workspace.package]
-version = "0.10.2"
+version = "0.10.3"
 edition = "2024"
 rust-version = "1.95.0"
 license = "MIT OR Apache-2.0"

Cargo.toml

@@ -29,6 +29,7 @@
 //! | `Workflow` | Workflow orchestration fails | Verify node registration and routing |
 //! | `Configuration` | Invalid node/workflow config | Check parameters and settings |
 //! | `RetryExhausted` | All retries exhausted | Increase retries or fix root cause |
+//! | `Timeout` | Per-attempt timeout reached | Increase `attempt_timeout` or speed up the task |
 //! | `Generic` | General errors | Check the specific error message |
 //!
 //! ## Using Errors in Your Nodes
@@ -178,6 +179,16 @@ pub enum CanoError {
     /// all configured retry attempts have been used up.
     RetryExhausted(String),
 
+    /// A bounded operation exceeded its deadline.
+    ///
+    /// Emitted by per-attempt task timeouts (see
+    /// [`crate::task::TaskConfig::with_attempt_timeout`]). The retry loop treats
+    /// this as a recoverable failure: each attempt that times out is retried
+    /// (subject to the configured `RetryMode`), and if all attempts time out
+    /// the loop wraps the final timeout error in
+    /// [`CanoError::RetryExhausted`].
+    Timeout(String),
+
     /// General-purpose error for other scenarios
     ///
     /// Use this for errors that don't fit the other categories.
@@ -230,6 +241,11 @@ impl CanoError {
         CanoError::RetryExhausted(msg.into())
     }
 
+    /// Create a new timeout error
+    pub fn timeout<S: Into<String>>(msg: S) -> Self {
+        CanoError::Timeout(msg.into())
+    }
+
     /// Create a new generic error
     pub fn generic<S: Into<String>>(msg: S) -> Self {
         CanoError::Generic(msg.into())
@@ -260,6 +276,7 @@ impl CanoError {
             CanoError::Workflow(msg) => msg,
             CanoError::Configuration(msg) => msg,
             CanoError::RetryExhausted(msg) => msg,
+            CanoError::Timeout(msg) => msg,
             CanoError::Generic(msg) => msg,
             CanoError::ResourceNotFound(msg) => msg,
             CanoError::ResourceTypeMismatch(msg) => msg,
@@ -277,6 +294,7 @@ impl CanoError {
             CanoError::Workflow(_) => "workflow",
             CanoError::Configuration(_) => "configuration",
             CanoError::RetryExhausted(_) => "retry_exhausted",
+            CanoError::Timeout(_) => "timeout",
             CanoError::Generic(_) => "generic",
             CanoError::ResourceNotFound(_) => "resource_not_found",
             CanoError::ResourceTypeMismatch(_) => "resource_type_mismatch",
@@ -295,6 +313,7 @@ impl std::fmt::Display for CanoError {
             CanoError::Workflow(msg) => write!(f, "Workflow error: {msg}"),
             CanoError::Configuration(msg) => write!(f, "Configuration error: {msg}"),
             CanoError::RetryExhausted(msg) => write!(f, "Retry exhausted: {msg}"),
+            CanoError::Timeout(msg) => write!(f, "Timeout error: {msg}"),
             CanoError::Generic(msg) => write!(f, "Error: {msg}"),
             CanoError::ResourceNotFound(msg) => write!(f, "Resource not found: {msg}"),
             CanoError::ResourceTypeMismatch(msg) => write!(f, "Resource type mismatch: {msg}"),
@@ -315,6 +334,7 @@ impl PartialEq for CanoError {
             (CanoError::Workflow(a), CanoError::Workflow(b)) => a == b,
             (CanoError::Configuration(a), CanoError::Configuration(b)) => a == b,
             (CanoError::RetryExhausted(a), CanoError::RetryExhausted(b)) => a == b,
+            (CanoError::Timeout(a), CanoError::Timeout(b)) => a == b,
             (CanoError::Generic(a), CanoError::Generic(b)) => a == b,
             (CanoError::ResourceNotFound(a), CanoError::ResourceNotFound(b)) => a == b,
             (CanoError::ResourceTypeMismatch(a), CanoError::ResourceTypeMismatch(b)) => a == b,
@@ -519,6 +539,30 @@ mod tests {
         assert_eq!(e, f);
     }
 
+    #[test]
+    fn test_timeout_constructor_and_category() {
+        let err = CanoError::timeout("attempt deadline reached");
+        assert_eq!(err.message(), "attempt deadline reached");
+        assert_eq!(err.category(), "timeout");
+        assert_eq!(format!("{err}"), "Timeout error: attempt deadline reached");
+    }
+
+    #[test]
+    fn test_timeout_partial_eq() {
+        let a = CanoError::timeout("d");
+        let b = CanoError::timeout("d");
+        assert_eq!(a, b);
+
+        let c = CanoError::timeout("x");
+        let d = CanoError::timeout("y");
+        assert_ne!(c, d);
+
+        // Distinct from same-message variants
+        let timeout = CanoError::timeout("k");
+        let workflow = CanoError::workflow("k");
+        assert_ne!(timeout, workflow);
+    }
+
     #[test]
     fn test_resource_not_found_distinct_from_resource_type_mismatch() {
         let not_found = CanoError::resource_not_found("key");

cano/src/error.rs

@@ -259,6 +259,12 @@ impl Default for RetryMode {
 pub struct TaskConfig {
     /// Retry strategy for failed executions
     pub retry_mode: RetryMode,
+    /// Optional per-attempt timeout. When set, each attempt inside
+    /// [`run_with_retries`] is wrapped with [`tokio::time::timeout`]; an
+    /// expired attempt produces a [`CanoError::Timeout`] and the retry loop
+    /// continues per `retry_mode`. `None` (the default) preserves the
+    /// previous behavior of letting attempts run unbounded.
+    pub attempt_timeout: Option<Duration>,
 }
 
 impl TaskConfig {
@@ -273,6 +279,7 @@ impl TaskConfig {
     pub fn minimal() -> Self {
         Self {
             retry_mode: RetryMode::None,
+            attempt_timeout: None,
         }
     }
 
@@ -291,6 +298,12 @@ impl TaskConfig {
     pub fn with_exponential_retry(self, max_retries: usize) -> Self {
         self.with_retry(RetryMode::exponential(max_retries))
     }
+
+    /// Apply a per-attempt timeout. Each retry attempt gets a fresh deadline.
+    pub fn with_attempt_timeout(mut self, timeout: Duration) -> Self {
+        self.attempt_timeout = Some(timeout);
+        self
+    }
 }
 
 /// Default implementation for retry logic that can be used by any task
@@ -326,7 +339,19 @@ where
         #[cfg(feature = "tracing")]
         debug!(attempt = attempt + 1, "Executing task attempt");
 
-        match run_fn().await {
+        let attempt_outcome = match config.attempt_timeout {
+            Some(d) => match tokio::time::timeout(d, run_fn()).await {
+                Ok(inner) => inner,
+                Err(_) => Err(CanoError::timeout(format!(
+                    "Task attempt {} exceeded attempt_timeout of {:?}",
+                    attempt + 1,
+                    d
+                ))),
+            },
+            None => run_fn().await,
+        };
+
+        match attempt_outcome {
             Ok(result) => {
                 #[cfg(feature = "tracing")]
                 info!(attempt = attempt + 1, "Task execution successful");
@@ -1169,6 +1194,83 @@ mod tests {
         assert_eq!(bare_called.load(Ordering::SeqCst), 0);
     }
 
+    #[tokio::test]
+    async fn test_attempt_timeout_triggers_retry() {
+        use std::sync::atomic::{AtomicUsize, Ordering};
+
+        let config = TaskConfig::new()
+            .with_fixed_retry(2, Duration::from_millis(1))
+            .with_attempt_timeout(Duration::from_millis(20));
+        let counter = Arc::new(AtomicUsize::new(0));
+        let counter_clone = Arc::clone(&counter);
+
+        let result = run_with_retries::<TaskResult<String>, _, _>(&config, || {
+            let counter = Arc::clone(&counter_clone);
+            async move {
+                counter.fetch_add(1, Ordering::SeqCst);
+                tokio::time::sleep(Duration::from_millis(200)).await;
+                Ok::<TaskResult<String>, CanoError>(TaskResult::Single("never".to_string()))
+            }
+        })
+        .await;
+
+        assert!(result.is_err());
+        let err = result.unwrap_err();
+        assert!(
+            matches!(err, CanoError::RetryExhausted(_)),
+            "expected RetryExhausted, got: {err}"
+        );
+        let msg = err.to_string();
+        assert!(
+            msg.contains("Timeout error") || msg.contains("attempt_timeout"),
+            "expected timeout context in error, got: {msg}"
+        );
+        // 1 initial + 2 retries — every attempt times out before the sleep returns.
+        assert_eq!(counter.load(Ordering::SeqCst), 3);
+    }
+
+    #[tokio::test]
+    async fn test_attempt_timeout_none_unchanged() {
+        let config = TaskConfig::new();
+        assert!(config.attempt_timeout.is_none());
+
+        let result = run_with_retries::<TaskResult<String>, _, _>(&config, || async {
+            tokio::time::sleep(Duration::from_millis(5)).await;
+            Ok::<TaskResult<String>, CanoError>(TaskResult::Single("ok".to_string()))
+        })
+        .await
+        .unwrap();
+        assert_eq!(result, TaskResult::Single("ok".to_string()));
+    }
+
+    #[tokio::test]
+    async fn test_attempt_timeout_resets_per_attempt() {
+        use std::sync::atomic::{AtomicUsize, Ordering};
+
+        // First attempt sleeps long enough to trip the timeout; second attempt
+        // returns immediately. Verifies each attempt gets a fresh deadline.
+        let config = TaskConfig::new()
+            .with_fixed_retry(1, Duration::from_millis(1))
+            .with_attempt_timeout(Duration::from_millis(30));
+        let counter = Arc::new(AtomicUsize::new(0));
+        let counter_clone = Arc::clone(&counter);
+
+        let result = run_with_retries::<TaskResult<String>, _, _>(&config, || {
+            let counter = Arc::clone(&counter_clone);
+            async move {
+                let n = counter.fetch_add(1, Ordering::SeqCst);
+                if n == 0 {
+                    tokio::time::sleep(Duration::from_millis(200)).await;
+                }
+                Ok::<TaskResult<String>, CanoError>(TaskResult::Single("ok".to_string()))
+            }
+        })
+        .await
+        .unwrap();
+        assert_eq!(result, TaskResult::Single("ok".to_string()));
+        assert_eq!(counter.load(Ordering::SeqCst), 2);
+    }
+
     #[tokio::test]
     async fn test_retry_exhausted_error_type() {
         use std::sync::atomic::{AtomicUsize, Ordering};