coven_database/

input_database.rs

//! Durable remote observations, independently committed while application writes run.

use super::{with_locked, Database};
use crate::{sqlite::DatabaseConnection, DbError};
use coven_format::path::{ObjectPath, PathError};
use coven_foundation::id_source::{DeviceId, KeyId, ObjectId};
use coven_foundation::locks::lock_unpoisoned;
use std::collections::{BTreeMap, BTreeSet};
use std::sync::{Arc, Mutex, Weak};

#[path = "input_bytes.rs"]
mod bytes;
pub use bytes::RetainedBytes;
#[path = "snapshot_staging.rs"]
pub(crate) mod staging;
#[cfg(any(test, feature = "test-utils"))]
pub use staging::StagedDamage;
pub use staging::{SnapshotStaging, StagedDigest};

type InputConnection = Arc<Mutex<Option<DatabaseConnection>>>;

fn read_local<R>(
    reader: &InputConnection,
    run: impl FnOnce(&DatabaseConnection) -> Result<R, DbError>,
) -> Result<R, DbError> {
    let guard = lock_unpoisoned(reader);
    let reader = guard.as_ref().ok_or(DbError::StoreClosed)?;
    reader.read_transaction(|| run(reader))
}

pub(crate) struct InputDatabase {
    reader: InputConnection,
    /// Shared with the input database's checkpointer (§4).
    writer: Arc<Mutex<DatabaseConnection>>,
    active: Arc<Mutex<BTreeMap<i64, Weak<InputRead>>>>,
    /// The next snapshot staging's number; 0 is the installing snapshot's.
    stagings: std::sync::atomic::AtomicI64,
    /// A test's request to change a staged record before the next load or
    /// file check reads it, as damage on this device's disk would.
    #[cfg(any(test, feature = "test-utils"))]
    damage_staged: std::sync::atomic::AtomicBool,
    /// A test's request to damage the next stagings that finish, and how
    /// many, before finishing checks them, as damage on this device's disk
    /// would.
    #[cfg(any(test, feature = "test-utils"))]
    damage_finishing: Mutex<Option<(staging::StagedDamage, u32)>>,
    /// A test's request to fail the next removal of the stagings a failed
    /// load or file check ends, as a disk error would.
    #[cfg(any(test, feature = "test-utils"))]
    fail_removal: std::sync::atomic::AtomicBool,
}

struct InputRead {
    id: i64,
    active: Weak<Mutex<BTreeMap<i64, Weak<InputRead>>>>,
    receiving: tokio::sync::Mutex<()>,
}

impl Drop for InputRead {
    fn drop(&mut self) {
        if let Some(active) = self.active.upgrade() {
            let mut active = lock_unpoisoned(&active);
            // A concurrent new reader may already have replaced this expired weak entry.
            if active.get(&self.id).is_some_and(|saved| saved.strong_count() == 0) {
                active.remove(&self.id);
            }
        }
    }
}

/// One observed object version. A replacement invalidates the preceding handle.
#[derive(Clone)]
pub struct RetainedInput {
    database: Database,
    id: i64,
    reading: Arc<InputRead>,
}

/// Encrypted bytes received for one provider observation; completion is separate.
pub struct RetainedRange {
    /// Byte offset in the encrypted object.
    pub offset: u64,
    /// The exact provider bytes.
    pub bytes: Vec<u8>,
}

/// Retained encrypted bytes have a maintained total and explicit eviction count.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct InputBudget {
    /// Maximum retained encrypted bytes, excluding checked metadata and references.
    pub maximum: u64,
    /// Currently retained encrypted bytes.
    pub retained: u64,
    /// Bytes explicitly evicted since this catalog was created.
    pub evicted: u64,
}

