saddle-runtime 0.2.0

//! Crate-private Runtime ownership for the fixed Observability writer.
//!
//! The first F08 Runtime slice owns request-terminal correlation and the writer
//! barriers. Cross-component route/Service/DB assembly remains deliberately
//! outside this module until its independently frozen bindings are available.

#![allow(dead_code)]

use std::{
    future::Future,
    pin::Pin,
    sync::{
        Mutex,
        atomic::{AtomicBool, AtomicU64, Ordering},
    },
    task::{Context, Poll, Waker},
    thread::{self, JoinHandle},
};

use saddle_admission::RequestMemory;
use saddle_observability::file::{
    ActiveCorrelationCall, CompletionResult, CompletionRole, CompletionSignal,
    CorrelationCallOutcome, CorrelationHealth, CorrelationRecord, CorrelationSinkIdentity,
    CorrelationSinkOwner, CorrelationSubmitError, FixedCompletion, FixedFailure, FixedFileLimits,
    FixedNotification, NormalizedSignedProviderFilesystemRuntimeEvidence,
    PreparedProductionFixedFileCore, fixed_core_layout, prepare_signed_provider_fixed_file_core,
};

use saddle_admission::{ObservabilityQueuePairAuthority, PrepairObservabilityQueueDomainOwner};
use saddle_core::{ComponentLifecycle, ErrorKind, LifecycleFuture, SaddleError};

const WRITER_NAME: &str = "sdl-log-writer";
const WRITER_STACK_BYTES: usize = 1_048_576;
const PRODUCTION_BLOCKS: usize = 4;
const PRODUCTION_BYTES: usize = 1_024;
const PRODUCTION_COMMANDS: usize = 6;
const PRODUCTION_PATH: usize = 256;
const PRODUCTION_DIRENT: usize = 4_096;
const PRODUCTION_ENTRIES: usize = 32;

type ProductionManagedFileWriter = ManagedFileWriter;

static WRITER_LEASED: AtomicBool = AtomicBool::new(false);
static REGISTRY_UNHEALTHY: AtomicBool = AtomicBool::new(false);
static CORRELATION_UNHEALTHY: AtomicBool = AtomicBool::new(false);
static NOTIFICATIONS: [NotificationSlot; 3] = [
    NotificationSlot::new(),
    NotificationSlot::new(),
    NotificationSlot::new(),
];

struct NotificationSlot {
    registered: AtomicBool,
    generation: AtomicU64,
    signalled: AtomicBool,
    // One reference to the caller's already allocated Tokio task. Under the
    // approved Tokio 1.53.1 fingerprint clone/drop only changes Header's
    // refcount; it neither allocates nor transfers the original task-storage
    // owner/reserve. Completion, cancellation and wait Drop clear this slot.
    waker: Mutex<Option<Waker>>,
}

impl NotificationSlot {
    const fn new() -> Self {
        Self {
            registered: AtomicBool::new(false),
            generation: AtomicU64::new(0),
            signalled: AtomicBool::new(false),
            waker: Mutex::new(None),
        }
    }

    fn register(&self) -> Result<(), ManagedWriterError> {
        self.registered
            .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
            .map(|_| ())
            .map_err(|_| ManagedWriterError::NotificationBusy)
    }

    fn clear(&self) {
        self.registered.store(false, Ordering::Release);
        self.signalled.store(false, Ordering::Release);
        match self.waker.lock() {
            Ok(mut waker) => *waker = None,
            Err(_) => REGISTRY_UNHEALTHY.store(true, Ordering::Release),
        }
    }

    fn reset(&self) {
        self.clear();
        self.generation.store(0, Ordering::Release);
    }
}

/// Lock-free publication plus a non-blocking fixed-slot wake attempt.
fn fixed_notify(token: usize, signal: CompletionSignal) {
    let Some(slot) = NOTIFICATIONS.get(token) else {
        REGISTRY_UNHEALTHY.store(true, Ordering::Release);
        return;
    };
    if token != signal.role as usize || signal.generation == 0 {
        REGISTRY_UNHEALTHY.store(true, Ordering::Release);
        return;
    }
    let previous = slot.generation.swap(signal.generation, Ordering::AcqRel);
    if signal.generation <= previous {
        REGISTRY_UNHEALTHY.store(true, Ordering::Release);
    }
    slot.signalled.store(true, Ordering::Release);
    if let Ok(mut registered) = slot.waker.try_lock() {
        if let Some(waker) = registered.take() {
            waker.wake();
        }
    }
}

