lodestone_client/

workflow_journal.rs

//! Crash-safe local recovery journal for bounded operator workflows.
//!
//! Product state remains authoritative in Core. This journal fixes local
//! effect identities and records observations so a caller can resolve or
//! resume after response loss without guessing.

use std::error::Error;
use std::fmt::{self, Display, Formatter};
use std::marker::PhantomData;
use std::path::Path;

use serde::de::DeserializeOwned;
use serde::{Deserialize, Serialize};

pub const WORKFLOW_JOURNAL_FORMAT_VERSION: u16 = 1;
pub const MAX_WORKFLOW_JOURNAL_INTENT_BYTES: usize = 64 * 1024;
pub const MAX_WORKFLOW_JOURNAL_RECORD_BYTES: usize = 64 * 1024;
pub const MAX_WORKFLOW_JOURNAL_RECORDS: u64 = 256;

pub(crate) mod sealed {
    pub trait Sealed {}
}

/// Closed local workflows admitted to the shared journal implementation.
#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum LocalOperatorWorkflowKind {
    DevelopmentMaterialize,
    DevelopmentPkiSuccessor,
    DevelopmentReconcile,
    ExactMutation,
    ProvisionRetainedAppend,
    RotateCredential,
}

/// A client-owned, closed, secret-free workflow intent.
///
/// The sealed trait keeps arbitrary JSON values out of the journal. New
/// workflows define their typed intent and transition in this crate.
pub trait WorkflowJournalIntent: sealed::Sealed + DeserializeOwned + Serialize + Send + Sync + 'static {
    type Transition: WorkflowJournalTransition;

    const KIND: LocalOperatorWorkflowKind;

    fn validate(&self) -> Result<(), WorkflowJournalError>;
}

/// One closed, secret-free transition for a client-owned operator workflow.
pub trait WorkflowJournalTransition: sealed::Sealed + DeserializeOwned + Serialize + Send + Sync + 'static {
    fn validate(&self) -> Result<(), WorkflowJournalError>;
}

/// Random identity of one exact local workflow transaction.
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct WorkflowTransactionId([u8; 32]);

impl WorkflowTransactionId {
    #[must_use]
    pub const fn as_bytes(self) -> [u8; 32] {
        self.0
    }

    fn is_zero(self) -> bool {
        self.0.iter().all(|byte| *byte == 0)
    }
}

/// Durable identity returned after one transition is fully published.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct WorkflowJournalRecordReceipt {
    sequence: u64,
    digest: [u8; 32],
}

impl WorkflowJournalRecordReceipt {
    #[must_use]
    pub const fn sequence(self) -> u64 {
        self.sequence
    }

    #[must_use]
    pub const fn digest(self) -> [u8; 32] {
        self.digest
    }
}

/// Open journal plus its fully verified transition history.
pub struct WorkflowJournalReplay<I: WorkflowJournalIntent> {
    journal: WorkflowJournal<I>,
    transitions: Vec<I::Transition>,
}

impl<I: WorkflowJournalIntent> WorkflowJournalReplay<I> {
    #[must_use]
    pub const fn journal(&self) -> &WorkflowJournal<I> {
        &self.journal
    }

    #[must_use]
    pub fn transitions(&self) -> &[I::Transition] {
        &self.transitions
    }

    #[must_use]
    pub fn into_parts(self) -> (WorkflowJournal<I>, Vec<I::Transition>) {
        (self.journal, self.transitions)
    }
}

/// Exclusive writer for one exact local operator workflow.
pub struct WorkflowJournal<I: WorkflowJournalIntent> {
    transaction_id: WorkflowTransactionId,
    intent_digest: [u8; 32],
    #[cfg(any(target_os = "linux", target_os = "macos"))]
    inner: std::sync::Arc<std::sync::Mutex<unix::JournalState>>,
    marker: PhantomData<I>,
}

impl<I: WorkflowJournalIntent> WorkflowJournal<I> {
    /// Creates or exactly resumes and locks one workflow journal.
    pub async fn open_or_create(
        path: impl AsRef<Path>,
        intent: I,
    ) -> Result<WorkflowJournalReplay<I>, WorkflowJournalError> {
        intent.validate()?;
        let path = path.as_ref().to_path_buf();

        #[cfg(any(target_os = "linux", target_os = "macos"))]
        {
            tokio::task::spawn_blocking(move || unix::open_or_create(path, intent))
                .await
                .map_err(|_| WorkflowJournalError::WorkerUnavailable)?
        }

        #[cfg(not(any(target_os = "linux", target_os = "macos")))]
        {
            let _ = (path, intent);
            Err(WorkflowJournalError::UnsupportedPlatform)
        }
    }

    #[must_use]
    pub const fn transaction_id(&self) -> WorkflowTransactionId {
        self.transaction_id
    }

    #[must_use]
    pub const fn intent_digest(&self) -> [u8; 32] {
        self.intent_digest
    }

