rsbinder 0.12.0

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

/*
 * Copyright (C) 2021 The Android Open Source Project
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

//! This module lets you use the Tokio `spawn_blocking` pool with AIDL in async
//! Rust code (enabled by the default `tokio` feature).
//!
//! It works by defining a type [`Tokio`], which you use as the generic
//! parameter `P` of the `IFooAsync<P>` trait generated by the AIDL compiler.
//! Obtain an async proxy either with the crate-root [`get_interface_async`] helper,
//! or by upgrading a sync handle with [`Strong::into_async`]:
//!
//! ```text
//! use rsbinder::Tokio;
//!
//! // Look up a service and cast straight to its async interface.
//! let svc = rsbinder::get_interface_async::<dyn SomeAsyncInterface<Tokio>>("name").await?;
//!
//! // Or upgrade an existing sync Strong handle:
//! let svc_async = sync_handle.into_async::<Tokio>();
//! ```
//!
//! [`Tokio`]: crate::Tokio
//! [`get_interface_async`]: crate::get_interface_async
//! [`Strong::into_async`]: crate::Strong::into_async

use crate::{
    hub, BinderAsyncPool, BinderAsyncRuntime, BoxFuture, DeathRecipient, FromIBinder, SIBinder,
    StatusCode, Strong, WIBinder,
};
use std::future::Future;
use std::pin::Pin;
use std::sync::{Arc, Mutex};
use std::task::{Context, Poll};

/// A `spawn_blocking` failure as a status; a panic resumes (none on Android: `panic = abort`).
pub(crate) fn join_error_status(e: tokio::task::JoinError) -> StatusCode {
    if e.is_panic() {
        std::panic::resume_unwind(e.into_panic());
    }
    if e.is_cancelled() {
        StatusCode::FailedTransaction
    } else {
        StatusCode::Unknown
    }
}

/// Retrieve an existing service for a particular interface — one
/// `getService` wire call, which does not block (see
/// [`crate::hub::try_get_interface`]). For an event-driven wait use
/// [`wait_for_interface_async`]; for a lookup that will not start an
/// unregistered lazy service, [`check_interface_async`] (except on Android
/// 15 r6+, where `check_*` is carried by `getService` too — see the `hub`
/// module doc).
///
/// The `_async` suffix is what separates the crate root's awaitable lookups
/// from the synchronous ones in [`crate::hub`]. It matches
/// [`crate::connect_async`], the facade entry point.
pub async fn get_interface_async<T: FromIBinder + ?Sized + 'static>(
    name: &str,
) -> Result<Strong<T>, StatusCode> {
    fn lookup<T: FromIBinder + ?Sized + 'static>(name: &str) -> Result<Strong<T>, StatusCode> {
        hub::try_get_interface::<T>(name)?.ok_or(StatusCode::NameNotFound)
    }

    if crate::is_handling_transaction() {
        // See comment in the BinderAsyncPool impl.
        return lookup::<T>(name);
    }

    let name = name.to_string();
    tokio::task::spawn_blocking(move || lookup::<T>(&name))
        .await
        .unwrap_or_else(|e| Err(join_error_status(e)))
}

/// Wait until the service named `name` is registered, then cast it to `T` —
/// the awaitable counterpart of [`crate::hub::wait_for_interface`] (AOSP
/// `binder_tokio::wait_for_interface`).
///
/// The contract is the synchronous one: the wait is **unbounded**, it resolves
/// as soon as the service appears, and it gives up with
/// [`StatusCode::NameNotFound`] only when the service manager is unreachable
/// or refuses `registerForNotifications` for `name` (an invalid name, an
/// SELinux `find` denial); a [`hub::default`] failure is returned as is. A
/// name registered under a different interface is
/// [`StatusCode::BadType`], exactly as in the sync path — the wait is not
/// retried on a descriptor mismatch.
///
/// # Dropping the future
///
/// Dropping it ends the wait: the drop marks it cancelled, the thread carrying
/// the wait leaves its condvar at once, and on its way out it unregisters the
/// callback it had registered with the service manager. That is what makes it
/// usable as one arm of a `select!` or under a `timeout`. Nothing is left
/// behind on the service manager — but the unregister is a wire call made
/// after the drop returns, so a caller that re-registers the same name
/// immediately can still see the old callback counted.
///
/// Shutting the runtime down counts as dropping it — which matters, because
/// `Runtime::drop` waits for blocking tasks: a wait that could not be
/// cancelled would hold the shutdown open for as long as the service stayed
/// unregistered.
///
/// # Cost
///
/// One thread of the Tokio **blocking pool** for as long as the wait lasts. The
/// pool is the same one every outbound binder call goes through, so a process
/// that parks many simultaneous waits there reduces what is left for calls;
/// service startup waits (a handful, at process start) are what this is for.
///
/// # What drives it
///
/// The wait is a single `spawn_blocking` task, and the blocking pool runs
/// independently of any scheduler: `onRegistration` arriving on a binder thread
/// and waking that task's condvar, and the per-second `getService` re-poll,
/// both proceed whatever the runtime is doing. What the runtime decides is how
/// this future — the handle to that task — is driven:
///
/// | How this future is driven | Completion | A `timeout` around it |
/// |---|---|---|
/// | awaited on a runtime that is being driven | observed | fires |
/// | awaited inside a handler this crate drives with [`TokioRuntime`] | observed — the wake comes from the blocking pool, not from the scheduler | only while that runtime's timer is advancing, which is the condition in [`TokioRuntime`]'s *Stalls* |
/// | `tokio::spawn`ed onto a current-thread runtime nobody is parked in | never — the task is not polled, so the wait never starts and no pool thread is taken | – |
///
/// The first poll is what submits the blocking task, so a runtime that stops
/// being driven afterwards leaves the wait holding its pool thread and polling
/// the service manager every second, with nobody there to take the result.
///
/// # From inside a transaction
///
/// Calling it while handling a transaction keeps that binder thread occupied
/// until the service appears, and the in-flight transaction goes unanswered for
/// that long — a `timeout` bounds the wait only under the condition in the
/// table above. Look services up before serving, not while serving.
///
/// Unlike the single-round-trip lookups, this does not run inline on the
/// calling thread: the wait goes to the blocking pool wherever it is called
/// from, so it is the reply that is delayed, not the kernel's deadlock
/// prevention that is lost.
///
/// # Thread pool
///
/// A binder thread pool ([`crate::ProcessState::start_thread_pool`]) is what
/// makes the wait event-driven; without one it degrades to the ~1s re-poll,
/// exactly as the synchronous path does.
pub async fn wait_for_interface_async<T: FromIBinder + ?Sized + 'static>(
    name: &str,
) -> Result<Strong<T>, StatusCode> {
    // Outside the closure: the drop guard must reach the wait's condvar after this future ends.
    let state = Arc::new(hub::WaiterState::default());
    let _cancel = CancelWaitOnDrop(state.clone());

    let name = name.to_string();
    let wait_state = state.clone();
    let res = tokio::task::spawn_blocking(move || {
        // `hub::default()` may probe the wire; casting here matches the sync path's `BadType`.
        let binder = hub::default()?
            .wait_for_service_cancellable(&name, wait_state)
            .map_err(|end| {
                end.log(&name);
                StatusCode::NameNotFound
            })?;
        FromIBinder::try_from(binder)
    })
    .await;
    res.unwrap_or_else(|e| Err(join_error_status(e)))
}

/// Cancels the wait on drop: a `spawn_blocking` task ignores `abort()`, so its state tells it.
struct CancelWaitOnDrop(Arc<hub::WaiterState>);

impl Drop for CancelWaitOnDrop {
    fn drop(&mut self) {
        self.0.cancel();
    }
}

/// Resolve `name` if it is registered **now**, without waiting — the awaitable
/// counterpart of [`crate::hub::check_interface`].
///
/// [`StatusCode::NameNotFound`] when the name is not registered. Unlike
/// [`get_interface_async`] this will not start an unregistered lazy service,
/// except on Android 15 r6+, where `check_*` is carried by `getService` —
/// see the `hub` module doc.
///
/// Being a single round trip, this follows [`get_interface_async`] and runs
/// **inline** when it is called while handling a transaction, so the kernel can
/// apply its deadlock prevention to the nested call.
/// [`wait_for_interface_async`] deliberately does not — an unbounded wait has
/// nothing to gain from staying on that thread and everything to lose.
pub async fn check_interface_async<T: FromIBinder + ?Sized + 'static>(
    name: &str,
) -> Result<Strong<T>, StatusCode> {
    fn lookup<T: FromIBinder + ?Sized + 'static>(name: &str) -> Result<Strong<T>, StatusCode> {
        hub::check_interface::<T>(name)
    }

    if crate::is_handling_transaction() {
        // See comment in the BinderAsyncPool impl.
        return lookup::<T>(name);
    }

    let name = name.to_string();
    tokio::task::spawn_blocking(move || lookup::<T>(&name))
        .await
        .unwrap_or_else(|e| Err(join_error_status(e)))
}

/// A future that completes when a binder dies — one `select!` arm instead of a
/// [`DeathRecipient`] implementation.
///
/// Obtained from [`death_signal`]. Completes with `()`; there is nothing to
/// report beyond the death itself, and the binder it watches is the one that
/// was passed in.
///
/// It is fused: once it has completed, polling it again returns
/// [`Poll::Ready`] immediately instead of panicking. That matters because the
/// future is [`Unpin`], so a `select!` arm written as `_ = &mut died` can be
/// polled after the death has already been reported.
///
/// Fused means *always ready*, not *ignored*: after the death, a `select!` arm
/// holding `&mut signal` is ready on every iteration, so a loop that keeps
/// selecting on it runs that arm over and over and makes no other progress.
/// Leave the loop from that arm, or disable it once it has fired — `select!`
/// takes a per-arm precondition for exactly this:
/// `_ = &mut died, if !dead => { dead = true; … }`.
///
/// Dropping it before the death unregisters the notification.
///
/// Dropping it never panics. The kernel `unlink_to_death` panics on a
/// poisoned recipients lock, and that poisoning is reachable without a bug:
/// `ProxyHandle::link_to_death` holds the write guard across
/// `talk_with_driver`, which panics if the driver under-consumes. The unlink
/// therefore runs under `catch_unwind` (as `bridge::unlink_all` does); a
/// caught panic is logged and leaves the death link registered.
pub struct DeathSignal {
    rx: tokio::sync::oneshot::Receiver<()>,
    /// Strong: the link keeps only a `Weak`, so without this the notification would never fire.
    recipient: Arc<SignalRecipient>,
    /// Strong: the proxy owns the subscription; if the caller drops it, a weak ref pends forever.
    binder: SIBinder,
    /// `false` for an already-dead binder, which never got a link to undo.
    linked: bool,
    /// Set once `rx` has yielded: a `oneshot` receiver panics if polled again (see "fused").
    fired: bool,
}

/// [`DeathRecipient`] that does nothing but wake a [`DeathSignal`].
struct SignalRecipient(Mutex<Option<tokio::sync::oneshot::Sender<()>>>);

impl SignalRecipient {
    fn fire(&self) {
        // `send` consumes the sender: only the first firing sends, a spurious second is dropped.
        if let Some(tx) = self.0.lock().unwrap_or_else(|e| e.into_inner()).take() {
            let _ = tx.send(());
        }
    }
}

impl DeathRecipient for SignalRecipient {
    fn binder_died(&self, _who: &WIBinder) {
        // Binder or RPC session thread; a `oneshot` send takes no binder-stack lock.
        self.fire();
    }
}

impl Future for DeathSignal {
    type Output = ();

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
        let this = self.get_mut();
        if this.fired {
            return Poll::Ready(());
        }
        // `Err(RecvError)` can't occur while we hold the recipient; resolving beats hanging.
        match Pin::new(&mut this.rx).poll(cx) {
            Poll::Ready(_) => {
                this.fired = true;
                Poll::Ready(())
            }
            Poll::Pending => Poll::Pending,
        }
    }
}

impl Drop for DeathSignal {
    fn drop(&mut self) {
        if !self.linked {
            return;
        }
        // Unlink may panic (see `DeathSignal` doc); a panic out of `Drop` mid-unwind aborts.
        let unlink = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
            self.binder.unlink_to_death_arc(&self.recipient)
        }));
        match unlink {
            Ok(Ok(())) => {}
            // Expected once the obituary has been delivered.
            Ok(Err(err)) => log::debug!("DeathSignal: unlink_to_death returned {err:?}"),
            Err(_) => {
                log::error!("DeathSignal: unlink_to_death panicked; a death link stays registered")
            }
        }
    }
}

/// Watch `binder` and complete when it dies.
///
/// ```ignore
/// let svc = rsbinder::get_interface_async::<dyn IFoo<Tokio>>("foo").await?;
/// let died = rsbinder::death_signal(&svc.as_binder())?;
/// tokio::select! {
///     r = svc.work() => r?,
///     _ = died => return Err(/* the service is gone */),
/// }
/// ```
///
/// An **already dead** binder is not an error: the answer to "wait until it
/// dies" is available immediately, so the returned future is already complete.
///
/// # Errors
///
/// - [`StatusCode::InvalidOperation`] for a local binder (`Binder<T>` does not
///   support death notification — there is no remote process to lose).
/// - [`StatusCode::InvalidOperation`] for an RPC proxy whose session nothing
///   serves (see *Requirements*).
/// - Whatever registering the notification reports otherwise.
///
/// # Requirements
///
/// - **Kernel binder:** the obituary is delivered by a thread reading the
///   driver, so the process needs a thread pool
///   ([`crate::ProcessState::start_thread_pool`]) or a thread inside a
///   transaction. A pure client with neither never observes the death, and this
///   future stays pending — the same condition as
///   [`crate::SIBinder::link_to_death_arc`].
/// - **RPC:** death is the session's connection dropping, so the session has to
///   be served (`spawn_serve`/`serve_blocking`, or a client with incoming
///   connections); otherwise the link is refused with
///   [`StatusCode::InvalidOperation`].
pub fn death_signal(binder: &SIBinder) -> Result<DeathSignal, StatusCode> {
    let (tx, rx) = tokio::sync::oneshot::channel();
    let recipient = Arc::new(SignalRecipient(Mutex::new(Some(tx))));

    match binder.link_to_death_arc(&recipient) {
        Ok(()) => Ok(DeathSignal {
            rx,
            recipient,
            binder: binder.clone(),
            linked: true,
            fired: false,
        }),
        // Already dead (both stacks answer `DeadObject` after the obituary): completed signal.
        Err(StatusCode::DeadObject) => {
            recipient.fire();
            Ok(DeathSignal {
                rx,
                recipient,
                binder: binder.clone(),
                linked: false,
                fired: false,
            })
        }
        Err(err) => Err(err),
    }
}

