haematite-cli 0.21.1

//! Bounded request dispatch. Socket handling owns framing and client admission.

mod text_search;
pub(in crate::serve) use text_search::{Words, not_configured};

use std::cell::Cell;
use std::io::{self, Write};
use std::num::{NonZeroU32, NonZeroU64, NonZeroUsize};
use std::sync::{Arc, Condvar, Mutex, TryLockError};
use std::time::{Duration, Instant};

use haematite::db::ReadOnlyDatabase;
use serde_json::value::RawValue;
use serde_json::{Value, json};

use super::embed::job::Bounds;
use super::embed::settings::Restart;
use super::embedding::{Driver, Guard, Worker};
use super::protocol::{
    BranchPage, Change, CollectionPage, CountsRebuild, DocumentPage, DraftsRun, EdgePage,
    EmbedRewalk, GraphPage, MutationContext, Operation, PROTOCOL, ResponseFailure, ReviewPage,
    Selection, StoreRead,
};
use super::query::Seal;
use super::query::admission::Slots;
use super::query::probe::Prepared;
use super::releaser::{self, Releaser};
use super::storage::{
    Captured, Catalogue, DiscardPasses, Failure, Held, Storage, edges, graph, reads,
};

use super::access::{self, Access, Source, Standalone};
use super::changes::{self, ChangesRequest};
use super::pass::{self, Gate};
use super::placement::{self, Placer};

thread_local! {
    static WRITER_DEADLINE: Cell<Option<(Instant, Duration)>> = const { Cell::new(None) };
}

struct RequestWait(Option<(Instant, Duration)>);
impl RequestWait {
    fn begin(timeout: Duration) -> Result<Self, Failure> {
        if timeout.is_zero() {
            return Err(Failure::new(
                "invalid_params",
                "writer timeout must be positive",
            ));
        }
        let end = Instant::now().checked_add(timeout).ok_or_else(|| {
            Failure::new(
                "invalid_params",
                "writer timeout exceeds the monotonic clock",
            )
        })?;
        Ok(Self(WRITER_DEADLINE.replace(Some((end, timeout)))))
    }
}
impl Drop for RequestWait {
    fn drop(&mut self) {
        WRITER_DEADLINE.set(self.0);
    }
}

#[derive(Default)]
struct WriterAvailable {
    active: Mutex<bool>,
    changed: Condvar,
}

pub struct Service {
    access_source: Arc<dyn Source>,
    /// The Lys gate every request on a connection is decided by; none for an
    /// owner serving without a directory.
    lys: Option<Arc<Gate>>,
    storage: Arc<Mutex<Storage>>,
    writer_available: Arc<WriterAvailable>,
    reader: ReadOnlyDatabase,
    catalogue: Catalogue,
    /// What came of the last discard's ask for a reclaim pass, for `store.stats`.
    discard_passes: DiscardPasses,
    pub max_page: NonZeroU32,
    pub append_prune_limits: super::storage::AppendPruneLimits,
    pub query: Option<Prepared>,
    /// The one limit on queries running at once, made from the settings.
    query_slots: Slots,
    /// The key this service's continuations are sealed with, made with it.
    query_seal: Seal,
    /// The named limits of `changes.list`.
    pub changes: changes::Settings,
    /// The release thread, once started; none in a service built for a test.
    releaser: Option<Releaser>,
    /// The embedding thread, once started; none without embed settings.
    embedding: Option<Driver>,
    /// The most vectors one text search compares: `query_scan_records`.
    pub embed_search_bound: Option<NonZeroU64>,
    /// Places each new collection and branch in Lys after its write commits
    /// (serve/placement.rs); none unless configured.
    placer: Option<Placer>,
}

impl Service {
    /// # Errors
    ///
    /// Propagates any error from `reader`.
    pub fn new(storage: Storage, max_page: NonZeroU32) -> Result<Self, Failure> {
        Self::with_query(storage, max_page, None)
    }

    /// Build the service with optional prepared query settings.
    ///
    /// # Errors
    ///
    /// Returns a storage reader, query admission, or continuation-key refusal.
    pub fn with_query(
        storage: Storage,
        max_page: NonZeroU32,
        query: Option<Prepared>,
    ) -> Result<Self, Failure> {
        Self::with_access_source(storage, max_page, query, Arc::new(Standalone))
    }

    pub(in crate::serve) fn with_access_source(
        storage: Storage,
        max_page: NonZeroU32,
        query: Option<Prepared>,
        access_source: Arc<dyn Source>,
    ) -> Result<Self, Failure> {
        let reader = storage.reader()?;
        let catalogue = storage.catalogue();
        let discard_passes = storage.discard_passes();
        let running = running_limit(query.as_ref().map(Prepared::settings))?;
        let embed_search_bound = query
            .as_ref()
            .map(|prepared| prepared.settings().scan_records);
        Ok(Self {
            access_source,
            lys: None,
            storage: Arc::new(Mutex::new(storage)),
            writer_available: Arc::default(),
            reader,
            catalogue,
            discard_passes,
            max_page,
            append_prune_limits: super::storage::AppendPruneLimits::default(),
            query,
            query_slots: Slots::new(running),
            query_seal: Seal::new()?,
            changes: changes::Settings::default(),
            releaser: None,
            embedding: None,
            embed_search_bound,
            placer: None,
        })
    }

    /// Decide every request through Lys (ACCESS-001 R2): a connection
    /// presents its pass with `session.pass`, and nothing in configuration
    /// grants a writer.
    #[must_use]
    pub fn with_lys(mut self, gate: Gate) -> Self {
        if let Some(placer) = &self.placer {
            placer.tell(gate.made());
        }
        self.lys = Some(Arc::new(gate));
        self
    }

    /// Place this store's collections and branches in Lys under its database
    /// (serve/placement.rs): every one it holds now, handed to `placer` at
    /// once, then each new one after the write that makes it commits.
    ///
    /// # Errors
    ///
    /// The store's branches or collections could not be listed.
    pub fn with_placer(mut self, placer: Placer) -> Result<Self, Failure> {
        if let Some(gate) = &self.lys {
            placer.tell(gate.made());
        }
        placer.existing(placement::held(&self.catalogue, &self.reader)?);
        self.placer = Some(placer);
        Ok(self)
    }

