beachcomber 0.3.1

A centralized daemon that caches shell state (git, battery, hostname, etc.) so every consumer reads from one fast cache instead of independently forking shells
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136

use crate::cache::Cache;
use crate::config::Config;
use crate::provider::registry::ProviderRegistry;
use crate::scheduler::Scheduler;
use crate::server::Server;
use crate::watcher_registry::WatcherRegistry;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use tokio_util::sync::CancellationToken;
use tracing::info;

pub fn pid_path_for_socket(socket_path: &Path) -> PathBuf {
    socket_path.with_file_name("daemon.pid")
}

pub fn is_daemon_running(socket_path: &Path) -> bool {
    std::os::unix::net::UnixStream::connect(socket_path).is_ok()
}

pub fn start_in_process(socket_path: PathBuf, config: Config) -> tokio::task::JoinHandle<()> {
    let cancel = CancellationToken::new();
    start_in_process_with_cancel(socket_path, config, cancel)
}

pub fn start_in_process_with_cancel(
    socket_path: PathBuf,
    config: Config,
    cancel: CancellationToken,
) -> tokio::task::JoinHandle<()> {
    tokio::spawn(async move {
        run_daemon_with_cancel(socket_path, config, cancel).await;
    })
}

async fn run_daemon_with_cancel(socket_path: PathBuf, config: Config, cancel: CancellationToken) {
    // Load env file before anything else — providers need these vars
    let env_count = config.load_env_file();
    if env_count > 0 {
        info!("Loaded {} environment variables from env file", env_count);
    }

    let watchers = Arc::new(WatcherRegistry::new());
    let cache = Arc::new(Cache::with_watchers(watchers.clone()));
    let registry = Arc::new(ProviderRegistry::with_config(&config));

    let (handle, scheduler) = Scheduler::new(cache.clone(), registry.clone(), config);

    let scheduler_handle = handle.clone();
    let scheduler_task = tokio::spawn(async move { scheduler.run().await });

    let server = Server::new(socket_path, cache, registry, Some(handle), watchers);

    tokio::select! {
        result = server.run() => {
            if let Err(e) = result {
                tracing::error!("Server error: {}", e);
            }
        }
        _ = cancel.cancelled() => {
            info!("Shutdown signal received");
            scheduler_handle.send(crate::scheduler::SchedulerMessage::Shutdown).await;
        }
    }

    let _ = scheduler_task.await;
    info!("Daemon shut down cleanly");
}

pub fn fork_daemon(binary_path: &str, socket_path: &Path) -> std::io::Result<()> {
    use std::process::Command;

    let pid_path = pid_path_for_socket(socket_path);

    if let Some(parent) = socket_path.parent() {
        std::fs::create_dir_all(parent)?;
    }

    // Open log file for the forked daemon's stderr.
    let config = crate::config::Config::load();
    let log_path = config.resolve_log_path();
    if let Some(parent) = log_path.parent() {
        let _ = std::fs::create_dir_all(parent);
    }
    let log_file = std::fs::OpenOptions::new()
        .create(true)
        .append(true)
        .open(&log_path)
        .unwrap_or_else(|_| {
            // Fall back to /dev/null if log file can't be opened.
            std::fs::File::open("/dev/null").unwrap()
        });

    let child = Command::new(binary_path)
        .arg("daemon")
        .arg("--socket")
        .arg(socket_path.as_os_str())
        .stdin(std::process::Stdio::null())
        .stdout(std::process::Stdio::null())
        .stderr(std::process::Stdio::from(log_file))
        .spawn()?;

    std::fs::write(&pid_path, child.id().to_string())?;

    Ok(())
}

pub fn wait_for_daemon(socket_path: &Path, max_attempts: u32) -> bool {
    let mut delay_ms = 10u64;
    for _ in 0..max_attempts {
        if is_daemon_running(socket_path) {
            return true;
        }
        std::thread::sleep(std::time::Duration::from_millis(delay_ms));
        delay_ms = (delay_ms * 2).min(500);
    }
    false
}

pub fn ensure_daemon(socket_path: &Path) -> std::io::Result<()> {
    if is_daemon_running(socket_path) {
        return Ok(());
    }

    let binary = std::env::current_exe()?.to_string_lossy().to_string();

    fork_daemon(&binary, socket_path)?;

    if !wait_for_daemon(socket_path, 8) {
        return Err(std::io::Error::new(
            std::io::ErrorKind::TimedOut,
            "Daemon failed to start within timeout",
        ));
    }

    Ok(())
}