diff options
Diffstat (limited to '')
| -rw-r--r-- | crates/daemon/src/api/control.rs (renamed from crates/turtle/src/atuin_daemon/components/sync.rs) | 186 |
1 files changed, 101 insertions, 85 deletions
diff --git a/crates/turtle/src/atuin_daemon/components/sync.rs b/crates/daemon/src/api/control.rs index 20d49839..16b4bd94 100644 --- a/crates/turtle/src/atuin_daemon/components/sync.rs +++ b/crates/daemon/src/api/control.rs @@ -1,29 +1,26 @@ -//! Sync component. -//! -//! Handles periodic synchronization with the Atuin cloud server. - use std::time::Duration; use eyre::Result; -use rand::Rng; -use tokio::sync::mpsc; +use rand::RngExt; use tokio::time::{self, MissedTickBehavior}; +use tonic::{Request, Response, Status}; +use tracing::{Level, instrument}; -use crate::atuin_client::{history::store::HistoryStore, record::sync, settings::Settings}; +use turtle_api::generated::{ + DAEMON_PROTOCOL_VERSION, + control::{ + ForceSyncReply, ForceSyncRequest, PathsReply, PathsRequest, StatusReply, StatusRequest, + control_server::{Control, ControlServer}, + }, +}; -use crate::atuin_daemon::{ - daemon::{Component, DaemonHandle}, +use crate::{ + DAEMON_VERSION, + aclient::{history::store::HistoryStore, record::sync, settings::Settings}, + daemon::DaemonHandle, events::DaemonEvent, }; -/// Commands that can be sent to the sync task. -enum SyncCommand { - /// Trigger an immediate sync. - ForceSync, - /// Stop the sync loop. - Stop, -} - /// Sync state - tracks whether we're in normal operation or retrying after failure. #[derive(Clone, Copy, PartialEq, Eq)] enum SyncState { @@ -34,77 +31,98 @@ enum SyncState { Retrying, } -/// Sync component - handles periodic cloud synchronization. +/// The Control gRPC service. /// -/// This component: -/// - Runs a background sync loop on a configurable interval -/// - Implements exponential backoff on sync failures -/// - Responds to [`ForceSync`] events for immediate sync -/// - Emits SyncCompleted/SyncFailed events -pub(crate) struct SyncComponent { - task_handle: Option<tokio::task::JoinHandle<()>>, - command_tx: Option<mpsc::Sender<SyncCommand>>, +/// 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, } -impl SyncComponent { - /// Create a new sync component. - pub(crate) fn new() -> Self { - Self { - task_handle: None, - command_tx: None, - } +impl ControlService { + /// Create a new control service with the given daemon handle. + pub(crate) fn new(handle: DaemonHandle) -> Self { + tokio::spawn(sync_loop(handle.clone())); + + Self { handle } } -} -impl Default for SyncComponent { - fn default() -> Self { - Self::new() + /// Get a tonic server for this service. + pub(crate) fn into_server(self) -> ControlServer<Self> { + ControlServer::new(self) } } #[tonic::async_trait] -impl Component for SyncComponent { - fn name(&self) -> &'static str { - "sync" - } +impl Control for ControlService { + #[instrument(skip_all, level = Level::INFO)] + async fn paths(&self, _request: Request<PathsRequest>) -> Result<Response<PathsReply>, Status> { + let settings = self.handle.settings().await; - async fn start(&mut self, handle: DaemonHandle) -> Result<()> { - let (cmd_tx, cmd_rx) = mpsc::channel(16); - self.command_tx = Some(cmd_tx); + let config = Settings::get_config_path() + .map_err(|e| Status::internal(format!("failed to get settings path: {e:?}")))?; - // Spawn the sync loop with its own copy of the handle - self.task_handle = Some(tokio::spawn(sync_loop(handle, cmd_rx))); + let reply = PathsReply { + config: config.to_string_lossy().to_string(), + db: settings.db_path.clone(), + socket: settings.daemon.socket_path.clone(), + }; - tracing::info!("sync component started"); - Ok(()) + Ok(Response::new(reply)) } - async fn handle_event(&mut self, event: &DaemonEvent) -> Result<()> { - match event { - DaemonEvent::ForceSync => { - tracing::info!("force sync requested"); - if let Some(tx) = &self.command_tx { - drop(tx.send(SyncCommand::ForceSync).await); - } - } - DaemonEvent::SyncFailed { error } => { - tracing::error!(?error, "Sync failed."); - } - _ => (), - } - Ok(()) + #[instrument(skip_all, level = Level::INFO)] + async fn status( + &self, + _request: Request<StatusRequest>, + ) -> Result<Response<StatusReply>, Status> { + let reply = StatusReply { + healthy: true, + version: DAEMON_VERSION.to_owned(), + pid: std::process::id(), + protocol: DAEMON_PROTOCOL_VERSION, + }; + + Ok(Response::new(reply)) } - async fn stop(&mut self) -> Result<()> { - if let Some(tx) = &self.command_tx { - drop(tx.send(SyncCommand::Stop).await); - } - if let Some(handle) = self.task_handle.take() { - // Give the task a moment to shut down gracefully - drop(time::timeout(Duration::from_secs(5), handle).await); - } - tracing::info!("sync component stopped"); - Ok(()) + #[instrument(skip_all, level = Level::INFO)] + async fn force_sync( + &self, + _request: Request<ForceSyncRequest>, + ) -> Result<Response<ForceSyncReply>, Status> { + self.handle.emit(DaemonEvent::ForceSync); + let event = self + .handle + .wait_for(|e| { + matches!( + e, + DaemonEvent::SyncFailed { .. } | DaemonEvent::SyncCompleted { .. } + ) + }) + .await + .map_err(|e| { + Status::internal(format!("failed to wait for sync response event: {e:?}")) + })?; + + let reply = match event { + DaemonEvent::SyncCompleted { + uploaded, + downloaded, + } => ForceSyncReply { + error: None, + uploaded: uploaded as u32, + downloaded: downloaded as u32, + }, + DaemonEvent::SyncFailed { error } => ForceSyncReply { + error: Some(error), + uploaded: 0, + downloaded: 0, + }, + _ => unreachable!(), + }; + + Ok(Response::new(reply)) } } @@ -113,7 +131,7 @@ impl Component for SyncComponent { /// 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, mut cmd_rx: mpsc::Receiver<SyncCommand>) { +async fn sync_loop(handle: DaemonHandle) { tracing::info!("sync loop starting"); // Clone settings since we need them across await points @@ -131,7 +149,7 @@ async fn sync_loop(handle: DaemonHandle, mut cmd_rx: mpsc::Receiver<SyncCommand> 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 max_interval: f64 = 60.0f64.mul_add(30.0, rand::rng().random_range(0.0..60.0)); let mut ticker = time::interval(Duration::from_secs(settings.daemon.sync_frequency)); @@ -141,6 +159,7 @@ async fn sync_loop(handle: DaemonHandle, mut cmd_rx: mpsc::Receiver<SyncCommand> let mut sync_state = SyncState::Idle; + let mut daemon_rx = handle.subscribe(); loop { tokio::select! { _ = ticker.tick() => { @@ -161,9 +180,9 @@ async fn sync_loop(handle: DaemonHandle, mut cmd_rx: mpsc::Receiver<SyncCommand> &settings, ).await; } - cmd = cmd_rx.recv() => { + cmd = daemon_rx.recv() => { match cmd { - Some(SyncCommand::ForceSync) => { + Ok(DaemonEvent::ForceSync) => { tracing::info!("executing force sync"); let settings = handle.settings().await; sync_state = do_sync_tick( @@ -173,11 +192,12 @@ async fn sync_loop(handle: DaemonHandle, mut cmd_rx: mpsc::Receiver<SyncCommand> max_interval, &settings, ).await; - } - Some(SyncCommand::Stop) | None => { + }, + Ok(DaemonEvent::ShutdownRequested) | Err(_) => { tracing::info!("sync loop stopping"); break; - } + }, + _ => () } } } @@ -217,14 +237,13 @@ async fn do_sync_tick( 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); + let mut rng = rand::rng(); + let mut new_interval = ticker.period().as_secs_f64() * rng.random_range(2.0..2.2); if new_interval > max_interval { new_interval = max_interval; @@ -256,9 +275,6 @@ async fn do_sync_tick( 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, |