object_store/

local.rs

// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License.  You may obtain a copy of the License at
//
//   http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.

//! An object store implementation for a local filesystem
use std::fs::{File, Metadata, OpenOptions, metadata, symlink_metadata};
use std::io::{ErrorKind, Read, Seek, SeekFrom, Write};
use std::ops::Range;
#[cfg(target_family = "unix")]
use std::os::unix::fs::FileExt;
#[cfg(target_family = "windows")]
use std::os::windows::fs::FileExt;
use std::sync::Arc;
use std::time::SystemTime;
use std::{collections::BTreeSet, io};
use std::{collections::VecDeque, path::PathBuf};

use async_trait::async_trait;
use bytes::Bytes;
use chrono::{DateTime, Utc};
use futures_util::{FutureExt, TryStreamExt};
use futures_util::{StreamExt, stream::BoxStream};
use parking_lot::Mutex;
use url::Url;
use walkdir::{DirEntry, WalkDir};

use crate::{
    Attributes, GetOptions, GetResult, GetResultPayload, ListResult, MultipartUpload, ObjectMeta,
    ObjectStore, PutMode, PutMultipartOptions, PutOptions, PutPayload, PutResult, Result,
    UploadPart, maybe_spawn_blocking,
    path::{Path, absolute_path_to_url},
    util::{InvalidGetRange, merge_ranges},
};
use crate::{CopyMode, CopyOptions, RenameOptions, RenameTargetMode};

/// A specialized `Error` for filesystem object store-related errors
#[derive(Debug, thiserror::Error)]
pub(crate) enum Error {
    #[error("Unable to walk dir: {}", source)]
    UnableToWalkDir { source: walkdir::Error },

    #[error("Unable to access metadata for {}: {}", path, source)]
    Metadata {
        source: Box<dyn std::error::Error + Send + Sync + 'static>,
        path: String,
    },

    #[error("Unable to copy data to file: {}", source)]
    UnableToCopyDataToFile { source: io::Error },

    #[error("Unable to rename file: {}", source)]
    UnableToRenameFile { source: io::Error },

    #[error("Unable to create dir {}: {}", path.display(), source)]
    UnableToCreateDir { source: io::Error, path: PathBuf },

    #[error("Unable to create file {}: {}", path.display(), source)]
    UnableToCreateFile { source: io::Error, path: PathBuf },

    #[error("Unable to delete file {}: {}", path.display(), source)]
    UnableToDeleteFile { source: io::Error, path: PathBuf },

    #[error("Unable to open file {}: {}", path.display(), source)]
    UnableToOpenFile { source: io::Error, path: PathBuf },

    #[error("Unable to read data from file {}: {}", path.display(), source)]
    UnableToReadBytes { source: io::Error, path: PathBuf },

    #[error("Out of range of file {}, expected: {}, actual: {}", path.display(), expected, actual)]
    OutOfRange {
        path: PathBuf,
        expected: u64,
        actual: u64,
    },

    #[error("Requested range was invalid")]
    InvalidRange { source: InvalidGetRange },

    #[error("Unable to copy file from {} to {}: {}", from.display(), to.display(), source)]
    UnableToCopyFile {
        from: PathBuf,
        to: PathBuf,
        source: io::Error,
    },

    #[error("NotFound")]
    NotFound { path: PathBuf, source: io::Error },

    #[error("Error seeking file {}: {}", path.display(), source)]
    Seek { source: io::Error, path: PathBuf },

    #[error("Unable to convert URL \"{}\" to filesystem path", url)]
    InvalidUrl { url: Url },

    #[error("AlreadyExists")]
    AlreadyExists { path: String, source: io::Error },

    #[error("Unable to canonicalize filesystem root: {}", path.display())]
    UnableToCanonicalize { path: PathBuf, source: io::Error },

    #[error("Filenames containing trailing '/#\\d+/' are not supported: {}", path)]
    InvalidPath { path: String },

    #[error("Unable to sync data to disk for {}: {}", path.display(), source)]
    UnableToSyncFile { source: io::Error, path: PathBuf },

    #[error("Upload aborted")]
    Aborted,
}

impl From<Error> for super::Error {
    fn from(source: Error) -> Self {
        match source {
            Error::NotFound { path, source } => Self::NotFound {
                path: path.to_string_lossy().to_string(),
                source: source.into(),
            },
            Error::AlreadyExists { path, source } => Self::AlreadyExists {
                path,
                source: source.into(),
            },
            _ => Self::Generic {
                store: "LocalFileSystem",
                source: Box::new(source),
            },
        }
    }
}

/// Explicitly close a file, checking for errors that would be silently ignored by Rust's `File::drop()`.
///
/// On network filesystems (e.g. NFS), `close()` can fail and indicate data loss.
fn close_file(file: File) -> std::result::Result<(), io::Error> {
    #[cfg(target_family = "unix")]
    {
        use std::os::fd::IntoRawFd;

        close_fd(file.into_raw_fd())
    }
    #[cfg(target_family = "windows")]
    {
        use std::os::windows::io::IntoRawHandle;

        close_handle(file.into_raw_handle())
    }
    #[cfg(not(any(target_family = "unix", target_family = "windows")))]
    {
        drop(file);
        Ok(())
    }
}

#[cfg(target_family = "unix")]
fn close_fd(fd: std::os::fd::RawFd) -> std::result::Result<(), io::Error> {
    nix::unistd::close(fd).map_err(|e| e.into())
}

#[cfg(target_family = "windows")]
fn close_handle(handle: std::os::windows::io::RawHandle) -> std::result::Result<(), io::Error> {
    // SAFETY: `handle` must be an owned handle, except when testing invalid handle errors.
    match unsafe { windows_sys::Win32::Foundation::CloseHandle(handle) } {
        0 => Err(io::Error::last_os_error()),
        _ => Ok(()),
    }
}

/// Local filesystem storage providing an [`ObjectStore`] interface to files on
/// local disk. Can optionally be created with a directory prefix
///
/// # Path Semantics
///
/// This implementation follows the [file URI] scheme outlined in [RFC 3986]. In
/// particular paths are delimited by `/`
///
/// [file URI]: https://en.wikipedia.org/wiki/File_URI_scheme
/// [RFC 3986]: https://www.rfc-editor.org/rfc/rfc3986
///
/// # Path Semantics
///
/// [`LocalFileSystem`] will expose the path semantics of the underlying filesystem, which may
/// have additional restrictions beyond those enforced by [`Path`].
///
/// For example:
///
/// * Windows forbids certain filenames, e.g. `COM0`,
/// * Windows forbids folders with trailing `.`
/// * Windows forbids certain ASCII characters, e.g. `<` or `|`
/// * OS X forbids filenames containing `:`
/// * Leading `-` are discouraged on Unix systems where they may be interpreted as CLI flags
/// * Filesystems may have restrictions on the maximum path or path segment length
/// * Filesystem support for non-ASCII characters is inconsistent
///
/// Additionally some filesystems, such as NTFS, are case-insensitive, whilst others like
/// FAT don't preserve case at all. Further some filesystems support non-unicode character
/// sequences, such as unpaired UTF-16 surrogates, and [`LocalFileSystem`] will error on
/// encountering such sequences.
///
/// Finally, filenames matching the regex `/.*#\d+/`, e.g. `foo.parquet#123`, are not supported
/// by [`LocalFileSystem`] as they are used to provide atomic writes. Such files will be ignored
/// for listing operations, and attempting to address such a file will error.
///
/// # Tokio Compatibility
///
/// Tokio discourages performing blocking IO on a tokio worker thread, however,
/// no major operating systems have stable async file APIs. Therefore if called from
/// a tokio context, this will use [`tokio::runtime::Handle::spawn_blocking`] to dispatch
/// IO to a blocking thread pool, much like `tokio::fs` does under-the-hood.
///
/// If not called from a tokio context, this will perform IO on the current thread with
/// no additional complexity or overheads
///
/// # Symlinks
///
/// [`LocalFileSystem`] will follow symlinks as normal, however, it is worth noting:
///
/// * Broken symlinks will be silently ignored by listing operations
/// * No effort is made to prevent breaking symlinks when deleting files
/// * Symlinks that resolve to paths outside the root **will** be followed
/// * Mutating a file through one or more symlinks will mutate the underlying file
/// * Deleting a path that resolves to a symlink will only delete the symlink
///
/// # Cross-Filesystem Copy
///
/// [`LocalFileSystem::copy_opts`] is implemented using [`std::fs::hard_link`], and therefore
/// does not support copying across filesystem boundaries.
///
#[derive(Clone, Debug)]
pub struct LocalFileSystem {
    config: Arc<Config>,
    // if you want to delete empty directories when deleting files
    automatic_cleanup: bool,
    // if true, fsync written files and their parent directories after writes
    fsync: bool,
}

#[derive(Debug)]
struct Config {
    root: Url,
}

impl std::fmt::Display for LocalFileSystem {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "LocalFileSystem({})", self.config.root)
    }
}

impl Default for LocalFileSystem {
    fn default() -> Self {
        Self::new()
    }
}

impl LocalFileSystem {
    /// Create new filesystem storage with no prefix
    pub fn new() -> Self {
        Self {
            config: Arc::new(Config {
                root: Url::parse("file:///").unwrap(),
            }),
            automatic_cleanup: false,
            fsync: false,
        }
    }

