vfs/engine/engines/

chunked.rs

use crate::engine::block::BlockStore;
use crate::engine::error::{VfsError, VfsResult};
use crate::engine::metadata::MetadataStore;
use crate::engine::types::{
    decode_unwritten_extents, encode_unwritten_extents, normalize_path, set_xattr_value,
    unwritten_after_allocate, unwritten_after_collapse, unwritten_after_insert,
    unwritten_after_truncate, unwritten_after_write, unwritten_after_zero, unwritten_byte_ranges,
    validate_xattr_name, BlockKey, ChunkEdit, ChunkRange, CreateInodeAttrs, Dentry, InodeMeta,
    InodePatch, InodeType, SnapshotId, Storage, Timespec, VirtualStat, DEFAULT_CHUNK_SIZE,
    DEFAULT_INLINE_THRESHOLD, INTERNAL_XATTR_PREFIX,
};
use crate::engine::vfs::{Snapshottable, VirtualFileSystem};
use async_trait::async_trait;
use std::collections::BTreeMap;

#[derive(Debug, Clone)]
pub struct ChunkedFsOptions {
    pub inline_threshold: usize,
    pub chunk_size: u32,
    pub uid: u32,
    pub gid: u32,
    pub file_mode: u32,
    pub dir_mode: u32,
}

impl Default for ChunkedFsOptions {
    fn default() -> Self {
        Self {
            inline_threshold: DEFAULT_INLINE_THRESHOLD,
            chunk_size: DEFAULT_CHUNK_SIZE,
            uid: 0,
            gid: 0,
            file_mode: 0o644,
            dir_mode: 0o755,
        }
    }
}

#[derive(Debug, Clone)]
pub struct ChunkedFs<M, B> {
    metadata: M,
    blocks: B,
    options: ChunkedFsOptions,
    adaptive_chunk_size: bool,
}

const MAX_ADAPTIVE_CHUNK_SIZE: u32 = 1024 * 1024;

impl<M, B> ChunkedFs<M, B> {
    pub fn new(metadata: M, blocks: B) -> Self {
        Self::with_options(metadata, blocks, ChunkedFsOptions::default())
    }

    pub fn with_options(metadata: M, blocks: B, options: ChunkedFsOptions) -> Self {
        Self {
            metadata,
            blocks,
            options,
            adaptive_chunk_size: false,
        }
    }

    pub fn with_adaptive_chunk_size(metadata: M, blocks: B, options: ChunkedFsOptions) -> Self {
        Self {
            metadata,
            blocks,
            options,
            adaptive_chunk_size: true,
        }
    }

    pub fn metadata(&self) -> &M {
        &self.metadata
    }

    pub fn blocks(&self) -> &B {
        &self.blocks
    }
}

impl<M: MetadataStore, B: BlockStore> ChunkedFs<M, B> {
    fn initial_chunk_size(&self, write_len: usize) -> u32 {
        if !self.adaptive_chunk_size {
            return self.options.chunk_size;
        }
        let write_len = u32::try_from(write_len).unwrap_or(u32::MAX);
        write_len
            .max(self.options.chunk_size)
            .min(MAX_ADAPTIVE_CHUNK_SIZE.max(self.options.chunk_size))
    }

    async fn write_existing_or_create(&self, path: &str, content: &[u8]) -> VfsResult<()> {
        let (parent, name) = self.metadata.resolve_parent(path).await?;
        let existing = self.metadata.lstat(path).await.ok();
        let (ino, mut xattrs) = match existing {
            Some(meta) => {
                if meta.kind == InodeType::Directory {
                    return Err(VfsError::eisdir(path));
                }
                (meta.ino, meta.xattrs)
            }
            None => {
                let chunk_size = self.initial_chunk_size(content.len());
                let storage = if content.len() <= self.options.inline_threshold {
                    Storage::Inline(content.to_vec())
                } else {
                    Storage::Chunked { chunk_size }
                };
                let meta = self
                    .metadata
                    .create(
                        parent.ino,
                        &name,
                        CreateInodeAttrs::file(
                            self.options.file_mode,
                            self.options.uid,
                            self.options.gid,
                            storage,
                        ),
                    )
                    .await?;
                if content.len() <= self.options.inline_threshold {
                    return Ok(());
                }
                (meta.ino, meta.xattrs)
            }
        };
        encode_unwritten_extents(&mut xattrs, &[]);

        if content.len() <= self.options.inline_threshold {
            let freed = self
                .metadata
                .set_attr(
                    ino,
                    InodePatch {
                        storage: Some(Storage::Inline(content.to_vec())),
                        size: Some(content.len() as u64),
                        xattrs: Some(xattrs),
                        ..InodePatch::default()
                    },
                )
                .await?;
            self.blocks.delete_many(&freed).await?;
            return Ok(());
        }

        let chunk_size = self.initial_chunk_size(content.len());
        let mut edits = Vec::new();
        for (index, chunk) in content.chunks(chunk_size as usize).enumerate() {
            let key = BlockKey::from_content(chunk);
            if !self.blocks.exists(&key).await? {
                self.blocks.put(&key, chunk).await?;
            }
            edits.push(ChunkEdit {
                index: index as u64,
                key,
                len: chunk.len() as u32,
            });
        }
        self.metadata
            .set_attr(
                ino,
                InodePatch {
                    storage: Some(Storage::Chunked { chunk_size }),
                    size: Some(content.len() as u64),
                    xattrs: Some(xattrs),
                    ..InodePatch::default()
                },
            )
            .await?;
        let freed = self
            .metadata
            .commit_write(
                ino,
                edits,
                content.len() as u64,
                dense_allocation(content.len() as u64),
            )
            .await?;
        self.blocks.delete_many(&freed).await?;
        Ok(())
    }

