feldera_storage/

lib.rs

//! Common Types and Trait Definition for Storage in Feldera.

use std::collections::HashSet;
use std::fmt::Debug;
use std::io::{Cursor, ErrorKind};
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::sync::atomic::AtomicI64;

use feldera_types::checkpoint::{CheckpointDependencies, CheckpointMetadata, PSpineBatches};
use feldera_types::config::{StorageBackendConfig, StorageConfig, StorageOptions};
use feldera_types::constants::{CHECKPOINT_DEPENDENCIES, CREATE_FILE_EXTENSION};
use serde::de::DeserializeOwned;
use tracing::warn;
use uuid::Uuid;

use crate::block::BlockLocation;
use crate::error::StorageError;
use crate::fbuf::FBuf;
use crate::file::FileId;

pub use object_store::path::{Path as StoragePath, PathPart as StoragePathPart};

pub mod block;
pub mod checkpoint_synchronizer;
pub mod error;
pub mod fbuf;
pub mod file;
pub mod histogram;
pub mod metrics;
pub mod tokio;

/// Helper function that appends to a [`PathBuf`].
pub fn append_to_path(p: PathBuf, s: &str) -> PathBuf {
    let mut p = p.into_os_string();
    p.push(s);
    p.into()
}

pub trait StorageBackendFactory: Sync {
    fn backend(&self) -> &'static str;
    fn create(
        &self,
        storage_config: &StorageConfig,
        backend_config: &StorageBackendConfig,
    ) -> Result<Arc<dyn StorageBackend>, StorageError>;
}

