Compare commits

..

3 Commits

Author SHA1 Message Date
overtrue 3dc92c6dff test(e2e): update security suite membership 2026-08-23 06:18:36 +08:00
overtrue d299b82b21 test(e2e): require tier SSRF guard evidence 2026-08-23 05:40:27 +08:00
overtrue fd6046faf2 test(e2e): pin multipart part number limit 2026-08-23 05:33:25 +08:00
21 changed files with 143 additions and 1041 deletions
+2 -2
View File
@@ -1,2 +1,2 @@
sha256-darwin=9f767b37ed8b1c82da62ea441462d75487785c8086e56f08fb6f6cd89c6e2e52
sha256-linux=fbdaf42b220958d4b1e8880e0f8b5a7992d38e21051bb60596dd4538424757d6
sha256-darwin=b8549d3362a69cca01c2a81f548bb06d5142d8a9ab4509487a656c8b3db1c164
sha256-linux=7ecd054965b4afa070af6deefdc37b5ca9f6a9b488dd5eef1ad0877378365b2f
+1 -2
View File
@@ -30,8 +30,7 @@ make build-docker BUILD_OS=ubuntu22.04
- Crate membership: `Cargo.toml` `[workspace].members`
- Architecture, layering, crate map: [ARCHITECTURE.md](ARCHITECTURE.md)
- Migration guardrails & readiness contracts: [docs/architecture/](docs/architecture/README.md)
- CI workflow steps: `.github/workflows/`; event, timeout, and required-status
matrix: [docs/testing/ci-gates.md](docs/testing/ci-gates.md)
- CI gates: `.github/workflows/ci.yml` (source of truth; never copy its steps into docs)
- Test-layer taxonomy, per-layer entry commands, serial/nextest rules, flake
policy: [docs/testing/README.md](docs/testing/README.md)
- Tier/ILM transition debugging (xl.meta inspection, versionId tracing):
-2
View File
@@ -70,8 +70,6 @@ make pre-pr
> For the full test-layer taxonomy (unit / ecstore black-box / e2e / s3s-e2e / S3 compatibility / chaos / fuzz / bench), each layer's entry command, the naming conventions the migration gate depends on, and the serial/nextest rules, see [docs/testing/README.md](docs/testing/README.md).
> For the event, timeout, required-status, and local reproduction matrix, see [docs/testing/ci-gates.md](docs/testing/ci-gates.md).
### 🔒 Automated Pre-commit Hooks
#### What `make pre-commit` and `make pre-pr` actually run
Generated
+27 -22
View File
@@ -1858,9 +1858,9 @@ dependencies = [
[[package]]
name = "cc"
version = "1.4.4"
version = "1.4.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0ad534f4357a5264cce5019c989cf66a4f0dc4e0d1b1d15f8aacec0ff7360273"
checksum = "509591b7bcd67f4ef775afad7662703b4935daaa6ec0e5605cfb1090b32a2b6d"
dependencies = [
"find-msvc-tools",
"jobserver",
@@ -2522,6 +2522,12 @@ dependencies = [
"subtle",
]
[[package]]
name = "cty"
version = "0.2.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b365fabc795046672053e29c954733ec3b05e4be654ab130fe8f1f94d7051f35"
[[package]]
name = "curve25519-dalek"
version = "4.1.3"
@@ -5982,6 +5988,15 @@ version = "0.2.16"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b6d2cec3eae94f9f509c767b45932f1ada8350c4bdb85af2fcab4a3c14807981"
[[package]]
name = "libmimalloc-sys"
version = "0.1.49"
source = "git+https://github.com/xonatius/mimalloc_rust.git?rev=6d4c41bb10c6d9da1d1b6f07b38c4cc051667f11#6d4c41bb10c6d9da1d1b6f07b38c4cc051667f11"
dependencies = [
"cc",
"cty",
]
[[package]]
name = "libredox"
version = "0.1.20"
@@ -6382,6 +6397,14 @@ dependencies = [
"synstructure 0.13.2",
]
[[package]]
name = "mimalloc"
version = "0.1.52"
source = "git+https://github.com/xonatius/mimalloc_rust.git?rev=6d4c41bb10c6d9da1d1b6f07b38c4cc051667f11#6d4c41bb10c6d9da1d1b6f07b38c4cc051667f11"
dependencies = [
"libmimalloc-sys",
]
[[package]]
name = "mime"
version = "0.3.17"
@@ -9139,11 +9162,13 @@ dependencies = [
"insta",
"jiff",
"libc",
"libmimalloc-sys",
"libsystemd",
"matchit 0.9.2",
"md-5 0.11.0",
"metrics",
"metrics-util",
"mimalloc",
"mime_guess",
"opentelemetry",
"opentelemetry_sdk",
@@ -9179,8 +9204,6 @@ dependencies = [
"rustfs-lock",
"rustfs-log-analyzer",
"rustfs-madmin",
"rustfs-mimalloc",
"rustfs-mimalloc-sys",
"rustfs-notify",
"rustfs-object-capacity",
"rustfs-object-data-cache",
@@ -9852,24 +9875,6 @@ dependencies = [
"tokio",
]
[[package]]
name = "rustfs-mimalloc"
version = "0.5.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a406f4aa07084301d485beec873af6dccc8e3f8762da244743df92038b1db1a6"
dependencies = [
"rustfs-mimalloc-sys",
]
[[package]]
name = "rustfs-mimalloc-sys"
version = "0.5.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c3051b819175f58445d4c369a72f0ab88149f3885ba8bea2aff3be01f53fe7cd"
dependencies = [
"cc",
]
[[package]]
name = "rustfs-notify"
version = "1.0.0-rc.3"
+2 -2
View File
@@ -350,8 +350,8 @@ russh-sftp = "2.4.0"
dav-server = "0.11.0"
# Performance Analysis and Memory Profiling
rustfs-mimalloc = { version = "0.5.0" }
rustfs-mimalloc-sys = { version = "0.5.0" }
mimalloc = { version = "0.1.52", git = "https://github.com/xonatius/mimalloc_rust.git", rev = "6d4c41bb10c6d9da1d1b6f07b38c4cc051667f11" }
libmimalloc-sys = { version = "0.1.49", git = "https://github.com/xonatius/mimalloc_rust.git", rev = "6d4c41bb10c6d9da1d1b6f07b38c4cc051667f11", features = ["extended"] }
hotpath = { version = "0.23.3", default-features = false }
# Snapshot testing for output format regression detection
insta = { version = "1.48" }
-5
View File
@@ -61,11 +61,6 @@ mod get_codec_streaming_compat_test;
#[cfg(test)]
mod version_id_regression_test;
// Receiver-side replication LWW (rustfs/backlog#1953): stale inbound
// replication metadata must not overwrite a newer local category state.
#[cfg(test)]
mod replication_lww_receiver_test;
// Data usage regression tests
#[cfg(test)]
mod data_usage_test;
@@ -1,150 +0,0 @@
#![cfg(test)]
// 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.
//! Receiver-side replication LWW over the wire (rustfs/backlog#1953, audit
//! A4/P1-6).
//!
//! In an active-active topology both sites' metadata states arrive at the
//! peer as authorized replication PUTs carrying per-category source
//! timestamps (`x-rustfs-source-replication-tagging-timestamp` header
//! family). Before the fix the receiver applied them unconditionally, so a
//! stale delivery overwrote a newer local state and the two sites diverged
//! permanently while both reported COMPLETED. This test drives one live
//! `rustfs` server with simulated inbound replication PUTs for the same
//! object version and asserts the newer tagging state wins regardless of
//! delivery order, while a stale delivery still succeeds at the object level
//! (a failure would loop through MRF re-delivering the stale value).
use crate::common::{RustFSTestEnvironment, init_logging};
use aws_sdk_s3::Client;
use aws_sdk_s3::primitives::ByteStream;
use aws_sdk_s3::types::{BucketVersioningStatus, VersioningConfiguration};
type TestResult = Result<(), Box<dyn std::error::Error + Send + Sync>>;
const HDR_SOURCE_REPLICATION_REQUEST: &str = "x-rustfs-source-replication-request";
const HDR_SOURCE_VERSION_ID: &str = "x-rustfs-source-version-id";
const HDR_SOURCE_MTIME: &str = "x-rustfs-source-mtime";
const HDR_SOURCE_TAGGING_TIMESTAMP: &str = "x-rustfs-source-replication-tagging-timestamp";
const SOURCE_MTIME: &str = "2026-01-01T00:00:00Z";
const T_STALE: &str = "2026-01-01T00:00:01Z";
const T_LOCAL: &str = "2026-02-01T00:00:00Z";
const T_NEWER: &str = "2026-03-01T00:00:00Z";
/// Simulated inbound authorized replication PUT: same object version, tags and
/// the source-authored tagging timestamp carried in transport headers.
async fn inbound_replication_put(
client: &Client,
bucket: &str,
key: &str,
version_id: &str,
tags: &str,
tagging_timestamp: &str,
) -> TestResult {
let version_id = version_id.to_string();
let tagging_timestamp = tagging_timestamp.to_string();
client
.put_object()
.bucket(bucket)
.key(key)
.body(ByteStream::from_static(b"lww-e2e-body"))
.tagging(tags)
.customize()
.mutate_request(move |req| {
req.headers_mut().insert(HDR_SOURCE_REPLICATION_REQUEST, "true");
req.headers_mut().insert(HDR_SOURCE_VERSION_ID, version_id.clone());
req.headers_mut().insert(HDR_SOURCE_MTIME, SOURCE_MTIME);
req.headers_mut()
.insert(HDR_SOURCE_TAGGING_TIMESTAMP, tagging_timestamp.clone());
})
.send()
.await?;
Ok(())
}
async fn tag_value(client: &Client, bucket: &str, key: &str, version_id: &str, tag_key: &str) -> Option<String> {
let tagging = client
.get_object_tagging()
.bucket(bucket)
.key(key)
.version_id(version_id)
.send()
.await
.expect("object tagging should be readable");
tagging
.tag_set()
.iter()
.find(|tag| tag.key() == tag_key)
.map(|tag| tag.value().to_string())
}
#[tokio::test(flavor = "multi_thread")]
async fn receiver_lww_keeps_newer_tags_across_delivery_orders() -> TestResult {
init_logging();
let mut env = RustFSTestEnvironment::new().await?;
env.start_rustfs_server(vec![]).await?;
let client = env.create_s3_client();
let bucket = "replication-lww-receiver";
let key = "object";
client.create_bucket().bucket(bucket).send().await?;
client
.put_bucket_versioning()
.bucket(bucket)
.versioning_configuration(
VersioningConfiguration::builder()
.status(BucketVersioningStatus::Enabled)
.build(),
)
.send()
.await?;
// First delivery establishes version V with tags stamped T_LOCAL.
let version_id = uuid::Uuid::new_v4().to_string();
inbound_replication_put(&client, bucket, key, &version_id, "site=local", T_LOCAL).await?;
assert_eq!(
tag_value(&client, bucket, key, &version_id, "site").await.as_deref(),
Some("local"),
"the first delivery must establish the tagged version"
);
// A stale delivery (older source timestamp) must succeed at the object
// level but must NOT overwrite the newer tags.
inbound_replication_put(&client, bucket, key, &version_id, "site=stale", T_STALE).await?;
assert_eq!(
tag_value(&client, bucket, key, &version_id, "site").await.as_deref(),
Some("local"),
"a stale inbound delivery must not overwrite newer tags (rustfs/backlog#1953)"
);
// A newer delivery still converges the version onto the newest state.
inbound_replication_put(&client, bucket, key, &version_id, "site=newer", T_NEWER).await?;
assert_eq!(
tag_value(&client, bucket, key, &version_id, "site").await.as_deref(),
Some("newer"),
"a newer inbound delivery must overwrite older tags"
);
client
.delete_object()
.bucket(bucket)
.key(key)
.version_id(&version_id)
.send()
.await?;
env.delete_test_bucket(bucket).await.ok();
Ok(())
}
+33 -23
View File
@@ -17,14 +17,14 @@
//! These tests verify that RustFS properly enforces security-sensitive
//! controls by issuing real requests against a running server and asserting
//! the concrete outcome of each control:
//! - DoS protection (oversized tagging payloads, excessive multipart parts)
//! - DoS protection (oversized tagging payloads, out-of-range multipart part numbers)
//! - SSRF prevention (internal/private endpoints rejected for tiering)
//! - Race condition handling (concurrent writes converge without corruption)
use crate::common::{RustFSTestEnvironment, awscurl_available, awscurl_put, init_logging};
use aws_sdk_s3::error::ProvideErrorMetadata;
use aws_sdk_s3::primitives::ByteStream;
use aws_sdk_s3::types::{CompletedMultipartUpload, CompletedPart, Tag, Tagging};
use aws_sdk_s3::types::{Tag, Tagging};
use std::error::Error;
use tracing::info;
@@ -88,9 +88,9 @@ async fn test_large_xml_body_rejection() -> Result<(), Box<dyn Error + Send + Sy
Ok(())
}
/// Excessive multipart parts must be rejected.
/// Multipart part numbers above the S3 limit must be rejected.
#[tokio::test]
async fn test_excessive_multipart_parts() -> Result<(), Box<dyn Error + Send + Sync>> {
async fn test_multipart_part_number_above_limit() -> Result<(), Box<dyn Error + Send + Sync>> {
init_logging();
let mut env = RustFSTestEnvironment::new().await?;
env.start_rustfs_server(vec![]).await?;
@@ -108,18 +108,23 @@ async fn test_excessive_multipart_parts() -> Result<(), Box<dyn Error + Send + S
let upload_id = create_result.upload_id().expect("upload_id should be present").to_string();
// Try to complete with too many parts (should be rejected).
let mut parts = Vec::new();
for i in 1..=10001 {
parts.push(CompletedPart::builder().part_number(i).e_tag(format!("etag-{i}")).build());
}
let result = client
.complete_multipart_upload()
client
.upload_part()
.bucket(&bucket_name)
.key("test-large")
.upload_id(&upload_id)
.multipart_upload(CompletedMultipartUpload::builder().set_parts(Some(parts)).build())
.part_number(10000)
.body(ByteStream::from_static(b"upper-bound part"))
.send()
.await?;
let result = client
.upload_part()
.bucket(&bucket_name)
.key("test-large")
.upload_id(&upload_id)
.part_number(10001)
.body(ByteStream::from_static(b"out-of-range part"))
.send()
.await;
@@ -133,7 +138,13 @@ async fn test_excessive_multipart_parts() -> Result<(), Box<dyn Error + Send + S
.await;
let _ = client.delete_bucket().bucket(&bucket_name).send().await;
assert!(result.is_err(), "Server should reject excessive multipart parts");
let err = result.expect_err("server must reject excessive multipart parts");
let code = err.as_service_error().and_then(ProvideErrorMetadata::code);
assert_eq!(
code,
Some("InvalidArgument"),
"Part number 10001 should be rejected with InvalidArgument, got code {code:?}, err: {err:?}"
);
env.stop_server();
Ok(())
@@ -217,12 +228,8 @@ async fn test_concurrent_object_operations() -> Result<(), Box<dyn Error + Send
/// Internal/private endpoints must be rejected as remote tier backends (SSRF).
///
/// This issues a real admin AddTier call (`PUT /rustfs/admin/v3/tier`) for each
/// internal/private endpoint and asserts the server rejects it (non-2xx, so the
/// signed request helper returns an error). An internal endpoint must never be
/// accepted as a tier backend. The rejection may originate from explicit
/// SSRF/internal-address filtering or from the backend connectivity/credential
/// validation performed during AddTier; either way the security-relevant
/// outcome — the internal endpoint is not accepted — is asserted here.
/// internal/private endpoint and asserts the request reaches the outbound URL
/// guard. Connectivity or credential failures do not prove SSRF protection.
///
/// The admin API is exercised via signed `awscurl` requests, matching the
/// pattern used by the other admin-API E2E tests in this crate; the test is
@@ -263,10 +270,13 @@ async fn test_tiering_url_validation() -> Result<(), Box<dyn Error + Send + Sync
})
.to_string();
let result = awscurl_put(&tier_url, &body, &env.access_key, &env.secret_key).await;
let err = awscurl_put(&tier_url, &body, &env.access_key, &env.secret_key)
.await
.expect_err("AddTier must reject internal endpoints");
let rendered = err.to_string();
assert!(
result.is_err(),
"AddTier must reject internal endpoint {endpoint}, but it was accepted: {result:?}"
rendered.contains("TierAddFailed") && rendered.contains("tier endpoint is not allowed"),
"AddTier rejected {endpoint} outside the outbound URL guard: {rendered}"
);
}
@@ -3868,7 +3868,6 @@ async fn replicate_object_with_multipart<S: ReplicationObjectIO>(ctx: MultipartR
actual_size,
object_info.etag.clone().unwrap_or_default(),
object_info.mod_time,
&put_opts.internal,
),
)
.await
@@ -472,7 +472,6 @@ pub(crate) fn replication_complete_multipart_options(
actual_size: String,
source_etag: String,
source_mtime: Option<OffsetDateTime>,
source_internal: &AdvancedPutOptions,
) -> PutObjectOptions {
let mut user_metadata = HashMap::new();
insert_header_map(&mut user_metadata, SUFFIX_REPLICATION_ACTUAL_OBJECT_SIZE, actual_size);
@@ -485,14 +484,6 @@ pub(crate) fn replication_complete_multipart_options(
// mtime must degrade to epoch so header() suppresses the header
// instead of asserting the replication time as the object's mtime.
source_mtime: source_mtime.unwrap_or(OffsetDateTime::UNIX_EPOCH),
// Carry the per-category LWW timestamps on the complete request as
// well: the receiver's CompleteMultipartUpload options builder
// parses the same headers, so the multipart transport gets the
// same receiver-side LWW as the single-PUT transport
// (rustfs/backlog#1953). Epoch values keep the headers suppressed.
tagging_timestamp: source_internal.tagging_timestamp,
retention_timestamp: source_internal.retention_timestamp,
legalhold_timestamp: source_internal.legalhold_timestamp,
replication_status: ReplicationStatusType::Replica,
replication_request: true,
..Default::default()
@@ -672,39 +663,20 @@ mod tests {
#[test]
fn replication_complete_multipart_options_sets_actual_size() {
let source_mtime = OffsetDateTime::from_unix_timestamp(1_716_170_000).expect("valid test timestamp");
let source_internal = AdvancedPutOptions {
tagging_timestamp: OffsetDateTime::from_unix_timestamp(1_716_170_100).expect("valid test timestamp"),
retention_timestamp: OffsetDateTime::from_unix_timestamp(1_716_170_200).expect("valid test timestamp"),
legalhold_timestamp: OffsetDateTime::from_unix_timestamp(1_716_170_300).expect("valid test timestamp"),
..Default::default()
};
let options = replication_complete_multipart_options(
"1024".to_string(),
"0123456789abcdef0123456789abcdef-3".to_string(),
Some(source_mtime),
&source_internal,
);
assert_eq!(options.internal.source_etag, "0123456789abcdef0123456789abcdef-3");
assert_eq!(options.internal.source_mtime, source_mtime);
// The complete request must carry the same per-category LWW timestamps
// as the initiate request; the receiver reads them from the complete
// headers (rustfs/backlog#1953).
assert_eq!(options.internal.tagging_timestamp, source_internal.tagging_timestamp);
assert_eq!(options.internal.retention_timestamp, source_internal.retention_timestamp);
assert_eq!(options.internal.legalhold_timestamp, source_internal.legalhold_timestamp);
// Absent source mtime must degrade to epoch (header suppressed), not
// the AdvancedPutOptions default of now_utc() — that default would
// stamp the replication time as the replica's mtime and break the
// multipart HEAD convergence. Unset category timestamps stay epoch so
// header() keeps suppressing them.
let options_no_mtime =
replication_complete_multipart_options("1024".to_string(), String::new(), None, &AdvancedPutOptions::default());
// multipart HEAD convergence.
let options_no_mtime = replication_complete_multipart_options("1024".to_string(), String::new(), None);
assert_eq!(options_no_mtime.internal.source_mtime.unix_timestamp(), 0);
assert_eq!(options_no_mtime.internal.tagging_timestamp.unix_timestamp(), 0);
assert_eq!(options_no_mtime.internal.retention_timestamp.unix_timestamp(), 0);
assert_eq!(options_no_mtime.internal.legalhold_timestamp.unix_timestamp(), 0);
assert_eq!(
get_header_map(&options.user_metadata, SUFFIX_REPLICATION_ACTUAL_OBJECT_SIZE).as_deref(),
@@ -2318,57 +2318,6 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
fi.set_data_moved();
}
// Receiver-side LWW (rustfs/backlog#1953): the multipart replication
// transport carries the category values at CreateMultipartUpload (in
// the staged upload metadata) and the source category timestamps on
// the complete request. Read the destination version under the held
// object write lock and keep any category this site modified more
// recently. Read failures (version absent on first replication, quorum
// errors) keep today's overwrite semantics: failing the complete would
// loop through MRF, re-delivering the stale value forever.
if crate::set_disk::ops::object::replication_lww_applicable(opts)
&& let Some(version_id) = fi.version_id
{
match self
.get_object_info(
bucket,
object,
&ObjectOptions {
version_id: Some(version_id.to_string()),
no_lock: true,
metadata_cache_safe: false,
versioned: opts.versioned,
version_suspended: opts.version_suspended,
..Default::default()
},
)
.await
{
Ok(existing) => {
let stored = crate::set_disk::ops::object::stored_replication_category_metadata(&existing);
crate::set_disk::ops::object::merge_replication_metadata_lww(&mut fi.metadata, &stored, opts);
}
// Version absent: first replication of this version, nothing
// local to compare — the normal path, not a degraded one.
Err(err) if is_err_object_not_found(&err) || is_err_version_not_found(&err) => {}
Err(err) => {
// Degraded path: without the stored state the inbound
// metadata is applied unchanged — exactly the overwrite
// LWW exists to prevent — so this must be operator-visible.
warn!(
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_SET_DISK,
bucket,
object,
version_id = %version_id,
error = %err,
state = "replication_lww_read_unavailable",
"SetDisk multipart replication LWW read skipped; inbound metadata applied without comparison"
);
}
}
}
for meta in parts_metadatas.iter_mut() {
if meta.has_valid_erasure_geometry() {
meta.size = fi.size;
@@ -7106,97 +7055,6 @@ mod tests {
.await
}
/// Receiver-side LWW on the multipart replication transport
/// (rustfs/backlog#1953): a metadata-only replication of a multipart
/// source object rides CreateMultipartUpload (category values in the
/// upload metadata) + CompleteMultipartUpload (category timestamps in
/// the complete options). A stale inbound tagging timestamp must not
/// overwrite a newer locally-tagged destination version.
#[tokio::test]
#[serial]
async fn complete_multipart_upload_stale_replication_tags_keep_local() {
use rustfs_utils::http::headers::AMZ_OBJECT_TAGGING;
use rustfs_utils::http::{SUFFIX_TAGGING_TIMESTAMP, get_str};
use time::format_description::well_known::Rfc3339;
const T_OLD: &str = "2026-01-01T00:00:00Z";
const T_LOCAL: &str = "2026-02-01T00:00:00Z";
let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
let bucket = "multipart-replication-lww-bucket";
let object = "object";
make_bucket_on_all(&disk_stores, bucket).await;
// Local destination version with newer tags.
let version_id = Uuid::new_v4();
let mut local_metadata = HashMap::new();
local_metadata.insert(AMZ_OBJECT_TAGGING.to_string(), "site=local".to_string());
rustfs_utils::http::insert_str(&mut local_metadata, SUFFIX_TAGGING_TIMESTAMP, T_LOCAL.to_string());
let mut local_reader = PutObjReader::from_vec(b"local body".to_vec());
set_disks
.put_object(
bucket,
object,
&mut local_reader,
&ObjectOptions {
versioned: true,
version_id: Some(version_id.to_string()),
user_defined: local_metadata,
// Explicit-version PUTs require the bucket Object Lock snapshot.
object_lock_config_snapshot: Some(Arc::new(crate::set_disk::ObjectLockConfigSnapshot::new(
crate::bucket::metadata_sys::ObjectLockConfigState::ConfirmedAbsent,
))),
..Default::default()
},
)
.await
.expect("local versioned put should commit");
// Inbound replication upload carrying older tags for the same version.
let mut inbound_metadata = HashMap::new();
inbound_metadata.insert(AMZ_OBJECT_TAGGING.to_string(), "site=remote".to_string());
rustfs_utils::http::insert_str(&mut inbound_metadata, SUFFIX_TAGGING_TIMESTAMP, T_OLD.to_string());
let create_opts = ObjectOptions {
versioned: true,
user_defined: inbound_metadata,
..Default::default()
};
let (upload_id, parts) =
stage_upload_with_create_opts(&set_disks, bucket, object, &payload(0x5a), &create_opts).await;
rewrite_staged_upload_version_id(&set_disks, bucket, object, &upload_id, Some(version_id)).await;
let complete_opts = ObjectOptions {
versioned: true,
replication_request: true,
replication_tagging_timestamp: Some(OffsetDateTime::parse(T_OLD, &Rfc3339).expect("test timestamp should parse")),
..Default::default()
};
set_disks
.clone()
.complete_multipart_upload(bucket, object, &upload_id, parts, &complete_opts)
.await
.expect("replication multipart completion should succeed even when a category keeps local values");
let info = set_disks
.get_object_info(
bucket,
object,
&ObjectOptions {
versioned: true,
version_id: Some(version_id.to_string()),
..Default::default()
},
)
.await
.expect("completed version should be readable");
assert_eq!(
info.user_tags.as_str(),
"site=local",
"older inbound multipart tags must not overwrite newer local tags"
);
assert_eq!(get_str(&info.user_defined, SUFFIX_TAGGING_TIMESTAMP).as_deref(), Some(T_LOCAL));
}
#[tokio::test]
#[serial]
async fn complete_multipart_upload_assigns_completion_version_id() {
+14 -472
View File
@@ -1881,110 +1881,6 @@ fn delete_file_info_with_replication_transport_metadata(fi: &FileInfo) -> FileIn
transported
}
/// True when an authorized replication write carries at least one per-category
/// source timestamp, i.e. receiver-side LWW has something to judge.
pub(in crate::set_disk) fn replication_lww_applicable(opts: &ObjectOptions) -> bool {
opts.replication_request
&& (opts.replication_tagging_timestamp.is_some()
|| opts.replication_retention_timestamp.is_some()
|| opts.replication_legalhold_timestamp.is_some())
}
/// The stored per-category state of a destination version, as compared by
/// [`merge_replication_metadata_lww`]. `ObjectInfo::from_file_info`
/// externalizes tags into `user_tags` (stripping the metadata key), so the
/// tag value is folded back into map form here.
pub(in crate::set_disk) fn stored_replication_category_metadata(existing: &ObjectInfo) -> HashMap<String, String> {
let mut stored = (*existing.user_defined).clone();
if !existing.user_tags.is_empty() {
stored.insert(rustfs_utils::http::headers::AMZ_OBJECT_TAGGING.to_string(), (*existing.user_tags).clone());
}
stored
}
/// Receiver-side last-writer-wins for authorized replication writes
/// (rustfs/backlog#1953, audit A4/P1-6). Metadata-only replication reuses the
/// whole-object transports, so in active-active topologies an inbound write
/// carries the source's tags / retention / legal hold verbatim and would
/// otherwise overwrite a category the destination modified more recently —
/// both sites end up permanently diverged while reporting COMPLETED.
///
/// Judged per category, only when the inbound request carries that category's
/// source timestamp (`ObjectOptions::replication_*_timestamp`):
/// - stored timestamp newer than inbound: the local category values and
/// timestamp are kept; the rest of the write proceeds per the inbound
/// metadata and the object-level result stays successful (failing the write
/// instead would loop through MRF, re-delivering the stale value forever);
/// - otherwise the inbound category wins and its internal timestamp key is
/// pinned to the source-authored time — the PUT path re-stamps the
/// object-lock timestamps with the receiver's clock
/// (`parse_object_lock_retention` / `parse_object_lock_legal_hold` insert
/// `now()` via `eval_metadata`), which would make the replica's clock the
/// LWW authority and wedge later convergence;
/// - no stored timestamp (pre-P1-6 data) or no inbound timestamp: the current
/// overwrite behavior is preserved.
///
/// Returns whether `inbound` was modified. Callers must hold the object write
/// lock so the stored values compared here are the ones being replaced.
pub(in crate::set_disk) fn merge_replication_metadata_lww(
inbound: &mut HashMap<String, String>,
existing: &HashMap<String, String>,
opts: &ObjectOptions,
) -> bool {
use rustfs_utils::http::headers::{
AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER, AMZ_OBJECT_LOCK_MODE_LOWER, AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER, AMZ_OBJECT_TAGGING,
};
use rustfs_utils::http::metadata_compat::{
SUFFIX_OBJECTLOCK_LEGALHOLD_TIMESTAMP, SUFFIX_OBJECTLOCK_RETENTION_TIMESTAMP, SUFFIX_TAGGING_TIMESTAMP, get_str,
remove_str,
};
use time::format_description::well_known::Rfc3339;
let categories: [(Option<OffsetDateTime>, &str, &[&str]); 3] = [
(opts.replication_tagging_timestamp, SUFFIX_TAGGING_TIMESTAMP, &[AMZ_OBJECT_TAGGING]),
(
opts.replication_retention_timestamp,
SUFFIX_OBJECTLOCK_RETENTION_TIMESTAMP,
&[AMZ_OBJECT_LOCK_MODE_LOWER, AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER],
),
(
opts.replication_legalhold_timestamp,
SUFFIX_OBJECTLOCK_LEGALHOLD_TIMESTAMP,
&[AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER],
),
];
let mut changed = false;
for (inbound_timestamp, timestamp_suffix, value_keys) in categories {
let Some(inbound_timestamp) = inbound_timestamp else { continue };
let is_category_value_key = |key: &str| value_keys.iter().any(|value_key| key.eq_ignore_ascii_case(value_key));
let stored_timestamp = get_str(existing, timestamp_suffix).and_then(|value| OffsetDateTime::parse(&value, &Rfc3339).ok());
if stored_timestamp.is_some_and(|stored| stored > inbound_timestamp) {
inbound.retain(|key, _| !is_category_value_key(key));
remove_str(inbound, timestamp_suffix);
for (key, value) in existing {
if is_category_value_key(key) {
inbound.insert(key.clone(), value.clone());
}
}
// Restore the winning timestamp via insert_str, not a verbatim key
// copy: a MinIO-written version may carry only the
// x-minio-internal- key, and the dual-key invariant requires every
// write to produce both keys.
if let Some(stored_value) = get_str(existing, timestamp_suffix) {
rustfs_utils::http::insert_str(inbound, timestamp_suffix, stored_value);
}
changed = true;
} else if let Ok(source_authored) = inbound_timestamp.format(&Rfc3339)
&& get_str(inbound, timestamp_suffix).as_deref() != Some(source_authored.as_str())
{
rustfs_utils::http::insert_str(inbound, timestamp_suffix, source_authored);
changed = true;
}
}
changed
}
impl SetDisks {
pub(in crate::set_disk) async fn persist_old_data_cleanup_receipts(
&self,
@@ -2228,13 +2124,26 @@ impl SetDisks {
let put_object_size = known_put_object_storage_size(data.size());
let shard_file_size_raw = erasure.shard_file_size(put_object_size);
let is_inline_buffer = storage_class_config.should_inline(shard_file_size_raw, erasure.data_shards, opts.versioned);
let is_inline_buffer =
storage_class_config.should_inline(shard_file_size_raw, erasure.data_shards, opts.versioned);
let collect_stage_timing = rustfs_io_metrics::put_stage_metrics_enabled() || issue3031_diag_enabled();
let shard_file_size = shard_file_size_raw;
let shard_size = erasure.shard_size();
let write_path = classify_put_write_path(is_inline_buffer, put_object_size, fi.erasure.block_size);
let direct_inline_commit = matches!(write_path, SmallWritePath::Inline);
{
use std::io::Write;
let msg = format!(
"INLINE_DEBUG: bucket={} obj={} size={} shard_fs={} ds={} bs={} inline={} direct={} path={} iblock={} ver={}\n",
bucket, object, put_object_size, shard_file_size_raw, erasure.data_shards, fi.erasure.block_size,
is_inline_buffer, direct_inline_commit, write_path.metric_label(), storage_class_config.inline_block(), opts.versioned
);
if let Ok(mut f) = std::fs::OpenOptions::new().create(true).append(true).open("/tmp/rustfs_inline_debug.log") {
let _ = f.write_all(msg.as_bytes());
}
let _ = std::io::stderr().write_all(msg.as_bytes());
}
rustfs_io_metrics::record_put_object_path(write_path.metric_label());
let writer_setup_stage_start = collect_stage_timing.then(Instant::now);
let (mut writers, errors) = if direct_inline_commit {
@@ -2666,22 +2575,6 @@ impl SetDisks {
if check_object_lock_for_deletion_with_state(object_lock_config.state(), &existing, false)?.is_some() {
return Err(StorageError::PrefixAccessDenied(bucket.to_string(), object.to_string()));
}
// Receiver-side LWW (rustfs/backlog#1953): reuse this
// commit-lock read of the destination version so a
// category (tags / retention / legal hold) modified
// more recently on this site is kept instead of being
// overwritten by the inbound replication metadata.
if replication_lww_applicable(opts) {
let stored = stored_replication_category_metadata(&existing);
let mut merged = parts_metadatas[response_metadata_slot].metadata.clone();
if merge_replication_metadata_lww(&mut merged, &stored, opts) {
for (pfi, disk) in parts_metadatas.iter_mut().zip(shuffle_disks.iter()) {
if disk.is_some() {
pfi.metadata = merged.clone();
}
}
}
}
}
Err(err) if is_err_object_not_found(&err) || is_err_version_not_found(&err) => {}
Err(err) => return Err(err),
@@ -8186,357 +8079,6 @@ mod replication_quota_safety_tests {
}
}
#[cfg(test)]
mod replication_lww_tests {
//! Receiver-side LWW for authorized replication writes (rustfs/backlog#1953,
//! audit A4/P1-6): an inbound replication PUT whose per-category timestamp
//! (tags / retention / legal hold) is older than the destination version's
//! stored timestamp must keep the local category values instead of
//! overwriting them; categories are judged independently and the write
//! itself still succeeds.
use super::hermetic_set_disks_support::hermetic_set_disks_isolated as hermetic_set_disks;
use super::*;
use crate::storage_api_contracts::object::{ObjectIO as _, ObjectOperations as _};
use rustfs_utils::http::headers::{
AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER, AMZ_OBJECT_LOCK_MODE_LOWER, AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER, AMZ_OBJECT_TAGGING,
};
use rustfs_utils::http::{
SUFFIX_OBJECTLOCK_LEGALHOLD_TIMESTAMP, SUFFIX_OBJECTLOCK_RETENTION_TIMESTAMP, SUFFIX_TAGGING_TIMESTAMP, get_str,
insert_str,
};
use time::format_description::well_known::Rfc3339;
const T_OLD: &str = "2026-01-01T00:00:00Z";
const T_LOCAL: &str = "2026-02-01T00:00:00Z";
const T_NEW: &str = "2026-03-01T00:00:00Z";
fn parse_ts(value: &str) -> OffsetDateTime {
OffsetDateTime::parse(value, &Rfc3339).expect("test timestamp should parse")
}
async fn make_bucket(disks: &[DiskStore], bucket: &str) {
for disk in disks {
disk.make_volume(bucket).await.expect("bucket volume should be created");
}
}
async fn put_version(set_disks: &Arc<SetDisks>, bucket: &str, object: &str, version_id: &str, opts: &ObjectOptions) {
let mut reader = PutObjReader::from_vec(b"lww-body".to_vec());
set_disks
.put_object(bucket, object, &mut reader, opts)
.await
.expect("versioned put should commit");
assert_eq!(opts.version_id.as_deref(), Some(version_id));
}
fn versioned_opts(version_id: &str, user_defined: HashMap<String, String>) -> ObjectOptions {
ObjectOptions {
versioned: true,
version_id: Some(version_id.to_string()),
user_defined,
// Explicit-version PUTs require the bucket Object Lock snapshot.
object_lock_config_snapshot: Some(Arc::new(ObjectLockConfigSnapshot::new(
crate::bucket::metadata_sys::ObjectLockConfigState::ConfirmedAbsent,
))),
..Default::default()
}
}
/// Local state: version `version_id` with tags "site=local" stamped `T_LOCAL`.
async fn seed_local_tagged_version(set_disks: &Arc<SetDisks>, bucket: &str, object: &str, version_id: &str) {
let mut user_defined = HashMap::new();
user_defined.insert(AMZ_OBJECT_TAGGING.to_string(), "site=local".to_string());
insert_str(&mut user_defined, SUFFIX_TAGGING_TIMESTAMP, T_LOCAL.to_string());
put_version(set_disks, bucket, object, version_id, &versioned_opts(version_id, user_defined)).await;
}
fn inbound_tagging_opts(version_id: &str, tags: &str, timestamp: &str) -> ObjectOptions {
let mut user_defined = HashMap::new();
user_defined.insert(AMZ_OBJECT_TAGGING.to_string(), tags.to_string());
insert_str(&mut user_defined, SUFFIX_TAGGING_TIMESTAMP, timestamp.to_string());
ObjectOptions {
replication_request: true,
replication_tagging_timestamp: Some(parse_ts(timestamp)),
..versioned_opts(version_id, user_defined)
}
}
async fn version_info(set_disks: &Arc<SetDisks>, bucket: &str, object: &str, version_id: &str) -> ObjectInfo {
set_disks
.get_object_info(bucket, object, &versioned_opts(version_id, HashMap::new()))
.await
.expect("version should be readable")
}
#[tokio::test]
async fn inbound_stale_tagging_keeps_newer_local_tags() {
let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
let bucket = "lww-tagging-stale";
let object = "object";
let version_id = Uuid::new_v4().to_string();
make_bucket(&disk_stores, bucket).await;
seed_local_tagged_version(&set_disks, bucket, object, &version_id).await;
put_version(
&set_disks,
bucket,
object,
&version_id,
&inbound_tagging_opts(&version_id, "site=remote", T_OLD),
)
.await;
let info = version_info(&set_disks, bucket, object, &version_id).await;
assert_eq!(
info.user_tags.as_str(),
"site=local",
"older inbound tags must not overwrite newer local tags"
);
assert_eq!(
get_str(&info.user_defined, SUFFIX_TAGGING_TIMESTAMP).as_deref(),
Some(T_LOCAL),
"the winning local tagging timestamp must be preserved"
);
}
#[tokio::test]
async fn inbound_newer_tagging_overwrites_local_tags() {
let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
let bucket = "lww-tagging-newer";
let object = "object";
let version_id = Uuid::new_v4().to_string();
make_bucket(&disk_stores, bucket).await;
seed_local_tagged_version(&set_disks, bucket, object, &version_id).await;
put_version(
&set_disks,
bucket,
object,
&version_id,
&inbound_tagging_opts(&version_id, "site=remote", T_NEW),
)
.await;
let info = version_info(&set_disks, bucket, object, &version_id).await;
assert_eq!(
info.user_tags.as_str(),
"site=remote",
"newer inbound tags must overwrite older local tags"
);
assert_eq!(get_str(&info.user_defined, SUFFIX_TAGGING_TIMESTAMP).as_deref(), Some(T_NEW));
}
#[tokio::test]
async fn inbound_wins_when_local_has_no_tagging_timestamp() {
let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
let bucket = "lww-tagging-no-local-ts";
let object = "object";
let version_id = Uuid::new_v4().to_string();
make_bucket(&disk_stores, bucket).await;
// Pre-P1-6 data: local tags without a stored tagging timestamp.
let mut user_defined = HashMap::new();
user_defined.insert(AMZ_OBJECT_TAGGING.to_string(), "site=local".to_string());
put_version(&set_disks, bucket, object, &version_id, &versioned_opts(&version_id, user_defined)).await;
put_version(
&set_disks,
bucket,
object,
&version_id,
&inbound_tagging_opts(&version_id, "site=remote", T_OLD),
)
.await;
let info = version_info(&set_disks, bucket, object, &version_id).await;
assert_eq!(
info.user_tags.as_str(),
"site=remote",
"without a local timestamp the inbound category must win (pre-LWW data compatibility)"
);
}
#[tokio::test]
async fn categories_are_judged_independently() {
let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
let bucket = "lww-category-independent";
let object = "object";
let version_id = Uuid::new_v4().to_string();
make_bucket(&disk_stores, bucket).await;
// Local: newer tags (T_LOCAL), older *cleared* retention (T_OLD) —
// timestamp key only, the shape a replicated retention clear stores.
// (An active local retention would already block the overwrite at the
// WORM gate; the LWW-reachable retention states are cleared/expired.)
let mut local = HashMap::new();
local.insert(AMZ_OBJECT_TAGGING.to_string(), "site=local".to_string());
insert_str(&mut local, SUFFIX_TAGGING_TIMESTAMP, T_LOCAL.to_string());
insert_str(&mut local, SUFFIX_OBJECTLOCK_RETENTION_TIMESTAMP, T_OLD.to_string());
put_version(&set_disks, bucket, object, &version_id, &versioned_opts(&version_id, local)).await;
// Inbound: older tags (T_OLD), newer retention (T_NEW).
let mut inbound = HashMap::new();
inbound.insert(AMZ_OBJECT_TAGGING.to_string(), "site=remote".to_string());
insert_str(&mut inbound, SUFFIX_TAGGING_TIMESTAMP, T_OLD.to_string());
inbound.insert(AMZ_OBJECT_LOCK_MODE_LOWER.to_string(), "COMPLIANCE".to_string());
inbound.insert(AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER.to_string(), "2028-01-01T00:00:00Z".to_string());
insert_str(&mut inbound, SUFFIX_OBJECTLOCK_RETENTION_TIMESTAMP, T_NEW.to_string());
let opts = ObjectOptions {
replication_request: true,
replication_tagging_timestamp: Some(parse_ts(T_OLD)),
replication_retention_timestamp: Some(parse_ts(T_NEW)),
..versioned_opts(&version_id, inbound)
};
put_version(&set_disks, bucket, object, &version_id, &opts).await;
let info = version_info(&set_disks, bucket, object, &version_id).await;
assert_eq!(info.user_tags.as_str(), "site=local", "the stale tagging category must keep local values");
assert_eq!(
info.user_defined.get(AMZ_OBJECT_LOCK_MODE_LOWER).map(String::as_str),
Some("COMPLIANCE"),
"the newer retention category must be applied in the same write"
);
assert_eq!(get_str(&info.user_defined, SUFFIX_OBJECTLOCK_RETENTION_TIMESTAMP).as_deref(), Some(T_NEW));
}
#[tokio::test]
async fn inbound_stale_legal_hold_keeps_local_value() {
let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
let bucket = "lww-legalhold-stale";
let object = "object";
let version_id = Uuid::new_v4().to_string();
make_bucket(&disk_stores, bucket).await;
// Local: legal hold released (OFF) at T_LOCAL. (A local hold that is
// still ON already blocks the overwrite at the WORM gate; the
// LWW-reachable divergence is a stale inbound ON resurrecting a hold
// that was released more recently on this site.)
let mut local = HashMap::new();
local.insert(AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER.to_string(), "OFF".to_string());
insert_str(&mut local, SUFFIX_OBJECTLOCK_LEGALHOLD_TIMESTAMP, T_LOCAL.to_string());
put_version(&set_disks, bucket, object, &version_id, &versioned_opts(&version_id, local)).await;
let mut inbound = HashMap::new();
inbound.insert(AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER.to_string(), "ON".to_string());
insert_str(&mut inbound, SUFFIX_OBJECTLOCK_LEGALHOLD_TIMESTAMP, T_OLD.to_string());
let opts = ObjectOptions {
replication_request: true,
replication_legalhold_timestamp: Some(parse_ts(T_OLD)),
..versioned_opts(&version_id, inbound)
};
put_version(&set_disks, bucket, object, &version_id, &opts).await;
let info = version_info(&set_disks, bucket, object, &version_id).await;
assert_eq!(
info.user_defined.get(AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER).map(String::as_str),
Some("OFF"),
"a stale inbound legal hold must not resurrect a hold released more recently"
);
assert_eq!(
get_str(&info.user_defined, SUFFIX_OBJECTLOCK_LEGALHOLD_TIMESTAMP).as_deref(),
Some(T_LOCAL)
);
}
/// Dual-key invariant under LWW: a MinIO-written destination version may
/// carry only the x-minio-internal timestamp key; when the local category
/// wins, the restored map must still hold BOTH compatibility keys.
#[test]
fn local_win_restores_both_internal_timestamp_keys_for_minio_only_metadata() {
let mut inbound = HashMap::new();
inbound.insert(AMZ_OBJECT_TAGGING.to_string(), "site=remote".to_string());
insert_str(&mut inbound, SUFFIX_TAGGING_TIMESTAMP, T_OLD.to_string());
let existing = HashMap::from([
(AMZ_OBJECT_TAGGING.to_string(), "site=local".to_string()),
("X-Minio-Internal-Tagging-Timestamp".to_string(), T_LOCAL.to_string()),
]);
let opts = ObjectOptions {
replication_request: true,
replication_tagging_timestamp: Some(parse_ts(T_OLD)),
..Default::default()
};
assert!(merge_replication_metadata_lww(&mut inbound, &existing, &opts));
assert_eq!(inbound.get(AMZ_OBJECT_TAGGING).map(String::as_str), Some("site=local"));
assert_eq!(
inbound.get("x-rustfs-internal-tagging-timestamp").map(String::as_str),
Some(T_LOCAL),
"the RustFS twin key must be materialized even when the source version only had the MinIO key"
);
assert_eq!(inbound.get("x-minio-internal-tagging-timestamp").map(String::as_str), Some(T_LOCAL));
}
/// When the inbound category wins, the stored timestamp must be the
/// source-authored one: the PUT path's eval_metadata stamps the
/// object-lock timestamps with the receiver's clock
/// (`parse_object_lock_retention`), which would otherwise make this
/// replica's clock the LWW authority and wedge later convergence.
#[tokio::test]
async fn inbound_win_pins_stored_timestamp_to_source_authored_value() {
let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
let bucket = "lww-retention-ts-pinned";
let object = "object";
let version_id = Uuid::new_v4().to_string();
make_bucket(&disk_stores, bucket).await;
// Local cleared retention at T_OLD.
let mut local = HashMap::new();
insert_str(&mut local, SUFFIX_OBJECTLOCK_RETENTION_TIMESTAMP, T_OLD.to_string());
put_version(&set_disks, bucket, object, &version_id, &versioned_opts(&version_id, local)).await;
// Inbound newer retention: the source authored T_LOCAL, but the PUT
// path's eval_metadata stomped the metadata key with receiver-now
// (simulated by T_NEW here).
let mut inbound = HashMap::new();
inbound.insert(AMZ_OBJECT_LOCK_MODE_LOWER.to_string(), "GOVERNANCE".to_string());
inbound.insert(AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER.to_string(), "2028-01-01T00:00:00Z".to_string());
insert_str(&mut inbound, SUFFIX_OBJECTLOCK_RETENTION_TIMESTAMP, T_NEW.to_string());
let opts = ObjectOptions {
replication_request: true,
replication_retention_timestamp: Some(parse_ts(T_LOCAL)),
..versioned_opts(&version_id, inbound)
};
put_version(&set_disks, bucket, object, &version_id, &opts).await;
let info = version_info(&set_disks, bucket, object, &version_id).await;
assert_eq!(
get_str(&info.user_defined, SUFFIX_OBJECTLOCK_RETENTION_TIMESTAMP).as_deref(),
Some(T_LOCAL),
"the stored category timestamp must be the source-authored time, not the receiver's clock"
);
assert_eq!(info.user_defined.get(AMZ_OBJECT_LOCK_MODE_LOWER).map(String::as_str), Some("GOVERNANCE"));
}
#[tokio::test]
async fn newer_local_tag_deletion_survives_stale_inbound_tags() {
let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
let bucket = "lww-tagging-deleted";
let object = "object";
let version_id = Uuid::new_v4().to_string();
make_bucket(&disk_stores, bucket).await;
// Local DeleteObjectTagging state: no tags, but a newer tagging timestamp.
let mut local = HashMap::new();
insert_str(&mut local, SUFFIX_TAGGING_TIMESTAMP, T_LOCAL.to_string());
put_version(&set_disks, bucket, object, &version_id, &versioned_opts(&version_id, local)).await;
put_version(
&set_disks,
bucket,
object,
&version_id,
&inbound_tagging_opts(&version_id, "site=remote", T_OLD),
)
.await;
let info = version_info(&set_disks, bucket, object, &version_id).await;
assert!(
info.user_tags.is_empty(),
"a newer local tag deletion must not be resurrected by older inbound tags"
);
assert_eq!(get_str(&info.user_defined, SUFFIX_TAGGING_TIMESTAMP).as_deref(), Some(T_LOCAL));
}
}
#[cfg(test)]
mod inline_put_commit_path_tests {
use super::hermetic_set_disks_support::hermetic_set_disks_isolated as hermetic_set_disks;
+1 -1
View File
@@ -3194,7 +3194,7 @@ impl ECStore {
// Default return value
let mut del_objects = vec![DeletedObject::default(); objects.len()];
let accounting = vec![None; objects.len()];
let mut accounting = vec![None; objects.len()];
let mut del_errs = Vec::with_capacity(objects.len());
for _ in 0..objects.len() {
@@ -271,7 +271,7 @@ pub(super) fn resolve_latest_object_info_candidates(
.filter(|candidate| latest_candidate_mod_time(candidate) == Some(latest_mod_time))
.collect::<Vec<_>>();
latest_candidates.sort_by_key(|candidate| std::cmp::Reverse(candidate.idx));
latest_candidates.sort_by(|left, right| right.idx.cmp(&left.idx));
let Some(winner) = latest_candidates.first() else {
return Err(Error::ErasureReadQuorum);
+3
View File
@@ -43,6 +43,9 @@ allow-git = [
# RustFS fork carrying presigned expiry and constant-time authentication fixes.
# owner: rustfs-maintainers review: 2026-10
"https://github.com/rustfs/s3s.git",
# MiMalloc fork pinned for hotpath allocation counting support.
# owner: houseme review: 2026-10
"https://github.com/xonatius/mimalloc_rust.git",
]
[bans]
-149
View File
@@ -1,149 +0,0 @@
# CI gate matrix
This file is the source of truth for which validation runs on each event, its
configured wall-clock budget, and whether it can block a merge. Test taxonomy,
naming, and nextest serialization rules remain in [README.md](README.md); e2e
membership and counts remain in
[e2e-suite-inventory.md](e2e-suite-inventory.md).
The distinction between **required** and **report-only** is load-bearing:
a failing job blocks a merge only when its exact check name is present in the
live `main` ruleset. A workflow name, a `merge_group` trigger, or a red PR check
does not make a job required by itself.
## Required merge checks
The live `main` ruleset (`6436880`) currently requires exactly these contexts:
| Required context | Producer | Validation |
|---|---|---|
| `CLA Check` | `.github/workflows/cla.yml` | Contributor agreement |
| `Quick Checks` | `.github/workflows/ci.yml` | Formatting and repository guard scripts |
| `Test and Lint` | `.github/workflows/ci.yml` | Clippy, workspace nextest excluding `e2e_test`, doctests, and migration proofs |
For pull requests limited to the paths excluded by the main CI workflow,
`.github/workflows/ci-docs-only.yml` reports `Quick Checks` and
`Test and Lint` under the same names. It runs the real quick checks and the
planning-document guard; it does not claim that Rust compilation or runtime
tests ran. Despite the workflow name, these paths also include selected deploy,
workflow, and lock files.
Verify the live rule rather than trusting this snapshot before changing merge
policy:
```bash
gh api repos/rustfs/rustfs/rulesets/6436880 \
--jq '.rules[] | select(.type == "required_status_checks") | .parameters'
```
The ruleset currently has `strict_required_status_checks_policy=false`.
`Continuous Integration` accepts `merge_group` events and runs `e2e-full` for
them, but `End-to-End Tests (full merge gate)` is not currently a required
context. Therefore the repository is prepared to test a merge-queue SHA, but
the workflow alone does not prove that every merge passed that lane.
## Pull request and merge matrix
Budgets below are job `timeout-minutes`, not typical runtimes. “Report-only”
means the result is visible and actionable but is not in the live required
context list.
| Event | Validation | Budget | Merge status | Reproduction |
|---|---|---:|---|---|
| PR, non-doc change | `Quick Checks` | 10 min | Required | `make pre-commit` (broader local umbrella) |
| PR, non-doc change | `Test and Lint` | 90 min | Required | `cargo nextest run --profile ci --all --exclude e2e_test` |
| PR, non-doc change | `Typos` | 10 min | Report-only | `typos` |
| PR, non-doc change | `ILM Integration (serial)` | 90 min | Report-only | Use the exact command in `.github/workflows/ci.yml` |
| PR, non-doc change | rio-v2 / swift / sftp test-and-lint variants | 90 min each | Report-only | `cargo nextest run` with the workflow's feature set |
| PR, non-doc change | `Build RustFS Debug Binary` | 30 min | Report-only; prerequisite for black-box lanes | `cargo build -p rustfs --bins` |
| PR, non-doc change | `io_uring Integration (real)` | 30 min | Report-only | `cargo test -p rustfs-ecstore --lib uring_ -- --test-threads=1 --nocapture` |
| PR, non-doc change | `End-to-End Tests` (`e2e-smoke` plus `s3s-e2e`) | 30 min | Report-only | `cargo nextest run --profile e2e-smoke -p e2e_test`; then `./scripts/e2e-run.sh ./target/debug/rustfs <data-dir>` |
| PR, non-doc change | `S3 Implemented Tests` | 60 min | Report-only | Build `rustfs`, then run `scripts/s3-tests/run.sh` with `DEPLOY_MODE=binary`, `TEST_MODE=single`, and `MAXFAIL=0` |
| PR, non-doc change | `S3 Lifecycle Behavior Tests` | 30 min | Report-only | Use the accelerated scanner environment in `.github/workflows/ci.yml` with `scripts/s3-tests/run.sh` |
| PR touching dependency or workflow inputs | Cargo Deny / Workflow Pin Report / Dependency Review | 20 / 5 / 30 min | Report-only | `cargo deny check`; `scripts/security/check_workflow_pins.sh` |
| PR touching architecture rules or architecture docs | `Architecture Migration Rules` | 10 min | Report-only | `scripts/check_architecture_migration_rules.sh` |
| PR touching Nix or workspace manifests | `Nix Build & Check` | 60 min | Report-only | `nix flake check` |
| PR limited to main-CI-excluded paths | companion `Quick Checks` and `Test and Lint` | 10 min each | Required | `git diff --check`; `make doc-paths-check` when documentation paths changed |
| `merge_group` | Standard CI plus `e2e-full` | 55 min for `e2e-full` | Standard required contexts only; `e2e-full` report-only | `cargo nextest run --profile e2e-full -p e2e_test` |
| Push to `main` | Standard CI plus `e2e-full` | 55 min for `e2e-full` | Post-merge detection | Same as `merge_group` |
| PR touching fuzz inputs or harness paths | Build plus five 60-second fuzz smoke targets | 60 min build; 30 min per target | Report-only | `MAX_TOTAL_TIME=60 ./scripts/fuzz/run.sh` |
| PR touching selected ecstore disk/format paths | `Rename Safety` on Windows | 60 min | Report-only | Run the four `cargo test -p rustfs-ecstore --lib <filter>` commands in `windows-filesystem.yml` on Windows |
The authoritative e2e filters live in `.config/nextest.toml`; extend a profile
instead of adding a second ad-hoc selector. Before a profile runs,
`scripts/check_test_wiring.py` compares its exact membership to the committed
digest so a silent test drop fails closed.
## Scheduled and manual validation
Scheduled lanes are independent fault domains. They do not block a pull
request, but their workflow-local gate can fail the run and scheduled failures
are routed to the shared failure-issue action. The scheduled-validation
watchdog and freshness workflow separately detect incomplete runs and missing
schedules.
| Cadence (UTC unless noted) | Workflow / validation | Budget | Verdict and artifacts | Reproduction |
|---|---|---:|---|---|
| Daily 02:17 | Fuzz: five nightly corpus targets | 60 min build; 60 min per target | Gate; corpus/crash artifacts, scheduled failure alert | `MAX_TOTAL_TIME=<seconds> ./scripts/fuzz/run.sh` |
| Daily 03:17 | MinIO interop (EC + SSE read parity) | 40 min | Gate; scheduled failure alert | Dispatch `minio-interop.yml` or follow its pinned Docker fixture steps |
| Daily 04:29 | Replication / cluster-fault / protocol e2e | 45 / 90 / 90 min | Three independent gates; JUnit, membership, and server logs | `cargo nextest run --profile e2e-repl-nightly -p e2e_test`; `--profile e2e-nightly`; `-j 1 --profile e2e-protocols` |
| Daily 06:31 | Warp performance A/B | 180 min | Regression budget gate; A/B summaries and server logs | `bash scripts/run_hotpath_warp_abba.sh --help` |
| Daily 00:07 Asia/Shanghai (16:07 UTC previous day) | Nightly GNU build and Vault lanes | 150 / 90 / 60 min | Build, live Vault, and HA failover gates | Use the commands and pinned Vault images in `nightly-gnu.yml` |
| Daily 03:23 | Security Audit | 20 / 5 min, plus 30 min on PR dependency review | Cargo Deny and workflow-pin gates; scheduled failure alert | `cargo deny check`; `scripts/security/check_workflow_pins.sh` |
| Daily 23:47 | Scheduled Validation Freshness | 10 min | Fails when a critical schedule was never created or is stale | Dispatch `scheduled-validation-freshness.yml` |
| Sunday 00:11 | Full `Continuous Integration` matrix | Per-job budgets above | Weekly variant coverage, including dormant rio-v2 binary/e2e lanes | Dispatch `ci.yml` |
| Sunday 01:13 | Seven-platform build matrix | 150 min per platform | Build/package integrity; scheduled failure alert | Dispatch `build.yml` with an exact platform set |
| Sunday 02:19 | Ceph s3-tests full sweep: single and real four-node, four shards each | 180 min per shard | Compatibility gate; report, JUnit, exact node IDs, and server logs | `scripts/s3-tests/run.sh` against an existing single or distributed target |
| Sunday 06:41 | Mint | 120 min | **Report-only by design**; per-suite PASS/FAIL/NA and raw `log.json` | Reproduce the pinned Docker sequence in `mint.yml` or dispatch it |
| Sunday 07:43 | Workspace line coverage | 120 min | Report-only trend; lcov and JSON retained 90 days | `make coverage` |
| Monthly, day 1 06:37 | Runner Hygiene | 15 min | Validates runner ephemerality; scheduled failure alert | Dispatch `runner-hygiene.yml` |
Manual `workflow_dispatch` exists for the scheduled workflows above. Manual
runs are debugging evidence and intentionally do not open scheduled-failure
issues. A manual performance run may explicitly allow a known regression; that
override must not be treated as an ordinary passing baseline.
## Release validation
Release validation is post-merge and tag-driven; it does not substitute for a
pull-request gate.
| Event | Validation | Budget | Result |
|---|---|---:|---|
| Push to `main` or weekly schedule | `Build and Release` platform matrix | 150 min per platform | Build artifacts for all selected targets; no release publication on a main push |
| Valid release or preview tag | `Build and Release` plus asset checks | 150 min per platform | Draft release, checksummed assets, and publish step |
| Successful non-preview release-tag build | Docker image build and image scan | 60 min build; 30 min scan | Multi-architecture images plus vulnerability report |
| Successful release-tag build | DEB/RPM packaging | 30 min per architecture | Packages and checksum files uploaded to the release |
| Successful non-preview release-tag build | Helm template test and package | 30 min build; 30 min publish | Versioned chart and repository index |
Use an exact preview tag for end-to-end release rehearsal. Manual dispatches
are backfill/debug paths and do not prove the automatic `workflow_run` chain.
## Evidence requirements
A green check is useful only when it proves the intended behavior ran:
- Record the exact commit SHA and run URL.
- Separate product failure from runner prerequisites, service readiness, and
cancellation. Repair the precondition, then rerun the exact workload.
- Preserve membership manifests, JUnit, raw compatibility logs, seeds, and
server logs where the workflow provides them.
- For a bug fix or a new fault checker, provide sensitivity evidence: the old
behavior or an intentional mutation must fail the new oracle, and the fixed
behavior must pass it.
- Never promote a report-only lane to required from one green run. Require at
least 14 days and 30 representative pull requests with at least 99% complete
execution, then update the ruleset and this table together.
## Change checklist
Update this file in the same pull request when any of these change:
- workflow triggers, job names, timeouts, or nextest profile ownership;
- required status contexts or strict/merge-queue policy;
- scheduled cadence, alert routing, artifact contract, or local reproduction;
- report-only versus gating semantics.
Do not copy per-module test counts here. Update
[e2e-suite-inventory.md](e2e-suite-inventory.md) and its enforced membership
digest instead.
+2 -2
View File
@@ -336,13 +336,13 @@ opentelemetry = { workspace = true }
tracing-opentelemetry = { workspace = true }
# Data structures
hashbrown = { workspace = true, features = ["serde", "rayon"] }
rustfs-mimalloc = { workspace = true }
mimalloc = { workspace = true }
[target.'cfg(target_os = "linux")'.dependencies]
libsystemd.workspace = true
[target.'cfg(not(target_os = "windows"))'.dependencies]
rustfs-mimalloc-sys.workspace = true
libmimalloc-sys.workspace = true
[dev-dependencies]
uuid = { workspace = true, features = ["v4", "v5", "fast-rng", "macro-diagnostics"] }
+7 -1
View File
@@ -369,8 +369,14 @@ pub fn allocator_reclaim_controller_snapshot(ctx: &CancellationToken) -> Allocat
}
#[cfg(not(target_os = "windows"))]
#[allow(unsafe_code)]
fn collect_allocator_memory(force: bool) -> Result<(), String> {
rustfs_mimalloc::MiMalloc::collect(force);
// SAFETY: `mi_collect` is provided by the active global allocator backend
// on this target family. It is explicitly intended to reclaim retained
// pages/segments and does not require additional invariants from the caller.
unsafe {
libmimalloc_sys::mi_collect(force);
}
Ok(())
}
+8 -10
View File
@@ -26,22 +26,22 @@ struct MiMallocAllocator;
unsafe impl GlobalAlloc for MiMallocAllocator {
unsafe fn alloc(&self, layout: Layout) -> *mut u8 {
// SAFETY: the caller upholds GlobalAlloc's contract for layout.
unsafe { rustfs_mimalloc::MiMalloc.alloc(layout) }
unsafe { mimalloc::MiMalloc.alloc(layout) }
}
unsafe fn alloc_zeroed(&self, layout: Layout) -> *mut u8 {
// SAFETY: the caller upholds GlobalAlloc's contract for layout.
unsafe { rustfs_mimalloc::MiMalloc.alloc_zeroed(layout) }
unsafe { mimalloc::MiMalloc.alloc_zeroed(layout) }
}
unsafe fn dealloc(&self, ptr: *mut u8, layout: Layout) {
// SAFETY: ptr and layout came from this allocator and are forwarded unchanged.
unsafe { rustfs_mimalloc::MiMalloc.dealloc(ptr, layout) }
unsafe { mimalloc::MiMalloc.dealloc(ptr, layout) }
}
unsafe fn realloc(&self, ptr: *mut u8, layout: Layout, new_size: usize) -> *mut u8 {
// SAFETY: ptr and layout came from this allocator and are forwarded unchanged.
unsafe { rustfs_mimalloc::MiMalloc.realloc(ptr, layout, new_size) }
unsafe { mimalloc::MiMalloc.realloc(ptr, layout, new_size) }
}
}
@@ -51,7 +51,7 @@ static GLOBAL: hotpath::CountingAllocator<MiMallocAllocator> = hotpath::Counting
#[cfg(not(all(feature = "hotpath", feature = "hotpath-alloc")))]
#[global_allocator]
static GLOBAL: rustfs_mimalloc::MiMalloc = rustfs_mimalloc::MiMalloc;
static GLOBAL: mimalloc::MiMalloc = mimalloc::MiMalloc;
fn main() {
let _hotpath_guard = hotpath::HotpathGuardBuilder::new("main").build();
@@ -71,9 +71,8 @@ mod tests {
allocation.extend_from_slice(&[7_u8; 64]);
assert_eq!(allocation.len(), 64);
let heap = rustfs_mimalloc::heap::Heap::main();
// SAFETY: the live Vec pointer is valid to inspect for heap ownership.
assert!(unsafe { heap.contains(allocation.as_ptr()) });
assert!(unsafe { libmimalloc_sys::mi_is_in_heap_region(allocation.as_ptr().cast()) });
}
#[test]
@@ -86,13 +85,12 @@ mod tests {
let layout = Layout::from_size_align(32, 8).expect("valid test allocation layout");
let grown_layout = Layout::from_size_align(64, 8).expect("valid grown test allocation layout");
let allocator = super::MiMallocAllocator;
let heap = rustfs_mimalloc::heap::Heap::main();
// SAFETY: The pointer is checked for null before use and later released
// through the same allocator with the corresponding layout.
let ptr = unsafe { allocator.alloc_zeroed(layout) };
assert!(!ptr.is_null());
assert!(unsafe { heap.contains(ptr) });
assert!(unsafe { libmimalloc_sys::mi_is_in_heap_region(ptr.cast()) });
assert!(unsafe { std::slice::from_raw_parts(ptr, 32).iter().all(|byte| *byte == 0) });
// SAFETY: `ptr` was allocated by `allocator` with `layout`; on failure
@@ -104,7 +102,7 @@ mod tests {
panic!("mimalloc realloc failed in allocator smoke test");
}
assert!(unsafe { heap.contains(grown_ptr) });
assert!(unsafe { libmimalloc_sys::mi_is_in_heap_region(grown_ptr.cast()) });
// SAFETY: `grown_ptr` was reallocated by `allocator` and is released
// with the matching grown layout.
unsafe { allocator.dealloc(grown_ptr, grown_layout) };
+36 -19
View File
@@ -17,7 +17,10 @@ use rustfs_io_metrics::{
record_cpu_usage, record_memory_usage, record_process_memory_split,
};
use serde::Serialize;
#[cfg(any(test, not(target_os = "windows")))]
use serde_json::Value;
#[cfg(not(target_os = "windows"))]
use std::ffi::CStr;
use std::path::Path;
use std::sync::{Arc, Mutex, OnceLock};
use std::time::Duration;
@@ -228,18 +231,7 @@ fn read_cgroup_memory_snapshot() -> Option<CgroupMemorySnapshot> {
read_cgroup_v2().or_else(read_cgroup_v1)
}
fn read_allocator_memory_snapshot() -> Option<AllocatorMemorySnapshot> {
let json = rustfs_mimalloc::MiMalloc::stats_json();
if json.is_empty() {
return None;
}
let observation = parse_mimalloc_stats_json(&json)?;
Some(AllocatorMemorySnapshot {
backend: crate::allocator_reclaim::allocator_backend(),
observation,
})
}
#[cfg(any(test, not(target_os = "windows")))]
fn numeric_json_value(value: &Value) -> Option<u64> {
match value {
Value::Number(number) => number
@@ -250,6 +242,7 @@ fn numeric_json_value(value: &Value) -> Option<u64> {
}
}
#[cfg(any(test, not(target_os = "windows")))]
fn numeric_json_field(value: &Value, field: &str) -> Option<u64> {
match value {
Value::Object(fields) => fields
@@ -261,6 +254,7 @@ fn numeric_json_field(value: &Value, field: &str) -> Option<u64> {
}
}
#[cfg(any(test, not(target_os = "windows")))]
fn mimalloc_stat_field(value: &Value, metric: &str, field: &str) -> Option<u64> {
match value {
Value::Object(fields) => {
@@ -277,10 +271,12 @@ fn mimalloc_stat_field(value: &Value, metric: &str, field: &str) -> Option<u64>
}
}
#[cfg(any(test, not(target_os = "windows")))]
fn mimalloc_stat_current(value: &Value, metric: &str) -> Option<u64> {
mimalloc_stat_field(value, metric, "current")
}
#[cfg(any(test, not(target_os = "windows")))]
fn mimalloc_stat_sum(value: &Value, metrics: &[&str], field: &str) -> Option<u64> {
metrics
.iter()
@@ -289,6 +285,7 @@ fn mimalloc_stat_sum(value: &Value, metrics: &[&str], field: &str) -> Option<u64
.filter(|value| *value > 0)
}
#[cfg(any(test, not(target_os = "windows")))]
fn parse_mimalloc_stats_json(stats_json: &str) -> Option<AllocatorMemoryObservation> {
let value = serde_json::from_str::<Value>(stats_json).ok()?;
let malloc_metrics = ["malloc_normal", "malloc_huge"];
@@ -315,6 +312,33 @@ fn parse_mimalloc_stats_json(stats_json: &str) -> Option<AllocatorMemoryObservat
}
}
#[cfg(not(target_os = "windows"))]
#[allow(unsafe_code)]
fn read_allocator_memory_snapshot() -> Option<AllocatorMemorySnapshot> {
// SAFETY: `mi_stats_get_json` returns a null-terminated JSON buffer owned by
// mimalloc when called with a null input buffer. The mimalloc API requires
// freeing that buffer with `mi_free`; parsing finishes before the buffer is freed.
let observation = unsafe {
let stats_ptr = libmimalloc_sys::mi_stats_get_json(0, std::ptr::null_mut());
if stats_ptr.is_null() {
return None;
}
let observation = CStr::from_ptr(stats_ptr).to_str().ok().and_then(parse_mimalloc_stats_json);
libmimalloc_sys::mi_free(stats_ptr.cast());
observation?
};
Some(AllocatorMemorySnapshot {
backend: crate::allocator_reclaim::allocator_backend(),
observation,
})
}
#[cfg(target_os = "windows")]
fn read_allocator_memory_snapshot() -> Option<AllocatorMemorySnapshot> {
None
}
fn configured_memory_observability_interval_secs() -> u64 {
rustfs_utils::get_env_u64(ENV_MEMORY_OBSERVABILITY_INTERVAL_SECS, DEFAULT_MEMORY_OBSERVABILITY_INTERVAL_SECS).max(1)
}
@@ -542,13 +566,6 @@ mod tests {
assert_eq!(parse_mimalloc_stats_json(r#"{ "allocator": "unknown" }"#), None);
}
#[test]
fn read_allocator_memory_snapshot_uses_mimalloc_stats_json() {
let snapshot = super::read_allocator_memory_snapshot();
#[cfg(not(target_os = "windows"))]
assert!(snapshot.is_some(), "allocator snapshot should be available on non-Windows");
}
#[test]
fn memory_observability_snapshot_reports_disabled_when_metrics_are_disabled() {
let snapshot = build_memory_observability_status_snapshot(false, 15, false);
+4 -5
View File
@@ -554,11 +554,10 @@ fn apply_replication_timestamps_from_headers(headers: &HeaderMap<HeaderValue>, o
// Persist into the internal metadata keys so a later outbound replication
// pass (replication_target_boundary) reads the source's modification
// times instead of falling back to mod_time. Receiver-side LWW happens at
// the set layer under the object write lock
// (ecstore set_disk::ops::object::merge_replication_metadata_lww,
// rustfs/backlog#1953): a category whose stored timestamp is newer than
// the inbound one keeps the local values.
// times instead of falling back to mod_time.
// TODO(P1-6): receiver-side LWW is still missing — when the stored
// per-category timestamp is newer than the inbound one, the existing
// tags/retention/legal-hold should win instead of being overwritten.
for (timestamp, suffix) in [
(opts.replication_tagging_timestamp, SUFFIX_TAGGING_TIMESTAMP),
(opts.replication_retention_timestamp, SUFFIX_OBJECTLOCK_RETENTION_TIMESTAMP),