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#[cfg(windows)]
306mod imp {
307    use std::net::{Ipv4Addr, SocketAddr, TcpListener, TcpStream};
308
309    use super::*;
310
311    /// A random token: the time, the process id and the address of a fresh allocation, hashed.
312    fn token() -> String {
313        use std::hash::{BuildHasher, Hasher};
314        let mut h = std::collections::hash_map::RandomState::new().build_hasher();
315        h.write_u128(std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map(|d| d.as_nanos()).unwrap_or(0));
316        h.write_u32(std::process::id());
317        let a = std::collections::hash_map::RandomState::new().build_hasher().finish();
318        format!("{:016x}{:016x}", h.finish(), a)
319    }
320
321    fn read_endpoint(endpoint: &Path) -> Option<(u16, String)> {
322        let text = std::fs::read_to_string(endpoint).ok()?;
323        let (port, token) = text.trim().split_once(' ')?;
324        Some((port.parse().ok()?, token.to_string()))
325    }
326
327    pub fn claim(endpoint: &Path, msg: impl Fn(String) -> Wire, on_forward: OnForward) -> Result<Launch> {
328        if let Some((port, token)) = read_endpoint(endpoint) {
329            let addr = SocketAddr::from((Ipv4Addr::LOCALHOST, port));
330            if let Ok(mut s) = TcpStream::connect_timeout(&addr, TIMEOUT) {
331                let _ = s.set_read_timeout(Some(TIMEOUT));
332                let _ = s.set_write_timeout(Some(TIMEOUT));
333                if send(&mut s, &msg(token)).is_ok() {
334                    return Ok(Launch::Forwarded);
335                }
336            }
337        }
338        let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).map_err(io)?;
339        let port = listener.local_addr().map_err(io)?.port();
340        let token = token();
341        let tmp = endpoint.with_extension("tmp");
342        std::fs::write(&tmp, format!("{port} {token}")).map_err(io)?;
343        std::fs::rename(&tmp, endpoint).map_err(io)?;
344        let stop = Arc::new(AtomicBool::new(false));
345        serve(
346            listener,
347            |l: &TcpListener| {
348                let (s, _) = l.accept()?;
349                s.set_read_timeout(Some(TIMEOUT))?;
350                s.set_write_timeout(Some(TIMEOUT))?;
351                Ok(s)
352            },
353            token,
354            stop.clone(),
355            on_forward,
356        );
357        let wake = Box::new(move || {
358            let _ = TcpStream::connect_timeout(&SocketAddr::from((Ipv4Addr::LOCALHOST, port)), Duration::from_millis(200));
359        });
360        Ok(Launch::Primary(PrimaryGuard { stop, endpoint: endpoint.to_path_buf(), wake }))
361    }
362}
363
364#[cfg(not(any(unix, windows)))]
365mod imp {
366    use super::*;
367
368    pub fn claim(_endpoint: &Path, _msg: impl Fn(String) -> Wire, _on_forward: OnForward) -> Result<Launch> {
369        Err(NativeError::Unsupported)
370    }
371}
372
373#[cfg(test)]
374mod tests {
375    use std::sync::mpsc::channel;
376
377    use super::*;
378
379    #[test]
380    fn app_ids_are_checked() {
381        assert!(valid_app_id("com.example.editor"));
382        for bad in ["", ".hidden", "a/b", "a b", "../x", &"x".repeat(65)] {
383            assert!(!valid_app_id(bad), "{bad:?}");
384        }
385        assert!(matches!(SingleInstance::new("a/b").claim(&[], |_| {}), Err(NativeError::Invalid(_))));
386    }
387
388    /// Socket paths are capped near 100 bytes: short folder and app names keep the tests passing
389    /// under a long `$TMPDIR` too.
390    fn short_dir() -> tempfile::TempDir {
391        tempfile::Builder::new().prefix("").rand_bytes(4).tempdir().unwrap()
392    }
393
394    #[cfg(any(unix, windows))]
395    #[test]
396    fn a_second_launch_forwards_its_arguments_to_the_first() {
397        let dir = short_dir();
398        let instance = SingleInstance::new("t.fwd").in_dir(dir.path());
399        let (tx, rx) = channel();
400        let first = instance.claim(&["first.txt".into()], move |m| tx.send(m).unwrap()).unwrap();
401        assert!(matches!(first, Launch::Primary(_)));
402        let second = instance.claim(&["a.txt".into(), "/abs/b.txt".into()], |_| panic!("not the primary")).unwrap();
403        assert!(matches!(second, Launch::Forwarded));
404        let got = rx.recv_timeout(Duration::from_secs(5)).unwrap();
405        assert_eq!(got.args, vec!["a.txt".to_string(), "/abs/b.txt".into()]);
406        assert_eq!(got.cwd, std::env::current_dir().unwrap());
407        // Garbage on the socket is ignored and the listener keeps going.
408        #[cfg(unix)]
409        {
410            let ep = instance.endpoint().unwrap();
411            let mut s = std::os::unix::net::UnixStream::connect(&ep).unwrap();
412            s.write_all(b"{not json}\n").unwrap();
413            drop(s);
414        }
415        assert!(matches!(instance.claim(&["c.txt".into()], |_| {}).unwrap(), Launch::Forwarded));
416        assert_eq!(rx.recv_timeout(Duration::from_secs(5)).unwrap().args, vec!["c.txt".to_string()]);
417        // When the first quits, the next launch becomes the first.
418        drop(first);
419        let third = instance.claim(&[], |_| {}).unwrap();
420        assert!(matches!(third, Launch::Primary(_)));
421    }
422
423    /// A crashed instance's socket can outlive it in a child process that inherited it (a helper it
424    /// spawned, or any process started while the socket was being made): connecting works, but the
425    /// connection closes without an answer. That is no running instance either: take it over.
426    #[cfg(unix)]
427    #[test]
428    fn a_socket_that_closes_without_answering_is_taken_over() {
429        let dir = short_dir();
430        let instance = SingleInstance::new("t.held").in_dir(dir.path());
431        let ep = instance.endpoint().unwrap();
432        let holder = std::os::unix::net::UnixListener::bind(&ep).unwrap();
433        let closer = std::thread::spawn(move || {
434            // accept one connection and close it unanswered, like a process that holds the socket
435            // but runs no listener
436            if let Ok((s, _)) = holder.accept() {
437                drop(s);
438            }
439            holder
440        });
441        assert!(matches!(instance.claim(&[], |_| {}).unwrap(), Launch::Primary(_)));
442        drop(closer.join());
443    }
444
445    #[cfg(unix)]
446    #[test]
447    fn a_stale_socket_from_a_crash_is_taken_over() {
448        let dir = short_dir();
449        let instance = SingleInstance::new("t.stale").in_dir(dir.path());
450        let ep = instance.endpoint().unwrap();
451        // A socket file nobody listens on.
452        drop(std::os::unix::net::UnixListener::bind(&ep).unwrap());
453        assert!(ep.exists());
454        assert!(matches!(instance.claim(&[], |_| {}).unwrap(), Launch::Primary(_)));
455    }
456
457    #[test]
458    fn oversized_or_wrong_token_messages_are_refused() {
459        struct Duplex(std::io::Cursor<Vec<u8>>, Vec<u8>);
460        impl Read for Duplex {
461            fn read(&mut self, b: &mut [u8]) -> std::io::Result<usize> {
462                self.0.read(b)
463            }
464        }
465        impl Write for Duplex {
466            fn write(&mut self, b: &[u8]) -> std::io::Result<usize> {
467                self.1.write(b)
468            }
469            fn flush(&mut self) -> std::io::Result<()> {
470                Ok(())
471            }
472        }
473        let line = |token: &str| format!("{{\"v\":1,\"token\":\"{token}\",\"cwd\":\"/\",\"args\":[\"x\"]}}\n").into_bytes();
474        let mut ok = Duplex(std::io::Cursor::new(line("t")), vec![]);
475        assert!(receive(&mut ok, "t").is_some());
476        assert_eq!(ok.1, b"ok\n");
477        let mut wrong = Duplex(std::io::Cursor::new(line("nope")), vec![]);
478        assert!(receive(&mut wrong, "t").is_none());
479        assert!(wrong.1.is_empty());
480        let mut huge = Duplex(std::io::Cursor::new(vec![b'x'; MAX_MESSAGE + 10]), vec![]);
481        assert!(receive(&mut huge, "").is_none());
482        let mut unterminated = Duplex(std::io::Cursor::new(b"{\"v\":1".to_vec()), vec![]);
483        assert!(receive(&mut unterminated, "").is_none());
484    }
485}
ashlar_native/single_instance.rs

ashlar_native/
single_instance.rs
