kunobi_daemon/local/
mod.rs

1//! Optional OS adapters. Unsafe calls are confined to these platform boundaries.
2//!
3//! [`ProcessHandle`] waits for a process to exit without polling its PID.
4use std::io;
5use std::time::{Duration, Instant};
6
7/// What names a local endpoint on this platform.
8///
9/// A Unix socket is a filesystem path; a named pipe is a name in its own
10/// namespace, with its own length limit and no directory. The crate does not
11/// derive one from the other: the endpoint name is where processes agree to
12/// meet, so a scheme imposed here would move the rendezvous of every consumer
13/// that already has one. A new Windows service can take its pipe name from
14/// `local::pipe_name::for_current_user`, which scopes it by user and service.
15#[cfg(unix)]
16pub type Endpoint = std::path::Path;
17/// What names a local endpoint on this platform.
18#[cfg(windows)]
19pub type Endpoint = str;
20
21/// How long to keep retrying a refused connection before giving up.
22///
23/// The peer may be up while it is still binding, so a single attempt is too
24/// strict. "Not running at all" is detected earlier and separately, so this
25/// budget is only ever spent on the genuinely transient case.
26pub const CONNECT_BUDGET: Duration = Duration::from_secs(5);
27
28/// Gap between connection attempts.
29pub const RETRY_INTERVAL: Duration = Duration::from_millis(50);
30
31/// A local connection: how to open one, who is on the far end, and how to take
32/// its halves apart.
33///
34/// Every method here has to exist on both platforms, which is the point. While
35/// connection setup lived in inherent methods the two adapters drifted: Windows
36/// never grew `connect_once_until`, and it carried its own copies of the budget
37/// and retry interval above. The compiler now compares them.
38pub trait Duplex: Sized + peer::PeerCredentials {
39    /// Receive half owned by the downstream pump.
40    type Reader: io::Read + Send + 'static;
41    /// Send half with an explicit half-close operation.
42    type Writer: crate::transport::WriteHalf;
43
44    /// Make one connection attempt bounded by an absolute deadline.
45    ///
46    /// Planned handoff wants exactly this: one try at one endpoint, without
47    /// spending the recovery retry budget on an endpoint that has moved.
48    fn connect_once_until(endpoint: &Endpoint, deadline: Instant) -> Result<Self, ConnectError>;
49
50    /// Retry availability until the deadline passes.
51    ///
52    /// Only [`ConnectError::is_transient`] failures are retried: a denial, an
53    /// endpoint too long to address or any other OS failure does not resolve by
54    /// waiting, and spending the budget first only makes the diagnosis slower.
55    fn connect_until(endpoint: &Endpoint, deadline: Instant) -> Result<Self, ConnectError> {
56        loop {
57            match Self::connect_once_until(endpoint, deadline) {
58                Ok(connected) => return Ok(connected),
59                Err(error) if !error.is_transient() => return Err(error),
60                Err(error) => {
61                    let remaining = deadline.saturating_duration_since(Instant::now());
62                    if remaining.is_zero() {
63                        return Err(error);
64                    }
65                    #[expect(
66                        clippy::disallowed_methods,
67                        reason = "A missing endpoint has no event to await; retry connects within the caller deadline."
68                    )]
69                    std::thread::sleep(RETRY_INTERVAL.min(remaining));
70                }
71            }
72        }
73    }
74
75    /// One attempt under [`CONNECT_BUDGET`].
76    fn connect_once(endpoint: &Endpoint) -> Result<Self, ConnectError> {
77        Self::connect_once_until(endpoint, Instant::now() + CONNECT_BUDGET)
78    }
79
80    /// Retry under [`CONNECT_BUDGET`].
81    fn connect(endpoint: &Endpoint) -> Result<Self, ConnectError> {
82        Self::connect_until(endpoint, Instant::now() + CONNECT_BUDGET)
83    }
84
85    /// Arm or clear one absolute deadline for session establishment.
86    fn set_read_deadline(&self, timeout: Option<Duration>) -> io::Result<()>;
87
88    /// Split ownership without serializing reads and writes.
89    fn split(self) -> io::Result<(Self::Reader, Self::Writer)>;
90}
91
92/// The outcome of trying to own an endpoint. Busy is ownership evidence, never
93/// application readiness proof.
94///
95/// Shared across platforms because the outcome has the same shape everywhere.
96/// `BindError` deliberately is not: a Unix socket can fail to prepare its parent
97/// directory or to unlink a proven stale inode, and a named pipe has neither.
98#[derive(Debug)]
99pub enum Bound<L> {
100    /// The caller owns this listener.
101    Won(L),
102    /// A listener or another serialized binder already owns the endpoint.
103    AlreadyRunning,
104}
105
106/// This platform's [`Duplex`].
107///
108/// Safe to name without a `cfg` because the trait fixes what both adapters
109/// offer; they cannot drift apart behind this alias.
110#[cfg(unix)]
111pub use unix::UnixDuplex as PlatformDuplex;
112/// This platform's [`Duplex`].
113#[cfg(windows)]
114pub use windows::WindowsDuplex as PlatformDuplex;
115
116pub(crate) use process::wait_child;
117pub use process::{ProcessHandle, process_has_exited, process_state};
118// Same name, same signature on both platforms. Everything else in `unix` and
119// `windows` differs (uid and child-reaping helpers on one side, a session-end
120// handler and an inherit guard on the other) and stays behind its module.
121#[cfg(all(unix, feature = "async"))]
122pub use unix::shutdown_signal;
123#[cfg(unix)]
124pub use unix::terminate_legacy_peer;
125#[cfg(all(windows, feature = "async"))]
126pub use windows::shutdown_signal;
127#[cfg(windows)]
128pub use windows::terminate_legacy_peer;
129
130/// What the OS can establish about a PID.
131///
132/// `Unknown` is its own state because the two useful answers need proof in
133/// opposite directions. Retiring a process's work needs proof it exited;
134/// treating it as still serving needs proof it runs. A denied or failed query
135/// proves neither, and a caller that folds it into either side eventually acts
136/// on a guess: waiting forever for a process that is gone, or starting a
137/// second owner next to one that is not.
138///
139/// `Alive` is about a PID, not a program: a PID is reused once its process is
140/// gone, on Unix and on Windows alike, so a PID that reads as alive may belong
141/// to an unrelated process by now. Before treating it as the daemon you
142/// started, confirm through its endpoint (connect and check the peer). To
143/// follow one process rather than a PID, keep a [`ProcessHandle`].
144#[derive(Clone, Copy, Debug, PartialEq, Eq)]
145#[non_exhaustive]
146pub enum ProcessState {
147    /// A process with this PID exists and has not exited. It may be a
148    /// different process from the one that first had the PID.
149    Alive,
150    /// The process exited, or no process has this PID any more.
151    Exited,
152    /// The OS did not say, for example because access to the process was
153    /// denied. Neither alive nor exited can be assumed.
154    Unknown,
155}
156
157/// Listener ownership for this platform.
158///
159/// Both sides now offer `acquire` under `local`, returning a blocking listener.
160/// Only Windows also offers `acquire_tokio`: a named pipe needs the Tokio
161/// wrapper chosen at creation, while a Unix descriptor converts to one
162/// afterwards, so asking this crate to do it would pull `interprocess` into
163/// every Unix build to save the caller two lines.
164#[cfg(unix)]
165pub use unix_socket as socket;
166/// Listener ownership for this platform.
167#[cfg(windows)]
168pub use windows_socket as socket;
169
170/// Connection establishment failure, before application traffic is sent.
171#[derive(Clone, Copy, Debug, PartialEq, Eq)]
172#[non_exhaustive]
173pub enum ConnectError {
174    /// Nothing accepted the connection in time: the endpoint does not exist
175    /// yet, refused the connection, is busy, or the deadline passed.
176    ConnectTimeout,
177    /// The OS denied access to the endpoint.
178    PermissionDenied,
179    /// The endpoint path does not fit in a Unix socket address, so no attempt
180    /// can succeed. See [`crate::socket_path`].
181    EndpointTooLong,
182    /// The OS failed the attempt for another reason, which waiting does not fix.
183    Failed(io::ErrorKind),
184}
185impl std::fmt::Display for ConnectError {
186    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
187        match self {
188            Self::ConnectTimeout => f.write_str("daemon connection timed out"),
189            Self::PermissionDenied => f.write_str("daemon endpoint permission denied"),
190            Self::EndpointTooLong => {
191                f.write_str("daemon endpoint path is too long for a Unix socket")
192            }
193            Self::Failed(kind) => write!(f, "daemon connection failed: {kind}"),
194        }
195    }
196}
197impl std::error::Error for ConnectError {}
198impl ConnectError {
199    /// Whether a later attempt can succeed: nothing accepted the connection in
200    /// time, which includes an endpoint that does not exist yet. Every failure
201    /// [`is_transient`] lists becomes [`Self::ConnectTimeout`].
202    pub fn is_transient(&self) -> bool {
203        matches!(self, Self::ConnectTimeout)
204    }
205}
206
207/// Whether a local connection or exchange that failed with `error` can
208/// succeed if repeated.
209///
210/// True when the endpoint does not exist yet or refused the connection, when
211/// the peer left mid-exchange, or when a deadline passed. A departing peer
212/// shows up as a reset, an end of stream or a broken pipe depending on timing
213/// the caller does not control, so all of them count: retrying only one makes
214/// surviving the same event a matter of chance. A denial and every other
215/// failure are final. [`ConnectError`] classifies connection setup through
216/// this list.
217pub fn is_transient(error: &io::Error) -> bool {
218    use io::ErrorKind::{
219        BrokenPipe, ConnectionAborted, ConnectionRefused, ConnectionReset, Interrupted,
220        NotConnected, NotFound, TimedOut, UnexpectedEof, WouldBlock,
221    };
222    matches!(
223        error.kind(),
224        NotFound
225            | ConnectionRefused
226            | ConnectionReset
227            | ConnectionAborted
228            | NotConnected
229            | UnexpectedEof
230            | BrokenPipe
231            | TimedOut
232            | WouldBlock
233            | Interrupted
234    ) || busy_pipe(error)
235}
236
237#[cfg(windows)]
238use windows::busy_pipe;
239/// Only a named pipe can be busy; Unix reports nothing outside the portable
240/// kinds above.
241#[cfg(unix)]
242fn busy_pipe(_: &io::Error) -> bool {
243    false
244}
245
246/// Classify a failed connection attempt, the same way on every platform.
247pub(crate) fn connect_error(error: &io::Error) -> ConnectError {
248    if error.kind() == io::ErrorKind::PermissionDenied {
249        ConnectError::PermissionDenied
250    } else if is_transient(error) {
251        ConnectError::ConnectTimeout
252    } else {
253        ConnectError::Failed(error.kind())
254    }
255}
256
257pub mod peer;
258mod process;
259#[cfg(unix)]
260pub mod unix;
261#[cfg(unix)]
262pub mod unix_socket;
263#[cfg(windows)]
264pub mod windows;
265
266#[cfg(test)]
267fn test_handshake(write: &mut impl io::Write, read: &mut impl io::Read, _: &str) -> io::Result<()> {
268    write.write_all(b"hello\n")?;
269    write.flush()?;
270    let mut reply = [0; 6];
271    read.read_exact(&mut reply)?;
272    if reply != *b"ready\n" {
273        return Err(io::Error::other("invalid test reply"));
274    }
275    Ok(())
276}
277
278#[cfg(windows)]
279pub mod windows_socket;
280
281#[cfg(windows)]
282pub mod pipe_name;
283// Composing a name needs no Windows call, so the other platforms test it too,
284// including the Linux mutation runs.
285#[cfg(all(test, not(windows)))]
286mod pipe_name;
287
288#[cfg(any(all(windows, feature = "launch"), test))]
289pub(crate) mod command_line;
290#[cfg(all(windows, feature = "launch"))]
291pub(crate) mod windows_spawn;
292
293#[cfg(test)]
294mod connect_error_tests {
295    use super::{ConnectError, connect_error};
296    use std::io::{self, ErrorKind};
297
298    #[test]
299    fn an_absent_or_departing_peer_is_worth_retrying() {
300        // A departing peer surfaces as a reset, an end of stream or a broken
301        // pipe depending on timing the caller does not control, so all three
302        // must be retried for a retry to be deterministic.
303        for kind in [
304            ErrorKind::NotFound,
305            ErrorKind::ConnectionRefused,
306            ErrorKind::TimedOut,
307            ErrorKind::WouldBlock,
308            ErrorKind::Interrupted,
309            ErrorKind::ConnectionReset,
310            ErrorKind::ConnectionAborted,
311            ErrorKind::NotConnected,
312            ErrorKind::UnexpectedEof,
313            ErrorKind::BrokenPipe,
314        ] {
315            assert!(super::is_transient(&kind.into()), "{kind:?}");
316            let error = connect_error(&kind.into());
317            assert_eq!(error, ConnectError::ConnectTimeout, "{kind:?}");
318            assert!(error.is_transient());
319        }
320        for kind in [
321            ErrorKind::PermissionDenied,
322            ErrorKind::InvalidInput,
323            ErrorKind::InvalidData,
324            ErrorKind::AddrInUse,
325            ErrorKind::Other,
326        ] {
327            assert!(!super::is_transient(&kind.into()), "{kind:?}");
328        }
329        assert_eq!(
330            connect_error(&ErrorKind::PermissionDenied.into()),
331            ConnectError::PermissionDenied
332        );
333        let other = connect_error(&ErrorKind::InvalidInput.into());
334        assert_eq!(other, ConnectError::Failed(ErrorKind::InvalidInput));
335        assert!(!other.is_transient());
336        assert!(!ConnectError::PermissionDenied.is_transient());
337        assert!(!ConnectError::EndpointTooLong.is_transient());
338    }
339
340    #[test]
341    fn a_busy_pipe_is_retried_only_on_windows() {
342        let busy = connect_error(&io::Error::from_raw_os_error(231));
343        if cfg!(windows) {
344            assert_eq!(busy, ConnectError::ConnectTimeout);
345        } else {
346            assert_ne!(busy, ConnectError::ConnectTimeout);
347        }
348        assert_eq!(
349            super::is_transient(&io::Error::from_raw_os_error(231)),
350            cfg!(windows)
351        );
352    }
353
354    #[test]
355    fn a_temporary_connection_failure_is_retried_until_the_endpoint_accepts() {
356        use super::{Duplex, Endpoint, PlatformDuplex, peer::PeerCredentials};
357        use std::sync::atomic::{AtomicUsize, Ordering};
358        static ATTEMPTS: AtomicUsize = AtomicUsize::new(0);
359        struct StartsAfterOneAttempt;
360        impl PeerCredentials for StartsAfterOneAttempt {
361            fn credentials(&self) -> io::Result<crate::peer::Credentials> {
362                unreachable!()
363            }
364        }
365        impl Duplex for StartsAfterOneAttempt {
366            type Reader = <PlatformDuplex as Duplex>::Reader;
367            type Writer = <PlatformDuplex as Duplex>::Writer;
368            fn connect_once_until(
369                _: &Endpoint,
370                _: std::time::Instant,
371            ) -> Result<Self, ConnectError> {
372                if ATTEMPTS.fetch_add(1, Ordering::Relaxed) == 0 {
373                    Err(ConnectError::ConnectTimeout)
374                } else {
375                    Ok(Self)
376                }
377            }
378            fn set_read_deadline(&self, _: Option<std::time::Duration>) -> io::Result<()> {
379                unreachable!()
380            }
381            fn split(self) -> io::Result<(Self::Reader, Self::Writer)> {
382                unreachable!()
383            }
384        }
385        #[cfg(unix)]
386        let endpoint = std::path::Path::new("unused");
387        #[cfg(windows)]
388        let endpoint = "unused";
389        assert!(
390            StartsAfterOneAttempt::connect_until(
391                endpoint,
392                std::time::Instant::now() + std::time::Duration::from_secs(1)
393            )
394            .is_ok()
395        );
396        assert_eq!(ATTEMPTS.load(Ordering::Relaxed), 2);
397    }
398
399    #[test]
400    fn every_failure_says_what_happened() {
401        let cases = [
402            (ConnectError::ConnectTimeout, "daemon connection timed out"),
403            (
404                ConnectError::PermissionDenied,
405                "daemon endpoint permission denied",
406            ),
407            (
408                ConnectError::EndpointTooLong,
409                "daemon endpoint path is too long for a Unix socket",
410            ),
411            (
412                ConnectError::Failed(ErrorKind::InvalidInput),
413                "daemon connection failed: invalid input parameter",
414            ),
415        ];
416        for (error, text) in cases {
417            assert_eq!(error.to_string(), text);
418        }
419    }
420}
421
422mod readiness;
kunobi_daemon/local/mod.rs

kunobi_daemon/local/
mod.rs
