//! Every `S3Client` method, against a real object store.
//!
//! WHY THIS FILE EXISTS. The crate had no integration test of any kind, and a
//! mutation run said what that costs: 114 of 206 mutants survived, 104 of them
//! inside `S3Client` (infra `1498fe2a`). `upload -> Ok(())` passed the whole
//! suite, because nothing in this repo ever called `upload`. The two worth
//! naming: `attempt < 3` can become `true` in all three multipart drivers,
//! which is an infinite retry on a permanent failure, and the part-offset
//! arithmetic in the copy driver can be wrong in either direction, which is
//! silent corruption of creator media -- the object completes and the bytes are
//! the wrong ones. Nothing short of a real round trip observes that.
//!
//! A round trip against a healthy MinIO is still not enough for the first of
//! those: a retry loop only runs when something fails, and nothing here ever
//! does. The last section of this file is a proxy that fails on purpose, which
//! is what reaches all three loops (infra `8baa89c6`).
//!
//! HOW TO RUN IT. The tier is astra, where MinIO is a local service:
//!
//!     set -a; . ~/.config/s3-storage-tests.env; set +a
//!     cargo nextest run --run-ignored all
//!
//! Every test is `#[ignore]`d so a developer machine reports them as SKIPPED
//! rather than passing silently. A test that did not run must never read as one
//! that did, which is the same rule the sweep applies to a cell it could not
//! evaluate.
//!
//! ISOLATION. `S3_TEST_*` names a bucket the credentials can reach and nothing
//! else: the key is scoped to it and MinIO refuses a write to the media buckets
//! (infra `6f7c7b46`, measured). Belt and braces anyway, every test works under
//! its own unique prefix and deletes it on the way out, so two runs cannot
//! collide and a leaked object is traceable to a run.
//!
//! MINIO IS NOT S3, and where they differ these tests assert the property that
//! matters rather than the exact bytes of a header. Multipart ETags are the
//! usual example: their format is not contractual, so nothing here reads one.

use std::io::{Read, Write};
use std::net::{Shutdown, TcpListener, TcpStream};
use std::sync::atomic::{AtomicU32, Ordering};
use std::sync::{Arc, Mutex, OnceLock};
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};

use s3_storage::{S3Client, S3Config};

const SKIP: &str = "live S3: set S3_TEST_* and run with --run-ignored (see the module docs)";

/// The bucket and credentials, or a panic naming what is missing.
///
/// A panic rather than a skip: these tests are `#[ignore]`d, so the only way to
/// reach this code is to have asked for them, and silently passing an asked-for
/// test because a variable was unset is the failure this whole file is about.
fn config() -> S3Config {
    fn var(name: &str) -> String {
        std::env::var(name)
            .unwrap_or_else(|_| panic!("{name} is not set; source ~/.config/s3-storage-tests.env"))
    }
    S3Config {
        endpoint: var("S3_TEST_ENDPOINT"),
        bucket: var("S3_TEST_BUCKET"),
        access_key: var("S3_TEST_ACCESS_KEY"),
        secret_key: var("S3_TEST_SECRET_KEY"),
        region: std::env::var("S3_TEST_REGION").unwrap_or_else(|_| "us-east-1".into()),
    }
}

async fn client() -> S3Client {
    S3Client::new(&config())
        .await
        .expect("building the client should not need the network")
}

/// A prefix no other run will use: the clock, the process, and a counter.
///
/// All three are needed. The counter because two tests in one process start
/// inside the same millisecond, and the pid because a mutation run has a dozen
/// copies of this suite against one bucket at once (infra `1498fe2a`), where
/// two runs sharing a prefix would delete each other's objects and report it as
/// a killed mutant.
fn prefix(what: &str) -> String {
    static SEQ: AtomicU32 = AtomicU32::new(0);
    let millis = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .expect("the clock is after 1970")
        .as_millis();
    let n = SEQ.fetch_add(1, Ordering::Relaxed);
    format!("run-{millis}-{}-{n}/{what}", std::process::id())
}

/// Bytes that are not all the same, so a wrong offset shows up as wrong content
/// rather than as an identical-looking block.
fn pattern(len: usize) -> Vec<u8> {
    (0..len).map(|i| (i % 251) as u8).collect()
}

/// Compare without printing megabytes on failure: say where they diverge.
fn assert_same_bytes(expected: &[u8], actual: &[u8], what: &str) {
    assert_eq!(
        expected.len(),
        actual.len(),
        "{what}: length differs, expected {} bytes and got {}",
        expected.len(),
        actual.len()
    );
    if let Some(at) = expected.iter().zip(actual).position(|(a, b)| a != b) {
        panic!(
            "{what}: first difference at byte {at} of {}, expected {:#04x} and got {:#04x}",
            expected.len(),
            expected[at],
            actual[at]
        );
    }
}

/// Best-effort teardown. A failed cleanup must not turn a passing test red, and
/// must not hide a failing one, so it reports and moves on.
async fn cleanup(s3: &S3Client, prefix: &str) {
    if let Err(e) = s3.delete_prefix(prefix).await {
        eprintln!("cleanup of {prefix} failed, leaving objects behind: {e}");
    }
}