/// Copy the current claim attempt's exact bytes from the store connection into
/// the input catalog, so applying a landed claim reads no provider bytes.
fn copy_claim(inputs: &InputDatabase, source: &DatabaseConnection, id: i64) -> Result<(), DbError> {
    inputs.write(|db| {
        db.internal_execute("DELETE FROM _coven_input_ranges WHERE object=?1", [id])?;
        crate::claim::read_attempt::<_, DbError>(source, |chunks| {
            let mut offset = 0u64;
            for chunk in chunks {
                let bytes = chunk?;
                let start = sql_offset(offset)?;
                offset += bytes.len() as u64;
                db.internal_execute(
                    "INSERT INTO _coven_input_ranges(object,start,end,bytes) VALUES(?1,?2,?3,?4)",
                    (id, start, sql_offset(offset)?, bytes),
                )?;
            }
            Ok(())
        })?;
        db.internal_execute("UPDATE _coven_input_objects SET received=1 WHERE id=?1", [id])?;
        evict(db, &inputs.active)
    })
}

impl InputDatabase {
    /// Prepare inputs for one slot's application, in one input transaction.
    pub(crate) fn prepare<T>(
        &self,
        consume: impl FnOnce(&crate::input_preparation::PreparedInputs<'_>) -> Result<T, DbError>,
    ) -> Result<T, DbError> {
        with_locked(&self.writer, |writer| {
            writer.transaction(|db| {
                let inputs = crate::input_preparation::PreparedInputs::new(db)?;
                let result = consume(&inputs)?;
                inputs.finish()?;
                Ok(result)
            })
        })
    }
    pub(crate) fn prepare_claim<T>(
        &self,
        attempt: crate::ClaimAttempt<impl std::io::Read>,
        consume: impl FnOnce(
            crate::claim_preparation::PreparedClaim<crate::claim_preparation::PreparedBytes<'_>>,
        ) -> Result<T, DbError>,
    ) -> Result<T, DbError> {
        with_locked(&self.writer, |writer| {
            writer.transaction(|db| crate::claim_preparation::prepare(db, attempt, consume))
        })
    }
    /// Drop an object's received bytes and its receipt, so its next read
    /// fetches it again.
    fn discard(&self, id: i64) -> Result<(), DbError> {
        self.write(|db| {
            db.internal_execute("DELETE FROM _coven_input_ranges WHERE object=?1", [id])?;
            db.internal_execute("UPDATE _coven_input_objects SET received=0 WHERE id=?1", [id])?;
            Ok(())
        })
    }

    fn release(&self, ids: Vec<i64>) -> Result<(), DbError> {
        let pending = self.read(|db| {
            let mut pending = Vec::new();
            for id in ids {
                if !db.query_row(
                    "SELECT released FROM _coven_input_objects WHERE id=?1",
                    [id],
                    |r| r.get::<_, bool>(0),
                )? {
                    pending.push(id);
                }
            }
            Ok(pending)
        })?;
        if pending.is_empty() {
            return Ok(());
        }
        self.write(|db| {
            for id in pending {
                db.internal_execute(
                    "UPDATE _coven_input_objects SET released=1 WHERE id=?1",
                    [id],
                )?;
            }
            Ok(())
        })
    }
    #[cfg(any(test, feature = "test-utils"))]
    pub(super) fn test_track_work(&self, counter: &crate::test_utils::WorkCounter) {
        lock_unpoisoned(&self.reader)
            .as_ref()
            .expect("open input reader")
            .test_track_work(counter.clone());
        lock_unpoisoned(&self.writer).test_track_work(counter.clone());
    }
    /// `writer` is shared with the input database's checkpointer (§4). Its
    /// temporary storage, on disk, holds prepared inputs and staged snapshots.
    pub(crate) fn new(
        reader: DatabaseConnection,
        writer: Arc<Mutex<DatabaseConnection>>,
    ) -> Result<Self, DbError> {
        with_locked(&writer, |writer| staging::create_tables(writer))?;
        Ok(Self {
            reader: Arc::new(Mutex::new(Some(reader))),
            writer,
            active: Arc::new(Mutex::new(BTreeMap::new())),
            stagings: std::sync::atomic::AtomicI64::new(1),
            #[cfg(any(test, feature = "test-utils"))]
            damage_staged: std::sync::atomic::AtomicBool::new(false),
            #[cfg(any(test, feature = "test-utils"))]
            damage_finishing: Mutex::new(None),
            #[cfg(any(test, feature = "test-utils"))]
            fail_removal: std::sync::atomic::AtomicBool::new(false),
        })
    }

    fn reading(&self, id: i64) -> Arc<InputRead> {
        let mut active = lock_unpoisoned(&self.active);
        if let Some(reading) = active.get(&id).and_then(Weak::upgrade) {
            return reading;
        }
        let reading = Arc::new(InputRead {
            id,
            active: Arc::downgrade(&self.active),
            receiving: tokio::sync::Mutex::new(()),
        });
        active.insert(id, Arc::downgrade(&reading));
        reading
    }

    pub(crate) fn close(self) -> Vec<DbError> {
        let reader = lock_unpoisoned(&self.reader).take().expect("open input reader");
        let mut failures: Vec<DbError> = reader.close().err().into_iter().collect();
        // The checkpointer stopped before, and let go of the writer.
        match Arc::try_unwrap(self.writer) {
            Ok(writer) => {
                if let Err(error) = coven_foundation::locks::into_unpoisoned(writer).close() {
                    failures.push(error);
                }
            }
            Err(_held) => failures.push(DbError::Invariant {
                what: "the input writer closes after its checkpointer stopped",
            }),
        }
        failures
    }

    fn read<R>(
        &self,
        run: impl FnOnce(&DatabaseConnection) -> Result<R, DbError>,
    ) -> Result<R, DbError> {
        read_local(&self.reader, run)
    }

    fn write<R>(
        &self,
        run: impl FnOnce(&DatabaseConnection) -> Result<R, DbError>,
    ) -> Result<R, DbError> {
        with_locked(&self.writer, |writer| {
            #[cfg(any(test, feature = "test-utils"))]
            writer.record_work(|work| work.writer_holds += 1);
            writer.transaction(run)
        })
    }
}

#[expect(
    clippy::map_err_ignore,
    reason = "a size past the target integer is TooLarge, which names the field and size"
)]
fn sql_offset(value: u64) -> Result<i64, DbError> {
    i64::try_from(value).map_err(|_| DbError::TooLarge {
        field: "retained input bytes",
        actual: value,
        maximum: i64::MAX as u64,
    })
}

