pitchfork-cli 2.13.1

Daemons with DX
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

//! IPC request handling and dispatch
//!
//! Handles incoming IPC requests from CLI clients and routes them to the appropriate handlers.

use super::{SUPERVISOR, Supervisor};
use crate::Result;
use crate::ipc::server::IpcServer;
use crate::ipc::{IpcRequest, IpcResponse};
use miette::IntoDiagnostic;

const VERSION: &str = env!("CARGO_PKG_VERSION");

impl Supervisor {
    /// Main IPC connection watch loop - reads and dispatches requests
    pub(crate) async fn conn_watch(&self, mut ipc: IpcServer) -> ! {
        loop {
            let (msg, send) = match ipc.read().await {
                Ok(msg) => msg,
                Err(e) => {
                    error!("failed to accept connection: {e:?}");
                    continue;
                }
            };
            debug!("received message: {msg:?}");
            tokio::spawn(async move {
                let rsp = SUPERVISOR
                    .handle_ipc(msg)
                    .await
                    .unwrap_or_else(|err| IpcResponse::Error(err.to_string()));
                if let Err(err) = send.send(rsp).await {
                    debug!("failed to send message: {err:?}");
                }
            });
        }
    }

    /// Handle a single IPC request and return the appropriate response
    pub(crate) async fn handle_ipc(&self, req: IpcRequest) -> Result<IpcResponse> {
        let rsp = match req {
            IpcRequest::Invalid { error } => {
                warn!("Invalid IPC request: {error}");
                return Ok(IpcResponse::Error(format!("Invalid request: {error}")));
            }
            IpcRequest::Connect => {
                debug!("received connect message (legacy, no version info)");
                IpcResponse::Ok
            }
            IpcRequest::ConnectV2 {
                version: client_version,
            } => {
                debug!("received connect message (client version: {client_version})");
                if client_version != VERSION {
                    warn!(
                        "Client version {client_version} differs from supervisor version {VERSION}. \
                            Restart the supervisor with: pitchfork supervisor start --force"
                    );
                }
                IpcResponse::ConnectOk {
                    version: VERSION.to_string(),
                }
            }
            IpcRequest::Stop { id } => {
                // id is already DaemonId, no validation needed
                self.stop(&id).await?
            }
            IpcRequest::Run(opts) => {
                // opts.id is already DaemonId, no validation needed
                self.run(opts).await?
            }
            IpcRequest::Enable { id } => {
                // id is already DaemonId, no validation needed
                if self.enable(&id).await? {
                    IpcResponse::Yes
                } else {
                    IpcResponse::No
                }
            }
            IpcRequest::Disable { id } => {
                // id is already DaemonId, no validation needed
                if self.disable(&id).await? {
                    IpcResponse::Yes
                } else {
                    IpcResponse::No
                }
            }
            IpcRequest::GetActiveDaemons => {
                let daemons = self.active_daemons().await;
                IpcResponse::ActiveDaemons(daemons)
            }
            IpcRequest::GetNotifications => {
                let notifications = self.get_notifications().await;
                IpcResponse::Notifications(notifications)
            }
            IpcRequest::UpdateShellDir { shell_pid, dir } => {
                let prev = self.get_shell_dir(shell_pid).await;
                self.set_shell_dir(shell_pid, dir.clone()).await?;
                // Cancel any pending autostops for daemons in the new directory
                self.cancel_pending_autostops_for_dir(&dir).await;
                if let Some(prev) = prev {
                    self.leave_dir(&prev).await?;
                }
                self.refresh().await?;
                IpcResponse::Ok
            }
            IpcRequest::Clean => {
                self.clean().await?;
                IpcResponse::Ok
            }
            IpcRequest::GetDisabledDaemons => {
                let disabled = self.state_file.lock().await.disabled.clone();
                IpcResponse::DisabledDaemons(disabled.into_iter().collect())
            }
            IpcRequest::SyncMdns => {
                self.sync_mdns().await;
                IpcResponse::MdnsSynced
            }
            IpcRequest::ReloadConfig => {
                tokio::task::spawn_blocking(|| {
                    crate::settings::reload_settings();
                    crate::logger::apply_settings();
                })
                .await
                .into_diagnostic()?;
                IpcResponse::ConfigReloaded
            }
        };
        // Ensure state is flushed to disk before returning the response
        // so that CLI commands reading StateFile::get() see fresh data.
        self.flush_state().await;
        Ok(rsp)
    }
}