fn fixed_notifications() -> [FixedNotification; 3] {
    std::array::from_fn(|token| FixedNotification::new(token, fixed_notify))
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(crate) enum ManagedWriterError {
    AlreadyOwned,
    ThreadSpawn,
    ThreadPanic,
    Completion(FixedFailure),
    CompletionStale,
    CompletionRecycle,
    NotificationBusy,
    NotificationUnhealthy,
    Submit,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(crate) struct RuntimeCorrelationIdentity {
    plan_digest: [u8; 32],
    owner_generation: u64,
    build_identity: [u8; 32],
    route_set_attestation: [u8; 32],
    sink: CorrelationSinkIdentity,
}

impl RuntimeCorrelationIdentity {
    fn new(
        binding: ([u8; 32], u64, [u8; 32], [u8; 32]),
        sink: CorrelationSinkIdentity,
    ) -> Result<Self, ManagedWriterError> {
        let (plan_digest, owner_generation, build_identity, route_set_attestation) = binding;
        if owner_generation == 0
            || [plan_digest, build_identity, route_set_attestation].contains(&[0; 32])
            || sink.sink_identity() == [0; 32]
        {
            return Err(ManagedWriterError::Submit);
        }
        Ok(Self {
            plan_digest,
            owner_generation,
            build_identity,
            route_set_attestation,
            sink,
        })
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(crate) enum RequestCorrelationOutcome {
    Success,
    Error,
    Cancel,
    Shutdown,
    Panic,
}

#[derive(Clone, Copy)]
pub(crate) struct RequestCorrelationContext<'a> {
    trace_id: &'a str,
    span_id: &'a str,
    route: &'a str,
    route_identity: [u8; 32],
    service: &'a str,
    operation: &'a str,
}

impl<'a> RequestCorrelationContext<'a> {
    fn record(self) -> CorrelationRecord<'a> {
        CorrelationRecord::started(
            saddle_observability::file::CorrelationCallKind::ExternalRoute,
            self.trace_id,
            self.span_id,
            None,
            self.route,
            self.service,
            self.operation,
        )
    }
}

impl RequestCorrelationOutcome {
    const fn fixed(self) -> CorrelationCallOutcome {
        match self {
            Self::Success => CorrelationCallOutcome::Success,
            Self::Cancel | Self::Shutdown => CorrelationCallOutcome::Cancelled,
            Self::Error | Self::Panic => CorrelationCallOutcome::Failure,
        }
    }
}

/// Runtime-private owner whose reserved finish block remains charged to the
/// current request until a terminal outcome is synchronously committed.
pub(crate) struct RequestCorrelationOwner<
    'a,
    const BLOCKS: usize,
    const BYTES: usize,
    const COMMANDS: usize,
> {
    call: Option<ActiveCorrelationCall<'a, BLOCKS, BYTES, COMMANDS>>,
    sink: &'a CorrelationSinkOwner<BLOCKS, BYTES, COMMANDS>,
    identity: RuntimeCorrelationIdentity,
    context: RequestCorrelationContext<'a>,
    service_parent_issued: bool,
}

impl<'a, const BLOCKS: usize, const BYTES: usize, const COMMANDS: usize>
    RequestCorrelationOwner<'a, BLOCKS, BYTES, COMMANDS>
{
    fn begin(
        sink: &'a CorrelationSinkOwner<BLOCKS, BYTES, COMMANDS>,
        identity: RuntimeCorrelationIdentity,
        memory: &RequestMemory,
        context: RequestCorrelationContext<'a>,
    ) -> Result<Self, CorrelationSubmitError> {
        if CORRELATION_UNHEALTHY.load(Ordering::Acquire) {
            return Err(CorrelationSubmitError::Unhealthy(FixedFailure::Invariant));
        }
        if context.route_identity == [0; 32] {
            CORRELATION_UNHEALTHY.store(true, Ordering::Release);
            return Err(CorrelationSubmitError::Encoding);
        }
        match sink.begin_call(memory, context.record()) {
            Ok(call) => Ok(Self {
                call: Some(call),
                sink,
                identity,
                context,
                service_parent_issued: false,
            }),
            Err(error) => {
                CORRELATION_UNHEALTHY.store(true, Ordering::Release);
                Err(error)
            }
        }
    }

    pub(crate) fn finish(
        mut self,
        outcome: RequestCorrelationOutcome,
    ) -> Result<(), CorrelationSubmitError> {
        let call = self.call.take().ok_or(CorrelationSubmitError::Encoding)?;
        match call.finish(outcome.fixed()) {
            Ok(()) => Ok(()),
            Err(error) => {
                CORRELATION_UNHEALTHY.store(true, Ordering::Release);
                Err(error)
            }
        }
    }

    pub(crate) fn issue_compiled_service_parent<'memory>(
        &mut self,
        memory: &'memory RequestMemory,
    ) -> Result<
        CompiledServiceCorrelationParent<'a, 'memory, BLOCKS, BYTES, COMMANDS>,
        CorrelationSubmitError,
    > {
        if self.service_parent_issued || CORRELATION_UNHEALTHY.load(Ordering::Acquire) {
            CORRELATION_UNHEALTHY.store(true, Ordering::Release);
            return Err(CorrelationSubmitError::Unhealthy(FixedFailure::Invariant));
        }
        self.service_parent_issued = true;
        Ok(CompiledServiceCorrelationParent {
            sink: self.sink,
            memory,
            identity: self.identity,
            trace_id: self.context.trace_id,
            parent_span_id: self.context.span_id,
            route: self.context.route,
            route_identity: self.context.route_identity,
        })
    }
}

/// Linear bridge from the already admitted Runtime request to the unique
/// compiled Service producer. Fields are private and the type has no Clone or
/// constructor, so generated/business code cannot forge or replay a parent.
///
/// ```compile_fail
/// use saddle_runtime::observability::CompiledServiceCorrelationParent;
/// fn replay(
///     value: &CompiledServiceCorrelationParent<'_, '_, 1, 1, 1>,
/// ) -> CompiledServiceCorrelationParent<'_, '_, 1, 1, 1> {
///     value.clone()
/// }
/// ```
///
/// ```compile_fail
/// use saddle_runtime::observability::CompiledServiceCorrelationParent;
/// let _ = CompiledServiceCorrelationParent::<1, 1, 1> { };
/// ```
#[doc(hidden)]
pub struct CompiledServiceCorrelationParent<
    'request,
    'memory,
    const BLOCKS: usize,
    const BYTES: usize,
    const COMMANDS: usize,
> {
    sink: &'request CorrelationSinkOwner<BLOCKS, BYTES, COMMANDS>,
    memory: &'memory RequestMemory,
    identity: RuntimeCorrelationIdentity,
    trace_id: &'request str,
    parent_span_id: &'request str,
    route: &'request str,
    route_identity: [u8; 32],
}

#[doc(hidden)]
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum CompiledServiceCorrelationOutcome {
    Success,
    Error,
    Cancelled,
    Shutdown,
    Panic,
}

/// Non-Clone typed Service child. Drop commits `abandoned`; explicit finish is
/// consuming and exactly once.
#[doc(hidden)]
pub struct CompiledServiceCorrelationChild<
    'request,
    const BLOCKS: usize,
    const BYTES: usize,
    const COMMANDS: usize,
> {
    call: Option<ActiveCorrelationCall<'request, BLOCKS, BYTES, COMMANDS>>,
}

impl<'request, const BLOCKS: usize, const BYTES: usize, const COMMANDS: usize>
    CompiledServiceCorrelationParent<'request, '_, BLOCKS, BYTES, COMMANDS>
{
    pub fn begin(
        self,
        span_id: &'request str,
        service: &'request str,
        operation: &'request str,
    ) -> Result<
        CompiledServiceCorrelationChild<'request, BLOCKS, BYTES, COMMANDS>,
        CorrelationSubmitError,
    > {
        if CORRELATION_UNHEALTHY.load(Ordering::Acquire)
            || self.identity.owner_generation == 0
            || self.identity.sink.sink_identity() == [0; 32]
            || self.route_identity == [0; 32]
        {
            CORRELATION_UNHEALTHY.store(true, Ordering::Release);
            return Err(CorrelationSubmitError::Unhealthy(FixedFailure::Invariant));
        }
        let call = self.sink.begin_call(
            self.memory,
            CorrelationRecord::started(
                saddle_observability::file::CorrelationCallKind::Service,
                self.trace_id,
                span_id,
                Some(self.parent_span_id),
                self.route,
                service,
                operation,
            ),
        )?;
        Ok(CompiledServiceCorrelationChild { call: Some(call) })
    }
}

impl<const BLOCKS: usize, const BYTES: usize, const COMMANDS: usize>
    CompiledServiceCorrelationChild<'_, BLOCKS, BYTES, COMMANDS>
{
    pub fn finish(
        mut self,
        outcome: CompiledServiceCorrelationOutcome,
    ) -> Result<(), CorrelationSubmitError> {
        let fixed = match outcome {
            CompiledServiceCorrelationOutcome::Success => CorrelationCallOutcome::Success,
            CompiledServiceCorrelationOutcome::Cancelled
            | CompiledServiceCorrelationOutcome::Shutdown => CorrelationCallOutcome::Cancelled,
            CompiledServiceCorrelationOutcome::Error | CompiledServiceCorrelationOutcome::Panic => {
                CorrelationCallOutcome::Failure
            }
        };
        let call = self.call.take().ok_or(CorrelationSubmitError::Encoding)?;
        match call.finish(fixed) {
            Ok(()) => Ok(()),
            Err(error) => {
                CORRELATION_UNHEALTHY.store(true, Ordering::Release);
                Err(error)
            }
        }
    }
}

impl<const BLOCKS: usize, const BYTES: usize, const COMMANDS: usize> Drop
    for CompiledServiceCorrelationChild<'_, BLOCKS, BYTES, COMMANDS>
{
    fn drop(&mut self) {
        let Some(call) = self.call.take() else {
            return;
        };
        if call.finish(CorrelationCallOutcome::Abandoned).is_err() {
            CORRELATION_UNHEALTHY.store(true, Ordering::Release);
            std::process::abort();
        }
    }
}

impl<const BLOCKS: usize, const BYTES: usize, const COMMANDS: usize> Drop
    for RequestCorrelationOwner<'_, BLOCKS, BYTES, COMMANDS>
{
    fn drop(&mut self) {
        let Some(call) = self.call.take() else {
            return;
        };
        if call.finish(CorrelationCallOutcome::Abandoned).is_err() {
            CORRELATION_UNHEALTHY.store(true, Ordering::Release);
            std::process::abort();
        }
    }
}

struct CompletionWait<const BLOCKS: usize, const BYTES: usize, const COMMANDS: usize> {
    completion: Option<FixedCompletion<BLOCKS, BYTES, COMMANDS>>,
    role: CompletionRole,
    registered: bool,
}

impl<const BLOCKS: usize, const BYTES: usize, const COMMANDS: usize>
    CompletionWait<BLOCKS, BYTES, COMMANDS>
{
    fn new(completion: FixedCompletion<BLOCKS, BYTES, COMMANDS>, role: CompletionRole) -> Self {
        Self {
            completion: Some(completion),
            role,
            registered: false,
        }
    }

    fn slot(&self) -> &'static NotificationSlot {
        &NOTIFICATIONS[self.role as usize]
    }

    fn unregister(&mut self) {
        if self.registered {
            self.slot().clear();
            self.registered = false;
        }
    }
}

