saddle-observability 0.2.0-rc.6

/// Fixed-resource queue and single Runtime-managed writer entry.
use std::{
    alloc::Layout,
    sync::{Arc, Condvar, Mutex, MutexGuard},
    time::{Duration, Instant},
};

use saddle_admission::{
    AdmissionError, ObservabilityBlockId, ObservabilityBlockLease, ObservabilityQueueDomain,
    ObservabilityQueuedBlock, RequestMemory,
};

use super::{
    FileStream,
    codec::{CodecError, EncodingTarget, JsonlRecord, encode_jsonl},
    completion::{
        CompletionError, CompletionResult, CompletionRole, CompletionSet, CompletionTicket,
    },
    linux_backend::{BackendError, LinuxFileBackend, ROTATION_RETRIES, WRITE_INTERRUPT_RETRIES},
    termination::{
        FilesystemProofError, VerifiedFilesystemServiceProof, WriterTerminationProof,
        derive_termination_proof, writer_work_proof,
    },
};

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct CompletionSignal {
    pub role: CompletionRole,
    pub generation: u64,
}

/// A caller-owned fixed notification slot.
///
/// The callback is part of the Runtime/Observability production contract: it
/// must be bounded, non-blocking, allocation-free and panic-free. Observability
/// stores exactly three copies prepared before the writer starts; it never
/// allocates a Waker, channel, waiter node or task.
#[derive(Clone, Copy)]
pub struct FixedNotification {
    token: usize,
    notify: fn(usize, CompletionSignal),
}

impl FixedNotification {
    pub const fn new(token: usize, notify: fn(usize, CompletionSignal)) -> Self {
        Self { token, notify }
    }

