ytsaurus-client 0.2.5

//! What the client actually puts on the wire.
//!
//! Everything else in this crate is checked against a cluster, which answers
//! the same whether or not the request was well made. These serve the request
//! from a socket in-process and read the bytes the client sent, which is the
//! only way to pin the things a cluster is too forgiving to notice:
//! compression the client asks for, the token it carries, and the header the
//! parameters travel in.
//!
//! The last section is a second question — not what a request looked like but
//! *which address* it went to, which no single listener can answer. It uses
//! two, and [`Proxy`] rather than [`capture`].

use std::io::{BufRead, BufReader, Read, Write};
use std::net::{TcpListener, TcpStream};

use ytsaurus_client::{
    Client, ClientError, DataFormat, Key, Method, OperationFilter, OperationParameters,
    RetryPolicy, RowRange, SkiffFormat, SkiffSchema, SkiffSchemaRef, SkiffWireType, TablePath,
    TraceContext, yson_build,
};

/// Serves exactly one request and returns its headers as text.
///
/// The reply is a valid `exists` answer, so the client finishes normally and
/// nothing retries into a second connection this listener would never accept.
fn capture(request_from: impl FnOnce(&str)) -> String {
    let listener = TcpListener::bind("127.0.0.1:0").expect("binds");
    let address = listener.local_addr().expect("has an address");

    let served = std::thread::spawn(move || {
        let (mut stream, _) = listener.accept().expect("accepts");
        let mut reader = BufReader::new(stream.try_clone().expect("clones"));

        let mut head = String::new();
        loop {
            let mut line = String::new();
            match reader.read_line(&mut line) {
                Ok(0) => break,
                Ok(_) if line == "\r\n" => break,
                Ok(_) => head.push_str(&line),
                Err(_) => break,
            }
        }

        // A command with a body — a table write — is only finished sending
        // when its body is read. Replying before then leaves the client writing
        // into a socket nobody is reading, which surfaces as a broken pipe
        // rather than as the request under test.
        if let Some(length) = content_length(&head) {
            let mut body = vec![0_u8; length];
            reader.read_exact(&mut body).ok();
        } else if head.to_lowercase().contains("transfer-encoding: chunked") {
            // The row-by-row writers stream, so their length is only known
            // when the terminating chunk arrives.
            drain_chunked(&mut reader);
        }

        let body = br#"{"value"=%true}"#;
        let mut reply = format!(
            "HTTP/1.1 200 OK\r\nContent-Length: {}\r\nContent-Type: application/x-yt-yson-text\r\n\r\n",
            body.len()
        )
        .into_bytes();
        reply.extend_from_slice(body);
        stream.write_all(&reply).expect("replies");
        stream.flush().ok();

        head
    });

    request_from(&format!("http://{address}"));
    served.join().expect("the listener thread finished")
}

/// Consumes a chunked body up to its terminating zero-length chunk.
fn drain_chunked(reader: &mut BufReader<std::net::TcpStream>) {
    loop {
        let mut header = String::new();
        if reader.read_line(&mut header).is_err() {
            return;
        }
        let size = usize::from_str_radix(header.trim(), 16).unwrap_or(0);
        if size == 0 {
            let mut trailer = String::new();
            reader.read_line(&mut trailer).ok();
            return;
        }
        let mut chunk = vec![0_u8; size + 2]; // the chunk and its CRLF
        if reader.read_exact(&mut chunk).is_err() {
            return;
        }
    }
}

/// The declared body length of a request, if it declared one.
fn content_length(head: &str) -> Option<usize> {
    head.lines()
        .find(|line| line.to_lowercase().starts_with("content-length:"))
        .and_then(|line| line.split(':').nth(1))
        .and_then(|value| value.trim().parse().ok())
}

/// One header of a captured request, exactly as it was sent.
///
/// The name is matched case-insensitively, because HTTP says a header name is;
/// the value comes back untouched, because some of them are case-sensitive and
/// lowercasing the whole request head — which the tests here otherwise do —
/// would hide that.
fn header_value(head: &str, name: &str) -> Option<String> {
    head.lines()
        .find(|line| {
            line.to_lowercase()
                .starts_with(&format!("{}:", name.to_lowercase()))
        })
        .map(|line| line[line.find(':').unwrap_or(0) + 1..].trim().to_owned())
}

/// The `X-YT-Parameters` header of a captured request.
fn parameters(head: &str) -> String {
    head.lines()
        .find(|line| line.to_lowercase().starts_with("x-yt-parameters:"))
        .map(|line| line[line.find(':').unwrap_or(0) + 1..].trim().to_owned())
        .unwrap_or_default()
}

