rc_core/admin/
replication.rs

1//! Typed contracts for read-only replication inspection.
2
3use std::collections::BTreeMap;
4
5use async_trait::async_trait;
6use jiff::Timestamp;
7use serde::{Deserialize, Serialize};
8use serde_json::Value;
9
10use crate::Result;
11
12/// Maximum encoded size accepted for one replication diff response.
13pub const MAX_REPLICATION_DIFF_RESPONSE_BYTES: usize = 8 * 1024 * 1024;
14/// Maximum encoded size accepted for metrics and MRF responses.
15pub const MAX_REPLICATION_INSPECTION_RESPONSE_BYTES: usize = 8 * 1024 * 1024;
16
17/// Scope of a replication observation. Unknown values are retained for forward compatibility.
18#[derive(Debug, Clone, PartialEq, Eq)]
19pub enum ReplicationMetricScope {
20    Unavailable,
21    NodeLocal,
22    ClusterAggregated,
23    PartialCluster,
24    Unknown(String),
25}
26
27impl ReplicationMetricScope {
28    pub fn as_str(&self) -> &str {
29        match self {
30            Self::Unavailable => "unavailable",
31            Self::NodeLocal => "node_local",
32            Self::ClusterAggregated => "cluster_aggregated",
33            Self::PartialCluster => "partial_cluster",
34            Self::Unknown(value) => value,
35        }
36    }
37}
38
39impl Serialize for ReplicationMetricScope {
40    fn serialize<S: serde::Serializer>(
41        &self,
42        serializer: S,
43    ) -> std::result::Result<S::Ok, S::Error> {
44        serializer.serialize_str(self.as_str())
45    }
46}
47
48impl<'de> Deserialize<'de> for ReplicationMetricScope {
49    fn deserialize<D: serde::Deserializer<'de>>(
50        deserializer: D,
51    ) -> std::result::Result<Self, D::Error> {
52        let value = String::deserialize(deserializer)?;
53        Ok(match value.as_str() {
54            "unavailable" => Self::Unavailable,
55            "node_local" => Self::NodeLocal,
56            "cluster_aggregated" => Self::ClusterAggregated,
57            "partial_cluster" => Self::PartialCluster,
58            _ => Self::Unknown(value),
59        })
60    }
61}
62
63#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
64pub struct ReplicationCountSize {
65    pub count: u64,
66    #[serde(rename = "bytes", alias = "size")]
67    pub size: u64,
68    #[serde(flatten, default)]
69    pub extra: BTreeMap<String, Value>,
70}
71
72#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
73pub struct ReplicationQueueMetric {
74    pub curr: ReplicationCountSize,
75    pub avg: ReplicationCountSize,
76    pub max: ReplicationCountSize,
77    pub last_minute: ReplicationCountSize,
78    #[serde(flatten, default)]
79    pub extra: BTreeMap<String, Value>,
80}
81
82#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
83pub struct ReplicationLatencyMetric {
84    pub avg: f64,
85    pub curr: f64,
86    pub max: f64,
87    #[serde(flatten, default)]
88    pub extra: BTreeMap<String, Value>,
89}
90
91#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
92pub struct ReplicationTransferRate {
93    pub avg: f64,
94    pub curr: f64,
95    pub peak: f64,
96    #[serde(flatten, default)]
97    pub extra: BTreeMap<String, Value>,
98}
99
100#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
101pub struct ReplicationTargetMetric {
102    pub replicated_size: u64,
103    pub replicated_count: u64,
104    pub failed: ReplicationCountSize,
105    #[serde(default)]
106    pub fail_stats: Option<ReplicationCountSize>,
107    pub latency: ReplicationLatencyMetric,
108    pub xfer_rate_lrg: ReplicationTransferRate,
109    pub xfer_rate_sml: ReplicationTransferRate,
110    pub bandwidth_limit_bytes_per_sec: u64,
111    pub current_bandwidth_bytes_per_sec: f64,
112    #[serde(default)]
113    pub latency_scope: Option<ReplicationMetricScope>,
114    #[serde(default)]
115    pub bandwidth_scope: Option<ReplicationMetricScope>,
116    #[serde(flatten, default)]
117    pub extra: BTreeMap<String, Value>,
118}
119
120#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
121pub struct ReplicationMetrics {
122    pub stats: BTreeMap<String, ReplicationTargetMetric>,
123    pub replica_size: u64,
124    pub replica_count: u64,
125    pub replicated_size: u64,
126    pub replicated_count: u64,
127    pub q_stat: ReplicationQueueMetric,
128    #[serde(default)]
129    pub provider_available: Option<bool>,
130    #[serde(default)]
131    pub cluster_complete: Option<bool>,
132    #[serde(default)]
133    pub observed_node_count: Option<u32>,
134    #[serde(default)]
135    pub expected_node_count: Option<u32>,
136    #[serde(default)]
137    pub queue_scope: Option<ReplicationMetricScope>,
138    #[serde(flatten, default)]
139    pub extra: BTreeMap<String, Value>,
140}
141
142#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
143pub struct ReplicationMrfTarget {
144    #[serde(rename = "ARN")]
145    pub arn: String,
146    #[serde(rename = "FailedCount")]
147    pub failed_count: u64,
148    #[serde(rename = "FailedSize")]
149    pub failed_size: u64,
150    #[serde(rename = "ObservationScope")]
151    pub observation_scope: ReplicationMetricScope,
152    #[serde(flatten, default)]
153    pub extra: BTreeMap<String, Value>,
154}
155
156#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
157pub struct ReplicationMrf {
158    #[serde(rename = "Bucket")]
159    pub bucket: String,
160    #[serde(rename = "Targets")]
161    pub targets: Vec<ReplicationMrfTarget>,
162    #[serde(rename = "TotalFailedCount")]
163    pub total_failed_count: u64,
164    #[serde(rename = "TotalFailedSize")]
165    pub total_failed_size: u64,
166    #[serde(rename = "QueuedCount")]
167    pub queued_count: u64,
168    #[serde(rename = "QueuedSize")]
169    pub queued_size: u64,
170    #[serde(rename = "PerObjectEntriesAvailable")]
171    pub per_object_entries_available: bool,
172    #[serde(rename = "RuntimeStatsAvailable")]
173    pub runtime_stats_available: bool,
174    #[serde(rename = "ClusterComplete")]
175    pub cluster_complete: bool,
176    #[serde(rename = "ObservedNodeCount")]
177    pub observed_node_count: u32,
178    #[serde(rename = "ExpectedNodeCount")]
179    pub expected_node_count: u32,
180    #[serde(rename = "DurableBacklogAvailable")]
181    pub durable_backlog_available: bool,
182    #[serde(rename = "DurableCount")]
183    pub durable_count: u64,
184    #[serde(rename = "DurableSize")]
185    pub durable_size: u64,
186    #[serde(rename = "PerTargetDurableEntriesAvailable")]
187    pub per_target_durable_entries_available: bool,
188    #[serde(flatten, default)]
189    pub extra: BTreeMap<String, Value>,
190}
191
192#[async_trait]
193pub trait ReplicationInspectionApi: Send + Sync {
194    async fn replication_metrics(&self, bucket: &str) -> Result<ReplicationMetrics>;
195    async fn replication_mrf(&self, bucket: &str) -> Result<ReplicationMrf>;
196}
197
198/// A bounded, on-demand scan of object versions that have not replicated.
199#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
200pub struct ReplicationDiff {
201    #[serde(rename = "Entries")]
202    pub entries: Vec<ReplicationDiffEntry>,
203    #[serde(rename = "IsTruncated")]
204    pub is_truncated: bool,
205    #[serde(rename = "ScannedVersions")]
206    pub scanned_versions: usize,
207    #[serde(flatten, default)]
208    pub extra: BTreeMap<String, Value>,
209}
210
211/// One pending or failed object version returned by a replication diff.
212#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
213pub struct ReplicationDiffEntry {
214    #[serde(rename = "Object")]
215    pub object: String,
216    #[serde(rename = "VersionID")]
217    pub version_id: Option<String>,
218    #[serde(rename = "Size")]
219    pub size_bytes: u64,
220    #[serde(rename = "IsDeleteMarker")]
221    pub delete_marker: bool,
222    #[serde(rename = "ReplicationStatus")]
223    pub replication_status: String,
224    #[serde(rename = "LastModified")]
225    pub last_modified: Option<Timestamp>,
226    #[serde(flatten, default)]
227    pub extra: BTreeMap<String, Value>,
228}
229
230/// Read-only RustFS replication diff operations.
231#[async_trait]
232pub trait ReplicationDiffApi: Send + Sync {
233    /// Scan a bucket, optionally below a prefix, for pending or failed versions.
234    async fn replication_diff(&self, bucket: &str, prefix: Option<&str>)
235    -> Result<ReplicationDiff>;
236}
237
238#[cfg(test)]
239mod tests {
240    use super::*;
241
242    #[test]
243    fn response_preserves_unknown_fields_and_typed_entries() {
244        let response: ReplicationDiff = serde_json::from_str(
245            r#"{
246                "Entries": [{
247                    "Object": "reports/a.json",
248                    "VersionID": "v1",
249                    "Size": 42,
250                    "IsDeleteMarker": false,
251                    "ReplicationStatus": "FAILED",
252                    "LastModified": "2026-07-21T04:00:00Z",
253                    "TargetDetail": {"attempts": 2}
254                }],
255                "IsTruncated": true,
256                "ScannedVersions": 10000,
257                "ServerRevision": 7
258            }"#,
259        )
260        .expect("typed replication diff");
261
262        assert_eq!(response.entries[0].size_bytes, 42);
263        assert_eq!(response.entries[0].version_id.as_deref(), Some("v1"));
264        assert_eq!(response.entries[0].extra["TargetDetail"]["attempts"], 2);
265        assert_eq!(response.extra["ServerRevision"], 7);
266    }
267
268    #[test]
269    fn response_accepts_delete_marker_without_version_or_timestamp() {
270        let response: ReplicationDiff = serde_json::from_str(
271            r#"{
272                "Entries": [{
273                    "Object": "removed.txt",
274                    "VersionID": null,
275                    "Size": 0,
276                    "IsDeleteMarker": true,
277                    "ReplicationStatus": "PENDING",
278                    "LastModified": null
279                }],
280                "IsTruncated": false,
281                "ScannedVersions": 1
282            }"#,
283        )
284        .expect("delete marker diff");
285
286        assert!(response.entries[0].delete_marker);
287        assert!(response.entries[0].version_id.is_none());
288        assert!(response.entries[0].last_modified.is_none());
289    }
290
291    #[test]
292    fn response_rejects_negative_sizes_and_malformed_timestamps() {
293        for payload in [
294            r#"{"Entries":[{"Object":"a","VersionID":null,"Size":-1,"IsDeleteMarker":false,"ReplicationStatus":"FAILED","LastModified":null}],"IsTruncated":false,"ScannedVersions":1}"#,
295            r#"{"Entries":[{"Object":"a","VersionID":null,"Size":1,"IsDeleteMarker":false,"ReplicationStatus":"FAILED","LastModified":"yesterday"}],"IsTruncated":false,"ScannedVersions":1}"#,
296        ] {
297            assert!(serde_json::from_str::<ReplicationDiff>(payload).is_err());
298        }
299    }
300
301    #[test]
302    fn response_requires_scan_completeness_fields() {
303        let payload = r#"{"Entries":[]}"#;
304        assert!(serde_json::from_str::<ReplicationDiff>(payload).is_err());
305    }
306
307    #[test]
308    fn metrics_distinguish_legacy_metadata_and_preserve_unknown_scope() {
309        let legacy: ReplicationMetrics = serde_json::from_str(r#"{"stats":{},"replica_size":0,"replica_count":0,"replicated_size":0,"replicated_count":0,"q_stat":{"curr":{"count":0,"size":0},"avg":{"count":0,"size":0},"max":{"count":0,"size":0},"last_minute":{"count":0,"size":0}}}"#).expect("legacy metrics");
310        assert_eq!(legacy.provider_available, None);
311        assert_eq!(legacy.cluster_complete, None);
312
313        let current: ReplicationMetrics = serde_json::from_str(r#"{"stats":{},"replica_size":0,"replica_count":0,"replicated_size":0,"replicated_count":0,"q_stat":{"curr":{"count":0,"size":0},"avg":{"count":0,"size":0},"max":{"count":0,"size":0},"last_minute":{"count":0,"size":0}},"provider_available":true,"cluster_complete":false,"observed_node_count":1,"expected_node_count":2,"queue_scope":"future_scope"}"#).expect("current metrics");
314        assert_eq!(current.provider_available, Some(true));
315        assert_eq!(
316            current.queue_scope,
317            Some(ReplicationMetricScope::Unknown("future_scope".into()))
318        );
319    }
320
321    #[test]
322    fn metrics_and_mrf_reject_negative_counters() {
323        let metrics = r#"{"stats":{},"replica_size":-1,"replica_count":0,"replicated_size":0,"replicated_count":0,"q_stat":{"curr":{"count":0,"size":0},"avg":{"count":0,"size":0},"max":{"count":0,"size":0},"last_minute":{"count":0,"size":0}}}"#;
324        assert!(serde_json::from_str::<ReplicationMetrics>(metrics).is_err());
325        let mrf = r#"{"Bucket":"b","Targets":[],"TotalFailedCount":-1,"TotalFailedSize":0,"QueuedCount":0,"QueuedSize":0,"PerObjectEntriesAvailable":false,"RuntimeStatsAvailable":true,"ClusterComplete":false,"ObservedNodeCount":1,"ExpectedNodeCount":2,"DurableBacklogAvailable":false,"DurableCount":0,"DurableSize":0,"PerTargetDurableEntriesAvailable":false}"#;
326        assert!(serde_json::from_str::<ReplicationMrf>(mrf).is_err());
327    }
328}
rc_core/admin/replication.rs

rc_core/admin/
replication.rs