    /// Create new filesystem storage with `prefix` applied to all paths
    ///
    /// Returns an error if the path does not exist
    ///
    pub fn new_with_prefix(prefix: impl AsRef<std::path::Path>) -> Result<Self> {
        let path = std::fs::canonicalize(&prefix).map_err(|source| {
            let path = prefix.as_ref().into();
            Error::UnableToCanonicalize { source, path }
        })?;

        Ok(Self {
            config: Arc::new(Config {
                root: absolute_path_to_url(path)?,
            }),
            automatic_cleanup: false,
            fsync: false,
        })
    }

    /// Return an absolute filesystem path of the given file location
    pub fn path_to_filesystem(&self, location: &Path) -> Result<PathBuf> {
        self.config.path_to_filesystem(location)
    }

    /// Enable automatic cleanup of empty directories when deleting files
    pub fn with_automatic_cleanup(mut self, automatic_cleanup: bool) -> Self {
        self.automatic_cleanup = automatic_cleanup;
        self
    }

    /// Enable `fsync` after writes for durability
    ///
    /// When enabled, [`LocalFileSystem`] calls [`File::sync_all`] on written files and fsyncs
    /// the affected parent directories before a write operation
    /// ([`put_opts`](ObjectStore::put_opts), [`copy_opts`](ObjectStore::copy_opts),
    /// [`rename_opts`](ObjectStore::rename_opts), and multipart upload completion) returns
    /// success. This guarantees that both the file contents and the directory entries pointing
    /// to them are durable on stable storage, matching the implicit durability contract of
    /// remote object stores such as S3 or GCS.
    ///
    /// This trades write throughput for durability and is **disabled by default**.
    ///
    /// Note that directory fsync is only performed on Unix; on other platforms (e.g. Windows)
    /// it is a no-op, as directories cannot be portably opened and synced.
    pub fn with_fsync(mut self, fsync: bool) -> Self {
        self.fsync = fsync;
        self
    }
}

impl Config {
    /// Return an absolute filesystem path of the given location
    fn prefix_to_filesystem(&self, location: &Path) -> Result<PathBuf> {
        let mut url = self.root.clone();
        url.path_segments_mut()
            .expect("url path")
            // technically not necessary as Path ignores empty segments
            // but avoids creating paths with "//" which look odd in error messages.
            .pop_if_empty()
            .extend(location.parts());

        url.to_file_path()
            .map_err(|_| Error::InvalidUrl { url }.into())
    }

    /// Return an absolute filesystem path of the given file location
    fn path_to_filesystem(&self, location: &Path) -> Result<PathBuf> {
        if !is_valid_file_path(location) {
            let path = location.as_ref().into();
            let error = Error::InvalidPath { path };
            return Err(error.into());
        }

        let path = self.prefix_to_filesystem(location)?;

        #[cfg(target_os = "windows")]
        let path = {
            let path = path.to_string_lossy();

            // Assume the first char is the drive letter and the next is a colon.
            let mut out = String::new();
            let drive = &path[..2]; // The drive letter and colon (e.g., "C:")
            let filepath = &path[2..].replace(':', "%3A"); // Replace subsequent colons
            out.push_str(drive);
            out.push_str(filepath);
            PathBuf::from(out)
        };

        Ok(path)
    }

    /// Resolves the provided absolute filesystem path to a [`Path`] prefix
    fn filesystem_to_path(&self, location: &std::path::Path) -> Result<Path> {
        Ok(Path::from_absolute_path_with_base(
            location,
            Some(&self.root),
        )?)
    }
}

fn is_valid_file_path(path: &Path) -> bool {
    match path.filename() {
        Some(p) => match p.split_once('#') {
            Some((_, suffix)) if !suffix.is_empty() => {
                // Valid if contains non-digits
                !suffix.as_bytes().iter().all(|x| x.is_ascii_digit())
            }
            _ => true,
        },
        None => false,
    }
}

#[async_trait]
impl ObjectStore for LocalFileSystem {
    async fn put_opts(
        &self,
        location: &Path,
        payload: PutPayload,
        opts: PutOptions,
    ) -> Result<PutResult> {
        if matches!(opts.mode, PutMode::Update(_)) {
            return Err(crate::Error::NotImplemented {
                operation: "`put_opts` with mode `PutMode::Update`".into(),
                implementer: self.to_string(),
            });
        }

        if !opts.attributes.is_empty() {
            return Err(crate::Error::NotImplemented {
                operation: "`put_opts` with `opts.attributes` specified".into(),
                implementer: self.to_string(),
            });
        }

        let path = self.path_to_filesystem(location)?;
        let fsync = self.fsync;
        maybe_spawn_blocking(move || {
            let (mut file, staging_path) = new_staged_upload(&path, fsync)?;
            let mut e_tag = None;

            let err = match payload.iter().try_for_each(|x| file.write_all(x)) {
                Ok(_) => {
                    let metadata = file.metadata().map_err(|e| Error::Metadata {
                        source: e.into(),
                        path: path.to_string_lossy().to_string(),
                    })?;
                    e_tag = Some(get_etag(&metadata));
                    // Atomically publish the staged file. When fsync is enabled the publish
                    // helpers flush the file's contents and the destination's parent directory to
                    // disk first, so a successful return is durable; the fsync calls are bundled
                    // into the helpers so a file-system modification can never be left unsynced.
                    match opts.mode {
                        PutMode::Overwrite => {
                            finish_staged_rename(file, &staging_path, &path, fsync).err()
                        }
                        PutMode::Create => {
                            finish_staged_hard_link(file, &staging_path, &path, fsync).err()
                        }
                        PutMode::Update(_) => unreachable!(),
                    }
                }
                Err(source) => Some(Error::UnableToCopyDataToFile { source }.into()),
            };

            if let Some(err) = err {
                let _ = std::fs::remove_file(&staging_path); // Attempt to cleanup
                return Err(err);
            }

            Ok(PutResult {
                e_tag,
                version: None,
                extensions: Default::default(),
            })
        })
        .await
    }

    async fn put_multipart_opts(
        &self,
        location: &Path,
        opts: PutMultipartOptions,
    ) -> Result<Box<dyn MultipartUpload>> {
        if !opts.attributes.is_empty() {
            return Err(crate::Error::NotImplemented {
                operation: "`put_multipart_opts` with `opts.attributes` specified".into(),
                implementer: self.to_string(),
            });
        }

        let dest = self.path_to_filesystem(location)?;
        let (file, src) = new_staged_upload(&dest, self.fsync)?;
        Ok(Box::new(LocalUpload::new(src, dest, file, self.fsync)))
    }

    async fn get_opts(&self, location: &Path, options: GetOptions) -> Result<GetResult> {
        let location = location.clone();
        let path = self.path_to_filesystem(&location)?;
        maybe_spawn_blocking(move || {
            let file = open_file(&path)?;
            let metadata = open_metadata(&file, &path)?;
            let meta = convert_metadata(metadata, location);
            options.check_preconditions(&meta)?;

            let range = match options.range {
                Some(r) => r
                    .as_range(meta.size)
                    .map_err(|source| Error::InvalidRange { source })?,
                None => 0..meta.size,
            };

            Ok(GetResult {
                payload: GetResultPayload::File(file, path),
                attributes: Attributes::default(),
                range,
                meta,
                extensions: Default::default(),
            })
        })
        .await
    }

    async fn get_ranges(&self, location: &Path, ranges: &[Range<u64>]) -> Result<Vec<Bytes>> {
        let path = self.path_to_filesystem(location)?;
        let ranges = ranges.to_vec();
        maybe_spawn_blocking(move || {
            // We do not read the metadata here, but error in `read_range` if necessary
            let mut file = File::open(&path).map_err(|e| map_open_error(e, &path))?;

            // Coalesce only contiguous/overlapping ranges (gap 0), so we never
            // read bytes that weren't requested while still collapsing adjacent
            // ranges into fewer reads.
            let fetch_ranges = merge_ranges(&ranges, 0);

            // Fast path: nothing coalesced, read each range directly in order.
            if fetch_ranges.len() == ranges.len() {
                return ranges
                    .iter()
                    .map(|r| read_range(&mut file, &path, r.clone()))
                    .collect();
            }

            let fetched = fetch_ranges
                .iter()
                .map(|r| read_range(&mut file, &path, r.clone()))
                .collect::<Result<Vec<_>>>()?;

            Ok(ranges
                .iter()
                .map(|range| {
                    let idx = fetch_ranges.partition_point(|v| v.start <= range.start) - 1;
                    let fetch_range = &fetch_ranges[idx];
                    let fetch_bytes = &fetched[idx];

                    let start = (range.start - fetch_range.start) as usize;
                    let end = (range.end - fetch_range.start) as usize;
                    fetch_bytes.slice(start..end.min(fetch_bytes.len()))
                })
                .collect())
        })
        .await
    }

