Merge branch 'main' into fix/bounded-transition-recovery

crates/e2e_test/src/upgrade_compatibility_test.rs

@@ -12,19 +12,30 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use crate::common::{RustFSTestClusterEnvironment, RustFSTestEnvironment, init_logging, rustfs_binary_path};
+use crate::common::{
+    RustFSTestClusterEnvironment, RustFSTestEnvironment, admin_request, init_logging, replication_fast_env, rustfs_binary_path,
+};
+use crate::fake_s3_target::{FAKE_ACCESS_KEY, FAKE_SECRET_KEY, FakeS3Target};
+use crate::replication_extension_test::{
+    LOOPBACK_REPLICATION_TARGET_ENV, ReplicationTargetOptions, put_bucket_replication, set_replication_target_with_options,
+};
 use aws_sdk_s3::Client;
 use aws_sdk_s3::error::ProvideErrorMetadata;
 use aws_sdk_s3::primitives::ByteStream;
 use aws_sdk_s3::types::{
-    BucketVersioningStatus, CompletedMultipartUpload, CompletedPart, ServerSideEncryption, VersioningConfiguration,
+    BucketLifecycleConfiguration, BucketVersioningStatus, CompletedMultipartUpload, CompletedPart, DefaultRetention,
+    ExpirationStatus, LifecycleExpiration, LifecycleRule, LifecycleRuleFilter, ObjectLockConfiguration, ObjectLockEnabled,
+    ObjectLockRetentionMode, ObjectLockRule, PublicAccessBlockConfiguration, ServerSideEncryption, ServerSideEncryptionByDefault,
+    ServerSideEncryptionConfiguration, ServerSideEncryptionRule, Tag, Tagging, VersioningConfiguration,
 };
+use http::{Method, StatusCode};
 use std::path::{Path, PathBuf};
 use std::time::Duration;
 use tokio::task::JoinSet;
 use tokio::time::{Instant, sleep};
 
 type TestResult = Result<(), Box<dyn std::error::Error + Send + Sync>>;
+type BoxError = Box<dyn std::error::Error + Send + Sync>;
 
 const SOURCE_BINARY_ENV: &str = "RUSTFS_UPGRADE_SOURCE_BINARY";
 const SSE_MASTER_KEY_ENV: &str = "RUSTFS_SSE_S3_MASTER_KEY";
@@ -40,6 +51,32 @@ const MULTIPART_UPLOADS_PER_WORKER: usize = 16;
 // comfortably covers that window plus CI scheduling jitter.
 const LISTING_CONVERGENCE_TIMEOUT: Duration = Duration::from_secs(30);
 
+// Bucket-configuration upgrade/rollback scenarios (rustfs#7172, #7183, #7089).
+const CONFIG_PLAIN_BUCKET: &str = "upgrade-config-plain";
+const CONFIG_ENCRYPTED_BUCKET: &str = "upgrade-config-encrypted";
+const CONFIG_REPLICATED_BUCKET: &str = "upgrade-config-replicated";
+const CONFIG_LOCKED_BUCKET: &str = "upgrade-config-locked";
+const CONFIG_REPLICA_BUCKET: &str = "upgrade-config-replica";
+const ROLLBACK_BUCKET: &str = "rollback-config-data";
+const ROLLBACK_REPLICA_BUCKET: &str = "rollback-config-replica";
+const BUCKET_QUOTA_BYTES: u64 = 64 * 1024 * 1024;
+const LIFECYCLE_RULE_ID: &str = "upgrade-expire-logs";
+const LIFECYCLE_PREFIX: &str = "logs/";
+const LIFECYCLE_DAYS: i32 = 30;
+const BUCKET_TAG_KEY: &str = "owner";
+const BUCKET_TAG_VALUE: &str = "upgrade-compatibility";
+const OBJECT_LOCK_DAYS: i32 = 1;
+// `set-bucket-quota` answers 503 until the scanner has made the bucket's usage
+// authoritative; the quota test uses the same 30s budget.
+const QUOTA_READINESS_TIMEOUT: Duration = Duration::from_secs(30);
+// Quota admission fails closed while a freshly started server has neither
+// authoritative usage nor a persisted degraded baseline for the bucket
+// (rustfs#5716), so a write to a quota-enabled bucket is retryable-503 for that
+// window. It is a restart property, not an upgrade property — the same window
+// opens on the very first start — so the write assertions ride it out instead
+// of treating it as an upgrade failure.
+const QUOTA_ADMISSION_WARMUP_TIMEOUT: Duration = Duration::from_secs(90);
+
 fn source_binary() -> Result<PathBuf, Box<dyn std::error::Error + Send + Sync>> {
     let path = std::env::var_os(SOURCE_BINARY_ENV)
         .map(PathBuf::from)
@@ -429,3 +466,653 @@ async fn rolling_upgrade_from_rc2_preserves_mixed_version_contracts() -> TestRes
 
     Ok(())
 }