#[test]
fn a_plain_write_replaces_and_says_nothing_about_it() {
    // The shape every version of this crate has sent. A path that grew
    // `<append=%false>` would be a new request for an unchanged meaning, and
    // the day it changed nobody would know which release did it.
    let head = capture(|proxy| {
        let client = Client::new(proxy).with_retries(RetryPolicy::none());
        client.write_table("//tmp/out", b"").expect("writes");
    });

    assert!(
        parameters(&head).contains(r#"path="//tmp/out""#),
        "the path is not a bare string:\n{head}"
    );
    assert!(
        !parameters(&head).contains("append"),
        "a replacing write mentioned append:\n{head}"
    );
}

#[test]
fn an_appending_write_carries_the_attribute_on_the_path() {
    // The attribute goes on the *path*, not beside it as a parameter of its
    // own. A cluster given `{path="//tmp/out";append=%true}` replaces the
    // table and reports success, which is the failure this pins.
    let head = capture(|proxy| {
        let client = Client::new(proxy).with_retries(RetryPolicy::none());
        client
            .write_table(TablePath::new("//tmp/out").append(), b"")
            .expect("writes");
    });

    assert!(
        parameters(&head).contains(r#"path=<append=%true>"//tmp/out""#),
        "the path does not carry the attribute:\n{head}"
    );
}

#[test]
fn all_three_writers_can_append() {
    // `write_table`, `write_table_rows` and `write_table_streaming` build the
    // same parameter block three times over, and only one of them is exercised
    // by an example. A path that lost its attribute on the streaming route
    // would replace a table the caller meant to add to, and the caller would
    // find out by losing rows.
    let row = std::collections::BTreeMap::from([("n", 1_i64)]);

    let heads = [
        (
            "write_table",
            capture(|proxy| {
                let client = Client::new(proxy).with_retries(RetryPolicy::none());
                client
                    .write_table(TablePath::new("//tmp/out").append(), b"")
                    .expect("writes");
            }),
        ),
        (
            "write_table_rows",
            capture(|proxy| {
                let client = Client::new(proxy).with_retries(RetryPolicy::none());
                client
                    .write_table_rows(TablePath::new("//tmp/out").append(), [row])
                    .expect("writes");
            }),
        ),
        (
            "write_table_streaming",
            capture(|proxy| {
                let client = Client::new(proxy).with_retries(RetryPolicy::none());
                client
                    .write_table_streaming(
                        TablePath::new("//tmp/out").append(),
                        std::io::Cursor::new(Vec::new()),
                    )
                    .expect("writes");
            }),
        ),
    ];

    for (command, head) in heads {
        assert!(
            parameters(&head).contains(r#"path=<append=%true>"//tmp/out""#),
            "{command} sent a path without the attribute:\n{head}"
        );
    }
}

// The reads below ignore their result on purpose: the stub's reply is not a
// binary YSON fragment, so the read errors *after* the request — the thing
// under test — is already on the wire.

#[test]
fn a_column_selection_travels_as_an_attribute_on_the_path() {
    // Same trap as append, read side: `columns` is an attribute ON the path —
    // the rich YPath reference lists it as recognized by `read_table` — and a
    // sibling parameter would be silently dropped.
    let head = capture(|proxy| {
        let client = Client::new(proxy).with_retries(RetryPolicy::none());
        let _ = client.read_table(TablePath::new("//tmp/wide").columns(["host", "status"]));
    });

    assert!(
        head.starts_with("GET /api/v4/read_table"),
        "not a read_table:\n{head}"
    );
    assert!(
        parameters(&head).contains(r#"path=<columns=[host;status]>"//tmp/wide""#),
        "the selection is not on the path:\n{head}"
    );
}

#[test]
fn a_row_range_travels_in_the_documented_limits() {
    // `0..2` is rows 0 and 1 on both sides of the wire: the reference says
    // every limit but key_bound is inclusive below and exclusive above, which
    // is exactly Rust's `..`.
    let head = capture(|proxy| {
        let client = Client::new(proxy).with_retries(RetryPolicy::none());
        let _ = client.read_table(TablePath::new("//tmp/t").range(0..2));
    });

    assert!(
        parameters(&head).contains(
            r#"path=<ranges=[{lower_limit={row_index=0};upper_limit={row_index=2}}]>"//tmp/t""#
        ),
        "the range is not on the path:\n{head}"
    );
}

#[test]
fn key_bounds_travel_in_the_clusters_representation() {
    // The two bounds `key` says natively go as `key`; the two it cannot go as
    // `key_bound=[relation;prefix]`, with the only relation the reference
    // allows on that side. Every literal is fixed, so asserting the rendered
    // text is safe.
    let plain = capture(|proxy| {
        let client = Client::new(proxy).with_retries(RetryPolicy::none());
        let _ = client.read_table(
            TablePath::new("//tmp/sorted")
                .range(RowRange::keys(Key::from("alice")..Key::from("bob"))),
        );
    });
    assert!(
        parameters(&plain).contains(
            r#"path=<ranges=[{lower_limit={key=[alice]};upper_limit={key=[bob]}}]>"//tmp/sorted""#
        ),
        "an inclusive..exclusive key range is not the plain key form:\n{plain}"
    );

    let bounds = capture(|proxy| {
        let client = Client::new(proxy).with_retries(RetryPolicy::none());
        let _ = client.read_table(TablePath::new("//tmp/sorted").range(RowRange::keys((
            std::ops::Bound::Excluded(Key::from("alice")),
            std::ops::Bound::Included(Key::from("bob")),
        ))));
    });
    assert!(
        parameters(&bounds).contains(
            r#"path=<ranges=[{lower_limit={key_bound=[">";[alice]]};upper_limit={key_bound=["<=";[bob]]}}]>"//tmp/sorted""#
        ),
        "the other two inclusivities are not key_bound:\n{bounds}"
    );
}

#[test]
fn an_exact_key_read_asks_in_the_clusters_word_for_it() {
    let head = capture(|proxy| {
        let client = Client::new(proxy).with_retries(RetryPolicy::none());
        let _ = client.read_table(
            TablePath::new("//tmp/sorted").range(RowRange::exact_key(Key::from("alice"))),
        );
    });

    assert!(
        parameters(&head).contains(r#"path=<ranges=[{exact={key=[alice]}}]>"//tmp/sorted""#),
        "the exact selector is not on the path:\n{head}"
    );
}

#[test]
fn all_three_readers_carry_the_selection() {
    // `read_table`, `read_table_rows` and `read_table_streaming` build their
    // parameter block separately; a selection lost on one route would quietly
    // read the whole table — the cost the selection exists to avoid.
    let selected = || TablePath::new("//tmp/wide").columns(["host"]).range(0..100);
    let expected = r#"path=<columns=[host];ranges=[{lower_limit={row_index=0};upper_limit={row_index=100}}]>"//tmp/wide""#;

    let heads = [
        (
            "read_table",
            capture(|proxy| {
                let client = Client::new(proxy).with_retries(RetryPolicy::none());
                let _ = client.read_table(selected());
            }),
        ),
        (
            "read_table_rows",
            capture(|proxy| {
                let client = Client::new(proxy).with_retries(RetryPolicy::none());
                let _ =
                    client.read_table_rows::<std::collections::BTreeMap<String, i64>>(selected());
            }),
        ),
        (
            "read_table_streaming",
            capture(|proxy| {
                let client = Client::new(proxy).with_retries(RetryPolicy::none());
                let _ = client.read_table_streaming(selected());
            }),
        ),
    ];

    for (command, head) in heads {
        assert!(
            parameters(&head).contains(expected),
            "{command} sent a path without the selection:\n{head}"
        );
    }
}

#[test]
fn a_skiff_read_merges_a_row_range_with_its_schema_columns() {
    // The Skiff format's fields are the column selection; a typed range says
    // which rows. Different questions, one path — both attributes must
    // arrive, or the read silently covers the wrong slice of the table.
    let head = capture(|proxy| {
        let client = Client::new(proxy).with_retries(RetryPolicy::none());
        let _ = client.read_skiff_table(TablePath::new("//tmp/t").range(0..2), &one_column());
    });

    // The whole rendering rather than two independent substrings: the claim
    // is that both attributes sit on the *path*, and a pair of `contains`
    // would still pass if `columns` moved off it to a sibling parameter —
    // which is precisely the trap `<append=%true>` taught. Attribute order is
    // the encoder's BTreeMap, so `columns` before `ranges` is byte order, not
    // insertion luck.
    assert!(
        parameters(&head).contains(
            r#"path=<columns=[n];ranges=[{lower_limit={row_index=0};upper_limit={row_index=2}}]>"//tmp/t""#
        ),
        "the Skiff read lost half the selection, or moved it off the path:\n{head}"
    );
}

#[test]
fn a_skiff_read_refuses_a_column_selection_spelled_into_the_string() {
    // The "one spelling of a selection" rule, in the one place it can be
    // broken with nothing typed at all: a Skiff read synthesises a `columns`
    // attribute out of the format's fields, so `//tmp/t{n}` arrives as two
    // column selections on one path. Measured in the wire shape this client
    // sends — attributes hung outside a YSON string node — the synthesised
    // attribute wins: `<columns=[n]>"//tmp/t{k}"` came back as column `n`, so
    // the tuple stays aligned with its schema and nothing decodes wrong. What
    // is lost is the caller's own `{n}`, discarded at 200 without a word. A
    // leading `<…>` is refused one step removed: the cluster takes it, but
    // this client cannot parse the block to know whether it names `columns`.
    let client = Client::new(&nowhere()).with_retries(RetryPolicy::none());

    for path in [
        "//tmp/t{n}",
        "<columns=[n]>//tmp/t",
        "<primary_medium=default>//tmp/t",
    ] {
        let error = client
            .read_skiff_table(path, &one_column())
            .expect_err("refused");
        assert!(
            matches!(&error, ClientError::Config(_)),
            "{path} was not refused locally: {error}"
        );
    }
}

#[test]
fn a_skiff_read_sends_a_string_spelled_row_range() {
    // The refusal above exists to stop a columns-versus-columns conflict, and
    // a row range is not one. Measured, `<columns=["n"]>"//tmp/t[#3:#5]"`
    // answered 200 with rows 3-4 carrying only `n`: ranges select rows, the
    // synthesised attribute selects columns, and they compose — neither is
    // discarded. `read_table` has always taken the string spelling, and a
    // Skiff read takes it too.
    let head = capture(|proxy| {
        let client = Client::new(proxy).with_retries(RetryPolicy::none());
        let _ = client.read_skiff_table("//tmp/t[#0:#2]", &one_column());
    });

    assert!(
        parameters(&head).contains(r#"path=<columns=[n]>"//tmp/t[#0:#2]""#),
        "the string-spelled range did not reach the cluster intact:\n{head}"
    );
}

#[test]
fn a_skiff_read_refuses_a_second_column_selection() {
    // Two projections — the schema's fields and TablePath::columns — with one
    // positional wire format between them. Whichever the cluster picked, the
    // decoder would disagree with it; refused before anything is sent.
    let client = Client::new(&nowhere()).with_retries(RetryPolicy::none());
    let error = client
        .read_skiff_table(TablePath::new("//tmp/t").columns(["n"]), &one_column())
        .expect_err("refused");

    assert!(
        matches!(&error, ClientError::Config(reason) if reason.contains("columns")),
        "not the local refusal: {error}"
    );
}

#[test]
fn a_write_with_a_read_selection_is_refused_before_anything_is_sent() {
    // The decision this crate makes about columns and ranges on a write:
    // refuse locally. The cluster's answer is to ignore them and replace the
    // whole table with a 200 — measured, and recorded in AGENTS.md — so
    // sending them is silent data loss with nicer syntax. The client points
    // at an address that answers nothing: had a request been attempted, the
    // error would be Transport, not Config.
    let client = Client::new(&nowhere()).with_retries(RetryPolicy::none());
    let row = std::collections::BTreeMap::from([("n", 1_i64)]);

    let refusals: Vec<(&str, ClientError)> = vec![
        (
            "write_table with columns",
            client
                .write_table(TablePath::new("//tmp/out").columns(["a"]), b"")
                .expect_err("refused"),
        ),
        (
            "write_table with a range",
            client
                .write_table(TablePath::new("//tmp/out").range(0..2), b"")
                .expect_err("refused"),
        ),
        (
            "write_table_rows with a range",
            client
                .write_table_rows(TablePath::new("//tmp/out").range(0..2), [row])
                .expect_err("refused"),
        ),
        (
            "write_table_streaming with columns",
            client
                .write_table_streaming(
                    TablePath::new("//tmp/out").columns(["a"]),
                    std::io::Cursor::new(Vec::new()),
                )
                .expect_err("refused"),
        ),
        (
            "write_skiff_table with a range",
            client
                .write_skiff_table(TablePath::new("//tmp/out").range(0..2), b"", &one_column())
                .expect_err("refused"),
        ),
    ];

    for (writer, error) in refusals {
        assert!(
            matches!(&error, ClientError::Config(_)),
            "{writer} did not refuse locally: {error}"
        );
    }
}

#[test]
fn a_write_path_string_spelling_a_selection_is_refused() {
    // The measured trap, verbatim: write_table_rows("//tmp/t[#0:#2]", rows)
    // replaced the whole table and returned success. The string is never
    // parsed, so the only write this client will send is one whose string is
    // a bare path.
    let client = Client::new(&nowhere()).with_retries(RetryPolicy::none());

    for path in ["//tmp/t[#0:#2]", "//tmp/t{a,b}", "<append=%true>//tmp/t"] {
        let error = client.write_table(path, b"").expect_err("refused");
        assert!(
            matches!(&error, ClientError::Config(_)),
            "{path} was not refused locally: {error}"
        );
    }
}

#[test]
fn an_escaped_bracket_is_a_node_name_and_still_writable() {
    // Rich YPath escapes a literal bracket as `\[`, so this is a table named
    // `t[x]`, not a range — refusing it would make that table unwritable.
    let head = capture(|proxy| {
        let client = Client::new(proxy).with_retries(RetryPolicy::none());
        client.write_table(r"//tmp/t\[x\]", b"").expect("writes");
    });

    assert!(
        head.starts_with("PUT /api/v4/write_table"),
        "the write was not sent:\n{head}"
    );
}

#[test]
fn a_read_keeps_passing_a_string_spelled_path_through() {
    // Reads honoured string-spelled ranges before TablePath modelled them,
    // and the cluster reads them correctly; refusing now would break working
    // code to protect it from nothing. The path travels as a bare string,
    // syntax and all — not parsed, not rewritten.
    let head = capture(|proxy| {
        let client = Client::new(proxy).with_retries(RetryPolicy::none());
        let _ = client.read_table("//tmp/t[#0:#2]");
    });

    assert!(
        parameters(&head).contains(r#"path="//tmp/t[#0:#2]""#),
        "the string-spelled path was rewritten:\n{head}"
    );
}

#[test]
fn a_read_refuses_a_selection_spelled_twice() {
    // A string that already carries `[…]` plus a typed range is the *same
    // kind* of selection spelled twice on one path. Measured, the attribute
    // this client hangs on the path wins and the string's half is discarded at
    // 200 with nothing said — the rows that come back are the ones the typed
    // range names, and the range the caller wrote into the string simply never
    // happens.
    //
    // Every reader has to refuse it, not just the buffered one: each builds
    // its parameter block separately, and the streaming reader is where a
    // doubled selection costs the most — it exists because the table is too
    // big to hold, so reading the wrong slice of it is the expensive mistake.
    // The address answers nothing, so a request that *was* attempted would
    // come back Transport rather than Config.
    let client = Client::new(&nowhere()).with_retries(RetryPolicy::none());
    let doubled = || TablePath::new("//tmp/t[#0:#2]").range(0..2);

    let refusals: Vec<(&str, ClientError)> = vec![
        (
            "read_table",
            client.read_table(doubled()).expect_err("refused"),
        ),
        (
            "read_table_rows",
            client
                .read_table_rows::<std::collections::BTreeMap<String, i64>>(doubled())
                .expect_err("refused"),
        ),
        (
            "read_table_streaming",
            // `expect_err` needs a `Debug` success value and a `TableReader`
            // has none, so the success case is spelled out.
            match client.read_table_streaming(doubled()) {
                Ok(_) => panic!("read_table_streaming did not refuse a doubled selection"),
                Err(error) => error,
            },
        ),
        (
            "read_skiff_table",
            client
                .read_skiff_table(TablePath::new("//tmp/t[#0:#2]").range(0..2), &one_column())
                .expect_err("refused"),
        ),
    ];

    for (reader, error) in refusals {
        assert!(
            matches!(&error, ClientError::Config(_)),
            "{reader} did not refuse locally: {error}"
        );
    }
}

#[test]
fn a_read_refuses_a_range_asking_for_rows_no_table_has() {
    // Measured on a local cluster: a backwards range is answered 200 with no
    // rows, in either selector; a negative `row_index` is *clamped to 0* and
    // answered 200 with real rows — `{lower_limit={row_index=-5}}` returned
    // all five rows of a five-row table, and `-5..2` returned rows 0 and 1.
    // Neither bound is honoured as written, and neither is reported.
    let client = Client::new(&nowhere()).with_retries(RetryPolicy::none());
    // From variables because clippy will not compile the literal `5..3` —
    // which is also how a caller reaches it: computed, from an offset that
    // came out wrong, where a silently wrong read is hardest to notice.
    let (from, to) = (5_i64, 3_i64);

    let refusals: Vec<(&str, ClientError)> = vec![
        (
            "rows(5..3)",
            client
                .read_table(TablePath::new("//tmp/t").range(from..to))
                .expect_err("refused"),
        ),
        (
            "rows(-5..2), which the cluster would have read as rows(0..2)",
            client
                .read_table(TablePath::new("//tmp/t").range(-5..2))
                .expect_err("refused"),
        ),
        (
            "keys(b..a), the same mistake in the other selector",
            client
                .read_table(
                    TablePath::new("//tmp/t").range(RowRange::keys(Key::from("b")..Key::from("a"))),
                )
                .expect_err("refused"),
        ),
    ];

    for (selection, error) in refusals {
        assert!(
            matches!(&error, ClientError::Config(_)),
            "{selection} did not refuse locally: {error}"
        );
    }
}

#[test]
fn a_read_sends_an_empty_column_selection() {
    // `columns([])` is a read that succeeds, not one that cannot: measured,
    // `<columns=[]>` answers 200 with one empty map per row, and it composes
    // with a range — `<columns=[];ranges=[…#0:#2]>` came back as two empty
    // maps. That counts the rows of a *range*, or probes whether a key range
    // holds any, with no column bytes on the wire — which `Client::row_count`
    // cannot do, reading as it does the whole-table `@row_count` attribute.
    // So it travels, on the path, as the empty list it is.
    let head = capture(|proxy| {
        let client = Client::new(proxy).with_retries(RetryPolicy::none());
        let _ = client.read_table(
            TablePath::new("//tmp/t")
                .columns(Vec::<String>::new())
                .range(0..2),
        );
    });

    assert!(
        parameters(&head).contains(
            r#"path=<columns=[];ranges=[{lower_limit={row_index=0};upper_limit={row_index=2}}]>"//tmp/t""#
        ),
        "the empty projection did not travel as an empty `columns` attribute:\n{head}"
    );
}

#[test]
fn an_abort_reason_cannot_break_out_of_its_header() {
    // The reason is caller's text and it travels in an HTTP header, so the
    // YSON encoder is the only thing between a chatty message and a forged
    // request. A raw newline here would end the header — and start whatever
    // came after it as a header of its own.
    let head = capture(|proxy| {
        let client = Client::new(proxy).with_retries(RetryPolicy::none());
        client
            .abort_operation(
                "1-2-3-4",
                Some("he said \"stop\"\r\nX-Forged: yes\nand meant it"),
            )
            .expect("aborts");
    });

    let params = parameters(&head);
    assert!(
        params.contains(r#"abort_message="he said \"stop\"\r\nX-Forged: yes\nand meant it""#),
        "the reason was not escaped as YSON text:\n{head}"
    );
    assert!(
        !head.lines().any(|line| line.starts_with("X-Forged")),
        "the reason smuggled a header into the request:\n{head}"
    );
    // One `X-YT-Parameters` line, holding the whole parameter block: the
    // cluster reads the header, not the lines under it.
    assert_eq!(
        head.lines()
            .filter(|line| line.to_lowercase().starts_with("x-yt-parameters:"))
            .count(),
        1,
        "{head}"
    );
}

#[test]
fn an_abort_carries_its_reason_and_no_mutation_id() {
    let head = capture(|proxy| {
        let client = Client::new(proxy).with_retries(RetryPolicy::none());
        client
            .abort_operation("1-2-3-4", Some("stopped by the test"))
            .expect("aborts");
    });

    let params = parameters(&head);
    assert!(head.starts_with("POST /api/v4/abort_operation"), "{head}");
    assert!(params.contains(r#"operation_id="1-2-3-4""#), "{params}");
    // The reason is what tells whoever finds the aborted operation later who
    // stopped it; dropping it silently would be the easy mistake here.
    assert!(
        params.contains(r#"abort_message="stopped by the test""#),
        "{params}"
    );
    // Deliberately absent, though this is a mutating command. The master's
    // mutation cache does not cover a scheduler command: a resend of the same
    // ID is answered `No such operation` rather than with the first response,
    // so a retry would report a successful abort as a failed one. Verified
    // against the cluster before this was changed.
    assert!(!params.contains("mutation_id"), "{params}");
    assert!(!params.contains("retry="), "{params}");
}

#[test]
fn an_abort_without_a_reason_sends_no_empty_message() {
    // An empty `abort_message` is a different statement from none, and it is
    // the one that would show up in the operation's error document as a blank
    // line under "aborted by user request".
    let head = capture(|proxy| {
        let client = Client::new(proxy).with_retries(RetryPolicy::none());
        client.abort_operation("1-2-3-4", None).expect("aborts");
    });

    assert!(
        !parameters(&head).contains("abort_message"),
        "{}",
        parameters(&head)
    );
}

#[test]
fn every_request_asks_for_a_compressed_answer() {
    // The proxy compresses when asked — a 67.7 MiB table came back as 400 KiB
    // — and `ureq` asks on its own because this crate turns its `gzip` feature
    // on. Nothing else here would notice if that feature were dropped: the
    // cluster answers the same either way, just larger.
    let head = capture(|proxy| {
        let client = Client::new(proxy).with_retries(RetryPolicy::none());
        assert_eq!(client.exists("//tmp").ok(), Some(true));
    });

    let lowercase = head.to_lowercase();
    assert!(
        lowercase.contains("accept-encoding: gzip"),
        "the request did not ask for compression:\n{head}"
    );
}

#[test]
fn the_parameters_travel_as_a_header_and_not_a_query_string() {
    let head = capture(|proxy| {
        let client = Client::new(proxy).with_retries(RetryPolicy::none());
        let _ = client.exists("//tmp/some/path");
    });

    assert!(
        head.starts_with("GET /api/v4/exists HTTP/1.1"),
        "the command is the path, with nothing appended:\n{head}"
    );
    // The body is where a data stream goes, so parameters cannot live there;
    // the query string is not where API v4 looks for them either.
    assert!(
        head.contains(r#"x-yt-parameters: {path="//tmp/some/path"}"#)
            || head.contains(r#"X-YT-Parameters: {path="//tmp/some/path"}"#),
        "parameters are not in the header the protocol names:\n{head}"
    );
    assert!(
        head.to_lowercase()
            .contains("x-yt-header-format: <format=text>yson"),
        "the header format must say how the other headers are encoded:\n{head}"
    );
}

#[test]
fn a_token_is_carried_as_an_oauth_authorization() {
    let head = capture(|proxy| {
        let client = Client::with_token(proxy, "secret-token").with_retries(RetryPolicy::none());
        let _ = client.exists("//tmp");
    });

    assert!(
        head.contains("authorization: OAuth secret-token")
            || head.contains("Authorization: OAuth secret-token"),
        "the token is not on the request:\n{head}"
    );
}

#[test]
fn an_unauthenticated_client_sends_no_authorization_at_all() {
    // Not an empty one: a header that is present and empty is a different
    // statement to a proxy than a header that is absent.
    let head = capture(|proxy| {
        let client = Client::new(proxy).with_retries(RetryPolicy::none());
        let _ = client.exists("//tmp");
    });

    assert!(
        !head.to_lowercase().contains("authorization:"),
        "an unauthenticated client sent an authorization header:\n{head}"
    );
}

#[test]
fn a_trace_context_travels_as_a_traceparent_header() {
    // The cluster traces itself, and a request that names a trace has its
    // proxy-side span put inside that one. The header is the W3C spelling,
    // which is what `TryParseTraceParent` in the proxy reads and what all three
    // official clients send; a header we spelled our own way would be dropped
    // in silence and the launch would simply not appear in the trace.
    let context = TraceContext::parse("00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01")
        .expect("a W3C traceparent");

    let head = capture(|proxy| {
        let client = Client::new(proxy)
            .with_retries(RetryPolicy::none())
            .with_trace_context(&context);

        // What will be sent, before it is sent — the read-back a caller logs
        // so the trace can be found again.
        assert_eq!(
            client.traceparent(),
            Some("00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01")
        );

        let _ = client.exists("//tmp");
    });

    // The header *name* is matched case-insensitively because HTTP says it is,
    // and the *value* case-sensitively because the standard says the hex is
    // lowercase. Lowercasing the whole line, as the tests around this one do,
    // would accept an uppercased id the standard does not allow.
    let value = header_value(&head, "traceparent");
    assert_eq!(
        value.as_deref(),
        Some("00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01"),
        "the trace context is not on the request as sent:\n{head}"
    );
}

#[test]
fn a_tracestate_travels_beside_the_traceparent() {
    // The standard pairs the two, and a participant that forwards one is
    // required to forward the other unmodified. The proxy ignores `tracestate`;
    // the caller's own backend is what keys off it, so dropping it here would
    // cost the sampling decision or the correlation key of everything
    // downstream of this hop and nothing would say so.
    let context = TraceContext::parse("00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01")
        .expect("a W3C traceparent")
        .with_tracestate("vendora=t61rcWkgMzE,vendorb=x9");

    let head = capture(|proxy| {
        let client = Client::new(proxy)
            .with_retries(RetryPolicy::none())
            .with_trace_context(&context);

        assert_eq!(client.tracestate(), Some("vendora=t61rcWkgMzE,vendorb=x9"));

        let _ = client.exists("//tmp");
    });

    assert_eq!(
        header_value(&head, "tracestate").as_deref(),
        Some("vendora=t61rcWkgMzE,vendorb=x9"),
        "the tracestate was dropped on the way to the cluster:\n{head}"
    );
    assert_eq!(
        header_value(&head, "traceparent").as_deref(),
        Some("00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01"),
        "a tracestate must not displace the traceparent it belongs to:\n{head}"
    );
}

#[test]
fn a_traced_client_sends_no_tracestate_it_was_not_given() {
    // A `tracestate` naming no vendor is not a smaller version of one; it is a
    // header the caller's backend has to decide what to do with.
    let context = TraceContext::parse("00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01")
        .expect("a W3C traceparent");

    let head = capture(|proxy| {
        let client = Client::new(proxy)
            .with_retries(RetryPolicy::none())
            .with_trace_context(&context);
        let _ = client.exists("//tmp");
    });

    assert!(
        !head.to_lowercase().contains("tracestate"),
        "a tracestate appeared from nowhere:\n{head}"
    );
}

#[test]
fn a_transaction_inherits_the_clients_trace() {
    // The doc on `with_trace_context` promises this, and it holds only because
    // `Transaction::start` clones the client rather than rebuilding one from
    // its parts — which is a plausible refactor, since it immediately overrides
    // the retries and the timeout on the ping client. A commit or a ping that
    // hung is named in that doc as the thing the trace is for, so the claim is
    // load-bearing and nothing else here would notice it breaking.
    let context = TraceContext::parse("00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01")
        .expect("a W3C traceparent");

    let head = capture(|proxy| {
        let client = Client::new(proxy)
            .with_retries(RetryPolicy::none())
            .with_trace_context(&context);
        // The reply is an `exists` answer rather than a transaction id, so the
        // start fails to decode and no ping thread outlives this. The request
        // is what is under test.
        let _ = client.start_transaction();
    });

    assert!(
        head.starts_with("POST /api/v4/start_transaction"),
        "the captured request is not the transaction start:\n{head}"
    );
    assert_eq!(
        header_value(&head, "traceparent").as_deref(),
        Some("00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01"),
        "a transaction started from a traced client left the trace:\n{head}"
    );
}

#[test]
fn a_client_without_a_trace_context_sends_no_traceparent() {
    // Sampling costs the cluster something, so a client that was not asked to
    // join a trace must not start one on its own.
    let head = capture(|proxy| {
        let client = Client::new(proxy).with_retries(RetryPolicy::none());
        let _ = client.exists("//tmp");
    });

    assert!(
        !head.to_lowercase().contains("traceparent"),
        "an untraced client sent a trace context:\n{head}"
    );
}

#[test]
fn the_hosts_lookup_carries_the_trace_like_a_command_does() {
    // `/hosts` is not a command and builds its own request, which is how it
    // once came to carry neither the token nor the timeout. The trace context
    // is one more thing it would miss, and a heavy-proxy lookup slow enough to
    // matter is exactly the one worth seeing in the trace.
    let context = TraceContext::parse("00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01")
        .expect("a W3C traceparent");

    let head = capture(|proxy| {
        let client = Client::with_token(proxy, "secret-token")
            .with_retries(RetryPolicy::none())
            .with_trace_context(&context);
        // The reply is an `exists` answer rather than a host list, so this
        // fails to decode; the request is what is under test.
        let _ = client.heavy_proxy();
    });

    assert!(
        head.starts_with("GET /hosts HTTP/1.1"),
        "the lookup is not the documented one:\n{head}"
    );
    assert_eq!(
        header_value(&head, "traceparent").as_deref(),
        Some("00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01"),
        "the hosts lookup dropped the trace context:\n{head}"
    );
    assert_eq!(
        header_value(&head, "authorization").as_deref(),
        Some("OAuth secret-token"),
        "the hosts lookup dropped the token:\n{head}"
    );
}

#[test]
fn a_raw_command_is_dressed_like_every_other_command() {
    // The reason the escape hatch is a `Client` method and not a bare `ureq`
    // agent handed to the caller. A raw command has to arrive looking like a
    // command — the token, the header format, the parameters header, the
    // compression — or every user of it reimplements this crate's transport
    // badly. A cluster answers the same either way, so nothing but this would
    // notice a regression.
    let head = capture(|proxy| {
        let client = Client::with_token(proxy, "secret-token").with_retries(RetryPolicy::none());
        let _ = client.raw_command(
            Method::Get,
            "get_supported_features",
            &yson_build::empty_map(),
            None,
        );
    });

    let lowercase = head.to_lowercase();
    assert!(
        head.starts_with("GET /api/v4/get_supported_features HTTP/1.1"),
        "the command name is the path, with nothing appended:\n{head}"
    );
    assert!(
        lowercase.contains("authorization: oauth secret-token"),
        "a raw command must carry the token:\n{head}"
    );
    assert!(
        lowercase.contains("x-yt-header-format: <format=text>yson")
            && lowercase.contains("x-yt-parameters: {}"),
        "a raw command must encode its parameters the way the protocol says:\n{head}"
    );
    assert!(
        lowercase.contains("accept-encoding: gzip"),
        "a raw command must ask for compression like the rest:\n{head}"
    );
}

/// A refusal that has to happen before the socket does.
#[test]
fn a_multi_table_skiff_write_is_refused_before_anything_is_sent() {
    // Nothing listens on this port, so reaching the transport at all would be
    // a connection error. A Config error is therefore proof that the format
    // was checked first, and that the caller is told what is actually wrong
    // with the request rather than what the stream looked like to a decoder
    // holding the wrong schema.
    let client = Client::new("http://127.0.0.1:1").with_retries(RetryPolicy::none());
    let two_tables = SkiffFormat::new(vec![
        SkiffSchemaRef::Inline(SkiffSchema::tuple([SkiffSchema::named(
            "a",
            SkiffWireType::Uint64,
        )])),
        SkiffSchemaRef::Inline(SkiffSchema::tuple([SkiffSchema::named(
            "b",
            SkiffWireType::Uint64,
        )])),
    ])
    .expect("two named tuples are a valid format");

    let error = client
        .write_table_with_format(
            // Truncated on purpose: with the checks the other way round this
            // is the byte that produces the misleading answer.
            TablePath::from("//tmp/out"),
            b"\x00",
            &DataFormat::skiff(two_tables),
        )
        .expect_err("direct table I/O takes exactly one table schema");

    assert!(matches!(error, ClientError::Config(_)), "{error:?}");
    assert!(
        error.to_string().contains("exactly one table schema"),
        "{error}"
    );
}

// ------------------------------------------------- the operation lifecycle

#[test]
fn a_suspend_says_what_to_do_with_the_running_jobs() {
    // Sent either way round, never left out. The two are different requests:
    // one lets the jobs that have started finish, the other throws their work
    // away — and a caller who asked for the second and got the first would
    // find out much later, from a cluster bill.
    for abort_running_jobs in [false, true] {
        let head = capture(|proxy| {
            let client = Client::new(proxy).with_retries(RetryPolicy::none());
            client
                .suspend_operation("1-2-3-4", abort_running_jobs)
                .expect("suspends");
        });

        let params = parameters(&head);
        assert!(head.starts_with("POST /api/v4/suspend_operation"), "{head}");
        assert!(params.contains(r#"operation_id="1-2-3-4""#), "{params}");
        assert!(
            params.contains(&format!("abort_running_jobs=%{abort_running_jobs}")),
            "{params}"
        );
    }
}

#[test]
fn the_scheduler_commands_carry_no_mutation_id() {
    // The master's mutation cache does not cover a scheduler command, which is
    // the fact `abort_operation` was built on and these three inherit: a
    // resend under the same ID is answered `No such operation` rather than
    // with the first response. Suspend is retried, but on its own idempotency
    // — a second suspend of a suspended operation is simply accepted — and not
    // by asking the cluster to deduplicate it.
    let heads = [
        (
            "suspend_operation",
            capture(|proxy| {
                let client = Client::new(proxy).with_retries(RetryPolicy::none());
                client
                    .suspend_operation("1-2-3-4", false)
                    .expect("suspends");
            }),
        ),
        (
            "resume_operation",
            capture(|proxy| {
                let client = Client::new(proxy).with_retries(RetryPolicy::none());
                client.resume_operation("1-2-3-4").expect("resumes");
            }),
        ),
        (
            "complete_operation",
            capture(|proxy| {
                let client = Client::new(proxy).with_retries(RetryPolicy::none());
                client.complete_operation("1-2-3-4").expect("completes");
            }),
        ),
    ];

    for (command, head) in heads {
        let params = parameters(&head);
        assert!(
            head.starts_with(&format!("POST /api/v4/{command}")),
            "a mutating command is a POST: {head}"
        );
        assert!(!params.contains("mutation_id"), "{command}: {params}");
        assert!(!params.contains("retry="), "{command}: {params}");
    }
}

#[test]
fn updated_parameters_travel_in_the_header_and_not_in_the_body() {
    // The command reference calls this command's input "structured", which
    // reads like a request body. The cluster's own registry says `null`, and
    // the registry is what the proxy implements: the parameters go in
    // `X-YT-Parameters` like every other command's.
    let head = capture(|proxy| {
        let client = Client::new(proxy).with_retries(RetryPolicy::none());
        client
            .update_operation_parameters(
                "1-2-3-4",
                &OperationParameters::new()
                    .with_pool("fast")
                    .with_weight(2.5),
            )
            .expect("updates");
    });

    let params = parameters(&head);
    assert!(
        head.starts_with("POST /api/v4/update_operation_parameters"),
        "{head}"
    );
    assert!(params.contains(r#"operation_id="1-2-3-4""#), "{params}");
    assert!(
        params.contains("parameters={pool=fast;weight=2.5}"),
        "the parameters are one nested dict, and the weight is a double: {params}"
    );
    assert!(
        !head.to_lowercase().contains("content-length: ")
            || head.to_lowercase().contains("content-length: 0"),
        "the command takes no body:\n{head}"
    );
}

#[test]
fn an_update_that_changes_nothing_is_refused_before_it_is_sent() {
    // Nothing listens on this port, so a Config error proves the check ran
    // first. The cluster answers an empty update with 200 and does nothing,
    // which is the shape of mistake that survives every test but the one that
    // reads the pool afterwards.
    let client = Client::new("http://127.0.0.1:1").with_retries(RetryPolicy::none());
    let error = client
        .update_operation_parameters("1-2-3-4", &OperationParameters::new())
        .expect_err("an empty update is not a request worth sending");

    assert!(matches!(error, ClientError::Config(_)), "{error:?}");
}

#[test]
fn an_alias_lookup_asks_for_runtime_information() {
    // Without it the cluster refuses outright: "Operation alias cannot be
    // resolved without using runtime information". A lookup that forgot this
    // would fail every time, and only against a real cluster.
    let head = capture(|proxy| {
        let client = Client::new(proxy).with_retries(RetryPolicy::none());
        let _ = client.get_operation_by_alias("*nightly", &["state"]);
    });

    let params = parameters(&head);
    assert!(head.starts_with("GET /api/v4/get_operation"), "{head}");
    assert!(params.contains(r#"operation_alias="*nightly""#), "{params}");
    assert!(params.contains("include_runtime=%true"), "{params}");
    assert!(params.contains("attributes=[state]"), "{params}");
    assert!(
        !params.contains("operation_id"),
        "an alias lookup names no id: {params}"
    );
}

#[test]
fn the_whole_operation_document_is_asked_for_by_naming_no_attributes() {
    // `attributes=[]` is a request for *no* attributes, which the cluster
    // answers with `{}`. Leaving the parameter out is what asks for
    // everything, so an empty slice must not be sent as an empty list.
    let head = capture(|proxy| {
        let client = Client::new(proxy).with_retries(RetryPolicy::none());
        let _ = client.get_operation("1-2-3-4", &[]);
    });

    let params = parameters(&head);
    assert_eq!(params, r#"{operation_id="1-2-3-4"}"#);
}

#[test]
fn a_filtered_listing_sends_its_filter_and_nothing_else() {
    let head = capture(|proxy| {
        let client = Client::new(proxy).with_retries(RetryPolicy::none());
        let _ = client.list_operations(
            &OperationFilter::new()
                .with_user("robot-loader")
                .with_state("running")
                .with_limit(20),
        );
    });

    assert!(head.starts_with("GET /api/v4/list_operations"), "{head}");
    assert_eq!(
        parameters(&head),
        "{limit=20;state=running;user=robot-loader}"
    );
}

#[test]
fn a_job_is_asked_for_by_operation_and_job() {
    let head = capture(|proxy| {
        let client = Client::new(proxy).with_retries(RetryPolicy::none());
        // The stub answers `{value=%true}`, which names no job; the request is
        // what this is about.
        let _ = client.get_job("1-2-3-4", "5-6-7-8");
    });

    assert!(head.starts_with("GET /api/v4/get_job "), "{head}");
    assert_eq!(
        parameters(&head),
        r#"{job_id="5-6-7-8";operation_id="1-2-3-4"}"#
    );
}

// ------------------------------------------- where a command is sent, and why
//
// A cluster answers a heavy command the same way whichever of its proxies was
// asked — that is the point of the roles — so nothing about the *answers* here
// could tell a routed client from an unrouted one. Two listeners and a record
// of which one was spoken to can.

/// A stand-in proxy that keeps serving, and remembers what it was asked.
///
/// [`capture`] serves exactly one request, which is all a wire *shape* needs.
/// Routing is about which address a request went to and how many there were,
/// so this keeps a list and stays up.
struct Proxy {
    address: std::net::SocketAddr,
    seen: std::sync::Arc<std::sync::Mutex<Vec<String>>>,
}

/// What a stand-in proxy does with `GET /hosts`.
#[derive(Clone)]
enum Hosts {
    /// Answers 200 with this JSON body.
    List(String),
    /// Answers 404 — a cluster with no such endpoint.
    Absent,
    /// Answers a status that says "not now": the shape of a proxy restarting.
    Status(u16),
    /// Accepts the question and never answers it.
    Hang,
    /// Answers this body, eventually — a cluster slower than the budget.
    Slow(std::time::Duration, String),
    /// Answers each ask with the next `(status, body)`; the last one repeats.
    /// A `/hosts` whose answer changes over time, for the refresh tests.
    Sequence(Vec<(u16, String)>),
}

/// What a stand-in proxy does with a **heavy** command.
///
/// The default stub serves everything, which is a single-node installation and
/// most of this file. It is also, for the tests that are about routing, a stub
/// that cannot fail the way the real thing does: a control proxy refuses a
/// heavy request outright, so a client that stopped routing looks exactly like
/// one that never did unless the stub refuses too.
#[derive(Clone, Copy, PartialEq, Eq)]
enum Role {
    /// Serves every command, whatever its weight.
    Any,
    /// Refuses a heavy command the way a real control proxy does.
    Control,
}

/// The commands a control proxy will not serve, from the cluster's own registry
/// — the `isHeavy` column of `REGISTER_ALL` in `driver.cpp`.
const HEAVY_COMMANDS: &[&str] = &[
    "write_table",
    "write_file",
    "read_table",
    "read_file",
    "get_job_input",
    "get_job_stderr",
];

impl Proxy {
    /// A proxy that answers `/hosts` with `hosts`, or with 404 when given
    /// `None` — the shape of a cluster that has no such endpoint.
    fn new(hosts: Option<String>) -> Self {
        Self::answering(hosts.map_or(Hosts::Absent, Hosts::List), 200)
    }

    /// A proxy that answers `/hosts` one way and commands with `commands`.
    fn answering(hosts: Hosts, commands: u16) -> Self {
        Self::in_role(hosts, commands, Role::Any)
    }

    /// A **control** proxy: it names heavy proxies and serves none of their
    /// work itself.
    fn control(hosts: String) -> Self {
        Self::in_role(Hosts::List(hosts), 200, Role::Control)
    }

    fn in_role(hosts: Hosts, commands: u16, role: Role) -> Self {
        let listener = TcpListener::bind("127.0.0.1:0").expect("binds");
        let address = listener.local_addr().expect("has an address");
        let seen = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));

        let served = std::sync::Arc::clone(&seen);
        std::thread::spawn(move || {
            for stream in listener.incoming() {
                let Ok(stream) = stream else { return };
                let hosts = hosts.clone();
                let seen = std::sync::Arc::clone(&served);
                // One thread per connection: `ureq` pools them, and a client
                // that opened a second while the first sat idle would deadlock
                // against a server that answered them in turn.
                std::thread::spawn(move || serve(stream, hosts, commands, role, seen));
            }
        });

        Self { address, seen }
    }

    /// The address to configure a client with.
    fn url(&self) -> String {
        format!("http://{}", self.address)
    }

    /// The address as `/hosts` names one: a bare host and port, no scheme.
    fn host(&self) -> String {
        self.address.to_string()
    }

    /// The request line of everything served so far.
    fn requests(&self) -> Vec<String> {
        self.seen
            .lock()
            .expect("nothing panicked holding it")
            .iter()
            .map(|head| head.lines().next().unwrap_or_default().to_owned())
            .collect()
    }

    /// Everything served so far, headers and all.
    fn heads(&self) -> Vec<String> {
        self.seen
            .lock()
            .expect("nothing panicked holding it")
            .clone()
    }
}

/// Answers requests on one connection until the client hangs up.
fn serve(
    mut stream: std::net::TcpStream,
    hosts: Hosts,
    commands: u16,
    role: Role,
    seen: std::sync::Arc<std::sync::Mutex<Vec<String>>>,
) {
    let mut reader = BufReader::new(stream.try_clone().expect("clones"));
    // Held open and never answered, for `Hosts::Hang`: dropping the socket
    // would be a connection reset, which is a different thing to measure.
    let mut hung = Vec::new();

    loop {
        let mut head = String::new();
        loop {
            let mut line = String::new();
            match reader.read_line(&mut line) {
                Ok(0) => return,
                Ok(_) if line == "\r\n" => break,
                Ok(_) => head.push_str(&line),
                Err(_) => return,
            }
        }
        if head.is_empty() {
            return;
        }

        // The body first, for the reason `capture` gives: a request is only
        // finished being sent when its body has been read.
        if let Some(length) = content_length(&head) {
            let mut body = vec![0_u8; length];
            if reader.read_exact(&mut body).is_err() {
                return;
            }
        } else if head.to_lowercase().contains("transfer-encoding: chunked") {
            drain_chunked(&mut reader);
        }

        let asked_where_to_go = head.starts_with("GET /hosts ");
        let heavy = HEAVY_COMMANDS
            .iter()
            .any(|command| head.contains(&format!("/api/v4/{command} ")));
        let asked_so_far = {
            let mut seen = seen.lock().expect("nothing panicked holding it");
            seen.push(head);
            seen.iter()
                .filter(|head| head.starts_with("GET /hosts "))
                .count()
        };

        let answer = if asked_where_to_go {
            match &hosts {
                Hosts::List(list) => reply(200, list.as_bytes()),
                Hosts::Absent => reply(404, b""),
                Hosts::Status(status) => reply(*status, b""),
                Hosts::Hang => {
                    hung.push(stream.try_clone().expect("clones"));
                    continue;
                }
                Hosts::Slow(delay, list) => {
                    std::thread::sleep(*delay);
                    reply(200, list.as_bytes())
                }
                Hosts::Sequence(answers) => {
                    let (status, body) = &answers[(asked_so_far - 1).min(answers.len() - 1)];
                    reply(*status, body.as_bytes())
                }
            }
        } else if role == Role::Control && heavy {
            refusal()
        } else {
            reply(commands, br#"{"value"=%true}"#)
        };

        if stream.write_all(&answer).is_err() {
            return;
        }
        stream.flush().ok();
    }
}

/// A response carrying `body`, on a connection that stays open.
fn reply(status: u16, body: &[u8]) -> Vec<u8> {
    let mut reply = format!(
        "HTTP/1.1 {status} .\r\nContent-Length: {}\r\nContent-Type: application/x-yt-yson-text\r\n\r\n",
        body.len()
    )
    .into_bytes();
    reply.extend_from_slice(body);
    reply
}

/// What a control proxy answers a heavy request with input data.
///
/// The cluster's own shape, from `TContext::TryRedirectHeavyRequests`: **503**
/// with `Retry-After`, carrying the reason in an `X-YT-Error` document — which
/// is where this client reads it, so the status is not what the caller sees.
fn refusal() -> Vec<u8> {
    let error =
        r#"{"code":1,"message":"Control proxy may not serve heavy requests with input data"}"#;
    format!("HTTP/1.1 503 .\r\nContent-Length: 0\r\nRetry-After: 60\r\nX-YT-Error: {error}\r\n\r\n")
        .into_bytes()
}

/// An address nothing is listening on, for a proxy that has gone away.
///
/// It has one job — a connection to it must be *refused*, at once — and one
/// hazard behind that job: the port must be one no other stub in the suite can
/// end up bound to, or "nowhere" becomes somewhere.
///
/// The obvious way is to bind `127.0.0.1:0`, read the port the OS chose, and
/// drop the listener so nothing is left answering. That refuses connections —
/// until the OS hands the freed port to another test's `TcpListener::bind`. The
/// range is cycled through before a port is reused, so within one binary it
/// almost never comes round; across a suite that binds thousands of sockets it
/// does, and then a request meant to go nowhere reaches a live listener (or the
/// stub that took the port receives a stray request). It is the kind of flake
/// that passes on a re-run and cannot be reproduced on demand.
///
/// A **privileged** port sidesteps the hazard entirely: `bind("127.0.0.1:0")`
/// draws from the ephemeral range only — 32768+ on Linux, 49152+ on macOS — so
/// nothing in this suite can ever be assigned a port below 1024, and no test
/// process binds one deliberately. Nothing is listening there, so a connection
/// is refused immediately on both platforms. The refusal is *verified* rather
/// than assumed: the first candidate that actually refuses is the one returned,
/// so a machine that happens to run something on one of these ports is skipped
/// rather than silently turning "nowhere" into a live host.
fn nowhere() -> String {
    for port in 1u16..=16 {
        let address = format!("127.0.0.1:{port}");
        let socket = address.parse().expect("a valid loopback address");
        match TcpStream::connect_timeout(&socket, std::time::Duration::from_millis(200)) {
            Err(error) if error.kind() == std::io::ErrorKind::ConnectionRefused => return address,
            // A listener answered, or the connect timed out: not nowhere. Try
            // the next reserved port rather than hand back a live one.
            _ => continue,
        }
    }
    panic!("no reserved loopback port refused a connection; cannot address nowhere");
}

/// A client that discovers, though it is talking to a listener on loopback.
///
/// Discovery is off for a loopback address by default — a local cluster cannot
/// be improved on and a tunnelled one cannot be followed — and every listener
/// in this file is on loopback. So the tests that are *about* discovery say so
/// explicitly, and the one that is about the default does not.
fn discovering(proxy: &Proxy) -> Client {
    Client::new(&proxy.url())
        .with_retries(RetryPolicy::none())
        .with_proxy_discovery(true)
}

#[test]
fn a_heavy_command_goes_to_the_proxy_the_cluster_names() {
    // The bug this file is the regression test for: every heavy command went
    // to whatever `YT_PROXY` held, which on an installation that separates
    // proxy roles is a control proxy, and a control proxy refuses one.
    let heavy = Proxy::new(None);
    let control = Proxy::new(Some(format!(r#"["{}"]"#, heavy.host())));

    Client::with_token(&control.url(), "secret-token")
        .with_retries(RetryPolicy::none())
        .with_proxy_discovery(true)
        .write_table("//tmp/t", b"")
        .expect("writes");

    assert_eq!(
        control.requests(),
        ["GET /hosts HTTP/1.1"],
        "the configured address served the upload itself"
    );
    assert_eq!(heavy.requests(), ["PUT /api/v4/write_table HTTP/1.1"]);

    // The other half of sending it elsewhere: it has to arrive dressed as a
    // command. A heavy proxy that is handed a request with no token answers by
    // blaming the caller's credentials.
    let head = &heavy.heads()[0];
    assert_eq!(
        header_value(head, "authorization").as_deref(),
        Some("OAuth secret-token"),
        "the upload reached the heavy proxy without its token:\n{head}"
    );
    assert!(
        parameters(head).contains(r#"path="//tmp/t""#),
        "the upload lost its parameters on the way:\n{head}"
    );
}

/// One table of one named column — enough to make a Skiff call well formed.
fn one_column() -> SkiffFormat {
    SkiffFormat::new(vec![SkiffSchemaRef::Inline(SkiffSchema::tuple([
        SkiffSchema::named("n", SkiffWireType::Int64),
    ]))])
    .expect("one named tuple is a valid format")
}

#[test]
fn every_heavy_shape_goes_there_and_the_cluster_is_asked_once() {
    // Buffered, streamed in, streamed out, files in both directions, Skiff in
    // both directions and a job's stderr: every route through the transport —
    // `call`, `upload`, `open` — that each had their own way of choosing an
    // address. And one lookup between all of them, because the answer is kept
    // until the refresh interval elapses — a minute by default, which nothing
    // here outlives — not one lookup per command. The interval itself is
    // pinned in
    // `a_stale_answer_is_refreshed_by_the_next_heavy_command_and_a_fresh_one_is_not`.
    //
    // The exact request lists below survive the random pick (#40) because the
    // pool has one host in it: `/hosts` names `heavy` and nobody else, so
    // there is nothing to choose between and the order is the order these
    // calls are made in. Nor does anything here empty that pool — the reads
    // fail on the *answer*, after a 200 the host served perfectly well, which
    // is not a failure attributable to the host and so drops nobody.
    //
    // The list is exact in both directions on purpose. Each of these call
    // sites is one word — `Repeatable::Heavy` — away from going to the control
    // proxy in silence, and `write_file` is the one that matters most:
    // `upload_worker`, `upload_current_exe` and `upload_worker_cached` all
    // funnel through it, so a single wrong word there sends every worker
    // upload back to a proxy that will not take it.
    let heavy = Proxy::new(None);
    let control = Proxy::new(Some(format!(r#"["{}"]"#, heavy.host())));
    let client = discovering(&control);
    let skiff = one_column();

    client.write_table("//tmp/t", b"").expect("buffered write");
    client
        .write_table_rows(
            "//tmp/t",
            [std::collections::BTreeMap::from([("n", 1_i64)])],
        )
        .expect("streamed write");
    client.write_file("//tmp/f", b"x").expect("file write");
    client
        .write_skiff_table("//tmp/t", b"", &skiff)
        .expect("skiff write");
    // The stub answers `{value=%true}`, which is not a table: these fail on
    // the answer, having asked the question this is about.
    let _ = client.read_table("//tmp/t");
    let _ = client.read_table_streaming("//tmp/t");
    let _ = client.read_skiff_table("//tmp/t", &skiff);
    let _ = client.read_file("//tmp/f");
    let _ = client.read_file_streaming("//tmp/f");
    let _ = client.get_job_stderr("1-2-3-4", "5-6-7-8");

    // The `get` is the buffered `read_file`'s completeness check — one light
    // command beside the heavy read, and light commands stay on the address
    // the caller configured. Routing it away would be the same mistake as not
    // routing the read, pointing the other way.
    assert_eq!(
        control.requests(),
        ["GET /hosts HTTP/1.1", "GET /api/v4/get HTTP/1.1"],
        "a heavy command was served by the control proxy"
    );
    assert_eq!(
        heavy.requests(),
        [
            "PUT /api/v4/write_table HTTP/1.1",
            "PUT /api/v4/write_table HTTP/1.1",
            "PUT /api/v4/write_file HTTP/1.1",
            "PUT /api/v4/write_table HTTP/1.1",
            "GET /api/v4/read_table HTTP/1.1",
            "GET /api/v4/read_table HTTP/1.1",
            "GET /api/v4/read_table HTTP/1.1",
            "GET /api/v4/read_file HTTP/1.1",
            "GET /api/v4/read_file HTTP/1.1",
            "GET /api/v4/get_job_stderr HTTP/1.1",
        ]
    );
}

#[test]
fn a_light_command_stays_where_the_client_was_pointed() {
    // Cypress, the scheduler and the master are the control proxy's own work.
    // Sending them to a heavy proxy would be the same mistake pointing the
    // other way, and asking `/hosts` before a `get` would put a round trip in
    // front of every one of them.
    let heavy = Proxy::new(None);
    let control = Proxy::new(Some(format!(r#"["{}"]"#, heavy.host())));
    let client = discovering(&control);

    let _ = client.exists("//tmp");
    let _ = client.create("table", "//tmp/t");
    let _ = client.abort_operation("1-2-3-4", None);

    assert_eq!(
        control.requests(),
        [
            "GET /api/v4/exists HTTP/1.1",
            "POST /api/v4/create HTTP/1.1",
            "POST /api/v4/abort_operation HTTP/1.1",
        ]
    );
    assert!(
        heavy.requests().is_empty(),
        "a light command was routed away: {:?}",
        heavy.requests()
    );
}

#[test]
fn a_cluster_that_names_no_heavy_proxy_keeps_serving_the_uploads_itself() {
    // The fallback that keeps a single-node installation working, and every
    // deployment that does not separate the roles. Asked once and then not
    // again: an empty answer is an answer.
    let control = Proxy::new(Some("[]".to_owned()));
    let client = discovering(&control);

    client.write_table("//tmp/t", b"").expect("writes");
    client.write_file("//tmp/f", b"x").expect("writes");

    assert_eq!(
        control.requests(),
        [
            "GET /hosts HTTP/1.1",
            "PUT /api/v4/write_table HTTP/1.1",
            "PUT /api/v4/write_file HTTP/1.1",
        ]
    );
}

#[test]
fn a_cluster_with_no_hosts_endpoint_is_not_asked_before_every_upload() {
    // `absent`, not merely empty: 404 is deterministic, so asking again would
    // cost a round trip per upload and buy nothing. A failure that might pass —
    // a timeout, a restarting proxy — is judged the other way, by the same
    // rule the retry policy uses.
    let control = Proxy::new(None);
    let client = discovering(&control);

    client.write_table("//tmp/t", b"").expect("writes");
    client.write_table("//tmp/t", b"").expect("writes");

    assert_eq!(
        control.requests(),
        [
            "GET /hosts HTTP/1.1",
            "PUT /api/v4/write_table HTTP/1.1",
            "PUT /api/v4/write_table HTTP/1.1",
        ]
    );
}

#[test]
fn a_heavy_proxy_that_cannot_be_reached_gives_the_configured_address_back() {
    // The measured trigger, from a single-node container reached from the
    // host: `172.17.0.2` is not local, so discovery runs, and `/hosts` answers
    // with a container-internal name nothing outside can dial.
    //
    // The first upload finds that out — it is not re-sent, because heavy
    // commands are not retried and a streamed body is gone by then. What must
    // not happen is what happened before: the answer thrown away, the same
    // question asked, the same dead host resolved, and every upload for the
    // rest of the client's life failing the same way. The client falls back to
    // the address the caller gave, which is serving perfectly well.
    let dead = nowhere();
    let control = Proxy::new(Some(format!(r#"["{dead}"]"#)));
    let client = discovering(&control);

    let first = client.write_table("//tmp/t", b"");
    let second = client.write_table("//tmp/t", b"");

    assert!(first.is_err(), "nothing was listening on {dead}");
    assert!(
        second.is_ok(),
        "the second upload was sent to the dead host too: {second:?}"
    );
    assert_eq!(
        control.requests(),
        ["GET /hosts HTTP/1.1", "PUT /api/v4/write_table HTTP/1.1"],
        "the client kept resolving an address it could not reach"
    );
}

#[test]
fn a_failure_at_a_discovered_proxy_says_which_one() {
    // `write_table: transport error: io: Connection refused` is a true report
    // about an address that appears nowhere in the caller's own code — the
    // client picked it out of a list the cluster gave it, and then said
    // nothing about the choice.
    let dead = nowhere();
    let control = Proxy::new(Some(format!(r#"["{dead}"]"#)));

    let error = discovering(&control)
        .write_table("//tmp/t", b"")
        .expect_err("nothing was listening");

    assert!(
        error
            .to_string()
            .starts_with(&format!("write_table at {dead}:")),
        "the failure did not name the proxy it went to: {error}"
    );
}

#[test]
fn a_failure_at_the_configured_address_is_not_dressed_up_as_a_routed_one() {
    // The other half of naming the host, and the guard that keeps the two
    // decisions apart. The caller typed the configured address, so naming it
    // back at them says nothing — and a failure there is not evidence about a
    // lookup that was never in charge of it.
    //
    // Both uploads fail with the same 503 from the same kind of stub; the only
    // difference is that the client chose where the first one went. That is
    // what the message has to reflect, and what "state is `At(x)` and `x` is
    // where this command went" is for.
    let heavy = Proxy::answering(Hosts::Absent, 503);
    let control = Proxy::answering(Hosts::List(format!(r#"["{}"]"#, heavy.host())), 503);
    let client = discovering(&control);

    let chosen = client.write_table("//tmp/t", b"").expect_err("503");
    let given = client.write_table("//tmp/t", b"").expect_err("503");

    assert!(
        chosen
            .to_string()
            .starts_with(&format!("write_table at {}:", heavy.host())),
        "{chosen}"
    );
    assert!(
        given.to_string().starts_with("write_table:"),
        "a failure at the address the caller gave was reported as a routed one: {given}"
    );
}

#[test]
fn a_settled_lookup_survives_a_failed_upload() {
    // A 404 `/hosts` is remembered as "this cluster serves its own heavy
    // commands", and an upload that then fails at *that* address says nothing
    // about the lookup — the caller chose the address, and there is no other
    // answer to go back for. Forgetting here would restart a settled question
    // on every 503, which is the opposite of what the fallback is for.
    //
    // **With the retry window set to nothing**, which is what makes this test
    // its name. At ten seconds a forgotten answer and a settled one look
    // identical for the whole life of the test: the state could be written as
    // `FellBack` here, or the guard that keeps this code out of a settled
    // lookup could be deleted outright, and this assertion would still pass.
    let control = Proxy::answering(Hosts::Absent, 503);
    let client = discovering(&control).with_hosts_retry_after(std::time::Duration::ZERO);

    assert!(client.write_table("//tmp/t", b"").is_err());
    assert!(client.write_table("//tmp/t", b"").is_err());

    assert_eq!(
        control.requests(),
        [
            "GET /hosts HTTP/1.1",
            "PUT /api/v4/write_table HTTP/1.1",
            "PUT /api/v4/write_table HTTP/1.1",
        ],
        "a settled lookup was restarted by a failure that had nothing to do with it"
    );
}

#[test]
fn a_lookup_that_did_not_settle_is_asked_again_and_a_settled_one_is_not() {
    // The distinction the whole `HeavyProxy` enum turns on, and the one nothing
    // could see: with the window fixed at ten seconds, no test in this suite
    // outlived one, so `Configured` and `FellBack` were observationally
    // identical — either could be written where the other was, and every test
    // stayed green. Set the window to nothing and the two say different things
    // about the very next upload.
    //
    // 404 is deterministic, so asking again would cost a round trip per upload
    // and buy nothing. 503 says only "not now".
    let settled = Proxy::answering(Hosts::Absent, 200);
    let client = discovering(&settled).with_hosts_retry_after(std::time::Duration::ZERO);

    client.write_table("//tmp/t", b"").expect("writes");
    client.write_table("//tmp/t", b"").expect("writes");

    assert_eq!(
        settled.requests(),
        [
            "GET /hosts HTTP/1.1",
            "PUT /api/v4/write_table HTTP/1.1",
            "PUT /api/v4/write_table HTTP/1.1",
        ],
        "a settled answer was asked about again"
    );

    // An answer this client declines in full is settled too: the names will be
    // the same names next time, and the domain rule the same rule.
    let refused = Proxy::new(Some(r#"["n0132-sas.somewhere-else.net"]"#.to_owned()));
    let client = discovering(&refused).with_hosts_retry_after(std::time::Duration::ZERO);

    client.write_table("//tmp/t", b"").expect("writes");
    client.write_table("//tmp/t", b"").expect("writes");

    assert_eq!(
        refused.requests(),
        [
            "GET /hosts HTTP/1.1",
            "PUT /api/v4/write_table HTTP/1.1",
            "PUT /api/v4/write_table HTTP/1.1",
        ],
        "an answer that was declined in full was asked for again"
    );

    let unsettled = Proxy::answering(Hosts::Status(503), 200);
    let client = discovering(&unsettled).with_hosts_retry_after(std::time::Duration::ZERO);

    client.write_table("//tmp/t", b"").expect("writes");
    client.write_table("//tmp/t", b"").expect("writes");

    assert_eq!(
        unsettled.requests(),
        [
            "GET /hosts HTTP/1.1",
            "PUT /api/v4/write_table HTTP/1.1",
            "GET /hosts HTTP/1.1",
            "PUT /api/v4/write_table HTTP/1.1",
        ],
        "a lookup that failed for a reason that might pass was never repeated"
    );
}

#[test]
fn the_control_proxy_in_these_tests_refuses_a_heavy_command_as_a_real_one_does() {
    // Everything below depends on this. A stub that cheerfully serves every
    // upload cannot tell a client that routed from one that did not, so a
    // regression that sends heavy commands back to the control proxy — the
    // whole of #30 — is invisible against it.
    let control = Proxy::control("[]".to_owned());

    let error = Client::new(&control.url())
        .with_retries(RetryPolicy::none())
        .write_table("//tmp/t", b"")
        .expect_err("a control proxy refuses a heavy request with input data");

    assert!(
        error
            .to_string()
            .contains("Control proxy may not serve heavy requests with input data"),
        "{error}"
    );
    // And the refusal says what the cluster cannot: that this client sent it
    // here on purpose, and which switch changes that.
    assert!(
        error.to_string().contains("with_proxy_discovery"),
        "the refusal says nothing about the routing that would have avoided it: {error}"
    );
}

#[test]
fn a_proxy_that_stumbles_is_dropped_from_the_pool_not_for_the_control_proxy() {
    // The failure that arrived through the fix for the previous one. A data
    // proxy answering a single transient 503 — a drain, a restart — used to
    // send the client back to the *configured* address for ten seconds, and on
    // the deployment this feature was written for that address is a balancer in
    // front of the control proxies. So every heavy command in that window was
    // answered `Control proxy may not serve heavy requests with input data`:
    // #30 itself, reproducible on demand, once per hiccup.
    //
    // `/hosts` had already named the alternatives. Now they are used — and the
    // pick is random (#40), so the shape here is not "first fails, second
    // takes over" but "the bad host costs exactly one command, ever": writes
    // land on the good proxy until chance routes one onto the bad, that one
    // fails and drops it, and every write after that succeeds. The bound is
    // 64 tries; the chance a coin lands one side 64 times is not a flake.
    let bad = Proxy::answering(Hosts::Absent, 503);
    let good = Proxy::new(None);
    let control = Proxy::control(format!(r#"["{}", "{}"]"#, bad.host(), good.host()));
    let client = discovering(&control);

    let mut failures = 0;
    for _ in 0..64 {
        if client.write_table("//tmp/t", b"").is_err() {
            failures += 1;
            break;
        }
    }
    assert_eq!(failures, 1, "the bad host was never picked in 64 writes");

    for _ in 0..8 {
        client
            .write_table("//tmp/t", b"")
            .expect("a write after the drop still reached the dropped host");
    }

    assert_eq!(
        control.requests(),
        ["GET /hosts HTTP/1.1"],
        "an upload went back to the control proxy, which is what refuses one"
    );
    assert_eq!(
        bad.requests(),
        ["PUT /api/v4/write_table HTTP/1.1"],
        "the host that failed was not dropped after its first failure"
    );
}

#[test]
fn heavy_commands_spread_across_the_pool_rather_than_piling_onto_the_first() {
    // `/hosts` is ordered by load, so "take `[0]`" reads like the obviously
    // right pick and is the pin: a client that keeps its one pick for life
    // never rebalances, and a draining host keeps every client that ever
    // picked it. Both official clients pick at random per command for exactly
    // this reason, and now so does this one.
    //
    // Sixty-four commands across a pool of three: the chance a fair pick
    // leaves any host unvisited is 3·(2/3)^64 ≈ 5e-12, so "every proxy served
    // something" is an assertion about the picker, not about luck — a
    // round-robin would pass it too, but a pick pinned to any subset fails.
    let (a, b, c) = (Proxy::new(None), Proxy::new(None), Proxy::new(None));
    let control = Proxy::control(format!(
        r#"["{}", "{}", "{}"]"#,
        a.host(),
        b.host(),
        c.host()
    ));
    let client = discovering(&control);

    std::thread::scope(|scope| {
        for _ in 0..64 {
            scope.spawn(|| client.write_table("//tmp/t", b"").expect("writes"));
        }
    });

    assert_eq!(
        control.requests(),
        ["GET /hosts HTTP/1.1"],
        "a fresh answer was re-asked, or an upload went to the control proxy"
    );
    for (name, proxy) in [("first", &a), ("second", &b), ("third", &c)] {
        assert!(
            !proxy.requests().is_empty(),
            "64 uploads over a pool of three never once landed on the {name} host"
        );
    }
}

#[test]
fn a_stale_answer_is_refreshed_by_the_next_heavy_command_and_a_fresh_one_is_not() {
    // The documentation's own strategy — "re-query the /hosts list every
    // minute or every few queries" — which this client declined for a release
    // and now follows (#40). Lazily, like the C++ SDK: the heavy command that
    // finds the answer stale asks first, and no background thread exists to
    // ask on its own.
    //
    // An interval of zero makes every answer already stale, so the second
    // write must re-ask; the mutation "never refresh" fails here. The default
    // interval is the other half — a fresh answer asked about once — and
    // `every_heavy_shape_goes_there_and_the_cluster_is_asked_once` already
    // pins it, but the pair belongs together: an hour-long interval here is
    // what tells "refreshed on the interval" from "refreshed every time".
    let heavy = Proxy::new(None);
    let stale = Proxy::new(Some(format!(r#"["{}"]"#, heavy.host())));
    let client = discovering(&stale).with_host_list_refresh_interval(std::time::Duration::ZERO);

    client.write_table("//tmp/t", b"").expect("writes");
    client.write_table("//tmp/t", b"").expect("writes");

    assert_eq!(
        stale.requests(),
        ["GET /hosts HTTP/1.1", "GET /hosts HTTP/1.1"],
        "a stale answer was not refreshed before the next heavy command"
    );
    assert_eq!(heavy.requests().len(), 2);

    let heavy = Proxy::new(None);
    let fresh = Proxy::new(Some(format!(r#"["{}"]"#, heavy.host())));
    let client =
        discovering(&fresh).with_host_list_refresh_interval(std::time::Duration::from_secs(3600));

    client.write_table("//tmp/t", b"").expect("writes");
    client.write_table("//tmp/t", b"").expect("writes");

    assert_eq!(
        fresh.requests(),
        ["GET /hosts HTTP/1.1"],
        "an answer well inside its interval was asked about again"
    );
    assert_eq!(heavy.requests().len(), 2);
}

/// An interval long enough for a test to do things "inside" it, short enough
/// to sleep past. The sleeps that wait it out add ~0.4 s to the suite each;
/// the alternative — an interval of zero — cannot tell "put off for another
/// interval" from "retried in front of the very next upload", which is the
/// distinction the refresh-failure tests exist to pin.
const TEST_INTERVAL: std::time::Duration = std::time::Duration::from_millis(300);

/// Sleeps `TEST_INTERVAL` out, with margin.
fn outlive_the_interval() {
    std::thread::sleep(TEST_INTERVAL + std::time::Duration::from_millis(100));
}

#[test]
fn a_refresh_that_fails_keeps_the_pool_and_waits_out_another_interval() {
    // A `/hosts` that answered once and then stopped — a coordinator
    // restarting — must not take routing down with it: the hosts in hand are
    // from an answer the cluster did give, and dropping them for the
    // configured address would route uploads to a control proxy because a
    // *lookup* hiccupped. And the failed question is put off for a whole
    // interval rather than hurried back on the short retry window: unlike the
    // initial lookup nothing is waiting on the answer, so retrying it sooner
    // would put a lookup's stall in front of heavy traffic several times a
    // minute for an answer the pool makes unnecessary. The third write here
    // rides the deferral — with the short window this small (zero) and the
    // interval outlived only once, a refresh retried on the wrong clock asks
    // a third time and fails the count.
    let heavy = Proxy::new(None);
    let flaky = Proxy::answering(
        Hosts::Sequence(vec![
            (200, format!(r#"["{}"]"#, heavy.host())),
            (503, String::new()),
        ]),
        200,
    );
    let client = discovering(&flaky)
        .with_host_list_refresh_interval(TEST_INTERVAL)
        .with_hosts_retry_after(std::time::Duration::ZERO);

    client.write_table("//tmp/t", b"").expect("writes");
    outlive_the_interval();
    client
        .write_table("//tmp/t", b"")
        .expect("a failed refresh dropped the pool");
    client
        .write_table("//tmp/t", b"")
        .expect("a failed refresh dropped the pool");

    assert_eq!(
        flaky.requests(),
        ["GET /hosts HTTP/1.1", "GET /hosts HTTP/1.1"],
        "a refresh that failed was retried in front of the very next upload"
    );
    assert_eq!(
        heavy.requests().len(),
        3,
        "an upload left the pool while the refresh was the only thing failing"
    );
}

#[test]
fn a_refresh_adopts_the_answer_it_fetched() {
    // The half of "refreshed" that a second `GET /hosts` alone cannot prove:
    // the fresh answer has to *replace* the pool, or the lookup is theatre.
    // The cluster here names one host and then a different one, and the
    // traffic has to move — which is also the property that restores a
    // dropped host, since a replacement pool is built from the answer alone.
    let (first, second) = (Proxy::new(None), Proxy::new(None));
    let moving = Proxy::answering(
        Hosts::Sequence(vec![
            (200, format!(r#"["{}"]"#, first.host())),
            (200, format!(r#"["{}"]"#, second.host())),
        ]),
        200,
    );
    let client = discovering(&moving).with_host_list_refresh_interval(std::time::Duration::ZERO);

    client.write_table("//tmp/t", b"").expect("writes");
    client.write_table("//tmp/t", b"").expect("writes");

    assert_eq!(
        first.requests(),
        ["PUT /api/v4/write_table HTTP/1.1"],
        "the first answer went unused, or was never given up"
    );
    assert_eq!(
        second.requests(),
        ["PUT /api/v4/write_table HTTP/1.1"],
        "the refreshed answer was fetched and thrown away"
    );
}

#[test]
fn a_dropped_host_is_restored_by_the_next_refresh() {
    // The deliberate price of dropping without a ban list: a host that is
    // *persistently* bad — the misissued certificate that motivated #40 — is
    // re-learned at one failed command per interval, in exchange for a host
    // that was merely draining coming back as soon as the cluster vouches for
    // it again. The proof it came back is that it can fail again: a second
    // failure after the interval is a command reaching a host that a
    // permanent drop would never have let another command reach.
    let bad = Proxy::answering(Hosts::Absent, 503);
    let good = Proxy::new(None);
    let control = Proxy::control(format!(r#"["{}", "{}"]"#, bad.host(), good.host()));
    let client = discovering(&control).with_host_list_refresh_interval(TEST_INTERVAL);

    let mut failures = 0;
    for _ in 0..64 {
        if client.write_table("//tmp/t", b"").is_err() {
            failures += 1;
            break;
        }
    }
    assert_eq!(failures, 1, "the bad host was never picked before the drop");

    outlive_the_interval();

    for _ in 0..64 {
        if client.write_table("//tmp/t", b"").is_err() {
            failures += 1;
            break;
        }
    }
    assert_eq!(
        failures, 2,
        "the dropped host was never restored by the refresh"
    );
    assert_eq!(
        bad.requests().len(),
        2,
        "restoration reached the bad host more (or less) than the drop accounts for"
    );
}

#[test]
fn a_refresh_that_answers_nobody_keeps_the_pool_in_hand() {
    // A fleet mid-rotation can briefly answer `[]`. Adopting that would end
    // routing — and an adopted empty pool would have nothing to pick from at
    // all — where waiting out another interval costs nothing: the hosts in
    // hand are from an answer the cluster did give. The uploads must keep
    // reaching the discovered host, not the configured address.
    let heavy = Proxy::new(None);
    let briefly_empty = Proxy::answering(
        Hosts::Sequence(vec![
            (200, format!(r#"["{}"]"#, heavy.host())),
            (200, "[]".to_owned()),
        ]),
        200,
    );
    let client =
        discovering(&briefly_empty).with_host_list_refresh_interval(std::time::Duration::ZERO);

    client.write_table("//tmp/t", b"").expect("writes");
    client.write_table("//tmp/t", b"").expect("writes");
    client.write_table("//tmp/t", b"").expect("writes");

    assert_eq!(
        heavy.requests().len(),
        3,
        "an empty refresh answer took the pool down with it"
    );
}

#[test]
fn an_emptied_pool_asks_the_cluster_again_after_the_window() {
    // The other half of the fallback's name: it ends. A pool whose every host
    // has been dropped uses the configured address for the retry window, and
    // the first heavy command past the window puts the question again rather
    // than living at the fallback for ever.
    let dead = nowhere();
    let control = Proxy::new(Some(format!(r#"["{dead}"]"#)));
    let client = discovering(&control).with_hosts_retry_after(std::time::Duration::ZERO);

    let first = client.write_table("//tmp/t", b"");
    assert!(first.is_err(), "nothing was listening on {dead}");
    let _ = client.write_table("//tmp/t", b"");

    let asked = control
        .requests()
        .iter()
        .filter(|line| line.starts_with("GET /hosts"))
        .count();
    assert_eq!(asked, 2, "the fallback never ended");
}

#[test]
fn an_interval_of_forever_disables_the_refresh() {
    // `Duration::MAX` never elapses, which has to hold by arithmetic that
    // cannot panic — the pool judges its age with `elapsed()` against the
    // interval rather than by adding the interval to an `Instant`, which
    // `Duration::MAX` would overflow.
    let heavy = Proxy::new(None);
    let control = Proxy::new(Some(format!(r#"["{}"]"#, heavy.host())));
    let client = discovering(&control).with_host_list_refresh_interval(std::time::Duration::MAX);

    client.write_table("//tmp/t", b"").expect("writes");
    client.write_table("//tmp/t", b"").expect("writes");

    assert_eq!(
        control.requests(),
        ["GET /hosts HTTP/1.1"],
        "an interval of forever still refreshed"
    );
    assert_eq!(heavy.requests().len(), 2);
}

#[test]
fn a_cluster_that_named_nobody_is_asked_again_an_interval_later() {
    // "The cluster names no heavy proxy" used to settle for ever, and a
    // permanent answer from one lookup is a pin of its own: a launcher whose
    // first upload landed in the few seconds of a rolling restart when
    // `/hosts` answers `[]` sent every heavy command to the control proxy for
    // the rest of its life. The verdict now expires with the same refresh
    // interval as a pool, so the discovery that ends the window is watched
    // here: an empty answer, then a real one, and the third upload routes.
    let heavy = Proxy::new(None);
    let recovering = Proxy::answering(
        Hosts::Sequence(vec![
            (200, "[]".to_owned()),
            (200, format!(r#"["{}"]"#, heavy.host())),
        ]),
        200,
    );
    let client = discovering(&recovering).with_host_list_refresh_interval(TEST_INTERVAL);

    client.write_table("//tmp/t", b"").expect("writes");
    client.write_table("//tmp/t", b"").expect("writes");
    assert!(
        heavy.requests().is_empty(),
        "an answer well inside its interval was given up early"
    );

    outlive_the_interval();
    client.write_table("//tmp/t", b"").expect("writes");

    assert_eq!(
        heavy.requests(),
        ["PUT /api/v4/write_table HTTP/1.1"],
        "a cluster that named nobody was never asked again"
    );
}

#[test]
fn a_proxy_that_refuses_heavy_work_is_given_up_like_one_that_cannot_be_reached() {
    // `/hosts` lists what `default_role_filter` says, and that is a coordinator
    // config parameter an operator can change — not a protocol guarantee that
    // the names are data proxies. A control proxy among them refuses with an
    // ordinary cluster error, code 1, which is hopeless to *retry* and plainly
    // the host's own fault: the same command is served happily next door. So
    // it is dropped from the pool exactly as an unreachable host is — at most
    // one command ever reaches it, and everything after the drop succeeds.
    let good = Proxy::new(None);
    let wrong_role = Proxy::control("[]".to_owned());
    let control = Proxy::control(format!(r#"["{}", "{}"]"#, wrong_role.host(), good.host()));
    let client = discovering(&control);

    let mut failures = 0;
    for _ in 0..64 {
        if client.write_table("//tmp/t", b"").is_err() {
            failures += 1;
            break;
        }
    }
    assert_eq!(failures, 1, "the wrong-role host was never picked");

    for _ in 0..8 {
        client
            .write_table("//tmp/t", b"")
            .expect("a write after the drop still reached the refusing host");
    }

    assert_eq!(
        wrong_role.requests().len(),
        1,
        "the refusing host was not dropped after its first refusal"
    );
    assert_eq!(control.requests(), ["GET /hosts HTTP/1.1"]);
}

#[test]
fn a_refusal_at_the_configured_address_says_what_would_have_routed_it() {
    // The silent half of #30: the client asked, was given a perfectly good
    // name, declined it under the domain rule, and said nothing. What the
    // operator sees is the cluster error the feature was built to prevent, with
    // nothing in it about `/hosts`, about a name, or about the one builder call
    // that would have used it.
    //
    // The client here is configured with `127.0.0.1`, a literal address, which
    // admits only itself — so `localhost` names the same machine and is still
    // refused.
    let elsewhere = Proxy::new(None);
    let named = format!("localhost:{}", elsewhere.address.port());
    let control = Proxy::control(format!(r#"["{named}"]"#));

    let error = discovering(&control)
        .write_table("//tmp/t", b"")
        .expect_err("a control proxy refuses a heavy request with input data");

    assert!(
        error.to_string().contains("with_heavy_proxies_anywhere"),
        "the refusal does not say that a name was declined: {error}"
    );
    assert!(
        elsewhere.requests().is_empty(),
        "the token went to a host the caller never named: {:?}",
        elsewhere.requests()
    );
}

#[test]
fn a_list_of_proxies_written_out_by_hand_is_what_is_used() {
    // The third mode, and the only one that is a boundary rather than a
    // heuristic: `anywhere` is all-or-nothing, and the domain rule is a guard
    // against a typo — on a shared platform, a parent domain is shared with
    // every other tenant of it.
    let allowed = Proxy::new(None);
    let refused = Proxy::new(None);
    let control = Proxy::control(format!(
        r#"["localhost:{}", "localhost:{}"]"#,
        refused.address.port(),
        allowed.address.port()
    ));

    Client::new(&control.url())
        .with_retries(RetryPolicy::none())
        .with_proxy_discovery(true)
        .with_heavy_proxies_in([format!("localhost:{}", allowed.address.port())])
        .write_table("//tmp/t", b"")
        .expect("writes");

    assert_eq!(control.requests(), ["GET /hosts HTTP/1.1"]);
    assert_eq!(allowed.requests(), ["PUT /api/v4/write_table HTTP/1.1"]);
    assert!(
        refused.requests().is_empty(),
        "a name outside the list was used: {:?}",
        refused.requests()
    );
}

#[test]
fn the_lookup_budget_can_be_raised_and_not_only_lowered() {
    // It used to be the smaller of 800 ms and the client's own timeout, so
    // `with_timeout` could only ever lower it: a cluster answering `/hosts` in
    // 900 ms was unroutable by any configuration at all. And 800 ms is not
    // always generous — the first heavy command is often a client's first
    // request, with DNS, TCP and a TLS handshake inside the same budget.
    let heavy = Proxy::new(None);
    let slow = Hosts::Slow(
        std::time::Duration::from_millis(1200),
        format!(r#"["{}"]"#, heavy.host()),
    );
    let control = Proxy::answering(slow, 200);

    Client::new(&control.url())
        .with_retries(RetryPolicy::none())
        .with_proxy_discovery(true)
        .write_table("//tmp/t", b"")
        .expect("writes");

    assert!(
        heavy.requests().is_empty(),
        "the default budget waited out a cluster it is meant to give up on"
    );

    Client::new(&control.url())
        .with_retries(RetryPolicy::none())
        .with_proxy_discovery(true)
        .with_hosts_timeout(std::time::Duration::from_secs(3))
        .write_table("//tmp/t", b"")
        .expect("writes");

    assert_eq!(
        heavy.requests(),
        ["PUT /api/v4/write_table HTTP/1.1"],
        "the budget could not be raised, so this cluster is unroutable"
    );
}

#[test]
fn a_mistake_at_the_heavy_proxy_does_not_send_the_client_back_to_ask() {
    // A table that does not exist will not exist over there either. Only a
    // failure another proxy could plausibly not have — a refused connection, a
    // 503, a banned proxy — is worth giving up a resolved address for; asking
    // again on every mistake would cost a lookup per typo.
    let heavy = Proxy::answering(Hosts::Absent, 404);
    let control = Proxy::new(Some(format!(r#"["{}"]"#, heavy.host())));
    let client = discovering(&control);

    assert!(client.write_table("//tmp/t", b"").is_err());
    assert!(client.write_table("//tmp/t", b"").is_err());

    assert_eq!(control.requests(), ["GET /hosts HTTP/1.1"]);
    assert_eq!(
        heavy.requests(),
        [
            "PUT /api/v4/write_table HTTP/1.1",
            "PUT /api/v4/write_table HTTP/1.1",
        ],
        "the client threw away a working address because a command was wrong"
    );
}

#[test]
fn a_discovery_off_client_never_touches_the_routing_state() {
    // Nothing to forget and nothing to ask, so a heavy failure costs no lock
    // and no lookup. The point is that `with_proxy_discovery(false)` is a
    // whole-feature off switch and not merely a "do not ask first".
    let control = Proxy::answering(Hosts::List(r#"["heavy.example.net"]"#.to_owned()), 503);

    let client = Client::new(&control.url())
        .with_retries(RetryPolicy::none())
        .with_proxy_discovery(false);

    assert!(client.write_table("//tmp/t", b"").is_err());
    assert!(client.write_table("//tmp/t", b"").is_err());

    assert_eq!(
        control.requests(),
        [
            "PUT /api/v4/write_table HTTP/1.1",
            "PUT /api/v4/write_table HTTP/1.1",
        ]
    );
}

#[test]
fn a_blank_name_in_the_answer_is_passed_over_rather_than_believed() {
    // `/hosts` is ordered best-first, so the first entry is the one to use —
    // unless it is not a host name at all. A blank one used to be filtered out
    // by hand; it is now one of the several shapes `heavy_base` refuses, and
    // the list is walked until something usable turns up.
    let heavy = Proxy::new(None);
    let control = Proxy::new(Some(format!(r#"["", "   ", "{}"]"#, heavy.host())));
    let client = discovering(&control);

    client.write_table("//tmp/t", b"").expect("writes");

    assert_eq!(control.requests(), ["GET /hosts HTTP/1.1"]);
    assert_eq!(heavy.requests(), ["PUT /api/v4/write_table HTTP/1.1"]);
}

#[test]
fn an_answer_that_is_not_a_list_of_host_names_settles_the_question() {
    // A body this client cannot read will not become readable by being asked
    // for again, so it is remembered exactly as an empty list is: the
    // configured address serves the uploads, and no lookup goes in front of
    // them.
    for nonsense in [
        "not json at all",
        r#"{"hosts": ["n0132-sas.example.net"]}"#,
        "[1, 2, 3]",
        "null",
    ] {
        let control = Proxy::answering(Hosts::List(nonsense.to_owned()), 200);
        let client = discovering(&control);

        client.write_table("//tmp/t", b"").expect("writes");
        client.write_table("//tmp/t", b"").expect("writes");

        assert_eq!(
            control.requests(),
            [
                "GET /hosts HTTP/1.1",
                "PUT /api/v4/write_table HTTP/1.1",
                "PUT /api/v4/write_table HTTP/1.1",
            ],
            "{nonsense:?} was asked about twice, or routed somewhere"
        );
    }
}

#[test]
fn a_host_outside_the_configured_domain_is_refused() {
    // The `/hosts` body decides where every heavy command goes, and a heavy
    // command carries the caller's OAuth token. On a plain-http base, forging
    // this body is exactly as easy as forging a `Location` header — which this
    // client refuses to follow. So a name from another domain is passed over,
    // and the upload goes to the address the caller actually chose.
    //
    // The client here is configured with `127.0.0.1`, which is a literal
    // address and therefore admits only itself: `localhost` names the same
    // machine and is still not the same host.
    let elsewhere = Proxy::new(None);
    let named = format!("localhost:{}", elsewhere.address.port());
    let control = Proxy::new(Some(format!(r#"["{named}"]"#)));

    discovering(&control)
        .write_table("//tmp/t", b"")
        .expect("writes");

    assert_eq!(
        control.requests(),
        ["GET /hosts HTTP/1.1", "PUT /api/v4/write_table HTTP/1.1"],
    );
    assert!(
        elsewhere.requests().is_empty(),
        "the token went to a host the caller never named: {:?}",
        elsewhere.requests()
    );
}

#[test]
fn an_installation_that_really_does_answer_elsewhere_can_opt_in() {
    // The escape hatch, so that an installation whose `/hosts` genuinely names
    // another domain — a vanity address, data proxies in a separate zone — is
    // not stranded by the default. Same two listeners as above, one builder
    // call apart.
    let heavy = Proxy::new(None);
    let named = format!("localhost:{}", heavy.address.port());
    let control = Proxy::new(Some(format!(r#"["{named}"]"#)));

    Client::new(&control.url())
        .with_retries(RetryPolicy::none())
        .with_proxy_discovery(true)
        .with_heavy_proxies_anywhere(true)
        .write_table("//tmp/t", b"")
        .expect("writes");

    assert_eq!(control.requests(), ["GET /hosts HTTP/1.1"]);
    assert_eq!(heavy.requests(), ["PUT /api/v4/write_table HTTP/1.1"]);
}

#[test]
fn the_lookup_has_its_own_budget_and_a_heavy_command_does_not_wait_out_the_clients() {
    // The lookup used to run under the client's full policy — five attempts,
    // one to eight seconds of backoff between them, two minutes each — while
    // holding the lock every other heavy command wants. A `/hosts` answering
    // 503 therefore cost a heavy command **15 s**, and one that hung cost it
    // **615 s**. Both measured; the second is why this test exists at all.
    //
    // The bound is loose because this is a clock, and the thing being ruled
    // out is two orders of magnitude away from it.
    for (what, hosts) in [
        ("503", Hosts::Status(503)),
        ("no answer at all", Hosts::Hang),
    ] {
        let control = Proxy::answering(hosts, 200);
        // The client's own policy, not `none()`: the point is that the lookup
        // does not inherit it.
        let client = Client::new(&control.url()).with_proxy_discovery(true);

        let started = std::time::Instant::now();
        client.write_table("//tmp/t", b"").expect("writes");
        let elapsed = started.elapsed();

        assert!(
            elapsed < std::time::Duration::from_secs(5),
            "a heavy command waited {elapsed:?} on a /hosts that answered {what}"
        );
    }
}

#[test]
fn waiting_threads_do_not_queue_up_behind_a_failing_lookup() {
    // Eight threads, one broken `/hosts`. Each used to wait out the lookup in
    // front of it and then perform its own, because a failure that might pass
    // left the state unasked: 8 x 5 attempts = 40 lookups and 240 s, measured.
    //
    // A lookup that fails now leaves an answer too — "the configured address,
    // for the next few seconds" — so the seven behind the first find a
    // decision rather than an invitation to repeat it.
    let control = Proxy::answering(Hosts::Hang, 200);
    let client = Client::new(&control.url()).with_proxy_discovery(true);

    let started = std::time::Instant::now();
    std::thread::scope(|scope| {
        for _ in 0..8 {
            scope.spawn(|| client.write_table("//tmp/t", b"").expect("writes"));
        }
    });
    let elapsed = started.elapsed();

    assert!(
        elapsed < std::time::Duration::from_secs(5),
        "eight threads took {elapsed:?}, which is a queue rather than one lookup"
    );
    let asked = control
        .requests()
        .iter()
        .filter(|line| line.starts_with("GET /hosts"))
        .count();
    assert!(asked <= 2, "{asked} lookups for one question");
}

#[test]
fn eight_threads_against_a_healthy_cluster_still_ask_exactly_once() {
    // The success path, which was already perfect and must stay that way: the
    // lock is held across the lookup precisely so that the seven threads
    // behind the first read its answer instead of asking again.
    let heavy = Proxy::new(None);
    let control = Proxy::new(Some(format!(r#"["{}"]"#, heavy.host())));
    let client = discovering(&control);

    std::thread::scope(|scope| {
        for _ in 0..8 {
            scope.spawn(|| client.write_table("//tmp/t", b"").expect("writes"));
        }
    });

    assert_eq!(control.requests(), ["GET /hosts HTTP/1.1"]);
    assert_eq!(heavy.requests().len(), 8);
}

#[test]
fn a_local_cluster_is_never_asked_where_to_send_a_heavy_command() {
    // The no-regression test, and the reason the others have to ask for
    // discovery explicitly. A cluster on loopback is this machine's own or a
    // tunnel to one: the address such a cluster publishes for itself is not
    // reachable from here, and following it would break every upload that
    // works today. So the default sends the request straight there, with no
    // lookup in front of it.
    let control = Proxy::new(Some(r#"["heavy.example.net"]"#.to_owned()));

    Client::new(&control.url())
        .with_retries(RetryPolicy::none())
        .write_table("//tmp/t", b"")
        .expect("writes");

    assert_eq!(control.requests(), ["PUT /api/v4/write_table HTTP/1.1"]);
}