pi_append_log 0.3.0

//! 使用异步文件命名空间的追加日志实现和默认布局。

use std::collections::HashMap;
use std::path::{Component, Path, PathBuf};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex, MutexGuard, OnceLock, Weak};

use futures_core::Stream;
use pi_async_fs::{
    CreateTargetEvidence, FileAccessMode, FileFlushMode, FileIo, FileNamespace, LocalFileNamespace,
    ReadGrowthLimit, RenameCommitEvidence,
};
use pi_result::{ClassifyErrorKind, InteropResultExt};
use tracing::warn;

use crate::format::BlockDecoder;
use crate::storage::{
    AppendLog, AppendLogBuilder, AppendLogVisitor, BlockVisitContext, BuildResult, Layout,
    ReadOrder,
};

/// 固定八位十进制文件名的默认布局。
#[derive(Clone)]
pub struct DefaultFileLayout {
    root: PathBuf,
}

impl DefaultFileLayout {
    /// 创建指定目录下的默认布局。
    pub fn new(root: impl Into<PathBuf>) -> Self {
        Self { root: root.into() }
    }
}

impl Layout for DefaultFileLayout {
    type StructureId = u64;
    type Name = PathBuf;

    fn segment_name(&self, structure_id: &Self::StructureId) -> Self::Name {
        self.root.join(format!("{structure_id:08}"))
    }

    fn archive_name(&self, structure_id: &Self::StructureId) -> Self::Name {
        self.root.join(format!("{structure_id:08}.archive"))
    }

    fn parse_segment_name(&self, name: &Self::Name) -> Option<Self::StructureId> {
        parse_name(name, &self.root, "segment")
    }

    fn parse_archive_name(&self, name: &Self::Name) -> Option<Self::StructureId> {
        parse_name(name, &self.root, "archive")
    }
}

fn parse_name(path: &Path, root: &Path, state: &str) -> Option<u64> {
    let relative = path.strip_prefix(root).ok()?;
    if relative.parent()?.as_os_str() != "" {
        return None;
    }
    let stem = relative.file_stem()?.to_str()?;
    if stem.len() != 8 || !stem.bytes().all(|byte| byte.is_ascii_digit()) {
        return None;
    }
    if state == "segment" {
        if relative.extension().is_some() {
            return None;
        }
    } else if relative.extension()?.to_str()? != state {
        return None;
    }
    let id = stem.parse::<u64>().ok()?;
    (id > 0).then_some(id)
}

struct NamespaceControl {
    operation: Arc<async_lock::Mutex<()>>,
    ready: AtomicBool,
    active_id: Mutex<Option<u64>>,
}

static OPERATIONS: OnceLock<Mutex<HashMap<PathBuf, Weak<NamespaceControl>>>> = OnceLock::new();

fn shared_control(root: &Path) -> Arc<NamespaceControl> {
    let key = stable_root_key(root);
    let registry = OPERATIONS.get_or_init(|| Mutex::new(HashMap::new()));
    let mut operations = registry
        .lock()
        .unwrap_or_else(|poisoned| poisoned.into_inner());
    if let Some(control) = operations.get(&key).and_then(Weak::upgrade) {
        return control;
    }
    let control = Arc::new(NamespaceControl {
        operation: Arc::new(async_lock::Mutex::new(())),
        ready: AtomicBool::new(true),
        active_id: Mutex::new(None),
    });
    operations.insert(key, Arc::downgrade(&control));
    control
}

fn stable_root_key(root: &Path) -> PathBuf {
    let mut normalized = PathBuf::new();
    for component in root.components() {
        match component {
            Component::CurDir => {}
            Component::ParentDir => {
                normalized.pop();
            }
            component => normalized.push(component.as_os_str()),
        }
    }
    #[cfg(windows)]
    {
        PathBuf::from(normalized.to_string_lossy().to_ascii_lowercase())
    }
    #[cfg(not(windows))]
    {
        normalized
    }
}