    /// Appends and synchronizes one transition before returning its receipt.
    ///
    /// Any publication error poisons this handle because the create-new link
    /// may have committed. Drop and reopen the journal to resolve it by replay.
    pub async fn append(
        &self,
        transition: I::Transition,
    ) -> Result<WorkflowJournalRecordReceipt, WorkflowJournalError> {
        transition.validate()?;

        #[cfg(any(target_os = "linux", target_os = "macos"))]
        {
            let inner = std::sync::Arc::clone(&self.inner);
            let transaction_id = self.transaction_id;
            let intent_digest = self.intent_digest;
            tokio::task::spawn_blocking(move || {
                let result = unix::append_transition(&inner, transaction_id, intent_digest, transition);
                // INVARIANT: release the worker's journal reference before the
                // join handle becomes ready so drop-and-reopen cannot race the
                // blocking scheduler's task-allocation cleanup.
                drop(inner);
                result
            })
            .await
            .map_err(|_| WorkflowJournalError::WorkerUnavailable)?
        }

        #[cfg(not(any(target_os = "linux", target_os = "macos")))]
        {
            let _ = transition;
            Err(WorkflowJournalError::UnsupportedPlatform)
        }
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum WorkflowJournalError {
    UnsupportedPlatform,
    InvalidPath,
    InsecurePath,
    Busy,
    IntentMismatch,
    InvalidIntent,
    InvalidTransition,
    LimitExceeded,
    Corrupt,
    Io,
    Codec,
    RandomUnavailable,
    ReopenRequired,
    WorkerUnavailable,
}

impl WorkflowJournalError {
    #[must_use]
    pub const fn code(self) -> &'static str {
        match self {
            Self::UnsupportedPlatform => "unsupported_platform",
            Self::InvalidPath => "invalid_path",
            Self::InsecurePath => "insecure_path",
            Self::Busy => "busy",
            Self::IntentMismatch => "intent_mismatch",
            Self::InvalidIntent => "invalid_intent",
            Self::InvalidTransition => "invalid_transition",
            Self::LimitExceeded => "limit_exceeded",
            Self::Corrupt => "corrupt",
            Self::Io => "io",
            Self::Codec => "codec",
            Self::RandomUnavailable => "random_unavailable",
            Self::ReopenRequired => "reopen_required",
            Self::WorkerUnavailable => "worker_unavailable",
        }
    }
}

impl Display for WorkflowJournalError {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        formatter.write_str(match self {
            Self::UnsupportedPlatform => "the platform cannot provide the required workflow journal guarantees",
            Self::InvalidPath => "the workflow journal path is not a canonical absolute directory path",
            Self::InsecurePath => "the workflow journal path or metadata is not private and no-follow safe",
            Self::Busy => "the workflow journal is locked by another writer",
            Self::IntentMismatch => "the workflow journal belongs to different invocation inputs",
            Self::InvalidIntent => "the workflow intent violates its closed contract",
            Self::InvalidTransition => "the workflow transition violates its closed contract",
            Self::LimitExceeded => "the workflow journal exceeds a fixed bound",
            Self::Corrupt => "the workflow journal is incomplete, forked, or corrupt",
            Self::Io => "the workflow journal filesystem operation failed",
            Self::Codec => "the workflow journal value is not canonical current-format JSON",
            Self::RandomUnavailable => "a workflow transaction identity could not be generated",
            Self::ReopenRequired => "the workflow journal must be reopened to resolve local publication",
            Self::WorkerUnavailable => "the bounded filesystem worker did not complete",
        })
    }
}

impl Error for WorkflowJournalError {}

#[derive(Deserialize, Serialize)]
#[serde(deny_unknown_fields)]
struct IntentEnvelope<I> {
    format_version: u16,
    workflow_kind: LocalOperatorWorkflowKind,
    transaction_id: WorkflowTransactionId,
    intent: I,
}

#[derive(Deserialize, Serialize)]
#[serde(deny_unknown_fields)]
struct TransitionEnvelope<T> {
    format_version: u16,
    transaction_id: WorkflowTransactionId,
    intent_digest: [u8; 32],
    sequence: u64,
    previous_record_digest: [u8; 32],
    transition: T,
}

#[cfg(any(target_os = "linux", target_os = "macos"))]
mod unix {
    use std::collections::BTreeSet;
    use std::ffi::OsStr;
    use std::fs::File;
    use std::io::{Read as _, Write as _};
    use std::marker::PhantomData;
    use std::os::unix::ffi::OsStrExt as _;
    use std::path::{Component, Path, PathBuf};
    use std::sync::{Arc, Mutex};

    use aws_lc_rs::digest;
    use rustix::fs::{
        AtFlags, Dir, FileType, FlockOperation, Mode, OFlags, flock, fstat, fsync, linkat, mkdirat, open, openat,
        unlinkat,
    };
    use serde::Serialize;
    use serde::de::DeserializeOwned;

    use super::{
        IntentEnvelope, MAX_WORKFLOW_JOURNAL_INTENT_BYTES, MAX_WORKFLOW_JOURNAL_RECORD_BYTES,
        MAX_WORKFLOW_JOURNAL_RECORDS, TransitionEnvelope, WORKFLOW_JOURNAL_FORMAT_VERSION, WorkflowJournal,
        WorkflowJournalError, WorkflowJournalIntent, WorkflowJournalRecordReceipt, WorkflowJournalReplay,
        WorkflowJournalTransition, WorkflowTransactionId,
    };

    const DIRECTORY_MODE: rustix::fs::RawMode = 0o700;
    const FILE_MODE: rustix::fs::RawMode = 0o600;
    const INTENT_NAME: &str = "intent.json";
    const INTENT_PENDING_NAME: &str = "intent.pending";
    const LOCK_NAME: &str = "journal.lock";
    const RECORDS_NAME: &str = "records";

    fn stat_value_u64<T: TryInto<u64>>(value: T) -> Option<u64> {
        value.try_into().ok()
    }

    pub(super) struct JournalState {
        _root: File,
        records: File,
        lock: File,
        last_sequence: u64,
        last_record_digest: [u8; 32],
        usable: bool,
    }

    impl Drop for JournalState {
        fn drop(&mut self) {
            // INVARIANT: a caller that drops and immediately resumes a local
            // workflow must not depend on deferred descriptor cleanup to
            // release the exclusive writer fence.
            let _ = flock(&self.lock, FlockOperation::Unlock);
        }
    }

