saddle_observability/

logger.rs

use std::{
    error::Error,
    fmt,
    future::Future,
    io::{self, Write},
    pin::Pin,
    sync::{
        Arc, Mutex, OnceLock,
        atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering},
        mpsc::{self, Receiver, SyncSender, TrySendError},
    },
    task::{Context, Poll, Waker},
    thread::{self, JoinHandle},
    time::{SystemTime, UNIX_EPOCH},
};

use saddle_core::{ComponentLifecycle, ErrorKind, LifecycleFuture, SaddleError, SpanId, TraceId};
use serde::Serialize;

use crate::{
    calendar_writer::{CalendarFileWriter, FileLoggingConfig},
    event::EventLevel,
};

static GLOBAL: OnceLock<Observer> = OnceLock::new();

/// Configuration for the bounded non-blocking logger.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct ObserverConfig {
    /// Maximum records waiting for the output worker. New records are dropped
    /// when the queue is full rather than blocking an async request.
    pub queue_capacity: usize,
}

impl Default for ObserverConfig {
    fn default() -> Self {
        Self {
            queue_capacity: 8_192,
        }
    }
}

#[derive(Debug)]
pub enum InitError {
    EmptyQueue,
    RandomSource,
    Output(io::Error),
    Spawn(io::Error),
}

impl fmt::Display for InitError {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::EmptyQueue => formatter.write_str("log queue capacity must be greater than zero"),
            Self::RandomSource => {
                formatter.write_str("operating system random source is unavailable")
            }
            Self::Output(error) => write!(formatter, "failed to initialize log output: {error}"),
            Self::Spawn(error) => write!(formatter, "failed to start log writer: {error}"),
        }
    }
}

impl Error for InitError {
    fn source(&self) -> Option<&(dyn Error + 'static)> {
        match self {
            Self::Spawn(error) | Self::Output(error) => Some(error),
            Self::EmptyQueue | Self::RandomSource => None,
        }
    }
}

/// The output operation whose first failure made log delivery unhealthy.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum OutputStage {
    Serialize,
    Record,
    Newline,
    Flush,
}

/// A flush or managed shutdown result that is safe to expose to framework code.
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum FlushError {
    OutputFailed(OutputStage),
    DroppedEvents(u64),
    OutputFailedAndDropped {
        stage: OutputStage,
        dropped_events: u64,
    },
    WriterStopped,
    AlreadyShuttingDown,
    CoordinatorUnavailable,
    WorkerPanicked,
}

impl fmt::Display for FlushError {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::OutputFailed(stage) => write!(formatter, "log output failed during {stage:?}"),
            Self::DroppedEvents(count) => write!(formatter, "{count} log events were dropped"),
            Self::OutputFailedAndDropped {
                stage,
                dropped_events,
            } => write!(
                formatter,
                "log output failed during {stage:?} and {dropped_events} events were dropped"
            ),
            Self::WriterStopped => formatter.write_str("log writer is not running"),
            Self::AlreadyShuttingDown => formatter.write_str("log writer shutdown already started"),
            Self::CoordinatorUnavailable => {
                formatter.write_str("failed to start log lifecycle coordinator")
            }
            Self::WorkerPanicked => formatter.write_str("log writer thread panicked"),
        }
    }
}

impl Error for FlushError {}

/// A non-blocking future returned by flush and shutdown lifecycle operations.
pub struct FlushFuture {
    completion: Arc<Completion>,
}

impl Future for FlushFuture {
    type Output = Result<(), FlushError>;

    fn poll(self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<Self::Output> {
        if let Some(result) = self.completion.result.lock().unwrap().take() {
            return Poll::Ready(result);
        }
        *self.completion.waker.lock().unwrap() = Some(context.waker().clone());
        if let Some(result) = self.completion.result.lock().unwrap().take() {
            Poll::Ready(result)
        } else {
            Poll::Pending
        }
    }
}

struct Completion {
    result: Mutex<Option<Result<(), FlushError>>>,
    waker: Mutex<Option<Waker>>,
}

impl Completion {
    fn pending() -> Arc<Self> {
        Arc::new(Self {
            result: Mutex::new(None),
            waker: Mutex::new(None),
        })
    }

    fn ready(result: Result<(), FlushError>) -> Arc<Self> {
        Arc::new(Self {
            result: Mutex::new(Some(result)),
            waker: Mutex::new(None),
        })
    }