+
+/// Child-process environment shared by both bucket-configuration scenarios.
+///
+/// The replication target is an in-process fake bound to `127.0.0.1`, which
+/// `set-remote-target` rejects as an SSRF risk without the loopback opt-in, and
+/// the proxy bypass keeps a developer's `HTTP_PROXY` from intercepting the
+/// server's outbound health check.
+fn bucket_config_server_env() -> Vec<(&'static str, &'static str)> {
+    let mut env = vec![
+        (SSE_MASTER_KEY_ENV, SSE_MASTER_KEY),
+        ("NO_PROXY", "127.0.0.1,localhost"),
+        ("HTTP_PROXY", ""),
+        ("HTTPS_PROXY", ""),
+        // Shorten the scanner cycle so the bucket's usage becomes authoritative
+        // in seconds; both `set-bucket-quota` and quota admission block on it.
+        ("RUSTFS_SCANNER_CYCLE", "1"),
+        ("RUSTFS_SCANNER_START_DELAY_SECS", "0"),
+    ];
+    env.extend_from_slice(LOOPBACK_REPLICATION_TARGET_ENV);
+    env.extend(replication_fast_env());
+    env
+}
+
+/// Restart `env` in place on the same data directory using an explicit binary.
+///
+/// [`RustFSTestEnvironment::restart_server_preserving_data`] always relaunches
+/// the workspace build, which is the upgrade direction only. The rollback
+/// scenario needs the reverse: stop the current build and bring the pinned
+/// previous release up on the metadata that build just wrote.
+async fn restart_from_binary(env: &mut RustFSTestEnvironment, binary: &Path, server_env: &[(&str, &str)]) -> TestResult {
+    env.stop_server();
+    env.start_rustfs_server_from_binary(binary, vec![], server_env).await
+}
+
+async fn set_bucket_quota(env: &RustFSTestEnvironment, bucket: &str, quota_bytes: u64) -> TestResult {
+    let path = format!("/rustfs/admin/v3/quota/{bucket}");
+    let body = serde_json::json!({ "quota": quota_bytes, "quota_type": "HARD" }).to_string();
+    let deadline = Instant::now() + QUOTA_READINESS_TIMEOUT;
+    loop {
+        let (status, response) =
+            admin_request(&env.url, Method::PUT, &path, Some(body.clone()), &env.access_key, &env.secret_key).await?;
+        if status.is_success() {
+            return Ok(());
+        }
+        if status != StatusCode::SERVICE_UNAVAILABLE || Instant::now() >= deadline {
+            return Err(format!("setting the quota of {bucket} failed: {status} {response}").into());
+        }
+        sleep(Duration::from_millis(500)).await;
+    }
+}
+
+/// PUT into a quota-enabled bucket, riding out the post-start quota-admission
+/// warm-up described on [`QUOTA_ADMISSION_WARMUP_TIMEOUT`].
+///
+/// Only `ServiceUnavailable` is retried: any other failure, and a warm-up that
+/// never ends, is a genuine regression and surfaces as an error.
+async fn put_object_through_quota_warmup(client: &Client, bucket: &str, key: &str, body: &'static [u8]) -> TestResult {
+    let deadline = Instant::now() + QUOTA_ADMISSION_WARMUP_TIMEOUT;
+    loop {
+        let result = client
+            .put_object()
+            .bucket(bucket)
+            .key(key)
+            .body(ByteStream::from_static(body))
+            .send()
+            .await;
+        let error = match result {
+            Ok(_) => return Ok(()),
+            Err(error) => error,
+        };
+        let retryable = error.as_service_error().and_then(ProvideErrorMetadata::code) == Some("ServiceUnavailable");
+        if !retryable || Instant::now() >= deadline {
+            return Err(format!("PUT {bucket}/{key} failed after the quota warm-up window: {error}").into());
+        }
+        sleep(Duration::from_millis(500)).await;
+    }
+}
+
+async fn get_bucket_quota(env: &RustFSTestEnvironment, bucket: &str) -> Result<Option<u64>, BoxError> {
+    let path = format!("/rustfs/admin/v3/quota/{bucket}");
+    let (status, response) = admin_request(&env.url, Method::GET, &path, None, &env.access_key, &env.secret_key).await?;
+    if status != StatusCode::OK {
+        return Err(format!("reading the quota of {bucket} failed: {status} {response}").into());
+    }
+    let quota: serde_json::Value = serde_json::from_str(&response)?;
+    Ok(quota.get("quota").and_then(serde_json::Value::as_u64))
+}
+
+/// `GET /rustfs/admin/v3/list-remote-targets?bucket=...`.
+///
+/// Returns an error for any non-200, because rustfs#7172 made this endpoint
+/// fail closed on a `bucket-targets.json` blob the running build cannot parse.
+/// An upgrade that misreads a blob written by the previous release therefore
+/// shows up here as an error, and a silently dropped target shows up as an
+/// empty list — the caller must distinguish the two.
+async fn list_remote_targets(env: &RustFSTestEnvironment, bucket: &str) -> Result<Vec<serde_json::Value>, BoxError> {
+    let path = format!("/rustfs/admin/v3/list-remote-targets?bucket={}", urlencoding::encode(bucket));
+    let (status, response) = admin_request(&env.url, Method::GET, &path, None, &env.access_key, &env.secret_key).await?;
+    if status != StatusCode::OK {
+        return Err(format!("list-remote-targets for {bucket} failed: {status} {response}").into());
+    }
+    Ok(serde_json::from_str(&response)?)
+}
+
+/// Assert that `bucket` still carries exactly the replication target `arn`.
+async fn assert_remote_target_preserved(env: &RustFSTestEnvironment, bucket: &str, arn: &str, context: &str) -> TestResult {
+    let targets = list_remote_targets(env, bucket).await?;
+    assert_eq!(
+        targets.len(),
+        1,
+        "{context}: list-remote-targets must still report the single configured target, got {targets:?}"
+    );
+    assert_eq!(
+        targets[0].get("arn").and_then(serde_json::Value::as_str),
+        Some(arn),
+        "{context}: the target ARN changed across the restart: {targets:?}"
+    );
+    Ok(())
+}
+
+/// Configure a replication target on `bucket` pointing at the in-process fake,
+/// then attach an enabled replication rule for it. Returns the target ARN.
+async fn configure_replication(
+    env: &RustFSTestEnvironment,
+    bucket: &str,
+    target: &FakeS3Target,
+    target_bucket: &str,
+) -> Result<String, BoxError> {
+    let arn = set_replication_target_with_options(
+        env,
+        bucket,
+        ReplicationTargetOptions {
+            endpoint: &target.address(),
+            access_key: FAKE_ACCESS_KEY,
+            secret_key: FAKE_SECRET_KEY,
+            target_bucket,
+            secure: false,
+            skip_tls_verify: false,
+            ca_cert_pem: None,
+        },
+    )
+    .await?;
+    put_bucket_replication(env, bucket, &arn).await?;
+    Ok(arn)
+}
+
+async fn put_default_sse_s3_encryption(client: &Client, bucket: &str) -> TestResult {
+    let configuration = ServerSideEncryptionConfiguration::builder()
+        .rules(
+            ServerSideEncryptionRule::builder()
+                .apply_server_side_encryption_by_default(
+                    ServerSideEncryptionByDefault::builder()
+                        .sse_algorithm(ServerSideEncryption::Aes256)
+                        .build()?,
+                )
+                .build(),
+        )
+        .build()?;
+    client
+        .put_bucket_encryption()
+        .bucket(bucket)
+        .server_side_encryption_configuration(configuration)
+        .send()
+        .await?;
+    Ok(())
+}
+
+async fn assert_default_sse_s3_encryption(client: &Client, bucket: &str, context: &str) -> TestResult {
+    let response = client.get_bucket_encryption().bucket(bucket).send().await?;
+    let rules = response
+        .server_side_encryption_configuration()
+        .ok_or("GetBucketEncryption omitted the configuration")?
+        .rules();
+    assert_eq!(rules.len(), 1, "{context}: expected exactly one encryption rule, got {rules:?}");
+    assert_eq!(
+        rules[0]
+            .apply_server_side_encryption_by_default()
+            .map(ServerSideEncryptionByDefault::sse_algorithm),
+        Some(&ServerSideEncryption::Aes256),
+        "{context}: the default encryption algorithm changed"
+    );
+    Ok(())
+}
+
+async fn put_bucket_tag(client: &Client, bucket: &str) -> TestResult {
+    let tagging = Tagging::builder()
+        .tag_set(Tag::builder().key(BUCKET_TAG_KEY).value(BUCKET_TAG_VALUE).build()?)
+        .build()?;
+    client.put_bucket_tagging().bucket(bucket).tagging(tagging).send().await?;
+    Ok(())
+}
+
+async fn assert_bucket_tag(client: &Client, bucket: &str, context: &str) -> TestResult {
+    let tags = client.get_bucket_tagging().bucket(bucket).send().await?;
+    let tag_set = tags.tag_set();
+    assert_eq!(tag_set.len(), 1, "{context}: expected exactly one bucket tag, got {tag_set:?}");
+    assert_eq!(tag_set[0].key(), BUCKET_TAG_KEY, "{context}: bucket tag key changed");
+    assert_eq!(tag_set[0].value(), BUCKET_TAG_VALUE, "{context}: bucket tag value changed");
+    Ok(())
+}
+
+async fn assert_versioning_enabled(client: &Client, bucket: &str, context: &str) -> TestResult {
+    let versioning = client.get_bucket_versioning().bucket(bucket).send().await?;
+    assert_eq!(
+        versioning.status(),
+        Some(&BucketVersioningStatus::Enabled),
+        "{context}: versioning is no longer Enabled on {bucket}"
+    );
+    Ok(())
+}
+
+fn bucket_policy_document(bucket: &str) -> serde_json::Value {
+    serde_json::json!({
+        "Version": "2012-10-17",
+        "Statement": [{
+            "Sid": "UpgradePublicRead",
+            "Effect": "Allow",
+            "Principal": { "AWS": ["*"] },
+            "Action": ["s3:GetObject"],
+            "Resource": [format!("arn:aws:s3:::{bucket}/public/*")]
+        }]
+    })
+}
+
+/// `GET .../on-demand-migration/{bucket}/status`.
+///
+/// The migration module defaults on from rustfs#7089, so a bucket that never
+/// configured a source must still answer `configured: false` rather than
+/// engaging the migration path.
+async fn assert_migration_not_configured(env: &RustFSTestEnvironment, bucket: &str) -> TestResult {
+    let path = format!("/rustfs/admin/v3/on-demand-migration/{bucket}/status");
+    let (status, response) = admin_request(&env.url, Method::GET, &path, None, &env.access_key, &env.secret_key).await?;
+    assert_eq!(
+        status,
+        StatusCode::OK,
+        "the migration status endpoint must answer for an unconfigured bucket: {status} {response}"
+    );
+    let body: serde_json::Value = serde_json::from_str(&response)?;
+    assert_eq!(
+        body.get("configured"),
+        Some(&serde_json::Value::Bool(false)),
+        "a bucket upgraded from the previous release must not look migration-configured: {body}"
+    );
+    Ok(())
+}
+
+/// A GET for a key that was never written must be a plain `NoSuchKey`.
+///
+/// With the migration module on by default this is the cheap proof that an
+/// unconfigured bucket never consults a source: any migration engagement would
+/// surface as a different status or error code here.
+async fn assert_missing_key_is_no_such_key(client: &Client, bucket: &str, key: &str) -> TestResult {
+    let error = client
+        .get_object()
+        .bucket(bucket)
+        .key(key)
+        .send()
+        .await
+        .expect_err("a key that was never written must not be readable");
+    assert_eq!(
+        error.raw_response().map(|response| response.status().as_u16()),
+        Some(404),
+        "a missing key must stay a 404 on a bucket with no migration configuration"
+    );
+    assert_eq!(
+        error.as_service_error().and_then(ProvideErrorMetadata::code),
+        Some("NoSuchKey"),
+        "a missing key must stay NoSuchKey on a bucket with no migration configuration"
+    );
+    Ok(())
+}
+
+/// Bucket configuration written by the pinned previous release must survive an
+/// upgrade to the current build unchanged, and must keep working.
+///
+/// This pins the three on-disk surfaces the on-demand-migration series moved:
+///
+/// * `BucketMetadata` grew two msgpack keys (encoded map length 44 -> 46), so
+///   every configuration read below decodes a 44-key blob on 46-key code.
+/// * rustfs#7172 made an unreadable `bucket-targets.json` / encryption /
+///   public-access-block / quota blob "present but unreadable" instead of
+///   silently defaulting, and made `list-remote-targets` fail closed on it. A
+///   replication target configured by the old release must therefore still be
+///   *listed*, not dropped and not an error.
+/// * rustfs#7183 made the object write path refuse a PUT when the bucket's
+///   encryption configuration cannot be read, so a misparsed SSE config would
+///   turn every PUT to that bucket into a 500.
+///
+/// Not covered on purpose: on-demand-migration configuration itself, which the
+/// previous release has no public API for — the reverse direction is asserted
+/// instead (an upgraded bucket reports `configured: false`).
+#[tokio::test]
+#[ignore = "requires a pinned previous RustFS release binary"]
+async fn direct_upgrade_from_previous_release_preserves_bucket_configuration() -> TestResult {
+    init_logging();
+    let previous_binary = source_binary()?;
+
+    // In-process: the fake target outlives both server processes, so the
+    // replication target stays reachable across the upgrade.
+    let replication_target = FakeS3Target::start().await?;
+    replication_target.create_bucket(CONFIG_REPLICA_BUCKET);
+
+    let mut env = RustFSTestEnvironment::new().await?;
+    let server_env = bucket_config_server_env();
+    env.start_rustfs_server_from_binary(&previous_binary, vec![], &server_env)
+        .await?;
+    let old_client = env.create_s3_client();
+
+    env.create_test_bucket(CONFIG_PLAIN_BUCKET).await?;
+    env.create_test_bucket(CONFIG_ENCRYPTED_BUCKET).await?;
+    env.create_test_bucket(CONFIG_REPLICATED_BUCKET).await?;
+    old_client
+        .create_bucket()
+        .bucket(CONFIG_LOCKED_BUCKET)
+        .object_lock_enabled_for_bucket(true)
+        .send()
+        .await?;
+
+    // Plain bucket: policy, tags, lifecycle, quota.
+    let policy = bucket_policy_document(CONFIG_PLAIN_BUCKET);
+    old_client
+        .put_bucket_policy()
+        .bucket(CONFIG_PLAIN_BUCKET)
+        .policy(policy.to_string())
+        .send()
+        .await?;
+    put_bucket_tag(&old_client, CONFIG_PLAIN_BUCKET).await?;
+    old_client
+        .put_bucket_lifecycle_configuration()
+        .bucket(CONFIG_PLAIN_BUCKET)
+        .lifecycle_configuration(
+            BucketLifecycleConfiguration::builder()
+                .rules(
+                    LifecycleRule::builder()
+                        .id(LIFECYCLE_RULE_ID)
+                        .status(ExpirationStatus::Enabled)
+                        .filter(LifecycleRuleFilter::builder().prefix(LIFECYCLE_PREFIX).build())
+                        .expiration(LifecycleExpiration::builder().days(LIFECYCLE_DAYS).build())
+                        .build()?,
+                )
+                .build()?,
+        )
+        .send()
+        .await?;
+    set_bucket_quota(&env, CONFIG_PLAIN_BUCKET, BUCKET_QUOTA_BYTES).await?;
+
+    // Encrypted bucket: SSE-S3 default encryption plus a fully restrictive
+    // public access block, both of which rustfs#7172 now fails closed on.
+    put_default_sse_s3_encryption(&old_client, CONFIG_ENCRYPTED_BUCKET).await?;
+    old_client
+        .put_public_access_block()
+        .bucket(CONFIG_ENCRYPTED_BUCKET)
+        .public_access_block_configuration(
+            PublicAccessBlockConfiguration::builder()
+                .block_public_acls(true)
+                .ignore_public_acls(true)
+                .block_public_policy(true)
+                .restrict_public_buckets(true)
+                .build(),
+        )
+        .send()
+        .await?;
+
+    // Replicated bucket: versioning, a validated remote target, a rule.
+    enable_versioning(&old_client, CONFIG_REPLICATED_BUCKET).await?;
+    let target_arn = configure_replication(&env, CONFIG_REPLICATED_BUCKET, &replication_target, CONFIG_REPLICA_BUCKET).await?;
+    assert_remote_target_preserved(&env, CONFIG_REPLICATED_BUCKET, &target_arn, "before the upgrade").await?;
+
+    // Object-lock bucket: a default GOVERNANCE retention on a fresh bucket.
+    old_client
+        .put_object_lock_configuration()
+        .bucket(CONFIG_LOCKED_BUCKET)
+        .object_lock_configuration(
+            ObjectLockConfiguration::builder()
+                .object_lock_enabled(ObjectLockEnabled::Enabled)
+                .rule(
+                    ObjectLockRule::builder()
+                        .default_retention(
+                            DefaultRetention::builder()
+                                .mode(ObjectLockRetentionMode::Governance)
+                                .days(OBJECT_LOCK_DAYS)
+                                .build(),
+                        )
+                        .build(),
+                )
+                .build(),
+        )
+        .send()
+        .await?;
+
+    let plain_key = "plain/written-by-previous";
+    let plain_bytes = b"plain object written by the previous RustFS release";
+    put_object_through_quota_warmup(&old_client, CONFIG_PLAIN_BUCKET, plain_key, plain_bytes).await?;
+
+    let encrypted_key = "encrypted/written-by-previous";
+    let encrypted_bytes = b"default-encrypted object written by the previous RustFS release";
+    old_client
+        .put_object()
+        .bucket(CONFIG_ENCRYPTED_BUCKET)
+        .key(encrypted_key)
+        .body(ByteStream::from_static(encrypted_bytes))
+        .send()
+        .await?;
+    assert_eq!(
+        read_object(&old_client, CONFIG_ENCRYPTED_BUCKET, encrypted_key, None)
+            .await?
+            .0,
+        Some(ServerSideEncryption::Aes256),
+        "the previous release must apply the bucket default encryption it just accepted"
+    );
+
+    // The multipart object lives in the default-encrypted bucket so the
+    // upgraded build has to reassemble parts *and* re-derive the object key.
+    let multipart_key = "encrypted/multipart-written-by-previous";
+    let multipart_parts = vec![vec![b'm'; 5 * 1024 * 1024], b"final multipart bytes".to_vec()];
+    let multipart_bytes = multipart_parts.concat();
+    write_multipart(&old_client, CONFIG_ENCRYPTED_BUCKET, multipart_key, &multipart_parts).await?;
+
+    let versioned_key = "versioned/written-by-previous";
+    let versioned_bytes = b"versioned object written by the previous RustFS release";
+    let versioned_id = old_client
+        .put_object()
+        .bucket(CONFIG_REPLICATED_BUCKET)
+        .key(versioned_key)
+        .body(ByteStream::from_static(versioned_bytes))
+        .send()
+        .await?
+        .version_id()
+        .ok_or("versioned PUT omitted version ID")?
+        .to_string();
+
+    env.restart_server_preserving_data(vec![], &server_env).await?;
+    let new_client = env.create_s3_client();
+
+    // Every configuration must read back unchanged on the upgraded build.
+    let upgraded_policy = new_client.get_bucket_policy().bucket(CONFIG_PLAIN_BUCKET).send().await?;
+    let upgraded_policy: serde_json::Value =
+        serde_json::from_str(upgraded_policy.policy().ok_or("GetBucketPolicy omitted the document")?)?;
+    assert_eq!(upgraded_policy, policy, "the bucket policy changed across the upgrade");
+    assert_bucket_tag(&new_client, CONFIG_PLAIN_BUCKET, "after the upgrade").await?;
+
+    let lifecycle = new_client
+        .get_bucket_lifecycle_configuration()
+        .bucket(CONFIG_PLAIN_BUCKET)
+        .send()
+        .await?;
+    let rules = lifecycle.rules();
+    assert_eq!(rules.len(), 1, "the lifecycle rule count changed across the upgrade: {rules:?}");
+    assert_eq!(rules[0].id(), Some(LIFECYCLE_RULE_ID));
+    assert_eq!(rules[0].status(), &ExpirationStatus::Enabled);
+    assert_eq!(
+        rules[0].expiration().and_then(LifecycleExpiration::days),
+        Some(LIFECYCLE_DAYS),
+        "the lifecycle expiration changed across the upgrade"
+    );
+
+    assert_eq!(
+        get_bucket_quota(&env, CONFIG_PLAIN_BUCKET).await?,
+        Some(BUCKET_QUOTA_BYTES),
+        "the bucket quota changed across the upgrade"
+    );
+
+    assert_default_sse_s3_encryption(&new_client, CONFIG_ENCRYPTED_BUCKET, "after the upgrade").await?;
+    let public_access_block = new_client
+        .get_public_access_block()
+        .bucket(CONFIG_ENCRYPTED_BUCKET)
+        .send()
+        .await?;
+    let public_access_block = public_access_block
+        .public_access_block_configuration()
+        .ok_or("GetPublicAccessBlock omitted the configuration")?;
+    assert_eq!(public_access_block.block_public_acls(), Some(true));
+    assert_eq!(public_access_block.ignore_public_acls(), Some(true));
+    assert_eq!(public_access_block.block_public_policy(), Some(true));
+    assert_eq!(public_access_block.restrict_public_buckets(), Some(true));
+
+    assert_versioning_enabled(&new_client, CONFIG_REPLICATED_BUCKET, "after the upgrade").await?;
+    // rustfs#7172: neither an empty list nor an error is acceptable here.
+    assert_remote_target_preserved(&env, CONFIG_REPLICATED_BUCKET, &target_arn, "after the upgrade").await?;
+    let replication = new_client
+        .get_bucket_replication()
+        .bucket(CONFIG_REPLICATED_BUCKET)
+        .send()
+        .await?;
+    let replication_rules = replication
+        .replication_configuration()
+        .ok_or("GetBucketReplication omitted the configuration")?
+        .rules();
+    assert_eq!(
+        replication_rules.len(),
+        1,
+        "the replication rule count changed across the upgrade: {replication_rules:?}"
+    );
+    assert_eq!(
+        replication_rules[0].destination().map(|destination| destination.bucket()),
+        Some(target_arn.as_str()),
+        "the replication rule no longer points at the configured target"
+    );
+
+    let object_lock = new_client
+        .get_object_lock_configuration()
+        .bucket(CONFIG_LOCKED_BUCKET)
+        .send()
+        .await?;
+    let object_lock = object_lock
+        .object_lock_configuration()
+        .ok_or("GetObjectLockConfiguration omitted the configuration")?;
+    assert_eq!(object_lock.object_lock_enabled(), Some(&ObjectLockEnabled::Enabled));
+    let retention = object_lock
+        .rule()
+        .and_then(ObjectLockRule::default_retention)
+        .ok_or("the object lock configuration lost its default retention")?;
+    assert_eq!(retention.mode(), Some(&ObjectLockRetentionMode::Governance));
+    assert_eq!(retention.days(), Some(OBJECT_LOCK_DAYS));
+
+    // rustfs#7183: a PUT into the default-encrypted bucket must still succeed
+    // and still come back encrypted.
+    let post_upgrade_encrypted_key = "encrypted/written-after-upgrade";
+    let post_upgrade_encrypted_bytes = b"default-encrypted object written by the current RustFS build";
+    new_client
+        .put_object()
+        .bucket(CONFIG_ENCRYPTED_BUCKET)
+        .key(post_upgrade_encrypted_key)
+        .body(ByteStream::from_static(post_upgrade_encrypted_bytes))
+        .send()
+        .await?;
+    let (encryption, body) = read_object(&new_client, CONFIG_ENCRYPTED_BUCKET, post_upgrade_encrypted_key, None).await?;
+    assert_eq!(
+        encryption,
+        Some(ServerSideEncryption::Aes256),
+        "a PUT after the upgrade lost the bucket default encryption"
+    );
+    assert_eq!(body, post_upgrade_encrypted_bytes);
+
+    let post_upgrade_plain_key = "plain/written-after-upgrade";
+    let post_upgrade_plain_bytes = b"plain object written by the current RustFS build";
+    put_object_through_quota_warmup(&new_client, CONFIG_PLAIN_BUCKET, post_upgrade_plain_key, post_upgrade_plain_bytes).await?;
+    let (encryption, body) = read_object(&new_client, CONFIG_PLAIN_BUCKET, post_upgrade_plain_key, None).await?;
+    assert_eq!(encryption, None, "a bucket without default encryption must not encrypt a PUT");
+    assert_eq!(body, post_upgrade_plain_bytes);
+
+    // Every object written by the previous release reads back byte-identical.
+    assert_eq!(read_object(&new_client, CONFIG_PLAIN_BUCKET, plain_key, None).await?.1, plain_bytes);
+    let (encryption, body) = read_object(&new_client, CONFIG_ENCRYPTED_BUCKET, encrypted_key, None).await?;
+    assert_eq!(encryption, Some(ServerSideEncryption::Aes256));
+    assert_eq!(body, encrypted_bytes);
+    let (encryption, body) = read_object(&new_client, CONFIG_ENCRYPTED_BUCKET, multipart_key, None).await?;
+    assert_eq!(encryption, Some(ServerSideEncryption::Aes256));
+    assert_eq!(body, multipart_bytes, "the multipart object did not survive the upgrade");
+    assert_eq!(
+        read_object(&new_client, CONFIG_REPLICATED_BUCKET, versioned_key, Some(&versioned_id))
+            .await?
+            .1,
+        versioned_bytes
+    );
+
+    // rustfs#7089: the migration module is on by default, but a bucket that
+    // never configured a source behaves exactly as before.
+    assert_migration_not_configured(&env, CONFIG_PLAIN_BUCKET).await?;
+    assert_missing_key_is_no_such_key(&new_client, CONFIG_PLAIN_BUCKET, "plain/never-written").await?;
+
+    replication_target.shutdown().await;
+    Ok(())
+}
+
+/// Rolling back to the pinned previous release must still read the bucket
+/// metadata the current build wrote.
+///
+/// This is the other half of the `BucketMetadata` 44 -> 46 key change: the
+/// current build writes a 46-key msgpack map with `OnDemandMigrationConfigJSON`
+/// and `OnDemandMigrationConfigUpdatedAt`, and the previous release's decoder
+/// has to skip those two unknown keys instead of failing the whole blob. If it
+/// did not, every configuration read below would come back empty or error and
+/// the rollback would silently discard the bucket's configuration.
+#[tokio::test]
+#[ignore = "requires a pinned previous RustFS release binary"]
+async fn rollback_to_previous_release_reads_current_bucket_metadata() -> TestResult {
+    init_logging();
+    let previous_binary = source_binary()?;
+
+    let replication_target = FakeS3Target::start().await?;
+    replication_target.create_bucket(ROLLBACK_REPLICA_BUCKET);
+
+    let mut env = RustFSTestEnvironment::new().await?;
+    let server_env = bucket_config_server_env();
+    env.start_rustfs_server_with_env(vec![], &server_env).await?;
+    let new_client = env.create_s3_client();
+
+    env.create_test_bucket(ROLLBACK_BUCKET).await?;
+    enable_versioning(&new_client, ROLLBACK_BUCKET).await?;
+    put_default_sse_s3_encryption(&new_client, ROLLBACK_BUCKET).await?;
+    put_bucket_tag(&new_client, ROLLBACK_BUCKET).await?;
+    let target_arn = configure_replication(&env, ROLLBACK_BUCKET, &replication_target, ROLLBACK_REPLICA_BUCKET).await?;
+    assert_remote_target_preserved(&env, ROLLBACK_BUCKET, &target_arn, "before the rollback").await?;
+
+    let single_key = "rollback/single";
+    let single_bytes = b"single-part object written by the current RustFS build";
+    let single_version = new_client
+        .put_object()
+        .bucket(ROLLBACK_BUCKET)
+        .key(single_key)
+        .body(ByteStream::from_static(single_bytes))
+        .send()
+        .await?
+        .version_id()
+        .ok_or("versioned PUT omitted version ID")?
+        .to_string();
+
+    let multipart_key = "rollback/multipart";
+    let multipart_parts = vec![vec![b'r'; 5 * 1024 * 1024], b"final rollback bytes".to_vec()];
+    let multipart_bytes = multipart_parts.concat();
+    write_multipart(&new_client, ROLLBACK_BUCKET, multipart_key, &multipart_parts).await?;
+
+    restart_from_binary(&mut env, &previous_binary, &server_env).await?;
+    let old_client = env.create_s3_client();
+
+    assert_versioning_enabled(&old_client, ROLLBACK_BUCKET, "after the rollback").await?;
+    assert_default_sse_s3_encryption(&old_client, ROLLBACK_BUCKET, "after the rollback").await?;
+    assert_bucket_tag(&old_client, ROLLBACK_BUCKET, "after the rollback").await?;
+    assert_remote_target_preserved(&env, ROLLBACK_BUCKET, &target_arn, "after the rollback").await?;
+
+    let (encryption, body) = read_object(&old_client, ROLLBACK_BUCKET, single_key, Some(&single_version)).await?;
+    assert_eq!(encryption, Some(ServerSideEncryption::Aes256));
+    assert_eq!(body, single_bytes);
+    let (encryption, body) = read_object(&old_client, ROLLBACK_BUCKET, multipart_key, None).await?;
+    assert_eq!(encryption, Some(ServerSideEncryption::Aes256));
+    assert_eq!(body, multipart_bytes, "the multipart object did not survive the rollback");
+
+    // A PUT on the rolled-back release must still honour the encryption
+    // configuration it decoded out of the current build's metadata blob.
+    let post_rollback_key = "rollback/written-after-rollback";
+    let post_rollback_bytes = b"object written by the previous RustFS release after the rollback";
+    old_client
+        .put_object()
+        .bucket(ROLLBACK_BUCKET)
+        .key(post_rollback_key)
+        .body(ByteStream::from_static(post_rollback_bytes))
+        .send()
+        .await?;
+    let (encryption, body) = read_object(&old_client, ROLLBACK_BUCKET, post_rollback_key, None).await?;
+    assert_eq!(
+        encryption,
+        Some(ServerSideEncryption::Aes256),
+        "the rolled-back release lost the bucket default encryption"
+    );
+    assert_eq!(body, post_rollback_bytes);
+
+    replication_target.shutdown().await;
+    Ok(())
+}

