Fix executors heartbeat flaw on startup (#82)

* Fix executors heartbeat flaw by setting them before spawning executors

* Fix worker startup command in README.md

Author: CCXLV

README.md

@@ -98,7 +98,7 @@ It picks the latest released worker based on your python version and installs it
 Running the worker is straightforward:
 
 ```bash
-fluxqueue start --tasks-module-path tasks
+fluxqueue start --tasks-module-path myapp/tasks
 ```
 
 In order the worker to disover your tasks you need to pass `--tasks-module-path` argument with the path to the tasks module. For more information please view the [documentation](https://fluxqueue.ccxlv.dev/tutorial/defininig_and_exposing_tasks).

crates/fluxqueue-worker/src/redis_client.rs

@@ -33,25 +33,30 @@ impl RedisClient {
         Ok(())
     }
 
-    pub async fn set_executor_heartbeat(&self, executor_ids: Arc<Vec<Arc<str>>>) -> Result<()> {
-        let mut conn = self.redis_pool.get().await?;
-
+    pub async fn set_executors_heartbeat(&self, executor_ids: Arc<Vec<Arc<str>>>) -> Result<()> {
         for id in executor_ids.iter() {
-            let heartbeat_key = keys::get_heartbeat_key(id);
-
-            let _: () = redis::cmd("SET")
-                .arg(heartbeat_key)
-                .arg(1)
-                .arg("EX")
-                .arg(15)
-                .query_async(&mut conn)
-                .await
-                .map_err(|e| anyhow::anyhow!("Failed to set executor heartbeat: {}", e))?;
+            self.set_executor_heartbeat(id).await?;
         }
 
         Ok(())
     }
 
+    pub async fn set_executor_heartbeat(&self, executor_id: &str) -> Result<()> {
+        let mut conn = self.redis_pool.get().await?;
+        let heartbeat_key = keys::get_heartbeat_key(executor_id);
+
+        let _: () = redis::cmd("SET")
+            .arg(heartbeat_key)
+            .arg(1)
+            .arg("EX")
+            .arg(15)
+            .query_async(&mut conn)
+            .await
+            .map_err(|e| anyhow::anyhow!("Failed to set executor heartbeat: {}", e))?;
+
+        Ok(())
+    }
+
     pub async fn cleanup_executors_registry(
         &self,
         queue_name: &str,

crates/fluxqueue-worker/src/worker.rs

@@ -63,6 +63,8 @@ pub async fn run_worker(
             .register_executor(&queue_name, &executor_id)
             .await?;
 
+        redis_client.set_executor_heartbeat(&executor_id).await?;
+
         executors.spawn(executor_loop(
             shutdown,
             queue_name,
@@ -249,7 +251,7 @@ async fn janitor_loop(
             hearbeat = async {
                 let executor_ids = Arc::clone(&executor_ids);
                 redis_client
-                    .set_executor_heartbeat(executor_ids)
+                    .set_executors_heartbeat(executor_ids)
                     .await
             } => {
                 match hearbeat {