    fn complete(&self, result: Result<(), FlushError>) {
        *self.result.lock().unwrap() = Some(result);
        if let Some(waker) = self.waker.lock().unwrap().take() {
            waker.wake();
        }
    }
}

/// Process-level structured logger and trace correlator.
///
/// Cloning this value is cheap. Emission uses `try_send`; output and JSON
/// serialization happen on the dedicated writer thread.
#[derive(Clone)]
pub struct Observer {
    pub(crate) inner: Arc<Inner>,
}

pub(crate) struct Inner {
    sender: SyncSender<Command>,
    dropped: AtomicU64,
    accepting: AtomicBool,
    emitting: AtomicUsize,
    shutdown_started: AtomicBool,
    worker: Mutex<Option<JoinHandle<()>>>,
    process_log: Mutex<Option<saddle_admission::ProcessLogStorage>>,
    ids: IdGenerator,
    pub(crate) metrics: crate::metrics::Metrics,
}

struct IdGenerator {
    trace_high: u64,
    trace_seed: u64,
    trace_counter: AtomicU64,
    span_seed: u64,
    span_counter: AtomicU64,
}

impl IdGenerator {
    fn from_seed(seed: [u8; 24]) -> Self {
        let mut high = u64::from_be_bytes(seed[0..8].try_into().unwrap());
        if high == 0 {
            high = 1;
        }
        Self {
            trace_high: high,
            trace_seed: u64::from_be_bytes(seed[8..16].try_into().unwrap()),
            trace_counter: AtomicU64::new(0),
            span_seed: u64::from_be_bytes(seed[16..24].try_into().unwrap()),
            span_counter: AtomicU64::new(0),
        }
    }

    fn trace_id(&self) -> TraceId {
        let low = self
            .trace_seed
            .wrapping_add(self.trace_counter.fetch_add(1, Ordering::Relaxed));
        TraceId::from_u128((u128::from(self.trace_high) << 64) | u128::from(low))
    }

    fn span_id(&self) -> SpanId {
        loop {
            let value = self
                .span_seed
                .wrapping_add(self.span_counter.fetch_add(1, Ordering::Relaxed));
            if value != 0 {
                return SpanId::from_u64(value);
            }
        }
    }
}

enum Command {
    Record(LogRecord),
    ProcessCall {
        bytes: saddle_admission::ExactStored<Vec<u8>>,
        dropped: u64,
    },
    // Source-encoded, bounded and self-contained. Never contains a request root.
    RootRecord {
        bytes: Vec<u8>,
        dropped: u64,
    },
    Flush {
        unreported_dropped: u64,
        response: mpsc::Sender<WorkerStatus>,
    },
    Shutdown {
        unreported_dropped: u64,
        response: mpsc::Sender<WorkerStatus>,
    },
}
pub(crate) fn root_queue_layout() -> std::alloc::Layout {
    std::alloc::Layout::new::<Command>()
}

#[derive(Clone, Copy, Debug, Default)]
struct WorkerStatus {
    failure: Option<OutputStage>,
    unreported_dropped: u64,
}

impl WorkerStatus {
    fn into_result(self) -> Result<(), FlushError> {
        match (self.failure, self.unreported_dropped) {
            (None, 0) => Ok(()),
            (Some(stage), 0) => Err(FlushError::OutputFailed(stage)),
            (None, dropped_events) => Err(FlushError::DroppedEvents(dropped_events)),
            (Some(stage), dropped_events) => Err(FlushError::OutputFailedAndDropped {
                stage,
                dropped_events,
            }),
        }
    }
}

#[derive(Serialize)]
pub(crate) struct LogRecord {
    pub timestamp_unix_ms: u128,
    pub level: EventLevel,
    pub event: &'static str,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub trace_id: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub span: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub span_id: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub parent: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub parent_span_id: Option<String>,
    #[serde(flatten)]
    pub data: serde_json::Map<String, serde_json::Value>,
    #[serde(skip_serializing_if = "is_zero")]
    pub dropped_events: u64,
}

const fn is_zero(value: &u64) -> bool {
    *value == 0
}

impl LogRecord {
    pub(crate) fn new(level: EventLevel, event: &'static str) -> Self {
        Self {
            timestamp_unix_ms: SystemTime::now()
                .duration_since(UNIX_EPOCH)
                .unwrap_or_default()
                .as_millis(),
            level,
            event,
            trace_id: None,
            span: None,
            span_id: None,
            parent: None,
            parent_span_id: None,
            data: serde_json::Map::new(),
            dropped_events: 0,
        }
    }
}

/// Initializes stdout logging once. Later calls return the first observer.
pub fn init(config: ObserverConfig) -> Result<&'static Observer, InitError> {
    if let Some(observer) = GLOBAL.get() {
        return Ok(observer);
    }

    let observer = Observer::with_writer(config, io::stdout())?;
    let _ = GLOBAL.set(observer);
    Ok(GLOBAL.get().expect("global observer was initialized"))
}

/// Initializes the process logger with the sole calendar-rotated file sink.
pub fn init_file(
    config: ObserverConfig,
    file: FileLoggingConfig,
) -> Result<&'static Observer, InitError> {
    if let Some(observer) = GLOBAL.get() {
        return Ok(observer);
    }
    let writer = CalendarFileWriter::open(file).map_err(InitError::Output)?;
    let observer = Observer::with_writer(config, writer)?;
    let _ = GLOBAL.set(observer);
    Ok(GLOBAL.get().expect("global observer was initialized"))
}

/// Returns the initialized process observer, if application startup installed it.
pub fn global() -> Option<&'static Observer> {
    GLOBAL.get()
}

impl Observer {
    /// Bind the original process framework domain before formal routed RPCs
    /// can emit ordinary call records. The writer never receives this handle.
    #[doc(hidden)]
    pub fn install_process_log_storage(&self, storage: saddle_admission::ProcessLogStorage) {
        let mut slot = self.inner.process_log.lock().unwrap_or_else(|p| p.into_inner());
        *slot = Some(storage);
    }