    pub(super) fn open_or_create<I: WorkflowJournalIntent>(
        path: PathBuf,
        intent: I,
    ) -> Result<WorkflowJournalReplay<I>, WorkflowJournalError> {
        let (parent_path, name) = split_absolute_target(&path)?;
        let parent = open_directory_chain(parent_path)?;
        validate_parent_directory(&parent)?;

        let root_created = match mkdirat(&parent, name, Mode::from_raw_mode(DIRECTORY_MODE)) {
            Ok(()) => true,
            Err(error) if error == rustix::io::Errno::EXIST => false,
            Err(_) => return Err(WorkflowJournalError::Io),
        };
        if root_created {
            fsync(&parent).map_err(|_| WorkflowJournalError::Io)?;
        }

        let root = File::from(open_directory_at(&parent, name)?);
        validate_private_directory(&root)?;
        validate_root_names(&root)?;

        let lock = open_lock_file(&root)?;
        match flock(&lock, FlockOperation::NonBlockingLockExclusive) {
            Ok(()) => {}
            Err(error) if error == rustix::io::Errno::AGAIN => return Err(WorkflowJournalError::Busy),
            Err(_) => return Err(WorkflowJournalError::Io),
        }

        // Repeat after the lock because the prior holder may have completed a
        // pending create-new publication between the first listing and lock.
        validate_root_names(&root)?;
        recover_pending(&root, INTENT_PENDING_NAME, INTENT_NAME)?;
        let records = open_or_create_records_directory(&root)?;
        recover_record_pendings(&records)?;

        let current_intent_bytes = encode_bounded(&intent, MAX_WORKFLOW_JOURNAL_INTENT_BYTES)?;
        let (transaction_id, intent_digest) =
            match read_optional_file(&root, INTENT_NAME, MAX_WORKFLOW_JOURNAL_INTENT_BYTES)? {
                Some(bytes) => {
                    let envelope: IntentEnvelope<I> = decode_canonical(&bytes)?;
                    if envelope.format_version != WORKFLOW_JOURNAL_FORMAT_VERSION
                        || envelope.workflow_kind != I::KIND
                        || envelope.transaction_id.is_zero()
                    {
                        return Err(WorkflowJournalError::Corrupt);
                    }
                    envelope.intent.validate()?;
                    if encode_bounded(&envelope.intent, MAX_WORKFLOW_JOURNAL_INTENT_BYTES)? != current_intent_bytes {
                        return Err(WorkflowJournalError::IntentMismatch);
                    }
                    (envelope.transaction_id, sha256(&bytes)?)
                }
                None => {
                    let transaction_id = generate_transaction_id()?;
                    let envelope = IntentEnvelope {
                        format_version: WORKFLOW_JOURNAL_FORMAT_VERSION,
                        workflow_kind: I::KIND,
                        transaction_id,
                        intent,
                    };
                    let bytes = encode_bounded(&envelope, MAX_WORKFLOW_JOURNAL_INTENT_BYTES)?;
                    atomic_publish(&root, INTENT_PENDING_NAME, INTENT_NAME, &bytes)?;
                    (transaction_id, sha256(&bytes)?)
                }
            };

        let (transitions, last_sequence, last_record_digest) =
            replay_records::<I::Transition>(&records, transaction_id, intent_digest)?;
        let journal = WorkflowJournal {
            transaction_id,
            intent_digest,
            inner: Arc::new(Mutex::new(JournalState {
                _root: root,
                records,
                lock,
                last_sequence,
                last_record_digest,
                usable: true,
            })),
            marker: PhantomData,
        };
        Ok(WorkflowJournalReplay { journal, transitions })
    }

    pub(super) fn append_transition<T: WorkflowJournalTransition>(
        inner: &Arc<Mutex<JournalState>>,
        transaction_id: WorkflowTransactionId,
        intent_digest: [u8; 32],
        transition: T,
    ) -> Result<WorkflowJournalRecordReceipt, WorkflowJournalError> {
        let mut state = inner.lock().map_err(|_| WorkflowJournalError::ReopenRequired)?;
        if !state.usable {
            return Err(WorkflowJournalError::ReopenRequired);
        }
        let sequence = state
            .last_sequence
            .checked_add(1)
            .ok_or(WorkflowJournalError::LimitExceeded)?;
        if sequence > MAX_WORKFLOW_JOURNAL_RECORDS {
            return Err(WorkflowJournalError::LimitExceeded);
        }
        let envelope = TransitionEnvelope {
            format_version: WORKFLOW_JOURNAL_FORMAT_VERSION,
            transaction_id,
            intent_digest,
            sequence,
            previous_record_digest: state.last_record_digest,
            transition,
        };
        let bytes = encode_bounded(&envelope, MAX_WORKFLOW_JOURNAL_RECORD_BYTES)?;
        let digest = sha256(&bytes)?;
        let final_name = record_name(sequence);
        let pending_name = pending_record_name(sequence);
        if let Err(error) = atomic_publish(&state.records, &pending_name, &final_name, &bytes) {
            state.usable = false;
            return Err(error);
        }
        state.last_sequence = sequence;
        state.last_record_digest = digest;
        Ok(WorkflowJournalRecordReceipt { sequence, digest })
    }

    fn split_absolute_target(path: &Path) -> Result<(&Path, &OsStr), WorkflowJournalError> {
        if !path.is_absolute()
            || path.components().any(|component| {
                matches!(
                    component,
                    Component::CurDir | Component::ParentDir | Component::Prefix(_)
                )
            })
        {
            return Err(WorkflowJournalError::InvalidPath);
        }
        let parent = path.parent().ok_or(WorkflowJournalError::InvalidPath)?;
        let name = path.file_name().ok_or(WorkflowJournalError::InvalidPath)?;
        if name.as_bytes().is_empty() || name.as_bytes().contains(&0) {
            return Err(WorkflowJournalError::InvalidPath);
        }
        Ok((parent, name))
    }

