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                },
238            )
239            .map_err(|error| {
240                Failure::new(
241                    if self.refs.get(name).is_some() {
242                        "durability_unknown"
243                    } else {
244                        "merge_hold_refused"
245                    },
246                    error.to_string(),
247                )
248            })?;
249        Ok(())
250    }
251
252    pub(super) fn merge_held_pin(
253        &self,
254        id: &str,
255        slot: &str,
256        pin: &MergePin,
257    ) -> Result<Captured, Failure> {
258        let name = holder_name(id, slot);
259        let record = self.refs.get(&name).ok_or_else(|| {
260            Failure::new("merge_holder_missing", "merge input reference is missing")
261        })?;
262        let at = Captured {
263            branch: name,
264            version: version(record),
265            shards: record.shards.clone(),
266        };
267        let bytes = reads::get_bytes(&self.reader, &at, HOLDER_KEY)?.ok_or_else(|| {
268            Failure::new(
269                "merge_holder_corrupt",
270                "merge input reference has no identity",
271            )
272        })?;
273        let held: MergePin = serde_json::from_slice(&bytes)
274            .map_err(|error| Failure::new("merge_holder_corrupt", error.to_string()))?;
275        if held != *pin || record.seq != 0 {
276            return Err(Failure::new(
277                "merge_holder_conflict",
278                "merge input identity changed",
279            ));
280        }
281        Ok(Captured {
282            branch: pin.branch.clone(),
283            version: pin.version,
284            shards: record
285                .shards
286                .iter()
287                .map(|shard| super::BranchShardRef {
288                    shard_id: shard.shard_id,
289                    fork_anchor: shard.fork_anchor,
290                    head: shard.fork_anchor,
291                })
292                .collect(),
293        })
294    }
295
296    /// Every merge plan still open (STOR-003 HR1): each shard's plan rows at
297    /// the service's roots, a page at a time. Abandoned plans rely on nothing.
298    ///
299    /// # Errors
300    /// `merge_row_corrupt` for a row that does not decode; storage failures.
301    pub(super) fn open_merge_plans(&self) -> Result<Vec<MergePlan>, Failure> {
302        let at = self.merge_service_selection()?;
303        let mut open = Vec::new();
304        for shard in &at.shards {
305            let mut from = PLAN_PREFIX.to_vec();
306            loop {
307                let page = self
308                    .reader
309                    .range_page_at(shard.shard_id, shard.head, &from, PLAN_END, PLAN_PAGE)
310                    .map_err(Failure::observer)?;
311                for (_key, bytes) in page.entries {
312                    let plan: MergePlan = serde_json::from_slice(&bytes)
313                        .map_err(|error| Failure::new("merge_row_corrupt", error.to_string()))?;
314                    if plan.state == MergeState::Open {
315                        open.push(plan);
316                    }
317                }
318                match page.next_from {
319                    Some(next) => from = next,
320                    None => break,
321                }
322            }
323        }
324        Ok(open)
325    }
326
327    pub(super) fn merge_service_selection(&self) -> Result<Captured, Failure> {
328        let record = self
329            .refs
330            .get(SERVICE)
331            .ok_or_else(|| Failure::new("service_missing", "merge control namespace is missing"))?;
332        Ok(Captured {
333            branch: SERVICE.to_owned(),
334            version: version(record),
335            shards: record.shards.clone(),
336        })
337    }
338
339    pub(super) fn confirm_merge_row(&mut self) -> Result<(), Failure> {
340        let at = self.merge_service_selection()?;
341        self.refs
342            .confirm_version(SERVICE, at.version.generation, at.version.sequence)
343            .map_err(|error| Failure::new("durability_unknown", error.to_string()))?;
344        self.publish_record(SERVICE)
345    }
346
347    fn write_merge_row(&mut self, plan: &MergePlan) -> Result<(), Failure> {
348        self.write_merge_service_rows(&[(
349            plan_key(&plan.id),
350            serde_json::to_vec(plan).map_err(Failure::storage)?,
351        )])
352    }
353
354    pub(super) fn write_merge_service_rows(
355        &mut self,
356        rows: &[(Vec<u8>, Vec<u8>)],
357    ) -> Result<(), Failure> {
358        let before = self.merge_service_selection()?.version;
359        let branch =
360            open_branch(SERVICE, &self.refs, &self.registry).map_err(Failure::branch_commit)?;
361        for (key, value) in rows {
362            branch
363                .put(self.writer.shard_for(key), key, value)
364                .map_err(Failure::branch)?;
365        }
366        self.writer
367            .commit(
368                &branch,
369                &self.registry,
370                CommitRequest {
371                    durability: CommitDurability::Durable {
372                        refs: &mut self.refs,
373                    },
374                    extra_parents: &[],
375                    timestamp: current_timestamp(),
376                },
377            )
378            .map_err(|error| {
379                Failure::new(
380                    if self
381                        .refs
382                        .get(SERVICE)
383                        .is_some_and(|record| version(record) != before)
384                    {
385                        "durability_unknown"
386                    } else {
387                        "merge_write_refused"
388                    },
389                    error.to_string(),
390                )
391            })?;
392        self.publish_record(SERVICE)
393    }
394}
395
396pub(super) fn holder_name(id: &str, slot: &str) -> String {
397    format!("@haem/merge/{}/{slot}", Hash::of(id.as_bytes()))
398}
399
400/// Where every merge plan row starts in the service namespace.
401const PLAN_PREFIX: &[u8] = b"j\0merge\0";
402
403/// Just past every key [`PLAN_PREFIX`] starts.
404const PLAN_END: &[u8] = b"j\0merge\x01";
405
406/// Plan rows read in one page of one shard.
407const PLAN_PAGE: NonZeroUsize = NonZeroUsize::MIN.saturating_add(255);
408
409fn plan_key(id: &str) -> Vec<u8> {
410    format!("j\0merge\0{}", Hash::of(id.as_bytes())).into_bytes()
411}
412
413#[cfg(test)]
414#[path = "merge_holders_tests.rs"]
415mod tests;
haematite_cli/serve/storage/merge_holders.rs

haematite_cli/serve/storage/
merge_holders.rs
