feat(admin): add replication diff/mrf and batch job admin endpoints (#4332)

Add MinIO-compatible admin endpoints under the /v3 prefix, wired to real
RustFS subsystems where they exist and documented as compatibility-only
where they do not.

Replication (backlog#611):
- POST /v3/replication/diff: bounded on-demand scan of object versions,
  returning versions whose replication status is PENDING or FAILED. Wired
  to list_object_versions and the per-version replication status. Requires
  the bucket to exist and have a replication configuration. The scan is
  capped (REPLICATION_DIFF_MAX_SCAN) and reports IsTruncated when partial.
- GET /v3/replication/mrf: reports the failed-replication backlog (MinIO's
  MRF concept) from the replication stats handle: aggregate failed count and
  size per target plus in-queue counters. RustFS records failed replications
  as aggregate counters and persists MRF entries without a runtime query API,
  so per-object MRF rows are not enumerable; PerObjectEntriesAvailable is
  always false to make that explicit.

Batch jobs (backlog#613):
- POST /v3/start-job, GET /v3/list-jobs, GET /v3/status-job,
  GET /v3/describe-job, DELETE /v3/cancel-job.
- RustFS has no batch-job execution engine (no scheduler, queue, or worker),
  so these implement MinIO request/response/error semantics only and never
  fake execution or success: start-job parses and validates the job
  definition then returns NotImplemented for known job types and
  InvalidRequest for unknown ones; list-jobs returns an empty list;
  status/describe/cancel validate jobId and return a no-such-job error.

All routes are registered with admin auth using existing AdminAction
variants (ReplicationDiff, GetReplicationMetricsAction, StartBatchJobAction,
ListBatchJobsAction, DescribeBatchJobAction, CancelBatchJobAction) and are
covered by the admin route-registration matrix and new handler unit tests.

Refs rustfs/backlog#611 #613

rustfs/src/admin/handlers/batch_job.rs

@@ -0,0 +1,336 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! MinIO-compatible batch-job admin endpoints.
+//!
+//! RustFS does not (yet) ship a batch-job execution engine: there is no job
+//! scheduler, queue, or background worker that can run `replicate`, `keyrotate`,
+//! or `expire` batch jobs. These handlers therefore implement the MinIO request,
+//! response, and error *semantics* only, so that MinIO-compatible admin clients
+//! (mc admin batch, madmin) receive deliberate, stable answers instead of a
+//! routing 404 or a silently faked success:
+//!
+//! * `POST /v3/start-job` parses the job definition, validates the declared job
+//!   type, and returns a deliberate `NotImplemented` compatibility error because
+//!   RustFS has no backend that can execute the job. Unknown job types are
+//!   rejected with `InvalidRequest`. No job is ever accepted or persisted, so no
+//!   execution or success is ever faked.
+//! * `GET /v3/list-jobs` returns an empty job list — no batch job can be running
+//!   because none can be started.
+//! * `GET /v3/status-job`, `GET /v3/describe-job`, and `DELETE /v3/cancel-job`
+//!   validate their `jobId` query parameter and return a `NoSuchJob`-style error,
+//!   since no job with that id can exist.
+//!
+//! When RustFS grows a real batch-job engine, these handlers should be rewired to
+//! it; the request parsing and response shapes here are intended to stay stable.
+
+use crate::admin::auth::validate_admin_request;
+use crate::admin::router::{AdminOperation, Operation, S3Router};
+use crate::admin::utils::read_compatible_admin_body;
+use crate::auth::{check_key_valid, get_session_token};
+use crate::server::{ADMIN_PREFIX, RemoteAddr};
+use http::{HeaderMap, HeaderValue, Uri};
+use hyper::{Method, StatusCode};
+use matchit::Params;
+use rustfs_config::MAX_ADMIN_REQUEST_BODY_SIZE;
+use rustfs_credentials::Credentials;
+use rustfs_policy::policy::action::{Action, AdminAction};
+use s3s::header::CONTENT_TYPE;
+use s3s::{Body, S3Error, S3ErrorCode, S3Request, S3Response, S3Result, s3_error};
+use serde::Serialize;
+use std::collections::HashMap;
+use tracing::warn;
+
+/// Job types recognised by the MinIO batch-job admin API.
+///
+/// RustFS cannot execute any of these today; the enum exists so that
+/// `start-job` can distinguish "known job type, unsupported backend"
+/// (`NotImplemented`) from "unknown job type" (`InvalidRequest`).
+const KNOWN_JOB_TYPES: &[&str] = &["replicate", "keyrotate", "expire"];
+
+fn extract_query_params(uri: &Uri) -> HashMap<String, String> {
+    let mut params = HashMap::new();
+    if let Some(query) = uri.query() {
+        for (key, value) in url::form_urlencoded::parse(query.as_bytes()) {
+            params.insert(key.into_owned(), value.into_owned());
+        }
+    }
+    params
+}
+
+async fn validate_batch_job_admin_request(req: &S3Request<Body>, action: AdminAction) -> S3Result<Credentials> {
+    let Some(input_cred) = req.credentials.as_ref() else {
+        return Err(s3_error!(InvalidRequest, "get cred failed"));
+    };
+
+    let (cred, owner) =
+        check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
+
+    let remote_addr = req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0));
+    validate_admin_request(&req.headers, &cred, owner, false, vec![Action::AdminAction(action)], remote_addr).await?;
+
+    Ok(cred)
+}
+
+fn json_response<T: Serialize>(status: StatusCode, value: &T) -> S3Result<S3Response<(StatusCode, Body)>> {
+    let data = serde_json::to_vec(value).map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("{e}")))?;
+    let mut headers = HeaderMap::new();
+    headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));
+    Ok(S3Response::with_headers((status, Body::from(data)), headers))
+}
+
+/// Best-effort detection of the declared job type from a MinIO batch-job
+/// definition body.
+///
+/// MinIO batch jobs are YAML documents whose top-level key names the job type,
+/// e.g. `replicate:`, `keyrotate:`, or `expire:`. RustFS has no YAML dependency
+/// and rejects every job anyway, so instead of pulling in a full parser we look
+/// for the first recognised top-level job-type key. Returns `Some(job_type)`
+/// when a known key is found, `None` otherwise.
+fn detect_job_type(body: &[u8]) -> Option<String> {
+    let text = std::str::from_utf8(body).ok()?;
+    for raw_line in text.lines() {
+        let line = raw_line.trim();
+        if line.is_empty() || line.starts_with('#') {
+            continue;
+        }
+        // A top-level mapping key looks like `replicate:` with no leading indentation.
+        if raw_line.starts_with(char::is_whitespace) {
+            continue;
+        }
+        let Some((key, _)) = line.split_once(':') else {
+            continue;
+        };
+        let key = key.trim().trim_matches('"').to_ascii_lowercase();
+        if KNOWN_JOB_TYPES.contains(&key.as_str()) {
+            return Some(key);
+        }
+    }
+    None
+}
+
+#[derive(Debug, Serialize)]
+struct ListBatchJobsResult {
+    jobs: Vec<serde_json::Value>,
+}
+
+pub fn register_batch_job_route(r: &mut S3Router<AdminOperation>) -> std::io::Result<()> {
+    r.insert(
+        Method::POST,
+        format!("{}{}", ADMIN_PREFIX, "/v3/start-job").as_str(),
+        AdminOperation(&StartBatchJobHandler {}),
+    )?;
+
+    r.insert(
+        Method::GET,
+        format!("{}{}", ADMIN_PREFIX, "/v3/list-jobs").as_str(),
+        AdminOperation(&ListBatchJobsHandler {}),
+    )?;
+
+    r.insert(
+        Method::GET,
+        format!("{}{}", ADMIN_PREFIX, "/v3/status-job").as_str(),
+        AdminOperation(&StatusBatchJobHandler {}),
+    )?;
+
+    r.insert(
+        Method::GET,
+        format!("{}{}", ADMIN_PREFIX, "/v3/describe-job").as_str(),
+        AdminOperation(&DescribeBatchJobHandler {}),
+    )?;
+
+    r.insert(
+        Method::DELETE,
+        format!("{}{}", ADMIN_PREFIX, "/v3/cancel-job").as_str(),
+        AdminOperation(&CancelBatchJobHandler {}),
+    )?;
+
+    Ok(())
+}
+
+/// `POST /v3/start-job`
+///
+/// Parses and validates the submitted job definition, then returns a deliberate
+/// `NotImplemented` compatibility error: RustFS has no batch-job engine to run
+/// the job. Unknown job types are rejected with `InvalidRequest`. No job is
+/// accepted, persisted, or reported as started.
+pub struct StartBatchJobHandler {}
+
+#[async_trait::async_trait]
+impl Operation for StartBatchJobHandler {
+    async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+        let cred = validate_batch_job_admin_request(&req, AdminAction::StartBatchJobAction).await?;
+
+        let body = read_compatible_admin_body(req.input, MAX_ADMIN_REQUEST_BODY_SIZE, req.uri.path(), &cred.secret_key).await?;
+
+        if body.trim_ascii().is_empty() {
+            return Err(s3_error!(InvalidRequest, "batch job definition is required"));
+        }
+
+        match detect_job_type(&body) {
+            Some(job_type) => {
+                warn!(
+                    job_type = %job_type,
+                    "batch job start requested but RustFS has no batch-job execution backend"
+                );
+                // Known job type, but no backend can run it: honest compatibility error.
+                Err(S3Error::with_message(
+                    S3ErrorCode::NotImplemented,
+                    format!("batch job type '{job_type}' is not supported: RustFS has no batch-job execution backend"),
+                ))
+            }
+            None => Err(s3_error!(
+                InvalidRequest,
+                "unable to determine batch job type from definition; expected one of: replicate, keyrotate, expire"
+            )),
+        }
+    }
+}
+
+/// `GET /v3/list-jobs`
+///
+/// No batch job can be running (none can be started), so this always returns an
+/// empty job list. The optional `jobType` filter is validated for compatibility.
+pub struct ListBatchJobsHandler {}
+
+#[async_trait::async_trait]
+impl Operation for ListBatchJobsHandler {
+    async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+        validate_batch_job_admin_request(&req, AdminAction::ListBatchJobsAction).await?;
+
+        let queries = extract_query_params(&req.uri);
+        if let Some(job_type) = queries.get("jobType").filter(|v| !v.is_empty()) {
+            let job_type = job_type.to_ascii_lowercase();
+            if !KNOWN_JOB_TYPES.contains(&job_type.as_str()) {
+                return Err(s3_error!(InvalidRequest, "invalid jobType filter: {job_type}"));
+            }
+        }
+
+        json_response(StatusCode::OK, &ListBatchJobsResult { jobs: Vec::new() })
+    }
+}
+
+/// `GET /v3/status-job`
+///
+/// No job can exist, so a well-formed `jobId` always yields a `NoSuchJob`-style
+/// error rather than a fabricated status.
+pub struct StatusBatchJobHandler {}
+
+#[async_trait::async_trait]
+impl Operation for StatusBatchJobHandler {
+    async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+        validate_batch_job_admin_request(&req, AdminAction::DescribeBatchJobAction).await?;
+        let job_id = require_job_id(&req.uri)?;
+        Err(no_such_job(&job_id))
+    }
+}
+
+/// `GET /v3/describe-job`
+///
+/// No job can exist, so a well-formed `jobId` always yields a `NoSuchJob`-style
+/// error rather than a fabricated definition.
+pub struct DescribeBatchJobHandler {}
+
+#[async_trait::async_trait]
+impl Operation for DescribeBatchJobHandler {
+    async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+        validate_batch_job_admin_request(&req, AdminAction::DescribeBatchJobAction).await?;
+        let job_id = require_job_id(&req.uri)?;
+        Err(no_such_job(&job_id))
+    }
+}
+
+/// `DELETE /v3/cancel-job`
+///
+/// No job can exist, so a well-formed `jobId` always yields a `NoSuchJob`-style
+/// error rather than a fabricated cancellation.
+pub struct CancelBatchJobHandler {}
+
+#[async_trait::async_trait]
+impl Operation for CancelBatchJobHandler {
+    async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+        validate_batch_job_admin_request(&req, AdminAction::CancelBatchJobAction).await?;
+        let job_id = require_job_id(&req.uri)?;
+        Err(no_such_job(&job_id))
+    }
+}
+
+fn require_job_id(uri: &Uri) -> S3Result<String> {
+    let queries = extract_query_params(uri);
+    match queries.get("jobId").filter(|v| !v.is_empty()) {
+        Some(job_id) => Ok(job_id.clone()),
+        None => Err(s3_error!(InvalidRequest, "jobId is required")),
+    }
+}
+
+fn no_such_job(job_id: &str) -> S3Error {
+    S3Error::with_message(S3ErrorCode::NoSuchKey, format!("no such batch job: {job_id}"))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::{detect_job_type, extract_query_params, require_job_id};
+    use http::Uri;
+
+    #[test]
+    fn detect_job_type_recognises_top_level_replicate_key() {
+        let body = b"replicate:\n  apiVersion: v1\n  source:\n    bucket: src\n";
+        assert_eq!(detect_job_type(body), Some("replicate".to_string()));
+    }
+
+    #[test]
+    fn detect_job_type_recognises_keyrotate_and_expire() {
+        assert_eq!(detect_job_type(b"keyrotate:\n  foo: bar\n"), Some("keyrotate".to_string()));
+        assert_eq!(detect_job_type(b"expire:\n  foo: bar\n"), Some("expire".to_string()));
+    }
+
+    #[test]
+    fn detect_job_type_skips_comments_and_blank_lines() {
+        let body = b"# a comment\n\nreplicate:\n  foo: bar\n";
+        assert_eq!(detect_job_type(body), Some("replicate".to_string()));
+    }
+
+    #[test]
+    fn detect_job_type_ignores_indented_keys() {
+        // An indented `replicate:` is a nested field, not the top-level job type.
+        let body = b"something:\n  replicate: true\n";
+        assert_eq!(detect_job_type(body), None);
+    }
+
+    #[test]
+    fn detect_job_type_returns_none_for_unknown_type() {
+        assert_eq!(detect_job_type(b"transmogrify:\n  foo: bar\n"), None);
+    }
+
+    #[test]
+    fn extract_query_params_decodes_job_id() {
+        let uri: Uri = "/rustfs/admin/v3/status-job?jobId=abc%2F123"
+            .parse()
+            .expect("uri should parse");
+        let params = extract_query_params(&uri);
+        assert_eq!(params.get("jobId"), Some(&"abc/123".to_string()));
+    }
+
+    #[test]
+    fn require_job_id_rejects_missing_and_empty() {
+        let missing: Uri = "/rustfs/admin/v3/status-job".parse().expect("uri should parse");
+        assert!(require_job_id(&missing).is_err());
+
+        let empty: Uri = "/rustfs/admin/v3/status-job?jobId=".parse().expect("uri should parse");
+        assert!(require_job_id(&empty).is_err());
+
+        let present: Uri = "/rustfs/admin/v3/status-job?jobId=job-1".parse().expect("uri should parse");
+        assert_eq!(require_job_id(&present).expect("job id"), "job-1");
+    }
+}