#[tokio::test]
#[ignore = "live S3"]
async fn an_object_round_trips_with_its_content_type() {
    let s3 = client().await;
    let key = prefix("round-trip/a.bin");
    let body = pattern(4096);

    s3.upload(
        &key,
        "application/octet-stream",
        body.clone(),
        Some("no-store"),
    )
    .await
    .expect("upload");

    let (got, content_type) = s3.download(&key).await.expect("download");
    assert_same_bytes(&body, &got, "round trip");
    assert_eq!(content_type, "application/octet-stream");

    assert!(s3.object_exists(&key).await.expect("exists"));
    assert_eq!(
        s3.object_size(&key).await.expect("size"),
        Some(body.len() as i64)
    );

    s3.delete(&key).await.expect("delete");
    assert!(
        !s3.object_exists(&key).await.expect("exists after delete"),
        "the object survived its own deletion"
    );
    assert_eq!(s3.object_size(&key).await.expect("size after delete"), None);

    cleanup(&s3, &key).await;
}

#[tokio::test]
#[ignore = "live S3"]
async fn download_head_reads_a_prefix_and_nothing_more() {
    let s3 = client().await;
    let key = prefix("head/a.bin");
    let body = pattern(10_000);
    s3.upload(&key, "application/octet-stream", body.clone(), None)
        .await
        .expect("upload");

    let head = s3.download_head(&key, 512).await.expect("ranged read");
    assert_same_bytes(&body[..512], &head, "ranged read");

    // Asking for more than the object holds is not an error, and the answer is
    // the object.
    let all = s3
        .download_head(&key, body.len() * 2)
        .await
        .expect("over-read");
    assert_same_bytes(&body, &all, "over-read");

    // Zero is answered without a request at all. It has to be: the range header
    // is built as `bytes=0-{len-1}`, which underflows at zero.
    assert!(
        s3.download_head(&key, 0)
            .await
            .expect("zero-length")
            .is_empty()
    );

    cleanup(&s3, &key).await;
}

#[tokio::test]
#[ignore = "live S3"]
async fn the_streaming_download_carries_the_same_bytes() {
    // `download` is `download_buf` plus a copy, and `download_buf` is the
    // aggregating form of this one, so the three share a path and only this one
    // hands the caller an unread stream.
    let s3 = client().await;
    let key = prefix("stream/a.bin");
    let body = pattern(200_000);
    s3.upload(&key, "application/octet-stream", body.clone(), None)
        .await
        .expect("upload");

    let stream = s3.download_stream(&key).await.expect("download_stream");
    let collected = stream.collect().await.expect("draining the stream");
    assert_same_bytes(&body, &collected.to_vec(), "streamed download");

    let (buffered, content_type) = s3.download_buf(&key).await.expect("download_buf");
    assert_same_bytes(&body, &buffered, "buffered download");
    assert_eq!(content_type, "application/octet-stream");

    cleanup(&s3, &key).await;
}

#[tokio::test]
#[ignore = "live S3"]
async fn a_missing_object_is_absent_rather_than_an_error() {
    let s3 = client().await;
    let key = prefix("missing/never-written.bin");
    assert!(!s3.object_exists(&key).await.expect("exists"));
    assert_eq!(s3.object_size(&key).await.expect("size"), None);
    assert!(
        s3.download(&key).await.is_err(),
        "downloading nothing should fail rather than return empty"
    );
}

#[tokio::test]
#[ignore = "live S3"]
async fn a_copy_carries_the_bytes_not_just_a_status() {
    let s3 = client().await;
    let root = prefix("copy");
    let src = format!("{root}/src.bin");
    let dst = format!("{root}/dst.bin");
    let from = format!("{root}/from.bin");
    let body = pattern(65_536);

    s3.upload(&src, "application/octet-stream", body.clone(), None)
        .await
        .expect("upload");

    s3.copy_object(&src, &dst).await.expect("copy_object");
    let (got, _) = s3.download(&dst).await.expect("download copy");
    assert_same_bytes(&body, &got, "copy_object");

    // Same bucket, named explicitly: the cross-bucket form with the only bucket
    // these credentials can reach.
    s3.copy_object_from(s3.bucket(), &src, &from)
        .await
        .expect("copy_object_from");
    let (got, _) = s3.download(&from).await.expect("download copy_from");
    assert_same_bytes(&body, &got, "copy_object_from");

    cleanup(&s3, &root).await;
}

#[tokio::test]
#[ignore = "live S3"]
async fn deleting_a_batch_reports_no_failures_and_removes_every_key() {
    let s3 = client().await;
    let root = prefix("batch");
    let keys: Vec<String> = (0..5).map(|i| format!("{root}/{i}.bin")).collect();
    for key in &keys {
        s3.upload(key, "application/octet-stream", pattern(32), None)
            .await
            .expect("upload");
    }
    // A key that was never written, because a batch in the wild has them and S3
    // treats deleting an absent key as success.
    let mut with_absent = keys.clone();
    with_absent.push(format!("{root}/never-written.bin"));

    let failures = s3
        .delete_objects(&with_absent)
        .await
        .expect("delete_objects");
    assert!(failures.is_empty(), "delete_objects reported {failures:?}");
    for key in &keys {
        assert!(
            !s3.object_exists(key).await.expect("exists"),
            "{key} survived the batch delete"
        );
    }

    cleanup(&s3, &root).await;
}

#[tokio::test]
#[ignore = "live S3"]
async fn delete_prefix_stops_at_the_prefix() {
    let s3 = client().await;
    let root = prefix("wipe");
    let doomed = format!("{root}/doomed");
    let kept = format!("{root}/kept");
    for i in 0..3 {
        s3.upload(
            &format!("{doomed}/{i}.bin"),
            "application/octet-stream",
            pattern(16),
            None,
        )
        .await
        .expect("upload");
    }
    s3.upload(
        &format!("{kept}/survivor.bin"),
        "application/octet-stream",
        pattern(16),
        None,
    )
    .await
    .expect("upload");

    s3.delete_prefix(&doomed).await.expect("delete_prefix");

    for i in 0..3 {
        assert!(
            !s3.object_exists(&format!("{doomed}/{i}.bin"))
                .await
                .expect("exists"),
            "object {i} survived delete_prefix"
        );
    }
    assert!(
        s3.object_exists(&format!("{kept}/survivor.bin"))
            .await
            .expect("exists"),
        "delete_prefix reached past its prefix, which is how a wipe becomes an incident"
    );

    cleanup(&s3, &root).await;
}

