blockworx_store/storage/

memory.rs

//! A container in a map, for a session with nothing to write to and for
//! tests on every target.
//!
//! The browser's facts are knobs here, so the paths that only origin-private
//! storage takes in earnest are driven natively: futures that are not ready
//! when they are made and a lock the host keeps rather than the container
//! ([`Memory::deferred`]), and a write that will not land
//! ([`Memory::refusing`]).

use core::future::Future;
use core::pin::Pin;
use core::task::{Context, Poll};
use std::cell::{Cell, RefCell};
use std::collections::{BTreeMap, BTreeSet};
use std::rc::Rc;

use super::{Entry, Name, Residency, Storage};
use crate::lock::{Claim, Holding};
use crate::record::WallTime;

/// A container's bytes, keyed by entry name, dying with the process.
///
/// A handle on a map rather than the map itself: cloning one hands a second
/// container the same bytes, which is what re-opening a container means
/// where there is no directory to point at twice.
#[derive(Clone, Debug)]
pub struct Memory(Rc<Held>);

#[derive(Debug)]
struct Held {
    name: RefCell<Name>,
    residency: Residency,
    pace: Pace,
    files: RefCell<BTreeMap<String, Vec<u8>>>,
    dirs: RefCell<BTreeSet<String>>,
    refused: RefCell<Option<Entry>>,
    /// Whether this map keeps its own lock, and whether that lock is held.
    /// `None` for one whose lock is the container's own entry.
    own_lock: Cell<Option<bool>>,
    reads: Cell<usize>,
    wrote: RefCell<Vec<Entry>>,
}

/// Whether this map's futures are ready the moment they are made.
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
enum Pace {
    Ready,
    Deferred,
}

impl Memory {
    /// Read as a directory is: an entry at a time, as it is asked for.
    pub fn new(name: &str) -> Self {
        Self::with(name, Residency::Lazy, Pace::Ready)
    }

    /// Read whole at open, with futures that are ready anyway — the
    /// residency without the waiting.
    pub fn resident(name: &str) -> Self {
        Self::with(name, Residency::Resident, Pace::Ready)
    }

    /// A map as asynchronous as origin-private storage is: every future
    /// answers `Pending` once before it completes, so nothing may resolve
    /// one by polling it — and, as [`Storage::release`] requires of such a
    /// storage, a lock of its own rather than an entry it could not remove
    /// as it is dropped.
    pub fn deferred(name: &str) -> Self {
        let deferred = Self::with(name, Residency::Resident, Pace::Deferred);
        deferred.0.own_lock.set(Some(false));
        deferred
    }

    /// A map that will not write `at`, so a storage that fails partway
    /// through what it was handed can be stood up without breaking one.
    #[must_use]
    pub fn refusing(self, at: Entry) -> Self {
        *self.0.refused.borrow_mut() = Some(at);
        self
    }

    fn with(name: &str, residency: Residency, pace: Pace) -> Self {
        Self(Rc::new(Held {
            name: RefCell::new(Name::new(name).unwrap_or_else(|| Name(name.to_owned()))),
            residency,
            pace,
            files: RefCell::default(),
            dirs: RefCell::default(),
            refused: RefCell::default(),
            own_lock: Cell::new(None),
            reads: Cell::new(0),
            wrote: RefCell::default(),
        }))
    }

    /// How many entries have been read out of this map — the count that
    /// must stop moving once a resident container is open.
    pub fn reads(&self) -> usize {
        self.0.reads.get()
    }

    /// What has been written to this map, in the order it landed.
    pub fn wrote(&self) -> Vec<Entry> {
        self.0.wrote.borrow().clone()
    }

    /// Answer `Pending` once where this map is the deferred kind, so a
    /// caller that resolves its futures by polling them is caught.
    async fn paced(&self) {
        if let Pace::Deferred = self.0.pace {
            Wait(false).await;
        }
    }

    fn read_one(&self) {
        self.0.reads.set(self.0.reads.get() + 1);
    }

    fn wrote_one(&self, at: &Entry) -> std::io::Result<()> {
        if self.0.refused.borrow().as_ref() == Some(at) {
            return Err(std::io::Error::new(
                std::io::ErrorKind::PermissionDenied,
                format!("{at} is an entry this map refuses to write"),
            ));
        }
        self.0.wrote.borrow_mut().push(at.clone());
        Ok(())
    }
}

