aboutsummaryrefslogtreecommitdiffstats
path: root/crates/daemon
diff options
context:
space:
mode:
authorBenedikt Peetz <benedikt.peetz@b-peetz.de>2026-07-20 15:13:12 +0200
committerBenedikt Peetz <benedikt.peetz@b-peetz.de>2026-07-20 15:13:12 +0200
commit1594307d9cd819b1ed739fa1ec048c6b27e6f4b0 (patch)
tree39b427e3cd6390bc218136454b809450016e0037 /crates/daemon
parentchore: Commit (diff)
downloadatuin-1594307d9cd819b1ed739fa1ec048c6b27e6f4b0.zip
chore: Commit
Diffstat (limited to 'crates/daemon')
-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
-rw-r--r--crates/daemon/src/components/history.rs44
-rw-r--r--crates/daemon/src/components/mod.rs20
-rw-r--r--crates/daemon/src/components/sync.rs285
-rw-r--r--crates/daemon/src/daemon.rs137
-rw-r--r--crates/daemon/src/lib.rs159
-rw-r--r--crates/daemon/src/main.rs223
-rw-r--r--crates/daemon/src/server.rs6
11 files changed, 394 insertions, 790 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 {
diff --git a/crates/daemon/src/components/history.rs b/crates/daemon/src/components/history.rs
deleted file mode 100644
index b476f627..00000000
--- a/crates/daemon/src/components/history.rs
+++ /dev/null
@@ -1,44 +0,0 @@
-//! History component.
-//!
-//! Handles command history lifecycle (start/end) and provides the History gRPC service.
-
-use crate::{
- aclient::{history::store::HistoryStore, settings::Settings},
- api::server::history::HistoryGrpcService,
-};
-use eyre::Result;
-
-use crate::{
- daemon::{Component, DaemonHandle},
- events::DaemonEvent,
-};
-
-#[tonic::async_trait]
-impl Component for HistoryGrpcService {
- fn name(&self) -> &'static str {
- "history"
- }
-
- async fn start(&mut self, handle: DaemonHandle) -> Result<()> {
- // Create the history store
- let host_id = Settings::host_id().await?;
- let history_store =
- HistoryStore::new(handle.store().clone(), host_id, *handle.encryption_key());
-
- *self.history_store.write().await = Some(history_store);
- *self.handle.write().await = Some(handle);
-
- tracing::info!("history component started");
- Ok(())
- }
-
- async fn handle_event(&mut self, _event: &DaemonEvent) -> Result<()> {
- // History component produces events but doesn't need to react to them
- Ok(())
- }
-
- async fn stop(&mut self) -> Result<()> {
- tracing::info!("history component stopped");
- Ok(())
- }
-}
diff --git a/crates/daemon/src/components/mod.rs b/crates/daemon/src/components/mod.rs
deleted file mode 100644
index 0b0319df..00000000
--- a/crates/daemon/src/components/mod.rs
+++ /dev/null
@@ -1,20 +0,0 @@
-//! Daemon components.
-//!
-//! Components are the building blocks of the daemon. Each component handles
-//! a specific domain and can:
-//!
-//! - Expose gRPC services
-//! - React to events
-//! - Spawn background tasks
-//!
-//! Available components:
-//!
-//! - [`history::HistoryComponent`]: Command history lifecycle management
-//! - [`semantic::SemanticComponent`]: In-memory semantic command captures
-//! - [`sync::SyncComponent`]: Cloud sync
-
-pub(crate) mod history;
-pub(crate) mod sync;
-
-pub(crate) use sync::SyncComponent;
-pub(crate) use history::HistoryComponent;
diff --git a/crates/daemon/src/components/sync.rs b/crates/daemon/src/components/sync.rs
deleted file mode 100644
index e200ad73..00000000
--- a/crates/daemon/src/components/sync.rs
+++ /dev/null
@@ -1,285 +0,0 @@
-//! 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 tokio::time::{self, MissedTickBehavior};
-
-use crate::aclient::{history::store::HistoryStore, record::sync, settings::Settings};
-
-use crate::{
- daemon::{Component, 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 {
- /// 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,
-}
-
-/// Sync component - handles periodic cloud synchronization.
-///
-/// 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>>,
-}
-
-impl SyncComponent {
- /// Create a new sync component.
- pub(crate) fn new() -> Self {
- Self {
- task_handle: None,
- command_tx: None,
- }
- }
-}
-
-impl Default for SyncComponent {
- fn default() -> Self {
- Self::new()
- }
-}
-
-#[tonic::async_trait]
-impl Component for SyncComponent {
- fn name(&self) -> &'static str {
- "sync"
- }
-
- async fn start(&mut self, handle: DaemonHandle) -> Result<()> {
- let (cmd_tx, cmd_rx) = mpsc::channel(16);
- self.command_tx = Some(cmd_tx);
-
- // Spawn the sync loop with its own copy of the handle
- self.task_handle = Some(tokio::spawn(sync_loop(handle, cmd_rx)));
-
- tracing::info!("sync component started");
- Ok(())
- }
-
- 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(())
- }
-
- 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(())
- }
-}
-
-/// 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, mut cmd_rx: mpsc::Receiver<SyncCommand>) {
- 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;
-
- 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 = cmd_rx.recv() => {
- match cmd {
- Some(SyncCommand::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;
- }
- Some(SyncCommand::Stop) | None => {
- 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/daemon.rs b/crates/daemon/src/daemon.rs
index 2e236f4e..f3eead19 100644
--- a/crates/daemon/src/daemon.rs
+++ b/crates/daemon/src/daemon.rs
@@ -4,7 +4,6 @@
//!
//! - [`DaemonState`]: Shared state owned by the daemon
//! - [`DaemonHandle`]: A lightweight, cloneable handle for accessing daemon state
-//! - [`Component`]: A trait for implementing daemon components
//! - [`Daemon`]: The main daemon orchestrator
//! - [`DaemonBuilder`]: Builder for constructing and configuring the daemon
@@ -25,7 +24,7 @@ use crate::events::DaemonEvent;
/// Shared state owned by the daemon.
///
-/// This contains all the resources that components and services need access to.
+/// This contains all the resources that services need access to.
/// The state is wrapped in an `Arc` and accessed via [`DaemonHandle`].
pub(crate) struct DaemonState {
// Event bus
@@ -48,7 +47,7 @@ pub(crate) struct DaemonState {
/// A lightweight handle to the daemon's shared state.
///
-/// This is the primary way for components, gRPC services, and spawned tasks to
+/// This is the primary way for gRPC services, and spawned tasks to
/// interact with the daemon. It provides access to:
///
/// - Event emission and subscription
@@ -92,8 +91,6 @@ impl DaemonHandle {
/// Subscribe to the event bus.
///
/// Returns a receiver that will receive all events emitted after this call.
- /// Useful for components that need to listen for events outside of the
- /// normal `handle_event` callback flow.
pub(crate) fn subscribe(&self) -> broadcast::Receiver<DaemonEvent> {
self.state.event_tx.subscribe()
}
@@ -148,92 +145,12 @@ impl std::fmt::Debug for DaemonHandle {
}
// ============================================================================
-// Component Trait
-// ============================================================================
-
-/// A daemon component that handles a specific domain.
-///
-/// Components are the building blocks of the daemon. Each component:
-///
-/// - Has a unique name for logging and debugging
-/// - Can optionally expose gRPC services
-/// - Receives a [`DaemonHandle`] on startup for accessing daemon resources
-/// - Handles events from the event bus
-/// - Performs cleanup on shutdown
-///
-/// # Lifecycle
-///
-/// 1. **Construction**: Component is created (usually via `new()`)
-/// 2. **Start**: `start()` is called with a [`DaemonHandle`]
-/// 3. **Running**: `handle_event()` is called for each event on the bus
-/// 4. **Shutdown**: `stop()` is called for cleanup
-///
-/// # Example
-///
-/// ```ignore
-/// pub(crate) struct MyComponent {
-/// handle: Option<DaemonHandle>,
-/// }
-///
-/// #[async_trait]
-/// impl Component for MyComponent {
-/// fn name(&self) -> &'static str { "my-component" }
-///
-/// async fn start(&mut self, handle: DaemonHandle) -> Result<()> {
-/// self.handle = Some(handle);
-/// Ok(())
-/// }
-///
-/// async fn handle_event(&mut self, event: &DaemonEvent) -> Result<()> {
-/// match event {
-/// DaemonEvent::SomeEvent => {
-/// // Handle the event
-/// if let Some(handle) = &self.handle {
-/// handle.emit(DaemonEvent::ResponseEvent);
-/// }
-/// }
-/// _ => {}
-/// }
-/// Ok(())
-/// }
-///
-/// async fn stop(&mut self) -> Result<()> {
-/// Ok(())
-/// }
-/// }
-/// ```
-#[tonic::async_trait]
-pub(crate) trait Component: Send + Sync {
- /// Human-readable name for logging and debugging.
- fn name(&self) -> &'static str;
-
- /// Called once at startup.
- ///
- /// Store the handle if you need to emit events or access daemon resources
- /// later. The handle is cheaply cloneable, so feel free to clone it for
- /// spawned tasks.
- async fn start(&mut self, handle: DaemonHandle) -> Result<()>;
-
- /// Handle an incoming event.
- ///
- /// Called for every event on the bus. To emit new events in response,
- /// use the handle stored during `start()`. Events emitted here will be
- /// processed in subsequent event loop iterations.
- async fn handle_event(&mut self, event: &DaemonEvent) -> Result<()>;
-
- /// Called on graceful shutdown.
- ///
- /// Use this to clean up resources, abort spawned tasks, etc.
- async fn stop(&mut self) -> Result<()>;
-}
-
-// ============================================================================
// Daemon
// ============================================================================
/// The main daemon orchestrator.
///
-/// The daemon manages components, runs the event loop, and coordinates startup
+/// The daemon runs the event loop, and coordinates startup
/// and shutdown. It is constructed via [`DaemonBuilder`].
///
/// # Event Loop
@@ -241,8 +158,6 @@ pub(crate) trait Component: Send + Sync {
/// The daemon runs a simple event loop:
///
/// 1. Wait for an event on the bus
-/// 2. Dispatch the event to all components (in registration order)
-/// 3. Components may emit new events in response
/// 4. Repeat until `ShutdownRequested` is received
///
/// Events emitted during handling are queued and processed in subsequent
@@ -264,23 +179,9 @@ impl Daemon {
self.handle.clone()
}
- /// Start all components.
- ///
- /// This must be called before `run_event_loop()`. It initializes all
- /// registered components with the daemon handle.
- pub(crate) async fn start_component(&mut self, component: &mut impl Component) -> Result<()> {
- tracing::info!(component = component.name(), "starting component");
- component
- .start(self.handle.clone())
- .await
- .with_context(|| format!("failed to start component: {}", component.name()))?;
- Ok(())
- }
-
/// Run the daemon event loop.
///
/// This processes events until a [`ShutdownRequested`] event is received.
- /// Components must be started first via `start_components()`.
pub(crate) async fn run_event_loop(&mut self) -> Result<()> {
let mut event_rx = self.handle.subscribe();
loop {
@@ -308,33 +209,8 @@ impl Daemon {
Ok(())
}
- /// Stop all components.
- ///
- /// This performs graceful shutdown of all components.
- pub(crate) async fn stop_components(&mut self) {
- for component in &mut self.components {
- tracing::info!(component = component.name(), "stopping component");
- if let Err(e) = component.stop().await {
- tracing::error!(
- component = component.name(),
- error = ?e,
- "error stopping component"
- );
- }
- }
- tracing::info!("all components stopped");
- }
-
async fn dispatch_event(&mut self, event: &DaemonEvent) {
- for component in &mut self.components {
- if let Err(e) = component.handle_event(event).await {
- tracing::error!(
- component = component.name(),
- error = ?e,
- "error handling event"
- );
- }
- }
+ todo!()
}
}
@@ -350,9 +226,6 @@ impl Daemon {
/// let daemon = Daemon::builder(settings)
/// .store(store)
/// .history_db(history_db)
-/// .component(HistoryComponent::new())
-/// .component(SearchComponent::new())
-/// .component(SyncComponent::new())
/// .build()
/// .await?;
///
@@ -362,7 +235,6 @@ pub(crate) struct DaemonBuilder {
settings: Settings,
store: Option<SqliteStore>,
history_db: Option<HistoryDatabase>,
- components: Vec<Arc<Box<dyn Component>>>,
}
impl DaemonBuilder {
@@ -372,7 +244,6 @@ impl DaemonBuilder {
settings,
store: None,
history_db: None,
- components: Vec::new(),
}
}
diff --git a/crates/daemon/src/lib.rs b/crates/daemon/src/lib.rs
index 3be39a1c..4f96e410 100644
--- a/crates/daemon/src/lib.rs
+++ b/crates/daemon/src/lib.rs
@@ -1,83 +1,78 @@
#![expect(unused_crate_dependencies, reason = "Didn't remove them yet")]
-use std::sync::Arc;
+use std::{
+ fs::{self, File, OpenOptions},
+ io::Write,
+ path::{Path, PathBuf},
+ time::{Duration, Instant},
+};
+
+use eyre::{Context, Result, bail, eyre};
+use fs4::fs_std::FileExt;
+use tokio::time::sleep;
-use crate::aclient::record::sqlite_store::SqliteStore;
-use crate::aclient::settings::Settings;
-use crate::api::server::control::ControlService;
use crate::{
- aclient::database::ClientSqlite as HistoryDatabase,
- api::generated::history::history_server::HistoryServer,
+ aclient::{
+ database::ClientSqlite as HistoryDatabase, record::sqlite_store::SqliteStore,
+ settings::Settings,
+ },
+ api::{
+ DAEMON_VERSION,
+ server::{control::ControlService, history::HistoryService},
+ },
+ daemon::Daemon,
};
-use eyre::Result;
pub mod aclient;
-pub(crate) mod api;
-pub(crate) mod components;
+pub mod api;
pub(crate) mod daemon;
pub(crate) mod events;
pub(crate) mod server;
-// Re-export core daemon types for convenience
-pub(crate) use daemon::Daemon;
-pub use events::DaemonEvent;
-
-// Re-export components
-pub(crate) use components::{HistoryComponent, SyncComponent};
-
-/// Boot the daemon using the new component-based architecture.
+/// Boot the daemon.
///
-/// This creates a daemon with the standard components (history, search, sync),
-/// starts the gRPC server with their services, and runs the event loop.
+/// This creates a daemon,
+/// starts the gRPC server with services, and runs the event loop.
pub async fn boot(
settings: Settings,
store: SqliteStore,
history_db: HistoryDatabase,
) -> Result<()> {
- // Create the components
- let history_component = Arc::new(HistoryComponent::new());
- let sync_component = Arc::new(Box::new(&SyncComponent::new()));
+ let pidfile_path = PathBuf::from(&settings.daemon.pidfile_path);
+ let _pidfile_guard = PidfileGuard::acquire(&pidfile_path)?;
- // Get the gRPC services before moving components into the daemon
- // (The services share state with the components via Arc)
- let history_service = HistoryServer::from_arc(Arc::clone(&history_component));
-
- // Build the daemon
let mut daemon = Daemon::builder(settings.clone())
.store(store)
.history_db(history_db)
- .component(history_component)
- .component(sync_component)
.build()?;
- let handle = daemon.handle();
+ let handle = {
+ let handle = daemon.handle();
- let control_service = ControlService::new(handle.clone());
+ // Spawn signal handler to emit ShutdownRequested on Ctrl+C/SIGTERM
+ let signal_handle = handle.clone();
+ tokio::spawn(async move {
+ shutdown_signal().await;
+ tracing::info!("received shutdown signal");
+ signal_handle.shutdown();
+ });
- // Start all components first (so gRPC services can work)
- daemon.start_components().await?;
+ handle
+ };
- // Spawn signal handler to emit ShutdownRequested on Ctrl+C/SIGTERM
- let signal_handle = handle.clone();
- tokio::spawn(async move {
- shutdown_signal().await;
- tracing::info!("received shutdown signal");
- signal_handle.shutdown();
- });
+ let history_service = HistoryService::new(handle.clone()).await?;
+ let control_service = ControlService::new(handle.clone());
server::run_grpc_server(
&settings,
- history_service,
+ history_service.into_server(),
control_service.into_server(),
handle,
)?;
daemon.run_event_loop().await?;
- // After the event loop exited, we shut-down the components.
- daemon.stop_components().await;
-
tracing::info!("daemon shut down complete");
Ok(())
}
@@ -95,3 +90,79 @@ async fn shutdown_signal() {
_ = int.recv() => {},
}
}
+
+struct PidfileGuard {
+ file: File,
+}
+
+impl PidfileGuard {
+ fn acquire(path: &Path) -> Result<Self> {
+ let mut file = open_lock_file(path)?;
+
+ if !file.try_lock_exclusive()? {
+ bail!(
+ "daemon already running (pidfile lock busy at {})",
+ path.display()
+ );
+ }
+
+ file.set_len(0)
+ .wrap_err_with(|| format!("could not truncate daemon pidfile {}", path.display()))?;
+ writeln!(file, "{}", std::process::id())
+ .and_then(|()| writeln!(file, "{DAEMON_VERSION}"))
+ .wrap_err_with(|| format!("could not write daemon pidfile {}", path.display()))?;
+
+ Ok(Self { file })
+ }
+}
+
+impl Drop for PidfileGuard {
+ fn drop(&mut self) {
+ drop(self.file.unlock());
+ }
+}
+
+fn open_lock_file(path: &Path) -> Result<File> {
+ if let Some(parent) = path.parent() {
+ fs::create_dir_all(parent)
+ .wrap_err_with(|| format!("could not create lock directory {}", parent.display()))?;
+ }
+
+ OpenOptions::new()
+ .read(true)
+ .write(true)
+ .create(true)
+ .truncate(false)
+ .open(path)
+ .wrap_err_with(|| format!("could not open lock file {}", path.display()))
+}
+
+async fn wait_for_lock(path: &Path, timeout: Duration) -> Result<File> {
+ const LOCK_POLL: Duration = Duration::from_millis(20);
+
+ let file = open_lock_file(path)?;
+ let start = Instant::now();
+
+ loop {
+ match file.try_lock_exclusive() {
+ Ok(true) => return Ok(file),
+ Ok(false) => {
+ if start.elapsed() >= timeout {
+ bail!("timed out waiting for lock at {}", path.display());
+ }
+
+ sleep(LOCK_POLL).await;
+ }
+ Err(err) => {
+ return Err(eyre!("could not lock {}: {err}", path.display()));
+ }
+ }
+ }
+}
+
+async fn wait_for_pidfile_available(path: &Path, timeout: Duration) -> Result<()> {
+ let file = wait_for_lock(path, timeout).await?;
+ file.unlock()
+ .wrap_err_with(|| format!("failed to unlock {}", path.display()))?;
+ Ok(())
+}
diff --git a/crates/daemon/src/main.rs b/crates/daemon/src/main.rs
index 174cf94b..1966e113 100644
--- a/crates/daemon/src/main.rs
+++ b/crates/daemon/src/main.rs
@@ -1,17 +1,12 @@
#![expect(unused_crate_dependencies, reason = "Didn't remove them yet")]
use clap::Parser;
-use eyre::{Result, WrapErr, bail, eyre};
-use fs4::fs_std::FileExt;
-use std::fs::{self, File, OpenOptions};
-use std::io::Write;
-use std::path::{Path, PathBuf};
-use std::time::{Duration, Instant};
-use tokio::time::sleep;
-use turtle_daemon::{
- aclient::{database::ClientSqlite, record::sqlite_store::SqliteStore, settings::Settings},
- client::{DaemonClientErrorKind, HistoryClient, classify_error},
+use eyre::{Result, WrapErr};
+use std::path::PathBuf;
+use turtle_daemon::aclient::{
+ database::ClientSqlite, record::sqlite_store::SqliteStore, settings::Settings,
};
+use turtle_daemon::api::client::{Probe, probe};
#[derive(Parser, Debug)]
#[command(infer_subcommands = true)]
@@ -22,12 +17,6 @@ pub(crate) enum Cmd {
#[arg(long)]
show_logs: bool,
},
-
- /// Show the daemon's current status
- Status,
-
- /// Stop the daemon gracefully
- Stop,
}
#[tokio::main]
@@ -36,208 +25,12 @@ async fn main() -> Result<()> {
let db_path = PathBuf::from(settings.db_path.as_str());
let record_store_path = PathBuf::from(settings.record_store_path.as_str());
- let db = ClientSqlite::new(db_path, settings.local_timeout).await?;
+ let history_db = ClientSqlite::new(db_path, settings.local_timeout).await?;
let sqlite_store = SqliteStore::new(record_store_path, settings.local_timeout).await?;
match Cmd::parse() {
- Cmd::Start { show_logs, .. } => start_cmd(settings, store, history_db, show_logs).await,
- Cmd::Status => status_cmd(&settings).await,
- Cmd::Stop => stop_cmd(&settings).await,
- }
-}
-
-const STARTUP_POLL: Duration = Duration::from_millis(40);
-const LEGACY_DAEMON_RESTART_MESSAGE: &str = "legacy daemon detected; restart daemon manually";
-
-struct PidfileGuard {
- file: File,
-}
-
-impl PidfileGuard {
- fn acquire(path: &Path) -> Result<Self> {
- let mut file = open_lock_file(path)?;
-
- if !file.try_lock_exclusive()? {
- bail!(
- "daemon already running (pidfile lock busy at {})",
- path.display()
- );
- }
-
- file.set_len(0)
- .wrap_err_with(|| format!("could not truncate daemon pidfile {}", path.display()))?;
- writeln!(file, "{}", std::process::id())
- .and_then(|()| writeln!(file, "{DAEMON_VERSION}"))
- .wrap_err_with(|| format!("could not write daemon pidfile {}", path.display()))?;
-
- Ok(Self { file })
- }
-}
-
-impl Drop for PidfileGuard {
- fn drop(&mut self) {
- drop(self.file.unlock());
- }
-}
-
-enum Probe {
- Ready(HistoryClient),
- NeedsRestart(String),
- Unreachable(eyre::Report),
-}
-
-fn is_legacy_daemon_error(err: &eyre::Report) -> bool {
- matches!(classify_error(err), DaemonClientErrorKind::Unimplemented)
-}
-
-fn open_lock_file(path: &Path) -> Result<File> {
- if let Some(parent) = path.parent() {
- fs::create_dir_all(parent)
- .wrap_err_with(|| format!("could not create lock directory {}", parent.display()))?;
- }
-
- OpenOptions::new()
- .read(true)
- .write(true)
- .create(true)
- .truncate(false)
- .open(path)
- .wrap_err_with(|| format!("could not open lock file {}", path.display()))
-}
-
-async fn wait_for_lock(path: &Path, timeout: Duration) -> Result<File> {
- const LOCK_POLL: Duration = Duration::from_millis(20);
-
- let file = open_lock_file(path)?;
- let start = Instant::now();
-
- loop {
- match file.try_lock_exclusive() {
- Ok(true) => return Ok(file),
- Ok(false) => {
- if start.elapsed() >= timeout {
- bail!("timed out waiting for lock at {}", path.display());
- }
-
- sleep(LOCK_POLL).await;
- }
- Err(err) => {
- return Err(eyre!("could not lock {}: {err}", path.display()));
- }
- }
- }
-}
-
-async fn wait_for_pidfile_available(path: &Path, timeout: Duration) -> Result<()> {
- let file = wait_for_lock(path, timeout).await?;
- file.unlock()
- .wrap_err_with(|| format!("failed to unlock {}", path.display()))?;
- Ok(())
-}
-
-async fn request_shutdown(settings: &Settings) {
- if let Ok(mut client) = connect_client(settings).await {
- drop(client.shutdown().await);
- }
-}
-
-fn startup_timeout(settings: &Settings) -> Duration {
- Duration::from_secs_f64(settings.local_timeout.max(0.5) + 2.0)
-}
-
-async fn status_cmd(settings: &Settings) -> Result<()> {
- match probe(settings).await {
- Probe::Ready(mut client) => {
- let status = client.status().await?;
- println!("Daemon running");
- println!(" PID: {}", status.pid);
- println!(" Version: {}", status.version);
- println!(" Protocol: {}", status.protocol);
- println!(" Healthy: {}", status.healthy);
- println!(" Socket: {}", settings.daemon.socket_path);
- }
- Probe::NeedsRestart(reason) => {
- println!("Daemon running (needs restart)");
- println!(" Reason: {reason}");
- }
- Probe::Unreachable(_) => {
- println!("Daemon is not running");
- }
- }
-
- Ok(())
-}
-
-async fn stop_cmd(settings: &Settings) -> Result<()> {
- let Ok(mut client) = connect_client(settings).await else {
- println!("Daemon is not running");
- return Ok(());
- };
-
- match client.shutdown().await {
- Ok(true) => {
- println!("Shutdown requested");
-
- let pidfile_path = PathBuf::from(&settings.daemon.pidfile_path);
- let timeout = Duration::from_secs(5);
- match wait_for_pidfile_available(&pidfile_path, timeout).await {
- Ok(()) => println!("Daemon stopped"),
- Err(_) => println!("Daemon may still be shutting down"),
- }
-
- Ok(())
+ Cmd::Start { show_logs, .. } => {
+ turtle_daemon::boot(settings, sqlite_store, history_db).await
}
- Ok(false) => bail!("Daemon rejected shutdown request"),
- Err(err) => Err(err.wrap_err("Failed to send shutdown request")),
- }
-}
-
-async fn start_cmd(
- settings: Settings,
- store: SqliteStore,
- history_db: ClientSqlite,
- show_logs: bool,
-) -> Result<()> {
- let pidfile_path = PathBuf::from(&settings.daemon.pidfile_path);
- let _pidfile_guard = PidfileGuard::acquire(&pidfile_path)?;
-
- turtle_daemon::boot(settings, store, history_db).await?;
-
- Ok(())
-}
-
-#[cfg(test)]
-mod tests {
- use super::{
- DAEMON_PROTOCOL_VERSION, DAEMON_VERSION, daemon_matches_expected, daemon_mismatch_message,
- };
-
- #[test]
- fn test_version_matches() {
- assert!(daemon_matches_expected(
- DAEMON_VERSION,
- DAEMON_PROTOCOL_VERSION
- ));
- }
-
- #[test]
- fn test_version_mismatch() {
- assert!(!daemon_matches_expected("0.0.0", DAEMON_PROTOCOL_VERSION));
- assert!(!daemon_matches_expected(DAEMON_VERSION, 999));
- assert!(!daemon_matches_expected("0.0.0", 999));
- }
-
- #[test]
- fn test_mismatch_message_version() {
- let msg = daemon_mismatch_message("0.0.0", DAEMON_PROTOCOL_VERSION);
- assert!(msg.contains("out of date"), "got: {msg}");
- assert!(msg.contains("0.0.0"));
- assert!(msg.contains(DAEMON_VERSION));
- }
-
- #[test]
- fn test_mismatch_message_protocol() {
- let msg = daemon_mismatch_message(DAEMON_VERSION, 999);
- assert!(msg.contains("protocol mismatch"), "got: {msg}");
}
}
diff --git a/crates/daemon/src/server.rs b/crates/daemon/src/server.rs
index 7ba8312c..3400ad62 100644
--- a/crates/daemon/src/server.rs
+++ b/crates/daemon/src/server.rs
@@ -5,7 +5,7 @@ use eyre::Result;
use eyre::{OptionExt, WrapErr};
#[cfg(unix)]
-use crate::api::server::{control::ControlService, history::HistoryGrpcService};
+use crate::api::server::{control::ControlService, history::HistoryService};
use crate::{
aclient::settings::Settings,
api::generated::{
@@ -21,7 +21,7 @@ use crate::{
#[cfg(unix)]
pub(crate) fn run_grpc_server(
settings: &Settings,
- history_service: HistoryServer<HistoryGrpcService>,
+ history_service: HistoryServer<HistoryService>,
control_service: ControlServer<ControlService>,
handle: DaemonHandle,
) -> Result<()> {
@@ -72,7 +72,7 @@ pub(crate) fn run_grpc_server(
let mut rx = handle.subscribe();
loop {
- use crate::DaemonEvent;
+ use crate::events::DaemonEvent;
match rx.recv().await {
Err(_) | Ok(DaemonEvent::ShutdownRequested) => break,