    fn file_chunk_size(&self, storage: &Storage) -> u32 {
        match storage {
            Storage::Chunked { chunk_size } => *chunk_size,
            Storage::Inline(_) | Storage::None => self.options.chunk_size,
        }
    }

    fn ensure_file<'a>(&self, path: &str, meta: &'a InodeMeta) -> VfsResult<&'a InodeMeta> {
        match meta.kind {
            InodeType::File => Ok(meta),
            InodeType::Directory => Err(VfsError::eisdir(path)),
            InodeType::Symlink => Err(VfsError::eopnotsupp("resolved symlink without target file")),
            InodeType::CharacterDevice | InodeType::BlockDevice | InodeType::Fifo => {
                Err(VfsError::new(
                    "ENXIO",
                    format!("device I/O requires kernel dispatch: {path}"),
                ))
            }
        }
    }

    async fn read_file_range(
        &self,
        meta: &InodeMeta,
        offset: u64,
        length: usize,
    ) -> VfsResult<Vec<u8>> {
        if length == 0 || offset >= meta.size {
            return Ok(Vec::new());
        }
        let available = meta.size.saturating_sub(offset).min(length as u64);
        let output_len = usize::try_from(available)
            .map_err(|_| VfsError::einval(format!("range length is too large: {available}")))?;

        match &meta.storage {
            Storage::Inline(data) => {
                let start = usize::try_from(offset).map_err(|_| {
                    VfsError::einval(format!("range offset is too large: {offset}"))
                })?;
                if start >= data.len() {
                    return Ok(vec![0; output_len]);
                }
                let end = start.saturating_add(output_len).min(data.len());
                let mut output = vec![0; output_len];
                output[..end - start].copy_from_slice(&data[start..end]);
                Ok(output)
            }
            Storage::None => Ok(vec![0; output_len]),
            Storage::Chunked { chunk_size } => {
                let chunk_size = u64::from(*chunk_size);
                let end_offset = offset
                    .checked_add(available)
                    .ok_or_else(|| VfsError::einval("range end overflows"))?;
                let start_index = offset / chunk_size;
                let end_index = end_offset.div_ceil(chunk_size);
                let chunks = self
                    .metadata
                    .get_chunks(
                        meta.ino,
                        ChunkRange {
                            start: start_index,
                            end: Some(end_index),
                        },
                    )
                    .await?;
                let mut output = vec![0; output_len];
                for chunk in chunks {
                    let chunk_start = chunk.index.saturating_mul(chunk_size);
                    let block = self.blocks.get(&chunk.key).await?;
                    let copy_start = offset.max(chunk_start);
                    let copy_end = end_offset.min(chunk_start.saturating_add(block.len() as u64));
                    if copy_start >= copy_end {
                        continue;
                    }
                    let output_start = usize::try_from(copy_start - offset)
                        .map_err(|_| VfsError::einval("range output offset is too large"))?;
                    let block_start = usize::try_from(copy_start - chunk_start)
                        .map_err(|_| VfsError::einval("range block offset is too large"))?;
                    let len = usize::try_from(copy_end - copy_start)
                        .map_err(|_| VfsError::einval("range copy length is too large"))?;
                    output[output_start..output_start + len]
                        .copy_from_slice(&block[block_start..block_start + len]);
                }
                Ok(output)
            }
        }
    }

    async fn put_chunk_edit(&self, index: u64, data: Vec<u8>) -> VfsResult<ChunkEdit> {
        let len = u32::try_from(data.len())
            .map_err(|_| VfsError::einval(format!("chunk is too large: {}", data.len())))?;
        let key = BlockKey::from_content(&data);
        if !self.blocks.exists(&key).await? {
            self.blocks.put(&key, &data).await?;
        }
        Ok(ChunkEdit { index, key, len })
    }

    async fn set_unwritten_extents(
        &self,
        meta: &InodeMeta,
        extents: &[(u64, u64)],
    ) -> VfsResult<()> {
        let mut xattrs = meta.xattrs.clone();
        encode_unwritten_extents(&mut xattrs, extents);
        let freed = self
            .metadata
            .set_attr(
                meta.ino,
                InodePatch {
                    xattrs: Some(xattrs),
                    ..InodePatch::default()
                },
            )
            .await?;
        self.blocks.delete_many(&freed).await
    }

    async fn write_chunked_range(
        &self,
        meta: &InodeMeta,
        content: &[u8],
        offset: u64,
    ) -> VfsResult<u64> {
        if content.is_empty() {
            return Ok(meta.size);
        }
        let content_len = u64::try_from(content.len()).map_err(|_| {
            VfsError::einval(format!("pwrite content is too large: {}", content.len()))
        })?;
        let end_offset = offset
            .checked_add(content_len)
            .ok_or_else(|| VfsError::einval("pwrite end offset overflows"))?;
        let new_size = meta.size.max(end_offset);

        if !matches!(meta.storage, Storage::Chunked { .. })
            && usize::try_from(new_size)
                .ok()
                .is_some_and(|len| len <= self.options.inline_threshold)
        {
            let old_len = usize::try_from(meta.size)
                .map_err(|_| VfsError::einval(format!("file is too large: {}", meta.size)))?;
            let mut data = self.read_file_range(meta, 0, old_len).await?;
            let start = usize::try_from(offset)
                .map_err(|_| VfsError::einval(format!("pwrite offset is too large: {offset}")))?;
            let end = start.saturating_add(content.len());
            if start > data.len() {
                data.resize(start, 0);
            }
            if end > data.len() {
                data.resize(end, 0);
            }
            data[start..end].copy_from_slice(content);
            let mut xattrs = meta.xattrs.clone();
            encode_unwritten_extents(
                &mut xattrs,
                &unwritten_after_write(
                    &decode_unwritten_extents(&meta.xattrs)?,
                    offset,
                    content_len,
                ),
            );
            let freed = self
                .metadata
                .set_attr(
                    meta.ino,
                    InodePatch {
                        storage: Some(Storage::Inline(data)),
                        size: Some(new_size),
                        allocated_extents: Some(allocation_after_write(
                            &meta.allocated_extents,
                            offset,
                            content_len,
                        )),
                        xattrs: Some(xattrs),
                        ..InodePatch::default()
                    },
                )
                .await?;
            self.blocks.delete_many(&freed).await?;
            return Ok(new_size);
        }

        let chunk_size = u64::from(match &meta.storage {
            Storage::Chunked { chunk_size } => *chunk_size,
            Storage::Inline(_) | Storage::None => self.initial_chunk_size(content.len()),
        });
        let start_index = offset / chunk_size;
        let end_index = end_offset.div_ceil(chunk_size);
        let existing_chunks = if matches!(meta.storage, Storage::Chunked { .. }) {
            self.metadata
                .get_chunks(
                    meta.ino,
                    ChunkRange {
                        start: start_index,
                        end: Some(end_index),
                    },
                )
                .await?
                .into_iter()
                .map(|chunk| (chunk.index, chunk.key))
                .collect::<BTreeMap<_, _>>()
        } else {
            BTreeMap::new()
        };

        let mut edits = Vec::new();
        if let Storage::Inline(data) = &meta.storage {
            let inline_chunks = u64::try_from(data.len())
                .map_err(|_| VfsError::einval("inline data is too large"))?
                .div_ceil(chunk_size);
            for index in 0..inline_chunks.min(start_index) {
                let chunk_start = index.saturating_mul(chunk_size);
                let chunk_len = chunk_size.min(new_size.saturating_sub(chunk_start));
                let mut chunk_data = vec![
                    0;
                    usize::try_from(chunk_len).map_err(|_| {
                        VfsError::einval("inline prefix chunk is too large")
                    })?
                ];
                let copy_start = usize::try_from(chunk_start)
                    .map_err(|_| VfsError::einval("inline prefix offset is too large"))?;
                let copy_end = data.len().min(copy_start.saturating_add(chunk_data.len()));
                chunk_data[..copy_end - copy_start].copy_from_slice(&data[copy_start..copy_end]);
                edits.push(self.put_chunk_edit(index, chunk_data).await?);
            }
        }
        for index in start_index..end_index {
            let chunk_start = index.saturating_mul(chunk_size);
            let chunk_len = chunk_size.min(new_size.saturating_sub(chunk_start));
            let mut chunk_data = vec![
                0;
                usize::try_from(chunk_len).map_err(|_| {
                    VfsError::einval("chunk length is too large")
                })?
            ];

            match &meta.storage {
                Storage::Inline(data) => {
                    let copy_start = chunk_start.min(data.len() as u64);
                    let copy_end = chunk_start.saturating_add(chunk_len).min(data.len() as u64);
                    if copy_start < copy_end {
                        let dst = usize::try_from(copy_start - chunk_start)
                            .map_err(|_| VfsError::einval("inline chunk offset is too large"))?;
                        let src = usize::try_from(copy_start)
                            .map_err(|_| VfsError::einval("inline source offset is too large"))?;
                        let len = usize::try_from(copy_end - copy_start)
                            .map_err(|_| VfsError::einval("inline copy length is too large"))?;
                        chunk_data[dst..dst + len].copy_from_slice(&data[src..src + len]);
                    }
                }
                Storage::Chunked { .. } => {
                    if let Some(key) = existing_chunks.get(&index) {
                        let old = self.blocks.get(key).await?;
                        let len = old.len().min(chunk_data.len());
                        chunk_data[..len].copy_from_slice(&old[..len]);
                    }
                }
                Storage::None => {}
            }

            let write_start = offset.max(chunk_start);
            let write_end = end_offset.min(chunk_start.saturating_add(chunk_len));
            if write_start < write_end {
                let dst = usize::try_from(write_start - chunk_start)
                    .map_err(|_| VfsError::einval("chunk write offset is too large"))?;
                let src = usize::try_from(write_start - offset)
                    .map_err(|_| VfsError::einval("content write offset is too large"))?;
                let len = usize::try_from(write_end - write_start)
                    .map_err(|_| VfsError::einval("chunk write length is too large"))?;
                chunk_data[dst..dst + len].copy_from_slice(&content[src..src + len]);
            }

            edits.push(self.put_chunk_edit(index, chunk_data).await?);
        }

        if !matches!(meta.storage, Storage::Chunked { .. }) {
            self.metadata
                .set_attr(
                    meta.ino,
                    InodePatch {
                        storage: Some(Storage::Chunked {
                            chunk_size: u32::try_from(chunk_size)
                                .map_err(|_| VfsError::einval("chunk size is too large"))?,
                        }),
                        size: Some(new_size),
                        ..InodePatch::default()
                    },
                )
                .await?;
        }
        let freed = self
            .metadata
            .commit_write(
                meta.ino,
                edits,
                new_size,
                allocation_after_write(&meta.allocated_extents, offset, content_len),
            )
            .await?;
        self.blocks.delete_many(&freed).await?;
        self.set_unwritten_extents(
            meta,
            &unwritten_after_write(
                &decode_unwritten_extents(&meta.xattrs)?,
                offset,
                content_len,
            ),
        )
        .await?;
        Ok(new_size)
    }
}

