haematite-cli 0.21.7

//! Each collection and branch placed in Lys under its database, so a grant on
//! `haematite.database <id>` reaches it (ACCESS-004).
//!
//! Lys decides a `haematite.collection` or `haematite.branch` as
//! `{database}.{name}` (serve/pass.rs `target`). A relation held on the
//! database flows to one only once Lys has recorded it placed there, through
//! `POST /api/apps/haematite/placements`, which takes haematite's own app
//! credential (`lys-app.haematite.<secret>`, the one `haematite lys-schema` sends).
//!
//! The service places a collection after the `collection.put` that names it
//! commits, a branch after the `branch.fork` that makes it commits, and, once
//! at start, every collection and branch the store holds, so a store made
//! before this, or a placement that failed, is healed by the next start.
//! Placing runs on a thread of its own, in order, off the request path: a
//! write never waits for Lys, and a failed placement never undoes a committed
//! write. A failure is logged by the resource's name, and that resource stays
//! unreachable through grants on the database until it is placed. Nothing
//! retries on a timer.
//!
//! Each placement's operation id is derived from the database, the kind and
//! the name, so the same placement sent again is the act Lys already kept,
//! never a second one. A resource Lys already holds placed under another act
//! (placed by hand, say) is refused `placement_invalid`, "already placed in a
//! parent"; that is logged as already placed, not as a failure. Lys's answer
//! does not say which parent it is in.

use std::collections::BTreeSet;
use std::num::NonZeroUsize;
use std::path::Path;
use std::sync::{Arc, Mutex, PoisonError, mpsc};
use std::time::Duration;

use haematite::db::ReadOnlyDatabase;
use haematite::tree::Hash;
use serde_json::{Value, json};

use super::pass::Made;
use super::protocol::{Change, Selection};
use super::storage::{Catalogue, Failure, reads};

const APP: &str = "haematite";
/// The one parent haematite's schema lists for a collection and a branch.
const PARENT: &str = "haematite.database";
/// How long one placement waits for Lys's answer before it is a failure.
const ANSWER_WITHIN: Duration = Duration::from_secs(10);
/// The words Lys refuses a placement with when the child is placed already.
const ALREADY_PLACED: &str = "already placed in a parent";
/// The collections read in one page when listing them at start.
const PAGE: NonZeroUsize = NonZeroUsize::MIN.saturating_add(255);

/// The kind of resource haematite places under its database.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Kind {
    Collection,
    Branch,
}

impl Kind {
    /// The kind as Lys names it.
    #[must_use]
    pub const fn lys(self) -> &'static str {
        match self {
            Self::Collection => "haematite.collection",
            Self::Branch => "haematite.branch",
        }
    }
}

/// One collection or branch, by its name in this store.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Child {
    pub kind: Kind,
    pub name: String,
}

/// The child a committed `change` makes, when it makes one Lys decides.
#[must_use]
pub fn made_by(change: &Change) -> Option<Child> {
    match change {
        Change::Collection(write) => Some(Child {
            kind: Kind::Collection,
            name: write.collection.clone(),
        }),
        Change::Fork(fork) => Some(Child {
            kind: Kind::Branch,
            name: fork.name.clone(),
        }),
        _ => None,
    }
}

/// Every branch the store holds now, then every collection on any of them.
///
/// Each is named once. A branch that cannot be read at its head (one being
/// discarded, say) is logged by name and its collections are passed over; the
/// next start reads it again.
///
/// # Errors
///
/// The branches could not be listed, or a collection page could not be read.
pub fn held(catalogue: &Catalogue, reader: &ReadOnlyDatabase) -> Result<Vec<Child>, Failure> {
    let mut collections = BTreeSet::new();
    let mut children = Vec::new();
    for branch in catalogue.branches(None, usize::MAX)? {
        let selection = Selection {
            branch: branch.branch.clone(),
            version: None,
        };
        let at = match catalogue.capture_held(&selection) {
            Ok(at) => at,
            Err(failure) => {
                eprintln!(
                    "haem/1 lys placement: branch {} could not be read at its head ({failure}), \
                     so its collections are not placed at this start",
                    branch.branch
                );
                continue;
            }
        };
        let mut after = None;
        loop {
            let page = reads::collections(reader, &at, after.as_deref(), PAGE)?;
            collections.extend(page.entries.into_iter().map(|collection| collection.name));
            match page.next_after {
                Some(next) => after = Some(next),
                None => break,
            }
        }
        children.push(Child {
            kind: Kind::Branch,
            name: branch.branch,
        });
    }
    children.extend(collections.into_iter().map(|name| Child {
        kind: Kind::Collection,
        name,
    }));
    Ok(children)
}

