ci(upgrade): support manual runs between any two release versions

The workflow_dispatch inputs already accept arbitrary release tags, but
the run failed late and unclearly when a tag had no .deb asset, and the
from_version default pointed at 1.0.0-rc.4-preview.1, whose release
ships no .deb at all - so scheduled runs died on a 404 while installing
the old package.

- Add a fail-fast preflight that resolves each requested tag via the
  GitHub release API and verifies the rustfs_<tag>_amd64.deb asset
  exists before the suite starts, with an actionable error message
  otherwise (e.g. 1.0.0-rc.4 ships only zip/sbom assets).
- Change the from_version default to 1.0.0-rc.3, the newest release
  that actually ships a .deb asset.
- Reword the from_version/to_version descriptions so manual triggers
  state the .deb-asset requirement and the nightly fallback.
- Pass PF_TESTING_GH_TOKEN as GH_TOKEN to the suite step for the gh api
  release lookups, matching the other functional workflows.

.github/workflows/rustfs-upgrade-test.yml

@@ -18,15 +18,15 @@ on:
   workflow_dispatch:
     inputs:
       from_version:
-        description: 'OLD RustFS release tag (e.g. 1.0.0-rc.4-preview.1)'
+        description: 'OLD RustFS release tag, e.g. 1.0.0-rc.3 (its release must ship a .deb asset). Leave empty for the default.'
         required: false
-        default: '1.0.0-rc.4-preview.1'
+        default: '1.0.0-rc.3'
       from_url:
         description: 'OLD .deb URL. Overrides from_version.'
         required: false
         type: string
       to_version:
-        description: 'NEW RustFS release tag (leave empty for latest nightly)'
+        description: 'NEW RustFS release tag, e.g. 1.0.0-rc.5 (any version with a .deb asset). Leave empty for latest nightly.'
         required: false
       to_url:
         description: 'NEW .deb URL. Overrides to_version / nightly default.'
@@ -145,6 +145,7 @@ jobs:
         continue-on-error: true
         env:
           LOG_FILE: /tmp/rustfs-upgrade.log
+          GH_TOKEN: ${{ secrets.PF_TESTING_GH_TOKEN }}
         run: |
           set -euo pipefail
           chmod +x auto-testing/rustfs-upgrade-test.sh
@@ -175,6 +176,29 @@ jobs:
           else
             ARGS+=(--to-url "${RUSTFS_NIGHTLY_PACKAGE_URL}")
           fi
+          # Fail fast with a clear message when a requested release tag has
+          # no .deb asset (e.g. 1.0.0-rc.4 ships only zips), instead of
+          # letting the suite die mid-run on a 404.
+          check_release_asset() {
+            local version="$1" tag asset url
+            [ -n "${version}" ] && [ "${version}" != "null" ] || return 0
+            tag="${version#v}"
+            asset="rustfs_${tag//-/.}_amd64.deb"
+            url="https://github.com/rustfs/rustfs/releases/download/${tag}/${asset}"
+            if ! gh api "repos/rustfs/rustfs/releases/tags/${tag}" --jq '.assets[].name' 2>/dev/null | grep -qxF "${asset}"; then
+              echo "ERROR: release ${tag} has no downloadable asset ${asset}:" >&2
+              echo "       ${url}" >&2
+              echo "Pick a tag whose release ships a .deb (check its release assets)." >&2
+              exit 1
+            fi
+            echo "resolved ${tag} -> ${url}"
+          }
+          if [ -z "${FROM_URL}" ]; then
+            check_release_asset "${FROM_VERSION}"
+          fi
+          if [ -z "${TO_URL}" ]; then
+            check_release_asset "${TO_VERSION}"
+          fi
           ./auto-testing/rustfs-upgrade-test.sh "${ARGS[@]}"
 
       - name: Generate report

crates/ecstore/src/bucket/on_demand_migration/source_client.rs

@@ -14,9 +14,9 @@
 
 //! Outbound client for an on-demand migration source bucket.
 //!
-//! `SourceClient` maps local keys onto a read-only `SourceBackend`. The
-//! S3 backend uses the shared remote builder and exposes the surface the
-//! migration path needs (HEAD, ranged streaming GET, ListObjectsV2, GetObjectTagging, a
+//! `SourceClient` wraps an `aws_sdk_s3::Client` built through the shared
+//! remote builder and exposes the read-only surface the migration path
+//! needs (HEAD, ranged streaming GET, ListObjectsV2, GetObjectTagging, a
 //! probe for admin validation). Every request carries the
 //! `source-proxy-request` anti-loop marker in both the `x-rustfs-` and
 //! `x-minio-` prefixes so a RustFS/MinIO source answers locally instead of