#[async_trait]
impl<M: MetadataStore, B: BlockStore> VirtualFileSystem for ChunkedFs<M, B> {
    async fn read_file(&self, path: &str) -> VfsResult<Vec<u8>> {
        let meta = self.metadata.resolve(path).await?;
        self.ensure_file(path, &meta)?;
        let len = usize::try_from(meta.size)
            .map_err(|_| VfsError::einval(format!("file is too large: {}", meta.size)))?;
        self.read_file_range(&meta, 0, len).await
    }

    async fn read_dir(&self, path: &str) -> VfsResult<Vec<String>> {
        let meta = self.metadata.resolve(path).await?;
        Ok(self
            .metadata
            .list_dir(meta.ino)
            .await?
            .into_iter()
            .map(|entry| entry.name)
            .collect())
    }

    async fn read_dir_with_types(&self, path: &str) -> VfsResult<Vec<Dentry>> {
        let meta = self.metadata.resolve(path).await?;
        Ok(self
            .metadata
            .list_dir(meta.ino)
            .await?
            .into_iter()
            .map(|entry| Dentry {
                name: entry.name,
                ino: entry.meta.ino,
                kind: entry.meta.kind,
            })
            .collect())
    }

    async fn write_file(&self, path: &str, content: &[u8]) -> VfsResult<()> {
        self.write_existing_or_create(path, content).await
    }

