ursula-stream 0.5.0-patch1

Durable Streams state machine for Ursula: bucket and stream commands, events, and offset bookkeeping.
Documentation 
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

//! Cold-tier reference state: flushed chunks, external segments, and the cold frontier.

use super::ColdChunkRef;
use super::ObjectPayloadRef;

#[derive(Debug, Clone, Default)]
pub(super) struct StreamColdState {
    cold_chunks: Vec<ColdChunkRef>,
    external_segments: Vec<ObjectPayloadRef>,
    cold_frontier: u64,
}

impl StreamColdState {
    pub(super) fn cold_chunks(&self) -> &[ColdChunkRef] {
        &self.cold_chunks
    }

    pub(super) fn external_segments(&self) -> &[ObjectPayloadRef] {
        &self.external_segments
    }

    pub(super) fn cold_generation(&self) -> u64 {
        0
    }

    pub(super) fn push_cold_chunk(&mut self, chunk: ColdChunkRef) {
        self.cold_frontier = chunk.end_offset;
        if chunk.shared_object {
            self.cold_chunks.push(chunk);
        }
    }

    pub(super) fn push_external_segment(&mut self, object: ObjectPayloadRef) {
        let frontier = self.cold_frontier.max(object.end_offset);
        self.cold_frontier = frontier;
    }

    pub(super) fn restore(
        cold_frontier_offset: u64,
        _cold_index_generation: u64,
        cold_chunks: Vec<ColdChunkRef>,
        external_segments: Vec<ObjectPayloadRef>,
    ) -> Self {
        Self {
            cold_chunks,
            external_segments,
            cold_frontier: cold_frontier_offset,
        }
    }

    pub(super) fn has_cold_objects(&self) -> bool {
        self.cold_frontier > 0 || !self.cold_chunks.is_empty() || !self.external_segments.is_empty()
    }

    pub(super) fn compact_before(&mut self, retained_offset: u64) -> Vec<String> {
        let mut dropped_cold_paths = Vec::new();
        self.cold_chunks.retain(|chunk| {
            let retain = chunk.end_offset > retained_offset;
            if !retain {
                dropped_cold_paths.push(chunk.s3_path.clone());
            }
            retain
        });
        self.external_segments
            .retain(|object| object.end_offset > retained_offset);
        dropped_cold_paths
    }

    pub(super) fn shared_object_paths(&self) -> impl Iterator<Item = &str> {
        self.cold_chunks
            .iter()
            .filter(|chunk| chunk.shared_object)
            .map(|chunk| chunk.s3_path.as_str())
    }

    pub(super) fn remove_shared_chunks(&mut self, old_chunks: &[ColdChunkRef]) -> bool {
        let before = self.cold_chunks.len();
        self.cold_chunks
            .retain(|chunk| !old_chunks.iter().any(|old| old == chunk));
        before.saturating_sub(self.cold_chunks.len()) == old_chunks.len()
    }

    pub(super) fn cold_frontier_offset(&self, retained_offset: u64) -> u64 {
        let external_segments = self.external_segments();
        let cold_frontier = self.cold_frontier.max(retained_offset);
        let mut ranges = Vec::with_capacity(1 + external_segments.len());
        if cold_frontier > retained_offset {
            ranges.push((retained_offset, cold_frontier));
        }
        ranges.extend(
            external_segments
                .iter()
                .map(|object| (object.start_offset, object.end_offset)),
        );
        ranges.sort_unstable();

        let mut frontier = retained_offset;
        for (start_offset, end_offset) in ranges {
            if end_offset <= frontier {
                continue;
            }
            if start_offset > frontier {
                break;
            }
            frontier = end_offset;
        }
        frontier
    }
}