Compare commits

..

6 Commits

Author SHA1 Message Date
houseme 7c3d620e09 docs(operations): split scanner excess alerts into English and Chinese pages
The page shipped Chinese-only; keep it as scanner-excess-alerts_zh.md and
add a faithful English translation at the original path, cross-linked at
the top of both.

Co-Authored-By: heihutu <heihutu@gmail.com>
2026-08-18 01:19:39 +08:00
houseme 8c9645924a feat(scanner): emit excess alerts as S3 notification events
The excess-versions / excess-version-size / excess-folders alerts were
metrics-and-logs only; consoles and external auditors had no way to hear
them (rustfs/backlog#1868, HS-04). MinIO emits s3:ObjectManyVersions /
s3:ObjectLargeVersions / s3:PrefixManyFolders for the same conditions —
RustFS carries those as EventName::Scanner* with s3:Scanner:* wire names
that already existed unpublished.

The three alert sites now also dispatch through the standard event
pipeline (send_event via the storage_api owner facade), carrying the
actual values and thresholds in req_params and UserAgent "Scanner".
Without a cooldown a single over-threshold object would re-emit on every
~60s scan cycle, so emissions are edge-held per (kind, bucket, object)
for 24h (RUSTFS_SCANNER_ALERT_COOLDOWN_SECS, 0 = every cycle), backed by
a process-global map with a 4096-key hard cap that clears rather than
grows. Metrics and structured logs stay level-triggered every cycle;
only the notification events are held back. A restart resets the
cooldown deliberately: one re-emission per still-hot key buys back
visibility after the restarts that accompany incident response.

Tests pin the edge-hold semantics (first fires, immediate re-check held,
independent keys, cooldown expiry re-fires, zero cooldown always emits,
hard bound) in one sequential test for the process-global map, and pin
the emitted wire names against EventName's canonical string forms so a
subscribed bucket notification can never silently stop matching.
docs/operations/scanner-excess-alerts.md documents the three events,
the metric-vs-event cadence difference, and the HS-15 threshold deltas
(alert_excess_folders 65538 vs MinIO 50000 is deliberate: Proxmox
Backup Server chunk layout compatibility).

Closes rustfs/backlog#1868.

Co-Authored-By: heihutu <heihutu@gmail.com>
2026-08-18 01:07:48 +08:00
唐小鸭 e0b87b0e7e fix(site-replication): admit only verifiable peer-edit fences (#6123) 2026-08-17 09:47:36 +00:00
houseme 984c705713 docs(ecstore): fix bitrot comment typo (#6168)
Co-authored-by: heihutu <heihutu@gmail.com>
2026-08-17 08:24:54 +00:00
houseme 23b17c2d5a feat(madmin): add a SigV4-signed admin client for heal and scanner APIs (HS-05) (#6166)
feat(madmin): add a SigV4-signed admin client for heal and scanner APIs

The madmin crate held only wire types; automation and mc-style tooling
had no way to drive the heal/scanner admin surface without hand-rolled
HTTP. Add `AdminClient`, which signs with the same rustfs-signer path
the server authenticates (UNSIGNED-PAYLOAD marker, matching RustFS peer
admin calls) and wraps:

- heal_start / heal_status / heal_stop over POST /rustfs/admin/v3/heal/
  (bucket/prefix path params percent-encoded per segment; stop models
  the server's two cancel branches: token-scoped task status vs
  path-scoped start-success receipt);
- background_heal_status, scanner_status (freshness typed), plus
  ilm_expiry_status / replacement_recovery_status passthroughs;
- a public get_json escape hatch for endpoints not wrapped yet.

Wire types follow the madmin-go model (SDK-owned mirrors pinned by
round-trip tests): HealOpts with serde defaults so partial settings
objects decode, HealScanMode accepting both the numeric and name
encodings, and status structs that type the fields operators branch on
while flattening unknown nested payloads verbatim so server additions
cannot break the client. Errors map to a closed AdminClientError enum
(InvalidEndpoint / Transport / HttpStatus with body / Decode).

Tests cover wire round-trips, path building, both stop branches, error
mapping, and — via a dependency-free raw-TCP test server — that signed
requests carry a SigV4 Authorization header, the right method/path/
query, and the expected JSON body.

Closes rustfs/backlog#1869 (first increment; single-sourcing the wire
structs server-side and an embedded-server e2e roundtrip are noted as
follow-ups there).

Co-authored-by: heihutu <heihutu@gmail.com>
2026-08-17 15:04:49 +08:00
houseme 89e2513205 feat(ecstore): pin bitrot algorithms with a startup self-test (HS-11) (#6165)
feat(ecstore): pin bitrot algorithms with a startup self-test

A drifted HighwayHash implementation fails silently: every shard reads
back corrupt, heal rewrites healthy data, and cross-platform clusters
disagree about which copy is good. Mirror MinIO's bitrotSelfTest by
verifying, once at process start:

- known-answer digests for HighwayHash256S / HighwayHash256SLegacy over
  a deterministic 4096-byte xorshift64* payload, plus the externally
  verifiable FIPS SHA-256 "abc" vector guarding the HashAlgorithm
  plumbing itself;
- an end-to-end roundtrip per streaming variant (encode -> size formula
  -> bitrot_verify -> BitrotReader read-back), over full blocks and a
  partial tail;
- tamper detection: one flipped byte in the final data block and one in
  the leading hash must both be rejected as a hash mismatch, not by an
  incidental read error.

The check costs microseconds and runs inline in
init_background_service_runtime before any shard can be written or
verified. Outcome surfaces as one structured bitrot_selftest log event,
the rustfs_bitrot_selftest_status gauge (1=passed / 0=failed / 2=skipped),
a bitrotSelftest field on the admin server-info response, and
RUSTFS_BITROT_SELFTEST_STRICT=on turns a failure into a startup error
(MinIO Fatal parity; the default only degrades the status so a bad build
cannot brick an existing fleet on upgrade).

Closes rustfs/backlog#1873 (HS-11).

Co-authored-by: heihutu <heihutu@gmail.com>
2026-08-17 15:04:23 +08:00
40 changed files with 3198 additions and 4310 deletions
Generated
+6
View File
@@ -9825,14 +9825,19 @@ name = "rustfs-madmin"
version = "1.0.0-rc.2"
dependencies = [
"hotpath",
"http 1.5.0",
"humantime",
"hyper",
"jiff",
"reqwest",
"rmp-serde",
"rustfs-signer",
"s3s",
"serde",
"serde_json",
"sysinfo",
"time",
"tokio",
]
[[package]]
@@ -10246,6 +10251,7 @@ dependencies = [
"rustfs-ecstore",
"rustfs-filemeta",
"rustfs-lock",
"rustfs-s3-types",
"rustfs-storage-api",
"rustfs-utils",
"s3s",
+1
View File
@@ -40,6 +40,7 @@ mak = "mak"
gae = "gae"
GAE = "GAE"
thr = "thr"
mis = "mis"
# s3-tests original test names (cannot be changed)
nonexisted = "nonexisted"
consts = "consts"
+2 -2
View File
@@ -4,8 +4,8 @@ This module is the shared failure-injection boundary for replication end-to-end
`FakeS3Target::start()` creates the listener. Add target buckets with `create_bucket`, point a RustFS remote target at `address()`, use `FAKE_ACCESS_KEY` / `FAKE_SECRET_KEY`, then enqueue per-operation faults with `inject`. Faults for one operation are consumed in FIFO order and do not consume faults queued for another operation. A fault is consumed only after `s3s` verifies the full request signature, so anonymous, other-access-key, and bad-signature traffic cannot disturb a script.
Supported data operations are HeadBucket, GetBucketVersioning, PUT/GET/HEAD/DELETE Object, Get/Put/Delete ObjectTagging (tags live per version; Put replaces the whole set, Delete clears it), and create/upload/complete/abort multipart upload. `create_bucket` models general-purpose buckets in S3's shared global namespace; account-regional namespace buckets and their `-an` names are intentionally out of scope. Buckets are versioned: PUT creates a version, DELETE without `versionId` creates a delete marker, and DELETE with `versionId` removes exactly that version. Internal source version IDs must be UUIDs and are stored canonically. Source mtime is honored only for source-replication PUT/DELETE requests; absent or invalid values use receipt time, matching RustFS, while multipart completion always uses receipt time. Replicated versions are ordered newest-first by source mtime so late older versions and delete markers do not become current. Equal mtimes prefer objects over delete markers, then canonical UUID order; RustFS's internal FileMeta signature tie-break is intentionally out of scope because it is not part of the target S3 protocol. Multipart part numbers follow S3's `1..=10000` range, and every completed part except the final part must be at least 5 MiB.
Supported data operations are HeadBucket, GetBucketVersioning, PUT/GET/HEAD/DELETE Object, and create/upload/complete/abort multipart upload. `create_bucket` models general-purpose buckets in S3's shared global namespace; account-regional namespace buckets and their `-an` names are intentionally out of scope. Buckets are versioned: PUT creates a version, DELETE without `versionId` creates a delete marker, and DELETE with `versionId` removes exactly that version. Internal source version IDs must be UUIDs and are stored canonically. Source mtime is honored only for source-replication PUT/DELETE requests; absent or invalid values use receipt time, matching RustFS, while multipart completion always uses receipt time. Replicated versions are ordered newest-first by source mtime so late older versions and delete markers do not become current. Equal mtimes prefer objects over delete markers, then canonical UUID order; RustFS's internal FileMeta signature tie-break is intentionally out of scope because it is not part of the target S3 protocol. Multipart part numbers follow S3's `1..=10000` range, and every completed part except the final part must be at least 5 MiB.
Fault actions cover HTTP 401/403/503 responses, pre-dispatch delay, connection abort when a logical request-body threshold is reached, streaming slow drain, and a deliberately wrong response ETag (including multipart-complete XML). `requests()` returns the ordered, credential-free request journal for assertions. Each record also journals a `ProxyHeaderSnapshot` — the read-proxy anti-loop marker (`x-{rustfs,minio}-source-proxy-request`), the replication-check exemption header, and the client SSE-C header family (algorithm and key-MD5 values; for the key itself only its presence) — so proxy tests can pin the exact wire contract.
Fault actions cover HTTP 401/403/503 responses, pre-dispatch delay, connection abort when a logical request-body threshold is reached, streaming slow drain, and a deliberately wrong response ETag (including multipart-complete XML). `requests()` returns the ordered, credential-free request journal for assertions.
The listener is loopback-only. It admits at most 64 active connections and two concurrently buffered request bodies; authenticated multipart-complete XML collection and assembly take both body permits. Keep-alive is disabled, request-header reads are bounded to 30 seconds, a parsed request is bounded to 65 seconds, and the complete connection lifetime is bounded to 100 seconds. It retains at most 256 buckets, 4,096 journal entries, 4,096 scripted faults, 4,096 object versions, 256 multipart uploads, and 10,000 multipart parts. Retained identifiers are capped at 1 KiB, user metadata at 2 KiB, and content type at 1 KiB. A PUT or uploaded part is capped at 64 MiB; a completed multipart object and all stored object/part data are capped at 128 MiB. Body drain, body-permit waits, delay, and slow-drain execution are bounded to 30 seconds; each slow-drain slice delay must be below that bound.
+5 -284
View File
@@ -30,12 +30,10 @@ use s3s::access::{S3Access, S3AccessContext};
use s3s::auth::SimpleAuth;
use s3s::dto::{
AbortMultipartUploadInput, AbortMultipartUploadOutput, CompleteMultipartUploadInput, CompleteMultipartUploadOutput,
CreateMultipartUploadInput, CreateMultipartUploadOutput, DeleteMarkerEntry, DeleteObjectInput, DeleteObjectOutput,
DeleteObjectTaggingInput, DeleteObjectTaggingOutput, ETag, GetBucketVersioningInput, GetBucketVersioningOutput,
GetObjectInput, GetObjectOutput, GetObjectTaggingInput, GetObjectTaggingOutput, HeadBucketInput, HeadBucketOutput,
CreateMultipartUploadInput, CreateMultipartUploadOutput, DeleteMarkerEntry, DeleteObjectInput, DeleteObjectOutput, ETag,
GetBucketVersioningInput, GetBucketVersioningOutput, GetObjectInput, GetObjectOutput, HeadBucketInput, HeadBucketOutput,
HeadObjectInput, HeadObjectOutput, ListObjectVersionsInput, ListObjectVersionsOutput, ObjectVersionId, PutObjectInput,
PutObjectOutput, PutObjectTaggingInput, PutObjectTaggingOutput, StreamingBlob, Tag, TagSet, Timestamp, TimestampFormat,
UploadPartInput, UploadPartOutput,
PutObjectOutput, StreamingBlob, Timestamp, TimestampFormat, UploadPartInput, UploadPartOutput,
};
use s3s::service::{S3Service, S3ServiceBuilder};
use s3s::validation::{AwsNameValidation, NameValidation};
@@ -90,13 +88,6 @@ const SOURCE_LEGALHOLD_TIMESTAMP_HEADERS: [&str; 2] = [
"x-rustfs-source-replication-legalhold-timestamp",
"x-minio-source-replication-legalhold-timestamp",
];
/// Wire prefix of the SSE-C passthrough replication transport headers
/// (`X-Rustfs-Replication-*`). In the default mode the fake stores them like a
/// RustFS target and echoes SSE-C evidence back on HEAD/GET; with
/// [`FakeS3Target::drop_unlisted_replication_headers`] it models MinIO /
/// generic S3, which silently discard unknown x-* headers.
const REPLICATION_SSE_TRANSPORT_PREFIX: &str = "x-rustfs-replication-";
const REPLICATION_SSEC_ALGORITHM_TRANSPORT_HEADER: &str = "x-rustfs-replication-ssec-algorithm";
const RESERVED_BUCKET_PREFIXES: [&str; 3] = ["xn--", "sthree-", "amzn-s3-demo-"];
const RESERVED_BUCKET_SUFFIXES: [&str; 6] = ["-s3alias", "--ol-s3", ".mrap", "--x-s3", "--table-s3", "-an"];
@@ -112,9 +103,6 @@ pub enum Operation {
GetObject,
HeadObject,
DeleteObject,
GetObjectTagging,
PutObjectTagging,
DeleteObjectTagging,
ListObjectVersions,
CreateMultipartUpload,
UploadPart,
@@ -161,42 +149,6 @@ impl ReplicationTimestampHeaders {
}
}
/// Read-proxy related headers observed on a request, journaled so proxy
/// tests can assert the exact wire contract: the anti-loop marker present,
/// the replication-check exemption absent, and the client SSE-C key family
/// forwarded verbatim. The SSE-C key value itself is never retained — only
/// its presence.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct ProxyHeaderSnapshot {
pub source_proxy_request: Option<String>,
pub replication_check: Option<String>,
pub ssec_algorithm: Option<String>,
pub ssec_key_present: bool,
pub ssec_key_md5: Option<String>,
/// Whether the request carried any `X-Rustfs-Replication-*` SSE-C
/// passthrough transport header, so fail-closed tests can assert the
/// sender really shipped the material a dropping target discarded.
pub ssec_transport_present: bool,
}
impl ProxyHeaderSnapshot {
fn from_headers(headers: &HeaderMap) -> Self {
Self {
source_proxy_request: header_value(headers, &["x-rustfs-source-proxy-request", "x-minio-source-proxy-request"])
.map(bounded_journal_value),
replication_check: header_value(headers, &["x-rustfs-source-replication-check", "x-minio-source-replication-check"])
.map(bounded_journal_value),
ssec_algorithm: header_value(headers, &["x-amz-server-side-encryption-customer-algorithm"])
.map(bounded_journal_value),
ssec_key_present: headers.contains_key("x-amz-server-side-encryption-customer-key"),
ssec_key_md5: header_value(headers, &["x-amz-server-side-encryption-customer-key-md5"]).map(bounded_journal_value),
ssec_transport_present: headers
.keys()
.any(|name| name.as_str().starts_with(REPLICATION_SSE_TRANSPORT_PREFIX)),
}
}
}
/// Credential-free request metadata retained for deterministic assertions.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RequestRecord {
@@ -211,7 +163,6 @@ pub struct RequestRecord {
pub content_length: Option<u64>,
pub consumed_bytes: Option<usize>,
pub replication_timestamps: ReplicationTimestampHeaders,
pub proxy_headers: ProxyHeaderSnapshot,
pub fault: Option<FaultAction>,
}
@@ -227,10 +178,6 @@ struct ControlState {
struct StoreState {
assign_own_version_ids: bool,
assign_own_multipart_version_ids: bool,
/// MinIO-like mode: silently discard non-whitelisted replication
/// transport headers instead of storing them (see
/// [`REPLICATION_SSE_TRANSPORT_PREFIX`]).
drop_unlisted_replication_headers: bool,
buckets: HashMap<String, BucketState>,
uploads: HashMap<String, MultipartState>,
total_bytes: usize,
@@ -252,12 +199,6 @@ struct ObjectVersion {
delete_marker: bool,
content_type: Option<String>,
metadata: Option<HashMap<String, String>>,
/// Object tags as ordered key/value pairs (PutObjectTagging replaces the
/// whole set, DeleteObjectTagging clears it).
tags: Vec<(String, String)>,
/// SSE-C passthrough transport headers stored with the version (RustFS
/// target behavior); empty when the drop mode discarded them.
replication_sse_headers: Vec<(String, String)>,
}
#[derive(Clone)]
@@ -267,7 +208,6 @@ struct MultipartState {
version_id: String,
content_type: Option<String>,
metadata: Option<HashMap<String, String>>,
replication_sse_headers: Vec<(String, String)>,
parts: BTreeMap<i32, MultipartPart>,
}
@@ -488,15 +428,6 @@ impl FakeS3Target {
/// Mint own version ids for the multipart path only — models a target
/// that adopts PutObject version ids but not CreateMultipartUpload ones.
/// MinIO-like mode: silently drop every `X-Rustfs-Replication-*` SSE-C
/// passthrough transport header instead of storing it. The default (off)
/// models a RustFS target, which preserves the headers and echoes SSE-C
/// evidence (`x-amz-server-side-encryption-customer-algorithm`) on
/// HEAD/GET of the replica.
pub fn drop_unlisted_replication_headers(&self, enabled: bool) {
lock(&self.backend.store).drop_unlisted_replication_headers = enabled;
}
pub fn assign_own_multipart_version_ids(&self, enabled: bool) {
lock(&self.backend.store).assign_own_multipart_version_ids = enabled;
}
@@ -638,7 +569,6 @@ impl S3Access for FaultAccess {
.and_then(|value| value.to_str().ok())
.and_then(|value| value.parse().ok());
let replication_timestamps = ReplicationTimestampHeaders::from_headers(context.headers());
let proxy_headers = ProxyHeaderSnapshot::from_headers(context.headers());
let fault = record_request(
&self.control,
operation,
@@ -646,7 +576,6 @@ impl S3Access for FaultAccess {
parsed,
content_length,
replication_timestamps,
proxy_headers,
);
if let Some(RequestFault {
action: FaultAction::Status(status),
@@ -686,9 +615,6 @@ fn operation_from_s3_name(name: &str) -> Operation {
"GetObject" => Operation::GetObject,
"HeadObject" => Operation::HeadObject,
"DeleteObject" => Operation::DeleteObject,
"GetObjectTagging" => Operation::GetObjectTagging,
"PutObjectTagging" => Operation::PutObjectTagging,
"DeleteObjectTagging" => Operation::DeleteObjectTagging,
"CreateMultipartUpload" => Operation::CreateMultipartUpload,
"UploadPart" => Operation::UploadPart,
"CompleteMultipartUpload" => Operation::CompleteMultipartUpload,
@@ -704,7 +630,6 @@ fn record_request(
parsed: ParsedRequest,
content_length: Option<u64>,
replication_timestamps: ReplicationTimestampHeaders,
proxy_headers: ProxyHeaderSnapshot,
) -> Option<RequestFault> {
let mut state = lock(control);
let action = parsed
@@ -730,7 +655,6 @@ fn record_request(
content_length,
consumed_bytes: None,
replication_timestamps,
proxy_headers,
fault: action.clone(),
});
action.map(|action| RequestFault { sequence, action })
@@ -797,15 +721,6 @@ fn parse_request(method: &Method, uri: &Uri) -> ParsedRequest {
(&Method::POST, true) if query.contains_key("uploads") => Operation::CreateMultipartUpload,
(&Method::POST, true) if upload_id.is_some() => Operation::CompleteMultipartUpload,
(&Method::DELETE, true) if upload_id.is_some() => Operation::AbortMultipartUpload,
(&Method::GET, true) if query.contains_key("tagging") && only_query_keys(&["tagging", "versionId"]) => {
Operation::GetObjectTagging
}
(&Method::PUT, true) if query.contains_key("tagging") && only_query_keys(&["tagging", "versionId"]) => {
Operation::PutObjectTagging
}
(&Method::DELETE, true) if query.contains_key("tagging") && only_query_keys(&["tagging", "versionId"]) => {
Operation::DeleteObjectTagging
}
// A replication PUT addresses the source version via `?versionId=`.
(&Method::PUT, true) if only_query_keys(&["versionId"]) => Operation::PutObject,
(&Method::GET, true) if only_query_keys(&["versionId"]) => Operation::GetObject,
@@ -873,29 +788,6 @@ fn new_version_id(headers: &HeaderMap, assign_own: bool) -> S3Result<String> {
Ok(version_id.to_string())
}
/// Capture the SSE-C passthrough transport headers a replication PUT carried.
/// Returns an empty set in the MinIO-like drop mode.
fn captured_replication_sse_headers(headers: &HeaderMap, drop_unlisted: bool) -> Vec<(String, String)> {
if drop_unlisted {
return Vec::new();
}
headers
.iter()
.filter(|(name, _)| name.as_str().starts_with(REPLICATION_SSE_TRANSPORT_PREFIX))
.filter_map(|(name, value)| Some((name.as_str().to_string(), value.to_str().ok()?.to_string())))
.collect()
}
/// SSE-C evidence a RustFS-like target echoes for a stored passthrough
/// replica: the customer algorithm restored from the transport headers.
fn stored_sse_customer_algorithm(version: &ObjectVersion) -> Option<String> {
version
.replication_sse_headers
.iter()
.find(|(name, _)| name == REPLICATION_SSEC_ALGORITHM_TRANSPORT_HEADER)
.map(|(_, value)| value.clone())
}
fn source_etag(headers: &HeaderMap) -> S3Result<Option<String>> {
header_value(headers, &SOURCE_ETAG_HEADERS)
.map(|value| validate_retained_identifier(value, "source ETag").map(|value| normalize_etag(&value)))
@@ -1243,33 +1135,6 @@ fn find_version(state: &StoreState, bucket: &str, key: &str, version_id: Option<
Ok(version.clone())
}
/// Replace (or clear, with an empty vec) the tag set of the addressed
/// version, returning its version id. Mirrors `find_version` addressing:
/// explicit version id or the latest version, delete markers rejected.
fn set_version_tags(
state: &mut StoreState,
bucket: &str,
key: &str,
version_id: Option<&str>,
tags: Vec<(String, String)>,
) -> S3Result<String> {
// Resolve first (immutable) so the error paths match find_version.
let resolved = find_version(state, bucket, key, version_id)?.version_id;
let versions = state
.buckets
.get_mut(bucket)
.expect("bucket existence checked by find_version")
.objects
.get_mut(key)
.expect("key existence checked by find_version");
let version = versions
.iter_mut()
.find(|version| version.version_id == resolved)
.expect("version existence checked by find_version");
version.tags = tags;
Ok(resolved)
}
#[async_trait]
impl S3 for FakeBackend {
async fn head_bucket(&self, req: S3Request<HeadBucketInput>) -> S3Result<S3Response<HeadBucketOutput>> {
@@ -1366,10 +1231,7 @@ impl S3 for FakeBackend {
let input = req.input;
let body = collect_stream(input.body, input.content_length, fault.as_ref(), &self.control).await?;
validate_stored_metadata(&input.content_type, &input.metadata)?;
let (assign_own, drop_unlisted) = {
let state = lock(&self.store);
(state.assign_own_version_ids, state.drop_unlisted_replication_headers)
};
let assign_own = lock(&self.store).assign_own_version_ids;
let version_id = new_version_id(&headers, assign_own)?;
let e_tag = match source_etag(&headers)? {
Some(value) => value,
@@ -1386,8 +1248,6 @@ impl S3 for FakeBackend {
delete_marker: false,
content_type: input.content_type,
metadata: input.metadata,
tags: Vec::new(),
replication_sse_headers: captured_replication_sse_headers(&headers, drop_unlisted),
};
upsert_version(&mut lock(&self.store), &input.bucket, input.key, version)?;
Ok(apply_response_fault(
@@ -1408,7 +1268,6 @@ impl S3 for FakeBackend {
let state = lock(&self.store);
find_version(&state, &input.bucket, &input.key, input.version_id.as_deref())?
};
let sse_customer_algorithm = stored_sse_customer_algorithm(&version);
Ok(apply_response_fault(
S3Response::new(GetObjectOutput {
body: Some(StreamingBlob::new(Body::from(version.body.clone()))),
@@ -1418,7 +1277,6 @@ impl S3 for FakeBackend {
e_tag: Some(ETag::Strong(version.e_tag)),
last_modified: Some(version.last_modified.clone()),
version_id: Some(version.version_id),
sse_customer_algorithm,
..Default::default()
}),
fault.as_ref(),
@@ -1433,7 +1291,6 @@ impl S3 for FakeBackend {
let state = lock(&self.store);
find_version(&state, &input.bucket, &input.key, input.version_id.as_deref())?
};
let sse_customer_algorithm = stored_sse_customer_algorithm(&version);
Ok(apply_response_fault(
S3Response::new(HeadObjectOutput {
content_length: Some(version.body.len() as i64),
@@ -1442,79 +1299,12 @@ impl S3 for FakeBackend {
e_tag: Some(ETag::Strong(version.e_tag)),
last_modified: Some(version.last_modified.clone()),
version_id: Some(version.version_id),
sse_customer_algorithm,
..Default::default()
}),
fault.as_ref(),
))
}
async fn get_object_tagging(&self, req: S3Request<GetObjectTaggingInput>) -> S3Result<S3Response<GetObjectTaggingOutput>> {
let fault = request_fault(&req);
apply_non_body_fault(fault.as_ref(), &self.control).await?;
let input = req.input;
let version = {
let state = lock(&self.store);
find_version(&state, &input.bucket, &input.key, input.version_id.as_deref())?
};
let tag_set: TagSet = version
.tags
.into_iter()
.map(|(key, value)| Tag {
key: Some(key),
value: Some(value),
})
.collect();
Ok(apply_response_fault(
S3Response::new(GetObjectTaggingOutput {
tag_set,
version_id: Some(ObjectVersionId::from(version.version_id)),
}),
fault.as_ref(),
))
}
async fn put_object_tagging(&self, req: S3Request<PutObjectTaggingInput>) -> S3Result<S3Response<PutObjectTaggingOutput>> {
let fault = request_fault(&req);
apply_non_body_fault(fault.as_ref(), &self.control).await?;
let input = req.input;
let tags = input
.tagging
.tag_set
.into_iter()
.map(|tag| (tag.key.unwrap_or_default(), tag.value.unwrap_or_default()))
.collect();
let version_id = {
let mut state = lock(&self.store);
set_version_tags(&mut state, &input.bucket, &input.key, input.version_id.as_deref(), tags)?
};
Ok(apply_response_fault(
S3Response::new(PutObjectTaggingOutput {
version_id: Some(ObjectVersionId::from(version_id)),
}),
fault.as_ref(),
))
}
async fn delete_object_tagging(
&self,
req: S3Request<DeleteObjectTaggingInput>,
) -> S3Result<S3Response<DeleteObjectTaggingOutput>> {
let fault = request_fault(&req);
apply_non_body_fault(fault.as_ref(), &self.control).await?;
let input = req.input;
let version_id = {
let mut state = lock(&self.store);
set_version_tags(&mut state, &input.bucket, &input.key, input.version_id.as_deref(), Vec::new())?
};
Ok(apply_response_fault(
S3Response::new(DeleteObjectTaggingOutput {
version_id: Some(ObjectVersionId::from(version_id)),
}),
fault.as_ref(),
))
}
async fn delete_object(&self, req: S3Request<DeleteObjectInput>) -> S3Result<S3Response<DeleteObjectOutput>> {
let fault = request_fault(&req);
apply_non_body_fault(fault.as_ref(), &self.control).await?;
@@ -1591,8 +1381,6 @@ impl S3 for FakeBackend {
delete_marker: true,
content_type: None,
metadata: None,
tags: Vec::new(),
replication_sse_headers: Vec::new(),
},
)?;
Ok(apply_response_fault(
@@ -1620,10 +1408,9 @@ impl S3 for FakeBackend {
ensure_upload_budget(&state)?;
validate_stored_metadata(&input.content_type, &input.metadata)?;
let upload_id = Uuid::new_v4().to_string();
// Read the flags before the mutable borrow of `state.uploads` below
// Read the flag before the mutable borrow of `state.uploads` below
// (and never re-lock the store: the mutex is not reentrant).
let mint_own = state.assign_own_version_ids || state.assign_own_multipart_version_ids;
let drop_unlisted = state.drop_unlisted_replication_headers;
let version_id = new_version_id(&headers, mint_own)?;
state.uploads.insert(
upload_id.clone(),
@@ -1633,7 +1420,6 @@ impl S3 for FakeBackend {
version_id,
content_type: input.content_type,
metadata: input.metadata,
replication_sse_headers: captured_replication_sse_headers(&headers, drop_unlisted),
parts: BTreeMap::new(),
},
);
@@ -1771,7 +1557,6 @@ impl S3 for FakeBackend {
version_id: upload.version_id.clone(),
content_type: upload.content_type.clone(),
metadata: upload.metadata.clone(),
replication_sse_headers: upload.replication_sse_headers.clone(),
parts: BTreeMap::new(),
},
selected,
@@ -1798,8 +1583,6 @@ impl S3 for FakeBackend {
delete_marker: false,
content_type: upload.content_type,
metadata: upload.metadata,
tags: Vec::new(),
replication_sse_headers: upload.replication_sse_headers,
};
let mut state = lock(&self.store);
let current = state
@@ -2004,65 +1787,6 @@ mod tests {
Ok(())
}
/// Default mode is RustFS-like: SSE-C passthrough transport headers are
/// stored and the customer algorithm is echoed on HEAD/GET. Drop mode is
/// MinIO-like: the headers are silently discarded, so no evidence comes
/// back — the exact difference the N2 fail-closed audit keys on. Both
/// modes journal that the sender shipped the transport headers.
#[tokio::test]
async fn ssec_passthrough_headers_echo_and_drop_modes() -> Result<(), BoxError> {
let target = FakeS3Target::start().await?;
target.create_bucket("target-bucket");
let client = client(&target);
let put_with_transport_headers = |key: &'static str| {
client
.put_object()
.bucket("target-bucket")
.key(key)
.body(ByteStream::from_static(b"ciphertext"))
.customize()
.map_request(move |mut request| {
let headers = request.headers_mut();
headers.insert("x-rustfs-replication-ssec-algorithm", "AES256");
headers.insert("x-rustfs-replication-ssec-key-md5", "AAAAAAAAAAAAAAAAAAAAAA==");
Ok::<_, std::convert::Infallible>(request)
})
.send()
};
put_with_transport_headers("kept").await?;
let head = client.head_object().bucket("target-bucket").key("kept").send().await?;
assert_eq!(head.sse_customer_algorithm(), Some("AES256"));
let get = client.get_object().bucket("target-bucket").key("kept").send().await?;
assert_eq!(get.sse_customer_algorithm(), Some("AES256"));
target.drop_unlisted_replication_headers(true);
put_with_transport_headers("dropped").await?;
let head = client.head_object().bucket("target-bucket").key("dropped").send().await?;
assert_eq!(head.sse_customer_algorithm(), None, "drop mode must discard SSE-C evidence");
let requests = target.requests();
for key in ["kept", "dropped"] {
let record = requests
.iter()
.find(|record| record.operation == Operation::PutObject && record.key.as_deref() == Some(key))
.expect("PUT must be journaled");
assert!(
record.proxy_headers.ssec_transport_present,
"the journal must prove the sender shipped the transport headers for {key}"
);
}
let plain_head = requests
.iter()
.find(|record| record.operation == Operation::HeadObject)
.expect("HEAD must be journaled");
assert!(!plain_head.proxy_headers.ssec_transport_present);
target.shutdown().await;
Ok(())
}
macro_rules! assert_sdk_error {
($error:expr, $status:expr, $code:expr) => {{
let error = &$error;
@@ -3328,7 +3052,6 @@ mod tests {
version_id: index.to_string(),
content_type: None,
metadata: None,
replication_sse_headers: Vec::new(),
parts: BTreeMap::new(),
},
);
@@ -3351,7 +3074,6 @@ mod tests {
},
Some(0),
ReplicationTimestampHeaders::default(),
ProxyHeaderSnapshot::default(),
);
}
let records = lock(&control).requests.clone();
@@ -3374,7 +3096,6 @@ mod tests {
},
None,
ReplicationTimestampHeaders::default(),
ProxyHeaderSnapshot::default(),
);
{
let bounded_records = lock(&bounded_control);
@@ -2610,20 +2610,17 @@ async fn test_replication_check_succeeds_with_remote_target() -> Result<(), Box<
assert_eq!(response.status(), StatusCode::OK);
let payload: serde_json::Value = response.json().await?;
assert_eq!(payload["Status"], "OK", "{payload}");
assert_eq!(payload["Status"], "OK");
assert_eq!(payload["ActiveMutation"], true);
assert_eq!(payload["Targets"].as_array().map(Vec::len), Some(1));
assert_eq!(payload["Targets"][0]["Status"], "OK", "{payload}");
assert_eq!(payload["Targets"][0]["Phases"]["Put"]["Status"], "OK", "{payload}");
assert_eq!(payload["Targets"][0]["Status"], "OK");
assert_eq!(payload["Targets"][0]["Phases"]["Put"]["Status"], "OK");
// A RustFS target adopts the source version id, so the P1-19
// version-identity probe passes.
assert_eq!(payload["Targets"][0]["Phases"]["VersionFidelity"]["Status"], "OK", "{payload}");
// A RustFS target preserves the SSE-C passthrough transport headers and
// echoes the customer algorithm on the replication-check HEAD (N2).
assert_eq!(payload["Targets"][0]["Phases"]["SsecPassthrough"]["Status"], "OK", "{payload}");
assert_eq!(payload["Targets"][0]["Phases"]["DeleteMarker"]["Status"], "OK", "{payload}");
assert_eq!(payload["Targets"][0]["Phases"]["VersionDelete"]["Status"], "OK", "{payload}");
assert_eq!(payload["Targets"][0]["Phases"]["Cleanup"]["Status"], "OK", "{payload}");
assert_eq!(payload["Targets"][0]["Phases"]["VersionFidelity"]["Status"], "OK");
assert_eq!(payload["Targets"][0]["Phases"]["DeleteMarker"]["Status"], "OK");
assert_eq!(payload["Targets"][0]["Phases"]["VersionDelete"]["Status"], "OK");
assert_eq!(payload["Targets"][0]["Phases"]["Cleanup"]["Status"], "OK");
let target_client = target_env.create_s3_client();
let versions = target_client
@@ -4652,410 +4649,6 @@ async fn test_bucket_replication_sse_c_multipart_passthrough() -> TestResult {
Ok(())
}
/// N2 (backlog#1675 P1-22): SSE-C passthrough replication to a target that
/// silently drops the `X-Rustfs-Replication-*` transport headers (MinIO-like
/// behavior, modeled by the fake target's drop mode) used to report COMPLETED
/// while the replica had irrecoverably lost its decryption material — the red
/// light this test was born failing on. Fail-closed contract now under test:
/// the first attempt PUTs, HEAD-backs the replica, finds no SSE-C evidence,
/// records the target Unsupported and reports FAILED; a second SSE-C object
/// fails without any PUT reaching the target (capability cache, proven from
/// the target journal); plaintext objects still replicate COMPLETED.
#[tokio::test]
#[serial]
async fn test_ssec_replication_fails_closed_when_target_drops_passthrough_headers() -> TestResult {
init_logging();
let target = FakeS3Target::start().await?;
let target_bucket = "ssec-drop-dst";
target.create_bucket(target_bucket);
target.drop_unlisted_replication_headers(true);
let mut source_env = RustFSTestEnvironment::new().await?;
let mut env_vars = replication_fast_env();
env_vars.extend_from_slice(LOOPBACK_REPLICATION_TARGET_ENV);
env_vars.extend_from_slice(&[("NO_PROXY", "127.0.0.1,localhost"), ("HTTP_PROXY", ""), ("HTTPS_PROXY", "")]);
source_env.start_rustfs_server_with_env(vec![], &env_vars).await?;
let source_bucket = "ssec-drop-src";
let source_client = source_env.create_s3_client();
source_client.create_bucket().bucket(source_bucket).send().await?;
enable_bucket_versioning(&source_env, source_bucket).await?;
let target_arn = set_replication_target_with_options(
&source_env,
source_bucket,
ReplicationTargetOptions {
endpoint: &target.address(),
access_key: FAKE_ACCESS_KEY,
secret_key: FAKE_SECRET_KEY,
target_bucket,
secure: false,
skip_tls_verify: false,
ca_cert_pem: None,
},
)
.await?;
put_bucket_replication(&source_env, source_bucket, &target_arn).await?;
let customer_key = BASE64_STANDARD.encode(REPL17_SSEC_KEY);
let customer_key_md5 = sse_customer_key_md5_base64(REPL17_SSEC_KEY);
let put_ssec = |key: &'static str| {
source_client
.put_object()
.bucket(source_bucket)
.key(key)
.body(ByteStream::from_static(b"ssec fail-closed payload"))
.sse_customer_algorithm("AES256")
.sse_customer_key(&customer_key)
.sse_customer_key_md5(&customer_key_md5)
.send()
};
// First SSE-C object: the audit must catch the dropped material.
put_ssec("ssec-first.txt").await?;
wait_for_source_replication_status(&source_client, source_bucket, "ssec-first.txt", "FAILED", true).await?;
let requests = target.take_requests();
let first_put = requests
.iter()
.find(|record| record.operation == FakeTargetOperation::PutObject && record.key.as_deref() == Some("ssec-first.txt"))
.ok_or("the first SSE-C object must have been PUT (capability was Unknown)")?;
assert!(
first_put.proxy_headers.ssec_transport_present,
"the replication PUT must have shipped the SSE-C transport headers the target then dropped"
);
assert!(
requests.iter().any(|record| {
record.operation == FakeTargetOperation::HeadObject
&& record.key.as_deref() == Some("ssec-first.txt")
&& record.sequence > first_put.sequence
&& record.proxy_headers.replication_check.as_deref() == Some("true")
}),
"the post-PUT HEAD-back audit must have run through the replication-check channel; journal: {requests:?}"
);
// Second SSE-C object: the cached Unsupported verdict fails it closed
// before any PUT — including MRF retries of the first object.
put_ssec("ssec-second.txt").await?;
wait_for_source_replication_status(&source_client, source_bucket, "ssec-second.txt", "FAILED", true).await?;
assert!(
!target.requests().iter().any(|record| {
record.operation == FakeTargetOperation::PutObject
&& record.key.as_deref() != Some("plain-control.txt")
&& record.proxy_headers.ssec_transport_present
}),
"no further SSE-C ciphertext may reach a target recorded Unsupported; journal: {:?}",
target.requests()
);
// The gate is scoped to SSE-C: plaintext replication keeps working.
source_client
.put_object()
.bucket(source_bucket)
.key("plain-control.txt")
.body(ByteStream::from_static(b"plaintext control payload"))
.send()
.await?;
wait_for_source_replication_status(&source_client, source_bucket, "plain-control.txt", "COMPLETED", false).await?;
assert!(target.has_object(target_bucket, "plain-control.txt"));
target.shutdown().await;
Ok(())
}
/// N2 (backlog#1675 P1-22): the admin replication-check must expose the same
/// verdict operators would otherwise only learn from failing SSE-C objects —
/// an SsecPassthrough probe phase that fails with the machine-readable
/// `BucketRemoteSsecPassthroughUnsupported` code against a header-dropping
/// target, with no probe residue left behind. The target's overall status
/// stays OK: unlike version-identity drift, dropped passthrough headers are
/// a capability limit, and a plaintext-only deployment against a MinIO-like
/// target must not turn red.
#[tokio::test]
#[serial]
async fn test_replication_check_flags_ssec_passthrough_dropping_target() -> TestResult {
init_logging();
let target = FakeS3Target::start().await?;
let target_bucket = "ssec-check-dst";
target.create_bucket(target_bucket);
target.drop_unlisted_replication_headers(true);
let mut source_env = RustFSTestEnvironment::new().await?;
let mut env_vars = replication_fast_env();
env_vars.extend_from_slice(LOOPBACK_REPLICATION_TARGET_ENV);
env_vars.extend_from_slice(&[("NO_PROXY", "127.0.0.1,localhost"), ("HTTP_PROXY", ""), ("HTTPS_PROXY", "")]);
source_env.start_rustfs_server_with_env(vec![], &env_vars).await?;
let source_bucket = "ssec-check-src";
let source_client = source_env.create_s3_client();
source_client.create_bucket().bucket(source_bucket).send().await?;
enable_bucket_versioning(&source_env, source_bucket).await?;
let target_arn = set_replication_target_with_options(
&source_env,
source_bucket,
ReplicationTargetOptions {
endpoint: &target.address(),
access_key: FAKE_ACCESS_KEY,
secret_key: FAKE_SECRET_KEY,
target_bucket,
secure: false,
skip_tls_verify: false,
ca_cert_pem: None,
},
)
.await?;
put_bucket_replication(&source_env, source_bucket, &target_arn).await?;
let response = run_replication_check(&source_env, source_bucket).await?;
assert_eq!(response.status(), StatusCode::OK);
let payload: serde_json::Value = response.json().await?;
assert_eq!(
payload["Status"], "OK",
"a capability-only SSE-C failure must not fail the check overall: {payload}"
);
let target_report = &payload["Targets"][0];
assert_eq!(target_report["Status"], "OK", "{payload}");
let ssec = &target_report["Phases"]["SsecPassthrough"];
assert_eq!(ssec["Status"], "FAILED", "SsecPassthrough phase must fail: {payload}");
assert_eq!(
ssec["Code"], "BucketRemoteSsecPassthroughUnsupported",
"the failure must carry the machine-readable code: {payload}"
);
// Basic replication of plaintext objects works on this target: every other
// phase passes, so the code is the discriminator operators branch on.
assert_eq!(target_report["Phases"]["Put"]["Status"], "OK", "{payload}");
assert_eq!(target_report["Phases"]["VersionFidelity"]["Status"], "OK", "{payload}");
assert_eq!(target_report["Phases"]["DeleteMarker"]["Status"], "OK", "{payload}");
assert_eq!(target_report["Phases"]["VersionDelete"]["Status"], "OK", "{payload}");
assert_eq!(target_report["Phases"]["Cleanup"]["Status"], "OK", "{payload}");
// The SSE-C probe PUT must have shipped the real transport header names —
// a mangled or missing header set would fail the phase for the wrong
// reason and mask a working target.
let requests = target.requests();
assert!(
requests
.iter()
.any(|record| record.operation == FakeTargetOperation::PutObject && record.proxy_headers.ssec_transport_present),
"the SSE-C probe PUT must carry the X-Rustfs-Replication-* transport headers; journal: {requests:?}"
);
// No probe residue, including the SSE-C probe version.
let probe_put = requests
.into_iter()
.find(|record| record.operation == FakeTargetOperation::PutObject)
.ok_or("the probe PUT never reached the fake target")?;
let probe_key = probe_put.key.ok_or("probe PUT journal record has no key")?;
assert!(
target.stored_versions(target_bucket, &probe_key).is_empty(),
"all probe versions must be cleaned up"
);
target.shutdown().await;
Ok(())
}
/// C1 (backlog#1675 P1-22): heal-path convergence for SSE-C. An SSE-C object
/// whose live replication failed during a target outage must converge through
/// the scanner/heal compensation once the target returns — passing the N2
/// HEAD-back audit against the recovered RustFS target — and the replica must
/// be readable with the customer key.
#[tokio::test]
#[serial]
async fn test_bucket_replication_sse_c_heals_after_target_outage() -> TestResult {
init_logging();
let (source_env, mut target_env, source_bucket, target_bucket) =
build_sse_replication_pair("ssec-heal", false, false).await?;
let source_client = source_env.create_s3_client();
let key = "ssec-heal-contract.txt";
let body = b"repl-22 ssec heal payload".to_vec();
let customer_key = BASE64_STANDARD.encode(REPL17_SSEC_KEY);
let customer_key_md5 = sse_customer_key_md5_base64(REPL17_SSEC_KEY);
// Target outage: the SSE-C write cannot replicate.
target_env.stop_server();
source_client
.put_object()
.bucket(&source_bucket)
.key(key)
.body(ByteStream::from(body.clone()))
.sse_customer_algorithm("AES256")
.sse_customer_key(&customer_key)
.sse_customer_key_md5(&customer_key_md5)
.send()
.await?;
// The failure is observable on the source (SSE-C HEAD needs the key).
let deadline = tokio::time::Instant::now() + Duration::from_secs(30);
loop {
let head = source_client
.head_object()
.bucket(&source_bucket)
.key(key)
.sse_customer_algorithm("AES256")
.sse_customer_key(&customer_key)
.sse_customer_key_md5(&customer_key_md5)
.send()
.await?;
match head.replication_status().map(|status| status.as_str()) {
Some("PENDING") | Some("FAILED") => break,
other => {
if tokio::time::Instant::now() >= deadline {
return Err(format!("source SSE-C object never reported PENDING/FAILED; last status={other:?}").into());
}
sleep(Duration::from_millis(200)).await;
}
}
}
// Recover the target in place; the source scanner re-drives the failure.
target_env
.restart_server_preserving_data(vec![], &[("NO_PROXY", "127.0.0.1,localhost"), ("HTTP_PROXY", ""), ("HTTPS_PROXY", "")])
.await?;
wait_for_source_replication_status(&source_client, &source_bucket, key, "COMPLETED", true).await?;
// The healed replica is a REPLICA (status surfaces on HEAD) readable with
// the customer key.
let target_client = target_env.create_s3_client();
let replica_head = target_client
.head_object()
.bucket(&target_bucket)
.key(key)
.sse_customer_algorithm("AES256")
.sse_customer_key(&customer_key)
.sse_customer_key_md5(&customer_key_md5)
.send()
.await?;
assert_eq!(
replica_head.replication_status().map(|status| status.as_str()),
Some("REPLICA"),
"the healed copy must carry REPLICA status"
);
let replica = target_client
.get_object()
.bucket(&target_bucket)
.key(key)
.sse_customer_algorithm("AES256")
.sse_customer_key(&customer_key)
.sse_customer_key_md5(&customer_key_md5)
.send()
.await?;
assert_eq!(replica.sse_customer_algorithm(), Some("AES256"));
assert_eq!(replica.body.collect().await?.into_bytes().as_ref(), body.as_slice());
Ok(())
}
/// C1 (backlog#1675 P1-22): existing-object resync for SSE-C. An SSE-C object
/// written BEFORE any replication config must reach the RustFS target through
/// the existing-object resync (`replicate_all` transport, N2-audited), land as
/// a REPLICA, and read back with the customer key.
#[tokio::test]
#[serial]
async fn test_bucket_replication_sse_c_existing_object_resync() -> TestResult {
init_logging();
let mut source_env = RustFSTestEnvironment::new().await?;
let mut source_process_env = replication_fast_env();
source_process_env.extend_from_slice(LOOPBACK_REPLICATION_TARGET_ENV);
source_process_env.extend_from_slice(FAST_SCANNER_ENV);
source_process_env.extend_from_slice(&[("NO_PROXY", "127.0.0.1,localhost"), ("HTTP_PROXY", ""), ("HTTPS_PROXY", "")]);
source_env.start_rustfs_server_with_env(vec![], &source_process_env).await?;
let mut target_env = RustFSTestEnvironment::new().await?;
target_env
.start_rustfs_server_without_cleanup_with_env(&[
("NO_PROXY", "127.0.0.1,localhost"),
("HTTP_PROXY", ""),
("HTTPS_PROXY", ""),
])
.await?;
let source_bucket = "ssec-existing-src";
let target_bucket = "ssec-existing-dst";
let source_client = source_env.create_s3_client();
let target_client = target_env.create_s3_client();
source_client.create_bucket().bucket(source_bucket).send().await?;
target_client.create_bucket().bucket(target_bucket).send().await?;
enable_bucket_versioning(&source_env, source_bucket).await?;
enable_bucket_versioning(&target_env, target_bucket).await?;
// The SSE-C object exists before any replication wiring.
let key = "ssec-existing-contract.txt";
let body = b"repl-22 ssec existing-object payload".to_vec();
let customer_key = BASE64_STANDARD.encode(REPL17_SSEC_KEY);
let customer_key_md5 = sse_customer_key_md5_base64(REPL17_SSEC_KEY);
source_client
.put_object()
.bucket(source_bucket)
.key(key)
.body(ByteStream::from(body.clone()))
.sse_customer_algorithm("AES256")
.sse_customer_key(&customer_key)
.sse_customer_key_md5(&customer_key_md5)
.send()
.await?;
// Wire replication (existing-object enabled) and drive a resync.
let target_arn = set_replication_target(&source_env, source_bucket, &target_env, target_bucket).await?;
put_bucket_replication(&source_env, source_bucket, &target_arn).await?;
let (reset_arn, reset_id) = start_bucket_replication_reset(&source_env, source_bucket).await?;
assert_eq!(reset_arn, target_arn);
let terminal = wait_for_replication_reset_target(&source_env, source_bucket, &target_arn, |status| {
status.reset_id == reset_id && matches!(status.status.as_str(), "Completed" | "Failed")
})
.await?;
assert_eq!(terminal.status, "Completed", "SSE-C existing-object resync must complete");
assert!(terminal.replicated_count >= 1, "the existing SSE-C object must have been resynced");
// The replica is a REPLICA (status surfaces on HEAD) readable with the
// customer key.
let replica_head = target_client
.head_object()
.bucket(target_bucket)
.key(key)
.sse_customer_algorithm("AES256")
.sse_customer_key(&customer_key)
.sse_customer_key_md5(&customer_key_md5)
.send()
.await?;
assert_eq!(
replica_head.replication_status().map(|status| status.as_str()),
Some("REPLICA"),
"the resynced copy must carry REPLICA status"
);
let replica = target_client
.get_object()
.bucket(target_bucket)
.key(key)
.sse_customer_algorithm("AES256")
.sse_customer_key(&customer_key)
.sse_customer_key_md5(&customer_key_md5)
.send()
.await?;
assert_eq!(replica.sse_customer_algorithm(), Some("AES256"));
assert_eq!(replica.body.collect().await?.into_bytes().as_ref(), body.as_slice());
// No plaintext leak: the replica stays unreadable without the key.
assert!(
target_client
.get_object()
.bucket(target_bucket)
.key(key)
.send()
.await
.is_err(),
"SSE-C replica must not be readable without the customer key"
);
Ok(())
}
/// backlog#1147 repl-17 / backlog#1783: SSE-S3 objects replicate by decrypting
/// at the source and re-encrypting on the target with the target's own KMS.
/// The property backlog#1291 pinned — never a silent plaintext replica — still
@@ -8824,304 +8417,3 @@ async fn test_scanner_never_compensates_when_existing_object_replication_disable
Ok(())
}
/// Shared setup for the P1-5 read-proxy scenarios (backlog#1675): a RustFS
/// source with an enabled replication rule pointing at the fake target, and
/// an object seeded DIRECTLY on the target — it exists remotely but not
/// locally, exactly the active-active replication-lag window the read proxy
/// serves.
async fn start_read_proxy_lab(
source_bucket: &str,
target_bucket: &str,
) -> Result<(FakeS3Target, RustFSTestEnvironment, Client, Client), Box<dyn Error + Send + Sync>> {
let target = FakeS3Target::start().await?;
target.create_bucket(target_bucket);
target.assign_own_version_ids(true);
let mut source_env = RustFSTestEnvironment::new().await?;
let mut process_env = replication_fast_env();
process_env.extend_from_slice(LOOPBACK_REPLICATION_TARGET_ENV);
process_env.extend_from_slice(&[
("NO_PROXY", "127.0.0.1,localhost"),
("HTTP_PROXY", ""),
("HTTPS_PROXY", ""),
("RUST_LOG", "error"),
]);
source_env.start_rustfs_server_with_env(vec![], &process_env).await?;
let source_client = source_env.create_s3_client();
source_client.create_bucket().bucket(source_bucket).send().await?;
enable_bucket_versioning(&source_env, source_bucket).await?;
let target_arn = set_replication_target_with_options(
&source_env,
source_bucket,
ReplicationTargetOptions {
endpoint: &target.address(),
access_key: FAKE_ACCESS_KEY,
secret_key: FAKE_SECRET_KEY,
target_bucket,
secure: false,
skip_tls_verify: false,
ca_cert_pem: None,
},
)
.await?;
put_bucket_replication(&source_env, source_bucket, &target_arn).await?;
let target_client = Client::from_conf(crate::common::build_test_s3_config(
target.endpoint(),
FAKE_ACCESS_KEY,
FAKE_SECRET_KEY,
None,
"read-proxy-e2e",
));
Ok((target, source_env, source_client, target_client))
}
/// P1-5 (backlog#1675): during the active-active replication lag window a
/// GET/HEAD for an object the local site does not have yet is proxied to the
/// replication target. Pins the wire contract: the anti-loop
/// `source-proxy-request` marker is sent, the replication worker's
/// `source-replication-check` SSE-C exemption is NEVER sent, client SSE-C
/// headers are forwarded verbatim, and an inbound request that was itself
/// proxied is answered locally (404) without touching the target.
#[tokio::test]
#[serial]
async fn test_get_and_head_proxy_unreplicated_object_to_replication_target() -> TestResult {
init_logging();
let source_bucket = "proxy-read-src";
let target_bucket = "proxy-read-dst";
let (target, source_env, source_client, target_client) = start_read_proxy_lab(source_bucket, target_bucket).await?;
let payload = b"proxy payload".to_vec();
target_client
.put_object()
.bucket(target_bucket)
.key("proxy-only")
.body(ByteStream::from(payload.clone()))
.send()
.await?;
target.take_requests();
// a. GET of the locally-missing object is served through the proxy.
let got = source_client
.get_object()
.bucket(source_bucket)
.key("proxy-only")
.send()
.await
.map_err(|err| format!("proxied GET failed: {}", err.into_service_error()))?;
assert_eq!(got.content_length, Some(payload.len() as i64));
let body = got.body.collect().await?.into_bytes();
assert_eq!(body.as_ref(), payload.as_slice(), "proxied GET must stream the target's body");
let get_record = target
.requests()
.into_iter()
.find(|record| record.operation == FakeTargetOperation::GetObject && record.key.as_deref() == Some("proxy-only"))
.ok_or("fake target never received the proxied GET")?;
assert_eq!(
get_record.proxy_headers.source_proxy_request.as_deref(),
Some("true"),
"proxied GET must carry the anti-loop source-proxy-request marker"
);
assert!(
get_record.proxy_headers.replication_check.is_none(),
"proxied GET must never carry the replication worker's source-replication-check exemption"
);
assert!(
get_record.proxy_headers.ssec_algorithm.is_none() && !get_record.proxy_headers.ssec_key_present,
"no client SSE-C headers were sent, so none may be forwarded"
);
// a2. Client SSE-C headers travel verbatim to the target (the target owns
// the real SSE-C decryption; the plaintext fake simply ignores them).
target.take_requests();
let ssec_key = "01234567890123456789012345678901";
let ssec_key_b64 = BASE64_STANDARD.encode(ssec_key);
let ssec_key_md5 = sse_customer_key_md5_base64(ssec_key);
let _ = source_client
.get_object()
.bucket(source_bucket)
.key("proxy-only")
.sse_customer_algorithm("AES256")
.sse_customer_key(&ssec_key_b64)
.sse_customer_key_md5(&ssec_key_md5)
.send()
.await
.map_err(|err| format!("proxied SSE-C GET failed: {}", err.into_service_error()))?;
let ssec_record = target
.requests()
.into_iter()
.find(|record| record.operation == FakeTargetOperation::GetObject && record.key.as_deref() == Some("proxy-only"))
.ok_or("fake target never received the proxied SSE-C GET")?;
assert_eq!(ssec_record.proxy_headers.ssec_algorithm.as_deref(), Some("AES256"));
assert!(ssec_record.proxy_headers.ssec_key_present, "SSE-C key header must be forwarded verbatim");
assert_eq!(ssec_record.proxy_headers.ssec_key_md5.as_deref(), Some(ssec_key_md5.as_str()));
assert!(ssec_record.proxy_headers.replication_check.is_none());
// b. HEAD of the locally-missing object is served through the proxy.
target.take_requests();
let head = source_client
.head_object()
.bucket(source_bucket)
.key("proxy-only")
.send()
.await
.map_err(|err| format!("proxied HEAD failed: {}", err.into_service_error()))?;
assert_eq!(head.content_length, Some(payload.len() as i64));
let head_record = target
.requests()
.into_iter()
.find(|record| record.operation == FakeTargetOperation::HeadObject && record.key.as_deref() == Some("proxy-only"))
.ok_or("fake target never received the proxied HEAD")?;
assert_eq!(head_record.proxy_headers.source_proxy_request.as_deref(), Some("true"));
assert!(head_record.proxy_headers.replication_check.is_none());
// c. Anti-loop: an inbound request that already carries the proxy marker
// is answered locally with 404 and never forwarded to the target.
target.take_requests();
let err = source_client
.get_object()
.bucket(source_bucket)
.key("proxy-only")
.customize()
.mutate_request(|req| {
req.headers_mut().insert("x-minio-source-proxy-request", "true");
})
.send()
.await
.expect_err("anti-loop GET must fail locally instead of proxying");
let service_err = err.into_service_error();
assert!(service_err.is_no_such_key(), "anti-loop GET must 404, got: {service_err}");
assert!(
!target
.requests()
.iter()
.any(|record| record.operation == FakeTargetOperation::GetObject),
"anti-loop GET must not reach the replication target; journal: {:?}",
target.requests()
);
// c2. MinIO ProxyHeaderSet parity: the header's mere PRESENCE disables
// proxying — "false" is exactly what a peer's replication worker sends on
// its convergence HEADs, and proxying that miss back would fake
// convergence.
target.take_requests();
let err = source_client
.get_object()
.bucket(source_bucket)
.key("proxy-only")
.customize()
.mutate_request(|req| {
req.headers_mut().insert("x-minio-source-proxy-request", "false");
})
.send()
.await
.expect_err("proxy-header-set GET must fail locally instead of proxying");
let service_err = err.into_service_error();
assert!(service_err.is_no_such_key(), "proxy-header-set GET must 404, got: {service_err}");
assert!(
!target
.requests()
.iter()
.any(|record| record.operation == FakeTargetOperation::GetObject),
"proxy-header-set GET must not reach the replication target; journal: {:?}",
target.requests()
);
// d. The replication worker's own convergence HEAD against the target
// must carry `source-proxy-request: false` (never proxied back) and the
// replication-check exemption. Trigger real replication and inspect the
// fake journal.
target.take_requests();
source_client
.put_object()
.bucket(source_bucket)
.key("worker-replicated")
.body(ByteStream::from_static(b"worker payload"))
.send()
.await?;
wait_for_target_request_version_id(&target, FakeTargetOperation::PutObject, "worker-replicated").await?;
let worker_head = target
.requests()
.into_iter()
.find(|record| record.operation == FakeTargetOperation::HeadObject && record.key.as_deref() == Some("worker-replicated"))
.ok_or_else(|| format!("replication worker never HEAD-ed the target; journal: {:?}", target.requests()))?;
assert_eq!(
worker_head.proxy_headers.source_proxy_request.as_deref(),
Some("false"),
"worker convergence HEAD must send source-proxy-request: false so the target answers locally"
);
assert_eq!(
worker_head.proxy_headers.replication_check.as_deref(),
Some("true"),
"worker convergence HEAD keeps the replication-check exemption"
);
drop(source_env);
target.shutdown().await;
Ok(())
}
/// P1-5 (backlog#1675): GetObjectTagging for an object missing locally is
/// proxied to the replication target with the anti-loop marker, mirroring
/// MinIO `proxyGetTaggingToRepTarget`.
#[tokio::test]
#[serial]
async fn test_get_object_tagging_proxies_unreplicated_object_to_replication_target() -> TestResult {
init_logging();
let source_bucket = "proxy-tag-src";
let target_bucket = "proxy-tag-dst";
let (target, source_env, source_client, target_client) = start_read_proxy_lab(source_bucket, target_bucket).await?;
target_client
.put_object()
.bucket(target_bucket)
.key("proxy-tagged")
.body(ByteStream::from_static(b"tagged payload"))
.send()
.await?;
target_client
.put_object_tagging()
.bucket(target_bucket)
.key("proxy-tagged")
.tagging(
aws_sdk_s3::types::Tagging::builder()
.tag_set(aws_sdk_s3::types::Tag::builder().key("team").value("storage").build()?)
.build()?,
)
.send()
.await?;
target.take_requests();
let tags = source_client
.get_object_tagging()
.bucket(source_bucket)
.key("proxy-tagged")
.send()
.await
.map_err(|err| format!("proxied GetObjectTagging failed: {}", err.into_service_error()))?;
assert_eq!(tags.tag_set.len(), 1, "proxied tagging read must return the target's tags");
assert_eq!(tags.tag_set[0].key.as_str(), "team");
assert_eq!(tags.tag_set[0].value.as_str(), "storage");
let record = target
.requests()
.into_iter()
.find(|record| record.operation == FakeTargetOperation::GetObjectTagging && record.key.as_deref() == Some("proxy-tagged"))
.ok_or("fake target never received the proxied GetObjectTagging")?;
assert_eq!(
record.proxy_headers.source_proxy_request.as_deref(),
Some("true"),
"proxied tagging read must carry the anti-loop marker"
);
assert!(record.proxy_headers.replication_check.is_none());
drop(source_env);
target.shutdown().await;
Ok(())
}
+10 -11
View File
@@ -32,7 +32,7 @@ pub mod bucket {
pub mod bucket_target_sys {
pub use crate::bucket::bucket_target_sys::{
AdvancedPutOptions, BucketTargetError, BucketTargetSys, PutObjectOptions, RemoveObjectOptions, S3ClientError,
SsecPassthroughCapability, TargetClient, append_version_id_query,
TargetClient, append_version_id_query,
};
}
@@ -198,13 +198,12 @@ pub mod bucket {
ReplicationType, ResyncOpts, ResyncStatusType, RuntimeReplicationTargetBacklog, TargetReplicationResyncStatus,
VersionPurgeStatusType, XferStats, commit_force_delete_intent, complete_force_delete_intent,
delete_replication_state_from_config, delete_replication_version_id, get_global_replication_pool,
get_global_replication_stats, get_proxy_targets, init_background_replication,
invalid_replication_config_status_field, persist_force_delete_intent, read_durable_mrf_backlog,
replication_state_to_filemeta, replication_status_to_filemeta, replication_statuses_map, replication_target_arns,
resync_start_conflict_id, should_remove_replication_target, should_schedule_delete_replication,
should_use_existing_delete_replication_info, should_use_existing_delete_replication_source,
unsupported_replication_config_field, validate_replication_config_structure, validate_replication_config_target_arns,
version_purge_status_to_filemeta,
get_global_replication_stats, init_background_replication, invalid_replication_config_status_field,
persist_force_delete_intent, read_durable_mrf_backlog, replication_state_to_filemeta, replication_status_to_filemeta,
replication_statuses_map, replication_target_arns, resync_start_conflict_id, should_remove_replication_target,
should_schedule_delete_replication, should_use_existing_delete_replication_info,
should_use_existing_delete_replication_source, unsupported_replication_config_field,
validate_replication_config_structure, validate_replication_config_target_arns, version_purge_status_to_filemeta,
};
}
@@ -374,14 +373,14 @@ pub mod error {
pub mod erasure {
pub use crate::erasure::coding::{
BitrotReader, BitrotWriter, BitrotWriterWrapper, CustomWriter, Erasure, ErasureConstructionError, ReedSolomonEncoder,
calc_shard_size, calc_shard_size_legacy,
BitrotReader, BitrotSelfTestError, BitrotWriter, BitrotWriterWrapper, CustomWriter, Erasure, ErasureConstructionError,
ReedSolomonEncoder, bitrot_self_test, calc_shard_size, calc_shard_size_legacy,
};
}
pub mod event {
pub use crate::event::name::EventName;
pub use crate::services::event_notification::{EventArgs, register_event_dispatch_hook};
pub use crate::services::event_notification::{EventArgs, register_event_dispatch_hook, send_event};
}
pub mod global {
+4 -316
View File
@@ -27,15 +27,10 @@ use aws_sdk_s3::config::SharedHttpClient;
use aws_sdk_s3::error::ProvideErrorMetadata;
use aws_sdk_s3::error::SdkError;
use aws_sdk_s3::operation::complete_multipart_upload::CompleteMultipartUploadOutput;
use aws_sdk_s3::operation::delete_object_tagging::{DeleteObjectTaggingError, DeleteObjectTaggingOutput};
use aws_sdk_s3::operation::get_object::{GetObjectError, GetObjectOutput};
use aws_sdk_s3::operation::get_object_tagging::{GetObjectTaggingError, GetObjectTaggingOutput};
use aws_sdk_s3::operation::head_bucket::HeadBucketError;
use aws_sdk_s3::operation::head_object::HeadObjectError;
use aws_sdk_s3::operation::put_object_tagging::{PutObjectTaggingError, PutObjectTaggingOutput};
use aws_sdk_s3::operation::upload_part::UploadPartOutput;
use aws_sdk_s3::primitives::ByteStream;
use aws_sdk_s3::types::Tagging as SdkTagging;
use aws_sdk_s3::types::{
ChecksumMode, CompletedMultipartUpload, CompletedPart, ObjectLockLegalHoldStatus, ObjectLockRetentionMode,
};
@@ -62,8 +57,8 @@ use rustfs_utils::http::{
is_rustfs_header, is_standard_header, is_storageclass_header,
};
use rustfs_utils::http::{
SUFFIX_FORCE_DELETE, SUFFIX_SOURCE_DELETEMARKER, SUFFIX_SOURCE_ETAG, SUFFIX_SOURCE_MTIME, SUFFIX_SOURCE_PROXY_REQUEST,
SUFFIX_SOURCE_REPLICATION_CHECK, SUFFIX_SOURCE_REPLICATION_LEGALHOLD_TIMESTAMP, SUFFIX_SOURCE_REPLICATION_REQUEST,
SUFFIX_FORCE_DELETE, SUFFIX_SOURCE_DELETEMARKER, SUFFIX_SOURCE_ETAG, SUFFIX_SOURCE_MTIME, SUFFIX_SOURCE_REPLICATION_CHECK,
SUFFIX_SOURCE_REPLICATION_LEGALHOLD_TIMESTAMP, SUFFIX_SOURCE_REPLICATION_REQUEST,
SUFFIX_SOURCE_REPLICATION_RETENTION_TIMESTAMP, SUFFIX_SOURCE_REPLICATION_TAGGING_TIMESTAMP, SUFFIX_SOURCE_VERSION_ID,
insert_header,
};
@@ -299,51 +294,9 @@ struct TargetClientBuildProbe {
release: Arc<tokio::sync::Semaphore>,
}
/// Whether a replication target preserves the SSE-C passthrough transport
/// headers (`X-Rustfs-Replication-*`) end to end.
///
/// A target that silently drops those headers (MinIO, generic S3) stores the
/// forwarded ciphertext without its decryption material — an unreadable
/// replica that used to report COMPLETED. The replication worker audits the
/// first passthrough PUT per target (HEAD-back for SSE-C evidence) and caches
/// the verdict here; a fresh `Unsupported` fails SSE-C replication closed
/// before any PUT is sent. Entries follow the `arn_remotes_map` lifecycle
/// (rebuilding or removing a target resets its capability to `Unknown`) and
/// additionally expire after [`SSEC_PASSTHROUGH_CAPABILITY_TTL`], after which
/// the next attempt re-audits.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum SsecPassthroughCapability {
#[default]
Unknown,
Supported,
Unsupported,
}
/// How long an audited SSE-C passthrough verdict stays authoritative.
///
/// Trade-off: without a TTL a verdict is sticky for the process lifetime —
/// an `Unsupported` target that gets upgraded (or re-probed only via
/// replication-check) would keep failing SSE-C replication forever, and the
/// fail-open twin: a `Supported` verdict would outlive a backend swapped
/// behind the same endpoint/ARN. With the TTL, a bad target costs at most
/// one wasted PUT+HEAD audit per TTL window, and a changed backend is
/// re-discovered within the same window.
pub const SSEC_PASSTHROUGH_CAPABILITY_TTL: Duration = Duration::from_secs(10 * 60);
/// A recorded SSE-C passthrough verdict plus when it was recorded, so reads
/// can report staleness against [`SSEC_PASSTHROUGH_CAPABILITY_TTL`].
#[derive(Debug, Clone, Copy)]
struct SsecPassthroughRecord {
capability: SsecPassthroughCapability,
recorded_at: Instant,
}
#[derive(Debug, Default)]
pub struct BucketTargetSys {
pub arn_remotes_map: Arc<RwLock<HashMap<String, ArnTarget>>>,
/// SSE-C passthrough capability verdicts keyed by target ARN. See
/// [`SsecPassthroughCapability`]; reset alongside `arn_remotes_map`.
ssec_passthrough_map: Arc<RwLock<HashMap<String, SsecPassthroughRecord>>>,
pub targets_map: Arc<RwLock<HashMap<String, Vec<BucketTarget>>>>,
pub h_mutex: Arc<RwLock<HashMap<String, EpHealth>>>,
target_h_mutex: Arc<RwLock<HashMap<String, EpHealth>>>,
@@ -364,7 +317,6 @@ impl BucketTargetSys {
fn new() -> Self {
Self {
arn_remotes_map: Arc::new(RwLock::new(HashMap::new())),
ssec_passthrough_map: Arc::new(RwLock::new(HashMap::new())),
targets_map: Arc::new(RwLock::new(HashMap::new())),
h_mutex: Arc::new(RwLock::new(HashMap::new())),
target_h_mutex: Arc::new(RwLock::new(HashMap::new())),
@@ -628,59 +580,19 @@ impl BucketTargetSys {
let update_mutex = self.target_update_mutex(bucket).await;
let _update_guard = update_mutex.lock().await;
// Lock order: targets_map, then arn_remotes_map, then target_h_mutex,
// then ssec_passthrough_map (always last; also taken standalone by the
// capability accessors).
// Lock order: targets_map, then arn_remotes_map, then target_h_mutex.
let mut targets_map = self.targets_map.write().await;
let mut arn_remotes_map = self.arn_remotes_map.write().await;
let mut health_map = self.target_h_mutex.write().await;
if let Some(targets) = targets_map.remove(bucket) {
let mut ssec_map = self.ssec_passthrough_map.write().await;
for target in targets {
arn_remotes_map.remove(&target.arn);
health_map.remove(&target.arn);
ssec_map.remove(&target.arn);
}
}
}
/// Cached SSE-C passthrough capability for a target ARN, plus whether the
/// verdict is older than [`SSEC_PASSTHROUGH_CAPABILITY_TTL`]. `(Unknown,
/// false)` when no verdict has been recorded since the target was built.
/// Staleness is computed here so the gate policy stays a pure function.
pub async fn ssec_passthrough_capability(&self, arn: &str) -> (SsecPassthroughCapability, bool) {
match self.ssec_passthrough_map.read().await.get(arn) {
Some(record) => (record.capability, record.recorded_at.elapsed() >= SSEC_PASSTHROUGH_CAPABILITY_TTL),
None => (SsecPassthroughCapability::Unknown, false),
}
}
/// Record an audited SSE-C passthrough verdict for a target ARN. Written by
/// the replication worker's HEAD-back audit and by the replication-check
/// SsecPassthrough probe phase.
pub async fn record_ssec_passthrough_capability(&self, arn: &str, capability: SsecPassthroughCapability) {
self.ssec_passthrough_map.write().await.insert(
arn.to_string(),
SsecPassthroughRecord {
capability,
recorded_at: Instant::now(),
},
);
}
/// Test hook: age an existing verdict so TTL expiry is observable without
/// waiting out the real window.
#[cfg(test)]
pub(crate) async fn backdate_ssec_passthrough_capability(&self, arn: &str, age: Duration) {
let backdated = Instant::now()
.checked_sub(age)
.expect("system uptime must exceed the backdate age");
if let Some(record) = self.ssec_passthrough_map.write().await.get_mut(arn) {
record.recorded_at = backdated;
}
}
pub async fn set_target(
&self,
bucket: &str,
@@ -1036,21 +948,15 @@ impl BucketTargetSys {
}
}
// Lock order: targets_map, then arn_remotes_map, then target_h_mutex,
// then ssec_passthrough_map (always last; also taken standalone by the
// capability accessors).
// Lock order: targets_map, then arn_remotes_map, then target_h_mutex.
let mut targets_map = self.targets_map.write().await;
let mut arn_remotes_map = self.arn_remotes_map.write().await;
let mut health_map = self.target_h_mutex.write().await;
// Remove existing targets
if let Some(existing_targets) = targets_map.remove(bucket) {
let mut ssec_map = self.ssec_passthrough_map.write().await;
for target in existing_targets {
arn_remotes_map.remove(&target.arn);
health_map.remove(&target.arn);
// A rebuilt/edited target may point at a different service:
// the SSE-C passthrough verdict must be re-audited from Unknown.
ssec_map.remove(&target.arn);
self.update_bandwidth_limit(bucket, &target.arn, 0);
}
}
@@ -1540,43 +1446,6 @@ fn resolve_put_api_version_id(source_version_id: &str) -> Option<&str> {
}
}
/// Resolve the S3 `versionId` for a proxied read against a remote target.
/// RustFS represents the null version internally as the nil UUID while the S3
/// API addresses it as the literal "null" (same mapping as
/// [`resolve_put_api_version_id`]); empty means "no version requested".
pub(crate) fn resolve_read_api_version_id(version_id: Option<String>) -> Option<String> {
let version_id = version_id?;
let trimmed = version_id.trim();
if trimmed.is_empty() {
None
} else if Uuid::parse_str(trimmed).is_ok_and(|uuid| uuid.is_nil()) {
Some(rustfs_filemeta::NULL_VERSION_ID.to_string())
} else {
Some(trimmed.to_string())
}
}
/// Outbound header set for a proxied read: the caller-provided passthrough
/// headers (client SSE-C key family, conditional headers) plus the anti-loop
/// `source-proxy-request` marker in both the x-rustfs- and x-minio- prefixes
/// (a MinIO target only understands the latter). Never adds
/// `source-replication-check`: that exemption channel belongs exclusively to
/// the replication worker's HEAD.
fn proxy_outbound_headers(mut extra_headers: HeaderMap) -> HeaderMap {
insert_header(&mut extra_headers, SUFFIX_SOURCE_PROXY_REQUEST, "true");
extra_headers
}
/// Copy `headers` onto an SDK request inside `customize().map_request` (runs
/// before signing, so the headers join the SigV4 canonical request).
fn apply_extra_headers(mut req: HttpRequest, headers: &HeaderMap) -> Result<HttpRequest, std::convert::Infallible> {
for (k, v) in headers.iter() {
req.headers_mut()
.insert(k.as_str().to_string(), v.to_str().unwrap_or("").to_string());
}
Ok(req)
}
/// Append `versionId=<id>` to an already-built request URI. aws-sdk-s3's
/// `PutObjectInput` / `CreateMultipartUploadInput` expose no version id
/// member, so the query is spliced in via `map_request`, which runs at
@@ -1982,13 +1851,6 @@ impl TargetClient {
// worker cannot hold; otherwise SSE-C replicas never converge on HEAD.
let mut headers = HeaderMap::new();
insert_header(&mut headers, SUFFIX_SOURCE_REPLICATION_CHECK, "true");
// `source-proxy-request: false` (MinIO `ProxyHeaderSet` semantics):
// the header's mere presence tells the receiver to answer LOCALLY
// instead of proxying the miss back to us. Without it, a not-found on
// the target gets read-proxied back to this source, echoes the source
// object with an identical ETag, and the worker concludes the object
// already converged — so it never actually replicates it.
insert_header(&mut headers, SUFFIX_SOURCE_PROXY_REQUEST, "false");
match self
.client
.head_object()
@@ -2013,129 +1875,6 @@ impl TargetClient {
}
}
/// HEAD used by the read-proxy path (GET/HEAD of an object not yet
/// replicated locally, MinIO `proxyHeadToRepTarget`).
///
/// Deliberately different from [`TargetClient::head_object`]: it must NOT
/// send `source-replication-check` — that header is the replication
/// worker's SSE-C metadata exemption channel. A proxied client request
/// instead forwards the client's own SSE-C headers (`extra_headers`) so
/// the target performs the real SSE-C validation/decryption. The
/// `source-proxy-request` marker is always added so the target does not
/// proxy the request onward (anti-loop).
pub async fn head_object_for_proxy(
&self,
bucket: &str,
object: &str,
version_id: Option<String>,
range: Option<String>,
part_number: Option<i32>,
extra_headers: HeaderMap,
) -> Result<HeadObjectOutput, SdkError<HeadObjectError>> {
let headers = proxy_outbound_headers(extra_headers);
self.client
.head_object()
.bucket(bucket)
.key(object)
.set_version_id(resolve_read_api_version_id(version_id))
.set_range(range)
.set_part_number(part_number)
.customize()
.map_request(move |req| apply_extra_headers(req, &headers))
.send()
.await
}
/// GET used by the read-proxy path (MinIO `proxyGetToReplicationTarget`).
/// Returns the streaming SDK output; callers must forward the body without
/// buffering it. Same header contract as [`Self::head_object_for_proxy`]:
/// anti-loop marker on, replication-check never sent, client SSE-C /
/// conditional headers forwarded verbatim via `extra_headers`.
pub async fn get_object(
&self,
bucket: &str,
object: &str,
version_id: Option<String>,
range: Option<String>,
part_number: Option<i32>,
extra_headers: HeaderMap,
) -> Result<GetObjectOutput, SdkError<GetObjectError>> {
let headers = proxy_outbound_headers(extra_headers);
self.client
.get_object()
.bucket(bucket)
.key(object)
.set_version_id(resolve_read_api_version_id(version_id))
.set_range(range)
.set_part_number(part_number)
.customize()
.map_request(move |req| apply_extra_headers(req, &headers))
.send()
.await
}
/// GetObjectTagging for the tagging read-proxy path
/// (MinIO `proxyGetTaggingToRepTarget`). Anti-loop marker always added.
pub async fn get_object_tagging(
&self,
bucket: &str,
object: &str,
version_id: Option<String>,
) -> Result<GetObjectTaggingOutput, SdkError<GetObjectTaggingError>> {
let headers = proxy_outbound_headers(HeaderMap::new());
self.client
.get_object_tagging()
.bucket(bucket)
.key(object)
.set_version_id(resolve_read_api_version_id(version_id))
.customize()
.map_request(move |req| apply_extra_headers(req, &headers))
.send()
.await
}
/// PutObjectTagging for the tagging proxy path
/// (MinIO `proxyTaggingToRepTarget`). Anti-loop marker always added.
pub async fn put_object_tagging(
&self,
bucket: &str,
object: &str,
version_id: Option<String>,
tagging: SdkTagging,
) -> Result<PutObjectTaggingOutput, SdkError<PutObjectTaggingError>> {
let headers = proxy_outbound_headers(HeaderMap::new());
self.client
.put_object_tagging()
.bucket(bucket)
.key(object)
.set_version_id(resolve_read_api_version_id(version_id))
.tagging(tagging)
.customize()
.map_request(move |req| apply_extra_headers(req, &headers))
.send()
.await
}
/// DeleteObjectTagging for the tagging proxy path
/// (MinIO `proxyTaggingToRepTarget`). Anti-loop marker always added.
pub async fn delete_object_tagging(
&self,
bucket: &str,
object: &str,
version_id: Option<String>,
) -> Result<DeleteObjectTaggingOutput, SdkError<DeleteObjectTaggingError>> {
let headers = proxy_outbound_headers(HeaderMap::new());
self.client
.delete_object_tagging()
.bucket(bucket)
.key(object)
.set_version_id(resolve_read_api_version_id(version_id))
.customize()
.map_request(move |req| apply_extra_headers(req, &headers))
.send()
.await
}
/// On success returns the version id the target assigned (from
/// `x-amz-version-id`), letting callers audit the version-identity
/// contract — a target that adopts the source version echoes it back.
@@ -2765,57 +2504,6 @@ mod tests {
assert_eq!(health.last_online, Some(now));
}
/// N2 TTL contract, both flip directions: a recorded verdict is fresh
/// until [`SSEC_PASSTHROUGH_CAPABILITY_TTL`], then reads as expired; a
/// re-audit that records the OPPOSITE verdict replaces it as fresh. The
/// worker gate maps expired verdicts to ProceedWithAudit (pinned in
/// `replication_target_boundary`), so together this proves an Unsupported
/// target recovers to Supported through the audit once its verdict ages
/// out — and a stale Supported one is re-proven rather than trusted.
#[tokio::test]
async fn ssec_passthrough_capability_ttl_expires_and_reaudit_flips_verdict() {
let sys = BucketTargetSys::default();
let arn = "arn:rustfs:replication:us-east-1:bucket:ssec-ttl";
let expired_age = SSEC_PASSTHROUGH_CAPABILITY_TTL + Duration::from_secs(1);
assert_eq!(
sys.ssec_passthrough_capability(arn).await,
(SsecPassthroughCapability::Unknown, false),
"an unrecorded target must read Unknown and never expired"
);
sys.record_ssec_passthrough_capability(arn, SsecPassthroughCapability::Unsupported)
.await;
assert_eq!(
sys.ssec_passthrough_capability(arn).await,
(SsecPassthroughCapability::Unsupported, false)
);
sys.backdate_ssec_passthrough_capability(arn, expired_age).await;
assert_eq!(
sys.ssec_passthrough_capability(arn).await,
(SsecPassthroughCapability::Unsupported, true),
"an aged-out Unsupported verdict must read expired so the gate re-audits"
);
// The re-audit against an upgraded target records Supported afresh.
sys.record_ssec_passthrough_capability(arn, SsecPassthroughCapability::Supported)
.await;
assert_eq!(
sys.ssec_passthrough_capability(arn).await,
(SsecPassthroughCapability::Supported, false),
"a fresh Supported verdict replaces the expired Unsupported one"
);
// And the fail-open twin: Supported also ages out.
sys.backdate_ssec_passthrough_capability(arn, expired_age).await;
assert_eq!(
sys.ssec_passthrough_capability(arn).await,
(SsecPassthroughCapability::Supported, true),
"an aged-out Supported verdict must read expired so the gate re-proves it"
);
}
#[tokio::test]
async fn list_targets_applies_health_stats_by_arn_and_preserves_endpoint_port() {
let sys = BucketTargetSys::default();
@@ -14,7 +14,6 @@ paths.
| `datatypes.rs` | ECStore compatibility re-export for resync status enums. | Re-exports `rustfs-replication` contracts while downstream facade consumers migrate. |
| `replication_object_decision_boundary.rs` | Object replication option DTOs, resync target projection, delete replication decisions, and multipart planning helpers. | Keeps ECStore runtime modules from importing object decision contracts directly from `rustfs-replication`. |
| `replication_pool.rs` | Replication queue, worker pool, MRF persistence, bucket stats, and delete/object scheduling. | Depends on bucket target sys, bucket metadata sys, metadata paths, queue contracts through the queue boundary, file metadata replication contracts through local boundaries, config storage, storage contracts through the replication storage boundary, runtime sources, and notification state. |
| `replication_proxy.rs` | Proxy-target selection for GET/HEAD/Tagging reads of objects not yet replicated locally (MinIO `getProxyTargets` parity: anti-loop, version-suspended, and no-config empty branches). | Uses replication config lookup, rule matching, and target clients through local boundaries. |
| `replication_queue_boundary.rs` | Queue/admission DTOs, heal queue DTOs, worker sizing, and backpressure helpers. | Keeps ECStore runtime modules from importing queue/backpressure contracts directly from `rustfs-replication`. |
| `replication_resync_boundary.rs` | Resync DTOs, status classifiers, persisted resync/MRF codec wrappers, and ECStore error mapping. | Keeps ECStore runtime modules from importing resync contract helpers directly from `rustfs-replication`. |
| `replication_resyncer.rs` | Object replication, delete replication, resync execution, target calls, and multipart target upload paths. | Depends on target calls and target config types through the replication target boundary, metadata paths and metadata systems through the replication metadata boundary, file metadata replication contracts through the filemeta boundary, object decisions and multipart planning through the object decision boundary, resync contracts through the resync boundary, queue DTOs through the queue boundary, error contracts through the error boundary, versioning systems, storage contracts through the replication storage boundary, config-derived storage class labels through the config store, runtime sources, notification events and local event host selection through the event sink, bandwidth reader wrapping, and SetDisks lock timing. |
@@ -29,7 +29,6 @@ mod replication_object_bridge;
mod replication_object_config;
mod replication_object_decision_boundary;
pub(crate) mod replication_pool;
mod replication_proxy;
mod replication_queue_boundary;
mod replication_resync_boundary;
mod replication_resyncer;
@@ -75,7 +74,6 @@ pub use replication_pool::{
get_global_replication_pool, get_global_replication_stats, init_background_replication, persist_force_delete_intent,
read_durable_mrf_backlog, resync_start_conflict_id,
};
pub use replication_proxy::get_proxy_targets;
pub use replication_queue_boundary::{
DeletedObjectReplicationInfo, ReplicationBatchAdmission, ReplicationHealQueueResult, ReplicationOperation,
ReplicationPriority, ReplicationQueueAdmission,
@@ -667,368 +667,6 @@ async fn acknowledge_mrf_recovery<S: ReplicationStorage>(
Err(EcstoreError::PreconditionFailed)
}
/// Acquires the MRF recovery leader lock for the startup replay.
/// Returns `None` (after logging) when the lock cannot be created or another
/// node is already processing the backlog.
async fn acquire_mrf_recovery_guard<S: ReplicationStorage>(storage: &Arc<S>) -> Option<rustfs_lock::NamespaceLockGuard> {
let recovery_lock = match storage
.new_ns_lock(
ReplicationMetadataStore::rustfs_meta_bucket(),
ReplicationMetadataStore::MRF_REPLICATION_RECOVERY_LOCK,
)
.await
{
Ok(lock) => lock,
Err(error) => {
warn!(
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REPLICATION,
error = %error,
"Failed to create the MRF recovery leader lock"
);
return None;
}
};
match recovery_lock
.get_write_lock_quiet(ReplicationLockTiming::acquire_timeout())
.await
{
Ok(guard) => Some(guard),
Err(_) => {
debug!(
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REPLICATION,
"Another node is already processing the MRF recovery backlog"
);
None
}
}
}
/// Reads and decodes the on-disk MRF recovery file.
/// Returns `None` when there is nothing to replay: missing file (publishes an
/// empty available summary), read failure, or corrupt data (quarantined).
async fn load_mrf_recovery_entries<S: ReplicationStorage>(storage: &Arc<S>) -> Option<Vec<MrfReplicateEntry>> {
let data = match ReplicationConfigStore::read(storage.clone(), ReplicationMetadataStore::MRF_REPLICATION_FILE).await {
Ok(d) => d,
Err(EcstoreError::ConfigNotFound) => {
set_durable_mrf_backlog_summary(DurableMrfBacklogSummary {
available: true,
buckets: Vec::new(),
});
return None;
}
Err(e) => {
warn!(
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REPLICATION,
error = %e,
"Failed to load MRF recovery file"
);
return None;
}
};
match decode_mrf_file(&data) {
Ok(v) => Some(v),
Err(e) => {
warn!(
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REPLICATION,
error = %e,
"Failed to decode MRF recovery file — preserving corrupt data"
);
quarantine_mrf_file(storage, &data).await;
None
}
}
}
/// Replays one MRF recovery entry by operation kind.
/// Returns `None` when the entry is skipped entirely (no admission outcome);
/// entries that must be retried later are pushed onto `retry_entries`.
async fn replay_mrf_entry<S: ReplicationStorage>(
entry: &MrfReplicateEntry,
storage: &Arc<S>,
retry_entries: &mut Vec<MrfReplicateEntry>,
) -> Option<ReplicationQueueAdmission> {
match entry.op {
MrfOpKind::Delete => replay_mrf_delete_entry(entry, storage, retry_entries).await,
MrfOpKind::Object | MrfOpKind::Heal | MrfOpKind::ExistingObject => {
replay_mrf_object_entry(entry, storage, retry_entries).await
}
MrfOpKind::Metadata => replay_mrf_metadata_entry(entry, storage, retry_entries).await,
}
}
/// Replays a delete-kind MRF entry: force-delete intents replay directly,
/// stale force-delete generations are skipped, and plain deletes are
/// reconstructed as heal deletes.
async fn replay_mrf_delete_entry<S: ReplicationStorage>(
entry: &MrfReplicateEntry,
storage: &Arc<S>,
retry_entries: &mut Vec<MrfReplicateEntry>,
) -> Option<ReplicationQueueAdmission> {
if should_replay_force_delete_intent(entry) {
let operation_id = entry.force_delete_id?;
let delete = force_delete_heal_replication_info(entry, operation_id);
if replicate_delete_with_outcome(delete, storage.clone()).await {
Some(ReplicationQueueAdmission::Queued)
} else {
Some(ReplicationQueueAdmission::Missed)
}
} else if entry.force_delete_id.is_some() {
Some(ReplicationQueueAdmission::Skipped)
} else {
replay_mrf_reconstructed_delete(entry, storage, retry_entries).await
}
}
/// Pure DTO construction: heal replication info for a replayed force-delete intent.
fn force_delete_heal_replication_info(entry: &MrfReplicateEntry, operation_id: uuid::Uuid) -> DeletedObjectReplicationInfo {
DeletedObjectReplicationInfo {
delete_object: ReplicationDeletedObject {
object_name: entry.object.clone(),
force_delete: true,
force_delete_id: Some(operation_id),
force_delete_target_arns: entry.target_arns.clone(),
force_delete_generation: entry.force_delete_generation,
..Default::default()
},
bucket: entry.bucket.clone(),
op_type: ReplicationType::Heal,
event_type: REPLICATE_HEAL_DELETE.to_string(),
..Default::default()
}
}
/// Reconstruct a heal delete and re-queue it. We do NOT call
/// get_object_info here because the delete-marker or version may
/// already be absent from the local store — that is expected.
async fn replay_mrf_reconstructed_delete<S: ReplicationStorage>(
entry: &MrfReplicateEntry,
storage: &Arc<S>,
retry_entries: &mut Vec<MrfReplicateEntry>,
) -> Option<ReplicationQueueAdmission> {
let versioned = ReplicationVersioningStore::prefix_enabled(&entry.bucket, &entry.object).await;
let oi = ObjectInfo {
bucket: entry.bucket.clone(),
name: entry.object.clone(),
version_id: entry.version_id,
delete_marker: entry.delete_marker,
..Default::default()
};
let dsc = resolve_mrf_delete_replicate_decision(entry, &oi, versioned, retry_entries).await?;
let dv = reconstructed_heal_delete_info(entry, &oi, &dsc);
if replicate_delete_with_outcome(dv, storage.clone()).await {
Some(ReplicationQueueAdmission::Queued)
} else {
Some(ReplicationQueueAdmission::Missed)
}
}
/// The MRF entry does not persist the replication decision and the
/// source object is gone, so re-derive the decision from the live
/// bucket config (mirroring get_heal_replicate_object_info) and set
/// it on the reconstructed delete. Without this the decision string
/// is empty and the delete replicates to zero targets — a silent
/// no-op that leaves replicas diverged (backlog#858 / #799 B9).
async fn resolve_mrf_delete_replicate_decision(
entry: &MrfReplicateEntry,
oi: &ObjectInfo,
versioned: bool,
retry_entries: &mut Vec<MrfReplicateEntry>,
) -> Option<ReplicateDecision> {
if entry.target_arns.is_empty() {
match ReplicationMetadataStore::optional_replication_config(&entry.bucket).await {
Ok(None) => None,
Err(_) => {
retry_entries.push(entry.clone());
None
}
Ok(Some(_)) => match check_replicate_delete_strict(
&entry.bucket,
&ObjectToDelete {
object_name: entry.object.clone(),
version_id: entry.version_id,
..Default::default()
},
oi,
&ObjectOptions {
versioned,
..Default::default()
},
None,
)
.await
{
Ok(dsc) => Some(dsc),
Err(_) => {
retry_entries.push(entry.clone());
None
}
},
}
} else {
Some(replicate_decision_for_admitted_targets(&entry.target_arns))
}
}
/// Pure DTO construction: reconstructed heal delete carrying the re-derived
/// replication decision.
fn reconstructed_heal_delete_info(
entry: &MrfReplicateEntry,
oi: &ObjectInfo,
dsc: &ReplicateDecision,
) -> DeletedObjectReplicationInfo {
let mut rstate = oi.replication_state();
rstate.replicate_decision_str = dsc.to_string();
let delete_marker_mtime = entry
.delete_marker_mtime
.and_then(|nanos| OffsetDateTime::from_unix_timestamp_nanos(i128::from(nanos)).ok());
DeletedObjectReplicationInfo {
delete_object: ReplicationDeletedObject {
object_name: entry.object.clone(),
version_id: entry.version_id,
delete_marker_version_id: entry.delete_marker_version_id,
delete_marker: entry.delete_marker,
delete_marker_mtime,
force_delete: entry.force_delete,
replication_state: Some(rstate),
..Default::default()
},
bucket: entry.bucket.clone(),
op_type: ReplicationType::Heal,
event_type: REPLICATE_HEAL_DELETE.to_string(),
..Default::default()
}
}
/// Replays an Object/Heal/ExistingObject MRF entry against the live source object.
async fn replay_mrf_object_entry<S: ReplicationStorage>(
entry: &MrfReplicateEntry,
storage: &Arc<S>,
retry_entries: &mut Vec<MrfReplicateEntry>,
) -> Option<ReplicationQueueAdmission> {
let opts = ObjectOptions {
version_id: entry.version_id.map(|u| u.to_string()),
..Default::default()
};
let oi = match storage.get_object_info(&entry.bucket, &entry.object, &opts).await {
Ok(oi) => oi,
Err(e) => {
debug!(
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REPLICATION,
bucket = %entry.bucket,
object = %entry.object,
error = %e,
"MRF recovery: source object lookup failed"
);
if should_retry_mrf_source_lookup(&e) {
retry_entries.push(entry.clone());
}
return None;
}
};
if entry.target_arns.is_empty() {
// Legacy entries predate target admission persistence. They cannot
// be safely attributed, so retain the old live-config fallback.
Some(queue_replication_heal(&entry.bucket, oi, entry.retry_count.max(0) as u32).await)
} else {
let roi = admitted_mrf_replicate_object(oi, entry, entry.op.replication_type());
if replicate_object_with_outcome(roi, storage.clone()).await.1 {
Some(ReplicationQueueAdmission::Queued)
} else {
Some(ReplicationQueueAdmission::Missed)
}
}
}
/// Replays a metadata-kind MRF entry against the live source object.
async fn replay_mrf_metadata_entry<S: ReplicationStorage>(
entry: &MrfReplicateEntry,
storage: &Arc<S>,
retry_entries: &mut Vec<MrfReplicateEntry>,
) -> Option<ReplicationQueueAdmission> {
let opts = ObjectOptions {
version_id: entry.version_id.map(|u| u.to_string()),
..Default::default()
};
let oi = match storage.get_object_info(&entry.bucket, &entry.object, &opts).await {
Ok(oi) => oi,
Err(e) => {
debug!(
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REPLICATION,
bucket = %entry.bucket,
object = %entry.object,
error = %e,
"MRF metadata recovery: source object lookup failed"
);
if should_retry_mrf_source_lookup(&e) {
retry_entries.push(entry.clone());
}
return None;
}
};
if entry.target_arns.is_empty() {
Some(queue_replication_metadata(&entry.bucket, oi, entry.retry_count.max(0) as u32).await)
} else {
let roi = admitted_mrf_replicate_object(oi, entry, ReplicationType::Metadata);
if replicate_object_with_outcome(roi, storage.clone()).await.1 {
Some(ReplicationQueueAdmission::Queued)
} else {
Some(ReplicationQueueAdmission::Missed)
}
}
}
/// Pure DTO construction: replicate-object info for an entry with persisted
/// admitted targets, carrying over the entry's retry count.
fn admitted_mrf_replicate_object(oi: ObjectInfo, entry: &MrfReplicateEntry, op_type: ReplicationType) -> ReplicateObjectInfo {
let dsc = replicate_decision_for_admitted_targets(&entry.target_arns);
let mut roi = replicate_object_info_from_object_info(oi, dsc, op_type);
roi.retry_count = entry.retry_count.max(0) as u32;
roi
}
/// Acknowledges the replayed MRF prefix and returns the retained backlog.
/// On acknowledgement failure the backlog is preserved for the next startup and
/// re-read (falling back to the replayed snapshot) so the published summary stays accurate.
async fn resolve_retained_mrf_entries<S: ReplicationStorage>(
storage: &Arc<S>,
recovery_guard: &rustfs_lock::NamespaceLockGuard,
entries: &[MrfReplicateEntry],
retry_entries: &[MrfReplicateEntry],
) -> Vec<MrfReplicateEntry> {
match acknowledge_mrf_recovery(storage.clone(), recovery_guard, entries, retry_entries).await {
Ok(retained) => retained,
Err(error) => {
warn!(
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REPLICATION,
error = %error,
"Failed to acknowledge the MRF recovery prefix; preserving it for the next startup"
);
match read_mrf_entries(storage.clone()).await {
Ok(current) => current,
Err(read_error) => {
warn!(
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REPLICATION,
error = %read_error,
"Failed to refresh the MRF backlog after acknowledgement failure"
);
entries.to_vec()
}
}
}
}
}
#[derive(Debug, thiserror::Error)]
#[error("replication resync {active_resync_id} is already active for {bucket}/{arn}")]
struct ResyncActiveConflictError {
@@ -1583,12 +1221,71 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
let storage = self.storage.clone();
let handle = tokio::spawn(async move {
let Some(recovery_guard) = acquire_mrf_recovery_guard(&storage).await else {
return;
let recovery_lock = match storage
.new_ns_lock(
ReplicationMetadataStore::rustfs_meta_bucket(),
ReplicationMetadataStore::MRF_REPLICATION_RECOVERY_LOCK,
)
.await
{
Ok(lock) => lock,
Err(error) => {
warn!(
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REPLICATION,
error = %error,
"Failed to create the MRF recovery leader lock"
);
return;
}
};
let recovery_guard = match recovery_lock
.get_write_lock_quiet(ReplicationLockTiming::acquire_timeout())
.await
{
Ok(guard) => guard,
Err(_) => {
debug!(
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REPLICATION,
"Another node is already processing the MRF recovery backlog"
);
return;
}
};
let Some(entries) = load_mrf_recovery_entries(&storage).await else {
return;
let data = match ReplicationConfigStore::read(storage.clone(), ReplicationMetadataStore::MRF_REPLICATION_FILE).await {
Ok(d) => d,
Err(EcstoreError::ConfigNotFound) => {
set_durable_mrf_backlog_summary(DurableMrfBacklogSummary {
available: true,
buckets: Vec::new(),
});
return;
}
Err(e) => {
warn!(
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REPLICATION,
error = %e,
"Failed to load MRF recovery file"
);
return;
}
};
let entries = match decode_mrf_file(&data) {
Ok(v) => v,
Err(e) => {
warn!(
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REPLICATION,
error = %e,
"Failed to decode MRF recovery file — preserving corrupt data"
);
quarantine_mrf_file(&storage, &data).await;
return;
}
};
set_durable_mrf_backlog_snapshot(durable_mrf_backlog_summary_from_entries(&entries));
@@ -1597,8 +1294,187 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
let mut retry_entries = Vec::new();
for entry in entries.iter() {
let Some(admission) = replay_mrf_entry(entry, &storage, &mut retry_entries).await else {
continue;
let admission = match entry.op {
MrfOpKind::Delete => {
if should_replay_force_delete_intent(entry) {
let Some(operation_id) = entry.force_delete_id else {
continue;
};
let delete = DeletedObjectReplicationInfo {
delete_object: ReplicationDeletedObject {
object_name: entry.object.clone(),
force_delete: true,
force_delete_id: Some(operation_id),
force_delete_target_arns: entry.target_arns.clone(),
force_delete_generation: entry.force_delete_generation,
..Default::default()
},
bucket: entry.bucket.clone(),
op_type: ReplicationType::Heal,
event_type: REPLICATE_HEAL_DELETE.to_string(),
..Default::default()
};
if replicate_delete_with_outcome(delete, storage.clone()).await {
ReplicationQueueAdmission::Queued
} else {
ReplicationQueueAdmission::Missed
}
} else if entry.force_delete_id.is_some() {
ReplicationQueueAdmission::Skipped
} else {
// Reconstruct a heal delete and re-queue it. We do NOT call
// get_object_info here because the delete-marker or version may
// already be absent from the local store — that is expected.
//
// The MRF entry does not persist the replication decision and the
// source object is gone, so re-derive the decision from the live
// bucket config (mirroring get_heal_replicate_object_info) and set
// it on the reconstructed delete. Without this the decision string
// is empty and the delete replicates to zero targets — a silent
// no-op that leaves replicas diverged (backlog#858 / #799 B9).
let versioned = ReplicationVersioningStore::prefix_enabled(&entry.bucket, &entry.object).await;
let oi = ObjectInfo {
bucket: entry.bucket.clone(),
name: entry.object.clone(),
version_id: entry.version_id,
delete_marker: entry.delete_marker,
..Default::default()
};
let dsc = if entry.target_arns.is_empty() {
match ReplicationMetadataStore::optional_replication_config(&entry.bucket).await {
Ok(None) => continue,
Err(_) => {
retry_entries.push(entry.clone());
continue;
}
Ok(Some(_)) => match check_replicate_delete_strict(
&entry.bucket,
&ObjectToDelete {
object_name: entry.object.clone(),
version_id: entry.version_id,
..Default::default()
},
&oi,
&ObjectOptions {
versioned,
..Default::default()
},
None,
)
.await
{
Ok(dsc) => dsc,
Err(_) => {
retry_entries.push(entry.clone());
continue;
}
},
}
} else {
replicate_decision_for_admitted_targets(&entry.target_arns)
};
let mut rstate = oi.replication_state();
rstate.replicate_decision_str = dsc.to_string();
let delete_marker_mtime = entry
.delete_marker_mtime
.and_then(|nanos| OffsetDateTime::from_unix_timestamp_nanos(i128::from(nanos)).ok());
let dv = DeletedObjectReplicationInfo {
delete_object: ReplicationDeletedObject {
object_name: entry.object.clone(),
version_id: entry.version_id,
delete_marker_version_id: entry.delete_marker_version_id,
delete_marker: entry.delete_marker,
delete_marker_mtime,
force_delete: entry.force_delete,
replication_state: Some(rstate),
..Default::default()
},
bucket: entry.bucket.clone(),
op_type: ReplicationType::Heal,
event_type: REPLICATE_HEAL_DELETE.to_string(),
..Default::default()
};
if replicate_delete_with_outcome(dv, storage.clone()).await {
ReplicationQueueAdmission::Queued
} else {
ReplicationQueueAdmission::Missed
}
}
}
MrfOpKind::Object | MrfOpKind::Heal | MrfOpKind::ExistingObject => {
let opts = ObjectOptions {
version_id: entry.version_id.map(|u| u.to_string()),
..Default::default()
};
let oi = match storage.get_object_info(&entry.bucket, &entry.object, &opts).await {
Ok(oi) => oi,
Err(e) => {
debug!(
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REPLICATION,
bucket = %entry.bucket,
object = %entry.object,
error = %e,
"MRF recovery: source object lookup failed"
);
if should_retry_mrf_source_lookup(&e) {
retry_entries.push(entry.clone());
}
continue;
}
};
if entry.target_arns.is_empty() {
// Legacy entries predate target admission persistence. They cannot
// be safely attributed, so retain the old live-config fallback.
queue_replication_heal(&entry.bucket, oi, entry.retry_count.max(0) as u32).await
} else {
let dsc = replicate_decision_for_admitted_targets(&entry.target_arns);
let mut roi = replicate_object_info_from_object_info(oi, dsc, entry.op.replication_type());
roi.retry_count = entry.retry_count.max(0) as u32;
if replicate_object_with_outcome(roi, storage.clone()).await.1 {
ReplicationQueueAdmission::Queued
} else {
ReplicationQueueAdmission::Missed
}
}
}
MrfOpKind::Metadata => {
let opts = ObjectOptions {
version_id: entry.version_id.map(|u| u.to_string()),
..Default::default()
};
let oi = match storage.get_object_info(&entry.bucket, &entry.object, &opts).await {
Ok(oi) => oi,
Err(e) => {
debug!(
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REPLICATION,
bucket = %entry.bucket,
object = %entry.object,
error = %e,
"MRF metadata recovery: source object lookup failed"
);
if should_retry_mrf_source_lookup(&e) {
retry_entries.push(entry.clone());
}
continue;
}
};
if entry.target_arns.is_empty() {
queue_replication_metadata(&entry.bucket, oi, entry.retry_count.max(0) as u32).await
} else {
let dsc = replicate_decision_for_admitted_targets(&entry.target_arns);
let mut roi = replicate_object_info_from_object_info(oi, dsc, ReplicationType::Metadata);
roi.retry_count = entry.retry_count.max(0) as u32;
if replicate_object_with_outcome(roi, storage.clone()).await.1 {
ReplicationQueueAdmission::Queued
} else {
ReplicationQueueAdmission::Missed
}
}
}
};
if admission == ReplicationQueueAdmission::Missed {
@@ -1608,7 +1484,29 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
}
}
let retained = resolve_retained_mrf_entries(&storage, &recovery_guard, &entries, &retry_entries).await;
let retained = match acknowledge_mrf_recovery(storage.clone(), &recovery_guard, &entries, &retry_entries).await {
Ok(retained) => retained,
Err(error) => {
warn!(
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REPLICATION,
error = %error,
"Failed to acknowledge the MRF recovery prefix; preserving it for the next startup"
);
match read_mrf_entries(storage.clone()).await {
Ok(current) => current,
Err(read_error) => {
warn!(
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REPLICATION,
error = %read_error,
"Failed to refresh the MRF backlog after acknowledgement failure"
);
entries.clone()
}
}
}
};
let retained_count = retained.len();
set_durable_mrf_backlog_snapshot(durable_mrf_backlog_summary_from_entries(&retained));
@@ -1,150 +0,0 @@
// 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.
//! Proxy-target selection for reads of objects not yet replicated locally
//! (MinIO `getProxyTargets`, bucket-replication.go).
//!
//! During the active-active replication lag window a GET/HEAD/Tagging request
//! for an object the local site does not have yet may be served by proxying to
//! a replication target. This module only *selects* the candidate targets; the
//! request-path callers perform the remote calls and response translation.
use std::sync::Arc;
use tracing::debug;
use super::replication_config_boundary::{ObjectOpts, ReplicationConfigurationExt as _};
use super::replication_object_config::get_replication_config;
use super::replication_storage_boundary::ObjectOptions;
use super::replication_target_boundary::{ReplicationTargetStore, TargetClient};
/// Returns the replication-target clients eligible to serve a proxied read of
/// `bucket/object`, in rule order. Mirrors MinIO's `getProxyTargets`:
///
/// - the `source-proxy-request` header family was present at all
/// (`opts.proxy_request` / `opts.proxy_header_set`, MinIO `ProxyRequest` /
/// `ProxyHeaderSet`) -> empty. "true" is the anti-loop marker of an
/// already-proxied client read; "false" is what a peer's replication
/// worker sends on convergence HEADs so the receiver answers locally —
/// proxying that miss back would echo the source object and fake
/// convergence, permanently skipping replication;
/// - the bucket's versioning is suspended for the object -> empty;
/// - no replication configuration / no matching rule -> empty;
/// - otherwise every distinct target ARN whose rules match the object,
/// resolved through the bucket target system, skipping targets that opted
/// out of proxying (`disable_proxy`).
pub async fn get_proxy_targets(bucket: &str, object: &str, opts: &ObjectOptions) -> Vec<Arc<TargetClient>> {
if opts.proxy_request || opts.proxy_header_set {
return Vec::new();
}
if opts.version_suspended {
return Vec::new();
}
let cfg = match get_replication_config(bucket).await {
Ok(Some(cfg)) => cfg,
Ok(None) => return Vec::new(),
Err(err) => {
debug!(bucket, object, error = %err, "read proxy: failed to load replication config; not proxying");
return Vec::new();
}
};
let arns = cfg.filter_target_arns(&ObjectOpts {
name: object.to_string(),
..Default::default()
});
let mut targets = Vec::with_capacity(arns.len());
for arn in arns {
let Some(client) = ReplicationTargetStore::remote_target_client(bucket, &arn).await else {
debug!(bucket, object, arn, "read proxy: no client for replication target ARN");
continue;
};
if client.disable_proxy {
continue;
}
targets.push(client);
}
targets
}
#[cfg(test)]
mod tests {
use super::*;
fn opts() -> ObjectOptions {
ObjectOptions::default()
}
/// Anti-loop: a request that was already proxied by a peer must never be
/// proxied onward, regardless of replication configuration.
#[tokio::test]
async fn proxy_request_yields_no_targets() {
let targets = get_proxy_targets(
"bucket",
"object",
&ObjectOptions {
proxy_request: true,
..opts()
},
)
.await;
assert!(targets.is_empty());
}
/// MinIO `ProxyHeaderSet` parity: the header family being present at all
/// disables proxying, even with the value "false" — that is what a
/// peer's replication worker sends on convergence HEADs.
#[tokio::test]
async fn proxy_header_set_yields_no_targets() {
let targets = get_proxy_targets(
"bucket",
"object",
&ObjectOptions {
proxy_header_set: true,
proxy_request: false,
..opts()
},
)
.await;
assert!(targets.is_empty());
}
/// Suspended versioning disables proxying (MinIO parity): the local null
/// version is authoritative and a remote read could resurrect data.
#[tokio::test]
async fn version_suspended_yields_no_targets() {
let targets = get_proxy_targets(
"bucket",
"object",
&ObjectOptions {
version_suspended: true,
..opts()
},
)
.await;
assert!(targets.is_empty());
}
/// A bucket without replication configuration has nothing to proxy to.
/// (No metadata system is running in unit tests, so the config lookup
/// resolves to "no configuration" — the same empty-result contract.)
#[tokio::test]
async fn missing_replication_config_yields_no_targets() {
let targets = get_proxy_targets("bucket-without-replication", "object", &opts()).await;
assert!(targets.is_empty());
}
}
@@ -15,14 +15,10 @@
use super::replication_error_boundary::{Error, Result};
use super::replication_filemeta_boundary::MrfReplicateEntry;
/// Kept test-only: the runtime consumer was the worker HEAD's fake proxy
/// counting (removed in backlog#1675 P1-5); the resyncer tests still pin the
/// classifier's semantics for the real client read-proxy failure accounting.
#[cfg(test)]
pub(crate) use rustfs_replication::should_count_head_proxy_failure;
pub use rustfs_replication::{BucketReplicationResyncStatus, ResyncOpts, ResyncStatusType, TargetReplicationResyncStatus};
pub(crate) use rustfs_replication::{
is_version_id_mismatch, resync_state_accepts_update, sanitize_resync_error_detail, should_auto_resume_resync,
should_count_head_proxy_failure,
};
#[allow(
File diff suppressed because it is too large Load Diff
@@ -1161,31 +1161,6 @@ mod tests {
assert!(all.contains_key("proxy-only-bucket"));
}
/// Pins the read-proxy metric contract (backlog#1675 P1-5): the API
/// strings the GET/HEAD/Tagging proxy paths record map onto the
/// get/head/tagging totals, and only unexpected failures raise the
/// failed counters.
#[tokio::test]
async fn test_proxy_stats_map_read_proxy_apis_to_totals() {
let stats = ReplicationStats::new();
stats.inc_proxy("proxy-bucket", "GetObject", false).await;
stats.inc_proxy("proxy-bucket", "GetObject", true).await;
stats.inc_proxy("proxy-bucket", "HeadObject", false).await;
stats.inc_proxy("proxy-bucket", "GetObjectTagging", false).await;
stats.inc_proxy("proxy-bucket", "PutObjectTagging", false).await;
stats.inc_proxy("proxy-bucket", "DeleteObjectTagging", true).await;
let metric = stats.get_proxy_stats("proxy-bucket").await;
assert_eq!(metric.get_total, 2);
assert_eq!(metric.get_failed, 1);
assert_eq!(metric.head_total, 1);
assert_eq!(metric.head_failed, 0);
assert_eq!(metric.get_tag_total, 1);
assert_eq!(metric.put_tag_total, 1);
assert_eq!(metric.delete_tag_total, 1);
assert_eq!(metric.delete_tag_failed, 1);
}
#[tokio::test]
async fn test_calculate_bucket_replication_stats_merges_resync_metrics() {
let stats = ReplicationStats::new();
@@ -36,8 +36,7 @@ use time::OffsetDateTime;
use time::format_description::well_known::Rfc3339;
pub(crate) use crate::bucket::bucket_target_sys::{
AdvancedPutOptions, PutObjectOptions, PutObjectPartOptions, RemoveObjectOptions, SsecPassthroughCapability, TargetClient,
resolve_read_api_version_id,
AdvancedPutOptions, PutObjectOptions, PutObjectPartOptions, RemoveObjectOptions, TargetClient,
};
#[cfg(test)]
pub(crate) use crate::bucket::target::BucketTarget;
@@ -66,8 +65,6 @@ static STANDARD_HEADERS: &[&str] = &[
];
const ERR_REPLICATION_ENCRYPTION_METADATA_UNSUPPORTED: &str = "replication source contains unsupported encryption metadata";
pub(crate) const ERR_REPLICATION_SSEC_PASSTHROUGH_UNSUPPORTED: &str = "replication target does not support SSE-C passthrough: the replica would lose its decryption material \
(run ?replication-check to re-probe)";
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum ReplicationSourceEncryption {
@@ -149,54 +146,6 @@ pub(crate) fn replication_object_is_ssec_encrypted(user_defined: &HashMap<String
rustfs_replication::is_ssec_encrypted(user_defined)
}
/// Fail-closed decision for an SSE-C passthrough replication attempt, derived
/// from the target's cached [`SsecPassthroughCapability`]. Pure so the policy
/// can migrate with the worker (M2) without dragging the cache along; the
/// caller computes `expired` from the cache record's age (see
/// `SSEC_PASSTHROUGH_CAPABILITY_TTL`).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum SsecPassthroughGate {
/// Not an SSE-C object, or the target has a fresh proof that it preserves
/// the passthrough transport headers: replicate without a HEAD-back audit.
Proceed,
/// No usable verdict — first SSE-C attempt since the target was (re)built,
/// or the recorded verdict (in either direction) aged out: PUT, then HEAD
/// the replica back and require SSE-C evidence before reporting COMPLETED.
ProceedWithAudit,
/// The target was recently proven to drop the passthrough headers: do not
/// send the PUT, report FAILED (the object stays on the normal MRF retry
/// channel and re-audits once the verdict expires).
FailClosed,
}
pub(crate) fn ssec_passthrough_gate(ssec: bool, capability: SsecPassthroughCapability, expired: bool) -> SsecPassthroughGate {
if !ssec {
return SsecPassthroughGate::Proceed;
}
// An expired verdict — Supported or Unsupported — must be re-earned: a
// stale Unsupported would otherwise stick forever after a target upgrade,
// and a stale Supported would fail open after a backend swap behind the
// same endpoint.
if expired {
return SsecPassthroughGate::ProceedWithAudit;
}
match capability {
SsecPassthroughCapability::Supported => SsecPassthroughGate::Proceed,
SsecPassthroughCapability::Unknown => SsecPassthroughGate::ProceedWithAudit,
SsecPassthroughCapability::Unsupported => SsecPassthroughGate::FailClosed,
}
}
/// True when a replication-check HEAD of the replica proves the SSE-C
/// material survived passthrough: a RustFS target restores the transport
/// headers into the stored SSE-C keys and its HEAD echoes
/// `x-amz-server-side-encryption-customer-algorithm` (the replication-check
/// exemption skips key validation but not the metadata echo). A target that
/// dropped the headers stored a plain object and echoes nothing.
pub(crate) fn ssec_passthrough_evidence_present(head: &HeadObjectOutput) -> bool {
head.sse_customer_algorithm.as_deref().is_some_and(|algo| !algo.is_empty())
}
pub(crate) struct ReplicationTargetStore;
impl ReplicationTargetStore {
@@ -216,17 +165,6 @@ impl ReplicationTargetStore {
BucketTargetSys::get().mark_target_offline(target_client).await
}
/// Returns the cached verdict and whether it has outlived its TTL.
pub(crate) async fn ssec_passthrough_capability(arn: &str) -> (SsecPassthroughCapability, bool) {
BucketTargetSys::get().ssec_passthrough_capability(arn).await
}
pub(crate) async fn record_ssec_passthrough_capability(arn: &str, capability: SsecPassthroughCapability) {
BucketTargetSys::get()
.record_ssec_passthrough_capability(arn, capability)
.await
}
#[cfg(test)]
pub(crate) async fn register_test_target(target_client: &Arc<TargetClient>) {
BucketTargetSys::get().arn_remotes_map.write().await.insert(
@@ -960,71 +898,6 @@ mod tests {
}
}
/// N2 fail-closed policy: SSE-C replication may only proceed silently
/// against a target with a FRESH proof that it preserves the passthrough
/// transport headers. Unknown targets must be audited; freshly-flagged
/// dropping targets must never receive the PUT; an expired verdict in
/// EITHER direction must be re-earned through the audit — a sticky
/// Unsupported would outlive a target upgrade, and a sticky Supported
/// would fail open after a backend swap behind the same endpoint.
#[test]
fn ssec_passthrough_gate_is_fail_closed_and_ttl_bounded() {
for capability in [
SsecPassthroughCapability::Unknown,
SsecPassthroughCapability::Supported,
SsecPassthroughCapability::Unsupported,
] {
for expired in [false, true] {
assert_eq!(
ssec_passthrough_gate(false, capability, expired),
SsecPassthroughGate::Proceed,
"non-SSE-C objects must never be gated on the passthrough capability"
);
}
}
assert_eq!(
ssec_passthrough_gate(true, SsecPassthroughCapability::Supported, false),
SsecPassthroughGate::Proceed
);
assert_eq!(
ssec_passthrough_gate(true, SsecPassthroughCapability::Unknown, false),
SsecPassthroughGate::ProceedWithAudit
);
assert_eq!(
ssec_passthrough_gate(true, SsecPassthroughCapability::Unsupported, false),
SsecPassthroughGate::FailClosed
);
// Expiry flips both directions back to the audit.
assert_eq!(
ssec_passthrough_gate(true, SsecPassthroughCapability::Unsupported, true),
SsecPassthroughGate::ProceedWithAudit,
"an expired Unsupported verdict must allow a re-audit (upgraded target recovers without operator action)"
);
assert_eq!(
ssec_passthrough_gate(true, SsecPassthroughCapability::Supported, true),
SsecPassthroughGate::ProceedWithAudit,
"an expired Supported verdict must be re-proven (backend swap behind the same endpoint must not fail open)"
);
}
#[test]
fn ssec_passthrough_evidence_requires_customer_algorithm_echo() {
let with_evidence = HeadObjectOutput::builder().sse_customer_algorithm("AES256").build();
assert!(ssec_passthrough_evidence_present(&with_evidence));
let empty_algorithm = HeadObjectOutput::builder().sse_customer_algorithm("").build();
assert!(
!ssec_passthrough_evidence_present(&empty_algorithm),
"an empty echo is not evidence of preserved SSE-C material"
);
let without_evidence = HeadObjectOutput::builder().e_tag("\"abc\"").content_length(8).build();
assert!(
!ssec_passthrough_evidence_present(&without_evidence),
"a plain HEAD response must classify the target as having dropped the material"
);
}
#[test]
fn replication_put_options_adds_ssec_checksum_metadata() {
let metadata = HashMap::from([(SSEC_ALGORITHM_HEADER.to_string(), "AES256".to_string())]);
+291 -12
View File
@@ -820,10 +820,263 @@ impl BitrotWriterWrapper {
}
}
// --- startup bitrot self-test (rustfs/backlog#1873, MinIO bitrotSelfTest parity) ---
//
// A broken hash implementation (bad SIMD feature combination, platform drift, a
// key-handling regression) fails silently: every shard reads back "corrupt",
// heal rewrites data that was fine, and cross-platform clusters disagree about
// which copy is healthy. The self-test below pins the algorithms the moment a
// process starts, so a drifted build announces itself instead of quietly
// rewriting objects. See docs/rustfs-heal-scanner-vs-minio-comprehensive-
// analysis-2026-08-16.md §6 HS-11.
/// Length of the deterministic self-test payload.
pub const BITROT_SELF_TEST_PAYLOAD_LEN: usize = 4096;
/// Known-answer digest of [`bitrot_self_test_payload`] under `HighwayHash256S`
/// (the production default). Pinned so any platform or build where the
/// implementation drifts fails startup instead of miss-hashing shards.
const BITROT_SELF_TEST_KAT_HIGHWAY_HASH256S: [u8; 32] = [
0xb9, 0x32, 0xa2, 0xaa, 0x4a, 0xb7, 0x33, 0x6a, 0xa3, 0xca, 0x7e, 0x61, 0x9d, 0x86, 0x52, 0x14, 0x6e, 0x7f, 0xd8, 0x9e, 0xea,
0x08, 0xd9, 0x8c, 0x33, 0x85, 0x87, 0x19, 0x30, 0xd6, 0xed, 0x06,
];
/// Known-answer digest of the same payload under `HighwayHash256SLegacy`.
const BITROT_SELF_TEST_KAT_HIGHWAY_HASH256S_LEGACY: [u8; 32] = [
0x98, 0x24, 0x71, 0x4f, 0x16, 0xbb, 0x48, 0x39, 0xed, 0x68, 0xfa, 0x63, 0x5e, 0xd9, 0x07, 0x61, 0xdf, 0x0a, 0xff, 0xcf, 0x7d,
0x8c, 0xa8, 0xc7, 0xc0, 0xb6, 0x6f, 0x05, 0xdb, 0xda, 0x5a, 0x22,
];
/// FIPS 180-2 test vector: SHA-256 of the ASCII string "abc". Unlike the
/// Highway digests above this one is externally verifiable, so it guards the
/// whole `HashAlgorithm` plumbing even for readers who distrust pinned
/// self-computed constants.
const BITROT_SELF_TEST_KAT_SHA256_ABC: [u8; 32] = [
0xba, 0x78, 0x16, 0xbf, 0x8f, 0x01, 0xcf, 0xea, 0x41, 0x41, 0x40, 0xde, 0x5d, 0xae, 0x22, 0x23, 0xb0, 0x03, 0x61, 0xa3, 0x96,
0x17, 0x7a, 0x9c, 0xb4, 0x10, 0xff, 0x61, 0xf2, 0x00, 0x15, 0xad,
];
/// Deterministic self-test payload: xorshift64* from a fixed seed, so every
/// platform and every run hashes the same 4096 bytes.
fn bitrot_self_test_payload() -> [u8; BITROT_SELF_TEST_PAYLOAD_LEN] {
let mut state = 0x9E37_79B9_7F4A_7C15u64;
let mut payload = [0u8; BITROT_SELF_TEST_PAYLOAD_LEN];
for byte in payload.iter_mut() {
state ^= state >> 12;
state ^= state << 25;
state ^= state >> 27;
*byte = state.wrapping_mul(0x2545_F491_4F6C_DD1D) as u8;
}
payload
}
/// Why a bitrot self-test failed.
#[derive(Debug)]
pub enum BitrotSelfTestError {
/// A known-answer digest mismatched the pinned constant.
KnownAnswerMismatch {
algorithm: &'static str,
got: String,
want: String,
},
/// A freshly encoded shard failed `bitrot_verify`.
RoundtripVerify { algorithm: &'static str, detail: String },
/// A verified roundtrip read back different bytes than were written.
RoundtripReadback { algorithm: &'static str },
/// A deliberately tampered shard was not rejected by `bitrot_verify`.
TamperNotRejected {
algorithm: &'static str,
tampered: &'static str,
},
}
impl std::fmt::Display for BitrotSelfTestError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::KnownAnswerMismatch { algorithm, got, want } => {
write!(f, "known-answer mismatch for {algorithm}: got {got}, want {want}")
}
Self::RoundtripVerify { algorithm, detail } => write!(f, "{algorithm} roundtrip shard failed verification: {detail}"),
Self::RoundtripReadback { algorithm } => write!(f, "{algorithm} roundtrip read back different bytes"),
Self::TamperNotRejected { algorithm, tampered } => {
write!(f, "{algorithm} tampered shard ({tampered}) was not rejected")
}
}
}
}
impl std::error::Error for BitrotSelfTestError {}
fn self_test_hex(bytes: &[u8]) -> String {
rustfs_utils::hex(bytes)
}
// (kept as a named one-liner so every KAT failure site reads the same; the
// underlying formatter is the shared `rustfs_utils::hex`)
/// Compare a digest against its pinned constant. Split out so a test can drive
/// it with a wrong constant and prove the mismatch path fires.
fn bitrot_kat_check(
algorithm: &'static str,
algo: &HashAlgorithm,
payload: &[u8],
expected: &[u8; 32],
) -> Result<(), BitrotSelfTestError> {
let digest = algo.hash_encode(payload);
let digest = digest.as_ref();
if digest.len() != expected.len() || digest != expected.as_slice() {
return Err(BitrotSelfTestError::KnownAnswerMismatch {
algorithm,
got: self_test_hex(digest),
want: self_test_hex(expected),
});
}
Ok(())
}
/// Encode `payload` with `shard_size` blocks, verify it end to end, and read
/// every block back through `BitrotReader` comparing bytes.
async fn bitrot_roundtrip_check(
algorithm: &'static str,
algo: HashAlgorithm,
payload: &[u8],
shard_size: usize,
) -> Result<(), BitrotSelfTestError> {
let mut writer = BitrotWriter::new(std::io::Cursor::new(Vec::<u8>::new()), shard_size, algo.clone());
for chunk in payload.chunks(shard_size) {
writer
.write(chunk)
.await
.map_err(|err| BitrotSelfTestError::RoundtripVerify {
algorithm,
detail: format!("encode failed: {err}"),
})?;
}
let encoded = writer.into_inner().into_inner();
let on_disk = bitrot_shard_file_size(payload.len(), shard_size, algo.clone());
if encoded.len() != on_disk {
return Err(BitrotSelfTestError::RoundtripVerify {
algorithm,
detail: format!("encoded {} bytes, size formula says {on_disk}", encoded.len()),
});
}
bitrot_verify(std::io::Cursor::new(encoded.clone()), on_disk, payload.len(), algo.clone(), shard_size)
.await
.map_err(|err| BitrotSelfTestError::RoundtripVerify {
algorithm,
detail: err.to_string(),
})?;
let mut reader = BitrotReader::new(std::io::Cursor::new(encoded), shard_size, algo, false);
let mut offset = 0usize;
while offset < payload.len() {
let want = shard_size.min(payload.len() - offset);
let mut buf = vec![0u8; want];
let read = reader
.read(&mut buf)
.await
.map_err(|err| BitrotSelfTestError::RoundtripVerify {
algorithm,
detail: format!("read back failed at offset {offset}: {err}"),
})?;
if read != want || buf[..read] != payload[offset..offset + read] {
return Err(BitrotSelfTestError::RoundtripReadback { algorithm });
}
offset += read;
}
Ok(())
}
/// Flip one byte and require `bitrot_verify` to reject the result.
async fn bitrot_tamper_check(
algorithm: &'static str,
algo: HashAlgorithm,
payload: &[u8],
shard_size: usize,
tampered: &'static str,
flip_at: usize,
) -> Result<(), BitrotSelfTestError> {
let mut writer = BitrotWriter::new(std::io::Cursor::new(Vec::<u8>::new()), shard_size, algo.clone());
for chunk in payload.chunks(shard_size) {
writer.write(chunk).await.expect("self-test encode should not fail");
}
let mut corrupt = writer.into_inner().into_inner();
let flip_index = flip_at % corrupt.len();
corrupt[flip_index] ^= 0x80;
let on_disk = bitrot_shard_file_size(payload.len(), shard_size, algo.clone());
match bitrot_verify(std::io::Cursor::new(corrupt), on_disk, payload.len(), algo, shard_size).await {
// The flipped byte must be rejected as a hash mismatch specifically, not
// by any incidental read error: an in-memory cursor cannot fail reads,
// so accepting any other failure here would mask a verify path that
// errors out before it ever compares hashes.
Err(err) if err.to_string().contains("hash mismatch") => Ok(()),
Ok(()) => Err(BitrotSelfTestError::TamperNotRejected { algorithm, tampered }),
Err(err) => Err(BitrotSelfTestError::RoundtripVerify {
algorithm,
detail: format!("tampered shard rejected with an unexpected error: {err}"),
}),
}
}
/// Verify every bitrot algorithm this crate can write or verify in production:
/// both streaming Highway variants roundtrip end to end (encode → size formula
/// → `bitrot_verify` → read back) and reject a flipped byte in both the data
/// and the leading hash, while all three hashed algorithms reproduce their
/// pinned known-answer digests.
///
/// Runs in well under a millisecond on 4 KiB of data; callers may run it inline
/// at startup. Pure CPU, no allocation beyond a few KiB of scratch.
pub async fn bitrot_self_test() -> Result<(), BitrotSelfTestError> {
let payload = bitrot_self_test_payload();
// Externally verifiable vector first: it guards the HashAlgorithm plumbing
// itself, before any self-pinned constants are consulted.
let abc = HashAlgorithm::SHA256.hash_encode(b"abc");
if abc.as_ref() != BITROT_SELF_TEST_KAT_SHA256_ABC.as_slice() {
return Err(BitrotSelfTestError::KnownAnswerMismatch {
algorithm: "SHA256",
got: self_test_hex(abc.as_ref()),
want: self_test_hex(&BITROT_SELF_TEST_KAT_SHA256_ABC),
});
}
bitrot_kat_check(
"HighwayHash256S",
&HashAlgorithm::HighwayHash256S,
&payload,
&BITROT_SELF_TEST_KAT_HIGHWAY_HASH256S,
)?;
bitrot_kat_check(
"HighwayHash256SLegacy",
&HashAlgorithm::HighwayHash256SLegacy,
&payload,
&BITROT_SELF_TEST_KAT_HIGHWAY_HASH256S_LEGACY,
)?;
for (algorithm, algo) in [
("HighwayHash256S", HashAlgorithm::HighwayHash256S),
("HighwayHash256SLegacy", HashAlgorithm::HighwayHash256SLegacy),
] {
// Full blocks plus a partial tail, exactly like a real part stripe.
let tail_len = 2 * 1024 + 333;
bitrot_roundtrip_check(algorithm, algo.clone(), &payload, 1024).await?;
bitrot_roundtrip_check(algorithm, algo.clone(), &payload[..tail_len], 1024).await?;
// One flipped byte in the final data block, one in the first leading
// hash: both must fail verification.
bitrot_tamper_check(algorithm, algo.clone(), &payload, 1024, "final data byte", payload.len() - 1).await?;
bitrot_tamper_check(algorithm, algo, &payload, 1024, "leading hash byte", 0).await?;
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::{
BitrotReader, BitrotWriter, BitrotWriterWrapper, CustomWriter, bitrot_shard_file_size, bitrot_verify, write_all_vectored,
BitrotReader, BitrotWriter, BitrotWriterWrapper, CustomWriter, bitrot_kat_check, bitrot_self_test,
bitrot_self_test_payload, bitrot_shard_file_size, bitrot_verify, write_all_vectored,
};
use super::{MAX_RETAINED_CHUNKS_PER_BLOCK, ShardChunkRead, ShardSource};
use bytes::Bytes;
@@ -1090,6 +1343,32 @@ mod tests {
}
}
#[test]
fn bitrot_self_test_payload_is_deterministic() {
// Two independent builds of the payload must agree byte for byte, or
// the pinned known-answer digests below would be meaningless.
assert_eq!(bitrot_self_test_payload(), bitrot_self_test_payload());
}
#[test]
fn bitrot_self_test_rejects_a_wrong_known_answer_digest() {
let payload = bitrot_self_test_payload();
let wrong = [0u8; 32];
let err = bitrot_kat_check("HighwayHash256S", &HashAlgorithm::HighwayHash256S, &payload, &wrong)
.expect_err("a zeroed digest must never match");
match err {
super::BitrotSelfTestError::KnownAnswerMismatch { algorithm, .. } => assert_eq!(algorithm, "HighwayHash256S"),
other => panic!("expected KnownAnswerMismatch, got {other:?}"),
}
}
#[tokio::test]
async fn bitrot_self_test_passes() {
bitrot_self_test()
.await
.expect("the pinned digests and roundtrip checks must all pass on this platform");
}
#[tokio::test]
async fn vectored_test_writers_cover_fallback_flush_and_shutdown_paths() {
let mut counting = VectoredCountingWriter::default();
@@ -1189,7 +1468,7 @@ mod tests {
let last = corrupt.len() - 1;
corrupt[last] ^= 0x80;
let err = bitrot_verify(
Cursor::new(corrupt),
std::io::Cursor::new(corrupt),
super::bitrot_shard_file_size(data.len(), shard_size, algo.clone()),
data.len(),
algo,
@@ -1282,7 +1561,7 @@ mod tests {
#[tokio::test]
async fn bitrot_reader_rejects_output_buffers_larger_than_shard_size() {
let mut reader = BitrotReader::new(Cursor::new(Vec::<u8>::new()), 4, HashAlgorithm::None, false);
let mut reader = BitrotReader::new(std::io::Cursor::new(Vec::<u8>::new()), 4, HashAlgorithm::None, false);
let mut out = [0u8; 5];
let err = reader
.read(&mut out)
@@ -1407,7 +1686,7 @@ mod tests {
(HashAlgorithm::HighwayHash256, true),
] {
let label = format!("{algo:?}");
let writer = Cursor::new(Vec::<u8>::new());
let writer = std::io::Cursor::new(Vec::<u8>::new());
let mut w = BitrotWriter::new(writer, shard_size, algo.clone());
w.write(&[7u8; 16]).await.unwrap();
let written = w.into_inner().into_inner();
@@ -1492,7 +1771,7 @@ mod tests {
}
async fn encode_one_block(payload: &[u8], shard_size: usize, algo: HashAlgorithm) -> Vec<u8> {
let mut w = BitrotWriter::new(Cursor::new(Vec::<u8>::new()), shard_size, algo);
let mut w = BitrotWriter::new(std::io::Cursor::new(Vec::<u8>::new()), shard_size, algo);
w.write(payload).await.unwrap();
w.into_inner().into_inner()
}
@@ -1600,7 +1879,7 @@ mod tests {
for algo in [HashAlgorithm::HighwayHash256S, HashAlgorithm::HighwayHash256SLegacy] {
for &size in &[1usize, 16, 17, 32, 40, 48] {
let payload: Vec<u8> = (0..size).map(|i| i as u8).collect();
let mut w = BitrotWriter::new(Cursor::new(Vec::<u8>::new()), shard_size, algo.clone());
let mut w = BitrotWriter::new(std::io::Cursor::new(Vec::<u8>::new()), shard_size, algo.clone());
for chunk in payload.chunks(shard_size) {
w.write(chunk).await.unwrap();
}
@@ -1674,14 +1953,14 @@ mod tests {
w.write(&data).await.expect("write shard");
let mut via_read = vec![0u8; SHARD];
let n1 = BitrotReader::new(Cursor::new(encoded.clone()), SHARD, algo.clone(), false)
let n1 = BitrotReader::new(std::io::Cursor::new(encoded.clone()), SHARD, algo.clone(), false)
.read(&mut via_read)
.await
.expect("read");
// A buffer with only capacity — no initialized bytes at all.
let mut via_append: Vec<u8> = Vec::with_capacity(SHARD);
let n2 = BitrotReader::new(Cursor::new(encoded), SHARD, algo.clone(), false)
let n2 = BitrotReader::new(std::io::Cursor::new(encoded), SHARD, algo.clone(), false)
.read_appending(&mut via_append, SHARD)
.await
.expect("read_appending");
@@ -1706,7 +1985,7 @@ mod tests {
encoded.truncate(encoded.len() - 1);
let mut out: Vec<u8> = Vec::with_capacity(SHARD);
let err = BitrotReader::new(Cursor::new(encoded), SHARD, algo.clone(), false)
let err = BitrotReader::new(std::io::Cursor::new(encoded), SHARD, algo.clone(), false)
.read_appending(&mut out, SHARD)
.await
.expect_err("a truncated shard must not succeed");
@@ -1732,7 +2011,7 @@ mod tests {
encoded[last] ^= 0xff;
let mut out: Vec<u8> = Vec::with_capacity(SHARD);
let err = BitrotReader::new(Cursor::new(encoded), SHARD, algo, false)
let err = BitrotReader::new(std::io::Cursor::new(encoded), SHARD, algo, false)
.read_appending(&mut out, SHARD)
.await
.expect_err("a corrupt shard must not verify");
@@ -1844,7 +2123,7 @@ mod tests {
"Cursor<Bytes> must be able to hand out a block, otherwise the fast path is dead code"
);
assert_eq!(mem.position(), 8, "taking a block must advance like a read of the same length");
let mut streamed = Cursor::new(encoded.clone());
let mut streamed = std::io::Cursor::new(encoded.clone());
assert!(
ShardSource::try_take_block(&mut streamed, 8).is_none(),
"a non-Bytes source must stay on the streaming path"
@@ -1872,7 +2151,7 @@ mod tests {
);
let mut via_stream: Vec<u8> = Vec::with_capacity(SHARD);
BitrotReader::new(Cursor::new(encoded), SHARD, algo, false)
BitrotReader::new(std::io::Cursor::new(encoded), SHARD, algo, false)
.read_appending(&mut via_stream, SHARD)
.await
.expect("streaming read");
-14
View File
@@ -277,20 +277,6 @@ pub struct ObjectOptions {
/// fence avoids recursively acquiring the read lock behind a queued writer.
pub bucket_lifecycle_lock_fence: Option<NamespaceLockFence>,
pub replication_request: bool,
/// True when the inbound request carried the
/// `{x-rustfs-,x-minio-}source-proxy-request` header family with the
/// value "true": the request was already proxied by a replication peer,
/// so this server must not proxy a local miss onward (anti-loop,
/// MinIO-compatible). The header only disables proxying — it grants no
/// capability — so no authorization gate is required to honor it.
pub proxy_request: bool,
/// True when the `source-proxy-request` header family was present at
/// all, regardless of value (MinIO's `ProxyHeaderSet`). A replication
/// peer sends `source-proxy-request: false` on its worker convergence
/// HEADs precisely so the receiver answers locally instead of proxying
/// back — otherwise a proxied 404->200 echo makes the worker believe the
/// object already converged and it never replicates it.
pub proxy_header_set: bool,
/// Source-cluster LWW timestamps carried by an authorized replication
/// request; None when the source never modified the category. Only the
/// replication-authorized options builders may set these.
+5
View File
@@ -37,7 +37,11 @@ hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu"]
[dependencies]
hotpath.workspace = true
humantime.workspace = true
http.workspace = true
hyper = { workspace = true, features = ["http2", "http1", "server"] }
reqwest = { workspace = true, features = ["json"] }
rustfs-signer.workspace = true
s3s.workspace = true
jiff = { workspace = true, features = ["serde"] }
serde = { workspace = true, features = ["derive"] }
serde_json = { workspace = true, features = ["raw_value"] }
@@ -49,3 +53,4 @@ doctest = false
[dev-dependencies]
rmp-serde.workspace = true
tokio = { workspace = true, features = ["macros", "rt-multi-thread", "net"] }
+851
View File
@@ -0,0 +1,851 @@
// 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.
//! Admin API HTTP client for heal and scanner management (rustfs/backlog#1869).
//!
//! [`AdminClient`] speaks the `/rustfs/admin/v3` surface with S3 SigV4
//! request signing (the same scheme the server's admin router authenticates),
//! so `mc`-style tooling and automation can drive heal start/query/cancel and
//! read background-heal / scanner status without hand-rolling HTTP.
//!
//! Wire structs in this module mirror the server-side shapes
//! (`rustfs/src/admin/handlers/heal.rs`, `handlers/scanner.rs`,
//! `rustfs-common/src/heal_channel.rs`), following the madmin-go model where
//! the SDK owns its own copies and round-trip tests pin the encoding. Deeply
//! nested status payloads that the server composes from runtime types are
//! carried through as `serde_json::Value` and flattened maps rather than
//! duplicated field-for-field, so the client cannot silently drift on fields
//! it never interprets.
use crate::heal_commands::HealResultItem;
use http::Method;
use serde::{Deserialize, Serialize, de};
use std::time::Duration;
/// Default admin API path prefix on a RustFS endpoint.
pub const DEFAULT_ADMIN_API_PREFIX: &str = "/rustfs/admin";
/// Default SigV4 region when the server has no explicit region configured.
pub const DEFAULT_REGION: &str = "us-east-1";
/// Scan mode for a heal request, mirroring the server's numeric-or-name wire
/// encoding (`0` unknown/default, `1` normal, `2` deep).
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub enum HealScanMode {
/// Server default; behaves as [`HealScanMode::Normal`].
#[default]
Unknown,
/// Metadata-level checks only.
Normal,
/// Full bitrot verification while healing.
Deep,
}
impl HealScanMode {
fn wire_number(self) -> u8 {
match self {
Self::Unknown => 0,
Self::Normal => 1,
Self::Deep => 2,
}
}
fn from_wire_number(value: u8) -> Option<Self> {
match value {
0 => Some(Self::Unknown),
1 => Some(Self::Normal),
2 => Some(Self::Deep),
_ => None,
}
}
fn from_wire_name(value: &str) -> Option<Self> {
match value {
"unknown" => Some(Self::Unknown),
"normal" => Some(Self::Normal),
"deep" => Some(Self::Deep),
_ => None,
}
}
}
impl Serialize for HealScanMode {
fn serialize<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
serializer.serialize_u8(self.wire_number())
}
}
impl<'de> Deserialize<'de> for HealScanMode {
fn deserialize<D: serde::Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
struct HealScanModeVisitor;
impl de::Visitor<'_> for HealScanModeVisitor {
type Value = HealScanMode;
fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
formatter.write_str("a heal scan mode number or name")
}
fn visit_u64<E: de::Error>(self, value: u64) -> Result<Self::Value, E> {
u8::try_from(value)
.ok()
.and_then(HealScanMode::from_wire_number)
.ok_or_else(|| E::custom(format!("unknown heal scan mode number: {value}")))
}
fn visit_str<E: de::Error>(self, value: &str) -> Result<Self::Value, E> {
HealScanMode::from_wire_name(value).ok_or_else(|| E::custom(format!("unknown heal scan mode name: {value}")))
}
}
deserializer.deserialize_any(HealScanModeVisitor)
}
}
/// Heal options for an admin heal request (mirror of the server body type).
/// Fields default on decode: a client should tolerate a server response whose
/// settings object omits fields it never set.
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct HealOpts {
#[serde(default)]
pub recursive: bool,
#[serde(rename = "dryRun", default)]
pub dry_run: bool,
#[serde(default)]
pub remove: bool,
#[serde(default)]
pub recreate: bool,
#[serde(rename = "scanMode", default)]
pub scan_mode: HealScanMode,
#[serde(rename = "updateParity", default)]
pub update_parity: bool,
#[serde(rename = "nolock", default)]
pub no_lock: bool,
#[serde(rename = "pool", default)]
pub pool: Option<usize>,
#[serde(rename = "set", default)]
pub set: Option<usize>,
}
/// Successful heal start / path-scoped cancel response.
#[derive(Debug, Clone, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct HealStartSuccess {
pub client_token: String,
pub client_address: String,
#[serde(default)]
pub start_time: String,
}
/// Heal task status response (query, cancel-with-token, start-then-poll).
#[derive(Debug, Clone, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct HealTaskStatus {
/// `running` | `finished` | `stopped` | `notFound`.
pub summary: String,
/// Failure detail for stopped tasks; empty otherwise.
#[serde(rename = "detail", default)]
pub failure_detail: String,
#[serde(default)]
pub start_time: String,
#[serde(default)]
pub settings: HealOpts,
#[serde(default)]
pub items: Vec<HealResultItem>,
#[serde(default)]
pub truncated: bool,
/// Live progress snapshot; the exact shape is owned by the heal runtime.
#[serde(default)]
pub progress: Option<serde_json::Value>,
}
/// `POST /v3/background-heal/status` response. Known top-level fields are
/// typed; the flattened heal info and operations matrix pass through verbatim.
#[derive(Debug, Clone, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct BackgroundHealStatus {
/// `disabled` | `uninitialized` | `idle` | `active` | `degraded`.
pub state: String,
#[serde(default)]
pub heal_queue_length: u64,
#[serde(default)]
pub heal_active_tasks: u64,
#[serde(default)]
pub cluster_status_complete: bool,
#[serde(default)]
pub progress: Option<serde_json::Value>,
/// Remaining wire fields (flattened `BackgroundHealInfo` plus the
/// priority-by-source operations matrix), carried verbatim.
#[serde(flatten)]
pub extra: serde_json::Map<String, serde_json::Value>,
}
/// `GET /v3/scanner/status` response, typed at the fields operators branch
/// on; everything else passes through verbatim.
#[derive(Debug, Clone, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct ScannerStatus {
pub enabled: bool,
/// `fresh` | `stale` | `unknown`; absent when the scanner never completed
/// a cycle.
#[serde(default)]
pub freshness: Option<ScannerFreshness>,
#[serde(flatten)]
pub extra: serde_json::Map<String, serde_json::Value>,
}
/// Freshness block of the scanner status response.
#[derive(Debug, Clone, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct ScannerFreshness {
/// `fresh` | `stale` | `unknown`.
pub state: String,
}
impl ScannerStatus {
/// Convenience accessor for the freshness state string.
pub fn freshness(&self) -> &str {
self.freshness
.as_ref()
.map(|freshness| freshness.state.as_str())
.unwrap_or("unknown")
}
}
/// Everything that can go wrong in an admin client call.
#[derive(Debug)]
pub enum AdminClientError {
/// The endpoint URL could not be parsed.
InvalidEndpoint(String),
/// Request build/send failed (DNS, connect, timeout, body read).
Transport(reqwest::Error),
/// The server answered a non-2xx status.
HttpStatus { status: u16, body: String },
/// The response body did not decode into the expected shape.
Decode { message: String },
}
impl std::fmt::Display for AdminClientError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::InvalidEndpoint(message) => write!(f, "invalid admin endpoint: {message}"),
Self::Transport(err) => write!(f, "admin request transport failure: {err}"),
Self::HttpStatus { status, body } => write!(f, "admin request failed with HTTP {status}: {body}"),
Self::Decode { message } => write!(f, "admin response decode failure: {message}"),
}
}
}
impl std::error::Error for AdminClientError {}
impl From<reqwest::Error> for AdminClientError {
fn from(err: reqwest::Error) -> Self {
Self::Transport(err)
}
}
/// A signed client for a RustFS admin API.
#[derive(Debug, Clone)]
pub struct AdminClient {
endpoint: reqwest::Url,
access_key: String,
secret_key: String,
session_token: String,
region: String,
api_prefix: String,
http: reqwest::Client,
}
impl AdminClient {
/// Build a client for `endpoint` (e.g. `http://127.0.0.1:9000`) using root
/// or admin credentials. Requests are SigV4-signed with the same scheme
/// the server's admin router authenticates.
pub fn new(endpoint: &str, access_key: &str, secret_key: &str) -> Result<Self, AdminClientError> {
let url = reqwest::Url::parse(endpoint).map_err(|err| AdminClientError::InvalidEndpoint(err.to_string()))?;
if url.host_str().is_none() {
return Err(AdminClientError::InvalidEndpoint("endpoint has no host".to_string()));
}
let http = reqwest::Client::builder()
.connect_timeout(Duration::from_secs(10))
.timeout(Duration::from_secs(30))
.build()
.map_err(AdminClientError::Transport)?;
Ok(Self {
endpoint: url,
access_key: access_key.to_string(),
secret_key: secret_key.to_string(),
session_token: String::new(),
region: DEFAULT_REGION.to_string(),
api_prefix: DEFAULT_ADMIN_API_PREFIX.to_string(),
http,
})
}
/// Attach an STS session token (signed as `x-amz-security-token`).
pub fn with_session_token(mut self, session_token: impl Into<String>) -> Self {
self.session_token = session_token.into();
self
}
/// Override the SigV4 region (defaults to `us-east-1`, matching a
/// region-less RustFS deployment).
pub fn with_region(mut self, region: impl Into<String>) -> Self {
self.region = region.into();
self
}
/// Override the admin API path prefix (defaults to `/rustfs/admin`).
pub fn with_api_prefix(mut self, prefix: impl Into<String>) -> Self {
self.api_prefix = prefix.into();
self
}
/// Start a heal. `bucket` empty and `prefix` empty heals the whole
/// deployment (requires `recursive` or a `pool`/`set` pair in `opts`,
/// enforced server-side); a bucket alone heals the bucket (the server
/// forces `recursive` for bucket heals).
pub async fn heal_start(
&self,
bucket: Option<&str>,
prefix: Option<&str>,
opts: &HealOpts,
force_start: bool,
) -> Result<HealStartSuccess, AdminClientError> {
let body = serde_json::to_vec(opts).map_err(|err| AdminClientError::Decode {
message: err.to_string(),
})?;
let mut query = Vec::new();
if force_start {
query.push(("forceStart", "true".to_string()));
}
self.post_json(&heal_path(bucket, prefix), &query, body).await
}
/// Query the status of the heal identified by `client_token` (the token
/// returned by [`Self::heal_start`]) at the path it was started on.
pub async fn heal_status(
&self,
bucket: Option<&str>,
prefix: Option<&str>,
client_token: &str,
) -> Result<HealTaskStatus, AdminClientError> {
self.post_json(&heal_path(bucket, prefix), &[("clientToken", client_token.to_string())], Vec::new())
.await
}
/// Stop a heal: with a `client_token` only that task is cancelled and its
/// final status returned; without one, every heal task at the path is
/// cancelled (the server answers with a start-success-shaped receipt).
pub async fn heal_stop(
&self,
bucket: Option<&str>,
prefix: Option<&str>,
client_token: Option<&str>,
) -> Result<HealStopOutcome, AdminClientError> {
let mut query = vec![("forceStop", "true".to_string())];
if let Some(token) = client_token {
query.push(("clientToken", token.to_string()));
}
match client_token {
Some(_) => {
let status: HealTaskStatus = self.post_json(&heal_path(bucket, prefix), &query, Vec::new()).await?;
Ok(HealStopOutcome::Stopped(status))
}
None => {
let success: HealStartSuccess = self.post_json(&heal_path(bucket, prefix), &query, Vec::new()).await?;
Ok(HealStopOutcome::PathStopped(success))
}
}
}
/// Cluster-aggregated background heal status.
pub async fn background_heal_status(&self) -> Result<BackgroundHealStatus, AdminClientError> {
self.get_json("/v3/background-heal/status").await
}
/// Data scanner status (enabled state, freshness, runtime config).
pub async fn scanner_status(&self) -> Result<ScannerStatus, AdminClientError> {
self.get_json("/v3/scanner/status").await
}
/// ILM expiry worker status. The payload is owned by the expiry
/// subsystem and still evolving; returned verbatim.
pub async fn ilm_expiry_status(&self) -> Result<serde_json::Value, AdminClientError> {
self.get_json("/v3/ilm/expiry/status").await
}
/// Durable replacement-recovery status (admin v4). The payload is owned
/// by the heal runtime; returned verbatim.
pub async fn replacement_recovery_status(&self) -> Result<serde_json::Value, AdminClientError> {
self.get_json("/v4/heal/replacement-recovery").await
}
/// Signed GET returning a decoded JSON body; escape hatch for endpoints
/// this client does not wrap yet.
pub async fn get_json<T: for<'de> Deserialize<'de>>(&self, path: &str) -> Result<T, AdminClientError> {
let url = self.url_for(path, &[])?;
let request = self.sign_and_build(Method::GET, url, Vec::new(), None).await?;
self.execute(request).await
}
/// Signed POST returning a decoded JSON body.
async fn post_json<T: for<'de> Deserialize<'de>>(
&self,
path: &str,
query: &[(&str, String)],
body: Vec<u8>,
) -> Result<T, AdminClientError> {
let content_type = if body.is_empty() { None } else { Some("application/json") };
let url = self.url_for(path, query)?;
let request = self.sign_and_build(Method::POST, url, body, content_type).await?;
self.execute(request).await
}
fn url_for(&self, path: &str, query: &[(&str, String)]) -> Result<reqwest::Url, AdminClientError> {
let mut url = self
.endpoint
.join(&format!("{}{}", self.api_prefix.trim_end_matches('/'), path))
.map_err(|err| AdminClientError::InvalidEndpoint(err.to_string()))?;
if !query.is_empty() {
let mut pairs = url.query_pairs_mut();
for (key, value) in query {
pairs.append_pair(key, value);
}
}
Ok(url)
}
/// Build a SigV4-signed request via the same signer the server trusts,
/// then hand the signed headers to the HTTP client. The signature covers
/// method, path, query, and an unsigned-payload marker — the same shape
/// RustFS itself sends for peer admin calls.
async fn sign_and_build(
&self,
method: Method,
url: reqwest::Url,
body: Vec<u8>,
content_type: Option<&str>,
) -> Result<reqwest::Request, AdminClientError> {
let authority = match (url.host_str(), url.port_or_known_default()) {
(Some(host), Some(port)) => format!("{host}:{port}"),
_ => return Err(AdminClientError::InvalidEndpoint("endpoint has no authority".to_string())),
};
let mut builder = http::Request::builder()
.method(method.clone())
.uri(url.as_str())
.header(http::header::HOST, &authority)
.header("x-amz-content-sha256", rustfs_signer::constants::UNSIGNED_PAYLOAD);
if let Some(content_type) = content_type {
builder = builder.header(http::header::CONTENT_TYPE, content_type);
}
let unsigned = builder
.body(s3s::Body::empty())
.map_err(|err| AdminClientError::InvalidEndpoint(format!("build request failed: {err}")))?;
let signed = rustfs_signer::sign_v4(
unsigned,
body.len() as i64,
&self.access_key,
&self.secret_key,
&self.session_token,
&self.region,
);
let mut request = self
.http
.request(method, url)
.body(body)
.build()
.map_err(AdminClientError::Transport)?;
let headers = request.headers_mut();
for (name, value) in signed.headers().iter() {
// HOST is owned by the HTTP client; the signed value above was
// built from the same URL authority, so they always agree.
if name == http::header::HOST {
continue;
}
headers.insert(name, value.clone());
}
Ok(request)
}
async fn execute<T: for<'de> Deserialize<'de>>(&self, request: reqwest::Request) -> Result<T, AdminClientError> {
let response = self.http.execute(request).await?;
let status = response.status();
let bytes = response.bytes().await?;
if !status.is_success() {
return Err(AdminClientError::HttpStatus {
status: status.as_u16(),
body: String::from_utf8_lossy(&bytes).into_owned(),
});
}
serde_json::from_slice(&bytes).map_err(|err| AdminClientError::Decode {
message: err.to_string(),
})
}
}
/// Response of [`AdminClient::heal_stop`]: cancelling a single tokened task
/// answers with that task's status, cancelling a whole path answers with a
/// start-success-shaped receipt.
#[derive(Debug, Clone)]
pub enum HealStopOutcome {
Stopped(HealTaskStatus),
PathStopped(HealStartSuccess),
}
fn heal_path(bucket: Option<&str>, prefix: Option<&str>) -> String {
match (bucket, prefix) {
(Some(bucket), Some(prefix)) if !bucket.is_empty() && !prefix.is_empty() => {
format!("/v3/heal/{}/{}", percent_encode_path_segment(bucket), percent_encode_path_segment(prefix))
}
(Some(bucket), Some(_)) | (Some(bucket), None) if !bucket.is_empty() => {
format!("/v3/heal/{}", percent_encode_path_segment(bucket))
}
_ => "/v3/heal/".to_string(),
}
}
/// Encode a single path segment (slashes are content, not separators, inside
/// bucket/prefix path params).
fn percent_encode_path_segment(segment: &str) -> String {
let mut out = String::with_capacity(segment.len());
for byte in segment.bytes() {
match byte {
b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' => out.push(byte as char),
_ => out.push_str(&format!("%{byte:02X}")),
}
}
out
}
#[cfg(test)]
mod tests {
use super::{
AdminClient, AdminClientError, BackgroundHealStatus, HealOpts, HealScanMode, HealStartSuccess, HealTaskStatus,
ScannerStatus, heal_path, percent_encode_path_segment,
};
use serde_json::json;
use std::sync::{Arc, Mutex};
#[test]
fn heal_paths_cover_root_bucket_and_prefix() {
assert_eq!(heal_path(None, None), "/v3/heal/");
assert_eq!(heal_path(Some(""), Some("")), "/v3/heal/");
assert_eq!(heal_path(Some("bucket"), None), "/v3/heal/bucket");
assert_eq!(heal_path(Some("bucket"), Some("pre/fix")), "/v3/heal/bucket/pre%2Ffix");
}
#[test]
fn path_segments_percent_encode_reserved_characters() {
assert_eq!(percent_encode_path_segment("a b"), "a%20b");
assert_eq!(percent_encode_path_segment("a/b"), "a%2Fb");
assert_eq!(percent_encode_path_segment("ü"), "%C3%BC");
}
#[test]
fn heal_opts_round_trip_through_the_server_wire_shape() {
let opts = HealOpts {
recursive: true,
dry_run: false,
remove: true,
recreate: false,
scan_mode: HealScanMode::Deep,
update_parity: true,
no_lock: false,
pool: Some(1),
set: Some(2),
};
let wire = serde_json::to_value(&opts).unwrap();
assert_eq!(wire["scanMode"], json!(2), "the server body decodes scanMode as a number");
let back: HealOpts = serde_json::from_value(wire).unwrap();
assert_eq!(back.scan_mode, HealScanMode::Deep);
assert_eq!(back.pool, Some(1));
}
#[test]
fn heal_scan_mode_accepts_both_wire_encodings() {
assert_eq!(serde_json::from_value::<HealScanMode>(json!(1)).unwrap(), HealScanMode::Normal);
assert_eq!(serde_json::from_value::<HealScanMode>(json!("deep")).unwrap(), HealScanMode::Deep);
assert!(serde_json::from_value::<HealScanMode>(json!(9)).is_err());
assert!(serde_json::from_value::<HealScanMode>(json!("sideways")).is_err());
}
#[test]
fn heal_task_status_decodes_the_server_response_shape() {
let raw = json!({
"summary": "finished",
"detail": "",
"startTime": "2026-08-17T00:00:00Z",
"settings": {"recursive": false, "scanMode": 1},
"items": [{
"resultId": 1, "type": "object", "bucket": "b", "object": "o", "versionId": "", "detail": "",
"parityBlocks": 2, "dataBlocks": 2, "diskCount": 4, "setCount": 1,
"before": {"drives": []}, "after": {"drives": []}, "objectSize": 128
}],
"truncated": false
});
let status: HealTaskStatus = serde_json::from_value(raw).unwrap();
assert_eq!(status.summary, "finished");
assert_eq!(status.items.len(), 1);
assert_eq!(status.settings.scan_mode, HealScanMode::Normal);
assert!(status.progress.is_none());
}
#[test]
fn background_heal_status_types_known_fields_and_passes_the_rest_through() {
let raw = json!({
"state": "active",
"bitrotStartTime": "t",
"healQueueLength": 3,
"healActiveTasks": 1,
"healOperations": {"queueLength": 3},
"clusterStatusComplete": true
});
let status: BackgroundHealStatus = serde_json::from_value(raw).unwrap();
assert_eq!(status.state, "active");
assert_eq!(status.heal_queue_length, 3);
assert!(status.cluster_status_complete);
assert!(status.extra.contains_key("healOperations"), "unknown nested payloads must pass through");
}
#[test]
fn scanner_status_defaults_freshness_to_unknown() {
let raw = json!({"enabled": true, "freshness": {"state": "stale"}, "metrics": {}});
let status: ScannerStatus = serde_json::from_value(raw).unwrap();
assert_eq!(status.freshness(), "stale");
let bare: ScannerStatus = serde_json::from_value(json!({"enabled": false})).unwrap();
assert_eq!(bare.freshness(), "unknown");
}
#[test]
fn invalid_endpoint_is_rejected_without_io() {
let err = AdminClient::new("not a url", "ak", "sk").unwrap_err();
assert!(matches!(err, AdminClientError::InvalidEndpoint(_)));
}
#[tokio::test]
async fn signed_requests_carry_sigv4_authorization_and_correct_target() {
let server = TestServer::spawn(r#"{"clientToken":"token-1","clientAddress":"127.0.0.1:9","startTime":"t"}"#, 200).await;
let client = AdminClient::new(&format!("http://{}", server.addr), "minioadmin", "minioadmin")
.expect("client builds against the test server");
let start: HealStartSuccess = client
.heal_start(
Some("bucket"),
None,
&HealOpts {
recursive: true,
..Default::default()
},
false,
)
.await
.expect("signed heal start decodes");
assert_eq!(start.client_token, "token-1");
let request = server.recorded();
assert_eq!(request.method, "POST");
assert_eq!(request.path, "/rustfs/admin/v3/heal/bucket");
assert!(!request.query.contains("forceStart"), "absent flags must not be sent");
let auth = request.header("authorization").expect("request must be signed");
assert!(auth.starts_with("AWS4-HMAC-SHA256"), "SigV4 scheme, got: {auth}");
assert!(auth.contains("Credential=minioadmin/"), "credentials must be in the Authorization header");
assert_eq!(
request.header("x-amz-content-sha256").as_deref(),
Some("UNSIGNED-PAYLOAD"),
"the client signs the same payload marker RustFS peer calls use"
);
assert_eq!(request.header("content-type").as_deref(), Some("application/json"));
assert!(request.body.contains("\"recursive\":true"));
}
#[tokio::test]
async fn query_sends_client_token_on_the_same_path() {
let body = r#"{"summary":"running","detail":"","settings":{"recursive":false},"items":[],"truncated":false}"#;
let server = TestServer::spawn(body, 200).await;
let client = AdminClient::new(&format!("http://{}", server.addr), "ak", "sk").unwrap();
let status = client
.heal_status(Some("bucket"), None, "token-1")
.await
.expect("status decodes");
assert_eq!(status.summary, "running");
let request = server.recorded();
assert_eq!(request.path, "/rustfs/admin/v3/heal/bucket");
assert!(request.query.contains("clientToken=token-1"));
assert!(!request.query.contains("forceStop"));
}
#[tokio::test]
async fn stop_without_token_takes_the_path_cancel_branch() {
let server = TestServer::spawn(r#"{"clientToken":"path","clientAddress":"c","startTime":"t"}"#, 200).await;
let client = AdminClient::new(&format!("http://{}", server.addr), "ak", "sk").unwrap();
let outcome = client.heal_stop(Some("bucket"), None, None).await.expect("path stop decodes");
assert!(matches!(outcome, super::HealStopOutcome::PathStopped(_)));
let request = server.recorded();
assert!(request.query.contains("forceStop=true"));
assert!(!request.query.contains("clientToken"));
}
#[tokio::test]
async fn http_error_status_maps_to_a_typed_error_with_body() {
let server = TestServer::spawn(r#"{"code":"AccessDenied","message":"denied"}"#, 403).await;
let client = AdminClient::new(&format!("http://{}", server.addr), "ak", "sk").unwrap();
let err = client.scanner_status().await.unwrap_err();
match err {
AdminClientError::HttpStatus { status, body } => {
assert_eq!(status, 403);
assert!(body.contains("AccessDenied"));
}
other => panic!("expected HttpStatus, got {other:?}"),
}
}
#[tokio::test]
async fn malformed_success_body_maps_to_a_decode_error() {
let server = TestServer::spawn("not json", 200).await;
let client = AdminClient::new(&format!("http://{}", server.addr), "ak", "sk").unwrap();
assert!(matches!(client.scanner_status().await.unwrap_err(), AdminClientError::Decode { .. }));
}
/// One recorded request, parsed off the wire with the minimum needed for
/// assertions: method, path, query, headers, body.
#[derive(Debug, Clone)]
struct RecordedRequest {
method: String,
path: String,
query: String,
headers: Vec<(String, String)>,
body: String,
}
impl RecordedRequest {
fn header(&self, name: &str) -> Option<String> {
self.headers
.iter()
.find(|(key, _)| key.eq_ignore_ascii_case(name))
.map(|(_, value)| value.clone())
}
}
/// Minimal HTTP/1.1 server: one canned response per connection, every
/// request recorded behind an `Arc<Mutex>`. Deliberately dependency-free —
/// the assertions only need the raw request bytes.
struct TestServer {
addr: std::net::SocketAddr,
requests: Arc<Mutex<Vec<RecordedRequest>>>,
}
impl TestServer {
async fn spawn(response_body: &'static str, status: u16) -> Self {
use tokio::io::{AsyncReadExt, AsyncWriteExt};
let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
.await
.expect("bind ephemeral port");
let addr = listener.local_addr().expect("local addr");
let requests: Arc<Mutex<Vec<RecordedRequest>>> = Arc::new(Mutex::new(Vec::new()));
let recorded = requests.clone();
tokio::spawn(async move {
let reason = if status == 200 { "OK" } else { "Forbidden" };
let response = format!(
"HTTP/1.1 {status} {reason}\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{response_body}",
response_body.len()
);
// Each request is a fresh connection (connection: close); a
// bounded loop serves every call a test makes while letting
// the task exit instead of lingering for the whole process.
for _ in 0..16 {
let Ok((mut stream, _)) = listener.accept().await else {
break;
};
let mut buffer = Vec::with_capacity(2048);
let mut chunk = [0u8; 2048];
// Read headers plus content-length body, or stop on close.
loop {
if let Some(end) = find_header_end(&buffer) {
let content_length = extract_content_length(&buffer[..end]);
if buffer.len() >= end + content_length {
break;
}
}
let n = match stream.read(&mut chunk).await {
Ok(0) | Err(_) => break,
Ok(n) => n,
};
buffer.extend_from_slice(&chunk[..n]);
if buffer.len() > 64 * 1024 {
break;
}
}
if let Some(request) = parse_request(&buffer) {
recorded.lock().expect("recorded lock").push(request);
}
let _ = stream.write_all(response.as_bytes()).await;
let _ = stream.shutdown().await;
}
});
Self { addr, requests }
}
fn recorded(&self) -> RecordedRequest {
self.requests
.lock()
.expect("recorded lock")
.last()
.cloned()
.expect("the client call must have produced one recorded request")
}
}
fn find_header_end(buffer: &[u8]) -> Option<usize> {
buffer.windows(4).position(|window| window == b"\r\n\r\n").map(|pos| pos + 4)
}
fn extract_content_length(headers: &[u8]) -> usize {
let text = String::from_utf8_lossy(headers).to_ascii_lowercase();
text.lines()
.find_map(|line| line.strip_prefix("content-length:"))
.and_then(|value| value.trim().parse().ok())
.unwrap_or(0)
}
fn parse_request(raw: &[u8]) -> Option<RecordedRequest> {
let end = find_header_end(raw)?;
let head = String::from_utf8_lossy(&raw[..end]);
let body = String::from_utf8_lossy(&raw[end..]).into_owned();
let mut lines = head.lines();
let request_line = lines.next()?;
let mut parts = request_line.split_whitespace();
let method = parts.next()?.to_string();
let target = parts.next()?.to_string();
let (path, query) = match target.split_once('?') {
Some((path, query)) => (path.to_string(), query.to_string()),
None => (target, String::new()),
};
let headers = lines
.filter_map(|line| line.split_once(':'))
.map(|(name, value)| (name.trim().to_string(), value.trim().to_string()))
.collect();
Some(RecordedRequest {
method,
path,
query,
headers,
body,
})
}
}
+2
View File
@@ -12,6 +12,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.
pub mod client;
pub mod group;
pub mod heal_commands;
pub mod health;
@@ -25,6 +26,7 @@ pub mod trace;
pub mod user;
pub mod utils;
pub use client::*;
pub use group::*;
pub use info_commands::*;
pub use policy::*;
+3
View File
@@ -108,6 +108,9 @@ temp-env = { workspace = true }
tempfile = { workspace = true }
uuid = { workspace = true, features = ["v4", "serde", "fast-rng", "macro-diagnostics"] }
tokio = { workspace = true, features = ["test-util", "fs", "rt-multi-thread"] }
# Test-only: pins the emitted scanner alert wire names against the canonical
# EventName string forms subscribers configure (rustfs/backlog#1868).
rustfs-s3-types.workspace = true
# Enables the shared MockWarmBackend / xl.meta assertion helpers exposed via
# the ecstore `api::tier::test_util` facade module (rustfs/backlog#1148 ilm-6).
rustfs-ecstore = { workspace = true, features = ["test-util"] }
+225 -4
View File
@@ -12,10 +12,10 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use std::collections::HashSet;
use std::collections::{HashMap, HashSet};
use std::fs::FileType;
use std::io::ErrorKind;
use std::sync::{Arc, Once};
use std::sync::{Arc, Mutex, Once};
use std::time::{Duration, Instant, SystemTime};
use crate::ReplTargetSizeSummary;
@@ -32,6 +32,7 @@ use crate::scanner_io::{
SCANNER_SKIP_FILE_ERROR, ScannerIODisk as _, is_scanner_metadata_corrupt_error, is_scanner_metadata_transient_error,
};
use crate::sleeper::DynamicSleeper;
use crate::storage_api::owner::{EcstoreEventArgs, ecstore_send_event};
use metrics::{counter, describe_counter};
use rustfs_common::heal_channel::{
HEAL_DELETE_DANGLING, HealAdmissionDropReason, HealAdmissionResult, HealChannelPriority, HealChannelRequest,
@@ -97,6 +98,101 @@ const METRIC_SCANNER_EXCESS_FOLDERS_TOTAL: &str = "rustfs_scanner_excess_folders
const METRIC_SCANNER_PENDING_HEAL_PRUNE_TOTAL: &str = "rustfs_scanner_pending_heal_prune_total";
const METRIC_SCANNER_PENDING_HEAL_MALFORMED_TOTAL: &str = "rustfs_scanner_pending_heal_malformed_total";
const MAX_PENDING_SCANNER_HEAL_RETRIES_PER_BUCKET: usize = 128;
// --- scanner excess alerts as S3 notification events (rustfs/backlog#1868) --
//
// The excess-versions / excess-version-size / excess-folders alerts were
// metrics-and-logs only; subscribers (consoles, external auditors) had no way
// to hear them. MinIO emits s3:ObjectManyVersions / s3:ObjectLargeVersions /
// s3:PrefixManyFolders for the same conditions — RustFS carries those as
// EventName::Scanner* with the wire names below. Without a cooldown a single
// over-threshold object would re-emit on every scan cycle (~a minute), so
// emissions are edge-held per (kind, bucket, object) for 24h.
/// `s3:Scanner:ManyVersions` (MinIO `s3:ObjectManyVersions`).
pub const EVENT_SCANNER_MANY_VERSIONS: &str = "s3:Scanner:ManyVersions";
/// `s3:Scanner:LargeVersions` (MinIO `s3:ObjectLargeVersions`).
pub const EVENT_SCANNER_LARGE_VERSIONS: &str = "s3:Scanner:LargeVersions";
/// `s3:Scanner:BigPrefix` (MinIO `s3:PrefixManyFolders`).
pub const EVENT_SCANNER_BIG_PREFIX: &str = "s3:Scanner:BigPrefix";
const ENV_SCANNER_ALERT_COOLDOWN_SECS: &str = "RUSTFS_SCANNER_ALERT_COOLDOWN_SECS";
const DEFAULT_SCANNER_ALERT_COOLDOWN_SECS: u64 = 86_400;
/// Hard cap on distinct cooldown keys; a pathological number of over-threshold
/// objects clears the map wholesale instead of growing without bound (the
/// worst case is one re-emission per still-hot key per scan cycle).
const MAX_SCANNER_ALERT_COOLDOWN_KEYS: usize = 4096;
/// Distinct alert kinds sharing one cooldown map.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
enum ScannerAlertKind {
ManyVersions,
LargeVersions,
BigPrefix,
}
type ScannerAlertCooldownKey = (ScannerAlertKind, String, String);
type ScannerAlertCooldownMap = HashMap<ScannerAlertCooldownKey, Instant>;
static SCANNER_ALERT_EMISSION_COOLDOWN: Mutex<Option<ScannerAlertCooldownMap>> = Mutex::new(None);
fn scanner_alert_cooldown() -> Duration {
let raw = std::env::var(ENV_SCANNER_ALERT_COOLDOWN_SECS)
.ok()
.and_then(|v| v.parse::<u64>().ok());
Duration::from_secs(raw.unwrap_or(DEFAULT_SCANNER_ALERT_COOLDOWN_SECS))
}
/// Edge-held emission gate: returns `true` (and records the cooldown) only
/// when this (kind, bucket, object) last fired longer than the cooldown ago —
/// or never. Metrics and logs stay level-triggered every cycle; only the
/// notification events are held back.
fn scanner_alert_emission_allows(kind: ScannerAlertKind, bucket: &str, object: &str, cooldown: Duration) -> bool {
let key = (kind, bucket.to_string(), object.to_string());
let mut guard = SCANNER_ALERT_EMISSION_COOLDOWN
.lock()
.unwrap_or_else(|poison| poison.into_inner());
let guard = guard.get_or_insert_with(ScannerAlertCooldownMap::new);
let now = Instant::now();
// Expired entries leave first; the cap is still exceeded only when live
// keys alone overflow it, in which case a wholesale clear trades one
// extra emission per hot key for a hard memory bound.
if guard.len() >= MAX_SCANNER_ALERT_COOLDOWN_KEYS {
guard.retain(|_, fired_at| now.duration_since(*fired_at) < cooldown);
if guard.len() >= MAX_SCANNER_ALERT_COOLDOWN_KEYS {
guard.clear();
}
}
match guard.get(&key) {
Some(fired_at) if now.duration_since(*fired_at) < cooldown => false,
_ => {
guard.insert(key, now);
true
}
}
}
/// Emit a scanner alert as an S3 notification event through the standard
/// dispatch pipeline. Fire-and-forget: the notify layer owns delivery,
/// retry, and target filtering; the scanner never waits on it.
fn emit_scanner_alert_event(event_name: &str, bucket: &str, object: &str, size: i64, details: &[(&str, String)]) {
let mut req_params = HashMap::with_capacity(details.len());
for (key, value) in details {
req_params.insert((*key).to_string(), value.clone());
}
ecstore_send_event(EcstoreEventArgs {
event_name: event_name.to_string(),
bucket_name: bucket.to_string(),
object: crate::ScannerObjectInfo {
bucket: bucket.to_string(),
name: object.to_string(),
size,
..Default::default()
},
req_params,
user_agent: "Scanner".to_string(),
..Default::default()
});
}
const MAX_PENDING_SCANNER_HEALS_PER_BUCKET: usize = 10_000;
static SCANNER_INLINE_HEAL_WARN_ONCE: Once = Once::new();
@@ -1197,6 +1293,7 @@ impl ScannerItem {
fn alert_excessive_versions(&self, remaining_versions: usize, cumulative_size: i64) {
ensure_scanner_alert_metrics_registered();
let (too_many_versions, too_large_versions) = should_alert_excessive_versions(remaining_versions, cumulative_size);
let object_path = self.object_path();
if too_many_versions {
global_metrics().record_scanner_source_executed(ScannerWorkSource::Alerts, 1);
counter!(
@@ -1204,13 +1301,26 @@ impl ScannerItem {
"bucket" => self.bucket.clone()
)
.increment(1);
if scanner_alert_emission_allows(ScannerAlertKind::ManyVersions, &self.bucket, &object_path, scanner_alert_cooldown())
{
emit_scanner_alert_event(
EVENT_SCANNER_MANY_VERSIONS,
&self.bucket,
&object_path,
cumulative_size,
&[
("versions", remaining_versions.to_string()),
("threshold", scanner_excess_versions_threshold().to_string()),
],
);
}
warn!(
target: "rustfs::scanner::folder",
event = EVENT_SCANNER_ALERT_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_FOLDER,
bucket = %self.bucket,
object = %self.object_path(),
object = %object_path,
versions = remaining_versions,
threshold = scanner_excess_versions_threshold(),
state = "excess_versions",
@@ -1224,13 +1334,31 @@ impl ScannerItem {
"bucket" => self.bucket.clone()
)
.increment(1);
if scanner_alert_emission_allows(
ScannerAlertKind::LargeVersions,
&self.bucket,
&object_path,
scanner_alert_cooldown(),
) {
emit_scanner_alert_event(
EVENT_SCANNER_LARGE_VERSIONS,
&self.bucket,
&object_path,
cumulative_size,
&[
("versions", remaining_versions.to_string()),
("cumulativeSize", cumulative_size.to_string()),
("threshold", scanner_excess_version_size_threshold().to_string()),
],
);
}
warn!(
target: "rustfs::scanner::folder",
event = EVENT_SCANNER_ALERT_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_FOLDER,
bucket = %self.bucket,
object = %self.object_path(),
object = %object_path,
versions = remaining_versions,
cumulative_size,
threshold = scanner_excess_version_size_threshold(),
@@ -1611,6 +1739,15 @@ impl FolderScanner {
"root" => self.root.clone()
)
.increment(1);
if scanner_alert_emission_allows(ScannerAlertKind::BigPrefix, &self.root, folder, scanner_alert_cooldown()) {
emit_scanner_alert_event(
EVENT_SCANNER_BIG_PREFIX,
&self.root,
folder,
0,
&[("folders", total_folders.to_string()), ("threshold", threshold.to_string())],
);
}
warn!(
target: "rustfs::scanner::folder",
event = EVENT_SCANNER_ALERT_STATE,
@@ -3076,6 +3213,90 @@ mod tests {
#[cfg(unix)]
use std::os::unix::fs::{PermissionsExt, symlink};
use std::sync::Mutex;
/// Reset the process-global alert cooldown map; test-only.
fn reset_alert_cooldowns() {
*SCANNER_ALERT_EMISSION_COOLDOWN
.lock()
.unwrap_or_else(|poison| poison.into_inner()) = Some(ScannerAlertCooldownMap::new());
}
/// The emitted event-name strings must be exactly what `EventName`
/// serializes, or a bucket notification subscribed to the documented name
/// would silently never match (rustfs/backlog#1868).
#[test]
fn scanner_alert_wire_names_match_canonical_event_names() {
use rustfs_s3_types::EventName;
assert_eq!(EVENT_SCANNER_MANY_VERSIONS, EventName::ScannerManyVersions.to_string());
assert_eq!(EVENT_SCANNER_LARGE_VERSIONS, EventName::ScannerLargeVersions.to_string());
assert_eq!(EVENT_SCANNER_BIG_PREFIX, EventName::ScannerBigPrefix.to_string());
}
fn cooldown_map_len() -> usize {
SCANNER_ALERT_EMISSION_COOLDOWN
.lock()
.unwrap_or_else(|poison| poison.into_inner())
.as_ref()
.map(|map| map.len())
.unwrap_or(0)
}
/// Backdate every recorded cooldown so the next check fires again.
fn expire_all_alert_cooldowns(cooldown: Duration) {
let now = Instant::now();
let mut guard = SCANNER_ALERT_EMISSION_COOLDOWN
.lock()
.unwrap_or_else(|poison| poison.into_inner());
if let Some(map) = guard.as_mut() {
for fired_at in map.values_mut() {
if let Some(expired) = now.checked_sub(cooldown + Duration::from_secs(1)) {
*fired_at = expired;
}
}
}
}
/// The emission gate is the only thing standing between an over-threshold
/// object and one S3 event per scan cycle, so its edge semantics get
/// pinned directly. All scenarios share one #[test] because the cooldown
/// map is process-global and parallel tests would read each other's
/// firings.
#[test]
fn scanner_alert_emission_is_edge_held_per_key_and_bounded() {
reset_alert_cooldowns();
let cooldown = Duration::from_secs(3600);
// First firing allows, an immediate re-check is held.
assert!(scanner_alert_emission_allows(ScannerAlertKind::ManyVersions, "bkt", "obj", cooldown));
assert!(!scanner_alert_emission_allows(ScannerAlertKind::ManyVersions, "bkt", "obj", cooldown));
// Different kind, object, and bucket are independent keys.
assert!(scanner_alert_emission_allows(ScannerAlertKind::LargeVersions, "bkt", "obj", cooldown));
assert!(scanner_alert_emission_allows(ScannerAlertKind::ManyVersions, "bkt", "other", cooldown));
assert!(scanner_alert_emission_allows(ScannerAlertKind::ManyVersions, "other", "obj", cooldown));
assert_eq!(cooldown_map_len(), 4);
// After the cooldown elapses the same key fires again.
expire_all_alert_cooldowns(cooldown);
assert!(scanner_alert_emission_allows(ScannerAlertKind::ManyVersions, "bkt", "obj", cooldown));
// A zero cooldown degenerates to always-emit (operators may want that).
assert!(scanner_alert_emission_allows(ScannerAlertKind::BigPrefix, "bkt", "dir", Duration::ZERO));
assert!(scanner_alert_emission_allows(ScannerAlertKind::BigPrefix, "bkt", "dir", Duration::ZERO));
// Hard bound: overflow the cap with zero-cooldown keys and confirm the
// map clears rather than growing past it.
reset_alert_cooldowns();
for index in 0..=(MAX_SCANNER_ALERT_COOLDOWN_KEYS + 8) {
let _ = scanner_alert_emission_allows(ScannerAlertKind::BigPrefix, "bkt", &format!("dir-{index}"), Duration::ZERO);
}
assert!(
cooldown_map_len() <= MAX_SCANNER_ALERT_COOLDOWN_KEYS,
"cooldown map must stay bounded, got {}",
cooldown_map_len()
);
}
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use temp_env::{with_var, with_var_unset};
use tracing_subscriber::fmt::MakeWriter;
+4 -3
View File
@@ -78,6 +78,7 @@ pub(crate) use rustfs_ecstore::api::disk::{
pub(crate) use rustfs_ecstore::api::error::{
Error as EcstoreErrorType, Result as EcstoreResultType, StorageError as EcstoreStorageError,
};
pub(crate) use rustfs_ecstore::api::event::{EventArgs as EcstoreEventArgs, send_event as ecstore_send_event};
#[cfg(test)]
pub(crate) use rustfs_ecstore::api::layout::{
EndpointServerPools as EcstoreEndpointServerPools, Endpoints as EcstoreEndpoints, PoolEndpoints as EcstorePoolEndpoints,
@@ -110,8 +111,8 @@ pub(crate) mod owner {
ECSTORE_BUCKET_META_PREFIX, ECSTORE_RUSTFS_META_BUCKET, ECSTORE_STORAGE_FORMAT_FILE, ECSTORE_STORAGECLASS_RRS,
ECSTORE_STORAGECLASS_STANDARD, ECSTORE_TRANSITION_COMPLETE, EcstoreBucketTargetSys, EcstoreBucketVersioningSys,
EcstoreDisk, EcstoreDiskAPI, EcstoreDiskBytes, EcstoreDiskError, EcstoreDiskInfo, EcstoreDiskInfoOptions,
EcstoreDiskLocation, EcstoreDiskResult, EcstoreErrorType, EcstoreEvaluator, EcstoreEvent, EcstoreLcEventSrc,
EcstoreLifecycle, EcstoreListPathRawOptions, EcstoreNsScannerOpenRequest, EcstoreObjectOpts,
EcstoreDiskLocation, EcstoreDiskResult, EcstoreErrorType, EcstoreEvaluator, EcstoreEvent, EcstoreEventArgs,
EcstoreLcEventSrc, EcstoreLifecycle, EcstoreListPathRawOptions, EcstoreNsScannerOpenRequest, EcstoreObjectOpts,
EcstoreReplicationConfigurationExt, EcstoreReplicationScannerBridge, EcstoreResultType, EcstoreScanGuard,
EcstoreSetDisks, EcstoreStorageError, EcstoreStore, EcstoreTierConfig, EcstoreVersioningApi,
ScannerReplicationHealObject, ScannerReplicationHealResult, ScannerReplicationQueueAdmission, ecstore_apply_expiry_rule,
@@ -121,7 +122,7 @@ pub(crate) mod owner {
ecstore_is_erasure_sd, ecstore_is_reserved_or_invalid_bucket, ecstore_list_path_raw,
ecstore_object_opts_from_object_info, ecstore_path2_bucket_object, ecstore_path2_bucket_object_with_base_path,
ecstore_read_config, ecstore_replace_bucket_usage_memory_from_info, ecstore_resolve_object_store_handle,
ecstore_save_config, scanner_replication_config_for_lifecycle_eval,
ecstore_save_config, ecstore_send_event, scanner_replication_config_for_lifecycle_eval,
};
#[cfg(test)]
+37
View File
@@ -0,0 +1,37 @@
# Scanner Excess Alerts: Metrics, S3 Events, and Thresholds
> 中文版:[scanner-excess-alerts_zh.md](scanner-excess-alerts_zh.md)
Date: 2026-08-18 (rustfs/backlog#1868 / HS-04; includes the HS-15 threshold-delta notes)
The background scanner detects three classes of "excess" conditions while it walks buckets and surfaces them as alerts. This page documents each alert's trigger condition, the subscribable S3 event, the cooldown semantics, and the threshold differences versus MinIO — for operators debugging alerts and for event consumers wiring up subscriptions.
## The three alerts
| Alert | Trigger (per scan cycle) | Metric | S3 event (RustFS wire name) | MinIO event name |
|---|---|---|---|---|
| Excess versions | Retained versions of one object ≥ `scanner:alert_excess_versions` | `rustfs_scanner_excess_object_versions_total{bucket}` | `s3:Scanner:ManyVersions` | `s3:ObjectManyVersions` |
| Excess version size | Cumulative bytes of all versions of one object ≥ `scanner:alert_excess_version_size` | `rustfs_scanner_excess_object_version_size_total{bucket}` | `s3:Scanner:LargeVersions` | `s3:ObjectLargeVersions` |
| Excess folders | Direct subfolders of one directory > `scanner:alert_excess_folders` | `rustfs_scanner_excess_folders_total{root}` | `s3:Scanner:BigPrefix` | `s3:PrefixManyFolders` |
Subscribe like any bucket notification: configure a notification on the target bucket with the RustFS wire name above (or the `s3:Scanner:*` wildcard). Events carry `UserAgent: Scanner` as their origin marker, and `req_params` holds the observed value and the threshold (`versions` / `cumulativeSize` / `folders` / `threshold`), so consumers can judge severity directly.
## Metrics and events fire on different cadences
- **Metrics and structured logs are level-triggered**: as long as the object stays over the threshold, every scan cycle counts and logs it (default cycle ≈ 60s; see `scanner:speed`).
- **S3 events are edge-triggered with a cooldown**: the same (alert kind, bucket, object) emits at most once per cooldown window — 24 hours by default (`RUSTFS_SCANNER_ALERT_COOLDOWN_SECS`; set it to 0 to emit every cycle). When the window lapses and the object is still over the threshold, the event fires again. The cooldown table lives in process memory with a 4096-entry hard cap; on overflow it is cleared and rebuilt (worst case: one extra emission per still-hot key).
- A process restart resets the cooldown (every still-over-threshold object emits once more after a restart) — deliberately: restarts usually accompany incident response, and the re-emission buys visibility.
## Threshold defaults and the MinIO deltas (HS-15)
| Config key | ENV | RustFS default | MinIO default | Notes |
|---|---|---|---|---|
| `scanner:alert_excess_versions` | `RUSTFS_SCANNER_ALERT_EXCESS_VERSIONS` | 100 | 100 | Identical |
| `scanner:alert_excess_version_size` | `RUSTFS_SCANNER_ALERT_EXCESS_VERSION_SIZE` | 1 TiB | 1 TB | Same order of magnitude; different unit basis (TiB vs TB) |
| `scanner:alert_excess_folders` | `RUSTFS_SCANNER_ALERT_EXCESS_FOLDERS` | 65538 | 50000 | **Deliberate divergence**: 65538 tolerates the Proxmox Backup Server chunk layout (65536 chunks per directory plus the directory's own entries); MinIO's 50000 would fire continuously for PBS users. Set it to 50000 explicitly to match MinIO behavior |
All three keys accept both env and admin config (`PUT /rustfs/admin/v3/config`, `scanner` subsystem); hot updates take effect immediately.
## Why the event names are mapped
RustFS's event enum (`rustfs_s3_types::EventName::ScannerManyVersions/LargeVersions/BigPrefix`) keeps the repo's established `s3:Scanner:*` wire names (literally different from MinIO's `s3:ObjectManyVersions`; the enum comments preserve the mapping). Subscribers should use the RustFS wire names in this page. If you need MinIO-literal compatibility, map the names on the console/consumer side — do not change the published wire names.
@@ -0,0 +1,37 @@
# Scanner 超限告警:指标、S3 事件与阈值
> English version: [scanner-excess-alerts.md](scanner-excess-alerts.md)
日期:2026-08-18rustfs/backlog#1868 / HS-04,含 HS-15 阈值差异说明)
后台 scanner 在扫描过程中检测三类"超限"状态并对外告警。本文说明每类告警的触发条件、可订阅的 S3 事件、冷却语义,以及与 MinIO 的阈值差异,供运维排障与事件消费方对接。
## 三类告警
| 告警 | 触发条件(任一扫描周期) | 指标 | S3 事件(RustFS wire 名) | MinIO 对应事件名 |
|---|---|---|---|---|
| 版本数超限 | 单对象保留版本数 ≥ `scanner:alert_excess_versions` | `rustfs_scanner_excess_object_versions_total{bucket}` | `s3:Scanner:ManyVersions` | `s3:ObjectManyVersions` |
| 版本总大小超限 | 单对象全部版本累计字节 ≥ `scanner:alert_excess_version_size` | `rustfs_scanner_excess_object_version_size_total{bucket}` | `s3:Scanner:LargeVersions` | `s3:ObjectLargeVersions` |
| 子目录数超限 | 单目录直接子目录数 > `scanner:alert_excess_folders` | `rustfs_scanner_excess_folders_total{root}` | `s3:Scanner:BigPrefix` | `s3:PrefixManyFolders` |
订阅方式与普通桶通知一致:对目标桶配置 notification,事件名填上表 RustFS wire 名(或通配 `s3:Scanner:*`)。事件以 `UserAgent: Scanner` 标记来源,`req_params` 携带实际值与阈值(`versions` / `cumulativeSize` / `folders` / `threshold`),便于消费方直接判断严重程度。
## 指标与事件的触发节奏不同
- **指标与结构化日志是电平触发**:只要对象仍在阈值之上,每个扫描周期都会计数/打日志(默认周期约 60s,见 `scanner:speed`)。
- **S3 事件是边沿触发 + 冷却**:同一 (告警类型, 桶, 对象) 在冷却窗口内只发一次,默认 24 小时(`RUSTFS_SCANNER_ALERT_COOLDOWN_SECS`,设 0 表示每周期都发)。窗口过后对象仍超限会再次发出。冷却表在进程内有 4096 条硬顶,超限清空重建(最坏情况是每个仍超限的 key 多发一次)。
- 进程重启会重置冷却(重启后每个仍超限的对象会再发一次)——这是有意为之:重启常伴随排障,重发提供可见性。
## 阈值默认值与 MinIO 差异(HS-15
| 配置键 | ENV | RustFS 默认 | MinIO 默认 | 差异说明 |
|---|---|---|---|---|
| `scanner:alert_excess_versions` | `RUSTFS_SCANNER_ALERT_EXCESS_VERSIONS` | 100 | 100 | 一致 |
| `scanner:alert_excess_version_size` | `RUSTFS_SCANNER_ALERT_EXCESS_VERSION_SIZE` | 1 TiB | 1 TB | 语义同量级,单位口径不同(TiB vs TB) |
| `scanner:alert_excess_folders` | `RUSTFS_SCANNER_ALERT_EXCESS_FOLDERS` | 65538 | 50000 | **有意差异**65538 兼容 Proxmox Backup Server 的 chunk 布局(每目录 65536 个 chunk + 目录自身条目),按 MinIO 的 50000 会对 PBS 用户持续误报。如需与 MinIO 行为一致可显式配置为 50000 |
三个键均支持 env 与 admin config`PUT /rustfs/admin/v3/config``scanner` 子系统)双通道,热更新即时生效。
## 事件名映射的由来
RustFS 的事件枚举(`rustfs_s3_types::EventName::ScannerManyVersions/LargeVersions/BigPrefix`)沿用仓库既有 wire 名 `s3:Scanner:*`(与 MinIO 的 `s3:ObjectManyVersions` 字面不同,枚举注释中保留了映射关系)。订阅方应以本文的 RustFS wire 名为准;如需 MinIO 字面兼容,请在 console/消费侧做名称映射,不要修改已发布的 wire 名。
+456 -267
View File
@@ -66,20 +66,18 @@ use rustfs_config::{
};
use rustfs_iam::error::is_err_no_such_service_account;
use rustfs_iam::federation::OIDC_VIRTUAL_PARENT_CLAIM;
use rustfs_iam::store::object::ObjectStore;
use rustfs_iam::store::{MappedPolicy, UserType, sr_wire_user_type, user_type_from_sr_wire};
use rustfs_iam::sys::{
IamSys, NewServiceAccountOpts, SITE_REPLICATOR_SERVICE_ACCOUNT, UpdateServiceAccountOpts, get_claims_from_token_with_secret,
NewServiceAccountOpts, SITE_REPLICATOR_SERVICE_ACCOUNT, UpdateServiceAccountOpts, get_claims_from_token_with_secret,
};
use rustfs_madmin::{
AddOrUpdateUserReq, BucketBandwidth, GroupAddRemove, GroupStatus, IDPSettings, InProgressMetric, InQueueMetric,
LDAPConfigSettings, LDAPSettings, OpenIDProviderSettings, PeerInfo, PeerSite, QStat, ReplProxyMetric, ReplicateAddStatus,
ReplicateEditStatus, ReplicateRemoveStatus, ResyncBucketStatus, SITE_REPL_API_VERSION, SR_IAM_ITEM_STS_ACC,
SR_IAM_ITEM_STS_ACC_LEGACY, SRBucketInfo, SRBucketMeta, SRBucketStatsSummary, SRGroupInfo, SRGroupStatsSummary, SRIAMItem,
SRIAMPolicy, SRIAMUser, SRILMExpiryStatsSummary, SRInfo, SRMetric, SRMetricsSummary, SRPeerError, SRPeerJoinReq,
SRPendingOperation, SRPolicyMapping, SRPolicyStatsSummary, SRRemoveReq, SRResyncOpStatus, SRRetryStats, SRSTSCredential,
SRSessionPolicy, SRSiteSummary, SRStateEditReq, SRStateInfo, SRStatusInfo, SRSvcAccChange, SRSvcAccCreate,
SRUserStatsSummary, SiteReplicationInfo, SyncStatus, WorkerStat,
SRIAMPolicy, SRILMExpiryStatsSummary, SRInfo, SRMetric, SRMetricsSummary, SRPeerError, SRPeerJoinReq, SRPendingOperation,
SRPolicyMapping, SRPolicyStatsSummary, SRRemoveReq, SRResyncOpStatus, SRRetryStats, SRSessionPolicy, SRSiteSummary,
SRStateEditReq, SRStateInfo, SRStatusInfo, SRSvcAccCreate, SRUserStatsSummary, SiteReplicationInfo, SyncStatus, WorkerStat,
};
use rustfs_policy::policy::{
Policy,
@@ -4586,10 +4584,7 @@ async fn build_metrics_summary(local_peer: &PeerInfo) -> SRMetricsSummary {
head_failed_total: non_negative_u64(node.proxy_head_failed),
put_tag_total: non_negative_u64(node.proxy_put_tag_total),
put_tag_failed_total: non_negative_u64(node.proxy_put_tag_failed),
get_tag_total: non_negative_u64(node.proxy_get_tag_total),
get_tag_failed_total: non_negative_u64(node.proxy_get_tag_failed),
remove_tag_total: non_negative_u64(node.proxy_delete_tag_total),
remove_tag_failed_total: non_negative_u64(node.proxy_delete_tag_failed),
..Default::default()
},
metrics,
uptime: node.uptime,
@@ -6028,7 +6023,10 @@ fn edit_generation_wall_clock() -> u64 {
/// node's clock behind the clock that fed the previous lifetime) mints
/// below the stale mark and the origin stays fenced — but only until real
/// time passes the previous lifetime's last allocation, because every later
/// allocation takes the wall-clock floor again. Bounded by the skew,
/// allocation takes the wall-clock floor again (and never longer than
/// [`PEER_EDIT_FENCE_STALENESS_WINDOW_NANOS`]: a regression past the window
/// leaves the mark implausibly distant and the origin runs unfenced
/// immediately). Bounded by the skew,
/// self-healing, and no rollback window beyond the plain counter's: a
/// delivery applies only at or above the receiver's mark, so the one
/// cross-lifetime interleaving that can apply stale content — a
@@ -6068,6 +6066,52 @@ fn peer_edit_fence(queries: &HashMap<String, String>) -> Option<(String, u64)> {
Some((origin.clone(), generation))
}
/// How far below the recorded high-water mark a delivery may sit and still
/// be fenced as stale. The distance a GENUINE superseded delivery can trail
/// its origin's mark is small: retransmissions re-run the sender flow and
/// mint a fresh generation (the retry queue keys on the bare path and never
/// replays a fenced URL), so only an in-flight straggler of the losing
/// fan-out race trails the mark, by delivery latency — minutes at the
/// outside. A mark further above than this window cannot be explained by
/// any genuine race, only by a forged fence (the shared service account
/// lets any peer stamp any origin) or by a persisted clock excursion the
/// origin has since left behind — and fencing on it would silently drop the
/// origin's real edits, so the stale check ignores it instead.
const PEER_EDIT_FENCE_STALENESS_WINDOW_NANOS: u64 = 24 * 60 * 60 * 1_000_000_000;
/// Whether an incoming fence may be honoured, as far as this site can vouch
/// for it. The sender's identity is unverifiable (shared service account),
/// so the check runs over what the receiving state knows: the claimed origin
/// must be a site this state currently replicates with — the same membership
/// rule the load-time mark pruning applies, so every mark recorded behind
/// this check is one a reload would keep — and not this site itself, which
/// never delivers edits to itself. The caller IGNORES an inadmissible fence
/// rather than failing the request: the delivery applies exactly as an
/// unstamped (pre-fence) delivery would, no high-water mark is read or
/// written, and the worst a forged fence achieves is forfeiting an ordering
/// guarantee its sender was never owed. The generation itself is NOT
/// bounded here: a genuine origin whose hybrid clock persisted a wall-clock
/// excursion allocates arbitrarily far in the future, and refusing to
/// record its marks would strip the ordering fence from exactly the
/// deliveries that still race — the staleness window on the read side is
/// what defuses forged marks instead.
fn peer_edit_fence_is_admissible(state: &SiteReplicationState, local_deployment_id: &str, fence: &(String, u64)) -> bool {
let (origin, generation) = fence;
if origin != local_deployment_id && state.peers.contains_key(origin) {
return true;
}
warn!(
event = EVENT_ADMIN_SITE_REPLICATION_STATE,
component = LOG_COMPONENT_ADMIN,
subsystem = LOG_SUBSYSTEM_SITE_REPLICATION,
result = "fence_origin_not_a_remote_peer",
origin = %origin,
generation = *generation,
"ignoring inadmissible peer-edit fence"
);
false
}
/// True when a strictly newer edit from the same origin site already landed
/// here. No lock on the sending side can order deliveries issued by two
/// nodes of that site, so ordering is decided here, on the generation the
@@ -6075,11 +6119,42 @@ fn peer_edit_fence(queries: &HashMap<String, String>) -> Option<(String, u64)> {
/// stale: one edit legitimately fans out several deliveries under a single
/// generation (the ILM-expiry edit sends every peer's record), and a replay of
/// an applied delivery re-applies the same edit idempotently.
///
/// A mark more than [`PEER_EDIT_FENCE_STALENESS_WINDOW_NANOS`] above the
/// delivery is implausible and does NOT fence: the shared service account
/// means any peer can stamp any origin, so a forged `u64::MAX`-scale mark
/// would otherwise silently swallow the origin's genuine edits for good.
/// Bounding the fence by distance instead of by an absolute ceiling keeps
/// ordering intact wherever the origin's clock actually operates — two
/// racing deliveries trail each other by seconds whether the hybrid clock
/// tracks wall time or persists a long-gone excursion far ahead of it —
/// while a mark no genuine race can explain merely downgrades the origin to
/// unfenced (pre-fence) delivery instead of dropping its edits. (One genuine
/// shape does land out here: a plain-counter straggler arriving after its
/// origin's first hybrid-clock edit. It gets the same downgrade — applied
/// unfenced — once, at upgrade time; fencing it instead would silence the
/// mirror case, a hybrid-clock origin downgraded back to the plain counter.)
fn peer_edit_delivery_is_stale(state: &SiteReplicationState, origin: &str, generation: u64) -> bool {
state
.applied_edit_generations
.get(origin)
.is_some_and(|applied| *applied > generation)
let Some(applied) = state.applied_edit_generations.get(origin) else {
return false;
};
if *applied <= generation {
return false;
}
if *applied - generation > PEER_EDIT_FENCE_STALENESS_WINDOW_NANOS {
warn!(
event = EVENT_ADMIN_SITE_REPLICATION_STATE,
component = LOG_COMPONENT_ADMIN,
subsystem = LOG_SUBSYSTEM_SITE_REPLICATION,
result = "fence_mark_beyond_staleness_window",
origin,
generation,
applied_mark = *applied,
"ignoring implausibly distant peer-edit high-water mark"
);
return false;
}
true
}
fn record_applied_peer_edit_generation(state: &mut SiteReplicationState, origin: &str, generation: u64) {
@@ -9283,16 +9358,247 @@ async fn apply_iam_item(item: SRIAMItem) -> S3Result<()> {
let incoming_updated_at = item.updated_at;
match item.r#type.as_str() {
"policy" => apply_iam_policy_item(&iam_sys, &item.name, item.policy).await,
"policy-mapping" => apply_iam_policy_mapping_item(&iam_sys, item.policy_mapping).await,
"group-info" => apply_iam_group_info_item(&iam_sys, item.group_info).await,
"policy" => {
if let Some(policy) = item.policy {
let policy: Policy =
serde_json::from_value(policy).map_err(|e| s3_error!(InvalidRequest, "invalid policy body: {}", e))?;
iam_sys.set_policy(&item.name, policy).await.map_err(ApiError::from)?;
} else {
iam_sys.delete_policy(&item.name, true).await.map_err(ApiError::from)?;
}
Ok(())
}
"policy-mapping" => {
let Some(mapping) = item.policy_mapping else {
return Err(s3_error!(InvalidRequest, "policyMapping is required"));
};
let user_type =
user_type_from_sr_wire(mapping.user_type).ok_or_else(|| s3_error!(InvalidRequest, "invalid userType"))?;
iam_sys
.policy_db_set(&mapping.user_or_group, user_type, mapping.is_group, &mapping.policy)
.await
.map_err(ApiError::from)?;
Ok(())
}
"group-info" => {
let Some(group_info) = item.group_info else {
return Err(s3_error!(InvalidRequest, "groupInfo is required"));
};
let update = group_info.update_req;
if !group_info_requires_upsert(&update) {
iam_sys
.remove_users_from_group(&update.group, update.members)
.await
.map_err(ApiError::from)?;
return Ok(());
}
iam_sys
.add_users_to_group(&update.group, update.members)
.await
.map_err(ApiError::from)?;
iam_sys
.set_group_status(&update.group, matches!(update.status, GroupStatus::Enabled))
.await
.map_err(ApiError::from)?;
Ok(())
}
// MinIO madmin-go sends `SRIAMItemSTSAcc = "sts-account"`. The legacy alias
// `sts-credential` (emitted by older RustFS releases) stays accepted permanently
// so mixed-version RustFS sites keep replicating STS credentials during rolling
// upgrades; it is a compatibility layer, not temporary code.
SR_IAM_ITEM_STS_ACC | SR_IAM_ITEM_STS_ACC_LEGACY => apply_iam_sts_account_item(&iam_sys, item.sts_credential).await,
"iam-user" => apply_iam_user_item(&iam_sys, item.iam_user, incoming_updated_at).await,
"service-account" => apply_iam_service_account_item(&iam_sys, item.svc_acc_change, incoming_updated_at).await,
SR_IAM_ITEM_STS_ACC | SR_IAM_ITEM_STS_ACC_LEGACY => {
let Some(sts_credential) = item.sts_credential else {
return Err(s3_error!(InvalidRequest, "stsCredential is required"));
};
let Some(secret) = current_token_signing_key() else {
return Err(s3_error!(InvalidRequest, "token signing key not initialized"));
};
let claims = get_claims_from_token_with_secret(&sts_credential.session_token, &secret)
.map_err(|e| s3_error!(InvalidRequest, "invalid STS session token: {e}"))?;
let expiration = claims
.get("exp")
.and_then(claims_unix_timestamp)
.map(OffsetDateTime::from_unix_timestamp)
.transpose()
.map_err(|e| s3_error!(InvalidRequest, "invalid STS expiry: {e}"))?;
let groups = string_list_claim(&claims, "groups");
let compatibility_policy = sts_replication_compatibility_policy(&claims, &sts_credential.parent_policy_mapping);
let cred = rustfs_credentials::Credentials {
access_key: sts_credential.access_key.clone(),
secret_key: sts_credential.secret_key.clone(),
session_token: sts_credential.session_token.clone(),
expiration,
status: "on".to_string(),
parent_user: sts_credential.parent_user.clone(),
groups,
claims: Some(claims),
..Default::default()
};
iam_sys
.set_temp_user(&sts_credential.access_key, &cred, compatibility_policy)
.await
.map_err(ApiError::from)?;
Ok(())
}
"iam-user" => {
let Some(user) = item.iam_user else {
return Err(s3_error!(InvalidRequest, "iamUser is required"));
};
if let Some(local) = iam_sys.get_user(&user.access_key).await
&& is_stale_update(local.update_at.unwrap_or(OffsetDateTime::UNIX_EPOCH), incoming_updated_at)
{
return Ok(());
}
if user.is_delete_req {
iam_sys.delete_user(&user.access_key, true).await.map_err(ApiError::from)?;
} else {
let Some(user_req) = user.user_req else {
return Err(s3_error!(InvalidRequest, "userReq is required"));
};
let is_status_only_update = user_req.secret_key.is_empty() && user_req.policy.is_none();
if is_status_only_update {
iam_sys
.set_user_status(&user.access_key, user_req.status)
.await
.map_err(ApiError::from)?;
} else {
iam_sys
.create_user(&user.access_key, &user_req)
.await
.map_err(ApiError::from)?;
}
}
Ok(())
}
"service-account" => {
let Some(change) = item.svc_acc_change else {
return Err(s3_error!(InvalidRequest, "serviceAccountChange is required"));
};
let envelope = change.oidc_service_account_envelope;
if let Some(create) = change.create {
let local_updated_at = iam_sys
.get_user(&create.access_key)
.await
.map(|local| local.update_at.unwrap_or(OffsetDateTime::UNIX_EPOCH));
let replicated_policy = if create.access_key == SITE_REPLICATOR_SERVICE_ACCOUNT {
if local_updated_at.is_some_and(|local_updated_at| is_stale_update(local_updated_at, incoming_updated_at)) {
return Ok(());
}
ReplicatedServiceAccountPolicy {
policy: Some(site_replicator_service_account_policy()?),
is_envelope: false,
}
} else {
let Some(replicated_policy) = decode_service_account_replication_policy(
&create,
envelope.as_ref(),
incoming_updated_at,
local_updated_at,
)?
else {
return Ok(());
};
replicated_policy
};
match iam_sys.get_service_account(&create.access_key).await {
Ok((existing, _)) => {
if existing.parent_user != create.parent {
return Err(s3_error!(
InvalidRequest,
"service account {} already exists with a different parent user",
create.access_key
));
}
iam_sys
.update_service_account(
&create.access_key,
UpdateServiceAccountOpts {
name: replicated_policy.metadata_for_existing_account(create.name),
description: replicated_policy.metadata_for_existing_account(create.description),
session_policy: replicated_policy.for_existing_account(),
secret_key: Some(create.secret_key),
expiration: create.expiration,
status: (!create.status.is_empty()).then_some(create.status),
parent_user: None,
allow_site_replicator_account: create.access_key == SITE_REPLICATOR_SERVICE_ACCOUNT,
},
)
.await
.map_err(ApiError::from)?;
}
Err(err) if is_err_no_such_service_account(&err) => {
iam_sys
.new_service_account(
&create.parent,
Some(create.groups),
NewServiceAccountOpts {
session_policy: replicated_policy.policy,
access_key: create.access_key,
secret_key: create.secret_key,
name: (!create.name.is_empty()).then_some(create.name),
description: (!create.description.is_empty()).then_some(create.description),
expiration: create.expiration,
allow_site_replicator_account: true,
claims: Some(create.claims),
},
)
.await
.map_err(ApiError::from)?;
}
Err(err) => return Err(ApiError::from(err).into()),
}
return Ok(());
}
if let Some(update) = change.update {
if let Some(local) = iam_sys.get_user(&update.access_key).await
&& is_stale_update(local.update_at.unwrap_or(OffsetDateTime::UNIX_EPOCH), incoming_updated_at)
{
return Ok(());
}
let allow_site_replicator_account = update.access_key == SITE_REPLICATOR_SERVICE_ACCOUNT;
let session_policy = if allow_site_replicator_account {
Some(site_replicator_service_account_policy()?)
} else {
update.session_policy.as_str().and_then(|raw| serde_json::from_str(raw).ok())
};
iam_sys
.update_service_account(
&update.access_key,
UpdateServiceAccountOpts {
session_policy,
secret_key: (!update.secret_key.is_empty()).then_some(update.secret_key),
name: (!update.name.is_empty()).then_some(update.name),
description: (!update.description.is_empty()).then_some(update.description),
expiration: update.expiration,
status: (!update.status.is_empty()).then_some(update.status),
// Peers replicate credentials, never the local parent binding:
// each site resolves its own parent from its own IAM.
parent_user: None,
allow_site_replicator_account,
},
)
.await
.map_err(ApiError::from)?;
return Ok(());
}
if let Some(delete) = change.delete {
if let Some(local) = iam_sys.get_user(&delete.access_key).await
&& is_stale_update(local.update_at.unwrap_or(OffsetDateTime::UNIX_EPOCH), incoming_updated_at)
{
return Ok(());
}
iam_sys
.delete_service_account(&delete.access_key, true)
.await
.map_err(ApiError::from)?;
return Ok(());
}
Err(s3_error!(InvalidRequest, "serviceAccountChange is empty"))
}
_ => Err(s3_error!(
NotImplemented,
"site replication IAM item type `{}` is not supported",
@@ -9301,252 +9607,6 @@ async fn apply_iam_item(item: SRIAMItem) -> S3Result<()> {
}
}
async fn apply_iam_policy_item(iam_sys: &IamSys<ObjectStore>, name: &str, policy: Option<Value>) -> S3Result<()> {
if let Some(policy) = policy {
let policy: Policy =
serde_json::from_value(policy).map_err(|e| s3_error!(InvalidRequest, "invalid policy body: {}", e))?;
iam_sys.set_policy(name, policy).await.map_err(ApiError::from)?;
} else {
iam_sys.delete_policy(name, true).await.map_err(ApiError::from)?;
}
Ok(())
}
async fn apply_iam_policy_mapping_item(iam_sys: &IamSys<ObjectStore>, policy_mapping: Option<SRPolicyMapping>) -> S3Result<()> {
let Some(mapping) = policy_mapping else {
return Err(s3_error!(InvalidRequest, "policyMapping is required"));
};
let user_type = user_type_from_sr_wire(mapping.user_type).ok_or_else(|| s3_error!(InvalidRequest, "invalid userType"))?;
iam_sys
.policy_db_set(&mapping.user_or_group, user_type, mapping.is_group, &mapping.policy)
.await
.map_err(ApiError::from)?;
Ok(())
}
async fn apply_iam_group_info_item(iam_sys: &IamSys<ObjectStore>, group_info: Option<SRGroupInfo>) -> S3Result<()> {
let Some(group_info) = group_info else {
return Err(s3_error!(InvalidRequest, "groupInfo is required"));
};
let update = group_info.update_req;
if !group_info_requires_upsert(&update) {
iam_sys
.remove_users_from_group(&update.group, update.members)
.await
.map_err(ApiError::from)?;
return Ok(());
}
iam_sys
.add_users_to_group(&update.group, update.members)
.await
.map_err(ApiError::from)?;
iam_sys
.set_group_status(&update.group, matches!(update.status, GroupStatus::Enabled))
.await
.map_err(ApiError::from)?;
Ok(())
}
async fn apply_iam_sts_account_item(iam_sys: &IamSys<ObjectStore>, sts_credential: Option<SRSTSCredential>) -> S3Result<()> {
let Some(sts_credential) = sts_credential else {
return Err(s3_error!(InvalidRequest, "stsCredential is required"));
};
let Some(secret) = current_token_signing_key() else {
return Err(s3_error!(InvalidRequest, "token signing key not initialized"));
};
let claims = get_claims_from_token_with_secret(&sts_credential.session_token, &secret)
.map_err(|e| s3_error!(InvalidRequest, "invalid STS session token: {e}"))?;
let expiration = claims
.get("exp")
.and_then(claims_unix_timestamp)
.map(OffsetDateTime::from_unix_timestamp)
.transpose()
.map_err(|e| s3_error!(InvalidRequest, "invalid STS expiry: {e}"))?;
let groups = string_list_claim(&claims, "groups");
let compatibility_policy = sts_replication_compatibility_policy(&claims, &sts_credential.parent_policy_mapping);
let cred = rustfs_credentials::Credentials {
access_key: sts_credential.access_key.clone(),
secret_key: sts_credential.secret_key.clone(),
session_token: sts_credential.session_token.clone(),
expiration,
status: "on".to_string(),
parent_user: sts_credential.parent_user.clone(),
groups,
claims: Some(claims),
..Default::default()
};
iam_sys
.set_temp_user(&sts_credential.access_key, &cred, compatibility_policy)
.await
.map_err(ApiError::from)?;
Ok(())
}
async fn apply_iam_user_item(
iam_sys: &IamSys<ObjectStore>,
iam_user: Option<SRIAMUser>,
incoming_updated_at: Option<OffsetDateTime>,
) -> S3Result<()> {
let Some(user) = iam_user else {
return Err(s3_error!(InvalidRequest, "iamUser is required"));
};
if let Some(local) = iam_sys.get_user(&user.access_key).await
&& is_stale_update(local.update_at.unwrap_or(OffsetDateTime::UNIX_EPOCH), incoming_updated_at)
{
return Ok(());
}
if user.is_delete_req {
iam_sys.delete_user(&user.access_key, true).await.map_err(ApiError::from)?;
} else {
let Some(user_req) = user.user_req else {
return Err(s3_error!(InvalidRequest, "userReq is required"));
};
let is_status_only_update = user_req.secret_key.is_empty() && user_req.policy.is_none();
if is_status_only_update {
iam_sys
.set_user_status(&user.access_key, user_req.status)
.await
.map_err(ApiError::from)?;
} else {
iam_sys
.create_user(&user.access_key, &user_req)
.await
.map_err(ApiError::from)?;
}
}
Ok(())
}
async fn apply_iam_service_account_item(
iam_sys: &IamSys<ObjectStore>,
svc_acc_change: Option<SRSvcAccChange>,
incoming_updated_at: Option<OffsetDateTime>,
) -> S3Result<()> {
let Some(change) = svc_acc_change else {
return Err(s3_error!(InvalidRequest, "serviceAccountChange is required"));
};
let envelope = change.oidc_service_account_envelope;
if let Some(create) = change.create {
let local_updated_at = iam_sys
.get_user(&create.access_key)
.await
.map(|local| local.update_at.unwrap_or(OffsetDateTime::UNIX_EPOCH));
let replicated_policy = if create.access_key == SITE_REPLICATOR_SERVICE_ACCOUNT {
if local_updated_at.is_some_and(|local_updated_at| is_stale_update(local_updated_at, incoming_updated_at)) {
return Ok(());
}
ReplicatedServiceAccountPolicy {
policy: Some(site_replicator_service_account_policy()?),
is_envelope: false,
}
} else {
let Some(replicated_policy) =
decode_service_account_replication_policy(&create, envelope.as_ref(), incoming_updated_at, local_updated_at)?
else {
return Ok(());
};
replicated_policy
};
match iam_sys.get_service_account(&create.access_key).await {
Ok((existing, _)) => {
if existing.parent_user != create.parent {
return Err(s3_error!(
InvalidRequest,
"service account {} already exists with a different parent user",
create.access_key
));
}
iam_sys
.update_service_account(
&create.access_key,
UpdateServiceAccountOpts {
name: replicated_policy.metadata_for_existing_account(create.name),
description: replicated_policy.metadata_for_existing_account(create.description),
session_policy: replicated_policy.for_existing_account(),
secret_key: Some(create.secret_key),
expiration: create.expiration,
status: (!create.status.is_empty()).then_some(create.status),
parent_user: None,
allow_site_replicator_account: create.access_key == SITE_REPLICATOR_SERVICE_ACCOUNT,
},
)
.await
.map_err(ApiError::from)?;
}
Err(err) if is_err_no_such_service_account(&err) => {
iam_sys
.new_service_account(
&create.parent,
Some(create.groups),
NewServiceAccountOpts {
session_policy: replicated_policy.policy,
access_key: create.access_key,
secret_key: create.secret_key,
name: (!create.name.is_empty()).then_some(create.name),
description: (!create.description.is_empty()).then_some(create.description),
expiration: create.expiration,
allow_site_replicator_account: true,
claims: Some(create.claims),
},
)
.await
.map_err(ApiError::from)?;
}
Err(err) => return Err(ApiError::from(err).into()),
}
return Ok(());
}
if let Some(update) = change.update {
if let Some(local) = iam_sys.get_user(&update.access_key).await
&& is_stale_update(local.update_at.unwrap_or(OffsetDateTime::UNIX_EPOCH), incoming_updated_at)
{
return Ok(());
}
let allow_site_replicator_account = update.access_key == SITE_REPLICATOR_SERVICE_ACCOUNT;
let session_policy = if allow_site_replicator_account {
Some(site_replicator_service_account_policy()?)
} else {
update.session_policy.as_str().and_then(|raw| serde_json::from_str(raw).ok())
};
iam_sys
.update_service_account(
&update.access_key,
UpdateServiceAccountOpts {
session_policy,
secret_key: (!update.secret_key.is_empty()).then_some(update.secret_key),
name: (!update.name.is_empty()).then_some(update.name),
description: (!update.description.is_empty()).then_some(update.description),
expiration: update.expiration,
status: (!update.status.is_empty()).then_some(update.status),
// Peers replicate credentials, never the local parent binding:
// each site resolves its own parent from its own IAM.
parent_user: None,
allow_site_replicator_account,
},
)
.await
.map_err(ApiError::from)?;
return Ok(());
}
if let Some(delete) = change.delete {
if let Some(local) = iam_sys.get_user(&delete.access_key).await
&& is_stale_update(local.update_at.unwrap_or(OffsetDateTime::UNIX_EPOCH), incoming_updated_at)
{
return Ok(());
}
iam_sys
.delete_service_account(&delete.access_key, true)
.await
.map_err(ApiError::from)?;
return Ok(());
}
Err(s3_error!(InvalidRequest, "serviceAccountChange is empty"))
}
fn claims_unix_timestamp(value: &Value) -> Option<i64> {
match value {
Value::Number(number) => number.as_i64(),
@@ -10718,6 +10778,11 @@ impl Operation for SRPeerEditHandler {
let outcome = update_site_replication_state_when_changed(move |state| {
let mut incoming = incoming;
let local_peer = local_peer_at_endpoint(commit_endpoint, state);
// The fence is self-reported — the shared service account means
// the sender cannot be identified — so it is honoured only after
// the admissibility check, against the same state it will gate.
let commit_fence =
commit_fence.filter(|fence| peer_edit_fence_is_admissible(state, &local_peer.deployment_id, fence));
// Ordering fence: the sending site allocates the generation under
// its state-object lock, so a delivery that lost the race carries
// a generation this site has already passed. Applying it would
@@ -13413,6 +13478,15 @@ mod tests {
handler_block.contains("record_applied_peer_edit_generation(state, origin, *generation);"),
"SRPeerEditHandler must record the applied generation so later stale deliveries are recognised"
);
// Fence hardening: origin and generation are self-reported by a
// caller the shared service account cannot identify, so the handler
// must pass the fence through the admissibility check — against the
// same state the fence gates, i.e. inside the transaction — before
// reading or raising any high-water mark.
assert!(
handler_block.contains(".filter(|fence| peer_edit_fence_is_admissible(state, &local_peer.deployment_id, fence))"),
"SRPeerEditHandler must admit a fence only through peer_edit_fence_is_admissible inside the state transaction"
);
// P1-15 PR2: both halves of the fence and the edit they fence share
// ONE transaction. Checking the fence against a state read outside the
// lock would let the check pass on one snapshot and the write land on
@@ -14789,6 +14863,121 @@ mod tests {
assert!(peer_edit_delivery_is_stale(&state, origin, generation - 1));
}
/// A fence is self-reported: every site authenticates peer traffic with
/// the same site-replicator credential, so a compromised peer can stamp
/// ANY origin with ANY generation. An origin the receiver does not
/// replicate with — or the receiver itself — is ignored and plants no
/// mark; a mark a compromised peer plants for a CURRENT origin cannot
/// silence that origin, because the staleness window refuses to fence on
/// a mark implausibly far above the genuine deliveries.
#[test]
fn forged_peer_edit_fences_cannot_poison_the_high_water_marks() {
let mut state = SiteReplicationState {
peers: BTreeMap::from([
(
"site-local".to_string(),
PeerInfo {
deployment_id: "site-local".to_string(),
..peer("local", "https://local.example:9000")
},
),
(
"site-victim".to_string(),
PeerInfo {
deployment_id: "site-victim".to_string(),
..peer("victim", "https://victim.example:9000")
},
),
]),
..Default::default()
};
// An origin outside the current membership is refused outright...
let unknown = ("site-unknown".to_string(), 4u64);
assert!(!peer_edit_fence_is_admissible(&state, "site-local", &unknown));
// No site delivers edits to itself: a fence claiming the receiver as
// its origin is forged by construction, current peer or not.
let own = ("site-local".to_string(), 4u64);
assert!(!peer_edit_fence_is_admissible(&state, "site-local", &own));
// A current remote peer's fence is admitted and works end to end.
let genuine = ("site-victim".to_string(), 1u64);
assert!(peer_edit_fence_is_admissible(&state, "site-local", &genuine));
assert!(!peer_edit_delivery_is_stale(&state, &genuine.0, genuine.1));
record_applied_peer_edit_generation(&mut state, &genuine.0, genuine.1);
assert_eq!(state.applied_edit_generations.get("site-victim"), Some(&1));
// A forged u64::MAX-scale mark CAN be recorded — the shared service
// account means the receiver cannot tell the stamp was forged — but
// it is inert: the victim's genuine hybrid-clock deliveries sit far
// more than the staleness window below it, so they keep applying
// instead of being silently acked-and-dropped.
record_applied_peer_edit_generation(&mut state, "site-victim", u64::MAX);
assert!(!peer_edit_delivery_is_stale(&state, "site-victim", edit_generation_wall_clock()));
}
/// The staleness window bounds the fence by DISTANCE from the mark, not
/// by an absolute clock ceiling, so ordering must hold wherever the
/// origin's hybrid clock actually operates. The regression that matters:
/// a temporary wall-clock excursion far in the future is persisted by
/// `next_peer_edit_generation` (`max(now, prev + 1)` never comes back
/// down), and two later edits g+1 then g can arrive in reverse order —
/// g must still be fenced, even though both generations dwarf the
/// receiver's clock. Conversely a mark further above a delivery than any
/// genuine race can explain must not fence it.
#[test]
fn peer_edit_fence_orders_a_persisted_future_clock_and_defuses_distant_marks() {
let mut state = SiteReplicationState {
peers: BTreeMap::from([(
"site-origin".to_string(),
PeerInfo {
deployment_id: "site-origin".to_string(),
..peer("origin", "https://origin.example:9000")
},
)]),
..Default::default()
};
// The origin's clock once jumped ten years ahead; the hybrid clock
// keeps allocating from there long after the clock was corrected.
let excursion = edit_generation_wall_clock() + 10 * 365 * 24 * 60 * 60 * 1_000_000_000;
let fence = ("site-origin".to_string(), excursion + 1);
assert!(peer_edit_fence_is_admissible(&state, "site-local", &fence));
record_applied_peer_edit_generation(&mut state, &fence.0, fence.1);
// The reverse delivery of the race: g arrives after g+1 landed.
// Without the fence it would commit last and roll g+1 back.
assert!(peer_edit_delivery_is_stale(&state, "site-origin", excursion));
// Equal generation (same edit's fan-out or a replay) still applies,
// as does the next edit.
assert!(!peer_edit_delivery_is_stale(&state, "site-origin", excursion + 1));
assert!(!peer_edit_delivery_is_stale(&state, "site-origin", excursion + 2));
// The window's exact boundary: a delivery trailing the mark by the
// full window is still fenced; one nanosecond further is not — that
// distance is no longer explicable by a genuine race, only by a
// forged mark or an excursion the origin has left behind.
let mark = fence.1;
// A straggler trailing by a concrete hour must still be fenced —
// pins the window's real magnitude, not just its symbolic boundary.
assert!(peer_edit_delivery_is_stale(&state, "site-origin", mark - 60 * 60 * 1_000_000_000));
assert!(peer_edit_delivery_is_stale(
&state,
"site-origin",
mark - PEER_EDIT_FENCE_STALENESS_WINDOW_NANOS
));
assert!(!peer_edit_delivery_is_stale(
&state,
"site-origin",
mark - PEER_EDIT_FENCE_STALENESS_WINDOW_NANOS - 1
));
// A pre-hybrid plain-counter origin trails such a mark by eons: it
// is not fenced (the rc.2-era downgrade case), it just runs
// unfenced until its counter regime catches up.
assert!(!peer_edit_delivery_is_stale(&state, "site-origin", 3));
}
/// P1-15 review follow-up: a site that leaves the mesh drops below two
/// peers, which clears its state object and restarts its generation
/// counter at zero. A mark left over from its previous membership would
+29
View File
@@ -417,6 +417,13 @@ struct SystemAdminDiscovery {
struct ServerInfoResponse {
info: InfoMessage,
admin_discovery: SystemAdminDiscovery,
/// Startup bitrot algorithm self-test outcome (rustfs/backlog#1873):
/// `passed` (algorithms verified at boot), `failed` (a drifted hash
/// implementation — the process is serving with degraded integrity
/// checking unless `RUSTFS_BITROT_SELFTEST_STRICT` aborted it), or
/// `unknown` (not yet run or disabled).
#[serde(rename = "bitrotSelftest")]
bitrot_selftest: &'static str,
}
#[derive(Serialize)]
@@ -433,6 +440,14 @@ fn system_admin_discovery(usecase: &DefaultAdminUsecase) -> SystemAdminDiscovery
}
}
fn bitrot_selftest_status_str() -> &'static str {
match crate::bitrot_selftest::bitrot_selftest_passed() {
Some(true) => "passed",
Some(false) => "failed",
None => "unknown",
}
}
#[async_trait::async_trait]
impl Operation for ServerInfoHandler {
async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
@@ -464,6 +479,7 @@ impl Operation for ServerInfoHandler {
let response = ServerInfoResponse {
info,
admin_discovery: system_admin_discovery(&usecase),
bitrot_selftest: bitrot_selftest_status_str(),
};
let data = serde_json::to_vec(&response).map_err(|e| {
@@ -1535,6 +1551,18 @@ mod tests {
);
}
/// The startup bitrot self-test outcome must surface in server info as one
/// of three closed-set strings, never an internal enum or a null
/// (rustfs/backlog#1873). This test pins the string mapping; whether the
/// process-global cell holds Some(true)/Some(false)/None is owned by
/// `crate::bitrot_selftest`'s own tests.
#[test]
fn bitrot_selftest_status_str_is_a_closed_set_of_operators_strings() {
let rendered = super::bitrot_selftest_status_str();
assert!(matches!(rendered, "passed" | "failed" | "unknown"));
assert_eq!(super::bitrot_selftest_status_str(), rendered);
}
#[test]
fn server_info_response_exposes_admin_discovery_paths() {
let usecase = DefaultAdminUsecase::without_context();
@@ -1556,6 +1584,7 @@ mod tests {
pools: None,
},
admin_discovery: system_admin_discovery(&usecase),
bitrot_selftest: super::bitrot_selftest_status_str(),
};
let value = serde_json::to_value(response).expect("server info response should serialize");
+8 -262
View File
@@ -17,8 +17,7 @@ use super::storage_api::bucket::metadata_sys;
use super::storage_api::bucket::replication::{self, BucketReplicationResyncStatus, BucketStats, ReplicationStatusType};
use super::storage_api::bucket::target::{BucketTarget, BucketTargetType, BucketTargets};
use super::storage_api::bucket::target_sys::{
BucketTargetSys, PutObjectOptions, RemoveObjectOptions, S3ClientError, SsecPassthroughCapability, TargetClient,
append_version_id_query,
BucketTargetSys, PutObjectOptions, RemoveObjectOptions, S3ClientError, TargetClient, append_version_id_query,
};
use super::storage_api::bucket::versioning_sys::BucketVersioningSys;
use super::storage_api::bucket::{AdminReplicationConfigExt as _, AdminVersioningConfigExt as _};
@@ -68,9 +67,6 @@ use rustfs_policy::policy::action::{Action, S3Action};
use rustfs_s3_types::EventName;
use rustfs_signer::pre_sign_v4;
use rustfs_utils::egress::{OutboundDnsResolver, OutboundPolicy};
use rustfs_utils::http::object_encryption_keys::{
REPLICATION_SSEC_ALGORITHM_HEADER, REPLICATION_SSEC_KEY_MD5_HEADER, REPLICATION_SSEC_ORIGINAL_SIZE_HEADER,
};
use rustfs_utils::http::{
SUFFIX_SOURCE_DELETEMARKER, SUFFIX_SOURCE_MTIME, SUFFIX_SOURCE_REPLICATION_CHECK, SUFFIX_SOURCE_REPLICATION_REQUEST,
SUFFIX_SOURCE_VERSION_ID, get_source_scheme, insert_header,
@@ -214,13 +210,6 @@ const REPLICATION_CHECK_ERROR_MAX_BYTES: usize = 512;
/// RustFS extension code (no madmin analogue): the target does not adopt the
/// source version id, breaking the version-identity replication contract.
const REPLICATION_CHECK_CODE_VERSION_MISMATCH: &str = "BucketRemoteTargetVersionMismatch";
/// RustFS extension code (no madmin analogue): the target drops the
/// `X-Rustfs-Replication-*` SSE-C passthrough headers, so an SSE-C replica
/// would lose its decryption material (N2 fail-closed).
const REPLICATION_CHECK_CODE_SSEC_PASSTHROUGH: &str = "BucketRemoteSsecPassthroughUnsupported";
/// Syntactically valid stand-in SSE-C key MD5 for the passthrough probe (the
/// probe object is never decrypted; it only has to round-trip the metadata).
const REPLICATION_CHECK_SSEC_PROBE_KEY_MD5: &str = "AAAAAAAAAAAAAAAAAAAAAA==";
#[derive(Debug, Clone, serde::Serialize)]
struct ReplicationCheckResponse {
@@ -262,8 +251,6 @@ struct ReplicationCheckPhases {
put: ReplicationCheckPhaseStatus,
#[serde(rename = "VersionFidelity")]
version_fidelity: ReplicationCheckPhaseStatus,
#[serde(rename = "SsecPassthrough")]
ssec_passthrough: ReplicationCheckPhaseStatus,
#[serde(rename = "DeleteMarker")]
delete_marker: ReplicationCheckPhaseStatus,
#[serde(rename = "VersionDelete")]
@@ -1866,7 +1853,7 @@ fn build_replication_check_response(mut targets: Vec<ReplicationCheckTargetStatu
let data = serde_json::to_vec(&ReplicationCheckResponse {
status: status.to_string(),
active_mutation: true,
mutation_description: "Writes probe objects (including an SSE-C passthrough probe), creates a delete marker, deletes the probe versions, and cleans up all probe artifacts on each target.",
mutation_description: "Writes a probe object, creates a delete marker, deletes the probe version, and cleans up all probe artifacts on each target.",
probe_namespace: REPLICATION_CHECK_PROBE_PREFIX,
targets,
})
@@ -2083,25 +2070,6 @@ async fn check_replication_target(
time: OffsetDateTime::now_utc(),
};
execute_replication_probe(&mut result, &mut operations).await;
// Sync the probe verdict into the runtime capability cache: the
// replication worker then fails SSE-C replication closed on a flagged
// target (or skips its own HEAD-back audit on a proven one) without
// re-learning what the probe just established.
match (result.phases.ssec_passthrough.status, result.phases.ssec_passthrough.code) {
("OK", _) => {
BucketTargetSys::get()
.record_ssec_passthrough_capability(&target.arn, SsecPassthroughCapability::Supported)
.await;
}
("FAILED", Some(REPLICATION_CHECK_CODE_SSEC_PASSTHROUGH)) => {
BucketTargetSys::get()
.record_ssec_passthrough_capability(&target.arn, SsecPassthroughCapability::Unsupported)
.await;
}
_ => {}
}
result
}
@@ -2120,15 +2088,6 @@ struct ReplicationProbePutOutcome {
response_version_id: Option<String>,
}
/// Outcome of the SSE-C passthrough probe: whether the HEAD-back of the probe
/// replica echoed SSE-C evidence (the customer-algorithm header a RustFS
/// target restores from the passthrough transport headers), plus the version
/// the target assigned so cleanup can address it.
struct ReplicationSsecProbeOutcome {
evidence_present: bool,
version_id: Option<String>,
}
struct ReplicationProbeMultipartError {
primary: S3ClientError,
cleanup_error: Option<String>,
@@ -2151,14 +2110,9 @@ trait ReplicationProbeOperations {
/// there: a target can adopt PutObject version ids and still mint its own
/// for CreateMultipartUpload.
async fn multipart_put(&mut self) -> Result<ReplicationProbePutOutcome, ReplicationProbeMultipartError>;
/// PUT a probe version carrying the SSE-C passthrough transport headers,
/// HEAD it back through the replication-check channel, and report whether
/// the SSE-C evidence survived. Cleanup of the created version is the
/// caller's job (the outcome carries its version id).
async fn ssec_passthrough_probe(&mut self) -> Result<ReplicationSsecProbeOutcome, S3ClientError>;
async fn create_delete_marker(&mut self, version_id: Option<&str>) -> Result<Option<String>, S3ClientError>;
async fn delete_version(&mut self, version_id: Option<&str>) -> Result<(), S3ClientError>;
async fn cleanup(&mut self, known_version_ids: [Option<&str>; 4]) -> Result<(), String>;
async fn cleanup(&mut self, known_version_ids: [Option<&str>; 3]) -> Result<(), String>;
}
struct RemoteReplicationProbeOperations<'a> {
@@ -2178,10 +2132,6 @@ impl ReplicationProbeOperations for RemoteReplicationProbeOperations<'_> {
multipart_put_replication_probe_object(self.client, self.bucket, self.key, self.time).await
}
async fn ssec_passthrough_probe(&mut self) -> Result<ReplicationSsecProbeOutcome, S3ClientError> {
ssec_passthrough_probe_object(self.client, self.bucket, self.key, self.time).await
}
async fn create_delete_marker(&mut self, version_id: Option<&str>) -> Result<Option<String>, S3ClientError> {
delete_replication_probe_object(
self.client,
@@ -2205,7 +2155,7 @@ impl ReplicationProbeOperations for RemoteReplicationProbeOperations<'_> {
.map(|_| ())
}
async fn cleanup(&mut self, known_version_ids: [Option<&str>; 4]) -> Result<(), String> {
async fn cleanup(&mut self, known_version_ids: [Option<&str>; 3]) -> Result<(), String> {
cleanup_replication_probe(self.client, self.bucket, self.key, known_version_ids).await
}
}
@@ -2226,7 +2176,6 @@ fn version_fidelity_error(api: &str, outcome: &ReplicationProbePutOutcome) -> Op
async fn execute_replication_probe(result: &mut ReplicationCheckTargetStatus, operations: &mut impl ReplicationProbeOperations) {
let mut probe_version_id = None;
let mut multipart_probe_version_id = None;
let mut ssec_probe_version_id = None;
let mut delete_marker_version_id = None;
let mut cleanup_required = true;
let mut multipart_cleanup_error = None;
@@ -2282,38 +2231,6 @@ async fn execute_replication_probe(result: &mut ReplicationCheckTargetStatus, op
}
}
// N2: probe SSE-C passthrough with the same transport headers live
// replication sends. A target that drops them (MinIO, generic S3) stores
// the probe as a plain object and echoes no SSE-C evidence on the
// HEAD-back; SSE-C replicas there would silently lose their decryption
// material, so the target must be flagged with a machine-readable code.
// Deliberately unlike VersionFidelity, a failed SsecPassthrough phase
// does NOT fail the target overall: version-identity drift breaks the
// replication contract for every object, while dropped SSE-C passthrough
// headers only limit a capability — a plaintext-only deployment against a
// MinIO target is perfectly healthy and must not turn red. The phase's
// own FAILED + machine-readable Code remains for madmin consumers (and
// the verdict still reaches the runtime capability cache).
if result.phases.put.status == "OK" && result.phases.version_fidelity.status == "OK" {
match operations.ssec_passthrough_probe().await {
Ok(outcome) => {
ssec_probe_version_id = outcome.version_id;
if outcome.evidence_present {
result.phases.ssec_passthrough = ReplicationCheckPhaseStatus::passed();
} else {
let error = "target drops SSE-C passthrough replication headers; \
SSE-C replicas would lose their decryption material on this target";
result.phases.ssec_passthrough =
ReplicationCheckPhaseStatus::failed_with_code(error, REPLICATION_CHECK_CODE_SSEC_PASSTHROUGH);
}
}
Err(err) => {
let error = format_replication_check_client_error(&err, ReplicationCheckFailureContext::ReplicateObject);
result.phases.ssec_passthrough = ReplicationCheckPhaseStatus::failed(&error);
}
}
}
if result.phases.put.status == "OK" && result.phases.version_fidelity.status == "OK" {
match operations.create_delete_marker(probe_version_id.as_deref()).await {
Ok(version_id) => {
@@ -2342,7 +2259,6 @@ async fn execute_replication_probe(result: &mut ReplicationCheckTargetStatus, op
.cleanup([
probe_version_id.as_deref(),
multipart_probe_version_id.as_deref(),
ssec_probe_version_id.as_deref(),
delete_marker_version_id.as_deref(),
])
.await
@@ -2637,72 +2553,6 @@ async fn put_replication_probe_object(
})
}
/// PUT a fresh probe version carrying the SSE-C passthrough transport headers
/// (the wire shape live SSE-C replication uses), then HEAD it back through the
/// worker channel (replication-check exemption + proxy suppression). A RustFS
/// target restores the transport headers into stored SSE-C metadata and its
/// HEAD echoes `x-amz-server-side-encryption-customer-algorithm`; a target
/// that dropped the headers echoes nothing. The probe body is never SSE-C
/// encrypted — only the metadata round-trip matters — and the version is
/// deleted by the shared probe cleanup.
async fn ssec_passthrough_probe_object(
target_client: &TargetClient,
target_bucket: &str,
probe_key: &str,
now: OffsetDateTime,
) -> Result<ReplicationSsecProbeOutcome, S3ClientError> {
let options = build_replication_probe_put_options(now);
let sent_version_id = options.internal.source_version_id.clone();
let mut headers = build_replication_probe_headers(&options);
// These are full wire names (not x-rustfs/x-minio suffixes), so they must
// be inserted verbatim — `insert_header` would mangle them.
for (name, value) in [
(REPLICATION_SSEC_ALGORITHM_HEADER, "AES256"),
(REPLICATION_SSEC_KEY_MD5_HEADER, REPLICATION_CHECK_SSEC_PROBE_KEY_MD5),
(REPLICATION_SSEC_ORIGINAL_SIZE_HEADER, "8"),
] {
let name = name
.parse::<HeaderName>()
.map_err(|err| S3ClientError::new(format!("invalid ssec probe header name: {err}")))?;
let value =
HeaderValue::from_str(value).map_err(|err| S3ClientError::new(format!("invalid ssec probe header value: {err}")))?;
headers.insert(name, value);
}
let query_version_id = sent_version_id.clone();
let response = target_client
.client
.put_object()
.bucket(target_bucket)
.key(probe_key)
.content_length(8)
.body(AwsByteStream::from_static(b"aaaaaaaa"))
.customize()
.map_request(move |mut req| {
for (key, value) in headers.clone() {
req.headers_mut().insert(key.expect("operation should succeed"), value);
}
let uri = append_version_id_query(req.uri(), &query_version_id);
req.set_uri(uri).map_err(std::io::Error::other)?;
Result::<_, std::io::Error>::Ok(req)
})
.send()
.await
.map_err(S3ClientError::from)?;
let version_id = response.version_id().map(ToOwned::to_owned);
let head_version = version_id.clone().or_else(|| Some(sent_version_id.clone()));
let head = target_client
.head_object(target_bucket, probe_key, head_version)
.await
.map_err(S3ClientError::from)?;
Ok(ReplicationSsecProbeOutcome {
evidence_present: head.sse_customer_algorithm().is_some_and(|algorithm| !algorithm.is_empty()),
version_id,
})
}
async fn delete_replication_probe_object(
target_client: &TargetClient,
target_bucket: &str,
@@ -3873,12 +3723,6 @@ mod tests {
/// Same, for the multipart leg: a target may mirror PutObject ids and
/// still mint its own at CreateMultipartUpload.
minted_multipart_version_id: Option<&'static str>,
/// Transport failure of the SSE-C passthrough probe itself.
ssec_probe_error: Option<&'static str>,
/// Models a MinIO-like target that drops the SSE-C passthrough
/// headers: the probe HEAD-back echoes no SSE-C evidence. The default
/// (false) models a RustFS target that preserves them.
ssec_evidence_missing: bool,
delete_marker_error: Option<&'static str>,
version_delete_error: Option<&'static str>,
cleanup_error: Option<&'static str>,
@@ -3916,17 +3760,6 @@ mod tests {
})
}
async fn ssec_passthrough_probe(&mut self) -> Result<ReplicationSsecProbeOutcome, S3ClientError> {
self.calls.push("ssec-probe");
match self.ssec_probe_error {
Some(code) => Err(scripted_probe_error(code)),
None => Ok(ReplicationSsecProbeOutcome {
evidence_present: !self.ssec_evidence_missing,
version_id: Some("ssec-version".to_string()),
}),
}
}
async fn create_delete_marker(&mut self, _version_id: Option<&str>) -> Result<Option<String>, S3ClientError> {
self.calls.push("delete-marker");
match self.delete_marker_error {
@@ -3943,7 +3776,7 @@ mod tests {
}
}
async fn cleanup(&mut self, known_version_ids: [Option<&str>; 4]) -> Result<(), String> {
async fn cleanup(&mut self, known_version_ids: [Option<&str>; 3]) -> Result<(), String> {
self.calls.push("cleanup");
self.cleanup_ids = known_version_ids
.into_iter()
@@ -3978,9 +3811,8 @@ mod tests {
assert_eq!(result.phases.version_fidelity.code, Some(REPLICATION_CHECK_CODE_VERSION_MISMATCH));
assert_eq!(result.phases.delete_marker.status, "SKIPPED");
assert_eq!(result.phases.version_delete.status, "SKIPPED");
assert_eq!(result.phases.ssec_passthrough.status, "SKIPPED");
assert_eq!(result.phases.cleanup.status, "OK");
assert_eq!(operations.cleanup_ids, [Some("target-minted-version".to_string()), None, None, None]);
assert_eq!(operations.cleanup_ids, [Some("target-minted-version".to_string()), None, None]);
}
#[tokio::test]
@@ -3993,70 +3825,6 @@ mod tests {
assert_eq!(result.status, "OK");
assert_eq!(result.phases.version_fidelity.status, "OK");
assert_eq!(result.phases.version_fidelity.code, None);
assert_eq!(result.phases.ssec_passthrough.status, "OK");
assert_eq!(result.phases.ssec_passthrough.code, None);
}
/// N2: a target that drops the SSE-C passthrough transport headers must
/// fail the SsecPassthrough phase with the machine-readable code while the
/// target overall stays OK — deliberately unlike VersionFidelity: this is
/// a capability limit, not a broken replication contract, and a
/// plaintext-only deployment against such a target must not turn red. The
/// other mutation phases keep running and the probe version is cleaned up.
#[tokio::test]
async fn replication_probe_flags_ssec_passthrough_dropping_target_without_failing_target() {
let mut result = replication_check_target("arn:a", "OK", None);
let mut operations = ScriptedReplicationProbe {
ssec_evidence_missing: true,
..Default::default()
};
execute_replication_probe(&mut result, &mut operations).await;
assert_eq!(
operations.calls,
[
"put",
"multipart-put",
"ssec-probe",
"delete-marker",
"version-delete",
"cleanup"
]
);
assert_eq!(result.status, "OK", "a capability-only failure must not fail the target overall");
assert_eq!(result.error, None);
assert_eq!(result.phases.ssec_passthrough.status, "FAILED");
assert_eq!(result.phases.ssec_passthrough.code, Some(REPLICATION_CHECK_CODE_SSEC_PASSTHROUGH));
assert_eq!(
operations.cleanup_ids,
[
Some("object-version".to_string()),
Some("multipart-version".to_string()),
Some("ssec-version".to_string()),
Some("marker-version".to_string())
]
);
}
/// A transport failure of the SSE-C probe is not evidence of a dropping
/// target: the phase fails without the capability code (the runtime cache
/// stays Unknown and the worker keeps auditing), and the target overall
/// stays OK.
#[tokio::test]
async fn replication_probe_ssec_transport_failure_carries_no_capability_code() {
let mut result = replication_check_target("arn:a", "OK", None);
let mut operations = ScriptedReplicationProbe {
ssec_probe_error: Some("InternalError"),
..Default::default()
};
execute_replication_probe(&mut result, &mut operations).await;
assert_eq!(result.status, "OK");
assert_eq!(result.phases.ssec_passthrough.status, "FAILED");
assert_eq!(result.phases.ssec_passthrough.code, None);
assert_eq!(operations.cleanup_ids[2], None, "a failed ssec probe leaves no version to clean");
}
#[tokio::test]
@@ -4087,23 +3855,12 @@ mod tests {
execute_replication_probe(&mut result, &mut operations).await;
assert_eq!(
operations.calls,
[
"put",
"multipart-put",
"ssec-probe",
"delete-marker",
"version-delete",
"cleanup"
]
);
assert_eq!(operations.calls, ["put", "multipart-put", "delete-marker", "version-delete", "cleanup"]);
assert_eq!(
operations.cleanup_ids,
[
Some("object-version".to_string()),
Some("multipart-version".to_string()),
Some("ssec-version".to_string()),
None
]
);
@@ -4123,23 +3880,12 @@ mod tests {
execute_replication_probe(&mut result, &mut operations).await;
assert_eq!(
operations.calls,
[
"put",
"multipart-put",
"ssec-probe",
"delete-marker",
"version-delete",
"cleanup"
]
);
assert_eq!(operations.calls, ["put", "multipart-put", "delete-marker", "version-delete", "cleanup"]);
assert_eq!(
operations.cleanup_ids,
[
Some("object-version".to_string()),
Some("multipart-version".to_string()),
Some("ssec-version".to_string()),
Some("marker-version".to_string())
]
);
-1
View File
@@ -196,7 +196,6 @@ pub(crate) mod bucket_target_sys {
pub(crate) type PutObjectOptions = super::ecstore_bucket::bucket_target_sys::PutObjectOptions;
pub(crate) type RemoveObjectOptions = super::ecstore_bucket::bucket_target_sys::RemoveObjectOptions;
pub(crate) type S3ClientError = super::ecstore_bucket::bucket_target_sys::S3ClientError;
pub(crate) type SsecPassthroughCapability = super::ecstore_bucket::bucket_target_sys::SsecPassthroughCapability;
pub(crate) type TargetClient = super::ecstore_bucket::bucket_target_sys::TargetClient;
}
+3 -248
View File
@@ -46,10 +46,9 @@ use super::storage_api::object_usecase::bucket::{
replication::{
DeleteReplicationConfigSnapshot, REPLICATE_INCOMING_DELETE, ReplicationStatusType, commit_force_delete_intent,
delete_replication_state_from_config, delete_replication_version_id, deleted_object_has_pending_replication_delete,
force_delete_target_set, get_read_proxy_targets, has_active_delete_rule, load_delete_config_snapshot,
must_replicate_object, persist_force_delete_intent, record_replication_proxy, schedule_object_replication,
schedule_replication_delete, schedule_replication_deletes, set_deleted_object_replication_state,
should_schedule_delete_replication, should_use_existing_delete_replication_info,
force_delete_target_set, has_active_delete_rule, load_delete_config_snapshot, must_replicate_object,
persist_force_delete_intent, schedule_object_replication, schedule_replication_delete, schedule_replication_deletes,
set_deleted_object_replication_state, should_schedule_delete_replication, should_use_existing_delete_replication_info,
},
tagging::decode_tags,
validate_restore_request,
@@ -6599,226 +6598,6 @@ impl DefaultObjectUsecase {
})
}
/// Headers a proxied read forwards verbatim to the replication target:
/// only the client's SSE-C key family, so the target performs the real
/// SSE-C decryption (never the replication-check exemption). HTTP
/// conditional headers (If-Match & co.) are deliberately NOT forwarded —
/// MinIO does not forward them either, and a remote 304/412 would leak a
/// conditional evaluation against a replica the local site never saw.
/// Range and part-number travel as typed SDK parameters instead.
fn proxy_read_passthrough_headers(headers: &HeaderMap) -> HeaderMap {
const FORWARDED: &[&str] = &[
"x-amz-server-side-encryption-customer-algorithm",
"x-amz-server-side-encryption-customer-key",
"x-amz-server-side-encryption-customer-key-md5",
];
let mut forwarded = HeaderMap::new();
for name in FORWARDED {
if let Ok(header_name) = http::HeaderName::from_str(name)
&& let Some(value) = headers.get(&header_name)
{
forwarded.insert(header_name, value.clone());
}
}
forwarded
}
/// True when a proxied SDK call failed because the target does not have
/// the object either (service-level not-found or a raw 404, which also
/// covers NoSuchVersion): the caller tries the next target silently.
fn proxy_sdk_error_is_not_found<E>(err: &aws_sdk_s3::error::SdkError<E>) -> bool {
err.raw_response().is_some_and(|resp| resp.status().as_u16() == 404)
}
/// Serve a GET whose local read failed with not-found by proxying to the
/// bucket's replication targets (MinIO `proxyGetToReplicationTarget`,
/// backlog#1675 P1-5). Returns None when no target can serve the object;
/// the caller then returns the original local error.
async fn proxy_get_object_to_replication_targets(
req: &S3Request<GetObjectInput>,
bucket: &str,
key: &str,
opts: &ObjectOptions,
) -> Option<GetObjectOutput> {
let targets = get_read_proxy_targets(bucket, key, opts).await;
if targets.is_empty() {
return None;
}
let extra_headers = Self::proxy_read_passthrough_headers(&req.headers);
let range = req
.headers
.get(http::header::RANGE)
.and_then(|value| value.to_str().ok())
.map(str::to_owned);
let part_number = req.input.part_number;
for target in targets {
match target
.get_object(
&target.bucket,
key,
opts.version_id.clone(),
range.clone(),
part_number,
extra_headers.clone(),
)
.await
{
Ok(remote) => {
// MinIO-aligned accounting: one total per proxy attempt
// (targets were available), one failed when no target
// served it — never per target.
record_replication_proxy(bucket, "GetObject", false).await;
return Some(Self::proxy_sdk_get_output_to_s3s(remote));
}
Err(err) if Self::proxy_sdk_error_is_not_found(&err) => {
debug!(bucket, key, arn = %target.arn, "read proxy: target does not have the object");
}
Err(err) => {
warn!(bucket, key, arn = %target.arn, error = %err, "read proxy: GET against replication target failed");
}
}
}
record_replication_proxy(bucket, "GetObject", true).await;
None
}
/// Serve a HEAD whose local lookup failed with not-found by proxying to
/// the bucket's replication targets (MinIO `proxyHeadToRepTarget`).
async fn proxy_head_object_to_replication_targets(
req: &S3Request<HeadObjectInput>,
bucket: &str,
key: &str,
opts: &ObjectOptions,
) -> Option<HeadObjectOutput> {
let targets = get_read_proxy_targets(bucket, key, opts).await;
if targets.is_empty() {
return None;
}
let extra_headers = Self::proxy_read_passthrough_headers(&req.headers);
let range = req
.headers
.get(http::header::RANGE)
.and_then(|value| value.to_str().ok())
.map(str::to_owned);
let part_number = req.input.part_number;
for target in targets {
match target
.head_object_for_proxy(
&target.bucket,
key,
opts.version_id.clone(),
range.clone(),
part_number,
extra_headers.clone(),
)
.await
{
Ok(remote) => {
// MinIO-aligned accounting: one total per proxy attempt,
// one failed when no target served it.
record_replication_proxy(bucket, "HeadObject", false).await;
return Some(Self::proxy_sdk_head_output_to_s3s(remote));
}
Err(err) if Self::proxy_sdk_error_is_not_found(&err) => {
debug!(bucket, key, arn = %target.arn, "read proxy: target does not have the object");
}
Err(err) => {
warn!(bucket, key, arn = %target.arn, error = %err, "read proxy: HEAD against replication target failed");
}
}
}
record_replication_proxy(bucket, "HeadObject", true).await;
None
}
/// Translate a proxied SDK GET response into the s3s output, forwarding
/// the body as a stream (no buffering, no local persistence).
fn proxy_sdk_get_output_to_s3s(remote: aws_sdk_s3::operation::get_object::GetObjectOutput) -> GetObjectOutput {
let body = remote.body;
let body_stream = tokio_util::io::ReaderStream::with_capacity(body.into_async_read(), 64 * 1024);
GetObjectOutput {
body: Some(StreamingBlob::wrap(body_stream)),
content_length: remote.content_length,
content_range: remote.content_range,
content_type: remote.content_type.as_deref().and_then(|v| ContentType::from_str(v).ok()),
content_encoding: remote.content_encoding,
content_disposition: remote.content_disposition,
content_language: remote.content_language,
cache_control: remote.cache_control,
accept_ranges: Some(ACCEPT_RANGES_BYTES.to_string()),
e_tag: remote.e_tag.as_deref().and_then(|v| ETag::from_str(v).ok()),
last_modified: remote
.last_modified
.and_then(|dt| OffsetDateTime::from_unix_timestamp_nanos(dt.as_nanos()).ok())
.map(Timestamp::from),
metadata: remote.metadata,
version_id: remote.version_id,
server_side_encryption: remote
.server_side_encryption
.map(|sse| ServerSideEncryption::from(sse.as_str().to_string())),
sse_customer_algorithm: remote.sse_customer_algorithm,
sse_customer_key_md5: remote.sse_customer_key_md5,
ssekms_key_id: remote.ssekms_key_id,
parts_count: remote.parts_count,
tag_count: remote.tag_count,
storage_class: remote.storage_class.map(|sc| StorageClass::from(sc.as_str().to_string())),
expiration: remote.expiration,
restore: remote.restore,
checksum_crc32: remote.checksum_crc32,
checksum_crc32c: remote.checksum_crc32_c,
checksum_crc64nvme: remote.checksum_crc64_nvme,
checksum_sha1: remote.checksum_sha1,
checksum_sha256: remote.checksum_sha256,
checksum_type: remote.checksum_type.map(|ct| ChecksumType::from(ct.as_str().to_string())),
..Default::default()
}
}
/// Translate a proxied SDK HEAD response into the s3s output.
///
/// Known gaps: the SDK's HeadObjectOutput does not model 206/Content-Range
/// for a ranged HEAD (the SDK exposes no content_range member on HEAD),
/// and s3s' typed HeadObjectOutput has no tag_count field (the local path
/// injects x-amz-tagging-count as a raw header) — both are dropped for
/// proxied HEADs.
fn proxy_sdk_head_output_to_s3s(remote: aws_sdk_s3::operation::head_object::HeadObjectOutput) -> HeadObjectOutput {
HeadObjectOutput {
content_length: remote.content_length,
content_type: remote.content_type.as_deref().and_then(|v| ContentType::from_str(v).ok()),
content_encoding: remote.content_encoding,
content_disposition: remote.content_disposition,
content_language: remote.content_language,
cache_control: remote.cache_control,
accept_ranges: Some(ACCEPT_RANGES_BYTES.to_string()),
e_tag: remote.e_tag.as_deref().and_then(|v| ETag::from_str(v).ok()),
last_modified: remote
.last_modified
.and_then(|dt| OffsetDateTime::from_unix_timestamp_nanos(dt.as_nanos()).ok())
.map(Timestamp::from),
metadata: remote.metadata,
version_id: remote.version_id,
server_side_encryption: remote
.server_side_encryption
.map(|sse| ServerSideEncryption::from(sse.as_str().to_string())),
sse_customer_algorithm: remote.sse_customer_algorithm,
sse_customer_key_md5: remote.sse_customer_key_md5,
ssekms_key_id: remote.ssekms_key_id,
parts_count: remote.parts_count,
storage_class: remote.storage_class.map(|sc| StorageClass::from(sc.as_str().to_string())),
expiration: remote.expiration,
restore: remote.restore,
checksum_crc32: remote.checksum_crc32,
checksum_crc32c: remote.checksum_crc32_c,
checksum_crc64nvme: remote.checksum_crc64_nvme,
checksum_sha1: remote.checksum_sha1,
checksum_sha256: remote.checksum_sha256,
checksum_type: remote.checksum_type.map(|ct| ChecksumType::from(ct.as_str().to_string())),
..Default::default()
}
}
#[instrument(name = "execute_get_object", level = "trace", skip(self, req))]
pub async fn execute_get_object(&self, req: S3Request<GetObjectInput>) -> S3Result<S3Response<GetObjectOutput>> {
self.execute_get_object_boxed(req).await
@@ -6944,19 +6723,6 @@ impl DefaultObjectUsecase {
{
Ok(prepared_read) => prepared_read,
Err(err) => {
// Active-active replication lag window: an object missing
// locally (and only missing — other errors keep their
// semantics) may still be served by proxying the GET to a
// replication target (backlog#1675 P1-5).
if matches!(*err.code(), S3ErrorCode::NoSuchKey | S3ErrorCode::NoSuchVersion)
&& let Some(output) = Self::proxy_get_object_to_replication_targets(&req, &bucket, &key, &opts).await
{
lifecycle.finish_ok();
let response = wrap_response_with_cors(&bucket, &req.method, &req.headers, output).await;
let result = Ok(response);
let _ = helper.version_id(version_id_for_event).complete(&result);
return result;
}
lifecycle.finish_err();
return Err(err);
}
@@ -8866,17 +8632,6 @@ impl DefaultObjectUsecase {
let msg = head_prefix_not_found_message(&bucket, &key, has_children);
return Err(S3Error::with_message(S3ErrorCode::NoSuchKey, msg));
}
// Active-active replication lag window: an object missing
// locally may still be served by proxying the HEAD to a
// replication target (backlog#1675 P1-5).
if let Some(output) = Self::proxy_head_object_to_replication_targets(&req, &bucket, &key, &opts).await {
let response = wrap_response_with_cors(&bucket, &req.method, &req.headers, output).await;
let result = Ok(response);
let _ = helper
.version_id(req.input.version_id.clone().unwrap_or_default())
.complete(&result);
return result;
}
return Err(S3Error::new(S3ErrorCode::NoSuchKey));
}
// Other errors, such as insufficient permissions, still return the original error
-18
View File
@@ -627,24 +627,6 @@ pub(crate) mod bucket {
#[cfg(test)]
pub(crate) use replication_contracts::replication_statuses_map;
/// Remote replication-target client used by the read-proxy path.
pub(crate) type ProxyTargetClient = crate::storage::storage_api::ecstore_bucket::bucket_target_sys::TargetClient;
/// Proxy-request metric recorder (get/head/tagging totals + failures).
pub(crate) use crate::storage::storage_api::record_replication_proxy;
/// Replication targets eligible to serve a proxied GET/HEAD/Tagging of
/// an object not present locally (MinIO `getProxyTargets`; empty when
/// the request was itself proxied, versioning is suspended, or no
/// replication rule matches). backlog#1675 P1-5.
pub(crate) async fn get_read_proxy_targets(
bucket: &str,
object: &str,
opts: &crate::storage::storage_api::StorageObjectOptions,
) -> Vec<Arc<ProxyTargetClient>> {
replication_contracts::get_proxy_targets(bucket, object, opts).await
}
pub(crate) async fn persist_force_delete_intent(
store: Arc<crate::storage::storage_api::ECStore>,
bucket: String,
+181
View File
@@ -0,0 +1,181 @@
// 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.
//! Startup bitrot algorithm self-test (rustfs/backlog#1873).
//!
//! A drifted hash implementation fails silently in production: every shard
//! reads back "corrupt", heal rewrites healthy data, and cross-platform
//! clusters disagree about which copy is good. [`run_startup_bitrot_self_test`]
//! pins the algorithms once at process start — the check itself runs in well
//! under a millisecond on 4 KiB, so it executes inline before background
//! services come up and the result is published before the server accepts
//! traffic.
//!
//! Outcome surface:
//! - one structured `bitrot_selftest` log event (`passed`/`failed`/`skipped`),
//! - the `rustfs_bitrot_selftest_status` gauge (1=passed, 0=failed, 2=skipped),
//! - [`bitrot_selftest_passed`] for admin/health surfaces,
//! - `RUSTFS_BITROT_SELFTEST_STRICT=on` turns a failure into a startup error
//! (MinIO `bitrotSelfTest` Fatal parity); the default only degrades the
//! status so a bad build cannot brick an existing fleet on upgrade.
use crate::storage_api::startup::background::{BitrotSelfTestError, bitrot_self_test};
use metrics::gauge;
use std::future::Future;
use std::io;
use std::sync::atomic::{AtomicU8, Ordering};
use std::time::Instant;
use tracing::{debug, error, info};
const LOG_COMPONENT_MAIN: &str = "main";
const LOG_SUBSYSTEM_STARTUP: &str = "startup";
const EVENT_BITROT_SELFTEST: &str = "bitrot_selftest";
const METRIC_BITROT_SELFTEST_STATUS: &str = "rustfs_bitrot_selftest_status";
/// Gauge values for [`METRIC_BITROT_SELFTEST_STATUS`].
const STATUS_PASSED: f64 = 1.0;
const STATUS_FAILED: f64 = 0.0;
const STATUS_SKIPPED: f64 = 2.0;
/// Internal cell values for [`BITROT_SELF_TEST_STATUS`].
const STATUS_CELL_UNSET: u8 = 0;
const STATUS_CELL_PASSED: u8 = 1;
const STATUS_CELL_FAILED: u8 = 2;
static BITROT_SELF_TEST_STATUS: AtomicU8 = AtomicU8::new(STATUS_CELL_UNSET);
/// Last recorded self-test outcome: `None` before the first run, then
/// `Some(true)` on a passing check and `Some(false)` on a failed one (a
/// skipped check never publishes, so it cannot read as a pass). The cell is
/// last-writer-wins rather than set-once: production runs the self-test once,
/// and last-writer-wins keeps tests that exercise both outcomes
/// order-independent.
pub fn bitrot_selftest_passed() -> Option<bool> {
match BITROT_SELF_TEST_STATUS.load(Ordering::Acquire) {
STATUS_CELL_UNSET => None,
STATUS_CELL_PASSED => Some(true),
STATUS_CELL_FAILED => Some(false),
_ => None,
}
}
/// Run the bitrot self-test and publish the outcome. In strict mode a failure
/// is returned as an error so the caller aborts startup.
pub(crate) async fn run_startup_bitrot_self_test(enabled: bool, strict: bool) -> io::Result<()> {
run_startup_bitrot_self_test_with(enabled, strict, bitrot_self_test).await
}
async fn run_startup_bitrot_self_test_with<F, Fut>(enabled: bool, strict: bool, run_check: F) -> io::Result<()>
where
F: FnOnce() -> Fut,
Fut: Future<Output = Result<(), BitrotSelfTestError>>,
{
if !enabled {
gauge!(METRIC_BITROT_SELFTEST_STATUS).set(STATUS_SKIPPED);
debug!(
target: "rustfs::main::run",
event = EVENT_BITROT_SELFTEST,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_STARTUP,
state = "skipped",
reason = "disabled",
"Bitrot self-test skipped"
);
return Ok(());
}
let started = Instant::now();
match run_check().await {
Ok(()) => {
BITROT_SELF_TEST_STATUS.store(STATUS_CELL_PASSED, Ordering::Release);
gauge!(METRIC_BITROT_SELFTEST_STATUS).set(STATUS_PASSED);
info!(
target: "rustfs::main::run",
event = EVENT_BITROT_SELFTEST,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_STARTUP,
state = "passed",
duration_us = started.elapsed().as_micros() as u64,
"Bitrot self-test passed"
);
}
Err(err) => {
BITROT_SELF_TEST_STATUS.store(STATUS_CELL_FAILED, Ordering::Release);
gauge!(METRIC_BITROT_SELFTEST_STATUS).set(STATUS_FAILED);
error!(
target: "rustfs::main::run",
event = EVENT_BITROT_SELFTEST,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_STARTUP,
state = "failed",
duration_us = started.elapsed().as_micros() as u64,
error = %err,
"Bitrot self-test failed"
);
if strict {
return Err(io::Error::other(format!("bitrot self-test failed: {err}")));
}
}
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::{BITROT_SELF_TEST_STATUS, STATUS_CELL_UNSET, bitrot_selftest_passed, run_startup_bitrot_self_test_with};
use crate::storage_api::startup::background::BitrotSelfTestError;
use std::future::ready;
use std::sync::atomic::Ordering;
fn failing_check() -> impl Future<Output = Result<(), BitrotSelfTestError>> {
ready(Err(BitrotSelfTestError::RoundtripReadback {
algorithm: "HighwayHash256S",
}))
}
/// All scenarios run sequentially inside one test: the status cell is
/// process-global, so parallel per-scenario tests would race the reset and
/// read each other's outcomes (the exact order-dependent flake class this
/// module exists to avoid).
#[tokio::test]
async fn startup_self_test_publishes_outcome_and_strict_gates_abort() {
BITROT_SELF_TEST_STATUS.store(STATUS_CELL_UNSET, Ordering::Release);
// Skipped: publishes nothing, never fails, never aborts.
run_startup_bitrot_self_test_with(false, true, || async { Ok(()) })
.await
.expect("a disabled self-test must not fail even in strict mode");
assert_eq!(bitrot_selftest_passed(), None, "a skipped run must leave the status unset");
// Passing: publishes Some(true), never fails.
run_startup_bitrot_self_test_with(true, false, || async { Ok(()) })
.await
.expect("a passing check must never fail startup");
assert_eq!(bitrot_selftest_passed(), Some(true), "a passing run must publish Some(true)");
// Failing, non-strict: publishes Some(false) but startup continues.
run_startup_bitrot_self_test_with(true, false, failing_check)
.await
.expect("a failed check must not abort startup in non-strict mode");
assert_eq!(bitrot_selftest_passed(), Some(false), "a failing run must publish Some(false)");
// Failing, strict: startup error carries the failure and the published
// outcome stays a failure.
let err = run_startup_bitrot_self_test_with(true, true, failing_check)
.await
.expect_err("strict mode must turn a failed check into a startup error");
assert!(err.to_string().contains("bitrot self-test failed"));
assert_eq!(bitrot_selftest_passed(), Some(false));
}
}
+1
View File
@@ -76,6 +76,7 @@ pub mod allocator_reclaim;
pub mod app;
pub mod auth;
pub mod auth_keystone;
pub(crate) mod bitrot_selftest;
pub mod capacity;
pub mod cluster_snapshot;
pub mod config;
+14
View File
@@ -33,6 +33,8 @@ pub(crate) const ENV_SCANNER_ENABLED: &str = "RUSTFS_SCANNER_ENABLED";
pub(crate) const ENV_SCANNER_ENABLED_DEPRECATED: &str = "RUSTFS_ENABLE_SCANNER";
pub(crate) const ENV_HEAL_ENABLED: &str = "RUSTFS_HEAL_ENABLED";
pub(crate) const ENV_HEAL_ENABLED_DEPRECATED: &str = "RUSTFS_ENABLE_HEAL";
pub(crate) const ENV_BITROT_SELFTEST_ENABLE: &str = "RUSTFS_BITROT_SELFTEST_ENABLE";
pub(crate) const ENV_BITROT_SELFTEST_STRICT: &str = "RUSTFS_BITROT_SELFTEST_STRICT";
static AUDIT_MODULE_ENABLED: AtomicBool = AtomicBool::new(rustfs_config::DEFAULT_AUDIT_ENABLE);
static NOTIFY_MODULE_ENABLED: AtomicBool = AtomicBool::new(rustfs_config::DEFAULT_NOTIFY_ENABLE);
@@ -47,6 +49,18 @@ pub(crate) fn heal_enabled_from_env() -> bool {
get_env_bool_with_aliases(ENV_HEAL_ENABLED, &[ENV_HEAL_ENABLED_DEPRECATED], true)
}
/// Whether the startup bitrot algorithm self-test runs, defaulting to on
/// (rustfs/backlog#1873).
pub(crate) fn bitrot_selftest_enabled_from_env() -> bool {
rustfs_utils::get_env_bool(ENV_BITROT_SELFTEST_ENABLE, true)
}
/// Whether a failed bitrot self-test aborts startup instead of only logging
/// and exposing a failed status, defaulting to off.
pub(crate) fn bitrot_selftest_strict_from_env() -> bool {
rustfs_utils::get_env_bool(ENV_BITROT_SELFTEST_STRICT, false)
}
/// Last published audit-module state.
pub fn is_audit_module_enabled() -> bool {
AUDIT_MODULE_ENABLED.load(Ordering::Relaxed)
+10 -1
View File
@@ -12,7 +12,10 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use crate::module_switches::{heal_enabled_from_env, scanner_enabled_from_env};
use crate::bitrot_selftest::run_startup_bitrot_self_test;
use crate::module_switches::{
bitrot_selftest_enabled_from_env, bitrot_selftest_strict_from_env, heal_enabled_from_env, scanner_enabled_from_env,
};
use crate::storage_api::startup::background::{ECStore, set_workload_admission_snapshot_provider};
use crate::workload_admission::RustFsWorkloadAdmissionSnapshotProvider;
use rustfs_concurrency::WorkloadAdmissionSnapshotProvider;
@@ -27,6 +30,12 @@ const LOG_SUBSYSTEM_STARTUP: &str = "startup";
const EVENT_BACKGROUND_SERVICES_CONFIGURED: &str = "background_services_configured";
pub(crate) async fn init_background_service_runtime(store: Arc<ECStore>) -> Result<bool> {
// Pin the bitrot algorithms before anything can write or verify a shard:
// the check costs well under a millisecond, and in strict mode a drifted
// build must abort here rather than after it has touched data
// (rustfs/backlog#1873).
run_startup_bitrot_self_test(bitrot_selftest_enabled_from_env(), bitrot_selftest_strict_from_env()).await?;
let _ = create_ahm_services_cancel_token();
let enable_scanner = scanner_enabled_from_env();
+38 -229
View File
@@ -12,15 +12,15 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use super::StorageVersioningConfigExt as _;
use super::{
BUCKET_ACCELERATE_CONFIG, BUCKET_LOGGING_CONFIG, BUCKET_REQUEST_PAYMENT_CONFIG, BUCKET_VERSIONING_CONFIG,
BUCKET_WEBSITE_CONFIG, BucketVersioningSys, OBJECT_LOCK_CONFIG, StorageError, check_retention_for_modification, decode_tags,
decode_tags_to_map, delete_bucket_metadata_config_if_incarnation, encode_tags, get_bucket_accelerate_config,
get_bucket_logging_config, get_bucket_object_lock_config, get_bucket_request_payment_config, get_bucket_website_config,
is_err_bucket_not_found, is_err_object_not_found, is_err_version_not_found, record_replication_proxy, serialize,
update_bucket_metadata_config_if_incarnation,
get_bucket_logging_config, get_bucket_object_lock_config, get_bucket_replication_config, get_bucket_request_payment_config,
get_bucket_website_config, is_err_bucket_not_found, is_err_object_not_found, is_err_version_not_found,
record_replication_proxy, serialize, update_bucket_metadata_config_if_incarnation,
};
use super::{StorageReplicationConfigExt as _, StorageVersioningConfigExt as _};
use crate::admin::handlers::site_replication::site_replication_bucket_meta_hook;
use crate::error::ApiError;
use crate::storage::access::{apply_bucket_generation_guard, bucket_config_mutation_incarnation, has_bypass_governance_header};
@@ -59,7 +59,7 @@ const LOG_SUBSYSTEM_OBJECT_LOCK: &str = "object_lock";
const LOG_SUBSYSTEM_TAGGING: &str = "tagging";
use crate::app::storage_api::object_usecase::bucket::replication::{
ReplicateDecision, get_read_proxy_targets, must_replicate_metadata, schedule_metadata_replication,
ReplicateDecision, must_replicate_metadata, schedule_metadata_replication,
};
use crate::storage::storage_api::ecfs_consumer::StorageObjectOptions as ObjectOptions;
@@ -105,152 +105,18 @@ impl FS {
&self.server_ctx
}
/// Not-found classifier for proxied SDK tagging calls: a raw 404 covers
/// NoSuchKey and NoSuchVersion alike; the caller silently tries the next
/// replication target.
fn proxy_sdk_error_is_not_found<E>(err: &aws_sdk_s3::error::SdkError<E>) -> bool {
err.raw_response().is_some_and(|resp| resp.status().as_u16() == 404)
async fn replication_tagging_enabled(bucket: &str, object: &str) -> bool {
get_bucket_replication_config(bucket)
.await
.map(|(cfg, _)| cfg.has_active_rules(object, true))
.unwrap_or(false)
}
/// Selector options for a tagging proxy. Reuses `get_opts` so the
/// anti-loop `source-proxy-request` header family and the bucket's
/// version-suspension state gate proxying exactly like GET/HEAD.
async fn tagging_proxy_opts(
bucket: &str,
object: &str,
version_id: Option<String>,
headers: &http::HeaderMap,
) -> Option<ObjectOptions> {
get_opts(bucket, object, version_id, None, headers).await.ok()
}
/// Serve a GetObjectTagging for an object missing locally by proxying to
/// the bucket's replication targets (MinIO `proxyGetTaggingToRepTarget`,
/// backlog#1675 P1-5). None means no target had the object.
async fn proxy_get_object_tagging(
bucket: &str,
object: &str,
version_id: Option<String>,
headers: &http::HeaderMap,
) -> Option<TagSet> {
let opts = Self::tagging_proxy_opts(bucket, object, version_id, headers).await?;
let targets = get_read_proxy_targets(bucket, object, &opts).await;
if targets.is_empty() {
return None;
async fn record_replication_tagging_metric(bucket: &str, object: &str, api: &str, is_err: bool) {
if !Self::replication_tagging_enabled(bucket, object).await {
return;
}
for target in targets {
match target
.get_object_tagging(&target.bucket, object, opts.version_id.clone())
.await
{
Ok(remote) => {
// MinIO-aligned accounting: one total per proxy attempt,
// one failed when no target served it.
record_replication_proxy(bucket, "GetObjectTagging", false).await;
return Some(
remote
.tag_set
.into_iter()
.map(|tag| Tag {
key: Some(tag.key),
value: Some(tag.value),
})
.collect(),
);
}
Err(err) if Self::proxy_sdk_error_is_not_found(&err) => {
debug!(bucket, object, arn = %target.arn, "tagging proxy: target does not have the object");
}
Err(err) => {
warn!(bucket, object, arn = %target.arn, error = %err, "tagging proxy: GetObjectTagging against replication target failed");
}
}
}
record_replication_proxy(bucket, "GetObjectTagging", true).await;
None
}
/// Apply a PutObjectTagging for an object missing locally on a
/// replication target (MinIO `proxyTaggingToRepTarget`).
async fn proxy_put_object_tagging(
bucket: &str,
object: &str,
version_id: Option<String>,
headers: &http::HeaderMap,
tag_set: &TagSet,
) -> Option<()> {
let opts = Self::tagging_proxy_opts(bucket, object, version_id, headers).await?;
let mut tagging = aws_sdk_s3::types::Tagging::builder();
for tag in tag_set {
let sdk_tag = aws_sdk_s3::types::Tag::builder()
.key(tag.key.clone().unwrap_or_default())
.value(tag.value.clone().unwrap_or_default())
.build()
.ok()?;
tagging = tagging.tag_set(sdk_tag);
}
let tagging = tagging.build().ok()?;
let targets = get_read_proxy_targets(bucket, object, &opts).await;
if targets.is_empty() {
return None;
}
for target in targets {
match target
.put_object_tagging(&target.bucket, object, opts.version_id.clone(), tagging.clone())
.await
{
Ok(_) => {
// MinIO-aligned accounting: one total per proxy attempt,
// one failed when no target served it.
record_replication_proxy(bucket, "PutObjectTagging", false).await;
return Some(());
}
Err(err) if Self::proxy_sdk_error_is_not_found(&err) => {
debug!(bucket, object, arn = %target.arn, "tagging proxy: target does not have the object");
}
Err(err) => {
warn!(bucket, object, arn = %target.arn, error = %err, "tagging proxy: PutObjectTagging against replication target failed");
}
}
}
record_replication_proxy(bucket, "PutObjectTagging", true).await;
None
}
/// Apply a DeleteObjectTagging for an object missing locally on a
/// replication target (MinIO `proxyTaggingToRepTarget`).
async fn proxy_delete_object_tagging(
bucket: &str,
object: &str,
version_id: Option<String>,
headers: &http::HeaderMap,
) -> Option<()> {
let opts = Self::tagging_proxy_opts(bucket, object, version_id, headers).await?;
let targets = get_read_proxy_targets(bucket, object, &opts).await;
if targets.is_empty() {
return None;
}
for target in targets {
match target
.delete_object_tagging(&target.bucket, object, opts.version_id.clone())
.await
{
Ok(_) => {
// MinIO-aligned accounting: one total per proxy attempt,
// one failed when no target served it.
record_replication_proxy(bucket, "DeleteObjectTagging", false).await;
return Some(());
}
Err(err) if Self::proxy_sdk_error_is_not_found(&err) => {
debug!(bucket, object, arn = %target.arn, "tagging proxy: target does not have the object");
}
Err(err) => {
warn!(bucket, object, arn = %target.arn, error = %err, "tagging proxy: DeleteObjectTagging against replication target failed");
}
}
}
record_replication_proxy(bucket, "DeleteObjectTagging", true).await;
None
record_replication_proxy(bucket, api, is_err).await;
}
pub async fn get_object_tag_conditions_for_policy(
@@ -581,27 +447,7 @@ impl S3 for FS {
let mut opts = get_opts(&bucket, &object, version_id.clone(), None, &req.headers)
.await
.map_err(ApiError::from)?;
let existing_object_info = match store.get_object_info(&bucket, &object, &opts).await {
Ok(info) => info,
Err(e) => {
// Replication lag window: apply the tagging delete on a
// replication target that already has the object
// (backlog#1675 P1-5). No local object exists, so no bucket
// notification event is emitted for the proxied write.
if (is_err_object_not_found(&e) || is_err_version_not_found(&e))
&& Self::proxy_delete_object_tagging(&bucket, &object, version_id.clone(), &req.headers)
.await
.is_some()
{
counter!("rustfs_delete_object_tagging_success").increment(1);
let duration = start_time.elapsed();
histogram!("rustfs_object_tagging_operation_duration_seconds", "operation" => "delete")
.record(duration.as_secs_f64());
return Ok(S3Response::new(DeleteObjectTaggingOutput { version_id }));
}
return Err(ApiError::from(e).into());
}
};
let existing_object_info = store.get_object_info(&bucket, &object, &opts).await.map_err(ApiError::from)?;
let dsc = must_replicate_metadata(
&bucket,
&object,
@@ -624,6 +470,7 @@ impl S3 for FS {
}
let delete_tags_result = store.delete_object_tags(&bucket, &object, &opts).await;
Self::record_replication_tagging_metric(&bucket, &object, "DeleteObjectTagging", delete_tags_result.is_err()).await;
let object_info = delete_tags_result.map_err(|e| {
error!(
component = LOG_COMPONENT_STORAGE,
@@ -1081,49 +928,32 @@ impl S3 for FS {
..Default::default()
};
let tags = match store.get_object_tags(bucket, object, &opts).await {
Ok(tags) => tags,
Err(e) => {
// Replication lag window: the object may exist on a
// replication target even though it is missing locally —
// proxy the tagging read there (backlog#1675 P1-5).
if (is_err_object_not_found(&e) || is_err_version_not_found(&e))
&& let Some(tag_set) =
Self::proxy_get_object_tagging(bucket, object, req.input.version_id.clone(), &req.headers).await
{
counter!("rustfs_get_object_tagging_success").increment(1);
let duration = start_time.elapsed();
histogram!("rustfs_object_tagging_operation_duration_seconds", "operation" => "get")
.record(duration.as_secs_f64());
return Ok(S3Response::new(GetObjectTaggingOutput {
tag_set,
version_id: req.input.version_id.clone(),
}));
}
if is_err_object_not_found(&e) {
debug!(
component = LOG_COMPONENT_STORAGE,
subsystem = LOG_SUBSYSTEM_TAGGING,
event = "object_tagging_not_found",
bucket = %bucket,
object = %object,
error = %e,
"Object tags not found"
);
return Err(s3_error!(NoSuchKey));
}
error!(
let tags_result = store.get_object_tags(bucket, object, &opts).await;
Self::record_replication_tagging_metric(bucket, object, "GetObjectTagging", tags_result.is_err()).await;
let tags = tags_result.map_err(|e| {
if is_err_object_not_found(&e) {
debug!(
component = LOG_COMPONENT_STORAGE,
subsystem = LOG_SUBSYSTEM_TAGGING,
event = "object_tagging_get_failed",
event = "object_tagging_not_found",
bucket = %bucket,
object = %object,
error = %e,
"Failed to load object tags"
"Object tags not found"
);
return Err(ApiError::from(e).into());
return s3_error!(NoSuchKey);
}
};
error!(
component = LOG_COMPONENT_STORAGE,
subsystem = LOG_SUBSYSTEM_TAGGING,
event = "object_tagging_get_failed",
bucket = %bucket,
object = %object,
error = %e,
"Failed to load object tags"
);
ApiError::from(e).into()
})?;
let tag_set = decode_tags(tags.as_str());
debug!(
@@ -1799,36 +1629,14 @@ impl S3 for FS {
return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
};
let tags = encode_tags(tagging.tag_set.clone());
let tags = encode_tags(tagging.tag_set);
debug!("Encoded tags: {}", tags);
let version_id = req.input.version_id.clone();
let mut opts = get_opts(&bucket, &object, version_id.clone(), None, &req.headers)
.await
.map_err(ApiError::from)?;
let existing_object_info = match store.get_object_info(&bucket, &object, &opts).await {
Ok(info) => info,
Err(e) => {
// Replication lag window: apply the tagging update on a
// replication target that already has the object
// (backlog#1675 P1-5). No local object exists, so no bucket
// notification event is emitted for the proxied write.
if (is_err_object_not_found(&e) || is_err_version_not_found(&e))
&& Self::proxy_put_object_tagging(&bucket, &object, version_id.clone(), &req.headers, &tagging.tag_set)
.await
.is_some()
{
counter!("rustfs_put_object_tagging_success").increment(1);
let duration = start_time.elapsed();
histogram!("rustfs_object_tagging_operation_duration_seconds", "operation" => "put")
.record(duration.as_secs_f64());
return Ok(S3Response::new(PutObjectTaggingOutput {
version_id: req.input.version_id.clone(),
}));
}
return Err(ApiError::from(e).into());
}
};
let existing_object_info = store.get_object_info(&bucket, &object, &opts).await.map_err(ApiError::from)?;
let dsc = must_replicate_metadata(
&bucket,
&object,
@@ -1851,6 +1659,7 @@ impl S3 for FS {
}
let put_tags_result = store.put_object_tags(&bucket, &object, &tags, &opts).await;
Self::record_replication_tagging_metric(&bucket, &object, "PutObjectTagging", put_tags_result.is_err()).await;
let object_info = put_tags_result.map_err(|e| {
error!("Failed to put object tags: {}", e);
counter!("rustfs_put_object_tagging_failure").increment(1);
+18 -18
View File
@@ -57,24 +57,24 @@ pub(crate) use storage_api::{
QuotaError, RUSTFS_META_BUCKET, RawFileInfo, ReadMultipleReq, ReadMultipleResp, ReadOptions, RenameDataResp,
ReplicationStats, ReplicationStatusType, Result, SERVICE_SIGNAL_REFRESH_CONFIG, SERVICE_SIGNAL_RELOAD_DYNAMIC,
StorageDeletedObject, StorageDiskRpcExt, StorageError, StorageGetObjectReader, StorageObjectInfo, StorageObjectOptions,
StorageObjectToDelete, StoragePeerS3ClientExt, StoragePutObjReader, StorageVersioningConfigExt, TONIC_RPC_PREFIX,
TierConfigMgr, UpdateMetadataOpts, VolumeInfo, WalkDirOptions, WorkloadAdmissionSnapshotProviderRef, WriteEncryption,
WritePlan, access_consumer, add_object_lock_years, all_local_disk, all_local_disk_path, check_retention_for_modification,
collect_local_metrics, compression_metadata_value, contract, decode_tags, decode_tags_to_map, delete_bucket_metadata_config,
delete_bucket_metadata_config_if_incarnation, disk_drive_path, disk_endpoint, ecfs_consumer, ecfs_extend_consumer,
ecstore_admin, ecstore_bucket, ecstore_capacity, ecstore_client, ecstore_cluster, ecstore_compression, ecstore_config,
ecstore_data_usage, ecstore_disk, ecstore_error, ecstore_event, ecstore_layout, ecstore_metrics, ecstore_notification,
ecstore_rebalance, ecstore_rio, ecstore_rpc, ecstore_set_disk, ecstore_storage, ecstore_tier, encode_tags,
find_local_disk_by_ref, get_bucket_accelerate_config, get_bucket_cors_config, get_bucket_logging_config, get_bucket_metadata,
get_bucket_notification_config, get_bucket_object_lock_config, get_bucket_request_payment_config, get_bucket_sse_config,
get_bucket_website_config, get_local_server_property, get_lock_acquire_timeout, head_prefix_consumer, helper_consumer,
init_background_replication, init_bucket_metadata_sys, init_ecstore_config, init_local_disks_with_instance_ctx,
init_lock_clients, is_err_bucket_not_found, is_err_object_not_found, is_err_version_not_found, is_valid_storage_class,
options_consumer, prewarm_local_disk_id_map_with_instance_ctx, read_config, record_replication_proxy, rpc_consumer,
runtime_sources_consumer, s3_api_consumer, serialize, table_catalog_path_hash, to_s3s_etag,
topology_snapshot_from_endpoint_pools_with_capabilities, try_migrate_bucket_metadata, try_migrate_iam_config,
try_migrate_server_config, update_bucket_metadata_config, update_bucket_metadata_config_if_incarnation, verify_rpc_signature,
wrap_reader,
StorageObjectToDelete, StoragePeerS3ClientExt, StoragePutObjReader, StorageReplicationConfigExt, StorageVersioningConfigExt,
TONIC_RPC_PREFIX, TierConfigMgr, UpdateMetadataOpts, VolumeInfo, WalkDirOptions, WorkloadAdmissionSnapshotProviderRef,
WriteEncryption, WritePlan, access_consumer, add_object_lock_years, all_local_disk, all_local_disk_path,
check_retention_for_modification, collect_local_metrics, compression_metadata_value, contract, decode_tags,
decode_tags_to_map, delete_bucket_metadata_config, delete_bucket_metadata_config_if_incarnation, disk_drive_path,
disk_endpoint, ecfs_consumer, ecfs_extend_consumer, ecstore_admin, ecstore_bucket, ecstore_capacity, ecstore_client,
ecstore_cluster, ecstore_compression, ecstore_config, ecstore_data_usage, ecstore_disk, ecstore_error, ecstore_event,
ecstore_layout, ecstore_metrics, ecstore_notification, ecstore_rebalance, ecstore_rio, ecstore_rpc, ecstore_set_disk,
ecstore_storage, ecstore_tier, encode_tags, find_local_disk_by_ref, get_bucket_accelerate_config, get_bucket_cors_config,
get_bucket_logging_config, get_bucket_metadata, get_bucket_notification_config, get_bucket_object_lock_config,
get_bucket_replication_config, get_bucket_request_payment_config, get_bucket_sse_config, get_bucket_website_config,
get_local_server_property, get_lock_acquire_timeout, head_prefix_consumer, helper_consumer, init_background_replication,
init_bucket_metadata_sys, init_ecstore_config, init_local_disks_with_instance_ctx, init_lock_clients,
is_err_bucket_not_found, is_err_object_not_found, is_err_version_not_found, is_valid_storage_class, options_consumer,
prewarm_local_disk_id_map_with_instance_ctx, read_config, record_replication_proxy, rpc_consumer, runtime_sources_consumer,
s3_api_consumer, serialize, table_catalog_path_hash, to_s3s_etag, topology_snapshot_from_endpoint_pools_with_capabilities,
try_migrate_bucket_metadata, try_migrate_iam_config, try_migrate_server_config, update_bucket_metadata_config,
update_bucket_metadata_config_if_incarnation, verify_rpc_signature, wrap_reader,
};
#[cfg(test)]
+3 -93
View File
@@ -19,10 +19,9 @@ use http::{HeaderMap, HeaderValue};
use rustfs_utils::http::{
AMZ_BUCKET_REPLICATION_STATUS, SUFFIX_FORCE_DELETE, SUFFIX_OBJECTLOCK_LEGALHOLD_TIMESTAMP,
SUFFIX_OBJECTLOCK_RETENTION_TIMESTAMP, SUFFIX_REPLICATION_ACTUAL_OBJECT_SIZE, SUFFIX_REPLICATION_SSEC_CRC,
SUFFIX_SOURCE_DELETEMARKER, SUFFIX_SOURCE_ETAG, SUFFIX_SOURCE_MTIME, SUFFIX_SOURCE_PROXY_REQUEST,
SUFFIX_SOURCE_REPLICATION_LEGALHOLD_TIMESTAMP, SUFFIX_SOURCE_REPLICATION_REQUEST,
SUFFIX_SOURCE_REPLICATION_RETENTION_TIMESTAMP, SUFFIX_SOURCE_REPLICATION_TAGGING_TIMESTAMP, SUFFIX_SOURCE_VERSION_ID,
SUFFIX_TAGGING_TIMESTAMP, get_header,
SUFFIX_SOURCE_DELETEMARKER, SUFFIX_SOURCE_ETAG, SUFFIX_SOURCE_MTIME, SUFFIX_SOURCE_REPLICATION_LEGALHOLD_TIMESTAMP,
SUFFIX_SOURCE_REPLICATION_REQUEST, SUFFIX_SOURCE_REPLICATION_RETENTION_TIMESTAMP,
SUFFIX_SOURCE_REPLICATION_TAGGING_TIMESTAMP, SUFFIX_SOURCE_VERSION_ID, SUFFIX_TAGGING_TIMESTAMP, get_header,
header_compat::{MINIO_ENCRYPTION_PREFIX, RUSTFS_ENCRYPTION_PREFIX},
insert_header_map, insert_str,
metadata_compat::{MINIO_INTERNAL_PREFIX, RUSTFS_INTERNAL_PREFIX},
@@ -277,19 +276,6 @@ pub async fn get_opts(
// Background scanner still performs full integrity checks asynchronously.
opts.skip_verify_bitrot = get_skip_verify_bitrot();
// Anti-loop markers for the replication read proxy
// (`{x-rustfs-,x-minio-}source-proxy-request` header family).
// MinIO semantics: the header being PRESENT at all (`ProxyHeaderSet`)
// disables proxying, whatever its value — a peer's replication worker
// sends "false" on its convergence HEADs so the receiver answers locally
// instead of proxying the miss back (a proxied echo would fake
// convergence and the object would never replicate). Deliberately not
// gated on replication authorization: the header only disables proxying
// (it grants nothing).
let proxy_header = get_header(headers, SUFFIX_SOURCE_PROXY_REQUEST);
opts.proxy_header_set = proxy_header.is_some();
opts.proxy_request = proxy_header.map(|v| v.as_ref() == "true").unwrap_or_default();
fill_conditional_writes_opts_from_header(headers, &mut opts)?;
Ok(opts)
@@ -2558,80 +2544,4 @@ mod tests {
}
}
}
/// The replication read-proxy anti-loop markers must be honored under
/// both interop prefixes (a MinIO peer sends x-minio-, a RustFS peer
/// sends both). `proxy_request` is set only for the literal value
/// "true", while `proxy_header_set` (MinIO `ProxyHeaderSet`) is set by
/// the header's mere presence — "false" (the replication worker's
/// convergence-HEAD marker) and arbitrary values included — so the
/// selector refuses to proxy either way.
#[tokio::test]
async fn test_get_opts_parses_source_proxy_request_under_both_prefixes() {
for header_name in ["x-rustfs-source-proxy-request", "x-minio-source-proxy-request"] {
let mut headers = HeaderMap::new();
headers.insert(header_name, HeaderValue::from_static("true"));
let opts = get_opts("test-bucket", "test-object", None, None, &headers)
.await
.expect("get_opts should succeed");
assert!(opts.proxy_request, "{header_name} must set opts.proxy_request");
assert!(opts.proxy_header_set, "{header_name} must set opts.proxy_header_set");
}
let opts = get_opts("test-bucket", "test-object", None, None, &HeaderMap::new())
.await
.expect("get_opts should succeed");
assert!(!opts.proxy_request, "absent header must leave proxy_request off");
assert!(!opts.proxy_header_set, "absent header must leave proxy_header_set off");
for (header_name, value) in [
("x-minio-source-proxy-request", "false"),
("x-rustfs-source-proxy-request", "false"),
("x-minio-source-proxy-request", "anything-else"),
] {
let mut headers = HeaderMap::new();
headers.insert(header_name, HeaderValue::from_static(value));
let opts = get_opts("test-bucket", "test-object", None, None, &headers)
.await
.expect("get_opts should succeed");
assert!(!opts.proxy_request, "{header_name}: non-'true' value must leave proxy_request off");
assert!(
opts.proxy_header_set,
"{header_name}: value {value:?} must still set proxy_header_set (presence disables proxying)"
);
}
}
/// Pin that the source-proxy-request transport family cannot be
/// materialized as bare stored metadata via an `x-*-meta-` disguise: the
/// reserved-key namespacing (`x-rustfs-source-` / `x-minio-source-`
/// prefixes in `is_reserved_user_metadata_key`) must keep covering it.
#[test]
fn test_source_proxy_request_family_is_reserved_user_metadata() {
let mut headers = HeaderMap::new();
headers.insert("x-amz-meta-x-minio-source-proxy-request", HeaderValue::from_static("true"));
headers.insert("x-rustfs-meta-x-rustfs-source-proxy-request", HeaderValue::from_static("true"));
// The bare transport header itself is not a user-metadata prefix and
// must never land in stored metadata at all.
headers.insert("x-minio-source-proxy-request", HeaderValue::from_static("true"));
let metadata = extract_metadata(&headers);
assert!(
!metadata.contains_key("x-minio-source-proxy-request"),
"bare source-proxy-request key must not be storable: {metadata:?}"
);
assert!(
!metadata.contains_key("x-rustfs-source-proxy-request"),
"bare source-proxy-request key must not be storable: {metadata:?}"
);
assert!(
metadata.contains_key("x-amz-meta-x-minio-source-proxy-request"),
"disguised key must be namespaced back under x-amz-meta-: {metadata:?}"
);
assert!(
metadata.contains_key("x-amz-meta-x-rustfs-source-proxy-request"),
"disguised key must be namespaced back under x-amz-meta-: {metadata:?}"
);
}
}
+22 -8
View File
@@ -569,6 +569,10 @@ pub(crate) mod ecstore_erasure {
pub(crate) use rustfs_ecstore::api::erasure::{BitrotReader, Erasure};
}
/// Startup bitrot algorithm self-test (rustfs/backlog#1873), re-exported for
/// the root facade's background-startup section.
pub(crate) use rustfs_ecstore::api::erasure::{BitrotSelfTestError, bitrot_self_test};
pub(crate) mod ecstore_storage {
#[cfg(test)]
pub(crate) use rustfs_ecstore::api::storage::init_local_disks;
@@ -805,10 +809,6 @@ impl StorageReplicationStatsHandle {
proxy_head_failed: metrics.proxied.head_failed,
proxy_put_tag_total: metrics.proxied.put_tag_total,
proxy_put_tag_failed: metrics.proxied.put_tag_failed,
proxy_get_tag_total: metrics.proxied.get_tag_total,
proxy_get_tag_failed: metrics.proxied.get_tag_failed,
proxy_delete_tag_total: metrics.proxied.delete_tag_total,
proxy_delete_tag_failed: metrics.proxied.delete_tag_failed,
replica_size: metrics.replica_size,
replica_count: metrics.replica_count,
}
@@ -845,10 +845,6 @@ pub(crate) struct ReplicationSiteMetricsSnapshot {
pub(crate) proxy_head_failed: i64,
pub(crate) proxy_put_tag_total: i64,
pub(crate) proxy_put_tag_failed: i64,
pub(crate) proxy_get_tag_total: i64,
pub(crate) proxy_get_tag_failed: i64,
pub(crate) proxy_delete_tag_total: i64,
pub(crate) proxy_delete_tag_failed: i64,
pub(crate) replica_size: i64,
pub(crate) replica_count: i64,
}
@@ -1495,6 +1491,12 @@ pub(crate) async fn get_bucket_object_lock_config(
ecstore_bucket::metadata_sys::get_object_lock_config(bucket).await
}
pub(crate) async fn get_bucket_replication_config(
bucket: &str,
) -> Result<(s3s::dto::ReplicationConfiguration, time::OffsetDateTime)> {
ecstore_bucket::metadata_sys::get_replication_config(bucket).await
}
pub(crate) async fn persist_force_delete_intent(
api: Arc<ECStore>,
entry: ecstore_bucket::replication::MrfReplicateEntry,
@@ -1840,6 +1842,18 @@ pub(crate) async fn find_local_disk_by_ref(disk_ref: &str) -> Option<DiskStore>
ecstore_storage::find_local_disk_by_ref(disk_ref).await
}
pub(crate) trait StorageReplicationConfigExt {
fn has_active_rules(&self, prefix: &str, recursive: bool) -> bool;
}
impl StorageReplicationConfigExt for s3s::dto::ReplicationConfiguration {
fn has_active_rules(&self, prefix: &str, recursive: bool) -> bool {
<s3s::dto::ReplicationConfiguration as ecstore_bucket::replication::ReplicationConfigurationExt>::has_active_rules(
self, prefix, recursive,
)
}
}
pub(crate) trait StorageVersioningConfigExt {
fn enabled(&self) -> bool;
}
+3 -1
View File
@@ -214,7 +214,9 @@ pub(crate) mod startup {
}
pub(crate) mod background {
pub(crate) use crate::storage::storage_api::{ECStore, set_workload_admission_snapshot_provider};
pub(crate) use crate::storage::storage_api::{
BitrotSelfTestError, ECStore, bitrot_self_test, set_workload_admission_snapshot_provider,
};
}
pub(crate) mod bucket_metadata {