From 19404805704598826c0d0d4ed6b225398efd0f95 Mon Sep 17 00:00:00 2001 From: Han Xu Date: Sun, 28 Jun 2026 06:57:29 +0000 Subject: [PATCH 1/3] feat(sync): claim tasks atomically --- crates/garmin-cli/src/storage/mod.rs | 2 +- crates/garmin-cli/src/storage/sync_db.rs | 129 ++++++++++++++++++++++- crates/garmin-cli/src/sync/task_queue.rs | 48 +++++++++ 3 files changed, 173 insertions(+), 6 deletions(-) diff --git a/crates/garmin-cli/src/storage/mod.rs b/crates/garmin-cli/src/storage/mod.rs index 7d97a82..16792e5 100644 --- a/crates/garmin-cli/src/storage/mod.rs +++ b/crates/garmin-cli/src/storage/mod.rs @@ -46,7 +46,7 @@ mod sync_db; pub use parquet::ParquetStore; pub use partitions::EntityType; -pub use sync_db::SyncDb; +pub use sync_db::{ActiveSyncTask, SyncDb}; use std::path::PathBuf; diff --git a/crates/garmin-cli/src/storage/sync_db.rs b/crates/garmin-cli/src/storage/sync_db.rs index 01f261e..e6542ee 100644 --- a/crates/garmin-cli/src/storage/sync_db.rs +++ b/crates/garmin-cli/src/storage/sync_db.rs @@ -6,12 +6,23 @@ use std::path::Path; -use chrono::{DateTime, NaiveDate, Utc}; +use chrono::{DateTime, NaiveDate, NaiveDateTime, Utc}; use rusqlite::{params, Connection, OptionalExtension}; use crate::db::models::{SyncPipeline, SyncState, SyncTask, SyncTaskType, TaskStatus}; use crate::error::{GarminError, Result}; +/// Task currently marked in progress in the sync queue. +#[derive(Debug, Clone)] +pub struct ActiveSyncTask { + pub id: i64, + pub task_type: SyncTaskType, + pub pipeline: SyncPipeline, + pub attempts: i32, + pub last_error: Option, + pub in_progress_at: Option>, +} + /// SQLite database for sync state and task queue pub struct SyncDb { conn: Connection, @@ -62,6 +73,7 @@ impl SyncDb { attempts INTEGER NOT NULL DEFAULT 0, last_error TEXT, created_at TEXT NOT NULL DEFAULT (datetime('now')), + in_progress_at TEXT, next_retry_at TEXT, completed_at TEXT ); @@ -105,6 +117,17 @@ impl SyncDb { })?; } + if !self.column_exists("sync_tasks", "in_progress_at")? { + self.conn + .execute("ALTER TABLE sync_tasks ADD COLUMN in_progress_at TEXT", []) + .map_err(|e| { + GarminError::Database(format!( + "Failed to add in_progress_at column to sync_tasks: {}", + e + )) + })?; + } + self.conn .execute( "CREATE INDEX IF NOT EXISTS idx_sync_tasks_pipeline_status @@ -544,10 +567,17 @@ impl SyncDb { pub fn mark_task_in_progress(&self, task_id: i64) -> Result<()> { self.conn .execute( - "UPDATE sync_tasks SET status = 'in_progress', attempts = attempts + 1 WHERE id = ?", + "UPDATE sync_tasks + SET status = 'in_progress', + attempts = attempts + 1, + in_progress_at = datetime('now'), + last_error = NULL + WHERE id = ?", params![task_id], ) - .map_err(|e| GarminError::Database(format!("Failed to mark task in progress: {}", e)))?; + .map_err(|e| { + GarminError::Database(format!("Failed to mark task in progress: {}", e)) + })?; Ok(()) } @@ -556,7 +586,11 @@ impl SyncDb { pub fn mark_task_completed(&self, task_id: i64) -> Result<()> { self.conn .execute( - "UPDATE sync_tasks SET status = 'completed', completed_at = datetime('now') WHERE id = ?", + "UPDATE sync_tasks + SET status = 'completed', + completed_at = datetime('now'), + in_progress_at = NULL + WHERE id = ?", params![task_id], ) .map_err(|e| GarminError::Database(format!("Failed to mark task completed: {}", e)))?; @@ -571,6 +605,7 @@ impl SyncDb { "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], @@ -585,7 +620,10 @@ impl SyncDb { let count = self .conn .execute( - "UPDATE sync_tasks SET status = 'pending' WHERE status = 'in_progress'", + "UPDATE sync_tasks + SET status = 'pending', + in_progress_at = NULL + WHERE status = 'in_progress'", [], ) .map_err(|e| GarminError::Database(format!("Failed to recover tasks: {}", e)))?; @@ -704,6 +742,53 @@ impl SyncDb { Ok((activities, gpx, health, performance)) } + /// List tasks currently marked in progress for a profile. + pub fn list_in_progress_tasks(&self, profile_id: i32) -> Result> { + let mut stmt = self + .conn + .prepare( + "SELECT id, task_data, pipeline, attempts, last_error, in_progress_at + FROM sync_tasks + WHERE profile_id = ? AND status = 'in_progress' + ORDER BY in_progress_at ASC, id ASC", + ) + .map_err(|e| { + GarminError::Database(format!("Failed to prepare active task query: {}", e)) + })?; + + let rows = stmt + .query_map(params![profile_id], |row| { + let task_data: String = row.get(1)?; + let task_type: SyncTaskType = serde_json::from_str(&task_data).map_err(|e| { + rusqlite::Error::FromSqlConversionFailure( + 1, + rusqlite::types::Type::Text, + Box::new(e), + ) + })?; + let pipeline_str: String = row.get(2)?; + let in_progress_at = row + .get::<_, Option>(5)? + .and_then(|s| parse_sqlite_datetime(&s)); + + Ok(ActiveSyncTask { + id: row.get(0)?, + task_type, + pipeline: parse_pipeline(&pipeline_str), + attempts: row.get(3)?, + last_error: row.get(4)?, + in_progress_at, + }) + }) + .map_err(|e| GarminError::Database(format!("Failed to query active tasks: {}", e)))?; + + let mut tasks = Vec::new(); + for row in rows { + tasks.push(row.map_err(|e| GarminError::Database(e.to_string()))?); + } + Ok(tasks) + } + /// Clean up old completed tasks pub fn cleanup_completed_tasks(&self, max_age_days: i32) -> Result { let count = self @@ -797,6 +882,17 @@ fn parse_status(s: &str) -> TaskStatus { } } +fn parse_sqlite_datetime(value: &str) -> Option> { + DateTime::parse_from_rfc3339(value) + .map(|dt| dt.with_timezone(&Utc)) + .ok() + .or_else(|| { + NaiveDateTime::parse_from_str(value, "%Y-%m-%d %H:%M:%S") + .ok() + .map(|dt| DateTime::::from_naive_utc_and_offset(dt, Utc)) + }) +} + #[cfg(test)] mod tests { use super::*; @@ -856,6 +952,29 @@ mod tests { assert!(next.is_none()); } + #[test] + fn test_list_in_progress_tasks_includes_started_at() { + let db = SyncDb::open_in_memory().unwrap(); + + let task = SyncTask::new( + 1, + SyncPipeline::Backfill, + SyncTaskType::DailyHealth { + date: NaiveDate::from_ymd_opt(2024, 12, 15).unwrap(), + }, + ); + + let id = db.push_task(&task).unwrap(); + db.mark_task_in_progress(id).unwrap(); + + let active = db.list_in_progress_tasks(1).unwrap(); + assert_eq!(active.len(), 1); + assert_eq!(active[0].id, id); + assert_eq!(active[0].pipeline, SyncPipeline::Backfill); + assert_eq!(active[0].attempts, 1); + assert!(active[0].in_progress_at.is_some()); + } + #[test] fn test_recover_in_progress() { let db = SyncDb::open_in_memory().unwrap(); diff --git a/crates/garmin-cli/src/sync/task_queue.rs b/crates/garmin-cli/src/sync/task_queue.rs index 4b30701..418c995 100644 --- a/crates/garmin-cli/src/sync/task_queue.rs +++ b/crates/garmin-cli/src/sync/task_queue.rs @@ -87,6 +87,20 @@ impl TaskQueue { self.sync_db.pop_task(self.profile_id, pipeline) } + /// Pop and mark the next round-robin task as in progress while holding the queue lock. + pub fn pop_round_robin_mark_in_progress( + &mut self, + pipeline: Option, + ) -> Result> { + let task = self.pop_round_robin_with_pipeline(pipeline)?; + if let Some(task) = &task { + if let Some(task_id) = task.id { + self.mark_in_progress(task_id)?; + } + } + Ok(task) + } + /// Mark a task as in progress pub fn mark_in_progress(&self, task_id: i64) -> Result<()> { self.sync_db.mark_task_in_progress(task_id) @@ -204,6 +218,15 @@ impl SharedTaskQueue { guard.pop_round_robin_with_pipeline(pipeline) } + /// Get and claim the next pending task using round-robin scheduling. + pub async fn pop_round_robin_mark_in_progress( + &self, + pipeline: Option, + ) -> Result> { + let mut guard = self.inner.lock().await; + guard.pop_round_robin_mark_in_progress(pipeline) + } + /// Add a task to the queue (thread-safe) pub async fn push(&self, task: SyncTask) -> Result { let guard = self.inner.lock().await; @@ -450,6 +473,31 @@ mod tests { assert!(matches!(first.task_type, SyncTaskType::Activities { .. })); } + #[test] + fn test_pop_round_robin_mark_in_progress_claims_task() { + let sync_db = SyncDb::open_in_memory().unwrap(); + let mut queue = TaskQueue::new(sync_db, 1, None); + + queue + .push(SyncTask::new( + 1, + SyncPipeline::Frontier, + SyncTaskType::DailyHealth { + date: NaiveDate::from_ymd_opt(2025, 1, 1).unwrap(), + }, + )) + .unwrap(); + + let claimed = queue.pop_round_robin_mark_in_progress(None).unwrap(); + assert!(claimed.is_some()); + + let next = queue.pop_round_robin_mark_in_progress(None).unwrap(); + assert!(next.is_none()); + + let (_pending, in_progress, _completed, _failed) = queue.count_by_status().unwrap(); + assert_eq!(in_progress, 1); + } + #[test] fn test_pop_with_pipeline_filters() { let sync_db = SyncDb::open_in_memory().unwrap(); From 73d5730a8aa61b6e4a8b1d9bf9eb3d2aa8413ab7 Mon Sep 17 00:00:00 2001 From: Han Xu Date: Sun, 5 Jul 2026 08:38:37 +0000 Subject: [PATCH 2/3] Fix atomic sync task claiming --- crates/garmin-cli/src/cli/commands/sync.rs | 104 ++++++- crates/garmin-cli/src/storage/sync_db.rs | 321 +++++++++++++++++---- crates/garmin-cli/src/sync/mod.rs | 15 +- crates/garmin-cli/src/sync/task_queue.rs | 61 +++- 4 files changed, 418 insertions(+), 83 deletions(-) diff --git a/crates/garmin-cli/src/cli/commands/sync.rs b/crates/garmin-cli/src/cli/commands/sync.rs index 10f6ab9..51a67a9 100644 --- a/crates/garmin-cli/src/cli/commands/sync.rs +++ b/crates/garmin-cli/src/cli/commands/sync.rs @@ -5,6 +5,7 @@ use std::path::Path; use crate::client::GarminClient; use crate::config::CredentialStore; +use crate::db::models::SyncTaskType; use crate::error::Result; use crate::storage::{default_storage_path, Storage, SyncDb}; use crate::sync::progress::SyncMode; @@ -122,12 +123,13 @@ pub async fn status(profile: Option, storage_path: Option) -> Re let performance_files = count_partition_files(&storage_path, "performance_metrics"); let track_files = count_partition_files(&storage_path, "track_points"); - // Get pending tasks - let pending_count = if let Some(pid) = profile_id { - sync_db.count_pending_tasks(pid, None)? - } else { - 0 - }; + // Get task status counts + let (pending_count, in_progress_count, completed_count, failed_count) = + if let Some(pid) = profile_id { + sync_db.count_tasks_by_status(pid)? + } else { + (0, 0, 0, 0) + }; println!("Storage: {}", storage_path.display()); println!("Profile: {}", profile_name); @@ -141,8 +143,33 @@ pub async fn status(profile: Option, storage_path: Option) -> Re println!(" Performance partitions: {:>4}", performance_files); println!(" Track point partitions: {:>4}", track_files); println!(); - if pending_count > 0 { - println!("Pending sync tasks: {}", pending_count); + if profile_id.is_some() { + println!("Sync tasks:"); + println!(" Pending: {:>4}", pending_count); + println!(" In progress: {:>4}", in_progress_count); + println!(" Failed: {:>4}", failed_count); + println!(" Completed: {:>4}", completed_count); + } + if let Some(pid) = profile_id { + if in_progress_count > 0 { + let active_tasks = sync_db.list_in_progress_tasks(pid)?; + println!(); + println!("Active tasks:"); + for task in active_tasks { + let started_at = task + .in_progress_at + .map(|dt| dt.to_rfc3339()) + .unwrap_or_else(|| "unknown".to_string()); + println!( + " #{} [{}] attempt {} started {} - {}", + task.id, + task.pipeline, + task.attempts, + started_at, + format_sync_task_type(&task.task_type) + ); + } + } } Ok(()) @@ -164,6 +191,49 @@ fn count_partition_files(storage_path: &Path, dirname: &str) -> usize { .unwrap_or(0) } +fn format_sync_task_type(task_type: &SyncTaskType) -> String { + match task_type { + SyncTaskType::Activities { + start, + limit, + min_date, + max_date, + } => { + let mut desc = format!("activities page start={} limit={}", start, limit); + if let Some(min_date) = min_date { + desc.push_str(&format!(" from={}", min_date)); + } + if let Some(max_date) = max_date { + desc.push_str(&format!(" to={}", max_date)); + } + desc + } + SyncTaskType::ActivityDetail { activity_id } => { + format!("activity detail activity_id={}", activity_id) + } + SyncTaskType::DownloadGpx { + activity_id, + activity_name, + activity_date, + } => { + let mut desc = format!("download GPX activity_id={}", activity_id); + if let Some(activity_date) = activity_date { + desc.push_str(&format!(" date={}", activity_date)); + } + if let Some(activity_name) = activity_name { + desc.push_str(&format!(" name=\"{}\"", activity_name)); + } + desc + } + SyncTaskType::DailyHealth { date } => format!("daily health date={}", date), + SyncTaskType::Performance { date } => format!("performance date={}", date), + SyncTaskType::Weight { from, to } => format!("weight from={} to={}", from, to), + SyncTaskType::GenerateEmbeddings { activity_ids } => { + format!("generate embeddings activities={}", activity_ids.len()) + } + } +} + /// Reset failed tasks to pending pub async fn reset(storage_path: Option) -> Result<()> { let storage_path = storage_path @@ -226,4 +296,22 @@ mod tests { assert_eq!(count_partition_files(temp_dir.path(), "activities"), 2); } + + #[test] + fn format_sync_task_type_describes_active_tasks_compactly() { + assert_eq!( + format_sync_task_type(&SyncTaskType::DailyHealth { + date: NaiveDate::from_ymd_opt(2026, 7, 5).unwrap() + }), + "daily health date=2026-07-05" + ); + assert_eq!( + format_sync_task_type(&SyncTaskType::DownloadGpx { + activity_id: 42, + activity_name: Some("Tempo Run".to_string()), + activity_date: Some("2026-07-04".to_string()), + }), + "download GPX activity_id=42 date=2026-07-04 name=\"Tempo Run\"" + ); + } } diff --git a/crates/garmin-cli/src/storage/sync_db.rs b/crates/garmin-cli/src/storage/sync_db.rs index e6542ee..b508cff 100644 --- a/crates/garmin-cli/src/storage/sync_db.rs +++ b/crates/garmin-cli/src/storage/sync_db.rs @@ -7,7 +7,7 @@ use std::path::Path; use chrono::{DateTime, NaiveDate, NaiveDateTime, Utc}; -use rusqlite::{params, Connection, OptionalExtension}; +use rusqlite::{params, Connection, OptionalExtension, Row}; use crate::db::models::{SyncPipeline, SyncState, SyncTask, SyncTaskType, TaskStatus}; use crate::error::{GarminError, Result}; @@ -458,34 +458,7 @@ impl SyncDb { }; self.conn - .query_row(query, params, |row| { - let task_data: String = row.get(3)?; - let task_type: SyncTaskType = serde_json::from_str(&task_data).unwrap(); - let pipeline_str: String = row.get(4)?; - let status_str: String = row.get(5)?; - - Ok(SyncTask { - id: Some(row.get(0)?), - profile_id: row.get(1)?, - task_type, - pipeline: parse_pipeline(&pipeline_str), - status: parse_status(&status_str), - attempts: row.get(6)?, - last_error: row.get(7)?, - created_at: row - .get::<_, Option>(8)? - .and_then(|s| DateTime::parse_from_rfc3339(&s).ok()) - .map(|dt| dt.with_timezone(&Utc)), - next_retry_at: row - .get::<_, Option>(9)? - .and_then(|s| DateTime::parse_from_rfc3339(&s).ok()) - .map(|dt| dt.with_timezone(&Utc)), - completed_at: row - .get::<_, Option>(10)? - .and_then(|s| DateTime::parse_from_rfc3339(&s).ok()) - .map(|dt| dt.with_timezone(&Utc)), - }) - }) + .query_row(query, params, sync_task_from_row) .optional() .map_err(|e| GarminError::Database(format!("Failed to pop task: {}", e))) } @@ -531,54 +504,106 @@ impl SyncDb { }; self.conn - .query_row(query, params, |row| { - let task_data: String = row.get(3)?; - let task_type: SyncTaskType = serde_json::from_str(&task_data).unwrap(); - let pipeline_str: String = row.get(4)?; - let status_str: String = row.get(5)?; - - Ok(SyncTask { - id: Some(row.get(0)?), - profile_id: row.get(1)?, - task_type, - pipeline: parse_pipeline(&pipeline_str), - status: parse_status(&status_str), - attempts: row.get(6)?, - last_error: row.get(7)?, - created_at: row - .get::<_, Option>(8)? - .and_then(|s| DateTime::parse_from_rfc3339(&s).ok()) - .map(|dt| dt.with_timezone(&Utc)), - next_retry_at: row - .get::<_, Option>(9)? - .and_then(|s| DateTime::parse_from_rfc3339(&s).ok()) - .map(|dt| dt.with_timezone(&Utc)), - completed_at: row - .get::<_, Option>(10)? - .and_then(|s| DateTime::parse_from_rfc3339(&s).ok()) - .map(|dt| dt.with_timezone(&Utc)), - }) - }) + .query_row(query, params, sync_task_from_row) .optional() .map_err(|e| GarminError::Database(format!("Failed to pop task by type: {}", e))) } + /// Atomically claim the next eligible task for a profile. + pub fn claim_next_task( + &self, + profile_id: i32, + pipeline: Option, + ) -> Result> { + let (query, params) = if let Some(pipeline) = pipeline { + ( + claim_task_query( + "profile_id = ? + AND pipeline = ? + AND status IN ('pending', 'failed') + AND (next_retry_at IS NULL OR next_retry_at <= datetime('now'))", + ), + params![profile_id, pipeline_name(pipeline)], + ) + } else { + ( + claim_task_query( + "profile_id = ? + AND status IN ('pending', 'failed') + AND (next_retry_at IS NULL OR next_retry_at <= datetime('now'))", + ), + params![profile_id], + ) + }; + + self.conn + .query_row(&query, params, sync_task_from_row) + .optional() + .map_err(|e| GarminError::Database(format!("Failed to claim task: {}", e))) + } + + /// Atomically claim the next eligible task for a profile and task type. + pub fn claim_next_task_by_type( + &self, + profile_id: i32, + task_type: &str, + pipeline: Option, + ) -> Result> { + let (query, params) = if let Some(pipeline) = pipeline { + ( + claim_task_query( + "profile_id = ? + AND task_type = ? + AND pipeline = ? + AND status IN ('pending', 'failed') + AND (next_retry_at IS NULL OR next_retry_at <= datetime('now'))", + ), + params![profile_id, task_type, pipeline_name(pipeline)], + ) + } else { + ( + claim_task_query( + "profile_id = ? + AND task_type = ? + AND status IN ('pending', 'failed') + AND (next_retry_at IS NULL OR next_retry_at <= datetime('now'))", + ), + params![profile_id, task_type], + ) + }; + + self.conn + .query_row(&query, params, sync_task_from_row) + .optional() + .map_err(|e| GarminError::Database(format!("Failed to claim task by type: {}", e))) + } + /// Mark a task as in progress pub fn mark_task_in_progress(&self, task_id: i64) -> Result<()> { - self.conn + let count = self + .conn .execute( "UPDATE sync_tasks SET status = 'in_progress', attempts = attempts + 1, in_progress_at = datetime('now'), last_error = NULL - WHERE id = ?", + WHERE id = ? + AND status IN ('pending', 'failed') + AND (next_retry_at IS NULL OR next_retry_at <= datetime('now'))", params![task_id], ) .map_err(|e| { GarminError::Database(format!("Failed to mark task in progress: {}", e)) })?; + if count == 0 { + return Err(GarminError::Database(format!( + "Task {} is not eligible to be marked in progress", + task_id + ))); + } + Ok(()) } @@ -882,6 +907,55 @@ fn parse_status(s: &str) -> TaskStatus { } } +fn claim_task_query(where_clause: &str) -> String { + format!( + "UPDATE sync_tasks + SET status = 'in_progress', + attempts = attempts + 1, + last_error = NULL, + in_progress_at = datetime('now') + WHERE id = ( + SELECT id + FROM sync_tasks + WHERE {where_clause} + ORDER BY + CASE WHEN status = 'failed' THEN 0 ELSE 1 END, + created_at ASC + LIMIT 1 + ) + RETURNING id, profile_id, task_type, task_data, pipeline, status, attempts, last_error, + created_at, next_retry_at, completed_at" + ) +} + +fn sync_task_from_row(row: &Row<'_>) -> rusqlite::Result { + let task_data: String = row.get(3)?; + let task_type: SyncTaskType = serde_json::from_str(&task_data).map_err(|e| { + rusqlite::Error::FromSqlConversionFailure(3, rusqlite::types::Type::Text, Box::new(e)) + })?; + let pipeline_str: String = row.get(4)?; + let status_str: String = row.get(5)?; + + Ok(SyncTask { + id: Some(row.get(0)?), + profile_id: row.get(1)?, + task_type, + pipeline: parse_pipeline(&pipeline_str), + status: parse_status(&status_str), + attempts: row.get(6)?, + last_error: row.get(7)?, + created_at: row + .get::<_, Option>(8)? + .and_then(|s| parse_sqlite_datetime(&s)), + next_retry_at: row + .get::<_, Option>(9)? + .and_then(|s| parse_sqlite_datetime(&s)), + completed_at: row + .get::<_, Option>(10)? + .and_then(|s| parse_sqlite_datetime(&s)), + }) +} + fn parse_sqlite_datetime(value: &str) -> Option> { DateTime::parse_from_rfc3339(value) .map(|dt| dt.with_timezone(&Utc)) @@ -896,6 +970,7 @@ fn parse_sqlite_datetime(value: &str) -> Option> { #[cfg(test)] mod tests { use super::*; + use std::sync::{Arc, Barrier}; #[test] fn test_profile_management() { @@ -1136,4 +1211,132 @@ mod tests { let (_activities, _gpx, health, _perf) = db.count_tasks_by_type(1, None).unwrap(); assert_eq!(health, 1); } + + #[test] + fn test_claim_next_task_marks_in_progress_once() { + let db = SyncDb::open_in_memory().unwrap(); + + let task = SyncTask::new( + 1, + SyncPipeline::Frontier, + SyncTaskType::DailyHealth { + date: NaiveDate::from_ymd_opt(2026, 7, 5).unwrap(), + }, + ); + let id = db.push_task(&task).unwrap(); + + let claimed = db.claim_next_task(1, None).unwrap().unwrap(); + assert_eq!(claimed.id, Some(id)); + assert_eq!(claimed.status, TaskStatus::InProgress); + assert_eq!(claimed.attempts, 1); + assert!(claimed.last_error.is_none()); + + let next = db.claim_next_task(1, None).unwrap(); + assert!(next.is_none()); + + let active = db.list_in_progress_tasks(1).unwrap(); + assert_eq!(active.len(), 1); + assert_eq!(active[0].attempts, 1); + } + + #[test] + fn test_claim_next_task_by_type_respects_pipeline() { + let db = SyncDb::open_in_memory().unwrap(); + + let frontier = SyncTask::new( + 1, + SyncPipeline::Frontier, + SyncTaskType::DailyHealth { + date: NaiveDate::from_ymd_opt(2026, 7, 5).unwrap(), + }, + ); + let backfill = SyncTask::new( + 1, + SyncPipeline::Backfill, + SyncTaskType::DailyHealth { + date: NaiveDate::from_ymd_opt(2026, 7, 4).unwrap(), + }, + ); + + let frontier_id = db.push_task(&frontier).unwrap(); + let backfill_id = db.push_task(&backfill).unwrap(); + + let claimed_backfill = db + .claim_next_task_by_type(1, "daily_health", Some(SyncPipeline::Backfill)) + .unwrap() + .unwrap(); + assert_eq!(claimed_backfill.id, Some(backfill_id)); + assert_eq!(claimed_backfill.pipeline, SyncPipeline::Backfill); + + let claimed_frontier = db + .claim_next_task_by_type(1, "daily_health", Some(SyncPipeline::Frontier)) + .unwrap() + .unwrap(); + assert_eq!(claimed_frontier.id, Some(frontier_id)); + assert_eq!(claimed_frontier.pipeline, SyncPipeline::Frontier); + } + + #[test] + fn test_claim_next_task_single_winner_across_connections() { + let temp_dir = tempfile::tempdir().unwrap(); + let db_path = temp_dir.path().join("sync.db"); + + let db = SyncDb::open(&db_path).unwrap(); + db.push_task(&SyncTask::new( + 1, + SyncPipeline::Frontier, + SyncTaskType::DailyHealth { + date: NaiveDate::from_ymd_opt(2026, 7, 5).unwrap(), + }, + )) + .unwrap(); + drop(db); + + let workers = 8; + let barrier = Arc::new(Barrier::new(workers)); + let mut handles = Vec::new(); + for _ in 0..workers { + let db_path = db_path.clone(); + let barrier = Arc::clone(&barrier); + handles.push(std::thread::spawn(move || { + let db = SyncDb::open(&db_path).unwrap(); + barrier.wait(); + db.claim_next_task(1, None) + .unwrap() + .map(|task| task.id.unwrap()) + })); + } + + let claimed: Vec = handles + .into_iter() + .filter_map(|handle| handle.join().unwrap()) + .collect(); + assert_eq!(claimed.len(), 1); + + let db = SyncDb::open(&db_path).unwrap(); + let active = db.list_in_progress_tasks(1).unwrap(); + assert_eq!(active.len(), 1); + assert_eq!(active[0].attempts, 1); + } + + #[test] + fn test_mark_task_in_progress_rejects_already_claimed_task() { + 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(); + + db.mark_task_in_progress(id).unwrap(); + assert!(db.mark_task_in_progress(id).is_err()); + + let active = db.list_in_progress_tasks(1).unwrap(); + assert_eq!(active[0].attempts, 1); + } } diff --git a/crates/garmin-cli/src/sync/mod.rs b/crates/garmin-cli/src/sync/mod.rs index 45dd5a1..4d32fed 100644 --- a/crates/garmin-cli/src/sync/mod.rs +++ b/crates/garmin-cli/src/sync/mod.rs @@ -1277,11 +1277,10 @@ async fn producer_loop( let mut empty_count = 0; loop { - // Pop next task - let next_task = match context.pipeline_filter { - Some(pipeline) => queue.pop_round_robin_with_pipeline(Some(pipeline)).await, - None => queue.pop_round_robin().await, - }; + // Claim next task + let next_task = queue + .pop_round_robin_mark_in_progress(context.pipeline_filter) + .await; let task = match next_task { Ok(Some(task)) => { @@ -1307,12 +1306,6 @@ async fn producer_loop( let task_id = task.id.unwrap(); context.in_flight.fetch_add(1, Ordering::Relaxed); - // Mark in progress - if let Err(e) = queue.mark_in_progress(task_id).await { - eprintln!("Failed to mark task in progress: {}", e); - continue; - } - // Update progress display update_progress_for_task(&task, &context.progress); diff --git a/crates/garmin-cli/src/sync/task_queue.rs b/crates/garmin-cli/src/sync/task_queue.rs index 418c995..b017310 100644 --- a/crates/garmin-cli/src/sync/task_queue.rs +++ b/crates/garmin-cli/src/sync/task_queue.rs @@ -92,13 +92,20 @@ impl TaskQueue { &mut self, pipeline: Option, ) -> Result> { - let task = self.pop_round_robin_with_pipeline(pipeline)?; - if let Some(task) = &task { - if let Some(task_id) = task.id { - self.mark_in_progress(task_id)?; + const TASK_TYPES: [&str; 4] = ["activities", "download_gpx", "performance", "daily_health"]; + + for _ in 0..TASK_TYPES.len() { + let idx = self.rr_index % TASK_TYPES.len(); + self.rr_index = self.rr_index.wrapping_add(1); + if let Some(task) = + self.sync_db + .claim_next_task_by_type(self.profile_id, TASK_TYPES[idx], pipeline)? + { + return Ok(Some(task)); } } - Ok(task) + + self.sync_db.claim_next_task(self.profile_id, pipeline) } /// Mark a task as in progress @@ -490,6 +497,9 @@ mod tests { let claimed = queue.pop_round_robin_mark_in_progress(None).unwrap(); assert!(claimed.is_some()); + let claimed = claimed.unwrap(); + assert_eq!(claimed.status, crate::db::models::TaskStatus::InProgress); + assert_eq!(claimed.attempts, 1); let next = queue.pop_round_robin_mark_in_progress(None).unwrap(); assert!(next.is_none()); @@ -498,6 +508,47 @@ mod tests { assert_eq!(in_progress, 1); } + #[test] + fn test_pop_round_robin_mark_in_progress_respects_pipeline_filter() { + let sync_db = SyncDb::open_in_memory().unwrap(); + let mut queue = TaskQueue::new(sync_db, 1, None); + + let frontier_task = SyncTask::new( + 1, + SyncPipeline::Frontier, + SyncTaskType::DailyHealth { + date: NaiveDate::from_ymd_opt(2025, 1, 1).unwrap(), + }, + ); + let backfill_task = SyncTask::new( + 1, + SyncPipeline::Backfill, + SyncTaskType::DailyHealth { + date: NaiveDate::from_ymd_opt(2025, 1, 2).unwrap(), + }, + ); + + let frontier_id = queue.push(frontier_task).unwrap(); + let backfill_id = queue.push(backfill_task).unwrap(); + + let claimed_backfill = queue + .pop_round_robin_mark_in_progress(Some(SyncPipeline::Backfill)) + .unwrap() + .unwrap(); + assert_eq!(claimed_backfill.id, Some(backfill_id)); + assert_eq!(claimed_backfill.pipeline, SyncPipeline::Backfill); + + let claimed_frontier = queue + .pop_round_robin_mark_in_progress(Some(SyncPipeline::Frontier)) + .unwrap() + .unwrap(); + assert_eq!(claimed_frontier.id, Some(frontier_id)); + assert_eq!(claimed_frontier.pipeline, SyncPipeline::Frontier); + + let next = queue.pop_round_robin_mark_in_progress(None).unwrap(); + assert!(next.is_none()); + } + #[test] fn test_pop_with_pipeline_filters() { let sync_db = SyncDb::open_in_memory().unwrap(); From ae6947b8e63718f9706cc662c9cd5a15a017d25f Mon Sep 17 00:00:00 2001 From: Han Xu Date: Sun, 5 Jul 2026 08:55:27 +0000 Subject: [PATCH 3/3] Recover only stale sync task claims --- crates/garmin-cli/src/storage/sync_db.rs | 88 +++++++++++++++++++++++- crates/garmin-cli/src/sync/task_queue.rs | 4 ++ 2 files changed, 89 insertions(+), 3 deletions(-) diff --git a/crates/garmin-cli/src/storage/sync_db.rs b/crates/garmin-cli/src/storage/sync_db.rs index b508cff..884eeb4 100644 --- a/crates/garmin-cli/src/storage/sync_db.rs +++ b/crates/garmin-cli/src/storage/sync_db.rs @@ -12,6 +12,8 @@ use rusqlite::{params, Connection, OptionalExtension, Row}; use crate::db::models::{SyncPipeline, SyncState, SyncTask, SyncTaskType, TaskStatus}; use crate::error::{GarminError, Result}; +const IN_PROGRESS_RECOVERY_AFTER_SECS: i64 = 30 * 60; + /// Task currently marked in progress in the sync queue. #[derive(Debug, Clone)] pub struct ActiveSyncTask { @@ -648,8 +650,12 @@ impl SyncDb { "UPDATE sync_tasks SET status = 'pending', in_progress_at = NULL - WHERE status = 'in_progress'", - [], + WHERE status = 'in_progress' + AND ( + in_progress_at IS NULL + OR in_progress_at <= datetime('now', '-' || ? || ' seconds') + )", + params![IN_PROGRESS_RECOVERY_AFTER_SECS], ) .map_err(|e| GarminError::Database(format!("Failed to recover tasks: {}", e)))?; @@ -814,6 +820,34 @@ impl SyncDb { Ok(tasks) } + #[cfg(test)] + pub(crate) fn set_in_progress_at_seconds_ago_for_test( + &self, + task_id: i64, + seconds_ago: i64, + ) -> Result<()> { + self.conn + .execute( + "UPDATE sync_tasks + SET in_progress_at = datetime('now', '-' || ? || ' seconds') + WHERE id = ?", + params![seconds_ago, task_id], + ) + .map_err(|e| GarminError::Database(format!("Failed to update test task: {}", e)))?; + Ok(()) + } + + #[cfg(test)] + pub(crate) fn clear_in_progress_at_for_test(&self, task_id: i64) -> Result<()> { + self.conn + .execute( + "UPDATE sync_tasks SET in_progress_at = NULL WHERE id = ?", + params![task_id], + ) + .map_err(|e| GarminError::Database(format!("Failed to update test task: {}", e)))?; + Ok(()) + } + /// Clean up old completed tasks pub fn cleanup_completed_tasks(&self, max_age_days: i32) -> Result { let count = self @@ -1051,7 +1085,7 @@ mod tests { } #[test] - fn test_recover_in_progress() { + fn test_recover_in_progress_recovers_stale_claims() { let db = SyncDb::open_in_memory().unwrap(); let task = SyncTask::new( @@ -1064,6 +1098,8 @@ mod tests { let id = db.push_task(&task).unwrap(); db.mark_task_in_progress(id).unwrap(); + db.set_in_progress_at_seconds_ago_for_test(id, IN_PROGRESS_RECOVERY_AFTER_SECS + 60) + .unwrap(); // Simulate crash recovery let recovered = db.recover_in_progress_tasks().unwrap(); @@ -1074,6 +1110,52 @@ mod tests { assert!(popped.is_some()); } + #[test] + fn test_recover_in_progress_preserves_fresh_claims() { + let db = SyncDb::open_in_memory().unwrap(); + + let id = db + .push_task(&SyncTask::new( + 1, + SyncPipeline::Frontier, + SyncTaskType::DailyHealth { + date: NaiveDate::from_ymd_opt(2024, 12, 15).unwrap(), + }, + )) + .unwrap(); + db.mark_task_in_progress(id).unwrap(); + + let recovered = db.recover_in_progress_tasks().unwrap(); + assert_eq!(recovered, 0); + + let active = db.list_in_progress_tasks(1).unwrap(); + assert_eq!(active.len(), 1); + assert_eq!(active[0].id, id); + } + + #[test] + fn test_recover_in_progress_recovers_legacy_claims_without_timestamp() { + let db = SyncDb::open_in_memory().unwrap(); + + let id = db + .push_task(&SyncTask::new( + 1, + SyncPipeline::Frontier, + SyncTaskType::DailyHealth { + date: NaiveDate::from_ymd_opt(2024, 12, 15).unwrap(), + }, + )) + .unwrap(); + db.mark_task_in_progress(id).unwrap(); + db.clear_in_progress_at_for_test(id).unwrap(); + + let recovered = db.recover_in_progress_tasks().unwrap(); + assert_eq!(recovered, 1); + + let popped = db.pop_task(1, None).unwrap().unwrap(); + assert_eq!(popped.id, Some(id)); + } + #[test] fn test_pop_task_scoped_by_profile() { let db = SyncDb::open_in_memory().unwrap(); diff --git a/crates/garmin-cli/src/sync/task_queue.rs b/crates/garmin-cli/src/sync/task_queue.rs index b017310..47b14ea 100644 --- a/crates/garmin-cli/src/sync/task_queue.rs +++ b/crates/garmin-cli/src/sync/task_queue.rs @@ -418,6 +418,10 @@ mod tests { ); let id = queue.push(task).unwrap(); queue.mark_in_progress(id).unwrap(); + queue + .sync_db + .set_in_progress_at_seconds_ago_for_test(id, 31 * 60) + .unwrap(); // Simulate crash recovery let recovered = queue.recover_in_progress().unwrap();