    async fn create_dir(&self, path: &str) -> VfsResult<()> {
        let (parent, name) = self.metadata.resolve_parent(path).await?;
        self.metadata
            .create(
                parent.ino,
                &name,
                CreateInodeAttrs::directory(
                    self.options.dir_mode,
                    self.options.uid,
                    self.options.gid,
                ),
            )
            .await?;
        Ok(())
    }

    async fn mkdir(&self, path: &str, recursive: bool) -> VfsResult<()> {
        if !recursive {
            return self.create_dir(path).await;
        }
        let normalized = normalize_path(path)?;
        let mut current = String::new();
        for part in normalized
            .trim_start_matches('/')
            .split('/')
            .filter(|p| !p.is_empty())
        {
            current.push('/');
            current.push_str(part);
            if !self.exists(&current).await {
                self.create_dir(&current).await?;
            }
        }
        Ok(())
    }

    async fn mknod(&self, path: &str, mode: u32, rdev: u64) -> VfsResult<()> {
        let (parent, name) = self.metadata.resolve_parent(path).await?;
        self.metadata
            .create(
                parent.ino,
                &name,
                CreateInodeAttrs::special_node(mode, self.options.uid, self.options.gid, rdev)?,
            )
            .await?;
        Ok(())
    }

    async fn exists(&self, path: &str) -> bool {
        self.metadata.resolve(path).await.is_ok()
    }

    async fn stat(&self, path: &str) -> VfsResult<VirtualStat> {
        Ok(self.metadata.resolve(path).await?.to_stat())
    }

    async fn lstat(&self, path: &str) -> VfsResult<VirtualStat> {
        Ok(self.metadata.lstat(path).await?.to_stat())
    }

    async fn remove_file(&self, path: &str) -> VfsResult<()> {
        let meta = self.metadata.lstat(path).await?;
        if meta.kind == InodeType::Directory {
            return Err(VfsError::eisdir(path));
        }
        let (parent, name) = self.metadata.resolve_parent(path).await?;
        let freed = self.metadata.remove(parent.ino, &name).await?;
        self.blocks.delete_many(&freed).await
    }

    async fn remove_dir(&self, path: &str) -> VfsResult<()> {
        let meta = self.metadata.lstat(path).await?;
        if meta.kind != InodeType::Directory {
            return Err(VfsError::enotdir(path));
        }
        let (parent, name) = self.metadata.resolve_parent(path).await?;
        let freed = self.metadata.remove(parent.ino, &name).await?;
        self.blocks.delete_many(&freed).await
    }

    async fn rename(&self, old_path: &str, new_path: &str) -> VfsResult<()> {
        let (src_parent, src) = self.metadata.resolve_parent(old_path).await?;
        let (dst_parent, dst) = self.metadata.resolve_parent(new_path).await?;
        let freed = self
            .metadata
            .rename(src_parent.ino, &src, dst_parent.ino, &dst)
            .await?;
        self.blocks.delete_many(&freed).await
    }

    async fn realpath(&self, path: &str) -> VfsResult<String> {
        self.metadata.resolve(path).await?;
        normalize_path(path)
    }

    async fn symlink(&self, target: &str, link_path: &str) -> VfsResult<()> {
        let (parent, name) = self.metadata.resolve_parent(link_path).await?;
        self.metadata
            .create(
                parent.ino,
                &name,
                CreateInodeAttrs::symlink(target.to_string(), self.options.uid, self.options.gid),
            )
            .await?;
        Ok(())
    }