    /// Verify a pass a connection presents, answering `{holder, expires}`.
    ///
    /// # Errors
    ///
    /// Refuses with the pass's stable name, or `invalid_operation` when this
    /// service is not served through Lys.
    pub fn present(&self, token: &str) -> Result<Value, Failure> {
        let Some(gate) = &self.lys else {
            return Err(Failure::new(
                "invalid_operation",
                "this service is not served through Lys, so it takes no pass",
            ));
        };
        gate.present(token, pass::now())
    }

    /// Run one page of the drafts Lys approved for this database, with the
    /// connection's pass `token` as the connector's (ACCESS-001 R2).
    ///
    /// # Errors
    ///
    /// `pass_missing` without a pass, `invalid_operation` when this service
    /// is not served through Lys, or where the run stopped, by name.
    pub fn run_drafts(&self, token: Option<&str>, run: &DraftsRun) -> Result<Value, Failure> {
        let Some(gate) = &self.lys else {
            return Err(Failure::new(
                "invalid_operation",
                "this service is not served through Lys, so it runs no drafts",
            ));
        };
        gate.run_drafts(self, token, run.after.as_deref())
    }

    /// Carry the request's writer deadline through approved draft execution.
    pub(in crate::serve) fn run_drafts_waiting(
        &self,
        token: Option<&str>,
        run: &DraftsRun,
        timeout: Duration,
    ) -> Result<Value, Failure> {
        let _wait = RequestWait::begin(timeout)?;
        self.run_drafts(token, run)
    }

    /// Follow `branch`'s head for one connection, giving each head frame to
    /// `send`, with the connection's pass expiring at `expires` seconds since
    /// the epoch as `clock` reads it (serve/follow.rs). The route was decided
    /// before this, as `head.follow`.
    ///
    /// Answers the refusal that ends the follow.
    ///
    /// # Errors
    ///
    /// The first error `send` returns: the client went away.
    pub fn follow(
        &self,
        branch: &str,
        expires: Option<u64>,
        clock: impl Fn() -> std::time::SystemTime,
        send: impl FnMut(&Value) -> io::Result<()>,
    ) -> io::Result<Failure> {
        super::follow::follow(&self.catalogue, branch, expires, clock, send)
    }

    pub(in crate::serve) fn follow_socket(
        &self,
        socket: &std::os::unix::net::UnixStream,
        branch: &str,
        expires: Option<u64>,
        send: impl FnMut(&Value) -> io::Result<()>,
    ) -> io::Result<Failure> {
        use std::sync::atomic::{AtomicBool, Ordering};
        let cancelled = Arc::new(AtomicBool::new(false));
        let closed = Arc::clone(&cancelled);
        let catalogue = self.catalogue.clone();
        let _monitor = super::follow::Monitor::start(socket.try_clone()?, move || {
            closed.store(true, Ordering::Release);
            catalogue.wake_followers();
        })?;
        super::follow::follow_cancellable(
            &self.catalogue,
            branch,
            expires,
            std::time::SystemTime::now,
            &cancelled,
            send,
        )
    }

    /// Decide `operation` from the connection's pass before it is dispatched.
    /// `words` are the request's exact text, drafted with it when Lys holds
    /// it for approval. A service without a Lys gate is its owner's, and
    /// decides nothing.
    ///
    /// # Errors
    ///
    /// Returns the first refusal of the request's subjects.
    pub fn authorize(
        &self,
        token: Option<&str>,
        operation: &Operation,
        words: Option<&str>,
    ) -> Result<(), Failure> {
        let Some(gate) = &self.lys else {
            return Ok(());
        };
        let absent = |kind: &str, name: &str| match kind {
            "branch" => self.catalogue.head(name).is_ok_and(|head| head.is_none()),
            "collection" => self.collection_absent(operation, name),
            _ => false,
        };
        gate.authorize_with(token, operation, words, pass::now(), &absent)
            .map(drop)
    }

    /// Whether a change's branch holds no collection named `collection`, at
    /// its head. Anything but a change, or a read that fails, answers `false`,
    /// so the collection's own refusal stands.
    fn collection_absent(&self, operation: &Operation, collection: &str) -> bool {
        let Operation::Mutate(mutation) = operation else {
            return false;
        };
        self.catalogue
            .capture_held(&Selection {
                branch: mutation.branch.clone(),
                version: None,
            })
            .and_then(|at| reads::collection(&self.reader, &at, collection))
            .is_ok_and(|found| found.is_none())
    }

    /// Starts the release thread over this service's storage. A write that
    /// finds a release running waits for it rather than being told `busy`.
    ///
    /// # Errors
    ///
    /// The decisions left at open could not be read.
    pub fn start_release(&mut self, settings: releaser::Settings) -> Result<(), Failure> {
        self.releaser = Some(Releaser::start(Arc::clone(&self.storage), settings)?);
        Ok(())
    }

    /// Starts the embedding thread over this service's storage with
    /// `worker`; the worker starts on that thread, off the serving path.
    ///
    /// # Errors
    ///
    /// The declared fields could not be read, or the thread not started.
    pub fn start_embedding(
        &mut self,
        worker: Arc<dyn Worker>,
        bounds: Bounds,
        writer_wait: std::time::Duration,
    ) -> Result<(), Failure> {
        self.start_embedding_with(worker, bounds, writer_wait, Restart::default())
    }

    /// As [`Self::start_embedding`], with a worker that dies started again
    /// as `restart` says (EWR1).
    ///
    /// # Errors
    ///
    /// The declared fields could not be read, or the thread not started.
    pub fn start_embedding_with(
        &mut self,
        worker: Arc<dyn Worker>,
        bounds: Bounds,
        writer_wait: std::time::Duration,
        restart: Restart,
    ) -> Result<(), Failure> {
        self.start_embedding_apart(worker, None, bounds, writer_wait, restart)
    }