rustfs/src/admin/handlers/mod.rs

@@ -15,6 +15,7 @@
 pub mod account_info;
 pub mod audit;
 mod audit_runtime_config;
+pub mod batch_job;
 pub mod bucket_meta;
 pub mod cluster_snapshot;
 pub mod config_admin;

rustfs/src/admin/handlers/replication.rs

@@ -19,10 +19,11 @@ use crate::admin::runtime_sources::{current_object_store_handle, current_replica
 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::BucketStats;
+use crate::admin::storage_api::bucket::replication::{BucketStats, ReplicationStatusType};
 use crate::admin::storage_api::bucket::target::{BucketTarget, BucketTargetType, Credentials as TargetCredentials, LatencyStat};
 use crate::admin::storage_api::bucket::target_sys::{BucketTargetError, BucketTargetSys};
 use crate::admin::storage_api::contract::bucket::{BucketOperations, BucketOptions};
+use crate::admin::storage_api::contract::list::ListOperations as _;
 use crate::admin::storage_api::error::StorageError;
 use crate::admin::utils::read_compatible_admin_body;
 use crate::auth::{check_key_valid, get_session_token};
@@ -36,7 +37,7 @@ use rustfs_credentials::Credentials;
 use rustfs_policy::policy::action::{Action, AdminAction};
 use s3s::header::CONTENT_TYPE;
 use s3s::{Body, S3Error, S3ErrorCode, S3Request, S3Response, S3Result, s3_error};
-use serde::Deserialize;
+use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
 use std::time::Duration;
 use time::OffsetDateTime;
@@ -247,6 +248,18 @@ pub fn register_replication_route(r: &mut S3Router<AdminOperation>) -> std::io::
         AdminOperation(&RemoveRemoteTargetHandler {}),
     )?;
 
