fix: Fix a bug where jobs would block event loop

crates/crontab_runner/src/lib.rs

@@ -48,8 +48,6 @@ pub async fn cron_main<'e>(
         let current_ts = round_date_minute(Local::now(), false);
         let ts_delta = current_ts - ts;
 
-        dbg!(ts_delta, current_ts, ts);
-
         match ts_delta.num_minutes().cmp(&0) {
             Ordering::Greater => {
                 warn!(

crates/crontab_types/src/lib.rs

@@ -1,3 +1,5 @@
+#![allow(clippy::non_canonical_partial_ord_impl)]
+
 use chrono::prelude::*;
 use getset::Getters;
 

examples/crontab.rs

@@ -43,12 +43,18 @@ async fn main() {
         .unwrap();
 
     WorkerOptions::default()
-        .concurrency(2)
+        .concurrency(10)
         .schema("example_simple_worker")
         .define_job("say_hello", say_hello)
+        .define_job("say_hello_2", say_hello)
         .pg_pool(pg_pool)
         // Run say_hello every two minutes with a backfill of 10 minutes
-        .with_crontab(r#"*/2 * * * * say_hello ?fill=10m {message:"Crontab"}"#)
+        .with_crontab(
+            r#"
+            */2 * * * * say_hello ?fill=10m {message:"Crontab"}
+            */1 * * * * say_hello_2 ?fill=10m {message:"Crontab"}
+        "#,
+        )
         .unwrap()
         .init()
         .await

justfile

@@ -0,0 +1,5 @@
+check-fmt:
+  cargo fmt --all -- --check
+
+check-clippy:
+  cargo clippy --all -- -D warnings

src/runner.rs

@@ -1,5 +1,5 @@
 use std::fmt::Debug;
-use std::future::{ready, Future};
+use std::future::Future;
 use std::pin::Pin;
 use std::time::Duration;
 use std::{collections::HashMap, time::Instant};
@@ -86,12 +86,15 @@ impl Worker {
         .await?;
 
         job_signal
-            .then(|source| process_one_job(self, source))
-            .try_for_each_concurrent(self.concurrency, |j| {
-                if let Some(j) = &j {
-                    debug!("Job id={} processed", j.id());
+            .map(Ok::<_, ProcessJobError>)
+            .try_for_each_concurrent(self.concurrency, |source| async move {
+                let res = process_one_job(self, source).await?;
+
+                if let Some(job) = res {
+                    debug!(job_id = job.id(), "Job processed");
                 }
-                ready(Ok(()))
+
+                Ok(())
             })
             .await?;
 
@@ -145,7 +148,7 @@ async fn process_one_job(
         Some(job) => {
             let job_result = run_job(&job, worker, &source).await;
             release_job(job_result, &job, worker).await.map_err(|e| {
-                error!("{:?}", e);
+                error!("Release job error : {:?}", e);
                 e
             })?;
             Ok(Some(job))

src/streams.rs

@@ -26,8 +26,8 @@ pub async fn job_signal_stream(
 
     let stream = stream::unfold((interval, pg_listener, shutdown_signal), |mut f| async {
         tokio::select! {
-            _ = (&mut f.0).tick() => Some((StreamSource::Polling, f)),
-            _ = (&mut f.1).recv() => Some((StreamSource::PgListener, f)),
+            _ = (f.0).tick() => Some((StreamSource::Polling, f)),
+            _ = (f.1).recv() => Some((StreamSource::PgListener, f)),
             _ = &mut f.2 => None,
         }
     });