inventory::collect!(&'static dyn StorageBackendFactory);

/// A storage backend.
pub trait StorageBackend: Send + Sync {
    /// Create a new file with the given `name`, automatically creating any
    /// parent directories within `name` that don't already exist.
    fn create_named(&self, name: &StoragePath) -> Result<Box<dyn FileWriter>, StorageError>;

    /// Creates a new persistent file used for writing data. The backend selects
    /// a name.
    fn create(&self) -> Result<Box<dyn FileWriter>, StorageError> {
        self.create_with_prefix(&StoragePath::default())
    }

    /// Creates a new persistent file used for writing data, giving the file's
    /// name the specified `prefix`. See also [`create`](Self::create).
    fn create_with_prefix(
        &self,
        prefix: &StoragePath,
    ) -> Result<Box<dyn FileWriter>, StorageError> {
        let uuid = Uuid::now_v7();
        let name = format!("{}{}{}", prefix, uuid, CREATE_FILE_EXTENSION);
        self.create_named(&name.into())
    }

    /// Opens `name` for reading.
    fn open(&self, name: &StoragePath) -> Result<Arc<dyn FileReader>, StorageError>;

    /// Returns the base directory path on the local file system if the storage backend
    /// uses local disk.
    fn file_system_path(&self) -> Option<&Path> {
        None
    }

    /// Calls `cb` with the name and type of each file under `parent`. This is a
    /// non-recursive list: it does not include files under sub-directories of
    /// `parent`.
    ///
    /// This method can report two classes of errors:
    ///
    /// - The return value indicates errors that could prevent `cb` from being
    ///   called for some or all of the files in the directory.  These errors
    ///   indicate that `parent` does not exist or cannot be (fully) read
    ///   successfully.  If more than one such error occurs, the method returns
    ///   the last one.
    ///
    /// - [DirEntry::file_type] in the argument to the callback reports errors
    ///   obtaining the file type.  Errors reported this way only mean that
    ///   there was a problem obtaining metadata for the file itself.
    fn list(&self, parent: &StoragePath, cb: &mut dyn FnMut(DirEntry)) -> Result<(), StorageError>;

    fn delete(&self, name: &StoragePath) -> Result<(), StorageError>;

    fn delete_recursive(&self, name: &StoragePath) -> Result<(), StorageError>;

    fn delete_if_exists(&self, name: &StoragePath) -> Result<(), StorageError> {
        match self.delete(name) {
            Err(error) if error.kind() == ErrorKind::NotFound => Ok(()),
            other => other,
        }
    }

    fn exists(&self, name: &StoragePath) -> Result<bool, StorageError> {
        match self.open(name) {
            Ok(_) => Ok(true),
            Err(error) if error.kind() == ErrorKind::NotFound => Ok(false),
            Err(error) => Err(error),
        }
    }

    /// Reads `name` and returns its contents.  The file `name` is relative to
    /// the base of the storage backend.
    fn read(&self, name: &StoragePath) -> Result<Arc<FBuf>, StorageError> {
        let reader = self.open(name)?;
        let size = reader.get_size()?.try_into().unwrap();
        reader.read_block(BlockLocation { offset: 0, size })
    }

    /// Writes `content` to `name`, automatically creating any parent
    /// directories within `name` that don't already exist.
    ///
    /// The caller must call `commit` on the returned file if it wants to make
    /// sure that the file is committed to stable storage.
    fn write(
        &self,
        name: &StoragePath,
        content: FBuf,
    ) -> Result<Arc<dyn FileCommitter>, StorageError> {
        let mut writer = self.create_named(name)?;
        writer.write_block(content)?;
        let reader = writer.complete()?;
        reader.mark_for_checkpoint();
        Ok(reader)
    }

    /// Flushes the directory entry metadata for `dir` to stable storage.
    ///
    /// POSIX `fsync(file)` guarantees the file's data is durable but says
    /// nothing about whether the file's name in its parent directory is
    /// durable. After renaming files into a directory or creating new
    /// subdirectories, the caller can use this barrier to make those
    /// directory entries durable in one syscall.
    ///
    /// Backends that do not have a meaningful notion of directory
    /// durability (e.g. object stores) may treat this as a no-op.
    fn fsync_dir(&self, _dir: &StoragePath) -> Result<(), StorageError> {
        Ok(())
    }

    /// Returns a value that represents the number of bytes of storage in use.
    /// The storage backend updates this value when its own functions cause more
    /// or less storage to be used:
    ///
    /// - Writing to a file.
    ///
    /// - Deleting a file (by dropping a [FileWriter] without completing, or by
    ///   dropping a [FileReader] without marking it for a checkpoint, or by
    ///   calling functions to delete files.
    ///
    /// The backend is *not* required to:
    ///
    /// - Initially report how much storage is in use. Instead, it just starts
    ///   out at zero. The client can traverse the storage itself and store the
    ///   correct initial value.
    ///
    /// - Detect changes made by a different backend or outside any backend.
    ///
    /// The value is signed because the problems above can cause it to become
    /// negative.
    fn usage(&self) -> Arc<AtomicI64>;
}

impl dyn StorageBackend {
    /// Creates and returns a new backend configured according to `config` and `options`.
    pub fn new(
        config: &StorageConfig,
        options: &StorageOptions,
    ) -> Result<Arc<Self>, StorageError> {
        Self::warn_about_tmpfs(config.path());
        for variable_provider in inventory::iter::<&dyn StorageBackendFactory> {
            if variable_provider.backend() == options.backend.to_string() {
                return variable_provider.create(config, &options.backend);
            }
        }
        Err(StorageError::BackendNotSupported(Box::new(
            options.backend.clone(),
        )))
    }

    fn is_tmpfs(_path: &Path) -> bool {
        #[cfg(target_os = "linux")]
        {
            use nix::sys::statfs;
            statfs::statfs(_path).is_ok_and(|s| s.filesystem_type() == statfs::TMPFS_MAGIC)
        }

        #[cfg(not(target_os = "linux"))]
        false
    }

    fn warn_about_tmpfs(path: &Path) {
        if Self::is_tmpfs(path) {
            static ONCE: std::sync::Once = std::sync::Once::new();
            ONCE.call_once(|| {
                warn!("initializing storage on in-memory tmpfs filesystem at {}; consider configuring physical storage", path.display())
            });
        }
    }

    pub fn gather_batches_for_checkpoint_uuid(
        &self,
        cpm: uuid::Uuid,
    ) -> Result<HashSet<StoragePath>, StorageError> {
        assert!(!cpm.is_nil());

        let checkpoint_dir: StoragePath = cpm.to_string().into();

        // `dependencies.json` holds the authoritative snapshot of every
        // batch referenced by the checkpoint at commit time. Prefer it
        // over the per-spine `pspine-batches-*.dat` scan below: a valid
        // commit always writes it, and it captures the full list in one
        // place so a single read suffices. See `CheckpointDependencies`
        // for the accepted JSON forms.
        let deps_path = checkpoint_dir.child(CHECKPOINT_DEPENDENCIES);
        match self.read_json::<CheckpointDependencies>(&deps_path) {
            Ok(deps) => {
                return Ok(deps
                    .batches()
                    .iter()
                    .map(|b| StoragePath::from(b.as_str()))
                    .collect());
            }
            Err(error) if error.kind() == ErrorKind::NotFound => {
                // Fall back to scanning per-spine sidecars. Predates the
                // `dependencies.json` snapshot entirely.
            }
            Err(error) => return Err(error),
        }

        // Legacy fallback. New checkpoints always carry a
        // `dependencies.json` (the early-return above), so this scan only
        // runs for checkpoints written before that file existed.
        // TODO: remove once no such old checkpoints remain in use.
        let mut spines = Vec::new();
        self.list(&checkpoint_dir, &mut |entry| {
            if let Some(filename) = entry.name.filename()
                && filename.starts_with("pspine-batches")
            {
                spines.push(entry.name);
            }
        })?;

        let mut batch_files_in_commit: HashSet<StoragePath> = HashSet::new();
        for spine in spines {
            let pspine_batches = self.read_json::<PSpineBatches>(&spine)?;
            for file in pspine_batches.files {
                batch_files_in_commit.insert(file.into());
            }
        }

        Ok(batch_files_in_commit)
    }

    pub fn gather_batches_for_checkpoint(
        &self,
        cpm: &CheckpointMetadata,
    ) -> Result<HashSet<StoragePath>, StorageError> {
        self.gather_batches_for_checkpoint_uuid(cpm.uuid)
    }
}

// For an explanation of the `+ '_` here, see:
// https://stackoverflow.com/questions/73495603/trait-problem-borrowed-data-escapes-outside-of-associated-function
impl dyn StorageBackend + '_ {
    /// Writes `content` to `name` as JSON, automatically creating any parent
    /// directories within `name` that don't already exist.
    ///
    /// The caller must call `commit` on the returned file if it wants to make
    /// sure that the file is committed to stable storage.
    pub fn write_json<V>(
        &self,
        name: &StoragePath,
        value: &V,
    ) -> Result<Arc<dyn FileCommitter>, StorageError>
    where
        V: serde::Serialize,
    {
        let mut content = FBuf::new();
        serde_json::to_writer(&mut content, value).unwrap();
        self.write(name, content)
    }

    /// Reads `name` as JSON.
    pub fn read_json<V>(&self, name: &StoragePath) -> Result<V, StorageError>
    where
        V: DeserializeOwned,
    {
        let content = self.read(name)?;
        serde_json::from_reader(Cursor::new(content.as_ref()))
            .map_err(|e| StorageError::JsonError(e.to_string()))
    }
}

