rsbinder 0.11.0

// Copyright 2022 Jeff Kim <hiking90@gmail.com>
// SPDX-License-Identifier: Apache-2.0

//! `RpcProxy` — client-side handle to a remote RPC object.
//!
//! A **distinct `IBinder` type** from `proxy::ProxyHandle`. It never
//! goes through the u32 kernel handle / `handle_to_proxy` / cache-pin
//! machinery — RPC has its own `RpcAddress` identity space and
//! its own ref-count. Android made `BpBinder` a dual-mode
//! `variant<BinderHandle, RpcHandle>`; because rsbinder's `IBinder` is
//! a trait, a separate type is cleaner.
//!
//! The generator emits `as_remote().ok_or(BadType)?` (a
//! [`RemoteProxy`](crate::RemoteProxy) trait object) instead of the
//! kernel-only `as_proxy().unwrap()`, so the **generated** `Bp*` stub
//! drives this proxy directly — the same single stub also drives the
//! kernel `ProxyHandle`.

use std::any::Any;
use std::mem::ManuallyDrop;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{self, Arc, OnceLock, RwLock};

use crate::binder::{DeathRecipient, IBinder, SIBinder, Stability, Transactable, WIBinder};
use crate::binder::{TransactionCode, TransactionFlags};
use crate::error::{Result, StatusCode};
use crate::parcel::Parcel;

use super::address::RpcAddress;
use super::session::RpcSessionInner;

/// A handle to a remote object reachable over an RPC session.
pub struct RpcProxy {
    addr: RpcAddress,
    /// The interface descriptor. The RPC wire transmits only an
    /// address (not a descriptor string), so a proxy resolved from
    /// `read_binder`/`get_root` starts empty and is stamped **once,
    /// in place** by the generated typed stub's `from_binder`
    /// (via `stamp_descriptor`).
    /// In-place — never a replacement proxy — keeps the dedup-cache
    /// identity and the single `DEC_STRONG` intact.
    descriptor: OnceLock<String>,
    /// **Strong** session ref (AOSP `BpBinder::RpcSessionBinder` holds
    /// `sp<RpcSession>`): a proxy alone keeps its session — and thus
    /// the connection — alive. The session's `remote_proxies` table is
    /// `Weak`, so the only cycle back to the session runs through a
    /// *local* object the peer holds; that is broken by
    /// [`RpcSessionInner::on_session_dead`], which runs when a serve loop
    /// ends, when a transaction fails on a lost connection, or on an
    /// explicit [`RpcSession::close_session`](super::RpcSession::close_session) —
    /// the last being the only break available to a session that neither
    /// serves nor transacts again.
    session: Arc<RpcSessionInner>,
    /// Death-notification state, mirroring the kernel
    /// [`ProxyHandle`](crate::proxy::ProxyHandle) exactly. RPC has no
    /// death *wire* message (AOSP `RpcState::sendObituaries`): an RPC
    /// object dies when its **session connection drops**, so the
    /// session fires every cached proxy's obituary when its serve loop
    /// ends (see [`RpcSessionInner::send_session_obituaries`]). Set
    /// once when the obituary is dispatched. Every access is under the
    /// `recipients` write lock (kernel `mLock` parity), which supplies
    /// the happens-before — hence `Relaxed` throughout.
    obituary_sent: AtomicBool,
    recipients: RwLock<Vec<sync::Weak<dyn DeathRecipient>>>,
}

impl RpcProxy {
    /// Identity of the owning session, for `write_binder`'s same-session check.
    pub(crate) fn session_ptr(&self) -> *const RpcSessionInner {
        Arc::as_ptr(&self.session)
    }

    pub(crate) fn new(addr: RpcAddress, session: Arc<RpcSessionInner>) -> Self {
        RpcProxy {
            addr,
            descriptor: OnceLock::new(),
            session,
            obituary_sent: AtomicBool::new(false),
            recipients: RwLock::new(Vec::new()),
        }
    }