    async fn readlink(&self, path: &str) -> VfsResult<String> {
        let meta = self.metadata.lstat(path).await?;
        if meta.kind != InodeType::Symlink {
            return Err(VfsError::einval(format!("not a symlink: {path}")));
        }
        Ok(meta.symlink_target.unwrap_or_default())
    }

    async fn link(&self, old_path: &str, new_path: &str) -> VfsResult<()> {
        let target = self.metadata.resolve(old_path).await?;
        let (parent, name) = self.metadata.resolve_parent(new_path).await?;
        self.metadata.link(parent.ino, &name, target.ino).await
    }

    async fn chmod(&self, path: &str, mode: u32) -> VfsResult<()> {
        let meta = self.metadata.resolve(path).await?;
        self.metadata
            .set_attr(
                meta.ino,
                InodePatch {
                    mode: Some(mode),
                    ..InodePatch::default()
                },
            )
            .await?;
        Ok(())
    }

    async fn chown(&self, path: &str, uid: u32, gid: u32) -> VfsResult<()> {
        let meta = self.metadata.resolve(path).await?;
        self.metadata
            .set_attr(
                meta.ino,
                InodePatch {
                    uid: Some(uid),
                    gid: Some(gid),
                    ..InodePatch::default()
                },
            )
            .await?;
        Ok(())
    }

    async fn lchown(&self, path: &str, uid: u32, gid: u32) -> VfsResult<()> {
        let meta = self.metadata.lstat(path).await?;
        self.metadata
            .set_attr(
                meta.ino,
                InodePatch {
                    uid: Some(uid),
                    gid: Some(gid),
                    ..InodePatch::default()
                },
            )
            .await?;
        Ok(())
    }

    async fn get_xattr(&self, path: &str, name: &str, follow_symlinks: bool) -> VfsResult<Vec<u8>> {
        validate_xattr_name(name)?;
        let meta = if follow_symlinks {
            self.metadata.resolve(path).await?
        } else {
            self.metadata.lstat(path).await?
        };
        meta.xattrs.get(name).cloned().ok_or_else(|| {
            VfsError::new(
                "ENODATA",
                format!("extended attribute does not exist: {name}"),
            )
        })
    }

    async fn list_xattrs(&self, path: &str, follow_symlinks: bool) -> VfsResult<Vec<String>> {
        let meta = if follow_symlinks {
            self.metadata.resolve(path).await?
        } else {
            self.metadata.lstat(path).await?
        };
        Ok(meta
            .xattrs
            .into_keys()
            .filter(|name| !name.starts_with(INTERNAL_XATTR_PREFIX))
            .collect())
    }

    async fn set_xattr(
        &self,
        path: &str,
        name: &str,
        value: &[u8],
        flags: u32,
        follow_symlinks: bool,
    ) -> VfsResult<()> {
        let meta = if follow_symlinks {
            self.metadata.resolve(path).await?
        } else {
            self.metadata.lstat(path).await?
        };
        let mut xattrs = meta.xattrs;
        set_xattr_value(&mut xattrs, name, value, flags)?;
        self.metadata
            .set_attr(
                meta.ino,
                InodePatch {
                    xattrs: Some(xattrs),
                    ..InodePatch::default()
                },
            )
            .await?;
        Ok(())
    }

    async fn remove_xattr(&self, path: &str, name: &str, follow_symlinks: bool) -> VfsResult<()> {
        validate_xattr_name(name)?;
        let meta = if follow_symlinks {
            self.metadata.resolve(path).await?
        } else {
            self.metadata.lstat(path).await?
        };
        let mut xattrs = meta.xattrs;
        if xattrs.remove(name).is_none() {
            return Err(VfsError::new(
                "ENODATA",
                format!("extended attribute does not exist: {name}"),
            ));
        }
        self.metadata
            .set_attr(
                meta.ino,
                InodePatch {
                    xattrs: Some(xattrs),
                    ..InodePatch::default()
                },
            )
            .await?;
        Ok(())
    }

    async fn utimes(&self, path: &str, atime_ms: u64, mtime_ms: u64) -> VfsResult<()> {
        let meta = self.metadata.resolve(path).await?;
        self.metadata
            .set_attr(
                meta.ino,
                InodePatch {
                    atime: Some(ms_to_timespec(atime_ms)),
                    mtime: Some(ms_to_timespec(mtime_ms)),
                    ..InodePatch::default()
                },
            )
            .await?;
        Ok(())
    }

    async fn set_atime(&self, path: &str, atime_ms: u64) -> VfsResult<()> {
        let meta = self.metadata.resolve(path).await?;
        self.metadata
            .set_attr(
                meta.ino,
                InodePatch {
                    atime: Some(ms_to_timespec(atime_ms)),
                    ..InodePatch::default()
                },
            )
            .await?;
        Ok(())
    }

