Fix sub-second queue delays in TWMQ

Sub-second delayed requeues in TWMQ were being truncated to 0s because queue scheduling used Duration::as_secs().

That meant values like 200ms were effectively treated as immediate retries, which could cause hot-looping at queue poll cadence.

This change adds a small helper that rounds any non-zero sub-second delay up to 1s and uses it consistently in:
- queue push scheduling
- hook scheduling
- multilane queue scheduling

Author: 0xFirekeeper

executors/src/solana_executor/worker.rs

@@ -221,7 +221,7 @@ impl DurableExecution for SolanaExecutorJobHandler {
     type JobData = SolanaExecutorJobData;
 
     #[tracing::instrument(
-        skip(self, job), 
+        skip(self, job),
         fields(
             transaction_id = job.job.id,
             stage = Self::stage_name()
@@ -648,7 +648,6 @@ impl SolanaExecutorJobHandler {
                                         let has_signatures = versioned_tx.signatures.iter().any(|sig| {
                                             sig.as_ref() != [0u8; 64]
                                         });
-                                        
                                         if has_signatures {
                                             error!(
                                                 transaction_id = %transaction_id,

twmq/src/hooks.rs

@@ -1,6 +1,6 @@
 use std::time::{SystemTime, UNIX_EPOCH};
 
-use crate::{DurableExecution, error::TwmqError, job::PushableJob};
+use crate::{DurableExecution, delay_to_queue_seconds, error::TwmqError, job::PushableJob};
 
 // A minimal transaction context that hooks can use
 pub struct TransactionContext<'a> {
@@ -51,7 +51,7 @@ impl<'a> TransactionContext<'a> {
             .sadd(job.queue.dedupe_set_name(), &job.options.id);
 
         if let Some(delay_options) = job.options.delay {
-            let process_at = now + delay_options.delay.as_secs();
+            let process_at = now + delay_to_queue_seconds(delay_options.delay);
             self.pipeline
                 .hset(
                     job.queue.job_meta_hash_name(&job.options.id),

twmq/src/lib.rs

@@ -28,6 +28,18 @@ pub use queue::IdempotencyMode;
 pub use redis;
 use tracing::Instrument;
 
+pub(crate) fn delay_to_queue_seconds(delay: Duration) -> u64 {
+    let delay_secs = delay.as_secs();
+
+    if delay.is_zero() {
+        0
+    } else if delay_secs == 0 {
+        1
+    } else {
+        delay_secs
+    }
+}
+
 // Trait for error types to implement user cancellation
 pub trait UserCancellable {
     fn user_cancelled() -> Self;
@@ -297,7 +309,7 @@ impl<H: DurableExecution> Queue<H> {
             position: RequeuePosition::Last,
         });
 
-        let delay_secs = delay.delay.as_secs();
+        let delay_secs = delay_to_queue_seconds(delay.delay);
         let position_string = delay.position.to_string();
 
         let _result: (i32, String) = script
@@ -1049,7 +1061,7 @@ impl<H: DurableExecution> Queue<H> {
 
         // Add to proper queue based on delay and position
         if let Some(delay_duration) = delay {
-            let delay_until = now + delay_duration.as_secs();
+            let delay_until = now + delay_to_queue_seconds(delay_duration);
             let pos_str = position.to_string();
 
             pipeline

twmq/src/multilane.rs

@@ -9,7 +9,7 @@ use tracing::Instrument;
 
 use crate::{
     CancelResult, DurableExecution, FailHookData, NackHookData, QueueInternalErrorHookData,
-    SuccessHookData, UserCancellable,
+    SuccessHookData, UserCancellable, delay_to_queue_seconds,
     error::TwmqError,
     hooks::TransactionContext,
     job::{
@@ -219,7 +219,7 @@ impl<H: DurableExecution> MultilaneQueue<H> {
             position: RequeuePosition::Last,
         });
 
-        let delay_secs = delay.delay.as_secs();
+        let delay_secs = delay_to_queue_seconds(delay.delay);
         let position_string = delay.position.to_string();
 
         let _result: (i32, String) = script
@@ -1226,7 +1226,7 @@ impl<H: DurableExecution> MultilaneQueue<H> {
                         .duration_since(UNIX_EPOCH)
                         .unwrap()
                         .as_secs();
-                    let delay_until = now + delay_duration.as_secs();
+                    let delay_until = now + delay_to_queue_seconds(*delay_duration);
                     let pos_str = position.to_string();
 
                     hook_pipeline