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

scv_server/
daemon.rs