    fn open_directory_chain(path: &Path) -> Result<File, WorkflowJournalError> {
        let mut components = path.components();
        if !matches!(components.next(), Some(Component::RootDir)) {
            return Err(WorkflowJournalError::InvalidPath);
        }
        let flags = directory_open_flags();
        let mut current = File::from(open("/", flags, Mode::empty()).map_err(map_open_error)?);
        for component in components {
            let Component::Normal(name) = component else {
                return Err(WorkflowJournalError::InvalidPath);
            };
            current = File::from(openat(&current, name, flags, Mode::empty()).map_err(map_open_error)?);
        }
        Ok(current)
    }

    fn open_directory_at(parent: &File, name: &OsStr) -> Result<std::os::fd::OwnedFd, WorkflowJournalError> {
        openat(parent, name, directory_open_flags(), Mode::empty()).map_err(map_open_error)
    }

    const fn directory_open_flags() -> OFlags {
        OFlags::RDONLY
            .union(OFlags::DIRECTORY)
            .union(OFlags::CLOEXEC)
            .union(OFlags::NOFOLLOW)
    }

    fn validate_parent_directory(directory: &File) -> Result<(), WorkflowJournalError> {
        let stat = fstat(directory).map_err(|_| WorkflowJournalError::Io)?;
        if FileType::from_raw_mode(stat.st_mode) != FileType::Directory
            || stat.st_uid != rustix::process::geteuid().as_raw()
            || stat.st_mode & 0o022 != 0
        {
            return Err(WorkflowJournalError::InsecurePath);
        }
        Ok(())
    }

    fn validate_private_directory(directory: &File) -> Result<(), WorkflowJournalError> {
        let stat = fstat(directory).map_err(|_| WorkflowJournalError::Io)?;
        if FileType::from_raw_mode(stat.st_mode) != FileType::Directory
            || stat.st_uid != rustix::process::geteuid().as_raw()
            || stat.st_mode & 0o777 != DIRECTORY_MODE
        {
            return Err(WorkflowJournalError::InsecurePath);
        }
        Ok(())
    }

    fn validate_root_names(root: &File) -> Result<(), WorkflowJournalError> {
        let names = list_names(root)?;
        if names
            .iter()
            .any(|name| name != LOCK_NAME && name != INTENT_NAME && name != INTENT_PENDING_NAME && name != RECORDS_NAME)
        {
            return Err(WorkflowJournalError::Corrupt);
        }
        Ok(())
    }

    fn open_lock_file(root: &File) -> Result<File, WorkflowJournalError> {
        let (lock, created) = open_or_create_private_file(root, LOCK_NAME, true)?;
        let stat = validate_private_file(&lock, Some(0), 1)?;
        if stat.st_size != 0 {
            return Err(WorkflowJournalError::Corrupt);
        }
        if created {
            lock.sync_all().map_err(|_| WorkflowJournalError::Io)?;
            fsync(root).map_err(|_| WorkflowJournalError::Io)?;
        }
        Ok(lock)
    }

    fn open_or_create_records_directory(root: &File) -> Result<File, WorkflowJournalError> {
        let created = match mkdirat(root, RECORDS_NAME, Mode::from_raw_mode(DIRECTORY_MODE)) {
            Ok(()) => true,
            Err(error) if error == rustix::io::Errno::EXIST => false,
            Err(_) => return Err(WorkflowJournalError::Io),
        };
        if created {
            fsync(root).map_err(|_| WorkflowJournalError::Io)?;
        }
        let records =
            File::from(openat(root, RECORDS_NAME, directory_open_flags(), Mode::empty()).map_err(map_open_error)?);
        validate_private_directory(&records)?;
        Ok(records)
    }

    fn open_or_create_private_file(
        directory: &File,
        name: &str,
        read_write: bool,
    ) -> Result<(File, bool), WorkflowJournalError> {
        let access = if read_write { OFlags::RDWR } else { OFlags::WRONLY };
        let create_flags = access | OFlags::CREATE | OFlags::EXCL | OFlags::CLOEXEC | OFlags::NOFOLLOW;
        match openat(directory, name, create_flags, Mode::from_raw_mode(FILE_MODE)) {
            Ok(fd) => Ok((File::from(fd), true)),
            Err(error) if error == rustix::io::Errno::EXIST => {
                let flags = access | OFlags::CLOEXEC | OFlags::NOFOLLOW | OFlags::NONBLOCK;
                let file = File::from(openat(directory, name, flags, Mode::empty()).map_err(map_open_error)?);
                Ok((file, false))
            }
            Err(error) => Err(map_open_error(error)),
        }
    }

    fn recover_record_pendings(records: &File) -> Result<(), WorkflowJournalError> {
        let names = list_names(records)?;
        for name in &names {
            if let Some(sequence) = parse_pending_record_name(name)? {
                if sequence == 0 || sequence > MAX_WORKFLOW_JOURNAL_RECORDS {
                    return Err(WorkflowJournalError::Corrupt);
                }
                recover_pending(records, name, &record_name(sequence))?;
            }
        }
        let names = list_names(records)?;
        if names.iter().any(|name| parse_record_name(name).is_err()) {
            return Err(WorkflowJournalError::Corrupt);
        }
        Ok(())
    }

