diff --git a/crates/garmin-cli/src/cli/commands/sync.rs b/crates/garmin-cli/src/cli/commands/sync.rs index 51a67a9..f79578e 100644 --- a/crates/garmin-cli/src/cli/commands/sync.rs +++ b/crates/garmin-cli/src/cli/commands/sync.rs @@ -1,12 +1,12 @@ //! Sync commands for garmin-cli -use chrono::NaiveDate; -use std::path::Path; +use chrono::{NaiveDate, Utc}; +use std::path::{Path, PathBuf}; -use crate::client::GarminClient; +use crate::client::{GarminClient, RequestLogger}; use crate::config::CredentialStore; use crate::db::models::SyncTaskType; -use crate::error::Result; +use crate::error::{GarminError, Result}; use crate::storage::{default_storage_path, Storage, SyncDb}; use crate::sync::progress::SyncMode; use crate::sync::{SyncEngine, SyncOptions, TaskQueue}; @@ -26,6 +26,8 @@ pub async fn run( dry_run: bool, backfill: bool, force: bool, + log: Option, + verbose: bool, ) -> Result<()> { let store = CredentialStore::new(profile.clone())?; let (_, oauth2) = refresh_token(&store).await?; @@ -56,10 +58,18 @@ pub async fn run( force, concurrency: 4, mode, + logger: None, }; + let (logger, log_path) = create_sync_logger(log, verbose)?; + if let Some(path) = &log_path { + println!("Writing sync log: {}", path.display()); + } + let mut opts = opts; + opts.logger = logger.clone(); + // Create sync engine - let client = GarminClient::new(); + let client = GarminClient::new().with_logger(logger); let mut engine = SyncEngine::with_storage(storage, client, oauth2)?; engine.run(opts).await?; @@ -124,12 +134,16 @@ pub async fn status(profile: Option, storage_path: Option) -> Re let track_files = count_partition_files(&storage_path, "track_points"); // 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) - }; + let (pending, in_progress, completed, failed) = if let Some(pid) = profile_id { + sync_db.count_tasks_by_status(pid)? + } else { + (0, 0, 0, 0) + }; + let active_tasks = if let Some(pid) = profile_id { + sync_db.list_in_progress_tasks(pid)? + } else { + Vec::new() + }; println!("Storage: {}", storage_path.display()); println!("Profile: {}", profile_name); @@ -145,30 +159,37 @@ pub async fn status(profile: Option, storage_path: Option) -> Re println!(); 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); + println!(" Pending: {:>4}", pending); + println!(" In progress: {:>4}", in_progress); + println!(" Failed: {:>4}", failed); + println!(" Completed: {:>4}", completed); } - 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) - ); - } + + if !active_tasks.is_empty() { + println!(); + println!("Active tasks:"); + for task in active_tasks { + let elapsed = task + .in_progress_at + .map(|started| { + let seconds = (Utc::now() - started).num_seconds().max(0) as u64; + format_elapsed(seconds) + }) + .unwrap_or_else(|| "unknown".to_string()); + let last_error = task + .last_error + .as_deref() + .map(|error| format!("; last error: {}", error)) + .unwrap_or_default(); + println!( + " #{} [{}] {} for {} (attempt {}{})", + task.id, + task.pipeline, + task_description(&task.task_type), + elapsed, + task.attempts, + last_error + ); } } @@ -191,49 +212,103 @@ fn count_partition_files(storage_path: &Path, dirname: &str) -> usize { .unwrap_or(0) } -fn format_sync_task_type(task_type: &SyncTaskType) -> String { +fn task_description(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::Activities { start, limit, .. } => { + format!("Activities {}-{}", start, start + limit) } 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 + let name = activity_name + .as_deref() + .map(str::to_string) + .unwrap_or_else(|| activity_id.to_string()); + activity_date + .as_deref() + .map(|date| format!("GPX {} {}", date, name)) + .unwrap_or_else(|| format!("GPX {}", name)) } - 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::DailyHealth { date } => format!("Health {}", date), + SyncTaskType::Performance { date } => format!("Performance {}", date), + SyncTaskType::ActivityDetail { activity_id } => format!("Activity detail {}", activity_id), + SyncTaskType::Weight { from, to } => format!("Weight {} -> {}", from, to), SyncTaskType::GenerateEmbeddings { activity_ids } => { - format!("generate embeddings activities={}", activity_ids.len()) + format!("Embeddings for {} activities", activity_ids.len()) } } } +fn format_elapsed(secs: u64) -> String { + let mins = secs / 60; + let remaining_secs = secs % 60; + + if mins > 0 { + format!("{}m {}s", mins, remaining_secs) + } else { + format!("{}s", secs) + } +} + +fn create_sync_logger( + log: Option, + verbose: bool, +) -> Result<(Option, Option)> { + let Some(path) = resolve_log_path(log, verbose)? else { + return Ok((None, None)); + }; + + if let Some(parent) = path.parent() { + std::fs::create_dir_all(parent).map_err(|e| { + GarminError::Io(std::io::Error::new( + e.kind(), + format!("Failed to create log directory {}: {}", parent.display(), e), + )) + })?; + } + + let file = std::fs::OpenOptions::new() + .create(true) + .append(true) + .open(&path) + .map_err(|e| { + GarminError::Io(std::io::Error::new( + e.kind(), + format!("Failed to open log file {}: {}", path.display(), e), + )) + })?; + + Ok((Some(RequestLogger::new(file, verbose)), Some(path))) +} + +fn resolve_log_path(log: Option, verbose: bool) -> Result> { + let Some(path) = log.map(PathBuf::from).or_else(|| { + verbose.then(|| { + std::env::current_dir() + .unwrap_or_else(|_| PathBuf::from(".")) + .join(default_log_filename()) + }) + }) else { + return Ok(None); + }; + + if path.extension().and_then(|ext| ext.to_str()) == Some("log") { + return Ok(Some(path)); + } + + if path.exists() && !path.is_dir() { + return Ok(Some(path)); + } + + Ok(Some(path.join(default_log_filename()))) +} + +fn default_log_filename() -> String { + format!("garmin-sync-{}.log", Utc::now().format("%Y%m%dT%H%M%SZ")) +} + /// Reset failed tasks to pending pub async fn reset(storage_path: Option) -> Result<()> { let storage_path = storage_path @@ -298,20 +373,20 @@ mod tests { } #[test] - fn format_sync_task_type_describes_active_tasks_compactly() { + fn task_description_describes_active_tasks_compactly() { assert_eq!( - format_sync_task_type(&SyncTaskType::DailyHealth { + task_description(&SyncTaskType::DailyHealth { date: NaiveDate::from_ymd_opt(2026, 7, 5).unwrap() }), - "daily health date=2026-07-05" + "Health 2026-07-05" ); assert_eq!( - format_sync_task_type(&SyncTaskType::DownloadGpx { + task_description(&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\"" + "GPX 2026-07-04 Tempo Run" ); } } diff --git a/crates/garmin-cli/src/main.rs b/crates/garmin-cli/src/main.rs index 765335e..ab2287c 100644 --- a/crates/garmin-cli/src/main.rs +++ b/crates/garmin-cli/src/main.rs @@ -356,6 +356,12 @@ enum SyncCommands { /// Force re-sync (ignore existing data) #[arg(long)] force: bool, + /// Directory or .log file path for detailed sync logs + #[arg(long)] + log: Option, + /// Include full JSON/text responses in the sync log + #[arg(long)] + verbose: bool, }, /// Show sync status Status { @@ -504,6 +510,8 @@ async fn main() -> garmin_cli::Result<()> { dry_run, backfill, force, + log, + verbose, } => { commands::sync_run( cli.profile, @@ -516,6 +524,8 @@ async fn main() -> garmin_cli::Result<()> { dry_run, backfill, force, + log, + verbose, ) .await } diff --git a/crates/garmin-cli/src/sync/mod.rs b/crates/garmin-cli/src/sync/mod.rs index bbf8bf5..864645e 100644 --- a/crates/garmin-cli/src/sync/mod.rs +++ b/crates/garmin-cli/src/sync/mod.rs @@ -19,13 +19,13 @@ use std::sync::Arc; use chrono::{DateTime, Duration, NaiveDate, NaiveDateTime, Utc}; use tokio::sync::{mpsc, Mutex as TokioMutex}; -use crate::client::{GarminClient, OAuth2Token}; +use crate::client::{GarminClient, OAuth2Token, RequestLogger}; use crate::db::models::{ Activity, DailyHealth, PerformanceMetrics, SyncPipeline, SyncTask, SyncTaskType, TrackPoint, }; use crate::storage::{ParquetStore, Storage, SyncDb}; use crate::{GarminError, Result}; -use std::io::{self, Write}; +use std::io::{self, IsTerminal, Write}; pub use progress::{PlanningStep, SharedProgress, SyncProgress}; pub use rate_limiter::{RateLimiter, SharedRateLimiter}; @@ -418,8 +418,15 @@ impl SyncEngine { // Set storage path and sync mode for display progress.set_storage_path(&self.storage.base_path().display().to_string()); progress.set_sync_mode(opts.mode); + progress.set_logger(opts.logger.clone()); + progress.set_worker_slots(opts.concurrency); println!("Using storage: {}", progress.get_storage_path()); println!("Planning sync..."); + progress.log(format!( + "Sync starting: storage={}, mode={}", + progress.get_storage_path(), + opts.mode + )); // Planning phase - updates progress instead of printing progress.set_planning_step(PlanningStep::FetchingProfile); @@ -468,6 +475,7 @@ impl SyncEngine { print_sync_overview(&progress); println!("Planning complete."); + progress.log("Planning complete."); let stats_result = if opts.dry_run { println!("Dry run mode - no changes will be made"); @@ -496,6 +504,7 @@ impl SyncEngine { if !opts.dry_run { println!("\nSync complete: {}", stats); + progress.log(format!("Sync complete: {}", stats)); } Ok(stats) } @@ -651,14 +660,14 @@ impl SyncEngine { // Wait for all producers to finish for h in producer_handles { if let Err(e) = h.await { - eprintln!("Producer error: {}", e); + progress.log(format!("Producer task join error: {}", e)); } } // Wait for all consumers to finish for h in consumer_handles { if let Err(e) = h.await { - eprintln!("Consumer error: {}", e); + progress.log(format!("Consumer task join error: {}", e)); } } @@ -1045,7 +1054,8 @@ fn print_sync_errors(progress: &SyncProgress) { } println!("\nRecent errors:"); - for error in errors.iter().take(5) { + let start = errors.len().saturating_sub(5); + for error in errors.iter().skip(start) { println!(" [{}] {}: {}", error.stream, error.item, error.error); } @@ -1055,11 +1065,28 @@ fn print_sync_errors(progress: &SyncProgress) { } async fn run_progress_reporter(progress: SharedProgress) { + let mut rendered_lines = 0usize; + let interactive = std::io::stdout().is_terminal(); + loop { - progress.print_simple_status(); + let lines = progress.status_lines(); + + if interactive { + if rendered_lines > 0 { + print!("\x1b[{}A", rendered_lines); + } + + for line in &lines { + print!("\x1b[2K{}\n", line); + } + rendered_lines = lines.len(); + } else { + println!("{}", lines.join("\n")); + } + + let _ = std::io::Write::flush(&mut std::io::stdout()); if progress.should_shutdown() || progress.is_complete() { - println!(); break; } @@ -1088,6 +1115,8 @@ pub struct SyncOptions { pub concurrency: usize, /// Sync mode (Latest or Backfill) pub mode: progress::SyncMode, + /// Optional detailed log writer + pub logger: Option, } impl Default for SyncOptions { @@ -1102,6 +1131,7 @@ impl Default for SyncOptions { force: false, concurrency: 4, mode: progress::SyncMode::Latest, + logger: None, } } } @@ -1241,6 +1271,7 @@ fn fail_progress_for_task(task: &SyncTask, progress: &SyncProgress, error: &str) } const MAX_IDLE_RETRIES: u32 = 10; +const TASK_TIMEOUT_SECS: u64 = 300; fn should_exit_when_idle(idle_loops: u32, in_flight: usize) -> bool { idle_loops >= MAX_IDLE_RETRIES && in_flight == 0 @@ -1281,13 +1312,69 @@ fn record_write_failure(data: &SyncData, progress: &SyncProgress, error: &str) { } } +fn task_progress_metadata(task: &SyncTask) -> Option<(&'static str, String)> { + match &task.task_type { + SyncTaskType::Activities { start, limit, .. } => Some(( + "Activities", + format!("Activities {}-{}", start, start + limit), + )), + SyncTaskType::DownloadGpx { + activity_id, + activity_name, + activity_date, + } => { + let name = activity_name + .as_deref() + .map(str::to_string) + .unwrap_or_else(|| activity_id.to_string()); + let item = activity_date + .as_deref() + .map(|date| format!("{} {}", date, name)) + .unwrap_or(name); + Some(("GPX", item)) + } + SyncTaskType::DailyHealth { date } => Some(("Health", date.to_string())), + SyncTaskType::Performance { date } => Some(("Performance", date.to_string())), + _ => None, + } +} + +fn sync_data_progress_metadata(data: &SyncData) -> (&'static str, String, i64) { + match data { + SyncData::Activities { + records, task_id, .. + } => { + let item = match (records.first(), records.last()) { + (Some(first), Some(last)) if first.activity_id != last.activity_id => { + format!("{}..{}", first.activity_id, last.activity_id) + } + (Some(first), _) => first.activity_id.to_string(), + _ => "empty page".to_string(), + }; + ("Activities", item, *task_id) + } + SyncData::Health { + record, task_id, .. + } => ("Health", record.date.to_string(), *task_id), + SyncData::Performance { + record, task_id, .. + } => ("Performance", record.date.to_string(), *task_id), + SyncData::TrackPoints { + activity_id, + date, + task_id, + .. + } => ("GPX", format!("{} ({})", date, activity_id), *task_id), + } +} + // ============================================================================= // Producer/Consumer Pipeline // ============================================================================= /// Producer loop: fetches data from Garmin API and sends to channel async fn producer_loop( - _id: usize, + id: usize, queue: SharedTaskQueue, tx: mpsc::Sender, context: ProducerContext, @@ -1316,7 +1403,9 @@ async fn producer_loop( continue; } Err(e) => { - eprintln!("Queue error: {}", e); + context + .progress + .log(format!("Producer P{} queue error: {}", id, e)); break; } }; @@ -1327,6 +1416,16 @@ async fn producer_loop( // Update progress display update_progress_for_task(&task, &context.progress); + if let Some((stream, item)) = task_progress_metadata(&task) { + context.progress.set_worker_task( + "P", + id, + Some(task_id), + stream, + item, + "checking cache", + ); + } // Skip tasks that already exist unless forcing if !context.force { @@ -1353,76 +1452,154 @@ async fn producer_loop( if should_skip { if let Err(e) = queue.mark_completed(task_id, claim_attempt).await { - eprintln!("Failed to mark task completed: {}", e); + context.progress.log(format!( + "Producer P{} failed to mark skipped task #{} completed: {}", + id, task_id, e + )); } complete_progress_for_task(&task, &context.progress); { let mut s = context.stats.lock().await; s.completed += 1; } + context.progress.clear_worker("P", id); context.in_flight.fetch_sub(1, Ordering::Relaxed); continue; } } // Acquire rate limiter permit + context + .progress + .update_worker_status("P", id, "waiting for rate limit"); let _permit = context.rate_limiter.acquire().await; context.progress.record_request(); + context.progress.update_worker_status("P", id, "fetching"); // Fetch data based on task type - let result = fetch_task_data( - &task, - &context.client, - &context.token, - &context.display_name, - context.profile_id, + let result = tokio::time::timeout( + std::time::Duration::from_secs(TASK_TIMEOUT_SECS), + fetch_task_data( + &task, + &context.client, + &context.token, + &context.display_name, + context.profile_id, + ), ) .await; match result { - Ok(data) => { + Ok(Ok(data)) => { context.rate_limiter.on_success(); + context + .progress + .update_worker_status("P", id, "waiting for writer"); // Send to consumer if tx.send(data).await.is_err() { // Channel closed, consumer is done context.in_flight.fetch_sub(1, Ordering::Relaxed); let backoff = Duration::seconds(60); + let error_msg = "Consumer channel closed"; let _ = queue - .mark_failed(task_id, claim_attempt, "Consumer channel closed", backoff) + .mark_failed(task_id, claim_attempt, error_msg, backoff) .await; + fail_progress_for_task(&task, &context.progress, error_msg); + { + let mut s = context.stats.lock().await; + s.failed += 1; + } + context.progress.log(format!( + "Producer P{} could not hand task #{} to writer: {}", + id, task_id, error_msg + )); + context.progress.clear_worker("P", id); break; } + context.progress.clear_worker("P", id); } - Err(GarminError::RateLimited) => { + Ok(Err(GarminError::RateLimited)) => { context.rate_limiter.on_rate_limit(); let backoff = Duration::seconds(60); if let Err(e) = queue .mark_failed(task_id, claim_attempt, "Rate limited", backoff) .await { - eprintln!("Failed to mark task as rate limited: {}", e); + context.progress.log(format!( + "Producer P{} failed to mark task #{} as rate limited: {}", + id, task_id, e + )); } fail_progress_for_task(&task, &context.progress, "Rate limited"); { let mut s = context.stats.lock().await; s.rate_limited += 1; } + context.progress.log(format!( + "Producer P{} hit rate limit on task #{}; retry scheduled in {}s", + id, + task_id, + backoff.num_seconds() + )); + context.progress.clear_worker("P", id); context.in_flight.fetch_sub(1, Ordering::Relaxed); } - Err(e) => { + Ok(Err(e)) => { let backoff = Duration::seconds(60); let error_msg = e.to_string(); if let Err(e) = queue .mark_failed(task_id, claim_attempt, &error_msg, backoff) .await { - eprintln!("Failed to mark task as failed: {}", e); + context.progress.log(format!( + "Producer P{} failed to mark task #{} as failed: {}", + id, task_id, e + )); + } + fail_progress_for_task(&task, &context.progress, &error_msg); + { + let mut s = context.stats.lock().await; + s.failed += 1; + } + context.progress.log(format!( + "Producer P{} failed task #{} ({}): {}", + id, + task_id, + task_progress_metadata(&task) + .map(|(_, item)| item) + .unwrap_or_else(|| "unknown task".to_string()), + error_msg + )); + context.progress.clear_worker("P", id); + context.in_flight.fetch_sub(1, Ordering::Relaxed); + } + Err(_) => { + let backoff = Duration::seconds(60); + let error_msg = format!("Timed out after {} seconds", TASK_TIMEOUT_SECS); + if let Err(e) = queue + .mark_failed(task_id, claim_attempt, &error_msg, backoff) + .await + { + context.progress.log(format!( + "Producer P{} failed to mark timed-out task #{} as failed: {}", + id, task_id, e + )); } fail_progress_for_task(&task, &context.progress, &error_msg); { let mut s = context.stats.lock().await; s.failed += 1; } + context.progress.log(format!( + "Producer P{} timed out task #{} ({}); retry scheduled in {}s", + id, + task_id, + task_progress_metadata(&task) + .map(|(_, item)| item) + .unwrap_or_else(|| "unknown task".to_string()), + backoff.num_seconds() + )); + context.progress.clear_worker("P", id); context.in_flight.fetch_sub(1, Ordering::Relaxed); } } @@ -2194,7 +2371,7 @@ fn parse_f64_stream_value(value: &str) -> Option { /// Consumer loop: receives data from channel and writes to Parquet async fn consumer_loop( - _id: usize, + id: usize, rx: Arc>>, parquet: Arc, queue: SharedTaskQueue, @@ -2214,6 +2391,9 @@ async fn consumer_loop( None => break, // Channel closed, all producers done }; + let (stream, item, active_task_id) = sync_data_progress_metadata(&data); + progress.set_worker_task("C", id, Some(active_task_id), stream, item, "writing"); + // Process and write data let result = match &data { SyncData::Activities { @@ -2346,7 +2526,10 @@ async fn consumer_loop( .mark_completed_with_followups(task_id, claim_attempt, &followup_tasks) .await { - eprintln!("Failed to mark task completed: {}", e); + progress.log(format!( + "Consumer C{} failed to mark task #{} completed: {}", + id, task_id, e + )); in_flight.fetch_sub(1, Ordering::Relaxed); continue; } @@ -2385,6 +2568,7 @@ async fn consumer_loop( } _ => {} } + progress.clear_worker("C", id); in_flight.fetch_sub(1, Ordering::Relaxed); } Err(e) => { @@ -2395,7 +2579,10 @@ async fn consumer_loop( .mark_failed(task_id, claim_attempt, &error_msg, backoff) .await { - eprintln!("Failed to mark task as failed: {}", e); + progress.log(format!( + "Consumer C{} failed to mark task #{} as failed: {}", + id, task_id, e + )); } // Update stats @@ -2405,7 +2592,11 @@ async fn consumer_loop( } record_write_failure(&data, &progress, &error_msg); - eprintln!("Write error for {}: {}", task_type, error_msg); + progress.log(format!( + "Consumer C{} write error for task #{} ({}): {}", + id, task_id, task_type, error_msg + )); + progress.clear_worker("C", id); in_flight.fetch_sub(1, Ordering::Relaxed); } } diff --git a/crates/garmin-cli/src/sync/progress.rs b/crates/garmin-cli/src/sync/progress.rs index cd89eac..11040a5 100644 --- a/crates/garmin-cli/src/sync/progress.rs +++ b/crates/garmin-cli/src/sync/progress.rs @@ -4,6 +4,8 @@ use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU8, Ordering}; use std::sync::{Arc, Mutex}; use std::time::Instant; +use crate::client::RequestLogger; + /// Sync mode - latest (recent data) vs backfill (historical) #[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] pub enum SyncMode { @@ -44,6 +46,30 @@ pub struct ErrorEntry { pub error: String, } +/// Current work assigned to a sync worker. +#[derive(Debug, Clone)] +pub struct WorkerProgress { + pub role: &'static str, + pub id: usize, + pub task_id: Option, + pub stream: &'static str, + pub item: String, + pub started_at: Instant, + pub status: String, +} + +impl WorkerProgress { + /// Human-readable worker label for compact status output. + pub fn label(&self) -> String { + format!("{}{}", self.role, self.id) + } + + /// How long this worker has been on the current task. + pub fn elapsed_str(&self) -> String { + format_elapsed(self.started_at.elapsed().as_secs()) + } +} + /// Progress tracking for a single data stream #[derive(Debug)] pub struct StreamProgress { @@ -207,6 +233,12 @@ pub struct SyncProgress { pub planning_step: Mutex, /// Oldest activity date found during planning pub oldest_activity_date: Mutex>, + /// Current task for each active worker + pub workers: Mutex>, + /// Optional detailed log writer + pub logger: Mutex>, + /// Number of producer workers to render + worker_slots: AtomicU32, } impl SyncProgress { @@ -231,6 +263,9 @@ impl SyncProgress { shutdown: AtomicBool::new(false), planning_step: Mutex::new(PlanningStep::FetchingProfile), oldest_activity_date: Mutex::new(None), + workers: Mutex::new(Vec::new()), + logger: Mutex::new(None), + worker_slots: AtomicU32::new(0), } } @@ -357,6 +392,106 @@ impl SyncProgress { self.errors.lock().unwrap().clone() } + /// Attach a detailed sync log writer. + pub fn set_logger(&self, logger: Option) { + *self.logger.lock().unwrap() = logger; + } + + /// Write a detailed log event if logging is enabled. + pub fn log(&self, message: impl AsRef) { + if let Some(logger) = self.logger.lock().unwrap().as_ref() { + logger.log(message); + } + } + + /// Set how many producer worker lines should be rendered. + pub fn set_worker_slots(&self, slots: usize) { + self.worker_slots.store(slots as u32, Ordering::Relaxed); + } + + /// Mark a worker as actively processing a task. + pub fn set_worker_task( + &self, + role: &'static str, + id: usize, + task_id: Option, + stream: &'static str, + item: String, + status: impl Into, + ) { + let mut workers = self.workers.lock().unwrap(); + if let Some(worker) = workers.iter_mut().find(|w| w.role == role && w.id == id) { + worker.task_id = task_id; + worker.stream = stream; + worker.item = item; + worker.started_at = Instant::now(); + worker.status = status.into(); + } else { + workers.push(WorkerProgress { + role, + id, + task_id, + stream, + item, + started_at: Instant::now(), + status: status.into(), + }); + } + } + + /// Update the displayed status for an active worker without resetting elapsed time. + pub fn update_worker_status(&self, role: &'static str, id: usize, status: impl Into) { + let mut workers = self.workers.lock().unwrap(); + if let Some(worker) = workers.iter_mut().find(|w| w.role == role && w.id == id) { + worker.status = status.into(); + } + } + + /// Clear a worker's current task. + pub fn clear_worker(&self, role: &'static str, id: usize) { + let mut workers = self.workers.lock().unwrap(); + workers.retain(|w| !(w.role == role && w.id == id)); + } + + /// Get current worker statuses. + pub fn get_workers(&self) -> Vec { + self.workers.lock().unwrap().clone() + } + + /// Get fixed-height terminal status lines. + pub fn status_lines(&self) -> Vec { + let act = &self.activities; + let gpx = &self.gpx; + let health = &self.health; + let perf = &self.performance; + let workers = self.get_workers(); + let slots = self.worker_slots.load(Ordering::Relaxed) as usize; + let width = terminal_width(); + + let mut lines = vec![truncate_line( + format!( + "Act {}/{} | GPX {}/{} | Health {}/{} | Perf {}/{} | elapsed {}", + act.get_completed(), + act.get_total(), + gpx.get_completed(), + gpx.get_total(), + health.get_completed(), + health.get_total(), + perf.get_completed(), + perf.get_total(), + self.elapsed_str(), + ), + width, + )]; + + for id in 0..slots { + let worker = workers.iter().find(|w| w.role == "P" && w.id == id); + lines.push(truncate_line(format_worker_line("P", id, worker), width)); + } + + lines + } + /// Get current task description (finds first active stream) pub fn get_current_task(&self) -> Option { // Check streams in order of typical processing @@ -479,27 +614,64 @@ impl SyncProgress { /// Print a status line for terminal progress reporting. pub fn print_simple_status(&self) { - let act = &self.activities; - let gpx = &self.gpx; - let health = &self.health; - let perf = &self.performance; - - print!( - "\rAct: {}/{} | GPX: {}/{} | Health: {}/{} | Perf: {}/{} | {} ", - act.get_completed(), - act.get_total(), - gpx.get_completed(), - gpx.get_total(), - health.get_completed(), - health.get_total(), - perf.get_completed(), - perf.get_total(), - self.elapsed_str(), - ); + print!("{}", self.status_lines().join("\n")); let _ = std::io::Write::flush(&mut std::io::stdout()); } } +fn format_worker_line(role: &str, id: usize, worker: Option<&WorkerProgress>) -> String { + match worker { + Some(worker) => { + let task = worker + .task_id + .map(|task_id| format!("#{} ", task_id)) + .unwrap_or_default(); + format!( + "{}{} {:<11} {}{} {:<22} {}", + role, + id, + worker.stream, + task, + worker.item, + worker.status, + worker.elapsed_str() + ) + } + None => format!("{}{} idle", role, id), + } +} + +fn terminal_width() -> usize { + std::env::var("COLUMNS") + .ok() + .and_then(|value| value.parse::().ok()) + .filter(|width| *width >= 40) + .unwrap_or(120) +} + +fn truncate_line(line: String, width: usize) -> String { + let count = line.chars().count(); + if count <= width { + return line; + } + + let keep = width.saturating_sub(1); + let mut truncated = line.chars().take(keep).collect::(); + truncated.push('~'); + truncated +} + +fn format_elapsed(secs: u64) -> String { + let mins = secs / 60; + let remaining_secs = secs % 60; + + if mins > 0 { + format!("{}m {}s", mins, remaining_secs) + } else { + format!("{}s", secs) + } +} + impl Default for SyncProgress { fn default() -> Self { Self::new() @@ -565,4 +737,28 @@ mod tests { assert_eq!(progress.total_completed(), 2); } + + #[test] + fn test_worker_progress_tracks_current_tasks() { + let progress = SyncProgress::new(); + + progress.set_worker_task( + "P", + 0, + Some(42), + "Health", + "2026-01-01".to_string(), + "fetching", + ); + let workers = progress.get_workers(); + assert_eq!(workers.len(), 1); + assert_eq!(workers[0].label(), "P0"); + assert_eq!(workers[0].task_id, Some(42)); + + progress.update_worker_status("P", 0, "waiting for writer"); + assert_eq!(progress.get_workers()[0].status, "waiting for writer"); + + progress.clear_worker("P", 0); + assert!(progress.get_workers().is_empty()); + } }