Compare commits

..

9 Commits

Author SHA1 Message Date
Zhengchao An 7e071dc0fa test(site-replication): cover retry endpoint matching 2026-08-16 06:17:58 +08:00
Zhengchao An e117a5d09f test(site-replication): preserve collapsed retry liability 2026-08-16 06:14:35 +08:00
Zhengchao An a26e97bbff fix(site-replication): gate test-only retry helper 2026-08-16 05:34:49 +08:00
overtrue 96a2f9c67f fix(site-replication): preserve retry accountability 2026-08-16 04:54:11 +08:00
唐小鸭 af802756c5 fix(site-replication): only a repair settles snapshot-escalated retry entries
Second review round: every iam-item / bucket-meta delivery shares a
constant path, so any later successful single-item delivery (a Bob
update) dequeued the escalated marker recording a possibly-unreplayed
deletion (a failed Alice delete) while the entity still existed
remotely. Ordinary settlement now skips escalated entries; only the
repair path — the operator's explicit accountability transfer — clears
them via dequeue_..._including_escalated. A new hook failure still
overwrites the marker and re-arms the drain. Regression covers
survive-ordinary-dequeue and repair-clears.
2026-08-16 01:02:42 +08:00
唐小鸭 971addca6e fix(site-replication): escalate snapshot-replayed retry entries instead of clearing them
Review: the bootstrap-plan snapshot cannot replay deletions — a deleted
IAM entity or absent bucket config produces no task, so clearing the
collapsed iam-item / bucket-meta entry after a successful snapshot
resend silently lost a failed delete and the peer kept stale state
permanently.

The drain now keeps those entries until remote absence is proven:
after a successful snapshot resend the entry is escalated
(failed=true, marker last_error) so it stays operator-visible in
retry_stats, and classification skips marked entries so the
once-per-failure-episode snapshot is not re-sent daily. A newer hook
failure overwrites the marker and re-arms the drain; a later full
delivery or a manual repair settles the entry. Escalation is
conditional on the snapshot timestamp, preserving the earlier
review's in-flight-failure guarantee. Bucket ops and peer edits keep
auto-settle — their replays are faithful.
2026-08-15 19:14:23 +08:00
唐小鸭 2c7d1f1f9f fix(site-replication): harden the retry drain against review findings
Adversarial review of the drain surfaced one real race and three cheap
hardenings:

- Conditional settlement for collapsed (constant-path) iam-item /
  bucket-meta entries: the snapshot resend proves delivery of the state
  as of plan-build time, so a hook failure stamped during the delivery
  window (a newer local commit the snapshot did not contain) must
  survive the snapshot's success instead of being cleared — previously
  the peer would silently diverge until the next same-path failure.
  (The operator repair path shares this collapse but keeps its existing
  unconditional settle; the drain runs every tick and needed the guard
  first.)
- Re-check the pending_* gates on the freshly loaded state: an endpoint
  refresh can commit its pending marker mid-tick without the lifecycle
  guard.
- Do not upsert retry events for peers that already left the state —
  remove_sites pruned their entries and they can never drain again.
- Correct the mutual-exclusion comment: the repair-execution lock does
  NOT close the dry-run -> execute window (dry-run takes no lock); that
  window fails safe via the preflight token, which hashes the
  replayable entries.