    async fn truncate(&self, path: &str, length: u64) -> VfsResult<()> {
        let meta = self.metadata.resolve(path).await?;
        self.ensure_file(path, &meta)?;
        if length == meta.size {
            return Ok(());
        }

        if usize::try_from(length)
            .ok()
            .is_some_and(|len| len <= self.options.inline_threshold)
        {
            let data = self
                .read_file_range(&meta, 0, usize::try_from(length).unwrap_or(0))
                .await?;
            let mut xattrs = meta.xattrs.clone();
            encode_unwritten_extents(
                &mut xattrs,
                &unwritten_after_truncate(&decode_unwritten_extents(&meta.xattrs)?, length),
            );
            let freed = self
                .metadata
                .set_attr(
                    meta.ino,
                    InodePatch {
                        storage: Some(Storage::Inline(data)),
                        size: Some(length),
                        allocated_extents: Some(allocation_after_truncate(
                            &meta.allocated_extents,
                            length,
                        )),
                        xattrs: Some(xattrs),
                        ..InodePatch::default()
                    },
                )
                .await?;
            self.blocks.delete_many(&freed).await?;
            return Ok(());
        }

        let chunk_size = u64::from(self.file_chunk_size(&meta.storage));
        let mut edits = Vec::new();
        if !matches!(meta.storage, Storage::Chunked { .. }) {
            let existing_len = meta.size.min(length);
            let mut offset = 0;
            while offset < existing_len {
                let len = (existing_len - offset).min(chunk_size);
                let data = self
                    .read_file_range(
                        &meta,
                        offset,
                        usize::try_from(len)
                            .map_err(|_| VfsError::einval("truncate chunk is too large"))?,
                    )
                    .await?;
                edits.push(self.put_chunk_edit(offset / chunk_size, data).await?);
                offset = offset.saturating_add(len);
            }
            self.metadata
                .set_attr(
                    meta.ino,
                    InodePatch {
                        storage: Some(Storage::Chunked {
                            chunk_size: self.options.chunk_size,
                        }),
                        size: Some(length),
                        ..InodePatch::default()
                    },
                )
                .await?;
        } else if length < meta.size && !length.is_multiple_of(chunk_size) {
            let final_index = length / chunk_size;
            let final_start = final_index.saturating_mul(chunk_size);
            let final_len = length - final_start;
            let data = self
                .read_file_range(
                    &meta,
                    final_start,
                    usize::try_from(final_len)
                        .map_err(|_| VfsError::einval("truncate final chunk is too large"))?,
                )
                .await?;
            edits.push(self.put_chunk_edit(final_index, data).await?);
        }

        let freed = self
            .metadata
            .commit_write(
                meta.ino,
                edits,
                length,
                allocation_after_truncate(&meta.allocated_extents, length),
            )
            .await?;
        self.blocks.delete_many(&freed).await?;
        self.set_unwritten_extents(
            &meta,
            &unwritten_after_truncate(&decode_unwritten_extents(&meta.xattrs)?, length),
        )
        .await
    }

    async fn allocate(&self, path: &str, offset: u64, length: u64) -> VfsResult<()> {
        let end = offset
            .checked_add(length)
            .ok_or_else(|| VfsError::einval("allocation range overflows"))?;
        let meta = self.metadata.resolve(path).await?;
        self.ensure_file(path, &meta)?;
        let mut xattrs = meta.xattrs.clone();
        encode_unwritten_extents(
            &mut xattrs,
            &unwritten_after_allocate(
                &decode_unwritten_extents(&meta.xattrs)?,
                &meta.allocated_extents,
                offset,
                length,
            ),
        );
        let freed = self
            .metadata
            .set_attr(
                meta.ino,
                InodePatch {
                    size: Some(meta.size.max(end)),
                    allocated_extents: Some(allocation_after_write(
                        &meta.allocated_extents,
                        offset,
                        length,
                    )),
                    xattrs: Some(xattrs),
                    ..InodePatch::default()
                },
            )
            .await?;
        self.blocks.delete_many(&freed).await
    }

    async fn punch_hole(&self, path: &str, offset: u64, length: u64) -> VfsResult<()> {
        const PUNCH_CHUNK_BYTES: u64 = 64 * 1024;

        let requested_end = offset
            .checked_add(length)
            .ok_or_else(|| VfsError::einval("hole-punch range overflows"))?;
        let meta = self.metadata.resolve(path).await?;
        self.ensure_file(path, &meta)?;
        let end = requested_end.min(meta.size);
        let mut cursor = offset.min(meta.size);
        while cursor < end {
            let chunk_len = usize::try_from((end - cursor).min(PUNCH_CHUNK_BYTES))
                .map_err(|_| VfsError::einval("hole-punch chunk is too large"))?;
            self.write_chunked_range(&meta, &vec![0; chunk_len], cursor)
                .await?;
            cursor += chunk_len as u64;
        }
        let freed = self
            .metadata
            .set_attr(
                meta.ino,
                InodePatch {
                    allocated_extents: Some(allocation_after_punch(
                        &meta.allocated_extents,
                        offset,
                        length,
                    )),
                    xattrs: Some({
                        let mut xattrs = meta.xattrs.clone();
                        encode_unwritten_extents(
                            &mut xattrs,
                            &unwritten_after_write(
                                &decode_unwritten_extents(&meta.xattrs)?,
                                offset,
                                length,
                            ),
                        );
                        xattrs
                    }),
                    ..InodePatch::default()
                },
            )
            .await?;
        self.blocks.delete_many(&freed).await
    }