fn same_root_path(left: &Path, right: &Path) -> bool {
    #[cfg(windows)]
    {
        left.to_string_lossy()
            .eq_ignore_ascii_case(&right.to_string_lossy())
    }
    #[cfg(not(windows))]
    {
        left == right
    }
}

fn validate_direct_child(root: &Path, path: &Path) -> pi_result::Result<()> {
    if path.file_name().is_some()
        && path
            .parent()
            .is_some_and(|parent| same_root_path(parent, root))
    {
        Ok(())
    } else {
        Err(kind_error(pi_result::ErrorKind::InvalidInput))
    }
}

fn validate_layout_paths_for_id<L>(root: &Path, layout: &L, id: u64) -> pi_result::Result<()>
where
    L: Layout<StructureId = u64, Name = PathBuf>,
{
    validate_direct_child(root, &layout.segment_name(&id))?;
    validate_direct_child(root, &layout.archive_name(&id))
}

/// 已封闭文件的归档句柄。
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct FileClosed {
    /// 产生此句柄的文件存储命名空间。
    namespace: PathBuf,
    /// 已封闭文件的八位逻辑编号。
    structure_id: u64,
}

/// 文件追加日志的初始化 Builder。
///
/// root 必须是绝对路径，并在 build 到 storage 的整个生命周期保持稳定。同一对象的
/// symlink、junction 或其它不同 locator 被视为不同 namespace；调用方不得通过别名并发访问。
pub struct FileAppendLogBuilder<L> {
    root: PathBuf,
    layout: L,
    namespace: LocalFileNamespace,
}

impl<L> FileAppendLogBuilder<L> {
    /// 创建文件追加日志 Builder。
    pub fn new(root: impl Into<PathBuf>, layout: L) -> Self {
        let root = root.into();
        Self {
            root,
            layout,
            namespace: LocalFileNamespace::new(),
        }
    }
}

/// 文件追加日志的运行期实现。
pub struct FileAppendLog<L> {
    state: Arc<Mutex<FileState<L>>>,
    control: Arc<NamespaceControl>,
    namespace: LocalFileNamespace,
}

struct FileState<L> {
    root: PathBuf,
    layout: L,
    active_id: u64,
    health: FileHealth,
}

#[derive(Clone, Copy, Eq, PartialEq)]
enum FileHealth {
    Ready,
    Invalid,
}

struct OperationLease {
    _guard: async_lock::MutexGuardArc<()>,
    _control: Arc<NamespaceControl>,
}

struct GuardedWriteBuffer<B> {
    buffer: B,
    lease: Arc<OperationLease>,
}

struct GuardedDetached<B> {
    buffer: B,
    _lease: Arc<OperationLease>,
}

struct GuardedRecovery<R> {
    recovery: R,
    lease: Arc<OperationLease>,
}

impl<B: AsRef<[u8]>> AsRef<[u8]> for GuardedWriteBuffer<B> {
    fn as_ref(&self) -> &[u8] {
        self.buffer.as_ref()
    }
}

impl<B: AsRef<[u8]>> AsRef<[u8]> for GuardedDetached<B> {
    fn as_ref(&self) -> &[u8] {
        self.buffer.as_ref()
    }
}

impl<B: pi_async_fs::DetachableWriteBuffer> pi_async_fs::DetachableWriteBuffer
    for GuardedWriteBuffer<B>
{
    type Detached = GuardedDetached<B::Detached>;
    type Recovery = GuardedRecovery<B::Recovery>;

    fn try_detach(
        self,
    ) -> pi_result::RawResult<(Self::Detached, Self::Recovery), pi_async_fs::BufferFailure<Self>>
    {
        match self.buffer.try_detach() {
            Ok((buffer, recovery)) => Ok((
                GuardedDetached {
                    buffer,
                    _lease: Arc::clone(&self.lease),
                },
                GuardedRecovery {
                    recovery,
                    lease: self.lease,
                },
            )),
            Err(failure) => {
                let (error, buffer, progress) = failure.into_parts();
                Err(pi_async_fs::BufferFailure::new(
                    error,
                    Self {
                        buffer,
                        lease: self.lease,
                    },
                    progress,
                ))
            }
        }
    }

    fn recover_from_detached(detached: Self::Detached, recovery: Self::Recovery) -> Self {
        Self {
            buffer: B::recover_from_detached(detached.buffer, recovery.recovery),
            lease: recovery.lease,
        }
    }
}

