Only enqueue a sched job if the worker removed it

src/periodic.rs

@@ -104,8 +104,8 @@ impl Builder {
             ..Default::default()
         };
 
-        pj.retry = self.retry.clone();
-        pj.queue = self.queue.clone();
+        pj.retry.clone_from(&self.retry);
+        pj.queue.clone_from(&self.queue);
         pj.args = self.args.clone().map(|a| a.to_string());
 
         pj.hydrate_attributes()?;

src/redis.rs

@@ -237,7 +237,7 @@ impl RedisConnection {
             .await
     }
 
-    pub async fn zrem<V>(&mut self, key: String, value: V) -> Result<bool, RedisError>
+    pub async fn zrem<V>(&mut self, key: String, value: V) -> Result<usize, RedisError>
     where
         V: ToRedisArgs + Send + Sync,
     {

src/scheduled.rs

@@ -21,13 +21,11 @@ impl Scheduled {
             let mut redis = self.redis.get().await?;
 
             let jobs: Vec<String> = redis
-                .zrangebyscore_limit(sorted_set.clone(), "-inf", now.timestamp(), 0, 100)
+                .zrangebyscore_limit(sorted_set.clone(), "-inf", now.timestamp(), 0, 10)
                 .await?;
 
-            n += jobs.len();
-
             for job in jobs {
-                if redis.zrem(sorted_set.clone(), job.clone()).await? {
+                if redis.zrem(sorted_set.clone(), job.clone()).await? > 0 {
                     let work = UnitOfWork::from_job_string(job)?;
 
                     debug!({
@@ -36,6 +34,8 @@ impl Scheduled {
                     },  "Enqueueing job");
 
                     work.enqueue_direct(&mut redis).await?;
+
+                    n += 1;
                 }
             }
         }

src/stats.rs

@@ -111,7 +111,7 @@ impl StatsPublisher {
             .arg(stats.beat.timestamp())
             .arg("info")
             .arg(serde_json::to_string(&stats.info)?)
-            .query_async(conn.unnamespaced_borrow_mut())
+            .query_async::<_, ()>(conn.unnamespaced_borrow_mut())
             .await?;
 
         conn.expire(self.identity.clone(), 30).await?;