crates/ecstore/Cargo.toml

@@ -31,6 +31,7 @@ workspace = true
 
 [features]
 default = []
+gcs = ["dep:google-cloud-storage", "dep:google-cloud-auth"]
 # Compiles the controlled list-objects namespace-journal chaos injector into a
 # production binary (it is always available to tests). Off by default so the
 # RUSTFS_LIST_OBJECTS_NAMESPACE_JOURNAL_CHAOS_* env vars cannot rewrite journal
@@ -212,8 +213,8 @@ aws-smithy-runtime-api = { workspace = true, features = ["http-1x"] }
 parking_lot = { workspace = true }
 base64-simd.workspace = true
 serde_urlencoded.workspace = true
-google-cloud-storage = { workspace = true }
-google-cloud-auth = { workspace = true }
+google-cloud-storage = { workspace = true, optional = true }
+google-cloud-auth = { workspace = true, optional = true }
 faster-hex = { workspace = true }
 quick-xml = { workspace = true }
 ratelimit = { workspace = true }

crates/ecstore/src/bucket/on_demand_migration/gcs.rs

@@ -124,6 +124,18 @@ impl GcsNativeSourceBackend {
         Ok(request)
     }
 
+    async fn send_object(&self, request: reqwest::Request) -> Result<reqwest::Response, SourceError> {
+        match self.http.send_object(request, NO_ERROR_CODE_HEADER).await {
+            Err(SourceError::NotFound) => {
+                // An XML object URL also returns 404 when its bucket is gone.
+                // Reuse the read-only listing probe before caching a key miss.
+                self.probe().await?;
+                Err(SourceError::NotFound)
+            }
+            result => result,
+        }
+    }
+
     /// Shared mapping for the XML API's HEAD and GET responses.
     fn head_from_response(headers: &HeaderMap) -> Result<SourceHead, SourceError> {
         if header(headers, "x-goog-encryption-key-sha256").is_some() {
@@ -164,7 +176,7 @@ impl GcsNativeSourceBackend {
 impl SourceBackend for GcsNativeSourceBackend {
     async fn head(&self, key: &str) -> Result<SourceHead, SourceError> {
         let request = self.request(Method::HEAD, self.object_url(key)?, HeaderMap::new()).await?;
-        let response = self.http.send(request, NO_ERROR_CODE_HEADER).await?;
+        let response = self.send_object(request).await?;
         Self::head_from_response(response.headers())
     }
 
@@ -177,7 +189,7 @@ impl SourceBackend for GcsNativeSourceBackend {
             );
         }
         let request = self.request(Method::GET, self.object_url(key)?, headers).await?;
-        let response = self.http.send(request, NO_ERROR_CODE_HEADER).await?;
+        let response = self.send_object(request).await?;
         let head = Self::head_from_response(response.headers())?;
         let content_range = header(response.headers(), "content-range").map(str::to_string);
         Ok(SourceGet {
@@ -488,6 +500,7 @@ mod tests {
             // request; the probe is the next one on the wire.
             ScriptedResponse::new(200, Vec::new(), "{}".to_string()),
             ScriptedResponse::new(404, Vec::new(), String::new()),
+            ScriptedResponse::new(200, Vec::new(), "{}".to_string()),
             ScriptedResponse::new(403, Vec::new(), String::new()),
         ])
         .await;
@@ -503,4 +516,49 @@ mod tests {
         )
         .await;
     }
+
+    #[tokio::test]
+    async fn listing_404_is_not_an_object_not_found() {
+        let (endpoint, _) = scripted_server(vec![ScriptedResponse::new(404, Vec::new(), String::new())]).await;
+        let err = backend(&endpoint)
+            .list(&SourceListRequest {
+                max_keys: 1,
+                ..Default::default()
+            })
+            .await
+            .expect_err("a failed bucket listing is not a per-object miss");
+        assert_eq!(err.class_label(), "other", "{err:?}");
+    }
+
+    #[tokio::test]
+    async fn object_404_requires_a_readable_source_bucket() {
+        for method in [Method::HEAD, Method::GET] {
+            for (probe_status, expected_class) in [
+                (200, "not_found"),
+                (404, "other"),
+                (403, "access_denied"),
+                (503, "throttled"),
+                (500, "server_error"),
+            ] {
+                let (endpoint, recorded) = scripted_server(vec![
+                    ScriptedResponse::new(404, Vec::new(), String::new()),
+                    ScriptedResponse::new(probe_status, Vec::new(), "{}".to_string()),
+                ])
+                .await;
+                let backend = backend(&endpoint);
+                let result = if method == Method::HEAD {
+                    backend.head("missing").await.map(|_| ())
+                } else {
+                    backend.get("missing", None).await.map(|_| ())
+                };
+                let error = result.expect_err("the object 404 must remain an error");
+                assert_eq!(error.class_label(), expected_class, "{method} with probe HTTP {probe_status}: {error:?}");
+                let recorded = recorded.lock().expect("recorder lock");
+                assert_eq!(recorded.len(), 2, "one bounded read-only probe per ambiguous object miss");
+                assert_eq!(recorded[0].method, method.as_str());
+                assert_eq!(recorded[1].method, "GET");
+                assert_eq!(recorded[1].target, "/storage/v1/b/legacy/o?maxResults=1");
+            }
+        }
+    }
 }

crates/ecstore/src/bucket/on_demand_migration/list_through.rs

@@ -1089,10 +1089,17 @@ mod tests {
 
     #[test]
     fn progress_tokens_preserve_v1_bytes_and_validate_v2_counts() {
+        fn framed(payload: &str) -> String {
+            format!("{LIST_THROUGH_TOKEN_PREFIX}{payload}")
+        }
+
         let token = progress_token(None, true, false);
         assert_eq!(
             token.encode(),
-            r#"{"t":"odm-list","v":1,"local":null,"local_done":true,"source":"A","source_done":false,"last_key":"last-key"}"#
+            concat!(
+                "\0odm-list:",
+                r#"{"t":"odm-list","v":1,"local":null,"local_done":true,"source":"A","source_done":false,"last_key":"last-key"}"#
+            )
         );
         for count in 1..MAX_LIST_NO_PROGRESS_PAGES {
             let token = progress_token(Some(count), true, false);
@@ -1100,16 +1107,17 @@ mod tests {
         }
         for version in [1, 2] {
             for value in ["null", "0", "16", "-1", "1.5", "256", "18446744073709551616", "\"1\""] {
-                let encoded = format!(r#"{{"t":"odm-list","v":{version},"no_progress":{value}}}"#);
+                let encoded = framed(&format!(r#"{{"t":"odm-list","v":{version},"no_progress":{value}}}"#));
                 assert_eq!(decode_continuation_token(&encoded), Err(ListThroughTokenError::Malformed), "{encoded}");
             }
         }
-        for encoded in [
+        for payload in [
             r#"{"t":"odm-list","v":1,"no_progress":1}"#,
             r#"{"t":"odm-list","v":2}"#,
             r#"{"t":"odm-list","v":2,"no_progress":1,"extra":true}"#,
         ] {
-            assert_eq!(decode_continuation_token(encoded), Err(ListThroughTokenError::Malformed), "{encoded}");
+            let encoded = framed(payload);
+            assert_eq!(decode_continuation_token(&encoded), Err(ListThroughTokenError::Malformed), "{encoded}");
         }
     }
 

crates/ecstore/src/bucket/on_demand_migration/mod.rs

@@ -30,6 +30,7 @@ mod backend_contract;
 pub mod backfill;
 pub mod breaker;
 pub mod config;
+#[cfg(feature = "gcs")]
 pub mod gcs;
 pub mod list_through;
 mod native_http;

crates/ecstore/src/bucket/on_demand_migration/native_http.rs

@@ -129,6 +129,23 @@ impl NativeHttp {
         &self,
         request: reqwest::Request,
         error_code_header: &str,
+    ) -> Result<reqwest::Response, SourceError> {
+        self.send_classified(request, error_code_header, false).await
+    }
+
+    pub(super) async fn send_object(
+        &self,
+        request: reqwest::Request,
+        error_code_header: &str,
+    ) -> Result<reqwest::Response, SourceError> {
+        self.send_classified(request, error_code_header, true).await
+    }
+
+    async fn send_classified(
+        &self,
+        request: reqwest::Request,
+        error_code_header: &str,
+        not_found_on_404_without_code: bool,
     ) -> Result<reqwest::Response, SourceError> {
         let response = self.client.execute(request).await.map_err(classify_transport_error)?;
         let status = response.status();
@@ -140,14 +157,14 @@ impl NativeHttp {
             .get(error_code_header)
             .and_then(|value| value.to_str().ok())
             .map(str::to_string);
-        Err(classify_status(
-            status.as_u16(),
-            None,
-            match &code {
-                Some(code) => format!("source returned HTTP {status} ({code})"),
-                None => format!("source returned HTTP {status}"),
-            },
-        ))
+        let message = match &code {
+            Some(code) => format!("source returned HTTP {status} ({code})"),
+            None => format!("source returned HTTP {status}"),
+        };
+        match classify_status(status.as_u16(), code.as_deref(), message) {
+            SourceError::Other(_) if not_found_on_404_without_code && status.as_u16() == 404 => Err(SourceError::NotFound),
+            err => Err(err),
+        }
     }
 }
 

crates/ecstore/src/bucket/on_demand_migration/source_client.rs

@@ -26,6 +26,7 @@
 //! forwarded: v1 rejects SSE-C source objects outright.
 
 use super::azure::AzureSourceBackend;
+#[cfg(feature = "gcs")]
 use super::gcs::GcsNativeSourceBackend;
 use super::list_through::{ListPageError, validate_list_page};
 use crate::bucket::remote_s3_client::{
@@ -334,8 +335,9 @@ const THROTTLE_CODES: &[&str] = &[
     "RequestLimitExceeded",
     "TooManyRequests",
     "RequestThrottled",
+    "ServerBusy",
 ];
-const NOT_FOUND_CODES: &[&str] = &["NoSuchKey"];
+const NOT_FOUND_CODES: &[&str] = &["NoSuchKey", "BlobNotFound"];
 const ACCESS_DENIED_CODES: &[&str] = &[
     "AccessDenied",
     "InvalidAccessKeyId",
@@ -343,6 +345,7 @@ const ACCESS_DENIED_CODES: &[&str] = &[
     "AllAccessDisabled",
     "ExpiredToken",
     "InvalidToken",
+    "AuthorizationPermissionMismatch",
 ];
 
 pub(super) fn classify_status(status: u16, code: Option<&str>, message: String) -> SourceError {
@@ -720,6 +723,9 @@ impl SourceClient {
                 )?;
                 Ok(Self::from_backend(Box::new(backend), spec))
             }
+            #[cfg(not(feature = "gcs"))]
+            SourceBackendSpec::Gcs(_) => Err(RemoteS3ClientError::BackendNotCompiled("gcs_native")),
+            #[cfg(feature = "gcs")]
             SourceBackendSpec::Gcs(gcs) => {
                 let backend = GcsNativeSourceBackend::new(
                     &spec.endpoint,
@@ -1111,6 +1117,28 @@ mod tests {
         }
     }
 
+    #[cfg(not(feature = "gcs"))]
+    #[tokio::test]
+    async fn gcs_backend_not_compiled_keeps_hmac_s3_available() {
+        let mut native = spec(None);
+        native.provider = SourceProvider::GcsNative;
+        native.credentials = None;
+        native.backend = SourceBackendSpec::Gcs(GcsSourceSpec {
+            service_account_json: "{}".to_string(),
+        });
+        assert!(matches!(
+            SourceClient::new(&native).await,
+            Err(RemoteS3ClientError::BackendNotCompiled("gcs_native"))
+        ));
+
+        let mut hmac = spec(None);
+        hmac.provider = SourceProvider::Gcs;
+        hmac.endpoint = "https://storage.googleapis.com".to_string();
+        SourceClient::new(&hmac)
+            .await
+            .expect("GCS HMAC uses the always-available S3 backend");
+    }
+
     async fn scripted_client(spec: &SourceClientSpec, responses: Vec<Scripted>) -> (SourceClient, Recorded) {
         let requests: Recorded = Arc::new(Mutex::new(Vec::new()));
         let connector = SharedHttpConnector::new(ScriptedConnector {
@@ -1817,6 +1845,7 @@ mod tests {
             ok(Vec::new(), CONTRACT_TAGGING),
             ok(Vec::new(), ""),
             status(404, ""),
+            ok(Vec::new(), ""),
             status(403, ACCESS_DENIED_BODY),
         ])
         .await;

crates/ecstore/src/bucket/on_demand_migration/sys.rs

@@ -85,6 +85,8 @@ pub static GLOBAL_ON_DEMAND_MIGRATION_SYS: OnceLock<OnDemandMigrationSys> = Once
 /// `resolve` as [`OdmLookup::Unavailable`] and through status snapshots.
 #[derive(Clone, Debug, PartialEq, Eq, thiserror::Error)]
 pub enum OdmStateError {
+    #[error("the {0} backend is not included in this build")]
+    BackendNotCompiled(&'static str),
     /// `source.credentials` is `null`; the shared client builder has no
     /// anonymous mode yet (rustfs/backlog#2149 follow-up).
     #[error("anonymous source access is not supported yet; configure source credentials")]
@@ -310,11 +312,12 @@ impl BucketOdmState {
         write_back: Option<Arc<dyn OdmWriteBack>>,
     ) -> Arc<Self> {
         let spec = source_client_spec(config);
-        let client = if config.source.credentials.is_none() {
+        let client = if config.source.credentials.is_none() && !config.source.provider.is_native() {
             Err(OdmStateError::AnonymousUnsupported)
         } else {
             SourceClient::new(&spec).await.map(Arc::new).map_err(|err| match err {
                 RemoteS3ClientError::MissingCredentials => OdmStateError::AnonymousUnsupported,
+                RemoteS3ClientError::BackendNotCompiled(provider) => OdmStateError::BackendNotCompiled(provider),
                 other => OdmStateError::ClientBuild(other.to_string()),
             })
         };
@@ -746,8 +749,9 @@ impl OnDemandMigrationSys {
     /// (client construction is async). Requires a Tokio runtime for the
     /// install path; without one the config is logged and skipped.
     pub fn publish(&'static self, bucket: &str, config: Option<&OnDemandMigrationConfig>) {
-        let generation = self.next_generation();
-        let Some(config) = self.desired(config) else {
+        let config = self.desired(config);
+        let generation = self.reserve_generation(bucket, config.is_some());
+        let Some(config) = config else {
             self.remove_with_generation(bucket, generation);
             return;
         };
@@ -777,7 +781,8 @@ impl OnDemandMigrationSys {
     /// Installs, rebuilds, or removes the bucket state for `config`.
     /// Idempotent: the same config on an installed bucket is a no-op.
     pub async fn apply(&self, bucket: &str, config: Option<&OnDemandMigrationConfig>) -> ApplyOutcome {
-        let generation = self.next_generation();
+        let config = self.desired(config);
+        let generation = self.reserve_generation(bucket, config.is_some());
         self.apply_with_generation(bucket, config, generation).await
     }
 
@@ -838,7 +843,7 @@ impl OnDemandMigrationSys {
 
     /// Removes a bucket's state (idempotent), cancelling its token.
     pub fn remove(&self, bucket: &str) -> ApplyOutcome {
-        let generation = self.next_generation();
+        let generation = self.reserve_generation(bucket, false);
         self.remove_with_generation(bucket, generation)
     }
 
@@ -879,8 +884,17 @@ impl OnDemandMigrationSys {
         snapshots
     }
 
-    fn next_generation(&self) -> u64 {
-        self.generation.fetch_add(1, Ordering::Relaxed) + 1
+    fn reserve_generation(&self, bucket: &str, installing: bool) -> u64 {
+        // Reserve a desired install before its async client build, under the
+        // same lock that orders removals. Unconfigured buckets need no slot.
+        let mut buckets = self.buckets.write();
+        let generation = self.generation.fetch_add(1, Ordering::Relaxed) + 1;
+        if installing {
+            buckets.entry(bucket.to_string()).or_default().generation = generation;
+        } else if let Some(slot) = buckets.get_mut(bucket) {
+            slot.generation = generation;
+        }
+        generation
     }
 
     fn desired<'c>(&self, config: Option<&'c OnDemandMigrationConfig>) -> Option<&'c OnDemandMigrationConfig> {
@@ -1076,6 +1090,45 @@ mod tests {
         assert_eq!(sys.apply("b", Some(&cfg)).await, ApplyOutcome::Rebuilt);
     }
 
+    #[tokio::test]
+    async fn native_azure_uses_provider_credentials_without_s3_credentials() {
+        let sys = enabled_sys();
+        let mut cfg = config(None);
+        cfg.source.provider = Provider::Azure;
+        cfg.source.endpoint = None;
+        cfg.source.credentials = None;
+        cfg.source.azure = Some(super::super::config::AzureSourceConfig {
+            account: "legacyaccount".to_string(),
+            account_key: Some("c2VjcmV0LWtleQ==".to_string()),
+            sas_token: None,
+        });
+        assert_eq!(sys.apply("b", Some(&cfg)).await, ApplyOutcome::Installed);
+        let state = ready_state(sys.resolve("b", "k"));
+        assert!(state.client().is_ok(), "native credentials must not be classified as anonymous S3");
+    }
+
+    #[cfg(not(feature = "gcs"))]
+    #[tokio::test]
+    async fn gcs_backend_not_compiled_is_unavailable_not_anonymous() {
+        let sys = enabled_sys();
+        let mut cfg = config(None);
+        cfg.source.provider = Provider::GcsNative;
+        cfg.source.credentials = None;
+        cfg.source.gcs = Some(super::super::config::GcsSourceConfig {
+            service_account_json: "{}".to_string(),
+        });
+        let encoded = cfg.to_json().expect("GCS config is serializable without the backend");
+        let restored: OnDemandMigrationConfig = serde_json::from_slice(&encoded).expect("GCS config stays readable");
+        assert_eq!(restored, cfg);
+        assert_eq!(sys.apply("b", Some(&cfg)).await, ApplyOutcome::Installed);
+        match sys.resolve("b", "k") {
+            Some(OdmLookup::Unavailable { error, .. }) => {
+                assert_eq!(error, OdmStateError::BackendNotCompiled("gcs_native"));
+            }
+            other => panic!("expected unavailable backend, got {other:?}"),
+        }
+    }
+
     #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
     async fn singleflight_admits_one_leader_per_key() {
         let sys = enabled_sys();
@@ -1291,21 +1344,31 @@ mod tests {
         assert!(state.is_cancelled());
     }
 
+    #[tokio::test]
+    async fn absent_config_updates_do_not_allocate_bucket_slots() {
+        let sys = enabled_sys();
+        for index in 0..1000 {
+            let bucket = format!("unconfigured-{index}");
+            assert_eq!(sys.apply(&bucket, None).await, ApplyOutcome::NotDesired);
+            assert_eq!(sys.remove(&bucket), ApplyOutcome::NotDesired);
+        }
+        assert!(sys.buckets.read().is_empty(), "unconfigured buckets must not accumulate tombstones");
+    }
+
     #[tokio::test]
     async fn stale_install_cannot_overwrite_a_later_removal() {
         let sys = enabled_sys();
         let cfg = config(None);
-        let older = sys.next_generation();
-        let newer = sys.next_generation();
+        let older = sys.reserve_generation("b", true);
+        let newer = sys.reserve_generation("b", false);
         assert_eq!(sys.remove_with_generation("b", newer), ApplyOutcome::NotDesired);
-        // The removal above did not create a slot; simulate an install that
-        // started before it and finishes after.
-        sys.apply_with_generation("b", Some(&cfg), older).await;
-        assert!(sys.state("b").is_some(), "no slot yet, so the older install lands");
+        assert_eq!(sys.apply_with_generation("b", Some(&cfg), older).await, ApplyOutcome::Superseded);
+        assert!(sys.state("b").is_none(), "removal must supersede an in-flight first install");
 
+        assert_eq!(sys.apply("b", Some(&cfg)).await, ApplyOutcome::Installed);
         let installed = sys.state("b").unwrap();
-        let older = sys.next_generation();
-        let newer = sys.next_generation();
+        let older = sys.reserve_generation("b", true);
+        let newer = sys.reserve_generation("b", false);
         assert_eq!(sys.remove_with_generation("b", newer), ApplyOutcome::Removed);
         assert!(installed.is_cancelled());
         assert_eq!(sys.apply_with_generation("b", Some(&cfg), older).await, ApplyOutcome::Superseded);

crates/ecstore/src/bucket/remote_s3_client.rs

@@ -180,6 +180,8 @@ impl RemoteS3EndpointSpec {
 
 #[derive(Debug, thiserror::Error)]
 pub enum RemoteS3ClientError {
+    #[error("the {0} backend is not included in this build")]
+    BackendNotCompiled(&'static str),
     #[error("remote endpoint requires credentials")]
     MissingCredentials,
     #[error("{0}")]

crates/ecstore/src/services/tier/mod.rs

@@ -25,6 +25,7 @@ pub(crate) mod tier_probe_intent;
 pub mod warm_backend;
 pub mod warm_backend_aliyun;
 pub mod warm_backend_azure;
+#[cfg(feature = "gcs")]
 pub mod warm_backend_gcs;
 pub mod warm_backend_huaweicloud;
 pub mod warm_backend_minio;

crates/ecstore/src/services/tier/warm_backend.rs

@@ -19,13 +19,14 @@
 #![allow(clippy::all)]
 
 use crate::error::is_err_bucket_not_found;
+#[cfg(feature = "gcs")]
+use crate::services::tier::warm_backend_gcs::WarmBackendGCS;
 use crate::services::tier::{
     tier::{ERR_TIER_BACKEND_IN_USE, ERR_TIER_INVALID_CONFIG, ERR_TIER_TYPE_UNSUPPORTED},
     tier_config::{TierConfig, TierType},
     tier_handlers::{ERR_TIER_BUCKET_NOT_FOUND, ERR_TIER_NOT_FOUND, ERR_TIER_PERM_ERR},
     warm_backend_aliyun::WarmBackendAliyun,
     warm_backend_azure::WarmBackendAzure,
-    warm_backend_gcs::WarmBackendGCS,
     warm_backend_huaweicloud::WarmBackendHuaweicloud,
     warm_backend_minio::WarmBackendMinIO,
     warm_backend_r2::WarmBackendR2,
@@ -912,6 +913,15 @@ pub async fn new_warm_backend(tier: &TierConfig, probe: bool) -> Result<WarmBack
                 });
             }
         }
+        #[cfg(not(feature = "gcs"))]
+        TierType::GCS => {
+            return Err(AdminError {
+                code: ERR_TIER_TYPE_UNSUPPORTED.code.clone(),
+                message: "This build does not include the GCS backend; rebuild with the gcs feature".to_string(),
+                status_code: StatusCode::NOT_IMPLEMENTED,
+            });
+        }
+        #[cfg(feature = "gcs")]
         TierType::GCS => {
             if let Some(gcs_config) = tier.gcs.as_ref() {
                 let dd = WarmBackendGCS::new(gcs_config, &tier.name).await;
@@ -1028,6 +1038,27 @@ mod tests {
 
     const PROBE_VERSION: &str = "remote-v2";
 
+    #[cfg(not(feature = "gcs"))]
+    #[tokio::test]
+    async fn gcs_backend_not_compiled_preserves_config() {
+        let json = r#"{"name":"ARCHIVE","type":"gcs","gcs":{"bucket":"archive","creds":"secret"}}"#;
+        let tier: TierConfig = serde_json::from_str(json).expect("GCS config remains readable without the backend");
+        assert_eq!(tier.tier_type, TierType::GCS);
+        let encoded = serde_json::to_vec(&tier).expect("GCS config remains writable");
+        let restored: TierConfig = serde_json::from_slice(&encoded).expect("GCS config round trips");
+        assert_eq!(restored.tier_type, TierType::GCS);
+        let restored_gcs = restored.gcs.as_ref().expect("GCS settings preserved");
+        assert_eq!(restored_gcs.bucket, "archive");
+        assert_eq!(restored_gcs.creds, "secret");
+        assert_eq!(tier.redacted().gcs.expect("redacted GCS settings").creds, "REDACTED");
+        let error = match new_warm_backend(&tier, false).await {
+            Ok(_) => panic!("an excluded GCS backend cannot be constructed"),
+            Err(error) => error,
+        };
+        assert_eq!(error.code, ERR_TIER_TYPE_UNSUPPORTED.code);
+        assert_eq!(error.status_code, StatusCode::NOT_IMPLEMENTED);
+    }
+
     struct CountingBackend {
         put_result: fn() -> Result<String, std::io::Error>,
         removes: Arc<AtomicUsize>,