haematite-cli 0.21.7

//! Private retained inputs and durable merge abandonment.

use haematite::branch::history::Key as HistoryKey;
use haematite::branch::{
    CommitDurability, CommitRequest, ForkWithChanges, ForkWrite, current_timestamp,
    open as open_branch, remove as remove_branch,
};
use haematite::tree::Hash;
use serde::{Deserialize, Serialize};
use std::num::NonZeroUsize;

use super::{Captured, Failure, SERVICE, Storage, branch_name, identifier, reads, version};
use crate::serve::protocol::{Selection, Version};

const HOLDER_KEY: &[u8] = b"b\0merge-holder";

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct MergePin {
    pub branch: String,
    pub version: Version,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum MergeState {
    Open,
    Abandoned,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct MergePlan {
    pub id: String,
    pub origin: MergePin,
    pub source: MergePin,
    pub target: MergePin,
    pub state: MergeState,
}

impl Storage {
    /// Retain three immutable input versions without changing a visible branch.
    ///
    /// # Errors
    /// Refuses moved pins, reused identities, abandoned plans and storage failures.
    pub fn merge_open(
        &mut self,
        id: &str,
        source: &MergePin,
        target: &MergePin,
    ) -> Result<MergePlan, Failure> {
        identifier(id)?;
        identifier(&source.branch)?;
        identifier(&target.branch)?;
        if let Some(plan) = self.merge_plan(id)? {
            if plan.source != *source || plan.target != *target {
                return Err(Failure::new(
                    "merge_identity_conflict",
                    "merge id has different pins",
                ));
            }
            if plan.state != MergeState::Open {
                return Err(Failure::new("merge_abandoned", "merge has been abandoned"));
            }
            self.confirm_merge_row()?;
            self.ensure_merge_holders(&plan)?;
            return Ok(plan);
        }
        self.check_merge_live_pin(source)?;
        self.check_merge_live_pin(target)?;
        let selected = self.capture_merge_pin(source)?;
        let provenance = self.fork_provenance(&selected)?;
        if provenance.branch != target.branch {
            return Err(Failure::new(
                "merge_target_mismatch",
                "target differs from fork origin",
            ));
        }
        self.retain_merge_live_pin(source)?;
        self.retain_merge_live_pin(target)?;
        let plan = MergePlan {
            id: id.to_owned(),
            origin: MergePin {
                branch: provenance.branch,
                version: provenance.version,
            },
            source: source.clone(),
            target: target.clone(),
            state: MergeState::Open,
        };
        self.ensure_merge_holders(&plan)?;
        self.write_merge_row(&plan)?;
        Ok(plan)
    }

    /// Read a merge's control row from the owner's current service reference.
    ///
    /// # Errors
    /// Refuses malformed rows and propagates point-read failures.
    pub fn merge_plan(&self, id: &str) -> Result<Option<MergePlan>, Failure> {
        identifier(id)?;
        let at = self.merge_service_selection()?;
        let Some(bytes) = reads::get_bytes(&self.reader, &at, &plan_key(id))? else {
            return Ok(None);
        };
        let plan: MergePlan = serde_json::from_slice(&bytes)
            .map_err(|error| Failure::new("merge_row_corrupt", error.to_string()))?;
        if plan.id != id || plan.origin.branch != plan.target.branch {
            return Err(Failure::new(
                "merge_row_corrupt",
                "merge row identity differs",
            ));
        }
        for pin in [&plan.origin, &plan.source, &plan.target] {
            identifier(&pin.branch)?;
        }
        Ok(Some(plan))
    }

    /// Commit abandonment before releasing any retained input reference.
    ///
    /// # Errors
    /// Refuses missing plans and reports uncertain commits or unlinks explicitly.
    pub fn merge_abandon(&mut self, id: &str) -> Result<MergePlan, Failure> {
        let mut plan = self
            .merge_plan(id)?
            .ok_or_else(|| Failure::new("merge_missing", "merge id has no control row"))?;
        if plan.state == MergeState::Open {
            plan.state = MergeState::Abandoned;
            self.write_merge_row(&plan)?;
        } else {
            self.confirm_merge_row()?;
        }
        self.confirmed_candidates
            .remove(&holder_name(id, "candidate"));
        for slot in ["origin", "source", "target", "candidate"] {
            remove_branch(&holder_name(id, slot), &mut self.refs)
                .map_err(|error| Failure::new("merge_release_unknown", error.to_string()))?;
        }
        Ok(plan)
    }

    fn retain_merge_live_pin(&mut self, pin: &MergePin) -> Result<(), Failure> {
        self.check_merge_live_pin(pin)?;
        let record = self
            .refs
            .get(&branch_name(&pin.branch))
            .ok_or_else(|| Failure::new("merge_pins_moved", "merge branch is missing"))?
            .clone();
        self.retain_current(&pin.branch, &record)
    }

    pub(super) fn check_merge_live_pin(&self, pin: &MergePin) -> Result<(), Failure> {
        let record = self
            .refs
            .get(&branch_name(&pin.branch))
            .ok_or_else(|| Failure::new("merge_pins_moved", "merge branch is missing"))?;
        if version(record) != pin.version || self.pending_discards.contains(&record.name) {
            return Err(Failure::new(
                "merge_pins_moved",
                "merge branch version changed",
            ));
        }
        Ok(())
    }

    fn capture_merge_pin(&self, pin: &MergePin) -> Result<Captured, Failure> {
        self.capture(&Selection {
            branch: pin.branch.clone(),
            version: Some(pin.version),
        })
    }

    fn ensure_merge_holders(&mut self, plan: &MergePlan) -> Result<(), Failure> {
        for (slot, pin) in [
            ("origin", &plan.origin),
            ("source", &plan.source),
            ("target", &plan.target),
        ] {
            self.ensure_merge_holder(&holder_name(&plan.id, slot), pin)?;
        }
        Ok(())
    }

    fn ensure_merge_holder(&mut self, name: &str, pin: &MergePin) -> Result<(), Failure> {
        let expected = self.capture_merge_pin(pin)?;
        if let Some(record) = self.refs.get(name) {
            let at = Captured {
                branch: name.to_owned(),
                version: version(record),
                shards: record.shards.clone(),
            };
            let bytes = reads::get_bytes(&self.reader, &at, HOLDER_KEY)?
                .ok_or_else(|| Failure::new("merge_holder_corrupt", "holder has no identity"))?;
            let held: MergePin = serde_json::from_slice(&bytes)
                .map_err(|error| Failure::new("merge_holder_corrupt", error.to_string()))?;
            if held != *pin
                || record.seq != 0
                // A holder is forked at the pinned version: its anchors are that version's heads.
                || !record
                    .shards
                    .iter()
                    .map(|held| (held.shard_id, held.fork_anchor))
                    .eq(expected.shards.iter().map(|pinned| (pinned.shard_id, pinned.head)))
            {
                return Err(Failure::new(
                    "merge_holder_conflict",
                    "holder differs from pinned input",
                ));
            }
            self.refs
                .confirm_version(name, at.version.generation, at.version.sequence)
                .map_err(|error| Failure::new("durability_unknown", error.to_string()))?;
            return Ok(());
        }
        let service =
            open_branch(SERVICE, &self.refs, &self.registry).map_err(Failure::branch_commit)?;
        let writes = [ForkWrite {
            shard_id: self.writer.shard_for(HOLDER_KEY),
            key: HOLDER_KEY.to_vec(),
            value: Some(serde_json::to_vec(pin).map_err(Failure::storage)?),
        }];
        self.writer
            .fork_kept(
                &service,
                HistoryKey {
                    branch: Hash::of(pin.branch.as_bytes()),
                    generation: pin.version.generation,
                    sequence: pin.version.sequence,
                },
                &mut self.refs,
                &self.registry,
                ForkWithChanges {
                    name,
                    timestamp: current_timestamp(),
                    writes: &writes,
                },
            )
            .map_err(|error| {
                Failure::new(
                    if self.refs.get(name).is_some() {
                        "durability_unknown"
                    } else {
                        "merge_hold_refused"
                    },
                    error.to_string(),
                )
            })?;
        Ok(())
    }

    pub(super) fn merge_held_pin(
        &self,
        id: &str,
        slot: &str,
        pin: &MergePin,
    ) -> Result<Captured, Failure> {
        let name = holder_name(id, slot);
        let record = self.refs.get(&name).ok_or_else(|| {
            Failure::new("merge_holder_missing", "merge input reference is missing")
        })?;
        let at = Captured {
            branch: name,
            version: version(record),
            shards: record.shards.clone(),
        };
        let bytes = reads::get_bytes(&self.reader, &at, HOLDER_KEY)?.ok_or_else(|| {
            Failure::new(
                "merge_holder_corrupt",
                "merge input reference has no identity",
            )
        })?;
        let held: MergePin = serde_json::from_slice(&bytes)
            .map_err(|error| Failure::new("merge_holder_corrupt", error.to_string()))?;
        if held != *pin || record.seq != 0 {
            return Err(Failure::new(
                "merge_holder_conflict",
                "merge input identity changed",
            ));
        }
        Ok(Captured {
            branch: pin.branch.clone(),
            version: pin.version,
            shards: record
                .shards
                .iter()
                .map(|shard| super::BranchShardRef {
                    shard_id: shard.shard_id,
                    fork_anchor: shard.fork_anchor,
                    head: shard.fork_anchor,
                })
                .collect(),
        })
    }

    /// Every merge plan still open (STOR-003 HR1): each shard's plan rows at
    /// the service's roots, a page at a time. Abandoned plans rely on nothing.
    ///
    /// # Errors
    /// `merge_row_corrupt` for a row that does not decode; storage failures.
    pub(super) fn open_merge_plans(&self) -> Result<Vec<MergePlan>, Failure> {
        let at = self.merge_service_selection()?;
        let mut open = Vec::new();
        for shard in &at.shards {
            let mut from = PLAN_PREFIX.to_vec();
            loop {
                let page = self
                    .reader
                    .range_page_at(shard.shard_id, shard.head, &from, PLAN_END, PLAN_PAGE)
                    .map_err(Failure::observer)?;
                for (_key, bytes) in page.entries {
                    let plan: MergePlan = serde_json::from_slice(&bytes)
                        .map_err(|error| Failure::new("merge_row_corrupt", error.to_string()))?;
                    if plan.state == MergeState::Open {
                        open.push(plan);
                    }
                }
                match page.next_from {
                    Some(next) => from = next,
                    None => break,
                }
            }
        }
        Ok(open)
    }

    pub(super) fn merge_service_selection(&self) -> Result<Captured, Failure> {
        let record = self
            .refs
            .get(SERVICE)
            .ok_or_else(|| Failure::new("service_missing", "merge control namespace is missing"))?;
        Ok(Captured {
            branch: SERVICE.to_owned(),
            version: version(record),
            shards: record.shards.clone(),
        })
    }

    pub(super) fn confirm_merge_row(&mut self) -> Result<(), Failure> {
        let at = self.merge_service_selection()?;
        self.refs
            .confirm_version(SERVICE, at.version.generation, at.version.sequence)
            .map_err(|error| Failure::new("durability_unknown", error.to_string()))?;
        self.publish_record(SERVICE)
    }

    fn write_merge_row(&mut self, plan: &MergePlan) -> Result<(), Failure> {
        self.write_merge_service_rows(&[(
            plan_key(&plan.id),
            serde_json::to_vec(plan).map_err(Failure::storage)?,
        )])
    }

    pub(super) fn write_merge_service_rows(
        &mut self,
        rows: &[(Vec<u8>, Vec<u8>)],
    ) -> Result<(), Failure> {
        let before = self.merge_service_selection()?.version;
        let branch =
            open_branch(SERVICE, &self.refs, &self.registry).map_err(Failure::branch_commit)?;
        for (key, value) in rows {
            branch
                .put(self.writer.shard_for(key), key, value)
                .map_err(Failure::branch)?;
        }
        self.writer
            .commit(
                &branch,
                &self.registry,
                CommitRequest {
                    durability: CommitDurability::Durable {
                        refs: &mut self.refs,
                    },
                    extra_parents: &[],
                    timestamp: current_timestamp(),
                },
            )
            .map_err(|error| {
                Failure::new(
                    if self
                        .refs
                        .get(SERVICE)
                        .is_some_and(|record| version(record) != before)
                    {
                        "durability_unknown"
                    } else {
                        "merge_write_refused"
                    },
                    error.to_string(),
                )
            })?;
        self.publish_record(SERVICE)
    }
}

pub(super) fn holder_name(id: &str, slot: &str) -> String {
    format!("@haem/merge/{}/{slot}", Hash::of(id.as_bytes()))
}

/// Where every merge plan row starts in the service namespace.
const PLAN_PREFIX: &[u8] = b"j\0merge\0";

/// Just past every key [`PLAN_PREFIX`] starts.
const PLAN_END: &[u8] = b"j\0merge\x01";

/// Plan rows read in one page of one shard.
const PLAN_PAGE: NonZeroUsize = NonZeroUsize::MIN.saturating_add(255);

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

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