struct AppendCommitGuard<L> {
    state: Arc<Mutex<FileState<L>>>,
    control: Arc<NamespaceControl>,
    armed: bool,
}

impl<L> AppendCommitGuard<L> {
    fn new(state: Arc<Mutex<FileState<L>>>, control: Arc<NamespaceControl>) -> Self {
        Self {
            state,
            control,
            armed: true,
        }
    }

    fn disarm(&mut self) {
        self.armed = false;
    }
}

impl<L> Drop for AppendCommitGuard<L> {
    fn drop(&mut self) {
        if self.armed {
            self.control.ready.store(false, Ordering::Release);
            self.state
                .lock()
                .unwrap_or_else(|poisoned| poisoned.into_inner())
                .health = FileHealth::Invalid;
        }
    }
}

impl<L> AppendLogBuilder for FileAppendLogBuilder<L>
where
    L: Layout<StructureId = u64, Name = PathBuf> + Clone + Send + Sync,
{
    type Storage = FileAppendLog<L>;

    async fn build<D, V>(
        self,
        decoder: &D,
        order: ReadOrder,
        visitor: &mut V,
    ) -> pi_result::Result<BuildResult<Self::Storage>>
    where
        D: BlockDecoder + Send + Sync,
        V: AppendLogVisitor + Send,
    {
        if !self.root.is_absolute() {
            return Err(kind_error(pi_result::ErrorKind::InvalidInput));
        }
        validate_layout_paths_for_id(&self.root, &self.layout, 1)?;
        let control = shared_control(&self.root);
        let build_operation = Arc::clone(&control.operation);
        let _operation = build_operation.lock().await;
        if !control.ready.load(Ordering::Acquire) {
            return Err(kind_error(pi_result::ErrorKind::InvalidState));
        }
        async {
            if let Err(failure) = self.namespace.create_dir_all(&self.root).await {
                return Err(failure.into_parts().0);
            }
            let discovered = discover(&self.namespace, &self.root, &self.layout).await?;
            let discovered_active_id = match discovered.active_id {
                Some(id) => id,
                None => {
                    let id = discovered.max_id.checked_add(1).unwrap_or(1);
                    validate_layout_paths_for_id(&self.root, &self.layout, id)?;
                    let path = self.layout.segment_name(&id);
                    let mut file =
                        match create_new(&self.namespace, &path, FileAccessMode::Append).await {
                            CreateOutcome::Created(file) => file,
                            CreateOutcome::Failed { error, evidence } => {
                                return Err(match evidence {
                                    CreateTargetEvidence::NotCreatedByOperation
                                        if error.classify_error_kind()
                                            == pi_result::ErrorKind::AlreadyExists =>
                                    {
                                        kind_error(pi_result::ErrorKind::Conflict)
                                    }
                                    _ => error,
                                });
                            }
                        };
                    file.flush(FileFlushMode::DataAndMetadata).await?;
                    id
                }
            };
            let active_id = {
                let mut shared_active_id = control
                    .active_id
                    .lock()
                    .unwrap_or_else(|poisoned| poisoned.into_inner());
                match *shared_active_id {
                    Some(id) => {
                        if discovered.active_id != Some(id) {
                            return Err(kind_error(pi_result::ErrorKind::Conflict));
                        }
                        validate_layout_paths_for_id(&self.root, &self.layout, id)?;
                        id
                    }
                    None => {
                        *shared_active_id = Some(discovered_active_id);
                        discovered_active_id
                    }
                }
            };

            let recovered_ids = discovered.closed_ids.clone();
            let mut structures = recovered_ids.clone();
            structures.push(active_id);
            structures.sort_unstable();
            let structure_count = structures.len();
            let mut stopped = false;
            for position in 0..structure_count {
                let structure_index = if matches!(order, ReadOrder::Forward) {
                    position
                } else {
                    structure_count - position - 1
                };
                let id = structures[structure_index];
                validate_layout_paths_for_id(&self.root, &self.layout, id)?;
                let is_active = id == active_id;
                let path = self.layout.segment_name(&id);
                let bytes = read_structure(&self.namespace, &path, decoder, is_active).await?;
                if !stopped {
                    stopped = visit_structure(&bytes, decoder, order, visitor)?;
                }
            }

            let namespace_path = self.root.clone();
            let storage = FileAppendLog {
                state: Arc::new(Mutex::new(FileState {
                    root: self.root,
                    layout: self.layout,
                    active_id,
                    health: FileHealth::Ready,
                })),
                control,
                namespace: self.namespace,
            };
            let recovered_closed = recovered_ids
                .into_iter()
                .map(|structure_id| FileClosed {
                    namespace: namespace_path.clone(),
                    structure_id,
                })
                .collect();
            Ok(BuildResult {
                storage,
                recovered_closed,
            })
        }
        .await
    }
}

