feat: activate native async dispatch

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

@@ -324,6 +324,19 @@ impl Scheduler {
         }
     }
 
+    /// Forget a job that was tracked but never reached the worker channel
+    /// (dispatch rollback). Unlike [`Self::release_in_flight`] this never wakes
+    /// the poller — the slot was returned by the same thread that took it, and
+    /// the send just failed, so an immediate retry would only spin.
+    fn untrack_in_flight(&self, job_id: &str) {
+        if self.config.max_in_flight.is_some() {
+            self.in_flight
+                .lock()
+                .unwrap_or_else(|p| p.into_inner())
+                .remove(job_id);
+        }
+    }
+
     /// How many of this task's jobs this worker currently has in flight.
     fn task_in_flight(&self, task_name: &str) -> usize {
         self.in_flight
@@ -385,6 +398,21 @@ impl Scheduler {
         self.shutdown.clone()
     }
 
+    /// `true` once every job this scheduler dispatched has had its result
+    /// handled — a "drain is done" signal for a binding's shutdown loop that
+    /// does not depend on the result channel disconnecting (which can hinge on
+    /// foreign-object lifetimes, e.g. a Python-held sender). `false` when
+    /// dispatch is unbounded: nothing is tracked, so emptiness proves nothing.
+    pub fn in_flight_settled(&self) -> bool {
+        self.config.max_in_flight.is_some()
+            && self
+                .in_flight
+                .lock()
+                .unwrap_or_else(|p| p.into_inner())
+                .len()
+                == 0
+    }
+
     pub fn register_queue_config(&mut self, queue_name: String, config: QueueConfig) {
         self.queue_configs.insert(queue_name, config);
     }
@@ -1181,6 +1209,121 @@ mod tests {
         );
     }
 