@@ -511,7 +511,7 @@ pub struct SourceObject {
 #[derive(Clone, Debug, Default, PartialEq, Eq)]
 pub struct SourcePage {
     pub objects: Vec<SourceObject>,
-    /// Rolled-up prefixes, in the same namespace as `objects`; always empty when the
+    /// Rolled-up prefixes, in the local namespace; always empty when the
     /// request carried no delimiter.
     pub common_prefixes: Vec<String>,
     pub is_truncated: bool,
@@ -519,8 +519,7 @@ pub struct SourcePage {
 }
 
 /// One `ListObjectsV2` page request against the source. Keys are given in the
-/// local namespace at `SourceClient`, and in the source namespace at
-/// `SourceBackend`; `SourceClient` maps them through `source_prefix`.
+/// local namespace; `SourceClient` maps them through `source_prefix`.
 #[derive(Clone, Debug, Default, PartialEq, Eq)]
 pub struct SourceListRequest<'a> {
     pub prefix: Option<&'a str>,
@@ -576,30 +575,8 @@ impl Intercept for SourceProxyMarkerInterceptor {
     }
 }
 
-/// Read-only provider operations in the source bucket namespace.
-///
-/// Implementations must preserve streaming, honor the requested range and
-/// pagination cursor, and classify failures without including credentials.
-/// `SourceClient` owns prefix mapping so every provider shares the same local
-/// namespace. Continuation tokens are opaque and must never be prefix-mapped.
-#[async_trait::async_trait]
-pub trait SourceBackend: Send + Sync {
-    async fn head(&self, key: &str) -> Result<SourceHead, SourceError>;
-    async fn get(&self, key: &str, range: Option<&HTTPRangeSpec>) -> Result<SourceGet, SourceError>;
-    async fn list(&self, request: &SourceListRequest<'_>) -> Result<SourcePage, SourceError>;
-    async fn tagging(&self, key: &str) -> Result<HashMap<String, String>, SourceError>;
-    /// Verify bucket access; `SourceClient` separately probes a filtered listing.
-    async fn probe(&self) -> Result<(), SourceError>;
-}
-
-/// S3-compatible implementation, including request signing and anti-loop headers.
-pub struct S3SourceBackend {
-    client: S3Client,
-    bucket: String,
-}
-
 pub struct SourceClient {
-    backend: Box<dyn SourceBackend>,
+    client: S3Client,
     endpoint: String,
     bucket: String,
     source_prefix: Option<String>,
@@ -632,10 +609,7 @@ impl SourceClient {
     fn from_config_builder(config: aws_sdk_s3::config::Builder, endpoint: String, spec: &SourceClientSpec) -> Self {
         let client = S3Client::from_conf(config.interceptor(SourceProxyMarkerInterceptor::new()).build());
         Self {
-            backend: Box::new(S3SourceBackend {
-                client,
-                bucket: spec.bucket.clone(),
-            }),
+            client,
             endpoint,
             bucket: spec.bucket.clone(),
             source_prefix: spec.source_prefix.clone().filter(|prefix| !prefix.is_empty()),
@@ -678,15 +652,35 @@ impl SourceClient {
     }
 
     pub async fn head_object(&self, key: &str) -> Result<SourceHead, SourceError> {
-        self.backend.head(&self.source_key(key)).await
+        let output = self
+            .client
+            .head_object()
+            .bucket(&self.bucket)
+            .key(self.source_key(key))
+            .send()
+            .await
+            .map_err(classify_sdk_error)?;
+        source_head_from_head_output(output)
     }
 
-    /// Streams the object, preserving an optional HTTP byte range.
+    /// Streams the object; `range` is passed through as an HTTP `Range`
+    /// header and omitted entirely when `None`.
     pub async fn get_object(&self, key: &str, range: Option<&HTTPRangeSpec>) -> Result<SourceGet, SourceError> {
-        self.backend.get(&self.source_key(key), range).await
+        let range = range.map(range_header_value).transpose()?;
+        let output = self
+            .client
+            .get_object()
+            .bucket(&self.bucket)
+            .key(self.source_key(key))
+            .set_range(range)
+            .send()
+            .await
+            .map_err(classify_sdk_error)?;
+        source_get_from_output(output)
     }
 
-    /// Lists one page under the local prefix.
+    /// Lists one page under the local `prefix`. Keys are returned in the
+    /// local namespace; entries outside `source_prefix` are skipped.
     pub async fn list_objects_v2(
         &self,
         prefix: Option<&str>,
@@ -702,81 +696,9 @@ impl SourceClient {
         .await
     }
 
-    /// Maps keys and common prefixes while leaving opaque cursors untouched.
+    /// [`Self::list_objects_v2`] with the delimiter and start-after the
+    /// list-through merge needs (rustfs/backlog#2164).
     pub async fn list_page(&self, request: &SourceListRequest<'_>) -> Result<SourcePage, SourceError> {
-        let prefix = self.source_key(request.prefix.unwrap_or_default());
-        let start_after = request.start_after.map(|key| self.source_key(key));
-        let mut page = self
-            .backend
-            .list(&SourceListRequest {
-                prefix: Some(&prefix),
-                start_after: start_after.as_deref(),
-                ..*request
-            })
-            .await?;
-        page.objects = page
-            .objects
-            .into_iter()
-            .filter_map(|object| self.local_object(object))
-            .collect();
-        page.common_prefixes = page
-            .common_prefixes
-            .into_iter()
-            .filter_map(|prefix| self.local_key(&prefix).map(str::to_string))
-            .collect();
-        Ok(page)
-    }
-
-    fn local_object(&self, mut object: SourceObject) -> Option<SourceObject> {
-        object.key = self.local_key(&object.key)?.to_string();
-        Some(object)
-    }
-
-    pub async fn get_object_tagging(&self, key: &str) -> Result<HashMap<String, String>, SourceError> {
-        self.backend.tagging(&self.source_key(key)).await
-    }
-
-    pub async fn probe(&self) -> Result<SourceProbe, SourceError> {
-        self.backend.probe().await?;
-        let page = self.list_objects_v2(None, None, 1).await?;
-        Ok(SourceProbe {
-            sample_object: page.objects.into_iter().next(),
-            has_more_objects: page.is_truncated,
-        })
-    }
-}
-
-#[async_trait::async_trait]
-impl SourceBackend for S3SourceBackend {
-    async fn head(&self, key: &str) -> Result<SourceHead, SourceError> {
-        let output = self
-            .client
-            .head_object()
-            .bucket(&self.bucket)
-            .key(key)
-            .send()
-            .await
-            .map_err(classify_sdk_error)?;
-        source_head_from_head_output(output)
-    }
-
-    /// Streams the object; `range` is passed through as an HTTP `Range`
-    /// header and omitted entirely when `None`.
-    async fn get(&self, key: &str, range: Option<&HTTPRangeSpec>) -> Result<SourceGet, SourceError> {
-        let range = range.map(range_header_value).transpose()?;
-        let output = self
-            .client
-            .get_object()
-            .bucket(&self.bucket)
-            .key(key)
-            .set_range(range)
-            .send()
-            .await
-            .map_err(classify_sdk_error)?;
-        source_get_from_output(output)
-    }
-
-    async fn list(&self, request: &SourceListRequest<'_>) -> Result<SourcePage, SourceError> {
         // `start_after` is silently ignored by S3 once a continuation token is
         // present; refuse the ambiguous pair rather than list from the wrong
         // position.
@@ -789,9 +711,9 @@ impl SourceBackend for S3SourceBackend {
             .client
             .list_objects_v2()
             .bucket(&self.bucket)
-            .prefix(request.prefix.unwrap_or_default())
+            .prefix(self.source_key(request.prefix.unwrap_or_default()))
             .set_delimiter(request.delimiter.map(str::to_string))
-            .set_start_after(request.start_after.map(str::to_string))
+            .set_start_after(request.start_after.map(|after| self.source_key(after)))
             .set_continuation_token(request.continuation_token.map(str::to_string))
             .max_keys(request.max_keys)
             .send()
@@ -809,13 +731,13 @@ impl SourceBackend for S3SourceBackend {
             .contents
             .unwrap_or_default()
             .into_iter()
-            .filter_map(s3_source_object)
+            .filter_map(|object| self.source_object(object))
             .collect();
         let common_prefixes = output
             .common_prefixes
             .unwrap_or_default()
             .into_iter()
-            .filter_map(|prefix| prefix.prefix)
+            .filter_map(|prefix| Some(self.local_key(prefix.prefix.as_deref()?)?.to_string()))
             .collect();
 
         Ok(SourcePage {
@@ -826,43 +748,48 @@ impl SourceBackend for S3SourceBackend {
         })
     }
 
-    async fn tagging(&self, key: &str) -> Result<HashMap<String, String>, SourceError> {
+    fn source_object(&self, object: SdkObject) -> Option<SourceObject> {
+        let key = self.local_key(object.key.as_deref()?)?.to_string();
+        let etag = normalize_etag(object.e_tag);
+        let is_multipart_etag = etag.as_deref().is_some_and(is_multipart_etag);
+        Some(SourceObject {
+            key,
+            etag,
+            size: object.size.and_then(|size| u64::try_from(size).ok()).unwrap_or(0),
+            last_modified: system_time(object.last_modified),
+            storage_class: object.storage_class.map(|class| class.as_str().to_string()),
+            is_multipart_etag,
+        })
+    }
+
+    pub async fn get_object_tagging(&self, key: &str) -> Result<HashMap<String, String>, SourceError> {
         let output = self
             .client
             .get_object_tagging()
             .bucket(&self.bucket)
-            .key(key)
+            .key(self.source_key(key))
             .send()
             .await
             .map_err(classify_sdk_error)?;
         Ok(output.tag_set.into_iter().map(|tag| (tag.key, tag.value)).collect())
     }
 
-    async fn probe(&self) -> Result<(), SourceError> {
+    /// Admin validation: HeadBucket plus a one-key listing under the prefix.
+    pub async fn probe(&self) -> Result<SourceProbe, SourceError> {
         self.client
             .head_bucket()
             .bucket(&self.bucket)
             .send()
             .await
             .map_err(classify_sdk_error)?;
-        Ok(())
+        let page = self.list_objects_v2(None, None, 1).await?;
+        Ok(SourceProbe {
+            sample_object: page.objects.into_iter().next(),
+            has_more_objects: page.is_truncated,
+        })
     }
 }
 
-fn s3_source_object(object: SdkObject) -> Option<SourceObject> {
-    let key = object.key?;
-    let etag = normalize_etag(object.e_tag);
-    let is_multipart_etag = etag.as_deref().is_some_and(is_multipart_etag);
-    Some(SourceObject {
-        key,
-        etag,
-        size: object.size.and_then(|size| u64::try_from(size).ok()).unwrap_or(0),
-        last_modified: system_time(object.last_modified),
-        storage_class: object.storage_class.map(|class| class.as_str().to_string()),
-        is_multipart_etag,
-    })
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -1267,68 +1194,6 @@ mod tests {
         assert!(requests[1].uri.contains("continuation-token=token-1"), "{}", requests[1].uri);
     }
 
-    #[tokio::test]
-    async fn list_page_maps_delimiter_prefixes_and_start_after_but_not_cursors() {
-        let body = r#"<ListBucketResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
-<IsTruncated>true</IsTruncated><NextContinuationToken>data/opaque</NextContinuationToken>
-<CommonPrefixes><Prefix>data/photos/</Prefix></CommonPrefixes>
-<CommonPrefixes><Prefix>outside/</Prefix></CommonPrefixes>
-</ListBucketResult>"#;
-        let (client, requests) = scripted_client(&spec(Some("data/")), vec![ok(Vec::new(), body), ok(Vec::new(), body)]).await;
-        let first = client
-            .list_page(&SourceListRequest {
-                prefix: Some("photos/"),
-                delimiter: Some("/"),
-                start_after: Some("photos/a"),
-                max_keys: 2,
-                ..Default::default()
-            })
-            .await
-            .expect("delimiter listing should succeed");
-        assert_eq!(first.common_prefixes, vec!["photos/"]);
-        assert_eq!(first.next_continuation_token.as_deref(), Some("data/opaque"));
-        let second = client
-            .list_page(&SourceListRequest {
-                continuation_token: first.next_continuation_token.as_deref(),
-                max_keys: 2,
-                ..Default::default()
-            })
-            .await
-            .expect("opaque continuation should succeed");
-        assert_eq!(second.common_prefixes, first.common_prefixes);
-        let requests = recorded(&requests);
-        let query = |request: &RecordedRequest| {
-            Url::parse(&request.uri)
-                .expect("request URI")
-                .query_pairs()
-                .into_owned()
-                .collect::<HashMap<_, _>>()
-        };
-        let first_query = query(&requests[0]);
-        assert_eq!(first_query.get("prefix").map(String::as_str), Some("data/photos/"));
-        assert_eq!(first_query.get("start-after").map(String::as_str), Some("data/photos/a"));
-        assert_eq!(first_query.get("delimiter").map(String::as_str), Some("/"));
-        let second_query = query(&requests[1]);
-        assert_eq!(second_query.get("continuation-token").map(String::as_str), Some("data/opaque"));
-        assert!(!second_query.contains_key("start-after"));
-    }
-
-    #[tokio::test]
-    async fn list_page_rejects_ambiguous_cursor_before_sending() {
-        let (client, requests) = scripted_client(&spec(Some("data/")), vec![]).await;
-        let err = client
-            .list_page(&SourceListRequest {
-                start_after: Some("a"),
-                continuation_token: Some("opaque"),
-                max_keys: 1,
-                ..Default::default()
-            })
-            .await
-            .expect_err("ambiguous list position must fail");
-        assert!(matches!(err, SourceError::Other(_)));
-        assert!(recorded(&requests).is_empty(), "invalid request must never reach the source");
-    }
-
     #[tokio::test]
     async fn list_objects_v2_rejects_truncated_page_without_token() {
         let (client, _) = scripted_client(&spec(None), vec![ok(Vec::new(), LIST_TRUNCATED_WITHOUT_TOKEN)]).await;
@@ -1492,14 +1357,11 @@ mod tests {
 
     fn prefix_client(prefix: Option<String>) -> SourceClient {
         SourceClient {
-            backend: Box::new(S3SourceBackend {
-                client: S3Client::from_conf(
-                    aws_sdk_s3::Config::builder()
-                        .behavior_version(aws_sdk_s3::config::BehaviorVersion::latest())
-                        .build(),
-                ),
-                bucket: "bucket".to_string(),
-            }),
+            client: S3Client::from_conf(
+                aws_sdk_s3::Config::builder()
+                    .behavior_version(aws_sdk_s3::config::BehaviorVersion::latest())
+                    .build(),
+            ),
             endpoint: "https://source.example.com".to_string(),
             bucket: "bucket".to_string(),
             source_prefix: prefix.filter(|prefix| !prefix.is_empty()),

crates/ecstore/src/bucket/sealed_credentials.rs

@@ -201,12 +201,6 @@ pub async fn unseal_secret(sealed: &SealedCredential, scope: &SealScope) -> Resu
 mod tests {
     use super::*;
     use parking_lot::Mutex;
-    use std::collections::BTreeMap;
-
-    fn encode_context(context: &HashMap<String, String>) -> String {
-        let ordered = context.iter().collect::<BTreeMap<_, _>>();
-        serde_json::to_string(&ordered).expect("context serializes")
-    }
 
     /// Stands in for the KMS-backed sealer: records the context it was called
     /// with, and refuses a ciphertext presented under a different one.
@@ -220,7 +214,7 @@ mod tests {
         async fn seal(&self, plaintext: &str, scope: &SealScope) -> Result<SealedCredential, SealedCredentialError> {
             let context = scope.encryption_context();
             self.sealed_contexts.lock().push(context.clone());
-            let mut bound = encode_context(&context);
+            let mut bound = serde_json::to_string(&context).expect("context serializes");
             bound.push('|');
             bound.push_str(plaintext);
             Ok(SealedCredential {
@@ -237,7 +231,7 @@ mod tests {
                 .decode_to_vec(sealed.ct.as_bytes())
                 .map_err(|err| SealedCredentialError::Malformed(err.to_string()))?;
             let bound = String::from_utf8(raw).map_err(|err| SealedCredentialError::Malformed(err.to_string()))?;
-            let expected = encode_context(&scope.encryption_context());
+            let expected = serde_json::to_string(&scope.encryption_context()).expect("context serializes");
             bound
                 .strip_prefix(&expected)
                 .and_then(|rest| rest.strip_prefix('|'))