impl<const BLOCKS: usize, const BYTES: usize, const COMMANDS: usize> Future
    for CompletionWait<BLOCKS, BYTES, COMMANDS>
{
    type Output = Result<(), ManagedWriterError>;

    fn poll(mut self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<Self::Output> {
        if REGISTRY_UNHEALTHY.load(Ordering::Acquire) {
            self.unregister();
            return Poll::Ready(Err(ManagedWriterError::NotificationUnhealthy));
        }
        let completion = self
            .completion
            .as_ref()
            .expect("pending completion remains owned");
        match completion.result() {
            CompletionResult::Ready(result) => {
                self.unregister();
                let completion = self.completion.take().expect("completion remains owned");
                if completion.recycle().is_err() {
                    return Poll::Ready(Err(ManagedWriterError::CompletionRecycle));
                }
                Poll::Ready(result.map_err(ManagedWriterError::Completion))
            }
            CompletionResult::Stale => {
                self.unregister();
                Poll::Ready(Err(ManagedWriterError::CompletionStale))
            }
            CompletionResult::Pending => {
                if !self.registered {
                    if let Err(error) = self.slot().register() {
                        return Poll::Ready(Err(error));
                    }
                    self.registered = true;
                }
                let slot = self.slot();
                match slot.waker.lock() {
                    Ok(mut waker) => {
                        if waker
                            .as_ref()
                            .is_none_or(|registered| !registered.will_wake(context.waker()))
                        {
                            *waker = Some(context.waker().clone());
                        }
                    }
                    Err(_) => {
                        REGISTRY_UNHEALTHY.store(true, Ordering::Release);
                        self.unregister();
                        return Poll::Ready(Err(ManagedWriterError::NotificationUnhealthy));
                    }
                }
                if slot.signalled.swap(false, Ordering::AcqRel) {
                    context.waker().wake_by_ref();
                }
                Poll::Pending
            }
        }
    }
}

impl<const BLOCKS: usize, const BYTES: usize, const COMMANDS: usize> Drop
    for CompletionWait<BLOCKS, BYTES, COMMANDS>
{
    fn drop(&mut self) {
        self.unregister();
    }
}

pub(crate) struct ManagedFileWriter {
    sink: CorrelationSinkOwner<PRODUCTION_BLOCKS, PRODUCTION_BYTES, PRODUCTION_COMMANDS>,
    identity: RuntimeCorrelationIdentity,
    writer: Option<JoinHandle<Result<(), FixedFailure>>>,
    shutdown_complete: bool,
}

impl ManagedFileWriter {
    pub(crate) async fn start_until<D>(
        prepared: PreparedProductionFixedFileCore,
        binding: ([u8; 32], u64, [u8; 32], [u8; 32]),
        deadline: D,
    ) -> Result<Self, ManagedWriterError>
    where
        D: Future<Output = ()>,
    {
        if WRITER_LEASED
            .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
            .is_err()
        {
            return Err(ManagedWriterError::AlreadyOwned);
        }
        REGISTRY_UNHEALTHY.store(false, Ordering::Release);
        CORRELATION_UNHEALTHY.store(false, Ordering::Release);
        for slot in &NOTIFICATIONS {
            slot.reset();
        }
        let (sink, writer_entry, startup, _termination) = prepared.into_runtime_managed_parts();
        let sink = CorrelationSinkOwner::from_fixed_sink(sink);
        let identity = match RuntimeCorrelationIdentity::new(binding, sink.identity()) {
            Ok(identity) => identity,
            Err(error) => {
                WRITER_LEASED.store(false, Ordering::Release);
                return Err(error);
            }
        };
        let writer = match thread::Builder::new()
            .name(WRITER_NAME.to_owned())
            .stack_size(WRITER_STACK_BYTES)
            .spawn(move || writer_entry.run_runtime_managed())
        {
            Ok(writer) => writer,
            Err(_) => {
                WRITER_LEASED.store(false, Ordering::Release);
                return Err(ManagedWriterError::ThreadSpawn);
            }
        };
        let startup = CompletionWait::new(startup, CompletionRole::Startup);
        tokio::pin!(startup);
        tokio::pin!(deadline);
        let startup = tokio::select! {
            biased;
            () = &mut deadline => std::process::abort(),
            result = &mut startup => result,
        };
        if let Err(error) = startup {
            let _ = writer.join();
            WRITER_LEASED.store(false, Ordering::Release);
            return Err(error);
        }
        Ok(Self {
            sink,
            identity,
            writer: Some(writer),
            shutdown_complete: false,
        })
    }

    pub(crate) fn identity(&self) -> RuntimeCorrelationIdentity {
        self.identity
    }

    pub(crate) fn begin_request<'a>(
        &'a self,
        memory: &RequestMemory,
        context: RequestCorrelationContext<'a>,
    ) -> Result<
        RequestCorrelationOwner<'a, PRODUCTION_BLOCKS, PRODUCTION_BYTES, PRODUCTION_COMMANDS>,
        CorrelationSubmitError,
    > {
        RequestCorrelationOwner::begin(&self.sink, self.identity, memory, context)
    }

    pub(crate) async fn flush(&self) -> Result<(), ManagedWriterError> {
        let completion = self
            .sink
            .try_flush()
            .map_err(|_| ManagedWriterError::Submit)?;
        CompletionWait::new(completion.into_completion(), CompletionRole::Flush).await
    }

    pub(crate) async fn shutdown_until<D>(mut self, deadline: D) -> Result<(), ManagedWriterError>
    where
        D: Future<Output = ()>,
    {
        // The fixed shutdown command is itself a sync barrier. It is queued
        // after every request finish record and therefore provides the final
        // flush-before-join ordering without a second waiter or retry loop.
        let completion = self
            .sink
            .try_shutdown()
            .map_err(|_| ManagedWriterError::Submit)?;
        let shutdown = CompletionWait::new(completion.into_completion(), CompletionRole::Shutdown);
        tokio::pin!(shutdown);
        tokio::pin!(deadline);
        tokio::select! {
            biased;
            () = &mut deadline => std::process::abort(),
            result = &mut shutdown => result?,
        }
        let writer = self.writer.take().ok_or(ManagedWriterError::ThreadPanic)?;
        let result = writer.join().map_err(|_| ManagedWriterError::ThreadPanic)?;
        result.map_err(ManagedWriterError::Completion)?;
        self.shutdown_complete = true;
        WRITER_LEASED.store(false, Ordering::Release);
        Ok(())
    }
}

