Add comprehensive job timeout functionality

Implements robust timeout detection and handling for job processing with
complete database persistence, statistics integration, and monitoring.

Core Features:
- Per-job timeout configuration with Job::with_timeout() builder
- Worker-level default timeouts with Worker::with_default_timeout()
- TimedOut job status for jobs exceeding timeout duration
- Async timeout detection using tokio::time::timeout
- Intelligent timeout precedence (job-specific overrides worker default)

Database Integration:
- timeout_seconds and timed_out_at columns in PostgreSQL and MySQL schemas
- mark_job_timed_out() method added to DatabaseQueue trait
- Complete timeout support in both database implementations
- Database queries updated for timeout field handling

Statistics & Monitoring:
- timed_out field in JobStatistics for timeout event tracking
- timed_out_count in QueueStats for per-queue timeout monitoring
- Timeout events included in error rate calculations
- JobEventType::TimedOut for comprehensive event tracking

Testing & Examples:
- 14 comprehensive tests covering timeout functionality and edge cases
- Enhanced PostgreSQL and MySQL examples with timeout demonstrations
- Various timeout scenarios (10s, 60s, 600s) and priority-based configs
- Complete test coverage for timeout detection, statistics, and database ops

Technical Implementation:
- Graceful timeout handling without affecting other jobs
- Proper async resource management with tokio timeout
- Backward compatible with existing job queue functionality
- Type-safe timeout configuration throughout the API

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <[email protected]>

Author: CodingAnarchy

CHANGELOG.md

@@ -5,6 +5,53 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [0.2.2] - 2025-06-26
+
+### Added
+- **Comprehensive Job Timeout Functionality**
+  - `TimedOut` job status for jobs that exceed their timeout duration
+  - Per-job timeout configuration with `Job::with_timeout()` builder method
+  - Worker-level default timeouts with `Worker::with_default_timeout()`
+  - Timeout detection using `tokio::time::timeout` for efficient async timeout handling
+  - `timeout` and `timed_out_at` fields added to `Job` struct for complete timeout tracking
+  - Automatic timeout event recording in statistics with `JobEventType::TimedOut`
+
+- **Enhanced Database Support for Timeouts**
+  - `timeout_seconds` and `timed_out_at` columns added to database schema
+  - `mark_job_timed_out()` method added to `DatabaseQueue` trait
+  - Complete timeout support in both PostgreSQL and MySQL implementations
+  - Database queries updated to handle timeout fields in job lifecycle operations
+  - Timeout counts integrated into queue statistics with `timed_out_count` field
+
+- **Timeout Statistics and Monitoring**
+  - `timed_out` field added to `JobStatistics` for timeout event tracking
+  - Timeout events included in error rate calculations for comprehensive metrics
+  - `timed_out_count` added to `QueueStats` for per-queue timeout monitoring
+  - Enhanced statistics display in examples showing timeout metrics
+  - Timeout event processing in `InMemoryStatsCollector`
+
+- **Comprehensive Testing**
+  - 14 new comprehensive tests covering timeout functionality
+  - Job timeout detection logic testing with edge cases
+  - Worker timeout configuration and precedence testing
+  - Timeout statistics integration testing
+  - Database operation interface testing for timeout methods
+  - Job lifecycle testing with timeout scenarios
+
+- **Enhanced Examples**
+  - Updated PostgreSQL example with timeout configuration demonstrations
+  - Updated MySQL example with various timeout scenarios (10s, 60s, 600s)
+  - Job timeout precedence examples (job-specific vs worker defaults)
+  - Priority-based timeout configuration examples (VIP vs standard jobs)
+  - Comprehensive timeout statistics display in both examples
+
+### Technical Implementation
+- Timeout precedence: job-specific timeout takes priority over worker default timeout
+- Graceful timeout handling: jobs are marked as `TimedOut` without affecting other jobs
+- Async timeout detection: uses `tokio::time::timeout` for efficient resource management
+- Database consistency: timeout information persisted and retrievable across job lifecycle
+- Statistics integration: timeout events fully integrated into existing statistics framework
+
 ## [0.2.1] - 2025-06-25
 
 ### Removed
@@ -107,6 +154,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ---
 
 ## Release Links
