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
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
//! `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. Its stage rows stay with it too: the merged
//! commit carries one stage row of its own naming the source at the version
//! merged and listing, in order, every stage row the source committed since
//! its fork (STOR-050 R1, serve/storage/lineage.rs). 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::lineage::{self, Found};
use super::mutations::PreparedWrite;
use super::review::diff_failure;
use super::{Captured, Failure, Storage, branch_name, version};
use crate::serve::protocol::{BranchMerge, Version};
/// Keys the source keeps: its receipts, its own control rows, the author
/// rows of its own requests and its stage rows. The merge's own author row
/// and stage row are the target's.
const KEPT_BY_SOURCE: [&[u8]; 4] = [
b"r\0",
b"b\0",
crate::serve::contracts::attribution::TAG,
lineage::TAG,
];
/// 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, and
/// the merged commit's stage row at `next` when the source has stages.
///
/// # 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,
next: Version,
) -> 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();
let mut found = Found::default();
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 {
// Added only: the source never rewrites a stage row, and
// the target's at the anchor are the source's too.
if let ObservedChange::Added { key, value } = &change
&& key.starts_with(lineage::TAG)
{
found.add(&merge.from, key, value)?;
}
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,
}
}
}
let source = Captured {
branch: merge.from.clone(),
version: head,
shards: source.shards.clone(),
};
writes.extend(lineage::merged_write(&self.reader, &source, found, next)?);
Ok(writes)
}
}