ashlar_native/
single_instance.rs

1//! One running instance per user: a second launch forwards its arguments (the files it was asked
2//! to open) to the first over a local socket, then exits.
3//!
4//! ```no_run
5//! # use ashlar_native::single_instance::{Launch, SingleInstance};
6//! let args: Vec<String> = std::env::args().skip(1).collect();
7//! match SingleInstance::new("com.example.editor").claim(&args, |msg| {
8//!     // On a listener thread: queue `msg.args` (resolved against `msg.cwd`) and wake the UI.
9//! }) {
10//!     Ok(Launch::Forwarded) => std::process::exit(0),
11//!     Ok(Launch::Primary(guard)) => { /* keep `guard` alive while the app runs */ }
12//!     Err(e) => { /* no single-instance support here: just run */ }
13//! }
14//! ```
15//!
16//! Transport: a Unix domain socket (macOS, Linux) in a per-user directory, readable only by the
17//! user; on Windows a loopback TCP port written to a per-user file with a random token the
18//! second launch must present. Messages are one JSON line, at most [`MAX_MESSAGE`] bytes.
19//! The handshake has timeouts, so a hung first instance never hangs the second.
20
21// The transport helpers are unused where there is no transport (the web).
22#![cfg_attr(not(any(unix, windows)), allow(dead_code))]
23
24use std::io::{BufRead, BufReader, Read, Write};
25use std::path::{Path, PathBuf};
26use std::sync::Arc;
27use std::sync::atomic::{AtomicBool, Ordering};
28use std::time::Duration;
29
30use crate::{NativeError, Result};
31
32/// Largest forwarded message.
33pub const MAX_MESSAGE: usize = 1024 * 1024;
34const TIMEOUT: Duration = Duration::from_secs(3);
35const ACK: &str = "ok";
36
37/// What a second launch sent.
38#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
39pub struct Forwarded {
40    /// The second launch's working directory (resolve relative arguments against it).
41    pub cwd: PathBuf,
42    /// Its arguments, without the program name.
43    pub args: Vec<String>,
44}
45
46#[derive(serde::Serialize, serde::Deserialize)]
47struct Wire {
48    v: u32,
49    #[serde(default)]
50    token: String,
51    cwd: PathBuf,
52    args: Vec<String>,
53}
54
55/// The outcome of [`SingleInstance::claim`].
56pub enum Launch {
57    /// This is the first instance; it listens while the guard lives.
58    Primary(PrimaryGuard),
59    /// Another instance is running and has the arguments: exit now.
60    Forwarded,
61}
62
63impl std::fmt::Debug for Launch {
64    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
65        f.write_str(match self {
66            Launch::Primary(_) => "Primary",
67            Launch::Forwarded => "Forwarded",
68        })
69    }
70}
71
72/// Stops listening (and removes the socket) when dropped.
73pub struct PrimaryGuard {
74    stop: Arc<AtomicBool>,
75    endpoint: PathBuf,
76    #[allow(dead_code)]
77    wake: Box<dyn Fn() + Send + Sync>,
78}
79
80impl Drop for PrimaryGuard {
81    fn drop(&mut self) {
82        self.stop.store(true, Ordering::SeqCst);
83        (self.wake)();
84        let _ = std::fs::remove_file(&self.endpoint);
85    }
86}
87
88/// Which app the instance belongs to, and where its endpoint lives.
89#[derive(Debug, Clone)]
90pub struct SingleInstance {
91    app_id: String,
92    dir: Option<PathBuf>,
93}
94
95/// An app id is a reverse-DNS-like name: ASCII letters, digits, `.`, `-`, `_`; at most 64.
96pub fn valid_app_id(id: &str) -> bool {
97    !id.is_empty() && id.len() <= 64 && id.bytes().all(|b| b.is_ascii_alphanumeric() || b"._-".contains(&b)) && !id.starts_with('.')
98}
99
100impl SingleInstance {
101    pub fn new(app_id: impl Into<String>) -> Self {
102        Self { app_id: app_id.into(), dir: None }
103    }
104
105    /// Keep the endpoint in `dir` instead of the per-user default (tests, sandboxes).
106    pub fn in_dir(mut self, dir: impl Into<PathBuf>) -> Self {
107        self.dir = Some(dir.into());
108        self
109    }
110
111    fn default_dir() -> Option<PathBuf> {
112        #[cfg(target_os = "linux")]
113        if let Some(d) = std::env::var_os("XDG_RUNTIME_DIR").filter(|d| !d.is_empty()) {
114            return Some(PathBuf::from(d));
115        }
116        // macOS: `$TMPDIR` is per user (`/var/folders/…`); Windows: `%TEMP%` is per user.
117        Some(std::env::temp_dir())
118    }
119
120    fn endpoint(&self) -> Result<PathBuf> {
121        if !valid_app_id(&self.app_id) {
122            return Err(NativeError::Invalid(format!("`{}` is not an app id", self.app_id.chars().take(64).collect::<String>())));
123        }
124        let dir = self.dir.clone().or_else(Self::default_dir).ok_or(NativeError::Unsupported)?;
125        #[cfg(unix)]
126        let name = {
127            // `/tmp` may be shared: keep users apart by name too.
128            let user = std::env::var("USER").unwrap_or_default().chars().filter(|c| c.is_ascii_alphanumeric()).take(32).collect::<String>();
129            format!("{}.{}.sock", self.app_id, user)
130        };
131        #[cfg(not(unix))]
132        let name = format!("{}.instance", self.app_id);
133        Ok(dir.join(name))
134    }
135
136    /// Become the first instance, or forward `args` to it. `on_forward` runs on a listener thread
137    /// for each later launch.
138    pub fn claim(&self, args: &[String], on_forward: impl Fn(Forwarded) + Send + 'static) -> Result<Launch> {
139        let endpoint = self.endpoint()?;
140        let cwd = std::env::current_dir().unwrap_or_default();
141        let msg = |token: String| Wire { v: 1, token, cwd: cwd.clone(), args: args.iter().take(4096).cloned().collect() };
142        imp::claim(&endpoint, msg, Box::new(on_forward))
143    }
144}
145
146/// Why forwarding failed.
147enum SendError {
148    /// The other end closed the connection without answering: no instance runs there (its socket
149    /// outlived it, for example held by a child process that inherited it).
150    Closed,
151    Failed(NativeError),
152}
153
154impl From<SendError> for NativeError {
155    fn from(e: SendError) -> Self {
156        match e {
157            SendError::Closed => NativeError::Os("the running instance closed the connection without accepting the files".into()),
158            SendError::Failed(e) => e,
159        }
160    }
161}
162
163/// Send one message on `stream` and wait for the ack.
164fn send<S: Read + Write>(stream: &mut S, wire: &Wire) -> std::result::Result<(), SendError> {
165    let mut line = serde_json::to_vec(wire).map_err(|e| SendError::Failed(NativeError::Os(crate::os_text(e))))?;
166    if line.len() > MAX_MESSAGE {
167        return Err(SendError::Failed(NativeError::Invalid("too many arguments to forward".into())));
168    }
169    line.push(b'\n');
170    let closed = |e: std::io::Error| match e.kind() {
171        std::io::ErrorKind::BrokenPipe
172        | std::io::ErrorKind::ConnectionReset
173        | std::io::ErrorKind::ConnectionAborted
174        | std::io::ErrorKind::NotConnected
175        | std::io::ErrorKind::UnexpectedEof => SendError::Closed,
176        _ => SendError::Failed(io(e)),
177    };
178    stream.write_all(&line).map_err(closed)?;
179    stream.flush().map_err(closed)?;
180    let mut reply = String::new();
181    BufReader::new(stream.take(16)).read_line(&mut reply).map_err(closed)?;
182    match reply.trim() {
183        r if r == ACK => Ok(()),
184        "" => Err(SendError::Closed),
185        _ => Err(SendError::Failed(NativeError::Os("the running instance did not accept the files".into()))),
186    }
187}
188
189/// Read one message from `stream`, check it, ack it.
190fn receive<S: Read + Write>(stream: &mut S, token: &str) -> Option<Forwarded> {
191    let mut line = Vec::new();
192    let mut reader = BufReader::new(Read::by_ref(stream).take(MAX_MESSAGE as u64 + 1));
193    reader.read_until(b'\n', &mut line).ok()?;
194    if line.len() > MAX_MESSAGE || line.last() != Some(&b'\n') {
195        return None;
196    }
197    let wire: Wire = serde_json::from_slice(&line).ok()?;
198    if wire.v != 1 || wire.token != token || wire.args.len() > 4096 {
199        return None;
200    }
201    let _ = stream.write_all(format!("{ACK}\n").as_bytes());
202    Some(Forwarded { cwd: wire.cwd, args: wire.args })
203}
204
205fn io(e: std::io::Error) -> NativeError {
206    NativeError::Os(crate::os_text(e))
207}
208
209type OnForward = Box<dyn Fn(Forwarded) + Send>;
210
211/// Accept connections until `stop`, handing each message to `on_forward`.
212fn serve<L, S>(listener: L, accept: impl Fn(&L) -> std::io::Result<S> + Send + 'static, token: String, stop: Arc<AtomicBool>, on_forward: OnForward)
213where
214    L: Send + 'static,
215    S: Read + Write + Send,
216{
217    let spawned = std::thread::Builder::new().name("single-instance".into()).spawn(move || {
218        while !stop.load(Ordering::SeqCst) {
219            let Ok(mut stream) = accept(&listener) else {
220                // Out of descriptors or similar: don't spin.
221                std::thread::sleep(Duration::from_millis(50));
222                continue;
223            };
224            if stop.load(Ordering::SeqCst) {
225                break;
226            }
227            let r = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
228                if let Some(msg) = receive(&mut stream, &token) {
229                    on_forward(msg);
230                }
231            }));
232            if r.is_err() {
233                log::warn!("single-instance: a forwarded launch's handler panicked");
234            }
235        }
236    });
237    if let Err(e) = spawned {
238        log::warn!("single-instance: no listener thread ({e})");
239    }
240}
241
242#[cfg(unix)]
243mod imp {
244    use std::os::unix::fs::PermissionsExt;
245    use std::os::unix::net::{UnixListener, UnixStream};
246
247    use super::*;
248
249    /// macOS limits socket paths to 104 bytes, Linux to 108.
250    const MAX_SOCKET_PATH: usize = 100;
251
252    fn connect(endpoint: &Path) -> std::io::Result<UnixStream> {
253        let s = UnixStream::connect(endpoint)?;
254        s.set_read_timeout(Some(TIMEOUT))?;
255        s.set_write_timeout(Some(TIMEOUT))?;
256        Ok(s)
257    }
258
259    pub fn claim(endpoint: &Path, msg: impl Fn(String) -> Wire, on_forward: OnForward) -> Result<Launch> {
260        if endpoint.as_os_str().len() > MAX_SOCKET_PATH {
261            return Err(NativeError::Invalid("the socket path is too long".into()));
262        }
263        for _ in 0..2 {
264            match UnixListener::bind(endpoint) {
265                Ok(listener) => {
266                    let _ = std::fs::set_permissions(endpoint, std::fs::Permissions::from_mode(0o600));
267                    let stop = Arc::new(AtomicBool::new(false));
268                    let wake_path = endpoint.to_path_buf();
269                    serve(
270                        listener,
271                        |l: &UnixListener| {
272                            let (s, _) = l.accept()?;
273                            s.set_read_timeout(Some(TIMEOUT))?;
274                            s.set_write_timeout(Some(TIMEOUT))?;
275                            Ok(s)
276                        },
277                        String::new(),
278                        stop.clone(),
279                        on_forward,
280                    );
281                    let wake = Box::new(move || {
282                        let _ = UnixStream::connect(&wake_path);
283                    });
284                    return Ok(Launch::Primary(PrimaryGuard { stop, endpoint: endpoint.to_path_buf(), wake }));
285                }
286                Err(e) if e.kind() == std::io::ErrorKind::AddrInUse => {
287                    match connect(endpoint).map_err(|_| SendError::Closed).and_then(|mut s| send(&mut s, &msg(String::new()))) {
288                        Ok(()) => return Ok(Launch::Forwarded),
289                        // Nobody listens, or something holds the socket but closes without answering
290                        // (a crashed instance's socket, inherited by a child it spawned): no instance
291                        // runs there. Remove it and retry.
292                        Err(SendError::Closed) => {
293                            let _ = std::fs::remove_file(endpoint);
294                        }
295                        Err(SendError::Failed(e)) => return Err(e),
296                    }
297                }
298                Err(e) => return Err(io(e)),
299            }
300        }
301        Err(NativeError::Os("could not claim the single-instance socket".into()))
302    }
303}
304
305/// `port token [pid]`, the Windows endpoint file (the pid is absent in files from older builds).
306#[cfg_attr(not(windows), allow(dead_code))]
307fn parse_endpoint(text: &str) -> Option<(u16, String, Option<u32>)> {
308    let mut parts = text.split_whitespace();
309    let port = parts.next()?.parse().ok()?;
310    let token = parts.next()?.to_string();
311    let pid = parts.next().and_then(|p| p.parse().ok());
312    Some((port, token, pid))
313}
314
315#[cfg(windows)]
316mod imp {
317    use std::net::{Ipv4Addr, SocketAddr, TcpListener, TcpStream};
318
319    use super::*;
320
321    /// A random token: the time, the process id and the address of a fresh allocation, hashed.
322    fn token() -> String {
323        use std::hash::{BuildHasher, Hasher};
324        let mut h = std::collections::hash_map::RandomState::new().build_hasher();
325        h.write_u128(std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map(|d| d.as_nanos()).unwrap_or(0));
326        h.write_u32(std::process::id());
327        let a = std::collections::hash_map::RandomState::new().build_hasher().finish();
328        format!("{:016x}{:016x}", h.finish(), a)
329    }
330
331    fn read_endpoint(endpoint: &Path) -> Option<(u16, String, Option<u32>)> {
332        parse_endpoint(&std::fs::read_to_string(endpoint).ok()?)
333    }
334
335    /// Let the first instance take the foreground: Windows refuses `SetForegroundWindow` to a
336    /// process in the background unless the foreground process (this launch) allows it.
337    #[allow(unsafe_code)]
338    fn allow_foreground(pid: u32) {
339        // SAFETY: a plain Win32 call taking a process id by value; failure only means the first
340        // instance may flash in the taskbar instead of coming to the front.
341        let _ = unsafe { windows::Win32::UI::WindowsAndMessaging::AllowSetForegroundWindow(pid) };
342    }
343
344    pub fn claim(endpoint: &Path, msg: impl Fn(String) -> Wire, on_forward: OnForward) -> Result<Launch> {
345        if let Some((port, token, pid)) = read_endpoint(endpoint) {
346            let addr = SocketAddr::from((Ipv4Addr::LOCALHOST, port));
347            if let Ok(mut s) = TcpStream::connect_timeout(&addr, TIMEOUT) {
348                if let Some(pid) = pid {
349                    allow_foreground(pid);
350                }
351                let _ = s.set_read_timeout(Some(TIMEOUT));
352                let _ = s.set_write_timeout(Some(TIMEOUT));
353                if send(&mut s, &msg(token)).is_ok() {
354                    return Ok(Launch::Forwarded);
355                }
356            }
357        }
358        let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).map_err(io)?;
359        let port = listener.local_addr().map_err(io)?.port();
360        let token = token();
361        let tmp = endpoint.with_extension("tmp");
362        std::fs::write(&tmp, format!("{port} {token} {}", std::process::id())).map_err(io)?;
363        std::fs::rename(&tmp, endpoint).map_err(io)?;
364        let stop = Arc::new(AtomicBool::new(false));
365        serve(
366            listener,
367            |l: &TcpListener| {
368                let (s, _) = l.accept()?;
369                s.set_read_timeout(Some(TIMEOUT))?;
370                s.set_write_timeout(Some(TIMEOUT))?;
371                Ok(s)
372            },
373            token,
374            stop.clone(),
375            on_forward,
376        );
377        let wake = Box::new(move || {
378            let _ = TcpStream::connect_timeout(&SocketAddr::from((Ipv4Addr::LOCALHOST, port)), Duration::from_millis(200));
379        });
380        Ok(Launch::Primary(PrimaryGuard { stop, endpoint: endpoint.to_path_buf(), wake }))
381    }
382}
383
384#[cfg(not(any(unix, windows)))]
385mod imp {
386    use super::*;
387
388    pub fn claim(_endpoint: &Path, _msg: impl Fn(String) -> Wire, _on_forward: OnForward) -> Result<Launch> {
389        Err(NativeError::Unsupported)
390    }
391}
392
393#[cfg(test)]
394mod tests {
395    use std::sync::mpsc::channel;
396
397    #[test]
398    fn endpoint_files_carry_an_optional_pid() {
399        assert_eq!(super::parse_endpoint("4242 abc 77\n"), Some((4242, "abc".into(), Some(77))));
400        assert_eq!(super::parse_endpoint("4242 abc"), Some((4242, "abc".into(), None)));
401        assert_eq!(super::parse_endpoint("x abc"), None);
402        assert_eq!(super::parse_endpoint("4242"), None);
403    }
404
405    use super::*;
406
407    #[test]
408    fn app_ids_are_checked() {
409        assert!(valid_app_id("com.example.editor"));
410        for bad in ["", ".hidden", "a/b", "a b", "../x", &"x".repeat(65)] {
411            assert!(!valid_app_id(bad), "{bad:?}");
412        }
413        assert!(matches!(SingleInstance::new("a/b").claim(&[], |_| {}), Err(NativeError::Invalid(_))));
414    }
415
416    /// Socket paths are capped near 100 bytes: short folder and app names keep the tests passing
417    /// under a long `$TMPDIR` too.
418    fn short_dir() -> tempfile::TempDir {
419        tempfile::Builder::new().prefix("").rand_bytes(4).tempdir().unwrap()
420    }
421
422    #[cfg(any(unix, windows))]
423    #[test]
424    fn a_second_launch_forwards_its_arguments_to_the_first() {
425        let dir = short_dir();
426        let instance = SingleInstance::new("t.fwd").in_dir(dir.path());
427        let (tx, rx) = channel();
428        let first = instance.claim(&["first.txt".into()], move |m| tx.send(m).unwrap()).unwrap();
429        assert!(matches!(first, Launch::Primary(_)));
430        let second = instance.claim(&["a.txt".into(), "/abs/b.txt".into()], |_| panic!("not the primary")).unwrap();
431        assert!(matches!(second, Launch::Forwarded));
432        let got = rx.recv_timeout(Duration::from_secs(5)).unwrap();
433        assert_eq!(got.args, vec!["a.txt".to_string(), "/abs/b.txt".into()]);
434        assert_eq!(got.cwd, std::env::current_dir().unwrap());
435        // Garbage on the socket is ignored and the listener keeps going.
436        #[cfg(unix)]
437        {
438            let ep = instance.endpoint().unwrap();
439            let mut s = std::os::unix::net::UnixStream::connect(&ep).unwrap();
440            s.write_all(b"{not json}\n").unwrap();
441            drop(s);
442        }
443        assert!(matches!(instance.claim(&["c.txt".into()], |_| {}).unwrap(), Launch::Forwarded));
444        assert_eq!(rx.recv_timeout(Duration::from_secs(5)).unwrap().args, vec!["c.txt".to_string()]);
445        // When the first quits, the next launch becomes the first.
446        drop(first);
447        let third = instance.claim(&[], |_| {}).unwrap();
448        assert!(matches!(third, Launch::Primary(_)));
449    }
450
451    /// A crashed instance's socket can outlive it in a child process that inherited it (a helper it
452    /// spawned, or any process started while the socket was being made): connecting works, but the
453    /// connection closes without an answer. That is no running instance either: take it over.
454    #[cfg(unix)]
455    #[test]
456    fn a_socket_that_closes_without_answering_is_taken_over() {
457        let dir = short_dir();
458        let instance = SingleInstance::new("t.held").in_dir(dir.path());
459        let ep = instance.endpoint().unwrap();
460        let holder = std::os::unix::net::UnixListener::bind(&ep).unwrap();
461        let closer = std::thread::spawn(move || {
462            // accept one connection and close it unanswered, like a process that holds the socket
463            // but runs no listener
464            if let Ok((s, _)) = holder.accept() {
465                drop(s);
466            }
467            holder
468        });
469        assert!(matches!(instance.claim(&[], |_| {}).unwrap(), Launch::Primary(_)));
470        drop(closer.join());
471    }
472
473    #[cfg(unix)]
474    #[test]
475    fn a_stale_socket_from_a_crash_is_taken_over() {
476        let dir = short_dir();
477        let instance = SingleInstance::new("t.stale").in_dir(dir.path());
478        let ep = instance.endpoint().unwrap();
479        // A socket file nobody listens on.
480        drop(std::os::unix::net::UnixListener::bind(&ep).unwrap());
481        assert!(ep.exists());
482        assert!(matches!(instance.claim(&[], |_| {}).unwrap(), Launch::Primary(_)));
483    }
484
485    #[test]
486    fn oversized_or_wrong_token_messages_are_refused() {
487        struct Duplex(std::io::Cursor<Vec<u8>>, Vec<u8>);
488        impl Read for Duplex {
489            fn read(&mut self, b: &mut [u8]) -> std::io::Result<usize> {
490                self.0.read(b)
491            }
492        }
493        impl Write for Duplex {
494            fn write(&mut self, b: &[u8]) -> std::io::Result<usize> {
495                self.1.write(b)
496            }
497            fn flush(&mut self) -> std::io::Result<()> {
498                Ok(())
499            }
500        }
501        let line = |token: &str| format!("{{\"v\":1,\"token\":\"{token}\",\"cwd\":\"/\",\"args\":[\"x\"]}}\n").into_bytes();
502        let mut ok = Duplex(std::io::Cursor::new(line("t")), vec![]);
503        assert!(receive(&mut ok, "t").is_some());
504        assert_eq!(ok.1, b"ok\n");
505        let mut wrong = Duplex(std::io::Cursor::new(line("nope")), vec![]);
506        assert!(receive(&mut wrong, "t").is_none());
507        assert!(wrong.1.is_empty());
508        let mut huge = Duplex(std::io::Cursor::new(vec![b'x'; MAX_MESSAGE + 10]), vec![]);
509        assert!(receive(&mut huge, "").is_none());
510        let mut unterminated = Duplex(std::io::Cursor::new(b"{\"v\":1".to_vec()), vec![]);
511        assert!(receive(&mut unterminated, "").is_none());
512    }
513}
ashlar_native/single_instance.rs

ashlar_native/
single_instance.rs
