rustfs_filemeta/filemeta/
transition_reconcile.rs

1// Copyright 2026 RustFS Team
2// SPDX-License-Identifier: Apache-2.0
3
4use super::{FileMeta, FileMetaVersion};
5use crate::{Error, Result, TRANSITION_COMPLETE, TransitionVersionState};
6use rustfs_utils::http::metadata_compat::{
7    SUFFIX_TRANSITION_TIER_DESTINATION_ID, SUFFIX_TRANSITIONED_VERSION_ID, SUFFIX_TRANSITIONED_VERSION_STATE, contains_key_bytes,
8    get_consistent_bytes, insert_bytes, remove_bytes,
9};
10use serde::{Deserialize, Serialize};
11use uuid::Uuid;
12
13const RECONCILE_SUFFIXES: [&str; 3] = [
14    SUFFIX_TRANSITIONED_VERSION_STATE,
15    SUFFIX_TRANSITIONED_VERSION_ID,
16    SUFFIX_TRANSITION_TIER_DESTINATION_ID,
17];
18
19/// The only fields a legacy transition repair is allowed to persist.
20#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
21#[serde(deny_unknown_fields)]
22pub struct TransitionStateReconcileTarget {
23    pub state: TransitionVersionState,
24    pub remote_version: Option<String>,
25    pub destination_id: String,
26}
27
28impl TransitionStateReconcileTarget {
29    pub fn validate(&self) -> Result<()> {
30        let valid_version = match self.state {
31            TransitionVersionState::KnownDisabled => self.remote_version.is_none(),
32            TransitionVersionState::SuspendedNull => self.remote_version.as_deref() == Some("null"),
33            TransitionVersionState::Exact => self.remote_version.as_deref().is_some_and(|value| {
34                !value.is_empty()
35                    && value.len() <= 1024
36                    && value != "null"
37                    && !value.chars().any(char::is_control)
38                    && !Uuid::parse_str(value).is_ok_and(|id| id.is_nil())
39            }),
40            TransitionVersionState::Unknown => false,
41        };
42        if !valid_version
43            || self.destination_id.len() != 64
44            || !self
45                .destination_id
46                .bytes()
47                .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
48        {
49            return Err(Error::FileCorrupt);
50        }
51        Ok(())
52    }
53}
54
55impl FileMeta {
56    /// Canonical identity of every version and inline byte, excluding only the
57    /// three repairable suffixes on the selected version. It survives a repair
58    /// and encoding-order changes, while detecting unrelated metadata changes.
59    pub fn transition_reconcile_generation(&self, version_id: Option<Uuid>) -> Result<Vec<u8>> {
60        if self
61            .versions
62            .iter()
63            .filter(|version| version.header.version_id.unwrap_or_default() == version_id.unwrap_or_default())
64            .count()
65            != 1
66        {
67            return Err(Error::FileCorrupt);
68        }
69        let (selected, _) = self.find_version(version_id)?;
70        let mut versions = Vec::with_capacity(self.versions.len());
71        for index in 0..self.versions.len() {
72            let mut version = self.get_idx(index)?;
73            if index == selected {
74                let object = version.object.as_mut().ok_or(Error::FileCorrupt)?;
75                for suffix in RECONCILE_SUFFIXES {
76                    remove_bytes(&mut object.meta_sys, suffix);
77                }
78            }
79            versions.push(version);
80        }
81        versions.sort_by_key(|version| version.get_version_id().unwrap_or_default());
82        let mut value = serde_json::to_value((&versions, &self.data)).map_err(|_| Error::FileCorrupt)?;
83        value.sort_all_objects();
84        serde_json::to_vec(&value).map_err(|_| Error::FileCorrupt)
85    }
86
87    /// Returns false for an already converged record. Callers must serialize
88    /// the read/check/commit and compare the observed metadata generation.
89    pub fn reconcile_transition_state(
90        &mut self,
91        version_id: Option<Uuid>,
92        target: &TransitionStateReconcileTarget,
93    ) -> Result<bool> {
94        target.validate()?;
95        let (index, mut version) = self.find_version(version_id)?;
96        let info = version.into_fileinfo("", "", true)?;
97        info.validate_for_metadata_read()?;
98        if info.transition_status != TRANSITION_COMPLETE
99            || info.transition_tier.is_empty()
100            || info.transitioned_objname.is_empty()
101        {
102            return Err(Error::FileCorrupt);
103        }
104        let object = version.object.as_mut().ok_or(Error::FileCorrupt)?;
105        let destination = get_consistent_bytes(&object.meta_sys, SUFFIX_TRANSITION_TIER_DESTINATION_ID);
106        if contains_key_bytes(&object.meta_sys, SUFFIX_TRANSITION_TIER_DESTINATION_ID)
107            && destination != Some(target.destination_id.as_bytes())
108        {
109            return Err(Error::FileCorrupt);
110        }
111        if contains_key_bytes(&object.meta_sys, SUFFIX_TRANSITIONED_VERSION_STATE) {
112            if info.transition_version_state == target.state
113                && info.transition_version == target.remote_version
114                && destination == Some(target.destination_id.as_bytes())
115            {
116                return Ok(false);
117            }
118            return Err(Error::FileCorrupt);
119        }
120        if info.transition_version_state != TransitionVersionState::Unknown
121            || info
122                .transition_version
123                .as_deref()
124                .filter(|value| !value.is_empty())
125                .is_some_and(|value| Some(value) != target.remote_version.as_deref())
126        {
127            return Err(Error::FileCorrupt);
128        }
129        insert_bytes(
130            &mut object.meta_sys,
131            SUFFIX_TRANSITIONED_VERSION_STATE,
132            target.state.as_str().as_bytes().to_vec(),
133        );
134        remove_bytes(&mut object.meta_sys, SUFFIX_TRANSITIONED_VERSION_ID);
135        if let Some(version) = &target.remote_version {
136            insert_bytes(&mut object.meta_sys, SUFFIX_TRANSITIONED_VERSION_ID, version.as_bytes().to_vec());
137        }
138        insert_bytes(
139            &mut object.meta_sys,
140            SUFFIX_TRANSITION_TIER_DESTINATION_ID,
141            target.destination_id.as_bytes().to_vec(),
142        );
143        version.into_fileinfo("", "", true)?.validate_for_metadata_read()?;
144        self.set_idx(index, version)?;
145        Ok(true)
146    }
147}
148
149/// A stale healer or metadata writer may carry the original absent fields.
150/// Preserve a proven binding for the same immutable transition, or reject an
151/// attempted change of meaning. A new payload/version or a delete is separate.
152pub(super) fn preserve_reconciled_transition(previous: &FileMetaVersion, next: &mut FileMetaVersion) -> Result<()> {
153    let (Some(previous_object), Some(next_object)) = (&previous.object, &mut next.object) else {
154        return Ok(());
155    };
156    if !contains_key_bytes(&previous_object.meta_sys, SUFFIX_TRANSITION_TIER_DESTINATION_ID)
157        || !contains_key_bytes(&previous_object.meta_sys, SUFFIX_TRANSITIONED_VERSION_STATE)
158        || previous_object.data_dir != next_object.data_dir
159    {
160        return Ok(());
161    }
162    let previous_info = previous.into_fileinfo("", "", true)?;
163    if previous_info.transition_version_state == TransitionVersionState::Unknown
164        || previous_info.transition_status != TRANSITION_COMPLETE
165    {
166        return Ok(());
167    }
168    let next_info = next.into_fileinfo("", "", true)?;
169    if previous_info.transition_tier != next_info.transition_tier
170        || previous_info.transitioned_objname != next_info.transitioned_objname
171        || previous_info.transition_status != next_info.transition_status
172        || previous_info.size != next_info.size
173        || previous_info.metadata.get("etag") != next_info.metadata.get("etag")
174    {
175        return Err(Error::FileCorrupt);
176    }
177    let destination =
178        get_consistent_bytes(&previous_object.meta_sys, SUFFIX_TRANSITION_TIER_DESTINATION_ID).ok_or(Error::FileCorrupt)?;
179    previous_info.validate_for_metadata_read()?;
180    TransitionStateReconcileTarget {
181        state: previous_info.transition_version_state,
182        remote_version: previous_info.transition_version.clone(),
183        destination_id: std::str::from_utf8(destination).map_err(|_| Error::FileCorrupt)?.to_string(),
184    }
185    .validate()?;
186    let next_object = next.object.as_mut().ok_or(Error::FileCorrupt)?;
187    if next_info
188        .transition_version
189        .as_ref()
190        .is_some_and(|version| Some(version) != previous_info.transition_version.as_ref())
191    {
192        return Err(Error::FileCorrupt);
193    }
194    if contains_key_bytes(&next_object.meta_sys, SUFFIX_TRANSITIONED_VERSION_STATE) {
195        if previous_info.transition_version_state != next_info.transition_version_state
196            || previous_info.transition_version != next_info.transition_version
197        {
198            return Err(Error::FileCorrupt);
199        }
200    } else if next_info.transition_version_state != TransitionVersionState::Unknown {
201        return Err(Error::FileCorrupt);
202    }
203    if contains_key_bytes(&next_object.meta_sys, SUFFIX_TRANSITION_TIER_DESTINATION_ID)
204        && get_consistent_bytes(&next_object.meta_sys, SUFFIX_TRANSITION_TIER_DESTINATION_ID) != Some(destination)
205    {
206        return Err(Error::FileCorrupt);
207    }
208    for suffix in RECONCILE_SUFFIXES {
209        remove_bytes(&mut next_object.meta_sys, suffix);
210        if let Some(value) = get_consistent_bytes(&previous_object.meta_sys, suffix) {
211            insert_bytes(&mut next_object.meta_sys, suffix, value.to_vec());
212        }
213    }
214    Ok(())
215}
216
217#[cfg(test)]
218mod tests {
219    use super::*;
220    use crate::{ErasureInfo, FileInfo, ObjectPartInfo};
221
222    fn legacy() -> (FileMeta, FileInfo) {
223        let info = FileInfo {
224            version_id: Some(Uuid::from_u128(1)),
225            data_dir: Some(Uuid::from_u128(2)),
226            mod_time: Some(time::OffsetDateTime::from_unix_timestamp(1_700_000_000).expect("fixture time")),
227            size: 7,
228            parts: vec![ObjectPartInfo {
229                number: 1,
230                size: 7,
231                actual_size: 7,
232                ..Default::default()
233            }],
234            erasure: ErasureInfo {
235                algorithm: "ReedSolomon".to_string(),
236                data_blocks: 2,
237                parity_blocks: 2,
238                block_size: 1024 * 1024,
239                index: 1,
240                distribution: vec![1, 2, 3, 4],
241                ..Default::default()
242            },
243            transition_status: TRANSITION_COMPLETE.to_string(),
244            transition_tier: "WARM".to_string(),
245            transitioned_objname: "remote-object".to_string(),
246            metadata: std::collections::HashMap::from([("etag".to_string(), "source-etag".to_string())]),
247            data: Some(bytes::Bytes::from_static(b"payload")),
248            ..Default::default()
249        };
250        let mut metadata = FileMeta::new();
251        metadata.add_version(info.clone()).expect("legacy fixture");
252        (metadata, info)
253    }
254
255    fn target(state: TransitionVersionState) -> TransitionStateReconcileTarget {
256        TransitionStateReconcileTarget {
257            state,
258            remote_version: match state {
259                TransitionVersionState::Exact => Some("opaque-version".to_string()),
260                TransitionVersionState::SuspendedNull => Some("null".to_string()),
261                _ => None,
262            },
263            destination_id: "ab".repeat(32),
264        }
265    }
266
267    #[test]
268    fn transition_reconcile_preserves_payload_and_generation_and_is_idempotent() {
269        for state in [
270            TransitionVersionState::KnownDisabled,
271            TransitionVersionState::SuspendedNull,
272            TransitionVersionState::Exact,
273        ] {
274            let (mut metadata, info) = legacy();
275            let mut other = info.clone();
276            other.version_id = Some(Uuid::from_u128(3));
277            other.data_dir = Some(Uuid::from_u128(4));
278            other.transition_status.clear();
279            other.transition_tier.clear();
280            other.transitioned_objname.clear();
281            other.data = Some(bytes::Bytes::from_static(b"other!!"));
282            metadata.add_version(other.clone()).expect("unrelated inline version");
283            let other_before = metadata.find_version(other.version_id).expect("unrelated version").1;
284            let original_data = metadata.data.clone();
285            let generation = metadata
286                .transition_reconcile_generation(info.version_id)
287                .expect("initial generation");
288            let target = target(state);
289            assert!(metadata.reconcile_transition_state(info.version_id, &target).expect("repair"));
290            let bytes = metadata.marshal_msg().expect("encode repair");
291            let mut reloaded = FileMeta::load(&bytes).expect("reload repair");
292            assert_eq!(reloaded.data, original_data);
293            assert_eq!(
294                reloaded
295                    .find_version(other.version_id)
296                    .expect("preserved unrelated version")
297                    .1,
298                other_before
299            );
300            assert_eq!(
301                reloaded
302                    .transition_reconcile_generation(info.version_id)
303                    .expect("repaired generation"),
304                generation
305            );
306            assert!(!reloaded.reconcile_transition_state(info.version_id, &target).expect("retry"));
307            let (_, version) = reloaded.find_version(info.version_id).expect("selected version");
308            let repaired = version.into_fileinfo("", "", true).expect("decode explicit state");
309            assert_eq!(repaired.transition_version_state, state);
310            assert_eq!(repaired.transition_version, target.remote_version);
311            assert_eq!(repaired.parts, info.parts);
312            for prefix in [
313                rustfs_utils::http::RUSTFS_INTERNAL_PREFIX,
314                rustfs_utils::http::MINIO_INTERNAL_PREFIX,
315            ] {
316                assert_eq!(
317                    version
318                        .object
319                        .as_ref()
320                        .expect("object")
321                        .meta_sys
322                        .get(&format!("{prefix}{SUFFIX_TRANSITIONED_VERSION_STATE}")),
323                    Some(&state.as_str().as_bytes().to_vec())
324                );
325            }
326        }
327    }
328
329    #[test]
330    fn transition_reconcile_generation_detects_unrelated_metadata_and_inline_changes() {
331        let (mut metadata, info) = legacy();
332        let original = metadata.transition_reconcile_generation(info.version_id).expect("generation");
333        let mut updated = info.clone();
334        updated.metadata.insert("user-tag".to_string(), "changed".to_string());
335        metadata.update_object_version(updated).expect("update unrelated field");
336        assert_ne!(metadata.transition_reconcile_generation(info.version_id).expect("generation"), original);
337        let (mut metadata, mut info) = legacy();
338        info.data = Some(bytes::Bytes::from_static(b"changed"));
339        metadata.add_version(info.clone()).expect("change inline bytes");
340        assert_ne!(metadata.transition_reconcile_generation(info.version_id).expect("generation"), original);
341    }
342
343    #[test]
344    fn transition_reconcile_binding_survives_stale_heal_and_metadata_writes() {
345        let (mut metadata, mut stale) = legacy();
346        let target = target(TransitionVersionState::Exact);
347        metadata
348            .reconcile_transition_state(stale.version_id, &target)
349            .expect("repair");
350        metadata
351            .add_version(stale.clone())
352            .expect("stale heal must preserve the binding");
353        stale.metadata.insert("user-tag".to_string(), "updated".to_string());
354        metadata
355            .update_object_version(stale.clone())
356            .expect("ordinary metadata update");
357        assert!(
358            !metadata
359                .reconcile_transition_state(stale.version_id, &target)
360                .expect("binding remains exact")
361        );
362    }
363
364    #[test]
365    fn transition_reconcile_rejects_explicit_unknown_and_conflicting_binding() {
366        let (mut metadata, mut info) = legacy();
367        rustfs_utils::http::insert_str(&mut info.metadata, SUFFIX_TRANSITIONED_VERSION_STATE, "unknown".to_string());
368        metadata.add_version(info.clone()).expect("explicit unknown fixture");
369        assert!(
370            metadata
371                .reconcile_transition_state(info.version_id, &target(TransitionVersionState::Exact))
372                .is_err()
373        );
374        let (mut metadata, mut stale) = legacy();
375        metadata
376            .reconcile_transition_state(stale.version_id, &target(TransitionVersionState::Exact))
377            .expect("repair");
378        stale.transition_version_state = TransitionVersionState::KnownDisabled;
379        assert!(
380            metadata.add_version(stale).is_err(),
381            "an explicit state cannot be replaced with a different model"
382        );
383    }
384}
rustfs_filemeta/filemeta/transition_reconcile.rs

rustfs_filemeta/filemeta/
transition_reconcile.rs
