From 523abec37c8a6363971b845cebc98af4e102f79a Mon Sep 17 00:00:00 2001 From: santoshkumarradha Date: Mon, 14 Sep 2026 09:56:32 -0400 Subject: [PATCH] fix: commit task splits atomically --- src/db/tasks.rs | 77 ++++++++++++++++++++++++++--------------------- src/db/tests.rs | 79 +++++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 123 insertions(+), 33 deletions(-) diff --git a/src/db/tasks.rs b/src/db/tasks.rs index 9c9341c..bad1983 100644 --- a/src/db/tasks.rs +++ b/src/db/tasks.rs @@ -468,6 +468,22 @@ pub(crate) fn row_to_task(row: &rusqlite::Row<'_>) -> rusqlite::Result { pub fn create_task(db: &Database, task: &Task, tags: &[String]) -> Result { let conn = db.lock()?; + insert_task(&conn, task, tags)?; + drop(conn); + let task = get_task(db, &task.id)?; + let _ = crate::db::insert_event( + db, + Some(&task.id), + Some(&task.project_id), + task.agent_id.as_deref(), + crate::models::EventType::TaskCreated, + Some(serde_json::json!({"title": task.title})), + crate::db::now_utc_naive(), + ); + Ok(task) +} + +fn insert_task(conn: &rusqlite::Connection, task: &Task, tags: &[String]) -> Result<()> { let mut task_with_defaults = task.clone(); if task_with_defaults.max_retries < 0 { task_with_defaults.max_retries = 0; @@ -527,18 +543,7 @@ pub fn create_task(db: &Database, task: &Task, tags: &[String]) -> Result for tag in tags { conn.execute(INSERT_TASK_TAG, params![&task_id, tag])?; } - drop(conn); - let task = get_task(db, &task_id)?; - let _ = crate::db::insert_event( - db, - Some(&task.id), - Some(&task.project_id), - task.agent_id.as_deref(), - crate::models::EventType::TaskCreated, - Some(serde_json::json!({"title": task.title})), - crate::db::now_utc_naive(), - ); - Ok(task) + Ok(()) } pub fn get_task(db: &Database, task_id: &str) -> Result { @@ -1450,15 +1455,6 @@ pub fn split_task(db: &Database, task_id: &str, parts: Vec) -> Result return Err(anyhow::anyhow!("split requires at least one part")); } - let parent = get_task(db, task_id)?; - { - let conn = db.lock()?; - conn.execute( - "UPDATE tasks SET is_composite = 1, status = 'pending', agent_id = NULL, updated_at = datetime('now') WHERE id = ?1", - params![task_id], - )?; - } - let mut seen_titles = HashSet::new(); for part in &parts { if !seen_titles.insert(part.title.clone()) { @@ -1466,6 +1462,20 @@ pub fn split_task(db: &Database, task_id: &str, parts: Vec) -> Result } } + // Hold the connection and write transaction through the entire mutation so + // concurrent claimers cannot observe children before their edges exist. + // Any validation or SQL failure rolls back the parent, children and events. + let mut conn = db.lock()?; + let tx = conn.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?; + let parent = tx + .query_row(SELECT_TASK_BY_ID, params![task_id], row_to_task) + .optional()? + .ok_or_else(|| PlandbError::NotFound(format!("task {task_id}")))?; + tx.execute( + "UPDATE tasks SET is_composite = 1, status = 'pending', agent_id = NULL, updated_at = datetime('now') WHERE id = ?1", + params![task_id], + )?; + let now = now_utc_naive(); let mut title_to_id = HashMap::new(); let mut created = Vec::new(); @@ -1528,12 +1538,16 @@ pub fn split_task(db: &Database, task_id: &str, parts: Vec) -> Result .map(|value| serde_json::Value::String(value.clone())); } - let created_task = create_task(db, &task, &[])?; + insert_task(&tx, &task, &[])?; + tx.execute( + "INSERT INTO events(task_id, project_id, event_type, payload, timestamp) VALUES (?1, ?2, 'task_created', ?3, ?4)", + params![&task.id, &task.project_id, serde_json::json!({"title": task.title}).to_string(), dt_to_sql(now)], + )?; if part.done.unwrap_or(false) { - done_ids.push(created_task.id.clone()); + done_ids.push(task.id.clone()); } - title_to_id.insert(part.title.clone(), created_task.id.clone()); - created.push(created_task.id); + title_to_id.insert(part.title.clone(), task.id.clone()); + created.push(task.id); } for part in &parts { @@ -1545,19 +1559,16 @@ pub fn split_task(db: &Database, task_id: &str, parts: Vec) -> Result let from_id = title_to_id.get(dep_title).ok_or_else(|| { anyhow::anyhow!("deps_on references unknown part: {}", dep_title) })?; - add_dependency( - db, - from_id, - to_id, - crate::models::DependencyKind::FeedsInto, - crate::models::DependencyCondition::All, - None, + tx.execute( + "INSERT INTO dependencies(from_task, to_task, kind, condition) VALUES (?1, ?2, 'feeds_into', 'all')", + params![from_id, to_id], )?; } } } - let _ = promote_ready_tasks(db)?; + tx.execute(PROMOTE_READY, params![dt_to_sql(now)])?; + tx.commit()?; Ok(SplitResult { parent_task_id: task_id.to_string(), diff --git a/src/db/tests.rs b/src/db/tests.rs index 8491fe7..59be8df 100644 --- a/src/db/tests.rs +++ b/src/db/tests.rs @@ -71,6 +71,85 @@ fn create_project_and_task() { assert_eq!(fetched.title, "first"); } +#[test] +fn split_failure_preserves_parent_children_dependencies_and_events() { + let db = init_db(":memory:").unwrap(); + let project = create_project(&db, "atomic split", None, None).unwrap(); + let parent = create_task( + &db, + &make_task(&project.id, "parent", TaskStatus::Ready), + &[], + ) + .unwrap(); + claim_task(&db, &parent.id, "owner").unwrap(); + let parent = start_task(&db, &parent.id).unwrap(); + let before_events = list_events(&db, EventFilters::default()).unwrap().len(); + + for spec in [ + r#"[{"title":"A"},{"title":"A"}]"#, + r#"[{"title":"A"},{"title":"B","deps_on":["missing"]}]"#, + // The second edge fails at the SQL constraint, after the first exists. + r#"[{"title":"A"},{"title":"B","deps_on":["A","A"]}]"#, + ] { + let parts = serde_json::from_str(spec).unwrap(); + assert!(split_task(&db, &parent.id, parts).is_err()); + let after = get_task(&db, &parent.id).unwrap(); + assert_eq!( + serde_json::to_value(&after).unwrap(), + serde_json::to_value(&parent).unwrap() + ); + assert!(list_subtree(&db, &parent.id).unwrap().is_empty()); + assert_eq!( + list_events(&db, EventFilters::default()).unwrap().len(), + before_events + ); + let count: i64 = db + .lock() + .unwrap() + .query_row("SELECT COUNT(*) FROM dependencies", [], |r| r.get(0)) + .unwrap(); + assert_eq!(count, 0); + } +} + +#[test] +fn split_commits_children_dependencies_results_and_events_together() { + let db = init_db(":memory:").unwrap(); + let project = create_project(&db, "successful split", None, None).unwrap(); + let parent = create_task( + &db, + &make_task(&project.id, "parent", TaskStatus::Ready), + &[], + ) + .unwrap(); + let parts = serde_json::from_str(r#"[{"title":"A","done":true,"result":"contract"},{"title":"B","deps_on":["A"]},{"title":"C","deps_on":["B"]}]"#).unwrap(); + let split = split_task(&db, &parent.id, parts).unwrap(); + assert_eq!(split.created.len(), 3); + let a = get_task(&db, &split.title_to_id["A"]).unwrap(); + assert_eq!(a.result, Some(serde_json::json!("contract"))); + assert_eq!( + get_task(&db, &split.title_to_id["B"]).unwrap().status, + TaskStatus::Ready + ); + assert_eq!( + get_task(&db, &split.title_to_id["C"]).unwrap().status, + TaskStatus::Pending + ); + assert!(get_task(&db, &parent.id).unwrap().is_composite); + assert_eq!( + list_events( + &db, + EventFilters { + event_type: Some(EventType::TaskCreated), + ..Default::default() + } + ) + .unwrap() + .len(), + 4 + ); +} + #[test] fn claim_task_concurrency_single_winner() { let db_path = test_db_path();