    fn recover_pending(directory: &File, pending_name: &str, final_name: &str) -> Result<(), WorkflowJournalError> {
        let pending = open_optional_file_handle(directory, pending_name, OFlags::RDONLY)?;
        let final_file = open_optional_file_handle(directory, final_name, OFlags::RDONLY)?;
        match (pending, final_file) {
            (None, _) => Ok(()),
            (Some(pending), None) => {
                validate_private_file(&pending, None, 1)?;
                unlinkat(directory, pending_name, AtFlags::empty()).map_err(|_| WorkflowJournalError::Io)?;
                fsync(directory).map_err(|_| WorkflowJournalError::Io)
            }
            (Some(pending), Some(final_file)) => {
                let pending_stat = validate_private_file(&pending, None, 2)?;
                let final_stat = validate_private_file(&final_file, None, 2)?;
                if pending_stat.st_dev != final_stat.st_dev
                    || pending_stat.st_ino != final_stat.st_ino
                    || pending_stat.st_size != final_stat.st_size
                {
                    return Err(WorkflowJournalError::Corrupt);
                }
                unlinkat(directory, pending_name, AtFlags::empty()).map_err(|_| WorkflowJournalError::Io)?;
                fsync(directory).map_err(|_| WorkflowJournalError::Io)?;
                validate_private_file(&final_file, None, 1)?;
                Ok(())
            }
        }
    }

    fn replay_records<T: WorkflowJournalTransition>(
        records: &File,
        transaction_id: WorkflowTransactionId,
        intent_digest: [u8; 32],
    ) -> Result<(Vec<T>, u64, [u8; 32]), WorkflowJournalError> {
        let names = list_names(records)?;
        if names.len()
            > usize::try_from(MAX_WORKFLOW_JOURNAL_RECORDS).map_err(|_| WorkflowJournalError::LimitExceeded)?
        {
            return Err(WorkflowJournalError::LimitExceeded);
        }
        let mut expected_sequence = 1_u64;
        let mut previous_digest = [0_u8; 32];
        let mut transitions = Vec::with_capacity(names.len());
        for name in names {
            let sequence = parse_record_name(&name)?;
            if sequence != expected_sequence {
                return Err(WorkflowJournalError::Corrupt);
            }
            let bytes = read_file(records, &name, MAX_WORKFLOW_JOURNAL_RECORD_BYTES)?;
            let envelope: TransitionEnvelope<T> = decode_canonical(&bytes)?;
            if envelope.format_version != WORKFLOW_JOURNAL_FORMAT_VERSION
                || envelope.transaction_id != transaction_id
                || envelope.intent_digest != intent_digest
                || envelope.sequence != sequence
                || envelope.previous_record_digest != previous_digest
            {
                return Err(WorkflowJournalError::Corrupt);
            }
            envelope.transition.validate()?;
            previous_digest = sha256(&bytes)?;
            transitions.push(envelope.transition);
            expected_sequence = expected_sequence
                .checked_add(1)
                .ok_or(WorkflowJournalError::LimitExceeded)?;
        }
        Ok((transitions, expected_sequence.saturating_sub(1), previous_digest))
    }

    fn atomic_publish(
        directory: &File,
        pending_name: &str,
        final_name: &str,
        bytes: &[u8],
    ) -> Result<(), WorkflowJournalError> {
        if open_optional_file_handle(directory, pending_name, OFlags::RDONLY)?.is_some()
            || open_optional_file_handle(directory, final_name, OFlags::RDONLY)?.is_some()
        {
            return Err(WorkflowJournalError::Corrupt);
        }
        let (mut pending, created) = open_or_create_private_file(directory, pending_name, false)?;
        if !created {
            return Err(WorkflowJournalError::Corrupt);
        }
        validate_private_file(&pending, Some(0), 1)?;
        pending.write_all(bytes).map_err(|_| WorkflowJournalError::Io)?;
        pending.sync_all().map_err(|_| WorkflowJournalError::Io)?;
        linkat(directory, pending_name, directory, final_name, AtFlags::empty())
            .map_err(|_| WorkflowJournalError::Io)?;
        fsync(directory).map_err(|_| WorkflowJournalError::Io)?;
        unlinkat(directory, pending_name, AtFlags::empty()).map_err(|_| WorkflowJournalError::Io)?;
        fsync(directory).map_err(|_| WorkflowJournalError::Io)?;
        let final_bytes = read_file(directory, final_name, bytes.len())?;
        if final_bytes != bytes {
            return Err(WorkflowJournalError::Corrupt);
        }
        Ok(())
    }

    fn read_optional_file(
        directory: &File,
        name: &str,
        maximum_bytes: usize,
    ) -> Result<Option<Vec<u8>>, WorkflowJournalError> {
        let Some(mut file) = open_optional_file_handle(directory, name, OFlags::RDONLY)? else {
            return Ok(None);
        };
        validate_private_file(&file, None, 1)?;
        read_bounded(&mut file, maximum_bytes).map(Some)
    }

    fn read_file(directory: &File, name: &str, maximum_bytes: usize) -> Result<Vec<u8>, WorkflowJournalError> {
        read_optional_file(directory, name, maximum_bytes)?.ok_or(WorkflowJournalError::Corrupt)
    }

    fn read_bounded(file: &mut File, maximum_bytes: usize) -> Result<Vec<u8>, WorkflowJournalError> {
        let capacity = maximum_bytes
            .checked_add(1)
            .ok_or(WorkflowJournalError::LimitExceeded)?;
        let limit = u64::try_from(capacity).map_err(|_| WorkflowJournalError::LimitExceeded)?;
        let mut bytes = Vec::new();
        file.take(limit)
            .read_to_end(&mut bytes)
            .map_err(|_| WorkflowJournalError::Io)?;
        if bytes.is_empty() || bytes.len() > maximum_bytes {
            return Err(WorkflowJournalError::LimitExceeded);
        }
        Ok(bytes)
    }