#[tokio::test]
#[ignore = "live S3"]
async fn a_multipart_upload_reads_back_byte_identical() {
    let s3 = client().await;
    let key = prefix("multipart/big.bin");
    // Three parts at the 5 MiB floor, the last one deliberately short: a driver
    // that gets the final part's length wrong passes on an exact multiple.
    let part_size = 5 * 1024 * 1024;
    let body = pattern(part_size * 2 + 1_234_567);

    let file = std::env::temp_dir().join(format!("s3-live-{}.bin", std::process::id()));
    std::fs::write(&file, &body).expect("writing the source file");

    let uploaded = s3
        .upload_multipart(&key, "application/octet-stream", &file, Some(part_size))
        .await;
    std::fs::remove_file(&file).ok();
    uploaded.expect("upload_multipart");

    assert_eq!(
        s3.object_size(&key).await.expect("size"),
        Some(body.len() as i64),
        "the assembled object is the wrong length"
    );
    let (got, _) = s3.download(&key).await.expect("download");
    assert_same_bytes(&body, &got, "multipart round trip");

    cleanup(&s3, &key).await;
}

#[tokio::test]
#[ignore = "live S3"]
async fn the_default_part_size_uploads_a_file_larger_than_one_part() {
    // `part_size: None` takes the 10 MiB default, which no other test exercises
    // and which nothing asserts: the default is written as an arithmetic
    // expression, and a wrong one is either a refusal at the 5 MiB floor or an
    // upload in more parts than intended. 12 MiB is two parts at the real
    // default and one at any smaller one.
    let s3 = client().await;
    let key = prefix("default-part/big.bin");
    let body = pattern(12 * 1024 * 1024);

    let file = std::env::temp_dir().join(format!("s3-live-default-{}.bin", std::process::id()));
    std::fs::write(&file, &body).expect("writing the source file");
    let uploaded = s3
        .upload_multipart(&key, "application/octet-stream", &file, None)
        .await;
    std::fs::remove_file(&file).ok();
    uploaded.expect("upload_multipart with the default part size");

    let (got, _) = s3.download(&key).await.expect("download");
    assert_same_bytes(&body, &got, "default part size round trip");

    cleanup(&s3, &key).await;
}

#[tokio::test]
#[ignore = "live S3"]
async fn a_part_size_below_the_s3_floor_is_refused_before_anything_is_created() {
    // The floor is 5 MiB and the check is `<`, so one byte under is refused and
    // exactly 5 MiB is not. Refused BEFORE the upload is created, which is why
    // this asserts on the pending-upload list as well as on the error: a
    // pre-flight that creates first and validates second strands an upload
    // nobody will ever complete or be billed for noticing.
    let s3 = client().await;
    let key = prefix("floor/rejected.bin");
    let file = std::env::temp_dir().join(format!("s3-live-floor-{}.bin", std::process::id()));
    std::fs::write(&file, pattern(1024)).expect("writing the source file");

    let err = s3
        .upload_multipart(
            &key,
            "application/octet-stream",
            &file,
            Some(5 * 1024 * 1024 - 1),
        )
        .await
        .expect_err("a part below the floor must be refused");
    assert!(
        err.contains("at least 5 MB"),
        "refused for the wrong reason: {err}"
    );
    std::fs::remove_file(&file).ok();

    assert!(
        s3.list_multipart_uploads_for_key(&key)
            .await
            .expect("list")
            .is_empty(),
        "the refusal created an upload and left it pending"
    );
}

#[tokio::test]
#[ignore = "live S3"]
async fn pending_uploads_are_listed_for_their_own_key_only() {
    // The listing filters a bucket-wide response down to one key. A filter that
    // widened would hand a caller somebody else's upload id, and
    // abort_multipart_upload would then cancel an upload that was going fine.
    let s3 = client().await;
    let root = prefix("listing");
    let mine = format!("{root}/mine.bin");
    let theirs = format!("{root}/theirs.bin");

    let mine_id = s3
        .create_multipart_upload(&mine, "application/octet-stream")
        .await
        .expect("create mine");
    let theirs_id = s3
        .create_multipart_upload(&theirs, "application/octet-stream")
        .await
        .expect("create theirs");

    let listed = s3
        .list_multipart_uploads_for_key(&mine)
        .await
        .expect("list mine");
    assert!(
        listed.contains(&mine_id),
        "our own upload is missing: {listed:?}"
    );
    assert!(
        !listed.contains(&theirs_id),
        "the listing reached past its key: {listed:?}"
    );

    s3.abort_multipart_upload(&mine, &mine_id)
        .await
        .expect("abort mine");
    s3.abort_multipart_upload(&theirs, &theirs_id)
        .await
        .expect("abort theirs");
}

