haematite_cli/serve/storage/
merge_journal.rs

1//! Immutable owner choices at privately held merge inputs.
2
3use std::num::NonZeroUsize;
4
5use haematite::tree::Hash;
6use serde::{Deserialize, Serialize};
7
8use super::reads::{PageLimits, bounded_key_page};
9use super::review::{choice, cursor};
10use super::{
11    Captured, Failure, MergePin, MergePlan, MergeState, Storage, branch_name, document_key, edges,
12    identifier, reads,
13};
14use crate::serve::protocol::Node;
15
16const TOKEN_PREFIX: &str = "merge-choice-v1:";
17
18#[path = "merge_coverage.rs"]
19pub(super) mod coverage;
20pub use coverage::{MergeCoverage, MissingChoice};
21
22#[path = "merge_candidate.rs"]
23mod candidate;
24pub use candidate::{CandidateAction, CandidateAdvance};
25
26#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
27#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
28pub enum MergeOwner {
29    Document {
30        collection: String,
31        document_id: String,
32    },
33    Edge {
34        relation: String,
35        from: Node,
36        to: Node,
37    },
38}
39
40impl MergeOwner {
41    fn key(&self) -> Result<Vec<u8>, Failure> {
42        match self {
43            Self::Document {
44                collection,
45                document_id,
46            } => {
47                identifier(collection)?;
48                identifier(document_id)?;
49                Ok(document_key(collection, document_id))
50            }
51            Self::Edge { relation, from, to } => edges::keys(from, relation, to).map(|keys| keys.0),
52        }
53    }
54}
55
56#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
57#[serde(rename_all = "snake_case")]
58pub enum MergeChoice {
59    TakeSource,
60    TakeTarget,
61    RemoveEdge,
62}
63
64#[derive(Debug, Clone, Serialize, Deserialize)]
65#[serde(deny_unknown_fields)]
66pub struct MergeChoose {
67    pub merge_id: String,
68    pub owner: MergeOwner,
69    pub token: String,
70    pub choice: MergeChoice,
71}
72
73#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
74#[serde(deny_unknown_fields)]
75pub struct MergeChoiceRow {
76    pub owner: MergeOwner,
77    pub choice: MergeChoice,
78}
79
80#[derive(Debug, Serialize)]
81pub struct RecordedChoice {
82    pub token: String,
83    pub row: MergeChoiceRow,
84}
85
86#[derive(Debug, Serialize)]
87pub struct MergeChoices {
88    pub merge_id: String,
89    pub revision: u64,
90    pub entries: Vec<RecordedChoice>,
91    pub next_after: Option<String>,
92}
93
94#[derive(Serialize, Deserialize)]
95#[serde(deny_unknown_fields)]
96struct ChoiceCursor {
97    merge_id: String,
98    revision: u64,
99    token: String,
100}
101
102#[derive(Serialize, Deserialize)]
103#[serde(deny_unknown_fields)]
104struct Revision {
105    number: u64,
106}
107
108struct HeldInputs {
109    origin: Captured,
110    target: Captured,
111    source: Captured,
112    pins: choice::Pins,
113}
114
115struct Triple {
116    ancestor: Option<Vec<u8>>,
117    target: Option<Vec<u8>>,
118    source: Option<Vec<u8>>,
119}
120
121impl Triple {
122    fn token(&self, pins: &choice::Pins, key: &[u8]) -> Result<String, Failure> {
123        pins.token(
124            key,
125            self.ancestor.as_deref(),
126            self.target.as_deref(),
127            self.source.as_deref(),
128        )
129    }
130
131    fn unchanged(&self) -> bool {
132        self.ancestor == self.target && self.ancestor == self.source
133    }
134}
135
136impl Storage {
137    /// Admit one immutable choice from owner bytes at the held input versions.
138    ///
139    /// # Errors
140    /// Refuses invalid owners, tokens, moved pins, unresolved endpoints and conflicting choices.
141    pub fn merge_choose(&mut self, request: &MergeChoose) -> Result<MergeChoiceRow, Failure> {
142        identifier(&request.merge_id)?;
143        validate_token(&request.token)?;
144        let key = request.owner.key()?;
145        validate_choice_kind(&request.owner, request.choice)?;
146        let plan = self.open_merge_plan(&request.merge_id)?;
147        let inputs = self.choice_inputs(&plan)?;
148        let values = self.choice_values(&inputs, &key)?;
149        if values.token(&inputs.pins, &key)? != request.token {
150            return Err(Failure::new(
151                "choice_token_mismatch",
152                "token differs from held owner bytes",
153            ));
154        }
155        let service = self.merge_service_selection()?;
156        let revision = self.choice_revision(&service, &request.merge_id)?;
157        let row = MergeChoiceRow {
158            owner: request.owner.clone(),
159            choice: request.choice,
160        };
161        if let Some(existing) = self.read_choice_row(&service, &request.merge_id, &request.token)? {
162            if revision == 0 || existing.owner != row.owner {
163                return Err(Failure::new(
164                    "choice_row_corrupt",
165                    "choice identity or revision differs",
166                ));
167            }
168            if existing != row {
169                return Err(Failure::new(
170                    "choice_conflict",
171                    "token already has a different choice",
172                ));
173            }
174            self.confirm_merge_row()?;
175            return Ok(existing);
176        }
177        self.admit_owner_choice(request, &inputs, &values, &service)?;
178        let next = revision.checked_add(1).ok_or_else(|| {
179            Failure::new(
180                "choice_revision_exhausted",
181                "choice journal revision cannot advance",
182            )
183        })?;
184        self.write_merge_service_rows(&[
185            (
186                choice_key(&request.merge_id, &request.token),
187                serde_json::to_vec(&row).map_err(Failure::storage)?,
188            ),
189            (
190                revision_key(&request.merge_id),
191                serde_json::to_vec(&Revision { number: next }).map_err(Failure::storage)?,
192            ),
193        ])?;
194        Ok(row)
195    }
196
197    /// Seek one merge's journal, refusing a cursor if that journal changed.
198    ///
199    /// # Errors
200    /// Refuses missing or abandoned plans, malformed rows and invalid or stale cursors.
201    pub fn merge_choices(
202        &self,
203        id: &str,
204        after: Option<&str>,
205        limit: NonZeroUsize,
206    ) -> Result<MergeChoices, Failure> {
207        self.open_merge_plan(id)?;
208        let service = self.merge_service_selection()?;
209        let revision = self.choice_revision(&service, id)?;
210        let prefix = choice_prefix(id);
211        let mut from = prefix.clone();
212        if let Some(after) = after {
213            let decoded: ChoiceCursor = cursor::decode(after)?;
214            if decoded.merge_id != id {
215                return Err(Failure::new(
216                    "invalid_continuation",
217                    "cursor belongs to another merge",
218                ));
219            }
220            validate_token(&decoded.token)
221                .map_err(|error| Failure::new("invalid_continuation", error.message))?;
222            if decoded.revision != revision {
223                return Err(Failure::new(
224                    "choice_journal_moved",
225                    "journal changed after this cursor",
226                ));
227            }
228            if self
229                .read_choice_row(&service, id, &decoded.token)?
230                .is_none()
231            {
232                return Err(Failure::new(
233                    "invalid_continuation",
234                    "cursor token is not in this journal",
235                ));
236            }
237            from.extend_from_slice(decoded.token.as_bytes());
238            from.push(0);
239        }
240        let (entries, next) = bounded_key_page(
241            &self.reader,
242            &service,
243            &prefix,
244            &from,
245            PageLimits {
246                rows: limit,
247                bytes: NonZeroUsize::MAX,
248            },
249            |token, bytes| {
250                Ok(RecordedChoice {
251                    row: decode_row(&token, &bytes)?,
252                    token,
253                })
254            },
255            |entry| {
256                serde_json::to_vec(entry)
257                    .map(|bytes| bytes.len())
258                    .map_err(Failure::storage)
259            },
260        )?;
261        if revision == 0 && !entries.is_empty() {
262            return Err(Failure::new(
263                "choice_row_corrupt",
264                "journal rows have no revision",
265            ));
266        }
267        let next_after = next
268            .map(|token| {
269                cursor::encode(&ChoiceCursor {
270                    merge_id: id.to_owned(),
271                    revision,
272                    token,
273                })
274            })
275            .transpose()?;
276        Ok(MergeChoices {
277            merge_id: id.to_owned(),
278            revision,
279            entries,
280            next_after,
281        })
282    }
283
284    fn open_merge_plan(&self, id: &str) -> Result<MergePlan, Failure> {
285        let plan = self
286            .merge_plan(id)?
287            .ok_or_else(|| Failure::new("merge_missing", "merge id has no control row"))?;
288        if plan.state != MergeState::Open {
289            return Err(Failure::new("merge_abandoned", "merge has been abandoned"));
290        }
291        Ok(plan)
292    }
293
294    fn choice_inputs(&self, plan: &MergePlan) -> Result<HeldInputs, Failure> {
295        let origin = self.merge_held_pin(&plan.id, "origin", &plan.origin)?;
296        let target = self.merge_held_pin(&plan.id, "target", &plan.target)?;
297        let source = self.merge_held_pin(&plan.id, "source", &plan.source)?;
298        self.check_choice_live_pin(&plan.target, &target)?;
299        self.check_choice_live_pin(&plan.source, &source)?;
300        let pins = choice::Pins::new(&origin, &target, &source)?;
301        Ok(HeldInputs {
302            origin,
303            target,
304            source,
305            pins,
306        })
307    }
308
309    fn check_choice_live_pin(&self, pin: &MergePin, held: &Captured) -> Result<(), Failure> {
310        self.check_merge_live_pin(pin)?;
311        let live = self
312            .refs
313            .get(&branch_name(&pin.branch))
314            .ok_or_else(|| Failure::new("merge_pins_moved", "merge input branch is missing"))?;
315        if !live
316            .shards
317            .iter()
318            .map(|shard| (shard.shard_id, shard.head))
319            .eq(held.shards.iter().map(|shard| (shard.shard_id, shard.head)))
320        {
321            return Err(Failure::new(
322                "merge_pins_moved",
323                "live roots differ from held inputs",
324            ));
325        }
326        Ok(())
327    }
328
329    fn choice_values(&self, inputs: &HeldInputs, key: &[u8]) -> Result<Triple, Failure> {
330        Ok(Triple {
331            ancestor: reads::get_bytes(&self.reader, &inputs.origin, key)?,
332            target: reads::get_bytes(&self.reader, &inputs.target, key)?,
333            source: reads::get_bytes(&self.reader, &inputs.source, key)?,
334        })
335    }
336
337    fn admit_owner_choice(
338        &self,
339        request: &MergeChoose,
340        inputs: &HeldInputs,
341        values: &Triple,
342        service: &Captured,
343    ) -> Result<(), Failure> {
344        if let MergeOwner::Edge { from, to, .. } = &request.owner {
345            let from_absent = self.endpoint_absent(request, inputs, service, from)?;
346            let to_absent = self.endpoint_absent(request, inputs, service, to)?;
347            let dangling = from_absent || to_absent;
348            if dangling {
349                if request.choice != MergeChoice::RemoveEdge {
350                    return Err(Failure::new(
351                        "choice_requires_remove_edge",
352                        "dangling edge requires removal",
353                    ));
354                }
355                if values.target.is_some() || values.source.is_some() {
356                    return Ok(());
357                }
358            }
359            if request.choice == MergeChoice::RemoveEdge {
360                return Err(Failure::new(
361                    "choice_not_needed",
362                    "edge has no settled dangling endpoint",
363                ));
364            }
365        }
366        if values.unchanged() {
367            return Err(Failure::new(
368                "choice_not_needed",
369                "owner has no difference at the pins",
370            ));
371        }
372        if values.target == values.source {
373            return Err(Failure::new(
374                "choice_same_on_both",
375                "owner is the same on both sides",
376            ));
377        }
378        Ok(())
379    }
380
381    fn endpoint_absent(
382        &self,
383        request: &MergeChoose,
384        inputs: &HeldInputs,
385        service: &Captured,
386        node: &Node,
387    ) -> Result<bool, Failure> {
388        let key = document_key(&node.collection, &node.document_id);
389        let values = self.choice_values(inputs, &key)?;
390        if values.target.is_some() == values.source.is_some() {
391            return Ok(values.target.is_none());
392        }
393        let token = values.token(&inputs.pins, &key)?;
394        let row = self
395            .read_choice_row(service, &request.merge_id, &token)?
396            .ok_or_else(|| {
397                Failure::new(
398                    "choice_depends_on_endpoint",
399                    "endpoint document must be chosen first",
400                )
401            })?;
402        let owner = MergeOwner::Document {
403            collection: node.collection.clone(),
404            document_id: node.document_id.clone(),
405        };
406        if row.owner != owner {
407            return Err(Failure::new(
408                "choice_row_corrupt",
409                "endpoint choice names another owner",
410            ));
411        }
412        match row.choice {
413            MergeChoice::TakeSource => Ok(values.source.is_none()),
414            MergeChoice::TakeTarget => Ok(values.target.is_none()),
415            MergeChoice::RemoveEdge => Err(Failure::new(
416                "choice_row_corrupt",
417                "document choice removes an edge",
418            )),
419        }
420    }
421
422    fn read_choice_row(
423        &self,
424        service: &Captured,
425        id: &str,
426        token: &str,
427    ) -> Result<Option<MergeChoiceRow>, Failure> {
428        reads::get_bytes(&self.reader, service, &choice_key(id, token))?
429            .map(|bytes| decode_row(token, &bytes))
430            .transpose()
431    }
432
433    fn choice_revision(&self, service: &Captured, id: &str) -> Result<u64, Failure> {
434        let stored = reads::get_bytes(&self.reader, service, &revision_key(id))?;
435        // A merge has no journal revision before its first choice.
436        stored.map_or(Ok(0), |bytes| {
437            let revision: Revision = serde_json::from_slice(&bytes)
438                .map_err(|error| Failure::new("choice_row_corrupt", error.to_string()))?;
439            if revision.number == 0 {
440                return Err(Failure::new(
441                    "choice_row_corrupt",
442                    "stored journal revision is zero",
443                ));
444            }
445            Ok(revision.number)
446        })
447    }
448}
449
450fn validate_token(token: &str) -> Result<(), Failure> {
451    let digest = token.strip_prefix(TOKEN_PREFIX).ok_or_else(|| {
452        Failure::new("choice_token_invalid", "choice token has an invalid prefix")
453    })?;
454    if digest.len() != 64
455        || !digest
456            .bytes()
457            .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
458    {
459        return Err(Failure::new(
460            "choice_token_invalid",
461            "choice token must contain a canonical digest",
462        ));
463    }
464    Ok(())
465}
466
467fn validate_choice_kind(owner: &MergeOwner, selected: MergeChoice) -> Result<(), Failure> {
468    if matches!(owner, MergeOwner::Document { .. }) && selected == MergeChoice::RemoveEdge {
469        return Err(Failure::new(
470            "choice_kind_mismatch",
471            "remove_edge requires an edge owner",
472        ));
473    }
474    Ok(())
475}
476
477fn decode_row(token: &str, bytes: &[u8]) -> Result<MergeChoiceRow, Failure> {
478    validate_token(token).map_err(|error| Failure::new("choice_row_corrupt", error.message))?;
479    let row: MergeChoiceRow = serde_json::from_slice(bytes)
480        .map_err(|error| Failure::new("choice_row_corrupt", error.to_string()))?;
481    row.owner
482        .key()
483        .map_err(|error| Failure::new("choice_row_corrupt", error.message))?;
484    validate_choice_kind(&row.owner, row.choice)
485        .map_err(|error| Failure::new("choice_row_corrupt", error.message))?;
486    Ok(row)
487}
488
489fn choice_prefix(id: &str) -> Vec<u8> {
490    format!("j\0merge-choice\0{}\0", Hash::of(id.as_bytes())).into_bytes()
491}
492
493fn choice_key(id: &str, token: &str) -> Vec<u8> {
494    let mut key = choice_prefix(id);
495    key.extend_from_slice(token.as_bytes());
496    key
497}
498
499fn revision_key(id: &str) -> Vec<u8> {
500    format!("j\0merge-choice-revision\0{}", Hash::of(id.as_bytes())).into_bytes()
501}
502
503#[cfg(test)]
504#[path = "merge_journal_tests.rs"]
505mod tests;
haematite_cli/serve/storage/merge_journal.rs

haematite_cli/serve/storage/
merge_journal.rs