    /// Fire `binder_died` on every registered recipient — called by the
    /// owning session when its connection drops (AOSP
    /// `BpBinder::sendObituary`, the RPC branch). Mirrors the kernel
    /// [`ProxyHandle::send_obituary`](crate::proxy::ProxyHandle) state
    /// machine **minus** the kernel-only `BC_CLEAR_DEATH_NOTIFICATION`
    /// / `flush_commands` (death over RPC is connection-drop-driven,
    /// not a wire command). Idempotent: a second call (e.g. a serve
    /// loop ending after a transact already observed the close) sees
    /// `obituary_sent == true` and returns.
    pub(crate) fn send_obituary(&self, who: &WIBinder) {
        let snapshot: Vec<sync::Weak<dyn DeathRecipient>> = {
            // All `obituary_sent` reads/writes that race `link`/`unlink`
            // happen under this write lock (kernel parity: `mLock`).
            // A poisoned recipients lock means a prior panic on this
            // proxy's death pathway — log + return best-effort instead
            // of panicking the serve loop.
            let mut recipients = match self.recipients.write() {
                Ok(g) => g,
                Err(_) => {
                    log::error!(
                        "RPC obituary skipped: recipients lock poisoned for addr {:?}",
                        self.addr
                    );
                    return;
                }
            };
            if self.obituary_sent.load(Ordering::Relaxed) {
                return;
            }
            let snapshot = std::mem::take(&mut *recipients);
            self.obituary_sent.store(true, Ordering::Relaxed);
            snapshot
        };
        // Callbacks outside the lock so a recipient may re-enter
        // `unlink_to_death`/`link_to_death` without self-deadlock
        // (AOSP unlocks before `reportOneDeath`). Panic-isolated so one
        // buggy recipient cannot abort the serve thread or starve the
        // rest.
        for weak in &snapshot {
            let Some(recipient) = weak.upgrade() else {
                continue;
            };
            let r = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
                recipient.binder_died(who);
            }));
            if let Err(payload) = r {
                let msg = payload
                    .downcast_ref::<&'static str>()
                    .copied()
                    .or_else(|| payload.downcast_ref::<String>().map(String::as_str))
                    .unwrap_or("<non-string panic payload>");
                log::error!(
                    "DeathRecipient panicked during binder_died for RPC addr {:?}: {msg}",
                    self.addr
                );
            }
        }
    }

    /// The remote object's RPC address.
    pub fn address(&self) -> RpcAddress {
        self.addr
    }

    /// Stamp the interface descriptor — known only to
    /// the generated typed stub at compile time — onto this
    /// **already-cached** proxy, in place. First write wins and is
    /// idempotent for the same interface (one wire address identifies
    /// one remote object = one interface). Never replaces the proxy:
    /// a replacement would send a second `DEC_STRONG` on drop and
    /// split the per-address dedup cache.
    ///
    /// Because the generated `from_binder` stamps *its own*
    /// `$descriptor` here **before** its `binder.descriptor() !=
    /// $descriptor` check, that check is self-referential for a fresh
    /// RPC proxy: unlike the kernel `ProxyHandle` path it does **not**
    /// validate the remote's actual interface. A wrong-interface cast
    /// is not rejected at `from_binder`; it surfaces as a transact-time
    /// `StatusCode` when the server rejects the interface token. This
    /// is inherent to the Android RPC wire (no descriptor transmitted),
    /// not a defect — `first write wins` additionally protects an
    /// in-use proxy from a later differing cast (the second
    /// `from_binder`'s descriptor check then returns `None`).
    ///
    /// **Return**: `true` if this call wrote the descriptor, `false`
    /// if it was already stamped. Surfacing the `OnceLock::set` result
    /// lets a caller distinguish fresh-stamp from already-stamped
    /// without an extra `descriptor()` round-trip; existing callers
    /// (generator `from_binder`) can keep ignoring the result with
    /// `let _ = …`.
    pub(crate) fn stamp_descriptor(&self, descriptor: &str) -> bool {
        self.descriptor.set(descriptor.to_string()).is_ok()
    }

    /// The stamped descriptor, or `""` if not yet stamped (a proxy
    /// fresh off the wire before its typed stub is built).
    fn descriptor_str(&self) -> &str {
        self.descriptor.get().map(String::as_str).unwrap_or("")
    }

    /// Build an RPC-mode request `Parcel` for `descriptor`, with the
    /// session's object hooks attached and the interface token written.
    /// Hand-written typed stubs call this, write their args, then
    /// [`RpcProxy::transact`].
    pub fn build_request(&self, descriptor: &str) -> Result<Parcel> {
        let inner = &self.session;
        let mut data = Parcel::new();
        // Enter RPC mode + stamp the negotiated FD policy (default
        // `None` ⇒ `ParcelFileDescriptor::serialize` rejects FDs).
        data.configure_rpc(
            inner.parcel_ops(),
            inner.fd_mode(),
            inner.records_fd_positions(),
        );
        super::session::write_rpc_interface_token(&mut data, descriptor)?;
        Ok(data)
    }

    /// Send an outbound transaction to the remote object. Returns the
    /// reply parcel (`None` for oneway).
    pub fn transact(
        &self,
        code: TransactionCode,
        data: &Parcel,
        flags: TransactionFlags,
    ) -> Result<Option<Parcel>> {
        self.session.client_transact(self.addr, code, data, flags)
    }
}

