1use std::env;
4use std::fmt;
5use std::io;
6use std::path::{Path, PathBuf};
7use std::process::{Command, Stdio};
8use std::time::{Duration, Instant};
9#[cfg(target_os = "linux")]
10use std::{fs::File, io::Read, os::fd::AsRawFd};
11
12#[cfg(windows)]
13use rmux_proto::DaemonStatusResponse;
14use rmux_proto::{Response, RmuxError};
15#[cfg(unix)]
16use rmux_sdk::bootstrap::startup_unix::{
17 connect_or_start_with, StartupError, StartupOutcome, DEFAULT_STARTUP_DEADLINE,
18 STARTUP_POLL_INTERVAL,
19};
20#[cfg(windows)]
21use rmux_sdk::bootstrap::startup_windows::{
22 connect_or_start_blocking_with, StartupError, StartupOutcome, DEFAULT_STARTUP_DEADLINE,
23 STARTUP_POLL_INTERVAL,
24};
25
26use crate::shell_quote::shell_quote_path;
27#[cfg(any(all(test, unix), not(any(unix, windows))))]
28use crate::ConnectResult;
29use crate::{default_socket_path, upgrade, ClientError, Connection};
30
31mod upgrade_restart;
32
33#[cfg(target_os = "linux")]
34const STARTUP_READY_EVENT_TIMEOUT: Duration = Duration::from_millis(20);
35#[cfg(windows)]
36const STARTUP_READY_EVENT_TIMEOUT: Duration = Duration::from_secs(2);
37#[cfg(not(any(unix, windows)))]
38const AUTO_START_TIMEOUT: Duration = Duration::from_secs(5);
39#[cfg(not(any(unix, windows)))]
40const POLL_INTERVAL: Duration = Duration::from_millis(50);
41
42pub const INTERNAL_DAEMON_FLAG: &str = "--__internal-daemon";
47
48const BINARY_OVERRIDE_ENV: &str = "RMUX_INTERNAL_BINARY_PATH";
49const BINARY_OVERRIDE_TEST_OPT_IN_ENV: &str = "RMUX_ALLOW_INTERNAL_BINARY_OVERRIDE";
50#[derive(Debug, Clone, PartialEq, Eq)]
52pub struct AutoStartConfig {
53 selection: AutoStartConfigSelection,
54 quiet: bool,
55 cwd: Option<PathBuf>,
56 web_frontend: Option<String>,
57 web_port: Option<u16>,
58 web_required: bool,
59 binary_override: Option<PathBuf>,
60}
61
62impl AutoStartConfig {
63 #[must_use]
65 pub const fn disabled() -> Self {
66 Self {
67 selection: AutoStartConfigSelection::Disabled,
68 quiet: true,
69 cwd: None,
70 web_frontend: None,
71 web_port: None,
72 web_required: false,
73 binary_override: None,
74 }
75 }
76
77 #[must_use]
79 pub fn default_files(quiet: bool, cwd: Option<PathBuf>) -> Self {
80 Self {
81 selection: AutoStartConfigSelection::Default,
82 quiet,
83 cwd,
84 web_frontend: None,
85 web_port: None,
86 web_required: false,
87 binary_override: None,
88 }
89 }
90
91 #[must_use]
93 pub fn custom_files(files: Vec<PathBuf>, quiet: bool, cwd: Option<PathBuf>) -> Self {
94 Self {
95 selection: AutoStartConfigSelection::Files(files),
96 quiet,
97 cwd,
98 web_frontend: None,
99 web_port: None,
100 web_required: false,
101 binary_override: None,
102 }
103 }
104
105 #[must_use]
107 pub const fn with_web_port(mut self, port: u16) -> Self {
108 self.web_port = Some(port);
109 self.web_required = true;
110 self
111 }
112
113 #[must_use]
115 pub fn with_web_frontend(mut self, frontend: String) -> Self {
116 self.web_frontend = Some(frontend);
117 self.web_required = true;
118 self
119 }
120
121 #[must_use]
123 pub const fn with_web_required(mut self) -> Self {
124 self.web_required = true;
125 self
126 }
127
128 #[must_use]
130 pub fn with_binary_override(mut self, binary_path: PathBuf) -> Self {
131 self.binary_override = Some(binary_path);
132 self
133 }
134
135 #[cfg(not(windows))]
136 #[cfg(not(any(unix, windows)))]
137 fn loads_startup_config(&self) -> bool {
138 !matches!(self.selection, AutoStartConfigSelection::Disabled)
139 }
140
141 fn append_hidden_daemon_args(&self, command: &mut Command) {
142 match &self.selection {
143 AutoStartConfigSelection::Disabled => {}
144 AutoStartConfigSelection::Default => {
145 command.arg("--config-default");
146 }
147 AutoStartConfigSelection::Files(files) => {
148 for file in files {
149 command.arg("--config-file").arg(file);
150 }
151 }
152 }
153
154 if self.quiet {
155 command.arg("--config-quiet");
156 }
157 if let Some(cwd) = &self.cwd {
158 command.arg("--config-cwd").arg(cwd);
159 }
160 if let Some(port) = self.web_port {
161 command.arg("--web-port").arg(port.to_string());
162 }
163 if let Some(frontend) = &self.web_frontend {
164 command.arg("--frontend-url").arg(frontend);
165 }
166 }
167}
168
169#[derive(Debug, Clone, PartialEq, Eq)]
171pub enum AutoStartConfigSelection {
172 Disabled,
174 Default,
176 Files(Vec<PathBuf>),
178}
179
180#[derive(Debug, Clone, Copy, PartialEq, Eq)]
182pub enum ServerConnectionProvenance {
183 StartedByCaller,
185 JoinedExisting,
187}
188
189pub struct EnsuredServerConnection {
191 connection: Connection,
192 provenance: ServerConnectionProvenance,
193}
194
195impl EnsuredServerConnection {
196 fn new(connection: Connection, provenance: ServerConnectionProvenance) -> Self {
197 Self {
198 connection,
199 provenance,
200 }
201 }
202
203 #[must_use]
205 pub const fn provenance(&self) -> ServerConnectionProvenance {
206 self.provenance
207 }
208
209 #[must_use]
211 pub fn into_connection(self) -> Connection {
212 self.connection
213 }
214}
215
216pub fn ensure_server_running(socket_path: &Path) -> Result<Connection, AutoStartError> {
223 ensure_server_running_with_config(socket_path, AutoStartConfig::disabled())
224}
225
226pub fn ensure_server_running_with_config(
228 socket_path: &Path,
229 config: AutoStartConfig,
230) -> Result<Connection, AutoStartError> {
231 ensure_server_running_with_config_outcome(socket_path, config)
232 .map(EnsuredServerConnection::into_connection)
233}
234
235#[cfg(windows)]
242pub fn ensure_server_running_with_config_outcome(
243 socket_path: &Path,
244 config: AutoStartConfig,
245) -> Result<EnsuredServerConnection, AutoStartError> {
246 ensure_server_running_windows(socket_path, config)
247}
248
249#[cfg(unix)]
256pub fn ensure_server_running_with_config_outcome(
257 socket_path: &Path,
258 config: AutoStartConfig,
259) -> Result<EnsuredServerConnection, AutoStartError> {
260 ensure_server_running_unix(socket_path, config)
261}
262
263#[cfg(not(any(unix, windows)))]
265pub fn ensure_server_running_with_config_outcome(
266 socket_path: &Path,
267 config: AutoStartConfig,
268) -> Result<EnsuredServerConnection, AutoStartError> {
269 ensure_server_running_polling(socket_path, config)
270}
271
272#[cfg(unix)]
273fn ensure_server_running_unix(
274 socket_path: &Path,
275 config: AutoStartConfig,
276) -> Result<EnsuredServerConnection, AutoStartError> {
277 let binary_path = rmux_binary_path(&config).map_err(AutoStartError::BinaryPath)?;
278 let launcher_binary_path = binary_path.clone();
279 let launcher_socket_path = socket_path.to_path_buf();
280 let launcher_config = config.clone();
281
282 let runtime = tokio::runtime::Builder::new_current_thread()
283 .enable_all()
284 .build()
285 .map_err(|error| AutoStartError::Client(ClientError::Io(error)))?;
286 let outcome = runtime.block_on(connect_or_start_with(
287 socket_path,
288 move || async move {
289 spawn_hidden_daemon_for(
290 &launcher_binary_path,
291 &launcher_socket_path,
292 &launcher_config,
293 )
294 },
295 DEFAULT_STARTUP_DEADLINE,
296 STARTUP_POLL_INTERVAL,
297 ));
298
299 let outcome =
300 outcome.map_err(|error| auto_start_error_from_startup(error, &binary_path, socket_path))?;
301 let provenance = startup_outcome_provenance(&outcome);
302 let connection = startup_outcome_into_connection(outcome)?;
303
304 let connection = probe_connected_server(connection, &config, socket_path)?;
305 let connection = upgrade_restart::ensure_daemon_fresh_or_restart(
306 connection,
307 socket_path,
308 &binary_path,
309 &config,
310 )?;
311 Ok(EnsuredServerConnection::new(connection, provenance))
312}
313
314#[cfg(unix)]
315fn startup_outcome_into_connection(outcome: StartupOutcome) -> Result<Connection, AutoStartError> {
316 let stream = outcome
317 .into_stream()
318 .into_std()
319 .map_err(|error| AutoStartError::Client(ClientError::Io(error)))?;
320 stream
321 .set_nonblocking(false)
322 .map_err(|error| AutoStartError::Client(ClientError::Io(error)))?;
323 Connection::new(stream).map_err(AutoStartError::Client)
324}
325
326#[cfg(windows)]
327fn ensure_server_running_windows(
328 socket_path: &Path,
329 config: AutoStartConfig,
330) -> Result<EnsuredServerConnection, AutoStartError> {
331 let binary_path = rmux_binary_path(&config).map_err(AutoStartError::BinaryPath)?;
332 let launcher_binary_path = binary_path.clone();
333 let launcher_socket_path = socket_path.to_path_buf();
334 let launcher_config = config.clone();
335
336 let outcome = connect_or_start_blocking_with(
337 socket_path,
338 move || {
339 spawn_hidden_daemon_for(
340 &launcher_binary_path,
341 &launcher_socket_path,
342 &launcher_config,
343 )
344 },
345 DEFAULT_STARTUP_DEADLINE,
346 STARTUP_POLL_INTERVAL,
347 );
348
349 let outcome =
350 outcome.map_err(|error| auto_start_error_from_startup(error, &binary_path, socket_path))?;
351 let provenance = startup_outcome_provenance(&outcome);
352 let connection = startup_outcome_into_connection(outcome)?;
353 let (connection, readiness_status) =
354 probe_connected_server_windows(connection, &config, socket_path)?;
355 let connection = upgrade_restart::ensure_daemon_fresh_or_restart_after_windows_readiness(
356 connection,
357 socket_path,
358 &binary_path,
359 &config,
360 readiness_status,
361 )?;
362 Ok(EnsuredServerConnection::new(connection, provenance))
363}
364
365#[cfg(any(unix, windows))]
366fn startup_outcome_provenance(outcome: &StartupOutcome) -> ServerConnectionProvenance {
367 if outcome.is_owner() {
368 ServerConnectionProvenance::StartedByCaller
369 } else {
370 ServerConnectionProvenance::JoinedExisting
371 }
372}
373
374#[cfg(windows)]
375fn startup_outcome_into_connection(outcome: StartupOutcome) -> Result<Connection, AutoStartError> {
376 Connection::new(outcome.into_stream()).map_err(AutoStartError::Client)
377}
378
379#[cfg(windows)]
380fn probe_connected_server_windows(
381 mut connection: Connection,
382 _config: &AutoStartConfig,
383 socket_path: &Path,
384) -> Result<(Connection, Option<DaemonStatusResponse>), AutoStartError> {
385 let deadline = Instant::now() + DEFAULT_STARTUP_DEADLINE;
386 let mut poll_attempt = 0_u32;
387 loop {
388 match probe_server_readiness_status(&mut connection) {
389 Ok(status) => return Ok((connection, status)),
390 Err(ClientError::Protocol(RmuxError::UnsupportedWireVersion { got, .. })) => {
391 return Err(AutoStartError::IncompatibleDaemon {
392 socket_path: socket_path.to_path_buf(),
393 message: upgrade::incompatible_daemon_message(&upgrade::IncompatibleDaemon {
394 daemon_version: None,
395 daemon_wire_version: Some(got),
396 }),
397 });
398 }
399 Err(error) if is_transient_connect_error(&error) && Instant::now() < deadline => {
400 let remaining = deadline.saturating_duration_since(Instant::now());
401 std::thread::sleep(startup_readiness_poll_sleep(&mut poll_attempt, remaining));
402 }
403 Err(error) => return Err(AutoStartError::Client(error)),
404 }
405 }
406}
407
408#[cfg(windows)]
409fn probe_server_readiness_status(
410 connection: &mut Connection,
411) -> Result<Option<DaemonStatusResponse>, ClientError> {
412 let response = connection.daemon_status()?;
413 match response {
414 Response::DaemonStatus(status) if status.config_loading => {
415 Err(ClientError::Io(io::Error::new(
416 io::ErrorKind::WouldBlock,
417 "daemon is still loading startup config",
418 )))
419 }
420 Response::DaemonStatus(status) => Ok(Some(status)),
421 Response::Error(_) => Ok(None),
422 other => Err(ClientError::Protocol(rmux_proto::RmuxError::Server(
423 format!("unexpected readiness response: {other:?}"),
424 ))),
425 }
426}
427
428fn probe_connected_server(
429 mut connection: Connection,
430 _config: &AutoStartConfig,
431 socket_path: &Path,
432) -> Result<Connection, AutoStartError> {
433 let deadline = Instant::now() + DEFAULT_STARTUP_DEADLINE;
434 let mut poll_attempt = 0_u32;
435 loop {
436 match probe_server_readiness(&mut connection) {
437 Ok(()) => return Ok(connection),
438 Err(ClientError::Protocol(RmuxError::UnsupportedWireVersion { got, .. })) => {
439 return Err(AutoStartError::IncompatibleDaemon {
440 socket_path: socket_path.to_path_buf(),
441 message: upgrade::incompatible_daemon_message(&upgrade::IncompatibleDaemon {
442 daemon_version: None,
443 daemon_wire_version: Some(got),
444 }),
445 });
446 }
447 Err(error) if is_transient_connect_error(&error) && Instant::now() < deadline => {
448 let remaining = deadline.saturating_duration_since(Instant::now());
449 std::thread::sleep(startup_readiness_poll_sleep(&mut poll_attempt, remaining));
450 }
451 Err(error) => return Err(AutoStartError::Client(error)),
452 }
453 }
454}
455
456fn startup_readiness_poll_sleep(poll_attempt: &mut u32, remaining: Duration) -> Duration {
457 #[cfg(windows)]
458 {
459 const INITIAL_POLL_MILLIS: u64 = 1;
460
461 let shift = (*poll_attempt).min(6);
462 *poll_attempt = (*poll_attempt).saturating_add(1);
463 let millis = INITIAL_POLL_MILLIS
464 .checked_shl(shift)
465 .unwrap_or(u64::MAX)
466 .min(STARTUP_POLL_INTERVAL.as_millis() as u64);
467 Duration::from_millis(millis).min(remaining)
468 }
469
470 #[cfg(not(windows))]
471 {
472 const INITIAL_POLL_MILLIS: u64 = 1;
473
474 let shift = (*poll_attempt).min(6);
475 *poll_attempt = (*poll_attempt).saturating_add(1);
476 let millis = INITIAL_POLL_MILLIS
477 .checked_shl(shift)
478 .unwrap_or(u64::MAX)
479 .min(STARTUP_POLL_INTERVAL.as_millis() as u64);
480 Duration::from_millis(millis).min(remaining)
481 }
482}
483
484#[cfg(unix)]
485fn auto_start_error_from_startup(
486 error: StartupError,
487 binary_path: &Path,
488 socket_path: &Path,
489) -> AutoStartError {
490 match error {
491 StartupError::Launcher { source } => AutoStartError::Launch {
492 path: binary_path.to_path_buf(),
493 error: source,
494 },
495 StartupError::StartupTimeout { waited, .. } => AutoStartError::TimedOut {
496 socket_path: socket_path.to_path_buf(),
497 waited,
498 },
499 error => AutoStartError::Client(ClientError::Io(io::Error::new(
500 startup_error_kind(&error),
501 error.to_string(),
502 ))),
503 }
504}
505
506#[cfg(windows)]
507fn auto_start_error_from_startup(
508 error: StartupError,
509 binary_path: &Path,
510 socket_path: &Path,
511) -> AutoStartError {
512 match error {
513 StartupError::Launcher { source } => AutoStartError::Launch {
514 path: binary_path.to_path_buf(),
515 error: source,
516 },
517 StartupError::StartupTimeout { waited, .. } => AutoStartError::TimedOut {
518 socket_path: socket_path.to_path_buf(),
519 waited,
520 },
521 error => AutoStartError::Client(ClientError::Io(io::Error::new(
522 startup_error_kind(&error),
523 error.to_string(),
524 ))),
525 }
526}
527
528#[cfg(unix)]
529fn startup_error_kind(error: &StartupError) -> io::ErrorKind {
530 match error {
531 StartupError::InvalidPath { .. } | StartupError::SymlinkRejected { .. } => {
532 io::ErrorKind::InvalidInput
533 }
534 StartupError::UnsafeOwner { .. }
535 | StartupError::UnsafePermissions { .. }
536 | StartupError::PeerCredentialMismatch { .. } => io::ErrorKind::PermissionDenied,
537 StartupError::Lock { source, .. } | StartupError::Filesystem { source, .. } => {
538 source.kind()
539 }
540 StartupError::Launcher { source } => source.kind(),
541 StartupError::StartupTimeout { .. } => io::ErrorKind::TimedOut,
542 }
543}
544
545#[cfg(windows)]
546fn startup_error_kind(error: &StartupError) -> io::ErrorKind {
547 match error {
548 StartupError::InvalidPipeName { .. } | StartupError::InvalidMutexName { .. } => {
549 io::ErrorKind::InvalidInput
550 }
551 StartupError::MutexAccessDenied { .. } | StartupError::PipeAccessDenied { .. } => {
552 io::ErrorKind::PermissionDenied
553 }
554 StartupError::MutexTimeout { .. }
555 | StartupError::PipeBusy { .. }
556 | StartupError::StartupTimeout { .. } => io::ErrorKind::TimedOut,
557 StartupError::PipeNotFound { .. } | StartupError::PipeNoData { .. } => {
558 io::ErrorKind::NotFound
559 }
560 StartupError::Mutex { source, .. } | StartupError::PipeIo { source, .. } => source.kind(),
561 StartupError::Launcher { source } => source.kind(),
562 }
563}
564
565#[cfg(not(any(unix, windows)))]
566fn ensure_server_running_polling(
567 socket_path: &Path,
568 config: AutoStartConfig,
569) -> Result<EnsuredServerConnection, AutoStartError> {
570 if config.loads_startup_config() {
571 return ensure_server_running_with_probe_outcome(
572 socket_path,
573 AUTO_START_TIMEOUT,
574 POLL_INTERVAL,
575 || crate::connect_or_absent(socket_path),
576 || launch_hidden_daemon(socket_path, &config),
577 |_| Ok(()),
578 );
579 }
580
581 ensure_server_running_with_probe_outcome(
582 socket_path,
583 AUTO_START_TIMEOUT,
584 POLL_INTERVAL,
585 || crate::connect_or_absent(socket_path),
586 || launch_hidden_daemon(socket_path, &config),
587 probe_server_readiness,
588 )
589}
590
591#[derive(Debug)]
593pub enum AutoStartError {
594 Client(ClientError),
596 BinaryPath(io::Error),
598 Launch {
600 path: PathBuf,
602 error: io::Error,
604 },
605 IncompatibleDaemon {
607 socket_path: PathBuf,
609 message: String,
611 },
612 TimedOut {
614 socket_path: PathBuf,
616 waited: Duration,
618 },
619}
620
621impl fmt::Display for AutoStartError {
622 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
623 match self {
624 Self::Client(error) => write!(formatter, "{error}"),
625 Self::BinaryPath(error) => {
626 write!(formatter, "failed to resolve rmux binary path: {error}")
627 }
628 Self::Launch { path, error } => {
629 write!(
630 formatter,
631 "failed to launch hidden rmux daemon '{}': {error}",
632 path.display()
633 )
634 }
635 Self::IncompatibleDaemon {
636 socket_path,
637 message,
638 } => write!(
639 formatter,
640 "rmux: {message} on '{}'.\nrmux: run `{}` to stop it, then retry.",
641 socket_path.display(),
642 incompatible_daemon_kill_server_command(socket_path)
643 ),
644 Self::TimedOut {
645 socket_path,
646 waited,
647 } => write!(
648 formatter,
649 "timed out after {}s waiting for rmux server socket '{}'. \
650 The hidden daemon may have exited before creating the socket; run `{}` to surface startup errors.",
651 waited.as_secs(),
652 socket_path.display(),
653 diagnostic_start_server_command(socket_path)
654 ),
655 }
656 }
657}
658
659impl std::error::Error for AutoStartError {
660 fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
661 match self {
662 Self::Client(error) => Some(error),
663 Self::BinaryPath(error) => Some(error),
664 Self::Launch { error, .. } => Some(error),
665 Self::IncompatibleDaemon { .. } => None,
666 Self::TimedOut { .. } => None,
667 }
668 }
669}
670
671impl From<ClientError> for AutoStartError {
672 fn from(error: ClientError) -> Self {
673 Self::Client(error)
674 }
675}
676
677#[cfg(not(any(unix, windows)))]
678fn ensure_server_running_with<ConnectFn, LaunchFn>(
679 socket_path: &Path,
680 timeout: Duration,
681 poll_interval: Duration,
682 connect: ConnectFn,
683 launch: LaunchFn,
684) -> Result<Connection, AutoStartError>
685where
686 ConnectFn: FnMut() -> Result<ConnectResult, ClientError>,
687 LaunchFn: FnMut() -> Result<(), AutoStartError>,
688{
689 ensure_server_running_with_probe(
690 socket_path,
691 timeout,
692 poll_interval,
693 connect,
694 launch,
695 probe_server_readiness,
696 )
697}
698
699#[cfg(any(all(test, unix), not(any(unix, windows))))]
700fn ensure_server_running_with_probe<ConnectFn, LaunchFn, ProbeFn>(
701 socket_path: &Path,
702 timeout: Duration,
703 poll_interval: Duration,
704 connect: ConnectFn,
705 launch: LaunchFn,
706 probe: ProbeFn,
707) -> Result<Connection, AutoStartError>
708where
709 ConnectFn: FnMut() -> Result<ConnectResult, ClientError>,
710 LaunchFn: FnMut() -> Result<(), AutoStartError>,
711 ProbeFn: FnMut(&mut Connection) -> Result<(), ClientError>,
712{
713 ensure_server_running_with_probe_outcome(
714 socket_path,
715 timeout,
716 poll_interval,
717 connect,
718 launch,
719 probe,
720 )
721 .map(EnsuredServerConnection::into_connection)
722}
723
724#[cfg(any(all(test, unix), not(any(unix, windows))))]
725fn ensure_server_running_with_probe_outcome<ConnectFn, LaunchFn, ProbeFn>(
726 socket_path: &Path,
727 timeout: Duration,
728 poll_interval: Duration,
729 mut connect: ConnectFn,
730 mut launch: LaunchFn,
731 mut probe: ProbeFn,
732) -> Result<EnsuredServerConnection, AutoStartError>
733where
734 ConnectFn: FnMut() -> Result<ConnectResult, ClientError>,
735 LaunchFn: FnMut() -> Result<(), AutoStartError>,
736 ProbeFn: FnMut(&mut Connection) -> Result<(), ClientError>,
737{
738 match connect().map_err(AutoStartError::Client)? {
739 ConnectResult::Connected(mut connection) => {
740 probe(&mut connection).map_err(AutoStartError::Client)?;
741 return Ok(EnsuredServerConnection::new(
742 connection,
743 ServerConnectionProvenance::JoinedExisting,
744 ));
745 }
746 ConnectResult::Absent => {}
747 }
748
749 launch()?;
750 let connection = wait_for_server(
751 socket_path,
752 timeout,
753 poll_interval,
754 &mut connect,
755 &mut probe,
756 )?;
757 Ok(EnsuredServerConnection::new(
758 connection,
759 ServerConnectionProvenance::StartedByCaller,
760 ))
761}
762
763#[cfg(any(all(test, unix), not(any(unix, windows))))]
764fn wait_for_server<ConnectFn, ProbeFn>(
765 socket_path: &Path,
766 timeout: Duration,
767 poll_interval: Duration,
768 connect: &mut ConnectFn,
769 probe: &mut ProbeFn,
770) -> Result<Connection, AutoStartError>
771where
772 ConnectFn: FnMut() -> Result<ConnectResult, ClientError>,
773 ProbeFn: FnMut(&mut Connection) -> Result<(), ClientError>,
774{
775 let start = Instant::now();
776 let deadline = start + timeout;
777
778 loop {
779 match connect() {
780 Ok(crate::ConnectResult::Connected(mut connection)) => match probe(&mut connection) {
781 Ok(()) => return Ok(connection),
782 Err(error) if is_transient_connect_error(&error) => {}
783 Err(error) => return Err(AutoStartError::Client(error)),
784 },
785 Ok(crate::ConnectResult::Absent) => {}
786 Err(error) if is_transient_connect_error(&error) => {}
787 Err(error) => return Err(AutoStartError::Client(error)),
788 }
789
790 let now = Instant::now();
791 if now >= deadline {
792 return Err(AutoStartError::TimedOut {
793 socket_path: socket_path.to_path_buf(),
794 waited: timeout,
795 });
796 }
797
798 std::thread::sleep(poll_interval.min(deadline.saturating_duration_since(now)));
799 }
800}
801
802fn is_transient_connect_error(error: &ClientError) -> bool {
803 matches!(
804 error,
805 ClientError::Io(io_error)
806 if matches!(
807 io_error.kind(),
808 io::ErrorKind::WouldBlock
809 | io::ErrorKind::Interrupted
810 | io::ErrorKind::TimedOut
811 )
812 )
813}
814
815fn incompatible_daemon_kill_server_command(socket_path: &Path) -> String {
816 if default_socket_path()
817 .ok()
818 .as_deref()
819 .is_some_and(|default_path| default_path == socket_path)
820 {
821 return "rmux kill-server".to_owned();
822 }
823
824 format!("rmux -S {} kill-server", shell_quote_path(socket_path))
825}
826
827fn diagnostic_start_server_command(socket_path: &Path) -> String {
828 if default_socket_path()
829 .ok()
830 .as_deref()
831 .is_some_and(|default_path| default_path == socket_path)
832 {
833 return "rmux start-server".to_owned();
834 }
835
836 format!("rmux -S {} start-server", shell_quote_path(socket_path))
837}
838
839fn probe_server_readiness(connection: &mut Connection) -> Result<(), ClientError> {
840 let response = connection.daemon_status()?;
841 match response {
842 Response::DaemonStatus(status) if status.config_loading => {
843 Err(ClientError::Io(io::Error::new(
844 io::ErrorKind::WouldBlock,
845 "daemon is still loading startup config",
846 )))
847 }
848 Response::DaemonStatus(_) => Ok(()),
849 Response::Error(_) => Ok(()),
850 other => Err(ClientError::Protocol(rmux_proto::RmuxError::Server(
851 format!("unexpected readiness response: {other:?}"),
852 ))),
853 }
854}
855
856#[cfg(not(any(unix, windows)))]
857fn launch_hidden_daemon(
858 socket_path: &Path,
859 config: &AutoStartConfig,
860) -> Result<(), AutoStartError> {
861 let binary_path = rmux_binary_path(config).map_err(AutoStartError::BinaryPath)?;
862 spawn_hidden_daemon_for(&binary_path, socket_path, config).map_err(|error| {
863 AutoStartError::Launch {
864 path: binary_path,
865 error,
866 }
867 })
868}
869
870fn spawn_hidden_daemon_for(
871 binary_path: &Path,
872 socket_path: &Path,
873 config: &AutoStartConfig,
874) -> io::Result<()> {
875 #[cfg(target_os = "linux")]
876 {
877 spawn_hidden_daemon_for_linux(binary_path, socket_path, config)
878 }
879
880 #[cfg(not(target_os = "linux"))]
881 {
882 spawn_hidden_daemon_for_polling(binary_path, socket_path, config)
883 }
884}
885
886#[cfg(not(target_os = "linux"))]
887fn spawn_hidden_daemon_for_polling(
888 binary_path: &Path,
889 socket_path: &Path,
890 config: &AutoStartConfig,
891) -> io::Result<()> {
892 #[cfg(windows)]
893 {
894 spawn_hidden_daemon_for_windows(binary_path, socket_path, config)
895 }
896
897 #[cfg(not(windows))]
898 {
899 let command = hidden_daemon_command(binary_path, socket_path, config, true);
900 spawn_hidden_daemon(command)
901 }
902}
903
904#[cfg(windows)]
905fn spawn_hidden_daemon_for_windows(
906 binary_path: &Path,
907 socket_path: &Path,
908 config: &AutoStartConfig,
909) -> io::Result<()> {
910 let ready = rmux_os::daemon::StartupReadyEvent::new()?;
911 let mut command = hidden_daemon_command(binary_path, socket_path, config, true);
912 append_startup_ready_event(&mut command, &ready);
913 spawn_hidden_daemon(command)?;
914 let _ = ready.wait(STARTUP_READY_EVENT_TIMEOUT);
915 Ok(())
916}
917
918#[cfg(windows)]
919fn append_startup_ready_event(command: &mut Command, ready: &rmux_os::daemon::StartupReadyEvent) {
920 command.arg("--startup-ready-event").arg(ready.name());
921}
922
923#[cfg(target_os = "linux")]
924fn spawn_hidden_daemon_for_linux(
925 binary_path: &Path,
926 socket_path: &Path,
927 config: &AutoStartConfig,
928) -> io::Result<()> {
929 let mut ready = StartupReadyEvent::new()?;
930 let mut command =
931 hidden_daemon_command_preserving_fd(binary_path, socket_path, config, true, ready.raw_fd());
932 ready.append_hidden_daemon_args(&mut command);
933 spawn_hidden_daemon(command)?;
934 ready.wait_for_signal(STARTUP_READY_EVENT_TIMEOUT);
935 Ok(())
936}
937
938#[cfg(target_os = "linux")]
939struct StartupReadyEvent {
940 file: File,
941}
942
943#[cfg(target_os = "linux")]
944impl StartupReadyEvent {
945 fn new() -> io::Result<Self> {
946 let fd = rustix::event::eventfd(
947 0,
948 rustix::event::EventfdFlags::NONBLOCK | rustix::event::EventfdFlags::CLOEXEC,
949 )
950 .map_err(io::Error::from)?;
951 Ok(Self { file: fd.into() })
952 }
953
954 fn append_hidden_daemon_args(&self, command: &mut Command) {
955 command
956 .arg("--startup-ready-fd")
957 .arg(self.file.as_raw_fd().to_string());
958 }
959
960 fn raw_fd(&self) -> i32 {
961 self.file.as_raw_fd()
962 }
963
964 fn wait_for_signal(&mut self, timeout: Duration) {
965 let deadline = Instant::now() + timeout;
966 let mut bytes = [0_u8; 8];
967 loop {
968 match self.file.read_exact(&mut bytes) {
969 Ok(()) => return,
970 Err(error)
971 if matches!(
972 error.kind(),
973 io::ErrorKind::WouldBlock | io::ErrorKind::Interrupted
974 ) && Instant::now() < deadline =>
975 {
976 std::thread::sleep(Duration::from_millis(1));
977 }
978 Err(_) => return,
979 }
980 }
981 }
982}
983
984#[cfg(target_os = "linux")]
985fn hidden_daemon_command_preserving_fd(
986 binary_path: &Path,
987 socket_path: &Path,
988 config: &AutoStartConfig,
989 allow_job_breakaway: bool,
990 preserved_fd: i32,
991) -> Command {
992 let mut command = hidden_daemon_command_base(binary_path, socket_path, config);
993 rmux_os::daemon::configure_hidden_daemon_command_preserving_fds(
994 &mut command,
995 allow_job_breakaway,
996 &[preserved_fd],
997 );
998 command
999}
1000
1001#[cfg(not(target_os = "linux"))]
1002fn hidden_daemon_command(
1003 binary_path: &Path,
1004 socket_path: &Path,
1005 config: &AutoStartConfig,
1006 allow_job_breakaway: bool,
1007) -> Command {
1008 let mut command = hidden_daemon_command_base(binary_path, socket_path, config);
1009 rmux_os::daemon::configure_hidden_daemon_command(&mut command, allow_job_breakaway);
1010 command
1011}
1012
1013fn hidden_daemon_command_base(
1014 binary_path: &Path,
1015 socket_path: &Path,
1016 config: &AutoStartConfig,
1017) -> Command {
1018 let mut command = Command::new(binary_path);
1019 command
1020 .arg(INTERNAL_DAEMON_FLAG)
1021 .arg(socket_path)
1022 .stdin(Stdio::null())
1023 .stdout(Stdio::null())
1024 .stderr(Stdio::null());
1025 config.append_hidden_daemon_args(&mut command);
1026 command
1027}
1028
1029fn spawn_hidden_daemon(mut command: Command) -> io::Result<()> {
1030 let child = rmux_os::daemon::spawn_hidden_daemon_command_requiring_job_breakaway(&mut command)?;
1031 drop(child);
1034 Ok(())
1035}
1036
1037fn rmux_binary_path(config: &AutoStartConfig) -> io::Result<PathBuf> {
1038 if let Some(path) = &config.binary_override {
1039 return Ok(path.clone());
1040 }
1041
1042 let current_exe = env::current_exe()?;
1043 let resolved_exe = std::fs::canonicalize(¤t_exe).ok();
1044 match env::var_os(BINARY_OVERRIDE_ENV).filter(|_| binary_override_enabled_for_tests()) {
1045 Some(path) => Ok(PathBuf::from(path)),
1046 None => Ok(hidden_daemon_binary_path_for_executable_paths(
1047 ¤t_exe,
1048 resolved_exe.as_deref(),
1049 config,
1050 )
1051 .unwrap_or(current_exe)),
1052 }
1053}
1054
1055fn binary_override_enabled_for_tests() -> bool {
1056 cfg!(debug_assertions)
1057 && env::var_os(BINARY_OVERRIDE_TEST_OPT_IN_ENV).is_some_and(|value| value == "1")
1058}
1059
1060#[cfg(all(test, unix))]
1061fn hidden_daemon_binary_path(current_exe: &Path) -> Option<PathBuf> {
1062 hidden_daemon_binary_path_for_executable_paths(current_exe, None, &AutoStartConfig::disabled())
1063}
1064
1065fn hidden_daemon_binary_path_for_executable_paths(
1066 current_exe: &Path,
1067 resolved_exe: Option<&Path>,
1068 config: &AutoStartConfig,
1069) -> Option<PathBuf> {
1070 hidden_daemon_binary_path_for_config(current_exe, config).or_else(|| {
1071 resolved_exe.and_then(|path| hidden_daemon_binary_path_for_config(path, config))
1072 })
1073}
1074
1075fn hidden_daemon_binary_path_for_config(
1076 current_exe: &Path,
1077 config: &AutoStartConfig,
1078) -> Option<PathBuf> {
1079 if config.web_required {
1080 return None;
1081 }
1082 let file_stem = current_exe.file_stem()?.to_str()?;
1083 if file_stem == "rmux-daemon" {
1084 return None;
1085 }
1086
1087 let mut candidate = current_exe.to_path_buf();
1088 let daemon_file_name = match current_exe
1089 .extension()
1090 .and_then(|extension| extension.to_str())
1091 {
1092 Some(extension) if !extension.is_empty() => format!("rmux-daemon.{extension}"),
1093 _ => "rmux-daemon".to_owned(),
1094 };
1095 candidate.set_file_name(daemon_file_name);
1096 candidate.is_file().then_some(candidate)
1097}
1098
1099#[cfg(all(test, unix))]
1100#[path = "auto_start/tests.rs"]
1101mod tests;
1102
1103#[cfg(all(test, windows))]
1104mod windows_tests {
1105 use std::time::Duration;
1106
1107 use super::{startup_readiness_poll_sleep, STARTUP_POLL_INTERVAL};
1108
1109 #[test]
1110 fn windows_startup_readiness_poll_uses_short_backoff() {
1111 let mut attempt = 0;
1112 let remaining = Duration::from_secs(1);
1113
1114 let sleeps = (0..8)
1115 .map(|_| startup_readiness_poll_sleep(&mut attempt, remaining))
1116 .collect::<Vec<_>>();
1117
1118 assert_eq!(
1119 sleeps,
1120 [
1121 Duration::from_millis(1),
1122 Duration::from_millis(2),
1123 Duration::from_millis(4),
1124 Duration::from_millis(8),
1125 Duration::from_millis(16),
1126 Duration::from_millis(32),
1127 STARTUP_POLL_INTERVAL,
1128 STARTUP_POLL_INTERVAL,
1129 ]
1130 );
1131 }
1132
1133 #[test]
1134 fn windows_startup_readiness_poll_respects_remaining_deadline() {
1135 let mut attempt = 6;
1136
1137 assert_eq!(
1138 startup_readiness_poll_sleep(&mut attempt, Duration::from_millis(7)),
1139 Duration::from_millis(7)
1140 );
1141 }
1142}