/// Use the Tokio `spawn_blocking` pool with AIDL.
pub enum Tokio {}

impl BinderAsyncPool for Tokio {
    fn spawn<'a, F1, F2, Fut, A, B, E>(spawn_me: F1, after_spawn: F2) -> BoxFuture<'a, Result<B, E>>
    where
        F1: FnOnce() -> A,
        F2: FnOnce(A) -> Fut,
        Fut: Future<Output = Result<B, E>>,
        F1: Send + 'static,
        F2: Send + 'a,
        Fut: Send + 'a,
        A: Send + 'static,
        B: Send + 'a,
        E: From<crate::StatusCode>,
    {
        // Inline: kernel deadlock prevention and RPC's thread-local `DRIVING` pin need this thread.
        if crate::is_handling_transaction() {
            // Blocking is harmless: `block_on` runs only this one task.
            let result = spawn_me();
            Box::pin(after_spawn(result))
        } else {
            let handle = tokio::task::spawn_blocking(spawn_me);
            Box::pin(async move {
                match handle.await {
                    Ok(res) => after_spawn(res).await,
                    Err(e) => Err(join_error_status(e).into()),
                }
            })
        }
    }
}

/// Wrapper around a Tokio [`Handle`], for providing a runtime to a binder
/// server.
///
/// An async server implementation is driven by [`BinderAsyncRuntime::block_on`]
/// on whatever thread delivered the transaction — a kernel binder looper, or an
/// RPC session thread. Those threads are outside the runtime, which is the case
/// `Handle::block_on` is built for.
///
/// The same `block_on` is also reached from a *synchronous* handle to a local
/// async service: `Bn*::new_async_binder` returns `Strong<dyn IFoo>`, whose
/// generated wrapper implements each sync method as `block_on(inner.method())`.
/// Calling that handle from a task on a multi-threaded runtime would otherwise
/// panic with "Cannot start a runtime from within a runtime", so this impl
/// releases the worker's core with [`tokio::task::block_in_place`] first.
///
/// Two independent things then decide what the call does, and reading them as
/// one list is how the failures get misattributed. **Where the call is made
/// from** decides whether it runs at all. **Which runtime the handle points at**
/// decides whether the future can finish. Either answer holds whatever the other
/// one is.
///
/// # Panics
///
/// Where the call is made from, whatever the handle points at. Tokio refuses a
/// `Handle::block_on` on a thread that is already executing inside a runtime,
/// and [`block_in_place`] — which steps back out of one — exists only for the
/// multi-threaded runtime. That is the whole rule:
///
/// | The calling thread | Result |
/// |---|---|
/// | Not executing inside a runtime | runs |
/// | Inside a **multi-threaded** runtime | runs — releasing the core first when the caller is a worker holding one, inline otherwise |
/// | Inside a multi-threaded runtime, within a [`LocalSet`] | **panic** — "can call blocking only when running on the multi-threaded runtime"; `block_in_place` is refused there |
/// | Inside a **current-thread** runtime | **panic** — "Cannot start a runtime from within a runtime"; there is no way back out |
///
/// Reading a position off that rule is the part worth doing carefully, because
/// "inside a runtime" is not the same as "on a runtime's thread":
///
/// - A `spawn_blocking` pool thread is **outside** the runtime it belongs to,
///   for either flavor. Those calls run.
/// - A `Runtime::block_on` body is **inside** — so a current-thread runtime's
///   `Runtime::block_on` body is the last row, not the first.
/// - **A handler this adapter is already driving is inside**, for as long as
///   `block_on` is driving it, and inside *the handle's* runtime whatever thread
///   it started on. So a service whose `TokioRuntime` holds a current-thread
///   handle cannot reach a second local service through a sync handle from
///   within a handler — that nested call is the last row. Reach it with
///   [`Strong::into_async`] instead, or drive the service from a
///   multi-threaded handle.
///
/// [`block_in_place`]: tokio::task::block_in_place
///
/// # Stalls
///
/// Which runtime the handle points at. This decides completion whatever the
/// caller.
///
/// `Handle::block_on` polls the future on the **calling** thread — not on a
/// worker — for either flavor, and that is all it does. Anything the handler
/// needs that runtime's *scheduler* to advance — a timer, IO readiness, or a
/// [`tokio::spawn`]ed task — needs that runtime to be running:
///
/// - A **multi-threaded** runtime always is: its workers drive the timer and IO
///   drivers and run spawned tasks.
/// - A **current-thread** runtime does so only while some thread sits inside
///   `Runtime::block_on` on it, and `Handle::block_on` is not that. With nobody
///   parked there, a handler that awaits any of those never completes and the
///   call never returns. A handler that needs none of them — ready on the first
///   poll, or suspending only on `yield_now` — still completes.
///
/// [`spawn_blocking`](tokio::task::spawn_blocking) is deliberately not in that
/// set: the blocking pool runs independently of the scheduler, so awaiting one
/// completes here too. That is why an *outbound* call through
/// [`BinderAsyncPool`] — which is `spawn_blocking` —
/// works from a handler this adapter is driving on a current-thread handle,
/// even though a `tokio::spawn`ed task in the same place would not.
///
/// # Cost
///
/// A call from a **worker** thread releases that worker's core for the duration
/// and offers it to the `spawn_blocking` pool. From every other position nothing
/// is released and the call runs inline. The pool spawns a thread on demand, so
/// the core is only left unclaimed when the pool is at its
/// `max_blocking_threads` cap with every thread busy; then the call still
/// completes, but a runtime with a *single* worker runs nothing else until it
/// returns. A second worker, or headroom under the cap, removes that.
///
/// So on a path that calls a local async service repeatedly, convert the handle
/// once with [`Strong::into_async`] and await it instead — that route dispatches
/// to the async service directly and never enters `block_on`.
///
/// [`Handle`]: tokio::runtime::Handle
/// [`LocalSet`]: tokio::task::LocalSet
/// [`Strong::into_async`]: crate::Strong::into_async
pub struct TokioRuntime<R>(pub R);

