From 23122daf655003187f858a97b97699d6569b513d Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Sat, 18 Jul 2026 16:47:41 +0530 Subject: [PATCH 1/7] feat(core): add CoDel load shedding to the dispatcher --- crates/taskito-core/src/scheduler/codel.rs | 152 ++++++++++++++++++ .../taskito-core/src/scheduler/maintenance.rs | 9 ++ crates/taskito-core/src/scheduler/mod.rs | 63 ++++++++ crates/taskito-core/src/scheduler/poller.rs | 56 +++++++ 4 files changed, 280 insertions(+) create mode 100644 crates/taskito-core/src/scheduler/codel.rs diff --git a/crates/taskito-core/src/scheduler/codel.rs b/crates/taskito-core/src/scheduler/codel.rs new file mode 100644 index 00000000..5b03e17a --- /dev/null +++ b/crates/taskito-core/src/scheduler/codel.rs @@ -0,0 +1,152 @@ +//! Controlled Delay (CoDel) load shedding for the dispatcher. +//! +//! Classic CoDel (RFC 8289), adapted from packet queues to task dispatch. A +//! job's *sojourn* is how long it waited past its eligibility (`now - +//! scheduled_at`). While the sojourn stays above `target` for a full `interval`, +//! the controller enters a dropping state and sheds jobs at an increasing rate +//! (`interval / sqrt(count)`), backing off the moment the sojourn recovers. +//! +//! Unlike a plain deadline this does **not** drop during a transient spike — +//! only sustained overload — which is the whole point of using CoDel over a +//! fixed max-age cutoff. Opt-in per queue; when unset the dispatcher never calls +//! in here. + +/// Per-queue CoDel tuning. `target_ms` is the acceptable steady-state sojourn; +/// `interval_ms` is the window the sojourn must stay above target before +/// shedding begins. +#[derive(Debug, Clone, Copy)] +pub struct CodelConfig { + pub target_ms: i64, + pub interval_ms: i64, +} + +/// Mutable controller state for one queue. Advanced once per candidate job in +/// dequeue order; persists across dispatch ticks. +#[derive(Debug, Default)] +pub struct CodelState { + /// Time at which shedding may begin, armed when the sojourn first exceeds + /// target and cleared when it recovers. `0` = not currently above target. + first_above_time: i64, + /// Scheduled time of the next drop while shedding. + drop_next: i64, + /// Drops in the current overload episode; drives the control law. + count: u32, + /// Whether the controller is currently shedding. + dropping: bool, +} + +impl CodelState { + /// Decide whether the candidate job with the given `sojourn_ms` should be + /// shed. Mutates the controller; call once per job in dequeue order. + pub fn should_drop(&mut self, sojourn_ms: i64, now: i64, cfg: &CodelConfig) -> bool { + if sojourn_ms < cfg.target_ms { + // Healthy: leave the dropping state and disarm. + self.first_above_time = 0; + self.dropping = false; + return false; + } + // At or above target. + if self.first_above_time == 0 { + // Just crossed above target — arm the interval before we may drop. + self.first_above_time = now.saturating_add(cfg.interval_ms); + return false; + } + if now < self.first_above_time { + // Above target, but not yet for a full interval — hold. + return false; + } + // Sustained above target for at least one interval: shed. + if !self.dropping { + self.dropping = true; + // A fresh episode restarts the rate at 1, but a quick relapse (soon + // after the last drop) resumes near the prior rate instead of ramping + // from scratch — the standard CoDel re-entry heuristic. + self.count = + if self.count > 2 && now.saturating_sub(self.drop_next) < 8 * cfg.interval_ms { + self.count - 2 + } else { + 1 + }; + self.drop_next = self.control_law(now, cfg); + return true; + } + if now >= self.drop_next { + self.count = self.count.saturating_add(1); + self.drop_next = self.control_law(now, cfg); + return true; + } + false + } + + /// Next drop time: spaced by `interval / sqrt(count)`, so the drop rate rises + /// as an episode persists and the sojourn refuses to fall. + fn control_law(&self, now: i64, cfg: &CodelConfig) -> i64 { + let spacing = (cfg.interval_ms as f64 / (self.count.max(1) as f64).sqrt()) as i64; + now.saturating_add(spacing.max(1)) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + const CFG: CodelConfig = CodelConfig { + target_ms: 100, + interval_ms: 1000, + }; + + #[test] + fn below_target_never_drops() { + let mut s = CodelState::default(); + for i in 0..100 { + let now = i * 50; + assert!(!s.should_drop(10, now, &CFG), "healthy sojourn must pass"); + } + } + + #[test] + fn transient_spike_does_not_drop() { + let mut s = CodelState::default(); + // Above target, but the episode never reaches a full interval before it + // recovers — a plain deadline would have dropped, CoDel must not. + assert!(!s.should_drop(500, 0, &CFG)); // arms first_above_time = 1000 + assert!(!s.should_drop(500, 500, &CFG)); // now < 1000: hold + assert!(!s.should_drop(10, 800, &CFG)); // recovered before the interval + assert!(!s.should_drop(500, 900, &CFG)); // re-arms, still no drop + } + + #[test] + fn sustained_overload_starts_dropping() { + let mut s = CodelState::default(); + assert!(!s.should_drop(500, 0, &CFG)); // arm at now+interval = 1000 + assert!(!s.should_drop(500, 999, &CFG)); // still under the interval + assert!(s.should_drop(500, 1000, &CFG)); // first drop at the interval edge + } + + #[test] + fn recovery_exits_dropping_state() { + let mut s = CodelState::default(); + assert!(!s.should_drop(500, 0, &CFG)); + assert!(s.should_drop(500, 1000, &CFG)); // dropping + // Sojourn recovers: must immediately stop dropping and re-arm cleanly. + assert!(!s.should_drop(10, 1100, &CFG)); + assert!(!s.should_drop(500, 1200, &CFG)); // re-arms, no immediate drop + assert!(!s.should_drop(500, 2100, &CFG)); // still under the new interval + assert!(s.should_drop(500, 2200, &CFG)); // drops again after a full interval + } + + #[test] + fn drop_rate_is_bounded_within_one_tick() { + // Many candidates evaluated at the same `now` while shedding: the control + // law spaces drops in time, so a single instant sheds at most one. + let mut s = CodelState::default(); + assert!(!s.should_drop(500, 0, &CFG)); + assert!(s.should_drop(500, 1000, &CFG)); + for _ in 0..10 { + assert!( + !s.should_drop(500, 1000, &CFG), + "no second drop at the same instant" + ); + } + } +} diff --git a/crates/taskito-core/src/scheduler/maintenance.rs b/crates/taskito-core/src/scheduler/maintenance.rs index e0b54bb1..76d09b12 100644 --- a/crates/taskito-core/src/scheduler/maintenance.rs +++ b/crates/taskito-core/src/scheduler/maintenance.rs @@ -324,6 +324,15 @@ impl Scheduler { let mut retried = 0u64; for entry in &candidates { + // Jobs shed by CoDel were intentionally dropped as stale; never let + // the auto-retry sweep resurrect them. + if entry + .error + .as_deref() + .is_some_and(|e| e.starts_with("codel:")) + { + continue; + } match self.storage.retry_dead(&entry.id) { Ok(new_id) => { info!( diff --git a/crates/taskito-core/src/scheduler/mod.rs b/crates/taskito-core/src/scheduler/mod.rs index 19241957..38c50c0d 100644 --- a/crates/taskito-core/src/scheduler/mod.rs +++ b/crates/taskito-core/src/scheduler/mod.rs @@ -1,3 +1,4 @@ +pub mod codel; mod maintenance; mod poller; mod result_handler; @@ -256,6 +257,13 @@ pub struct Scheduler { circuit_breaker: CircuitBreaker, task_configs: HashMap, queue_configs: HashMap, + /// Opt-in per-queue CoDel tuning, kept out of `QueueConfig` so enabling it + /// never forces every backend to thread a new field. Empty = no shedding. + codel_configs: HashMap, + /// Per-queue CoDel controller state, keyed by queue name. Only touched when + /// 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>, queues: Vec, config: SchedulerConfig, shutdown: Arc, @@ -317,6 +325,8 @@ impl Scheduler { circuit_breaker, task_configs: HashMap::new(), queue_configs: HashMap::new(), + codel_configs: HashMap::new(), + codel_states: Mutex::new(HashMap::new()), queues, config, shutdown: Arc::new(Notify::new()), @@ -471,6 +481,12 @@ impl Scheduler { self.queue_configs.insert(queue_name, config); } + /// Enable opt-in CoDel load shedding on a queue. Orthogonal to + /// [`register_queue_config`] so the common (no-CoDel) path never pays for it. + pub fn register_queue_codel(&mut self, queue_name: String, config: codel::CodelConfig) { + self.codel_configs.insert(queue_name, config); + } + pub fn register_task(&mut self, task_name: String, config: TaskConfig) { if let Some(ref cb_config) = config.circuit_breaker { if let Err(e) = self.circuit_breaker.register(&task_name, cb_config) { @@ -1600,6 +1616,53 @@ mod tests { assert_eq!(pending.len(), 1); } + #[test] + fn test_codel_sheds_stale_job_to_dlq() { + // Drive the real dispatch shed path with controlled time. CoDel arms on + // the first stale candidate, then sheds a later one once the sojourn has + // stayed above target for a full interval — dead-lettering it with a + // reserved `codel:` reason (never dropping during a transient spike). + let mut scheduler = test_scheduler(); + scheduler.register_queue_codel( + "default".to_string(), + codel::CodelConfig { + target_ms: 100, + interval_ms: 1000, + }, + ); + let job = scheduler.storage.enqueue(make_job("stale")).unwrap(); + let base = job.scheduled_at; + + // First stale candidate (sojourn 500 > target 100): only arms, kept. + let kept = scheduler + .codel_admit(vec![job.clone()], base + 500) + .unwrap(); + assert_eq!(kept.len(), 1, "the first stale job arms but is not shed"); + + // A full interval later, still stale: shed to the DLQ. + let kept = scheduler + .codel_admit(vec![job.clone()], base + 500 + 1000) + .unwrap(); + assert!(kept.is_empty(), "a sustained-overload job is shed"); + + let dead = scheduler.storage.list_dead(10, 0).unwrap(); + assert!( + dead.iter() + .any(|d| d.error.as_deref().is_some_and(|e| e.starts_with("codel:"))), + "shed job is dead-lettered with a codel: reason" + ); + } + + #[test] + fn test_codel_leaves_uncapped_queues_untouched() { + // With no CoDel config, the admit filter is a no-op fast path. + let scheduler = test_scheduler(); + let job = scheduler.storage.enqueue(make_job("fresh")).unwrap(); + let ancient = job.scheduled_at + 10_000_000; + let kept = scheduler.codel_admit(vec![job], ancient).unwrap(); + assert_eq!(kept.len(), 1, "no codel config => nothing is ever shed"); + } + #[test] fn test_try_dispatch_reschedules_on_closed_channel() { // Regression: when the worker channel is closed (worker pool diff --git a/crates/taskito-core/src/scheduler/poller.rs b/crates/taskito-core/src/scheduler/poller.rs index 1aabaf50..18bc462a 100644 --- a/crates/taskito-core/src/scheduler/poller.rs +++ b/crates/taskito-core/src/scheduler/poller.rs @@ -109,6 +109,13 @@ impl Scheduler { None => return Ok(false), }; + // CoDel: shed a stale job before claiming (opt-in per queue). The job is + // still Pending here, so the DLQ move is uniform across backends. + let job = match self.codel_admit(vec![job], now)?.pop() { + Some(j) => j, + None => return Ok(false), + }; + // A fresh cache for a single job loads each count at most once — the // same one query the old code issued. let mut counts = GateCounts::default(); @@ -161,6 +168,14 @@ impl Scheduler { return Ok(false); } + // CoDel: shed stale jobs before claiming (opt-in per queue). Dropped + // jobs are still Pending, so the DLQ move is uniform across backends and + // leaves no execution claim to unwind. + let jobs = self.codel_admit(jobs, now)?; + if jobs.is_empty() { + return Ok(false); + } + // Claim the entire batch in one round-trip instead of one per job. A // storage error leaves the batch claim atomic-nothing (failed statement // / rolled-back txn), so degrade to the proven single-job path where @@ -204,6 +219,47 @@ impl Scheduler { Ok(dispatched_any) } + /// CoDel admission: drop stale jobs before they are claimed (opt-in per + /// queue). Returns the jobs that survive; shed jobs are dead-lettered with a + /// reserved `codel:` reason so the auto-retry sweep leaves them alone. + /// + /// A job's sojourn is measured from `scheduled_at` (time waiting *past* its + /// eligibility), so an intentional delay is never counted as staleness. The + /// empty-map short-circuit keeps this free for queues that never opted in. + pub(super) fn codel_admit(&self, jobs: Vec, now: i64) -> Result> { + if self.codel_configs.is_empty() { + return Ok(jobs); + } + let mut states = self.codel_states.lock().unwrap(); + let mut kept = Vec::with_capacity(jobs.len()); + for job in jobs { + let cfg = match self.codel_configs.get(&job.queue).copied() { + Some(cfg) => cfg, + None => { + kept.push(job); + continue; + } + }; + let sojourn = now.saturating_sub(job.scheduled_at).max(0); + let state = states.entry(job.queue.clone()).or_default(); + if state.should_drop(sojourn, now, &cfg) { + let reason = format!( + "codel: sojourn {sojourn}ms exceeded target {}ms under sustained overload", + cfg.target_ms + ); + self.storage + .move_to_dlq(&job, &reason, Some("{\"codel\":true}"))?; + warn!( + "codel shed {} on queue '{}' (sojourn {sojourn}ms)", + job.id, job.queue + ); + } else { + kept.push(job); + } + } + Ok(kept) + } + /// Run the post-dequeue pipeline for a single job: soft pre-claim gates, /// exactly-once claim, then the shared post-claim tail. Returns `Ok(true)` /// if the job was dispatched, `Ok(false)` if it was gated/rolled back. Used From e39402fc5edb2ffa4cc612c8b36b43c68521ca84 Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Sat, 18 Jul 2026 16:47:58 +0530 Subject: [PATCH 2/7] feat(python): add set_queue_codel --- crates/taskito-python/src/py_queue/worker.rs | 18 ++++ sdks/python/taskito/mixins/runtime_config.py | 20 ++++ sdks/python/tests/core/test_codel.py | 105 +++++++++++++++++++ 3 files changed, 143 insertions(+) create mode 100644 sdks/python/tests/core/test_codel.py diff --git a/crates/taskito-python/src/py_queue/worker.rs b/crates/taskito-python/src/py_queue/worker.rs index 3ba4c7b3..46d47760 100644 --- a/crates/taskito-python/src/py_queue/worker.rs +++ b/crates/taskito-python/src/py_queue/worker.rs @@ -16,6 +16,21 @@ use super::PyQueue; use crate::async_worker::AsyncWorkerPool; use crate::py_config::PyTaskConfig; +/// Parse the optional per-queue CoDel config from a `queue_configs` entry. Both +/// `codel_target_ms` and `codel_interval_ms` must be present and positive for +/// shedding to be enabled on the queue. +fn parse_codel(cfg: &serde_json::Value) -> Option { + let target_ms = cfg.get("codel_target_ms").and_then(|v| v.as_i64())?; + let interval_ms = cfg.get("codel_interval_ms").and_then(|v| v.as_i64())?; + if target_ms <= 0 || interval_ms <= 0 { + return None; + } + Some(taskito_core::scheduler::codel::CodelConfig { + target_ms, + interval_ms, + }) +} + /// Mesh-aware scheduler bridge: receives jobs from the scheduler's /// intermediate channel, pushes them into the local deque with affinity /// sorting, then drains the deque to the real dispatcher channel. @@ -427,6 +442,9 @@ impl PyQueue { .get("max_concurrent") .and_then(|v| v.as_i64()) .map(|v| v as i32); + if let Some(codel) = parse_codel(&cfg) { + scheduler.register_queue_codel(queue_name.clone(), codel); + } scheduler.register_queue_config( queue_name, taskito_core::scheduler::QueueConfig { diff --git a/sdks/python/taskito/mixins/runtime_config.py b/sdks/python/taskito/mixins/runtime_config.py index a17f040a..4ba58c5c 100644 --- a/sdks/python/taskito/mixins/runtime_config.py +++ b/sdks/python/taskito/mixins/runtime_config.py @@ -104,3 +104,23 @@ def set_queue_max_pending(self, queue_name: str, max_pending: int) -> None: if max_pending < 0: raise ValueError("max_pending must be non-negative") self._max_pending[queue_name] = max_pending + + def set_queue_codel(self, queue_name: str, target_ms: int, interval_ms: int) -> None: + """Enable opt-in CoDel load shedding on a queue. + + Under sustained overload — when a job's wait past its eligibility stays + above ``target_ms`` for a full ``interval_ms`` — the dispatcher sheds the + stalest jobs to the dead-letter queue (reason prefixed ``codel:``) + instead of running them stale. A transient spike is never shed, and CoDel + entries are excluded from DLQ auto-retry. Takes effect at ``run_worker``. + + Args: + queue_name: Queue name (e.g. ``"default"``). + target_ms: Acceptable steady-state wait (ms) past eligibility. + interval_ms: Window the wait must stay above target before shedding. + """ + if target_ms <= 0 or interval_ms <= 0: + raise ValueError("target_ms and interval_ms must be positive") + config = self._queue_configs.setdefault(queue_name, {}) + config["codel_target_ms"] = target_ms + config["codel_interval_ms"] = interval_ms diff --git a/sdks/python/tests/core/test_codel.py b/sdks/python/tests/core/test_codel.py new file mode 100644 index 00000000..a2000fe9 --- /dev/null +++ b/sdks/python/tests/core/test_codel.py @@ -0,0 +1,105 @@ +"""S27 — opt-in CoDel load shedding. + +Behavioral (timing-based): a slow task with concurrency 1 backs the queue up, so +later jobs' sojourn stays above target for a full interval and CoDel sheds the +stalest ones to the DLQ. The controller's algorithm itself is unit-tested in Rust +(``scheduler::codel``); here we assert the end-to-end shed path and its +invariants. +""" + +from __future__ import annotations + +import threading +import time +from pathlib import Path + +import pytest + +from taskito import Queue + + +def test_set_queue_codel_validates() -> None: + q = Queue(db_path=":memory:", workers=1) + with pytest.raises(ValueError): + q.set_queue_codel("default", target_ms=0, interval_ms=10) + with pytest.raises(ValueError): + q.set_queue_codel("default", target_ms=10, interval_ms=-1) + + +def test_codel_sheds_stale_jobs_under_overload(tmp_path: Path) -> None: + q = Queue(db_path=str(tmp_path / "codel.db"), workers=1, scheduler_batch_size=1) + q.set_queue_codel("default", target_ms=1, interval_ms=30) + + @q.task(name="slow") + def slow() -> None: + time.sleep(0.05) + + total = 20 + for _ in range(total): + q.enqueue("slow") + + worker = threading.Thread(target=q.run_worker, daemon=True) + worker.start() + + # Poll until every job is accounted for (ran or shed) and at least one was + # shed — an aggregate-convergence check, never an instantaneous snapshot. + deadline = time.time() + 25 + codel_dead: list[dict] = [] + stats = q.stats() + try: + while time.time() < deadline: + dead = q.dead_letters(limit=100) + codel_dead = [d for d in dead if str(d.get("error", "")).startswith("codel:")] + stats = q.stats() + if stats["completed"] + stats["dead"] == total and len(codel_dead) >= 1: + break + time.sleep(0.1) + finally: + q.shutdown() + worker.join(timeout=5) + + assert len(codel_dead) >= 1, "sustained overload should shed at least one stale job" + # Every shed job is a CoDel drop (the task never fails on its own). + assert stats["dead"] == len(codel_dead) + # Nothing is lost: every job either ran to completion or was shed. + assert stats["completed"] + stats["dead"] == total + + +def test_codel_drops_are_not_auto_retried(tmp_path: Path) -> None: + # A CoDel-shed entry must not be resurrected by DLQ auto-retry. + q = Queue( + db_path=str(tmp_path / "codel_retry.db"), + workers=1, + scheduler_batch_size=1, + dlq_auto_retry_delay=0, + dlq_auto_retry_max=5, + ) + q.set_queue_codel("default", target_ms=1, interval_ms=30) + + @q.task(name="slow") + def slow() -> None: + time.sleep(0.05) + + for _ in range(20): + q.enqueue("slow") + + worker = threading.Thread(target=q.run_worker, daemon=True) + worker.start() + try: + deadline = time.time() + 25 + while time.time() < deadline: + stats = q.stats() + if stats["pending"] == 0 and stats["running"] == 0 and stats["dead"] > 0: + break + time.sleep(0.1) + # Give auto-retry a few cycles to (not) fire. + time.sleep(1.0) + finally: + q.shutdown() + worker.join(timeout=5) + + dead = q.dead_letters(limit=100) + codel_dead = [d for d in dead if str(d.get("error", "")).startswith("codel:")] + assert codel_dead, "expected at least one CoDel-shed entry" + # Still dead — auto-retry left them alone. + assert q.stats()["dead"] == len(codel_dead) From 2cb8df5a02a801dee5c362b3b49f37ccde862dde Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Sat, 18 Jul 2026 16:48:22 +0530 Subject: [PATCH 3/7] feat(node): add per-queue CoDel config --- crates/taskito-node/src/config.rs | 5 ++ crates/taskito-node/src/convert/mod.rs | 2 +- .../taskito-node/src/convert/task_config.rs | 17 ++++++ crates/taskito-node/src/worker.rs | 3 + sdks/node/src/types.ts | 7 +++ sdks/node/src/worker.ts | 2 + sdks/node/test/core/codel.test.ts | 55 +++++++++++++++++++ 7 files changed, 90 insertions(+), 1 deletion(-) create mode 100644 sdks/node/test/core/codel.test.ts diff --git a/crates/taskito-node/src/config.rs b/crates/taskito-node/src/config.rs index 95d43f62..97ce31c9 100644 --- a/crates/taskito-node/src/config.rs +++ b/crates/taskito-node/src/config.rs @@ -148,6 +148,11 @@ pub struct QueueConfigInput { pub name: String, pub max_concurrent: Option, pub rate_limit: Option, + /// Opt-in CoDel load shedding. Both must be set and positive to enable it: + /// `codelTargetMs` = acceptable steady-state wait; `codelIntervalMs` = window + /// the wait must stay above target before jobs are shed to the DLQ. + pub codel_target_ms: Option, + pub codel_interval_ms: Option, } /// Opt-in mesh overlay for a worker: decentralized peer discovery (SWIM gossip) diff --git a/crates/taskito-node/src/convert/mod.rs b/crates/taskito-node/src/convert/mod.rs index af0d7b33..f2fed243 100644 --- a/crates/taskito-node/src/convert/mod.rs +++ b/crates/taskito-node/src/convert/mod.rs @@ -27,7 +27,7 @@ pub use stats::{ dead_job_to_js, job_error_to_js, metric_to_js, stats_to_js, status_code, worker_to_js, JsDeadJob, JsDeadJobPage, JsJobError, JsMetric, JsStats, JsWorkerRow, }; -pub use task_config::{queue_config, task_config}; +pub use task_config::{queue_codel, queue_config, task_config}; #[cfg(feature = "workflows")] pub use workflow::{ node_to_js, run_to_js, JsFanOutCompletion, JsWorkflowAdvance, JsWorkflowNode, diff --git a/crates/taskito-node/src/convert/task_config.rs b/crates/taskito-node/src/convert/task_config.rs index e54b67fe..d90c1368 100644 --- a/crates/taskito-node/src/convert/task_config.rs +++ b/crates/taskito-node/src/convert/task_config.rs @@ -118,3 +118,20 @@ pub fn queue_config(input: &QueueConfigInput) -> Result { max_concurrent: input.max_concurrent, }) } + +/// Build the optional per-queue CoDel config: both bounds must be present and +/// positive. Kept separate from [`queue_config`] since the core registers CoDel +/// through its own path, not on `QueueConfig`. +pub fn queue_codel( + input: &QueueConfigInput, +) -> Option { + match (input.codel_target_ms, input.codel_interval_ms) { + (Some(target_ms), Some(interval_ms)) if target_ms > 0 && interval_ms > 0 => { + Some(taskito_core::scheduler::codel::CodelConfig { + target_ms, + interval_ms, + }) + } + _ => None, + } +} diff --git a/crates/taskito-node/src/worker.rs b/crates/taskito-node/src/worker.rs index 31685c91..7540a64f 100644 --- a/crates/taskito-node/src/worker.rs +++ b/crates/taskito-node/src/worker.rs @@ -127,6 +127,9 @@ pub fn start_worker( scheduler.register_task(input.name.clone(), crate::convert::task_config(input)?); } for input in options.queue_configs.iter().flatten() { + if let Some(codel) = crate::convert::queue_codel(input) { + scheduler.register_queue_codel(input.name.clone(), codel); + } scheduler.register_queue_config(input.name.clone(), crate::convert::queue_config(input)?); } let scheduler = Arc::new(scheduler); diff --git a/sdks/node/src/types.ts b/sdks/node/src/types.ts index eb19697c..f27e772c 100644 --- a/sdks/node/src/types.ts +++ b/sdks/node/src/types.ts @@ -173,6 +173,13 @@ export interface QueueLimits { * (a non-atomic count-then-insert), so it applies even with no worker running. */ maxPending?: number; + /** + * Opt-in CoDel load shedding. Under sustained overload — a job's wait past its + * eligibility staying above `targetMs` for a full `intervalMs` — the + * dispatcher sheds the stalest jobs to the DLQ (reason prefixed `codel:`) + * rather than running them stale. A transient spike is never shed. + */ + codel?: { targetMs: number; intervalMs: number }; } /** A task handler plus its registration options. */ diff --git a/sdks/node/src/worker.ts b/sdks/node/src/worker.ts index 581af3ed..572a5d29 100644 --- a/sdks/node/src/worker.ts +++ b/sdks/node/src/worker.ts @@ -356,5 +356,7 @@ function buildQueueConfigs(limits: ReadonlyMap): QueueConfi name, maxConcurrent: limit.maxConcurrent, rateLimit: limit.rateLimit, + codelTargetMs: limit.codel?.targetMs, + codelIntervalMs: limit.codel?.intervalMs, })); } diff --git a/sdks/node/test/core/codel.test.ts b/sdks/node/test/core/codel.test.ts new file mode 100644 index 00000000..92f25983 --- /dev/null +++ b/sdks/node/test/core/codel.test.ts @@ -0,0 +1,55 @@ +// S27 — opt-in CoDel load shedding. Behavioral (timing-based): a slow task at +// concurrency 1 backs the queue up, so later jobs stay above target for a full +// interval and CoDel sheds the stalest ones to the DLQ. The controller algorithm +// itself is unit-tested in Rust (`scheduler::codel`). + +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-codel-")), "queue.db"); + return new Queue({ dbPath }); +} + +const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms)); + +it("sheds stale jobs to the DLQ under sustained overload", async () => { + const queue = newQueue(); + queue.configureQueue("default", { codel: { targetMs: 1, intervalMs: 30 } }); + queue.task("slow", async () => { + await sleep(50); + }); + + const total = 20; + for (let i = 0; i < total; i++) queue.enqueue("slow"); + + worker = queue.runWorker({ concurrency: 1, batchSize: 1 }); + + // Poll until every job is accounted for (ran or shed) AND at least one was + // shed — an aggregate-convergence check, never an instantaneous snapshot. + const deadline = Date.now() + 25_000; + let codelDead = 0; + let stats = await queue.stats(); + while (Date.now() < deadline) { + const dead = await queue.deadLetters(100); + codelDead = dead.filter((d) => (d.error ?? "").startsWith("codel:")).length; + stats = await queue.stats(); + if (stats.completed + stats.dead === total && codelDead >= 1) break; + await sleep(100); + } + + expect(codelDead).toBeGreaterThanOrEqual(1); + // Every dead job is a CoDel drop (the task never throws), and nothing is lost. + expect(stats.dead).toBe(codelDead); + expect(stats.completed + stats.dead).toBe(total); +}); From 1abe90defb89c241ebfc3de89f1412db70f7d68d Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Sat, 18 Jul 2026 16:48:37 +0530 Subject: [PATCH 4/7] feat(java): add per-queue CoDel config --- crates/taskito-java/src/convert.rs | 13 +++ crates/taskito-java/src/worker.rs | 29 ++++++- .../org/byteveda/taskito/DefaultTaskito.java | 27 +++++- .../java/org/byteveda/taskito/Taskito.java | 10 +++ .../org/byteveda/taskito/worker/Worker.java | 14 ++++ .../org/byteveda/taskito/core/CodelTest.java | 82 +++++++++++++++++++ 6 files changed, 173 insertions(+), 2 deletions(-) create mode 100644 sdks/java/src/test/java/org/byteveda/taskito/core/CodelTest.java diff --git a/crates/taskito-java/src/convert.rs b/crates/taskito-java/src/convert.rs index d166f150..6b586c91 100644 --- a/crates/taskito-java/src/convert.rs +++ b/crates/taskito-java/src/convert.rs @@ -300,6 +300,19 @@ 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. + 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. +#[derive(Deserialize, Default)] +#[serde(rename_all = "camelCase", default)] +pub struct QueueConfigSpec { + pub name: String, + pub codel_target_ms: Option, + pub codel_interval_ms: 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 7f0f5f69..256662b3 100644 --- a/crates/taskito-java/src/worker.rs +++ b/crates/taskito-java/src/worker.rs @@ -15,6 +15,7 @@ use jni::JNIEnv; use taskito_core::resilience::circuit_breaker::CircuitBreakerConfig; use taskito_core::resilience::rate_limiter::RateLimitConfig; use taskito_core::resilience::retry::RetryPolicy; +use taskito_core::scheduler::codel::CodelConfig; use taskito_core::scheduler::{ResultOutcome, TaskConfig}; use taskito_core::worker::WorkerDispatcher; use taskito_core::{Scheduler, SchedulerConfig, Storage, StorageBackend}; @@ -24,7 +25,9 @@ use taskito_core::job::now_millis; use taskito_core::storage::models::NewSubscriptionRow; use crate::backend::QueueHandle; -use crate::convert::{parse_json, SubscriptionSpec, TaskRetryConfig, WorkerOptions}; +use crate::convert::{ + parse_json, QueueConfigSpec, SubscriptionSpec, TaskRetryConfig, WorkerOptions, +}; use crate::dispatcher::{JavaDispatcher, Registry, TaskOutcome}; use crate::ffi::{guard, read_bytes, read_string}; use crate::handle::{self, drop_handle, into_handle}; @@ -120,6 +123,9 @@ 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); + } let scheduler = Arc::new(scheduler); let shutdown = scheduler.shutdown_handle(); let heartbeat_stop = Arc::new(Notify::new()); @@ -315,6 +321,27 @@ fn build_task_policies( Ok(built) } +/// 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() +} + /// Parse an optional rate spec, naming the offending task and option so a typo /// is actionable. Several options share this `"100/m"` grammar. fn parse_rate_spec( 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 6550d137..3942b9b3 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java @@ -94,6 +94,8 @@ final class DefaultTaskito implements Taskito { private final List subscriptions = new CopyOnWriteArrayList<>(); // Opt-in per-queue admission caps (queue -> max pending). Absent = uncapped. private final Map maxPending = new ConcurrentHashMap<>(); + // Opt-in per-queue CoDel config (queue -> [targetMs, intervalMs]). + private final Map codelConfigs = new ConcurrentHashMap<>(); DefaultTaskito(QueueBackend backend, Serializer serializer, Map codecs) { this.backend = backend; @@ -194,6 +196,28 @@ public Taskito gate(String taskName, EnqueueGate gate) { return this; } + @Override + public Taskito codel(String queue, long targetMs, long intervalMs) { + if (targetMs <= 0 || intervalMs <= 0) { + throw new IllegalArgumentException("targetMs and intervalMs must be positive"); + } + codelConfigs.put(queue, new long[] {targetMs, intervalMs}); + return this; + } + + /** Serialize the per-queue CoDel configs into the worker-options wire shape. */ + private List> encodeQueueConfigs() { + List> specs = new ArrayList<>(codelConfigs.size()); + codelConfigs.forEach((queue, codel) -> { + Map spec = new LinkedHashMap<>(); + spec.put("name", queue); + spec.put("codelTargetMs", codel[0]); + spec.put("codelIntervalMs", codel[1]); + specs.add(spec); + }); + return specs; + } + @Override public Taskito intercept(Interceptor interceptor) { interceptors.add(interceptor); @@ -1085,7 +1109,8 @@ public Worker.Builder worker() { // The live subscription list is shared so subscribe() calls made before // start() are registered under the started worker's id. return Worker.builder(backend, serializer, middleware, resources.forWorker(), codecs) - .subscriptions(subscriptions); + .subscriptions(subscriptions) + .queueConfigs(encodeQueueConfigs()); } @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 e84c53cc..0babd332 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/Taskito.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/Taskito.java @@ -130,6 +130,16 @@ static Builder builder() { */ Taskito maxPending(String queue, int cap); + /** + * Enable opt-in CoDel load shedding on {@code queue}. Under sustained + * overload — a job's wait past its eligibility staying above {@code targetMs} + * for a full {@code intervalMs} — a running worker sheds the stalest jobs to + * the dead-letter queue (reason prefixed {@code codel:}) instead of running + * them stale. A transient spike is never shed. Takes effect for workers + * started after this call. Returns {@code this}. + */ + Taskito codel(String queue, long targetMs, long intervalMs); + // ── 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 d5731116..caef5c24 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 @@ -190,6 +190,7 @@ public static final class Builder { private final Map>> listeners = new EnumMap<>(EventName.class); private List subscriptions = List.of(); + private List> queueConfigs = List.of(); private List queues; private int concurrency; private Integer channelCapacity; @@ -274,6 +275,16 @@ public Builder subscriptions(List subscriptions) { return this; } + /** + * Per-queue scheduler config in wire shape (currently CoDel). Populated + * from the owning {@code Taskito} via {@code Taskito.worker()}; a manually + * built worker leaves it empty. + */ + public Builder queueConfigs(List> queueConfigs) { + this.queueConfigs = queueConfigs; + return this; + } + public Builder queues(String... queues) { this.queues = Arrays.asList(queues); return this; @@ -457,6 +468,9 @@ private String encodeOptions() { if (!subscriptions.isEmpty()) { options.put("subscriptions", encodeSubscriptions()); } + if (!queueConfigs.isEmpty()) { + options.put("queueConfigs", queueConfigs); + } // Presence, not emptiness: an empty Retention encodes as `{}` and // disables retention, which the core distinguishes from an omitted // option (recommended defaults). Do NOT change to a non-empty check. diff --git a/sdks/java/src/test/java/org/byteveda/taskito/core/CodelTest.java b/sdks/java/src/test/java/org/byteveda/taskito/core/CodelTest.java new file mode 100644 index 00000000..9aed74d2 --- /dev/null +++ b/sdks/java/src/test/java/org/byteveda/taskito/core/CodelTest.java @@ -0,0 +1,82 @@ +package org.byteveda.taskito.core; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.nio.file.Path; +import java.time.Duration; +import java.util.List; +import org.byteveda.taskito.Taskito; +import org.byteveda.taskito.model.DeadJob; +import org.byteveda.taskito.model.QueueStats; +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; + +/** + * S27 — opt-in CoDel load shedding. The controller algorithm is unit-tested in + * Rust (`scheduler::codel`); this drives the end-to-end shed path with a slow + * task at concurrency 1 so the queue backs up and stale jobs are shed to the DLQ. + */ +class CodelTest { + + @Test + void codelValidatesBounds(@TempDir Path dir) { + try (Taskito queue = + Taskito.builder().url(dir.resolve("v.db").toString()).open()) { + assertThrows(IllegalArgumentException.class, () -> queue.codel("default", 0, 10)); + assertThrows(IllegalArgumentException.class, () -> queue.codel("default", 10, -1)); + } + } + + @Test + @Timeout(60) + void shedsStaleJobsUnderOverload(@TempDir Path dir) throws Exception { + try (Taskito queue = + Taskito.builder().url(dir.resolve("codel.db").toString()).open()) { + queue.codel("default", 1, 30); + Task slow = Task.of("slow", String.class); + + int total = 20; + for (int i = 0; i < total; i++) { + queue.enqueue(slow, "x"); + } + + try (Worker worker = queue.worker() + .concurrency(1) + .batchSize(1) + .handle(slow, payload -> { + try { + Thread.sleep(50); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + return null; + }) + .start()) { + // Poll until every job is accounted for and at least one was shed. + long deadline = System.nanoTime() + Duration.ofSeconds(45).toNanos(); + long codelDead = 0; + QueueStats stats = queue.stats(); + while (System.nanoTime() < deadline) { + List dead = queue.listDead(100, 0); + codelDead = dead.stream() + .filter(d -> d.error != null && d.error.startsWith("codel:")) + .count(); + stats = queue.stats(); + if (stats.completed + stats.dead == total && codelDead >= 1) { + break; + } + Thread.sleep(100); + } + + assertTrue(codelDead >= 1, "sustained overload should shed at least one stale job"); + assertEquals(codelDead, stats.dead, "every dead job is a CoDel drop"); + assertEquals(total, stats.completed + stats.dead, "no job is lost"); + } + } + } +} From dd1dd33884d780f6588b5c56a74842ab94586f88 Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Sat, 18 Jul 2026 19:31:08 +0530 Subject: [PATCH 5/7] fix(java): bind CoDel config at worker start, not builder creation --- .../org/byteveda/taskito/DefaultTaskito.java | 2 +- .../org/byteveda/taskito/worker/Worker.java | 19 +++++---- .../org/byteveda/taskito/core/CodelTest.java | 42 +++++++++++++++++++ 3 files changed, 55 insertions(+), 8 deletions(-) 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 3942b9b3..a2e01a48 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java @@ -1110,7 +1110,7 @@ public Worker.Builder worker() { // start() are registered under the started worker's id. return Worker.builder(backend, serializer, middleware, resources.forWorker(), codecs) .subscriptions(subscriptions) - .queueConfigs(encodeQueueConfigs()); + .queueConfigs(this::encodeQueueConfigs); } @Override 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 caef5c24..0fbb8929 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 @@ -18,6 +18,7 @@ import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; import java.util.function.Consumer; +import java.util.function.Supplier; import org.byteveda.taskito.autoscale.AutoscaleOptions; import org.byteveda.taskito.autoscale.Autoscaler; import org.byteveda.taskito.errors.SerializationException; @@ -190,7 +191,7 @@ public static final class Builder { private final Map>> listeners = new EnumMap<>(EventName.class); private List subscriptions = List.of(); - private List> queueConfigs = List.of(); + private Supplier>> queueConfigs = List::of; private List queues; private int concurrency; private Integer channelCapacity; @@ -276,11 +277,13 @@ public Builder subscriptions(List subscriptions) { } /** - * Per-queue scheduler config in wire shape (currently CoDel). Populated - * from the owning {@code Taskito} via {@code Taskito.worker()}; a manually - * built worker leaves it empty. + * 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 + * obtained (e.g. {@code Taskito.codel(...)}) is still picked up — matching + * that method's documented timing. A manually built worker leaves it empty. */ - public Builder queueConfigs(List> queueConfigs) { + public Builder queueConfigs(Supplier>> queueConfigs) { this.queueConfigs = queueConfigs; return this; } @@ -468,8 +471,10 @@ private String encodeOptions() { if (!subscriptions.isEmpty()) { options.put("subscriptions", encodeSubscriptions()); } - if (!queueConfigs.isEmpty()) { - options.put("queueConfigs", queueConfigs); + // Resolve at start() so config set after the builder was obtained is seen. + List> resolvedQueueConfigs = queueConfigs.get(); + if (!resolvedQueueConfigs.isEmpty()) { + options.put("queueConfigs", resolvedQueueConfigs); } // Presence, not emptiness: an empty Retention encodes as `{}` and // disables retention, which the core distinguishes from an omitted diff --git a/sdks/java/src/test/java/org/byteveda/taskito/core/CodelTest.java b/sdks/java/src/test/java/org/byteveda/taskito/core/CodelTest.java index 9aed74d2..26b939c9 100644 --- a/sdks/java/src/test/java/org/byteveda/taskito/core/CodelTest.java +++ b/sdks/java/src/test/java/org/byteveda/taskito/core/CodelTest.java @@ -79,4 +79,46 @@ void shedsStaleJobsUnderOverload(@TempDir Path dir) throws Exception { } } } + + @Test + @Timeout(60) + void codelConfiguredAfterBuilderIsStillApplied(@TempDir Path dir) throws Exception { + try (Taskito queue = + Taskito.builder().url(dir.resolve("late.db").toString()).open()) { + Task slow = Task.of("slow", String.class); + int total = 20; + for (int i = 0; i < total; i++) { + queue.enqueue(slow, "x"); + } + + // Obtain the builder BEFORE configuring CoDel, then configure it — the + // late-bound queue-config source must pick this up at start(). + Worker.Builder builder = queue.worker().concurrency(1).batchSize(1).handle(slow, payload -> { + try { + Thread.sleep(50); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + return null; + }); + queue.codel("default", 1, 30); + + try (Worker worker = builder.start()) { + long deadline = System.nanoTime() + Duration.ofSeconds(45).toNanos(); + long codelDead = 0; + QueueStats stats = queue.stats(); + while (System.nanoTime() < deadline) { + codelDead = queue.listDead(100, 0).stream() + .filter(d -> d.error != null && d.error.startsWith("codel:")) + .count(); + stats = queue.stats(); + if (stats.completed + stats.dead == total && codelDead >= 1) { + break; + } + Thread.sleep(100); + } + assertTrue(codelDead >= 1, "CoDel set after the builder was obtained must still apply"); + } + } + } } From f5f3609ef9355cc97d35483749030db32f132f3e Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Sat, 18 Jul 2026 19:31:14 +0530 Subject: [PATCH 6/7] fix(node): validate CoDel bounds in configureQueue --- sdks/node/src/queue.ts | 10 ++++++++++ sdks/node/test/core/codel.test.ts | 14 ++++++++++++++ 2 files changed, 24 insertions(+) diff --git a/sdks/node/src/queue.ts b/sdks/node/src/queue.ts index 8c0963cf..33911668 100644 --- a/sdks/node/src/queue.ts +++ b/sdks/node/src/queue.ts @@ -392,6 +392,16 @@ export class Queue { if (limits.maxPending !== undefined && limits.maxPending < 0) { throw new RangeError("maxPending must be non-negative"); } + // CoDel bounds cross to native as i64 — reject non-positive/non-integer + // (0, negatives, fractions, NaN, Infinity) here rather than silently coercing. + if (limits.codel !== undefined) { + for (const key of ["targetMs", "intervalMs"] as const) { + const value = limits.codel[key]; + if (!Number.isInteger(value) || value <= 0) { + throw new RangeError(`codel.${key} must be a positive integer`); + } + } + } this.queueLimits.set(name, limits); } diff --git a/sdks/node/test/core/codel.test.ts b/sdks/node/test/core/codel.test.ts index 92f25983..1994ea9f 100644 --- a/sdks/node/test/core/codel.test.ts +++ b/sdks/node/test/core/codel.test.ts @@ -23,6 +23,20 @@ function newQueue(): Queue { const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms)); +it("rejects non-positive-integer CoDel bounds at configureQueue", () => { + const queue = newQueue(); + for (const bad of [0, -1, 1.5, Number.NaN, Number.POSITIVE_INFINITY]) { + expect(() => + queue.configureQueue("default", { codel: { targetMs: bad, intervalMs: 30 } }), + ).toThrow(RangeError); + expect(() => + queue.configureQueue("default", { codel: { targetMs: 30, intervalMs: bad } }), + ).toThrow(RangeError); + } + // A valid pair is accepted. + queue.configureQueue("default", { codel: { targetMs: 1, intervalMs: 30 } }); +}); + it("sheds stale jobs to the DLQ under sustained overload", async () => { const queue = newQueue(); queue.configureQueue("default", { codel: { targetMs: 1, intervalMs: 30 } }); From 219e2651fd1dbe861b4e115b3be19e7489fabc23 Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Sat, 18 Jul 2026 19:38:22 +0530 Subject: [PATCH 7/7] fix(node): snapshot validated queue limits in configureQueue --- sdks/node/src/queue.ts | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/sdks/node/src/queue.ts b/sdks/node/src/queue.ts index 33911668..b0165618 100644 --- a/sdks/node/src/queue.ts +++ b/sdks/node/src/queue.ts @@ -402,7 +402,12 @@ export class Queue { } } } - this.queueLimits.set(name, limits); + // Snapshot after validating: `limits` is caller-owned, so store a copy (with + // a copied `codel`) so a later mutation can't slip past the checks above. + this.queueLimits.set(name, { + ...limits, + codel: limits.codel === undefined ? undefined : { ...limits.codel }, + }); } /** Register middleware (execution + outcome hooks). Runs in registration order. */