blockworx_store/storage/

mod.rs

//! Where one container's bytes live.
//!
//! The `.bwx` layout — which file, in what order, fsync'd before what — is
//! the container's business and is written once, over this trait. What a
//! [`Storage`] answers is narrower: read, write, append, truncate, list and
//! remove, by container-relative [`Entry`] name, the three facts about the
//! container itself (what it is called, where it is, how much of it is
//! worth holding in memory), and the lock that says who may write it.
//!
//! **Async below, synchronous above.** The editor's read path is
//! synchronous and stays so, while origin-private storage in a browser is
//! not: every method answers a future, and a storage whose futures are
//! ready the moment they are made ([`Native`], [`Memory`]) is resolved with
//! one poll ([`ready_now`]). A storage whose futures are *not* ready is
//! read into memory at open instead ([`Residency::Resident`]), so the reads
//! the editor makes never reach it and the writes it makes are owed to the
//! storage rather than made through it
//! ([`Container::drain`](crate::container::Container::drain)).

pub mod erased;
pub mod memory;
pub mod native;

use core::future::Future;
use std::borrow::Cow;
use std::path::Path;

use super::lock::Claim;
use super::record::WallTime;

pub use erased::Any;
pub use memory::Memory;
pub use native::Native;

/// One file of a container, named relative to it: `manifest.jsonl`,
/// `revs/000012.json.gz`. Never a platform path — the storage is what
/// knows whether there is one.
#[derive(Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Debug)]
pub struct Entry(Cow<'static, str>);

impl Entry {
    /// The container itself, which is what [`Storage::list`] is asked for
    /// when the question is what is in it.
    pub const ROOT: Self = Self::fixed("");

    pub const fn fixed(name: &'static str) -> Self {
        Self(Cow::Borrowed(name))
    }

    /// The entry `name` spells, for a caller reading names back rather than
    /// naming them in the source — a listing, an archive.
    pub fn named(name: &str) -> Self {
        Self(Cow::Owned(name.to_owned()))
    }

    /// `dir/name` — the one spelling of a nested entry, so no two callers
    /// can build it differently.
    pub fn under(dir: &str, name: &str) -> Self {
        Self(Cow::Owned(format!("{dir}/{name}")))
    }

    pub fn as_str(&self) -> &str {
        &self.0
    }
}

impl std::fmt::Display for Entry {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(&self.0)
    }
}

/// What a container is called, wherever it is kept.
///
/// One segment, never a path: a rename gives the document another name and
/// never another home, which is what a rename box is for. The container
/// extension is part of it, because the container's own name *is* the
/// document's.
#[derive(Clone, PartialEq, Eq, PartialOrd, Ord, Debug)]
pub struct Name(String);

/// The document format's extension, on a container.
pub const CONTAINER_EXTENSION: &str = "bwx";

impl Name {
    /// The name as it is spelled, or `None` for one that is no name:
    /// blank, hidden, or more than one segment.
    pub fn new(name: &str) -> Option<Self> {
        let name = name.trim();
        let one_segment = Path::new(name).components().count() == 1;
        if name.is_empty() || name.starts_with('.') || !one_segment {
            return None;
        }
        Some(Self(name.to_owned()))
    }

    /// What the user typed, as a container is named: the suffix is
    /// appended rather than substituted, since a document called `notes.v2`
    /// should not become `notes.bwx`.
    pub fn of_document(name: &str) -> Option<Self> {
        let Self(name) = Self::new(name)?;
        if named_as_a_container(&name) {
            return Some(Self(name));
        }
        Some(Self(format!("{name}.{CONTAINER_EXTENSION}")))
    }

    /// The container an archive's file name says it carries: `engine.bwx.zip`
    /// carries `engine.bwx`.
    pub fn of_archive(file: &str) -> Option<Self> {
        Self::of_document(file.strip_suffix(".zip").unwrap_or(file))
    }

    pub fn as_str(&self) -> &str {
        &self.0
    }
}

impl std::fmt::Display for Name {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(&self.0)
    }
}

