krabka_object_store/
multipart.rs

1//! S3's incomplete-multipart listing, which `object_store` 0.13 does not expose.
2
3use {
4    crate::{
5        ObjectStoreError, S3Config,
6        build::build_s3_store,
7        control_plane::{s3_bucket_url, s3_http_client, signed_s3_get_xml},
8    },
9    serde::Deserialize,
10};
11
12/// One upload that S3 has initiated but not completed or aborted.
13#[derive(Debug, Clone, PartialEq, Eq)]
14pub struct IncompleteMultipartUpload {
15    /// Object key the upload would create.
16    pub key: String,
17    /// Backend upload identifier.
18    pub upload_id: String,
19}
20
21#[derive(Debug, Deserialize)]
22#[serde(rename_all = "PascalCase")]
23struct ListResponse {
24    #[serde(default, rename = "Upload")]
25    uploads: Vec<IncompleteUpload>,
26    #[serde(default)]
27    is_truncated: bool,
28    next_key_marker: Option<String>,
29    next_upload_id_marker: Option<String>,
30}
31
32#[derive(Debug, Deserialize)]
33#[serde(rename_all = "PascalCase")]
34struct IncompleteUpload {
35    key: String,
36    upload_id: String,
37}
38
39/// Lists every incomplete S3 multipart upload under `prefix`.
40///
41/// `object_store` exposes multipart creation and abort but not S3's
42/// `ListMultipartUploads` operation, so the verifier makes this one signed S3
43/// request directly with the same credential provider.
44///
45/// # Errors
46///
47/// Returns an error when the S3 configuration, credentials, request, or XML
48/// response is invalid.
49pub async fn list_s3_multipart_uploads(
50    cfg: &S3Config,
51    prefix: Option<&str>,
52) -> Result<Vec<IncompleteMultipartUpload>, ObjectStoreError> {
53    let store = build_s3_store(cfg)?;
54    let credential = store.credentials().get_credential().await?;
55    let client = s3_http_client(cfg)?;
56    let bucket_url = s3_bucket_url(cfg)?;
57    let mut key_marker = None;
58    let mut upload_id_marker = None;
59    let mut found = Vec::new();
60
61    loop {
62        let mut url = bucket_url.clone();
63        {
64            let mut query = url.query_pairs_mut();
65            query.append_pair("uploads", "");
66            if let Some(prefix) = prefix {
67                query.append_pair("prefix", prefix);
68            }
69            if let Some(marker) = key_marker.as_deref() {
70                query.append_pair("key-marker", marker);
71            }
72            if let Some(marker) = upload_id_marker.as_deref() {
73                query.append_pair("upload-id-marker", marker);
74            }
75        }
76        let page: ListResponse =
77            signed_s3_get_xml(&client, &credential, cfg, url, "ListMultipartUploads").await?;
78        found.extend(
79            page.uploads
80                .into_iter()
81                .map(|upload| IncompleteMultipartUpload {
82                    key: upload.key,
83                    upload_id: upload.upload_id,
84                }),
85        );
86        if !page.is_truncated {
87            break;
88        }
89        key_marker = page.next_key_marker;
90        upload_id_marker = page.next_upload_id_marker;
91        if key_marker.is_none() && upload_id_marker.is_none() {
92            return Err(ObjectStoreError::Backend(
93                "truncated ListMultipartUploads response has no continuation marker".into(),
94            ));
95        }
96    }
97    Ok(found)
98}
99
100#[cfg(test)]
101mod tests {
102    use {
103        super::*,
104        crate::test_support::{HttpPage, HttpServer, s3_config, serve_http_pages},
105        assert2::check,
106        quick_xml::de,
107    };
108
109    async fn with_listing<T>(
110        pages: Vec<HttpPage>,
111        prefix: Option<&str>,
112        extract: impl FnOnce(Result<Vec<IncompleteMultipartUpload>, ObjectStoreError>) -> T,
113    ) -> T {
114        let HttpServer { endpoint, server } = serve_http_pages(pages, true).await;
115        let result = extract(list_s3_multipart_uploads(&config(endpoint), prefix).await);
116        server.await.unwrap();
117        result
118    }
119
120    async fn check_listing_error(pages: Vec<HttpPage>, reason: &str) {
121        let error = with_listing(pages, None, Result::unwrap_err).await;
122        check!(error.to_string().contains(reason));
123    }
124
125    fn config(endpoint: String) -> S3Config { s3_config("bucket", endpoint) }
126
127    #[test]
128    fn parses_incomplete_uploads_and_pagination() {
129        let page: ListResponse = de::from_str(
130            "<ListMultipartUploadsResult>
131                <IsTruncated>true</IsTruncated>
132                <NextKeyMarker>worm/b</NextKeyMarker>
133                <NextUploadIdMarker>next</NextUploadIdMarker>
134                <Upload><Key>worm/a</Key><UploadId>one</UploadId></Upload>
135                <Upload><Key>worm/b</Key><UploadId>two</UploadId></Upload>
136            </ListMultipartUploadsResult>",
137        )
138        .unwrap();
139
140        check!(page.is_truncated);
141        check!(page.next_key_marker.as_deref() == Some("worm/b"));
142        check!(page.next_upload_id_marker.as_deref() == Some("next"));
143        check!(page.uploads.len() == 2);
144        check!(page.uploads[0].key == "worm/a");
145        check!(page.uploads[0].upload_id == "one");
146    }
147
148    #[tokio::test]
149    async fn lists_incomplete_uploads_under_the_prefix() {
150        let body = "<ListMultipartUploadsResult><IsTruncated>false</IsTruncated><Upload><Key>worm/a</Key><UploadId>one</UploadId></Upload></ListMultipartUploadsResult>";
151        let uploads = with_listing(
152            vec![HttpPage::new(
153                "GET /bucket?uploads=&prefix=worm%2F HTTP/1.1",
154                "200 OK",
155                body,
156            )],
157            Some("worm/"),
158            Result::unwrap,
159        )
160        .await;
161
162        check!(
163            uploads
164                == vec![IncompleteMultipartUpload {
165                    key: "worm/a".into(),
166                    upload_id: "one".into(),
167                }]
168        );
169    }
170
171    #[tokio::test]
172    async fn follows_multipart_pagination_markers() {
173        let first = "<ListMultipartUploadsResult><IsTruncated>true</IsTruncated><NextKeyMarker>worm/a</NextKeyMarker><NextUploadIdMarker>one</NextUploadIdMarker><Upload><Key>worm/a</Key><UploadId>one</UploadId></Upload></ListMultipartUploadsResult>";
174        let second = "<ListMultipartUploadsResult><IsTruncated>false</IsTruncated><Upload><Key>worm/b</Key><UploadId>two</UploadId></Upload></ListMultipartUploadsResult>";
175        let uploads = with_listing(vec![
176            HttpPage::new(
177                "GET /bucket?uploads=&prefix=worm%2F HTTP/1.1",
178                "200 OK",
179                first,
180            ),
181            HttpPage::new(
182                "GET /bucket?uploads=&prefix=worm%2F&key-marker=worm%2Fa&upload-id-marker=one HTTP/1.1",
183                "200 OK",
184                second,
185            ),
186        ], Some("worm/"), Result::unwrap).await;
187
188        check!(uploads.len() == 2);
189        check!(uploads[1].key == "worm/b");
190        check!(uploads[1].upload_id == "two");
191    }
192
193    #[tokio::test]
194    async fn rejects_truncated_page_without_markers() {
195        let body = "<ListMultipartUploadsResult><IsTruncated>true</IsTruncated></ListMultipartUploadsResult>";
196        check_listing_error(
197            vec![HttpPage::new(
198                "GET /bucket?uploads= HTTP/1.1",
199                "200 OK",
200                body,
201            )],
202            "no continuation marker",
203        )
204        .await;
205    }
206
207    #[tokio::test]
208    async fn reports_unsuccessful_multipart_listing() {
209        check_listing_error(
210            vec![HttpPage::new(
211                "GET /bucket?uploads= HTTP/1.1",
212                "403 Forbidden",
213                "denied",
214            )],
215            "403 Forbidden: denied",
216        )
217        .await;
218    }
219}
krabka_object_store/multipart.rs

krabka_object_store/
multipart.rs
