From 3229ce05e3cfdffe0e74dac0ac0befcae1ab9fc1 Mon Sep 17 00:00:00 2001 From: overtrue Date: Sat, 22 Aug 2026 08:26:53 +0800 Subject: [PATCH] fix(ecstore): repair durable ILM receipt recovery --- .../src/bucket/lifecycle/durable_namespace.rs | 28 +++++++++++++--- .../bucket/lifecycle/tier_delete_journal.rs | 20 ++++++++---- crates/ecstore/src/core/pools.rs | 32 +++++++++++++++---- 3 files changed, 63 insertions(+), 17 deletions(-) diff --git a/crates/ecstore/src/bucket/lifecycle/durable_namespace.rs b/crates/ecstore/src/bucket/lifecycle/durable_namespace.rs index 81a8f38df..ad14c0fff 100644 --- a/crates/ecstore/src/bucket/lifecycle/durable_namespace.rs +++ b/crates/ecstore/src/bucket/lifecycle/durable_namespace.rs @@ -169,7 +169,7 @@ pub(crate) enum DurableIlmRecordCheckpoint { ManualTransitionJob { content_sha256: String, identity_sha256: String, - updated_at_unix_nanos: i128, + updated_at_unix_nanos: i64, state: manual_transition_job::ManualTransitionJobState, scan_completed: bool, cancel_requested: bool, @@ -181,7 +181,7 @@ pub(crate) enum DurableIlmRecordCheckpoint { ManualTransitionScope { content_sha256: String, identity_sha256: String, - updated_at_unix_nanos: i128, + updated_at_unix_nanos: i64, }, ManualTransitionTask { content_sha256: String, @@ -817,13 +817,15 @@ pub(crate) fn validate_durable_ilm_record(path: &str, data: &[u8]) -> Result Result, je: &Jen ) .await?; } - record_tier_delete_journal_decommission_terminal(&api, je).await?; - remove_tier_delete_journal_entry(api, je).await -} - -async fn record_tier_delete_journal_decommission_terminal(api: &Arc, je: &Jentry) -> std::io::Result<()> { let path = tier_delete_journal_object_name(je); let data = encode_tier_delete_journal_entry(je).map_err(std::io::Error::other)?; - api.record_durable_ilm_decommission_terminal(&path, &data) + let target_pool_indices = api + .record_durable_ilm_decommission_terminal_target_pools(&path, &data) .await - .map_err(std::io::Error::other) + .map_err(std::io::Error::other)?; + if !target_pool_indices.is_empty() { + for target_pool_idx in target_pool_indices { + match config_boundary::delete_config(api.pools[target_pool_idx].clone(), &path).await { + Ok(()) | Err(Error::ConfigNotFound) => {} + Err(err) => return Err(std::io::Error::other(err)), + } + } + return Ok(()); + } + remove_tier_delete_journal_entry(api, je).await } fn object_info_references_tier_delete(info: &ObjectInfo, je: &Jentry) -> std::io::Result { diff --git a/crates/ecstore/src/core/pools.rs b/crates/ecstore/src/core/pools.rs index cc6cda6e6..1a86a3356 100644 --- a/crates/ecstore/src/core/pools.rs +++ b/crates/ecstore/src/core/pools.rs @@ -5478,7 +5478,7 @@ impl ECStore { } } - async fn advance_durable_ilm_decommission_receipts(&self, path: &str, data: &[u8], terminal: bool) -> Result<()> { + async fn advance_durable_ilm_decommission_receipts(&self, path: &str, data: &[u8], terminal: bool) -> Result> { let active_runs = { let pool_meta = self.pool_meta.read().await; pool_meta @@ -5495,20 +5495,25 @@ impl ECStore { .collect::>() }; if active_runs.is_empty() { - return Ok(()); + return Ok(Vec::new()); } let stage = if terminal { "terminal" } else { "progress" }; let record = validate_durable_ilm_record(path, data) .map_err(|err| Error::other(format!("{stage} durable ILM record is invalid at path `{path}`: {err}")))?; + let mut terminal_target_pool_indices = Vec::new(); for (source_pool_idx, run_token) in active_runs { let receipt_path = decommission_durable_ilm_receipt_path(&run_token, path, record.id_kind, &record.id); let mut receipt_found = false; for pool_idx in 0..self.pools.len() { if pool_idx != source_pool_idx { - receipt_found |= self + let found = self .advance_durable_ilm_decommission_receipt(pool_idx, &receipt_path, &record, terminal) .await?; + receipt_found |= found; + if terminal && found && !terminal_target_pool_indices.contains(&pool_idx) { + terminal_target_pool_indices.push(pool_idx); + } } } if terminal && !receipt_found { @@ -5518,14 +5523,27 @@ impl ECStore { ))); } } - Ok(()) + Ok(terminal_target_pool_indices) } pub(crate) async fn record_durable_ilm_decommission_progress(&self, path: &str, data: &[u8]) -> Result<()> { - self.advance_durable_ilm_decommission_receipts(path, data, false).await + self.advance_durable_ilm_decommission_receipts(path, data, false) + .await + .map(|_| ()) } pub(crate) async fn record_durable_ilm_decommission_terminal(&self, path: &str, data: &[u8]) -> Result<()> { + self.record_durable_ilm_decommission_terminal_target_pools(path, data) + .await + .map(|_| ()) + } + + /// Record terminal proof and return its non-source receipt pools for targeted cleanup. + pub(crate) async fn record_durable_ilm_decommission_terminal_target_pools( + &self, + path: &str, + data: &[u8], + ) -> Result> { self.advance_durable_ilm_decommission_receipts(path, data, true).await } @@ -7021,6 +7039,7 @@ mod pools_tests { let job_bytes = job.encode().expect("large manual job should remain within its record limit"); assert!(job_bytes.len() > DECOMMISSION_DURABLE_ILM_RECEIPT_MAX_SIZE); let record = validate_durable_ilm_record(&path, &job_bytes).expect("large manual job should validate"); + let expected_checkpoint = record.checkpoint.clone(); let mut receipt = DecommissionDurableIlmReceipt::new(&path, &record); receipt.terminal_checkpoint = Some(record.checkpoint); @@ -7029,7 +7048,8 @@ mod pools_tests { assert!(encoded.len() <= DECOMMISSION_DURABLE_ILM_RECEIPT_MAX_SIZE); assert_eq!(decoded.source_path, path); - assert!(decoded.terminal_checkpoint.is_some()); + assert_eq!(decoded.checkpoint, expected_checkpoint); + assert_eq!(decoded.terminal_checkpoint, Some(expected_checkpoint)); } #[test]