    fn delete_stream(
        &self,
        locations: BoxStream<'static, Result<Path>>,
    ) -> BoxStream<'static, Result<Path>> {
        let config = Arc::clone(&self.config);
        let automatic_cleanup = self.automatic_cleanup;
        locations
            .map(move |location| {
                let config = Arc::clone(&config);
                maybe_spawn_blocking(move || {
                    let location = location?;
                    // `with_fsync` does not apply to standalone deletes; only create-mode rename
                    // fsyncs its internal source removal as part of the durable copy-and-delete.
                    Self::delete_location(config, automatic_cleanup, &location, false)?;
                    Ok(location)
                })
            })
            .buffered(10)
            .boxed()
    }

    fn list(&self, prefix: Option<&Path>) -> BoxStream<'static, Result<ObjectMeta>> {
        Self::list_with_maybe_offset(Arc::clone(&self.config), prefix, None)
    }

    fn list_with_offset(
        &self,
        prefix: Option<&Path>,
        offset: &Path,
    ) -> BoxStream<'static, Result<ObjectMeta>> {
        Self::list_with_maybe_offset(Arc::clone(&self.config), prefix, Some(offset))
    }

    async fn list_with_delimiter(&self, prefix: Option<&Path>) -> Result<ListResult> {
        let config = Arc::clone(&self.config);

        let prefix = prefix.cloned().unwrap_or_default();
        let resolved_prefix = config.prefix_to_filesystem(&prefix)?;

        maybe_spawn_blocking(move || {
            let walkdir = WalkDir::new(&resolved_prefix)
                .min_depth(1)
                .max_depth(1)
                .follow_links(true);

            let mut common_prefixes = BTreeSet::new();
            let mut objects = Vec::new();

            for entry_res in walkdir.into_iter().map(convert_walkdir_result) {
                if let Some(entry) = entry_res? {
                    let is_directory = entry.file_type().is_dir();
                    let entry_location = config.filesystem_to_path(entry.path())?;
                    if !is_directory && !is_valid_file_path(&entry_location) {
                        continue;
                    }

                    let mut parts = match entry_location.prefix_match(&prefix) {
                        Some(parts) => parts,
                        None => continue,
                    };

                    let common_prefix = match parts.next() {
                        Some(p) => p,
                        None => continue,
                    };

                    drop(parts);

                    if is_directory {
                        common_prefixes.insert(prefix.clone().join(common_prefix));
                    } else if let Some(metadata) = convert_entry(entry, entry_location)? {
                        objects.push(metadata);
                    }
                }
            }

            Ok(ListResult {
                common_prefixes: common_prefixes.into_iter().collect(),
                objects,
                extensions: Default::default(),
            })
        })
        .await
    }

    async fn copy_opts(&self, from: &Path, to: &Path, options: CopyOptions) -> Result<()> {
        let CopyOptions {
            mode,
            extensions: _,
        } = options;

        let from = self.path_to_filesystem(from)?;
        let to = self.path_to_filesystem(to)?;
        let fsync = self.fsync;

        match mode {
            CopyMode::Overwrite => {
                let mut id = 0;
                // In order to make this atomic we:
                //
                // - hard link to a hidden temporary file
                // - atomically rename this temporary file into place
                //
                // This is necessary because hard_link returns an error if the destination already exists
                maybe_spawn_blocking(move || {
                    loop {
                        let staged = staged_upload_path(&to, &id.to_string());
                        // Stage via a temporary hard link; the source is already durable so the
                        // staging link itself needs no fsync (the publish rename below fsyncs the
                        // shared parent directory).
                        match std::fs::hard_link(&from, &staged) {
                            // `rename` bundles in the fsync of `to`'s parent directory.
                            Ok(_) => match rename(&staged, &to, fsync) {
                                Ok(_) => return Ok(()),
                                Err(source) => {
                                    let _ = std::fs::remove_file(&staged); // Attempt to clean up
                                    return Err(Error::UnableToCopyFile { from, to, source }.into());
                                }
                            },
                            Err(source) => match source.kind() {
                                ErrorKind::AlreadyExists => id += 1,
                                ErrorKind::NotFound => match from.exists() {
                                    true => create_parent_dirs(&to, source, fsync)?,
                                    false => {
                                        return Err(Error::NotFound { path: from, source }.into());
                                    }
                                },
                                _ => {
                                    return Err(Error::UnableToCopyFile { from, to, source }.into());
                                }
                            },
                        }
                    }
                })
                .await
            }
            CopyMode::Create => {
                maybe_spawn_blocking(move || {
                    loop {
                        // The source is an existing object that is already durable, so no file
                        // sync is needed; `hard_link` bundles in the fsync of `to`'s parent dir.
                        match hard_link(&from, &to, fsync) {
                            Ok(_) => return Ok(()),
                            Err(source) => match source.kind() {
                                ErrorKind::AlreadyExists => {
                                    return Err(Error::AlreadyExists {
                                        path: to.to_str().unwrap().to_string(),
                                        source,
                                    }
                                    .into());
                                }
                                ErrorKind::NotFound => match from.exists() {
                                    true => create_parent_dirs(&to, source, fsync)?,
                                    false => {
                                        return Err(Error::NotFound { path: from, source }.into());
                                    }
                                },
                                _ => {
                                    return Err(Error::UnableToCopyFile { from, to, source }.into());
                                }
                            },
                        }
                    }
                })
                .await
            }
        }
    }

    async fn rename_opts(&self, from: &Path, to: &Path, options: RenameOptions) -> Result<()> {
        let RenameOptions {
            target_mode,
            extensions,
        } = options;

        match target_mode {
            // optimized implementation
            RenameTargetMode::Overwrite => {
                let from = self.path_to_filesystem(from)?;
                let to = self.path_to_filesystem(to)?;
                let fsync = self.fsync;
                maybe_spawn_blocking(move || {
                    loop {
                        // Unlike multipart `complete`, there is no freshly written file to
                        // `sync_all` here: `from` is an existing, already-durable object and a
                        // rename only mutates directory entries. `rename` bundles in the fsync of
                        // both affected directories (destination, and source if it differs).
                        match rename(&from, &to, fsync) {
                            Ok(_) => return Ok(()),
                            Err(source) => match source.kind() {
                                ErrorKind::NotFound => match from.exists() {
                                    true => create_parent_dirs(&to, source, fsync)?,
                                    false => {
                                        return Err(Error::NotFound { path: from, source }.into());
                                    }
                                },
                                _ => {
                                    return Err(Error::UnableToCopyFile { from, to, source }.into());
                                }
                            },
                        }
                    }
                })
                .await
            }
            // fall-back to copy & delete
            RenameTargetMode::Create => {
                self.copy_opts(
                    from,
                    to,
                    CopyOptions {
                        mode: CopyMode::Create,
                        extensions,
                    },
                )
                .await?;
                let config = Arc::clone(&self.config);
                let automatic_cleanup = self.automatic_cleanup;
                let fsync = self.fsync;
                let from = from.clone();
                maybe_spawn_blocking(move || {
                    Self::delete_location(config, automatic_cleanup, &from, fsync)
                })
                .await?;
                Ok(())
            }
        }
    }
}

impl LocalFileSystem {
    fn delete_location(
        config: Arc<Config>,
        automatic_cleanup: bool,
        location: &Path,
        fsync: bool,
    ) -> Result<()> {
        let path = config.path_to_filesystem(location)?;
        if let Err(e) = std::fs::remove_file(&path) {
            Err(match e.kind() {
                ErrorKind::NotFound => Error::NotFound { path, source: e }.into(),
                _ => Error::UnableToDeleteFile { path, source: e }.into(),
            })
        } else {
            if fsync {
                fsync_parent_dir(&path).map_err(|source| Error::UnableToSyncFile {
                    source,
                    path: path.clone(),
                })?;
            }

            if !automatic_cleanup {
                return Ok(());
            }

            let root = &config.root;
            let root = root
                .to_file_path()
                .map_err(|_| Error::InvalidUrl { url: root.clone() })?;

            // here we will try to traverse up and delete an empty dir if possible until we reach the root or get an error
            let mut parent = path.parent();

            while let Some(loc) = parent {
                if loc != root && std::fs::remove_dir(loc).is_ok() {
                    parent = loc.parent();
                } else {
                    break;
                }
            }

            Ok(())
        }
    }

    fn list_with_maybe_offset(
        config: Arc<Config>,
        prefix: Option<&Path>,
        maybe_offset: Option<&Path>,
    ) -> BoxStream<'static, Result<ObjectMeta>> {
        let root_path = match prefix {
            Some(prefix) => match config.prefix_to_filesystem(prefix) {
                Ok(path) => path,
                Err(e) => return futures_util::future::ready(Err(e)).into_stream().boxed(),
            },
            None => config.root.to_file_path().unwrap(),
        };

        let walkdir = WalkDir::new(root_path)
            // Don't include the root directory itself
            .min_depth(1)
            .follow_links(true);

        let maybe_offset = maybe_offset.cloned();

        let s = walkdir.into_iter().flat_map(move |result_dir_entry| {
            // Apply offset filter before proceeding, to reduce statx file system calls
            // This matters for NFS mounts
            if let (Some(offset), Ok(entry)) = (maybe_offset.as_ref(), result_dir_entry.as_ref()) {
                let location = config.filesystem_to_path(entry.path());
                match location {
                    Ok(path) if path <= *offset => return None,
                    Err(e) => return Some(Err(e)),
                    _ => {}
                }
            }

            let entry = match convert_walkdir_result(result_dir_entry).transpose()? {
                Ok(entry) => entry,
                Err(e) => return Some(Err(e)),
            };

            if !entry.path().is_file() {
                return None;
            }

            match config.filesystem_to_path(entry.path()) {
                Ok(path) => match is_valid_file_path(&path) {
                    true => convert_entry(entry, path).transpose(),
                    false => None,
                },
                Err(e) => Some(Err(e)),
            }
        });

