coven_database/

file_database.rs

//! File metadata operations through the owning database, never through app SQL.

use super::{Database, DatabaseAccess, DatabaseInner, Mode, Writer};
use crate::internal_schema::CACHE_NAME_USED_SQL;
use crate::DatabaseChanges;
use crate::{
    file_queue, file_ref, sqlite::DatabaseConnection, write_schema::WriteSchema, DbError, FileRef,
    FileUpload,
};
use coven_crypto::SecretBytes;
use coven_foundation::task::CarriesBug;
use coven_foundation::{
    files::{FileName, StoreDir},
    id_source::{DeviceId, FileId, ObjectId},
};
use std::sync::Arc;

#[path = "file_cache.rs"]
pub(crate) mod cache;
pub(crate) use cache::{
    clear_entries as clear_cached, declare_budgets, kept as cache_kept, object_path,
    object_path_of, publish as publish_cached, remove_entries as remove_cached,
    set_budget as set_cache_budget, trim_entries as trim_cached,
};

/// File queue and cache access, on the store's writer or a read-only
/// handle (`FileDatabase<ReadOnly>`), which reads and caches files and can
/// change only its local cache metadata. The upload and pin queue and the
/// file records are the writer's. Construct beside the database at the
/// application composition root.
///
/// ```compile_fail
/// async fn completes_nothing(files: &coven_database::FileDatabase<coven_database::ReadOnly>) {
///     files.complete_upload(1, ()).await;
/// }
/// ```
/// ```compile_fail
/// fn queues_nothing(files: &coven_database::FileDatabase<coven_database::ReadOnly>) {
///     files.queued_work_changes();
/// }
/// ```
/// ```compile_fail
/// async fn records_nothing(files: &coven_database::FileDatabase<coven_database::ReadOnly>) {
///     files.fail_upload(1, (), None).await;
/// }
/// ```
pub struct FileDatabase<M: Mode = Writer> {
    database: Database<M>,
}

// Written out: `derive(Clone)` would ask the mode to be `Clone`.
impl<M: Mode> Clone for FileDatabase<M> {
    fn clone(&self) -> Self {
        Self { database: self.database.clone() }
    }
}

impl<M: Mode> FileDatabase<M> {
    /// Use an open store's handle for its file operations.
    pub fn new(database: Database<M>) -> Self {
        Self { database }
    }

    /// Where a bug in the file owner's work that nobody waits for goes, as
    /// its database's does (E15).
    pub fn bugs(&self) -> &coven_foundation::task::BugSinkRef {
        self.database.bugs()
    }

    /// The writer's file owner, which keeps file records and the upload and
    /// pin queue; a read-only handle keeps none.
    pub fn records(&self) -> Option<&FileDatabase<Writer>> {
        M::records(self)
    }

    /// Run `run` on the handle's writable connection: the store's writer,
    /// or a read-only handle's cache connection.
    async fn run<T: Send + 'static, E: From<DbError> + CarriesBug + Send + 'static>(
        &self,
        run: impl FnOnce(&mut DatabaseConnection, &WriteSchema, &StoreDir, DeviceId) -> Result<T, E>
            + Send
            + 'static,
    ) -> Result<T, E> {
        self.database
            .call(move |open| {
                let access = M::access(open);
                access.with_writer(|writer| {
                    run(writer, &access.write_schema, &access.directory, access.device)
                })
            })
            .await
    }

    /// The file this device has a record of under this id.
    pub async fn file_ref(&self, file: FileId) -> Result<FileRef, DbError> {
        self.database.file_ref(file).await
    }
    /// Open and check the local source of a file still on this device only,
    /// while it uploads. `None` means its bytes come from the cache or
    /// storage.
    pub async fn open_local(
        &self,
        file: &FileRef,
    ) -> Result<Option<crate::LocalFileStream>, crate::LocalFileError> {
        self.database.open_local_file(file).await
    }

