saddle-runtime 0.3.24

//! AR uses the real admitted owner/JoinSet. No database execution is implied.
use super::*;
use crate::request_task::reserved::{
    ReservedBorrowedFuture, ReservedRequestFailure, ReservedTaskContext,
};
use saddle_core::request_context::ContextLabel;
use std::{
    alloc::Layout,
    sync::atomic::{AtomicBool, Ordering},
};

#[derive(Clone, Copy, Debug)]
enum Case {
    Ready,
    PrePoll,
    Pending,
    Withdraw,
}
struct Input {
    case: Case,
    entered: Arc<AtomicBool>,
    observer: Observer,
}
type Owner = ReservedDispatchOwner<Input>;
fn body(
    owner: &mut Owner,
    mut context: ReservedTaskContext,
) -> impl Future<Output = Result<u32, ReservedRequestFailure>> + Send + '_ {
    async move {
        owner.1.entered.store(true, Ordering::SeqCst);
        assert_eq!(
            owner.2.as_ref().unwrap().receipt.decision(),
            ProfuseGwCapacityDecision::Accepted
        );
        if matches!(owner.1.case, Case::Ready) {
            // Same consumer ordering as Service: validated identity exists here,
            // not at reservation, and the original event is consumed only here.
            let (call, _) = owner
                .1
                .observer
                .start_external_call_checked(
                    "app",
                    "module",
                    "service",
                    "route",
                    Some("safe-trace"),
                )
                .unwrap();
            let event = EventContext::new(
                saddle_observability::RequestIdentity::new("request").unwrap(),
                saddle_observability::RouteIdentity::new("route").unwrap(),
                1,
            )
            .unwrap();
            let admission = owner.2.take().unwrap();
            assert!(owner.2.take().is_none());
            context.publish(saddle_core::RequestIdentityGroup::from_validated(
                call.context(), "request", "route", 1,
                saddle_core::ContextFact::NotEstablished,
            ).unwrap()).unwrap();
            let (dispatch, submission) = owner.0.take().unwrap().bind_reserved_observation(
                owner.1.observer.clone(),
                call.context().clone(),
                event,
                admission,
                &context.view(),
                None,
            );
            assert_unavailable(submission);
            owner.0 = Some(dispatch);
            Ok(41)
        } else {
            std::future::pending().await
        }
    }
}