fn observation(db: &DatabaseConnection, path: &str) -> Result<Option<(i64, Vec<u8>)>, DbError> {
    Ok(db
        .query("SELECT id,identity FROM _coven_input_objects WHERE path=?1", [path], |r| {
            Ok((r.get(0)?, r.get(1)?))
        })?
        .into_iter()
        .next())
}

// Nonoverlapping ranges need at most the immediately preceding range plus
// ranges beginning inside the requested interval. Seek to that boundary.
const RANGES: &str = "SELECT max(start,?2),substr(bytes,max(start,?2)-start+1,min(end,?3)-max(start,?2)) FROM _coven_input_ranges WHERE object=?1 AND start>=coalesce((SELECT start FROM _coven_input_ranges WHERE object=?1 AND start<=?2 ORDER BY start DESC LIMIT 1),?2) AND start<?3 AND end>?2 ORDER BY start";

fn budget(db: &DatabaseConnection) -> Result<InputBudget, DbError> {
    db.query_row(
        "SELECT maximum,retained,evicted FROM _coven_input_budget WHERE singleton=1",
        [],
        |r| {
            Ok(InputBudget {
                maximum: r.get::<_, i64>(0)? as u64,
                retained: r.get::<_, i64>(1)? as u64,
                evicted: r.get::<_, i64>(2)? as u64,
            })
        },
    )
}

