saddle-boundary 0.3.29

use super::*;
use crate::*;
use http_body_util::{BodyExt, Full};
use prost::Message;
use saddle_admission::*;
use std::sync::atomic::{AtomicUsize, Ordering};
fn process() -> ProfuseGwLightweightProcessOwner {
    let pending = freeze_deployment_resource_budget(
        1, 32, 5000, 1, 1, 1_000_000, 1_000_000, 1_000_000, 1_000_000,
    )
    .unwrap();
    let (app, listener) = saddle_core::BootstrapRendezvousIssuer::issue()
        .freeze_application(saddle_core::GeneratedApplicationFreezeSource::new(
            "app",
            b"descriptor",
            &["route"],
        ))
        .unwrap();
    let listener = listener
        .freeze_listener(saddle_core::ListenerStartupFreezeSource::new(
            "app",
            "127.0.0.1:8000".parse().unwrap(),
            "127.0.0.1:9000".parse().unwrap(),
            Duration::from_secs(5),
        ))
        .ok()
        .unwrap();
    let (whole, receipt) = saddle_core::pair_bootstrap_rendezvous(app, listener)
        .ok()
        .unwrap();
    prepare_profusegw_lightweight_profile(
        bind_deployment_resource_budget_bootstrap(pending, whole, receipt)
            .ok()
            .unwrap(),
    )
    .ok()
    .unwrap()
}
fn execution(process: &ProfuseGwLightweightProcessOwner) -> ProfuseGwLightweightExecutionOwner {
    let mut ingress = process.verified_profile().try_ingress().unwrap();
    match process
        .verified_profile()
        .try_promote_ingress(&mut ingress, 0)
    {
        ProfuseGwLightweightObservedAdmissionOutcome::Ready(permit, _) => permit.into_execution(),
        _ => panic!("pre-RPC requests must coexist"),
    }
}
fn request() -> InvokeRequest {
    InvokeRequest::unary(
        "r",
        "c",
        InvocationTarget::new("unit", "function").unwrap(),
        (SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .unwrap()
            .as_millis()
            + 5000) as i64,
        CallerContext {
            trace_info: Some(proto::TraceInfo {
                trace_id: "trace".into(),
                rpc_id: "0.1".into(),
            }),
            ldc_info: Some(proto::LdcInfo {
                zone: "z".into(),
                idc: "i".into(),
                env: "test".into(),
            }),
        },
        vec![],
    )
    .unwrap()
}
fn owned_request(stage: &RpcStagePermit) -> ReadOnlyInput<InvokeRequest> {
    let original = request();
    stage.framework_input(|b| Ok(InvokeRequest {
        request_id: b.copy_text(&[&original.request_id])?,
        call_id: b.copy_text(&[&original.call_id])?,
        business_unit: b.copy_text(&[&original.business_unit])?,
        function: b.copy_text(&[&original.function])?,
        deadline_unix_ms: original.deadline_unix_ms,
        caller_context: Some(CallerContext {
            trace_info: Some(proto::TraceInfo {
                trace_id: b.copy_text(&["trace"])?, rpc_id: b.copy_text(&["0.1"])?,
            }),
            ldc_info: Some(proto::LdcInfo {
                zone: b.copy_text(&["z"])?, idc: b.copy_text(&["i"])?, env: b.copy_text(&["test"])?,
            }),
        }),
        payload: b.write_bytes(65537, |bytes| { bytes.fill(7); Ok(()) })?,
    })).unwrap()
}
#[test]
fn shared_last_reference_retains_stage_and_storage_after_call_close() {
    use hyper::rt::Executor;
    struct Dropped(Arc<AtomicUsize>);
    impl Future for Dropped {
        type Output = ();
        fn poll(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<()> { Poll::Pending }
    }
    impl Drop for Dropped {
        fn drop(&mut self) { self.0.fetch_add(1, Ordering::SeqCst); }
    }
    let process = process();
    let execution = execution(&process);
    let stage = execution.try_begin_rpc_stage().unwrap();
    let baseline = process.resource_snapshot().framework_charged;
    let escaped = Arc::new(std::sync::Mutex::new(None));
    let handoff = escaped.clone();
    let call = RpcCall::new(stage, |executor| {
        *handoff.lock().unwrap() = Some(executor.clone());
        std::future::ready(())
    }).unwrap();
    assert_eq!(process.rpc_stage_snapshot().operations, 1);
    drop(call);
    assert_eq!(process.rpc_stage_snapshot().operations, 1);
    assert!(process.resource_snapshot().framework_charged > baseline);
    let discarded = Arc::new(AtomicUsize::new(0));
    escaped.lock().unwrap().as_ref().unwrap()
        .execute(Box::pin(Dropped(discarded.clone())));
    assert_eq!(discarded.load(Ordering::SeqCst), 1);
    assert_eq!(process.rpc_stage_snapshot().operations, 1);
    drop(escaped.lock().unwrap().take());
    assert_eq!(process.rpc_stage_snapshot().operations, 1);
    let recovered = execution.try_begin_rpc_stage().unwrap();
    assert_eq!(process.rpc_stage_snapshot().operations, 1);
    assert_eq!(process.resource_snapshot().framework_charged, baseline);
    drop(recovered);
    assert_eq!(process.rpc_stage_snapshot().operations, 0);
}

#[test]
fn framework_shortage_prevents_routed_rpc_construction_before_peer_effect() {
    let process = process();
    let execution = execution(&process);
    let stage = execution.try_begin_rpc_stage().unwrap();
    let snapshot = process.resource_snapshot();
    let available = snapshot.framework_capacity - snapshot.framework_charged;
    let permit_bytes = std::mem::size_of::<StoragePermit>();
    let hold = process.try_process_storage(StorageDemand::separate(&[(
        Layout::array::<u8>(available - permit_bytes - 1).unwrap(), 1,
    )]).unwrap()).unwrap();
    let calls = AtomicUsize::new(0);
    let result = RpcCall::new(stage, |_| {
        calls.fetch_add(1, Ordering::SeqCst);
        std::future::ready(())
    });
    assert!(matches!(result, Err(AdmissionError::FrameworkReserveExceeded { .. })));
    assert_eq!(calls.load(Ordering::SeqCst), 0, "no connector or peer operation may start");
    drop(hold);
    assert_eq!(process.rpc_stage_snapshot().operations, 0);
}

#[test]
fn late_driver_waker_is_a_real_last_strong_owner() {
    use hyper::rt::Executor;
    struct CountWake(AtomicUsize);
    impl std::task::Wake for CountWake {
        fn wake(self: Arc<Self>) { self.0.fetch_add(1, Ordering::SeqCst); }
        fn wake_by_ref(self: &Arc<Self>) { self.0.fetch_add(1, Ordering::SeqCst); }
    }
    let process = process();
    let execution = execution(&process);
    let stage = execution.try_begin_rpc_stage().unwrap();
    let baseline = process.resource_snapshot().framework_charged;
    let escaped = Arc::new(std::sync::Mutex::new(None::<Waker>));
    let handoff = escaped.clone();
    let mut call = Box::pin(RpcCall::new(stage, |executor| async move {
        executor.execute(Box::pin(std::future::poll_fn(move |cx| {
            *handoff.lock().unwrap() = Some(cx.waker().clone());
            Poll::Pending
        })));
        std::future::pending::<()>().await;
    }).unwrap());
    let count = Arc::new(CountWake(AtomicUsize::new(0)));
    let caller_waker = Waker::from(count.clone());
    let mut cx = Context::from_waker(&caller_waker);
    assert!(call.as_mut().poll(&mut cx).is_pending());
    assert!(call.as_mut().poll(&mut cx).is_pending());
    assert!(escaped.lock().unwrap().is_some());
    drop(call);
    assert_eq!(process.rpc_stage_snapshot().operations, 1);
    assert!(process.resource_snapshot().framework_charged > baseline);
    let before_late_wake = count.0.load(Ordering::SeqCst);
    escaped.lock().unwrap().as_ref().unwrap().wake_by_ref();
    assert_eq!(count.0.load(Ordering::SeqCst), before_late_wake,
        "a closed RPC must not wake its former business task");
    drop(escaped.lock().unwrap().take());
    assert_eq!(process.rpc_stage_snapshot().operations, 1);
    let recovered = execution.try_begin_rpc_stage().unwrap();
    drop(recovered);
    assert_eq!(process.rpc_stage_snapshot().operations, 0);
    assert_eq!(process.resource_snapshot().framework_charged, baseline);
    // The formal owner creates no Weak reference. The stage's bytes are
    // reclaimed only after Arc::into_inner frees its last control block.
}
#[test]
fn h2_driver_waker_bridge_forwards_all_safe_wake_operations() {
    use hyper::rt::Executor;
    struct CountingWake(AtomicUsize, std::sync::Mutex<Option<std::sync::Weak<Shared>>>);
    impl CountingWake {
        fn observe(&self) {
            if let Some(shared) = self.1.lock().unwrap().as_ref().and_then(|weak| weak.upgrade()) {
                assert!(shared.state.try_lock().is_ok(), "callback ran under driver state lock");
            }
            self.0.fetch_add(1, Ordering::SeqCst);
        }
    }
    impl std::task::Wake for CountingWake {
        fn wake(self: Arc<Self>) { self.observe(); }
        fn wake_by_ref(self: &Arc<Self>) { self.observe(); }
    }
    let process = process();
    let execution = execution(&process);
    let stage = execution.try_begin_rpc_stage().unwrap();
    let count = Arc::new(CountingWake(AtomicUsize::new(0), std::sync::Mutex::new(None)));
    let caller_waker = Waker::from(count.clone());
    let mut cx = Context::from_waker(&caller_waker);
    let mut call = Box::pin(RpcCall::new(stage, |executor| async move {
        executor.execute(Box::pin(std::future::poll_fn(|cx| {
            let copy = cx.waker().clone();
            copy.wake_by_ref();
            copy.wake();
            Poll::Ready(())
        })));
        std::future::pending::<()>().await;
    }).unwrap());
    *count.1.lock().unwrap() = Some(Arc::downgrade(call.executor.0.as_ref().unwrap()));
    assert!(call.as_mut().poll(&mut cx).is_pending());
    assert!(call.as_mut().poll(&mut cx).is_pending());
    assert!(count.0.load(Ordering::SeqCst) >= 2);
    *count.1.lock().unwrap() = None;
    drop(call);
    assert_eq!(process.rpc_stage_snapshot().operations, 0);
}
#[test]
fn unpaid_caller_waker_allocation_returns_to_request_audit() {
    use hyper::rt::Executor;
    const CHILD: &str = "SADDLE_RPC_UNPAID_WAKER_CHILD";
    if std::env::var_os(CHILD).is_none() {
        use std::os::unix::process::ExitStatusExt;
        let status = std::process::Command::new(std::env::current_exe().unwrap())
            .arg("unpaid_caller_waker_allocation_returns_to_request_audit")
            .env(CHILD, "1")
            .status().unwrap();
        assert_eq!(status.signal(), Some(6), "unpaid callback must fail the original account");
        return;
    }
    struct UnpaidWake;
    impl std::task::Wake for UnpaidWake {
        fn wake(self: Arc<Self>) { std::hint::black_box(vec![0_u8; 32]); }
        fn wake_by_ref(self: &Arc<Self>) { std::hint::black_box(vec![0_u8; 32]); }
    }
    let process = process();
    let execution = execution(&process);
    let stage = execution.try_begin_rpc_stage().unwrap();
    let caller_waker = Waker::from(Arc::new(UnpaidWake));
    let mut cx = Context::from_waker(&caller_waker);
    let mut call = Box::pin(RpcCall::new(stage, |executor| async move {
        executor.execute(Box::pin(std::future::poll_fn(|cx| {
            cx.waker().wake_by_ref();
            Poll::Ready(())
        })));
        std::future::pending::<()>().await;
    }).unwrap());
    assert!(execution.poll_database_query(call.as_mut(), &mut cx).is_pending());
    assert!(execution.poll_database_query(call.as_mut(), &mut cx).is_pending());
    drop(call);
    execution.cancel_observed();
    std::process::exit(0);
}
#[tokio::test]
async fn real_h2_drivers_complete_or_cancel_before_rpc_credit_returns() {
    let process = Arc::new(process());
    let first = execution(&process);
    let second = execution(&process);
    let third = execution(&process);
    assert_eq!(process.rpc_stage_snapshot().operations, 0);
    for cancel in [false, true] {
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let endpoint =
            ProfuseContractEndpoint::new(format!("http://{}", listener.local_addr().unwrap()))
                .unwrap();
        let entered = Arc::new(tokio::sync::Notify::new());
        let release = Arc::new(tokio::sync::Notify::new());
        let count = Arc::new(AtomicUsize::new(0));
        let seen = count.clone();
        let arrived = entered.clone();
        let gate = release.clone();
        let peer = tokio::spawn(async move {
            let (stream, _) = listener.accept().await.unwrap();
            let service = hyper::service::service_fn(
                move |request: hyper::Request<hyper::body::Incoming>| {
                    let (arrived, gate, seen) = (arrived.clone(), gate.clone(), seen.clone());
                    async move {
                        let bytes = request.into_body().collect().await.unwrap().to_bytes();
                        assert!(bytes.len() >= 5);
                        let request = InvokeRequest::decode(&bytes[5..]).unwrap();
                        assert_eq!(request.payload.len(), 65537);
                        assert!(request.payload.iter().all(|byte| *byte == 7));
                        seen.fetch_add(1, Ordering::SeqCst);
                        arrived.notify_one();
                        gate.notified().await;
                        let response = proto::InvokeResponse {
                            request_id: "r".into(),
                            call_id: "c".into(),
                            outcome: Some(proto::invoke_response::Outcome::Completed(
                                proto::Completed { result: vec![] },
                            )),
                        }
                        .encode_to_vec();
                        let mut bytes = vec![0];
                        bytes.extend_from_slice(&(response.len() as u32).to_be_bytes());
                        bytes.extend(response);
                        let body =
                            Full::new(hyper::body::Bytes::from(bytes)).with_trailers(async {
                                let mut trailers = hyper::HeaderMap::new();
                                trailers.insert("grpc-status", "0".parse().unwrap());
                                Some(Ok::<_, std::convert::Infallible>(trailers))
                            });
                        Ok::<_, std::convert::Infallible>(
                            hyper::Response::builder()
                                .header("content-type", "application/grpc")
                                .body(body)
                                .unwrap(),
                        )
                    }
                },
            );
            hyper::server::conn::http2::Builder::new(hyper_util::rt::TokioExecutor::new())
                .serve_connection(hyper_util::rt::TokioIo::new(stream), service)
                .await
        });
        let stage = first.try_begin_rpc_stage().unwrap();
        let before = process.resource_snapshot().framework_charged;
        let baseline = process.resource_snapshot().charged;
        // Cancellation before any codec poll must release the same owned fields.
        let unpolled = owned_request(&stage);
        assert_eq!(process.resource_snapshot().charged - baseline, unpolled.storage_bytes());
        drop(unpolled);
        assert_eq!(process.resource_snapshot().charged, baseline);
        let owned = owned_request(&stage);
        assert!(owned.storage_bytes() >= 65537, "payload allocation must have its own bill");
        assert_eq!(process.resource_snapshot().charged - baseline, owned.storage_bytes());
        let mut call = Box::pin(
            RpcCall::new(stage, |executor| async move {
                let boundary = TonicBoundary::connect_until_executor(
                    endpoint,
                    request().deadline_unix_ms,
                    None,
                    Some(executor),
                )
                .await?;
                boundary.invoke_owned(owned).await
            })
            .unwrap(),
        );
        tokio::select! {
            result=&mut call=>panic!("peer must hold call: {result:?}"),
            ()=entered.notified()=>(),
            ()=tokio::time::sleep(Duration::from_secs(2))=>panic!("local executor failed to drive real HTTP/2"),
        }
        assert_eq!(count.load(Ordering::SeqCst), 1);
        // Encoder has consumed/destroyed native fields; the peer can remain
        // held without conservatively charging those already freed copies.
        assert_eq!(process.resource_snapshot().charged, baseline);
        assert_eq!(
            second.try_begin_rpc_stage().unwrap_err(),
            AdmissionError::RpcCreditsExhausted
        );
        assert!(process.resource_snapshot().framework_charged > before);
        if !cancel {
            release.notify_one();
            assert!(
                tokio::time::timeout(Duration::from_secs(2), &mut call)
                    .await
                    .unwrap()
                    .unwrap()
                    .is_ok()
            );
        }
        // The fixed formal H2 route destroys the client future first, then
        // every connection driver. Its Tokio registration clears stored
        // Wakers on socket destruction, leaving the executor anchor as the
        // sole strong owner before Arc::into_inner performs final teardown.
        call.as_mut().get_mut().close();
        drop(call);
        assert_eq!(process.rpc_stage_snapshot().operations, 0);
        assert_eq!(process.resource_snapshot().framework_charged, before);
        let successor = third.try_begin_rpc_stage().unwrap();
        drop(successor);
        // This is the real server connection, not an outer-client-future probe.
        tokio::time::timeout(Duration::from_secs(2), peer)
            .await
            .expect("driver socket remains alive")
            .unwrap()
            .ok();
        assert_eq!(count.load(Ordering::SeqCst), 1, "no replay");
    }
    // Executor bookkeeping has no fixed two-driver/task ceiling. These are
    // driver futures, not a second queue of application requests.
    let dropped = Arc::new(AtomicUsize::new(0));
    let polled = Arc::new(AtomicUsize::new(0));
    let premature = Arc::new(AtomicUsize::new(0));
    for scarcity in [false, true] {
        let (profile, drops, polls, early) = (
            process.clone(),
            dropped.clone(),
            polled.clone(),
            premature.clone(),
        );
        let count = if scarcity { 1 } else { 33 };
        let mut call = Box::pin(
            RpcCall::new(
                first.try_begin_rpc_stage().unwrap(),
                |executor| async move {
                    for _ in 0..count {
                        hyper::rt::Executor::execute(
                            &executor,
                            Box::pin(Witness {
                                process: profile.clone(),
                                dropped: drops.clone(),
                                polled: polls.clone(),
                                premature: early.clone(),
                            }) as Driver,
                        );
                    }
                    std::future::pending::<()>().await;
                },
            )
            .unwrap(),
        );
        let hold = if scarcity {
            let snapshot = process.resource_snapshot();
            let available = snapshot.framework_capacity - snapshot.framework_charged;
            Some(
                process
                    .try_process_storage(
                        StorageDemand::separate(&[(
                            Layout::array::<u8>(available - std::mem::size_of::<StoragePermit>())
                                .unwrap(),
                            1,
                        )])
                        .unwrap(),
                    )
                    .unwrap(),
            )
        } else {
            None
        };
        let before_poll = polled.load(Ordering::SeqCst);
        let mut cx = Context::from_waker(Waker::noop());
        let result = call.as_mut().poll(&mut cx);
        if scarcity {
            assert!(matches!(
                result,
                Poll::Ready(Err(AdmissionError::FrameworkReserveExceeded { .. }))
            ));
            assert_eq!(
                polled.load(Ordering::SeqCst),
                before_poll,
                "rejected driver must never run"
            );
        } else {
            assert!(result.is_pending());
            assert!(call.as_mut().poll(&mut cx).is_pending());
            assert_eq!(polled.load(Ordering::SeqCst) - before_poll, 33);
        }
        drop(call);
        drop(hold);
        assert_eq!(process.rpc_stage_snapshot().operations, 0);
    }
    assert_eq!(dropped.load(Ordering::SeqCst), 34);
    assert_eq!(
        premature.load(Ordering::SeqCst),
        0,
        "RPC credit refunded before physical driver destructor"
    );
    first.cancel();
    second.cancel();
    third.cancel();
    assert!(process.resource_snapshot().healthy);
    Arc::try_unwrap(process).ok().unwrap().finish().unwrap();
}

struct Witness {
    process: Arc<ProfuseGwLightweightProcessOwner>,
    dropped: Arc<AtomicUsize>,
    polled: Arc<AtomicUsize>,
    premature: Arc<AtomicUsize>,
}
impl Future for Witness {
    type Output = ();
    fn poll(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<()> {
        self.polled.fetch_add(1, Ordering::SeqCst);
        Poll::Pending
    }
}
impl Drop for Witness {
    fn drop(&mut self) {
        if self.process.rpc_stage_snapshot().operations != 1 {
            self.premature.fetch_add(1, Ordering::SeqCst);
        }
        self.dropped.fetch_add(1, Ordering::SeqCst);
    }
}