feat(replication): support temporary target credentials (#6860)

crates/ecstore/src/api/mod.rs

@@ -194,10 +194,9 @@ pub mod bucket {
             BucketReplicationResyncStatus, BucketReplicationStat, BucketReplicationStats, BucketStats,
             DeleteReplicationConfigSnapshot, DeletedObjectReplicationInfo, DurableMrfBacklog, DynReplicationPool, InQueueMetric,
             MrfOpKind, MrfReplicateEntry, MustReplicateOptions, ObjectOpts, OperatorRuleContract,
-            REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION, REMOTE_TARGET_READ_ONLY_HISTORICAL_FIELDS,
+            REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION, REMOTE_TARGET_UNSUPPORTED_FIELDS, REMOTE_TARGET_WRITABLE_FIELDS,
-            REMOTE_TARGET_UNSUPPORTED_FIELDS, REMOTE_TARGET_WRITABLE_FIELDS, REPLICATE_INCOMING_DELETE,
+            REPLICATE_INCOMING_DELETE, REPLICATION_CAPABILITY_CONTRACT_VERSION, REPLICATION_READ_ONLY_HISTORICAL_FIELDS,
-            REPLICATION_CAPABILITY_CONTRACT_VERSION, REPLICATION_READ_ONLY_HISTORICAL_FIELDS, REPLICATION_WRITABLE_FIELDS,
+            REPLICATION_WRITABLE_FIELDS, ReplicateDecision, ReplicateObjectInfo, ReplicationBatchAdmission, ReplicationConfig,
-            ReplicateDecision, ReplicateObjectInfo, ReplicationBatchAdmission, ReplicationConfig,
             ReplicationConfigStructureError, ReplicationConfigurationExt, ReplicationDeleteScheduleInput,
             ReplicationDeleteStateSource, ReplicationHealQueueResult, ReplicationObjectBridge, ReplicationObjectIO,
             ReplicationOperation, ReplicationPoolTrait, ReplicationPriority, ReplicationQueueAdmission, ReplicationScannerBridge,

crates/ecstore/src/bucket/bucket_target_sys.rs

@@ -22,6 +22,7 @@ use crate::bucket::target::{self, BucketTarget, BucketTargets, Credentials};
 use crate::bucket::versioning_sys::BucketVersioningSys;
 use crate::runtime::sources as runtime_sources;
 use aws_credential_types::Credentials as SdkCredentials;
+use aws_credential_types::provider::{ProvideCredentials, error::CredentialsError, future};
 use aws_sdk_s3::config::Region as SdkRegion;
 use aws_sdk_s3::config::SharedHttpClient;
 use aws_sdk_s3::error::ProvideErrorMetadata;
@@ -77,7 +78,7 @@ use std::str::FromStr as _;
 use std::sync::Arc;
 use std::sync::OnceLock;
 use std::sync::Weak;
-use std::time::{Duration, Instant};
+use std::time::{Duration, Instant, SystemTime};
 use time::{OffsetDateTime, format_description::well_known::Rfc3339};
 use tokio::sync::Mutex;
 use tokio::sync::RwLock;
@@ -89,6 +90,71 @@ use uuid::Uuid;
 
 const MAX_CONCURRENT_TARGET_HEALTH_CHECKS: usize = 16;
 const REDACTED_CREDENTIAL: &str = "<redacted>";
+const EXPIRED_REMOTE_TARGET_CREDENTIALS: &str = "remote target credentials have expired";
+
+#[derive(Clone)]
+struct RemoteTargetCredentialsProvider {
+    credentials: SdkCredentials,
+}
+
+impl RemoteTargetCredentialsProvider {
+    fn resolve_at(&self, now: SystemTime) -> aws_credential_types::provider::Result {
+        if self.credentials.expiry().is_some_and(|expiration| expiration <= now) {
+            return Err(CredentialsError::provider_error(std::io::Error::other(EXPIRED_REMOTE_TARGET_CREDENTIALS)));
+        }
+        Ok(self.credentials.clone())
+    }
+}
+
+impl fmt::Debug for RemoteTargetCredentialsProvider {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("RemoteTargetCredentialsProvider")
+            .field("temporary", &self.credentials.session_token().is_some())
+            .field("expiration", &self.credentials.expiry())
+            .finish()
+    }
+}
+
+impl ProvideCredentials for RemoteTargetCredentialsProvider {
+    fn provide_credentials<'a>(&'a self) -> future::ProvideCredentials<'a>
+    where
+        Self: 'a,
+    {
+        future::ProvideCredentials::ready(self.resolve_at(SystemTime::now()))
+    }
+
+    fn fallback_on_interrupt(&self) -> Option<SdkCredentials> {
+        self.resolve_at(SystemTime::now()).ok()
+    }
+}
+
+fn remote_target_sdk_credentials(
+    credentials: &Credentials,
+    account_id: &str,
+    now: SystemTime,
+) -> Result<SdkCredentials, &'static str> {
+    let session_token = credentials.effective_session_token();
+    let expiration = credentials.effective_expiration().map(SystemTime::from);
+    if expiration.is_some() && session_token.is_none() {
+        return Err("remote target credential expiration requires a session token");
+    }
+    if expiration.is_some_and(|expiration| expiration <= now) {
+        return Err(EXPIRED_REMOTE_TARGET_CREDENTIALS);
+    }
+
+    let mut builder = SdkCredentials::builder()
+        .access_key_id(credentials.access_key.clone())
+        .secret_access_key(credentials.secret_key.clone())
+        .account_id(account_id.to_string())
+        .provider_name("bucket_target_sys");
+    if let Some(session_token) = session_token {
+        builder = builder.session_token(session_token.to_string());
+    }
+    if let Some(expiration) = expiration {
+        builder = builder.expiry(expiration);
+    }
+    Ok(builder.build())
+}
 
 pub type HeadObjectSdkError = Box<SdkError<HeadObjectError>>;
 pub type GetObjectSdkError = Box<SdkError<GetObjectError>>;
@@ -845,13 +911,26 @@ impl BucketTargetSys {
         Ok(BucketTargets { targets: new_targets })
     }
 
