From 1594307d9cd819b1ed739fa1ec048c6b27e6f4b0 Mon Sep 17 00:00:00 2001 From: Benedikt Peetz Date: Mon, 20 Jul 2026 15:13:12 +0200 Subject: chore: Commit --- crates/daemon/src/api/server/control.rs | 203 +++++++++++++++++++++++++++++++- 1 file changed, 202 insertions(+), 1 deletion(-) (limited to 'crates/daemon/src/api/server/control.rs') diff --git a/crates/daemon/src/api/server/control.rs b/crates/daemon/src/api/server/control.rs index 63fef340..8d1ec7b8 100644 --- a/crates/daemon/src/api/server/control.rs +++ b/crates/daemon/src/api/server/control.rs @@ -1,7 +1,13 @@ +use std::time::Duration; + +use eyre::Result; +use rand::Rng; +use tokio::time::{self, MissedTickBehavior}; use tonic::{Request, Response, Status}; use tracing::{Level, instrument}; use crate::{ + aclient::{history::store::HistoryStore, record::sync, settings::Settings}, api::{ DAEMON_PROTOCOL_VERSION, DAEMON_VERSION, generated::control::{ @@ -10,20 +16,37 @@ use crate::{ }, }, daemon::DaemonHandle, + events::DaemonEvent, }; +/// Sync state - tracks whether we're in normal operation or retrying after failure. +#[derive(Clone, Copy, PartialEq, Eq)] +enum SyncState { + /// Normal operation. Periodic syncs only run if [`auto_sync`] is enabled. + Idle, + /// Retrying after a sync failure. Retries continue regardless of [`auto_sync`] + /// until the sync succeeds. + Retrying, +} + /// The Control gRPC service. /// /// This service is used by external processes to inject events into the daemon. /// It's not a component - it's part of the daemon's core infrastructure. pub(crate) struct ControlService { handle: DaemonHandle, + task_handle: tokio::task::JoinHandle<()>, } impl ControlService { /// Create a new control service with the given daemon handle. pub(crate) fn new(handle: DaemonHandle) -> Self { - Self { handle } + let task_handle = tokio::spawn(sync_loop(handle.clone())); + + Self { + handle, + task_handle, + } } /// Get a tonic server for this service. @@ -59,3 +82,181 @@ impl Control for ControlService { Ok(Response::new(reply)) } } + +/// The main sync loop. +/// +/// This runs in a spawned task and handles periodic sync as well as +/// force sync requests. +#[expect(clippy::significant_drop_tightening, reason = "false positive")] +async fn sync_loop(handle: DaemonHandle) { + tracing::info!("sync loop starting"); + + // Clone settings since we need them across await points + let settings = handle.settings().await.clone(); + let host_id = match Settings::host_id().await { + Ok(id) => id, + Err(e) => { + tracing::error!("failed to get host id, sync disabled: {e}"); + return; + } + }; + + // Create the stores we need + let encryption_key = *handle.encryption_key(); + let history_store = HistoryStore::new(handle.store().clone(), host_id, encryption_key); + + // Don't backoff by more than 30 mins (with a random jitter of up to 1 min) + let max_interval: f64 = 60.0f64.mul_add(30.0, rand::thread_rng().gen_range(0.0..60.0)); + + let mut ticker = time::interval(Duration::from_secs(settings.daemon.sync_frequency)); + + // IMPORTANT: without this, if we miss ticks because a sync takes ages or is otherwise delayed, + // we may end up running a lot of syncs in a hot loop. + ticker.set_missed_tick_behavior(MissedTickBehavior::Skip); + + let mut sync_state = SyncState::Idle; + + let mut daemon_rx = handle.subscribe(); + loop { + tokio::select! { + _ = ticker.tick() => { + let settings = handle.settings().await; + + // Skip periodic ticks if auto_sync is disabled AND we're not retrying + // a previous failure. Retries must continue regardless of auto_sync. + if !settings.sync.auto && sync_state == SyncState::Idle { + tracing::debug!("auto_sync disabled, skipping periodic sync tick"); + continue; + } + + sync_state = do_sync_tick( + &handle, + &history_store, + &mut ticker, + max_interval, + &settings, + ).await; + } + cmd = daemon_rx.recv() => { + match cmd { + Ok(DaemonEvent::ForceSync) => { + tracing::info!("executing force sync"); + let settings = handle.settings().await; + sync_state = do_sync_tick( + &handle, + &history_store, + &mut ticker, + max_interval, + &settings, + ).await; + }, + Ok(DaemonEvent::ShutdownRequested) | Err(_) => { + tracing::info!("sync loop stopping"); + break; + }, + _ => () + } + } + } + } +} + +/// Execute a single sync tick. +/// +/// Returns the new sync state: `Idle` on success, `Retrying` on failure. +async fn do_sync_tick( + handle: &DaemonHandle, + history_store: &HistoryStore, + ticker: &mut time::Interval, + max_interval: f64, + settings: &Settings, +) -> SyncState { + tracing::info!("sync tick"); + + // Check if logged in + let logged_in = match settings.sync.have_sync_user() { + Ok(v) => v, + Err(e) => { + tracing::warn!("failed to check login status, skipping sync tick: {e}"); + return SyncState::Idle; + } + }; + + if !logged_in { + tracing::debug!("not logged in, skipping sync tick"); + return SyncState::Idle; + } + + // Perform the sync + let res = sync::sync(settings, handle.store(), handle.encryption_key()).await; + + match res { + Err(e) => { + tracing::error!("sync tick failed with {e}"); + + // Emit failure event + handle.emit(DaemonEvent::SyncFailed { + error: e.to_string(), + }); + + // Exponential backoff + let mut rng = rand::thread_rng(); + let mut new_interval = ticker.period().as_secs_f64() * rng.gen_range(2.0..2.2); + + if new_interval > max_interval { + new_interval = max_interval; + } + + *ticker = time::interval_at( + time::Instant::now() + Duration::from_secs(new_interval as u64), + Duration::from_secs(new_interval as u64), + ); + ticker.reset_after(Duration::from_secs(new_interval as u64)); + ticker.set_missed_tick_behavior(MissedTickBehavior::Skip); + + tracing::error!("backing off, next sync tick in {new_interval}"); + + SyncState::Retrying + } + Ok((uploaded_count, downloaded_records)) => { + tracing::info!( + uploaded = uploaded_count, + downloaded = downloaded_records.len(), + "sync complete" + ); + + // Build history from downloaded records + if let Err(e) = history_store + .incremental_build(handle.history_db(), &downloaded_records) + .await + { + tracing::error!("failed to build history from downloaded records: {e}"); + } + + // Emit the records added event (for search indexing) + handle.emit(DaemonEvent::RecordsAdded(downloaded_records.clone())); + + // Emit sync completed event + handle.emit(DaemonEvent::SyncCompleted { + uploaded: uploaded_count as usize, + downloaded: downloaded_records.len(), + }); + + // Reset backoff on success + if ticker.period().as_secs() != settings.daemon.sync_frequency { + *ticker = time::interval_at( + time::Instant::now() + Duration::from_secs(settings.daemon.sync_frequency), + Duration::from_secs(settings.daemon.sync_frequency), + ); + ticker.set_missed_tick_behavior(MissedTickBehavior::Skip); + } + + // Store sync time + if let Err(e) = Settings::save_sync_time().await { + tracing::error!("failed to save sync time: {e}"); + } + + SyncState::Idle + } + } +} -- cgit v1.3.1