fn evict(
    db: &DatabaseConnection,
    active: &Mutex<BTreeMap<i64, Weak<InputRead>>>,
) -> Result<(), DbError> {
    // A released object can be borrowed by another consumer. Keep its bytes
    // until that consumer drops its handle, including across network awaits.
    let active = lock_unpoisoned(active);
    let before = budget(db)?;
    let mut retained = before.retained;
    let mut after = 0;
    while retained > before.maximum {
        let next = db.query(
            "SELECT r.id,r.object FROM _coven_input_ranges r JOIN _coven_input_objects o ON o.id=r.object WHERE o.released=1 AND r.id>?1 ORDER BY r.id LIMIT 1",
            [after], |r| Ok((r.get::<_,i64>(0)?, r.get::<_,i64>(1)?)),
        )?.into_iter().next();
        let Some((id, object)) = next else {
            return Err(DbError::InputBudgetExceeded {
                required: retained,
                maximum: before.maximum,
            });
        };
        after = id;
        if active.get(&object).is_some_and(|input| input.strong_count() > 0) {
            continue;
        }
        // Evict a complete input together: retaining only its suffix would
        // turn the next read into separate requests around an internal gap.
        db.internal_execute("DELETE FROM _coven_input_ranges WHERE object=?1", [object])?;
        retained = budget(db)?.retained;
    }
    let evicted = before.retained - retained;
    if evicted > 0 {
        db.internal_execute(
            "UPDATE _coven_input_budget SET evicted=evicted+?1 WHERE singleton=1",
            [sql_offset(evicted)?],
        )?;
        tracing::debug!(bytes = evicted, retained, "evicted retained remote input bytes");
    }
    Ok(())
}

impl InputDatabase {
    /// Delete the records of the objects `select` picks, except those a live
    /// handle holds: a reader in flight keeps its rows. The cascades take each
    /// record's ranges, file evidence and unread parts.
    fn forget(&self, select: &str, params: impl rusqlite::Params) -> Result<(), DbError> {
        self.write(|db| {
            let active = lock_unpoisoned(&self.active);
            for object in db.query(select, params, |r| r.get::<_, i64>(0))? {
                if active.get(&object).is_some_and(|input| input.strong_count() > 0) {
                    continue;
                }
                db.internal_execute("DELETE FROM _coven_input_objects WHERE id=?1", [object])?;
            }
            Ok(())
        })
    }

    fn forget_slots_below(&self, first: u64) -> Result<(), DbError> {
        self.forget("SELECT id FROM _coven_input_objects WHERE slot<?1", [sql_offset(first)?])
    }

    pub(crate) fn forget_path(&self, path: &str) -> Result<(), DbError> {
        self.forget("SELECT id FROM _coven_input_objects WHERE path=?1", [path])
    }
}

impl Database {
    /// Forget the input records of the slots below a checked floor's `first`:
    /// nothing reads them again (§16.4). A record a live read holds stays.
    pub async fn forget_slot_inputs_below(
        &self,
        first: coven_format::value::Slot,
    ) -> Result<(), DbError> {
        self.call(move |inner| inner.inputs.forget_slots_below(first.get())).await
    }

