coreshift_core/

transport.rs

// This Source Code Form is subject to the terms of the Mozilla Public
// License, v. 2.0. If a copy of the MPL was not distributed with this
// file, You can obtain one at https://mozilla.org/MPL/2.0/

//! Transport seam: commands, one-shot replies, and persistent watch sinks.
//!
//! This module is the message-shaped boundary between a daemon's
//! transport-agnostic channel logic and its transport backends (the
//! abstract-socket daemon today, a Binder backend later). The channel logic
//! sees **commands and watchers**, never byte streams: each backend's only job
//! is to turn its transport's native events into [`TransportEvent`]s and push
//! structured update values into [`WatcherSink::push`].
//!
//! The seam is generic over the command and reply payload types (`C` and `R`);
//! the channel logic (e.g. `FgCtx`/`FpsCtx`) provides the domain types and
//! never knows which backend delivered them.
//!
//! ### Two shapes, kept distinct
//!
//! - [`ReplySink`] is **one-shot**: the single reply to a command, consumed
//!   exactly once. The socket backend answers in-line (no hop); a Binder
//!   backend unblocks a waiting transaction thread.
//! - [`WatcherId`] / [`WatcherSink`] are **persistent**: a long-lived push
//!   target with its own close / death lifecycle, outliving any single
//!   transaction.
//!
//! Backends may implement both with the same underlying handle (a socket
//! backend uses one fd for both roles over its lifetime); the seam does not
//! assume that in general.
//!
//! ### Single-owner watcher registry
//!
//! A watch registration is itself a command, so it flows through the same
//! ingestion path as `ping`/`status`. The reactor thread is therefore the only
//! actor that ever allocates a [`WatcherId`]; [`TransportEvent::WatcherClosed`]
//! and [`TransportEvent::WatcherDied`] flow back through the same path and are
//! processed on that same thread.
//!
//! ### What ships here, and what ships with the Binder backend
//!
//! This module ships the message types only. The bounded per-channel ingestion
//! queue + eventfd is Binder-backend machinery: only a Binder backend's
//! `on_transact` fires on an arbitrary threadpool thread (a genuine cross-thread
//! boundary), while the socket backend's thread already *is* the owning reactor
//! thread and answers in-line with zero hop. The queue is added when the Binder
//! backend lands, not as dead machinery before any consumer exists.

use crate::CoreError;

/// An opaque long-lived handle identifying one watch subscriber.
///
/// Allocated by the backend's owning thread; never interpreted by the channel
/// logic. Backends should hand out monotonically increasing values from a
/// per-loop counter.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
pub struct WatcherId(u64);

impl WatcherId {
    /// Construct a watcher id from a raw value.
    ///
    /// The caller (the backend's owning thread) is responsible for uniqueness
    /// within its loop.
    pub fn new(value: u64) -> Self {
        Self(value)
    }

    /// The raw identifier value.
    pub fn as_u64(&self) -> u64 {
        self.0
    }
}

/// Delivery closure backing a one-shot reply sink.
type ReplySend<R> = Box<dyn FnOnce(R) -> Result<(), CoreError> + Send>;
/// Push closure backing a persistent watcher sink.
type WatcherPush<U> = Box<dyn FnMut(&U) -> Result<(), CoreError> + Send>;

/// One-shot reply target for a command.
///
/// Consumed exactly once via [`Self::send`]; the closure performs the
/// backend-specific delivery (socket: serialize and write to the peer fd;
/// binder: send through a oneshot channel to the blocked transaction thread).
pub struct ReplySink<R> {
    send: ReplySend<R>,
}

impl<R> ReplySink<R> {
    /// Wrap a delivery closure as a reply sink.
    pub fn new(send: impl FnOnce(R) -> Result<(), CoreError> + Send + 'static) -> Self {
        Self {
            send: Box::new(send),
        }
    }

    /// Deliver the reply, consuming the sink.
    ///
    /// ### Errors
    /// Returns the backend's delivery error (e.g. the peer went away); the
    /// caller may log it and drop the sink.
    ///
    /// ### Drop semantics
    /// Dropping a [`ReplySink`] without calling [`Self::send`] loses the reply
    /// (the closure is dropped). For a Binder backend the waiting transaction
    /// thread observes the oneshot disconnect and errors out; the socket
    /// backend simply never writes.
    pub fn send(self, reply: R) -> Result<(), CoreError> {
        (self.send)(reply)
    }
}

impl<R> std::fmt::Debug for ReplySink<R> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("ReplySink").finish_non_exhaustive()
    }
}

/// Persistent push target for watch updates.
///
/// The channel logic holds one [`WatcherSink`] per registered watcher and
/// calls [`Self::push`] to deliver a structured update value. A failed push
/// (e.g. `EPIPE` on a gone socket) tells the channel logic the watcher is
/// dead; the corresponding [`TransportEvent::WatcherClosed`] / `WatcherDied`
/// still flows back through the ingestion path for cleanup.
pub struct WatcherSink<U> {
    push: WatcherPush<U>,
}

impl<U> WatcherSink<U> {
    /// Wrap a push closure as a watcher sink.
    pub fn new(push: impl FnMut(&U) -> Result<(), CoreError> + Send + 'static) -> Self {
        Self {
            push: Box::new(push),
        }
    }

