diff --git a/crates/taskito-core/src/scheduler/mod.rs b/crates/taskito-core/src/scheduler/mod.rs index 15931a7c..3a979fcb 100644 --- a/crates/taskito-core/src/scheduler/mod.rs +++ b/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>, + /// 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, queues: Vec, config: SchedulerConfig, shutdown: Arc, @@ -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 { diff --git a/crates/taskito-core/src/scheduler/poller.rs b/crates/taskito-core/src/scheduler/poller.rs index 18bc462a..f3869193 100644 --- a/crates/taskito-core/src/scheduler/poller.rs +++ b/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); diff --git a/crates/taskito-core/src/storage/diesel_common/jobs.rs b/crates/taskito-core/src/storage/diesel_common/jobs.rs index 27d54b38..f37d511d 100644 --- a/crates/taskito-core/src/storage/diesel_common/jobs.rs +++ b/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> { + 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> { 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 = - Self::scan_dequeue_candidates(conn, queue_name, now, namespace, 100)?; + let candidates: Vec = 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, ) -> Result> { 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> { + 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> { 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 = Self::scan_dequeue_candidates( - conn, queue_name, now, namespace, scan_limit, + conn, queue_name, now, namespace, scan_limit, order, )?; let mut claimed: Vec = Vec::with_capacity(max.min(candidates.len())); @@ -720,13 +761,15 @@ 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, ) -> Result> { let mut claimed: Vec = Vec::new(); for queue_name in queues { @@ -734,7 +777,9 @@ macro_rules! impl_diesel_job_ops { 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) diff --git a/crates/taskito-core/src/storage/mod.rs b/crates/taskito-core/src/storage/mod.rs index 198eca75..a055297a 100644 --- a/crates/taskito-core/src/storage/mod.rs +++ b/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, + 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, ) -> $crate::error::Result> { - 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, ) -> $crate::error::Result> { - 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, ) -> Result> { - 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, ) -> Result> { - 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>) -> Result<()> { delegate!(self, complete, id, result_bytes) diff --git a/crates/taskito-core/src/storage/postgres/jobs.rs b/crates/taskito-core/src/storage/postgres/jobs.rs index 3b1b3454..c12602a0 100644 --- a/crates/taskito-core/src/storage/postgres/jobs.rs +++ b/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> { + 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() diff --git a/crates/taskito-core/src/storage/redis_backend/jobs/dequeue.rs b/crates/taskito-core/src/storage/redis_backend/jobs/dequeue.rs index 96b37bde..a2898c1a 100644 --- a/crates/taskito-core/src/storage/redis_backend/jobs/dequeue.rs +++ b/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, ) -> Result> { 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, ) -> Result> { let mut claimed: Vec = Vec::new(); for queue_name in queues { diff --git a/crates/taskito-core/src/storage/sqlite/jobs.rs b/crates/taskito-core/src/storage/sqlite/jobs.rs index 72daaa77..0625bf15 100644 --- a/crates/taskito-core/src/storage/sqlite/jobs.rs +++ b/crates/taskito-core/src/storage/sqlite/jobs.rs @@ -44,18 +44,33 @@ impl SqliteStorage { now: i64, namespace: Option<&str>, limit: i64, + order: crate::storage::DispatchOrder, ) -> diesel::result::QueryResult> { let mut query = jobs::table .filter(jobs::queue.eq(queue_name)) .filter(jobs::status.eq(JobStatus::Pending as i32)) .filter(jobs::scheduled_at.le(now)) - .order((jobs::priority.desc(), jobs::scheduled_at.asc())) .limit(limit) .into_boxed(); query = match namespace { Some(ns) => query.filter(jobs::namespace.eq(ns)), None => query.filter(jobs::namespace.is_null()), }; + // Priority always wins; the (scheduled_at, id) tie-break flips with the + // dispatch order. `id` is a UUIDv7, so it is a deterministic time-ordered + // final key — without it, same-`scheduled_at` ties are engine-order. + query = match order { + crate::storage::DispatchOrder::Fifo => query.order(( + jobs::priority.desc(), + jobs::scheduled_at.asc(), + jobs::id.asc(), + )), + crate::storage::DispatchOrder::Lifo => query.order(( + jobs::priority.desc(), + jobs::scheduled_at.desc(), + jobs::id.desc(), + )), + }; query.select(NarrowJobRow::as_select()).load(conn) } } diff --git a/crates/taskito-core/src/storage/sqlite/tests.rs b/crates/taskito-core/src/storage/sqlite/tests.rs index 63ffee44..1fe7956c 100644 --- a/crates/taskito-core/src/storage/sqlite/tests.rs +++ b/crates/taskito-core/src/storage/sqlite/tests.rs @@ -237,7 +237,13 @@ fn test_dequeue_batch_from_across_queues() { let queues = vec!["qa".to_string(), "qb".to_string()]; let claimed = storage - .dequeue_batch_from(&queues, now_millis() + 1000, None, 10) + .dequeue_batch_from( + &queues, + now_millis() + 1000, + None, + 10, + &std::collections::HashMap::new(), + ) .unwrap(); assert_eq!(claimed.len(), 3, "claims across both queues"); @@ -836,6 +842,85 @@ fn test_cascade_cancel_on_dlq() { assert!(b.error.unwrap().contains("dependency failed")); } +#[test] +fn test_dequeue_lifo_vs_fifo_order() { + use crate::storage::DispatchOrder; + use std::collections::HashMap; + + let storage = test_storage(); + let base = now_millis(); + let now = base + 1000; + + // Same priority, staggered eligibility: index 4 is the newest. + for i in 0..5i64 { + let mut job = make_job("t"); + job.queue = "lifoq".to_string(); + job.scheduled_at = base + i; + storage.enqueue(job).unwrap(); + } + let mut orders = HashMap::new(); + orders.insert("lifoq".to_string(), DispatchOrder::Lifo); + let jobs = storage + .dequeue_batch_from(&["lifoq".to_string()], now, None, 5, &orders) + .unwrap(); + let sched: Vec = jobs.iter().map(|j| j.scheduled_at).collect(); + assert_eq!( + sched, + vec![base + 4, base + 3, base + 2, base + 1, base], + "LIFO dispatches newest-first within a priority" + ); + + // FIFO (default, empty map) keeps oldest-first. + for i in 0..5i64 { + let mut job = make_job("t"); + job.queue = "fifoq".to_string(); + job.scheduled_at = base + i; + storage.enqueue(job).unwrap(); + } + let jobs = storage + .dequeue_batch_from(&["fifoq".to_string()], now, None, 5, &HashMap::new()) + .unwrap(); + let sched: Vec = jobs.iter().map(|j| j.scheduled_at).collect(); + assert_eq!( + sched, + vec![base, base + 1, base + 2, base + 3, base + 4], + "FIFO dispatches oldest-first (the default)" + ); +} + +#[test] +fn test_dequeue_lifo_priority_still_dominates() { + use crate::storage::DispatchOrder; + use std::collections::HashMap; + + let storage = test_storage(); + let base = now_millis(); + let now = base + 1000; + + // A high-priority OLD job and a low-priority NEW job on a LIFO queue. + let mut old_high = make_job("t"); + old_high.queue = "pq".to_string(); + old_high.priority = 10; + old_high.scheduled_at = base; // oldest + storage.enqueue(old_high).unwrap(); + + let mut new_low = make_job("t"); + new_low.queue = "pq".to_string(); + new_low.priority = 0; + new_low.scheduled_at = base + 100; // newest + storage.enqueue(new_low).unwrap(); + + let mut orders = HashMap::new(); + orders.insert("pq".to_string(), DispatchOrder::Lifo); + let jobs = storage + .dequeue_batch_from(&["pq".to_string()], now, None, 2, &orders) + .unwrap(); + // Priority wins even under LIFO: the high-priority job goes first despite + // being older; LIFO only reorders same-priority ties. + assert_eq!(jobs[0].priority, 10); + assert_eq!(jobs[1].priority, 0); +} + #[test] fn test_count_running_by_task() { let storage = test_storage(); diff --git a/crates/taskito-core/src/storage/traits.rs b/crates/taskito-core/src/storage/traits.rs index d98e92a1..a86351b0 100644 --- a/crates/taskito-core/src/storage/traits.rs +++ b/crates/taskito-core/src/storage/traits.rs @@ -4,7 +4,7 @@ use crate::storage::records::{ CircuitBreakerState, JobError, LockInfo, NewPeriodicTask, NewSubscription, PeriodicTask, RateLimitState, ReplayEntry, Subscription, TaskLogEntry, TaskMetric, WorkerInfo, }; -use crate::storage::{DeadJob, QueueStats, SubscriptionBacklogStats}; +use crate::storage::{DeadJob, DispatchOrder, QueueStats, SubscriptionBacklogStats}; /// Trait abstracting the storage backend for the task queue. /// @@ -27,12 +27,14 @@ pub trait Storage: Send + Sync + Clone { /// to `Running`. `None` when no job is eligible. `namespace = None` /// matches only namespace-less jobs. fn dequeue(&self, queue_name: &str, now: i64, namespace: Option<&str>) -> Result>; - /// [`dequeue`](Self::dequeue) across several queues, checked in order. + /// [`dequeue`](Self::dequeue) across several queues, checked in order. Each + /// queue uses its dispatch order from `orders` (absent = the `Fifo` default). fn dequeue_from( &self, queues: &[String], now: i64, namespace: Option<&str>, + orders: &std::collections::HashMap, ) -> Result>; /// Atomically claim up to `max` ready jobs from a single queue in one /// transaction. Returns the claimed jobs (now in `Running` state). May @@ -45,13 +47,15 @@ pub trait Storage: Send + Sync + Clone { max: usize, ) -> Result>; /// 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. Each queue uses its dispatch order + /// from `orders` (absent = the `Fifo` default). fn dequeue_batch_from( &self, queues: &[String], now: i64, namespace: Option<&str>, max: usize, + orders: &std::collections::HashMap, ) -> Result>; /// Mark a job completed with its result, moving it from `jobs` into /// `archived_jobs` in one transaction. diff --git a/crates/taskito-core/tests/rust/storage_tests.rs b/crates/taskito-core/tests/rust/storage_tests.rs index af5b4355..951c2d1e 100644 --- a/crates/taskito-core/tests/rust/storage_tests.rs +++ b/crates/taskito-core/tests/rust/storage_tests.rs @@ -1323,10 +1323,35 @@ fn test_enqueue_unique_batch(s: &impl Storage) { ); } +/// A `Lifo` orders map plumbs through `dequeue_batch_from` on every backend and +/// claims exactly the eligible jobs. Order is asserted per-backend in the +/// SQLite unit tests; Redis is a documented FIFO fallback, so this shared test +/// only checks the set of claimed jobs, not their order. +fn test_dispatch_order_lifo_map(s: &impl Storage) { + use std::collections::HashMap; + let q = "q-dispatch-order"; + let mut ids = std::collections::HashSet::new(); + for _ in 0..4 { + ids.insert(s.enqueue(make_job(q, "ord")).unwrap().id); + } + let mut orders = HashMap::new(); + orders.insert(q.to_string(), taskito_core::storage::DispatchOrder::Lifo); + let claimed = s + .dequeue_batch_from(&[q.to_string()], now_millis() + 1000, None, 10, &orders) + .unwrap(); + let claimed_ids: std::collections::HashSet = + claimed.into_iter().map(|j| j.id).collect(); + assert_eq!( + claimed_ids, ids, + "LIFO map claims exactly the eligible jobs" + ); +} + fn run_storage_tests(s: &impl Storage) { test_enqueue_and_get(s); test_dequeue(s); test_dequeue_batch(s); + test_dispatch_order_lifo_map(s); test_complete(s); test_fail(s); test_retry(s); diff --git a/crates/taskito-java/src/convert.rs b/crates/taskito-java/src/convert.rs index 0824be0f..451eab90 100644 --- a/crates/taskito-java/src/convert.rs +++ b/crates/taskito-java/src/convert.rs @@ -300,19 +300,20 @@ pub struct WorkerOptions { pub subscriptions: Option>, /// Per-table retention windows for auto-cleanup, in seconds. pub retention: Option, - /// Per-queue config registered with the scheduler at start. Currently - /// carries opt-in CoDel load shedding. + /// Per-queue config registered with the scheduler at start. Carries opt-in + /// CoDel load shedding and dispatch order. pub queue_configs: Option>, } /// Per-queue scheduler config. Only queues with a value here are registered; an -/// entry with no positive CoDel bounds is ignored. +/// entry with no positive CoDel bounds and a non-`lifo` order is a no-op. #[derive(Deserialize, Default)] #[serde(rename_all = "camelCase", default)] pub struct QueueConfigSpec { pub name: String, pub codel_target_ms: Option, pub codel_interval_ms: Option, + pub dispatch_order: Option, } /// Per-table retention windows in seconds. An unset field keeps that table diff --git a/crates/taskito-java/src/worker.rs b/crates/taskito-java/src/worker.rs index d41a87ff..e0d4f30e 100644 --- a/crates/taskito-java/src/worker.rs +++ b/crates/taskito-java/src/worker.rs @@ -123,8 +123,16 @@ fn start_worker( for (name, policy) in task_policies { scheduler.register_task(name, policy); } - for (name, codel) in build_queue_codels(options.queue_configs.take()) { - scheduler.register_queue_codel(name, codel); + for spec in options.queue_configs.take().unwrap_or_default() { + if let Some(codel) = queue_codel_from_spec(&spec) { + scheduler.register_queue_codel(spec.name.clone(), codel); + } + if spec.dispatch_order.as_deref() == Some("lifo") { + scheduler.register_queue_dispatch_order( + spec.name, + taskito_core::storage::DispatchOrder::Lifo, + ); + } } let scheduler = Arc::new(scheduler); let shutdown = scheduler.shutdown_handle(); @@ -323,23 +331,18 @@ fn build_task_policies( /// Build the per-queue CoDel configs. Only queues with positive bounds are /// registered; anything else is dropped so an empty spec is a harmless no-op. -fn build_queue_codels(configs: Option>) -> Vec<(String, CodelConfig)> { - configs - .unwrap_or_default() - .into_iter() - .filter_map( - |spec| match (spec.codel_target_ms, spec.codel_interval_ms) { - (Some(target_ms), Some(interval_ms)) if target_ms > 0 && interval_ms > 0 => Some(( - spec.name, - CodelConfig { - target_ms, - interval_ms, - }, - )), - _ => None, - }, - ) - .collect() +/// Extract a valid CoDel config from a queue spec: both bounds set and positive, +/// else `None` (the queue keeps default dispatch, no shedding). +fn queue_codel_from_spec(spec: &QueueConfigSpec) -> Option { + match (spec.codel_target_ms, spec.codel_interval_ms) { + (Some(target_ms), Some(interval_ms)) if target_ms > 0 && interval_ms > 0 => { + Some(CodelConfig { + target_ms, + interval_ms, + }) + } + _ => None, + } } /// Parse an optional rate spec, naming the offending task and option so a typo diff --git a/crates/taskito-node/src/config.rs b/crates/taskito-node/src/config.rs index 97ce31c9..3aea223f 100644 --- a/crates/taskito-node/src/config.rs +++ b/crates/taskito-node/src/config.rs @@ -153,6 +153,10 @@ pub struct QueueConfigInput { /// the wait must stay above target before jobs are shed to the DLQ. pub codel_target_ms: Option, pub codel_interval_ms: Option, + /// `"lifo"` dispatches newest-first within a priority under overload; + /// anything else (or unset) keeps the fair `"fifo"` default. Honored on + /// SQLite/Postgres; Redis is FIFO-only. + pub dispatch_order: Option, } /// Opt-in mesh overlay for a worker: decentralized peer discovery (SWIM gossip) diff --git a/crates/taskito-node/src/worker.rs b/crates/taskito-node/src/worker.rs index 7540a64f..cf01db8a 100644 --- a/crates/taskito-node/src/worker.rs +++ b/crates/taskito-node/src/worker.rs @@ -130,6 +130,12 @@ pub fn start_worker( if let Some(codel) = crate::convert::queue_codel(input) { scheduler.register_queue_codel(input.name.clone(), codel); } + if input.dispatch_order.as_deref() == Some("lifo") { + scheduler.register_queue_dispatch_order( + input.name.clone(), + taskito_core::storage::DispatchOrder::Lifo, + ); + } scheduler.register_queue_config(input.name.clone(), crate::convert::queue_config(input)?); } let scheduler = Arc::new(scheduler); diff --git a/crates/taskito-python/src/py_queue/worker.rs b/crates/taskito-python/src/py_queue/worker.rs index b9baa48c..d44cbd30 100644 --- a/crates/taskito-python/src/py_queue/worker.rs +++ b/crates/taskito-python/src/py_queue/worker.rs @@ -445,6 +445,12 @@ impl PyQueue { if let Some(codel) = parse_codel(&cfg) { scheduler.register_queue_codel(queue_name.clone(), codel); } + if let Some("lifo") = cfg.get("dispatch_order").and_then(|v| v.as_str()) { + scheduler.register_queue_dispatch_order( + queue_name.clone(), + taskito_core::storage::DispatchOrder::Lifo, + ); + } scheduler.register_queue_config( queue_name, taskito_core::scheduler::QueueConfig { diff --git a/sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java b/sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java index a2e01a48..7d9b884f 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java @@ -96,6 +96,8 @@ final class DefaultTaskito implements Taskito { private final Map maxPending = new ConcurrentHashMap<>(); // Opt-in per-queue CoDel config (queue -> [targetMs, intervalMs]). private final Map codelConfigs = new ConcurrentHashMap<>(); + // Opt-in per-queue dispatch order (queue -> "lifo"). Absent = "fifo". + private final Map dispatchOrders = new ConcurrentHashMap<>(); DefaultTaskito(QueueBackend backend, Serializer serializer, Map codecs) { this.backend = backend; @@ -205,17 +207,38 @@ public Taskito codel(String queue, long targetMs, long intervalMs) { return this; } - /** Serialize the per-queue CoDel configs into the worker-options wire shape. */ + @Override + public Taskito dispatchOrder(String queue, String order) { + if (!"fifo".equals(order) && !"lifo".equals(order)) { + throw new IllegalArgumentException("order must be 'fifo' or 'lifo'"); + } + dispatchOrders.put(queue, order); + return this; + } + + /** + * Serialize the per-queue scheduler config (CoDel + dispatch order) into the + * worker-options wire shape, one spec per queue that has either set. + */ private List> encodeQueueConfigs() { - List> specs = new ArrayList<>(codelConfigs.size()); + Map> byQueue = new LinkedHashMap<>(); codelConfigs.forEach((queue, codel) -> { - Map spec = new LinkedHashMap<>(); - spec.put("name", queue); + Map spec = byQueue.computeIfAbsent(queue, DefaultTaskito::queueSpec); spec.put("codelTargetMs", codel[0]); spec.put("codelIntervalMs", codel[1]); - specs.add(spec); }); - return specs; + dispatchOrders.forEach((queue, order) -> { + Map spec = byQueue.computeIfAbsent(queue, DefaultTaskito::queueSpec); + spec.put("dispatchOrder", order); + }); + return new ArrayList<>(byQueue.values()); + } + + /** A fresh wire spec carrying just the queue name; callers add their keys. */ + private static Map queueSpec(String queue) { + Map spec = new LinkedHashMap<>(); + spec.put("name", queue); + return spec; } @Override diff --git a/sdks/java/src/main/java/org/byteveda/taskito/Taskito.java b/sdks/java/src/main/java/org/byteveda/taskito/Taskito.java index 0babd332..e8cf309a 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/Taskito.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/Taskito.java @@ -140,6 +140,15 @@ static Builder builder() { */ Taskito codel(String queue, long targetMs, long intervalMs); + /** + * Set a queue's same-priority dispatch order. {@code "lifo"} runs + * newest-first under overload (a freshness lever); {@code "fifo"} (default) + * is the fair oldest-first ordering. Priority always dominates. Honored on + * SQLite/Postgres; the Redis backend is FIFO-only. Takes effect for workers + * started after this call. Returns {@code this}. + */ + Taskito dispatchOrder(String queue, String order); + // ── Producer ──────────────────────────────────────────────────── /** Enqueue a typed payload using the task's default options; returns the job id. */ diff --git a/sdks/java/src/main/java/org/byteveda/taskito/worker/Worker.java b/sdks/java/src/main/java/org/byteveda/taskito/worker/Worker.java index 0fbb8929..2c4f729f 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/worker/Worker.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/worker/Worker.java @@ -277,9 +277,9 @@ public Builder subscriptions(List subscriptions) { } /** - * Late-bound per-queue scheduler config in wire shape (currently CoDel). - * Supplied by the owning {@code Taskito} via {@code Taskito.worker()} and - * resolved at {@code start()}, so config set *after* the builder was + * Late-bound per-queue scheduler config in wire shape (CoDel + dispatch + * order). Supplied by the owning {@code Taskito} via {@code Taskito.worker()} + * and resolved at {@code start()}, so config set *after* the builder was * obtained (e.g. {@code Taskito.codel(...)}) is still picked up — matching * that method's documented timing. A manually built worker leaves it empty. */ diff --git a/sdks/java/src/test/java/org/byteveda/taskito/core/DispatchOrderTest.java b/sdks/java/src/test/java/org/byteveda/taskito/core/DispatchOrderTest.java new file mode 100644 index 00000000..e96fc206 --- /dev/null +++ b/sdks/java/src/test/java/org/byteveda/taskito/core/DispatchOrderTest.java @@ -0,0 +1,72 @@ +package org.byteveda.taskito.core; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; + +import java.nio.file.Path; +import java.time.Duration; +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; +import org.byteveda.taskito.Taskito; +import org.byteveda.taskito.task.Task; +import org.byteveda.taskito.worker.Worker; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; +import org.junit.jupiter.api.io.TempDir; + +/** + * S25 — opt-in LIFO dispatch order. Jobs are enqueued before the worker starts, + * so they pile up pending; a concurrency-1 worker then dispatches them one at a + * time and we record the order. UUIDv7 ids make same-millisecond ties + * deterministic, so LIFO is exactly newest-first. + */ +class DispatchOrderTest { + + @Test + void validatesOrder(@TempDir Path dir) { + try (Taskito queue = + Taskito.builder().url(dir.resolve("v.db").toString()).open()) { + assertThrows(IllegalArgumentException.class, () -> queue.dispatchOrder("default", "sideways")); + } + } + + private static List runAndRecord(Taskito queue, int total) throws Exception { + List order = new CopyOnWriteArrayList<>(); + Task rec = Task.of("rec", Integer.class); + for (int i = 0; i < total; i++) { + queue.enqueue(rec, i); + } + try (Worker worker = queue.worker() + .concurrency(1) + .batchSize(1) + .handle(rec, order::add) + .start()) { + long deadline = System.nanoTime() + Duration.ofSeconds(20).toNanos(); + while (System.nanoTime() < deadline && order.size() < total) { + Thread.sleep(50); + } + } + return order; + } + + @Test + @Timeout(40) + void lifoDispatchesNewestFirst(@TempDir Path dir) throws Exception { + try (Taskito queue = + Taskito.builder().url(dir.resolve("lifo.db").toString()).open()) { + queue.dispatchOrder("default", "lifo"); + List order = runAndRecord(queue, 6); + assertEquals(List.of(5, 4, 3, 2, 1, 0), order, "LIFO runs newest-first"); + } + } + + @Test + @Timeout(40) + void fifoIsTheDefault(@TempDir Path dir) throws Exception { + try (Taskito queue = + Taskito.builder().url(dir.resolve("fifo.db").toString()).open()) { + List order = runAndRecord(queue, 6); + assertEquals(List.of(0, 1, 2, 3, 4, 5), order, "FIFO runs oldest-first by default"); + } + } +} diff --git a/sdks/node/src/types.ts b/sdks/node/src/types.ts index f27e772c..c78ba97f 100644 --- a/sdks/node/src/types.ts +++ b/sdks/node/src/types.ts @@ -180,6 +180,12 @@ export interface QueueLimits { * rather than running them stale. A transient spike is never shed. */ codel?: { targetMs: number; intervalMs: number }; + /** + * Same-priority dispatch order. `"lifo"` runs newest-first under overload (a + * freshness lever); `"fifo"` (default) is the fair oldest-first ordering. + * Priority always dominates. Honored on SQLite/Postgres; Redis is FIFO-only. + */ + dispatchOrder?: "fifo" | "lifo"; } /** A task handler plus its registration options. */ diff --git a/sdks/node/src/worker.ts b/sdks/node/src/worker.ts index 572a5d29..1c1c46a3 100644 --- a/sdks/node/src/worker.ts +++ b/sdks/node/src/worker.ts @@ -358,5 +358,6 @@ function buildQueueConfigs(limits: ReadonlyMap): QueueConfi rateLimit: limit.rateLimit, codelTargetMs: limit.codel?.targetMs, codelIntervalMs: limit.codel?.intervalMs, + dispatchOrder: limit.dispatchOrder, })); } diff --git a/sdks/node/test/core/dispatch-order.test.ts b/sdks/node/test/core/dispatch-order.test.ts new file mode 100644 index 00000000..3ac74693 --- /dev/null +++ b/sdks/node/test/core/dispatch-order.test.ts @@ -0,0 +1,52 @@ +// S25 — opt-in LIFO dispatch order. Jobs are enqueued before the worker starts, +// so all pile up pending; a concurrency-1 worker then dispatches them one at a +// time and we record the order. UUIDv7 ids make same-millisecond ties +// deterministic, so LIFO is exactly newest-first. + +import { mkdtempSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterEach, expect, it } from "vitest"; +import { Queue, type Worker } from "../../src/index"; + +let worker: Worker | undefined; + +afterEach(() => { + worker?.stop(); + worker = undefined; +}); + +function newQueue(): Queue { + const dbPath = join(mkdtempSync(join(tmpdir(), "taskito-order-")), "queue.db"); + return new Queue({ dbPath }); +} + +const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms)); + +async function runAndRecord(queue: Queue, total: number): Promise { + const order: number[] = []; + queue.task("rec", (n: number) => { + order.push(n); + }); + for (let i = 0; i < total; i++) queue.enqueue("rec", [i]); + + worker = queue.runWorker({ concurrency: 1, batchSize: 1 }); + const deadline = Date.now() + 20_000; + while (Date.now() < deadline && order.length < total) { + await sleep(50); + } + return order; +} + +it("LIFO dispatches newest-first", async () => { + const queue = newQueue(); + queue.configureQueue("default", { dispatchOrder: "lifo" }); + const order = await runAndRecord(queue, 6); + expect(order).toEqual([5, 4, 3, 2, 1, 0]); +}); + +it("FIFO is the default", async () => { + const queue = newQueue(); + const order = await runAndRecord(queue, 6); + expect(order).toEqual([0, 1, 2, 3, 4, 5]); +}); diff --git a/sdks/python/taskito/mixins/runtime_config.py b/sdks/python/taskito/mixins/runtime_config.py index 4ba58c5c..1165e10d 100644 --- a/sdks/python/taskito/mixins/runtime_config.py +++ b/sdks/python/taskito/mixins/runtime_config.py @@ -124,3 +124,24 @@ def set_queue_codel(self, queue_name: str, target_ms: int, interval_ms: int) -> config = self._queue_configs.setdefault(queue_name, {}) config["codel_target_ms"] = target_ms config["codel_interval_ms"] = interval_ms + + def set_queue_dispatch_order(self, queue_name: str, order: str) -> None: + """Set a queue's dispatch order for same-priority jobs. + + ``"fifo"`` (default) runs oldest-first — the fair ordering a durable + queue is expected to keep. ``"lifo"`` runs newest-first, a freshness + lever for workloads that would rather run recent work than clear a stale + backlog in order under overload. Priority always dominates; this only + breaks ties. Takes effect at ``run_worker``. + + Note: honored on SQLite and PostgreSQL. The Redis backend is FIFO-only + (its single-score index can't express per-priority LIFO without a + second sorted set); ``"lifo"`` is accepted but ignored there. + + Args: + queue_name: Queue name (e.g. ``"default"``). + order: ``"fifo"`` or ``"lifo"``. + """ + if order not in ("fifo", "lifo"): + raise ValueError("order must be 'fifo' or 'lifo'") + self._queue_configs.setdefault(queue_name, {})["dispatch_order"] = order diff --git a/sdks/python/tests/core/test_dispatch_order.py b/sdks/python/tests/core/test_dispatch_order.py new file mode 100644 index 00000000..258e233c --- /dev/null +++ b/sdks/python/tests/core/test_dispatch_order.py @@ -0,0 +1,68 @@ +"""S25 — opt-in LIFO dispatch order. + +Jobs are enqueued before the worker starts, so all pile up pending; a +concurrency-1 worker then dispatches them one at a time and we record the order. +Same-priority ties break by ``(scheduled_at, id)`` direction — with UUIDv7 ids, +LIFO is deterministically newest-first even when jobs share a millisecond. +""" + +from __future__ import annotations + +import threading +import time +from pathlib import Path + +import pytest + +from taskito import Queue + + +def _run_and_record(queue: Queue, total: int) -> list[int]: + order: list[int] = [] + lock = threading.Lock() + + @queue.task(name="rec") + def rec(n: int) -> None: + with lock: + order.append(n) + + # Enqueue everything before the worker exists so the queue is fully backed + # up when dispatch begins. + for i in range(total): + queue.enqueue("rec", args=(i,)) + + worker = threading.Thread(target=queue.run_worker, daemon=True) + worker.start() + try: + deadline = time.time() + 20 + while time.time() < deadline: + with lock: + if len(order) == total: + break + time.sleep(0.05) + finally: + queue.shutdown() + worker.join(timeout=5) + + with lock: + return list(order) + + +def test_set_queue_dispatch_order_validates() -> None: + q = Queue(db_path=":memory:", workers=1) + with pytest.raises(ValueError): + q.set_queue_dispatch_order("default", "sideways") + + +def test_lifo_dispatches_newest_first(tmp_path: Path) -> None: + q = Queue(db_path=str(tmp_path / "lifo.db"), workers=1, scheduler_batch_size=1) + q.set_queue_dispatch_order("default", "lifo") + order = _run_and_record(q, 6) + assert order == [5, 4, 3, 2, 1, 0], "LIFO runs newest-first" + + +def test_fifo_is_the_default(tmp_path: Path) -> None: + q = Queue(db_path=str(tmp_path / "fifo.db"), workers=1, scheduler_batch_size=1) + # No dispatch_order set — the fair default. + order = _run_and_record(q, 6) + assert order == [0, 1, 2, 3, 4, 5], "FIFO runs oldest-first by default"