saddle-boundary 0.3.37

//! Application-owned standard unary connections. Only tonic's private executor
//! can register drivers; neither callbacks nor arbitrary futures are public API.
use saddle_admission::{AdmissionError, ExactStored, RpcStagePermit, StandardRpcPoolStorage,
    StandardRpcPoolPhysicalOwner, StandardRpcRequestWake, StorageDemand, StoragePermit};
use std::{alloc::Layout, future::Future, pin::Pin, sync::{Arc, Mutex,
    atomic::{AtomicBool, AtomicUsize, Ordering}}, task::{Context, Poll, Wake, Waker}, time::Instant};
type Driver = Pin<Box<dyn Future<Output = ()> + Send>>;
// Locked hyper 1.9.0: ConnTask + ClientTask + tonic worker, plus one
// exclusive stream's optional Pipe and SendWhen (proto/h2/client.rs).
const TASKS_PER_CONNECTION: usize = 5;

/// Startup-frozen alias and authority template. Each index belongs to one pool
/// (one application); actual zone resolution participates in the connection key.
#[doc(hidden)]
pub struct FrozenConnectionTarget { alias: String, template: String }
impl FrozenConnectionTarget {
    pub fn new(alias: &str, template: &str) -> Result<Self, AdmissionError> {
        if alias.is_empty() { return Err(AdmissionError::InvalidConfiguration); }
        crate::ProfuseContractAuthorityTemplate::new(template.to_owned())
            .map_err(|_| AdmissionError::InvalidConfiguration)?;
        Ok(Self { alias: alias.to_owned(), template: template.to_owned() })
    }
}
struct Target { alias: ExactStored<String>, template: ExactStored<String> }
#[derive(Clone, Copy, PartialEq, Eq)]
enum Phase { Vacant, Connecting, Leased, Idle, Retiring }
struct Task { future: Option<Driver>, polling: bool, persistent: bool }
impl Task { fn empty() -> Self { Self { future: None, polling: false, persistent: false } } }
struct Slot {
    generation: u64, connection_generation: u64, target: usize, phase: Phase,
    authority: Option<ExactStored<String>>, channel: Option<tonic::transport::Channel>,
    tasks: [Task; TASKS_PER_CONNECTION], failure: Option<AdmissionError>,
    stage: Option<RpcStagePermit>, request_wake: Option<StandardRpcRequestWake>,
}
impl Slot {
    fn vacant(generation: u64) -> Self { Self { generation, connection_generation: generation,
        target: 0, phase: Phase::Vacant, authority: None, channel: None,
        tasks: std::array::from_fn(|_| Task::empty()), failure: None, stage: None, request_wake: None } }
}
struct State {
    slots: ExactStored<Box<[Option<Slot>]>>, targets: ExactStored<Box<[Option<Target>]>>,
    accepting: bool, started: bool, stopping: bool, joined: bool, root_wake: Option<Waker>,
    task_bill: Option<StoragePermit>,
}
#[repr(C)]
struct SharedPrefix { state: Mutex<State>, handles: AtomicUsize, closed: AtomicBool }
#[repr(C)]
struct Shared { state: Mutex<State>, handles: AtomicUsize, closed: AtomicBool,
    arc_bill: StoragePermit, storage: StandardRpcPoolStorage }