    async fn zero_range(
        &self,
        path: &str,
        offset: u64,
        length: u64,
        keep_size: bool,
    ) -> VfsResult<()> {
        const ZERO_CHUNK_BYTES: u64 = 64 * 1024;

        let end = offset
            .checked_add(length)
            .ok_or_else(|| VfsError::einval("zero range overflows"))?;
        if length == 0 {
            return Err(VfsError::einval("zero range length must be nonzero"));
        }
        let original = self.metadata.resolve(path).await?;
        self.ensure_file(path, &original)?;
        let zero_end = if keep_size {
            end.min(original.size)
        } else {
            end
        };
        let mut cursor = offset.min(zero_end);
        while cursor < zero_end {
            let chunk_len = usize::try_from((zero_end - cursor).min(ZERO_CHUNK_BYTES))
                .map_err(|_| VfsError::einval("zero range chunk is too large"))?;
            self.pwrite(path, &vec![0; chunk_len], cursor).await?;
            cursor += chunk_len as u64;
        }
        let updated = self.metadata.resolve(path).await?;
        let freed = self
            .metadata
            .set_attr(
                updated.ino,
                InodePatch {
                    size: keep_size.then_some(original.size),
                    allocated_extents: Some(allocation_after_write(
                        &updated.allocated_extents,
                        offset,
                        length,
                    )),
                    xattrs: Some({
                        let mut xattrs = original.xattrs.clone();
                        encode_unwritten_extents(
                            &mut xattrs,
                            &unwritten_after_zero(
                                &decode_unwritten_extents(&original.xattrs)?,
                                &original.allocated_extents,
                                offset,
                                length,
                            ),
                        );
                        xattrs
                    }),
                    ..InodePatch::default()
                },
            )
            .await?;
        self.blocks.delete_many(&freed).await
    }

    async fn insert_range(&self, path: &str, offset: u64, length: u64) -> VfsResult<()> {
        validate_shift_range(offset, length)?;
        let meta = self.metadata.resolve(path).await?;
        self.ensure_file(path, &meta)?;
        if offset >= meta.size {
            return Err(VfsError::einval("insert range offset must be before EOF"));
        }
        let mut data = self
            .read_file_range(
                &meta,
                0,
                usize::try_from(meta.size)
                    .map_err(|_| VfsError::einval("insert source is too large"))?,
            )
            .await?;
        let start = usize::try_from(offset)
            .map_err(|_| VfsError::einval("insert range offset is too large"))?;
        let insert_len = usize::try_from(length)
            .map_err(|_| VfsError::einval("insert range length is too large"))?;
        data.splice(start..start, std::iter::repeat_n(0, insert_len));
        let extents = allocation_after_insert(&meta.allocated_extents, offset, length);
        let unwritten =
            unwritten_after_insert(&decode_unwritten_extents(&meta.xattrs)?, offset, length);
        self.write_existing_or_create(path, &data).await?;
        let updated = self.metadata.resolve(path).await?;
        let freed = self
            .metadata
            .set_attr(
                updated.ino,
                InodePatch {
                    allocated_extents: Some(extents),
                    xattrs: Some({
                        let mut xattrs = meta.xattrs.clone();
                        encode_unwritten_extents(&mut xattrs, &unwritten);
                        xattrs
                    }),
                    ..InodePatch::default()
                },
            )
            .await?;
        self.blocks.delete_many(&freed).await
    }

    async fn collapse_range(&self, path: &str, offset: u64, length: u64) -> VfsResult<()> {
        validate_shift_range(offset, length)?;
        let end = offset
            .checked_add(length)
            .ok_or_else(|| VfsError::einval("collapse range overflows"))?;
        let meta = self.metadata.resolve(path).await?;
        self.ensure_file(path, &meta)?;
        if end >= meta.size {
            return Err(VfsError::einval("collapse range must end before EOF"));
        }
        let mut data = self
            .read_file_range(
                &meta,
                0,
                usize::try_from(meta.size)
                    .map_err(|_| VfsError::einval("collapse source is too large"))?,
            )
            .await?;
        let start = usize::try_from(offset)
            .map_err(|_| VfsError::einval("collapse range offset is too large"))?;
        let end = usize::try_from(end)
            .map_err(|_| VfsError::einval("collapse range end is too large"))?;
        data.drain(start..end);
        let extents = allocation_after_collapse(&meta.allocated_extents, offset, length);
        let unwritten =
            unwritten_after_collapse(&decode_unwritten_extents(&meta.xattrs)?, offset, length);
        self.write_existing_or_create(path, &data).await?;
        let updated = self.metadata.resolve(path).await?;
        let freed = self
            .metadata
            .set_attr(
                updated.ino,
                InodePatch {
                    allocated_extents: Some(extents),
                    xattrs: Some({
                        let mut xattrs = meta.xattrs.clone();
                        encode_unwritten_extents(&mut xattrs, &unwritten);
                        xattrs
                    }),
                    ..InodePatch::default()
                },
            )
            .await?;
        self.blocks.delete_many(&freed).await
    }

    async fn allocated_ranges(&self, path: &str) -> VfsResult<Vec<(u64, u64)>> {
        let meta = self.metadata.resolve(path).await?;
        self.ensure_file(path, &meta)?;
        Ok(allocation_byte_ranges(&meta.allocated_extents, meta.size))
    }

    async fn unwritten_ranges(&self, path: &str) -> VfsResult<Vec<(u64, u64)>> {
        let meta = self.metadata.resolve(path).await?;
        self.ensure_file(path, &meta)?;
        unwritten_byte_ranges(&meta.xattrs, meta.size)
    }

    async fn pread(&self, path: &str, offset: u64, length: usize) -> VfsResult<Vec<u8>> {
        let meta = self.metadata.resolve(path).await?;
        self.ensure_file(path, &meta)?;
        self.read_file_range(&meta, offset, length).await
    }

