use std::sync::{Arc, Mutex, MutexGuard};
use saddle_core::{ErrorKind, Result, SaddleError};
use crate::compiled_route::OfficialCompiledDriverFinalizer;
struct State {
armed: bool,
submitted: Option<OfficialCompiledDriverFinalizer>,
consumed: bool,
}
#[derive(Clone)]
pub struct PendingDriverFinalizerSlot {
state: Arc<Mutex<State>>,
}
impl PendingDriverFinalizerSlot {
pub(crate) fn new() -> Self {
Self {
state: Arc::new(Mutex::new(State {
armed: false,
submitted: None,
consumed: false,
})),
}
}
pub fn arm(&self) {
let mut state = lock(&self.state);
if state.armed || state.consumed {
std::process::abort();
}
state.armed = true;
}
pub fn submit(&self, finalizer: OfficialCompiledDriverFinalizer) {
let mut state = lock(&self.state);
if !state.armed || state.submitted.is_some() || state.consumed {
std::process::abort();
}
state.submitted = Some(finalizer);
}
pub(crate) fn finish(
&self,
runtime: tokio::runtime::Runtime,
application_result: Result<()>,
) -> Result<()> {
let pending = {
let mut state = lock(&self.state);
if state.consumed {
std::process::abort();
}
state.consumed = true;
if state.armed {
Some(
state
.submitted
.take()
.unwrap_or_else(|| std::process::abort()),
)
} else {
None
}
};
let finalizer_result = match pending {
Some(pending) => pending
.bind_runtime(runtime)
.finish()
.map_err(|_| finalization_error())
.and_then(|report| {
if report.ledger.healthy && !report.watermark.breached && !report.task_failed {
Ok(())
} else {
Err(finalization_error())
}
}),
None => {
drop(runtime);
Ok(())
}
};
application_result.and(finalizer_result)
}
}
fn lock(state: &Mutex<State>) -> MutexGuard<'_, State> {
state.lock().unwrap_or_else(|_| std::process::abort())
}
fn finalization_error() -> SaddleError {
SaddleError::new(
ErrorKind::Infrastructure,
"runtime.driver_finalization_failed",
"the managed Runtime driver did not finalize cleanly",
)
}