/// A directory entry read by [StorageBackend::list].
pub struct DirEntry {
    /// File name.
    pub name: StoragePath,

    /// File type, if it could be obtained.
    pub file_type: Result<StorageFileType, StorageError>,
}

/// A file being read or written.
pub trait FileRw {
    /// Returns the file's unique ID.
    fn file_id(&self) -> FileId;

    /// Returns the file's path.
    fn path(&self) -> &StoragePath;
}

/// A file being written.
///
/// The file can't be read until it is completed with
/// [FileWriter::complete]. Until then, the file is temporary and will be
/// deleted if it is dropped.
pub trait FileWriter: Send + Sync + FileRw {
    /// Writes `data` at the end of the file. len()` must be a multiple of 512.
    /// Returns the data that was written encapsulated in an `Arc`.
    fn write_block(&mut self, data: FBuf) -> Result<Arc<FBuf>, StorageError>;

    /// Completes writing of a file and returns a reader for the file.
    ///
    /// The file will be deleted if the reader is dropped without calling
    /// [FileReader::mark_for_checkpoint].
    ///
    /// The file is not necessarily committed to stable storage before calling
    /// `commit` on the returned file.
    fn complete(self: Box<Self>) -> Result<Arc<dyn FileReader>, StorageError>;
}

/// Allows a file to be committed to stable storage.
///
/// This is a supertrait of [FileReader] that only allows the commit operation.
/// It's somewhat surprising that a file that can't be written can be committed,
/// but it makes sense in the context of [FileWriter::complete] returning a
/// [FileReader] that isn't necessarily committed yet.  Making this a separate
/// trait allows code to split off a `FileCommitter` to hand to a piece of code
/// that only needs to be able to commit it.
pub trait FileCommitter: Send + Sync + Debug + FileRw {
    /// Commits the file to stable storage.
    fn commit(&self) -> Result<(), StorageError>;
}

