feat: add opt-in LIFO dispatch order

crates/taskito-core/src/scheduler/mod.rs

@@ -322,6 +322,10 @@ pub struct Scheduler {
     /// a queue has CoDel configured; `codel_configs.is_empty()` short-circuits
     /// the common no-CoDel case before this lock is ever taken.
     codel_states: Mutex<HashMap<String, codel::CodelState>>,
+    /// Opt-in per-queue dispatch order (queue → Lifo). Absent = the fair Fifo
+    /// default. Passed straight to the storage dequeue so each queue is scanned
+    /// with its own tie-break direction.
+    dispatch_orders: HashMap<String, crate::storage::DispatchOrder>,
     queues: Vec<String>,
     config: SchedulerConfig,
     shutdown: Arc<Notify>,
@@ -387,6 +391,7 @@ impl Scheduler {
             queue_configs: HashMap::new(),
             codel_configs: HashMap::new(),
             codel_states: Mutex::new(HashMap::new()),
+            dispatch_orders: HashMap::new(),
             queues,
             config,
             shutdown: Arc::new(Notify::new()),
@@ -550,6 +555,16 @@ impl Scheduler {
         self.codel_configs.insert(queue_name, config);
     }
 
+    /// Set a queue's dispatch order. `Lifo` runs newest-first within a priority
+    /// under overload; the default is `Fifo`. Priority always dominates.
+    pub fn register_queue_dispatch_order(
+        &mut self,
+        queue_name: String,
+        order: crate::storage::DispatchOrder,
+    ) {
+        self.dispatch_orders.insert(queue_name, order);
+    }
+
     /// Set a task's resilience policy and register its circuit breaker, if any.
     pub fn register_task(&mut self, task_name: String, config: TaskConfig) {
         if let Some(ref cb_config) = config.circuit_breaker {

crates/taskito-core/src/scheduler/poller.rs

@@ -101,10 +101,12 @@ impl Scheduler {
             return Ok(false);
         }
 
-        let job = match self
-            .storage
-            .dequeue_from(&active_queues, now, self.namespace.as_deref())?
-        {
+        let job = match self.storage.dequeue_from(
+            &active_queues,
+            now,
+            self.namespace.as_deref(),
+            &self.dispatch_orders,
+        )? {
             Some(j) => j,
             None => return Ok(false),
         };
@@ -163,6 +165,7 @@ impl Scheduler {
             now,
             self.namespace.as_deref(),
             budget,
+            &self.dispatch_orders,
         )?;
         if jobs.is_empty() {
             return Ok(false);

crates/taskito-core/src/storage/diesel_common/jobs.rs

@@ -550,13 +550,31 @@ macro_rules! impl_diesel_job_ops {
                 queue_name: &str,
                 now: i64,
                 namespace: Option<&str>,
+            ) -> Result<Option<Job>> {
+                self.dequeue_ordered(
+                    queue_name,
+                    now,
+                    namespace,
+                    $crate::storage::DispatchOrder::Fifo,
+                )
+            }
+
+            /// [`dequeue`](Self::dequeue) with an explicit dispatch order — the
+            /// per-queue path used by [`dequeue_from`](Self::dequeue_from).
+            pub fn dequeue_ordered(
+                &self,
+                queue_name: &str,
+                now: i64,
+                namespace: Option<&str>,
+                order: $crate::storage::DispatchOrder,
             ) -> Result<Option<Job>> {
                 self.write_transaction(|conn| {
                     // Narrow candidate scan (no payload/result blobs). Postgres
                     // applies FOR UPDATE SKIP LOCKED so concurrent workers claim
                     // disjoint rows; SQLite serializes writers via BEGIN IMMEDIATE.
-                    let candidates: Vec<NarrowJobRow> =
-                        Self::scan_dequeue_candidates(conn, queue_name, now, namespace, 100)?;
+                    let candidates: Vec<NarrowJobRow> = Self::scan_dequeue_candidates(
+                        conn, queue_name, now, namespace, 100, order,
+                    )?;
 
                     for row in candidates {
                         // Skip expired jobs — archive them as cancelled. The
@@ -613,15 +631,18 @@ macro_rules! impl_diesel_job_ops {
                 })
             }
 
-            /// Dequeue from multiple queues, checking each in order.
+            /// Dequeue from multiple queues, checking each in order. Each queue
+            /// is scanned with its own dispatch order (`orders`, default Fifo).
             pub fn dequeue_from(
                 &self,
                 queues: &[String],
                 now: i64,
                 namespace: Option<&str>,
+                orders: &std::collections::HashMap<String, $crate::storage::DispatchOrder>,
             ) -> Result<Option<Job>> {
                 for queue_name in queues {
-                    if let Some(job) = self.dequeue(queue_name, now, namespace)? {
+                    let order = $crate::storage::order_for(orders, queue_name);
+                    if let Some(job) = self.dequeue_ordered(queue_name, now, namespace, order)? {
                         return Ok(Some(job));
                     }
                 }
@@ -643,6 +664,26 @@ macro_rules! impl_diesel_job_ops {
                 now: i64,
                 namespace: Option<&str>,
                 max: usize,
+            ) -> Result<Vec<Job>> {
+                self.dequeue_batch_ordered(
+                    queue_name,
+                    now,
+                    namespace,
+                    max,
+                    $crate::storage::DispatchOrder::Fifo,
+                )
+            }
+
+            /// [`dequeue_batch`](Self::dequeue_batch) with an explicit dispatch
+            /// order — the per-queue path used by
+            /// [`dequeue_batch_from`](Self::dequeue_batch_from).
+            pub fn dequeue_batch_ordered(
+                &self,
+                queue_name: &str,
+                now: i64,
+                namespace: Option<&str>,
+                max: usize,
+                order: $crate::storage::DispatchOrder,
             ) -> Result<Vec<Job>> {
                 if max == 0 {
                     return Ok(Vec::new());
@@ -658,7 +699,7 @@ macro_rules! impl_diesel_job_ops {
                     // Postgres applies FOR UPDATE SKIP LOCKED). Only claimed
                     // winners load their payload inline below.
                     let candidates: Vec<NarrowJobRow> = Self::scan_dequeue_candidates(
-                        conn, queue_name, now, namespace, scan_limit,
+                        conn, queue_name, now, namespace, scan_limit, order,
                     )?;
 
                     let mut claimed: Vec<Job> = Vec::with_capacity(max.min(candidates.len()));
@@ -720,21 +761,25 @@ macro_rules! impl_diesel_job_ops {
             }
 
             /// Claim up to `max` ready jobs across the given queues, checking
-            /// each in order until the budget is exhausted.
+            /// each in order until the budget is exhausted. Each queue uses its
+            /// own dispatch order (`orders`, default Fifo).
             pub fn dequeue_batch_from(
                 &self,
                 queues: &[String],
                 now: i64,
                 namespace: Option<&str>,
                 max: usize,
+                orders: &std::collections::HashMap<String, $crate::storage::DispatchOrder>,
             ) -> Result<Vec<Job>> {
                 let mut claimed: Vec<Job> = Vec::new();
                 for queue_name in queues {
                     if claimed.len() >= max {
                         break;
                     }
                     let remaining = max - claimed.len();
-                    let mut batch = self.dequeue_batch(queue_name, now, namespace, remaining)?;
+                    let order = $crate::storage::order_for(orders, queue_name);
+                    let mut batch =
+                        self.dequeue_batch_ordered(queue_name, now, namespace, remaining, order)?;
                     claimed.append(&mut batch);
                 }
                 Ok(claimed)

crates/taskito-core/src/storage/mod.rs

@@ -152,6 +152,28 @@ pub struct QueueStats {
     pub cancelled: i64,
 }
 
+/// Order in which same-priority jobs are dispatched from a queue. Priority
+/// always dominates; this only breaks ties. `Fifo` (oldest-first) is the fair
+/// default; `Lifo` (newest-first) is opt-in for freshness-sensitive workloads
+/// that would rather run recent work than clear a stale backlog in order.
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
+pub enum DispatchOrder {
+    /// Oldest-first — the fair default a durable queue is expected to keep.
+    #[default]
+    Fifo,
+    /// Newest-first — opt-in freshness lever for same-priority ties under load.
+    Lifo,
+}
+
+/// Resolve a queue's dispatch order from a queue→order map, defaulting to
+/// `Fifo` for any queue not explicitly set to `Lifo`.
+pub(crate) fn order_for(
+    orders: &std::collections::HashMap<String, DispatchOrder>,
+    queue: &str,
+) -> DispatchOrder {
+    orders.get(queue).copied().unwrap_or_default()
+}
+
 /// Per-subscription backlog/lag snapshot for the pub/sub dashboard. One entry
 /// per *registered* subscription — durable or ephemeral, active or paused —
 /// even at zero backlog, so the dashboard renders the full subscriber list
@@ -376,8 +398,9 @@ macro_rules! impl_storage {
                 queues: &[String],
                 now: i64,
                 namespace: Option<&str>,
+                orders: &std::collections::HashMap<String, $crate::storage::DispatchOrder>,
             ) -> $crate::error::Result<Option<$crate::job::Job>> {
-                self.dequeue_from(queues, now, namespace)
+                self.dequeue_from(queues, now, namespace, orders)
             }
             fn dequeue_batch(
                 &self,
@@ -394,8 +417,9 @@ macro_rules! impl_storage {
                 now: i64,
                 namespace: Option<&str>,
                 max: usize,
+                orders: &std::collections::HashMap<String, $crate::storage::DispatchOrder>,
             ) -> $crate::error::Result<Vec<$crate::job::Job>> {
-                self.dequeue_batch_from(queues, now, namespace, max)
+                self.dequeue_batch_from(queues, now, namespace, max, orders)
             }
             fn complete(
                 &self,
@@ -1099,8 +1123,9 @@ impl Storage for StorageBackend {
         queues: &[String],
         now: i64,
         namespace: Option<&str>,
+        orders: &std::collections::HashMap<String, DispatchOrder>,
     ) -> Result<Option<Job>> {
-        delegate!(self, dequeue_from, queues, now, namespace)
+        delegate!(self, dequeue_from, queues, now, namespace, orders)
     }
     fn dequeue_batch(
         &self,
@@ -1117,8 +1142,17 @@ impl Storage for StorageBackend {
         now: i64,
         namespace: Option<&str>,
         max: usize,
+        orders: &std::collections::HashMap<String, DispatchOrder>,
     ) -> Result<Vec<Job>> {
-        delegate!(self, dequeue_batch_from, queues, now, namespace, max)
+        delegate!(
+            self,
+            dequeue_batch_from,
+            queues,
+            now,
+            namespace,
+            max,
+            orders
+        )
     }
     fn complete(&self, id: &str, result_bytes: Option<Vec<u8>>) -> Result<()> {
         delegate!(self, complete, id, result_bytes)

crates/taskito-core/src/storage/postgres/jobs.rs

@@ -45,24 +45,63 @@ impl PostgresStorage {
         now: i64,
         namespace: Option<&str>,
         limit: i64,
+        order: crate::storage::DispatchOrder,
     ) -> diesel::result::QueryResult<Vec<NarrowJobRow>> {
+        use crate::storage::DispatchOrder;
         let base = jobs::table
             .filter(jobs::queue.eq(queue_name))
             .filter(jobs::status.eq(JobStatus::Pending as i32))
             .filter(jobs::scheduled_at.le(now));
 
-        match namespace {
-            Some(ns) => base
+        // `SKIP LOCKED` needs a concrete (un-boxed) statement, so the namespace
+        // filter and the order direction are each branched rather than built
+        // dynamically. Priority always dominates; the (scheduled_at, id)
+        // tie-break flips with the dispatch order (UUIDv7 `id` = deterministic
+        // time-ordered final key).
+        match (namespace, order) {
+            (Some(ns), DispatchOrder::Fifo) => base
                 .filter(jobs::namespace.eq(ns))
-                .order((jobs::priority.desc(), jobs::scheduled_at.asc()))
+                .order((
+                    jobs::priority.desc(),
+                    jobs::scheduled_at.asc(),
+                    jobs::id.asc(),
+                ))
                 .limit(limit)
                 .select(NarrowJobRow::as_select())
                 .for_update()
                 .skip_locked()
                 .load(conn),
-            None => base
+            (Some(ns), DispatchOrder::Lifo) => base
+                .filter(jobs::namespace.eq(ns))
+                .order((
+                    jobs::priority.desc(),
+                    jobs::scheduled_at.desc(),
+                    jobs::id.desc(),
+                ))
+                .limit(limit)
+                .select(NarrowJobRow::as_select())
+                .for_update()
+                .skip_locked()
+                .load(conn),
+            (None, DispatchOrder::Fifo) => base
+                .filter(jobs::namespace.is_null())
+                .order((
+                    jobs::priority.desc(),
+                    jobs::scheduled_at.asc(),
+                    jobs::id.asc(),
+                ))
+                .limit(limit)
+                .select(NarrowJobRow::as_select())
+                .for_update()
+                .skip_locked()
+                .load(conn),
+            (None, DispatchOrder::Lifo) => base
                 .filter(jobs::namespace.is_null())
-                .order((jobs::priority.desc(), jobs::scheduled_at.asc()))
+                .order((
+                    jobs::priority.desc(),
+                    jobs::scheduled_at.desc(),
+                    jobs::id.desc(),
+                ))
                 .limit(limit)
                 .select(NarrowJobRow::as_select())
                 .for_update()

crates/taskito-core/src/storage/redis_backend/jobs/dequeue.rs

@@ -165,12 +165,17 @@ impl RedisStorage {
         Ok(None)
     }
 
-    /// [`dequeue`](Self::dequeue) across several queues, checked in order.
+    /// Dequeue across queues. `orders` is accepted for cross-backend signature
+    /// parity but **ignored**: Redis packs priority and `scheduled_at` into a
+    /// single ZSET score, so per-priority LIFO would need a second score-inverted
+    /// sorted set (a backfill migration of every pending row). Documented as an
+    /// exception, like the `list_jobs_after` Redis note; Redis stays FIFO.
     pub fn dequeue_from(
         &self,
         queues: &[String],
         now: i64,
         namespace: Option<&str>,
+        _orders: &std::collections::HashMap<String, crate::storage::DispatchOrder>,
     ) -> Result<Option<Job>> {
         for queue_name in queues {
             if let Some(job) = self.dequeue(queue_name, now, namespace)? {
@@ -307,13 +312,15 @@ impl RedisStorage {
     }
 
     /// Claim up to `max` ready jobs across the given queues, checking each in
-    /// order until the budget is exhausted.
+    /// order until the budget is exhausted. `orders` is accepted but ignored —
+    /// see [`dequeue_from`](Self::dequeue_from) for why Redis stays FIFO.
     pub fn dequeue_batch_from(
         &self,
         queues: &[String],
         now: i64,
         namespace: Option<&str>,
         max: usize,
+        _orders: &std::collections::HashMap<String, crate::storage::DispatchOrder>,
     ) -> Result<Vec<Job>> {
         let mut claimed: Vec<Job> = Vec::new();
         for queue_name in queues {