update local concurrency interface

Author: d4mr

twmq/src/lib.rs

@@ -327,13 +327,13 @@ where
         Ok(count)
     }
 
-    pub async fn work(self: Arc<Self>, local_concurrency: usize) -> Result<(), TwmqError> {
+    pub async fn work(self: Arc<Self>) -> Result<(), TwmqError> {
         // Local semaphore to limit concurrency per instance
-        let semaphore = Arc::new(Semaphore::new(local_concurrency));
+        let semaphore = Arc::new(Semaphore::new(self.options.local_concurrency));
 
         // Start worker
         tokio::spawn(async move {
-            let mut interval = tokio::time::interval(Duration::from_millis(100));
+            let mut interval = tokio::time::interval(self.options.polling_interval);
 
             loop {
                 interval.tick().await;

twmq/src/queue.rs

@@ -1,9 +1,13 @@
 use std::time::Duration;
 
+#[derive(Clone, Debug)]
 pub struct QueueOptions {
     pub max_success: usize,
     pub max_failed: usize,
     pub lease_duration: Duration,
+    pub local_concurrency: usize,
+
+    pub polling_interval: Duration,
 
     /// If true, always poll for jobs even if there are no available permits
     /// This is important, because polling is how delayed and timed out jobs are handled
@@ -17,6 +21,8 @@ impl Default for QueueOptions {
         Self {
             max_success: 1000,
             max_failed: 10000,
+            local_concurrency: 100,
+            polling_interval: Duration::from_millis(100),
             lease_duration: Duration::from_secs(30),
             always_poll: false,
         }

twmq/tests/basic.rs

@@ -95,7 +95,7 @@ async fn test_queue_push_and_process_job() {
     let worker_queue_ref = Arc::clone(&queue);
     let worker_handle = tokio::spawn(async move {
         // The worker will loop internally, so we expect it to pick up the job
-        if let Err(e) = worker_queue_ref.work(1).await {
+        if let Err(e) = worker_queue_ref.work().await {
             eprintln!("Worker for queue {} failed: {:?}", queue_name_clone, e);
         }
     });

twmq/tests/basic_hook.rs

@@ -17,6 +17,7 @@ use twmq::{
     DurableExecution, Queue,
     hooks::TransactionContext,
     job::{Job, JobResult, JobStatus, RequeuePosition},
+    queue::QueueOptions,
 };
 
 // Helper to clean up Redis keys for a given queue name pattern
@@ -203,12 +204,15 @@ async fn test_cross_queue_job_scheduling() {
     println!("Creating main queue: {}", main_queue_name);
     println!("Creating webhook queue: {}", webhook_queue_name);
 
+    let mut queue_options = QueueOptions::default();
+    queue_options.local_concurrency = 1;
+
     // Create webhook queue
     let webhook_queue = Arc::new(
         Queue::<WebhookJobPayload, WebhookJobOutput, WebhookJobErrorData, ()>::new(
             REDIS_URL,
             &webhook_queue_name,
-            None,
+            Some(queue_options.clone()),
             (),
         )
         .await
@@ -220,7 +224,7 @@ async fn test_cross_queue_job_scheduling() {
         Queue::<MainJobPayload, TestJobOutput, TestJobErrorData, Arc<WebhookQueue>>::new(
             REDIS_URL,
             &main_queue_name,
-            None,
+            Some(queue_options),
             webhook_queue.clone(),
         )
         .await
@@ -252,7 +256,7 @@ async fn test_cross_queue_job_scheduling() {
     let main_worker = {
         let queue = main_queue.clone();
         tokio::spawn(async move {
-            if let Err(e) = queue.work(1).await {
+            if let Err(e) = queue.work().await {
                 eprintln!("Main worker failed: {:?}", e);
             }
         })
@@ -261,7 +265,7 @@ async fn test_cross_queue_job_scheduling() {
     let webhook_worker = {
         let queue = webhook_queue.clone();
         tokio::spawn(async move {
-            if let Err(e) = queue.work(1).await {
+            if let Err(e) = queue.work().await {
                 eprintln!("Webhook worker failed: {:?}", e);
             }
         })

twmq/tests/lease_expiry.rs

@@ -141,6 +141,8 @@ async fn test_job_lease_expiry() {
         max_success: 1000,
         max_failed: 1000,
         lease_duration,
+        polling_interval: Duration::from_millis(100),
+        local_concurrency: 1,
         always_poll: true,
     };
 
@@ -189,7 +191,7 @@ async fn test_job_lease_expiry() {
     let worker = {
         let queue = queue.clone();
         tokio::spawn(async move {
-            if let Err(e) = queue.work(1).await {
+            if let Err(e) = queue.work().await {
                 tracing::error!("Lease worker failed: {:?}", e);
             }
         })
@@ -290,7 +292,9 @@ async fn test_multiple_job_lease_expiry() {
     let queue_options = QueueOptions {
         max_success: 1000,
         max_failed: 1000,
+        local_concurrency: 3,
         lease_duration,
+        polling_interval: Duration::from_millis(100),
         always_poll: true,
     };
 
@@ -328,7 +332,7 @@ async fn test_multiple_job_lease_expiry() {
     let worker = {
         let queue = queue.clone();
         tokio::spawn(async move {
-            if let Err(e) = queue.work(3).await {
+            if let Err(e) = queue.work().await {
                 tracing::error!("Multi-lease worker failed: {:?}", e);
             }
         })

twmq/tests/nack.rs

@@ -14,6 +14,7 @@ use twmq::{
     DurableExecution, Queue,
     hooks::TransactionContext,
     job::{Job, JobResult, JobStatus, RequeuePosition},
+    queue::QueueOptions,
     redis::aio::ConnectionManager,
 };
 
@@ -168,11 +169,14 @@ async fn test_job_retry_attempts() {
     tracing::info!("Creating retry queue: {}", queue_name);
     tracing::info!("Job should succeed after {} attempts", desired_attempts);
 
+    let mut queue_options = QueueOptions::default();
+    queue_options.local_concurrency = 1;
+
     let queue = Arc::new(
         Queue::<RetryJobPayload, RetryJobOutput, RetryJobErrorData, ()>::new(
             REDIS_URL,
             &queue_name,
-            None,
+            Some(queue_options),
             (),
         )
         .await
@@ -202,7 +206,7 @@ async fn test_job_retry_attempts() {
     let worker = {
         let queue = queue.clone();
         tokio::spawn(async move {
-            if let Err(e) = queue.work(1).await {
+            if let Err(e) = queue.work().await {
                 tracing::error!("Retry worker failed: {:?}", e);
             }
         })
@@ -297,11 +301,14 @@ async fn test_different_retry_counts() {
         // Reset counters
         RETRY_JOB_FINAL_SUCCESS.store(false, Ordering::SeqCst);
 
+        let mut queue_options = QueueOptions::default();
+        queue_options.local_concurrency = 1;
+
         let queue = Arc::new(
             Queue::<RetryJobPayload, RetryJobOutput, RetryJobErrorData, ()>::new(
                 REDIS_URL,
                 &queue_name,
-                None,
+                Some(queue_options),
                 (),
             )
             .await
@@ -326,7 +333,7 @@ async fn test_different_retry_counts() {
         let worker = {
             let queue = queue.clone();
             tokio::spawn(async move {
-                if let Err(e) = queue.work(1).await {
+                if let Err(e) = queue.work().await {
                     tracing::error!("Worker failed: {:?}", e);
                 }
             })