gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-09-07 04:25:54 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(tier): add durable probe intent protocol

Browse Source
This commit is contained in:
cxymds
2026-09-04 23:16:21 +08:00
parent 923bde6904
commit 1638e9b516
6 changed files with 1550 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/bucket/lifecycle/durable_namespace.rs
+234 -1
View File
@@ -25,6 +25,7 @@ use super::{
manual_transition_job, tier_delete_journal, transition_transaction,
};
use crate::error::{Error, Result};
use crate::services::tier::tier_probe_intent;
pub(crate) const ILM_META_PREFIX: &str = "ilm";
const ILM_META_OBJECT_PREFIX: &str = "ilm/";
@@ -35,6 +36,7 @@ pub(crate) enum DurableIlmRecordKind {
TierDeleteJournal,
TierDeleteDispatchManifest,
TransitionTransaction,
TierProbeIntent,
ManualTransitionJob,
ManualTransitionScope,
ManualTransitionTask,
@@ -73,6 +75,12 @@ pub(crate) const TRANSITION_TRANSACTION_NAMESPACE: DurableIlmNamespace = Durable
max_record_size: transition_transaction::MAX_TRANSITION_TRANSACTION_SIZE,
kind: DurableIlmRecordKind::TransitionTransaction,
};
pub(crate) const TIER_PROBE_INTENT_NAMESPACE: DurableIlmNamespace = DurableIlmNamespace {
name: "tier-probe-intent",
prefix: tier_probe_intent::TIER_PROBE_INTENT_RECORD_PREFIX,
max_record_size: tier_probe_intent::MAX_TIER_PROBE_INTENT_SIZE,
kind: DurableIlmRecordKind::TierProbeIntent,
};
pub(crate) const MANUAL_TRANSITION_JOB_NAMESPACE: DurableIlmNamespace = DurableIlmNamespace {
name: "manual-transition-job",
prefix: "ilm/manual-transition/jobs",
@@ -98,11 +106,12 @@ pub(crate) const MANUAL_TRANSITION_WORKER_RESULT_NAMESPACE: DurableIlmNamespace
kind: DurableIlmRecordKind::ManualTransitionWorkerResult,
};
pub(crate) const DURABLE_ILM_NAMESPACES: [DurableIlmNamespace; 8] = [
pub(crate) const DURABLE_ILM_NAMESPACES: [DurableIlmNamespace; 9] = [
TIER_DELETE_JOURNAL_NAMESPACE,
TIER_DELETE_JOURNAL_V6_NAMESPACE,
TIER_DELETE_DISPATCH_MANIFEST_NAMESPACE,
TRANSITION_TRANSACTION_NAMESPACE,
TIER_PROBE_INTENT_NAMESPACE,
MANUAL_TRANSITION_JOB_NAMESPACE,
MANUAL_TRANSITION_SCOPE_NAMESPACE,
MANUAL_TRANSITION_TASK_NAMESPACE,
@@ -200,6 +209,15 @@ pub(crate) enum DurableIlmRecordCheckpoint {
revision: u64,
state: transition_transaction::TransitionTransactionState,
},
TierProbeIntent {
content_sha256: String,
identity_sha256: String,
remote_version_sha256: String,
remote_version_known: bool,
owner_fence_sha256: String,
revision: u64,
state: tier_probe_intent::TierProbeIntentState,
},
ManualTransitionJob {
content_sha256: String,
identity_sha256: String,
@@ -232,6 +250,7 @@ impl DurableIlmRecordCheckpoint {
| Self::TierDeleteDispatchManifest { content_sha256, .. }
| Self::TierDeleteDispatchParent { content_sha256, .. }
| Self::TransitionTransaction { content_sha256, .. }
| Self::TierProbeIntent { content_sha256, .. }
| Self::ManualTransitionJob { content_sha256, .. }
| Self::ManualTransitionScope { content_sha256, .. }
| Self::ManualTransitionTask { content_sha256 }
@@ -421,6 +440,32 @@ impl DurableIlmRecordCheckpoint {
.is_some_and(|expected_revision| *next_revision == expected_revision)
&& (!previous_remote_version_known || previous_remote_version == next_remote_version)
}
(
Self::TierProbeIntent {
identity_sha256: previous_identity,
remote_version_sha256: previous_remote_version,
remote_version_known: previous_remote_version_known,
owner_fence_sha256: previous_owner_fence,
revision: previous_revision,
state: previous_state,
..
},
Self::TierProbeIntent {
identity_sha256: next_identity,
remote_version_sha256: next_remote_version,
owner_fence_sha256: next_owner_fence,
revision: next_revision,
state: next_state,
..
},
) => {
previous_identity == next_identity
&& previous_owner_fence == next_owner_fence
&& next_revision
.checked_sub(*previous_revision)
.is_some_and(|distance| distance == 1 && tier_probe_state_reaches(*previous_state, *next_state, distance))
&& (!previous_remote_version_known || previous_remote_version == next_remote_version)
}
(
Self::ManualTransitionJob {
content_sha256: previous_content,
@@ -500,6 +545,14 @@ impl DurableIlmRecordCheckpoint {
/// after the exact terminal ETag and terminal receipt were committed, to
/// purge older object versions exposed by that deletion.
pub(crate) fn is_predecessor_of_terminal(&self, terminal: &Self) -> bool {
if let Self::TierProbeIntent { state, .. } = terminal
&& !matches!(
state,
tier_probe_intent::TierProbeIntentState::AbortedNoRemote | tier_probe_intent::TierProbeIntentState::Completed
)
{
return false;
}
if self == terminal || self.validate_successor(terminal).is_ok() {
return true;
}
@@ -568,6 +621,37 @@ impl DurableIlmRecordCheckpoint {
}
})
}
(
Self::TierProbeIntent {
identity_sha256: previous_identity,
remote_version_sha256: previous_remote_version,
remote_version_known: previous_remote_version_known,
owner_fence_sha256: previous_owner_fence,
revision: previous_revision,
state: previous_state,
..
},
Self::TierProbeIntent {
identity_sha256: terminal_identity,
remote_version_sha256: terminal_remote_version,
owner_fence_sha256: terminal_owner_fence,
revision: terminal_revision,
state: terminal_state,
..
},
) => {
previous_identity == terminal_identity
&& previous_owner_fence == terminal_owner_fence
&& matches!(
terminal_state,
tier_probe_intent::TierProbeIntentState::AbortedNoRemote
| tier_probe_intent::TierProbeIntentState::Completed
)
&& terminal_revision
.checked_sub(*previous_revision)
.is_some_and(|distance| tier_probe_state_reaches(*previous_state, *terminal_state, distance))
&& (!previous_remote_version_known || previous_remote_version == terminal_remote_version)
}
_ => false,
}
}
@@ -606,6 +690,23 @@ fn transition_state_distance(
}
}
fn tier_probe_state_reaches(
from: tier_probe_intent::TierProbeIntentState,
to: tier_probe_intent::TierProbeIntentState,
revision_distance: u64,
) -> bool {
use tier_probe_intent::TierProbeIntentState::{AbortedNoRemote, CleanupPending, Completed, UploadOutcomeUnknown, Uploaded};
match (from, to) {
(UploadOutcomeUnknown, Uploaded | CleanupPending | AbortedNoRemote) => revision_distance == 1,
(UploadOutcomeUnknown, Completed) => matches!(revision_distance, 2 | 3),
(Uploaded, CleanupPending) => revision_distance == 1,
(Uploaded, Completed) => revision_distance == 2,
(CleanupPending, Completed) => revision_distance == 1,
_ => false,
}
}
fn manual_job_state_reaches(
from: manual_transition_job::ManualTransitionJobState,
to: manual_transition_job::ManualTransitionJobState,
@@ -1082,6 +1183,42 @@ pub(crate) fn validate_durable_ilm_record(path: &str, data: &[u8]) -> Result<Val
},
)
}
DurableIlmRecordKind::TierProbeIntent => {
let probe_id = tier_probe_intent::tier_probe_intent_id_from_record_object_name(path)
.map_err(|err| Error::other(err.to_string()))?;
let intent =
tier_probe_intent::TierProbeIntent::decode(probe_id, data).map_err(|err| Error::other(err.to_string()))?;
let canonical =
tier_probe_intent::tier_probe_intent_record_object_name(probe_id).map_err(|err| Error::other(err.to_string()))?;
if canonical != path {
return Err(Error::other("tier probe intent path is not canonical"));
}
let identity_sha256 = checkpoint_hash(&(
intent.probe_id,
&intent.operation,
&intent.tier_name,
intent.destination_id,
&intent.probe_object,
&intent.creator_id,
intent.creator_epoch,
intent.created_at_unix_nanos,
))?;
let remote_version_sha256 = checkpoint_hash(&intent.remote_version)?;
let owner_fence_sha256 = checkpoint_hash(&intent.owner)?;
(
"probe_id",
probe_id.to_string(),
DurableIlmRecordCheckpoint::TierProbeIntent {
content_sha256,
identity_sha256,
remote_version_sha256,
remote_version_known: !intent.remote_version.is_unknown(),
owner_fence_sha256,
revision: intent.revision,
state: intent.state,
},
)
}
DurableIlmRecordKind::ManualTransitionJob => {
let job_id = manual_transition_job::manual_transition_job_id_from_record_object_name(path)
.map_err(|err| Error::other(err.to_string()))?;
@@ -1237,6 +1374,102 @@ mod tests {
}
}
fn tier_probe_intent_fixture() -> tier_probe_intent::TierProbeIntent {
let probe_id = Uuid::parse_str("36e2220e-9ad2-495b-b3bc-c4d2caf70a31").expect("fixture uuid should parse");
tier_probe_intent::TierProbeIntent {
probe_id,
revision: 1,
state: tier_probe_intent::TierProbeIntentState::UploadOutcomeUnknown,
operation: tier_probe_intent::TierProbeOperationIdentity::Verify {
config_etag: "config-etag".to_string(),
backend_identity: [1; 32],
},
tier_name: "COLD-A".to_string(),
destination_id: [1; 32],
probe_object: tier_probe_intent::tier_probe_object_name(probe_id),
creator_id: "node-a".to_string(),
creator_epoch: Uuid::parse_str("76746062-c05a-40b7-9e38-d2722d7e0332").expect("fixture creator epoch should parse"),
created_at_unix_nanos: 1_780_000_000_000_000_000,
owner: tier_probe_intent::TierProbeOwnerFence {
owner_id: "node-a".to_string(),
owner_epoch: Uuid::parse_str("76746062-c05a-40b7-9e38-d2722d7e0332").expect("fixture owner epoch should parse"),
not_after_unix_nanos: 1_780_000_900_000_000_000,
},
remote_version: tier_probe_intent::TierProbeRemoteVersion::default(),
}
}
fn tier_probe_checkpoint(intent: &tier_probe_intent::TierProbeIntent) -> DurableIlmRecordCheckpoint {
let path =
tier_probe_intent::tier_probe_intent_record_object_name(intent.probe_id).expect("tier probe path should build");
let encoded = intent.encode().expect("tier probe intent should encode");
let namespace = classify_durable_ilm_record(&path)
.expect("tier probe namespace should classify")
.expect("tier probe intent should be durable");
assert_eq!(namespace, &TIER_PROBE_INTENT_NAMESPACE);
validate_durable_ilm_record(&path, &encoded)
.expect("tier probe intent should validate")
.checkpoint
}
#[test]
fn tier_probe_intent_checkpoint_tracks_exact_monotonic_generations() {
let initial_intent = tier_probe_intent_fixture();
let initial = tier_probe_checkpoint(&initial_intent);
let mut uploaded_intent = initial_intent.clone();
uploaded_intent
.advance(
tier_probe_intent::TierProbeIntentState::Uploaded,
tier_probe_intent::TierProbeRemoteVersion::versioned("opaque-v1"),
)
.expect("uploaded state should advance");
let uploaded = tier_probe_checkpoint(&uploaded_intent);
initial
.validate_successor(&uploaded)
.expect("durable receipt may adopt the exact uploaded generation");
let mut cleanup_intent = uploaded_intent.clone();
cleanup_intent
.advance(
tier_probe_intent::TierProbeIntentState::CleanupPending,
uploaded_intent.remote_version.clone(),
)
.expect("cleanup state should advance");
let cleanup = tier_probe_checkpoint(&cleanup_intent);
uploaded
.validate_successor(&cleanup)
.expect("durable receipt may adopt the exact cleanup generation");
let mut completed_intent = cleanup_intent.clone();
completed_intent
.advance(tier_probe_intent::TierProbeIntentState::Completed, cleanup_intent.remote_version.clone())
.expect("completed state should advance");
let completed = tier_probe_checkpoint(&completed_intent);
cleanup
.validate_successor(&completed)
.expect("durable receipt may adopt the exact terminal generation");
assert!(
initial.is_predecessor_of_terminal(&completed),
"terminal cleanup must recognize the full acknowledged-PUT path"
);
assert!(
initial.validate_successor(&completed).is_err(),
"ordinary receipt advancement must not skip intermediate generations"
);
assert!(
!initial.is_predecessor_of_terminal(&uploaded),
"a nonterminal generation must not be accepted as terminal proof"
);
let mut rebound = uploaded_intent;
rebound.owner.owner_epoch = Uuid::new_v4();
assert!(
rebound.encode().is_err(),
"dormant v1 must reject owner takeover before producing a checkpoint"
);
}
#[test]
fn tier_delete_dispatch_manifest_namespace_validates_monotonic_branches() {
use tier_delete_journal::TierDeleteDispatchManifestState::{Aborted, Aborting, Completed, DispatchAuthorized, Preparing};
crates/ecstore/src/services/tier/mod.rs
+1
View File
@@ -21,6 +21,7 @@ pub mod tier_gen;
pub mod tier_handlers;
pub(crate) mod tier_mutation_intent;
pub mod tier_mutation_peer;
pub(crate) mod tier_probe_intent;
pub mod warm_backend;
pub mod warm_backend_aliyun;
pub mod warm_backend_azure;
crates/ecstore/src/services/tier/tier_probe_intent.rs
+1130
View File
File diff suppressed because it is too large Load Diff
crates/ecstore/src/store/init.rs
+146
View File
@@ -864,6 +864,11 @@ mod tests {
save_tier_mutation_intent_record, save_tier_mutation_intent_record_if_current,
},
tier_mutation_peer::{TierMutationPeerError, TierMutationPeerState, handle_tier_mutation_peer_request},
tier_probe_intent::{
TierProbeIntent, TierProbeIntentState, TierProbeOperationIdentity, TierProbeOwnerFence, TierProbeRemoteVersion,
delete_tier_probe_intent_record_if_current, load_tier_probe_intent_record,
save_tier_probe_intent_record_if_absent, save_tier_probe_intent_record_if_current,
},
warm_backend::{TransitionCandidateProbe, WarmBackend},
},
set_disk::SetDiskTransitionUploadedCommitBarrier as TransitionUploadedCommitBarrier,
@@ -17196,6 +17201,147 @@ mod tests {
assert!(matches!(err, Error::ConfigNotFound));
}
#[cfg(feature = "test-util")]
#[tokio::test]
#[serial_test::serial(storage_class_env)]
async fn tier_probe_intent_store_enforces_create_cas_and_terminal_delete_preconditions() {
let temp_dir = tempfile::tempdir().expect("create temp store dir");
let (_ctx, store, _shutdown) =
without_storage_class_env(build_isolated_test_store(temp_dir.path(), "tier-probe-intent-cas", &[4])).await;
let probe_id = uuid::Uuid::new_v4();
let creator_epoch = uuid::Uuid::new_v4();
let initial = TierProbeIntent {
probe_id,
revision: 1,
state: TierProbeIntentState::UploadOutcomeUnknown,
operation: TierProbeOperationIdentity::Verify {
config_etag: "config-etag".to_string(),
backend_identity: [1; 32],
},
tier_name: "COLD-A".to_string(),
destination_id: [1; 32],
probe_object: format!("rustfs-tier-probe-{probe_id}"),
creator_id: "node-a".to_string(),
creator_epoch,
created_at_unix_nanos: 1_780_000_000_000_000_000,
owner: TierProbeOwnerFence {
owner_id: "node-a".to_string(),
owner_epoch: creator_epoch,
not_after_unix_nanos: 1_780_000_900_000_000_000,
},
remote_version: TierProbeRemoteVersion::default(),
};
save_tier_probe_intent_record_if_absent(store.clone(), &initial)
.await
.expect("initial probe intent should persist with create-only semantics");
let duplicate = save_tier_probe_intent_record_if_absent(store.clone(), &initial)
.await
.expect_err("duplicate create must fail closed");
assert!(matches!(duplicate, Error::PreconditionFailed));
let observed_initial = load_tier_probe_intent_record(store.clone(), probe_id)
.await
.expect("initial probe intent should load with an ETag");
assert_eq!(observed_initial.intent(), &initial);
let nonterminal_delete = delete_tier_probe_intent_record_if_current(store.clone(), &observed_initial)
.await
.expect_err("nonterminal evidence must not be deleted");
assert!(nonterminal_delete.to_string().contains("must be terminal"));
let mut fabricated_current_intent = initial.clone();
fabricated_current_intent.tier_name = "COLD-B".to_string();
let mut fabricated_successor = fabricated_current_intent.clone();
fabricated_successor
.advance(
TierProbeIntentState::Uploaded,
TierProbeRemoteVersion::versioned(uuid::Uuid::new_v4().to_string()),
)
.expect("fabricated successor should be internally valid");
let fabricated_current = observed_initial.with_intent_for_test(fabricated_current_intent.clone());
let crossed_cas = save_tier_probe_intent_record_if_current(store.clone(), &fabricated_current, &fabricated_successor)
.await
.expect_err("a live ETag must not authorize a different caller record");
assert!(matches!(crossed_cas, Error::PreconditionFailed));
assert_eq!(
load_tier_probe_intent_record(store.clone(), probe_id)
.await
.expect("crossed CAS must retain the authoritative record")
.intent(),
&initial
);
let mut fabricated_terminal_intent = fabricated_current_intent;
fabricated_terminal_intent
.advance(TierProbeIntentState::AbortedNoRemote, TierProbeRemoteVersion::default())
.expect("fabricated terminal should be internally valid");
let fabricated_terminal = observed_initial.with_intent_for_test(fabricated_terminal_intent);
let crossed_delete = delete_tier_probe_intent_record_if_current(store.clone(), &fabricated_terminal)
.await
.expect_err("a live ETag must not delete for a different caller record");
assert!(matches!(crossed_delete, Error::PreconditionFailed));
assert_eq!(
load_tier_probe_intent_record(store.clone(), probe_id)
.await
.expect("crossed delete must retain the authoritative record")
.intent(),
&initial
);
let remote_version = TierProbeRemoteVersion::versioned(uuid::Uuid::new_v4().to_string());
let mut uploaded = observed_initial.intent().clone();
uploaded
.advance(TierProbeIntentState::Uploaded, remote_version.clone())
.expect("known PUT result should advance");
save_tier_probe_intent_record_if_current(store.clone(), &observed_initial, &uploaded)
.await
.expect("the matching initial ETag should admit one successor");
let stale_cas = save_tier_probe_intent_record_if_current(store.clone(), &observed_initial, &uploaded)
.await
.expect_err("a consumed ETag must not overwrite the current generation");
assert!(matches!(stale_cas, Error::PreconditionFailed));
let observed_uploaded = load_tier_probe_intent_record(store.clone(), probe_id)
.await
.expect("uploaded generation should load");
assert_eq!(observed_uploaded.intent(), &uploaded);
let mut cleanup = observed_uploaded.intent().clone();
cleanup
.advance(TierProbeIntentState::CleanupPending, remote_version.clone())
.expect("known candidate should become cleanup-pending");
save_tier_probe_intent_record_if_current(store.clone(), &observed_uploaded, &cleanup)
.await
.expect("cleanup generation should persist by exact ETag");
let observed_cleanup = load_tier_probe_intent_record(store.clone(), probe_id)
.await
.expect("cleanup generation should load");
let mut completed = observed_cleanup.intent().clone();
completed
.advance(TierProbeIntentState::Completed, remote_version)
.expect("exact cleanup should become terminal");
save_tier_probe_intent_record_if_current(store.clone(), &observed_cleanup, &completed)
.await
.expect("terminal generation should persist by exact ETag");
let stale_terminal = observed_cleanup.with_intent_for_test(completed.clone());
let stale_delete = delete_tier_probe_intent_record_if_current(store.clone(), &stale_terminal)
.await
.expect_err("a stale ETag must not delete terminal evidence");
assert!(matches!(stale_delete, Error::PreconditionFailed));
let observed_completed = load_tier_probe_intent_record(store.clone(), probe_id)
.await
.expect("terminal generation should remain after stale delete");
assert_eq!(observed_completed.intent(), &completed);
delete_tier_probe_intent_record_if_current(store.clone(), &observed_completed)
.await
.expect("the exact terminal ETag should delete the record");
assert!(matches!(load_tier_probe_intent_record(store, probe_id).await, Err(Error::ConfigNotFound)));
}
#[cfg(feature = "test-util")]
#[tokio::test]
#[serial_test::serial(storage_class_env)]