mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-16 18:08:21 +00:00
Compare commits
12 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| c8e7bb0fca | |||
| 9e6e02ea09 | |||
| 39274fc37c | |||
| 33eff4c3c4 | |||
| a2f16aa066 | |||
| 4c8b9f87e1 | |||
| 3272730c13 | |||
| 1862112d0c | |||
| cd0ac02879 | |||
| 6cf9cf7bb5 | |||
| f1f86ee9d0 | |||
| 1eef0de003 |
@@ -11,7 +11,6 @@
|
||||
// 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,7 +82,6 @@ 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,11 +126,23 @@ 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>;
|
||||
@@ -140,9 +152,21 @@ 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.";
|
||||
@@ -280,6 +304,10 @@ 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();
|
||||
@@ -570,6 +598,10 @@ 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>>,
|
||||
@@ -2868,6 +2900,10 @@ 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
|
||||
}
|
||||
@@ -3352,6 +3388,10 @@ 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;
|
||||
@@ -4339,6 +4379,10 @@ 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();
|
||||
@@ -4373,6 +4417,10 @@ 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 || {
|
||||
@@ -4391,6 +4439,10 @@ 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,
|
||||
@@ -5145,6 +5197,10 @@ 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 {
|
||||
@@ -7422,6 +7478,10 @@ 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,6 +759,10 @@ 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)
|
||||
}
|
||||
@@ -1257,6 +1261,10 @@ 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,
|
||||
@@ -1320,6 +1328,10 @@ 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,
|
||||
@@ -1455,6 +1467,10 @@ 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,25 +15,35 @@
|
||||
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;
|
||||
|
||||
pub(crate) type LifecycleReplicationConfig = ReplicationLifecycleConfig;
|
||||
|
||||
#[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(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,6 +14,10 @@
|
||||
|
||||
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,6 +331,10 @@ 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,6 +148,10 @@ 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,6 +385,10 @@ 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
|
||||
@@ -395,6 +399,10 @@ 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) {
|
||||
@@ -405,6 +413,10 @@ 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,
|
||||
@@ -485,6 +497,10 @@ 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,8 +50,16 @@ 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),
|
||||
|
||||
@@ -60,12 +60,14 @@ 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(),
|
||||
@@ -84,6 +86,7 @@ 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
|
||||
@@ -890,6 +893,10 @@ 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;
|
||||
@@ -1104,6 +1111,10 @@ 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;
|
||||
@@ -1617,6 +1628,7 @@ 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,
|
||||
@@ -1721,6 +1733,7 @@ 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"));
|
||||
|
||||
@@ -13,7 +13,6 @@
|
||||
// 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,6 +136,7 @@ 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)
|
||||
@@ -151,6 +152,7 @@ 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 {
|
||||
|
||||
@@ -193,6 +193,7 @@ 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,
|
||||
@@ -208,6 +209,7 @@ 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,6 +899,7 @@ 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);
|
||||
}
|
||||
|
||||
@@ -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, ReplicationLifecycleConfig};
|
||||
pub(crate) use replication_lifecycle_bridge::ReplicationLifecycleBridge;
|
||||
pub(crate) use replication_migration_bridge::ReplicationMigrationBridge;
|
||||
pub use replication_object_bridge::ReplicationObjectBridge;
|
||||
pub use replication_object_config::{DeleteReplicationConfigSnapshot, ReplicationConfig};
|
||||
|
||||
@@ -37,6 +37,10 @@ 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,15 +24,27 @@ 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,
|
||||
@@ -45,6 +57,10 @@ 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,
|
||||
@@ -54,6 +70,10 @@ 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,17 +19,33 @@ 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,6 +77,10 @@ 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,6 +231,10 @@ 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,6 +217,10 @@ 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,
|
||||
@@ -712,6 +716,10 @@ 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
|
||||
@@ -940,6 +948,10 @@ 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,
|
||||
@@ -1180,6 +1192,10 @@ 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;
|
||||
}
|
||||
@@ -1651,6 +1667,10 @@ 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>,
|
||||
@@ -1664,6 +1684,10 @@ 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>,
|
||||
@@ -1678,6 +1702,10 @@ 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>,
|
||||
@@ -1691,6 +1719,10 @@ 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,11 +21,31 @@ 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,6 +122,10 @@ 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();
|
||||
@@ -328,6 +332,10 @@ 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>,
|
||||
@@ -544,6 +552,10 @@ 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,6 +340,10 @@ 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,6 +59,10 @@ 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,
|
||||
|
||||
@@ -73,23 +73,6 @@ 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>,
|
||||
@@ -100,6 +83,10 @@ 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);
|
||||
{
|
||||
@@ -186,15 +173,27 @@ 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",
|
||||
@@ -204,6 +203,10 @@ 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,6 +90,10 @@ 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());
|
||||
|
||||
@@ -637,14 +637,23 @@ 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);
|
||||
@@ -770,6 +779,7 @@ 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);
|
||||
}
|
||||
@@ -850,6 +860,7 @@ 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 {
|
||||
@@ -980,11 +991,13 @@ 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)
|
||||
}
|
||||
@@ -1026,21 +1039,6 @@ 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)]
|
||||
@@ -1072,10 +1070,6 @@ 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,
|
||||
@@ -1438,20 +1432,6 @@ 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 {
|
||||
|
||||
@@ -48,6 +48,7 @@ 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(),
|
||||
|
||||
@@ -178,6 +178,7 @@ 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 {
|
||||
|
||||
@@ -665,6 +665,7 @@ 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<()>,
|
||||
@@ -1016,13 +1017,29 @@ 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).
|
||||
@@ -1036,7 +1053,15 @@ 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;
|
||||
@@ -1095,12 +1120,14 @@ 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);
|
||||
}
|
||||
@@ -1456,6 +1483,7 @@ 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);
|
||||
}
|
||||
@@ -1673,12 +1701,20 @@ 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
|
||||
@@ -1687,6 +1723,10 @@ 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)
|
||||
@@ -2142,6 +2182,7 @@ 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()
|
||||
@@ -2447,6 +2488,10 @@ 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,
|
||||
@@ -2454,6 +2499,10 @@ 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,
|
||||
@@ -2519,6 +2568,10 @@ 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.
|
||||
@@ -3071,6 +3124,7 @@ 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,
|
||||
@@ -4704,6 +4758,10 @@ 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,
|
||||
@@ -5490,6 +5548,7 @@ 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();
|
||||
@@ -5567,15 +5626,24 @@ 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();
|
||||
@@ -6488,6 +6556,7 @@ 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)
|
||||
|
||||
@@ -13,7 +13,6 @@
|
||||
// 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;
|
||||
@@ -1114,6 +1113,10 @@ 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,
|
||||
@@ -1372,6 +1375,7 @@ 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)
|
||||
}
|
||||
|
||||
@@ -571,6 +571,10 @@ 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?;
|
||||
@@ -583,6 +587,7 @@ 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
|
||||
}
|
||||
@@ -1809,10 +1814,6 @@ 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,
|
||||
@@ -2858,13 +2859,6 @@ 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
|
||||
|
||||
@@ -704,6 +704,7 @@ 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>,
|
||||
|
||||
@@ -180,11 +180,13 @@ 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),
|
||||
@@ -666,6 +668,60 @@ 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);
|
||||
@@ -703,51 +759,7 @@ pub(in crate::set_disk) async fn data_read_early_stop_inline_body_miss_reason(
|
||||
if !can_try_inline_data_shards_direct(object_size, candidate.erasure.block_size) {
|
||||
return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_SIZE);
|
||||
}
|
||||
|
||||
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),
|
||||
}
|
||||
None
|
||||
}
|
||||
|
||||
fn data_read_inline_missing_shards_are_pending(
|
||||
@@ -1836,6 +1848,7 @@ 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>],
|
||||
@@ -2126,6 +2139,7 @@ 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,
|
||||
@@ -2442,6 +2456,7 @@ 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;
|
||||
@@ -2449,10 +2464,14 @@ 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 {
|
||||
@@ -2548,6 +2567,7 @@ 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;
|
||||
@@ -2555,6 +2575,7 @@ 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 {
|
||||
@@ -2563,6 +2584,9 @@ 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 {
|
||||
@@ -2596,6 +2620,14 @@ 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;
|
||||
}
|
||||
@@ -2955,6 +2987,7 @@ 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,
|
||||
@@ -2988,14 +3021,11 @@ 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() {
|
||||
@@ -3134,6 +3164,7 @@ 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,
|
||||
@@ -3143,6 +3174,7 @@ 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,
|
||||
@@ -3261,6 +3293,7 @@ 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,
|
||||
@@ -5073,6 +5106,7 @@ 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,
|
||||
@@ -5297,6 +5331,7 @@ 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";
|
||||
}
|
||||
|
||||
@@ -5734,6 +5769,130 @@ 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
|
||||
@@ -7081,7 +7240,7 @@ mod tests {
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
||||
async fn bounded_non_inline_data_get_hedges_then_waits_for_full_fanout() {
|
||||
async fn bounded_non_inline_data_get_immediately_forces_full_fanout() {
|
||||
const DISKS: usize = 4;
|
||||
let bucket = "bounded-data-get-hedge-bucket";
|
||||
let object = "bounded-data-get-hedge-object";
|
||||
@@ -7112,7 +7271,7 @@ mod tests {
|
||||
}
|
||||
})
|
||||
.await
|
||||
.expect("bounded data-read fanout should hedge by starting the spare disk");
|
||||
.expect("bounded non-inline data-read fanout should immediately schedule the spare disk");
|
||||
|
||||
let pending = tokio::time::timeout(BARRIER_PAUSE_GUARD, &mut read).await;
|
||||
assert!(
|
||||
@@ -7128,7 +7287,7 @@ mod tests {
|
||||
assert_eq!(
|
||||
calls.total(disk_call_counters::KIND_READ_VERSION),
|
||||
DISKS as u64,
|
||||
"bounded data-read fanout should issue the paused disk plus one spare hedge"
|
||||
"bounded non-inline data-read fanout should issue the paused disk plus the remaining spare"
|
||||
);
|
||||
assert_eq!(diagnostics.total_responses(), DISKS);
|
||||
assert_eq!(parts_metadata.iter().filter(|fi| fi.name == object).count(), DISKS);
|
||||
@@ -7155,16 +7314,42 @@ 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 (parts_metadata, errs, diagnostics) =
|
||||
SetDisks::read_all_fileinfo_observed(&disks, bucket, bucket, object, "", true, false, false, true, 2)
|
||||
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)
|
||||
.await
|
||||
.expect("default data-read metadata should resolve");
|
||||
});
|
||||
|
||||
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");
|
||||
|
||||
assert_eq!(
|
||||
calls.total(disk_call_counters::KIND_READ_VERSION),
|
||||
DISKS as u64,
|
||||
"default non-inline GET data-read metadata must keep full fanout for read-failure tolerance"
|
||||
"default non-inline GET data-read metadata must keep full fanout without waiting for a quorum miss first"
|
||||
);
|
||||
assert_eq!(diagnostics.total_responses(), DISKS);
|
||||
assert_eq!(parts_metadata.iter().filter(|fi| fi.name == object).count(), DISKS);
|
||||
|
||||
@@ -42,12 +42,20 @@ 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
|
||||
}
|
||||
@@ -56,14 +64,26 @@ 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
|
||||
}
|
||||
@@ -72,6 +92,10 @@ 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
|
||||
}
|
||||
@@ -84,6 +108,10 @@ 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
|
||||
}
|
||||
|
||||
@@ -39,7 +39,6 @@
|
||||
//! - `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)]
|
||||
|
||||
@@ -103,9 +102,7 @@ use crate::storage_api_contracts::{
|
||||
};
|
||||
use crate::store::utils::is_reserved_or_invalid_bucket;
|
||||
use crate::{
|
||||
bucket::lifecycle::bucket_lifecycle_ops::{
|
||||
LifecycleOps, gen_transition_objname, get_transitioned_object_reader_with_tier_manager, put_restore_opts,
|
||||
},
|
||||
bucket::lifecycle::bucket_lifecycle_ops::{LifecycleOps, get_transitioned_object_reader_with_tier_manager, put_restore_opts},
|
||||
cache_value::metacache_set::{ListPathRawOptions, list_path_raw},
|
||||
config::storageclass,
|
||||
disk::{
|
||||
@@ -177,15 +174,14 @@ 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;
|
||||
@@ -624,7 +620,9 @@ 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
|
||||
@@ -698,7 +696,15 @@ 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";
|
||||
@@ -708,7 +714,12 @@ 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 = false;
|
||||
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";
|
||||
|
||||
// --- Multipart Reader-Setup Prefetch Configuration (backlog#870) ---
|
||||
|
||||
@@ -910,6 +921,10 @@ 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)
|
||||
@@ -1110,7 +1125,10 @@ mod prepared_get_object_metadata_tests {
|
||||
|
||||
assert_eq!(object_size, payload.len() as i64);
|
||||
assert_eq!(restored, payload);
|
||||
assert_eq!(calls_total, 4, "default production GET should eagerly schedule the full metadata fanout");
|
||||
assert_eq!(
|
||||
calls_total, 4,
|
||||
"default production inline GET should schedule the initial bounded quorum plus one hedge"
|
||||
);
|
||||
assert_eq!(
|
||||
recorder.histogram_values(
|
||||
"rustfs_io_get_object_metadata_fanout_scheduled",
|
||||
@@ -1684,6 +1702,95 @@ 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).
|
||||
///
|
||||
@@ -1709,6 +1816,10 @@ 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(|| {
|
||||
@@ -1748,6 +1859,10 @@ 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();
|
||||
@@ -2181,6 +2296,7 @@ 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,
|
||||
@@ -2786,6 +2902,7 @@ 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,
|
||||
@@ -3061,11 +3178,13 @@ 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)
|
||||
@@ -3659,6 +3778,7 @@ 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
|
||||
@@ -4237,6 +4357,7 @@ 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())
|
||||
}
|
||||
@@ -4512,15 +4633,6 @@ 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>],
|
||||
@@ -5297,6 +5409,7 @@ 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,
|
||||
@@ -5305,6 +5418,7 @@ 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,
|
||||
|
||||
@@ -453,7 +453,9 @@ impl SetDisks {
|
||||
..Default::default()
|
||||
};
|
||||
|
||||
let write_lock_guard = if !opts.no_lock {
|
||||
// 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 ns_lock = self.new_ns_lock(bucket, object).await?;
|
||||
Some(
|
||||
ns_lock
|
||||
@@ -996,7 +998,7 @@ impl SetDisks {
|
||||
readers.push(None);
|
||||
continue;
|
||||
}
|
||||
Err(e) => {
|
||||
Err(_e) => {
|
||||
readers.push(None);
|
||||
continue;
|
||||
}
|
||||
@@ -1545,6 +1547,9 @@ 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);
|
||||
};
|
||||
@@ -1716,6 +1721,10 @@ 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,6 +66,7 @@ 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 => {
|
||||
@@ -79,6 +80,7 @@ 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,
|
||||
@@ -143,6 +145,7 @@ 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<_>>();
|
||||
@@ -153,6 +156,10 @@ 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
|
||||
@@ -432,6 +439,10 @@ 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,
|
||||
@@ -440,6 +451,10 @@ impl SetDisks {
|
||||
Ok((new_disks, healing > 0))
|
||||
}
|
||||
|
||||
#[allow(
|
||||
dead_code,
|
||||
reason = "reached only from get_online_disk_with_healing, itself uncalled in this port (backlog#1823)"
|
||||
)]
|
||||
pub(in crate::set_disk) async fn get_online_disk_with_healing_and_info(
|
||||
&self,
|
||||
incl_healing: bool,
|
||||
|
||||
@@ -415,6 +415,7 @@ 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();
|
||||
@@ -557,6 +558,7 @@ 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,
|
||||
@@ -1398,7 +1400,7 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
|
||||
|
||||
let mut count = max_parts;
|
||||
|
||||
for (i, part) in object_parts.iter().enumerate() {
|
||||
for part in object_parts.iter() {
|
||||
if let Some(err) = &part.error {
|
||||
warn!("list_object_parts part error: {:?}", &err);
|
||||
}
|
||||
@@ -2041,8 +2043,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={}",
|
||||
p.part_num, bucket, object
|
||||
"complete_multipart_upload checksum add_part failed part_id={}, bucket={}, object={}, err={}",
|
||||
p.part_num, bucket, object, err
|
||||
);
|
||||
return Err(Error::InvalidPart(p.part_num, ext_part.etag.clone(), p.etag.clone().unwrap_or_default()));
|
||||
}
|
||||
@@ -2087,8 +2089,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={}",
|
||||
wtcs.encoded, checksum.encoded
|
||||
"complete_multipart_upload checksum matches failed want={}, got={}, err={}",
|
||||
wtcs.encoded, checksum.encoded, err
|
||||
);
|
||||
return Err(Error::other(format!(
|
||||
"complete_multipart_upload checksum matches failed want={}, got={}",
|
||||
|
||||
@@ -3507,6 +3507,10 @@ struct TransitionUploadedSaveProbeState {
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
#[allow(
|
||||
dead_code,
|
||||
reason = "installed by set_disk tests behind `--features test-util` (backlog#1823)"
|
||||
)]
|
||||
struct TransitionUploadedSaveProbe {
|
||||
state: Arc<TransitionUploadedSaveProbeState>,
|
||||
}
|
||||
@@ -3517,6 +3521,10 @@ 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(),
|
||||
@@ -3533,6 +3541,10 @@ 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)
|
||||
}
|
||||
@@ -3738,6 +3750,10 @@ struct TransitionCommitBarrierState {
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
#[allow(
|
||||
dead_code,
|
||||
reason = "installed by set_disk tests behind `--features test-util` (backlog#1823)"
|
||||
)]
|
||||
struct TransitionCommitBarrier {
|
||||
state: Arc<TransitionCommitBarrierState>,
|
||||
}
|
||||
@@ -3748,14 +3764,26 @@ 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)
|
||||
}
|
||||
@@ -3778,12 +3806,20 @@ 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();
|
||||
}
|
||||
@@ -5620,7 +5656,9 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
|
||||
// TODO: Lifecycle
|
||||
|
||||
let mut version_found = true;
|
||||
let (mut goi, write_quorum, gerr) = self.get_object_info_and_quorum(bucket, object, &opts).await;
|
||||
// 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;
|
||||
if let Some(err) = &gerr
|
||||
&& goi.name.is_empty()
|
||||
{
|
||||
@@ -6374,7 +6412,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 let Some(disk) = disk {
|
||||
if disk.is_some() {
|
||||
continue;
|
||||
}
|
||||
let _ = self
|
||||
@@ -10170,6 +10208,288 @@ 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,
|
||||
|
||||
@@ -116,6 +116,7 @@ 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),
|
||||
@@ -1826,6 +1827,7 @@ 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()
|
||||
}
|
||||
@@ -3935,7 +3937,7 @@ mod tests {
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn metadata_early_stop_bounded_fanout_defaults_to_disabled() {
|
||||
fn metadata_early_stop_bounded_fanout_defaults_to_enabled() {
|
||||
temp_env::with_vars(
|
||||
[
|
||||
(ENV_RUSTFS_GET_METADATA_EARLY_STOP_ENABLE, Some("true")),
|
||||
@@ -3944,20 +3946,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("true"))], || {
|
||||
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_DATA_READ_EARLY_STOP_ENABLE, Some("false")),
|
||||
(ENV_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT, Some("false")),
|
||||
(ENV_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT, Some("true")),
|
||||
],
|
||||
|| {
|
||||
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());
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
@@ -18,8 +18,6 @@
|
||||
//! 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,14 +34,6 @@ 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,13 +12,6 @@
|
||||
# # 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}"
|
||||
@@ -53,10 +46,7 @@ if [ -n "${MINIO_LAB_PYTHON_IMAGE:-}" ]; then
|
||||
fi
|
||||
|
||||
echo ">> building ${IMAGE}"
|
||||
# ${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}"
|
||||
docker build -f "${SCRIPT_DIR}/Dockerfile" -t "${IMAGE}" "${build_args[@]}" "${SCRIPT_DIR}"
|
||||
|
||||
echo ">> capturing fixtures into ${FIXTURE_REL}"
|
||||
docker run --rm -v "${REPO_ROOT}:/repo" "${IMAGE}" \
|
||||
@@ -64,7 +54,6 @@ docker run --rm -v "${REPO_ROOT}:/repo" "${IMAGE}" \
|
||||
--root "/repo/${FIXTURE_REL}" \
|
||||
--work-root /tmp/minio-lab-work \
|
||||
--minio-binary /usr/local/bin/minio \
|
||||
--endpoint "${MINIO_LAB_ENDPOINT:-http://127.0.0.1:9000}" \
|
||||
${case_args[@]+"${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,
|
||||
schema,
|
||||
partition_spec,
|
||||
write_order,
|
||||
mut schema,
|
||||
mut partition_spec,
|
||||
mut write_order,
|
||||
stage_create,
|
||||
mut properties,
|
||||
} = request;
|
||||
@@ -3031,6 +3031,9 @@ 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,6 +1873,66 @@ 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");
|
||||
@@ -1990,6 +2050,142 @@ 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");
|
||||
@@ -2247,7 +2443,11 @@ 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");
|
||||
|
||||
assert_eq!(metadata["last-column-id"], 9);
|
||||
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);
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
||||
@@ -56,7 +56,6 @@
|
||||
//! ```
|
||||
|
||||
// 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};
|
||||
@@ -264,6 +263,10 @@ 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,
|
||||
|
||||
@@ -31,7 +31,6 @@
|
||||
//! ```
|
||||
|
||||
// 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
|
||||
|
||||
@@ -12,8 +12,6 @@
|
||||
// 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
|
||||
@@ -29,18 +27,6 @@ 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,6 +71,10 @@ 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()
|
||||
@@ -78,6 +82,10 @@ 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>> {
|
||||
@@ -157,6 +165,10 @@ 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)>> {
|
||||
@@ -166,6 +178,10 @@ 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,6 +19,7 @@ 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);
|
||||
@@ -103,6 +104,10 @@ 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))
|
||||
}
|
||||
@@ -1357,6 +1362,188 @@ 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()
|
||||
@@ -2414,11 +2601,35 @@ 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,
|
||||
}
|
||||
|
||||
@@ -87,6 +87,10 @@ 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,
|
||||
@@ -94,6 +98,10 @@ 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,
|
||||
@@ -116,6 +124,10 @@ 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,
|
||||
}
|
||||
@@ -129,14 +141,26 @@ 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)
|
||||
}
|
||||
@@ -169,6 +193,10 @@ 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)
|
||||
}
|
||||
@@ -185,6 +213,10 @@ 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)
|
||||
}
|
||||
@@ -225,14 +257,26 @@ 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}");
|
||||
@@ -244,6 +288,10 @@ 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}");
|
||||
|
||||
@@ -18,8 +18,6 @@
|
||||
//! 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,
|
||||
@@ -87,8 +85,20 @@ 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;
|
||||
@@ -123,9 +133,25 @@ 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";
|
||||
|
||||
@@ -1076,6 +1076,10 @@ 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,
|
||||
}
|
||||
|
||||
@@ -1083,6 +1087,10 @@ 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,
|
||||
}
|
||||
|
||||
@@ -1090,6 +1098,10 @@ 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,
|
||||
}
|
||||
|
||||
@@ -1409,12 +1421,20 @@ 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,
|
||||
@@ -1423,11 +1443,19 @@ 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,
|
||||
@@ -1436,6 +1464,10 @@ 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,
|
||||
@@ -1446,17 +1478,29 @@ 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,
|
||||
@@ -1465,10 +1509,18 @@ 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)
|
||||
}
|
||||
|
||||
@@ -164,6 +164,10 @@ 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,
|
||||
@@ -193,8 +197,16 @@ 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(
|
||||
@@ -250,6 +262,10 @@ 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(
|
||||
@@ -495,6 +511,10 @@ 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),
|
||||
@@ -615,6 +635,10 @@ 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
|
||||
}
|
||||
@@ -1006,6 +1030,10 @@ 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,
|
||||
@@ -1529,6 +1557,10 @@ 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,
|
||||
@@ -1541,6 +1573,10 @@ 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,
|
||||
|
||||
@@ -1270,6 +1270,10 @@ 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,
|
||||
@@ -1690,6 +1694,10 @@ 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,
|
||||
@@ -3006,6 +3014,10 @@ where
|
||||
table_compaction_planning_report(&self.backend, table_bucket, &namespace, &table, &entry, ¤t_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,
|
||||
@@ -3559,6 +3571,10 @@ 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,
|
||||
@@ -3573,6 +3589,10 @@ 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,
|
||||
@@ -3754,6 +3774,10 @@ 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,
|
||||
|
||||
Reference in New Issue
Block a user