pitchfork_cli/ipc/
mod.rs

1use crate::Result;
2use crate::daemon::{Daemon, RunOptions};
3use crate::daemon_id::DaemonId;
4use crate::env;
5#[cfg(unix)]
6use crate::error::IpcError;
7use interprocess::local_socket::Name;
8#[cfg(unix)]
9use interprocess::local_socket::{GenericFilePath, ToFsName};
10#[cfg(windows)]
11use interprocess::local_socket::{GenericNamespaced, ToNsName};
12use miette::{Context, IntoDiagnostic};
13#[cfg(unix)]
14use std::path::Path;
15use std::path::PathBuf;
16
17pub(crate) mod batch;
18pub(crate) mod client;
19pub(crate) mod server;
20
21// #[derive(Debug, Clone, serde::Serialize, serde::Deserialize, strum::Display, strum::EnumIs)]
22// pub enum IpcMessage {
23//     Connect(String),
24//     ConnectOK,
25//     Run(String, Vec<String>),
26//     Stop(String),
27//     DaemonAlreadyRunning(String),
28//     DaemonAlreadyStopped(String),
29//     DaemonStart(Daemon),
30//     DaemonStop { name: String },
31//     DaemonFailed { name: String, error: String },
32//     Response(String),
33// }
34
35#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, strum::Display, strum::EnumIs)]
36#[allow(clippy::large_enum_variant)]
37pub enum IpcRequest {
38    Connect,
39    /// Versioned connect handshake (v2): client sends its version so the supervisor can
40    /// detect mismatches. Kept as a separate variant so the wire format of `Connect`
41    /// (unit variant) stays unchanged for backward compatibility with older supervisors.
42    ConnectV2 {
43        version: String,
44    },
45    Clean,
46    Stop {
47        id: DaemonId,
48    },
49    GetActiveDaemons,
50    GetDisabledDaemons,
51    /// Shut the supervisor down gracefully, as on SIGTERM, for
52    /// `supervisor stop`. Answered with `ShuttingDown` once new starts are
53    /// frozen; an older supervisor answers `Error`, and is sent the signal.
54    Shutdown,
55    Run(RunOptions),
56    Enable {
57        id: DaemonId,
58    },
59    Disable {
60        id: DaemonId,
61    },
62    UpdateShellDir {
63        shell_pid: u32,
64        dir: PathBuf,
65    },
66    GetNotifications,
67    /// Notify the supervisor that the slug registry has changed (e.g. `proxy add/remove`).
68    /// The supervisor should re-read slugs and update mDNS records accordingly.
69    SyncMdns,
70    /// Notify the supervisor that settings have changed.
71    /// The supervisor should reload settings from config files.
72    ReloadConfig,
73    /// Enter or replace a project session for a host PID in a directory.
74    ProjectEnter {
75        pid: u32,
76        dir: PathBuf,
77    },
78    /// Leave a project session for a host PID in a directory.
79    ProjectLeave {
80        pid: u32,
81        dir: PathBuf,
82    },
83    /// List all tracked project sessions with live liveness status filled in
84    /// by the supervisor.
85    GetProjectSessions,
86    /// A daemon's log sink reporting a line the supervisor needs to act on:
87    /// one matching the readiness pattern, one that should fire the
88    /// `on_output` hook, or both.
89    ///
90    /// The sink owns the daemon's output stream, so it does the matching and
91    /// tells the supervisor rather than the other way around. Sent by
92    /// `pitchfork log-sink`, not by any user-facing command.
93    SinkOutputLine {
94        id: DaemonId,
95        /// Identifies the start attempt this sink belongs to, so a report from
96        /// a sink still draining a previous attempt cannot satisfy a retry.
97        token: u64,
98        /// Whether this line passed the `on_output` hook's filter and debounce.
99        /// A line reported only because it matched the readiness pattern must
100        /// not fire a hook that filters for something else.
101        fires_hook: bool,
102        line: String,
103    },
104    /// Ask the supervisor for the URL of the web UI, if it is running.
105    /// Reflects the actual bound address, not static config.
106    GetWebUrl,
107    /// Remove stopped daemon registrations matching the supplied filters.
108    /// Appended to preserve the wire indexes of existing variants.
109    CleanFiltered {
110        namespaces: Vec<String>,
111        daemons: Vec<DaemonId>,
112        prune: bool,
113    },
114    /// These daemons are being started explicitly: none of them is to be
115    /// stopped for inactivity from now on, even if the proxy started it.
116    /// Sent before the start itself, and answered once any idle stop already
117    /// under way for one of them has finished. Appended to preserve the wire
118    /// indexes of existing variants.
119    ClaimDaemons {
120        ids: Vec<DaemonId>,
121    },
122    /// The supervisor's records of these daemons, whatever their status.
123    /// Missing ids are left out. Appended to preserve the wire indexes of
124    /// existing variants.
125    GetDaemons {
126        ids: Vec<DaemonId>,
127    },
128    /// Invalid request (failed to deserialize)
129    #[serde(skip)]
130    Invalid {
131        error: String,
132    },
133}
134
135/// A snapshot of a single project session, returned by `GetProjectSessions`.
136///
137/// `liveness_title` is the title recorded at enter time. `alive` and
138/// `current_title` are filled in by the supervisor from its `PROCS` singleton
139/// so the client does not need process introspection.
140#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
141pub struct ProjectSessionInfo {
142    pub pid: u32,
143    pub directory: PathBuf,
144    #[serde(skip_serializing_if = "Option::is_none", default)]
145    pub liveness_title: Option<String>,
146    pub alive: bool,
147    #[serde(skip_serializing_if = "Option::is_none", default)]
148    pub current_title: Option<String>,
149}
150
151#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, strum::Display, strum::EnumIs)]
152pub enum IpcResponse {
153    Ok,
154    /// Successful connect handshake, includes supervisor version for mismatch detection
155    ConnectOk {
156        version: String,
157    },
158    Yes,
159    No,
160    Error(String),
161    Notifications(Vec<(log::LevelFilter, String)>),
162    ActiveDaemons(Vec<Daemon>),
163    DisabledDaemons(Vec<DaemonId>),
164    /// The supervisor is stopping its daemons and will exit; `budget_ms` is
165    /// how long that may take, after which it may be killed.
166    ShuttingDown {
167        budget_ms: u64,
168    },
169    DaemonAlreadyRunning,
170    DaemonStart {
171        daemon: Daemon,
172    },
173    DaemonFailed {
174        error: String,
175    },
176    /// Port conflict detected with detailed process information
177    PortConflict {
178        port: u16,
179        process: String,
180        pid: u32,
181    },
182    /// No available ports found after exhausting auto-bump attempts
183    NoAvailablePort {
184        start_port: u16,
185        attempts: u32,
186    },
187    DaemonReady {
188        daemon: Daemon,
189    },
190    DaemonFailedWithCode {
191        exit_code: Option<i32>,
192        /// Ports resolved by the failed attempt, so in-process retry hooks
193        /// observe the attempt's actual (post-bump) ports.
194        #[serde(default)]
195        resolved_ports: Vec<u16>,
196    },
197    /// Process was not running but had a PID record (unexpected exit)
198    DaemonWasNotRunning,
199    /// mDNS sync completed (or was a no-op if LAN mode is disabled)
200    MdnsSynced,
201    /// Settings reloaded from config files
202    ConfigReloaded,
203    /// URL of the running web UI, or `None` if the web UI is not running
204    WebUrl {
205        url: Option<String>,
206    },
207    /// Failed to kill the process (still running)
208    DaemonStopFailed {
209        error: String,
210    },
211    /// Daemon exists but is not running (no PID)
212    DaemonNotRunning,
213    DaemonNotFound,
214    /// Snapshot of all project sessions (response to `GetProjectSessions`).
215    ProjectSessions(Vec<ProjectSessionInfo>),
216    /// Number of daemon registrations removed by `CleanFiltered`.
217    Cleaned {
218        count: u64,
219    },
220    /// Records answering `GetDaemons`.
221    Daemons(Vec<Daemon>),
222}
223
224/// Bytes a socket path may occupy in `sockaddr_un.sun_path` on this platform
225/// (108 on Linux, 104 on macOS and the BSDs). Matches the check `interprocess`
226/// makes when it builds the address, so an over-long path is caught here first.
227#[cfg(unix)]
228const SOCKET_PATH_CAPACITY: usize = {
229    // SAFETY: `sockaddr_un` is plain data, for which all-zero bytes are valid.
230    let sun = unsafe { std::mem::zeroed::<libc::sockaddr_un>() };
231    sun.sun_path.len()
232};
233
234/// Fail with an actionable error when a socket path cannot fit in `sun_path`.
235///
236/// Without this the connection attempt fails, five retries later, with
237/// `local socket name length exceeds capacity of sun_path of sockaddr_un`,
238/// which names neither the path nor what to change.
239#[cfg(unix)]
240fn check_socket_path(path: &Path, capacity: usize) -> Result<()> {
241    use std::os::unix::ffi::OsStrExt;
242    let len = path.as_os_str().as_bytes().len();
243    if len <= capacity {
244        return Ok(());
245    }
246    let over = len - capacity;
247    let help = format!(
248        "the socket is {path} and is {over} byte(s) over the limit.\n\
249         Set PITCHFORK_STATE_DIR (or XDG_STATE_HOME) to a shorter directory, for example \
250         PITCHFORK_STATE_DIR=/tmp/pitchfork, so that <state dir>/sock/main.sock is at most \
251         {capacity} bytes.",
252        path = path.display()
253    );
254    Err(IpcError::SocketPathTooLong {
255        path: path.to_path_buf(),
256        len,
257        limit: capacity,
258        help,
259    }
260    .into())
261}
262
263fn fs_name(name: &str) -> Result<Name<'_>> {
264    // Unix: use a filesystem path for the AF_UNIX socket.
265    #[cfg(unix)]
266    {
267        let path = env::IPC_SOCK_DIR.join(name).with_extension("sock");
268        check_socket_path(&path, SOCKET_PATH_CAPACITY)?;
269        let fs_name = path.to_fs_name::<GenericFilePath>().into_diagnostic()?;
270        Ok(fs_name)
271    }
272    // Windows: named pipes use a flat namespace (\\.\pipe\<name>) that
273    // cannot contain path separators. Derive a unique pipe name from the
274    // state directory to preserve test isolation when multiple supervisors
275    // run concurrently with different PITCHFORK_STATE_DIR values.
276    //
277    // Use a hash of the state directory path rather than character replacement
278    // to guarantee injectivity: `C:\a.b` and `C:\a\b` would both flatten to
279    // `C--a-b` with the old approach, causing pipe name collisions.
280    #[cfg(windows)]
281    {
282        let state_dir = env::PITCHFORK_STATE_DIR.to_string_lossy();
283        // FNV-1a hash: deterministic, stable across Rust versions.
284        // DefaultHasher's algorithm is not guaranteed stable, which would
285        // break IPC if the CLI and supervisor were ever compiled with
286        // different toolchains.
287        let mut hash: u64 = 0xcbf29ce484222325;
288        for byte in state_dir.bytes() {
289            hash ^= byte as u64;
290            hash = hash.wrapping_mul(0x100000001b3);
291        }
292        let pipe_name = format!("pitchfork-{hash:016x}-{name}");
293        pipe_name
294            .to_ns_name::<GenericNamespaced>()
295            .into_diagnostic()
296    }
297}
298
299/// Human-readable location of the supervisor's IPC endpoint, for messages.
300pub(crate) fn socket_display() -> String {
301    #[cfg(unix)]
302    {
303        env::IPC_SOCK_MAIN.display().to_string()
304    }
305    #[cfg(windows)]
306    {
307        "the supervisor named pipe".to_string()
308    }
309}
310
311/// Whether a supervisor is accepting connections on the IPC socket right now.
312///
313/// This is the ground truth for "is a supervisor running": the state-file
314/// record can be lost or rewritten while the supervisor keeps serving, and a
315/// supervisor started on that basis would take the socket over and leave the
316/// original running but unreachable. A stale socket file left by a crashed
317/// supervisor refuses connections, so it does not count.
318pub(crate) async fn supervisor_listening() -> bool {
319    use interprocess::local_socket::traits::tokio::Stream as _;
320    let Ok(name) = fs_name("main") else {
321        return false;
322    };
323    let connect = interprocess::local_socket::tokio::Stream::connect(name);
324    // A live supervisor accepts at once (the kernel queues the connection);
325    // the timeout only guards against a platform where connecting blocks.
326    match tokio::time::timeout(std::time::Duration::from_secs(1), connect).await {
327        Ok(Ok(_)) => true,
328        Ok(Err(err)) => {
329            trace!("no supervisor listening on the IPC socket: {err}");
330            false
331        }
332        Err(_) => {
333            debug!("timed out probing the IPC socket; treating it as not listening");
334            false
335        }
336    }
337}
338
339/// Encode an IPC message as JSON.
340///
341/// Messages are framed by a trailing NUL byte, which JSON text never contains.
342/// A binary encoding would, so it cannot use this framing.
343fn serialize<T: serde::Serialize>(msg: &T) -> Result<Vec<u8>> {
344    serde_json::to_vec(msg)
345        .into_diagnostic()
346        .wrap_err("failed to serialize IPC message as JSON")
347}
348
349fn deserialize<T: serde::de::DeserializeOwned>(bytes: &[u8]) -> Result<T> {
350    let mut bytes = bytes.to_vec();
351    bytes.pop();
352    let preview = std::str::from_utf8(&bytes).unwrap_or("<binary>");
353    trace!("msg: {preview:?}");
354    serde_json::from_slice(&bytes)
355        .into_diagnostic()
356        .wrap_err("failed to deserialize IPC JSON response")
357}
358
359#[cfg(test)]
360mod tests {
361    use super::*;
362
363    #[cfg(unix)]
364    #[test]
365    fn socket_path_over_sun_path_capacity_names_path_length_and_fix() {
366        let fits = PathBuf::from(format!("/{}", "a".repeat(SOCKET_PATH_CAPACITY - 1)));
367        assert_eq!(fits.as_os_str().len(), SOCKET_PATH_CAPACITY);
368        assert!(check_socket_path(&fits, SOCKET_PATH_CAPACITY).is_ok());
369
370        let long = PathBuf::from(format!(
371            "/{}/sock/main.sock",
372            "a".repeat(SOCKET_PATH_CAPACITY)
373        ));
374        let err = check_socket_path(&long, SOCKET_PATH_CAPACITY).unwrap_err();
375        let len = long.as_os_str().len();
376        let message = err.to_string();
377        assert!(message.contains(&format!("{len} bytes")), "{message}");
378        assert!(
379            message.contains(&format!("allows {SOCKET_PATH_CAPACITY}")),
380            "{message}"
381        );
382        let help = err.help().expect("help text").to_string();
383        assert!(help.contains(&long.display().to_string()), "{help}");
384        assert!(help.contains("PITCHFORK_STATE_DIR"), "{help}");
385        assert!(help.contains("XDG_STATE_HOME"), "{help}");
386        assert!(
387            matches!(
388                err.downcast_ref::<IpcError>(),
389                Some(IpcError::SocketPathTooLong { len: l, limit, .. })
390                    if *l == len && *limit == SOCKET_PATH_CAPACITY
391            ),
392            "{err:?}"
393        );
394    }
395
396    #[cfg(unix)]
397    #[test]
398    fn socket_path_capacity_matches_the_platform() {
399        // 108 on Linux, 104 on macOS and the BSDs.
400        assert!((100..=108).contains(&SOCKET_PATH_CAPACITY));
401    }
402
403    #[test]
404    fn filtered_clean_ipc_round_trips() {
405        let request = IpcRequest::CleanFiltered {
406            namespaces: vec!["worktree".to_string()],
407            daemons: vec![DaemonId::new("worktree", "api")],
408            prune: true,
409        };
410        let mut bytes = serialize(&request).unwrap();
411        bytes.push(b'\n');
412        let decoded: IpcRequest = deserialize(&bytes).unwrap();
413        match decoded {
414            IpcRequest::CleanFiltered {
415                namespaces,
416                daemons,
417                prune,
418            } => {
419                assert_eq!(namespaces, ["worktree"]);
420                assert_eq!(daemons, [DaemonId::new("worktree", "api")]);
421                assert!(prune);
422            }
423            other => panic!("unexpected request: {other:?}"),
424        }
425
426        let mut bytes = serialize(&IpcResponse::Cleaned { count: 3 }).unwrap();
427        bytes.push(b'\n');
428        let decoded: IpcResponse = deserialize(&bytes).unwrap();
429        assert!(matches!(decoded, IpcResponse::Cleaned { count: 3 }));
430    }
431
432    fn round_trip<T: serde::Serialize + serde::de::DeserializeOwned>(value: &T) -> T {
433        let mut bytes = serialize(value).unwrap();
434        bytes.push(b'\n');
435        deserialize(&bytes).unwrap()
436    }
437
438    #[test]
439    fn claim_daemons_ipc_round_trips() {
440        let ids = vec![DaemonId::new("proj", "api"), DaemonId::new("proj", "db")];
441        match round_trip(&IpcRequest::ClaimDaemons { ids: ids.clone() }) {
442            IpcRequest::ClaimDaemons { ids: decoded } => assert_eq!(decoded, ids),
443            other => panic!("unexpected request: {other:?}"),
444        }
445    }
446
447    /// The idle-shutdown ownership crosses IPC in both structs, alongside
448    /// fields that are skipped when empty.
449    #[test]
450    fn proxy_idle_timeout_survives_the_ipc_encoding() {
451        let daemon = Daemon {
452            id: DaemonId::new("proj", "api"),
453            proxy_idle_timeout_ms: Some(900_000),
454            ..Default::default()
455        };
456        match round_trip(&IpcResponse::ActiveDaemons(vec![daemon])) {
457            IpcResponse::ActiveDaemons(daemons) => {
458                assert_eq!(daemons[0].proxy_idle_timeout_ms, Some(900_000));
459                assert!(!daemons[0].oneshot);
460            }
461            other => panic!("unexpected response: {other:?}"),
462        }
463
464        let opts = RunOptions {
465            id: DaemonId::new("proj", "api"),
466            proxy_idle_timeout_ms: Some(900_000),
467            ..Default::default()
468        };
469        match round_trip(&IpcRequest::Run(opts)) {
470            IpcRequest::Run(opts) => assert_eq!(opts.proxy_idle_timeout_ms, Some(900_000)),
471            other => panic!("unexpected request: {other:?}"),
472        }
473    }
474
475    /// A start replaces the saved record only when it says so: a request from
476    /// an older client, which does not send the flag, keeps merging.
477    #[test]
478    fn replaces_saved_record_is_opt_in_across_ipc() {
479        let full = RunOptions {
480            id: DaemonId::new("proj", "api"),
481            replaces_saved_record: true,
482            ..Default::default()
483        };
484        match round_trip(&IpcRequest::Run(full)) {
485            IpcRequest::Run(opts) => assert!(opts.replaces_saved_record),
486            other => panic!("unexpected request: {other:?}"),
487        }
488
489        let older_client = RunOptions {
490            id: DaemonId::new("proj", "api"),
491            ..Default::default()
492        };
493        let bytes = serialize(&IpcRequest::Run(older_client)).unwrap();
494        assert!(!String::from_utf8_lossy(&bytes).contains("replaces_saved_record"));
495        match round_trip(&IpcRequest::Run(RunOptions {
496            id: DaemonId::new("proj", "api"),
497            ..Default::default()
498        })) {
499            IpcRequest::Run(opts) => assert!(!opts.replaces_saved_record),
500            other => panic!("unexpected request: {other:?}"),
501        }
502    }
503}
pitchfork_cli/ipc/mod.rs

pitchfork_cli/ipc/
mod.rs
