feat: add opt-in CoDel load shedding

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"
+            );
+        }
+    }
+}

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!(

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<String, TaskConfig>,
     queue_configs: HashMap<String, QueueConfig>,
+    /// 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<String, codel::CodelConfig>,
+    /// 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<HashMap<String, codel::CodelState>>,
     queues: Vec<String>,
     config: SchedulerConfig,
     shutdown: Arc<Notify>,
@@ -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

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<Job>, now: i64) -> Result<Vec<Job>> {
+        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

crates/taskito-java/src/convert.rs

@@ -300,6 +300,19 @@ pub struct WorkerOptions {
     pub subscriptions: Option<Vec<SubscriptionSpec>>,
     /// Per-table retention windows for auto-cleanup, in seconds.
     pub retention: Option<RetentionSpec>,
+    /// Per-queue config registered with the scheduler at start. Currently
+    /// carries opt-in CoDel load shedding.
+    pub queue_configs: Option<Vec<QueueConfigSpec>>,
+}
+
+/// 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<i64>,
+    pub codel_interval_ms: Option<i64>,
 }
 
 /// Per-table retention windows in seconds. An unset field keeps that table