aboutsummaryrefslogtreecommitdiffstats
path: root/crates/daemon/src/api
diff options
context:
space:
mode:
Diffstat (limited to '')
-rw-r--r--crates/daemon/src/api/client/mod.rs30
-rw-r--r--crates/daemon/src/api/mod.rs6
-rw-r--r--crates/daemon/src/api/server/control.rs203
-rw-r--r--crates/daemon/src/api/server/history.rs71
4 files changed, 264 insertions, 46 deletions
diff --git a/crates/daemon/src/api/client/mod.rs b/crates/daemon/src/api/client/mod.rs
index d6cbbe85..71fa0e37 100644
--- a/crates/daemon/src/api/client/mod.rs
+++ b/crates/daemon/src/api/client/mod.rs
@@ -64,10 +64,37 @@ pub fn classify_error(error: &eyre::Report) -> DaemonClientErrorKind {
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<Channel>,
}
@@ -139,13 +166,14 @@ impl HistoryClient {
// ============================================================================
/// Client for the Control gRPC service.
+#[derive(Debug)]
pub struct ControlClient {
client: ControlServiceClient<Channel>,
}
impl ControlClient {
/// Connect to the daemon's control service.
- pub(crate) async fn new(path: String) -> Result<Self> {
+ pub async fn new(path: String) -> Result<Self> {
let log_path = path.clone();
let channel = Endpoint::try_from("http://atuin_local_daemon:0")?
.connect_with_connector(service_fn(move |_: Uri| {
diff --git a/crates/daemon/src/api/mod.rs b/crates/daemon/src/api/mod.rs
index e6f8f1f0..b7f82a7b 100644
--- a/crates/daemon/src/api/mod.rs
+++ b/crates/daemon/src/api/mod.rs
@@ -1,6 +1,6 @@
-pub(crate) mod client;
-pub(crate) mod server;
+pub mod client;
pub(crate) mod generated;
+pub(crate) mod server;
-const DAEMON_VERSION: &str = env!("CARGO_PKG_VERSION");
+pub(crate) const DAEMON_VERSION: &str = env!("CARGO_PKG_VERSION");
const DAEMON_PROTOCOL_VERSION: u32 = 1;
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
+ }
+ }
+}
diff --git a/crates/daemon/src/api/server/history.rs b/crates/daemon/src/api/server/history.rs
index 032876a6..989c7895 100644
--- a/crates/daemon/src/api/server/history.rs
+++ b/crates/daemon/src/api/server/history.rs
@@ -8,13 +8,16 @@ use tonic::{Request, Response, Status};
use tracing::{Level, instrument};
use crate::{
- aclient::history::{History, HistoryId, store::HistoryStore},
+ aclient::{
+ history::{History, HistoryId, store::HistoryStore},
+ settings::Settings,
+ },
api::{
DAEMON_PROTOCOL_VERSION,
generated::history::{
EndHistoryReply, EndHistoryRequest, HistoryEntry, HistoryEventKind, StartHistoryReply,
StartHistoryRequest, TailHistoryReply, TailHistoryRequest,
- history_server::History as HistorySvc,
+ history_server::{History as HistorySvc, HistoryServer},
},
},
daemon::DaemonHandle,
@@ -24,25 +27,33 @@ use crate::{
/// The gRPC service implementation.
///
/// This is a thin wrapper that delegates to the component's shared state.
-pub(crate) struct HistoryGrpcService {
+pub(crate) struct HistoryService {
/// Commands currently running (not yet completed).
running: DashMap<HistoryId, History>,
/// Handle to the daemon (set during start).
- pub(crate) handle: tokio::sync::RwLock<Option<DaemonHandle>>,
+ pub(crate) handle: DaemonHandle,
/// History store for pushing records (set during start).
- pub(crate) history_store: tokio::sync::RwLock<Option<HistoryStore>>,
+ pub(crate) history_store: HistoryStore,
}
-impl HistoryGrpcService {
- /// Create a new history component.
- pub(crate) fn new() -> Self {
- Self {
+impl HistoryService {
+ pub(crate) async fn new(handle: DaemonHandle) -> Result<Self> {
+ 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: tokio::sync::RwLock::new(None),
- history_store: tokio::sync::RwLock::new(None),
- }
+ handle,
+ history_store,
+ })
+ }
+
+ /// Get a tonic server for this service.
+ pub(crate) fn into_server(self) -> HistoryServer<Self> {
+ HistoryServer::new(self)
}
}
@@ -65,7 +76,7 @@ fn history_to_tail_reply(kind: HistoryEventKind, history: History) -> TailHistor
}
#[tonic::async_trait]
-impl HistorySvc for HistoryGrpcService {
+impl HistorySvc for HistoryService {
type TailHistoryStream = Pin<Box<dyn Stream<Item = Result<TailHistoryReply, Status>> + Send>>;
#[instrument(skip_all, level = Level::INFO)]
@@ -93,10 +104,7 @@ impl HistorySvc for HistoryGrpcService {
.build()
.into();
- // Emit the event
- if let Some(handle) = self.handle.read().await.as_ref() {
- handle.emit(DaemonEvent::HistoryStarted(h.clone()));
- }
+ self.handle.emit(DaemonEvent::HistoryStarted(h.clone()));
let id = h.id.clone();
tracing::info!(id = id.to_string(), "start history");
@@ -130,19 +138,7 @@ impl HistorySvc for HistoryGrpcService {
value => i64::try_from(value).expect("failed to get i64 duration"),
};
- // Get the handle and store to save the history
- let handle_guard = self.handle.read().await;
- let handle = handle_guard
- .as_ref()
- .ok_or_else(|| Status::internal("component not initialized"))?;
-
- let store_guard = self.history_store.read().await;
- let history_store = store_guard
- .as_ref()
- .ok_or_else(|| Status::internal("component not initialized"))?;
-
- // Save to database
- handle
+ self.handle
.history_db()
.save(&history)
.await
@@ -150,14 +146,13 @@ impl HistorySvc for HistoryGrpcService {
tracing::info!(id = id.0, duration = history.duration, "end history");
- // Push to record store
- let (record_id, idx) = history_store
+ let (record_id, idx) = self
+ .history_store
.push(history.clone())
.await
.map_err(|e| Status::internal(format!("failed to push record to store: {e:?}")))?;
- // Emit the event
- handle.emit(DaemonEvent::HistoryEnded(history));
+ self.handle.emit(DaemonEvent::HistoryEnded(history));
let reply = EndHistoryReply {
id: record_id.0.to_string(),
@@ -180,13 +175,7 @@ impl HistorySvc for HistoryGrpcService {
&self,
_request: Request<TailHistoryRequest>,
) -> Result<Response<Self::TailHistoryStream>, Status> {
- let handle_guard = self.handle.read().await;
- let handle = handle_guard
- .as_ref()
- .cloned()
- .ok_or_else(|| Status::internal("component not initialized"))?;
-
- let mut rx = handle.subscribe();
+ let mut rx = self.handle.subscribe();
let (tx, out_rx) = tokio::sync::mpsc::channel::<Result<TailHistoryReply, Status>>(128);
tokio::spawn(async move {
generated by cgit v1.3.1 (git 2.54.0) at 2026-08-28 05:46:33 +0000