/// Started-but-unpublished writer owner. It is created only from the paired
/// seventh queue owner and the physical provider half; no listener authority
/// is involved in preparation.
#[doc(hidden)]
pub struct StartupObservabilityOwner {
    writer: Mutex<Option<ProductionManagedFileWriter>>,
    pair_authority: Mutex<Option<ObservabilityQueuePairAuthority>>,
    shutdown_budget: std::time::Duration,
}

impl StartupObservabilityOwner {
    pub(crate) async fn prepare(
        physical: NormalizedSignedProviderFilesystemRuntimeEvidence,
        queue: PrepairObservabilityQueueDomainOwner,
        pair_authority: ObservabilityQueuePairAuthority,
        shutdown_budget: std::time::Duration,
    ) -> Result<Self, ManagedWriterError> {
        let binding = pair_authority.writer_startup_binding();
        let layout = fixed_core_layout::<
            PRODUCTION_BLOCKS,
            PRODUCTION_BYTES,
            PRODUCTION_COMMANDS,
            PRODUCTION_PATH,
            PRODUCTION_DIRENT,
            PRODUCTION_ENTRIES,
        >()
        .map_err(ManagedWriterError::Completion)?;
        let prepared = prepare_signed_provider_fixed_file_core(
            queue.into_domain(),
            layout.total,
            FixedFileLimits::candidate(),
            physical,
            fixed_notifications(),
        )
        .map_err(ManagedWriterError::Completion)?;
        let writer = ProductionManagedFileWriter::start_until(
            prepared,
            binding,
            tokio::time::sleep(shutdown_budget),
        )
        .await?;
        Ok(Self {
            writer: Mutex::new(Some(writer)),
            pair_authority: Mutex::new(Some(pair_authority)),
            shutdown_budget,
        })
    }