    /// As [`Self::start_embedding_with`], with every text search's question
    /// asked of `questions` when given, a worker of its own that the walk
    /// never uses (SL1). The question worker never commits, so it never
    /// takes the admission; the walk's commits go through it (JOIN1).
    ///
    /// # Errors
    ///
    /// The declared fields could not be read, or the thread not started.
    pub fn start_embedding_apart(
        &mut self,
        worker: Arc<dyn Worker>,
        questions: Option<Arc<dyn Worker>>,
        bounds: Bounds,
        writer_wait: std::time::Duration,
        restart: Restart,
    ) -> Result<(), Failure> {
        let storage = Arc::clone(&self.storage);
        let driver = Driver::start_apart(storage, worker, questions, bounds, writer_wait, restart)?;
        self.embedding = Some(driver);
        Ok(())
    }

    /// The embedding thread, once started.
    #[must_use]
    pub const fn embedding(&self) -> Option<&Driver> {
        self.embedding.as_ref()
    }

    #[cfg(test)]
    pub(in crate::serve) fn start_embedding_steps(
        &mut self,
        worker: Arc<dyn Worker>,
        bounds: Bounds,
        restart: Restart,
    ) -> Result<(), Failure> {
        self.embedding = Some(Driver::for_steps(&self.storage, worker, bounds, restart)?);
        Ok(())
    }

    #[cfg(test)]
    pub(in crate::serve) fn embedding_step_at(
        &self,
        bounds: Bounds,
        now: std::time::Instant,
    ) -> Result<bool, String> {
        self.embedding
            .as_ref()
            .ok_or("embedding driver missing")?
            .step_at(&self.storage, bounds, now)
    }

    /// The release thread, once started.
    #[must_use]
    pub const fn releaser(&self) -> Option<&Releaser> {
        self.releaser.as_ref()
    }

    pub(crate) fn activate_reclaim(&self) -> Result<(), Failure> {
        self.admission()?.activate_reclaim()
    }

    /// Wait until no reclaim pass runs, so a test can look at the files.
    ///
    /// # Errors
    ///
    /// The writer is held or the pass failed.
    #[cfg(test)]
    pub fn quiesce_reclaim(&self) -> Result<(), Failure> {
        self.admission()?.quiesce_reclaim()
    }

    /// Each shard's reclaim (SL2: a test's diagnosis).
    ///
    /// # Errors
    /// `writer_unavailable` when the storage lock is poisoned.
    #[cfg(test)]
    pub fn reclaim_status(&self) -> Result<haematite::db::ReclaimStatus, Failure> {
        Ok(self.admission()?.reclaim_status())
    }

    fn releasing(&self) -> bool {
        self.releaser.as_ref().is_some_and(Releaser::is_releasing)
    }

