GenarrativeAI/Genarrative
10
0
Fork 0
Code Issues 20 Pull Requests 8 Actions Packages Projects 1 Releases Wiki Activity

3D 产物下载的重试改为覆盖整个响应体

Browse Source 
- download_model / download_rendered_image 把「取响应头 + 读完 body」作为一次可重试事务，
  每次尝试都重新取响应头并整体重下，不做断点续传
- body 传输失败与长度不一致归一成可重试的传输错误（不再带 HTTP 200），HTTP 状态只留在文案里
- 新增 TripoArtifactBytes（读完的产物）并公开 download_model / download_rendered_image 的 max_bytes 上限；
  超限按 OutputSchema 失败且不重试，流式句柄 TripoDownloadedArtifact 收回 crate 内部
- api-server 的 read_artifact 改成对完整产物做落库前校验（只判空），体积上限转由调用方传入
- smoke 示例改为一次写入完整字节；新增「body 中途断开后整体重下成功」与「超限不重试」两个用例
- 同步技术方案与实施计划：把「只重试响应头」的描述改成「重试包住整个 body」
This commit is contained in:
王德宇
2026-09-23 17:01:40 +08:00
parent 347d066fc5
commit b33a8a141c
9 changed files with 254 additions and 101 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/project-memory/plans/【实施计划】Tripo生成Worker执行链路与API路由-2026-09-21.md
+1 -1
View File
@@ -49,7 +49,7 @@ Milestone Spec: `docs/project-memory/plans/【里程碑】Tripo生成Worker执
- 验收：跨 owner 的资源 / 素材被拒绝。
5. **产物落库**
- provider 产物下载用 `next_chunk` 收进 `Vec<u8>`（流式直传 OSS 留 TODO），模型与预览分别 OSS 写入并记录 `content_type` / `content_length` / `sha256`。
- provider 产物下载由 `platform-tripo` 的下载入口读完整个 body（body 中途失败整体重下，体积上限由调用方传入），模型与预览分别 OSS 写入并记录 `content_type` / `content_length` / `sha256`（流式直传 OSS 留 TODO）。
- 复用现有原子落库路径写入 `assetKind = "model3d"` 的资源 / 素材、预览图引用与 job 终态；不新增资源列。
- 按「新增编辑器 `assetKind` 接入清单」对照表执行：把两个 Tripo job kind 加进 `EDITOR_GENERATION_OPERATION_KINDS`（否则原子落库整笔事务被拒）；画布回填只走 placement-only，`model3d` 不进 `EDITOR_CANVAS_ASSET_KINDS`，也不动前端 `CANVAS_ASSET_KIND_TAG_OPTIONS`。
- 验收：两个 job kind 在白名单内；`canvasCompletion` 回填的 layer 只有 `layerId` / `resourceId`，无 `assetKind` / `src` / `objectKey`；用户标签覆盖、快速编辑、改造 capability 三项保持不进入。
docs/technical/【技术方案】Tripo 3D模型Provider集成-2026-09-18.md
+2 -2
View File
@@ -44,7 +44,7 @@ TODO：图生输入当前由 api-server 自己读站内对象的字节再上传
`get_task` 按 `TripoTaskHandle` 做单次查询，返回通用 `TripoTaskSnapshot`，由 `task_type` 决定 `output` 的具体 variant；adapter 不做轮询、不阻塞等待。
`download_model` 接受 `TripoTaskSnapshot`，先确认快照确实处于完成态，再从严格 endpoint 结果中取得已校验的 `model_url`，由 provider 自己的无鉴权 reqwest client 打开签名 URL 并返回 `TripoDownloadedArtifact` 流包装。包装只公开 `url`、`content_type`、`content_length`、`filename(name)` 和 `next_chunk()`；不提供完整 `Vec<u8>`，调用方必须逐块消费。`filename(name)` 的扩展名来自远端地址，只接受短的 ASCII 字母数字，其余退回 `glb`，避免远端地址里的 `%2F` 解码后拼出跨目录路径。provider 在流结束时校验实际接收字节数与 `Content-Length`，响应体中断或长度不一致按结构化请求错误失败。产物下载不设总超时：几十 MB 的流只要还在出数据就不该被判失败、更不该重试后从零重传，因此下载链路只按「无数据推进」判超时，预算取 `TripoSettings::request_timeout`（连接用 `connect_timeout`，响应体用 `read_timeout`，响应头之前由显式的 `tokio::time::timeout` 兜住），超过预算没有数据推进即按传输失败返回；SDK 保持第三方原样，不承担产物下载。smoke example 使用异步文件写入逐块落盘；未来接入 OSS 时应把同一数据流直接送入 OSS 分片上传，不经过完整内存缓冲。
`download_model`（及预览图的 `download_rendered_image`）接受 `TripoTaskSnapshot` 与调用方给的体积上限 `max_bytes`，先确认快照确实处于完成态，再从严格 endpoint 结果中取得已校验的产物地址，由 provider 自己的无鉴权 reqwest client 打开签名 URL，读完整个 body 后返回 `TripoArtifactBytes`（`url`、`content_type`、`content_length`、`bytes`、`filename(name)`）。流式句柄 `TripoDownloadedArtifact`（`next_chunk()`）保留为内部实现：**重试必须包住整个 body** —— 只重试「拿到响应头」这一步的时候，几十 MB 的字节其实是在之后才传输的，一次 CDN 抖动就会毁掉一次已经扣费、provider 任务也跑完的生成。因此每次尝试都重新取响应头并整体重下（不做断点续传，签名地址会过期），body 读取失败与长度不一致都归一成可重试的传输错误，重试上限与退避沿用 `TripoSettings::retries`。`filename(name)` 的扩展名来自远端地址，只接受短的 ASCII 字母数字，其余退回 `glb`，避免远端地址里的 `%2F` 解码后拼出跨目录路径。每次读取都校验实际接收字节数与 `Content-Length`，读到 `max_bytes` 之上直接按输出违约失败（不重试）：宁可失败退款也不要把 api-server 内存打满。产物下载不设总超时：几十 MB 的流只要还在出数据就不该被判失败，重下只发生在传输真的断了的时候，因此下载链路只按「无数据推进」判超时，预算取 `TripoSettings::request_timeout`（连接用 `connect_timeout`，响应体用 `read_timeout`，响应头之前由显式的 `tokio::time::timeout` 兜住），超过预算没有数据推进即按传输失败返回；SDK 保持第三方原样，不承担产物下载。smoke example 一次写入完整字节；未来接入 OSS 时应把同一数据流直接送入 OSS 分片上传，不经过完整内存缓冲。
TODO：等待上游 `tripo-rust-sdk` 提供原生 artifact stream API 后，删除 provider-side reqwest 下载器，改由 SDK stream 直接承接。
@@ -55,5 +55,5 @@ TODO：等待上游 `tripo-rust-sdk` 提供原生 artifact stream API 后，删
- 三个入口的输入校验可拒绝空白 `prompt` / `input`、视图不足两张和空白 `taskId`；组合校验在调用 SDK 前完成。
- provider 错误统一为 adapter 错误类型。
- shared contracts 的 ts-rs binding 无 feature 开关且始终可生成。
- 产物下载只按「无数据推进」判超时：连接、响应头与每个数据块都必须有进展，慢速但持续的下载可完整收完，中途断流按请求错误失败。
- 产物下载只按「无数据推进」判超时：连接、响应头与每个数据块都必须有进展，慢速但持续的下载可完整收完，中途断流按可重试的传输错误失败并由下载入口整体重下。
- 真实 Provider smoke 已覆盖三个入口；example 在任务未完成时跳过下载并继续跑后续入口，只打印脱敏后的结果结构和下载产物信息。
server-rs/crates/api-server/src/tripo3d/artifacts.rs
+12 -36
View File
@@ -1,45 +1,31 @@
//! provider 产物的读取。
//!
//! provider 侧提供的是流式句柄，这里按字节收口成 `Vec<u8>` 交给 OSS 写入。
//! 下载与「body 中途失败整体重下」都在 `platform-tripo` 的下载入口里完成（重试上限
//! 取自 provider 客户端配置），这里只做 api-server 自己的口径：产物不能为空。
//! TODO(stream): provider SDK 与 `platform-oss` 都支持流式 / 分片后，这里应当直接
//! 把流转交 OSS，不再把几十 MB 的模型完整读进 api-server 内存。
use axum::http::StatusCode;
use platform_tripo::TripoDownloadedArtifact;
use platform_tripo::TripoArtifactBytes;
use serde_json::json;
use crate::http_error::AppError;
use super::{errors::map_provider_error, provider::TRIPO_PROVIDER};
use super::provider::TRIPO_PROVIDER;
/// 单次 attempt 内允许读取的单个产物上限。超过它说明响应异常或产物口径变了，
/// 单次下载允许的单个产物上限。超过它说明响应异常或产物口径变了，
/// 宁可失败退款也不要把 worker 的内存打满。
const MODEL3D_MAX_ARTIFACT_BYTES: u64 = 512 * 1024 * 1024;
pub(crate) const MODEL3D_MAX_ARTIFACT_BYTES: u64 = 512 * 1024 * 1024;
pub(crate) struct DownloadedArtifact {
pub(crate) bytes: Vec<u8>,
pub(crate) content_type: String,
}
pub(crate) async fn read_artifact(
mut artifact: TripoDownloadedArtifact,
) -> Result<DownloadedArtifact, AppError> {
let content_type = artifact.content_type.take().unwrap_or_default();
// 按声明长度预分配（封顶到单产物上限，防止 provider 谎报长度撑爆内存）：
// 几百 MB 的模型逐块 append 会反复搬移，预分配把峰值与拷贝都压下来。
let claimed_bytes = artifact
.content_length
.unwrap_or(0)
.min(MODEL3D_MAX_ARTIFACT_BYTES) as usize;
let mut bytes = Vec::with_capacity(claimed_bytes);
while let Some(chunk) = artifact.next_chunk().await.map_err(map_provider_error)? {
let next_len = bytes.len() as u64 + chunk.len() as u64;
if next_len > MODEL3D_MAX_ARTIFACT_BYTES {
return Err(artifact_too_large(next_len));
}
bytes.extend_from_slice(chunk.as_ref());
}
if bytes.is_empty() {
/// 完整读出的产物在落库前的最后一道检查：空产物按上游内容不合法失败，
/// 不写出一个 0 字节的对象再让下游去猜。
pub(crate) fn read_artifact(artifact: TripoArtifactBytes) -> Result<DownloadedArtifact, AppError> {
if artifact.bytes.is_empty() {
return Err(
AppError::from_status(StatusCode::BAD_GATEWAY).with_details(json!({
"provider": TRIPO_PROVIDER,
@@ -49,17 +35,7 @@ pub(crate) async fn read_artifact(
);
}
Ok(DownloadedArtifact {
bytes,
content_type,
bytes: artifact.bytes,
content_type: artifact.content_type.unwrap_or_default(),
})
}
fn artifact_too_large(actual_bytes: u64) -> AppError {
AppError::from_status(StatusCode::BAD_GATEWAY).with_details(json!({
"provider": TRIPO_PROVIDER,
"reason": "model3d-artifact-too-large",
"message": "provider 返回的 3D 产物超过单次尝试允许的体积上限。",
"actualBytes": actual_bytes,
"maxBytes": MODEL3D_MAX_ARTIFACT_BYTES,
}))
}
server-rs/crates/api-server/src/tripo3d/worker.rs
+5 -7
View File
@@ -33,7 +33,7 @@ use crate::{
};
use super::{
artifacts::read_artifact,
artifacts::{MODEL3D_MAX_ARTIFACT_BYTES, read_artifact},
errors::map_provider_error,
image_source::resolve_image_input,
job::{MODEL3D_PROVIDER_KIND, Model3dJobRequest, parse_model3d_job_request},
@@ -85,18 +85,16 @@ async fn run_model3d_job(
let snapshot = poll_until_terminal(&client, &handle, provider_deadline).await?;
let model = read_artifact(
client
.download_model(&snapshot)
.download_model(&snapshot, MODEL3D_MAX_ARTIFACT_BYTES)
.await
.map_err(map_provider_error)?,
)
.await?;
)?;
let preview = read_artifact(
client
.download_rendered_image(&snapshot)
.download_rendered_image(&snapshot, MODEL3D_MAX_ARTIFACT_BYTES)
.await
.map_err(map_provider_error)?,
)
.await?;
)?;
let (preview_width, preview_height) = preview_dimensions(preview.bytes.as_slice())?;
let stored_model = store_model3d_artifact(
state,
server-rs/crates/platform-tripo/examples/tripo_generation_smoke.rs
+12 -11
View File
@@ -26,6 +26,9 @@ use tripo3d_sdk::{ClientOptions, TripoClient, WaitOptions};
const SAMPLE_IMAGE_URL: &str = "https://www.rustacean.net/assets/rustacean-flat-happy.png";
/// 冒烟脚本单次允许的产物体积上限，与 api-server 的口径一致（512 MiB）。
const SMOKE_MAX_ARTIFACT_BYTES: u64 = 512 * 1024 * 1024;
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
let api_key = required_env("TRIPO_API_KEY")?;
@@ -178,26 +181,24 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
continue;
}
let downloaded = match client.download_model(&snapshot).await {
Ok(downloaded) => downloaded,
let artifact = match client
.download_model(&snapshot, SMOKE_MAX_ARTIFACT_BYTES)
.await
{
Ok(artifact) => artifact,
Err(error) => {
eprintln!("{name} skipped: artifact download failed: {error}");
continue;
}
};
let filename = downloaded.filename(name);
let filename = artifact.filename(name);
let mut file = File::create(&filename).await?;
let mut bytes = 0u64;
let mut downloaded = downloaded;
while let Some(chunk) = downloaded.next_chunk().await? {
bytes += chunk.len() as u64;
file.write_all(&chunk).await?;
}
file.write_all(&artifact.bytes).await?;
println!(
"{name} artifact={} bytes={} url={}",
filename,
bytes,
downloaded.url.redacted()
artifact.bytes.len(),
artifact.url.redacted()
);
}
server-rs/crates/platform-tripo/src/common/client.rs
+116 -14
View File
@@ -4,8 +4,8 @@ use shared_contracts::model3d::common::Model3dTaskStatus;
use tripo3d_sdk::TripoClient;
use super::{
TripoDownloadedArtifact, TripoError, TripoSettings, TripoTaskHandle, TripoTaskOutput,
TripoTaskSnapshot, TripoUrl, map_task, validate_task_id,
TripoArtifactBytes, TripoDownloadedArtifact, TripoError, TripoSettings, TripoTaskHandle,
TripoTaskOutput, TripoTaskSnapshot, TripoUrl, map_task, validate_task_id,
};
pub struct TripoProviderClient {
@@ -57,14 +57,16 @@ impl TripoProviderClient {
map_task(task)
}
/// 下载完整的模型产物（响应头 + 整个 body）；`max_bytes` 是单次读取的硬上限。
pub async fn download_model(
&self,
task: &TripoTaskSnapshot,
) -> Result<TripoDownloadedArtifact, TripoError> {
max_bytes: u64,
) -> Result<TripoArtifactBytes, TripoError> {
let output = completed_output(task)?;
// TODO SDK upstream: expose a streaming artifact API; then replace this
// provider-side reqwest client with the SDK stream and remove the duplicate downloader.
self.download_artifact(&task.handle.task_id, output.model_url())
self.download_artifact(&task.handle.task_id, output.model_url(), max_bytes)
.await
}
@@ -75,23 +77,36 @@ impl TripoProviderClient {
pub async fn download_rendered_image(
&self,
task: &TripoTaskSnapshot,
) -> Result<TripoDownloadedArtifact, TripoError> {
max_bytes: u64,
) -> Result<TripoArtifactBytes, TripoError> {
let output = completed_output(task)?;
self.download_artifact(&task.handle.task_id, output.rendered_image_url())
self.download_artifact(&task.handle.task_id, output.rendered_image_url(), max_bytes)
.await
}
/// 下载一个产物并读完它的 body。
///
/// 重试必须包住整个 body：`download_artifact_headers` 拿到响应头就返回，真正的
/// 几十 MB 字节在 `next_chunk` 里才传输；只重试「拿到头」这一步等于没覆盖传输阶段，
/// 一次 CDN 抖动就会毁掉一次已经扣费、provider 任务也已经跑完的生成。
async fn download_artifact(
&self,
task_id: &str,
url: &TripoUrl,
) -> Result<TripoDownloadedArtifact, TripoError> {
max_bytes: u64,
) -> Result<TripoArtifactBytes, TripoError> {
let total_attempts = self.artifact_retries.saturating_add(1);
let mut attempt = 1;
loop {
match self.download_artifact_once(task_id, url).await {
Ok(downloaded) => return Ok(downloaded),
// 每次尝试都从「取响应头」开始：body 已经读到的字节随连接一起丢掉，
// 重下的是完整产物，不做断点续传（签名地址会过期，续传也只是徒增状态）。
let result = match self.download_artifact_headers(task_id, url).await {
Ok(headers) => headers.read_all(max_bytes).await,
Err(error) => Err(error),
};
match result {
Ok(artifact) => return Ok(artifact),
Err(error) if attempt < total_attempts && error.is_retryable() => {
tokio::time::sleep(download_backoff(attempt)).await;
attempt += 1;
@@ -101,7 +116,7 @@ impl TripoProviderClient {
}
}
async fn download_artifact_once(
async fn download_artifact_headers(
&self,
task_id: &str,
url: &TripoUrl,
@@ -143,6 +158,7 @@ impl TripoProviderClient {
.map(str::to_owned);
let content_length = response.content_length();
Ok(TripoDownloadedArtifact::new(
task_id.to_string(),
url.clone(),
content_type,
content_length,
@@ -215,15 +231,22 @@ mod tests {
/// 测试用的「无数据推进」预算：够短让用例跑得快，又给 CI 抖动留出余量。
const STALL_TIMEOUT: Duration = Duration::from_millis(300);
/// 测试用产物体积上限：足够大，只有专门的超限用例才会撞上。
const TEST_MAX_ARTIFACT_BYTES: u64 = 1024 * 1024;
/// 正常推进的间隔：明显小于预算，保证用例只在「真的没数据」时才失败。
const PROGRESS_INTERVAL: Duration = Duration::from_millis(100);
fn test_client() -> TripoProviderClient {
test_client_with_retries(0)
}
/// 带产物重试次数的测试客户端：验证「整体重下」需要至少一次重试。
fn test_client_with_retries(retries: u32) -> TripoProviderClient {
let settings = TripoSettings::new(
"test-key".to_string(),
"http://127.0.0.1:1".to_string(),
STALL_TIMEOUT,
0,
retries,
"genarrative-test-tripo/1".to_string(),
)
.expect("测试配置必须合法");
@@ -314,7 +337,7 @@ mod tests {
async fn artifact_download_times_out_when_response_headers_never_arrive() {
let addr = spawn_silent_server().await;
let error = match test_client()
.download_artifact("task-1", &artifact_url(addr))
.download_artifact("task-1", &artifact_url(addr), TEST_MAX_ARTIFACT_BYTES)
.await
{
Ok(_) => panic!("响应头一直不来必须按无数据推进失败"),
@@ -327,7 +350,7 @@ mod tests {
async fn artifact_download_times_out_when_body_stops_progressing() {
let addr = spawn_stalling_body_server().await;
let mut artifact = test_client()
.download_artifact_once("task-1", &artifact_url(addr))
.download_artifact_headers("task-1", &artifact_url(addr))
.await
.expect("响应头与第一块数据必须正常返回");
@@ -343,13 +366,14 @@ mod tests {
.await
.expect_err("传输中途断流必须按无数据推进失败");
assert!(matches!(error, TripoError::Request { .. }), "{error:?}");
assert!(error.is_retryable(), "传输阶段失败必须可重试：{error}");
}
#[tokio::test]
async fn artifact_download_survives_slow_but_progressing_body() {
let addr = spawn_slow_progress_server().await;
let mut artifact = test_client()
.download_artifact_once("task-1", &artifact_url(addr))
.download_artifact_headers("task-1", &artifact_url(addr))
.await
.expect("响应头必须正常返回");
@@ -363,4 +387,82 @@ mod tests {
}
assert_eq!(received.len(), 10, "慢但持续的下载必须完整收完");
}
/// 第一次连接在 body 中途结束：下载入口必须整体重下，第二次读完完整产物。
#[tokio::test]
async fn artifact_download_retries_the_whole_body_after_a_broken_transfer() {
const COMPLETE_BODY: &[u8] = b"glTF\x02\x00\x00\x00complete-glb";
let listener = TcpListener::bind("127.0.0.1:0")
.await
.expect("mock server should bind");
let addr = listener
.local_addr()
.expect("mock server should expose addr");
tokio::spawn(async move {
// 第一次：响应头与一小块数据后结束传输，声明长度远大于实收字节。
let Ok((mut first, _)) = listener.accept().await else {
return;
};
read_request(&mut first).await;
let head = "HTTP/1.1 200 OK\r\n\
Content-Type: model/gltf-binary\r\n\
Content-Length: 64\r\n\
\r\n";
let _ = first.write_all(head.as_bytes()).await;
let _ = first.write_all(b"glTF").await;
let _ = first.flush().await;
let _ = first.shutdown().await;
// 第二次：完整产物。
let Ok((mut second, _)) = listener.accept().await else {
return;
};
read_request(&mut second).await;
let head = format!(
"HTTP/1.1 200 OK\r\nContent-Type: model/gltf-binary\r\nContent-Length: {}\r\n\r\n",
COMPLETE_BODY.len()
);
let _ = second.write_all(head.as_bytes()).await;
let _ = second.write_all(COMPLETE_BODY).await;
let _ = second.flush().await;
// 第一次连接保持打开：客户端读到的是「提前结束」而不是连接被重置。
tokio::time::sleep(Duration::from_secs(30)).await;
drop(first);
});
let artifact = test_client_with_retries(1)
.download_artifact("task-1", &artifact_url(addr), TEST_MAX_ARTIFACT_BYTES)
.await
.expect("body 中途失败必须整体重下并成功");
assert_eq!(artifact.bytes, COMPLETE_BODY);
assert_eq!(artifact.content_length, Some(COMPLETE_BODY.len() as u64));
}
/// 产物超过调用方给的体积上限：按输出违约失败，且不重试。
#[tokio::test]
async fn artifact_download_rejects_bodies_over_the_caller_limit() {
const BODY: &[u8] = b"glTF\x02\x00\x00\x00oversized";
let addr = spawn_mock_server(|mut socket| async move {
read_request(&mut socket).await;
let head = format!(
"HTTP/1.1 200 OK\r\nContent-Type: model/gltf-binary\r\nContent-Length: {}\r\n\r\n",
BODY.len()
);
let _ = socket.write_all(head.as_bytes()).await;
let _ = socket.write_all(BODY).await;
let _ = socket.flush().await;
})
.await;
let error = test_client_with_retries(2)
.download_artifact("task-1", &artifact_url(addr), 8)
.await
.expect_err("超过上限的产物必须被拒绝");
assert!(
matches!(error, TripoError::OutputSchema { .. }),
"{error:?}"
);
assert!(!error.is_retryable(), "体积超限不是重试能解决的问题");
}
}
server-rs/crates/platform-tripo/src/common/mod.rs
+2 -1
View File
@@ -12,8 +12,9 @@ pub use config::TripoSettings;
pub use error::{TripoError, TripoField, TripoValidationReason};
pub(crate) use extra::extra_fields;
pub(crate) use mapping::map_task;
pub(crate) use types::TripoDownloadedArtifact;
pub use types::{
TripoDownloadedArtifact, TripoTaskFailure, TripoTaskHandle, TripoTaskOutput, TripoTaskSnapshot,
TripoArtifactBytes, TripoTaskFailure, TripoTaskHandle, TripoTaskOutput, TripoTaskSnapshot,
TripoTaskType, TripoUrl,
};
pub(crate) use validation::{
server-rs/crates/platform-tripo/src/common/types.rs
+103 -28
View File
@@ -122,25 +122,49 @@ pub struct TripoTaskSnapshot {
pub completed_at: Option<String>,
}
/// provider 产物的下载句柄。模型与预览图共用同一条流式读取路径，
/// provider 产物的下载句柄（响应头阶段）。模型与预览图共用同一条流式读取路径，
/// 因此类型名按“产物”而不是“模型”命名。
pub struct TripoDownloadedArtifact {
pub url: TripoUrl,
pub content_type: Option<String>,
pub content_length: Option<u64>,
///
/// 只在本 crate 内使用：对外只暴露读完的 [`TripoArtifactBytes`]，流式消费等
/// 「产物直送 OSS」落地时再决定是否公开。
///
/// 拿到的只是响应头：body 由 [`TripoDownloadedArtifact::next_chunk`] 逐块读出。
/// [`crate::TripoProviderClient`] 的下载入口会把「响应头 + 读完 body」作为一个整体
/// 重试单元（见 [`TripoDownloadedArtifact::read_all`]），所以这里的失败不等于调用方失败。
pub(crate) struct TripoDownloadedArtifact {
task_id: String,
pub(crate) url: TripoUrl,
pub(crate) content_type: Option<String>,
pub(crate) content_length: Option<u64>,
response: Response,
status: u16,
received: u64,
}
/// 已完整读出的 provider 产物：响应头元数据 + 全部字节。
///
/// 与 [`TripoDownloadedArtifact`] 的区别是「读完了」：body 中途失败会被下载入口整体重下，
/// 因此拿到这个值就代表这次读取是完整的。（几十 MB 的产物当前一次性读进内存，api-server
/// 侧同样如此；未来接 OSS 流式 / 分片上传时再回到流式句柄。）
#[derive(Debug)]
pub struct TripoArtifactBytes {
pub url: TripoUrl,
pub content_type: Option<String>,
/// provider 声明的长度，仅用于读取时的完整性校验与内存预分配，不是落库口径。
pub content_length: Option<u64>,
pub bytes: Vec<u8>,
}
impl TripoDownloadedArtifact {
pub(crate) fn new(
task_id: String,
url: TripoUrl,
content_type: Option<String>,
content_length: Option<u64>,
response: Response,
) -> Self {
Self {
task_id,
url,
content_type,
content_length,
@@ -150,15 +174,14 @@ impl TripoDownloadedArtifact {
}
}
pub async fn next_chunk(&mut self) -> Result<Option<Bytes>, TripoError> {
pub(crate) async fn next_chunk(&mut self) -> Result<Option<Bytes>, TripoError> {
let chunk = self
.response
.chunk()
.await
.map_err(|error| TripoError::Request {
message: format!("failed to read artifact response body: {error}"),
status: Some(self.status),
})?;
// 传输错误不是 HTTP 状态：`status` 描述的是响应头，写进这里会让
// `is_retryable` 把 200 当成「不值得重试」，CDN 抖动就再也重试不了。
.map_err(|error| body_read_error(&self.task_id, self.status, format!("{error}")))?;
match chunk {
Some(chunk) => {
self.received += chunk.len() as u64;
@@ -167,13 +190,15 @@ impl TripoDownloadedArtifact {
None => {
if let Some(expected) = self.content_length {
if self.received != expected {
return Err(TripoError::Request {
message: format!(
// body 提前结束同样按传输失败处理：整段重下才有机会拿到完整字节。
return Err(body_read_error(
&self.task_id,
self.status,
format!(
"artifact length mismatch: expected {expected} bytes, received {}",
self.received
),
status: Some(self.status),
});
));
}
}
Ok(None)
@@ -181,27 +206,77 @@ impl TripoDownloadedArtifact {
}
}
/// 读完整个响应体。单次读取：失败不在这里重试（只有一份连接），
/// 由下载入口用新连接整体重下。
///
/// `max_bytes` 是单次读取的硬上限：超过它说明响应异常或产物口径变了，
/// 宁可失败退款也不要把调用方内存打满。
pub(crate) async fn read_all(
mut self,
max_bytes: u64,
) -> Result<TripoArtifactBytes, TripoError> {
let mut bytes =
Vec::with_capacity(self.content_length.unwrap_or(0).min(max_bytes) as usize);
while let Some(chunk) = self.next_chunk().await? {
let next_len = bytes.len() as u64 + chunk.len() as u64;
if next_len > max_bytes {
return Err(TripoError::OutputSchema {
task_id: self.task_id.clone(),
message: format!(
"artifact exceeds the {max_bytes} byte limit implied by the caller: {} bytes and still streaming",
next_len
),
});
}
bytes.extend_from_slice(chunk.as_ref());
}
Ok(TripoArtifactBytes {
url: self.url,
content_type: self.content_type,
content_length: self.content_length,
bytes,
})
}
}
impl TripoArtifactBytes {
/// 产物落盘用的文件名。远端地址不可信，扩展名只接受短的字母数字 token：
/// `Url::path_segments` 已做百分号解码，`%2F` / `%5C` 这类编码分隔符会直接
/// 进到扩展名里，不过滤就会拼出跨目录的路径。
pub fn filename(&self, name: &str) -> String {
let extension = self
.url
.0
.path_segments()
.and_then(|segments| segments.last())
.and_then(|segment| segment.rsplit_once('.'))
.map(|(_, extension)| extension)
.filter(|extension| {
!extension.is_empty()
&& extension.len() <= MAX_ARTIFACT_FILE_EXTENSION_LEN
&& extension.bytes().all(|byte| byte.is_ascii_alphanumeric())
})
.unwrap_or_else(|| fallback_artifact_extension(self.content_type.as_deref()));
format!("{}.{extension}", sanitize_artifact_file_stem(name))
artifact_filename(name, self.content_type.as_deref(), &self.url)
}
}
/// 响应体读取失败：统一成可重试的传输错误，并把 HTTP 状态留在文案里供排障。
fn body_read_error(task_id: &str, status: u16, detail: String) -> TripoError {
TripoError::Request {
message: format!(
"failed to read artifact response body for task {task_id} (HTTP {status}): {detail}"
),
status: None,
}
}
/// 产物落盘用的文件名。远端地址不可信，扩展名只接受短的字母数字 token：
/// `Url::path_segments` 已做百分号解码，`%2F` / `%5C` 这类编码分隔符会直接
/// 进到扩展名里，不过滤就会拼出跨目录的路径。
fn artifact_filename(name: &str, content_type: Option<&str>, url: &TripoUrl) -> String {
let extension = url
.0
.path_segments()
.and_then(|segments| segments.last())
.and_then(|segment| segment.rsplit_once('.'))
.map(|(_, extension)| extension)
.filter(|extension| {
!extension.is_empty()
&& extension.len() <= MAX_ARTIFACT_FILE_EXTENSION_LEN
&& extension.bytes().all(|byte| byte.is_ascii_alphanumeric())
})
.unwrap_or_else(|| fallback_artifact_extension(content_type));
format!("{}.{extension}", sanitize_artifact_file_stem(name))
}
/// URL 路径给不出可用扩展名时按 content type 兜底：同一条下载路径也服务预览图，
/// 一律写 `.glb` 会把 PNG / JPEG 产物写成模型文件名。
fn fallback_artifact_extension(content_type: Option<&str>) -> &'static str {
server-rs/crates/platform-tripo/src/lib.rs
+1 -1
View File
@@ -9,7 +9,7 @@ pub mod multiview_to_model;
pub mod text_to_model;
pub use common::{
TripoDownloadedArtifact, TripoError, TripoField, TripoProviderClient, TripoSettings,
TripoArtifactBytes, TripoError, TripoField, TripoProviderClient, TripoSettings,
TripoTaskFailure, TripoTaskHandle, TripoTaskOutput, TripoTaskSnapshot, TripoTaskType, TripoUrl,
TripoValidationReason,
};