fn assert_unavailable(submission: saddle_observability::AdmissionCapacitySubmission) {
    use saddle_observability::DiagnosticSubmission::OutputUnavailable;
    assert_eq!(submission.cpu, OutputUnavailable);
    assert_eq!(submission.memory, OutputUnavailable);
    assert_eq!(submission.database, OutputUnavailable);
    assert_eq!(submission.profuse_contract, OutputUnavailable);
}
fn factory<'a>(
    owner: &'a mut Owner,
    context: ReservedTaskContext,
) -> ReservedBorrowedFuture<'a, u32> {
    Box::pin(body(owner, context))
}
fn returned_layout<I, R>(_: impl FnOnce(I) -> R) -> Layout {
    Layout::new::<R>()
}
fn body_layout() -> Layout {
    returned_layout(|(o, c): (&'static mut Owner, ReservedTaskContext)| body(o, c))
}

#[test]
fn admission_event_survives_owner_lifecycle() {
    tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .unwrap()
        .block_on(async {
            for case in [Case::Ready, Case::PrePoll, Case::Pending, Case::Withdraw] {
                let shared = Arc::new(SharedRuntimeProcess {
                    process: Mutex::new(Some(ProfuseGwRuntimeProcess::new(
                        crate::request_task::reserved::tests::process(),
                    ))),
                });
                let lease = ProfuseGwProcessLease {
                    shared: shared.clone(),
                };
                let startup = lease.take_database_startup_half().unwrap();
                let entered = Arc::new(AtomicBool::new(false));
                let input = Input {
                    case,
                    entered: entered.clone(),
                    observer: Observer::with_writer(
                        saddle_observability::ObserverConfig::default(),
                        std::io::sink(),
                    )
                    .unwrap(),
                };
                let ReservedDispatchOutcome::Ready {
                    root,
                    future,
                    mut ticket,
                } = lease.try_reserved_dispatch(
                    ContextLabel::checked("app").unwrap(),
                    None,
                    body_layout(),
                    &[],
                    input,
                    factory,
                )
                else {
                    panic!("real permit fit")
                };
                let held = root
                    .view(saddle_core::request_context::RequestViewPhase::Reading)
                    .unwrap();
                let before_join = held.test_available_storage();
                let joined = if matches!(case, Case::Withdraw) {
                    ticket.withdraw(future).ok().unwrap()
                } else {
                    let mut tasks = tokio::task::JoinSet::new();
                    let handle = tasks.spawn(future);
                    ticket.bind(handle.id()).unwrap();
                    if matches!(case, Case::PrePoll) {
                        handle.abort();
                    }
                    if matches!(case, Case::Pending) {
                        while !entered.load(Ordering::SeqCst) {
                            tokio::task::yield_now().await;
                        }
                        handle.abort();
                    }
                    if matches!(case, Case::Ready) {
                        while !handle.is_finished() {
                            tokio::task::yield_now().await;
                        }
                    }
                    assert!(matches!(
                        lease.try_admit(),
                        ProfuseGwCoordinatorAdmissionOutcome::CapacityRejected(_)
                    ));
                    ticket
                        .complete(tasks.join_next().await.unwrap())
                        .ok()
                        .unwrap()
                };
                drop(root);
                let mut recovered = joined.recover(Default::default()).ok().unwrap();
                assert!(
                    held.test_available_storage() > before_join,
                    "task storage released only after matching recovery"
                );
                if matches!(case, Case::Ready) {
                    assert_eq!(recovered.result.take().unwrap().ok(), Some(41));
                    assert!(recovered.owner.2.is_none());
                    assert!(recovered.owner.0.as_ref().unwrap().observation.is_some());
                } else {
                    assert_eq!(
                        recovered.owner.2.as_ref().unwrap().receipt.decision(),
                        ProfuseGwCapacityDecision::Accepted
                    );
                    assert!(recovered.owner.0.as_ref().unwrap().observation.is_none());
                    if matches!(case, Case::PrePoll | Case::Withdraw) {
                        assert!(!entered.load(Ordering::SeqCst));
                    }
                    // Terminal owner consumes the original event after the
                    // matching join, using the still-live original root guard.
                    assert_unavailable(held.record_admission(
                        recovered.owner.2.take().unwrap(),
                        &recovered.owner.1.observer,
                        None,
                        saddle_observability::AdmissionConstruction::Cancelled,
                    ));
                    assert!(recovered.owner.2.is_none());
                }
                finish_profusegw_without_database(recovered.owner.0.take().unwrap(), ())
                    .ok()
                    .unwrap();
                // Output unavailable was returned per dimension, not reported written.
                drop(recovered);
                drop(held);
                match lease.try_admit() {
                    ProfuseGwCoordinatorAdmissionOutcome::Ready(d, _) => d.cancel(),
                    _ => panic!("original profile retry"),
                }
                drop((startup, lease));
                shared
                    .process
                    .lock()
                    .unwrap()
                    .take()
                    .unwrap()
                    .finish()
                    .unwrap();
                println!("AR_LIFECYCLE {case:?} ORIGINAL_OWNER=RECOVERED ZERO=PASS");
            }
            println!(
                "AR_LAYOUT event={} owner={} body={} demand={}",
                Layout::new::<ProfuseGwAdmissionEvent>().size(),
                Layout::new::<Owner>().size(),
                body_layout().size(),
                crate::request_task::reserved::dispatch_storage_bytes_for::<Input, u32, _>(
                    &factory,
                    body_layout(),
                    &[]
                )
                .unwrap()
            );
        });
}

#[test]
fn admitted_failures_restore_event_input_and_uncalled_factory() {
    tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .unwrap()
        .block_on(async {
            for construction in [false, true] {
                let shared = Arc::new(SharedRuntimeProcess {
                    process: Mutex::new(Some(ProfuseGwRuntimeProcess::new(
                        crate::request_task::reserved::tests::process(),
                    ))),
                });
                let lease = ProfuseGwProcessLease {
                    shared: shared.clone(),
                };
                let startup = lease.take_database_startup_half().unwrap();
                let entered = Arc::new(AtomicBool::new(false));
                let input = Input {
                    case: Case::Ready,
                    entered: entered.clone(),
                    observer: Observer::with_writer(
                        saddle_observability::ObserverConfig::default(),
                        std::io::sink(),
                    )
                    .unwrap(),
                };
                let backing = [(Layout::from_size_align(1_000_000, 1).unwrap(), 1)];
                crate::request_task::reserved::FAIL_AR_CONSTRUCTION
                    .with(|flag| flag.set(construction));
                let outcome = lease.try_reserved_dispatch(
                    ContextLabel::checked("app").unwrap(),
                    None,
                    body_layout(),
                    if construction { &[] } else { &backing },
                    input,
                    factory,
                );
                let ReservedDispatchOutcome::Rejected {
                    input,
                    make,
                    reason: ReservedDispatchRejection::Admitted { admission, failure },
                } = outcome
                else {
                    panic!("admitted failure must return original event")
                };
                assert_eq!(
                    admission.receipt.decision(),
                    ProfuseGwCapacityDecision::Accepted
                );
                assert_eq!(admission.receipt.used(), 1);
                assert!(!entered.load(Ordering::SeqCst));
                assert!(Arc::ptr_eq(&input.entered, &entered));
                assert!(matches!(
                    (construction, failure),
                    (false, ReservedDispatchConstructionFailure::Storage(_))
                        | (true, ReservedDispatchConstructionFailure::Context(_))
                ));
                assert_unavailable(admission.record_unrooted(
                    &input.observer,
                    None,
                    &ContextLabel::checked("app").unwrap(),
                    saddle_core::RequestViewPhase::Admitted,
                    saddle_observability::AdmissionConstruction::Failed(if construction {
                        saddle_observability::AdmissionConstructionFailure::Context
                    } else {
                        saddle_observability::AdmissionConstructionFailure::Storage
                    }),
                ));
                // Original event is still caller-owned; retry is a new decision,
                // not an attempt to reuse that old receipt as admission authority.
                let ReservedDispatchOutcome::Ready {
                    root,
                    future,
                    ticket,
                } = lease.try_reserved_dispatch(
                    ContextLabel::checked("app").unwrap(),
                    None,
                    body_layout(),
                    &[],
                    input,
                    make,
                )
                else {
                    panic!("withdrawn resources and exact factory usable")
                };
                let mut recovered = ticket
                    .withdraw(future)
                    .ok()
                    .unwrap()
                    .recover(Default::default())
                    .ok()
                    .unwrap();
                assert_eq!(
                    recovered.owner.2.as_ref().unwrap().receipt.decision(),
                    ProfuseGwCapacityDecision::Accepted
                );
                recovered.owner.0.take().unwrap().cancel();
                drop((recovered, root, startup, lease));
                shared
                    .process
                    .lock()
                    .unwrap()
                    .take()
                    .unwrap()
                    .finish()
                    .unwrap();
                println!(
                    "AR_FAILURE construction={construction} INPUT_FACTORY_EVENT=RESTORED ZERO=PASS"
                );
            }
        });
}

type RefusalOwner = Option<ProfuseGwAdmissionEvent>;
fn refusal<'a>(
    owner: &'a mut RefusalOwner,
    _: ReservedTaskContext,
) -> ReservedBorrowedFuture<'a, u32> {
    Box::pin(async move {
        assert!(owner.is_some());
        Ok(41)
    })
}
#[test]
fn rejection_sixteen_slots_keep_original_event_until_last_reference() {
    tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap().block_on(async {
        let shared=Arc::new(SharedRuntimeProcess{process:Mutex::new(Some(ProfuseGwRuntimeProcess::new(crate::request_task::reserved::tests::process())))});
        let lease=ProfuseGwProcessLease{shared:shared.clone()};
        let startup=lease.take_database_startup_half().unwrap();
        let ProfuseGwCoordinatorAdmissionOutcome::Ready(active,accepted)=lease.try_admit() else {panic!("active")};
        let mut slots=Vec::new();
        for index in 0..17 {
            let ReservedDispatchOutcome::Rejected{reason:ReservedDispatchRejection::Capacity(event),..}=lease.try_reserved_dispatch(ContextLabel::checked("app").unwrap(),None,body_layout(),&[],Input{case:Case::Ready,entered:Arc::new(AtomicBool::new(false)),observer:Observer::with_writer(saddle_observability::ObserverConfig::default(),std::io::sink()).unwrap()},factory) else {panic!("original rejected event")};
            assert_eq!(event.receipt.decision(),ProfuseGwCapacityDecision::CapacityRejected);
            let slot=lease.try_reserved_rejection(ContextLabel::checked("app").unwrap(),None,Layout::new::<[u8;64]>(),&[],Some(event),refusal);
            if let Ok(slot)=slot {assert!(index<16);slots.push(slot);}
            else {
                let (original,make,error)=slot.err().unwrap();
                println!("AR_REJECTION actual_tasks={} refusal={error:?}",slots.len());
                assert!(slots.len()<=16 && !slots.is_empty());
                assert_eq!(original.as_ref().unwrap().receipt.decision(),ProfuseGwCapacityDecision::CapacityRejected);
                // Original byte reserve may run out before sixteen complete
                // tasks fit. Fill ONLY remaining slot identities with genuine
                // minimal profile reservations to isolate the slot-lifetime
                // oracle. These are not claimed as constructed tasks.
                let mut slot_only=Vec::new();
                {
                    let process=shared.process.lock().unwrap();
                    let profile=process.as_ref().unwrap().admission.verified_profile();
                    for _ in slots.len()..16 {
                        let minimal=saddle_admission::StorageDemand::embedded(Layout::new::<()>(),Layout::new::<()>(),&[]).unwrap();
                        slot_only.push(profile.try_rejection_storage(minimal).unwrap());
                    }
                }
                let (root,future,ticket)=slots.pop().unwrap();
                let held=root.view(saddle_core::request_context::RequestViewPhase::Reading).unwrap();
                let recovered=ticket.withdraw(future).ok().unwrap().recover(Default::default()).ok().unwrap();
                assert!(recovered.owner.is_some());drop((recovered,root));
                // Matching recovery does not refund the slot while a view survives.
                let (original,make,_)=lease.try_reserved_rejection(ContextLabel::checked("app").unwrap(),None,Layout::new::<[u8;64]>(),&[],original,make).err().expect("last view holds reservation");
                drop(held);
                slots.push(lease.try_reserved_rejection(ContextLabel::checked("app").unwrap(),None,Layout::new::<[u8;64]>(),&[],original,make).ok().expect("last reference refunds original slot"));
                drop(slot_only);
                break;
            }
        }
        for (root,future,ticket) in slots {
            let recovered=ticket.withdraw(future).ok().unwrap().recover(Default::default()).ok().unwrap();
            assert_eq!(recovered.owner.as_ref().unwrap().receipt.decision(),ProfuseGwCapacityDecision::CapacityRejected);
            drop((recovered,root));
        }
        active.cancel();drop((accepted,startup,lease));shared.process.lock().unwrap().take().unwrap().finish().unwrap();
        println!("AR_REJECTION SLOTS=16 EXCESS=RESTORED LAST_REFERENCE=REFUNDED IDENTITY=NOT_ESTABLISHED ZERO=PASS");
    });
}