Merge remote-tracking branch 'origin/cxymds/fix-1358-fleet-version-gate' into cxymds/fix-1358-gcs-exact-generation

crates/config/src/constants/object.rs

@@ -116,6 +116,39 @@ pub const ENV_OBJECT_GET_SKIP_BITROT_VERIFY: &str = "RUSTFS_OBJECT_GET_SKIP_BITR
 /// Default: bitrot verification is enabled on GetObject reads (do not skip).
 pub const DEFAULT_OBJECT_GET_SKIP_BITROT_VERIFY: bool = false;
 
+/// Request writing the complete remote-tier version state into object metadata.
+///
+/// This remains ineffective until
+/// [`ENV_TIER_REMOTE_VERSION_STATE_FLEET_CONFIRMED`] is also enabled.
+pub const ENV_TIER_REMOTE_VERSION_STATE_WRITE: &str = "RUSTFS_TIER_REMOTE_VERSION_STATE_WRITE";
+pub const DEFAULT_TIER_REMOTE_VERSION_STATE_WRITE: bool = false;
+
+/// Operator-attested fleet-wide confirmation for
+/// [`ENV_TIER_REMOTE_VERSION_STATE_WRITE`].
+///
+/// This flag is an operational contract, not automatic capability discovery.
+/// Operators may enable it only after every node that can write or read
+/// transitioned object metadata supports the remote version-state schema and
+/// semantics. Keeping the confirmation separate makes a single-node request or
+/// a writer whose local opt-in is removed fail closed.
+pub const ENV_TIER_REMOTE_VERSION_STATE_FLEET_CONFIRMED: &str = "RUSTFS_TIER_REMOTE_VERSION_STATE_FLEET_CONFIRMED";
+pub const DEFAULT_TIER_REMOTE_VERSION_STATE_FLEET_CONFIRMED: bool = false;
+
+const _: () = assert!(!DEFAULT_TIER_REMOTE_VERSION_STATE_WRITE);
+const _: () = assert!(!DEFAULT_TIER_REMOTE_VERSION_STATE_FLEET_CONFIRMED);
+
+#[cfg(test)]
+mod remote_version_state_tests {
+    #[test]
+    fn remote_version_state_gate_uses_stable_environment_names() {
+        assert_eq!(super::ENV_TIER_REMOTE_VERSION_STATE_WRITE, "RUSTFS_TIER_REMOTE_VERSION_STATE_WRITE");
+        assert_eq!(
+            super::ENV_TIER_REMOTE_VERSION_STATE_FLEET_CONFIRMED,
+            "RUSTFS_TIER_REMOTE_VERSION_STATE_FLEET_CONFIRMED"
+        );
+    }
+}
+
 // =============================================================================
 // Concurrent Request Fix - Timeout and Backpressure Configuration
 // =============================================================================

crates/ecstore/src/set_disk/ops/object.rs