    fn admission(&self) -> Result<Guard<'_, Storage>, Failure> {
        let Some((end, timeout)) = WRITER_DEADLINE.get() else {
            return self.admission_now();
        };
        let started = Instant::now();
        loop {
            if Instant::now() >= end {
                return Err(Failure::new(
                    "busy",
                    format!(
                        "writer lock wait reached timeout_ms={} after {} ms",
                        timeout.as_millis(),
                        started.elapsed().as_millis()
                    ),
                ));
            }
            match self.admission_now() {
                Ok(guard) => {
                    if Instant::now() < end {
                        return Ok(guard);
                    }
                    drop(guard);
                    return Err(Failure::new(
                        "busy",
                        format!(
                            "writer lock wait reached timeout_ms={} after {} ms",
                            timeout.as_millis(),
                            started.elapsed().as_millis()
                        ),
                    ));
                }
                Err(failure) if failure.code == "busy" => {
                    self.wait_writer(end, started, timeout)?;
                }
                Err(failure) => return Err(failure),
            }
        }
    }

    fn wait_writer(
        &self,
        end: Instant,
        started: Instant,
        timeout: Duration,
    ) -> Result<(), Failure> {
        let unavailable = || {
            Failure::new(
                "writer_unavailable",
                "writer wait state requires process recovery",
            )
        };
        let busy = || {
            Failure::new(
                "busy",
                format!(
                    "writer lock wait reached timeout_ms={} after {} ms",
                    timeout.as_millis(),
                    started.elapsed().as_millis()
                ),
            )
        };
        if Instant::now() >= end {
            return Err(busy());
        }
        let waited = {
            let mut active = self
                .writer_available
                .active
                .lock()
                .map_err(|_| unavailable())?;
            if !*active {
                *active = true;
                #[cfg(test)]
                self.writer_available.changed.notify_all();
                let storage = Arc::clone(&self.storage);
                let ready = Arc::clone(&self.writer_available);
                if let Err(error) = std::thread::Builder::new()
                    .name("haematite-writer-ready".to_owned())
                    .spawn(move || {
                        // Waiting for the lock never carries a mutation beyond its caller's deadline.
                        match storage.lock() {
                            Ok(guard) => drop(guard),
                            Err(error) => eprintln!("haematite writer_unavailable: {error}"),
                        }
                        match ready.active.lock() {
                            Ok(mut active) => {
                                *active = false;
                                ready.changed.notify_all();
                            }
                            Err(error) => {
                                eprintln!("haematite writer_wait_unavailable: {error}");
                                ready.changed.notify_all();
                            }
                        }
                    })
                {
                    *active = false;
                    self.writer_available.changed.notify_all();
                    return Err(Failure::new(
                        "writer_unavailable",
                        format!("writer wait thread could not start: {error}"),
                    ));
                }
            }
            let left = end
                .checked_duration_since(Instant::now())
                .filter(|left| !left.is_zero())
                .ok_or_else(busy)?;
            self.writer_available
                .changed
                .wait_timeout_while(active, left, |active| *active)
                .map_err(|_| unavailable())
        }?;
        let timed_out = waited.1.timed_out();
        drop(waited);
        if timed_out {
            return Err(busy());
        }
        Ok(())
    }

    fn admission_now(&self) -> Result<Guard<'_, Storage>, Failure> {
        if let Some(driver) = &self.embedding {
            return match WRITER_DEADLINE.get() {
                Some((end, _)) => driver.admission().foreground_until(&self.storage, end),
                None => driver
                    .admission()
                    .foreground(&self.storage, self.releasing()),
            };
        }
        let unavailable = || {
            Failure::new(
                "writer_unavailable",
                "writer state requires process recovery",
            )
        };
        match self.storage.try_lock() {
            Ok(guard) => Ok(Guard::plain(guard)),
            Err(TryLockError::WouldBlock)
                if self.releasing() && WRITER_DEADLINE.get().is_none() =>
            {
                self.storage
                    .lock()
                    .map(Guard::plain)
                    .map_err(|_| unavailable())
            }
            Err(TryLockError::WouldBlock) => Err(Failure::new(
                "busy",
                "another mutation is being admitted; retry this request identity",
            )),
            Err(TryLockError::Poisoned(_)) => Err(unavailable()),
        }
    }

    /// # Errors
    ///
    /// Returns a refusal with mutation context if dispatch fails.
    /// The answer as a value, for a caller in this process. A stored body in
    /// a `query.run` or `changes.list` answer is equal as a value to what was
    /// written, but a value cannot keep a number's text (`1e2` becomes `1e+2`):
    /// the transport answers through [`Self::dispatch_text`] (NB1).
    pub fn dispatch(&self, operation: Operation) -> Result<Value, ResponseFailure> {
        let mutation = MutationContext::from_operation(&operation);
        let result = access::request(self.access_source.as_ref())
            .and_then(|access| self.dispatch_with_entry_budget(&access, operation, None))
            .and_then(Answer::into_value);
        result.map_err(|failure| ResponseFailure::new(failure, mutation))
    }

    /// Dispatch with the request's configured bound on writer admission.
    ///
    /// # Errors
    ///
    /// Returns a dispatch refusal, invalid timeout, or `busy` naming an exhausted wait.
    pub fn dispatch_waiting(
        &self,
        operation: Operation,
        timeout: std::time::Duration,
    ) -> Result<Value, ResponseFailure> {
        let mutation = MutationContext::from_operation(&operation);
        let _wait = RequestWait::begin(timeout)
            .map_err(|failure| ResponseFailure::new(failure, mutation))?;
        self.dispatch(operation)
    }

    /// Dispatch response text with the request's writer admission bound.
    ///
    /// # Errors
    ///
    /// Returns a dispatch refusal, invalid timeout, or `busy` naming an exhausted wait.
    pub fn dispatch_text_waiting(
        &self,
        operation: Operation,
        limit: NonZeroUsize,
        timeout: Duration,
    ) -> Result<Box<RawValue>, ResponseFailure> {
        let mutation = MutationContext::from_operation(&operation);
        let _wait = RequestWait::begin(timeout)
            .map_err(|failure| ResponseFailure::new(failure, mutation))?;
        self.dispatch_text(operation, limit)
    }

    /// The answer as the text the transport writes (NB1): a `query.run` or
    /// `changes.list` answer carries every stored body as the text it is
    /// stored as, never decoded and encoded again; any other answer is its
    /// value written once. Bounded as [`Self::dispatch_bounded`].
    ///
    /// # Errors
    ///
    /// Returns a refusal with mutation context if dispatch fails or the text
    /// is over `response_byte_limit`.
    pub fn dispatch_text(
        &self,
        operation: Operation,
        response_byte_limit: NonZeroUsize,
    ) -> Result<Box<RawValue>, ResponseFailure> {
        let mutation = MutationContext::from_operation(&operation);
        let access = access::request(self.access_source.as_ref())
            .map_err(|failure| ResponseFailure::new(failure, mutation.clone()))?;
        let text = self
            .dispatch_with_entry_budget(&access, operation, Some(response_byte_limit))
            .and_then(Answer::into_text)
            .map_err(|failure| ResponseFailure::new(failure, mutation.clone()))?;
        if text.get().len() > response_byte_limit.get() {
            return Err(ResponseFailure::after_dispatch(
                Failure::new(
                    "response_limit",
                    "response exceeds the configured byte budget; use a smaller page",
                ),
                mutation,
            ));
        }
        Ok(text)
    }

    /// Dispatch with a cumulative serialized-entry budget for collection and
    /// document pages, then verify the complete result against the same bound.
    /// The transport still owns the final envelope serialization and frame cap.
    ///
    /// # Errors
    ///
    /// Propagates any error from `dispatch_with_entry_budget` or `fits_serialized`.
    pub fn dispatch_bounded(
        &self,
        operation: Operation,
        response_byte_limit: NonZeroUsize,
    ) -> Result<Value, ResponseFailure> {
        let mutation = MutationContext::from_operation(&operation);
        let access = access::request(self.access_source.as_ref())
            .map_err(|failure| ResponseFailure::new(failure, mutation.clone()))?;
        let result = self
            .dispatch_with_entry_budget(&access, operation, Some(response_byte_limit))
            .and_then(Answer::into_value)
            .map_err(|failure| ResponseFailure::new(failure, mutation.clone()))?;
        if !fits_serialized(&result, response_byte_limit)
            .map_err(|failure| ResponseFailure::after_dispatch(failure, mutation.clone()))?
        {
            return Err(ResponseFailure::after_dispatch(
                Failure::new(
                    "response_limit",
                    "response exceeds the configured byte budget; use a smaller page",
                ),
                mutation,
            ));
        }
        Ok(result)
    }

    fn dispatch_with_entry_budget(
        &self,
        access: &Access,
        operation: Operation,
        response_byte_limit: Option<NonZeroUsize>,
    ) -> Result<Answer, Failure> {
        match operation {
            Operation::Mutate(request) => {
                if matches!(&request.change, Change::Discard(_)) && !access.can_read_store() {
                    return Err(Failure::new(
                        "discard_access_refused",
                        "whole store authority is required",
                    ));
                }
                // A merge carries every key the source changed.
                if matches!(&request.change, Change::Merge(_)) && !access.can_read_store() {
                    return Err(Failure::new(
                        "merge_access_refused",
                        "whole store authority is required",
                    ));
                }
                let mut writer = self.admission()?;
                let receipt = writer.mutate_with_append_prune_limits(
                    &request,
                    self.max_page,
                    self.append_prune_limits,
                )?;
                // A discard leaves rows of its generation for release to take.
                let discarded = matches!(&request.change, Change::Discard(_));
                let release_due = self.releaser.as_ref().is_some_and(|releaser| {
                    discarded || releaser.due(writer.advanced_since_release(&request.branch))
                });
                drop(writer);
                if release_due && let Some(releaser) = &self.releaser {
                    if discarded {
                        releaser.discarded(&request.branch);
                    } else {
                        releaser.wake(&request.branch);
                    }
                }
                if let Some(driver) = &self.embedding {
                    wake_embedding(driver, &request.branch, &request.change);
                }
                // A pass issued before this commit cannot carry what it made;
                // the gate asks Lys live for it (serve/pass.rs).
                if let Some(gate) = &self.lys
                    && let Some(child) = placement::made_by(&request.change)
                {
                    gate.placed(
                        child.kind.lys(),
                        &child.name,
                        pass::now(),
                        self.placer.is_some(),
                    );
                }
                // After the commit, never before: a placement that fails
                // leaves the write standing (serve/placement.rs).
                if let Some(placer) = &self.placer
                    && let Some(child) = placement::made_by(&request.change)
                {
                    placer.committed(child);
                }
                serde_json::to_value(receipt)
                    .map(Answer::Value)
                    .map_err(Failure::storage)
            }
            Operation::Receipt(query) => serde_json::to_value(self.admission()?.receipt(&query)?)
                .map(Answer::Value)
                .map_err(Failure::storage),
            Operation::CountsRebuild(request) => {
                self.counts_rebuild(access, &request).map(Answer::Value)
            }
            Operation::EmbedRewalk(request) => {
                self.embed_rewalk(access, &request).map(Answer::Value)
            }
            Operation::QueryRun(request) => self
                .query_run(access, &request, response_byte_limit)
                .map(Answer::Text),
            Operation::ChangesList(request) => self
                .changes_list(access, &request, response_byte_limit)
                .map(Answer::Text),
            read => self
                .dispatch_read(access, read, response_byte_limit)
                .map(Answer::Value),
        }
    }

    fn dispatch_read(
        &self,
        access: &Access,
        operation: Operation,
        response_byte_limit: Option<NonZeroUsize>,
    ) -> Result<Value, Failure> {
        match operation {
            Operation::Hello(agreement) => {
                agreement.validate(agreement.max_frame)?;
                Ok(
                    json!({"protocol": PROTOCOL, "capabilities": [super::protocol::RECEIPT_CAPABILITY], "max_page": self.max_page}),
                )
            }
            Operation::QueryCheck(request) => Ok(super::query::check::check(&request)),
            Operation::QueryWords(request) => super::query::check::words(&request),
            Operation::Branches(page) => self.branch_page(&page),
            Operation::Collections(page) => {
                self.collection_page(access, &page, response_byte_limit)
            }
            Operation::Get(read) => {
                let at = self.capture(&read.selection, false)?;
                let document = reads::document(
                    access,
                    &self.reader,
                    &at,
                    &read.collection,
                    &read.document_id,
                )?;
                Ok(json!({"selection":view(&at), "document":document}))
            }
            Operation::List(page) => self.document_page(access, &page, response_byte_limit),
            Operation::Review(page) => self.branch_review(&page),
            Operation::ReviewCounts(page) => self.branch_review_counts(&page),
            Operation::MergePrepare(page) => self.branch_merge_prepare(&page),
            Operation::MergeDangling(page) => self.branch_merge_dangling(&page),
            Operation::History(page) => self.history_page(&page),
            Operation::EdgesOut(page) => self.edge_page(access, &page, false, response_byte_limit),
            Operation::EdgesIn(page) => self.edge_page(access, &page, true, response_byte_limit),
            Operation::GraphExport(page) => self.graph_export(access, &page, response_byte_limit),
            Operation::VectorGet(read) => {
                let at = self.capture(&read.selection, false)?;
                let vector = super::storage::vectors::get(
                    access,
                    &self.reader,
                    &at,
                    &read.collection,
                    &read.field,
                    &read.document_id,
                )?;
                Ok(json!({"selection":view(&at), "vector":vector}))
            }
            Operation::VectorList(page) => {
                let limit = self.limit(page.limit)?;
                let at = self.capture(&page.selection, page.after.is_some())?;
                let (vectors, next_after) = super::storage::vectors::list(
                    access,
                    &self.reader,
                    &at,
                    &page.collection,
                    &page.field,
                    page.after.as_deref(),
                    (limit, response_limit(response_byte_limit)?),
                )?;
                Ok(json!({"selection":view(&at), "vectors":vectors, "next_after":next_after}))
            }
            Operation::VectorSearch(search) => {
                let k = self.limit(search.k)?;
                let at = self.capture(&search.selection, false)?;
                let (hits, scanned) =
                    super::storage::vectors::search(access, &self.reader, &at, &search, k)?;
                Ok(json!({"selection":view(&at), "hits":hits, "scanned":scanned}))
            }
            Operation::VectorSearchText(search) => self.search_text(access, &search),
            Operation::StoreStats(read) => self.store_stats(access, &read),
            Operation::CountsAudit(selection) => self.counts_audit(access, &selection),
            Operation::VectorFields(read) => self.vector_fields(access, &read),
            Operation::Relations(read) => {
                let at = self.capture(&read.selection, false)?;
                let relations = super::storage::relations::list(
                    access,
                    &self.reader,
                    &at,
                    response_limit(response_byte_limit)?,
                )?;
                Ok(json!({"selection":view(&at), "relations":relations}))
            }
            Operation::Mutate(_)
            | Operation::Receipt(_)
            | Operation::CountsRebuild(_)
            | Operation::EmbedRewalk(_) => Err(Failure::new(
                "invalid_operation",
                "mutation must use writer admission",
            )),
            Operation::QueryRun(_) | Operation::ChangesList(_) => Err(Failure::new(
                "invalid_operation",
                "query.run and changes.list are answered as text",
            )),
            Operation::Present(_) | Operation::DraftsRun(_) | Operation::Follow(_) => {
                Err(Failure::new(
                    "invalid_operation",
                    "session.pass, drafts.run and head.follow are answered by the connection \
                     that asks them, with its pass",
                ))
            }
        }
    }

    fn history_page(&self, page: &super::protocol::HistoryPage) -> Result<Value, Failure> {
        let limit = self.limit(page.limit)?;
        let rows = self
            .catalogue
            .history(&page.branch, page.after, limit.get() + 1)?;
        let more = rows.len() > limit.get();
        let rows: Vec<_> = rows.into_iter().take(limit.get()).collect();
        let next = more.then(|| rows.last().map(|row| row.version)).flatten();
        Ok(json!({"entries":rows.iter().map(view).collect::<Vec<_>>(), "next_after":next}))
    }

    fn branch_page(&self, page: &BranchPage) -> Result<Value, Failure> {
        let limit = self.limit(page.limit)?;
        let rows = self
            .catalogue
            .branches(page.after.as_deref(), limit.get() + 1)?;
        let more = rows.len() > limit.get();
        let rows: Vec<_> = rows.into_iter().take(limit.get()).collect();
        let next = more
            .then(|| rows.last().map(|row| row.branch.clone()))
            .flatten();
        Ok(json!({"entries": rows.iter().map(view).collect::<Vec<_>>(), "next_after": next}))
    }

    fn counts_rebuild(&self, access: &Access, request: &CountsRebuild) -> Result<Value, Failure> {
        if !access.can_read_store() {
            return Err(Failure::new(
                "counts_access_refused",
                "whole store authority is required",
            ));
        }
        let mut writer = self.admission()?;
        serde_json::to_value(writer.rebuild_counts(request)?).map_err(Failure::storage)
    }

    /// EMB-E2 (j): opens one field's rewalk, then wakes its walk.
    fn embed_rewalk(&self, access: &Access, request: &EmbedRewalk) -> Result<Value, Failure> {
        if !access.can_read_store() {
            return Err(Failure::new(
                "embed_rewalk_access_refused",
                "whole store authority is required",
            ));
        }
        let Some(driver) = &self.embedding else {
            return Err(Failure::new(
                "embed_not_configured",
                "this service embeds no field",
            ));
        };
        let (branch, collection, field) = (&request.branch, &request.collection, &request.field);
        let through = self
            .admission()?
            .embed_rewalk_longer(branch, collection, field)?;
        driver.declared(branch, collection, field);
        Ok(
            json!({"branch": branch, "collection": collection, "field": field,
            "through": through}),
        )
    }

    fn counts_audit(&self, access: &Access, selection: &Selection) -> Result<Value, Failure> {
        let at = self.capture(selection, false)?;
        let mut answer = serde_json::to_value(Storage::audit_counts(access, &self.reader, &at)?)
            .map_err(Failure::storage)?;
        answer
            .as_object_mut()
            .ok_or_else(|| Failure::new("answer_shape", "audit answer must be an object"))?
            .insert("selection".to_owned(), view(&at));
        Ok(answer)
    }

    /// Reads at two captures of its own and never takes the writer.
    fn changes_list(
        &self,
        access: &Access,
        request: &ChangesRequest,
        response_byte_limit: Option<NonZeroUsize>,
    ) -> Result<Box<RawValue>, Failure> {
        let context = changes::Context {
            access,
            reader: &self.reader,
            catalogue: &self.catalogue,
            settings: &self.changes,
            seal: &self.query_seal,
        };
        changes::list(&context, request, response_byte_limit)
    }

    fn branch_review(&self, page: &ReviewPage) -> Result<Value, Failure> {
        let result = super::storage::review::page(
            &self.reader,
            &self.catalogue,
            page,
            self.limit(page.limit)?,
        )?;
        // Whole-review totals are collected through the separate counting pages.
        Ok(
            json!({"branch": result.selection.branch, "selection": view(&result.selection),
            "from": result.from, "diff": result.diff, "next_after": result.next_after,
            "changes": null, "conflicts": null, "dangling": null, "base_moved": result.base_moved}),
        )
    }

    fn branch_review_counts(&self, page: &ReviewPage) -> Result<Value, Failure> {
        let result = super::storage::review::counts::page(
            &self.reader,
            &self.catalogue,
            page,
            self.limit(page.limit)?,
        )?;
        Ok(json!({
            "branch": result.selection.branch, "selection": view(&result.selection),
            "from": result.from, "parent": view(&result.parent), "partial": result.partial,
            "base_moved": result.base_moved, "done": result.next_after.is_none(),
            "next_after": result.next_after,
        }))
    }

    fn branch_merge_prepare(&self, page: &ReviewPage) -> Result<Value, Failure> {
        let result = super::storage::review::prepare::page(
            &self.reader,
            &self.catalogue,
            page,
            self.limit(page.limit)?,
        )?;
        Ok(json!({
            "branch": result.selection.branch, "selection": view(&result.selection),
            "from": result.from, "parent": view(&result.parent), "lines": result.lines,
            "done": result.next_after.is_none(), "next_after": result.next_after,
        }))
    }

    fn branch_merge_dangling(&self, page: &ReviewPage) -> Result<Value, Failure> {
        let result = super::storage::review::dangling::page(
            &self.reader,
            &self.catalogue,
            page,
            self.limit(page.limit)?,
        )?;
        Ok(json!({
            "branch": result.selection.branch, "selection": view(&result.selection),
            "from": result.from, "parent": view(&result.parent), "lines": result.lines,
            "done": result.next_after.is_none(), "next_after": result.next_after,
        }))
    }

    fn collection_page(
        &self,
        access: &Access,
        page: &CollectionPage,
        response_byte_limit: Option<NonZeroUsize>,
    ) -> Result<Value, Failure> {
        let limit = self.limit(page.limit)?;
        let at = self.capture(&page.selection, page.after.is_some())?;
        let result = if let Some(byte_limit) = response_byte_limit {
            reads::collections_bounded(
                access,
                &self.reader,
                &at,
                page.after.as_deref(),
                limit,
                byte_limit,
            )?
        } else {
            let result = reads::collections(&self.reader, &at, page.after.as_deref(), limit)?;
            reads::Page {
                branch: result.branch,
                version: result.version,
                entries: reads::render_collections(access, &self.reader, &at, result.entries)?,
                next_after: result.next_after,
            }
        };
        Ok(json!({
            "selection": view(&at), "entries":result.entries, "next_after":result.next_after,
        }))
    }

    fn document_page(
        &self,
        access: &Access,
        page: &DocumentPage,
        response_byte_limit: Option<NonZeroUsize>,
    ) -> Result<Value, Failure> {
        reads::validate_document_content(page.content.as_deref())?;
        reads::validate_document_prefix(page.prefix.as_deref(), page.after.as_deref())?;
        let limit = self.limit(page.limit)?;
        let at = self.capture(&page.selection, page.after.is_some())?;
        let result =
            reads::list_documents(access, &self.reader, &at, page, limit, response_byte_limit)?;
        Ok(json!({"selection":view(&at), "entries":result.entries, "next_after":result.next_after}))
    }

    fn store_stats(&self, access: &Access, read: &StoreRead) -> Result<Value, Failure> {
        let at = self.capture(&read.selection, false)?;
        let stats = reads::stats(access, &self.reader, &at)?;
        let mut answer = json!({
            "selection":view(&at), "collections":stats.collections,
            "embedded_fields":stats.embedded_fields,
            "embedded_fields_reason":stats.embedded_fields_reason,
            "live_bytes":null, "history_bytes":null, "reclaimable_bytes":null,
        });
        stats.documents.insert(&mut answer, "documents")?;
        stats.edges.insert(&mut answer, "edges")?;
        stats.vectors.insert(&mut answer, "vectors")?;
        stats
            .collection_count
            .insert(&mut answer, "collection_count")?;
        stats.relation_count.insert(&mut answer, "relation_count")?;
        if let Some(object) = answer.as_object_mut() {
            let release = self
                .releaser
                .as_ref()
                .map_or_else(|| json!({"state":"not_started"}), Releaser::status);
            object.insert("release".to_owned(), release);
            object.insert("discard_pass".to_owned(), self.discard_passes.status());
            let embedding = self
                .embedding
                .as_ref()
                .map_or_else(|| json!({"state":"not_started"}), Driver::status);
            object.insert("embedding".to_owned(), embedding);
        }
        Ok(answer)
    }

    fn vector_fields(&self, access: &Access, read: &StoreRead) -> Result<Value, Failure> {
        let at = self.capture(&read.selection, false)?;
        let mut listing = super::storage::vectors::fields(access, &self.reader, &at)?;
        // EWR1: a field this service embeds says what its driver is doing.
        if let Some(driver) = &self.embedding {
            listing.statused(|collection, field| super::storage::vectors::FieldState {
                status: driver.field_status(&at.branch, collection, field),
                longer_than_model: driver.longer_than_model(&at.branch, collection, field),
                embed_input_empty: driver.embed_input_empty(&at.branch, collection, field),
                windows: driver.windows(&at.branch, collection, field),
                failure: driver.field_failure(&at.branch, collection, field),
            });
        }
        Ok(json!({"selection":view(&at), "fields":listing.fields,
            "embedded_fields":listing.embedded_fields}))
    }

    fn edge_page(
        &self,
        access: &Access,
        page: &EdgePage,
        incoming: bool,
        response_byte_limit: Option<NonZeroUsize>,
    ) -> Result<Value, Failure> {
        let limit = self.limit(page.limit)?;
        let at = self.capture(&page.selection, page.after.is_some())?;
        let query = if incoming {
            edges::visible_incoming
        } else {
            edges::visible_out
        };
        let result = query(
            &self.reader,
            &at,
            &edges::VisibleEdges {
                access,
                node: &page.node,
                relation: page.relation.as_deref(),
                after: page.after.as_deref(),
                limits: reads::PageLimits {
                    rows: limit,
                    bytes: response_limit(response_byte_limit)?,
                },
            },
        )?;
        Ok(json!({"selection":view(&at), "edges":result.entries, "next_after":result.next_after}))
    }

    fn graph_export(
        &self,
        access: &Access,
        page: &GraphPage,
        response_byte_limit: Option<NonZeroUsize>,
    ) -> Result<Value, Failure> {
        let limit = self.limit(page.limit)?;
        let at = self.capture(&page.selection, page.after.is_some())?;
        if graph::is_filtered(page) {
            let result = graph::export(
                &self.reader,
                &at,
                access,
                page,
                reads::PageLimits {
                    rows: limit,
                    bytes: response_limit(response_byte_limit)?,
                },
                self.max_page,
            )?;
            return Ok(json!({"selection":view(&at), "edges":result.edges,
                "nodes":result.nodes, "next_after":result.next_after,
                "examination_ended":result.examination_ended}));
        }
        let result = edges::visible_export(
            &self.reader,
            &at,
            access,
            page.after.as_deref(),
            reads::PageLimits {
                rows: limit,
                bytes: response_limit(response_byte_limit)?,
            },
        )?;
        Ok(json!({"selection":view(&at), "edges":result.entries, "next_after":result.next_after}))
    }

    /// The capture a read answers from, held until the answer is built: no
    /// reclaim pass gives its roots back while the read runs (STOR-003
    /// H1b2-A1).
    fn capture(&self, selection: &Selection, continuation: bool) -> Result<Held, Failure> {
        if continuation && selection.version.is_none() {
            return Err(Failure::new(
                "version_required",
                "continuation requires the captured version",
            ));
        }
        self.catalogue.capture_held(selection)
    }

    fn limit(&self, limit: NonZeroU32) -> Result<NonZeroUsize, Failure> {
        if limit > self.max_page {
            return Err(Failure::new(
                "page_limit",
                format!("maximum page size is {}", self.max_page),
            ));
        }
        let value = usize::try_from(limit.get()).map_err(Failure::storage)?;
        NonZeroUsize::new(value)
            .filter(|limit| limit.get() < usize::MAX)
            .ok_or_else(|| Failure::new("page_limit", "page must allow one lookahead entry"))
    }
}