impl std::fmt::Debug for Shared {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("StandardRpcPoolOwner").field("closed", &self.closed.load(Ordering::Acquire)).finish()
    }
}
impl StandardRpcPoolPhysicalOwner for Shared {
    fn closed(&self) -> bool { self.closed.load(Ordering::Acquire) }
    fn destroy_last(self: Arc<Self>) {
        let Self { state, handles: _, closed: _, arc_bill, storage } =
            Arc::into_inner(self).unwrap_or_else(|| std::process::abort());
        // Actual Arc block deallocation precedes its bill. Arrays and nested
        // owned strings also destroy their real blocks before their invoices.
        drop(state); drop(arc_bill); drop(storage);
    }
}
impl Shared {
    fn wake_root(&self) {
        let wake = self.state.lock().unwrap_or_else(|p| p.into_inner()).root_wake.take();
        if let Some(wake) = wake { wake.wake(); }
    }
    fn request_stop(&self) {
        { let mut state = self.state.lock().unwrap_or_else(|p| p.into_inner());
          state.accepting = false; state.stopping = true; }
        self.wake_root();
    }
}
impl Wake for Shared {
    fn wake(self: Arc<Self>) { self.wake_root(); release(self); }
    fn wake_by_ref(self: &Arc<Self>) { self.wake_root(); }
}
fn release(shared: Arc<Shared>) {
    let reclaim = shared.storage.reclaimer();
    drop(shared); reclaim.reclaim();
}
fn copy_text(storage: &StandardRpcPoolStorage, text: &str) -> Result<ExactStored<String>, AdmissionError> {
    let layout = Layout::array::<u8>(text.len()).map_err(|_| AdmissionError::SizeOverflow)?;
    let permit = storage.try_reserve(StorageDemand::separate(&[(layout, 1)])?)?;
    Ok(permit.allocate_exact(layout, || text.to_owned()))
}
/// Concrete assembly handle; leases are exclusive and intentionally not Clone.
#[doc(hidden)]
pub struct StandardConnectionPool { shared: Option<Arc<Shared>> }
impl Clone for StandardConnectionPool {
    fn clone(&self) -> Self {
        let shared = self.shared.as_ref().unwrap();
        shared.handles.fetch_add(1, Ordering::Relaxed);
        Self { shared: Some(shared.clone()) }
    }
}
impl Drop for StandardConnectionPool {
    fn drop(&mut self) {
        if let Some(shared) = self.shared.take() {
            if shared.handles.fetch_sub(1, Ordering::AcqRel) == 1 {
                let mut state = shared.state.lock().unwrap_or_else(|p| p.into_inner());
                state.accepting = false;
                if !state.started { shared.closed.store(true, Ordering::Release); }
                drop(state);
                shared.request_stop();
            }
            release(shared);
        }
    }
}
impl StandardConnectionPool {
    pub fn new(root: StoragePermit, capacity: usize, targets: Vec<FrozenConnectionTarget>) -> Result<Self, AdmissionError> {
        let storage = root.into_standard_rpc_pool_storage(capacity)?;
        if targets.is_empty() { return Err(AdmissionError::InvalidConfiguration); }
        for (i, target) in targets.iter().enumerate() {
            if targets[..i].iter().any(|v| v.alias == target.alias) {
                return Err(AdmissionError::InvalidConfiguration);
            }
        }
        let layout = Layout::array::<Option<Slot>>(capacity).map_err(|_| AdmissionError::SizeOverflow)?;
        let mut slots = storage.try_reserve(StorageDemand::separate(&[(layout, 1)])?)?
            .allocate_exact_none_slice::<Slot>(capacity);
        for slot in slots.get_mut() { *slot = Some(Slot::vacant(0)); }
        let layout = Layout::array::<Option<Target>>(targets.len()).map_err(|_| AdmissionError::SizeOverflow)?;
        let mut frozen = storage.try_reserve(StorageDemand::separate(&[(layout, 1)])?)?
            .allocate_exact_none_slice::<Target>(targets.len());
        for (target, input) in frozen.get_mut().iter_mut().zip(targets) {
            *target = Some(Target { alias: copy_text(&storage, &input.alias)?, template: copy_text(&storage, &input.template)? });
        }
        let demand = StorageDemand::embedded(Layout::new::<SharedPrefix>(), Layout::new::<StandardRpcPoolStorage>(),
            &[(Layout::new::<[usize; 2]>(), 1)])?;
        debug_assert_eq!(demand.bytes(), std::mem::size_of::<Shared>() + 2 * std::mem::size_of::<usize>());
        let arc_bill = storage.try_reserve(demand)?;
        let shared = Arc::new(Shared { state: Mutex::new(State { slots, targets: frozen, accepting: false,
            started: false, stopping: false, joined: false, root_wake: None, task_bill: None }),
            handles: AtomicUsize::new(1), closed: AtomicBool::new(false), arc_bill, storage });
        if let Err(error) = shared.storage.install(shared.clone()) {
            // A competing startup owner may win installation. Free this Arc
            // block before refunding its embedded invoice on rollback too.
            shared.destroy_last();
            return Err(error);
        }
        Ok(Self { shared: Some(shared) })
    }
    pub fn target_index(&self, alias: &str) -> Option<usize> {
        self.shared.as_ref().unwrap().state.lock().unwrap_or_else(|p| p.into_inner())
            .targets.get().iter().position(|t| t.as_ref().is_some_and(|t| t.alias.get() == alias))
    }
    /// No user endpoint/config enters the cache. The actual authority must
    /// match the frozen alias template with a valid, bounded zone substitution.
    pub fn try_acquire_authority(&self, target: usize, authority: &str) -> Result<ConnectionLease, AdmissionError> {
        let shared = self.shared.as_ref().unwrap();
        let mut state = shared.state.lock().unwrap_or_else(|p| p.into_inner());
        let frozen = state.targets.get().get(target).and_then(Option::as_ref)
            .ok_or(AdmissionError::InvalidConfiguration)?;
        if !matches_template(frozen.template.get(), authority) { return Err(AdmissionError::InvalidConfiguration); }
        if !state.accepting { return Err(AdmissionError::AccountClosed); }
        let matching = state.slots.get().iter().position(|s| s.as_ref().is_some_and(|s|
            s.phase == Phase::Idle && s.target == target && s.authority.as_ref().is_some_and(|a| a.get() == authority)));
        let vacant = state.slots.get().iter().position(|s| s.as_ref().is_some_and(|s| s.phase == Phase::Vacant));
        let index = match matching.or(vacant) {
            Some(index) => index,
            None => {
                // Idle eviction is retired by the root, physically, before
                // any replacement admission. No unbounded wait queue.
                if let Some(slot) = state.slots.get_mut().iter_mut().flatten().find(|s| s.phase == Phase::Idle) {
                    slot.phase = Phase::Retiring;
                }
                drop(state); shared.wake_root(); return Err(AdmissionError::CapacityRejected);
            }
        };
        let slot = state.slots.get_mut()[index].as_mut().unwrap();
        let generation = slot.generation.checked_add(1).ok_or(AdmissionError::SizeOverflow)?;
        if slot.phase == Phase::Vacant {
            slot.authority = Some(copy_text(&shared.storage, authority)?);
            slot.connection_generation = generation; slot.target = target; slot.phase = Phase::Connecting;
        } else { slot.phase = Phase::Leased; }
        slot.generation = generation;
        Ok(ConnectionLease { shared: Some(shared.clone()), index, generation })
    }
    pub fn try_acquire(&self, target: usize) -> Result<ConnectionLease, AdmissionError> {
        // Non-templated targets need no transient string allocation.
        let shared = self.shared.as_ref().unwrap();
        let state = shared.state.lock().unwrap_or_else(|p| p.into_inner());
        let frozen = state.targets.get().get(target).and_then(Option::as_ref)
            .ok_or(AdmissionError::InvalidConfiguration)?;
        let template = frozen.template.get().as_str();
        // Avoid holding the mutex across the actual acquisition: the frozen
        // storage lifetime is the pool's, but no borrowed lock guard escapes.
        let (ptr_target, templated) = (target, template.contains("{zone}"));
        if templated { return Err(AdmissionError::InvalidConfiguration); }
        // A startup-frozen source copy stays process billed during this call.
        let authority = copy_text(&shared.storage, template)?;
        drop(state);
        self.try_acquire_authority(ptr_target, authority.get())
    }
    pub fn stop_admission(&self) {
        let shared = self.shared.as_ref().unwrap();
        shared.state.lock().unwrap_or_else(|p| p.into_inner()).accepting = false;
    }
    pub fn start(&self) -> Result<StandardDriverService, AdmissionError> {
        if !cfg!(all(target_os = "linux", target_arch = "x86_64", target_pointer_width = "64")) {
            return Err(AdmissionError::InvalidConfiguration);
        }
        let shared = self.shared.as_ref().unwrap();
        if !shared.storage.startup_allowed() { return Err(AdmissionError::InvalidConfiguration); }
        let runtime = tokio::runtime::Handle::try_current().map_err(|_| AdmissionError::InvalidConfiguration)?;
        let mut state = shared.state.lock().unwrap_or_else(|p| p.into_inner());
        if state.started || state.stopping { return Err(AdmissionError::InvalidConfiguration); }
        let layout = saddle_runtime::request_task::storage::task_cell(
            saddle_runtime::request_task::storage::TaskBackendLayouts {
                header: Layout::from_size_align(32, 8).unwrap(), scheduler: Layout::new::<usize>(),
                task_id: Layout::new::<u64>(), trailer: Layout::from_size_align(48, 8).unwrap(), cell_alignment: 128,
                joinset_entry: Layout::from_size_align(56, 8).unwrap(), joinset_shared: Layout::from_size_align(72, 8).unwrap(),
            }, Layout::new::<DriverRoot>(), Layout::new::<Result<(), tokio::task::JoinError>>())
            .map_err(|_| AdmissionError::SizeOverflow)?;
        state.task_bill = Some(shared.storage.try_reserve(StorageDemand::separate(&[(layout, 1)])?)?);
        let worker = runtime.spawn(DriverRoot { shared: shared.clone() });
        state.started = true; state.accepting = true;
        Ok(StandardDriverService { shared: Some(shared.clone()), worker: Some(worker) })
    }
    #[cfg(test)]
    pub async fn wait_retired(&self) {
        tokio::time::timeout(std::time::Duration::from_secs(5), std::future::poll_fn(|cx| {
            let shared = self.shared.as_ref().unwrap();
            let pending = shared.state.lock().unwrap_or_else(|p| p.into_inner()).slots.get().iter()
                .any(|s| s.as_ref().is_none_or(|s| s.phase == Phase::Retiring));
            if pending { shared.wake_root(); cx.waker().wake_by_ref(); Poll::Pending } else { Poll::Ready(()) }
        })).await.expect("root retires before bounded deadline");
    }
    #[cfg(test)]
    pub fn test_idle_count(&self) -> usize { self.shared.as_ref().unwrap().state.lock().unwrap().slots.get().iter()
        .flatten().filter(|s| s.phase == Phase::Idle).count() }
}
fn matches_template(template: &str, authority: &str) -> bool {
    match template.split_once("{zone}") {
        None => template == authority,
        Some((prefix, suffix)) => authority.strip_prefix(prefix).and_then(|v| v.strip_suffix(suffix))
            .is_some_and(|zone| !zone.is_empty() && zone.len() <= crate::ingress::MAX_ID_BYTES
                && zone.bytes().all(|b| b.is_ascii_alphanumeric() || b == b'-' || b == b'_')),
    }
}
/// Linear generation lease; dropping it always retires the connection.
#[doc(hidden)]
pub struct ConnectionLease { shared: Option<Arc<Shared>>, index: usize, generation: u64 }
impl ConnectionLease {
    pub fn generation(&self) -> u64 { self.generation }
    pub fn attach_stage(&mut self, stage: RpcStagePermit) -> Result<(), AdmissionError> {
        let request_wake = StandardRpcRequestWake::prepare(&stage)?;
        let shared = self.shared.as_ref().unwrap();
        let mut state = shared.state.lock().unwrap_or_else(|p| p.into_inner());
        let slot = state.slots.get_mut()[self.index].as_mut().ok_or(AdmissionError::AccountClosed)?;
        if slot.generation != self.generation || slot.stage.is_some() || !matches!(slot.phase, Phase::Connecting | Phase::Leased) {
            return Err(AdmissionError::AccountClosed);
        }
        slot.stage = Some(stage); slot.request_wake = Some(request_wake); Ok(())
    }
    pub(crate) fn executor(&self) -> StandardExecutor {
        let shared = self.shared.as_ref().unwrap();
        let generation = shared.state.lock().unwrap_or_else(|p| p.into_inner()).slots.get()[self.index]
            .as_ref().filter(|slot| slot.generation == self.generation)
            .map_or(self.generation, |slot| slot.connection_generation);
        StandardExecutor { shared: Some(shared.clone()), index: self.index, generation }
    }
    pub fn failure(&self) -> Option<AdmissionError> {
        self.shared.as_ref().unwrap().state.lock().unwrap_or_else(|p| p.into_inner()).slots.get()[self.index]
            .as_ref().filter(|s| s.generation == self.generation && s.phase != Phase::Vacant)
            .map_or(Some(AdmissionError::AccountClosed), |s| s.failure)
    }
    fn retain_transport_owner(&self) -> Result<RpcStagePermit, AdmissionError> {
        let state = self.shared.as_ref().unwrap().state.lock().unwrap_or_else(|p| p.into_inner());
        state.slots.get()[self.index].as_ref().filter(|s| s.generation == self.generation)
            .and_then(|s| s.stage.as_ref()).ok_or(AdmissionError::AccountClosed)?.retain_physical_owner()
    }
    // This private generic helper is reached only by the concrete tonic
    // connect/ready/unary methods below, never by an application callback.
    async fn transport<F, T>(&self, future: F) -> Result<T, AdmissionError>
    where F: Future<Output = T> + Send, T: Send {
        let (audit, request_wake, permit) = {
            let state = self.shared.as_ref().unwrap().state.lock().unwrap_or_else(|p| p.into_inner());
            let slot = state.slots.get()[self.index].as_ref().filter(|s| s.generation == self.generation)
                .ok_or(AdmissionError::AccountClosed)?;
            let stage = slot.stage.as_ref().ok_or(AdmissionError::AccountClosed)?;
            (stage.dependency_audit(), slot.request_wake.as_ref().ok_or(AdmissionError::AccountClosed)?.clone(),
                stage.try_storage(StorageDemand::separate(&[])?)?)
        };
        let mut result = None;
        let private = async { result = Some(future.await); };
        fn future_layout<F>(_: &F) -> Layout { Layout::new::<F>() }
        let layout = future_layout(&private);
        let backing = permit.try_reserve(StorageDemand::separate(&[(layout, 1)])?)?;
        let heap = backing.allocate_exact(layout, || Some(Box::pin(private)));
        let mut transport = TransportPoll { heap, audit, request_wake };
        let outcome = std::future::poll_fn(|cx| {
            if let Some(error) = self.failure() { return Poll::Ready(Err(error)); }
            let waker = transport.request_wake.waker(cx.waker());
            let mut bridge = Context::from_waker(&waker);
            let future = transport.heap.get_mut().as_mut().unwrap();
            match transport.audit.poll_driver(future.as_mut(), &mut bridge) {
                Poll::Ready(()) => Poll::Ready(Ok(())), Poll::Pending => Poll::Pending,
            }
        }).await;
        drop(transport);
        outcome?;
        Ok(result.expect("transport completion owns its output"))
    }
    pub(crate) async fn connect(&self, deadline: Instant) -> Result<tonic::transport::Channel, TransportFailure> {
        let private = async {
            let shared = self.shared.as_ref().unwrap();
            let endpoint = {
                let state = shared.state.lock().unwrap_or_else(|p| p.into_inner());
                let slot = state.slots.get()[self.index].as_ref().filter(|s| s.generation == self.generation)
                    .ok_or(TransportFailure::Resource(AdmissionError::AccountClosed))?;
                if let Some(channel) = &slot.channel { return Ok(channel.clone()); }
                tonic::transport::Endpoint::from_shared(slot.authority.as_ref().unwrap().get().clone())
                    .map_err(TransportFailure::Connection)?
            };
            // No persistent timeout or first-request connect_timeout: every
            // wait inherits this invocation's unchanged absolute deadline.
            let endpoint = endpoint.concurrency_limit(1).buffer_size(1).executor(self.executor());
            let connect = endpoint.connect();
            let channel = tokio::time::timeout_at(deadline.into(), connect).await
                .map_err(|_| TransportFailure::Deadline)?.map_err(TransportFailure::Connection)?;
            let mut state = shared.state.lock().unwrap_or_else(|p| p.into_inner());
            let slot = state.slots.get_mut()[self.index].as_mut().filter(|s|
                s.generation == self.generation)
                .ok_or(TransportFailure::Resource(AdmissionError::AccountClosed))?;
            if let Some(error) = slot.failure { return Err(TransportFailure::Resource(error)); }
            if slot.phase != Phase::Connecting { return Err(TransportFailure::Resource(AdmissionError::AccountClosed)); }
            slot.channel = Some(channel.clone()); slot.phase = Phase::Leased;
            Ok(channel)
        };
        self.transport(private).await.map_err(TransportFailure::Resource)?
    }
    /// Fixed raw transport only. Request construction, schema conversion and
    /// error recording remain in the enclosing original-request attempt.
    pub(crate) async fn unary_transport(
        &self, request: tonic::Request<saddle_admission::ReadOnlyInput<Vec<u8>>>,
        memory: saddle_admission::RequestMemory, path: &'static str, deadline: Instant,
    ) -> Result<saddle_admission::ReadOnlyInput<Vec<u8>>, TransportFailure> {
        self.unary_transport_recorded(request, memory, path, deadline, &mut false).await
    }
    pub(crate) async fn unary_transport_recorded(
        &self, request: tonic::Request<saddle_admission::ReadOnlyInput<Vec<u8>>>,
        memory: saddle_admission::RequestMemory, path: &'static str, deadline: Instant, sent: &mut bool,
    ) -> Result<saddle_admission::ReadOnlyInput<Vec<u8>>, TransportFailure> {
        let channel = self.connect(deadline).await?;
        let wire_stage = self.retain_transport_owner().map_err(TransportFailure::Resource)?;
        let codec_stage = self.retain_transport_owner().map_err(TransportFailure::Resource)?;
        let private = async move {
            let mut client = tonic::client::Grpc::new(channel)
                .max_encoding_message_size(crate::MAX_UNARY_PAYLOAD_BYTES)
                .max_decoding_message_size(crate::MAX_UNARY_PAYLOAD_BYTES);
            tokio::time::timeout_at(deadline.into(), client.ready()).await
                .map_err(|_| TransportFailure::Deadline)?.map_err(TransportFailure::Ready)?;
            let remaining = deadline.checked_duration_since(Instant::now()).filter(|d| !d.is_zero())
                .ok_or(TransportFailure::Deadline)?;
            let mut request = request.map(|wire| crate::standard_codec::LeasedWire { wire, stage: wire_stage });
            request.set_timeout(remaining);
            *sent = true;
            let response = tokio::time::timeout_at(deadline.into(), client.unary(request,
                tonic::codegen::http::uri::PathAndQuery::from_static(path),
                crate::standard_codec::LeasedCodec { memory, stage: codec_stage })).await
                .map_err(|_| TransportFailure::Deadline)?.map_err(TransportFailure::Status)?.into_inner();
            response.map_err(TransportFailure::Resource)
        };
        self.transport(private).await.map_err(TransportFailure::Resource)?
    }
    /// Wait for private stream tasks and request references to be physically released.
    /// Uses the original call deadline and request audit wake bridge.
    pub(crate) async fn drain_response(&self, deadline: Instant) -> Result<(), TransportFailure> {
        tokio::time::timeout_at(deadline.into(), std::future::poll_fn(|cx| {
            let shared = self.shared.as_ref().unwrap();
            let registration = {
                let state = shared.state.lock().unwrap_or_else(|p| p.into_inner());
                state.slots.get()[self.index].as_ref()
                    .filter(|slot| slot.generation == self.generation)
                    .and_then(|slot| slot.request_wake.clone())
            };
            if let Some(registration) = registration {
                // Original RawWaker callbacks execute outside the pool lock.
                // Release our temporary owner before testing physical refs.
                drop(registration.waker(cx.waker()));
                drop(registration);
            }
            let state = shared.state.lock().unwrap_or_else(|p| p.into_inner());
            let Some(slot) = state.slots.get()[self.index].as_ref() else {
                return Poll::Ready(Err(TransportFailure::Resource(AdmissionError::AccountClosed)));
            };
            if slot.generation != self.generation || slot.phase != Phase::Leased {
                return Poll::Ready(Err(TransportFailure::Resource(AdmissionError::AccountClosed)));
            }
            if let Some(error) = slot.failure { return Poll::Ready(Err(TransportFailure::Resource(error))); }
            let (Some(wake), Some(stage)) = (slot.request_wake.as_ref(), slot.stage.as_ref()) else {
                return Poll::Ready(Err(TransportFailure::Resource(AdmissionError::AccountClosed)));
            };
            // Registration precedes the completion observation, so a driver
            // completion cannot be lost between the check and this poll.
            if slot.tasks.iter().all(|t| t.persistent || (t.future.is_none() && !t.polling))
                && stage.standard_transport_references_clear() && wake.references_clear() {
                Poll::Ready(Ok(()))
            } else { Poll::Pending }
        })).await.map_err(|_| TransportFailure::Deadline)?
    }
    pub fn finish(mut self, response_complete: bool) { self.finish_inner(response_complete); }
    fn finish_inner(&mut self, response_complete: bool) {
        let Some(shared) = self.shared.take() else { return; };
        let (wake, stage, idle) = {
            let mut state = shared.state.lock().unwrap_or_else(|p| p.into_inner());
            let accepting = state.accepting;
            if let Some(slot) = state.slots.get_mut()[self.index].as_mut().filter(|s| s.generation == self.generation) {
                let clear = slot.stage.as_ref().is_some_and(RpcStagePermit::standard_transport_references_clear)
                    && slot.request_wake.as_ref().is_some_and(StandardRpcRequestWake::references_clear);
                let idle = response_complete && accepting && clear && slot.failure.is_none() && slot.channel.is_some()
                    && slot.tasks.iter().any(|t| t.persistent && (t.future.is_some() || t.polling))
                    && slot.tasks.iter().all(|t| t.persistent || (t.future.is_none() && !t.polling));
                if idle { slot.phase = Phase::Idle; (slot.request_wake.take(), slot.stage.take(), true) }
                else { slot.phase = Phase::Retiring; (None, None, false) }
            } else { (None, None, false) }
        };
        if let Some(wake) = wake { wake.close(); drop(wake); }
        drop(stage);
        if !idle { shared.wake_root(); }
        release(shared);
    }
    #[cfg(test)]
    pub fn test_driver_waker(&self) -> Waker { Waker::from(self.shared.as_ref().unwrap().clone()) }
    #[cfg(test)]
    pub fn test_driver_count(&self) -> usize { self.shared.as_ref().unwrap().state.lock().unwrap().slots.get()[self.index]
        .as_ref().unwrap().tasks.iter().filter(|t| t.future.is_some() || t.polling).count() }
}
impl Drop for ConnectionLease { fn drop(&mut self) { self.finish_inner(false); } }
struct TransportPoll<F: Future<Output = ()> + Send> {
    heap: ExactStored<Option<Pin<Box<F>>>>,
    audit: saddle_admission::RpcDependencyAudit,
    request_wake: StandardRpcRequestWake,
}
impl<F: Future<Output = ()> + Send> Drop for TransportPoll<F> {
    fn drop(&mut self) {
        if let Some(future) = self.heap.get_mut().take() { self.audit.drop_driver(future); }
        // ExactStored drops an empty physical pointer before its invoice.
    }
}
#[derive(Debug)]
pub(crate) enum TransportFailure {
    Resource(AdmissionError), Connection(tonic::transport::Error),
    Ready(tonic::transport::Error), Status(tonic::Status), Deadline,
}
impl std::fmt::Display for TransportFailure {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self { Self::Resource(e) => e.fmt(f), Self::Connection(e) => e.fmt(f),
            Self::Ready(e) => e.fmt(f), Self::Status(e) => e.fmt(f), Self::Deadline => f.write_str("original request deadline elapsed") }
    }
}
impl std::error::Error for TransportFailure {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match self { Self::Resource(e) => Some(e), Self::Connection(e) => Some(e),
            Self::Ready(e) => Some(e), Self::Status(e) => Some(e), Self::Deadline => None }
    }
}
/// Private tonic adapter. It carries process ownership and a connection epoch,
/// never an original request account. Registration allocates no framework task.
pub(crate) struct StandardExecutor { shared: Option<Arc<Shared>>, index: usize, generation: u64 }
impl Clone for StandardExecutor {
    fn clone(&self) -> Self { Self { shared: self.shared.clone(), index: self.index, generation: self.generation } }
}
impl Drop for StandardExecutor { fn drop(&mut self) { if let Some(shared) = self.shared.take() { release(shared); } } }
impl hyper::rt::Executor<Driver> for StandardExecutor {
    fn execute(&self, future: Driver) {
        let shared = self.shared.as_ref().unwrap();
        let mut future = Some(future);
        {
            let mut state = shared.state.lock().unwrap_or_else(|p| p.into_inner());
            if let Some(slot) = state.slots.get_mut()[self.index].as_mut().filter(|s|
                s.connection_generation == self.generation && matches!(s.phase, Phase::Connecting | Phase::Leased | Phase::Idle)) {
                if let Some(task) = slot.tasks.iter_mut().find(|t| t.future.is_none() && !t.polling) {
                    task.persistent = slot.phase == Phase::Connecting;
                    task.future = future.take();
                } else { slot.failure = Some(AdmissionError::CapacityRejected); slot.phase = Phase::Retiring; }
            }
        }
        drop(future); shared.wake_root();
    }
}
/// One root task, created before requests and joined by the application owner.
#[doc(hidden)]
pub struct StandardDriverService { shared: Option<Arc<Shared>>, worker: Option<tokio::task::JoinHandle<()>> }
impl StandardDriverService {
    pub async fn shutdown(&mut self) -> Result<(), tokio::task::JoinError> {
        let Some(worker) = self.worker.as_mut() else { return Ok(()); };
        let shared = self.shared.as_ref().unwrap();
        shared.request_stop();
        worker.await?;
        drop(self.worker.take());
        let mut state = shared.state.lock().unwrap_or_else(|p| p.into_inner());
        state.joined = true;
        // Tokio cell and all root wakers are gone at the matching join.
        drop(state.task_bill.take());
        shared.closed.store(true, Ordering::Release);
        Ok(())
    }
}
impl Drop for StandardDriverService {
    fn drop(&mut self) {
        if let Some(worker) = self.worker.take() {
            if let Some(shared) = self.shared.as_ref() { shared.request_stop(); }
            worker.abort(); // No join claim or refund on an abandoned service.
        }
        if let Some(shared) = self.shared.take() { release(shared); }
    }
}
struct DriverRoot { shared: Arc<Shared> }
impl Future for DriverRoot {
    type Output = ();
    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
        let shared = &self.shared;
        let count = {
            let mut state = shared.state.lock().unwrap_or_else(|p| p.into_inner());
            let previous = state.root_wake.replace(cx.waker().clone());
            drop(previous);
            state.slots.get().len()
        };
        for index in 0..count {
            let retiring = {
                let mut state = shared.state.lock().unwrap_or_else(|p| p.into_inner());
                let stopping = state.stopping;
                let slot = state.slots.get_mut()[index].as_mut().unwrap();
                if stopping || slot.failure.is_some() { slot.phase = Phase::Retiring; }
                if slot.phase == Phase::Retiring { state.slots.get_mut()[index].take() } else { None }
            };
            if let Some(mut slot) = retiring {
                if let Some(wake) = &slot.request_wake { wake.clone().notify(); wake.close(); }
                drop(slot.channel.take());
                for task in &mut slot.tasks { drop(task.future.take()); }
                // Dependency bytes/wakers have been physically released first.
                drop(slot.request_wake.take()); drop(slot.stage.take()); drop(slot.authority.take());
                shared.state.lock().unwrap_or_else(|p| p.into_inner()).slots.get_mut()[index] = Some(Slot::vacant(slot.generation));
                continue;
            }
            for task_index in 0..TASKS_PER_CONNECTION {
                let task = {
                    let mut state = shared.state.lock().unwrap_or_else(|p| p.into_inner());
                    let task = &mut state.slots.get_mut()[index].as_mut().unwrap().tasks[task_index];
                    let future = task.future.take(); task.polling = future.is_some(); future
                };
                if let Some(mut future) = task {
                    let waker = Waker::from(shared.clone());
                    let result = future.as_mut().poll(&mut Context::from_waker(&waker));
                    let mut state = shared.state.lock().unwrap_or_else(|p| p.into_inner());
                    let slot = state.slots.get_mut()[index].as_mut().unwrap();
                    if result.is_pending() { slot.tasks[task_index].polling = false; slot.tasks[task_index].future = Some(future); }
                    else {
                        if slot.tasks[task_index].persistent {
                            slot.failure = Some(AdmissionError::AccountClosed); slot.phase = Phase::Retiring;
                        }
                        drop(state); drop(future);
                        let notify = {
                            let mut state = shared.state.lock().unwrap_or_else(|p| p.into_inner());
                            let slot = state.slots.get_mut()[index].as_mut().unwrap();
                            slot.tasks[task_index].polling = false;
                            slot.request_wake.clone()
                        };
                        if let Some(notify) = notify { notify.notify(); }
                        shared.wake_root();
                    }
                }
            }
        }
        let mut state = shared.state.lock().unwrap_or_else(|p| p.into_inner());
        if state.stopping && state.slots.get().iter().flatten().all(|s| s.phase == Phase::Vacant) {
            let wake = state.root_wake.take(); drop(state); drop(wake); Poll::Ready(())
        } else {
            if state.stopping { cx.waker().wake_by_ref(); }
            Poll::Pending
        }
    }
}