#[tokio::test]
#[ignore = "live S3"]
async fn a_multipart_copy_preserves_every_byte() {
    let s3 = client().await;
    let root = prefix("multipart-copy");
    let src = format!("{root}/src.bin");
    let dst = format!("{root}/dst.bin");
    let part_size = 5 * 1024 * 1024;
    let body = pattern(part_size * 2 + 999_983);

    let file = std::env::temp_dir().join(format!("s3-live-copy-{}.bin", std::process::id()));
    std::fs::write(&file, &body).expect("writing the source file");
    let uploaded = s3
        .upload_multipart(&src, "application/octet-stream", &file, Some(part_size))
        .await;
    std::fs::remove_file(&file).ok();
    uploaded.expect("upload_multipart");

    s3.copy_object_multipart(
        s3.bucket(),
        &src,
        &dst,
        "application/octet-stream",
        body.len() as u64,
        Some(part_size),
    )
    .await
    .expect("copy_object_multipart");

    let (got, _) = s3.download(&dst).await.expect("download");
    // THE ASSERTION THIS FILE WAS WRITTEN FOR. A copy driver whose part offsets
    // are off by one part duplicates or drops a range, the object still
    // completes, and nothing but the bytes says so.
    assert_same_bytes(&body, &got, "multipart copy");

    cleanup(&s3, &root).await;
}

#[tokio::test]
#[ignore = "live S3"]
async fn an_aborted_upload_leaves_nothing_behind() {
    let s3 = client().await;
    let key = prefix("abort/pending.bin");

    let upload_id = s3
        .create_multipart_upload(&key, "application/octet-stream")
        .await
        .expect("create_multipart_upload");
    let pending = s3
        .list_multipart_uploads_for_key(&key)
        .await
        .expect("list_multipart_uploads_for_key");
    assert!(
        pending.contains(&upload_id),
        "the upload we just created is not listed as pending: {pending:?}"
    );

    s3.abort_multipart_upload(&key, &upload_id)
        .await
        .expect("abort_multipart_upload");
    let after = s3
        .list_multipart_uploads_for_key(&key)
        .await
        .expect("list after abort");
    assert!(
        !after.contains(&upload_id),
        "the aborted upload is still pending, which is a bill nobody sees: {after:?}"
    );
    assert!(
        !s3.object_exists(&key).await.expect("exists"),
        "an aborted upload produced an object"
    );
}

#[tokio::test]
#[ignore = "live S3"]
async fn a_presigned_url_actually_fetches_and_actually_uploads() {
    let s3 = client().await;
    let root = prefix("presign");
    let download_key = format!("{root}/download.bin");
    let upload_key = format!("{root}/upload.bin");
    let body = pattern(2048);

    s3.upload(
        &download_key,
        "application/octet-stream",
        body.clone(),
        None,
    )
    .await
    .expect("upload");

    let url = s3
        .presign_download(&download_key, 300)
        .await
        .expect("presign_download");
    let fetched = http_get(&url).expect("fetching the presigned URL");
    assert_same_bytes(&body, &fetched, "presigned download");

    let put_url = s3
        .presign_upload(
            &upload_key,
            "application/octet-stream",
            300,
            None,
            Some(body.len() as i64),
        )
        .await
        .expect("presign_upload");
    http_put(&put_url, "application/octet-stream", &body).expect("PUT to the presigned URL");
    let (got, _) = s3
        .download(&upload_key)
        .await
        .expect("download what was PUT");
    assert_same_bytes(&body, &got, "presigned upload");

    cleanup(&s3, &root).await;
}

#[tokio::test]
#[ignore = "live S3"]
async fn connectivity_answers_for_the_configured_bucket() {
    let s3 = client().await;
    s3.check_connectivity().await.expect("check_connectivity");

    let mut wrong = config();
    wrong.bucket = format!("{}-does-not-exist", wrong.bucket);
    let s3 = S3Client::new(&wrong).await.expect("client");
    assert!(
        s3.check_connectivity().await.is_err(),
        "connectivity passed against a bucket that does not exist, so it is not checking the bucket"
    );
}

// ---------------------------------------------------------------------------
// A minimal HTTP client, because a presigned URL is only worth anything if
// something outside this crate can use it.
//
// Raw TCP rather than a dependency: the tier's endpoint is MinIO on localhost
// over plain HTTP, and adding an HTTP stack to dev-dependencies to issue two
// requests would pull a TLS backend into a crate that deliberately pins its own
// (see Cargo.toml on `rustls-ring`). If the tier ever points at an https
// endpoint, these two helpers are what has to change, and they will fail loudly
// rather than quietly skip.
// ---------------------------------------------------------------------------

fn split_url(url: &str) -> (String, String) {
    let rest = url
        .strip_prefix("http://")
        .unwrap_or_else(|| panic!("this helper speaks plain HTTP only, got: {url}"));
    match rest.split_once('/') {
        Some((host, path)) => (host.to_string(), format!("/{path}")),
        None => (rest.to_string(), "/".to_string()),
    }
}

fn read_response(mut stream: TcpStream) -> Result<Vec<u8>, String> {
    let mut raw = Vec::new();
    stream
        .read_to_end(&mut raw)
        .map_err(|e| format!("reading the response: {e}"))?;
    let split = raw
        .windows(4)
        .position(|w| w == b"\r\n\r\n")
        .ok_or_else(|| "no header terminator in the response".to_string())?;
    let headers = String::from_utf8_lossy(&raw[..split]).to_string();
    let status = headers
        .lines()
        .next()
        .unwrap_or_default()
        .split_whitespace()
        .nth(1)
        .unwrap_or_default()
        .to_string();
    if !status.starts_with('2') {
        return Err(format!("HTTP {status}: {headers}"));
    }
    Ok(raw[split + 4..].to_vec())
}

