haematite_cli/serve/storage/
merge_holders.rs

1//! Private retained inputs and durable merge abandonment.
2
3use haematite::branch::history::Key as HistoryKey;
4use haematite::branch::{
5    CommitDurability, CommitRequest, ForkWithChanges, ForkWrite, current_timestamp,
6    open as open_branch, remove as remove_branch,
7};
8use haematite::tree::Hash;
9use serde::{Deserialize, Serialize};
10use std::num::NonZeroUsize;
11
12use super::{Captured, Failure, SERVICE, Storage, branch_name, identifier, reads, version};
13use crate::serve::protocol::{Selection, Version};
14
15const HOLDER_KEY: &[u8] = b"b\0merge-holder";
16
17#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
18#[serde(deny_unknown_fields)]
19pub struct MergePin {
20    pub branch: String,
21    pub version: Version,
22}
23
24#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
25#[serde(rename_all = "snake_case")]
26pub enum MergeState {
27    Open,
28    Abandoned,
29}
30
31#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
32#[serde(deny_unknown_fields)]
33pub struct MergePlan {
34    pub id: String,
35    pub origin: MergePin,
36    pub source: MergePin,
37    pub target: MergePin,
38    pub state: MergeState,
39}
40
41impl Storage {
42    /// Retain three immutable input versions without changing a visible branch.
43    ///
44    /// # Errors
45    /// Refuses moved pins, reused identities, abandoned plans and storage failures.
46    pub fn merge_open(
47        &mut self,
48        id: &str,
49        source: &MergePin,
50        target: &MergePin,
51    ) -> Result<MergePlan, Failure> {
52        identifier(id)?;
53        identifier(&source.branch)?;
54        identifier(&target.branch)?;
55        if let Some(plan) = self.merge_plan(id)? {
56            if plan.source != *source || plan.target != *target {
57                return Err(Failure::new(
58                    "merge_identity_conflict",
59                    "merge id has different pins",
60                ));
61            }
62            if plan.state != MergeState::Open {
63                return Err(Failure::new("merge_abandoned", "merge has been abandoned"));
64            }
65            self.confirm_merge_row()?;
66            self.ensure_merge_holders(&plan)?;
67            return Ok(plan);
68        }
69        self.check_merge_live_pin(source)?;
70        self.check_merge_live_pin(target)?;
71        let selected = self.capture_merge_pin(source)?;
72        let provenance = self.fork_provenance(&selected)?;
73        if provenance.branch != target.branch {
74            return Err(Failure::new(
75                "merge_target_mismatch",
76                "target differs from fork origin",
77            ));
78        }
79        self.retain_merge_live_pin(source)?;
80        self.retain_merge_live_pin(target)?;
81        let plan = MergePlan {
82            id: id.to_owned(),
83            origin: MergePin {
84                branch: provenance.branch,
85                version: provenance.version,
86            },
87            source: source.clone(),
88            target: target.clone(),
89            state: MergeState::Open,
90        };
91        self.ensure_merge_holders(&plan)?;
92        self.write_merge_row(&plan)?;
93        Ok(plan)
94    }
95
96    /// Read a merge's control row from the owner's current service reference.
97    ///
98    /// # Errors
99    /// Refuses malformed rows and propagates point-read failures.
100    pub fn merge_plan(&self, id: &str) -> Result<Option<MergePlan>, Failure> {
101        identifier(id)?;
102        let at = self.merge_service_selection()?;
103        let Some(bytes) = reads::get_bytes(&self.reader, &at, &plan_key(id))? else {
104            return Ok(None);
105        };
106        let plan: MergePlan = serde_json::from_slice(&bytes)
107            .map_err(|error| Failure::new("merge_row_corrupt", error.to_string()))?;
108        if plan.id != id || plan.origin.branch != plan.target.branch {
109            return Err(Failure::new(
110                "merge_row_corrupt",
111                "merge row identity differs",
112            ));
113        }
114        for pin in [&plan.origin, &plan.source, &plan.target] {
115            identifier(&pin.branch)?;
116        }
117        Ok(Some(plan))
118    }
119
120    /// Commit abandonment before releasing any retained input reference.
121    ///
122    /// # Errors
123    /// Refuses missing plans and reports uncertain commits or unlinks explicitly.
124    pub fn merge_abandon(&mut self, id: &str) -> Result<MergePlan, Failure> {
125        let mut plan = self
126            .merge_plan(id)?
127            .ok_or_else(|| Failure::new("merge_missing", "merge id has no control row"))?;
128        if plan.state == MergeState::Open {
129            plan.state = MergeState::Abandoned;
130            self.write_merge_row(&plan)?;
131        } else {
132            self.confirm_merge_row()?;
133        }
134        self.confirmed_candidates
135            .remove(&holder_name(id, "candidate"));
136        for slot in ["origin", "source", "target", "candidate"] {
137            remove_branch(&holder_name(id, slot), &mut self.refs)
138                .map_err(|error| Failure::new("merge_release_unknown", error.to_string()))?;
139        }
140        Ok(plan)
141    }
142
143    fn retain_merge_live_pin(&mut self, pin: &MergePin) -> Result<(), Failure> {
144        self.check_merge_live_pin(pin)?;
145        let record = self
146            .refs
147            .get(&branch_name(&pin.branch))
148            .ok_or_else(|| Failure::new("merge_pins_moved", "merge branch is missing"))?
149            .clone();
150        self.retain_current(&pin.branch, &record)
151    }
152
153    pub(super) fn check_merge_live_pin(&self, pin: &MergePin) -> Result<(), Failure> {
154        let record = self
155            .refs
156            .get(&branch_name(&pin.branch))
157            .ok_or_else(|| Failure::new("merge_pins_moved", "merge branch is missing"))?;
158        if version(record) != pin.version || self.pending_discards.contains(&record.name) {
159            return Err(Failure::new(
160                "merge_pins_moved",
161                "merge branch version changed",
162            ));
163        }
164        Ok(())
165    }
166
167    fn capture_merge_pin(&self, pin: &MergePin) -> Result<Captured, Failure> {
168        self.capture(&Selection {
169            branch: pin.branch.clone(),
170            version: Some(pin.version),
171        })
172    }
173
174    fn ensure_merge_holders(&mut self, plan: &MergePlan) -> Result<(), Failure> {
175        for (slot, pin) in [
176            ("origin", &plan.origin),
177            ("source", &plan.source),
178            ("target", &plan.target),
179        ] {
180            self.ensure_merge_holder(&holder_name(&plan.id, slot), pin)?;
181        }
182        Ok(())
183    }
184
185    fn ensure_merge_holder(&mut self, name: &str, pin: &MergePin) -> Result<(), Failure> {
186        let expected = self.capture_merge_pin(pin)?;
187        if let Some(record) = self.refs.get(name) {
188            let at = Captured {
189                branch: name.to_owned(),
190                version: version(record),
191                shards: record.shards.clone(),
192            };
193            let bytes = reads::get_bytes(&self.reader, &at, HOLDER_KEY)?
194                .ok_or_else(|| Failure::new("merge_holder_corrupt", "holder has no identity"))?;
195            let held: MergePin = serde_json::from_slice(&bytes)
196                .map_err(|error| Failure::new("merge_holder_corrupt", error.to_string()))?;
197            if held != *pin
198                || record.seq != 0
199                // A holder is forked at the pinned version: its anchors are that version's heads.
200                || !record
201                    .shards
202                    .iter()
203                    .map(|held| (held.shard_id, held.fork_anchor))
204                    .eq(expected.shards.iter().map(|pinned| (pinned.shard_id, pinned.head)))
205            {
206                return Err(Failure::new(
207                    "merge_holder_conflict",
208                    "holder differs from pinned input",
209                ));
210            }
211            self.refs
212                .confirm_version(name, at.version.generation, at.version.sequence)
213                .map_err(|error| Failure::new("durability_unknown", error.to_string()))?;
214            return Ok(());
215        }
216        let service =
217            open_branch(SERVICE, &self.refs, &self.registry).map_err(Failure::branch_commit)?;
218        let writes = [ForkWrite {
219            shard_id: self.writer.shard_for(HOLDER_KEY),
220            key: HOLDER_KEY.to_vec(),
221            value: Some(serde_json::to_vec(pin).map_err(Failure::storage)?),
222        }];
223        self.writer
224            .fork_kept(
225                &service,
226                HistoryKey {
227                    branch: Hash::of(pin.branch.as_bytes()),
228                    generation: pin.version.generation,
229                    sequence: pin.version.sequence,
230                },
231                &mut self.refs,
232                &self.registry,
233                ForkWithChanges {
234                    name,
235                    timestamp: current_timestamp(),
236                    writes: &writes,
237                    note: None,
238                },
239            )
240            .map_err(|error| {
241                Failure::new(
242                    if self.refs.get(name).is_some() {
243                        "durability_unknown"
244                    } else {
245                        "merge_hold_refused"
246                    },
247                    error.to_string(),
248                )
249            })?;
250        Ok(())
251    }
252
253    pub(super) fn merge_held_pin(
254        &self,
255        id: &str,
256        slot: &str,
257        pin: &MergePin,
258    ) -> Result<Captured, Failure> {
259        let name = holder_name(id, slot);
260        let record = self.refs.get(&name).ok_or_else(|| {
261            Failure::new("merge_holder_missing", "merge input reference is missing")
262        })?;
263        let at = Captured {
264            branch: name,
265            version: version(record),
266            shards: record.shards.clone(),
267        };
268        let bytes = reads::get_bytes(&self.reader, &at, HOLDER_KEY)?.ok_or_else(|| {
269            Failure::new(
270                "merge_holder_corrupt",
271                "merge input reference has no identity",
272            )
273        })?;
274        let held: MergePin = serde_json::from_slice(&bytes)
275            .map_err(|error| Failure::new("merge_holder_corrupt", error.to_string()))?;
276        if held != *pin || record.seq != 0 {
277            return Err(Failure::new(
278                "merge_holder_conflict",
279                "merge input identity changed",
280            ));
281        }
282        Ok(Captured {
283            branch: pin.branch.clone(),
284            version: pin.version,
285            shards: record
286                .shards
287                .iter()
288                .map(|shard| super::BranchShardRef {
289                    shard_id: shard.shard_id,
290                    fork_anchor: shard.fork_anchor,
291                    head: shard.fork_anchor,
292                })
293                .collect(),
294        })
295    }
296
297    /// Every merge plan still open (STOR-003 HR1): each shard's plan rows at
298    /// the service's roots, a page at a time. Abandoned plans rely on nothing.
299    ///
300    /// # Errors
301    /// `merge_row_corrupt` for a row that does not decode; storage failures.
302    pub(super) fn open_merge_plans(&self) -> Result<Vec<MergePlan>, Failure> {
303        let at = self.merge_service_selection()?;
304        let mut open = Vec::new();
305        for shard in &at.shards {
306            let mut from = PLAN_PREFIX.to_vec();
307            loop {
308                let page = self
309                    .reader
310                    .range_page_at(shard.shard_id, shard.head, &from, PLAN_END, PLAN_PAGE)
311                    .map_err(Failure::observer)?;
312                for (_key, bytes) in page.entries {
313                    let plan: MergePlan = serde_json::from_slice(&bytes)
314                        .map_err(|error| Failure::new("merge_row_corrupt", error.to_string()))?;
315                    if plan.state == MergeState::Open {
316                        open.push(plan);
317                    }
318                }
319                match page.next_from {
320                    Some(next) => from = next,
321                    None => break,
322                }
323            }
324        }
325        Ok(open)
326    }
327
328    pub(super) fn merge_service_selection(&self) -> Result<Captured, Failure> {
329        let record = self
330            .refs
331            .get(SERVICE)
332            .ok_or_else(|| Failure::new("service_missing", "merge control namespace is missing"))?;
333        Ok(Captured {
334            branch: SERVICE.to_owned(),
335            version: version(record),
336            shards: record.shards.clone(),
337        })
338    }
339
340    pub(super) fn confirm_merge_row(&mut self) -> Result<(), Failure> {
341        let at = self.merge_service_selection()?;
342        self.refs
343            .confirm_version(SERVICE, at.version.generation, at.version.sequence)
344            .map_err(|error| Failure::new("durability_unknown", error.to_string()))?;
345        self.publish_record(SERVICE)
346    }
347
348    fn write_merge_row(&mut self, plan: &MergePlan) -> Result<(), Failure> {
349        self.write_merge_service_rows(&[(
350            plan_key(&plan.id),
351            serde_json::to_vec(plan).map_err(Failure::storage)?,
352        )])
353    }
354
355    pub(super) fn write_merge_service_rows(
356        &mut self,
357        rows: &[(Vec<u8>, Vec<u8>)],
358    ) -> Result<(), Failure> {
359        let before = self.merge_service_selection()?.version;
360        let branch =
361            open_branch(SERVICE, &self.refs, &self.registry).map_err(Failure::branch_commit)?;
362        for (key, value) in rows {
363            branch
364                .put(self.writer.shard_for(key), key, value)
365                .map_err(Failure::branch)?;
366        }
367        self.writer
368            .commit(
369                &branch,
370                &self.registry,
371                CommitRequest {
372                    durability: CommitDurability::Durable {
373                        refs: &mut self.refs,
374                    },
375                    extra_parents: &[],
376                    timestamp: current_timestamp(),
377                    note: None,
378                },
379            )
380            .map_err(|error| {
381                Failure::new(
382                    if self
383                        .refs
384                        .get(SERVICE)
385                        .is_some_and(|record| version(record) != before)
386                    {
387                        "durability_unknown"
388                    } else {
389                        "merge_write_refused"
390                    },
391                    error.to_string(),
392                )
393            })?;
394        self.publish_record(SERVICE)
395    }
396}
397
398pub(super) fn holder_name(id: &str, slot: &str) -> String {
399    format!("@haem/merge/{}/{slot}", Hash::of(id.as_bytes()))
400}
401
402/// Where every merge plan row starts in the service namespace.
403const PLAN_PREFIX: &[u8] = b"j\0merge\0";
404
405/// Just past every key [`PLAN_PREFIX`] starts.
406const PLAN_END: &[u8] = b"j\0merge\x01";
407
408/// Plan rows read in one page of one shard.
409const PLAN_PAGE: NonZeroUsize = NonZeroUsize::MIN.saturating_add(255);
410
411fn plan_key(id: &str) -> Vec<u8> {
412    format!("j\0merge\0{}", Hash::of(id.as_bytes())).into_bytes()
413}
414
415#[cfg(test)]
416#[path = "merge_holders_tests.rs"]
417mod tests;
haematite_cli/serve/storage/merge_holders.rs

haematite_cli/serve/storage/
merge_holders.rs