        // If no tokio context, return iterator directly as no
        // need to perform chunked spawn_blocking reads
        if tokio::runtime::Handle::try_current().is_err() {
            return futures_util::stream::iter(s).boxed();
        }

        // Otherwise list in batches of CHUNK_SIZE
        const CHUNK_SIZE: usize = 1024;

        let buffer = VecDeque::with_capacity(CHUNK_SIZE);
        futures_util::stream::try_unfold((s, buffer), |(mut s, mut buffer)| async move {
            if buffer.is_empty() {
                (s, buffer) = tokio::task::spawn_blocking(move || {
                    for _ in 0..CHUNK_SIZE {
                        match s.next() {
                            Some(r) => buffer.push_back(r),
                            None => break,
                        }
                    }
                    (s, buffer)
                })
                .await?;
            }

            match buffer.pop_front() {
                Some(Err(e)) => Err(e),
                Some(Ok(meta)) => Ok(Some((meta, (s, buffer)))),
                None => Ok(None),
            }
        })
        .boxed()
    }
}

/// Creates the parent directories of `path` or returns an error based on `source` if no parent
///
/// When `fsync` is true, every directory created here is fsynced, up to and including the first
/// pre-existing ancestor (whose entry list also changed), so the new directory entries are durable.
fn create_parent_dirs(path: &std::path::Path, source: io::Error, fsync: bool) -> Result<()> {
    let parent = path.parent().ok_or_else(|| {
        let path = path.to_path_buf();
        Error::UnableToCreateFile { path, source }
    })?;

    // Record the deepest already-existing ancestor *before* creating any directories, so that
    // afterwards we know exactly which directories are new and need to be fsynced.
    let first_existing = fsync.then(|| {
        let mut dir = parent;
        while !dir.exists() {
            match dir.parent() {
                Some(p) => dir = p,
                None => break,
            }
        }
        dir.to_path_buf()
    });

    std::fs::create_dir_all(parent).map_err(|source| {
        let path = parent.into();
        Error::UnableToCreateDir { source, path }
    })?;

    if let Some(first_existing) = first_existing {
        // Walk from `parent` up to `first_existing`, fsyncing each directory whose entries changed.
        let mut dir = parent;
        loop {
            fsync_dir(dir).map_err(|source| Error::UnableToSyncFile {
                source,
                path: dir.into(),
            })?;
            if dir == first_existing {
                break;
            }
            dir = match dir.parent() {
                Some(p) => p,
                None => break,
            };
        }
    }
    Ok(())
}

/// Renames `from` to `to`, then — when `fsync` is enabled — fsyncs the destination's parent
/// directory (and the source's too, if it differs) so the moved directory entries are durable.
///
/// The directory fsync is bundled in deliberately: every durable rename goes through here, so the
/// post-rename fsync can never be forgotten at an individual call site.
fn rename(from: &std::path::Path, to: &std::path::Path, fsync: bool) -> io::Result<()> {
    std::fs::rename(from, to)?;
    if fsync {
        fsync_parent_dir(to)?;
        // A cross-directory move also removes an entry from the source directory.
        if from.parent() != to.parent() {
            fsync_parent_dir(from)?;
        }
    }
    Ok(())
}

/// Hard-links `original` to `link`, then — when `fsync` is enabled — fsyncs `link`'s parent
/// directory so the new directory entry is durable.
///
/// As with [`rename`], the directory fsync is bundled in so it cannot be forgotten at a call site.
fn hard_link(original: &std::path::Path, link: &std::path::Path, fsync: bool) -> io::Result<()> {
    std::fs::hard_link(original, link)?;
    if fsync {
        fsync_parent_dir(link)?;
    }
    Ok(())
}

/// Durably publishes the freshly-written staging file `file` (located at `src`) to `dest` via a
/// rename.
///
/// When `fsync` is enabled, the file's contents are flushed before — and `dest`'s parent
/// directory after — the rename, so a successful return is durable. The file is always closed
/// before the rename (checking for close errors): required for NFS error detection and for some
/// FUSE filesystems (e.g. Blobfuse) that only commit the data on close.
fn finish_staged_rename(
    file: File,
    src: &std::path::Path,
    dest: &std::path::Path,
    fsync: bool,
) -> Result<()> {
    sync_and_close(file, src, fsync)?;
    rename(src, dest, fsync).map_err(|source| Error::UnableToRenameFile { source })?;
    Ok(())
}

/// Like [`finish_staged_rename`] but publishes via a hard link (`PutMode::Create` semantics): the
/// staging file is linked to `dest` and then removed. Returns [`Error::AlreadyExists`] if `dest`
/// already exists.
fn finish_staged_hard_link(
    file: File,
    src: &std::path::Path,
    dest: &std::path::Path,
    fsync: bool,
) -> Result<()> {
    sync_and_close(file, src, fsync)?;
    match hard_link(src, dest, fsync) {
        Ok(()) => {
            let _ = std::fs::remove_file(src); // Attempt to cleanup
            Ok(())
        }
        Err(source) => match source.kind() {
            ErrorKind::AlreadyExists => Err(Error::AlreadyExists {
                path: dest.to_str().unwrap().to_string(),
                source,
            }
            .into()),
            _ => Err(Error::UnableToRenameFile { source }.into()),
        },
    }
}

/// Flushes the freshly-written `file`'s contents to disk (when `fsync` is enabled) and then
/// closes it, checking for close errors that dropping the [`File`] would silently ignore.
fn sync_and_close(file: File, path: &std::path::Path, fsync: bool) -> Result<()> {
    if fsync {
        file.sync_all().map_err(|source| Error::UnableToSyncFile {
            source,
            path: path.into(),
        })?;
    }
    close_file(file).map_err(|source| Error::UnableToCopyDataToFile { source })?;
    Ok(())
}

/// Fsyncs the parent directory of `path` so a change to its directory entries (e.g. one just made
/// by a rename or hard link) is durable. A no-op when `path` has no parent.
fn fsync_parent_dir(path: &std::path::Path) -> io::Result<()> {
    match path.parent() {
        Some(parent) => fsync_dir(parent),
        None => Ok(()),
    }
}

/// Fsyncs `dir_path` so that changes to its directory entries are durable.
///
/// This is only meaningful on Unix; on other platforms (e.g. Windows) directories cannot be
/// portably opened as a [`File`] and synced, so this is a no-op.
fn fsync_dir(dir_path: &std::path::Path) -> io::Result<()> {
    #[cfg(target_family = "unix")]
    {
        File::open(dir_path)?.sync_all()
    }
    #[cfg(not(target_family = "unix"))]
    {
        let _ = dir_path;
        Ok(())
    }
}

/// Generates a unique file path `{base}#{suffix}`, returning the opened `File` and `path`
///
/// Creates any directories if necessary, fsyncing them when `fsync` is enabled
fn new_staged_upload(base: &std::path::Path, fsync: bool) -> Result<(File, PathBuf)> {
    let mut multipart_id = 1;
    loop {
        let suffix = multipart_id.to_string();
        let path = staged_upload_path(base, &suffix);
        let mut options = OpenOptions::new();
        match options.read(true).write(true).create_new(true).open(&path) {
            Ok(f) => return Ok((f, path)),
            Err(source) => match source.kind() {
                ErrorKind::AlreadyExists => multipart_id += 1,
                ErrorKind::NotFound => create_parent_dirs(&path, source, fsync)?,
                _ => return Err(Error::UnableToOpenFile { source, path }.into()),
            },
        }
    }
}

/// Returns the unique upload for the given path and suffix
fn staged_upload_path(dest: &std::path::Path, suffix: &str) -> PathBuf {
    let mut staging_path = dest.as_os_str().to_owned();
    staging_path.push("#");
    staging_path.push(suffix);
    staging_path.into()
}

#[derive(Debug)]
struct LocalUpload {
    /// The upload state
    state: Arc<UploadState>,
    /// The location of the temporary file
    src: Option<PathBuf>,
    /// The next offset to write into the file
    offset: u64,
    /// Whether to fsync the file and its parent directory on completion
    fsync: bool,
}

#[derive(Debug)]
struct UploadState {
    dest: PathBuf,
    file: Mutex<Option<File>>,
}

impl LocalUpload {
    pub(crate) fn new(src: PathBuf, dest: PathBuf, file: File, fsync: bool) -> Self {
        Self {
            state: Arc::new(UploadState {
                dest,
                file: Mutex::new(Some(file)),
            }),
            src: Some(src),
            offset: 0,
            fsync,
        }
    }
}

#[async_trait]
impl MultipartUpload for LocalUpload {
    fn put_part(&mut self, data: PutPayload) -> UploadPart {
        let offset = self.offset;
        self.offset += data.content_length() as u64;

        let s = Arc::clone(&self.state);
        maybe_spawn_blocking(move || {
            let mut guard = s.file.lock();
            let file = guard.as_mut().ok_or(Error::Aborted)?;
            file.seek(SeekFrom::Start(offset)).map_err(|source| {
                let path = s.dest.clone();
                Error::Seek { source, path }
            })?;

            data.iter()
                .try_for_each(|x| file.write_all(x))
                .map_err(|source| Error::UnableToCopyDataToFile { source })?;

            Ok(())
        })
        .boxed()
    }

