From fb271932756340a529f6abceecc665fc72665422 Mon Sep 17 00:00:00 2001 From: Vicente Reig Rincon de Arellano Date: Sun, 5 Jul 2026 17:05:51 +0200 Subject: [PATCH] fix stale sync claim ownership --- crates/garmin-cli/src/storage/sync_db.rs | 253 ++++++++++++++++++++--- crates/garmin-cli/src/sync/mod.rs | 165 ++++++++++++--- crates/garmin-cli/src/sync/task_queue.rs | 45 +++- 3 files changed, 392 insertions(+), 71 deletions(-) diff --git a/crates/garmin-cli/src/storage/sync_db.rs b/crates/garmin-cli/src/storage/sync_db.rs index 884eeb4..6a3cb92 100644 --- a/crates/garmin-cli/src/storage/sync_db.rs +++ b/crates/garmin-cli/src/storage/sync_db.rs @@ -400,26 +400,7 @@ impl SyncDb { /// Push a task to the queue pub fn push_task(&self, task: &SyncTask) -> Result { - let task_data = serde_json::to_string(&task.task_type) - .map_err(|e| GarminError::Database(format!("Failed to serialize task: {}", e)))?; - - self.conn - .execute( - "INSERT INTO sync_tasks (profile_id, task_type, task_data, pipeline, status, attempts, last_error) - VALUES (?, ?, ?, ?, ?, ?, ?)", - params![ - task.profile_id, - task_type_name(&task.task_type), - task_data, - pipeline_name(task.pipeline), - task.status.to_string(), - task.attempts, - task.last_error, - ], - ) - .map_err(|e| GarminError::Database(format!("Failed to push task: {}", e)))?; - - Ok(self.conn.last_insert_rowid()) + insert_task(&self.conn, task) } /// Pop the next task from the queue for a profile @@ -610,35 +591,109 @@ impl SyncDb { } /// Mark a task as completed - pub fn mark_task_completed(&self, task_id: i64) -> Result<()> { - self.conn + pub fn mark_task_completed(&self, task_id: i64, claim_attempt: i32) -> Result<()> { + let count = self + .conn .execute( "UPDATE sync_tasks SET status = 'completed', completed_at = datetime('now'), in_progress_at = NULL - WHERE id = ?", - params![task_id], + WHERE id = ? + AND attempts = ? + AND status = 'in_progress'", + params![task_id, claim_attempt], + ) + .map_err(|e| GarminError::Database(format!("Failed to mark task completed: {}", e)))?; + + if count == 0 { + return Err(GarminError::Database(format!( + "Task {} claim attempt {} is not eligible to complete", + task_id, claim_attempt + ))); + } + + Ok(()) + } + + /// Mark a task as completed and enqueue follow-up tasks in the same claim-guarded transaction. + pub fn mark_task_completed_with_followups( + &self, + task_id: i64, + claim_attempt: i32, + followups: &[SyncTask], + ) -> Result<()> { + let tx = self.conn.unchecked_transaction().map_err(|e| { + GarminError::Database(format!( + "Failed to start task completion transaction: {}", + e + )) + })?; + + let count = tx + .execute( + "UPDATE sync_tasks + SET status = 'completed', + completed_at = datetime('now'), + in_progress_at = NULL + WHERE id = ? + AND attempts = ? + AND status = 'in_progress'", + params![task_id, claim_attempt], ) .map_err(|e| GarminError::Database(format!("Failed to mark task completed: {}", e)))?; + if count == 0 { + return Err(GarminError::Database(format!( + "Task {} claim attempt {} is not eligible to complete", + task_id, claim_attempt + ))); + } + + for task in followups { + insert_task(&tx, task)?; + } + + tx.commit().map_err(|e| { + GarminError::Database(format!( + "Failed to commit task completion transaction: {}", + e + )) + })?; + Ok(()) } /// Mark a task as failed - pub fn mark_task_failed(&self, task_id: i64, error: &str, retry_delay_secs: i64) -> Result<()> { - self.conn + pub fn mark_task_failed( + &self, + task_id: i64, + claim_attempt: i32, + error: &str, + retry_delay_secs: i64, + ) -> Result<()> { + let count = self + .conn .execute( "UPDATE sync_tasks SET status = 'failed', last_error = ?, in_progress_at = NULL, next_retry_at = datetime('now', '+' || ? || ' seconds') - WHERE id = ?", - params![error, retry_delay_secs, task_id], + WHERE id = ? + AND attempts = ? + AND status = 'in_progress'", + params![error, retry_delay_secs, task_id, claim_attempt], ) .map_err(|e| GarminError::Database(format!("Failed to mark task failed: {}", e)))?; + if count == 0 { + return Err(GarminError::Database(format!( + "Task {} claim attempt {} is not eligible to fail", + task_id, claim_attempt + ))); + } + Ok(()) } @@ -924,6 +979,28 @@ fn pipeline_name(pipeline: SyncPipeline) -> &'static str { } } +fn insert_task(conn: &Connection, task: &SyncTask) -> Result { + let task_data = serde_json::to_string(&task.task_type) + .map_err(|e| GarminError::Database(format!("Failed to serialize task: {}", e)))?; + + conn.execute( + "INSERT INTO sync_tasks (profile_id, task_type, task_data, pipeline, status, attempts, last_error) + VALUES (?, ?, ?, ?, ?, ?, ?)", + params![ + task.profile_id, + task_type_name(&task.task_type), + task_data, + pipeline_name(task.pipeline), + task.status.to_string(), + task.attempts, + task.last_error, + ], + ) + .map_err(|e| GarminError::Database(format!("Failed to push task: {}", e)))?; + + Ok(conn.last_insert_rowid()) +} + fn parse_pipeline(s: &str) -> SyncPipeline { match s { "backfill" => SyncPipeline::Backfill, @@ -1054,7 +1131,7 @@ mod tests { assert_eq!(popped.id, Some(id)); db.mark_task_in_progress(id).unwrap(); - db.mark_task_completed(id).unwrap(); + db.mark_task_completed(id, 1).unwrap(); // Should be no more pending tasks let next = db.pop_task(1, None).unwrap(); @@ -1288,7 +1365,7 @@ mod tests { let id = db.push_task(&task).unwrap(); db.mark_task_in_progress(id).unwrap(); - db.mark_task_failed(id, "boom", 60).unwrap(); + db.mark_task_failed(id, 1, "boom", 60).unwrap(); let (_activities, _gpx, health, _perf) = db.count_tasks_by_type(1, None).unwrap(); assert_eq!(health, 1); @@ -1421,4 +1498,120 @@ mod tests { let active = db.list_in_progress_tasks(1).unwrap(); assert_eq!(active[0].attempts, 1); } + + #[test] + fn test_stale_claim_owner_cannot_complete_newer_claim() { + let db = SyncDb::open_in_memory().unwrap(); + + let id = db + .push_task(&SyncTask::new( + 1, + SyncPipeline::Frontier, + SyncTaskType::DailyHealth { + date: NaiveDate::from_ymd_opt(2026, 7, 5).unwrap(), + }, + )) + .unwrap(); + + let first_claim = db.claim_next_task(1, None).unwrap().unwrap(); + assert_eq!(first_claim.id, Some(id)); + assert_eq!(first_claim.attempts, 1); + + db.set_in_progress_at_seconds_ago_for_test(id, IN_PROGRESS_RECOVERY_AFTER_SECS + 60) + .unwrap(); + assert_eq!(db.recover_in_progress_tasks().unwrap(), 1); + + let second_claim = db.claim_next_task(1, None).unwrap().unwrap(); + assert_eq!(second_claim.id, Some(id)); + assert_eq!(second_claim.attempts, 2); + + assert!(db.mark_task_completed(id, first_claim.attempts).is_err()); + + let active = db.list_in_progress_tasks(1).unwrap(); + assert_eq!(active.len(), 1); + assert_eq!(active[0].id, id); + assert_eq!(active[0].attempts, second_claim.attempts); + + db.mark_task_completed(id, second_claim.attempts).unwrap(); + let active = db.list_in_progress_tasks(1).unwrap(); + assert!(active.is_empty()); + } + + #[test] + fn test_stale_claim_owner_cannot_enqueue_followups() { + let db = SyncDb::open_in_memory().unwrap(); + + let id = db + .push_task(&SyncTask::new( + 1, + SyncPipeline::Frontier, + SyncTaskType::Activities { + start: 0, + limit: 50, + min_date: None, + max_date: None, + }, + )) + .unwrap(); + + let first_claim = db.claim_next_task(1, None).unwrap().unwrap(); + assert_eq!(first_claim.id, Some(id)); + assert_eq!(first_claim.attempts, 1); + + db.set_in_progress_at_seconds_ago_for_test(id, IN_PROGRESS_RECOVERY_AFTER_SECS + 60) + .unwrap(); + assert_eq!(db.recover_in_progress_tasks().unwrap(), 1); + + let second_claim = db.claim_next_task(1, None).unwrap().unwrap(); + assert_eq!(second_claim.id, Some(id)); + assert_eq!(second_claim.attempts, 2); + + let followups = vec![ + SyncTask::new( + 1, + SyncPipeline::Frontier, + SyncTaskType::DownloadGpx { + activity_id: 42, + activity_name: Some("Run".to_string()), + activity_date: Some("2026-07-05".to_string()), + }, + ), + SyncTask::new( + 1, + SyncPipeline::Frontier, + SyncTaskType::Activities { + start: 50, + limit: 50, + min_date: None, + max_date: None, + }, + ), + ]; + + assert!(db + .mark_task_completed_with_followups(id, first_claim.attempts, &followups) + .is_err()); + + let active = db.list_in_progress_tasks(1).unwrap(); + assert_eq!(active.len(), 1); + assert_eq!(active[0].id, id); + assert_eq!(active[0].attempts, second_claim.attempts); + + let by_type = db + .count_tasks_by_type(1, Some(SyncPipeline::Frontier)) + .unwrap(); + assert_eq!(by_type, (1, 0, 0, 0)); + let by_status = db.count_tasks_by_status(1).unwrap(); + assert_eq!(by_status, (0, 1, 0, 0)); + + db.mark_task_completed_with_followups(id, second_claim.attempts, &followups) + .unwrap(); + + let by_type = db + .count_tasks_by_type(1, Some(SyncPipeline::Frontier)) + .unwrap(); + assert_eq!(by_type, (1, 1, 0, 0)); + let by_status = db.count_tasks_by_status(1).unwrap(); + assert_eq!(by_status, (2, 0, 1, 0)); + } } diff --git a/crates/garmin-cli/src/sync/mod.rs b/crates/garmin-cli/src/sync/mod.rs index 25ae66e..bbf8bf5 100644 --- a/crates/garmin-cli/src/sync/mod.rs +++ b/crates/garmin-cli/src/sync/mod.rs @@ -47,13 +47,19 @@ enum SyncData { gpx_tasks: Vec, next_page: Option, task_id: i64, + claim_attempt: i32, }, /// Daily health record - Health { record: DailyHealth, task_id: i64 }, + Health { + record: DailyHealth, + task_id: i64, + claim_attempt: i32, + }, /// Performance metrics record Performance { record: PerformanceMetrics, task_id: i64, + claim_attempt: i32, }, /// Track points from GPX TrackPoints { @@ -62,9 +68,20 @@ enum SyncData { date: NaiveDate, points: Vec, task_id: i64, + claim_attempt: i32, }, } +struct ConsumerWriteOutcome { + write_result: Result<()>, + task_id: i64, + claim_attempt: i32, + task_type: &'static str, + followup_tasks: Vec, + gpx_followups: u32, + activity_followups: u32, +} + /// Sync engine for orchestrating data synchronization pub struct SyncEngine { storage: Storage, @@ -1305,6 +1322,7 @@ async fn producer_loop( }; let task_id = task.id.unwrap(); + let claim_attempt = task.attempts; context.in_flight.fetch_add(1, Ordering::Relaxed); // Update progress display @@ -1334,7 +1352,7 @@ async fn producer_loop( }; if should_skip { - if let Err(e) = queue.mark_completed(task_id).await { + if let Err(e) = queue.mark_completed(task_id, claim_attempt).await { eprintln!("Failed to mark task completed: {}", e); } complete_progress_for_task(&task, &context.progress); @@ -1370,7 +1388,7 @@ async fn producer_loop( context.in_flight.fetch_sub(1, Ordering::Relaxed); let backoff = Duration::seconds(60); let _ = queue - .mark_failed(task_id, "Consumer channel closed", backoff) + .mark_failed(task_id, claim_attempt, "Consumer channel closed", backoff) .await; break; } @@ -1378,7 +1396,10 @@ async fn producer_loop( Err(GarminError::RateLimited) => { context.rate_limiter.on_rate_limit(); let backoff = Duration::seconds(60); - if let Err(e) = queue.mark_failed(task_id, "Rate limited", backoff).await { + if let Err(e) = queue + .mark_failed(task_id, claim_attempt, "Rate limited", backoff) + .await + { eprintln!("Failed to mark task as rate limited: {}", e); } fail_progress_for_task(&task, &context.progress, "Rate limited"); @@ -1391,7 +1412,10 @@ async fn producer_loop( Err(e) => { let backoff = Duration::seconds(60); let error_msg = e.to_string(); - if let Err(e) = queue.mark_failed(task_id, &error_msg, backoff).await { + if let Err(e) = queue + .mark_failed(task_id, claim_attempt, &error_msg, backoff) + .await + { eprintln!("Failed to mark task as failed: {}", e); } fail_progress_for_task(&task, &context.progress, &error_msg); @@ -1568,6 +1592,7 @@ async fn fetch_task_data( profile_id: i32, ) -> Result { let task_id = task.id.unwrap(); + let claim_attempt = task.attempts; match &task.task_type { SyncTaskType::Activities { @@ -1663,6 +1688,7 @@ async fn fetch_task_data( gpx_tasks, next_page, task_id, + claim_attempt, }) } @@ -1688,6 +1714,7 @@ async fn fetch_task_data( date, points, task_id, + claim_attempt, }) } @@ -1818,7 +1845,11 @@ async fn fetch_task_data( raw_json: Some(health), }; - Ok(SyncData::Health { record, task_id }) + Ok(SyncData::Health { + record, + task_id, + claim_attempt, + }) } SyncTaskType::Performance { date } => { @@ -1919,7 +1950,11 @@ async fn fetch_task_data( raw_json: None, }; - Ok(SyncData::Performance { record, task_id }) + Ok(SyncData::Performance { + record, + task_id, + claim_attempt, + }) } _ => { @@ -2186,13 +2221,17 @@ async fn consumer_loop( gpx_tasks, next_page, task_id, + claim_attempt, } => { // Write activities to Parquet let write_result = parquet.upsert_activities_async(records).await; + let mut followup_tasks = Vec::new(); + let mut gpx_followups = 0u32; + let mut activity_followups = 0u32; + if write_result.is_ok() { - // Queue GPX tasks and update progress totals - let mut gpx_added = 0u32; + // Build follow-up tasks, but persist them only after the claim token is accepted. for gpx_task in gpx_tasks { let should_skip = match &gpx_task.task_type { SyncTaskType::DownloadGpx { @@ -2211,60 +2250,113 @@ async fn consumer_loop( continue; } - if let Err(e) = queue.push(gpx_task.clone()).await { - eprintln!("Failed to queue GPX task: {}", e); - } else { - gpx_added += 1; - } - } - if gpx_added > 0 { - progress.gpx.add_total(gpx_added); + followup_tasks.push(gpx_task.clone()); + gpx_followups += 1; } - // Queue next page if there is one if let Some(next) = next_page { - if let Err(e) = queue.push(next.clone()).await { - eprintln!("Failed to queue next page: {}", e); - } - progress.activities.add_total(1); + followup_tasks.push(next.clone()); + activity_followups = 1; } } - (write_result, *task_id, "Activities") + ConsumerWriteOutcome { + write_result, + task_id: *task_id, + claim_attempt: *claim_attempt, + task_type: "Activities", + followup_tasks, + gpx_followups, + activity_followups, + } } - SyncData::Health { record, task_id } => { + SyncData::Health { + record, + task_id, + claim_attempt, + } => { let result = parquet .upsert_daily_health_async(std::slice::from_ref(record)) .await; - (result, *task_id, "Health") + ConsumerWriteOutcome { + write_result: result, + task_id: *task_id, + claim_attempt: *claim_attempt, + task_type: "Health", + followup_tasks: Vec::new(), + gpx_followups: 0, + activity_followups: 0, + } } - SyncData::Performance { record, task_id } => { + SyncData::Performance { + record, + task_id, + claim_attempt, + } => { let result = parquet .upsert_performance_metrics_async(std::slice::from_ref(record)) .await; - (result, *task_id, "Performance") + ConsumerWriteOutcome { + write_result: result, + task_id: *task_id, + claim_attempt: *claim_attempt, + task_type: "Performance", + followup_tasks: Vec::new(), + gpx_followups: 0, + activity_followups: 0, + } } SyncData::TrackPoints { date, points, task_id, + claim_attempt, .. } => { let result = parquet.write_track_points_async(*date, points).await; - (result, *task_id, "GPX") + ConsumerWriteOutcome { + write_result: result, + task_id: *task_id, + claim_attempt: *claim_attempt, + task_type: "GPX", + followup_tasks: Vec::new(), + gpx_followups: 0, + activity_followups: 0, + } } }; - let (write_result, task_id, task_type) = result; + let ConsumerWriteOutcome { + write_result, + task_id, + claim_attempt, + task_type, + followup_tasks, + gpx_followups, + activity_followups, + } = result; match write_result { Ok(()) => { - // Mark task completed - if let Err(e) = queue.mark_completed(task_id).await { + // Mark task completed and persist follow-up work behind the same claim token. + if let Err(e) = queue + .mark_completed_with_followups(task_id, claim_attempt, &followup_tasks) + .await + { eprintln!("Failed to mark task completed: {}", e); + in_flight.fetch_sub(1, Ordering::Relaxed); + continue; + } + + if gpx_followups > 0 { + progress.gpx.add_total(gpx_followups); + } + + if activity_followups > 0 { + progress.activities.add_total(activity_followups); } // Update stats @@ -2299,7 +2391,10 @@ async fn consumer_loop( // Mark task failed let backoff = Duration::seconds(60); let error_msg = e.to_string(); - if let Err(e) = queue.mark_failed(task_id, &error_msg, backoff).await { + if let Err(e) = queue + .mark_failed(task_id, claim_attempt, &error_msg, backoff) + .await + { eprintln!("Failed to mark task as failed: {}", e); } @@ -2446,7 +2541,11 @@ mod tests { raw_json: None, }; - let data = SyncData::Health { record, task_id: 1 }; + let data = SyncData::Health { + record, + task_id: 1, + claim_attempt: 1, + }; record_write_failure(&data, &progress, "write failed"); assert_eq!(progress.health.get_failed(), 1); diff --git a/crates/garmin-cli/src/sync/task_queue.rs b/crates/garmin-cli/src/sync/task_queue.rs index 47b14ea..e92e151 100644 --- a/crates/garmin-cli/src/sync/task_queue.rs +++ b/crates/garmin-cli/src/sync/task_queue.rs @@ -114,14 +114,31 @@ impl TaskQueue { } /// Mark a task as completed - pub fn mark_completed(&self, task_id: i64) -> Result<()> { - self.sync_db.mark_task_completed(task_id) + pub fn mark_completed(&self, task_id: i64, claim_attempt: i32) -> Result<()> { + self.sync_db.mark_task_completed(task_id, claim_attempt) + } + + /// Mark a task as completed and enqueue follow-up tasks atomically + pub fn mark_completed_with_followups( + &self, + task_id: i64, + claim_attempt: i32, + followups: &[SyncTask], + ) -> Result<()> { + self.sync_db + .mark_task_completed_with_followups(task_id, claim_attempt, followups) } /// Mark a task as failed with retry - pub fn mark_failed(&self, task_id: i64, error: &str, retry_after: Duration) -> Result<()> { + pub fn mark_failed( + &self, + task_id: i64, + claim_attempt: i32, + error: &str, + retry_after: Duration, + ) -> Result<()> { self.sync_db - .mark_task_failed(task_id, error, retry_after.num_seconds()) + .mark_task_failed(task_id, claim_attempt, error, retry_after.num_seconds()) } /// Recover tasks that were in progress (crashed) @@ -247,20 +264,32 @@ impl SharedTaskQueue { } /// Mark a task as completed (thread-safe) - pub async fn mark_completed(&self, task_id: i64) -> Result<()> { + pub async fn mark_completed(&self, task_id: i64, claim_attempt: i32) -> Result<()> { + let guard = self.inner.lock().await; + guard.mark_completed(task_id, claim_attempt) + } + + /// Mark a task as completed and enqueue follow-up tasks atomically (thread-safe) + pub async fn mark_completed_with_followups( + &self, + task_id: i64, + claim_attempt: i32, + followups: &[SyncTask], + ) -> Result<()> { let guard = self.inner.lock().await; - guard.mark_completed(task_id) + guard.mark_completed_with_followups(task_id, claim_attempt, followups) } /// Mark a task as failed with retry (thread-safe) pub async fn mark_failed( &self, task_id: i64, + claim_attempt: i32, error: &str, retry_after: Duration, ) -> Result<()> { let guard = self.inner.lock().await; - guard.mark_failed(task_id, error, retry_after) + guard.mark_failed(task_id, claim_attempt, error, retry_after) } /// Get count of pending tasks (thread-safe) @@ -364,7 +393,7 @@ mod tests { let id = queue.push(task).unwrap(); queue.mark_in_progress(id).unwrap(); - queue.mark_completed(id).unwrap(); + queue.mark_completed(id, 1).unwrap(); // Should not pop completed tasks let popped = queue.pop().unwrap();