use super::*;
use saddle_runtime::profusegw::{ProfuseGwReservedScopeFailure, ProfuseGwSuspendedScope};
use saddle_runtime::request_task::reserved::ReservedRequestFailure;
type Cancel = tokio::sync::futures::OwnedNotified;
pub(super) struct ReservedDbOwner<T> {
pub(super) serial: Option<SerialScope>,
suspended: Option<ProfuseGwSuspendedScope<Cancel, T>>,
state: Arc<RequestState>,
process: Arc<saddle_db::internal::StartupManagedDatabaseProcessCapability>,
}
impl<T> ReservedDbOwner<T> {
pub(super) async fn resolve(
&mut self,
result: Result<
saddle_db::internal::ReservedDatabaseCompletion<Cancel, T>,
ProfuseGwReservedScopeFailure,
>,
context: &saddle_runtime::request_task::reserved::ReservedTaskContext,
) -> Result<T, ScopeDatabaseError> {
match result {
Ok(completion) => {
self.suspended = Some(completion.suspended);
self.retain(completion.unretained);
Ok(self.publish().await)
}
Err(error) => {
let class = match &error {
ProfuseGwReservedScopeFailure::Preparation(reason) => {
use saddle_core::{
BoundedDiagnostic, BoundedDiagnosticCause, CaptureSite,
DiagnosticCategory, DiagnosticCode, DiagnosticStage,
};
let code = DiagnosticCode::new("service.database.preparation").unwrap();
self.retain(Some(context.view().source_description(
&PreparationDescription(reason),
BoundedDiagnostic::capture(
DiagnosticCategory::UnexpectedError,
CaptureSite::FirstObserved,
BoundedDiagnosticCause::new(DiagnosticStage::RequestDb, code),
),
code,
self.state.diagnostic_handle.as_ref(),
saddle_observability::root_diagnostic::RootRequestEvent::Database,
Default::default(),
)));
ScopeDatabaseError::Resource
}
ProfuseGwReservedScopeFailure::Execution(reason) => match reason {
saddle_runtime::profusegw::ProfuseGwScopeFailure::Stopped(
saddle_runtime::profusegw::ProfuseGwScopeStop::TimedOut,
) => ScopeDatabaseError::TimedOut,
saddle_runtime::profusegw::ProfuseGwScopeFailure::Stopped(
saddle_runtime::profusegw::ProfuseGwScopeStop::Cancelled,
) => ScopeDatabaseError::Cancelled,
saddle_runtime::profusegw::ProfuseGwScopeFailure::Panicked => {
ScopeDatabaseError::Panicked
}
_ => ScopeDatabaseError::State,
},
};
let suspended = self.process.finish_reserved_without_connection(
self.serial
.take()
.expect("preparation returns original owner"),
error,
);
let (_original, terminal) = suspended
.into_response_parts()
.unwrap_or_else(|_| std::process::abort());
*self.state.terminal.lock().unwrap() = RequestTerminal::PostDatabase(terminal);
self.state.completed.notify_one();
Err(class)
}
}
}
async fn publish(&mut self) -> T {
let suspended = self
.suspended
.as_mut()
.expect("D completion saved before publication");
let (terminal, value) = match suspended.resume().await {
Ok((scope, value)) => (RequestTerminal::Serial(scope), value),
Err(_) => {
let (value, terminal) = self
.suspended
.take()
.unwrap()
.into_response_parts()
.unwrap_or_else(|_| std::process::abort());
(RequestTerminal::PostDatabase(terminal), value)
}
};
self.suspended = None;
*self.state.terminal.lock().unwrap() = terminal;
self.state.completed.notify_one();
value
}
fn retain(&self, failure: Option<ReservedRequestFailure>) {
if let Some(failure) = failure {
let mut slot = self.state.reserved_failure.lock().unwrap();
assert!(
slot.is_none(),
"pending original failure prohibits later DB execution"
);
*slot = Some(failure);
}
}
}
impl<T> Drop for ReservedDbOwner<T> {
fn drop(&mut self) {
let terminal = if let Some(suspended) = self.suspended.take() {
let (_, terminal) = suspended
.into_response_parts()
.unwrap_or_else(|_| std::process::abort());
Some(terminal)
} else if let Some(scope) = self.serial.take() {
let suspended = self.process.finish_reserved_without_connection(scope, ());
let (_, terminal) = suspended
.into_response_parts()
.unwrap_or_else(|_| std::process::abort());
Some(terminal)
} else {
None
};
if let Some(terminal) = terminal {
*self.state.terminal.lock().unwrap() = RequestTerminal::PostDatabase(terminal);
self.state.completed.notify_one();
}
}
}
impl DatabaseRequest {
pub(super) fn reserved_owner<T>(
&mut self,
) -> Result<
(
Arc<saddle_runtime::request_task::reserved::ReservedTaskContext>,
ReservedDbOwner<T>,
),
ScopeDatabaseError,
> {
let context = self
.state
.reserved_context
.clone()
.ok_or(ScopeDatabaseError::State)?;
if self.state.reserved_failure.lock().unwrap().is_some() {
return Err(ScopeDatabaseError::State);
}
let process = self
.process
.clone()
.ok_or(ScopeDatabaseError::Unavailable)?;
let serial = self.begin_serial()?;
Ok((
context,
ReservedDbOwner {
serial: Some(serial),
suspended: None,
state: self.state.clone(),
process,
},
))
}
pub(crate) fn take_reserved_failure(&self) -> Option<ReservedRequestFailure> {
self.state.reserved_failure.lock().unwrap().take()
}
pub async fn constructed_transaction<B>(
&mut self,
isolation: saddle_db::internal::TransactionIsolation,
body: B,
) -> saddle_db::internal::ScopeTransactionOutcome<B::Value, B::Error>
where
B: saddle_db::internal::ReservedTransactionBody<Cancel> + Send,
{
use saddle_db::internal::{ScopeTransactionAbort as A, ScopeTransactionOutcome as O};
let Some(context) = self.state.reserved_context.clone() else {
return O::Rejected(A::Technical(ScopeDatabaseError::State));
};
if self.state.reserved_failure.lock().unwrap().is_some() {
return O::Rejected(A::Technical(ScopeDatabaseError::State));
}
let Some(process) = self.process.clone() else {
return O::Rejected(A::Technical(ScopeDatabaseError::Unavailable));
};
let scope = match self.begin_serial() {
Ok(scope) => scope,
Err(error) => return O::Rejected(A::Technical(error)),
};
let mut owner = ReservedDbOwner {
serial: Some(scope),
suspended: None,
state: self.state.clone(),
process: process.clone(),
};
match process
.transaction_scope_reserved(&mut owner.serial, &context, isolation, body)
.await
{
Ok(completion) => {
owner.suspended = Some(completion.suspended);
owner.retain(completion.unretained);
owner.publish().await.outcome
}
Err(error) => {
let class = match &error {
ProfuseGwReservedScopeFailure::Preparation(reason) => {
use saddle_core::{
BoundedDiagnostic, BoundedDiagnosticCause, CaptureSite,
DiagnosticCategory, DiagnosticCode, DiagnosticStage,
};
let code = DiagnosticCode::new("service.database.preparation").unwrap();
let failure = context.view().source_description(
&PreparationDescription(reason),
BoundedDiagnostic::capture(
DiagnosticCategory::UnexpectedError,
CaptureSite::FirstObserved,
BoundedDiagnosticCause::new(DiagnosticStage::RequestDb, code),
),
code,
self.diagnostic_handle.as_ref(),
saddle_observability::root_diagnostic::RootRequestEvent::Database,
Default::default(),
);
owner.retain(Some(failure));
ScopeDatabaseError::Resource
}
ProfuseGwReservedScopeFailure::Execution(reason) => match reason {
saddle_runtime::profusegw::ProfuseGwScopeFailure::Stopped(
saddle_runtime::profusegw::ProfuseGwScopeStop::TimedOut,
) => ScopeDatabaseError::TimedOut,
saddle_runtime::profusegw::ProfuseGwScopeFailure::Stopped(
saddle_runtime::profusegw::ProfuseGwScopeStop::Cancelled,
) => ScopeDatabaseError::Cancelled,
saddle_runtime::profusegw::ProfuseGwScopeFailure::Panicked => {
ScopeDatabaseError::Panicked
}
saddle_runtime::profusegw::ProfuseGwScopeFailure::AlreadySupervised
| saddle_runtime::profusegw::ProfuseGwScopeFailure::ScopeAlreadyEntered => {
ScopeDatabaseError::State
}
},
};
let suspended =
process.finish_reserved_without_connection(owner.serial.take().unwrap(), error);
let (_original, terminal) = suspended
.into_response_parts()
.unwrap_or_else(|_| std::process::abort());
*owner.state.terminal.lock().unwrap() = RequestTerminal::PostDatabase(terminal);
owner.state.completed.notify_one();
O::Rejected(A::Technical(class))
}
}
}
}
#[derive(Debug)]
struct PreparationDescription<'a>(&'a saddle_runtime::request_task::reserved::ReservedContextError);
impl std::fmt::Display for PreparationDescription<'_> {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "database scope preparation: {:?}", self.0)
}
}