    /// Deliver one update value to the subscriber.
    ///
    /// ### Errors
    /// Returns the backend's delivery error (e.g. the peer went away); the
    /// caller should evict the watcher and stop pushing.
    pub fn push(&mut self, update: &U) -> Result<(), CoreError> {
        (self.push)(update)
    }
}

impl<U> std::fmt::Debug for WatcherSink<U> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("WatcherSink").finish_non_exhaustive()
    }
}

/// A transport-agnostic command or watcher-lifecycle event.
///
/// Backends produce these; channel logic consumes them. `C` is the domain
/// command type, `R` the domain reply type.
#[derive(Debug)]
pub enum TransportEvent<C, R> {
    /// One command with the caller identity captured at delivery time and a
    /// one-shot reply target.
    Command {
        /// The domain command.
        cmd: C,
        /// The calling uid captured synchronously by the backend when the
        /// command arrived (`None` when the transport exposes no caller
        /// identity, e.g. a socket backend with no auth policy).
        calling_uid: Option<u32>,
        /// The one-shot reply target, consumed exactly once.
        reply: ReplySink<R>,
    },
    /// A watcher's transport went away (socket EOF / hangup, or a client
    /// unwatch). The id was allocated by this loop.
    WatcherClosed(WatcherId),
    /// A watcher died remotely (binder death-recipient callback). The id was
    /// allocated by this loop.
    WatcherDied(WatcherId),
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::Arc;
    use std::sync::atomic::{AtomicU32, AtomicUsize, Ordering};

    #[test]
    fn watcher_id_round_trips_value() {
        let id = WatcherId::new(7);
        assert_eq!(id.as_u64(), 7);
        assert_eq!(WatcherId::new(7), id);
        assert_ne!(WatcherId::new(8), id);
    }

    #[test]
    fn reply_sink_delivers_exactly_once() {
        let delivered = Arc::new(AtomicU32::new(0));
        let captured = Arc::clone(&delivered);
        let sink = ReplySink::new(move |reply: u32| {
            captured.fetch_add(reply, Ordering::SeqCst);
            Ok(())
        });
        sink.send(42).unwrap();
        assert_eq!(delivered.load(Ordering::SeqCst), 42);
    }

    #[test]
    fn reply_sink_propagates_backend_error() {
        let sink = ReplySink::new(|_: u32| Err(CoreError::sys(libc::EPIPE, "reply")));
        let err = sink.send(1).unwrap_err();
        assert_eq!(err.raw_os_error(), Some(libc::EPIPE));
    }

    #[test]
    fn watcher_sink_pushes_borrowed_updates() {
        let seen = Arc::new(std::sync::Mutex::new(Vec::new()));
        let captured = Arc::clone(&seen);
        let mut sink = WatcherSink::new(move |update: &u32| {
            captured.lock().unwrap().push(*update);
            Ok(())
        });
        sink.push(&1).unwrap();
        sink.push(&2).unwrap();
        assert_eq!(*seen.lock().unwrap(), vec![1, 2]);
    }

    #[test]
    fn watcher_sink_propagates_push_error() {
        let mut sink = WatcherSink::new(|_: &u32| Err(CoreError::sys(libc::EPIPE, "push")));
        let err = sink.push(&1).unwrap_err();
        assert_eq!(err.raw_os_error(), Some(libc::EPIPE));
    }

    #[derive(Debug)]
    enum Cmd {
        Ping,
    }
    #[derive(Debug)]
    enum Reply {
        Pong,
    }

    #[test]
    fn transport_event_carries_command_and_identity() {
        let answered = Arc::new(AtomicUsize::new(0));
        let captured = Arc::clone(&answered);
        let event = TransportEvent::Command {
            cmd: Cmd::Ping,
            calling_uid: Some(1000),
            reply: ReplySink::new(move |r: Reply| {
                assert!(matches!(r, Reply::Pong));
                captured.fetch_add(1, Ordering::SeqCst);
                Ok(())
            }),
        };
        match event {
            TransportEvent::Command {
                cmd,
                calling_uid,
                reply,
            } => {
                assert!(matches!(cmd, Cmd::Ping));
                assert_eq!(calling_uid, Some(1000));
                reply.send(Reply::Pong).unwrap();
            }
            _ => panic!("expected command event"),
        }
        assert_eq!(answered.load(Ordering::SeqCst), 1);
    }

    #[test]
    fn transport_event_carries_watcher_close_and_died() {
        let id = WatcherId::new(3);
        match TransportEvent::<Cmd, Reply>::WatcherClosed(id) {
            TransportEvent::WatcherClosed(got) => assert_eq!(got, id),
            _ => panic!("expected watcher closed"),
        }
        match TransportEvent::<Cmd, Reply>::WatcherDied(id) {
            TransportEvent::WatcherDied(got) => assert_eq!(got, id),
            _ => panic!("expected watcher died"),
        }
    }

    #[test]
    fn sinks_are_debug_but_opaque() {
        let sink: ReplySink<u32> = ReplySink::new(|_| Ok(()));
        let text = format!("{sink:?}");
        assert!(text.starts_with("ReplySink"));

        let watcher: WatcherSink<u32> = WatcherSink::new(|_| Ok(()));
        let text = format!("{watcher:?}");
        assert!(text.starts_with("WatcherSink"));
    }
}