Skip to main content

max / makenotwork

Reach the three multipart retry loops with a proxy that fails on purpose Every line of the three hand-written retry loops runs only on a transient failure, and a healthy MinIO never produces one, so the live suite could not observe them however thorough it was: `attempt < 3` mutated to `true` is an infinite retry against a permanent failure, and the backoff arithmetic could become almost any expression with nothing to notice. The proxy is a TcpListener in front of the endpoint that closes the first N connections whose request head matches, and forwards the rest. Raw TCP for the reason the presign helpers give: this crate pins its own TLS backend. It inserts Connection: close into the forwarded head so one connection carries one request and the fault count means what it says. Six tests: a retried completion, part upload and copy part each write the right bytes afterwards, and each of the three loops gives up rather than looping when the failure never clears. The give-up tests are bounded by a timeout rather than an assertion, because an assertion after an infinite loop never runs. The backoff is read off the loop's own delay_ms field, not off the clock. That was measured, not assumed: with the expression replaced by a flat 200ms a whole-operation timing assertion still passed, because the SDK retries underneath ours with a backoff near a second and swamps the difference between 200ms and 800ms. Verified by hand-mutating both the guard and the backoff in each of the three loops: six mutants, six killed, no survivors. tokio gains the `time` feature it was already using. It compiled without it because an aws crate happens to enable it, and a feature you get by unification is one you lose when a dependency changes its mind.
Author: Max Johnson <me@maxj.phd> · 2026-08-28 18:30 UTC
Signed with PGP, not checked
Commit: 77dda81e863bfe6ae4a6f4992db24465536b4c16
Parent: be39295
3 files changed, +504 insertions, -21 deletions
M shared/s3-storage/Cargo.lock +17 -17
@@ -2085,27 +2085,27 @@
2085 2085
2086 2086 [[patch.unused]]
2087 2087 name = "quasi-axum"
2088 - version = "0.65.0"
2088 + version = "0.74.0"
2089 2089
2090 2090 [[patch.unused]]
2091 2091 name = "quasi-basics"
2092 - version = "0.65.0"
2092 + version = "0.74.0"
2093 2093
2094 2094 [[patch.unused]]
2095 2095 name = "quasi-http"
2096 - version = "0.65.0"
2096 + version = "0.74.0"
2097 2097
2098 2098 [[patch.unused]]
2099 2099 name = "quasi-immediate"
2100 - version = "0.65.0"
2100 + version = "0.74.0"
2101 2101
2102 2102 [[patch.unused]]
2103 2103 name = "quasi-notifs"
2104 - version = "0.65.0"
2104 + version = "0.74.0"
2105 2105
2106 2106 [[patch.unused]]
2107 2107 name = "quasi-router"
2108 - version = "0.65.0"
2108 + version = "0.74.0"
2109 2109
2110 2110 [[patch.unused]]
2111 2111 name = "quasi-store"
@@ -2113,28 +2113,20 @@
2113 2113
2114 2114 [[patch.unused]]
2115 2115 name = "quasi-tauri"
2116 - version = "0.65.0"
2116 + version = "0.74.0"
2117 2117
2118 2118 [[patch.unused]]
2119 2119 name = "quasi-webview"
2120 - version = "0.65.0"
2120 + version = "0.74.0"
2121 2121
2122 2122 [[patch.unused]]
2123 2123 name = "synckit-client"
2124 - version = "0.9.1"
2124 + version = "0.10.0"
2125 2125
2126 2126 [[patch.unused]]
2127 2127 name = "synckit-config"
2128 2128 version = "0.2.0"
2129 2129
2130 - [[patch.unused]]
2131 - name = "quasi-type"
2132 - version = "0.1.0"
2133 -
2134 - [[patch.unused]]
2135 - name = "docengine"
2136 - version = "0.7.0"
2137 -
2138 2130 [[patch.unused]]
2139 2131 name = "kberg"
2140 2132 version = "0.1.0"
@@ -2150,3 +2142,11 @@
2150 2142 [[patch.unused]]
2151 2143 name = "tagtree"
2152 2144 version = "0.4.1"
2145 +
2146 + [[patch.unused]]
2147 + name = "docengine"
2148 + version = "0.7.0"
2149 +
2150 + [[patch.unused]]
2151 + name = "quasi-type"
2152 + version = "0.1.2"
M shared/s3-storage/Cargo.toml +5 -2
@@ -18,11 +18,14 @@
18 18 aws-config = { version = "1.8", default-features = false, features = ["behavior-version-latest"] }
19 19 aws-smithy-http-client = { version = "1.1", features = ["rustls-ring"] }
20 20 bytes = "1"
21 - tokio = { version = "1", features = ["fs", "io-util"] }
21 + # `time` is what the retry loops sleep on. It compiled without it because an
22 + # aws crate happens to enable the feature, and a feature you get by unification
23 + # is one you lose when a dependency changes its mind.
24 + tokio = { version = "1", features = ["fs", "io-util", "time"] }
22 25 tracing = "0.1"
23 26
24 27 [dev-dependencies]
25 - tokio = { version = "1", features = ["macros", "rt"] }
28 + tokio = { version = "1", features = ["macros", "rt", "time"] }
26 29
27 30 [lints.rust]
28 31 unused = "warn"
M shared/s3-storage/tests/live_s3.rs +482 -2
@@ -10,6 +10,11 @@
10 10 //! silent corruption of creator media -- the object completes and the bytes are
11 11 //! the wrong ones. Nothing short of a real round trip observes that.
12 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 + //!
13 18 //! HOW TO RUN IT. The tier is astra, where MinIO is a local service:
14 19 //!
15 20 //! set -a; . ~/.config/s3-storage-tests.env; set +a
@@ -31,9 +36,10 @@
31 36 //! usual example: their format is not contractual, so nothing here reads one.
32 37
33 38 use std::io::{Read, Write};
34 - use std::net::TcpStream;
39 + use std::net::{Shutdown, TcpListener, TcpStream};
35 40 use std::sync::atomic::{AtomicU32, Ordering};
36 - use std::time::{SystemTime, UNIX_EPOCH};
41 + use std::sync::{Arc, Mutex, OnceLock};
42 + use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
37 43
38 44 use s3_storage::{S3Client, S3Config};
39 45
@@ -667,6 +673,555 @@
667 673 read_response(stream)
668 674 }
669 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;
Lines truncated