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

laser_wire/
source.rs