fn named_as_a_container(name: &str) -> bool {
    Path::new(name)
        .extension()
        .is_some_and(|ext| ext.eq_ignore_ascii_case(CONTAINER_EXTENSION))
}

/// Which document a door names, as the shell that opens it spells one: a
/// path on a desktop, a name in the library in a browser.
///
/// Opaque to everything in between. The kernel raises it and carries it
/// back; only the shell that minted it resolves it, against its own
/// storage — which is why "where the documents are" never has to become a
/// fact of the crates under it.
#[derive(Clone, PartialEq, Eq, PartialOrd, Ord, Debug, serde::Serialize, serde::Deserialize)]
#[serde(transparent)]
pub struct DocumentRef(String);

impl DocumentRef {
    pub fn new(named: impl Into<String>) -> Self {
        Self(named.into())
    }

    pub fn as_str(&self) -> &str {
        &self.0
    }
}

impl std::fmt::Display for DocumentRef {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(&self.0)
    }
}

impl From<&Path> for DocumentRef {
    fn from(path: &Path) -> Self {
        Self(path.to_string_lossy().into_owned())
    }
}

impl From<&DocumentRef> for std::path::PathBuf {
    fn from(named: &DocumentRef) -> Self {
        Self::from(&named.0)
    }
}

/// How much of a container is worth holding in memory.
///
/// The storage says which, because it is a fact about where the bytes are:
/// a directory is read as it is asked for, and a storage whose reads are
/// really asynchronous is read once at open so that every read the editor
/// makes afterwards is a memory hit.
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
pub enum Residency {
    Lazy,
    Resident,
}

/// Where one container's bytes live.
pub trait Storage {
    /// What this container is called.
    fn name(&self) -> Name;

    /// What a report calls `at` — a platform path where there is one. Not
    /// a promise that anything is there.
    fn names(&self, at: &Entry) -> String;

    fn residency(&self) -> Residency;

    /// Where these bytes sit on a filesystem, for a shell that can name
    /// one. `None` for a storage that has no such thing.
    fn disk_path(&self) -> Option<&Path> {
        None
    }

    /// The bytes at `at`, or `None` where nothing is — which is not the
    /// same as a read that did not work.
    fn read<'a>(
        &'a self,
        at: &'a Entry,
    ) -> impl Future<Output = std::io::Result<Option<Vec<u8>>>> + 'a;

    /// Whether anything stands at `at`. Separate from [`Self::read`]
    /// because a payload is filed under a content hash, so the answer is
    /// all a write needs — and reading a megabyte to learn it is already
    /// there is the one thing that would make every commit cost the
    /// artwork on the page.
    fn exists<'a>(&'a self, at: &'a Entry) -> impl Future<Output = std::io::Result<bool>> + 'a;

    /// Whole file or nothing: a reader caught mid-write sees the previous
    /// contents, never half of each.
    fn write<'a>(
        &'a self,
        at: &'a Entry,
        bytes: &'a [u8],
    ) -> impl Future<Output = std::io::Result<()>> + 'a;

    /// Add `bytes` to the end of `at`, durably before returning.
    fn append<'a>(
        &'a self,
        at: &'a Entry,
        bytes: &'a [u8],
    ) -> impl Future<Output = std::io::Result<()>> + 'a;

    fn truncate<'a>(
        &'a self,
        at: &'a Entry,
        len: u64,
    ) -> impl Future<Output = std::io::Result<()>> + 'a;

    /// What `dir` holds, by name, in no particular order.
    fn list<'a>(
        &'a self,
        dir: &'a Entry,
    ) -> impl Future<Output = std::io::Result<Vec<String>>> + 'a;

    fn remove<'a>(&'a self, at: &'a Entry) -> impl Future<Output = std::io::Result<()>> + 'a;

    /// Make `dir` a place entries can be written.
    fn create_dir<'a>(&'a self, dir: &'a Entry) -> impl Future<Output = std::io::Result<()>> + 'a;

    /// Call this container something else, leaving everything in it where
    /// it is — which is what renaming the document means.
    fn rename<'a>(&'a mut self, to: &'a Name) -> impl Future<Output = std::io::Result<()>> + 'a;

    /// Remove the container itself. Only ever asked of one that has been
    /// emptied.
    fn discard(&self) -> impl Future<Output = std::io::Result<()>> + '_;

    /// Claim the right to write this container.
    ///
    /// The default is the [lock entry](super::lock) a container has always
    /// carried: a row naming the process that holds it, broken when that
    /// process is gone. A storage whose host keeps locks of its own
    /// answers from that instead — which is not a nicety in a browser,
    /// where nothing can be asked whether the tab that left an entry
    /// behind is still there.
    fn claim(&self, now: WallTime) -> impl Future<Output = std::io::Result<Claim>> + '_ {
        super::lock::claim(self, now)
    }

    /// Give the lock back.
    ///
    /// Synchronous where the claim is not, because a container releases
    /// its lock as it is dropped and a `Drop` cannot await. Which is why a
    /// storage whose futures are not ready must hold its lock somewhere it
    /// can let go of in one call — its host's lock manager — rather than
    /// in the entry the default writes: an entry it could not remove here
    /// would outlive every session that took it.
    fn release(&self) {
        super::lock::release(self);
    }
}

