laminar_core/checkpoint/committed_checkpoint/
view_log_references.rs

1//! View-log segments across the participants of one cut. A participant that adopts the log of
2//! one that left lists the leaver's segments by reference, so a cut is checked as a whole: no
3//! segment is listed twice, and no departed segment is lost.
4
5use std::collections::btree_map::Entry;
6
7use super::{BTreeMap, CheckpointManifest};
8use crate::checkpoint::{
9    view_log_segment_limit, ByteRange, CheckpointAssignmentFence, StateChunkId, ViewLogSegment,
10};
11
12/// One segment a survivor adopts from a participant that left, with the view-log object that
13/// holds its bytes.
14#[derive(Debug, Clone, PartialEq, Eq)]
15pub struct AdoptedSegment {
16    /// The segment as the departed participant listed it.
17    pub segment: ViewLogSegment,
18    /// Exact length of the object holding the segment.
19    pub object_length: u64,
20    /// SHA-256 of that whole object.
21    pub object_sha256: String,
22}
23
24/// The segments of one view a survivor adopts, in the departed log's order.
25#[derive(Debug, Clone, PartialEq, Eq)]
26pub struct AdoptedViewLog {
27    /// Catalog name of the view.
28    pub view: String,
29    /// Adopted segments.
30    pub segments: Vec<AdoptedSegment>,
31}
32
33/// The view logs `adopter` adopts when the participants of `cut` that `target` lacks leave.
34///
35/// Every participant derives the same plan from the same cut and target. Each departed log goes
36/// whole to the survivor (a participant of both) holding the fewest segments so far, ties to
37/// the lowest node id, in order of departed participant and then view.
38///
39/// # Errors
40/// Returns an error when departed logs have no survivor to adopt them, when a survivor would
41/// hold more segments than its share of the cluster budget, or when a departed segment's object
42/// is untracked.
43pub fn adopted_view_logs(
44    cut: &[CheckpointManifest],
45    target: &CheckpointAssignmentFence,
46    adopter: u64,
47) -> Result<Vec<AdoptedViewLog>, String> {
48    plan_adoption(cut, target, Some(adopter))
49}
50
51/// Whether the survivors of `target` can adopt the logs of the participants of `cut` it lacks.
52///
53/// # Errors
54/// Returns the error [`adopted_view_logs`] would return to any survivor.
55pub fn check_view_log_adoption(
56    cut: &[CheckpointManifest],
57    target: &CheckpointAssignmentFence,
58) -> Result<(), String> {
59    plan_adoption(cut, target, None).map(drop)
60}
61
62fn plan_adoption(
63    cut: &[CheckpointManifest],
64    target: &CheckpointAssignmentFence,
65    adopter: Option<u64>,
66) -> Result<Vec<AdoptedViewLog>, String> {
67    let limit = view_log_segment_limit(Some(target.participants.len()));
68    let mut manifests: Vec<&CheckpointManifest> = cut.iter().collect();
69    manifests.sort_unstable_by_key(|manifest| manifest.participant_id);
70    let (survivors, departed): (Vec<_>, Vec<_>) = manifests
71        .into_iter()
72        .partition(|manifest| target.contains(manifest.participant_id));
73    let mut load: BTreeMap<u64, usize> = survivors
74        .iter()
75        .map(|manifest| (manifest.participant_id, segment_count(manifest)))
76        .collect();
77    let mut logs = Vec::new();
78    for manifest in departed {
79        for log in manifest
80            .view_logs
81            .iter()
82            .filter(|log| !log.segments.is_empty())
83        {
84            let (&chosen, held) = load
85                .iter_mut()
86                .min_by_key(|(id, held)| (**held, **id))
87                .ok_or_else(|| {
88                    format!(
89                        "participant {} left with view logs, and no surviving participant can \
90                         adopt them",
91                        manifest.participant_id
92                    )
93                })?;
94            *held = held.saturating_add(log.segments.len());
95            if *held > limit {
96                return Err(format!(
97                    "participant {chosen} would hold {held} view-log segments after adopting \
98                     view '{}' from participant {}; its share of the cluster budget is {limit}",
99                    log.view, manifest.participant_id
100                ));
101            }
102            if adopter != Some(chosen) {
103                continue;
104            }
105            let segments = log
106                .segments
107                .iter()
108                .map(|segment| adopted_segment(manifest, segment))
109                .collect::<Result<_, _>>()?;
110            logs.push(AdoptedViewLog {
111                view: log.view.clone(),
112                segments,
113            });
114        }
115    }
116    Ok(logs)
117}
118
119fn segment_count(manifest: &CheckpointManifest) -> usize {
120    manifest
121        .view_logs
122        .iter()
123        .map(|log| log.segments.len())
124        .sum()
125}
126
127fn adopted_segment(
128    manifest: &CheckpointManifest,
129    segment: &ViewLogSegment,
130) -> Result<AdoptedSegment, String> {
131    let written = manifest
132        .view_log
133        .iter()
134        .map(|object| (object.chunk, object.object_length, &object.sha256));
135    let referenced = manifest
136        .referenced_view_logs
137        .iter()
138        .map(|object| (object.chunk, object.object_length, &object.sha256));
139    let (_, object_length, object_sha256) = written
140        .chain(referenced)
141        .find(|(chunk, _, _)| *chunk == segment.chunk)
142        .ok_or_else(|| {
143            format!(
144                "participant {} lists view-log segment {} of untracked object {:?}",
145                manifest.participant_id, segment.sequence, segment.chunk
146            )
147        })?;
148    Ok(AdoptedSegment {
149        segment: segment.clone(),
150        object_length,
151        object_sha256: object_sha256.clone(),
152    })
153}
154
155/// Every view-log segment of a cut is listed once, and every participant that writes or
156/// references a view-log object agrees on its length and digest.
157///
158/// # Errors
159/// Returns an error naming the first segment listed twice or the first object participants
160/// disagree on.
161pub fn validate_cut_view_logs<'a>(
162    manifests: impl IntoIterator<Item = &'a CheckpointManifest>,
163) -> Result<(), String> {
164    let mut objects = BTreeMap::<StateChunkId, (u64, &str)>::new();
165    let mut segments = BTreeMap::<(StateChunkId, ByteRange), &str>::new();
166    for manifest in manifests {
167        let written = manifest
168            .view_log
169            .iter()
170            .map(|object| (object.chunk, object.object_length, object.sha256.as_str()));
171        let referenced = manifest
172            .referenced_view_logs
173            .iter()
174            .map(|object| (object.chunk, object.object_length, object.sha256.as_str()));
175        for (chunk, length, sha256) in written.chain(referenced) {
176            match objects.entry(chunk) {
177                Entry::Vacant(entry) => {
178                    entry.insert((length, sha256));
179                }
180                Entry::Occupied(entry) if *entry.get() != (length, sha256) => {
181                    return Err(format!(
182                        "participants disagree on the length or digest of view-log object {chunk:?}"
183                    ));
184                }
185                Entry::Occupied(_) => {}
186            }
187        }
188        for log in &manifest.view_logs {
189            for segment in &log.segments {
190                if segments
191                    .insert((segment.chunk, segment.range), &log.view)
192                    .is_some()
193                {
194                    return Err(format!(
195                        "view-log object {:?} bytes {}+{} are listed twice in one cut",
196                        segment.chunk, segment.range.offset, segment.range.length
197                    ));
198                }
199            }
200        }
201    }
202    Ok(())
203}
204
205/// Every segment the departed participants' view logs held at the predecessor cut is listed
206/// exactly once, by the same view, in this cut.
207///
208/// # Errors
209/// Returns an error naming the first departed segment this cut lost or lists twice.
210pub fn validate_departed_view_logs<'a>(
211    departed: impl IntoIterator<Item = &'a CheckpointManifest>,
212    cut: impl IntoIterator<Item = &'a CheckpointManifest>,
213) -> Result<(), String> {
214    let mut listed = BTreeMap::<(&str, StateChunkId, ByteRange, &str), u32>::new();
215    for log in cut.into_iter().flat_map(|manifest| &manifest.view_logs) {
216        for segment in &log.segments {
217            let key = (
218                log.view.as_str(),
219                segment.chunk,
220                segment.range,
221                segment.sha256.as_str(),
222            );
223            *listed.entry(key).or_default() += 1;
224        }
225    }
226    for manifest in departed {
227        for log in &manifest.view_logs {
228            for segment in &log.segments {
229                let key = (
230                    log.view.as_str(),
231                    segment.chunk,
232                    segment.range,
233                    segment.sha256.as_str(),
234                );
235                let count = listed.get(&key).copied().unwrap_or(0);
236                if count != 1 {
237                    return Err(format!(
238                        "participant {} left, and view '{}' lists its segment {} of object \
239                         {:?} {count} times instead of once",
240                        manifest.participant_id, log.view, segment.sequence, segment.chunk
241                    ));
242                }
243            }
244        }
245    }
246    Ok(())
247}
248
249#[cfg(test)]
250mod tests {
251    use std::num::NonZeroU32;
252
253    use super::*;
254    use crate::checkpoint::{
255        checkpoint_sha256, NodeDataObject, ReferencedStateChunk, ViewLog, ViewLogSegment,
256    };
257
258    const OBJECT: &[u8] = b"rows of participant one";
259
260    fn chunk(participant_id: u64, checkpoint_id: u64) -> StateChunkId {
261        StateChunkId {
262            participant_id,
263            checkpoint_id,
264        }
265    }
266
267    fn segment(chunk: StateChunkId, sequence: u64) -> ViewLogSegment {
268        ViewLogSegment {
269            partition: 0,
270            sequence,
271            chunk,
272            range: ByteRange {
273                offset: sequence * 4,
274                length: 4,
275            },
276            sha256: checkpoint_sha256(&[u8::try_from(sequence).unwrap()]),
277            rows: 1,
278            event_time: None,
279        }
280    }
281
282    /// Participant `participant` at checkpoint 5, listing `segments` of view `v`; segments of
283    /// another checkpoint's object are references.
284    fn manifest(participant: u64, segments: Vec<ViewLogSegment>) -> CheckpointManifest {
285        let mut manifest = CheckpointManifest::new(5, 5);
286        manifest.participant_id = participant;
287        let own = chunk(participant, 5);
288        let mut referenced = BTreeMap::<StateChunkId, u32>::new();
289        for segment in segments.iter().filter(|segment| segment.chunk != own) {
290            *referenced.entry(segment.chunk).or_default() += 1;
291        }
292        manifest.view_log = segments
293            .iter()
294            .any(|segment| segment.chunk == own)
295            .then(|| NodeDataObject {
296                chunk: own,
297                object_length: 64,
298                sha256: checkpoint_sha256(OBJECT),
299            });
300        manifest.referenced_view_logs = referenced
301            .into_iter()
302            .map(|(chunk, count)| ReferencedStateChunk {
303                chunk,
304                object_length: 64,
305                sha256: checkpoint_sha256(OBJECT),
306                ref_count: NonZeroU32::new(count).unwrap(),
307            })
308            .collect();
309        manifest.view_logs = vec![ViewLog {
310            view: "v".into(),
311            next_sequence: 8,
312            frontier: None,
313            idle: false,
314            segments,
315        }];
316        manifest
317    }
318
319    #[test]
320    fn a_segment_is_listed_once_and_objects_agree_across_participants() {
321        let old = chunk(1, 3);
322        let first = manifest(1, vec![segment(old, 0)]);
323        let second = manifest(2, vec![segment(old, 1)]);
324        validate_cut_view_logs([&first, &second]).unwrap();
325
326        let twice = manifest(2, vec![segment(old, 0)]);
327        assert!(validate_cut_view_logs([&first, &twice])
328            .unwrap_err()
329            .contains("listed twice"));
330
331        let mut disagreeing = second;
332        disagreeing.referenced_view_logs[0].object_length = 65;
333        assert!(validate_cut_view_logs([&first, &disagreeing])
334            .unwrap_err()
335            .contains("disagree"));
336    }
337
338    fn fence(participants: &[u64]) -> CheckpointAssignmentFence {
339        let owners: Vec<u64> = participants.iter().copied().cycle().take(4).collect();
340        CheckpointAssignmentFence::from_owner_map(
341            9,
342            &owners,
343            participants
344                .iter()
345                .map(|&node_id| crate::checkpoint::CheckpointParticipant {
346                    node_id,
347                    boot_incarnation: uuid::Uuid::from_u128(u128::from(node_id)),
348                })
349                .collect(),
350        )
351        .unwrap()
352    }
353
354    fn with_view(mut manifest: CheckpointManifest, view: &str) -> CheckpointManifest {
355        manifest.view_logs[0].view = view.into();
356        manifest
357    }
358
359    #[test]
360    fn each_departed_log_goes_whole_to_the_least_loaded_survivor() {
361        let leaver = chunk(3, 4);
362        let mut departed = manifest(3, vec![segment(leaver, 0), segment(leaver, 1)]);
363        let mut second = with_view(manifest(3, vec![segment(leaver, 2)]), "w");
364        departed.view_logs.append(&mut second.view_logs);
365        // Survivor 1 holds two segments, survivor 2 one.
366        let one = manifest(1, vec![segment(chunk(1, 5), 0), segment(chunk(1, 5), 1)]);
367        let two = manifest(2, vec![segment(chunk(2, 5), 0)]);
368        let cut = [one, two, departed];
369        let target = fence(&[1, 2]);
370
371        // `v` (two segments) goes to survivor 2, which then holds three, so `w` goes to 1.
372        let to_two = adopted_view_logs(&cut, &target, 2).unwrap();
373        assert_eq!(to_two.len(), 1);
374        assert_eq!(to_two[0].view, "v");
375        assert_eq!(to_two[0].segments.len(), 2);
376        assert_eq!(to_two[0].segments[0].object_length, 64);
377        let to_one = adopted_view_logs(&cut, &target, 1).unwrap();
378        assert_eq!(to_one[0].view, "w");
379        assert_eq!(to_one[0].segments[0].segment, segment(leaver, 2));
380
381        // The same cut in another order makes the same plan.
382        let reordered = [cut[2].clone(), cut[0].clone(), cut[1].clone()];
383        assert_eq!(adopted_view_logs(&reordered, &target, 2).unwrap(), to_two);
384
385        // A participant joining in the target adopts nothing.
386        assert!(adopted_view_logs(&cut, &fence(&[1, 2, 4]), 4)
387            .unwrap()
388            .is_empty());
389    }
390
391    #[test]
392    fn departed_logs_need_a_survivor_with_room() {
393        let leaver = chunk(3, 4);
394        let cut = [manifest(3, vec![segment(leaver, 0)])];
395        assert!(adopted_view_logs(&cut, &fence(&[4]), 4)
396            .unwrap_err()
397            .contains("no surviving participant"));
398
399        // Two survivors share the cluster budget, and both already hold their whole share.
400        let share = view_log_segment_limit(Some(2));
401        let mut first = manifest(1, vec![segment(chunk(1, 5), 0)]);
402        first.view_logs[0].segments = vec![segment(chunk(1, 5), 0); share];
403        let mut second = manifest(2, vec![segment(chunk(2, 5), 0)]);
404        second.view_logs[0].segments = vec![segment(chunk(2, 5), 0); share];
405        let cut = [first, second, manifest(3, vec![segment(leaver, 0)])];
406        let error = check_view_log_adoption(&cut, &fence(&[1, 2])).unwrap_err();
407        assert!(error.contains("share of the cluster budget"), "{error}");
408        assert_eq!(
409            adopted_view_logs(&cut, &fence(&[1, 2]), 2).unwrap_err(),
410            error
411        );
412    }
413
414    #[test]
415    fn a_departed_participants_segments_are_listed_exactly_once() {
416        let leaver_object = chunk(3, 4);
417        let leaver = manifest(
418            3,
419            vec![segment(leaver_object, 0), segment(leaver_object, 1)],
420        );
421        let adopter = manifest(
422            1,
423            vec![
424                segment(chunk(1, 5), 2),
425                segment(leaver_object, 0),
426                segment(leaver_object, 1),
427            ],
428        );
429        let other = manifest(2, vec![segment(chunk(2, 5), 3)]);
430        validate_departed_view_logs([&leaver], [&adopter, &other]).unwrap();
431
432        let lost = manifest(1, vec![segment(leaver_object, 0)]);
433        assert!(validate_departed_view_logs([&leaver], [&lost, &other])
434            .unwrap_err()
435            .contains("0 times"));
436
437        let doubled = manifest(2, vec![segment(leaver_object, 1)]);
438        assert!(validate_departed_view_logs([&leaver], [&adopter, &doubled])
439            .unwrap_err()
440            .contains("2 times"));
441
442        let mut renamed = adopter;
443        renamed.view_logs[0].view = "other".into();
444        assert!(validate_departed_view_logs([&leaver], [&renamed]).is_err());
445    }
446}
laminar_core/checkpoint/committed_checkpoint/view_log_references.rs

laminar_core/checkpoint/committed_checkpoint/
view_log_references.rs