/// `op-` and 32 hex digits of a hash of the database, the kind and the name:
/// the same placement is always the same act.
#[must_use]
pub fn operation_id(database: &str, child: &Child) -> String {
    let words = format!(
        "haem placement\0{database}\0{}\0{}",
        child.kind.lys(),
        child.name
    );
    let digest = Hash::of(words.as_bytes()).to_string();
    format!("op-{}", digest.get(..32).unwrap_or(&digest))
}

/// What came of one placement.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Outcome {
    /// Lys kept it now, or had kept it under this same act before.
    Placed,
    /// Lys holds the child placed under another act.
    AlreadyPlaced,
}

/// Lys's placement route for one database, with haematite's app credential.
pub struct Lys {
    http: reqwest::Client,
    address: url::Url,
    bearer: String,
    database: String,
}

impl Lys {
    /// Lys at `issuer`, placing under `haematite.database <database>`, with
    /// the app credential in `credential_file`.
    ///
    /// # Errors
    ///
    /// The file is not an absolute path or not one line of haematite's Lys
    /// app credential, the issuer is not an address, or the client cannot be
    /// made; each by name, never with the credential.
    pub fn new(issuer: &str, database: &str, credential_file: &Path) -> Result<Self, String> {
        if !credential_file.is_absolute() {
            return Err(format!(
                "lys placement_credential_file {} must be an absolute path",
                credential_file.display()
            ));
        }
        let bearer = super::lys_schema::read_credential(credential_file)?;
        let address = url::Url::parse(issuer)
            .and_then(|origin| origin.join(&format!("api/apps/{APP}/placements")))
            .map_err(|_| format!("the lys issuer {issuer} is not an address"))?;
        let http = reqwest::Client::builder()
            .timeout(ANSWER_WITHIN)
            .build()
            .map_err(|_| "the Lys placement client could not be made".to_owned())?;
        Ok(Self {
            http,
            address,
            bearer,
            database: database.to_owned(),
        })
    }

    /// The child's id as Lys decides it: `{database}.{name}`, as
    /// serve/pass.rs `target` names it.
    fn id(&self, child: &Child) -> String {
        format!("{}.{}", self.database, child.name)
    }

    /// The placement request for `child`: under the database, not restricted.
    #[must_use]
    pub fn body(&self, child: &Child) -> Value {
        json!({
            "operation": operation_id(&self.database, child),
            "child": {"kind": child.kind.lys(), "id": self.id(child)},
            "parent": {"kind": PARENT, "id": self.database},
            "restricted": false,
        })
    }

    /// Place `child` under the database.
    ///
    /// # Errors
    ///
    /// Lys could not be reached or answered outside the route's contract, or
    /// refused the placement, by its refusal's name.
    pub async fn place(&self, child: &Child) -> Result<Outcome, String> {
        let body = self.body(child);
        let response = self
            .http
            .post(self.address.clone())
            .bearer_auth(&self.bearer)
            .header("content-type", "application/json")
            .body(body.to_string())
            .send()
            .await
            .map_err(|_| "lys_could_not_be_asked: Lys could not be reached".to_owned())?;
        let status = response.status();
        let text = response
            .text()
            .await
            .map_err(|_| "lys_could_not_be_asked: Lys's answer could not be read".to_owned())?;
        let answer: Value = serde_json::from_str(&text)
            .map_err(|_| format!("lys_could_not_be_asked: Lys answered {status} without JSON"))?;
        if status.is_success() {
            if answer["operation"] != body["operation"] {
                return Err(format!(
                    "lys_could_not_be_asked: Lys answered {status} for another operation"
                ));
            }
            return Ok(Outcome::Placed);
        }
        let refusal = answer["refusal"].as_str().unwrap_or("lys_refused");
        let reason = answer["reason"].as_str().unwrap_or_default();
        if refusal == "placement_invalid" && reason.contains(ALREADY_PLACED) {
            return Ok(Outcome::AlreadyPlaced);
        }
        Err(format!(
            "{refusal}: Lys refused the placement ({status}): {reason}"
        ))
    }