impl<L> AppendLog for FileAppendLog<L>
where
    L: Layout<StructureId = u64, Name = PathBuf> + Clone + Send + Sync,
{
    type Closed = FileClosed;

    async fn append<'a, B>(
        &'a self,
        block: B,
        options: crate::AppendOptions,
    ) -> pi_result::Result<u64>
    where
        B: pi_async_fs::DetachableWriteBuffer + Send + 'a,
        B::Detached: Send,
        B::Recovery: Send + 'a,
    {
        async {
            if block.as_ref().is_empty() {
                return Err(kind_error(pi_result::ErrorKind::InvalidInput));
            }
            let operation = Arc::clone(&self.control.operation);
            let lease = Arc::new(OperationLease {
                _guard: operation.lock_arc().await,
                _control: Arc::clone(&self.control),
            });
            self.ensure_ready().await?;
            let path = self.active_path()?;
            let mut file = self.open_active_append(&path).await?;
            let mut guard =
                AppendCommitGuard::new(Arc::clone(&self.state), Arc::clone(&self.control));
            let guarded = GuardedWriteBuffer {
                buffer: block,
                lease: Arc::clone(&lease),
            };
            if let Err(failure) = file.append(guarded).await {
                return Err(failure.into_parts().0);
            }
            if options.durable {
                file.flush(FileFlushMode::DataAndMetadata).await?;
            }
            let size = file.byte_len().await?;
            guard.disarm();
            Ok(size)
        }
        .await
    }

    async fn append_stream<'a, S, B>(
        &'a self,
        stream: S,
        options: crate::AppendOptions,
    ) -> pi_result::Result<u64>
    where
        S: Stream<Item = pi_result::Result<B>> + Send + 'a,
        B: pi_async_fs::DetachableWriteBuffer + Send + 'a,
        B::Detached: Send,
        B::Recovery: Send + 'a,
    {
        async {
            let operation = Arc::clone(&self.control.operation);
            let lease = Arc::new(OperationLease {
                _guard: operation.lock_arc().await,
                _control: Arc::clone(&self.control),
            });
            self.ensure_ready().await?;
            let mut stream = std::pin::pin!(stream);
            let first = loop {
                match std::future::poll_fn(|context| stream.as_mut().poll_next(context)).await {
                    Some(Ok(block)) if !block.as_ref().is_empty() => break block,
                    Some(Ok(_)) => {}
                    Some(Err(error)) => return Err(error),
                    None => return Err(kind_error(pi_result::ErrorKind::InvalidInput)),
                }
            };
            let path = self.active_path()?;
            let mut file = self.open_active_append(&path).await?;
            let first = GuardedWriteBuffer {
                buffer: first,
                lease: Arc::clone(&lease),
            };
            let mut guard =
                AppendCommitGuard::new(Arc::clone(&self.state), Arc::clone(&self.control));
            if let Err(failure) = file.append(first).await {
                return Err(failure.into_parts().0);
            }
            while let Some(item) =
                std::future::poll_fn(|context| stream.as_mut().poll_next(context)).await
            {
                let block = item?;
                if block.as_ref().is_empty() {
                    continue;
                }
                let block = GuardedWriteBuffer {
                    buffer: block,
                    lease: Arc::clone(&lease),
                };
                if let Err(failure) = file.append(block).await {
                    return Err(failure.into_parts().0);
                }
            }
            if options.durable {
                file.flush(FileFlushMode::DataAndMetadata).await?;
            }
            let size = file.byte_len().await?;
            guard.disarm();
            Ok(size)
        }
        .await
    }

    /// 错误或取消可能使实例进入 Invalid；后续操作返回 InvalidState，调用方须重建。
    async fn rotate(&self) -> pi_result::Result<Option<Self::Closed>> {
        let _operation = self.control.operation.lock().await;
        self.ensure_ready().await?;
        let (active_id, active_path, next_id, next_path, root) = {
            let state = self.lock_state()?;
            let active_id = self.shared_active_id()?;
            validate_layout_paths_for_id(&state.root, &state.layout, active_id)?;
            let next_id = active_id
                .checked_add(1)
                .ok_or_else(|| kind_error(pi_result::ErrorKind::InvalidState))?;
            validate_layout_paths_for_id(&state.root, &state.layout, next_id)?;
            (
                active_id,
                state.layout.segment_name(&active_id),
                next_id,
                state.layout.segment_name(&next_id),
                state.root.clone(),
            )
        };
        let length = match flush_append_barrier(&self.namespace, &active_path).await {
            Ok(length) => length,
            Err(error) => {
                self.set_health(FileHealth::Invalid)?;
                return Err(error);
            }
        };
        if length == 0 {
            return Ok(None);
        }

        match create_new(&self.namespace, &next_path, FileAccessMode::Append).await {
            CreateOutcome::Created(mut file) => {
                if let Err(error) = file.flush(FileFlushMode::DataAndMetadata).await {
                    self.set_health(FileHealth::Invalid)?;
                    return Err(error);
                }
                *self
                    .control
                    .active_id
                    .lock()
                    .unwrap_or_else(|poisoned| poisoned.into_inner()) = Some(next_id);
                self.lock_state()?.active_id = next_id;
                self.set_health(FileHealth::Ready)?;
                Ok(Some(FileClosed {
                    namespace: root,
                    structure_id: active_id,
                }))
            }
            CreateOutcome::Failed { error, evidence: _ } => {
                self.set_health(FileHealth::Invalid)?;
                Err(error)
            }
        }
    }

    async fn archive(&self, closed: Self::Closed) -> pi_result::Result<()> {
        let _operation = self.control.operation.lock().await;
        self.ensure_ready().await?;
        let (segment_path, archive_path) = {
            let state = self.lock_state()?;
            if !same_root_path(&closed.namespace, &state.root)
                || closed.structure_id == self.shared_active_id()?
            {
                return Err(kind_error(pi_result::ErrorKind::InvalidInput));
            }
            validate_layout_paths_for_id(&state.root, &state.layout, closed.structure_id)?;
            (
                state.layout.segment_name(&closed.structure_id),
                state.layout.archive_name(&closed.structure_id),
            )
        };
        match observe_archive_state(&self.namespace, &segment_path, &archive_path).await? {
            ArchiveState::Both => Err(kind_error(pi_result::ErrorKind::Conflict)),
            ArchiveState::Neither => Err(kind_error(pi_result::ErrorKind::NotFound)),
            ArchiveState::ArchiveOnly => {
                match observe_archive_state(&self.namespace, &segment_path, &archive_path).await? {
                    ArchiveState::ArchiveOnly => Ok(()),
                    ArchiveState::Both => Err(kind_error(pi_result::ErrorKind::Conflict)),
                    ArchiveState::Neither => Err(kind_error(pi_result::ErrorKind::NotFound)),
                    ArchiveState::SegmentOnly => Err(kind_error(pi_result::ErrorKind::Conflict)),
                }
            }
            ArchiveState::SegmentOnly => {
                match rename(&self.namespace, &segment_path, &archive_path).await {
                    RenameOutcome::Renamed => Ok(()),
                    RenameOutcome::Failed { error, evidence } => match evidence {
                        RenameCommitEvidence::RenamedByOperation => Ok(()),
                        RenameCommitEvidence::NotRenamedByOperation => Err(error),
                        RenameCommitEvidence::Unknown => {
                            match observe_archive_state(
                                &self.namespace,
                                &segment_path,
                                &archive_path,
                            )
                            .await
                            {
                                Ok(ArchiveState::ArchiveOnly) => Ok(()),
                                Ok(ArchiveState::Both) => {
                                    Err(kind_error(pi_result::ErrorKind::Conflict))
                                }
                                Ok(ArchiveState::Neither) => {
                                    Err(kind_error(pi_result::ErrorKind::NotFound))
                                }
                                Ok(ArchiveState::SegmentOnly) | Err(_) => Err(error),
                            }
                        }
                        _ => Err(error),
                    },
                }
            }
        }
    }
}