/// A future that answers `Pending` once before it is ready.
struct Wait(bool);

impl Future for Wait {
    type Output = ();

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
        if self.0 {
            return Poll::Ready(());
        }
        self.0 = true;
        cx.waker().wake_by_ref();
        Poll::Pending
    }
}

impl Default for Memory {
    fn default() -> Self {
        Self::new("session.bwx")
    }
}

/// The name `dir` holds `key` under, or `None` where `key` is not in it.
/// The root holds everything with no separator left in it.
fn within<'a>(dir: &str, key: &'a str) -> Option<&'a str> {
    let rest = if dir.is_empty() {
        key
    } else {
        key.strip_prefix(dir)?.strip_prefix('/')?
    };
    (!rest.contains('/')).then_some(rest)
}

impl Storage for Memory {
    fn name(&self) -> Name {
        self.0.name.borrow().clone()
    }

    fn names(&self, at: &Entry) -> String {
        format!("{at} of {}", self.0.name.borrow())
    }

    fn residency(&self) -> Residency {
        self.0.residency
    }

    async fn read<'a>(&'a self, at: &'a Entry) -> std::io::Result<Option<Vec<u8>>> {
        self.paced().await;
        self.read_one();
        Ok(self.0.files.borrow().get(at.as_str()).cloned())
    }

    async fn exists<'a>(&'a self, at: &'a Entry) -> std::io::Result<bool> {
        self.paced().await;
        self.read_one();
        Ok(self.0.files.borrow().contains_key(at.as_str()))
    }

    async fn write<'a>(&'a self, at: &'a Entry, bytes: &'a [u8]) -> std::io::Result<()> {
        self.paced().await;
        self.wrote_one(at)?;
        self.0
            .files
            .borrow_mut()
            .insert(at.as_str().to_owned(), bytes.to_vec());
        Ok(())
    }

    async fn append<'a>(&'a self, at: &'a Entry, bytes: &'a [u8]) -> std::io::Result<()> {
        self.paced().await;
        self.wrote_one(at)?;
        self.0
            .files
            .borrow_mut()
            .entry(at.as_str().to_owned())
            .or_default()
            .extend_from_slice(bytes);
        Ok(())
    }

    async fn truncate<'a>(&'a self, at: &'a Entry, len: u64) -> std::io::Result<()> {
        self.paced().await;
        self.wrote_one(at)?;
        if let Some(bytes) = self.0.files.borrow_mut().get_mut(at.as_str()) {
            bytes.truncate(len as usize);
        }
        Ok(())
    }

    async fn list<'a>(&'a self, dir: &'a Entry) -> std::io::Result<Vec<String>> {
        self.paced().await;
        self.read_one();
        let mut names: Vec<String> = self
            .0
            .dirs
            .borrow()
            .iter()
            .filter_map(|key| within(dir.as_str(), key))
            .map(ToOwned::to_owned)
            .collect();
        names.extend(
            self.0
                .files
                .borrow()
                .keys()
                .filter_map(|key| within(dir.as_str(), key))
                .map(ToOwned::to_owned),
        );
        Ok(names)
    }

    async fn remove<'a>(&'a self, at: &'a Entry) -> std::io::Result<()> {
        self.paced().await;
        let gone = self.0.files.borrow_mut().remove(at.as_str()).is_some()
            || self.0.dirs.borrow_mut().remove(at.as_str());
        if gone {
            return Ok(());
        }
        Err(std::io::Error::from(std::io::ErrorKind::NotFound))
    }

    async fn create_dir<'a>(&'a self, dir: &'a Entry) -> std::io::Result<()> {
        self.paced().await;
        if !dir.as_str().is_empty() {
            self.0.dirs.borrow_mut().insert(dir.as_str().to_owned());
        }
        Ok(())
    }

    async fn rename<'a>(&'a mut self, to: &'a Name) -> std::io::Result<()> {
        self.paced().await;
        *self.0.name.borrow_mut() = to.clone();
        Ok(())
    }

    async fn discard(&self) -> std::io::Result<()> {
        self.paced().await;
        self.0.files.borrow_mut().clear();
        self.0.dirs.borrow_mut().clear();
        Ok(())
    }

    async fn claim(&self, now: WallTime) -> std::io::Result<Claim> {
        self.paced().await;
        match self.0.own_lock.get() {
            None => crate::lock::claim(self, now).await,
            Some(true) => Ok(Claim::Held(Holding::Elsewhere)),
            Some(false) => {
                self.0.own_lock.set(Some(true));
                Ok(Claim::Taken)
            }
        }
    }

    fn release(&self) {
        match self.0.own_lock.get() {
            None => crate::lock::release(self),
            Some(_) => self.0.own_lock.set(Some(false)),
        }
    }
}

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

    const MANIFEST: Entry = Entry::fixed("manifest.jsonl");

    #[test]
    fn entries_round_trip_and_list_under_the_directory_that_holds_them() {
        let memory = Memory::new("doc.bwx");
        ready_now(memory.create_dir(&Entry::fixed("revs"))).expect("the directory");
        ready_now(memory.append(&MANIFEST, b"one\n")).expect("a row");
        ready_now(memory.append(&MANIFEST, b"two\n")).expect("another row");
        let rev = Entry::under("revs", "000001.json");
        ready_now(memory.write(&rev, b"a document")).expect("a rev");

        assert_eq!(
            ready_now(memory.read(&MANIFEST)).expect("it reads back"),
            Some(b"one\ntwo\n".to_vec()),
        );
        let mut root = ready_now(memory.list(&Entry::ROOT)).expect("it lists");
        root.sort();
        assert_eq!(root, ["manifest.jsonl", "revs"]);
        assert_eq!(
            ready_now(memory.list(&Entry::fixed("revs"))).expect("it lists"),
            ["000001.json"],
            "a nested entry was listed at the root, or not under its directory",
        );

        ready_now(memory.truncate(&MANIFEST, 4)).expect("the truncate");
        assert_eq!(
            ready_now(memory.read(&MANIFEST)).expect("it reads back"),
            Some(b"one\n".to_vec()),
        );
        ready_now(memory.remove(&rev)).expect("the rev goes");
        assert_eq!(
            ready_now(memory.read(&rev)).expect("the read answers"),
            None
        );
    }

    #[test]
    fn a_map_is_read_as_it_is_asked_and_can_be_asked_to_be_read_whole() {
        assert_eq!(Memory::new("a").residency(), Residency::Lazy);
        assert_eq!(Memory::resident("a").residency(), Residency::Resident);
        assert_eq!(Memory::deferred("a").residency(), Residency::Resident);
    }

    /// Two handles on one map are one container's bytes seen twice — how a
    /// container is re-opened where there is no directory to point at.
    #[test]
    fn a_clone_is_the_same_map_and_not_a_copy_of_it() {
        let first = Memory::new("doc.bwx");
        let second = first.clone();
        ready_now(first.write(&MANIFEST, b"a row\n")).expect("the write lands");

        assert_eq!(
            ready_now(second.read(&MANIFEST)).expect("it reads back"),
            Some(b"a row\n".to_vec()),
        );
    }

    /// The promise the synchronous read path rests on, broken deliberately:
    /// a deferred map's futures cannot be resolved by polling them once.
    #[test]
    fn a_deferred_map_refuses_to_be_read_as_if_it_were_ready() {
        let deferred = Memory::deferred("doc.bwx");
        assert_eq!(
            ready_now(deferred.read(&MANIFEST))
                .expect_err("a deferred future must not resolve in one poll")
                .kind(),
            std::io::ErrorKind::WouldBlock,
        );
        assert_eq!(
            crate::fixture::block_on(deferred.read(&MANIFEST)).expect("driven, it answers"),
            None,
        );
    }

    #[test]
    fn a_refused_entry_is_the_only_one_that_does_not_write() {
        let memory = Memory::new("doc.bwx").refusing(MANIFEST);
        assert_eq!(
            ready_now(memory.append(&MANIFEST, b"a row\n"))
                .expect_err("the refusal")
                .kind(),
            std::io::ErrorKind::PermissionDenied,
        );
        ready_now(memory.write(&Entry::fixed("document.json"), b"{}")).expect("another entry");
        assert_eq!(memory.wrote(), [Entry::fixed("document.json")]);
    }
}