New tests pin the (Some settled, None failed) generation-settle
combination the peer-edit drain depends on, and the snapshot-relative
conditional settlement.
2026-08-15 11:01:58 +08:00
唐小鸭 5328e8b958 feat(site-replication): drain the retry queue from the reconcile tick
The retry queue recorded every failed peer delivery but had no
consumer — missed IAM/bucket metadata updates drifted until an operator
ran a manual repair (backlog#1675 P1-3). The 600s reconcile tick now
drains the queue behind the existing lifecycle guard and pending_*
gates.

Replay discipline (pinned by the red-light tests):
- IAM / bucket-meta entries collapse per (peer, path) and their bodies
  are not persisted, so the only faithful replay is the current
  bootstrap-plan snapshot (SiteReplicationRepairTask reuse) sent to the
  affected peer.
- make-with-versioning / configure-replication ops are re-derived from
  the CURRENT plan for their bucket — never the recorded path, whose
  query can carry an expired one-shot bootstrap token; an op whose
  bucket left the plan is provably stale and is settled.
- Peer edits are re-sent as the current peer records under a freshly
  allocated generation inside one state transaction — the recorded
  generation is stale by definition and the receiver would fence it.
- Destructive bucket ops (delete-bucket / force-delete-bucket) and
  internal: marker records (pending-endpoint-refresh backup store) are
  never background-replayed.
- Exponential backoff (600s * 2^(n-1), 24h ceiling) gates each attempt;
  an unreachable peer's transport failure re-queues its events so the
  backoff still advances.

Concurrency: the drain takes the repair-execution config lock — the
operator repair preflight token hashes the replayable retry events, so
settling them between dry-run and execute would strand the operator on
a stale preflight. Lock order matches repair (lifecycle guard -> repair
execution lock -> state object lock); success/failure settlement reuses
the generation-fenced upsert/settle semantics from #5882/#6097.

Scope note: the drain settles 'delivered once and failed' entries. A
hook that never fired (crash between local commit and send) leaves no
entry; a low-frequency plan-diff catch-all remains follow-up work, and
manual repair stays authoritative.
2026-08-15 10:42:36 +08:00
唐小鸭 1e16e06f8a test(site-replication): pin the background retry-drain discipline
Red-light evidence for backlog#1675 P1-3: the retry queue has no
background consumer — every failed peer delivery waits for a manual
repair. The new tests specify the drain rules before the drain exists:

- classification: IAM / bucket-meta entries collapse per (peer, path)
  with no persisted body, so only a bootstrap-plan snapshot resend is a
  faithful replay; make-with-versioning / configure-replication are
  re-derivable per bucket; peer edits are re-sent under a fresh
  generation; destructive bucket ops and internal: marker records
  (pending-endpoint-refresh backup store) are never background-replayed
- exponential backoff (600s * 2^(n-1), 24h ceiling) gates every
  attempt, otherwise a dead peer's entries hit the failed threshold
  within 30 minutes of reconcile ticks
- the actionable subset respects classification, current peer
  membership and backoff

All fail against the placeholder implementations (no consumer).
2026-08-15 10:19:28 +08:00
125 changed files with 3045 additions and 13557 deletions
Generated
-4
View File
@@ -278,7 +278,6 @@ checksum = "312c1ea69e5fe9966e0029fb95aca8790100b85aff4f0d3b00a9337c74069a9c"
dependencies = [
"bigdecimal",
"bon",
"crc32fast",
"digest 0.11.3",
"log",
"miniz_oxide 0.9.1",
@@ -290,11 +289,9 @@ dependencies = [
"serde",
"serde_bytes",
"serde_json",
"snap",
"strum",
"thiserror 2.0.20",
"uuid",
"zstd",
]
[[package]]
@@ -9203,7 +9200,6 @@ dependencies = [
"serial_test",
"sha2 0.11.0",
"shadow-rs",
"snap",
"socket2",
"subtle",
"sysinfo",
+1 -1
View File
@@ -171,7 +171,7 @@ tower = { version = "0.5.3" }
tower-http = { version = "0.7.0" }
# Serialization and Data Formats
apache-avro = { version = "0.22.0", features = ["snappy", "zstandard"] }
apache-avro = "0.22.0"
bytes = { version = "1.12.1" }
bytesize = "2.7.0"
byteorder = "1.5.0"
+1
View File
@@ -11,6 +11,7 @@
// 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.
#![allow(dead_code)]
use base64_simd::STANDARD;
+1 -91
View File
@@ -76,18 +76,6 @@ const SOURCE_MTIME_HEADERS: [&str; 2] = ["x-rustfs-source-mtime", "x-minio-sourc
const SOURCE_REPLICATION_REQUEST_HEADERS: [&str; 2] =
["x-rustfs-source-replication-request", "x-minio-source-replication-request"];
const SOURCE_ETAG_HEADERS: [&str; 2] = ["x-rustfs-source-etag", "x-minio-source-etag"];
const SOURCE_TAGGING_TIMESTAMP_HEADERS: [&str; 2] = [
"x-rustfs-source-replication-tagging-timestamp",
"x-minio-source-replication-tagging-timestamp",
];
const SOURCE_RETENTION_TIMESTAMP_HEADERS: [&str; 2] = [
"x-rustfs-source-replication-retention-timestamp",
"x-minio-source-replication-retention-timestamp",
];
const SOURCE_LEGALHOLD_TIMESTAMP_HEADERS: [&str; 2] = [
"x-rustfs-source-replication-legalhold-timestamp",
"x-minio-source-replication-legalhold-timestamp",
];
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"];
@@ -130,25 +118,6 @@ pub enum FaultAction {
WrongEtag,
}
/// Replication LWW timestamp headers observed on a request, journaled so
/// sender-side tests can assert what a real target would receive.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct ReplicationTimestampHeaders {
pub tagging: Option<String>,
pub retention: Option<String>,
pub legalhold: Option<String>,
}
impl ReplicationTimestampHeaders {
fn from_headers(headers: &HeaderMap) -> Self {
Self {
tagging: header_value(headers, &SOURCE_TAGGING_TIMESTAMP_HEADERS).map(bounded_journal_value),
retention: header_value(headers, &SOURCE_RETENTION_TIMESTAMP_HEADERS).map(bounded_journal_value),
legalhold: header_value(headers, &SOURCE_LEGALHOLD_TIMESTAMP_HEADERS).map(bounded_journal_value),
}
}
}
/// Credential-free request metadata retained for deterministic assertions.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RequestRecord {
@@ -162,7 +131,6 @@ pub struct RequestRecord {
pub part_number: Option<i32>,
pub content_length: Option<u64>,
pub consumed_bytes: Option<usize>,
pub replication_timestamps: ReplicationTimestampHeaders,
pub fault: Option<FaultAction>,
}
@@ -568,15 +536,7 @@ impl S3Access for FaultAccess {
.get(CONTENT_LENGTH)
.and_then(|value| value.to_str().ok())
.and_then(|value| value.parse().ok());
let replication_timestamps = ReplicationTimestampHeaders::from_headers(context.headers());
let fault = record_request(
&self.control,
operation,
context.method().clone(),
parsed,
content_length,
replication_timestamps,
);
let fault = record_request(&self.control, operation, context.method().clone(), parsed, content_length);
if let Some(RequestFault {
action: FaultAction::Status(status),
..
@@ -629,7 +589,6 @@ fn record_request(
method: Method,
parsed: ParsedRequest,
content_length: Option<u64>,
replication_timestamps: ReplicationTimestampHeaders,
) -> Option<RequestFault> {
let mut state = lock(control);
let action = parsed
@@ -654,7 +613,6 @@ fn record_request(
part_number: parsed.part_number,
content_length,
consumed_bytes: None,
replication_timestamps,
fault: action.clone(),
});
action.map(|action| RequestFault { sequence, action })
@@ -1741,52 +1699,6 @@ mod tests {
.await?)
}
#[tokio::test]
async fn journals_replication_timestamp_headers() -> Result<(), BoxError> {
let target = FakeS3Target::start().await?;
target.create_bucket("target-bucket");
let client = client(&target);
client
.put_object()
.bucket("target-bucket")
.key("plain")
.body(ByteStream::from_static(b"plain"))
.send()
.await?;
client
.put_object()
.bucket("target-bucket")
.key("stamped")
.body(ByteStream::from_static(b"stamped"))
.customize()
.map_request(move |mut request| {
let headers = request.headers_mut();
headers.insert("x-rustfs-source-replication-tagging-timestamp", "2026-01-02T03:04:05Z");
headers.insert("x-minio-source-replication-retention-timestamp", "2026-01-02T03:04:06Z");
headers.insert("x-rustfs-source-replication-legalhold-timestamp", "2026-01-02T03:04:07Z");
Ok::<_, std::convert::Infallible>(request)
})
.send()
.await?;
let requests = target.requests();
let plain = requests
.iter()
.find(|record| record.operation == Operation::PutObject && record.key.as_deref() == Some("plain"))
.expect("plain PUT must be journaled");
assert_eq!(plain.replication_timestamps, ReplicationTimestampHeaders::default());
let stamped = requests
.iter()
.find(|record| record.operation == Operation::PutObject && record.key.as_deref() == Some("stamped"))
.expect("stamped PUT must be journaled");
assert_eq!(stamped.replication_timestamps.tagging.as_deref(), Some("2026-01-02T03:04:05Z"));
assert_eq!(stamped.replication_timestamps.retention.as_deref(), Some("2026-01-02T03:04:06Z"));
assert_eq!(stamped.replication_timestamps.legalhold.as_deref(), Some("2026-01-02T03:04:07Z"));
Ok(())
}
macro_rules! assert_sdk_error {
($error:expr, $status:expr, $code:expr) => {{
let error = &$error;
@@ -3073,7 +2985,6 @@ mod tests {
part_number: None,
},
Some(0),
ReplicationTimestampHeaders::default(),
);
}
let records = lock(&control).requests.clone();
@@ -3095,7 +3006,6 @@ mod tests {
part_number: None,
},
None,
ReplicationTimestampHeaders::default(),
);
{
let bounded_records = lock(&bounded_control);
+12 -14
View File
@@ -135,8 +135,7 @@ pub mod bucket {
pub use crate::bucket::metadata_sys::ConfigWriteLockProbe;
pub use crate::bucket::metadata_sys::{
BucketMetadataMutationGuard, BucketMetadataSys, ObjectLockConfigState, acquire_bucket_metadata_transaction_lock,
acquire_bucket_metadata_transaction_lock_for_incarnation, capture_bucket_metadata_incarnation, delete,
delete_if_incarnation, delete_under_transaction_lock, get, get_accelerate_config, get_bucket_policy,
capture_bucket_metadata_incarnation, delete, delete_if_incarnation, get, get_accelerate_config, get_bucket_policy,
get_bucket_policy_raw, get_bucket_targets_config, get_config_from_disk, get_cors_config, get_durability_config,
get_global_bucket_metadata_sys, get_lifecycle_config, get_logging_config, get_notification_config,
get_object_lock_config, get_object_lock_config_state, get_public_access_block_config, get_quota_config,
@@ -185,18 +184,17 @@ pub mod bucket {
mrf_backlog_observability_snapshot,
};
pub use crate::bucket::replication::{
BucketReplicationResyncStatus, BucketReplicationStat, BucketReplicationStats, BucketStats,
DeleteReplicationConfigSnapshot, DeletedObjectReplicationInfo, DurableMrfBacklog, DynReplicationPool, InQueueMetric,
MrfOpKind, MrfReplicateEntry, MustReplicateOptions, ObjectOpts, REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION,
REMOTE_TARGET_UNSUPPORTED_FIELDS, REMOTE_TARGET_WRITABLE_FIELDS, REPLICATE_INCOMING_DELETE,
REPLICATION_CAPABILITY_CONTRACT_VERSION, REPLICATION_READ_ONLY_HISTORICAL_FIELDS, REPLICATION_WRITABLE_FIELDS,
ReplicateDecision, ReplicateObjectInfo, ReplicationBatchAdmission, ReplicationConfig,
ReplicationConfigStructureError, ReplicationConfigurationExt, ReplicationDeleteScheduleInput,
ReplicationDeleteStateSource, ReplicationHealQueueResult, ReplicationObjectBridge, ReplicationObjectIO,
ReplicationOperation, ReplicationPoolTrait, ReplicationPriority, ReplicationQueueAdmission, ReplicationScannerBridge,
ReplicationState, ReplicationStats, ReplicationStatusType, ReplicationStorage, ReplicationTargetValidationError,
ReplicationType, ResyncOpts, ResyncStatusType, RuntimeReplicationTargetBacklog, TargetReplicationResyncStatus,
VersionPurgeStatusType, XferStats, commit_force_delete_intent, complete_force_delete_intent,
BucketReplicationResyncStatus, BucketReplicationStats, BucketStats, DeleteReplicationConfigSnapshot,
DeletedObjectReplicationInfo, DurableMrfBacklog, DynReplicationPool, MrfOpKind, MrfReplicateEntry,
MustReplicateOptions, ObjectOpts, REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION, REMOTE_TARGET_UNSUPPORTED_FIELDS,
REMOTE_TARGET_WRITABLE_FIELDS, REPLICATE_INCOMING_DELETE, REPLICATION_CAPABILITY_CONTRACT_VERSION,
REPLICATION_READ_ONLY_HISTORICAL_FIELDS, REPLICATION_WRITABLE_FIELDS, ReplicateDecision, ReplicateObjectInfo,
ReplicationBatchAdmission, ReplicationConfig, ReplicationConfigStructureError, ReplicationConfigurationExt,
ReplicationDeleteScheduleInput, ReplicationDeleteStateSource, ReplicationHealQueueResult, ReplicationObjectBridge,
ReplicationObjectIO, ReplicationOperation, ReplicationPoolTrait, ReplicationPriority, ReplicationQueueAdmission,
ReplicationScannerBridge, ReplicationState, ReplicationStats, ReplicationStatusType, ReplicationStorage,
ReplicationTargetValidationError, ReplicationType, ResyncOpts, ResyncStatusType, RuntimeReplicationTargetBacklog,
TargetReplicationResyncStatus, VersionPurgeStatusType, 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, init_background_replication, invalid_replication_config_status_field,
persist_force_delete_intent, read_durable_mrf_backlog, replication_state_to_filemeta, replication_status_to_filemeta,
+5 -70
View File
@@ -58,9 +58,7 @@ use rustfs_utils::http::{
};
use rustfs_utils::http::{
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,
SUFFIX_SOURCE_REPLICATION_REQUEST, SUFFIX_SOURCE_VERSION_ID, insert_header,
};
use rustls_pki_types::pem::PemObject;
use serde::{Deserialize, Serialize};
@@ -82,6 +80,7 @@ use tracing::warn;
use url::Url;
use uuid::Uuid;
const DEFAULT_HEALTH_CHECK_RELOAD_DURATION: Duration = Duration::from_secs(30 * 60);
const MAX_CONCURRENT_TARGET_HEALTH_CHECKS: usize = 16;
const REDACTED_CREDENTIAL: &str = "<redacted>";
@@ -1477,12 +1476,9 @@ impl Default for AdvancedPutOptions {
replication_status: ReplicationStatusType::Pending,
source_mtime: OffsetDateTime::now_utc(),
replication_request: false,
// UNIX_EPOCH means "never modified": header() must not emit a
// timestamp header for it, otherwise a receiver would treat an
// unset category as a modification made right now.
retention_timestamp: OffsetDateTime::UNIX_EPOCH,
tagging_timestamp: OffsetDateTime::UNIX_EPOCH,
legalhold_timestamp: OffsetDateTime::UNIX_EPOCH,
retention_timestamp: OffsetDateTime::now_utc(),
tagging_timestamp: OffsetDateTime::now_utc(),
legalhold_timestamp: OffsetDateTime::now_utc(),
replication_validity_check: false,
}
}
@@ -1679,16 +1675,6 @@ impl PutObjectOptions {
);
}
for (suffix, timestamp) in [
(SUFFIX_SOURCE_REPLICATION_TAGGING_TIMESTAMP, self.internal.tagging_timestamp),
(SUFFIX_SOURCE_REPLICATION_RETENTION_TIMESTAMP, self.internal.retention_timestamp),
(SUFFIX_SOURCE_REPLICATION_LEGALHOLD_TIMESTAMP, self.internal.legalhold_timestamp),
] {
if timestamp.unix_timestamp() != 0 {
insert_header(&mut header, suffix, timestamp.format(&Rfc3339).unwrap_or_default());
}
}
if self.internal.replication_request {
insert_header(&mut header, SUFFIX_SOURCE_REPLICATION_REQUEST, "true");
}
@@ -2856,57 +2842,6 @@ mod tests {
);
}
#[test]
fn put_object_headers_carry_replication_timestamp_headers() {
// MinIO receivers resolve concurrent tag/retention/legal-hold edits by
// last-writer-wins on these headers (object-api-options.go parses them
// as RFC3339); a replica without them loses every conflict resolution.
let mut opts = PutObjectOptions::default();
opts.internal.replication_request = true;
let tagging = OffsetDateTime::from_unix_timestamp(1_700_000_001).expect("valid timestamp");
let retention = OffsetDateTime::from_unix_timestamp(1_700_000_002).expect("valid timestamp");
let legalhold = OffsetDateTime::from_unix_timestamp(1_700_000_003).expect("valid timestamp");
opts.internal.tagging_timestamp = tagging;
opts.internal.retention_timestamp = retention;
opts.internal.legalhold_timestamp = legalhold;
let header = opts.header();
for (suffix, expected) in [
("source-replication-tagging-timestamp", tagging),
("source-replication-retention-timestamp", retention),
("source-replication-legalhold-timestamp", legalhold),
] {
assert_eq!(
rustfs_utils::http::get_header(&header, suffix).as_deref(),
Some(expected.format(&Rfc3339).expect("RFC3339 timestamp").as_str()),
"replication put requests must carry the {suffix} header"
);
}
}
#[test]
fn put_object_headers_omit_unset_replication_timestamps() {
// UNIX_EPOCH means "never modified on the source"; sending it would
// make the receiver treat an unset category as a fresh modification.
let mut opts = PutObjectOptions::default();
opts.internal.replication_request = true;
opts.internal.tagging_timestamp = OffsetDateTime::UNIX_EPOCH;
opts.internal.retention_timestamp = OffsetDateTime::UNIX_EPOCH;
opts.internal.legalhold_timestamp = OffsetDateTime::UNIX_EPOCH;
let header = opts.header();
for suffix in [
"source-replication-tagging-timestamp",
"source-replication-retention-timestamp",
"source-replication-legalhold-timestamp",
] {
assert!(
rustfs_utils::http::get_header(&header, suffix).is_none(),
"unset {suffix} must not be sent to replication targets"
);
}
}
#[tokio::test]
async fn get_remote_target_client_internal_rejects_loopback_endpoint() {
let sys = BucketTargetSys::default();
@@ -126,23 +126,11 @@ const EVENT_LIFECYCLE_EXPIRED_DETECTED: &str = "lifecycle_expired_detected";
const EVENT_LIFECYCLE_NOT_ENQUEUED: &str = "lifecycle_not_enqueued";
const EVENT_LIFECYCLE_DELETE_DISPATCHED: &str = "lifecycle_delete_dispatched";
const EVENT_LIFECYCLE_DELETE_COMPLETED: &str = "lifecycle_delete_completed";
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
const EVENT_LIFECYCLE_TIER_AUDIT: &str = "lifecycle_tier_audit";
const EVENT_LIFECYCLE_TIER_OPERATION_FAILED: &str = "lifecycle_tier_operation_failed";
const EVENT_LIFECYCLE_DELETE_FAILED: &str = "lifecycle_delete_failed";
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
pub type TimeFn = Arc<dyn Fn() -> Pin<Box<dyn Future<Output = ()> + Send>> + Send + Sync + 'static>;
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
pub type TraceFn =
Arc<dyn Fn(String, HashMap<String, String>) -> Pin<Box<dyn Future<Output = ()> + Send>> + Send + Sync + 'static>;
pub type ExpiryOpType = Box<dyn ExpiryOp + Send + Sync + 'static>;
@@ -152,21 +140,9 @@ static TIER_FREE_VERSION_RECOVERY_STARTED: OnceLock<()> = OnceLock::new();
static MANUAL_TRANSITION_JOB_RECOVERY_STARTED: OnceLock<()> = OnceLock::new();
pub const AMZ_OBJECT_TAGGING: &str = "X-Amz-Tagging";
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
pub const AMZ_TAG_COUNT: &str = "x-amz-tagging-count";
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
pub const AMZ_TAG_DIRECTIVE: &str = "X-Amz-Tagging-Directive";
pub const AMZ_ENCRYPTION_AES: &str = "AES256";
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
pub const AMZ_ENCRYPTION_KMS: &str = "aws:kms";
pub const ERR_INVALID_STORAGECLASS: &str = "invalid tier.";
@@ -304,10 +280,6 @@ impl LifecycleSys {
}
}
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
pub fn trace(oi: &ObjectInfo) -> TraceFn {
let bucket = oi.bucket.clone();
let name = oi.name.clone();
@@ -598,10 +570,6 @@ async fn delete_free_version_remote_object(
Ok(())
}
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
async fn delete_free_version_remote_object_then<T, F, Fut>(
oi: &ObjectInfo,
tier_config_mgr: &Arc<RwLock<TierConfigMgr>>,
@@ -2900,10 +2868,6 @@ fn stale_upload_default_due(initiated: OffsetDateTime, default_expiry: StdDurati
initiated + time::Duration::seconds(default_expiry.as_secs() as i64)
}
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
async fn stale_upload_current_size(set: &Arc<SetDisks>, metadata: &HashMap<String, String>, upload_dir: &str) -> Option<usize> {
stale_upload_current_size_with_opts(set, metadata, upload_dir, false).await
}
@@ -3388,10 +3352,6 @@ pub async fn validate_transition_tier(lc: &BucketLifecycleConfiguration) -> Resu
Ok(())
}
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
fn mark_delete_opts_skip_decommissioned_on_remote_success(opts: &mut ObjectOptions, remote_delete_succeeded: bool) {
if remote_delete_succeeded {
opts.skip_decommissioned = true;
@@ -4379,10 +4339,6 @@ pub async fn expire_transitioned_object(
Ok(dobj)
}
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
pub fn gen_transition_objname(bucket: &str) -> Result<String, Error> {
let us = Uuid::new_v4().to_string();
let mut hasher = Sha256::new();
@@ -4417,10 +4373,6 @@ pub async fn transition_object(api: Arc<ECStore>, oi: &ObjectInfo, lae: LcAuditE
result
}
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
pub fn audit_tier_actions(_tier: &str, bytes: i64) -> TimeFn {
let tier = _tier.to_string();
Arc::new(move || {
@@ -4439,10 +4391,6 @@ pub fn audit_tier_actions(_tier: &str, bytes: i64) -> TimeFn {
})
}
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
pub async fn get_transitioned_object_reader(
bucket: &str,
object: &str,
@@ -5197,10 +5145,6 @@ async fn lifecycle_delete_config_snapshot(api: &ECStore, oi: &ObjectInfo) -> Res
ReplicationObjectBridge::delete_request_config(api, &oi.bucket).await
}
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
pub async fn apply_lifecycle_action(event: &lifecycle::Event, src: &LcEventSrc, oi: &ObjectInfo) -> bool {
let mut success = false;
match event.action {
@@ -7478,10 +7422,6 @@ mod tests {
// process environment while `env::set_var`/`env::remove_var` is active.
// SAFETY: keep this note adjacent to the allowance for the repository guard.
#[allow(unsafe_code)]
#[allow(
dead_code,
reason = "transition-queue env fixture kept for tests that scope those vars; no test uses it today (backlog#1823)"
)]
async fn with_transition_queue_env_async<F, Fut>(capacity: Option<&str>, timeout_ms: Option<&str>, test_fn: F)
where
F: FnOnce() -> Fut,
@@ -759,10 +759,6 @@ pub struct ManualTransitionWorkerResultRecord {
}
impl ManualTransitionWorkerResultRecord {
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
pub fn new(job_id: Uuid, task_key: impl Into<String>, result: ManualTransitionWorkerResult) -> Self {
Self::new_with_reason(job_id, task_key, result, None)
}
@@ -1261,10 +1257,6 @@ pub(crate) async fn save_manual_transition_task_if_absent(
}
}
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
pub async fn load_manual_transition_task_record(
api: Arc<ECStore>,
job_id: Uuid,
@@ -1328,10 +1320,6 @@ async fn scan_manual_transition_task_journal(api: Arc<ECStore>, job_id: Uuid) ->
}
}
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
pub async fn load_manual_transition_worker_result_stats(
api: Arc<ECStore>,
job_id: Uuid,
@@ -1467,10 +1455,6 @@ async fn scan_manual_transition_worker_result_journal(
}
}
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
pub async fn reconcile_manual_transition_worker_results(
api: Arc<ECStore>,
job_id: Uuid,
@@ -15,35 +15,25 @@
use rustfs_common::metrics::IlmAction;
use crate::bucket::lifecycle::lifecycle::ObjectOpts;
use crate::bucket::replication::ReplicationLifecycleBridge;
pub(crate) use crate::bucket::replication::ReplicationStatusType;
#[cfg(test)]
pub(crate) use crate::bucket::replication::VersionPurgeStatusType;
pub(crate) use crate::bucket::replication::{
DeleteReplicationConfigSnapshot, ReplicationObjectBridge, replication_state_to_filemeta,
};
use crate::bucket::replication::{ReplicationLifecycleBridge, ReplicationLifecycleConfig};
use crate::storage_api_contracts::object::DeletedObject;
#[allow(
dead_code,
reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
)]
pub(crate) type LifecycleReplicationConfig = ReplicationLifecycleConfig;
pub(crate) fn has_pending_version_purge(obj: &ObjectOpts) -> bool {
obj.version_purge_status.is_pending()
}
#[allow(
dead_code,
reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
)]
pub(crate) fn has_pending_object_replication(obj: &ObjectOpts) -> bool {
replication_status_blocks_lifecycle(&obj.replication_status)
}
#[allow(
dead_code,
reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
)]
pub(crate) fn has_pending_lifecycle_replication(obj: &ObjectOpts) -> bool {
has_pending_object_replication(obj) || has_pending_version_purge(obj)
}
@@ -14,10 +14,6 @@
use std::collections::HashMap;
#[allow(
dead_code,
reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
)]
pub(crate) fn decode_tags_to_map(tags: &str) -> HashMap<String, String> {
crate::bucket::tagging::decode_tags_to_map(tags)
}
@@ -331,10 +331,6 @@ where
persist_tier_delete_journal_entry(api, &committed).await
}
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
pub async fn abort_tier_delete_journal_entry<S>(api: Arc<S>, je: &Jentry) -> std::io::Result<()>
where
S: ObjectOperations<
@@ -148,10 +148,6 @@ struct RecoveryCursor {
object: String,
}
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
pub async fn recover_tier_free_versions(
api: Arc<ECStore>,
limit: usize,
@@ -385,10 +385,6 @@ impl ExpiryOp for Jentry {
}
}
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
pub async fn delete_object_from_remote_tier(obj_name: &str, rv_id: &str, tier_name: &str) -> Result<(), std::io::Error> {
let result = delete_object_from_remote_tier_raw(obj_name, rv_id, tier_name).await;
if let Err(err) = &result
@@ -399,10 +395,6 @@ pub async fn delete_object_from_remote_tier(obj_name: &str, rv_id: &str, tier_na
result
}
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
async fn delete_object_from_remote_tier_raw(obj_name: &str, rv_id: &str, tier_name: &str) -> Result<(), std::io::Error> {
#[cfg(test)]
if let Some(result) = run_remote_tier_delete_test_hook(obj_name, rv_id, tier_name) {
@@ -413,10 +405,6 @@ async fn delete_object_from_remote_tier_raw(obj_name: &str, rv_id: &str, tier_na
delete_object_from_remote_tier_raw_with_manager(obj_name, rv_id, tier_name, &tier_config_mgr).await
}
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
async fn delete_object_from_remote_tier_raw_with_manager(
obj_name: &str,
rv_id: &str,
@@ -497,10 +485,6 @@ pub enum RemoteTierDeleteOutcome {
AlreadyRemoved,
}
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
pub async fn delete_object_from_remote_tier_idempotent(
obj_name: &str,
rv_id: &str,
@@ -50,16 +50,8 @@ pub type Result<T> = std::result::Result<T, TransitionTransactionError>;
#[derive(Debug, thiserror::Error)]
pub enum TransitionTransactionError {
#[error("transition transaction already exists")]
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
AlreadyExists,
#[error("transition transaction is not found")]
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
NotFound,
#[error("transition transaction is corrupt: {0}")]
Corrupt(&'static str),
-31
View File
@@ -60,14 +60,12 @@ struct ConfigWriteLockProbeState {
static CONFIG_WRITE_LOCK_PROBES: std::sync::OnceLock<StdMutex<Vec<Arc<ConfigWriteLockProbeState>>>> = std::sync::OnceLock::new();
#[cfg(any(test, feature = "test-util"))]
#[allow(dead_code, reason = "installed by tests behind `--features test-util` (backlog#1823)")]
pub struct ConfigWriteLockProbe {
state: Arc<ConfigWriteLockProbeState>,
}
#[cfg(any(test, feature = "test-util"))]
impl ConfigWriteLockProbe {
#[allow(dead_code, reason = "installed by tests behind `--features test-util` (backlog#1823)")]
pub fn install(bucket: &str) -> Self {
let state = Arc::new(ConfigWriteLockProbeState {
bucket: bucket.to_string(),
@@ -86,7 +84,6 @@ impl ConfigWriteLockProbe {
Self { state }
}
#[allow(dead_code, reason = "installed by tests behind `--features test-util` (backlog#1823)")]
pub async fn wait_until_attempted(&self) {
tokio::time::timeout(Duration::from_secs(30), self.state.arrived.notified())
.await
@@ -659,16 +656,6 @@ pub async fn update_under_transaction_lock(
update_under_config_write_guard(get_bucket_metadata_sys()?, guard, config_file, data).await
}
/// Clear one config file while the caller holds this bucket's transaction lock.
pub async fn delete_under_transaction_lock(
guard: &BucketMetadataMutationGuard,
bucket: &str,
config_file: &str,
) -> Result<OffsetDateTime> {
guard.ensure_valid(bucket)?;
delete_under_config_write_guard(get_bucket_metadata_sys()?, guard, config_file).await
}
pub async fn update_quota_if_incarnation(
bucket: &str,
data: Vec<u8>,
@@ -808,14 +795,6 @@ pub async fn acquire_bucket_metadata_transaction_lock(bucket: &str) -> Result<Bu
acquire_config_write_guard(get_bucket_metadata_sys()?, bucket).await
}
/// Acquire the bucket transaction lock only if its incarnation still matches.
pub async fn acquire_bucket_metadata_transaction_lock_for_incarnation(
bucket: &str,
expected_incarnation_id: Uuid,
) -> Result<BucketMetadataMutationGuard> {
acquire_config_write_guard_for_incarnation(get_bucket_metadata_sys()?, bucket, Some(expected_incarnation_id)).await
}
pub(crate) async fn acquire_bucket_metadata_transaction_lock_in(
ctx: &crate::runtime::instance::InstanceContext,
bucket: &str,
@@ -893,10 +872,6 @@ pub async fn get_bucket_policy_raw(bucket: &str) -> Result<(String, OffsetDateTi
bucket_meta_sys.get_bucket_policy_raw(bucket).await
}
#[allow(
dead_code,
reason = "free-function facade over the live BucketMetadataSys::get_bucket_acl_config; no caller in this port (backlog#1823)"
)]
pub async fn get_bucket_acl_config(bucket: &str) -> Result<(String, OffsetDateTime)> {
let bucket_meta_sys_lock = get_bucket_metadata_sys()?;
let bucket_meta_sys = bucket_meta_sys_lock.read().await;
@@ -1111,10 +1086,6 @@ pub async fn get_config_from_disk(bucket: &str) -> Result<BucketMetadata> {
bucket_meta_sys.get_config_from_disk(bucket).await
}
#[allow(
dead_code,
reason = "ambient-facade variant of the live created_at_in; no caller in this port (backlog#1823)"
)]
pub async fn created_at(bucket: &str) -> Result<OffsetDateTime> {
let bucket_meta_sys_lock = get_bucket_metadata_sys()?;
let bucket_meta_sys = bucket_meta_sys_lock.read().await;
@@ -1628,7 +1599,6 @@ impl BucketMetadataSys {
/// [`Self::update`], with the payload computed from the loaded metadata
/// instead of supplied up front. Loads through this system's own store so
/// the read and the persisted write target the same instance.
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
async fn update_config_with<F>(&self, bucket: &str, config_file: &str, mutate: F) -> Result<OffsetDateTime>
where
F: FnOnce(&BucketMetadata) -> Result<Vec<u8>> + Send,
@@ -1733,7 +1703,6 @@ impl BucketMetadataSys {
/// A miss is never published as an authoritative default, and a snapshot
/// read before delete plus same-name recreation cannot replace the new
/// generation.
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
pub(crate) async fn reload_from_store(&self, bucket: &str) -> Result<()> {
if is_meta_bucketname(bucket) {
return Err(Error::other("errInvalidArgument"));
+1
View File
@@ -13,6 +13,7 @@
// limitations under the License.
// #730: bucket subsystems still contain staged ECStore migration code.
#![allow(dead_code)]
pub mod bandwidth;
pub mod bucket_target_sys;
@@ -136,7 +136,6 @@ pub fn add_years(dt: OffsetDateTime, years: i32) -> OffsetDateTime {
/// Check if an object has legal hold enabled.
/// Returns true if legal hold is ON.
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
fn has_legal_hold(user_defined: &std::collections::HashMap<String, String>) -> bool {
let lhold = objectlock::get_object_legalhold_meta(user_defined);
matches!(lhold.status, Some(ref st) if st.as_str() == ObjectLockLegalHoldStatus::ON)
@@ -152,7 +151,6 @@ fn has_legal_hold(user_defined: &std::collections::HashMap<String, String>) -> b
/// # Returns
/// * `true` if the object is locked (cannot be deleted/modified)
/// * `false` if the object is not locked
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
pub fn is_object_locked_by_metadata(user_defined: &std::collections::HashMap<String, String>, is_delete_marker: bool) -> bool {
// Delete markers are never locked
if is_delete_marker {
-2
View File
@@ -193,7 +193,6 @@ pub enum QuotaError {
}
#[derive(Debug, Serialize)]
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
pub struct QuotaErrorResponse {
#[serde(rename = "Code")]
pub code: String,
@@ -209,7 +208,6 @@ pub struct QuotaErrorResponse {
}
impl QuotaErrorResponse {
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
pub fn new(quota_error: &QuotaError, request_id: &str, host_id: &str) -> Self {
match quota_error {
QuotaError::QuotaExceeded { .. } => Self {
@@ -899,7 +899,6 @@ async fn save_ledger_locked(
}
#[cfg(any(test, feature = "test-util"))]
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
pub fn fail_next_quota_ledger_save_for_test() {
FAIL_NEXT_LEDGER_SAVE.store(true, std::sync::atomic::Ordering::SeqCst);
}
+2 -2
View File
@@ -60,7 +60,7 @@ pub use replication_filemeta_boundary::{
pub(crate) use replication_filemeta_boundary::{
replication_state_from_filemeta, replication_status_from_filemeta, version_purge_status_from_filemeta,
};
pub(crate) use replication_lifecycle_bridge::ReplicationLifecycleBridge;
pub(crate) use replication_lifecycle_bridge::{ReplicationLifecycleBridge, ReplicationLifecycleConfig};
pub(crate) use replication_migration_bridge::ReplicationMigrationBridge;
pub use replication_object_bridge::ReplicationObjectBridge;
pub use replication_object_config::{DeleteReplicationConfigSnapshot, ReplicationConfig};
@@ -81,6 +81,6 @@ pub use replication_queue_boundary::{
pub use replication_resync_boundary::{BucketReplicationResyncStatus, ResyncOpts, TargetReplicationResyncStatus};
pub use replication_scanner_bridge::ReplicationScannerBridge;
pub use replication_state::{ReplicationStats, RuntimeReplicationTargetBacklog};
pub use replication_stats_boundary::{BucketReplicationStat, BucketReplicationStats, BucketStats, InQueueMetric, XferStats};
pub use replication_stats_boundary::{BucketReplicationStats, BucketStats};
pub use replication_storage_boundary::{ReplicationObjectIO, ReplicationStorage};
pub(crate) use replication_target_config_bridge::ReplicationTargetConfigBridge;
@@ -37,10 +37,6 @@ impl ReplicationConfigStore {
com::read_config_limited(api, file, max_bytes).await
}
#[allow(
dead_code,
reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
)]
pub(crate) async fn read_no_lock<S>(api: Arc<S>, file: &str) -> Result<Vec<u8>>
where
S: ReplicationObjectIO,
@@ -24,27 +24,15 @@ use super::replication_storage_boundary::{
DeletedObject, ObjectInfo, ObjectOptions, ObjectToDelete, deleted_object_for_replication,
};
#[allow(
dead_code,
reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
)]
pub(crate) type ReplicationLifecycleConfig = ReplicationConfig;
pub(crate) struct ReplicationLifecycleBridge;
impl ReplicationLifecycleBridge {
#[allow(
dead_code,
reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
)]
pub(crate) fn new_config(config: ReplicationConfiguration) -> ReplicationLifecycleConfig {
ReplicationConfig::new(Some(config), None)
}
#[allow(
dead_code,
reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
)]
pub(crate) fn has_pending_version_purge(
config: &ReplicationLifecycleConfig,
object_name: &str,
@@ -57,10 +45,6 @@ impl ReplicationLifecycleBridge {
.is_some_and(|config| config.has_active_rules(object_name, true))
}
#[allow(
dead_code,
reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
)]
pub(crate) async fn check_delete_replication(
bucket: &str,
object: &ObjectToDelete,
@@ -70,10 +54,6 @@ impl ReplicationLifecycleBridge {
check_replicate_delete(bucket, object, source, opts, None).await
}
#[allow(
dead_code,
reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
)]
pub(crate) fn version_delete_replication_state(decision: &ReplicateDecision) -> ReplicationState {
let pending_status = decision.pending_status();
ReplicationState {
@@ -19,33 +19,17 @@ use time::OffsetDateTime;
use super::replication_error_boundary::Result;
use crate::bucket::msgp_decode;
#[allow(
dead_code,
reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
)]
pub(crate) struct ReplicationMsgpCodec;
impl ReplicationMsgpCodec {
#[allow(
dead_code,
reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
)]
pub(crate) fn read_ext8_time<R: Read>(rd: &mut R) -> Result<OffsetDateTime> {
msgp_decode::read_msgp_ext8_time(rd)
}
#[allow(
dead_code,
reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
)]
pub(crate) fn skip_value<R: Read>(rd: &mut R) -> Result<()> {
msgp_decode::skip_msgp_value(rd)
}
#[allow(
dead_code,
reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
)]
pub(crate) fn write_time<W: Write>(wr: &mut W, time: OffsetDateTime) -> Result<()> {
msgp_decode::write_msgp_time(wr, time)
}
@@ -77,10 +77,6 @@ impl ReplicationObjectBridge {
load_delete_request_config_in(ctx, bucket).await
}
#[allow(
dead_code,
reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
)]
pub(crate) async fn delete_config_snapshot_in(
ctx: &ReplicationInstanceContext,
bucket: &str,
@@ -231,10 +231,6 @@ pub(crate) async fn load_delete_replication_config(
delete_snapshot_from_metadata(ReplicationMetadataStore::delete_metadata(bucket).await?)
}
#[allow(
dead_code,
reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
)]
pub(crate) async fn load_delete_replication_config_in(
ctx: &ReplicationInstanceContext,
bucket: &str,
@@ -217,10 +217,6 @@ impl DurableMrfBacklogTracker {
}
}
#[allow(
dead_code,
reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
)]
fn durable_mrf_backlog_tracker_from_entries(entries: &[MrfReplicateEntry]) -> DurableMrfBacklogTracker {
let mut tracker = DurableMrfBacklogTracker {
available: true,
@@ -716,10 +712,6 @@ pub struct ReplicationPool<S: ReplicationStorage> {
// MRF worker lifecycle
mrf_worker_cancellations: Mutex<Vec<CancellationToken>>,
#[allow(
dead_code,
reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
)]
mrf_stop_tx: Sender<()>,
// Worker size tracking
@@ -948,10 +940,6 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
}
/// Resizes worker priority and counts
#[allow(
dead_code,
reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
)]
pub async fn resize_worker_priority(
&self,
pri: ReplicationPriority,
@@ -1192,10 +1180,6 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
}
/// Queues an MRF save operation
#[allow(
dead_code,
reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
)]
async fn queue_mrf_save(&self, entry: MrfReplicateEntry) {
let _ = self.queue_mrf_save_admission(entry, "mrf_worker").await;
}
@@ -1667,10 +1651,6 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
}
/// Worker function for handling regular replication operations
#[allow(
dead_code,
reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
)]
async fn add_worker(
&self,
mut rx: Receiver<ReplicationOperation>,
@@ -1684,10 +1664,6 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
}
/// Worker function for handling large object replication operations
#[allow(
dead_code,
reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
)]
async fn add_large_worker(
&self,
mut rx: Receiver<ReplicationOperation>,
@@ -1702,10 +1678,6 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
}
/// Worker function for handling MRF (Most Recent Failures) operations
#[allow(
dead_code,
reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
)]
async fn add_mrf_worker(
&self,
mut rx: Receiver<ReplicationOperation>,
@@ -1719,10 +1691,6 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
}
/// Delete resync metadata from replication resync state in memory
#[allow(
dead_code,
reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
)]
pub async fn delete_resync_metadata(&self, bucket: &str) {
let mut status_map = self.resyncer.status_map.write().await;
status_map.remove(bucket);
@@ -21,31 +21,11 @@ pub(crate) use rustfs_replication::{
should_count_head_proxy_failure,
};
#[allow(
dead_code,
reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
)]
pub(crate) const RESYNC_META_FORMAT: u16 = rustfs_replication::resync::RESYNC_META_FORMAT;
#[allow(
dead_code,
reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
)]
pub(crate) const RESYNC_META_VERSION: u16 = rustfs_replication::resync::RESYNC_META_VERSION;
pub(crate) const RESYNC_FILE_MAX_BYTES: usize = rustfs_replication::RESYNC_FILE_MAX_BYTES;
#[allow(
dead_code,
reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
)]
pub(crate) const WIRE_ZERO_TIME_UNIX: i64 = rustfs_replication::resync::WIRE_ZERO_TIME_UNIX;
#[allow(
dead_code,
reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
)]
pub(crate) const MRF_META_FORMAT: u16 = rustfs_replication::mrf::MRF_META_FORMAT;
#[allow(
dead_code,
reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
)]
pub(crate) const MRF_META_VERSION: u16 = rustfs_replication::mrf::MRF_META_VERSION;
fn map_replication_error(err: rustfs_replication::Error) -> Error {
@@ -122,10 +122,6 @@ const REPLICATION_TARGET_OFFLINE_ERROR_MARKERS: &[&str] = &[
"tcp connect error",
];
#[allow(
dead_code,
reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
)]
const RESYNC_TIME_INTERVAL: TokioDuration = TokioDuration::from_secs(60);
static WARNED_MONITOR_UNINIT: std::sync::Once = std::sync::Once::new();
@@ -332,10 +328,6 @@ fn bounded_resync_max_jobs(value: usize) -> usize {
#[derive(Debug)]
pub struct ReplicationResyncer {
pub status_map: Arc<RwLock<HashMap<String, BucketReplicationResyncStatus>>>,
#[allow(
dead_code,
reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
)]
pub worker_size: usize,
pub(crate) cancel_tokens: Arc<RwLock<HashMap<ResyncCancelKey, CancellationToken>>>,
resync_admission: Arc<Semaphore>,
@@ -552,10 +544,6 @@ impl ReplicationResyncer {
.is_some_and(|status| status.failed_count > 0)
}
#[allow(
dead_code,
reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
)]
pub async fn persist_to_disk<S>(&self, cancel_token: CancellationToken, api: Arc<S>)
where
S: ReplicationObjectIO,
@@ -340,10 +340,6 @@ impl ReplicationStats {
}
/// Site replication update replica statistics
#[allow(
dead_code,
reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
)]
fn sr_update_replica_stat(&self, size: i64) {
self.sr_stats.replica_size.fetch_add(size, Ordering::Relaxed);
self.sr_stats.replica_count.fetch_add(1, Ordering::Relaxed);
@@ -708,12 +704,6 @@ impl ReplicationStats {
} else {
BucketReplicationStats::new()
};
// Stamp the serializable failure windows from the live samples: the
// samples themselves do not cross the peer-RPC wire, so this snapshot
// is what cluster aggregation and the metrics endpoints see.
for stat in replication_stats.stats.values_mut() {
stat.fail_stats.refresh_windows();
}
let uptime = if cache.contains_key(bucket) {
SystemTime::now()
.duration_since(SystemTime::UNIX_EPOCH)
@@ -15,9 +15,7 @@
#[cfg(test)]
pub(crate) use rustfs_replication::FailStats;
pub(crate) use rustfs_replication::{
ActiveWorkerStat, ProxyMetric, ProxyStatsCache, QueueCache, ReplicationMetricScope, SRMetricsSummary,
ActiveWorkerStat, BucketReplicationStat, InQueueMetric, ProxyMetric, ProxyStatsCache, QueueCache, ReplicationMetricScope,
SRMetricsSummary, XferStats,
};
// Public so the admin wire DTOs (rustfs/src/admin/replication_metrics_wire.rs)
// can project the internal stats onto the minio-go response shapes through
// the storage_api facade chain.
pub use rustfs_replication::{BucketReplicationStat, BucketReplicationStats, BucketStats, InQueueMetric, XferStats};
pub use rustfs_replication::{BucketReplicationStats, BucketStats};
@@ -27,10 +27,8 @@ use rustfs_utils::http::{
AMZ_OBJECT_TAGGING, AMZ_SERVER_SIDE_ENCRYPTION, AMZ_SERVER_SIDE_ENCRYPTION_KMS_CONTEXT, AMZ_SERVER_SIDE_ENCRYPTION_KMS_ID,
AMZ_STORAGE_CLASS, AMZ_TAG_COUNT, CACHE_CONTROL, CONTENT_DISPOSITION, CONTENT_ENCODING, CONTENT_LANGUAGE, CONTENT_TYPE,
HeaderExt as _, SUFFIX_OBJECTLOCK_LEGALHOLD_TIMESTAMP, SUFFIX_OBJECTLOCK_RETENTION_TIMESTAMP,
SUFFIX_REPLICATION_ACTUAL_OBJECT_SIZE, SUFFIX_REPLICATION_SSEC_CRC, SUFFIX_SOURCE_REPLICATION_LEGALHOLD_TIMESTAMP,
SUFFIX_SOURCE_REPLICATION_RETENTION_TIMESTAMP, SUFFIX_SOURCE_REPLICATION_TAGGING_TIMESTAMP, SUFFIX_TAGGING_TIMESTAMP,
get_str, insert_header_map, is_internal_key, is_object_encryption_marker, is_replication_stripped_encryption_key,
ssec_replication_transport_header,
SUFFIX_REPLICATION_ACTUAL_OBJECT_SIZE, SUFFIX_REPLICATION_SSEC_CRC, SUFFIX_TAGGING_TIMESTAMP, get_str, insert_header_map,
is_internal_key, is_object_encryption_marker, is_replication_stripped_encryption_key, ssec_replication_transport_header,
};
use time::OffsetDateTime;
use time::format_description::well_known::Rfc3339;
@@ -121,27 +119,6 @@ fn classify_replication_source_encryption(metadata: &HashMap<String, String>) ->
}
}
fn is_legacy_source_replication_timestamp_key(key: &str) -> bool {
fn has_prefix_and_suffix(key: &str, prefix: &str, suffix: &str) -> bool {
let key = key.as_bytes();
key.len() == prefix.len() + suffix.len()
&& key[..prefix.len()].eq_ignore_ascii_case(prefix.as_bytes())
&& key[prefix.len()..].eq_ignore_ascii_case(suffix.as_bytes())
}
[
SUFFIX_SOURCE_REPLICATION_TAGGING_TIMESTAMP,
SUFFIX_SOURCE_REPLICATION_RETENTION_TIMESTAMP,
SUFFIX_SOURCE_REPLICATION_LEGALHOLD_TIMESTAMP,
]
.iter()
.any(|suffix| {
["x-rustfs-", "x-minio-"]
.iter()
.any(|prefix| has_prefix_and_suffix(key, prefix, suffix))
})
}
pub(crate) fn replication_object_is_ssec_encrypted(user_defined: &HashMap<String, String>) -> bool {
rustfs_replication::is_ssec_encrypted(user_defined)
}
@@ -199,11 +176,6 @@ pub(crate) fn replication_put_object_options(sc: &str, object_info: &ObjectInfo)
continue;
}
if is_legacy_source_replication_timestamp_key(key) {
meta.insert(format!("x-amz-meta-{key}"), value.to_string());
continue;
}
if is_internal_key(key) || is_standard_header(key) {
continue;
}
@@ -287,23 +259,15 @@ pub(crate) fn replication_put_object_options(sc: &str, object_info: &ObjectInfo)
if !tags.is_empty() {
put_options.user_tags = tags;
put_options.internal.tagging_timestamp =
if let Some(timestamp) = get_str(&object_info.user_defined, SUFFIX_TAGGING_TIMESTAMP) {
OffsetDateTime::parse(&timestamp, &Rfc3339)
.map_err(|err| Error::other(format!("Failed to parse tagging timestamp: {err}")))?
} else {
object_info.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH)
};
}
}
// Load the stored tagging timestamp independently of whether any tags
// remain: DeleteObjectTagging leaves the object tagless but stamps this
// key, and the deletion's LWW timestamp must still reach the replica.
// With no stored key, fall back to mod_time only while tags exist
// (MinIO parity); a tagless object without the key was never tagged and
// keeps the epoch default (no header).
put_options.internal.tagging_timestamp = if let Some(timestamp) = get_str(&object_info.user_defined, SUFFIX_TAGGING_TIMESTAMP)
{
OffsetDateTime::parse(&timestamp, &Rfc3339)
.map_err(|err| Error::other(format!("Failed to parse tagging timestamp: {err}")))?
} else if !put_options.user_tags.is_empty() {
object_info.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH)
} else {
OffsetDateTime::UNIX_EPOCH
};
let metadata = &*object_info.user_defined;
@@ -319,15 +283,13 @@ pub(crate) fn replication_put_object_options(sc: &str, object_info: &ObjectInfo)
put_options.cache_control = cache_control.to_string();
}
if let Some(mode) = metadata.lookup(AMZ_OBJECT_LOCK_MODE).filter(|mode| !mode.is_empty()) {
if let Some(mode) = metadata.lookup(AMZ_OBJECT_LOCK_MODE) {
put_options.mode = Some(ObjectLockRetentionMode::from(mode.to_uppercase().as_str()));
}
if let Some(retain_until_date) = metadata.lookup(AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE) {
if !retain_until_date.is_empty() {
put_options.retain_until_date = OffsetDateTime::parse(retain_until_date, &Rfc3339)
.map_err(|err| Error::other(format!("Failed to parse retain until date: {err}")))?;
}
put_options.retain_until_date = OffsetDateTime::parse(retain_until_date, &Rfc3339)
.map_err(|err| Error::other(format!("Failed to parse retain until date: {err}")))?;
put_options.internal.retention_timestamp =
if let Some(timestamp) = get_str(&object_info.user_defined, SUFFIX_OBJECTLOCK_RETENTION_TIMESTAMP) {
OffsetDateTime::parse(&timestamp, &Rfc3339).unwrap_or(OffsetDateTime::UNIX_EPOCH)
@@ -732,110 +694,6 @@ mod tests {
assert!(options.internal.replication_request);
}
/// DeleteObjectTagging leaves the object tagless but stamps the
/// tagging-timestamp internal key; the deletion's LWW timestamp must
/// still be loaded (and therefore sent) so the replica can order the
/// deletion against concurrent tag edits.
#[test]
fn replication_put_options_carry_tagging_timestamp_after_tag_deletion() {
let mut metadata = std::collections::HashMap::new();
rustfs_utils::http::insert_str(&mut metadata, SUFFIX_TAGGING_TIMESTAMP, "2026-01-02T03:04:05Z".to_string());
let object_info = ObjectInfo {
user_defined: Arc::new(metadata),
user_tags: Arc::new(String::new()),
mod_time: Some(OffsetDateTime::UNIX_EPOCH),
version_id: Some(Uuid::nil()),
..Default::default()
};
let (options, _) = replication_put_object_options("", &object_info).expect("build put options");
assert!(options.user_tags.is_empty());
assert_eq!(
options.internal.tagging_timestamp,
OffsetDateTime::parse("2026-01-02T03:04:05Z", &Rfc3339).expect("valid timestamp"),
"the stored tagging timestamp must load independently of remaining tags"
);
// A tagless object without the stored key was never tagged: the epoch
// default keeps the header unsent.
let untagged = ObjectInfo {
user_tags: Arc::new(String::new()),
mod_time: Some(OffsetDateTime::from_unix_timestamp(1_700_000_000).expect("timestamp")),
version_id: Some(Uuid::nil()),
..Default::default()
};
let (options, _) = replication_put_object_options("", &untagged).expect("build put options");
assert_eq!(options.internal.tagging_timestamp, OffsetDateTime::UNIX_EPOCH);
}
#[test]
fn replication_put_options_do_not_promote_legacy_user_timestamp_metadata() {
let legacy_keys = [
"x-rustfs-source-replication-tagging-timestamp",
"x-rustfs-source-replication-retention-timestamp",
"x-rustfs-source-replication-legalhold-timestamp",
"x-minio-source-replication-tagging-timestamp",
"x-minio-source-replication-retention-timestamp",
"x-minio-source-replication-legalhold-timestamp",
];
let object_info = ObjectInfo {
user_defined: Arc::new(
legacy_keys
.iter()
.map(|key| (key.to_string(), "2099-01-02T03:04:05Z".to_string()))
.collect(),
),
..Default::default()
};
let (options, _) = replication_put_object_options("", &object_info).expect("build put options");
for legacy_key in legacy_keys {
assert!(!options.user_metadata.contains_key(legacy_key));
assert_eq!(
options
.user_metadata
.get(&format!("x-amz-meta-{legacy_key}"))
.map(String::as_str),
Some("2099-01-02T03:04:05Z")
);
}
assert_eq!(options.internal.tagging_timestamp, OffsetDateTime::UNIX_EPOCH);
assert_eq!(options.internal.retention_timestamp, OffsetDateTime::UNIX_EPOCH);
assert_eq!(options.internal.legalhold_timestamp, OffsetDateTime::UNIX_EPOCH);
}
#[test]
fn replication_put_options_carry_retention_timestamp_after_clear() {
let mut metadata = HashMap::from([
(AMZ_OBJECT_LOCK_MODE.to_string(), String::new()),
(AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE.to_string(), String::new()),
]);
rustfs_utils::http::insert_str(&mut metadata, SUFFIX_OBJECTLOCK_RETENTION_TIMESTAMP, "2026-01-02T03:04:05Z".to_string());
let object_info = ObjectInfo {
user_defined: Arc::new(metadata),
..Default::default()
};
let (options, _) = replication_put_object_options("", &object_info).expect("retention clear must replicate");
assert!(options.mode.is_none());
assert_eq!(options.retain_until_date, OffsetDateTime::UNIX_EPOCH);
assert_eq!(
options.internal.retention_timestamp,
OffsetDateTime::parse("2026-01-02T03:04:05Z", &Rfc3339).expect("valid timestamp")
);
let headers = options.header();
assert!(!headers.contains_key(AMZ_OBJECT_LOCK_MODE));
assert!(!headers.contains_key(AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE));
assert_eq!(
rustfs_utils::http::get_header(&headers, SUFFIX_SOURCE_REPLICATION_RETENTION_TIMESTAMP).as_deref(),
Some("2026-01-02T03:04:05Z")
);
}
#[test]
fn replication_put_options_strip_encryption_metadata_from_plaintext_objects() {
use rustfs_utils::http::object_encryption_keys::{INTERNAL_ENCRYPTION_ORIGINAL_SIZE_HEADER, SSEC_ORIGINAL_SIZE_HEADER};
+4 -52
View File
@@ -40,14 +40,7 @@ impl ARN {
impl Display for ARN {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
// The `minio` partition is deliberate: madmin-go's ParseARN
// hard-rejects any other partition, so native mc/madmin tooling can
// only decode remote-target ARNs minted in this form (backlog#1675
// P1-7). Legacy `arn:rustfs:` ARNs persisted by older releases stay
// readable via the FromStr whitelist below; runtime matching between
// targets and replication rules is by full-string equality, so mixed
// partitions coexist safely.
write!(f, "arn:minio:{}:{}:{}:{}", self.arn_type, self.region, self.id, self.bucket)
write!(f, "arn:rustfs:{}:{}:{}:{}", self.arn_type, self.region, self.id, self.bucket)
}
}
@@ -55,12 +48,7 @@ impl FromStr for ARN {
type Err = std::io::Error;
fn from_str(s: &str) -> Result<Self, Self::Err> {
// Partition whitelist, not just an `arn:` check: `BucketTargetType::
// from_str(...).unwrap_or_default()` below never fails, so this is
// the only structural gate rejecting foreign ARNs. `arn:rustfs:` is
// the legacy partition and must stay accepted forever (persisted
// bucket-targets.json / replication configs from older releases).
if !s.starts_with("arn:minio:") && !s.starts_with("arn:rustfs:") {
if !s.starts_with("arn:rustfs:") {
return Err(std::io::Error::new(std::io::ErrorKind::InvalidInput, "Invalid ARN format"));
}
@@ -113,50 +101,14 @@ mod tests {
}
/// RustFS commonly generates ARNs with an empty region:
/// `arn:minio:replication::<deployment_id>:<bucket>`.
/// `arn:rustfs:replication::<deployment_id>:<bucket>`.
#[test]
fn from_str_handles_empty_region_segment() {
let parsed = ARN::from_str("arn:minio:replication::depl-123:bucket-a").expect("valid ARN must parse");
let parsed = ARN::from_str("arn:rustfs:replication::depl-123:bucket-a").expect("valid ARN must parse");
assert_eq!(parsed.arn_type, BucketTargetType::ReplicationService);
assert_eq!(parsed.region, "", "region segment is empty in this form");
assert_eq!(parsed.id, "depl-123");
assert_eq!(parsed.bucket, "bucket-a");
}
/// madmin-go's `ParseARN` hard-rejects anything that does not start with
/// `arn:minio:`, so generated ARNs must use the `minio` partition or the
/// native mc/madmin tooling cannot decode remote-target listings.
#[test]
fn display_emits_minio_partition() {
let arn = ARN::new(
BucketTargetType::ReplicationService,
"depl-123".to_string(),
String::new(),
"bucket-a".to_string(),
);
assert_eq!(arn.to_string(), "arn:minio:replication::depl-123:bucket-a");
}
/// Persisted bucket-targets.json files from older RustFS releases carry
/// `arn:rustfs:` ARNs; the legacy partition must stay parseable forever.
#[test]
fn from_str_accepts_legacy_rustfs_partition() {
let parsed = ARN::from_str("arn:rustfs:replication:us-east-1:depl-123:bucket-a").expect("legacy ARN must parse");
assert_eq!(parsed.arn_type, BucketTargetType::ReplicationService);
assert_eq!(parsed.region, "us-east-1");
assert_eq!(parsed.id, "depl-123");
assert_eq!(parsed.bucket, "bucket-a");
}
/// The partition whitelist is the only structural gate: `BucketTargetType::
/// from_str(...).unwrap_or_default()` never fails, so any 6-segment string
/// would otherwise parse as `type=None`.
#[test]
fn from_str_rejects_unknown_partition() {
assert!(ARN::from_str("arn:aws:replication::depl-123:bucket-a").is_err());
assert!(ARN::from_str("not-an-arn").is_err());
}
}
@@ -59,10 +59,6 @@ impl fmt::Debug for Credentials {
}
#[derive(Debug, Deserialize, Serialize, Default, Clone)]
#[allow(
dead_code,
reason = "MinIO-parity bucket-target service discriminator with no caller in this port (backlog#1823)"
)]
pub enum ServiceType {
#[default]
Replication,
+17 -20
View File
@@ -73,6 +73,23 @@ pub fn check_valid_bucket_name_strict(bucket_name: &str) -> Result<()> {
check_bucket_name_common(bucket_name, true)
}
pub fn check_valid_object_name_prefix(object_name: &str) -> Result<()> {
if object_name.len() > 1024 {
return Err(Error::other("Object name cannot be longer than 1024 characters"));
}
if !object_name.is_ascii() {
return Err(Error::other("Object name with non-UTF-8 strings are not supported"));
}
Ok(())
}
pub fn check_valid_object_name(object_name: &str) -> Result<()> {
if object_name.trim().is_empty() {
return Err(Error::other("Object name cannot be empty"));
}
check_valid_object_name_prefix(object_name)
}
pub fn deserialize<T>(input: &[u8]) -> xml::DeResult<T>
where
T: for<'xml> xml::Deserialize<'xml>,
@@ -83,10 +100,6 @@ where
Ok(ans)
}
#[allow(
dead_code,
reason = "xml serialize helper with no caller in this port; the live sibling is deserialize (backlog#1823)"
)]
pub fn serialize_content<T: xml::SerializeContent>(val: &T) -> xml::SerResult<String> {
let mut buf = Vec::with_capacity(256);
{
@@ -173,27 +186,15 @@ pub fn is_valid_object_name(object: &str) -> bool {
/// Client-facing reason attached to rejections of object keys that Win32/NTFS
/// cannot represent as file paths (issue #3299). Deployments on Linux/macOS
/// accept the full S3 key character set.
#[allow(
dead_code,
reason = "live on Windows: callers sit inside the #[cfg(target_os = \"windows\")] block in check_object_name_for_length_and_slash (backlog#1823)"
)]
pub const WINDOWS_RESERVED_CHARACTERS_REASON: &str =
"object key contains characters unsupported on Windows hosts (one of ':', '*', '?', '\"', '|', '<', '>')";
/// Client-facing reason for path segments Windows can store but not address
/// afterwards (issue #3449): trailing dot/space or reserved DOS device names.
#[allow(
dead_code,
reason = "live on Windows: callers sit inside the #[cfg(target_os = \"windows\")] block in check_object_name_for_length_and_slash (backlog#1823)"
)]
pub const WINDOWS_RESERVED_SEGMENT_REASON: &str = "object key contains a path segment unsupported on Windows hosts (trailing dot or space, or a reserved device name such as NUL/CON/COM1)";
/// Reserved DOS device names that shadow regular files on Windows, even when
/// an extension is appended (e.g. `NUL.txt` resolves to the `NUL` device).
#[allow(
dead_code,
reason = "live on Windows: callers sit inside the #[cfg(target_os = \"windows\")] block in check_object_name_for_length_and_slash (backlog#1823)"
)]
const WINDOWS_RESERVED_NAMES: &[&str] = &[
"CON", "PRN", "AUX", "NUL", "COM1", "COM2", "COM3", "COM4", "COM5", "COM6", "COM7", "COM8", "COM9", "LPT1", "LPT2", "LPT3",
"LPT4", "LPT5", "LPT6", "LPT7", "LPT8", "LPT9",
@@ -203,10 +204,6 @@ const WINDOWS_RESERVED_NAMES: &[&str] = &[
/// the Win32 API cannot address afterwards (issue #3449): segments ending in a
/// dot or a space, and reserved DOS device names — bare or with an extension
/// (`NUL.txt`), matching classic Win32 path resolution semantics.
#[allow(
dead_code,
reason = "live on Windows: callers sit inside the #[cfg(target_os = \"windows\")] block in check_object_name_for_length_and_slash (backlog#1823)"
)]
pub fn object_name_has_windows_incompatible_segment(object: &str) -> bool {
object.split(['/', '\\']).any(|segment| {
if segment.ends_with('.') || segment.ends_with(' ') {
@@ -90,10 +90,6 @@ impl BucketVersioningSys {
/// caller's own instance context so a second in-process store never
/// answers with the first instance's versioning state; falls back to the
/// ambient system when the instance cell is not initialized.
#[allow(
dead_code,
reason = "instance-scoped seam (backlog#1052) with no caller in this port (backlog#1823)"
)]
pub(crate) async fn get_in(ctx: &crate::runtime::instance::InstanceContext, bucket: &str) -> Result<VersioningConfiguration> {
if bucket == RUSTFS_META_BUCKET || bucket.starts_with(RUSTFS_META_BUCKET) {
return Ok(VersioningConfiguration::default());
@@ -15,7 +15,6 @@
use crate::disk::disk_store::{get_drive_walkdir_peek_timeout, get_drive_walkdir_stall_timeout};
use crate::disk::error::DiskError;
use crate::disk::{self, DiskAPI, DiskStore, WalkDirOptions};
use futures::future::join_all;
use metrics::counter;
use rustfs_filemeta::{MetaCacheEntries, MetaCacheEntry, MetacacheReader, is_io_eof};
use std::{
@@ -656,7 +655,6 @@ async fn list_path_raw_inner(
errs.push(None);
}
let mut pending_entries: Vec<Option<MetaCacheEntry>> = vec![None; readers.len()];
let mut peek_outcomes: Vec<Option<PeekOutcome>> = std::iter::repeat_with(|| None).take(readers.len()).collect();
loop {
let mut current = MetaCacheEntry::default();
@@ -678,21 +676,6 @@ async fn list_path_raw_inner(
let mut has_err = 0;
let mut agree = 0;
// Start every missing head read in the same round so one stalled
// disk cannot multiply the wait budget by the erasure-set width.
// Outcomes are still consumed below in stable disk-index order.
let concurrent_peeks = readers.iter_mut().enumerate().filter_map(|(i, reader)| {
if errs[i].is_some() || pending_entries[i].is_some() {
return None;
}
let cancel = &revjob_rx;
Some(async move { (i, peek_with_timeout(cancel, reader, peek_timeout).await) })
});
for (i, outcome) in join_all(concurrent_peeks).await {
peek_outcomes[i] = Some(outcome);
}
for (i, r) in readers.iter_mut().enumerate() {
if errs[i].is_some() {
has_err += 1;
@@ -702,10 +685,7 @@ async fn list_path_raw_inner(
let entry = if let Some(entry) = pending_entries[i].take() {
entry
} else {
let Some(outcome) = peek_outcomes[i].take() else {
return Err(DiskError::Unexpected);
};
match outcome {
match peek_with_timeout(&revjob_rx, r, peek_timeout).await {
PeekOutcome::Ready(res) => {
if let Some(entry) = res {
// info!("read entry disk: {}, name: {}", i, entry.name);
@@ -1315,36 +1295,6 @@ mod tests {
assert_eq!(err, DiskError::Timeout);
}
#[tokio::test(start_paused = true)]
async fn list_path_raw_bounds_multiple_stalled_readers_by_one_peek_deadline() {
let peek_timeout = Duration::from_millis(20);
let started = tokio::time::Instant::now();
let err = list_path_raw(
CancellationToken::new(),
ListPathRawOptions {
disks: vec![None, None, None, None],
min_disks: 1,
test_reader_behaviors: vec![
TestReaderBehavior::Stall,
TestReaderBehavior::Stall,
TestReaderBehavior::Stall,
TestReaderBehavior::Stall,
],
peek_timeout: Some(peek_timeout),
..Default::default()
},
)
.await
.expect_err("all stalled readers should fail the listing");
assert_eq!(err, DiskError::Timeout);
assert_eq!(
started.elapsed(),
peek_timeout,
"reader deadlines must overlap instead of accumulating once per disk"
);
}
#[tokio::test]
async fn list_path_raw_waits_past_producer_stall_for_slow_progressing_reader() {
let entry = MetaCacheEntry {
@@ -229,6 +229,17 @@ pub fn http_resp_to_error_response(
err_resp
}
pub fn err_transfer_acceleration_bucket(bucket_name: &str) -> ErrorResponse {
ErrorResponse {
status_code: StatusCode::BAD_REQUEST,
code: S3ErrorCode::InvalidArgument,
message: "The name of the bucket used for Transfer Acceleration must be DNS-compliant and must not contain periods .."
.to_string(),
bucket_name: bucket_name.to_string(),
..Default::default()
}
}
pub fn err_entity_too_large(total_size: i64, max_object_size: i64, bucket_name: &str, object_name: &str) -> ErrorResponse {
let msg = format!(
"Your proposed upload size {} exceeds the maximum allowed object size {} for single PUT operation.",
@@ -284,6 +295,16 @@ pub fn err_invalid_argument(message: &str) -> ErrorResponse {
}
}
pub fn err_api_not_supported(message: &str) -> ErrorResponse {
ErrorResponse {
status_code: StatusCode::NOT_IMPLEMENTED,
code: S3ErrorCode::Custom("APINotSupported".into()),
message: message.to_string(),
request_id: "rustfs".to_string(),
..Default::default()
}
}
#[cfg(test)]
mod tests {
use super::*;
@@ -135,10 +135,6 @@ impl Object {
Self { ..Default::default() }
}
#[allow(
dead_code,
reason = "MinIO-parity reader surface with no caller in this port (backlog#1823)"
)]
fn do_get_request(&self, request: &GetRequest) -> Result<GetResponse, std::io::Error> {
let _ = request.did_offset_change;
let _ = request.offset;
@@ -154,20 +150,12 @@ impl Object {
))
}
#[allow(
dead_code,
reason = "MinIO-parity Object reader method with no caller in this port (backlog#1823)"
)]
fn set_offset(&mut self, bytes_read: i64) -> Result<(), std::io::Error> {
self.curr_offset += bytes_read;
Ok(())
}
#[allow(
dead_code,
reason = "MinIO-parity Object reader method with no caller in this port (backlog#1823)"
)]
fn read(&mut self, b: &[u8]) -> Result<i64, std::io::Error> {
let mut read_req = GetRequest {
is_read_op: true,
@@ -192,10 +180,6 @@ impl Object {
Ok(response.size)
}
#[allow(
dead_code,
reason = "MinIO-parity Object reader method with no caller in this port (backlog#1823)"
)]
fn stat(&self) -> Result<ObjectInfo, std::io::Error> {
if !self.is_started || !self.object_info_set {
let _ = self.do_get_request(&GetRequest {
@@ -208,10 +192,6 @@ impl Object {
Ok(self.object_info.clone())
}
#[allow(
dead_code,
reason = "MinIO-parity Object reader method with no caller in this port (backlog#1823)"
)]
fn read_at(&mut self, b: &[u8], offset: i64) -> Result<i64, std::io::Error> {
self.curr_offset = offset;
@@ -239,10 +219,6 @@ impl Object {
Ok(response.size)
}
#[allow(
dead_code,
reason = "MinIO-parity Object reader method with no caller in this port (backlog#1823)"
)]
fn seek(&mut self, offset: i64, whence: i64) -> Result<i64, std::io::Error> {
if !self.is_started || !self.object_info_set {
let seek_req = GetRequest {
@@ -277,10 +253,6 @@ impl Object {
Ok(self.curr_offset)
}
#[allow(
dead_code,
reason = "MinIO-parity Object reader method with no caller in this port (backlog#1823)"
)]
fn close(&mut self) -> Result<(), std::io::Error> {
self.is_closed = true;
Ok(())
+1 -1
View File
@@ -37,7 +37,7 @@ use crate::client::{
api_put_object_common::optimal_part_info,
api_put_object_multipart::UploadPartParams,
api_s3_datatypes::{CompleteMultipartUpload, CompletePart, ObjectPart},
constants::{ISO8601_DATEFORMAT, MAX_MULTIPART_PUT_OBJECT_SIZE, MIN_PART_SIZE},
constants::{ISO8601_DATEFORMAT, MAX_MULTIPART_PUT_OBJECT_SIZE, MIN_PART_SIZE, TOTAL_WORKERS},
credentials::SignatureType,
transition_api::{ReaderImpl, TransitionClient, UploadInfo},
utils::{is_amz_header, is_minio_header, is_rustfs_header, is_standard_header, is_storageclass_header},
@@ -30,6 +30,10 @@ pub fn is_object(reader: &ReaderImpl) -> bool {
matches!(reader, ReaderImpl::ObjectBody(_))
}
pub fn is_read_at(reader: ReaderImpl) -> bool {
matches!(reader, ReaderImpl::ObjectBody(_))
}
pub fn optimal_part_info(object_size: i64, configured_part_size: u64) -> Result<(i64, i64, i64), std::io::Error> {
let unknown_size;
let mut object_size = object_size;
@@ -81,6 +81,18 @@ async fn read_multipart_part(reader: &mut ReaderImpl, want: usize) -> Result<Vec
}
}
pub struct UploadedPartRes {
pub error: std::io::Error,
pub part_num: i64,
pub size: i64,
pub part: ObjectPart,
}
pub struct UploadPartReq {
pub part_num: i64,
pub part: ObjectPart,
}
impl TransitionClient {
pub async fn put_object_multipart_stream(
self: Arc<Self>,
+38 -18
View File
@@ -29,6 +29,10 @@ use crate::client::utils::base64_decode;
use super::transition_api;
pub struct ListAllMyBucketsResult {
pub owner: Owner,
}
#[derive(Debug, Default, Serialize, Deserialize)]
pub struct CommonPrefix {
pub prefix: String,
@@ -85,10 +89,6 @@ pub struct ListVersionsResult {
pub next_version_id_marker: String,
}
#[allow(
dead_code,
reason = "fields of a MinIO-parity list result that this port builds but never reads back (backlog#1823)"
)]
pub struct ListBucketResult {
common_prefixes: Vec<CommonPrefix>,
contents: Vec<transition_api::ObjectInfo>,
@@ -102,10 +102,6 @@ pub struct ListBucketResult {
prefix: String,
}
#[allow(
dead_code,
reason = "fields of a MinIO-parity list result that this port builds but never reads back (backlog#1823)"
)]
pub struct ListMultipartUploadsResult {
bucket: String,
key_marker: String,
@@ -121,15 +117,16 @@ pub struct ListMultipartUploadsResult {
common_prefixes: Vec<CommonPrefix>,
}
#[allow(
dead_code,
reason = "fields of a MinIO-parity list result that this port builds but never reads back (backlog#1823)"
)]
pub struct Initiator {
id: String,
display_name: String,
}
pub struct CopyObjectResult {
pub etag: String,
pub last_modified: OffsetDateTime,
}
#[derive(Debug, Clone)]
pub struct ObjectPart {
pub etag: String,
@@ -263,7 +260,6 @@ pub struct CompletePart {
}
impl CompletePart {
#[allow(dead_code, reason = "MinIO-parity accessor with no caller in this port (backlog#1823)")]
fn checksum(&self, t: &ChecksumMode) -> String {
match t {
ChecksumMode::ChecksumCRC32C => {
@@ -288,6 +284,11 @@ impl CompletePart {
}
}
pub struct CopyObjectPartResult {
pub etag: String,
pub last_modified: OffsetDateTime,
}
#[derive(Debug, Default, serde::Serialize)]
#[serde(rename = "CompleteMultipartUpload")]
pub struct CompleteMultipartUpload {
@@ -356,10 +357,10 @@ impl CompleteMultipartUpload {
}
}
#[allow(
dead_code,
reason = "live via quick_xml::de::from_str in bucket_cache.rs; serde deserialization is not a construction (backlog#1823)"
)]
pub struct CreateBucketConfiguration {
pub location: String,
}
#[derive(serde::Serialize)]
pub struct DeleteObject {
//api has
@@ -367,6 +368,21 @@ pub struct DeleteObject {
pub version_id: String,
}
pub struct DeletedObject {
//s3s has
pub key: String,
pub version_id: String,
pub deletemarker: bool,
pub deletemarker_version_id: String,
}
pub struct NonDeletedObject {
pub key: String,
pub code: String,
pub message: String,
pub version_id: String,
}
#[derive(serde::Serialize)]
pub struct DeleteMultiObjects {
pub quiet: bool,
@@ -386,7 +402,6 @@ impl DeleteMultiObjects {
Ok(buf)
}
#[allow(dead_code, reason = "MinIO-parity XML helper with no caller in this port (backlog#1823)")]
pub fn unmarshal(buf: &[u8]) -> Result<Self, std::io::Error> {
#[derive(Debug, Deserialize)]
struct WireDeleteObject {
@@ -421,3 +436,8 @@ impl DeleteMultiObjects {
})
}
}
pub struct DeleteMultiObjectsResult {
pub deleted_objects: Vec<DeletedObject>,
pub undeleted_objects: Vec<NonDeletedObject>,
}
-4
View File
@@ -365,10 +365,6 @@ mod tests {
pub struct Checksum {
checksum_type: ChecksumMode,
r: Vec<u8>,
#[allow(
dead_code,
reason = "checksum bookkeeping field kept beside the value it guards (backlog#1823)"
)]
computed: bool,
}
+3
View File
@@ -32,5 +32,8 @@ pub const MAX_MULTIPART_PUT_OBJECT_SIZE: i64 = 1024 * 1024 * 1024 * 1024 * 5;
pub const UNSIGNED_PAYLOAD: &str = "UNSIGNED-PAYLOAD";
pub const UNSIGNED_PAYLOAD_TRAILER: &str = "STREAMING-UNSIGNED-PAYLOAD-TRAILER";
pub const TOTAL_WORKERS: i64 = 4;
pub const SIGN_V4_ALGORITHM: &str = "AWS4-HMAC-SHA256";
pub const ISO8601_DATEFORMAT: &[FormatItem<'_>] =
format_description!("[year]-[month]-[day]T[hour]:[minute]:[second].[subsecond]Z");
+19 -12
View File
@@ -67,10 +67,6 @@ impl<P: Provider + Default> Credentials<P> {
Ok(self.creds.clone())
}
#[allow(
dead_code,
reason = "MinIO-parity credential surface with no caller in this port (backlog#1823)"
)]
fn expire(&mut self) {
self.force_refresh = true;
}
@@ -137,10 +133,6 @@ impl Provider for Static {
#[derive(Debug, Clone, Default)]
pub struct STSError {
#[allow(
dead_code,
reason = "MinIO-parity STS error detail that this port never reads back (backlog#1823)"
)]
pub r#type: String,
pub code: String,
pub message: String,
@@ -149,10 +141,6 @@ pub struct STSError {
#[derive(Debug, Clone, thiserror::Error)]
pub struct ErrorResponse {
pub sts_error: STSError,
#[allow(
dead_code,
reason = "MinIO-parity STS error detail that this port never reads back (backlog#1823)"
)]
pub request_id: String,
}
@@ -170,3 +158,22 @@ impl ErrorResponse {
return self.sts_error.message.clone();
}
}
pub fn xml_decoder<T>(body: &[u8]) -> Result<T, Error>
where
for<'de> T: Deserialize<'de>,
{
match std::str::from_utf8(body) {
Ok(xml_body) => quick_xml::de::from_str::<T>(xml_body).map_err(|err| Error::new(ErrorKind::InvalidData, err.to_string())),
Err(err) => Err(Error::new(ErrorKind::InvalidData, err.to_string())),
}
}
pub fn xml_decode_and_body<T>(body_reader: &[u8]) -> Result<(Vec<u8>, T), std::io::Error>
where
for<'de> T: Deserialize<'de>,
{
let body = body_reader.to_vec();
let parsed = xml_decoder(&body)?;
Ok((body, parsed))
}
+1
View File
@@ -13,6 +13,7 @@
// limitations under the License.
// #730: S3 client compatibility models are kept while ECStore callers move to narrower facades.
#![allow(dead_code)]
pub mod admin_handler_utils;
pub mod api_error_response;
@@ -77,6 +77,39 @@ fn part_number_to_rangespec(oi: ObjectInfo, part_number: usize) -> Option<HTTPRa
})
}
fn get_compressed_offsets(oi: ObjectInfo, offset: i64) -> (i64, i64, i64, i64, u64) {
let mut skip_length: i64 = 0;
let mut cumulative_actual_size: i64 = 0;
let mut first_part_idx: i64 = 0;
let mut compressed_offset: i64 = 0;
let mut part_skip: i64 = 0;
let mut decrypt_skip: i64 = 0;
let mut seq_num: u64 = 0;
for (i, part) in oi.parts.iter().enumerate() {
cumulative_actual_size += part.actual_size as i64;
if cumulative_actual_size <= offset {
compressed_offset += part.size as i64;
} else {
first_part_idx = i as i64;
skip_length = cumulative_actual_size - part.actual_size as i64;
break;
}
}
skip_length = offset - skip_length;
let parts: &[ObjectPartInfo] = &oi.parts;
if skip_length > 0
&& parts.len() > first_part_idx as usize
&& parts[first_part_idx as usize].index.as_ref().is_some_and(|idx| idx.len() > 0)
{
let _ = part_skip;
let _ = decrypt_skip;
let _ = seq_num;
}
(compressed_offset, part_skip, first_part_idx, decrypt_skip, seq_num)
}
pub fn new_getobjectreader<'a>(
rs: &Option<HTTPRangeSpec>,
oi: &'a ObjectInfo,
@@ -23,7 +23,6 @@ const X_OBS_VERSION_ID: &str = "x-obs-version-id";
const MAX_REMOTE_VERSION_ID_LEN: usize = 1024;
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
#[allow(dead_code, reason = "bucket versioning states kept as a complete vocabulary (backlog#1823)")]
pub(crate) enum BucketVersioningState {
Unknown,
Disabled,
@@ -48,7 +47,6 @@ impl RemoteVersion {
}
}
#[allow(dead_code, reason = "MinIO-parity accessor with no caller in this port (backlog#1823)")]
pub(crate) fn exact_request_id(&self) -> Result<Option<&str>, Error> {
match self {
Self::Unknown => Err(Error::new(
@@ -101,10 +101,6 @@ where
const C_UNKNOWN: i32 = -1;
const C_OFFLINE: i32 = 0;
#[allow(
dead_code,
reason = "reachable only from the unused transition client methods below (backlog#1823)"
)]
const C_ONLINE: i32 = 1;
fn invalid_utf8_header_error(scope: &str, header_name: &str) -> std::io::Error {
@@ -324,10 +320,6 @@ impl TransitionClient {
Ok(client)
}
#[allow(
dead_code,
reason = "MinIO-parity transition client surface with no caller in this port (backlog#1823)"
)]
fn endpoint_url(&self) -> Url {
self.endpoint_url.clone()
}
@@ -356,20 +348,12 @@ impl TransitionClient {
.to_string())
}
#[allow(
dead_code,
reason = "MinIO-parity transition client method with no caller in this port (backlog#1823)"
)]
fn trace_errors_only_off(&self) {
if let Ok(mut trace_errors_only) = self.trace_errors_only.lock() {
*trace_errors_only = false;
}
}
#[allow(
dead_code,
reason = "MinIO-parity transition client method with no caller in this port (backlog#1823)"
)]
fn trace_off(&self) {
if let Ok(mut is_trace_enabled) = self.is_trace_enabled.lock() {
*is_trace_enabled = false;
@@ -379,20 +363,12 @@ impl TransitionClient {
}
}
#[allow(
dead_code,
reason = "MinIO-parity transition client method with no caller in this port (backlog#1823)"
)]
fn set_s3_transfer_accelerate(&self, accelerate_endpoint: &str) {
if let Ok(mut endpoint) = self.s3_accelerate_endpoint.lock() {
*endpoint = accelerate_endpoint.to_string();
}
}
#[allow(
dead_code,
reason = "MinIO-parity transition client method with no caller in this port (backlog#1823)"
)]
fn set_s3_enable_dual_stack(&self, enabled: bool) {
if let Ok(mut dual_stack) = self.s3_dual_stack_enabled.lock() {
*dual_stack = enabled;
@@ -422,18 +398,10 @@ impl TransitionClient {
(hash_algos, hash_sums)
}
#[allow(
dead_code,
reason = "MinIO-parity transition client method with no caller in this port (backlog#1823)"
)]
fn is_online(&self) -> bool {
!self.is_offline()
}
#[allow(
dead_code,
reason = "MinIO-parity transition client method with no caller in this port (backlog#1823)"
)]
fn mark_offline(&self) {
self.health_status
.compare_exchange(C_ONLINE, C_OFFLINE, Ordering::SeqCst, Ordering::SeqCst);
@@ -443,18 +411,10 @@ impl TransitionClient {
self.health_status.load(Ordering::SeqCst) == C_OFFLINE
}
#[allow(
dead_code,
reason = "MinIO-parity transition client method with no caller in this port (backlog#1823)"
)]
fn health_check(hc_duration: Duration) {
let _ = hc_duration;
}
#[allow(
dead_code,
reason = "MinIO-parity transition client method with no caller in this port (backlog#1823)"
)]
fn dump_http(&self, req: &Request<s3s::Body>, resp: &Response<Incoming>) -> Result<(), std::io::Error> {
let mut resp_trace: Vec<u8>;
@@ -1142,7 +1102,6 @@ impl Default for ObjectInfo {
}
impl ObjectInfo {
#[allow(dead_code, reason = "MinIO-parity accessor with no caller in this port (backlog#1823)")]
pub(crate) fn remote_version(
&self,
capabilities: ProviderVersionCapabilities,
+4
View File
@@ -48,6 +48,10 @@ lazy_static! {
};
}
pub fn is_standard_query_value(qs_key: &str) -> bool {
SUPPORTED_QUERY_VALUES[qs_key]
}
pub fn is_storageclass_header(header_key: &str) -> bool {
header_key.to_lowercase() == X_AMZ_STORAGE_CLASS.as_str().to_lowercase()
}
-37
View File
@@ -190,17 +190,6 @@ pub(crate) const GET_METADATA_CACHE_REASON_VERSION_SUSPENDED: &str = "version_su
pub(crate) const GET_METADATA_CACHE_REASON_VERSIONED: &str = "versioned";
pub(crate) const GET_METADATA_EARLY_STOP_REASON_CONFLICTING_METADATA: &str = "conflicting_metadata";
pub(crate) const GET_METADATA_EARLY_STOP_REASON_DELETE_MARKER: &str = "delete_marker";
pub(crate) const GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_BODY_VERIFY: &str = "data_read_inline_body_verify";
pub(crate) const GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_DELETED: &str = "data_read_inline_deleted";
pub(crate) const GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_GEOMETRY: &str = "data_read_inline_geometry";
pub(crate) const GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_IDENTITY_MISMATCH: &str = "data_read_inline_identity_mismatch";
pub(crate) const GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_MISSING_PAYLOAD: &str = "data_read_inline_missing_payload";
pub(crate) const GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_MISSING_SHARD: &str = "data_read_inline_missing_shard";
pub(crate) const GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_NOT_INLINE: &str = "data_read_inline_not_inline";
pub(crate) const GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_PART_SHAPE: &str = "data_read_inline_part_shape";
pub(crate) const GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_REMOTE: &str = "data_read_inline_remote";
pub(crate) const GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_SIZE: &str = "data_read_inline_size";
pub(crate) const GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_TRANSFORMED: &str = "data_read_inline_transformed";
pub(crate) const GET_METADATA_EARLY_STOP_REASON_ERROR: &str = "error";
pub(crate) const GET_METADATA_EARLY_STOP_REASON_INSUFFICIENT_QUORUM: &str = "insufficient_quorum";
pub(crate) const GET_METADATA_EARLY_STOP_REASON_NOT_FOUND: &str = "not_found";
@@ -562,32 +551,6 @@ mod tests {
assert_eq!(GET_METADATA_CACHE_REASON_VERSIONED, "versioned");
assert_eq!(GET_METADATA_EARLY_STOP_REASON_CONFLICTING_METADATA, "conflicting_metadata");
assert_eq!(GET_METADATA_EARLY_STOP_REASON_DELETE_MARKER, "delete_marker");
assert_eq!(
GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_BODY_VERIFY,
"data_read_inline_body_verify"
);
assert_eq!(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_DELETED, "data_read_inline_deleted");
assert_eq!(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_GEOMETRY, "data_read_inline_geometry");
assert_eq!(
GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_IDENTITY_MISMATCH,
"data_read_inline_identity_mismatch"
);
assert_eq!(
GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_MISSING_PAYLOAD,
"data_read_inline_missing_payload"
);
assert_eq!(
GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_MISSING_SHARD,
"data_read_inline_missing_shard"
);
assert_eq!(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_NOT_INLINE, "data_read_inline_not_inline");
assert_eq!(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_PART_SHAPE, "data_read_inline_part_shape");
assert_eq!(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_REMOTE, "data_read_inline_remote");
assert_eq!(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_SIZE, "data_read_inline_size");
assert_eq!(
GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_TRANSFORMED,
"data_read_inline_transformed"
);
assert_eq!(GET_METADATA_EARLY_STOP_REASON_ERROR, "error");
assert_eq!(GET_METADATA_EARLY_STOP_REASON_INSUFFICIENT_QUORUM, "insufficient_quorum");
assert_eq!(GET_METADATA_EARLY_STOP_REASON_NOT_FOUND, "not_found");
+33 -13
View File
@@ -637,23 +637,14 @@ impl Default for DiskOperationMetrics {
}
impl DiskOperationMetrics {
#[allow(
dead_code,
reason = "internal metrics recorder reached only from record() below (backlog#1823)"
)]
fn record_call(&mut self) {
self.lifetime_calls.fetch_add(1, Ordering::Relaxed);
}
#[allow(
dead_code,
reason = "internal metrics recorder reached only from record() below (backlog#1823)"
)]
fn record_latency(&mut self, now_sec: u64, elapsed: Duration) {
self.record_latency_atomic(now_sec, elapsed);
}
#[allow(dead_code, reason = "metrics roll-up with no caller in this port (backlog#1823)")]
fn record(&mut self, now_sec: u64, elapsed: Duration) {
self.record_call();
self.record_latency(now_sec, elapsed);
@@ -779,7 +770,6 @@ impl DiskHealthTracker {
}
/// Set disk as faulty
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
pub fn set_faulty(&self) {
self.status.store(DISK_HEALTH_FAULTY, Ordering::Release);
}
@@ -860,7 +850,6 @@ impl DiskHealthTracker {
became_offline
}
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
pub fn mark_offline(&self, endpoint: &Endpoint, reason: &'static str) -> bool {
let current = self.runtime_state();
if current == RuntimeDriveHealthState::Offline {
@@ -991,13 +980,11 @@ impl DiskHealthTracker {
}
/// Get waiting operations count
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
pub fn waiting_count(&self) -> u32 {
self.waiting.load(Ordering::Relaxed)
}
/// Get last success timestamp
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
pub fn last_success(&self) -> i64 {
self.last_success.load(Ordering::Acquire)
}
@@ -1039,6 +1026,21 @@ impl Default for DiskHealthTracker {
}
}
/// Health check context key for tracking disk operations
#[derive(Debug, Clone)]
struct HealthDiskCtxKey;
#[derive(Debug)]
struct HealthDiskCtxValue {
last_success: Arc<AtomicI64>,
}
impl HealthDiskCtxValue {
fn log_success(&self) {
self.last_success.store(current_unix_nanos(), Ordering::Relaxed);
}
}
/// LocalDiskWrapper wraps a DiskStore with health tracking capabilities.
/// This is similar to Go's xlStorageDiskIDCheck.
#[derive(Debug, Clone)]
@@ -1070,6 +1072,10 @@ impl LocalDiskWrapper {
)
}
pub(crate) fn new_with_health(disk: Arc<LocalDisk>, health_check: bool, health: Arc<DiskHealthTracker>) -> Self {
Self::new_with_health_and_metrics(disk, health_check, health, Arc::new(DiskHealthMetricEpoch::default()))
}
pub(crate) fn new_with_reconnect_state(
disk: Arc<LocalDisk>,
health_check: bool,
@@ -1432,6 +1438,20 @@ impl LocalDiskWrapper {
}
}
async fn check_id(&self, want_id: Option<Uuid>) -> Result<()> {
if want_id.is_none() {
return Ok(());
}
let stored_disk_id = self.disk.get_disk_id().await?;
if stored_disk_id != want_id {
return Err(Error::other(format!("Disk ID mismatch wanted {want_id:?}, got {stored_disk_id:?}")));
}
Ok(())
}
/// Check if disk ID is stale
async fn check_disk_stale(&self) -> Result<()> {
let Some(current_disk_id) = *self.disk_id.read().await else {
-1
View File
@@ -48,7 +48,6 @@ pub fn to_volume_error(io_err: std::io::Error) -> std::io::Error {
}
}
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
pub fn to_disk_error(io_err: std::io::Error) -> std::io::Error {
match io_err.kind() {
std::io::ErrorKind::NotFound => DiskError::DiskNotFound.into(),
-1
View File
@@ -178,7 +178,6 @@ pub async fn remove(path: impl AsRef<Path>) -> io::Result<()> {
}
}
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
pub async fn remove_all(path: impl AsRef<Path>) -> io::Result<()> {
// Try remove_file first; fall back to remove_dir_all if it's a directory
match fs::remove_file(path.as_ref()).await {
-69
View File
@@ -665,7 +665,6 @@ async fn remove_empty_directory_tree_under_mount_lease(
}
#[cfg(unix)]
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
async fn remove_empty_directory_tree_with(
root: &Path,
before_descend: impl FnMut(&Path) -> std::io::Result<()>,
@@ -1017,29 +1016,13 @@ fn record_direct_read_page_fault_delta(path: &'static str, stage: &'static str,
/// When enabled, shard reads bypass the page cache using O_DIRECT flag.
/// Requires aligned buffers (typically 512 bytes or 4096 bytes).
/// Default: false (uses page cache via mmap/pread).
#[allow(
dead_code,
reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)"
)]
const ENV_RUSTFS_OBJECT_DIRECT_IO_READ_ENABLE: &str = "RUSTFS_OBJECT_DIRECT_IO_READ_ENABLE";
#[allow(
dead_code,
reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)"
)]
const DEFAULT_RUSTFS_OBJECT_DIRECT_IO_READ_ENABLE: bool = false;
/// Minimum shard size threshold for O_DIRECT reads.
/// Only shards larger than this threshold will use O_DIRECT.
/// Default: 4MB.
#[allow(
dead_code,
reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)"
)]
const ENV_RUSTFS_OBJECT_DIRECT_IO_READ_THRESHOLD: &str = "RUSTFS_OBJECT_DIRECT_IO_READ_THRESHOLD";
#[allow(
dead_code,
reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)"
)]
const DEFAULT_RUSTFS_OBJECT_DIRECT_IO_READ_THRESHOLD: usize = 4 * 1024 * 1024;
/// Enable O_DIRECT for erasure shard / multipart part data writes (Linux only).
@@ -1053,15 +1036,7 @@ const DEFAULT_RUSTFS_OBJECT_DIRECT_IO_READ_THRESHOLD: usize = 4 * 1024 * 1024;
/// EINVAL/EOPNOTSUPP (tmpfs, overlayfs, 9p, ...) latch the path off and fall
/// back to buffered writes for the whole disk. Non-Linux always falls back.
/// Default: false (buffered writes via the page cache, as before).
#[allow(
dead_code,
reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)"
)]
const ENV_RUSTFS_OBJECT_DIRECT_IO_WRITE_ENABLE: &str = "RUSTFS_OBJECT_DIRECT_IO_WRITE_ENABLE";
#[allow(
dead_code,
reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)"
)]
const DEFAULT_RUSTFS_OBJECT_DIRECT_IO_WRITE_ENABLE: bool = false;
const ENV_RUSTFS_OBJECT_MMAP_POPULATE_ENABLE: &str = "RUSTFS_OBJECT_MMAP_POPULATE_ENABLE";
const DEFAULT_RUSTFS_OBJECT_MMAP_POPULATE_ENABLE: bool = false;
@@ -1120,14 +1095,12 @@ macro_rules! cached_read_env {
cached_read_env! {
/// Check if O_DIRECT reads are enabled.
#[allow(dead_code, reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)")]
fn is_direct_io_read_enabled() -> bool =
rustfs_utils::get_env_bool(ENV_RUSTFS_OBJECT_DIRECT_IO_READ_ENABLE, DEFAULT_RUSTFS_OBJECT_DIRECT_IO_READ_ENABLE);
}
cached_read_env! {
/// Check if O_DIRECT shard/part data writes are enabled.
#[allow(dead_code, reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)")]
fn is_direct_io_write_enabled() -> bool =
rustfs_utils::get_env_bool(ENV_RUSTFS_OBJECT_DIRECT_IO_WRITE_ENABLE, DEFAULT_RUSTFS_OBJECT_DIRECT_IO_WRITE_ENABLE);
}
@@ -1483,7 +1456,6 @@ pub(crate) fn effective_durability(volume: &str) -> DurabilityMode {
cached_read_env! {
/// Get the O_DIRECT read threshold size.
#[allow(dead_code, reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)")]
fn get_direct_io_read_threshold() -> usize =
rustfs_utils::get_env_usize(ENV_RUSTFS_OBJECT_DIRECT_IO_READ_THRESHOLD, DEFAULT_RUSTFS_OBJECT_DIRECT_IO_READ_THRESHOLD);
}
@@ -1701,20 +1673,12 @@ impl DirectIoWriteState {
/// Target staging size for O_DIRECT writes, rounded up to the DIO alignment.
/// Bounds the per-writer aligned bounce buffer and batches many shard blocks
/// into one positioned write to keep the syscall count low.
#[allow(
dead_code,
reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)"
)]
const DIRECT_WRITE_STAGING_BYTES: usize = 1024 * 1024;
/// Aligned bounce-buffer capacity for a given DIO alignment: the target staging
/// size rounded up to a whole multiple of `align` so the buffer address, every
/// flushed batch length, and every write offset stay alignment-correct.
/// Platform-independent (no O_DIRECT), so it is unit-tested on any host.
#[allow(
dead_code,
reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)"
)]
fn direct_write_staging_capacity(align: usize) -> usize {
debug_assert!(align.is_power_of_two() && align >= 512);
DIRECT_WRITE_STAGING_BYTES.div_ceil(align) * align
@@ -1723,10 +1687,6 @@ fn direct_write_staging_capacity(align: usize) -> usize {
/// Split `filled` staged bytes into the alignment-sized prefix written with
/// O_DIRECT and the sub-alignment tail written buffered. Platform-independent,
/// so the tail-boundary math is unit-tested on any host.
#[allow(
dead_code,
reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)"
)]
fn direct_write_tail_split(filled: usize, align: usize) -> (usize, usize) {
let aligned = filled - (filled % align);
(aligned, filled - aligned)
@@ -2182,7 +2142,6 @@ fn set_delete_version_fail_after_data_staged(path: &str) {
}
#[cfg(test)]
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
pub(crate) fn set_delete_version_fail_after_commit(root: &Path, path: &str) {
DELETE_VERSION_FAIL_AFTER_COMMIT
.lock()
@@ -2488,10 +2447,6 @@ enum SyncMode {
FileOnly,
}
#[allow(
dead_code,
reason = "reclaim bookkeeping fields written by Drop but never read back (backlog#1823)"
)]
struct FileCacheReclaimWriter {
inner: File,
reclaim_len: usize,
@@ -2499,10 +2454,6 @@ struct FileCacheReclaimWriter {
reclaimed: bool,
}
#[allow(
dead_code,
reason = "reclaim bookkeeping fields written by Drop but never read back (backlog#1823)"
)]
struct FileCacheReclaimReader {
inner: File,
reclaim_offset: u64,
@@ -2568,10 +2519,6 @@ impl<R: AsyncRead + Unpin> AsyncRead for StallTimeoutReader<R> {
}
}
#[allow(
dead_code,
reason = "reclaim metrics emitter reached only from the Linux-gated reclaim paths (backlog#1823)"
)]
fn record_file_cache_reclaim_success(kind: &'static str, reclaim_len: usize, started: std::time::Instant) {
// Runs per read-stream page-cache reclaim window; skip the whole emission
// (three metric-key constructions) when general metrics are disabled.
@@ -3124,7 +3071,6 @@ impl LocalIoBackend for StdBackend {
use memmap2::MmapOptions;
use std::time::{Duration as StdDuration, Instant as StdInstant};
#[allow(dead_code, reason = "mmap copy result slot kept beside the mapping it owns (backlog#1823)")]
struct MmapCopyReadResult {
bytes: Bytes,
access_check_duration: StdDuration,
@@ -4758,10 +4704,6 @@ fn build_local_io_backend(root: PathBuf) -> Arc<dyn LocalIoBackend> {
Arc::new(StdBackend::new(root))
}
#[allow(
dead_code,
reason = "path cache and cwd slots retained beside the disk root they derive from (backlog#1823)"
)]
pub struct LocalDisk {
pub root: PathBuf,
publication_root: os::PublicationRoot,
@@ -5548,7 +5490,6 @@ impl LocalDisk {
Ok(Self::resolve_abs_path_from(&self.root, path.as_ref()))
}
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
fn io_resolve_abs_path(&self, path: impl AsRef<Path>) -> PathBuf {
let path_ref = path.as_ref();
let path_str = path_ref.to_string_lossy();
@@ -5626,24 +5567,15 @@ impl LocalDisk {
}
// Check if a path is valid
#[allow(
dead_code,
reason = "method wrapper over the live free function check_local_disk_valid_path; no caller in this port (backlog#1823)"
)]
fn check_valid_path<P: AsRef<Path>>(&self, path: P) -> Result<()> {
check_local_disk_valid_path(self.io_root(), path)
}
#[allow(
dead_code,
reason = "method wrapper over the live free function reject_local_disk_symlink_components; no caller in this port (backlog#1823)"
)]
fn reject_symlink_components(&self, path: &Path) -> Result<()> {
reject_local_disk_symlink_components(self.io_root(), path)
}
// Batch path generation with single lock acquisition
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
fn get_object_paths_batch(&self, requests: &[(String, String)]) -> Result<Vec<PathBuf>> {
let mut results = Vec::with_capacity(requests.len());
let mut cache_misses = Vec::new();
@@ -6556,7 +6488,6 @@ impl LocalDisk {
Ok(f)
}
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
async fn open_file_read_only(&self, path: impl AsRef<Path>) -> Result<File> {
let f = super::fs::open_file(path.as_ref(), O_RDONLY).await.map_err(to_file_error)?;
Ok(f)
+1 -5
View File
@@ -13,6 +13,7 @@
// limitations under the License.
// #730: disk abstractions still carry staged health and direct-I/O migration paths.
#![allow(dead_code)]
pub mod disk_store;
pub mod endpoint;
@@ -1113,10 +1114,6 @@ pub struct DiskInfo {
}
#[derive(Clone, Debug, Default)]
#[allow(
dead_code,
reason = "MinIO-parity disk info shape with no constructor in this port (backlog#1823)"
)]
pub struct Info {
pub total: u64,
pub free: u64,
@@ -1375,7 +1372,6 @@ pub fn conv_part_err_to_int(err: &Option<Error>) -> usize {
}
}
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
pub fn has_part_err(part_errs: &[usize]) -> bool {
part_errs.iter().any(|err| *err != CHECK_PART_SUCCESS)
}
+11 -5
View File
@@ -571,10 +571,6 @@ fn regular_files(dir: &Path) -> io::Result<Vec<PathBuf>> {
/// Fdatasync every regular file directly inside `dir`, then fsync the directory
/// itself.
#[allow(
dead_code,
reason = "reached only through sync_dir_files, whose callers are tests (backlog#1823)"
)]
pub fn sync_dir_files_std(dir: impl AsRef<Path>) -> io::Result<()> {
for entry in std::fs::read_dir(dir.as_ref())? {
let entry = entry?;
@@ -587,7 +583,6 @@ pub fn sync_dir_files_std(dir: impl AsRef<Path>) -> io::Result<()> {
/// Async wrapper around [`sync_dir_files_std`]. Large directories flush files
/// concurrently, bounded both per directory and process-wide.
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
pub async fn sync_dir_files(dir: impl AsRef<Path>) -> io::Result<()> {
sync_dir_files_with_limiter(dir, Arc::new(Semaphore::new(MAX_PARALLEL_FILE_SYNCS))).await
}
@@ -1814,6 +1809,10 @@ impl RenameCommitGuard {
})
}
pub(crate) fn lock_destination_directory_for_path_access(&self, directory: &Path) -> io::Result<RenameDestinationPathGuard> {
self.destination_directory_guard(directory, false)
}
pub(crate) fn create_destination_directory_for_path_access(
&self,
directory: &Path,
@@ -2859,6 +2858,13 @@ pub async fn os_mkdir_all(dir_path: impl AsRef<Path>, base_dir: impl AsRef<Path>
Ok(())
}
/// Check if a file exists.
/// Returns true if the file exists, false otherwise.
#[tracing::instrument(level = "debug", skip_all)]
pub fn file_exists(path: impl AsRef<Path>) -> bool {
std::fs::metadata(path.as_ref()).map(|_| true).unwrap_or(false)
}
/// Whether an [`io::Error`] means "the directory is not empty".
///
/// POSIX lets `rmdir`/`rename` report a non-empty directory as either
-1
View File
@@ -704,7 +704,6 @@ pub(crate) async fn create_bitrot_reader_from_bytes_with_stage_metrics(
}
#[allow(clippy::too_many_arguments)]
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
pub fn create_deferred_bitrot_reader(
inline_data: Option<Bytes>,
disk: Option<DiskStore>,
-6
View File
@@ -277,12 +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,
/// 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.
pub replication_tagging_timestamp: Option<OffsetDateTime>,
pub replication_retention_timestamp: Option<OffsetDateTime>,
pub replication_legalhold_timestamp: Option<OffsetDateTime>,
/// Authorized SSE-C replication passthrough: the body is already
/// ciphertext, so the write path must not encrypt or compress it and
/// stores the restored encryption metadata verbatim. Only the
+49 -432
View File
@@ -32,22 +32,15 @@ use crate::diagnostics::get::{
GET_METADATA_CACHE_REASON_NOT_READ_DATA, GET_METADATA_CACHE_REASON_PART_NUMBER,
GET_METADATA_CACHE_REASON_RAW_DATA_MOVEMENT_READ, GET_METADATA_CACHE_REASON_USABLE, GET_METADATA_CACHE_REASON_VERSION_ID,
GET_METADATA_CACHE_REASON_VERSION_SUSPENDED, GET_METADATA_CACHE_REASON_VERSIONED,
GET_METADATA_EARLY_STOP_REASON_CONFLICTING_METADATA, GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_BODY_VERIFY,
GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_DELETED, GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_GEOMETRY,
GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_IDENTITY_MISMATCH,
GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_MISSING_PAYLOAD,
GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_MISSING_SHARD, GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_NOT_INLINE,
GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_PART_SHAPE, GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_REMOTE,
GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_SIZE, GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_TRANSFORMED,
GET_METADATA_EARLY_STOP_REASON_DELETE_MARKER, GET_METADATA_EARLY_STOP_REASON_ERROR,
GET_METADATA_EARLY_STOP_REASON_INSUFFICIENT_QUORUM, GET_METADATA_EARLY_STOP_REASON_NOT_FOUND,
GET_METADATA_EARLY_STOP_REASON_UNSAFE_REQUEST, GET_METADATA_EARLY_STOP_REASON_VALID_QUORUM,
GET_METADATA_EARLY_STOP_REASON_VERSION_MATCH_QUORUM, GET_METADATA_EARLY_STOP_REASON_VERSION_NOT_FOUND,
GET_METADATA_RESPONSE_CORRUPT, GET_METADATA_RESPONSE_DISK_NOT_FOUND, GET_METADATA_RESPONSE_ERROR,
GET_METADATA_RESPONSE_IGNORED, GET_METADATA_RESPONSE_NOT_FOUND, GET_METADATA_RESPONSE_TIMEOUT, GET_METADATA_RESPONSE_VALID,
GET_METADATA_RESPONSE_VERSION_NOT_FOUND, GET_OBJECT_PATH_CODEC_STREAMING, GET_OBJECT_PATH_DIRECT_MEMORY,
GET_OBJECT_PATH_INTERNAL_META, GET_OBJECT_PATH_LEGACY_DUPLEX, GET_OBJECT_PATH_SET_DISK, GET_STAGE_DECODE,
GET_STAGE_METADATA_CACHE_LOOKUP, GET_STAGE_METADATA_RESOLVE, GET_STAGE_RANGE, GET_STAGE_READER_SETUP,
GET_METADATA_EARLY_STOP_REASON_CONFLICTING_METADATA, GET_METADATA_EARLY_STOP_REASON_DELETE_MARKER,
GET_METADATA_EARLY_STOP_REASON_ERROR, GET_METADATA_EARLY_STOP_REASON_INSUFFICIENT_QUORUM,
GET_METADATA_EARLY_STOP_REASON_NOT_FOUND, GET_METADATA_EARLY_STOP_REASON_UNSAFE_REQUEST,
GET_METADATA_EARLY_STOP_REASON_VALID_QUORUM, GET_METADATA_EARLY_STOP_REASON_VERSION_MATCH_QUORUM,
GET_METADATA_EARLY_STOP_REASON_VERSION_NOT_FOUND, GET_METADATA_RESPONSE_CORRUPT, GET_METADATA_RESPONSE_DISK_NOT_FOUND,
GET_METADATA_RESPONSE_ERROR, GET_METADATA_RESPONSE_IGNORED, GET_METADATA_RESPONSE_NOT_FOUND, GET_METADATA_RESPONSE_TIMEOUT,
GET_METADATA_RESPONSE_VALID, GET_METADATA_RESPONSE_VERSION_NOT_FOUND, GET_OBJECT_PATH_CODEC_STREAMING,
GET_OBJECT_PATH_DIRECT_MEMORY, GET_OBJECT_PATH_INTERNAL_META, GET_OBJECT_PATH_LEGACY_DUPLEX, GET_OBJECT_PATH_SET_DISK,
GET_STAGE_DECODE, GET_STAGE_METADATA_CACHE_LOOKUP, GET_STAGE_METADATA_RESOLVE, GET_STAGE_RANGE, GET_STAGE_READER_SETUP,
GET_STAGE_READER_SETUP_DROP_PENDING, GET_STAGE_READER_SETUP_SCHEDULE, GET_STAGE_READER_SETUP_WAIT_QUORUM,
GET_STAGE_READER_TASK_BITROT_READER_INIT, GET_STAGE_READER_TASK_FILE_OPEN, GET_STAGE_READER_TASK_READER_CONSTRUCTION,
GetObjectFailureReason, classify_disk_error, get_stage_timer_if_enabled, record_get_object_pipeline_failure,
@@ -180,13 +173,11 @@ pub(in crate::set_disk) enum GetCodecStreamingReaderBuildOutcome {
Fallback(GetCodecStreamingFallbackReason),
}
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
pub(in crate::set_disk) struct MultipartCodecStreamingReader {
pub(in crate::set_disk) readers: VecDeque<Box<dyn AsyncRead + Unpin + Send + Sync>>,
}
impl MultipartCodecStreamingReader {
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
pub(in crate::set_disk) fn new(readers: Vec<Box<dyn AsyncRead + Unpin + Send + Sync>>) -> Self {
Self {
readers: VecDeque::from(readers),
@@ -661,49 +652,36 @@ pub(in crate::set_disk) fn metadata_early_stop_candidate_matches(left: &FileInfo
&& left.erasure.distribution == right.erasure.distribution
}
pub(in crate::set_disk) async fn data_read_early_stop_inline_body_miss_reason(
pub(in crate::set_disk) async fn data_read_early_stop_inline_body_verified(
bucket: &str,
object: &str,
candidate: &FileInfo,
parts_metadata: &[FileInfo],
disks: &[Option<DiskStore>],
) -> Option<&'static str> {
// `inline_data` excludes remote objects; this diagnostic reports them separately.
if !rustfs_utils::http::contains_key_str(&candidate.metadata, rustfs_utils::http::SUFFIX_INLINE_DATA) {
return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_NOT_INLINE);
}
if candidate.is_compressed()
) -> bool {
if !candidate.inline_data()
|| candidate.is_compressed()
|| candidate
.metadata
.keys()
.any(|key| rustfs_utils::http::is_object_encryption_marker(key))
|| candidate.is_remote()
|| candidate.deleted
|| candidate.size <= 0
|| candidate.parts.len() != 1
|| !candidate.has_valid_erasure_geometry()
{
return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_TRANSFORMED);
}
if candidate.is_remote() {
return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_REMOTE);
}
if candidate.deleted {
return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_DELETED);
}
if candidate.size <= 0 {
return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_SIZE);
}
if candidate.parts.len() != 1 {
return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_PART_SHAPE);
}
if !candidate.has_valid_erasure_geometry() {
return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_GEOMETRY);
return false;
}
let Ok(object_size) = usize::try_from(candidate.size) else {
return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_SIZE);
return false;
};
if candidate.parts.first().is_none_or(|part| part.size != object_size) {
return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_PART_SHAPE);
return false;
}
if !can_try_inline_data_shards_direct(object_size, candidate.erasure.block_size) {
return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_SIZE);
return false;
}
let Ok(erasure) = coding::Erasure::try_new_with_options(
@@ -712,18 +690,18 @@ pub(in crate::set_disk) async fn data_read_early_stop_inline_body_miss_reason(
candidate.erasure.block_size,
candidate.uses_legacy_checksum,
) else {
return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_GEOMETRY);
return false;
};
let data_files =
match collect_inline_data_shard_fileinfos_by_index_or_reason(parts_metadata, candidate, erasure.data_shards, |index| {
let Some(data_files) =
collect_inline_data_shard_fileinfos_by_index(parts_metadata, candidate, erasure.data_shards, |index| {
disks.get(index).is_some_and(Option::is_some)
}) {
Ok(data_files) => data_files,
Err(reason) => return Some(reason),
};
})
else {
return false;
};
let Some(part) = candidate.parts.first() else {
return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_PART_SHAPE);
return false;
};
let checksum_info = candidate.erasure.get_checksum_info(part.number);
let checksum_algo = if candidate.uses_legacy_checksum && checksum_info.algorithm == HashAlgorithm::HighwayHash256S {
@@ -743,70 +721,12 @@ pub(in crate::set_disk) async fn data_read_early_stop_inline_body_miss_reason(
let Ok(mut readers) =
build_inline_bitrot_readers_from_refs(&data_files, bucket, object, read_length, shard_size, &checksum_algo, false).await
else {
return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_BODY_VERIFY);
};
match try_read_inline_data_shards_direct(&mut readers, erasure.data_shards, read_length, object_size).await {
Some(body) if body.len() == object_size => None,
_ => Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_BODY_VERIFY),
}
}
fn data_read_inline_missing_shards_are_pending(
candidate: &FileInfo,
parts_metadata: &[FileInfo],
errors: &[Option<DiskError>],
disks: &[Option<DiskStore>],
fanout_order: &[usize],
scheduled_fanout_len: usize,
) -> bool {
let Ok(erasure) = coding::Erasure::try_new_with_options(
candidate.erasure.data_blocks,
candidate.erasure.parity_blocks,
candidate.erasure.block_size,
candidate.uses_legacy_checksum,
) else {
return false;
};
let distribution = &candidate.erasure.distribution;
let mut data_shards_seen_or_pending = vec![false; erasure.data_shards];
let mut missing_pending_data_shards = 0usize;
for (disk_index, file_info) in parts_metadata.iter().enumerate() {
let Some(&block_index) = distribution.get(disk_index) else {
return false;
};
if block_index == 0 || block_index > erasure.data_shards {
continue;
}
if !disks.get(disk_index).is_some_and(Option::is_some) {
return false;
}
let data_slot = block_index - 1;
if file_info.name.is_empty() {
let scheduled_and_not_failed = fanout_order
.get(..scheduled_fanout_len)
.is_some_and(|scheduled_disks| scheduled_disks.contains(&disk_index))
&& errors.get(disk_index).is_some_and(Option::is_none);
if scheduled_and_not_failed {
data_shards_seen_or_pending[data_slot] = true;
missing_pending_data_shards = missing_pending_data_shards.saturating_add(1);
continue;
}
return false;
}
if file_info.erasure.index != block_index
|| !file_info.has_valid_erasure_geometry()
|| !metadata_early_stop_candidate_matches(file_info, candidate)
|| file_info.data.as_ref().is_none_or(|data| data.is_empty())
{
return false;
}
data_shards_seen_or_pending[data_slot] = true;
}
missing_pending_data_shards > 0 && data_shards_seen_or_pending.into_iter().all(|seen_or_pending| seen_or_pending)
try_read_inline_data_shards_direct(&mut readers, erasure.data_shards, read_length, object_size)
.await
.is_some_and(|body| body.len() == object_size)
}
pub(in crate::set_disk) fn classify_metadata_response_error(err: &DiskError) -> &'static str {
@@ -1838,7 +1758,6 @@ pub(in crate::set_disk) async fn create_bitrot_readers_until_quorum_all_shards(
}
#[allow(clippy::too_many_arguments)]
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
pub(in crate::set_disk) async fn create_bitrot_readers_until_quorum(
files: &[FileInfo],
disks: &[Option<DiskStore>],
@@ -2129,7 +2048,6 @@ pub(in crate::set_disk) async fn create_data_block_bitrot_readers(
setup
}
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
pub(in crate::set_disk) async fn collect_read_multiple_results<F>(
tasks: Vec<F>,
read_quorum: usize,
@@ -2551,7 +2469,6 @@ impl SetDisks {
let mut next_fanout_index = 0usize;
let mut scheduled_count = 0usize;
let mut force_full_wait = false;
let mut final_miss_reason_override = None;
let spawn_read_version =
|join_set: &mut JoinSet<(usize, disk::error::Result<FileInfo>, Duration)>, index: usize, disk: Option<DiskStore>| {
let task_opts = opts;
@@ -2594,7 +2511,6 @@ impl SetDisks {
}
while let Some(result) = join_set.join_next().await {
let mut defer_pending_inline_data_shard = false;
match result {
Ok((index, res, elapsed)) => match res {
Ok(file_info) => {
@@ -2625,43 +2541,17 @@ impl SetDisks {
.or_else(|| accumulator.version_early_stop_decision())
{
let should_return_early = if read_data {
match accumulator.candidate.as_ref() {
Some(candidate) => match data_read_early_stop_inline_body_miss_reason(
bucket.as_ref(),
object.as_ref(),
candidate,
&ress,
disks,
)
.await
{
None => true,
Some(reason) => {
final_miss_reason_override = Some(reason);
if bounded_fanout
&& reason == GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_MISSING_SHARD
&& data_read_inline_missing_shards_are_pending(
candidate,
&ress,
&errors,
disks,
&fanout_order,
next_fanout_index,
)
{
defer_pending_inline_data_shard = true;
} else {
force_full_wait = true;
}
false
}
},
None => {
force_full_wait = true;
final_miss_reason_override = Some(GET_METADATA_EARLY_STOP_REASON_INSUFFICIENT_QUORUM);
false
let allow_data_read_early_stop = match accumulator.candidate.as_ref() {
Some(candidate) => {
data_read_early_stop_inline_body_verified(bucket.as_ref(), object.as_ref(), candidate, &ress, disks)
.await
}
None => false,
};
if !allow_data_read_early_stop {
force_full_wait = true;
}
allow_data_read_early_stop
} else {
true
};
@@ -2698,7 +2588,6 @@ impl SetDisks {
let pending_responses = join_set.len();
let should_hedge_single_pending_data_read = read_data
&& !force_full_wait
&& !defer_pending_inline_data_shard
&& pending_responses == 1
&& accumulator.can_still_reach_early_stop_with_pending(pending_responses);
if bounded_fanout && force_full_wait {
@@ -2711,7 +2600,6 @@ impl SetDisks {
next_fanout_index = next_fanout_index.saturating_add(1);
}
} else if bounded_fanout
&& !defer_pending_inline_data_shard
&& next_fanout_index < disks.len()
&& (!accumulator.can_still_reach_early_stop_with_pending(pending_responses)
|| should_hedge_single_pending_data_read)
@@ -2725,12 +2613,7 @@ impl SetDisks {
}
}
let accumulator_miss_reason = accumulator.final_miss_reason();
let final_miss_reason = match (final_miss_reason_override, accumulator_miss_reason) {
(Some(reason), GET_METADATA_EARLY_STOP_REASON_INSUFFICIENT_QUORUM) => reason,
_ => accumulator_miss_reason,
};
rustfs_io_metrics::record_get_object_metadata_early_stop_miss(metrics_path, final_miss_reason);
rustfs_io_metrics::record_get_object_metadata_early_stop_miss(metrics_path, accumulator.final_miss_reason());
rustfs_io_metrics::record_get_object_metadata_early_stop_saved_responses(metrics_path, 0);
rustfs_io_metrics::record_get_object_metadata_fanout_lifecycle(metrics_path, scheduled_count, scheduled_count, 0);
let diagnostics = MetadataFanoutDiagnostics::new(fanout_start.elapsed(), observations);
@@ -2959,7 +2842,6 @@ impl SetDisks {
(meta_file_infos, errs)
}
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
pub(in crate::set_disk) async fn read_multiple_files(
disks: &[Option<DiskStore>],
req: ReadMultipleReq,
@@ -3139,7 +3021,6 @@ pub(in crate::set_disk) struct RenameDataCommit {
pub(in crate::set_disk) committed_file_info: FileInfo,
}
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
type RenameDataLegacyTuple = (
Vec<Option<DiskStore>>,
RenameConvergence,
@@ -3149,7 +3030,6 @@ type RenameDataLegacyTuple = (
);
impl RenameDataCommit {
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
fn into_legacy_tuple(self) -> RenameDataLegacyTuple {
(
self.online_disks,
@@ -3268,7 +3148,6 @@ impl SetDisks {
#[tracing::instrument(level = "debug", skip(disks, file_infos))]
#[allow(clippy::type_complexity)]
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
pub(in crate::set_disk) async fn rename_data(
disks: &[Option<DiskStore>],
src_bucket: &str,
@@ -5081,7 +4960,6 @@ fn is_cleanup_not_found(e: &DiskError) -> bool {
/// normalized to `DiskNotFound`: a panic is not a "disk absent" condition and
/// must not be silently swallowed as an ignorable error (fixes the historical
/// `Unexpected`/`DiskNotFound` misclassification).
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
fn map_cleanup_join_result(joined: std::result::Result<Option<DiskError>, tokio::task::JoinError>) -> Option<DiskError> {
match joined {
Ok(res) => res,
@@ -5306,7 +5184,6 @@ pub(in crate::set_disk) mod rename_fanout_barrier_phase {
/// The per-disk old-data-dir cleanup phase of the commit fan-out.
pub const CLEANUP: &str = "cleanup";
/// The per-disk `read_version` phase of metadata read fan-out.
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
pub const READ_VERSION: &str = "read_version";
}
@@ -5875,20 +5752,9 @@ mod tests {
object: &str,
payload: &[u8],
uses_legacy_checksum: bool,
) -> Vec<FileInfo> {
inline_metadata_fanout_fileinfos_with_geometry(bucket, object, payload, uses_legacy_checksum, 2, 2).await
}
async fn inline_metadata_fanout_fileinfos_with_geometry(
bucket: &str,
object: &str,
payload: &[u8],
uses_legacy_checksum: bool,
data_shards: usize,
parity_shards: usize,
) -> Vec<FileInfo> {
let distribution_key = metadata_distribution_key(bucket, object);
let mut base = FileInfo::new(&distribution_key, data_shards, parity_shards);
let mut base = FileInfo::new(&distribution_key, 2, 2);
base.volume = bucket.to_string();
base.name = object.to_string();
base.size = i64::try_from(payload.len()).expect("test payload should fit i64");
@@ -5951,21 +5817,6 @@ mod tests {
install_inline_metadata_fanout_files(disks, bucket, object, files).await;
}
async fn install_inline_metadata_fanout_fileinfo_with_geometry(
disks: &[Option<DiskStore>],
bucket: &str,
object: &str,
payload: &[u8],
data_shards: usize,
parity_shards: usize,
mutate: impl FnOnce(&mut [FileInfo]),
) {
let mut files =
inline_metadata_fanout_fileinfos_with_geometry(bucket, object, payload, false, data_shards, parity_shards).await;
mutate(&mut files);
install_inline_metadata_fanout_files(disks, bucket, object, files).await;
}
async fn install_inline_metadata_fanout_files(disks: &[Option<DiskStore>], bucket: &str, object: &str, files: Vec<FileInfo>) {
let distribution = files
.first()
@@ -6186,118 +6037,6 @@ mod tests {
drop(dirs);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn bounded_metadata_early_stop_waits_for_pending_inline_data_shard() {
const DISKS: usize = 6;
const DATA_SHARDS: usize = 4;
const PARITY_SHARDS: usize = 2;
let bucket = "bounded-inline-data-get-pending-shard-bucket";
let object =
object_with_initial_data_shards(bucket, "bounded-inline-data-get-pending-shard-object", DATA_SHARDS, DATA_SHARDS);
let (dirs, disks) = call_counter_local_disks(bucket, DISKS).await;
install_inline_metadata_fanout_fileinfo_with_geometry(
&disks,
bucket,
&object,
b"verified inline payload",
DATA_SHARDS,
PARITY_SHARDS,
|_| {},
)
.await;
temp_env::async_with_vars(
[
("RUSTFS_GET_METADATA_EARLY_STOP_ENABLE", Some("true")),
("RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE", Some("true")),
("RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT", Some("true")),
],
async {
let fanout_order = bounded_metadata_fanout_order(bucket, &object, DISKS, PARITY_SHARDS);
let distribution_key = metadata_distribution_key(bucket, &object);
let distribution = FileInfo::new(&distribution_key, DATA_SHARDS, PARITY_SHARDS)
.erasure
.distribution;
let paused_data_disk = *fanout_order
.iter()
.take(DATA_SHARDS)
.find(|disk_index| {
distribution
.get(**disk_index)
.is_some_and(|block_index| (1..=DATA_SHARDS).contains(block_index))
})
.expect("initial fanout should include a data shard to pause");
let hedged_parity_disk = fanout_order[DATA_SHARDS];
let unscheduled_parity_disk = fanout_order[DATA_SHARDS + 1];
let barrier = rename_fanout_barrier::arm(&object, paused_data_disk, rename_fanout_barrier::PHASE_READ_VERSION);
let tracker = rename_fanout_barrier::observe_tasks(&object);
let calls = disk_call_counters::observe(&object);
let disks_for_read = disks.clone();
let object_for_read = object.clone();
let mut read = tokio::spawn(async move {
SetDisks::read_all_fileinfo_observed(
&disks_for_read,
bucket,
bucket,
&object_for_read,
"",
true,
false,
false,
true,
PARITY_SHARDS,
)
.await
});
tokio::time::timeout(BARRIER_PAUSE_GUARD, barrier.wait_until_paused())
.await
.expect("initial data shard should pause before returning");
tokio::time::timeout(BARRIER_PAUSE_GUARD, async {
while calls.for_disk(disk_call_counters::KIND_READ_VERSION, hedged_parity_disk) == 0 {
tokio::task::yield_now().await;
}
})
.await
.expect("bounded fanout should hedge one parity disk while the data shard is pending");
assert!(
tokio::time::timeout(BARRIER_PAUSE_GUARD, &mut read).await.is_err(),
"inline data-read early-stop must wait for a scheduled missing data shard instead of forcing full wait"
);
barrier.release();
let (parts_metadata, errs, diagnostics) = read
.await
.expect("metadata read task should not panic")
.expect("pending data shard should let the inline verifier finish");
assert_eq!(
calls.total(disk_call_counters::KIND_READ_VERSION),
5,
"pending data-shard defer should not schedule the final parity disk"
);
assert_eq!(
calls.for_disk(disk_call_counters::KIND_READ_VERSION, unscheduled_parity_disk),
0,
"the remaining parity disk must stay unissued when pending data verification succeeds"
);
assert_eq!(
tracker.running(),
0,
"early-stop should drain spawned read_version tasks before returning"
);
assert_eq!(diagnostics.total_responses(), 5);
assert_eq!(parts_metadata.iter().filter(|fi| fi.name == object).count(), 5);
assert!(errs.iter().all(Option::is_none));
},
)
.await;
drop(dirs);
}
#[tokio::test]
async fn data_read_early_stop_verifies_legacy_inline_checksum_payload() {
let bucket = "legacy-inline-data-get-fanout-bucket";
@@ -6328,133 +6067,11 @@ mod tests {
.clone();
assert!(
data_read_early_stop_inline_body_miss_reason(bucket, object, &candidate, &parts_metadata, &disks)
.await
.is_none(),
data_read_early_stop_inline_body_verified(bucket, object, &candidate, &parts_metadata, &disks).await,
"legacy inline metadata must use the legacy bitrot shard sizing and checksum algorithm"
);
}
#[tokio::test]
async fn data_read_early_stop_reports_inline_miss_reasons() {
let bucket = "inline-data-get-miss-reason-bucket";
let object = "inline-data-get-miss-reason-object";
let payload = b"verified inline payload";
let (_dirs, disks) = call_counter_local_disks(bucket, 4).await;
let files = inline_metadata_fanout_fileinfos_with_mode(bucket, object, payload, false).await;
let distribution = files
.first()
.map(|file| file.erasure.distribution.clone())
.expect("fixture should include metadata");
let order = bounded_metadata_fanout_order(bucket, object, 4, 2);
let mut parts_metadata = vec![FileInfo::default(); 4];
for disk_index in order.into_iter().take(3) {
let block_index = distribution
.get(disk_index)
.copied()
.expect("fixture distribution should cover every disk");
parts_metadata[disk_index] = files
.get(block_index.checked_sub(1).expect("erasure block indexes are one-based"))
.expect("fixture should include every distributed shard")
.clone();
}
let candidate = parts_metadata
.iter()
.find(|file| file.name == object)
.expect("fixture should include observed metadata")
.clone();
let data_disk = distribution
.iter()
.position(|block_index| *block_index == 1)
.expect("fixture distribution should include first data shard");
assert_eq!(
data_read_early_stop_inline_body_miss_reason(bucket, object, &candidate, &parts_metadata, &disks).await,
None
);
let mut not_inline = candidate.clone();
rustfs_utils::http::remove_str(&mut not_inline.metadata, rustfs_utils::http::SUFFIX_INLINE_DATA);
assert_eq!(
data_read_early_stop_inline_body_miss_reason(bucket, object, &not_inline, &parts_metadata, &disks).await,
Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_NOT_INLINE)
);
let mut remote = candidate.clone();
remote.transition_status = TRANSITION_COMPLETE.to_string();
assert_eq!(
data_read_early_stop_inline_body_miss_reason(bucket, object, &remote, &parts_metadata, &disks).await,
Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_REMOTE)
);
let mut transformed = candidate.clone();
rustfs_utils::http::insert_str(&mut transformed.metadata, rustfs_utils::http::SUFFIX_COMPRESSION, "zstd".to_string());
assert_eq!(
data_read_early_stop_inline_body_miss_reason(bucket, object, &transformed, &parts_metadata, &disks).await,
Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_TRANSFORMED)
);
let mut deleted = candidate.clone();
deleted.deleted = true;
assert_eq!(
data_read_early_stop_inline_body_miss_reason(bucket, object, &deleted, &parts_metadata, &disks).await,
Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_DELETED)
);
let mut zero_size = candidate.clone();
zero_size.size = 0;
assert_eq!(
data_read_early_stop_inline_body_miss_reason(bucket, object, &zero_size, &parts_metadata, &disks).await,
Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_SIZE)
);
let mut multipart = candidate.clone();
multipart.parts.push(multipart.parts[0].clone());
assert_eq!(
data_read_early_stop_inline_body_miss_reason(bucket, object, &multipart, &parts_metadata, &disks).await,
Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_PART_SHAPE)
);
let mut invalid_geometry = candidate.clone();
invalid_geometry.erasure.data_blocks = 0;
assert_eq!(
data_read_early_stop_inline_body_miss_reason(bucket, object, &invalid_geometry, &parts_metadata, &disks).await,
Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_GEOMETRY)
);
let mut missing_shard = parts_metadata.clone();
missing_shard[data_disk] = FileInfo::default();
assert_eq!(
data_read_early_stop_inline_body_miss_reason(bucket, object, &candidate, &missing_shard, &disks).await,
Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_MISSING_SHARD)
);
let mut missing_payload = parts_metadata.clone();
missing_payload[data_disk].data = None;
assert_eq!(
data_read_early_stop_inline_body_miss_reason(bucket, object, &candidate, &missing_payload, &disks).await,
Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_MISSING_PAYLOAD)
);
let mut identity_mismatch = parts_metadata.clone();
identity_mismatch[data_disk].version_id = Some(Uuid::new_v4());
assert_eq!(
data_read_early_stop_inline_body_miss_reason(bucket, object, &candidate, &identity_mismatch, &disks).await,
Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_IDENTITY_MISMATCH)
);
let mut corrupt = parts_metadata.clone();
if let Some(data) = corrupt[data_disk].data.as_mut() {
let mut corrupt_data = data.to_vec();
corrupt_data[0] ^= 0x01;
*data = Bytes::from(corrupt_data);
}
assert_eq!(
data_read_early_stop_inline_body_miss_reason(bucket, object, &candidate, &corrupt, &disks).await,
Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_BODY_VERIFY)
);
}
#[test]
#[serial_test::serial]
fn metadata_fanout_lifecycle_records_real_early_stop_abort() {
@@ -6544,7 +6161,7 @@ mod tests {
&[
("path", GET_OBJECT_PATH_INTERNAL_META),
("decision", "miss"),
("reason", GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_BODY_VERIFY),
("reason", GET_METADATA_EARLY_STOP_REASON_INSUFFICIENT_QUORUM),
],
),
1,
@@ -6556,7 +6173,7 @@ mod tests {
&[
("path", GET_OBJECT_PATH_LEGACY_DUPLEX),
("decision", "miss"),
("reason", GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_BODY_VERIFY),
("reason", GET_METADATA_EARLY_STOP_REASON_INSUFFICIENT_QUORUM),
],
),
0,
@@ -7151,7 +6768,7 @@ mod tests {
}
#[tokio::test]
async fn bounded_metadata_early_stop_defaults_keep_non_inline_data_get_full_fanout() {
async fn bounded_metadata_early_stop_defaults_keep_data_get_full_fanout() {
const DISKS: usize = 4;
let bucket = "bounded-data-get-default-bucket";
let object = "bounded-data-get-default-object";
@@ -7174,7 +6791,7 @@ mod tests {
assert_eq!(
calls.total(disk_call_counters::KIND_READ_VERSION),
DISKS as u64,
"default non-inline GET data-read metadata must keep full fanout for read-failure tolerance"
"default GET data-read metadata must keep full fanout for read-failure tolerance"
);
assert_eq!(diagnostics.total_responses(), DISKS);
assert_eq!(parts_metadata.iter().filter(|fi| fi.name == object).count(), DISKS);
-28
View File
@@ -42,20 +42,12 @@ impl<'a> SetDisksCtx<'a> {
}
/// The borrowed core, for state not yet fronted by a typed accessor.
#[allow(
dead_code,
reason = "SetDisks split seam (backlog#815) with no caller in this port (backlog#1823)"
)]
pub(crate) fn core(&self) -> &'a SetDisks {
self.core
}
// --- Immutable topology / config (fixed after construction) ---
#[allow(
dead_code,
reason = "SetDisks split seam (backlog#815) with no caller in this port (backlog#1823)"
)]
pub(crate) fn set_index(&self) -> usize {
self.core.set_index
}
@@ -64,26 +56,14 @@ impl<'a> SetDisksCtx<'a> {
self.core.pool_index
}
#[allow(
dead_code,
reason = "SetDisks split seam (backlog#815) with no caller in this port (backlog#1823)"
)]
pub(crate) fn set_drive_count(&self) -> usize {
self.core.set_drive_count
}
#[allow(
dead_code,
reason = "SetDisks split seam (backlog#815) with no caller in this port (backlog#1823)"
)]
pub(crate) fn default_parity_count(&self) -> usize {
self.core.default_parity_count
}
#[allow(
dead_code,
reason = "SetDisks split seam (backlog#815) with no caller in this port (backlog#1823)"
)]
pub(crate) fn set_endpoints(&self) -> &'a [Endpoint] {
&self.core.set_endpoints
}
@@ -92,10 +72,6 @@ impl<'a> SetDisksCtx<'a> {
&self.core.format
}
#[allow(
dead_code,
reason = "SetDisks split seam (backlog#815) with no caller in this port (backlog#1823)"
)]
pub(crate) fn locker_owner(&self) -> &'a str {
&self.core.locker_owner
}
@@ -108,10 +84,6 @@ impl<'a> SetDisksCtx<'a> {
// --- Locker trio ---
#[allow(
dead_code,
reason = "SetDisks split seam (backlog#815) with no caller in this port (backlog#1823)"
)]
pub(crate) fn lockers(&self) -> &'a [Arc<dyn LockClient>] {
&self.core.lockers
}
+45 -78
View File
@@ -39,6 +39,7 @@
//! - `metadata.rs`, `replication.rs`, `shard_source.rs` — supporting helpers.
// #730: SetDisks still hosts staged read/heal/write migration helpers.
#![allow(dead_code)]
#![allow(unused_imports)]
#![allow(unused_variables)]
@@ -58,10 +59,7 @@ use crate::client::{object_api_utils::get_raw_etag, transition_api::ReaderImpl};
use crate::cluster::rpc::heal_bucket_local_on_disks;
use crate::data_usage::record_compression_total_memory;
use crate::diagnostics::get::{
GET_CODEC_STREAMING_OBJECT_CLASS_PLAIN_SINGLE_PART, GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_GEOMETRY,
GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_IDENTITY_MISMATCH,
GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_MISSING_PAYLOAD,
GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_MISSING_SHARD, GET_OBJECT_PATH_BODY_CACHE, GET_OBJECT_PATH_CODEC_STREAMING,
GET_CODEC_STREAMING_OBJECT_CLASS_PLAIN_SINGLE_PART, GET_OBJECT_PATH_BODY_CACHE, GET_OBJECT_PATH_CODEC_STREAMING,
GET_OBJECT_PATH_CODEC_STREAMING_LEGACY_ENGINE, GET_OBJECT_PATH_CODEC_STREAMING_RUSTFS_ENGINE, GET_OBJECT_PATH_DIRECT_MEMORY,
GET_OBJECT_PATH_EMPTY, GET_OBJECT_PATH_INLINE_DIRECT, GET_OBJECT_PATH_INTERNAL_META, GET_OBJECT_PATH_LEGACY_DUPLEX,
GET_OBJECT_PATH_REMOTE_TRANSITION, GET_OBJECT_PATH_SET_DISK, GET_STAGE_DECODE, GET_STAGE_EMIT, GET_STAGE_INLINE_PREPARE,
@@ -102,7 +100,9 @@ use crate::storage_api_contracts::{
};
use crate::store::utils::is_reserved_or_invalid_bucket;
use crate::{
bucket::lifecycle::bucket_lifecycle_ops::{LifecycleOps, get_transitioned_object_reader_with_tier_manager, put_restore_opts},
bucket::lifecycle::bucket_lifecycle_ops::{
LifecycleOps, gen_transition_objname, get_transitioned_object_reader_with_tier_manager, put_restore_opts,
},
cache_value::metacache_set::{ListPathRawOptions, list_path_raw},
config::storageclass,
disk::{
@@ -621,9 +621,7 @@ fn adaptive_duplex_buffer_size(object_size: i64) -> usize {
// Each flag has a corresponding `*_ROLLOUT_PCT` for percentage-based gradual rollout.
// ============================================================================
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
const DISK_ONLINE_TIMEOUT: Duration = Duration::from_secs(1);
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
const DISK_HEALTH_CACHE_TTL: Duration = Duration::from_millis(750);
const GET_OBJECT_METADATA_CACHE_TTL: Duration = Duration::from_secs(2); // Increased from 250ms to 2s
const DEFAULT_GET_OBJECT_METADATA_CACHE_MAX_ENTRIES: usize = 4096; // Increased from 1024 to 4096
@@ -691,28 +689,19 @@ const DEFAULT_RUSTFS_GET_SMALL_OBJECT_DIRECT_MEMORY_THRESHOLD: usize = 128 * 102
const ENV_RUSTFS_GET_METADATA_EARLY_STOP_ENABLE: &str = "RUSTFS_GET_METADATA_EARLY_STOP_ENABLE";
// Enabled by default (backlog#872): the early-stop path only engages for
// requests `should_allow_metadata_early_stop` classifies as safe (latest-version
// reads by default, without version_id / healing / free-version needs) and still
// requires a full read-quorum agreement before stopping. Data-read requests add
// a separate inline-shard verifier before cancelling the remaining fanout. Set
// metadata-only reads by default, without version_id / healing / free-version
// needs) and still requires a full read-quorum agreement before stopping. Set
// the env var to `false` to fall back to full-wait metadata fanout.
const DEFAULT_RUSTFS_GET_METADATA_EARLY_STOP_ENABLE: bool = true;
#[allow(
dead_code,
reason = "percentage-rollout facet of the metadata early-stop switch; its predicate has no caller while the sibling enable flag is live (backlog#1823)"
)]
const ENV_RUSTFS_GET_METADATA_EARLY_STOP_ROLLOUT_PCT: &str = "RUSTFS_GET_METADATA_EARLY_STOP_ROLLOUT_PCT";
#[allow(
dead_code,
reason = "percentage-rollout facet of the metadata early-stop switch; its predicate has no caller while the sibling enable flag is live (backlog#1823)"
)]
const DEFAULT_RUSTFS_GET_METADATA_EARLY_STOP_ROLLOUT_PCT: u32 = 100;
const ENV_RUSTFS_GET_METADATA_VERSION_EARLY_STOP_ENABLE: &str = "RUSTFS_GET_METADATA_VERSION_EARLY_STOP_ENABLE";
const DEFAULT_RUSTFS_GET_METADATA_VERSION_EARLY_STOP_ENABLE: bool = false;
const ENV_RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE: &str = "RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE";
const DEFAULT_RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE: bool = true;
const DEFAULT_RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE: bool = false;
const ENV_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT: &str = "RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT";
const DEFAULT_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT: bool = false;
@@ -917,16 +906,18 @@ mod prepared_get_object_metadata_tests {
.expect("test should find an object whose initial fanout covers both data shards")
}
#[allow(
dead_code,
reason = "test fixture no assertion in this module uses today; the live namesake lives in io_primitives tests (backlog#1823)"
)]
fn bounded_spare_disk_index(bucket: &str, object: &str) -> usize {
*bounded_metadata_fanout_order(bucket, object, 4, 2)
.get(3)
.expect("4-disk test geometry should leave one bounded spare disk")
}
fn bounded_slow_initial_disk_index(bucket: &str, object: &str) -> usize {
*bounded_metadata_fanout_order(bucket, object, 4, 2)
.get(2)
.expect("4-disk test geometry should include a third initial metadata disk")
}
#[tokio::test]
async fn prepared_metadata_is_consumed_exactly_once() {
let snapshot = GetObjectFileInfo::owned(FileInfo::default(), Vec::new(), Vec::new());
@@ -1045,7 +1036,7 @@ mod prepared_get_object_metadata_tests {
#[test]
#[serial_test::serial(body_cache_hook)]
fn inline_data_read_early_stop_defaults_return_exact_body() {
fn inline_data_read_early_stop_reader_returns_exact_body() {
let runtime = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
@@ -1077,14 +1068,14 @@ mod prepared_get_object_metadata_tests {
temp_env::async_with_vars(
[
("RUSTFS_GET_METADATA_EARLY_STOP_ENABLE", None::<&str>),
("RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE", None::<&str>),
("RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT", None::<&str>),
("RUSTFS_GET_METADATA_EARLY_STOP_ENABLE", Some("true")),
("RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE", Some("true")),
("RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT", Some("true")),
],
async {
let slow_parity_disk = bounded_spare_disk_index(bucket, &object);
let slow_initial_disk = bounded_slow_initial_disk_index(bucket, &object);
let barrier =
rename_fanout_barrier::arm(&object, slow_parity_disk, rename_fanout_barrier::PHASE_READ_VERSION);
rename_fanout_barrier::arm(&object, slow_initial_disk, rename_fanout_barrier::PHASE_READ_VERSION);
let calls = disk_call_counters::observe(&object);
let set_disks_for_read = Arc::clone(&set_disks);
let opts_for_read = opts.clone();
@@ -1097,10 +1088,10 @@ mod prepared_get_object_metadata_tests {
tokio::time::timeout(READ_VERSION_BARRIER_GUARD, barrier.wait_until_paused())
.await
.expect("default inline GET should pause a slow parity metadata read");
.expect("bounded inline GET should pause a slow initial metadata read");
let mut reader = tokio::time::timeout(READ_VERSION_BARRIER_GUARD, &mut open_reader)
.await
.expect("default production inline GET should return before the paused parity metadata response")
.expect("production inline GET should return before the paused metadata response")
.expect("inline GET reader task should not panic")
.expect("inline GET reader should open");
let object_size = reader.object_info.size;
@@ -1121,14 +1112,14 @@ mod prepared_get_object_metadata_tests {
assert_eq!(object_size, payload.len() as i64);
assert_eq!(restored, payload);
assert_eq!(calls_total, 4, "default production GET should eagerly schedule the full metadata fanout");
assert_eq!(calls_total, 4, "bounded production GET should schedule the initial quorum plus one spare");
assert_eq!(
recorder.histogram_values(
"rustfs_io_get_object_metadata_fanout_scheduled",
&[("path", GET_OBJECT_PATH_LEGACY_DUPLEX)]
),
vec![4.0],
"default production GET should record all scheduled metadata tasks"
"bounded production GET should record all scheduled metadata tasks"
);
assert_eq!(
recorder.histogram_values(
@@ -1136,7 +1127,7 @@ mod prepared_get_object_metadata_tests {
&[("path", GET_OBJECT_PATH_LEGACY_DUPLEX)]
),
vec![3.0],
"default production GET should record only observed metadata responses as completed"
"bounded production GET should record only observed metadata responses as completed"
);
assert_eq!(
recorder.histogram_values(
@@ -1144,7 +1135,7 @@ mod prepared_get_object_metadata_tests {
&[("path", GET_OBJECT_PATH_LEGACY_DUPLEX)]
),
vec![1.0],
"default production GET should record the aborted slow parity metadata task"
"bounded production GET should record the aborted slow metadata task"
);
}
@@ -1291,9 +1282,9 @@ mod prepared_get_object_metadata_tests {
temp_env::async_with_vars(
[
("RUSTFS_GET_METADATA_EARLY_STOP_ENABLE", None::<&str>),
("RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE", None::<&str>),
("RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT", None::<&str>),
("RUSTFS_GET_METADATA_EARLY_STOP_ENABLE", Some("true")),
("RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE", Some("true")),
("RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT", Some("true")),
],
async {
let calls = disk_call_counters::observe(&object);
@@ -1720,10 +1711,6 @@ fn is_multipart_reader_setup_prefetch_enabled() -> bool {
}
}
#[allow(
dead_code,
reason = "percentage-rollout facet of the metadata early-stop switch; its predicate has no caller while the sibling enable flag is live (backlog#1823)"
)]
fn get_metadata_early_stop_rollout_pct() -> u32 {
static CACHED: OnceLock<u32> = OnceLock::new();
*CACHED.get_or_init(|| {
@@ -1763,10 +1750,6 @@ fn should_use_codec_streaming(config: GetCodecStreamingConfig, bucket: &str, obj
}
/// Should this specific request use metadata early-stop?
#[allow(
dead_code,
reason = "percentage-rollout facet of the metadata early-stop switch; its predicate has no caller while the sibling enable flag is live (backlog#1823)"
)]
pub fn should_use_metadata_early_stop(bucket: &str, object: &str) -> bool {
let base = is_get_metadata_early_stop_enabled();
let pct = get_metadata_early_stop_rollout_pct();
@@ -2200,7 +2183,6 @@ fn classify_get_codec_streaming_object_class(
GetCodecStreamingObjectClass::PlainSinglePart
}
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
fn is_get_small_object_direct_memory_eligible_with_threshold(
range: &Option<HTTPRangeSpec>,
object_info: &ObjectInfo,
@@ -2806,7 +2788,6 @@ pub struct SetDisks {
/// Stable namespace shared by every object lock created for this set.
set_lock_namespace: Arc<str>,
pub format: FormatV3,
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
disk_health_cache: Arc<RwLock<Vec<Option<DiskHealthEntry>>>>,
get_object_metadata_cache: moka::future::Cache<GetObjectMetadataCacheKey, Arc<GetObjectMetadataCacheEntry>>,
get_object_metadata_cache_hash_builder: std::collections::hash_map::RandomState,
@@ -3082,13 +3063,11 @@ struct GetObjectMetadataCacheEntry {
#[derive(Clone, Debug)]
struct DiskHealthEntry {
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
last_check: Instant,
online: bool,
}
impl DiskHealthEntry {
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
fn cached_value(&self) -> Option<bool> {
if self.last_check.elapsed() <= DISK_HEALTH_CACHE_TTL {
Some(self.online)
@@ -3682,7 +3661,6 @@ fn multipart_put_large_batch_min_size_bytes() -> usize {
})
}
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
fn classify_small_write_path(is_inline_buffer: bool, object_size: i64, block_size: usize) -> SmallWritePath {
if should_use_inline_small_fast_path(is_inline_buffer, object_size, block_size) {
SmallWritePath::Inline
@@ -3885,20 +3863,11 @@ fn inline_erasure_shard_file_offset(
}
fn collect_inline_data_shard_fileinfos_by_index<'a>(
parts_metadata: &'a [FileInfo],
fi: &FileInfo,
data_shards: usize,
disk_is_online: impl FnMut(usize) -> bool,
) -> Option<Vec<&'a FileInfo>> {
collect_inline_data_shard_fileinfos_by_index_or_reason(parts_metadata, fi, data_shards, disk_is_online).ok()
}
fn collect_inline_data_shard_fileinfos_by_index_or_reason<'a>(
parts_metadata: &'a [FileInfo],
fi: &FileInfo,
data_shards: usize,
mut disk_is_online: impl FnMut(usize) -> bool,
) -> std::result::Result<Vec<&'a FileInfo>, &'static str> {
) -> Option<Vec<&'a FileInfo>> {
let distribution = &fi.erasure.distribution;
let mut data_files = vec![None; data_shards];
@@ -3906,35 +3875,27 @@ fn collect_inline_data_shard_fileinfos_by_index_or_reason<'a>(
if !disk_is_online(disk_index) {
continue;
}
let Some(&block_index) = distribution.get(disk_index) else {
return Err(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_GEOMETRY);
};
let block_index = *distribution.get(disk_index)?;
if block_index == 0 || block_index > data_shards {
continue;
}
if file_info.name.is_empty() {
return Err(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_MISSING_SHARD);
}
if file_info.erasure.index != block_index {
return Err(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_IDENTITY_MISMATCH);
continue;
}
if !file_info.has_valid_erasure_geometry() {
return Err(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_GEOMETRY);
continue;
}
if !core::io_primitives::metadata_early_stop_candidate_matches(file_info, fi) {
return Err(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_IDENTITY_MISMATCH);
continue;
}
if file_info.data.as_ref().is_none_or(|data| data.is_empty()) {
return Err(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_MISSING_PAYLOAD);
continue;
}
data_files[block_index - 1] = Some(file_info);
}
data_files
.into_iter()
.collect::<Option<Vec<_>>>()
.ok_or(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_MISSING_SHARD)
data_files.into_iter().collect()
}
impl SetDisks {
@@ -4261,7 +4222,6 @@ fn check_object_lock_retention_update(bucket: &str, object: &str, obj_info: &Obj
///
/// Fail closed: when bucket metadata cannot be resolved the check stays on, so
/// object-lock protection is never skipped because of a metadata lookup miss.
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
pub(crate) fn object_lock_delete_check_required(bucket_meta: Option<&crate::bucket::metadata::BucketMetadata>) -> bool {
bucket_meta.is_none_or(|meta| meta.object_locking())
}
@@ -4537,6 +4497,15 @@ impl Hash for ObjProps {
}
}
#[derive(Default, Clone, Debug)]
pub struct HealEntryResult {
pub bytes: usize,
pub success: bool,
pub skipped: bool,
pub entry_done: bool,
pub name: String,
}
fn is_object_dangling(
meta_arr: &[FileInfo],
errs: &[Option<DiskError>],
@@ -5313,7 +5282,6 @@ pub fn is_valid_storage_class(storage_class: &str) -> bool {
}
/// Returns true if the storage class is a cold storage tier that requires special handling
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
pub fn is_cold_storage_class(storage_class: &str) -> bool {
matches!(
storage_class,
@@ -5322,7 +5290,6 @@ pub fn is_cold_storage_class(storage_class: &str) -> bool {
}
/// Returns true if the storage class is an infrequent access tier
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
pub fn is_infrequent_access_class(storage_class: &str) -> bool {
matches!(
storage_class,
-4
View File
@@ -1716,10 +1716,6 @@ impl SetDisks {
Ok((result, None))
}
#[allow(
dead_code,
reason = "lock-taking wrapper over the live heal_object_dir_locked; only comments reference it (backlog#1823)"
)]
#[tracing::instrument(level = "trace", skip(self), fields(bucket = %bucket, object = %object))]
pub(in crate::set_disk) async fn heal_object_dir(
&self,
@@ -66,7 +66,6 @@ impl crate::storage_api_contracts::namespace::NamespaceLocking for SetDisks {
}
impl SetDisks {
#[allow(dead_code, reason = "lock diagnostics formatter with no caller in this port (backlog#1823)")]
pub(in crate::set_disk) fn format_lock_error(&self, bucket: &str, object: &str, mode: &str, err: &LockResult) -> String {
match err {
LockResult::Timeout => {
@@ -80,7 +79,6 @@ impl SetDisks {
}
}
#[allow(dead_code, reason = "lock diagnostics formatter with no caller in this port (backlog#1823)")]
pub(in crate::set_disk) fn format_lock_error_from_error(
&self,
bucket: &str,
@@ -145,7 +143,6 @@ impl SetDisks {
disks
}
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
pub(in crate::set_disk) async fn get_online_disks(&self) -> Vec<Option<DiskStore>> {
let snapshot = self.drive_membership_snapshot().await;
let mut disks = snapshot.strict_online_candidates().into_iter().map(Some).collect::<Vec<_>>();
@@ -156,10 +153,6 @@ impl SetDisks {
disks
}
#[allow(
dead_code,
reason = "local-only sibling of the test-covered get_online_disks; no caller in this port (backlog#1823)"
)]
pub(in crate::set_disk) async fn get_online_local_disks(&self) -> Vec<Option<DiskStore>> {
let snapshot = self.drive_membership_snapshot().await;
let mut disks = snapshot
@@ -439,10 +432,6 @@ impl SetDisks {
Ok((disk, fm))
}
#[allow(
dead_code,
reason = "MinIO-parity healing-disk accessor with no caller in this port (backlog#1823)"
)]
pub(in crate::set_disk) async fn get_online_disk_with_healing(
&self,
incl_healing: bool,
@@ -451,10 +440,6 @@ impl SetDisks {
Ok((new_disks, healing > 0))
}
#[allow(
dead_code,
reason = "reached only from get_online_disk_with_healing, itself uncalled in this port (backlog#1823)"
)]
pub(in crate::set_disk) async fn get_online_disk_with_healing_and_info(
&self,
incl_healing: bool,
@@ -415,7 +415,6 @@ fn reduce_quorum_part_numbers(object_parts: Vec<Vec<String>>, read_quorum: usize
/// never returned, but flips `is_truncated` to `true` and yields a
/// `next_upload_id_marker` pointing at the last returned upload so the caller can
/// resume paging.
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
fn paginate_upload_page(remaining: &[MultipartInfo], max_uploads: usize) -> (Vec<MultipartInfo>, bool, Option<String>) {
let is_truncated = remaining.len() > max_uploads;
let page: Vec<MultipartInfo> = remaining.iter().take(max_uploads).cloned().collect();
@@ -558,7 +557,6 @@ impl SetDisks {
}
#[tracing::instrument(level = "debug", skip(self))]
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
pub(super) async fn check_upload_id_exists(
&self,
bucket: &str,
+11 -329
View File
@@ -3507,10 +3507,6 @@ struct TransitionUploadedSaveProbeState {
}
#[cfg(test)]
#[allow(
dead_code,
reason = "installed by set_disk tests behind `--features test-util` (backlog#1823)"
)]
struct TransitionUploadedSaveProbe {
state: Arc<TransitionUploadedSaveProbeState>,
}
@@ -3521,10 +3517,6 @@ static TRANSITION_UPLOADED_SAVE_PROBE: std::sync::OnceLock<std::sync::Mutex<Opti
#[cfg(test)]
impl TransitionUploadedSaveProbe {
#[allow(
dead_code,
reason = "installed by set_disk tests behind `--features test-util` (backlog#1823)"
)]
fn install(bucket: &str, object: &str) -> Self {
let state = Arc::new(TransitionUploadedSaveProbeState {
bucket: bucket.to_string(),
@@ -3541,10 +3533,6 @@ impl TransitionUploadedSaveProbe {
Self { state }
}
#[allow(
dead_code,
reason = "installed by set_disk tests behind `--features test-util` (backlog#1823)"
)]
fn attempts(&self) -> usize {
self.state.attempts.load(std::sync::atomic::Ordering::Acquire)
}
@@ -3750,10 +3738,6 @@ struct TransitionCommitBarrierState {
}
#[cfg(test)]
#[allow(
dead_code,
reason = "installed by set_disk tests behind `--features test-util` (backlog#1823)"
)]
struct TransitionCommitBarrier {
state: Arc<TransitionCommitBarrierState>,
}
@@ -3764,26 +3748,14 @@ static TRANSITION_COMMIT_BARRIER: std::sync::OnceLock<std::sync::Mutex<Option<Ar
#[cfg(test)]
impl TransitionCommitBarrier {
#[allow(
dead_code,
reason = "installed by set_disk tests behind `--features test-util` (backlog#1823)"
)]
fn install_before_lock_lost_check(bucket: &str, object: &str) -> Self {
Self::install_at(bucket, object, TransitionCommitPause::BeforeLockLost)
}
#[allow(
dead_code,
reason = "installed by set_disk tests behind `--features test-util` (backlog#1823)"
)]
fn install(bucket: &str, object: &str) -> Self {
Self::install_at(bucket, object, TransitionCommitPause::BeforeLeaseValidation)
}
#[allow(
dead_code,
reason = "installed by set_disk tests behind `--features test-util` (backlog#1823)"
)]
fn install_after_lease_check(bucket: &str, object: &str) -> Self {
Self::install_at(bucket, object, TransitionCommitPause::AfterLeaseValidation)
}
@@ -3806,20 +3778,12 @@ impl TransitionCommitBarrier {
Self { state }
}
#[allow(
dead_code,
reason = "installed by set_disk tests behind `--features test-util` (backlog#1823)"
)]
async fn wait_until_paused(&self) {
tokio::time::timeout(Duration::from_secs(30), self.state.arrived.notified())
.await
.expect("transition should reach the deterministic commit barrier");
}
#[allow(
dead_code,
reason = "installed by set_disk tests behind `--features test-util` (backlog#1823)"
)]
fn release(&self) {
self.state.release.notify_one();
}
@@ -7541,7 +7505,7 @@ mod get_object_downstream_close_accounting_tests {
use super::hermetic_set_disks_support::hermetic_set_disks;
use super::*;
use crate::diagnostics::get::{
GET_METADATA_EARLY_STOP_REASON_NOT_FOUND, GET_OBJECT_PATH_INTERNAL_META, GET_STAGE_DECODE, GET_STAGE_EMIT,
GET_METADATA_EARLY_STOP_REASON_UNSAFE_REQUEST, GET_OBJECT_PATH_INTERNAL_META, GET_STAGE_DECODE, GET_STAGE_EMIT,
GetObjectFailureReason,
};
use crate::disk::RUSTFS_META_BUCKET;
@@ -7673,8 +7637,8 @@ mod get_object_downstream_close_accounting_tests {
legacy_completed,
internal_cancelled,
legacy_cancelled,
internal_not_found_miss,
legacy_not_found_miss,
internal_unsafe_miss,
legacy_unsafe_miss,
internal_saved,
legacy_saved,
) = metrics::with_local_recorder(&recorder, || {
@@ -7750,7 +7714,7 @@ mod get_object_downstream_close_accounting_tests {
&[
("path", GET_OBJECT_PATH_INTERNAL_META),
("decision", "miss"),
("reason", GET_METADATA_EARLY_STOP_REASON_NOT_FOUND),
("reason", GET_METADATA_EARLY_STOP_REASON_UNSAFE_REQUEST),
],
),
recorder.counter_value(
@@ -7758,7 +7722,7 @@ mod get_object_downstream_close_accounting_tests {
&[
("path", GET_OBJECT_PATH_LEGACY_DUPLEX),
("decision", "miss"),
("reason", GET_METADATA_EARLY_STOP_REASON_NOT_FOUND),
("reason", GET_METADATA_EARLY_STOP_REASON_UNSAFE_REQUEST),
],
),
recorder.histogram_values(
@@ -7809,21 +7773,21 @@ mod get_object_downstream_close_accounting_tests {
"internal metadata lifecycle cancelled count must not leak into legacy_duplex"
);
assert_eq!(
internal_not_found_miss, 1,
"internal metadata not-found early-stop miss must retain its path label"
internal_unsafe_miss, 1,
"internal metadata unsafe early-stop miss must retain its path label"
);
assert_eq!(
legacy_not_found_miss, 0,
"internal metadata not-found early-stop miss must not leak into legacy_duplex"
legacy_unsafe_miss, 0,
"internal metadata unsafe early-stop miss must not leak into legacy_duplex"
);
assert_eq!(
internal_saved,
vec![0.0],
"internal metadata not-found miss must record zero saved responses on internal_meta"
"internal metadata unsafe miss must record zero saved responses on internal_meta"
);
assert!(
legacy_saved.is_empty(),
"internal metadata not-found miss saved responses must not leak into legacy_duplex"
"internal metadata unsafe miss saved responses must not leak into legacy_duplex"
);
}
}
@@ -10206,288 +10170,6 @@ mod transition_upload_integrity_tests {
assert!(backend.contains(remote_object).await, "committed remote object should remain available");
}
/// Compresses `plaintext` with the codec the PUT path uses, so the stored
/// bytes round-trip through the read path's decompressor.
async fn compress_for_storage(plaintext: &[u8]) -> Vec<u8> {
let mut reader = crate::io_support::rio::compression_reader(
Cursor::new(plaintext.to_vec()),
rustfs_utils::CompressionAlgorithm::default(),
false,
);
let mut compressed = Vec::new();
reader.read_to_end(&mut compressed).await.expect("plaintext should compress");
assert!(compressed.len() < plaintext.len(), "test payload must actually compress");
compressed
}
/// Writes a genuinely compressed object: stored data is `compressed`, and the
/// metadata marks it compressed with the plaintext length as its actual size,
/// exactly as the app-layer compress path records it.
async fn write_compressed_source(
set_disks: &Arc<SetDisks>,
disk_stores: &[DiskStore],
bucket: &str,
object: &str,
plaintext: &[u8],
compressed: &[u8],
) -> ObjectInfo {
for disk in disk_stores {
disk.make_volume(bucket).await.expect("bucket volume should be created");
}
let mut user_defined = HashMap::new();
rustfs_utils::http::insert_str(
&mut user_defined,
rustfs_utils::http::SUFFIX_COMPRESSION,
crate::io_support::rio::compression_metadata_value(rustfs_utils::CompressionAlgorithm::default()),
);
rustfs_utils::http::insert_str(&mut user_defined, rustfs_utils::http::SUFFIX_ACTUAL_SIZE, plaintext.len().to_string());
let stream = crate::io_support::rio::HashReader::from_stream(
Cursor::new(compressed.to_vec()),
compressed.len() as i64,
plaintext.len() as i64,
None,
None,
false,
)
.expect("hash reader over compressed bytes");
let mut reader = PutObjReader::new(stream);
set_disks
.put_object(
bucket,
object,
&mut reader,
&ObjectOptions {
no_lock: true,
user_defined,
..Default::default()
},
)
.await
.expect("compressed object should be written")
}
async fn read_transitioned(
set_disks: &Arc<SetDisks>,
bucket: &str,
object: &str,
range: Option<HTTPRangeSpec>,
opts: &ObjectOptions,
) -> (Vec<u8>, i64) {
let mut reader = set_disks
.get_object_reader(bucket, object, range, HeaderMap::new(), opts)
.await
.expect("transitioned object reader should open");
let published_size = reader.object_info.size;
let mut body = Vec::new();
reader
.stream
.read_to_end(&mut body)
.await
.expect("transitioned body should drain");
(body, published_size)
}
/// Transition uploads the object's STORED bytes, so a tiered read has to
/// apply the same transform an erasure read would. #6107 routed this path
/// through `ReadPlan` to stop serving an encrypted object's ciphertext;
/// compression rides the same plan, and nothing pinned it (backlog#1851).
/// Without the transform this GET returns the compressed bytes under the
/// compressed size — silent corruption for every client of a compressed
/// object that ILM has moved to a warm tier.
#[tokio::test]
#[serial_test::serial]
async fn transitioned_compressed_object_get_returns_plaintext() {
let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
let bucket = "transitioned-compressed-get-bucket";
let object = "object.txt";
let plaintext = b"transitioned compressed objects must decompress on read ".repeat(20_000);
let compressed = compress_for_storage(&plaintext).await;
let original = write_compressed_source(&set_disks, &disk_stores, bucket, object, &plaintext, &compressed).await;
let opts = ObjectOptions {
no_lock: true,
..Default::default()
};
let (local_body, local_size) = read_transitioned(&set_disks, bucket, object, None, &opts).await;
assert_eq!(local_body, plaintext, "control: the pre-transition read must decompress");
assert_eq!(
local_size,
plaintext.len() as i64,
"control: the pre-transition read publishes the plaintext size"
);
let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase();
let backend = register_mock_tier(&runtime_sources::global_tier_config_mgr(), &tier_name).await;
set_disks
.transition_object(bucket, object, &transition_options(&original, tier_name))
.await
.expect("transition should commit");
let put_versions = backend.put_versions().await;
assert_eq!(put_versions.len(), 1, "transition should upload one remote candidate");
let remote_bytes = backend
.bytes(&put_versions[0].0)
.await
.expect("remote candidate should be stored");
assert_eq!(
remote_bytes, compressed,
"transition uploads the stored representation; the read side is what has to decode it"
);
let (body, published_size) = read_transitioned(&set_disks, bucket, object, None, &opts).await;
assert_eq!(body, plaintext, "a tiered read must return the object's content, not its stored bytes");
assert_eq!(
published_size,
plaintext.len() as i64,
"a tiered read must publish the plaintext size, not the compressed one"
);
}
/// A ranged tiered read is expressed in plaintext coordinates, so the plan
/// has to translate it into the remote copy's compressed extent and skip
/// into the decompressed stream — the same translation the erasure path does.
#[tokio::test]
#[serial_test::serial]
async fn transitioned_compressed_object_range_get_returns_plaintext_slice() {
let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
let bucket = "transitioned-compressed-range-bucket";
let object = "object.txt";
let plaintext = b"ranged reads of transitioned compressed objects must land in plaintext ".repeat(20_000);
let compressed = compress_for_storage(&plaintext).await;
let original = write_compressed_source(&set_disks, &disk_stores, bucket, object, &plaintext, &compressed).await;
let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase();
register_mock_tier(&runtime_sources::global_tier_config_mgr(), &tier_name).await;
set_disks
.transition_object(bucket, object, &transition_options(&original, tier_name))
.await
.expect("transition should commit");
let opts = ObjectOptions {
no_lock: true,
..Default::default()
};
// Deliberately past the compressed size, so a range still measured in
// stored coordinates could not produce this slice.
let start = compressed.len() as i64 + 4096;
let end = start + 511;
let range = HTTPRangeSpec {
is_suffix_length: false,
start,
end,
};
let (body, published_size) = read_transitioned(&set_disks, bucket, object, Some(range), &opts).await;
let expected = &plaintext[start as usize..=end as usize];
assert_eq!(body, expected, "a ranged tiered read must return that plaintext slice");
assert_eq!(published_size, expected.len() as i64, "a ranged tiered read publishes the slice length");
}
/// The restore copy-back re-writes the object under its original metadata,
/// which still says "compressed". It therefore has to keep receiving the
/// STORED bytes: `restore_request_active` holds it on the plan's `Plain`
/// branch, and decompressing there would write plaintext under compressed
/// metadata.
#[tokio::test]
#[serial_test::serial]
async fn restore_read_of_transitioned_compressed_object_keeps_stored_bytes() {
let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
let bucket = "transitioned-compressed-restore-bucket";
let object = "object.txt";
let plaintext = b"restore copy-back must keep the stored representation intact ".repeat(20_000);
let compressed = compress_for_storage(&plaintext).await;
let original = write_compressed_source(&set_disks, &disk_stores, bucket, object, &plaintext, &compressed).await;
let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase();
register_mock_tier(&runtime_sources::global_tier_config_mgr(), &tier_name).await;
set_disks
.transition_object(bucket, object, &transition_options(&original, tier_name))
.await
.expect("transition should commit");
let oi = set_disks
.get_object_info(
bucket,
object,
&ObjectOptions {
no_lock: true,
..Default::default()
},
)
.await
.expect("transitioned metadata should resolve");
let restore_opts = ObjectOptions {
no_lock: true,
part_number: Some(1),
transition: TransitionOptions {
restore_request: s3s::dto::RestoreRequest {
days: Some(1),
..Default::default()
},
..Default::default()
},
..Default::default()
};
let mut reader = get_transitioned_object_reader_with_tier_manager(
bucket,
object,
&None,
&HeaderMap::new(),
&oi,
&restore_opts,
&set_disks.ctx.tier_config_mgr(),
set_disks.ctx.object_encryption_resolver(),
)
.await
.expect("restore read of the tiered copy should open");
let published_size = reader.object_info.size;
let mut body = Vec::new();
reader.stream.read_to_end(&mut body).await.expect("restore body should drain");
assert_eq!(body, compressed, "a restore read must copy the stored bytes back verbatim");
assert_eq!(
published_size,
compressed.len() as i64,
"a restore read must keep publishing the stored size"
);
}
/// Plain objects must keep streaming the remote bytes through untouched:
/// their plan is `Plain`, so the tiered read stays byte-identical.
#[tokio::test]
#[serial_test::serial]
async fn transitioned_plain_object_get_is_unchanged() {
let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
let bucket = "transitioned-plain-get-bucket";
let object = "object.bin";
let payload = b"plain transitioned objects must keep reading back byte-identical ".repeat(1024);
let original = write_source(&set_disks, &disk_stores, bucket, object, &payload).await;
let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase();
register_mock_tier(&runtime_sources::global_tier_config_mgr(), &tier_name).await;
set_disks
.transition_object(bucket, object, &transition_options(&original, tier_name))
.await
.expect("transition should commit");
let opts = ObjectOptions {
no_lock: true,
..Default::default()
};
let (body, published_size) = read_transitioned(&set_disks, bucket, object, None, &opts).await;
assert_eq!(body, payload);
assert_eq!(published_size, payload.len() as i64);
let range = HTTPRangeSpec {
is_suffix_length: false,
start: 100,
end: 611,
};
let (ranged_body, ranged_size) = read_transitioned(&set_disks, bucket, object, Some(range), &opts).await;
assert_eq!(ranged_body, &payload[100..=611]);
assert_eq!(ranged_size, payload.len() as i64, "a plain ranged read keeps publishing the object size");
}
async fn corrupt_beyond_read_quorum(
temp_dirs: &[tempfile::TempDir],
bucket: &str,
+3 -35
View File
@@ -116,7 +116,6 @@ impl SetDisks {
.then_some(GET_METADATA_CACHE_REASON_DIST_ERASURE)
}
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
async fn cached_get_object_fileinfo(&self, bucket: &str, object: &str) -> Option<Arc<GetObjectMetadataCacheEntry>> {
match self.lookup_cached_get_object_fileinfo(bucket, object).await {
MetadataCacheLookup::Hit(entry) => Some(entry),
@@ -1827,7 +1826,6 @@ fn get_object_metadata_cache_request_bypass_reason(bucket: &str, opts: &ObjectOp
.then_some(GET_METADATA_CACHE_REASON_META_BUCKET)
}
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
fn is_get_object_metadata_cache_request_eligible(bucket: &str, opts: &ObjectOptions, read_data: bool) -> bool {
get_object_metadata_cache_request_bypass_reason(bucket, opts, read_data).is_none()
}
@@ -3888,15 +3886,13 @@ mod tests {
assert!(metadata_early_stop_permitted(true, true, false, "", false, false));
// observe=false (non-observed fanout) also disables early-stop.
assert!(!metadata_early_stop_permitted(true, false, false, "", false, false));
// Whole/latest data-read metadata is now allowed by default;
// the inline verifier still decides whether it can stop early.
assert!(metadata_early_stop_permitted(true, true, true, "", false, false));
assert!(!metadata_early_stop_permitted(true, true, true, "", false, false));
},
);
}
#[test]
fn metadata_early_stop_allows_safe_data_reads_by_default() {
fn metadata_early_stop_keeps_data_reads_opt_in_by_default() {
temp_env::with_vars(
[
(ENV_RUSTFS_GET_METADATA_EARLY_STOP_ENABLE, Some("true")),
@@ -3904,7 +3900,7 @@ mod tests {
(ENV_RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE, None),
],
|| {
assert!(should_allow_metadata_early_stop(true, "", false, false));
assert!(!should_allow_metadata_early_stop(true, "", false, false));
assert!(!should_allow_metadata_early_stop(true, "version-id", false, false));
assert!(should_allow_metadata_early_stop(false, "", false, false));
assert!(!should_allow_metadata_early_stop(false, "version-id", false, false));
@@ -3936,34 +3932,6 @@ mod tests {
);
}
#[test]
fn metadata_early_stop_bounded_fanout_defaults_to_disabled() {
temp_env::with_vars(
[
(ENV_RUSTFS_GET_METADATA_EARLY_STOP_ENABLE, Some("true")),
(ENV_RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE, None),
(ENV_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT, None),
],
|| {
assert!(is_get_metadata_data_read_early_stop_enabled());
assert!(!is_get_metadata_early_stop_bounded_fanout_enabled());
},
);
temp_env::with_vars([(ENV_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT, Some("true"))], || {
assert!(is_get_metadata_early_stop_bounded_fanout_enabled());
});
temp_env::with_vars(
[
(ENV_RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE, Some("false")),
(ENV_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT, Some("false")),
],
|| {
assert!(!is_get_metadata_data_read_early_stop_enabled());
assert!(!is_get_metadata_early_stop_bounded_fanout_enabled());
},
);
}
#[test]
fn metadata_early_stop_rejects_healing_and_free_version_requests() {
temp_env::with_vars(
+4 -1
View File
@@ -135,7 +135,10 @@ impl FileMeta {
let i = buf.len() as u64;
// check version, buf = buf[8..]
let (buf, _, _) = Self::check_xl2_v1(buf)?;
let (buf, _, _) = Self::check_xl2_v1(buf).map_err(|e| {
error!("failed to check XL2 v1 format: {}", e);
e
})?;
if buf.len() < 5 {
error!(
+2 -7
View File
@@ -82,8 +82,8 @@ impl Error {
/// Whether a heal operation can be retried without changing its inputs.
pub(crate) fn is_recoverable_heal(&self) -> bool {
match self {
Error::TaskCancelled | Error::TaskTimeout => false,
Error::TransientSkip { .. } => true,
Error::TaskCancelled => false,
Error::TaskTimeout | Error::TransientSkip { .. } => true,
Error::Storage(err) => {
err.is_quorum_error()
|| matches!(
@@ -165,9 +165,4 @@ mod tests {
assert!(Error::Storage(EcstoreError::DiskNotFound).is_recoverable_heal());
assert!(Error::Storage(EcstoreError::VolumeNotFound).is_recoverable_heal());
}
#[test]
fn task_timeout_is_terminal() {
assert!(!Error::TaskTimeout.is_recoverable_heal());
}
}
+20 -30
View File
@@ -673,12 +673,6 @@ fn retry_request_for_result(task: &HealTask, result: &Result<()>) -> Option<(Hea
Some((request, delay, error))
}
async fn retry_request_for_result_with_budget(task: &HealTask, result: &Result<()>) -> Option<(HealRequest, Duration, String)> {
let (_, delay, error) = retry_request_for_result(task, result)?;
let request = task.retry_request_with_remaining_timeout().await.ok()?;
Some((request, delay, error))
}
fn recoverable_heal_retry_delay(retry_attempt: u32) -> Duration {
let retry_attempt = retry_attempt.clamp(1, 5);
let delay = Duration::from_secs(2_u64.saturating_pow(retry_attempt));
@@ -696,7 +690,7 @@ pub struct HealConfig {
pub max_concurrent_heals: usize,
/// Maximum concurrent heal tasks allowed for a single erasure set
pub max_concurrent_per_set: usize,
/// Aggregate task execution timeout across recoverable retries
/// Task timeout
pub task_timeout: Duration,
/// Queue size
pub queue_size: usize,
@@ -3112,7 +3106,7 @@ impl HealManager {
"Heal scheduler task started"
);
let result = task.execute().await;
let retry_request = retry_request_for_result_with_budget(task.as_ref(), &result).await;
let retry_request = retry_request_for_result(task.as_ref(), &result);
match &result {
Ok(_) => {
debug!(
@@ -4545,25 +4539,6 @@ mod tests {
assert!(retry_error.contains("Lock acquisition timeout"));
}
#[tokio::test]
async fn retry_request_for_result_preserves_remaining_timeout_budget() {
let storage: Arc<dyn HealStorageAPI> = Arc::new(MockStorage);
let mut request = HealRequest::object("retry-transition".to_string(), "object".to_string(), None);
request.options.timeout = Some(Duration::from_secs(60));
let task = HealTask::from_request(request, storage);
let result = task.execute().await;
let (retry_request, _, _) = retry_request_for_result_with_budget(&task, &result)
.await
.expect("read quorum failure should retain the unused timeout budget");
let remaining = retry_request
.options
.timeout
.expect("configured timeout should remain present");
assert!(remaining < Duration::from_secs(60));
assert!(remaining > Duration::from_secs(59));
}
#[test]
fn test_retry_request_for_incomplete_heal_rename() {
let storage: Arc<dyn HealStorageAPI> = Arc::new(MockStorage);
@@ -6079,7 +6054,7 @@ mod tests {
process_manager_queue_once(&manager).await;
let defaulted_status = tokio::time::timeout(Duration::from_secs(1), async {
loop {
if let Ok(status @ HealTaskStatus::Timeout) = manager.get_task_status(&defaulted_id).await {
if let Ok(status @ HealTaskStatus::Retrying { .. }) = manager.get_task_status(&defaulted_id).await {
break status;
}
tokio::task::yield_now().await;
@@ -6087,8 +6062,23 @@ mod tests {
})
.await
.expect("configured timeout should finish the task");
assert_eq!(defaulted_status, HealTaskStatus::Timeout);
assert!(manager.retrying_heals.lock().await.get(&defaulted_id).is_none());
assert!(matches!(defaulted_status, HealTaskStatus::Retrying { .. }));
assert_eq!(
manager
.retrying_heals
.lock()
.await
.get(&defaulted_id)
.expect("timed out task should retain its retry request")
.request
.options
.timeout,
Some(Duration::ZERO)
);
manager
.cancel_task(&defaulted_id)
.await
.expect("retrying timeout task should be cancelled");
let mut explicit = bucket_request("explicit-timeout", HealPriority::Normal, HealRequestSource::Admin);
explicit.options.timeout = Some(Duration::from_secs(60));
+1 -39
View File
@@ -196,7 +196,7 @@ pub struct HealOptions {
/// Whether to skip namespace locking
#[serde(default)]
pub no_lock: bool,
/// Aggregate execution timeout across recoverable manager retries
/// Timeout
pub timeout: Option<Duration>,
/// pool index
pub pool_index: Option<usize>,
@@ -442,14 +442,6 @@ impl HealTask {
}
}
pub(crate) async fn retry_request_with_remaining_timeout(&self) -> Result<HealRequest> {
let mut request = self.retry_request();
if self.options.timeout.is_some() {
request.options.timeout = self.remaining_timeout().await?;
}
Ok(request)
}
pub(crate) fn from_replacement_recovery_request(
request: HealRequest,
storage: Arc<dyn HealStorageAPI>,
@@ -2665,36 +2657,6 @@ mod tests {
use super::super::storage_api::status::BucketInfo;
#[tokio::test]
async fn retry_request_carries_remaining_timeout_budget() {
let storage: Arc<dyn HealStorageAPI> = Arc::new(MockStorage::default());
let mut request = HealRequest::bucket("bucket".to_string());
request.options.timeout = Some(Duration::from_secs(100));
let task = HealTask::from_request(request, storage.clone());
*task.task_start_instant.write().await = Some(Instant::now() - Duration::from_secs(40));
let retry = task
.retry_request_with_remaining_timeout()
.await
.expect("first retry should retain the unused timeout budget");
let first_remaining = retry.options.timeout.expect("configured timeout should remain present");
assert!(first_remaining <= Duration::from_secs(60));
assert!(first_remaining > Duration::from_secs(59));
let retry_task = HealTask::from_request(retry, storage);
*retry_task.task_start_instant.write().await = Some(Instant::now() - Duration::from_secs(20));
let second_retry = retry_task
.retry_request_with_remaining_timeout()
.await
.expect("second retry should retain only the unused aggregate budget");
let second_remaining = second_retry
.options
.timeout
.expect("configured timeout should remain present");
assert!(second_remaining <= Duration::from_secs(40));
assert!(second_remaining > Duration::from_secs(39));
}
#[test]
fn format_result_requires_every_requested_target_to_be_ok() {
let result = HealResultItem {
+2
View File
@@ -18,6 +18,8 @@
//! data encryption keys using master keys. It abstracts the encryption
//! operations so that different backends can share the same encryption logic.
#![allow(dead_code)] // Trait methods may be used by implementations
use crate::error::{KmsError, Result};
use crate::persisted_observability::{BoundedUnknownFieldName, UnknownFieldSummary};
use async_trait::async_trait;
+3 -3
View File
@@ -258,7 +258,7 @@ pub struct SRLDAPUser {
pub api_version: Option<String>,
}
#[derive(Debug, Serialize, Deserialize, Default)]
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
pub struct SRIAMUser {
#[serde(rename = "accessKey", default)]
pub access_key: String,
@@ -270,7 +270,7 @@ pub struct SRIAMUser {
pub api_version: Option<String>,
}
#[derive(Debug, Serialize, Deserialize, Default)]
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
pub struct SRGroupInfo {
#[serde(rename = "updateReq", default)]
pub update_req: GroupAddRemove,
@@ -346,7 +346,7 @@ pub struct SRCredInfo {
pub api_version: Option<String>,
}
#[derive(Debug, Serialize, Deserialize, Default)]
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
pub struct SRIAMItem {
#[serde(default)]
pub r#type: String,
+1 -32
View File
@@ -520,14 +520,6 @@ struct FailureSample {
pub struct FailStats {
pub count: i64,
pub size: i64,
/// Rolling-window snapshots refreshed at collection time
/// ([`Self::refresh_windows`]). The raw samples (`recent`) are process
/// local (serde-skipped), so these fields are what survives the peer-RPC
/// wire and [`Self::merge`]-based cluster aggregation.
#[serde(default)]
pub last_minute: FailedMetric,
#[serde(default)]
pub last_hour: FailedMetric,
#[serde(skip)]
recent: VecDeque<FailureSample>,
}
@@ -545,17 +537,6 @@ impl FailStats {
self.prune(observed_at);
}
/// Recompute the serializable rolling-window snapshots from the local
/// samples. Called at the collection point (per-node stats snapshot),
/// never on the failure hot path — the two deque scans are O(window) and
/// `add_size` runs under the bucket-stats write lock. Only meaningful on
/// the live per-node struct: a deserialized or merged struct has no
/// samples, and refreshing it would wipe the aggregated windows.
pub fn refresh_windows(&mut self) {
self.last_minute = self.recent_since(Duration::from_secs(60));
self.last_hour = self.recent_since(Duration::from_secs(3600));
}
fn prune(&mut self, observed_at: Instant) {
while self
.recent
@@ -584,16 +565,6 @@ impl FailStats {
Self {
count: self.count.saturating_add(other.count),
size: self.size.saturating_add(other.size),
// The window snapshots sum across nodes; the raw samples do not
// travel and stay empty on aggregated structs.
last_minute: FailedMetric {
count: self.last_minute.count.saturating_add(other.last_minute.count),
size: self.last_minute.size.saturating_add(other.last_minute.size),
},
last_hour: FailedMetric {
count: self.last_hour.count.saturating_add(other.last_hour.count),
size: self.last_hour.size.saturating_add(other.last_hour.size),
},
recent: VecDeque::new(),
}
}
@@ -665,9 +636,7 @@ impl BucketReplicationStat {
}
pub fn update_xfer_rate(&mut self, size: i64, duration: Duration) {
// Same boundary as the worker-pool split and minio-go's
// Large/Small transfer-summary labels: >= 128 MiB is "large".
if size >= crate::runtime::MIN_LARGE_OBJ_SIZE {
if size > 1024 * 1024 {
self.xfer_rate_lrg.add_size(size, duration);
} else {
self.xfer_rate_sml.add_size(size, duration);
@@ -6,18 +6,9 @@
#
# The MinIO release is pinned so the captured fixture format is reproducible;
# this is the release the interop tests were validated against.
#
# Both base images are build args so a network that cannot reach Docker Hub can
# point them at a mirror carrying the same content — quay.io publishes the MinIO
# releases, and public.ecr.aws mirrors the official Python images. CI keeps the
# Docker Hub defaults. Override with:
# --build-arg MINIO_IMAGE=quay.io/minio/minio:RELEASE.2025-09-07T16-13-09Z \
# --build-arg PYTHON_IMAGE=public.ecr.aws/docker/library/python:3.12-slim
ARG MINIO_IMAGE=minio/minio:RELEASE.2025-09-07T16-13-09Z
ARG PYTHON_IMAGE=python:3.12-slim
FROM ${MINIO_IMAGE} AS minio
FROM minio/minio:RELEASE.2025-09-07T16-13-09Z AS minio
FROM ${PYTHON_IMAGE}
FROM python:3.12-slim
RUN apt-get update \
&& apt-get install -y --no-install-recommends openssl ca-certificates \
&& rm -rf /var/lib/apt/lists/*
@@ -22,18 +22,6 @@ Use the automated path when you want the lab to:
- upload a predefined SSE fixture case
- export the generated backend tree into the lab layout
## Networks without Docker Hub access
`capture_via_docker.sh` pulls its two base images from Docker Hub by default. Where that registry is unreachable, point the build at mirrors carrying the same content — quay.io publishes the MinIO releases and public.ecr.aws mirrors the official Python images:
```bash
MINIO_LAB_MINIO_IMAGE=quay.io/minio/minio:RELEASE.2025-09-07T16-13-09Z \
MINIO_LAB_PYTHON_IMAGE=public.ecr.aws/docker/library/python:3.12-slim \
./capture_via_docker.sh
```
Pin the MinIO tag to the same release the Dockerfile names; an unpinned `:latest` captures whatever format that day's build writes, which is not what the interop tests were validated against.
## Layout
The default root is `artifacts/minio-fixture-lab`, which is already ignored by the repository.
@@ -34,19 +34,8 @@ if [ "${cases[0]}" != "all" ]; then
done
fi
# Base images are overridable so a network without Docker Hub access can point
# them at a mirror (see the Dockerfile header). Unset by default, which keeps the
# Dockerfile's Docker Hub defaults for CI.
build_args=()
if [ -n "${MINIO_LAB_MINIO_IMAGE:-}" ]; then
build_args+=(--build-arg "MINIO_IMAGE=${MINIO_LAB_MINIO_IMAGE}")
fi
if [ -n "${MINIO_LAB_PYTHON_IMAGE:-}" ]; then
build_args+=(--build-arg "PYTHON_IMAGE=${MINIO_LAB_PYTHON_IMAGE}")
fi
echo ">> building ${IMAGE}"
docker build -f "${SCRIPT_DIR}/Dockerfile" -t "${IMAGE}" "${build_args[@]}" "${SCRIPT_DIR}"
docker build -f "${SCRIPT_DIR}/Dockerfile" -t "${IMAGE}" "${SCRIPT_DIR}"
echo ">> capturing fixtures into ${FIXTURE_REL}"
docker run --rm -v "${REPO_ROOT}:/repo" "${IMAGE}" \
+1 -1
View File
@@ -102,7 +102,7 @@ bytes.workspace = true
hex-simd.workspace = true
[dev-dependencies]
tracing-subscriber = { workspace = true, features = ["json", "env-filter", "time"] }
tracing-subscriber = { workspace = true, features = ["env-filter", "time"] }
serial_test = { workspace = true }
temp-env = { workspace = true }
tempfile = { workspace = true }
+12 -111
View File
@@ -65,7 +65,6 @@ const LOG_SUBSYSTEM_FOLDER: &str = "folder";
const LOG_SUBSYSTEM_LIFECYCLE: &str = "lifecycle";
const LOG_SUBSYSTEM_HEAL: &str = "heal";
const EVENT_SCANNER_FOLDER_STATE: &str = "scanner_folder_state";
const EVENT_SCANNER_METADATA_CORRUPT: &str = "scanner_metadata_corrupt";
const EVENT_SCANNER_LIFECYCLE_ACTION: &str = "scanner_lifecycle_action";
const EVENT_SCANNER_HEAL_ADMISSION: &str = "scanner_heal_admission";
const EVENT_SCANNER_ALERT_STATE: &str = "scanner_alert_state";
@@ -2155,34 +2154,17 @@ impl FolderScanner {
self.record_failed(&item.path);
if should_log_failed_object(into.failed_objects) {
if let GetSizeFailureAction::HealMetadata { object } = &failure_action {
error!(
target: "rustfs::scanner::folder",
event = EVENT_SCANNER_METADATA_CORRUPT,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_FOLDER,
drive = %self.local_disk.path().display(),
bucket = %item.bucket,
object = %object,
metadata_path = %item.path,
failed_objects = into.failed_objects,
state = "metadata_corrupt",
error = %e,
"Scanner detected corrupt object metadata"
);
} else {
warn!(
target: "rustfs::scanner::folder",
event = EVENT_SCANNER_FOLDER_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_FOLDER,
path = %item.path,
failed_objects = into.failed_objects,
state = "get_size_failed",
error = %e,
"Scanner folder failed to get object size"
);
}
warn!(
target: "rustfs::scanner::folder",
event = EVENT_SCANNER_FOLDER_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_FOLDER,
path = %item.path,
failed_objects = into.failed_objects,
state = "get_size_failed",
error = %e,
"Scanner folder failed to get object size"
);
}
}
@@ -3072,59 +3054,12 @@ mod tests {
use crate::{DiskOption, Endpoint, STORAGE_FORMAT_FILE, TierStats, new_disk, storageclass};
use rustfs_filemeta::{FileInfo, FileMeta};
use serial_test::serial;
use std::io::Write;
#[cfg(unix)]
use std::os::unix::fs::{PermissionsExt, symlink};
use std::sync::Mutex;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use temp_env::{with_var, with_var_unset};
use tracing_subscriber::fmt::MakeWriter;
use uuid::Uuid;
#[derive(Clone, Default)]
struct CapturedLogs {
buffer: Arc<Mutex<Vec<u8>>>,
}
struct CapturedLogWriter {
buffer: Arc<Mutex<Vec<u8>>>,
}
impl CapturedLogs {
fn contents(&self) -> String {
let buffer = self
.buffer
.lock()
.expect("captured logs mutex should not be poisoned")
.clone();
String::from_utf8(buffer).expect("captured logs should be valid UTF-8")
}
}
impl Write for CapturedLogWriter {
fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
self.buffer
.lock()
.expect("captured logs mutex should not be poisoned")
.extend_from_slice(buf);
Ok(buf.len())
}
fn flush(&mut self) -> std::io::Result<()> {
Ok(())
}
}
impl<'a> MakeWriter<'a> for CapturedLogs {
type Writer = CapturedLogWriter;
fn make_writer(&'a self) -> Self::Writer {
CapturedLogWriter {
buffer: Arc::clone(&self.buffer),
}
}
}
#[test]
fn scanner_size_summary_application_saturates_usage_counters() {
let target = "arn:minio:replication::target".to_string();
@@ -4607,19 +4542,9 @@ mod tests {
assert!(budget.entries_visited() >= 1);
}
#[tokio::test(flavor = "current_thread")]
#[tokio::test]
#[serial]
async fn test_scan_folder_corrupt_xl_meta_stops_erasure_data_dir_descent() {
let logs = CapturedLogs::default();
let subscriber = tracing_subscriber::fmt()
.json()
.with_max_level(tracing::Level::ERROR)
.with_writer(logs.clone())
.with_ansi(false)
.without_time()
.finish();
let _subscriber_guard = tracing::subscriber::set_default(subscriber);
let (mut scanner, temp_dir) = build_test_scanner().await;
let _guard = TestGuard::new(60, 100, &mut scanner, temp_dir.clone());
@@ -4671,30 +4596,6 @@ mod tests {
assert!(!budget.budget_elapsed());
assert_eq!(budget.reason(), None);
let captured = logs.contents();
assert!(
!captured.contains("failed to check XL2 v1 format"),
"the context-free filemeta parser error must not be emitted"
);
let events = captured
.lines()
.map(|line| serde_json::from_str::<serde_json::Value>(line).expect("captured scanner log should be valid JSON"))
.filter(|line| line["fields"]["event"] == EVENT_SCANNER_METADATA_CORRUPT)
.collect::<Vec<_>>();
assert_eq!(
events.len(),
1,
"one corrupt metadata observation must emit one scanner-owned diagnostic event"
);
let fields = &events[0]["fields"];
assert_eq!(fields["component"], LOG_COMPONENT_SCANNER);
assert_eq!(fields["subsystem"], LOG_SUBSYSTEM_FOLDER);
assert_eq!(fields["drive"], temp_dir.to_string_lossy().as_ref());
assert_eq!(fields["bucket"], "bucket");
assert_eq!(fields["object"], "object");
assert_eq!(fields["metadata_path"], metadata_path.to_string_lossy().as_ref());
assert_eq!(fields["state"], "metadata_corrupt");
let retry_budget = ScannerCycleBudget::new_with_progress_tracking(
&parent,
crate::scanner_budget::ScannerCycleBudgetConfig {
+11
View File
@@ -3849,6 +3849,17 @@ impl ScannerIODisk for Disk {
let fivs = match meta.get_file_info_versions(item.bucket.as_str(), item.object_path().as_str(), false) {
Ok(versions) => versions,
Err(e) => {
error!(
target: "rustfs::scanner::io",
event = EVENT_SCANNER_DISK_BUCKET_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_IO,
bucket = %item.bucket,
object = %item.object_path(),
state = "file_info_versions_failed",
error = %e,
"Scanner disk bucket failed to resolve file info versions"
);
return Err(scanner_metadata_corrupt_error(
format!("failed to resolve file info versions: {e}"),
&item.bucket,
+298 -303
View File
@@ -1069,314 +1069,309 @@ mod serial_tests {
#[serial]
#[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-1)"]
async fn test_transition_and_restore_flows() {
async move {
let (disk_paths, ecstore) = setup_test_env().await;
let (disk_paths, ecstore) = setup_test_env().await;
let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase();
let backend = register_mock_tier(&tier_name).await;
let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase();
let backend = register_mock_tier(&tier_name).await;
let put_bucket = format!("test-immediate-put-{}", &Uuid::new_v4().simple().to_string()[..8]);
let put_object = "test/object.txt";
let put_payload = b"Hello, immediate transition!";
let put_bucket = format!("test-immediate-put-{}", &Uuid::new_v4().simple().to_string()[..8]);
let put_object = "test/object.txt";
let put_payload = b"Hello, immediate transition!";
create_test_bucket(&ecstore, put_bucket.as_str()).await;
set_bucket_lifecycle_transition_with_tier(put_bucket.as_str(), &tier_name)
create_test_bucket(&ecstore, put_bucket.as_str()).await;
set_bucket_lifecycle_transition_with_tier(put_bucket.as_str(), &tier_name)
.await
.expect("Failed to set lifecycle configuration");
let mut reader = PutObjReader::from_vec(put_payload.to_vec());
let mut metadata = HashMap::new();
metadata.insert("content-type".to_string(), "text/plain".to_string());
ecstore
.put_object(
put_bucket.as_str(),
put_object,
&mut reader,
&ObjectOptions {
user_defined: metadata,
..Default::default()
},
)
.await
.expect("Failed to upload transition metadata test object");
enqueue_transition_for_existing_objects(ecstore.clone(), put_bucket.as_str())
.await
.expect("Failed to enqueue transitioned put object");
let put_info = wait_for_transition(&ecstore, put_bucket.as_str(), put_object, TRANSITION_WAIT_TIMEOUT)
.await
.expect("object should transition after enqueueing existing objects");
assert_eq!(put_info.transitioned_object.status, "complete");
assert_eq!(put_info.transitioned_object.tier, tier_name);
assert!(backend.contains(&put_info.transitioned_object.name).await);
{
let transitioned = backend
.stored(&put_info.transitioned_object.name)
.await
.expect("Failed to set lifecycle configuration");
let mut reader = PutObjReader::from_vec(put_payload.to_vec());
let mut metadata = HashMap::new();
metadata.insert("content-type".to_string(), "text/plain".to_string());
ecstore
.put_object(
put_bucket.as_str(),
put_object,
&mut reader,
&ObjectOptions {
user_defined: metadata,
..Default::default()
},
)
.await
.expect("Failed to upload transition metadata test object");
enqueue_transition_for_existing_objects(ecstore.clone(), put_bucket.as_str())
.await
.expect("Failed to enqueue transitioned put object");
let put_info = wait_for_transition(&ecstore, put_bucket.as_str(), put_object, TRANSITION_WAIT_TIMEOUT)
.await
.expect("object should transition after enqueueing existing objects");
assert_eq!(put_info.transitioned_object.status, "complete");
assert_eq!(put_info.transitioned_object.tier, tier_name);
assert!(backend.contains(&put_info.transitioned_object.name).await);
{
let transitioned = backend
.stored(&put_info.transitioned_object.name)
.await
.expect("transitioned object should be present in mock backend");
assert_eq!(transitioned.metadata.get("content-type"), Some(&"text/plain".to_string()));
assert!(
!transitioned.metadata.contains_key("x-amz-replication-status"),
"transitioned objects must not inherit replication status defaults"
);
assert!(
!transitioned.metadata.contains_key("x-amz-object-lock-legal-hold"),
"transitioned objects must not invent object lock headers"
);
}
// Cross-shard xl.meta transition assertion helper (rustfs/backlog#1148 ilm-6):
// every disk must agree on the transition tuple for the object.
let put_meta = assert_transition_meta_consistent(&disk_paths, put_bucket.as_str(), put_object).await;
assert_eq!(put_meta.status, "complete");
assert_eq!(put_meta.tier, tier_name);
let multipart_bucket = format!("test-immediate-mpu-{}", &Uuid::new_v4().simple().to_string()[..8]);
let multipart_object = "test/multipart.txt";
create_test_bucket(&ecstore, multipart_bucket.as_str()).await;
set_bucket_lifecycle_transition_with_tier(multipart_bucket.as_str(), &tier_name)
.await
.expect("Failed to set lifecycle configuration");
let upload = ecstore
.new_multipart_upload(multipart_bucket.as_str(), multipart_object, &ObjectOptions::default())
.await
.expect("Failed to create multipart upload");
let part_data = b"multipart immediate transition";
let mut reader = PutObjReader::from_vec(part_data.to_vec());
let part = ecstore
.put_object_part(
multipart_bucket.as_str(),
multipart_object,
&upload.upload_id,
1,
&mut reader,
&ObjectOptions::default(),
)
.await
.expect("Failed to upload multipart part");
ecstore
.clone()
.complete_multipart_upload(
multipart_bucket.as_str(),
multipart_object,
&upload.upload_id,
vec![CompletePart {
part_num: 1,
etag: part.etag.clone(),
..Default::default()
}],
&ObjectOptions::default(),
)
.await
.expect("Failed to complete multipart upload");
enqueue_transition_for_existing_objects(ecstore.clone(), multipart_bucket.as_str())
.await
.expect("Failed to enqueue transitioned multipart object");
let multipart_info =
wait_for_transition(&ecstore, multipart_bucket.as_str(), multipart_object, TRANSITION_WAIT_TIMEOUT)
.await
.expect("object should transition after enqueueing existing objects");
assert_eq!(multipart_info.transitioned_object.status, "complete");
assert_eq!(multipart_info.transitioned_object.tier, tier_name);
assert!(backend.contains(&multipart_info.transitioned_object.name).await);
let src_bucket = format!("test-immediate-copy-src-{}", &Uuid::new_v4().simple().to_string()[..8]);
let dst_bucket = format!("test-immediate-copy-dst-{}", &Uuid::new_v4().simple().to_string()[..8]);
let src_object = "test/source.txt";
let dst_object = "test/copied.txt";
let payload = b"copy object immediate transition";
create_test_bucket(&ecstore, src_bucket.as_str()).await;
create_test_bucket(&ecstore, dst_bucket.as_str()).await;
set_bucket_lifecycle_transition_with_tier(dst_bucket.as_str(), &tier_name)
.await
.expect("Failed to set destination lifecycle configuration");
upload_test_object(&ecstore, src_bucket.as_str(), src_object, payload).await;
let mut src_info = ecstore
.get_object_info(src_bucket.as_str(), src_object, &ObjectOptions::default())
.await
.expect("Failed to load source object info");
src_info.put_object_reader = Some(PutObjReader::from_vec(payload.to_vec()));
ecstore
.copy_object(
src_bucket.as_str(),
src_object,
dst_bucket.as_str(),
dst_object,
&mut src_info,
&ObjectOptions::default(),
&ObjectOptions::default(),
)
.await
.expect("Failed to copy object");
enqueue_transition_for_existing_objects(ecstore.clone(), dst_bucket.as_str())
.await
.expect("Failed to enqueue transitioned copied object");
let copy_info = wait_for_transition(&ecstore, dst_bucket.as_str(), dst_object, TRANSITION_WAIT_TIMEOUT)
.await
.expect("copied object should transition after enqueueing existing objects");
assert_eq!(copy_info.transitioned_object.status, "complete");
assert_eq!(copy_info.transitioned_object.tier, tier_name);
assert!(backend.contains(&copy_info.transitioned_object.name).await);
let bucket_name = format!("test-lifecycle-update-{}", &Uuid::new_v4().simple().to_string()[..8]);
let object_name = "test/existing.txt";
let payload = b"existing object before lifecycle";
create_test_bucket(&ecstore, bucket_name.as_str()).await;
upload_test_object(&ecstore, bucket_name.as_str(), object_name, payload).await;
set_bucket_lifecycle_transition_with_tier(bucket_name.as_str(), &tier_name)
.await
.expect("Failed to set lifecycle configuration");
enqueue_transition_for_existing_objects(ecstore.clone(), bucket_name.as_str())
.await
.expect("Failed to enqueue transition for existing objects");
let info = wait_for_transition(&ecstore, bucket_name.as_str(), object_name, TRANSITION_WAIT_TIMEOUT)
.await
.expect("existing object should transition after lifecycle update");
assert_eq!(info.transitioned_object.status, "complete");
assert_eq!(info.transitioned_object.tier, tier_name);
assert!(backend.contains(&info.transitioned_object.name).await);
let bucket_name = format!("test-restore-mpu-{}", &Uuid::new_v4().simple().to_string()[..8]);
let object_name = "test/restore.txt";
let part1 = vec![b'a'; 5 * 1024 * 1024];
let part2 = b"restored-tail".to_vec();
let expected = [part1.clone(), part2.clone()].concat();
create_test_bucket(&ecstore, bucket_name.as_str()).await;
set_bucket_lifecycle_transition_with_tier(bucket_name.as_str(), &tier_name)
.await
.expect("Failed to set lifecycle configuration");
let upload = ecstore
.new_multipart_upload(bucket_name.as_str(), object_name, &ObjectOptions::default())
.await
.expect("Failed to create multipart upload");
let mut part1_reader = PutObjReader::from_vec(part1);
let uploaded_part1 = ecstore
.put_object_part(
bucket_name.as_str(),
object_name,
&upload.upload_id,
1,
&mut part1_reader,
&ObjectOptions::default(),
)
.await
.expect("Failed to upload first multipart part");
let mut part2_reader = PutObjReader::from_vec(part2);
let uploaded_part2 = ecstore
.put_object_part(
bucket_name.as_str(),
object_name,
&upload.upload_id,
2,
&mut part2_reader,
&ObjectOptions::default(),
)
.await
.expect("Failed to upload second multipart part");
ecstore
.clone()
.complete_multipart_upload(
bucket_name.as_str(),
object_name,
&upload.upload_id,
vec![
CompletePart {
part_num: 1,
etag: uploaded_part1.etag.clone(),
..Default::default()
},
CompletePart {
part_num: 2,
etag: uploaded_part2.etag.clone(),
..Default::default()
},
],
&ObjectOptions::default(),
)
.await
.expect("Failed to complete multipart upload");
enqueue_transition_for_existing_objects(ecstore.clone(), bucket_name.as_str())
.await
.expect("Failed to enqueue transitioned restore object");
let transitioned = wait_for_transition(&ecstore, bucket_name.as_str(), object_name, TRANSITION_WAIT_TIMEOUT)
.await
.expect("multipart object should transition after enqueueing existing objects");
assert_eq!(transitioned.parts.len(), 2);
ecstore
.clone()
.restore_transitioned_object(
bucket_name.as_str(),
object_name,
&ObjectOptions {
transition: TransitionOptions {
restore_request: RestoreRequest {
days: Some(1),
description: None,
glacier_job_parameters: None,
output_location: None,
select_parameters: None,
tier: None,
type_: None,
},
..Default::default()
},
..Default::default()
},
)
.await
.expect("Failed to restore transitioned multipart object");
let restored = ecstore
.get_object_info(bucket_name.as_str(), object_name, &ObjectOptions::default())
.await
.expect("Failed to load restored object info");
assert_eq!(restored.parts.len(), 2);
assert!(restored.restore_expires.is_some());
assert!(!restored.restore_ongoing);
let mut reader = ecstore
.get_object_reader(bucket_name.as_str(), object_name, None, http::HeaderMap::new(), &ObjectOptions::default())
.await
.expect("Failed to read restored object");
let mut data = Vec::new();
reader
.stream
.read_to_end(&mut data)
.await
.expect("Failed to consume restored object stream");
assert_eq!(data, expected);
.expect("transitioned object should be present in mock backend");
assert_eq!(transitioned.metadata.get("content-type"), Some(&"text/plain".to_string()));
assert!(
!transitioned.metadata.contains_key("x-amz-replication-status"),
"transitioned objects must not inherit replication status defaults"
);
assert!(
!transitioned.metadata.contains_key("x-amz-object-lock-legal-hold"),
"transitioned objects must not invent object lock headers"
);
}
.boxed_local()
.await;
// Cross-shard xl.meta transition assertion helper (rustfs/backlog#1148 ilm-6):
// every disk must agree on the transition tuple for the object.
let put_meta = assert_transition_meta_consistent(&disk_paths, put_bucket.as_str(), put_object).await;
assert_eq!(put_meta.status, "complete");
assert_eq!(put_meta.tier, tier_name);
let multipart_bucket = format!("test-immediate-mpu-{}", &Uuid::new_v4().simple().to_string()[..8]);
let multipart_object = "test/multipart.txt";
create_test_bucket(&ecstore, multipart_bucket.as_str()).await;
set_bucket_lifecycle_transition_with_tier(multipart_bucket.as_str(), &tier_name)
.await
.expect("Failed to set lifecycle configuration");
let upload = ecstore
.new_multipart_upload(multipart_bucket.as_str(), multipart_object, &ObjectOptions::default())
.await
.expect("Failed to create multipart upload");
let part_data = b"multipart immediate transition";
let mut reader = PutObjReader::from_vec(part_data.to_vec());
let part = ecstore
.put_object_part(
multipart_bucket.as_str(),
multipart_object,
&upload.upload_id,
1,
&mut reader,
&ObjectOptions::default(),
)
.await
.expect("Failed to upload multipart part");
ecstore
.clone()
.complete_multipart_upload(
multipart_bucket.as_str(),
multipart_object,
&upload.upload_id,
vec![CompletePart {
part_num: 1,
etag: part.etag.clone(),
..Default::default()
}],
&ObjectOptions::default(),
)
.await
.expect("Failed to complete multipart upload");
enqueue_transition_for_existing_objects(ecstore.clone(), multipart_bucket.as_str())
.await
.expect("Failed to enqueue transitioned multipart object");
let multipart_info = wait_for_transition(&ecstore, multipart_bucket.as_str(), multipart_object, TRANSITION_WAIT_TIMEOUT)
.await
.expect("object should transition after enqueueing existing objects");
assert_eq!(multipart_info.transitioned_object.status, "complete");
assert_eq!(multipart_info.transitioned_object.tier, tier_name);
assert!(backend.contains(&multipart_info.transitioned_object.name).await);
let src_bucket = format!("test-immediate-copy-src-{}", &Uuid::new_v4().simple().to_string()[..8]);
let dst_bucket = format!("test-immediate-copy-dst-{}", &Uuid::new_v4().simple().to_string()[..8]);
let src_object = "test/source.txt";
let dst_object = "test/copied.txt";
let payload = b"copy object immediate transition";
create_test_bucket(&ecstore, src_bucket.as_str()).await;
create_test_bucket(&ecstore, dst_bucket.as_str()).await;
set_bucket_lifecycle_transition_with_tier(dst_bucket.as_str(), &tier_name)
.await
.expect("Failed to set destination lifecycle configuration");
upload_test_object(&ecstore, src_bucket.as_str(), src_object, payload).await;
let mut src_info = ecstore
.get_object_info(src_bucket.as_str(), src_object, &ObjectOptions::default())
.await
.expect("Failed to load source object info");
src_info.put_object_reader = Some(PutObjReader::from_vec(payload.to_vec()));
ecstore
.copy_object(
src_bucket.as_str(),
src_object,
dst_bucket.as_str(),
dst_object,
&mut src_info,
&ObjectOptions::default(),
&ObjectOptions::default(),
)
.await
.expect("Failed to copy object");
enqueue_transition_for_existing_objects(ecstore.clone(), dst_bucket.as_str())
.await
.expect("Failed to enqueue transitioned copied object");
let copy_info = wait_for_transition(&ecstore, dst_bucket.as_str(), dst_object, TRANSITION_WAIT_TIMEOUT)
.await
.expect("copied object should transition after enqueueing existing objects");
assert_eq!(copy_info.transitioned_object.status, "complete");
assert_eq!(copy_info.transitioned_object.tier, tier_name);
assert!(backend.contains(&copy_info.transitioned_object.name).await);
let bucket_name = format!("test-lifecycle-update-{}", &Uuid::new_v4().simple().to_string()[..8]);
let object_name = "test/existing.txt";
let payload = b"existing object before lifecycle";
create_test_bucket(&ecstore, bucket_name.as_str()).await;
upload_test_object(&ecstore, bucket_name.as_str(), object_name, payload).await;
set_bucket_lifecycle_transition_with_tier(bucket_name.as_str(), &tier_name)
.await
.expect("Failed to set lifecycle configuration");
enqueue_transition_for_existing_objects(ecstore.clone(), bucket_name.as_str())
.await
.expect("Failed to enqueue transition for existing objects");
let info = wait_for_transition(&ecstore, bucket_name.as_str(), object_name, TRANSITION_WAIT_TIMEOUT)
.await
.expect("existing object should transition after lifecycle update");
assert_eq!(info.transitioned_object.status, "complete");
assert_eq!(info.transitioned_object.tier, tier_name);
assert!(backend.contains(&info.transitioned_object.name).await);
let bucket_name = format!("test-restore-mpu-{}", &Uuid::new_v4().simple().to_string()[..8]);
let object_name = "test/restore.txt";
let part1 = vec![b'a'; 5 * 1024 * 1024];
let part2 = b"restored-tail".to_vec();
let expected = [part1.clone(), part2.clone()].concat();
create_test_bucket(&ecstore, bucket_name.as_str()).await;
set_bucket_lifecycle_transition_with_tier(bucket_name.as_str(), &tier_name)
.await
.expect("Failed to set lifecycle configuration");
let upload = ecstore
.new_multipart_upload(bucket_name.as_str(), object_name, &ObjectOptions::default())
.await
.expect("Failed to create multipart upload");
let mut part1_reader = PutObjReader::from_vec(part1);
let uploaded_part1 = ecstore
.put_object_part(
bucket_name.as_str(),
object_name,
&upload.upload_id,
1,
&mut part1_reader,
&ObjectOptions::default(),
)
.await
.expect("Failed to upload first multipart part");
let mut part2_reader = PutObjReader::from_vec(part2);
let uploaded_part2 = ecstore
.put_object_part(
bucket_name.as_str(),
object_name,
&upload.upload_id,
2,
&mut part2_reader,
&ObjectOptions::default(),
)
.await
.expect("Failed to upload second multipart part");
ecstore
.clone()
.complete_multipart_upload(
bucket_name.as_str(),
object_name,
&upload.upload_id,
vec![
CompletePart {
part_num: 1,
etag: uploaded_part1.etag.clone(),
..Default::default()
},
CompletePart {
part_num: 2,
etag: uploaded_part2.etag.clone(),
..Default::default()
},
],
&ObjectOptions::default(),
)
.await
.expect("Failed to complete multipart upload");
enqueue_transition_for_existing_objects(ecstore.clone(), bucket_name.as_str())
.await
.expect("Failed to enqueue transitioned restore object");
let transitioned = wait_for_transition(&ecstore, bucket_name.as_str(), object_name, TRANSITION_WAIT_TIMEOUT)
.await
.expect("multipart object should transition after enqueueing existing objects");
assert_eq!(transitioned.parts.len(), 2);
ecstore
.clone()
.restore_transitioned_object(
bucket_name.as_str(),
object_name,
&ObjectOptions {
transition: TransitionOptions {
restore_request: RestoreRequest {
days: Some(1),
description: None,
glacier_job_parameters: None,
output_location: None,
select_parameters: None,
tier: None,
type_: None,
},
..Default::default()
},
..Default::default()
},
)
.await
.expect("Failed to restore transitioned multipart object");
let restored = ecstore
.get_object_info(bucket_name.as_str(), object_name, &ObjectOptions::default())
.await
.expect("Failed to load restored object info");
assert_eq!(restored.parts.len(), 2);
assert!(restored.restore_expires.is_some());
assert!(!restored.restore_ongoing);
let mut reader = ecstore
.get_object_reader(bucket_name.as_str(), object_name, None, http::HeaderMap::new(), &ObjectOptions::default())
.await
.expect("Failed to read restored object");
let mut data = Vec::new();
reader
.stream
.read_to_end(&mut data)
.await
.expect("Failed to consume restored object stream");
assert_eq!(data, expected);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
-29
View File
@@ -50,14 +50,6 @@ pub const SUFFIX_SOURCE_DELETEMARKER: &str = "source-deletemarker";
pub const SUFFIX_SOURCE_PROXY_REQUEST: &str = "source-proxy-request";
pub const SUFFIX_SOURCE_REPLICATION_REQUEST: &str = "source-replication-request";
pub const SUFFIX_SOURCE_REPLICATION_CHECK: &str = "source-replication-check";
// LWW timestamps for replicated tag/retention/legal-hold modifications. MinIO
// declares these with mixed case (internal/http/headers.go:
// X-Minio-Source-Replication-Tagging-Timestamp / -Retention-Timestamp /
// -LegalHold-Timestamp); HTTP header names compare case-insensitively, so the
// lowercase suffix forms interoperate. Values are RFC3339 on the wire.
pub const SUFFIX_SOURCE_REPLICATION_TAGGING_TIMESTAMP: &str = "source-replication-tagging-timestamp";
pub const SUFFIX_SOURCE_REPLICATION_RETENTION_TIMESTAMP: &str = "source-replication-retention-timestamp";
pub const SUFFIX_SOURCE_REPLICATION_LEGALHOLD_TIMESTAMP: &str = "source-replication-legalhold-timestamp";
pub const SUFFIX_REPLICATION_SSEC_CRC: &str = "replication-ssec-crc";
/// Returns true if the key is object-encryption metadata understood by RustFS or MinIO.
@@ -204,27 +196,6 @@ mod tests {
assert_eq!(get_object_encryption_original_size(&metadata).expect("valid size"), Some(42));
}
#[test]
fn replication_timestamp_headers_match_minio_wire_names() {
let mut headers = HeaderMap::new();
insert_header(&mut headers, SUFFIX_SOURCE_REPLICATION_TAGGING_TIMESTAMP, "2026-01-02T03:04:05Z");
insert_header(&mut headers, SUFFIX_SOURCE_REPLICATION_RETENTION_TIMESTAMP, "2026-01-02T03:04:06Z");
insert_header(&mut headers, SUFFIX_SOURCE_REPLICATION_LEGALHOLD_TIMESTAMP, "2026-01-02T03:04:07Z");
// The exact names MinIO's object-api-options.go reads (its Get()
// canonicalizes case, so a case-insensitive match is wire-equivalent).
for name in [
"X-Minio-Source-Replication-Tagging-Timestamp",
"X-Minio-Source-Replication-Retention-Timestamp",
"X-Minio-Source-Replication-LegalHold-Timestamp",
"x-rustfs-source-replication-tagging-timestamp",
"x-rustfs-source-replication-retention-timestamp",
"x-rustfs-source-replication-legalhold-timestamp",
] {
assert!(headers.contains_key(name), "replication timestamp header {name} must be written");
}
}
#[test]
fn test_get_header() {
let mut headers = HeaderMap::new();
@@ -61,17 +61,17 @@ catalog extension.
| Area | Status | Covered behavior |
|---|---|---|
| Catalog config | Supported | `GET /v1/config` advertises RustFS catalog defaults and only the supported OpenAPI REST paths in `endpoints`. RustFS administration, maintenance, migration, diagnostics, refs, and metadata-location extensions remain available but are not presented as standard Iceberg REST endpoints. |
| Catalog config | Supported | `GET /v1/config` advertises RustFS catalog defaults and route capabilities. |
| Table bucket discovery | Supported | `PUT` and `GET /v1/buckets/{warehouse}` enable and inspect table bucket state. |
| Namespaces | Supported | Create, list, load, existence check, and drop namespace routes are registered on both catalog prefixes. List responses support Iceberg REST `pageSize`/`pageToken` pagination with context-bound tokens and bounded catalog-store reads. Namespace identifiers are limited to 512 ASCII characters so persisted paths and stateless continuation tokens remain bounded. |
| Tables | Supported | Create, register, list, load, existence check, commit, metadata-location get/update, and drop table routes are registered on both catalog prefixes. Table and view listings support Iceberg REST `pageSize`/`pageToken` pagination with context-bound tokens and bounded catalog-store reads. Commit identifiers must match the URL resource; unknown requirements, updates, and snapshot operations fail as bad requests; staged create, register overwrite, purge-on-drop, and v3-only encryption-key updates return an explicit unsupported-operation response. Standard statistics, partition statistics, and schema/spec cleanup updates are accepted. |
| Commit CAS | Supported | Single-table commits validate base metadata, expected version token, referenced object existence, warehouse scope, and Iceberg commit requirements before advancing the current metadata pointer. Externally supplied metadata transitions preserve monotonic column, partition, and sequence assignment watermarks and immutable definitions for retained schemas, partition specs, sort orders, and snapshots. Standard commits preserve the normal commit-token file name and use an immutable-table-scoped fallback when rename followed by source-name reuse would otherwise collide at the same generation and commit ID. The catalog does not advertise `idempotency-key-lifetime`; clients must treat standard mutation-wide `Idempotency-Key` semantics as unsupported. |
| Tables | Supported | Create, register, list, load, existence check, commit, metadata-location get/update, and drop table routes are registered on both catalog prefixes. Table and view listings support Iceberg REST `pageSize`/`pageToken` pagination with context-bound tokens and bounded catalog-store reads. |
| Commit CAS | Supported | Single-table commits validate base metadata, expected version token, referenced object existence, warehouse scope, and Iceberg commit requirements before advancing the current metadata pointer. Standard commits preserve the normal commit-token file name and use an immutable-table-scoped fallback when rename followed by source-name reuse would otherwise collide at the same generation and commit ID. |
| Commit recovery | Supported | Commit log, idempotency lookup, diagnostics, and recovery routes expose staged/finalization gaps and repair safe idempotency gaps without moving the table pointer. |
| Snapshot refs | Supported | Refs can be listed, created or replaced, and deleted through catalog commits. `main` is protected and refs with explicit retention require forced delete. |
| Iceberg views | Supported | Basic create, list, load, replace, existence check, and drop routes persist view metadata with view-scoped authorization. Replace identifiers must match the URL resource, `schema-id: -1` resolves to the last added schema, one commit timestamp is used consistently, and only Iceberg view format version 1 is accepted. |
| Table credentials endpoint | Supported | Returns an empty `storage-credentials` list by default. Returns table-scoped temporary credentials only when credential vending is enabled. Credential responses set `Cache-Control: no-store, private`, `Pragma: no-cache`, and `Expires: 0`. |
| Iceberg views | Supported | Basic create, list, load, replace, existence check, and drop routes persist view metadata with view-scoped authorization. |
| Table credentials endpoint | Supported | Returns an empty `storage-credentials` list by default. Returns table-scoped temporary credentials only when credential vending is enabled. |
| Catalog diagnostics and export | Supported | Exposes recovery state, consistency state, backing manifest, recoverable commit-log WAL state, strong backing migration target, single-active-writer policy, and scale validation matrix. |
| Catalog import and rollback | Supported | Import/register and online rollback use catalog validation and commit paths rather than direct pointer mutation. Online rollback accepts only a forward-safe metadata target that preserves assignment watermarks and retained definitions. Restoring an older target that lowers those watermarks is an offline disaster-recovery operation and requires every writer to be stopped. |
| Catalog import and rollback | Supported | Import/register and rollback use catalog validation and commit paths rather than direct pointer mutation. |
| External catalog bridge | Supported operator path | Operator-supplied metadata pointer sync/import is supported for external catalog identity boundaries. Online vendor SDK polling and policy mirroring are not claimed. |
| Multi-table transactions | Not claimed | RustFS currently claims single-table commit atomicity only. |
Generated
+6 -6
View File
@@ -2,11 +2,11 @@
"nodes": {
"nixpkgs": {
"locked": {
"lastModified": 1786719841,
"narHash": "sha256-QcpQOT0NQEFkI77t+YXPZqDJc35iIodG7zinieOwFUg=",
"lastModified": 1785975029,
"narHash": "sha256-X44cn5rzytELc3NNoQsh0aLkjWA/QzPfc6HPQmsG3sU=",
"owner": "NixOS",
"repo": "nixpkgs",
"rev": "8be7bd0c83f12e2e3bbba07c9044d6fed9e66f7f",
"rev": "70ce234312134a463ba7728e94da2486a1d237ac",
"type": "github"
},
"original": {
@@ -29,11 +29,11 @@
]
},
"locked": {
"lastModified": 1786849542,
"narHash": "sha256-MS8cOa/ii1+QA8R3JWVzqEIoXimu9n50GwQDiBVTFOM=",
"lastModified": 1786247265,
"narHash": "sha256-cVTcTqAhzSNMOVddtBhFpuv+Vx6/SgrhgZWJiI61H24=",
"owner": "oxalica",
"repo": "rust-overlay",
"rev": "b211eadeba8b180da9453ec3413a8a3535c85b3f",
"rev": "6df7076ea0f5697e3242719a4c71211f00411cf5",
"type": "github"
},
"original": {
+1 -2
View File
@@ -278,8 +278,6 @@ rustfs-signer.workspace = true
serde = { workspace = true, features = ["derive"] }
serde_json = { workspace = true, features = ["raw_value"] }
serde_urlencoded = { workspace = true }
snap.workspace = true
zstd.workspace = true
# Cryptography and Security
rustls = { workspace = true, default-features = false, features = ["aws-lc-rs", "logging", "tls12", "prefer-post-quantum", "std"] }
@@ -357,6 +355,7 @@ rcgen = { workspace = true }
rustfs-test-utils.workspace = true
# diagnose_e2e fixtures (archives are generated in-test, never checked in)
zip = { workspace = true }
zstd = { 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"] }
+7 -8
View File
@@ -40,7 +40,7 @@ use s3s::{Body, S3Request, S3Response, S3Result, s3_error};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::sync::LazyLock;
use tracing::{error, info, warn};
use tracing::{Span, error, info, warn};
const LOG_COMPONENT_ADMIN_API: &str = "admin_api";
const LOG_SUBSYSTEM_AUDIT_TARGET: &str = "audit_target";
@@ -278,6 +278,8 @@ pub struct AuditTargetConfig {}
#[async_trait::async_trait]
impl Operation for AuditTargetConfig {
async fn call(&self, req: S3Request<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
let span = Span::current();
let _enter = span.enter();
let (target_type, target_name) = extract_target_params(&params)?;
authorize_audit_admin_request(&req, AdminAction::SetBucketTargetAction).await?;
@@ -343,6 +345,8 @@ pub struct ListAuditTargets {}
#[async_trait::async_trait]
impl Operation for ListAuditTargets {
async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
let span = Span::current();
let _enter = span.enter();
authorize_audit_admin_request(&req, AdminAction::GetBucketTargetAction).await?;
let mut runtime_statuses = HashMap::new();
@@ -366,6 +370,8 @@ pub struct RemoveAuditTarget {}
#[async_trait::async_trait]
impl Operation for RemoveAuditTarget {
async fn call(&self, req: S3Request<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
let span = Span::current();
let _enter = span.enter();
let (target_type, target_name) = extract_target_params(&params)?;
authorize_audit_admin_request(&req, AdminAction::SetBucketTargetAction).await?;
@@ -832,13 +838,6 @@ mod tests {
extract_block_between_markers(src, "impl Operation for ListAuditTargets", "pub struct RemoveAuditTarget");
let delete_block = extract_block_between_markers(src, "impl Operation for RemoveAuditTarget", "#[cfg(test)]");
for block in [put_block, list_block, delete_block] {
assert!(
!block.contains(".enter()"),
"async audit handlers must rely on request-future instrumentation instead of holding span guards across awaits"
);
}
assert!(
put_block.contains("authorize_audit_admin_request(&req, AdminAction::SetBucketTargetAction).await?;"),
"audit target writes should require SetBucketTargetAction"
+9 -8
View File
@@ -43,7 +43,7 @@ use s3s::{Body, S3Request, S3Response, S3Result, s3_error};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::sync::LazyLock;
use tracing::{error, info, warn};
use tracing::{Span, error, info, warn};
const LOG_COMPONENT_ADMIN_API: &str = "admin_api";
@@ -333,6 +333,8 @@ pub struct NotificationTarget {}
#[async_trait::async_trait]
impl Operation for NotificationTarget {
async fn call(&self, req: S3Request<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
let span = Span::current();
let _enter = span.enter();
let (target_type, target_name) = extract_target_params(&params)?;
let context = app_context_from_req(&req);
@@ -399,6 +401,8 @@ pub struct ListNotificationTargets {}
#[async_trait::async_trait]
impl Operation for ListNotificationTargets {
async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
let span = Span::current();
let _enter = span.enter();
authorize_notification_admin_request(&req, AdminAction::GetBucketTargetAction).await?;
refresh_persisted_module_switches_from_store().await.map_err(|err| {
warn!(
@@ -435,6 +439,8 @@ pub struct ListTargetsArns {}
#[async_trait::async_trait]
impl Operation for ListTargetsArns {
async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
let span = Span::current();
let _enter = span.enter();
authorize_notification_admin_request(&req, AdminAction::GetBucketTargetAction).await?;
if let Some(reason) = notification_target_operation_block_reason(
"querying notification target ARNs for bucket associations from the console",
@@ -479,6 +485,8 @@ pub struct RemoveNotificationTarget {}
#[async_trait::async_trait]
impl Operation for RemoveNotificationTarget {
async fn call(&self, req: S3Request<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
let span = Span::current();
let _enter = span.enter();
let (target_type, target_name) = extract_target_params(&params)?;
let context = app_context_from_req(&req);
@@ -998,13 +1006,6 @@ mod tests {
extract_block_between_markers(src, "impl Operation for ListTargetsArns", "pub struct RemoveNotificationTarget");
let delete_block = extract_block_between_markers(src, "impl Operation for RemoveNotificationTarget", "fn extract_param");
for block in [put_block, list_block, arns_block, delete_block] {
assert!(
!block.contains(".enter()"),
"async notification handlers must rely on request-future instrumentation instead of holding span guards across awaits"
);
}
assert!(
put_block.contains("authorize_notification_admin_request(&req, AdminAction::SetBucketTargetAction).await?;"),
"notification target writes should require SetBucketTargetAction"
+19 -353
View File
@@ -535,13 +535,8 @@ impl Operation for GetReplicationMetricsHandler {
let bucket_stats = cluster_replication_stats(bucket, app_context_from_req(&req)).await;
// Same minio-go `replication.Metrics` wire shape as
// `?replication-metrics` — the internal snake_case stats are the peer
// RPC wire format and must not leak here.
let data = serde_json::to_vec(&crate::admin::replication_metrics_wire::MetricsWire::from(
&bucket_stats.replication_stats,
))
.map_err(|_| S3Error::with_message(S3ErrorCode::InternalError, "serialize failed"))?;
let data = serde_json::to_vec(&bucket_stats.replication_stats)
.map_err(|_| S3Error::with_message(S3ErrorCode::InternalError, "serialize failed"))?;
let mut headers = HeaderMap::new();
headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));
Ok(S3Response::with_headers((StatusCode::OK, Body::from(data)), headers))
@@ -1158,7 +1153,7 @@ struct MrfResponse {
fn build_mrf_response(
bucket: String,
bucket_stats: &BucketStats,
durable: &crate::admin::storage_api::replication::DurableMrfBacklog,
durable: crate::admin::storage_api::replication::DurableMrfBacklog,
) -> MrfResponse {
let observation_scope = if bucket_stats.replication_stats.cluster_complete {
"cluster_aggregated"
@@ -1228,10 +1223,7 @@ fn build_mrf_response(
total_failed_size,
queued_count: queued.count,
queued_size: queued.bytes,
// The default (non-aggregate) response mode streams the durable
// backlog per object, so the enumerable API exists whenever the
// backlog is readable.
per_object_entries_available: durable.available,
per_object_entries_available: false,
runtime_stats_available: bucket_stats.replication_stats.provider_available,
cluster_complete: bucket_stats.replication_stats.cluster_complete,
observed_node_count: bucket_stats.replication_stats.observed_node_count,
@@ -1243,165 +1235,23 @@ fn build_mrf_response(
}
}
/// One durable MRF backlog entry rendered for the default (madmin-compatible)
/// stream. Field names are the exact json tags of madmin-go `ReplicationMRF`
/// (replication-api.go), which `mc replicate backlog` decodes one JSON
/// document at a time. `Size` and `TargetARNs` are RustFS extension keys with
/// no madmin counterpart; Go decoders ignore unknown keys.
#[derive(Debug, Serialize)]
struct MrfEntryDocument {
/// The durable backlog is a cluster-shared ledger with no per-node
/// attribution, so the madmin `nodeName` tag is always empty.
#[serde(rename = "nodeName")]
node_name: String,
#[serde(rename = "bucket")]
bucket: String,
#[serde(rename = "object")]
object: String,
#[serde(rename = "versionId")]
version_id: String,
#[serde(rename = "retryCount")]
retry_count: i32,
#[serde(rename = "Size")]
size: i64,
#[serde(rename = "TargetARNs", skip_serializing_if = "Vec::is_empty")]
target_arns: Vec<String>,
}
/// Upper bound on the number of documents one stream response emits. The
/// durable ledger is not bounded by the in-memory pending cap (recovery can
/// persist far larger generations), and the body is buffered before send, so
/// an unbounded read could stage hundreds of MB per request. The handler
/// rejects a response beyond this bound instead of returning a partial 200.
const REPLICATION_MRF_MAX_STREAM_ENTRIES: usize = 10_000;
/// Project the durable backlog into madmin `ReplicationMRF` documents,
/// scoped to `bucket` when it is non-empty (madmin allows an empty bucket to
/// mean "across all buckets"), bounded by
/// [`REPLICATION_MRF_MAX_STREAM_ENTRIES`]. Returns the documents and whether
/// the backlog was truncated.
fn mrf_entry_documents(
bucket: &str,
durable: &crate::admin::storage_api::replication::DurableMrfBacklog,
) -> (Vec<MrfEntryDocument>, bool) {
let mut documents = Vec::new();
let mut truncated = false;
for entry in durable
.entries
.iter()
.filter(|entry| bucket.is_empty() || entry.bucket == bucket)
{
if documents.len() >= REPLICATION_MRF_MAX_STREAM_ENTRIES {
truncated = true;
break;
}
documents.push(MrfEntryDocument {
node_name: String::new(),
bucket: entry.bucket.clone(),
object: entry.object.clone(),
// Delete-marker purge entries track the marker version separately;
// fall back to it so those rows still carry a version identity.
// The nil UUID is RustFS's in-memory null-version sentinel and
// must leave as the S3 wire token, not a zero UUID.
version_id: entry
.version_id
.or(entry.delete_marker_version_id)
.map(|v| {
if v.is_nil() {
rustfs_filemeta::NULL_VERSION_ID.to_string()
} else {
v.to_string()
}
})
.unwrap_or_default(),
retry_count: entry.retry_count,
size: entry.size,
target_arns: entry.target_arns.clone(),
});
}
(documents, truncated)
}
/// Render the MRF backlog as a response body.
///
/// Default (madmin-compatible) mode emits one `ReplicationMRF` JSON document
/// per line with no envelope — madmin's `BucketReplicationMRF` reads the body
/// with a `json.Decoder` loop, so an envelope object would decode as a single
/// entry whose `"Bucket"` key case-insensitively matches
/// `ReplicationMRF.Bucket` (a phantom row in `mc replicate backlog`), and an
/// empty backlog must render an empty body so the loop ends on io.EOF with
/// zero rows.
///
/// `aggregate=true` (RustFS extension) keeps the enveloped counter shape;
/// backlog-source health (`RuntimeStatsAvailable`/`DurableBacklogAvailable`)
/// is only representable there — an unreadable ledger fails the stream
/// request outright in the handler (madmin only decodes the body of a 200,
/// so an empty stream would read as a healthy zero-row backlog).
fn render_mrf_backlog(
response: &MrfResponse,
durable: &crate::admin::storage_api::replication::DurableMrfBacklog,
aggregate: bool,
) -> Result<(Vec<u8>, bool), serde_json::Error> {
if aggregate {
return Ok((serde_json::to_vec(response)?, false));
}
let (documents, truncated) = mrf_entry_documents(&response.bucket, durable);
let mut data = Vec::new();
for entry in documents {
serde_json::to_writer(&mut data, &entry)?;
data.push(b'\n');
}
Ok((data, truncated))
}
fn ensure_complete_mrf_stream(truncated: bool) -> S3Result<()> {
if truncated {
return Err(S3Error::with_message(
S3ErrorCode::ServiceUnavailable,
"durable MRF backlog exceeds the stream limit; narrow the bucket scope or drain the backlog".to_string(),
));
}
Ok(())
}
/// `GET /v3/replication/mrf`
///
/// Reports the failed-replication backlog (MinIO's MRF concept) for a bucket.
///
/// The default response is a madmin-compatible stream of `ReplicationMRF`
/// documents built from the durable backlog ledger (in-memory failures that
/// have not been flushed yet — the persister runs every few seconds — are not
/// visible). `?aggregate=true` (RustFS extension) returns the enveloped
/// runtime + durable counter shape instead; `PerTargetDurableEntriesAvailable`
/// is false there when the durable backlog includes older entries that cannot
/// be attributed to a target.
///
/// The madmin `node` parameter is accepted but has no filtering effect: the
/// durable ledger is cluster-shared with no per-node attribution, so every
/// node serves the same (complete) backlog.
///
/// Authorization: the stream requires `admin:ReplicationDiff` (it enumerates
/// object names and version ids, MinIO parity); `?aggregate=true` carries no
/// object identities and requires only `admin:GetReplicationMetrics`.
/// Compatibility note: MinIO returns a stream of individual MRF entries. RustFS
/// deliberately returns aggregate runtime and durable counters instead.
/// `PerObjectEntriesAvailable` remains false until an enumerable API exists.
/// `PerTargetDurableEntriesAvailable` is false when the durable backlog includes
/// older entries that cannot be attributed to a target.
pub struct ReplicationMrfHandler {}
#[async_trait::async_trait]
impl Operation for ReplicationMrfHandler {
async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
let queries = extract_query_params(&req.uri);
let aggregate = queries.get("aggregate").map(String::as_str) == Some("true");
// The default stream enumerates object names and version ids, which
// a metrics-only principal must not see; gate it on the same action
// MinIO uses for this endpoint. The aggregate counters carry no
// object identities and keep the metrics action.
let action = if aggregate {
AdminAction::GetReplicationMetricsAction
} else {
AdminAction::ReplicationDiff
};
validate_replication_admin_request(&req, action).await?;
validate_replication_admin_request(&req, AdminAction::GetReplicationMetricsAction).await?;
let queries = extract_query_params(&req.uri);
let Some(bucket) = queries.get("bucket").filter(|b| !b.is_empty()).cloned() else {
return Err(s3_error!(InvalidRequest, "bucket is required"));
};
@@ -1425,48 +1275,14 @@ impl Operation for ReplicationMrfHandler {
return Err(ApiError::from(err).into());
}
if let Some(node) = queries.get("node").filter(|node| !node.is_empty() && node.as_str() != "all") {
// The durable backlog ledger is cluster-shared with no per-node
// attribution, so a node-scoped request still sees the complete
// (superset) backlog.
debug!(node = %node, "replication mrf node filter has no effect on the cluster-shared backlog");
}
let durable = crate::admin::storage_api::replication::read_durable_mrf_backlog(store).await;
let bucket_stats = cluster_replication_stats(&bucket, app_context_from_req(&req)).await;
let response = build_mrf_response(bucket, &bucket_stats, &durable);
let response = build_mrf_response(bucket, &bucket_stats, durable);
if !durable.available && !aggregate {
// The madmin stream has no envelope to carry source health, and
// madmin only decodes the body of a 200 — an empty stream would
// read as a clean, healthy zero-row backlog. Fail loudly instead;
// aggregate mode still reports the availability fields.
tracing::warn!(
bucket = %response.bucket,
"durable MRF backlog is unreadable; failing the stream request — use aggregate=true to see source health"
);
return Err(S3Error::with_message(
S3ErrorCode::ServiceUnavailable,
"durable MRF backlog is unreadable; retry, or use aggregate=true for source health".to_string(),
));
}
let (data, truncated) = render_mrf_backlog(&response, &durable, aggregate)
let data = serde_json::to_vec(&response)
.map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("serialize failed: {e}")))?;
let mut headers = HeaderMap::new();
headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));
if truncated {
tracing::warn!(
event = "replication_mrf_stream_rejected",
component = "admin",
subsystem = "replication",
result = "rejected",
bucket = %response.bucket,
max_entries = REPLICATION_MRF_MAX_STREAM_ENTRIES,
"replication mrf stream exceeds the response limit"
);
}
ensure_complete_mrf_stream(truncated)?;
Ok(S3Response::with_headers((StatusCode::OK, Body::from(data)), headers))
}
}
@@ -1476,8 +1292,7 @@ mod tests {
use super::{
REMOTE_TARGET_UNSUPPORTED_FIELDS, REMOTE_TARGET_WRITABLE_FIELDS, RemoteTargetCredentialsRequest, RemoteTargetRequest,
ReplicationDiffEntry, SUPPORTED_REMOTE_TARGET_API, TargetUpdateOp, build_mrf_response, extract_query_params,
parse_remote_target_update_ops, render_mrf_backlog, render_replication_diff, unique_replication_peers,
validate_remote_target_tls_settings,
parse_remote_target_update_ops, render_replication_diff, unique_replication_peers, validate_remote_target_tls_settings,
};
use crate::admin::storage_api::bucket::target::{BucketTarget, LatencyStat};
use crate::admin::storage_api::replication::{BucketStats, DurableMrfBacklog, MrfOpKind, MrfReplicateEntry};
@@ -1695,7 +1510,7 @@ mod tests {
],
};
let response = build_mrf_response("bucket-a".to_string(), &stats, &durable);
let response = build_mrf_response("bucket-a".to_string(), &stats, durable);
let json = serde_json::to_value(response).expect("MRF response should serialize");
assert_eq!(json["TotalFailedCount"], 3);
@@ -1707,9 +1522,7 @@ mod tests {
assert_eq!(json["RuntimeStatsAvailable"], true);
assert_eq!(json["ClusterComplete"], false);
assert_eq!(json["Targets"][0]["ObservationScope"], "partial_cluster");
// The bare stream enumerates the durable backlog per object, so a
// readable backlog advertises the enumerable API.
assert_eq!(json["PerObjectEntriesAvailable"], true);
assert_eq!(json["PerObjectEntriesAvailable"], false);
assert_eq!(json["PerTargetDurableEntriesAvailable"], true);
let targets = json["Targets"].as_array().expect("targets should serialize as an array");
@@ -1756,7 +1569,7 @@ mod tests {
}],
};
let response = build_mrf_response("bucket-a".to_string(), &stats, &durable);
let response = build_mrf_response("bucket-a".to_string(), &stats, durable);
let json = serde_json::to_value(response).expect("MRF response should serialize");
assert_eq!(json["DurableBacklogAvailable"], true);
@@ -1774,7 +1587,7 @@ mod tests {
#[test]
fn mrf_response_distinguishes_unavailable_sources_from_valid_zero() {
let unavailable = build_mrf_response("bucket-a".to_string(), &BucketStats::default(), &DurableMrfBacklog::default());
let unavailable = build_mrf_response("bucket-a".to_string(), &BucketStats::default(), DurableMrfBacklog::default());
let unavailable_json = serde_json::to_value(unavailable).expect("unavailable response should serialize");
assert_eq!(unavailable_json["RuntimeStatsAvailable"], false);
assert_eq!(unavailable_json["DurableBacklogAvailable"], false);
@@ -1787,7 +1600,7 @@ mod tests {
let valid_empty = build_mrf_response(
"bucket-a".to_string(),
&valid_empty_stats,
&DurableMrfBacklog {
DurableMrfBacklog {
available: true,
entries: Vec::new(),
},
@@ -1800,153 +1613,6 @@ mod tests {
assert_eq!(valid_empty_json["PerTargetDurableEntriesAvailable"], true);
}
fn sample_durable_backlog() -> DurableMrfBacklog {
DurableMrfBacklog {
available: true,
entries: vec![
MrfReplicateEntry {
bucket: "bucket-a".to_string(),
object: "object-a".to_string(),
version_id: Some(uuid::Uuid::from_u128(7)),
retry_count: 2,
size: 250,
op: MrfOpKind::Object,
target_arns: vec!["arn-a".to_string()],
..Default::default()
},
MrfReplicateEntry {
bucket: "other-bucket".to_string(),
object: "object-b".to_string(),
version_id: None,
retry_count: 0,
size: 999,
op: MrfOpKind::Object,
target_arns: Vec::new(),
..Default::default()
},
],
}
}
/// madmin's `BucketReplicationMRF` decodes the body one `ReplicationMRF`
/// JSON document at a time; the default response must therefore be a bare
/// document stream with madmin's exact json tags, not an envelope.
#[test]
fn mrf_stream_renders_bare_madmin_documents() {
let durable = sample_durable_backlog();
let response = build_mrf_response("bucket-a".to_string(), &BucketStats::default(), &durable);
let (body, _) = render_mrf_backlog(&response, &durable, false).expect("stream body should serialize");
let text = String::from_utf8(body).expect("body should be utf-8");
let lines: Vec<&str> = text.lines().filter(|line| !line.trim().is_empty()).collect();
// Only the entry matching the requested bucket is streamed.
assert_eq!(lines.len(), 1, "expected one MRF document, got: {text}");
let doc: serde_json::Value = serde_json::from_str(lines[0]).expect("each line should be a JSON document");
assert_eq!(doc["bucket"], "bucket-a");
assert_eq!(doc["object"], "object-a");
assert_eq!(doc["versionId"], uuid::Uuid::from_u128(7).to_string());
assert_eq!(doc["retryCount"], 2);
// madmin `ReplicationMRF` has a `nodeName` tag; the durable backlog is
// cluster-shared, so RustFS reports an empty node name.
assert_eq!(doc["nodeName"], "");
// The envelope keys must not leak into the stream: a `"Bucket"` key
// would case-insensitively populate `ReplicationMRF.Bucket` and render
// a phantom row in `mc replicate backlog`.
assert!(doc.get("Bucket").is_none());
assert!(doc.get("Targets").is_none());
}
/// An empty backlog must produce an empty body: madmin's decoder loop then
/// terminates on io.EOF with zero rows instead of one phantom row.
#[test]
fn mrf_stream_renders_empty_body_for_no_entries() {
let durable = DurableMrfBacklog {
available: true,
entries: Vec::new(),
};
let response = build_mrf_response("bucket-a".to_string(), &BucketStats::default(), &durable);
let (body, _) = render_mrf_backlog(&response, &durable, false).expect("stream body should serialize");
assert!(
body.is_empty(),
"empty backlog must serialize to an empty body, got: {}",
String::from_utf8_lossy(&body)
);
}
/// `?aggregate=true` (RustFS extension) keeps the enveloped counter shape.
#[test]
fn mrf_aggregate_envelope_retains_counters() {
let durable = sample_durable_backlog();
let response = build_mrf_response("bucket-a".to_string(), &BucketStats::default(), &durable);
let (body, _) = render_mrf_backlog(&response, &durable, true).expect("aggregate body should serialize");
let json: serde_json::Value = serde_json::from_slice(&body).expect("aggregate body should be one JSON object");
assert_eq!(json["Bucket"], "bucket-a");
assert_eq!(json["DurableCount"], 1);
assert_eq!(json["DurableBacklogAvailable"], true);
// The bare stream is an enumerable per-object API, so the aggregate
// shell now truthfully advertises it whenever the backlog is readable.
assert_eq!(json["PerObjectEntriesAvailable"], true);
}
/// The nil UUID is RustFS's in-memory null-version sentinel; the wire
/// token is `null`, never the zero UUID (second review round).
#[test]
fn mrf_stream_maps_nil_version_to_null_token() {
let durable = DurableMrfBacklog {
available: true,
entries: vec![MrfReplicateEntry {
bucket: "bucket-a".to_string(),
object: "null-version-object".to_string(),
version_id: Some(uuid::Uuid::nil()),
retry_count: 1,
size: 10,
op: MrfOpKind::Object,
..Default::default()
}],
};
let response = build_mrf_response("bucket-a".to_string(), &BucketStats::default(), &durable);
let (body, truncated) = render_mrf_backlog(&response, &durable, false).expect("stream body should serialize");
assert!(!truncated);
let doc: serde_json::Value =
serde_json::from_str(String::from_utf8(body).expect("utf-8").lines().next().expect("one line"))
.expect("line should be a JSON document");
assert_eq!(doc["versionId"], "null");
}
/// The durable ledger is not bounded by the in-memory pending cap; the
/// stream must stop at the documented bound and signal truncation
/// (second review round).
#[test]
fn mrf_stream_truncates_at_the_documented_bound() {
let entries = (0..super::REPLICATION_MRF_MAX_STREAM_ENTRIES + 1)
.map(|index| MrfReplicateEntry {
bucket: "bucket-a".to_string(),
object: format!("object-{index}"),
retry_count: 1,
op: MrfOpKind::Object,
..Default::default()
})
.collect();
let durable = DurableMrfBacklog {
available: true,
entries,
};
let response = build_mrf_response("bucket-a".to_string(), &BucketStats::default(), &durable);
let (body, truncated) = render_mrf_backlog(&response, &durable, false).expect("stream body should serialize");
assert!(truncated, "one entry past the bound must signal truncation");
assert_eq!(
String::from_utf8(body).expect("utf-8").lines().count(),
super::REPLICATION_MRF_MAX_STREAM_ENTRIES
);
let error = super::ensure_complete_mrf_stream(truncated).expect_err("partial streams must not return 200");
assert_eq!(error.code(), &s3s::S3ErrorCode::ServiceUnavailable);
}
#[test]
fn test_extract_query_params_decodes_percent_encoded_values() {
let uri: Uri = "/rustfs/admin/v3/list-remote-targets?bucket=foo%2Fbar&flag=a+b"
File diff suppressed because it is too large Load Diff
@@ -29,6 +29,6 @@ impl Operation for RestLoadCredentialsHandler {
let issuer = IamTableCredentialIssuer::from_request(&req)?;
let response =
load_credentials_response(&store, &warehouse, &namespace, &table, &issuer, Some(&principal.credentials)).await?;
build_sensitive_json_response(StatusCode::OK, &response)
build_json_response(StatusCode::OK, &response)
}
}
File diff suppressed because it is too large Load Diff
@@ -125,9 +125,7 @@ impl Operation for RestLoadTableHandler {
ensure_table_bucket_enabled_from_extensions(&req.extensions, &warehouse).await?;
let metadata_backend = table_catalog_backend_from_extensions(&req.extensions)?;
let store = table_catalog_store_from_backend(metadata_backend.clone())?;
let snapshot_selection = rest_table_snapshot_selection_from_query(&req.uri)?;
let mut response = load_table_response(&store, &metadata_backend, &warehouse, &namespace, &table).await?;
apply_rest_table_snapshot_selection(&mut response.metadata, snapshot_selection);
let response = load_table_response(&store, &metadata_backend, &warehouse, &namespace, &table).await?;
build_json_response(StatusCode::OK, &response)
}
}
@@ -160,7 +158,7 @@ impl Operation for RestCommitTableHandler {
let principal = authorize_table_catalog_resource_request(&req, &resource, AdminAction::CommitTableAction).await?;
install_table_catalog_s3_request_info(&mut req, &principal)?;
ensure_table_bucket_enabled_from_extensions(&req.extensions, &warehouse).await?;
let request = read_rest_commit_table_request(std::mem::take(&mut req.input)).await?;
let request = read_json_body::<RestCommitTableRequest>(std::mem::take(&mut req.input)).await?;
let metadata_backend = table_catalog_backend_from_extensions(&req.extensions)?;
let store = table_catalog_store_from_backend(metadata_backend.clone())?;
let commit_backend = TableCommitObjectBackend::for_request(metadata_backend, req);
@@ -180,14 +178,6 @@ impl Operation for RestDropTableHandler {
let table = table_name_from_params(&params)?;
let resource = TableCatalogResource::table(&warehouse, &namespace, &table);
authorize_table_catalog_resource_request(&req, &resource, AdminAction::DeleteTableAction).await?;
let purge_requested = rest_purge_requested_from_query(&req.uri)?;
if purge_requested {
return Err(iceberg_rest_error(
ICEBERG_ERROR_UNSUPPORTED_OPERATION,
StatusCode::NOT_ACCEPTABLE,
"purgeRequested=true is not supported",
));
}
ensure_table_bucket_enabled_from_extensions(&req.extensions, &warehouse).await?;
let store = table_catalog_store_from_extensions(&req.extensions)?;
drop_table_in_store(&store, &warehouse, &namespace, &table).await?;
File diff suppressed because it is too large Load Diff
@@ -43,9 +43,8 @@ impl Operation for RestCreateViewHandler {
let metadata_backend = table_catalog_backend_from_extensions(&req.extensions)?;
let store = table_catalog_store_from_backend(metadata_backend.clone())?;
let table_bucket_enabled = table_bucket_enabled_from_extensions(&req.extensions, &warehouse).await?;
let publication_backend = TableCommitObjectBackend::preauthorized(metadata_backend);
let response =
create_view_response(&store, &publication_backend, &warehouse, &namespace, request, table_bucket_enabled).await?;
create_view_response(&store, &metadata_backend, &warehouse, &namespace, request, table_bucket_enabled).await?;
build_json_response(StatusCode::OK, &response)
}
}
@@ -88,20 +87,17 @@ pub struct RestReplaceViewHandler {}
#[async_trait::async_trait]
impl Operation for RestReplaceViewHandler {
async fn call(&self, mut req: S3Request<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
async fn call(&self, req: S3Request<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
let warehouse = warehouse_from_params(&params)?;
let namespace = namespace_from_params(&params)?;
let view = view_name_from_params(&params)?;
let resource = TableCatalogResource::view(&warehouse, &namespace, &view);
let principal = authorize_table_catalog_resource_request(&req, &resource, AdminAction::CommitTableAction).await?;
install_table_catalog_s3_request_info(&mut req, &principal)?;
authorize_table_catalog_resource_request(&req, &resource, AdminAction::CommitTableAction).await?;
ensure_table_bucket_enabled_from_extensions(&req.extensions, &warehouse).await?;
let request = read_rest_commit_view_request(std::mem::take(&mut req.input)).await?;
let request = read_json_body::<RestCommitViewRequest>(req.input).await?;
let metadata_backend = table_catalog_backend_from_extensions(&req.extensions)?;
let store = table_catalog_store_from_backend(metadata_backend.clone())?;
let commit_backend = TableCommitObjectBackend::for_request(metadata_backend, req);
let result = replace_view_response(&store, &commit_backend, &warehouse, &namespace, &view, request).await;
let response = commit_backend.finish(result).await?;
let response = replace_view_response(&store, &metadata_backend, &warehouse, &namespace, &view, request).await?;
build_json_response(StatusCode::OK, &response)
}
}
-1
View File
@@ -17,7 +17,6 @@ mod auth;
pub mod console;
pub mod handlers;
mod plugin_contract;
pub(crate) mod replication_metrics_wire;
// Contract inventory is validated by tests before later runtime integration.
#[allow(dead_code)]
pub(crate) mod route_policy;
@@ -1,673 +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.
//! Serialize-only wire projections of the internal replication statistics
//! onto the minio-go `replication.Metrics` / `replication.MetricsV2` json
//! shapes consumed by `mc replicate status` (`?replication-metrics[=2]` and
//! the admin `replicationmetrics` endpoint).
//!
//! Red line: the internal `BucketStats` family in
//! `crates/replication/src/stats.rs` is ALSO the intra-cluster peer-RPC wire
//! format — `node_service.rs` encodes it with `rmp_serde::to_vec_named`, so
//! its Rust field names travel between nodes as msgpack map keys. Renaming
//! those serde names would break mixed-version clusters mid rolling upgrade.
//! All madmin/minio-go interop therefore happens in these DTOs; never add
//! `#[serde(rename)]` to the internal structs instead.
//!
//! Field names below are the exact json tags of minio-go
//! `pkg/replication/replication.go` (v7.0.91). Keys minio-go does not know
//! are RustFS extensions; Go decoders ignore unknown keys. `max`/`peak` are
//! both emitted for the queue peak because the MinIO server writes `max`
//! while minio-go reads `peak` (an upstream drift); emitting both keeps every
//! decoder working.
use serde::Serialize;
use std::collections::HashMap;
use std::time::Duration;
use crate::admin::storage_api::replication::{
BucketReplicationStat as InternalReplicationStat, BucketReplicationStats as InternalReplicationStats, BucketStats,
InQueueMetric as InternalInQueueMetric, XferStats as InternalXferStats,
};
/// minio-go `replication.RStat`.
#[derive(Debug, Default, Clone, Copy, Serialize)]
pub(crate) struct RStatWire {
#[serde(rename = "count")]
pub count: f64,
#[serde(rename = "bytes")]
pub bytes: i64,
}
/// minio-go `replication.TimedErrStats`.
#[derive(Debug, Default, Clone, Copy, Serialize)]
pub(crate) struct TimedErrStatsWire {
#[serde(rename = "lastMinute")]
pub last_minute: RStatWire,
#[serde(rename = "lastHour")]
pub last_hour: RStatWire,
#[serde(rename = "totals")]
pub totals: RStatWire,
}
impl TimedErrStatsWire {
fn add(self, other: TimedErrStatsWire) -> TimedErrStatsWire {
fn add(a: RStatWire, b: RStatWire) -> RStatWire {
RStatWire {
count: a.count + b.count,
bytes: a.bytes.saturating_add(b.bytes),
}
}
TimedErrStatsWire {
last_minute: add(self.last_minute, other.last_minute),
last_hour: add(self.last_hour, other.last_hour),
totals: add(self.totals, other.totals),
}
}
}
/// minio-go `replication.QStat`.
#[derive(Debug, Default, Clone, Copy, Serialize)]
pub(crate) struct QStatWire {
#[serde(rename = "count")]
pub count: f64,
#[serde(rename = "bytes")]
pub bytes: f64,
}
/// minio-go `replication.InQueueMetric`, with the queue peak emitted under
/// both `peak` (minio-go tag) and `max` (MinIO server tag).
#[derive(Debug, Default, Clone, Copy, Serialize)]
pub(crate) struct InQueueMetricWire {
#[serde(rename = "curr")]
pub curr: QStatWire,
#[serde(rename = "avg")]
pub avg: QStatWire,
#[serde(rename = "max")]
pub max: QStatWire,
#[serde(rename = "peak")]
pub peak: QStatWire,
}
impl From<&InternalInQueueMetric> for InQueueMetricWire {
fn from(metric: &InternalInQueueMetric) -> Self {
fn qstat(bytes: i64, count: i64) -> QStatWire {
QStatWire {
count: count as f64,
bytes: bytes as f64,
}
}
let peak = qstat(metric.max.bytes, metric.max.count);
InQueueMetricWire {
curr: qstat(metric.curr.bytes, metric.curr.count),
avg: qstat(metric.avg.bytes, metric.avg.count),
max: peak,
peak,
}
}
}
/// minio-go `replication.XferStats`.
#[derive(Debug, Default, Clone, Copy, Serialize)]
pub(crate) struct XferStatsWire {
#[serde(rename = "avgRate")]
pub avg_rate: f64,
#[serde(rename = "peakRate")]
pub peak_rate: f64,
#[serde(rename = "currRate")]
pub curr_rate: f64,
}
#[derive(Default)]
struct XferStatsAverage {
sum: XferStatsWire,
active: u32,
}
impl XferStatsAverage {
fn add_active(&mut self, stats: XferStatsWire) {
if stats.peak_rate <= 0.0 {
return;
}
self.add_raw(stats);
self.active += 1;
}
fn add_raw(&mut self, stats: XferStatsWire) {
self.sum.avg_rate += stats.avg_rate;
self.sum.curr_rate += stats.curr_rate;
self.sum.peak_rate = self.sum.peak_rate.max(stats.peak_rate);
}
fn finish(self) -> XferStatsWire {
let active = self.active;
self.finish_with_divisor(active)
}
fn finish_with_divisor(self, divisor: u32) -> XferStatsWire {
if divisor == 0 {
return self.sum;
}
XferStatsWire {
avg_rate: self.sum.avg_rate / f64::from(divisor),
peak_rate: self.sum.peak_rate,
curr_rate: self.sum.curr_rate / f64::from(divisor),
}
}
}
impl From<&InternalXferStats> for XferStatsWire {
fn from(stats: &InternalXferStats) -> Self {
XferStatsWire {
avg_rate: stats.avg,
peak_rate: stats.peak,
curr_rate: stats.curr,
}
}
}
/// minio-go `replication.WorkerStat`. RustFS does not track per-bucket worker
/// occupancy yet, so this always reports zeros.
#[derive(Debug, Default, Clone, Copy, Serialize)]
pub(crate) struct WorkerStatWire {
#[serde(rename = "curr")]
pub curr: i32,
#[serde(rename = "avg")]
pub avg: f32,
#[serde(rename = "max")]
pub max: i32,
}
/// minio-go `replication.ReplMRFStats`. RustFS does not track the 5-minute /
/// dropped MRF windows, so this always reports zeros; the durable backlog is
/// enumerable via `/v3/replication/mrf` instead.
#[derive(Debug, Default, Clone, Copy, Serialize)]
pub(crate) struct ReplMrfStatsWire {
#[serde(rename = "failedCount_last5min")]
pub last_failed_count: u64,
#[serde(rename = "droppedCount_since_uptime")]
pub total_dropped_count: u64,
#[serde(rename = "droppedBytes_since_uptime")]
pub total_dropped_bytes: u64,
}
/// minio-go `replication.CounterSummary`.
#[derive(Debug, Default, Clone, Copy, Serialize)]
pub(crate) struct CounterSummaryWire {
#[serde(rename = "last1hr")]
pub last1hr: u64,
#[serde(rename = "last1m")]
pub last1m: u64,
#[serde(rename = "total")]
pub total: u64,
}
/// minio-go `replication.TargetMetrics` (one remote target / ARN).
#[derive(Debug, Default, Serialize)]
pub(crate) struct TargetMetricsWire {
#[serde(rename = "replicationCount")]
pub replicated_count: i64,
#[serde(rename = "completedReplicationSize")]
pub replicated_size: i64,
/// Bandwidth limit for this target. The tag says "bits" but both MinIO
/// and minio-go treat the value as bytes/sec; keep bytes/sec.
#[serde(rename = "limitInBits")]
pub bandwidth_limit_bytes_per_sec: i64,
#[serde(rename = "currentBandwidth")]
pub current_bandwidth_bytes_per_sec: f64,
#[serde(rename = "failed")]
pub failed: TimedErrStatsWire,
#[serde(rename = "failedReplicationSize")]
pub failed_size: i64,
#[serde(rename = "failedReplicationCount")]
pub failed_count: i64,
}
fn target_timed_err_stats(stat: &InternalReplicationStat) -> TimedErrStatsWire {
// Cluster aggregation merges FailStats without the process-local samples,
// so the serializable window snapshots (refreshed at each node's
// collection point, summed by merge) are authoritative here; the live
// samples only ever agree with or lag them, so take the larger.
let sampled_minute = stat.fail_stats.recent_since(Duration::from_secs(60));
let sampled_hour = stat.fail_stats.recent_since(Duration::from_secs(3600));
let window = |sampled_count: i64, sampled_size: i64, snapshot_count: i64, snapshot_size: i64| RStatWire {
count: sampled_count.max(snapshot_count) as f64,
bytes: sampled_size.max(snapshot_size),
};
TimedErrStatsWire {
last_minute: window(
sampled_minute.count,
sampled_minute.size,
stat.fail_stats.last_minute.count,
stat.fail_stats.last_minute.size,
),
last_hour: window(
sampled_hour.count,
sampled_hour.size,
stat.fail_stats.last_hour.count,
stat.fail_stats.last_hour.size,
),
totals: RStatWire {
count: stat.failed.count as f64,
bytes: stat.failed.size,
},
}
}
impl From<&InternalReplicationStat> for TargetMetricsWire {
fn from(stat: &InternalReplicationStat) -> Self {
TargetMetricsWire {
replicated_count: stat.replicated_count,
replicated_size: stat.replicated_size,
bandwidth_limit_bytes_per_sec: stat.bandwidth_limit_bytes_per_sec,
current_bandwidth_bytes_per_sec: stat.current_bandwidth_bytes_per_sec,
failed: target_timed_err_stats(stat),
failed_size: stat.failed.size,
failed_count: stat.failed.count,
}
}
}
/// minio-go `replication.Metrics` — the `currStats` member of `MetricsV2` and
/// the whole v1 response body. The trailing snake_case fields are RustFS
/// source-health extension keys (ignored by Go decoders) carried over from
/// the previous response shape.
#[derive(Debug, Default, Serialize)]
pub(crate) struct MetricsWire {
#[serde(rename = "Stats")]
pub stats: HashMap<String, TargetMetricsWire>,
#[serde(rename = "completedReplicationSize")]
pub replicated_size: i64,
#[serde(rename = "replicaSize")]
pub replica_size: i64,
#[serde(rename = "replicaCount")]
pub replica_count: i64,
#[serde(rename = "replicationCount")]
pub replicated_count: i64,
#[serde(rename = "failed")]
pub failed: TimedErrStatsWire,
#[serde(rename = "queued")]
pub queued: InQueueMetricWire,
// RustFS extension keys (source health of the aggregation).
pub provider_available: bool,
pub cluster_complete: bool,
pub observed_node_count: u32,
pub expected_node_count: u32,
}
impl From<&InternalReplicationStats> for MetricsWire {
fn from(stats: &InternalReplicationStats) -> Self {
let mut failed = TimedErrStatsWire::default();
let mut targets = HashMap::with_capacity(stats.stats.len());
for (arn, stat) in &stats.stats {
let target = TargetMetricsWire::from(stat);
failed = failed.add(target.failed);
targets.insert(arn.clone(), target);
}
MetricsWire {
stats: targets,
replicated_size: stats.replicated_size,
replica_size: stats.replica_size,
replica_count: stats.replica_count,
replicated_count: stats.replicated_count,
failed,
queued: InQueueMetricWire::from(&stats.q_stat),
provider_available: stats.provider_available,
cluster_complete: stats.cluster_complete,
observed_node_count: stats.observed_node_count,
expected_node_count: stats.expected_node_count,
}
}
}
/// minio-go `replication.ReplQNodeStats`.
#[derive(Debug, Default, Serialize)]
pub(crate) struct ReplQNodeStatsWire {
#[serde(rename = "nodeName")]
pub node_name: String,
#[serde(rename = "uptime")]
pub uptime: i64,
#[serde(rename = "activeWorkers")]
pub workers: WorkerStatWire,
#[serde(rename = "transferSummary")]
pub xfer_stats: XferSummaryWire,
#[serde(rename = "tgtTransferStats")]
pub tgt_xfer_stats: TargetXferSummaryWire,
#[serde(rename = "queueStats")]
pub q_stats: InQueueMetricWire,
#[serde(rename = "mrfStats")]
pub mrf_stats: ReplMrfStatsWire,
#[serde(rename = "retries")]
pub retries: CounterSummaryWire,
#[serde(rename = "errors")]
pub errors: CounterSummaryWire,
}
/// minio-go `replication.ReplQueueStats`.
#[derive(Debug, Default, Serialize)]
pub(crate) struct ReplQueueStatsWire {
#[serde(rename = "nodes")]
pub nodes: Vec<ReplQNodeStatsWire>,
}
/// minio-go `replication.MetricsV2` — the `?replication-metrics=2` body.
#[derive(Debug, Default, Serialize)]
pub(crate) struct MetricsV2Wire {
#[serde(rename = "uptime")]
pub uptime: i64,
#[serde(rename = "currStats")]
pub current_stats: MetricsWire,
#[serde(rename = "queueStats")]
pub queue_stats: ReplQueueStatsWire,
#[serde(rename = "downtimeInfo")]
pub downtime_info: HashMap<String, serde_json::Value>,
}
/// `transferSummary` map keyed by minio-go `MetricName` (Large/Small/Total).
type XferSummaryWire = HashMap<&'static str, XferStatsWire>;
/// `tgtTransferStats` map keyed by target ARN.
type TargetXferSummaryWire = HashMap<String, XferSummaryWire>;
fn transfer_summaries(stats: &InternalReplicationStats) -> (XferSummaryWire, TargetXferSummaryWire) {
let mut per_target: TargetXferSummaryWire = HashMap::new();
let mut large_summary = XferStatsAverage::default();
let mut small_summary = XferStatsAverage::default();
let mut total_summary = XferStatsAverage::default();
let mut active_targets = 0;
for (arn, stat) in &stats.stats {
let large = XferStatsWire::from(&stat.xfer_rate_lrg);
let small = XferStatsWire::from(&stat.xfer_rate_sml);
let mut target_total = XferStatsAverage::default();
target_total.add_active(large);
target_total.add_active(small);
let total = target_total.finish();
per_target.insert(arn.clone(), HashMap::from([("Large", large), ("Small", small), ("Total", total)]));
if large.peak_rate > 0.0 || small.peak_rate > 0.0 {
active_targets += 1;
large_summary.add_raw(large);
small_summary.add_raw(small);
total_summary.add_raw(large);
total_summary.add_raw(small);
}
}
let summary = HashMap::from([
("Large", large_summary.finish_with_divisor(active_targets)),
("Small", small_summary.finish_with_divisor(active_targets)),
("Total", total_summary.finish_with_divisor(active_targets)),
]);
(summary, per_target)
}
impl MetricsV2Wire {
/// Project the aggregated internal stats onto the `MetricsV2` shape.
///
/// The aggregation path leaves `queue_stats.nodes` empty today, so a
/// single node entry is synthesized from the bucket queue snapshot —
/// `mc replicate status` derives its queue/worker panels from
/// `queueStats.nodes` and treats an empty list as "no data".
pub(crate) fn from_stats(bucket_stats: &BucketStats, node_name: &str) -> Self {
let (xfer_stats, tgt_xfer_stats) = transfer_summaries(&bucket_stats.replication_stats);
let mut nodes: Vec<ReplQNodeStatsWire> = bucket_stats
.queue_stats
.nodes
.iter()
.map(|node| ReplQNodeStatsWire {
node_name: node_name.to_string(),
uptime: bucket_stats.uptime,
q_stats: InQueueMetricWire::from(&node.q_stats),
..Default::default()
})
.collect();
if nodes.is_empty() {
nodes.push(ReplQNodeStatsWire {
node_name: node_name.to_string(),
uptime: bucket_stats.uptime,
q_stats: InQueueMetricWire::from(&bucket_stats.replication_stats.q_stat),
xfer_stats: xfer_stats.clone(),
tgt_xfer_stats: tgt_xfer_stats.clone(),
..Default::default()
});
} else {
// Attach the transfer summaries to the first node; the internal
// snapshot does not attribute transfer rates per node.
if let Some(first) = nodes.first_mut() {
first.xfer_stats = xfer_stats.clone();
first.tgt_xfer_stats = tgt_xfer_stats.clone();
}
}
MetricsV2Wire {
uptime: bucket_stats.uptime,
current_stats: MetricsWire::from(&bucket_stats.replication_stats),
queue_stats: ReplQueueStatsWire { nodes },
downtime_info: HashMap::new(),
}
}
}
#[cfg(test)]
mod tests {
use super::*;
fn sample_bucket_stats() -> BucketStats {
let mut stats = BucketStats {
uptime: 42,
..Default::default()
};
stats.replication_stats.replica_count = 2;
stats.replication_stats.replica_size = 128;
stats.replication_stats.replicated_count = 9;
stats.replication_stats.replicated_size = 4096;
let target = stats
.replication_stats
.stats
.entry("arn:minio:replication::t:b".to_string())
.or_default();
target.replicated_count = 9;
target.replicated_size = 4096;
target.failed.count = 3;
target.failed.size = 900;
target.bandwidth_limit_bytes_per_sec = 1024;
target.current_bandwidth_bytes_per_sec = 512.5;
stats
.replication_stats
.q_stat
.curr
.now_count
.store(4, std::sync::atomic::Ordering::Relaxed);
stats
.replication_stats
.q_stat
.curr
.now_bytes
.store(1200, std::sync::atomic::Ordering::Relaxed);
stats.replication_stats.q_stat = stats.replication_stats.q_stat.snapshot();
stats
}
#[test]
fn metrics_wire_matches_minio_go_tags() {
let stats = sample_bucket_stats();
let json = serde_json::to_value(MetricsWire::from(&stats.replication_stats)).expect("v1 wire should serialize");
assert_eq!(json["replicaCount"], 2);
assert_eq!(json["replicaSize"], 128);
assert_eq!(json["replicationCount"], 9);
assert_eq!(json["completedReplicationSize"], 4096);
assert_eq!(json["queued"]["curr"]["count"], 4.0);
assert_eq!(json["queued"]["curr"]["bytes"], 1200.0);
let target = &json["Stats"]["arn:minio:replication::t:b"];
assert_eq!(target["replicationCount"], 9);
assert_eq!(target["completedReplicationSize"], 4096);
assert_eq!(target["limitInBits"], 1024);
assert_eq!(target["currentBandwidth"], 512.5);
// failed is the madmin TimedErrStats envelope, not the internal
// {count,size} pair.
assert_eq!(target["failed"]["totals"]["count"], 3.0);
assert_eq!(target["failed"]["totals"]["bytes"], 900);
assert!(target["failed"].get("count").is_none());
// Aggregate failed mirrors the per-target totals.
assert_eq!(json["failed"]["totals"]["count"], 3.0);
}
#[test]
fn metrics_v2_wire_synthesizes_queue_node() {
let stats = sample_bucket_stats();
let json = serde_json::to_value(MetricsV2Wire::from_stats(&stats, "node-1:9000")).expect("v2 wire should serialize");
assert_eq!(json["uptime"], 42);
assert_eq!(json["currStats"]["replicaCount"], 2);
let node = &json["queueStats"]["nodes"][0];
assert_eq!(node["nodeName"], "node-1:9000");
assert_eq!(node["uptime"], 42);
assert_eq!(node["queueStats"]["curr"]["count"], 4.0);
// The queue peak is emitted under both the minio-go tag (`peak`) and
// the MinIO server tag (`max`).
assert_eq!(node["queueStats"]["peak"], node["queueStats"]["max"]);
assert!(node["activeWorkers"].get("curr").is_some());
assert!(node["transferSummary"].get("Total").is_some());
assert_eq!(json["downtimeInfo"], serde_json::json!({}));
}
/// minio-go's transferSummary labels mean >= 128 MiB for Large; the
/// producer must bin on the same boundary (MIN_LARGE_OBJ_SIZE, shared
/// with the worker-pool split), or a 2 MiB replication shows under Large
/// while Small stays zero.
#[test]
fn transfer_summary_bins_on_the_128_mib_boundary() {
const MIB: i64 = 1024 * 1024;
let mut stats = BucketStats::default();
let stat = stats
.replication_stats
.stats
.entry("arn:minio:replication::t:b".to_string())
.or_default();
stat.update_xfer_rate(2 * MIB, std::time::Duration::from_secs(1));
stat.update_xfer_rate(127 * MIB, std::time::Duration::from_secs(1));
stat.update_xfer_rate(128 * MIB, std::time::Duration::from_secs(1));
let json = serde_json::to_value(MetricsV2Wire::from_stats(&stats, "node-1")).expect("v2 wire should serialize");
let summary = &json["queueStats"]["nodes"][0]["tgtTransferStats"]["arn:minio:replication::t:b"];
let small_peak = summary["Small"]["peakRate"].as_f64().expect("Small peakRate");
let large_peak = summary["Large"]["peakRate"].as_f64().expect("Large peakRate");
assert!(
(small_peak - (127 * MIB) as f64).abs() < 1.0,
"2 MiB and 127 MiB transfers must bin as Small (peak {small_peak})"
);
assert!(
(large_peak - (128 * MIB) as f64).abs() < 1.0,
"exactly 128 MiB must bin as Large (peak {large_peak})"
);
}
#[test]
fn transfer_summaries_average_active_bins_and_targets() {
let mut stats = BucketStats::default();
let first = stats.replication_stats.stats.entry("target-a".to_string()).or_default();
first.xfer_rate_sml.avg = 50.0;
first.xfer_rate_sml.curr = 40.0;
first.xfer_rate_sml.peak = 60.0;
first.xfer_rate_lrg.avg = 100.0;
first.xfer_rate_lrg.curr = 80.0;
first.xfer_rate_lrg.peak = 120.0;
let second = stats.replication_stats.stats.entry("target-b".to_string()).or_default();
second.xfer_rate_sml.avg = 30.0;
second.xfer_rate_sml.curr = 20.0;
second.xfer_rate_sml.peak = 40.0;
let json = serde_json::to_value(MetricsV2Wire::from_stats(&stats, "node-1")).expect("v2 wire should serialize");
let node = &json["queueStats"]["nodes"][0];
let target_a = &node["tgtTransferStats"]["target-a"]["Total"];
assert_eq!(target_a["avgRate"], 75.0);
assert_eq!(target_a["currRate"], 60.0);
assert_eq!(target_a["peakRate"], 120.0);
let summary = &node["transferSummary"];
assert_eq!(summary["Small"]["avgRate"], 40.0);
assert_eq!(summary["Small"]["currRate"], 30.0);
assert_eq!(summary["Large"]["avgRate"], 50.0);
assert_eq!(summary["Total"]["avgRate"], 90.0);
assert_eq!(summary["Total"]["currRate"], 70.0);
assert_eq!(summary["Total"]["peakRate"], 120.0);
}
/// Review regression: both metrics endpoints aggregate first, and the
/// FailStats merge drops the process-local samples — the rolling windows
/// must survive a peer-RPC round trip plus aggregation and still reach
/// the wire body.
#[test]
fn failure_windows_survive_aggregation_before_serialization() {
// Node A: live failure; the windows are stamped at the collection
// point (get_latest_replication_stats calls refresh_windows before
// the stats cross the wire), never on the failure hot path.
let mut node_a = crate::admin::storage_api::replication::BucketReplicationStat::default();
node_a.fail_stats.add_size(512, None::<&std::io::Error>);
node_a.fail_stats.refresh_windows();
node_a.failed = node_a.fail_stats.to_metric();
// Node A's stats cross the peer RPC wire: the samples are dropped,
// the window snapshots travel.
let encoded = rmp_serde::to_vec_named(&node_a).expect("stat should encode");
let remote: crate::admin::storage_api::replication::BucketReplicationStat =
rmp_serde::from_slice(&encoded).expect("stat should decode");
// Aggregation merges the remote stat with an empty local one.
let merged_fail = remote.fail_stats.merge(&Default::default());
let aggregated = crate::admin::storage_api::replication::BucketReplicationStat {
failed: merged_fail.to_metric(),
fail_stats: merged_fail,
..Default::default()
};
let mut stats = BucketStats::default();
stats
.replication_stats
.stats
.insert("arn:minio:replication::t:b".to_string(), aggregated);
let json = serde_json::to_value(MetricsWire::from(&stats.replication_stats)).expect("wire should serialize");
let failed = &json["Stats"]["arn:minio:replication::t:b"]["failed"];
assert_eq!(failed["totals"]["count"], 1.0);
assert_eq!(
failed["lastMinute"]["count"], 1.0,
"the rolling minute window must survive RPC + aggregation"
);
assert_eq!(failed["lastMinute"]["bytes"], 512);
assert_eq!(failed["lastHour"]["count"], 1.0);
}
/// Pin the intra-cluster peer-RPC wire format of the internal stats: it
/// is msgpack with the Rust field names as map keys
/// (`rmp_serde::to_vec_named` in node_service.rs). If someone "fixes"
/// the interop bug by renaming the internal serde fields instead of using
/// these DTOs, this test fails and points them here.
#[test]
fn internal_bucket_stats_rpc_wire_stays_snake_case() {
let stats = sample_bucket_stats();
let encoded = rmp_serde::to_vec_named(&stats).expect("internal stats should encode");
let value: serde_json::Value = rmp_serde::from_slice(&encoded).expect("named msgpack should decode generically");
assert!(
value.get("replication_stats").is_some(),
"peer RPC key replication_stats must not be renamed"
);
assert!(value["replication_stats"].get("q_stat").is_some());
assert!(value.get("queue_stats").is_some());
assert!(value.get("proxy_stats").is_some());
let decoded: BucketStats = rmp_serde::from_slice(&encoded).expect("round-trip through the peer RPC wire");
assert_eq!(decoded.replication_stats.replica_count, 2);
}
}
+1 -4
View File
@@ -1459,13 +1459,10 @@ pub const ADMIN_ROUTE_POLICY_SPECS: &[AdminRouteSpec] = &[
REPLICATION_DIFF,
RouteRiskLevel::Sensitive,
),
// The default stream enumerates object names/version ids and requires
// ReplicationDiff (MinIO parity); only ?aggregate=true relaxes to
// GetReplicationMetrics in the handler.
admin(
HttpMethod::Get,
"/rustfs/admin/v3/replication/mrf",
REPLICATION_DIFF,
GET_REPLICATION_METRICS,
RouteRiskLevel::Sensitive,
),
];
+13 -67
View File
@@ -1548,8 +1548,7 @@ async fn build_replication_metrics_response(
let bucket_stats = apply_replication_metrics_bandwidth_report(bucket_stats, collect_replication_metrics_bandwidth(bucket));
let bucket_stats = apply_replication_metrics_runtime_fields(bucket_stats, route, replication_metrics_uptime_seconds());
let node_name = crate::runtime_sources::current_local_node_name().await.unwrap_or_default();
let body = serialize_replication_metrics_body(&bucket_stats, route, &node_name)?;
let body = serialize_replication_metrics_body(&bucket_stats, route)?;
let mut resp = S3Response::with_status(Body::from(body), StatusCode::OK);
resp.headers
@@ -1609,24 +1608,12 @@ fn apply_replication_metrics_runtime_fields(
bucket_stats
}
/// Serialize the metrics body in the minio-go wire shapes
/// (`replication.Metrics` for v1, `replication.MetricsV2` for v2). The
/// internal `BucketStats` serde names are the intra-cluster peer-RPC wire
/// format and must never appear here — see
/// `crate::admin::replication_metrics_wire`.
fn serialize_replication_metrics_body(
bucket_stats: &BucketStats,
route: ReplicationExtRoute,
node_name: &str,
) -> S3Result<Vec<u8>> {
use crate::admin::replication_metrics_wire::{MetricsV2Wire, MetricsWire};
fn serialize_replication_metrics_body(bucket_stats: &BucketStats, route: ReplicationExtRoute) -> S3Result<Vec<u8>> {
match route {
ReplicationExtRoute::MetricsV1 => {
serde_json::to_vec(&MetricsWire::from(&bucket_stats.replication_stats)).map_err(|e| s3_error!(InternalError, "{e}"))
}
ReplicationExtRoute::MetricsV2 => {
serde_json::to_vec(&MetricsV2Wire::from_stats(bucket_stats, node_name)).map_err(|e| s3_error!(InternalError, "{e}"))
serde_json::to_vec(&bucket_stats.replication_stats).map_err(|e| s3_error!(InternalError, "{e}"))
}
ReplicationExtRoute::MetricsV2 => serde_json::to_vec(bucket_stats).map_err(|e| s3_error!(InternalError, "{e}")),
ReplicationExtRoute::Check | ReplicationExtRoute::ResetStart | ReplicationExtRoute::ResetStatus => {
Err(s3_error!(InternalError, "invalid route for metrics response"))
}
@@ -4160,37 +4147,22 @@ mod tests {
assert!(err.message().unwrap_or_default().contains("rule-stale"));
}
/// The v1 body must decode into minio-go `replication.Metrics` (exact
/// json tags); Go's decoder matches case-insensitively but does not
/// ignore underscores, so the internal snake_case names read as all-zero.
#[test]
fn serialize_replication_metrics_body_v1_returns_minio_go_metrics_shape() {
fn serialize_replication_metrics_body_v1_returns_replication_stats_only() {
let mut stats = BucketStats {
uptime: 99,
..Default::default()
};
stats.replication_stats.replica_count = 7;
stats.replication_stats.replicated_size = 2048;
stats
.replication_stats
.stats
.entry("arn:minio:replication::t:b".to_string())
.or_default()
.replicated_count = 5;
stats.proxy_stats.put_total = 3;
let body = serialize_replication_metrics_body(&stats, ReplicationExtRoute::MetricsV1, "node-1:9000")
.expect("metrics v1 body should serialize");
let body =
serialize_replication_metrics_body(&stats, ReplicationExtRoute::MetricsV1).expect("metrics v1 body should serialize");
let payload: serde_json::Value = serde_json::from_slice(&body).expect("body should be json");
assert_eq!(payload["replicaCount"], 7);
assert_eq!(payload["completedReplicationSize"], 2048);
assert_eq!(payload["Stats"]["arn:minio:replication::t:b"]["replicationCount"], 5);
assert_eq!(payload["replica_count"], 7);
assert!(payload.get("uptime").is_none());
assert!(payload.get("proxy_stats").is_none());
// The internal snake_case names must not leak into the wire body.
assert!(payload.get("replica_count").is_none());
assert!(payload.get("q_stat").is_none());
}
#[test]
@@ -4276,48 +4248,22 @@ mod tests {
assert_eq!(target.current_bandwidth_bytes_per_sec, 3000.0);
}
/// The v2 body must decode into minio-go `replication.MetricsV2`
/// (`uptime`/`currStats`/`queueStats`); `mc replicate status` reads
/// `currStats` and `queueStats.nodes` and silently shows zeros when the
/// keys do not match.
#[test]
fn serialize_replication_metrics_body_v2_returns_minio_go_metrics_v2_shape() {
fn serialize_replication_metrics_body_v2_returns_full_bucket_stats() {
let mut stats = BucketStats {
uptime: 99,
..Default::default()
};
stats.replication_stats.replica_count = 7;
stats
.replication_stats
.q_stat
.curr
.now_count
.store(4, std::sync::atomic::Ordering::Relaxed);
stats
.replication_stats
.q_stat
.curr
.now_bytes
.store(1200, std::sync::atomic::Ordering::Relaxed);
stats.replication_stats.q_stat = stats.replication_stats.q_stat.snapshot();
stats.proxy_stats.put_total = 3;
let body = serialize_replication_metrics_body(&stats, ReplicationExtRoute::MetricsV2, "node-1:9000")
.expect("metrics v2 body should serialize");
let body =
serialize_replication_metrics_body(&stats, ReplicationExtRoute::MetricsV2).expect("metrics v2 body should serialize");
let payload: serde_json::Value = serde_json::from_slice(&body).expect("body should be json");
assert_eq!(payload["uptime"], 99);
assert_eq!(payload["currStats"]["replicaCount"], 7);
assert_eq!(payload["currStats"]["queued"]["curr"]["count"], 4.0);
// The queue snapshot must surface at least one node: mc derives the
// worker/queue panels from queueStats.nodes and treats an empty list
// as "no data".
assert_eq!(payload["queueStats"]["nodes"][0]["queueStats"]["curr"]["count"], 4.0);
assert_eq!(payload["queueStats"]["nodes"][0]["uptime"], 99);
// The internal snake_case names must not leak into the wire body.
assert!(payload.get("replication_stats").is_none());
assert!(payload.get("queue_stats").is_none());
assert!(payload.get("proxy_stats").is_none());
assert_eq!(payload["replication_stats"]["replica_count"], 7);
assert_eq!(payload["proxy_stats"]["put_total"], 3);
}
#[test]
-32
View File
@@ -321,34 +321,6 @@ pub(crate) mod metadata_sys {
crate::storage::storage_api::acquire_bucket_metadata_transaction_lock(bucket).await
}
pub(crate) async fn acquire_bucket_metadata_transaction_lock_for_incarnation(
bucket: &str,
expected_incarnation_id: uuid::Uuid,
) -> Result<super::ecstore_bucket::metadata_sys::BucketMetadataMutationGuard> {
super::ecstore_bucket::metadata_sys::acquire_bucket_metadata_transaction_lock_for_incarnation(
bucket,
expected_incarnation_id,
)
.await
}
pub(crate) async fn update_under_transaction_lock(
guard: &super::ecstore_bucket::metadata_sys::BucketMetadataMutationGuard,
bucket: &str,
config_file: &str,
data: Vec<u8>,
) -> Result<OffsetDateTime> {
super::ecstore_bucket::metadata_sys::update_under_transaction_lock(guard, bucket, config_file, data).await
}
pub(crate) async fn delete_under_transaction_lock(
guard: &super::ecstore_bucket::metadata_sys::BucketMetadataMutationGuard,
bucket: &str,
config_file: &str,
) -> Result<OffsetDateTime> {
super::ecstore_bucket::metadata_sys::delete_under_transaction_lock(guard, bucket, config_file).await
}
pub(crate) async fn update_bucket_targets_under_transaction_lock(
guard: &super::ecstore_bucket::metadata_sys::BucketMetadataMutationGuard,
bucket: &str,
@@ -445,10 +417,6 @@ pub(crate) mod replication {
};
pub(crate) type BucketReplicationResyncStatus = super::ecstore_bucket::replication::BucketReplicationResyncStatus;
pub(crate) type BucketStats = super::ecstore_bucket::replication::BucketStats;
pub(crate) type BucketReplicationStats = super::ecstore_bucket::replication::BucketReplicationStats;
pub(crate) type BucketReplicationStat = super::ecstore_bucket::replication::BucketReplicationStat;
pub(crate) type InQueueMetric = super::ecstore_bucket::replication::InQueueMetric;
pub(crate) type XferStats = super::ecstore_bucket::replication::XferStats;
pub(crate) type ReplicationStatusType = super::ecstore_bucket::replication::ReplicationStatusType;
pub(crate) type ResyncOpts = super::ecstore_bucket::replication::ResyncOpts;
pub(crate) type ResyncStatusType = super::ecstore_bucket::replication::ResyncStatusType;
+2 -39
View File
@@ -1158,19 +1158,6 @@ fn lifecycle_has_expiry_rules(config: &BucketLifecycleConfiguration) -> bool {
})
}
/// Status-independent presence of the expiry subset that site replication
/// propagates (`replicateILMExpiry`): expiration / noncurrent-version
/// expiration only. Distinct from [`lifecycle_has_expiry_rules`], which
/// filters on ENABLED for scanner scheduling — editing a Disabled expiry rule
/// must still advance the replication axis. Del-marker expiration and
/// abort-multipart are site-local and never travel.
fn lifecycle_rules_have_expiry(config: &BucketLifecycleConfiguration) -> bool {
config
.rules
.iter()
.any(|rule| rule.expiration.is_some() || rule.noncurrent_version_expiration.is_some())
}
fn lifecycle_has_abort_multipart_rules(config: &BucketLifecycleConfiguration) -> bool {
config.rules.iter().any(|rule| {
rule.status == ExpirationStatus::from_static(ExpirationStatus::ENABLED)
@@ -2199,24 +2186,7 @@ impl DefaultBucketUsecase {
return Err(s3_error!(InvalidArgument, "{err}"));
}
// Stamp the expiry axis only when the expiry subset can have changed
// (MinIO: HasExpiry() || expiryRuleRemoved). Site-replication peers
// judge lc-config staleness on this axis; a transition-only edit that
// advanced it would let this site's stale expiry subset shadow — and
// roll back — a newer peer expiry edit fleet-wide.
let previous_expiry_updated_at = match metadata_sys::get_lifecycle_config(&bucket).await {
Ok((previous, _)) => {
if lifecycle_rules_have_expiry(&input_cfg) || lifecycle_rules_have_expiry(&previous) {
Some(Timestamp::from(time::OffsetDateTime::now_utc()))
} else {
previous.expiry_updated_at
}
}
// No previous config (or unreadable): stamping is the
// conservative pre-existing behavior.
Err(_) => lifecycle_rules_have_expiry(&input_cfg).then(|| Timestamp::from(time::OffsetDateTime::now_utc())),
};
input_cfg.expiry_updated_at = previous_expiry_updated_at;
input_cfg.expiry_updated_at = Some(Timestamp::from(time::OffsetDateTime::now_utc()));
let data = serialize_config(&input_cfg)?;
update_bucket_config_for_incarnation(&bucket, BUCKET_LIFECYCLE_CONFIG, data, expected_incarnation_id)
.await
@@ -2227,14 +2197,7 @@ impl DefaultBucketUsecase {
let mut item = sr_bucket_meta_item(bucket.clone(), "lc-config");
item.expiry_lc_config =
Some(serialize_config(&input_cfg).and_then(|bytes| String::from_utf8(bytes).map_err(to_internal_error))?);
// The item travels with the expiry axis, not the wall clock: a site
// whose expiry knowledge is old (or absent — UNIX_EPOCH) must not
// out-rank newer peer expiry state at the receivers.
item.expiry_updated_at = input_cfg
.expiry_updated_at
.clone()
.map(time::OffsetDateTime::from)
.or(Some(time::OffsetDateTime::UNIX_EPOCH));
item.expiry_updated_at = item.updated_at;
if let Err(err) = site_replication_bucket_meta_hook(item).await {
warn!(bucket = %bucket, error = ?err, "site replication bucket lifecycle hook failed");
}

Some files were not shown because too many files have changed in this diff Show More