Files
Genarrative/server-rs/crates/platform-oss/src/lib.rs
T
k88936 422c8931d6
Project CI / Repository checks (push) Successful in 2m45s
Project CI / Frontend tests (push) Successful in 3m22s
Project CI / Backend tests (push) Successful in 5m53s
Project CI / Native shell tests (push) Successful in 16m45s
Feat/AGC错误报告 (#240)
实现:
在rust内存里维护错误事件队列, webview调用tauri command传入, 后台任务agent工具等直接插入
把rust , webview console的日志统一写到AppData文件夹下的(滚动保存的)日志文件.
rust对传入的错误进行筛选,脱敏, 防抖,后通知前端提醒用户.
用户提醒是一个不阻塞的小UI, 展开后可以选择错误上报, 可以附加文字描述
上传时附带最近日志, 错误堆栈等信息

元数据存在数据库, 考虑到字符串信息很难查询, 所以在api server打包成zip存在OSS.
管理页面新增错误报告的查看页面

![shotmd-1788328504.jpg](/attachments/641b43c8-deed-47d8-8c78-36adb9a2548f)
![shotmd-1788328495.jpg](/attachments/33c6ada3-dd4d-44b6-9859-f18f944eb653)
![shotmd-1788328222.jpg](/attachments/59267df9-f624-4e47-bb9b-b69d77f595c4)
![shotmd-1788328230.jpg](/attachments/63ea6a86-9a51-412b-a98c-dce454bcc968)

---------

Co-authored-by: 段舒康 <kdletters@qq.com>
Reviewed-on: http://192.168.35.82/git/GenarrativeAI/Genarrative/pulls/240
Co-authored-by: 王德宇 <kvtodev@outlook.com>
Co-committed-by: 王德宇 <kvtodev@outlook.com>
2026-09-03 10:02:05 +08:00

3374 lines
118 KiB
Rust
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
use std::{collections::BTreeMap, error::Error, fmt, future::Future, sync::Arc, time::Instant};
use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64_STANDARD};
use bytes::Bytes;
use hmac::{Hmac, Mac};
use reqwest::Method;
use serde::{Deserialize, Serialize};
use serde_json::{Value, json};
use sha2::{Digest, Sha256};
use time::{Duration, OffsetDateTime, format_description::well_known::Rfc3339};
use tokio::{sync::Semaphore, time::sleep};
use tracing::{info, warn};
type HmacSha256 = Hmac<Sha256>;
pub const DEFAULT_POST_EXPIRE_SECONDS: u64 = 10 * 60;
pub const DEFAULT_READ_EXPIRE_SECONDS: u64 = 10 * 60;
pub const DEFAULT_POST_MAX_SIZE_BYTES: u64 = 20 * 1024 * 1024;
pub const DEFAULT_SUCCESS_ACTION_STATUS: u16 = 200;
pub const DEFAULT_METADATA_TOTAL_BYTES_LIMIT: usize = 8 * 1024;
pub const DEFAULT_IMMUTABLE_CACHE_CONTROL: &str = "public, max-age=31536000, immutable";
const OSS_V4_ALGORITHM: &str = "OSS4-HMAC-SHA256";
const OSS_V4_REQUEST: &str = "aliyun_v4_request";
const OSS_V4_SERVICE: &str = "oss";
const OSS_UNSIGNED_PAYLOAD: &str = "UNSIGNED-PAYLOAD";
const OSS_PROVIDER: &str = "aliyun-oss";
pub const LEGACY_PUBLIC_PREFIXES: [&str; 14] = [
"generated-character-drafts",
"generated-characters",
"generated-animations",
"generated-big-fish-assets",
"generated-square-hole-assets",
"generated-wooden-fish-assets",
"generated-match3d-assets",
"generated-puzzle-assets",
"generated-puzzle-clear-assets",
"generated-jump-hop-assets",
"generated-custom-world-scenes",
"generated-custom-world-covers",
"generated-bark-battle-assets",
"generated-qwen-sprites",
];
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum OssObjectAccess {
Public,
Private,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum LegacyAssetPrefix {
EditorAgent,
AgcErrorReports,
CharacterDrafts,
Characters,
Animations,
BigFishAssets,
SquareHoleAssets,
WoodenFishAssets,
Match3DAssets,
PuzzleAssets,
PuzzleClearAssets,
JumpHopAssets,
CustomWorldScenes,
CustomWorldCovers,
BarkBattleAssets,
QwenSprites,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct OssConfig {
bucket: String,
endpoint: String,
access_key_id: String,
access_key_secret: String,
default_read_expire_seconds: u64,
default_post_expire_seconds: u64,
default_post_max_size_bytes: u64,
default_success_action_status: u16,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct OssPostObjectRequest {
pub prefix: LegacyAssetPrefix,
pub path_segments: Vec<String>,
pub file_name: String,
pub content_type: Option<String>,
pub access: OssObjectAccess,
pub metadata: BTreeMap<String, String>,
pub max_size_bytes: Option<u64>,
pub expire_seconds: Option<u64>,
pub success_action_status: Option<u16>,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct OssSignedGetObjectUrlRequest {
pub object_key: String,
pub expire_seconds: Option<u64>,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct OssHeadObjectRequest {
pub object_key: String,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct OssGetObjectRequest {
pub object_key: String,
pub max_bytes: usize,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct OssDeleteObjectRequest {
pub object_key: String,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct OssPutObjectRequest {
pub prefix: LegacyAssetPrefix,
pub path_segments: Vec<String>,
pub file_name: String,
pub content_type: Option<String>,
pub access: OssObjectAccess,
pub metadata: BTreeMap<String, String>,
pub body: Vec<u8>,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
pub struct OssPostObjectResponse {
#[serde(rename = "signatureVersion")]
pub signature_version: &'static str,
pub provider: &'static str,
pub bucket: String,
pub endpoint: String,
pub host: String,
#[serde(rename = "objectKey")]
pub object_key: String,
#[serde(rename = "legacyPublicPath")]
pub legacy_public_path: String,
#[serde(rename = "contentType", skip_serializing_if = "Option::is_none")]
pub content_type: Option<String>,
pub access: OssObjectAccess,
#[serde(rename = "keyPrefix")]
pub key_prefix: String,
#[serde(rename = "expiresAt")]
pub expires_at: String,
#[serde(rename = "maxSizeBytes")]
pub max_size_bytes: u64,
#[serde(rename = "successActionStatus")]
pub success_action_status: u16,
#[serde(rename = "formFields")]
pub form_fields: OssPostObjectFormFields,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
pub struct OssSignedGetObjectUrlResponse {
pub provider: &'static str,
pub bucket: String,
pub endpoint: String,
pub host: String,
#[serde(rename = "objectKey")]
pub object_key: String,
#[serde(rename = "expiresAt")]
pub expires_at: String,
#[serde(rename = "signedUrl")]
pub signed_url: String,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct OssHeadObjectResponse {
pub bucket: String,
pub object_key: String,
pub content_length: u64,
pub content_type: Option<String>,
pub etag: Option<String>,
pub last_modified: Option<String>,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
pub struct OssPutObjectResponse {
pub provider: &'static str,
pub bucket: String,
pub endpoint: String,
pub host: String,
#[serde(rename = "objectKey")]
pub object_key: String,
#[serde(rename = "legacyPublicPath")]
pub legacy_public_path: String,
#[serde(rename = "contentType", skip_serializing_if = "Option::is_none")]
pub content_type: Option<String>,
#[serde(rename = "contentLength")]
pub content_length: u64,
pub access: OssObjectAccess,
#[serde(skip_serializing_if = "Option::is_none")]
pub etag: Option<String>,
#[serde(rename = "lastModified", skip_serializing_if = "Option::is_none")]
pub last_modified: Option<String>,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
pub struct OssPostObjectFormFields {
pub key: String,
pub policy: String,
#[serde(rename = "x-oss-signature-version")]
pub signature_version: String,
#[serde(rename = "x-oss-credential")]
pub credential: String,
#[serde(rename = "x-oss-date")]
pub date: String,
#[serde(rename = "x-oss-signature")]
pub signature: String,
#[serde(rename = "success_action_status")]
pub success_action_status: String,
#[serde(rename = "Content-Type", skip_serializing_if = "Option::is_none")]
pub content_type: Option<String>,
#[serde(rename = "Cache-Control", skip_serializing_if = "Option::is_none")]
pub cache_control: Option<String>,
#[serde(flatten)]
pub metadata: BTreeMap<String, String>,
}
#[derive(Clone, Debug)]
pub struct OssClient {
config: OssConfig,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum OssRequestOperation {
Put,
Head,
Get,
Delete,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct OssRequestError {
pub status: Option<u16>,
pub timeout: bool,
pub connect: bool,
pub transport: bool,
pub oss_code: Option<String>,
pub oss_request_id: Option<String>,
pub operation: OssRequestOperation,
pub message: String,
}
#[derive(Debug, PartialEq, Eq)]
pub enum OssError {
InvalidConfig(String),
InvalidRequest(String),
ObjectNotFound(String),
Request(OssRequestError),
SerializePolicy(String),
Sign(String),
}
// 平台 OSS 错误只先归类,不在 platform 层绑定 HTTP status。
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum OssErrorKind {
InvalidConfig,
InvalidRequest,
ObjectNotFound,
Request,
SerializePolicy,
Sign,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct OssRequestAttemptContext {
pub frame_index: usize,
pub operation: &'static str,
}
const CHARACTER_ANIMATION_OSS_MAX_ATTEMPTS: u32 = 3;
const CHARACTER_ANIMATION_OSS_RETRY_DELAYS_MS: [u64; 2] = [250, 500];
const CHARACTER_ANIMATION_OSS_ERROR_BODY_MAX_BYTES: usize = 16 * 1024;
const OSS_ERROR_CODE_MAX_BYTES: usize = 128;
const OSS_REQUEST_ID_MAX_BYTES: usize = 256;
struct PreparedHeadObject {
object_key: String,
target_url: reqwest::Url,
}
struct PreparedPutObject {
object_key: String,
target_url: reqwest::Url,
content_type: Option<String>,
headers: BTreeMap<String, String>,
content_length: u64,
access: OssObjectAccess,
body: Bytes,
}
impl LegacyAssetPrefix {
pub fn parse(raw: &str) -> Option<Self> {
let normalized = raw
.trim()
.trim_start_matches('/')
.trim_end_matches('/')
.trim_end_matches('*')
.trim_end_matches('/');
match normalized {
"editor-agent" => Some(Self::EditorAgent),
"agc" => Some(Self::AgcErrorReports),
"generated-character-drafts" => Some(Self::CharacterDrafts),
"generated-characters" => Some(Self::Characters),
"generated-animations" => Some(Self::Animations),
"generated-big-fish-assets" => Some(Self::BigFishAssets),
"generated-square-hole-assets" => Some(Self::SquareHoleAssets),
"generated-wooden-fish-assets" => Some(Self::WoodenFishAssets),
"generated-match3d-assets" => Some(Self::Match3DAssets),
"generated-puzzle-assets" => Some(Self::PuzzleAssets),
"generated-puzzle-clear-assets" => Some(Self::PuzzleClearAssets),
"generated-jump-hop-assets" => Some(Self::JumpHopAssets),
"generated-custom-world-scenes" => Some(Self::CustomWorldScenes),
"generated-custom-world-covers" => Some(Self::CustomWorldCovers),
"generated-bark-battle-assets" => Some(Self::BarkBattleAssets),
"generated-qwen-sprites" => Some(Self::QwenSprites),
_ => None,
}
}
pub fn as_str(&self) -> &'static str {
match self {
Self::EditorAgent => "editor-agent",
Self::AgcErrorReports => "agc",
Self::CharacterDrafts => "generated-character-drafts",
Self::Characters => "generated-characters",
Self::Animations => "generated-animations",
Self::BigFishAssets => "generated-big-fish-assets",
Self::SquareHoleAssets => "generated-square-hole-assets",
Self::WoodenFishAssets => "generated-wooden-fish-assets",
Self::Match3DAssets => "generated-match3d-assets",
Self::PuzzleAssets => "generated-puzzle-assets",
Self::PuzzleClearAssets => "generated-puzzle-clear-assets",
Self::JumpHopAssets => "generated-jump-hop-assets",
Self::CustomWorldScenes => "generated-custom-world-scenes",
Self::CustomWorldCovers => "generated-custom-world-covers",
Self::BarkBattleAssets => "generated-bark-battle-assets",
Self::QwenSprites => "generated-qwen-sprites",
}
}
pub fn as_public_path_prefix(&self) -> String {
format!("/{}", self.as_str())
}
pub fn from_object_key(raw: &str) -> Option<Self> {
let normalized = raw.trim().trim_start_matches('/').trim();
let prefix = normalized.split('/').next()?;
match Self::parse(prefix) {
// agc/error-reports is a server-only write prefix, never a caller-supplied object key.
Some(Self::AgcErrorReports) => None,
other => other,
}
}
}
impl OssConfig {
#[allow(clippy::too_many_arguments)]
pub fn new(
bucket: String,
endpoint: String,
access_key_id: String,
access_key_secret: String,
default_read_expire_seconds: u64,
default_post_expire_seconds: u64,
default_post_max_size_bytes: u64,
default_success_action_status: u16,
) -> Result<Self, OssError> {
let bucket = normalize_required_value(bucket, "OSS bucket 不能为空")?;
let endpoint = normalize_endpoint(&endpoint)?;
let access_key_id = normalize_required_value(access_key_id, "OSS AccessKeyId 不能为空")?;
let access_key_secret =
normalize_required_value(access_key_secret, "OSS AccessKeySecret 不能为空")?;
if default_read_expire_seconds == 0 {
return Err(OssError::InvalidConfig(
"OSS 私有读签名有效期必须大于 0".to_string(),
));
}
if default_post_expire_seconds == 0 {
return Err(OssError::InvalidConfig(
"OSS PostObject 签名有效期必须大于 0".to_string(),
));
}
if default_post_max_size_bytes == 0 {
return Err(OssError::InvalidConfig(
"OSS PostObject 最大上传大小必须大于 0".to_string(),
));
}
if !(100..=999).contains(&default_success_action_status) {
return Err(OssError::InvalidConfig(
"OSS success_action_status 必须是三位 HTTP 状态码".to_string(),
));
}
Ok(Self {
bucket,
endpoint,
access_key_id,
access_key_secret,
default_read_expire_seconds,
default_post_expire_seconds,
default_post_max_size_bytes,
default_success_action_status,
})
}
pub fn upload_host(&self) -> String {
format!("https://{}.{}", self.bucket, self.endpoint)
}
pub fn endpoint(&self) -> &str {
&self.endpoint
}
pub fn bucket(&self) -> &str {
&self.bucket
}
pub fn access_key_id(&self) -> &str {
&self.access_key_id
}
pub fn access_key_secret(&self) -> &str {
&self.access_key_secret
}
}
impl OssClient {
pub fn new(config: OssConfig) -> Self {
Self { config }
}
pub fn config_bucket(&self) -> &str {
self.config.bucket()
}
pub fn sign_post_object(
&self,
request: OssPostObjectRequest,
) -> Result<OssPostObjectResponse, OssError> {
let started_at = Instant::now();
let requested_prefix = request.prefix.as_str();
let requested_content_type = request
.content_type
.as_deref()
.map(str::trim)
.unwrap_or("")
.to_string();
let requested_metadata_count = request.metadata.len();
let result = (|| {
let max_size_bytes = request
.max_size_bytes
.unwrap_or(self.config.default_post_max_size_bytes);
let expire_seconds = request
.expire_seconds
.unwrap_or(self.config.default_post_expire_seconds);
let success_action_status = request
.success_action_status
.unwrap_or(self.config.default_success_action_status);
if max_size_bytes == 0 {
return Err(OssError::InvalidRequest(
"maxSizeBytes 必须大于 0".to_string(),
));
}
if expire_seconds == 0 {
return Err(OssError::InvalidRequest(
"expireSeconds 必须大于 0".to_string(),
));
}
if !(100..=999).contains(&success_action_status) {
return Err(OssError::InvalidRequest(
"successActionStatus 必须是三位 HTTP 状态码".to_string(),
));
}
let sanitized_segments = request
.path_segments
.iter()
.map(|segment| sanitize_path_segment(segment))
.filter(|segment| !segment.is_empty())
.collect::<Vec<_>>();
let file_name = sanitize_file_name(&request.file_name)?;
let object_key = build_object_key(request.prefix, &sanitized_segments, &file_name);
let legacy_public_path = format!("/{}", object_key);
let content_type = normalize_optional_value(request.content_type);
let metadata = normalize_metadata(request.metadata)?;
let cache_control = Some(DEFAULT_IMMUTABLE_CACHE_CONTROL.to_string());
let expires_at = OffsetDateTime::now_utc()
.checked_add(Duration::seconds(i64::try_from(expire_seconds).map_err(
|_| OssError::InvalidRequest("expireSeconds 超出可支持范围".to_string()),
)?))
.ok_or_else(|| {
OssError::InvalidRequest("expireSeconds 计算结果溢出".to_string())
})?;
let expires_at = expires_at.format(&Rfc3339).map_err(|error| {
OssError::SerializePolicy(format!("格式化过期时间失败:{error}"))
})?;
let signed_at = OffsetDateTime::now_utc();
let signature_scope = build_v4_signature_scope(&self.config.endpoint, signed_at)?;
let signature_date = build_v4_signature_date(signed_at)?;
let credential = format!("{}/{}", self.config.access_key_id, signature_scope);
let policy_json = build_policy_json(
&self.config.bucket,
&object_key,
&expires_at,
max_size_bytes,
success_action_status,
content_type.as_deref(),
cache_control.as_deref(),
&metadata,
&credential,
&signature_date,
);
let policy = serde_json::to_string(&policy_json).map_err(|error| {
OssError::SerializePolicy(format!("序列化 policy 失败:{error}"))
})?;
let encoded_policy = BASE64_STANDARD.encode(policy.as_bytes());
let signature = sign_v4_content(
&self.config.access_key_secret,
&signature_scope,
&encoded_policy,
)?;
Ok(OssPostObjectResponse {
signature_version: "v4",
provider: OSS_PROVIDER,
bucket: self.config.bucket.clone(),
endpoint: self.config.endpoint.clone(),
host: self.config.upload_host(),
object_key: object_key.clone(),
legacy_public_path,
content_type: content_type.clone(),
access: request.access,
key_prefix: build_key_prefix(request.prefix, &sanitized_segments),
expires_at,
max_size_bytes,
success_action_status,
form_fields: OssPostObjectFormFields {
key: object_key,
policy: encoded_policy,
signature_version: OSS_V4_ALGORITHM.to_string(),
credential,
date: signature_date,
signature,
success_action_status: success_action_status.to_string(),
content_type,
cache_control,
metadata,
},
})
})();
match &result {
Ok(response) => info!(
provider = OSS_PROVIDER,
operation = "sign_post_object",
bucket = %response.bucket,
endpoint = %response.endpoint,
object_key = %response.object_key,
key_prefix = %response.key_prefix,
access = oss_access_label(response.access),
content_type = %response.content_type.as_deref().unwrap_or(""),
max_size_bytes = response.max_size_bytes,
success_action_status = response.success_action_status,
metadata_count = response.form_fields.metadata.len(),
expires_at = %response.expires_at,
elapsed_ms = elapsed_ms(started_at),
"OSS PostObject 签名完成"
),
Err(error) => warn!(
provider = OSS_PROVIDER,
operation = "sign_post_object",
bucket = %self.config.bucket(),
endpoint = %self.config.endpoint(),
key_prefix = requested_prefix,
content_type = %requested_content_type,
metadata_count = requested_metadata_count,
error_kind = oss_error_kind_label(error),
message = %error,
elapsed_ms = elapsed_ms(started_at),
"OSS PostObject 签名失败"
),
}
result
}
// 私有 bucket 的对象读取统一走短期签名 URL,避免把长期主凭证下发给浏览器。
pub fn sign_get_object_url(
&self,
request: OssSignedGetObjectUrlRequest,
) -> Result<OssSignedGetObjectUrlResponse, OssError> {
self.sign_get_object_url_with_normalizer(
request,
normalize_object_key,
"sign_get_object_url",
)
}
pub fn sign_internal_get_object_url(
&self,
request: OssSignedGetObjectUrlRequest,
) -> Result<OssSignedGetObjectUrlResponse, OssError> {
self.sign_get_object_url_with_normalizer(
request,
normalize_editor_agent_messages_object_key,
"sign_internal_get_object_url",
)
}
fn sign_get_object_url_with_normalizer(
&self,
request: OssSignedGetObjectUrlRequest,
normalize: fn(&str) -> Result<String, OssError>,
operation: &'static str,
) -> Result<OssSignedGetObjectUrlResponse, OssError> {
let started_at = Instant::now();
let requested_object_key = request
.object_key
.trim()
.trim_start_matches('/')
.trim()
.to_string();
let result = (|| {
let expire_seconds = request
.expire_seconds
.unwrap_or(self.config.default_read_expire_seconds);
if expire_seconds == 0 {
return Err(OssError::InvalidRequest(
"expireSeconds 必须大于 0".to_string(),
));
}
let object_key = normalize(&request.object_key)?;
let expires_at = OffsetDateTime::now_utc()
.checked_add(Duration::seconds(i64::try_from(expire_seconds).map_err(
|_| OssError::InvalidRequest("expireSeconds 超出可支持范围".to_string()),
)?))
.ok_or_else(|| {
OssError::InvalidRequest("expireSeconds 计算结果溢出".to_string())
})?;
let expires_at_text = expires_at
.format(&Rfc3339)
.map_err(|error| OssError::Sign(format!("格式化过期时间失败:{error}")))?;
let signed_at = OffsetDateTime::now_utc();
let signed_at_text = build_v4_signature_date(signed_at)?;
let signature_scope = build_v4_signature_scope(&self.config.endpoint, signed_at)?;
let credential = format!("{}/{}", self.config.access_key_id, signature_scope);
let mut query = BTreeMap::from([
("x-oss-additional-headers".to_string(), "host".to_string()),
(
"x-oss-signature-version".to_string(),
OSS_V4_ALGORITHM.to_string(),
),
("x-oss-credential".to_string(), credential),
("x-oss-date".to_string(), signed_at_text),
("x-oss-expires".to_string(), expire_seconds.to_string()),
]);
let canonical_uri = build_v4_canonical_uri(&self.config.bucket, Some(&object_key));
let object_url_path = format!("/{}", encode_url_path(&object_key));
let additional_headers = "host";
let canonical_headers =
format!("host:{}.{}\n", self.config.bucket(), self.config.endpoint());
let canonical_query = build_canonical_query_string(&query);
let canonical_request = build_v4_canonical_request(
Method::GET.as_str(),
&canonical_uri,
&canonical_query,
&canonical_headers,
additional_headers,
OSS_UNSIGNED_PAYLOAD,
);
let string_to_sign = build_v4_string_to_sign(
query["x-oss-date"].as_str(),
&signature_scope,
&canonical_request,
);
let signature = sign_v4_content(
&self.config.access_key_secret,
&signature_scope,
&string_to_sign,
)?;
query.insert("x-oss-signature".to_string(), signature);
let signed_url = format!(
"{}{}?{}",
self.config.upload_host(),
object_url_path,
build_canonical_query_string(&query)
);
Ok(OssSignedGetObjectUrlResponse {
provider: OSS_PROVIDER,
bucket: self.config.bucket.clone(),
endpoint: self.config.endpoint.clone(),
host: self.config.upload_host(),
object_key,
expires_at: expires_at_text,
signed_url,
})
})();
match &result {
Ok(response) => info!(
provider = OSS_PROVIDER,
operation,
bucket = %response.bucket,
endpoint = %response.endpoint,
object_key = %response.object_key,
expires_at = %response.expires_at,
elapsed_ms = elapsed_ms(started_at),
"OSS GetObject 读签名完成"
),
Err(error) => warn!(
provider = OSS_PROVIDER,
operation,
bucket = %self.config.bucket(),
endpoint = %self.config.endpoint(),
object_key = %requested_object_key,
error_kind = oss_error_kind_label(error),
message = %error,
elapsed_ms = elapsed_ms(started_at),
"OSS GetObject 读签名失败"
),
}
result
}
// 上传完成确认前,服务端必须自己探测一次对象,不能只相信客户端回传的 object_key。
pub async fn head_object(
&self,
client: &reqwest::Client,
request: OssHeadObjectRequest,
) -> Result<OssHeadObjectResponse, OssError> {
let started_at = Instant::now();
let requested_object_key = request
.object_key
.trim()
.trim_start_matches('/')
.trim()
.to_string();
let mut response_status = None;
let result = async {
let object_key = normalize_object_key(&request.object_key)?;
let target_url =
build_object_url(&self.config.bucket, &self.config.endpoint, &object_key).map_err(
|error| {
request_error(
OssRequestOperation::Head,
&format!("构造 OSS 对象 URL 失败:{error}"),
)
},
)?;
let response = send_signed_request(
client,
&self.config,
Method::HEAD,
Some(&object_key),
target_url,
OssRequestOperation::Head,
)
.await?;
response_status = Some(response.status().as_u16());
if response.status() == reqwest::StatusCode::NOT_FOUND {
return Err(OssError::ObjectNotFound(format!(
"OSS 对象不存在:{}",
request.object_key
)));
}
if !response.status().is_success() {
return Err(request_status_error(
OssRequestOperation::Head,
response.status().as_u16(),
format!("OSS HEAD Object 失败,状态码:{}", response.status()),
));
}
let headers = response.headers();
let content_length = headers
.get(reqwest::header::CONTENT_LENGTH)
.and_then(|value| value.to_str().ok())
.and_then(|value| value.parse::<u64>().ok())
.unwrap_or(0);
let content_type = headers
.get(reqwest::header::CONTENT_TYPE)
.and_then(|value| value.to_str().ok())
.map(|value| value.to_string());
let etag = headers
.get(reqwest::header::ETAG)
.and_then(|value| value.to_str().ok())
.map(|value| value.trim_matches('"').to_string());
let last_modified = headers
.get(reqwest::header::LAST_MODIFIED)
.and_then(|value| value.to_str().ok())
.map(|value| value.to_string());
Ok(OssHeadObjectResponse {
bucket: self.config.bucket.clone(),
object_key,
content_length,
content_type,
etag,
last_modified,
})
}
.await;
match &result {
Ok(response) => info!(
provider = OSS_PROVIDER,
operation = "head_object",
bucket = %response.bucket,
endpoint = %self.config.endpoint(),
object_key = %response.object_key,
status = response_status.unwrap_or(reqwest::StatusCode::OK.as_u16()),
status_class = http_status_class_from_option(response_status),
content_length = response.content_length,
content_type = %response.content_type.as_deref().unwrap_or(""),
etag_present = response.etag.is_some(),
last_modified_present = response.last_modified.is_some(),
elapsed_ms = elapsed_ms(started_at),
"OSS HEAD Object 完成"
),
Err(error) => warn!(
provider = OSS_PROVIDER,
operation = "head_object",
bucket = %self.config.bucket(),
endpoint = %self.config.endpoint(),
object_key = %requested_object_key,
status = response_status.unwrap_or_default(),
status_class = http_status_class_from_option(response_status),
error_kind = oss_error_kind_label(error),
message = %error,
elapsed_ms = elapsed_ms(started_at),
"OSS HEAD Object 失败"
),
}
result
}
pub async fn get_object(
&self,
client: &reqwest::Client,
request: OssGetObjectRequest,
) -> Result<Vec<u8>, OssError> {
let key = normalize_internal_object_key(&request.object_key)?;
let target = build_object_url(&self.config.bucket, &self.config.endpoint, &key)
.map_err(|e| request_error(OssRequestOperation::Get, &e.to_string()))?;
let mut response = send_signed_request(
client,
&self.config,
Method::GET,
Some(&key),
target,
OssRequestOperation::Get,
)
.await?;
if response.status() == reqwest::StatusCode::NOT_FOUND {
return Err(OssError::ObjectNotFound(format!("OSS 对象不存在:{key}")));
}
if !response.status().is_success() {
return Err(request_status_error(
OssRequestOperation::Get,
response.status().as_u16(),
format!("OSS GET Object 失败,状态码:{}", response.status()),
));
}
if response
.content_length()
.is_some_and(|length| length > request.max_bytes as u64)
{
return Err(OssError::InvalidRequest("OSS 对象超过读取上限".to_string()));
}
let mut bytes = Vec::new();
while let Some(chunk) = response
.chunk()
.await
.map_err(|e| request_error_from_reqwest(OssRequestOperation::Get, e))?
{
if bytes.len().saturating_add(chunk.len()) > request.max_bytes {
return Err(OssError::InvalidRequest("OSS 对象超过读取上限".to_string()));
}
bytes.extend_from_slice(&chunk);
}
Ok(bytes)
}
pub async fn delete_object(
&self,
client: &reqwest::Client,
request: OssDeleteObjectRequest,
) -> Result<(), OssError> {
let key = normalize_internal_object_key(&request.object_key)?;
let target = build_object_url(&self.config.bucket, &self.config.endpoint, &key)
.map_err(|e| request_error(OssRequestOperation::Delete, &e.to_string()))?;
let response = send_signed_request(
client,
&self.config,
Method::DELETE,
Some(&key),
target,
OssRequestOperation::Delete,
)
.await?;
if response.status() == reqwest::StatusCode::NOT_FOUND || response.status().is_success() {
return Ok(());
}
Err(request_status_error(
OssRequestOperation::Delete,
response.status().as_u16(),
format!("OSS DELETE Object 失败,状态码:{}", response.status()),
))
}
// AI 生成资源默认由服务端上传 OSS,Web 端只拿签名读地址,不直接持有写权限。
pub async fn put_object(
&self,
client: &reqwest::Client,
request: OssPutObjectRequest,
) -> Result<OssPutObjectResponse, OssError> {
let started_at = Instant::now();
let requested_prefix = request.prefix.as_str();
let requested_content_type = request
.content_type
.as_deref()
.map(str::trim)
.unwrap_or("")
.to_string();
let requested_content_length = request.body.len();
let requested_metadata_count = request.metadata.len();
let mut response_status = None;
let result = async {
if request.body.is_empty() {
return Err(OssError::InvalidRequest(
"服务端上传对象内容不能为空".to_string(),
));
}
let sanitized_segments = request
.path_segments
.iter()
.map(|segment| sanitize_path_segment(segment))
.filter(|segment| !segment.is_empty())
.collect::<Vec<_>>();
let file_name = sanitize_file_name(&request.file_name)?;
let object_key = build_object_key(request.prefix, &sanitized_segments, &file_name);
let content_type = normalize_optional_value(request.content_type);
let headers = build_put_object_headers(request.metadata)?;
let target_url =
build_object_url(&self.config.bucket, &self.config.endpoint, &object_key).map_err(
|error| {
request_error(
OssRequestOperation::Put,
&format!("构造 OSS 对象 URL 失败:{error}"),
)
},
)?;
let content_length = u64::try_from(request.body.len())
.map_err(|_| OssError::InvalidRequest("上传对象大小超出可支持范围".to_string()))?;
let builder = signed_request_builder(
client,
&self.config,
Method::PUT,
Some(&object_key),
target_url,
content_type.as_deref(),
&headers,
)?
.header(reqwest::header::CONTENT_LENGTH, content_length)
.body(request.body);
let response = builder
.send()
.await
.map_err(|error| request_error_from_reqwest(OssRequestOperation::Put, error))?;
response_status = Some(response.status().as_u16());
if !response.status().is_success() {
return Err(request_status_error(
OssRequestOperation::Put,
response.status().as_u16(),
format!("OSS PutObject 失败,状态码:{}", response.status()),
));
}
let headers = response.headers();
let etag = headers
.get(reqwest::header::ETAG)
.and_then(|value| value.to_str().ok())
.map(|value| value.trim_matches('"').to_string());
let last_modified = headers
.get(reqwest::header::LAST_MODIFIED)
.and_then(|value| value.to_str().ok())
.map(|value| value.to_string());
Ok(OssPutObjectResponse {
provider: OSS_PROVIDER,
bucket: self.config.bucket.clone(),
endpoint: self.config.endpoint.clone(),
host: self.config.upload_host(),
legacy_public_path: format!("/{object_key}"),
object_key,
content_type,
content_length,
access: request.access,
etag,
last_modified,
})
}
.await;
match &result {
Ok(response) => info!(
provider = OSS_PROVIDER,
operation = "put_object",
bucket = %response.bucket,
endpoint = %response.endpoint,
object_key = %response.object_key,
access = oss_access_label(response.access),
status = response_status.unwrap_or(reqwest::StatusCode::OK.as_u16()),
status_class = http_status_class_from_option(response_status),
content_length = response.content_length,
content_type = %response.content_type.as_deref().unwrap_or(""),
etag_present = response.etag.is_some(),
last_modified_present = response.last_modified.is_some(),
elapsed_ms = elapsed_ms(started_at),
"OSS PutObject 上传完成"
),
Err(error) => warn!(
provider = OSS_PROVIDER,
operation = "put_object",
bucket = %self.config.bucket(),
endpoint = %self.config.endpoint(),
key_prefix = requested_prefix,
content_length = requested_content_length,
content_type = %requested_content_type,
metadata_count = requested_metadata_count,
status = response_status.unwrap_or_default(),
status_class = http_status_class_from_option(response_status),
error_kind = oss_error_kind_label(error),
message = %error,
elapsed_ms = elapsed_ms(started_at),
"OSS PutObject 上传失败"
),
}
result
}
/// 角色动画帧专用的可重试 PUT。调用方传入进程级并发限制器,单次网络 attempt
/// 独占一个 permit,退避等待期间不会占用 permit。
pub async fn put_object_with_retry(
&self,
client: &reqwest::Client,
request: OssPutObjectRequest,
io_limiter: Arc<Semaphore>,
attempt_context: OssRequestAttemptContext,
) -> Result<OssPutObjectResponse, OssError> {
let prepared = self.prepare_put_object(request)?;
let object_key = prepared.object_key.clone();
run_animation_request_with_retry(io_limiter, attempt_context, &object_key, || {
self.put_object_once(client, &prepared)
})
.await
}
/// 角色动画帧专用的可重试 HEAD。HEAD 与 PUT 独立重试,HEAD 失败不会重新上传 PUT。
pub async fn head_object_with_retry(
&self,
client: &reqwest::Client,
request: OssHeadObjectRequest,
io_limiter: Arc<Semaphore>,
attempt_context: OssRequestAttemptContext,
) -> Result<OssHeadObjectResponse, OssError> {
let prepared = self.prepare_head_object(request)?;
let object_key = prepared.object_key.clone();
run_animation_request_with_retry(io_limiter, attempt_context, &object_key, || {
self.head_object_once(client, &prepared)
})
.await
}
fn prepare_head_object(
&self,
request: OssHeadObjectRequest,
) -> Result<PreparedHeadObject, OssError> {
let object_key = normalize_object_key(&request.object_key)?;
let target_url = build_object_url(&self.config.bucket, &self.config.endpoint, &object_key)
.map_err(|error| {
request_error(
OssRequestOperation::Head,
&format!("构造 OSS 对象 URL 失败:{error}"),
)
})?;
Ok(PreparedHeadObject {
object_key,
target_url,
})
}
async fn head_object_once(
&self,
client: &reqwest::Client,
prepared: &PreparedHeadObject,
) -> Result<(OssHeadObjectResponse, u16), OssError> {
let response = send_signed_request(
client,
&self.config,
Method::HEAD,
Some(&prepared.object_key),
prepared.target_url.clone(),
OssRequestOperation::Head,
)
.await?;
if response.status() == reqwest::StatusCode::NOT_FOUND {
return Err(OssError::ObjectNotFound(format!(
"OSS 对象不存在:{}",
prepared.object_key
)));
}
if !response.status().is_success() {
return Err(request_status_error(
OssRequestOperation::Head,
response.status().as_u16(),
format!("OSS HEAD Object 失败,状态码:{}", response.status()),
));
}
let headers = response.headers();
let content_length = headers
.get(reqwest::header::CONTENT_LENGTH)
.and_then(|value| value.to_str().ok())
.and_then(|value| value.parse::<u64>().ok())
.unwrap_or(0);
let content_type = headers
.get(reqwest::header::CONTENT_TYPE)
.and_then(|value| value.to_str().ok())
.map(str::to_string);
let etag = headers
.get(reqwest::header::ETAG)
.and_then(|value| value.to_str().ok())
.map(|value| value.trim_matches('"').to_string());
let last_modified = headers
.get(reqwest::header::LAST_MODIFIED)
.and_then(|value| value.to_str().ok())
.map(str::to_string);
Ok((
OssHeadObjectResponse {
bucket: self.config.bucket.clone(),
object_key: prepared.object_key.clone(),
content_length,
content_type,
etag,
last_modified,
},
response.status().as_u16(),
))
}
fn prepare_put_object(
&self,
request: OssPutObjectRequest,
) -> Result<PreparedPutObject, OssError> {
if request.body.is_empty() {
return Err(OssError::InvalidRequest(
"服务端上传对象内容不能为空".to_string(),
));
}
let sanitized_segments = request
.path_segments
.iter()
.map(|segment| sanitize_path_segment(segment))
.filter(|segment| !segment.is_empty())
.collect::<Vec<_>>();
let file_name = sanitize_file_name(&request.file_name)?;
let object_key = build_object_key(request.prefix, &sanitized_segments, &file_name);
let content_type = normalize_optional_value(request.content_type);
let headers = build_put_object_headers(request.metadata)?;
let target_url = build_object_url(&self.config.bucket, &self.config.endpoint, &object_key)
.map_err(|error| {
request_error(
OssRequestOperation::Put,
&format!("构造 OSS 对象 URL 失败:{error}"),
)
})?;
let content_length = u64::try_from(request.body.len())
.map_err(|_| OssError::InvalidRequest("上传对象大小超出可支持范围".to_string()))?;
Ok(PreparedPutObject {
object_key,
target_url,
content_type,
headers,
content_length,
access: request.access,
body: Bytes::from(request.body),
})
}
async fn put_object_once(
&self,
client: &reqwest::Client,
prepared: &PreparedPutObject,
) -> Result<(OssPutObjectResponse, u16), OssError> {
let response = signed_request_builder(
client,
&self.config,
Method::PUT,
Some(&prepared.object_key),
prepared.target_url.clone(),
prepared.content_type.as_deref(),
&prepared.headers,
)?
.header(reqwest::header::CONTENT_LENGTH, prepared.content_length)
.body(prepared.body.clone())
.send()
.await
.map_err(|error| request_error_from_reqwest(OssRequestOperation::Put, error))?;
if !response.status().is_success() {
return Err(request_status_error_from_oss_put_response(response).await);
}
let headers = response.headers();
let etag = headers
.get(reqwest::header::ETAG)
.and_then(|value| value.to_str().ok())
.map(|value| value.trim_matches('"').to_string());
let last_modified = headers
.get(reqwest::header::LAST_MODIFIED)
.and_then(|value| value.to_str().ok())
.map(str::to_string);
Ok((
OssPutObjectResponse {
provider: OSS_PROVIDER,
bucket: self.config.bucket.clone(),
endpoint: self.config.endpoint.clone(),
host: self.config.upload_host(),
legacy_public_path: format!("/{}", prepared.object_key),
object_key: prepared.object_key.clone(),
content_type: prepared.content_type.clone(),
content_length: prepared.content_length,
access: prepared.access,
etag,
last_modified,
},
response.status().as_u16(),
))
}
}
fn request_error(operation: OssRequestOperation, message: &str) -> OssError {
OssError::Request(OssRequestError {
status: None,
timeout: false,
connect: false,
transport: false,
oss_code: None,
oss_request_id: None,
operation,
message: message.to_string(),
})
}
async fn request_status_error_from_oss_put_response(mut response: reqwest::Response) -> OssError {
let status = response.status();
let header_request_id = response
.headers()
.get("x-oss-request-id")
.and_then(|value| value.to_str().ok())
.and_then(|value| normalize_oss_error_field(value, OSS_REQUEST_ID_MAX_BYTES));
let body_read = if status == reqwest::StatusCode::BAD_REQUEST {
read_bounded_oss_error_body(&mut response).await
} else {
OssErrorBodyRead {
body: Vec::new(),
read_failure: None,
}
};
request_status_error_from_oss_parts(
OssRequestOperation::Put,
status.as_u16(),
header_request_id,
&body_read.body,
body_read.read_failure,
)
}
/// 400 错误响应体读取失败的原因。仅在部分响应体尚未解析出确定性
/// OSS 错误码时参与重试判定,否则只体现在 message 里。
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
struct OssErrorBodyReadFailure {
timeout: bool,
}
struct OssErrorBodyRead {
body: Vec<u8>,
read_failure: Option<OssErrorBodyReadFailure>,
}
async fn read_bounded_oss_error_body(response: &mut reqwest::Response) -> OssErrorBodyRead {
let mut body = Vec::new();
let mut read_failure = None;
while body.len() < CHARACTER_ANIMATION_OSS_ERROR_BODY_MAX_BYTES {
let chunk = match response.chunk().await {
Ok(Some(chunk)) => chunk,
Ok(None) => break,
Err(error) => {
// 断流/超时不丢弃已读字节:部分响应体可能已含确定性错误码。
read_failure = Some(OssErrorBodyReadFailure {
timeout: error.is_timeout(),
});
break;
}
};
let remaining = CHARACTER_ANIMATION_OSS_ERROR_BODY_MAX_BYTES - body.len();
body.extend_from_slice(&chunk[..chunk.len().min(remaining)]);
if chunk.len() > remaining {
break;
}
}
OssErrorBodyRead { body, read_failure }
}
fn request_status_error_from_oss_parts(
operation: OssRequestOperation,
status: u16,
header_request_id: Option<String>,
body: &[u8],
body_read_failure: Option<OssErrorBodyReadFailure>,
) -> OssError {
let header_request_id = header_request_id
.as_deref()
.and_then(|value| normalize_oss_error_field(value, OSS_REQUEST_ID_MAX_BYTES));
let bounded_body = &body[..body.len().min(CHARACTER_ANIMATION_OSS_ERROR_BODY_MAX_BYTES)];
let oss_code = extract_oss_error_xml_field(bounded_body, "Code", OSS_ERROR_CODE_MAX_BYTES);
let xml_request_id =
extract_oss_error_xml_field(bounded_body, "RequestId", OSS_REQUEST_ID_MAX_BYTES);
let oss_request_id = header_request_id.or(xml_request_id);
// 确定性错误码优先:已解析出 Code 时,响应体读取失败只保留在 message 里,
// 不改变重试语义;错误码缺失时才按读取失败归类为可重试的超时/传输错误。
let unclassified_read_failure = if oss_code.is_some() {
None
} else {
body_read_failure
};
let timeout = (status == reqwest::StatusCode::BAD_REQUEST.as_u16()
&& oss_code.as_deref() == Some("RequestTimeout"))
|| unclassified_read_failure.is_some_and(|failure| failure.timeout);
let transport = unclassified_read_failure.is_some_and(|failure| !failure.timeout);
let mut message = format!("OSS PutObject 失败,状态码:{status}");
if let Some(oss_code) = oss_code.as_deref() {
message.push_str(&format!(",OSS 错误码:{oss_code}"));
}
if let Some(oss_request_id) = oss_request_id.as_deref() {
message.push_str(&format!(",OSS Request ID:{oss_request_id}"));
}
if body_read_failure.is_some() {
message.push_str(",错误响应体读取失败");
}
OssError::Request(OssRequestError {
status: Some(status),
timeout,
connect: false,
transport,
oss_code,
oss_request_id,
operation,
message,
})
}
fn extract_oss_error_xml_field(body: &[u8], field: &str, max_bytes: usize) -> Option<String> {
let body = std::str::from_utf8(body).ok()?;
let start_tag = format!("<{field}>");
let end_tag = format!("</{field}>");
let value_start = body.find(&start_tag)? + start_tag.len();
let value_end = value_start + body[value_start..].find(&end_tag)?;
normalize_oss_error_field(&body[value_start..value_end], max_bytes)
}
fn normalize_oss_error_field(value: &str, max_bytes: usize) -> Option<String> {
let value = value.trim();
if value.is_empty() || value.len() > max_bytes || value.chars().any(char::is_control) {
return None;
}
Some(value.to_string())
}
async fn run_animation_request_with_retry<T, F, Fut>(
io_limiter: Arc<Semaphore>,
attempt_context: OssRequestAttemptContext,
object_key: &str,
mut attempt_request: F,
) -> Result<T, OssError>
where
F: FnMut() -> Fut,
Fut: Future<Output = Result<(T, u16), OssError>>,
{
for attempt in 1..=CHARACTER_ANIMATION_OSS_MAX_ATTEMPTS {
let attempt_started_at = Instant::now();
let permit_wait_started_at = Instant::now();
let permit = io_limiter
.acquire()
.await
.map_err(|_| OssError::InvalidConfig("角色动画 OSS 并发限制器已关闭".to_string()))?;
let permit_wait_ms = elapsed_ms(permit_wait_started_at);
let result = attempt_request().await;
drop(permit);
let retryable = result.as_ref().err().is_some_and(oss_error_is_retryable);
let will_retry = retryable && attempt < CHARACTER_ANIMATION_OSS_MAX_ATTEMPTS;
let retry_delay_ms = if will_retry {
CHARACTER_ANIMATION_OSS_RETRY_DELAYS_MS[(attempt - 1) as usize]
} else {
0
};
let success_status = result.as_ref().ok().map(|(_, status)| *status);
log_animation_request_attempt(
attempt_context,
object_key,
attempt,
retryable,
will_retry,
retry_delay_ms,
permit_wait_ms,
elapsed_ms(attempt_started_at),
success_status,
result.as_ref().err(),
);
match result {
Ok((response, _status)) => return Ok(response),
Err(_error) if will_retry => {
sleep(std::time::Duration::from_millis(retry_delay_ms)).await
}
Err(error) => return Err(error),
}
}
unreachable!("角色动画 OSS 重试循环必须返回结果")
}
fn request_error_from_reqwest(operation: OssRequestOperation, error: reqwest::Error) -> OssError {
let status = error.status().map(|status| status.as_u16());
let timeout = error.is_timeout();
let connect = error.is_connect();
let transport = !timeout && !connect && (error.is_request() || error.is_body());
OssError::Request(OssRequestError {
status,
timeout,
connect,
transport,
oss_code: None,
oss_request_id: None,
operation,
message: format!("请求 OSS 失败:{error}"),
})
}
fn request_status_error(operation: OssRequestOperation, status: u16, message: String) -> OssError {
OssError::Request(OssRequestError {
status: Some(status),
timeout: false,
connect: false,
transport: false,
oss_code: None,
oss_request_id: None,
operation,
message,
})
}
fn oss_error_is_retryable(error: &OssError) -> bool {
let OssError::Request(request_error) = error else {
return false;
};
match (request_error.status, request_error.oss_code.as_deref()) {
(Some(400), Some("RequestTimeout")) => true,
// 400 是唯一会读取错误响应体的状态码:错误码缺失且响应体读取
// 超时/断流时,无法证明是确定性 400,按传输错误重试。
(Some(400), None) if request_error.timeout || request_error.transport => true,
(Some(408 | 429 | 500..=599), _) => true,
(Some(_), _) => false,
(None, _) => request_error.timeout || request_error.connect || request_error.transport,
}
}
fn request_error_details(
error: Option<&OssError>,
) -> (Option<u16>, bool, bool, bool, Option<&str>, Option<&str>) {
match error {
Some(OssError::Request(request_error)) => (
request_error.status,
request_error.timeout,
request_error.connect,
request_error.transport,
request_error.oss_code.as_deref(),
request_error.oss_request_id.as_deref(),
),
Some(OssError::ObjectNotFound(_)) => (Some(404), false, false, false, None, None),
_ => (None, false, false, false, None, None),
}
}
fn log_animation_request_attempt(
context: OssRequestAttemptContext,
object_key: &str,
attempt: u32,
retryable: bool,
will_retry: bool,
retry_delay_ms: u64,
permit_wait_ms: u64,
elapsed_ms: u64,
success_status: Option<u16>,
error: Option<&OssError>,
) {
let (error_status, timeout, connect, transport, oss_code, oss_request_id) =
request_error_details(error);
let status = success_status.or(error_status);
if error.is_none() {
info!(
provider = OSS_PROVIDER,
frame_index = context.frame_index,
object_key,
operation = context.operation,
attempt,
max_attempts = CHARACTER_ANIMATION_OSS_MAX_ATTEMPTS,
retryable,
will_retry,
retry_delay_ms,
permit_wait_ms,
timeout,
connect,
transport,
oss_code = oss_code.unwrap_or_default(),
oss_request_id = oss_request_id.unwrap_or_default(),
status = status.unwrap_or_default(),
elapsed_ms,
"角色动画 OSS 请求 attempt 完成"
);
return;
}
warn!(
provider = OSS_PROVIDER,
frame_index = context.frame_index,
object_key,
operation = context.operation,
attempt,
max_attempts = CHARACTER_ANIMATION_OSS_MAX_ATTEMPTS,
retryable,
will_retry,
retry_delay_ms,
permit_wait_ms,
timeout,
connect,
transport,
oss_code = oss_code.unwrap_or_default(),
oss_request_id = oss_request_id.unwrap_or_default(),
status = status.unwrap_or_default(),
elapsed_ms,
error_kind = error.map(oss_error_kind_label),
message = error.map(ToString::to_string).unwrap_or_default(),
"角色动画 OSS 请求 attempt 完成"
);
}
impl fmt::Display for OssError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::InvalidConfig(message)
| Self::InvalidRequest(message)
| Self::SerializePolicy(message)
| Self::Sign(message) => f.write_str(message),
Self::ObjectNotFound(message) => f.write_str(message),
Self::Request(error) => f.write_str(&error.message),
}
}
}
impl Error for OssError {}
impl OssError {
pub fn kind(&self) -> OssErrorKind {
match self {
Self::InvalidConfig(_) => OssErrorKind::InvalidConfig,
Self::InvalidRequest(_) => OssErrorKind::InvalidRequest,
Self::ObjectNotFound(_) => OssErrorKind::ObjectNotFound,
Self::Request(_) => OssErrorKind::Request,
Self::SerializePolicy(_) => OssErrorKind::SerializePolicy,
Self::Sign(_) => OssErrorKind::Sign,
}
}
}
fn elapsed_ms(started_at: Instant) -> u64 {
started_at.elapsed().as_millis().min(u64::MAX as u128) as u64
}
fn oss_access_label(access: OssObjectAccess) -> &'static str {
match access {
OssObjectAccess::Public => "public",
OssObjectAccess::Private => "private",
}
}
fn oss_error_kind_label(error: &OssError) -> &'static str {
match error.kind() {
OssErrorKind::InvalidConfig => "invalid_config",
OssErrorKind::InvalidRequest => "invalid_request",
OssErrorKind::ObjectNotFound => "object_not_found",
OssErrorKind::Request => "request",
OssErrorKind::SerializePolicy => "serialize_policy",
OssErrorKind::Sign => "sign",
}
}
fn http_status_class_from_option(status: Option<u16>) -> &'static str {
status.map(http_status_class).unwrap_or("unknown")
}
fn http_status_class(status: u16) -> &'static str {
match status {
100..=199 => "1xx",
200..=299 => "2xx",
300..=399 => "3xx",
400..=499 => "4xx",
500..=599 => "5xx",
_ => "unknown",
}
}
fn build_policy_json(
bucket: &str,
object_key: &str,
expires_at: &str,
max_size_bytes: u64,
success_action_status: u16,
content_type: Option<&str>,
cache_control: Option<&str>,
metadata: &BTreeMap<String, String>,
credential: &str,
signature_date: &str,
) -> Value {
let mut conditions = vec![
json!({ "bucket": bucket }),
json!(["eq", "$key", object_key]),
json!(["content-length-range", 1, max_size_bytes]),
json!([
"eq",
"$success_action_status",
success_action_status.to_string()
]),
json!(["eq", "$x-oss-signature-version", OSS_V4_ALGORITHM]),
json!(["eq", "$x-oss-credential", credential]),
json!(["eq", "$x-oss-date", signature_date]),
];
if let Some(content_type) = content_type {
conditions.push(json!(["eq", "$content-type", content_type]));
}
if let Some(cache_control) = cache_control {
conditions.push(json!(["eq", "$Cache-Control", cache_control]));
}
for (key, value) in metadata {
conditions.push(json!(["eq", format!("${key}"), value]));
}
json!({
"expiration": expires_at,
"conditions": conditions,
})
}
fn build_object_url(
bucket: &str,
endpoint: &str,
object_key: &str,
) -> Result<reqwest::Url, String> {
let mut url = reqwest::Url::parse(&format!("https://{bucket}.{endpoint}/"))
.map_err(|error| error.to_string())?;
url = url
.join(object_key.trim_start_matches('/'))
.map_err(|error| error.to_string())?;
Ok(url)
}
fn build_object_key(
prefix: LegacyAssetPrefix,
path_segments: &[String],
file_name: &str,
) -> String {
let mut parts = Vec::with_capacity(path_segments.len() + 2);
parts.push(prefix.as_str().to_string());
parts.extend(path_segments.iter().cloned());
parts.push(file_name.to_string());
parts.join("/")
}
fn normalize_object_key(raw: &str) -> Result<String, OssError> {
let normalized = raw.trim().trim_start_matches('/').trim().to_string();
if normalized.is_empty() {
return Err(OssError::InvalidRequest("objectKey 不能为空".to_string()));
}
match LegacyAssetPrefix::from_object_key(&normalized) {
Some(LegacyAssetPrefix::EditorAgent) | None => {
return Err(OssError::InvalidRequest(
"objectKey 必须落在受支持的 OSS 前缀下".to_string(),
));
}
Some(_) => {}
}
validate_object_key_segments(&normalized)?;
Ok(normalized)
}
fn normalize_editor_agent_messages_object_key(raw: &str) -> Result<String, OssError> {
let normalized = raw.trim().trim_start_matches('/').trim().to_string();
if normalized.is_empty() {
return Err(OssError::InvalidRequest("objectKey 不能为空".to_string()));
}
validate_object_key_segments(&normalized)?;
let mut segments = normalized.split('/');
let prefix = segments.next();
let file_name = segments.next();
if prefix != Some(LegacyAssetPrefix::EditorAgent.as_str())
|| segments.next().is_some()
|| !file_name
.map(|value| {
value.starts_with("editor-agent-conv-")
&& value.ends_with(".json")
&& value.len() > "editor-agent-conv-.json".len()
})
.unwrap_or(false)
{
return Err(OssError::InvalidRequest(
"objectKey 必须是 editor-agent 消息文档".to_string(),
));
}
Ok(normalized)
}
fn normalize_internal_object_key(raw: &str) -> Result<String, OssError> {
let normalized = raw.trim().trim_start_matches('/').trim().to_string();
validate_object_key_segments(&normalized)?;
if normalized.starts_with("agc/error-reports/v1/") {
Ok(normalized)
} else {
Err(OssError::InvalidRequest(
"objectKey 不属于内部对象前缀".to_string(),
))
}
}
fn validate_object_key_segments(normalized: &str) -> Result<(), OssError> {
let segments = normalized.split('/').collect::<Vec<_>>();
if segments.len() < 2 {
return Err(OssError::InvalidRequest(
"objectKey 至少需要包含前缀和文件名".to_string(),
));
}
for segment in &segments {
if segment.is_empty() || *segment == "." || *segment == ".." {
return Err(OssError::InvalidRequest(
"objectKey 包含非法路径片段".to_string(),
));
}
if segment.contains('\\') {
return Err(OssError::InvalidRequest(
"objectKey 不能包含反斜杠".to_string(),
));
}
}
Ok(())
}
fn build_key_prefix(prefix: LegacyAssetPrefix, path_segments: &[String]) -> String {
let mut parts = Vec::with_capacity(path_segments.len() + 1);
parts.push(prefix.as_str().to_string());
parts.extend(path_segments.iter().cloned());
parts.join("/")
}
fn normalize_metadata(
metadata: BTreeMap<String, String>,
) -> Result<BTreeMap<String, String>, OssError> {
let mut normalized = BTreeMap::new();
for (key, value) in metadata {
let key = key.trim();
let value = value.trim();
if key.is_empty() || value.is_empty() {
continue;
}
let key = normalize_metadata_key(key);
normalized.insert(key, value.to_string());
}
let total_bytes = normalized
.iter()
.map(|(key, value)| key.len() + value.len())
.sum::<usize>();
if total_bytes > DEFAULT_METADATA_TOTAL_BYTES_LIMIT {
return Err(OssError::InvalidRequest(format!(
"x-oss-meta-* 总大小不能超过 {} 字节",
DEFAULT_METADATA_TOTAL_BYTES_LIMIT
)));
}
Ok(normalized)
}
fn build_put_object_headers(
metadata: BTreeMap<String, String>,
) -> Result<BTreeMap<String, String>, OssError> {
// 中文注释:生成资产 object key 含会话与 asset id,内容不可变,适合交给浏览器/CDN 长缓存。
let mut headers = BTreeMap::from([(
"Cache-Control".to_string(),
DEFAULT_IMMUTABLE_CACHE_CONTROL.to_string(),
)]);
headers.extend(normalize_metadata(metadata)?);
Ok(headers)
}
fn normalize_metadata_key(raw: &str) -> String {
let stripped = raw
.trim()
.trim_start_matches("x-oss-meta-")
.trim()
.to_ascii_lowercase();
let sanitized = stripped
.chars()
.map(|character| match character {
'a'..='z' | '0'..='9' | '-' => character,
'_' | ' ' | '/' | '.' => '-',
_ => '-',
})
.collect::<String>();
let sanitized = collapse_dashes(&sanitized);
format!(
"x-oss-meta-{}",
if sanitized.is_empty() {
"metadata".to_string()
} else {
sanitized
}
)
}
fn sanitize_path_segment(raw: &str) -> String {
let normalized = raw
.trim()
.to_ascii_lowercase()
.chars()
.map(|character| match character {
'a'..='z' | '0'..='9' | '-' | '_' => character,
_ => '-',
})
.collect::<String>();
collapse_dashes(&normalized)
}
fn sanitize_file_name(raw: &str) -> Result<String, OssError> {
let trimmed = raw.trim();
if trimmed.is_empty() {
return Err(OssError::InvalidRequest("fileName 不能为空".to_string()));
}
let file_name = trimmed.rsplit(['/', '\\']).next().unwrap_or(trimmed).trim();
if file_name.is_empty() {
return Err(OssError::InvalidRequest("fileName 不能为空".to_string()));
}
let (raw_stem, raw_extension) = match file_name.rsplit_once('.') {
Some((stem, extension)) if !stem.trim().is_empty() && !extension.trim().is_empty() => {
(stem, Some(extension))
}
_ => (file_name, None),
};
let stem = raw_stem
.trim()
.to_ascii_lowercase()
.chars()
.map(|character| match character {
'a'..='z' | '0'..='9' | '-' | '_' => character,
_ => '-',
})
.collect::<String>();
let stem = collapse_dashes(&stem);
let stem = if stem.is_empty() {
"file".to_string()
} else {
stem
};
let extension = raw_extension
.map(|extension| {
extension
.trim()
.to_ascii_lowercase()
.chars()
.filter(|character| character.is_ascii_alphanumeric())
.collect::<String>()
})
.filter(|extension| !extension.is_empty());
Ok(match extension {
Some(extension) => format!("{stem}.{extension}"),
None => stem,
})
}
fn normalize_required_value(value: String, message: &str) -> Result<String, OssError> {
let value = value.trim().to_string();
if value.is_empty() {
return Err(OssError::InvalidConfig(message.to_string()));
}
Ok(value)
}
fn normalize_optional_value(value: Option<String>) -> Option<String> {
value.and_then(|value| {
let value = value.trim().to_string();
if value.is_empty() { None } else { Some(value) }
})
}
fn normalize_endpoint(raw: &str) -> Result<String, OssError> {
let endpoint = raw
.trim()
.trim_start_matches("https://")
.trim_start_matches("http://")
.trim_matches('/')
.to_string();
if endpoint.is_empty() {
return Err(OssError::InvalidConfig("OSS endpoint 不能为空".to_string()));
}
Ok(endpoint)
}
fn collapse_dashes(value: &str) -> String {
value
.chars()
.fold(
(String::new(), false),
|(mut output, last_is_dash), character| {
let is_dash = character == '-';
if is_dash && last_is_dash {
return (output, true);
}
output.push(character);
(output, is_dash)
},
)
.0
.trim_matches('-')
.to_string()
}
async fn send_signed_request(
client: &reqwest::Client,
config: &OssConfig,
method: Method,
object_key: Option<&str>,
target_url: reqwest::Url,
operation: OssRequestOperation,
) -> Result<reqwest::Response, OssError> {
signed_request_builder(
client,
config,
method,
object_key,
target_url,
None,
&BTreeMap::new(),
)?
.send()
.await
.map_err(|error| request_error_from_reqwest(operation, error))
}
fn signed_request_builder(
client: &reqwest::Client,
config: &OssConfig,
method: Method,
object_key: Option<&str>,
target_url: reqwest::Url,
content_type: Option<&str>,
oss_headers: &BTreeMap<String, String>,
) -> Result<reqwest::RequestBuilder, OssError> {
let signed_at = OffsetDateTime::now_utc();
let signed_at_text = build_v4_signature_date(signed_at)?;
let signature_scope = build_v4_signature_scope(config.endpoint(), signed_at)?;
let object_path = object_key.map(str::trim).filter(|value| !value.is_empty());
let canonical_uri = build_v4_canonical_uri(config.bucket(), object_path);
let body_sha256 = OSS_UNSIGNED_PAYLOAD.to_string();
let mut signed_headers = BTreeMap::from([
(
"host".to_string(),
format!("{}.{}", config.bucket(), config.endpoint()),
),
("x-oss-content-sha256".to_string(), body_sha256.clone()),
("x-oss-date".to_string(), signed_at_text.clone()),
]);
if let Some(content_type) = content_type {
signed_headers.insert("content-type".to_string(), content_type.to_string());
}
for (key, value) in oss_headers {
signed_headers.insert(key.to_ascii_lowercase(), value.trim().to_string());
}
let canonical_headers = build_v4_canonical_headers(&signed_headers);
let additional_headers = build_v4_additional_headers(&signed_headers);
let canonical_request = build_v4_canonical_request(
method.as_str(),
&canonical_uri,
"",
&canonical_headers,
&additional_headers,
&body_sha256,
);
let string_to_sign =
build_v4_string_to_sign(&signed_at_text, &signature_scope, &canonical_request);
let signature = sign_v4_content(
config.access_key_secret(),
&signature_scope,
&string_to_sign,
)?;
let mut builder = client
.request(method, target_url)
.header("x-oss-content-sha256", body_sha256)
.header("x-oss-date", signed_at_text)
.header(
"Authorization",
format!(
"{OSS_V4_ALGORITHM} Credential={}/{},AdditionalHeaders={},Signature={}",
config.access_key_id(),
signature_scope,
additional_headers,
signature
),
);
if let Some(content_type) = content_type {
builder = builder.header(reqwest::header::CONTENT_TYPE, content_type);
}
for (key, value) in oss_headers {
builder = builder.header(key.as_str(), value.as_str());
}
Ok(builder)
}
fn build_v4_signature_scope(endpoint: &str, signed_at: OffsetDateTime) -> Result<String, OssError> {
let date = format_v4_signature_scope_date(signed_at);
let region = extract_oss_region(endpoint)?;
Ok(format!("{date}/{region}/{OSS_V4_SERVICE}/{OSS_V4_REQUEST}"))
}
fn build_v4_signature_date(signed_at: OffsetDateTime) -> Result<String, OssError> {
// 中文注释:time::Time 的 Display 在小时小于 10 时不会稳定补零,OSS V4 必须使用固定宽度 UTC 时间。
Ok(format!(
"{}T{:02}{:02}{:02}Z",
format_v4_signature_scope_date(signed_at),
signed_at.hour(),
signed_at.minute(),
signed_at.second()
))
}
fn format_v4_signature_scope_date(signed_at: OffsetDateTime) -> String {
format!(
"{:04}{:02}{:02}",
signed_at.year(),
signed_at.month() as u8,
signed_at.day()
)
}
fn build_v4_canonical_uri(bucket: &str, object_key: Option<&str>) -> String {
match object_key.map(str::trim).filter(|value| !value.is_empty()) {
Some(object_key) => format!(
"/{}/{}",
encode_url_query_value(bucket),
encode_url_path(object_key.trim_start_matches('/'))
),
None => format!("/{}/", encode_url_query_value(bucket)),
}
}
fn extract_oss_region(endpoint: &str) -> Result<String, OssError> {
endpoint
.trim()
.trim_start_matches("https://")
.trim_start_matches("http://")
.split('.')
.next()
.and_then(|segment| segment.strip_prefix("oss-"))
.map(str::to_string)
.filter(|region| !region.is_empty())
.ok_or_else(|| {
OssError::InvalidConfig(format!("OSS endpoint 无法解析 region,当前值:{endpoint}"))
})
}
fn sign_v4_content(
access_key_secret: &str,
signature_scope: &str,
content: &str,
) -> Result<String, OssError> {
let signing_key = build_v4_signing_key(access_key_secret, signature_scope)?;
Ok(hex_sha256_hmac(&signing_key, content.as_bytes()))
}
fn build_v4_signing_key(
access_key_secret: &str,
signature_scope: &str,
) -> Result<Vec<u8>, OssError> {
let mut parts = signature_scope.split('/');
let date = parts
.next()
.ok_or_else(|| OssError::Sign("OSS V4 签名 scope 缺少日期".to_string()))?;
let region = parts
.next()
.ok_or_else(|| OssError::Sign("OSS V4 签名 scope 缺少 region".to_string()))?;
let service = parts
.next()
.ok_or_else(|| OssError::Sign("OSS V4 签名 scope 缺少 service".to_string()))?;
let request = parts
.next()
.ok_or_else(|| OssError::Sign("OSS V4 签名 scope 缺少 request".to_string()))?;
let date_key = hmac_sha256_raw(format!("aliyun_v4{access_key_secret}").as_bytes(), date)?;
let region_key = hmac_sha256_raw(&date_key, region)?;
let service_key = hmac_sha256_raw(&region_key, service)?;
hmac_sha256_raw(&service_key, request)
}
fn hmac_sha256_raw(key: &[u8], content: &str) -> Result<Vec<u8>, OssError> {
let mut signer = HmacSha256::new_from_slice(key)
.map_err(|error| OssError::Sign(format!("初始化 HMAC-SHA256 失败:{error}")))?;
signer.update(content.as_bytes());
Ok(signer.finalize().into_bytes().to_vec())
}
fn hex_sha256_hmac(key: &[u8], content: &[u8]) -> String {
let mut signer = HmacSha256::new_from_slice(key).expect("HMAC-SHA256 accepts keys of any size");
signer.update(content);
hex_lower(&signer.finalize().into_bytes())
}
fn build_v4_canonical_request(
method: &str,
canonical_uri: &str,
canonical_query: &str,
canonical_headers: &str,
signed_headers: &str,
payload_hash: &str,
) -> String {
format!(
"{method}\n{canonical_uri}\n{canonical_query}\n{canonical_headers}\n{signed_headers}\n{payload_hash}"
)
}
fn build_v4_string_to_sign(
signature_date: &str,
signature_scope: &str,
canonical_request: &str,
) -> String {
format!(
"{OSS_V4_ALGORITHM}\n{signature_date}\n{signature_scope}\n{}",
sha256_hex(canonical_request.as_bytes())
)
}
fn sha256_hex(content: &[u8]) -> String {
let mut hasher = Sha256::new();
hasher.update(content);
hex_lower(&hasher.finalize())
}
fn hex_lower(bytes: &[u8]) -> String {
bytes
.iter()
.map(|byte| format!("{byte:02x}"))
.collect::<String>()
}
fn build_v4_canonical_headers(headers: &BTreeMap<String, String>) -> String {
headers
.iter()
.map(|(key, value)| format!("{}:{}\n", key.to_ascii_lowercase(), value.trim()))
.collect::<String>()
}
fn build_v4_additional_headers(headers: &BTreeMap<String, String>) -> String {
let mut additional_headers = headers
.keys()
.map(|key| key.to_ascii_lowercase())
.filter(|key| key != "content-type" && key != "content-md5" && !key.starts_with("x-oss-"))
.collect::<Vec<_>>();
additional_headers.sort();
additional_headers.join(";")
}
fn build_canonical_query_string(params: &BTreeMap<String, String>) -> String {
params
.iter()
.map(|(key, value)| {
format!(
"{}={}",
encode_url_query_value(key),
encode_url_query_value(value)
)
})
.collect::<Vec<_>>()
.join("&")
}
fn encode_url_path(path: &str) -> String {
path.split('/')
.map(encode_url_query_value)
.collect::<Vec<_>>()
.join("/")
}
fn encode_url_query_value(value: &str) -> String {
let mut encoded = String::with_capacity(value.len());
for byte in value.bytes() {
match byte {
b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' => {
encoded.push(byte as char)
}
_ => {
use std::fmt::Write as _;
let _ = write!(&mut encoded, "%{byte:02X}");
}
}
}
encoded
}
#[cfg(test)]
mod tests {
use super::*;
use std::sync::atomic::{AtomicUsize, Ordering};
fn retryable_request_error(
status: Option<u16>,
timeout: bool,
connect: bool,
transport: bool,
) -> OssError {
OssError::Request(OssRequestError {
status,
timeout,
connect,
transport,
oss_code: None,
oss_request_id: None,
operation: OssRequestOperation::Put,
message: "mock request failure".to_string(),
})
}
fn animation_attempt_context(operation: &'static str) -> OssRequestAttemptContext {
OssRequestAttemptContext {
frame_index: 1,
operation,
}
}
#[test]
fn oss_error_kind_is_stable_for_adapter_mapping() {
assert_eq!(
OssError::InvalidConfig("bad config".to_string()).kind(),
OssErrorKind::InvalidConfig
);
assert_eq!(
OssError::ObjectNotFound("missing".to_string()).kind(),
OssErrorKind::ObjectNotFound
);
assert_eq!(
OssError::Request(OssRequestError {
status: None,
timeout: false,
connect: false,
transport: true,
oss_code: None,
oss_request_id: None,
operation: OssRequestOperation::Put,
message: "network".to_string(),
})
.kind(),
OssErrorKind::Request
);
}
#[test]
fn object_not_found_attempt_log_details_keep_404_status() {
let error = OssError::ObjectNotFound("missing".to_string());
assert_eq!(
request_error_details(Some(&error)),
(Some(404), false, false, false, None, None)
);
assert_eq!(oss_error_kind_label(&error), "object_not_found");
assert!(!oss_error_is_retryable(&error));
}
#[test]
fn oss_request_timeout_400_is_retryable_and_prefers_header_request_id() {
let body = br#"<?xml version="1.0" encoding="UTF-8"?>
<Error><Code>RequestTimeout</Code><RequestId>xml-request-id</RequestId></Error>"#;
let error = request_status_error_from_oss_parts(
OssRequestOperation::Put,
400,
Some("header-request-id".to_string()),
body,
None,
);
let OssError::Request(request_error) = &error else {
panic!("OSS status failure should remain a request error");
};
assert_eq!(request_error.status, Some(400));
assert!(request_error.timeout);
assert_eq!(request_error.oss_code.as_deref(), Some("RequestTimeout"));
assert_eq!(
request_error.oss_request_id.as_deref(),
Some("header-request-id")
);
assert!(oss_error_is_retryable(&error));
}
#[test]
fn oss_request_timeout_400_uses_xml_request_id_when_header_is_missing() {
let body = br#"<Error>
<Code>RequestTimeout</Code><RequestId>xml-request-id</RequestId>
</Error>"#;
let error =
request_status_error_from_oss_parts(OssRequestOperation::Put, 400, None, body, None);
let OssError::Request(request_error) = &error else {
panic!("OSS status failure should remain a request error");
};
assert_eq!(
request_error.oss_request_id.as_deref(),
Some("xml-request-id")
);
assert!(oss_error_is_retryable(&error));
}
#[test]
fn other_oss_400_errors_and_malformed_xml_are_not_retryable() {
for body in [
b"<Error><Code>InvalidArgument</Code></Error>".as_slice(),
b"<Error><Code>RequestTimeout".as_slice(),
b"not xml".as_slice(),
] {
let error = request_status_error_from_oss_parts(
OssRequestOperation::Put,
400,
None,
body,
None,
);
assert!(!oss_error_is_retryable(&error));
}
let error = request_status_error_from_oss_parts(
OssRequestOperation::Put,
403,
None,
b"<Error><Code>RequestTimeout</Code></Error>",
None,
);
assert!(!oss_error_is_retryable(&error));
}
#[test]
fn oss_error_xml_fields_beyond_body_limit_are_ignored() {
let mut body = vec![b' '; CHARACTER_ANIMATION_OSS_ERROR_BODY_MAX_BYTES];
body.extend_from_slice(
b"<Error><Code>RequestTimeout</Code><RequestId>late</RequestId></Error>",
);
let error =
request_status_error_from_oss_parts(OssRequestOperation::Put, 400, None, &body, None);
let OssError::Request(request_error) = &error else {
panic!("OSS status failure should remain a request error");
};
assert_eq!(request_error.oss_code, None);
assert_eq!(request_error.oss_request_id, None);
assert!(!request_error.timeout);
assert!(!oss_error_is_retryable(&error));
}
#[test]
fn oss_400_without_code_and_broken_body_read_is_retryable() {
for (read_timeout, expect_transport) in [(true, false), (false, true)] {
let error = request_status_error_from_oss_parts(
OssRequestOperation::Put,
400,
None,
b"<Error><Code>Request",
Some(OssErrorBodyReadFailure {
timeout: read_timeout,
}),
);
let OssError::Request(request_error) = &error else {
panic!("OSS status failure should remain a request error");
};
assert_eq!(request_error.status, Some(400));
assert_eq!(request_error.oss_code, None);
assert_eq!(request_error.timeout, read_timeout);
assert_eq!(request_error.transport, expect_transport);
assert!(request_error.message.contains("错误响应体读取失败"));
assert!(oss_error_is_retryable(&error));
}
}
#[test]
fn oss_400_with_parsed_code_keeps_deterministic_semantics_on_read_failure() {
let error = request_status_error_from_oss_parts(
OssRequestOperation::Put,
400,
None,
b"<Error><Code>InvalidArgument</Code><RequestId>partial",
Some(OssErrorBodyReadFailure { timeout: true }),
);
let OssError::Request(request_error) = &error else {
panic!("OSS status failure should remain a request error");
};
assert_eq!(request_error.oss_code.as_deref(), Some("InvalidArgument"));
assert!(!request_error.timeout);
assert!(!request_error.transport);
assert!(request_error.message.contains("错误响应体读取失败"));
assert!(!oss_error_is_retryable(&error));
let error = request_status_error_from_oss_parts(
OssRequestOperation::Put,
400,
None,
b"<Error><Code>RequestTimeout</Code>",
Some(OssErrorBodyReadFailure { timeout: false }),
);
assert!(oss_error_is_retryable(&error));
}
const MOCK_PUT_BODY: &[u8] = b"animation-frame-bytes";
/// 极简 HTTP/1.1 mock:读完整个 PUT 请求后返回 400 与部分 XML 响应体
/// (Content-Length 大于实际发送字节),`stall_before_close` 决定挂住
/// 连接触发客户端读超时,还是直接断开触发传输错误。
async fn spawn_broken_error_body_server(stall_before_close: bool) -> std::net::SocketAddr {
use tokio::io::{AsyncReadExt, AsyncWriteExt};
let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
.await
.expect("mock server should bind");
let addr = listener
.local_addr()
.expect("mock server should expose its addr");
tokio::spawn(async move {
let Ok((mut socket, _)) = listener.accept().await else {
return;
};
let mut received = Vec::new();
let mut buffer = [0u8; 4096];
while !received.ends_with(MOCK_PUT_BODY) {
match socket.read(&mut buffer).await {
Ok(0) | Err(_) => return,
Ok(read) => received.extend_from_slice(&buffer[..read]),
}
}
let response = "HTTP/1.1 400 Bad Request\r\n\
x-oss-request-id: mock-request-id\r\n\
Content-Type: application/xml\r\n\
Content-Length: 4096\r\n\
\r\n\
<Error><Code>Request";
if socket.write_all(response.as_bytes()).await.is_err() {
return;
}
let _ = socket.flush().await;
if stall_before_close {
tokio::time::sleep(std::time::Duration::from_secs(5)).await;
}
});
addr
}
#[tokio::test]
async fn oss_400_with_broken_error_body_stream_is_retryable_transport() {
let addr = spawn_broken_error_body_server(false).await;
let response = reqwest::Client::new()
.put(format!("http://{addr}/generated-animations/frame01.png"))
.body(MOCK_PUT_BODY.to_vec())
.send()
.await
.expect("response headers should arrive before the body breaks");
assert_eq!(response.status(), reqwest::StatusCode::BAD_REQUEST);
let error = request_status_error_from_oss_put_response(response).await;
let OssError::Request(request_error) = &error else {
panic!("OSS status failure should remain a request error");
};
assert_eq!(request_error.status, Some(400));
assert_eq!(request_error.oss_code, None);
assert!(!request_error.timeout);
assert!(request_error.transport);
assert_eq!(
request_error.oss_request_id.as_deref(),
Some("mock-request-id")
);
assert!(request_error.message.contains("错误响应体读取失败"));
assert!(oss_error_is_retryable(&error));
}
#[tokio::test]
async fn oss_400_with_stalled_error_body_stream_is_retryable_timeout() {
let addr = spawn_broken_error_body_server(true).await;
let client = reqwest::Client::builder()
.timeout(std::time::Duration::from_millis(300))
.build()
.expect("test client should build");
let response = client
.put(format!("http://{addr}/generated-animations/frame01.png"))
.body(MOCK_PUT_BODY.to_vec())
.send()
.await
.expect("response headers should arrive before the body stalls");
assert_eq!(response.status(), reqwest::StatusCode::BAD_REQUEST);
let error = request_status_error_from_oss_put_response(response).await;
let OssError::Request(request_error) = &error else {
panic!("OSS status failure should remain a request error");
};
assert_eq!(request_error.status, Some(400));
assert_eq!(request_error.oss_code, None);
assert!(request_error.timeout);
assert!(!request_error.transport);
assert!(oss_error_is_retryable(&error));
}
#[tokio::test]
async fn reqwest_builder_error_is_not_retryable_transport() {
let error = reqwest::Client::new()
.put("https://example.com")
.header("x-oss-meta-invalid", "first line\nsecond line")
.send()
.await
.expect_err("invalid header must fail while building the request");
assert!(error.is_builder());
let error = request_error_from_reqwest(OssRequestOperation::Put, error);
let OssError::Request(request_error) = &error else {
panic!("builder failure should remain an OSS request error");
};
assert_eq!(request_error.status, None);
assert!(!request_error.timeout);
assert!(!request_error.connect);
assert!(!request_error.transport);
assert!(!oss_error_is_retryable(&error));
}
#[tokio::test]
async fn animation_retry_retries_transport_then_succeeds() {
let attempts = Arc::new(AtomicUsize::new(0));
let attempts_for_request = attempts.clone();
let result = run_animation_request_with_retry(
Arc::new(Semaphore::new(8)),
animation_attempt_context("source_put"),
"generated-animations/editor/layer/task/green-screen-frame01.png",
move || {
let attempts = attempts_for_request.clone();
async move {
if attempts.fetch_add(1, Ordering::SeqCst) == 0 {
Err(retryable_request_error(None, false, false, true))
} else {
Ok(("uploaded", 201))
}
}
},
)
.await;
assert_eq!(result.expect("second attempt should succeed"), "uploaded");
assert_eq!(attempts.load(Ordering::SeqCst), 2);
}
#[tokio::test]
async fn animation_retry_retries_timeout_and_retryable_statuses() {
for (status, timeout, connect, transport) in [
(None, true, false, false),
(Some(408), false, false, false),
(Some(429), false, false, false),
(Some(500), false, false, false),
(Some(502), false, false, false),
(Some(503), false, false, false),
(Some(504), false, false, false),
] {
let attempts = Arc::new(AtomicUsize::new(0));
let attempts_for_request = attempts.clone();
let result = run_animation_request_with_retry(
Arc::new(Semaphore::new(8)),
animation_attempt_context("final_put"),
"generated-animations/editor/layer/task/frame01.png",
move || {
let attempts = attempts_for_request.clone();
async move {
if attempts.fetch_add(1, Ordering::SeqCst) == 0 {
Err(retryable_request_error(status, timeout, connect, transport))
} else {
Ok(((), 204))
}
}
},
)
.await;
assert!(result.is_ok(), "status={status:?} should retry");
assert_eq!(attempts.load(Ordering::SeqCst), 2, "status={status:?}");
}
}
#[tokio::test]
async fn animation_retry_does_not_retry_deterministic_statuses() {
for status in [400, 403, 404] {
let attempts = Arc::new(AtomicUsize::new(0));
let attempts_for_request = attempts.clone();
let result = run_animation_request_with_retry(
Arc::new(Semaphore::new(8)),
animation_attempt_context("final_head"),
"generated-animations/editor/layer/task/frame01.png",
move || {
let attempts = attempts_for_request.clone();
async move {
attempts.fetch_add(1, Ordering::SeqCst);
Err::<((), u16), _>(retryable_request_error(
Some(status),
false,
false,
false,
))
}
},
)
.await;
assert!(result.is_err(), "status={status} should fail");
assert_eq!(attempts.load(Ordering::SeqCst), 1, "status={status}");
}
}
#[tokio::test]
async fn animation_retry_returns_final_error_after_three_attempts() {
let attempts = Arc::new(AtomicUsize::new(0));
let attempts_for_request = attempts.clone();
let result = run_animation_request_with_retry(
Arc::new(Semaphore::new(8)),
animation_attempt_context("final_put"),
"generated-animations/editor/layer/task/frame01.png",
move || {
let attempts = attempts_for_request.clone();
async move {
attempts.fetch_add(1, Ordering::SeqCst);
Err::<((), u16), _>(retryable_request_error(Some(503), false, false, false))
}
},
)
.await;
assert!(result.is_err());
assert_eq!(attempts.load(Ordering::SeqCst), 3);
}
#[tokio::test]
async fn animation_retry_keeps_head_retry_independent_from_successful_put() {
let put_attempts = Arc::new(AtomicUsize::new(0));
let put_attempts_for_request = put_attempts.clone();
run_animation_request_with_retry(
Arc::new(Semaphore::new(8)),
animation_attempt_context("final_put"),
"generated-animations/editor/layer/task/frame01.png",
move || {
put_attempts_for_request.fetch_add(1, Ordering::SeqCst);
async { Ok::<_, OssError>(((), 204)) }
},
)
.await
.expect("PUT should succeed once");
let head_attempts = Arc::new(AtomicUsize::new(0));
let head_attempts_for_request = head_attempts.clone();
run_animation_request_with_retry(
Arc::new(Semaphore::new(8)),
animation_attempt_context("final_head"),
"generated-animations/editor/layer/task/frame01.png",
move || {
let head_attempts = head_attempts_for_request.clone();
async move {
if head_attempts.fetch_add(1, Ordering::SeqCst) == 0 {
Err(retryable_request_error(Some(503), false, false, false))
} else {
Ok(((), 204))
}
}
},
)
.await
.expect("HEAD should succeed on its retry");
assert_eq!(put_attempts.load(Ordering::SeqCst), 1);
assert_eq!(head_attempts.load(Ordering::SeqCst), 2);
}
#[tokio::test]
async fn animation_retry_keeps_permit_bound_at_eight_in_flight_requests() {
let limiter = Arc::new(Semaphore::new(8));
let current = Arc::new(AtomicUsize::new(0));
let maximum = Arc::new(AtomicUsize::new(0));
let mut tasks = Vec::new();
for frame_index in 0..32 {
let limiter = limiter.clone();
let current_for_request = current.clone();
let maximum_for_request = maximum.clone();
tasks.push(tokio::spawn(async move {
run_animation_request_with_retry(
limiter,
OssRequestAttemptContext {
frame_index: frame_index + 1,
operation: "source_put",
},
"generated-animations/editor/layer/task/green-screen-frame01.png",
move || {
let current = current_for_request.clone();
let maximum = maximum_for_request.clone();
async move {
let in_flight = current.fetch_add(1, Ordering::SeqCst) + 1;
maximum.fetch_max(in_flight, Ordering::SeqCst);
tokio::time::sleep(std::time::Duration::from_millis(5)).await;
current.fetch_sub(1, Ordering::SeqCst);
Ok::<_, OssError>(((), 204))
}
},
)
.await
}));
}
for task in tasks {
task.await
.expect("mock request task should join")
.expect("request should pass");
}
assert!(maximum.load(Ordering::SeqCst) <= 8);
}
#[tokio::test]
async fn animation_retry_reuses_object_key_and_body_across_attempts() {
let client = OssClient::new(
OssConfig::new(
"bucket".to_string(),
"oss-cn-shanghai.aliyuncs.com".to_string(),
"access-key".to_string(),
"access-secret".to_string(),
60,
60,
1024,
204,
)
.expect("test OSS config should be valid"),
);
let prepared = client
.prepare_put_object(OssPutObjectRequest {
prefix: LegacyAssetPrefix::Animations,
path_segments: vec![
"editor".to_string(),
"layer".to_string(),
"task".to_string(),
],
file_name: "frame01.png".to_string(),
content_type: Some("image/png".to_string()),
access: OssObjectAccess::Private,
metadata: BTreeMap::new(),
body: vec![1, 2, 3, 4],
})
.expect("test request should be prepared");
let expected_key = prepared.object_key.clone();
let expected_body = prepared.body.clone();
let expected_key_for_request = expected_key.clone();
let attempts = Arc::new(AtomicUsize::new(0));
let attempts_for_request = attempts.clone();
let result = run_animation_request_with_retry(
Arc::new(Semaphore::new(8)),
animation_attempt_context("final_put"),
&expected_key_for_request,
move || {
let attempts = attempts_for_request.clone();
let key = prepared.object_key.clone();
let body = prepared.body.clone();
let expected_key = expected_key.clone();
let expected_body = expected_body.clone();
async move {
assert_eq!(key, expected_key);
assert_eq!(body, expected_body);
if attempts.fetch_add(1, Ordering::SeqCst) == 0 {
Err(retryable_request_error(Some(503), false, false, false))
} else {
Ok(((), 204))
}
}
},
)
.await;
assert!(result.is_ok());
assert_eq!(attempts.load(Ordering::SeqCst), 2);
}
#[test]
fn structured_log_labels_are_stable() {
assert_eq!(
oss_error_kind_label(&OssError::InvalidRequest("bad input".to_string())),
"invalid_request"
);
assert_eq!(oss_access_label(OssObjectAccess::Private), "private");
assert_eq!(http_status_class(204), "2xx");
assert_eq!(http_status_class(404), "4xx");
assert_eq!(http_status_class_from_option(None), "unknown");
}
fn build_client() -> OssClient {
OssClient::new(
OssConfig::new(
"genarrative-assets".to_string(),
"oss-cn-shanghai.aliyuncs.com".to_string(),
"test-access-key-id".to_string(),
"test-access-key-secret".to_string(),
DEFAULT_READ_EXPIRE_SECONDS,
DEFAULT_POST_EXPIRE_SECONDS,
DEFAULT_POST_MAX_SIZE_BYTES,
DEFAULT_SUCCESS_ACTION_STATUS,
)
.expect("OSS config should be valid"),
)
}
#[test]
fn parse_legacy_prefix_accepts_public_style_path() {
assert_eq!(
LegacyAssetPrefix::parse("/generated-characters/*"),
Some(LegacyAssetPrefix::Characters)
);
assert_eq!(
LegacyAssetPrefix::parse("/generated-puzzle-assets/*"),
Some(LegacyAssetPrefix::PuzzleAssets)
);
assert_eq!(
LegacyAssetPrefix::parse("/generated-match3d-assets/*"),
Some(LegacyAssetPrefix::Match3DAssets)
);
assert_eq!(
LegacyAssetPrefix::parse("/generated-wooden-fish-assets/*"),
Some(LegacyAssetPrefix::WoodenFishAssets)
);
assert!(LEGACY_PUBLIC_PREFIXES.contains(&"generated-puzzle-assets"));
assert!(LEGACY_PUBLIC_PREFIXES.contains(&"generated-match3d-assets"));
assert!(LEGACY_PUBLIC_PREFIXES.contains(&"generated-bark-battle-assets"));
assert!(LEGACY_PUBLIC_PREFIXES.contains(&"generated-wooden-fish-assets"));
assert_eq!(LegacyAssetPrefix::parse("unknown"), None);
}
#[test]
fn build_v4_signature_date_zero_pads_single_digit_time_parts() {
let signed_at =
OffsetDateTime::from_unix_timestamp(1_771_477_389).expect("timestamp should be valid");
assert_eq!(
build_v4_signature_date(signed_at).expect("date should format"),
"20260219T050309Z"
);
assert_eq!(
build_v4_signature_scope("oss-cn-shanghai.aliyuncs.com", signed_at)
.expect("scope should format"),
"20260219/cn-shanghai/oss/aliyun_v4_request"
);
}
#[test]
fn sign_post_object_returns_bucket_and_object_key_for_private_storage_truth() {
let client = build_client();
let mut metadata = BTreeMap::new();
metadata.insert("asset-kind".to_string(), "character-visual".to_string());
metadata.insert("origin".to_string(), "browser-upload".to_string());
let response = client
.sign_post_object(OssPostObjectRequest {
prefix: LegacyAssetPrefix::Characters,
path_segments: vec![
"Hero_001".to_string(),
"Visual".to_string(),
"Asset_01".to_string(),
],
file_name: "Master.PNG".to_string(),
content_type: Some("image/png".to_string()),
access: OssObjectAccess::Public,
metadata,
max_size_bytes: Some(5 * 1024 * 1024),
expire_seconds: Some(300),
success_action_status: Some(200),
})
.expect("post object signature should build");
assert_eq!(
response.object_key,
"generated-characters/hero_001/visual/asset_01/master.png"
);
assert_eq!(
response.legacy_public_path,
"/generated-characters/hero_001/visual/asset_01/master.png"
);
assert_eq!(response.bucket, "genarrative-assets".to_string());
assert_eq!(
response.form_fields.signature_version,
OSS_V4_ALGORITHM.to_string()
);
assert!(
response
.form_fields
.credential
.starts_with("test-access-key-id/")
);
assert!(
response
.form_fields
.credential
.ends_with("/cn-shanghai/oss/aliyun_v4_request")
);
assert_eq!(response.form_fields.date.len(), "20260507T120000Z".len());
assert_eq!(
response.form_fields.metadata.get("x-oss-meta-asset-kind"),
Some(&"character-visual".to_string())
);
}
#[test]
fn sign_post_object_embeds_policy_constraints() {
let client = build_client();
let response = client
.sign_post_object(OssPostObjectRequest {
prefix: LegacyAssetPrefix::QwenSprites,
path_segments: vec!["_drafts".to_string(), "master".to_string()],
file_name: "candidate-01.png".to_string(),
content_type: Some("image/png".to_string()),
access: OssObjectAccess::Private,
metadata: BTreeMap::new(),
max_size_bytes: Some(1024),
expire_seconds: Some(60),
success_action_status: Some(200),
})
.expect("post object signature should build");
let decoded_policy = BASE64_STANDARD
.decode(response.form_fields.policy.as_bytes())
.expect("policy should be valid base64");
let policy: Value =
serde_json::from_slice(&decoded_policy).expect("policy should be valid json");
assert_eq!(
policy["conditions"][0]["bucket"],
Value::String("genarrative-assets".to_string())
);
assert_eq!(
policy["conditions"][1],
json!([
"eq",
"$key",
"generated-qwen-sprites/_drafts/master/candidate-01.png"
])
);
assert_eq!(
policy["conditions"][2],
json!(["content-length-range", 1, 1024])
);
assert_eq!(
policy["conditions"][3],
json!(["eq", "$success_action_status", "200"])
);
assert_eq!(
policy["conditions"][4],
json!(["eq", "$x-oss-signature-version", "OSS4-HMAC-SHA256"])
);
assert_eq!(
policy["conditions"][5],
json!(["eq", "$x-oss-credential", response.form_fields.credential])
);
assert_eq!(
policy["conditions"][6],
json!(["eq", "$x-oss-date", response.form_fields.date])
);
assert_eq!(
policy["conditions"][7],
json!(["eq", "$content-type", "image/png"])
);
assert_eq!(
policy["conditions"][8],
json!(["eq", "$Cache-Control", DEFAULT_IMMUTABLE_CACHE_CONTROL])
);
assert_eq!(
response.form_fields.cache_control,
Some(DEFAULT_IMMUTABLE_CACHE_CONTROL.to_string())
);
assert_eq!(response.bucket, "genarrative-assets".to_string());
}
#[test]
fn sanitize_file_name_rejects_empty_input() {
let error = sanitize_file_name(" ").expect_err("empty file name should fail");
assert_eq!(
error,
OssError::InvalidRequest("fileName 不能为空".to_string())
);
}
#[test]
fn sanitize_file_name_falls_back_when_stem_has_no_ascii_body() {
assert_eq!(
sanitize_file_name("剪贴板素材.png").expect("file name should sanitize"),
"file.png"
);
}
#[test]
fn sign_get_object_url_returns_signed_private_read_url() {
let client = build_client();
let response = client
.sign_get_object_url(OssSignedGetObjectUrlRequest {
object_key: "generated-characters/hero_001/visual/asset_01/master.png".to_string(),
expire_seconds: Some(300),
})
.expect("signed get url should build");
assert_eq!(response.bucket, "genarrative-assets".to_string());
assert_eq!(
response.object_key,
"generated-characters/hero_001/visual/asset_01/master.png".to_string()
);
assert!(response
.signed_url
.starts_with("https://genarrative-assets.oss-cn-shanghai.aliyuncs.com/generated-characters/hero_001/visual/asset_01/master.png?"));
assert!(
response
.signed_url
.contains("x-oss-signature-version=OSS4-HMAC-SHA256")
);
assert!(
response
.signed_url
.contains("x-oss-credential=test-access-key-id%2F")
);
assert!(response.signed_url.contains("&x-oss-expires=300"));
assert!(response.signed_url.contains("&x-oss-signature="));
}
#[test]
fn sign_get_object_url_uses_square_hole_object_key_without_bucket_prefix() {
let client = OssClient::new(
OssConfig::new(
"xushi-dev".to_string(),
"oss-cn-shanghai.aliyuncs.com".to_string(),
"test-access-key-id".to_string(),
"test-access-key-secret".to_string(),
DEFAULT_READ_EXPIRE_SECONDS,
DEFAULT_POST_EXPIRE_SECONDS,
DEFAULT_POST_MAX_SIZE_BYTES,
DEFAULT_SUCCESS_ACTION_STATUS,
)
.expect("OSS config should be valid"),
);
let response = client
.sign_get_object_url(OssSignedGetObjectUrlRequest {
object_key: "generated-square-hole-assets/square-hole-session-546d881972684be2980a2a882cd0cc71/square-hole-profile-134411276ce1469cbe398f946a25d7f8/square-hole-shape-image/rabbit-option/asset-1777979289912039/image.png".to_string(),
expire_seconds: Some(300),
})
.expect("square hole object key should build signed url");
assert_eq!(response.bucket, "xushi-dev".to_string());
assert_eq!(
response.object_key,
"generated-square-hole-assets/square-hole-session-546d881972684be2980a2a882cd0cc71/square-hole-profile-134411276ce1469cbe398f946a25d7f8/square-hole-shape-image/rabbit-option/asset-1777979289912039/image.png".to_string()
);
assert!(response
.signed_url
.starts_with("https://xushi-dev.oss-cn-shanghai.aliyuncs.com/generated-square-hole-assets/square-hole-session-546d881972684be2980a2a882cd0cc71/square-hole-profile-134411276ce1469cbe398f946a25d7f8/square-hole-shape-image/rabbit-option/asset-1777979289912039/image.png?"));
}
#[test]
fn sign_get_object_url_rejects_unsupported_prefix() {
let client = build_client();
let error = client
.sign_get_object_url(OssSignedGetObjectUrlRequest {
object_key: "workflow-cache/task-1.json".to_string(),
expire_seconds: Some(300),
})
.expect_err("unsupported prefix should fail");
assert_eq!(
error,
OssError::InvalidRequest("objectKey 必须落在受支持的 OSS 前缀下".to_string())
);
}
#[test]
fn sign_internal_get_object_url_allows_editor_agent_messages_document() {
let client = build_client();
let response = client
.sign_internal_get_object_url(OssSignedGetObjectUrlRequest {
object_key: "editor-agent/editor-agent-conv-1.json".to_string(),
expire_seconds: Some(300),
})
.expect("editor agent messages document should build signed url");
assert_eq!(
response.object_key,
"editor-agent/editor-agent-conv-1.json".to_string()
);
assert!(response.signed_url.starts_with(
"https://genarrative-assets.oss-cn-shanghai.aliyuncs.com/editor-agent/editor-agent-conv-1.json?"
));
}
#[test]
fn sign_internal_get_object_url_rejects_non_editor_agent_messages_document() {
let client = build_client();
for object_key in [
"workflow-cache/task-1.json",
"generated-characters/hero/master.png",
"editor-agent/editor-agent-conv-1/part.json",
"editor-agent/other.json",
"editor-agent/editor-agent-conv-1.png",
] {
let error = client
.sign_internal_get_object_url(OssSignedGetObjectUrlRequest {
object_key: object_key.to_string(),
expire_seconds: Some(300),
})
.expect_err("non editor agent messages document should fail");
assert_eq!(
error,
OssError::InvalidRequest("objectKey 必须是 editor-agent 消息文档".to_string())
);
}
}
#[test]
fn legacy_prefix_can_be_resolved_from_object_key() {
assert_eq!(
LegacyAssetPrefix::from_object_key(
"generated-custom-world-scenes/profile_01/landmark_01/scene.png"
),
Some(LegacyAssetPrefix::CustomWorldScenes)
);
assert_eq!(
LegacyAssetPrefix::from_object_key(
"generated-wooden-fish-assets/session/profile/hit_object/asset/image.png"
),
Some(LegacyAssetPrefix::WoodenFishAssets)
);
assert_eq!(
LegacyAssetPrefix::from_object_key("workflow-cache/demo.json"),
None
);
}
#[test]
fn put_object_request_reuses_generated_object_key_contract() {
let request = OssPutObjectRequest {
prefix: LegacyAssetPrefix::CustomWorldCovers,
path_segments: vec!["Profile 001".to_string(), "asset_01".to_string()],
file_name: "Cover.PNG".to_string(),
content_type: Some(" image/png ".to_string()),
access: OssObjectAccess::Private,
metadata: BTreeMap::from([
("asset_kind".to_string(), "custom_world_cover".to_string()),
("source job id".to_string(), "job_001".to_string()),
]),
body: b"cover-bytes".to_vec(),
};
let sanitized_segments = request
.path_segments
.iter()
.map(|segment| sanitize_path_segment(segment))
.filter(|segment| !segment.is_empty())
.collect::<Vec<_>>();
let file_name = sanitize_file_name(&request.file_name).expect("file name should sanitize");
let object_key = build_object_key(request.prefix, &sanitized_segments, &file_name);
let metadata = normalize_metadata(request.metadata).expect("metadata should normalize");
assert_eq!(
object_key,
"generated-custom-world-covers/profile-001/asset_01/cover.png"
);
assert_eq!(
metadata.get("x-oss-meta-asset-kind"),
Some(&"custom_world_cover".to_string())
);
assert_eq!(
metadata.get("x-oss-meta-source-job-id"),
Some(&"job_001".to_string())
);
}
#[test]
fn canonicalized_oss_headers_matches_sorted_v4_header_shape() {
let headers = BTreeMap::from([
(
"Cache-Control".to_string(),
DEFAULT_IMMUTABLE_CACHE_CONTROL.to_string(),
),
(
"x-oss-meta-source-job-id".to_string(),
" job_001 ".to_string(),
),
(
"x-oss-meta-asset-kind".to_string(),
"character_visual".to_string(),
),
]);
assert_eq!(
build_v4_canonical_headers(&headers),
"cache-control:public, max-age=31536000, immutable\nx-oss-meta-asset-kind:character_visual\nx-oss-meta-source-job-id:job_001\n"
);
}
#[test]
fn additional_headers_include_plain_headers_and_skip_oss_managed_headers() {
let headers = BTreeMap::from([
(
"host".to_string(),
"genarrative-assets.oss-cn-beijing.aliyuncs.com".to_string(),
),
("content-type".to_string(), "image/png".to_string()),
(
"Cache-Control".to_string(),
DEFAULT_IMMUTABLE_CACHE_CONTROL.to_string(),
),
("x-oss-date".to_string(), "20260507T120000Z".to_string()),
(
"x-oss-meta-asset-kind".to_string(),
"puzzle-cover".to_string(),
),
]);
assert_eq!(build_v4_additional_headers(&headers), "cache-control;host");
}
#[test]
fn put_object_headers_include_immutable_cache_control_for_generated_assets() {
let headers = build_put_object_headers(BTreeMap::from([(
"asset-kind".to_string(),
"puzzle-cover".to_string(),
)]))
.expect("headers should build");
assert_eq!(
headers.get("Cache-Control"),
Some(&DEFAULT_IMMUTABLE_CACHE_CONTROL.to_string())
);
assert_eq!(
headers.get("x-oss-meta-asset-kind"),
Some(&"puzzle-cover".to_string())
);
}
#[tokio::test]
async fn put_object_rejects_empty_body_before_calling_oss() {
let client = build_client();
let error = client
.put_object(
&reqwest::Client::new(),
OssPutObjectRequest {
prefix: LegacyAssetPrefix::Characters,
path_segments: vec!["hero".to_string()],
file_name: "master.png".to_string(),
content_type: Some("image/png".to_string()),
access: OssObjectAccess::Private,
metadata: BTreeMap::new(),
body: Vec::new(),
},
)
.await
.expect_err("empty server upload should fail before network");
assert_eq!(
error,
OssError::InvalidRequest("服务端上传对象内容不能为空".to_string())
);
}
}