fn response_limit(limit: Option<NonZeroUsize>) -> Result<NonZeroUsize, Failure> {
    let bytes = limit.map_or(usize::MAX, NonZeroUsize::get);
    NonZeroUsize::new(bytes)
        .ok_or_else(|| Failure::new("response_limit", "response byte limit is zero"))
}

fn running_limit(settings: Option<&super::query::Settings>) -> Result<NonZeroUsize, Failure> {
    let running = settings.map_or(1, |settings| settings.max_running.get());
    NonZeroUsize::new(running)
        .ok_or_else(|| Failure::new("query_limit", "query running limit is zero"))
}

fn view(at: &Captured) -> Value {
    json!({"branch":at.branch, "version":at.version,
        "roots":at.shards.iter().map(|shard| json!({"shard":shard.shard_id, "root":shard.head.to_string()})).collect::<Vec<_>>()})
}

/// What a verb answers (NB1): a value, or text that carries stored bodies as
/// they are stored.
enum Answer {
    Value(Value),
    Text(Box<RawValue>),
}

impl Answer {
    /// The answer as a value; text is read back once, equal as a value.
    fn into_value(self) -> Result<Value, Failure> {
        match self {
            Self::Value(value) => Ok(value),
            Self::Text(text) => serde_json::from_str(text.get()).map_err(Failure::storage),
        }
    }

