fix(ecstore): restore odm source contract tests (#7215)

* fix(ecstore): restore odm source contract tests

Co-Authored-By: heihutu <heihutu@gmail.com>

Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* 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 <heihutu@gmail.com>

Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* fix(odm): distinguish missing GCS buckets from object misses (#7221)

---------

Co-authored-by: zhi22915 <qiuzgang@gmail.com>
Co-authored-by: Zhengchao An <anzhengchao@gmail.com>

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<reqwest::Response, SourceError> {
+        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<SourceHead, SourceError> {
         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<SourceHead, SourceError> {
         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");
+            }
+        }
+    }
 }

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}");
         }
     }
 

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<reqwest::Response, SourceError> {
+        self.send_classified(request, error_code_header, false).await
+    }
+
+    pub(super) async fn send_object(
+        &self,
+        request: reqwest::Request,
+        error_code_header: &str,
+    ) -> Result<reqwest::Response, SourceError> {
+        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<reqwest::Response, SourceError> {
         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(
+        let message = match &code {
-            status.as_u16(),
+            Some(code) => format!("source returned HTTP {status} ({code})"),
-            None,
+            None => format!("source returned HTTP {status}"),
-            match &code {
+        };
-                Some(code) => format!("source returned HTTP {status} ({code})"),
+        match classify_status(status.as_u16(), code.as_deref(), message) {
-                None => format!("source returned HTTP {status}"),
+            SourceError::Other(_) if not_found_on_404_without_code && status.as_u16() == 404 => Err(SourceError::NotFound),
-            },
+            err => Err(err),
-        ))
+        }
     }
 }
 

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;

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")