Skip to main content

max / makenotwork

44.6 KB · 1231 lines History Blame Raw
1 //! Every `S3Client` method, against a real object store.
2 //!
3 //! WHY THIS FILE EXISTS. The crate had no integration test of any kind, and a
4 //! mutation run said what that costs: 114 of 206 mutants survived, 104 of them
5 //! inside `S3Client` (infra `1498fe2a`). `upload -> Ok(())` passed the whole
6 //! suite, because nothing in this repo ever called `upload`. The two worth
7 //! naming: `attempt < 3` can become `true` in all three multipart drivers,
8 //! which is an infinite retry on a permanent failure, and the part-offset
9 //! arithmetic in the copy driver can be wrong in either direction, which is
10 //! silent corruption of creator media -- the object completes and the bytes are
11 //! the wrong ones. Nothing short of a real round trip observes that.
12 //!
13 //! A round trip against a healthy MinIO is still not enough for the first of
14 //! those: a retry loop only runs when something fails, and nothing here ever
15 //! does. The last section of this file is a proxy that fails on purpose, which
16 //! is what reaches all three loops (infra `8baa89c6`).
17 //!
18 //! HOW TO RUN IT. The tier is astra, where MinIO is a local service:
19 //!
20 //! set -a; . ~/.config/s3-storage-tests.env; set +a
21 //! cargo nextest run --run-ignored all
22 //!
23 //! Every test is `#[ignore]`d so a developer machine reports them as SKIPPED
24 //! rather than passing silently. A test that did not run must never read as one
25 //! that did, which is the same rule the sweep applies to a cell it could not
26 //! evaluate.
27 //!
28 //! ISOLATION. `S3_TEST_*` names a bucket the credentials can reach and nothing
29 //! else: the key is scoped to it and MinIO refuses a write to the media buckets
30 //! (infra `6f7c7b46`, measured). Belt and braces anyway, every test works under
31 //! its own unique prefix and deletes it on the way out, so two runs cannot
32 //! collide and a leaked object is traceable to a run.
33 //!
34 //! MINIO IS NOT S3, and where they differ these tests assert the property that
35 //! matters rather than the exact bytes of a header. Multipart ETags are the
36 //! usual example: their format is not contractual, so nothing here reads one.
37
38 use std::io::{Read, Write};
39 use std::net::{Shutdown, TcpListener, TcpStream};
40 use std::sync::atomic::{AtomicU32, Ordering};
41 use std::sync::{Arc, Mutex, OnceLock};
42 use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
43
44 use s3_storage::{S3Client, S3Config};
45
46 const SKIP: &str = "live S3: set S3_TEST_* and run with --run-ignored (see the module docs)";
47
48 /// The bucket and credentials, or a panic naming what is missing.
49 ///
50 /// A panic rather than a skip: these tests are `#[ignore]`d, so the only way to
51 /// reach this code is to have asked for them, and silently passing an asked-for
52 /// test because a variable was unset is the failure this whole file is about.
53 fn config() -> S3Config {
54 fn var(name: &str) -> String {
55 std::env::var(name)
56 .unwrap_or_else(|_| panic!("{name} is not set; source ~/.config/s3-storage-tests.env"))
57 }
58 S3Config {
59 endpoint: var("S3_TEST_ENDPOINT"),
60 bucket: var("S3_TEST_BUCKET"),
61 access_key: var("S3_TEST_ACCESS_KEY"),
62 secret_key: var("S3_TEST_SECRET_KEY"),
63 region: std::env::var("S3_TEST_REGION").unwrap_or_else(|_| "us-east-1".into()),
64 }
65 }
66
67 async fn client() -> S3Client {
68 S3Client::new(&config())
69 .await
70 .expect("building the client should not need the network")
71 }
72
73 /// A prefix no other run will use: the clock, the process, and a counter.
74 ///
75 /// All three are needed. The counter because two tests in one process start
76 /// inside the same millisecond, and the pid because a mutation run has a dozen
77 /// copies of this suite against one bucket at once (infra `1498fe2a`), where
78 /// two runs sharing a prefix would delete each other's objects and report it as
79 /// a killed mutant.
80 fn prefix(what: &str) -> String {
81 static SEQ: AtomicU32 = AtomicU32::new(0);
82 let millis = SystemTime::now()
83 .duration_since(UNIX_EPOCH)
84 .expect("the clock is after 1970")
85 .as_millis();
86 let n = SEQ.fetch_add(1, Ordering::Relaxed);
87 format!("run-{millis}-{}-{n}/{what}", std::process::id())
88 }
89
90 /// Bytes that are not all the same, so a wrong offset shows up as wrong content
91 /// rather than as an identical-looking block.
92 fn pattern(len: usize) -> Vec<u8> {
93 (0..len).map(|i| (i % 251) as u8).collect()
94 }
95
96 /// Compare without printing megabytes on failure: say where they diverge.
97 fn assert_same_bytes(expected: &[u8], actual: &[u8], what: &str) {
98 assert_eq!(
99 expected.len(),
100 actual.len(),
101 "{what}: length differs, expected {} bytes and got {}",
102 expected.len(),
103 actual.len()
104 );
105 if let Some(at) = expected.iter().zip(actual).position(|(a, b)| a != b) {
106 panic!(
107 "{what}: first difference at byte {at} of {}, expected {:#04x} and got {:#04x}",
108 expected.len(),
109 expected[at],
110 actual[at]
111 );
112 }
113 }
114
115 /// Best-effort teardown. A failed cleanup must not turn a passing test red, and
116 /// must not hide a failing one, so it reports and moves on.
117 async fn cleanup(s3: &S3Client, prefix: &str) {
118 if let Err(e) = s3.delete_prefix(prefix).await {
119 eprintln!("cleanup of {prefix} failed, leaving objects behind: {e}");
120 }
121 }
122
123 #[tokio::test]
124 #[ignore = "live S3"]
125 async fn an_object_round_trips_with_its_content_type() {
126 let s3 = client().await;
127 let key = prefix("round-trip/a.bin");
128 let body = pattern(4096);
129
130 s3.upload(
131 &key,
132 "application/octet-stream",
133 body.clone(),
134 Some("no-store"),
135 )
136 .await
137 .expect("upload");
138
139 let (got, content_type) = s3.download(&key).await.expect("download");
140 assert_same_bytes(&body, &got, "round trip");
141 assert_eq!(content_type, "application/octet-stream");
142
143 assert!(s3.object_exists(&key).await.expect("exists"));
144 assert_eq!(
145 s3.object_size(&key).await.expect("size"),
146 Some(body.len() as i64)
147 );
148
149 s3.delete(&key).await.expect("delete");
150 assert!(
151 !s3.object_exists(&key).await.expect("exists after delete"),
152 "the object survived its own deletion"
153 );
154 assert_eq!(s3.object_size(&key).await.expect("size after delete"), None);
155
156 cleanup(&s3, &key).await;
157 }
158
159 #[tokio::test]
160 #[ignore = "live S3"]
161 async fn download_head_reads_a_prefix_and_nothing_more() {
162 let s3 = client().await;
163 let key = prefix("head/a.bin");
164 let body = pattern(10_000);
165 s3.upload(&key, "application/octet-stream", body.clone(), None)
166 .await
167 .expect("upload");
168
169 let head = s3.download_head(&key, 512).await.expect("ranged read");
170 assert_same_bytes(&body[..512], &head, "ranged read");
171
172 // Asking for more than the object holds is not an error, and the answer is
173 // the object.
174 let all = s3
175 .download_head(&key, body.len() * 2)
176 .await
177 .expect("over-read");
178 assert_same_bytes(&body, &all, "over-read");
179
180 // Zero is answered without a request at all. It has to be: the range header
181 // is built as `bytes=0-{len-1}`, which underflows at zero.
182 assert!(
183 s3.download_head(&key, 0)
184 .await
185 .expect("zero-length")
186 .is_empty()
187 );
188
189 cleanup(&s3, &key).await;
190 }
191
192 #[tokio::test]
193 #[ignore = "live S3"]
194 async fn the_streaming_download_carries_the_same_bytes() {
195 // `download` is `download_buf` plus a copy, and `download_buf` is the
196 // aggregating form of this one, so the three share a path and only this one
197 // hands the caller an unread stream.
198 let s3 = client().await;
199 let key = prefix("stream/a.bin");
200 let body = pattern(200_000);
201 s3.upload(&key, "application/octet-stream", body.clone(), None)
202 .await
203 .expect("upload");
204
205 let stream = s3.download_stream(&key).await.expect("download_stream");
206 let collected = stream.collect().await.expect("draining the stream");
207 assert_same_bytes(&body, &collected.to_vec(), "streamed download");
208
209 let (buffered, content_type) = s3.download_buf(&key).await.expect("download_buf");
210 assert_same_bytes(&body, &buffered, "buffered download");
211 assert_eq!(content_type, "application/octet-stream");
212
213 cleanup(&s3, &key).await;
214 }
215
216 #[tokio::test]
217 #[ignore = "live S3"]
218 async fn a_missing_object_is_absent_rather_than_an_error() {
219 let s3 = client().await;
220 let key = prefix("missing/never-written.bin");
221 assert!(!s3.object_exists(&key).await.expect("exists"));
222 assert_eq!(s3.object_size(&key).await.expect("size"), None);
223 assert!(
224 s3.download(&key).await.is_err(),
225 "downloading nothing should fail rather than return empty"
226 );
227 }
228
229 #[tokio::test]
230 #[ignore = "live S3"]
231 async fn a_copy_carries_the_bytes_not_just_a_status() {
232 let s3 = client().await;
233 let root = prefix("copy");
234 let src = format!("{root}/src.bin");
235 let dst = format!("{root}/dst.bin");
236 let from = format!("{root}/from.bin");
237 let body = pattern(65_536);
238
239 s3.upload(&src, "application/octet-stream", body.clone(), None)
240 .await
241 .expect("upload");
242
243 s3.copy_object(&src, &dst).await.expect("copy_object");
244 let (got, _) = s3.download(&dst).await.expect("download copy");
245 assert_same_bytes(&body, &got, "copy_object");
246
247 // Same bucket, named explicitly: the cross-bucket form with the only bucket
248 // these credentials can reach.
249 s3.copy_object_from(s3.bucket(), &src, &from)
250 .await
251 .expect("copy_object_from");
252 let (got, _) = s3.download(&from).await.expect("download copy_from");
253 assert_same_bytes(&body, &got, "copy_object_from");
254
255 cleanup(&s3, &root).await;
256 }
257
258 #[tokio::test]
259 #[ignore = "live S3"]
260 async fn deleting_a_batch_reports_no_failures_and_removes_every_key() {
261 let s3 = client().await;
262 let root = prefix("batch");
263 let keys: Vec<String> = (0..5).map(|i| format!("{root}/{i}.bin")).collect();
264 for key in &keys {
265 s3.upload(key, "application/octet-stream", pattern(32), None)
266 .await
267 .expect("upload");
268 }
269 // A key that was never written, because a batch in the wild has them and S3
270 // treats deleting an absent key as success.
271 let mut with_absent = keys.clone();
272 with_absent.push(format!("{root}/never-written.bin"));
273
274 let failures = s3
275 .delete_objects(&with_absent)
276 .await
277 .expect("delete_objects");
278 assert!(failures.is_empty(), "delete_objects reported {failures:?}");
279 for key in &keys {
280 assert!(
281 !s3.object_exists(key).await.expect("exists"),
282 "{key} survived the batch delete"
283 );
284 }
285
286 cleanup(&s3, &root).await;
287 }
288
289 #[tokio::test]
290 #[ignore = "live S3"]
291 async fn delete_prefix_stops_at_the_prefix() {
292 let s3 = client().await;
293 let root = prefix("wipe");
294 let doomed = format!("{root}/doomed");
295 let kept = format!("{root}/kept");
296 for i in 0..3 {
297 s3.upload(
298 &format!("{doomed}/{i}.bin"),
299 "application/octet-stream",
300 pattern(16),
301 None,
302 )
303 .await
304 .expect("upload");
305 }
306 s3.upload(
307 &format!("{kept}/survivor.bin"),
308 "application/octet-stream",
309 pattern(16),
310 None,
311 )
312 .await
313 .expect("upload");
314
315 s3.delete_prefix(&doomed).await.expect("delete_prefix");
316
317 for i in 0..3 {
318 assert!(
319 !s3.object_exists(&format!("{doomed}/{i}.bin"))
320 .await
321 .expect("exists"),
322 "object {i} survived delete_prefix"
323 );
324 }
325 assert!(
326 s3.object_exists(&format!("{kept}/survivor.bin"))
327 .await
328 .expect("exists"),
329 "delete_prefix reached past its prefix, which is how a wipe becomes an incident"
330 );
331
332 cleanup(&s3, &root).await;
333 }
334
335 #[tokio::test]
336 #[ignore = "live S3"]
337 async fn a_multipart_upload_reads_back_byte_identical() {
338 let s3 = client().await;
339 let key = prefix("multipart/big.bin");
340 // Three parts at the 5 MiB floor, the last one deliberately short: a driver
341 // that gets the final part's length wrong passes on an exact multiple.
342 let part_size = 5 * 1024 * 1024;
343 let body = pattern(part_size * 2 + 1_234_567);
344
345 let file = std::env::temp_dir().join(format!("s3-live-{}.bin", std::process::id()));
346 std::fs::write(&file, &body).expect("writing the source file");
347
348 let uploaded = s3
349 .upload_multipart(&key, "application/octet-stream", &file, Some(part_size))
350 .await;
351 std::fs::remove_file(&file).ok();
352 uploaded.expect("upload_multipart");
353
354 assert_eq!(
355 s3.object_size(&key).await.expect("size"),
356 Some(body.len() as i64),
357 "the assembled object is the wrong length"
358 );
359 let (got, _) = s3.download(&key).await.expect("download");
360 assert_same_bytes(&body, &got, "multipart round trip");
361
362 cleanup(&s3, &key).await;
363 }
364
365 #[tokio::test]
366 #[ignore = "live S3"]
367 async fn the_default_part_size_uploads_a_file_larger_than_one_part() {
368 // `part_size: None` takes the 10 MiB default, which no other test exercises
369 // and which nothing asserts: the default is written as an arithmetic
370 // expression, and a wrong one is either a refusal at the 5 MiB floor or an
371 // upload in more parts than intended. 12 MiB is two parts at the real
372 // default and one at any smaller one.
373 let s3 = client().await;
374 let key = prefix("default-part/big.bin");
375 let body = pattern(12 * 1024 * 1024);
376
377 let file = std::env::temp_dir().join(format!("s3-live-default-{}.bin", std::process::id()));
378 std::fs::write(&file, &body).expect("writing the source file");
379 let uploaded = s3
380 .upload_multipart(&key, "application/octet-stream", &file, None)
381 .await;
382 std::fs::remove_file(&file).ok();
383 uploaded.expect("upload_multipart with the default part size");
384
385 let (got, _) = s3.download(&key).await.expect("download");
386 assert_same_bytes(&body, &got, "default part size round trip");
387
388 cleanup(&s3, &key).await;
389 }
390
391 #[tokio::test]
392 #[ignore = "live S3"]
393 async fn a_part_size_below_the_s3_floor_is_refused_before_anything_is_created() {
394 // The floor is 5 MiB and the check is `<`, so one byte under is refused and
395 // exactly 5 MiB is not. Refused BEFORE the upload is created, which is why
396 // this asserts on the pending-upload list as well as on the error: a
397 // pre-flight that creates first and validates second strands an upload
398 // nobody will ever complete or be billed for noticing.
399 let s3 = client().await;
400 let key = prefix("floor/rejected.bin");
401 let file = std::env::temp_dir().join(format!("s3-live-floor-{}.bin", std::process::id()));
402 std::fs::write(&file, pattern(1024)).expect("writing the source file");
403
404 let err = s3
405 .upload_multipart(
406 &key,
407 "application/octet-stream",
408 &file,
409 Some(5 * 1024 * 1024 - 1),
410 )
411 .await
412 .expect_err("a part below the floor must be refused");
413 assert!(
414 err.contains("at least 5 MB"),
415 "refused for the wrong reason: {err}"
416 );
417 std::fs::remove_file(&file).ok();
418
419 assert!(
420 s3.list_multipart_uploads_for_key(&key)
421 .await
422 .expect("list")
423 .is_empty(),
424 "the refusal created an upload and left it pending"
425 );
426 }
427
428 #[tokio::test]
429 #[ignore = "live S3"]
430 async fn pending_uploads_are_listed_for_their_own_key_only() {
431 // The listing filters a bucket-wide response down to one key. A filter that
432 // widened would hand a caller somebody else's upload id, and
433 // abort_multipart_upload would then cancel an upload that was going fine.
434 let s3 = client().await;
435 let root = prefix("listing");
436 let mine = format!("{root}/mine.bin");
437 let theirs = format!("{root}/theirs.bin");
438
439 let mine_id = s3
440 .create_multipart_upload(&mine, "application/octet-stream")
441 .await
442 .expect("create mine");
443 let theirs_id = s3
444 .create_multipart_upload(&theirs, "application/octet-stream")
445 .await
446 .expect("create theirs");
447
448 let listed = s3
449 .list_multipart_uploads_for_key(&mine)
450 .await
451 .expect("list mine");
452 assert!(
453 listed.contains(&mine_id),
454 "our own upload is missing: {listed:?}"
455 );
456 assert!(
457 !listed.contains(&theirs_id),
458 "the listing reached past its key: {listed:?}"
459 );
460
461 s3.abort_multipart_upload(&mine, &mine_id)
462 .await
463 .expect("abort mine");
464 s3.abort_multipart_upload(&theirs, &theirs_id)
465 .await
466 .expect("abort theirs");
467 }
468
469 #[tokio::test]
470 #[ignore = "live S3"]
471 async fn a_multipart_copy_preserves_every_byte() {
472 let s3 = client().await;
473 let root = prefix("multipart-copy");
474 let src = format!("{root}/src.bin");
475 let dst = format!("{root}/dst.bin");
476 let part_size = 5 * 1024 * 1024;
477 let body = pattern(part_size * 2 + 999_983);
478
479 let file = std::env::temp_dir().join(format!("s3-live-copy-{}.bin", std::process::id()));
480 std::fs::write(&file, &body).expect("writing the source file");
481 let uploaded = s3
482 .upload_multipart(&src, "application/octet-stream", &file, Some(part_size))
483 .await;
484 std::fs::remove_file(&file).ok();
485 uploaded.expect("upload_multipart");
486
487 s3.copy_object_multipart(
488 s3.bucket(),
489 &src,
490 &dst,
491 "application/octet-stream",
492 body.len() as u64,
493 Some(part_size),
494 )
495 .await
496 .expect("copy_object_multipart");
497
498 let (got, _) = s3.download(&dst).await.expect("download");
499 // THE ASSERTION THIS FILE WAS WRITTEN FOR. A copy driver whose part offsets
500 // are off by one part duplicates or drops a range, the object still
501 // completes, and nothing but the bytes says so.
502 assert_same_bytes(&body, &got, "multipart copy");
503
504 cleanup(&s3, &root).await;
505 }
506
507 #[tokio::test]
508 #[ignore = "live S3"]
509 async fn an_aborted_upload_leaves_nothing_behind() {
510 let s3 = client().await;
511 let key = prefix("abort/pending.bin");
512
513 let upload_id = s3
514 .create_multipart_upload(&key, "application/octet-stream")
515 .await
516 .expect("create_multipart_upload");
517 let pending = s3
518 .list_multipart_uploads_for_key(&key)
519 .await
520 .expect("list_multipart_uploads_for_key");
521 assert!(
522 pending.contains(&upload_id),
523 "the upload we just created is not listed as pending: {pending:?}"
524 );
525
526 s3.abort_multipart_upload(&key, &upload_id)
527 .await
528 .expect("abort_multipart_upload");
529 let after = s3
530 .list_multipart_uploads_for_key(&key)
531 .await
532 .expect("list after abort");
533 assert!(
534 !after.contains(&upload_id),
535 "the aborted upload is still pending, which is a bill nobody sees: {after:?}"
536 );
537 assert!(
538 !s3.object_exists(&key).await.expect("exists"),
539 "an aborted upload produced an object"
540 );
541 }
542
543 #[tokio::test]
544 #[ignore = "live S3"]
545 async fn a_presigned_url_actually_fetches_and_actually_uploads() {
546 let s3 = client().await;
547 let root = prefix("presign");
548 let download_key = format!("{root}/download.bin");
549 let upload_key = format!("{root}/upload.bin");
550 let body = pattern(2048);
551
552 s3.upload(
553 &download_key,
554 "application/octet-stream",
555 body.clone(),
556 None,
557 )
558 .await
559 .expect("upload");
560
561 let url = s3
562 .presign_download(&download_key, 300)
563 .await
564 .expect("presign_download");
565 let fetched = http_get(&url).expect("fetching the presigned URL");
566 assert_same_bytes(&body, &fetched, "presigned download");
567
568 let put_url = s3
569 .presign_upload(
570 &upload_key,
571 "application/octet-stream",
572 300,
573 None,
574 Some(body.len() as i64),
575 )
576 .await
577 .expect("presign_upload");
578 http_put(&put_url, "application/octet-stream", &body).expect("PUT to the presigned URL");
579 let (got, _) = s3
580 .download(&upload_key)
581 .await
582 .expect("download what was PUT");
583 assert_same_bytes(&body, &got, "presigned upload");
584
585 cleanup(&s3, &root).await;
586 }
587
588 #[tokio::test]
589 #[ignore = "live S3"]
590 async fn connectivity_answers_for_the_configured_bucket() {
591 let s3 = client().await;
592 s3.check_connectivity().await.expect("check_connectivity");
593
594 let mut wrong = config();
595 wrong.bucket = format!("{}-does-not-exist", wrong.bucket);
596 let s3 = S3Client::new(&wrong).await.expect("client");
597 assert!(
598 s3.check_connectivity().await.is_err(),
599 "connectivity passed against a bucket that does not exist, so it is not checking the bucket"
600 );
601 }
602
603 // ---------------------------------------------------------------------------
604 // A minimal HTTP client, because a presigned URL is only worth anything if
605 // something outside this crate can use it.
606 //
607 // Raw TCP rather than a dependency: the tier's endpoint is MinIO on localhost
608 // over plain HTTP, and adding an HTTP stack to dev-dependencies to issue two
609 // requests would pull a TLS backend into a crate that deliberately pins its own
610 // (see Cargo.toml on `rustls-ring`). If the tier ever points at an https
611 // endpoint, these two helpers are what has to change, and they will fail loudly
612 // rather than quietly skip.
613 // ---------------------------------------------------------------------------
614
615 fn split_url(url: &str) -> (String, String) {
616 let rest = url
617 .strip_prefix("http://")
618 .unwrap_or_else(|| panic!("this helper speaks plain HTTP only, got: {url}"));
619 match rest.split_once('/') {
620 Some((host, path)) => (host.to_string(), format!("/{path}")),
621 None => (rest.to_string(), "/".to_string()),
622 }
623 }
624
625 fn read_response(mut stream: TcpStream) -> Result<Vec<u8>, String> {
626 let mut raw = Vec::new();
627 stream
628 .read_to_end(&mut raw)
629 .map_err(|e| format!("reading the response: {e}"))?;
630 let split = raw
631 .windows(4)
632 .position(|w| w == b"\r\n\r\n")
633 .ok_or_else(|| "no header terminator in the response".to_string())?;
634 let headers = String::from_utf8_lossy(&raw[..split]).to_string();
635 let status = headers
636 .lines()
637 .next()
638 .unwrap_or_default()
639 .split_whitespace()
640 .nth(1)
641 .unwrap_or_default()
642 .to_string();
643 if !status.starts_with('2') {
644 return Err(format!("HTTP {status}: {headers}"));
645 }
646 Ok(raw[split + 4..].to_vec())
647 }
648
649 fn http_get(url: &str) -> Result<Vec<u8>, String> {
650 let (host, path) = split_url(url);
651 let mut stream = TcpStream::connect(&host).map_err(|e| format!("connecting to {host}: {e}"))?;
652 let req = format!("GET {path} HTTP/1.1\r\nHost: {host}\r\nConnection: close\r\n\r\n");
653 stream
654 .write_all(req.as_bytes())
655 .map_err(|e| format!("writing the request: {e}"))?;
656 read_response(stream)
657 }
658
659 fn http_put(url: &str, content_type: &str, body: &[u8]) -> Result<Vec<u8>, String> {
660 let (host, path) = split_url(url);
661 let mut stream = TcpStream::connect(&host).map_err(|e| format!("connecting to {host}: {e}"))?;
662 let head = format!(
663 "PUT {path} HTTP/1.1\r\nHost: {host}\r\nContent-Type: {content_type}\r\n\
664 Content-Length: {}\r\nConnection: close\r\n\r\n",
665 body.len()
666 );
667 stream
668 .write_all(head.as_bytes())
669 .map_err(|e| format!("writing the request head: {e}"))?;
670 stream
671 .write_all(body)
672 .map_err(|e| format!("writing the body: {e}"))?;
673 read_response(stream)
674 }
675
676 // ---------------------------------------------------------------------------
677 // A deliberately unreliable proxy, because a healthy MinIO never fails.
678 //
679 // WHY. Three multipart drivers carry a hand-written retry loop -- `attempt < 3`
680 // and `200 * (1 << ((attempt - 1) * 2))`, three verbatim copies. Every line of
681 // all three runs only on a transient failure, so the tests above cannot reach
682 // them however thorough they are, and a mutation run says so: `attempt < 3`
683 // surviving as `true` is an infinite retry against a permanent failure, and the
684 // backoff can become almost any expression with nothing to notice (infra
685 // `8baa89c6`). The only way to observe a retry from outside is to cause one.
686 //
687 // HOW. A `TcpListener` on loopback that forwards to the real endpoint, except
688 // that it closes the first N connections whose request head matches -- no
689 // response, which the SDK sees as a transport error. `S3_TEST_ENDPOINT` is
690 // pointed at the proxy for that one test.
691 //
692 // Raw TCP again, and for the reason the presign helpers above give: this crate
693 // pins its own TLS backend and a proxy dependency would drag another one in.
694 //
695 // ONE REQUEST PER CONNECTION. The proxy inserts `Connection: close` into the
696 // forwarded head, so MinIO answers and hangs up rather than keeping the socket
697 // for the next request. Without it the fault counter would mean "connections"
698 // while the assertions mean "requests", and connection reuse would decide the
699 // difference. SigV4 signs a named header list that never includes `Connection`,
700 // so the insertion does not disturb the signature.
701 // ---------------------------------------------------------------------------
702
703 /// A proxy in front of the object store that fails on purpose.
704 ///
705 /// It lives for the rest of the test process: the accept loop has no shutdown
706 /// path, because a test binary that is about to exit does not need one and the
707 /// alternative is a nonblocking loop that spins.
708 struct FaultProxy {
709 endpoint: String,
710 injected: Arc<AtomicU32>,
711 }
712
713 impl FaultProxy {
714 /// Forward to the configured endpoint, closing the first `fail_first`
715 /// connections whose request head starts with `method` and contains
716 /// `needle`. `u32::MAX` fails every one of them, which is the permanent
717 /// failure case.
718 fn start(method: &'static str, needle: &'static str, fail_first: u32) -> Self {
719 let (upstream, _) = split_url(&config().endpoint);
720 let listener = TcpListener::bind("127.0.0.1:0").expect("binding the fault proxy");
721 let port = listener
722 .local_addr()
723 .expect("the proxy has an address")
724 .port();
725 let injected = Arc::new(AtomicU32::new(0));
726 let counter = Arc::clone(&injected);
727
728 std::thread::spawn(move || {
729 for conn in listener.incoming() {
730 let Ok(client) = conn else { continue };
731 let upstream = upstream.clone();
732 let counter = Arc::clone(&counter);
733 std::thread::spawn(move || {
734 proxy_one(client, &upstream, method, needle, fail_first, &counter);
735 });
736 }
737 });
738
739 Self {
740 endpoint: format!("http://127.0.0.1:{port}"),
741 injected,
742 }
743 }
744
745 /// How many faults were actually injected. Asserted rather than assumed: a
746 /// test that passes because the proxy never matched anything is a test that
747 /// proves nothing, and it would look identical to a passing retry.
748 fn injected(&self) -> u32 {
749 self.injected.load(Ordering::SeqCst)
750 }
751 }
752
753 fn proxy_one(
754 mut client: TcpStream,
755 upstream: &str,
756 method: &'static str,
757 needle: &'static str,
758 fail_first: u32,
759 injected: &AtomicU32,
760 ) {
761 // Bound both sides. A hung socket here would surface as a test that never
762 // finishes, which is worse than one that fails.
763 let timeout = Some(Duration::from_secs(30));
764 client.set_read_timeout(timeout).ok();
765 client.set_write_timeout(timeout).ok();
766
767 // Byte at a time to the header terminator, so the body is left in the
768 // socket for the pump below rather than half-read into this buffer. Heads
769 // are a kilobyte or so; this is a test.
770 let mut head = Vec::new();
771 let mut byte = [0u8; 1];
772 while head.len() < 64 * 1024 {
773 match client.read(&mut byte) {
774 Ok(0) | Err(_) => return,
775 Ok(_) => head.push(byte[0]),
776 }
777 if head.ends_with(b"\r\n\r\n") {
778 break;
779 }
780 }
781 let head = String::from_utf8_lossy(&head).to_string();
782
783 if head.starts_with(method) && head.contains(needle) {
784 let claimed = injected
785 .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |n| {
786 (n < fail_first).then_some(n + 1)
787 })
788 .is_ok();
789 if claimed {
790 // Hang up without answering.
791 client.shutdown(Shutdown::Both).ok();
792 return;
793 }
794 }
795
796 let Ok(mut server) = TcpStream::connect(upstream) else {
797 return;
798 };
799 server.set_read_timeout(timeout).ok();
800 server.set_write_timeout(timeout).ok();
801
802 let (request_line, rest) = head.split_once("\r\n").unwrap_or((head.as_str(), ""));
803 let forwarded = format!("{request_line}\r\nConnection: close\r\n{rest}");
804 if server.write_all(forwarded.as_bytes()).is_err() {
805 return;
806 }
807
808 let mut from_server = server.try_clone().expect("cloning the upstream socket");
809 let mut to_client = client.try_clone().expect("cloning the client socket");
810 let back = std::thread::spawn(move || {
811 std::io::copy(&mut from_server, &mut to_client).ok();
812 to_client.shutdown(Shutdown::Write).ok();
813 });
814 std::io::copy(&mut client, &mut server).ok();
815 server.shutdown(Shutdown::Write).ok();
816 back.join().ok();
817 }
818
819 // ---------------------------------------------------------------------------
820 // Reading the backoff off the crate's own warning, because the wall clock
821 // cannot see it.
822 //
823 // A lower bound on elapsed time proves a sleep happened. It does NOT prove the
824 // sleep was the right length, and that was measured rather than assumed: with
825 // `200 * (1 << ((attempt - 1) * 2))` replaced by a flat `200`, an upload that
826 // should have cost 200ms + 800ms of backoff still finished inside the one-second
827 // bound and the test passed. The SDK runs its own retry policy under ours and
828 // its backoff is close to a second per attempt, so it swamps the difference
829 // between 200ms and 800ms in any whole-operation timing.
830 //
831 // So the delay expression is read where it is unambiguous: each loop logs
832 // `delay_ms` on the warning it emits before sleeping. A mutant that changes the
833 // arithmetic changes that number. Pairing the two -- the field for the value,
834 // the clock for the fact that a sleep occurred -- covers both halves, and
835 // neither covers both alone.
836 // ---------------------------------------------------------------------------
837
838 /// Every `delay_ms` the crate has logged in this process, in order.
839 fn delays() -> &'static Mutex<Vec<u64>> {
840 static DELAYS: OnceLock<Mutex<Vec<u64>>> = OnceLock::new();
841 static INSTALLED: OnceLock<()> = OnceLock::new();
842 let cell = DELAYS.get_or_init(|| Mutex::new(Vec::new()));
843 INSTALLED.get_or_init(|| {
844 // Global, and that is sound here because nextest runs one test per
845 // process -- which is how this file is documented to be run. Under a
846 // shared-process `cargo test` two concurrent retry tests would append
847 // to one list, so each test reads only the tail it appended.
848 tracing::subscriber::set_global_default(DelayCollector)
849 .expect("no other subscriber should be installed in a test process");
850 });
851 cell
852 }
853
854 struct DelayCollector;
855
856 impl tracing::Subscriber for DelayCollector {
857 fn enabled(&self, _: &tracing::Metadata<'_>) -> bool {
858 true
859 }
860 fn new_span(&self, _: &tracing::span::Attributes<'_>) -> tracing::Id {
861 tracing::Id::from_u64(1)
862 }
863 fn record(&self, _: &tracing::Id, _: &tracing::span::Record<'_>) {}
864 fn record_follows_from(&self, _: &tracing::Id, _: &tracing::Id) {}
865 fn event(&self, event: &tracing::Event<'_>) {
866 struct Pick(Option<u64>);
867 impl tracing::field::Visit for Pick {
868 fn record_u64(&mut self, field: &tracing::field::Field, value: u64) {
869 if field.name() == "delay_ms" {
870 self.0 = Some(value);
871 }
872 }
873 fn record_debug(&mut self, _: &tracing::field::Field, _: &dyn std::fmt::Debug) {}
874 }
875 let mut pick = Pick(None);
876 event.record(&mut pick);
877 if let Some(ms) = pick.0 {
878 delays()
879 .lock()
880 .expect("the delay list is not poisoned")
881 .push(ms);
882 }
883 }
884 fn enter(&self, _: &tracing::Id) {}
885 fn exit(&self, _: &tracing::Id) {}
886 }
887
888 /// The delays logged since `from`, which is where the caller's own work started.
889 fn delays_since(from: usize) -> Vec<u64> {
890 delays().lock().expect("the delay list is not poisoned")[from..].to_vec()
891 }
892
893 fn delays_so_far() -> usize {
894 delays()
895 .lock()
896 .expect("the delay list is not poisoned")
897 .len()
898 }
899
900 /// The same client the other tests build, pointed at a proxy.
901 async fn client_via(proxy: &FaultProxy) -> S3Client {
902 let mut config = config();
903 config.endpoint.clone_from(&proxy.endpoint);
904 S3Client::new(&config)
905 .await
906 .expect("building the client should not need the network")
907 }
908
909 /// Two parts and a short tail, small enough that the parts are not the point:
910 /// what is being exercised is the completion that follows them.
911 async fn upload_through(proxy: &FaultProxy, key: &str) -> (Result<(), String>, Vec<u8>, Duration) {
912 let s3 = client_via(proxy).await;
913 let part_size = 5 * 1024 * 1024;
914 let body = pattern(part_size + 1_000);
915 let file =
916 std::env::temp_dir().join(format!("s3-fault-{}-{}.bin", std::process::id(), key.len()));
917 std::fs::write(&file, &body).expect("writing the source file");
918
919 // The timeout is load-bearing, not defensive. `attempt < 3` mutated to
920 // `true` is a loop that never ends, and an assertion placed after the call
921 // never runs: the test hangs instead of failing, which reads as a mutant
922 // that survived. A bound turns the hang into a red test.
923 let started = Instant::now();
924 let result = tokio::time::timeout(
925 Duration::from_mins(2),
926 s3.upload_multipart(key, "application/octet-stream", &file, Some(part_size)),
927 )
928 .await
929 .unwrap_or_else(|_| Err("upload_multipart never returned".to_string()));
930 let elapsed = started.elapsed();
931 std::fs::remove_file(&file).ok();
932 (result, body, elapsed)
933 }
934
935 /// The completion is POST with an `uploadId`; nothing else in a multipart
936 /// upload is. `CreateMultipartUpload` is a POST too, but carries `uploads`
937 /// rather than `uploadId=`.
938 const COMPLETE: (&str, &str) = ("POST", "uploadId=");
939
940 /// A part, uploaded or copied: both are a PUT carrying a part number.
941 const PART: (&str, &str) = ("PUT", "partNumber=");
942
943 // How many closed connections it takes to spend one of the crate's three
944 // attempts. Not one: the SDK runs its own retry policy underneath, so a single
945 // closed connection is absorbed before the crate's loop ever sees an error.
946 // These are MEASURED against MinIO on astra rather than derived from the SDK's
947 // defaults, because the number that matters is what the two policies do
948 // together, and a derived number would silently rot when either changes.
949 const PER_ATTEMPT: u32 = 3;
950 const FAIL_ONE_ATTEMPT: u32 = PER_ATTEMPT;
951 const FAIL_TWO_ATTEMPTS: u32 = PER_ATTEMPT * 2;
952
953 #[tokio::test]
954 #[ignore = "live S3"]
955 async fn a_retried_completion_still_writes_the_right_bytes() {
956 let proxy = FaultProxy::start(COMPLETE.0, COMPLETE.1, FAIL_ONE_ATTEMPT);
957 let key = prefix("fault/retried.bin");
958 let mark = delays_so_far();
959 let (result, body, elapsed) = upload_through(&proxy, &key).await;
960 result.expect("the upload should survive one failed completion");
961
962 assert!(
963 proxy.injected() > 0,
964 "the proxy never matched a completion, so nothing was retried and this test proved nothing"
965 );
966 // One failed attempt, one backoff, and it is the first rung: 200ms.
967 assert_eq!(
968 delays_since(mark),
969 vec![200],
970 "the completion loop backed off with the wrong delays"
971 );
972 assert!(
973 elapsed >= Duration::from_millis(200),
974 "returned in {elapsed:?}, so the delay was logged and not slept"
975 );
976
977 // The retry is only interesting if the object is right afterwards. A driver
978 // that retried and assembled the wrong parts would pass a bare `is_ok`.
979 let s3 = client().await;
980 let (got, _) = s3.download(&key).await.expect("download");
981 assert_same_bytes(&body, &got, "the object written across a retry");
982
983 cleanup(&s3, &key).await;
984 }
985
986 #[tokio::test]
987 #[ignore = "live S3"]
988 async fn two_failures_walk_up_the_backoff() {
989 let proxy = FaultProxy::start(COMPLETE.0, COMPLETE.1, FAIL_TWO_ATTEMPTS);
990 let key = prefix("fault/twice.bin");
991 let mark = delays_so_far();
992 let (result, body, elapsed) = upload_through(&proxy, &key).await;
993 result.expect("the upload should survive two failed completions");
994
995 // THE ASSERTION THE BACKOFF ARITHMETIC EXISTS FOR. `200 * (1 << ((attempt
996 // - 1) * 2))` is 200 then 800; a flat 200, a doubling, or an off-by-one on
997 // the exponent all produce a different second number.
998 assert_eq!(
999 delays_since(mark),
1000 vec![200, 800],
1001 "the completion loop backed off with the wrong delays"
1002 );
1003 assert!(
1004 elapsed >= Duration::from_secs(1),
1005 "returned in {elapsed:?}, faster than the 200ms + 800ms it says it slept"
1006 );
1007
1008 let s3 = client().await;
1009 let (got, _) = s3.download(&key).await.expect("download");
1010 assert_same_bytes(&body, &got, "the object written across two retries");
1011 cleanup(&s3, &key).await;
1012 }
1013
1014 #[tokio::test]
1015 #[ignore = "live S3"]
1016 async fn a_permanent_failure_gives_up_rather_than_looping() {
1017 let proxy = FaultProxy::start(COMPLETE.0, COMPLETE.1, u32::MAX);
1018 let key = prefix("fault/permanent.bin");
1019 let mark = delays_so_far();
1020 let (result, _, elapsed) = upload_through(&proxy, &key).await;
1021
1022 let error = result.expect_err("a completion that never succeeds must not report success");
1023 assert!(
1024 error.contains("after retries"),
1025 "gave up with the wrong error, which suggests it left the retry loop by another path \
1026 -- or never left it at all, which is what the timeout reports: {error}"
1027 );
1028
1029 // Three attempts means two backoffs and then a decision, not a third sleep.
1030 // This is the `attempt < 3` guard read as a number: a loop that ran once
1031 // more would log a third delay.
1032 assert_eq!(
1033 delays_since(mark),
1034 vec![200, 800],
1035 "gave up after the wrong number of attempts"
1036 );
1037
1038 // The `attempt < 3` -> `true` mutant is an infinite loop, and the only way
1039 // to fail a test on an infinite loop is a wall clock. Measured: the mutant
1040 // makes this test fail on `upload_multipart never returned`.
1041 assert!(
1042 elapsed < Duration::from_mins(1),
1043 "took {elapsed:?} to give up, which is the shape of a retry loop that does not stop"
1044 );
1045
1046 let s3 = client().await;
1047 cleanup(&s3, &key).await;
1048 }
1049
1050 #[tokio::test]
1051 #[ignore = "live S3"]
1052 async fn a_retried_part_upload_still_writes_the_right_bytes() {
1053 // The second of the three loops, in `run_multipart_upload`. A part is a PUT
1054 // carrying a part number; the completion that follows is a POST, so this
1055 // proxy leaves it alone.
1056 let proxy = FaultProxy::start(PART.0, PART.1, FAIL_ONE_ATTEMPT);
1057 let key = prefix("fault/part.bin");
1058 let mark = delays_so_far();
1059 let (result, body, elapsed) = upload_through(&proxy, &key).await;
1060 result.expect("the upload should survive one failed part");
1061
1062 assert!(proxy.injected() > 0, "no part upload was ever failed");
1063 assert_eq!(
1064 delays_since(mark),
1065 vec![200],
1066 "the part loop backed off with the wrong delays"
1067 );
1068 assert!(
1069 elapsed >= Duration::from_millis(200),
1070 "returned in {elapsed:?}"
1071 );
1072
1073 // A part driver that retried by re-sending the wrong slice of the file
1074 // completes an object of the right length and the wrong contents.
1075 let s3 = client().await;
1076 let (got, _) = s3.download(&key).await.expect("download");
1077 assert_same_bytes(&body, &got, "the object written across a failed part");
1078 cleanup(&s3, &key).await;
1079 }
1080
1081 #[tokio::test]
1082 #[ignore = "live S3"]
1083 async fn a_retried_copy_part_still_copies_every_byte() {
1084 // The third loop, in `run_multipart_copy`. Same shape as a part upload from
1085 // the wire's point of view, which is why one needle reaches both: what
1086 // separates them is which driver the test drives.
1087 let s3 = client().await;
1088 let root = prefix("fault/copy");
1089 let src = format!("{root}/src.bin");
1090 let dst = format!("{root}/dst.bin");
1091 let part_size = 5 * 1024 * 1024;
1092 let body = pattern(part_size + 1_000);
1093
1094 let file = std::env::temp_dir().join(format!("s3-fault-copy-{}.bin", std::process::id()));
1095 std::fs::write(&file, &body).expect("writing the source file");
1096 let uploaded = s3
1097 .upload_multipart(&src, "application/octet-stream", &file, Some(part_size))
1098 .await;
1099 std::fs::remove_file(&file).ok();
1100 uploaded.expect("the source upload runs against the real endpoint");
1101
1102 let proxy = FaultProxy::start(PART.0, PART.1, FAIL_ONE_ATTEMPT);
1103 let faulty = client_via(&proxy).await;
1104 let mark = delays_so_far();
1105 let started = Instant::now();
1106 let copied = tokio::time::timeout(
1107 Duration::from_mins(2),
1108 faulty.copy_object_multipart(
1109 s3.bucket(),
1110 &src,
1111 &dst,
1112 "application/octet-stream",
1113 body.len() as u64,
1114 Some(part_size),
1115 ),
1116 )
1117 .await
1118 .unwrap_or_else(|_| Err("copy_object_multipart never returned".to_string()));
1119 let elapsed = started.elapsed();
1120 copied.expect("the copy should survive one failed part");
1121
1122 assert!(proxy.injected() > 0, "no copy part was ever failed");
1123 assert_eq!(
1124 delays_since(mark),
1125 vec![200],
1126 "the copy loop backed off with the wrong delays"
1127 );
1128 assert!(
1129 elapsed >= Duration::from_millis(200),
1130 "returned in {elapsed:?}"
1131 );
1132
1133 let (got, _) = s3.download(&dst).await.expect("download");
1134 // The offset arithmetic is what a retried copy part can get wrong, and only
1135 // the bytes say so: the object completes either way.
1136 assert_same_bytes(&body, &got, "the object copied across a failed part");
1137 cleanup(&s3, &root).await;
1138 }
1139
1140 #[tokio::test]
1141 #[ignore = "live S3"]
1142 async fn a_part_that_never_uploads_gives_up_rather_than_looping() {
1143 // The part loop's own `attempt < 3`. Its completion counterpart above
1144 // cannot reach this one: a part that fails forever never gets to a
1145 // completion.
1146 let proxy = FaultProxy::start(PART.0, PART.1, u32::MAX);
1147 let key = prefix("fault/part-permanent.bin");
1148 let mark = delays_so_far();
1149 let (result, _, elapsed) = upload_through(&proxy, &key).await;
1150
1151 let error = result.expect_err("a part that never uploads must not report success");
1152 assert!(
1153 error.contains("after retries"),
1154 "gave up with the wrong error, or never gave up at all: {error}"
1155 );
1156 assert_eq!(
1157 delays_since(mark),
1158 vec![200, 800],
1159 "the part loop gave up after the wrong number of attempts"
1160 );
1161 assert!(
1162 elapsed < Duration::from_mins(1),
1163 "took {elapsed:?} to give up, which is the shape of a retry loop that does not stop"
1164 );
1165
1166 let s3 = client().await;
1167 cleanup(&s3, &key).await;
1168 }
1169
1170 #[tokio::test]
1171 #[ignore = "live S3"]
1172 async fn a_copy_part_that_never_lands_gives_up_rather_than_looping() {
1173 let s3 = client().await;
1174 let root = prefix("fault/copy-permanent");
1175 let src = format!("{root}/src.bin");
1176 let dst = format!("{root}/dst.bin");
1177 let part_size = 5 * 1024 * 1024;
1178 let body = pattern(part_size + 1_000);
1179
1180 let file = std::env::temp_dir().join(format!("s3-fault-copyp-{}.bin", std::process::id()));
1181 std::fs::write(&file, &body).expect("writing the source file");
1182 let uploaded = s3
1183 .upload_multipart(&src, "application/octet-stream", &file, Some(part_size))
1184 .await;
1185 std::fs::remove_file(&file).ok();
1186 uploaded.expect("the source upload runs against the real endpoint");
1187
1188 let proxy = FaultProxy::start(PART.0, PART.1, u32::MAX);
1189 let faulty = client_via(&proxy).await;
1190 let mark = delays_so_far();
1191 let started = Instant::now();
1192 let copied = tokio::time::timeout(
1193 Duration::from_mins(2),
1194 faulty.copy_object_multipart(
1195 s3.bucket(),
1196 &src,
1197 &dst,
1198 "application/octet-stream",
1199 body.len() as u64,
1200 Some(part_size),
1201 ),
1202 )
1203 .await
1204 .unwrap_or_else(|_| Err("copy_object_multipart never returned".to_string()));
1205 let elapsed = started.elapsed();
1206
1207 let error = copied.expect_err("a copy part that never lands must not report success");
1208 assert!(
1209 error.contains("after retries"),
1210 "gave up with the wrong error, or never gave up at all: {error}"
1211 );
1212 assert_eq!(
1213 delays_since(mark),
1214 vec![200, 800],
1215 "the copy loop gave up after the wrong number of attempts"
1216 );
1217 assert!(
1218 elapsed < Duration::from_mins(1),
1219 "took {elapsed:?} to give up, which is the shape of a retry loop that does not stop"
1220 );
1221
1222 cleanup(&s3, &root).await;
1223 }
1224
1225 /// Named so `cargo nextest list` shows why the file looks empty on a laptop.
1226 #[test]
1227 #[ignore = "live S3"]
1228 fn these_tests_need_a_live_object_store() {
1229 println!("{SKIP}");
1230 }
1231