fn http_get(url: &str) -> Result<Vec<u8>, String> {
    let (host, path) = split_url(url);
    let mut stream = TcpStream::connect(&host).map_err(|e| format!("connecting to {host}: {e}"))?;
    let req = format!("GET {path} HTTP/1.1\r\nHost: {host}\r\nConnection: close\r\n\r\n");
    stream
        .write_all(req.as_bytes())
        .map_err(|e| format!("writing the request: {e}"))?;
    read_response(stream)
}

fn http_put(url: &str, content_type: &str, body: &[u8]) -> Result<Vec<u8>, String> {
    let (host, path) = split_url(url);
    let mut stream = TcpStream::connect(&host).map_err(|e| format!("connecting to {host}: {e}"))?;
    let head = format!(
        "PUT {path} HTTP/1.1\r\nHost: {host}\r\nContent-Type: {content_type}\r\n\
         Content-Length: {}\r\nConnection: close\r\n\r\n",
        body.len()
    );
    stream
        .write_all(head.as_bytes())
        .map_err(|e| format!("writing the request head: {e}"))?;
    stream
        .write_all(body)
        .map_err(|e| format!("writing the body: {e}"))?;
    read_response(stream)
}

// ---------------------------------------------------------------------------
// A deliberately unreliable proxy, because a healthy MinIO never fails.
//
// WHY. Three multipart drivers carry a hand-written retry loop -- `attempt < 3`
// and `200 * (1 << ((attempt - 1) * 2))`, three verbatim copies. Every line of
// all three runs only on a transient failure, so the tests above cannot reach
// them however thorough they are, and a mutation run says so: `attempt < 3`
// surviving as `true` is an infinite retry against a permanent failure, and the
// backoff can become almost any expression with nothing to notice (infra
// `8baa89c6`). The only way to observe a retry from outside is to cause one.
//
// HOW. A `TcpListener` on loopback that forwards to the real endpoint, except
// that it closes the first N connections whose request head matches -- no
// response, which the SDK sees as a transport error. `S3_TEST_ENDPOINT` is
// pointed at the proxy for that one test.
//
// Raw TCP again, and for the reason the presign helpers above give: this crate
// pins its own TLS backend and a proxy dependency would drag another one in.
//
// ONE REQUEST PER CONNECTION. The proxy inserts `Connection: close` into the
// forwarded head, so MinIO answers and hangs up rather than keeping the socket
// for the next request. Without it the fault counter would mean "connections"
// while the assertions mean "requests", and connection reuse would decide the
// difference. SigV4 signs a named header list that never includes `Connection`,
// so the insertion does not disturb the signature.
// ---------------------------------------------------------------------------

/// A proxy in front of the object store that fails on purpose.
///
/// It lives for the rest of the test process: the accept loop has no shutdown
/// path, because a test binary that is about to exit does not need one and the
/// alternative is a nonblocking loop that spins.
struct FaultProxy {
    endpoint: String,
    injected: Arc<AtomicU32>,
}

impl FaultProxy {
    /// Forward to the configured endpoint, closing the first `fail_first`
    /// connections whose request head starts with `method` and contains
    /// `needle`. `u32::MAX` fails every one of them, which is the permanent
    /// failure case.
    fn start(method: &'static str, needle: &'static str, fail_first: u32) -> Self {
        let (upstream, _) = split_url(&config().endpoint);
        let listener = TcpListener::bind("127.0.0.1:0").expect("binding the fault proxy");
        let port = listener
            .local_addr()
            .expect("the proxy has an address")
            .port();
        let injected = Arc::new(AtomicU32::new(0));
        let counter = Arc::clone(&injected);

        std::thread::spawn(move || {
            for conn in listener.incoming() {
                let Ok(client) = conn else { continue };
                let upstream = upstream.clone();
                let counter = Arc::clone(&counter);
                std::thread::spawn(move || {
                    proxy_one(client, &upstream, method, needle, fail_first, &counter);
                });
            }
        });

        Self {
            endpoint: format!("http://127.0.0.1:{port}"),
            injected,
        }
    }

    /// How many faults were actually injected. Asserted rather than assumed: a
    /// test that passes because the proxy never matched anything is a test that
    /// proves nothing, and it would look identical to a passing retry.
    fn injected(&self) -> u32 {
        self.injected.load(Ordering::SeqCst)
    }
}

fn proxy_one(
    mut client: TcpStream,
    upstream: &str,
    method: &'static str,
    needle: &'static str,
    fail_first: u32,
    injected: &AtomicU32,
) {
    // Bound both sides. A hung socket here would surface as a test that never
    // finishes, which is worse than one that fails.
    let timeout = Some(Duration::from_secs(30));
    client.set_read_timeout(timeout).ok();
    client.set_write_timeout(timeout).ok();

    // Byte at a time to the header terminator, so the body is left in the
    // socket for the pump below rather than half-read into this buffer. Heads
    // are a kilobyte or so; this is a test.
    let mut head = Vec::new();
    let mut byte = [0u8; 1];
    while head.len() < 64 * 1024 {
        match client.read(&mut byte) {
            Ok(0) | Err(_) => return,
            Ok(_) => head.push(byte[0]),
        }
        if head.ends_with(b"\r\n\r\n") {
            break;
        }
    }
    let head = String::from_utf8_lossy(&head).to_string();

    if head.starts_with(method) && head.contains(needle) {
        let claimed = injected
            .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |n| {
                (n < fail_first).then_some(n + 1)
            })
            .is_ok();
        if claimed {
            // Hang up without answering.
            client.shutdown(Shutdown::Both).ok();
            return;
        }
    }

    let Ok(mut server) = TcpStream::connect(upstream) else {
        return;
    };
    server.set_read_timeout(timeout).ok();
    server.set_write_timeout(timeout).ok();

    let (request_line, rest) = head.split_once("\r\n").unwrap_or((head.as_str(), ""));
    let forwarded = format!("{request_line}\r\nConnection: close\r\n{rest}");
    if server.write_all(forwarded.as_bytes()).is_err() {
        return;
    }

    let mut from_server = server.try_clone().expect("cloning the upstream socket");
    let mut to_client = client.try_clone().expect("cloning the client socket");
    let back = std::thread::spawn(move || {
        std::io::copy(&mut from_server, &mut to_client).ok();
        to_client.shutdown(Shutdown::Write).ok();
    });
    std::io::copy(&mut client, &mut server).ok();
    server.shutdown(Shutdown::Write).ok();
    back.join().ok();
}