    async fn complete(&mut self) -> Result<PutResult> {
        let src = self.src.take().ok_or(Error::Aborted)?;
        let s = Arc::clone(&self.state);
        let fsync = self.fsync;
        maybe_spawn_blocking(move || {
            // Ensure no inflight writes
            let mut guard = s.file.lock();
            let file = guard.take().ok_or(Error::Aborted)?;

            let metadata = file.metadata().map_err(|e| Error::Metadata {
                source: e.into(),
                path: src.to_string_lossy().to_string(),
            })?;

            // Durably publish the freshly-written staging file: flush its contents, close it, then
            // rename it into place and fsync the destination's parent directory (the fsync calls
            // are bundled into the helper and only run when fsync is enabled).
            finish_staged_rename(file, &src, &s.dest, fsync)?;

            Ok(PutResult {
                e_tag: Some(get_etag(&metadata)),
                version: None,
                extensions: Default::default(),
            })
        })
        .await
    }

    async fn abort(&mut self) -> Result<()> {
        let src = self.src.take().ok_or(Error::Aborted)?;
        maybe_spawn_blocking(move || {
            std::fs::remove_file(&src)
                .map_err(|source| Error::UnableToDeleteFile { source, path: src })?;
            Ok(())
        })
        .await
    }
}

impl Drop for LocalUpload {
    fn drop(&mut self) {
        if let Some(src) = self.src.take() {
            // Try to clean up intermediate file ignoring any error
            match tokio::runtime::Handle::try_current() {
                Ok(r) => drop(r.spawn_blocking(move || std::fs::remove_file(src))),
                Err(_) => drop(std::fs::remove_file(src)),
            };
        }
    }
}

pub(crate) fn chunked_stream(
    mut file: File,
    path: PathBuf,
    range: Range<u64>,
    chunk_size: usize,
) -> BoxStream<'static, Result<Bytes, super::Error>> {
    futures_util::stream::once(async move {
        let requested = range.end - range.start;

        let (file, path) = maybe_spawn_blocking(move || {
            file.seek(SeekFrom::Start(range.start as _))
                .map_err(|err| map_seek_error(err, &file, &path, range.start))?;
            Ok((file, path))
        })
        .await?;

        let stream = futures_util::stream::try_unfold(
            (file, path, requested),
            move |(mut file, path, remaining)| {
                maybe_spawn_blocking(move || {
                    if remaining == 0 {
                        return Ok(None);
                    }

                    let to_read = remaining.min(chunk_size as u64);
                    let cap = usize::try_from(to_read).map_err(|_e| Error::InvalidRange {
                        source: InvalidGetRange::TooLarge {
                            requested: to_read,
                            max: usize::MAX as u64,
                        },
                    })?;
                    let mut buffer = Vec::with_capacity(cap);
                    let read = (&mut file)
                        .take(to_read)
                        .read_to_end(&mut buffer)
                        .map_err(|e| Error::UnableToReadBytes {
                            source: e,
                            path: path.clone(),
                        })?;

                    Ok(Some((buffer.into(), (file, path, remaining - read as u64))))
                })
            },
        );
        Ok::<_, super::Error>(stream)
    })
    .try_flatten()
    .boxed()
}

pub(crate) fn read_range(
    file: &mut File,
    path: &std::path::Path,
    range: Range<u64>,
) -> Result<Bytes> {
    let requested = range.end - range.start;

    let mut buf = Vec::with_capacity(requested as usize);

    #[cfg(any(target_family = "unix", target_family = "windows"))]
    {
        buf.resize(requested as usize, 0_u8);

        let mut buf_slice = &mut buf[..];
        let mut offset = range.start;

        while !buf_slice.is_empty() {
            #[cfg(target_family = "unix")]
            let read_result = file.read_at(buf_slice, offset);

            #[cfg(target_family = "windows")]
            let read_result = file.seek_read(buf_slice, offset);

            match read_result {
                Ok(0) => break,
                Ok(n) => {
                    let tmp = buf_slice;
                    buf_slice = &mut tmp[n..];
                    offset += n as u64;
                }
                // This error is recoverable
                Err(e) if e.kind() == ErrorKind::Interrupted => {}
                Err(source) => {
                    let error = Error::UnableToReadBytes {
                        source,
                        path: path.into(),
                    };

                    return Err(error.into());
                }
            }
        }

        // If we reached EOF before filling the buffer
        if !buf_slice.is_empty() {
            let metadata = open_metadata(file, path)?;
            let file_len = metadata.len();

            // If none of the range is satisfiable we should error, e.g. if the start offset is beyond the
            // extents of the file, or if its at the end of the file and wants to read a non-empty range.
            // if range.start > file_len || (range.start == file_len && !range.is_empty()) {
            if range.start >= file_len {
                return Err(Error::InvalidRange {
                    source: InvalidGetRange::StartTooLarge {
                        requested: range.start,
                        length: file_len,
                    },
                }
                .into());
            }

            let expected = range.end.min(file_len) - range.start;

            let error = Error::OutOfRange {
                path: path.into(),
                expected,
                actual: offset - range.start,
            };

            return Err(error.into());
        }
    }
    #[cfg(all(not(windows), not(unix)))]
    {
        file.seek(SeekFrom::Start(range.start))
            .map_err(|err| map_seek_error(err, file, path, range.start))?;

        let read = file.take(requested).read_to_end(&mut buf).map_err(|err| {
            // try to read metadata to give a better error in case of directory
            if let Err(e) = open_metadata(file, path) {
                return e;
            }
            Error::UnableToReadBytes {
                source: err,
                path: path.to_path_buf(),
            }
        })? as u64;

        if read != requested {
            let metadata = open_metadata(file, path)?;
            let file_len = metadata.len();

            if range.start >= file_len {
                return Err(Error::InvalidRange {
                    source: InvalidGetRange::StartTooLarge {
                        requested: range.start,
                        length: file_len,
                    },
                }
                .into());
            }

            let expected = range.end.min(file_len) - range.start;
            if read != expected {
                return Err(Error::OutOfRange {
                    path: path.to_path_buf(),
                    expected,
                    actual: read,
                }
                .into());
            }
        }
    }

    Ok(buf.into())
}

fn open_file(path: &std::path::Path) -> Result<File, Error> {
    File::open(path).map_err(|e| map_open_error(e, path))
}

fn open_metadata(file: &File, path: &std::path::Path) -> Result<Metadata, Error> {
    let metadata = file.metadata().map_err(|e| map_open_error(e, path))?;
    if metadata.is_dir() {
        Err(Error::NotFound {
            path: PathBuf::from(path),
            source: io::Error::new(ErrorKind::NotFound, "is directory"),
        })
    } else {
        Ok(metadata)
    }
}

/// Translates errors from opening a file into a more specific [`Error`] when possible
fn map_open_error(source: io::Error, path: &std::path::Path) -> Error {
    let path = PathBuf::from(path);
    match source.kind() {
        ErrorKind::NotFound => Error::NotFound { path, source },
        _ => Error::UnableToOpenFile { path, source },
    }
}

/// Translates errors from attempting to a file into a more specific [`Error`] when possible
fn map_seek_error(source: io::Error, file: &File, path: &std::path::Path, requested: u64) -> Error {
    // if we can't seek, check if start is out of bounds to give
    // a better error. Don't read metadata before to avoid
    // an extra syscall in the common case
    let m = match open_metadata(file, path) {
        Err(e) => return e,
        Ok(m) => m,
    };
    if requested >= m.len() {
        return Error::InvalidRange {
            source: InvalidGetRange::StartTooLarge {
                requested,
                length: m.len(),
            },
        };
    }
    Error::Seek {
        source,
        path: PathBuf::from(path),
    }
}

fn convert_entry(entry: DirEntry, location: Path) -> Result<Option<ObjectMeta>> {
    match entry.metadata() {
        Ok(metadata) => Ok(Some(convert_metadata(metadata, location))),
        Err(e) => {
            if let Some(io_err) = e.io_error() {
                if io_err.kind() == ErrorKind::NotFound {
                    return Ok(None);
                }
            }
            Err(Error::Metadata {
                source: e.into(),
                path: location.to_string(),
            })?
        }
    }
}

fn last_modified(metadata: &Metadata) -> DateTime<Utc> {
    metadata
        .modified()
        .expect("Modified file time should be supported on this platform")
        .into()
}

fn get_etag(metadata: &Metadata) -> String {
    let inode = get_inode(metadata);
    let size = metadata.len();
    let mtime = metadata
        .modified()
        .ok()
        .and_then(|mtime| mtime.duration_since(SystemTime::UNIX_EPOCH).ok())
        .unwrap_or_default()
        .as_micros();

    // Use an ETag scheme based on that used by many popular HTTP servers
    // <https://httpd.apache.org/docs/2.2/mod/core.html#fileetag>
    // <https://stackoverflow.com/questions/47512043/how-etags-are-generated-and-configured>
    format!("\"{inode:x}-{mtime:x}-{size:x}\"")
}

fn convert_metadata(metadata: Metadata, location: Path) -> ObjectMeta {
    let last_modified = last_modified(&metadata);

    ObjectMeta {
        location,
        last_modified,
        size: metadata.len(),
        e_tag: Some(get_etag(&metadata)),
        version: None,
    }
}

