Remove waits table and add logos

README.md

@@ -1,4 +1,9 @@
-# Taskturbine
+<h1 align="center">
+    <img src="https://raw.githubusercontent.com/markstory/taskturbine/d25fd8b00646a6107ed9cfcd7017acce58962949/assets/taskturbine-logo-color.svg" />
+    <br />
+    <br />
+    Taskturbine
+</h1>
 
 Taskturbine is a cross-language durable task framework for Rust & Python, using postgres for task storage.
 

taskturbine-app/src/app.rs

@@ -732,6 +732,10 @@ async fn claim_tasks(worker: Arc<Worker>, work_send: Sender<ClaimedTask>) {
                     }
                 }
             }
+            _ = work_send.closed() => {
+                log::debug!("No more receivers for work_send, shutting down claim_tasks");
+                break;
+            }
             _ = guard.wait() => {
                 log::debug!("Shutting down claim_tasks");
                 break;
@@ -748,6 +752,10 @@ async fn process_task(worker: Arc<Worker>, work_channel: Receiver<ClaimedTask>)
             Ok(task) = work_channel.recv() => {
                 worker.execute_task(task).await;
             }
+            Err(_) = work_channel.recv() => {
+                log::debug!("work_channel closed, shutting down process_task");
+                break;
+            }
             _ = guard.wait() => {
                 log::debug!("Shutting down process_task");
                 work_channel.close();
@@ -1197,14 +1205,14 @@ mod tests {
             send.send(item).await.expect("Failed to send claimed task");
         }
 
-        // Kill the worker process after 1 seconds
         let mut join = JoinSet::new();
-        join.spawn(
-            elegant_departure::tokio::depart()
-                .on_completion(tokio::time::sleep(Duration::from_secs(1))),
-        );
-
         join.spawn(process_task(worker, recv));
+
+        // Kill the worker process after 1 seconds
+        join.spawn(async move {
+            tokio::time::sleep(Duration::from_secs(1)).await;
+            send.close();
+        });
         join.join_all().await;
 
         let task = storage
@@ -1231,16 +1239,16 @@ mod tests {
 
         // Kill the worker process after 1 seconds
         let mut join = JoinSet::new();
-        join.spawn(
-            elegant_departure::tokio::depart()
-                .on_completion(tokio::time::sleep(Duration::from_secs(1))),
-        );
-
         join.spawn(claim_tasks(worker, send));
+        join.spawn(async move {
+            tokio::time::sleep(Duration::from_secs(1)).await;
+            let claimed = recv.recv().await.expect("recv should be ok");
+            assert_eq!(claimed.task_id, spawn_res.task_id);
+
+            recv.close();
+        });
         join.join_all().await;
 
-        let claimed = recv.recv().await.expect("recv should be ok");
-        assert_eq!(claimed.task_id, spawn_res.task_id);
         let task = storage
             .get_task(spawn_res.task_id)
             .await
@@ -1264,15 +1272,11 @@ mod tests {
 
         // Kill the worker process after 1 seconds
         let mut join = JoinSet::new();
-        join.spawn(
-            elegant_departure::tokio::depart()
-                .on_completion(tokio::time::sleep(Duration::from_secs(1))),
-        );
 
         // close the channel and then start the claim_tasks task.
         send.close();
-        join.spawn(claim_tasks(worker, send));
 
+        join.spawn(claim_tasks(worker, send));
         join.join_all().await;
 
         let recv_res = recv.try_recv();

taskturbine-core/migrations/0003_run_wait_event.sql

@@ -0,0 +1,2 @@
+ALTER TABLE taskturbine.runs ADD COLUMN wait_event_name TEXT;
+CREATE INDEX runs_wait_event ON taskturbine.runs (wait_event_name);