@@ -2116,26 +2116,53 @@ async fn pause_transition_commit(bucket: &str, object: &str, pause: TransitionCo
 
 fn persisted_transition_version(
     remote_version: &str,
+) -> std::io::Result<(Option<String>, rustfs_filemeta::TransitionVersionState)> {
+    persisted_transition_version_with_gate(remote_version, remote_version_state_writer_enabled())
+}
+
+fn remote_version_state_writer_enabled() -> bool {
+    remote_version_state_writer_enabled_for(
+        rustfs_utils::get_env_bool(
+            rustfs_config::ENV_TIER_REMOTE_VERSION_STATE_WRITE,
+            rustfs_config::DEFAULT_TIER_REMOTE_VERSION_STATE_WRITE,
+        ),
+        rustfs_utils::get_env_bool(
+            rustfs_config::ENV_TIER_REMOTE_VERSION_STATE_FLEET_CONFIRMED,
+            rustfs_config::DEFAULT_TIER_REMOTE_VERSION_STATE_FLEET_CONFIRMED,
+        ),
+    )
+}
+
+fn remote_version_state_writer_enabled_for(requested: bool, fleet_confirmed: bool) -> bool {
+    requested && fleet_confirmed
+}
+
+fn persisted_transition_version_with_gate(
+    remote_version: &str,
+    remote_version_state_writer_enabled: bool,
 ) -> std::io::Result<(Option<String>, rustfs_filemeta::TransitionVersionState)> {
     if remote_version.is_empty() {
         return Err(std::io::Error::new(
             std::io::ErrorKind::Unsupported,
-            "a missing remote tier object version remains unknown until the cluster capability gate is active",
+            "a missing remote tier object version does not prove that bucket versioning is disabled",
         ));
     }
-    let version_id = Uuid::parse_str(remote_version).map_err(|_| {
+
-        std::io::Error::new(
+    match Uuid::parse_str(remote_version) {
-            std::io::ErrorKind::Unsupported,
+        Ok(version_id) if version_id.is_nil() => Err(std::io::Error::new(
-            "opaque remote tier versions require the cluster capability gate",
-        )
-    })?;
-    if version_id.is_nil() {
-        return Err(std::io::Error::new(
             std::io::ErrorKind::InvalidData,
             "remote tier returned a nil object version ID",
-        ));
+        )),
+        Ok(_) => Ok((Some(remote_version.to_string()), rustfs_filemeta::TransitionVersionState::Exact)),
+        Err(_) if !remote_version_state_writer_enabled => Err(std::io::Error::new(
+            std::io::ErrorKind::Unsupported,
+            "opaque remote tier versions require the operator-attested fleet gate",
+        )),
+        Err(_) if remote_version == "null" => {
+            Ok((Some(remote_version.to_string()), rustfs_filemeta::TransitionVersionState::SuspendedNull))
+        }
+        Err(_) => Ok((Some(remote_version.to_string()), rustfs_filemeta::TransitionVersionState::Exact)),
     }
-    Ok((Some(remote_version.to_string()), rustfs_filemeta::TransitionVersionState::Exact))
 }
 
 #[cfg(test)]
@@ -2297,7 +2324,10 @@ mod transition_upload_completion_tests {
 
 #[cfg(test)]
 mod transition_version_id_tests {
-    use super::{TransitionUploadCandidate, persisted_transition_version};
+    use super::{
+        TransitionUploadCandidate, persisted_transition_version, persisted_transition_version_with_gate,
+        remote_version_state_writer_enabled_for,
+    };
     use rustfs_filemeta::TransitionVersionState;
     use uuid::Uuid;
 
@@ -2333,6 +2363,44 @@ mod transition_version_id_tests {
             "opaque-version-token"
         );
     }
+
+    #[test]
+    fn remote_version_state_writer_requires_request_and_fleet_confirmation() {
+        for (case, requested, fleet_confirmed, expected) in [
+            ("old defaults", false, false, false),
+            ("missing fleet confirmation", true, false, false),
+            ("missing local opt-in", false, true, false),
+            ("explicitly unconfirmed fleet", true, false, false),
+            ("rolled-back writer", false, true, false),
+            ("fully upgraded fleet", true, true, true),
+        ] {
+            assert_eq!(remote_version_state_writer_enabled_for(requested, fleet_confirmed), expected, "{case}");
+        }
+    }
+
+    #[test]
+    fn fleet_gate_enables_null_and_opaque_remote_version_states() {
+        for (remote_version, expected) in [
+            ("null", (Some("null".to_string()), TransitionVersionState::SuspendedNull)),
+            (
+                "opaque-version-token",
+                (Some("opaque-version-token".to_string()), TransitionVersionState::Exact),
+            ),
+        ] {
+            assert!(
+                persisted_transition_version_with_gate(remote_version, false).is_err(),
+                "missing fleet confirmation must reject {remote_version:?}"
+            );
+            assert_eq!(
+                persisted_transition_version_with_gate(remote_version, true).expect("fleet-confirmed state must be persisted"),
+                expected
+            );
+        }
+        assert!(
+            persisted_transition_version_with_gate("", true).is_err(),
+            "an empty PUT response must remain unknown without authoritative bucket state"
+        );
+    }
 }
 
 #[async_trait::async_trait]