From e0c96a67e9a17fc51ab7251b502fc12065f44c50 Mon Sep 17 00:00:00 2001 From: Mark Story Date: Sun, 23 Aug 2026 16:01:16 -0400 Subject: [PATCH 1/9] Add a migration to add wait metadata to runs table. --- taskturbine-core/migrations/0003_run_wait_event.sql | 3 +++ taskturbine-core/src/storage.rs | 4 ++++ 2 files changed, 7 insertions(+) create mode 100644 taskturbine-core/migrations/0003_run_wait_event.sql diff --git a/taskturbine-core/migrations/0003_run_wait_event.sql b/taskturbine-core/migrations/0003_run_wait_event.sql new file mode 100644 index 0000000..67a69ed --- /dev/null +++ b/taskturbine-core/migrations/0003_run_wait_event.sql @@ -0,0 +1,3 @@ +ALTER TABLE taskturbine.runs ADD COLUMN wait_event_name TEXT; +ALTER TABLE taskturbine.runs ADD COLUMN wait_timeout TIMESTAMPTZ; +CREATE INDEX runs_wait_event ON taskturbine.runs (wait_event_name); diff --git a/taskturbine-core/src/storage.rs b/taskturbine-core/src/storage.rs index c89cab6..55c357a 100644 --- a/taskturbine-core/src/storage.rs +++ b/taskturbine-core/src/storage.rs @@ -770,6 +770,7 @@ impl Storage { .await .map_err(StorageError::SqlError)?; + // TODO do this conditionally only if the run has a wait attached. self.clear_waits(run_id, &mut atomic).await?; atomic.commit().await.map_err(StorageError::SqlError)?; @@ -972,6 +973,7 @@ impl Storage { .await .map_err(StorageError::SqlError)?; + // TODO only do this if the run has a wait attached. self.clear_waits(run_id, &mut *conn).await?; Ok(()) @@ -1191,6 +1193,8 @@ impl Storage { timeout: Duration, ) -> Result<(), StorageError> { let timeout = Utc::now() + timeout; + // TODO If `taskturbine.runs` had a nullable wait_id column, we could optimize + // fail_run/complete_run to not delete waits that don't exist. sqlx::query( "INSERT INTO taskturbine.waits (usecase, task_id, run_id, step_name, event_name, timeout_at, created_at) VALUES ($1, $2, $3, $4, $5, $6, NOW()) From 5cc82521a7abee365195025becfc5b59445d7c15 Mon Sep 17 00:00:00 2001 From: Mark Story Date: Wed, 26 Aug 2026 00:00:01 -0400 Subject: [PATCH 2/9] Simpler schema Since runs already have an available_at time, having a wait timeout seems redundant. Instead, we can rely on await_event() setting the sleep time, and upkeep loops/retries continuing the sleep cycle until the task max ages out, or is fulfilled. --- taskturbine-core/migrations/0003_run_wait_event.sql | 1 - 1 file changed, 1 deletion(-) diff --git a/taskturbine-core/migrations/0003_run_wait_event.sql b/taskturbine-core/migrations/0003_run_wait_event.sql index 67a69ed..2a12287 100644 --- a/taskturbine-core/migrations/0003_run_wait_event.sql +++ b/taskturbine-core/migrations/0003_run_wait_event.sql @@ -1,3 +1,2 @@ ALTER TABLE taskturbine.runs ADD COLUMN wait_event_name TEXT; -ALTER TABLE taskturbine.runs ADD COLUMN wait_timeout TIMESTAMPTZ; CREATE INDEX runs_wait_event ON taskturbine.runs (wait_event_name); From 283f0c2a4e83e8b1f01bef028c675f9e040ba7c0 Mon Sep 17 00:00:00 2001 From: Mark Story Date: Wed, 26 Aug 2026 00:04:54 -0400 Subject: [PATCH 3/9] Remove usage of taskturbine.waits Waits are simple enough that there isn't a real need for a full separate table that requires more upkeep operations. Instead, runs can wait for events with an additional column. An indexed column lets us search runs with reasonable efficiency. --- taskturbine-core/src/storage.rs | 154 ++++++++++---------------------- 1 file changed, 46 insertions(+), 108 deletions(-) diff --git a/taskturbine-core/src/storage.rs b/taskturbine-core/src/storage.rs index 55c357a..38bf383 100644 --- a/taskturbine-core/src/storage.rs +++ b/taskturbine-core/src/storage.rs @@ -320,12 +320,12 @@ impl Storage { Ok(res) } - /// Testing helper: get waits for a run + /// Testing helper: get runs waiting on an event #[cfg(feature = "test")] - pub async fn get_wait_by_run_id(&self, run_id: RunId) -> Result, StorageError> { - let res = sqlx::query("SELECT * FROM taskturbine.waits WHERE run_id = $1") - .bind(run_id) - .fetch_optional(&self.pool) + pub async fn get_runs_waiting(&self, event_name: &str) -> Result, StorageError> { + let res = sqlx::query("SELECT * FROM taskturbine.runs WHERE wait_event_name = $1") + .bind(event_name) + .fetch_all(&self.pool) .await .map_err(StorageError::SqlError)?; @@ -745,7 +745,7 @@ impl Storage { } let res = sqlx::query( "UPDATE taskturbine.runs as run - SET state = $1, completed_at = NOW(), result = $2 + SET state = $1, completed_at = NOW(), wait_event_name = NULL, result = $2 WHERE run_id = $3", ) .bind(TaskState::Completed) @@ -770,9 +770,6 @@ impl Storage { .await .map_err(StorageError::SqlError)?; - // TODO do this conditionally only if the run has a wait attached. - self.clear_waits(run_id, &mut atomic).await?; - atomic.commit().await.map_err(StorageError::SqlError)?; Ok(()) @@ -828,22 +825,6 @@ impl Storage { Ok(task) } - /// Clear waits on runs that we are no longer interested in - /// as the run is complete or cancelled. - async fn clear_waits( - &self, - run_id: RunId, - conn: &mut PgConnection, - ) -> Result<(), StorageError> { - sqlx::query("DELETE FROM taskturbine.waits WHERE run_id = $1") - .bind(run_id) - .execute(&mut *conn) - .await - .map_err(StorageError::SqlError)?; - - Ok(()) - } - /// Mark a run as failed with the provided reason. /// If an retry_at is not provided, the next retry time will be calculated /// based on the task's retry_ attributes. @@ -891,8 +872,9 @@ impl Storage { let res = sqlx::query( "UPDATE taskturbine.runs SET state = $1, - completed_at = NOW(), - failure_reason = $2 + completed_at = NOW(), + failure_reason = $2, + wait_event_name = NULL WHERE run_id = $3", ) .bind(TaskState::Failed) @@ -973,9 +955,6 @@ impl Storage { .await .map_err(StorageError::SqlError)?; - // TODO only do this if the run has a wait attached. - self.clear_waits(run_id, &mut *conn).await?; - Ok(()) } @@ -1006,6 +985,7 @@ impl Storage { &run.get::("task_id"), &run_id, wait_for, + None, ) .await?; @@ -1138,7 +1118,7 @@ impl Storage { return Err(StorageError::NotRunning(run_id.0)); } - // Check for an event that was received while we were sleeping/running. + // Check for an event that was received while we were sleeping/starting up. if let Some(payload) = event { // There was an event, store a checkpoint and return self.store_checkpoint(&mut atomic, &task_id, &run_id, step_name, &payload) @@ -1157,19 +1137,9 @@ impl Storage { let timeout = timeout.unwrap_or_else(|| { Duration::from_secs(self.config.await_event_default_timeout_secs as u64) }); - // Record the event wait - self.store_wait( - &mut atomic, - &task_id, - &run_id, - step_name, - event_name, - timeout, - ) - .await?; // Suspend the current run and mark the task as sleeping - self.suspend_run(&mut atomic, &task_id, &run_id, timeout) + self.suspend_run(&mut atomic, &task_id, &run_id, timeout, Some(event_name)) .await?; atomic.commit().await.map_err(StorageError::SqlError)?; @@ -1180,46 +1150,6 @@ impl Storage { }) } - /// Store a wait for a task - /// It is assumed that event_name are globally unique, and on a conflict, - /// wait record is updated to reflect the provided run information. - async fn store_wait( - &self, - conn: &mut PgConnection, - task_id: &TaskId, - run_id: &RunId, - step_name: &str, - event_name: &str, - timeout: Duration, - ) -> Result<(), StorageError> { - let timeout = Utc::now() + timeout; - // TODO If `taskturbine.runs` had a nullable wait_id column, we could optimize - // fail_run/complete_run to not delete waits that don't exist. - sqlx::query( - "INSERT INTO taskturbine.waits (usecase, task_id, run_id, step_name, event_name, timeout_at, created_at) - VALUES ($1, $2, $3, $4, $5, $6, NOW()) - ON CONFLICT (usecase, event_name) - DO UPDATE - SET usecase = EXCLUDED.usecase, - task_id = EXCLUDED.task_id, - run_id = EXCLUDED.run_id, - step_name = EXCLUDED.step_name, - timeout_at = EXCLUDED.timeout_at, - created_at = EXCLUDED.created_at" - ) - .bind(&self.config.usecase) - .bind(task_id.0) - .bind(run_id) - .bind(step_name) - .bind(event_name) - .bind(timeout) - .execute(conn) - .await - .map_err(StorageError::SqlError)?; - - Ok(()) - } - /// Record a checkpoint for a task at a given step. /// If the checkpoint already exists, it will be updated with the run_id and state. async fn store_checkpoint( @@ -1281,13 +1211,17 @@ impl Storage { } } - /// Advance a task and run to sleeping state until available_at + /// Advance a task and run to sleeping state until available_at has elapsed + /// + /// Can optionally set a event_name to wait for. Event names are assumed to be unique + /// within a usecase. async fn suspend_run( &self, conn: &mut PgConnection, task_id: &TaskId, run_id: &RunId, available_in: Duration, + wait_event_name: Option<&str> ) -> Result<(), StorageError> { let available_at = Utc::now() + available_in; sqlx::query( @@ -1296,13 +1230,15 @@ impl Storage { SET state = $1, claimed_by = NULL, claim_expires_at = NULL, - available_at = $2 - WHERE run_id = $3 + available_at = $2, + wait_event_name = $3 + WHERE run_id = $4 ) - UPDATE taskturbine.tasks SET state = $1 WHERE task_id = $4", + UPDATE taskturbine.tasks SET state = $1 WHERE task_id = $5", ) .bind(TaskState::Sleeping) .bind(available_at) + .bind(wait_event_name) .bind(run_id) .bind(task_id) .execute(&mut *conn) @@ -1339,17 +1275,17 @@ impl Storage { // Clear any valid waits, and wake up those runs. sqlx::query( "WITH matching_waits AS ( - DELETE FROM taskturbine.waits - WHERE event_name = $1 - AND usecase = $2 - AND (timeout_at IS NULL OR timeout_at >= NOW()) - RETURNING run_id + SELECT r.run_id + FROM taskturbine.runs AS r + INNER JOIN taskturbine.tasks AS t ON r.task_id = t.task_id + WHERE r.wait_event_name = $1 AND t.usecase = $2 ), updated_runs AS ( UPDATE taskturbine.runs SET state = $3, available_at = NOW(), claimed_by = NULL, + wait_event_name = NULL, claim_expires_at = NULL WHERE run_id IN (SELECT run_id FROM matching_waits) RETURNING task_id @@ -1582,13 +1518,10 @@ mod tests { .await; assert!(res.is_ok()); - let wait_res = storage.get_wait_by_run_id(spawned.run_id).await; - assert!(wait_res.is_ok()); - assert!( - wait_res.unwrap().is_none(), - "wait should be deleted on run completion" - ); + let waiting_runs = storage.get_runs_waiting("event_name").await; + assert!(waiting_runs.is_ok()); + assert_eq!(waiting_runs.unwrap().len(), 0, "no runs waiting"); } #[tokio::test] @@ -1693,10 +1626,10 @@ mod tests { ) .await; assert!(res.is_ok()); - let wait_res = storage.get_wait_by_run_id(spawned.run_id).await; - assert!(wait_res.is_ok()); - let wait = wait_res.unwrap(); - assert!(wait.is_none(), "wait should be deleted on fail"); + let waiting = storage.get_runs_waiting("event_name").await; + assert!(waiting.is_ok()); + let waits = waiting.unwrap(); + assert_eq!(waits.len(), 0, "no waits remaining"); } #[tokio::test] @@ -1729,6 +1662,10 @@ mod tests { assert!(res.is_err()); let err = res.err().unwrap(); assert!(matches!(err, StorageError::NotRunning(_))); + + let waiting = storage.get_runs_waiting("event_name").await; + assert!(waiting.is_ok()); + assert_eq!(waiting.unwrap().len(), 0, "no wait stored"); } #[tokio::test] @@ -1908,17 +1845,18 @@ mod tests { .await; assert!(res.is_ok()); - let res = storage.get_wait_by_run_id(spawned.run_id).await; - let opt = res.unwrap(); - assert!(opt.is_some(), "a wait should be saved"); + let res = storage.get_runs_waiting(&event_id).await; + let rows = res.unwrap(); + assert_eq!(rows.len(), 1, "a wait should be saved"); - // Capture an event which should wait up the task + // Capture an event which should update the task to pending let res = storage.emit_event(&event_id, b"payload data").await; + dbg!(&res); assert!(res.is_ok()); - let res = storage.get_wait_by_run_id(spawned.run_id).await; - let opt = res.unwrap(); - assert!(opt.is_none(), "no wait should remain"); + let res = storage.get_runs_waiting(&event_id).await; + let rows = res.unwrap(); + assert_eq!(rows.len(), 0, "should be no waits remaining"); let run = storage.get_run(spawned.run_id).await.unwrap(); assert_eq!(run.state, TaskState::Pending); From 97a73e6edde8132f27b60560e54f5e87b761029a Mon Sep 17 00:00:00 2001 From: Mark Story Date: Mon, 7 Sep 2026 23:15:11 -0400 Subject: [PATCH 4/9] Add black and white logo --- assets/taskturbine-logo.svg | 151 ++++++++++++++++++++++++++++++++++++ 1 file changed, 151 insertions(+) create mode 100644 assets/taskturbine-logo.svg diff --git a/assets/taskturbine-logo.svg b/assets/taskturbine-logo.svg new file mode 100644 index 0000000..0c211e1 --- /dev/null +++ b/assets/taskturbine-logo.svg @@ -0,0 +1,151 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + From 83c92362b1407bef7c058e66d7f108a69931d07e Mon Sep 17 00:00:00 2001 From: Mark Story Date: Mon, 7 Sep 2026 23:16:14 -0400 Subject: [PATCH 5/9] Add color logo --- assets/taskturbine-logo-color.svg | 129 ++++++++++++++++++++++++++++++ 1 file changed, 129 insertions(+) create mode 100644 assets/taskturbine-logo-color.svg diff --git a/assets/taskturbine-logo-color.svg b/assets/taskturbine-logo-color.svg new file mode 100644 index 0000000..151bb5b --- /dev/null +++ b/assets/taskturbine-logo-color.svg @@ -0,0 +1,129 @@ + + + + + + + + + + + + + + + + + + + + + + From d3070bba93d79f6c6197de3253c8a8106c62ee96 Mon Sep 17 00:00:00 2001 From: Mark Story Date: Mon, 7 Sep 2026 23:16:25 -0400 Subject: [PATCH 6/9] Fix formatting --- taskturbine-core/src/storage.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/taskturbine-core/src/storage.rs b/taskturbine-core/src/storage.rs index 38bf383..52f48f0 100644 --- a/taskturbine-core/src/storage.rs +++ b/taskturbine-core/src/storage.rs @@ -1221,7 +1221,7 @@ impl Storage { task_id: &TaskId, run_id: &RunId, available_in: Duration, - wait_event_name: Option<&str> + wait_event_name: Option<&str>, ) -> Result<(), StorageError> { let available_at = Utc::now() + available_in; sqlx::query( From d25fd8b00646a6107ed9cfcd7017acce58962949 Mon Sep 17 00:00:00 2001 From: Mark Story Date: Mon, 7 Sep 2026 23:54:39 -0400 Subject: [PATCH 7/9] Don't use elegant_departure in tests Use channel close and receiver close to shutdown channels instead. This helped handle additional failure scenarios that could have resulted in stuck workers before. --- taskturbine-app/src/app.rs | 40 +++++++++++++++++++++----------------- 1 file changed, 22 insertions(+), 18 deletions(-) diff --git a/taskturbine-app/src/app.rs b/taskturbine-app/src/app.rs index 6de5408..5aca733 100644 --- a/taskturbine-app/src/app.rs +++ b/taskturbine-app/src/app.rs @@ -732,6 +732,10 @@ async fn claim_tasks(worker: Arc, work_send: Sender) { } } } + _ = 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, work_channel: Receiver) 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(); From dac6dcfdae8c7631233d314ecbb243c602342d5c Mon Sep 17 00:00:00 2001 From: Mark Story Date: Mon, 7 Sep 2026 23:56:47 -0400 Subject: [PATCH 8/9] Add logo to readme --- README.md | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/README.md b/README.md index 340bd7c..050cc0a 100644 --- a/README.md +++ b/README.md @@ -1,4 +1,8 @@ -# Taskturbine +

+ +
+ Taskturbine +

Taskturbine is a cross-language durable task framework for Rust & Python, using postgres for task storage. From a3cc51d64950b6f7a1bc164ee9d5bd6a63b7216e Mon Sep 17 00:00:00 2001 From: Mark Story Date: Mon, 7 Sep 2026 23:58:57 -0400 Subject: [PATCH 9/9] One more line break plz --- README.md | 1 + 1 file changed, 1 insertion(+) diff --git a/README.md b/README.md index 050cc0a..1e7e079 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,7 @@


+
Taskturbine