blockworx_store/

container.rs

//! The `.bwx` container: a manifest, the revs it names, and the artwork
//! they place.
//!
//! ```text
//! doc.bwx/
//!   revs/           the whole document at every rev, compressed — THE document
//!   manifest.jsonl  one appended row per rev: who, when, where, what it named
//!   assets/         content-addressed artwork
//!   lock            advisory single-writer lock
//!   .gitattributes  how git must treat them
//! ```
//!
//! Which file, in what order, and fsync'd before what is written here and
//! once, over a [`Storage`] that says only where the bytes go — a
//! directory of real files on a desktop, a map in a test. On a filesystem
//! that makes the container a directory rather than a packed file, because
//! appends, content-addressed blobs and per-writer sidecars all want to be
//! separate files, and because git and text tools then handle a container
//! natively.

use std::cell::RefCell;
use std::collections::{BTreeMap, VecDeque};

use super::assets::{AssetFormat, AssetSink, AssetSource};
use super::lock::{Claim, Holding};
use super::manifest::{self, BreakReport, Row};
use super::record::{Digest, WallTime};
use super::revs::{self, REVS};
use super::storage::{Entry, Name, Residency, Storage, ready_now};

use blockworx_doc::{block_model::Asset, document::Document, hash::AssetHash, rev::Rev};

pub const MANIFEST: Entry = Entry::fixed("manifest.jsonl");
pub const LOCK: Entry = Entry::fixed("lock");
pub const GITATTRIBUTES: Entry = Entry::fixed(".gitattributes");
/// The directory a container files its artwork in.
pub const ASSETS: &str = "assets";