+- [0.2.2](https://github.com/CodingAnarchy/hammerwork/releases/tag/v0.2.2)
 - [0.2.1](https://github.com/CodingAnarchy/hammerwork/releases/tag/v0.2.1)
 - [0.2.0](https://github.com/CodingAnarchy/hammerwork/releases/tag/v0.2.0)
 - [0.1.0](https://github.com/CodingAnarchy/hammerwork/releases/tag/v0.1.0)

examples/mysql_example.rs

@@ -86,15 +86,17 @@ async fn main() -> std::result::Result<(), Box<dyn std::error::Error>> {
         }) as std::pin::Pin<Box<dyn std::future::Future<Output = Result<()>> + Send>>
     });
 
-    // Create workers for different queues with statistics collection
+    // Create workers for different queues with statistics collection and timeout configuration
     let image_worker = Worker::new(queue.clone(), "image_processing".to_string(), image_handler)
         .with_poll_interval(tokio::time::Duration::from_secs(1))
         .with_max_retries(2)
+        .with_default_timeout(tokio::time::Duration::from_secs(120)) // 2-minute timeout for image processing
         .with_stats_collector(Arc::clone(&stats_collector));
 
     let email_worker = Worker::new(queue.clone(), "email_queue".to_string(), email_handler)
         .with_poll_interval(tokio::time::Duration::from_millis(500))
         .with_max_retries(3)
+        .with_default_timeout(tokio::time::Duration::from_secs(30)) // 30-second timeout for emails
         .with_stats_collector(Arc::clone(&stats_collector));
 
     let mut worker_pool = WorkerPool::new()
@@ -107,15 +109,15 @@ async fn main() -> std::result::Result<(), Box<dyn std::error::Error>> {
     {
         use hammerwork::queue::DatabaseQueue;
 
-        // Create various test jobs including some that will fail
+        // Create various test jobs with timeout configurations
         let successful_image_job = Job::new(
             "image_processing".to_string(),
             json!({
                 "image_url": "https://example.com/photo.jpg",
                 "resize": "800x600",
                 "format": "webp"
             }),
-        );
+        ).with_timeout(std::time::Duration::from_secs(60)); // Custom 1-minute timeout for this job
 
         let failing_image_job = Job::new(
             "image_processing".to_string(),
@@ -124,7 +126,7 @@ async fn main() -> std::result::Result<(), Box<dyn std::error::Error>> {
                 "resize": "800x600",
                 "format": "webp"
             }),
-        );
+        ); // Uses worker default timeout (2 minutes)
 
         let email_job = Job::new(
             "email_queue".to_string(),
@@ -133,7 +135,31 @@ async fn main() -> std::result::Result<(), Box<dyn std::error::Error>> {
                 "subject": "Your image has been processed",
                 "template": "image_ready"
             }),