    /// Cached encrypted chunk or header (-1). Its bytes are read with no
    /// writer held (§21.2); an entry evicted meanwhile reads as not cached.
    pub async fn cached(&self, file: &FileRef, index: i64) -> Result<Option<Vec<u8>>, DbError> {
        let file = file.clone();
        self.database.call(move |open| cache::read(M::access(open), &file, index)).await
    }
    /// Write the stored bytes of chunk `index`, or the header (-1), back
    /// over the cached copy of them that failed its check (§17.8). The
    /// bytes are written with no writer held (§21.2).
    pub async fn repair_cached(
        &self,
        file: &FileRef,
        index: i64,
        bytes: Vec<u8>,
    ) -> Result<(), DbError> {
        let file = file.clone();
        self.database.call(move |open| cache::repair(M::access(open), &file, index, &bytes)).await
    }
    /// Keep one verified range of `chunks` consecutive encrypted chunks from
    /// `first`, or the header (`first` = -1), durably before publishing its
    /// cache metadata in one transaction. The bytes are written with no
    /// writer held (§21.2), on a blocking worker that finishes even if the
    /// caller stops waiting.
    pub async fn cache_range(
        &self,
        file: &FileRef,
        first: i64,
        chunks: i64,
        name: FileName,
        bytes: Vec<u8>,
    ) -> Result<(), DbError> {
        let file = file.clone();
        self.database
            .call(move |open| cache::put(M::access(open), &file, first, chunks, &name, &bytes))
            .await
    }
    /// Remove the bytes of cache entries the writer just chose, once it is
    /// free (§21.2, Appendix C.3). Returns the failures, whose records stay for the
    /// next attempt.
    async fn remove_released(
        &self,
        chosen: crate::file_removals::Chosen,
    ) -> Result<Vec<DbError>, DbError> {
        self.database
            .call(move |open| Ok::<_, DbError>(M::access(open).remove_released(chosen)))
            .await
    }
    async fn removed(&self, chosen: crate::file_removals::Chosen) -> Result<(), DbError> {
        let failures = self.remove_released(chosen).await?;
        crate::write_failure::after_cleanup(Ok(()), failures)
    }
    /// Evict unpinned chunks in least-recently-read order, independently per namespace.
    pub async fn trim_cache(&self, namespace: &str) -> Result<(), DbError> {
        let namespace = namespace.to_owned();
        let chosen = self
            .run(move |db, _, _, _| {
                let paths = cache::trim(db, &namespace)?;
                Ok::<_, DbError>(crate::file_removals::choose_cache(&paths))
            })
            .await?;
        self.removed(chosen).await
    }
    /// Plaintext bytes absent from this namespace's cache; headers and tags
    /// are excluded from application download progress. A file only on this
    /// device is wholly local.
    #[cfg(any(test, feature = "test-utils"))]
    pub async fn missing_bytes(&self, file: &FileRef) -> Result<u64, DbError> {
        let file = file.clone();
        self.run(move |db, _, _, device| {
            Ok(file.size() - crate::cache_progress::covered(db, &file, device)?)
        })
        .await
    }
    /// The namespace's budget, declared or since set.
    pub async fn budget(&self, namespace: &str) -> Result<u64, DbError> {
        let namespace = namespace.to_owned();
        self.run(move |db, _, _, _| cache::budget(db, &namespace)).await
    }
}

impl FileDatabase {
    /// The files this process is sending now, which the upload worker marks
    /// as each attempt starts and ends (§17.9).
    pub fn uploads_in_flight(&self) -> crate::InFlight {
        self.database.uploads_in_flight()
    }

    /// The files this process is downloading to keep whole now, which each
    /// pin's download marks as it starts and ends (§17.8).
    pub fn downloads_in_flight(&self) -> crate::InFlight {
        self.database.downloads_in_flight()
    }

