Compare commits

..

2 Commits

Author SHA1 Message Date
overtrue 2aa4e411e5 test(interop): fix the fixture lab's all-cases and SSE-C capture paths
Two defects kept most of the fixture matrix uncapturable, both found by actually running the lab rather than reading it.

`./capture_via_docker.sh all` never worked. With no case ids to forward, `case_args` is an empty array, and `${arr[@]}` under `set -u` aborts on bash 3.2 — still the default /bin/bash on macOS — with "case_args[@]: unbound variable". The build_args array this branch added has the same shape and would have hit it on the CI path, where no mirror overrides are set. Both now use ${arr[@]+"${arr[@]}"}, which is empty-safe on 3.2.

The SSE-C cases could not be captured at all: MinIO refuses SSE-C over a plain-HTTP connection, and the script pinned the lab to its default http endpoint with no way to override. lab.py already provisions a self-signed certificate and speaks https end to end, so this only needed an endpoint passthrough — MINIO_LAB_ENDPOINT, documented alongside the requirement.

With both fixed, one command captures the full six-case matrix (SSE-S3 / SSE-KMS / SSE-C x singlepart / multipart), which is what backlog#1638's D4 matrix has to be measured against.

Refs rustfs/backlog#1638.
2026-08-16 22:25:27 +08:00
overtrue 5571d4830b test(interop): let the MinIO fixture lab build from registry mirrors
The fixture lab's throwaway image hardcoded two Docker Hub base images, so on a network that cannot reach registry-1.docker.io the capture script fails at build time and no MinIO fixtures can be generated at all. That matters more than it looks: the interop reader tests are #[ignore]d precisely because the fixtures are gitignored and must be generated locally, so an unreachable registry silently keeps the whole MinIO SSE interop lane unverifiable.

Both base images become build args with the current Docker Hub values as defaults, and the capture script forwards MINIO_LAB_MINIO_IMAGE / MINIO_LAB_PYTHON_IMAGE when set. CI keeps the defaults; a restricted network can point at quay.io (which publishes the MinIO releases) and public.ecr.aws (which mirrors the official Python images), both verified to produce a working lab image and real fixtures.

The README documents the override and warns against an unpinned :latest MinIO tag, since the captured on-disk format is exactly what the interop tests assert against.