#[cfg(unix)]
/// We include the inode when available to yield an ETag more resistant to collisions
/// and as used by popular web servers such as [Apache](https://httpd.apache.org/docs/2.2/mod/core.html#fileetag)
fn get_inode(metadata: &Metadata) -> u64 {
    std::os::unix::fs::MetadataExt::ino(metadata)
}

#[cfg(not(unix))]
/// On platforms where an inode isn't available, fallback to just relying on size and mtime
fn get_inode(_metadata: &Metadata) -> u64 {
    0
}

/// Convert walkdir results and converts not-found errors into `None`.
/// Convert broken symlinks to `None`.
fn convert_walkdir_result(
    res: std::result::Result<DirEntry, walkdir::Error>,
) -> Result<Option<DirEntry>> {
    match res {
        Ok(entry) => {
            // To check for broken symlink: call symlink_metadata() - it does not traverse symlinks);
            // if ok: check if entry is symlink; and try to read it by calling metadata().
            match symlink_metadata(entry.path()) {
                Ok(attr) => {
                    if attr.is_symlink() {
                        let target_metadata = metadata(entry.path());
                        match target_metadata {
                            Ok(_) => {
                                // symlink is valid
                                Ok(Some(entry))
                            }
                            Err(_) => {
                                // this is a broken symlink, return None
                                Ok(None)
                            }
                        }
                    } else {
                        Ok(Some(entry))
                    }
                }
                Err(_) => Ok(None),
            }
        }

        Err(walkdir_err) => match walkdir_err.io_error() {
            Some(io_err) => match io_err.kind() {
                ErrorKind::NotFound => Ok(None),
                _ => Err(Error::UnableToWalkDir {
                    source: walkdir_err,
                }
                .into()),
            },
            None => Err(Error::UnableToWalkDir {
                source: walkdir_err,
            }
            .into()),
        },
    }
}

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

    use futures_util::TryStreamExt;
    use tempfile::TempDir;

    #[cfg(target_family = "unix")]
    use std::os::unix::fs::PermissionsExt;

    #[cfg(target_family = "unix")]
    use tempfile::NamedTempFile;

    use crate::{ObjectStoreExt, integration::*};

    use super::*;

    #[tokio::test]
    #[cfg(target_family = "unix")]
    async fn file_test() {
        let root = TempDir::new().unwrap();
        let integration = LocalFileSystem::new_with_prefix(root.path()).unwrap();

        put_get_delete_list(&integration).await;
        list_with_offset_exclusivity(&integration).await;
        get_opts(&integration).await;
        list_uses_directories_correctly(&integration).await;
        list_with_delimiter(&integration).await;
        rename_and_copy(&integration).await;
        copy_if_not_exists(&integration).await;
        copy_rename_nonexistent_object(&integration).await;
        stream_get(&integration).await;
        put_opts(&integration, false).await;
    }

    #[tokio::test]
    #[cfg(target_family = "unix")]
    async fn file_test_fsync() {
        // Run the full integration suite with fsync enabled to ensure the durability code
        // paths (file sync + directory fsync on put/copy/rename/multipart, including recursive
        // directory creation) behave identically to the default.
        let root = TempDir::new().unwrap();
        let integration = LocalFileSystem::new_with_prefix(root.path())
            .unwrap()
            .with_fsync(true);

        put_get_delete_list(&integration).await;
        list_with_offset_exclusivity(&integration).await;
        get_opts(&integration).await;
        list_uses_directories_correctly(&integration).await;
        list_with_delimiter(&integration).await;
        rename_and_copy(&integration).await;
        copy_if_not_exists(&integration).await;
        copy_rename_nonexistent_object(&integration).await;
        stream_get(&integration).await;
        put_opts(&integration, false).await;
    }

    #[tokio::test]
    async fn fsync_creates_nested_dirs() {
        // Exercises the recursive directory fsync in `create_parent_dirs`: every directory
        // component is newly created, so each must be synced up to the pre-existing root.
        let root = TempDir::new().unwrap();
        let integration = LocalFileSystem::new_with_prefix(root.path())
            .unwrap()
            .with_fsync(true);

        let data = Bytes::from("arbitrary data");

        // `put` (overwrite) into a deeply nested, non-existent directory tree
        let location = Path::from("a/b/c/d/put_file");
        integration
            .put(&location, data.clone().into())
            .await
            .unwrap();
        let read = integration
            .get(&location)
            .await
            .unwrap()
            .bytes()
            .await
            .unwrap();
        assert_eq!(read, data);

        // multipart upload into another nested tree
        let location = Path::from("e/f/g/multipart_file");
        let mut upload = integration.put_multipart(&location).await.unwrap();
        upload.put_part(data.clone().into()).await.unwrap();
        upload.complete().await.unwrap();
        let read = integration
            .get(&location)
            .await
            .unwrap()
            .bytes()
            .await
            .unwrap();
        assert_eq!(read, data);
    }

    #[tokio::test]
    #[cfg(target_family = "unix")]
    async fn fsync_rename_if_not_exists_propagates_source_delete_sync_error() {
        let root = TempDir::new().unwrap();
        let integration = LocalFileSystem::new_with_prefix(root.path())
            .unwrap()
            .with_fsync(true);

        let source = Path::from("source_dir/source_file");
        let dest = Path::from("dest_dir/dest_file");
        integration.put(&source, "data".into()).await.unwrap();

        let source_dir = root.path().join("source_dir");
        let original_permissions = fs::metadata(&source_dir).unwrap().permissions();
        // Allow unlinking the file from the directory, but prevent opening the
        // directory for fsync. Without the source-parent fsync, rename succeeds;
        // with it, the sync error is propagated after the source is deleted.
        fs::set_permissions(&source_dir, fs::Permissions::from_mode(0o300)).unwrap();

        let result = integration.rename_if_not_exists(&source, &dest).await;
        fs::set_permissions(&source_dir, original_permissions).unwrap();

        match result {
            Err(crate::Error::Generic { source, .. }) => {
                assert!(source.to_string().contains("Unable to sync data to disk"));
            }
            _ => panic!("expected source parent fsync to fail"),
        }

        let read = integration.get(&dest).await.unwrap().bytes().await.unwrap();
        assert_eq!(read, Bytes::from("data"));
        assert!(matches!(
            integration.get(&source).await.unwrap_err(),
            crate::Error::NotFound { .. }
        ));
    }

    #[test]
    #[cfg(target_family = "unix")]
    fn test_non_tokio() {
        let root = TempDir::new().unwrap();
        let integration = LocalFileSystem::new_with_prefix(root.path()).unwrap();
        futures_executor::block_on(async move {
            put_get_delete_list(&integration).await;
            list_uses_directories_correctly(&integration).await;
            list_with_delimiter(&integration).await;

            // Can't use stream_get test as WriteMultipart uses a tokio JoinSet
            let p = Path::from("manual_upload");
            let mut upload = integration.put_multipart(&p).await.unwrap();
            upload.put_part("123".into()).await.unwrap();
            upload.put_part("45678".into()).await.unwrap();
            let r = upload.complete().await.unwrap();

            let get = integration.get(&p).await.unwrap();
            assert_eq!(get.meta.e_tag.as_ref().unwrap(), r.e_tag.as_ref().unwrap());
            let actual = get.bytes().await.unwrap();
            assert_eq!(actual.as_ref(), b"12345678");
        });
    }

    #[tokio::test]
    async fn creates_dir_if_not_present() {
        let root = TempDir::new().unwrap();
        let integration = LocalFileSystem::new_with_prefix(root.path()).unwrap();

        let location = Path::from("nested/file/test_file");

        let data = Bytes::from("arbitrary data");

        integration
            .put(&location, data.clone().into())
            .await
            .unwrap();

        let read_data = integration
            .get(&location)
            .await
            .unwrap()
            .bytes()
            .await
            .unwrap();
        assert_eq!(&*read_data, data);
    }

    #[tokio::test]
    async fn unknown_length() {
        let root = TempDir::new().unwrap();
        let integration = LocalFileSystem::new_with_prefix(root.path()).unwrap();

        let location = Path::from("some_file");

        let data = Bytes::from("arbitrary data");

        integration
            .put(&location, data.clone().into())
            .await
            .unwrap();

        let read_data = integration
            .get(&location)
            .await
            .unwrap()
            .bytes()
            .await
            .unwrap();
        assert_eq!(&*read_data, data);
    }

    #[tokio::test]
    async fn range_request_start_beyond_end_of_file() {
        let root = TempDir::new().unwrap();
        let integration = LocalFileSystem::new_with_prefix(root.path()).unwrap();

        let location = Path::from("some_file");

        let data = Bytes::from("arbitrary data");

        integration
            .put(&location, data.clone().into())
            .await
            .unwrap();

        integration
            .get_range(&location, 100..200)
            .await
            .expect_err("Should error with start range beyond end of file");
    }

    #[tokio::test]
    async fn range_request_beyond_end_of_file() {
        let root = TempDir::new().unwrap();
        let integration = LocalFileSystem::new_with_prefix(root.path()).unwrap();

        let location = Path::from("some_file");

        let data = Bytes::from("arbitrary data");

        integration
            .put(&location, data.clone().into())
            .await
            .unwrap();

        let read_data = integration.get_range(&location, 0..100).await.unwrap();
        assert_eq!(&*read_data, data);
    }

    #[tokio::test]
    async fn get_ranges_coalesces() {
        let root = TempDir::new().unwrap();
        let integration = LocalFileSystem::new_with_prefix(root.path()).unwrap();
        let location = Path::from("some_file");

        // Position-dependent content so a misrouted slice is detectable.
        let data: Bytes = (0..4096u32)
            .map(|i| (i % 251) as u8)
            .collect::<Vec<_>>()
            .into();
        integration
            .put(&location, data.clone().into())
            .await
            .unwrap();

        let cases: Vec<Vec<Range<u64>>> = vec![
            vec![0..16, 16..32, 32..48],                    // adjacent -> coalesced
            vec![0..16, 64..80, 1000..1016],                // gapped -> not coalesced
            vec![100..120, 0..50, 40..60, 0..50, 110..130], // out-of-order, overlapping, duplicate
            vec![],                                         // empty
        ];

        for ranges in cases {
            let got = integration.get_ranges(&location, &ranges).await.unwrap();
            assert_eq!(got.len(), ranges.len());
            for (range, bytes) in ranges.iter().zip(got) {
                assert_eq!(
                    bytes,
                    data.slice(range.start as usize..range.end as usize),
                    "mismatch for range {range:?}"
                );
            }
        }
    }

    #[tokio::test]
    #[cfg(target_family = "unix")]
    // Fails on github actions runner (which runs the tests as root)
    #[ignore]
    async fn bubble_up_io_errors() {
        use std::{fs::set_permissions, os::unix::prelude::PermissionsExt};

        let root = TempDir::new().unwrap();

        // make non-readable
        let metadata = root.path().metadata().unwrap();
        let mut permissions = metadata.permissions();
        permissions.set_mode(0o000);
        set_permissions(root.path(), permissions).unwrap();

        let store = LocalFileSystem::new_with_prefix(root.path()).unwrap();

        let mut stream = store.list(None);
        let mut any_err = false;
        while let Some(res) = stream.next().await {
            if res.is_err() {
                any_err = true;
            }
        }
        assert!(any_err);

        // `list_with_delimiter
        assert!(store.list_with_delimiter(None).await.is_err());
    }

    const NON_EXISTENT_NAME: &str = "nonexistentname";

    #[tokio::test]
    async fn get_nonexistent_location() {
        let root = TempDir::new().unwrap();
        let integration = LocalFileSystem::new_with_prefix(root.path()).unwrap();

        let location = Path::from(NON_EXISTENT_NAME);

        let err = get_nonexistent_object(&integration, Some(location))
            .await
            .unwrap_err();
        if let crate::Error::NotFound { path, source } = err {
            let source_variant = source.downcast_ref::<std::io::Error>();
            assert!(
                matches!(source_variant, Some(std::io::Error { .. }),),
                "got: {source_variant:?}"
            );
            assert!(path.ends_with(NON_EXISTENT_NAME), "{}", path);
        } else {
            panic!("unexpected error type: {err:?}");
        }
    }

    #[tokio::test]
    async fn root() {
        let integration = LocalFileSystem::new();

        let canonical = std::path::Path::new("Cargo.toml").canonicalize().unwrap();
        let url = Url::from_directory_path(&canonical).unwrap();
        let path = Path::parse(url.path()).unwrap();

        let roundtrip = integration.path_to_filesystem(&path).unwrap();

        // Needed as on Windows canonicalize returns extended length path syntax
        // C:\Users\circleci -> \\?\C:\Users\circleci
        let roundtrip = roundtrip.canonicalize().unwrap();

        assert_eq!(roundtrip, canonical);

        integration.head(&path).await.unwrap();
    }

    #[tokio::test]
    #[cfg(target_family = "windows")]
    async fn test_list_root() {
        let fs = LocalFileSystem::new();
        let r = fs.list_with_delimiter(None).await.unwrap_err().to_string();

        assert!(
            r.contains("Unable to convert URL \"file:///\" to filesystem path"),
            "{}",
            r
        );
    }

    #[tokio::test]
    #[cfg(target_os = "linux")]
    async fn test_list_root() {
        let fs = LocalFileSystem::new();
        fs.list_with_delimiter(None).await.unwrap();
    }

    #[cfg(target_family = "unix")]
    async fn check_list(integration: &LocalFileSystem, prefix: Option<&Path>, expected: &[&str]) {
        let result: Vec<_> = integration.list(prefix).try_collect().await.unwrap();

        let mut strings: Vec<_> = result.iter().map(|x| x.location.as_ref()).collect();
        strings.sort_unstable();
        assert_eq!(&strings, expected)
    }

    #[tokio::test]
    #[cfg(target_family = "unix")]
    async fn test_symlink() {
        let root = TempDir::new().unwrap();
        let integration = LocalFileSystem::new_with_prefix(root.path()).unwrap();

        let subdir = root.path().join("a");
        std::fs::create_dir(&subdir).unwrap();
        let file = subdir.join("file.parquet");
        std::fs::write(file, "test").unwrap();

        check_list(&integration, None, &["a/file.parquet"]).await;
        integration
            .head(&Path::from("a/file.parquet"))
            .await
            .unwrap();

        // Follow out of tree symlink
        let other = NamedTempFile::new().unwrap();
        std::os::unix::fs::symlink(other.path(), root.path().join("test.parquet")).unwrap();

        // Should return test.parquet even though out of tree
        check_list(&integration, None, &["a/file.parquet", "test.parquet"]).await;

        // Can fetch test.parquet
        integration.head(&Path::from("test.parquet")).await.unwrap();

        // Follow in tree symlink
        std::os::unix::fs::symlink(&subdir, root.path().join("b")).unwrap();
        check_list(
            &integration,
            None,
            &["a/file.parquet", "b/file.parquet", "test.parquet"],
        )
        .await;
        check_list(&integration, Some(&Path::from("b")), &["b/file.parquet"]).await;

        // Can fetch through symlink
        integration
            .head(&Path::from("b/file.parquet"))
            .await
            .unwrap();

        // Ignore broken symlink
        std::os::unix::fs::symlink(root.path().join("foo.parquet"), root.path().join("c")).unwrap();

        check_list(
            &integration,
            None,
            &["a/file.parquet", "b/file.parquet", "test.parquet"],
        )
        .await;

        let mut r = integration.list_with_delimiter(None).await.unwrap();
        r.common_prefixes.sort_unstable();
        assert_eq!(r.common_prefixes.len(), 2);
        assert_eq!(r.common_prefixes[0].as_ref(), "a");
        assert_eq!(r.common_prefixes[1].as_ref(), "b");
        assert_eq!(r.objects.len(), 1);
        assert_eq!(r.objects[0].location.as_ref(), "test.parquet");

        let r = integration
            .list_with_delimiter(Some(&Path::from("a")))
            .await
            .unwrap();
        assert_eq!(r.common_prefixes.len(), 0);
        assert_eq!(r.objects.len(), 1);
        assert_eq!(r.objects[0].location.as_ref(), "a/file.parquet");

        // Deleting a symlink doesn't delete the source file
        integration
            .delete(&Path::from("test.parquet"))
            .await
            .unwrap();
        assert!(other.path().exists());

        check_list(&integration, None, &["a/file.parquet", "b/file.parquet"]).await;

        // Deleting through a symlink deletes both files
        integration
            .delete(&Path::from("b/file.parquet"))
            .await
            .unwrap();

        check_list(&integration, None, &[]).await;

        // Adding a file through a symlink creates in both paths
        integration
            .put(&Path::from("b/file.parquet"), vec![0, 1, 2].into())
            .await
            .unwrap();

        check_list(&integration, None, &["a/file.parquet", "b/file.parquet"]).await;
    }

    #[tokio::test]
    async fn invalid_path() {
        let root = TempDir::new().unwrap();
        let root = root.path().join("🙀");
        std::fs::create_dir(root.clone()).unwrap();

        // Invalid paths supported above root of store
        let integration = LocalFileSystem::new_with_prefix(root.clone()).unwrap();

        let directory = Path::from("directory");
        let object = directory.clone().join("child.txt");
        let data = Bytes::from("arbitrary");
        integration.put(&object, data.clone().into()).await.unwrap();
        integration.head(&object).await.unwrap();
        let result = integration.get(&object).await.unwrap();
        assert_eq!(result.bytes().await.unwrap(), data);

        flatten_list_stream(&integration, None).await.unwrap();
        flatten_list_stream(&integration, Some(&directory))
            .await
            .unwrap();

        let result = integration
            .list_with_delimiter(Some(&directory))
            .await
            .unwrap();
        assert_eq!(result.objects.len(), 1);
        assert!(result.common_prefixes.is_empty());
        assert_eq!(result.objects[0].location, object);

        let emoji = root.join("💀");
        std::fs::write(emoji, "foo").unwrap();

        // Can list illegal file
        let mut paths = flatten_list_stream(&integration, None).await.unwrap();
        paths.sort_unstable();

        assert_eq!(
            paths,
            vec![
                Path::parse("directory/child.txt").unwrap(),
                Path::parse("💀").unwrap()
            ]
        );
    }

    #[tokio::test]
    async fn list_hides_incomplete_uploads() {
        let root = TempDir::new().unwrap();
        let integration = LocalFileSystem::new_with_prefix(root.path()).unwrap();
        let location = Path::from("some_file");

        let data = PutPayload::from("arbitrary data");
        let mut u1 = integration.put_multipart(&location).await.unwrap();
        u1.put_part(data.clone()).await.unwrap();

        let mut u2 = integration.put_multipart(&location).await.unwrap();
        u2.put_part(data).await.unwrap();

        let list = flatten_list_stream(&integration, None).await.unwrap();
        assert_eq!(list.len(), 0);

        assert_eq!(
            integration
                .list_with_delimiter(None)
                .await
                .unwrap()
                .objects
                .len(),
            0
        );
    }

    #[tokio::test]
    async fn test_path_with_offset() {
        let root = TempDir::new().unwrap();
        let integration = LocalFileSystem::new_with_prefix(root.path()).unwrap();

        let root_path = root.path();
        for i in 0..5 {
            let filename = format!("test{i}.parquet");
            let file = root_path.join(filename);
            std::fs::write(file, "test").unwrap();
        }
        let filter_str = "test";
        let filter = String::from(filter_str);
        let offset_str = filter + "1";
        let offset = Path::from(offset_str.clone());

        // Use list_with_offset to retrieve files
        let res = integration.list_with_offset(None, &offset);
        let offset_paths: Vec<_> = res.map_ok(|x| x.location).try_collect().await.unwrap();
        let mut offset_files: Vec<_> = offset_paths
            .iter()
            .map(|x| String::from(x.filename().unwrap()))
            .collect();

        // Check result with direct filesystem read
        let files = fs::read_dir(root_path).unwrap();
        let filtered_files = files
            .filter_map(Result::ok)
            .filter_map(|d| {
                d.file_name().to_str().and_then(|f| {
                    if f.contains(filter_str) {
                        Some(String::from(f))
                    } else {
                        None
                    }
                })
            })
            .collect::<Vec<_>>();

        let mut expected_offset_files: Vec<_> = filtered_files
            .iter()
            .filter(|s| **s > offset_str)
            .cloned()
            .collect();

        fn do_vecs_match<T: PartialEq>(a: &[T], b: &[T]) -> bool {
            let matching = a.iter().zip(b.iter()).filter(|&(a, b)| a == b).count();
            matching == a.len() && matching == b.len()
        }

        offset_files.sort();
        expected_offset_files.sort();

        // println!("Expected Offset Files: {:?}", expected_offset_files);
        // println!("Actual Offset Files: {:?}", offset_files);

        assert_eq!(offset_files.len(), expected_offset_files.len());
        assert!(do_vecs_match(&expected_offset_files, &offset_files));
    }

    #[tokio::test]
    async fn filesystem_filename_with_percent() {
        let temp_dir = TempDir::new().unwrap();
        let integration = LocalFileSystem::new_with_prefix(temp_dir.path()).unwrap();
        let filename = "L%3ABC.parquet";

        std::fs::write(temp_dir.path().join(filename), "foo").unwrap();

        let res: Vec<_> = integration.list(None).try_collect().await.unwrap();
        assert_eq!(res.len(), 1);
        assert_eq!(res[0].location.as_ref(), filename);

        let res = integration.list_with_delimiter(None).await.unwrap();
        assert_eq!(res.objects.len(), 1);
        assert_eq!(res.objects[0].location.as_ref(), filename);
    }

    #[tokio::test]
    async fn relative_paths() {
        LocalFileSystem::new_with_prefix(".").unwrap();
        LocalFileSystem::new_with_prefix("..").unwrap();
        LocalFileSystem::new_with_prefix("../..").unwrap();

        let integration = LocalFileSystem::new();
        let path = Path::from_filesystem_path(".").unwrap();
        integration.list_with_delimiter(Some(&path)).await.unwrap();
    }

    #[test]
    fn test_valid_path() {
        let cases = [
            ("foo#123/test.txt", true),
            ("foo#123/test#23.txt", true),
            ("foo#123/test#34", false),
            ("foo😁/test#34", false),
            ("foo/test#😁34", true),
        ];

        for (case, expected) in cases {
            let path = Path::parse(case).unwrap();
            assert_eq!(is_valid_file_path(&path), expected);
        }
    }

    #[tokio::test]
    async fn test_intermediate_files() {
        let root = TempDir::new().unwrap();
        let integration = LocalFileSystem::new_with_prefix(root.path()).unwrap();

        let a = Path::parse("foo#123/test.txt").unwrap();
        integration.put(&a, "test".into()).await.unwrap();

        let list = flatten_list_stream(&integration, None).await.unwrap();
        assert_eq!(list, vec![a.clone()]);

        std::fs::write(root.path().join("bar#123"), "test").unwrap();

        // Should ignore file
        let list = flatten_list_stream(&integration, None).await.unwrap();
        assert_eq!(list, vec![a.clone()]);

        let b = Path::parse("bar#123").unwrap();
        let err = integration.get(&b).await.unwrap_err().to_string();
        assert_eq!(
            err,
            "Generic LocalFileSystem error: Filenames containing trailing '/#\\d+/' are not supported: bar#123"
        );

        let c = Path::parse("foo#123.txt").unwrap();
        integration.put(&c, "test".into()).await.unwrap();

        let mut list = flatten_list_stream(&integration, None).await.unwrap();
        list.sort_unstable();
        assert_eq!(list, vec![c, a]);
    }

    #[tokio::test]
    #[cfg(target_os = "windows")]
    async fn filesystem_filename_with_colon() {
        let root = TempDir::new().unwrap();
        let integration = LocalFileSystem::new_with_prefix(root.path()).unwrap();
        let path = Path::parse("file%3Aname.parquet").unwrap();
        let location = Path::parse("file:name.parquet").unwrap();

        integration.put(&location, "test".into()).await.unwrap();
        let list = flatten_list_stream(&integration, None).await.unwrap();
        assert_eq!(list, vec![path.clone()]);

        let result = integration
            .get(&location)
            .await
            .unwrap()
            .bytes()
            .await
            .unwrap();
        assert_eq!(result, Bytes::from("test"));
    }

    #[tokio::test]
    async fn delete_dirs_automatically() {
        let root = TempDir::new().unwrap();
        let integration = LocalFileSystem::new_with_prefix(root.path())
            .unwrap()
            .with_automatic_cleanup(true);
        let location = Path::from("nested/file/test_file");
        let data = Bytes::from("arbitrary data");

        integration
            .put(&location, data.clone().into())
            .await
            .unwrap();

        let read_data = integration
            .get(&location)
            .await
            .unwrap()
            .bytes()
            .await
            .unwrap();

        assert_eq!(&*read_data, data);
        assert!(fs::read_dir(root.path()).unwrap().count() > 0);
        integration.delete(&location).await.unwrap();
        assert!(fs::read_dir(root.path()).unwrap().count() == 0);
    }

    #[test]
    #[cfg(target_family = "unix")]
    fn test_close_file_detects_error_unix() {
        let err = super::close_fd(-1).unwrap_err();
        assert_eq!(err.raw_os_error(), Some(nix::libc::EBADF), "got: {err:?}");
    }

    #[test]
    #[cfg(target_family = "windows")]
    fn test_close_file_detects_error_windows() {
        let err = super::close_handle(std::ptr::null_mut()).unwrap_err();
        assert_eq!(
            err.raw_os_error(),
            Some(windows_sys::Win32::Foundation::ERROR_INVALID_HANDLE as i32),
            "got: {err:?}"
        );
    }
}