    /// The answer as text; a value is written once, text is passed on as it is.
    fn into_text(self) -> Result<Box<RawValue>, Failure> {
        match self {
            Self::Value(value) => serde_json::value::to_raw_value(&value).map_err(Failure::storage),
            Self::Text(text) => Ok(text),
        }
    }
}

pub(super) fn fits_serialized(value: &Value, limit: NonZeroUsize) -> Result<bool, Failure> {
    let mut output = CappedWriter {
        remaining: limit.get(),
        exceeded: false,
    };
    match serde_json::to_writer(&mut output, value) {
        Ok(()) => Ok(true),
        Err(_) if output.exceeded => Ok(false),
        Err(error) => Err(Failure::storage(error)),
    }
}

struct CappedWriter {
    remaining: usize,
    exceeded: bool,
}

impl Write for CappedWriter {
    fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
        if bytes.len() > self.remaining {
            self.exceeded = true;
            return Err(io::Error::other("serialized response exceeds byte limit"));
        }
        self.remaining -= bytes.len();
        Ok(bytes.len())
    }

    fn flush(&mut self) -> io::Result<()> {
        Ok(())
    }
}

/// Wake the embedding thread for a change that writes a record's text or
/// declares a field; nothing else is waited on.
fn wake_embedding(driver: &Driver, branch: &str, change: &Change) {
    match change {
        Change::Put(write) => driver.wrote(branch, &write.collection),
        Change::AppendPrune(write) => driver.wrote(branch, &write.collection),
        Change::Delete(delete) => driver.wrote(branch, &delete.collection),
        Change::DeleteMany(delete) => driver.wrote(branch, &delete.collection),
        Change::PutMany(import) => driver.wrote(branch, &import.collection),
        Change::VectorDeclare(declare) => {
            driver.declared(branch, &declare.collection, &declare.field);
        }
        _ => {}
    }
}

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

