mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-22 20:36:38 +00:00
Compare commits
4 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| a206895fad | |||
| da90d02c15 | |||
| a930152d5a | |||
| 3cee88f313 |
@@ -3425,6 +3425,44 @@ mod tests {
|
||||
assert!(mutexes.contains_key("second"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn update_all_targets_publishes_disable_proxy_on_target_client() {
|
||||
// The read-proxy selector (replication_proxy::get_proxy_targets) skips
|
||||
// targets whose TargetClient carries disable_proxy — the persisted
|
||||
// per-target opt-out must survive client publication.
|
||||
let sys = BucketTargetSys::default();
|
||||
let target = |arn: &str, disable_proxy: bool| BucketTarget {
|
||||
arn: arn.to_string(),
|
||||
endpoint: "192.168.1.10:9000".to_string(),
|
||||
target_bucket: "target-bucket".to_string(),
|
||||
region: "us-east-1".to_string(),
|
||||
disable_proxy,
|
||||
credentials: Some(Credentials {
|
||||
access_key: "access".to_string(),
|
||||
secret_key: "secret".to_string(),
|
||||
session_token: None,
|
||||
expiration: None,
|
||||
}),
|
||||
..Default::default()
|
||||
};
|
||||
let targets = BucketTargets {
|
||||
targets: vec![target("arn:proxied", false), target("arn:opted-out", true)],
|
||||
};
|
||||
|
||||
sys.update_all_targets("bucket", Some(&targets)).await;
|
||||
|
||||
let proxied = sys
|
||||
.get_remote_target_client("bucket", "arn:proxied")
|
||||
.await
|
||||
.expect("client should be published");
|
||||
assert!(!proxied.disable_proxy);
|
||||
let opted_out = sys
|
||||
.get_remote_target_client("bucket", "arn:opted-out")
|
||||
.await
|
||||
.expect("client should be published");
|
||||
assert!(opted_out.disable_proxy, "disable_proxy must reach the published TargetClient");
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
||||
async fn target_updates_serialize_client_build_through_publication_per_bucket() {
|
||||
let sys = Arc::new(BucketTargetSys::default());
|
||||
|
||||
@@ -1130,6 +1130,23 @@ fn should_cleanup_decommission_source_entry(decommissioned: usize, total_version
|
||||
decommissioned.saturating_add(expired) == total_versions
|
||||
}
|
||||
|
||||
/// Disposition reason logged for tier free-version records that decommission
|
||||
/// skips instead of migrating.
|
||||
const DECOMMISSION_FREE_VERSION_SKIP_REASON: &str = "tier_free_version_not_migrated";
|
||||
|
||||
/// Counts the tier free-version records present in a decommission entry
|
||||
/// inventory. The exact loader (`load_file_info_versions_exact`) keeps these
|
||||
/// records inline in `versions` instead of separating them into
|
||||
/// `free_versions`, and the migration loop then routes them through the
|
||||
/// generic delete-marker path: the free-version flag and its remote-tier
|
||||
/// identity are never carried to the target pool, and a lone record is skipped
|
||||
/// by the empty-delete-marker rule. Accounting for them here keeps the final
|
||||
/// sweep from silently omitting records whose free-version disposition was
|
||||
/// dropped (see docs/architecture/decommission-compatibility.md).
|
||||
fn decommission_free_versions_skipped(fivs: &FileInfoVersions) -> usize {
|
||||
fivs.versions.iter().filter(|version| version.tier_free_version()).count()
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
#[allow(
|
||||
dead_code,
|
||||
@@ -3097,6 +3114,22 @@ impl ECStore {
|
||||
fivs.versions
|
||||
.sort_by_key(|v| (v.mod_time.is_none(), std::cmp::Reverse(v.mod_time)));
|
||||
|
||||
let skipped_free_versions = decommission_free_versions_skipped(&fivs);
|
||||
if skipped_free_versions > 0 {
|
||||
debug!(
|
||||
event = EVENT_DECOMMISSION_ENTRY,
|
||||
component = LOG_COMPONENT_ECSTORE,
|
||||
subsystem = LOG_SUBSYSTEM_POOLS,
|
||||
pool_index = idx,
|
||||
bucket = %bucket,
|
||||
object = %entry.name,
|
||||
skipped_free_versions,
|
||||
reason = DECOMMISSION_FREE_VERSION_SKIP_REASON,
|
||||
state = "free_versions_skipped",
|
||||
"Decommission skipped free-version migration"
|
||||
);
|
||||
}
|
||||
|
||||
let mut decommissioned: usize = 0;
|
||||
let mut expired: usize = 0;
|
||||
let mut cleanup_preflight_allowed_missing = Vec::new();
|
||||
@@ -5458,11 +5491,12 @@ pub(crate) fn fallback_free_capacity_dedup(disks: &[rustfs_madmin::Disk]) -> usi
|
||||
#[cfg(test)]
|
||||
mod pools_tests {
|
||||
use super::{
|
||||
DECOMMISSION_PROGRESS_SAVE_INTERVAL, DECOMMISSION_PROGRESS_SAVE_ITEM_THRESHOLD, DECOMMISSION_PROGRESS_SAVE_RETRY_BACKOFF,
|
||||
DecomBucketInfo, DecommissionStartPoolState, DecommissionTerminalState, ListCallback, PoolDecommissionInfo, PoolMeta,
|
||||
PoolSpaceInfo, PoolStatus, apply_decommission_status_space_info, bind_decommission_cancelers,
|
||||
bind_missing_decommission_cancelers, cancel_decommission_canceler, classify_decommission_terminal_state,
|
||||
count_decommission_item, decommission_cancel_signal_result, decommission_item_size, decommission_meta_bucket_options,
|
||||
DECOMMISSION_FREE_VERSION_SKIP_REASON, DECOMMISSION_PROGRESS_SAVE_INTERVAL, DECOMMISSION_PROGRESS_SAVE_ITEM_THRESHOLD,
|
||||
DECOMMISSION_PROGRESS_SAVE_RETRY_BACKOFF, DecomBucketInfo, DecommissionStartPoolState, DecommissionTerminalState,
|
||||
ListCallback, PoolDecommissionInfo, PoolMeta, PoolSpaceInfo, PoolStatus, apply_decommission_status_space_info,
|
||||
bind_decommission_cancelers, bind_missing_decommission_cancelers, cancel_decommission_canceler,
|
||||
classify_decommission_terminal_state, count_decommission_item, decommission_cancel_signal_result,
|
||||
decommission_free_versions_skipped, decommission_item_size, decommission_meta_bucket_options,
|
||||
decommission_start_pool_state, dedup_indices, default_decommission_bucket_concurrency,
|
||||
ensure_decommission_cancel_allowed, ensure_decommission_clear_allowed, ensure_decommission_listing_disks_available,
|
||||
ensure_decommission_not_rebalancing, ensure_decommission_start_allowed, ensure_decommission_start_keeps_active_pool,
|
||||
@@ -6762,6 +6796,27 @@ mod pools_tests {
|
||||
assert!(!should_cleanup_decommission_source_entry(2, 2, 1));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn decommission_free_version_accounting_reports_skipped_records() {
|
||||
let mut fivs = FileInfoVersions::default();
|
||||
assert_eq!(decommission_free_versions_skipped(&fivs), 0);
|
||||
|
||||
fivs.versions.push(FileInfo {
|
||||
name: "object.txt".to_string(),
|
||||
..Default::default()
|
||||
});
|
||||
let mut free_one = FileInfo::default();
|
||||
free_one.set_tier_free_version();
|
||||
fivs.versions.push(free_one);
|
||||
let mut free_two = FileInfo::default();
|
||||
free_two.set_tier_free_version();
|
||||
free_two.transition_tier = "WARM".to_string();
|
||||
fivs.versions.push(free_two);
|
||||
|
||||
assert_eq!(decommission_free_versions_skipped(&fivs), 2);
|
||||
assert_eq!(DECOMMISSION_FREE_VERSION_SKIP_REASON, "tier_free_version_not_migrated");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_pool_meta_update_after_rejects_out_of_range_index() {
|
||||
let mut meta = PoolMeta::default();
|
||||
|
||||
@@ -297,10 +297,16 @@ impl ECStore {
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::bucket::metadata_sys;
|
||||
use crate::core::pools::{PoolDecommissionInfo, PoolStatus};
|
||||
use crate::disk::{DiskOption, format::FormatV3, new_disk};
|
||||
use crate::layout::endpoints::{Endpoints, PoolEndpoints};
|
||||
use crate::disk::{DeleteOptions, DiskOption, format::FormatV3, new_disk};
|
||||
use crate::layout::endpoints::{EndpointServerPools, Endpoints, PoolEndpoints};
|
||||
use crate::runtime::instance::InstanceContext;
|
||||
use crate::storage_api_contracts::bucket::{BucketOperations, MakeBucketOptions};
|
||||
use crate::storage_api_contracts::object::{ObjectIO as _, ObjectOperations};
|
||||
use crate::store::init_format::{load_format_erasure, save_format_file};
|
||||
use crate::store::init_local_disks_with_instance_ctx;
|
||||
use tokio_util::sync::CancellationToken;
|
||||
|
||||
async fn minimal_heal_pool(pool_idx: usize) -> Arc<Sets> {
|
||||
let format = FormatV3::new(1, 1);
|
||||
@@ -347,6 +353,51 @@ mod tests {
|
||||
}
|
||||
}
|
||||
|
||||
async fn multi_pool_heal_store() -> (tempfile::TempDir, Arc<ECStore>, CancellationToken) {
|
||||
let temp_dir = tempfile::tempdir().expect("multi-pool heal test directory should be created");
|
||||
let mut pool_endpoints = Vec::new();
|
||||
for pool_index in 0..2 {
|
||||
let mut endpoints = Vec::new();
|
||||
for disk_index in 0..4 {
|
||||
let disk_path = temp_dir.path().join(format!("pool{pool_index}-disk{disk_index}"));
|
||||
tokio::fs::create_dir_all(&disk_path)
|
||||
.await
|
||||
.expect("multi-pool heal test disk should be created");
|
||||
let mut endpoint = Endpoint::try_from(disk_path.to_str().expect("disk path should be utf8"))
|
||||
.expect("test endpoint should parse");
|
||||
endpoint.set_pool_index(pool_index);
|
||||
endpoint.set_set_index(0);
|
||||
endpoint.set_disk_index(disk_index);
|
||||
endpoints.push(endpoint);
|
||||
}
|
||||
pool_endpoints.push(PoolEndpoints {
|
||||
legacy: false,
|
||||
set_count: 1,
|
||||
drives_per_set: 4,
|
||||
endpoints: Endpoints::from(endpoints),
|
||||
cmd_line: format!("heal-owner-pool-{pool_index}"),
|
||||
platform: "test".to_string(),
|
||||
});
|
||||
}
|
||||
|
||||
let endpoint_pools = EndpointServerPools::from(pool_endpoints);
|
||||
let instance_ctx = Arc::new(InstanceContext::new());
|
||||
init_local_disks_with_instance_ctx(&instance_ctx, endpoint_pools.clone())
|
||||
.await
|
||||
.expect("multi-pool local disks should initialize");
|
||||
let shutdown = CancellationToken::new();
|
||||
let store = ECStore::new_with_instance_ctx(
|
||||
"127.0.0.1:0".parse().expect("test address should parse"),
|
||||
endpoint_pools,
|
||||
shutdown.clone(),
|
||||
instance_ctx,
|
||||
)
|
||||
.await
|
||||
.expect("multi-pool test store should initialize");
|
||||
metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
|
||||
(temp_dir, store, shutdown)
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn heal_object_pool_scope_selects_only_requested_pool() {
|
||||
let store = minimal_heal_store().await;
|
||||
@@ -506,6 +557,229 @@ mod tests {
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial_test::serial]
|
||||
async fn unscoped_heal_object_suspended_owner_semantics() {
|
||||
let (_temp_dir, store, shutdown) = multi_pool_heal_store().await;
|
||||
let bucket = format!("heal-owner-{}", Uuid::new_v4().simple());
|
||||
let active_object = "active-owner";
|
||||
let suspended_only_object = "suspended-only";
|
||||
let duplicate_object = "duplicate-owner";
|
||||
let marker_object = "marker-owner";
|
||||
let quorum_object = "quorum-owner";
|
||||
store
|
||||
.make_bucket(&bucket, &MakeBucketOptions::default())
|
||||
.await
|
||||
.expect("bucket should be created in all pools");
|
||||
|
||||
let mut active_reader = PutObjReader::from_vec(b"active owner".to_vec());
|
||||
store.pools[0]
|
||||
.put_object(&bucket, active_object, &mut active_reader, &ObjectOptions::default())
|
||||
.await
|
||||
.expect("active owner object should be written");
|
||||
let active_disks = store.pools[0].disk_set[0].disks.read().await.clone();
|
||||
let missing_active_disk = active_disks[0].clone().expect("active disk should be online");
|
||||
missing_active_disk
|
||||
.delete(
|
||||
&bucket,
|
||||
active_object,
|
||||
DeleteOptions {
|
||||
recursive: true,
|
||||
immediate: true,
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect("active owner shard should be removed for repair");
|
||||
assert!(
|
||||
missing_active_disk.read_xl(&bucket, active_object, false).await.is_err(),
|
||||
"the active owner fixture must start with one missing metadata copy"
|
||||
);
|
||||
|
||||
let mut suspended_reader = PutObjReader::from_vec(b"suspended owner".to_vec());
|
||||
store.pools[1]
|
||||
.put_object(&bucket, suspended_only_object, &mut suspended_reader, &ObjectOptions::default())
|
||||
.await
|
||||
.expect("suspended owner object should be written");
|
||||
for (pool_index, mod_time) in [1_i64, 2_i64].into_iter().enumerate() {
|
||||
let mut duplicate_reader = PutObjReader::from_vec(format!("duplicate-pool-{pool_index}").into_bytes());
|
||||
store.pools[pool_index]
|
||||
.put_object(
|
||||
&bucket,
|
||||
duplicate_object,
|
||||
&mut duplicate_reader,
|
||||
&ObjectOptions {
|
||||
mod_time: Some(OffsetDateTime::UNIX_EPOCH + time::Duration::seconds(mod_time)),
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect("duplicate owner object should be written");
|
||||
}
|
||||
let duplicate_missing_disk = store.pools[0].disk_set[0].disks.read().await[0]
|
||||
.clone()
|
||||
.expect("duplicate active owner disk should be online");
|
||||
duplicate_missing_disk
|
||||
.delete(
|
||||
&bucket,
|
||||
duplicate_object,
|
||||
DeleteOptions {
|
||||
recursive: true,
|
||||
immediate: true,
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect("duplicate active owner shard should be removed for repair");
|
||||
let history_version = Uuid::new_v4();
|
||||
let mut history_reader = PutObjReader::from_vec(b"marker history".to_vec());
|
||||
store.pools[0]
|
||||
.put_object(
|
||||
&bucket,
|
||||
marker_object,
|
||||
&mut history_reader,
|
||||
&ObjectOptions {
|
||||
versioned: true,
|
||||
version_id: Some(history_version.to_string()),
|
||||
mod_time: Some(OffsetDateTime::UNIX_EPOCH + time::Duration::seconds(1)),
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect("versioned marker history should be written");
|
||||
store.pools[0]
|
||||
.delete_object(
|
||||
&bucket,
|
||||
marker_object,
|
||||
ObjectOptions {
|
||||
versioned: true,
|
||||
mod_time: Some(OffsetDateTime::UNIX_EPOCH + time::Duration::seconds(2)),
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect("delete marker should be written");
|
||||
let mut quorum_reader = PutObjReader::from_vec(b"quorum boundary".to_vec());
|
||||
store.pools[0]
|
||||
.put_object(&bucket, quorum_object, &mut quorum_reader, &ObjectOptions::default())
|
||||
.await
|
||||
.expect("quorum boundary object should be written");
|
||||
{
|
||||
let mut pool_meta = store.pool_meta.write().await;
|
||||
let mut next = PoolMeta::new(&store.pools, &pool_meta);
|
||||
next.pools[1].decommission = Some(PoolDecommissionInfo {
|
||||
start_time: Some(OffsetDateTime::UNIX_EPOCH),
|
||||
..Default::default()
|
||||
});
|
||||
*pool_meta = next;
|
||||
}
|
||||
|
||||
let (_, duplicate_owner) = store
|
||||
.get_latest_object_info_with_idx(&bucket, duplicate_object, &ObjectOptions::default())
|
||||
.await
|
||||
.expect("duplicate owner should resolve");
|
||||
assert_eq!(duplicate_owner, 1, "latest duplicate must win when all pools are eligible");
|
||||
let (_, active_duplicate_owner) = store
|
||||
.get_latest_object_info_with_idx(
|
||||
&bucket,
|
||||
duplicate_object,
|
||||
&ObjectOptions {
|
||||
skip_decommissioned: true,
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect("active duplicate owner should resolve");
|
||||
assert_eq!(
|
||||
active_duplicate_owner, 0,
|
||||
"suspended duplicate must be excluded from active owner selection"
|
||||
);
|
||||
let (duplicate_result, duplicate_err) = store
|
||||
.handle_heal_object(&bucket, duplicate_object, "", &HealOpts::default())
|
||||
.await
|
||||
.expect("duplicate owner heal should complete through the production path");
|
||||
assert_eq!(duplicate_result.object, duplicate_object);
|
||||
assert!(duplicate_err.is_none(), "active duplicate should be repaired: {duplicate_err:?}");
|
||||
assert!(
|
||||
duplicate_missing_disk.read_xl(&bucket, duplicate_object, false).await.is_ok(),
|
||||
"production heal must repair the active duplicate owner rather than the suspended owner"
|
||||
);
|
||||
let (marker_info, marker_owner) = store
|
||||
.get_latest_object_info_with_idx(
|
||||
&bucket,
|
||||
marker_object,
|
||||
&ObjectOptions {
|
||||
skip_decommissioned: true,
|
||||
versioned: true,
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect("latest delete marker should resolve");
|
||||
assert_eq!(marker_owner, 0);
|
||||
assert!(marker_info.delete_marker, "latest version must preserve delete-marker semantics");
|
||||
|
||||
let (active_result, active_err) = store
|
||||
.handle_heal_object(&bucket, active_object, "", &HealOpts::default())
|
||||
.await
|
||||
.expect("unscoped active-owner heal should complete");
|
||||
assert_eq!(active_result.object, active_object);
|
||||
assert!(active_err.is_none(), "active owner must be selected even with a suspended pool");
|
||||
assert!(
|
||||
missing_active_disk.read_xl(&bucket, active_object, false).await.is_ok(),
|
||||
"active owner heal must write the missing disk metadata: result={active_result:?}, err={active_err:?}"
|
||||
);
|
||||
assert!(
|
||||
store.pools[1]
|
||||
.get_object_info(&bucket, active_object, &ObjectOptions::default())
|
||||
.await
|
||||
.is_err(),
|
||||
"the suspended pool must not be written for an active-owner object"
|
||||
);
|
||||
|
||||
let (suspended_result, suspended_err) = store
|
||||
.handle_heal_object(&bucket, suspended_only_object, "", &HealOpts::default())
|
||||
.await
|
||||
.expect("unscoped suspended-only heal should return a terminal result");
|
||||
assert!(suspended_result.object.is_empty());
|
||||
assert!(matches!(suspended_err, Some(Error::FileNotFound)));
|
||||
assert!(
|
||||
store.pools[1]
|
||||
.get_object_info(&bucket, suspended_only_object, &ObjectOptions::default())
|
||||
.await
|
||||
.is_ok(),
|
||||
"suspended-only data must remain untouched when unscoped heal reports absent"
|
||||
);
|
||||
|
||||
let (_, explicit_err) = store
|
||||
.handle_heal_object(
|
||||
&bucket,
|
||||
suspended_only_object,
|
||||
"",
|
||||
&HealOpts {
|
||||
pool: Some(1),
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect("explicit suspended-owner heal should return a mapped error");
|
||||
assert!(matches!(explicit_err, Some(Error::SlowDown)));
|
||||
|
||||
let original_quorum_disks = store.pools[0].disk_set[0].disks.read().await.clone();
|
||||
let surviving_quorum_disk = original_quorum_disks[3].clone();
|
||||
*store.pools[0].disk_set[0].disks.write().await = vec![None, None, None, surviving_quorum_disk];
|
||||
let (_, quorum_err) = store
|
||||
.handle_heal_object(&bucket, quorum_object, "", &HealOpts::default())
|
||||
.await
|
||||
.expect("quorum boundary heal should return a mapped result");
|
||||
*store.pools[0].disk_set[0].disks.write().await = original_quorum_disks;
|
||||
assert!(
|
||||
matches!(quorum_err, Some(Error::ErasureReadQuorum)),
|
||||
"quorum-boundary heal must preserve quorum error, got {quorum_err:?}"
|
||||
);
|
||||
shutdown.cancel();
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn handle_heal_format_continues_after_a_pool_error() {
|
||||
let canonical_format = FormatV3::new(1, 3);
|
||||
|
||||
@@ -90,6 +90,22 @@ fn legacy_data_key_for_version(version_id: Option<Uuid>) -> Option<String> {
|
||||
pub const TRANSITION_COMPLETE: &str = "complete";
|
||||
pub const TRANSITION_PENDING: &str = "pending";
|
||||
|
||||
/// xl.meta key marking a tier free-version record.
|
||||
///
|
||||
/// A free version is a delete-marker-shaped cleanup hint appended by
|
||||
/// [`MetaObject::delete_version`] when a version whose remote transition
|
||||
/// completed is removed from xl.meta; it carries the remote tier identity for
|
||||
/// an idempotent remote delete and is never a user-visible version
|
||||
/// (`num_versions` excludes it). While the record exists it is consumed by the
|
||||
/// lifecycle free-version recovery scan and the usage scanner, which re-enqueue
|
||||
/// the pending remote delete, and by heal metadata walks. On S3 and lifecycle
|
||||
/// delete paths the same obligation is also carried by a committed tier-journal
|
||||
/// entry; deletes without such an entry (for example a removed version whose
|
||||
/// transition state decodes as unknown) rely on this record alone until the
|
||||
/// worker removes it after a successful remote delete. Decommission does not
|
||||
/// preserve these semantics: its exact inventory keeps the records inline in
|
||||
/// `versions` and the migration loop treats them as ordinary delete markers —
|
||||
/// see docs/architecture/decommission-compatibility.md.
|
||||
pub const FREE_VERSION: &str = "free-version";
|
||||
|
||||
pub const TRANSITION_STATUS: &str = "transition-status";
|
||||
|
||||
@@ -2725,6 +2725,15 @@ impl MetaObject {
|
||||
self.meta_sys.retain(|k, _| !k.starts_with("X-Amz-Restore"));
|
||||
}
|
||||
|
||||
/// Builds the free-version cleanup record appended when a transitioned
|
||||
/// version is removed from xl.meta. The record keeps the remote tier
|
||||
/// identity so the lifecycle worker can issue the idempotent remote delete
|
||||
/// and only then remove the record; until then the recovery scan and the
|
||||
/// usage scanner keep re-enqueueing it. S3 and lifecycle deletes also
|
||||
/// persist a committed tier-journal entry for the same remote delete, so a
|
||||
/// record destroyed without its remote delete (as decommission does when it
|
||||
/// treats these records as ordinary delete markers) strands only the
|
||||
/// journal-less cases — see docs/architecture/decommission-compatibility.md.
|
||||
pub fn init_free_version(&self, fi: &FileInfo) -> Result<(FileMetaVersion, bool)> {
|
||||
if fi.skip_tier_free_version() {
|
||||
return Ok((FileMetaVersion::default(), false));
|
||||
|
||||
@@ -60,7 +60,9 @@ pub const REPLICATION_READ_ONLY_HISTORICAL_FIELDS: &[&str] = &[
|
||||
"Destination.ReplicationTime",
|
||||
];
|
||||
|
||||
pub const REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION: u32 = 1;
|
||||
// v2: disableProxy moved from unsupported to writable (per-target read-proxy
|
||||
// opt-out is accepted by set-remote-target and the `proxy` update op).
|
||||
pub const REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION: u32 = 2;
|
||||
|
||||
pub const REMOTE_TARGET_WRITABLE_FIELDS: &[&str] = &[
|
||||
"sourcebucket",
|
||||
@@ -83,9 +85,12 @@ pub const REMOTE_TARGET_WRITABLE_FIELDS: &[&str] = &[
|
||||
// madmin default of 60s); the per-target health-check interval is not
|
||||
// yet applied — the heartbeat keeps its global env-configured interval.
|
||||
"healthCheckDuration",
|
||||
// Per-target read-proxy opt-out, consumed by the proxy-target selector
|
||||
// (contract v2; previously only importable via MinIO bucket-targets.json).
|
||||
"disableProxy",
|
||||
];
|
||||
|
||||
pub const REMOTE_TARGET_UNSUPPORTED_FIELDS: &[&str] = &["disableProxy", "edge", "edgeSyncBeforeExpiry"];
|
||||
pub const REMOTE_TARGET_UNSUPPORTED_FIELDS: &[&str] = &["edge", "edgeSyncBeforeExpiry"];
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
|
||||
pub struct ObjectOpts {
|
||||
|
||||
@@ -153,6 +153,91 @@ No migration step is required for these decisions because this note documents th
|
||||
current RustFS behavior. Changing either decision later requires an operator
|
||||
compatibility note and updated characterization tests.
|
||||
|
||||
## Tier Free Versions During Decommission
|
||||
|
||||
A tier free version is an internal xl.meta record (`rustfs_filemeta::FREE_VERSION`,
|
||||
flagged `XL_FLAG_FREE_VERSION`) shaped like a delete marker. It is created by
|
||||
`MetaObject::init_free_version` when a version whose remote transition completed is
|
||||
deleted locally: the visible version is removed and the record keeps the remote-tier
|
||||
identity (tier, object name, version id, state, destination id) needed for an
|
||||
idempotent remote delete. Free versions are not user-visible versions; `num_versions`
|
||||
and all listing/GET paths exclude them.
|
||||
|
||||
### Lifecycle And Consumers
|
||||
|
||||
Creation: any local delete that removes a version whose transition status is
|
||||
`complete` appends the record via `MetaObject::delete_version` →
|
||||
`init_free_version` (skipped only when `skip_tier_free_version` is set, as on
|
||||
data-movement copies). The same deletes also persist a durable tier-journal
|
||||
entry on every user-facing path: S3 single deletes (`execute_delete_object` →
|
||||
`delete_object_with_tier_delete_journal`), S3 batch deletes, lifecycle expiry,
|
||||
and lifecycle delete-all all prepare and commit a journal entry around the
|
||||
delete. A journal entry is omitted when the removed version's transition state
|
||||
decodes as `TransitionVersionState::Unknown`, or on internal journal-less
|
||||
delete paths that never touch transitioned user objects.
|
||||
|
||||
Consumption while the record exists: the background recovery loop started by
|
||||
`init_background_expiry` (spawned by `spawn_tier_free_version_recovery_once`,
|
||||
enabled by default) scans disks for pending records and re-enqueues them; the
|
||||
usage scanner does the same; the lifecycle worker then deletes the remote tier
|
||||
object idempotently and only afterwards removes the local record. Heal walks
|
||||
include free-version records in metadata healing. Transition planning,
|
||||
replication, restore, GET, listings, and usage aggregation never depend on
|
||||
them.
|
||||
|
||||
### Decommission Handling
|
||||
|
||||
The exact decommission inventory loader (`load_file_info_versions_exact` via
|
||||
`get_all_file_info_versions`) keeps free-version records inline in `versions`; it
|
||||
never populates `free_versions`, so the source-cleanup preflight comparison of
|
||||
`free_versions` is vacuous for decommission. The migration loop then routes every
|
||||
record through the generic delete-marker handling:
|
||||
|
||||
- a record that is the only remaining version without replication is skipped by the
|
||||
empty-delete-marker rule and counted as done;
|
||||
- any other record is copied to the target pool as an ordinary delete marker with the
|
||||
same version id and mod time.
|
||||
|
||||
In both cases the free-version flag and its remote-tier identity are dropped:
|
||||
decommission neither preserves free-version semantics nor performs or reschedules the
|
||||
pending remote-tier delete. Source cleanup then removes the original records together
|
||||
with the source xl.meta.
|
||||
|
||||
Allowed physical-delete timing: the source record may be removed once the migration
|
||||
loop has dispositioned it (copied as a plain marker or skipped as lone), which
|
||||
happens regardless of whether its remote-tier delete was ever performed.
|
||||
|
||||
### Reference-Audit Result
|
||||
|
||||
No cluster-local consumer resolves a free version after decommission finishes: GET,
|
||||
listing, transition planning, replication, restore, and heal operate either on
|
||||
user-visible versions or while the record still exists. The remote exposure is
|
||||
bounded:
|
||||
|
||||
- On every user-facing delete path the remote-delete obligation is durably carried
|
||||
by the committed tier-journal entry, which the tier sweeper processes
|
||||
independently of xl.meta; the free-version record is an idempotent second
|
||||
pointer, not the only one. Dropping it during decommission therefore does not
|
||||
orphan the remote object.
|
||||
- Residual exposure: for records whose version state decoded as `Unknown` no
|
||||
journal entry exists, so dropping the unconsumed record loses that cleanup hint
|
||||
and the remote-tier object is orphaned. The same applies to any future internal
|
||||
delete path that removes transitioned versions without a journal entry.
|
||||
|
||||
Copying a pending record as an ordinary delete marker also adds a user-visible
|
||||
tombstone to the target pool's version history that the source never exposed.
|
||||
|
||||
Because of the residual journal-less case, decommission must account for every
|
||||
free-version record instead of omitting it silently:
|
||||
|
||||
- `decommission_free_versions_skipped` counts the records per decommission entry;
|
||||
- entries with a non-zero count log `state = "free_versions_skipped"` with reason
|
||||
`tier_free_version_not_migrated`.
|
||||
|
||||
Regression guard:
|
||||
|
||||
- `decommission_free_version_accounting_reports_skipped_records`
|
||||
|
||||
## Regression Guard
|
||||
|
||||
The queued multi-pool contract is guarded by:
|
||||
|
||||
@@ -73,6 +73,8 @@ enum TargetUpdateOp {
|
||||
/// Connection group: credentials plus endpoint, target bucket, and TLS settings.
|
||||
Credentials,
|
||||
Sync,
|
||||
/// Per-target read-proxy opt-out (`disableProxy`).
|
||||
Proxy,
|
||||
Bandwidth,
|
||||
Path,
|
||||
}
|
||||
@@ -81,12 +83,13 @@ fn parse_remote_target_update_ops(queries: &HashMap<String, String>) -> S3Result
|
||||
const SUPPORTED_OPS: &[(&str, TargetUpdateOp)] = &[
|
||||
("creds", TargetUpdateOp::Credentials),
|
||||
("sync", TargetUpdateOp::Sync),
|
||||
("proxy", TargetUpdateOp::Proxy),
|
||||
("bandwidth", TargetUpdateOp::Bandwidth),
|
||||
("path", TargetUpdateOp::Path),
|
||||
];
|
||||
// Present in the MinIO wire contract, but they drive target fields this
|
||||
// version rejects as unsupported — fail loudly instead of silently ignoring.
|
||||
const UNSUPPORTED_OPS: &[&str] = &["proxy", "healthcheck", "edge", "edgeSyncBeforeExpiry"];
|
||||
const UNSUPPORTED_OPS: &[&str] = &["healthcheck", "edge", "edgeSyncBeforeExpiry"];
|
||||
|
||||
for key in UNSUPPORTED_OPS {
|
||||
if queries.get(*key).is_some_and(|value| value == "true") {
|
||||
@@ -312,11 +315,10 @@ impl RemoteTargetRequest {
|
||||
));
|
||||
}
|
||||
|
||||
for (unsupported, configured) in
|
||||
REMOTE_TARGET_UNSUPPORTED_FIELDS
|
||||
.iter()
|
||||
.copied()
|
||||
.zip([self.disable_proxy, self.edge, self.edge_sync_before_expiry])
|
||||
for (unsupported, configured) in REMOTE_TARGET_UNSUPPORTED_FIELDS
|
||||
.iter()
|
||||
.copied()
|
||||
.zip([self.edge, self.edge_sync_before_expiry])
|
||||
{
|
||||
if configured {
|
||||
return Err(s3_error!(
|
||||
@@ -702,6 +704,7 @@ impl Operation for SetRemoteTargetHandler {
|
||||
target.deployment_id = remote_target.deployment_id.clone();
|
||||
}
|
||||
TargetUpdateOp::Sync => target.replication_sync = remote_target.replication_sync,
|
||||
TargetUpdateOp::Proxy => target.disable_proxy = remote_target.disable_proxy,
|
||||
TargetUpdateOp::Bandwidth => target.bandwidth_limit = remote_target.bandwidth_limit,
|
||||
TargetUpdateOp::Path => target.path = remote_target.path.clone(),
|
||||
}
|
||||
@@ -1520,6 +1523,7 @@ mod tests {
|
||||
("update", "true"),
|
||||
("creds", "true"),
|
||||
("sync", "true"),
|
||||
("proxy", "true"),
|
||||
("bandwidth", "true"),
|
||||
("path", "true"),
|
||||
]))
|
||||
@@ -1529,6 +1533,7 @@ mod tests {
|
||||
vec![
|
||||
TargetUpdateOp::Credentials,
|
||||
TargetUpdateOp::Sync,
|
||||
TargetUpdateOp::Proxy,
|
||||
TargetUpdateOp::Bandwidth,
|
||||
TargetUpdateOp::Path
|
||||
]
|
||||
@@ -2070,7 +2075,6 @@ mod tests {
|
||||
("credentials.session_token", serde_json::json!("session-token")),
|
||||
("credentials.expiration", serde_json::json!("2026-01-01T00:00:00Z")),
|
||||
("api", serde_json::json!("s3v2")),
|
||||
("disableProxy", serde_json::json!(true)),
|
||||
("edge", serde_json::json!(true)),
|
||||
("edgeSyncBeforeExpiry", serde_json::json!(true)),
|
||||
] {
|
||||
@@ -2300,6 +2304,44 @@ mod tests {
|
||||
assert!(!REMOTE_TARGET_UNSUPPORTED_FIELDS.contains(&"healthCheckDuration"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn remote_target_disable_proxy_is_declared_writable_edge_stays_unsupported() {
|
||||
assert!(REMOTE_TARGET_WRITABLE_FIELDS.contains(&"disableProxy"));
|
||||
assert!(!REMOTE_TARGET_UNSUPPORTED_FIELDS.contains(&"disableProxy"));
|
||||
// edge sync has no implementation behind it — it must stay rejected.
|
||||
assert!(REMOTE_TARGET_UNSUPPORTED_FIELDS.contains(&"edge"));
|
||||
assert!(REMOTE_TARGET_UNSUPPORTED_FIELDS.contains(&"edgeSyncBeforeExpiry"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn remote_target_create_accepts_disable_proxy() {
|
||||
let mut request = valid_remote_target_request();
|
||||
request["disableProxy"] = serde_json::json!(true);
|
||||
|
||||
let target = serde_json::from_value::<RemoteTargetRequest>(request)
|
||||
.expect("request should deserialize")
|
||||
.into_bucket_target()
|
||||
.expect("disableProxy is a supported per-target read-proxy opt-out");
|
||||
|
||||
assert!(target.disable_proxy);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn update_body_with_proxy_op_toggles_disable_proxy_without_credentials() {
|
||||
// Mirrors the other partial-update groups: a proxy-only update body may
|
||||
// omit the connection fields entirely.
|
||||
let body = serde_json::json!({
|
||||
"arn": "arn:rustfs:replication:us-east-1:dep:target",
|
||||
"type": "replication",
|
||||
"disableProxy": true
|
||||
});
|
||||
let request: RemoteTargetRequest = serde_json::from_value(body).expect("partial update body should deserialize");
|
||||
let target = request
|
||||
.into_update_bucket_target(&[TargetUpdateOp::Proxy])
|
||||
.expect("proxy-only update must not require credentials");
|
||||
assert!(target.disable_proxy);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn remote_target_capability_fields_do_not_overlap() {
|
||||
for field in REMOTE_TARGET_UNSUPPORTED_FIELDS {
|
||||
|
||||
@@ -1262,7 +1262,9 @@ mod tests {
|
||||
assert_eq!(response.summary.manual_transition_jobs.state, CapabilityState::Supported);
|
||||
assert_eq!(response.replication.contract_version, 1);
|
||||
assert_eq!(response.replication.bucket_replication.contract_version, 1);
|
||||
assert_eq!(response.replication.remote_targets.contract_version, 1);
|
||||
// v2: disableProxy moved from unsupported to writable (per-target
|
||||
// read-proxy opt-out reached the admin API).
|
||||
assert_eq!(response.replication.remote_targets.contract_version, 2);
|
||||
assert_eq!(response.replication.bucket_replication.status.state, CapabilityState::Supported);
|
||||
assert_eq!(response.replication.remote_targets.status.state, CapabilityState::Supported);
|
||||
assert_eq!(
|
||||
@@ -1293,7 +1295,15 @@ mod tests {
|
||||
.remote_targets
|
||||
.fields
|
||||
.iter()
|
||||
.any(|field| field.name == "disableProxy" && field.state == super::ReplicationFieldState::Unsupported)
|
||||
.any(|field| field.name == "disableProxy" && field.state == super::ReplicationFieldState::Supported)
|
||||
);
|
||||
assert!(
|
||||
response
|
||||
.replication
|
||||
.remote_targets
|
||||
.fields
|
||||
.iter()
|
||||
.any(|field| field.name == "edge" && field.state == super::ReplicationFieldState::Unsupported)
|
||||
);
|
||||
assert!(
|
||||
response
|
||||
@@ -1364,7 +1374,7 @@ mod tests {
|
||||
assert_eq!(value["summary"]["manual_transition_jobs"]["state"], "supported");
|
||||
assert_eq!(value["replication"]["contract_version"], 1);
|
||||
assert_eq!(value["replication"]["bucket_replication"]["contract_version"], 1);
|
||||
assert_eq!(value["replication"]["remote_targets"]["contract_version"], 1);
|
||||
assert_eq!(value["replication"]["remote_targets"]["contract_version"], 2);
|
||||
assert_eq!(value["replication"]["bucket_replication"]["status"]["state"], "supported");
|
||||
assert_eq!(value["replication"]["remote_targets"]["status"]["state"], "supported");
|
||||
assert_eq!(
|
||||
@@ -1383,7 +1393,14 @@ mod tests {
|
||||
.as_array()
|
||||
.expect("remote target fields should be an array")
|
||||
.iter()
|
||||
.any(|field| field["name"] == "disableProxy" && field["state"] == "unsupported")
|
||||
.any(|field| field["name"] == "disableProxy" && field["state"] == "supported")
|
||||
);
|
||||
assert!(
|
||||
value["replication"]["remote_targets"]["fields"]
|
||||
.as_array()
|
||||
.expect("remote target fields should be an array")
|
||||
.iter()
|
||||
.any(|field| field["name"] == "edge" && field["state"] == "unsupported")
|
||||
);
|
||||
assert!(
|
||||
value["replication"]["remote_targets"]["fields"]
|
||||
|
||||
Reference in New Issue
Block a user