pi_append_log 0.4.0

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

use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::sync::atomic::{AtomicI64, Ordering};

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

use crate::format::BlockDecoder;
use crate::storage::{
    AppendLog, AppendLogBuilder, AppendLogVisitor, BlockVisitContext, BuildResult, ClosedStructure,
    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(structure_key(*structure_id))
    }

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

    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)
}

fn validate_direct_child(root: &Path, path: &Path) -> pi_result::Result<()> {
    if path.file_name().is_some() && path.parent().is_some_and(|parent| 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))
}

// 文件后端的 Layout 固定使用 u64 身份；八位是最小宽度，宽 ID 不截断。
// key 不包含资源路径、owner 或归档后缀，确保恢复与轮转使用同一身份规范。
fn structure_key(structure_id: u64) -> String {
    format!("{structure_id:08}")
}

/// 已封闭文件的归档句柄。
///
/// 句柄由产生它的 owner 归档；同一 owner 的 `Arc<FileAppendLog<_>>` 克隆仍属该 owner。
/// 稳定 key 是至少八位的十进制结构 ID，不含 root、owner 或归档后缀；自定义 Layout
/// 的资源命名不影响此规范，宽于八位的 ID 保留完整数字。owner 重建和归档不改变 key，
/// 但不同 root 或删除重建可能重复，外部应使用 `(log_identity, key)`。key 不能构造
/// `FileClosed` 或授权归档；旧 owner 的句柄即使 key 相同仍会被新 owner 拒绝。
pub struct FileClosed {
    owner: Arc<OwnerToken>,
    structure_id: u64,
    structure_key: String,
}

impl Clone for FileClosed {
    fn clone(&self) -> Self {
        Self {
            owner: Arc::clone(&self.owner),
            structure_id: self.structure_id,
            structure_key: self.structure_key.clone(),
        }
    }
}

impl std::fmt::Debug for FileClosed {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("FileClosed")
            .field("structure_id", &self.structure_id)
            .finish()
    }
}

impl PartialEq for FileClosed {
    fn eq(&self, other: &Self) -> bool {
        Arc::ptr_eq(&self.owner, &other.owner) && self.structure_id == other.structure_id
    }
}

impl Eq for FileClosed {}

impl ClosedStructure for FileClosed {
    fn structure_key(&self) -> &str {
        &self.structure_key
    }
}

struct OwnerToken;

/// 文件追加日志的初始化 Builder。
///
/// root 必须是绝对路径，并在 build 到 storage 的整个生命周期保持稳定。Builder 不检测
/// 或拒绝重复 build；调用方必须排除同一物理 root 的其它 owner、其它进程、旁路 I/O 以及
/// symlink、junction、大小写或其它 locator 别名的并发访问。
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(),
        }
    }
}

/// 文件追加日志的运行期实现。
///
/// 一个物理 root 在一个 owner 生命周期内只能对应一个通过 `build` 获得的实例；调用方
/// 负责遵守该合同。多任务应共享同一个 `Arc<FileAppendLog<L>>`，而不是重复 build。单 owner
/// 内 append、append_stream、rotate 和 archive 使用同一个 operation 串行化。
/// `LocalFileNamespace` 仅表示文件资源范围和 I/O 语境，不承担跨实例协调。
pub struct FileAppendLog<L> {
    root: PathBuf,
    layout: L,
    active_id: AtomicI64,
    owner: Arc<OwnerToken>,
    operation: Arc<async_lock::Mutex<()>>,
    namespace: LocalFileNamespace,
}

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

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<'a> {
    active_id: &'a AtomicI64,
    armed: bool,
}