/// Resolve a future that promised to be ready the moment it was made.
///
/// The editor's read path is synchronous, so a storage whose
/// [`Storage::residency`] is [`Residency::Lazy`] is read through this: one
/// poll, no executor, no waker. A storage that answers `Pending` here has
/// broken that promise — a programming error, reported as one rather than
/// blocking a frame on a spin.
pub fn ready_now<T, E: From<std::io::Error>>(
    future: impl Future<Output = Result<T, E>>,
) -> Result<T, E> {
    use core::task::{Context, Poll};
    let mut future = core::pin::pin!(future);
    match future
        .as_mut()
        .poll(&mut Context::from_waker(core::task::Waker::noop()))
    {
        Poll::Ready(done) => done,
        Poll::Pending => Err(std::io::Error::new(
            std::io::ErrorKind::WouldBlock,
            "this storage is read as if it were ready, and it was not",
        )
        .into()),
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn a_name_is_one_segment_and_carries_the_container_extension() {
        assert_eq!(
            Name::of_document("motor-controller").map(|n| n.to_string()),
            Some("motor-controller.bwx".to_owned()),
        );
        assert_eq!(
            Name::of_document("  motor-controller  ").map(|n| n.to_string()),
            Some("motor-controller.bwx".to_owned()),
            "the box's whitespace is not part of the name",
        );
        assert_eq!(
            Name::of_document("notes.v2").map(|n| n.to_string()),
            Some("notes.v2.bwx".to_owned()),
            "the suffix is appended, so a dotted name keeps all of itself",
        );
        assert_eq!(
            Name::of_document("engine.bwx").map(|n| n.to_string()),
            Some("engine.bwx".to_owned()),
            "a name already carrying the extension does not gain a second",
        );
        for refused in [
            "",
            "   ",
            ".hidden",
            "..",
            "sub/engine",
            "/elsewhere/engine",
        ] {
            assert_eq!(
                Name::of_document(refused),
                None,
                "{refused:?} was taken for a document name",
            );
        }
    }

    #[test]
    fn an_entry_is_named_relative_to_its_container() {
        assert_eq!(
            Entry::under("revs", "000012.json.gz").as_str(),
            "revs/000012.json.gz"
        );
        assert_eq!(Entry::ROOT.as_str(), "");
    }

    /// The promise the synchronous read path rests on, and what happens
    /// when a storage breaks it.
    #[test]
    fn a_future_that_is_not_ready_is_reported_rather_than_waited_on() {
        assert_eq!(
            ready_now(core::future::ready(std::io::Result::Ok(7))).expect("ready"),
            7
        );
        let refusal = ready_now(core::future::pending::<std::io::Result<u8>>())
            .expect_err("a pending future cannot be resolved");
        assert_eq!(refusal.kind(), std::io::ErrorKind::WouldBlock);
    }
}