+    async fn mark_refresh_attempt(&self, arn: &str) {
+        // Rate-limit a failed config fetch as well as a failed client build.
+        // A successful rebuild replaces this timestamp during publication.
+        self.arn_remotes_map
+            .write()
+            .await
+            .entry(arn.to_string())
+            .or_default()
+            .last_refresh = OffsetDateTime::now_utc();
+    }
+
     pub async fn mark_refresh_in_progress(&self, bucket: &str, arn: &str) {
         let mut arn_errs = self.arn_errs_map.write().await;
-        arn_errs.entry(arn.to_string()).or_insert_with(|| ArnErrs {
+        let err = arn_errs.entry(arn.to_string()).or_insert_with(|| ArnErrs {
-            bucket: bucket.to_string(),
-            update_in_progress: true,
             count: 1,
+            bucket: bucket.to_string(),
+            ..Default::default()
         });
+        err.update_in_progress = true;
+        err.bucket = bucket.to_string();
     }
 
     pub async fn mark_refresh_done(&self, bucket: &str, arn: &str) {
@@ -863,15 +942,21 @@ impl BucketTargetSys {
     }
 
     pub async fn is_reloading_target(&self, _bucket: &str, arn: &str) -> bool {
-        let arn_errs = self.arn_errs_map.read().await;
+        self.arn_errs_map
-        arn_errs.get(arn).map(|err| err.update_in_progress).unwrap_or(false)
+            .read()
+            .await
+            .get(arn)
+            .is_some_and(|err| err.update_in_progress)
     }
 
-    pub async fn inc_arn_errs(&self, _bucket: &str, arn: &str) {
+    pub async fn inc_arn_errs(&self, bucket: &str, arn: &str) {
         let mut arn_errs = self.arn_errs_map.write().await;
-        if let Some(err) = arn_errs.get_mut(arn) {
+        let err = arn_errs.entry(arn.to_string()).or_insert_with(|| ArnErrs {
-            err.count += 1;
+            bucket: bucket.to_string(),
-        }
+            ..Default::default()
+        });
+        err.count += 1;
+        err.bucket = bucket.to_string();
     }
 
     pub async fn get_remote_target_client(&self, bucket: &str, arn: &str) -> Option<Arc<TargetClient>> {
@@ -884,15 +969,15 @@ impl BucketTargetSys {
                 .unwrap_or((None, None))
         };
 
-        if let Some(cli) = cli {
+        let credentials_expired = cli
+            .as_ref()
+            .is_some_and(|client| client.credentials_expired_at(jiff::Timestamp::now()));
+        if let Some(cli) = cli
+            && !credentials_expired
+        {
             return Some(cli);
         }
 
-        // TODO(backlog): spawn an async task to proactively reload the replication target
-        if self.is_reloading_target(bucket, arn).await {
-            return None;
-        }
-
         if let Some(last_refresh) = last_refresh {
             let now = OffsetDateTime::now_utc();
             if now - last_refresh < Duration::from_secs(60 * 5) {
@@ -900,16 +985,24 @@ impl BucketTargetSys {
             }
         }
 
+        // The existing per-bucket publication lock is also the reload claim:
+        // try-locking keeps the request path non-blocking, is cancellation-safe,
+        // and prevents a stale reload from publishing after a credential update.
+        let update_mutex = self.target_update_mutex(bucket).await;
+        let Ok(update_guard) = update_mutex.try_lock() else {
+            return None;
+        };
+        self.mark_refresh_attempt(arn).await;
+
         match get_bucket_targets_config(bucket).await {
             Ok(bucket_targets) => {
-                self.mark_refresh_in_progress(bucket, arn).await;
+                self.update_all_targets_locked(bucket, Some(&bucket_targets)).await;
-                self.update_all_targets(bucket, Some(&bucket_targets)).await;
-                self.mark_refresh_done(bucket, arn).await;
             }
             Err(e) => {
                 error!("get bucket targets config error:{}", e);
             }
         };
+        drop(update_guard);
 
         let cli = self
             .arn_remotes_map
@@ -917,8 +1010,10 @@ impl BucketTargetSys {
             .await
             .get(arn)
             .and_then(|target| target.client.clone());
-        if cli.is_some() {
+        if let Some(cli) = cli
-            return cli;
+            && !cli.credentials_expired_at(jiff::Timestamp::now())
+        {
+            return Some(cli);
         }
 
         self.inc_arn_errs(bucket, arn).await;
@@ -948,12 +1043,13 @@ impl BucketTargetSys {
             });
         };
 
-        let creds = SdkCredentials::builder()
+        let creds = remote_target_sdk_credentials(credentials, &target.reset_id, SystemTime::now()).map_err(|error| {
-            .access_key_id(credentials.access_key.clone())
+            BucketTargetError::RemoteTargetConnectionErr {
-            .secret_access_key(credentials.secret_key.clone())
+                bucket: target.target_bucket.clone(),
-            .account_id(target.reset_id.clone())
+                access_key: credentials.access_key.clone(),
-            .provider_name("bucket_target_sys")
+                error: error.to_string(),
-            .build();
+            }
+        })?;
 
         let endpoint = if target.secure {
             format!("https://{}", target.endpoint)
@@ -973,7 +1069,7 @@ impl BucketTargetSys {
 
         let mut config_builder = S3Config::builder()
             .endpoint_url(endpoint.clone())
-            .credentials_provider(SharedCredentialsProvider::new(creds))
+            .credentials_provider(SharedCredentialsProvider::new(RemoteTargetCredentialsProvider { credentials: creds }))
             .region(SdkRegion::new(target.region.clone()))
             .behavior_version(aws_sdk_s3::config::BehaviorVersion::latest());
 
@@ -1047,6 +1143,13 @@ impl BucketTargetSys {
         let update_mutex = self.target_update_mutex(bucket).await;
         let _update_guard = update_mutex.lock().await;
 
+        self.update_all_targets_locked(bucket, targets).await;
+    }
+
+    /// Builds and publishes one bucket snapshot while its update mutex is held.
+    /// Keeping persisted-config reads under the same mutex prevents a stale
+    /// reload from overwriting a concurrent credential rotation.
+    async fn update_all_targets_locked(&self, bucket: &str, targets: Option<&BucketTargets>) {
         let mut clients = Vec::new();
         if let Some(new_targets) = targets {
             for target in &new_targets.targets {
@@ -1078,6 +1181,17 @@ impl BucketTargetSys {
             && !new_targets.is_empty()
         {
             for (target, client) in clients {
+                // Keep a timestamped placeholder for configured targets whose
+                // client cannot be built. Replication records these attempts as
+                // failed, while the placeholder prevents every object from
+                // triggering another metadata reload/client build for five minutes.
+                arn_remotes_map.insert(
+                    target.arn.clone(),
+                    ArnTarget {
+                        client: None,
+                        last_refresh: OffsetDateTime::now_utc(),
+                    },
+                );
                 match client {
                     Ok(client) => {
                         arn_remotes_map.insert(
@@ -1090,11 +1204,6 @@ impl BucketTargetSys {
                         health_map.insert(client.arn.clone(), target_health(&client));
                         self.update_bandwidth_limit(bucket, &target.arn, target.bandwidth_limit);
                     }
-                    // The target stays in `targets_map`, so it keeps showing up in
-                    // `bucket remote ls` while no client exists to replicate through it —
-                    // replication then drops every object for this ARN. Without this the
-                    // rejection (loopback endpoint, bad CA, unparseable URL) left no trace
-                    // anywhere.
                     Err(err) => warn!(
                         bucket = %bucket,
                         arn = %target.arn,
@@ -1962,6 +2071,13 @@ pub struct TargetClient {
 }
 
 impl TargetClient {
+    fn credentials_expired_at(&self, now: jiff::Timestamp) -> bool {
+        self.credentials
+            .as_ref()
+            .and_then(Credentials::effective_expiration)
+            .is_some_and(|expiration| expiration <= now)
+    }
+
     pub fn to_url(&self) -> Url {
         Url::parse(&self.endpoint).unwrap()
     }
@@ -2557,6 +2673,26 @@ mod tests {
         }
     }
 
+    #[derive(Clone, Debug)]
+    struct RecordingAuthConnector {
+        signed_requests: Arc<std::sync::Mutex<Vec<(bool, bool)>>>,
+    }
+
+    impl SmithyHttpConnector for RecordingAuthConnector {
+        fn call(&self, request: HttpRequest) -> HttpConnectorFuture {
+            let has_expected_token = request.headers().get("x-amz-security-token") == Some("temporary-session-token");
+            let has_authorization = request.headers().contains_key("authorization");
+            self.signed_requests
+                .lock()
+                .expect("recorded auth request lock should not be poisoned")
+                .push((has_expected_token, has_authorization));
+            HttpConnectorFuture::ready(Ok(HttpResponse::new(
+                aws_smithy_runtime_api::http::StatusCode::try_from(200_u16).expect("200 should be a valid response status"),
+                SdkBody::empty(),
+            )))
+        }
+    }
+
     fn recording_target_client() -> (TargetClient, Arc<std::sync::Mutex<Vec<String>>>) {
         let request_uris = Arc::new(std::sync::Mutex::new(Vec::new()));
         let connector = SharedHttpConnector::new(RecordingHttpConnector {
@@ -2582,6 +2718,150 @@ mod tests {
         )
     }
 
+    #[test]
+    fn remote_target_sdk_credentials_preserve_temporary_credential_fields() {
+        let now = SystemTime::UNIX_EPOCH + Duration::from_secs(1_000);
+        let expiration = SystemTime::UNIX_EPOCH + Duration::from_secs(2_000);
+        let credentials = Credentials {
+            access_key: "access".to_string(),
+            secret_key: "secret".to_string(),
+            session_token: Some("temporary-session-token".to_string()),
+            expiration: Some(jiff::Timestamp::try_from(expiration).expect("test expiration should convert")),
+        };
+
+        let sdk_credentials =
+            remote_target_sdk_credentials(&credentials, "account", now).expect("unexpired temporary credentials should build");
+
+        assert_eq!(sdk_credentials.session_token(), Some("temporary-session-token"));
+        assert_eq!(sdk_credentials.expiry(), Some(expiration));
+        assert_eq!(sdk_credentials.account_id().map(|id| id.as_str()), Some("account"));
+    }
+
+    #[test]
+    fn remote_target_sdk_credentials_normalize_go_zero_expiration() {
+        let credentials = Credentials {
+            access_key: "access".to_string(),
+            secret_key: "secret".to_string(),
+            session_token: None,
+            expiration: Some("0001-01-01T00:00:00Z".parse().expect("Go zero time should parse")),
+        };
+
+        let sdk_credentials = remote_target_sdk_credentials(&credentials, "", SystemTime::now())
+            .expect("Go zero expiration should remain compatible with static credentials");
+
+        assert!(sdk_credentials.session_token().is_none());
+        assert!(sdk_credentials.expiry().is_none());
+    }
+
+    #[test]
+    fn remote_target_sdk_credentials_reject_invalid_expiration_boundaries() {
+        let expiration = SystemTime::UNIX_EPOCH + Duration::from_secs(2_000);
+        let mut credentials = Credentials {
+            access_key: "access".to_string(),
+            secret_key: "secret".to_string(),
+            session_token: None,
+            expiration: Some(jiff::Timestamp::try_from(expiration).expect("test expiration should convert")),
+        };
+
+        assert_eq!(
+            remote_target_sdk_credentials(&credentials, "", SystemTime::UNIX_EPOCH + Duration::from_secs(1_000))
+                .expect_err("expiration without a session token must fail"),
+            "remote target credential expiration requires a session token"
+        );
+
+        credentials.session_token = Some("temporary-session-token".to_string());
+        assert_eq!(
+            remote_target_sdk_credentials(&credentials, "", expiration)
+                .expect_err("credentials expire at the exact expiration boundary"),
+            EXPIRED_REMOTE_TARGET_CREDENTIALS
+        );
+    }
+
+    #[test]
+    fn remote_target_credentials_provider_fails_closed_after_expiration() {
+        let expiration = SystemTime::UNIX_EPOCH + Duration::from_secs(2_000);
+        let provider = RemoteTargetCredentialsProvider {
+            credentials: SdkCredentials::new(
+                "access",
+                "secret",
+                Some("temporary-session-token".to_string()),
+                Some(expiration),
+                "test",
+            ),
+        };
+
+        assert!(provider.resolve_at(expiration - Duration::from_nanos(1)).is_ok());
+        let err = provider
+            .resolve_at(expiration)
+            .expect_err("expired credentials must not be returned");
+        assert_eq!(err.source().map(ToString::to_string).as_deref(), Some(EXPIRED_REMOTE_TARGET_CREDENTIALS));
+        assert!(!format!("{provider:?}").contains("temporary-session-token"));
+        assert!(!format!("{provider:?}").contains("secret"));
+    }
+
+    #[test]
+    fn target_client_detects_expiration_for_cache_refresh() {
+        let expiration: jiff::Timestamp = "2099-01-01T00:00:00Z".parse().expect("expiration should parse");
+        let (mut client, _) = recording_target_client();
+        client.credentials = Some(Credentials {
+            access_key: "access".to_string(),
+            secret_key: "secret".to_string(),
+            session_token: Some("temporary-session-token".to_string()),
+            expiration: Some(expiration),
+        });
+
+        assert!(!client.credentials_expired_at("2098-12-31T23:59:59Z".parse().expect("pre-expiration timestamp should parse")));
+        assert!(client.credentials_expired_at(expiration));
+
+        client.credentials.as_mut().expect("credentials should exist").expiration =
+            Some("0001-01-01T00:00:00Z".parse().expect("Go zero time should parse"));
+        assert!(!client.credentials_expired_at(jiff::Timestamp::now()));
+    }
+
+    #[tokio::test]
+    async fn temporary_credentials_add_security_token_to_sigv4_requests() {
+        let signed_requests = Arc::new(std::sync::Mutex::new(Vec::new()));
+        let connector = SharedHttpConnector::new(RecordingAuthConnector {
+            signed_requests: Arc::clone(&signed_requests),
+        });
+        let http_client = http_client_fn(move |_settings, _components| connector.clone());
+        let credentials = Credentials {
+            access_key: "access".to_string(),
+            secret_key: "secret".to_string(),
+            session_token: Some("temporary-session-token".to_string()),
+            expiration: Some("2099-01-01T00:00:00Z".parse().expect("future expiration should parse")),
+        };
+        let sdk_credentials = remote_target_sdk_credentials(&credentials, "", SystemTime::now())
+            .expect("unexpired temporary credentials should build");
+        let client = S3Client::from_conf(
+            S3Config::builder()
+                .endpoint_url("https://target.example")
+                .credentials_provider(SharedCredentialsProvider::new(RemoteTargetCredentialsProvider {
+                    credentials: sdk_credentials,
+                }))
+                .region(SdkRegion::new("us-east-1"))
+                .http_client(http_client)
+                .behavior_version(aws_sdk_s3::config::BehaviorVersion::latest())
+                .build(),
+        );
+
+        client
+            .head_bucket()
+            .bucket("target-bucket")
+            .send()
+            .await
+            .expect("recording connector should accept the signed request");
+
+        assert_eq!(
+            signed_requests
+                .lock()
+                .expect("recorded auth request lock should not be poisoned")
+                .as_slice(),
+            &[(true, true)],
+            "SigV4 request must include both authorization and the session-token header"
+        );
+    }
+
     fn spawn_https_server(cert: &rcgen::CertifiedKey<rcgen::KeyPair>, requests: usize) -> (u16, std::thread::JoinHandle<()>) {
         use std::io::{Read, Write};
 
@@ -3513,6 +3793,29 @@ mod tests {
         assert!(mutexes.contains_key("second"));
     }
 
+    #[tokio::test]
+    async fn target_refresh_attempt_updates_retry_timestamp_and_error_count() {
+        let sys = BucketTargetSys::default();
+
+        sys.mark_refresh_attempt("arn:reload").await;
+        let last_refresh = sys.arn_remotes_map.read().await["arn:reload"].last_refresh;
+        assert!(OffsetDateTime::now_utc() - last_refresh < Duration::from_secs(5));
+
+        sys.inc_arn_errs("bucket", "arn:reload").await;
+        sys.inc_arn_errs("bucket", "arn:reload").await;
+        let errors = sys.arn_errs_map.read().await;
+        assert_eq!(errors["arn:reload"].count, 2);
+        assert_eq!(errors["arn:reload"].bucket, "bucket");
+        drop(errors);
+
+        sys.mark_refresh_in_progress("bucket", "arn:reload").await;
+        assert!(sys.is_reloading_target("bucket", "arn:reload").await);
+        sys.mark_refresh_done("bucket", "arn:reload").await;
+        assert!(!sys.is_reloading_target("bucket", "arn:reload").await);
+        sys.mark_refresh_in_progress("bucket", "arn:reload").await;
+        assert!(sys.is_reloading_target("bucket", "arn:reload").await);
+    }
+
     #[tokio::test]
     async fn update_all_targets_publishes_disable_proxy_on_target_client() {
         // The read-proxy selector (replication_proxy::get_proxy_targets) skips
@@ -3551,6 +3854,88 @@ mod tests {
         assert!(opted_out.disable_proxy, "disable_proxy must reach the published TargetClient");
     }
 
+    #[tokio::test]
+    async fn update_all_targets_keeps_failed_client_placeholder() {
+        let sys = BucketTargetSys::default();
+        let target = BucketTarget {
+            arn: "arn:expired".to_string(),
+            endpoint: "192.168.1.10:9000".to_string(),
+            target_bucket: "target-bucket".to_string(),
+            region: "us-east-1".to_string(),
+            credentials: Some(Credentials {
+                access_key: "access".to_string(),
+                secret_key: "secret".to_string(),
+                session_token: Some("temporary-session-token".to_string()),
+                expiration: Some("2000-01-01T00:00:00Z".parse().expect("expired timestamp should parse")),
+            }),
+            ..Default::default()
+        };
+        let targets = BucketTargets { targets: vec![target] };
+
+        sys.update_all_targets("bucket", Some(&targets)).await;
+
+        let remotes = sys.arn_remotes_map.read().await;
+        let placeholder = remotes
+            .get("arn:expired")
+            .expect("configured target should retain a cache entry");
+        assert!(placeholder.client.is_none());
+        assert!(OffsetDateTime::now_utc() - placeholder.last_refresh < Duration::from_secs(5));
+        drop(remotes);
+        assert!(sys.get_remote_target_client("bucket", "arn:expired").await.is_none());
+    }
+
+    #[tokio::test]
+    async fn credential_rotation_atomically_replaces_published_client() {
+        let sys = BucketTargetSys::default();
+        let target = |session_token: &str| BucketTarget {
+            arn: "arn:rotating".to_string(),
+            endpoint: "192.168.1.10:9000".to_string(),
+            target_bucket: "target-bucket".to_string(),
+            region: "us-east-1".to_string(),
+            credentials: Some(Credentials {
+                access_key: "access".to_string(),
+                secret_key: "secret".to_string(),
+                session_token: Some(session_token.to_string()),
+                expiration: None,
+            }),
+            ..Default::default()
+        };
+
+        sys.update_all_targets(
+            "bucket",
+            Some(&BucketTargets {
+                targets: vec![target("old-session-token")],
+            }),
+        )
+        .await;
+        let old_client = sys
+            .get_remote_target_client("bucket", "arn:rotating")
+            .await
+            .expect("initial client should be published");
+
+        sys.update_all_targets(
+            "bucket",
+            Some(&BucketTargets {
+                targets: vec![target("new-session-token")],
+            }),
+        )
+        .await;
+        let new_client = sys
+            .get_remote_target_client("bucket", "arn:rotating")
+            .await
+            .expect("rotated client should be published");
+
+        assert!(!Arc::ptr_eq(&old_client, &new_client));
+        assert_eq!(
+            old_client.credentials.as_ref().and_then(Credentials::effective_session_token),
+            Some("old-session-token")
+        );
+        assert_eq!(
+            new_client.credentials.as_ref().and_then(Credentials::effective_session_token),
+            Some("new-session-token")
+        );
+    }
+
     #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
     async fn target_updates_serialize_client_build_through_publication_per_bucket() {
         let sys = Arc::new(BucketTargetSys::default());

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

@@ -44,14 +44,14 @@ mod replication_versioning_boundary;
 mod runtime_boundary;
 
 pub use replication_config_boundary::{
-    ObjectOpts, OperatorRuleContract, REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION, REMOTE_TARGET_READ_ONLY_HISTORICAL_FIELDS,
+    ObjectOpts, OperatorRuleContract, REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION, REMOTE_TARGET_UNSUPPORTED_FIELDS,
-    REMOTE_TARGET_UNSUPPORTED_FIELDS, REMOTE_TARGET_WRITABLE_FIELDS, REPLICATION_CAPABILITY_CONTRACT_VERSION,
+    REMOTE_TARGET_WRITABLE_FIELDS, REPLICATION_CAPABILITY_CONTRACT_VERSION, REPLICATION_READ_ONLY_HISTORICAL_FIELDS,
-    REPLICATION_READ_ONLY_HISTORICAL_FIELDS, REPLICATION_WRITABLE_FIELDS, ReplicationConfigStructureError,
+    REPLICATION_WRITABLE_FIELDS, ReplicationConfigStructureError, ReplicationConfigurationExt, ReplicationTargetValidationError,
-    ReplicationConfigurationExt, ReplicationTargetValidationError, assign_site_replication_rule_priorities,
+    assign_site_replication_rule_priorities, invalid_replication_config_status_field, is_site_replication_role,
-    invalid_replication_config_status_field, is_site_replication_role, is_site_replication_rule,
+    is_site_replication_rule, merge_incoming_replication_config, merge_user_replication_config,
-    merge_incoming_replication_config, merge_user_replication_config, replication_target_arn_deployment_id,
+    replication_target_arn_deployment_id, replication_target_arns, should_remove_replication_target,
-    replication_target_arns, should_remove_replication_target, site_replication_rule_deployment_id,
+    site_replication_rule_deployment_id, unsupported_replication_config_field, validate_replication_config_structure,
-    unsupported_replication_config_field, validate_replication_config_structure, validate_replication_config_target_arns,
+    validate_replication_config_target_arns,
 };
 pub(crate) use replication_filemeta_boundary::version_purge_statuses_map;
 pub use replication_filemeta_boundary::{

crates/ecstore/src/bucket/replication/replication_config_boundary.rs

@@ -13,12 +13,12 @@
 // limitations under the License.
 
 pub use rustfs_replication::{
-    ObjectOpts, OperatorRuleContract, REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION, REMOTE_TARGET_READ_ONLY_HISTORICAL_FIELDS,
+    ObjectOpts, OperatorRuleContract, REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION, REMOTE_TARGET_UNSUPPORTED_FIELDS,
-    REMOTE_TARGET_UNSUPPORTED_FIELDS, REMOTE_TARGET_WRITABLE_FIELDS, REPLICATION_CAPABILITY_CONTRACT_VERSION,
+    REMOTE_TARGET_WRITABLE_FIELDS, REPLICATION_CAPABILITY_CONTRACT_VERSION, REPLICATION_READ_ONLY_HISTORICAL_FIELDS,
-    REPLICATION_READ_ONLY_HISTORICAL_FIELDS, REPLICATION_WRITABLE_FIELDS, ReplicationConfigStructureError,
+    REPLICATION_WRITABLE_FIELDS, ReplicationConfigStructureError, ReplicationConfigurationExt, ReplicationRuleExt,
-    ReplicationConfigurationExt, ReplicationRuleExt, ReplicationTargetValidationError, assign_site_replication_rule_priorities,
+    ReplicationTargetValidationError, assign_site_replication_rule_priorities, invalid_replication_config_status_field,
-    invalid_replication_config_status_field, is_site_replication_role, is_site_replication_rule,
+    is_site_replication_role, is_site_replication_rule, merge_incoming_replication_config, merge_user_replication_config,
-    merge_incoming_replication_config, merge_user_replication_config, replication_target_arn_deployment_id,
+    replication_target_arn_deployment_id, replication_target_arns, should_remove_replication_target,
-    replication_target_arns, should_remove_replication_target, site_replication_rule_deployment_id,
+    site_replication_rule_deployment_id, unsupported_replication_config_field, validate_replication_config_structure,
-    unsupported_replication_config_field, validate_replication_config_structure, validate_replication_config_target_arns,
+    validate_replication_config_target_arns,
 };

crates/ecstore/src/bucket/replication/replication_object_config.rs

@@ -436,16 +436,21 @@ pub(crate) async fn check_replicate_delete_strict(
     }
 
     for target in decision.targets_map.values_mut() {
-        if let Some(client) = ReplicationTargetStore::remote_target_client(bucket, &target.arn).await {
+        let replicate_sync = ReplicationTargetStore::remote_target_client(bucket, &target.arn)
-            target.synchronous = client.replicate_sync;
+            .await
-        } else {
+            .map(|client| client.replicate_sync);
-            target.replicate = false;
+        apply_target_delivery_mode(target, replicate_sync);
-            target.synchronous = false;
-        }
     }
     Ok(decision)
 }
 
+fn apply_target_delivery_mode(target: &mut ReplicateTargetDecision, replicate_sync: Option<bool>) {
+    // A missing runtime client is a delivery failure, not a rule mismatch.
+    // Preserve admission and fall back to the asynchronous worker, which can
+    // persist FAILED state for the heal/retry path.
+    target.synchronous = replicate_sync.unwrap_or(false);
+}
+
 pub(crate) fn check_replicate_delete_with_snapshot(
     dobj: &ObjectToDelete,
     oi: &ObjectInfo,
@@ -629,6 +634,23 @@ mod tests {
         }));
     }
 
+    #[test]
+    fn missing_target_client_preserves_delete_admission_as_async() {
+        let mut target = ReplicateTargetDecision::new("arn:target".to_string(), true, true);
+
+        apply_target_delivery_mode(&mut target, None);
+
+        assert!(target.replicate, "a runtime client miss must not erase the replication rule decision");
+        assert!(
+            !target.synchronous,
+            "unavailable synchronous targets must fall back to the async retry path"
+        );
+
+        apply_target_delivery_mode(&mut target, Some(true));
+        assert!(target.replicate);
+        assert!(target.synchronous);
+    }
+
     #[test]
     fn must_replicate_options_preserve_request_flag() {
         let user_defined = HashMap::new();

crates/ecstore/src/bucket/replication/replication_resyncer.rs

@@ -102,6 +102,7 @@ const BACKGROUND_WALKDIR_TIMEOUT: TokioDuration = TokioDuration::from_secs(60);
 const ENV_REPL_RESYNC_MAX_JOBS: &str = "RUSTFS_REPL_RESYNC_MAX_JOBS";
 const DEFAULT_REPL_RESYNC_MAX_JOBS: usize = 2;
 const MAX_REPL_RESYNC_MAX_JOBS: usize = 32;
+const TARGET_CLIENT_UNAVAILABLE_ERROR: &str = "replication target client is unavailable";
 use uuid::Uuid;
 
 const EVENT_RESYNC_STATUS_UPDATE_SKIPPED: &str = "replication_resync_status_update_skipped";
@@ -1847,19 +1848,7 @@ pub(crate) async fn replicate_delete_with_outcome<S: ReplicationStorage>(
                 reason = "target_client_missing",
                 "Skipping replication delete because target client is unavailable"
             );
-            send_local_event(EventArgs {
+            rinfos.targets.push(unavailable_delete_target_info(&dobj, &tgt_entry.arn));
-                event_name: EventName::ObjectReplicationNotTracked.to_string(),
-                bucket_name: bucket.clone(),
-                object: ObjectInfo {
-                    bucket: bucket.clone(),
-                    name: dobj.delete_object.object_name.clone(),
-                    version_id,
-                    delete_marker: dobj.delete_object.delete_marker,
-                    ..Default::default()
-                },
-                user_agent: "Internal: [Replication]".to_string(),
-                ..Default::default()
-            });
             continue;
         };
 
@@ -2584,6 +2573,32 @@ async fn replicate_force_delete_to_targets<S: ReplicationStorage>(dobj: &Deleted
     all_succeeded
 }
 
+fn unavailable_delete_target_info(dobj: &DeletedObjectReplicationInfo, arn: &str) -> ReplicatedTargetInfo {
+    let mut rinfo = dobj
+        .delete_object
+        .replication_state
+        .as_ref()
+        .map(|state| state.target_state(arn))
+        .unwrap_or_else(|| ReplicatedTargetInfo {
+            arn: arn.to_string(),
+            ..Default::default()
+        });
+    rinfo.op_type = dobj.op_type;
+    if is_version_delete_replication(&dobj.delete_object) {
+        if rinfo.version_purge_status != VersionPurgeStatusType::Complete {
+            rinfo.version_purge_status = VersionPurgeStatusType::Failed;
+            rinfo.error = Some(TARGET_CLIENT_UNAVAILABLE_ERROR.to_string());
+        }
+    } else if rinfo.prev_replication_status == ReplicationStatusType::Completed && dobj.op_type != ReplicationType::ExistingObject
+    {
+        rinfo.replication_status = ReplicationStatusType::Completed;
+    } else {
+        rinfo.replication_status = ReplicationStatusType::Failed;
+        rinfo.error = Some(TARGET_CLIENT_UNAVAILABLE_ERROR.to_string());
+    }
+    rinfo
+}
+
 async fn replicate_delete_to_target(dobj: &DeletedObjectReplicationInfo, tgt_client: Arc<TargetClient>) -> ReplicatedTargetInfo {
     let version_id = if let Some(version_id) = &dobj.delete_object.delete_marker_version_id {
         version_id.to_owned()
@@ -2796,6 +2811,10 @@ pub(crate) async fn replicate_object_with_outcome<S: ReplicationStorage>(
     };
 
     let mut join_set = JoinSet::new();
+    let mut rinfos = ReplicatedInfos {
+        replication_timestamp: Some(OffsetDateTime::now_utc()),
+        targets: Vec::with_capacity(tgt_arns.len()),
+    };
 
     for arn in tgt_arns {
         let Some(tgt_client) = ReplicationTargetStore::remote_target_client(&bucket, &arn).await else {
@@ -2803,7 +2822,8 @@ pub(crate) async fn replicate_object_with_outcome<S: ReplicationStorage>(
             // stays unreachable would flood the log from the replication hot path. The
             // condition is reported once per pass by the site-replication reconciler and
             // once per rebuild by `update_all_targets`, which is where an operator can act
-            // on it; the per-object event below still records each dropped object.
+            // on it; the FAILED state below preserves retry visibility and the
+            // aggregate result emits the user-visible failure event once.
             debug!(
                 event = EVENT_RESYNC_RUNTIME_SKIPPED,
                 component = LOG_COMPONENT_ECSTORE,
@@ -2812,15 +2832,9 @@ pub(crate) async fn replicate_object_with_outcome<S: ReplicationStorage>(
                 object = %object,
                 arn = %arn,
                 reason = "target_client_missing",
-                "Replication rule has no bucket target for its destination ARN; object not replicated"
+                "Replication target client unavailable"
             );
-            send_local_event(EventArgs {
+            rinfos.targets.push(unavailable_object_target_info(&roi, &arn));
-                event_name: EventName::ObjectReplicationNotTracked.to_string(),
-                bucket_name: bucket.clone(),
-                object: roi.to_object_info(),
-                user_agent: "Internal: [Replication]".to_string(),
-                ..Default::default()
-            });
             continue;
         };
 
@@ -2835,11 +2849,6 @@ pub(crate) async fn replicate_object_with_outcome<S: ReplicationStorage>(
         });
     }
 
-    let mut rinfos = ReplicatedInfos {
-        replication_timestamp: Some(OffsetDateTime::now_utc()),
-        targets: Vec::with_capacity(join_set.len()),
-    };
-
     while let Some(result) = join_set.join_next().await {
         match result {
             Ok(tgt_info) => {
@@ -2945,6 +2954,23 @@ pub(crate) async fn replicate_object_with_outcome<S: ReplicationStorage>(
     (merged_state, state_persisted)
 }
 
+fn unavailable_object_target_info(roi: &ReplicateObjectInfo, arn: &str) -> ReplicatedTargetInfo {
+    ReplicatedTargetInfo {
+        arn: arn.to_string(),
+        size: roi.actual_size,
+        replication_action: if roi.op_type == ReplicationType::Object {
+            ReplicationAction::All
+        } else {
+            ReplicationAction::Metadata
+        },
+        op_type: roi.op_type,
+        replication_status: ReplicationStatusType::Failed,
+        prev_replication_status: roi.target_replication_status(arn),
+        error: Some(TARGET_CLIENT_UNAVAILABLE_ERROR.to_string()),
+        ..Default::default()
+    }
+}
+
 trait ReplicateObjectInfoExt {
     async fn replicate_object<S: ReplicationObjectIO>(
         &self,
@@ -4157,6 +4183,88 @@ async fn replicate_multipart_parts_and_complete<S: ReplicationObjectIO>(
 #[cfg(test)]
 mod tests {
     use super::super::replication_filemeta_boundary::ReplicateTargetDecision;
+
+    #[test]
+    fn unavailable_object_target_is_persisted_as_failed() {
+        let arn = "arn:object-target";
+        let roi = ReplicateObjectInfo {
+            actual_size: 42,
+            op_type: ReplicationType::Object,
+            replication_status_internal: Some(format!("{arn}=PENDING;")),
+            ..Default::default()
+        };
+
+        let target_info = unavailable_object_target_info(&roi, arn);
+        let merged = get_replication_state(
+            &ReplicatedInfos {
+                replication_timestamp: Some(OffsetDateTime::now_utc()),
+                targets: vec![target_info.clone()],
+            },
+            &ReplicationState::default(),
+            None,
+        );
+
+        assert_eq!(target_info.replication_status, ReplicationStatusType::Failed);
+        assert_eq!(target_info.prev_replication_status, ReplicationStatusType::Pending);
+        assert_eq!(target_info.replication_action, ReplicationAction::All);
+        assert_eq!(target_info.error.as_deref(), Some(TARGET_CLIENT_UNAVAILABLE_ERROR));
+        assert_eq!(merged.targets.get(arn), Some(&ReplicationStatusType::Failed));
+    }
+
+    #[test]
+    fn unavailable_delete_target_is_failed_without_overwriting_completed_state() {
+        let arn = "arn:delete-target";
+        let mut previous_state = ReplicationState::default();
+        previous_state.targets.insert(arn.to_string(), ReplicationStatusType::Pending);
+        let mut dobj = DeletedObjectReplicationInfo {
+            delete_object: ReplicationDeletedObject {
+                delete_marker: true,
+                replication_state: Some(previous_state),
+                ..Default::default()
+            },
+            op_type: ReplicationType::Delete,
+            ..Default::default()
+        };
+
+        let failed = unavailable_delete_target_info(&dobj, arn);
+        assert_eq!(failed.replication_status, ReplicationStatusType::Failed);
+        assert_eq!(failed.prev_replication_status, ReplicationStatusType::Pending);
+        assert_eq!(failed.error.as_deref(), Some(TARGET_CLIENT_UNAVAILABLE_ERROR));
+
+        dobj.delete_object
+            .replication_state
+            .as_mut()
+            .expect("previous state should exist")
+            .targets
+            .insert(arn.to_string(), ReplicationStatusType::Completed);
+        let completed = unavailable_delete_target_info(&dobj, arn);
+        assert_eq!(completed.replication_status, ReplicationStatusType::Completed);
+        assert!(completed.error.is_none());
+    }
+
+    #[test]
+    fn unavailable_version_purge_target_is_persisted_as_failed() {
+        let arn = "arn:purge-target";
+        let mut previous_state = ReplicationState::default();
+        previous_state
+            .purge_targets
+            .insert(arn.to_string(), VersionPurgeStatusType::Pending);
+        let dobj = DeletedObjectReplicationInfo {
+            delete_object: ReplicationDeletedObject {
+                version_id: Some(Uuid::new_v4()),
+                replication_state: Some(previous_state),
+                ..Default::default()
+            },
+            op_type: ReplicationType::Delete,
+            ..Default::default()
+        };
+
+        let target_info = unavailable_delete_target_info(&dobj, arn);
+
+        assert_eq!(target_info.version_purge_status, VersionPurgeStatusType::Failed);
+        assert_eq!(target_info.error.as_deref(), Some(TARGET_CLIENT_UNAVAILABLE_ERROR));
+    }
+
     fn resync_target_state(resync_id: &str, status: ResyncStatusType, replicated_count: i64) -> TargetReplicationResyncStatus {
         TargetReplicationResyncStatus {
             resync_id: resync_id.to_string(),

crates/ecstore/src/bucket/target/bucket_target.rs

@@ -25,6 +25,8 @@ use time::OffsetDateTime;
 use url::Url;
 
 const REDACTED_CREDENTIAL: &str = "<redacted>";
+const GO_YEAR_ONE_START_UNIX_SECONDS: i64 = -62_135_596_800;
+const GO_YEAR_TWO_START_UNIX_SECONDS: i64 = -62_104_060_800;
 
 #[derive(Deserialize, Serialize, Default, Clone)]
 pub struct Credentials {
@@ -41,6 +43,26 @@ pub struct Credentials {
 }
 
 impl Credentials {
+    /// Returns the session token used for request signing.
+    ///
+    /// MinIO-compatible payloads may carry an empty token. Treat whitespace-only
+    /// values as absent without rewriting a real token, whose bytes are opaque.
+    pub fn effective_session_token(&self) -> Option<&str> {
+        self.session_token.as_deref().filter(|token| !token.trim().is_empty())
+    }
+
+    /// Returns the credential expiry after normalizing Go's zero `time.Time`.
+    ///
+    /// Go JSON encoders emit year 1 for an unset `time.Time`; persisted MinIO
+    /// target metadata can therefore contain that sentinel even for static
+    /// credentials.
+    pub fn effective_expiration(&self) -> Option<Timestamp> {
+        self.expiration.filter(|expiration| {
+            let unix_seconds = expiration.as_second();
+            !(GO_YEAR_ONE_START_UNIX_SECONDS..GO_YEAR_TWO_START_UNIX_SECONDS).contains(&unix_seconds)
+        })
+    }
+
     pub fn redacted(&self) -> Self {
         Self {
             access_key: self.access_key.clone(),
@@ -355,6 +377,24 @@ mod tests {
     use std::time::Duration;
     use time::OffsetDateTime;
 
+    #[test]
+    fn credential_effective_values_normalize_only_compatibility_sentinels() {
+        let mut credentials = Credentials {
+            access_key: "access".to_string(),
+            secret_key: "secret".to_string(),
+            session_token: Some("  ".to_string()),
+            expiration: Some("0001-01-01T08:00:00+08:00".parse().expect("Go zero time should parse")),
+        };
+
+        assert!(credentials.effective_session_token().is_none());
+        assert!(credentials.effective_expiration().is_none());
+
+        credentials.session_token = Some(" opaque token ".to_string());
+        credentials.expiration = Some("2099-01-01T00:00:00Z".parse().expect("future timestamp should parse"));
+        assert_eq!(credentials.effective_session_token(), Some(" opaque token "));
+        assert_eq!(credentials.effective_expiration(), credentials.expiration);
+    }
+
     #[test]
     fn test_bucket_target_json_deserialize() {
         let json = r#"

crates/replication/src/config.rs

@@ -60,9 +60,7 @@ pub const REPLICATION_READ_ONLY_HISTORICAL_FIELDS: &[&str] = &[
     "Destination.ReplicationTime",
 ];
 
-// v3: temporary-credential fields are advertised as read-only historical
+// v3: remote targets accept temporary credential session tokens and expiry.
-// metadata. They remain decodable for MinIO and persisted-data compatibility,
-// but set-remote-target rejects them until refresh and rotation are supported.
 pub const REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION: u32 = 3;
 
 pub const REMOTE_TARGET_WRITABLE_FIELDS: &[&str] = &[
@@ -70,6 +68,8 @@ pub const REMOTE_TARGET_WRITABLE_FIELDS: &[&str] = &[
     "endpoint",
     "credentials.accessKey",
     "credentials.secretKey",
+    "credentials.sessionToken",
+    "credentials.expiration",
     "targetbucket",
     "secure",
     "path",
@@ -91,8 +91,6 @@ pub const REMOTE_TARGET_WRITABLE_FIELDS: &[&str] = &[
     "disableProxy",
 ];
 
-pub const REMOTE_TARGET_READ_ONLY_HISTORICAL_FIELDS: &[&str] = &["credentials.sessionToken", "credentials.expiration"];
-
 pub const REMOTE_TARGET_UNSUPPORTED_FIELDS: &[&str] = &["edge", "edgeSyncBeforeExpiry"];
 
 #[derive(Debug, Clone, Serialize, Deserialize, Default)]

crates/replication/src/lib.rs

@@ -29,15 +29,14 @@ mod storage_api;
 pub mod tagging;
 
 pub use config::{
-    ObjectOpts, OperatorRuleContract, REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION, REMOTE_TARGET_READ_ONLY_HISTORICAL_FIELDS,
+    ObjectOpts, OperatorRuleContract, REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION, REMOTE_TARGET_UNSUPPORTED_FIELDS,
-    REMOTE_TARGET_UNSUPPORTED_FIELDS, REMOTE_TARGET_WRITABLE_FIELDS, REPLICATION_CAPABILITY_CONTRACT_VERSION,
+    REMOTE_TARGET_WRITABLE_FIELDS, REPLICATION_CAPABILITY_CONTRACT_VERSION, REPLICATION_READ_ONLY_HISTORICAL_FIELDS,
-    REPLICATION_READ_ONLY_HISTORICAL_FIELDS, REPLICATION_WRITABLE_FIELDS, ReplicationConfigStructureError,
+    REPLICATION_WRITABLE_FIELDS, ReplicationConfigStructureError, ReplicationConfigurationExt, ReplicationTargetValidationError,
-    ReplicationConfigurationExt, ReplicationTargetValidationError, active_replication_rule_destination_arns,
+    active_replication_rule_destination_arns, assign_site_replication_rule_priorities, invalid_replication_config_status_field,
-    assign_site_replication_rule_priorities, invalid_replication_config_status_field, is_reconciler_owned_site_replication_rule,
+    is_reconciler_owned_site_replication_rule, is_site_replication_role, is_site_replication_rule,
-    is_site_replication_role, is_site_replication_rule, merge_incoming_replication_config, merge_user_replication_config,
+    merge_incoming_replication_config, merge_user_replication_config, replication_target_arn_deployment_id,
-    replication_target_arn_deployment_id, replication_target_arns, should_remove_replication_target,
+    replication_target_arns, should_remove_replication_target, site_replication_rule_deployment_id,
-    site_replication_rule_deployment_id, unsupported_replication_config_field, validate_replication_config_structure,
+    unsupported_replication_config_field, validate_replication_config_structure, validate_replication_config_target_arns,
-    validate_replication_config_target_arns,
 };
 pub use delete::{
     DeletedObjectReplicationInfo, delete_marker_purge_mrf_entry, delete_marker_purge_version_id,

rustfs/src/admin/handlers/replication.rs

@@ -23,9 +23,9 @@ use crate::admin::storage_api::bucket::metadata::BUCKET_TARGETS_FILE;
 use crate::admin::storage_api::bucket::metadata_sys;
 use crate::admin::storage_api::bucket::metadata_sys::get_replication_config;
 use crate::admin::storage_api::bucket::replication::REMOTE_TARGET_UNSUPPORTED_FIELDS;
-use crate::admin::storage_api::bucket::replication::{BucketStats, ReplicationStatusType};
 #[cfg(test)]
-use crate::admin::storage_api::bucket::replication::{REMOTE_TARGET_READ_ONLY_HISTORICAL_FIELDS, REMOTE_TARGET_WRITABLE_FIELDS};
+use crate::admin::storage_api::bucket::replication::REMOTE_TARGET_WRITABLE_FIELDS;
+use crate::admin::storage_api::bucket::replication::{BucketStats, ReplicationStatusType};
 use crate::admin::storage_api::bucket::target::{
     BucketTarget, BucketTargetType, Credentials as TargetCredentials, LatencyStat, duration_from_secs_or_nanos,
 };
@@ -57,13 +57,6 @@ use url::Host;
 
 const SUPPORTED_REMOTE_TARGET_API: &str = "s3v4";
 
-/// Go encodes the zero `time.Time` as the year-1 instant
-/// (`0001-01-01T00:00:00Z`, possibly re-encoded with an offset); no real
-/// credential expiry lives in year 1, so any such timestamp means "unset".
-fn is_go_zero_time(timestamp: Timestamp) -> bool {
-    timestamp.to_zoned(jiff::tz::TimeZone::UTC).year() == 1
-}
-
 /// Field groups a `set-remote-target?update=true` request may modify, mirroring
 /// MinIO's `TargetUpdateType` / `GetTargetUpdateOps` query contract: the update
 /// overlays only the requested groups onto the stored target, so a client can
@@ -250,7 +243,7 @@ impl RemoteTargetRequest {
 
     fn into_bucket_target(self) -> S3Result<BucketTarget> {
         self.validate_connection_fields()?;
-        self.into_bucket_target_common()
+        self.into_bucket_target_common(true)
     }
 
     /// Partial-update parse: only the field groups named by `ops` are validated;
@@ -259,43 +252,18 @@ impl RemoteTargetRequest {
         if self.arn.trim().is_empty() {
             return Err(s3_error!(InvalidRequest, "arn is required for update"));
         }
-        if ops.contains(&TargetUpdateOp::Credentials) {
+        let replacing_credentials = ops.contains(&TargetUpdateOp::Credentials);
+        if replacing_credentials {
             self.validate_connection_fields()?;
         }
-        self.into_bucket_target_common()
+        self.into_bucket_target_common(replacing_credentials)
     }
 
-    fn into_bucket_target_common(self) -> S3Result<BucketTarget> {
+    fn into_bucket_target_common(self, validate_credentials: bool) -> S3Result<BucketTarget> {
         if !self.target_type.is_valid() {
             return Err(s3_error!(InvalidRequest, "type is invalid"));
         }
 
-        if self
-            .credentials
-            .session_token
-            .as_deref()
-            .is_some_and(|token| !token.trim().is_empty())
-        {
-            return Err(s3_error!(
-                InvalidRequest,
-                "remote target field credentials.session_token is not supported by this RustFS version"
-            ));
-        }
-
-        // Go's `omitempty` never elides a zero `time.Time`, so every madmin
-        // marshal carries `"expiration":"0001-01-01T00:00:00Z"`; only a real
-        // (non-year-1) expiry means the client wants expiring credentials.
-        if self
-            .credentials
-            .expiration
-            .is_some_and(|expiration| !is_go_zero_time(expiration))
-        {
-            return Err(s3_error!(
-                InvalidRequest,
-                "remote target field credentials.expiration is not supported by this RustFS version"
-            ));
-        }
-
         if !self.api.is_empty() && self.api != SUPPORTED_REMOTE_TARGET_API {
             return Err(s3_error!(
                 InvalidRequest,
@@ -317,9 +285,17 @@ impl RemoteTargetRequest {
         }
 
         let mut credentials = TargetCredentials::from(self.credentials);
-        // Past the check above the expiration can only be the zero-value
+        credentials.expiration = credentials.effective_expiration();
-        // sentinel, i.e. "no expiration" — never persist it.
+        if validate_credentials && credentials.expiration.is_some() && credentials.effective_session_token().is_none() {
-        credentials.expiration = None;
+            return Err(s3_error!(InvalidRequest, "credentials.expiration requires credentials.session_token"));
+        }
+        if validate_credentials
+            && credentials
+                .expiration
+                .is_some_and(|expiration| expiration <= Timestamp::now())
+        {
+            return Err(s3_error!(InvalidRequest, "credentials.expiration must be in the future"));
+        }
 
         Ok(BucketTarget {
             source_bucket: self.source_bucket,
@@ -1504,10 +1480,10 @@ impl Operation for ReplicationMrfHandler {
 #[cfg(test)]
 mod tests {
     use super::{
-        REMOTE_TARGET_READ_ONLY_HISTORICAL_FIELDS, REMOTE_TARGET_UNSUPPORTED_FIELDS, REMOTE_TARGET_WRITABLE_FIELDS,
+        REMOTE_TARGET_UNSUPPORTED_FIELDS, REMOTE_TARGET_WRITABLE_FIELDS, RemoteTargetCredentialsRequest, RemoteTargetRequest,
-        RemoteTargetCredentialsRequest, RemoteTargetRequest, ReplicationDiffEntry, SUPPORTED_REMOTE_TARGET_API, TargetUpdateOp,
+        ReplicationDiffEntry, SUPPORTED_REMOTE_TARGET_API, TargetUpdateOp, build_mrf_response, extract_query_params,
-        build_mrf_response, extract_query_params, parse_remote_target_update_ops, render_mrf_backlog, render_replication_diff,
+        parse_remote_target_update_ops, render_mrf_backlog, render_replication_diff, unique_replication_peers,
-        unique_replication_peers, validate_remote_target_tls_settings,
+        validate_remote_target_tls_settings,
     };
     use crate::admin::storage_api::bucket::target::{BucketTarget, Credentials as TargetCredentials, LatencyStat};
     use crate::admin::storage_api::replication::{BucketStats, DurableMrfBacklog, MrfOpKind, MrfReplicateEntry};
@@ -2104,27 +2080,13 @@ mod tests {
 
     #[test]
     fn remote_target_request_rejects_unimplemented_fields() {
-        for (field, value, historical_field) in [
+        for (field, value) in [
-            (
+            ("api", serde_json::json!("s3v2")),
-                "credentials.session_token",
+            ("edge", serde_json::json!(true)),
-                serde_json::json!("session-token"),
+            ("edgeSyncBeforeExpiry", serde_json::json!(true)),
-                Some("credentials.sessionToken"),
-            ),
-            (
-                "credentials.expiration",
-                serde_json::json!("2026-01-01T00:00:00Z"),
-                Some("credentials.expiration"),
-            ),
-            ("api", serde_json::json!("s3v2"), None),
-            ("edge", serde_json::json!(true), None),
-            ("edgeSyncBeforeExpiry", serde_json::json!(true), None),
         ] {
             let mut request = valid_remote_target_request();
-            if let Some((credential_field, credential_name)) = field.split_once('.') {
+            request[field] = value;
-                request[credential_field][credential_name] = value;
-            } else {
-                request[field] = value;
-            }
             let request: RemoteTargetRequest =
                 serde_json::from_value(request).expect("unsupported field should still deserialize");
             let err = request
@@ -2133,15 +2095,86 @@ mod tests {
 
             assert!(err.to_string().contains(field));
             assert!(err.to_string().contains("not supported by this RustFS version"));
-            if let Some(historical_field) = historical_field {
-                assert!(
-                    REMOTE_TARGET_READ_ONLY_HISTORICAL_FIELDS.contains(&historical_field),
-                    "rejected field {field} must be advertised as historical-only"
-                );
-            }
         }
     }
 
+    #[test]
+    fn remote_target_request_accepts_temporary_credentials() {
+        let mut request = valid_remote_target_request();
+        request["credentials"]["sessionToken"] = serde_json::json!("session-token");
+        request["credentials"]["expiration"] = serde_json::json!("2099-01-01T00:00:00Z");
+
+        let target = serde_json::from_value::<RemoteTargetRequest>(request)
+            .expect("temporary credentials should deserialize")
+            .into_bucket_target()
+            .expect("unexpired temporary credentials should be accepted");
+        let credentials = target.credentials.expect("credentials should be preserved");
+
+        assert_eq!(credentials.session_token.as_deref(), Some("session-token"));
+        assert_eq!(
+            serde_json::to_value(credentials.expiration.expect("expiration should be preserved"))
+                .expect("expiration should serialize"),
+            serde_json::json!("2099-01-01T00:00:00Z")
+        );
+        for field in ["credentials.sessionToken", "credentials.expiration"] {
+            assert!(REMOTE_TARGET_WRITABLE_FIELDS.contains(&field));
+        }
+    }
+
+    #[test]
+    fn remote_target_request_accepts_session_token_without_expiration() {
+        let mut request = valid_remote_target_request();
+        request["credentials"]["session_token"] = serde_json::json!("session-token");
+
+        let target = serde_json::from_value::<RemoteTargetRequest>(request)
+            .expect("temporary credentials should deserialize")
+            .into_bucket_target()
+            .expect("a session token without a reported expiration should remain compatible");
+
+        assert_eq!(
+            target
+                .credentials
+                .as_ref()
+                .and_then(|credentials| credentials.session_token.as_deref()),
+            Some("session-token")
+        );
+        assert!(target.credentials.and_then(|credentials| credentials.expiration).is_none());
+    }
+
+    #[test]
+    fn remote_target_request_rejects_expiration_without_session_token() {
+        let mut request = valid_remote_target_request();
+        request["credentials"]["expiration"] = serde_json::json!("2099-01-01T00:00:00Z");
+
+        let err = serde_json::from_value::<RemoteTargetRequest>(request)
+            .expect("request should deserialize")
+            .into_bucket_target()
+            .expect_err("an expiring credential bundle requires a session token");
+
+        assert!(
+            err.to_string()
+                .contains("credentials.expiration requires credentials.session_token")
+        );
+    }
+
+    #[test]
+    fn remote_target_request_rejects_expired_temporary_credentials_without_leaking_secrets() {
+        let mut request = valid_remote_target_request();
+        request["credentials"]["secretKey"] = serde_json::json!("secret-must-not-leak");
+        request["credentials"]["sessionToken"] = serde_json::json!("session-token-must-not-leak");
+        request["credentials"]["expiration"] = serde_json::json!("2000-01-01T00:00:00Z");
+
+        let err = serde_json::from_value::<RemoteTargetRequest>(request)
+            .expect("request should deserialize")
+            .into_bucket_target()
+            .expect_err("expired credentials must fail before persistence");
+        let message = err.to_string();
+
+        assert!(message.contains("credentials.expiration must be in the future"));
+        assert!(!message.contains("secret-must-not-leak"));
+        assert!(!message.contains("session-token-must-not-leak"));
+    }
+
     #[test]
     fn remote_target_request_accepts_real_madmin_add_marshal() {
         let request: RemoteTargetRequest =
@@ -2192,6 +2225,32 @@ mod tests {
         );
     }
 
+    #[test]
+    fn non_credential_update_accepts_redacted_temporary_credential_round_trip() {
+        // list-remote-targets redacts both the secret and session token but
+        // intentionally retains the non-secret expiration. mc echoes that
+        // shape on a sync-only update; the credentials group is not applied.
+        let body = serde_json::json!({
+            "endpoint": "192.168.1.10:9000",
+            "credentials": {
+                "accessKey": "access",
+                "expiration": "2099-01-01T00:00:00Z"
+            },
+            "targetbucket": "target",
+            "arn": "arn:rustfs:replication:us-east-1:dep:target",
+            "type": "replication",
+            "replicationSync": true
+        });
+
+        let target = serde_json::from_value::<RemoteTargetRequest>(body)
+            .expect("redacted mc round-trip should deserialize")
+            .into_update_bucket_target(&[TargetUpdateOp::Sync])
+            .expect("a sync-only update must ignore the redacted credential group");
+
+        assert!(target.replication_sync);
+        assert!(target.credentials.and_then(|credentials| credentials.expiration).is_some());
+    }
+
     #[test]
     fn remote_target_request_ignores_client_supplied_latency() {
         // Latency is a server-measured runtime stat; mc echoes the
@@ -2238,22 +2297,6 @@ mod tests {
         assert!(!target.ca_cert_pem.is_empty());
     }
 
-    #[test]
-    fn remote_target_request_validation_does_not_echo_credential_values() {
-        let mut request = valid_remote_target_request();
-        request["credentials"]["session_token"] = serde_json::json!("session-token-must-not-leak");
-
-        let request: RemoteTargetRequest = serde_json::from_value(request).expect("request should deserialize");
-        let err = request
-            .into_bucket_target()
-            .expect_err("session tokens must be rejected before persistence");
-        let message = err.to_string();
-
-        assert!(message.contains("credentials.session_token"));
-        assert!(!message.contains("session-token-must-not-leak"));
-        assert!(!message.contains("secret"));
-    }
-
     #[test]
     fn remote_target_request_accepts_go_duration_wire_values() {
         // `mc replicate add` defaults `--healthcheck-seconds` to 60; madmin
@@ -2514,17 +2557,6 @@ mod tests {
 
     #[test]
     fn remote_target_capability_fields_do_not_overlap() {
-        for field in REMOTE_TARGET_READ_ONLY_HISTORICAL_FIELDS {
-            assert!(
-                !REMOTE_TARGET_WRITABLE_FIELDS.contains(field),
-                "remote target field {field} cannot be both historical-only and writable"
-            );
-            assert!(
-                !REMOTE_TARGET_UNSUPPORTED_FIELDS.contains(field),
-                "remote target field {field} cannot be both historical-only and unsupported"
-            );
-        }
-
         for field in REMOTE_TARGET_UNSUPPORTED_FIELDS {
             assert!(
                 !REMOTE_TARGET_WRITABLE_FIELDS.contains(field),

rustfs/src/admin/handlers/system.rs

@@ -24,9 +24,8 @@ use crate::admin::runtime_sources::{
     DefaultAdminUsecase, QueryServerInfoRequest, current_endpoints_handle, default_admin_usecase, object_store_from_req,
 };
 use crate::admin::storage_api::bucket::replication::{
-    REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION, REMOTE_TARGET_READ_ONLY_HISTORICAL_FIELDS, REMOTE_TARGET_UNSUPPORTED_FIELDS,
+    REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION, REMOTE_TARGET_UNSUPPORTED_FIELDS, REMOTE_TARGET_WRITABLE_FIELDS,
-    REMOTE_TARGET_WRITABLE_FIELDS, REPLICATION_CAPABILITY_CONTRACT_VERSION, REPLICATION_READ_ONLY_HISTORICAL_FIELDS,
+    REPLICATION_CAPABILITY_CONTRACT_VERSION, REPLICATION_READ_ONLY_HISTORICAL_FIELDS, REPLICATION_WRITABLE_FIELDS,
-    REPLICATION_WRITABLE_FIELDS,
 };
 use crate::admin::storage_api::cluster::{
     CapabilityState, CapabilityStatus, ObservabilitySnapshotProvider, TopologySnapshot, TopologySnapshotProvider,
@@ -730,15 +729,6 @@ impl ReplicationCapabilities {
                         name,
                         state: ReplicationFieldState::Supported,
                     })
-                    .chain(
-                        REMOTE_TARGET_READ_ONLY_HISTORICAL_FIELDS
-                            .iter()
-                            .copied()
-                            .map(|name| ReplicationFieldCapability {
-                                name,
-                                state: ReplicationFieldState::ReadOnlyHistorical,
-                            }),
-                    )
                     .chain(
                         REMOTE_TARGET_UNSUPPORTED_FIELDS
                             .iter()
@@ -1306,7 +1296,7 @@ mod tests {
         assert_eq!(response.summary.manual_transition_jobs.state, CapabilityState::Supported);
         assert_eq!(response.replication.contract_version, 1);
         assert_eq!(response.replication.bucket_replication.contract_version, 1);
-        // v3: temporary-credential fields are explicitly historical-only.
+        // v3: temporary-credential fields are writable and used for signing.
         assert_eq!(response.replication.remote_targets.contract_version, 3);
         assert_eq!(response.replication.bucket_replication.status.state, CapabilityState::Supported);
         assert_eq!(response.replication.remote_targets.status.state, CapabilityState::Supported);
@@ -1363,8 +1353,8 @@ mod tests {
                     .remote_targets
                     .fields
                     .iter()
-                    .any(|field| field.name == name && field.state == super::ReplicationFieldState::ReadOnlyHistorical),
+                    .any(|field| field.name == name && field.state == super::ReplicationFieldState::Supported),
-                "remote target field {name} must be advertised as historical-only"
+                "remote target field {name} must be advertised as writable"
             );
         }
         assert_eq!(response.manual_transition_jobs.contract_version, 1);
@@ -1469,8 +1459,8 @@ mod tests {
                     .as_array()
                     .expect("remote target fields should be an array")
                     .iter()
-                    .any(|field| field["name"] == name && field["state"] == "read_only_historical"),
+                    .any(|field| field["name"] == name && field["state"] == "supported"),
-                "serialized remote target field {name} must be historical-only"
+                "serialized remote target field {name} must be writable"
             );
         }
         assert_eq!(value["manual_transition_jobs"]["contract_version"], 1);

rustfs/src/admin/storage_api.rs

@@ -443,10 +443,10 @@ pub(crate) mod quota {
 
 pub(crate) mod replication {
     pub(crate) use super::ecstore_bucket::replication::{
-        OperatorRuleContract, REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION, REMOTE_TARGET_READ_ONLY_HISTORICAL_FIELDS,
+        OperatorRuleContract, REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION, REMOTE_TARGET_UNSUPPORTED_FIELDS,
-        REMOTE_TARGET_UNSUPPORTED_FIELDS, REMOTE_TARGET_WRITABLE_FIELDS, REPLICATION_CAPABILITY_CONTRACT_VERSION,
+        REMOTE_TARGET_WRITABLE_FIELDS, REPLICATION_CAPABILITY_CONTRACT_VERSION, REPLICATION_READ_ONLY_HISTORICAL_FIELDS,
-        REPLICATION_READ_ONLY_HISTORICAL_FIELDS, REPLICATION_WRITABLE_FIELDS, assign_site_replication_rule_priorities,
+        REPLICATION_WRITABLE_FIELDS, assign_site_replication_rule_priorities, merge_incoming_replication_config,
-        merge_incoming_replication_config, replication_target_arn_deployment_id,
+        replication_target_arn_deployment_id,
     };
     pub(crate) type BucketReplicationResyncStatus = super::ecstore_bucket::replication::BucketReplicationResyncStatus;
     pub(crate) type BucketStats = super::ecstore_bucket::replication::BucketStats;