impl<'a> AppendCommitGuard<'a> {
    fn new(active_id: &'a AtomicI64) -> Self {
        Self {
            active_id,
            armed: true,
        }
    }

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

impl Drop for AppendCommitGuard<'_> {
    fn drop(&mut self) {
        if self.armed {
            self.active_id.store(-1, Ordering::Release);
        }
    }
}

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 _operation = Arc::new(async_lock::Mutex::new(())).lock_arc().await;
        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 = discovered_active_id;
            let active_atomic = i64::try_from(active_id)
                .ok()
                .filter(|id| *id > 0)
                .ok_or_else(|| kind_error(pi_result::ErrorKind::ResourceExhausted))?;

            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 {
                    let key = structure_key(id);
                    stopped = visit_structure(&bytes, &key, decoder, order, visitor)?;
                }
            }

            let owner = Arc::new(OwnerToken);
            let storage = FileAppendLog {
                root: self.root,
                layout: self.layout,
                active_id: AtomicI64::new(active_atomic),
                owner: Arc::clone(&owner),
                operation: Arc::new(async_lock::Mutex::new(())),
                namespace: self.namespace,
            };
            let recovered_closed = recovered_ids
                .into_iter()
                .map(|structure_id| FileClosed {
                    owner: Arc::clone(&owner),
                    structure_id,
                    structure_key: structure_key(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.operation);
            let lease = Arc::new(OperationLease {
                _guard: operation.lock_arc().await,
            });
            self.ensure_ready().await?;
            let path = self.active_path()?;
            let mut file = self.open_active_append(&path).await?;
            let mut guard = AppendCommitGuard::new(&self.active_id);
            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.operation);
            let lease = Arc::new(OperationLease {
                _guard: operation.lock_arc().await,
            });
            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(&self.active_id);
            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.operation.lock().await;
        self.ensure_ready().await?;
        let active_id = self.active_id()?;
        let active_path = self.layout.segment_name(&active_id);
        let next_id = active_id
            .checked_add(1)
            .ok_or_else(|| kind_error(pi_result::ErrorKind::InvalidState))?;
        let next_atomic = i64::try_from(next_id)
            .ok()
            .filter(|id| *id > 0)
            .ok_or_else(|| kind_error(pi_result::ErrorKind::ResourceExhausted))?;
        validate_layout_paths_for_id(&self.root, &self.layout, next_id)?;
        let length = match flush_append_barrier(&self.namespace, &active_path).await {
            Ok(length) => length,
            Err(error) => {
                self.invalidate();
                return Err(error);
            }
        };
        if length == 0 {
            return Ok(None);
        }
        let next_path = self.layout.segment_name(&next_id);
        match create_new(&self.namespace, &next_path, FileAccessMode::Append).await {
            CreateOutcome::Created(mut file) => {
                if let Err(error) = file.flush(FileFlushMode::DataAndMetadata).await {
                    self.invalidate();
                    return Err(error);
                }
                self.active_id.store(next_atomic, Ordering::Release);
                Ok(Some(FileClosed {
                    owner: Arc::clone(&self.owner),
                    structure_id: active_id,
                    structure_key: structure_key(active_id),
                }))
            }
            CreateOutcome::Failed { error, .. } => {
                self.invalidate();
                Err(error)
            }
        }
    }

    async fn archive(&self, closed: Self::Closed) -> pi_result::Result<()> {
        let _operation = self.operation.lock().await;
        self.ensure_ready().await?;
        if !Arc::ptr_eq(&closed.owner, &self.owner) || closed.structure_id == self.active_id()? {
            return Err(kind_error(pi_result::ErrorKind::InvalidInput));
        }
        let segment_path = self.layout.segment_name(&closed.structure_id);
        let archive_path = self.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 => Ok(()),
            ArchiveState::SegmentOnly => {
                match rename(&self.namespace, &segment_path, &archive_path).await {
                    RenameOutcome::Renamed => Ok(()),
                    RenameOutcome::Failed { error, evidence } => match evidence {
                        RenameCommitEvidence::RenamedByOperation => Ok(()),
                        _ => 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?;
    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],
    structure_key: &str,
    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 {
                structure_key,
                is_first_in_structure: is_first,
                is_last_in_structure: is_last,
            },
        )? {
            return Ok(true);
        }
    }
    Ok(false)
}

impl<L> FileAppendLog<L> {
    fn active_id(&self) -> pi_result::Result<u64> {
        let value = self.active_id.load(Ordering::Acquire);
        if value <= 0 {
            return Err(kind_error(pi_result::ErrorKind::InvalidState));
        }
        u64::try_from(value).map_err(|_| kind_error(pi_result::ErrorKind::InvalidState))
    }

    fn invalidate(&self) {
        self.active_id.store(-1, Ordering::Release);
    }

    async fn ensure_ready(&self) -> pi_result::Result<()>
    where
        L: Layout<StructureId = u64, Name = PathBuf>,
    {
        let active_id = self.active_id()?;
        let path = self.layout.segment_name(&active_id);
        match self.namespace.try_exists(&path).await {
            Ok(true) => Ok(()),
            Ok(false) => {
                self.invalidate();
                Err(kind_error(pi_result::ErrorKind::InvalidState))
            }
            Err(error) => {
                self.invalidate();
                Err(error)
            }
        }
    }

    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.invalidate();
                Err(error)
            }
        }
    }

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

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