rusty_cat/
http_breakpoint.rs

1//! Breakpoint upload/download protocol plugins.
2//!
3//! - The executor handles scheduling, chunk I/O, retries, progress, and state.
4//! - Protocol plugins handle business-specific request/response semantics.
5
6use crate::error::{InnerErrorCode, MeowError};
7use crate::transfer_task::TransferTask;
8use async_trait::async_trait;
9
10pub use crate::download_trait::{BreakpointDownload, DownloadHeadCtx, DownloadRangeGetCtx};
11pub use crate::upload_trait::{BreakpointUpload, UploadChunkCtx, UploadPrepareCtx};
12use reqwest::header::{HeaderMap, HeaderName, HeaderValue};
13use reqwest::multipart;
14use reqwest::Method;
15
16#[derive(Debug, Clone, Default)]
17pub struct UploadResumeInfo {
18    /// File ID returned by server when upload is already complete.
19    pub completed_file_id: Option<String>,
20    /// Suggested next byte offset (`nextByte`) from server.
21    ///
22    /// Range: `>= 0`.
23    pub next_byte: Option<u64>,
24    /// Opaque, provider-defined identifier of the in-flight multipart/resumable
25    /// upload session (for example an Aliyun OSS multipart `UploadId`).
26    ///
27    /// This is **not** a credential and is safe to persist. Persisting it lets
28    /// callers abort an orphaned session out-of-band after a crash, so that
29    /// uncommitted parts/blocks stop accruing storage cost. Providers that have
30    /// no separate session id (such as Azure Block Blob, whose resume state is
31    /// the uncommitted block list keyed by the blob URL) leave this `None`.
32    ///
33    /// Note: the bundled executor only consumes [`UploadResumeInfo::completed_file_id`]
34    /// and [`UploadResumeInfo::next_byte`]; it does not forward this id to the
35    /// progress pipeline. To persist it, hold the protocol instance and read its
36    /// accessor (for OSS, [`crate::aliyun_oss_direct::AliOssDirectUpload::current_upload_id`]),
37    /// or consume this value from a custom executor.
38    pub provider_upload_id: Option<String>,
39}
40
41/// Upload HTTP request body payload.
42#[derive(Debug)]
43pub enum UploadBody {
44    Multipart(multipart::Form),
45    Binary(Vec<u8>),
46}
47
48/// Upload request description returned by protocol plugin.
49#[derive(Debug)]
50pub struct UploadRequest {
51    /// HTTP method for upload call.
52    pub method: Method,
53    /// Target request URL.
54    pub url: String,
55    /// Request headers.
56    pub headers: HeaderMap,
57    /// Body payload.
58    pub body: UploadBody,
59}
60
61impl UploadRequest {
62    /// Creates an upload request using URL/method/headers from task.
63    ///
64    /// # Examples
65    ///
66    /// ```no_run
67    /// use rusty_cat::api::{TransferTask, UploadBody, UploadRequest};
68    ///
69    /// fn make_request(task: &TransferTask, bytes: Vec<u8>) {
70    ///     let req = UploadRequest::from_task(task, UploadBody::Binary(bytes));
71    ///     let _ = req;
72    /// }
73    /// ```
74    pub fn from_task(task: &TransferTask, body: UploadBody) -> Self {
75        Self {
76            method: task.method(),
77            url: task.url().to_string(),
78            headers: task.headers().clone(),
79            body,
80        }
81    }
82}
83
84/// Parses default upload response JSON payload into [`UploadResumeInfo`].
85fn parse_default_upload_response(body: &str) -> Result<UploadResumeInfo, MeowError> {
86    if body.trim().is_empty() {
87        crate::meow_trace_log!(
88            "upload_protocol",
89            "empty upload response body, fallback default"
90        );
91        return Ok(UploadResumeInfo::default());
92    }
93    let v: DefaultUploadResp = serde_json::from_str(body).map_err(|e| {
94        crate::meow_error_log!(
95            "upload_protocol",
96            "upload response parse failed: body_len={} err={}",
97            body.len(),
98            e
99        );
100        MeowError::from_code(
101            InnerErrorCode::ResponseParseError,
102            format!("upload response json: {e}, body: {body}"),
103        )
104    })?;
105    crate::meow_trace_log!(
106        "upload_protocol",
107        "upload response parsed: file_id_present={} next_byte={:?}",
108        v.file_id.is_some(),
109        v.next_byte
110    );
111    Ok(UploadResumeInfo {
112        completed_file_id: v.file_id,
113        next_byte: v.next_byte.map(|n| if n < 0 { 0u64 } else { n as u64 }),
114        // The default multipart protocol carries no provider-side session id.
115        provider_upload_id: None,
116    })
117}
118
119/// Sends upload request and returns response body as string.
120async fn send_upload_request(
121    client: &reqwest::Client,
122    req: UploadRequest,
123) -> Result<String, MeowError> {
124    // Capture the URL for logging before it is moved into the request builder.
125    // Logging-only binding; does not change request behavior.
126    let log_url = req.url.clone();
127    let mut builder = client.request(req.method, req.url).headers(req.headers);
128    builder = match req.body {
129        UploadBody::Multipart(form) => builder.multipart(form),
130        UploadBody::Binary(bytes) => builder.body(bytes),
131    };
132    let resp = builder.send().await.map_err(|e| {
133        crate::log::emit_lazy(|| {
134            crate::log::Log::error(
135                "upload_send",
136                format!(
137                    "upload request send() transport error: {}",
138                    crate::log::redact_secrets(&e.to_string())
139                ),
140            )
141            .with_url(log_url.as_str())
142        });
143        map_reqwest(e)
144    })?;
145    let status = resp.status();
146    let body = resp.text().await.map_err(|e| {
147        crate::log::emit_lazy(|| {
148            crate::log::Log::error(
149                "upload_send",
150                format!(
151                    "upload response body read error: {}",
152                    crate::log::redact_secrets(&e.to_string())
153                ),
154            )
155            .with_http_status(status.as_u16())
156            .with_url(log_url.as_str())
157        });
158        map_reqwest(e)
159    })?;
160    if !status.is_success() {
161        crate::log::emit_lazy(|| {
162            crate::log::Log::error(
163                "upload_send",
164                format!(
165                    "upload HTTP non-2xx status={} body={}",
166                    status,
167                    crate::log::redact_secrets(&body)
168                ),
169            )
170            .with_http_status(status.as_u16())
171            .with_url(log_url.as_str())
172        });
173        return Err(MeowError::from_code(
174            InnerErrorCode::ResponseStatusError,
175            format!("upload HTTP {status}: {body}"),
176        )
177        .with_http_status(status.as_u16()));
178    }
179    Ok(body)
180}
181
182/// Maps `reqwest` errors into SDK errors.
183fn map_reqwest(e: reqwest::Error) -> MeowError {
184    MeowError::from_source(InnerErrorCode::HttpError, e.to_string(), e)
185}
186
187#[derive(Debug, Clone)]
188pub struct DefaultStyleUpload {
189    /// Optional business category sent in multipart form.
190    pub category: String,
191}
192
193impl Default for DefaultStyleUpload {
194    fn default() -> Self {
195        Self {
196            category: String::new(),
197        }
198    }
199}
200
201/// Multipart field key: file md5/signature.
202const KEY_FILE_MD5: &str = "fileMd5";
203/// Multipart field key: file name.
204const KEY_FILE_NAME: &str = "fileName";
205/// Multipart field key: business category.
206const KEY_CATEGORY: &str = "category";
207/// Multipart field key: total file size.
208const KEY_TOTAL_SIZE: &str = "totalSize";
209/// Multipart field key: current chunk offset.
210const KEY_OFFSET: &str = "offset";
211/// Multipart field key: current chunk byte length.
212const KEY_PART_SIZE: &str = "partSize";
213/// Multipart field key: file binary part.
214const KEY_FILE: &str = "file";
215/// Multipart part file name used for chunk payload.
216const KEY_UPLOAD_CHUNK_DATA: &str = "upload_chunk_data";
217
218#[derive(serde::Deserialize)]
219struct DefaultUploadResp {
220    #[serde(rename = "fileId")]
221    file_id: Option<String>,
222    #[serde(rename = "nextByte")]
223    next_byte: Option<i64>,
224}
225
226#[async_trait]
227impl BreakpointUpload for DefaultStyleUpload {
228    /// Sends prepare request for default multipart upload protocol.
229    async fn prepare(&self, ctx: UploadPrepareCtx<'_>) -> Result<UploadResumeInfo, MeowError> {
230        let form = multipart::Form::new()
231            .text(KEY_FILE_MD5, ctx.task.file_sign().to_string())
232            .text(KEY_FILE_NAME, ctx.task.file_name().to_string())
233            .text(KEY_CATEGORY, self.category.clone())
234            .text(KEY_TOTAL_SIZE, ctx.task.total_size().to_string());
235        let req = UploadRequest::from_task(ctx.task, UploadBody::Multipart(form));
236        let body = send_upload_request(ctx.client, req).await?;
237        parse_default_upload_response(&body)
238    }
239
240    /// Sends one upload chunk for default multipart upload protocol.
241    ///
242    /// The chunk payload is forwarded to `reqwest` as a stream built from the
243    /// shared [`bytes::Bytes`] handle, so no per-chunk `Vec::to_vec` copy is
244    /// required. Retries clone the same `Bytes` (refcount only) instead of
245    /// re-allocating the chunk buffer.
246    async fn upload_chunk(&self, ctx: UploadChunkCtx<'_>) -> Result<UploadResumeInfo, MeowError> {
247        let chunk_len = ctx.chunk.len();
248        let body = reqwest::Body::from(ctx.chunk.clone());
249        let part = multipart::Part::stream_with_length(body, chunk_len as u64)
250            .file_name(KEY_UPLOAD_CHUNK_DATA)
251            .mime_str("application/octet-stream")
252            .map_err(|e| {
253                let err_text = e.to_string();
254                let offset = ctx.offset;
255                let byte_len = chunk_len as u64;
256                let url = ctx.task.url().to_string();
257                crate::log::emit_lazy(move || {
258                    crate::log::Log::error(
259                        "upload_chunk",
260                        format!(
261                            "multipart chunk Part build failed: {}",
262                            crate::log::redact_secrets(&err_text)
263                        ),
264                    )
265                    .with_offset(offset)
266                    .with_byte_len(byte_len)
267                    .with_url(url.as_str())
268                });
269                MeowError::from_code(InnerErrorCode::HttpError, e.to_string())
270            })?;
271
272        let form = multipart::Form::new()
273            .part(KEY_FILE, part)
274            .text(KEY_FILE_MD5, ctx.task.file_sign().to_string())
275            .text(KEY_FILE_NAME, ctx.task.file_name().to_string())
276            .text(KEY_CATEGORY, self.category.clone())
277            .text(KEY_OFFSET, ctx.offset.to_string())
278            .text(KEY_PART_SIZE, chunk_len.to_string())
279            .text(KEY_TOTAL_SIZE, ctx.task.total_size().to_string());
280        let req = UploadRequest::from_task(ctx.task, UploadBody::Multipart(form));
281        let body = send_upload_request(ctx.client, req).await?;
282        parse_default_upload_response(&body)
283    }
284}
285
286/// HTTP behavior config for breakpoint range download.
287///
288/// This is usually provided by caller in task config. If missing, enqueue logic
289/// fills it from [`crate::meow_config::MeowConfig`].
290#[derive(Debug, Clone, PartialEq, Eq)]
291pub struct BreakpointDownloadHttpConfig {
292    /// `Accept` header used by range GET chunk requests.
293    ///
294    /// Typical value: `application/octet-stream`.
295    pub range_accept: String,
296}
297
298impl Default for BreakpointDownloadHttpConfig {
299    fn default() -> Self {
300        Self {
301            range_accept: DEFAULT_RANGE_ACCEPT.to_string(),
302        }
303    }
304}
305
306pub(crate) const DEFAULT_RANGE_ACCEPT: &str = "application/octet-stream";
307
308/// Inserts header pair into map when both name/value are valid.
309pub(crate) fn insert_header(map: &mut HeaderMap, name: &str, value: &str) {
310    if let (Ok(n), Ok(v)) = (
311        HeaderName::from_bytes(name.as_bytes()),
312        HeaderValue::from_str(value),
313    ) {
314        map.insert(n, v);
315    }
316}
317
318/// Default range download protocol.
319///
320/// It sets `Range` and `Accept` headers and reads total size from
321/// `Content-Length`.
322#[derive(Debug, Clone, Default)]
323pub struct StandardRangeDownload;
324
325impl BreakpointDownload for StandardRangeDownload {
326    fn supports_parallel_parts(&self) -> bool {
327        true
328    }
329}
330
331#[cfg(test)]
332mod tests {
333    use super::{BreakpointUpload, DefaultStyleUpload, StandardRangeDownload};
334
335    #[test]
336    fn default_style_upload_is_serial_only() {
337        // The default multipart protocol trusts the server's single-cursor
338        // `nextByte`, so it must NOT advertise out-of-order safety.
339        assert!(!DefaultStyleUpload::default().supports_parallel_parts());
340    }
341
342    #[test]
343    fn standard_range_download_supports_parallel_parts() {
344        use crate::download_trait::BreakpointDownload;
345        assert!(StandardRangeDownload.supports_parallel_parts());
346    }
347}
rusty_cat/http_breakpoint.rs

rusty_cat/
http_breakpoint.rs