// ---------------------------------------------------------------------------
// Reading the backoff off the crate's own warning, because the wall clock
// cannot see it.
//
// A lower bound on elapsed time proves a sleep happened. It does NOT prove the
// sleep was the right length, and that was measured rather than assumed: with
// `200 * (1 << ((attempt - 1) * 2))` replaced by a flat `200`, an upload that
// should have cost 200ms + 800ms of backoff still finished inside the one-second
// bound and the test passed. The SDK runs its own retry policy under ours and
// its backoff is close to a second per attempt, so it swamps the difference
// between 200ms and 800ms in any whole-operation timing.
//
// So the delay expression is read where it is unambiguous: each loop logs
// `delay_ms` on the warning it emits before sleeping. A mutant that changes the
// arithmetic changes that number. Pairing the two -- the field for the value,
// the clock for the fact that a sleep occurred -- covers both halves, and
// neither covers both alone.
// ---------------------------------------------------------------------------

/// Every `delay_ms` the crate has logged in this process, in order.
fn delays() -> &'static Mutex<Vec<u64>> {
    static DELAYS: OnceLock<Mutex<Vec<u64>>> = OnceLock::new();
    static INSTALLED: OnceLock<()> = OnceLock::new();
    let cell = DELAYS.get_or_init(|| Mutex::new(Vec::new()));
    INSTALLED.get_or_init(|| {
        // Global, and that is sound here because nextest runs one test per
        // process -- which is how this file is documented to be run. Under a
        // shared-process `cargo test` two concurrent retry tests would append
        // to one list, so each test reads only the tail it appended.
        tracing::subscriber::set_global_default(DelayCollector)
            .expect("no other subscriber should be installed in a test process");
    });
    cell
}

struct DelayCollector;

