thruput_node_types/
parallel_impls.rs

1use crate::ParallelBatch;
2use xdr_common::{BytesM, VecM};
3
4impl ParallelBatch {
5    /// An empty batch with no segments and no data.
6    pub fn empty() -> Self {
7        Self {
8            segment_tokens: VecM(vec![]), // if VecM can't be const, use a fn
9            segment_bytes: VecM(vec![]),
10            xdr: BytesM(vec![]),
11        }
12    }
13
14    /// Total number of items across all segments.
15    ///
16    /// A `u64`, so the sum of up to `u32::MAX` per-segment `u32` counts
17    /// cannot overflow.
18    pub fn items(&self) -> u64 {
19        self.segment_tokens.iter().map(|&t| u64::from(t)).sum()
20    }
21
22    /// Number of segments in this batch.
23    ///
24    /// Uses the minimum of `segment_tokens` and `segment_bytes` lengths
25    /// so the result is always consistent with [`segment()`](Self::segment).
26    pub fn segments(&self) -> usize {
27        self.segment_tokens.len().min(self.segment_bytes.len())
28    }
29
30    /// True if the batch contains no items.
31    pub fn is_empty(&self) -> bool {
32        self.segment_tokens.iter().all(|&t| t == 0)
33    }
34
35    /// Check internal consistency of segment metadata against the
36    /// packed data.
37    pub fn validate(&self) -> Result<(), BatchValidationError> {
38        if self.segment_tokens.len() != self.segment_bytes.len() {
39            return Err(BatchValidationError::SegmentCountMismatch {
40                tokens: self.segment_tokens.len(),
41                bytes: self.segment_bytes.len(),
42            });
43        }
44        let expected: u64 = self.segment_bytes.iter().map(|&b| b as u64).sum();
45        let actual = self.xdr.len() as u64;
46        if expected != actual {
47            return Err(BatchValidationError::ByteLengthMismatch {
48                expected,
49                actual,
50            });
51        }
52        Ok(())
53    }
54
55    /// Byte slice for segment `i`. Returns None if out of bounds
56    /// or if the offset falls outside xdr.
57    pub fn segment(&self, i: usize) -> Option<&[u8]> {
58        if i >= self.segments() {
59            return None;
60        }
61        // Sum in u64 and convert checked: on a 32-bit target the offset of a
62        // corrupt batch can exceed usize.
63        let offset: u64 =
64            self.segment_bytes[..i].iter().map(|&b| u64::from(b)).sum();
65        let offset = usize::try_from(offset).ok()?;
66        let len = usize::try_from(self.segment_bytes[i]).ok()?;
67        self.xdr.0.get(offset..offset.checked_add(len)?)
68    }
69}
70
71#[derive(Debug, Clone, PartialEq, Eq)]
72pub enum BatchValidationError {
73    /// `segment_tokens` and `segment_bytes` have different lengths.
74    SegmentCountMismatch { tokens: usize, bytes: usize },
75    /// Sum of `segment_bytes` doesn't match `xdr` length.
76    ByteLengthMismatch { expected: u64, actual: u64 },
77}
78
79#[cfg(feature = "std")]
80impl std::fmt::Display for BatchValidationError {
81    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
82        match self {
83            Self::SegmentCountMismatch { tokens, bytes } => {
84                write!(
85                    f,
86                    "segment count mismatch: segment_tokens has {tokens} entries, segment_bytes has {bytes}"
87                )
88            }
89            Self::ByteLengthMismatch { expected, actual } => {
90                write!(
91                    f,
92                    "byte length mismatch: segment_bytes sum to {expected}, xdr is {actual} bytes"
93                )
94            }
95        }
96    }
97}
98
99#[cfg(test)]
100mod tests {
101    use super::*;
102    use crate::{
103        ParallelNondeterministicProposalDataBatch,
104        ParallelSignedTransactionBatch, ParallelSignedTransactionHashBatch,
105    };
106    use xdr_common::{BytesM, VecM};
107
108    /// Helper: build a batch with the given segments.
109    /// Each segment is a `(token_count, payload)` pair.
110    fn make_batch(segments: &[(u32, &[u8])]) -> ParallelBatch {
111        let mut segment_tokens = Vec::new();
112        let mut segment_bytes = Vec::new();
113        let mut xdr = Vec::new();
114        for &(tokens, payload) in segments {
115            segment_tokens.push(tokens);
116            segment_bytes.push(payload.len() as u32);
117            xdr.extend_from_slice(payload);
118        }
119        ParallelBatch {
120            segment_tokens: VecM(segment_tokens),
121            segment_bytes: VecM(segment_bytes),
122            xdr: BytesM(xdr),
123        }
124    }
125
126    // ── Validation ──────────────────────────────────────────────────
127
128    #[test]
129    fn validate_empty_batch() {
130        let batch = ParallelBatch::empty();
131        assert!(batch.validate().is_ok());
132    }
133
134    #[test]
135    fn validate_single_segment() {
136        let batch = make_batch(&[(3, &[0xAA; 10])]);
137        assert!(batch.validate().is_ok());
138    }
139
140    #[test]
141    fn validate_multi_segment() {
142        let batch = make_batch(&[(1, &[1, 2, 3]), (2, &[4, 5]), (5, &[6])]);
143        assert!(batch.validate().is_ok());
144    }
145
146    #[test]
147    fn validate_segment_count_mismatch() {
148        let batch = ParallelBatch {
149            segment_tokens: VecM(vec![1, 2]),
150            segment_bytes: VecM(vec![4]),
151            xdr: BytesM(vec![0; 4]),
152        };
153        assert_eq!(
154            batch.validate(),
155            Err(BatchValidationError::SegmentCountMismatch {
156                tokens: 2,
157                bytes: 1,
158            })
159        );
160    }
161
162    #[test]
163    fn segments_and_segment_consistent_on_mismatched_vecs() {
164        // segment_tokens longer than segment_bytes
165        let batch = ParallelBatch {
166            segment_tokens: VecM(vec![1, 2, 3]),
167            segment_bytes: VecM(vec![4, 5]),
168            xdr: BytesM(vec![0; 9]),
169        };
170        assert!(batch.validate().is_err());
171        // segments() should be capped to the shorter vec
172        assert_eq!(batch.segments(), 2);
173        assert!(batch.segment(0).is_some());
174        assert!(batch.segment(1).is_some());
175        assert_eq!(batch.segment(2), None);
176
177        // segment_bytes longer than segment_tokens
178        let batch = ParallelBatch {
179            segment_tokens: VecM(vec![1]),
180            segment_bytes: VecM(vec![3, 4]),
181            xdr: BytesM(vec![0; 7]),
182        };
183        assert!(batch.validate().is_err());
184        assert_eq!(batch.segments(), 1);
185        assert!(batch.segment(0).is_some());
186        assert_eq!(batch.segment(1), None);
187    }
188
189    #[test]
190    fn validate_byte_length_too_short() {
191        let batch = ParallelBatch {
192            segment_tokens: VecM(vec![1]),
193            segment_bytes: VecM(vec![10]),
194            xdr: BytesM(vec![0; 5]), // only 5, expected 10
195        };
196        assert_eq!(
197            batch.validate(),
198            Err(BatchValidationError::ByteLengthMismatch {
199                expected: 10,
200                actual: 5,
201            })
202        );
203    }
204
205    #[test]
206    fn validate_byte_length_too_long() {
207        let batch = ParallelBatch {
208            segment_tokens: VecM(vec![1]),
209            segment_bytes: VecM(vec![3]),
210            xdr: BytesM(vec![0; 8]), // 8 > 3
211        };
212        assert_eq!(
213            batch.validate(),
214            Err(BatchValidationError::ByteLengthMismatch {
215                expected: 3,
216                actual: 8,
217            })
218        );
219    }
220
221    // ── Utility methods ─────────────────────────────────────────────
222
223    #[test]
224    fn empty_batch_properties() {
225        let batch = ParallelBatch::empty();
226        assert_eq!(batch.items(), 0);
227        assert_eq!(batch.segments(), 0);
228        assert!(batch.is_empty());
229        assert_eq!(batch.segment(0), None);
230    }
231
232    #[test]
233    fn items_sums_across_segments() {
234        let batch = make_batch(&[(3, &[0; 2]), (7, &[0; 4]), (1, &[0; 1])]);
235        assert_eq!(batch.items(), 11);
236        assert_eq!(batch.segments(), 3);
237        assert!(!batch.is_empty());
238    }
239
240    #[test]
241    fn segment_returns_correct_slices() {
242        let batch = make_batch(&[
243            (1, &[0xAA, 0xBB]),
244            (1, &[0xCC]),
245            (1, &[0xDD, 0xEE, 0xFF]),
246        ]);
247        assert_eq!(batch.segment(0), Some([0xAA, 0xBB].as_slice()));
248        assert_eq!(batch.segment(1), Some([0xCC].as_slice()));
249        assert_eq!(batch.segment(2), Some([0xDD, 0xEE, 0xFF].as_slice()));
250        assert_eq!(batch.segment(3), None);
251    }
252
253    #[test]
254    fn segment_out_of_bounds() {
255        let batch = make_batch(&[(1, &[0x01])]);
256        assert_eq!(batch.segment(1), None);
257        assert_eq!(batch.segment(100), None);
258    }
259
260    // ── XDR round-trip (CXS) ───────────────────────────────────────
261
262    #[test]
263    fn xdr_cxs_roundtrip_empty() {
264        let batch = ParallelBatch::empty();
265        let encoded = xdrs::cxs::encode(&batch).unwrap();
266        let decoded: ParallelBatch = xdrs::cxs::decode(&encoded).unwrap();
267        assert_eq!(batch, decoded);
268    }
269
270    #[test]
271    fn xdr_cxs_roundtrip_populated() {
272        let batch = make_batch(&[(3, &[1, 2, 3, 4, 5]), (2, &[6, 7, 8])]);
273        let encoded = xdrs::cxs::encode(&batch).unwrap();
274        let decoded: ParallelBatch = xdrs::cxs::decode(&encoded).unwrap();
275        assert_eq!(batch, decoded);
276    }
277
278    #[test]
279    fn xdr_txs_roundtrip_empty() {
280        let batch = ParallelBatch::empty();
281        let encoded = xdrs::txs::encode(&batch).unwrap();
282        let decoded: ParallelBatch = xdrs::txs::decode(&encoded).unwrap();
283        assert_eq!(batch, decoded);
284    }
285
286    #[test]
287    fn xdr_txs_roundtrip_populated() {
288        let batch = make_batch(&[(10, &[0xFF; 32]), (5, &[0x42; 16])]);
289        let encoded = xdrs::txs::encode(&batch).unwrap();
290        let decoded: ParallelBatch = xdrs::txs::decode(&encoded).unwrap();
291        assert_eq!(batch, decoded);
292    }
293
294    // ── XDR round-trip for newtype wrappers ─────────────────────────
295
296    #[test]
297    fn xdr_roundtrip_signed_transaction_batch() {
298        let inner = make_batch(&[(2, &[0xAA, 0xBB, 0xCC])]);
299        let batch = ParallelSignedTransactionBatch::from(inner.clone());
300        let encoded = xdrs::cxs::encode(&batch).unwrap();
301        let decoded: ParallelSignedTransactionBatch =
302            xdrs::cxs::decode(&encoded).unwrap();
303        assert_eq!(*decoded, inner);
304    }
305
306    #[test]
307    fn xdr_roundtrip_signed_transaction_hash_batch() {
308        let inner = make_batch(&[(1, &[0xDE, 0xAD])]);
309        let batch = ParallelSignedTransactionHashBatch::from(inner.clone());
310        let encoded = xdrs::txs::encode(&batch).unwrap();
311        let decoded: ParallelSignedTransactionHashBatch =
312            xdrs::txs::decode(&encoded).unwrap();
313        assert_eq!(*decoded, inner);
314    }
315
316    #[test]
317    fn xdr_roundtrip_nondeterministic_proposal_data_batch() {
318        let inner = make_batch(&[(4, &[1; 8]), (3, &[2; 6])]);
319        let batch =
320            ParallelNondeterministicProposalDataBatch::from(inner.clone());
321        let encoded = xdrs::cxs::encode(&batch).unwrap();
322        let decoded: ParallelNondeterministicProposalDataBatch =
323            xdrs::cxs::decode(&encoded).unwrap();
324        assert_eq!(*decoded, inner);
325    }
326
327    // ── Serde JSON round-trip ───────────────────────────────────────
328
329    #[cfg(feature = "serde")]
330    #[test]
331    fn serde_json_roundtrip_empty() {
332        let batch = ParallelBatch::empty();
333        let json = serde_json::to_string(&batch).unwrap();
334        let decoded: ParallelBatch = serde_json::from_str(&json).unwrap();
335        assert_eq!(batch, decoded);
336    }
337
338    #[cfg(feature = "serde")]
339    #[test]
340    fn serde_json_roundtrip_populated() {
341        let batch = make_batch(&[(3, &[1, 2, 3, 4, 5]), (2, &[6, 7, 8])]);
342        let json = serde_json::to_string(&batch).unwrap();
343        let decoded: ParallelBatch = serde_json::from_str(&json).unwrap();
344        assert_eq!(batch, decoded);
345    }
346
347    #[cfg(feature = "serde")]
348    #[test]
349    fn serde_json_roundtrip_newtypes() {
350        let inner = make_batch(&[(1, &[0xFF; 4])]);
351
352        let tx = ParallelSignedTransactionBatch::from(inner.clone());
353        let json = serde_json::to_string(&tx).unwrap();
354        let decoded: ParallelSignedTransactionBatch =
355            serde_json::from_str(&json).unwrap();
356        assert_eq!(*decoded, inner);
357
358        let hash = ParallelSignedTransactionHashBatch::from(inner.clone());
359        let json = serde_json::to_string(&hash).unwrap();
360        let decoded: ParallelSignedTransactionHashBatch =
361            serde_json::from_str(&json).unwrap();
362        assert_eq!(*decoded, inner);
363
364        let nd =
365            ParallelNondeterministicProposalDataBatch::from(inner.clone());
366        let json = serde_json::to_string(&nd).unwrap();
367        let decoded: ParallelNondeterministicProposalDataBatch =
368            serde_json::from_str(&json).unwrap();
369        assert_eq!(*decoded, inner);
370    }
371
372    // ── Newtype conversions ─────────────────────────────────────────
373
374    #[test]
375    fn newtype_from_and_into_inner() {
376        let inner = make_batch(&[(5, &[0x01; 10])]);
377        let tx = ParallelSignedTransactionBatch::from(inner.clone());
378        assert_eq!(tx.into_inner(), inner);
379
380        let hash = ParallelSignedTransactionHashBatch::new(inner.clone());
381        assert_eq!(hash.into_inner(), inner);
382
383        let nd =
384            ParallelNondeterministicProposalDataBatch::from(inner.clone());
385        assert_eq!(nd.into_inner(), inner);
386    }
387
388    #[test]
389    fn newtype_deref_provides_validation() {
390        let inner = make_batch(&[(2, &[0xAB; 6])]);
391        let tx = ParallelSignedTransactionBatch::from(inner);
392        // Deref allows calling ParallelBatch methods directly
393        assert!(tx.validate().is_ok());
394        assert_eq!(tx.items(), 2);
395        assert_eq!(tx.segments(), 1);
396        assert_eq!(tx.segment(0), Some([0xAB; 6].as_slice()));
397    }
398
399    #[test]
400    fn items_does_not_overflow_u32() {
401        let batch = ParallelBatch {
402            segment_tokens: VecM(vec![u32::MAX, u32::MAX]),
403            segment_bytes: VecM(vec![0, 0]),
404            xdr: BytesM(vec![]),
405        };
406        assert_eq!(batch.items(), 2 * u64::from(u32::MAX));
407        assert!(!batch.is_empty());
408    }
409
410    #[test]
411    fn is_empty_with_zero_token_segments() {
412        let batch = make_batch(&[(0, &[]), (0, &[])]);
413        assert!(batch.is_empty());
414        assert_eq!(batch.items(), 0);
415    }
416
417    #[test]
418    fn segment_with_corrupt_lengths_is_none() {
419        // segment_bytes claim far more data than xdr holds; the offset of
420        // the last segment exceeds u32 and must not panic or wrap.
421        let batch = ParallelBatch {
422            segment_tokens: VecM(vec![1, 1, 1]),
423            segment_bytes: VecM(vec![u32::MAX, u32::MAX, 1]),
424            xdr: BytesM(vec![0; 4]),
425        };
426        assert!(batch.validate().is_err());
427        assert_eq!(batch.segment(0), None);
428        assert_eq!(batch.segment(2), None);
429    }
430}
thruput_node_types/parallel_impls.rs

thruput_node_types/
parallel_impls.rs