    fn open_optional_file_handle(
        directory: &File,
        name: &str,
        access: OFlags,
    ) -> Result<Option<File>, WorkflowJournalError> {
        // NONBLOCK prevents a corrupt reserved name backed by a FIFO from
        // hanging the bounded worker before the file-type check can reject it.
        let flags = access | OFlags::CLOEXEC | OFlags::NOFOLLOW | OFlags::NONBLOCK;
        match openat(directory, name, flags, Mode::empty()) {
            Ok(fd) => Ok(Some(File::from(fd))),
            Err(error) if error == rustix::io::Errno::NOENT => Ok(None),
            Err(error) => Err(map_open_error(error)),
        }
    }

    fn validate_private_file(
        file: &File,
        expected_size: Option<u64>,
        expected_links: u64,
    ) -> Result<rustix::fs::Stat, WorkflowJournalError> {
        let stat = fstat(file).map_err(|_| WorkflowJournalError::Io)?;
        let size = u64::try_from(stat.st_size).map_err(|_| WorkflowJournalError::Corrupt)?;
        let links = stat_value_u64(stat.st_nlink).ok_or(WorkflowJournalError::Corrupt)?;
        if FileType::from_raw_mode(stat.st_mode) != FileType::RegularFile
            || stat.st_uid != rustix::process::geteuid().as_raw()
            || stat.st_mode & 0o777 != FILE_MODE
            || links != expected_links
            || expected_size.is_some_and(|expected| expected != size)
        {
            return Err(WorkflowJournalError::InsecurePath);
        }
        Ok(stat)
    }

    fn list_names(directory: &File) -> Result<BTreeSet<String>, WorkflowJournalError> {
        let mut names = BTreeSet::new();
        let entries = Dir::read_from(directory).map_err(|_| WorkflowJournalError::Io)?;
        for entry in entries {
            let entry = entry.map_err(|_| WorkflowJournalError::Io)?;
            let bytes = entry.file_name().to_bytes();
            if bytes == b"." || bytes == b".." {
                continue;
            }
            let name = std::str::from_utf8(bytes).map_err(|_| WorkflowJournalError::Corrupt)?;
            if !names.insert(name.to_owned()) {
                return Err(WorkflowJournalError::Corrupt);
            }
        }
        Ok(names)
    }

    fn parse_record_name(name: &str) -> Result<u64, WorkflowJournalError> {
        let Some(digits) = name.strip_suffix(".json") else {
            return Err(WorkflowJournalError::Corrupt);
        };
        parse_sequence_digits(digits)
    }

    fn parse_pending_record_name(name: &str) -> Result<Option<u64>, WorkflowJournalError> {
        let Some(rest) = name.strip_prefix('.') else {
            return Ok(None);
        };
        let Some(digits) = rest.strip_suffix(".pending") else {
            return Err(WorkflowJournalError::Corrupt);
        };
        parse_sequence_digits(digits).map(Some)
    }

    fn parse_sequence_digits(digits: &str) -> Result<u64, WorkflowJournalError> {
        if digits.len() != 20 || digits.bytes().any(|byte| !byte.is_ascii_digit()) {
            return Err(WorkflowJournalError::Corrupt);
        }
        let sequence = digits.parse::<u64>().map_err(|_| WorkflowJournalError::Corrupt)?;
        if record_name(sequence) != format!("{digits}.json") {
            return Err(WorkflowJournalError::Corrupt);
        }
        Ok(sequence)
    }

    fn record_name(sequence: u64) -> String {
        format!("{sequence:020}.json")
    }

    fn pending_record_name(sequence: u64) -> String {
        format!(".{sequence:020}.pending")
    }

    fn encode_bounded<T: Serialize>(value: &T, maximum_bytes: usize) -> Result<Vec<u8>, WorkflowJournalError> {
        let bytes = serde_json::to_vec(value).map_err(|_| WorkflowJournalError::Codec)?;
        if bytes.is_empty() || bytes.len() > maximum_bytes {
            return Err(WorkflowJournalError::LimitExceeded);
        }
        Ok(bytes)
    }

    fn decode_canonical<T: DeserializeOwned + Serialize>(bytes: &[u8]) -> Result<T, WorkflowJournalError> {
        let value = serde_json::from_slice(bytes).map_err(|_| WorkflowJournalError::Codec)?;
        if serde_json::to_vec(&value).map_err(|_| WorkflowJournalError::Codec)? != bytes {
            return Err(WorkflowJournalError::Codec);
        }
        Ok(value)
    }

    fn sha256(bytes: &[u8]) -> Result<[u8; 32], WorkflowJournalError> {
        let value = digest::digest(&digest::SHA256, bytes);
        let source = value.as_ref();
        if source.len() != 32 {
            return Err(WorkflowJournalError::Codec);
        }
        let mut output = [0_u8; 32];
        output.copy_from_slice(source);
        Ok(output)
    }

    fn generate_transaction_id() -> Result<WorkflowTransactionId, WorkflowJournalError> {
        let mut bytes = [0_u8; 32];
        getrandom::fill(&mut bytes).map_err(|_| WorkflowJournalError::RandomUnavailable)?;
        let identity = WorkflowTransactionId(bytes);
        if identity.is_zero() {
            return Err(WorkflowJournalError::RandomUnavailable);
        }
        Ok(identity)
    }

    fn map_open_error(error: rustix::io::Errno) -> WorkflowJournalError {
        if error == rustix::io::Errno::LOOP || error == rustix::io::Errno::NOTDIR {
            WorkflowJournalError::InsecurePath
        } else {
            WorkflowJournalError::Io
        }
    }
}

