forge_browser/
channel.rs

1//! Frames over a Unix socket: length-prefixed, so one read is one frame.
2//!
3//! **The server listens and each host connects.** A profile's service binds
4//! the socket before spawning its host, and a host that finds no socket
5//! keeps painting into its ring and tries again rather than failing.
6
7use std::io::{ErrorKind, Read as _, Write as _};
8use std::os::unix::net::{UnixListener, UnixStream};
9use std::path::{Path, PathBuf};
10use std::sync::atomic::{AtomicBool, Ordering};
11use std::sync::{Arc, Mutex};
12use std::thread::JoinHandle;
13use std::time::Duration;
14
15use crate::frame::Frame;
16
17/// The largest frame the channel accepts: a 4K lossless WebP fits under it,
18/// and anything past it is a corrupt length rather than a frame.
19pub const MAX_FRAME_LEN: usize = 64 * 1024 * 1024;
20
21/// How long a write may block before the host treats the reader as gone. It
22/// bounds one syscall; [`write()`] bounds the whole frame besides, so a reader
23/// draining a byte at a time cannot stretch a frame out for ever.
24const WRITE_TIMEOUT: Duration = Duration::from_secs(2);
25
26/// How long the accept loop sleeps while no host is connected. **One Hz**:
27/// a host that dies leaves the reader idling until its relaunch, and a
28/// hundred wakeups a second for nothing was the cost of a 10 ms poll.
29const ACCEPT_POLL: Duration = Duration::from_secs(1);
30
31/// How long a read waits before the loop looks at its stop flag again.
32const READ_POLL: Duration = Duration::from_millis(200);
33
34/// How long a frame may stall mid-read before the connection is treated as
35/// gone. The host's own write is bounded at two seconds, so five is a wedge.
36const STALL: Duration = Duration::from_secs(5);
37
38/// What one [`read_full`] produced.
39enum Filled {
40    /// Every byte of the buffer arrived.
41    Complete,
42    /// Nothing arrived and the caller may look at its stop flag.
43    Quiet,
44    /// The host closed its end.
45    Closed,
46}
47
48/// Fill `buf` from the stream.
49///
50/// **Progress survives the poll timeout.** The stream answers a read with
51/// nothing available at any point - a frame's body arrives in many chunks,
52/// and the reader polls so it can look at its stop flag - so a timeout
53/// mid-frame resumes where it left off rather than failing the frame. Only a
54/// frame's FIRST byte may report [`Filled::Quiet`]: once a frame has begun,
55/// nothing arriving for [`STALL`] is a wedge.
56fn read_full(
57    stream: &mut UnixStream,
58    buf: &mut [u8],
59    may_be_quiet: bool,
60) -> Result<Filled, String> {
61    let mut filled = 0;
62    let mut stalled = Duration::ZERO;
63    while filled < buf.len() {
64        match stream.read(&mut buf[filled..]) {
65            Ok(0) => return Ok(Filled::Closed),
66            Ok(read) => {
67                filled += read;
68                stalled = Duration::ZERO;
69            }
70            Err(why) if matches!(why.kind(), ErrorKind::WouldBlock | ErrorKind::TimedOut) => {
71                if may_be_quiet && filled == 0 {
72                    return Ok(Filled::Quiet);
73                }
74                stalled += READ_POLL;
75                if stalled >= STALL {
76                    return Err("a frame stalled mid-read".to_owned());
77                }
78            }
79            Err(why) if why.kind() == ErrorKind::Interrupted => {}
80            Err(why) => return Err(format!("the frame was not read: {why}")),
81        }
82    }
83    Ok(Filled::Complete)
84}
85
86/// The socket a profile's host writes frames to, under the profile itself.
87///
88/// **Only a fallback now**: a profile under app-support is deeper than a
89/// Unix socket's ~104-byte address allows, so the server passes the short
90/// path from [`frame_socket_path`] and this is what a host launched by hand
91/// (and the tests) falls back to.
92pub fn socket_path(profile: &Path) -> PathBuf {
93    profile.join("frames.sock")
94}
95
96/// The private directory short sockets live in: `/tmp/forge-<uid>/`.
97///
98/// **A socket address is about 104 bytes**, and a profile's own directory is
99/// deep under app-support, so both the frame socket and the chooser socket
100/// live here. `/tmp` is shared, so the directory is created 0700 and checked
101/// to be this user's own.
102pub fn private_dir(profile: &Path) -> Result<PathBuf, String> {
103    use std::os::unix::fs::{MetadataExt as _, PermissionsExt as _};
104
105    let uid = std::fs::metadata(profile)
106        .map_err(|why| {
107            format!("the profile directory {} cannot be read: {why}", profile.display())
108        })?
109        .uid();
110    let dir = Path::new("/tmp").join(format!("forge-{uid}"));
111    std::fs::create_dir_all(&dir)
112        .map_err(|why| format!("the socket directory {} cannot be made: {why}", dir.display()))?;
113    std::fs::set_permissions(&dir, std::fs::Permissions::from_mode(0o700)).map_err(|why| {
114        format!("the socket directory {} cannot be made private: {why}", dir.display())
115    })?;
116    let meta = std::fs::metadata(&dir)
117        .map_err(|why| format!("the socket directory {} cannot be read: {why}", dir.display()))?;
118    if meta.uid() != uid || meta.permissions().mode() & 0o777 != 0o700 {
119        return Err(format!(
120            "the socket directory {} is not this user's private directory",
121            dir.display()
122        ));
123    }
124    Ok(dir)
125}
126
127/// The frame socket the server binds for one profile: a short path, with the
128/// server's pid so two instances never share one.
129pub fn frame_socket_path(profile: &Path, id: u16) -> Result<PathBuf, String> {
130    Ok(private_dir(profile)?.join(format!("frames-{}-{id}.sock", std::process::id())))
131}
132
133/// Connect to a profile's socket as the writing side.
134pub fn connect(path: &Path) -> Result<UnixStream, String> {
135    let stream = UnixStream::connect(path)
136        .map_err(|why| format!("the frame socket {} did not accept: {why}", path.display()))?;
137    stream
138        .set_write_timeout(Some(WRITE_TIMEOUT))
139        .map_err(|why| format!("the frame socket's write timeout cannot be set: {why}"))?;
140    Ok(stream)
141}
142
143/// The kind byte a message's body carries: one frame, or one audio packet.
144pub const KIND_FRAME: u8 = 1;
145pub const KIND_AUDIO: u8 = 2;
146
147/// One audio packet from the page: interleaved 16-bit PCM.
148///
149/// **PCM, not encoded audio**, because both ends are ours: the host gets
150/// float samples from CEF and the client's player takes them directly.
151#[derive(Clone, Debug, PartialEq, Eq)]
152pub struct Audio {
153    /// The profile whose page made it.
154    pub profile: u16,
155    /// The profile's tab whose page made it.
156    pub tab: u16,
157    /// Monotonic per tab, so a dropped packet is visible.
158    pub seq: u32,
159    pub sample_rate: u32,
160    pub channels: u8,
161    /// Interleaved samples, little-endian on the wire.
162    pub samples: Vec<i16>,
163}
164
165impl Audio {
166    /// The header's own length; the samples follow it.
167    pub const HEADER_LEN: usize = 13;
168
169    pub fn encode(&self) -> Vec<u8> {
170        let mut out = Vec::with_capacity(Self::HEADER_LEN + self.samples.len() * 2);
171        out.extend_from_slice(&self.profile.to_le_bytes());
172        out.extend_from_slice(&self.tab.to_le_bytes());
173        out.extend_from_slice(&self.seq.to_le_bytes());
174        out.extend_from_slice(&self.sample_rate.to_le_bytes());
175        out.push(self.channels);
176        for sample in &self.samples {
177            out.extend_from_slice(&sample.to_le_bytes());
178        }
179        out
180    }
181
182    pub fn decode(bytes: &[u8]) -> Result<Self, String> {
183        if bytes.len() < Self::HEADER_LEN {
184            return Err(format!(
185                "an audio packet is at least {} bytes, got {}",
186                Self::HEADER_LEN,
187                bytes.len()
188            ));
189        }
190        let profile = u16::from_le_bytes([bytes[0], bytes[1]]);
191        let tab = u16::from_le_bytes([bytes[2], bytes[3]]);
192        let seq = u32::from_le_bytes([bytes[4], bytes[5], bytes[6], bytes[7]]);
193        let sample_rate = u32::from_le_bytes([bytes[8], bytes[9], bytes[10], bytes[11]]);
194        let channels = bytes[12];
195        let rest = &bytes[Self::HEADER_LEN..];
196        if !rest.len().is_multiple_of(2) {
197            return Err("an audio packet carries a whole number of samples".to_owned());
198        }
199        let samples =
200            rest.as_chunks::<2>().0.iter().map(|pair| i16::from_le_bytes(*pair)).collect();
201        Ok(Self { profile, tab, seq, sample_rate, channels, samples })
202    }
203}
204
205/// Write one message: its length, its kind, then the body.
206///
207/// **The whole message is bounded, not just one syscall.** The stream's own
208/// write timeout bounds each `write`; a peer that drains a byte at a time
209/// would still hold the message for hours, so the deadline is checked with it
210/// and the message is abandoned whole - the next one carries the page.
211fn write_message(stream: &mut UnixStream, kind: u8, body: &[u8]) -> Result<(), String> {
212    let len = u32::try_from(body.len() + 1)
213        .map_err(|why| format!("a message past the channel's length: {why}"))?;
214    let mut bytes = Vec::with_capacity(5 + body.len());
215    bytes.extend_from_slice(&len.to_le_bytes());
216    bytes.push(kind);
217    bytes.extend_from_slice(body);
218    let deadline = std::time::Instant::now() + WRITE_TIMEOUT;
219    while !bytes.is_empty() {
220        match stream.write(&bytes) {
221            Ok(0) => return Err("the frame socket closed mid-message".to_owned()),
222            Ok(written) => {
223                bytes.drain(..written);
224            }
225            Err(why) if why.kind() == ErrorKind::Interrupted => {}
226            Err(why) if matches!(why.kind(), ErrorKind::WouldBlock | ErrorKind::TimedOut) => {
227                if std::time::Instant::now() >= deadline {
228                    return Err("the message was not written within its bound".to_owned());
229                }
230            }
231            Err(why) => return Err(format!("the message was not written: {why}")),
232        }
233        if std::time::Instant::now() >= deadline {
234            return Err("the message was not written within its bound".to_owned());
235        }
236    }
237    Ok(())
238}
239
240/// Write one frame.
241pub fn write(stream: &mut UnixStream, frame: &Frame) -> Result<(), String> {
242    write_message(stream, KIND_FRAME, &frame.encode())
243}
244
245/// Write one audio packet.
246pub fn write_audio(stream: &mut UnixStream, audio: &Audio) -> Result<(), String> {
247    write_message(stream, KIND_AUDIO, &audio.encode())
248}
249
250/// What one read from a host's socket produced.
251#[derive(Debug)]
252pub enum Reception {
253    /// A whole frame.
254    Frame(Frame),
255    /// One audio packet.
256    Audio(Audio),
257    /// Nothing yet: the read timed out, and the loop looks at its stop flag.
258    Quiet,
259    /// The host closed its end.
260    Closed,
261}
262
263/// One message from a host's socket, as the server routes it.
264#[derive(Debug)]
265pub enum ChannelMessage {
266    Frame(Frame),
267    Audio(Audio),
268}
269
270/// Read one message from a host's socket.
271pub fn read(stream: &mut UnixStream) -> Result<Reception, String> {
272    let mut length = [0_u8; 4];
273    match read_full(stream, &mut length, true)? {
274        Filled::Complete => {}
275        Filled::Quiet => return Ok(Reception::Quiet),
276        Filled::Closed => return Ok(Reception::Closed),
277    }
278    let len = usize::try_from(u32::from_le_bytes(length)).unwrap_or(usize::MAX);
279    if !(1..=MAX_FRAME_LEN).contains(&len) {
280        return Err(format!("a message length of {len} is not a message"));
281    }
282    let mut body = vec![0_u8; len];
283    match read_full(stream, &mut body, false)? {
284        Filled::Complete => {}
285        // A body that began cannot be quiet: the stall bound answers first.
286        Filled::Quiet => return Ok(Reception::Quiet),
287        Filled::Closed => return Ok(Reception::Closed),
288    }
289    let (kind, payload) = body.split_first().ok_or_else(|| "an empty message".to_owned())?;
290    match *kind {
291        KIND_FRAME => Frame::decode(payload).map(Reception::Frame),
292        KIND_AUDIO => Audio::decode(payload).map(Reception::Audio),
293        other => Err(format!("a message kind of {other} is not one this channel knows")),
294    }
295}
296
297/// Bind `path` and hand every frame a host writes to `sink`.
298///
299/// **The reader runs on its own thread** and is stopped through the returned
300/// guard: the server keeps one per open profile, and a closed profile's
301/// thread ends rather than waiting on an accept that will never return.
302///
303/// **Each connection gets a reader of its own.** A host opens one connection
304/// per tab, and accepting one stream at a time would leave every tab after
305/// the first waiting in the backlog with its frames never read.
306pub fn serve<F>(path: &Path, sink: F) -> Result<Reader, String>
307where
308    F: FnMut(ChannelMessage) + Send + 'static,
309{
310    let _ = std::fs::remove_file(path);
311    let listener = UnixListener::bind(path)
312        .map_err(|why| format!("the frame socket {} cannot be bound: {why}", path.display()))?;
313    listener
314        .set_nonblocking(true)
315        .map_err(|why| format!("the frame socket cannot be made non-blocking: {why}"))?;
316    let stop = Arc::new(AtomicBool::new(false));
317    let own_stop = Arc::clone(&stop);
318    let own_path = path.to_path_buf();
319    let wake = path.to_path_buf();
320    let sink = Arc::new(Mutex::new(sink));
321    let thread = std::thread::spawn(move || {
322        let mut connections: Vec<JoinHandle<()>> = Vec::new();
323        while !own_stop.load(Ordering::Acquire) {
324            match listener.accept() {
325                Ok((stream, _)) => {
326                    let sink = Arc::clone(&sink);
327                    let stop = Arc::clone(&own_stop);
328                    let path = own_path.clone();
329                    connections.push(std::thread::spawn(move || {
330                        read_connection(stream, &path, &stop, &sink);
331                    }));
332                }
333                Err(why) if why.kind() == ErrorKind::WouldBlock => {
334                    std::thread::sleep(ACCEPT_POLL);
335                }
336                Err(why) => {
337                    eprintln!(
338                        "cef host: the frame socket accept failed (event_name \
339                         cef_frames_accept_failed, {}): {why}",
340                        own_path.display()
341                    );
342                    break;
343                }
344            }
345        }
346        for connection in connections {
347            let _ = connection.join();
348        }
349        let _ = std::fs::remove_file(&own_path);
350    });
351    Ok(Reader { stop, wake, path: path.to_path_buf(), thread: Some(thread) })
352}
353
354/// Read one host connection until it closes or the profile's reader stops.
355fn read_connection<F>(mut stream: UnixStream, path: &Path, stop: &AtomicBool, sink: &Mutex<F>)
356where
357    F: FnMut(ChannelMessage),
358{
359    if let Err(why) = stream.set_read_timeout(Some(READ_POLL)) {
360        eprintln!("cef host: the frame socket's read timeout was refused: {why}");
361    }
362    // **Blocking, so the timeout is the only bound.** An accepted socket can
363    // inherit the listener's non-blocking flag, and then a body read fails
364    // the moment no byte is ready.
365    if let Err(why) = stream.set_nonblocking(false) {
366        eprintln!("cef host: the frame socket cannot be made blocking: {why}");
367    }
368    loop {
369        if stop.load(Ordering::Acquire) {
370            return;
371        }
372        match read(&mut stream) {
373            Ok(Reception::Frame(frame)) => {
374                (sink.lock().unwrap_or_else(std::sync::PoisonError::into_inner))(
375                    ChannelMessage::Frame(frame),
376                );
377            }
378            Ok(Reception::Audio(audio)) => {
379                (sink.lock().unwrap_or_else(std::sync::PoisonError::into_inner))(
380                    ChannelMessage::Audio(audio),
381                );
382            }
383            Ok(Reception::Quiet) => {}
384            Ok(Reception::Closed) => break,
385            Err(why) => {
386                eprintln!(
387                    "cef host: the frame stream ended (event_name cef_frames_read_failed, {}): \
388                     {why}",
389                    path.display()
390                );
391                break;
392            }
393        }
394    }
395}
396
397/// One profile's frame reader, as the server holds it.
398pub struct Reader {
399    stop: Arc<AtomicBool>,
400    wake: PathBuf,
401    path: PathBuf,
402    thread: Option<JoinHandle<()>>,
403}
404
405impl Reader {
406    /// End the reader: the flag stops the loop, and a connection to the
407    /// socket wakes an accept already waiting when it was set.
408    pub fn stop(mut self) {
409        self.end();
410    }
411
412    fn end(&mut self) {
413        self.stop.store(true, Ordering::Release);
414        let _ = UnixStream::connect(&self.wake);
415        if let Some(thread) = self.thread.take() {
416            let _ = thread.join();
417        }
418        let _ = std::fs::remove_file(&self.path);
419    }
420}
421
422impl Drop for Reader {
423    fn drop(&mut self) {
424        self.end();
425    }
426}
427
428#[cfg(test)]
429mod tests {
430    use super::frame_socket_path;
431
432    /// **A profile name can be 64 characters, and its socket must still
433    /// bind.** A Unix socket's address is about 104 bytes; the profile tree
434    /// under app-support plus a full-length name is far past it, which is
435    /// why the frame socket lives in the short private directory (#2539 made
436    /// an omitted profile the seat's own, so names are long by default).
437    #[test]
438    fn a_full_length_profile_name_still_leaves_a_short_socket() {
439        let dir = tempfile::tempdir().expect("a temp dir");
440        let profile = dir.path().join("p".repeat(64)).join("cef");
441        std::fs::create_dir_all(&profile).expect("the profile dir");
442        let path = frame_socket_path(&profile, 1).expect("the frame socket path");
443        assert!(
444            path.as_os_str().len() < 104,
445            "{} is {} bytes, past a Unix socket's address",
446            path.display(),
447            path.as_os_str().len(),
448        );
449        assert!(
450            path.starts_with(std::path::Path::new("/tmp")),
451            "the short path is outside the profile tree: {}",
452            path.display(),
453        );
454    }
455}
forge_browser/channel.rs

forge_browser/
channel.rs