    /// A read-only query on one committed state, on the read pool, so
    /// read-only decisions never take the writer (§21.2).
    async fn read<T: Send + 'static, E: From<DbError> + CarriesBug + Send + 'static>(
        &self,
        read: impl FnOnce(&DatabaseConnection) -> Result<T, E> + Send + 'static,
    ) -> Result<T, E> {
        self.database.call(move |inner| inner.access.with_committed(read)).await
    }
    /// The upload of a file this device uploads, its move, this device's
    /// record about the upload's bytes and its move's record, read in one
    /// committed state: the local facts its status is judged with (§17.6).
    /// `None` when this device doesn't upload the file.
    pub async fn local_evidence<P: crate::PendingValue>(
        &self,
        file: &FileRef,
        upload_subject: impl FnOnce(&crate::FileLocation) -> P::Subject + Send + 'static,
        move_subject: impl FnOnce(crate::MoveId) -> P::Subject + Send + 'static,
    ) -> Result<Option<LocalEvidence<P>>, DbError>
    where
        P::Subject: Send,
    {
        let file = file.clone();
        self.run(move |db, _, _, device| {
            db.read_transaction(|| {
                if file.place() != (crate::FilePlace::DeviceOnly { device }) {
                    return Ok(None);
                }
                let Some(upload) = file_queue::upload_of(db, file.id())? else {
                    return Ok(None);
                };
                let (location, of) = db.query_row(
                    "SELECT location,move FROM _coven_file_uploads WHERE id=?1",
                    [upload],
                    |r| Ok((r.get::<_, String>(0)?, r.get::<_, crate::MoveId>(1)?)),
                )?;
                let location = crate::FileLocation::decode(&location)?;
                let record = crate::pending::find::<P>(db, &upload_subject(&location))?;
                let moving = crate::pending::find::<P>(db, &move_subject(of))?;
                let original = crate::user_file::read(db, file.id())?.map(|original| original.path);
                Ok(Some(LocalEvidence { upload, location, of, original, record, moving }))
            })
        })
        .await
    }
    /// Fail a move because a user's original changed before its upload
    /// (§17.3): none of its files switches, each stays on this device only,
    /// and its uploads retire, their stored bytes becoming deletion
    /// candidates. `record` is the move's own and says which original. A
    /// move that already ended keeps its state.
    pub async fn fail_move<P: crate::PendingValue>(
        &self,
        upload: i64,
        record: P,
    ) -> Result<(), DbError> {
        self.run(move |db, _, _, _| {
            db.state_transaction(|db| {
                let Some(of) = record.move_of() else {
                    return Err(DbError::Invariant {
                        what: "a failed move's record names the move",
                    });
                };
                if crate::file_moves::of_upload(db, upload)? != of {
                    return Err(DbError::Invariant {
                        what: "a move fails through one of its own uploads",
                    });
                }
                if crate::file_moves::state(db, of)? != Some(crate::file_moves::MoveState::Moving) {
                    return Ok(());
                }
                crate::file_moves::end_unswitched(db, of, crate::file_moves::MoveState::Failed)?;
                crate::pending::save(db, &record)
            })
        })
        .await
    }
    /// Clear the record `subject` names if `checked` says the read or upload
    /// of `upload`'s original that just succeeded settles it, by the record
    /// and the original's path. The predicate and deletion share the
    /// pending transaction, which also ends the upload's retry wait the
    /// failure armed. Returns the original's path when it cleared.
    pub async fn clear_original<P: crate::PendingValue>(
        &self,
        upload: i64,
        subject: P::Subject,
        checked: impl Fn(&P, &std::path::Path) -> bool + Send + 'static,
    ) -> Result<Option<std::path::PathBuf>, DbError>
    where
        P::Subject: Send + 'static,
    {
        self.run(move |db, _, _, _| {
            db.state_transaction(|db| {
                let original: Option<Option<Vec<u8>>> = db
                    .query("SELECT original FROM _coven_file_uploads WHERE id=?1", [upload], |r| {
                        r.get(0)
                    })?
                    .pop();
                let Some(Some(original)) = original else {
                    return Ok(None);
                };
                let path = crate::user_file::decode_path(original)?;
                if crate::pending::find::<P>(db, &subject)?
                    .is_some_and(|record| checked(&record, &path))
                {
                    crate::pending::clear(db, &subject)?;
                    crate::retry::clear(db, crate::RetrySubject::Upload(upload))?;
                    return Ok(Some(path));
                }
                Ok(None)
            })
        })
        .await
    }
    /// Retire the uploads no file waits for any more, except those whose
    /// objects are in `in_flight`, whose attempts retire them as they end
    /// (§17.9). Each retired upload's pending record goes with it, and its
    /// stored path becomes a deletion candidate unless something else
    /// protects it (§17.10). Returns the retired queue ids.
    pub async fn retire_unlinked<S: serde::Serialize + Send + 'static>(
        &self,
        in_flight: std::collections::BTreeSet<ObjectId>,
        subject: impl Fn(&crate::FileLocation) -> S + Send + 'static,
    ) -> Result<Vec<i64>, DbError> {
        // An idle run writes nothing: only work found takes the writer.
        let database = &self.database;
        let work = database
            .call(move |inner| {
                inner
                    .access
                    .with_committed(|db| Ok::<_, DbError>(!file_queue::retirable(db)?.is_empty()))
            })
            .await?;
        if !work {
            return Ok(Vec::new());
        }
        database
            .call(move |inner| {
                // The released copies go once the writer is free (§21.2). A
                // move whose last waiting upload retires switches its files
                // in this write (§17.3).
                inner.with_files("retire uploads", |writer, files| {
                    writer.local_write(
                        &inner.access.write_schema,
                        inner.access.device,
                        inner.clock.now(),
                        files,
                        |files| files.retire_unlinked(&in_flight, subject),
                    )
                })
            })
            .await
    }
    /// Record this device's first unmet condition for the stored bytes a
    /// pin downloads, replacing an earlier reason. It keeps no wait: only the
    /// app's next pin or read tries again (§17.7).
    pub async fn fail_download<P: crate::PendingValue>(&self, record: P) -> Result<(), DbError> {
        self.database.record_pending(record).await
    }
    /// Clear only conditions the completed download actually checked. The
    /// predicate and deletion share the pending transaction.
    pub async fn clear_download_pending<P: crate::PendingValue>(
        &self,
        subject: P::Subject,
        checked: impl Fn(&P) -> bool + Send + 'static,
    ) -> Result<(), DbError>
    where
        P::Subject: Send + 'static,
    {
        self.database.clear_checked::<P>(subject, checked).await
    }
    /// Clear only conditions the completed local read actually checked. The
    /// predicate and deletion share the pending transaction. Any queued
    /// upload wait ends with it; return the upload subject so the caller can
    /// discard its in-memory timer.
    pub async fn clear_upload_pending<P: crate::PendingValue>(
        &self,
        object: (DeviceId, ObjectId),
        subject: P::Subject,
        checked: impl Fn(&P) -> bool + Send + 'static,
    ) -> Result<Option<crate::RetrySubject>, DbError>
    where
        P::Subject: Send + 'static,
    {
        self.database.clear_pending_if::<P>(object, subject, checked).await
    }
    /// Record a local source failure only while its upload remains queued.
    /// The retry is selected from the current upload wait and committed with
    /// the record, so concurrent upload completion cannot leave a stale failure.
    /// A `next` that fails records nothing and returns its own error; only a
    /// failed save fails the handle, as any lost pending state does (E5).
    pub async fn fail_local_read<P, E>(
        &self,
        file: FileId,
        record: P,
        next: impl FnOnce(Option<crate::RetryWait>) -> Result<crate::RetryWait, E> + Send + 'static,
    ) -> Result<Option<(crate::RetrySubject, crate::RetryWait)>, E>
    where
        P: crate::PendingValue,
        E: From<DbError> + CarriesBug + Send + 'static,
    {
        self.run(move |db, _, _, _| {
            db.state_transaction(|db| {
                let Some(id) = file_queue::upload_of(db, file)? else {
                    return Ok(Ok(None));
                };
                let subject = crate::RetrySubject::Upload(id);
                // Nothing is written yet: a refused wait leaves the
                // transaction with nothing to commit.
                let wait = match next(crate::retry::find(db, subject)?) {
                    Ok(wait) => wait,
                    Err(refused) => return Ok(Err(refused)),
                };
                crate::pending::save(db, &record)?;
                crate::retry::save(db, subject, &wait)?;
                Ok(Ok(Some((subject, wait))))
            })
        })
        .await?
    }

    /// Durable queue: each upload some file still waits for.
    pub async fn uploads(&self) -> Result<Vec<FileUpload>, DbError> {
        self.read(file_queue::read).await
    }
    /// Every queued upload, those no file waits for included.
    #[cfg(any(test, feature = "test-utils"))]
    pub async fn all_uploads(&self) -> Result<Vec<FileUpload>, DbError> {
        self.read(file_queue::read_all).await
    }
    /// The queue ids of the bound uploads, which the pause leaves running
    /// (§17.4).
    pub async fn bound_uploads(&self) -> Result<std::collections::BTreeSet<i64>, DbError> {
        self.read(file_queue::bound).await
    }
    /// Whether the app paused uploads, as persisted (E7).
    pub async fn uploads_paused(&self) -> Result<bool, DbError> {
        self.read(file_queue::paused).await
    }
    /// Whether a self-replacement pauses this device's writes, which holds
    /// a trailing upload's completion until it ends (§12.3, §17.4).
    pub async fn writes_paused(&self) -> Result<bool, DbError> {
        self.read(crate::operation::writes_paused).await
    }
    /// Whether the app moves files only on unmetered networks, as persisted
    /// (E7).
    pub async fn unmetered_only(&self) -> Result<bool, DbError> {
        self.read(file_queue::unmetered_only).await
    }
    /// Each queued upload with this device's pending record about its file,
    /// read in one committed state: the reason an earlier attempt left decides
    /// whether automatic work retries it (§17.9).
    pub async fn upload_work<P: crate::PendingValue>(
        &self,
        subject: impl Fn(&crate::FileLocation) -> P::Subject + Send + 'static,
    ) -> Result<Vec<(FileUpload, Option<P>)>, DbError> {
        self.run(move |db, _, _, _| {
            db.read_transaction(|| {
                file_queue::read(db)?
                    .into_iter()
                    // A stored upload only waits for the rest of its move.
                    .filter(|upload| !upload.stored)
                    .map(|upload| {
                        let record = crate::pending::find::<P>(db, &subject(&upload.location))?;
                        Ok((upload, record))
                    })
                    .collect()
            })
        })
        .await
    }
    /// Each upload no file waits for any more that still records a provider
    /// session, with this device's pending record about its file, read in one
    /// committed state: its session is aborted before it retires (§17.9).
    pub async fn abandoned_uploads<P: crate::PendingValue>(
        &self,
        subject: impl Fn(&crate::FileLocation) -> P::Subject + Send + 'static,
    ) -> Result<Vec<(FileUpload, Option<P>)>, DbError> {
        self.run(move |db, _, _, _| {
            db.read_transaction(|| {
                file_queue::abandoned(db)?
                    .into_iter()
                    .map(|upload| {
                        let record = crate::pending::find::<P>(db, &subject(&upload.location))?;
                        Ok((upload, record))
                    })
                    .collect()
            })
        })
        .await
    }
    /// Forget the session of upload `id` once storage has aborted it, if no
    /// file waits for the upload: it then retires with the others (§17.9).
    /// One a file waits for again keeps its record, and its next attempt
    /// finds the session gone and starts another.
    pub async fn forget_session(&self, id: i64) -> Result<(), DbError> {
        self.run(move |db, _, _, _| db.transaction(|db| crate::file_queue::forget_session(db, id)))
            .await
    }
    /// Record a failed attempt as this device's pending record about its file,
    /// with its retry wait or none when it waits for a prerequisite or the app,
    /// in one transaction (§17.9, §19.1).
    pub async fn fail_upload<P: crate::PendingValue>(
        &self,
        id: i64,
        record: P,
        retry: Option<crate::RetryWait>,
    ) -> Result<(), DbError> {
        self.run(move |db, _, _, _| {
            db.transaction(|db| {
                if !db.query_row(
                    "SELECT EXISTS(SELECT 1 FROM _coven_file_uploads WHERE id=?1)",
                    [id],
                    |r| r.get::<_, bool>(0),
                )? {
                    return Err(DbError::Invariant {
                        what: "the upload recording progress is queued",
                    });
                }
                crate::retry::save_failure(
                    db,
                    &record,
                    crate::RetrySubject::Upload(id),
                    retry.as_ref(),
                )
            })
        })
        .await
    }
    /// Read the first-read plaintext hash of one queued chunk, keyed by the
    /// upload's location and owned by the queue whatever its files became.
    pub async fn upload_chunk_hash(
        &self,
        location: &crate::FileLocation,
        index: u64,
    ) -> Result<coven_crypto::ContentHash, DbError> {
        let location = location.clone();
        self.read(move |db| file_queue::chunk_hash(db, &location, index)).await
    }
    /// Open the source an upload sends, from the facts its queue entry fixed
    /// at staging; later row changes cannot replace them. `None` means the
    /// upload is no longer queued. It is opened with no writer held.
    pub async fn open_upload_source(
        &self,
        upload: &FileUpload,
    ) -> Result<Option<crate::LocalFileStream>, crate::LocalFileError> {
        let (id, object, size) = (upload.id, upload.location.object(), upload.size);
        self.database.call(move |inner| open_upload_source(&inner.access, id, object, size)).await
    }

    /// Persist the provider recording after each confirmed change in progress,
    /// with the bytes it records as confirmed, which `coven_uploads` gives the
    /// app (E4). Storage alone reads the recording, so the caller gives both
    /// from the same session.
    #[expect(
        clippy::map_err_ignore,
        reason = "a byte count past SQLite's integers breaks the named invariant"
    )]
    pub async fn record_session(
        &self,
        id: i64,
        session: SecretBytes,
        progress: crate::UploadProgress,
    ) -> Result<(), DbError> {
        let count = |bytes: u64| {
            i64::try_from(bytes)
                .map_err(|_| DbError::Invariant { what: "an upload's size fits SQLite" })
        };
        let confirmed = count(progress.bytes_confirmed)?;
        let total = count(progress.bytes_total)?;
        self.run(move |db, _, _, _| {
            db.transaction(|db| {
                changed(crate::file_queue::record_session(
                    db,
                    id,
                    session.as_bytes(),
                    confirmed,
                    total,
                )?)
            })
        })
        .await
    }
    /// Complete an upload once storage confirms its complete bytes (§17.9):
    /// one local transaction switches each file the upload stores to the
    /// cloud, as this device's write, and removes the queue row, its chunk
    /// hashes, the copy kept for it and this device's pending record about
    /// it. A file gone meanwhile switches nothing. Afterwards this device
    /// reads the files from the cache or storage. A write that can't be made
    /// now fails the transaction, and the upload stays queued. An upload no
    /// file waits for any more, of a move that failed or was cancelled while
    /// it was sent, retires instead, storing no file.
    pub async fn complete_upload<S: serde::Serialize + Send + 'static>(
        &self,
        id: i64,
        subject: S,
    ) -> Result<crate::UploadCompletion, DbError> {
        let subject = crate::pending::SubjectKey::of(&subject)?;
        self.database.call(move |inner| complete(inner, id, subject)).await
    }
    /// Reserve a whole cache file of `file` while it is written and checked,
    /// for a pin; a read-only handle caches individual chunks only.
    pub async fn reserve_cache_file(
        &self,
        file: &FileRef,
        name: FileName,
    ) -> Result<CacheReservation, DbError> {
        let database = &self.database;
        let writing = BytesWriting::of(database).await;
        let file = file.clone();
        database
            .call(move |inner| {
                inner.access.with_writer(|writer| {
                    writer.transaction(|db| {
                        if db.query_row(
                            &format!("SELECT {CACHE_NAME_USED_SQL}"),
                            [name.as_str()],
                            |r| r.get::<_, bool>(0),
                        )? {
                            return Err(DbError::FileNameReused { name: name.as_str().to_owned() });
                        }
                        crate::file_removals::record_cached(db, name.as_str())?;
                        Ok(())
                    })?;
                    inner.staging.reserve([crate::file_removals::OwnedFile::Cached(name.clone())]);
                    Ok::<_, DbError>(())
                })?;
                Ok(CacheReservation {
                    reserved: inner.staging.clone(),
                    file,
                    name,
                    writing: Some(writing),
                    partial: false,
                })
            })
            .await
    }

    /// Observe the upload worker's work: newly queued uploads, and uploads a
    /// file stopped waiting for. Subscribing before reading the work list
    /// prevents missed commits.
    pub fn queued_work_changes(&self) -> DatabaseChanges {
        // A newly queued upload, an upload a file stopped waiting for, which
        // the worker then retires (§17.9), and a self-replacement's write
        // pause starting or ending, which holds trailing uploads'
        // completions (§12.3, §17.4).
        let reads = vec![
            crate::observation::TableRead {
                table: "_coven_write_pause".into(),
                columns: ["paused".into()].into(),
                scope: crate::row_scope::RowScope::All,
                inserted_only: false,
            },
            crate::observation::TableRead {
                table: "_coven_file_uploads".into(),
                columns: ["location".into()].into(),
                scope: crate::row_scope::RowScope::All,
                inserted_only: true,
            },
            crate::observation::TableRead {
                table: "_coven_upload_files".into(),
                columns: ["file".into()].into(),
                scope: crate::row_scope::RowScope::All,
                inserted_only: false,
            },
        ];
        self.database.observer.changes(reads, true)
    }
}

