rustfs_protos/
heal_control.rs

1// Copyright 2024 RustFS Team
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15//! Structural wire codec for heal-control commands.
16//!
17//! Decoding proves only that a payload has a supported, bounded shape. It does
18//! not establish coordinator authority, freshness, or replay protection. The
19//! decoded fields intentionally remain opaque until an execution layer adds
20//! contextual lease and nonce validation.
21
22use rmp_serde::Deserializer;
23use rustfs_heal_contracts::heal_channel::{
24    HealAdmissionDropReason, HealAdmissionResult, HealChannelPriority, HealChannelRequest, HealChannelResponse,
25    HealRequestSource, HealScanMode,
26};
27use serde::{Deserialize, Serialize, de::SeqAccess, de::Visitor};
28use std::{fmt, io::Cursor, io::Write};
29
30const ENVELOPE_VERSION: u8 = 2;
31pub const ENVELOPE_MAX_SIZE: usize = 64 * 1024;
32pub const RESULT_MAX_SIZE: usize = 16 * 1024 * 1024;
33pub const NONCE_SIZE: usize = 16;
34pub const MAX_LIFETIME_MS: i64 = 30_000;
35const MAX_CLOCK_SKEW_MS: i64 = 5_000;
36
37#[derive(Debug, Clone, Copy, PartialEq, Eq)]
38pub struct RequestMetadata {
39    pub nonce: [u8; NONCE_SIZE],
40    pub issued_at_unix_ms: i64,
41    pub expires_at_unix_ms: i64,
42    pub coordinator_epoch: u64,
43}
44
45impl RequestMetadata {
46    pub const fn new(nonce: [u8; NONCE_SIZE], issued_at_unix_ms: i64, expires_at_unix_ms: i64, coordinator_epoch: u64) -> Self {
47        Self {
48            nonce,
49            issued_at_unix_ms,
50            expires_at_unix_ms,
51            coordinator_epoch,
52        }
53    }
54}
55
56#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
57#[serde(rename_all = "snake_case")]
58enum Priority {
59    Low,
60    Normal,
61    High,
62    Critical,
63}
64
65impl From<HealChannelPriority> for Priority {
66    fn from(priority: HealChannelPriority) -> Self {
67        match priority {
68            HealChannelPriority::Low => Self::Low,
69            HealChannelPriority::Normal => Self::Normal,
70            HealChannelPriority::High => Self::High,
71            HealChannelPriority::Critical => Self::Critical,
72        }
73    }
74}
75
76impl From<Priority> for HealChannelPriority {
77    fn from(priority: Priority) -> Self {
78        match priority {
79            Priority::Low => Self::Low,
80            Priority::Normal => Self::Normal,
81            Priority::High => Self::High,
82            Priority::Critical => Self::Critical,
83        }
84    }
85}
86
87#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
88#[serde(rename_all = "camelCase", deny_unknown_fields)]
89pub struct StartCommand {
90    disk: Option<String>,
91    #[serde(default, skip_serializing_if = "Vec::is_empty")]
92    heal_endpoints: Vec<String>,
93    bucket: String,
94    object_prefix: Option<String>,
95    object_version_id: Option<String>,
96    force_start: bool,
97    priority: Priority,
98    pool_index: Option<u64>,
99    set_index: Option<u64>,
100    scan_mode: Option<HealScanMode>,
101    remove_corrupted: Option<bool>,
102    recreate_missing: Option<bool>,
103    update_parity: Option<bool>,
104    recursive: Option<bool>,
105    dry_run: Option<bool>,
106    #[serde(default, skip_serializing_if = "Option::is_none")]
107    no_lock: Option<bool>,
108    timeout_seconds: Option<u64>,
109    source: HealRequestSource,
110}
111
112impl TryFrom<HealChannelRequest> for StartCommand {
113    type Error = String;
114
115    fn try_from(request: HealChannelRequest) -> Result<Self, Self::Error> {
116        Ok(Self {
117            disk: request.disk,
118            heal_endpoints: request.heal_endpoints,
119            bucket: request.bucket,
120            object_prefix: request.object_prefix,
121            object_version_id: request.object_version_id,
122            force_start: request.force_start,
123            priority: request.priority.into(),
124            pool_index: request
125                .pool_index
126                .map(u64::try_from)
127                .transpose()
128                .map_err(|_| "heal pool index exceeds wire range".to_string())?,
129            set_index: request
130                .set_index
131                .map(u64::try_from)
132                .transpose()
133                .map_err(|_| "heal set index exceeds wire range".to_string())?,
134            scan_mode: request.scan_mode,
135            remove_corrupted: request.remove_corrupted,
136            recreate_missing: request.recreate_missing,
137            update_parity: request.update_parity,
138            recursive: request.recursive,
139            dry_run: request.dry_run,
140            no_lock: request.no_lock,
141            timeout_seconds: request.timeout_seconds,
142            source: request.source,
143        })
144    }
145}
146
147impl StartCommand {
148    fn into_channel_request(self, request_id: String) -> Result<HealChannelRequest, String> {
149        Ok(HealChannelRequest {
150            id: request_id,
151            disk: self.disk,
152            heal_endpoints: self.heal_endpoints,
153            bucket: self.bucket,
154            object_prefix: self.object_prefix,
155            object_version_id: self.object_version_id,
156            force_start: self.force_start,
157            priority: self.priority.into(),
158            pool_index: self
159                .pool_index
160                .map(usize::try_from)
161                .transpose()
162                .map_err(|_| "heal pool index exceeds platform range".to_string())?,
163            set_index: self
164                .set_index
165                .map(usize::try_from)
166                .transpose()
167                .map_err(|_| "heal set index exceeds platform range".to_string())?,
168            scan_mode: self.scan_mode,
169            remove_corrupted: self.remove_corrupted,
170            recreate_missing: self.recreate_missing,
171            update_parity: self.update_parity,
172            recursive: self.recursive,
173            dry_run: self.dry_run,
174            no_lock: self.no_lock,
175            timeout_seconds: self.timeout_seconds,
176            source: self.source,
177        })
178    }
179}
180
181#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
182#[serde(tag = "action", rename_all = "snake_case", deny_unknown_fields)]
183pub enum Command {
184    Start {
185        request: StartCommand,
186    },
187    Query {
188        heal_path: String,
189        client_token: String,
190        /// Incremental result cursor (HS-06): only items with a sequence
191        /// greater than this are returned. Absent = legacy full snapshot.
192        /// Optional + defaulted so older peers stay wire-compatible.
193        #[serde(default, skip_serializing_if = "Option::is_none")]
194        since_seq: Option<u64>,
195    },
196    Cancel {
197        heal_path: String,
198        client_token: String,
199    },
200}
201
202#[derive(Debug)]
203pub enum ExecutableCommand {
204    Start {
205        request: HealChannelRequest,
206    },
207    Query {
208        heal_path: String,
209        client_token: String,
210        since_seq: Option<u64>,
211    },
212    Cancel {
213        heal_path: String,
214        client_token: String,
215    },
216}
217
218#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
219#[serde(rename_all = "camelCase", deny_unknown_fields)]
220pub struct Envelope {
221    version: u8,
222    request_id: String,
223    nonce: [u8; NONCE_SIZE],
224    issued_at_unix_ms: i64,
225    expires_at_unix_ms: i64,
226    coordinator_epoch: u64,
227    command: Command,
228}
229
230impl Envelope {
231    fn new(request_id: String, metadata: RequestMetadata, command: Command) -> Result<Self, String> {
232        let envelope = Self {
233            version: ENVELOPE_VERSION,
234            request_id,
235            nonce: metadata.nonce,
236            issued_at_unix_ms: metadata.issued_at_unix_ms,
237            expires_at_unix_ms: metadata.expires_at_unix_ms,
238            coordinator_epoch: metadata.coordinator_epoch,
239            command,
240        };
241        envelope.validate()?;
242        Ok(envelope)
243    }
244
245    pub fn start(request: HealChannelRequest, metadata: RequestMetadata) -> Result<Self, String> {
246        let request_id = request.id.clone();
247        Self::new(
248            request_id,
249            metadata,
250            Command::Start {
251                request: request.try_into()?,
252            },
253        )
254    }
255
256    pub fn query(
257        request_id: String,
258        metadata: RequestMetadata,
259        heal_path: String,
260        client_token: String,
261        since_seq: Option<u64>,
262    ) -> Result<Self, String> {
263        Self::new(
264            request_id,
265            metadata,
266            Command::Query {
267                heal_path,
268                client_token,
269                since_seq,
270            },
271        )
272    }
273
274    pub fn cancel(
275        request_id: String,
276        metadata: RequestMetadata,
277        heal_path: String,
278        client_token: String,
279    ) -> Result<Self, String> {
280        Self::new(request_id, metadata, Command::Cancel { heal_path, client_token })
281    }
282
283    fn validate(&self) -> Result<(), String> {
284        if self.version != ENVELOPE_VERSION {
285            return Err(format!("unsupported heal control envelope version: {}", self.version));
286        }
287        validate_uuid(&self.request_id, "request")?;
288        if self.coordinator_epoch == 0 {
289            return Err("heal control coordinator epoch is zero".to_string());
290        }
291        if self.nonce.iter().all(|byte| *byte == 0) {
292            return Err("heal control nonce is zero".to_string());
293        }
294        if self.issued_at_unix_ms < 0 || self.expires_at_unix_ms <= self.issued_at_unix_ms {
295            return Err("heal control request lifetime is invalid".to_string());
296        }
297        if self.expires_at_unix_ms - self.issued_at_unix_ms > MAX_LIFETIME_MS {
298            return Err("heal control request lifetime exceeds limit".to_string());
299        }
300        match &self.command {
301            Command::Start { .. } => {}
302            Command::Query { client_token, .. } if client_token.is_empty() => {
303                return Err("heal control client token is empty".to_string());
304            }
305            Command::Query { .. } | Command::Cancel { .. } => {}
306        }
307        Ok(())
308    }
309
310    pub fn validate_execution(&self, now_unix_ms: i64, expected_coordinator_epoch: u64) -> Result<(), String> {
311        self.validate()?;
312        if self.coordinator_epoch != expected_coordinator_epoch {
313            return Err("heal control coordinator epoch does not match".to_string());
314        }
315        if self.issued_at_unix_ms > now_unix_ms.saturating_add(MAX_CLOCK_SKEW_MS) {
316            return Err("heal control request was issued in the future".to_string());
317        }
318        if self.expires_at_unix_ms <= now_unix_ms {
319            return Err("heal control request expired".to_string());
320        }
321        Ok(())
322    }
323
324    pub fn into_execution(self) -> Result<(String, u64, ExecutableCommand), String> {
325        let command = match self.command {
326            Command::Start { request } => ExecutableCommand::Start {
327                request: request.into_channel_request(self.request_id.clone())?,
328            },
329            Command::Query {
330                heal_path,
331                client_token,
332                since_seq,
333            } => ExecutableCommand::Query {
334                heal_path,
335                client_token,
336                since_seq,
337            },
338            Command::Cancel { heal_path, client_token } => ExecutableCommand::Cancel { heal_path, client_token },
339        };
340        Ok((self.request_id, self.coordinator_epoch, command))
341    }
342
343    pub const fn expires_at_unix_ms(&self) -> i64 {
344        self.expires_at_unix_ms
345    }
346}
347
348#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
349#[serde(rename_all = "snake_case")]
350pub enum Admission {
351    Accepted,
352    Merged,
353    Full,
354    DroppedQueueFull,
355    DroppedPolicy,
356    /// Admin start rejected because the same target is already owned and
357    /// cannot be merged under the selected overlap policy.
358    DroppedAlreadyRunning,
359    /// Admin start rejected because its scope overlaps an existing owner.
360    DroppedOverlappingPaths,
361}
362
363impl From<HealAdmissionResult> for Admission {
364    fn from(result: HealAdmissionResult) -> Self {
365        match result {
366            HealAdmissionResult::Accepted => Self::Accepted,
367            HealAdmissionResult::Merged => Self::Merged,
368            HealAdmissionResult::Full => Self::Full,
369            HealAdmissionResult::Dropped(HealAdmissionDropReason::QueueFull) => Self::DroppedQueueFull,
370            HealAdmissionResult::Dropped(HealAdmissionDropReason::PolicyDropped) => Self::DroppedPolicy,
371            HealAdmissionResult::Dropped(HealAdmissionDropReason::AlreadyRunning) => Self::DroppedAlreadyRunning,
372            HealAdmissionResult::Dropped(HealAdmissionDropReason::OverlappingPaths) => Self::DroppedOverlappingPaths,
373        }
374    }
375}
376
377impl Admission {
378    pub const fn is_admitted(self) -> bool {
379        matches!(self, Self::Accepted | Self::Merged)
380    }
381
382    pub const fn into_heal_admission_result(self) -> HealAdmissionResult {
383        match self {
384            Self::Accepted => HealAdmissionResult::Accepted,
385            Self::Merged => HealAdmissionResult::Merged,
386            Self::Full => HealAdmissionResult::Full,
387            Self::DroppedQueueFull => HealAdmissionResult::Dropped(HealAdmissionDropReason::QueueFull),
388            Self::DroppedPolicy => HealAdmissionResult::Dropped(HealAdmissionDropReason::PolicyDropped),
389            Self::DroppedAlreadyRunning => HealAdmissionResult::Dropped(HealAdmissionDropReason::AlreadyRunning),
390            Self::DroppedOverlappingPaths => HealAdmissionResult::Dropped(HealAdmissionDropReason::OverlappingPaths),
391        }
392    }
393}
394
395#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
396#[serde(tag = "result", rename_all = "snake_case", deny_unknown_fields)]
397pub enum Outcome {
398    Start {
399        task_id: String,
400        admission: Admission,
401    },
402    Channel {
403        success: bool,
404        #[serde(deserialize_with = "deserialize_bounded_optional_bytes")]
405        data: Option<Vec<u8>>,
406        error: Option<String>,
407    },
408}
409
410#[derive(Debug)]
411struct BoundedBytes(Vec<u8>);
412struct BoundedBytesVisitor;
413
414impl<'de> Visitor<'de> for BoundedBytesVisitor {
415    type Value = BoundedBytes;
416
417    fn expecting(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
418        write!(formatter, "at most {RESULT_MAX_SIZE} bytes")
419    }
420
421    fn visit_seq<A>(self, mut sequence: A) -> Result<Self::Value, A::Error>
422    where
423        A: SeqAccess<'de>,
424    {
425        if sequence.size_hint().is_some_and(|length| length > RESULT_MAX_SIZE) {
426            return Err(serde::de::Error::custom("heal control response data exceeds size limit"));
427        }
428
429        let mut bytes = Vec::new();
430        while let Some(byte) = sequence.next_element()? {
431            if bytes.len() == RESULT_MAX_SIZE {
432                return Err(serde::de::Error::custom("heal control response data exceeds size limit"));
433            }
434            bytes.push(byte);
435        }
436        Ok(BoundedBytes(bytes))
437    }
438}
439
440impl<'de> Deserialize<'de> for BoundedBytes {
441    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
442    where
443        D: serde::Deserializer<'de>,
444    {
445        deserializer.deserialize_seq(BoundedBytesVisitor)
446    }
447}
448
449fn deserialize_bounded_optional_bytes<'de, D>(deserializer: D) -> Result<Option<Vec<u8>>, D::Error>
450where
451    D: serde::Deserializer<'de>,
452{
453    Ok(Option::<BoundedBytes>::deserialize(deserializer)?.map(|bytes| bytes.0))
454}
455
456#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
457#[serde(rename_all = "camelCase", deny_unknown_fields)]
458pub struct ResultEnvelope {
459    version: u8,
460    request_id: String,
461    coordinator_epoch: u64,
462    outcome: Outcome,
463}
464
465impl ResultEnvelope {
466    pub fn new(request_id: String, coordinator_epoch: u64, outcome: Outcome) -> Result<Self, String> {
467        let result = Self {
468            version: ENVELOPE_VERSION,
469            request_id,
470            coordinator_epoch,
471            outcome,
472        };
473        result.validate()?;
474        Ok(result)
475    }
476
477    pub fn channel(request_id: String, coordinator_epoch: u64, response: HealChannelResponse) -> Result<Self, String> {
478        if response.request_id != request_id {
479            return Err("heal control channel response request ID does not match".to_string());
480        }
481        Self::new(
482            request_id,
483            coordinator_epoch,
484            Outcome::Channel {
485                success: response.success,
486                data: response.data,
487                error: response.error,
488            },
489        )
490    }
491
492    fn validate(&self) -> Result<(), String> {
493        if self.version != ENVELOPE_VERSION {
494            return Err(format!("unsupported heal control result version: {}", self.version));
495        }
496        if self.coordinator_epoch == 0 {
497            return Err("heal control result coordinator epoch is zero".to_string());
498        }
499        validate_uuid(&self.request_id, "result request")?;
500        match &self.outcome {
501            Outcome::Start { task_id, .. } => validate_uuid(task_id, "result task")?,
502            Outcome::Channel {
503                success: false,
504                error: None,
505                ..
506            } => {
507                return Err("failed heal control result has no error".to_string());
508            }
509            Outcome::Channel {
510                success: false,
511                data: Some(_),
512                ..
513            } => {
514                return Err("failed heal control result contains data".to_string());
515            }
516            Outcome::Channel {
517                success: false,
518                error: Some(error),
519                ..
520            } if error.is_empty() => {
521                return Err("failed heal control result has an empty error".to_string());
522            }
523            Outcome::Channel { .. } => {}
524        }
525        Ok(())
526    }
527
528    pub fn into_outcome(self, expected_request_id: &str, expected_epoch: u64) -> Result<Outcome, String> {
529        self.validate()?;
530        if self.request_id != expected_request_id {
531            return Err("heal control result request ID does not match".to_string());
532        }
533        if self.coordinator_epoch != expected_epoch {
534            return Err("heal control result coordinator epoch does not match".to_string());
535        }
536        Ok(self.outcome)
537    }
538}
539
540fn validate_uuid(value: &str, field: &str) -> Result<(), String> {
541    let parsed = uuid::Uuid::parse_str(value).map_err(|_| format!("heal control {field} ID is not a UUID"))?;
542    if parsed.is_nil() || parsed.to_string() != value {
543        return Err(format!("heal control {field} ID is not canonical"));
544    }
545    Ok(())
546}
547
548fn decode<T: for<'de> Deserialize<'de>>(data: &[u8], value_name: &str, max_size: usize) -> Result<T, String> {
549    if data.len() > max_size {
550        return Err(format!("{value_name} exceeds size limit"));
551    }
552    let mut deserializer = Deserializer::new(Cursor::new(data));
553    let value = T::deserialize(&mut deserializer).map_err(|err| format!("failed to decode {value_name}: {err}"))?;
554    if usize::try_from(deserializer.get_ref().position()).ok() != Some(data.len()) {
555        return Err(format!("{value_name} contains trailing data"));
556    }
557    Ok(value)
558}
559
560pub fn encode_envelope(envelope: &Envelope) -> Result<Vec<u8>, String> {
561    envelope.validate()?;
562    encode_bounded(envelope, "heal control envelope", ENVELOPE_MAX_SIZE)
563}
564
565pub fn decode_envelope(data: &[u8]) -> Result<Envelope, String> {
566    let envelope: Envelope = decode(data, "heal control envelope", ENVELOPE_MAX_SIZE)?;
567    envelope.validate()?;
568    Ok(envelope)
569}
570
571pub fn encode_result(result: &ResultEnvelope) -> Result<Vec<u8>, String> {
572    result.validate()?;
573    encode_bounded(result, "heal control result", RESULT_MAX_SIZE)
574}
575
576pub fn decode_result(data: &[u8]) -> Result<ResultEnvelope, String> {
577    let result: ResultEnvelope = decode(data, "heal control result", RESULT_MAX_SIZE)?;
578    result.validate()?;
579    Ok(result)
580}
581
582fn encode_bounded(value: &impl Serialize, value_name: &str, max_size: usize) -> Result<Vec<u8>, String> {
583    struct BoundedWriter {
584        bytes: Vec<u8>,
585        exceeded: bool,
586        max_size: usize,
587    }
588
589    impl Write for BoundedWriter {
590        fn write(&mut self, data: &[u8]) -> std::io::Result<usize> {
591            let remaining = self.max_size.saturating_sub(self.bytes.len());
592            if data.len() > remaining {
593                self.exceeded = true;
594                return Err(std::io::Error::other("heal control value exceeds size limit"));
595            }
596            self.bytes.extend_from_slice(data);
597            Ok(data.len())
598        }
599
600        fn flush(&mut self) -> std::io::Result<()> {
601            Ok(())
602        }
603    }
604
605    let mut writer = BoundedWriter {
606        bytes: Vec::with_capacity(1024),
607        exceeded: false,
608        max_size,
609    };
610    let result = value.serialize(&mut rmp_serde::Serializer::new(&mut writer).with_struct_map());
611    if writer.exceeded {
612        return Err(format!("{value_name} exceeds size limit"));
613    }
614    result.map_err(|err| format!("failed to encode {value_name}: {err}"))?;
615    Ok(writer.bytes)
616}
617
618#[cfg(test)]
619mod tests {
620    use super::{
621        Admission, ENVELOPE_MAX_SIZE, Envelope, ExecutableCommand, Outcome, RESULT_MAX_SIZE, RequestMetadata, ResultEnvelope,
622        decode_envelope, decode_result, encode_result,
623    };
624    use rustfs_heal_contracts::heal_channel::{HealChannelRequest, HealChannelResponse, HealRequestSource};
625    use serde::de::{DeserializeSeed, SeqAccess, Visitor, value::Error as ValueError};
626
627    fn test_request(request_id: String) -> HealChannelRequest {
628        HealChannelRequest {
629            id: request_id,
630            bucket: "bucket".to_string(),
631            object_prefix: Some("prefix".to_string()),
632            pool_index: Some(1),
633            set_index: Some(2),
634            source: HealRequestSource::Admin,
635            ..Default::default()
636        }
637    }
638
639    fn replacement_test_request(request_id: String) -> HealChannelRequest {
640        let mut request = test_request(request_id);
641        request.heal_endpoints = vec!["http://node1:9000/drive2".to_string()];
642        request
643    }
644
645    fn metadata(byte: u8, epoch: u64) -> RequestMetadata {
646        RequestMetadata::new([byte; 16], 1_000, 2_000, epoch)
647    }
648
649    #[test]
650    fn round_trips_all_commands_and_results() {
651        let request_id = uuid::Uuid::new_v4().to_string();
652        let start = Envelope::start(replacement_test_request(request_id), metadata(1, 7)).unwrap();
653        let query = Envelope::query(
654            uuid::Uuid::new_v4().to_string(),
655            metadata(2, 7),
656            "bucket/prefix".to_string(),
657            "token".to_string(),
658            None,
659        )
660        .unwrap();
661        let cancel = Envelope::cancel(
662            uuid::Uuid::new_v4().to_string(),
663            metadata(3, 7),
664            "bucket/prefix".to_string(),
665            "token".to_string(),
666        )
667        .unwrap();
668
669        for envelope in [start, query, cancel] {
670            let encoded = super::encode_envelope(&envelope).unwrap();
671            assert_eq!(decode_envelope(&encoded).unwrap(), envelope);
672        }
673
674        let result = ResultEnvelope::new(
675            uuid::Uuid::new_v4().to_string(),
676            7,
677            Outcome::Start {
678                task_id: uuid::Uuid::new_v4().to_string(),
679                admission: Admission::Merged,
680            },
681        )
682        .unwrap();
683        let encoded = super::encode_result(&result).unwrap();
684        assert_eq!(decode_result(&encoded).unwrap(), result);
685
686        let channel_request_id = uuid::Uuid::new_v4().to_string();
687        let channel = ResultEnvelope::channel(
688            channel_request_id.clone(),
689            7,
690            HealChannelResponse {
691                request_id: channel_request_id,
692                success: true,
693                data: Some(vec![1, 2, 3]),
694                error: None,
695            },
696        )
697        .unwrap();
698        let encoded = super::encode_result(&channel).unwrap();
699        assert_eq!(decode_result(&encoded).unwrap(), channel);
700    }
701
702    #[test]
703    fn v2_start_round_trips_no_lock() {
704        for no_lock in [None, Some(false), Some(true)] {
705            let request_id = uuid::Uuid::new_v4().to_string();
706            let mut request = test_request(request_id);
707            request.no_lock = no_lock;
708
709            let envelope = Envelope::start(request, metadata(1, 7)).expect("v2 start should encode nolock");
710            let encoded = super::encode_envelope(&envelope).expect("v2 start should serialize");
711            let decoded = super::decode_envelope(&encoded).expect("v2 start should deserialize");
712            let (_, _, ExecutableCommand::Start { request }) = decoded.into_execution().expect("v2 start command should decode")
713            else {
714                panic!("expected start command");
715            };
716
717            assert_eq!(request.no_lock, no_lock);
718        }
719    }
720
721    #[test]
722    fn v2_command_and_result_wire_fixtures_are_stable() {
723        let request_id = "00112233-4455-6677-8899-aabbccddeeff".to_string();
724        let start = Envelope::start(
725            test_request(request_id.clone()),
726            RequestMetadata::new([0x11; 16], 1_700_000_000_000, 1_700_000_030_000, 9),
727        )
728        .unwrap();
729        let envelope = Envelope::query(
730            request_id.clone(),
731            RequestMetadata::new([0x11; 16], 1_700_000_000_000, 1_700_000_030_000, 9),
732            "bucket/prefix".to_string(),
733            "client-token".to_string(),
734            None,
735        )
736        .unwrap();
737        let cancel = Envelope::cancel(
738            request_id.clone(),
739            RequestMetadata::new([0x11; 16], 1_700_000_000_000, 1_700_000_030_000, 9),
740            "bucket/prefix".to_string(),
741            "client-token".to_string(),
742        )
743        .unwrap();
744        let start_result = ResultEnvelope::new(
745            request_id.clone(),
746            9,
747            Outcome::Start {
748                task_id: "ffeeddcc-bbaa-9988-7766-554433221100".to_string(),
749                admission: Admission::DroppedPolicy,
750            },
751        )
752        .unwrap();
753        let result = ResultEnvelope::new(
754            request_id,
755            9,
756            Outcome::Channel {
757                success: true,
758                data: Some(vec![1, 2, 3]),
759                error: None,
760            },
761        )
762        .unwrap();
763
764        let envelope_hex = super::encode_envelope(&envelope)
765            .unwrap()
766            .iter()
767            .map(|byte| format!("{byte:02x}"))
768            .collect::<String>();
769        let result_hex = super::encode_result(&result)
770            .unwrap()
771            .iter()
772            .map(|byte| format!("{byte:02x}"))
773            .collect::<String>();
774        let start_hex = super::encode_envelope(&start)
775            .unwrap()
776            .iter()
777            .map(|byte| format!("{byte:02x}"))
778            .collect::<String>();
779        let cancel_hex = super::encode_envelope(&cancel)
780            .unwrap()
781            .iter()
782            .map(|byte| format!("{byte:02x}"))
783            .collect::<String>();
784        let start_result_hex = super::encode_result(&start_result)
785            .unwrap()
786            .iter()
787            .map(|byte| format!("{byte:02x}"))
788            .collect::<String>();
789        assert_eq!(
790            start_hex,
791            "87a776657273696f6e02a9726571756573744964d92430303131323233332d343435352d363637372d383839392d616162626363646465656666a56e6f6e6365dc001011111111111111111111111111111111ae6973737565644174556e69784d73cf0000018bcfe56800af657870697265734174556e69784d73cf0000018bcfe5dd30b0636f6f7264696e61746f7245706f636809a7636f6d6d616e6482a6616374696f6ea57374617274a772657175657374de0010a46469736bc0a66275636b6574a66275636b6574ac6f626a656374507265666978a6707265666978af6f626a65637456657273696f6e4964c0aa666f7263655374617274c2a87072696f72697479a66e6f726d616ca9706f6f6c496e64657801a8736574496e64657802a87363616e4d6f6465c0af72656d6f7665436f72727570746564c0af72656372656174654d697373696e67c0ac757064617465506172697479c0a9726563757273697665c0a664727952756ec0ae74696d656f75745365636f6e6473c0a6736f75726365a561646d696e"
792        );
793        assert_eq!(
794            cancel_hex,
795            "87a776657273696f6e02a9726571756573744964d92430303131323233332d343435352d363637372d383839392d616162626363646465656666a56e6f6e6365dc001011111111111111111111111111111111ae6973737565644174556e69784d73cf0000018bcfe56800af657870697265734174556e69784d73cf0000018bcfe5dd30b0636f6f7264696e61746f7245706f636809a7636f6d6d616e6483a6616374696f6ea663616e63656ca96865616c5f70617468ad6275636b65742f707265666978ac636c69656e745f746f6b656eac636c69656e742d746f6b656e"
796        );
797        assert_eq!(
798            start_result_hex,
799            "84a776657273696f6e02a9726571756573744964d92430303131323233332d343435352d363637372d383839392d616162626363646465656666b0636f6f7264696e61746f7245706f636809a76f7574636f6d6583a6726573756c74a57374617274a77461736b5f6964d92466666565646463632d626261612d393938382d373736362d353534343333323231313030a961646d697373696f6eae64726f707065645f706f6c696379"
800        );
801        assert_eq!(
802            envelope_hex,
803            "87a776657273696f6e02a9726571756573744964d92430303131323233332d343435352d363637372d383839392d616162626363646465656666a56e6f6e6365dc001011111111111111111111111111111111ae6973737565644174556e69784d73cf0000018bcfe56800af657870697265734174556e69784d73cf0000018bcfe5dd30b0636f6f7264696e61746f7245706f636809a7636f6d6d616e6483a6616374696f6ea57175657279a96865616c5f70617468ad6275636b65742f707265666978ac636c69656e745f746f6b656eac636c69656e742d746f6b656e"
804        );
805        assert_eq!(
806            result_hex,
807            "84a776657273696f6e02a9726571756573744964d92430303131323233332d343435352d363637372d383839392d616162626363646465656666b0636f6f7264696e61746f7245706f636809a76f7574636f6d6584a6726573756c74a76368616e6e656ca773756363657373c3a46461746193010203a56572726f72c0"
808        );
809    }
810
811    #[test]
812    fn rejects_invalid_and_untrusted_shapes() {
813        let request_id = uuid::Uuid::new_v4().to_string();
814        assert!(Envelope::start(test_request(request_id.clone()), metadata(0, 7)).is_err());
815        assert!(Envelope::start(test_request(request_id.clone()), metadata(1, 0)).is_err());
816        assert!(Envelope::start(test_request(request_id.clone()), RequestMetadata::new([1; 16], 1_000, 31_001, 7),).is_err());
817        assert!(Envelope::query(request_id.clone(), metadata(1, 7), String::new(), String::new(), None).is_err());
818        assert!(Envelope::cancel(request_id.clone(), metadata(1, 7), String::new(), String::new()).is_ok());
819
820        let mut noncanonical_request = test_request(request_id.to_uppercase());
821        assert!(Envelope::start(noncanonical_request.clone(), metadata(1, 7)).is_err());
822        noncanonical_request.id = uuid::Uuid::nil().to_string();
823        assert!(Envelope::start(noncanonical_request, metadata(1, 7)).is_err());
824
825        let boundary = RequestMetadata::new([1; 16], 1_000, 1_000 + super::MAX_LIFETIME_MS, 7);
826        assert!(Envelope::start(test_request(request_id.clone()), boundary).is_ok());
827        assert!(
828            Envelope::start(
829                test_request(request_id.clone()),
830                RequestMetadata::new([1; 16], 1_000, 1_001 + super::MAX_LIFETIME_MS, 7),
831            )
832            .is_err()
833        );
834
835        let envelope = Envelope::start(test_request(request_id.clone()), metadata(1, 7)).unwrap();
836        let mut trailing = super::encode_envelope(&envelope).unwrap();
837        trailing.push(0);
838        assert!(decode_envelope(&trailing).unwrap_err().contains("trailing data"));
839        assert!(
840            decode_envelope(&vec![0; ENVELOPE_MAX_SIZE + 1])
841                .unwrap_err()
842                .contains("size limit")
843        );
844
845        let oversized = Envelope::query(
846            uuid::Uuid::new_v4().to_string(),
847            metadata(1, 7),
848            "x".repeat(ENVELOPE_MAX_SIZE),
849            "token".to_string(),
850            None,
851        )
852        .unwrap();
853        let error = super::encode_envelope(&oversized).unwrap_err();
854        assert!(error.contains("size limit"), "{error}");
855
856        let mut unknown = serde_json::to_value(&envelope).unwrap();
857        unknown
858            .as_object_mut()
859            .unwrap()
860            .insert("unknown".to_string(), serde_json::Value::Bool(true));
861        let unknown = rmp_serde::to_vec_named(&unknown).unwrap();
862        assert!(decode_envelope(&unknown).unwrap_err().contains("unknown field"));
863
864        let mut read_repair = serde_json::to_value(&envelope).expect("start envelope should serialize");
865        read_repair["command"]["request"]
866            .as_object_mut()
867            .expect("start request must be an object")
868            .insert("readRepair".to_string(), serde_json::Value::Bool(true));
869        let encoded = rmp_serde::to_vec_named(&read_repair).expect("invalid start fixture should encode");
870        let error = decode_envelope(&encoded).expect_err("RPC must reject an Admin readRepair field");
871        assert!(error.contains("unknown field") && error.contains("readRepair"), "{error}");
872
873        let executable =
874            Envelope::start(test_request(request_id.clone()), RequestMetadata::new([1; 16], 10_000, 20_000, 7)).unwrap();
875        assert!(executable.validate_execution(15_000, 7).is_ok());
876        assert!(executable.validate_execution(20_000, 7).unwrap_err().contains("expired"));
877        assert!(executable.validate_execution(4_999, 7).unwrap_err().contains("future"));
878        let wrong_epoch =
879            Envelope::start(test_request(request_id.clone()), RequestMetadata::new([1; 16], 10_000, 20_000, 8)).unwrap();
880        assert!(wrong_epoch.validate_execution(15_000, 7).unwrap_err().contains("epoch"));
881
882        assert!(
883            ResultEnvelope::channel(
884                uuid::Uuid::new_v4().to_string(),
885                7,
886                HealChannelResponse {
887                    request_id: uuid::Uuid::new_v4().to_string(),
888                    success: true,
889                    data: None,
890                    error: None,
891                },
892            )
893            .unwrap_err()
894            .contains("does not match")
895        );
896
897        assert!(
898            ResultEnvelope::new(
899                request_id.clone(),
900                7,
901                Outcome::Channel {
902                    success: true,
903                    data: None,
904                    error: Some("status detail".to_string()),
905                },
906            )
907            .is_ok()
908        );
909
910        let large_result = ResultEnvelope::new(
911            request_id.clone(),
912            7,
913            Outcome::Channel {
914                success: true,
915                data: Some(vec![7; ENVELOPE_MAX_SIZE + 1]),
916                error: None,
917            },
918        )
919        .unwrap();
920        let large_result = encode_result(&large_result).expect("status results may exceed the request envelope limit");
921        assert!(large_result.len() > ENVELOPE_MAX_SIZE);
922        assert!(decode_result(&large_result).is_ok());
923        assert!(
924            ResultEnvelope::new(
925                request_id.clone(),
926                7,
927                Outcome::Channel {
928                    success: false,
929                    data: None,
930                    error: None,
931                },
932            )
933            .unwrap_err()
934            .contains("has no error")
935        );
936        assert!(
937            ResultEnvelope::new(
938                request_id.clone(),
939                0,
940                Outcome::Channel {
941                    success: true,
942                    data: None,
943                    error: None,
944                },
945            )
946            .unwrap_err()
947            .contains("epoch is zero")
948        );
949        assert!(
950            ResultEnvelope::new(
951                "not-a-uuid".to_string(),
952                7,
953                Outcome::Channel {
954                    success: true,
955                    data: None,
956                    error: None,
957                },
958            )
959            .unwrap_err()
960            .contains("not a UUID")
961        );
962        assert!(
963            ResultEnvelope::new(
964                request_id.clone(),
965                7,
966                Outcome::Start {
967                    task_id: "not-a-uuid".to_string(),
968                    admission: Admission::Accepted,
969                },
970            )
971            .unwrap_err()
972            .contains("not a UUID")
973        );
974        assert!(
975            ResultEnvelope::new(
976                request_id.clone(),
977                7,
978                Outcome::Channel {
979                    success: false,
980                    data: Some(vec![1]),
981                    error: Some("failed".to_string()),
982                },
983            )
984            .is_err()
985        );
986        assert!(
987            ResultEnvelope::new(
988                request_id.clone(),
989                7,
990                Outcome::Channel {
991                    success: false,
992                    data: None,
993                    error: Some(String::new()),
994                },
995            )
996            .is_err()
997        );
998
999        let valid_result = ResultEnvelope::new(
1000            request_id,
1001            7,
1002            Outcome::Channel {
1003                success: true,
1004                data: Some(vec![1]),
1005                error: None,
1006            },
1007        )
1008        .unwrap();
1009        let mut length_bomb = rmp_serde::to_vec_named(&valid_result).unwrap();
1010        let data_key = length_bomb
1011            .windows(6)
1012            .position(|window| window == b"\xa4data\x91")
1013            .expect("fixture must encode data as a one-element array");
1014        length_bomb.splice(data_key + 5..=data_key + 5, [0xdd, 0xff, 0xff, 0xff, 0xff]);
1015        assert!(decode_result(&length_bomb).is_err());
1016
1017        struct OversizedHint;
1018
1019        impl<'de> SeqAccess<'de> for OversizedHint {
1020            type Error = ValueError;
1021
1022            fn next_element_seed<T>(&mut self, _seed: T) -> Result<Option<T::Value>, Self::Error>
1023            where
1024                T: DeserializeSeed<'de>,
1025            {
1026                Ok(None)
1027            }
1028
1029            fn size_hint(&self) -> Option<usize> {
1030                Some(RESULT_MAX_SIZE + 1)
1031            }
1032        }
1033
1034        let error = super::BoundedBytesVisitor.visit_seq(OversizedHint).unwrap_err();
1035        assert!(error.to_string().contains("size limit"));
1036    }
1037}
rustfs_protos/heal_control.rs

rustfs_protos/
heal_control.rs