/// A readable file.
pub trait FileReader: Send + Sync + Debug + FileRw + FileCommitter {
    /// Marks a file to be part of a checkpoint.
    ///
    /// This is used to prevent the file from being deleted when it is dropped.
    /// This is only useful for files obtained via [FileWriter::complete],
    /// because files that were opened with [StorageBackend::open] are never
    /// deleted on drop.
    fn mark_for_checkpoint(&self);

    /// Reads data at `location` from the file.  If successful, the result will
    /// be exactly the requested length; that is, this API treats read past EOF
    /// as an error.
    fn read_block(&self, location: BlockLocation) -> Result<Arc<FBuf>, StorageError>;

    /// Initiates an asynchronous read.  When the read completes, `callback`
    /// will be called.
    ///
    /// The default implementation is not actually asynchronous.
    #[allow(clippy::type_complexity)]
    fn read_async(
        &self,
        blocks: Vec<BlockLocation>,
        callback: Box<dyn FnOnce(Vec<Result<Arc<FBuf>, StorageError>>) + Send>,
    ) {
        default_read_async(self, blocks, callback);
    }

    /// Returns the file's size in bytes.
    fn get_size(&self) -> Result<u64, StorageError>;
}

/// Default implementation for [FileReader::read_async].
///
/// This implementation is not actually asynchronous.
#[allow(clippy::type_complexity)]
pub fn default_read_async<R>(
    reader: &R,
    blocks: Vec<BlockLocation>,
    callback: Box<dyn FnOnce(Vec<Result<Arc<FBuf>, StorageError>>) + Send>,
) where
    R: FileReader + ?Sized,
{
    callback(
        blocks
            .into_iter()
            .map(|location| reader.read_block(location))
            .collect(),
    )
}

#[derive(Copy, Clone, Debug, PartialEq, Eq)]
pub enum StorageFileType {
    /// A regular file.
    File {
        /// File size in bytes.
        size: u64,
    },

    /// A directory.
    ///
    /// Only some kinds of storage backends support directories. The ones that
    /// don't still allow files to be named hierarchically, but they don't
    /// support creating or deleting directories independently from the files in
    /// them. That is, with such a backend, a directory is effectively created
    /// by creating a file in it, and is effectively deleted when the last file
    /// in it is deleted.
    Directory,

    /// Something else.
    Other,
}