/// The RPC proxy implements the same generalized
/// [`RemoteProxy`](crate::RemoteProxy) trait as the kernel
/// `ProxyHandle`, so the one generated `Bp*` stub drives either stack
/// (generator emits `as_remote()`). `prepare_transact` writes
/// the interface token from the descriptor stamped in place by the
/// generated `from_binder` (`stamp_descriptor`).
impl crate::binder::RemoteProxy for RpcProxy {
    fn prepare_transact(&self, write_header: bool) -> Result<Parcel> {
        let inner = &self.session;
        let mut data = Parcel::new();
        data.configure_rpc(
            inner.parcel_ops(),
            inner.fd_mode(),
            inner.records_fd_positions(),
        );
        if write_header {
            super::session::write_rpc_interface_token(&mut data, self.descriptor_str())?;
        }
        Ok(data)
    }

    fn submit_transact(
        &self,
        code: TransactionCode,
        data: &Parcel,
        flags: TransactionFlags,
    ) -> Result<Option<Parcel>> {
        RpcProxy::transact(self, code, data, flags)
    }
}

impl Drop for RpcProxy {
    fn drop(&mut self) {
        // Last `Arc<RpcProxy>` for this address is going away: tell the
        // peer to drop its strong ref. Best-effort — never panic in
        // Drop, and a dead session simply means the peer is already
        // gone.
        //
        // `queue_dec_strong` is a non-blocking mpsc enqueue handled by
        // the session's reaper thread, rather than a synchronous send:
        // on a single-slot session a synchronous `find_conn` would
        // `cv.wait` if another thread already drives the slot, blocking
        // the user's hot path for the full peer round-trip.
        let inner = &self.session;
        inner.queue_dec_strong(self.addr);
        // Identity-checked: if this proxy's `Arc` already hit 0 and
        // a concurrent `read_binder` re-cached a fresh live proxy
        // for the same address, this stale `Drop` must NOT evict
        // that successor (see `forget_remote_if`).
        inner.forget_remote_if(&self.addr, self as *const RpcProxy as *const ());
    }
}

