diff options
Diffstat (limited to '')
| -rw-r--r-- | crates/daemon/src/lib.rs | 159 |
1 files changed, 115 insertions, 44 deletions
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(()) +} |