From a9f01dbbdb56f9f59123c222dd642e2a004eee55 Mon Sep 17 00:00:00 2001 From: houseme Date: Sun, 6 Sep 2026 01:26:08 +0800 Subject: [PATCH] fix(ecstore): restore odm source contract tests (#7215) * fix(ecstore): restore odm source contract tests Co-Authored-By: heihutu Co-Authored-By: zhi22915 * test(ci): initialize replication evidence in chain test Run the replication workflow's evidence initialization before the chain handoff self-test executes the suite step. This keeps the test model aligned with the workflow-provided LOG_FILE and TMPDIR values. Co-Authored-By: heihutu Co-Authored-By: zhi22915 * fix(odm): distinguish missing GCS buckets from object misses (#7221) --------- Co-authored-by: zhi22915 Co-authored-by: Zhengchao An --- .../src/bucket/on_demand_migration/gcs.rs | 62 ++++++++++++++++++- .../on_demand_migration/list_through.rs | 16 +++-- .../bucket/on_demand_migration/native_http.rs | 33 +++++++--- .../on_demand_migration/source_client.rs | 5 +- scripts/test_security_workflow.py | 2 + 5 files changed, 103 insertions(+), 15 deletions(-) diff --git a/crates/ecstore/src/bucket/on_demand_migration/gcs.rs b/crates/ecstore/src/bucket/on_demand_migration/gcs.rs index 707e1cc82..874ddb0e3 100644 --- a/crates/ecstore/src/bucket/on_demand_migration/gcs.rs +++ b/crates/ecstore/src/bucket/on_demand_migration/gcs.rs @@ -124,6 +124,18 @@ impl GcsNativeSourceBackend { Ok(request) } + async fn send_object(&self, request: reqwest::Request) -> Result { + match self.http.send_object(request, NO_ERROR_CODE_HEADER).await { + Err(SourceError::NotFound) => { + // An XML object URL also returns 404 when its bucket is gone. + // Reuse the read-only listing probe before caching a key miss. + self.probe().await?; + Err(SourceError::NotFound) + } + result => result, + } + } + /// Shared mapping for the XML API's HEAD and GET responses. fn head_from_response(headers: &HeaderMap) -> Result { if header(headers, "x-goog-encryption-key-sha256").is_some() { @@ -164,7 +176,7 @@ impl GcsNativeSourceBackend { impl SourceBackend for GcsNativeSourceBackend { async fn head(&self, key: &str) -> Result { let request = self.request(Method::HEAD, self.object_url(key)?, HeaderMap::new()).await?; - let response = self.http.send(request, NO_ERROR_CODE_HEADER).await?; + let response = self.send_object(request).await?; Self::head_from_response(response.headers()) } @@ -177,7 +189,7 @@ impl SourceBackend for GcsNativeSourceBackend { ); } let request = self.request(Method::GET, self.object_url(key)?, headers).await?; - let response = self.http.send(request, NO_ERROR_CODE_HEADER).await?; + let response = self.send_object(request).await?; let head = Self::head_from_response(response.headers())?; let content_range = header(response.headers(), "content-range").map(str::to_string); Ok(SourceGet { @@ -488,6 +500,7 @@ mod tests { // request; the probe is the next one on the wire. ScriptedResponse::new(200, Vec::new(), "{}".to_string()), ScriptedResponse::new(404, Vec::new(), String::new()), + ScriptedResponse::new(200, Vec::new(), "{}".to_string()), ScriptedResponse::new(403, Vec::new(), String::new()), ]) .await; @@ -503,4 +516,49 @@ mod tests { ) .await; } + + #[tokio::test] + async fn listing_404_is_not_an_object_not_found() { + let (endpoint, _) = scripted_server(vec![ScriptedResponse::new(404, Vec::new(), String::new())]).await; + let err = backend(&endpoint) + .list(&SourceListRequest { + max_keys: 1, + ..Default::default() + }) + .await + .expect_err("a failed bucket listing is not a per-object miss"); + assert_eq!(err.class_label(), "other", "{err:?}"); + } + + #[tokio::test] + async fn object_404_requires_a_readable_source_bucket() { + for method in [Method::HEAD, Method::GET] { + for (probe_status, expected_class) in [ + (200, "not_found"), + (404, "other"), + (403, "access_denied"), + (503, "throttled"), + (500, "server_error"), + ] { + let (endpoint, recorded) = scripted_server(vec![ + ScriptedResponse::new(404, Vec::new(), String::new()), + ScriptedResponse::new(probe_status, Vec::new(), "{}".to_string()), + ]) + .await; + let backend = backend(&endpoint); + let result = if method == Method::HEAD { + backend.head("missing").await.map(|_| ()) + } else { + backend.get("missing", None).await.map(|_| ()) + }; + let error = result.expect_err("the object 404 must remain an error"); + assert_eq!(error.class_label(), expected_class, "{method} with probe HTTP {probe_status}: {error:?}"); + let recorded = recorded.lock().expect("recorder lock"); + assert_eq!(recorded.len(), 2, "one bounded read-only probe per ambiguous object miss"); + assert_eq!(recorded[0].method, method.as_str()); + assert_eq!(recorded[1].method, "GET"); + assert_eq!(recorded[1].target, "/storage/v1/b/legacy/o?maxResults=1"); + } + } + } } diff --git a/crates/ecstore/src/bucket/on_demand_migration/list_through.rs b/crates/ecstore/src/bucket/on_demand_migration/list_through.rs index dd5a6a236..124d82d8f 100644 --- a/crates/ecstore/src/bucket/on_demand_migration/list_through.rs +++ b/crates/ecstore/src/bucket/on_demand_migration/list_through.rs @@ -1089,10 +1089,17 @@ mod tests { #[test] fn progress_tokens_preserve_v1_bytes_and_validate_v2_counts() { + fn framed(payload: &str) -> String { + format!("{LIST_THROUGH_TOKEN_PREFIX}{payload}") + } + let token = progress_token(None, true, false); assert_eq!( token.encode(), - r#"{"t":"odm-list","v":1,"local":null,"local_done":true,"source":"A","source_done":false,"last_key":"last-key"}"# + concat!( + "\0odm-list:", + r#"{"t":"odm-list","v":1,"local":null,"local_done":true,"source":"A","source_done":false,"last_key":"last-key"}"# + ) ); for count in 1..MAX_LIST_NO_PROGRESS_PAGES { let token = progress_token(Some(count), true, false); @@ -1100,16 +1107,17 @@ mod tests { } for version in [1, 2] { for value in ["null", "0", "16", "-1", "1.5", "256", "18446744073709551616", "\"1\""] { - let encoded = format!(r#"{{"t":"odm-list","v":{version},"no_progress":{value}}}"#); + let encoded = framed(&format!(r#"{{"t":"odm-list","v":{version},"no_progress":{value}}}"#)); assert_eq!(decode_continuation_token(&encoded), Err(ListThroughTokenError::Malformed), "{encoded}"); } } - for encoded in [ + for payload in [ r#"{"t":"odm-list","v":1,"no_progress":1}"#, r#"{"t":"odm-list","v":2}"#, r#"{"t":"odm-list","v":2,"no_progress":1,"extra":true}"#, ] { - assert_eq!(decode_continuation_token(encoded), Err(ListThroughTokenError::Malformed), "{encoded}"); + let encoded = framed(payload); + assert_eq!(decode_continuation_token(&encoded), Err(ListThroughTokenError::Malformed), "{encoded}"); } } diff --git a/crates/ecstore/src/bucket/on_demand_migration/native_http.rs b/crates/ecstore/src/bucket/on_demand_migration/native_http.rs index 881db697d..8e5c12dd4 100644 --- a/crates/ecstore/src/bucket/on_demand_migration/native_http.rs +++ b/crates/ecstore/src/bucket/on_demand_migration/native_http.rs @@ -129,6 +129,23 @@ impl NativeHttp { &self, request: reqwest::Request, error_code_header: &str, + ) -> Result { + self.send_classified(request, error_code_header, false).await + } + + pub(super) async fn send_object( + &self, + request: reqwest::Request, + error_code_header: &str, + ) -> Result { + self.send_classified(request, error_code_header, true).await + } + + async fn send_classified( + &self, + request: reqwest::Request, + error_code_header: &str, + not_found_on_404_without_code: bool, ) -> Result { let response = self.client.execute(request).await.map_err(classify_transport_error)?; let status = response.status(); @@ -140,14 +157,14 @@ impl NativeHttp { .get(error_code_header) .and_then(|value| value.to_str().ok()) .map(str::to_string); - Err(classify_status( - status.as_u16(), - None, - match &code { - Some(code) => format!("source returned HTTP {status} ({code})"), - None => format!("source returned HTTP {status}"), - }, - )) + let message = match &code { + Some(code) => format!("source returned HTTP {status} ({code})"), + None => format!("source returned HTTP {status}"), + }; + match classify_status(status.as_u16(), code.as_deref(), message) { + SourceError::Other(_) if not_found_on_404_without_code && status.as_u16() == 404 => Err(SourceError::NotFound), + err => Err(err), + } } } diff --git a/crates/ecstore/src/bucket/on_demand_migration/source_client.rs b/crates/ecstore/src/bucket/on_demand_migration/source_client.rs index 618fff00e..6d1ba136a 100644 --- a/crates/ecstore/src/bucket/on_demand_migration/source_client.rs +++ b/crates/ecstore/src/bucket/on_demand_migration/source_client.rs @@ -334,8 +334,9 @@ const THROTTLE_CODES: &[&str] = &[ "RequestLimitExceeded", "TooManyRequests", "RequestThrottled", + "ServerBusy", ]; -const NOT_FOUND_CODES: &[&str] = &["NoSuchKey"]; +const NOT_FOUND_CODES: &[&str] = &["NoSuchKey", "BlobNotFound"]; const ACCESS_DENIED_CODES: &[&str] = &[ "AccessDenied", "InvalidAccessKeyId", @@ -343,6 +344,7 @@ const ACCESS_DENIED_CODES: &[&str] = &[ "AllAccessDisabled", "ExpiredToken", "InvalidToken", + "AuthorizationPermissionMismatch", ]; pub(super) fn classify_status(status: u16, code: Option<&str>, message: String) -> SourceError { @@ -1817,6 +1819,7 @@ mod tests { ok(Vec::new(), CONTRACT_TAGGING), ok(Vec::new(), ""), status(404, ""), + ok(Vec::new(), ""), status(403, ACCESS_DENIED_BODY), ]) .await; diff --git a/scripts/test_security_workflow.py b/scripts/test_security_workflow.py index b96b3e3f5..23268bf73 100644 --- a/scripts/test_security_workflow.py +++ b/scripts/test_security_workflow.py @@ -355,6 +355,8 @@ fi self.assertFalse(any(line.strip().startswith("continue-on-error:") for line in self.steps[handoff])) self.assertIn(" if: always()", self.steps["Cleanup environment (after)"]) self.assertLess(list(self.steps).index("Cleanup environment (after)"), list(self.steps).index(handoff)) + initialized = self.run_step("Initialize functional evidence") + self.assertEqual(initialized.returncode, 0, initialized.stderr) suite = self.directory / "auto-testing/rustfs-replication-test.sh" suite.write_text('#!/bin/sh\nprintf "suite failed\\n" >> "$EXECUTED"\nexit 17\n') failed = self.run_step("Run replication suite")