+    r.insert(
+        Method::POST,
+        format!("{}{}", ADMIN_PREFIX, "/v3/replication/diff").as_str(),
+        AdminOperation(&ReplicationDiffHandler {}),
+    )?;
+
+    r.insert(
+        Method::GET,
+        format!("{}{}", ADMIN_PREFIX, "/v3/replication/mrf").as_str(),
+        AdminOperation(&ReplicationMrfHandler {}),
+    )?;
+
     Ok(())
 }
 
@@ -584,6 +597,301 @@ impl Operation for RemoveRemoteTargetHandler {
     }
 }
 
+/// Upper bound on the number of object versions scanned per `POST
+/// /v3/replication/diff` request. RustFS has no persisted per-object
+/// replication-diff index, so the diff is computed by scanning object versions
+/// on demand. Cap the work so a single admin call cannot walk an arbitrarily
+/// large bucket. When the scan is truncated, `is_truncated` is set on the
+/// response so clients know the diff is partial.
+const REPLICATION_DIFF_MAX_SCAN: usize = 10_000;
+
+/// Number of object versions requested per `list_object_versions` page while
+/// computing a replication diff.
+const REPLICATION_DIFF_PAGE_SIZE: i32 = 1_000;
+
+/// A single object version whose replication is not yet complete, reported by
+/// `POST /v3/replication/diff`. Field names mirror MinIO's `madmin.DiffInfo`
+/// so MinIO-compatible admin clients can parse the response.
+#[derive(Debug, Serialize)]
+struct ReplicationDiffEntry {
+    #[serde(rename = "Object")]
+    object: String,
+    #[serde(rename = "VersionID", skip_serializing_if = "Option::is_none")]
+    version_id: Option<String>,
+    #[serde(rename = "Size")]
+    size: i64,
+    #[serde(rename = "IsDeleteMarker")]
+    is_delete_marker: bool,
+    #[serde(rename = "ReplicationStatus")]
+    replication_status: String,
+    #[serde(rename = "LastModified", skip_serializing_if = "Option::is_none")]
+    last_modified: Option<String>,
+}
+
+/// Response body for `POST /v3/replication/diff`.
+///
+/// `entries` lists object versions with a `PENDING` or `FAILED` replication
+/// status. `is_truncated` indicates the on-demand scan hit
+/// [`REPLICATION_DIFF_MAX_SCAN`] before reaching the end of the bucket, so the
+/// diff is partial and should be re-run with a narrower prefix.
+#[derive(Debug, Serialize)]
+struct ReplicationDiffResponse {
+    #[serde(rename = "Entries")]
+    entries: Vec<ReplicationDiffEntry>,
+    #[serde(rename = "IsTruncated")]
+    is_truncated: bool,
+    #[serde(rename = "ScannedVersions")]
+    scanned_versions: usize,
+}
+
+#[derive(Debug, Default, Deserialize)]
+#[serde(deny_unknown_fields)]
+struct ReplicationDiffRequest {
+    #[serde(default)]
+    prefix: String,
+}
+
+/// `POST /v3/replication/diff`
+///
+/// Computes, on demand, the set of object versions in a bucket whose replication
+/// is still `PENDING` or has `FAILED`. RustFS stores the replication status on
+/// each object version (`x-amz-replication-status`) but has no pre-built diff
+/// index, so this handler scans object versions (bounded by
+/// [`REPLICATION_DIFF_MAX_SCAN`]) and returns the not-yet-replicated versions.
+///
+/// The bucket must exist and have a replication configuration, matching MinIO's
+/// behavior of returning `ReplicationConfigurationNotFoundError` otherwise.
+pub struct ReplicationDiffHandler {}
+
+#[async_trait::async_trait]
+impl Operation for ReplicationDiffHandler {
+    async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+        let cred = validate_replication_admin_request(&req, AdminAction::ReplicationDiff).await?;
+
+        let queries = extract_query_params(&req.uri);
+        let Some(bucket) = queries.get("bucket").filter(|b| !b.is_empty()).cloned() else {
+            return Err(s3_error!(InvalidRequest, "bucket is required"));
+        };
+
+        let Some(store) = current_object_store_handle() else {
+            return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
+        };
+
+        store
+            .get_bucket_info(&bucket, &BucketOptions::default())
+            .await
+            .map_err(ApiError::from)?;
+
+        // A replication diff is only meaningful for a bucket that is configured
+        // for replication; mirror MinIO's not-found semantics otherwise.
+        if let Err(err) = get_replication_config(&bucket).await {
+            if err == StorageError::ConfigNotFound {
+                return Err(S3Error::with_message(
+                    S3ErrorCode::ReplicationConfigurationNotFoundError,
+                    "replication configuration not found".to_string(),
+                ));
+            }
+            return Err(ApiError::from(err).into());
+        }
+
+        // Optional prefix can be supplied either as a query parameter (MinIO
+        // clients) or, for RustFS clients, as a small JSON body.
+        let mut prefix = queries.get("prefix").cloned().unwrap_or_default();
+        let body = read_compatible_admin_body(req.input, MAX_ADMIN_REQUEST_BODY_SIZE, req.uri.path(), &cred.secret_key)
+            .await
+            .unwrap_or_default();
+        if prefix.is_empty() && !body.trim_ascii().is_empty() {
+            match serde_json::from_slice::<ReplicationDiffRequest>(&body) {
+                Ok(parsed) => prefix = parsed.prefix,
+                Err(e) => return Err(s3_error!(InvalidRequest, "invalid replication diff request body: {e}")),
+            }
+        }
+
+        let mut entries: Vec<ReplicationDiffEntry> = Vec::new();
+        let mut scanned_versions: usize = 0;
+        let mut marker: Option<String> = None;
+        let mut version_marker: Option<String> = None;
+        let mut is_truncated = false;
+
+        'scan: loop {
+            let listing = store
+                .clone()
+                .list_object_versions(&bucket, &prefix, marker.clone(), version_marker.clone(), None, REPLICATION_DIFF_PAGE_SIZE)
+                .await
+                .map_err(ApiError::from)?;
+
+            for object in &listing.objects {
+                scanned_versions += 1;
+
+                if matches!(object.replication_status, ReplicationStatusType::Pending | ReplicationStatusType::Failed) {
+                    entries.push(ReplicationDiffEntry {
+                        object: object.name.clone(),
+                        version_id: object.version_id.map(|v| v.to_string()),
+                        size: object.size,
+                        is_delete_marker: object.delete_marker,
+                        replication_status: object.replication_status.as_str().to_string(),
+                        last_modified: object
+                            .mod_time
+                            .and_then(|t| t.format(&time::format_description::well_known::Rfc3339).ok()),
+                    });
+                }
+
+                if scanned_versions >= REPLICATION_DIFF_MAX_SCAN {
+                    // We stopped early; the diff is partial.
+                    is_truncated =
+                        listing.is_truncated || listing.next_marker.is_some() || listing.next_version_idmarker.is_some();
+                    break 'scan;
+                }
+            }
+
+            if !listing.is_truncated {
+                break;
+            }
+            marker = listing.next_marker;
+            version_marker = listing.next_version_idmarker;
+            if marker.is_none() && version_marker.is_none() {
+                break;
+            }
+        }
+
+        debug!(
+            bucket = %bucket,
+            prefix = %prefix,
+            scanned = scanned_versions,
+            pending_or_failed = entries.len(),
+            truncated = is_truncated,
+            "computed replication diff"
+        );
+
+        let response = ReplicationDiffResponse {
+            entries,
+            is_truncated,
+            scanned_versions,
+        };
+        let data = serde_json::to_vec(&response)
+            .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("serialize failed: {e}")))?;
+        let mut headers = HeaderMap::new();
+        headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));
+        Ok(S3Response::with_headers((StatusCode::OK, Body::from(data)), headers))
+    }
+}
+
+/// Failed-replication backlog for one remote target (ARN), summarised from the
+/// replication stats. RustFS tracks failed replications as aggregate counters,
+/// not an enumerable per-object MRF queue, so this reports the aggregate.
+#[derive(Debug, Serialize)]
+struct MrfTargetBacklog {
+    #[serde(rename = "ARN")]
+    arn: String,
+    #[serde(rename = "FailedCount")]
+    failed_count: i64,
+    #[serde(rename = "FailedSize")]
+    failed_size: i64,
+}
+
+/// Response body for `GET /v3/replication/mrf`.
+///
+/// MinIO streams individual MRF (most-recently-failed) entries. RustFS persists
+/// MRF entries to disk but does not expose a runtime query over that queue, so
+/// this endpoint reports the aggregate failed-replication backlog and the
+/// in-queue counters instead. See the handler docs and the PR limitations note.
+#[derive(Debug, Serialize)]
+struct MrfResponse {
+    #[serde(rename = "Bucket")]
+    bucket: String,
+    #[serde(rename = "Targets")]
+    targets: Vec<MrfTargetBacklog>,
+    #[serde(rename = "TotalFailedCount")]
+    total_failed_count: i64,
+    #[serde(rename = "TotalFailedSize")]
+    total_failed_size: i64,
+    #[serde(rename = "QueuedCount")]
+    queued_count: i64,
+    #[serde(rename = "QueuedSize")]
+    queued_size: i64,
+    #[serde(rename = "PerObjectEntriesAvailable")]
+    per_object_entries_available: bool,
+}
+
+/// `GET /v3/replication/mrf`
+///
+/// Reports the failed-replication backlog (MinIO's MRF concept) for a bucket.
+///
+/// Compatibility note: MinIO returns a stream of individual MRF entries drained
+/// from its in-memory MRF queue. RustFS records failed replications as aggregate
+/// counters in the replication stats and persists MRF entries to disk without a
+/// runtime query API, so this handler returns the aggregate failed and queued
+/// counts per target instead of per-object rows. `PerObjectEntriesAvailable` is
+/// always `false` to make that limitation explicit to clients.
+pub struct ReplicationMrfHandler {}
+
+#[async_trait::async_trait]
+impl Operation for ReplicationMrfHandler {
+    async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+        validate_replication_admin_request(&req, AdminAction::GetReplicationMetricsAction).await?;
+
+        let queries = extract_query_params(&req.uri);
+        let Some(bucket) = queries.get("bucket").filter(|b| !b.is_empty()).cloned() else {
+            return Err(s3_error!(InvalidRequest, "bucket is required"));
+        };
+
+        let Some(store) = current_object_store_handle() else {
+            return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
+        };
+
+        store
+            .get_bucket_info(&bucket, &BucketOptions::default())
+            .await
+            .map_err(ApiError::from)?;
+
+        if let Err(err) = get_replication_config(&bucket).await {
+            if err == StorageError::ConfigNotFound {
+                return Err(S3Error::with_message(
+                    S3ErrorCode::ReplicationConfigurationNotFoundError,
+                    "replication configuration not found".to_string(),
+                ));
+            }
+            return Err(ApiError::from(err).into());
+        }
+
+        let bucket_stats = match current_replication_stats_handle() {
+            Some(s) => s.get_latest_replication_stats(&bucket).await,
+            None => BucketStats::default(),
+        };
+
+        let mut targets: Vec<MrfTargetBacklog> = Vec::new();
+        let mut total_failed_count: i64 = 0;
+        let mut total_failed_size: i64 = 0;
+        for (arn, stat) in &bucket_stats.replication_stats.stats {
+            total_failed_count += stat.failed.count;
+            total_failed_size += stat.failed.size;
+            targets.push(MrfTargetBacklog {
+                arn: arn.clone(),
+                failed_count: stat.failed.count,
+                failed_size: stat.failed.size,
+            });
+        }
+        targets.sort_by(|a, b| a.arn.cmp(&b.arn));
+
+        let queued = &bucket_stats.replication_stats.q_stat.curr;
+        let response = MrfResponse {
+            bucket,
+            targets,
+            total_failed_count,
+            total_failed_size,
+            queued_count: queued.count,
+            queued_size: queued.bytes,
+            per_object_entries_available: false,
+        };
+
+        let data = serde_json::to_vec(&response)
+            .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("serialize failed: {e}")))?;
+        let mut headers = HeaderMap::new();
+        headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));
+        Ok(S3Response::with_headers((StatusCode::OK, Body::from(data)), headers))
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::{RemoteTargetRequest, extract_query_params, validate_remote_target_tls_settings};

