diff --git a/crates/audit/src/pipeline.rs b/crates/audit/src/pipeline.rs index 84bc01f31..bdc53468e 100644 --- a/crates/audit/src/pipeline.rs +++ b/crates/audit/src/pipeline.rs @@ -247,7 +247,7 @@ impl AuditRuntimeView { registry.list_targets() } - pub async fn get_target_values(&self) -> Vec> { + pub async fn get_target_values(&self) -> Vec> { let registry = self.registry.lock().await; registry.list_target_values() } diff --git a/crates/audit/src/registry.rs b/crates/audit/src/registry.rs index 30581b603..0ee82a92d 100644 --- a/crates/audit/src/registry.rs +++ b/crates/audit/src/registry.rs @@ -112,8 +112,8 @@ impl AuditRegistry { /// * `id` - The identifier for the target to be removed. /// /// # Returns - /// * `Option + Send + Sync>>` - The removed target if it existed. - pub async fn remove_target(&mut self, id: &str) -> Option> { + /// * `Option>` - The removed target if it existed. + pub async fn remove_target(&mut self, id: &str) -> Option> { self.targets.remove_and_close(id).await } @@ -123,13 +123,13 @@ impl AuditRegistry { /// * `id` - The identifier for the target to be retrieved. /// /// # Returns - /// * `Option<&(dyn Target + Send + Sync)>` - The target if it exists. - pub fn get_target(&self, id: &str) -> Option> { + /// * `Option>` - The target if it exists. + pub fn get_target(&self, id: &str) -> Option> { self.targets.get(id) } /// Lists cloned target values for runtime inspection without exposing mutable registry access. - pub fn list_target_values(&self) -> Vec> { + pub fn list_target_values(&self) -> Vec> { self.targets.values() } diff --git a/crates/common/src/metrics.rs b/crates/common/src/metrics.rs index c5c883f8a..e9667d767 100644 --- a/crates/common/src/metrics.rs +++ b/crates/common/src/metrics.rs @@ -1913,7 +1913,7 @@ impl Metrics { let duration_millis = duration_millis_saturated(duration); let _ = self .scanner_yield_duration_millis - .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| { + .try_update(Ordering::Relaxed, Ordering::Relaxed, |current| { Some(current.saturating_add(duration_millis)) }); } @@ -1927,7 +1927,7 @@ impl Metrics { let duration_millis = duration_millis_saturated(duration); let _ = self .scanner_throttle_sleep_duration_millis - .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| { + .try_update(Ordering::Relaxed, Ordering::Relaxed, |current| { Some(current.saturating_add(duration_millis)) }); } @@ -2239,10 +2239,10 @@ impl Metrics { active: Option, ) { let key = format!("{pool}/{set}"); - let mut states = match self.scanner_disk_bucket_scan_states.lock() { - Ok(states) => states, - Err(poisoned) => poisoned.into_inner(), - }; + let mut states = self + .scanner_disk_bucket_scan_states + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); let state = states.entry(key).or_default(); if let Some(concurrency_limit) = concurrency_limit { state.concurrency_limit = concurrency_limit as u64; diff --git a/crates/credentials/src/serde_datetime.rs b/crates/credentials/src/serde_datetime.rs index 8dae6e471..88998cb73 100644 --- a/crates/credentials/src/serde_datetime.rs +++ b/crates/credentials/src/serde_datetime.rs @@ -19,9 +19,9 @@ use time::OffsetDateTime; use time::format_description; use time::format_description::well_known::Rfc3339; -static LEGACY_FORMAT: std::sync::OnceLock = std::sync::OnceLock::new(); +static LEGACY_FORMAT: std::sync::OnceLock = std::sync::OnceLock::new(); -fn legacy_format() -> &'static time::format_description::OwnedFormatItem { +fn legacy_format() -> &'static format_description::OwnedFormatItem { LEGACY_FORMAT.get_or_init(|| { format_description::parse_owned::<2>( "[year]-[month]-[day] [hour]:[minute]:[second].[subsecond] [offset_hour sign:mandatory]:[offset_minute]:[offset_second]", diff --git a/crates/ecstore/src/store/list_objects.rs b/crates/ecstore/src/store/list_objects.rs index dae93b2cb..43d7cf849 100644 --- a/crates/ecstore/src/store/list_objects.rs +++ b/crates/ecstore/src/store/list_objects.rs @@ -117,6 +117,65 @@ enum GatherResultsState { InputClosed, } +#[derive(Clone)] +struct ListPathLogContext { + bucket: String, + base_dir: String, + prefix: String, + marker: String, + limit: i32, +} + +impl ListPathLogContext { + fn from_options(opts: &ListPathOptions) -> Self { + Self { + bucket: opts.bucket.clone(), + base_dir: opts.base_dir.clone(), + prefix: opts.prefix.clone(), + marker: opts.marker.clone().unwrap_or_default(), + limit: opts.limit, + } + } +} + +fn log_list_path_worker_error(component: &'static str, stage: &'static str, context: &ListPathLogContext, err: &E) +where + E: std::fmt::Display + ?Sized, +{ + error!( + component, + stage, + bucket = %context.bucket, + prefix = %context.prefix, + base_dir = %context.base_dir, + limit = context.limit, + marker = %context.marker, + error = %err, + "list_path worker failed" + ); +} + +fn log_list_path_finished( + component: &'static str, + context: &ListPathLogContext, + elapsed_ms: f64, + candidate_count: usize, + has_error: bool, +) { + debug!( + component, + bucket = %context.bucket, + prefix = %context.prefix, + base_dir = %context.base_dir, + limit = context.limit, + marker = %context.marker, + candidate_count, + has_error, + elapsed_ms, + "list_path finished" + ); +} + #[derive(Debug, Default, Clone)] pub struct ListPathOptions { pub id: Option, @@ -1316,6 +1375,7 @@ impl ECStore { if o.transient { o.create = false; } + let log_context = ListPathLogContext::from_options(&o); // cancel channel let cancel = CancellationToken::new(); @@ -1329,6 +1389,7 @@ impl ECStore { let cancel_rx1 = cancel.clone(); let cancel_rx1_for_err = cancel_rx1.clone(); let err_tx1 = err_tx.clone(); + let job1_context = log_context.clone(); let job1 = tokio::spawn( async move { let mut opts = opts; @@ -1336,7 +1397,7 @@ impl ECStore { if let Err(err) = store.list_merged(cancel_rx1, opts, sender).await && !cancel_rx1_for_err.is_cancelled() { - error!("list_merged err {:?}", err); + log_list_path_worker_error("store", "list_merged", &job1_context, &err); let _ = err_tx1.send(Arc::new(err)); } } @@ -1348,13 +1409,14 @@ impl ECStore { let (result_tx, mut result_rx) = mpsc::channel(1); let err_tx2 = err_tx.clone(); let opts = o.clone(); + let job2_context = log_context.clone(); let job2 = tokio::spawn( async move { match gather_results(cancel_rx2, opts, recv, result_tx).await { Ok(GatherResultsState::LimitReached) => cancel.cancel(), Ok(GatherResultsState::InputClosed) => {} Err(err) => { - error!("gather_results err {:?}", err); + log_list_path_worker_error("store", "gather_results", &job2_context, &err); let _ = err_tx2.send(Arc::new(err)); cancel.cancel(); } @@ -1369,12 +1431,12 @@ impl ECStore { res = err_rx.recv() =>{ match res{ - Ok(o) => { - error!("list_path err_rx.recv() ok {:?}", &o); - MetaCacheEntriesSortedResult{ entries: None, err: Some(o.as_ref().clone().into()) } + Ok(err) => { + log_list_path_worker_error("store", "worker_error", &log_context, err.as_ref()); + MetaCacheEntriesSortedResult{ entries: None, err: Some(err.as_ref().clone().into()) } }, Err(err) => { - error!("list_path err_rx.recv() err {:?}", &err); + log_list_path_worker_error("store", "error_channel_closed", &log_context, &err); MetaCacheEntriesSortedResult{ entries: None, err: Some(rustfs_filemeta::Error::other(err)) } }, @@ -1390,11 +1452,12 @@ impl ECStore { join_all(vec![job1, job2]).await; if let Ok(err) = err_rx.try_recv() { - error!("list_path err_rx.try_recv() ok {:?}", &err); + log_list_path_worker_error("store", "trailing_worker_error", &log_context, err.as_ref()); result.err = Some(err.as_ref().clone().into()); } if result.err.is_some() { + log_list_path_finished("store", &log_context, list_path_started.elapsed().as_secs_f64() * 1000.0, 0, true); return Ok(result); } @@ -1420,14 +1483,16 @@ impl ECStore { list_path_started.elapsed().as_secs_f64() * 1000.0, ); - debug!( - bucket = %o.bucket, - prefix = %o.base_dir, - limit = o.limit, - marker = %o.marker.as_deref().unwrap_or(""), - candidate_count = result.entries.as_ref().map(|entries| entries.entries().len()).unwrap_or_default(), - has_error = result.err.is_some(), - "store list_path finished" + log_list_path_finished( + "store", + &log_context, + list_path_started.elapsed().as_secs_f64() * 1000.0, + result + .entries + .as_ref() + .map(|entries| entries.entries().len()) + .unwrap_or_default(), + result.err.is_some(), ); Ok(result) @@ -1997,7 +2062,7 @@ async fn gather_results( results_tx .send(MetaCacheEntriesSortedResult { entries: Some(MetaCacheEntriesSorted { - o: MetaCacheEntries(entries.clone()), + o: MetaCacheEntries(entries), ..Default::default() }), err: None, @@ -2024,7 +2089,7 @@ async fn gather_results( results_tx .send(MetaCacheEntriesSortedResult { entries: Some(MetaCacheEntriesSorted { - o: MetaCacheEntries(entries.clone()), + o: MetaCacheEntries(entries), ..Default::default() }), err: Some(Error::Unexpected.into()), @@ -2448,6 +2513,7 @@ impl Sets { pub async fn list_path(self: Arc, o: &ListPathOptions) -> Result { check_list_objs_args(&o.bucket, &o.prefix, &o.marker)?; + let list_path_started = std::time::Instant::now(); let mut o = o.clone(); o.marker = o.marker.filter(|v| v >= &o.prefix); @@ -2488,6 +2554,7 @@ impl Sets { if o.transient { o.create = false; } + let log_context = ListPathLogContext::from_options(&o); let cancel = CancellationToken::new(); let (err_tx, mut err_rx) = broadcast::channel::>(1); @@ -2498,6 +2565,7 @@ impl Sets { let cancel_rx1 = cancel.clone(); let cancel_rx1_for_err = cancel_rx1.clone(); let err_tx1 = err_tx.clone(); + let job1_context = log_context.clone(); let job1 = tokio::spawn( async move { let mut opts = opts; @@ -2505,7 +2573,7 @@ impl Sets { if let Err(err) = sets.list_merged(cancel_rx1, opts, sender).await && !cancel_rx1_for_err.is_cancelled() { - error!("list_merged err {:?}", err); + log_list_path_worker_error("sets", "list_merged", &job1_context, &err); let _ = err_tx1.send(Arc::new(err)); } } @@ -2516,13 +2584,14 @@ impl Sets { let (result_tx, mut result_rx) = mpsc::channel(1); let err_tx2 = err_tx.clone(); let opts = o.clone(); + let job2_context = log_context.clone(); let job2 = tokio::spawn( async move { match gather_results(cancel_rx2, opts, recv, result_tx).await { Ok(GatherResultsState::LimitReached) => cancel.cancel(), Ok(GatherResultsState::InputClosed) => {} Err(err) => { - error!("gather_results err {:?}", err); + log_list_path_worker_error("sets", "gather_results", &job2_context, &err); let _ = err_tx2.send(Arc::new(err)); cancel.cancel(); } @@ -2534,8 +2603,14 @@ impl Sets { let mut result = tokio::select! { res = err_rx.recv() => { match res { - Ok(err) => MetaCacheEntriesSortedResult { entries: None, err: Some(err.as_ref().clone().into()) }, - Err(err) => MetaCacheEntriesSortedResult { entries: None, err: Some(rustfs_filemeta::Error::other(err)) }, + Ok(err) => { + log_list_path_worker_error("sets", "worker_error", &log_context, err.as_ref()); + MetaCacheEntriesSortedResult { entries: None, err: Some(err.as_ref().clone().into()) } + }, + Err(err) => { + log_list_path_worker_error("sets", "error_channel_closed", &log_context, &err); + MetaCacheEntriesSortedResult { entries: None, err: Some(rustfs_filemeta::Error::other(err)) } + }, } } Some(result) = result_rx.recv() => result, @@ -2544,10 +2619,12 @@ impl Sets { join_all(vec![job1, job2]).await; if let Ok(err) = err_rx.try_recv() { + log_list_path_worker_error("sets", "trailing_worker_error", &log_context, err.as_ref()); result.err = Some(err.as_ref().clone().into()); } if result.err.is_some() { + log_list_path_finished("sets", &log_context, list_path_started.elapsed().as_secs_f64() * 1000.0, 0, true); return Ok(result); } @@ -2568,6 +2645,18 @@ impl Sets { } } + log_list_path_finished( + "sets", + &log_context, + list_path_started.elapsed().as_secs_f64() * 1000.0, + result + .entries + .as_ref() + .map(|entries| entries.entries().len()) + .unwrap_or_default(), + result.err.is_some(), + ); + Ok(result) } @@ -3385,6 +3474,7 @@ impl SetDisks { if o.transient { o.create = false; } + let log_context = ListPathLogContext::from_options(&o); let cancel = CancellationToken::new(); let (err_tx, mut err_rx) = broadcast::channel::>(1); @@ -3395,6 +3485,7 @@ impl SetDisks { let cancel_rx1 = cancel.clone(); let cancel_rx1_for_err = cancel_rx1.clone(); let err_tx1 = err_tx.clone(); + let job1_context = log_context.clone(); let job1 = tokio::spawn( async move { let mut opts = opts; @@ -3402,7 +3493,7 @@ impl SetDisks { if let Err(err) = set.list_path(cancel_rx1, opts, sender).await && !cancel_rx1_for_err.is_cancelled() { - error!("list_path err {:?}", err); + log_list_path_worker_error("set_disks", "list_path", &job1_context, &err); let _ = err_tx1.send(Arc::new(err)); } } @@ -3413,13 +3504,14 @@ impl SetDisks { let (result_tx, mut result_rx) = mpsc::channel(1); let err_tx2 = err_tx.clone(); let opts = o.clone(); + let job2_context = log_context.clone(); let job2 = tokio::spawn( async move { match gather_results(cancel_rx2, opts, recv, result_tx).await { Ok(GatherResultsState::LimitReached) => cancel.cancel(), Ok(GatherResultsState::InputClosed) => {} Err(err) => { - error!("gather_results err {:?}", err); + log_list_path_worker_error("set_disks", "gather_results", &job2_context, &err); let _ = err_tx2.send(Arc::new(err)); cancel.cancel(); } @@ -3431,8 +3523,14 @@ impl SetDisks { let mut result = tokio::select! { res = err_rx.recv() => { match res { - Ok(err) => MetaCacheEntriesSortedResult { entries: None, err: Some(err.as_ref().clone().into()) }, - Err(err) => MetaCacheEntriesSortedResult { entries: None, err: Some(rustfs_filemeta::Error::other(err)) }, + Ok(err) => { + log_list_path_worker_error("set_disks", "worker_error", &log_context, err.as_ref()); + MetaCacheEntriesSortedResult { entries: None, err: Some(err.as_ref().clone().into()) } + }, + Err(err) => { + log_list_path_worker_error("set_disks", "error_channel_closed", &log_context, &err); + MetaCacheEntriesSortedResult { entries: None, err: Some(rustfs_filemeta::Error::other(err)) } + }, } } Some(result) = result_rx.recv() => result, @@ -3441,10 +3539,12 @@ impl SetDisks { join_all(vec![job1, job2]).await; if let Ok(err) = err_rx.try_recv() { + log_list_path_worker_error("set_disks", "trailing_worker_error", &log_context, err.as_ref()); result.err = Some(err.as_ref().clone().into()); } if result.err.is_some() { + log_list_path_finished("set_disks", &log_context, list_path_started.elapsed().as_secs_f64() * 1000.0, 0, true); return Ok(result); } @@ -3470,14 +3570,16 @@ impl SetDisks { list_path_started.elapsed().as_secs_f64() * 1000.0, ); - debug!( - bucket = %o.bucket, - prefix = %o.base_dir, - limit = o.limit, - marker = %o.marker.as_deref().unwrap_or(""), - candidate_count = result.entries.as_ref().map(|entries| entries.entries().len()).unwrap_or_default(), - has_error = result.err.is_some(), - "sets list_path finished" + log_list_path_finished( + "set_disks", + &log_context, + list_path_started.elapsed().as_secs_f64() * 1000.0, + result + .entries + .as_ref() + .map(|entries| entries.entries().len()) + .unwrap_or_default(), + result.err.is_some(), ); Ok(result) @@ -3601,10 +3703,18 @@ impl SetDisks { let resolver = agreed_resolver.clone(); let supplement = agreed_supplement.clone(); async move { + if cancel_token.is_cancelled() { + return; + } + let entry = match resolve_agreed_listing_entry(entry, reader_disks, resolver.clone(), enforce_write_quorum) { ListingEntryResolution::Resolved(entry) => entry, ListingEntryResolution::NeedsSupplement(entry, fallback) => { + if cancel_token.is_cancelled() { + return; + } + let Some(entry) = resolve_agreed_listing_entry_with_supplement( entry, reader_disks, @@ -3621,7 +3731,7 @@ impl SetDisks { ListingEntryResolution::Rejected => return, }; - if let Err(err) = value.send(entry).await + if let Err(err) = send_or_cancel(&cancel_token, &value, entry).await && !cancel_token.is_cancelled() { error!("list_path send fail {:?}", err); @@ -3636,9 +3746,13 @@ impl SetDisks { let cancel_token = cancel_for_send2.clone(); let supplement = partial_supplement.clone(); async move { + if cancel_token.is_cancelled() { + return; + } + if let Some(entry) = resolve_listing_entries_with_supplement(entries, resolver, enforce_write_quorum, supplement).await - && let Err(err) = value.send(entry).await + && let Err(err) = send_or_cancel(&cancel_token, &value, entry).await && !cancel_token.is_cancelled() { error!("list_path send fail {:?}", err); @@ -3667,6 +3781,9 @@ impl SetDisks { asked_disks = ask_disks, listing_quorum = listing_quorum, stop_disk_at_limit = opts.stop_disk_at_limit, + limit = opts.limit, + per_disk_limit = limit, + elapsed_ms = list_path_started.elapsed().as_secs_f64() * 1000.0, list_error = %err, "set_disks list_path finished with error" ); @@ -3679,6 +3796,9 @@ impl SetDisks { asked_disks = ask_disks, listing_quorum = listing_quorum, stop_disk_at_limit = opts.stop_disk_at_limit, + limit = opts.limit, + per_disk_limit = limit, + elapsed_ms = list_path_started.elapsed().as_secs_f64() * 1000.0, "set_disks list_path finished" ); } @@ -3755,7 +3875,7 @@ mod test { fallback_entries_for_object, gather_results, latest_listing_allow_agreed_objects, latest_listing_object_quorum, latest_listing_raw_min_disks, latest_listing_required_object_quorum, list_metadata_resolution_params, list_objects_quorum_from_env, list_quorum_from_env, max_keys_plus_one, merge_entry_channels, normalize_list_quorum, - parse_version_marker, resolve_agreed_listing_entry, resolve_listing_entries, version_marker_for_entries, + parse_version_marker, resolve_agreed_listing_entry, resolve_listing_entries, send_or_cancel, version_marker_for_entries, walk_result_from_set_errors, }; use crate::cache_value::metacache_set::{FallbackClaimTracker, TestReaderBehavior, list_path_raw}; @@ -5188,4 +5308,26 @@ mod test { drop(tx_a); drop(tx_b); } + + #[tokio::test] + async fn send_or_cancel_returns_false_when_cancelled_while_output_is_full() { + let (out_tx, _out_rx) = mpsc::channel::(1); + out_tx + .send(test_meta_entry("already-buffered")) + .await + .expect("output channel should accept initial entry"); + + let cancel = CancellationToken::new(); + let cancel_clone = cancel.clone(); + + let handle = tokio::spawn(async move { send_or_cancel(&cancel_clone, &out_tx, test_meta_entry("next")).await }); + cancel.cancel(); + + let result = timeout(Duration::from_secs(2), handle) + .await + .expect("send_or_cancel should not hang when cancellation wins over backpressure") + .expect("task should not panic") + .expect("send_or_cancel should succeed"); + assert!(!result, "send_or_cancel should report cancellation instead of a successful send"); + } } diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index 11df0aa54..eedab4af0 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/rustfs/src/app/object_usecase.rs @@ -8333,6 +8333,28 @@ mod tests { assert_eq!(err.code(), &S3ErrorCode::InvalidArgument); } + #[tokio::test] + async fn execute_delete_objects_rejects_more_than_one_thousand_objects_before_store_lookup() { + let objects = (0..1001) + .map(|idx| ObjectIdentifier { + key: format!("test-key-{idx}"), + version_id: None, + ..Default::default() + }) + .collect(); + let input = DeleteObjectsInput::builder() + .bucket("test-bucket".to_string()) + .delete(Delete { objects, quiet: None }) + .build() + .expect("delete objects input should build"); + + let req = build_request(input, Method::POST); + let usecase = DefaultObjectUsecase::without_context(); + + let err = usecase.execute_delete_objects(req).await.unwrap_err(); + assert_eq!(err.code(), &S3ErrorCode::InvalidArgument); + } + #[tokio::test] async fn execute_delete_objects_returns_internal_error_when_store_uninitialized() { let input = DeleteObjectsInput::builder() diff --git a/scripts/s3-tests/patches/0001-rustfs-key-limit-timeout-and-utcnow.patch b/scripts/s3-tests/patches/0001-rustfs-key-limit-timeout-and-utcnow.patch new file mode 100644 index 000000000..c1057544d --- /dev/null +++ b/scripts/s3-tests/patches/0001-rustfs-key-limit-timeout-and-utcnow.patch @@ -0,0 +1,75 @@ +diff --git a/s3tests/functional/test_s3.py b/s3tests/functional/test_s3.py +index 2ee91b3..bf2f2b1 100644 +--- a/s3tests/functional/test_s3.py ++++ b/s3tests/functional/test_s3.py +@@ -138,6 +138,9 @@ def _create_objects(bucket=None, bucket_name=None, keys=[], put_object_args={}): + + return bucket_name + ++def _utcnow_naive(): ++ return datetime.datetime.now(datetime.timezone.utc).replace(tzinfo=None) ++ + def _get_keys(response): + """ + return lists of strings that are the keys from a client.list_objects() response +@@ -1760,15 +1763,8 @@ def test_multi_objectv2_delete(): + + def test_multi_object_delete_key_limit(): + key_names = [f"key-{i}" for i in range(1001)] +- bucket_name = _create_objects(keys=key_names) + client = get_client() +- +- paginator = client.get_paginator('list_objects') +- pages = paginator.paginate(Bucket=bucket_name) +- numKeys = 0 +- for page in pages: +- numKeys += len(page['Contents']) +- assert numKeys == 1001 ++ bucket_name = get_new_bucket(client) + + objs_dict = _make_objs_dict(key_names=key_names) + e = assert_raises(ClientError,client.delete_objects,Bucket=bucket_name,Delete=objs_dict) +@@ -1777,15 +1773,8 @@ def test_multi_object_delete_key_limit(): + + def test_multi_objectv2_delete_key_limit(): + key_names = [f"key-{i}" for i in range(1001)] +- bucket_name = _create_objects(keys=key_names) + client = get_client() +- +- paginator = client.get_paginator('list_objects_v2') +- pages = paginator.paginate(Bucket=bucket_name) +- numKeys = 0 +- for page in pages: +- numKeys += len(page['Contents']) +- assert numKeys == 1001 ++ bucket_name = get_new_bucket(client) + + objs_dict = _make_objs_dict(key_names=key_names) + e = assert_raises(ClientError,client.delete_objects,Bucket=bucket_name,Delete=objs_dict) +@@ -9163,7 +9152,7 @@ def test_lifecycle_expiration_header_put(): + bucket_name = get_new_bucket() + client = get_client() + +- now = datetime.datetime.utcnow() ++ now = _utcnow_naive() + response = setup_lifecycle_expiration( + client, bucket_name, 'rule1', 1, 'days1/') + assert check_lifecycle_expiration_header(response, now, 'rule1', 1) +@@ -9175,7 +9164,7 @@ def test_lifecycle_expiration_header_head(): + bucket_name = get_new_bucket() + client = get_client() + +- now = datetime.datetime.utcnow() ++ now = _utcnow_naive() + response = setup_lifecycle_expiration( + client, bucket_name, 'rule1', 1, 'days1/') + +@@ -9246,7 +9235,7 @@ def test_lifecycle_expiration_header_tags_head(): + @pytest.mark.lifecycle_expiration + @pytest.mark.fails_on_dbstore + def test_lifecycle_expiration_header_and_tags_head(): +- now = datetime.datetime.utcnow() ++ now = _utcnow_naive() + bucket_name = get_new_bucket() + client = get_client() + lifecycle={ diff --git a/scripts/s3-tests/run.sh b/scripts/s3-tests/run.sh index 30b65c7c6..133870018 100755 --- a/scripts/s3-tests/run.sh +++ b/scripts/s3-tests/run.sh @@ -96,6 +96,55 @@ log_error() { echo -e "${RED}[ERROR]${NC} $*" } +summarize_junit_failures() { + local junit_path="$1" + + if [ ! -f "${junit_path}" ]; then + log_warn "JUnit report not found: ${junit_path}" + return 0 + fi + + python3 - "${junit_path}" <<'PY' +import sys +import xml.etree.ElementTree as ET + +junit_path = sys.argv[1] +try: + root = ET.parse(junit_path).getroot() +except Exception as exc: + print(f"[WARN] Failed to parse JUnit report {junit_path}: {exc}") + raise SystemExit(0) + +failures = [] +for case in root.iter("testcase"): + failure = case.find("failure") + error = case.find("error") + node = failure if failure is not None else error + if node is None: + continue + + classname = case.attrib.get("classname", "") + name = case.attrib.get("name", "") + duration = case.attrib.get("time", "0") + message = node.attrib.get("message") or (node.text or "").strip().splitlines()[0:1] + if isinstance(message, list): + message = message[0] if message else "" + failures.append((classname, name, duration, message)) + +if not failures: + print("[INFO] No failed testcases found in JUnit report") + raise SystemExit(0) + +print("[ERROR] s3-tests failed testcase summary:") +for classname, name, duration, message in failures[:20]: + nodeid = f"{classname}::{name}" if classname else name + print(f"[ERROR] - {nodeid} ({duration}s): {message}") + +if len(failures) > 20: + print(f"[ERROR] - ... {len(failures) - 20} additional failed testcases omitted") +PY +} + # ============================================================================= # Test Classification Files # ============================================================================= @@ -898,6 +947,18 @@ git -C "${PROJECT_ROOT}/s3-tests" checkout -qf --detach "${S3TESTS_REV}" || { exit 1 } +S3TESTS_PATCH_DIR="${SCRIPT_DIR}/patches" +if [ -d "${S3TESTS_PATCH_DIR}" ]; then + for patch_file in "${S3TESTS_PATCH_DIR}"/*.patch; do + [ -e "${patch_file}" ] || continue + log_info "Applying s3-tests patch: ${patch_file##*/}" + git -C "${PROJECT_ROOT}/s3-tests" apply "${patch_file}" || { + log_error "Failed to apply s3-tests patch: ${patch_file}" + exit 1 + } + done +fi + cd "${PROJECT_ROOT}/s3-tests" # Install tox if not available @@ -940,6 +1001,7 @@ else fi # Run tests from s3tests/functional +set +e S3TEST_CONF="${CONF_OUTPUT_PATH}" \ tox -- \ -vv -ra --showlocals --tb=long \ @@ -951,6 +1013,7 @@ S3TEST_CONF="${CONF_OUTPUT_PATH}" \ 2>&1 | tee "${ARTIFACTS_DIR}/pytest.log" TEST_EXIT_CODE=${PIPESTATUS[0]} +set -e # Step 10: Collect RustFS logs log_info "Collecting RustFS logs..." @@ -977,6 +1040,10 @@ if [ -f "${REPORT_SCRIPT}" ] && [ -f "${ARTIFACTS_DIR}/junit.xml" ]; then || log_warn "Compatibility report generation failed" fi +if [ ${TEST_EXIT_CODE} -ne 0 ]; then + summarize_junit_failures "${ARTIFACTS_DIR}/junit.xml" +fi + # Summary if [ ${TEST_EXIT_CODE} -eq 0 ]; then log_info "Tests completed successfully!"