#[derive(Clone, Copy)]
enum ArchiveState {
    SegmentOnly,
    ArchiveOnly,
    Both,
    Neither,
}

async fn observe_archive_state(
    namespace: &LocalFileNamespace,
    segment: &Path,
    archive: &Path,
) -> pi_result::Result<ArchiveState> {
    let segment_exists = namespace.try_exists(&segment.to_path_buf()).await?;
    let archive_exists = namespace.try_exists(&archive.to_path_buf()).await?;
    Ok(match (segment_exists, archive_exists) {
        (true, true) => ArchiveState::Both,
        (true, false) => ArchiveState::SegmentOnly,
        (false, true) => ArchiveState::ArchiveOnly,
        (false, false) => ArchiveState::Neither,
    })
}

async fn discover<L>(
    namespace: &LocalFileNamespace,
    root: &Path,
    layout: &L,
) -> pi_result::Result<Discovered>
where
    L: Layout<StructureId = u64, Name = PathBuf>,
{
    let stream = namespace.read_dir(root.to_path_buf()).await?;
    let mut stream = std::pin::pin!(stream);
    let mut segment_ids = Vec::new();
    let mut max_id = 0;
    while let Some(entry) = std::future::poll_fn(|context| stream.as_mut().poll_next(context)).await
    {
        let path = entry?.locator;
        if let Some(id) = layout.parse_segment_name(&path) {
            validate_layout_paths_for_id(root, layout, id)?;
            if segment_ids.contains(&id) {
                return Err(kind_error(pi_result::ErrorKind::Conflict));
            }
            segment_ids.push(id);
            max_id = max_id.max(id);
        } else if layout.parse_archive_name(&path).is_some() {
            // Archives do not participate in recovery or next-id allocation.
        }
    }
    segment_ids.sort_unstable();
    segment_ids.dedup();
    let active_id = segment_ids.last().copied();
    let closed_ids = segment_ids
        .into_iter()
        .filter(|id| Some(*id) != active_id)
        .collect();
    Ok(Discovered {
        active_id,
        closed_ids,
        max_id,
    })
}

