From 71e9e81257a8012fcd9f768cbee03b10a4366e82 Mon Sep 17 00:00:00 2001 From: cxymds Date: Thu, 2 Jul 2026 16:32:33 +0800 Subject: [PATCH] fix: finalize replication resync workers (#4175) --- .../replication/replication_resyncer.rs | 135 ++++++++++++++++-- 1 file changed, 127 insertions(+), 8 deletions(-) diff --git a/crates/ecstore/src/bucket/replication/replication_resyncer.rs b/crates/ecstore/src/bucket/replication/replication_resyncer.rs index 41b2cbc8b..dcee31d61 100644 --- a/crates/ecstore/src/bucket/replication/replication_resyncer.rs +++ b/crates/ecstore/src/bucket/replication/replication_resyncer.rs @@ -74,7 +74,7 @@ use time::OffsetDateTime; use time::format_description::well_known::Rfc3339; use tokio::io::AsyncRead; use tokio::sync::RwLock; -use tokio::task::JoinSet; +use tokio::task::{JoinHandle, JoinSet}; use tokio::time::Duration as TokioDuration; use tokio_util::io::ReaderStream; use tokio_util::sync::CancellationToken; @@ -98,6 +98,40 @@ const RESYNC_TIME_INTERVAL: TokioDuration = TokioDuration::from_secs(60); static WARNED_MONITOR_UNINIT: std::sync::Once = std::sync::Once::new(); +async fn finish_resync_workers( + worker_txs: Vec>, + results_tx: tokio::sync::broadcast::Sender, + futures: Vec>, + abort: bool, +) -> bool { + drop(worker_txs); + drop(results_tx); + + if abort { + for future in &futures { + future.abort(); + } + } + + let mut failed = false; + for result in join_all(futures).await { + if let Err(err) = result + && !(abort && err.is_cancelled()) + { + failed = true; + error!( + event = EVENT_RESYNC_TASK_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + error = %err, + "Replication resync task failed" + ); + } + } + + failed +} + fn has_raw_status(err: &SdkError, status: u16) -> bool { err.raw_response().is_some_and(|r| r.status().as_u16() == status) } @@ -361,6 +395,15 @@ impl ReplicationResyncer { bucket_status.last_update = Some(now); } + async fn target_has_resync_failures(&self, opts: &ResyncOpts) -> bool { + self.status_map + .read() + .await + .get(&opts.bucket) + .and_then(|status| status.targets_map.get(&opts.arn)) + .is_some_and(|status| status.failed_count > 0) + } + pub async fn persist_to_disk(&self, cancel_token: CancellationToken, api: Arc) where S: ReplicationObjectIO, @@ -616,6 +659,7 @@ impl ReplicationResyncer { .await; return; } + drop(tx); let status = { self.status_map @@ -785,12 +829,25 @@ impl ReplicationResyncer { error = %err, "Failed to receive resync object info" ); + let worker_failed = finish_resync_workers(worker_txs, results_tx, futures, false).await; + if worker_failed { + error!( + event = EVENT_RESYNC_TASK_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + bucket = %opts.bucket, + arn = %opts.arn, + reason = "worker_join_failed_after_object_info_error", + "Replication resync worker cleanup observed task failure" + ); + } self.resync_bucket_mark_status(ResyncStatusType::ResyncFailed, opts.clone(), storage.clone()) .await; return; } if cancellation_token.is_cancelled() { + finish_resync_workers(worker_txs, results_tx, futures, true).await; self.resync_bucket_mark_status(ResyncStatusType::ResyncCanceled, opts.clone(), storage.clone()) .await; return; @@ -814,6 +871,7 @@ impl ReplicationResyncer { } if cancellation_token.is_cancelled() { + finish_resync_workers(worker_txs, results_tx, futures, true).await; self.resync_bucket_mark_status(ResyncStatusType::ResyncCanceled, opts.clone(), storage.clone()) .await; return; @@ -832,20 +890,33 @@ impl ReplicationResyncer { error = %err, "Failed to send resync object to worker" ); + let worker_failed = finish_resync_workers(worker_txs, results_tx, futures, false).await; + if worker_failed { + error!( + event = EVENT_RESYNC_TASK_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + bucket = %opts.bucket, + arn = %opts.arn, + reason = "worker_join_failed_after_queue_send_error", + "Replication resync worker cleanup observed task failure" + ); + } self.resync_bucket_mark_status(ResyncStatusType::ResyncFailed, opts.clone(), storage.clone()) .await; return; } } - for worker_tx in worker_txs { - drop(worker_tx); - } + let worker_failed = finish_resync_workers(worker_txs, results_tx, futures, false).await; + let target_failed = self.target_has_resync_failures(&opts).await; + let status = if worker_failed || target_failed { + ResyncStatusType::ResyncFailed + } else { + ResyncStatusType::ResyncCompleted + }; - join_all(futures).await; - - self.resync_bucket_mark_status(ResyncStatusType::ResyncCompleted, opts.clone(), storage.clone()) - .await; + self.resync_bucket_mark_status(status, opts.clone(), storage.clone()).await; } } @@ -3465,6 +3536,54 @@ mod tests { assert!(!token_b.is_cancelled()); } + #[tokio::test] + async fn test_finish_resync_workers_closes_result_collector() { + let (worker_tx, mut worker_rx) = tokio::sync::mpsc::channel::(1); + let (results_tx, mut results_rx) = tokio::sync::broadcast::channel::(1); + let worker = tokio::spawn(async move { while worker_rx.recv().await.is_some() {} }); + let collector = tokio::spawn(async move { while results_rx.recv().await.is_ok() {} }); + + let failed = tokio::time::timeout( + TokioDuration::from_secs(1), + finish_resync_workers(vec![worker_tx], results_tx, vec![worker, collector], false), + ) + .await + .expect("resync worker cleanup should not hang after closing senders"); + + assert!(!failed); + } + + #[tokio::test] + async fn test_finish_resync_workers_reports_join_failure() { + let (results_tx, _results_rx) = tokio::sync::broadcast::channel::(1); + let failed_worker = tokio::spawn(async { + panic!("intentional resync worker failure"); + }); + + let failed = finish_resync_workers(Vec::new(), results_tx, vec![failed_worker], false).await; + + assert!(failed); + } + + #[tokio::test] + async fn test_target_has_resync_failures_reads_accumulated_stats() { + let resyncer = ReplicationResyncer::new().await; + let opts = ResyncOpts { + bucket: "bucket".to_string(), + arn: "arn:replication::dest".to_string(), + resync_id: "run-new".to_string(), + resync_before: None, + }; + let status = TargetReplicationResyncStatus { + failed_count: 1, + ..Default::default() + }; + + resyncer.inc_stats(&status, opts.clone()).await; + + assert!(resyncer.target_has_resync_failures(&opts).await); + } + #[test] fn test_resync_state_accepts_update_only_for_matching_run() { let current = TargetReplicationResyncStatus {