gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-05 12:57:42 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(ecstore): make transitioned deletes durable (#5644)

Browse Source 
* fix(ecstore): make transitioned deletes durable

* fix(ecstore): journal force deletes

* fix(ecstore): journal force deletes
This commit is contained in:
cxymds
2026-08-03 02:00:11 +08:00
committed by GitHub
parent 8a65017f36
commit 2ce670837c
13 changed files with 823 additions and 264 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs
+8
View File
@@ -5928,6 +5928,8 @@ mod tests {
backend_identity: Some([1; 32]),
version_id_exact: true,
version_state: rustfs_filemeta::TransitionVersionState::Exact,
state: crate::bucket::lifecycle::tier_sweeper::TierDeleteJournalState::Committed,
source: None,
};
let err = state
@@ -6040,6 +6042,8 @@ mod tests {
backend_identity: Some([1; 32]),
version_id_exact: true,
version_state: rustfs_filemeta::TransitionVersionState::Exact,
state: crate::bucket::lifecycle::tier_sweeper::TierDeleteJournalState::Committed,
source: None,
};
state
@@ -10146,6 +10150,8 @@ mod tests {
backend_identity: Some(identity),
version_id_exact: false,
version_state: rustfs_filemeta::TransitionVersionState::Unknown,
state: crate::bucket::lifecycle::tier_sweeper::TierDeleteJournalState::Committed,
source: None,
};
let err = crate::bucket::lifecycle::tier_delete_journal::process_tier_delete_journal_entry(ecstore, &je)
@@ -10185,6 +10191,8 @@ mod tests {
backend_identity: Some(identity),
version_id_exact: true,
version_state: rustfs_filemeta::TransitionVersionState::Exact,
state: crate::bucket::lifecycle::tier_sweeper::TierDeleteJournalState::Committed,
source: None,
};
crate::set_disk::cleanup_rejected_transition_upload_durably(
crates/ecstore/src/bucket/lifecycle/tier_delete_journal.rs
+276 -6
View File
@@ -20,8 +20,10 @@ use tokio_util::sync::CancellationToken;
use tracing::{debug, warn};
use crate::bucket::lifecycle::config_boundary;
use crate::bucket::lifecycle::runtime_boundary;
use crate::bucket::lifecycle::tier_sweeper::{
Jentry, delete_confirmed_transition_candidate_exact_with_manager_and_identity,
Jentry, TierDeleteJournalState, TierDeleteSourceIdentity,
delete_confirmed_transition_candidate_exact_with_manager_and_identity,
delete_object_from_remote_tier_idempotent_with_manager_and_identity,
};
use crate::disk::RUSTFS_META_BUCKET;
@@ -30,7 +32,7 @@ use crate::object_api::{GetObjectReader, ObjectInfo, ObjectOptions, PutObjReader
use crate::services::tier::tier::tier_destination_id_from_metadata;
use crate::storage_api_contracts::{
list::ListOperations as _,
object::{DeletedObject, ObjectIO, ObjectOperations, ObjectToDelete},
object::{DeletedObject, HTTPPreconditions, ObjectIO, ObjectOperations, ObjectToDelete},
range::HTTPRangeSpec,
};
use crate::store::ECStore;
@@ -46,6 +48,7 @@ const TIER_DELETE_JOURNAL_RECOVERY_TIMEOUT: Duration = Duration::from_secs(300);
const TIER_DELETE_JOURNAL_VERSION: u8 = 2;
const TIER_DELETE_JOURNAL_EXACT_VERSION: u8 = 3;
const TIER_DELETE_JOURNAL_STATE_VERSION: u8 = 4;
const TIER_DELETE_JOURNAL_TRANSACTION_VERSION: u8 = 5;
pub(crate) const TIER_DELETE_JOURNAL_PREFIX: &str = "ilm/tier-delete-journal/";
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
@@ -61,13 +64,22 @@ struct PersistedTierDeleteJournalEntry {
version_id_exact: Option<bool>,
#[serde(default, skip_serializing_if = "Option::is_none")]
version_state: Option<rustfs_filemeta::TransitionVersionState>,
#[serde(default, skip_serializing_if = "Option::is_none")]
state: Option<TierDeleteJournalState>,
#[serde(default, skip_serializing_if = "Option::is_none")]
source: Option<TierDeleteSourceIdentity>,
}
impl PersistedTierDeleteJournalEntry {
fn from_jentry(je: &Jentry) -> Result<Self> {
validate_version_state(je.version_state, &je.version_id, je.version_id_exact)?;
let legacy_unknown = je.version_state == rustfs_filemeta::TransitionVersionState::Unknown;
let version = if legacy_unknown {
let version = if je.source.is_some() || je.state == TierDeleteJournalState::Prepared {
if je.backend_identity.is_none() {
return Err(Error::other("tier delete transaction is missing its backend identity"));
}
TIER_DELETE_JOURNAL_TRANSACTION_VERSION
} else if legacy_unknown {
if je.backend_identity.is_some() {
TIER_DELETE_JOURNAL_VERSION
} else {
@@ -87,6 +99,10 @@ impl PersistedTierDeleteJournalEntry {
backend_identity: je.backend_identity,
version_id_exact: je.version_id_exact.then_some(true),
version_state: (!legacy_unknown).then_some(je.version_state),
state: (version == TIER_DELETE_JOURNAL_TRANSACTION_VERSION).then_some(je.state),
source: (version == TIER_DELETE_JOURNAL_TRANSACTION_VERSION)
.then(|| je.source.clone())
.flatten(),
})
}
@@ -101,14 +117,21 @@ impl PersistedTierDeleteJournalEntry {
}
if self.version != TIER_DELETE_JOURNAL_EXACT_VERSION
&& self.version != TIER_DELETE_JOURNAL_STATE_VERSION
&& self.version != TIER_DELETE_JOURNAL_TRANSACTION_VERSION
&& self.version_id_exact.unwrap_or(false)
{
return Err(Error::other(
"legacy tier delete journal entry has an unsupported exact version constraint",
));
}
let (backend_identity, version_id_exact, version_state) = match self.version {
1 => (None, false, rustfs_filemeta::TransitionVersionState::Unknown),
let (backend_identity, version_id_exact, version_state, state, source) = match self.version {
1 => (
None,
false,
rustfs_filemeta::TransitionVersionState::Unknown,
TierDeleteJournalState::Committed,
None,
),
TIER_DELETE_JOURNAL_VERSION => (
Some(
self.backend_identity
@@ -116,6 +139,8 @@ impl PersistedTierDeleteJournalEntry {
),
false,
rustfs_filemeta::TransitionVersionState::Unknown,
TierDeleteJournalState::Committed,
None,
),
TIER_DELETE_JOURNAL_EXACT_VERSION => {
if self.version_id.is_empty() || self.version_id_exact != Some(true) {
@@ -128,6 +153,8 @@ impl PersistedTierDeleteJournalEntry {
),
true,
rustfs_filemeta::TransitionVersionState::Exact,
TierDeleteJournalState::Committed,
None,
)
}
TIER_DELETE_JOURNAL_STATE_VERSION => {
@@ -143,6 +170,31 @@ impl PersistedTierDeleteJournalEntry {
),
exact,
state,
TierDeleteJournalState::Committed,
None,
)
}
TIER_DELETE_JOURNAL_TRANSACTION_VERSION => {
let state = self
.state
.ok_or_else(|| Error::other("tier delete journal v5 entry is missing its state"))?;
let source = self
.source
.ok_or_else(|| Error::other("tier delete journal v5 entry is missing its source identity"))?;
let exact = self.version_id_exact.unwrap_or(false);
let version_state = self
.version_state
.ok_or_else(|| Error::other("tier delete journal v5 entry is missing its version state"))?;
validate_version_state(version_state, &self.version_id, exact)?;
(
Some(
self.backend_identity
.ok_or_else(|| Error::other("tier delete journal v5 entry is missing its backend identity"))?,
),
exact,
version_state,
state,
Some(source),
)
}
version => return Err(Error::other(format!("unsupported tier delete journal version {version}"))),
@@ -154,6 +206,8 @@ impl PersistedTierDeleteJournalEntry {
backend_identity,
version_id_exact,
version_state,
state,
source,
})
}
}
@@ -201,6 +255,20 @@ pub(crate) fn tier_delete_journal_object_name(je: &Jentry) -> String {
hasher.update([0]);
hasher.update(b"exact-version-id");
}
if let Some(source) = &je.source {
hasher.update([0]);
hasher.update(source.bucket.as_bytes());
hasher.update([0]);
hasher.update(source.object.as_bytes());
hasher.update([0]);
hasher.update(source.version_id.as_deref().unwrap_or_default().as_bytes());
hasher.update([0]);
hasher.update(source.data_dir.as_deref().unwrap_or_default().as_bytes());
hasher.update([0]);
hasher.update(source.etag.as_deref().unwrap_or_default().as_bytes());
hasher.update([0]);
hasher.update(source.mod_time.as_deref().unwrap_or_default().as_bytes());
}
format!(
"{TIER_DELETE_JOURNAL_PREFIX}{}.json",
rustfs_utils::crypto::hex(hasher.finalize().as_slice())
@@ -246,6 +314,66 @@ where
.map_err(std::io::Error::other)
}
pub async fn commit_tier_delete_journal_entry<S>(api: Arc<S>, je: &Jentry) -> std::io::Result<()>
where
S: ObjectIO<
Error = Error,
RangeSpec = HTTPRangeSpec,
HeaderMap = http::HeaderMap,
ObjectOptions = ObjectOptions,
ObjectInfo = ObjectInfo,
GetObjectReader = GetObjectReader,
PutObjectReader = PutObjReader,
>,
{
let mut committed = je.clone();
committed.state = TierDeleteJournalState::Committed;
persist_tier_delete_journal_entry(api, &committed).await
}
pub async fn abort_tier_delete_journal_entry<S>(api: Arc<S>, je: &Jentry) -> std::io::Result<()>
where
S: ObjectOperations<
Error = Error,
ObjectInfo = ObjectInfo,
ObjectOptions = ObjectOptions,
FileInfo = FileInfo,
ObjectToDelete = ObjectToDelete,
DeletedObject = DeletedObject,
>,
{
remove_tier_delete_journal_entry(api, je).await
}
pub async fn abort_prepared_tier_delete_journal_entry(api: Arc<ECStore>, je: &Jentry) -> std::io::Result<()> {
let name = tier_delete_journal_object_name(je);
let (data, metadata) = match config_boundary::read_config_with_metadata(api.clone(), &name, &ObjectOptions::default()).await {
Ok(result) => result,
Err(Error::ConfigNotFound) | Err(Error::FileNotFound) => return Ok(()),
Err(err) => return Err(std::io::Error::other(err)),
};
let current = decode_tier_delete_journal_entry(&data).map_err(std::io::Error::other)?;
if current.state != TierDeleteJournalState::Prepared {
return Ok(());
}
let etag = metadata
.etag
.ok_or_else(|| std::io::Error::other("prepared tier delete journal has no entity tag"))?;
match config_boundary::delete_config_if_match(api, &name, &etag).await {
Ok(()) | Err(Error::ConfigNotFound) => Ok(()),
Err(Error::PreconditionFailed) => Err(std::io::Error::new(
std::io::ErrorKind::WouldBlock,
"prepared tier delete journal changed before abort",
)),
Err(err) => Err(std::io::Error::other(err)),
}
}
pub(crate) async fn enqueue_committed_tier_delete_journal_entry(je: &Jentry) -> std::io::Result<()> {
let expiry_state = runtime_boundary::expiry_state_handle();
expiry_state.write().await.enqueue_tier_journal_entry(je)
}
pub async fn remove_tier_delete_journal_entry<S>(api: Arc<S>, je: &Jentry) -> std::io::Result<()>
where
S: ObjectOperations<
@@ -264,6 +392,13 @@ where
}
pub async fn process_tier_delete_journal_entry(api: Arc<ECStore>, je: &Jentry) -> std::io::Result<()> {
if je.state == TierDeleteJournalState::Prepared {
return reconcile_prepared_tier_delete_journal_entry(api, je).await;
}
process_committed_tier_delete_journal_entry(api, je).await
}
async fn process_committed_tier_delete_journal_entry(api: Arc<ECStore>, je: &Jentry) -> std::io::Result<()> {
if je.version_state == rustfs_filemeta::TransitionVersionState::Unknown {
return Err(std::io::Error::new(
std::io::ErrorKind::InvalidData,
@@ -296,6 +431,87 @@ pub async fn process_tier_delete_journal_entry(api: Arc<ECStore>, je: &Jentry) -
remove_tier_delete_journal_entry(api, je).await
}
async fn reconcile_prepared_tier_delete_journal_entry(api: Arc<ECStore>, je: &Jentry) -> std::io::Result<()> {
let (data, metadata) =
config_boundary::read_config_with_metadata(api.clone(), &tier_delete_journal_object_name(je), &ObjectOptions::default())
.await
.map_err(std::io::Error::other)?;
let current = decode_tier_delete_journal_entry(&data).map_err(std::io::Error::other)?;
if current.state != TierDeleteJournalState::Prepared {
return Err(std::io::Error::new(
std::io::ErrorKind::WouldBlock,
"prepared tier delete journal changed before reconciliation",
));
}
let Some(etag) = metadata.etag else {
return Err(std::io::Error::new(
std::io::ErrorKind::InvalidData,
"prepared tier delete journal has no entity tag",
));
};
let source = je
.source
.as_ref()
.ok_or_else(|| std::io::Error::new(std::io::ErrorKind::InvalidData, "prepared tier delete journal has no source"))?;
match api
.get_object_info(&source.bucket, &source.object, &source.lookup_options())
.await
{
Ok(info) if source.matches(&info) => {
match config_boundary::delete_config_if_match(api, &tier_delete_journal_object_name(&current), &etag).await {
Ok(()) => Ok(()),
Err(Error::PreconditionFailed) => Err(std::io::Error::new(
std::io::ErrorKind::WouldBlock,
"prepared tier delete journal changed before abort",
)),
Err(err) => Err(std::io::Error::other(err)),
}
}
Ok(_info) if source.has_stable_identity() => {
commit_prepared_tier_delete_journal_entry_if_current(api, current, etag).await
}
Ok(_) => Err(std::io::Error::new(
std::io::ErrorKind::WouldBlock,
"prepared tier delete journal source identity is not sufficient to confirm deletion",
)),
Err(Error::ObjectNotFound(_, _)) | Err(Error::FileNotFound) | Err(Error::FileVersionNotFound) => {
commit_prepared_tier_delete_journal_entry_if_current(api, current, etag).await
}
Err(err) => Err(std::io::Error::other(err)),
}
}
async fn commit_prepared_tier_delete_journal_entry_if_current(
api: Arc<ECStore>,
mut committed: Jentry,
etag: String,
) -> std::io::Result<()> {
committed.state = TierDeleteJournalState::Committed;
let data = encode_tier_delete_journal_entry(&committed).map_err(std::io::Error::other)?;
match config_boundary::save_config_with_opts(
api.clone(),
&tier_delete_journal_object_name(&committed),
data,
&ObjectOptions {
max_parity: true,
http_preconditions: Some(HTTPPreconditions {
if_match: Some(etag),
..Default::default()
}),
..Default::default()
},
)
.await
{
Ok(()) => process_committed_tier_delete_journal_entry(api, &committed).await,
Err(Error::PreconditionFailed) => Err(std::io::Error::new(
std::io::ErrorKind::WouldBlock,
"prepared tier delete journal changed before commit",
)),
Err(err) => Err(std::io::Error::other(err)),
}
}
pub async fn recover_tier_delete_journal_entries(
api: Arc<ECStore>,
limit: usize,
@@ -482,10 +698,13 @@ mod tests {
decode_tier_delete_journal_entry, encode_tier_delete_journal_entry, record_tier_delete_journal_backend_identity,
tier_delete_journal_object_name,
};
use crate::bucket::lifecycle::tier_sweeper::Jentry;
use crate::bucket::lifecycle::tier_sweeper::{Jentry, TierDeleteJournalState, TierDeleteSourceIdentity};
use crate::error::Result;
use crate::object_api::ObjectInfo;
use std::time::Duration;
use time::OffsetDateTime;
use tokio_util::sync::CancellationToken;
use uuid::Uuid;
fn journal_entry() -> Jentry {
Jentry {
@@ -495,6 +714,8 @@ mod tests {
backend_identity: Some([7; 32]),
version_id_exact: true,
version_state: rustfs_filemeta::TransitionVersionState::Exact,
state: TierDeleteJournalState::Committed,
source: None,
}
}
@@ -513,6 +734,55 @@ mod tests {
assert_eq!(decoded.version_state, je.version_state);
}
#[test]
fn tier_delete_transaction_roundtrips_prepared_source_identity() {
let mut je = journal_entry();
je.state = TierDeleteJournalState::Prepared;
je.source = Some(TierDeleteSourceIdentity {
bucket: "bucket".to_string(),
object: "object".to_string(),
version_id: Some("version".to_string()),
versioned: true,
version_suspended: false,
data_dir: Some("data-dir".to_string()),
etag: Some("etag".to_string()),
mod_time: Some("mod-time".to_string()),
});
let encoded = encode_tier_delete_journal_entry(&je).expect("prepared transaction should encode");
let value: serde_json::Value = serde_json::from_slice(&encoded).expect("transaction should be JSON");
assert_eq!(value["version"], serde_json::json!(5));
assert_eq!(value["state"], serde_json::json!("Prepared"));
assert!(value["source"].is_object());
let decoded = decode_tier_delete_journal_entry(&encoded).expect("prepared transaction should decode");
assert_eq!(decoded.state, TierDeleteJournalState::Prepared);
assert_eq!(decoded.source, je.source);
}
#[test]
fn tier_delete_source_identity_rejects_recreated_object() {
let version_id = Uuid::from_u128(1);
let data_dir = Uuid::from_u128(2);
let mod_time = OffsetDateTime::UNIX_EPOCH + time::Duration::seconds(1);
let info = ObjectInfo {
bucket: "bucket".to_string(),
name: "object".to_string(),
version_id: Some(version_id),
data_dir: Some(data_dir),
mod_time: Some(mod_time),
..Default::default()
};
let source = TierDeleteSourceIdentity::from_object_info("bucket", "object", &info, true, false);
assert!(source.matches(&info));
let recreated = ObjectInfo {
data_dir: Some(Uuid::from_u128(3)),
..info
};
assert!(!source.matches(&recreated));
}
#[test]
fn tier_delete_journal_roundtrips_exact_put_response_constraint() {
let mut exact = journal_entry();
crates/ecstore/src/bucket/lifecycle/tier_sweeper.rs
+115
View File
@@ -23,10 +23,12 @@ use crate::bucket::lifecycle::bucket_lifecycle_ops::ExpiryOp;
use crate::bucket::lifecycle::lifecycle::{self, ObjectOpts};
use crate::bucket::lifecycle::tier_delete_journal::persist_tier_delete_journal_entry;
use crate::client::signer_error::error_chain_contains_signer_header_marker;
use crate::object_api::ObjectInfo;
use crate::services::tier::tier::{TierConfigMgr, TierDestinationId, TierOperationLease};
use crate::storage_api_contracts::lifecycle::TransitionedObject;
use crate::store::ECStore;
use rustfs_utils::get_env_usize;
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use std::any::Any;
use std::collections::VecDeque;
@@ -257,6 +259,8 @@ impl ObjSweeper {
rustfs_filemeta::TransitionVersionState::SuspendedNull | rustfs_filemeta::TransitionVersionState::Exact
),
version_state: self.transition_version_state,
state: TierDeleteJournalState::Committed,
source: None,
});
}
None
@@ -285,6 +289,76 @@ impl ObjSweeper {
}
}
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
pub(crate) enum TierDeleteJournalState {
Prepared,
Committed,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(deny_unknown_fields)]
pub(crate) struct TierDeleteSourceIdentity {
pub(crate) bucket: String,
pub(crate) object: String,
pub(crate) version_id: Option<String>,
pub(crate) versioned: bool,
pub(crate) version_suspended: bool,
pub(crate) data_dir: Option<String>,
pub(crate) etag: Option<String>,
pub(crate) mod_time: Option<String>,
}
impl TierDeleteSourceIdentity {
pub(crate) fn from_object_info(
bucket: &str,
object: &str,
info: &ObjectInfo,
versioned: bool,
version_suspended: bool,
) -> Self {
Self {
bucket: bucket.to_string(),
object: object.to_string(),
version_id: info.version_id.map(|id| id.to_string()),
versioned,
version_suspended,
data_dir: info.data_dir.map(|id| id.to_string()),
etag: info.etag.clone(),
mod_time: info.mod_time.map(|time| time.to_string()),
}
}
pub(crate) fn lookup_options(&self) -> crate::object_api::ObjectOptions {
crate::object_api::ObjectOptions {
version_id: self.version_id.clone(),
versioned: self.versioned,
version_suspended: self.version_suspended,
..Default::default()
}
}
pub(crate) fn matches(&self, info: &ObjectInfo) -> bool {
if self.bucket != info.bucket {
return false;
}
if let Some(version_id) = &self.version_id {
return info.version_id.map(|id| id.to_string()).as_deref() == Some(version_id.as_str())
&& self.data_dir == info.data_dir.map(|id| id.to_string());
}
if self.data_dir.is_some() {
return self.data_dir == info.data_dir.map(|id| id.to_string());
}
self.etag.is_some()
&& self.etag == info.etag
&& self.mod_time.is_some()
&& self.mod_time == info.mod_time.map(|time| time.to_string())
}
pub(crate) fn has_stable_identity(&self) -> bool {
self.version_id.is_some() || self.data_dir.is_some() || (self.etag.is_some() && self.mod_time.is_some())
}
}
#[derive(Debug, Clone)]
#[allow(unused_assignments)]
pub struct Jentry {
@@ -294,6 +368,8 @@ pub struct Jentry {
pub(crate) backend_identity: Option<TierDestinationId>,
pub(crate) version_id_exact: bool,
pub(crate) version_state: rustfs_filemeta::TransitionVersionState,
pub(crate) state: TierDeleteJournalState,
pub(crate) source: Option<TierDeleteSourceIdentity>,
}
impl ExpiryOp for Jentry {
@@ -554,9 +630,48 @@ pub fn transitioned_force_delete_journal_entry(
rustfs_filemeta::TransitionVersionState::SuspendedNull | rustfs_filemeta::TransitionVersionState::Exact
),
version_state: transition_version_state,
state: TierDeleteJournalState::Committed,
source: None,
})
}
pub(crate) fn attach_tier_delete_source(
je: &mut Jentry,
bucket: &str,
object: &str,
info: &ObjectInfo,
versioned: bool,
version_suspended: bool,
) {
je.state = TierDeleteJournalState::Prepared;
je.source = Some(TierDeleteSourceIdentity::from_object_info(
bucket,
object,
info,
versioned,
version_suspended,
));
}
pub(crate) fn transitioned_delete_journal_entry_for_source(
version_id: Option<Uuid>,
versioned: bool,
suspended: bool,
bucket: &str,
object: &str,
source: &ObjectInfo,
) -> Option<Jentry> {
let mut je = transitioned_delete_journal_entry(
version_id,
versioned,
suspended,
&source.transitioned_object,
source.transition_version_state,
)?;
attach_tier_delete_source(&mut je, bucket, object, source, versioned, suspended);
Some(je)
}
#[cfg(test)]
mod test {
use crate::client::signer_error::invalid_utf8_header_error;
crates/ecstore/src/core/sets.rs
+3 -1
View File
@@ -339,7 +339,9 @@ impl Sets {
futures.push(set.delete_object(bucket, object, opt.clone()));
}
let _results = join_all(futures).await;
if let Some(err) = join_all(futures).await.into_iter().find_map(Result::err) {
return Err(err);
}
Ok(())
}
crates/ecstore/src/object_api/types.rs
+3
View File
@@ -112,6 +112,9 @@ pub struct ObjectOptions {
pub want_checksum: Option<Checksum>,
pub skip_verify_bitrot: bool,
pub capacity_scope_token: Option<Uuid>,
/// Storage-owned journal writer used by the atomic delete path. This is
/// populated only by the `ECStore` wrapper that holds the namespace locks.
pub tier_delete_journal_api: Option<Arc<crate::store::ECStore>>,
}
impl ObjectOptions {
crates/ecstore/src/services/tier/tier.rs
+2
View File
@@ -9383,6 +9383,8 @@ mod tests {
backend_identity: Some(current_identity),
version_id_exact: true,
version_state: rustfs_filemeta::TransitionVersionState::Exact,
state: crate::bucket::lifecycle::tier_sweeper::TierDeleteJournalState::Committed,
source: None,
};
journal_store
.insert_config_object(
crates/ecstore/src/set_disk/ops/object.rs
+88 -4
View File
@@ -24,10 +24,14 @@ use super::bitrot_self_verify::{BitrotSelfVerifyTarget, drop_failed_writer_disks
use crate::set_disk::read::GetObjectDownstreamWriter;
use crate::bucket::lifecycle::{
tier_delete_journal::{persist_tier_delete_journal_entry, remove_tier_delete_journal_entry},
tier_delete_journal::{
enqueue_committed_tier_delete_journal_entry, persist_tier_delete_journal_entry,
record_tier_delete_journal_backend_identity, remove_tier_delete_journal_entry,
},
tier_sweeper::{
Jentry, RemoteTierDeleteOutcome, delete_confirmed_transition_candidate_exact_with_lease_idempotent,
delete_object_from_remote_tier_with_lease_idempotent,
Jentry, RemoteTierDeleteOutcome, TierDeleteJournalState,
delete_confirmed_transition_candidate_exact_with_lease_idempotent, delete_object_from_remote_tier_with_lease_idempotent,
transitioned_delete_journal_entry_for_source,
},
transition_transaction::{
TransitionRemoteVersion, TransitionSourceIdentity, TransitionSourceVersionMode, TransitionTransaction,
@@ -1851,6 +1855,8 @@ pub(crate) async fn cleanup_rejected_transition_upload_durably(
} else {
rustfs_filemeta::TransitionVersionState::Exact
},
state: TierDeleteJournalState::Committed,
source: None,
};
let journal_error = if let Some(api) = api.as_ref() {
@@ -3208,6 +3214,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
object_lock_delete_check_required(metadata_sys::get_in(&self.ctx, bucket).await.ok().as_deref());
let mut vers_map: HashMap<&String, FileInfoVersions> = HashMap::new();
let mut journal_entries: Vec<(usize, Jentry)> = Vec::new();
for (i, dobj) in objects.iter().enumerate() {
if del_errs[i].is_some() {
@@ -3232,7 +3239,8 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
let marker_delete = dobj.version_id.is_none() || dobj.synthetic_version_id;
let replication_needs_source = replicate_delete
&& (!marker_delete || delete_config_snapshot.active_delete_marker_rules_require_tags(&replication_object_name));
let (goi, gerr) = if object_lock_checks_required || replication_needs_source {
let (goi, gerr) = if object_lock_checks_required || replication_needs_source || opts.tier_delete_journal_api.is_some()
{
let (goi, _write_quorum, gerr) = self.get_object_info_and_quorum(bucket, &dobj.object_name, &check_opts).await;
(goi, gerr)
} else {
@@ -3241,8 +3249,14 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
let source_missing = gerr
.as_ref()
.is_some_and(|err| is_err_object_not_found(err) || is_err_version_not_found(err));
let explicit_delete_marker = opts.tier_delete_journal_api.is_some()
&& dobj.version_id.is_some()
&& goi.delete_marker
&& goi.version_id == version_id
&& matches!(gerr.as_ref(), Some(StorageError::MethodNotAllowed));
if let Some(err) = gerr.as_ref()
&& !source_missing
&& !explicit_delete_marker
{
del_errs[i] = Some(err.clone());
continue;
@@ -3255,6 +3269,23 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
continue;
}
if opts.tier_delete_journal_api.is_some()
&& let Some(mut je) = transitioned_delete_journal_entry_for_source(
version_id,
versioned,
version_suspended,
bucket,
&replication_object_name,
&goi,
)
{
if let Err(err) = record_tier_delete_journal_backend_identity(&mut je, &goi.user_defined) {
del_errs[i] = Some(Error::other(err));
continue;
}
journal_entries.push((i, je));
}
let mut admitted = dobj.clone();
admitted.object_name = replication_object_name;
if admitted.synthetic_version_id {
@@ -3300,6 +3331,11 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
}
}
if goi.delete_marker && dobj.version_id.is_some() && goi.version_id == version_id {
vr.deleted = true;
vr.mod_time = goi.mod_time;
}
let v = {
if vers_map.contains_key(&dobj.object_name) {
let val = vers_map.get_mut(&dobj.object_name).unwrap();
@@ -3372,6 +3408,23 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
return (del_objects, del_errs);
}
let mut persisted_journal_entries = Vec::with_capacity(journal_entries.len());
if let Some(api) = opts.tier_delete_journal_api.as_ref() {
for (idx, mut je) in journal_entries {
if let Err(err) = persist_tier_delete_journal_entry(Arc::clone(api), &je).await {
del_errs[idx] = Some(Error::other(err));
continue;
}
je.state = TierDeleteJournalState::Prepared;
persisted_journal_entries.push((idx, je));
}
}
for fi_vers in &mut vers {
fi_vers.versions.retain(|fi| del_errs[fi.idx].is_none());
}
vers.retain(|fi_vers| !fi_vers.versions.is_empty());
let rollback_dir = Uuid::new_v4();
let disks = self.disks.read().await;
@@ -3535,6 +3588,37 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
// TODO: add_partial
if let Some(api) = opts.tier_delete_journal_api.as_ref() {
for (idx, je) in persisted_journal_entries {
if del_errs[idx].is_none() {
let mut committed = je;
committed.state = TierDeleteJournalState::Committed;
if let Err(err) = persist_tier_delete_journal_entry(Arc::clone(api), &committed).await {
warn!(
object = %committed.obj_name,
tier = %committed.tier_name,
error = ?err,
"batch tier delete committed locally but journal commit failed; recovery will retry"
);
} else if let Err(err) = enqueue_committed_tier_delete_journal_entry(&committed).await {
warn!(
object = %committed.obj_name,
tier = %committed.tier_name,
error = ?err,
"batch tier delete journal committed but could not be queued; recovery will retry"
);
}
} else if let Err(err) = remove_tier_delete_journal_entry(Arc::clone(api), &je).await {
warn!(
object = %je.obj_name,
tier = %je.tier_name,
error = ?err,
"failed to remove aborted batch tier delete journal"
);
}
}
}
if dist_erasure {
self.release_dist_delete_object_locks_batch(dist_batch_lock_ids).await;
}
crates/ecstore/src/store/init.rs
+4
View File
@@ -1379,6 +1379,8 @@ mod tests {
backend_identity: Some(identity_a),
version_id_exact: true,
version_state: rustfs_filemeta::TransitionVersionState::Exact,
state: crate::bucket::lifecycle::tier_sweeper::TierDeleteJournalState::Committed,
source: None,
};
let entry_b = Jentry {
obj_name: "remote-b".to_string(),
@@ -1387,6 +1389,8 @@ mod tests {
backend_identity: Some(identity_b),
version_id_exact: true,
version_state: rustfs_filemeta::TransitionVersionState::Exact,
state: crate::bucket::lifecycle::tier_sweeper::TierDeleteJournalState::Committed,
source: None,
};
let remove_a = backend_a.arm_failing_remove_barrier().await;
persist_tier_delete_journal_entry(store_a.clone(), &entry_a)
crates/ecstore/src/store/object.rs
+316 -3
View File
@@ -13,6 +13,16 @@
// limitations under the License.
use super::*;
use crate::bucket::lifecycle::{
tier_delete_journal::{
abort_prepared_tier_delete_journal_entry as abort_prepared_journal_entry_if_current, commit_tier_delete_journal_entry,
enqueue_committed_tier_delete_journal_entry, persist_tier_delete_journal_entry,
record_tier_delete_journal_backend_identity,
},
tier_sweeper::{
Jentry, attach_tier_delete_source, transitioned_delete_journal_entry_for_source, transitioned_force_delete_journal_entry,
},
};
use crate::bucket::replication::ReplicationObjectBridge;
use crate::disk::OldCurrentSize;
use crate::object_api::DeleteLockFence;
@@ -33,6 +43,181 @@ use std::{
};
use tokio::io::{AsyncRead, ReadBuf};
const FORCE_DELETE_LIST_PAGE_SIZE: i32 = 1_000;
fn build_tier_delete_journal_entry(
bucket: &str,
object: &str,
opts: &ObjectOptions,
source: &ObjectInfo,
) -> Result<Option<Jentry>> {
let version_id = opts.version_id.as_deref().map(Uuid::parse_str).transpose()?;
let source_object = decode_dir_object(object);
let Some(mut je) = (if opts.delete_prefix {
transitioned_force_delete_journal_entry(&source.transitioned_object, source.transition_version_state).map(|mut je| {
attach_tier_delete_source(&mut je, bucket, source_object.as_str(), source, opts.versioned, opts.version_suspended);
je
})
} else {
transitioned_delete_journal_entry_for_source(
version_id,
opts.versioned,
opts.version_suspended,
bucket,
source_object.as_str(),
source,
)
}) else {
return Ok(None);
};
record_tier_delete_journal_backend_identity(&mut je, &source.user_defined).map_err(Error::other)?;
Ok(Some(je))
}
async fn prepare_tier_delete_journal_entry(
api: &Arc<ECStore>,
bucket: &str,
object: &str,
opts: &ObjectOptions,
source: &ObjectInfo,
) -> Result<Option<Jentry>> {
let Some(je) = build_tier_delete_journal_entry(bucket, object, opts, source)? else {
return Ok(None);
};
persist_tier_delete_journal_entry(Arc::clone(api), &je)
.await
.map_err(Error::other)?;
Ok(Some(je))
}
async fn abort_prepared_tier_delete_journal_entry(api: &Arc<ECStore>, je: &Jentry) {
if let Err(err) = abort_prepared_journal_entry_if_current(Arc::clone(api), je).await {
warn!(
object = %je.obj_name,
tier = %je.tier_name,
error = ?err,
"failed to remove aborted tier delete journal"
);
}
}
async fn abort_prepared_tier_delete_journal_entries(api: &Arc<ECStore>, entries: &[Jentry]) {
for entry in entries {
abort_prepared_tier_delete_journal_entry(api, entry).await;
}
}
async fn commit_prepared_tier_delete_journal_entry(api: &Arc<ECStore>, je: &Jentry) {
if let Err(err) = commit_tier_delete_journal_entry(Arc::clone(api), je).await {
warn!(
object = %je.obj_name,
tier = %je.tier_name,
error = ?err,
"tier delete committed locally but journal commit failed; recovery will retry"
);
return;
}
let mut committed = je.clone();
committed.state = crate::bucket::lifecycle::tier_sweeper::TierDeleteJournalState::Committed;
if let Err(err) = enqueue_committed_tier_delete_journal_entry(&committed).await {
warn!(
object = %je.obj_name,
tier = %je.tier_name,
error = ?err,
"tier delete journal committed but could not be queued; recovery will retry"
);
}
}
async fn commit_prepared_tier_delete_journal_entries(api: &Arc<ECStore>, entries: &[Jentry]) {
for entry in entries {
commit_prepared_tier_delete_journal_entry(api, entry).await;
}
}
async fn prepare_prefix_tier_delete_journal_entries(
api: &Arc<ECStore>,
bucket: &str,
prefix: &str,
opts: &ObjectOptions,
) -> Result<Vec<Jentry>> {
let mut marker = None;
let mut version_marker = None;
let mut entries = Vec::new();
loop {
let page = Arc::clone(api)
.list_object_versions_for_lifecycle(
bucket,
prefix,
marker.clone(),
version_marker.clone(),
None,
FORCE_DELETE_LIST_PAGE_SIZE,
)
.await?;
for source in page.objects {
if let Some(entry) = build_tier_delete_journal_entry(bucket, &source.name, opts, &source)? {
entries.push(entry);
}
}
if !page.is_truncated {
break;
}
let next_marker = page
.next_marker
.ok_or_else(|| Error::other("truncated force delete listing has no next marker"))?;
let next_version_marker = page.next_version_idmarker;
if marker.as_deref() == Some(next_marker.as_str()) && version_marker == next_version_marker {
return Err(Error::other("force delete listing marker did not advance"));
}
marker = Some(next_marker);
version_marker = next_version_marker;
}
let mut persisted = Vec::with_capacity(entries.len());
for entry in entries {
if let Err(err) = persist_tier_delete_journal_entry(Arc::clone(api), &entry).await {
abort_prepared_tier_delete_journal_entries(api, &persisted).await;
return Err(Error::other(err));
}
persisted.push(entry);
}
Ok(persisted)
}
async fn delete_prefix_with_tier_delete_journal(
store: &ECStore,
bucket: &str,
object: &str,
opts: &ObjectOptions,
tier_journal_api: Option<&Arc<ECStore>>,
) -> Result<()> {
let journal_entry = if let Some(api) = tier_journal_api {
Some(prepare_prefix_tier_delete_journal_entries(api, bucket, object, opts).await?)
} else {
None
};
let result = store.delete_prefix(bucket, object, opts).await;
match result {
Ok(()) => {
if let (Some(api), Some(entries)) = (tier_journal_api, journal_entry.as_ref()) {
commit_prepared_tier_delete_journal_entries(api, entries).await;
}
Ok(())
}
Err(err) => {
if let (Some(api), Some(entries)) = (tier_journal_api, journal_entry.as_ref()) {
abort_prepared_tier_delete_journal_entries(api, entries).await;
}
Err(err)
}
}
}
/// A GET whose object identity has been resolved while its namespace read lock
/// remains held, but whose body reader has not been constructed yet.
///
@@ -1092,8 +1277,49 @@ impl ECStore {
purged
}
pub async fn delete_object_with_tier_delete_journal(
self: &Arc<Self>,
bucket: &str,
object: &str,
opts: ObjectOptions,
) -> Result<ObjectInfo> {
let result = self
.handle_delete_object_with_journal(bucket, object, opts, Some(Arc::clone(self)))
.await;
if result.is_ok() {
list_objects::observe_list_objects_mutation(self, bucket).await;
}
result
}
pub async fn delete_objects_with_tier_delete_journal(
self: &Arc<Self>,
bucket: &str,
objects: Vec<ObjectToDelete>,
opts: ObjectOptions,
) -> (Vec<DeletedObject>, Vec<Option<Error>>) {
let result = self
.handle_delete_objects_with_journal(bucket, objects, opts, Some(Arc::clone(self)))
.await;
let success_count = result.1.iter().filter(|err| err.is_none()).count();
if success_count > 0 {
list_objects::observe_list_objects_mutations(self, bucket, success_count).await;
}
result
}
#[instrument(skip(self))]
pub(super) async fn handle_delete_object(&self, bucket: &str, object: &str, opts: ObjectOptions) -> Result<ObjectInfo> {
self.handle_delete_object_with_journal(bucket, object, opts, None).await
}
pub(super) async fn handle_delete_object_with_journal(
&self,
bucket: &str,
object: &str,
opts: ObjectOptions,
tier_journal_api: Option<Arc<ECStore>>,
) -> Result<ObjectInfo> {
check_del_obj_args(bucket, object)?;
let object = if opts.delete_prefix && !opts.delete_prefix_object {
@@ -1103,11 +1329,12 @@ impl ECStore {
};
let object = object.as_str();
let mut opts = opts;
opts.tier_delete_journal_api = tier_journal_api.clone();
if opts.delete_prefix && !opts.delete_prefix_object {
// Prefix deletes cover multiple object keys; an exact lock on the prefix string
// would not protect child objects.
self.delete_prefix(bucket, object, &opts).await?;
delete_prefix_with_tier_delete_journal(self, bucket, object, &opts, tier_journal_api.as_ref()).await?;
return Ok(ObjectInfo::default());
}
@@ -1119,7 +1346,7 @@ impl ECStore {
};
if opts.delete_prefix {
self.delete_prefix(bucket, object, &opts).await?;
delete_prefix_with_tier_delete_journal(self, bucket, object, &opts, tier_journal_api.as_ref()).await?;
return Ok(ObjectInfo::default());
}
@@ -1215,8 +1442,25 @@ impl ECStore {
));
}
let journal_entry = if let Some(api) = tier_journal_api.as_ref() {
prepare_tier_delete_journal_entry(api, bucket, object, &opts, &pinfo.object_info).await?
} else {
None
};
if !errs.is_empty() && !opts.versioned && !opts.version_suspended {
let mut obj = self.delete_object_from_all_pools(bucket, object, &opts, errs).await?;
let mut obj = match self.delete_object_from_all_pools(bucket, object, &opts, errs).await {
Ok(obj) => obj,
Err(err) => {
if let (Some(api), Some(je)) = (tier_journal_api.as_ref(), journal_entry.as_ref()) {
abort_prepared_tier_delete_journal_entry(api, je).await;
}
return Err(err);
}
};
if let (Some(api), Some(je)) = (tier_journal_api.as_ref(), journal_entry.as_ref()) {
commit_prepared_tier_delete_journal_entry(api, je).await;
}
obj.name = decode_dir_object(object);
return Ok(obj);
}
@@ -1224,6 +1468,9 @@ impl ECStore {
for pool in self.pools.iter() {
match pool.delete_object(bucket, object, opts.clone()).await {
Ok(res) => {
if let (Some(api), Some(je)) = (tier_journal_api.as_ref(), journal_entry.as_ref()) {
commit_prepared_tier_delete_journal_entry(api, je).await;
}
let mut obj = res;
obj.name = decode_dir_object(object);
return Ok(obj);
@@ -1236,6 +1483,10 @@ impl ECStore {
}
}
if let (Some(api), Some(je)) = (tier_journal_api.as_ref(), journal_entry.as_ref()) {
abort_prepared_tier_delete_journal_entry(api, je).await;
}
if let Some(ver) = opts.version_id {
return Err(StorageError::VersionNotFound(bucket.to_owned(), object.to_owned(), ver));
}
@@ -1249,6 +1500,16 @@ impl ECStore {
bucket: &str,
objects: Vec<ObjectToDelete>,
opts: ObjectOptions,
) -> (Vec<DeletedObject>, Vec<Option<Error>>) {
self.handle_delete_objects_with_journal(bucket, objects, opts, None).await
}
pub(super) async fn handle_delete_objects_with_journal(
&self,
bucket: &str,
objects: Vec<ObjectToDelete>,
opts: ObjectOptions,
tier_journal_api: Option<Arc<ECStore>>,
) -> (Vec<DeletedObject>, Vec<Option<Error>>) {
// encode object name
let objects: Vec<ObjectToDelete> = objects
@@ -1269,6 +1530,7 @@ impl ECStore {
}
let mut opts = opts;
opts.tier_delete_journal_api = tier_journal_api;
if opts.delete_replication_config_snapshot.is_none() {
match ReplicationObjectBridge::delete_request_config_in(&self.ctx, bucket).await {
Ok(snapshot) => opts.delete_replication_config_snapshot = Some(Arc::new(snapshot)),
@@ -1625,6 +1887,7 @@ impl ECStore {
mod tests {
use super::*;
use crate::bucket::lifecycle::core::TRANSITION_COMPLETE;
use crate::bucket::lifecycle::tier_sweeper::TierDeleteJournalState;
use crate::bucket::replication::{
ReplicationState, ReplicationStatusType, VersionPurgeStatusType, replication_state_to_filemeta, replication_statuses_map,
version_purge_statuses_map,
@@ -1640,6 +1903,7 @@ mod tests {
};
use crate::set_disk::SetDisks;
use crate::storage_api_contracts::bucket::MakeBucketOptions;
use crate::storage_api_contracts::lifecycle::TransitionedObject;
use bytes::Bytes;
use std::io::Cursor;
use std::sync::Arc;
@@ -1660,6 +1924,55 @@ mod tests {
struct BodyCacheHookGuard;
#[test]
fn tier_delete_entry_is_prepared_and_bound_to_source_generation() {
let identity = [9_u8; 32];
let mut metadata = HashMap::new();
rustfs_utils::http::metadata_compat::insert_str(
&mut metadata,
rustfs_utils::http::metadata_compat::SUFFIX_TRANSITION_TIER_DESTINATION_ID,
rustfs_utils::crypto::hex(identity),
);
let version_id = Uuid::from_u128(1);
let data_dir = Uuid::from_u128(2);
let source = ObjectInfo {
bucket: "bucket".to_string(),
name: "object".to_string(),
version_id: Some(version_id),
data_dir: Some(data_dir),
user_defined: Arc::new(metadata),
transitioned_object: TransitionedObject {
name: "remote/object".to_string(),
version_id: "remote-version".to_string(),
tier: "WARM".to_string(),
status: TRANSITION_COMPLETE.to_string(),
..Default::default()
},
transition_version_state: rustfs_filemeta::TransitionVersionState::Exact,
..Default::default()
};
let entry = build_tier_delete_journal_entry(
"bucket",
"object",
&ObjectOptions {
version_id: Some(version_id.to_string()),
versioned: true,
..Default::default()
},
&source,
)
.expect("transition source should produce a journal entry")
.expect("completed transition should be journaled");
assert_eq!(entry.state, TierDeleteJournalState::Prepared);
assert_eq!(entry.backend_identity, Some(identity));
let data_dir_string = data_dir.to_string();
assert_eq!(
entry.source.as_ref().and_then(|source| source.data_dir.as_deref()),
Some(data_dir_string.as_str())
);
}
impl Drop for BodyCacheHookGuard {
fn drop(&mut self) {
clear_get_object_body_cache_hook();
rustfs/src/app/object_usecase.rs
+7 -192
View File
@@ -32,7 +32,6 @@ use super::storage_api::object_usecase::bucket::{
bucket_lifecycle_audit::LcEventSrc,
bucket_lifecycle_ops::{enqueue_transition_immediate, post_restore_opts},
lifecycle::{self, TransitionOptions},
tier_delete_journal, tier_sweeper,
},
metadata_sys,
object_lock::{
@@ -108,8 +107,8 @@ use super::storage_api::object_usecase::{
validate_ssec_for_read, wrap_response_with_cors,
};
use crate::app::runtime_sources::{
AppContext, current_app_context, current_expiry_state_handle, current_notify_interface_for_context,
current_object_data_cache_for_context, current_object_store_handle_for_context,
AppContext, current_app_context, current_notify_interface_for_context, current_object_data_cache_for_context,
current_object_store_handle_for_context,
};
use crate::config::RustFSBufferConfig;
use crate::delete_tail_activity::{DeleteTailActivityGuard, DeleteTailStage};
@@ -751,53 +750,6 @@ fn tune_reader_stream_buffer_size(
selected_size
}
async fn enqueue_transitioned_delete_cleanup(
store: Arc<ECStore>,
bucket: &str,
object: &str,
opts: &ObjectOptions,
existing: Option<&ObjectInfo>,
) -> std::io::Result<()> {
let Some(existing) = existing else {
return Ok(());
};
let _activity_guard = DeleteTailActivityGuard::new(DeleteTailStage::Cleanup);
let je = if opts.delete_prefix {
tier_sweeper::transitioned_force_delete_journal_entry(&existing.transitioned_object, existing.transition_version_state)
} else {
let version_id = opts.version_id.as_ref().and_then(|v| Uuid::parse_str(v).ok());
tier_sweeper::transitioned_delete_journal_entry(
version_id,
opts.versioned,
opts.version_suspended,
&existing.transitioned_object,
existing.transition_version_state,
)
};
let Some(mut je) = je else {
return Ok(());
};
tier_delete_journal::record_tier_delete_journal_backend_identity(&mut je, &existing.user_defined)?;
tier_delete_journal::persist_tier_delete_journal_entry(store, &je).await?;
let expiry_state = current_expiry_state_handle();
let mut expiry_state = expiry_state.write().await;
if let Err(err) = expiry_state.enqueue_tier_journal_entry(&je) {
warn!(
bucket,
object,
remote_object = %existing.transitioned_object.name,
remote_version_id = %existing.transitioned_object.version_id,
tier = %existing.transitioned_object.tier,
error = ?err,
"transitioned object cleanup journal persisted but was not queued"
);
}
Ok(())
}
pin_project! {
struct ExtractArchiveEtagReader<R> {
#[pin]
@@ -6932,7 +6884,7 @@ impl DefaultObjectUsecase {
invalidate_object_data_cache_objects_before_mutation(&cache_adapter, &bucket, cache_keys_before_delete.iter()).await;
let (dobjs, errs) = store
.delete_objects(
.delete_objects_with_tier_delete_journal(
&bucket,
object_to_delete.clone(),
ObjectOptions {
@@ -6969,27 +6921,6 @@ impl DefaultObjectUsecase {
{
delete_results[didx].delete_object = Some(dobjs[i].clone());
let (versioned, version_suspended) = object_versioning[i];
if let Err(err) = enqueue_transitioned_delete_cleanup(
store.clone(),
&bucket,
&object_to_delete[i].object_name,
&ObjectOptions {
version_id: object_to_delete[i].version_id.map(|v| v.to_string()),
versioned,
version_suspended,
..Default::default()
},
existing_object_infos[i].as_ref(),
)
.await
{
warn!(
bucket = %bucket,
object = %object_to_delete[i].object_name,
error = ?err,
"failed to persist transitioned object cleanup journal"
);
}
let creates_delete_marker = object_to_delete[i].version_id.is_none() && versioned && !version_suspended;
if creates_delete_marker {
record_bucket_delete_marker_memory(&bucket).await;
@@ -7246,7 +7177,10 @@ impl DefaultObjectUsecase {
}
let obj_info = {
match store.delete_object(&bucket, &key, opts.clone()).await {
match store
.delete_object_with_tier_delete_journal(&bucket, &key, opts.clone())
.await
{
Ok(obj) => obj,
Err(err) => {
if is_err_bucket_not_found(&err) {
@@ -7263,16 +7197,6 @@ impl DefaultObjectUsecase {
}
};
if let Err(err) =
enqueue_transitioned_delete_cleanup(store.clone(), &bucket, &key, &opts, existing_object_info.as_ref()).await
{
warn!(
bucket = %bucket,
object = %key,
error = ?err,
"failed to persist transitioned object cleanup journal"
);
}
if force_delete {
let _ = invalidate_object_data_cache_prefix_after_delete(&cache_adapter, &bucket, &key).await;
} else {
@@ -9864,115 +9788,6 @@ mod tests {
(store, context)
}
#[tokio::test]
#[serial_test::serial]
async fn transitioned_delete_cleanup_persists_known_state_and_rejects_unknown_state() {
let store = crate::app::gating_test_env::shared_gating_ecstore().await;
if current_app_context().is_none() {
crate::app::runtime_sources::install_test_app_context(Arc::clone(&store)).await;
}
let identity = [11_u8; 32];
let mut metadata = HashMap::new();
rustfs_utils::http::metadata_compat::insert_str(
&mut metadata,
rustfs_utils::http::metadata_compat::SUFFIX_TRANSITION_TIER_DESTINATION_ID,
rustfs_utils::crypto::hex(identity),
);
let mut current = ObjectInfo {
user_defined: Arc::new(metadata),
..Default::default()
};
current.transitioned_object.status = lifecycle::TRANSITION_COMPLETE.to_string();
current.transitioned_object.tier = "WARM".to_string();
current.transitioned_object.name = "remote/identity-bound".to_string();
current.transitioned_object.version_id = "remote-version".to_string();
current.transition_version_state = rustfs_filemeta::TransitionVersionState::Exact;
let journal_name = |remote_object: &str, backend_identity: Option<[u8; 32]>, version_id_exact: bool| {
use sha2::{Digest, Sha256};
let mut hasher = Sha256::new();
hasher.update(b"WARM");
hasher.update([0]);
hasher.update(remote_object.as_bytes());
hasher.update([0]);
hasher.update(b"remote-version");
if let Some(backend_identity) = backend_identity {
hasher.update([0]);
hasher.update(backend_identity);
}
if version_id_exact {
hasher.update([0]);
hasher.update(b"exact-version-id");
}
format!("ilm/tier-delete-journal/{}.json", rustfs_utils::crypto::hex(hasher.finalize().as_slice()))
};
enqueue_transitioned_delete_cleanup(store.clone(), "bucket", "identity-bound", &ObjectOptions::default(), Some(&current))
.await
.expect("normal transitioned delete should persist an identity-bound journal");
let mut identity_bound = store
.get_object_reader(
".rustfs.sys",
&journal_name("remote/identity-bound", Some(identity), true),
None,
http::HeaderMap::new(),
&ObjectOptions::default(),
)
.await
.expect("identity-bound journal should be readable");
let mut identity_bound_data = Vec::new();
tokio::io::AsyncReadExt::read_to_end(&mut identity_bound.stream, &mut identity_bound_data)
.await
.expect("identity-bound journal body should be readable");
let identity_bound: serde_json::Value =
serde_json::from_slice(&identity_bound_data).expect("identity-bound journal should decode as JSON");
assert_eq!(identity_bound["version"], serde_json::json!(4));
assert_eq!(identity_bound["backend_identity"], serde_json::json!(identity));
assert_eq!(identity_bound["version_id_exact"], serde_json::json!(true));
assert_eq!(identity_bound["version_state"], serde_json::json!("exact"));
current.user_defined = Arc::new(HashMap::new());
current.transitioned_object.name = "remote/legacy".to_string();
current.transition_version_state = rustfs_filemeta::TransitionVersionState::Unknown;
enqueue_transitioned_delete_cleanup(
store.clone(),
"bucket",
"legacy",
&ObjectOptions {
delete_prefix: true,
..Default::default()
},
Some(&current),
)
.await
.expect("unknown force-delete cleanup should fail closed without a journal");
let legacy_err = match store
.get_object_reader(
".rustfs.sys",
&journal_name("remote/legacy", None, false),
None,
http::HeaderMap::new(),
&ObjectOptions::default(),
)
.await
{
Ok(_) => panic!("unknown remote version state must not persist a delete journal"),
Err(err) => err,
};
assert!(
matches!(
&legacy_err,
StorageError::FileNotFound
| StorageError::ObjectNotFound(_, _)
| StorageError::FileVersionNotFound
| StorageError::VersionNotFound(_, _, _)
| StorageError::VolumeNotFound
),
"unknown remote version state must leave no journal, got {legacy_err:?}"
);
}
async fn put_real_cold_fill_object(store: &Arc<ECStore>, bucket: &str, object: &str, body: &[u8]) -> ObjectInfo {
let mut reader = PutObjReader::from_vec(body.to_vec());
store
rustfs/src/app/runtime_sources.rs
+1 -5
View File
@@ -13,7 +13,6 @@
// limitations under the License.
use crate::app::object_data_cache::ObjectDataCacheAdapter;
use crate::app::storage_api::runtime_sources::ExpiryState;
#[cfg(test)]
use crate::app::storage_api::runtime_sources::TierConfigMgr;
use crate::runtime_sources as root_runtime_sources;
@@ -24,6 +23,7 @@ pub(crate) use crate::runtime_sources::{
use rustfs_s3select_api::{QueryResult, server::dbms::DatabaseManagerSystem};
use s3s::dto::SelectObjectContentInput;
use std::sync::Arc;
#[cfg(test)]
use tokio::sync::RwLock;
pub(crate) fn current_app_context() -> Option<Arc<AppContext>> {
@@ -55,10 +55,6 @@ pub(crate) async fn current_s3select_db(
.await
}
pub(crate) fn current_expiry_state_handle() -> Arc<RwLock<ExpiryState>> {
root_runtime_sources::current_expiry_state_handle().unwrap_or_else(ExpiryState::new)
}
#[cfg(test)]
pub(crate) fn current_tier_config_handle() -> Arc<RwLock<TierConfigMgr>> {
root_runtime_sources::current_tier_config_handle().unwrap_or_else(TierConfigMgr::new)
rustfs/src/app/storage_api.rs
-52
View File
@@ -210,7 +210,6 @@ pub(crate) mod runtime {
}
pub(crate) mod runtime_sources {
pub(crate) type ExpiryState = super::runtime::ExpiryState;
#[cfg(test)]
pub(crate) type TierConfigMgr = super::runtime::TierConfigMgr;
}
@@ -373,57 +372,6 @@ pub(crate) mod bucket {
}
}
pub(crate) use lifecycle_contract as lifecycle;
pub(crate) mod tier_delete_journal {
use std::sync::Arc;
pub(crate) fn record_tier_delete_journal_backend_identity(
je: &mut super::tier_sweeper::Jentry,
metadata: &std::collections::HashMap<String, String>,
) -> std::io::Result<()> {
crate::storage::storage_api::ecstore_bucket::lifecycle::tier_delete_journal::record_tier_delete_journal_backend_identity(je, metadata)
}
pub(crate) async fn persist_tier_delete_journal_entry(
api: Arc<crate::storage::storage_api::ECStore>,
je: &super::tier_sweeper::Jentry,
) -> std::io::Result<()> {
crate::storage::storage_api::ecstore_bucket::lifecycle::tier_delete_journal::persist_tier_delete_journal_entry(
api, je,
)
.await
}
}
pub(crate) mod tier_sweeper {
pub(crate) type Jentry = crate::storage::storage_api::ecstore_bucket::lifecycle::tier_sweeper::Jentry;
pub(crate) fn transitioned_delete_journal_entry(
version_id: Option<uuid::Uuid>,
versioned: bool,
suspended: bool,
transitioned: &super::super::super::storage_contracts::TransitionedObject,
transition_version_state: rustfs_filemeta::TransitionVersionState,
) -> Option<Jentry> {
crate::storage::storage_api::ecstore_bucket::lifecycle::tier_sweeper::transitioned_delete_journal_entry(
version_id,
versioned,
suspended,
transitioned,
transition_version_state,
)
}
pub(crate) fn transitioned_force_delete_journal_entry(
transitioned: &super::super::super::storage_contracts::TransitionedObject,
transition_version_state: rustfs_filemeta::TransitionVersionState,
) -> Option<Jentry> {
crate::storage::storage_api::ecstore_bucket::lifecycle::tier_sweeper::transitioned_force_delete_journal_entry(
transitioned,
transition_version_state,
)
}
}
}
pub(crate) mod metadata {
rustfs/src/runtime_sources.rs
-1
View File
@@ -35,7 +35,6 @@ pub(crate) use context::{
resolve_bucket_monitor_handle as current_bucket_monitor_handle, resolve_buffer_config as current_buffer_config,
resolve_daily_tier_stats as current_daily_tier_stats, resolve_deployment_id as current_deployment_id,
resolve_encryption_service as current_encryption_service, resolve_endpoints_handle as current_endpoints_handle,
resolve_expiry_state_handle as current_expiry_state_handle,
resolve_federated_identity_service as current_federated_identity_service, resolve_iam_handle as current_iam_handle,
resolve_iam_ready as current_iam_ready, resolve_internode_metrics as current_internode_metrics,
resolve_kms_runtime_service_manager as current_kms_runtime_service_manager,