/// Bytes reserved for one file's whole encrypted cache file while it is
/// written, until a prepared pin takes it (§17.8).
pub struct CacheReservation {
    /// The names the store that reserved this file reserves.
    reserved: Arc<crate::reserved_names::ReservedNames>,
    file: FileRef,
    name: FileName,
    /// Held until a prepared pin takes the copy or it drops.
    writing: Option<BytesWriting>,
    /// Whether the file is kept as a partial download, its row naming it.
    partial: bool,
}
impl CacheReservation {
    /// The cache file's name.
    pub fn name(&self) -> &FileName {
        &self.name
    }
}
/// What a reservation holds while its bytes are still written, a pin's
/// whole cache file or a device-only call's copies: the lease closing the
/// store waits for, and the open store, so that once its last handle drops
/// the store stays closing, and a reopen waits, until the writing ends.
/// Opening removes recorded bytes, and would delete bytes this writer still
/// owns (Appendix C.3, F.4).
struct BytesWriting {
    _lease: tokio::sync::OwnedRwLockReadGuard<()>,
    _store: Arc<std::sync::RwLock<Option<DatabaseInner>>>,
}
impl BytesWriting {
    /// Hold `database`'s store while bytes are written, once no close has
    /// begun waiting for its file work.
    async fn of(database: &Database) -> Self {
        Self {
            _lease: database.writer.file_tasks.clone().read_owned().await,
            _store: database.inner.clone(),
        }
    }
}
impl Drop for CacheReservation {
    fn drop(&mut self) {
        // The name goes before the store the writing holds can close.
        let Some(_writing) = self.writing.take() else {
            return;
        };
        self.reserved.release([crate::file_removals::OwnedFile::Cached(self.name.clone())]);
    }
}
/// One upload's completion (§17.9): a local write that switches the files
/// the upload stores to the cloud, then retires the upload.
fn complete(
    inner: &DatabaseInner,
    id: i64,
    subject: crate::pending::SubjectKey,
) -> Result<crate::UploadCompletion, DbError> {
    inner.with_files("complete an upload", |writer, files| {
        writer.local_write(
            &inner.access.write_schema,
            inner.access.device,
            inner.clock.now(),
            files,
            |files| {
                files.complete_upload(
                    id,
                    Box::new(move |db| crate::pending::clear_key(db, &subject)),
                )
            },
        )
    })
}
/// Find a queued upload's source under the writer, then open it once the
/// writer is free: opening is file IO (§21.2).
fn open_upload_source<L>(
    access: &DatabaseAccess<L>,
    upload: i64,
    object: ObjectId,
    size: u64,
) -> Result<Option<crate::LocalFileStream>, crate::LocalFileError> {
    let source = access.with_writer(|db| file_ref::queued_source(db, upload))?;
    source
        .map(|source| file_ref::open_source(&access.directory, &object.to_string(), size, source))
        .transpose()
}