    pub(crate) fn emit_process_call(&self, record: &mut crate::call::ProcessCallRecord<'_>) {
        struct Counter(usize);
        impl Write for Counter {
            fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
                self.0 = self.0.checked_add(bytes.len()).ok_or(io::ErrorKind::OutOfMemory)?;
                Ok(bytes.len())
            }
            fn flush(&mut self) -> io::Result<()> { Ok(()) }
        }
        self.inner.emitting.fetch_add(1, Ordering::AcqRel);
        let submitted = (|| {
            if !self.inner.accepting.load(Ordering::Acquire) { return false; }
            let storage = self.inner.process_log.lock().unwrap_or_else(|p| p.into_inner()).clone();
            let Some(storage) = storage else { return false; };
            let dropped = self.inner.dropped.swap(0, Ordering::Relaxed);
            record.dropped_events = dropped;
            let mut counter = Counter(0);
            if serde_json::to_writer(&mut counter, record).is_err() {
                self.inner.dropped.fetch_add(dropped, Ordering::Relaxed);
                return false;
            }
            let Ok(layout) = std::alloc::Layout::array::<u8>(counter.0) else {
                self.inner.dropped.fetch_add(dropped, Ordering::Relaxed);
                return false;
            };
            let Ok(permit) = storage.try_reserve(layout) else {
                self.inner.dropped.fetch_add(dropped, Ordering::Relaxed);
                return false;
            };
            let mut bytes = permit.allocate_exact(layout, || Vec::with_capacity(counter.0));
            if serde_json::to_writer(bytes.get_mut(), record).is_err() || bytes.get().len() != counter.0 {
                drop(bytes);
                self.inner.dropped.fetch_add(dropped, Ordering::Relaxed);
                return false;
            }
            match self.inner.sender.try_send(Command::ProcessCall { bytes, dropped }) {
                Ok(()) => true,
                Err(TrySendError::Full(command)) => {
                    drop(command);
                    self.inner.dropped.fetch_add(dropped, Ordering::Relaxed);
                    false
                }
                Err(TrySendError::Disconnected(command)) => {
                    drop(command);
                    self.inner.metrics.logger_output(true);
                    self.inner.dropped.fetch_add(dropped, Ordering::Relaxed);
                    false
                }
            }
        })();
        if !submitted {
            self.inner.metrics.logger_dropped(1);
            self.inner.dropped.fetch_add(1, Ordering::Relaxed);
        }
        self.inner.emitting.fetch_sub(1, Ordering::Release);
    }
    /// Creates an observer with a framework-owned writer.
    ///
    /// Application startup should normally use [`init`]. This constructor is
    /// useful for embedding and deterministic tests.
    pub fn with_writer(
        config: ObserverConfig,
        writer: impl Write + Send + 'static,
    ) -> Result<Self, InitError> {
        let mut seed = [0_u8; 24];
        getrandom::fill(&mut seed).map_err(|_| InitError::RandomSource)?;
        Self::with_writer_and_seed(config, writer, Ok(seed))
    }

    fn with_writer_and_seed(
        config: ObserverConfig,
        writer: impl Write + Send + 'static,
        seed: Result<[u8; 24], InitError>,
    ) -> Result<Self, InitError> {
        if config.queue_capacity == 0 {
            return Err(InitError::EmptyQueue);
        }
        let ids = IdGenerator::from_seed(seed?);
        let (sender, receiver) = mpsc::sync_channel(config.queue_capacity);
        let worker = thread::Builder::new()
            .name("saddle-log-writer".to_owned())
            .spawn(move || write_records(receiver, writer))
            .map_err(InitError::Spawn)?;
        Ok(Self {
            inner: Arc::new(Inner {
                sender,
                dropped: AtomicU64::new(0),
                accepting: AtomicBool::new(true),
                emitting: AtomicUsize::new(0),
                shutdown_started: AtomicBool::new(false),
                worker: Mutex::new(Some(worker)),
                process_log: Mutex::new(None),
                ids,
                metrics: crate::metrics::Metrics::default(),
            }),
        })
    }

    pub(crate) fn new_trace_id(&self) -> TraceId {
        self.inner.ids.trace_id()
    }

    pub(crate) fn new_span_id(&self) -> SpanId {
        self.inner.ids.span_id()
    }

    pub(crate) fn emit(&self, mut record: LogRecord) {
        if !self.inner.accepting.load(Ordering::Acquire) {
            return;
        }
        self.inner.emitting.fetch_add(1, Ordering::AcqRel);
        if !self.inner.accepting.load(Ordering::Acquire) {
            self.inner.emitting.fetch_sub(1, Ordering::AcqRel);
            return;
        }

        record.dropped_events = self.inner.dropped.swap(0, Ordering::Relaxed);
        match self.inner.sender.try_send(Command::Record(record)) {
            Ok(()) => {}
            Err(TrySendError::Full(Command::Record(record))) => {
                self.inner.metrics.logger_dropped(1);
                self.inner
                    .dropped
                    .fetch_add(record.dropped_events.saturating_add(1), Ordering::Relaxed);
            }
            Err(TrySendError::Disconnected(_)) => {
                self.inner.metrics.logger_dropped(1);
                self.inner.metrics.logger_output(true);
                self.inner.dropped.fetch_add(1, Ordering::Relaxed);
            }
            Err(TrySendError::Full(_)) => unreachable!("emit only sends records"),
        }
        self.inner.emitting.fetch_sub(1, Ordering::Release);
    }

    pub(crate) fn emit_root_record<T: Serialize>(&self, record: &T) -> crate::DiagnosticSubmission {
        use crate::DiagnosticSubmission as Submission;
        self.inner.emitting.fetch_add(1, Ordering::AcqRel);
        let result = (|| {
            if !self.inner.accepting.load(Ordering::Acquire) {
                return Submission::Closed;
            }
            let mut frame = crate::diagnostic::FixedDiagnosticBytes {
                bytes: [0; 8192],
                len: 0,
            };
            if serde_json::to_writer(&mut frame, record).is_err() {
                return Submission::EncodingFailed;
            }
            let dropped = self.inner.dropped.swap(0, Ordering::Relaxed);
            // Only initialized bytes move into the existing logger domain. This
            // allocation and source frame peak are separate R0 layout inputs.
            let bytes = frame.bytes[..frame.len].to_vec();
            match self
                .inner
                .sender
                .try_send(Command::RootRecord { bytes, dropped })
            {
                Ok(()) => Submission::Enqueued,
                Err(TrySendError::Full(Command::RootRecord { dropped, .. })) => {
                    self.inner.dropped.fetch_add(dropped, Ordering::Relaxed);
                    Submission::Full
                }
                Err(TrySendError::Disconnected(_)) => {
                    self.inner.dropped.fetch_add(dropped, Ordering::Relaxed);
                    self.inner.metrics.logger_output(true);
                    Submission::Closed
                }
                Err(TrySendError::Full(_)) => unreachable!("root record submission"),
            }
        })();
        if result != Submission::Enqueued {
            self.inner.dropped.fetch_add(1, Ordering::Relaxed);
            self.inner.metrics.logger_dropped(1);
        }
        self.inner.emitting.fetch_sub(1, Ordering::Release);
        result
    }

    /// Returns a fixed-size, allocation-free copy of the current process metrics.
    pub fn metrics_snapshot(&self) -> crate::MetricsSnapshot {
        self.inner.metrics.snapshot()
    }

    /// Flushes records accepted before the lifecycle coordinator reaches the
    /// writer. The returned future never blocks the async executor thread.
    pub fn flush(&self) -> FlushFuture {
        if !self.inner.accepting.load(Ordering::Acquire) {
            return FlushFuture {
                completion: Completion::ready(Err(FlushError::WriterStopped)),
            };
        }
        let completion = Completion::pending();
        let future = FlushFuture {
            completion: completion.clone(),
        };
        let inner = self.inner.clone();
        if thread::Builder::new()
            .name("saddle-log-flush".to_owned())
            .spawn(move || coordinate_flush(inner, completion.clone()))
            .is_err()
        {
            future
                .completion
                .complete(Err(FlushError::CoordinatorUnavailable));
        }
        future
    }

    /// Stops admission, drains accepted records, flushes output and joins the
    /// writer without blocking the async executor thread.
    pub fn shutdown(&self) -> FlushFuture {
        if self.inner.shutdown_started.swap(true, Ordering::AcqRel) {
            return FlushFuture {
                completion: Completion::ready(Err(FlushError::AlreadyShuttingDown)),
            };
        }
        self.inner.accepting.store(false, Ordering::Release);
        let completion = Completion::pending();
        let future = FlushFuture {
            completion: completion.clone(),
        };
        let inner = self.inner.clone();
        if thread::Builder::new()
            .name("saddle-log-shutdown".to_owned())
            .spawn(move || coordinate_shutdown(inner, completion.clone()))
            .is_err()
        {
            self.inner.accepting.store(true, Ordering::Release);
            self.inner.shutdown_started.store(false, Ordering::Release);
            future
                .completion
                .complete(Err(FlushError::CoordinatorUnavailable));
        }
        future
    }

    pub fn dropped_events(&self) -> u64 {
        self.inner.dropped.load(Ordering::Relaxed)
    }
}

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

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

    fn shutdown(&self) -> LifecycleFuture<'_> {
        let shutdown = Observer::shutdown(self);
        Box::pin(async move {
            shutdown.await.map_err(|_| {
                SaddleError::new(
                    ErrorKind::Infrastructure,
                    "observability.shutdown_failed",
                    "structured log shutdown failed",
                )
            })
        })
    }
}