// build 调用方必须独占 namespace；EOF 与长度复核只检测意外受管变化，不能防御
// 不合作的同长度路径替换。Truncate 资源由 pi_async_fs 绑定打开时的稳定身份。
async fn read_structure<D: BlockDecoder>(
    namespace: &LocalFileNamespace,
    path: &Path,
    decoder: &D,
    active: bool,
) -> pi_result::Result<Vec<u8>> {
    let mut file = namespace
        .open(&path.to_path_buf(), FileAccessMode::Read)
        .await?;
    let original_len = file.byte_len().await?;
    let read_limit = original_len
        .checked_add(1)
        .and_then(|length| usize::try_from(length).ok())
        .ok_or_else(|| kind_error(pi_result::ErrorKind::ResourceExhausted))?;
    let mut bytes = Vec::new();
    let outcome = file
        .read_to_end(&mut bytes, ReadGrowthLimit::new(read_limit))
        .await?;
    drop(file);
    if !outcome.is_end_of_file() || bytes.len() as u64 != original_len {
        return Err(kind_error(pi_result::ErrorKind::Conflict));
    }

    let boundary = match decoder.find_last_complete(&bytes)? {
        Some(end) => end,
        None if bytes.is_empty() => return Ok(bytes),
        None if active => 0,
        None => return Err(kind_error(pi_result::ErrorKind::Corrupted)),
    };
    if boundary < bytes.len() {
        if !active {
            return Err(kind_error(pi_result::ErrorKind::Corrupted));
        }
        truncate_structure(namespace, path, original_len, boundary as u64).await?;
        bytes.truncate(boundary);
    }
    validate_structure(&bytes, decoder)?;
    Ok(bytes)
}

