mj_controller/
web_viewer.rs

1//! Listener recovery lives beside the daemon, independently of session control.
2
3use std::net::SocketAddr;
4use std::sync::{Mutex, PoisonError};
5use std::time::Duration;
6
7use crate::server::{
8    ServerOptions, WebListenerProcess, WebViewerAccess, WebViewerRecovery, run_server_on_listener,
9};
10use anyhow::{Context, Result, bail, ensure};
11use tokio::net::TcpListener;
12use tokio::sync::{mpsc, watch};
13use tokio_util::sync::CancellationToken;
14
15pub struct ViewerControl {
16    access: watch::Sender<WebViewerAccess>,
17    commands: Mutex<Option<mpsc::Sender<WebViewerRecovery>>>,
18}
19
20impl Default for ViewerControl {
21    fn default() -> Self {
22        Self::new()
23    }
24}
25
26impl ViewerControl {
27    pub fn new() -> Self {
28        Self {
29            access: watch::channel(WebViewerAccess::Starting).0,
30            commands: Mutex::new(None),
31        }
32    }
33
34    pub fn access(&self) -> WebViewerAccess {
35        self.access.borrow().clone()
36    }
37
38    pub fn publish(&self, access: WebViewerAccess) {
39        self.access.send_replace(access);
40    }
41
42    pub fn recover(&self, action: WebViewerRecovery) -> Result<()> {
43        // Serialize checking and queueing so two attached clients cannot start two recoveries.
44        let commands = self.commands.lock().unwrap_or_else(PoisonError::into_inner);
45        ensure!(
46            matches!(self.access(), WebViewerAccess::Failed { .. }),
47            "The viewer is already running or starting. Refresh its status before retrying."
48        );
49        let sender = commands
50            .as_ref()
51            .context("Viewer recovery is unavailable")?;
52        let previous = self.access();
53        if matches!(&action, WebViewerRecovery::StopAndRetry(_)) {
54            self.conflict_address()?;
55        }
56        self.publish(WebViewerAccess::Starting);
57        if let Err(error) = sender.try_send(action) {
58            self.publish(previous);
59            return Err(error).context("A viewer recovery is already in progress");
60        }
61        Ok(())
62    }
63
64    pub fn conflict_address(&self) -> Result<SocketAddr> {
65        match self.access() {
66            WebViewerAccess::Failed {
67                address,
68                port_conflict: true,
69                ..
70            } => Ok(address),
71            _ => bail!("The viewer no longer has a port conflict. Refresh its status."),
72        }
73    }
74}
75
76/// Retain the controller and authentication while the listener waits for recovery.
77pub async fn serve(
78    options: ServerOptions,
79    ready: WebViewerAccess,
80    control: &ViewerControl,
81    report: impl Fn(WebViewerAccess),
82) -> Result<()> {
83    let (commands, mut requests) = mpsc::channel(1);
84    *control
85        .commands
86        .lock()
87        .unwrap_or_else(PoisonError::into_inner) = Some(commands);
88    let result = serve_inner(options, ready, &mut requests, report).await;
89    *control
90        .commands
91        .lock()
92        .unwrap_or_else(PoisonError::into_inner) = None;
93    result
94}
95
96async fn serve_inner(
97    options: ServerOptions,
98    ready: WebViewerAccess,
99    requests: &mut mpsc::Receiver<WebViewerRecovery>,
100    report: impl Fn(WebViewerAccess),
101) -> Result<()> {
102    let mut address = options.bind;
103    let mut recovery_failure = None;
104    loop {
105        let failure = if let Some(message) = recovery_failure.take() {
106            WebViewerAccess::Failed {
107                address,
108                message,
109                port_conflict: true,
110            }
111        } else {
112            report(WebViewerAccess::Starting);
113            let listener = tokio::select! {
114                _ = options.shutdown.cancelled() => return Ok(()),
115                result = TcpListener::bind(address) => result,
116            };
117            match listener {
118                Ok(listener) => {
119                    address = listener.local_addr().context("read reserved viewer port")?;
120                    report(ready_at(&ready, address.port())?);
121                    let result = run_server_on_listener(options.clone(), listener).await;
122                    if options.shutdown.is_cancelled() {
123                        return result;
124                    }
125                    let message = match result {
126                        Ok(()) => {
127                            "The web viewer stopped unexpectedly. Try starting it again.".to_owned()
128                        }
129                        Err(error) => {
130                            tracing::warn!(error = format!("{error:#}"), %address, "web viewer stopped");
131                            format!("The web viewer stopped: {error}")
132                        }
133                    };
134                    WebViewerAccess::Failed {
135                        address,
136                        message,
137                        port_conflict: false,
138                    }
139                }
140                Err(error) => {
141                    let port_conflict = error.kind() == std::io::ErrorKind::AddrInUse;
142                    let message = if port_conflict {
143                        format!("Port {} is already in use.", address.port())
144                    } else {
145                        format!("Could not listen on {address}: {error}")
146                    };
147                    WebViewerAccess::Failed {
148                        address,
149                        message,
150                        port_conflict,
151                    }
152                }
153            }
154        };
155        report(failure);
156        let request = tokio::select! {
157            _ = options.shutdown.cancelled() => return Ok(()),
158            request = requests.recv() => request.context("viewer recovery channel closed")?,
159        };
160        report(WebViewerAccess::Starting);
161        match request {
162            WebViewerRecovery::Retry => {}
163            WebViewerRecovery::AnotherPort => address.set_port(0),
164            WebViewerRecovery::StopAndRetry(process) => {
165                let cancellation = options.shutdown.clone();
166                let result = stop_listener(address, process, cancellation).await;
167                if let Err(error) = result {
168                    recovery_failure = Some(format!("Could not stop the server: {error:#}"));
169                }
170            }
171        }
172    }
173}
174
175fn ready_at(ready: &WebViewerAccess, port: u16) -> Result<WebViewerAccess> {
176    let WebViewerAccess::Ready {
177        viewer_url,
178        viewer_code,
179        qr_login_url,
180        fallback_reason,
181    } = ready
182    else {
183        bail!("viewer startup is missing its access details");
184    };
185    fn with_port(value: &str, port: u16) -> Result<String> {
186        let mut url = url::Url::parse(value).context("parse viewer URL")?;
187        url.set_port(Some(port))
188            .map_err(|()| anyhow::anyhow!("viewer URL cannot have a port"))?;
189        Ok(url.into())
190    }
191    Ok(WebViewerAccess::Ready {
192        viewer_url: with_port(viewer_url, port)?,
193        viewer_code: viewer_code.clone(),
194        qr_login_url: qr_login_url
195            .as_deref()
196            .map(|url| with_port(url, port))
197            .transpose()?,
198        fallback_reason: fallback_reason.clone(),
199    })
200}
201
202pub fn inspect_listener(address: SocketAddr) -> Result<Vec<WebListenerProcess>> {
203    let pids = listener_pids(address)?;
204    let mut system = sysinfo::System::new();
205    let own_pid = sysinfo::Pid::from_u32(std::process::id());
206    let mut requested = pids
207        .iter()
208        .map(|pid| sysinfo::Pid::from_u32(*pid))
209        .collect::<Vec<_>>();
210    if !requested.contains(&own_pid) {
211        requested.push(own_pid);
212    }
213    system.refresh_processes_specifics(
214        sysinfo::ProcessesToUpdate::Some(&requested),
215        true,
216        sysinfo::ProcessRefreshKind::new()
217            .with_user(sysinfo::UpdateKind::Always)
218            .with_cmd(sysinfo::UpdateKind::Always)
219            .with_exe(sysinfo::UpdateKind::Always),
220    );
221    let own_user = system
222        .process(own_pid)
223        .and_then(|process| process.user_id());
224    Ok(pids
225        .into_iter()
226        .filter_map(|pid| {
227            let process = system.process(sysinfo::Pid::from_u32(pid))?;
228            let executable = process.exe().map(std::path::Path::to_path_buf).unwrap_or_default();
229            let is_mj = executable.file_stem().is_some_and(|name| name == "mj")
230                && process.cmd().get(1).is_some_and(|arg| arg == "daemon-run");
231            let reason = if pid == std::process::id() {
232                Some("This is the current daemon; stopping it would disconnect this dashboard.")
233            } else if own_user.is_none() || process.user_id() != own_user {
234                Some("This process belongs to another user or its owner cannot be verified.")
235            } else if !is_mj {
236                Some("This is not an identified Mjolnir server. Stop it in its own application.")
237            } else if !cfg!(target_os = "linux") {
238                Some("Safe stopping is unavailable on this platform. Stop this server in its application or use another port.")
239            } else {
240                None
241            };
242            Some(WebListenerProcess {
243                pid,
244                name: process.name().to_string_lossy().into_owned(),
245                executable,
246                started_at: process.start_time(),
247                stop_disabled_reason: reason.map(str::to_owned),
248            })
249        })
250        .collect())
251}
252
253async fn stop_listener(
254    address: SocketAddr,
255    expected: WebListenerProcess,
256    cancel: CancellationToken,
257) -> Result<()> {
258    let pid = expected.pid;
259    let signal_cancel = cancel.clone();
260    let mut task = tokio::task::spawn_blocking(move || {
261        // Acquire a stable process handle before re-inspecting, so PID reuse cannot redirect a signal.
262        #[cfg(target_os = "linux")]
263        let process_handle = open_process_handle(pid)?;
264        let current = inspect_listener(address)?
265            .into_iter()
266            .find(|process| process.pid == pid)
267            .context("That process no longer owns this listener. Inspect the port again.")?;
268        ensure!(
269            current == expected,
270            "The listener's identity changed. Inspect the port again."
271        );
272        ensure!(
273            current.stop_disabled_reason.is_none(),
274            "{}",
275            current.stop_disabled_reason.unwrap_or_default()
276        );
277        ensure!(
278            !signal_cancel.is_cancelled(),
279            "Viewer shutdown cancelled the stop request"
280        );
281        #[cfg(target_os = "linux")]
282        {
283            signal_process(&process_handle)
284        }
285        #[cfg(not(target_os = "linux"))]
286        {
287            bail!(
288                "Safe process termination is unavailable on this platform. Use another port or stop the identified server in its application."
289            )
290        }
291    });
292    tokio::select! {
293        result = &mut task => result.context("listener stop task failed")??,
294        _ = cancel.cancelled() => {
295            // Inspection is bounded; observe its result even when shutdown wins.
296            match task.await {
297                Ok(Ok(())) => {},
298                Ok(Err(error)) => tracing::warn!(%error, "listener stop failed during shutdown"),
299                Err(error) => tracing::warn!(%error, "listener stop task failed during shutdown"),
300            }
301            bail!("Viewer shutdown interrupted recovery");
302        }
303    }
304    let deadline = tokio::time::Instant::now() + Duration::from_secs(10);
305    loop {
306        tokio::select! {
307            _ = cancel.cancelled() => bail!("Viewer shutdown interrupted recovery"),
308            _ = tokio::time::sleep(Duration::from_millis(100)) => {}
309        }
310        match TcpListener::bind(address).await {
311            Ok(listener) => {
312                drop(listener);
313                return Ok(());
314            }
315            Err(error) if error.kind() == std::io::ErrorKind::AddrInUse => {}
316            Err(error) => return Err(error).context("check listener after stopping the server"),
317        }
318        ensure!(
319            tokio::time::Instant::now() < deadline,
320            "Port {} is still occupied after waiting 10 seconds. No force kill was sent. Use another port or inspect again.",
321            address.port()
322        );
323    }
324}
325
326#[cfg(target_os = "linux")]
327fn signal_process(handle: &std::os::fd::OwnedFd) -> Result<()> {
328    use std::os::fd::AsRawFd;
329    // SAFETY: the owned pidfd names the inspected process, never a recycled PID.
330    let result = unsafe {
331        libc::syscall(
332            libc::SYS_pidfd_send_signal,
333            handle.as_raw_fd(),
334            libc::SIGTERM,
335            std::ptr::null::<libc::siginfo_t>(),
336            0,
337        )
338    };
339    ensure!(result == 0, "{}", std::io::Error::last_os_error());
340    Ok(())
341}
342
343#[cfg(target_os = "linux")]
344fn open_process_handle(pid: u32) -> Result<std::os::fd::OwnedFd> {
345    use std::os::fd::FromRawFd;
346    // SAFETY: pidfd_open has no pointer arguments; success yields an owned descriptor.
347    let fd = unsafe { libc::syscall(libc::SYS_pidfd_open, pid, 0) };
348    ensure!(
349        fd >= 0,
350        "Cannot safely open process {pid}: {}",
351        std::io::Error::last_os_error()
352    );
353    // SAFETY: the new descriptor is owned by this call.
354    Ok(unsafe { std::os::fd::OwnedFd::from_raw_fd(fd as i32) })
355}
356
357#[cfg(target_os = "linux")]
358fn listener_pids(address: SocketAddr) -> Result<Vec<u32>> {
359    use std::collections::BTreeSet;
360    use std::fs;
361    use std::path::Path;
362    let mut inodes = BTreeSet::new();
363    for (path, ipv6) in [("/proc/net/tcp", false), ("/proc/net/tcp6", true)] {
364        let contents = match fs::read_to_string(path) {
365            Ok(contents) => contents,
366            Err(error) if error.kind() == std::io::ErrorKind::NotFound && ipv6 => continue,
367            Err(error) => return Err(error).with_context(|| format!("read {path}")),
368        };
369        for line in contents.lines().skip(1) {
370            let fields = line.split_whitespace().collect::<Vec<_>>();
371            ensure!(fields.len() >= 10, "Invalid listener information in {path}");
372            if fields[3] != "0A" {
373                continue;
374            }
375            let candidate = proc_address(fields[1], ipv6)?;
376            if addresses_overlap(address, candidate) {
377                inodes.insert(format!("socket:[{}]", fields[9]));
378            }
379        }
380    }
381    if inodes.is_empty() {
382        return Ok(Vec::new());
383    }
384    let mut pids = BTreeSet::new();
385    for entry in fs::read_dir("/proc").context("inspect running processes")? {
386        let entry = entry?;
387        let Some(pid) = entry
388            .file_name()
389            .to_str()
390            .and_then(|name| name.parse::<u32>().ok())
391        else {
392            continue;
393        };
394        let descriptors = match fs::read_dir(entry.path().join("fd")) {
395            Ok(entries) => entries,
396            Err(error)
397                if matches!(
398                    error.kind(),
399                    std::io::ErrorKind::PermissionDenied | std::io::ErrorKind::NotFound
400                ) =>
401            {
402                continue;
403            }
404            Err(error) => return Err(error).context("inspect listener process descriptors"),
405        };
406        for descriptor in descriptors {
407            let descriptor = descriptor?;
408            let target = match fs::read_link(descriptor.path()) {
409                Ok(target) => target,
410                Err(error)
411                    if matches!(
412                        error.kind(),
413                        std::io::ErrorKind::PermissionDenied | std::io::ErrorKind::NotFound
414                    ) =>
415                {
416                    continue;
417                }
418                Err(error) => return Err(error).context("inspect listener socket ownership"),
419            };
420            if inodes.iter().any(|inode| target == Path::new(inode)) {
421                pids.insert(pid);
422                break;
423            }
424        }
425    }
426    Ok(pids.into_iter().collect())
427}
428
429#[cfg(target_os = "linux")]
430fn proc_address(value: &str, ipv6: bool) -> Result<SocketAddr> {
431    let (host, port) = value.split_once(':').context("invalid listener address")?;
432    let port = u16::from_str_radix(port, 16).context("invalid listener port")?;
433    if ipv6 {
434        ensure!(host.len() == 32 && host.is_ascii(), "invalid IPv6 listener");
435        let mut bytes = [0; 16];
436        for (index, bytes) in bytes.chunks_mut(4).enumerate() {
437            bytes.copy_from_slice(
438                &u32::from_str_radix(&host[index * 8..index * 8 + 8], 16)?.to_ne_bytes(),
439            );
440        }
441        Ok(SocketAddr::new(
442            std::net::Ipv6Addr::from(bytes).into(),
443            port,
444        ))
445    } else {
446        Ok(SocketAddr::new(
447            std::net::Ipv4Addr::from(u32::from_str_radix(host, 16)?.to_ne_bytes()).into(),
448            port,
449        ))
450    }
451}
452
453#[cfg(target_os = "linux")]
454fn addresses_overlap(a: SocketAddr, b: SocketAddr) -> bool {
455    a.port() == b.port()
456        && (a.ip().is_unspecified()
457            || b.ip().is_unspecified()
458            || a.ip() == b.ip()
459            || a.ip().to_canonical() == b.ip().to_canonical())
460}
461
462#[cfg(not(target_os = "linux"))]
463fn listener_pids(address: SocketAddr) -> Result<Vec<u32>> {
464    use crate::targets::{CancellableProcessExecutor, CommandExecutor, CommandSpec};
465    let executor = CancellableProcessExecutor::new(std::sync::Arc::new(
466        std::sync::atomic::AtomicBool::new(false),
467    ))
468    .with_deadline(Duration::from_secs(5));
469    let output = executor
470        .execute(&CommandSpec::new(
471            "lsof",
472            [
473                "-nP".to_owned(),
474                "-a".into(),
475                format!("-iTCP:{}", address.port()),
476                "-sTCP:LISTEN".into(),
477                "-Fp".into(),
478            ],
479        ))
480        .context("Could not inspect this port; lsof must be installed")?;
481    ensure!(
482        output.status == 0 || (output.status == 1 && output.stderr.is_empty()),
483        "Could not inspect this port: {}",
484        String::from_utf8_lossy(&output.stderr)
485    );
486    String::from_utf8(output.stdout)?
487        .lines()
488        .filter_map(|line| line.strip_prefix('p'))
489        .map(|pid| pid.parse().context("invalid listener PID"))
490        .collect()
491}
492
493#[cfg(test)]
494mod tests {
495    use super::*;
496    use crate::server::{ServerRequests, ViewerSnapshot};
497    use std::collections::BTreeMap;
498    use std::sync::Arc;
499    use tokio::io::{AsyncReadExt, AsyncWriteExt};
500
501    fn options(address: SocketAddr) -> ServerOptions {
502        ServerOptions::new(
503            address,
504            watch::channel(ViewerSnapshot::default()).1,
505            watch::channel(BTreeMap::new()).1,
506            ServerRequests {
507                action_tx: mpsc::channel(1).0,
508                bundle_tx: mpsc::channel(1).0,
509                receipt_tx: mpsc::channel(1).0,
510                preflight_tx: mpsc::channel(1).0,
511                move_preparation_tx: mpsc::channel(1).0,
512                client_state_tx: mpsc::channel(1).0,
513                dictation_tx: mpsc::channel(1).0,
514            },
515        )
516        .unwrap()
517    }
518
519    fn ready(address: SocketAddr) -> WebViewerAccess {
520        WebViewerAccess::Ready {
521            viewer_url: format!("http://{address}"),
522            viewer_code: "123456".into(),
523            qr_login_url: None,
524            fallback_reason: None,
525        }
526    }
527
528    async fn wait_access(
529        control: &ViewerControl,
530        predicate: impl Fn(&WebViewerAccess) -> bool,
531    ) -> WebViewerAccess {
532        let mut updates = control.access.subscribe();
533        tokio::time::timeout(Duration::from_secs(10), async {
534            loop {
535                let access = updates.borrow_and_update().clone();
536                if predicate(&access) {
537                    return access;
538                }
539                updates.changed().await.unwrap();
540            }
541        })
542        .await
543        .expect("viewer state did not arrive")
544    }
545
546    async fn http_response(address: SocketAddr) -> String {
547        tokio::time::timeout(Duration::from_secs(3), async {
548            let mut stream = tokio::net::TcpStream::connect(address).await.unwrap();
549            stream
550                .write_all(b"GET / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n")
551                .await
552                .unwrap();
553            let mut response = String::new();
554            stream.read_to_string(&mut response).await.unwrap();
555            response
556        })
557        .await
558        .expect("viewer did not serve HTTP")
559    }
560
561    fn spawn_viewer(
562        options: ServerOptions,
563        control: Arc<ViewerControl>,
564    ) -> tokio::task::JoinHandle<Result<()>> {
565        tokio::spawn(async move {
566            let ready = ready(options.bind);
567            serve(options, ready, &control, |access| control.publish(access)).await
568        })
569    }
570
571    /// Read one API route over a fresh connection, the way a CLI client and
572    /// `probe_api` do. A stalled listener shows up here as a timeout, because
573    /// a new connection has to be accepted before anything can answer.
574    async fn api_response(address: SocketAddr, deadline: Duration) -> String {
575        tokio::time::timeout(deadline, async {
576            let mut stream = tokio::net::TcpStream::connect(address).await.unwrap();
577            stream
578                .write_all(
579                    b"GET /api/v1/sessions HTTP/1.1\r\nHost: localhost\r\n\
580                      Authorization: Bearer test-api-token\r\nConnection: close\r\n\r\n",
581                )
582                .await
583                .unwrap();
584            let mut response = String::new();
585            stream.read_to_string(&mut response).await.unwrap();
586            response
587        })
588        .await
589        .expect("the API did not answer while the control loop was busy")
590    }
591
592    /// The phone server runs its control loop beside this listener. The loop
593    /// takes long turns — it builds snapshots, waits on the daemon's locks,
594    /// and follows every session — and while it does, a cheap read must still
595    /// be answered. `run_server` therefore puts the listener on its own task
596    /// through `ViewerServer` instead of polling it in the loop's `select!`,
597    /// which is what left `GET /api/v1/sessions` unanswered past the client's
598    /// ten-second timeout during a remote provisioning run (issue 1061).
599    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
600    async fn the_api_answers_while_the_control_loop_takes_a_long_turn() {
601        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
602        let address = listener.local_addr().unwrap();
603        let mut options = options(address);
604        options.set_api_token("test-api-token".to_owned());
605        let cancel = options.shutdown.clone();
606        let mut server = crate::server_runtime::ViewerServer::spawn(
607            crate::server::run_server_on_listener(options, listener),
608        );
609
610        let (busy_tx, busy_rx) = tokio::sync::oneshot::channel();
611        // A turn that occupies its task for far longer than the client waits.
612        let control = async move {
613            let _ = busy_tx.send(());
614            std::thread::sleep(Duration::from_secs(3));
615            Ok::<(), anyhow::Error>(())
616        };
617        let request = tokio::spawn(async move {
618            busy_rx.await.unwrap();
619            api_response(address, Duration::from_secs(1)).await
620        });
621
622        // The same composition `run_server` uses: whatever the control loop is
623        // doing, the server keeps its own schedule.
624        tokio::select! {
625            result = server.stopped() => panic!("the viewer stopped early: {result:?}"),
626            result = control => result.unwrap(),
627        }
628        let response = request.await.unwrap();
629        assert!(
630            response.starts_with("HTTP/1.1 200"),
631            "the sessions list must answer during a long control-loop turn: {response}"
632        );
633        assert!(response.contains("\"sessions\""), "{response}");
634        cancel.cancel();
635    }
636
637    #[tokio::test]
638    async fn occupied_port_can_recover_on_another_reserved_port_and_serve_http() {
639        let occupied = TcpListener::bind("127.0.0.1:0").await.unwrap();
640        let address = occupied.local_addr().unwrap();
641        let options = options(address);
642        let cancel = options.shutdown.clone();
643        let control = Arc::new(ViewerControl::new());
644        let task = spawn_viewer(options, control.clone());
645        let failure = wait_access(&control, |access| {
646            matches!(access, WebViewerAccess::Failed { .. })
647        })
648        .await;
649        assert!(
650            matches!(failure, WebViewerAccess::Failed { address: failed, port_conflict: true, .. } if failed == address)
651        );
652        control.recover(WebViewerRecovery::AnotherPort).unwrap();
653        assert!(
654            control.recover(WebViewerRecovery::AnotherPort).is_err(),
655            "concurrent recovery must be rejected"
656        );
657        let access = wait_access(&control, |access| {
658            matches!(access, WebViewerAccess::Ready { .. })
659        })
660        .await;
661        let WebViewerAccess::Ready { viewer_url, .. } = access else {
662            unreachable!()
663        };
664        let url = url::Url::parse(&viewer_url).unwrap();
665        let actual = SocketAddr::new(address.ip(), url.port().unwrap());
666        assert_ne!(actual.port(), address.port());
667        assert_ne!(actual.port(), 0);
668        assert!(http_response(actual).await.starts_with("HTTP/1.1 200"));
669        assert!(
670            TcpListener::bind(address).await.is_err(),
671            "recovery must leave the existing listener alone"
672        );
673        assert!(
674            control.recover(WebViewerRecovery::Retry).is_err(),
675            "a healthy viewer must not be restarted"
676        );
677        cancel.cancel();
678        tokio::time::timeout(Duration::from_secs(3), task)
679            .await
680            .unwrap()
681            .unwrap()
682            .unwrap();
683    }
684
685    #[tokio::test]
686    async fn retry_uses_the_original_port_after_its_owner_releases_it() {
687        let occupied = TcpListener::bind("127.0.0.1:0").await.unwrap();
688        let address = occupied.local_addr().unwrap();
689        let options = options(address);
690        let cancel = options.shutdown.clone();
691        let control = Arc::new(ViewerControl::new());
692        let task = spawn_viewer(options, control.clone());
693        wait_access(&control, |access| {
694            matches!(access, WebViewerAccess::Failed { .. })
695        })
696        .await;
697        drop(occupied);
698        // Concurrent process-spawning tests can inherit the listening socket
699        // between fork and exec. Dropping our descriptor alone does not prove
700        // the OS has released it. Wait for that precondition before Retry.
701        // Bind without listening so the probe cannot become another inherited
702        // listener; SO_REUSEADDR matches the server's bind behavior.
703        tokio::time::timeout(Duration::from_secs(3), async {
704            loop {
705                let probe = tokio::net::TcpSocket::new_v4().unwrap();
706                probe.set_reuseaddr(true).unwrap();
707                match probe.bind(address) {
708                    Ok(()) => break,
709                    Err(error) if error.kind() == std::io::ErrorKind::AddrInUse => {
710                        tokio::time::sleep(Duration::from_millis(10)).await;
711                    }
712                    Err(error) => panic!("cannot check listener release: {error}"),
713                }
714            }
715        })
716        .await
717        .expect("original listener was not released");
718        control.recover(WebViewerRecovery::Retry).unwrap();
719        let access = wait_access(&control, |access| {
720            matches!(
721                access,
722                WebViewerAccess::Ready { .. } | WebViewerAccess::Failed { .. }
723            )
724        })
725        .await;
726        assert!(
727            matches!(&access, WebViewerAccess::Ready { viewer_url, .. } if viewer_url == &format!("http://{address}/")),
728            "retry did not restore the original listener: {access:?}"
729        );
730        assert!(http_response(address).await.starts_with("HTTP/1.1 200"));
731        cancel.cancel();
732        tokio::time::timeout(Duration::from_secs(3), task)
733            .await
734            .unwrap()
735            .unwrap()
736            .unwrap();
737    }
738
739    #[tokio::test]
740    async fn shutdown_does_not_wait_for_a_port_conflict_to_be_resolved() {
741        let occupied = TcpListener::bind("127.0.0.1:0").await.unwrap();
742        let options = options(occupied.local_addr().unwrap());
743        let cancel = options.shutdown.clone();
744        let control = Arc::new(ViewerControl::new());
745        let task = spawn_viewer(options, control.clone());
746        wait_access(&control, |access| {
747            matches!(access, WebViewerAccess::Failed { .. })
748        })
749        .await;
750        cancel.cancel();
751        tokio::time::timeout(Duration::from_millis(500), task)
752            .await
753            .unwrap()
754            .unwrap()
755            .unwrap();
756        assert!(control.recover(WebViewerRecovery::Retry).is_err());
757    }
758
759    #[test]
760    fn changing_ports_preserves_https_hostname_and_login_credentials() {
761        let ready = WebViewerAccess::Ready {
762            viewer_url: "https://host.tailnet.ts.net:37650/".into(),
763            viewer_code: "123456".into(),
764            qr_login_url: Some("https://host.tailnet.ts.net:37650/auth/login?token=secret".into()),
765            fallback_reason: None,
766        };
767        let changed = ready_at(&ready, 49152).unwrap();
768        assert!(
769            matches!(changed, WebViewerAccess::Ready { viewer_url, viewer_code, qr_login_url: Some(login), .. }
770            if viewer_url == "https://host.tailnet.ts.net:49152/" && viewer_code == "123456" && login == "https://host.tailnet.ts.net:49152/auth/login?token=secret")
771        );
772        let ipv6 = ready_at(&self::ready("[::1]:0".parse().unwrap()), 49152).unwrap();
773        assert!(
774            matches!(ipv6, WebViewerAccess::Ready { viewer_url, .. } if viewer_url == "http://[::1]:49152/")
775        );
776    }
777
778    #[cfg(target_os = "linux")]
779    #[tokio::test]
780    async fn inspection_finds_real_listener_and_refuses_to_stop_the_current_daemon() {
781        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
782        let address = listener.local_addr().unwrap();
783        let pids = listener_pids(address).unwrap();
784        assert!(
785            pids.contains(&std::process::id()),
786            "socket {address} owners: {pids:?}; expected {}",
787            std::process::id()
788        );
789        let processes = tokio::task::spawn_blocking(move || inspect_listener(address))
790            .await
791            .unwrap()
792            .unwrap();
793        let own = processes
794            .into_iter()
795            .find(|process| process.pid == std::process::id())
796            .expect("listener owner must be found");
797        assert!(
798            own.stop_disabled_reason
799                .as_ref()
800                .unwrap()
801                .contains("current daemon")
802        );
803        let error = stop_listener(address, own, CancellationToken::new())
804            .await
805            .unwrap_err();
806        assert!(error.to_string().contains("current daemon"));
807        assert!(TcpListener::bind(address).await.is_err());
808    }
809
810    #[cfg(target_os = "linux")]
811    #[tokio::test]
812    async fn stopping_rejects_a_stale_inspected_identity() {
813        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
814        let address = listener.local_addr().unwrap();
815        let mut own = tokio::task::spawn_blocking(move || inspect_listener(address))
816            .await
817            .unwrap()
818            .unwrap()
819            .into_iter()
820            .find(|process| process.pid == std::process::id())
821            .unwrap();
822        own.started_at = own.started_at.saturating_sub(1);
823        let error = stop_listener(address, own, CancellationToken::new())
824            .await
825            .unwrap_err();
826        assert!(error.to_string().contains("identity changed"));
827    }
828}
mj_controller/web_viewer.rs

mj_controller/
web_viewer.rs