fn coordinate_flush(inner: Arc<Inner>, completion: Arc<Completion>) {
    let dropped = inner.dropped.swap(0, Ordering::AcqRel);
    let (response, receiver) = mpsc::channel();
    let result = if inner
        .sender
        .send(Command::Flush {
            unreported_dropped: dropped,
            response,
        })
        .is_err()
    {
        Err(FlushError::WriterStopped)
    } else {
        receiver
            .recv()
            .map_err(|_| FlushError::WriterStopped)
            .and_then(WorkerStatus::into_result)
    };
    completion.complete(result);
}

fn coordinate_shutdown(inner: Arc<Inner>, completion: Arc<Completion>) {
    while inner.emitting.load(Ordering::Acquire) != 0 {
        thread::yield_now();
    }
    let dropped = inner.dropped.swap(0, Ordering::AcqRel);
    let (response, receiver) = mpsc::channel();
    let mut result = if inner
        .sender
        .send(Command::Shutdown {
            unreported_dropped: dropped,
            response,
        })
        .is_err()
    {
        Err(FlushError::WriterStopped)
    } else {
        receiver
            .recv()
            .map_err(|_| FlushError::WriterStopped)
            .and_then(WorkerStatus::into_result)
    };

    if let Some(worker) = inner.worker.lock().unwrap().take() {
        if worker.join().is_err() {
            result = Err(FlushError::WorkerPanicked);
        }
    }
    inner.process_log.lock().unwrap_or_else(|p| p.into_inner()).take();
    completion.complete(result);
}