Refs rustfs/backlog#1638.
2026-08-16 21:23:17 +08:00
58 changed files with 202 additions and 1774 deletions
+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;
@@ -82,6 +82,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>";
@@ -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),
-13
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
@@ -893,10 +890,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 +1104,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 +1617,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 +1721,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);
}
+1 -1
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};
@@ -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);
@@ -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());
+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>,
+56 -241
View File
@@ -180,13 +180,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),
@@ -668,60 +666,6 @@ pub(in crate::set_disk) async fn data_read_early_stop_inline_body_miss_reason(
parts_metadata: &[FileInfo],
disks: &[Option<DiskStore>],
) -> Option<&'static str> {
if let Some(reason) = data_read_early_stop_inline_candidate_miss_reason(candidate) {
return Some(reason);
}
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 Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_GEOMETRY);
};
let data_files =
match collect_inline_data_shard_fileinfos_by_index_or_reason(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),
};
let Some(part) = candidate.parts.first() else {
return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_PART_SHAPE);
};
let Ok(object_size) = usize::try_from(candidate.size) else {
return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_SIZE);
};
let checksum_info = candidate.erasure.get_checksum_info(part.number);
let checksum_algo = if candidate.uses_legacy_checksum && checksum_info.algorithm == HashAlgorithm::HighwayHash256S {
HashAlgorithm::HighwayHash256SLegacy
} else {
checksum_info.algorithm
};
let read_length = inline_erasure_shard_file_offset(
0,
object_size,
object_size,
candidate.erasure.block_size,
erasure.data_shards,
candidate.uses_legacy_checksum,
);
let shard_size = inline_erasure_shard_size(candidate.erasure.block_size, erasure.data_shards, candidate.uses_legacy_checksum);
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_early_stop_inline_candidate_miss_reason(candidate: &FileInfo) -> 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);
@@ -759,7 +703,51 @@ fn data_read_early_stop_inline_candidate_miss_reason(candidate: &FileInfo) -> Op
if !can_try_inline_data_shards_direct(object_size, candidate.erasure.block_size) {
return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_SIZE);
}
None
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 Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_GEOMETRY);
};
let data_files =
match collect_inline_data_shard_fileinfos_by_index_or_reason(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),
};
let Some(part) = candidate.parts.first() else {
return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_PART_SHAPE);
};
let checksum_info = candidate.erasure.get_checksum_info(part.number);
let checksum_algo = if candidate.uses_legacy_checksum && checksum_info.algorithm == HashAlgorithm::HighwayHash256S {
HashAlgorithm::HighwayHash256SLegacy
} else {
checksum_info.algorithm
};
let read_length = inline_erasure_shard_file_offset(
0,
object_size,
object_size,
candidate.erasure.block_size,
erasure.data_shards,
candidate.uses_legacy_checksum,
);
let shard_size = inline_erasure_shard_size(candidate.erasure.block_size, erasure.data_shards, candidate.uses_legacy_checksum);
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(
@@ -1848,7 +1836,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>],
@@ -2139,7 +2126,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,
@@ -2456,7 +2442,6 @@ impl SetDisks {
let bucket: Arc<str> = Arc::from(bucket);
let object: Arc<str> = Arc::from(object);
let version_id: Arc<str> = Arc::from(version_id);
let slowtail_fault = get_metadata_slowtail_fault_request(bucket.as_ref(), object.as_ref(), read_data);
let futures = disks.iter().enumerate().map(|(disk_index, disk)| {
let disk = disk.clone();
let task_opts = opts;
@@ -2464,14 +2449,10 @@ impl SetDisks {
let bucket = bucket.clone();
let object = object.clone();
let version_id = version_id.clone();
let slowtail_fault = slowtail_fault.clone();
tokio::spawn(async move {
let response_start = observe.then(Instant::now);
let result = if let Some(disk) = disk {
Self::record_read_version_call(&object, disk_index);
if let Some(delay) = slowtail_fault.as_ref().and_then(|fault| fault.delay_for_disk(disk_index)) {
tokio::time::sleep(delay).await;
}
disk.read_version(&org_bucket, &bucket, &object, &version_id, &task_opts)
.await
} else {
@@ -2567,7 +2548,6 @@ impl SetDisks {
let mut scheduled_count = 0usize;
let mut force_full_wait = false;
let mut final_miss_reason_override = None;
let slowtail_fault = get_metadata_slowtail_fault_request(bucket.as_ref(), object.as_ref(), read_data);
let spawn_read_version =
|join_set: &mut JoinSet<(usize, disk::error::Result<FileInfo>, Duration)>, index: usize, disk: Option<DiskStore>| {
let task_opts = opts;
@@ -2575,7 +2555,6 @@ impl SetDisks {
let bucket = bucket.clone();
let object = object.clone();
let version_id = version_id.clone();
let slowtail_fault = slowtail_fault.clone();
join_set.spawn(async move {
let response_start = Instant::now();
let result = if let Some(disk) = disk {
@@ -2584,9 +2563,6 @@ impl SetDisks {
Self::record_read_version_call(&object, index);
#[cfg(test)]
Self::read_version_fanout_barrier(&object, index).await;
if let Some(delay) = slowtail_fault.as_ref().and_then(|fault| fault.delay_for_disk(index)) {
tokio::time::sleep(delay).await;
}
disk.read_version(&org_bucket, &bucket, &object, &version_id, &task_opts)
.await
} else {
@@ -2620,14 +2596,6 @@ impl SetDisks {
Ok(file_info) => {
observations.push(MetadataFanoutObservation::from_file_info(&file_info, elapsed));
accumulator.observe_file_info(&file_info);
if bounded_fanout
&& read_data
&& !force_full_wait
&& let Some(reason) = data_read_early_stop_inline_candidate_miss_reason(&file_info)
{
force_full_wait = true;
final_miss_reason_override.get_or_insert(reason);
}
if let Some(slot) = ress.get_mut(index) {
*slot = file_info;
}
@@ -2987,7 +2955,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,
@@ -3021,11 +2988,14 @@ impl SetDisks {
});
}
let (ress, _errors) = match collect_read_multiple_results(futures, read_quorum).await {
let (ress, errors) = match collect_read_multiple_results(futures, read_quorum).await {
Ok(collected) => collected,
Err(()) => return empty_quorum_result(),
};
// debug!("ReadMultipleResp ress {:?}", ress);
// debug!("ReadMultipleResp errors {:?}", errors);
let mut ret = Vec::with_capacity(req.files.len());
for want in req.files.iter() {
@@ -3164,7 +3134,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,
@@ -3174,7 +3143,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,
@@ -3293,7 +3261,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,
@@ -5106,7 +5073,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,
@@ -5331,7 +5297,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";
}
@@ -5769,130 +5734,6 @@ mod tests {
(dirs, disks)
}
#[test]
fn metadata_slowtail_fault_delay_parses_and_filters_request() {
temp_env::with_vars(
[
(ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DELAY_MS, Some("25")),
(ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DISKS, Some("1,3")),
(ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_BUCKET, Some("bench-bucket")),
(ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_OBJECT_PREFIX, Some("objects/")),
],
|| {
assert_eq!(
get_metadata_slowtail_fault_delay("bench-bucket", "objects/000001", 3, true),
Some(Duration::from_millis(25))
);
assert!(get_metadata_slowtail_fault_delay("bench-bucket", "objects/000001", 2, true).is_none());
assert!(get_metadata_slowtail_fault_delay("other-bucket", "objects/000001", 3, true).is_none());
assert!(get_metadata_slowtail_fault_delay("bench-bucket", "other/000001", 3, true).is_none());
assert!(get_metadata_slowtail_fault_delay("bench-bucket", "objects/000001", 3, false).is_none());
},
);
}
#[test]
fn metadata_slowtail_fault_delay_disables_invalid_disk_list() {
temp_env::with_vars(
[
(ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DELAY_MS, Some("25")),
(ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DISKS, Some("1,nope")),
],
|| {
assert!(get_metadata_slowtail_fault_delay("bucket", "object", 1, true).is_none());
},
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn metadata_slowtail_fault_delays_only_data_read_metadata_task() {
const DISKS: usize = 4;
let bucket = "metadata-slowtail-fault-bucket";
let object = "objects/metadata-slowtail-fault-object";
let (dirs, disks) = call_counter_local_disks(bucket, DISKS).await;
install_metadata_fanout_fileinfo(&disks, bucket, object, None).await;
temp_env::async_with_vars(
[
(ENV_RUSTFS_GET_METADATA_EARLY_STOP_ENABLE, Some("false")),
(ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DELAY_MS, Some("150")),
(ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DISKS, Some("3")),
(ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_BUCKET, Some(bucket)),
(ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_OBJECT_PREFIX, Some("objects/")),
],
async {
let read_without_data =
SetDisks::read_all_fileinfo_observed(&disks, bucket, bucket, object, "", false, false, false, true, 2);
tokio::time::timeout(Duration::from_millis(100), read_without_data)
.await
.expect("non-data metadata fanout must not be delayed by the data-read slowtail hook")
.expect("metadata fanout without read_data should resolve");
let mut read_with_data = Box::pin(SetDisks::read_all_fileinfo_observed(
&disks, bucket, bucket, object, "", true, false, false, true, 2,
));
assert!(
tokio::time::timeout(Duration::from_millis(40), &mut read_with_data)
.await
.is_err(),
"data-read metadata fanout must wait for the injected slow read_version response"
);
let (parts_metadata, errs, diagnostics) = tokio::time::timeout(Duration::from_secs(2), read_with_data)
.await
.expect("injected slowtail should eventually complete")
.expect("data-read metadata fanout should resolve");
assert_eq!(parts_metadata.iter().filter(|fi| fi.name == object).count(), DISKS);
assert!(errs.iter().all(Option::is_none));
assert_eq!(diagnostics.total_responses(), DISKS);
},
)
.await;
drop(dirs);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn metadata_slowtail_fault_delays_early_stop_metadata_task() {
const DISKS: usize = 4;
let bucket = "metadata-slowtail-early-stop-bucket";
let object = "objects/metadata-slowtail-early-stop-object";
let (dirs, disks) = call_counter_local_disks(bucket, DISKS).await;
install_metadata_fanout_fileinfo(&disks, bucket, object, None).await;
temp_env::async_with_vars(
[
(ENV_RUSTFS_GET_METADATA_EARLY_STOP_ENABLE, Some("true")),
(ENV_RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE, Some("true")),
(ENV_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT, Some("false")),
(ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DELAY_MS, Some("150")),
(ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DISKS, Some("3")),
(ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_BUCKET, Some(bucket)),
(ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_OBJECT_PREFIX, Some("objects/")),
],
async {
let mut read_with_data = Box::pin(SetDisks::read_all_fileinfo_observed(
&disks, bucket, bucket, object, "", true, false, false, true, 2,
));
assert!(
tokio::time::timeout(Duration::from_millis(40), &mut read_with_data)
.await
.is_err(),
"early-stop metadata fanout must still wait for the injected slow response after fallback to full wait"
);
let (parts_metadata, errs, diagnostics) = tokio::time::timeout(Duration::from_secs(2), read_with_data)
.await
.expect("injected early-stop slowtail should eventually complete")
.expect("early-stop metadata fanout should resolve");
assert_eq!(parts_metadata.iter().filter(|fi| fi.name == object).count(), DISKS);
assert!(errs.iter().all(Option::is_none));
assert_eq!(diagnostics.total_responses(), DISKS);
},
)
.await;
drop(dirs);
}
/// Demo / regression guard for the backlog#1325 per-disk call counters.
///
/// The metadata fan-out issues each `read_version` inside its own
@@ -7240,7 +7081,7 @@ mod tests {
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn bounded_non_inline_data_get_immediately_forces_full_fanout() {
async fn bounded_non_inline_data_get_hedges_then_waits_for_full_fanout() {
const DISKS: usize = 4;
let bucket = "bounded-data-get-hedge-bucket";
let object = "bounded-data-get-hedge-object";
@@ -7271,7 +7112,7 @@ mod tests {
}
})
.await
.expect("bounded non-inline data-read fanout should immediately schedule the spare disk");
.expect("bounded data-read fanout should hedge by starting the spare disk");
let pending = tokio::time::timeout(BARRIER_PAUSE_GUARD, &mut read).await;
assert!(
@@ -7287,7 +7128,7 @@ mod tests {
assert_eq!(
calls.total(disk_call_counters::KIND_READ_VERSION),
DISKS as u64,
"bounded non-inline data-read fanout should issue the paused disk plus the remaining spare"
"bounded data-read fanout should issue the paused disk plus one spare hedge"
);
assert_eq!(diagnostics.total_responses(), DISKS);
assert_eq!(parts_metadata.iter().filter(|fi| fi.name == object).count(), DISKS);
@@ -7314,42 +7155,16 @@ mod tests {
("RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT", None::<&str>),
],
async {
let barrier = rename_fanout_barrier::arm(object, 2, rename_fanout_barrier::PHASE_READ_VERSION);
let calls = disk_call_counters::observe(object);
let disks_for_read = disks.clone();
let mut read = tokio::spawn(async move {
SetDisks::read_all_fileinfo_observed(&disks_for_read, bucket, bucket, object, "", true, false, false, true, 2)
let (parts_metadata, errs, diagnostics) =
SetDisks::read_all_fileinfo_observed(&disks, bucket, bucket, object, "", true, false, false, true, 2)
.await
});
tokio::time::timeout(BARRIER_PAUSE_GUARD, barrier.wait_until_paused())
.await
.expect("default bounded non-inline read should schedule the paused metadata task");
tokio::time::timeout(BARRIER_PAUSE_GUARD, async {
while calls.for_disk(disk_call_counters::KIND_READ_VERSION, 3) == 0 {
tokio::task::yield_now().await;
}
})
.await
.expect(
"default bounded non-inline read should immediately force full fanout after the first non-inline response",
);
let pending = tokio::time::timeout(BARRIER_PAUSE_GUARD, &mut read).await;
assert!(
pending.is_err(),
"default non-inline data reads must not return before the paused metadata response"
);
barrier.release();
let (parts_metadata, errs, diagnostics) = read
.await
.expect("metadata read task should not panic")
.expect("default data-read metadata should resolve");
.expect("default data-read metadata should resolve");
assert_eq!(
calls.total(disk_call_counters::KIND_READ_VERSION),
DISKS as u64,
"default non-inline GET data-read metadata must keep full fanout without waiting for a quorum miss first"
"default non-inline 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
}
+17 -131
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)]
@@ -102,7 +103,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::{
@@ -174,14 +177,15 @@ use std::future::Future;
use std::hash::{BuildHasher, Hash, Hasher};
use std::mem::{self};
use std::pin::Pin;
use std::sync::OnceLock;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, OnceLock};
use std::task::{Context, Poll};
use std::time::{Instant, SystemTime, UNIX_EPOCH};
use std::{
collections::{HashMap, HashSet},
io::{Cursor, Write},
path::Path,
sync::Arc,
time::Duration,
};
use time::OffsetDateTime;
@@ -620,9 +624,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
@@ -696,15 +698,7 @@ const ENV_RUSTFS_GET_METADATA_EARLY_STOP_ENABLE: &str = "RUSTFS_GET_METADATA_EAR
// 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";
@@ -714,12 +708,7 @@ const ENV_RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE: &str = "RUSTFS_GET_ME
const DEFAULT_RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE: bool = true;
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 = true;
const ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DELAY_MS: &str = "RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DELAY_MS";
const ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DISKS: &str = "RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DISKS";
const ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_BUCKET: &str = "RUSTFS_GET_METADATA_SLOWTAIL_FAULT_BUCKET";
const ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_OBJECT_PREFIX: &str = "RUSTFS_GET_METADATA_SLOWTAIL_FAULT_OBJECT_PREFIX";
const DEFAULT_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT: bool = false;
// --- Multipart Reader-Setup Prefetch Configuration (backlog#870) ---
@@ -921,10 +910,6 @@ 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)
@@ -1125,10 +1110,7 @@ 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 inline GET should schedule the initial bounded quorum plus one hedge"
);
assert_eq!(calls_total, 4, "default production GET should eagerly schedule the full metadata fanout");
assert_eq!(
recorder.histogram_values(
"rustfs_io_get_object_metadata_fanout_scheduled",
@@ -1702,95 +1684,6 @@ fn is_get_metadata_early_stop_bounded_fanout_enabled() -> bool {
}
}
#[derive(Debug)]
struct GetMetadataSlowtailFaultConfig {
delay: Duration,
disks: Arc<[usize]>,
bucket: Option<String>,
object_prefix: Option<String>,
}
#[derive(Clone, Debug)]
struct GetMetadataSlowtailFaultRequest {
delay: Duration,
disks: Arc<[usize]>,
}
impl GetMetadataSlowtailFaultRequest {
fn delay_for_disk(&self, disk_index: usize) -> Option<Duration> {
self.disks.contains(&disk_index).then_some(self.delay)
}
}
fn parse_get_metadata_slowtail_fault_disks(raw: &str) -> Option<Vec<usize>> {
let mut disks = Vec::new();
for item in raw.split(',').map(str::trim).filter(|item| !item.is_empty()) {
let Ok(index) = item.parse::<usize>() else {
return None;
};
if !disks.contains(&index) {
disks.push(index);
}
}
(!disks.is_empty()).then_some(disks)
}
fn load_get_metadata_slowtail_fault_config() -> Option<GetMetadataSlowtailFaultConfig> {
let delay_ms = rustfs_utils::get_env_u64(ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DELAY_MS, 0);
if delay_ms == 0 {
return None;
}
let disks = parse_get_metadata_slowtail_fault_disks(&std::env::var(ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DISKS).ok()?)?;
let bucket = std::env::var(ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_BUCKET)
.ok()
.filter(|value| !value.is_empty());
let object_prefix = std::env::var(ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_OBJECT_PREFIX)
.ok()
.filter(|value| !value.is_empty());
Some(GetMetadataSlowtailFaultConfig {
delay: Duration::from_millis(delay_ms),
disks: Arc::from(disks.into_boxed_slice()),
bucket,
object_prefix,
})
}
fn get_metadata_slowtail_fault_request(bucket: &str, object: &str, read_data: bool) -> Option<GetMetadataSlowtailFaultRequest> {
if !read_data {
return None;
}
#[cfg(test)]
let config = load_get_metadata_slowtail_fault_config();
#[cfg(test)]
let config = config.as_ref()?;
#[cfg(not(test))]
let config = ({
static CACHED: OnceLock<Option<GetMetadataSlowtailFaultConfig>> = OnceLock::new();
CACHED.get_or_init(load_get_metadata_slowtail_fault_config).as_ref()
})?;
if let Some(expected_bucket) = &config.bucket
&& expected_bucket != bucket
{
return None;
}
if let Some(expected_prefix) = &config.object_prefix
&& !object.starts_with(expected_prefix)
{
return None;
}
Some(GetMetadataSlowtailFaultRequest {
delay: config.delay,
disks: config.disks.clone(),
})
}
#[cfg(test)]
fn get_metadata_slowtail_fault_delay(bucket: &str, object: &str, disk_index: usize, read_data: bool) -> Option<Duration> {
get_metadata_slowtail_fault_request(bucket, object, read_data)?.delay_for_disk(disk_index)
}
/// Check if multipart reads prefetch the next part's bitrot reader setup
/// while the current part decodes (backlog#870).
///
@@ -1816,10 +1709,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(|| {
@@ -1859,10 +1748,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();
@@ -2296,7 +2181,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,
@@ -2902,7 +2786,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,
@@ -3178,13 +3061,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)
@@ -3778,7 +3659,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
@@ -4357,7 +4237,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())
}
@@ -4633,6 +4512,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>],
@@ -5409,7 +5297,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,
@@ -5418,7 +5305,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,
+2 -11
View File
@@ -453,9 +453,7 @@ impl SetDisks {
..Default::default()
};
// Bound, not `_`: this guard must live to the end of the scope. A bare
// `_` would drop it here and release the namespace write lock.
let _write_lock_guard = if !opts.no_lock {
let write_lock_guard = if !opts.no_lock {
let ns_lock = self.new_ns_lock(bucket, object).await?;
Some(
ns_lock
@@ -998,7 +996,7 @@ impl SetDisks {
readers.push(None);
continue;
}
Err(_e) => {
Err(e) => {
readers.push(None);
continue;
}
@@ -1547,9 +1545,6 @@ impl SetDisks {
for candidate in candidates.iter_mut().filter(|candidate| candidate.local_payload) {
for (disk_index, disk) in disks.iter().enumerate() {
// Only the #[cfg(test)] fault-injection branch below reads this.
#[cfg(not(test))]
let _ = disk_index;
let Some(disk) = disk else {
return Ok(DanglingDeleteSafety::UnsafeToDelete);
};
@@ -1721,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,
+5 -7
View File
@@ -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,
@@ -1400,7 +1398,7 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
let mut count = max_parts;
for part in object_parts.iter() {
for (i, part) in object_parts.iter().enumerate() {
if let Some(err) = &part.error {
warn!("list_object_parts part error: {:?}", &err);
}
@@ -2043,8 +2041,8 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
&& let Err(err) = checksum.add_part(&cs, ext_part.actual_size)
{
error!(
"complete_multipart_upload checksum add_part failed part_id={}, bucket={}, object={}, err={}",
p.part_num, bucket, object, err
"complete_multipart_upload checksum add_part failed part_id={}, bucket={}, object={}",
p.part_num, bucket, object
);
return Err(Error::InvalidPart(p.part_num, ext_part.etag.clone(), p.etag.clone().unwrap_or_default()));
}
@@ -2089,8 +2087,8 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
}
} else if let Err(err) = wtcs.matches(&checksum_combined, uploaded_parts.len() as i32) {
error!(
"complete_multipart_upload checksum matches failed want={}, got={}, err={}",
wtcs.encoded, checksum.encoded, err
"complete_multipart_upload checksum matches failed want={}, got={}",
wtcs.encoded, checksum.encoded
);
return Err(Error::other(format!(
"complete_multipart_upload checksum matches failed want={}, got={}",
+2 -322
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();
}
@@ -5656,9 +5620,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
// TODO: Lifecycle
let mut version_found = true;
// delete_object_version below derives its own majority quorum from the
// disk array, so the object-derived quorum here is unused.
let (mut goi, _write_quorum, gerr) = self.get_object_info_and_quorum(bucket, object, &opts).await;
let (mut goi, write_quorum, gerr) = self.get_object_info_and_quorum(bucket, object, &opts).await;
if let Some(err) = &gerr
&& goi.name.is_empty()
{
@@ -6412,7 +6374,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
self.record_capacity_scope_if_needed(opts.capacity_scope_token, &disks);
for disk in disks.iter() {
if disk.is_some() {
if let Some(disk) = disk {
continue;
}
let _ = self
@@ -10208,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,
+6 -8
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()
}
@@ -3937,7 +3935,7 @@ mod tests {
}
#[test]
fn metadata_early_stop_bounded_fanout_defaults_to_enabled() {
fn metadata_early_stop_bounded_fanout_defaults_to_disabled() {
temp_env::with_vars(
[
(ENV_RUSTFS_GET_METADATA_EARLY_STOP_ENABLE, Some("true")),
@@ -3946,20 +3944,20 @@ mod tests {
],
|| {
assert!(is_get_metadata_data_read_early_stop_enabled());
assert!(is_get_metadata_early_stop_bounded_fanout_enabled());
assert!(!is_get_metadata_early_stop_bounded_fanout_enabled());
},
);
temp_env::with_vars([(ENV_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT, Some("false"))], || {
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("true")),
(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());
assert!(!is_get_metadata_early_stop_bounded_fanout_enabled());
},
);
}
+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;
@@ -34,6 +34,14 @@ MINIO_LAB_PYTHON_IMAGE=public.ecr.aws/docker/library/python:3.12-slim \
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.
## Capturing the SSE-C cases
MinIO refuses SSE-C over a plain-HTTP connection, so the `sse-c-*` cases cannot be captured against the default endpoint — `./capture_via_docker.sh all` fails on the first SSE-C upload with `InvalidRequest ... must be made over a secure connection`. The lab provisions its own self-signed certificate; point it at the HTTPS endpoint to capture them:
```bash
MINIO_LAB_ENDPOINT=https://127.0.0.1:9000 ./capture_via_docker.sh all
```
## Layout
The default root is `artifacts/minio-fixture-lab`, which is already ignored by the repository.
@@ -12,6 +12,13 @@
# # ignored interop tests consume
# ./capture_via_docker.sh sse-s3-singlepart-64k # specific case id(s)
# ./capture_via_docker.sh all # full SSE/size matrix
#
# The SSE-C cases are not reachable over the default plain-HTTP endpoint: MinIO
# refuses SSE-C unless the connection is secure ("Requests specifying Server
# Side Encryption with Customer provided keys must be made over a secure
# connection"). Capture those by pointing the lab at its self-signed HTTPS
# endpoint, which it provisions itself:
# MINIO_LAB_ENDPOINT=https://127.0.0.1:9000 ./capture_via_docker.sh all
set -euo pipefail
IMAGE="${MINIO_LAB_IMAGE:-rustfs-minio-lab:latest}"
@@ -46,7 +53,10 @@ if [ -n "${MINIO_LAB_PYTHON_IMAGE:-}" ]; then
fi
echo ">> building ${IMAGE}"
docker build -f "${SCRIPT_DIR}/Dockerfile" -t "${IMAGE}" "${build_args[@]}" "${SCRIPT_DIR}"
# ${arr[@]+"${arr[@]}"} rather than "${arr[@]}": under `set -u`, bash 3.2 —
# still the default /bin/bash on macOS — treats an empty array expansion as an
# unbound variable and aborts.
docker build -f "${SCRIPT_DIR}/Dockerfile" -t "${IMAGE}" ${build_args[@]+"${build_args[@]}"} "${SCRIPT_DIR}"
echo ">> capturing fixtures into ${FIXTURE_REL}"
docker run --rm -v "${REPO_ROOT}:/repo" "${IMAGE}" \
@@ -54,6 +64,7 @@ docker run --rm -v "${REPO_ROOT}:/repo" "${IMAGE}" \
--root "/repo/${FIXTURE_REL}" \
--work-root /tmp/minio-lab-work \
--minio-binary /usr/local/bin/minio \
"${case_args[@]}"
--endpoint "${MINIO_LAB_ENDPOINT:-http://127.0.0.1:9000}" \
${case_args[@]+"${case_args[@]}"}
echo ">> done — fixtures under ${REPO_ROOT}/${FIXTURE_REL}/cases/"
@@ -2995,9 +2995,9 @@ fn table_entry_from_create_table_request(
let CreateTableRequest {
name,
location,
mut schema,
mut partition_spec,
mut write_order,
schema,
partition_spec,
write_order,
stage_create,
mut properties,
} = request;
@@ -3031,9 +3031,6 @@ fn table_entry_from_create_table_request(
let metadata_location =
crate::table_catalog::default_table_metadata_file_path(namespace, &table, &next_metadata_file_name(1, &table_id));
crate::table_catalog::assign_fresh_create_schema_ids(&mut schema, partition_spec.as_mut(), write_order.as_mut())
.map_err(catalog_store_error)?;
let entry = crate::table_catalog::TableEntry {
version: crate::table_catalog::TABLE_CATALOG_ENTRY_VERSION,
table_bucket: bucket.to_string(),
@@ -1873,66 +1873,6 @@ fn create_table_request_accepts_standard_iceberg_rest_shape() {
assert_eq!(request.name, "events");
}
#[test]
fn create_table_assigns_positive_ids_to_spark_schema() {
let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
let request: CreateTableRequest = serde_json::from_value(serde_json::json!({
"name": "events",
"schema": {
"type": "struct",
"schema-id": 0,
"fields": [
{"id": 0, "name": "id", "required": false, "type": "long"},
{"id": 1, "name": "payload", "required": false, "type": "string"}
]
},
"partition-spec": {"spec-id": 0, "fields": []},
"properties": {"owner": "spark"}
}))
.expect("Spark create table request should parse");
let (_, metadata) = table_entry_from_create_table_request("warehouse", &namespace, request)
.expect("catalog should assign positive field IDs");
assert_eq!(metadata["schemas"][0]["fields"][0]["id"], 1);
assert_eq!(metadata["schemas"][0]["fields"][1]["id"], 2);
assert_eq!(metadata["last-column-id"], 2);
}
#[test]
fn create_table_assigns_fresh_id_to_negative_temporary_field_id() {
let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
let request: CreateTableRequest = serde_json::from_value(serde_json::json!({
"name": "events",
"schema": {
"type": "struct",
"identifier-field-ids": [-1],
"fields": [{"id": -1, "name": "id", "required": true, "type": "long"}]
},
"partition-spec": {
"fields": [{"source-id": -1, "name": "id", "transform": "identity"}]
},
"write-order": {
"fields": [{
"source-id": -1,
"transform": "identity",
"direction": "asc",
"null-order": "nulls-first"
}]
}
}))
.expect("create table request with a negative temporary field ID should parse");
let (_, metadata) = table_entry_from_create_table_request("warehouse", &namespace, request)
.expect("catalog should replace the negative temporary field ID");
assert_eq!(metadata["schemas"][0]["fields"][0]["id"], 1);
assert_eq!(metadata["schemas"][0]["identifier-field-ids"], serde_json::json!([1]));
assert_eq!(metadata["partition-specs"][0]["fields"][0]["source-id"], 1);
assert_eq!(metadata["sort-orders"][0]["fields"][0]["source-id"], 1);
assert_eq!(metadata["last-column-id"], 1);
}
#[test]
fn create_table_request_honors_supported_format_version_property() {
let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
@@ -2050,142 +1990,6 @@ fn catalog_assigns_read_only_schema_spec_and_sort_order_ids() {
assert_eq!(updated["default-sort-order-id"], 0);
}
#[test]
fn create_table_assigns_fresh_schema_field_ids_and_rewrites_references() {
let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
let request: CreateTableRequest = serde_json::from_value(serde_json::json!({
"name": "events",
"schema": {
"type": "struct",
"schema-id": 41,
"identifier-field-ids": [0],
"fields": [
{"id": 0, "name": "id", "required": true, "type": "long"},
{
"id": 10,
"name": "details",
"required": false,
"type": {
"type": "struct",
"fields": [{"id": 11, "name": "category", "required": false, "type": "string"}]
}
},
{
"id": 20,
"name": "tags",
"required": false,
"type": {
"type": "list",
"element-id": 21,
"element-required": false,
"element": "string"
}
},
{
"id": 30,
"name": "attributes",
"required": false,
"type": {
"type": "map",
"key-id": 31,
"key": "string",
"value-id": 32,
"value-required": false,
"value": {
"type": "struct",
"fields": [{"id": 33, "name": "score", "required": false, "type": "int"}]
}
}
}
]
},
"partition-spec": {
"spec-id": 42,
"fields": [{"source-id": 0, "name": "id", "transform": "identity"}]
},
"write-order": {
"order-id": 43,
"fields": [{
"source-id": 11,
"transform": "identity",
"direction": "asc",
"null-order": "nulls-first"
}]
}
}))
.expect("create table request should parse");
let (_, metadata) =
table_entry_from_create_table_request("warehouse", &namespace, request).expect("catalog should assign fresh field IDs");
let schema = &metadata["schemas"][0];
assert_eq!(schema["fields"][0]["id"], 1);
assert_eq!(schema["fields"][1]["id"], 2);
assert_eq!(schema["fields"][2]["id"], 3);
assert_eq!(schema["fields"][3]["id"], 4);
assert_eq!(schema["fields"][1]["type"]["fields"][0]["id"], 5);
assert_eq!(schema["fields"][2]["type"]["element-id"], 6);
assert_eq!(schema["fields"][3]["type"]["key-id"], 7);
assert_eq!(schema["fields"][3]["type"]["value-id"], 8);
assert_eq!(schema["fields"][3]["type"]["value"]["fields"][0]["id"], 9);
assert_eq!(schema["identifier-field-ids"], serde_json::json!([1]));
assert_eq!(metadata["last-column-id"], 9);
assert_eq!(metadata["partition-specs"][0]["fields"][0]["source-id"], 1);
assert_eq!(metadata["sort-orders"][0]["fields"][0]["source-id"], 5);
}
#[test]
fn create_table_rejects_duplicate_temporary_schema_field_ids() {
let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
let request: CreateTableRequest = serde_json::from_value(serde_json::json!({
"name": "events",
"schema": {
"type": "struct",
"fields": [
{"id": 0, "name": "id", "required": false, "type": "long"},
{"id": 0, "name": "payload", "required": false, "type": "string"}
]
}
}))
.expect("create table request should parse");
let error = table_entry_from_create_table_request("warehouse", &namespace, request)
.expect_err("duplicate temporary field IDs must be rejected");
assert_eq!(error.message(), Some("duplicate create schema field id 0"));
}
#[test]
fn create_table_rejects_excessive_schema_nesting() {
let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
let mut field_type = serde_json::Value::from("long");
for element_id in 1..=crate::table_catalog::ICEBERG_MAX_SCHEMA_NESTING_DEPTH + 1 {
field_type = serde_json::json!({
"type": "list",
"element-id": element_id,
"element-required": false,
"element": field_type
});
}
let request = CreateTableRequest {
name: "events".to_string(),
location: None,
schema: serde_json::json!({
"type": "struct",
"fields": [{"id": 0, "name": "nested", "required": false, "type": field_type}]
}),
partition_spec: None,
write_order: None,
stage_create: false,
properties: BTreeMap::new(),
};
let error = table_entry_from_create_table_request("warehouse", &namespace, request)
.expect_err("excessively nested create schemas must be rejected");
assert_eq!(error.message(), Some("create schema exceeds the maximum nesting depth"));
}
#[test]
fn standard_commit_binds_new_specs_and_sort_orders_to_current_schema() {
let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
@@ -2443,11 +2247,7 @@ fn create_table_counts_collection_ids_in_last_column_id() {
let (_, metadata) =
table_entry_from_create_table_request("warehouse", &namespace, request).expect("table metadata should be created");
let schema = &metadata["schemas"][0];
assert_eq!(schema["fields"][0]["type"]["element-id"], 3);
assert_eq!(schema["fields"][1]["type"]["key-id"], 4);
assert_eq!(schema["fields"][1]["type"]["value-id"], 5);
assert_eq!(metadata["last-column-id"], 5);
assert_eq!(metadata["last-column-id"], 9);
}
#[test]
+1 -4
View File
@@ -56,6 +56,7 @@
//! ```
// Allow dead_code for public API that may be used by external modules or future features
#![allow(dead_code)]
use parking_lot::{Mutex, RwLock};
use std::collections::{HashMap, HashSet};
@@ -263,10 +264,6 @@ pub struct ResourceUsage {
/// Deadlock detector.
pub struct DeadlockDetector {
/// Configuration.
#[allow(
dead_code,
reason = "policy snapshot retained beside the detector it configures (backlog#1823)"
)]
config: RequestHangDetectionPolicy,
/// Shared concurrency facade policy.
policy: DeadlockMonitorPolicy,
+1
View File
@@ -31,6 +31,7 @@
//! ```
// Allow dead_code for public API that may be used by external modules or future features
#![allow(dead_code)]
//! # Key Features
//!
//! - Early lock release after metadata read
+14
View File
@@ -12,6 +12,8 @@
// See the License for the specific language governing permissions and
// limitations under the License.
#![allow(dead_code)]
//! Facade modules for incremental S3 API extraction from `ecfs.rs`.
//!
//! This file intentionally starts as skeleton-only. Behavior remains in place
@@ -27,6 +29,18 @@ pub(crate) mod common;
pub(crate) mod multipart;
pub(crate) mod tagging;
pub(crate) fn default_bucket_usecase() -> DefaultBucketUsecase {
DefaultBucketUsecase::from_global()
}
pub(crate) fn default_multipart_usecase() -> DefaultMultipartUsecase {
DefaultMultipartUsecase::from_global()
}
pub(crate) fn default_object_usecase() -> DefaultObjectUsecase {
DefaultObjectUsecase::from_global()
}
/// Resolve the object use-case for a server's request path (backlog#1052 S6):
/// bind it to the server's own application context so it resolves that
/// server's store instead of the ambient process default.
@@ -71,10 +71,6 @@ pub(crate) struct DecodedManifest {
pub partition_spec_id: Option<i32>,
}
#[allow(
dead_code,
reason = "exercised by table_catalog/tests.rs; the lib target cannot see test-only consumers (backlog#1823)"
)]
pub(crate) fn manifest_paths_from_manifest_list_avro(data: &[u8]) -> TableCatalogStoreResult<Vec<String>> {
Ok(manifest_list_references_from_manifest_list_avro(data)?
.into_iter()
@@ -82,10 +78,6 @@ pub(crate) fn manifest_paths_from_manifest_list_avro(data: &[u8]) -> TableCatalo
.collect())
}
#[allow(
dead_code,
reason = "exercised by table_catalog/tests.rs; the lib target cannot see test-only consumers (backlog#1823)"
)]
pub(crate) fn manifest_list_references_from_manifest_list_avro(
data: &[u8],
) -> TableCatalogStoreResult<Vec<ManifestListReference>> {
@@ -165,10 +157,6 @@ pub(crate) async fn decode_manifest_list_avro_async(data: Vec<u8>) -> TableCatal
.map_err(|err| TableCatalogStoreError::Internal(format!("manifest-list parser task failed: {err}")))?
}
#[allow(
dead_code,
reason = "exercised by table_catalog/tests.rs; the lib target cannot see test-only consumers (backlog#1823)"
)]
pub(crate) fn file_references_from_manifest_avro(
data: &[u8],
) -> TableCatalogStoreResult<Vec<(String, TableMetadataMaintenanceObjectKind)>> {
@@ -178,10 +166,6 @@ pub(crate) fn file_references_from_manifest_avro(
.collect())
}
#[allow(
dead_code,
reason = "exercised by table_catalog/tests.rs; the lib target cannot see test-only consumers (backlog#1823)"
)]
pub(crate) fn data_file_references_from_manifest_avro(data: &[u8]) -> TableCatalogStoreResult<Vec<ManifestDataFileReference>> {
Ok(decode_manifest_avro(data)?.references)
}
@@ -19,7 +19,6 @@ use futures::{StreamExt, TryStreamExt, stream};
use super::super::*;
const ICEBERG_MAX_USER_FIELD_ID: i32 = i32::MAX - 200;
pub(crate) const ICEBERG_MAX_SCHEMA_NESTING_DEPTH: usize = 128;
fn normalize_warehouse_object_prefix(object_prefix: &str, max_prefix_depth: Option<usize>) -> TableCatalogStoreResult<String> {
let object_prefix = object_prefix.strip_suffix('/').unwrap_or(object_prefix);
@@ -104,10 +103,6 @@ pub(crate) fn table_warehouse_index_entry(entry: &TableEntry) -> TableCatalogSto
})
}
#[allow(
dead_code,
reason = "exercised by table_catalog/tests.rs; the lib target cannot see test-only consumers (backlog#1823)"
)]
fn table_warehouse_data_dir_path(entry: &TableEntry) -> TableCatalogStoreResult<String> {
Ok(format!("{}{}", table_warehouse_object_prefix(entry)?, DATA_DIR))
}
@@ -1362,188 +1357,6 @@ fn validate_iceberg_schema(schema: &serde_json::Value, label: &str) -> TableCata
Ok(validate_iceberg_schema_fields(schema, label)?.field_ids)
}
pub(crate) fn assign_fresh_create_schema_ids(
schema: &mut serde_json::Value,
partition_spec: Option<&mut serde_json::Value>,
sort_order: Option<&mut serde_json::Value>,
) -> TableCatalogStoreResult<()> {
let mut assigner = FreshCreateSchemaIdAssigner::new();
assigner.assign_schema(schema)?;
assigner.remap_identifier_field_ids(schema)?;
if let Some(partition_spec) = partition_spec {
assigner.remap_source_ids(partition_spec, "partition spec")?;
}
if let Some(sort_order) = sort_order {
assigner.remap_source_ids(sort_order, "sort order")?;
}
Ok(())
}
struct FreshCreateSchemaIdAssigner {
next_id: i32,
old_to_new: BTreeMap<i32, i32>,
}
impl FreshCreateSchemaIdAssigner {
fn new() -> Self {
Self {
next_id: 1,
old_to_new: BTreeMap::new(),
}
}
fn assign_schema(&mut self, schema: &mut serde_json::Value) -> TableCatalogStoreResult<()> {
let schema = schema
.as_object_mut()
.ok_or_else(|| TableCatalogStoreError::Invalid("create schema must be a JSON object".to_string()))?;
if schema.get("type").and_then(serde_json::Value::as_str) != Some("struct") {
return Err(TableCatalogStoreError::Invalid("create schema type must be struct".to_string()));
}
let fields = schema
.get_mut("fields")
.and_then(serde_json::Value::as_array_mut)
.ok_or_else(|| TableCatalogStoreError::Invalid("create schema fields must be an array".to_string()))?;
self.assign_struct_fields(fields, 0)
}
fn assign_struct_fields(&mut self, fields: &mut [serde_json::Value], depth: usize) -> TableCatalogStoreResult<()> {
for field in fields.iter_mut() {
let field = field
.as_object_mut()
.ok_or_else(|| TableCatalogStoreError::Invalid("create schema fields must be JSON objects".to_string()))?;
self.assign_object_id(field, "id", "create schema field id")?;
}
for field in fields {
let field = field
.as_object_mut()
.ok_or_else(|| TableCatalogStoreError::Invalid("create schema fields must be JSON objects".to_string()))?;
let field_type = field
.get_mut("type")
.ok_or_else(|| TableCatalogStoreError::Invalid("create schema field type is required".to_string()))?;
self.assign_type_ids(field_type, depth)?;
}
Ok(())
}
fn assign_type_ids(&mut self, field_type: &mut serde_json::Value, depth: usize) -> TableCatalogStoreResult<()> {
if field_type.is_string() {
return Ok(());
}
if depth >= ICEBERG_MAX_SCHEMA_NESTING_DEPTH {
return Err(TableCatalogStoreError::Invalid(
"create schema exceeds the maximum nesting depth".to_string(),
));
}
let nested_depth = depth + 1;
let field_type = field_type.as_object_mut().ok_or_else(|| {
TableCatalogStoreError::Invalid("create schema field type must be a string or JSON object".to_string())
})?;
match field_type.get("type").and_then(serde_json::Value::as_str) {
Some("struct") => {
let fields = field_type
.get_mut("fields")
.and_then(serde_json::Value::as_array_mut)
.ok_or_else(|| TableCatalogStoreError::Invalid("create schema struct fields must be an array".to_string()))?;
self.assign_struct_fields(fields, nested_depth)
}
Some("list") => {
self.assign_object_id(field_type, "element-id", "create schema list element-id")?;
let element = field_type
.get_mut("element")
.ok_or_else(|| TableCatalogStoreError::Invalid("create schema list element is required".to_string()))?;
self.assign_type_ids(element, nested_depth)
}
Some("map") => {
self.assign_object_id(field_type, "key-id", "create schema map key-id")?;
self.assign_object_id(field_type, "value-id", "create schema map value-id")?;
let key = field_type
.get_mut("key")
.ok_or_else(|| TableCatalogStoreError::Invalid("create schema map key is required".to_string()))?;
self.assign_type_ids(key, nested_depth)?;
let value = field_type
.get_mut("value")
.ok_or_else(|| TableCatalogStoreError::Invalid("create schema map value is required".to_string()))?;
self.assign_type_ids(value, nested_depth)
}
_ => Err(TableCatalogStoreError::Invalid(
"create schema contains an unsupported field type".to_string(),
)),
}
}
fn assign_object_id(
&mut self,
object: &mut serde_json::Map<String, serde_json::Value>,
field: &str,
label: &str,
) -> TableCatalogStoreResult<()> {
let old_id = required_i32_value(object, field, label)?;
let entry = match self.old_to_new.entry(old_id) {
std::collections::btree_map::Entry::Occupied(_) => {
return Err(TableCatalogStoreError::Invalid(format!("duplicate create schema field id {old_id}")));
}
std::collections::btree_map::Entry::Vacant(entry) => entry,
};
let new_id = self.next_id;
if new_id > ICEBERG_MAX_USER_FIELD_ID {
return Err(TableCatalogStoreError::Invalid(
"create schema exceeds the available Iceberg field ID range".to_string(),
));
}
self.next_id = new_id.checked_add(1).ok_or_else(|| {
TableCatalogStoreError::Invalid("create schema exceeds the available Iceberg field ID range".to_string())
})?;
entry.insert(new_id);
object.insert(field.to_string(), serde_json::Value::from(new_id));
Ok(())
}
fn remap_identifier_field_ids(&self, schema: &mut serde_json::Value) -> TableCatalogStoreResult<()> {
let Some(identifier_field_ids) = schema
.as_object_mut()
.and_then(|schema| schema.get_mut("identifier-field-ids"))
else {
return Ok(());
};
let identifier_field_ids = identifier_field_ids
.as_array_mut()
.ok_or_else(|| TableCatalogStoreError::Invalid("create schema identifier-field-ids must be an array".to_string()))?;
for field_id in identifier_field_ids {
let old_id = required_i32(field_id, "create schema identifier field id")?;
let new_id = self.old_to_new.get(&old_id).ok_or_else(|| {
TableCatalogStoreError::Invalid(format!(
"create schema identifier field id {old_id} does not reference a schema field"
))
})?;
*field_id = serde_json::Value::from(*new_id);
}
Ok(())
}
fn remap_source_ids(&self, value: &mut serde_json::Value, label: &str) -> TableCatalogStoreResult<()> {
let value = value
.as_object_mut()
.ok_or_else(|| TableCatalogStoreError::Invalid(format!("{label} must be a JSON object")))?;
let Some(fields) = value.get_mut("fields") else {
return Ok(());
};
let fields = fields
.as_array_mut()
.ok_or_else(|| TableCatalogStoreError::Invalid(format!("{label} fields must be an array")))?;
for field in fields {
let field = field
.as_object_mut()
.ok_or_else(|| TableCatalogStoreError::Invalid(format!("{label} fields must be JSON objects")))?;
let old_id = required_i32_value(field, "source-id", &format!("{label} source-id"))?;
let new_id = self.old_to_new.get(&old_id).ok_or_else(|| {
TableCatalogStoreError::Invalid(format!("{label} source-id {old_id} does not reference the create schema"))
})?;
field.insert("source-id".to_string(), serde_json::Value::from(*new_id));
}
Ok(())
}
}
fn validate_iceberg_schema_fields(schema: &serde_json::Value, label: &str) -> TableCatalogStoreResult<IcebergSchemaFields> {
let schema = schema
.as_object()
@@ -2601,35 +2414,11 @@ struct SnapshotGraphManifestLocation {
sequence_number: Option<i64>,
min_sequence_number: Option<i64>,
added_snapshot_id: Option<i64>,
#[allow(
dead_code,
reason = "exercised by table_catalog/tests.rs; the lib target cannot see test-only consumers (backlog#1823)"
)]
added_files_count: Option<u64>,
#[allow(
dead_code,
reason = "exercised by table_catalog/tests.rs; the lib target cannot see test-only consumers (backlog#1823)"
)]
existing_files_count: Option<u64>,
#[allow(
dead_code,
reason = "exercised by table_catalog/tests.rs; the lib target cannot see test-only consumers (backlog#1823)"
)]
deleted_files_count: Option<u64>,
#[allow(
dead_code,
reason = "exercised by table_catalog/tests.rs; the lib target cannot see test-only consumers (backlog#1823)"
)]
added_rows_count: Option<u64>,
#[allow(
dead_code,
reason = "exercised by table_catalog/tests.rs; the lib target cannot see test-only consumers (backlog#1823)"
)]
existing_rows_count: Option<u64>,
#[allow(
dead_code,
reason = "exercised by table_catalog/tests.rs; the lib target cannot see test-only consumers (backlog#1823)"
)]
deleted_rows_count: Option<u64>,
from_manifest_list: bool,
}
-48
View File
@@ -87,10 +87,6 @@ impl Namespace {
}
#[derive(Debug, Clone, PartialEq, Eq)]
#[allow(
dead_code,
reason = "exercised by table_catalog/tests.rs; the lib target cannot see test-only consumers (backlog#1823)"
)]
pub struct TableIdentifier {
warehouse: IdentifierSegment,
namespace: Namespace,
@@ -98,10 +94,6 @@ pub struct TableIdentifier {
}
impl TableIdentifier {
#[allow(
dead_code,
reason = "exercised by table_catalog/tests.rs; the lib target cannot see test-only consumers (backlog#1823)"
)]
pub fn new(warehouse: IdentifierSegment, namespace: Namespace, name: IdentifierSegment) -> Self {
Self {
warehouse,
@@ -124,10 +116,6 @@ impl TableIdentifier {
}
#[derive(Debug, Clone, PartialEq, Eq)]
#[allow(
dead_code,
reason = "exercised by table_catalog/tests.rs; the lib target cannot see test-only consumers (backlog#1823)"
)]
pub struct TablePathResolver {
reserved_prefix: &'static str,
}
@@ -141,26 +129,14 @@ impl Default for TablePathResolver {
}
impl TablePathResolver {
#[allow(
dead_code,
reason = "exercised by table_catalog/tests.rs; the lib target cannot see test-only consumers (backlog#1823)"
)]
pub fn current_pointer_path(&self, table: &TableIdentifier) -> String {
format!("{}/{}", self.table_root(table), CURRENT_POINTER_FILE)
}
#[allow(
dead_code,
reason = "exercised by table_catalog/tests.rs; the lib target cannot see test-only consumers (backlog#1823)"
)]
pub fn metadata_dir_path(&self, table: &TableIdentifier) -> String {
format!("{}/{}", self.table_root(table), METADATA_DIR)
}
#[allow(
dead_code,
reason = "exercised by table_catalog/tests.rs; the lib target cannot see test-only consumers (backlog#1823)"
)]
pub fn metadata_file_path(&self, table: &TableIdentifier, metadata_file_name: &str) -> String {
format!("{}/{}", self.metadata_dir_path(table), metadata_file_name)
}
@@ -193,10 +169,6 @@ pub(crate) fn default_namespace_root_prefix() -> String {
)
}
#[allow(
dead_code,
reason = "exercised by table_catalog/tests.rs; the lib target cannot see test-only consumers (backlog#1823)"
)]
pub(crate) fn default_namespace_marker_path(namespace: &Namespace) -> String {
format!("{}{}/{}", default_namespace_root_prefix(), namespace.storage_id(), NAMESPACE_MARKER_FILE)
}
@@ -213,10 +185,6 @@ pub(crate) fn default_table_bucket_publication_lock_path() -> String {
rustfs_common::table_catalog::TABLE_BUCKET_PUBLICATION_LOCK_PATH.to_string()
}
#[allow(
dead_code,
reason = "exercised by table_catalog/tests.rs; the lib target cannot see test-only consumers (backlog#1823)"
)]
pub(crate) fn default_table_marker_path(namespace: &Namespace, table: &IdentifierSegment) -> String {
format!("{}{}/{}", default_table_root_prefix(namespace), table.as_str(), TABLE_MARKER_FILE)
}
@@ -257,26 +225,14 @@ pub(crate) fn default_table_metadata_file_path(
format!("{}/{}", default_table_metadata_dir_path(namespace, table), metadata_file_name)
}
#[allow(
dead_code,
reason = "exercised by table_catalog/tests.rs; the lib target cannot see test-only consumers (backlog#1823)"
)]
pub(crate) fn default_table_current_pointer_path(namespace: &Namespace, table: &IdentifierSegment) -> String {
format!("{}{}/{}", default_table_root_prefix(namespace), table.as_str(), CURRENT_POINTER_FILE)
}
#[allow(
dead_code,
reason = "exercised by table_catalog/tests.rs; the lib target cannot see test-only consumers (backlog#1823)"
)]
pub(crate) fn default_table_lifecycle_path(namespace: &Namespace, table: &IdentifierSegment) -> String {
format!("{}{}/{}", default_table_root_prefix(namespace), table.as_str(), LIFECYCLE_FILE)
}
#[allow(
dead_code,
reason = "exercised by table_catalog/tests.rs; the lib target cannot see test-only consumers (backlog#1823)"
)]
pub(crate) fn namespace_name_from_marker_path(object_key: &str) -> Option<String> {
let prefix = default_namespace_root_prefix();
let suffix = format!("/{NAMESPACE_MARKER_FILE}");
@@ -288,10 +244,6 @@ pub(crate) fn namespace_name_from_marker_path(object_key: &str) -> Option<String
.map(|value| value.replace('/', "."))
}
#[allow(
dead_code,
reason = "exercised by table_catalog/tests.rs; the lib target cannot see test-only consumers (backlog#1823)"
)]
pub(crate) fn table_name_from_marker_path(namespace: &Namespace, object_key: &str) -> Option<String> {
let prefix = default_table_root_prefix(namespace);
let suffix = format!("/{TABLE_MARKER_FILE}");
+2 -28
View File
@@ -18,6 +18,8 @@
//! S3 object behavior. It defines the stable internal boundary that later
//! catalog routes and object guards can share.
#![allow(dead_code)]
use std::{
collections::{BTreeMap, BTreeSet},
num::NonZeroUsize,
@@ -85,20 +87,8 @@ pub(crate) const RESERVED_CATALOG_OBJECT_MESSAGE: &str = "Object key is reserved
pub(crate) const TABLE_BUCKET_CATALOG_TYPE: &str = "iceberg-rest";
pub(crate) const TABLE_BUCKET_CONFIG_VERSION: u16 = 1;
pub(crate) const DEFAULT_WAREHOUSE_ID: &str = "default";
#[allow(
dead_code,
reason = "exercised by table_catalog/tests.rs; the lib target cannot see test-only consumers (backlog#1823)"
)]
pub(crate) const TABLE_NAMESPACE_MARKER_VERSION: u16 = 1;
#[allow(
dead_code,
reason = "exercised by table_catalog/tests.rs; the lib target cannot see test-only consumers (backlog#1823)"
)]
pub(crate) const TABLE_RESOURCE_MARKER_VERSION: u16 = 1;
#[allow(
dead_code,
reason = "exercised by table_catalog/tests.rs; the lib target cannot see test-only consumers (backlog#1823)"
)]
pub(crate) const TABLE_METADATA_POINTER_VERSION: u16 = 1;
pub(crate) const TABLE_CATALOG_ENTRY_VERSION: u16 = 1;
pub(crate) const TABLE_WAREHOUSE_INDEX_STATE_VERSION: u16 = 2;
@@ -133,25 +123,9 @@ const WAREHOUSE_ROOT: &str = "warehouses";
const NAMESPACE_ROOT: &str = "namespaces";
const TABLE_ROOT: &str = "tables";
const VIEW_ROOT: &str = "views";
#[allow(
dead_code,
reason = "exercised by table_catalog/tests.rs; the lib target cannot see test-only consumers (backlog#1823)"
)]
const NAMESPACE_MARKER_FILE: &str = "namespace.json";
#[allow(
dead_code,
reason = "exercised by table_catalog/tests.rs; the lib target cannot see test-only consumers (backlog#1823)"
)]
const TABLE_MARKER_FILE: &str = "table.json";
#[allow(
dead_code,
reason = "exercised by table_catalog/tests.rs; the lib target cannot see test-only consumers (backlog#1823)"
)]
const CURRENT_POINTER_FILE: &str = "current.json";
#[allow(
dead_code,
reason = "exercised by table_catalog/tests.rs; the lib target cannot see test-only consumers (backlog#1823)"
)]
const LIFECYCLE_FILE: &str = "lifecycle.json";
const METADATA_DIR: &str = "metadata";
const DATA_DIR: &str = "data";
-52
View File
@@ -1076,10 +1076,6 @@ pub(crate) enum TableCatalogBackingKind {
#[serde(rename_all = "SCREAMING_SNAKE_CASE")]
pub(crate) enum TableCatalogAuthority {
RustfsSysObject,
#[allow(
dead_code,
reason = "exercised by table_catalog/tests.rs; the lib target cannot see test-only consumers (backlog#1823)"
)]
LinearizableMetadataKv,
}
@@ -1087,10 +1083,6 @@ pub(crate) enum TableCatalogAuthority {
#[serde(rename_all = "SCREAMING_SNAKE_CASE")]
pub(crate) enum TableCatalogConsistencyMode {
ConditionalObjectCas,
#[allow(
dead_code,
reason = "exercised by table_catalog/tests.rs; the lib target cannot see test-only consumers (backlog#1823)"
)]
LinearizableCas,
}
@@ -1098,10 +1090,6 @@ pub(crate) enum TableCatalogConsistencyMode {
#[serde(rename_all = "SCREAMING_SNAKE_CASE")]
pub(crate) enum TableCatalogDurabilityMode {
StagedCommitLogBeforePointerUpdate,
#[allow(
dead_code,
reason = "exercised by table_catalog/tests.rs; the lib target cannot see test-only consumers (backlog#1823)"
)]
WalBeforeStateMachineApply,
}
@@ -1421,20 +1409,12 @@ pub(crate) struct TableCommitRecoveryReport {
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
#[allow(
dead_code,
reason = "exercised by table_catalog/tests.rs; the lib target cannot see test-only consumers (backlog#1823)"
)]
pub(crate) struct NamespaceMarker {
pub version: u16,
pub namespace: String,
}
impl NamespaceMarker {
#[allow(
dead_code,
reason = "exercised by table_catalog/tests.rs; the lib target cannot see test-only consumers (backlog#1823)"
)]
pub fn new(namespace: &Namespace) -> Self {
Self {
version: TABLE_NAMESPACE_MARKER_VERSION,
@@ -1443,19 +1423,11 @@ impl NamespaceMarker {
}
}
#[allow(
dead_code,
reason = "exercised by table_catalog/tests.rs; the lib target cannot see test-only consumers (backlog#1823)"
)]
pub(crate) fn namespace_marker_json(namespace: &Namespace) -> Result<Vec<u8>, serde_json::Error> {
serde_json::to_vec(&NamespaceMarker::new(namespace))
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
#[allow(
dead_code,
reason = "exercised by table_catalog/tests.rs; the lib target cannot see test-only consumers (backlog#1823)"
)]
pub(crate) struct TableMarker {
pub version: u16,
pub namespace: String,
@@ -1464,10 +1436,6 @@ pub(crate) struct TableMarker {
}
impl TableMarker {
#[allow(
dead_code,
reason = "exercised by table_catalog/tests.rs; the lib target cannot see test-only consumers (backlog#1823)"
)]
pub fn new(namespace: &Namespace, table: &IdentifierSegment) -> Self {
Self {
version: TABLE_RESOURCE_MARKER_VERSION,
@@ -1478,29 +1446,17 @@ impl TableMarker {
}
}
#[allow(
dead_code,
reason = "exercised by table_catalog/tests.rs; the lib target cannot see test-only consumers (backlog#1823)"
)]
pub(crate) fn table_marker_json(namespace: &Namespace, table: &IdentifierSegment) -> Result<Vec<u8>, serde_json::Error> {
serde_json::to_vec(&TableMarker::new(namespace, table))
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[allow(
dead_code,
reason = "exercised by table_catalog/tests.rs; the lib target cannot see test-only consumers (backlog#1823)"
)]
pub(crate) struct TableMetadataPointer {
pub version: u16,
pub metadata_location: String,
}
impl TableMetadataPointer {
#[allow(
dead_code,
reason = "exercised by table_catalog/tests.rs; the lib target cannot see test-only consumers (backlog#1823)"
)]
pub fn new(metadata_location: String) -> Self {
Self {
version: TABLE_METADATA_POINTER_VERSION,
@@ -1509,18 +1465,10 @@ impl TableMetadataPointer {
}
}
#[allow(
dead_code,
reason = "exercised by table_catalog/tests.rs; the lib target cannot see test-only consumers (backlog#1823)"
)]
pub(crate) fn table_metadata_pointer_json(metadata_location: String) -> Result<Vec<u8>, serde_json::Error> {
serde_json::to_vec(&TableMetadataPointer::new(metadata_location))
}
#[allow(
dead_code,
reason = "exercised by table_catalog/tests.rs; the lib target cannot see test-only consumers (backlog#1823)"
)]
pub(crate) fn parse_table_metadata_pointer(data: &[u8]) -> Result<TableMetadataPointer, serde_json::Error> {
serde_json::from_slice(data)
}
-36
View File
@@ -164,10 +164,6 @@ pub(crate) trait TableCatalogStore: Send + Sync {
))
}
#[allow(
dead_code,
reason = "exercised by table_catalog/tests.rs; the lib target cannot see test-only consumers (backlog#1823)"
)]
async fn list_namespaces_page(
&self,
table_bucket: &str,
@@ -197,16 +193,8 @@ pub(crate) trait TableCatalogStore: Send + Sync {
async fn drop_namespace(&self, table_bucket: &str, namespace: &str) -> TableCatalogStoreResult<()>;
#[allow(
dead_code,
reason = "declared trait method: implementors provide it but no caller dispatches through the trait yet (backlog#1823)"
)]
async fn create_table(&self, entry: TableEntry) -> TableCatalogStoreResult<()>;
#[allow(
dead_code,
reason = "declared trait method: implementors provide it but no caller dispatches through the trait yet (backlog#1823)"
)]
async fn register_table(&self, entry: TableEntry) -> TableCatalogStoreResult<()>;
async fn register_table_with_publication(
@@ -262,10 +250,6 @@ pub(crate) trait TableCatalogStore: Send + Sync {
///
/// Callers publishing client-supplied Iceberg metadata must validate its logical shape and the physical graph of
/// newly introduced or changed snapshots before invoking this persistence boundary.
#[allow(
dead_code,
reason = "declared trait method: implementors provide it but no caller dispatches through the trait yet (backlog#1823)"
)]
async fn commit_table(&self, request: TableCommitRequest) -> TableCatalogStoreResult<TableCommitResult>;
async fn commit_table_with_publication(
@@ -511,10 +495,6 @@ pub(crate) struct TableCatalogLockGuard {
}
impl TableCatalogLockGuard {
#[allow(
dead_code,
reason = "exercised by table_catalog/tests.rs; the lib target cannot see test-only consumers (backlog#1823)"
)]
pub(crate) fn stable(guard: impl Send + 'static) -> Self {
Self {
_guard: Box::new(guard),
@@ -635,10 +615,6 @@ pub(crate) trait TableCatalogObjectBackend: Clone + Send + Sync + 'static {
async fn object_exists(&self, bucket: &str, object: &str) -> TableCatalogStoreResult<bool>;
#[allow(
dead_code,
reason = "exercised by table_catalog/tests.rs; the lib target cannot see test-only consumers (backlog#1823)"
)]
async fn object_exists_unlocked(&self, bucket: &str, object: &str) -> TableCatalogStoreResult<bool> {
self.object_exists(bucket, object).await
}
@@ -1030,10 +1006,6 @@ where
}
}
#[allow(
dead_code,
reason = "exercised by table_catalog/tests.rs; the lib target cannot see test-only consumers (backlog#1823)"
)]
pub(crate) fn backing_mode(&self) -> TableCatalogBackingMode {
match self {
Self::ObjectBacked(_) => TableCatalogBackingMode::ObjectBacked,
@@ -1557,10 +1529,6 @@ where
}
}
#[allow(
dead_code,
reason = "exercised by table_catalog/tests.rs; the lib target cannot see test-only consumers (backlog#1823)"
)]
pub(crate) async fn get_external_catalog_bridge(
&self,
table_bucket: &str,
@@ -1573,10 +1541,6 @@ where
}
}
#[allow(
dead_code,
reason = "exercised by table_catalog/tests.rs; the lib target cannot see test-only consumers (backlog#1823)"
)]
pub(crate) async fn put_external_catalog_bridge(
&self,
entry: ExternalCatalogBridgeEntry,
-24
View File
@@ -1270,10 +1270,6 @@ where
Ok(Some((entry, etag)))
}
#[allow(
dead_code,
reason = "exercised by table_catalog/tests.rs; the lib target cannot see test-only consumers (backlog#1823)"
)]
async fn write_table_entry(
&self,
entry: TableEntry,
@@ -1694,10 +1690,6 @@ where
Ok(config)
}
#[allow(
dead_code,
reason = "exercised by table_catalog/tests.rs; the lib target cannot see test-only consumers (backlog#1823)"
)]
pub(crate) async fn put_table_bucket_maintenance_config(
&self,
table_bucket: &str,
@@ -3014,10 +3006,6 @@ where
table_compaction_planning_report(&self.backend, table_bucket, &namespace, &table, &entry, &current_metadata, config).await
}
#[allow(
dead_code,
reason = "exercised by table_catalog/tests.rs; the lib target cannot see test-only consumers (backlog#1823)"
)]
pub(crate) async fn commit_table_compaction(
&self,
table_bucket: &str,
@@ -3571,10 +3559,6 @@ where
Ok(report)
}
#[allow(
dead_code,
reason = "exercised by table_catalog/tests.rs; the lib target cannot see test-only consumers (backlog#1823)"
)]
pub(crate) async fn delete_table_metadata_maintenance_candidates(
&self,
table_bucket: &str,
@@ -3589,10 +3573,6 @@ where
.await
}
#[allow(
dead_code,
reason = "exercised by table_catalog/tests.rs; the lib target cannot see test-only consumers (backlog#1823)"
)]
pub(crate) async fn run_table_metadata_maintenance(
&self,
table_bucket: &str,
@@ -3774,10 +3754,6 @@ where
Ok(report)
}
#[allow(
dead_code,
reason = "exercised by table_catalog/tests.rs; the lib target cannot see test-only consumers (backlog#1823)"
)]
pub(in crate::table_catalog) async fn delete_table_metadata_maintenance_report(
&self,
table_bucket: &str,