1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
//! `branch.merge`: a reviewed branch merged into the branch the mutation
//! names, in that branch's one ordinary commit with its receipt
//! (docs/design/storage/serve-merge.md, what must hold 4).
//!
//! Only a merge nothing has to choose is served: every shard of the target
//! must still be at the source's fork anchor, so the merged tree is the
//! source's own and no conflict can exist (what must hold 1). A target that
//! moved since the fork is refused `merge_target_moved`; a source that moved
//! since its review is refused `merge_source_moved`, never reviewed again in
//! silence. The source's changes are carried as they are, its derived rows
//! with them, since they were derived over that same base (what must hold 2);
//! its receipts and its own control rows (fork provenance, merge markers)
//! are its own and stay with it. The source is left as it is (what must
//! hold 5). The changes are read in bounded pages, but the commit holds every
//! one of them, so its memory grows with the size of the source's change.
use std::num::NonZeroUsize;
use haematite::db::ObservedChange;
use haematite::tree::DiffBudget;
use super::mutations::PreparedWrite;
use super::review::diff_failure;
use super::{Captured, Failure, Storage, branch_name, version};
use crate::serve::protocol::BranchMerge;
/// Keys the source keeps: its receipts and its own control rows.
const KEPT_BY_SOURCE: [&[u8]; 2] = [b"r\0", b"b\0"];
/// One page of the source's changes.
const PAGE: DiffBudget = DiffBudget {
max_changes: NonZeroUsize::MIN.saturating_add(4095),
max_nodes: NonZeroUsize::MIN.saturating_add(65_535),
};
impl Storage {
/// The writes that bring the target at `at` to `merge.from`'s tree.
///
/// # Errors
///
/// `merge_self`, `branch_missing`, `discard_pending`, `merge_source_moved`
/// or `merge_target_moved` by name, or a failure reading the changes.
pub(super) fn merge_writes(
&self,
at: &Captured,
merge: &BranchMerge,
) -> Result<Vec<PreparedWrite>, Failure> {
if merge.from == at.branch {
return Err(Failure::new(
"merge_self",
"a branch cannot be merged into itself",
));
}
let source = self
.refs
.get(&branch_name(&merge.from))
.ok_or_else(|| Failure::new("branch_missing", &merge.from))?;
if self.pending_discards.contains(&source.name) {
return Err(Failure::new(
"discard_pending",
"the source's removal requires receipt reconciliation",
));
}
let head = version(source);
if head != merge.reviewed {
return Err(Failure::new(
"merge_source_moved",
format!(
"{} is at {}.{}, not the reviewed {}.{}; review it again",
merge.from,
head.generation,
head.sequence,
merge.reviewed.generation,
merge.reviewed.sequence
),
));
}
let mut target = at.shards.clone();
target.sort_unstable_by_key(|shard| shard.shard_id);
let mut from = source.shards.clone();
from.sort_unstable_by_key(|shard| shard.shard_id);
if target.len() != from.len()
|| target.iter().zip(&from).any(|(target, from)| {
(target.shard_id, target.head) != (from.shard_id, from.fork_anchor)
})
{
return Err(Failure::new(
"merge_target_moved",
format!(
"{} has changed since {} was forked from it, so the merge would have to choose",
at.branch, merge.from
),
));
}
let mut writes = Vec::new();
for (target, from) in target.iter().zip(&from) {
let mut after: Option<Vec<u8>> = None;
loop {
let page = self
.reader
.changes_page(
target.shard_id,
target.head,
from.head,
after.as_deref(),
PAGE,
)
.map_err(diff_failure)?;
for change in page.entries {
if KEPT_BY_SOURCE
.iter()
.any(|prefix| change.key().starts_with(prefix))
{
continue;
}
writes.push(match change {
ObservedChange::Added { key, value }
| ObservedChange::Modified {
key,
new_value: value,
..
} => (key, Some(value)),
ObservedChange::Removed { key, .. } => (key, None),
});
}
match page.next_after {
Some(next) => after = Some(next),
None => break,
}
}
}
Ok(writes)
}
}