use std::net::SocketAddr;
use std::path::{Path, PathBuf};
use std::time::Duration;
use anyhow::{Context, Result, bail};
use sqlx::SqlitePool;
use tokio::net::UdpSocket;
use tokio::time::{Instant, sleep_until, timeout};
#[cfg(unix)]
use std::os::unix::fs::MetadataExt;
use tracing::{debug, info, warn};
use crate::config::AppConfig;
use crate::db::open_db;
use crate::signals::shutdown_signal;
use crate::sync::SyncHttpClient;
use crate::sync::wire::{DAEMON_INCOMPLETE_RESCHEDULE_MS, DAEMON_SYNC_PAGE_BUDGET};
mod service;
pub use service::{
ServiceInstallArgs, ServiceRepairArgs, install, repair, restart, status_snapshot, uninstall,
};
const BINARY_CHECK_INTERVAL: Duration = Duration::from_secs(30);
#[derive(Clone, Debug, PartialEq, Eq)]
struct BinaryFingerprint {
path: PathBuf,
len: u64,
modified_ns: Option<u128>,
#[cfg(unix)]
dev: u64,
#[cfg(unix)]
ino: u64,
}
pub struct DaemonRunArgs {
pub db_path: PathBuf,
pub config: AppConfig,
}
pub async fn run(args: DaemonRunArgs) -> Result<()> {
if !args.config.sync.enabled {
bail!("error sync-disabled hint=\"set sync.enabled = true in config.yaml\"");
}
let server = args
.config
.sync
.server_url
.clone()
.context("error sync-server-required hint=\"set sync.server_url in config.yaml\"")?;
let wake_addr = args.config.wake_addr()?;
let interval_seconds = args.config.sync_interval_seconds();
let pool = open_db(&args.db_path).await?;
let socket = UdpSocket::bind(wake_addr).await.with_context(|| {
format!("could not bind daemon wake address {wake_addr}; is another daemon running?")
})?;
info!(
db = %args.db_path.display(),
server = %server,
wake_addr = %wake_addr,
interval_seconds,
"daemon starting"
);
println!(
"daemon db={} server={} wake={}",
args.db_path.display(),
server,
wake_addr
);
let binary_fingerprint = current_binary_fingerprint()?;
let client = SyncHttpClient::new().context("build daemon sync HTTP client")?;
info!(server = %server, http_client_id = %client.id(), "daemon sync client ready");
run_loop(
pool,
server,
socket,
interval_seconds,
args.config.sync_auth_token().map(str::to_string),
client,
binary_fingerprint,
)
.await
}
async fn run_loop(
pool: SqlitePool,
server: String,
socket: UdpSocket,
interval_seconds: u64,
auth_token: Option<String>,
client: SyncHttpClient,
binary_fingerprint: BinaryFingerprint,
) -> Result<()> {
let mut wake_buf = [0_u8; 16];
let mut backoff_seconds = 1_u64;
let mut next_sync = Instant::now();
let mut next_binary_check = Instant::now() + BINARY_CHECK_INTERVAL;
loop {
tokio::select! {
_ = shutdown_signal() => {
info!("daemon shutting down");
break;
}
result = socket.recv_from(&mut wake_buf) => {
if let Err(err) = result {
warn!(error = %err, "daemon wake receive failed");
eprintln!("daemon wake failed: {err}");
} else {
debug!("daemon wake received");
}
drain_wakes(&socket, &mut wake_buf);
next_sync = Instant::now();
}
_ = sleep_until(next_binary_check) => {
if binary_changed(&binary_fingerprint)? {
info!(path = %binary_fingerprint.path.display(), "daemon executable changed");
println!("daemon-executable-changed path={}", binary_fingerprint.path.display());
break;
}
next_binary_check = Instant::now() + BINARY_CHECK_INTERVAL;
}
_ = sleep_until(next_sync) => {
match timeout(
Duration::from_secs(35),
sync_once(&pool, &server, auth_token.as_deref(), &client),
)
.await
{
Ok(Ok(summary)) => {
backoff_seconds = 1;
next_sync = if summary.complete {
Instant::now() + Duration::from_secs(interval_seconds)
} else {
Instant::now() + Duration::from_millis(DAEMON_INCOMPLETE_RESCHEDULE_MS)
};
}
Ok(Err(err)) => {
warn!(error = %err, backoff_seconds, "daemon sync failed");
eprintln!("daemon sync failed: {err}");
next_sync = Instant::now() + Duration::from_secs(backoff_seconds);
backoff_seconds = (backoff_seconds * 2).min(300);
}
Err(_) => {
warn!(backoff_seconds, "daemon sync timed out");
eprintln!("daemon sync failed: timed out");
next_sync = Instant::now() + Duration::from_secs(backoff_seconds);
backoff_seconds = (backoff_seconds * 2).min(300);
}
}
}
}
}
Ok(())
}
fn current_binary_fingerprint() -> Result<BinaryFingerprint> {
let path = std::env::current_exe().context("resolve current executable")?;
binary_fingerprint(&path)
}
fn binary_changed(initial: &BinaryFingerprint) -> Result<bool> {
Ok(binary_fingerprint(&initial.path)? != *initial)
}
fn binary_fingerprint(path: &Path) -> Result<BinaryFingerprint> {
let path = path.canonicalize().unwrap_or_else(|_| path.to_path_buf());
let metadata = std::fs::metadata(&path)
.with_context(|| format!("read executable metadata {}", path.display()))?;
let modified_ns = metadata
.modified()
.ok()
.and_then(|modified| modified.duration_since(std::time::UNIX_EPOCH).ok())
.map(|duration| duration.as_nanos());
Ok(BinaryFingerprint {
path,
len: metadata.len(),
modified_ns,
#[cfg(unix)]
dev: metadata.dev(),
#[cfg(unix)]
ino: metadata.ino(),
})
}
fn drain_wakes(socket: &UdpSocket, wake_buf: &mut [u8]) {
while socket.try_recv_from(wake_buf).is_ok() {}
}
async fn sync_once(
pool: &SqlitePool,
server: &str,
auth_token: Option<&str>,
client: &SyncHttpClient,
) -> Result<crate::sync::SyncSummary> {
let mut conn = pool.acquire().await?;
let summary = crate::sync::run_sync_with_page_budget_using_client(
&mut conn,
server,
auth_token,
Some(DAEMON_SYNC_PAGE_BUDGET),
client,
)
.await?;
info!(
pushed = summary.pushed,
pulled = summary.pulled,
cursor = summary.cursor,
complete = summary.complete,
pages = summary.pages,
request_bytes = summary.request_bytes,
request_wire_bytes = summary.request_wire_bytes,
response_decoded_bytes = summary.response_decoded_bytes,
response_compression = summary.response_compression,
apply_ms = summary.apply_ms,
"daemon sync completed"
);
println!(
"daemon-synced pushed={} pulled={} cursor={} complete={} pages={}",
summary.pushed, summary.pulled, summary.cursor, summary.complete, summary.pages
);
Ok(summary)
}
pub fn wake(addr: SocketAddr) {
let bind_addr = SocketAddr::new(addr.ip(), 0);
match std::net::UdpSocket::bind(bind_addr).and_then(|socket| socket.send_to(b"1", addr)) {
Ok(_) => debug!(wake_addr = %addr, "daemon wake sent"),
Err(err) => warn!(wake_addr = %addr, error = %err, "daemon wake send failed"),
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn binary_fingerprint_changes_when_file_changes() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("aven");
std::fs::write(&path, "one").unwrap();
let initial = binary_fingerprint(&path).unwrap();
std::thread::sleep(Duration::from_millis(5));
std::fs::write(&path, "two-two").unwrap();
assert!(binary_changed(&initial).unwrap());
}
}