-        );
+        ).with_timeout(std::time::Duration::from_secs(10)); // Quick timeout for email sending
+
+        // High-priority email with longer timeout and more retries
+        let priority_email_job = Job::new(
+            "email_queue".to_string(),
+            json!({
+                "to": "[email protected]",
+                "subject": "VIP Image Processing Complete",
+                "template": "vip_notification",
+                "priority": "high"
+            }),
+        ).with_timeout(std::time::Duration::from_secs(45)) // Longer timeout for VIP
+         .with_max_attempts(5); // More retry attempts for important emails
+
+        // Large image processing job with extended timeout
+        let large_image_job = Job::new(
+            "image_processing".to_string(),
+            json!({
+                "image_url": "https://example.com/huge_photo.raw",
+                "resize": "4K",
+                "format": "png",
+                "effects": ["sharpen", "color_correct", "denoise"]
+            }),
+        ).with_timeout(std::time::Duration::from_secs(600)) // 10-minute timeout for large processing
+         .with_max_attempts(2); // Fewer retries for expensive operations
 
         // Schedule a delayed job
         let delayed_job = Job::with_delay(
@@ -149,38 +175,49 @@ async fn main() -> std::result::Result<(), Box<dyn std::error::Error>> {
         let job1_id = queue.enqueue(successful_image_job).await?;
         let job2_id = queue.enqueue(failing_image_job).await?;
         let job3_id = queue.enqueue(email_job).await?;
-        let job4_id = queue.enqueue(delayed_job).await?;
-
-        info!("Enqueued test jobs: {}, {}, {}, {}", job1_id, job2_id, job3_id, job4_id);
+        let job4_id = queue.enqueue(priority_email_job).await?;
+        let job5_id = queue.enqueue(large_image_job).await?;
+        let job6_id = queue.enqueue(delayed_job).await?;
+
+        info!("Enqueued test jobs with various timeout configurations:");
+        info!("  {} - image processing with 60s timeout", job1_id);
+        info!("  {} - failing image job (uses worker default 2min timeout)", job2_id);
+        info!("  {} - email with 10s timeout", job3_id);
+        info!("  {} - VIP email with 45s timeout", job4_id);
+        info!("  {} - large image with 10min timeout", job5_id);
+        info!("  {} - delayed email (1min delay)", job6_id);
 
         // Let jobs process for a bit
         tokio::time::sleep(tokio::time::Duration::from_secs(8)).await;
 
-        // Demonstrate statistics collection
-        info!("=== Job Processing Statistics ===");
+        // Demonstrate statistics collection with timeout tracking
+        info!("=== Job Processing Statistics (Including Timeouts) ===");
         let system_stats = stats_collector.get_system_statistics(std::time::Duration::from_secs(300)).await?;
-        info!("System Stats - Total: {}, Completed: {}, Failed: {}, Dead: {}, Error Rate: {:.2}%",
+        info!("System Stats - Total: {}, Completed: {}, Failed: {}, Dead: {}, Timed Out: {}, Error Rate: {:.2}%",
             system_stats.total_processed,
             system_stats.completed,
             system_stats.failed,
             system_stats.dead,
+            system_stats.timed_out,
             system_stats.error_rate * 100.0
         );
 
         // Get statistics for each queue
         let image_stats = stats_collector.get_queue_statistics("image_processing", std::time::Duration::from_secs(300)).await?;
-        info!("Image Processing Stats - Total: {}, Completed: {}, Failed: {}, Avg Time: {:.2}ms",
+        info!("Image Processing Stats - Total: {}, Completed: {}, Failed: {}, Timed Out: {}, Avg Time: {:.2}ms",
             image_stats.total_processed,
             image_stats.completed,
             image_stats.failed,
+            image_stats.timed_out,
             image_stats.avg_processing_time_ms
         );
 
         let email_stats = stats_collector.get_queue_statistics("email_queue", std::time::Duration::from_secs(300)).await?;
-        info!("Email Queue Stats - Total: {}, Completed: {}, Failed: {}, Avg Time: {:.2}ms",
+        info!("Email Queue Stats - Total: {}, Completed: {}, Failed: {}, Timed Out: {}, Avg Time: {:.2}ms",
             email_stats.total_processed,
             email_stats.completed,
             email_stats.failed,
+            email_stats.timed_out,
             email_stats.avg_processing_time_ms
         );
 
@@ -204,14 +241,15 @@ async fn main() -> std::result::Result<(), Box<dyn std::error::Error>> {
             dead_summary.dead_jobs_by_queue
         );
 
-        // Show queue statistics from database
+        // Show queue statistics from database (including timeout counts)
         let all_queue_stats = queue.get_all_queue_stats().await?;
         for queue_stat in all_queue_stats {
-            info!("Queue '{}' - Pending: {}, Running: {}, Dead: {}, Completed: {}",
+            info!("Queue '{}' - Pending: {}, Running: {}, Dead: {}, Timed Out: {}, Completed: {}",
                 queue_stat.queue_name,
                 queue_stat.pending_count,
                 queue_stat.running_count,
                 queue_stat.dead_count,
+                queue_stat.timed_out_count,
                 queue_stat.completed_count
             );
         }
@@ -226,7 +264,21 @@ async fn main() -> std::result::Result<(), Box<dyn std::error::Error>> {
         }
     }
 
-    info!("MySQL example completed successfully with statistics and dead job management.");
+    info!("MySQL example completed successfully!");
+    info!("Demonstrated features:");
+    info!("  ✓ Job timeout configuration (per-job and worker defaults)");
+    info!("  ✓ Various timeout scenarios (10s email, 60s image, 10min large jobs)");
+    info!("  ✓ Timeout statistics tracking and monitoring");
+    info!("  ✓ Dead job management and error analysis");
+    info!("  ✓ Comprehensive queue statistics with timeout counts");
+    info!("");
+    info!("Timeout Features Demonstrated:");
+    info!("  • Worker default timeouts (30s for email, 2min for image processing)");
+    info!("  • Job-specific timeouts (10s, 45s, 60s, 600s examples)");
+    info!("  • Priority-based timeout configuration (VIP vs standard jobs)");
+    info!("  • Timeout event tracking in statistics");
+    info!("  • Database timeout count tracking");
+    info!("");
     info!("In a real application, you would start the worker pool to run indefinitely:");
     info!("worker_pool.start().await?;");
 

examples/postgres_example.rs

@@ -52,15 +52,17 @@ async fn main() -> std::result::Result<(), Box<dyn std::error::Error>> {
         }) as std::pin::Pin<Box<dyn std::future::Future<Output = Result<()>> + Send>>
     });
 
-    // Create and start workers with statistics collection
+    // Create and start workers with statistics collection and timeout configuration
     let worker1 = Worker::new(queue.clone(), "email".to_string(), handler.clone())
         .with_poll_interval(tokio::time::Duration::from_secs(1))
         .with_max_retries(3)
+        .with_default_timeout(tokio::time::Duration::from_secs(30)) // 30 second default timeout
         .with_stats_collector(Arc::clone(&stats_collector));
 
     let worker2 = Worker::new(queue.clone(), "notifications".to_string(), handler.clone())
         .with_poll_interval(tokio::time::Duration::from_secs(2))
         .with_max_retries(2)
+        .with_default_timeout(tokio::time::Duration::from_secs(60)) // 60 second default timeout
         .with_stats_collector(Arc::clone(&stats_collector));
 
     let mut worker_pool = WorkerPool::new()
@@ -73,47 +75,75 @@ async fn main() -> std::result::Result<(), Box<dyn std::error::Error>> {
     {
         use hammerwork::queue::DatabaseQueue;
 
+        // Create jobs with various timeout configurations
         let job1 = Job::new(
             "email".to_string(),
             json!({"to": "[email protected]", "subject": "Hello"}),
-        );
+        ); // Uses worker default timeout (30s)
+        
         let job2 = Job::new(
             "notifications".to_string(),
             json!({"message": "Welcome!", "user_id": 123}),
-        );
+        ).with_timeout(std::time::Duration::from_secs(10)); // Custom 10-second timeout
+        
         let job3 = Job::new("email".to_string(), json!({"action": "fail"})); // This will fail
-        let job4 = Job::new("email".to_string(), json!({"action": "fail"})); // This will also fail and become dead
+        
+        let job4 = Job::new("email".to_string(), json!({"action": "fail"})) // This will also fail and become dead
+            .with_timeout(std::time::Duration::from_secs(5)); // Short timeout for demo
+        
+        // Job with a very long timeout for heavy processing
+        let job5 = Job::new(
+            "email".to_string(),
+            json!({"to": "[email protected]", "subject": "Heavy Processing"}),
+        ).with_timeout(std::time::Duration::from_secs(300)) // 5-minute timeout
+         .with_max_attempts(5); // More retry attempts for important jobs
 
         let job1_id = queue.enqueue(job1).await?;
         let job2_id = queue.enqueue(job2).await?;
         let job3_id = queue.enqueue(job3).await?;
         let job4_id = queue.enqueue(job4).await?;
+        let job5_id = queue.enqueue(job5).await?;
 
-        info!("Enqueued test jobs: {}, {}, {}, {}", job1_id, job2_id, job3_id, job4_id);
+        info!("Enqueued test jobs with timeouts:");
+        info!("  {} - uses worker default timeout (30s)", job1_id);
+        info!("  {} - custom 10s timeout", job2_id);
+        info!("  {} - will fail (uses default timeout)", job3_id);
+        info!("  {} - will fail with 5s timeout", job4_id);
+        info!("  {} - heavy processing with 5min timeout", job5_id);
 
         // Let jobs process for a bit
         tokio::time::sleep(tokio::time::Duration::from_secs(5)).await;
 
-        // Demonstrate statistics collection
-        info!("=== Job Processing Statistics ===");
+        // Demonstrate statistics collection with timeout tracking
+        info!("=== Job Processing Statistics (Including Timeouts) ===");
         let system_stats = stats_collector.get_system_statistics(std::time::Duration::from_secs(300)).await?;
-        info!("System Stats - Total: {}, Completed: {}, Failed: {}, Dead: {}, Error Rate: {:.2}%",
+        info!("System Stats - Total: {}, Completed: {}, Failed: {}, Dead: {}, Timed Out: {}, Error Rate: {:.2}%",
             system_stats.total_processed,
             system_stats.completed,
             system_stats.failed,
             system_stats.dead,
+            system_stats.timed_out,
             system_stats.error_rate * 100.0
         );
 
         // Get queue-specific statistics
         let email_stats = stats_collector.get_queue_statistics("email", std::time::Duration::from_secs(300)).await?;
-        info!("Email Queue Stats - Total: {}, Completed: {}, Failed: {}, Avg Processing Time: {:.2}ms",
+        info!("Email Queue Stats - Total: {}, Completed: {}, Failed: {}, Timed Out: {}, Avg Processing Time: {:.2}ms",
             email_stats.total_processed,
             email_stats.completed,
             email_stats.failed,
+            email_stats.timed_out,
             email_stats.avg_processing_time_ms
         );
 
+        let notifications_stats = stats_collector.get_queue_statistics("notifications", std::time::Duration::from_secs(300)).await?;
+        info!("Notifications Queue Stats - Total: {}, Completed: {}, Timed Out: {}, Avg Processing Time: {:.2}ms",
+            notifications_stats.total_processed,
+            notifications_stats.completed,
+            notifications_stats.timed_out,
+            notifications_stats.avg_processing_time_ms
+        );
+
         // Demonstrate dead job management
         info!("=== Dead Job Management ===");
         let dead_jobs = queue.get_dead_jobs(Some(10), None).await?;
@@ -131,15 +161,30 @@ async fn main() -> std::result::Result<(), Box<dyn std::error::Error>> {
             dead_summary.error_patterns
         );
 
-        // Demonstrate queue statistics from database
+        // Demonstrate queue statistics from database (including timeout counts)
         let queue_stats = queue.get_queue_stats("email").await?;
-        info!("Email Queue DB Stats - Pending: {}, Running: {}, Dead: {}, Completed: {}",
+        info!("Email Queue DB Stats - Pending: {}, Running: {}, Dead: {}, Timed Out: {}, Completed: {}",
             queue_stats.pending_count,
             queue_stats.running_count,
             queue_stats.dead_count,
+            queue_stats.timed_out_count,
             queue_stats.completed_count
         );
 
+        // Show all queue statistics
+        let all_queue_stats = queue.get_all_queue_stats().await?;
+        info!("=== All Queue Statistics ===");
+        for stats in &all_queue_stats {
+            info!("Queue '{}' - Pending: {}, Running: {}, Dead: {}, Timed Out: {}, Completed: {}",
+                stats.queue_name,
+                stats.pending_count,
+                stats.running_count,
+                stats.dead_count,
+                stats.timed_out_count,
+                stats.completed_count
+            );
+        }
+
         // If there are dead jobs, demonstrate retry functionality
         if !dead_jobs.is_empty() {
             let dead_job_id = dead_jobs[0].id;
@@ -149,7 +194,19 @@ async fn main() -> std::result::Result<(), Box<dyn std::error::Error>> {
         }
     }
 
-    info!("Example completed successfully. Check the statistics and dead job management features.");
+    info!("Example completed successfully!");
+    info!("Demonstrated features:");
+    info!("  ✓ Job timeout configuration (per-job and worker defaults)");
+    info!("  ✓ Timeout statistics tracking and monitoring");
+    info!("  ✓ Dead job management and retry functionality");
+    info!("  ✓ Comprehensive job processing statistics");
+    info!("");
+    info!("Timeout Features Shown:");
+    info!("  • Worker default timeouts (30s for email, 60s for notifications)");
+    info!("  • Job-specific timeouts (10s, 5s, 300s examples)");
+    info!("  • Timeout event tracking in statistics");
+    info!("  • Timeout counts in database queue statistics");
+    info!("");
     info!("In a real application, you would start the worker pool to run indefinitely:");
     info!("worker_pool.start().await?;");