async fn truncate_structure(
    namespace: &LocalFileNamespace,
    path: &Path,
    original_len: u64,
    recovered_len: u64,
) -> pi_result::Result<()> {
    let mut file = namespace
        .open(&path.to_path_buf(), FileAccessMode::Truncate)
        .await?;
    if file.byte_len().await? != original_len {
        return Err(kind_error(pi_result::ErrorKind::Conflict));
    }
    file.truncate(recovered_len).await?;
    file.flush(FileFlushMode::DataAndMetadata).await?;
    let structure = path
        .file_name()
        .map(|name| name.to_string_lossy())
        .unwrap_or_default();
    warn!(
        structure = %structure,
        original_len,
        recovered_len,
        truncated_bytes = original_len - recovered_len,
        "truncated invalid active append-log tail"
    );
    Ok(())
}

async fn flush_append_barrier(
    namespace: &LocalFileNamespace,
    path: &Path,
) -> pi_result::Result<u64> {
    let mut file = namespace
        .open(&path.to_path_buf(), FileAccessMode::Append)
        .await?;
    file.flush(FileFlushMode::DataAndMetadata).await?;
    file.byte_len().await
}

enum CreateOutcome {
    Created(pi_async_fs::LocalFile),
    Failed {
        error: pi_result::Error,
        evidence: CreateTargetEvidence,
    },
}

async fn create_new(
    namespace: &LocalFileNamespace,
    path: &Path,
    access: FileAccessMode,
) -> CreateOutcome {
    match namespace.create_new(&path.to_path_buf(), access).await {
        Ok(file) => CreateOutcome::Created(file),
        Err(failure) => {
            let (error, evidence) = failure.into_parts();
            CreateOutcome::Failed { error, evidence }
        }
    }
}

enum RenameOutcome {
    Renamed,
    Failed {
        error: pi_result::Error,
        evidence: RenameCommitEvidence,
    },
}

async fn rename(
    namespace: &LocalFileNamespace,
    source: &Path,
    destination: &Path,
) -> RenameOutcome {
    match namespace
        .rename(&source.to_path_buf(), &destination.to_path_buf())
        .await
    {
        Ok(()) => RenameOutcome::Renamed,
        Err(failure) => {
            let (error, evidence) = failure.into_parts();
            RenameOutcome::Failed { error, evidence }
        }
    }
}

struct Discovered {
    active_id: Option<u64>,
    closed_ids: Vec<u64>,
    max_id: u64,
}

fn validate_structure<D: BlockDecoder>(bytes: &[u8], decoder: &D) -> pi_result::Result<()> {
    let mut offset = 0;
    while offset < bytes.len() {
        let decoded = decoder.decode_forward(&bytes[offset..])?;
        offset = offset
            .checked_add(decoded.encoded_len())
            .ok_or_else(|| kind_error(pi_result::ErrorKind::Corrupted))?;
    }
    Ok(())
}