    fn fire(self, ticket: CompletionTicket) {
        (self.notify)(
            self.token,
            CompletionSignal {
                role: ticket.role(),
                generation: ticket.generation(),
            },
        );
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum FixedFailure {
    Layout,
    Directory,
    Lock,
    Recovery(FileStream),
    Open(FileStream),
    Write(FileStream),
    Sync(FileStream),
    Rotate(FileStream),
    Retention(FileStream),
    FilesystemProof(FilesystemProofError),
    Invariant,
}

#[derive(Debug, Eq, PartialEq)]
pub enum SubmitError {
    Admission(AdmissionError),
    Backpressured,
    CompletionBusy,
    Encoding,
    OutstandingEncoding,
    ShuttingDown,
    Unhealthy(FixedFailure),
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct FixedCoreLayout {
    pub shared_allocation_upper_bound: usize,
    pub payload_blocks: usize,
    pub writer_inline: usize,
    pub backend_inline: usize,
    pub termination_inline: usize,
    pub total: usize,
}

#[derive(Clone, Copy, Debug)]
pub struct FixedFileLimits {
    pub rotate_bytes: u64,
    pub retained_files: usize,
    pub retention_age: Duration,
    pub sync_bytes: u64,
    pub sync_interval: Duration,
    #[cfg(test)]
    pub(super) fail: Option<FixedFailure>,
}

impl FixedFileLimits {
    pub const fn candidate() -> Self {
        Self {
            rotate_bytes: 128 * 1024 * 1024,
            retained_files: 8,
            retention_age: Duration::from_secs(7 * 24 * 60 * 60),
            sync_bytes: 4 * 1024 * 1024,
            sync_interval: Duration::from_secs(1),
            #[cfg(test)]
            fail: None,
        }
    }

    fn valid(self) -> bool {
        self.rotate_bytes != 0
            && self.retained_files != 0
            && self.sync_bytes != 0
            && !self.sync_interval.is_zero()
    }
}

#[derive(Clone, Copy)]
struct PayloadBlock<const BYTES: usize> {
    generation: u64,
    length: usize,
    bytes: [u8; BYTES],
}

impl<const BYTES: usize> PayloadBlock<BYTES> {
    const fn empty() -> Self {
        Self {
            generation: 0,
            length: 0,
            bytes: [0; BYTES],
        }
    }
}

enum CommandState {
    Vacant,
    Encoding(ObservabilityBlockId),
    Cancelled,
    Record {
        stream: FileStream,
        block: ObservabilityBlockId,
        length: usize,
    },
    Flush(CompletionTicket),
    Shutdown(CompletionTicket),
}

struct CommandSlot {
    generation: u64,
    state: CommandState,
    queued: Option<ObservabilityQueuedBlock>,
}

impl CommandSlot {
    const fn vacant() -> Self {
        Self {
            generation: 0,
            state: CommandState::Vacant,
            queued: None,
        }
    }

    fn clear(&mut self) {
        self.state = CommandState::Vacant;
        self.queued = None;
    }
}

struct State<const BLOCKS: usize, const BYTES: usize, const COMMANDS: usize> {
    domain: ObservabilityQueueDomain,
    payloads: [PayloadBlock<BYTES>; BLOCKS],
    commands: [CommandSlot; COMMANDS],
    completions: CompletionSet<FixedFailure>,
    head: usize,
    tail: usize,
    occupied: usize,
    accepting: bool,
    failure: Option<FixedFailure>,
    notification_outbox: [Option<CompletionTicket>; 3],
}

struct Inner<const BLOCKS: usize, const BYTES: usize, const COMMANDS: usize> {
    state: Mutex<State<BLOCKS, BYTES, COMMANDS>>,
    changed: Condvar,
    notifications: [FixedNotification; 3],
}

pub struct FixedFileSink<const BLOCKS: usize, const BYTES: usize, const COMMANDS: usize> {
    inner: Arc<Inner<BLOCKS, BYTES, COMMANDS>>,
}

impl<const BLOCKS: usize, const BYTES: usize, const COMMANDS: usize> Clone
    for FixedFileSink<BLOCKS, BYTES, COMMANDS>
{
    fn clone(&self) -> Self {
        Self {
            inner: Arc::clone(&self.inner),
        }
    }
}

pub(super) struct FixedEncodingLease<const BLOCKS: usize, const BYTES: usize, const COMMANDS: usize>
{
    inner: Arc<Inner<BLOCKS, BYTES, COMMANDS>>,
    admission: Option<ObservabilityBlockLease>,
    block: ObservabilityBlockId,
    command: usize,
    command_generation: u64,
    cursor: usize,
    active: bool,
}

pub struct FixedCompletion<const BLOCKS: usize, const BYTES: usize, const COMMANDS: usize> {
    inner: Arc<Inner<BLOCKS, BYTES, COMMANDS>>,
    ticket: CompletionTicket,
    active: bool,
}

impl<const BLOCKS: usize, const BYTES: usize, const COMMANDS: usize>
    FixedCompletion<BLOCKS, BYTES, COMMANDS>
{
    pub fn result(&self) -> CompletionResult<FixedFailure> {
        lock(&self.inner).completions.result(self.ticket)
    }

    pub fn recycle(mut self) -> Result<(), SubmitError> {
        lock(&self.inner)
            .completions
            .recycle(self.ticket)
            .map_err(|_| SubmitError::CompletionBusy)?;
        self.active = false;
        Ok(())
    }

    pub fn cancel(mut self) -> Result<CompletionResult<FixedFailure>, SubmitError> {
        let mut state = lock(&self.inner);
        let result = state
            .completions
            .cancel(self.ticket)
            .map_err(|_| SubmitError::CompletionBusy)?;
        if matches!(result, CompletionResult::Ready(_)) {
            state
                .completions
                .recycle(self.ticket)
                .map_err(|_| SubmitError::CompletionBusy)?;
        }
        self.active = false;
        Ok(result)
    }
}

impl<const BLOCKS: usize, const BYTES: usize, const COMMANDS: usize> Drop
    for FixedCompletion<BLOCKS, BYTES, COMMANDS>
{
    fn drop(&mut self) {
        if self.active {
            let mut state = lock(&self.inner);
            if matches!(
                state.completions.cancel(self.ticket),
                Ok(CompletionResult::Ready(_))
            ) {
                let _ = state.completions.recycle(self.ticket);
            }
            self.active = false;
        }
    }
}

#[derive(Clone, Copy)]
struct StreamProgress {
    written: u64,
    dirty: u64,
    deadline: Option<Instant>,
}

impl StreamProgress {
    const fn clean() -> Self {
        Self {
            written: 0,
            dirty: 0,
            deadline: None,
        }
    }
}

pub struct FixedWriterEntry<
    const BLOCKS: usize,
    const BYTES: usize,
    const COMMANDS: usize,
    const PATH: usize,
    const DIRENT_BYTES: usize,
    const ENTRIES: usize,
> {
    inner: Arc<Inner<BLOCKS, BYTES, COMMANDS>>,
    directory: [u8; PATH],
    directory_length: usize,
    record: [u8; BYTES],
    streams: [StreamProgress; 4],
    limits: FixedFileLimits,
}

pub struct PreparedFixedFileCore<
    const BLOCKS: usize,
    const BYTES: usize,
    const COMMANDS: usize,
    const PATH: usize,
    const DIRENT_BYTES: usize,
    const ENTRIES: usize,
> {
    pub sink: FixedFileSink<BLOCKS, BYTES, COMMANDS>,
    pub writer: FixedWriterEntry<BLOCKS, BYTES, COMMANDS, PATH, DIRENT_BYTES, ENTRIES>,
    pub startup: FixedCompletion<BLOCKS, BYTES, COMMANDS>,
    pub termination: WriterTerminationProof,
}

pub fn fixed_core_layout<
    const BLOCKS: usize,
    const BYTES: usize,
    const COMMANDS: usize,
    const PATH: usize,
    const DIRENT_BYTES: usize,
    const ENTRIES: usize,
>() -> Result<FixedCoreLayout, FixedFailure> {
    let shared = Layout::new::<Inner<BLOCKS, BYTES, COMMANDS>>()
        .size()
        .checked_add(2 * std::mem::size_of::<usize>())
        .and_then(|value| {
            value.checked_add(std::mem::align_of::<Inner<BLOCKS, BYTES, COMMANDS>>() - 1)
        })
        .ok_or(FixedFailure::Layout)?;
    let payload_blocks = BLOCKS.checked_mul(BYTES).ok_or(FixedFailure::Layout)?;
    let writer_inline = std::mem::size_of::<
        FixedWriterEntry<BLOCKS, BYTES, COMMANDS, PATH, DIRENT_BYTES, ENTRIES>,
    >();
    let backend_inline = std::mem::size_of::<LinuxFileBackend<PATH, DIRENT_BYTES, ENTRIES>>();
    let termination_inline = std::mem::size_of::<WriterTerminationProof>();
    let total = shared
        .checked_add(writer_inline)
        .and_then(|value| value.checked_add(backend_inline))
        .and_then(|value| value.checked_add(termination_inline))
        .ok_or(FixedFailure::Layout)?;
    Ok(FixedCoreLayout {
        shared_allocation_upper_bound: shared,
        payload_blocks,
        writer_inline,
        backend_inline,
        termination_inline,
        total,
    })
}

pub fn prepare_fixed_file_core<
    const BLOCKS: usize,
    const BYTES: usize,
    const COMMANDS: usize,
    const PATH: usize,
    const DIRENT_BYTES: usize,
    const ENTRIES: usize,
>(
    domain: ObservabilityQueueDomain,
    directory: &[u8],
    process_state_reserve: usize,
    limits: FixedFileLimits,
    filesystem_service: VerifiedFilesystemServiceProof,
    notifications: [FixedNotification; 3],
) -> Result<PreparedFixedFileCore<BLOCKS, BYTES, COMMANDS, PATH, DIRENT_BYTES, ENTRIES>, FixedFailure>
{
    if BLOCKS == 0
        || BYTES == 0
        || COMMANDS < BLOCKS.saturating_add(2)
        || PATH == 0
        || DIRENT_BYTES < 256
        || ENTRIES == 0
        || directory.len() >= PATH
        || directory.contains(&0)
        || !limits.valid()
    {
        return Err(FixedFailure::Layout);
    }
    let snapshot = domain.snapshot().map_err(|_| FixedFailure::Invariant)?;
    if snapshot.capacity != BLOCKS || snapshot.block_bytes != BYTES || snapshot.free != BLOCKS {
        return Err(FixedFailure::Layout);
    }
    let layout = fixed_core_layout::<BLOCKS, BYTES, COMMANDS, PATH, DIRENT_BYTES, ENTRIES>()?;
    if process_state_reserve < layout.total {
        return Err(FixedFailure::Layout);
    }
    let rotation_retries = usize::try_from(ROTATION_RETRIES).map_err(|_| FixedFailure::Layout)?;
    let write_interrupt_retries =
        usize::try_from(WRITE_INTERRUPT_RETRIES).map_err(|_| FixedFailure::Layout)?;
    let work = writer_work_proof::<BLOCKS, BYTES, ENTRIES>(
        limits,
        rotation_retries,
        write_interrupt_retries,
    )
    .map_err(FixedFailure::FilesystemProof)?;
    let termination = derive_termination_proof(work, filesystem_service)
        .map_err(FixedFailure::FilesystemProof)?;
    let mut completions = CompletionSet::new();
    let startup = completions
        .acquire(CompletionRole::Startup)
        .map_err(|_| FixedFailure::Invariant)?;
    let inner = Arc::new(Inner {
        state: Mutex::new(State {
            domain,
            payloads: [PayloadBlock::empty(); BLOCKS],
            commands: std::array::from_fn(|_| CommandSlot::vacant()),
            completions,
            head: 0,
            tail: 0,
            occupied: 0,
            accepting: true,
            failure: None,
            notification_outbox: [None; 3],
        }),
        changed: Condvar::new(),
        notifications,
    });
    let mut fixed_directory = [0; PATH];
    fixed_directory[..directory.len()].copy_from_slice(directory);
    Ok(PreparedFixedFileCore {
        sink: FixedFileSink {
            inner: Arc::clone(&inner),
        },
        writer: FixedWriterEntry {
            inner: Arc::clone(&inner),
            directory: fixed_directory,
            directory_length: directory.len(),
            record: [0; BYTES],
            streams: [StreamProgress::clean(); 4],
            limits,
        },
        startup: FixedCompletion {
            inner,
            ticket: startup,
            active: true,
        },
        termination,
    })
}

impl<const BLOCKS: usize, const BYTES: usize, const COMMANDS: usize>
    FixedFileSink<BLOCKS, BYTES, COMMANDS>
{
    pub(super) fn begin_record(
        &self,
        memory: &RequestMemory,
    ) -> Result<FixedEncodingLease<BLOCKS, BYTES, COMMANDS>, SubmitError> {
        let mut state = lock(&self.inner);
        ensure_accepting(&state)?;
        if state.occupied == COMMANDS {
            return Err(SubmitError::Backpressured);
        }
        let admission = memory
            .try_observability_block(&state.domain)
            .map_err(SubmitError::Admission)?;
        let block = admission.block();
        if block.slot() >= BLOCKS {
            drop(admission);
            fail_state(&mut state, FixedFailure::Invariant);
            self.inner.changed.notify_all();
            drop(state);
            fire_outbox(&self.inner);
            return Err(SubmitError::Unhealthy(FixedFailure::Invariant));
        }
        let command = state.tail;
        if !matches!(state.commands[command].state, CommandState::Vacant) {
            drop(admission);
            fail_state(&mut state, FixedFailure::Invariant);
            self.inner.changed.notify_all();
            drop(state);
            fire_outbox(&self.inner);
            return Err(SubmitError::Unhealthy(FixedFailure::Invariant));
        }
        let Some(command_generation) = state.commands[command].generation.checked_add(1) else {
            drop(admission);
            fail_state(&mut state, FixedFailure::Invariant);
            self.inner.changed.notify_all();
            drop(state);
            fire_outbox(&self.inner);
            return Err(SubmitError::Unhealthy(FixedFailure::Invariant));
        };
        state.commands[command].generation = command_generation;
        state.commands[command].state = CommandState::Encoding(block);
        let payload = &mut state.payloads[block.slot()];
        payload.generation = block.generation();
        payload.length = 0;
        payload.bytes.fill(0);
        state.tail = (state.tail + 1) % COMMANDS;
        state.occupied += 1;
        Ok(FixedEncodingLease {
            inner: Arc::clone(&self.inner),
            admission: Some(admission),
            block,
            command,
            command_generation,
            cursor: 0,
            active: true,
        })
    }

    pub fn try_flush(&self) -> Result<FixedCompletion<BLOCKS, BYTES, COMMANDS>, SubmitError> {
        self.enqueue_completion(CompletionRole::Flush, false)
    }

    pub fn try_shutdown(&self) -> Result<FixedCompletion<BLOCKS, BYTES, COMMANDS>, SubmitError> {
        self.enqueue_completion(CompletionRole::Shutdown, true)
    }

    fn enqueue_completion(
        &self,
        role: CompletionRole,
        shutdown: bool,
    ) -> Result<FixedCompletion<BLOCKS, BYTES, COMMANDS>, SubmitError> {
        let mut state = lock(&self.inner);
        ensure_accepting(&state)?;
        if shutdown
            && state
                .commands
                .iter()
                .any(|slot| matches!(slot.state, CommandState::Encoding(_)))
        {
            return Err(SubmitError::OutstandingEncoding);
        }
        if state.occupied == COMMANDS {
            return Err(SubmitError::Backpressured);
        }
        let ticket = match state.completions.acquire(role) {
            Ok(ticket) => ticket,
            Err(CompletionError::Busy | CompletionError::Stale) => {
                return Err(SubmitError::CompletionBusy);
            }
            Err(CompletionError::GenerationExhausted) => {
                fail_state(&mut state, FixedFailure::Invariant);
                self.inner.changed.notify_all();
                drop(state);
                fire_outbox(&self.inner);
                return Err(SubmitError::Unhealthy(FixedFailure::Invariant));
            }
        };
        let command = state.tail;
        state.commands[command].state = if shutdown {
            CommandState::Shutdown(ticket)
        } else {
            CommandState::Flush(ticket)
        };
        state.tail = (state.tail + 1) % COMMANDS;
        state.occupied += 1;
        if shutdown {
            state.accepting = false;
        }
        self.inner.changed.notify_one();
        Ok(FixedCompletion {
            inner: Arc::clone(&self.inner),
            ticket,
            active: true,
        })
    }
}

impl<const BLOCKS: usize, const BYTES: usize, const COMMANDS: usize>
    FixedEncodingLease<BLOCKS, BYTES, COMMANDS>
{
    pub(super) fn encode_and_commit(
        mut self,
        stream: FileStream,
        record: JsonlRecord<'_>,
    ) -> Result<(), SubmitError> {
        encode_jsonl(&mut self, record).map_err(|_| SubmitError::Encoding)?;
        self.commit(stream)
    }

    fn commit(mut self, stream: FileStream) -> Result<(), SubmitError> {
        let mut state = lock(&self.inner);
        validate_encoding(&state, &self)?;
        if self.cursor == 0
            || state.payloads[self.block.slot()].bytes[self.cursor - 1] != b'\n'
            || state.payloads[self.block.slot()].bytes[..self.cursor - 1].contains(&b'\n')
        {
            cancel_encoding(&mut state, self.command, self.block);
            self.active = false;
            drop(state);
            drop(self.admission.take());
            return Err(SubmitError::Encoding);
        }
        let admission = self
            .admission
            .take()
            .expect("active encoding owns Admission lease");
        let queued = match admission.commit() {
            Ok(queued) => queued,
            Err(error) => {
                cancel_encoding(&mut state, self.command, self.block);
                self.active = false;
                drop(state);
                return Err(SubmitError::Admission(error));
            }
        };
        let command = &mut state.commands[self.command];
        command.state = CommandState::Record {
            stream,
            block: self.block,
            length: self.cursor,
        };
        command.queued = Some(queued);
        state.payloads[self.block.slot()].length = self.cursor;
        self.active = false;
        self.inner.changed.notify_one();
        Ok(())
    }
}

impl<const BLOCKS: usize, const BYTES: usize, const COMMANDS: usize> EncodingTarget
    for FixedEncodingLease<BLOCKS, BYTES, COMMANDS>
{
    fn append(&mut self, bytes: &[u8]) -> Result<(), CodecError> {
        let end = self
            .cursor
            .checked_add(bytes.len())
            .ok_or(CodecError::Capacity)?;
        if end > BYTES {
            return Err(CodecError::Capacity);
        }
        let mut state = lock(&self.inner);
        validate_encoding(&state, self).map_err(|_| CodecError::Capacity)?;
        state.payloads[self.block.slot()].bytes[self.cursor..end].copy_from_slice(bytes);
        self.cursor = end;
        Ok(())
    }
}

impl<const BLOCKS: usize, const BYTES: usize, const COMMANDS: usize> Drop
    for FixedEncodingLease<BLOCKS, BYTES, COMMANDS>
{
    fn drop(&mut self) {
        if !self.active {
            return;
        }
        let mut state = lock(&self.inner);
        if encoding_matches(&state, self) {
            cancel_encoding(&mut state, self.command, self.block);
            self.inner.changed.notify_one();
        } else if state.failure.is_none() {
            fail_state(&mut state, FixedFailure::Invariant);
        }
        self.active = false;
        drop(state);
        drop(self.admission.take());
        fire_outbox(&self.inner);
    }
}

impl<
    const BLOCKS: usize,
    const BYTES: usize,
    const COMMANDS: usize,
    const PATH: usize,
    const DIRENT_BYTES: usize,
    const ENTRIES: usize,
> FixedWriterEntry<BLOCKS, BYTES, COMMANDS, PATH, DIRENT_BYTES, ENTRIES>
{
    pub fn run(mut self) -> Result<(), FixedFailure> {
        let mut backend = match LinuxFileBackend::<PATH, DIRENT_BYTES, ENTRIES>::prepare(
            &self.directory[..self.directory_length],
        ) {
            Ok(backend) => backend,
            Err(error) => {
                let failure = map_backend(error);
                self.finish_startup(Err(failure));
                return Err(failure);
            }
        };
        self.finish_startup(Ok(()));
        loop {
            match self.next_command() {
                WriterCommand::Wait(duration) => {
                    self.wait(duration);
                    self.sync_due(&backend)?;
                }
                WriterCommand::Cancelled => self.clear_head(None)?,
                WriterCommand::Record {
                    stream,
                    block,
                    length,
                } => {
                    self.fail_if_requested(FixedFailure::Write(stream))?;
                    backend
                        .write_all(stream, &self.record[..length])
                        .map_err(map_backend)
                        .or_else(|failure| self.fail(failure))?;
                    self.after_write(&mut backend, stream, length)?;
                    self.clear_head(Some(block))?;
                }
                WriterCommand::Flush(ticket) => {
                    let result = self.sync_barrier(&backend);
                    self.complete_barrier(ticket, result)?;
                }
                WriterCommand::Shutdown(ticket) => {
                    let result = self.sync_barrier(&backend);
                    self.complete_barrier(ticket, result)?;
                    return result;
                }
            }
        }
    }

    fn finish_startup(&self, result: Result<(), FixedFailure>) {
        let mut state = lock(&self.inner);
        let ticket = CompletionTicket::initial_startup();
        if matches!(state.completions.complete(ticket, result), Ok(true)) {
            queue_notification(&mut state, ticket);
        }
        if let Err(failure) = result {
            fail_state(&mut state, failure);
        }
        self.inner.changed.notify_all();
        drop(state);
        fire_outbox(&self.inner);
    }

    fn next_command(&mut self) -> WriterCommand {
        let state = lock(&self.inner);
        if state.occupied == 0
            || matches!(state.commands[state.head].state, CommandState::Encoding(_))
        {
            return WriterCommand::Wait(self.next_wait());
        }
        let command = &state.commands[state.head];
        match command.state {
            CommandState::Cancelled => WriterCommand::Cancelled,
            CommandState::Record {
                stream,
                block,
                length,
            } => {
                let payload = &state.payloads[block.slot()];
                if payload.generation != block.generation() || payload.length != length {
                    return WriterCommand::Wait(Duration::ZERO);
                }
                self.record[..length].copy_from_slice(&payload.bytes[..length]);
                WriterCommand::Record {
                    stream,
                    block,
                    length,
                }
            }
            CommandState::Flush(ticket) => WriterCommand::Flush(ticket),
            CommandState::Shutdown(ticket) => WriterCommand::Shutdown(ticket),
            CommandState::Vacant | CommandState::Encoding(_) => WriterCommand::Wait(Duration::ZERO),
        }
    }

    fn next_wait(&self) -> Duration {
        let now = Instant::now();
        self.streams
            .iter()
            .filter_map(|stream| stream.deadline)
            .min()
            .map_or(self.limits.sync_interval, |deadline| {
                deadline.saturating_duration_since(now)
            })
    }

    fn wait(&self, duration: Duration) {
        let state = lock(&self.inner);
        if state.occupied != 0
            && !matches!(state.commands[state.head].state, CommandState::Encoding(_))
        {
            return;
        }
        let _ = self
            .inner
            .changed
            .wait_timeout(state, duration)
            .expect("fixed file core mutex poisoned; terminate");
    }

    fn after_write(
        &mut self,
        backend: &mut LinuxFileBackend<PATH, DIRENT_BYTES, ENTRIES>,
        stream: FileStream,
        length: usize,
    ) -> Result<(), FixedFailure> {
        let bytes = u64::try_from(length).unwrap_or(u64::MAX);
        let (rotate, sync) = {
            let progress = &mut self.streams[stream.index()];
            progress.written = progress.written.saturating_add(bytes);
            progress.dirty = progress.dirty.saturating_add(bytes);
            progress
                .deadline
                .get_or_insert_with(|| Instant::now() + self.limits.sync_interval);
            (
                progress.written >= self.limits.rotate_bytes,
                progress.dirty >= self.limits.sync_bytes,
            )
        };
        if rotate {
            self.fail_if_requested(FixedFailure::Rotate(stream))?;
            if let Err(failure) = backend
                .rotate(
                    stream,
                    self.limits.retained_files,
                    u64::try_from(self.limits.retention_age.as_millis()).unwrap_or(u64::MAX),
                )
                .map_err(map_backend)
            {
                return self.fail(failure);
            }
            self.streams[stream.index()] = StreamProgress::clean();
        } else if sync {
            self.fail_if_requested(FixedFailure::Sync(stream))?;
            if let Err(failure) = backend.sync_stream(stream).map_err(map_backend) {
                return self.fail(failure);
            }
            let progress = &mut self.streams[stream.index()];
            progress.dirty = 0;
            progress.deadline = None;
        }
        Ok(())
    }

    fn sync_due(
        &mut self,
        backend: &LinuxFileBackend<PATH, DIRENT_BYTES, ENTRIES>,
    ) -> Result<(), FixedFailure> {
        let now = Instant::now();
        for stream in FileStream::ALL {
            let due = self.streams[stream.index()]
                .deadline
                .is_some_and(|deadline| deadline <= now);
            if due {
                self.fail_if_requested(FixedFailure::Sync(stream))?;
                if let Err(failure) = backend.sync_stream(stream).map_err(map_backend) {
                    return self.fail(failure);
                }
                let progress = &mut self.streams[stream.index()];
                progress.dirty = 0;
                progress.deadline = None;
            }
        }
        Ok(())
    }

    fn sync_barrier(
        &self,
        backend: &LinuxFileBackend<PATH, DIRENT_BYTES, ENTRIES>,
    ) -> Result<(), FixedFailure> {
        for stream in FileStream::ALL {
            self.fail_if_requested(FixedFailure::Sync(stream))?;
            backend.sync_stream(stream).map_err(map_backend)?;
        }
        Ok(())
    }

    #[cfg(test)]
    fn fail_if_requested(&self, failure: FixedFailure) -> Result<(), FixedFailure> {
        if self.limits.fail == Some(failure) {
            self.fail(failure)
        } else {
            Ok(())
        }
    }

    #[cfg(not(test))]
    const fn fail_if_requested(&self, _failure: FixedFailure) -> Result<(), FixedFailure> {
        Ok(())
    }

    fn clear_head(&self, expected: Option<ObservabilityBlockId>) -> Result<(), FixedFailure> {
        let mut state = lock(&self.inner);
        let head = state.head;
        if let Some(block) = expected {
            let matches = matches!(
                state.commands[head].state,
                CommandState::Record {
                    block: current,
                    ..
                } if current == block
            );
            if !matches {
                drop(state);
                return self.fail(FixedFailure::Invariant);
            }
            state.payloads[block.slot()].length = 0;
        } else if !matches!(state.commands[head].state, CommandState::Cancelled) {
            drop(state);
            return self.fail(FixedFailure::Invariant);
        }
        state.commands[head].clear();
        state.head = (head + 1) % COMMANDS;
        state.occupied -= 1;
        self.inner.changed.notify_all();
        Ok(())
    }

    fn complete_barrier(
        &self,
        ticket: CompletionTicket,
        result: Result<(), FixedFailure>,
    ) -> Result<(), FixedFailure> {
        let mut state = lock(&self.inner);
        let head = state.head;
        let notify = state
            .completions
            .complete(ticket, result)
            .map_err(|_| FixedFailure::Invariant)?;
        if notify {
            queue_notification(&mut state, ticket);
        }
        state.commands[head].clear();
        state.head = (head + 1) % COMMANDS;
        state.occupied -= 1;
        self.inner.changed.notify_all();
        if let Err(failure) = result {
            fail_locked_state(&mut state, failure);
            drop(state);
            fire_outbox(&self.inner);
            return Err(failure);
        }
        drop(state);
        fire_outbox(&self.inner);
        Ok(())
    }

    fn fail<T>(&self, failure: FixedFailure) -> Result<T, FixedFailure> {
        let mut state = lock(&self.inner);
        fail_locked_state(&mut state, failure);
        self.inner.changed.notify_all();
        drop(state);
        fire_outbox(&self.inner);
        Err(failure)
    }
}

enum WriterCommand {
    Wait(Duration),
    Cancelled,
    Record {
        stream: FileStream,
        block: ObservabilityBlockId,
        length: usize,
    },
    Flush(CompletionTicket),
    Shutdown(CompletionTicket),
}

fn ensure_accepting<const BLOCKS: usize, const BYTES: usize, const COMMANDS: usize>(
    state: &State<BLOCKS, BYTES, COMMANDS>,
) -> Result<(), SubmitError> {
    if let Some(failure) = state.failure {
        Err(SubmitError::Unhealthy(failure))
    } else if !state.accepting {
        Err(SubmitError::ShuttingDown)
    } else {
        Ok(())
    }
}

fn encoding_matches<const BLOCKS: usize, const BYTES: usize, const COMMANDS: usize>(
    state: &State<BLOCKS, BYTES, COMMANDS>,
    lease: &FixedEncodingLease<BLOCKS, BYTES, COMMANDS>,
) -> bool {
    state.commands[lease.command].generation == lease.command_generation
        && matches!(
            state.commands[lease.command].state,
            CommandState::Encoding(block) if block == lease.block
        )
        && state.payloads[lease.block.slot()].generation == lease.block.generation()
}

fn validate_encoding<const BLOCKS: usize, const BYTES: usize, const COMMANDS: usize>(
    state: &State<BLOCKS, BYTES, COMMANDS>,
    lease: &FixedEncodingLease<BLOCKS, BYTES, COMMANDS>,
) -> Result<(), SubmitError> {
    if let Some(failure) = state.failure {
        return Err(SubmitError::Unhealthy(failure));
    }
    if encoding_matches(state, lease) {
        Ok(())
    } else {
        Err(SubmitError::Unhealthy(FixedFailure::Invariant))
    }
}

fn cancel_encoding<const BLOCKS: usize, const BYTES: usize, const COMMANDS: usize>(
    state: &mut State<BLOCKS, BYTES, COMMANDS>,
    command: usize,
    block: ObservabilityBlockId,
) {
    state.commands[command].state = CommandState::Cancelled;
    state.payloads[block.slot()].length = 0;
}

fn fail_state<const BLOCKS: usize, const BYTES: usize, const COMMANDS: usize>(
    state: &mut State<BLOCKS, BYTES, COMMANDS>,
    failure: FixedFailure,
) {
    state.failure.get_or_insert(failure);
    state.accepting = false;
    let notifications = state.completions.fail_pending(failure);
    for ticket in notifications.into_iter().flatten() {
        queue_notification(state, ticket);
    }
}

fn fail_locked_state<const BLOCKS: usize, const BYTES: usize, const COMMANDS: usize>(
    state: &mut State<BLOCKS, BYTES, COMMANDS>,
    failure: FixedFailure,
) {
    fail_state(state, failure);
    for command in &mut state.commands {
        command.clear();
    }
    for payload in &mut state.payloads {
        payload.length = 0;
    }
    state.head = 0;
    state.tail = 0;
    state.occupied = 0;
}

fn queue_notification<const BLOCKS: usize, const BYTES: usize, const COMMANDS: usize>(
    state: &mut State<BLOCKS, BYTES, COMMANDS>,
    ticket: CompletionTicket,
) {
    state.notification_outbox[ticket.role() as usize] = Some(ticket);
}

fn fire_outbox<const BLOCKS: usize, const BYTES: usize, const COMMANDS: usize>(
    inner: &Inner<BLOCKS, BYTES, COMMANDS>,
) {
    let signals = {
        let mut state = lock(inner);
        std::mem::take(&mut state.notification_outbox)
    };
    for ticket in signals.into_iter().flatten() {
        inner.notifications[ticket.role() as usize].fire(ticket);
    }
}

fn lock<const BLOCKS: usize, const BYTES: usize, const COMMANDS: usize>(
    inner: &Inner<BLOCKS, BYTES, COMMANDS>,
) -> MutexGuard<'_, State<BLOCKS, BYTES, COMMANDS>> {
    inner
        .state
        .lock()
        .expect("fixed file core mutex poisoned; terminate")
}

const fn map_backend(error: BackendError) -> FixedFailure {
    match error {
        BackendError::Directory
        | BackendError::DirectoryExceeded
        | BackendError::PathExceeded
        | BackendError::InvalidPath => FixedFailure::Directory,
        BackendError::Lock => FixedFailure::Lock,
        BackendError::Recovery(stream) => FixedFailure::Recovery(stream),
        BackendError::Open(stream) => FixedFailure::Open(stream),
        BackendError::Write(stream) => FixedFailure::Write(stream),
        BackendError::Sync(stream) => FixedFailure::Sync(stream),
        BackendError::Rotate(stream) => FixedFailure::Rotate(stream),
        BackendError::Retention(stream) => FixedFailure::Retention(stream),
    }
}

#[cfg(test)]
mod tests {
    use std::{
        future::Future,
        os::unix::ffi::OsStrExt,
        pin::Pin,
        sync::atomic::{AtomicU64, Ordering},
        task::{Context, Poll, Waker},
        thread,
        time::Duration,
    };

    use saddle_admission::{AttemptOutcome, ProcessLedger, ResourceConfig};

    use super::*;
    use crate::file::codec::{Field, Scalar};

    const BLOCKS: usize = 4;
    const BYTES: usize = 1024;
    const COMMANDS: usize = 6;
    const PATH: usize = 256;
    const DIRENT: usize = 4096;
    const ENTRIES: usize = 32;
    const TEST_NO_THRESHOLD: u64 = u64::MAX / 8;
    const BUILD_IDENTITY: [u8; 32] = [6; 32];

    type Prepared = PreparedFixedFileCore<BLOCKS, BYTES, COMMANDS, PATH, DIRENT, ENTRIES>;

    struct TestFilesystemService;

    impl crate::file::DeploymentFilesystemServiceAttestation for TestFilesystemService {
        fn target(&self) -> crate::file::FilesystemTarget {
            crate::file::FilesystemTarget::LinuxX86_64
        }

        fn approved(&self) -> bool {
            true
        }

        fn reproducible_measurement(&self) -> bool {
            true
        }

        fn conservative_upper_bound(&self) -> bool {
            true
        }

        fn work_domain(&self) -> crate::file::FilesystemWorkDomain {
            crate::file::FilesystemWorkDomain {
                max_write_ops: u64::MAX,
                max_write_bytes: u64::MAX,
                max_file_sync_ops: u64::MAX,
                max_file_sync_bytes: u64::MAX,
                max_directory_sync_ops: u64::MAX,
                max_hard_link_ops: u64::MAX,
                max_rename_ops: 0,
                max_unlink_ops: u64::MAX,
                max_create_ops: u64::MAX,
                max_scan_ops: u64::MAX,
                max_scan_entries: u64::MAX,
                max_parallel_filesystem_ops: 1,
                advisory_lock: true,
                hard_link_no_replace: true,
                directory_fsync: true,
                data_fsync: true,
            }
        }

        fn service_rates(&self) -> crate::file::FilesystemServiceRates {
            crate::file::FilesystemServiceRates {
                write_op_nanos: 1,
                write_byte_nanos: 1,
                file_sync_op_nanos: 1,
                file_sync_byte_nanos: 1,
                directory_sync_op_nanos: 1,
                hard_link_op_nanos: 1,
                rename_op_nanos: 0,
                unlink_op_nanos: 1,
                create_op_nanos: 1,
                scan_op_nanos: 1,
                scan_entry_nanos: 1,
            }
        }

        fn calibration_identity(&self) -> [u8; 32] {
            [1; 32]
        }

        fn environment_identity(&self) -> [u8; 32] {
            [2; 32]
        }

        fn filesystem_identity(&self) -> [u8; 32] {
            [3; 32]
        }

        fn mount_identity(&self) -> [u8; 32] {
            [4; 32]
        }

        fn service_attestation(&self) -> [u8; 32] {
            [5; 32]
        }

        fn build_identity(&self) -> [u8; 32] {
            BUILD_IDENTITY
        }
    }

    fn test_filesystem_service() -> VerifiedFilesystemServiceProof {
        crate::file::verify_filesystem_service(
            TestFilesystemService,
            crate::file::FilesystemDeploymentIdentity {
                target: crate::file::FilesystemTarget::LinuxX86_64,
                environment: [2; 32],
                filesystem: [3; 32],
                mount: [4; 32],
                build: BUILD_IDENTITY,
            },
        )
        .unwrap()
    }

    static NEXT: AtomicU64 = AtomicU64::new(0);
    static FAILURE_NOTIFIED: [AtomicU64; 3] =
        [AtomicU64::new(0), AtomicU64::new(0), AtomicU64::new(0)];
    static CANCELLATION_NOTIFIED: [AtomicU64; 3] =
        [AtomicU64::new(0), AtomicU64::new(0), AtomicU64::new(0)];
    static SHUTDOWN_CANCELLATION_NOTIFIED: [AtomicU64; 3] =
        [AtomicU64::new(0), AtomicU64::new(0), AtomicU64::new(0)];

    fn ignore_notification(_: usize, _: CompletionSignal) {}

    const fn notifications() -> [FixedNotification; 3] {
        [FixedNotification::new(0, ignore_notification); 3]
    }

    fn count_failure_notification(token: usize, signal: CompletionSignal) {
        assert_eq!(token, signal.role as usize);
        FAILURE_NOTIFIED[token].fetch_add(1, Ordering::Relaxed);
    }

    fn failure_notifications() -> [FixedNotification; 3] {
        for count in &FAILURE_NOTIFIED {
            count.store(0, Ordering::Relaxed);
        }
        std::array::from_fn(|token| FixedNotification::new(token, count_failure_notification))
    }

    fn count_cancellation_notification(token: usize, signal: CompletionSignal) {
        assert_eq!(token, signal.role as usize);
        CANCELLATION_NOTIFIED[token].fetch_add(1, Ordering::Relaxed);
    }

    fn cancellation_notifications() -> [FixedNotification; 3] {
        for count in &CANCELLATION_NOTIFIED {
            count.store(0, Ordering::Relaxed);
        }
        std::array::from_fn(|token| FixedNotification::new(token, count_cancellation_notification))
    }

    fn count_shutdown_cancellation_notification(token: usize, signal: CompletionSignal) {
        assert_eq!(token, signal.role as usize);
        SHUTDOWN_CANCELLATION_NOTIFIED[token].fetch_add(1, Ordering::Relaxed);
    }

    fn shutdown_cancellation_notifications() -> [FixedNotification; 3] {
        for count in &SHUTDOWN_CANCELLATION_NOTIFIED {
            count.store(0, Ordering::Relaxed);
        }
        std::array::from_fn(|token| {
            FixedNotification::new(token, count_shutdown_cancellation_notification)
        })
    }

    fn directory(name: &str) -> std::path::PathBuf {
        std::env::temp_dir().join(format!(
            "saddle-fixed-core-{}-{name}",
            NEXT.fetch_add(1, Ordering::Relaxed)
        ))
    }

    fn ledger_and_core(path: &std::path::Path) -> (ProcessLedger, Prepared) {
        ledger_and_core_with_limits(
            path,
            FixedFileLimits {
                rotate_bytes: TEST_NO_THRESHOLD,
                retained_files: 2,
                retention_age: Duration::from_secs(60),
                sync_bytes: TEST_NO_THRESHOLD,
                sync_interval: Duration::from_millis(10),
                fail: None,
            },
        )
    }

    fn ledger_and_core_with_limits(
        path: &std::path::Path,
        limits: FixedFileLimits,
    ) -> (ProcessLedger, Prepared) {
        ledger_and_core_with_limits_and_notifications(path, limits, notifications())
    }

    fn ledger_and_core_with_limits_and_notifications(
        path: &std::path::Path,
        limits: FixedFileLimits,
        fixed_notifications: [FixedNotification; 3],
    ) -> (ProcessLedger, Prepared) {
        let layout = fixed_core_layout::<BLOCKS, BYTES, COMMANDS, PATH, DIRENT, ENTRIES>().unwrap();
        let base = ProcessLedger::minimum_process_state_reserve(4).unwrap();
        let queue = ProcessLedger::observability_queue_state_reserve(BLOCKS).unwrap();
        let process_state_reserve = base + queue + layout.total;
        let framework_reserve = BLOCKS * BYTES + 4096;
        let task_reserve = 4096;
        let managed_capacity = 16 * BYTES;
        let fixed = 64 + framework_reserve + task_reserve + process_state_reserve + 256 + 128;
        let ledger = ProcessLedger::new(ResourceConfig {
            managed_capacity,
            entry_reserve: 64,
            framework_reserve,
            task_reserve,
            process_state_reserve,
            system_estimate: 256,
            safety_margin: 128,
            process_limit: managed_capacity + fixed,
            max_active_requests: 4,
        })
        .unwrap();
        let domain = ledger.prepare_observability_queue(BLOCKS, BYTES).unwrap();
        let core = prepare_fixed_file_core(
            domain,
            path.as_os_str().as_bytes(),
            layout.total,
            limits,
            test_filesystem_service(),
            fixed_notifications,
        )
        .unwrap();
        (ledger, core)
    }

    fn poll_once<F: Future>(future: Pin<&mut F>) -> Poll<F::Output> {
        let waker = Waker::noop();
        let mut context = Context::from_waker(waker);
        future.poll(&mut context)
    }

    fn wait_completion(
        completion: &FixedCompletion<BLOCKS, BYTES, COMMANDS>,
    ) -> Result<(), FixedFailure> {
        let deadline = Instant::now() + Duration::from_secs(2);
        loop {
            match completion.result() {
                CompletionResult::Pending if Instant::now() < deadline => {
                    thread::sleep(Duration::from_millis(1));
                }
                CompletionResult::Pending => panic!("writer completion did not converge"),
                CompletionResult::Ready(result) => return result,
                CompletionResult::Stale => panic!("completion became stale before recycle"),
            }
        }
    }

    fn record<'a>(fields: &'a [Field<'a>]) -> JsonlRecord<'a> {
        JsonlRecord {
            timestamp_unix_ms: 1,
            level: "info",
            event: "domain.event",
            trace_id: Some("0123456789abcdef0123456789abcdef"),
            span_id: Some("0123456789abcdef"),
            parent_span_id: None,
            fields,
            dropped_events: 0,
        }
    }

    #[test]
    fn admission_lease_maps_to_fixed_payload_and_writer_recycles_linearly() {
        let path = directory("routing");
        let (ledger, core) = ledger_and_core(&path);
        let sink = core.sink;
        let startup = core.startup;
        let writer = thread::spawn(move || core.writer.run());
        wait_completion(&startup).unwrap();

        let mut envelope = ledger
            .try_envelope(2 * BYTES, 2048, |memory| {
                let fields = [Field {
                    name: "message",
                    value: Scalar::String("直接编码"),
                }];
                sink.begin_record(memory)
                    .unwrap()
                    .encode_and_commit(FileStream::Event, record(&fields))
                    .unwrap();
                async {}
            })
            .unwrap();
        let Poll::Ready(Ok(report)) = poll_once(Pin::new(&mut envelope)) else {
            panic!("fixed encoding request must complete");
        };
        assert_eq!(report.escape_allocations, 0);
        assert_eq!(report.escape_bytes, 0);
        drop(envelope);

        let flush = sink.try_flush().unwrap();
        wait_completion(&flush).unwrap();
        flush.recycle().unwrap();
        let shutdown = sink.try_shutdown().unwrap();
        wait_completion(&shutdown).unwrap();
        assert_eq!(writer.join().unwrap(), Ok(()));
        assert!(
            std::fs::read(path.join("saddle-event.jsonl"))
                .unwrap()
                .ends_with(b"}\n")
        );
        drop(startup);
        drop(shutdown);
        drop(sink);
        assert!(ledger.try_shutdown().unwrap().healthy);
        std::fs::remove_dir_all(path).unwrap();
    }

    #[test]
    fn dropped_encoding_rolls_back_and_cancelled_fifo_slot_is_reused() {
        let path = directory("rollback");
        let (ledger, core) = ledger_and_core(&path);
        let sink = core.sink;
        let startup = core.startup;
        let writer = thread::spawn(move || core.writer.run());
        wait_completion(&startup).unwrap();

        let mut envelope = ledger
            .try_envelope(2 * BYTES, 2048, |memory| {
                let lease = sink.begin_record(memory).unwrap();
                let first = lease.block;
                drop(lease);
                let next = sink.begin_record(memory).unwrap();
                assert_eq!(next.block.slot(), first.slot());
                assert_ne!(next.block.generation(), first.generation());
                drop(next);
                async {}
            })
            .unwrap();
        assert!(matches!(
            poll_once(Pin::new(&mut envelope)),
            Poll::Ready(Ok(_))
        ));
        drop(envelope);
        let shutdown = sink.try_shutdown().unwrap();
        wait_completion(&shutdown).unwrap();
        assert_eq!(writer.join().unwrap(), Ok(()));
        drop(startup);
        drop(shutdown);
        drop(sink);
        assert!(ledger.try_shutdown().unwrap().healthy);
        std::fs::remove_dir_all(path).unwrap();
    }

    #[test]
    fn shutdown_rejects_outstanding_encoding_owner_then_converges_at_zero() {
        let path = directory("encoding-owner");
        let (ledger, core) = ledger_and_core(&path);
        let sink = core.sink;
        let startup = core.startup;
        let writer = thread::spawn(move || core.writer.run());
        wait_completion(&startup).unwrap();

        let mut envelope = ledger
            .try_envelope(2 * BYTES, 2048, |memory| {
                let lease = sink.begin_record(memory).unwrap();
                assert!(matches!(
                    sink.try_shutdown(),
                    Err(SubmitError::OutstandingEncoding)
                ));
                drop(lease);
                async {}
            })
            .unwrap();
        assert!(matches!(
            poll_once(Pin::new(&mut envelope)),
            Poll::Ready(Ok(_))
        ));
        drop(envelope);

        let shutdown = sink.try_shutdown().unwrap();
        wait_completion(&shutdown).unwrap();
        assert_eq!(writer.join().unwrap(), Ok(()));
        drop(startup);
        drop(shutdown);
        drop(sink);
        assert!(ledger.try_shutdown().unwrap().healthy);
        std::fs::remove_dir_all(path).unwrap();
    }

    #[test]
    fn cancelled_shutdown_waiter_does_not_cancel_durability_or_ownership_return() {
        let path = directory("cancelled-shutdown");
        let (ledger, core) = ledger_and_core_with_limits_and_notifications(
            &path,
            FixedFileLimits {
                rotate_bytes: TEST_NO_THRESHOLD,
                retained_files: 2,
                retention_age: Duration::from_secs(60),
                sync_bytes: TEST_NO_THRESHOLD,
                sync_interval: Duration::from_millis(10),
                fail: None,
            },
            shutdown_cancellation_notifications(),
        );
        let sink = core.sink;
        let startup = core.startup;
        let shutdown = sink.try_shutdown().unwrap();
        assert_eq!(shutdown.cancel().unwrap(), CompletionResult::Pending);
        assert_eq!(core.writer.run(), Ok(()));
        assert_eq!(
            SHUTDOWN_CANCELLATION_NOTIFIED[CompletionRole::Shutdown as usize]
                .load(Ordering::Relaxed),
            0
        );
        drop(startup);
        drop(sink);
        assert!(ledger.try_shutdown().unwrap().healthy);
        std::fs::remove_dir_all(path).unwrap();
    }

    #[test]
    fn layout_and_fixed_completion_fail_closed_without_fallback() {
        let path = directory("layout");
        let layout = fixed_core_layout::<BLOCKS, BYTES, COMMANDS, PATH, DIRENT, ENTRIES>().unwrap();
        let base = ProcessLedger::minimum_process_state_reserve(1).unwrap();
        let queue = ProcessLedger::observability_queue_state_reserve(BLOCKS).unwrap();
        let process_state_reserve = base + queue + layout.total;
        let framework_reserve = BLOCKS * BYTES;
        let fixed = framework_reserve + process_state_reserve + 5;
        let ledger = ProcessLedger::new(ResourceConfig {
            managed_capacity: BYTES,
            entry_reserve: 1,
            framework_reserve,
            task_reserve: 1,
            process_state_reserve,
            system_estimate: 1,
            safety_margin: 1,
            process_limit: BYTES + fixed,
            max_active_requests: 1,
        })
        .unwrap();
        let domain = ledger.prepare_observability_queue(BLOCKS, BYTES).unwrap();
        assert!(matches!(
            prepare_fixed_file_core::<BLOCKS, BYTES, COMMANDS, PATH, DIRENT, ENTRIES>(
                domain,
                path.as_os_str().as_bytes(),
                layout.total - 1,
                FixedFileLimits::candidate(),
                test_filesystem_service(),
                notifications(),
            ),
            Err(FixedFailure::Layout)
        ));
        assert!(ledger.try_shutdown().unwrap().healthy);

        let ledger = ProcessLedger::new(ResourceConfig {
            managed_capacity: BYTES,
            entry_reserve: 1,
            framework_reserve,
            task_reserve: 1,
            process_state_reserve,
            system_estimate: 1,
            safety_margin: 1,
            process_limit: BYTES + fixed,
            max_active_requests: 1,
        })
        .unwrap();
        let mismatched = ledger
            .prepare_observability_queue(BLOCKS, BYTES / 2)
            .unwrap();
        assert!(matches!(
            prepare_fixed_file_core::<BLOCKS, BYTES, COMMANDS, PATH, DIRENT, ENTRIES>(
                mismatched,
                path.as_os_str().as_bytes(),
                layout.total,
                FixedFileLimits::candidate(),
                test_filesystem_service(),
                notifications(),
            ),
            Err(FixedFailure::Layout)
        ));
        assert!(ledger.try_shutdown().unwrap().healthy);
    }

    #[test]
    fn startup_backend_failure_is_sticky_and_completes_fixed_cell() {
        let path = directory("bad-parent").join("missing").join("nested");
        let (ledger, core) = ledger_and_core(&path);
        let startup = core.startup;
        let sink = core.sink;
        assert!(core.writer.run().is_err());
        assert!(matches!(
            startup.result(),
            CompletionResult::Ready(Err(FixedFailure::Directory))
        ));
        assert!(matches!(
            sink.try_flush(),
            Err(SubmitError::Unhealthy(FixedFailure::Directory))
        ));
        drop(startup);
        drop(sink);
        assert!(ledger.try_shutdown().unwrap().healthy);
    }

    #[test]
    fn cancelled_startup_waiter_is_reclaimed_without_notification() {
        let path = directory("cancelled-startup");
        let limits = FixedFileLimits {
            rotate_bytes: TEST_NO_THRESHOLD,
            retained_files: 2,
            retention_age: Duration::from_secs(60),
            sync_bytes: TEST_NO_THRESHOLD,
            sync_interval: Duration::from_millis(10),
            fail: None,
        };
        let (ledger, core) = ledger_and_core_with_limits_and_notifications(
            &path,
            limits,
            cancellation_notifications(),
        );
        assert_eq!(core.startup.cancel().unwrap(), CompletionResult::Pending);
        let shutdown = core.sink.try_shutdown().unwrap();
        assert_eq!(core.writer.run(), Ok(()));
        assert_eq!(
            CANCELLATION_NOTIFIED[CompletionRole::Startup as usize].load(Ordering::Relaxed),
            0
        );
        assert_eq!(
            CANCELLATION_NOTIFIED[CompletionRole::Shutdown as usize].load(Ordering::Relaxed),
            1
        );
        shutdown.recycle().unwrap();
        drop(core.sink);
        assert!(ledger.try_shutdown().unwrap().healthy);
        std::fs::remove_dir_all(path).unwrap();
    }

    #[test]
    fn full_admission_domain_is_atomic_and_request_allocation_free() {
        let path = directory("full");
        let (ledger, core) = ledger_and_core(&path);
        let sink = core.sink;
        let mut envelope = ledger
            .try_envelope((BLOCKS + 1) * BYTES, 4096, |memory| {
                let leases: [Option<FixedEncodingLease<BLOCKS, BYTES, COMMANDS>>; BLOCKS] =
                    std::array::from_fn(|_| Some(sink.begin_record(memory).unwrap()));
                assert!(matches!(
                    sink.begin_record(memory),
                    Err(SubmitError::Admission(
                        AdmissionError::ObservabilityQueueFull
                    ))
                ));
                drop(leases);
                async {}
            })
            .unwrap();
        let Poll::Ready(Ok(report)) = poll_once(Pin::new(&mut envelope)) else {
            panic!("full-domain request must roll back cleanly");
        };
        assert_eq!(report.escape_allocations, 0);
        assert_eq!(report.escape_bytes, 0);
        drop(envelope);
        drop(core.startup);
        drop(core.writer);
        drop(sink);
        assert!(ledger.try_shutdown().unwrap().healthy);
        let _ = std::fs::remove_dir_all(path);
    }

    #[test]
    fn writer_failure_is_global_sticky_and_fails_queued_barrier() {
        let path = directory("writer-failure");
        let limits = FixedFileLimits {
            rotate_bytes: TEST_NO_THRESHOLD,
            retained_files: 2,
            retention_age: Duration::from_secs(60),
            sync_bytes: TEST_NO_THRESHOLD,
            sync_interval: Duration::from_secs(1),
            fail: Some(FixedFailure::Write(FileStream::Event)),
        };
        let (ledger, core) =
            ledger_and_core_with_limits_and_notifications(&path, limits, failure_notifications());
        let sink = core.sink;
        let startup = core.startup;
        let mut envelope = ledger
            .try_envelope(2 * BYTES, 2048, |memory| {
                sink.begin_record(memory)
                    .unwrap()
                    .encode_and_commit(FileStream::Event, record(&[]))
                    .unwrap();
                async {}
            })
            .unwrap();
        assert!(matches!(
            poll_once(Pin::new(&mut envelope)),
            Poll::Ready(Ok(_))
        ));
        drop(envelope);
        let flush = sink.try_flush().unwrap();
        let shutdown = sink.try_shutdown().unwrap();
        assert_eq!(
            core.writer.run(),
            Err(FixedFailure::Write(FileStream::Event))
        );
        assert_eq!(
            flush.result(),
            CompletionResult::Ready(Err(FixedFailure::Write(FileStream::Event)))
        );
        assert_eq!(
            shutdown.result(),
            CompletionResult::Ready(Err(FixedFailure::Write(FileStream::Event)))
        );
        assert_eq!(
            FAILURE_NOTIFIED[CompletionRole::Startup as usize].load(Ordering::Relaxed),
            1
        );
        assert_eq!(
            FAILURE_NOTIFIED[CompletionRole::Flush as usize].load(Ordering::Relaxed),
            1
        );
        assert_eq!(
            FAILURE_NOTIFIED[CompletionRole::Shutdown as usize].load(Ordering::Relaxed),
            1
        );
        assert!(matches!(
            sink.try_flush(),
            Err(SubmitError::Unhealthy(FixedFailure::Write(
                FileStream::Event
            )))
        ));
        drop(startup);
        drop(flush);
        drop(shutdown);
        drop(sink);
        assert!(ledger.try_shutdown().unwrap().healthy);
        std::fs::remove_dir_all(path).unwrap();
    }

    #[test]
    fn attempt_outcome_can_hold_fixed_encoding_without_public_domain_access() {
        let path = directory("pending");
        let (ledger, core) = ledger_and_core(&path);
        let sink = core.sink;
        let outcome = ledger.attempt_or_register(2 * BYTES, 2048, |memory| {
            let lease = sink.begin_record(memory).unwrap();
            async move {
                drop(lease);
            }
        });
        let AttemptOutcome::Ready(mut envelope) = outcome else {
            panic!("fixed request should be admitted");
        };
        assert!(matches!(
            poll_once(Pin::new(&mut envelope)),
            Poll::Ready(Ok(_))
        ));
        drop(envelope);
        drop(core.startup);
        drop(core.writer);
        drop(sink);
        assert!(ledger.try_shutdown().unwrap().healthy);
        let _ = std::fs::remove_dir_all(path);
    }
}