    /// A seekable local input for applying the current claim without a provider
    /// reread after successful creation. One cache identity is reused across
    /// claims; the journal remains the authoritative copy until settlement.
    #[cfg(any(test, feature = "test-utils"))]
    pub async fn claim_input(&self) -> Result<RetainedBytes, DbError> {
        let (reader, reading, length) = self.call(|inner| {
            inner.access.with_committed(|source| {
                let upload = crate::claim::upload(source)?;
                let id = inner.inputs.write(|db: &DatabaseConnection| {
                    db.internal_execute("DELETE FROM _coven_input_objects WHERE path='local-claim'", [])?;
                    db.query_row("INSERT INTO _coven_input_objects(path,identity,observation) VALUES('local-claim',x'',x'') RETURNING id", [], |row| row.get(0))
                })?;
                let reading = inner.inputs.reading(id);
                copy_claim(&inner.inputs, source, id)?;
                inner.inputs.release(vec![id])?;
                Ok::<_, DbError>((inner.inputs.reader.clone(), reading, upload.size))
            })
        }).await?;
        Ok(RetainedBytes::new(reader, reading, length))
    }
    /// The retained provider observation of an exact path, if any. A drained
    /// writer's discovered objects are found here instead of by probing (§6.1).
    pub async fn retained_observation(&self, path: String) -> Result<Option<Vec<u8>>, DbError> {
        self.call(move |inner| {
            inner.inputs.read(|db| {
                Ok(db
                    .query(
                        "SELECT observation FROM _coven_input_objects WHERE path=?1",
                        [&path],
                        |r| r.get(0),
                    )?
                    .into_iter()
                    .next())
            })
        })
        .await
    }

    /// Retain a provider observation before reading it. Immutable identity changes
    /// return `None` and preserve the preceding evidence. A replacement discards
    /// only that path's preceding version. Equal observations acquire no writer.
    pub async fn observe_input(
        &self,
        path: String,
        identity: Vec<u8>,
        observation_bytes: Vec<u8>,
        replaceable: bool,
    ) -> Result<Option<RetainedInput>, DbError> {
        let slot = ObjectPath::parse(&path)
            .map_err(|PathError| DbError::Invariant {
                what: "an observed input names a storage path",
            })?
            .slot_number()
            .map(|slot| sql_offset(slot.get()))
            .transpose()?;
        let id = self.call(move |inner| {
            let previous = inner.inputs.read(|db| observation(db, &path))?;
            match previous {
                Some((id, saved)) if saved == identity => return Ok(Some(id)),
                Some(_) if !replaceable => return Ok(None),
                _ => (),
            }
            inner.inputs.write(|db: &DatabaseConnection| {
                match observation(db, &path)? {
                    Some((id, saved)) if saved == identity => return Ok(Some(id)),
                    Some(_) if !replaceable => return Ok(None),
                    _ => (),
                }
                db.internal_execute("DELETE FROM _coven_input_objects WHERE path=?1", [&path])?;
                db.query_row("INSERT INTO _coven_input_objects(path,slot,identity,observation) VALUES(?1,?2,?3,?4) RETURNING id", (&path, slot, &identity, &observation_bytes), |r| r.get(0)).map(Some)
            })
        }).await?;
        match id {
            None => Ok(None),
            Some(id) => {
                let reading =
                    self.call(move |inner| Ok::<_, DbError>(inner.inputs.reading(id))).await?;
                Ok(Some(RetainedInput { database: self.clone(), id, reading }))
            }
        }
    }

    /// Configure retained encrypted input independently of application file caches.
    /// Lowering the budget evicts oldest completed inputs atomically with accounting.
    /// Checked facts and file protection are never evicted with their bytes.
    #[cfg(any(test, feature = "test-utils"))]
    pub async fn set_input_budget(&self, maximum: u64) -> Result<InputBudget, DbError> {
        self.call(move |inner| {
            let previous = inner.inputs.read(budget)?;
            if previous.maximum == maximum {
                return Ok(previous);
            }
            inner.inputs.write(|db: &DatabaseConnection| {
                db.internal_execute(
                    "UPDATE _coven_input_budget SET maximum=?1 WHERE singleton=1",
                    [sql_offset(maximum)?],
                )?;
                evict(db, &inner.inputs.active)?;
                budget(db)
            })
        })
        .await
    }

    /// Report byte retention and explicit eviction without scanning cached chunks.
    #[cfg(any(test, feature = "test-utils"))]
    pub async fn input_budget(&self) -> Result<InputBudget, DbError> {
        self.call(|inner| inner.inputs.read(budget)).await
    }