fn visit_structure<D, V>(
    bytes: &[u8],
    decoder: &D,
    order: ReadOrder,
    visitor: &mut V,
) -> pi_result::Result<bool>
where
    D: BlockDecoder,
    V: AppendLogVisitor + Send,
{
    let mut ranges = Vec::new();
    let mut offset = 0;
    while offset < bytes.len() {
        let decoded = decoder.decode_forward(&bytes[offset..])?;
        let next_offset = offset
            .checked_add(decoded.encoded_len())
            .ok_or_else(|| kind_error(pi_result::ErrorKind::Corrupted))?;
        ranges.push((offset, next_offset));
        offset = next_offset;
    }
    if matches!(order, ReadOrder::Backward) {
        ranges.reverse();
    }
    for (index, (start, end)) in ranges.iter().enumerate() {
        let is_backward = matches!(order, ReadOrder::Backward);
        let is_first = if is_backward {
            index + 1 == ranges.len()
        } else {
            index == 0
        };
        let is_last = if is_backward {
            index == 0
        } else {
            index + 1 == ranges.len()
        };
        if visitor.visit(
            &bytes[*start..*end],
            BlockVisitContext {
                is_first_in_structure: is_first,
                is_last_in_structure: is_last,
            },
        )? {
            return Ok(true);
        }
    }
    Ok(false)
}

impl<L> FileAppendLog<L> {
    fn shared_active_id(&self) -> pi_result::Result<u64> {
        self.control
            .active_id
            .lock()
            .map_err(|_| kind_error(pi_result::ErrorKind::InvalidState))?
            .as_ref()
            .copied()
            .ok_or_else(|| kind_error(pi_result::ErrorKind::InvalidState))
    }

    fn lock_state(&self) -> pi_result::Result<MutexGuard<'_, FileState<L>>> {
        self.state
            .lock()
            .into_external_error_by(|_| pi_result::ErrorKind::InvalidState)
    }

    // 仅防御实例已知部分成功或活动名称缺失，不防御不合作方的同路径替换。
    async fn ensure_ready(&self) -> pi_result::Result<()>
    where
        L: Layout<StructureId = u64, Name = PathBuf>,
    {
        if !self.control.ready.load(Ordering::Acquire) {
            return Err(kind_error(pi_result::ErrorKind::InvalidState));
        }
        let path = {
            let state = self.lock_state()?;
            if state.health != FileHealth::Ready {
                return Err(kind_error(pi_result::ErrorKind::InvalidState));
            }
            state.layout.segment_name(&self.shared_active_id()?)
        };
        match self.namespace.try_exists(&path).await {
            Ok(true) => Ok(()),
            Ok(false) => {
                self.set_health(FileHealth::Invalid)?;
                Err(kind_error(pi_result::ErrorKind::InvalidState))
            }
            Err(error) => {
                self.set_health(FileHealth::Invalid)?;
                Err(error)
            }
        }
    }

    fn set_health(&self, health: FileHealth) -> pi_result::Result<()> {
        self.control
            .ready
            .store(health == FileHealth::Ready, Ordering::Release);
        self.lock_state()?.health = health;
        Ok(())
    }

    async fn open_active_append(&self, path: &Path) -> pi_result::Result<pi_async_fs::LocalFile> {
        match self
            .namespace
            .open(&path.to_path_buf(), FileAccessMode::Append)
            .await
        {
            Ok(file) => Ok(file),
            Err(error) => {
                self.set_health(FileHealth::Invalid)?;
                Err(error)
            }
        }
    }

    fn active_path(&self) -> pi_result::Result<PathBuf>
    where
        L: Layout<StructureId = u64, Name = PathBuf>,
    {
        let state = self.lock_state()?;
        Ok(state.layout.segment_name(&self.shared_active_id()?))
    }
}

fn kind_error(kind: pi_result::ErrorKind) -> pi_result::Error {
    pi_result::error_stack::Report::new(kind)
}