From 966a80c4199a49898cc7d8641012d520ce6b2efa Mon Sep 17 00:00:00 2001 From: Benedikt Peetz Date: Mon, 20 Jul 2026 19:30:40 +0200 Subject: chore: Commit --- crates/daemon/src/api/client/mod.rs | 272 ------------------------------- crates/daemon/src/api/control.rs | 280 ++++++++++++++++++++++++++++++++ crates/daemon/src/api/generated.rs | 28 ---- crates/daemon/src/api/history.rs | 253 +++++++++++++++++++++++++++++ crates/daemon/src/api/mod.rs | 8 +- crates/daemon/src/api/server/control.rs | 278 ------------------------------- crates/daemon/src/api/server/history.rs | 250 ---------------------------- crates/daemon/src/api/server/mod.rs | 2 - 8 files changed, 535 insertions(+), 836 deletions(-) delete mode 100644 crates/daemon/src/api/client/mod.rs create mode 100644 crates/daemon/src/api/control.rs delete mode 100644 crates/daemon/src/api/generated.rs create mode 100644 crates/daemon/src/api/history.rs delete mode 100644 crates/daemon/src/api/server/control.rs delete mode 100644 crates/daemon/src/api/server/history.rs delete mode 100644 crates/daemon/src/api/server/mod.rs (limited to 'crates/daemon/src/api') diff --git a/crates/daemon/src/api/client/mod.rs b/crates/daemon/src/api/client/mod.rs deleted file mode 100644 index c588fb09..00000000 --- a/crates/daemon/src/api/client/mod.rs +++ /dev/null @@ -1,272 +0,0 @@ -use eyre::{Context as EyreContext, Result}; -use time::OffsetDateTime; -use tonic::Code; -use tonic::transport::{Channel, Endpoint, Uri}; -use tower::service_fn; - -use hyper_util::rt::TokioIo; - -#[cfg(unix)] -use tokio::net::UnixStream; - -use crate::api::generated; -use crate::api::generated::control::{ForceSyncReply, ForceSyncRequest, PathsReply, PathsRequest}; -use crate::api::generated::history::{HistoryEntry, HistoryRequest}; -use crate::{ - aclient::history::History, - api::{ - DAEMON_PROTOCOL_VERSION, DAEMON_VERSION, - generated::{ - control::{ - StatusReply, StatusRequest, control_client::ControlClient as ControlServiceClient, - }, - history::{ - EndHistoryReply, EndHistoryRequest, StartHistoryReply, StartHistoryRequest, - TailHistoryRequest, history_client::HistoryClient as HistoryServiceClient, - }, - }, - }, -}; - -pub use crate::api::generated::history::{HistoryEventKind, TailHistoryReply}; - -fn normalize_optional_field(value: &str) -> Option { - let trimmed = value.trim(); - if trimmed.is_empty() { - None - } else { - Some(trimmed.to_owned()) - } -} - -pub fn history_entry_to_history(entry: HistoryEntry) -> History { - let timestamp = OffsetDateTime::from_unix_timestamp_nanos(i128::from(entry.timestamp)) - .expect("Daemon history timestamp should always be valid"); - - History { - id: entry.id.into(), - timestamp, - duration: entry.duration, - exit: entry.exit, - command: entry.command, - cwd: entry.cwd, - session: entry.session, - hostname: entry.hostname, - author: entry.author, - intent: normalize_optional_field(&entry.intent), - deleted_at: None, - } -} - -#[must_use] -pub fn daemon_matches_expected(version: &str, protocol: u32) -> bool { - version == DAEMON_VERSION && protocol == DAEMON_PROTOCOL_VERSION -} - -#[must_use] -pub fn daemon_mismatch_message(version: &str, protocol: u32) -> String { - if protocol == DAEMON_PROTOCOL_VERSION { - format!("daemon is out of date: expected {DAEMON_VERSION}, got {version}") - } else { - format!("daemon protocol mismatch: expected {DAEMON_PROTOCOL_VERSION}, got {protocol}") - } -} - -#[derive(Clone, Copy, Debug, Eq, PartialEq)] -pub enum DaemonClientErrorKind { - Connect, - Unavailable, - Unimplemented, - Other, -} - -#[must_use] -pub fn classify_error(error: &eyre::Report) -> DaemonClientErrorKind { - for cause in error.chain() { - if cause.downcast_ref::().is_some() { - return DaemonClientErrorKind::Connect; - } - - if let Some(status) = cause.downcast_ref::() { - return match status.code() { - Code::Unavailable => DaemonClientErrorKind::Unavailable, - Code::Unimplemented => DaemonClientErrorKind::Unimplemented, - _ => DaemonClientErrorKind::Other, - }; - } - } - - DaemonClientErrorKind::Other -} - -#[derive(Debug)] -pub enum Probe { - Ready(ControlClient), - NeedsRestart(String), - Unreachable(eyre::Report), -} - -/// Check if a client can reach the daemon. -pub async fn probe(path: String) -> Probe { - let mut client = match ControlClient::new(path).await { - Ok(client) => client, - Err(err) => return Probe::Unreachable(err), - }; - - match client.status().await { - Ok(status) => { - if daemon_matches_expected(&status.version, status.protocol) { - Probe::Ready(client) - } else { - Probe::NeedsRestart(daemon_mismatch_message(&status.version, status.protocol)) - } - } - Err(err) => Probe::Unreachable(err), - } -} - -// ============================================================================ -// History Client -// ============================================================================ - -#[derive(Debug)] -pub struct HistoryClient { - client: HistoryServiceClient, -} - -pub struct Range { - pub start: OffsetDateTime, - pub end: OffsetDateTime, -} - -// Wrap the grpc client -impl HistoryClient { - #[cfg(unix)] - pub async fn new(path: String) -> Result { - use eyre::Context; - - let log_path = path.clone(); - let channel = Endpoint::try_from("http://atuin_local_daemon:0")? - .connect_with_connector(service_fn(move |_: Uri| { - let path = path.clone(); - - async move { - Ok::<_, std::io::Error>(TokioIo::new(UnixStream::connect(path.clone()).await?)) - } - })) - .await - .wrap_err_with(|| { - format!( - "failed to connect to local atuin daemon at {}. Is it running?", - &log_path - ) - })?; - - let client = HistoryServiceClient::new(channel); - - Ok(Self { client }) - } - - pub async fn start_history(&mut self, h: History) -> Result { - let req = StartHistoryRequest { - command: h.command, - cwd: h.cwd, - hostname: h.hostname, - session: h.session, - timestamp: h.timestamp.unix_timestamp_nanos() as u64, - author: h.author, - intent: h.intent.unwrap_or_default(), - }; - - Ok(self.client.start_history(req).await?.into_inner()) - } - - pub async fn history(&mut self, session: String, range: Option) -> Result> { - let req = HistoryRequest { - session, - range: range.map(|r| generated::history::Range { - start: r.start.unix_timestamp() as u64, - end: r.end.unix_timestamp() as u64, - }), - }; - - let reply = self.client.history(req).await?.into_inner(); - - Ok(reply - .entries - .into_iter() - .map(history_entry_to_history) - .collect()) - } - - pub async fn end_history( - &mut self, - id: String, - duration: u64, - exit: i64, - ) -> Result { - let req = EndHistoryRequest { id, exit, duration }; - - Ok(self.client.end_history(req).await?.into_inner()) - } - - pub async fn tail_history(&mut self) -> Result> { - Ok(self - .client - .tail_history(TailHistoryRequest {}) - .await? - .into_inner()) - } -} - -// ============================================================================ -// Control Client -// ============================================================================ - -/// Client for the Control gRPC service. -#[derive(Debug)] -pub struct ControlClient { - client: ControlServiceClient, -} - -impl ControlClient { - /// Connect to the daemon's control service. - pub async fn new(path: String) -> Result { - let log_path = path.clone(); - let channel = Endpoint::try_from("http://atuin_local_daemon:0")? - .connect_with_connector(service_fn(move |_: Uri| { - let path = path.clone(); - - async move { - Ok::<_, std::io::Error>(TokioIo::new(UnixStream::connect(path.clone()).await?)) - } - })) - .await - .wrap_err_with(|| { - format!( - "failed to connect to local atuin daemon at {}. Is it running?", - &log_path - ) - })?; - - let client = ControlServiceClient::new(channel); - - Ok(Self { client }) - } - - pub async fn paths(&mut self) -> Result { - Ok(self.client.paths(PathsRequest {}).await?.into_inner()) - } - - pub async fn force_sync(&mut self) -> Result { - Ok(self - .client - .force_sync(ForceSyncRequest {}) - .await? - .into_inner()) - } - - pub async fn status(&mut self) -> Result { - Ok(self.client.status(StatusRequest {}).await?.into_inner()) - } -} diff --git a/crates/daemon/src/api/control.rs b/crates/daemon/src/api/control.rs new file mode 100644 index 00000000..a9d9cff3 --- /dev/null +++ b/crates/daemon/src/api/control.rs @@ -0,0 +1,280 @@ +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 turtle::generated::{ + DAEMON_PROTOCOL_VERSION, + control::{ + ForceSyncReply, ForceSyncRequest, PathsReply, PathsRequest, StatusReply, StatusRequest, + control_server::{Control, ControlServer}, + }, +}; + +use crate::{ + DAEMON_VERSION, + aclient::{history::store::HistoryStore, record::sync, settings::Settings}, + 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 { + let task_handle = tokio::spawn(sync_loop(handle.clone())); + + Self { + handle, + task_handle, + } + } + + /// Get a tonic server for this service. + pub(crate) fn into_server(self) -> ControlServer { + ControlServer::new(self) + } +} + +#[tonic::async_trait] +impl Control for ControlService { + #[instrument(skip_all, level = Level::INFO)] + async fn paths(&self, _request: Request) -> Result, Status> { + let settings = self.handle.settings().await; + + let config = Settings::get_config_path() + .map_err(|e| Status::internal(format!("failed to get settings path: {e:?}")))?; + + let reply = PathsReply { + config: config.to_string_lossy().to_string(), + db: settings.db_path.clone(), + socket: settings.daemon.socket_path.clone(), + }; + + Ok(Response::new(reply)) + } + + #[instrument(skip_all, level = Level::INFO)] + async fn status( + &self, + _request: Request, + ) -> Result, Status> { + let reply = StatusReply { + healthy: true, + version: DAEMON_VERSION.to_owned(), + pid: std::process::id(), + protocol: DAEMON_PROTOCOL_VERSION, + }; + + Ok(Response::new(reply)) + } + + #[instrument(skip_all, level = Level::INFO)] + async fn force_sync( + &self, + _request: Request, + ) -> Result, Status> { + let reply = ForceSyncReply { accepted: false }; + + 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 + } + } +} diff --git a/crates/daemon/src/api/generated.rs b/crates/daemon/src/api/generated.rs deleted file mode 100644 index 304edcd9..00000000 --- a/crates/daemon/src/api/generated.rs +++ /dev/null @@ -1,28 +0,0 @@ -#![expect( - unreachable_pub, - unused_qualifications, - clippy::doc_markdown, - clippy::default_trait_access, - clippy::too_many_lines, - clippy::trivially_copy_pass_by_ref, - clippy::allow_attributes, - clippy::derive_partial_eq_without_eq, - reason = "All of these lints are triggered by the generated code" -)] - -/// Semantic command capture gRPC service types. -pub(crate) mod semantic { - tonic::include_proto!("semantic"); -} - -/// History module for the daemon gRPC history service. -/// -/// This module contains the proto-generated types for the history gRPC service. -pub(crate) mod history { - tonic::include_proto!("history"); -} - -/// Control module for external control. -pub(crate) mod control { - tonic::include_proto!("control"); -} diff --git a/crates/daemon/src/api/history.rs b/crates/daemon/src/api/history.rs new file mode 100644 index 00000000..bcd2ee5a --- /dev/null +++ b/crates/daemon/src/api/history.rs @@ -0,0 +1,253 @@ +use std::pin::Pin; + +use dashmap::DashMap; +use eyre::Result; +use time::OffsetDateTime; +use tokio_stream::Stream; +use tonic::{Request, Response, Status}; +use tracing::{Level, instrument}; + +use crate::{ + aclient::{ + database::{ClientSqlite, current_context}, + history::store::HistoryStore, + settings::Settings, + }, + daemon::DaemonHandle, + events::DaemonEvent, +}; +use turtle::{ + generated::{ + DAEMON_PROTOCOL_VERSION, + history::{ + EndHistoryReply, EndHistoryRequest, HistoryEntry, HistoryEventKind, HistoryReply, + HistoryRequest, StartHistoryReply, StartHistoryRequest, TailHistoryReply, + TailHistoryRequest, + history_server::{History as HistorySvc, HistoryServer}, + }, + }, + history::{History, HistoryId}, +}; + +/// The gRPC service implementation. +/// +/// This is a thin wrapper that delegates to the component's shared state. +pub(crate) struct HistoryService { + /// Commands currently running (not yet completed). + running: DashMap, + + /// Handle to the daemon (set during start). + handle: DaemonHandle, + + /// History store for pushing records + history_store: HistoryStore, + + history_db: ClientSqlite, +} + +impl HistoryService { + pub(crate) async fn new(handle: DaemonHandle, history_db: ClientSqlite) -> Result { + let host_id = Settings::host_id().await?; + let history_store = + HistoryStore::new(handle.store().clone(), host_id, *handle.encryption_key()); + + Ok(Self { + running: DashMap::new(), + handle, + history_store, + history_db, + }) + } + + /// Get a tonic server for this service. + pub(crate) fn into_server(self) -> HistoryServer { + HistoryServer::new(self) + } +} + +fn history_to_reply(history: History) -> HistoryEntry { + HistoryEntry { + timestamp: history.timestamp.unix_timestamp_nanos() as u64, + id: history.id.0, + command: history.command, + cwd: history.cwd, + session: history.session, + hostname: history.hostname, + author: history.author, + intent: history.intent.unwrap_or_default(), + exit: history.exit, + duration: history.duration, + } +} + +#[tonic::async_trait] +impl HistorySvc for HistoryService { + type TailHistoryStream = Pin> + Send>>; + + #[instrument(skip_all, level = Level::INFO)] + async fn history( + &self, + request: Request, + ) -> Result, Status> { + let req = request.into_inner(); + + let context = current_context(req.session) + .await + .map_err(|e| Status::internal(format!("failed to aquire context: {e:?}")))?; + + let entries = if let Some(range) = req.range { + let from = OffsetDateTime::from_unix_timestamp(range.start as i64).unwrap(); + let to = OffsetDateTime::from_unix_timestamp(range.end as i64).unwrap(); + + self.history_db.range(from, to).await + } else { + self.history_db + .list(&[], &context, None, false, false) + .await + } + .map_err(|e| Status::internal(format!("failed to read db: {e:?}")))? + .into_iter() + .map(history_to_reply) + .collect(); + + Ok(Response::new(HistoryReply { entries })) + } + + #[instrument(skip_all, level = Level::INFO)] + async fn start_history( + &self, + request: Request, + ) -> Result, Status> { + let req = request.into_inner(); + + let timestamp = OffsetDateTime::from_unix_timestamp_nanos(i128::from(req.timestamp)) + .map_err(|_| { + Status::invalid_argument( + "failed to parse timestamp as unix time (expected nanos since epoch)", + ) + })?; + + let h: History = History::daemon() + .timestamp(timestamp) + .command(req.command) + .cwd(req.cwd) + .session(req.session) + .hostname(req.hostname) + .author(req.author) + .intent(req.intent) + .build() + .into(); + + self.handle.emit(DaemonEvent::HistoryStarted(h.clone())); + + let id = h.id.clone(); + tracing::info!(id = id.to_string(), "start history"); + self.running.insert(id.clone(), h); + + let reply = StartHistoryReply { + id: id.to_string(), + version: env!("CARGO_PKG_VERSION").to_string(), + protocol: DAEMON_PROTOCOL_VERSION, + }; + + Ok(Response::new(reply)) + } + + #[instrument(skip_all, level = Level::INFO)] + async fn end_history( + &self, + request: Request, + ) -> Result, Status> { + let req = request.into_inner(); + let id = HistoryId(req.id); + + if let Some((_, mut history)) = self.running.remove(&id) { + history.exit = req.exit; + history.duration = match req.duration { + 0 => i64::try_from( + (OffsetDateTime::now_utc() - history.timestamp).whole_nanoseconds(), + ) + .expect("failed to convert calculated duration to i64"), + value => i64::try_from(value).expect("failed to get i64 duration"), + }; + + self.handle + .history_db() + .save(&history) + .await + .map_err(|e| Status::internal(format!("failed to write to db: {e:?}")))?; + + tracing::info!(id = id.0, duration = history.duration, "end history"); + + let (record_id, idx) = self + .history_store + .push(history.clone()) + .await + .map_err(|e| Status::internal(format!("failed to push record to store: {e:?}")))?; + + self.handle.emit(DaemonEvent::HistoryEnded(history)); + + let reply = EndHistoryReply { + id: record_id.0.to_string(), + idx, + version: env!("CARGO_PKG_VERSION").to_string(), + protocol: DAEMON_PROTOCOL_VERSION, + }; + + return Ok(Response::new(reply)); + } + + Err(Status::not_found(format!( + "could not find history with id: {id}" + ))) + } + + #[instrument(skip_all, level = Level::INFO)] + #[expect(clippy::significant_drop_tightening, reason = "Would be a logic-bug")] + async fn tail_history( + &self, + _request: Request, + ) -> Result, Status> { + let mut rx = self.handle.subscribe(); + let (tx, out_rx) = tokio::sync::mpsc::channel::>(128); + + tokio::spawn(async move { + loop { + let event = match rx.recv().await { + Ok(event) => event, + Err(tokio::sync::broadcast::error::RecvError::Lagged(skipped)) => { + drop( + tx.send(Err(Status::resource_exhausted(format!( + "tail stream lagged behind and dropped {skipped} events" + )))) + .await, + ); + break; + } + Err(tokio::sync::broadcast::error::RecvError::Closed) => break, + }; + + let reply = match event { + DaemonEvent::HistoryStarted(history) => Some(TailHistoryReply { + kind: HistoryEventKind::Started.into(), + history: Some(history_to_reply(history)), + }), + DaemonEvent::HistoryEnded(history) => Some(TailHistoryReply { + kind: HistoryEventKind::Ended.into(), + history: Some(history_to_reply(history)), + }), + _ => None, + }; + + if let Some(reply) = reply + && tx.send(Ok(reply)).await.is_err() + { + break; + } + } + }); + + let stream = tokio_stream::wrappers::ReceiverStream::new(out_rx); + Ok(Response::new(Box::pin(stream))) + } +} diff --git a/crates/daemon/src/api/mod.rs b/crates/daemon/src/api/mod.rs index b7f82a7b..8d475fe9 100644 --- a/crates/daemon/src/api/mod.rs +++ b/crates/daemon/src/api/mod.rs @@ -1,6 +1,2 @@ -pub mod client; -pub(crate) mod generated; -pub(crate) mod server; - -pub(crate) const DAEMON_VERSION: &str = env!("CARGO_PKG_VERSION"); -const DAEMON_PROTOCOL_VERSION: u32 = 1; +pub(crate) mod control; +pub(crate) mod history; diff --git a/crates/daemon/src/api/server/control.rs b/crates/daemon/src/api/server/control.rs deleted file mode 100644 index a5e26355..00000000 --- a/crates/daemon/src/api/server/control.rs +++ /dev/null @@ -1,278 +0,0 @@ -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::{ - ForceSyncReply, ForceSyncRequest, PathsReply, PathsRequest, StatusReply, StatusRequest, - control_server::{Control, ControlServer}, - }, - }, - 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 { - let task_handle = tokio::spawn(sync_loop(handle.clone())); - - Self { - handle, - task_handle, - } - } - - /// Get a tonic server for this service. - pub(crate) fn into_server(self) -> ControlServer { - ControlServer::new(self) - } -} - -#[tonic::async_trait] -impl Control for ControlService { - #[instrument(skip_all, level = Level::INFO)] - async fn paths(&self, _request: Request) -> Result, Status> { - let settings = self.handle.settings().await; - - let config = Settings::get_config_path() - .map_err(|e| Status::internal(format!("failed to get settings path: {e:?}")))?; - - let reply = PathsReply { - config: config.to_string_lossy().to_string(), - db: settings.db_path.clone(), - socket: settings.daemon.socket_path.clone(), - }; - - Ok(Response::new(reply)) - } - - #[instrument(skip_all, level = Level::INFO)] - async fn status( - &self, - _request: Request, - ) -> Result, Status> { - let reply = StatusReply { - healthy: true, - version: DAEMON_VERSION.to_owned(), - pid: std::process::id(), - protocol: DAEMON_PROTOCOL_VERSION, - }; - - Ok(Response::new(reply)) - } - - #[instrument(skip_all, level = Level::INFO)] - async fn force_sync( - &self, - _request: Request, - ) -> Result, Status> { - let reply = ForceSyncReply { accepted: false }; - - 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 - } - } -} diff --git a/crates/daemon/src/api/server/history.rs b/crates/daemon/src/api/server/history.rs deleted file mode 100644 index 0edf3b94..00000000 --- a/crates/daemon/src/api/server/history.rs +++ /dev/null @@ -1,250 +0,0 @@ -use std::pin::Pin; - -use dashmap::DashMap; -use eyre::Result; -use time::OffsetDateTime; -use tokio_stream::Stream; -use tonic::{Request, Response, Status}; -use tracing::{Level, instrument}; - -use crate::{ - aclient::{ - database::{ClientSqlite, current_context}, - history::{History, HistoryId, store::HistoryStore}, - settings::Settings, - }, - api::{ - DAEMON_PROTOCOL_VERSION, - generated::history::{ - EndHistoryReply, EndHistoryRequest, HistoryEntry, HistoryEventKind, HistoryReply, - HistoryRequest, StartHistoryReply, StartHistoryRequest, TailHistoryReply, - TailHistoryRequest, - history_server::{History as HistorySvc, HistoryServer}, - }, - }, - daemon::DaemonHandle, - events::DaemonEvent, -}; - -/// The gRPC service implementation. -/// -/// This is a thin wrapper that delegates to the component's shared state. -pub(crate) struct HistoryService { - /// Commands currently running (not yet completed). - running: DashMap, - - /// Handle to the daemon (set during start). - handle: DaemonHandle, - - /// History store for pushing records - history_store: HistoryStore, - - history_db: ClientSqlite, -} - -impl HistoryService { - pub(crate) async fn new(handle: DaemonHandle, history_db: ClientSqlite) -> Result { - let host_id = Settings::host_id().await?; - let history_store = - HistoryStore::new(handle.store().clone(), host_id, *handle.encryption_key()); - - Ok(Self { - running: DashMap::new(), - handle, - history_store, - history_db, - }) - } - - /// Get a tonic server for this service. - pub(crate) fn into_server(self) -> HistoryServer { - HistoryServer::new(self) - } -} - -fn history_to_reply(history: History) -> HistoryEntry { - HistoryEntry { - timestamp: history.timestamp.unix_timestamp_nanos() as u64, - id: history.id.0, - command: history.command, - cwd: history.cwd, - session: history.session, - hostname: history.hostname, - author: history.author, - intent: history.intent.unwrap_or_default(), - exit: history.exit, - duration: history.duration, - } -} - -#[tonic::async_trait] -impl HistorySvc for HistoryService { - type TailHistoryStream = Pin> + Send>>; - - #[instrument(skip_all, level = Level::INFO)] - async fn history( - &self, - request: Request, - ) -> Result, Status> { - let req = request.into_inner(); - - let context = current_context(req.session) - .await - .map_err(|e| Status::internal(format!("failed to aquire context: {e:?}")))?; - - let entries = if let Some(range) = req.range { - let from = OffsetDateTime::from_unix_timestamp(range.start as i64).unwrap(); - let to = OffsetDateTime::from_unix_timestamp(range.end as i64).unwrap(); - - self.history_db.range(from, to).await - } else { - self.history_db - .list(&[], &context, None, false, false) - .await - } - .map_err(|e| Status::internal(format!("failed to read db: {e:?}")))? - .into_iter() - .map(history_to_reply) - .collect(); - - Ok(Response::new(HistoryReply { entries })) - } - - #[instrument(skip_all, level = Level::INFO)] - async fn start_history( - &self, - request: Request, - ) -> Result, Status> { - let req = request.into_inner(); - - let timestamp = OffsetDateTime::from_unix_timestamp_nanos(i128::from(req.timestamp)) - .map_err(|_| { - Status::invalid_argument( - "failed to parse timestamp as unix time (expected nanos since epoch)", - ) - })?; - - let h: History = History::daemon() - .timestamp(timestamp) - .command(req.command) - .cwd(req.cwd) - .session(req.session) - .hostname(req.hostname) - .author(req.author) - .intent(req.intent) - .build() - .into(); - - self.handle.emit(DaemonEvent::HistoryStarted(h.clone())); - - let id = h.id.clone(); - tracing::info!(id = id.to_string(), "start history"); - self.running.insert(id.clone(), h); - - let reply = StartHistoryReply { - id: id.to_string(), - version: env!("CARGO_PKG_VERSION").to_string(), - protocol: DAEMON_PROTOCOL_VERSION, - }; - - Ok(Response::new(reply)) - } - - #[instrument(skip_all, level = Level::INFO)] - async fn end_history( - &self, - request: Request, - ) -> Result, Status> { - let req = request.into_inner(); - let id = HistoryId(req.id); - - if let Some((_, mut history)) = self.running.remove(&id) { - history.exit = req.exit; - history.duration = match req.duration { - 0 => i64::try_from( - (OffsetDateTime::now_utc() - history.timestamp).whole_nanoseconds(), - ) - .expect("failed to convert calculated duration to i64"), - value => i64::try_from(value).expect("failed to get i64 duration"), - }; - - self.handle - .history_db() - .save(&history) - .await - .map_err(|e| Status::internal(format!("failed to write to db: {e:?}")))?; - - tracing::info!(id = id.0, duration = history.duration, "end history"); - - let (record_id, idx) = self - .history_store - .push(history.clone()) - .await - .map_err(|e| Status::internal(format!("failed to push record to store: {e:?}")))?; - - self.handle.emit(DaemonEvent::HistoryEnded(history)); - - let reply = EndHistoryReply { - id: record_id.0.to_string(), - idx, - version: env!("CARGO_PKG_VERSION").to_string(), - protocol: DAEMON_PROTOCOL_VERSION, - }; - - return Ok(Response::new(reply)); - } - - Err(Status::not_found(format!( - "could not find history with id: {id}" - ))) - } - - #[instrument(skip_all, level = Level::INFO)] - #[expect(clippy::significant_drop_tightening, reason = "Would be a logic-bug")] - async fn tail_history( - &self, - _request: Request, - ) -> Result, Status> { - let mut rx = self.handle.subscribe(); - let (tx, out_rx) = tokio::sync::mpsc::channel::>(128); - - tokio::spawn(async move { - loop { - let event = match rx.recv().await { - Ok(event) => event, - Err(tokio::sync::broadcast::error::RecvError::Lagged(skipped)) => { - drop( - tx.send(Err(Status::resource_exhausted(format!( - "tail stream lagged behind and dropped {skipped} events" - )))) - .await, - ); - break; - } - Err(tokio::sync::broadcast::error::RecvError::Closed) => break, - }; - - let reply = match event { - DaemonEvent::HistoryStarted(history) => Some(TailHistoryReply { - kind: HistoryEventKind::Started.into(), - history: Some(history_to_reply(history)), - }), - DaemonEvent::HistoryEnded(history) => Some(TailHistoryReply { - kind: HistoryEventKind::Ended.into(), - history: Some(history_to_reply(history)), - }), - _ => None, - }; - - if let Some(reply) = reply - && tx.send(Ok(reply)).await.is_err() - { - break; - } - } - }); - - let stream = tokio_stream::wrappers::ReceiverStream::new(out_rx); - Ok(Response::new(Box::pin(stream))) - } -} diff --git a/crates/daemon/src/api/server/mod.rs b/crates/daemon/src/api/server/mod.rs deleted file mode 100644 index 8d475fe9..00000000 --- a/crates/daemon/src/api/server/mod.rs +++ /dev/null @@ -1,2 +0,0 @@ -pub(crate) mod control; -pub(crate) mod history; -- cgit v1.3.1