    /// The log line for `child` not placed for `error`.
    fn not_placed(&self, child: &Child, error: &str) -> String {
        format!(
            "haem/1 lys placement failed: {} {} was not placed under {PARENT} {} ({error}); \
             the write that made it stands, and it stays unreachable through grants on the \
             database until it is placed: the next start of this service places it",
            child.kind.lys(),
            self.id(child),
            self.database
        )
    }
}

enum Work {
    /// Every collection and branch the store held at start.
    Existing(Vec<Child>),
    /// One made by a write that committed.
    Committed(Child),
}

/// Who is told when each placement lands: the gate, which a refused act
/// waits on before asking Lys about what was just made (serve/pass.rs).
type Told = Arc<Mutex<Option<Arc<Made>>>>;

/// The thread that places, in the order it is given them.
pub struct Placer {
    work: mpsc::Sender<Work>,
    told: Told,
}

impl Placer {
    /// Start the placing thread for `lys`. It ends when the placer is dropped,
    /// after what it was already given.
    ///
    /// # Errors
    ///
    /// The thread or its runtime could not start.
    pub fn start(lys: Lys) -> Result<Self, String> {
        let runtime = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .map_err(|error| format!("the Lys placement runtime could not start: {error}"))?;
        let (work, queue) = mpsc::channel();
        let told: Told = Arc::new(Mutex::new(None));
        let telling = Arc::clone(&told);
        std::thread::Builder::new()
            .name("haematite-lys-placement".to_owned())
            .spawn(move || {
                for work in &queue {
                    run(&lys, &runtime, &work, &telling);
                }
            })
            .map_err(|error| format!("the Lys placement thread could not start: {error}"))?;
        Ok(Self { work, told })
    }

    /// Tell `made` as each placement from now on lands.
    pub fn tell(&self, made: Arc<Made>) {
        *self.told.lock().unwrap_or_else(PoisonError::into_inner) = Some(made);
    }

    /// Place every collection and branch the store holds at start.
    pub fn existing(&self, children: Vec<Child>) {
        self.send(Work::Existing(children));
    }

    /// Place one made by a write that committed.
    pub fn committed(&self, child: Child) {
        self.send(Work::Committed(child));
    }

    fn send(&self, work: Work) {
        if self.work.send(work).is_err() {
            eprintln!(
                "haem/1 lys placement failed: the placement thread has stopped, so what was \
                 just made stays unreachable through grants on the database until the next \
                 start of this service places it"
            );
        }
    }
}

/// Say that `child`'s placement was answered, to whoever is told.
fn landed(told: &Told, child: &Child) {
    if let Some(made) = told.lock().unwrap_or_else(PoisonError::into_inner).as_ref() {
        made.landed(child.kind.lys(), &child.name);
    }
}

fn run(lys: &Lys, runtime: &tokio::runtime::Runtime, work: &Work, told: &Told) {
    match work {
        Work::Committed(child) => {
            match runtime.block_on(lys.place(child)) {
                Ok(Outcome::Placed) => eprintln!(
                    "haem/1 lys placement: placed {} {} under {PARENT} {}",
                    child.kind.lys(),
                    lys.id(child),
                    lys.database
                ),
                Ok(Outcome::AlreadyPlaced) => eprintln!(
                    "haem/1 lys placement: {} {} is already placed in a parent by another act",
                    child.kind.lys(),
                    lys.id(child)
                ),
                Err(error) => eprintln!("{}", lys.not_placed(child, &error)),
            }
            // placed or not, the answer is in: a gate waiting on it asks Lys now
            landed(told, child);
        }
        Work::Existing(children) => {
            let (mut placed, mut already, mut failed) = (0_usize, 0_usize, 0_usize);
            for child in children {
                match runtime.block_on(lys.place(child)) {
                    Ok(Outcome::Placed) => placed += 1,
                    Ok(Outcome::AlreadyPlaced) => already += 1,
                    Err(error) => {
                        failed += 1;
                        eprintln!("{}", lys.not_placed(child, &error));
                    }
                }
            }
            eprintln!(
                "haem/1 lys placement at start: of {} collections and branches under {PARENT} \
                 {}, {placed} placed, {already} already placed by another act, {failed} not placed",
                children.len(),
                lys.database
            );
        }
    }
}

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