#[cfg(test)]
mod tests {
    use std::path::PathBuf;

    use super::*;

    #[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
    #[serde(deny_unknown_fields)]
    struct TestIntent {
        value: String,
    }

    #[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
    #[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
    enum TestTransition {
        EffectPrepared { action: u64 },
        EffectObserved { action: u64 },
    }

    impl sealed::Sealed for TestIntent {}
    impl WorkflowJournalIntent for TestIntent {
        type Transition = TestTransition;

        const KIND: LocalOperatorWorkflowKind = LocalOperatorWorkflowKind::RotateCredential;

        fn validate(&self) -> Result<(), WorkflowJournalError> {
            if self.value.is_empty() || self.value.len() > 128 {
                return Err(WorkflowJournalError::InvalidIntent);
            }
            Ok(())
        }
    }

    impl sealed::Sealed for TestTransition {}
    impl WorkflowJournalTransition for TestTransition {
        fn validate(&self) -> Result<(), WorkflowJournalError> {
            let action = match self {
                Self::EffectPrepared { action } | Self::EffectObserved { action } => *action,
            };
            if action == 0 {
                return Err(WorkflowJournalError::InvalidTransition);
            }
            Ok(())
        }
    }

    #[cfg(any(target_os = "linux", target_os = "macos"))]
    struct TestDirectory(PathBuf);