impl tracing::Subscriber for DelayCollector {
    fn enabled(&self, _: &tracing::Metadata<'_>) -> bool {
        true
    }
    fn new_span(&self, _: &tracing::span::Attributes<'_>) -> tracing::Id {
        tracing::Id::from_u64(1)
    }
    fn record(&self, _: &tracing::Id, _: &tracing::span::Record<'_>) {}
    fn record_follows_from(&self, _: &tracing::Id, _: &tracing::Id) {}
    fn event(&self, event: &tracing::Event<'_>) {
        struct Pick(Option<u64>);
        impl tracing::field::Visit for Pick {
            fn record_u64(&mut self, field: &tracing::field::Field, value: u64) {
                if field.name() == "delay_ms" {
                    self.0 = Some(value);
                }
            }
            fn record_debug(&mut self, _: &tracing::field::Field, _: &dyn std::fmt::Debug) {}
        }
        let mut pick = Pick(None);
        event.record(&mut pick);
        if let Some(ms) = pick.0 {
            delays()
                .lock()
                .expect("the delay list is not poisoned")
                .push(ms);
        }
    }
    fn enter(&self, _: &tracing::Id) {}
    fn exit(&self, _: &tracing::Id) {}
}

/// The delays logged since `from`, which is where the caller's own work started.
fn delays_since(from: usize) -> Vec<u64> {
    delays().lock().expect("the delay list is not poisoned")[from..].to_vec()
}

fn delays_so_far() -> usize {
    delays()
        .lock()
        .expect("the delay list is not poisoned")
        .len()
}

/// The same client the other tests build, pointed at a proxy.
async fn client_via(proxy: &FaultProxy) -> S3Client {
    let mut config = config();
    config.endpoint.clone_from(&proxy.endpoint);
    S3Client::new(&config)
        .await
        .expect("building the client should not need the network")
}

/// Two parts and a short tail, small enough that the parts are not the point:
/// what is being exercised is the completion that follows them.
async fn upload_through(proxy: &FaultProxy, key: &str) -> (Result<(), String>, Vec<u8>, Duration) {
    let s3 = client_via(proxy).await;
    let part_size = 5 * 1024 * 1024;
    let body = pattern(part_size + 1_000);
    let file =
        std::env::temp_dir().join(format!("s3-fault-{}-{}.bin", std::process::id(), key.len()));
    std::fs::write(&file, &body).expect("writing the source file");

    // The timeout is load-bearing, not defensive. `attempt < 3` mutated to
    // `true` is a loop that never ends, and an assertion placed after the call
    // never runs: the test hangs instead of failing, which reads as a mutant
    // that survived. A bound turns the hang into a red test.
    let started = Instant::now();
    let result = tokio::time::timeout(
        Duration::from_mins(2),
        s3.upload_multipart(key, "application/octet-stream", &file, Some(part_size)),
    )
    .await
    .unwrap_or_else(|_| Err("upload_multipart never returned".to_string()));
    let elapsed = started.elapsed();
    std::fs::remove_file(&file).ok();
    (result, body, elapsed)
}

/// The completion is POST with an `uploadId`; nothing else in a multipart
/// upload is. `CreateMultipartUpload` is a POST too, but carries `uploads`
/// rather than `uploadId=`.
const COMPLETE: (&str, &str) = ("POST", "uploadId=");

/// A part, uploaded or copied: both are a PUT carrying a part number.
const PART: (&str, &str) = ("PUT", "partNumber=");

// How many closed connections it takes to spend one of the crate's three
// attempts. Not one: the SDK runs its own retry policy underneath, so a single
// closed connection is absorbed before the crate's loop ever sees an error.
// These are MEASURED against MinIO on astra rather than derived from the SDK's
// defaults, because the number that matters is what the two policies do
// together, and a derived number would silently rot when either changes.
const PER_ATTEMPT: u32 = 3;
const FAIL_ONE_ATTEMPT: u32 = PER_ATTEMPT;
const FAIL_TWO_ATTEMPTS: u32 = PER_ATTEMPT * 2;

#[tokio::test]
#[ignore = "live S3"]
async fn a_retried_completion_still_writes_the_right_bytes() {
    let proxy = FaultProxy::start(COMPLETE.0, COMPLETE.1, FAIL_ONE_ATTEMPT);
    let key = prefix("fault/retried.bin");
    let mark = delays_so_far();
    let (result, body, elapsed) = upload_through(&proxy, &key).await;
    result.expect("the upload should survive one failed completion");

    assert!(
        proxy.injected() > 0,
        "the proxy never matched a completion, so nothing was retried and this test proved nothing"
    );
    // One failed attempt, one backoff, and it is the first rung: 200ms.
    assert_eq!(
        delays_since(mark),
        vec![200],
        "the completion loop backed off with the wrong delays"
    );
    assert!(
        elapsed >= Duration::from_millis(200),
        "returned in {elapsed:?}, so the delay was logged and not slept"
    );

    // The retry is only interesting if the object is right afterwards. A driver
    // that retried and assembled the wrong parts would pass a bare `is_ok`.
    let s3 = client().await;
    let (got, _) = s3.download(&key).await.expect("download");
    assert_same_bytes(&body, &got, "the object written across a retry");

    cleanup(&s3, &key).await;
}

#[tokio::test]
#[ignore = "live S3"]
async fn two_failures_walk_up_the_backoff() {
    let proxy = FaultProxy::start(COMPLETE.0, COMPLETE.1, FAIL_TWO_ATTEMPTS);
    let key = prefix("fault/twice.bin");
    let mark = delays_so_far();
    let (result, body, elapsed) = upload_through(&proxy, &key).await;
    result.expect("the upload should survive two failed completions");

    // THE ASSERTION THE BACKOFF ARITHMETIC EXISTS FOR. `200 * (1 << ((attempt
    // - 1) * 2))` is 200 then 800; a flat 200, a doubling, or an off-by-one on
    // the exponent all produce a different second number.
    assert_eq!(
        delays_since(mark),
        vec![200, 800],
        "the completion loop backed off with the wrong delays"
    );
    assert!(
        elapsed >= Duration::from_secs(1),
        "returned in {elapsed:?}, faster than the 200ms + 800ms it says it slept"
    );

    let s3 = client().await;
    let (got, _) = s3.download(&key).await.expect("download");
    assert_same_bytes(&body, &got, "the object written across two retries");
    cleanup(&s3, &key).await;
}

#[tokio::test]
#[ignore = "live S3"]
async fn a_permanent_failure_gives_up_rather_than_looping() {
    let proxy = FaultProxy::start(COMPLETE.0, COMPLETE.1, u32::MAX);
    let key = prefix("fault/permanent.bin");
    let mark = delays_so_far();
    let (result, _, elapsed) = upload_through(&proxy, &key).await;

    let error = result.expect_err("a completion that never succeeds must not report success");
    assert!(
        error.contains("after retries"),
        "gave up with the wrong error, which suggests it left the retry loop by another path \
         -- or never left it at all, which is what the timeout reports: {error}"
    );

    // Three attempts means two backoffs and then a decision, not a third sleep.
    // This is the `attempt < 3` guard read as a number: a loop that ran once
    // more would log a third delay.
    assert_eq!(
        delays_since(mark),
        vec![200, 800],
        "gave up after the wrong number of attempts"
    );

    // The `attempt < 3` -> `true` mutant is an infinite loop, and the only way
    // to fail a test on an infinite loop is a wall clock. Measured: the mutant
    // makes this test fail on `upload_multipart never returned`.
    assert!(
        elapsed < Duration::from_mins(1),
        "took {elapsed:?} to give up, which is the shape of a retry loop that does not stop"
    );

    let s3 = client().await;
    cleanup(&s3, &key).await;
}

#[tokio::test]
#[ignore = "live S3"]
async fn a_retried_part_upload_still_writes_the_right_bytes() {
    // The second of the three loops, in `run_multipart_upload`. A part is a PUT
    // carrying a part number; the completion that follows is a POST, so this
    // proxy leaves it alone.
    let proxy = FaultProxy::start(PART.0, PART.1, FAIL_ONE_ATTEMPT);
    let key = prefix("fault/part.bin");
    let mark = delays_so_far();
    let (result, body, elapsed) = upload_through(&proxy, &key).await;
    result.expect("the upload should survive one failed part");

    assert!(proxy.injected() > 0, "no part upload was ever failed");
    assert_eq!(
        delays_since(mark),
        vec![200],
        "the part loop backed off with the wrong delays"
    );
    assert!(
        elapsed >= Duration::from_millis(200),
        "returned in {elapsed:?}"
    );

    // A part driver that retried by re-sending the wrong slice of the file
    // completes an object of the right length and the wrong contents.
    let s3 = client().await;
    let (got, _) = s3.download(&key).await.expect("download");
    assert_same_bytes(&body, &got, "the object written across a failed part");
    cleanup(&s3, &key).await;
}

#[tokio::test]
#[ignore = "live S3"]
async fn a_retried_copy_part_still_copies_every_byte() {
    // The third loop, in `run_multipart_copy`. Same shape as a part upload from
    // the wire's point of view, which is why one needle reaches both: what
    // separates them is which driver the test drives.
    let s3 = client().await;
    let root = prefix("fault/copy");
    let src = format!("{root}/src.bin");
    let dst = format!("{root}/dst.bin");
    let part_size = 5 * 1024 * 1024;
    let body = pattern(part_size + 1_000);

    let file = std::env::temp_dir().join(format!("s3-fault-copy-{}.bin", std::process::id()));
    std::fs::write(&file, &body).expect("writing the source file");
    let uploaded = s3
        .upload_multipart(&src, "application/octet-stream", &file, Some(part_size))
        .await;
    std::fs::remove_file(&file).ok();
    uploaded.expect("the source upload runs against the real endpoint");

    let proxy = FaultProxy::start(PART.0, PART.1, FAIL_ONE_ATTEMPT);
    let faulty = client_via(&proxy).await;
    let mark = delays_so_far();
    let started = Instant::now();
    let copied = tokio::time::timeout(
        Duration::from_mins(2),
        faulty.copy_object_multipart(
            s3.bucket(),
            &src,
            &dst,
            "application/octet-stream",
            body.len() as u64,
            Some(part_size),
        ),
    )
    .await
    .unwrap_or_else(|_| Err("copy_object_multipart never returned".to_string()));
    let elapsed = started.elapsed();
    copied.expect("the copy should survive one failed part");

    assert!(proxy.injected() > 0, "no copy part was ever failed");
    assert_eq!(
        delays_since(mark),
        vec![200],
        "the copy loop backed off with the wrong delays"
    );
    assert!(
        elapsed >= Duration::from_millis(200),
        "returned in {elapsed:?}"
    );

    let (got, _) = s3.download(&dst).await.expect("download");
    // The offset arithmetic is what a retried copy part can get wrong, and only
    // the bytes say so: the object completes either way.
    assert_same_bytes(&body, &got, "the object copied across a failed part");
    cleanup(&s3, &root).await;
}

#[tokio::test]
#[ignore = "live S3"]
async fn a_part_that_never_uploads_gives_up_rather_than_looping() {
    // The part loop's own `attempt < 3`. Its completion counterpart above
    // cannot reach this one: a part that fails forever never gets to a
    // completion.
    let proxy = FaultProxy::start(PART.0, PART.1, u32::MAX);
    let key = prefix("fault/part-permanent.bin");
    let mark = delays_so_far();
    let (result, _, elapsed) = upload_through(&proxy, &key).await;

    let error = result.expect_err("a part that never uploads must not report success");
    assert!(
        error.contains("after retries"),
        "gave up with the wrong error, or never gave up at all: {error}"
    );
    assert_eq!(
        delays_since(mark),
        vec![200, 800],
        "the part loop gave up after the wrong number of attempts"
    );
    assert!(
        elapsed < Duration::from_mins(1),
        "took {elapsed:?} to give up, which is the shape of a retry loop that does not stop"
    );

    let s3 = client().await;
    cleanup(&s3, &key).await;
}

#[tokio::test]
#[ignore = "live S3"]
async fn a_copy_part_that_never_lands_gives_up_rather_than_looping() {
    let s3 = client().await;
    let root = prefix("fault/copy-permanent");
    let src = format!("{root}/src.bin");
    let dst = format!("{root}/dst.bin");
    let part_size = 5 * 1024 * 1024;
    let body = pattern(part_size + 1_000);

    let file = std::env::temp_dir().join(format!("s3-fault-copyp-{}.bin", std::process::id()));
    std::fs::write(&file, &body).expect("writing the source file");
    let uploaded = s3
        .upload_multipart(&src, "application/octet-stream", &file, Some(part_size))
        .await;
    std::fs::remove_file(&file).ok();
    uploaded.expect("the source upload runs against the real endpoint");

    let proxy = FaultProxy::start(PART.0, PART.1, u32::MAX);
    let faulty = client_via(&proxy).await;
    let mark = delays_so_far();
    let started = Instant::now();
    let copied = tokio::time::timeout(
        Duration::from_mins(2),
        faulty.copy_object_multipart(
            s3.bucket(),
            &src,
            &dst,
            "application/octet-stream",
            body.len() as u64,
            Some(part_size),
        ),
    )
    .await
    .unwrap_or_else(|_| Err("copy_object_multipart never returned".to_string()));
    let elapsed = started.elapsed();

    let error = copied.expect_err("a copy part that never lands must not report success");
    assert!(
        error.contains("after retries"),
        "gave up with the wrong error, or never gave up at all: {error}"
    );
    assert_eq!(
        delays_since(mark),
        vec![200, 800],
        "the copy loop gave up after the wrong number of attempts"
    );
    assert!(
        elapsed < Duration::from_mins(1),
        "took {elapsed:?} to give up, which is the shape of a retry loop that does not stop"
    );

    cleanup(&s3, &root).await;
}

/// Named so `cargo nextest list` shows why the file looks empty on a laptop.
#[test]
#[ignore = "live S3"]
fn these_tests_need_a_live_object_store() {
    println!("{SKIP}");
}