rustfs/src/admin/mod.rs

@@ -33,8 +33,8 @@ mod console_test;
 mod route_registration_test;
 
 use handlers::{
-    audit, bucket_meta, cluster_snapshot, config_admin, extensions, heal, health, kms, module_switch, object_zip_download, oidc,
+    audit, batch_job, bucket_meta, cluster_snapshot, config_admin, extensions, heal, health, kms, module_switch,
-    plugins_catalog, plugins_instances, pools, profile_admin, quota as quota_handler, rebalance,
+    object_zip_download, oidc, plugins_catalog, plugins_instances, pools, profile_admin, quota as quota_handler, rebalance,
     replication as replication_handler, scanner, site_replication, sts, system, table_catalog, tier, tls_debug, user,
 };
 use router::{AdminOperation, S3Router};
@@ -79,6 +79,7 @@ fn register_admin_routes(r: &mut S3Router<AdminOperation>) -> std::io::Result<()
     plugins_instances::register_plugin_instance_route(r)?;
 
     replication_handler::register_replication_route(r)?;
+    batch_job::register_batch_job_route(r)?;
     site_replication::register_site_replication_route(r)?;
     profile_admin::register_profiling_route(r)?;
     tls_debug::register_tls_debug_route(r)?;

rustfs/src/admin/route_policy.rs