impl BinderAsyncRuntime for TokioRuntime<tokio::runtime::Handle> {
    fn block_on<F: Future>(&self, future: F) -> F::Output {
        match tokio::runtime::Handle::try_current() {
            // Multi-thread: `block_in_place` frees a worker's core; inline on its blocking threads.
            Ok(handle) if handle.runtime_flavor() == tokio::runtime::RuntimeFlavor::MultiThread => {
                tokio::task::block_in_place(|| self.0.block_on(future))
            }
            // No runtime, or current-thread: panics only inside it; its blocking pool may block.
            _ => self.0.block_on(future),
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::binder::{IBinder, Stability};
    use std::mem::ManuallyDrop;
    use std::sync::atomic::{AtomicUsize, Ordering};

    /// Records what [`death_signal`] does; `unlink_panics` models a poisoned `recipients` lock.
    struct FakeRemote {
        link: Option<StatusCode>,
        unlink_panics: bool,
        unlinked: AtomicUsize,
        strong: AtomicUsize,
        recipient: Mutex<Option<std::sync::Weak<dyn DeathRecipient>>>,
    }

    impl FakeRemote {
        fn new(link: Option<StatusCode>, unlink_panics: bool) -> Arc<Self> {
            Arc::new(Self {
                link,
                unlink_panics,
                unlinked: AtomicUsize::new(0),
                strong: AtomicUsize::new(0),
                recipient: Mutex::new(None),
            })
        }

        /// Deliver an obituary as a worker thread does: via the stored recipient, only a `Weak`.
        fn fire_obituary(&self, who: &WIBinder) {
            let recipient = self
                .recipient
                .lock()
                .unwrap_or_else(|e| e.into_inner())
                .clone();
            let recipient = recipient.expect("no recipient registered").upgrade();
            recipient
                .expect("the signal dropped its recipient")
                .binder_died(who);
        }

        fn unlink_count(&self) -> usize {
            self.unlinked.load(Ordering::SeqCst)
        }
    }

    impl IBinder for FakeRemote {
        fn link_to_death(
            &self,
            recipient: std::sync::Weak<dyn DeathRecipient>,
        ) -> crate::Result<()> {
            if let Some(err) = self.link {
                return Err(err);
            }
            *self.recipient.lock().unwrap_or_else(|e| e.into_inner()) = Some(recipient);
            Ok(())
        }
        fn unlink_to_death(
            &self,
            recipient: std::sync::Weak<dyn DeathRecipient>,
        ) -> crate::Result<()> {
            assert!(!self.unlink_panics, "injected unlink panic");
            // Match by `Weak::ptr_eq` like the real proxy: a wrong-recipient unlink must not count.
            let mut slot = self.recipient.lock().unwrap_or_else(|e| e.into_inner());
            if !slot
                .as_ref()
                .is_some_and(|stored| std::sync::Weak::ptr_eq(stored, &recipient))
            {
                return Err(StatusCode::NameNotFound);
            }
            *slot = None;
            self.unlinked.fetch_add(1, Ordering::SeqCst);
            Ok(())
        }
        fn ping_binder(&self) -> crate::Result<()> {
            Ok(())
        }
        fn as_any(&self) -> &dyn std::any::Any {
            self
        }
        fn as_transactable(&self) -> Option<&dyn crate::Transactable> {
            None
        }
        fn descriptor(&self) -> &str {
            "rsbinder.test.IFakeRemote"
        }
        fn is_remote(&self) -> bool {
            true
        }
        fn stability(&self) -> Stability {
            Stability::default()
        }
        fn inc_strong(&self, _: &SIBinder) -> crate::Result<()> {
            self.strong.fetch_add(1, Ordering::SeqCst);
            Ok(())
        }
        fn attempt_inc_strong(&self) -> bool {
            true
        }
        fn dec_strong(&self, _: Option<ManuallyDrop<SIBinder>>) -> crate::Result<()> {
            // Like a destroyed proxy's subscription, the last strong drop clears the recipient.
            if self.strong.fetch_sub(1, Ordering::SeqCst) == 1 {
                *self.recipient.lock().unwrap_or_else(|e| e.into_inner()) = None;
            }
            Ok(())
        }
        fn inc_weak(&self, _: &WIBinder) -> crate::Result<()> {
            Ok(())
        }
        fn dec_weak(&self) -> crate::Result<()> {
            Ok(())
        }
    }

    /// Poll once without a runtime; a real one would only add a way for the test to hang.
    fn poll_once(signal: &mut DeathSignal) -> Poll<()> {
        let mut cx = Context::from_waker(std::task::Waker::noop());
        Pin::new(signal).poll(&mut cx)
    }

    #[test]
    fn completes_when_the_obituary_arrives() {
        let fake = FakeRemote::new(None, false);
        let binder = SIBinder::new(fake.clone()).unwrap();
        let mut signal = death_signal(&binder).unwrap();

        assert_eq!(poll_once(&mut signal), Poll::Pending);
        fake.fire_obituary(&SIBinder::downgrade(&binder));
        assert_eq!(poll_once(&mut signal), Poll::Ready(()));
    }

    #[test]
    fn an_already_dead_binder_is_complete_at_once() {
        // `DeadObject` from the link is not an error: the binder already died.
        let fake = FakeRemote::new(Some(StatusCode::DeadObject), false);
        let binder = SIBinder::new(fake.clone()).unwrap();
        let mut signal = death_signal(&binder).unwrap();

        assert_eq!(poll_once(&mut signal), Poll::Ready(()));
        drop(signal);
        // Nothing was linked, so nothing is unlinked.
        assert_eq!(fake.unlink_count(), 0);
    }

    #[test]
    fn a_local_binder_has_no_death_notification() {
        let fake = FakeRemote::new(Some(StatusCode::InvalidOperation), false);
        let binder = SIBinder::new(fake).unwrap();
        assert_eq!(
            death_signal(&binder).err(),
            Some(StatusCode::InvalidOperation)
        );
    }

    #[test]
    fn dropping_the_signal_unlinks_once() {
        let fake = FakeRemote::new(None, false);
        let binder = SIBinder::new(fake.clone()).unwrap();

        let signal = death_signal(&binder).unwrap();
        assert_eq!(fake.unlink_count(), 0);
        drop(signal);
        assert_eq!(fake.unlink_count(), 1);
    }

    #[test]
    fn a_panicking_unlink_stays_inside_drop() {
        // A poisoned-lock unlink panics, and a panic leaving `Drop` mid-unwind aborts.
        let fake = FakeRemote::new(None, true);
        let binder = SIBinder::new(fake).unwrap();
        let signal = death_signal(&binder).unwrap();

        let escaped = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| drop(signal)));
        assert!(escaped.is_ok(), "the injected unlink panic escaped Drop");
    }

    #[test]
    fn the_signal_survives_the_binder_handle_going_away() {
        // Obituary after the handle drop: a weak signal ref would let the fake drop its recipient.
        let fake = FakeRemote::new(None, false);
        let (mut signal, who) = {
            let binder = SIBinder::new(fake.clone()).unwrap();
            (death_signal(&binder).unwrap(), SIBinder::downgrade(&binder))
        };

        assert_eq!(poll_once(&mut signal), Poll::Pending);
        fake.fire_obituary(&who);
        assert_eq!(poll_once(&mut signal), Poll::Ready(()));
    }

    /// AC-11-1.6 — usable as a `select!` arm from a spawned task.
    #[test]
    fn death_signal_is_send_and_static() {
        fn assert_send_static<T: Send + 'static>() {}
        assert_send_static::<DeathSignal>();
    }
}