    /// The most retained bytes any input write has required since this was
    /// first called, before its evictions, sampled on every change.
    #[cfg(any(test, feature = "test-utils"))]
    pub async fn input_peak(&self) -> Result<u64, DbError> {
        self.call(|inner| {
            with_locked(&inner.inputs.writer, |writer| {
                writer.batch(
                    "CREATE TEMP TABLE IF NOT EXISTS _coven_input_peak(peak INTEGER NOT NULL);
                     INSERT INTO _coven_input_peak SELECT retained FROM main._coven_input_budget WHERE NOT EXISTS(SELECT 1 FROM _coven_input_peak);
                     CREATE TEMP TRIGGER IF NOT EXISTS _coven_input_peak_change AFTER UPDATE OF retained ON main._coven_input_budget BEGIN
                         UPDATE _coven_input_peak SET peak=max(peak,NEW.retained);
                     END;",
                )?;
                writer.query_row("SELECT peak FROM temp._coven_input_peak", [], |row| {
                    row.get::<_, i64>(0)
                })
            })
        })
        .await
        .map(|peak| peak as u64)
    }
}

impl RetainedInput {
    /// Retain the confirmed current claim under its provider observation without
    /// downloading bytes already owned by the journal. The caller has settled
    /// publication and checked the floor before associating these bytes.
    pub async fn keep_claim_bytes(&self) -> Result<(), DbError> {
        if self.received().await? {
            return Ok(());
        }
        let id = self.id;
        self.database
            .call(move |inner| {
                inner.access.with_committed(|source| copy_claim(&inner.inputs, source, id))
            })
            .await
    }
    /// A seekable reader over received bytes, pinned against eviction. It never
    /// fetches missing bytes or acquires the application writer. Consumers may
    /// use it inside a slot transaction after completing transport and floor checks.
    pub async fn local_bytes(&self, length: u64) -> Result<RetainedBytes, DbError> {
        let reader =
            self.database.call(|inner| Ok::<_, DbError>(inner.inputs.reader.clone())).await?;
        Ok(RetainedBytes::new(reader, self.reading.clone(), length))
    }
    /// The first completed provider observation for this identity, including
    /// native metadata whose representation can differ in later listings.
    pub async fn observation(&self) -> Result<Vec<u8>, DbError> {
        self.read(|db, id| {
            db.query_row("SELECT observation FROM _coven_input_objects WHERE id=?1", [id], |r| {
                r.get(0)
            })
        })
        .await
    }
    /// Share an in-flight range across consumers of the same durable observation.
    /// The guard spans checking retained ranges, fetching gaps, and committing them.
    pub async fn receiving(&self) -> tokio::sync::MutexGuard<'_, ()> {
        self.reading.receiving.lock().await
    }
    async fn read<R: Send + 'static>(
        &self,
        run: impl FnOnce(&DatabaseConnection, i64) -> Result<R, DbError> + Send + 'static,
    ) -> Result<R, DbError> {
        let id = self.id;
        self.database.call(move |inner| inner.inputs.read(|db| run(db, id))).await
    }

    async fn write<R: Send + 'static>(
        &self,
        run: impl FnOnce(&DatabaseConnection, i64) -> Result<R, DbError> + Send + 'static,
    ) -> Result<R, DbError> {
        let id = self.id;
        self.database.call(move |inner| inner.inputs.write(|db| run(db, id))).await
    }

    /// Cached portions of a requested interval, clipped so memory is bounded by it.
    pub async fn ranges(&self, start: u64, end: u64) -> Result<Vec<RetainedRange>, DbError> {
        let start = sql_offset(start)?;
        let end = sql_offset(end)?;
        self.read(move |db, id| {
            db.query(RANGES, (id, start, end), |r| {
                Ok(RetainedRange { offset: r.get::<_, i64>(0)? as u64, bytes: r.get(1)? })
            })
        })
        .await
    }

    /// Commit newly fetched, nonoverlapping bytes. Refilling an evicted range is
    /// a new insertion, while retained bytes must be consumed without fetching.
    pub async fn keep_range(&self, offset: u64, bytes: Vec<u8>) -> Result<(), DbError> {
        self.keep_received_range(offset, bytes, false).await
    }

    /// Whether a whole response (or all of its bounded ranges) reached checked
    /// transport EOF. This records receipt, never content authentication.
    pub async fn received(&self) -> Result<bool, DbError> {
        self.read(|db, id| {
            db.query_row("SELECT received FROM _coven_input_objects WHERE id=?1", [id], |r| {
                r.get(0)
            })
        })
        .await
    }

    /// How many of the object's bytes are retained: its ranges never overlap,
    /// so the object is held whole when this equals its size.
    pub async fn retained_length(&self) -> Result<u64, DbError> {
        self.read(|db, id| {
            db.query_row(
                "SELECT coalesce(sum(end-start),0) FROM _coven_input_ranges WHERE object=?1",
                [id],
                |r| r.get::<_, i64>(0),
            )
        })
        .await
        .map(|length| length as u64)
    }

    /// Commit received bytes and, at successful EOF, the transport completion
    /// fact in one transaction. Neither fact can survive a failed byte commit.
    pub async fn keep_received_range(
        &self,
        offset: u64,
        bytes: Vec<u8>,
        complete: bool,
    ) -> Result<(), DbError> {
        if bytes.is_empty() && (!complete || self.received().await?) {
            return Ok(());
        }
        let end = offset
            .checked_add(bytes.len() as u64)
            .ok_or(DbError::Invariant { what: "an input range's end fits u64" })?;
        let offset = sql_offset(offset)?;
        let end = sql_offset(end)?;
        let id = self.id;
        self.database.call(move |inner| inner.inputs.write(|db: &DatabaseConnection| {
            if !bytes.is_empty() {
                db.internal_execute("INSERT INTO _coven_input_ranges(object,start,end,bytes) VALUES(?1,?2,?3,?4)", (id,offset,end,bytes))?;
            }
            if complete {
                db.internal_execute("UPDATE _coven_input_objects SET received=1 WHERE id=?1", [id])?;
            }
            evict(db, &inner.inputs.active)
        })).await
    }

    /// The complete input has been applied (with reversible inputs committed),
    /// or permanently refused. Release its encrypted ranges for explicit eviction;
    /// checked metadata, file protection, and the refusal remain durable.
    pub async fn release_ranges(self) -> Result<(), DbError> {
        let id = self.id;
        self.database.call(move |inner| inner.inputs.release(vec![id])).await
    }

    /// Drop the received bytes, which a second read of the object
    /// contradicted, so the next read fetches the object again (§19.1).
    pub async fn discard(self) -> Result<(), DbError> {
        let id = self.id;
        self.database.call(move |inner| inner.inputs.discard(id)).await
    }

    /// The checked file evidence of these bytes, if any (§17.10).
    pub async fn files(&self) -> Result<Option<FileEvidence>, DbError> {
        self.read(|db, id| {
            let (complete, schema): (Option<bool>, Option<u32>) = db.query_row(
                "SELECT files_complete,files_schema FROM _coven_input_objects WHERE id=?1",
                [id],
                |r| Ok((r.get(0)?, r.get(1)?)),
            )?;
            Ok(match (complete, schema) {
                (None, _) => None,
                (Some(false), Some(version)) => {
                    Some(FileEvidence::Unproven(UnprovenFiles::Schema(version)))
                }
                (Some(false), None) => {
                    let parts = db
                        .query(
                            "SELECT audience,key FROM _coven_input_unread WHERE object=?1",
                            [id],
                            |r| {
                                Ok((
                                    crate::write_encoding::audience(r, 0)?,
                                    r.get::<_, KeyId>(1)?,
                                ))
                            },
                        )?
                        .into_iter()
                        .collect::<BTreeSet<_>>();
                    if parts.is_empty() {
                        return Err(DbError::DamagedDatabase);
                    }
                    Some(FileEvidence::Unproven(UnprovenFiles::Parts(parts)))
                }
                (Some(true), _) => Some(FileEvidence::Complete(
                    db.query(
                        "SELECT device,stored_object FROM _coven_input_files WHERE object=?1 ORDER BY device,stored_object",
                        [id],
                        |r| Ok((r.get::<_, DeviceId>(0)?, r.get::<_, ObjectId>(1)?)),
                    )?
                    .into_iter()
                    .collect(),
                )),
            })
        })
        .await
    }

    /// Record references only after complete authentication. Unproven evidence
    /// proves no file absent and names what must change before these bytes
    /// are checked again.
    pub async fn keep_files(&self, files: FileEvidence) -> Result<(), DbError> {
        if self.files().await?.as_ref() == Some(&files) {
            return Ok(());
        }
        self.write(move |db: &DatabaseConnection, id| {
            db.internal_execute("DELETE FROM _coven_input_files WHERE object=?1", [id])?;
            db.internal_execute("DELETE FROM _coven_input_unread WHERE object=?1", [id])?;
            let (complete, schema) = match &files {
                FileEvidence::Complete(_) => (true, None),
                FileEvidence::Unproven(UnprovenFiles::Schema(version)) => (false, Some(*version)),
                FileEvidence::Unproven(UnprovenFiles::Parts(_)) => (false, None),
            };
            db.internal_execute(
                "UPDATE _coven_input_objects SET files_complete=?2,files_schema=?3 WHERE id=?1",
                (id, complete, schema),
            )?;
            match files {
                FileEvidence::Complete(files) => {
                    for (device, object) in files {
                        db.internal_execute(
                            "INSERT INTO _coven_input_files(object,device,stored_object) VALUES(?1,?2,?3)",
                            (id, device, object),
                        )?;
                    }
                }
                FileEvidence::Unproven(UnprovenFiles::Parts(parts)) => {
                    if parts.is_empty() {
                        return Err(DbError::Invariant {
                            what: "unproven file evidence names a part",
                        });
                    }
                    for (audience, key) in parts {
                        db.internal_execute(
                            "INSERT INTO _coven_input_unread(object,audience,key) VALUES(?1,?2,?3)",
                            (
                                id,
                                &audience,
                                key,
                            ),
                        )?;
                    }
                }
                FileEvidence::Unproven(UnprovenFiles::Schema(_)) => (),
            }
            Ok(())
        })
        .await
    }
}

