rc_core/
multipart_copy.rs

1//! Pure planning and request types for multipart server-side copies.
2
3use std::collections::HashMap;
4use std::sync::Arc;
5use std::sync::atomic::{AtomicBool, Ordering};
6
7use crate::{Error, ObjectInfo, Result};
8use tokio::sync::Notify;
9
10/// Largest object that can be copied with a single S3 CopyObject request.
11pub const S3_SINGLE_COPY_MAX_SIZE: u64 = 5 * 1024 * 1024 * 1024;
12
13/// Smallest non-final part accepted by S3 multipart uploads.
14pub const S3_MULTIPART_COPY_MIN_PART_SIZE: u64 = 5 * 1024 * 1024;
15
16/// Largest part accepted by S3 multipart uploads.
17pub const S3_MULTIPART_COPY_MAX_PART_SIZE: u64 = 5 * 1024 * 1024 * 1024;
18
19/// Largest number of parts accepted by S3 multipart uploads.
20pub const S3_MULTIPART_COPY_MAX_PARTS: usize = 10_000;
21
22/// Largest object accepted by S3.
23pub const S3_MAX_OBJECT_SIZE: u64 = 5 * 1024 * 1024 * 1024 * 1024;
24
25/// Default multipart copy part size.
26pub const DEFAULT_MULTIPART_COPY_PART_SIZE: u64 = 64 * 1024 * 1024;
27
28/// Cooperative cancellation control for multipart server-side copies.
29///
30/// Callers must signal this control rather than dropping the copy future when
31/// they need the backend to await multipart upload cleanup.
32#[derive(Debug, Clone, Default)]
33pub struct MultipartCopyCancellation {
34    inner: Arc<MultipartCopyCancellationInner>,
35}
36
37#[derive(Debug, Default)]
38struct MultipartCopyCancellationInner {
39    cancelled: AtomicBool,
40    notify: Notify,
41}
42
43impl MultipartCopyCancellation {
44    /// Create a cancellation control in the active state.
45    pub fn new() -> Self {
46        Self::default()
47    }
48
49    /// Request cooperative cancellation.
50    pub fn cancel(&self) {
51        if !self.inner.cancelled.swap(true, Ordering::AcqRel) {
52            self.inner.notify.notify_waiters();
53        }
54    }
55
56    /// Return whether cancellation has been requested.
57    pub fn is_cancelled(&self) -> bool {
58        self.inner.cancelled.load(Ordering::Acquire)
59    }
60
61    /// Wait until cancellation is requested without losing an early signal.
62    pub async fn cancelled(&self) {
63        let notified = self.inner.notify.notified();
64        tokio::pin!(notified);
65        // Register before reading the flag so cancellation cannot occur in the
66        // gap between the state check and awaiting the notification.
67        notified.as_mut().enable();
68        if self.is_cancelled() {
69            return;
70        }
71        notified.await;
72    }
73}
74
75/// A byte range assigned to one UploadPartCopy request.
76#[derive(Debug, Clone, PartialEq, Eq)]
77pub struct MultipartCopyPart {
78    /// One-based S3 part number.
79    pub part_number: i32,
80    /// Inclusive first byte offset.
81    pub start: u64,
82    /// Inclusive last byte offset.
83    pub end_inclusive: u64,
84    /// Number of logical bytes in the part.
85    pub size: u64,
86}
87
88/// A validated multipart server-side copy plan.
89#[derive(Debug, Clone, PartialEq, Eq)]
90pub struct MultipartCopyPlan {
91    /// Total logical source size.
92    pub object_size: u64,
93    /// Planned size of every non-final part.
94    pub part_size: u64,
95    /// Ordered, gap-free byte ranges.
96    pub parts: Vec<MultipartCopyPart>,
97}
98
99impl MultipartCopyPlan {
100    /// Build a plan that satisfies S3 part size and count limits.
101    pub fn new(object_size: u64, preferred_part_size: Option<u64>) -> Result<Self> {
102        if object_size == 0 {
103            return Err(Error::InvalidPath(
104                "Multipart copy source size must be greater than zero".to_string(),
105            ));
106        }
107        if object_size > S3_MAX_OBJECT_SIZE {
108            return Err(Error::InvalidPath(format!(
109                "Multipart copy source size exceeds the S3 object limit of {S3_MAX_OBJECT_SIZE} bytes"
110            )));
111        }
112
113        let preferred_part_size = preferred_part_size.unwrap_or(DEFAULT_MULTIPART_COPY_PART_SIZE);
114        if !(S3_MULTIPART_COPY_MIN_PART_SIZE..=S3_MULTIPART_COPY_MAX_PART_SIZE)
115            .contains(&preferred_part_size)
116        {
117            return Err(Error::InvalidPath(format!(
118                "Multipart copy part size must be between {S3_MULTIPART_COPY_MIN_PART_SIZE} and {S3_MULTIPART_COPY_MAX_PART_SIZE} bytes"
119            )));
120        }
121
122        // Raising a caller's preferred size is necessary when its requested size
123        // would exceed S3's 10,000-part ceiling.
124        let minimum_size_for_part_limit = object_size.div_ceil(S3_MULTIPART_COPY_MAX_PARTS as u64);
125        let part_size = preferred_part_size
126            .max(minimum_size_for_part_limit)
127            .max(S3_MULTIPART_COPY_MIN_PART_SIZE);
128        if part_size > S3_MULTIPART_COPY_MAX_PART_SIZE {
129            return Err(Error::InvalidPath(format!(
130                "Multipart copy requires a part larger than the S3 limit of {S3_MULTIPART_COPY_MAX_PART_SIZE} bytes"
131            )));
132        }
133
134        let part_count = object_size.div_ceil(part_size);
135        if part_count > S3_MULTIPART_COPY_MAX_PARTS as u64 {
136            return Err(Error::InvalidPath(format!(
137                "Multipart copy requires {part_count} parts, exceeding the S3 limit of {S3_MULTIPART_COPY_MAX_PARTS}"
138            )));
139        }
140
141        let mut parts = Vec::with_capacity(part_count as usize);
142        let mut start = 0;
143        for part_index in 0..part_count {
144            let end_exclusive = (start + part_size).min(object_size);
145            parts.push(MultipartCopyPart {
146                part_number: i32::try_from(part_index + 1)
147                    .expect("S3 multipart copy plans contain at most 10,000 parts"),
148                start,
149                end_inclusive: end_exclusive - 1,
150                size: end_exclusive - start,
151            });
152            start = end_exclusive;
153        }
154
155        Ok(Self {
156            object_size,
157            part_size,
158            parts,
159        })
160    }
161}
162
163/// Return whether S3 requires multipart server-side copy for this object size.
164pub const fn requires_multipart_copy(object_size: u64) -> bool {
165    object_size > S3_SINGLE_COPY_MAX_SIZE
166}
167
168/// Inputs required for a consistent multipart server-side copy.
169#[derive(Debug, Clone, PartialEq, Eq)]
170pub struct MultipartCopyOptions {
171    /// Logical source size used to build part ranges.
172    pub source_size: u64,
173    /// ETag sent as `x-amz-copy-source-if-match` on every part.
174    pub source_etag: String,
175    /// Exact historical source version, when selected.
176    pub source_version_id: Option<String>,
177    /// Optional preferred part size.
178    pub preferred_part_size: Option<u64>,
179    /// Destination content type inherited from the source.
180    pub content_type: Option<String>,
181    /// Destination user metadata inherited from the source.
182    pub metadata: HashMap<String, String>,
183}
184
185impl MultipartCopyOptions {
186    /// Build options with source identity and safe defaults.
187    pub fn new(source_size: u64, source_etag: impl Into<String>) -> Result<Self> {
188        let source_etag = source_etag.into();
189        if source_etag.is_empty() {
190            return Err(Error::InvalidPath(
191                "Multipart copy source ETag cannot be empty".to_string(),
192            ));
193        }
194        Ok(Self {
195            source_size,
196            source_etag,
197            source_version_id: None,
198            preferred_part_size: None,
199            content_type: None,
200            metadata: HashMap::new(),
201        })
202    }
203
204    /// Validate the request and return its pure copy plan.
205    pub fn plan(&self) -> Result<MultipartCopyPlan> {
206        if self.source_etag.is_empty() {
207            return Err(Error::InvalidPath(
208                "Multipart copy source ETag cannot be empty".to_string(),
209            ));
210        }
211        if self.source_version_id.as_deref().is_some_and(str::is_empty) {
212            return Err(Error::InvalidPath(
213                "Source version ID cannot be empty".to_string(),
214            ));
215        }
216        MultipartCopyPlan::new(self.source_size, self.preferred_part_size)
217    }
218}
219
220/// Successful multipart copy details.
221#[derive(Debug, Clone)]
222pub struct MultipartCopyResult {
223    /// Destination object metadata.
224    pub object: ObjectInfo,
225    /// Service upload identifier used for the completed operation.
226    pub upload_id: String,
227    /// Number of successfully completed parts.
228    pub part_count: usize,
229    /// Logical bytes copied.
230    pub bytes_copied: u64,
231}
232
233/// Callback invoked with cumulative successfully copied logical bytes.
234pub type MultipartCopyProgress<'a> = dyn Fn(u64) + Send + Sync + 'a;
235
236#[cfg(test)]
237mod tests {
238    use super::*;
239
240    #[test]
241    fn copy_threshold_is_strictly_greater_than_five_gibibytes() {
242        assert!(!requires_multipart_copy(S3_SINGLE_COPY_MAX_SIZE));
243        assert!(requires_multipart_copy(S3_SINGLE_COPY_MAX_SIZE + 1));
244    }
245
246    #[tokio::test]
247    async fn cancellation_control_remembers_early_and_late_signals() {
248        let early = MultipartCopyCancellation::new();
249        early.cancel();
250        early.cancel();
251        early.cancelled().await;
252        assert!(early.is_cancelled());
253
254        let late = MultipartCopyCancellation::new();
255        let waiter = late.clone();
256        let task = tokio::spawn(async move {
257            waiter.cancelled().await;
258        });
259        late.cancel();
260        task.await.expect("cancellation waiter");
261    }
262
263    #[tokio::test]
264    async fn cancellation_control_handles_registration_races() {
265        for _ in 0..1_000 {
266            let cancellation = MultipartCopyCancellation::new();
267            let waiter = cancellation.clone();
268            let task = tokio::spawn(async move {
269                waiter.cancelled().await;
270            });
271            tokio::task::yield_now().await;
272            cancellation.cancel();
273            tokio::time::timeout(std::time::Duration::from_secs(1), task)
274                .await
275                .expect("cancellation waiter must not lose notification")
276                .expect("cancellation waiter task");
277        }
278    }
279
280    #[test]
281    fn plan_uses_inclusive_gap_free_ranges() {
282        let object_size = S3_MULTIPART_COPY_MIN_PART_SIZE * 2 + 1;
283        let plan = MultipartCopyPlan::new(object_size, Some(S3_MULTIPART_COPY_MIN_PART_SIZE))
284            .expect("build multipart copy plan");
285
286        assert_eq!(plan.parts.len(), 3);
287        assert_eq!(
288            plan.parts,
289            vec![
290                MultipartCopyPart {
291                    part_number: 1,
292                    start: 0,
293                    end_inclusive: S3_MULTIPART_COPY_MIN_PART_SIZE - 1,
294                    size: S3_MULTIPART_COPY_MIN_PART_SIZE,
295                },
296                MultipartCopyPart {
297                    part_number: 2,
298                    start: S3_MULTIPART_COPY_MIN_PART_SIZE,
299                    end_inclusive: S3_MULTIPART_COPY_MIN_PART_SIZE * 2 - 1,
300                    size: S3_MULTIPART_COPY_MIN_PART_SIZE,
301                },
302                MultipartCopyPart {
303                    part_number: 3,
304                    start: S3_MULTIPART_COPY_MIN_PART_SIZE * 2,
305                    end_inclusive: object_size - 1,
306                    size: 1,
307                },
308            ]
309        );
310        assert_eq!(
311            plan.parts.iter().map(|part| part.size).sum::<u64>(),
312            object_size
313        );
314    }
315
316    #[test]
317    fn plan_increases_part_size_to_stay_within_ten_thousand_parts() {
318        let object_size =
319            S3_MULTIPART_COPY_MIN_PART_SIZE * (S3_MULTIPART_COPY_MAX_PARTS as u64 + 1);
320        let plan = MultipartCopyPlan::new(object_size, Some(S3_MULTIPART_COPY_MIN_PART_SIZE))
321            .expect("adjust undersized preferred part");
322
323        assert_eq!(plan.parts.len(), S3_MULTIPART_COPY_MAX_PARTS);
324        assert_eq!(plan.part_size, object_size.div_ceil(10_000));
325        assert_eq!(
326            plan.parts.last().map(|part| part.end_inclusive),
327            Some(object_size - 1)
328        );
329    }
330
331    #[test]
332    fn plan_accepts_exact_s3_boundaries() {
333        let exact_max_parts_size =
334            S3_MULTIPART_COPY_MIN_PART_SIZE * S3_MULTIPART_COPY_MAX_PARTS as u64;
335        let plan =
336            MultipartCopyPlan::new(exact_max_parts_size, Some(S3_MULTIPART_COPY_MIN_PART_SIZE))
337                .expect("accept exactly ten thousand parts");
338        assert_eq!(plan.parts.len(), S3_MULTIPART_COPY_MAX_PARTS);
339
340        let max_object = MultipartCopyPlan::new(S3_MAX_OBJECT_SIZE, None)
341            .expect("accept maximum S3 object size");
342        assert!(max_object.parts.len() <= S3_MULTIPART_COPY_MAX_PARTS);
343        assert!(max_object.part_size <= S3_MULTIPART_COPY_MAX_PART_SIZE);
344    }
345
346    #[test]
347    fn plan_rejects_zero_oversized_objects_and_invalid_part_sizes() {
348        for (object_size, part_size) in [
349            (0, None),
350            (S3_MAX_OBJECT_SIZE + 1, None),
351            (1, Some(S3_MULTIPART_COPY_MIN_PART_SIZE - 1)),
352            (1, Some(S3_MULTIPART_COPY_MAX_PART_SIZE + 1)),
353        ] {
354            assert!(
355                MultipartCopyPlan::new(object_size, part_size).is_err(),
356                "object_size={object_size}, part_size={part_size:?}"
357            );
358        }
359    }
360
361    #[test]
362    fn options_reject_empty_source_identity() {
363        assert!(MultipartCopyOptions::new(1, "").is_err());
364
365        let mut options =
366            MultipartCopyOptions::new(1, "etag").expect("valid multipart copy options");
367        options.source_etag.clear();
368        assert!(options.plan().is_err());
369
370        options.source_etag = "etag".to_string();
371        options.source_version_id = Some(String::new());
372        assert!(options.plan().is_err());
373    }
374}
rc_core/multipart_copy.rs

rc_core/
multipart_copy.rs
