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

rustfs_protos/
heal_control.rs