    #[cfg(any(target_os = "linux", target_os = "macos"))]
    impl TestDirectory {
        fn create() -> Result<Self, Box<dyn Error>> {
            use std::os::unix::fs::PermissionsExt as _;

            let mut nonce = [0_u8; 16];
            getrandom::fill(&mut nonce)?;
            let suffix: String = nonce.iter().map(|byte| format!("{byte:02x}")).collect();
            let path = std::env::temp_dir()
                .canonicalize()?
                .join(format!("lodestone-workflow-journal-{suffix}"));
            std::fs::create_dir(&path)?;
            std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o700))?;
            Ok(Self(path))
        }

        fn journal(&self) -> PathBuf {
            self.0.join("transaction")
        }
    }

    #[cfg(any(target_os = "linux", target_os = "macos"))]
    impl Drop for TestDirectory {
        fn drop(&mut self) {
            let _ = std::fs::remove_dir_all(&self.0);
        }
    }

    fn intent(value: &str) -> TestIntent {
        TestIntent {
            value: value.to_owned(),
        }
    }

    #[cfg(any(target_os = "linux", target_os = "macos"))]
    #[tokio::test]
    async fn create_append_and_exact_replay_are_canonical_and_private() -> Result<(), Box<dyn Error>> {
        use std::os::unix::fs::{MetadataExt as _, PermissionsExt as _};

        let directory = TestDirectory::create()?;
        let path = directory.journal();
        let replay = WorkflowJournal::<TestIntent>::open_or_create(&path, intent("same")).await?;
        assert!(replay.transitions().is_empty());
        let transaction_id = replay.journal().transaction_id();
        let first = replay
            .journal()
            .append(TestTransition::EffectPrepared { action: 7 })
            .await?;
        assert_eq!(first.sequence(), 1);
        drop(replay);

        let resumed = WorkflowJournal::<TestIntent>::open_or_create(&path, intent("same")).await?;
        assert_eq!(resumed.journal().transaction_id(), transaction_id);
        assert_eq!(resumed.transitions(), &[TestTransition::EffectPrepared { action: 7 }]);
        let second = resumed
            .journal()
            .append(TestTransition::EffectObserved { action: 7 })
            .await?;
        assert_eq!(second.sequence(), 2);

        let root = std::fs::metadata(&path)?;
        let intent_metadata = std::fs::metadata(path.join("intent.json"))?;
        let record_metadata = std::fs::metadata(path.join("records/00000000000000000001.json"))?;
        assert_eq!(root.permissions().mode() & 0o777, 0o700);
        assert_eq!(intent_metadata.permissions().mode() & 0o777, 0o600);
        assert_eq!(intent_metadata.nlink(), 1);
        assert_eq!(record_metadata.permissions().mode() & 0o777, 0o600);
        assert_eq!(record_metadata.nlink(), 1);
        Ok(())
    }

    #[cfg(any(target_os = "linux", target_os = "macos"))]
    #[tokio::test]
    async fn concurrent_writer_and_different_intent_fail_closed() -> Result<(), Box<dyn Error>> {
        let directory = TestDirectory::create()?;
        let path = directory.journal();
        let first = WorkflowJournal::<TestIntent>::open_or_create(&path, intent("same")).await?;
        let busy = WorkflowJournal::<TestIntent>::open_or_create(&path, intent("same")).await;
        assert!(matches!(busy, Err(WorkflowJournalError::Busy)));
        drop(first);
        let mismatch = WorkflowJournal::<TestIntent>::open_or_create(&path, intent("different")).await;
        match mismatch {
            Err(WorkflowJournalError::IntentMismatch) => {}
            Err(error) => return Err(format!("unexpected mismatch result: {error:?}").into()),
            Ok(_) => return Err("different intent unexpectedly reopened the journal".into()),
        }
        Ok(())
    }

    #[cfg(any(target_os = "linux", target_os = "macos"))]
    #[tokio::test]
    async fn noncanonical_record_gap_and_unknown_file_are_refused() -> Result<(), Box<dyn Error>> {
        use std::io::Write as _;

        let directory = TestDirectory::create()?;
        let path = directory.journal();
        let replay = WorkflowJournal::<TestIntent>::open_or_create(&path, intent("same")).await?;
        replay
            .journal()
            .append(TestTransition::EffectPrepared { action: 7 })
            .await?;
        drop(replay);

        let record = path.join("records/00000000000000000001.json");
        std::fs::OpenOptions::new()
            .append(true)
            .open(&record)?
            .write_all(b" ")?;
        let noncanonical = WorkflowJournal::<TestIntent>::open_or_create(&path, intent("same")).await;
        assert!(matches!(noncanonical, Err(WorkflowJournalError::Codec)));

        std::fs::remove_dir_all(&path)?;
        let replay = WorkflowJournal::<TestIntent>::open_or_create(&path, intent("same")).await?;
        replay
            .journal()
            .append(TestTransition::EffectPrepared { action: 7 })
            .await?;
        drop(replay);
        std::fs::rename(
            path.join("records/00000000000000000001.json"),
            path.join("records/00000000000000000002.json"),
        )?;
        let gap = WorkflowJournal::<TestIntent>::open_or_create(&path, intent("same")).await;
        assert!(matches!(gap, Err(WorkflowJournalError::Corrupt)));

        std::fs::remove_dir_all(&path)?;
        let replay = WorkflowJournal::<TestIntent>::open_or_create(&path, intent("same")).await?;
        drop(replay);
        std::fs::write(path.join("unexpected"), b"unexpected")?;
        let unknown = WorkflowJournal::<TestIntent>::open_or_create(&path, intent("same")).await;
        assert!(matches!(unknown, Err(WorkflowJournalError::Corrupt)));
        Ok(())
    }

    #[cfg(any(target_os = "linux", target_os = "macos"))]
    #[tokio::test]
    async fn known_pending_links_are_recovered_but_symlinks_are_refused() -> Result<(), Box<dyn Error>> {
        use std::os::unix::fs::{PermissionsExt as _, symlink};

        let directory = TestDirectory::create()?;
        let path = directory.journal();
        let replay = WorkflowJournal::<TestIntent>::open_or_create(&path, intent("same")).await?;
        replay
            .journal()
            .append(TestTransition::EffectPrepared { action: 7 })
            .await?;
        drop(replay);

        let final_record = path.join("records/00000000000000000001.json");
        let pending_record = path.join("records/.00000000000000000001.pending");
        std::fs::hard_link(&final_record, &pending_record)?;
        let resumed = WorkflowJournal::<TestIntent>::open_or_create(&path, intent("same")).await?;
        assert!(!pending_record.exists());
        assert_eq!(resumed.transitions().len(), 1);
        drop(resumed);

        std::fs::remove_dir_all(&path)?;
        let target = directory.0.join("target");
        std::fs::create_dir(&target)?;
        std::fs::set_permissions(&target, std::fs::Permissions::from_mode(0o700))?;
        symlink(&target, &path)?;
        let symlinked = WorkflowJournal::<TestIntent>::open_or_create(&path, intent("same")).await;
        assert!(matches!(symlinked, Err(WorkflowJournalError::InsecurePath)));
        Ok(())
    }

    #[cfg(any(target_os = "linux", target_os = "macos"))]
    #[tokio::test]
    async fn reserved_fifo_is_refused_without_blocking_the_worker() -> Result<(), Box<dyn Error>> {
        let directory = TestDirectory::create()?;
        let path = directory.journal();
        let replay = WorkflowJournal::<TestIntent>::open_or_create(&path, intent("same")).await?;
        drop(replay);

        let intent_path = path.join("intent.json");
        std::fs::remove_file(&intent_path)?;
        let status = std::process::Command::new("mkfifo").arg(&intent_path).status()?;
        assert!(status.success());

        let corrupt = WorkflowJournal::<TestIntent>::open_or_create(&path, intent("same")).await;
        assert!(matches!(corrupt, Err(WorkflowJournalError::InsecurePath)));
        Ok(())
    }

    #[cfg(any(target_os = "linux", target_os = "macos"))]
    #[tokio::test]
    async fn uncommitted_pending_inode_is_removed_without_becoming_history() -> Result<(), Box<dyn Error>> {
        use std::os::unix::fs::PermissionsExt as _;

        let directory = TestDirectory::create()?;
        let path = directory.journal();
        let replay = WorkflowJournal::<TestIntent>::open_or_create(&path, intent("same")).await?;
        drop(replay);

        let pending = path.join("records/.00000000000000000001.pending");
        std::fs::write(&pending, b"partial")?;
        std::fs::set_permissions(&pending, std::fs::Permissions::from_mode(0o600))?;
        let resumed = WorkflowJournal::<TestIntent>::open_or_create(&path, intent("same")).await?;
        assert!(!pending.exists());
        assert!(resumed.transitions().is_empty());
        Ok(())
    }

    #[cfg(any(target_os = "linux", target_os = "macos"))]
    #[tokio::test]
    async fn unknown_intent_format_is_refused_without_compatibility_decode() -> Result<(), Box<dyn Error>> {
        let directory = TestDirectory::create()?;
        let path = directory.journal();
        let replay = WorkflowJournal::<TestIntent>::open_or_create(&path, intent("same")).await?;
        drop(replay);

        let intent_path = path.join("intent.json");
        let bytes = std::fs::read(&intent_path)?;
        let mut envelope: IntentEnvelope<TestIntent> = serde_json::from_slice(&bytes)?;
        envelope.format_version = WORKFLOW_JOURNAL_FORMAT_VERSION.saturating_add(1);
        std::fs::write(&intent_path, serde_json::to_vec(&envelope)?)?;
        let result = WorkflowJournal::<TestIntent>::open_or_create(&path, intent("same")).await;
        assert!(matches!(result, Err(WorkflowJournalError::Corrupt)));
        Ok(())
    }
}