scv_server/
daemon.rs

1//! The daemon: the socket listener and its lock, supervision of components
2//! and delegated runs, and shutdown. [`run_stdio`] serves one connection
3//! without a daemon.
4
5use std::{path::Path, sync::Arc, time::Duration};
6
7use anyhow::{Context, Result, anyhow};
8use scv_client::Layout;
9use scv_tools::delegation::{self as delegations, DelegationRegistry};
10use tokio::{
11    net::{UnixListener, UnixStream},
12    sync::Mutex,
13};
14use tokio_util::{sync::CancellationToken, task::TaskTracker};
15
16use crate::{
17    components,
18    config::{ConfigOverrides, Instance},
19    confirm,
20    connection::run_managed,
21    disk, restart,
22};
23
24/// Serve one session over stdin and stdout for the instance at `layout`.
25pub async fn run_stdio(layout: &Layout, overrides: ConfigOverrides) -> Result<()> {
26    let stdin = tokio::io::stdin();
27    let stdout = tokio::io::stdout();
28    let tasks = TaskTracker::new();
29    let registry = instance_delegations(layout);
30    // Without a daemon, a later `scv exec` is what cleans up after an earlier
31    // one that was killed; this runs alongside the session.
32    tokio::spawn(reconcile_delegations(Arc::clone(&registry)));
33    let result = run_managed(
34        stdin,
35        stdout,
36        Instance {
37            layout: layout.clone(),
38            overrides,
39        },
40        None,
41        registry,
42        CancellationToken::new(),
43        tasks.clone(),
44    )
45    .await;
46    tasks.close();
47    tasks.wait().await;
48    result
49}
50
51/// Run the authoritative server on the instance's Unix socket.
52pub async fn run_socket(layout: &Layout, overrides: ConfigOverrides) -> Result<()> {
53    let socket = layout.socket();
54    let path = socket.as_path();
55    let instance = Instance {
56        layout: layout.clone(),
57        overrides,
58    };
59    if let Some(parent) = path.parent() {
60        tokio::fs::create_dir_all(parent)
61            .await
62            .context("create SCV socket directory")?;
63        #[cfg(unix)]
64        {
65            use std::os::unix::fs::PermissionsExt;
66            std::fs::set_permissions(parent, std::fs::Permissions::from_mode(0o700))
67                .context("secure SCV socket directory")?;
68        }
69    }
70    let _lock = SocketLock::acquire(path)?;
71    for warning in instance.claude_warnings() {
72        tracing::warn!("{warning}");
73    }
74    // Nothing reads an older release's files; say so once rather than let
75    // them look like live configuration.
76    if let Ok(strays) = layout.strays() {
77        for stray in strays.into_iter().filter(|stray| stray.legacy) {
78            tracing::warn!(
79                "{} is from an older SCV layout and is not used; see `scv config show`",
80                stray.path.display()
81            );
82        }
83    }
84    if path.exists() {
85        if UnixStream::connect(path).await.is_ok() {
86            return Err(anyhow!(
87                "SCV server is already running at {}",
88                path.display()
89            ));
90        }
91        use std::os::unix::fs::FileTypeExt;
92        if !std::fs::symlink_metadata(path)?.file_type().is_socket() {
93            return Err(anyhow!(
94                "refusing to remove a non-socket at SCV socket path"
95            ));
96        }
97        tokio::fs::remove_file(path)
98            .await
99            .with_context(|| format!("remove stale SCV socket {}", path.display()))?;
100    }
101    let listener = UnixListener::bind(path)
102        .with_context(|| format!("bind SCV server socket {}", path.display()))?;
103    #[cfg(unix)]
104    {
105        use std::os::unix::fs::PermissionsExt;
106        std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o600))
107            .context("secure SCV socket")?;
108    }
109    let hub = scv_channels::hub::Hub::new(Some(layout.last_owner()));
110    // Before any account starts: its recovery needs to know whether this
111    // start is a planned restart.
112    let startup = restart::startup(layout, &hub);
113    let components = Arc::new(Mutex::new(components::Components::with_hub(
114        instance.clone(),
115        std::env::current_dir()?,
116        Arc::clone(&hub),
117    )?));
118    let registry = instance_delegations(layout);
119    // Descendants a delegated agent leaves behind reparent to the daemon, not init.
120    if !delegations::become_child_subreaper() {
121        tracing::debug!("SCV daemon is not a child subreaper on this platform");
122    }
123    let cancellation = CancellationToken::new();
124    let restarter = restart::Restarter::new(
125        instance.clone(),
126        Arc::clone(&hub),
127        Arc::clone(&registry),
128        &components,
129        cancellation.clone(),
130    );
131    let disk = tokio::spawn(disk::monitor(
132        instance.clone(),
133        Arc::clone(&hub),
134        restarter.notifier().clone(),
135        cancellation.clone(),
136    ));
137    let _disk_abort = AbortGuard(disk.abort_handle());
138    let confirmer = confirm::Confirmer::new(
139        hub,
140        Arc::clone(&registry),
141        restarter.notifier().clone(),
142        cancellation.clone(),
143    );
144    {
145        let mut components = components.lock().await;
146        components.set_restarter(Arc::clone(&restarter));
147        components.set_confirmer(confirmer);
148    }
149    let notices = tokio::spawn(restart::announce(
150        layout.clone(),
151        startup,
152        restarter.notifier().clone(),
153        cancellation.clone(),
154    ));
155    let _notices_abort = AbortGuard(notices.abort_handle());
156    let monitor = tokio::spawn(restart::monitor(
157        restarter.notifier().clone(),
158        cancellation.clone(),
159    ));
160    let _monitor_abort = AbortGuard(monitor.abort_handle());
161    let delegation_registry = Arc::clone(&registry);
162    let delegation_cancel = cancellation.clone();
163    let mut delegation_task = tokio::spawn(async move {
164        // The first tick is immediate: orphans from before a restart go first.
165        let mut interval = tokio::time::interval(DELEGATION_RECONCILE_INTERVAL);
166        loop {
167            tokio::select! {
168                biased;
169                () = delegation_cancel.cancelled() => break,
170                _ = interval.tick() => {
171                    reconcile_delegations(Arc::clone(&delegation_registry)).await;
172                    let zombies = delegations::reap_orphaned_zombies();
173                    if zombies > 0 {
174                        tracing::debug!("Reaped {zombies} exited orphan processes");
175                    }
176                }
177            }
178        }
179    });
180    let _delegation_abort = AbortGuard(delegation_task.abort_handle());
181    let tasks = TaskTracker::new();
182    let mut clients = tokio::task::JoinSet::new();
183    let refresh_components = components.clone();
184    let refresh_cancel = cancellation.clone();
185    let mut refresh_task = tokio::spawn(async move {
186        let mut refresh = tokio::time::interval(Duration::from_secs(2));
187        loop {
188            tokio::select! {
189                biased;
190                () = refresh_cancel.cancelled() => break,
191                _ = refresh.tick() => {
192                    tokio::select! {
193                        biased;
194                        () = refresh_cancel.cancelled() => break,
195                        result = async { refresh_components.lock().await.reconcile().await } => {
196                            if result.is_err() { tracing::warn!("Component account discovery failed"); }
197                        }
198                    }
199                }
200            }
201        }
202    });
203    let _refresh_abort = AbortGuard(refresh_task.abort_handle());
204    let mut terminate = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())?;
205    let result = loop {
206        tokio::select! {
207            accepted = listener.accept() => {
208                let (stream, _) = match accepted { Ok(value) => value, Err(error) => break Err(error.into()) };
209                let instance = instance.clone();
210                let components = components.clone();
211                let registry = Arc::clone(&registry);
212                let cancellation = cancellation.clone();
213                let tasks = tasks.clone();
214                clients.spawn(async move {
215                    let (reader, writer) = stream.into_split();
216                    if let Err(error) = run_managed(reader, writer, instance, Some(components), registry, cancellation, tasks).await {
217                        tracing::warn!(error = format!("{error:#}"), "SCV socket client stopped");
218                    }
219                });
220            }
221            _ = clients.join_next(), if !clients.is_empty() => {},
222            _ = tokio::signal::ctrl_c() => break Ok(()),
223            _ = terminate.recv() => break Ok(()),
224        }
225    };
226    drop(listener);
227    cancellation.cancel();
228    let _ = (&mut refresh_task).await;
229    let _ = (&mut delegation_task).await;
230    components.lock().await.shutdown().await;
231    if tokio::time::timeout(Duration::from_secs(8), async {
232        while clients.join_next().await.is_some() {}
233    })
234    .await
235    .is_err()
236    {
237        clients.abort_all();
238        while clients.join_next().await.is_some() {}
239    }
240    tasks.close();
241    tasks.wait().await;
242    let _ = tokio::fs::remove_file(path).await;
243    restart::clean_shutdown(layout);
244    result
245}
246
247pub(crate) const DELEGATION_RECONCILE_INTERVAL: Duration = Duration::from_secs(60);
248
249/// The delegation registry for this process's SCV instance.
250pub(crate) fn instance_delegations(layout: &Layout) -> Arc<DelegationRegistry> {
251    Arc::new(DelegationRegistry::new(layout))
252}
253
254/// Stop orphaned delegations of this instance and log what was stopped.
255pub(crate) async fn reconcile_delegations(registry: Arc<DelegationRegistry>) {
256    let report = registry.reconcile().await;
257    if !report.reaped.is_empty() {
258        tracing::info!(
259            "Reaped {} orphaned delegations: {}",
260            report.reaped.len(),
261            report.reaped.join(", ")
262        );
263    }
264    if report.removed > 0 {
265        tracing::debug!(
266            "Removed {} delegation records whose processes had exited",
267            report.removed
268        );
269    }
270    if report.stale_markers > 0 {
271        tracing::debug!(
272            "Removed {} conversation markers whose SCV process had exited",
273            report.stale_markers
274        );
275    }
276}
277
278/// A persistent advisory lock closes the stale-socket unlink/bind race.
279pub(crate) struct SocketLock(std::fs::File);
280impl SocketLock {
281    pub(crate) fn acquire(socket: &Path) -> Result<Self> {
282        use std::os::unix::{fs::OpenOptionsExt, io::AsRawFd};
283        let file = std::fs::OpenOptions::new()
284            .read(true)
285            .write(true)
286            .create(true)
287            .truncate(false)
288            .mode(0o600)
289            .custom_flags(libc::O_NOFOLLOW)
290            .open(socket.with_extension("lock"))?;
291        // SAFETY: flock operates on this owned, live file descriptor.
292        if unsafe { libc::flock(file.as_raw_fd(), libc::LOCK_EX | libc::LOCK_NB) } != 0 {
293            return Err(anyhow!("SCV daemon already owns this socket"));
294        }
295        Ok(Self(file))
296    }
297}
298impl Drop for SocketLock {
299    fn drop(&mut self) {
300        use std::os::unix::io::AsRawFd;
301        // SAFETY: the descriptor remains live until this drop returns.
302        unsafe {
303            libc::flock(self.0.as_raw_fd(), libc::LOCK_UN);
304        }
305    }
306}
307
308/// Aborts a task when dropped, so a task the daemon spawned never outlives it.
309pub(crate) struct AbortGuard(pub(crate) tokio::task::AbortHandle);
310impl Drop for AbortGuard {
311    fn drop(&mut self) {
312        self.0.abort();
313    }
314}
315
316#[cfg(test)]
317mod tests;
scv_server/daemon.rs

scv_server/
daemon.rs