@@ -82,6 +82,12 @@ const SITE_REPLICATION_OPERATION: AdminActionRef = AdminActionRef::new("SiteRepl
 const SITE_REPLICATION_REMOVE: AdminActionRef = AdminActionRef::new("SiteReplicationRemoveAction");
 const SITE_REPLICATION_RESYNC: AdminActionRef = AdminActionRef::new("SiteReplicationResyncAction");
 const STORAGE_INFO: AdminActionRef = AdminActionRef::new("StorageInfoAdminAction");
+// MinIO admin compat: batch jobs and replication diff.
+const START_BATCH_JOB: AdminActionRef = AdminActionRef::new("StartBatchJobAction");
+const LIST_BATCH_JOBS: AdminActionRef = AdminActionRef::new("ListBatchJobsAction");
+const DESCRIBE_BATCH_JOB: AdminActionRef = AdminActionRef::new("DescribeBatchJobAction");
+const CANCEL_BATCH_JOB: AdminActionRef = AdminActionRef::new("CancelBatchJobAction");
+const REPLICATION_DIFF: AdminActionRef = AdminActionRef::new("ReplicationDiff");
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
 pub enum DeferredRoutePolicyReason {
@@ -1172,6 +1178,35 @@ pub const ADMIN_ROUTE_POLICY_SPECS: &[AdminRouteSpec] = &[
         COMMIT_TABLE,
         RouteRiskLevel::High,
     ),
+    // MinIO admin compat: batch job lifecycle (backlog#613).
+    admin(HttpMethod::Post, "/rustfs/admin/v3/start-job", START_BATCH_JOB, RouteRiskLevel::High),
+    admin(HttpMethod::Get, "/rustfs/admin/v3/list-jobs", LIST_BATCH_JOBS, RouteRiskLevel::Sensitive),
+    admin(
+        HttpMethod::Get,
+        "/rustfs/admin/v3/status-job",
+        DESCRIBE_BATCH_JOB,
+        RouteRiskLevel::Sensitive,
+    ),
+    admin(
+        HttpMethod::Get,
+        "/rustfs/admin/v3/describe-job",
+        DESCRIBE_BATCH_JOB,
+        RouteRiskLevel::Sensitive,
+    ),
+    admin(HttpMethod::Delete, "/rustfs/admin/v3/cancel-job", CANCEL_BATCH_JOB, RouteRiskLevel::High),
+    // MinIO admin compat: replication diff / MRF inspection (backlog#611).
+    admin(
+        HttpMethod::Post,
+        "/rustfs/admin/v3/replication/diff",
+        REPLICATION_DIFF,
+        RouteRiskLevel::Sensitive,
+    ),
+    admin(
+        HttpMethod::Get,
+        "/rustfs/admin/v3/replication/mrf",
+        GET_REPLICATION_METRICS,
+        RouteRiskLevel::Sensitive,
+    ),
 ];
 
 pub const DEFERRED_ADMIN_ROUTE_POLICIES: &[DeferredAdminRoutePolicy] = &[

rustfs/src/admin/route_registration_test.rs

@@ -244,6 +244,13 @@ fn expected_admin_route_matrix() -> Vec<RouteMatrixEntry> {
         admin_route(Method::GET, "/v3/replicationmetrics"),
         admin_route(Method::PUT, "/v3/set-remote-target"),
         admin_route(Method::DELETE, "/v3/remove-remote-target"),
+        admin_route(Method::POST, "/v3/replication/diff"),
+        admin_route(Method::GET, "/v3/replication/mrf"),
+        admin_route(Method::POST, "/v3/start-job"),
+        admin_route(Method::GET, "/v3/list-jobs"),
+        admin_route(Method::GET, "/v3/status-job"),
+        admin_route(Method::GET, "/v3/describe-job"),
+        admin_route(Method::DELETE, "/v3/cancel-job"),
         admin_route(Method::PUT, "/v3/site-replication/add"),
         admin_route(Method::PUT, "/v3/site-replication/remove"),
         admin_route(Method::GET, "/v3/site-replication/info"),
@@ -1125,6 +1132,13 @@ fn test_register_routes_cover_representative_admin_paths() {
     assert_route(&router, Method::PUT, &admin_path("/v3/import-bucket-metadata"));
     assert_route(&router, Method::GET, &admin_path("/v3/list-remote-targets"));
     assert_route(&router, Method::PUT, &admin_path("/v3/set-remote-target"));
+    assert_route(&router, Method::POST, &admin_path("/v3/replication/diff"));
+    assert_route(&router, Method::GET, &admin_path("/v3/replication/mrf"));
+    assert_route(&router, Method::POST, &admin_path("/v3/start-job"));
+    assert_route(&router, Method::GET, &admin_path("/v3/list-jobs"));
+    assert_route(&router, Method::GET, &admin_path("/v3/status-job"));
+    assert_route(&router, Method::GET, &admin_path("/v3/describe-job"));
+    assert_route(&router, Method::DELETE, &admin_path("/v3/cancel-job"));
     assert_route(&router, Method::PUT, &admin_path("/v3/site-replication/add"));
     assert_route(&router, Method::PUT, &admin_path("/v3/site-replication/remove"));
     assert_route(&router, Method::GET, &admin_path("/v3/site-replication/info"));