laser_wire/
source.rs

1use crate::agent::wire_id;
2use crate::error::InvalidError;
3use crate::validate::Validate;
4use serde::{Deserialize, Serialize};
5use std::collections::BTreeSet;
6
7pub const MAX_SOURCE_PARTITIONS: usize = 65_536;
8
9wire_id!(
10    /// Durable identity of one physical Iggy cluster across replica replacement and restore.
11    PhysicalClusterIncarnation
12);
13
14#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
15pub struct SourceIncarnation {
16    pub cluster: PhysicalClusterIncarnation,
17    pub stream_id: u32,
18    pub topic_id: u32,
19    pub partition_id: u32,
20    pub partition_created_revision: u64,
21}
22
23impl Validate for SourceIncarnation {
24    fn validate(&self) -> Result<(), InvalidError> {
25        if self.cluster.as_u128() == 0 {
26            return Err(InvalidError::new(
27                "physical cluster incarnation must be nonzero",
28            ));
29        }
30        if self.partition_created_revision == 0 {
31            return Err(InvalidError::new(
32                "partition created revision must be nonzero",
33            ));
34        }
35        Ok(())
36    }
37}
38
39#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
40pub struct SourceScope {
41    pub stream: String,
42    pub topic: String,
43}
44
45impl SourceScope {
46    pub fn new(stream: impl Into<String>, topic: impl Into<String>) -> Self {
47        Self {
48            stream: stream.into(),
49            topic: topic.into(),
50        }
51    }
52}
53
54impl Validate for SourceScope {
55    fn validate(&self) -> Result<(), InvalidError> {
56        validate_source_name("stream", &self.stream)?;
57        validate_source_name("topic", &self.topic)
58    }
59}
60
61#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
62pub struct SourcePartitionCut {
63    pub incarnation: SourceIncarnation,
64    pub retained_start: u64,
65    pub end_exclusive: u64,
66}
67
68impl Validate for SourcePartitionCut {
69    fn validate(&self) -> Result<(), InvalidError> {
70        self.incarnation.validate()?;
71        if self.retained_start > self.end_exclusive {
72            return Err(InvalidError::new(format!(
73                "retained start {} is above end-exclusive head {}",
74                self.retained_start, self.end_exclusive
75            )));
76        }
77        Ok(())
78    }
79}
80
81#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
82pub struct SourceCut {
83    pub metadata_revision: u64,
84    pub partitions: Vec<SourcePartitionCut>,
85}
86
87impl Validate for SourceCut {
88    fn validate(&self) -> Result<(), InvalidError> {
89        if self.metadata_revision == 0 {
90            return Err(InvalidError::new(
91                "source metadata revision must be nonzero",
92            ));
93        }
94        if self.partitions.is_empty() || self.partitions.len() > MAX_SOURCE_PARTITIONS {
95            return Err(InvalidError::new(format!(
96                "source cut must contain 1..={MAX_SOURCE_PARTITIONS} partitions"
97            )));
98        }
99        let mut partitions = BTreeSet::new();
100        let mut previous_partition = None;
101        let mut namespace = None;
102        for partition in &self.partitions {
103            partition.validate()?;
104            let current = (
105                partition.incarnation.cluster,
106                partition.incarnation.stream_id,
107                partition.incarnation.topic_id,
108            );
109            if namespace.is_some_and(|expected| expected != current) {
110                return Err(InvalidError::new(
111                    "source cut partitions must share one cluster, stream, and topic",
112                ));
113            }
114            namespace = Some(current);
115            if !partitions.insert(partition.incarnation.partition_id) {
116                return Err(InvalidError::new(format!(
117                    "source cut repeats partition {}",
118                    partition.incarnation.partition_id
119                )));
120            }
121            if previous_partition
122                .is_some_and(|previous| previous >= partition.incarnation.partition_id)
123            {
124                return Err(InvalidError::new(
125                    "source cut partitions must be ordered by ascending partition id",
126                ));
127            }
128            previous_partition = Some(partition.incarnation.partition_id);
129        }
130        Ok(())
131    }
132}
133
134fn validate_source_name(label: &str, value: &str) -> Result<(), InvalidError> {
135    if value.is_empty() {
136        return Err(InvalidError::new(format!("{label} must not be empty")));
137    }
138    if value.len() > 255 {
139        return Err(InvalidError::new(format!(
140            "{label} is {} bytes, exceeds cap 255",
141            value.len()
142        )));
143    }
144    if value.chars().any(char::is_control) {
145        return Err(InvalidError::new(format!(
146            "{label} contains a control character"
147        )));
148    }
149    Ok(())
150}
151
152#[cfg(test)]
153mod tests {
154    use super::*;
155
156    fn incarnation(partition_id: u32, created_revision: u64) -> SourceIncarnation {
157        SourceIncarnation {
158            cluster: PhysicalClusterIncarnation::from_u128(1),
159            stream_id: 10,
160            topic_id: 20,
161            partition_id,
162            partition_created_revision: created_revision,
163        }
164    }
165
166    #[test]
167    fn given_a_recreated_partition_when_incarnations_are_compared_then_should_differ() {
168        assert_ne!(incarnation(1, 7), incarnation(1, 8));
169        incarnation(1, 7).validate().expect("incarnation is valid");
170    }
171
172    #[test]
173    fn given_duplicate_partitions_or_inverted_heads_when_a_source_cut_is_validated_then_should_reject()
174     {
175        let duplicate = SourceCut {
176            metadata_revision: 9,
177            partitions: vec![
178                SourcePartitionCut {
179                    incarnation: incarnation(1, 7),
180                    retained_start: 0,
181                    end_exclusive: 10,
182                },
183                SourcePartitionCut {
184                    incarnation: incarnation(1, 7),
185                    retained_start: 2,
186                    end_exclusive: 10,
187                },
188            ],
189        };
190        assert!(duplicate.validate().is_err());
191
192        let inverted = SourcePartitionCut {
193            incarnation: incarnation(1, 7),
194            retained_start: 11,
195            end_exclusive: 10,
196        };
197        assert!(inverted.validate().is_err());
198    }
199}
laser_wire/source.rs

laser_wire/
source.rs
