use super::*;
use saddle_observability::root_diagnostic::RootRequestEvent;
use saddle_runtime::profusegw::ProfuseGwIngressDeadline;
use saddle_runtime::request_task::reserved::{
ReservedBorrowedFuture, ReservedRequestFailure, ReservedTaskContext,
};
pub(super) struct RecognizedInput {
pub accepted: saddle_boundary::ingress::AcceptedIngress,
pub input_peak: usize,
}
pub(super) struct ReadState<B, C, D> {
pub factory: Option<consumer_storage::NormalFactory<B, C, D>>,
pub deadline: ProfuseGwIngressDeadline,
pub root: Option<saddle_runtime::request_task::reserved::ReservedRequestRoot>,
pub retained: reserved_entry::EntryOwner,
}
pub(super) type Owner<B, C, D> = (saddle_admission::ProfuseGwIngressOwner, ReadState<B, C, D>);
pub(super) fn factory<B: Send + Sync + 'static, C: Send + 'static, D: Send + Sync + 'static>(
owner: &mut Owner<B, C, D>,
task: ReservedTaskContext,
) -> ReservedBorrowedFuture<'_, ()> {
Box::pin(read(owner, task))
}
pub(super) fn body_layout<B: Send + Sync + 'static, C: Send + 'static, D: Send + Sync + 'static>()
-> std::alloc::Layout {
fn out<A, R>(_: impl FnOnce(A) -> R) -> std::alloc::Layout {
std::alloc::Layout::new::<R>()
}
out(|(owner, task): (&'static mut Owner<B, C, D>, ReservedTaskContext)| read(owner, task))
}
async fn read<B: Send + Sync + 'static, C: Send + 'static, D: Send + Sync + 'static>(
owner: &mut Owner<B, C, D>,
task: ReservedTaskContext,
) -> std::result::Result<(), ReservedRequestFailure> {
let state = &mut owner.1;
state.root = Some(task.retain_root());
state.retained.view = Some(task.view());
let factory = state.factory.as_mut().expect("single recognition owner");
let view = task.view();
let output = factory.diagnostic_handle.as_ref();
let io = |error: std::io::Error| {
let status = match error.kind() {
std::io::ErrorKind::PermissionDenied => 401,
std::io::ErrorKind::OutOfMemory => 503,
std::io::ErrorKind::TimedOut => 408,
std::io::ErrorKind::FileTooLarge => 413,
_ => 400,
};
(
status,
view.source_error(
&error,
output,
saddle_core::DiagnosticStage::RequestDecode,
RootRequestEvent::Ingress,
Default::default(),
),
)
};
let result=async {
let memory=owner.0.memory().map_err(|e|io(ingress_memory_error(e)))?;
let head=tokio::select! {
biased;
()=state.deadline.elapsed()=>return Err(io(std::io::ErrorKind::TimedOut.into())),
result=read_request_head_with_memory(&mut factory.socket,state.deadline.unix_ms(),factory.ingress_token.as_deref(),Some(&memory))=>result.map_err(&io)?,
};
state.deadline.tighten(head.deadline_unix_ms).map_err(|error| {
#[derive(Debug)]
struct DeadlineDescription(saddle_runtime::alpha1_ingress::AbsoluteDeadlineError);
impl std::fmt::Display for DeadlineDescription {
fn fmt(&self,f:&mut std::fmt::Formatter<'_>)->std::fmt::Result { write!(f,"{:?}",self.0) }
}
let status=if error==saddle_runtime::alpha1_ingress::AbsoluteDeadlineError::Expired {408} else {400};
let code=saddle_core::DiagnosticCode::new("service.ingress.deadline").unwrap();
(status,view.source_description(&DeadlineDescription(error),saddle_core::BoundedDiagnostic::capture(
saddle_core::DiagnosticCategory::ExpectedRejection,saddle_core::CaptureSite::FirstObserved,
saddle_core::BoundedDiagnosticCause::new(saddle_core::DiagnosticStage::RequestDecode,code)),code,output,RootRequestEvent::Ingress,Default::default()))
})?;
let request=tokio::select! {
biased;
()=state.deadline.elapsed()=>return Err(io(std::io::ErrorKind::TimedOut.into())),
result=read_request_body(&mut factory.socket,head)=>result.map_err(&io)?,
};
factory.adapter.validate_managed_transport(&request.method,&request.path,&request.content_type,&request.identity,&request.body)
.map_err(|error|(error.http_status(),view.source_error(&error,output,saddle_core::DiagnosticStage::RequestDecode,RootRequestEvent::Ingress,Default::default())))?;
let document=request.body.decode_input(&memory).map_err(|error| {
let status=if error==saddle_admission::AdmissionError::InvalidJson {400} else {503};
(status,view.source_error(&error,output,saddle_core::DiagnosticStage::RequestDecode,RootRequestEvent::Ingress,Default::default()))
})?;
let accepted=factory.adapter.accept_managed(&request.method,&request.path,&request.content_type,&request.identity,&request.body,document)
.map_err(|error| { (error.http_status(),view.source_error(&error,output,saddle_core::DiagnosticStage::RequestDecode,RootRequestEvent::Ingress,Default::default())) })?;
let input_peak = match factory.input_planner {
Some(plan) => plan(&accepted).map_err(|error| {
let status = if error == saddle_admission::AdmissionError::InvalidJson {400} else {503};
(status,view.source_error(&error,output,saddle_core::DiagnosticStage::RequestDecode,RootRequestEvent::Ingress,Default::default()))
})?,
None => 0,
};
Ok(RecognizedInput { accepted, input_peak })
}.await;
match result {
Ok(input) => {
factory.preparsed = Some(input);
Ok(())
}
Err((status, source)) => {
let bytes = super::terminal_status(status);
if state.deadline.is_elapsed() {
reserved_entry::deliver_now(
&factory.socket,
&bytes,
view,
output,
&mut state.retained.delivery,
);
} else {
reserved_entry::deliver(
&mut factory.socket,
&bytes,
state.deadline.instant(),
view,
output,
&mut state.retained.delivery,
)
.await;
}
Err(source)
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use saddle_runtime::profusegw::*;
use saddle_runtime::request_task::reserved::try_prepare_ingress_task;
#[test]
fn real_socket_recognition_retains_root_body_and_storage_only_account() {
let mut coordinator = std::pin::pin!(coordinate_profusegw_app_run(
reserved_entry::lifecycle_tests::budget(),
|process| std::future::ready(run_profusegw_owned_application(
process,
|lease| async move {
let startup = lease.take_database_startup_half().unwrap();
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let mut client = TcpStream::connect(listener.local_addr().unwrap())
.await
.unwrap();
let (socket, _) = listener.accept().await.unwrap();
let observer = saddle_observability::Observer::with_writer(
Default::default(),
std::io::sink(),
)
.unwrap();
let input = consumer_storage::NormalFactory {
socket,
input_planner: None,
preparsed: None,
adapter: saddle_boundary::ingress::ProfuseGwListenerAdapter::new("app")
.unwrap(),
deployment: Some(()),
business: Some(BusinessConfig::unit()),
ingress_token: None,
dispatch: Arc::new(()),
observer,
admission: None,
database: None,
diagnostic_handle: None,
};
let ingress = lease.try_ingress().unwrap();
let state = ReadState {
factory: Some(input),
deadline: ProfuseGwIngressDeadline::start(Duration::from_secs(5)).unwrap(),
root: None,
retained: reserved_entry::EntryOwner::new(),
};
let (root, future, mut ticket) = try_prepare_ingress_task(
saddle_core::ContextLabel::checked("app").unwrap(),
None,
body_layout::<(), (), ()>(),
&[],
(ingress, state),
factory::<(), (), ()>,
)
.ok()
.unwrap();
let mut tasks = tokio::task::JoinSet::new();
let abort = tasks.spawn(future);
ticket.bind(abort.id()).unwrap();
let body=br#"{"requestData":{"key":"value"},"profuseGwContext":{"userInfo":{"userId":"u"},"traceInfo":{"rpcId":"0"},"ldcInfo":{"zone":"z","idc":"i","env":"test"}},"target":{"app":"app","interfaceId":"route"}}"#;
let head = format!(
"POST /saddle/v1/ingress/profusegw/invoke HTTP/1.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nX-Request-Id: r\r\nX-Call-Id: c\r\n\r\n",
body.len()
);
client.write_all(head.as_bytes()).await.unwrap();
tokio::time::sleep(Duration::from_millis(5)).await;
assert!(lease.resource_snapshot().unwrap().charged >= MAX_HEAD_BYTES);
client.write_all(body).await.unwrap();
let joined = ticket
.complete(tasks.join_next().await.unwrap())
.ok()
.unwrap();
let recovered = joined.recover(Default::default()).ok().unwrap();
assert!(matches!(recovered.result, Some(Ok(()))));
let (ingress, mut state) = recovered.owner;
let input = state.factory.take().unwrap();
let parsed = input.preparsed.as_ref().unwrap();
assert_eq!(parsed.accepted.interface_id,"route");
let saddle_boundary::ingress::AcceptedRequestData::Managed(document)=&parsed.accepted.request_data else { panic!("managed formal input") };
assert_eq!(document.root().field("requestData").unwrap().field("key").unwrap().text(),Some("value"));
let input_bytes=document.storage_bytes();
assert!(state.root.is_some());
drop(ingress);
assert!(lease.resource_snapshot().unwrap().healthy);
assert!(lease.resource_snapshot().unwrap().charged >= input_bytes);
drop(input);
drop(state);
drop(root);
assert_eq!(lease.resource_snapshot().unwrap().active_accounts, 0);
drop((startup, lease));
assert!(
std::process::Command::new("kill")
.args(["-TERM", &std::process::id().to_string()])
.status()
.unwrap()
.success()
);
Ok(saddle_runtime::Application::new())
}
))
));
let outcome = coordinator
.as_mut()
.poll(&mut std::task::Context::from_waker(std::task::Waker::noop()));
assert!(matches!(outcome, std::task::Poll::Ready(Ok(()))));
}
}