Compare commits

...

4 Commits

Author SHA1 Message Date
overtrue a206895fad Merge remote-tracking branch 'origin/main' into overtrue/docs-1923-free-version-disposition 2026-08-22 21:24:49 +08:00
cxymds da90d02c15 test(ecstore): cover suspended-owner heal semantics (#6348)
* test(ecstore): cover suspended-owner heal semantics

* test(heal): cover suspended owner production path
2026-08-22 20:45:03 +08:00
唐小鸭 a930152d5a fix(admin): expose per-target disableProxy through remote target admin API (#6376)
The read-proxy selector already honors a target's disable_proxy flag
(PR #6172), but the admin API still rejected the field, so the only way
to set it was importing a MinIO-written bucket-targets.json.

- move disableProxy from REMOTE_TARGET_UNSUPPORTED_FIELDS to
  REMOTE_TARGET_WRITABLE_FIELDS (set-remote-target create accepts it)
- add TargetUpdateOp::Proxy so set-remote-target?update=true&proxy=true
  overlays only the proxy group (MinIO TargetUpdateType parity)
- bump REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION 1 -> 2 and update the
  runtime capability pin tests
- keep edge/edgeSyncBeforeExpiry rejected (no implementation behind them)
- pin that a published TargetClient carries disable_proxy, the field the
  proxy-target selector consults

Refs rustfs/backlog#1950
2026-08-22 20:43:52 +08:00
overtrue 3cee88f313 feat(ecstore): account for tier free versions in decommission sweep
Tier free versions (xl.meta cleanup records for deleted transitioned
versions) are not migrated as free versions during decommission: the
exact inventory keeps them inline in versions and the migration loop
routes them through the generic delete-marker path, dropping the flag
and remote-tier identity. Reference audit across GET, heal, ILM,
transition, replication, and restore found no cluster-local consumer
that resolves a free version after decommission; on user-facing delete
paths the remote-delete obligation is also carried by a committed
tier-journal entry, leaving only journal-less records (transition
state unknown) exposed to remote orphaning.

- count and log skipped free versions per decommission entry with
  disposition reason tier_free_version_not_migrated instead of
  omitting them silently
- document free-version lifecycle, non-migration invariant, allowed
  physical-delete timing, and the reference-audit result in
  docs/architecture/decommission-compatibility.md
- state the invariant in doc comments at the filemeta free-version
  sites
- guard the accounting with
  decommission_free_version_accounting_reports_skipped_records

Closes rustfs/backlog#1923
2026-08-22 18:37:54 +08:00
9 changed files with 561 additions and 20 deletions
@@ -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());
+60 -5
View File
@@ -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();
+276 -2
View File
@@ -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);
+16
View File
@@ -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";
+9
View File
@@ -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));
+7 -2
View File
@@ -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:
+49 -7
View File
@@ -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 {
+21 -4
View File
@@ -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"]