#[cfg(not(target_arch = "wasm32"))]
#[cfg(test)]
mod not_wasm_tests {
    use std::time::Duration;
    use tempfile::TempDir;

    use crate::local::LocalFileSystem;
    use crate::{ObjectStoreExt, Path, PutPayload};

    #[tokio::test]
    async fn test_cleanup_intermediate_files() {
        let root = TempDir::new().unwrap();
        let integration = LocalFileSystem::new_with_prefix(root.path()).unwrap();

        let location = Path::from("some_file");
        let data = PutPayload::from_static(b"hello");
        let mut upload = integration.put_multipart(&location).await.unwrap();
        upload.put_part(data).await.unwrap();

        let file_count = std::fs::read_dir(root.path()).unwrap().count();
        assert_eq!(file_count, 1);
        drop(upload);

        for _ in 0..100 {
            tokio::time::sleep(Duration::from_millis(1)).await;
            let file_count = std::fs::read_dir(root.path()).unwrap().count();
            if file_count == 0 {
                return;
            }
        }
        panic!("Failed to cleanup file in 100ms")
    }
}

#[cfg(target_family = "unix")]
#[cfg(test)]
mod unix_test {
    use std::fs::OpenOptions;

    use nix::sys::stat;
    use nix::unistd;
    use tempfile::TempDir;

    use crate::local::LocalFileSystem;
    use crate::{ObjectStoreExt, Path};

    #[tokio::test]
    async fn test_fifo() {
        let filename = "some_file";
        let root = TempDir::new().unwrap();
        let integration = LocalFileSystem::new_with_prefix(root.path()).unwrap();
        let path = root.path().join(filename);
        unistd::mkfifo(&path, stat::Mode::S_IRWXU).unwrap();

        // Need to open read and write side in parallel
        let spawned =
            tokio::task::spawn_blocking(|| OpenOptions::new().write(true).open(path).unwrap());

        let location = Path::from(filename);
        integration.head(&location).await.unwrap();
        integration.get(&location).await.unwrap();

        spawned.await.unwrap();
    }
}