    async fn pwrite(&self, path: &str, content: &[u8], offset: u64) -> VfsResult<()> {
        let meta = self.metadata.resolve(path).await?;
        self.ensure_file(path, &meta)?;
        self.write_chunked_range(&meta, content, offset).await?;
        Ok(())
    }

    async fn append(&self, path: &str, content: &[u8]) -> VfsResult<u64> {
        let meta = self.metadata.resolve(path).await?;
        self.ensure_file(path, &meta)?;
        let len = meta
            .size
            .checked_add(u64::try_from(content.len()).map_err(|_| {
                VfsError::einval(format!("append content is too large: {}", content.len()))
            })?)
            .ok_or_else(|| VfsError::einval("append size overflows"))?;
        self.write_chunked_range(&meta, content, meta.size).await?;
        Ok(len)
    }

    async fn sync(&self, _path: &str) -> VfsResult<()> {
        self.blocks.sync().await?;
        self.metadata.flush().await
    }
}

#[async_trait]
impl<M: MetadataStore, B: BlockStore> Snapshottable for ChunkedFs<M, B> {
    async fn snapshot(&self, root: u64) -> VfsResult<SnapshotId> {
        self.metadata.snapshot(root).await
    }

    async fn fork(&self, snap: SnapshotId) -> VfsResult<u64> {
        self.metadata.fork(snap).await
    }
}

fn ms_to_timespec(ms: u64) -> Timespec {
    Timespec {
        sec: (ms / 1_000) as i64,
        nsec: ((ms % 1_000) * 1_000_000) as u32,
    }
}

fn dense_allocation(size: u64) -> Vec<(u64, u64)> {
    if size == 0 {
        Vec::new()
    } else {
        vec![(0, size.div_ceil(512))]
    }
}

fn allocation_after_write(existing: &[(u64, u64)], offset: u64, length: u64) -> Vec<(u64, u64)> {
    if length == 0 {
        return existing.to_vec();
    }
    let mut pending = (offset / 512, offset.saturating_add(length).div_ceil(512));
    let mut merged = Vec::with_capacity(existing.len() + 1);
    for &(start, end) in existing {
        if end < pending.0 {
            merged.push((start, end));
        } else if pending.1 < start {
            merged.push(pending);
            pending = (start, end);
        } else {
            pending.0 = pending.0.min(start);
            pending.1 = pending.1.max(end);
        }
    }
    merged.push(pending);
    merged
}

fn allocation_after_truncate(existing: &[(u64, u64)], size: u64) -> Vec<(u64, u64)> {
    let end = size.div_ceil(512);
    existing
        .iter()
        .filter_map(|(start, extent_end)| {
            let extent_end = (*extent_end).min(end);
            (*start < extent_end).then_some((*start, extent_end))
        })
        .collect()
}

fn allocation_after_punch(existing: &[(u64, u64)], offset: u64, length: u64) -> Vec<(u64, u64)> {
    let start = offset.div_ceil(512);
    let end = offset.saturating_add(length) / 512;
    if start >= end {
        return existing.to_vec();
    }
    existing
        .iter()
        .flat_map(|&(extent_start, extent_end)| {
            [
                (extent_start, extent_end.min(start)),
                (extent_start.max(end), extent_end),
            ]
            .into_iter()
            .filter(|(part_start, part_end)| part_start < part_end)
        })
        .collect()
}

fn validate_shift_range(offset: u64, length: u64) -> VfsResult<()> {
    if length == 0 || !offset.is_multiple_of(512) || !length.is_multiple_of(512) {
        return Err(VfsError::einval(
            "insert/collapse range requires a nonzero 512-byte-aligned range",
        ));
    }
    Ok(())
}

fn allocation_after_insert(existing: &[(u64, u64)], offset: u64, length: u64) -> Vec<(u64, u64)> {
    let start = offset / 512;
    let shift = length / 512;
    normalize_extents(existing.iter().flat_map(|&(extent_start, extent_end)| {
        if extent_end <= start {
            vec![(extent_start, extent_end)]
        } else if extent_start >= start {
            vec![(extent_start + shift, extent_end + shift)]
        } else {
            vec![(extent_start, start), (start + shift, extent_end + shift)]
        }
    }))
}

fn allocation_after_collapse(existing: &[(u64, u64)], offset: u64, length: u64) -> Vec<(u64, u64)> {
    let start = offset / 512;
    let end = start + length / 512;
    normalize_extents(existing.iter().flat_map(|&(extent_start, extent_end)| {
        let mut parts = Vec::with_capacity(2);
        if extent_start < start {
            parts.push((extent_start, extent_end.min(start)));
        }
        if extent_end > end {
            parts.push((
                extent_start.max(end) - (end - start),
                extent_end - (end - start),
            ));
        }
        parts
    }))
}

fn normalize_extents(extents: impl IntoIterator<Item = (u64, u64)>) -> Vec<(u64, u64)> {
    let mut merged = Vec::<(u64, u64)>::new();
    for (start, end) in extents.into_iter().filter(|(start, end)| start < end) {
        if let Some(last) = merged.last_mut().filter(|last| start <= last.1) {
            last.1 = last.1.max(end);
        } else {
            merged.push((start, end));
        }
    }
    merged
}

fn allocation_byte_ranges(extents: &[(u64, u64)], size: u64) -> Vec<(u64, u64)> {
    extents
        .iter()
        .filter_map(|&(start, end)| {
            let start = start.saturating_mul(512).min(size);
            let end = end.saturating_mul(512).min(size);
            (start < end).then_some((start, end))
        })
        .collect()
}