/// What checked retained bytes establish about the uploaded files they name
/// (§17.10).
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum FileEvidence {
    /// Every fixed file path the bytes name, after complete authentication.
    Complete(BTreeSet<(DeviceId, ObjectId)>),
    /// The bytes cannot prove any file absent, and are not read again for that
    /// purpose until this prerequisite changes.
    Unproven(UnprovenFiles),
}

/// Why retained bytes cannot yet name every file they refer to (§17.10).
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum UnprovenFiles {
    /// Parts sealed for these audiences under these keys, which this device
    /// could not open: it lacked the key or could not read the audience.
    Parts(BTreeSet<(coven_merge::Audience, KeyId)>),
    /// A schema version this version cannot interpret.
    Schema(u32),
}

impl UnprovenFiles {
    /// Whether reading the same bytes again could now succeed: one of the
    /// parts now opens, or the schema version is now supported.
    pub fn changed(
        &self,
        opens: impl Fn(&coven_merge::Audience, KeyId) -> bool,
        versions: &std::ops::RangeInclusive<u32>,
    ) -> bool {
        match self {
            Self::Parts(parts) => parts.iter().any(|(audience, key)| opens(audience, *key)),
            Self::Schema(version) => versions.contains(version),
        }
    }
}

#[cfg(test)]
#[path = "input_database_tests.rs"]
mod tests;