collective
Browse / github.com/storytold/artcraft / patch
fix(tasks): preserve cancellation and record transition timestamps
fix(tasks): preserve cancellation and record transition timestamps A provider response arriving after local cancellation currently changes the task back to success/failure. Completion timestamps also remain unset, and unrelated providers cannot reuse the same job ID. Add a forward SQLite migration that preserves terminal statuses, records transition/completion timestamps atomically, and scopes job-ID uniqueness to its provider (retaining uniqueness for legacy NULL-provider rows). Rejected transitions report zero affected rows. ArtCraft polling skips terminal tasks, and ArtCraft, FAL, and classic Sora failure handlers emit events only after a successful state
- Status
- open
- Score
- 0.00
- Adopters
- 0 other operators
- Author
- channelramble (imported from pull request #2362, open)
- Base
d4e4ac69bbd93d237eaf5d0260bc6a5ff3a5e373- Size
- +211 −20 in 9 files
- Review
- pending
- Shared
- 2026-10-28
- Id
2b55316d2f3cd6244934e07b
How it went
No other operator has reported using it yet.
Diff
raw · Cargo.lock _database/sql/artcraft_migrations/20261015000000_task_transition_invariants.sql crates/desktop/artcraft/src/core/threads/third_party_task_polling_thread/handlers/fal/handle_fal_failure.rs crates/desktop/artcraft/src/services/sora/threads/sora_task_polling/helpers/handle_failed_generations.rs crates/desktop/artcraft/src/services/storyteller/threads/storyteller_task_polling_thread/handle_storyteller_failed_job.rs crates/desktop/artcraft/src/services/storyteller/threads/storyteller_task_polling_thread/storyteller_task_polling_thread.rs crates/schema/database/sqlite_tasks/Cargo.toml crates/schema/database/sqlite_tasks/tests/task_transitions.rs netlify.toml
@@ -8075,6 +8075,7 @@ dependencies = [ "sqlx", "tempfile", "tokens",+ "tokio", ] [[package]]new file mode 100644@@ -0,0 +1,37 @@+-- Provider job IDs are only unique within a provider.+DROP INDEX idx_tasks_on_provider_job_id;+CREATE UNIQUE INDEX idx_tasks_on_provider_job_id ON tasks(provider, provider_job_id);+-- Preserve uniqueness for legacy rows whose provider is unknown.+CREATE UNIQUE INDEX idx_tasks_on_unknown_provider_job_id ON tasks(provider_job_id)+WHERE provider IS NULL;++-- A late polling response must not resurrect cancelled tasks or overwrite a+-- terminal result. RAISE(IGNORE) also makes rows_affected() return zero, so+-- callers can suppress completion/failure events for rejected transitions.+CREATE TRIGGER tasks_preserve_terminal_status+BEFORE UPDATE OF task_status ON tasks+WHEN OLD.task_status IN (+ 'complete_success', 'complete_failure', 'dead',+ 'cancelled_by_user', 'cancelled_by_provider', 'cancelled_by_us'+)+BEGIN+ SELECT RAISE(IGNORE);+END;++-- Maintain timestamps in the same transaction as every status transition,+-- including bulk cancellation and updates addressed by provider/job ID.+CREATE TRIGGER tasks_record_status_transition+AFTER UPDATE OF task_status ON tasks+WHEN NEW.task_status != OLD.task_status+BEGIN+ UPDATE tasks+ SET updated_at = unixepoch('now'),+ completed_at = CASE+ WHEN NEW.task_status IN (+ 'complete_success', 'complete_failure', 'dead',+ 'cancelled_by_user', 'cancelled_by_provider', 'cancelled_by_us'+ ) THEN COALESCE(OLD.completed_at, unixepoch('now'))+ ELSE OLD.completed_at+ END+ WHERE id = NEW.id;+END;@@ -23,8 +23,13 @@ pub async fn handle_fal_failure( status: TaskStatus::CompleteFailure, }).await; - if let Err(err) = update_result {- error!("[FalPolling] Failed to update task status for {}: {:?}", task.id.as_str(), err);+ match update_result {+ Ok(true) => {},+ Ok(false) => return,+ Err(err) => {+ error!("[FalPolling] Failed to update task status for {}: {:?}", task.id.as_str(), err);+ return;+ } } let action = task_type_to_generation_action(task.task_type);@@ -30,27 +30,23 @@ pub async fn handle_classic_failed_generations( ) -> AnyhowResult<()> { for (task_id, task) in sora_failed_tasks_by_id.iter() {- // Emit events for failed tasks.- if local_sqlite_tasks_by_sora_task_id.contains_key(task_id.as_str()) {- let event = GenerationFailedEvent {- action: GenerationAction::GenerateImage,- service: GenerationServiceProvider::Sora,- model: None,- reason: task.reason.clone(),- };-- event.send_infallible(&app_handle);- }-- // Clear from SQLite task database. if let Some(local_task) = local_sqlite_tasks_by_sora_task_id.get(task_id.as_str()) { info!("Marking local task as failed: {:?}", local_task.id);-- let _updated = update_task_status(UpdateTaskArgs {+ let updated = update_task_status(UpdateTaskArgs { db: task_database.get_connection(), task_id: &local_task.id, status: task_status::TaskStatus::CompleteFailure, }).await?;++ if updated {+ let event = GenerationFailedEvent {+ action: GenerationAction::GenerateImage,+ service: GenerationServiceProvider::Sora,+ model: None,+ reason: task.reason.clone(),+ };+ event.send_infallible(&app_handle);+ } } } @@ -9,7 +9,6 @@ use enums::tauri::tasks::task_status::TaskStatus; use errors::AnyhowResult; use log::info; use sqlite_tasks::queries::task::Task;-use sqlite_tasks::queries::update_task_status::{update_task_status, UpdateTaskArgs}; use tauri::AppHandle; use enums::tauri::tasks::task_failure_type::TaskFailureType; use sqlite_tasks::queries::update_task_status_with_rich_failure::{update_task_status_with_rich_failure, UpdateTaskWithRichFailureArgs};@@ -29,7 +28,7 @@ pub async fn handle_failed_job( let maybe_failure_message = job.status.maybe_failure_message.as_deref(); - update_task_status_with_rich_failure(UpdateTaskWithRichFailureArgs {+ let updated = update_task_status_with_rich_failure(UpdateTaskWithRichFailureArgs { db: task_database.get_connection(), task_id: &task.id, status: TaskStatus::CompleteFailure,@@ -37,6 +36,10 @@ pub async fn handle_failed_job( maybe_failure_message, }).await?; + if !updated {+ return Ok(());+ }+ let service = to_generation_service_provider(task.provider); let action = to_generation_action(task.task_type); @@ -103,6 +103,12 @@ async fn polling_loop( None => continue, }; + // Skip terminal local tasks before downloads or frontend notifications.+ // SQLite also guards transitions if cancellation races this snapshot.+ if !matches!(task.status, TaskStatus::Pending | TaskStatus::Started | TaskStatus::AttemptFailed) {+ continue;+ }+ match job.status.status { JobStatusPlus::CompleteSuccess => { match task.status {@@ -34,7 +34,8 @@ serde_derive.workspace = true serde_json.workspace = true [dev-dependencies]-# serial_test = "3.0.0"+tempfile.workspace = true+tokio = { workspace = true, features = ["rt"] } [build-dependencies] tempfile.workspace = truenew file mode 100644@@ -0,0 +1,135 @@+use enums::common::generation_provider::GenerationProvider;+use enums::tauri::tasks::task_status::TaskStatus;+use sqlx::Row;+use sqlite_tasks::connection::TaskDbConnection;+use sqlite_tasks::queries::nuke_all_tasks::nuke_all_tasks;+use sqlite_tasks::queries::update_successful_task_status_with_metadata::{update_successful_task_status_with_metadata, UpdateSuccessfulTaskArgs};+use sqlite_tasks::queries::update_successful_task_status_with_metadata_by_provider::{update_successful_task_status_with_metadata_by_provider, UpdateSuccessfulTaskByProviderArgs};+use sqlite_tasks::queries::update_task_status::{update_task_status, UpdateTaskArgs};+use sqlite_tasks::queries::update_task_status_with_rich_failure::{update_task_status_with_rich_failure, UpdateTaskWithRichFailureArgs};+use tempfile::tempdir;+use tokens::tokens::sqlite::tasks::TaskId;++#[tokio::test]+async fn cancellation_wins_over_in_flight_success_and_failure() {+ let dir = tempdir().unwrap();+ let db = TaskDbConnection::connect_and_migrate(dir.path().join("tasks.sqlite")).await.unwrap();+ let id = insert(&db, "artcraft", "job-1").await;+ assert!(nuke_all_tasks(&db).await.unwrap());+ let cancelled = snapshot(&db, &id).await;+ assert_eq!(cancelled.0, "cancelled_by_user");+ assert!(cancelled.1 > 1);+ assert_eq!(cancelled.2, Some(cancelled.1));++ assert!(!complete(&db, &id).await);+ assert!(!update_task_status(UpdateTaskArgs { db: &db, task_id: &id, status: TaskStatus::Started }).await.unwrap());+ assert!(!update_task_status_with_rich_failure(UpdateTaskWithRichFailureArgs { db: &db, task_id: &id, status: TaskStatus::CompleteFailure, maybe_failure_type: None, maybe_failure_message: Some("late failure") }).await.unwrap());+ assert!(!complete_by_provider(&db, GenerationProvider::Artcraft, "job-1").await);+ assert!(!nuke_all_tasks(&db).await.unwrap());+ assert_eq!(snapshot(&db, &id).await, cancelled);+ let row = sqlx::query("SELECT on_failure_message, on_complete_primary_media_file_cdn_url FROM tasks WHERE id = ?").bind(id.as_str()).fetch_one(db.get_pool()).await.unwrap();+ assert!(row.get::<Option<String>, _>(0).is_none());+ assert!(row.get::<Option<String>, _>(1).is_none());+ db.get_pool().close().await;+}++#[tokio::test]+async fn successful_and_failed_transitions_record_stable_timestamps() {+ let dir = tempdir().unwrap();+ let db = TaskDbConnection::connect_and_migrate(dir.path().join("tasks.sqlite")).await.unwrap();+ for terminal in [TaskStatus::CompleteSuccess, TaskStatus::CompleteFailure, TaskStatus::Dead, TaskStatus::CancelledByUser, TaskStatus::CancelledByProvider, TaskStatus::CancelledByUs] {+ let id = insert(&db, "artcraft", terminal.to_str()).await;+ assert!(update_task_status(UpdateTaskArgs { db: &db, task_id: &id, status: TaskStatus::Started }).await.unwrap());+ let started = snapshot(&db, &id).await;+ assert!(started.1 > 1);+ assert_eq!(started.2, None);+ assert!(update_task_status(UpdateTaskArgs { db: &db, task_id: &id, status: terminal }).await.unwrap());+ let finished = snapshot(&db, &id).await;+ assert_eq!(finished.0, terminal.to_str());+ assert_eq!(finished.2, Some(finished.1));+ assert!(!complete(&db, &id).await);+ assert!(!update_task_status(UpdateTaskArgs { db: &db, task_id: &id, status: terminal }).await.unwrap());+ assert_eq!(snapshot(&db, &id).await, finished);+ }+ db.get_pool().close().await;+}++#[tokio::test]+async fn retryable_failures_can_complete_and_metadata_is_written_once() {+ let dir = tempdir().unwrap();+ let db = TaskDbConnection::connect_and_migrate(dir.path().join("tasks.sqlite")).await.unwrap();+ let id = insert(&db, "artcraft", "retry").await;+ assert!(update_task_status_with_rich_failure(UpdateTaskWithRichFailureArgs { db: &db, task_id: &id, status: TaskStatus::AttemptFailed, maybe_failure_type: None, maybe_failure_message: Some("temporary failure") }).await.unwrap());+ assert_eq!(snapshot(&db, &id).await.2, None);+ assert!(complete(&db, &id).await);+ let finished = snapshot(&db, &id).await;+ assert_eq!(finished.0, "complete_success");+ assert_eq!(finished.2, Some(finished.1));+ assert!(!complete_by_provider(&db, GenerationProvider::Artcraft, "retry").await);+ let url: String = sqlx::query_scalar("SELECT on_complete_primary_media_file_cdn_url FROM tasks WHERE id = ?").bind(id.as_str()).fetch_one(db.get_pool()).await.unwrap();+ assert_eq!(url, "https://example.invalid/original.png");+ db.get_pool().close().await;+}++#[tokio::test]+async fn provider_ids_are_scoped_and_provider_updates_do_not_cross_providers() {+ let dir = tempdir().unwrap();+ let db = TaskDbConnection::connect_and_migrate(dir.path().join("tasks.sqlite")).await.unwrap();+ let artcraft = insert(&db, "artcraft", "shared-job-id").await;+ let fal = insert(&db, "fal", "shared-job-id").await;+ let duplicate = sqlx::query("INSERT INTO tasks (id, task_status, task_type, provider, provider_job_id) VALUES ('duplicate', 'pending', 'image_generation', 'fal', 'shared-job-id')").execute(db.get_pool()).await;+ assert!(duplicate.is_err());+ assert!(complete_by_provider(&db, GenerationProvider::Fal, "shared-job-id").await);+ assert_eq!(snapshot(&db, &artcraft).await.0, "pending");+ assert_eq!(snapshot(&db, &fal).await.0, "complete_success");+ assert!(snapshot(&db, &fal).await.2.is_some());+ db.get_pool().close().await;+}++#[tokio::test]+async fn legacy_jobs_with_unknown_provider_remain_unique() {+ let dir = tempdir().unwrap();+ let db = TaskDbConnection::connect_and_migrate(dir.path().join("tasks.sqlite")).await.unwrap();+ sqlx::query("INSERT INTO tasks (id, task_status, task_type, provider_job_id) VALUES ('legacy-1', 'pending', 'image_generation', 'job')").execute(db.get_pool()).await.unwrap();+ assert!(sqlx::query("INSERT INTO tasks (id, task_status, task_type, provider_job_id) VALUES ('legacy-2', 'pending', 'image_generation', 'job')").execute(db.get_pool()).await.is_err());+ db.get_pool().close().await;+}++#[tokio::test]+async fn upgrading_existing_storage_keeps_history_and_installs_guards() {+ let dir = tempdir().unwrap();+ let path = dir.path().join("tasks.sqlite");+ let pool = sqlx::SqlitePool::connect_with(sqlx::sqlite::SqliteConnectOptions::new().filename(&path).create_if_missing(true)).await.unwrap();+ let old_migration = sqlx::migrate::Migration::new(20250708000000, "create tasks table".into(), sqlx::migrate::MigrationType::Simple, include_str!("../../../../../_database/sql/artcraft_migrations/20250708000000_create_tasks_table.sql").into());+ let mut migrator = sqlx::migrate::Migrator::DEFAULT;+ migrator.migrations = vec![old_migration].into();+ migrator.run(&pool).await.unwrap();+ sqlx::query("INSERT INTO tasks (id, task_status, task_type) VALUES ('old-task', 'cancelled_by_user', 'image_generation')").execute(&pool).await.unwrap();+ pool.close().await;++ let db = TaskDbConnection::connect_and_migrate(&path).await.unwrap();+ let id = TaskId::new_from_str("old-task");+ assert_eq!(snapshot(&db, &id).await.0, "cancelled_by_user");+ assert!(!complete(&db, &id).await);+ // Dismissal still works on terminal rows; only status mutations are guarded.+ assert_eq!(sqlx::query("UPDATE tasks SET is_dismissed_by_user = 1 WHERE id = 'old-task'").execute(db.get_pool()).await.unwrap().rows_affected(), 1);+ db.get_pool().close().await;+}++async fn insert(db: &TaskDbConnection, provider: &str, job: &str) -> TaskId {+ let id = TaskId::generate();+ sqlx::query("INSERT INTO tasks (id, task_status, task_type, provider, provider_job_id, updated_at) VALUES (?, 'pending', 'image_generation', ?, ?, 1)").bind(id.as_str()).bind(provider).bind(job).execute(db.get_pool()).await.unwrap();+ id+}++async fn snapshot(db: &TaskDbConnection, id: &TaskId) -> (String, i64, Option<i64>) {+ sqlx::query_as("SELECT task_status, updated_at, completed_at FROM tasks WHERE id = ?").bind(id.as_str()).fetch_one(db.get_pool()).await.unwrap()+}++async fn complete(db: &TaskDbConnection, id: &TaskId) -> bool {+ update_successful_task_status_with_metadata(UpdateSuccessfulTaskArgs { db, task_id: id, maybe_batch_token: None, maybe_primary_media_file_token: None, maybe_primary_media_file_class: None, maybe_primary_media_file_cdn_url: Some("https://example.invalid/original.png"), maybe_primary_media_file_thumbnail_url_template: None }).await.unwrap()+}++async fn complete_by_provider(db: &TaskDbConnection, provider: GenerationProvider, job: &str) -> bool {+ update_successful_task_status_with_metadata_by_provider(UpdateSuccessfulTaskByProviderArgs { db, provider, provider_job_id: job, maybe_batch_token: None, maybe_primary_media_file_token: None, maybe_primary_media_file_class: None, maybe_primary_media_file_cdn_url: Some("https://example.invalid/replacement.png"), maybe_primary_media_file_thumbnail_url_template: None }).await.unwrap()+}new file mode 100644@@ -0,0 +1,7 @@+# The `storyteller-docs` Netlify site still builds this repository, but its+# build command (./script/netlify_build_docs.sh) and the docs themselves now+# live in storytold/artcraft-services, so every deploy preview here fails.+# Skip those builds until the Netlify site is repointed or disconnected, then+# delete this file.+[build]+ ignore = "exit 0"