rmux_server/
daemon.rs

1#[cfg(all(test, unix))]
2use std::fs;
3use std::io;
4#[cfg(windows)]
5use std::io::{Read, Write};
6use std::path::{Path, PathBuf};
7use std::sync::{Arc, Mutex as StdMutex};
8#[cfg(windows)]
9use std::time::Duration;
10
11use tokio::sync::oneshot;
12use tokio::task::JoinHandle;
13
14use rmux_core::events::SubscriptionLimits;
15#[cfg(windows)]
16use rmux_ipc::connect_blocking;
17use rmux_ipc::LocalEndpoint;
18#[cfg(windows)]
19use rmux_ipc::LocalListener;
20#[cfg(windows)]
21use rmux_proto::{
22    encode_frame, FrameDecoder, HasSessionRequest, Request, Response, RmuxError, SessionName,
23};
24
25use crate::listener;
26use crate::listener_options::ServeOptions;
27#[cfg(windows)]
28use crate::server_access::current_owner_uid;
29#[cfg(unix)]
30use crate::unix_socket::bind_unix_listener_at;
31#[cfg(unix)]
32use crate::unix_socket::real_user_id;
33#[cfg(all(test, unix))]
34use crate::unix_socket::{
35    ensure_parent_directory, indicates_stale_socket, remove_stale_socket_if_needed,
36};
37
38#[cfg(all(test, unix))]
39const FALLBACK_SOCKET_ROOT: &str = "/tmp";
40const DEFAULT_WEB_PORT: u16 = 9777;
41
42/// Computes the default RMUX daemon socket path.
43///
44/// The path uses an rmux-specific per-user directory so it cannot collide with
45/// a real tmux server socket.
46pub fn default_socket_path() -> io::Result<PathBuf> {
47    rmux_ipc::default_endpoint().map(LocalEndpoint::into_path)
48}
49
50#[cfg(all(test, unix))]
51fn socket_root_from_env(tmpdir: Option<&std::ffi::OsStr>) -> io::Result<PathBuf> {
52    let tmpdir = tmpdir
53        .filter(|value| !value.is_empty())
54        .map(PathBuf::from)
55        .into_iter();
56    let candidates = tmpdir.chain(std::iter::once(PathBuf::from(FALLBACK_SOCKET_ROOT)));
57
58    for candidate in candidates {
59        if let Ok(resolved) = fs::canonicalize(&candidate) {
60            return Ok(resolved);
61        }
62    }
63
64    Err(io::Error::new(
65        io::ErrorKind::NotFound,
66        "no suitable rmux socket directory",
67    ))
68}
69
70/// Daemon configuration for a single RMUX server instance.
71#[derive(Debug, Clone, PartialEq, Eq)]
72pub struct DaemonConfig {
73    socket_path: PathBuf,
74    config_load: ConfigLoadOptions,
75    subscription_limits: SubscriptionLimits,
76    web_frontend: Option<String>,
77    web_port: u16,
78    web_required: bool,
79}
80
81impl DaemonConfig {
82    /// Builds a daemon configuration for the given socket path.
83    #[must_use]
84    pub fn new(socket_path: PathBuf) -> Self {
85        Self {
86            socket_path,
87            config_load: ConfigLoadOptions::disabled(),
88            subscription_limits: SubscriptionLimits::default(),
89            web_frontend: None,
90            web_port: DEFAULT_WEB_PORT,
91            web_required: false,
92        }
93    }
94
95    /// Builds a daemon configuration using the default spec socket path.
96    pub fn with_default_socket_path() -> io::Result<Self> {
97        Ok(Self::new(default_socket_path()?))
98    }
99
100    /// Returns the configured local IPC endpoint path.
101    #[must_use]
102    pub fn socket_path(&self) -> &Path {
103        &self.socket_path
104    }
105
106    /// Returns the startup config loading policy.
107    #[must_use]
108    pub const fn config_load(&self) -> &ConfigLoadOptions {
109        &self.config_load
110    }
111
112    /// Returns the pane-output subscription limits.
113    #[must_use]
114    pub fn subscription_limits(&self) -> SubscriptionLimits {
115        self.subscription_limits
116    }
117
118    /// Returns the configured web-share listener port.
119    #[must_use]
120    pub const fn web_port(&self) -> u16 {
121        self.web_port
122    }
123
124    /// Returns whether this daemon startup requires the web listener to bind.
125    #[must_use]
126    pub const fn web_required(&self) -> bool {
127        self.web_required
128    }
129
130    /// Returns the optional external web-share frontend origin.
131    #[must_use]
132    pub fn web_frontend(&self) -> Option<&str> {
133        self.web_frontend.as_deref()
134    }
135
136    /// Enables RMUX default startup config loading.
137    #[must_use]
138    pub fn with_default_config_load(mut self, quiet: bool, cwd: Option<PathBuf>) -> Self {
139        self.config_load = ConfigLoadOptions {
140            selection: ConfigFileSelection::Default,
141            quiet,
142            cwd,
143        };
144        self
145    }
146
147    /// Overrides pane-output subscription limits for this daemon.
148    #[must_use]
149    pub fn with_subscription_limits(mut self, subscription_limits: SubscriptionLimits) -> Self {
150        self.subscription_limits = subscription_limits;
151        self
152    }
153
154    /// Overrides the web-share listener port.
155    #[must_use]
156    pub const fn with_web_port(mut self, port: u16) -> Self {
157        self.web_port = port;
158        self.web_required = true;
159        self
160    }
161
162    /// Overrides the frontend origin used in generated web-share URLs.
163    #[must_use]
164    pub fn with_web_frontend(mut self, frontend: String) -> Self {
165        self.web_frontend = Some(frontend);
166        self.web_required = true;
167        self
168    }
169
170    /// Enables explicit `-f` startup config loading.
171    #[must_use]
172    pub fn with_config_files(
173        mut self,
174        files: Vec<PathBuf>,
175        quiet: bool,
176        cwd: Option<PathBuf>,
177    ) -> Self {
178        self.config_load = ConfigLoadOptions {
179            selection: ConfigFileSelection::Files(files),
180            quiet,
181            cwd,
182        };
183        self
184    }
185}
186
187/// Startup config loading policy for a daemon.
188#[derive(Debug, Clone, PartialEq, Eq)]
189pub struct ConfigLoadOptions {
190    selection: ConfigFileSelection,
191    quiet: bool,
192    cwd: Option<PathBuf>,
193}
194
195impl ConfigLoadOptions {
196    /// Builds a config policy that performs no startup config loading.
197    #[must_use]
198    pub const fn disabled() -> Self {
199        Self {
200            selection: ConfigFileSelection::Disabled,
201            quiet: true,
202            cwd: None,
203        }
204    }
205
206    /// Returns the selected config files mode.
207    #[must_use]
208    pub const fn selection(&self) -> &ConfigFileSelection {
209        &self.selection
210    }
211
212    /// Returns whether missing files should be suppressed.
213    #[must_use]
214    pub const fn quiet(&self) -> bool {
215        self.quiet
216    }
217
218    /// Returns the startup client's current working directory.
219    #[must_use]
220    pub fn cwd(&self) -> Option<&Path> {
221        self.cwd.as_deref()
222    }
223}
224
225/// Config file selection mode for daemon startup.
226#[derive(Debug, Clone, PartialEq, Eq)]
227pub enum ConfigFileSelection {
228    /// Do not load config files.
229    Disabled,
230    /// Load RMUX default config files, with a filtered tmux config fallback.
231    Default,
232    /// Load the explicit `-f` files in order.
233    Files(Vec<PathBuf>),
234}
235
236/// RMUX daemon launcher — call [`bind`](Self::bind) to start listening.
237#[derive(Debug, Clone, PartialEq, Eq)]
238pub struct ServerDaemon {
239    config: DaemonConfig,
240}
241
242#[derive(Debug, Clone)]
243pub(crate) struct ShutdownHandle {
244    sender: Arc<StdMutex<Option<oneshot::Sender<()>>>>,
245}
246
247impl ShutdownHandle {
248    pub(crate) fn new() -> (Self, oneshot::Receiver<()>) {
249        let (sender, receiver) = oneshot::channel();
250        (
251            Self {
252                sender: Arc::new(StdMutex::new(Some(sender))),
253            },
254            receiver,
255        )
256    }
257
258    pub(crate) fn request_shutdown(&self) {
259        if let Some(sender) = self.sender.lock().expect("shutdown sender").take() {
260            let _ = sender.send(());
261        }
262    }
263}
264
265impl ServerDaemon {
266    /// Creates a daemon launcher for the given configuration.
267    #[must_use]
268    pub fn new(config: DaemonConfig) -> Self {
269        Self { config }
270    }
271
272    /// Binds the local IPC endpoint, starts accepting requests, and returns a handle.
273    pub async fn bind(self) -> io::Result<ServerHandle> {
274        #[cfg(unix)]
275        {
276            let bound_listener = bind_unix_listener_at(self.config.socket_path())?;
277            let (shutdown_handle, shutdown_receiver) = ShutdownHandle::new();
278            let signal_watcher = crate::signals::SignalWatcher::install()?;
279            let socket_path = self.config.socket_path().to_path_buf();
280            let owner_uid = real_user_id()?;
281            let serve_options = ServeOptions::new(
282                self.config.config_load().clone(),
283                self.config.subscription_limits(),
284                owner_uid,
285            )
286            .with_web_options(
287                self.config.web_port(),
288                self.config.web_frontend().map(str::to_owned),
289                self.config.web_required(),
290            )
291            .with_socket_identity(bound_listener.identity)
292            .with_server_signals(signal_watcher);
293
294            let task = tokio::spawn(listener::serve(
295                bound_listener.listener,
296                socket_path.clone(),
297                shutdown_handle.clone(),
298                shutdown_receiver,
299                serve_options,
300            ));
301
302            Ok(ServerHandle {
303                socket_path,
304                shutdown_handle,
305                task: Some(task),
306            })
307        }
308
309        #[cfg(windows)]
310        {
311            let endpoint = LocalEndpoint::from_path(self.config.socket_path().to_path_buf());
312            let listener = bind_windows_listener(&endpoint)?;
313            let (shutdown_handle, shutdown_receiver) = ShutdownHandle::new();
314            let socket_path = self.config.socket_path().to_path_buf();
315            let owner_uid = current_owner_uid();
316            let serve_options = ServeOptions::new(
317                self.config.config_load().clone(),
318                self.config.subscription_limits(),
319                owner_uid,
320            )
321            .with_web_options(
322                self.config.web_port(),
323                self.config.web_frontend().map(str::to_owned),
324                self.config.web_required(),
325            );
326
327            let task = tokio::spawn(listener::serve(
328                listener,
329                socket_path.clone(),
330                shutdown_handle.clone(),
331                shutdown_receiver,
332                serve_options,
333            ));
334
335            Ok(ServerHandle {
336                socket_path,
337                shutdown_handle,
338                task: Some(task),
339            })
340        }
341    }
342}
343
344#[cfg(windows)]
345fn bind_windows_listener(endpoint: &LocalEndpoint) -> io::Result<LocalListener> {
346    match LocalListener::bind(endpoint) {
347        Ok(listener) => Ok(listener),
348        Err(bind_error) => Err(windows_bind_error(endpoint, bind_error)),
349    }
350}
351
352#[cfg(windows)]
353fn windows_bind_error(endpoint: &LocalEndpoint, bind_error: io::Error) -> io::Error {
354    if windows_pipe_responds(endpoint) {
355        return io::Error::new(
356            io::ErrorKind::AddrInUse,
357            format!(
358                "Windows named pipe '{}' is already held by a responsive rmux-compatible server",
359                endpoint.as_path().display()
360            ),
361        );
362    }
363
364    io::Error::new(
365        bind_error.kind(),
366        format!(
367            "failed to bind Windows named pipe '{}': {bind_error}. Another process may still be holding this endpoint",
368            endpoint.as_path().display()
369        ),
370    )
371}
372
373#[cfg(windows)]
374fn windows_pipe_responds(endpoint: &LocalEndpoint) -> bool {
375    let endpoint = endpoint.clone();
376    std::thread::spawn(move || windows_protocol_probe(&endpoint).unwrap_or(false))
377        .join()
378        .unwrap_or(false)
379}
380
381#[cfg(windows)]
382fn windows_protocol_probe(endpoint: &LocalEndpoint) -> io::Result<bool> {
383    let mut stream = connect_blocking(endpoint, Duration::from_millis(100))?;
384    stream.set_write_timeout(Some(Duration::from_millis(100)))?;
385    stream.set_read_timeout(Some(Duration::from_millis(100)))?;
386
387    let request = Request::HasSession(HasSessionRequest {
388        target: SessionName::new("__rmux_probe__").map_err(io::Error::other)?,
389    });
390    let frame = encode_frame(&request).map_err(io::Error::other)?;
391    stream.write_all(&frame)?;
392    stream.flush()?;
393
394    let mut decoder = FrameDecoder::new();
395    let mut buffer = [0_u8; 512];
396    loop {
397        let bytes_read = match stream.read(&mut buffer) {
398            Ok(0) => return Ok(false),
399            Ok(bytes_read) => bytes_read,
400            Err(error) if error.kind() == io::ErrorKind::TimedOut => return Ok(false),
401            Err(error) => return Err(error),
402        };
403        decoder.push_bytes(&buffer[..bytes_read]);
404        match decoder.next_frame::<Response>() {
405            Ok(Some(Response::HasSession(_))) => return Ok(true),
406            Ok(Some(_response)) => return Ok(false),
407            Ok(None) => continue,
408            Err(RmuxError::IncompleteFrame { .. }) => continue,
409            Err(_error) => return Ok(false),
410        }
411    }
412}
413
414/// Handle to a running RMUX daemon; dropping it triggers shutdown.
415#[derive(Debug)]
416pub struct ServerHandle {
417    socket_path: PathBuf,
418    shutdown_handle: ShutdownHandle,
419    task: Option<JoinHandle<io::Result<()>>>,
420}
421
422impl ServerHandle {
423    /// Returns the bound local IPC endpoint path for the running daemon.
424    #[must_use]
425    pub fn socket_path(&self) -> &Path {
426        &self.socket_path
427    }
428
429    /// Waits for the daemon task to exit after an external shutdown request.
430    pub async fn wait(mut self) -> io::Result<()> {
431        if let Some(task) = self.task.take() {
432            return task.await.map_err(io::Error::other)?;
433        }
434
435        Ok(())
436    }
437
438    /// Requests shutdown and waits for socket cleanup to complete.
439    pub async fn shutdown(mut self) -> io::Result<()> {
440        self.request_shutdown();
441
442        if let Some(task) = self.task.take() {
443            return task.await.map_err(io::Error::other)?;
444        }
445
446        Ok(())
447    }
448
449    fn request_shutdown(&mut self) {
450        self.shutdown_handle.request_shutdown();
451    }
452}
453
454impl Drop for ServerHandle {
455    fn drop(&mut self) {
456        self.request_shutdown();
457    }
458}
459
460#[cfg(all(test, unix))]
461#[path = "daemon_tests/unix.rs"]
462mod tests;
463
464#[cfg(all(test, windows))]
465#[path = "daemon_tests/windows.rs"]
466mod tests;
rmux_server/daemon.rs

rmux_server/
daemon.rs