impl IBinder for RpcProxy {
    /// Register a death recipient. Death over RPC = the **session
    /// connection dropping** (AOSP `RpcState::sendObituaries`): the
    /// recipient fires when the session's serve loop ends. Mirrors the
    /// kernel [`ProxyHandle::link_to_death`](crate::proxy::ProxyHandle)
    /// minus the kernel `requestDeathNotification` IPC (RPC has no
    /// death wire message).
    ///
    /// **Detection requires the session to be served.** AOSP rejects an
    /// RPC `linkToDeath` outright unless `getMaxIncomingThreads() >= 1`;
    /// the rsbinder analogue is that the obituary is delivered by
    /// [`RpcSession::serve_blocking`](super::session::RpcSession) on
    /// connection loss, so a peer that wants death notification must
    /// run a serve loop (it already does for nested callbacks). A
    /// session that is never served still registers the recipient and
    /// delivers **lazily**: the first transaction that fails on the lost
    /// connection runs the same death sequence (obituaries + local
    /// object release) — a documented rsbinder model property, faithful
    /// to AOSP's incoming-thread requirement.
    ///
    /// A client built with
    /// [`RpcUnixClientConfig::incoming_connections`](super::session::RpcUnixClientConfig::incoming_connections)
    /// `≥ 1` *is* served — by the threads on its incoming (callback)
    /// connections — so it observes the drop at once (AOSP
    /// `onSessionAllIncomingThreadsEnded`) with no serve loop of its own.
    fn link_to_death(&self, recipient: sync::Weak<dyn DeathRecipient>) -> Result<()> {
        // Lock first, then check `obituary_sent` — kernel/AOSP ordering
        // (`BpBinder::linkToDeath` checks `mObitsSent` under `mLock`).
        let mut recipients = self.recipients.write().map_err(|_| {
            // A poisoned recipients lock means a prior panic on this
            // proxy's death pathway. Surface as `DeadObject` so a
            // caller can recover (drop the binder, re-establish) rather
            // than panicking out of a public `IBinder` method.
            StatusCode::DeadObject
        })?;
        if self.obituary_sent.load(Ordering::Relaxed) {
            // Connection already dropped — AOSP returns DEAD_OBJECT.
            return Err(StatusCode::DeadObject);
        }
        recipients.push(recipient);
        Ok(())
    }

    /// Unregister a death recipient (single-position, order-preserving
    /// — kernel `removeAt(i)` parity, *not* `retain`, so a duplicate
    /// registration keeps its remaining subscriptions).
    fn unlink_to_death(&self, recipient: sync::Weak<dyn DeathRecipient>) -> Result<()> {
        let mut recipients = self.recipients.write().map_err(|_| {
            // A poisoned recipients lock means a prior panic on this
            // proxy's death pathway. Surface as `DeadObject` so a
            // caller can recover (drop the binder, re-establish) rather
            // than panicking out of a public `IBinder` method.
            StatusCode::DeadObject
        })?;
        if self.obituary_sent.load(Ordering::Relaxed) {
            return Err(StatusCode::DeadObject);
        }
        let Some(i) = recipients
            .iter()
            .position(|r| sync::Weak::ptr_eq(r, &recipient))
        else {
            return Err(StatusCode::NameNotFound);
        };
        recipients.remove(i);
        Ok(())
    }

    fn ping_binder(&self) -> Result<()> {
        // PING_TRANSACTION round-trip (no payload, no reply body).
        let data = Parcel::new();
        self.session
            .client_transact(self.addr, crate::binder::PING_TRANSACTION, &data, 0)?;
        Ok(())
    }

    fn as_any(&self) -> &dyn Any {
        self
    }

    fn as_transactable(&self) -> Option<&dyn Transactable> {
        None
    }

    fn descriptor(&self) -> &str {
        self.descriptor_str()
    }

    fn is_remote(&self) -> bool {
        true
    }

    // RPC ref-count is driven by the wire `DEC_STRONG` (sent from
    // `Drop`), not by these `SIBinder`/`WIBinder` clone/drop hooks —
    // same no-op shape as `ProxyHandle` under the cache-pin model.
    fn inc_strong(&self, _strong: &SIBinder) -> Result<()> {
        Ok(())
    }

    fn attempt_inc_strong(&self) -> bool {
        true
    }

    fn dec_strong(&self, _strong: Option<ManuallyDrop<SIBinder>>) -> Result<()> {
        Ok(())
    }

    fn inc_weak(&self, _weak: &WIBinder) -> Result<()> {
        Ok(())
    }

    fn dec_weak(&self) -> Result<()> {
        Ok(())
    }

    fn stability(&self) -> Stability {
        Stability::default()
    }
}