fn write_records(receiver: Receiver<Command>, mut writer: impl Write) {
    let mut status = WorkerStatus::default();
    while let Ok(command) = receiver.recv() {
        match command {
            Command::Record(record) => write_record(&mut writer, record, &mut status),
            Command::ProcessCall { bytes, dropped } => {
                status.unreported_dropped = status.unreported_dropped.saturating_add(dropped);
                if status.failure.is_none() {
                    if writer.write_all(bytes.get()).is_err() {
                        status.failure = Some(OutputStage::Record);
                    } else if writer.write_all(b"\n").is_err() {
                        status.failure = Some(OutputStage::Newline);
                    }
                }
                drop(bytes);
            }
            Command::RootRecord { bytes, dropped } => {
                status.unreported_dropped = status.unreported_dropped.saturating_add(dropped);
                if status.failure.is_none() {
                    if writer.write_all(&bytes).is_err() {
                        status.failure = Some(OutputStage::Record);
                    } else if writer.write_all(b"\n").is_err() {
                        status.failure = Some(OutputStage::Newline);
                    }
                }
            }
            Command::Flush {
                unreported_dropped,
                response,
            } => {
                status.unreported_dropped =
                    status.unreported_dropped.saturating_add(unreported_dropped);
                flush_writer(&mut writer, &mut status);
                let _ = response.send(status);
            }
            Command::Shutdown {
                unreported_dropped,
                response,
            } => {
                status.unreported_dropped =
                    status.unreported_dropped.saturating_add(unreported_dropped);
                flush_writer(&mut writer, &mut status);
                let _ = response.send(status);
                break;
            }
        }
    }
}

fn write_record(writer: &mut impl Write, record: LogRecord, status: &mut WorkerStatus) {
    if status.failure.is_some() {
        status.unreported_dropped = status
            .unreported_dropped
            .saturating_add(record.dropped_events);
        return;
    }

    let dropped = record.dropped_events;
    let bytes = match serde_json::to_vec(&record) {
        Ok(bytes) => bytes,
        Err(_) => {
            status.failure = Some(OutputStage::Serialize);
            status.unreported_dropped = status.unreported_dropped.saturating_add(dropped);
            return;
        }
    };
    if writer.write_all(&bytes).is_err() {
        status.failure = Some(OutputStage::Record);
        status.unreported_dropped = status.unreported_dropped.saturating_add(dropped);
        return;
    }
    if writer.write_all(b"\n").is_err() {
        status.failure = Some(OutputStage::Newline);
        status.unreported_dropped = status.unreported_dropped.saturating_add(dropped);
    }
}

fn flush_writer(writer: &mut impl Write, status: &mut WorkerStatus) {
    if writer.flush().is_err() && status.failure.is_none() {
        status.failure = Some(OutputStage::Flush);
    }
}

#[cfg(test)]
mod tests {
    use std::{
        sync::{Condvar, mpsc},
        task::{Wake, Waker},
    };

    use super::*;

    struct ThreadWaker(thread::Thread);

    impl Wake for ThreadWaker {
        fn wake(self: Arc<Self>) {
            self.0.unpark();
        }
    }

    fn block_on<T>(future: impl Future<Output = T>) -> T {
        let waker = Waker::from(Arc::new(ThreadWaker(thread::current())));
        let mut context = Context::from_waker(&waker);
        let mut future = std::pin::pin!(future);
        loop {
            match future.as_mut().poll(&mut context) {
                Poll::Ready(output) => return output,
                Poll::Pending => thread::park(),
            }
        }
    }

    struct BlockingWriter {
        entered: Option<mpsc::Sender<()>>,
        release: Arc<(Mutex<bool>, Condvar)>,
        fail_after_release: bool,
    }

    impl Write for BlockingWriter {
        fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
            if let Some(entered) = self.entered.take() {
                let _ = entered.send(());
                let (lock, condition) = &*self.release;
                let mut released = lock.lock().unwrap();
                while !*released {
                    released = condition.wait(released).unwrap();
                }
                if self.fail_after_release {
                    return Err(io::Error::other("injected writer failure"));
                }
            }
            Ok(bytes.len())
        }

