saddle-runtime 0.3.21

//! Framework composition only. Facade owns the ONE hook and emergency writer.
use saddle_core::{
    CaptureSite, Diagnostic, DiagnosticCategory, DiagnosticCause, DiagnosticCode, DiagnosticStage,
    SaddleError,
};
use saddle_observability::{CallContext, EmergencyDiagnosticHandle, EventContext, Observer};
use std::{
    cell::RefCell,
    future::Future,
    panic::{AssertUnwindSafe, PanicHookInfo, catch_unwind, resume_unwind},
    sync::OnceLock,
};

static OUTPUT: OnceLock<EmergencyDiagnosticHandle> = OnceLock::new();

/// Output observation, NOT a transaction/resource/durable-write receipt.
#[derive(Debug)]
pub struct RuntimeDiagnosticExit {
    pub shutdown: saddle_observability::DiagnosticShutdown,
    pub snapshot: saddle_observability::DiagnosticOutputSnapshot,
}

pub(crate) fn close_output(
    mut owner: saddle_observability::EmergencyDiagnostics,
    deadline: Option<std::time::Instant>,
) -> RuntimeDiagnosticExit {
    let shutdown = loop {
        let status = owner.shutdown();
        if status != saddle_observability::DiagnosticShutdown::Pending {
            break status;
        }
        let remaining = deadline
            .map(|d| d.saturating_duration_since(std::time::Instant::now()))
            .unwrap_or_default();
        if remaining.is_zero() {
            break status;
        }
        // Same absolute shutdown budget. No new async timer, writer join or
        // wait after expiration. Startup failure without a budget is immediate.
        std::thread::sleep(remaining.min(std::time::Duration::from_millis(1)));
    };
    RuntimeDiagnosticExit {
        shutdown,
        snapshot: owner.snapshot(),
    }
}
type RequestContext = (Observer, CallContext, EventContext);
struct Frame {
    stage: DiagnosticStage,
    task: &'static str,
    context: Option<RequestContext>,
    panic: Option<Diagnostic>,
}
thread_local! { static CURRENT: RefCell<Option<Frame>> = const { RefCell::new(None) }; }

/// Called once by Facade with its existing writer handle. No thread/hook is created.
pub fn install_output(handle: EmergencyDiagnosticHandle) -> Result<(), EmergencyDiagnosticHandle> {
    OUTPUT.set(handle)
}

/// Called by Facade's sole panic hook. False means no Runtime poll context;
/// the Facade must capture that unassociated panic itself. Never reads payload.
pub fn capture_current_panic(info: &PanicHookInfo<'_>) -> bool {
    CURRENT.with(|slot| {
        let Ok(mut slot) = slot.try_borrow_mut() else {
            return false;
        };
        let Some(frame) = slot.as_mut() else {
            return false;
        };
        let diagnostic = Diagnostic::capture_panic(info, frame.stage).with_task(code(frame.task));
        emit(&diagnostic, frame.context.as_ref());
        frame.panic = Some(diagnostic);
        true
    })
}

fn code(value: &'static str) -> DiagnosticCode {
    DiagnosticCode::new(value).expect("Runtime diagnostic codes are static schema identifiers")
}

pub(crate) fn emit(diagnostic: &Diagnostic, context: Option<&RequestContext>) {
    if let Some(output) = OUTPUT.get() {
        if let Some((observer, call, event)) = context {
            let _ = observer.record_diagnostic(diagnostic, output, Some((call, event)));
        } else {
            let _ = output.submit(diagnostic);
        }
    }
    // Absence/Full/Closed is NOT a delivery acknowledgement. Facade retains
    // the writer owner and must inspect its snapshot at the ORIGINAL deadline.
}

#[track_caller]
pub(crate) fn failure(
    stage: DiagnosticStage,
    category: DiagnosticCategory,
    name: &'static str,
) -> Diagnostic {
    Diagnostic::capture(
        category,
        CaptureSite::FirstObserved,
        DiagnosticCause::new(stage, code(name)),
    )
}

/// Invoke at the existing catch boundary, INSIDE Admission's admitted poll.
/// CURRENT is replaced/restored per synchronous poll, never held across await.
pub(crate) fn catching<T>(
    stage: DiagnosticStage,
    task: &'static str,
    context: Option<RequestContext>,
    f: impl FnOnce() -> T,
) -> Result<T, Diagnostic> {
    let previous = CURRENT.with(|slot| {
        slot.replace(Some(Frame {
            stage,
            task,
            context,
            panic: None,
        }))
    });
    let outcome = catch_unwind(AssertUnwindSafe(f));
    let frame = CURRENT
        .with(|slot| slot.replace(previous))
        .expect("matching synchronous diagnostic frame");
    match outcome {
        Ok(value) => Ok(value),
        Err(payload) => {
            // An inner managed poll has already captured and emitted this
            // occurrence. resume_unwind does not invoke the hook a second time.
            if let Ok(diagnostic) = payload.downcast::<Diagnostic>() {
                return Err(*diagnostic);
            }
            let diagnostic = frame.panic.unwrap_or_else(|| {
                let d = failure(
                    stage,
                    DiagnosticCategory::Panic,
                    "runtime.panic_without_hook",
                )
                .with_task(code(task));
                emit(&d, frame.context.as_ref());
                d
            });
            Err(diagnostic)
        }
    }
}

pub(crate) async fn task<F: Future>(
    future: F,
    stage: DiagnosticStage,
    name: &'static str,
) -> F::Output {
    let mut future = std::pin::pin!(future);
    std::future::poll_fn(|cx| {
        match catching(stage, name, None, || {
            let poll = future.as_mut().poll(cx);
            #[cfg(test)]
            if poll.is_ready()
                && name == "runtime.managed_request_task"
                && std::env::var_os("RUNTIME_DIAGNOSTIC_MANAGER_FAULT").is_some()
            {
                // Private fixed manager failure AFTER the envelope returned,
                // not a handler panic; compiled out of published artifacts.
                panic!("DIAGNOSTIC_PRIVATE_SENTINEL");
            }
            poll
        }) {
            Ok(poll) => poll,
            Err(diagnostic) => resume_unwind(Box::new(diagnostic)),
        }
    })
    .await
}

pub(crate) fn joined(error: tokio::task::JoinError) {
    if error.is_cancelled() {
        return;
    }
    if error.is_panic() {
        let payload = error.into_panic();
        if payload.is::<Diagnostic>() {
            return;
        } // already emitted at original poll
    }
    let d = failure(
        DiagnosticStage::BackgroundTask,
        DiagnosticCategory::UnexpectedError,
        "runtime.task_join_failed",
    );
    emit(&d, None);
}

#[track_caller]
pub(crate) fn attach(
    error: SaddleError,
    stage: DiagnosticStage,
    name: &'static str,
) -> SaddleError {
    if error.diagnostic().is_some() {
        error
    } else {
        error.with_diagnostic(failure(stage, DiagnosticCategory::UnexpectedError, name))
    }
}

pub(crate) fn report(error: &SaddleError) {
    if let Some(d) = error.diagnostic() {
        emit(d, None);
    }
}

pub(crate) fn cleanup(
    error: SaddleError,
    primary: Option<&SaddleError>,
    stage: DiagnosticStage,
) -> SaddleError {
    let mut error = attach(error, stage, "runtime.cleanup_failed");
    if let Some(primary) = primary {
        error = error.during_cleanup_of(primary);
    }
    report(&error);
    error
}