rc_core/
replication.rs

1//! Bucket replication configuration types
2//!
3//! Domain types for S3 bucket replication configuration and
4//! RustFS admin API remote target management.
5
6use serde::{Deserialize, Serialize};
7use std::fmt;
8use std::time::Duration;
9
10/// Overall or per-target outcome of an active replication check.
11#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
12pub enum ReplicationCheckStatus {
13    #[serde(rename = "OK")]
14    Ok,
15    #[serde(rename = "FAILED")]
16    Failed,
17}
18
19/// Outcome of one phase in an active replication check.
20#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
21pub enum ReplicationCheckPhaseState {
22    #[serde(rename = "OK")]
23    Ok,
24    #[serde(rename = "FAILED")]
25    Failed,
26    #[serde(rename = "SKIPPED")]
27    Skipped,
28}
29
30/// One phase of the server's active target validation state machine.
31#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
32pub struct ReplicationCheckPhase {
33    #[serde(rename = "Status")]
34    pub status: ReplicationCheckPhaseState,
35    #[serde(rename = "Error", default)]
36    pub error: Option<String>,
37}
38
39/// All phases reported for one configured replication target.
40#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
41pub struct ReplicationCheckPhases {
42    #[serde(rename = "Bucket")]
43    pub bucket: ReplicationCheckPhase,
44    #[serde(rename = "Versioning")]
45    pub versioning: ReplicationCheckPhase,
46    #[serde(rename = "ObjectLock")]
47    pub object_lock: ReplicationCheckPhase,
48    #[serde(rename = "Put")]
49    pub put: ReplicationCheckPhase,
50    #[serde(rename = "DeleteMarker")]
51    pub delete_marker: ReplicationCheckPhase,
52    #[serde(rename = "VersionDelete")]
53    pub version_delete: ReplicationCheckPhase,
54    #[serde(rename = "Cleanup")]
55    pub cleanup: ReplicationCheckPhase,
56}
57
58/// Structured outcome for one configured replication target.
59#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
60pub struct ReplicationCheckTarget {
61    #[serde(rename = "Arn")]
62    pub target_arn: String,
63    #[serde(rename = "Bucket")]
64    pub bucket: String,
65    #[serde(rename = "Status")]
66    pub status: ReplicationCheckStatus,
67    #[serde(rename = "Error", default)]
68    pub error: Option<String>,
69    #[serde(rename = "Phases")]
70    pub phases: ReplicationCheckPhases,
71}
72
73/// Result of the active replication-target validation extension.
74#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
75pub struct ReplicationCheckResult {
76    /// Optional explicit contract version for forward-compatible servers.
77    #[serde(rename = "Version", default, skip_serializing_if = "Option::is_none")]
78    pub contract_version: Option<u32>,
79    #[serde(rename = "Status")]
80    pub status: ReplicationCheckStatus,
81    #[serde(rename = "ActiveMutation")]
82    pub active_mutation: bool,
83    #[serde(rename = "MutationDescription")]
84    pub mutation_description: String,
85    #[serde(rename = "ProbeNamespace")]
86    pub probe_namespace: String,
87    #[serde(rename = "Targets")]
88    pub targets: Vec<ReplicationCheckTarget>,
89    /// True only for the successful empty-body contract used by older servers.
90    #[serde(skip)]
91    pub legacy_empty_response: bool,
92}
93
94impl ReplicationCheckResult {
95    /// Represent the legacy server contract that returned no target detail.
96    pub fn legacy_success() -> Self {
97        Self {
98            contract_version: None,
99            status: ReplicationCheckStatus::Ok,
100            active_mutation: true,
101            mutation_description: "Legacy active write/delete replication probe".to_string(),
102            probe_namespace: String::new(),
103            targets: Vec::new(),
104            legacy_empty_response: true,
105        }
106    }
107
108    pub fn succeeded(&self) -> bool {
109        self.status == ReplicationCheckStatus::Ok
110    }
111}
112
113/// Options for starting a RustFS bucket replication resync.
114#[derive(Debug, Clone, PartialEq, Eq, Default)]
115pub struct ReplicationResyncStartOptions {
116    /// Select a configured replication target. The server may infer the only target.
117    pub target_arn: Option<String>,
118    /// Only resync objects older than this duration.
119    pub older_than: Option<Duration>,
120    /// Caller-supplied operation identifier. The server generates one when omitted.
121    pub reset_id: Option<String>,
122}
123
124/// A target accepted by a bucket replication resync start request.
125#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
126pub struct ReplicationResyncStartResult {
127    pub target_arn: String,
128    pub reset_id: String,
129}
130
131/// Server state for a bucket replication resync target.
132#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
133#[serde(rename_all = "snake_case")]
134pub enum ReplicationResyncState {
135    NotStarted,
136    Pending,
137    Ongoing,
138    Completed,
139    Failed,
140    Canceled,
141    Unknown,
142}
143
144impl ReplicationResyncState {
145    pub fn from_server(value: &str) -> Self {
146        match value {
147            "" => Self::NotStarted,
148            "Pending" => Self::Pending,
149            "Ongoing" => Self::Ongoing,
150            "Completed" => Self::Completed,
151            "Failed" => Self::Failed,
152            "Canceled" => Self::Canceled,
153            _ => Self::Unknown,
154        }
155    }
156}
157
158/// Status for one configured replication target.
159#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
160pub struct ReplicationResyncTargetStatus {
161    pub target_arn: String,
162    pub reset_id: String,
163    pub reset_before: Option<jiff::Timestamp>,
164    pub started_at: Option<jiff::Timestamp>,
165    /// RustFS beta.10 calls this `EndTime`, but populates it from last-update time.
166    pub last_updated_at: Option<jiff::Timestamp>,
167    pub state: ReplicationResyncState,
168    /// Exact state string returned by the server, including future values.
169    pub server_state: String,
170    pub replicated_count: u64,
171    pub replicated_size: u64,
172    pub failed_count: u64,
173    pub failed_size: u64,
174    pub current_bucket: Option<String>,
175    pub current_object: Option<String>,
176    /// Optional server detail. RustFS beta.10 does not reliably persist this field.
177    pub error: Option<String>,
178}
179
180/// Current server-side bucket replication resync state.
181#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default)]
182pub struct ReplicationResyncStatus {
183    pub targets: Vec<ReplicationResyncTargetStatus>,
184}
185
186// ==================== S3 Replication Config Types ====================
187
188/// Full replication configuration for a bucket (S3 API)
189#[derive(Debug, Clone, Serialize, Deserialize)]
190pub struct ReplicationConfiguration {
191    /// Role ARN or empty for per-rule destination ARNs
192    #[serde(default)]
193    pub role: String,
194
195    /// Replication rules
196    pub rules: Vec<ReplicationRule>,
197}
198
199/// A single replication rule
200#[derive(Debug, Clone, Serialize, Deserialize)]
201#[serde(rename_all = "camelCase")]
202pub struct ReplicationRule {
203    /// Rule identifier
204    pub id: String,
205
206    /// Rule priority (higher = more important)
207    pub priority: i32,
208
209    /// Whether the rule is enabled or disabled
210    pub status: ReplicationRuleStatus,
211
212    /// Key prefix filter
213    #[serde(skip_serializing_if = "Option::is_none")]
214    pub prefix: Option<String>,
215
216    /// Tag-based filter
217    #[serde(skip_serializing_if = "Option::is_none")]
218    pub tags: Option<std::collections::HashMap<String, String>>,
219
220    /// Destination bucket ARN and optional storage class
221    pub destination: ReplicationDestination,
222
223    /// Whether to replicate delete markers
224    #[serde(skip_serializing_if = "Option::is_none")]
225    pub delete_marker_replication: Option<bool>,
226
227    /// Whether to replicate existing objects
228    #[serde(skip_serializing_if = "Option::is_none")]
229    pub existing_object_replication: Option<bool>,
230
231    /// Whether to replicate version deletes
232    #[serde(skip_serializing_if = "Option::is_none")]
233    pub delete_replication: Option<bool>,
234}
235
236/// Rule status
237#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
238pub enum ReplicationRuleStatus {
239    Enabled,
240    Disabled,
241}
242
243impl fmt::Display for ReplicationRuleStatus {
244    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
245        match self {
246            ReplicationRuleStatus::Enabled => write!(f, "Enabled"),
247            ReplicationRuleStatus::Disabled => write!(f, "Disabled"),
248        }
249    }
250}
251
252impl std::str::FromStr for ReplicationRuleStatus {
253    type Err = String;
254
255    fn from_str(s: &str) -> Result<Self, Self::Err> {
256        match s.to_lowercase().as_str() {
257            "enabled" => Ok(ReplicationRuleStatus::Enabled),
258            "disabled" => Ok(ReplicationRuleStatus::Disabled),
259            _ => Err(format!("Invalid replication rule status: {s}")),
260        }
261    }
262}
263
264/// Replication destination
265#[derive(Debug, Clone, Serialize, Deserialize)]
266#[serde(rename_all = "camelCase")]
267pub struct ReplicationDestination {
268    /// Destination bucket ARN
269    pub bucket_arn: String,
270
271    /// Optional storage class override at destination
272    #[serde(skip_serializing_if = "Option::is_none")]
273    pub storage_class: Option<String>,
274}
275
276// ==================== Admin API Remote Target Types ====================
277
278/// Remote bucket target for replication (matches RustFS admin API JSON format)
279#[derive(Debug, Clone, Serialize, Deserialize, Default)]
280pub struct BucketTarget {
281    #[serde(rename = "sourcebucket", default)]
282    pub source_bucket: String,
283
284    #[serde(default)]
285    pub endpoint: String,
286
287    #[serde(default)]
288    pub credentials: Option<BucketTargetCredentials>,
289
290    #[serde(rename = "targetbucket", default)]
291    pub target_bucket: String,
292
293    #[serde(default)]
294    pub secure: bool,
295
296    #[serde(
297        rename = "skipTlsVerify",
298        default,
299        skip_serializing_if = "Option::is_none"
300    )]
301    pub skip_tls_verify: Option<bool>,
302
303    #[serde(rename = "caCertPem", default, skip_serializing_if = "Option::is_none")]
304    pub ca_cert_pem: Option<String>,
305
306    #[serde(default)]
307    pub path: String,
308
309    #[serde(default)]
310    pub api: String,
311
312    #[serde(default)]
313    pub arn: String,
314
315    #[serde(rename = "type", default)]
316    pub target_type: String,
317
318    #[serde(default)]
319    pub region: String,
320
321    #[serde(alias = "bandwidth", default)]
322    pub bandwidth_limit: i64,
323
324    #[serde(rename = "replicationSync", default)]
325    pub replication_sync: bool,
326
327    #[serde(default)]
328    pub storage_class: String,
329
330    #[serde(rename = "healthCheckDuration", default)]
331    pub health_check_duration: u64,
332
333    #[serde(rename = "disableProxy", default)]
334    pub disable_proxy: bool,
335
336    #[serde(rename = "isOnline", default)]
337    pub online: bool,
338}
339
340/// Credentials for a remote bucket target
341#[derive(Debug, Clone, Serialize, Deserialize, Default)]
342pub struct BucketTargetCredentials {
343    #[serde(rename = "accessKey")]
344    pub access_key: String,
345    #[serde(rename = "secretKey")]
346    pub secret_key: String,
347}
348
349impl fmt::Display for ReplicationRule {
350    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
351        write!(
352            f,
353            "{} (priority={}, {})",
354            self.id, self.priority, self.status
355        )
356    }
357}
358
359#[cfg(test)]
360mod tests {
361    use super::*;
362
363    #[test]
364    fn test_replication_rule_status_display() {
365        assert_eq!(ReplicationRuleStatus::Enabled.to_string(), "Enabled");
366        assert_eq!(ReplicationRuleStatus::Disabled.to_string(), "Disabled");
367    }
368
369    #[test]
370    fn test_replication_rule_status_from_str() {
371        assert_eq!(
372            "enabled".parse::<ReplicationRuleStatus>().unwrap(),
373            ReplicationRuleStatus::Enabled
374        );
375        assert!("invalid".parse::<ReplicationRuleStatus>().is_err());
376    }
377
378    #[test]
379    fn resync_state_preserves_empty_and_future_wire_values() {
380        assert_eq!(
381            ReplicationResyncState::from_server("Ongoing"),
382            ReplicationResyncState::Ongoing
383        );
384        assert_eq!(
385            ReplicationResyncState::from_server(""),
386            ReplicationResyncState::NotStarted
387        );
388        assert_eq!(
389            ReplicationResyncState::from_server("FutureState"),
390            ReplicationResyncState::Unknown
391        );
392    }
393
394    #[test]
395    fn resync_start_options_distinguish_server_and_caller_ids() {
396        let generated = ReplicationResyncStartOptions {
397            target_arn: Some("arn:rustfs:replication::id:dest".to_string()),
398            older_than: Some(Duration::from_secs(3600)),
399            reset_id: None,
400        };
401        let explicit = ReplicationResyncStartOptions {
402            reset_id: Some("reset-1".to_string()),
403            ..generated.clone()
404        };
405
406        assert_eq!(generated.reset_id, None);
407        assert_eq!(explicit.reset_id.as_deref(), Some("reset-1"));
408    }
409
410    #[test]
411    fn test_replication_configuration_serialization() {
412        let config = ReplicationConfiguration {
413            role: "arn:aws:iam::123456789:role/replication".to_string(),
414            rules: vec![ReplicationRule {
415                id: "rule-1".to_string(),
416                priority: 1,
417                status: ReplicationRuleStatus::Enabled,
418                prefix: Some("data/".to_string()),
419                tags: None,
420                destination: ReplicationDestination {
421                    bucket_arn: "arn:aws:s3:::dest-bucket".to_string(),
422                    storage_class: None,
423                },
424                delete_marker_replication: Some(true),
425                existing_object_replication: Some(true),
426                delete_replication: None,
427            }],
428        };
429
430        let json = serde_json::to_string_pretty(&config).unwrap();
431        let decoded: ReplicationConfiguration = serde_json::from_str(&json).unwrap();
432        assert_eq!(decoded.rules.len(), 1);
433        assert_eq!(decoded.rules[0].id, "rule-1");
434        assert_eq!(decoded.rules[0].priority, 1);
435    }
436
437    #[test]
438    fn test_bucket_target_serialization() {
439        let target = BucketTarget {
440            source_bucket: "my-bucket".to_string(),
441            endpoint: "http://remote:9000".to_string(),
442            credentials: Some(BucketTargetCredentials {
443                access_key: "admin".to_string(),
444                secret_key: "secret".to_string(),
445            }),
446            target_bucket: "dest-bucket".to_string(),
447            secure: false,
448            skip_tls_verify: Some(true),
449            target_type: "replication".to_string(),
450            region: "us-east-1".to_string(),
451            replication_sync: true,
452            ..Default::default()
453        };
454
455        let json = serde_json::to_string(&target).unwrap();
456        assert!(json.contains("sourcebucket"));
457        assert!(json.contains("targetbucket"));
458        assert!(json.contains("replicationSync"));
459        assert!(json.contains("skipTlsVerify"));
460
461        let decoded: BucketTarget = serde_json::from_str(&json).unwrap();
462        assert_eq!(decoded.source_bucket, "my-bucket");
463        assert_eq!(decoded.target_bucket, "dest-bucket");
464        assert!(decoded.replication_sync);
465        assert_eq!(decoded.skip_tls_verify, Some(true));
466    }
467
468    #[test]
469    fn test_bucket_target_deserialization_from_backend() {
470        let json = r#"{"sourcebucket":"src","endpoint":"http://host:9000","credentials":{"accessKey":"ak","secretKey":"sk"},"targetbucket":"dst","secure":false,"path":"","api":"","arn":"arn:rustfs:replication::id:dst","type":"replication","region":"","bandwidth":0,"replicationSync":false,"storage_class":"","healthCheckDuration":0,"disableProxy":false,"isOnline":true}"#;
471        let target: BucketTarget = serde_json::from_str(json).unwrap();
472        assert_eq!(target.source_bucket, "src");
473        assert_eq!(target.target_bucket, "dst");
474        assert!(target.online);
475        assert_eq!(target.target_type, "replication");
476    }
477
478    #[test]
479    fn test_bucket_target_serialization_includes_ca_cert_pem_content() {
480        let pem = "-----BEGIN CERTIFICATE-----\nMIIB\n-----END CERTIFICATE-----\n";
481        let target = BucketTarget {
482            source_bucket: "my-bucket".to_string(),
483            endpoint: "remote:9000".to_string(),
484            target_bucket: "dest-bucket".to_string(),
485            secure: true,
486            skip_tls_verify: Some(false),
487            ca_cert_pem: Some(pem.to_string()),
488            target_type: "replication".to_string(),
489            ..Default::default()
490        };
491
492        let json = serde_json::to_string(&target).unwrap();
493        assert!(json.contains("\"skipTlsVerify\":false"));
494        assert!(json.contains("caCertPem"));
495        assert!(!json.contains("ca.pem"));
496    }
497}
rc_core/replication.rs

rc_core/
replication.rs