        fn flush(&mut self) -> io::Result<()> {
            Ok(())
        }
    }

    struct FailingWriter {
        fail_write: Option<usize>,
        writes: usize,
        fail_flush: bool,
    }

    impl Write for FailingWriter {
        fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
            self.writes += 1;
            if self.fail_write == Some(self.writes) {
                Err(io::Error::other("injected writer failure"))
            } else {
                Ok(bytes.len())
            }
        }

        fn flush(&mut self) -> io::Result<()> {
            if self.fail_flush {
                Err(io::Error::other("injected flush failure"))
            } else {
                Ok(())
            }
        }
    }

    fn observer(writer: impl Write + Send + 'static, capacity: usize) -> Observer {
        Observer::with_writer_and_seed(
            ObserverConfig {
                queue_capacity: capacity,
            },
            writer,
            Ok([7; 24]),
        )
        .unwrap()
    }

    #[test]
    fn managed_call_log_outlives_request_account_while_writer_is_blocked() {
        // The new process-owned variant adds no command-ring slot bytes:
        // the unchanged ordinary LogRecord still determines Command's size.
        assert_eq!(std::mem::size_of::<Command>(), std::mem::size_of::<LogRecord>());
        assert_eq!(std::mem::size_of::<Command>(), 192);
        assert_eq!(ObserverConfig::default().queue_capacity * (8 + std::mem::size_of::<Command>()),
            1_638_400);
        use saddle_admission::{
            ProfuseGwLightweightObservedAdmissionOutcome, StorageDemand,
            bind_deployment_resource_budget_bootstrap, freeze_deployment_resource_budget,
            prepare_profusegw_lightweight_profile,
        };
        use saddle_core::{
            BootstrapRendezvousIssuer, GeneratedApplicationFreezeSource,
            ListenerStartupFreezeSource, RpcCorrelationId, pair_bootstrap_rendezvous,
        };
        let pending = freeze_deployment_resource_budget(
            1, 32, 5_000, 1, 1, 1_000_000, 1_000_000, 1_000_000, 1_000_000,
        ).unwrap().reserve_known_database_memory(0, 0, 0).ok().unwrap();
        let (application, listener) = BootstrapRendezvousIssuer::issue()
            .freeze_application(GeneratedApplicationFreezeSource::new(
                "app", b"descriptor", &["route"],
            )).unwrap();
        let listener = listener.freeze_listener(ListenerStartupFreezeSource::new(
            "app", "127.0.0.1:8000".parse().unwrap(),
            "127.0.0.1:9000".parse().unwrap(),
            std::time::Duration::from_millis(5_000),
        )).ok().unwrap();
        let (whole, receipt) = pair_bootstrap_rendezvous(application, listener).ok().unwrap();
        let budget = bind_deployment_resource_budget_bootstrap(pending, whole, receipt).ok().unwrap();
        let process = prepare_profusegw_lightweight_profile(budget).ok().unwrap();
        let root = process.try_process_storage(StorageDemand::separate(&[(
            std::alloc::Layout::array::<u8>(16).unwrap(), 1,
        )]).unwrap()).unwrap();
        let baseline = process.resource_snapshot().framework_charged;
        let (entered, entered_rx) = mpsc::channel();
        let release = Arc::new((Mutex::new(false), Condvar::new()));
        let logger = observer(BlockingWriter {
            entered: Some(entered), release: Arc::clone(&release), fail_after_release: false,
        }, 2);
        logger.install_process_log_storage(process.process_log_storage());
        let (call, _) = logger.start_managed_external_call_with_rpc(
            "app", "zone", "interface", "operation", None,
            RpcCorrelationId::new("0").unwrap(),
        ).unwrap();
        entered_rx.recv_timeout(std::time::Duration::from_secs(2)).unwrap();
        let admitted = match process.verified_profile().try_admit_observed() {
            ProfuseGwLightweightObservedAdmissionOutcome::Ready(permit, _) => permit,
            _ => panic!("fixture admission failed"),
        };
        let original = saddle_core::SaddleError::new(
            saddle_core::ErrorKind::Internal, "original.failure", "original",
        );
        call.fail(&original);
        assert_eq!(original.code(), "original.failure");
        admitted.into_execution().cancel_observed();
        assert_eq!(process.resource_snapshot().active_accounts, 0);
        assert!(process.resource_snapshot().framework_charged > baseline,
            "queued process log must remain charged after request cleanup");
        // The worker holds the first record while the second is queued.
        // Additional managed call records must be dropped synchronously.
        let (full, _) = logger.start_managed_external_call_with_rpc(
            "app", "zone", "interface", "full", None,
            RpcCorrelationId::new("1").unwrap(),
        ).unwrap();
        full.fail(&original);
        assert!(logger.dropped_events() > 0);
        *release.0.lock().unwrap() = true;
        release.1.notify_all();
        assert!(matches!(block_on(logger.flush()), Err(FlushError::DroppedEvents(_))));
        assert_eq!(process.resource_snapshot().framework_charged, baseline);

        // Exhaust the *same* framework domain with a physical child owner;
        // encoding must fail to reserve before allocating and remain lossy.
        let available = process.resource_snapshot().framework_capacity
            - process.resource_snapshot().framework_charged;
        let filler = process.try_test_process_storage_child(StorageDemand::separate(&[(
            std::alloc::Layout::array::<u8>(
                available - 64 - std::mem::size_of::<saddle_admission::StoragePermit>(),
            ).unwrap(), 1,
        )]).unwrap()).unwrap();
        let charged = process.resource_snapshot().framework_charged;
        let before_drop = logger.dropped_events();
        let (short, _) = logger.start_managed_external_call_with_rpc(
            "app", "zone", "interface", "short", None,
            RpcCorrelationId::new("2").unwrap(),
        ).unwrap();
        short.fail(&original);
        assert!(logger.dropped_events() >= before_drop + 2);
        assert_eq!(process.resource_snapshot().framework_charged, charged);
        drop(filler);
        assert!(matches!(block_on(logger.shutdown()), Err(FlushError::DroppedEvents(_))));
        let after_shutdown = process.resource_snapshot().framework_charged;
        let (closed, _) = logger.start_managed_external_call_with_rpc(
            "app", "zone", "interface", "closed", None,
            RpcCorrelationId::new("3").unwrap(),
        ).unwrap();
        closed.fail(&original);
        assert_eq!(process.resource_snapshot().framework_charged, after_shutdown);
        assert_eq!(process.resource_snapshot().framework_charged, baseline);
        drop(logger);
        drop(root);
        process.finish().unwrap();
    }

    #[test]
    fn root_contract_ordinary_uses_same_view_and_bounded_queue() {
        use crate::{
            DiagnosticSubmission, RootDiagnosticScope, RootOutcomeFacts, RootRequestEvent,
        };
        use saddle_core::{
            ContextFact, ContextLabel, RequestLocalFacts, RequestRootPublisher, RequestViewPhase,
        };
        let root = RequestRootPublisher::create(
            ContextLabel::checked("app").unwrap(),
            ContextFact::NotEstablished,
        )
        .unwrap();
        let view = root
            .reference()
            .view(RequestLocalFacts::new(RequestViewPhase::Handler));
        let expected = serde_json::to_value(&view).unwrap();
        let mut frame = crate::diagnostic::FixedDiagnosticBytes {
            bytes: [0; 8192],
            len: 0,
        };
        serde_json::to_writer(&mut frame, &view).unwrap();
        let ordinary_context: serde_json::Value =
            serde_json::from_slice(&frame.bytes[..frame.len]).unwrap();
        assert_eq!(ordinary_context, expected);
        let (entered, wait) = mpsc::channel();
        let release = Arc::new((Mutex::new(false), Condvar::new()));
        struct Captured {
            writer: BlockingWriter,
            bytes: Arc<Mutex<Vec<u8>>>,
        }
        impl Write for Captured {
            fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
                let n = self.writer.write(bytes)?;
                self.bytes.lock().unwrap().extend_from_slice(&bytes[..n]);
                Ok(n)
            }
            fn flush(&mut self) -> io::Result<()> {
                self.writer.flush()
            }
        }
        struct Unblock(Arc<(Mutex<bool>, Condvar)>);
        impl Drop for Unblock {
            fn drop(&mut self) {
                *self.0.0.lock().unwrap() = true;
                self.0.1.notify_all();
            }
        }
        let unblock = Unblock(Arc::clone(&release));
        let bytes = Arc::new(Mutex::new(Vec::new()));
        let logger = observer(
            Captured {
                writer: BlockingWriter {
                    entered: Some(entered),
                    release: Arc::clone(&release),
                    fail_after_release: false,
                },
                bytes: Arc::clone(&bytes),
            },
            1,
        );
        let scope = RootDiagnosticScope::new(&view, None);
        assert_eq!(
            scope.ordinary(
                &logger,
                RootRequestEvent::Handler,
                RootOutcomeFacts::default()
            ),
            DiagnosticSubmission::Enqueued
        );
        wait.recv_timeout(std::time::Duration::from_secs(5))
            .unwrap();
        assert_eq!(
            scope.ordinary(
                &logger,
                RootRequestEvent::Response,
                RootOutcomeFacts::default()
            ),
            DiagnosticSubmission::Enqueued
        );
        assert_eq!(
            scope.ordinary(
                &logger,
                RootRequestEvent::Response,
                RootOutcomeFacts::default()
            ),
            DiagnosticSubmission::Full
        );
        drop((view, root));
        drop(unblock);
        assert!(matches!(
            block_on(logger.shutdown()),
            Err(FlushError::DroppedEvents(1))
        ));
        let bytes = bytes.lock().unwrap();
        let rows: Vec<serde_json::Value> = serde_json::Deserializer::from_slice(&bytes)
            .into_iter()
            .map(Result::unwrap)
            .collect();
        assert_eq!(rows.len(), 2);
        assert!(rows.iter().all(|row| row["context"] == expected));
    }

    #[test]
    fn root_contract_encoded_ordinary_readback_and_failure_use_original_writer() {
        let record = serde_json::json!({"event":"request_stage", "context":{"local_request":5}});
        let (tx, rx) = mpsc::sync_channel(1);
        tx.try_send(Command::RootRecord {
            bytes: serde_json::to_vec(&record).unwrap(),
            dropped: 0,
        })
        .unwrap_or_else(|_| panic!("empty queue"));
        drop(tx);
        let mut bytes = Vec::new();
        write_records(rx, &mut bytes);
        assert_eq!(
            serde_json::from_slice::<serde_json::Value>(&bytes).unwrap(),
            record
        );
        let logger = observer(
            FailingWriter {
                fail_write: Some(1),
                writes: 0,
                fail_flush: false,
            },
            1,
        );
        assert_eq!(
            logger.emit_root_record(&record),
            crate::DiagnosticSubmission::Enqueued
        );
        assert!(matches!(
            block_on(logger.shutdown()),
            Err(FlushError::OutputFailed(OutputStage::Record))
        ));
    }

    #[test]
    fn random_source_failure_is_reported_during_initialization() {
        let result = Observer::with_writer_and_seed(
            ObserverConfig::default(),
            io::sink(),
            Err(InitError::RandomSource),
        );
        assert!(matches!(result, Err(InitError::RandomSource)));
    }

    #[test]
    fn initialized_generator_produces_non_zero_unique_ids_without_more_io() {
        let observer = observer(io::sink(), 8);
        let first_trace = observer.new_trace_id();
        let second_trace = observer.new_trace_id();
        let first_span = observer.new_span_id();
        let second_span = observer.new_span_id();
        assert_ne!(first_trace.as_u128(), 0);
        assert_ne!(first_trace, second_trace);
        assert_ne!(first_span.as_u64(), 0);
        assert_ne!(first_span, second_span);
        block_on(observer.shutdown()).unwrap();
    }

    #[test]
    fn observer_is_a_managed_component() {
        let observer = observer(io::sink(), 8);
        assert_eq!(ComponentLifecycle::name(&observer), "observability");
        block_on(ComponentLifecycle::start(&observer)).unwrap();
        block_on(ComponentLifecycle::shutdown(&observer)).unwrap();
    }

    #[test]
    fn full_queue_is_reported_by_shutdown_without_waiting_in_emit() {
        let (entered_sender, entered_receiver) = mpsc::channel();
        let release = Arc::new((Mutex::new(false), Condvar::new()));
        let writer = BlockingWriter {
            entered: Some(entered_sender),
            release: release.clone(),
            fail_after_release: false,
        };
        let observer = observer(writer, 1);

        observer.emit(LogRecord::new(EventLevel::Info, "test.first"));
        entered_receiver.recv().unwrap();
        observer.emit(LogRecord::new(EventLevel::Info, "test.queued"));
        observer.emit(LogRecord::new(EventLevel::Info, "test.dropped"));
        assert_eq!(observer.dropped_events(), 1);

        let shutdown = observer.shutdown();
        let (lock, condition) = &*release;
        *lock.lock().unwrap() = true;
        condition.notify_one();
        assert_eq!(block_on(shutdown), Err(FlushError::DroppedEvents(1)));
    }

    #[test]
    fn shutdown_reports_writer_failure_and_unreported_drops_together() {
        let (entered_sender, entered_receiver) = mpsc::channel();
        let release = Arc::new((Mutex::new(false), Condvar::new()));
        let writer = BlockingWriter {
            entered: Some(entered_sender),
            release: release.clone(),
            fail_after_release: true,
        };
        let observer = observer(writer, 1);

        observer.emit(LogRecord::new(EventLevel::Info, "test.first"));
        entered_receiver.recv().unwrap();
        observer.emit(LogRecord::new(EventLevel::Info, "test.queued"));
        observer.emit(LogRecord::new(EventLevel::Info, "test.dropped"));
        let shutdown = observer.shutdown();

        let (lock, condition) = &*release;
        *lock.lock().unwrap() = true;
        condition.notify_one();
        assert_eq!(
            block_on(shutdown),
            Err(FlushError::OutputFailedAndDropped {
                stage: OutputStage::Record,
                dropped_events: 1,
            })
        );
    }

    #[test]
    fn record_write_failure_persists_until_shutdown() {
        let observer = observer(
            FailingWriter {
                fail_write: Some(1),
                writes: 0,
                fail_flush: false,
            },
            8,
        );
        observer.emit(LogRecord::new(EventLevel::Info, "test.record"));
        assert_eq!(
            block_on(observer.shutdown()),
            Err(FlushError::OutputFailed(OutputStage::Record))
        );
    }

    #[test]
    fn newline_failure_persists_until_flush() {
        let observer = observer(
            FailingWriter {
                fail_write: Some(2),
                writes: 0,
                fail_flush: false,
            },
            8,
        );
        observer.emit(LogRecord::new(EventLevel::Info, "test.newline"));
        assert_eq!(
            block_on(observer.flush()),
            Err(FlushError::OutputFailed(OutputStage::Newline))
        );
        let _ = block_on(observer.shutdown());
    }

    #[test]
    fn flush_failure_is_reported() {
        let observer = observer(
            FailingWriter {
                fail_write: None,
                writes: 0,
                fail_flush: true,
            },
            8,
        );
        assert_eq!(
            block_on(observer.flush()),
            Err(FlushError::OutputFailed(OutputStage::Flush))
        );
        assert_eq!(
            block_on(observer.shutdown()),
            Err(FlushError::OutputFailed(OutputStage::Flush))
        );
    }
}