haematite-cli 0.21.8

//! Immutable owner choices at privately held merge inputs.

use std::num::NonZeroUsize;

use haematite::tree::Hash;
use serde::{Deserialize, Serialize};

use super::reads::{PageLimits, bounded_key_page};
use super::review::{choice, cursor};
use super::{
    Captured, Failure, MergePin, MergePlan, MergeState, Storage, branch_name, document_key, edges,
    identifier, reads,
};
use crate::serve::protocol::Node;

const TOKEN_PREFIX: &str = "merge-choice-v1:";

#[path = "merge_coverage.rs"]
pub(super) mod coverage;
pub use coverage::{MergeCoverage, MissingChoice};

#[path = "merge_candidate.rs"]
mod candidate;
pub use candidate::{CandidateAction, CandidateAdvance};

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
pub enum MergeOwner {
    Document {
        collection: String,
        document_id: String,
    },
    Edge {
        relation: String,
        from: Node,
        to: Node,
    },
}

impl MergeOwner {
    fn key(&self) -> Result<Vec<u8>, Failure> {
        match self {
            Self::Document {
                collection,
                document_id,
            } => {
                identifier(collection)?;
                identifier(document_id)?;
                Ok(document_key(collection, document_id))
            }
            Self::Edge { relation, from, to } => edges::keys(from, relation, to).map(|keys| keys.0),
        }
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum MergeChoice {
    TakeSource,
    TakeTarget,
    RemoveEdge,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct MergeChoose {
    pub merge_id: String,
    pub owner: MergeOwner,
    pub token: String,
    pub choice: MergeChoice,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct MergeChoiceRow {
    pub owner: MergeOwner,
    pub choice: MergeChoice,
}

#[derive(Debug, Serialize)]
pub struct RecordedChoice {
    pub token: String,
    pub row: MergeChoiceRow,
}

#[derive(Debug, Serialize)]
pub struct MergeChoices {
    pub merge_id: String,
    pub revision: u64,
    pub entries: Vec<RecordedChoice>,
    pub next_after: Option<String>,
}

#[derive(Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct ChoiceCursor {
    merge_id: String,
    revision: u64,
    token: String,
}

#[derive(Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct Revision {
    number: u64,
}

struct HeldInputs {
    origin: Captured,
    target: Captured,
    source: Captured,
    pins: choice::Pins,
}

struct Triple {
    ancestor: Option<Vec<u8>>,
    target: Option<Vec<u8>>,
    source: Option<Vec<u8>>,
}

impl Triple {
    fn token(&self, pins: &choice::Pins, key: &[u8]) -> Result<String, Failure> {
        pins.token(
            key,
            self.ancestor.as_deref(),
            self.target.as_deref(),
            self.source.as_deref(),
        )
    }

    fn unchanged(&self) -> bool {
        self.ancestor == self.target && self.ancestor == self.source
    }
}

impl Storage {
    /// Admit one immutable choice from owner bytes at the held input versions.
    ///
    /// # Errors
    /// Refuses invalid owners, tokens, moved pins, unresolved endpoints and conflicting choices.
    pub fn merge_choose(&mut self, request: &MergeChoose) -> Result<MergeChoiceRow, Failure> {
        identifier(&request.merge_id)?;
        validate_token(&request.token)?;
        let key = request.owner.key()?;
        validate_choice_kind(&request.owner, request.choice)?;
        let plan = self.open_merge_plan(&request.merge_id)?;
        let inputs = self.choice_inputs(&plan)?;
        let values = self.choice_values(&inputs, &key)?;
        if values.token(&inputs.pins, &key)? != request.token {
            return Err(Failure::new(
                "choice_token_mismatch",
                "token differs from held owner bytes",
            ));
        }
        let service = self.merge_service_selection()?;
        let revision = self.choice_revision(&service, &request.merge_id)?;
        let row = MergeChoiceRow {
            owner: request.owner.clone(),
            choice: request.choice,
        };
        if let Some(existing) = self.read_choice_row(&service, &request.merge_id, &request.token)? {
            if revision == 0 || existing.owner != row.owner {
                return Err(Failure::new(
                    "choice_row_corrupt",
                    "choice identity or revision differs",
                ));
            }
            if existing != row {
                return Err(Failure::new(
                    "choice_conflict",
                    "token already has a different choice",
                ));
            }
            self.confirm_merge_row()?;
            return Ok(existing);
        }
        self.admit_owner_choice(request, &inputs, &values, &service)?;
        let next = revision.checked_add(1).ok_or_else(|| {
            Failure::new(
                "choice_revision_exhausted",
                "choice journal revision cannot advance",
            )
        })?;
        self.write_merge_service_rows(&[
            (
                choice_key(&request.merge_id, &request.token),
                serde_json::to_vec(&row).map_err(Failure::storage)?,
            ),
            (
                revision_key(&request.merge_id),
                serde_json::to_vec(&Revision { number: next }).map_err(Failure::storage)?,
            ),
        ])?;
        Ok(row)
    }

    /// Seek one merge's journal, refusing a cursor if that journal changed.
    ///
    /// # Errors
    /// Refuses missing or abandoned plans, malformed rows and invalid or stale cursors.
    pub fn merge_choices(
        &self,
        id: &str,
        after: Option<&str>,
        limit: NonZeroUsize,
    ) -> Result<MergeChoices, Failure> {
        self.open_merge_plan(id)?;
        let service = self.merge_service_selection()?;
        let revision = self.choice_revision(&service, id)?;
        let prefix = choice_prefix(id);
        let mut from = prefix.clone();
        if let Some(after) = after {
            let decoded: ChoiceCursor = cursor::decode(after)?;
            if decoded.merge_id != id {
                return Err(Failure::new(
                    "invalid_continuation",
                    "cursor belongs to another merge",
                ));
            }
            validate_token(&decoded.token)
                .map_err(|error| Failure::new("invalid_continuation", error.message))?;
            if decoded.revision != revision {
                return Err(Failure::new(
                    "choice_journal_moved",
                    "journal changed after this cursor",
                ));
            }
            if self
                .read_choice_row(&service, id, &decoded.token)?
                .is_none()
            {
                return Err(Failure::new(
                    "invalid_continuation",
                    "cursor token is not in this journal",
                ));
            }
            from.extend_from_slice(decoded.token.as_bytes());
            from.push(0);
        }
        let (entries, next) = bounded_key_page(
            &self.reader,
            &service,
            &prefix,
            &from,
            PageLimits {
                rows: limit,
                bytes: NonZeroUsize::MAX,
            },
            |token, bytes| {
                Ok(RecordedChoice {
                    row: decode_row(&token, &bytes)?,
                    token,
                })
            },
            |entry| {
                serde_json::to_vec(entry)
                    .map(|bytes| bytes.len())
                    .map_err(Failure::storage)
            },
        )?;
        if revision == 0 && !entries.is_empty() {
            return Err(Failure::new(
                "choice_row_corrupt",
                "journal rows have no revision",
            ));
        }
        let next_after = next
            .map(|token| {
                cursor::encode(&ChoiceCursor {
                    merge_id: id.to_owned(),
                    revision,
                    token,
                })
            })
            .transpose()?;
        Ok(MergeChoices {
            merge_id: id.to_owned(),
            revision,
            entries,
            next_after,
        })
    }

    fn open_merge_plan(&self, id: &str) -> Result<MergePlan, Failure> {
        let plan = self
            .merge_plan(id)?
            .ok_or_else(|| Failure::new("merge_missing", "merge id has no control row"))?;
        if plan.state != MergeState::Open {
            return Err(Failure::new("merge_abandoned", "merge has been abandoned"));
        }
        Ok(plan)
    }

    fn choice_inputs(&self, plan: &MergePlan) -> Result<HeldInputs, Failure> {
        let origin = self.merge_held_pin(&plan.id, "origin", &plan.origin)?;
        let target = self.merge_held_pin(&plan.id, "target", &plan.target)?;
        let source = self.merge_held_pin(&plan.id, "source", &plan.source)?;
        self.check_choice_live_pin(&plan.target, &target)?;
        self.check_choice_live_pin(&plan.source, &source)?;
        let pins = choice::Pins::new(&origin, &target, &source)?;
        Ok(HeldInputs {
            origin,
            target,
            source,
            pins,
        })
    }

    fn check_choice_live_pin(&self, pin: &MergePin, held: &Captured) -> Result<(), Failure> {
        self.check_merge_live_pin(pin)?;
        let live = self
            .refs
            .get(&branch_name(&pin.branch))
            .ok_or_else(|| Failure::new("merge_pins_moved", "merge input branch is missing"))?;
        if !live
            .shards
            .iter()
            .map(|shard| (shard.shard_id, shard.head))
            .eq(held.shards.iter().map(|shard| (shard.shard_id, shard.head)))
        {
            return Err(Failure::new(
                "merge_pins_moved",
                "live roots differ from held inputs",
            ));
        }
        Ok(())
    }

    fn choice_values(&self, inputs: &HeldInputs, key: &[u8]) -> Result<Triple, Failure> {
        Ok(Triple {
            ancestor: reads::get_bytes(&self.reader, &inputs.origin, key)?,
            target: reads::get_bytes(&self.reader, &inputs.target, key)?,
            source: reads::get_bytes(&self.reader, &inputs.source, key)?,
        })
    }

    fn admit_owner_choice(
        &self,
        request: &MergeChoose,
        inputs: &HeldInputs,
        values: &Triple,
        service: &Captured,
    ) -> Result<(), Failure> {
        if let MergeOwner::Edge { from, to, .. } = &request.owner {
            let from_absent = self.endpoint_absent(request, inputs, service, from)?;
            let to_absent = self.endpoint_absent(request, inputs, service, to)?;
            let dangling = from_absent || to_absent;
            if dangling {
                if request.choice != MergeChoice::RemoveEdge {
                    return Err(Failure::new(
                        "choice_requires_remove_edge",
                        "dangling edge requires removal",
                    ));
                }
                if values.target.is_some() || values.source.is_some() {
                    return Ok(());
                }
            }
            if request.choice == MergeChoice::RemoveEdge {
                return Err(Failure::new(
                    "choice_not_needed",
                    "edge has no settled dangling endpoint",
                ));
            }
        }
        if values.unchanged() {
            return Err(Failure::new(
                "choice_not_needed",
                "owner has no difference at the pins",
            ));
        }
        if values.target == values.source {
            return Err(Failure::new(
                "choice_same_on_both",
                "owner is the same on both sides",
            ));
        }
        Ok(())
    }

    fn endpoint_absent(
        &self,
        request: &MergeChoose,
        inputs: &HeldInputs,
        service: &Captured,
        node: &Node,
    ) -> Result<bool, Failure> {
        let key = document_key(&node.collection, &node.document_id);
        let values = self.choice_values(inputs, &key)?;
        if values.target.is_some() == values.source.is_some() {
            return Ok(values.target.is_none());
        }
        let token = values.token(&inputs.pins, &key)?;
        let row = self
            .read_choice_row(service, &request.merge_id, &token)?
            .ok_or_else(|| {
                Failure::new(
                    "choice_depends_on_endpoint",
                    "endpoint document must be chosen first",
                )
            })?;
        let owner = MergeOwner::Document {
            collection: node.collection.clone(),
            document_id: node.document_id.clone(),
        };
        if row.owner != owner {
            return Err(Failure::new(
                "choice_row_corrupt",
                "endpoint choice names another owner",
            ));
        }
        match row.choice {
            MergeChoice::TakeSource => Ok(values.source.is_none()),
            MergeChoice::TakeTarget => Ok(values.target.is_none()),
            MergeChoice::RemoveEdge => Err(Failure::new(
                "choice_row_corrupt",
                "document choice removes an edge",
            )),
        }
    }

    fn read_choice_row(
        &self,
        service: &Captured,
        id: &str,
        token: &str,
    ) -> Result<Option<MergeChoiceRow>, Failure> {
        reads::get_bytes(&self.reader, service, &choice_key(id, token))?
            .map(|bytes| decode_row(token, &bytes))
            .transpose()
    }

    fn choice_revision(&self, service: &Captured, id: &str) -> Result<u64, Failure> {
        let stored = reads::get_bytes(&self.reader, service, &revision_key(id))?;
        // A merge has no journal revision before its first choice.
        stored.map_or(Ok(0), |bytes| {
            let revision: Revision = serde_json::from_slice(&bytes)
                .map_err(|error| Failure::new("choice_row_corrupt", error.to_string()))?;
            if revision.number == 0 {
                return Err(Failure::new(
                    "choice_row_corrupt",
                    "stored journal revision is zero",
                ));
            }
            Ok(revision.number)
        })
    }
}

fn validate_token(token: &str) -> Result<(), Failure> {
    let digest = token.strip_prefix(TOKEN_PREFIX).ok_or_else(|| {
        Failure::new("choice_token_invalid", "choice token has an invalid prefix")
    })?;
    if digest.len() != 64
        || !digest
            .bytes()
            .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
    {
        return Err(Failure::new(
            "choice_token_invalid",
            "choice token must contain a canonical digest",
        ));
    }
    Ok(())
}

fn validate_choice_kind(owner: &MergeOwner, selected: MergeChoice) -> Result<(), Failure> {
    if matches!(owner, MergeOwner::Document { .. }) && selected == MergeChoice::RemoveEdge {
        return Err(Failure::new(
            "choice_kind_mismatch",
            "remove_edge requires an edge owner",
        ));
    }
    Ok(())
}

fn decode_row(token: &str, bytes: &[u8]) -> Result<MergeChoiceRow, Failure> {
    validate_token(token).map_err(|error| Failure::new("choice_row_corrupt", error.message))?;
    let row: MergeChoiceRow = serde_json::from_slice(bytes)
        .map_err(|error| Failure::new("choice_row_corrupt", error.to_string()))?;
    row.owner
        .key()
        .map_err(|error| Failure::new("choice_row_corrupt", error.message))?;
    validate_choice_kind(&row.owner, row.choice)
        .map_err(|error| Failure::new("choice_row_corrupt", error.message))?;
    Ok(row)
}

fn choice_prefix(id: &str) -> Vec<u8> {
    format!("j\0merge-choice\0{}\0", Hash::of(id.as_bytes())).into_bytes()
}

fn choice_key(id: &str, token: &str) -> Vec<u8> {
    let mut key = choice_prefix(id);
    key.extend_from_slice(token.as_bytes());
    key
}

fn revision_key(id: &str) -> Vec<u8> {
    format!("j\0merge-choice-revision\0{}", Hash::of(id.as_bytes())).into_bytes()
}

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