+    #[test]
+    fn test_full_channel_rollback_does_not_leak_in_flight_slot() {
+        // A job that fails the channel hand-off must not keep its in-flight
+        // slot: a leaked entry shrinks the dispatch cap for the life of the
+        // worker and keeps the drain-exit signal from ever settling.
+        let storage =
+            StorageBackend::Sqlite(crate::storage::sqlite::SqliteStorage::in_memory().unwrap());
+        let config = SchedulerConfig {
+            max_in_flight: Some(8),
+            ..SchedulerConfig::default()
+        };
+        let scheduler = Scheduler::new(storage, vec!["default".to_string()], config, None);
+        scheduler.storage.enqueue(make_job("full_chan")).unwrap();
+
+        // Capacity-1 channel pre-filled with a sentinel so the hand-off fails.
+        let (tx, _rx) = make_channel(1);
+        let sentinel = scheduler.storage.enqueue(make_job("sentinel")).unwrap();
+        tx.try_send(sentinel).expect("pre-fill should succeed");
+
+        assert!(!scheduler.tick_dispatch(&tx), "hand-off must not succeed");
+        assert_eq!(
+            scheduler.in_flight_remaining(),
+            Some(8),
+            "failed hand-off must not occupy an in-flight slot"
+        );
+    }
+
+    #[test]
+    fn test_closed_channel_rollback_untracks_in_flight() {
+        // A job tracked before a failed hand-off must be untracked by the
+        // rollback, or the slot leaks and `in_flight_settled` can never turn
+        // true again — every later shutdown would wait out the drain timeout.
+        let storage =
+            StorageBackend::Sqlite(crate::storage::sqlite::SqliteStorage::in_memory().unwrap());
+        let config = SchedulerConfig {
+            max_in_flight: Some(8),
+            ..SchedulerConfig::default()
+        };
+        let scheduler = Scheduler::new(storage, vec!["default".to_string()], config, None);
+        scheduler.storage.enqueue(make_job("closed_chan")).unwrap();
+
+        let (tx, rx) = make_channel(16);
+        drop(rx);
+        assert!(
+            !scheduler.tick_dispatch(&tx),
+            "closed channel must not count as a dispatch"
+        );
+
+        assert_eq!(
+            scheduler.in_flight_remaining(),
+            Some(8),
+            "rolled-back job must not occupy an in-flight slot"
+        );
+        let pending = scheduler
+            .storage
+            .list_jobs(Some(JobStatus::Pending as i32), None, None, 10, 0, None)
+            .unwrap();
+        assert_eq!(
+            pending.len(),
+            1,
+            "job returned to Pending for another worker"
+        );
+    }
+
+    #[test]
+    fn test_in_flight_settled_tracks_dispatch_lifecycle() {
+        // Bounded dispatch: settled when idle, unsettled while a job is out,
+        // settled again once its result is handled. Unbounded dispatch tracks
+        // nothing, so it must never report settled.
+        let storage =
+            StorageBackend::Sqlite(crate::storage::sqlite::SqliteStorage::in_memory().unwrap());
+        let config = SchedulerConfig {
+            max_in_flight: Some(4),
+            ..SchedulerConfig::default()
+        };
+        let scheduler = Scheduler::new(storage, vec!["default".to_string()], config, None);
+        assert!(
+            scheduler.in_flight_settled(),
+            "idle bounded scheduler is settled"
+        );
+
+        scheduler.storage.enqueue(make_job("settle")).unwrap();
+        let (tx, mut rx) = make_channel(16);
+        while scheduler.tick_dispatch(&tx) {}
+        let job = rx.try_recv().expect("job dispatched");
+        assert!(
+            !scheduler.in_flight_settled(),
+            "a dispatched job keeps the scheduler unsettled"
+        );
+
+        scheduler
+            .handle_result(JobResult::Success {
+                job_id: job.id.clone(),
+                result: None,
+                task_name: "settle".to_string(),
+                wall_time_ns: 1,
+            })
+            .unwrap();
+        assert!(
+            scheduler.in_flight_settled(),
+            "handled result settles the scheduler"
+        );
+
+        let unbounded = Scheduler::new(
+            StorageBackend::Sqlite(crate::storage::sqlite::SqliteStorage::in_memory().unwrap()),
+            vec!["default".to_string()],
+            SchedulerConfig::default(),
+            None,
+        );
+        assert!(
+            !unbounded.in_flight_settled(),
+            "unbounded dispatch can never prove it settled"
+        );
+    }
+
     /// Scheduler with an in-flight cap and a per-task bulkhead on `task_name`.
     fn bulkhead_scheduler(max_in_flight: usize, task_name: &str, per_task: usize) -> Scheduler {
         let storage =
@@ -1442,17 +1585,16 @@ mod tests {
     }
 
     #[test]
-    fn test_try_dispatch_reschedules_on_full_channel() {
-        // Same regression when the channel is *full* rather than closed —
-        // the worker pool is behind but still alive. The job must come back
-        // to Pending so the next tick has a chance to dispatch it once the
-        // pool drains.
+    fn test_try_dispatch_skips_when_channel_full() {
+        // When the channel is *full* rather than closed, the poller must not
+        // dequeue at all: a claim taken now could only roll back on `try_send`,
+        // churning storage for nothing. The job stays Pending, untouched, and
+        // dispatch resumes once the pool drains.
         let scheduler = test_scheduler();
         let job = scheduler.storage.enqueue(make_job("ch_full_task")).unwrap();
 
-        // Capacity-1 channel pre-filled with a sentinel job. The poller's
-        // `try_send` will see `TrySendError::Full`.
-        let (tx, _rx) = make_channel(1);
+        // Capacity-1 channel pre-filled with a sentinel job.
+        let (tx, mut rx) = make_channel(1);
         let sentinel = scheduler.storage.enqueue(make_job("sentinel")).unwrap();
         tx.try_send(sentinel).expect("pre-fill should succeed");
 
@@ -1461,13 +1603,23 @@ mod tests {
 
         let after = scheduler.storage.get_job(&job.id).unwrap().unwrap();
         assert_eq!(after.status, JobStatus::Pending);
-        assert!(after.scheduled_at > now_millis());
+        assert_eq!(
+            after.scheduled_at, job.scheduled_at,
+            "a skipped job keeps its schedule — it was never touched"
+        );
 
         let claims = scheduler
             .storage
             .list_claims_by_worker("scheduler")
             .unwrap();
         assert!(!claims.contains(&job.id));
+
+        // Draining the channel restores dispatch.
+        let _ = rx.try_recv().expect("sentinel was in the channel");
+        assert!(
+            scheduler.tick_dispatch(&tx),
+            "freed capacity resumes dispatch"
+        );
     }
 
     #[test]

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

@@ -93,6 +93,14 @@ impl Scheduler {
             return Ok(false);
         }
 
+        // A full channel means every dequeue would claim and then immediately
+        // roll back on `try_send` — pure storage churn. Wait for the pool to
+        // drain; `try_send` below still guards the race where it fills between
+        // this check and the hand-off.
+        if job_tx.capacity() == 0 {
+            return Ok(false);
+        }
+
         let job = match self
             .storage
             .dequeue_from(&active_queues, now, self.namespace.as_deref())?
@@ -126,8 +134,13 @@ impl Scheduler {
         // Size the batch to the worker pool's free capacity so we never claim
         // more jobs than the channel can immediately accept. `try_send` still
         // guards each hand-off, but pre-sizing avoids needless claim/rollback
-        // churn. Always claim at least one.
-        let mut budget = self.config.batch_size.min(job_tx.capacity().max(1));
+        // churn — including the full-channel case, where any claim would
+        // immediately roll back.
+        let capacity = job_tx.capacity();
+        if capacity == 0 {
+            return Ok(false);
+        }
+        let mut budget = self.config.batch_size.min(capacity);
 
         // Never claim past the in-flight cap: a drained batch that outran the
         // workers would mark the surplus `Running` and starve peer schedulers.
@@ -289,15 +302,20 @@ impl Scheduler {
         // must reverse the claim — otherwise the job sits in `Running` until
         // the stale-reaper times it out, which surfaces as a *timeout* in
         // metrics and middleware (wrong outcome for a job that never ran).
+        //
+        // Track BEFORE the send: once the job is in the channel its result can
+        // arrive on the drain thread at any moment, and a release that races
+        // ahead of the insert would leak the slot forever. Tracking first means
+        // the entry always exists by the time a result can be handled; the
+        // failure arms untrack what never left.
         let job_id = job.id.clone();
         let task_name = job.task_name.clone();
+        self.track_in_flight(&job_id, &task_name);
         match job_tx.try_send(job) {
-            Ok(()) => {
-                self.track_in_flight(&job_id, &task_name);
-                Ok(true)
-            }
+            Ok(()) => Ok(true),
             Err(TrySendError::Full(job)) => {
                 warn!("worker channel full; rescheduling job {job_id} (worker pool is behind)",);
+                self.untrack_in_flight(&job_id);
                 counts.release(&job.task_name, &job.queue);
                 self.rollback_claim_and_reschedule(
                     &job_id,
@@ -309,6 +327,7 @@ impl Scheduler {
                 warn!(
                     "worker channel closed; rescheduling job {job_id} (worker pool shutting down)",
                 );
+                self.untrack_in_flight(&job_id);
                 counts.release(&job.task_name, &job.queue);
                 self.rollback_claim_and_reschedule(
                     &job_id,