/// Two divergent `manifest.jsonl`s are concurrent editing through the back door,
/// which this design deliberately has no answer for, so the conflict is
/// surfaced at merge time instead of waiting for a replay to refuse the
/// result.
const GITATTRIBUTES_TEMPLATE: &str = "\
# A blockworx container. revs/ holds the document at every rev and
# manifest.jsonl is the trail that names them.
#
# The merge driver named here is not built in. Define it once, globally:
#     git config --global merge.conflict-always.name 'never auto-merge'
#     git config --global merge.conflict-always.driver false
# Without it git falls back to the text merge, and a merged manifest will
# be refused by the integrity chain at load — loudly, but later.
manifest.jsonl merge=conflict-always
revs/** binary
assets/** binary
lock export-ignore
";

/// Why a container is open without the right to write it. Every arm is
/// something the user must be told, which is why this is not a bool.
#[derive(Debug)]
pub enum ReadOnlyReason {
    /// Another live process holds the advisory lock.
    Locked(Holding),
    /// Opened to be read rather than edited — someone else's document being
    /// looked at. The lock was never claimed, which is the
    /// whole point: a reader must not stand between an owner and their own
    /// document.
    Reading,
    /// The manifest does not verify; only the prefix before the break is
    /// real.
    HistoryBroken(BreakReport),
    /// A row could not be appended, so this session's document and the
    /// files have parted ways. Nothing more is written until the container
    /// is reopened.
    WriteFailed(std::io::Error),
}

impl std::fmt::Display for ReadOnlyReason {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            ReadOnlyReason::Locked(held) => {
                write!(f, "another blockworx ({held}) has this diagram open")
            }
            ReadOnlyReason::Reading => write!(f, "this container was opened to be read"),
            ReadOnlyReason::HistoryBroken(report) => {
                write!(f, "the manifest does not verify at {report}")
            }
            ReadOnlyReason::WriteFailed(error) => {
                write!(f, "a row could not be appended: {error}")
            }
        }
    }
}

/// Whether this handle may write. The writable arm *is* the lock: it is
/// taken when the container is attached and given up when the handle
/// closes or is demoted, so no caller has a guard to remember.
pub enum Access {
    Writable,
    ReadOnly(ReadOnlyReason),
}

/// Why a container could not be opened at all — as opposed to opened
/// read-only, which is an outcome rather than a failure.
#[derive(Debug, thiserror::Error)]
pub enum ContainerError {
    #[error("{0} already holds a diagram")]
    Exists(String),
    #[error("{0} is not a blockworx diagram: it has no {MANIFEST}")]
    NotAContainer(String),
    #[error(transparent)]
    Io(#[from] std::io::Error),
}

/// What of the container is kept in memory.
///
/// A storage read as it is asked for keeps nothing: every read is a file
/// read, as it has always been, and every write is made before the call
/// that made it returns. A storage whose reads are really asynchronous —
/// origin-private storage in a browser — is read whole when it is
/// attached, so the editor's synchronous read path never reaches it, and
/// its writes are made to that memory and owed to the storage
/// ([`Container::drain`]).
enum Kept {
    Nothing,
    Everything(RefCell<Held>),
}

#[derive(Default)]
struct Held {
    bytes: BTreeMap<Entry, Vec<u8>>,
    owed: VecDeque<(Entry, Op)>,
}

/// One write a container has made to its memory and not yet to its
/// storage.
#[derive(Clone, Debug)]
enum Op {
    Write(Vec<u8>),
    Append(Vec<u8>),
    Truncate(u64),
}

/// How many of the writes it owed a container has made.
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
pub struct Drained(usize);

impl Drained {
    pub const NOTHING: Self = Self(0);

    pub fn performed(self) -> usize {
        self.0
    }
}

pub struct Container<S: Storage> {
    storage: S,
    access: Access,
    kept: Kept,
}

impl<S: Storage> Container<S> {
    /// Lay out a new container and take its lock.
    ///
    /// # Errors
    /// [`ContainerError::Exists`] if the storage already holds a manifest,
    /// or the underlying I/O failure.
    pub fn create(storage: S, now: WallTime) -> Result<Self, ContainerError> {
        ready_now(Self::creating(storage, now))
    }

    /// [`Self::create`] for a storage whose futures are not ready when they
    /// are made.
    ///
    /// The layout goes down through the storage itself rather than through
    /// the journal: there is no container to owe a write to until this
    /// returns.
    ///
    /// # Errors
    /// As [`Self::create`].
    pub async fn creating(storage: S, now: WallTime) -> Result<Self, ContainerError> {
        Self::creating_holding(storage, now, b"").await
    }

    /// The same, with `rows` already in it — the prefix Save-as, which
    /// copies an existing manifest's lines rather than re-writing them.
    /// The bytes go down before the lock is taken, so nothing can append
    /// into the middle of them.
    ///
    /// # Errors
    /// As [`Self::create`].
    pub fn create_holding(storage: S, now: WallTime, rows: &[u8]) -> Result<Self, ContainerError> {
        ready_now(Self::creating_holding(storage, now, rows))
    }

    /// # Errors
    /// As [`Self::create_holding`].
    pub async fn creating_holding(
        storage: S,
        now: WallTime,
        rows: &[u8],
    ) -> Result<Self, ContainerError> {
        if storage.exists(&MANIFEST).await? {
            return Err(ContainerError::Exists(storage.names(&Entry::ROOT)));
        }
        storage.create_dir(&Entry::fixed(ASSETS)).await?;
        storage.create_dir(&Entry::fixed(REVS)).await?;
        storage
            .write(&GITATTRIBUTES, GITATTRIBUTES_TEMPLATE.as_bytes())
            .await?;
        // Whole-file, so the entries themselves and not just their contents
        // are down: a power cut between here and the first append must not
        // leave a directory that has a manifest in it only sometimes.
        storage.write(&MANIFEST, rows).await?;
        Self::attach(storage, now).await
    }

    /// Open an existing container, taking its lock if nobody holds it.
    ///
    /// # Errors
    /// [`ContainerError::NotAContainer`] when there is no manifest to open, or
    /// the underlying I/O failure.
    pub fn open(storage: S, now: WallTime) -> Result<Self, ContainerError> {
        ready_now(Self::opening(storage, now))
    }

    /// [`Self::open`] for a storage whose futures are not ready when they
    /// are made — which, being a resident one, is where the container is
    /// read into memory.
    ///
    /// # Errors
    /// As [`Self::open`].
    pub async fn opening(storage: S, now: WallTime) -> Result<Self, ContainerError> {
        Self::require_manifest(&storage).await?;
        Self::attach(storage, now).await
    }

    /// Open an existing container to be read and never written, leaving the
    /// lock untouched.
    ///
    /// The honest mode for a reader: claiming the lock merely to look
    /// would either demote the session editing that container or hold a
    /// lock against one nobody is writing, and a handle that will refuse
    /// every write anyway has no business taking either.
    ///
    /// # Errors
    /// As [`Self::open`].
    pub fn reading(storage: S) -> Result<Self, ContainerError> {
        ready_now(Self::opening_to_read(storage))
    }

    /// # Errors
    /// As [`Self::reading`].
    pub async fn opening_to_read(storage: S) -> Result<Self, ContainerError> {
        Self::require_manifest(&storage).await?;
        Ok(Self {
            kept: Self::kept(&storage).await?,
            storage,
            access: Access::ReadOnly(ReadOnlyReason::Reading),
        })
    }

    async fn require_manifest(storage: &S) -> Result<(), ContainerError> {
        if storage.exists(&MANIFEST).await? {
            Ok(())
        } else {
            Err(ContainerError::NotAContainer(storage.names(&Entry::ROOT)))
        }
    }

    async fn attach(storage: S, now: WallTime) -> Result<Self, ContainerError> {
        let access = match storage.claim(now).await? {
            Claim::Held(held) => Access::ReadOnly(ReadOnlyReason::Locked(held)),
            Claim::Taken => Access::Writable,
        };
        Ok(Self {
            kept: Self::kept(&storage).await?,
            storage,
            access,
        })
    }

    /// Read the container in, for a storage that says its reads are not
    /// ready when they are made: the manifest, then every rev and every
    /// payload.
    async fn kept(storage: &S) -> Result<Kept, ContainerError> {
        if let Residency::Lazy = storage.residency() {
            return Ok(Kept::Nothing);
        }
        let mut held = Held::default();
        if let Some(bytes) = storage.read(&MANIFEST).await? {
            held.bytes.insert(MANIFEST, bytes);
        }
        for dir in [REVS, ASSETS] {
            let within = storage.list(&Entry::fixed(dir)).await.unwrap_or_default();
            for name in within {
                let at = Entry::under(dir, &name);
                if let Some(bytes) = storage.read(&at).await? {
                    held.bytes.insert(at, bytes);
                }
            }
        }
        Ok(Kept::Everything(RefCell::new(held)))
    }

    pub fn storage(&self) -> &S {
        &self.storage
    }

    /// What this container is called — the document's own name.
    pub fn name(&self) -> Name {
        self.storage.name()
    }

    pub fn access(&self) -> &Access {
        &self.access
    }

    /// The bytes at `at`, off the shelf when the container is held in
    /// memory and out of the storage when it is not.
    ///
    /// # Errors
    /// The read that did not work — which is not the same as nothing being
    /// there.
    pub fn read(&self, at: &Entry) -> std::io::Result<Option<Vec<u8>>> {
        match &self.kept {
            Kept::Nothing => ready_now(self.storage.read(at)),
            Kept::Everything(held) => Ok(held.borrow().bytes.get(at).cloned()),
        }
    }

    /// # Errors
    /// The read that did not work.
    pub fn exists(&self, at: &Entry) -> std::io::Result<bool> {
        match &self.kept {
            Kept::Nothing => ready_now(self.storage.exists(at)),
            Kept::Everything(held) => Ok(held.borrow().bytes.contains_key(at)),
        }
    }

    /// # Errors
    /// The write that did not land — which a held container's cannot,
    /// since it has only reached memory when this returns.
    pub fn write(&self, at: &Entry, bytes: &[u8]) -> std::io::Result<()> {
        let Kept::Everything(held) = &self.kept else {
            return ready_now(self.storage.write(at, bytes));
        };
        let mut held = held.borrow_mut();
        held.bytes.insert(at.clone(), bytes.to_vec());
        held.owed.push_back((at.clone(), Op::Write(bytes.to_vec())));
        Ok(())
    }

    fn append(&self, at: &Entry, bytes: &[u8]) -> std::io::Result<()> {
        let Kept::Everything(held) = &self.kept else {
            return ready_now(self.storage.append(at, bytes));
        };
        let mut held = held.borrow_mut();
        held.bytes
            .entry(at.clone())
            .or_default()
            .extend_from_slice(bytes);
        held.owed
            .push_back((at.clone(), Op::Append(bytes.to_vec())));
        Ok(())
    }

    fn truncate(&self, at: &Entry, len: u64) -> std::io::Result<()> {
        let Kept::Everything(held) = &self.kept else {
            return ready_now(self.storage.truncate(at, len));
        };
        let mut held = held.borrow_mut();
        if let Some(bytes) = held.bytes.get_mut(at) {
            bytes.truncate(len as usize);
        }
        held.owed.push_back((at.clone(), Op::Truncate(len)));
        Ok(())
    }

    /// How many writes this container has made to its memory and not yet
    /// to its storage. Zero for a container that writes as it goes.
    pub fn pending(&self) -> usize {
        match &self.kept {
            Kept::Nothing => 0,
            Kept::Everything(held) => held.borrow().owed.len(),
        }
    }

    /// Make the writes this container owes, one at a time, in the order it
    /// made them — the payload before the rev, the rev before the row that
    /// names it, as a container that writes as it goes makes them.
    ///
    /// A write that does not land leaves this session's document ahead of
    /// the files, so the container gives up its lock, as a failed append
    /// does where there is no journal. What it had not yet written goes
    /// with the lock: nothing more is written to a container this session
    /// has parted from.
    ///
    /// # Errors
    /// The write that did not land.
    pub async fn drain(&mut self) -> std::io::Result<Drained> {
        let mut performed = 0;
        while let Some((at, op)) = self.next_owed() {
            let done = match &op {
                Op::Write(bytes) => self.storage.write(&at, bytes).await,
                Op::Append(bytes) => self.storage.append(&at, bytes).await,
                Op::Truncate(len) => self.storage.truncate(&at, *len).await,
            };
            if let Err(error) = done {
                self.forget_what_is_owed();
                let echo = std::io::Error::other(error.to_string());
                self.demote(ReadOnlyReason::WriteFailed(error));
                return Err(echo);
            }
            performed += 1;
        }
        Ok(Drained(performed))
    }

    fn next_owed(&self) -> Option<(Entry, Op)> {
        match &self.kept {
            Kept::Nothing => None,
            Kept::Everything(held) => held.borrow_mut().owed.pop_front(),
        }
    }

    fn forget_what_is_owed(&self) {
        if let Kept::Everything(held) = &self.kept {
            held.borrow_mut().owed.clear();
        }
    }

    /// Rename this container — which renames the document, since the
    /// container's name is what the document is called.
    ///
    /// The session carries on through it: every entry is named relative to
    /// the container, so an edit made after the rename lands in the same
    /// manifest the edits before it did, and the lock moves with what it
    /// locks.
    ///
    /// # Errors
    /// [`std::io::ErrorKind::AlreadyExists`] when something already stands
    /// under that name, or the rename's own failure.
    pub fn rename(&mut self, to: &Name) -> std::io::Result<()> {
        ready_now(self.renaming_to(to))
    }

    /// [`Self::rename`] for a storage whose futures are not ready when they
    /// are made.
    ///
    /// # Errors
    /// As [`Self::rename`].
    pub async fn renaming_to(&mut self, to: &Name) -> std::io::Result<()> {
        self.storage.rename(to).await
    }

    /// Give up the right to write, releasing the lock with it. Called when
    /// the manifest turns out not to verify, and when an append fails.
    pub fn demote(&mut self, reason: ReadOnlyReason) {
        self.release();
        self.access = Access::ReadOnly(reason);
    }

    /// Let go of the lock, if this handle was holding it.
    fn release(&mut self) {
        if let Access::Writable = self.access {
            self.storage.release();
        }
    }

    /// # Errors
    /// The underlying read failure.
    pub fn read_manifest(&self) -> std::io::Result<String> {
        let bytes = self.read(&MANIFEST)?.unwrap_or_default();
        String::from_utf8(bytes)
            .map_err(|why| std::io::Error::new(std::io::ErrorKind::InvalidData, why.to_string()))
    }

    /// Where this container's artwork is read and written.
    pub fn assets(&self) -> Artwork<'_, S> {
        Artwork(self)
    }

    /// Where this container keeps the document at every rev.
    pub fn revs(&self) -> Revs<'_, S> {
        Revs(self)
    }

    /// Write the document as it stands at `rev` into `revs/`, its payloads
    /// into `assets/`, and hand back the digest of the rev's bytes.
    ///
    /// Called before the row that names the rev is appended, so the
    /// manifest never names bytes a crash could still take away — and the
    /// payloads land before the rev that references them, for the same
    /// reason one level down.
    ///
    /// # Errors
    /// [`WriteRefusal::ReadOnly`] if this handle does not hold the lock,
    /// or the write that did not land.
    pub fn write_rev(&self, rev: Rev, document: &Document) -> Result<Digest, WriteRefusal> {
        let Access::Writable = &self.access else {
            return Err(WriteRefusal::ReadOnly);
        };
        Ok(revs::write(
            &mut self.revs(),
            rev,
            document,
            &mut self.assets(),
        )?)
    }

    /// The document this container holds at `rev`, payloads and all.
    ///
    /// # Errors
    /// [`revs::RevFault`]: no file, one that is not a document this build
    /// reads, or artwork the container cannot hand back.
    pub fn read_rev(&self, rev: Rev) -> Result<Document, revs::RevFault> {
        let document = revs::read(&self.revs(), rev)?;
        Ok(revs::attached(document, &self.assets())?)
    }

    /// Append one row and make it durable before returning: the line and
    /// its newline in one write, then an fsync. A crash anywhere in here
    /// loses at most this row, and loses it *visibly* — the manifest then
    /// ends without a newline, which is how load knows to drop it.
    ///
    /// # Errors
    /// [`WriteRefusal::ReadOnly`] if this handle does not hold the lock,
    /// or the underlying write failure.
    pub fn append_row(&self, row: &Row) -> Result<(), WriteRefusal> {
        let Access::Writable = &self.access else {
            return Err(WriteRefusal::ReadOnly);
        };
        let mut line = row.canonical_bytes();
        line.push(b'\n');
        Ok(self.append(&MANIFEST, &line)?)
    }

    /// Cut a partial trailing row off, so the next append does not splice
    /// itself onto half a line.
    ///
    /// # Errors
    /// As [`Self::append_row`].
    pub fn truncate_manifest(&self, bytes: u64) -> Result<(), WriteRefusal> {
        let Access::Writable = &self.access else {
            return Err(WriteRefusal::ReadOnly);
        };
        Ok(self.truncate(&MANIFEST, bytes)?)
    }
}

/// What a library shows of a container without opening it: its newest rev,
/// and when that rev was written.
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
pub struct Glance {
    pub rev: Rev,
    pub written: WallTime,
}

impl<S: Storage> Container<S> {
    /// A glance at the container in `storage`, read without touching its
    /// lock. `None` for one no rev has been written into yet.
    ///
    /// # Errors
    /// As [`Self::reading`], or the manifest that could not be read.
    pub async fn glance(storage: S) -> Result<Option<Glance>, ContainerError> {
        let container = Self::opening_to_read(storage).await?;
        let text = container.read_manifest()?;
        let history = manifest::history(&manifest::scan(&text).rows);
        Ok(history.rows.last().map(|row| Glance {
            rev: row.rev,
            written: row.wall_time,
        }))
    }
}

impl<S: Storage> Drop for Container<S> {
    /// The lock's lifetime is the writable handle's lifetime, rather than
    /// something a caller has to remember.
    fn drop(&mut self) {
        self.release();
    }
}

/// A container's rev files, as the shelf a rev is read off and written to.
pub struct Revs<'a, S: Storage>(&'a Container<S>);

impl<S: Storage> revs::Backing for Revs<'_, S> {
    fn put(&mut self, at: Rev, bytes: &[u8]) -> std::io::Result<()> {
        self.0.write(&revs::entry(at), bytes)
    }

    fn get(&self, at: Rev) -> std::io::Result<Option<Vec<u8>>> {
        self.0.read(&revs::entry(at))
    }

    fn names(&self, at: Rev) -> String {
        self.0.storage.names(&revs::entry(at))
    }
}

/// A container's `assets/` directory.
pub struct Artwork<'a, S: Storage>(&'a Container<S>);

/// Where a payload of `hash` is filed, given the format that reads it.
pub fn asset_entry(hash: AssetHash, format: AssetFormat) -> Entry {
    Entry::under(ASSETS, &format.file_name(hash))
}

impl<S: Storage> AssetSource for Artwork<'_, S> {
    fn names(&self, hash: AssetHash) -> String {
        self.0
            .storage
            .names(&Entry::under(ASSETS, &hash.to_string()))
    }

    fn find(&self, hash: AssetHash) -> std::io::Result<(AssetFormat, Vec<u8>)> {
        for format in AssetFormat::ALL {
            if let Some(bytes) = self.0.read(&asset_entry(hash, format))? {
                return Ok((format, bytes));
            }
        }
        Err(std::io::Error::from(std::io::ErrorKind::NotFound))
    }
}

impl<S: Storage> AssetSink for Artwork<'_, S> {
    /// Create-only: a file already under a content hash's name already
    /// holds those bytes, so there is nothing a second write could say.
    fn put(&mut self, hash: AssetHash, asset: &Asset) -> std::io::Result<()> {
        let at = asset_entry(hash, AssetFormat::of(asset));
        if self.0.exists(&at)? {
            return Ok(());
        }
        self.0.write(&at, asset.bytes())
    }
}

#[derive(Debug, thiserror::Error)]
pub enum WriteRefusal {
    #[error("this diagram is open read-only")]
    ReadOnly,
    #[error(transparent)]
    Io(#[from] std::io::Error),
}

/// What became of a container offered for [`discard_pristine`].
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
pub enum Discarded {
    Removed,
    /// Something is in it, so it stays: a row in the manifest, artwork, or a
    /// file this layout never writes.
    Kept,
}

/// The files [`Container::create`] lays down. Anything else in a container
/// belongs to whoever put it there.
const LAID_DOWN: [Entry; 3] = [MANIFEST, LOCK, GITATTRIBUTES];

/// The directories it lays down. Each must be empty for the container to
/// count as pristine.
const LAID_DOWN_DIRS: [&str; 2] = [ASSETS, REVS];

/// Remove a container nothing was ever written into, so a launch and a quit
/// leave the documents directory as they found it.
///
/// Deliberately timid: an empty manifest is necessary but not sufficient, and
/// only a container holding exactly what [`Container::create`] laid down is
/// removed. Anything else — a row, an asset, a file the user put beside
/// the manifest — and the container stays, because deleting something we did
/// not write every part of is not ours to decide. The caller narrows this
/// further: only a container *this session created* is ever offered.
///
/// # Errors
/// The read or the removal that did not work.
pub async fn discard_pristine<S: Storage>(storage: &S) -> std::io::Result<Discarded> {
    let manifest = storage
        .read(&MANIFEST)
        .await?
        .ok_or_else(|| std::io::Error::from(std::io::ErrorKind::NotFound))?;
    if !manifest.is_empty() {
        return Ok(Discarded::Kept);
    }
    let mut ours = Vec::new();
    for name in storage.list(&Entry::ROOT).await? {
        if let Some(dir) = LAID_DOWN_DIRS.into_iter().find(|dir| *dir == name) {
            // A directory that cannot be listed is not one we emptied, so
            // it counts as something in the container rather than as a
            // reason to fail.
            if !matches!(storage.list(&Entry::fixed(dir)).await.as_deref(), Ok([])) {
                return Ok(Discarded::Kept);
            }
            ours.push(Entry::fixed(dir));
        } else if let Some(at) = LAID_DOWN.into_iter().find(|at| at.as_str() == name) {
            ours.push(at);
        } else {
            return Ok(Discarded::Kept);
        }
    }
    for at in ours {
        storage.remove(&at).await?;
    }
    storage.discard().await?;
    Ok(Discarded::Removed)
}

/// The same cleanup, narrowed by the lock instead of by a caller's memory of
/// what it made: a container nobody holds and nothing was ever written into.
///
/// This is how a host with no exit to sweep on asks the question — a browser
/// tab that is closed runs nothing, so the containers a session left behind
/// are judged the next time one starts, and the lock is what tells a newborn
/// nobody kept from one another view has open.
///
/// # Errors
/// The claim, the read or the removal that did not work.
pub async fn discard_unclaimed<S: Storage>(
    storage: &S,
    now: WallTime,
) -> std::io::Result<Discarded> {
    if !matches!(storage.claim(now).await?, Claim::Taken) {
        return Ok(Discarded::Kept);
    }
    let swept = discard_pristine(storage).await;
    // A removed container took its lock with it; there is something to give
    // back only where it stayed.
    if !matches!(swept, Ok(Discarded::Removed)) {
        storage.release();
    }
    swept
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::fixture::{self, block_on};
    use crate::storage::{Memory, Native};

    fn native(dir: &crate::temp::TempDir, name: &str) -> Native {
        Native::at(dir.join(name))
    }

    #[test]
    fn a_created_container_holds_the_whole_template() {
        let dir = fixture::dir("container-layout");
        let root = dir.join("doc.bwx");
        let container = Container::create(Native::at(&root), WallTime::EPOCH)
            .expect("the container is laid out");

        assert!(matches!(container.access(), Access::Writable));
        assert_eq!(container.read_manifest().expect("an empty manifest"), "");
        assert!(root.join(ASSETS).is_dir());
        assert!(root.join(REVS).is_dir(), "a new container has no revs/");
        assert!(
            root.join(LOCK.as_str()).is_file(),
            "a writable handle holds the lock"
        );
        let attributes =
            std::fs::read_to_string(root.join(GITATTRIBUTES.as_str())).expect("the git template");
        assert!(attributes.contains("manifest.jsonl merge=conflict-always"));
        assert!(attributes.contains("revs/** binary"));
        assert!(
            !attributes.contains("document.json"),
            "the template still names a file this layout does not write",
        );
    }

    #[test]
    fn creating_over_an_existing_container_is_refused() {
        let dir = fixture::dir("container-exists");
        let root = dir.join("doc.bwx");
        let first =
            Container::create(Native::at(&root), WallTime::EPOCH).expect("the first is laid out");
        drop(first);

        assert!(matches!(
            Container::create(Native::at(&root), WallTime::EPOCH),
            Err(ContainerError::Exists(_)),
        ));
        assert!(matches!(
            Container::open(native(&dir, "nothing-here"), WallTime::EPOCH),
            Err(ContainerError::NotAContainer(_)),
        ));
    }

    /// The lock is the writable handle's lifetime, so closing a container
    /// hands the next one the right to write.
    #[test]
    fn dropping_a_container_releases_its_lock() {
        let dir = fixture::dir("container-lock-lifetime");
        let root = dir.join("doc.bwx");
        let container =
            Container::create(Native::at(&root), WallTime::EPOCH).expect("the container");
        assert!(matches!(
            Container::open(Native::at(&root), WallTime::EPOCH)
                .expect("a second handle opens")
                .access(),
            Access::ReadOnly(ReadOnlyReason::Locked(_)),
        ));

        drop(container);
        assert!(matches!(
            Container::open(Native::at(&root), WallTime::EPOCH)
                .expect("a third handle opens")
                .access(),
            Access::Writable,
        ));
    }

    /// The property renaming a document rests on: the session keeps the
    /// container through the move, so an append after the rename lands in
    /// the renamed manifest — and the lock, which sits inside what moved,
    /// is released under its new name.
    #[test]
    fn a_renamed_container_keeps_its_lock_and_its_open_manifest() {
        let dir = fixture::dir("container-rename");
        let root = dir.join("doc.bwx");
        let renamed = dir.join("engine.bwx");
        let mut container =
            Container::create(Native::at(&root), WallTime::EPOCH).expect("the container");
        let row = fixture::rows(1).pop().expect("one row");
        container.append_row(&row).expect("the first row lands");

        container
            .rename(&Name::of_document("engine").expect("a name"))
            .expect("the container moves");

        assert_eq!(container.name().to_string(), "engine.bwx");
        assert!(!root.exists(), "the old name still stands");
        assert!(
            renamed.join(LOCK.as_str()).is_file(),
            "the lock did not travel with the container",
        );
        container
            .append_row(&row)
            .expect("an append after the rename");
        assert_eq!(
            std::fs::read_to_string(renamed.join(MANIFEST.as_str()))
                .expect("the renamed manifest")
                .lines()
                .count(),
            2,
            "the session kept appending to the file it had open",
        );
        assert!(
            matches!(
                Container::open(Native::at(&renamed), WallTime::EPOCH)
                    .expect("a second handle")
                    .access(),
                Access::ReadOnly(ReadOnlyReason::Locked(_)),
            ),
            "the lock was dropped somewhere in the move",
        );

        drop(container);
        assert!(
            !renamed.join(LOCK.as_str()).exists(),
            "releasing the lock removed the wrong path",
        );
    }

    #[test]
    fn a_rename_onto_something_that_exists_is_refused() {
        let dir = fixture::dir("container-rename-onto");
        let root = dir.join("doc.bwx");
        let taken = dir.join("taken.bwx");
        let mut container =
            Container::create(Native::at(&root), WallTime::EPOCH).expect("the container");
        Container::create(Native::at(&taken), WallTime::EPOCH).expect("the container in the way");

        let refusal = container
            .rename(&Name::of_document("taken").expect("a name"))
            .expect_err("renaming over a container must be refused");
        assert_eq!(refusal.kind(), std::io::ErrorKind::AlreadyExists);
        assert_eq!(
            container.name().to_string(),
            "doc.bwx",
            "the store moved anyway"
        );
        assert!(
            taken.join(MANIFEST.as_str()).is_file(),
            "the container in the way is gone"
        );
    }

    /// The cleanup, and everything it refuses to do: only a container holding
    /// exactly what `create` laid down, with an empty log, is removed.
    #[test]
    fn only_a_container_with_nothing_in_it_is_discarded() {
        let dir = fixture::dir("container-discard");

        let empty = dir.join("empty.bwx");
        drop(Container::create(Native::at(&empty), WallTime::EPOCH).expect("the empty container"));
        assert_eq!(
            block_on(discard_pristine(&Native::at(&empty))).expect("the sweep answers"),
            Discarded::Removed,
        );
        assert!(!empty.exists(), "a pristine container was left behind");

        let written = dir.join("written.bwx");
        let container =
            Container::create(Native::at(&written), WallTime::EPOCH).expect("the container");
        let row = fixture::rows(1).pop().expect("one row");
        container.append_row(&row).expect("the row lands");
        drop(container);
        assert_eq!(
            block_on(discard_pristine(&Native::at(&written))).expect("the sweep answers"),
            Discarded::Kept,
            "a container with a row in it was removed",
        );
        assert!(written.join(MANIFEST.as_str()).is_file());

        let theirs = dir.join("theirs.bwx");
        drop(Container::create(Native::at(&theirs), WallTime::EPOCH).expect("the container"));
        std::fs::write(theirs.join("notes.txt"), b"mine").expect("a file of the user's own");
        assert_eq!(
            block_on(discard_pristine(&Native::at(&theirs))).expect("the sweep answers"),
            Discarded::Kept,
            "something we did not write was removed",
        );
        assert!(
            theirs.join(MANIFEST.as_str()).is_file(),
            "and it was half-removed"
        );
    }

    /// What the sweep asks where there is no exit to run it on: the lock
    /// stands in for the caller's memory of what it made, so a container
    /// another view has open survives a sweep that would otherwise take it.
    #[test]
    fn a_held_container_survives_the_unclaimed_sweep() {
        let held = Memory::deferred("held.bwx");
        drop(block_on(Container::creating(held.clone(), WallTime::EPOCH)).expect("the container"));
        let claimed = block_on(Storage::claim(&held, WallTime::EPOCH)).expect("the claim");
        assert!(
            matches!(claimed, Claim::Taken),
            "precondition: the lock is ours to hold"
        );

        assert_eq!(
            block_on(discard_unclaimed(&held, WallTime::EPOCH)).expect("the sweep answers"),
            Discarded::Kept,
            "a container someone holds was swept",
        );
        assert!(
            block_on(held.exists(&MANIFEST)).expect("the read"),
            "and it was half-removed",
        );

        held.release();
        assert_eq!(
            block_on(discard_unclaimed(&held, WallTime::EPOCH)).expect("the sweep answers"),
            Discarded::Removed,
            "a container nobody holds and nobody wrote to was left behind",
        );
        assert!(
            !block_on(held.exists(&MANIFEST)).expect("the read"),
            "and it was half-removed",
        );
    }

    #[test]
    fn a_read_only_handle_refuses_to_write() {
        let dir = fixture::dir("container-read-only");
        let root = dir.join("doc.bwx");
        let _held = Container::create(Native::at(&root), WallTime::EPOCH).expect("the container");
        let second = Container::open(Native::at(&root), WallTime::EPOCH).expect("a second handle");

        let row = fixture::rows(1).pop().expect("one row");
        assert!(matches!(
            second.append_row(&row),
            Err(WriteRefusal::ReadOnly)
        ));
        assert!(matches!(
            second.truncate_manifest(0),
            Err(WriteRefusal::ReadOnly)
        ));
    }

    /// A container read whole at open answers out of memory, and every
    /// write keeps that memory in step — the property the browser's
    /// synchronous read path rests on.
    #[test]
    fn a_resident_container_reads_back_what_it_was_written() {
        let document = fixture::documents(1).pop().expect("one document");
        let held = Container::create(Memory::resident("doc.bwx"), WallTime::EPOCH)
            .expect("the container is laid out");
        let at = blockworx_doc::fixtures::rev(1);
        held.write_rev(at, &document).expect("the rev lands");

        assert!(
            matches!(held.kept, Kept::Everything(_)),
            "precondition: this container is held in memory",
        );
        assert_eq!(held.read_rev(at).expect("it reads back"), document);
        let row = fixture::rows(1).pop().expect("one row");
        held.append_row(&row).expect("the row lands");
        assert_eq!(
            held.read_manifest().expect("the manifest").lines().count(),
            1,
        );
    }
}