#[cfg(test)]
#[path = "service_merge_prepare_tests.rs"]
mod merge_prepare_tests;

#[cfg(test)]
#[path = "service_relationship_access_tests.rs"]
mod relationship_access_tests;

#[cfg(test)]
#[path = "service_access_tests.rs"]
mod access_tests;

#[cfg(test)]
#[path = "service_mutation_outcome_tests.rs"]
mod mutation_outcome_tests;

#[cfg(test)]
#[path = "service_held_tests.rs"]
mod held_tests;

#[cfg(test)]
mod count_access_tests {
    use super::Service;
    use crate::serve::access::TestSource;
    use crate::serve::protocol::{CountsRebuild, Operation, Selection};
    use crate::serve::storage::Storage;
    use haematite::db::DatabaseConfig;
    use haematite::store::NodeCacheBudget;
    use std::error::Error;
    use std::num::NonZeroU32;
    use std::sync::Arc;

    #[test]
    fn whole_store_authority_is_required_without_disclosing_or_rebuilding()
    -> Result<(), Box<dyn Error>> {
        let directory = tempfile::tempdir()?;
        let storage = Storage::create(
            DatabaseConfig {
                data_dir: directory.path().join("database"),
                shard_count: 1,
                distributed: None,
                executor_threads: Some(1),
                node_cache_budget: Some(NodeCacheBudget::Unlimited),
                retention: haematite::Retention::KeepAll,
            },
            NodeCacheBudget::Unlimited,
            fixture_long_wait_interval(),
        )?;
        let before = storage
            .capture(&Selection {
                branch: "main".to_owned(),
                version: None,
            })?
            .version;
        let source = Arc::new(TestSource::denying(&[("records", "one")]));
        let service = Service::with_access_source(storage, NonZeroU32::MIN, None, source)?;
        let error = service
            .dispatch(Operation::CountsRebuild(CountsRebuild {
                branch: "main".to_owned(),
                request_id: "withheld".to_owned(),
            }))
            .err()
            .ok_or("restricted rebuild was admitted")?;
        let error = serde_json::to_value(error)?;
        assert_eq!(error["code"], "counts_access_refused");
        assert_eq!(error["outcome"], "not_applied");
        let answer = service.dispatch(Operation::CountsAudit(Selection {
            branch: "main".to_owned(),
            version: None,
        }))?;
        assert_eq!(answer["result"], "cannot_tell");
        assert_eq!(answer["reason"], "withheld");
        assert!(answer.get("buckets").is_none());
        assert_eq!(
            answer["selection"]["version"]["sequence"],
            before.sequence.to_string()
        );
        Ok(())
    }

    const fn fixture_long_wait_interval() -> std::time::Duration {
        std::time::Duration::from_secs(1)
    }
}

#[cfg(test)]
#[path = "service_writer_wait_tests.rs"]
mod writer_wait_tests;