    pub(crate) fn take_pair_authority(
        &self,
    ) -> Result<ObservabilityQueuePairAuthority, ManagedWriterError> {
        let writer = self
            .writer
            .lock()
            .map_err(|_| ManagedWriterError::NotificationUnhealthy)?;
        let writer = writer.as_ref().ok_or(ManagedWriterError::ThreadPanic)?;
        if writer.sink.verify().health() != CorrelationHealth::Accepting {
            return Err(ManagedWriterError::NotificationUnhealthy);
        }
        self.pair_authority
            .lock()
            .map_err(|_| ManagedWriterError::NotificationUnhealthy)?
            .take()
            .ok_or(ManagedWriterError::NotificationUnhealthy)
    }

    pub(crate) async fn shutdown_prepared(&self) -> Result<(), ManagedWriterError> {
        let writer = self
            .writer
            .lock()
            .map_err(|_| ManagedWriterError::NotificationUnhealthy)?
            .take();
        let Some(writer) = writer else {
            return Err(ManagedWriterError::ThreadPanic);
        };
        writer
            .shutdown_until(tokio::time::sleep(self.shutdown_budget))
            .await
    }
}

impl ComponentLifecycle for StartupObservabilityOwner {
    fn name(&self) -> &'static str {
        "observability"
    }

    fn start(&self) -> LifecycleFuture<'_> {
        Box::pin(async { Ok(()) })
    }

    fn shutdown(&self) -> LifecycleFuture<'_> {
        Box::pin(async move {
            self.shutdown_prepared()
                .await
                .map_err(|_| writer_lifecycle_error())
        })
    }
}

fn writer_lifecycle_error() -> SaddleError {
    SaddleError::new(
        ErrorKind::Internal,
        "runtime.observability_writer_lifecycle_failed",
        "managed observability writer lifecycle failed",
    )
}

impl Drop for ManagedFileWriter {
    fn drop(&mut self) {
        if !self.shutdown_complete {
            std::process::abort();
        }
    }
}