impl Database {
    /// Settle what other devices' writes left for this device's files, in
    /// one write of its own (§17.2): delete each record they left with no
    /// reference. An audience reloading, or a pause of this device's writes,
    /// leaves its files for a later call.
    pub async fn settle_files(&self) -> Result<(), DbError> {
        self.call(move |inner| {
            let waiting = inner.access.with_committed(|db| {
                db.query_row(
                    "SELECT EXISTS(SELECT 1 FROM _coven_unreferenced_files)
                        AND NOT EXISTS(SELECT 1 FROM _coven_write_pause)",
                    [],
                    |r| r.get::<_, bool>(0),
                )
            })?;
            if !waiting {
                return Ok(());
            }
            inner.with_files("settle files", |writer, files| {
                writer.local_write(
                    &inner.access.write_schema,
                    inner.access.device,
                    inner.clock.now(),
                    files,
                    |files| files.settle(),
                )
            })
        })
        .await
    }
}

fn changed(count: usize) -> Result<(), DbError> {
    if count == 1 {
        Ok(())
    } else {
        Err(DbError::Invariant { what: "a queued upload's update changes exactly its row" })
    }
}

/// What this device knows locally about a file it uploads (§17.6): the
/// upload, where its bytes go, its move, and the records about each.
#[derive(Debug)]
pub struct LocalEvidence<P> {
    /// The upload's queue id.
    pub upload: i64,
    /// Where its bytes go, with their key.
    pub location: crate::FileLocation,
    /// The move it belongs to.
    pub of: crate::MoveId,
    /// The file's original, for a user's file.
    pub original: Option<std::path::PathBuf>,
    /// This device's record about the upload's bytes.
    pub record: Option<P>,
    /// The move's record.
    pub moving: Option<P>,
}

#[path = "file_database_pins.rs"]
mod pins;

#[path = "file_database_partial_pins.rs"]
mod partial_pins;
pub(crate) use partial_pins::remove as remove_partial_pin;

#[path = "file_database_prepared_pin.rs"]
mod prepared_pin;
pub use prepared_pin::PreparedPin;

#[path = "file_database_copies.rs"]
pub(crate) mod copies;
pub use copies::{CopyReservation, FolderCopy, PreparedDeviceOnly};

#[cfg(test)]
#[path = "file_database_tests.rs"]
pub(crate) mod tests;