gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-09-07 20:46:11 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(odm): preserve cursor compatibility and native source semantics (#7238)

Browse Source
This commit is contained in:
Zhengchao An
2026-09-06 14:12:56 +08:00
committed by GitHub
parent 760c9d65be
commit 1a88870809
17 changed files with 2370 additions and 120 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rustfs/fixtures/on_demand_migration/source_config_e2a.rs
+80
View File
@@ -0,0 +1,80 @@
// Strict source reader frozen from e2a921bc1608823c8efec955d7463ab8350a8a01.
// Wire declarations and credential Debug are copied verbatim; runtime methods are omitted.
use serde::{Deserialize, Serialize};
use std::fmt;
const REDACTED: &str = "REDACTED";
/// The external S3-compatible source bucket.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct SourceConfig {
pub provider: Provider,
/// `http(s)://host[:port]` with no path or query. Optional only for
/// [`Provider::Aws`], where it is derived from `region`.
#[serde(default)]
pub endpoint: Option<String>,
pub region: String,
pub bucket: String,
#[serde(default)]
pub path_style: PathStyle,
/// `None` means anonymous access to a public source bucket.
#[serde(default)]
pub credentials: Option<SourceCredentials>,
#[serde(default)]
pub tls: TlsConfig,
}
/// Source vendor family. `azure` is deliberately absent from this version.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum Provider {
/// Generic S3-compatible endpoint.
S3,
Aws,
Minio,
Rustfs,
R2,
/// GCS XML interoperability API with HMAC keys.
Gcs,
}
/// Bucket addressing style. `auto` is resolved by the source client builder.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum PathStyle {
#[default]
Auto,
Path,
Virtual,
}
/// Static credentials for the source. `Debug` never prints the secret or
/// the session token.
#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct SourceCredentials {
pub access_key: String,
pub secret_key: String,
#[serde(default)]
pub session_token: Option<String>,
}
impl fmt::Debug for SourceCredentials {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("SourceCredentials")
.field("access_key", &self.access_key)
.field("secret_key", &REDACTED)
.field("session_token", &self.session_token.as_ref().map(|_| REDACTED))
.finish()
}
}
#[derive(Debug, Clone, PartialEq, Eq, Default, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct TlsConfig {
#[serde(default)]
pub skip_verify: bool,
#[serde(default)]
pub ca_cert_pem: Option<String>,
}
rustfs/src/app/bucket_list_through.rs
+1076 -19
View File
File diff suppressed because it is too large Load Diff
rustfs/src/app/bucket_usecase.rs
+18 -4
View File
@@ -2783,8 +2783,21 @@ impl DefaultBucketUsecase {
} else {
(None, None)
};
let (object_infos, degraded) = match source_state {
Some(state) => {
let (object_infos, degraded) = match (source_state, merged_token.as_ref()) {
(None, Some(token)) if params.max_keys == 0 => {
// No source was consulted, so retain every unconsumed side and
// the original wire format without spending its progress budget.
let is_truncated = !token.local_done || !token.source_done;
(
StorageListObjectsV2Info {
is_truncated,
next_continuation_token: params.decoded_continuation_token.clone().filter(|_| is_truncated),
..Default::default()
},
false,
)
}
(Some(state), _) => {
let outcome = list_through::merged_list_objects_v2(
&store,
&state,
@@ -2797,12 +2810,12 @@ impl DefaultBucketUsecase {
.await?;
(outcome.info, outcome.degraded)
}
None => {
(None, _) => {
let cursor = list_through::local_cursor(params.decoded_continuation_token.as_deref(), merged_token.as_ref());
match cursor {
list_through::LocalListCursor::Exhausted => (StorageListObjectsV2Info::default(), false),
list_through::LocalListCursor::Token(token) => {
let infos = store
let mut infos = store
.list_objects_v2(
&bucket,
&params.prefix,
@@ -2815,6 +2828,7 @@ impl DefaultBucketUsecase {
)
.await
.map_err(ApiError::from)?;
list_through::preserve_framed_local_cursor(&mut infos, merged_token.as_ref());
(infos, false)
}
}
rustfs/src/on_demand_migration/azure.rs
+433 -18
View File
@@ -163,6 +163,32 @@ impl AzureSourceBackend {
Ok(request)
}
/// A missing blob is distinct from a missing container or version. Only
/// object reads may use BlobNotFound as positive evidence of absence.
async fn send_object_request(&self, request: reqwest::Request) -> Result<reqwest::Response, SourceError> {
let is_head = request.method() == Method::HEAD;
let versioned = request
.url()
.query_pairs()
.any(|(name, _)| name.eq_ignore_ascii_case("versionid") || name.eq_ignore_ascii_case("snapshot"));
let response = self.http.execute(request).await?;
if response.status() == http::StatusCode::NOT_FOUND && !versioned {
match header(response.headers(), HEADER_ERROR_CODE) {
Some("BlobNotFound") => return Err(SourceError::NotFound),
None | Some("ResourceNotFound") if is_head => {
// HEAD may omit an error code. One successful container
// probe proves key absence; a failed probe keeps its error.
// These are two independently timed requests, not one deadline.
drop(response);
self.probe().await?;
return Err(SourceError::NotFound);
}
_ => {}
}
}
NativeHttp::check_response(response, Some(HEADER_ERROR_CODE))
}
/// Shared mapping for Get Blob and Get Blob Properties.
fn head_from_response(headers: &HeaderMap) -> Result<SourceHead, SourceError> {
// A customer-provided key means the service holds ciphertext it cannot
@@ -189,7 +215,7 @@ impl AzureSourceBackend {
impl SourceBackend for AzureSourceBackend {
async fn head(&self, key: &str) -> Result<SourceHead, SourceError> {
let request = self.request(Method::HEAD, self.blob_url(key)?, HeaderMap::new())?;
let response = self.http.send(request, HEADER_ERROR_CODE).await?;
let response = self.send_object_request(request).await?;
Self::head_from_response(response.headers())
}
@@ -202,7 +228,7 @@ impl SourceBackend for AzureSourceBackend {
);
}
let request = self.request(Method::GET, self.blob_url(key)?, headers)?;
let response = self.http.send(request, HEADER_ERROR_CODE).await?;
let response = self.send_object_request(request).await?;
let head = Self::head_from_response(response.headers())?;
let content_range = header(response.headers(), "content-range").map(str::to_string);
Ok(SourceGet {
@@ -240,7 +266,7 @@ impl SourceBackend for AzureSourceBackend {
}
let request = self.request(Method::GET, url, HeaderMap::new())?;
let response = self.http.send(request, HEADER_ERROR_CODE).await?;
let response = self.http.send(request, Some(HEADER_ERROR_CODE)).await?;
let body = read_text(response, MAX_XML_BYTES).await?;
let listing = parse_list_blobs(&body)?;
@@ -256,7 +282,7 @@ impl SourceBackend for AzureSourceBackend {
let mut url = self.blob_url(key)?;
url.query_pairs_mut().append_pair("comp", "tags");
let request = self.request(Method::GET, url, HeaderMap::new())?;
let response = self.http.send(request, HEADER_ERROR_CODE).await?;
let response = self.http.send(request, Some(HEADER_ERROR_CODE)).await?;
let body = read_text(response, MAX_XML_BYTES).await?;
parse_blob_tags(&body)
}
@@ -265,7 +291,7 @@ impl SourceBackend for AzureSourceBackend {
let mut url = self.container_url()?;
url.query_pairs_mut().append_pair("restype", "container");
let request = self.request(Method::HEAD, url, HeaderMap::new())?;
self.http.send(request, HEADER_ERROR_CODE).await?;
self.http.send(request, Some(HEADER_ERROR_CODE)).await?;
Ok(())
}
}
@@ -343,9 +369,9 @@ struct AzureListing {
#[derive(Default)]
struct BlobEntry {
name: String,
name: Option<String>,
etag: Option<String>,
size: u64,
size: Option<u64>,
last_modified: Option<std::time::SystemTime>,
access_tier: Option<String>,
}
@@ -358,6 +384,7 @@ fn parse_list_blobs(xml: &str) -> Result<AzureListing, SourceError> {
let mut next_marker = None;
let mut blob: Option<BlobEntry> = None;
let mut in_blob_prefix = false;
let mut blob_prefix: Option<String> = None;
// Open container elements. quick-xml reports a truncated document as a
// plain end of input, so a non-zero depth at EOF is the only signal that
// the page was cut short and must not be read as a complete listing.
@@ -367,6 +394,9 @@ fn parse_list_blobs(xml: &str) -> Result<AzureListing, SourceError> {
match reader.read_event() {
Ok(Event::Start(start)) => {
let name = local_name(start.name().as_ref());
if matches!(name.as_str(), "blob" | "blobprefix") && (blob.is_some() || in_blob_prefix) {
return Err(SourceError::Other("source listing entries must not be nested".to_string()));
}
match name.as_str() {
"blob" => {
depth += 1;
@@ -385,27 +415,35 @@ fn parse_list_blobs(xml: &str) -> Result<AzureListing, SourceError> {
} else {
text
};
apply_list_field(&name, text, &mut blob, &mut prefixes, &mut next_marker, in_blob_prefix);
apply_list_field(&name, text, &mut blob, &mut blob_prefix, &mut next_marker, in_blob_prefix)?;
}
}
}
Ok(Event::Empty(empty)) => {
let name = local_name(empty.name().as_ref());
if matches!(name.as_str(), "blob" | "blobprefix") {
return Err(SourceError::Other("source listing entry has no name".to_string()));
}
let text = if name == "name" {
decode_list_name(&empty, String::new())?
} else {
String::new()
};
apply_list_field(&name, text, &mut blob, &mut prefixes, &mut next_marker, in_blob_prefix);
apply_list_field(&name, text, &mut blob, &mut blob_prefix, &mut next_marker, in_blob_prefix)?;
}
Ok(Event::End(end)) => match local_name(end.name().as_ref()).as_str() {
"blob" => {
depth = depth.saturating_sub(1);
if let Some(entry) = blob.take() {
objects.push(SourceObject {
key: entry.name,
key: entry
.name
.filter(|name| !name.is_empty())
.ok_or_else(|| SourceError::Other("source listing object has no name".to_string()))?,
etag: entry.etag,
size: entry.size,
size: entry
.size
.ok_or_else(|| SourceError::Other("source listing object has no valid size".to_string()))?,
last_modified: entry.last_modified,
storage_class: entry.access_tier,
// Azure ETags carry no part count; the listing
@@ -417,6 +455,12 @@ fn parse_list_blobs(xml: &str) -> Result<AzureListing, SourceError> {
"blobprefix" => {
depth = depth.saturating_sub(1);
in_blob_prefix = false;
prefixes.push(
blob_prefix
.take()
.filter(|name| !name.is_empty())
.ok_or_else(|| SourceError::Other("source listing prefix has no name".to_string()))?,
);
}
"properties" | "blobs" | "enumerationresults" => depth = depth.saturating_sub(1),
_ => {}
@@ -478,16 +522,22 @@ fn apply_list_field(
name: &str,
text: String,
blob: &mut Option<BlobEntry>,
prefixes: &mut Vec<String>,
blob_prefix: &mut Option<String>,
next_marker: &mut Option<String>,
in_blob_prefix: bool,
) {
) -> Result<(), SourceError> {
match name {
"name" => {
if in_blob_prefix {
prefixes.push(text);
if blob_prefix.is_some() {
return Err(SourceError::Other("source listing prefix has duplicate names".to_string()));
}
*blob_prefix = Some(text);
} else if let Some(entry) = blob.as_mut() {
entry.name = text;
if entry.name.is_some() {
return Err(SourceError::Other("source listing object has duplicate names".to_string()));
}
entry.name = Some(text);
}
}
"nextmarker" => *next_marker = Some(text),
@@ -498,7 +548,14 @@ fn apply_list_field(
}
"content-length" => {
if let Some(entry) = blob.as_mut() {
entry.size = text.trim().parse().unwrap_or(0);
if entry.size.is_some() {
return Err(SourceError::Other("source listing object has duplicate sizes".to_string()));
}
entry.size = Some(
text.trim()
.parse()
.map_err(|_| SourceError::Other("source listing object has no valid size".to_string()))?,
);
}
}
"last-modified" => {
@@ -513,6 +570,7 @@ fn apply_list_field(
}
_ => {}
}
Ok(())
}
/// Parses a `Get Blob Tags` response.
@@ -599,7 +657,7 @@ mod tests {
use super::*;
use crate::on_demand_migration::backend_contract::{BackendCapabilities, assert_backend_contract};
use crate::on_demand_migration::source_client::SourceError;
use crate::on_demand_migration::test_http_fixture::{ScriptedResponse, scripted_server};
use crate::on_demand_migration::test_http_fixture::{ScriptedResponse, assert_requests, scripted_server};
const LIST_PAGE: &str = r#"<?xml version="1.0" encoding="utf-8"?>
<EnumerationResults ServiceEndpoint="https://acct.blob.core.windows.net/" ContainerName="legacy">
@@ -769,6 +827,168 @@ mod tests {
assert!(parse_blob_tags("<Tags><TagSet>").is_err(), "a truncated tag set must fail");
}
#[tokio::test]
async fn native_listing_rejects_missing_or_invalid_required_object_fields() {
for entry in [
"<Blob />",
"<Blob><Properties><Content-Length>1</Content-Length></Properties></Blob>",
"<Blob><Name /><Properties><Content-Length>1</Content-Length></Properties></Blob>",
"<Blob><Name>broken</Name></Blob>",
"<Blob><Name>broken</Name><Properties><Content-Length /></Properties></Blob>",
"<Blob><Name>broken</Name><Properties><Content-Length>-1</Content-Length></Properties></Blob>",
"<Blob><Name>broken</Name><Properties><Content-Length>18446744073709551616</Content-Length></Properties></Blob>",
"<Blob><Name>broken</Name><Properties><Content-Length>not-a-size</Content-Length></Properties></Blob>",
"<BlobPrefix />",
"<BlobPrefix><Name /></BlobPrefix>",
"<BlobPrefix></BlobPrefix>",
] {
// Reject the entire page even if a valid object precedes the bad
// entry, so callers cannot expose partial data or advance its cursor.
let body = format!(
"<EnumerationResults><Blobs><Blob><Name>valid</Name><Properties><Content-Length>1</Content-Length></Properties></Blob>{entry}</Blobs><NextMarker>next</NextMarker></EnumerationResults>"
);
let (endpoint, recorded) = scripted_server(vec![ScriptedResponse::new(200, Vec::new(), body)]).await;
let err = backend(&endpoint, Credential::SharedKey(vec![7_u8; 32]))
.list(&SourceListRequest {
prefix: Some("dir/"),
delimiter: Some("/"),
continuation_token: Some("opaque+/="),
max_keys: 2,
..Default::default()
})
.await
.expect_err("malformed object must reject the complete native page");
assert!(matches!(err, SourceError::Other(_)), "{entry}: {err:?}");
assert!(!err.is_retryable());
assert_requests(
&recorded,
&[(
"GET",
"/legacy?restype=container&comp=list&prefix=dir%2F&delimiter=%2F&marker=opaque%2B%2F%3D&maxresults=2",
)],
);
}
}
#[tokio::test]
async fn native_listing_rejects_duplicate_fields_and_nested_entries() {
for entry in [
"<Blob><Name>a</Name><Name>b</Name><Properties><Content-Length>1</Content-Length></Properties></Blob>",
"<Blob><Name /><Name>b</Name><Properties><Content-Length>1</Content-Length></Properties></Blob>",
"<Blob><Name>a</Name><Properties><Content-Length>1</Content-Length><Content-Length>2</Content-Length></Properties></Blob>",
"<BlobPrefix><Name>a/</Name><Name>b/</Name></BlobPrefix>",
"<BlobPrefix><Name /><Name>b/</Name></BlobPrefix>",
"<Blob><Name>a</Name><Properties><Content-Length>1</Content-Length></Properties><Blob><Name>b</Name><Properties><Content-Length>2</Content-Length></Properties></Blob></Blob>",
"<Blob><Name>a</Name><Properties><Content-Length>1</Content-Length></Properties><BlobPrefix><Name>b/</Name></BlobPrefix></Blob>",
"<BlobPrefix><Name>a/</Name><Blob><Name>b</Name><Properties><Content-Length>2</Content-Length></Properties></Blob></BlobPrefix>",
"<BlobPrefix><Name>a/</Name><BlobPrefix><Name>b/</Name></BlobPrefix></BlobPrefix>",
] {
let body = format!(
"<EnumerationResults><Blobs><Blob><Name>valid</Name><Properties><Content-Length>0</Content-Length></Properties></Blob>{entry}</Blobs><NextMarker>next</NextMarker></EnumerationResults>"
);
let (endpoint, recorded) = scripted_server(vec![ScriptedResponse::new(200, Vec::new(), body)]).await;
let result = backend(&endpoint, Credential::SharedKey(vec![7_u8; 32]))
.list(&SourceListRequest {
delimiter: Some("/"),
continuation_token: Some("opaque+/="),
max_keys: 2,
..Default::default()
})
.await;
let err = result.expect_err("ambiguous entries must reject the entire page and its cursor");
assert!(matches!(err, SourceError::Other(_)), "{entry}: {err:?}");
assert!(!err.is_retryable(), "{entry}: {err:?}");
assert_requests(
&recorded,
&[(
"GET",
"/legacy?restype=container&comp=list&delimiter=%2F&marker=opaque%2B%2F%3D&maxresults=2",
)],
);
}
}
#[tokio::test]
async fn native_listing_preserves_zero_size_unicode_prefixes_and_opaque_cursors() {
let body = "<EnumerationResults><Blobs><Blob><Name>目录/空 &amp; file</Name><Properties><Content-Length>0</Content-Length></Properties></Blob><BlobPrefix><Name>目录/子/</Name></BlobPrefix></Blobs><NextMarker>opaque+/=</NextMarker></EnumerationResults>";
let (endpoint, recorded) = scripted_server(vec![ScriptedResponse::new(200, Vec::new(), body.to_string())]).await;
let page = backend(&endpoint, Credential::SharedKey(vec![7_u8; 32]))
.list(&SourceListRequest {
max_keys: 2,
..Default::default()
})
.await
.expect("valid native page");
assert_eq!(page.objects.len(), 1);
assert_eq!(page.objects[0].key, "目录/空 & file");
assert_eq!(page.objects[0].size, 0);
assert_eq!(page.common_prefixes, ["目录/子/"]);
assert!(page.is_truncated);
assert_eq!(page.next_continuation_token.as_deref(), Some("opaque+/="));
assert_requests(&recorded, &[("GET", "/legacy?restype=container&comp=list&maxresults=2")]);
}
#[tokio::test]
async fn encoded_listing_preserves_required_field_and_entry_validation() {
for (entry, expected_error) in [
(
r#"<Blob><Name Encoded="true">a%2Fb</Name></Blob>"#,
"source listing object has no valid size",
),
(
r#"<Blob><Name Encoded="true">a%2Fb</Name><Properties><Content-Length>-1</Content-Length></Properties></Blob>"#,
"source listing object has no valid size",
),
(
r#"<Blob><Name Encoded="true">a%2Fb</Name><Name>a/b</Name><Properties><Content-Length>1</Content-Length></Properties></Blob>"#,
"source listing object has duplicate names",
),
(
r#"<Blob><Name Encoded="true">a%2Fb</Name><Properties><Content-Length>1</Content-Length><Content-Length>2</Content-Length></Properties></Blob>"#,
"source listing object has duplicate sizes",
),
(
r#"<BlobPrefix><Name Encoded="true">a%2F</Name><Name>a/</Name></BlobPrefix>"#,
"source listing prefix has duplicate names",
),
(r#"<BlobPrefix><Name Encoded="true" /></BlobPrefix>"#, "source listing prefix has no name"),
(
r#"<Blob><Name Encoded="true">a%2Fb</Name><Properties><Content-Length>1</Content-Length></Properties><Blob><Name Encoded="true">c%2Fd</Name><Properties><Content-Length>2</Content-Length></Properties></Blob></Blob>"#,
"source listing entries must not be nested",
),
(
r#"<BlobPrefix><Name Encoded="true">a%2F</Name><BlobPrefix><Name Encoded="true">b%2F</Name></BlobPrefix></BlobPrefix>"#,
"source listing entries must not be nested",
),
] {
let body = format!(
r#"<EnumerationResults><Blobs><Blob><Name Encoded="true">valid%252F</Name><Properties><Content-Length>0</Content-Length></Properties></Blob>{entry}</Blobs><NextMarker Encoded="true">opaque%2B+marker</NextMarker></EnumerationResults>"#
);
let (endpoint, recorded) = scripted_server(vec![ScriptedResponse::new(200, Vec::new(), body)]).await;
let err = backend(&endpoint, Credential::SharedKey(vec![7_u8; 32]))
.list(&SourceListRequest {
delimiter: Some("/"),
continuation_token: Some("opaque%2B+marker"),
max_keys: 2,
..Default::default()
})
.await
.expect_err("encoded names cannot bypass whole-page validation");
assert!(!err.is_retryable(), "{entry}: {err:?}");
let SourceError::Other(message) = err else {
panic!("wrong error class for {entry}: {err:?}");
};
assert_eq!(message, expected_error, "{entry}");
assert_requests(
&recorded,
&[(
"GET",
"/legacy?restype=container&comp=list&delimiter=%2F&marker=opaque%252B%2Bmarker&maxresults=2",
)],
);
}
}
#[test]
fn blob_tags_parse_into_the_shared_tag_map() {
let tags = parse_blob_tags(TAGS).expect("tags should parse");
@@ -1251,6 +1471,184 @@ mod tests {
]
}
#[tokio::test]
async fn object_not_found_requires_provider_evidence_or_one_successful_head_probe() {
for method in [Method::HEAD, Method::GET] {
for (status, code, expected) in [
(404, Some("BlobNotFound"), "not_found"),
(403, Some("BlobNotFound"), "access_denied"),
(404, Some("ContainerNotFound"), "other"),
(404, Some("BlobVersionNotFound"), "other"),
(404, Some("UnrecognizedError"), "other"),
(404, None, if method == Method::HEAD { "not_found" } else { "other" }),
(404, Some("ResourceNotFound"), if method == Method::HEAD { "not_found" } else { "other" }),
] {
let probes = method == Method::HEAD && status == 404 && matches!(code, None | Some("ResourceNotFound"));
let headers = code
.map(|value| vec![(HEADER_ERROR_CODE, value.to_string())])
.unwrap_or_default();
let mut responses = vec![ScriptedResponse::new(status, headers, "untrusted-error-body".to_string())];
if probes {
responses.push(ScriptedResponse::new(200, Vec::new(), String::new()));
}
let (endpoint, recorded) = scripted_server(responses).await;
let backend = backend(&endpoint, Credential::SharedKey(vec![7_u8; 32]));
let result = if method == Method::HEAD {
backend.head("missing").await.map(|_| ())
} else {
backend.get("missing", None).await.map(|_| ())
};
let err = result.expect_err("object error must remain an error");
assert_eq!(err.class_label(), expected, "{method} {status} {code:?}: {err:?}");
assert!(!err.is_retryable(), "{err:?}");
assert!(!err.to_string().contains("untrusted-error-body"));
let mut requests = vec![(method.as_str(), "/legacy/missing")];
if probes {
requests.push(("HEAD", "/legacy?restype=container"));
}
assert_requests(&recorded, &requests);
}
}
}
#[tokio::test]
async fn s3_not_found_alias_never_proves_native_object_absence() {
for selector in [None, Some("versionid"), Some("snapshot")] {
for operation in ["head", "get", "list", "tags", "probe"] {
if selector.is_some() && !matches!(operation, "head" | "get") {
continue;
}
for (status, expected, retryable) in [
(403, "access_denied", false),
(404, "other", false),
(416, "other", false),
(500, "server_error", true),
] {
let (endpoint, recorded) = scripted_server(vec![ScriptedResponse::new(
status,
vec![(HEADER_ERROR_CODE, "NoSuchKey".to_string())],
"untrusted-error-body".to_string(),
)])
.await;
let credential = selector.map_or_else(
|| Credential::SharedKey(vec![7_u8; 32]),
|selector| Credential::Sas(vec![(selector.to_string(), "old-version".to_string())]),
);
let backend = backend(&endpoint, credential);
let result = match operation {
"head" => backend.head("missing").await.map(|_| ()),
"get" => backend.get("missing", None).await.map(|_| ()),
"list" => backend.list(&SourceListRequest::default()).await.map(|_| ()),
"tags" => backend.tagging("missing").await.map(|_| ()),
"probe" => backend.probe().await,
_ => unreachable!(),
};
let err = result.expect_err("an S3 error alias is not Azure absence evidence");
assert_eq!(err.class_label(), expected, "{operation} {selector:?} HTTP {status}: {err:?}");
assert_eq!(err.is_retryable(), retryable, "{operation} {selector:?} HTTP {status}: {err:?}");
if status == 500 {
assert!(matches!(err, SourceError::ServerError(500)));
}
assert!(!err.to_string().contains("untrusted-error-body"));
let (method, mut target) = match operation {
"head" => ("HEAD", "/legacy/missing".to_string()),
"get" => ("GET", "/legacy/missing".to_string()),
"list" => ("GET", "/legacy?restype=container&comp=list".to_string()),
"tags" => ("GET", "/legacy/missing?comp=tags".to_string()),
"probe" => ("HEAD", "/legacy?restype=container".to_string()),
_ => unreachable!(),
};
if let Some(selector) = selector {
target.push_str(&format!("?{selector}=old-version"));
}
assert_requests(&recorded, &[(method, target.as_str())]);
}
}
}
}
#[tokio::test]
async fn ambiguous_head_preserves_the_container_probe_failure() {
for (status, expected, retryable) in [
(403, "access_denied", false),
(404, "other", false),
(429, "throttled", true),
(500, "server_error", true),
(503, "throttled", true),
] {
let (endpoint, recorded) = scripted_server(vec![
ScriptedResponse::new(404, Vec::new(), String::new()),
// A BlobNotFound header on a container request cannot prove
// that the object is missing, regardless of this status.
ScriptedResponse::new(status, vec![(HEADER_ERROR_CODE, "BlobNotFound".to_string())], String::new()),
])
.await;
let err = backend(&endpoint, Credential::SharedKey(vec![7_u8; 32]))
.head("missing")
.await
.expect_err("failed probe must not become object absence");
assert_eq!(err.class_label(), expected, "probe {status}: {err:?}");
assert_eq!(err.is_retryable(), retryable, "probe {status}: {err:?}");
if status == 500 {
assert!(matches!(err, SourceError::ServerError(500)));
}
assert_requests(&recorded, &[("HEAD", "/legacy/missing"), ("HEAD", "/legacy?restype=container")]);
}
}
#[tokio::test]
async fn version_and_snapshot_absence_are_not_missing_current_blobs() {
for selector in ["versionid", "snapshot"] {
for code in [None, Some("BlobNotFound"), Some("ResourceNotFound")] {
for method in [Method::HEAD, Method::GET] {
let headers = code
.map(|value| vec![(HEADER_ERROR_CODE, value.to_string())])
.unwrap_or_default();
let (endpoint, recorded) = scripted_server(vec![ScriptedResponse::new(404, headers, String::new())]).await;
let backend = backend(&endpoint, Credential::Sas(vec![(selector.to_string(), "old-version".to_string())]));
let result = if method == Method::HEAD {
backend.head("object").await.map(|_| ())
} else {
backend.get("object", None).await.map(|_| ())
};
let err = result.expect_err("missing selected version must remain a source error");
assert!(matches!(err, SourceError::Other(_)), "{method} {selector} {code:?}: {err:?}");
assert_requests(&recorded, &[(method.as_str(), &format!("/legacy/object?{selector}=old-version"))]);
}
}
}
}
#[tokio::test]
async fn blob_not_found_header_is_not_object_absence_for_list_or_tags() {
for tags in [false, true] {
let (endpoint, recorded) = scripted_server(vec![ScriptedResponse::new(
404,
vec![(HEADER_ERROR_CODE, "BlobNotFound".to_string())],
String::new(),
)])
.await;
let backend = backend(&endpoint, Credential::SharedKey(vec![7_u8; 32]));
let result = if tags {
backend.tagging("missing").await.map(|_| ())
} else {
backend.list(&SourceListRequest::default()).await.map(|_| ())
};
assert!(matches!(result, Err(SourceError::Other(_))), "tags={tags}: {result:?}");
assert_requests(
&recorded,
&[(
"GET",
if tags {
"/legacy/missing?comp=tags"
} else {
"/legacy?restype=container&comp=list"
},
)],
);
}
}
#[tokio::test]
async fn azure_backend_satisfies_the_shared_backend_contract() {
let mut ranged = contract_blob_headers();
@@ -1258,7 +1656,7 @@ mod tests {
// A HEAD reports the object size with no body, exactly as Azure does.
let mut head_only = contract_blob_headers();
head_only.push(("Content-Length", "5".to_string()));
let (endpoint, _) = scripted_server(vec![
let (endpoint, recorded) = scripted_server(vec![
ScriptedResponse::new(200, head_only, String::new()),
ScriptedResponse::new(200, contract_blob_headers(), "hello".to_string()),
ScriptedResponse::new(206, ranged, "ell".to_string()),
@@ -1288,6 +1686,23 @@ mod tests {
},
)
.await;
assert_requests(
&recorded,
&[
("HEAD", "/legacy/dir/a.txt"),
("GET", "/legacy/dir/a.txt"),
("GET", "/legacy/dir/a.txt"),
("GET", "/legacy?restype=container&comp=list&prefix=dir%2F&delimiter=%2F&maxresults=2"),
(
"GET",
"/legacy?restype=container&comp=list&prefix=dir%2F&delimiter=%2F&marker=cursor-1&maxresults=2",
),
("GET", "/legacy/dir/a.txt?comp=tags"),
("HEAD", "/legacy?restype=container"),
("HEAD", "/legacy/missing"),
("HEAD", "/legacy/secret"),
],
);
}
#[tokio::test]
rustfs/src/on_demand_migration/config.rs
+45 -2
View File
@@ -106,12 +106,12 @@ pub struct SourceConfig {
pub tls: TlsConfig,
/// Required for [`Provider::Azure`] and rejected for every other
/// provider.
#[serde(default)]
#[serde(default, skip_serializing_if = "Option::is_none")]
pub azure: Option<AzureSourceConfig>,
/// Required for [`Provider::GcsNative`] and rejected for every other
/// provider. [`Provider::Gcs`] keeps using `credentials` because it
/// speaks the S3 interoperability API.
#[serde(default)]
#[serde(default, skip_serializing_if = "Option::is_none")]
pub gcs: Option<GcsSourceConfig>,
}
@@ -808,6 +808,10 @@ impl EndpointKey {
mod tests {
use super::*;
mod before_native_sources {
include!("../../fixtures/on_demand_migration/source_config_e2a.rs");
}
const FULL_JSON: &str = r#"{
"version": 1,
"enabled": true,
@@ -878,6 +882,32 @@ mod tests {
assert_eq!(minimal.policy.source_timeout.first_byte_ms, 15_000);
}
#[test]
fn s3_config_writes_remain_readable_by_the_strict_pre_native_reader() {
// FULL_JSON is the complete config fixture already present in e2a921bc.
for provider in ["s3", "aws", "minio", "rustfs", "r2", "gcs"] {
let mut old_wire: serde_json::Value = serde_json::from_str(FULL_JSON).expect("historical config fixture");
old_wire["source"]["provider"] = provider.into();
let config = OnDemandMigrationConfig::from_json(&serde_json::to_vec(&old_wire).expect("historical wire"))
.expect("current reader accepts the historical source");
let wire = config.to_json().expect("persist current config");
let actual: serde_json::Value = serde_json::from_slice(&wire).expect("persisted config JSON");
let old_source: before_native_sources::SourceConfig = serde_json::from_value(actual["source"].clone())
.expect("an existing S3 source must remain readable by the strict e2a source consumer");
assert_eq!(serde_json::to_value(old_source).expect("old reader wire"), old_wire["source"]);
assert_eq!(actual, old_wire, "provider={provider}: no existing config field or value may change");
for field in ["azure", "gcs"] {
let mut rejected = old_wire["source"].clone();
rejected[field] = serde_json::Value::Null;
assert!(
serde_json::from_value::<before_native_sources::SourceConfig>(rejected).is_err(),
"the frozen old reader must reject {field}, even when null"
);
}
}
}
#[test]
fn unknown_fields_are_rejected_at_every_level() {
for (label, json) in [
@@ -1080,6 +1110,19 @@ mod tests {
for cfg in [azure_cfg(), gcs_native_cfg()] {
let json = cfg.to_json().expect("config must serialize");
assert_eq!(OnDemandMigrationConfig::from_json(&json).expect("config must parse"), cfg);
let wire: serde_json::Value = serde_json::from_slice(&json).expect("native config JSON");
let (present, absent, expected) = match cfg.source.provider {
Provider::Azure => ("azure", "gcs", serde_json::to_value(&cfg.source.azure).expect("Azure block")),
Provider::GcsNative => ("gcs", "azure", serde_json::to_value(&cfg.source.gcs).expect("GCS block")),
_ => unreachable!("native fixture"),
};
assert!(expected.is_object(), "native credentials must be present");
assert_eq!(wire["source"][present], expected);
assert!(wire["source"].get(absent).is_none());
assert!(
serde_json::from_value::<before_native_sources::SourceConfig>(wire["source"].clone()).is_err(),
"native providers still require upgraded readers"
);
}
// The wire labels are part of the admin contract.
assert!(
rustfs/src/on_demand_migration/gcs.rs
+209 -12
View File
@@ -55,10 +55,6 @@ use url::Url;
/// Read-only object scope: this backend never writes to the source.
const READ_ONLY_SCOPE: &str = "https://www.googleapis.com/auth/devstorage.read_only";
const METADATA_PREFIX: &str = "x-goog-meta-";
/// GCS reports its error code in the response body, not a header; the shared
/// transport takes a header name, so it is given one that never matches and
/// classification falls back to the status.
const NO_ERROR_CODE_HEADER: &str = "x-goog-unused-error-code";
/// One `objects.list` page is small; refuse an unbounded document.
const MAX_JSON_BYTES: usize = 8 * 1024 * 1024;
@@ -125,7 +121,7 @@ impl GcsNativeSourceBackend {
}
async fn send_object(&self, request: reqwest::Request) -> Result<reqwest::Response, SourceError> {
match self.http.send_object(request, NO_ERROR_CODE_HEADER).await {
match self.http.send_object(request, None).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.
@@ -225,7 +221,7 @@ impl SourceBackend for GcsNativeSourceBackend {
}
let request = self.request(Method::GET, url, HeaderMap::new()).await?;
let response = self.http.send(request, NO_ERROR_CODE_HEADER).await?;
let response = self.http.send(request, None).await?;
let body = read_text(response, MAX_JSON_BYTES).await?;
parse_objects_list(&body)
}
@@ -245,7 +241,7 @@ impl SourceBackend for GcsNativeSourceBackend {
let mut url = self.objects_url()?;
url.query_pairs_mut().append_pair("maxResults", "1");
let request = self.request(Method::GET, url, HeaderMap::new()).await?;
let response = self.http.send(request, NO_ERROR_CODE_HEADER).await?;
let response = self.http.send(request, None).await?;
read_text(response, MAX_JSON_BYTES)
.await
.and_then(|body| parse_objects_list(&body))?;
@@ -284,28 +280,38 @@ struct ListedObject {
fn parse_objects_list(body: &str) -> Result<SourcePage, SourceError> {
let listing: ObjectsList =
serde_json::from_str(body).map_err(|err| SourceError::Other(format!("source listing is not valid JSON: {err}")))?;
if listing.prefixes.iter().any(|prefix| prefix.is_empty()) {
return Err(SourceError::Other("source listing prefix has no name".to_string()));
}
let next_continuation_token = listing.next_page_token.filter(|token| !token.is_empty());
let objects = listing
.items
.into_iter()
.map(|item| {
if item.name.is_empty() {
return Err(SourceError::Other("source listing object has no name".to_string()));
}
let size = item
.size
.and_then(|size| size.parse::<u64>().ok())
.ok_or_else(|| SourceError::Other("source listing object has no valid size".to_string()))?;
let etag = item
.md5_hash
.as_deref()
.and_then(base64_md5_to_hex)
.or_else(|| item.etag.map(|etag| etag.trim_matches('"').to_string()))
.filter(|etag| !etag.is_empty());
SourceObject {
Ok(SourceObject {
key: item.name,
etag,
size: item.size.and_then(|size| size.parse().ok()).unwrap_or(0),
size,
last_modified: item.updated.as_deref().and_then(parse_http_timestamp),
storage_class: item.storage_class,
// GCS never encodes a part count in a digest or an ETag.
is_multipart_etag: false,
}
})
})
.collect();
.collect::<Result<_, SourceError>>()?;
Ok(SourcePage {
objects,
@@ -319,7 +325,7 @@ fn parse_objects_list(body: &str) -> Result<SourcePage, SourceError> {
mod tests {
use super::*;
use crate::on_demand_migration::backend_contract::{BackendCapabilities, assert_backend_contract};
use crate::on_demand_migration::test_http_fixture::{ScriptedResponse, scripted_server};
use crate::on_demand_migration::test_http_fixture::{ScriptedResponse, assert_requests, scripted_server};
use google_cloud_auth::credentials::anonymous::Builder as AnonymousBuilder;
const LIST_PAGE_ONE: &str = r#"{
@@ -573,6 +579,197 @@ mod tests {
}
}
#[tokio::test]
async fn native_listing_rejects_missing_or_invalid_required_object_fields() {
for entry in [
r#"{"size":"1"}"#,
r#"{"name":"","size":"1"}"#,
r#"{"name":"broken"}"#,
r#"{"name":"broken","size":null}"#,
r#"{"name":"broken","size":""}"#,
r#"{"name":"broken","size":"-1"}"#,
r#"{"name":"broken","size":"18446744073709551616"}"#,
r#"{"name":"broken","size":"not-a-size"}"#,
r#"{"name":"broken","size":1}"#,
] {
let body = format!(r#"{{"items":[{{"name":"valid","size":"1"}},{entry}],"nextPageToken":"next"}}"#);
let (endpoint, recorded) = scripted_server(vec![ScriptedResponse::new(200, Vec::new(), body)]).await;
let err = backend(&endpoint)
.list(&SourceListRequest {
prefix: Some("dir/"),
delimiter: Some("/"),
continuation_token: Some("opaque+/="),
max_keys: 2,
..Default::default()
})
.await
.expect_err("malformed object must reject the complete native page");
assert!(matches!(err, SourceError::Other(_)), "{entry}: {err:?}");
assert!(!err.is_retryable());
assert_requests(
&recorded,
&[(
"GET",
"/storage/v1/b/legacy/o?prefix=dir%2F&delimiter=%2F&pageToken=opaque%2B%2F%3D&maxResults=2",
)],
);
}
}
#[tokio::test]
async fn native_listing_rejects_empty_prefix_entries() {
for body in [
r#"{"items":[{"name":"valid","size":"1"}],"prefixes":[""],"nextPageToken":"next"}"#,
r#"{"prefixes":[""],"nextPageToken":"next"}"#,
r#"{"prefixes":["目录/子/",""],"nextPageToken":"next"}"#,
] {
let (endpoint, recorded) = scripted_server(vec![ScriptedResponse::new(200, Vec::new(), body.to_string())]).await;
let result = backend(&endpoint)
.list(&SourceListRequest {
delimiter: Some("/"),
continuation_token: Some("opaque+/="),
max_keys: 2,
..Default::default()
})
.await;
let err = result.expect_err("an empty prefix must reject the entire page and its cursor");
assert!(matches!(err, SourceError::Other(_)), "{body}: {err:?}");
assert!(!err.is_retryable());
assert_requests(
&recorded,
&[("GET", "/storage/v1/b/legacy/o?delimiter=%2F&pageToken=opaque%2B%2F%3D&maxResults=2")],
);
}
let body = r#"{"prefixes":["目录/子/"],"nextPageToken":"opaque+/="}"#;
let (endpoint, recorded) = scripted_server(vec![ScriptedResponse::new(200, Vec::new(), body.to_string())]).await;
let page = backend(&endpoint)
.list(&SourceListRequest {
delimiter: Some("/"),
max_keys: 1,
..Default::default()
})
.await
.expect("a valid prefix-only page must remain usable");
assert!(page.objects.is_empty());
assert_eq!(page.common_prefixes, ["目录/子/"]);
assert!(page.is_truncated);
assert_eq!(page.next_continuation_token.as_deref(), Some("opaque+/="));
assert_requests(&recorded, &[("GET", "/storage/v1/b/legacy/o?delimiter=%2F&maxResults=1")]);
}
#[tokio::test]
async fn native_listing_preserves_zero_size_unicode_prefixes_and_opaque_cursors() {
let body = r#"{"items":[{"name":"目录/空 & file","size":"0"}],"prefixes":["目录/子/"],"nextPageToken":"opaque+/="}"#;
let (endpoint, recorded) = scripted_server(vec![ScriptedResponse::new(200, Vec::new(), body.to_string())]).await;
let page = backend(&endpoint)
.list(&SourceListRequest {
max_keys: 2,
..Default::default()
})
.await
.expect("valid native page");
assert_eq!(page.objects.len(), 1);
assert_eq!(page.objects[0].key, "目录/空 & file");
assert_eq!(page.objects[0].size, 0);
assert_eq!(page.common_prefixes, ["目录/子/"]);
assert!(page.is_truncated);
assert_eq!(page.next_continuation_token.as_deref(), Some("opaque+/="));
assert_requests(&recorded, &[("GET", "/storage/v1/b/legacy/o?maxResults=2")]);
}
#[tokio::test]
async fn missing_object_head_requires_one_successful_bucket_probe() {
for (status, body, expected, retryable) in [
(200, "{}", "not_found", false),
(403, "", "access_denied", false),
(404, "", "other", false),
(429, "", "throttled", true),
(500, "", "server_error", true),
(503, "", "throttled", true),
(200, "not JSON", "other", false),
] {
let (endpoint, recorded) = scripted_server(vec![
ScriptedResponse::new(404, Vec::new(), String::new()),
ScriptedResponse::new(status, Vec::new(), body.to_string()),
])
.await;
let err = backend(&endpoint).head("missing").await.expect_err("missing HEAD must fail");
assert_eq!(err.class_label(), expected, "probe {status} {body:?}: {err:?}");
assert_eq!(err.is_retryable(), retryable, "probe {status} {body:?}: {err:?}");
if status == 500 {
assert!(matches!(err, SourceError::ServerError(500)));
}
assert_requests(&recorded, &[("HEAD", "/legacy/missing"), ("GET", "/storage/v1/b/legacy/o?maxResults=1")]);
}
}
#[tokio::test]
async fn denied_object_reads_do_not_probe_or_become_object_absence() {
for method in [Method::HEAD, Method::GET] {
let (endpoint, recorded) = scripted_server(vec![ScriptedResponse::new(
403,
vec![("x-goog-unused-error-code", "NoSuchKey".to_string())],
"untrusted-error-body".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 err = result.expect_err("denied object read must remain a failure");
assert_eq!(err.class_label(), "access_denied");
assert!(!err.is_retryable());
assert!(!err.to_string().contains("untrusted-error-body"));
assert_requests(&recorded, &[(method.as_str(), "/legacy/missing")]);
}
}
#[tokio::test]
async fn non_object_errors_ignore_untrusted_error_code_headers() {
for probe in [false, true] {
for (status, expected, retryable) in [(403, "access_denied", false), (500, "server_error", true)] {
let (endpoint, recorded) = scripted_server(vec![ScriptedResponse::new(
status,
vec![("x-goog-unused-error-code", "NoSuchKey".to_string())],
"untrusted-error-body".to_string(),
)])
.await;
let backend = backend(&endpoint);
let result = if probe {
backend.probe().await
} else {
backend
.list(&SourceListRequest {
max_keys: 2,
..Default::default()
})
.await
.map(|_| ())
};
let err = result.expect_err("a synthetic provider header cannot change the source status");
assert_eq!(err.class_label(), expected, "probe={probe} status={status}: {err:?}");
assert_eq!(err.is_retryable(), retryable);
assert!(!err.to_string().contains("untrusted-error-body"));
if status == 500 {
assert!(matches!(err, SourceError::ServerError(500)));
}
assert_requests(
&recorded,
&[(
"GET",
if probe {
"/storage/v1/b/legacy/o?maxResults=1"
} else {
"/storage/v1/b/legacy/o?maxResults=2"
},
)],
);
}
}
}
/// GCS states its error code in the response body, which this backend never
/// reads, so every class must follow from the status alone. The classes are
/// what the runtime acts on: only `NotFound` is negative-cached, and only a
rustfs/src/on_demand_migration/list_through.rs
+294 -33
View File
@@ -94,13 +94,16 @@ pub struct MergePick {
}
/// The continuation-token envelope. Opaque to clients: it is serialized as
/// framed JSON and then base64-encoded by the same helper as a local marker.
/// JSON, optionally framed, then base64-encoded like a local marker.
///
/// A `null` cursor with `done = false` means "list that side from the start";
/// `done = true` means the side is finished and must not be listed again.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ListThroughToken {
/// Transport framing observed by the decoder, never an envelope field.
#[serde(skip)]
pub framed: bool,
/// Envelope marker, always [`LIST_THROUGH_TOKEN_TAG`].
pub t: String,
pub v: u32,
@@ -127,6 +130,7 @@ pub struct ListThroughToken {
impl ListThroughToken {
fn new(local: SideCursor, source: SideCursor, last_key: Option<String>) -> Self {
Self {
framed: false,
t: LIST_THROUGH_TOKEN_TAG.to_string(),
v: LIST_THROUGH_TOKEN_VERSION,
local: local.token,
@@ -141,7 +145,12 @@ impl ListThroughToken {
pub fn encode(&self) -> String {
// The envelope is built here from owned strings, so serialization
// cannot fail; the fallback keeps the signature infallible.
format!("{LIST_THROUGH_TOKEN_PREFIX}{}", serde_json::to_string(self).unwrap_or_default())
let json = serde_json::to_string(self).unwrap_or_default();
if self.framed {
format!("{LIST_THROUGH_TOKEN_PREFIX}{json}")
} else {
json
}
}
}
@@ -165,16 +174,30 @@ pub enum ListThroughTokenError {
/// Classifies an already base64-decoded continuation token.
///
/// Only a framed JSON object is read as a merged token;
/// anything else is a local marker, so a bucket that turns `list_through` off
/// keeps paginating with the tokens it handed out. A token that *is* an
/// envelope but was tampered with (unknown version, unknown field, truncated
/// JSON) is an error, never a silent fallback.
/// Framed envelopes and complete historical writer envelopes are merged tokens.
/// Partial JSON-shaped keys remain local markers. A key identical to a complete
/// historical envelope is inherently ambiguous and retains merged semantics.
/// Recognized envelopes share the same version, count and field validation.
pub fn decode_continuation_token(decoded: &str) -> Result<ListThroughCursor, ListThroughTokenError> {
let Some(payload) = decoded.strip_prefix(LIST_THROUGH_TOKEN_PREFIX) else {
return Ok(ListThroughCursor::Local(decoded.to_string()));
let (payload, framed) = match decoded.strip_prefix(LIST_THROUGH_TOKEN_PREFIX) {
Some(payload) => (payload, true),
None if decoded.starts_with('{') => (decoded, false),
None => return Ok(ListThroughCursor::Local(decoded.to_string())),
};
let value = serde_json::from_str::<serde_json::Value>(payload).map_err(|_| ListThroughTokenError::Malformed)?;
let value = match serde_json::from_str::<serde_json::Value>(payload) {
Ok(value) => value,
Err(_) if framed => return Err(ListThroughTokenError::Malformed),
Err(_) => return Ok(ListThroughCursor::Local(decoded.to_string())),
};
// RUSTFS_COMPAT_TODO(odm-list-bare-envelope): old writers issued bare JSON. Remove after all supported readers understand framing and outstanding bare listings have drained or explicitly restarted.
if !framed
&& (value.get("t").and_then(serde_json::Value::as_str) != Some(LIST_THROUGH_TOKEN_TAG)
|| ["v", "local", "local_done", "source", "source_done", "last_key"]
.iter()
.any(|field| value.get(field).is_none()))
{
return Ok(ListThroughCursor::Local(decoded.to_string()));
}
if value.get("t").and_then(serde_json::Value::as_str) != Some(LIST_THROUGH_TOKEN_TAG) {
return Err(ListThroughTokenError::Malformed);
}
@@ -198,7 +221,10 @@ pub fn decode_continuation_token(decoded: &str) -> Result<ListThroughCursor, Lis
None => return Err(ListThroughTokenError::Malformed),
}
serde_json::from_value::<ListThroughToken>(value)
.map(|token| ListThroughCursor::Merged(Box::new(token)))
.map(|mut token| {
token.framed = framed;
ListThroughCursor::Merged(Box::new(token))
})
.map_err(|_| ListThroughTokenError::Malformed)
}
@@ -643,6 +669,126 @@ impl Default for SourceListRateLimiter {
}
}
// Frozen framed-only codec from e1608fbd9ca934d157b5de46c80b4393f2dd3dd6.
// Keep its own DTO and constants: current-reader round trips cannot establish
// whether a deployed framed-only reader accepts the bytes we issue.
#[cfg(test)]
pub(crate) mod e160_framed_reader {
use serde::{Deserialize, Serialize};
/// The continuation-token version used by ordinary progressing pages.
pub const LIST_THROUGH_TOKEN_VERSION: u32 = 1;
const LIST_THROUGH_PROGRESS_TOKEN_VERSION: u32 = 2;
/// The sixteenth consecutive merged page without a key or new EOF fails.
/// This also bounds legitimate sparse listings; it is not a cycle detector.
pub const MAX_LIST_NO_PROGRESS_PAGES: u8 = 16;
/// Envelope marker. A bucket that is *not* merging hands out the local
/// listing's own marker, so the decoder needs a positive signal before it
/// treats an opaque token as a merged one.
const LIST_THROUGH_TOKEN_TAG: &str = "odm-list";
// Object keys cannot contain NUL (bucket::utils::is_valid_object_prefix),
// so this framing cannot collide with a local key used as an opaque marker.
const LIST_THROUGH_TOKEN_PREFIX: &str = "\0odm-list:";
/// The continuation-token envelope. Opaque to clients: it is serialized as
/// framed JSON and then base64-encoded by the same helper as a local marker.
///
/// A `null` cursor with `done = false` means "list that side from the start";
/// `done = true` means the side is finished and must not be listed again.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ListThroughToken {
/// Envelope marker, always [`LIST_THROUGH_TOKEN_TAG`].
pub t: String,
pub v: u32,
#[serde(default)]
pub local: Option<String>,
#[serde(default)]
pub local_done: bool,
#[serde(default)]
pub source: Option<String>,
#[serde(default)]
pub source_done: bool,
/// Last entry the previous page consumed. A side whose page was only
/// partially consumed is re-listed from the same cursor and everything at
/// or below this key is dropped, which is delimiter-safe: a rolled-up
/// common prefix compares as itself, never as its members.
#[serde(default)]
pub last_key: Option<String>,
/// Consecutive empty truncated merged pages, present only in v2 tokens.
/// Ordinary v1 tokens retain their original serialized shape.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub no_progress: Option<u8>,
}
impl ListThroughToken {
pub fn encode(&self) -> String {
// The envelope is built here from owned strings, so serialization
// cannot fail; the fallback keeps the signature infallible.
format!("{LIST_THROUGH_TOKEN_PREFIX}{}", serde_json::to_string(self).unwrap_or_default())
}
}
/// What a decoded (base64-stripped) continuation token turned out to be.
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum ListThroughCursor {
/// A plain local listing marker: the bucket was not merging when the token
/// was issued, or the client is paginating a non-merged listing.
Local(String),
Merged(Box<ListThroughToken>),
}
#[derive(Clone, Debug, PartialEq, Eq, thiserror::Error)]
pub enum ListThroughTokenError {
#[error("continuation token version {0} is not supported")]
UnsupportedVersion(u32),
/// The message never echoes the token: it is client-controlled input.
#[error("continuation token is malformed")]
Malformed,
}
/// Classifies an already base64-decoded continuation token.
///
/// Only a framed JSON object is read as a merged token;
/// anything else is a local marker, so a bucket that turns `list_through` off
/// keeps paginating with the tokens it handed out. A token that *is* an
/// envelope but was tampered with (unknown version, unknown field, truncated
/// JSON) is an error, never a silent fallback.
pub fn decode_continuation_token(decoded: &str) -> Result<ListThroughCursor, ListThroughTokenError> {
let Some(payload) = decoded.strip_prefix(LIST_THROUGH_TOKEN_PREFIX) else {
return Ok(ListThroughCursor::Local(decoded.to_string()));
};
let value = serde_json::from_str::<serde_json::Value>(payload).map_err(|_| ListThroughTokenError::Malformed)?;
if value.get("t").and_then(serde_json::Value::as_str) != Some(LIST_THROUGH_TOKEN_TAG) {
return Err(ListThroughTokenError::Malformed);
}
match value.get("v").and_then(serde_json::Value::as_u64) {
Some(version) if version == u64::from(LIST_THROUGH_TOKEN_VERSION) => {
// v1 readers reject this field even when it is null or zero.
if value.get("no_progress").is_some() {
return Err(ListThroughTokenError::Malformed);
}
}
Some(version) if version == u64::from(LIST_THROUGH_PROGRESS_TOKEN_VERSION) => {
if !value
.get("no_progress")
.and_then(serde_json::Value::as_u64)
.is_some_and(|count| (1..u64::from(MAX_LIST_NO_PROGRESS_PAGES)).contains(&count))
{
return Err(ListThroughTokenError::Malformed);
}
}
Some(version) => return Err(ListThroughTokenError::UnsupportedVersion(version.min(u64::from(u32::MAX)) as u32)),
None => return Err(ListThroughTokenError::Malformed),
}
serde_json::from_value::<ListThroughToken>(value)
.map(|token| ListThroughCursor::Merged(Box::new(token)))
.map_err(|_| ListThroughTokenError::Malformed)
}
}
#[cfg(test)]
mod tests {
use super::*;
@@ -797,6 +943,7 @@ mod tests {
#[test]
fn a_degraded_page_keeps_the_source_cursor_for_the_next_one() {
let resume = ListThroughToken {
framed: false,
t: LIST_THROUGH_TOKEN_TAG.to_string(),
v: LIST_THROUGH_TOKEN_VERSION,
local: Some("local-1".to_string()),
@@ -1033,7 +1180,7 @@ mod tests {
#[test]
fn token_round_trips_and_rejects_tampering() {
let token = ListThroughToken::new(
let mut token = ListThroughToken::new(
SideCursor {
token: Some("l".to_string()),
done: false,
@@ -1041,6 +1188,7 @@ mod tests {
SideCursor { token: None, done: true },
Some("k".to_string()),
);
token.framed = true;
let encoded = token.encode();
assert_eq!(decode_continuation_token(&encoded), Ok(ListThroughCursor::Merged(Box::new(token))));
@@ -1089,35 +1237,148 @@ 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(),
concat!(
"\0odm-list:",
r#"{"t":"odm-list","v":1,"local":null,"local_done":true,"source":"A","source_done":false,"last_key":"last-key"}"#
)
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);
assert_eq!(decode_continuation_token(&token.encode()), Ok(ListThroughCursor::Merged(Box::new(token))));
}
for version in [1, 2] {
for value in ["null", "0", "16", "-1", "1.5", "256", "18446744073709551616", "\"1\""] {
let encoded = framed(&format!(r#"{{"t":"odm-list","v":{version},"no_progress":{value}}}"#));
for framed in [false, true] {
let prefix = if framed { LIST_THROUGH_TOKEN_PREFIX } else { "" };
for count in 1..MAX_LIST_NO_PROGRESS_PAGES {
let mut token = progress_token(Some(count), true, false);
token.framed = framed;
assert_eq!(decode_continuation_token(&token.encode()), Ok(ListThroughCursor::Merged(Box::new(token))));
}
// Bare recognition requires the complete shape emitted by old writers;
// partial JSON objects are also valid local keys.
for version in [1, 2] {
for value in ["null", "0", "16", "-1", "1.5", "256", "18446744073709551616", "\"1\""] {
let encoded = format!(
r#"{prefix}{{"t":"odm-list","v":{version},"local":null,"local_done":true,"source":"A","source_done":false,"last_key":"last-key","no_progress":{value}}}"#
);
assert_eq!(decode_continuation_token(&encoded), Err(ListThroughTokenError::Malformed), "{encoded}");
}
}
for encoded in [
r#"{"t":"odm-list","v":1,"local":null,"local_done":true,"source":"A","source_done":false,"last_key":"last-key","no_progress":1}"#,
r#"{"t":"odm-list","v":2,"local":null,"local_done":true,"source":"A","source_done":false,"last_key":"last-key"}"#,
r#"{"t":"odm-list","v":2,"local":null,"local_done":true,"source":"A","source_done":false,"last_key":"last-key","no_progress":1,"extra":true}"#,
r#"{"t":"odm-list","v":2,"local":null,"local_done":true,"source":"A","source_done":false,"last_key":"last-key","no_progress":1,"framed":true}"#,
] {
let encoded = format!("{prefix}{encoded}");
assert_eq!(decode_continuation_token(&encoded), Err(ListThroughTokenError::Malformed), "{encoded}");
}
let bumped = format!("{prefix}{}", token.encode().replace("\"v\":1", "\"v\":9"));
assert_eq!(decode_continuation_token(&bumped), Err(ListThroughTokenError::UnsupportedVersion(9)));
}
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}"#,
}
// Frozen decoder from 447f3c704, before framing was introduced. Keeping this
// independent of the current decoder catches a default-writer rollout break.
fn decode_before_framing(decoded: &str) -> Result<ListThroughCursor, ListThroughTokenError> {
if !decoded.starts_with('{') {
return Ok(ListThroughCursor::Local(decoded.to_string()));
}
let Ok(value) = serde_json::from_str::<serde_json::Value>(decoded) else {
// Not JSON at all: an object key may legitimately start with '{'.
return Ok(ListThroughCursor::Local(decoded.to_string()));
};
if value.get("t").and_then(serde_json::Value::as_str) != Some(LIST_THROUGH_TOKEN_TAG) {
return Ok(ListThroughCursor::Local(decoded.to_string()));
}
match value.get("v").and_then(serde_json::Value::as_u64) {
Some(version) if version == u64::from(LIST_THROUGH_TOKEN_VERSION) => {
// v1 readers reject this field even when it is null or zero.
if value.get("no_progress").is_some() {
return Err(ListThroughTokenError::Malformed);
}
}
Some(version) if version == u64::from(LIST_THROUGH_PROGRESS_TOKEN_VERSION) => {
if !value
.get("no_progress")
.and_then(serde_json::Value::as_u64)
.is_some_and(|count| (1..u64::from(MAX_LIST_NO_PROGRESS_PAGES)).contains(&count))
{
return Err(ListThroughTokenError::Malformed);
}
}
Some(version) => return Err(ListThroughTokenError::UnsupportedVersion(version.min(u64::from(u32::MAX)) as u32)),
None => return Err(ListThroughTokenError::Malformed),
}
serde_json::from_value::<ListThroughToken>(value)
.map(|token| ListThroughCursor::Merged(Box::new(token)))
.map_err(|_| ListThroughTokenError::Malformed)
}
#[test]
fn historical_writer_fixtures_and_default_output_remain_readable() {
for (wire, version, count) in [
(
r#"{"t":"odm-list","v":1,"local":"local-2","local_done":false,"source":"source-2","source_done":false,"last_key":"k"}"#,
1,
None,
),
(
r#"{"t":"odm-list","v":2,"local":"local-2","local_done":false,"source":"source-2","source_done":false,"last_key":"k","no_progress":15}"#,
2,
Some(15),
),
] {
let encoded = framed(payload);
assert_eq!(decode_continuation_token(&encoded), Err(ListThroughTokenError::Malformed), "{encoded}");
let ListThroughCursor::Merged(mut token) = decode_continuation_token(wire).expect("historical issued token") else {
panic!("a historical cursor must not silently become a local marker, even if a key has identical JSON");
};
assert_eq!(token.local.as_deref(), Some("local-2"));
assert_eq!(token.source.as_deref(), Some("source-2"));
assert_eq!(token.last_key.as_deref(), Some("k"));
assert_eq!(token.v, version);
assert_eq!(token.no_progress, count);
assert!(!token.framed);
assert_eq!(token.encode(), wire, "bare output retains the historical bytes");
assert_eq!(decode_before_framing(&token.encode()), Ok(ListThroughCursor::Merged(token.clone())));
token.framed = true;
let framed = format!("\0odm-list:{wire}");
assert_eq!(token.encode(), framed, "framing leaves the JSON payload unchanged");
assert_eq!(decode_continuation_token(&framed), Ok(ListThroughCursor::Merged(token)));
}
}
#[test]
fn frozen_e160_reader_distinguishes_framing_and_keeps_strict_budget_validation() {
use super::e160_framed_reader as old;
for raw in [
r#"{"t":"odm-list","v":1,"local":"local-2","local_done":false,"source":"source-2","source_done":false,"last_key":"k"}"#,
r#"{"t":"odm-list","v":2,"local":"local-2","local_done":false,"source":"source-2","source_done":false,"last_key":"k","no_progress":15}"#,
] {
assert_eq!(old::decode_continuation_token(raw), Ok(old::ListThroughCursor::Local(raw.to_string())));
let framed = format!("\0odm-list:{raw}");
let old::ListThroughCursor::Merged(old_token) = old::decode_continuation_token(&framed).expect("old writer bytes")
else {
panic!("e160 recognizes its own frame");
};
assert_eq!(old_token.encode(), framed);
let ListThroughCursor::Merged(current) = decode_continuation_token(&framed).expect("dual reader") else {
panic!("dual readers preserve old framed chains");
};
assert_eq!(current.encode(), framed);
assert_eq!(current.local, old_token.local);
assert_eq!(current.local_done, old_token.local_done);
assert_eq!(current.source, old_token.source);
assert_eq!(current.source_done, old_token.source_done);
assert_eq!(current.last_key, old_token.last_key);
assert_eq!(current.v, old_token.v);
assert_eq!(current.no_progress, old_token.no_progress);
}
for count in ["null", "0", "16", "-1", "1.5", "\"1\"", "256"] {
let raw = format!(
"\0odm-list:{{\"t\":\"odm-list\",\"v\":2,\"local\":null,\"local_done\":true,\"source\":\"A\",\"source_done\":false,\"last_key\":null,\"no_progress\":{count}}}"
);
assert_eq!(
old::decode_continuation_token(&raw),
Err(old::ListThroughTokenError::Malformed),
"{count}"
);
assert_eq!(decode_continuation_token(&raw), Err(ListThroughTokenError::Malformed), "{count}");
}
}
rustfs/src/on_demand_migration/native_http.rs
+28 -11
View File
@@ -99,6 +99,7 @@ impl NativeHttp {
pub(super) fn for_test(endpoint: Url) -> Self {
Self {
client: reqwest::Client::builder()
.no_proxy()
.redirect(reqwest::redirect::Policy::none())
.build()
.expect("test http client should build"),
@@ -130,13 +131,13 @@ impl NativeHttp {
}
/// Sends the request and returns the response only for a 2xx status.
/// Non-2xx statuses are classified from the status and the provider's own
/// Non-2xx statuses are classified from the status and an optional provider
/// error-code header; response bodies are not read, so no provider message
/// can smuggle credentials or markup into a log line.
pub(super) async fn send(
&self,
request: reqwest::Request,
error_code_header: &str,
error_code_header: Option<&str>,
) -> Result<reqwest::Response, SourceError> {
self.send_classified(request, error_code_header, false).await
}
@@ -145,7 +146,7 @@ impl NativeHttp {
pub(super) async fn send_object(
&self,
request: reqwest::Request,
error_code_header: &str,
error_code_header: Option<&str>,
) -> Result<reqwest::Response, SourceError> {
self.send_classified(request, error_code_header, true).await
}
@@ -153,26 +154,42 @@ impl NativeHttp {
async fn send_classified(
&self,
request: reqwest::Request,
error_code_header: &str,
error_code_header: Option<&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 response = self.execute(request).await?;
let status = response.status();
match Self::check_response(response, error_code_header) {
Err(SourceError::Other(_)) if not_found_on_404_without_code && status.as_u16() == 404 => Err(SourceError::NotFound),
result => result,
}
}
pub(super) async fn execute(&self, request: reqwest::Request) -> Result<reqwest::Response, SourceError> {
self.client.execute(request).await.map_err(classify_transport_error)
}
pub(super) fn check_response(
response: reqwest::Response,
error_code_header: Option<&str>,
) -> Result<reqwest::Response, SourceError> {
let status = response.status();
if status.is_success() {
return Ok(response);
}
let code = response
.headers()
.get(error_code_header)
let code = error_code_header
.and_then(|header| response.headers().get(header))
.and_then(|value| value.to_str().ok())
.map(str::to_string);
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),
match classify_status(status.as_u16(), code.as_deref(), message.clone()) {
// Native object absence needs provider-specific evidence or a
// successful bucket probe, never an alias from the S3 classifier.
SourceError::NotFound => Err(classify_status(status.as_u16(), None, message)),
error => Err(error),
}
}
}
rustfs/src/on_demand_migration/source_client.rs
+39 -8
View File
@@ -337,7 +337,7 @@ const THROTTLE_CODES: &[&str] = &[
"RequestThrottled",
"ServerBusy",
];
const NOT_FOUND_CODES: &[&str] = &["NoSuchKey", "BlobNotFound"];
const NOT_FOUND_CODES: &[&str] = &["NoSuchKey"];
const ACCESS_DENIED_CODES: &[&str] = &[
"AccessDenied",
"InvalidAccessKeyId",
@@ -1813,10 +1813,11 @@ mod tests {
/// The S3 backend behind the scripted connector, without the prefix-mapping
/// client on top: the contract is a property of the backend itself.
async fn scripted_s3_backend(responses: Vec<Scripted>) -> S3SourceBackend {
async fn scripted_s3_backend(responses: Vec<Scripted>) -> (S3SourceBackend, Recorded) {
let spec = spec(None);
let requests: Recorded = Arc::new(Mutex::new(Vec::new()));
let connector = SharedHttpConnector::new(ScriptedConnector {
requests: Arc::new(Mutex::new(Vec::new())),
requests: Arc::clone(&requests),
responses: Arc::new(Mutex::new(responses.into_iter().collect())),
});
let http_client = http_client_fn(move |_settings, _components| connector.clone());
@@ -1826,17 +1827,20 @@ mod tests {
.expect("test spec should build")
.http_client(http_client)
.interceptor(SourceProxyMarkerInterceptor::new());
S3SourceBackend {
client: S3Client::from_conf(config.build()),
bucket: spec.bucket.clone(),
}
(
S3SourceBackend {
client: S3Client::from_conf(config.build()),
bucket: spec.bucket.clone(),
},
requests,
)
}
#[tokio::test]
async fn s3_backend_satisfies_the_shared_backend_contract() {
let mut ranged = contract_object_headers(3);
ranged.push(("content-range", "bytes 1-3/5".to_string()));
let backend = scripted_s3_backend(vec![
let (backend, requests) = scripted_s3_backend(vec![
ok(contract_object_headers(5), ""),
ok(contract_object_headers(5), "hello"),
ok(ranged, "ell"),
@@ -1845,6 +1849,7 @@ mod tests {
ok(Vec::new(), CONTRACT_TAGGING),
ok(Vec::new(), ""),
status(404, ""),
// An object HEAD 404 requires the existing S3 bucket HEAD probe.
ok(Vec::new(), ""),
status(403, ACCESS_DENIED_BODY),
])
@@ -1859,6 +1864,32 @@ mod tests {
},
)
.await;
let requests = recorded(&requests);
let actual: Vec<_> = requests
.iter()
.map(|request| {
(
request.method.as_str(),
url::Url::parse(&request.uri).expect("recorded S3 URL").path().to_string(),
)
})
.collect();
let expected = [
("HEAD", "/source-bucket/dir/a.txt"),
("GET", "/source-bucket/dir/a.txt"),
("GET", "/source-bucket/dir/a.txt"),
("GET", "/source-bucket/"),
("GET", "/source-bucket/"),
("GET", "/source-bucket/dir/a.txt"),
("HEAD", "/source-bucket/"),
("HEAD", "/source-bucket/missing"),
("HEAD", "/source-bucket/"),
("HEAD", "/source-bucket/secret"),
];
assert_eq!(actual, expected.map(|(method, path)| (method, path.to_string())));
for request in &requests {
assert_outbound_markers(request);
}
}
fn prefix_client(prefix: Option<String>) -> SourceClient {
rustfs/src/on_demand_migration/test_http_fixture.rs
+10
View File
@@ -56,6 +56,16 @@ impl RecordedRequest {
pub(super) type Recorder = Arc<Mutex<Vec<RecordedRequest>>>;
/// Checks the full request sequence, including the absence of extra probes.
pub(super) fn assert_requests(recorder: &Recorder, expected: &[(&str, &str)]) {
let recorded = recorder.lock().expect("recorder lock");
let actual: Vec<_> = recorded
.iter()
.map(|request| (request.method.as_str(), request.target.as_str()))
.collect();
assert_eq!(actual, expected, "unexpected native source request sequence");
}
/// Binds a loopback listener that answers `responses` in order and returns its
/// origin plus the recorder. The task ends once the script is exhausted.
pub(super) async fn scripted_server(responses: Vec<ScriptedResponse>) -> (Url, Recorder) {