autumn-web 0.9.0

//! Confirming that a published `_acme-challenge` TXT record is actually visible
//! in public DNS (issue #1620).
//!
//! A DNS-01 challenge fails — and burns an ACME authorization — if the CA
//! queries before the record has propagated. So after writing a record, autumn
//! waits until **every** configured resolver returns the expected value, bounded
//! by `[server.tls.acme.dns] propagation_timeout_secs`. The timeout error names
//! the exact record, value and resolver that never caught up, because "DNS-01
//! failed" without that is unactionable.
//!
//! # Why a hand-rolled query
//!
//! One question type (`TXT`), one class (`IN`), against explicit resolver
//! addresses — the answer parsing is ~100 lines and, unlike a stub-resolver
//! crate, it lets the wait be driven deterministically in tests against an
//! in-process UDP server. It is also the same code `autumn doctor` uses for its
//! DNS-01 preflight, so the check and the runtime agree by construction.
//!
//! This is a **DNS client for one narrow purpose**, not a resolver.
//!
//! # Why the probe goes to the AUTHORITATIVE servers
//!
//! The obvious implementation — ask `1.1.1.1` whether the record is there yet —
//! is quietly broken. The first probe fires the instant the provider's API
//! returns, which is *before* the record is live on the zone's own nameservers,
//! so the recursive resolver answers `NXDOMAIN` and **caches that negatively**
//! for the zone's SOA minimum (RFC 2308). Route 53 defaults to 900s and
//! Cloudflare to 1800s — both longer than the 300s propagation budget. Every
//! later probe then reads the cached negative answer, the wait times out, and
//! the next hourly attempt repeats it. Forever.
//!
//! So the configured `[server.tls.acme.dns] resolvers` are used to *discover*
//! the zone's authoritative nameservers ([`authoritative_resolvers`]), and the
//! propagation probe is sent to those directly with recursion **not** desired —
//! which is also what the CA effectively does. The configured resolvers remain
//! the fallback when discovery fails (a split-horizon setup, a resolver that
//! will not answer `NS`), because a recursive probe is still better than none.

// autumn-determinism-gate: production code in this module must read time and
// mint identifiers through the framework's injected seams (ClockSource /
// Entropy), never `Instant::now()` / `Utc::now()` / `SystemTime::now()` /
// `Uuid::new_v4()` directly. See CONTRIBUTING.md "Determinism seam gate"
// (issue #1797). Justify exceptions with
// #[allow(clippy::disallowed_methods, reason = "…")] at the narrowest scope.
#![cfg_attr(not(test), deny(clippy::disallowed_methods))]

use std::collections::BTreeMap;
use std::net::SocketAddr;
use std::time::Duration;

use futures::future::BoxFuture;

use super::TxtRecord;

/// `A` query type (RFC 1035 §3.2.2).
const QTYPE_A: u16 = 1;
/// `NS` query type.
const QTYPE_NS: u16 = 2;
/// `AAAA` query type (RFC 3596).
const QTYPE_AAAA: u16 = 28;
/// Most candidate zones NS discovery asks about at the same time.
const MAX_ZONE_PROBES: usize = 8;
/// Most nameserver names of one zone that discovery resolves.
const MAX_NAMESERVERS: usize = 8;
/// `CNAME` record type.
const QTYPE_CNAME: u16 = 5;
/// Most CNAME hops [`DnsAnswer::txt_values_via_cnames`] follows.
const MAX_CNAME_HOPS: usize = 8;
/// `TXT` query type.
const QTYPE_TXT: u16 = 16;
/// `OPT` pseudo-record type (EDNS0, RFC 6891).
const QTYPE_OPT: u16 = 41;
/// `IN` class.
const QCLASS_IN: u16 = 1;
/// Recursion-desired flag in the header's second 16-bit word.
const FLAG_RECURSION_DESIRED: u16 = 0x0100;
/// Query/response flag: set on a response.
const FLAG_RESPONSE: u16 = 0x8000;
/// Truncation flag.
const FLAG_TRUNCATED: u16 = 0x0200;
/// How many compression pointers a single name may follow before the message is
/// rejected as malformed. Bounds the classic decompression loop.
const MAX_NAME_POINTERS: usize = 32;
/// Fixed DNS header length.
const HEADER_LEN: usize = 12;
/// Maximum UDP answer we read. 4 KiB comfortably holds a handful of 43-byte
/// challenge values plus overhead; a larger answer sets the TC bit, which is
/// reported rather than silently truncated.
const MAX_RESPONSE: usize = 4096;

/// What one resolver said about a TXT name.
#[derive(Debug, Clone, PartialEq, Eq)]
#[non_exhaustive]
pub struct TxtAnswer {
    /// The TXT values in the answer section (character-strings concatenated per
    /// record, as RFC 1035 §3.3.14 requires).
    pub values: Vec<String>,
    /// The response code: `0` NOERROR, `3` NXDOMAIN, …
    pub rcode: u8,
}

impl TxtAnswer {
    /// Whether the name exists but currently carries no TXT record.
    #[must_use]
    pub const fn is_nxdomain(&self) -> bool {
        self.rcode == 3
    }
}

/// Sends one DNS query to one server.
///
/// A trait so the propagation wait and the authoritative-server discovery can be
/// driven deterministically in tests, against scripted answers rather than the
/// network.
pub trait DnsLookup: Send + Sync {
    /// Ask `server` a question of type `qtype` about `name`.
    ///
    /// # Errors
    ///
    /// Returns a message for a transport failure, a malformed or unrelated
    /// answer, or a server-side failure (`SERVFAIL`, `REFUSED`). An absent record
    /// is `Ok` with no matching records, not an error — it simply has not
    /// propagated yet.
    fn query<'a>(
        &'a self,
        server: SocketAddr,
        name: &'a str,
        qtype: u16,
        recursion_desired: bool,
    ) -> BoxFuture<'a, Result<DnsAnswer, String>>;
}

/// The production [`DnsLookup`]: a UDP query straight to the server.
pub struct UdpDnsLookup {
    timeout: Duration,
}

impl UdpDnsLookup {
    /// Build a lookup with a per-query timeout.
    #[must_use]
    pub const fn new(timeout: Duration) -> Self {
        Self { timeout }
    }
}

impl Default for UdpDnsLookup {
    fn default() -> Self {
        Self::new(Duration::from_secs(5))
    }
}

impl DnsLookup for UdpDnsLookup {
    fn query<'a>(
        &'a self,
        server: SocketAddr,
        name: &'a str,
        qtype: u16,
        recursion_desired: bool,
    ) -> BoxFuture<'a, Result<DnsAnswer, String>> {
        Box::pin(async move {
            let id = query_id();
            let query = encode_query(id, name, qtype, recursion_desired)?;
            let socket = tokio::net::UdpSocket::bind(unspecified_bind(server))
                .await
                .map_err(|e| format!("could not open a UDP socket to query {server}: {e}"))?;
            socket
                .connect(server)
                .await
                .map_err(|e| format!("could not connect to {server}: {e}"))?;
            socket
                .send(&query)
                .await
                .map_err(|e| format!("could not send a query for {name} to {server}: {e}"))?;
            let mut buf = vec![0_u8; MAX_RESPONSE];
            let read = tokio::time::timeout(self.timeout, socket.recv(&mut buf))
                .await
                .map_err(|_| {
                    format!(
                        "{server} did not answer a query for {name} within {}s",
                        self.timeout.as_secs()
                    )
                })?
                .map_err(|e| format!("could not read the answer from {server}: {e}"))?;
            parse_response(id, name, &buf[..read])
                .map_err(|e| format!("{server} answered a query for {name}: {e}"))
        })
    }
}

/// Discover the addresses of the nameservers **authoritative** for `fqdn`.
///
/// Asks `NS` for the label suffixes of `fqdn`, most specific first, through
/// `recursive` (the configured resolvers), takes the most specific suffix
/// that answers with nameserver names, and resolves those names to IPv4 and
/// IPv6 addresses.
///
/// Returns an empty vector when discovery fails at any step — the caller then
/// falls back to probing the recursive resolvers, which is worse but not
/// nothing. See the module docs for why the authoritative probe matters.
pub async fn authoritative_resolvers(
    fqdn: &str,
    recursive: &[SocketAddr],
    lookup: &dyn DnsLookup,
) -> Vec<SocketAddr> {
    discover(fqdn, recursive, &Failed::default(), lookup).await
}

/// `authoritative_resolvers`, skipping the servers in `failed` and adding
/// each server that fails to it.
async fn discover(
    fqdn: &str,
    recursive: &[SocketAddr],
    failed: &Failed,
    lookup: &dyn DnsLookup,
) -> Vec<SocketAddr> {
    use futures::{FutureExt as _, StreamExt as _};
    // The `_acme-challenge` label stays on: it can itself be a delegated zone
    // with its own NS records, which is a recommended way to give an ACME
    // client credentials that reach nothing else. Stripping it skipped that
    // zone and asked the PARENT's nameservers, which answer a referral rather
    // than the TXT value — so a correctly published record could only ever time
    // out. It is simply the most specific candidate, tried first (issue #1620).
    let base = normalize_name(fqdn);
    let labels: Vec<&str> = base.split('.').filter(|l| !l.is_empty()).collect();
    let zones: Vec<String> = (0..labels.len().saturating_sub(1))
        .map(|start| labels[start..].join("."))
        .collect();
    // Candidate zones are asked `MAX_ZONE_PROBES` at a time, most specific
    // first, so a hostname with many labels cannot start a lookup for each at
    // once. A resolver that fails a query is not asked again, so one that
    // drops packets costs one query timeout in total (#2642).
    // Boxed first: a lazily mapped iterator here makes the future not `Send`
    // for every lifetime, which the spawned tasks need.
    let probes: Vec<BoxFuture<'_, Vec<String>>> = zones
        .iter()
        .map(|zone| {
            first_found(
                recursive,
                failed,
                zone,
                &[QTYPE_NS],
                lookup,
                DnsAnswer::ns_names,
            )
            .boxed()
        })
        .collect();
    let mut ns_answers = futures::stream::iter(probes).buffered(MAX_ZONE_PROBES);
    while let Some(mut names) = ns_answers.next().await {
        if names.is_empty() {
            continue;
        }
        // A zone can list many names; only the first `MAX_NAMESERVERS` are
        // resolved, so one tenant's zone cannot start a lookup for each.
        names.sort();
        names.dedup();
        names.truncate(MAX_NAMESERVERS);
        let ip_answers = futures::future::join_all(names.iter().map(|name| {
            first_found(
                recursive,
                failed,
                name,
                &[QTYPE_A, QTYPE_AAAA],
                lookup,
                DnsAnswer::ip_addrs,
            )
        }))
        .await;
        let mut addrs = Vec::new();
        for ip in ip_answers.into_iter().flatten() {
            let socket = SocketAddr::new(ip, 53);
            if !addrs.contains(&socket) {
                addrs.push(socket);
            }
        }
        if !addrs.is_empty() {
            return addrs;
        }
    }
    Vec::new()
}

/// Servers that failed a query during one lookup.
type Failed = std::sync::Mutex<Vec<SocketAddr>>;

/// Ask every server in `servers` that has not failed yet for every type in
/// `qtypes` at once, with recursion desired, and return what `pick` finds.
///
/// For each type, the first answer with records counts. It returns once each
/// type has records or has no query left, so a server that drops packets
/// delays only a type that no other server has records for. A server that
/// fails is added to `failed` and is not asked again.
async fn first_found<T>(
    servers: &[SocketAddr],
    failed: &Failed,
    name: &str,
    qtypes: &[u16],
    lookup: &dyn DnsLookup,
    pick: impl Fn(&DnsAnswer) -> Vec<T>,
) -> Vec<T> {
    use futures::{FutureExt as _, StreamExt as _};
    let live: Vec<SocketAddr> = {
        let failed = failed
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        servers
            .iter()
            .copied()
            .filter(|server| !failed.contains(server))
            .collect()
    };
    let mut pending: futures::stream::FuturesUnordered<_> = live
        .iter()
        .flat_map(|&server| {
            qtypes.iter().enumerate().map(move |(slot, &qtype)| {
                lookup
                    .query(server, name, qtype, true)
                    .map(move |answer| (server, slot, answer))
            })
        })
        .collect();
    let mut outstanding = vec![live.len(); qtypes.len()];
    let mut found: Vec<Vec<T>> = qtypes.iter().map(|_| Vec::new()).collect();
    while let Some((server, slot, answer)) = pending.next().await {
        outstanding[slot] -= 1;
        if let Ok(answer) = answer {
            if found[slot].is_empty() {
                found[slot] = pick(&answer);
            }
        } else {
            let mut failed = failed
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner);
            if !failed.contains(&server) {
                failed.push(server);
            }
        }
        if found
            .iter()
            .zip(&outstanding)
            .all(|(found, left)| !found.is_empty() || *left == 0)
        {
            break;
        }
    }
    found.into_iter().flatten().collect()
}

/// Every TXT value published at `fqdn`, for a custom domain's ownership check
/// (#2642).
///
/// Asks the zone's authoritative nameservers, found through `recursive`, with
/// recursion not desired: their answer is fresh even when a recursive resolver
/// cached a negative answer from before the tenant published. Also asks
/// `recursive`, which covers a failed discovery and a record delegated by CNAME
/// to another zone, which only a recursive resolver follows. Values from every
/// server that answered are merged, following any CNAME chain in each answer.
/// When an authoritative answer ends at a CNAME to another zone, the target
/// is asked of its own authoritative servers, up to the CNAME hop bound.
///
/// All servers are asked at the same time, so a server that drops packets
/// costs one query timeout, not one per server. Discovery and the
/// authoritative queries stop at `deadline`; the recursive answers still count.
///
/// # Errors
///
/// Returns the last error when no server answered at all.
pub async fn txt_values(
    fqdn: &str,
    recursive: &[SocketAddr],
    lookup: &dyn DnsLookup,
    deadline: Duration,
) -> Result<Vec<String>, String> {
    // Each authoritative answer is kept as it arrives, so the deadline drops
    // only the servers still silent, not answers already in.
    let collected: std::sync::Mutex<Vec<(String, Result<DnsAnswer, String>)>> =
        std::sync::Mutex::new(Vec::new());
    let keep = |asked: &str, answer: Result<DnsAnswer, String>| {
        collected
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .push((asked.to_owned(), answer));
    };
    let from_authoritative = async {
        use futures::StreamExt as _;
        let mut asked: Vec<String> = Vec::new();
        let mut name = normalize_name(fqdn);
        // Shared by every hop, so a server that drops packets costs one
        // timeout for the whole chain.
        let failed = Failed::default();
        for _ in 0..=MAX_CNAME_HOPS {
            asked.push(name.clone());
            let servers = discover(&name, recursive, &failed, lookup).await;
            let mut pending: futures::stream::FuturesUnordered<_> = servers
                .iter()
                .map(|server| lookup.query(*server, &name, QTYPE_TXT, false))
                .collect();
            // The servers of one zone agree, so the first CNAME target is
            // followed at once and the rest of this hop is not waited for.
            let mut target = None;
            while let Some(answer) = pending.next().await {
                if let Ok(answer) = &answer {
                    target = answer.unresolved_cname_target(&name);
                }
                keep(&name, answer);
                if target.is_some() {
                    break;
                }
            }
            drop(pending);
            match target {
                Some(target) if !asked.contains(&target) => name = target,
                _ => break,
            }
        }
    };
    let from_authoritative = async {
        if tokio::time::timeout(deadline, from_authoritative)
            .await
            .is_err()
        {
            keep(
                fqdn,
                Err(format!(
                    "the authoritative TXT lookup for {fqdn} did not finish within {}s",
                    deadline.as_secs()
                )),
            );
        }
    };
    let from_recursive = async {
        ask_txt_of_all(recursive, fqdn, true, lookup)
            .await
            .into_iter()
            .map(|answer| (fqdn.to_owned(), answer))
            .collect::<Vec<_>>()
    };
    let ((), recursive) = futures::future::join(from_authoritative, from_recursive).await;
    let authoritative = collected
        .into_inner()
        .unwrap_or_else(std::sync::PoisonError::into_inner);
    let mut values: Vec<String> = Vec::new();
    let mut answered = false;
    let mut last_error = "no resolvers were configured".to_owned();
    for (asked, result) in authoritative.into_iter().chain(recursive) {
        match result {
            Ok(answer) => {
                answered = true;
                for value in answer.txt_values_via_cnames(&asked) {
                    if !values.contains(&value) {
                        values.push(value);
                    }
                }
            }
            Err(e) => last_error = e,
        }
    }
    if answered {
        Ok(values)
    } else {
        Err(last_error)
    }
}

/// Ask every server in `servers` for `fqdn`'s TXT record at the same time.
async fn ask_txt_of_all(
    servers: &[SocketAddr],
    fqdn: &str,
    recursion_desired: bool,
    lookup: &dyn DnsLookup,
) -> Vec<Result<DnsAnswer, String>> {
    futures::future::join_all(
        servers
            .iter()
            .map(|server| lookup.query(*server, fqdn, QTYPE_TXT, recursion_desired)),
    )
    .await
}

/// Query one resolver for a name's TXT values, blocking.
///
/// The same wire code as [`UdpDnsLookup`], for `autumn doctor`'s synchronous
/// check path.
///
/// # Errors
///
/// As [`DnsLookup::query`].
pub fn lookup_txt_blocking(
    resolver: SocketAddr,
    name: &str,
    timeout: Duration,
) -> Result<TxtAnswer, String> {
    let id = query_id();
    let query = encode_query(id, name, QTYPE_TXT, true)?;
    let socket = std::net::UdpSocket::bind(unspecified_bind(resolver))
        .map_err(|e| format!("could not open a UDP socket to query {resolver}: {e}"))?;
    socket
        .set_read_timeout(Some(timeout))
        .map_err(|e| format!("could not set a read timeout: {e}"))?;
    socket
        .connect(resolver)
        .map_err(|e| format!("could not connect to resolver {resolver}: {e}"))?;
    socket
        .send(&query)
        .map_err(|e| format!("could not send a TXT query for {name} to {resolver}: {e}"))?;
    let mut buf = vec![0_u8; MAX_RESPONSE];
    let read = socket
        .recv(&mut buf)
        .map_err(|e| format!("resolver {resolver} did not answer a TXT query for {name}: {e}"))?;
    parse_txt_response(id, name, &buf[..read])
        .map_err(|e| format!("resolver {resolver} answered a TXT query for {name}: {e}"))
}

/// The wildcard local address to bind before querying `resolver`, matching its
/// address family.
///
/// Built rather than parsed, so there is no fallible step and no `# Panics`
/// caveat on the query functions.
const fn unspecified_bind(resolver: SocketAddr) -> SocketAddr {
    match resolver {
        SocketAddr::V4(_) => {
            SocketAddr::new(std::net::IpAddr::V4(std::net::Ipv4Addr::UNSPECIFIED), 0)
        }
        SocketAddr::V6(_) => {
            SocketAddr::new(std::net::IpAddr::V6(std::net::Ipv6Addr::UNSPECIFIED), 0)
        }
    }
}

/// A per-query transaction id.
///
/// Not a security boundary — the query goes to an explicitly configured resolver
/// over a connected socket — but a distinct id per query means a late answer to
/// a previous query is rejected rather than mistaken for this one's.
#[allow(
    clippy::disallowed_methods,
    reason = "the id goes to a real DNS server and must differ across real processes"
)]
fn query_id() -> u16 {
    use std::sync::atomic::{AtomicU16, Ordering};
    static NEXT: AtomicU16 = AtomicU16::new(1);
    // Mix in the low bits of the clock so two processes (or a restart) do not
    // walk the same sequence.
    let counter = NEXT.fetch_add(1, Ordering::Relaxed);
    let clock = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map_or(0, |d| u16::try_from(d.subsec_nanos() & 0xFFFF).unwrap_or(0));
    counter ^ clock
}

/// Encode a `TXT`/`IN` query for `name`, as [`encode_query`] with `QTYPE_TXT`.
///
/// # Errors
///
/// As [`encode_query`].
pub fn encode_txt_query(id: u16, name: &str) -> Result<Vec<u8>, String> {
    encode_query(id, name, QTYPE_TXT, true)
}

/// Encode a DNS query for `name` of type `qtype`.
///
/// An EDNS0 `OPT` record advertising a 4096-byte buffer is always included, so a
/// record set that would not fit a bare 512-byte UDP answer comes back whole
/// instead of truncated (see [`MAX_RESPONSE`]).
///
/// `recursion_desired` is `false` for the authoritative probe: those servers are
/// authoritative for the name, so recursion is both unnecessary and usually
/// refused.
///
/// # Errors
///
/// Returns a message when a label is empty or longer than 63 bytes, or the whole
/// name exceeds 255 bytes.
pub fn encode_query(
    id: u16,
    name: &str,
    qtype: u16,
    recursion_desired: bool,
) -> Result<Vec<u8>, String> {
    let name = name.trim().trim_end_matches('.');
    if name.is_empty() {
        return Err("cannot query an empty DNS name".to_owned());
    }
    let mut out = Vec::with_capacity(HEADER_LEN + name.len() + 17);
    out.extend_from_slice(&id.to_be_bytes());
    let flags = if recursion_desired {
        FLAG_RECURSION_DESIRED
    } else {
        0
    };
    out.extend_from_slice(&flags.to_be_bytes());
    out.extend_from_slice(&1_u16.to_be_bytes()); // QDCOUNT
    out.extend_from_slice(&0_u16.to_be_bytes()); // ANCOUNT
    out.extend_from_slice(&0_u16.to_be_bytes()); // NSCOUNT
    out.extend_from_slice(&1_u16.to_be_bytes()); // ARCOUNT: the EDNS0 OPT below

    let mut encoded_len = 1; // the root label
    for label in name.split('.') {
        if label.is_empty() {
            return Err(format!("DNS name `{name}` has an empty label"));
        }
        let len = u8::try_from(label.len())
            .ok()
            .filter(|len| *len <= 63)
            .ok_or_else(|| format!("DNS label `{label}` is longer than 63 bytes"))?;
        encoded_len += 1 + label.len();
        out.push(len);
        out.extend_from_slice(label.as_bytes());
    }
    if encoded_len > 255 {
        return Err(format!(
            "DNS name `{name}` is longer than 255 bytes encoded"
        ));
    }
    out.push(0); // root label
    out.extend_from_slice(&qtype.to_be_bytes());
    out.extend_from_slice(&QCLASS_IN.to_be_bytes());

    // EDNS0 OPT (RFC 6891 §6.1.2): root name, type OPT, CLASS = the UDP payload
    // size we can accept, zero TTL/flags, zero RDLENGTH.
    out.push(0);
    out.extend_from_slice(&QTYPE_OPT.to_be_bytes());
    out.extend_from_slice(&u16::try_from(MAX_RESPONSE).unwrap_or(4096).to_be_bytes());
    out.extend_from_slice(&0_u32.to_be_bytes());
    out.extend_from_slice(&0_u16.to_be_bytes());
    Ok(out)
}

/// One resource record's payload, decoded for the types this client asks for.
#[derive(Debug, Clone, PartialEq, Eq)]
#[non_exhaustive]
pub enum Rdata {
    /// A `TXT` record's character-strings, concatenated (RFC 1035 §3.3.14).
    Txt(String),
    /// The domain name in an `NS` (or `CNAME`) record, decompressed.
    Name(String),
    /// An `A` record's address.
    A(std::net::Ipv4Addr),
    /// An `AAAA` record's address.
    Aaaa(std::net::Ipv6Addr),
    /// A record type this client does not decode.
    Other,
}

/// One decoded answer-section record.
#[derive(Debug, Clone, PartialEq, Eq)]
#[non_exhaustive]
pub struct ResourceRecord {
    /// The record's owner name, lowercased and without the trailing dot.
    pub name: String,
    /// The record type.
    pub rtype: u16,
    /// The decoded payload.
    pub rdata: Rdata,
}

/// A parsed DNS response.
#[derive(Debug, Clone, PartialEq, Eq)]
#[non_exhaustive]
pub struct DnsAnswer {
    /// The response code: `0` NOERROR, `3` NXDOMAIN, …
    pub rcode: u8,
    /// The answer-section records.
    pub records: Vec<ResourceRecord>,
}

impl DnsAnswer {
    /// The `TXT` values whose owner name is `name`.
    #[must_use]
    pub fn txt_values(&self, name: &str) -> Vec<String> {
        let wanted = normalize_name(name);
        self.records
            .iter()
            .filter(|r| r.name == wanted)
            .filter_map(|r| match &r.rdata {
                Rdata::Txt(value) => Some(value.clone()),
                _ => None,
            })
            .collect()
    }

    /// The `TXT` values at `name`, or at the end of the CNAME chain that
    /// starts at `name`, as a recursive resolver returns it.
    ///
    /// A tenant can delegate its ownership record with a CNAME. The answer
    /// then holds the CNAME at `name` and the TXT under the target, so
    /// [`txt_values`](Self::txt_values) alone finds nothing. The chain is
    /// bounded, so a CNAME loop cannot run forever.
    #[must_use]
    pub fn txt_values_via_cnames(&self, name: &str) -> Vec<String> {
        let mut owner = normalize_name(name);
        let mut values = self.txt_values(&owner);
        for _ in 0..MAX_CNAME_HOPS {
            let next = self.records.iter().find_map(|r| match &r.rdata {
                Rdata::Name(target) if r.rtype == QTYPE_CNAME && r.name == owner => {
                    Some(normalize_name(target))
                }
                _ => None,
            });
            let Some(next) = next else {
                break;
            };
            owner = next;
            values.extend(self.txt_values(&owner));
        }
        values
    }

    /// The end of the CNAME chain that starts at `name`, when the chain has
    /// at least one hop and the answer carries no TXT value along it.
    ///
    /// An authoritative server answers only for its own zone, so a CNAME to
    /// another zone arrives without the target's records.
    #[must_use]
    pub fn unresolved_cname_target(&self, name: &str) -> Option<String> {
        if !self.txt_values_via_cnames(name).is_empty() {
            return None;
        }
        let start = normalize_name(name);
        let mut owner = start.clone();
        for _ in 0..MAX_CNAME_HOPS {
            let next = self.records.iter().find_map(|r| match &r.rdata {
                Rdata::Name(target) if r.rtype == QTYPE_CNAME && r.name == owner => {
                    Some(normalize_name(target))
                }
                _ => None,
            });
            let Some(next) = next else {
                break;
            };
            owner = next;
        }
        (owner != start).then_some(owner)
    }

    /// The `NS` names in the answer.
    #[must_use]
    pub fn ns_names(&self) -> Vec<String> {
        self.records
            .iter()
            .filter(|r| r.rtype == QTYPE_NS)
            .filter_map(|r| match &r.rdata {
                Rdata::Name(name) => Some(name.clone()),
                _ => None,
            })
            .collect()
    }

    /// The `A` and `AAAA` addresses in the answer.
    #[must_use]
    pub fn ip_addrs(&self) -> Vec<std::net::IpAddr> {
        self.records
            .iter()
            .filter_map(|r| match &r.rdata {
                Rdata::A(addr) => Some(std::net::IpAddr::V4(*addr)),
                Rdata::Aaaa(addr) => Some(std::net::IpAddr::V6(*addr)),
                _ => None,
            })
            .collect()
    }

    /// The `A` addresses in the answer.
    #[must_use]
    pub fn a_addrs(&self) -> Vec<std::net::Ipv4Addr> {
        self.records
            .iter()
            .filter_map(|r| match &r.rdata {
                Rdata::A(addr) => Some(*addr),
                _ => None,
            })
            .collect()
    }
}

/// Lowercase a domain name and drop any trailing dot, so two spellings of the
/// same name compare equal.
fn normalize_name(name: &str) -> String {
    name.trim().trim_end_matches('.').to_ascii_lowercase()
}

/// Parse a DNS response into the TXT values it carries for `name`, following
/// a CNAME chain in the answer (see [`DnsAnswer::txt_values_via_cnames`]).
///
/// # Errors
///
/// As [`parse_response`].
pub fn parse_txt_response(id: u16, name: &str, msg: &[u8]) -> Result<TxtAnswer, String> {
    let answer = parse_response(id, name, msg)?;
    Ok(TxtAnswer {
        values: answer.txt_values_via_cnames(name),
        rcode: answer.rcode,
    })
}

/// Parse a DNS response to a query for `name`.
///
/// Validates that the message is a *response* to *this* query — the QR bit is
/// set, the transaction id matches, and the echoed question is the name that was
/// asked — before reading the answer section. Without those checks an unrelated
/// or reflected datagram would be read as propagation evidence.
///
/// # Errors
///
/// Returns a message for a malformed message, a mismatched id or question, a
/// truncated (`TC`) answer, or a server-side failure rcode. `NXDOMAIN` and an
/// empty `NOERROR` answer are **not** errors — they mean "not published yet".
pub fn parse_response(id: u16, name: &str, msg: &[u8]) -> Result<DnsAnswer, String> {
    if msg.len() < HEADER_LEN {
        return Err(format!(
            "the response is {} bytes, shorter than a DNS header",
            msg.len()
        ));
    }
    let response_id = u16::from_be_bytes([msg[0], msg[1]]);
    if response_id != id {
        return Err(format!(
            "transaction id {response_id:#06x} does not match the query's {id:#06x} (a late \
             answer to a previous query)"
        ));
    }
    let flags = u16::from_be_bytes([msg[2], msg[3]]);
    if flags & FLAG_RESPONSE == 0 {
        return Err("the datagram is a query, not a response".to_owned());
    }
    if flags & FLAG_TRUNCATED != 0 {
        return Err(
            "the answer was truncated (TC) even with EDNS0; the record set is too large for a \
             UDP answer"
                .to_owned(),
        );
    }
    let rcode = u8::try_from(flags & 0x000F).unwrap_or(0);
    match rcode {
        // NOERROR and NXDOMAIN both mean "the resolver answered"; whether the
        // value is there yet is the caller's decision.
        0 | 3 => {}
        2 => return Err("SERVFAIL — the zone's nameservers did not answer".to_owned()),
        5 => return Err("REFUSED — the resolver refused the query".to_owned()),
        other => return Err(format!("rcode {other}")),
    }
    let qdcount = u16::from_be_bytes([msg[4], msg[5]]);
    let ancount = u16::from_be_bytes([msg[6], msg[7]]);

    let mut offset = HEADER_LEN;
    for index in 0..qdcount {
        let (question, next) = read_name(msg, offset)?;
        // The echoed question must be the name that was asked. A server that
        // answers a different question is answering someone else's query.
        if index == 0 && question != normalize_name(name) {
            return Err(format!(
                "the response echoes the question `{question}`, not the queried `{}`",
                normalize_name(name)
            ));
        }
        offset = next
            .checked_add(4)
            .filter(|end| *end <= msg.len())
            .ok_or_else(|| "the question section is truncated".to_owned())?;
    }

    let mut records = Vec::new();
    for _ in 0..ancount {
        let (owner, next) = read_name(msg, offset)?;
        offset = next;
        let header_end = offset
            .checked_add(10)
            .filter(|end| *end <= msg.len())
            .ok_or_else(|| "an answer record header is truncated".to_owned())?;
        let rtype = u16::from_be_bytes([msg[offset], msg[offset + 1]]);
        let rdlength = usize::from(u16::from_be_bytes([msg[offset + 8], msg[offset + 9]]));
        let rdata_end = header_end
            .checked_add(rdlength)
            .filter(|end| *end <= msg.len())
            .ok_or_else(|| "an answer record's RDATA is truncated".to_owned())?;
        let rdata = match rtype {
            QTYPE_TXT => Rdata::Txt(decode_txt_rdata(&msg[header_end..rdata_end])?),
            // An NS target is a domain name in the message, so it may be
            // compressed against an earlier one — decode it against the WHOLE
            // message rather than the RDATA slice.
            QTYPE_NS | QTYPE_CNAME => Rdata::Name(read_name(msg, header_end)?.0),
            QTYPE_A if rdlength == 4 => Rdata::A(std::net::Ipv4Addr::new(
                msg[header_end],
                msg[header_end + 1],
                msg[header_end + 2],
                msg[header_end + 3],
            )),
            QTYPE_AAAA if rdlength == 16 => {
                let mut octets = [0_u8; 16];
                octets.copy_from_slice(&msg[header_end..rdata_end]);
                Rdata::Aaaa(std::net::Ipv6Addr::from(octets))
            }
            _ => Rdata::Other,
        };
        records.push(ResourceRecord {
            name: owner,
            rtype,
            rdata,
        });
        offset = rdata_end;
    }
    Ok(DnsAnswer { rcode, records })
}

/// Read a (possibly compressed) domain name, returning it and the offset just
/// past its encoding in the record stream.
///
/// Following a compression pointer does not advance the returned offset past the
/// two pointer bytes — that is what the record stream contains. The number of
/// pointers followed is bounded by [`MAX_NAME_POINTERS`], so a message whose
/// pointers form a cycle is rejected instead of looping forever.
fn read_name(msg: &[u8], start: usize) -> Result<(String, usize), String> {
    let mut labels: Vec<String> = Vec::new();
    let mut offset = start;
    let mut after: Option<usize> = None;
    let mut pointers = 0;
    loop {
        let len = *msg
            .get(offset)
            .ok_or_else(|| "a domain name runs past the end of the message".to_owned())?;
        if len & 0xC0 == 0xC0 {
            let low = *msg
                .get(offset + 1)
                .ok_or_else(|| "a compression pointer is truncated".to_owned())?;
            pointers += 1;
            if pointers > MAX_NAME_POINTERS {
                return Err("a domain name follows too many compression pointers".to_owned());
            }
            // The record stream continues after the pointer, wherever the name
            // itself is stored.
            after.get_or_insert(offset + 2);
            let target = usize::from(u16::from_be_bytes([len & 0x3F, low]));
            if target >= msg.len() || target >= offset {
                // A pointer must point BACKWARDS; anything else is malformed and
                // is the shape a decompression cycle takes.
                return Err("a compression pointer does not point backwards".to_owned());
            }
            offset = target;
            continue;
        }
        if len & 0xC0 != 0 {
            return Err(format!("unsupported DNS label type {:#04x}", len & 0xC0));
        }
        let label_start = offset + 1;
        let label_end = label_start
            .checked_add(usize::from(len))
            .filter(|end| *end <= msg.len())
            .ok_or_else(|| "a domain name label is truncated".to_owned())?;
        if len == 0 {
            return Ok((
                labels.join(".").to_ascii_lowercase(),
                after.unwrap_or(label_start),
            ));
        }
        labels.push(String::from_utf8_lossy(&msg[label_start..label_end]).into_owned());
        offset = label_end;
    }
}

/// Decode TXT RDATA: one or more length-prefixed character-strings,
/// concatenated (RFC 1035 §3.3.14 / RFC 7208 §3.3).
fn decode_txt_rdata(rdata: &[u8]) -> Result<String, String> {
    let mut out = String::new();
    let mut offset = 0;
    while offset < rdata.len() {
        let len = usize::from(rdata[offset]);
        let start = offset + 1;
        let end = start
            .checked_add(len)
            .filter(|end| *end <= rdata.len())
            .ok_or_else(|| "a TXT character-string runs past its RDATA".to_owned())?;
        out.push_str(&String::from_utf8_lossy(&rdata[start..end]));
        offset = end;
    }
    Ok(out)
}

/// Which of `expected` are not present in `observed`.
///
/// Pure, so the propagation decision is testable without a resolver.
#[must_use]
pub fn missing_values(expected: &[String], observed: &[String]) -> Vec<String> {
    expected
        .iter()
        .filter(|value| !observed.contains(value))
        .cloned()
        .collect()
}

/// Group records into `fqdn → expected values`, so an apex + wildcard order's
/// two values at one name are checked as a set.
#[must_use]
pub fn group_by_name(records: &[TxtRecord]) -> BTreeMap<String, Vec<String>> {
    let mut grouped: BTreeMap<String, Vec<String>> = BTreeMap::new();
    for record in records {
        let values = grouped.entry(record.fqdn.clone()).or_default();
        if !values.contains(&record.value) {
            values.push(record.value.clone());
        }
    }
    grouped
}

/// Why the propagation wait gave up, in operator-facing form.
///
/// Kept separate from the message so the wait's own logic is testable and the
/// wording lives in one place.
#[derive(Debug, Clone, PartialEq, Eq)]
#[non_exhaustive]
pub struct PropagationTimeout {
    /// The record name that never carried every expected value.
    pub fqdn: String,
    /// The specific values still missing.
    pub missing: Vec<String>,
    /// The resolver that still did not see them.
    pub resolver: String,
    /// What that resolver last returned for the name, or the error it gave.
    pub observed: String,
    /// The budget that elapsed, in seconds.
    pub waited_secs: u64,
}

impl std::fmt::Display for PropagationTimeout {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(
            f,
            "DNS-01 propagation timed out after {}s: the TXT record `{}` still does not carry {} \
             at resolver {} ({}). Check that the record was written to the zone that actually \
             serves this name and that its NS delegation is live, then raise \
             [server.tls.acme.dns] propagation_timeout_secs if the provider is simply slow",
            self.waited_secs,
            self.fqdn,
            self.missing
                .iter()
                .map(|v| format!("`{v}`"))
                .collect::<Vec<_>>()
                .join(", "),
            self.resolver,
            self.observed
        )
    }
}

/// Which nameservers to probe for each challenge name.
///
/// A multi-domain order spans as many zones as it has base domains, and one
/// zone's authoritative nameservers are not authoritative for another: asked
/// about a name they do not serve they answer REFUSED or a referral, never the
/// TXT value. Probing every name at one zone's servers therefore cannot
/// succeed — the wait burns its whole budget and then the caller deletes every
/// record, so an otherwise-correct multi-domain order fails every time.
///
/// Targets are held per name, with a `fallback` (the configured recursive
/// resolvers) used for any name whose authoritative set could not be
/// discovered. Discovery failing for one zone must not drag the others down to
/// the fallback (issue #1620).
#[derive(Debug, Clone, Default)]
pub struct ProbeTargets {
    /// Per challenge FQDN, in first-seen order. A `Vec` rather than a map: an
    /// order has a handful of names at most, and insertion order keeps the
    /// probe sequence (and so any timeout message) predictable.
    per_name: Vec<(String, Vec<SocketAddr>)>,
    fallback: Vec<SocketAddr>,
}

impl ProbeTargets {
    /// Targets that probe `fallback` for every name.
    #[must_use]
    pub fn flat(fallback: &[SocketAddr]) -> Self {
        Self {
            per_name: Vec::new(),
            fallback: fallback.to_vec(),
        }
    }

    /// Probe `servers` — the zone's own authoritative nameservers — for `fqdn`,
    /// replacing any previous entry for it.
    pub fn set_authoritative(&mut self, fqdn: &str, servers: Vec<SocketAddr>) {
        match self.per_name.iter_mut().find(|(name, _)| name == fqdn) {
            Some((_, existing)) => *existing = servers,
            None => self.per_name.push((fqdn.to_owned(), servers)),
        }
    }

    /// The servers to probe for `fqdn`, and whether they are authoritative for
    /// it.
    ///
    /// The flag decides the query's recursion-desired bit, and getting it wrong
    /// breaks the probe in one direction or the other. An authoritative server
    /// answers for its own zone with RD=0, which is what this wants: it reads
    /// the zone directly and cannot plant a negative cache entry. A *recursive*
    /// resolver asked with RD=0 answers only from cache — so a fallback probe
    /// with RD=0 gets an empty answer or `REFUSED` for a name nothing has looked
    /// up yet, forever, and a correctly published record never appears
    /// (issue #1620).
    #[must_use]
    pub fn for_name(&self, fqdn: &str) -> (&[SocketAddr], bool) {
        self.per_name
            .iter()
            .find(|(name, _)| name == fqdn)
            .map_or((self.fallback.as_slice(), false), |(_, servers)| {
                (servers.as_slice(), true)
            })
    }
}

/// Wait until every record in `records` is visible at every server probed for
/// its name, or the budget runs out.
///
/// # Errors
///
/// Returns a [`PropagationTimeout`] rendering naming the exact record, values
/// and resolver that never caught up.
pub async fn wait_for_propagation(
    records: &[TxtRecord],
    targets: &ProbeTargets,
    timeout: Duration,
    poll_interval: Duration,
    lookup: &dyn DnsLookup,
) -> Result<(), String> {
    if records.is_empty() {
        return Ok(());
    }
    let wanted = group_by_name(records);
    // Checked per name rather than once: a name can only be confirmed if
    // *something* answers for it, and with per-zone targets one name having no
    // server is possible while another has several.
    if let Some((fqdn, _)) = wanted
        .iter()
        .find(|(fqdn, _)| targets.for_name(fqdn).0.is_empty())
    {
        return Err(format!(
            "no nameserver to probe for {fqdn}, so DNS-01 propagation cannot be confirmed: \
             the zone's authoritative servers could not be discovered and \
             [server.tls.acme.dns] resolvers is empty"
        ));
    }
    let started = tokio::time::Instant::now();
    let deadline = started + timeout;
    let mut last_gap: Option<PropagationTimeout> = None;

    loop {
        let mut all_visible = true;
        'round: for (fqdn, expected) in &wanted {
            let (servers, authoritative) = targets.for_name(fqdn);
            for resolver in servers {
                // Recursion is desired only on the FALLBACK path. An
                // authoritative server answers for its own zone without it, and
                // asking it to recurse is meaningless. A recursive resolver
                // asked with RD=0 answers purely from cache, so a name it has
                // never looked up comes back empty or REFUSED no matter how
                // long the wait runs — the fallback must ask it to resolve.
                let (missing, observed) = match lookup
                    .query(*resolver, fqdn, QTYPE_TXT, !authoritative)
                    .await
                {
                    Ok(answer) => {
                        let values = answer.txt_values(fqdn);
                        let missing = missing_values(expected, &values);
                        let observed = if answer.rcode == 3 {
                            "the name does not exist there yet".to_owned()
                        } else {
                            format!("it currently returns {} TXT value(s)", values.len())
                        };
                        (missing, observed)
                    }
                    // A resolver error is a not-yet, not a hard failure: a
                    // freshly-created name often SERVFAILs while the zone
                    // catches up. It is recorded so the timeout can report it.
                    Err(e) => (expected.clone(), e),
                };
                if !missing.is_empty() {
                    last_gap = Some(PropagationTimeout {
                        fqdn: fqdn.clone(),
                        missing,
                        resolver: resolver.to_string(),
                        observed,
                        // Filled in below with the ELAPSED time; the loop may
                        // run one round past the deadline, so the configured
                        // budget would systematically understate the wait.
                        waited_secs: 0,
                    });
                    all_visible = false;
                    break 'round;
                }
            }
        }
        if all_visible {
            return Ok(());
        }
        // Re-check the budget only after a full round, so a wait configured with
        // a single-probe budget still probes once before giving up.
        let now = tokio::time::Instant::now();
        if now >= deadline {
            break;
        }
        // Loop back for another round even when the sleep crossed the deadline:
        // the budget is spent on probing, and the deadline check at the top of
        // the next iteration is what ends the wait.
        tokio::time::sleep(poll_interval.min(deadline - now)).await;
    }

    Err(last_gap.map_or_else(
        || "DNS-01 propagation could not be confirmed".to_owned(),
        |mut gap| {
            gap.waited_secs = started.elapsed().as_secs();
            gap.to_string()
        },
    ))
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::{Arc, Mutex};

    fn query_name(msg: &[u8]) -> String {
        let mut offset = HEADER_LEN;
        let mut labels = Vec::new();
        loop {
            let len = usize::from(msg[offset]);
            if len == 0 {
                break;
            }
            labels.push(String::from_utf8_lossy(&msg[offset + 1..offset + 1 + len]).into_owned());
            offset += 1 + len;
        }
        labels.join(".")
    }

    /// The question section of a query for `name`, with the header's counts
    /// rewritten for a response — the base every fake answer below is built on.
    ///
    /// Deliberately drops the query's EDNS0 OPT record: it lives in the
    /// ADDITIONAL section, so anything appended here must follow the question
    /// directly or the parser would read the OPT as the first answer.
    fn response_head(id: u16, name: &str, ancount: u16, rcode: u8) -> Vec<u8> {
        let query = encode_txt_query(id, name).expect("question encodes");
        // Header + QNAME + QTYPE/QCLASS, stopping before the OPT record.
        let mut end = HEADER_LEN;
        while query[end] != 0 {
            end += 1 + usize::from(query[end]);
        }
        end += 1 + 4;
        let mut msg = query[..end].to_vec();
        let flags: u16 = FLAG_RESPONSE | FLAG_RECURSION_DESIRED | u16::from(rcode);
        msg[2..4].copy_from_slice(&flags.to_be_bytes());
        msg[6..8].copy_from_slice(&ancount.to_be_bytes());
        msg[10..12].copy_from_slice(&0_u16.to_be_bytes()); // ARCOUNT: no OPT here
        msg
    }

    /// Append one answer record whose owner name is a compression pointer to the
    /// question's name — the shape a real resolver sends.
    fn push_answer(msg: &mut Vec<u8>, rtype: u16, rdata: &[u8]) {
        msg.extend_from_slice(&[0xC0, u8::try_from(HEADER_LEN).expect("header fits u8")]);
        msg.extend_from_slice(&rtype.to_be_bytes());
        msg.extend_from_slice(&QCLASS_IN.to_be_bytes());
        msg.extend_from_slice(&60_u32.to_be_bytes());
        msg.extend_from_slice(
            &u16::try_from(rdata.len())
                .expect("rdata fits u16")
                .to_be_bytes(),
        );
        msg.extend_from_slice(rdata);
    }

    /// Build a TXT answer for `name` carrying `values`.
    fn txt_response(id: u16, name: &str, values: &[&str], rcode: u8) -> Vec<u8> {
        let mut msg = response_head(
            id,
            name,
            u16::try_from(values.len()).expect("answer count fits u16"),
            rcode,
        );
        for value in values {
            let mut rdata = vec![u8::try_from(value.len()).expect("string fits u8")];
            rdata.extend_from_slice(value.as_bytes());
            push_answer(&mut msg, QTYPE_TXT, &rdata);
        }
        msg
    }

    /// Encode a domain name as uncompressed DNS labels.
    fn encode_labels(name: &str) -> Vec<u8> {
        let mut out = Vec::new();
        for label in name.split('.').filter(|l| !l.is_empty()) {
            out.push(u8::try_from(label.len()).expect("label fits u8"));
            out.extend_from_slice(label.as_bytes());
        }
        out.push(0);
        out
    }

    /// Build an `NS` answer for `zone` naming `servers`.
    fn ns_response(id: u16, zone: &str, servers: &[&str]) -> Vec<u8> {
        let mut msg = response_head(
            id,
            zone,
            u16::try_from(servers.len()).expect("answer count fits u16"),
            0,
        );
        for server in servers {
            push_answer(&mut msg, QTYPE_NS, &encode_labels(server));
        }
        msg
    }

    /// Build an `A` answer for `name` carrying `addrs`.
    fn a_response(id: u16, name: &str, addrs: &[[u8; 4]]) -> Vec<u8> {
        let mut msg = response_head(
            id,
            name,
            u16::try_from(addrs.len()).expect("answer count fits u16"),
            0,
        );
        for addr in addrs {
            push_answer(&mut msg, QTYPE_A, addr);
        }
        msg
    }

    #[test]
    fn a_query_encodes_the_name_as_labels() {
        let msg = encode_txt_query(0x1234, "_acme-challenge.myapp.com").expect("encodes");
        assert_eq!(&msg[0..2], &[0x12, 0x34]);
        assert_eq!(u16::from_be_bytes([msg[2], msg[3]]), FLAG_RECURSION_DESIRED);
        assert_eq!(u16::from_be_bytes([msg[4], msg[5]]), 1, "QDCOUNT");
        assert_eq!(query_name(&msg), "_acme-challenge.myapp.com");
        // QTYPE/QCLASS sit right after the root label, BEFORE the EDNS0 OPT
        // record the query now appends (see `a_query_advertises_an_edns0_buffer`).
        let qtype_at = msg.len() - 11 - 4;
        assert_eq!(&msg[qtype_at..qtype_at + 4], &[0, 16, 0, 1], "TXT/IN");
        // A trailing dot is the same name.
        assert_eq!(
            encode_txt_query(1, "myapp.com.").unwrap(),
            encode_txt_query(1, "myapp.com").unwrap()
        );
    }

    #[test]
    fn a_query_rejects_unencodable_names() {
        assert!(encode_txt_query(1, "  ").is_err());
        assert!(encode_txt_query(1, "a..b").is_err());
        assert!(encode_txt_query(1, &format!("{}.com", "x".repeat(64))).is_err());
        let long = std::iter::repeat_n("abcdefghij", 30)
            .collect::<Vec<_>>()
            .join(".");
        assert!(encode_txt_query(1, &long).is_err());
    }

    #[test]
    fn a_response_with_a_compression_pointer_parses() {
        let id = 0xABCD;
        let msg = txt_response(
            id,
            "_acme-challenge.myapp.com",
            &["value-one", "value-two"],
            0,
        );
        let answer = parse_txt_response(id, "_acme-challenge.myapp.com", &msg).expect("parses");
        assert_eq!(answer.rcode, 0);
        assert_eq!(answer.values, vec!["value-one", "value-two"]);
    }

    #[test]
    fn nxdomain_is_not_an_error_it_is_not_published_yet() {
        let id = 7;
        let msg = txt_response(id, "_acme-challenge.myapp.com", &[], 3);
        let answer = parse_txt_response(id, "_acme-challenge.myapp.com", &msg).expect("parses");
        assert!(answer.is_nxdomain());
        assert!(answer.values.is_empty());
    }

    #[test]
    fn server_failures_are_errors() {
        for (rcode, needle) in [(2_u8, "SERVFAIL"), (5, "REFUSED")] {
            let msg = txt_response(9, "x.myapp.com", &[], rcode);
            let err = parse_txt_response(9, "x.myapp.com", &msg)
                .expect_err("a server failure must surface");
            assert!(err.contains(needle), "got: {err}");
        }
    }

    #[test]
    fn a_mismatched_transaction_id_is_rejected() {
        let msg = txt_response(1, "x.myapp.com", &["v"], 0);
        let err = parse_txt_response(2, "x.myapp.com", &msg).expect_err("id must match");
        assert!(err.contains("transaction id"), "got: {err}");
    }

    #[test]
    fn a_truncated_answer_is_reported_rather_than_silently_short() {
        let mut msg = txt_response(3, "x.myapp.com", &["v"], 0);
        let flags = u16::from_be_bytes([msg[2], msg[3]]) | FLAG_TRUNCATED;
        msg[2..4].copy_from_slice(&flags.to_be_bytes());
        let err = parse_txt_response(3, "x.myapp.com", &msg).expect_err("TC must surface");
        assert!(err.contains("truncated"), "got: {err}");
    }

    #[test]
    fn malformed_messages_never_panic() {
        let good = txt_response(4, "x.myapp.com", &["v"], 0);
        // Every prefix of a well-formed message must be rejected, not panic.
        for len in 0..good.len() {
            let _ = parse_txt_response(4, "x.myapp.com", &good[..len]);
        }
        // A record claiming more RDATA than is present.
        let mut lying = good;
        let rdlen_at = lying.len() - 1 - 1 - "v".len();
        lying[rdlen_at..rdlen_at + 2].copy_from_slice(&9999_u16.to_be_bytes());
        assert!(parse_txt_response(4, "x.myapp.com", &lying).is_err());
    }

    // RFC 1035 §3.3.14: TXT RDATA is one or more character-strings, and a value
    // longer than 255 bytes arrives split. They concatenate.
    #[test]
    fn multi_string_txt_rdata_concatenates() {
        assert_eq!(decode_txt_rdata(&[3, b'a', b'b', b'c']).unwrap(), "abc");
        assert_eq!(
            decode_txt_rdata(&[2, b'a', b'b', 2, b'c', b'd']).unwrap(),
            "abcd"
        );
        assert_eq!(decode_txt_rdata(&[]).unwrap(), "");
        assert!(decode_txt_rdata(&[5, b'a']).is_err());
    }

    // EDNS0 (RFC 6891): without an OPT record advertising a bigger buffer, a
    // server caps the answer at 512 bytes and sets TC — which the propagation
    // wait would then retry for its whole budget against a record that IS
    // published.
    #[test]
    fn a_query_advertises_an_edns0_buffer() {
        let msg = encode_txt_query(0x1234, "_acme-challenge.myapp.com").expect("encodes");
        assert_eq!(u16::from_be_bytes([msg[10], msg[11]]), 1, "ARCOUNT");
        // The OPT record is the last 11 bytes: root name, type, class (= the
        // advertised UDP payload size), TTL, RDLENGTH.
        let opt = &msg[msg.len() - 11..];
        assert_eq!(opt[0], 0, "OPT's owner name is the root");
        assert_eq!(u16::from_be_bytes([opt[1], opt[2]]), QTYPE_OPT);
        assert_eq!(
            usize::from(u16::from_be_bytes([opt[3], opt[4]])),
            MAX_RESPONSE,
            "the advertised buffer must match what we actually read"
        );
    }

    // The propagation probe must not be recursive: it goes to the zone's own
    // nameservers, which answer for the name directly.
    #[test]
    fn recursion_can_be_turned_off() {
        let recursive = encode_query(1, "myapp.com", QTYPE_TXT, true).expect("encodes");
        let authoritative = encode_query(1, "myapp.com", QTYPE_TXT, false).expect("encodes");
        assert_eq!(
            u16::from_be_bytes([recursive[2], recursive[3]]),
            FLAG_RECURSION_DESIRED
        );
        assert_eq!(u16::from_be_bytes([authoritative[2], authoritative[3]]), 0);
    }

    // A datagram without the QR bit is a query, not an answer — reading one as
    // propagation evidence would let a reflected packet satisfy the wait.
    #[test]
    fn a_query_is_not_accepted_as_a_response() {
        let mut msg = txt_response(11, "x.myapp.com", &["v"], 0);
        let flags = u16::from_be_bytes([msg[2], msg[3]]) & !FLAG_RESPONSE;
        msg[2..4].copy_from_slice(&flags.to_be_bytes());
        let err = parse_response(11, "x.myapp.com", &msg).expect_err("QR must be set");
        assert!(err.contains("not a response"), "got: {err}");
    }

    // An answer to a DIFFERENT question is somebody else's answer.
    #[test]
    fn a_response_echoing_another_question_is_rejected() {
        let msg = txt_response(12, "other.myapp.com", &["v"], 0);
        let err = parse_response(12, "x.myapp.com", &msg).expect_err("question must match");
        assert!(err.contains("echoes the question"), "got: {err}");
    }

    // TXT records belonging to a different owner name in the same answer must
    // not count towards this name's propagation.
    #[test]
    fn txt_values_are_scoped_to_their_owner_name() {
        let answer = DnsAnswer {
            rcode: 0,
            records: vec![
                ResourceRecord {
                    name: "_acme-challenge.myapp.com".to_owned(),
                    rtype: QTYPE_TXT,
                    rdata: Rdata::Txt("mine".to_owned()),
                },
                ResourceRecord {
                    name: "_acme-challenge.other.com".to_owned(),
                    rtype: QTYPE_TXT,
                    rdata: Rdata::Txt("theirs".to_owned()),
                },
            ],
        };
        assert_eq!(
            answer.txt_values("_acme-challenge.myapp.com"),
            vec!["mine".to_owned()]
        );
        assert_eq!(
            answer.txt_values("_ACME-CHALLENGE.MyApp.com."),
            vec!["mine"]
        );
    }

    #[test]
    fn ns_and_a_records_decode() {
        let msg = ns_response(21, "myapp.com", &["ns1.provider.net", "ns2.provider.net"]);
        let answer = parse_response(21, "myapp.com", &msg).expect("parses");
        assert_eq!(
            answer.ns_names(),
            vec!["ns1.provider.net".to_owned(), "ns2.provider.net".to_owned()]
        );

        let msg = a_response(
            22,
            "ns1.provider.net",
            &[[192, 0, 2, 10], [198, 51, 100, 7]],
        );
        let answer = parse_response(22, "ns1.provider.net", &msg).expect("parses");
        assert_eq!(
            answer.a_addrs(),
            vec![
                std::net::Ipv4Addr::new(192, 0, 2, 10),
                std::net::Ipv4Addr::new(198, 51, 100, 7)
            ]
        );
    }

    // Decoding a compressed name is the one place a hostile message can loop
    // forever. A pointer must point strictly backwards, and the hop count is
    // bounded either way.
    #[test]
    fn a_self_referential_compression_pointer_is_rejected_not_looped() {
        // A pointer at offset 12 that points at itself.
        let mut msg = vec![0_u8; 14];
        msg[12] = 0xC0;
        msg[13] = 12;
        let err = read_name(&msg, 12).expect_err("a self-pointer must be rejected");
        assert!(err.contains("backwards"), "got: {err}");

        // …and one pointing forwards, the other half of the same cycle.
        let mut msg = vec![0_u8; 32];
        msg[12] = 0xC0;
        msg[13] = 20;
        msg[20] = 0xC0;
        msg[21] = 12;
        assert!(read_name(&msg, 12).is_err());
    }

    #[test]
    fn a_backwards_compression_pointer_resolves_and_advances_past_the_pointer() {
        // "myapp.com" at offset 12, then a pointer to it at offset 23.
        let mut msg = vec![0_u8; 12];
        msg.extend_from_slice(&encode_labels("myapp.com"));
        let pointer_at = msg.len();
        msg.extend_from_slice(&[0xC0, 12]);
        let (name, next) = read_name(&msg, pointer_at).expect("resolves");
        assert_eq!(name, "myapp.com");
        assert_eq!(
            next,
            pointer_at + 2,
            "the record stream continues after the two pointer bytes"
        );
    }

    /// A lookup that answers `NS` and `A` from a script, so authoritative
    /// discovery can be driven without a network.
    struct DiscoveryLookup {
        ns: std::collections::HashMap<String, Vec<String>>,
        a: std::collections::HashMap<String, Vec<std::net::Ipv4Addr>>,
        asked: Mutex<Vec<(String, u16)>>,
    }

    impl DiscoveryLookup {
        fn new(ns: &[(&str, &[&str])], a: &[(&str, &[[u8; 4]])]) -> Arc<Self> {
            Arc::new(Self {
                ns: ns
                    .iter()
                    .map(|(zone, servers)| {
                        (
                            (*zone).to_owned(),
                            servers.iter().map(|s| (*s).to_owned()).collect(),
                        )
                    })
                    .collect(),
                a: a.iter()
                    .map(|(name, addrs)| {
                        (
                            (*name).to_owned(),
                            addrs
                                .iter()
                                .map(|o| std::net::Ipv4Addr::new(o[0], o[1], o[2], o[3]))
                                .collect(),
                        )
                    })
                    .collect(),
                asked: Mutex::new(Vec::new()),
            })
        }
    }

    impl DnsLookup for DiscoveryLookup {
        fn query<'a>(
            &'a self,
            _server: SocketAddr,
            name: &'a str,
            qtype: u16,
            _recursion_desired: bool,
        ) -> BoxFuture<'a, Result<DnsAnswer, String>> {
            self.asked.lock().unwrap().push((name.to_owned(), qtype));
            let owner = normalize_name(name);
            let records = match qtype {
                QTYPE_NS => self
                    .ns
                    .get(&owner)
                    .map(|servers| {
                        servers
                            .iter()
                            .map(|s| ResourceRecord {
                                name: owner.clone(),
                                rtype: QTYPE_NS,
                                rdata: Rdata::Name(s.clone()),
                            })
                            .collect()
                    })
                    .unwrap_or_default(),
                QTYPE_A => self
                    .a
                    .get(&owner)
                    .map(|addrs| {
                        addrs
                            .iter()
                            .map(|addr| ResourceRecord {
                                name: owner.clone(),
                                rtype: QTYPE_A,
                                rdata: Rdata::A(*addr),
                            })
                            .collect()
                    })
                    .unwrap_or_default(),
                _ => Vec::new(),
            };
            Box::pin(async move {
                Ok(DnsAnswer {
                    rcode: if records.is_empty() { 3 } else { 0 },
                    records,
                })
            })
        }
    }

    // The whole point of the authoritative probe: the zone's own nameservers are
    // found from the challenge FQDN, with the `_acme-challenge` label dropped and
    // the NS names resolved to addresses.
    #[tokio::test]
    async fn authoritative_discovery_finds_the_zones_nameservers() {
        let lookup = DiscoveryLookup::new(
            &[("myapp.com", &["ns1.provider.net", "ns2.provider.net"])],
            &[
                ("ns1.provider.net", &[[192, 0, 2, 10]]),
                ("ns2.provider.net", &[[198, 51, 100, 7]]),
            ],
        );
        let found = authoritative_resolvers(
            "_acme-challenge.myapp.com",
            &[resolver(53)],
            lookup.as_ref(),
        )
        .await;
        assert_eq!(
            found,
            vec![
                SocketAddr::from(([192, 0, 2, 10], 53)),
                SocketAddr::from(([198, 51, 100, 7], 53)),
            ]
        );
        // The challenge name is TRIED first — it can itself be a delegated zone
        // (#1620) — and when it is not, the walk asks about the zone that
        // actually holds the record.
        let asked = lookup.asked.lock().unwrap().clone();
        assert!(
            asked.contains(&("myapp.com".to_owned(), QTYPE_NS)),
            "got: {asked:?}"
        );
        assert!(
            asked.contains(&("_acme-challenge.myapp.com".to_owned(), QTYPE_NS)),
            "the challenge name must be offered as the most specific zone candidate: {asked:?}"
        );
    }

    // A delegated sub-zone wins over its parent: the most specific suffix that
    // answers NS is the zone that actually serves the record.
    #[tokio::test]
    async fn discovery_prefers_the_most_specific_delegated_zone() {
        let lookup = DiscoveryLookup::new(
            &[
                ("tenants.myapp.com", &["ns1.sub.net"]),
                ("myapp.com", &["ns1.parent.net"]),
            ],
            &[
                ("ns1.sub.net", &[[192, 0, 2, 1]]),
                ("ns1.parent.net", &[[192, 0, 2, 2]]),
            ],
        );
        let found = authoritative_resolvers(
            "_acme-challenge.tenants.myapp.com",
            &[resolver(53)],
            lookup.as_ref(),
        )
        .await;
        assert_eq!(found, vec![SocketAddr::from(([192, 0, 2, 1], 53))]);
    }

    /// Regression (#1620): `_acme-challenge.<domain>` can be a delegated zone
    /// with its own NS records — a recommended way to scope an ACME credential
    /// to nothing but the challenge name.
    ///
    /// Discovery used to strip the label unconditionally, so it asked the
    /// PARENT's nameservers. Those answer a referral rather than the TXT value,
    /// and the probe is sent with recursion disabled, so a correctly published
    /// record could only ever time out.
    #[tokio::test]
    async fn discovery_finds_a_delegated_challenge_zone() {
        let lookup = DiscoveryLookup::new(
            &[
                ("_acme-challenge.myapp.com", &["ns1.challenge.net"]),
                ("myapp.com", &["ns1.parent.net"]),
            ],
            &[
                ("ns1.challenge.net", &[[192, 0, 2, 9]]),
                ("ns1.parent.net", &[[192, 0, 2, 2]]),
            ],
        );
        let found = authoritative_resolvers(
            "_acme-challenge.myapp.com",
            &[resolver(53)],
            lookup.as_ref(),
        )
        .await;
        assert_eq!(
            found,
            vec![SocketAddr::from(([192, 0, 2, 9], 53))],
            "the delegated challenge zone's own nameserver must win over the parent's"
        );
    }

    /// …and when `_acme-challenge` is NOT delegated (the ordinary case), the
    /// walk falls through to the parent zone exactly as before.
    #[tokio::test]
    async fn discovery_falls_through_to_the_parent_when_the_label_is_not_delegated() {
        let lookup = DiscoveryLookup::new(
            &[("myapp.com", &["ns1.parent.net"])],
            &[("ns1.parent.net", &[[192, 0, 2, 2]])],
        );
        let found = authoritative_resolvers(
            "_acme-challenge.myapp.com",
            &[resolver(53)],
            lookup.as_ref(),
        )
        .await;
        assert_eq!(found, vec![SocketAddr::from(([192, 0, 2, 2], 53))]);
    }

    // Discovery is best-effort: when it finds nothing the caller falls back to
    // the configured resolvers rather than failing the order.
    #[tokio::test]
    async fn discovery_returns_empty_when_nothing_answers() {
        let lookup = DiscoveryLookup::new(&[], &[]);
        assert!(
            authoritative_resolvers(
                "_acme-challenge.myapp.com",
                &[resolver(53)],
                lookup.as_ref()
            )
            .await
            .is_empty()
        );
        // …and an NS answer whose names do not resolve is also a miss.
        let lookup = DiscoveryLookup::new(&[("myapp.com", &["ns1.provider.net"])], &[]);
        assert!(
            authoritative_resolvers(
                "_acme-challenge.myapp.com",
                &[resolver(53)],
                lookup.as_ref()
            )
            .await
            .is_empty()
        );
    }

    #[test]
    fn missing_values_compares_sets() {
        let expected = vec!["a".to_owned(), "b".to_owned()];
        assert!(
            missing_values(&expected, &["a".to_owned(), "b".to_owned(), "z".to_owned()]).is_empty()
        );
        assert_eq!(
            missing_values(&expected, &["a".to_owned()]),
            vec!["b".to_owned()]
        );
    }

    // An apex + wildcard order publishes two DIFFERENT values at ONE name; the
    // wait must require both before telling the CA to validate.
    #[test]
    fn grouping_merges_two_values_at_one_name() {
        let records = vec![
            TxtRecord::new("myapp.com", "value-apex"),
            TxtRecord::new("myapp.com", "value-wildcard"),
        ];
        let grouped = group_by_name(&records);
        assert_eq!(grouped.len(), 1);
        assert_eq!(
            grouped["_acme-challenge.myapp.com"],
            vec!["value-apex".to_owned(), "value-wildcard".to_owned()]
        );
    }

    /// A scripted lookup: a STABLE answer per resolver, repeated on every round.
    ///
    /// Keyed per resolver rather than a shared queue, because the wait polls in
    /// rounds — a queue would hand round two whatever round one did not consume,
    /// so "this resolver is the lagging one" could not be expressed at all.
    /// A resolver with no scripted answer returns NXDOMAIN (not published yet).
    struct ScriptedLookup {
        answers: std::collections::HashMap<String, Result<TxtAnswer, String>>,
        calls: Mutex<Vec<(String, String)>>,
    }

    impl ScriptedLookup {
        fn new(answers: Vec<(SocketAddr, Result<TxtAnswer, String>)>) -> Arc<Self> {
            Arc::new(Self {
                answers: answers
                    .into_iter()
                    .map(|(addr, answer)| (addr.to_string(), answer))
                    .collect(),
                calls: Mutex::new(Vec::new()),
            })
        }

        fn call_count(&self) -> usize {
            self.calls.lock().unwrap().len()
        }
    }

    impl DnsLookup for ScriptedLookup {
        fn query<'a>(
            &'a self,
            server: SocketAddr,
            name: &'a str,
            qtype: u16,
            _recursion_desired: bool,
        ) -> BoxFuture<'a, Result<DnsAnswer, String>> {
            self.calls
                .lock()
                .unwrap()
                .push((server.to_string(), name.to_owned()));
            // These tests script TXT answers only; discovery queries answer
            // empty so `authoritative_resolvers` falls back, which is the path
            // under test here.
            if qtype != QTYPE_TXT {
                return Box::pin(async move {
                    Ok(DnsAnswer {
                        rcode: 0,
                        records: Vec::new(),
                    })
                });
            }
            let owner = normalize_name(name);
            let answer = self
                .answers
                .get(&server.to_string())
                .cloned()
                .unwrap_or_else(|| {
                    Ok(TxtAnswer {
                        values: Vec::new(),
                        rcode: 3,
                    })
                })
                .map(|txt| DnsAnswer {
                    rcode: txt.rcode,
                    records: txt
                        .values
                        .into_iter()
                        .map(|value| ResourceRecord {
                            name: owner.clone(),
                            rtype: QTYPE_TXT,
                            rdata: Rdata::Txt(value),
                        })
                        .collect(),
                });
            Box::pin(async move { answer })
        }
    }

    fn resolver(port: u16) -> SocketAddr {
        SocketAddr::from(([127, 0, 0, 1], port))
    }

    // #2642: the ownership lookup merges what every answering server saw, and
    // one resolver failing does not hide another's answer.
    #[tokio::test]
    async fn txt_values_merge_every_answer_and_survive_one_failing_resolver() {
        let lookup = ScriptedLookup::new(vec![
            (resolver(53), Err("timed out".to_owned())),
            (
                resolver(5353),
                Ok(TxtAnswer {
                    values: vec!["token-b".to_owned(), "v=spf1 -all".to_owned()],
                    rcode: 0,
                }),
            ),
        ]);
        let values = txt_values(
            "_autumn-challenge.app.clientco.com",
            &[resolver(53), resolver(5353)],
            lookup.as_ref(),
            DEADLINE,
        )
        .await
        .unwrap();
        assert_eq!(values, vec!["token-b".to_owned(), "v=spf1 -all".to_owned()]);
    }

    // #2642: a TXT record delegated by CNAME is read at the chain's end.
    #[test]
    fn txt_values_follow_a_cname_chain_in_the_answer() {
        let record = |name: &str, rtype: u16, rdata: Rdata| ResourceRecord {
            name: name.to_owned(),
            rtype,
            rdata,
        };
        let answer = DnsAnswer {
            rcode: 0,
            records: vec![
                record(
                    "_autumn-challenge.app.clientco.com",
                    QTYPE_CNAME,
                    Rdata::Name("verify.dns-host.net".to_owned()),
                ),
                record(
                    "verify.dns-host.net",
                    QTYPE_TXT,
                    Rdata::Txt("token-b".to_owned()),
                ),
                record(
                    "unrelated.example",
                    QTYPE_TXT,
                    Rdata::Txt("other".to_owned()),
                ),
            ],
        };
        assert_eq!(
            answer.txt_values_via_cnames("_autumn-challenge.app.clientco.com."),
            vec!["token-b".to_owned()]
        );

        // A loop ends at the hop bound instead of spinning.
        let looped = DnsAnswer {
            rcode: 0,
            records: vec![
                record("a.test", QTYPE_CNAME, Rdata::Name("b.test".to_owned())),
                record("b.test", QTYPE_CNAME, Rdata::Name("a.test".to_owned())),
            ],
        };
        assert!(looped.txt_values_via_cnames("a.test").is_empty());
    }

    // The wire parser decodes a CNAME's target, so the chain can be walked.
    #[test]
    fn a_cname_answer_record_decodes_its_target() {
        let name = "_autumn-challenge.app.clientco.com";
        let mut msg = response_head(0x4242, name, 1, 0);
        push_answer(&mut msg, QTYPE_CNAME, &encode_labels("verify.dns-host.net"));
        let answer = parse_response(0x4242, name, &msg).unwrap();
        assert_eq!(
            answer.records[0].rdata,
            Rdata::Name("verify.dns-host.net".to_owned())
        );
    }

    #[tokio::test]
    async fn txt_values_fail_only_when_no_server_answers() {
        let lookup = ScriptedLookup::new(vec![(resolver(53), Err("REFUSED".to_owned()))]);
        let err = txt_values(
            "_autumn-challenge.app.clientco.com",
            &[resolver(53)],
            lookup.as_ref(),
            DEADLINE,
        )
        .await
        .unwrap_err();
        assert!(err.contains("REFUSED"), "{err}");

        // NXDOMAIN is an answer: nothing is published yet.
        let lookup = ScriptedLookup::new(vec![]);
        let values = txt_values(
            "_autumn-challenge.app.clientco.com",
            &[resolver(53)],
            lookup.as_ref(),
            DEADLINE,
        )
        .await
        .unwrap();
        assert!(values.is_empty());
    }

    const DEADLINE: Duration = Duration::from_secs(6);

    /// Answers every TXT query after `txt_delay` with one value naming the
    /// server, and every discovery query after `discovery_delay` with nothing.
    struct SlowLookup {
        txt_delay: Duration,
        discovery_delay: Duration,
    }

    impl DnsLookup for SlowLookup {
        fn query<'a>(
            &'a self,
            server: SocketAddr,
            name: &'a str,
            qtype: u16,
            _recursion_desired: bool,
        ) -> BoxFuture<'a, Result<DnsAnswer, String>> {
            Box::pin(async move {
                if qtype != QTYPE_TXT {
                    tokio::time::sleep(self.discovery_delay).await;
                    return Ok(DnsAnswer {
                        rcode: 0,
                        records: Vec::new(),
                    });
                }
                tokio::time::sleep(self.txt_delay).await;
                Ok(DnsAnswer {
                    rcode: 0,
                    records: vec![ResourceRecord {
                        name: normalize_name(name),
                        rtype: QTYPE_TXT,
                        rdata: Rdata::Txt(format!("from-{}", server.port())),
                    }],
                })
            })
        }
    }

    // #2642: slow servers are asked at the same time, so one lookup costs one
    // server timeout, not one per server.
    #[tokio::test(start_paused = true)]
    async fn txt_values_ask_every_server_at_once() {
        let lookup = SlowLookup {
            txt_delay: Duration::from_secs(3),
            discovery_delay: Duration::ZERO,
        };
        let started = tokio::time::Instant::now();
        let values = txt_values(
            "_autumn-challenge.app.clientco.com",
            &[resolver(53), resolver(54), resolver(55), resolver(56)],
            &lookup,
            DEADLINE,
        )
        .await
        .unwrap();
        assert_eq!(values.len(), 4, "{values:?}");
        assert!(
            started.elapsed() < Duration::from_secs(4),
            "took {:?}",
            started.elapsed()
        );
    }

    // #2642: a discovery that stalls stops at the deadline, and the recursive
    // resolvers' answers still count.
    #[tokio::test(start_paused = true)]
    async fn a_stalled_discovery_stops_at_the_deadline_and_keeps_recursive_answers() {
        let lookup = SlowLookup {
            txt_delay: Duration::ZERO,
            discovery_delay: Duration::from_secs(3600),
        };
        let started = tokio::time::Instant::now();
        let values = txt_values(
            "_autumn-challenge.app.clientco.com",
            &[resolver(53)],
            &lookup,
            DEADLINE,
        )
        .await
        .unwrap();
        assert_eq!(values, vec!["from-53".to_owned()]);
        assert!(
            started.elapsed() <= DEADLINE,
            "took {:?}",
            started.elapsed()
        );
    }

    /// A tenant delegated its ownership record by CNAME after a recursive
    /// resolver cached NXDOMAIN. Authoritative servers answer each name only
    /// with what they own; recursive servers still answer NXDOMAIN.
    struct DelegatedLookup;

    impl DnsLookup for DelegatedLookup {
        fn query<'a>(
            &'a self,
            _server: SocketAddr,
            name: &'a str,
            qtype: u16,
            recursion_desired: bool,
        ) -> BoxFuture<'a, Result<DnsAnswer, String>> {
            let owner = normalize_name(name);
            let record = |rtype: u16, rdata: Rdata| ResourceRecord {
                name: owner.clone(),
                rtype,
                rdata,
            };
            let records = match (qtype, recursion_desired, owner.as_str()) {
                (QTYPE_NS, _, _) => vec![record(QTYPE_NS, Rdata::Name("ns.test".to_owned()))],
                (QTYPE_A, _, "ns.test") => {
                    vec![record(
                        QTYPE_A,
                        Rdata::A(std::net::Ipv4Addr::new(192, 0, 2, 1)),
                    )]
                }
                (QTYPE_TXT, false, "_autumn-challenge.app.clientco.com") => vec![record(
                    QTYPE_CNAME,
                    Rdata::Name("verify.dns-host.net".to_owned()),
                )],
                (QTYPE_TXT, false, "verify.dns-host.net") => {
                    vec![record(QTYPE_TXT, Rdata::Txt("token-b".to_owned()))]
                }
                _ => Vec::new(),
            };
            let rcode = if records.is_empty() { 3 } else { 0 };
            Box::pin(async move { Ok(DnsAnswer { rcode, records }) })
        }
    }

    // #2642: a CNAME target missing from the authoritative answer is asked of
    // its own zone's servers, so a stale recursive NXDOMAIN does not hide it.
    #[tokio::test]
    async fn txt_values_ask_a_cname_target_of_its_own_nameservers() {
        let values = txt_values(
            "_autumn-challenge.app.clientco.com",
            &[resolver(53)],
            &DelegatedLookup,
            DEADLINE,
        )
        .await
        .unwrap();
        assert_eq!(values, vec!["token-b".to_owned()]);
    }

    /// `DelegatedLookup` with only `clientco.com` and `dns-host.net` as zones,
    /// and resolver 1 dropping every packet. Each answer takes 10ms.
    struct DeadDelegatedLookup;

    impl DnsLookup for DeadDelegatedLookup {
        fn query<'a>(
            &'a self,
            server: SocketAddr,
            name: &'a str,
            qtype: u16,
            recursion_desired: bool,
        ) -> BoxFuture<'a, Result<DnsAnswer, String>> {
            Box::pin(async move {
                if server == resolver(1) {
                    tokio::time::sleep(Duration::from_secs(3)).await;
                    return Err(format!("resolver {server} did not answer"));
                }
                tokio::time::sleep(Duration::from_millis(10)).await;
                let owner = normalize_name(name);
                if qtype == QTYPE_NS && owner != "clientco.com" && owner != "dns-host.net" {
                    return Ok(DnsAnswer {
                        rcode: 0,
                        records: Vec::new(),
                    });
                }
                DelegatedLookup
                    .query(server, name, qtype, recursion_desired)
                    .await
            })
        }
    }

    // Codex review on #2936: a resolver that drops packets costs one timeout
    // for the whole CNAME chain, not one per hop.
    #[tokio::test(start_paused = true)]
    async fn a_dead_resolver_costs_one_timeout_across_cname_hops() {
        let values = txt_values(
            "_autumn-challenge.app.clientco.com",
            &[resolver(1), resolver(2)],
            &DeadDelegatedLookup,
            DEADLINE,
        )
        .await
        .unwrap();
        assert_eq!(values, vec!["token-b".to_owned()]);
    }

    /// Resolver 1 and nameserver `ns2.test` drop every packet. `ns1.test`
    /// answers the ownership record with a CNAME to `dns-host.net`, whose
    /// nameserver has the token. Resolver 2 still caches NXDOMAIN. Each answer
    /// takes 10ms.
    struct SilentPeerLookup;

    impl DnsLookup for SilentPeerLookup {
        fn query<'a>(
            &'a self,
            server: SocketAddr,
            name: &'a str,
            qtype: u16,
            recursion_desired: bool,
        ) -> BoxFuture<'a, Result<DnsAnswer, String>> {
            Box::pin(async move {
                if server == resolver(1) || server == SocketAddr::from(([192, 0, 2, 2], 53)) {
                    tokio::time::sleep(Duration::from_secs(3)).await;
                    return Err(format!("{server} did not answer"));
                }
                tokio::time::sleep(Duration::from_millis(10)).await;
                let owner = normalize_name(name);
                let record = |rtype: u16, rdata: Rdata| ResourceRecord {
                    name: owner.clone(),
                    rtype,
                    rdata,
                };
                let ns = |host: &str| record(QTYPE_NS, Rdata::Name(host.to_owned()));
                let a =
                    |last: u8| record(QTYPE_A, Rdata::A(std::net::Ipv4Addr::new(192, 0, 2, last)));
                let records = match (qtype, recursion_desired, owner.as_str()) {
                    (QTYPE_NS, _, "clientco.com") => vec![ns("ns1.test"), ns("ns2.test")],
                    (QTYPE_NS, _, "dns-host.net") => vec![ns("ns3.test")],
                    (QTYPE_A, _, "ns1.test") => vec![a(1)],
                    (QTYPE_A, _, "ns2.test") => vec![a(2)],
                    (QTYPE_A, _, "ns3.test") => vec![a(3)],
                    (QTYPE_TXT, false, "_autumn-challenge.app.clientco.com") => vec![record(
                        QTYPE_CNAME,
                        Rdata::Name("verify.dns-host.net".to_owned()),
                    )],
                    (QTYPE_TXT, false, "verify.dns-host.net") => {
                        vec![record(QTYPE_TXT, Rdata::Txt("token-b".to_owned()))]
                    }
                    _ => Vec::new(),
                };
                Ok(DnsAnswer { rcode: 0, records })
            })
        }
    }

    // Codex review on #2936: a CNAME target is asked at once, not after every
    // nameserver of the current hop has answered.
    #[tokio::test(start_paused = true)]
    async fn a_cname_target_is_asked_without_waiting_for_a_silent_nameserver() {
        let values = txt_values(
            "_autumn-challenge.app.clientco.com",
            &[resolver(1), resolver(2)],
            &SilentPeerLookup,
            DEADLINE,
        )
        .await
        .unwrap();
        assert_eq!(values, vec!["token-b".to_owned()]);
    }

    /// Resolver 1 drops every packet. Resolver 2 answers, but still caches
    /// NXDOMAIN for the ownership record. Only the zone's own nameserver has
    /// the token.
    struct PartlyDeadLookup;

    impl DnsLookup for PartlyDeadLookup {
        fn query<'a>(
            &'a self,
            server: SocketAddr,
            name: &'a str,
            qtype: u16,
            recursion_desired: bool,
        ) -> BoxFuture<'a, Result<DnsAnswer, String>> {
            Box::pin(async move {
                if server == resolver(1) {
                    tokio::time::sleep(Duration::from_secs(3)).await;
                    return Err(format!("resolver {server} did not answer"));
                }
                let owner = normalize_name(name);
                let record = |rtype: u16, rdata: Rdata| ResourceRecord {
                    name: owner.clone(),
                    rtype,
                    rdata,
                };
                let records = match (qtype, recursion_desired, owner.as_str()) {
                    (QTYPE_NS, _, "clientco.com") => {
                        vec![record(QTYPE_NS, Rdata::Name("ns.test".to_owned()))]
                    }
                    (QTYPE_A, _, "ns.test") => {
                        vec![record(
                            QTYPE_A,
                            Rdata::A(std::net::Ipv4Addr::new(192, 0, 2, 1)),
                        )]
                    }
                    (QTYPE_TXT, false, "_autumn-challenge.app.clientco.com") => {
                        vec![record(QTYPE_TXT, Rdata::Txt("token-b".to_owned()))]
                    }
                    _ => Vec::new(),
                };
                let rcode = if records.is_empty() { 3 } else { 0 };
                Ok(DnsAnswer { rcode, records })
            })
        }
    }

    // Codex review on #2936: a resolver that drops packets costs discovery one
    // timeout in total, not one per zone candidate, so the zone's nameserver
    // is still reached inside the deadline.
    #[tokio::test(start_paused = true)]
    async fn a_dead_resolver_does_not_stall_discovery_past_the_deadline() {
        let started = tokio::time::Instant::now();
        let found = authoritative_resolvers(
            "_autumn-challenge.app.clientco.com",
            &[resolver(1), resolver(2)],
            &PartlyDeadLookup,
        )
        .await;
        assert_eq!(found, vec![SocketAddr::from(([192, 0, 2, 1], 53))]);
        assert!(
            started.elapsed() <= Duration::from_secs(3),
            "took {:?}",
            started.elapsed()
        );

        let values = txt_values(
            "_autumn-challenge.app.clientco.com",
            &[resolver(1), resolver(2)],
            &PartlyDeadLookup,
            DEADLINE,
        )
        .await
        .unwrap();
        assert_eq!(values, vec!["token-b".to_owned()]);
    }

    /// Resolver 1 drops every packet, and so does the zone's second
    /// nameserver. The first nameserver has the token at once; resolver 2
    /// still caches NXDOMAIN.
    struct SlowAuthoritativeLookup;

    impl DnsLookup for SlowAuthoritativeLookup {
        fn query<'a>(
            &'a self,
            server: SocketAddr,
            name: &'a str,
            qtype: u16,
            recursion_desired: bool,
        ) -> BoxFuture<'a, Result<DnsAnswer, String>> {
            Box::pin(async move {
                if server == resolver(1) {
                    tokio::time::sleep(Duration::from_secs(3)).await;
                    return Err(format!("resolver {server} did not answer"));
                }
                if server == SocketAddr::from(([192, 0, 2, 2], 53)) {
                    tokio::time::sleep(Duration::from_secs(60)).await;
                    return Err(format!("resolver {server} did not answer"));
                }
                let owner = normalize_name(name);
                let record = |rtype: u16, rdata: Rdata| ResourceRecord {
                    name: owner.clone(),
                    rtype,
                    rdata,
                };
                let records = match (qtype, recursion_desired, owner.as_str()) {
                    (QTYPE_NS, _, "clientco.com") => vec![
                        record(QTYPE_NS, Rdata::Name("ns1.test".to_owned())),
                        record(QTYPE_NS, Rdata::Name("ns2.test".to_owned())),
                    ],
                    (QTYPE_A, _, "ns1.test") => {
                        vec![record(
                            QTYPE_A,
                            Rdata::A(std::net::Ipv4Addr::new(192, 0, 2, 1)),
                        )]
                    }
                    (QTYPE_A, _, "ns2.test") => {
                        vec![record(
                            QTYPE_A,
                            Rdata::A(std::net::Ipv4Addr::new(192, 0, 2, 2)),
                        )]
                    }
                    (QTYPE_TXT, false, "_autumn-challenge.app.clientco.com") => {
                        vec![record(QTYPE_TXT, Rdata::Txt("token-b".to_owned()))]
                    }
                    _ => Vec::new(),
                };
                let rcode = if records.is_empty() { 3 } else { 0 };
                Ok(DnsAnswer { rcode, records })
            })
        }
    }

    // Codex review on #2936: an answer that arrived before the deadline counts,
    // even when another authoritative server is still silent at the deadline.
    #[tokio::test(start_paused = true)]
    async fn answers_in_before_the_deadline_survive_a_silent_nameserver() {
        let values = txt_values(
            "_autumn-challenge.app.clientco.com",
            &[resolver(1), resolver(2)],
            &SlowAuthoritativeLookup,
            DEADLINE,
        )
        .await
        .unwrap();
        assert_eq!(values, vec!["token-b".to_owned()]);
    }

    #[test]
    fn an_aaaa_answer_record_decodes_its_address() {
        let name = "ns6.test";
        let mut msg = response_head(0x4243, name, 1, 0);
        push_answer(
            &mut msg,
            QTYPE_AAAA,
            &[0x20, 0x01, 0x0d, 0xb8, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1],
        );
        let answer = parse_response(0x4243, name, &msg).unwrap();
        assert_eq!(
            answer.ip_addrs(),
            vec!["2001:db8::1".parse::<std::net::IpAddr>().unwrap()]
        );
    }

    /// NS and address answers from one table; every other query is empty.
    struct TableLookup {
        ns: Vec<(&'static str, &'static str)>,
        addrs: Vec<(&'static str, std::net::IpAddr)>,
        ns_delay: Duration,
        aaaa_delay: Duration,
        in_flight: std::sync::atomic::AtomicUsize,
        peak: std::sync::atomic::AtomicUsize,
        addr_queries: std::sync::atomic::AtomicUsize,
    }

    impl TableLookup {
        fn new(
            ns: Vec<(&'static str, &'static str)>,
            addrs: Vec<(&'static str, std::net::IpAddr)>,
            ns_delay: Duration,
        ) -> Self {
            Self {
                ns,
                addrs,
                ns_delay,
                aaaa_delay: Duration::ZERO,
                in_flight: std::sync::atomic::AtomicUsize::new(0),
                peak: std::sync::atomic::AtomicUsize::new(0),
                addr_queries: std::sync::atomic::AtomicUsize::new(0),
            }
        }
    }

    impl DnsLookup for TableLookup {
        fn query<'a>(
            &'a self,
            _server: SocketAddr,
            name: &'a str,
            qtype: u16,
            _recursion_desired: bool,
        ) -> BoxFuture<'a, Result<DnsAnswer, String>> {
            use std::sync::atomic::Ordering::SeqCst;
            Box::pin(async move {
                let owner = normalize_name(name);
                let record = |rtype: u16, rdata: Rdata| ResourceRecord {
                    name: owner.clone(),
                    rtype,
                    rdata,
                };
                let records = match qtype {
                    QTYPE_NS => {
                        let now = self.in_flight.fetch_add(1, SeqCst) + 1;
                        self.peak.fetch_max(now, SeqCst);
                        tokio::time::sleep(self.ns_delay).await;
                        self.in_flight.fetch_sub(1, SeqCst);
                        self.ns
                            .iter()
                            .filter(|(zone, _)| *zone == owner)
                            .map(|(_, ns)| record(QTYPE_NS, Rdata::Name((*ns).to_owned())))
                            .collect()
                    }
                    QTYPE_A | QTYPE_AAAA => {
                        self.addr_queries.fetch_add(1, SeqCst);
                        if qtype == QTYPE_AAAA {
                            tokio::time::sleep(self.aaaa_delay).await;
                        }
                        self.addrs
                            .iter()
                            .filter(|(host, _)| *host == owner)
                            .filter_map(|(_, ip)| match (qtype, ip) {
                                (QTYPE_A, std::net::IpAddr::V4(v4)) => {
                                    Some(record(QTYPE_A, Rdata::A(*v4)))
                                }
                                (QTYPE_AAAA, std::net::IpAddr::V6(v6)) => {
                                    Some(record(QTYPE_AAAA, Rdata::Aaaa(*v6)))
                                }
                                _ => None,
                            })
                            .collect()
                    }
                    _ => Vec::new(),
                };
                Ok(DnsAnswer { rcode: 0, records })
            })
        }
    }

    /// #2478: a nameserver one resolver cannot resolve is still found
    /// through the next resolver.
    #[tokio::test]
    async fn a_nameserver_one_resolver_cannot_resolve_is_not_dropped() {
        struct SplitView;
        impl DnsLookup for SplitView {
            fn query<'a>(
                &'a self,
                server: SocketAddr,
                name: &'a str,
                qtype: u16,
                _recursion_desired: bool,
            ) -> BoxFuture<'a, Result<DnsAnswer, String>> {
                let owner = normalize_name(name);
                let record = |rtype, rdata| ResourceRecord {
                    name: owner.clone(),
                    rtype,
                    rdata,
                };
                let records = match (qtype, owner.as_str(), server.port()) {
                    (QTYPE_NS, "clientco.com", _) => vec![
                        record(QTYPE_NS, Rdata::Name("ns1.test".to_owned())),
                        record(QTYPE_NS, Rdata::Name("ns2.test".to_owned())),
                    ],
                    (QTYPE_A, "ns1.test", _) => {
                        vec![record(QTYPE_A, Rdata::A([192, 0, 2, 1].into()))]
                    }
                    // Only the second resolver can resolve `ns2.test`.
                    (QTYPE_A, "ns2.test", 2) => {
                        vec![record(QTYPE_A, Rdata::A([192, 0, 2, 2].into()))]
                    }
                    _ => Vec::new(),
                };
                Box::pin(async move { Ok(DnsAnswer { rcode: 0, records }) })
            }
        }

        let mut found = authoritative_resolvers(
            "_acme-challenge.clientco.com",
            &[resolver(1), resolver(2)],
            &SplitView,
        )
        .await;
        found.sort();
        assert_eq!(
            found,
            vec![
                SocketAddr::from(([192, 0, 2, 1], 53)),
                SocketAddr::from(([192, 0, 2, 2], 53)),
            ]
        );
    }

    // Codex review on #2936: a zone whose nameservers have only IPv6
    // addresses is still found.
    #[tokio::test]
    async fn discovery_finds_ipv6_only_nameservers() {
        let lookup = TableLookup::new(
            vec![("clientco.com", "ns6.test")],
            vec![("ns6.test", "2001:db8::1".parse().unwrap())],
            Duration::ZERO,
        );
        let found = authoritative_resolvers(
            "_autumn-challenge.app.clientco.com",
            &[resolver(53)],
            &lookup,
        )
        .await;
        assert_eq!(
            found,
            vec!["[2001:db8::1]:53".parse::<SocketAddr>().unwrap()]
        );
    }

    // Codex review on #2936: a hostname with many labels does not start an NS
    // lookup for every candidate zone at once.
    #[tokio::test(start_paused = true)]
    async fn discovery_caps_how_many_zones_it_asks_at_once() {
        let lookup = TableLookup::new(
            vec![("clientco.com", "ns1.test")],
            vec![("ns1.test", "192.0.2.1".parse().unwrap())],
            Duration::from_millis(100),
        );
        let host = format!("{}clientco.com", "a.".repeat(60));
        let found = authoritative_resolvers(&host, &[resolver(53)], &lookup).await;
        assert_eq!(found, vec![SocketAddr::from(([192, 0, 2, 1], 53))]);
        let peak = lookup.peak.load(std::sync::atomic::Ordering::SeqCst);
        assert!(peak <= MAX_ZONE_PROBES, "{peak} NS lookups at once");
    }

    // Codex review on #2936: a resolver that drops packets is not asked
    // again, so a hostname with many labels still reaches its zone's
    // nameserver after one query timeout.
    #[tokio::test(start_paused = true)]
    async fn a_dead_resolver_costs_one_timeout_for_a_long_hostname() {
        let started = tokio::time::Instant::now();
        let host = format!("{}_autumn-challenge.app.clientco.com", "a.".repeat(24));
        let found =
            authoritative_resolvers(&host, &[resolver(1), resolver(2)], &PartlyDeadLookup).await;
        assert_eq!(found, vec![SocketAddr::from(([192, 0, 2, 1], 53))]);
        assert!(
            started.elapsed() <= Duration::from_secs(3),
            "took {:?}",
            started.elapsed()
        );
    }

    // Codex review on #2936: a large NS set does not start an address lookup
    // for every name.
    #[tokio::test]
    async fn discovery_caps_how_many_nameservers_it_resolves() {
        const NAMES: [&str; 40] = [
            "ns00.test",
            "ns01.test",
            "ns02.test",
            "ns03.test",
            "ns04.test",
            "ns05.test",
            "ns06.test",
            "ns07.test",
            "ns08.test",
            "ns09.test",
            "ns10.test",
            "ns11.test",
            "ns12.test",
            "ns13.test",
            "ns14.test",
            "ns15.test",
            "ns16.test",
            "ns17.test",
            "ns18.test",
            "ns19.test",
            "ns20.test",
            "ns21.test",
            "ns22.test",
            "ns23.test",
            "ns24.test",
            "ns25.test",
            "ns26.test",
            "ns27.test",
            "ns28.test",
            "ns29.test",
            "ns30.test",
            "ns31.test",
            "ns32.test",
            "ns33.test",
            "ns34.test",
            "ns35.test",
            "ns36.test",
            "ns37.test",
            "ns38.test",
            "ns39.test",
        ];
        let lookup = TableLookup::new(
            NAMES.iter().map(|name| ("clientco.com", *name)).collect(),
            NAMES
                .iter()
                .zip(1_u8..)
                .map(|(name, n)| (*name, std::net::IpAddr::from([192, 0, 2, n])))
                .collect(),
            Duration::ZERO,
        );
        let found = authoritative_resolvers(
            "_autumn-challenge.app.clientco.com",
            &[resolver(53)],
            &lookup,
        )
        .await;
        assert_eq!(found.len(), MAX_NAMESERVERS);
        let asked = lookup
            .addr_queries
            .load(std::sync::atomic::Ordering::SeqCst);
        assert!(asked <= 2 * MAX_NAMESERVERS, "{asked} address lookups");
    }

    // Codex review on #2936: a dual-stack nameserver keeps both addresses,
    // even when the A answer comes first.
    #[tokio::test(start_paused = true)]
    async fn discovery_keeps_both_addresses_of_a_dual_stack_nameserver() {
        let mut lookup = TableLookup::new(
            vec![("clientco.com", "ns.test")],
            vec![
                ("ns.test", "192.0.2.1".parse().unwrap()),
                ("ns.test", "2001:db8::1".parse().unwrap()),
            ],
            Duration::ZERO,
        );
        lookup.aaaa_delay = Duration::from_millis(100);
        let found = authoritative_resolvers(
            "_autumn-challenge.app.clientco.com",
            &[resolver(53)],
            &lookup,
        )
        .await;
        assert_eq!(
            found,
            vec![
                SocketAddr::from(([192, 0, 2, 1], 53)),
                "[2001:db8::1]:53".parse::<SocketAddr>().unwrap(),
            ]
        );
    }

    /// Resolver 1 knows only the nameserver's A record and answers AAAA with
    /// nothing at once. Resolver 2 knows only its AAAA record and is slower.
    struct SplitStackLookup;

    impl DnsLookup for SplitStackLookup {
        fn query<'a>(
            &'a self,
            server: SocketAddr,
            name: &'a str,
            qtype: u16,
            _recursion_desired: bool,
        ) -> BoxFuture<'a, Result<DnsAnswer, String>> {
            Box::pin(async move {
                let owner = normalize_name(name);
                let record = |rtype: u16, rdata: Rdata| ResourceRecord {
                    name: owner.clone(),
                    rtype,
                    rdata,
                };
                if server == resolver(2) {
                    tokio::time::sleep(Duration::from_millis(100)).await;
                }
                let records = match (qtype, server == resolver(1), owner.as_str()) {
                    (QTYPE_NS, _, "clientco.com") => {
                        vec![record(QTYPE_NS, Rdata::Name("ns.test".to_owned()))]
                    }
                    (QTYPE_A, true, "ns.test") => {
                        vec![record(
                            QTYPE_A,
                            Rdata::A(std::net::Ipv4Addr::new(192, 0, 2, 1)),
                        )]
                    }
                    (QTYPE_AAAA, false, "ns.test") => vec![record(
                        QTYPE_AAAA,
                        Rdata::Aaaa("2001:db8::1".parse().unwrap()),
                    )],
                    _ => Vec::new(),
                };
                Ok(DnsAnswer { rcode: 0, records })
            })
        }
    }

    // Codex review on #2936: an empty AAAA answer from one resolver does not
    // stop the wait for another resolver's AAAA records.
    #[tokio::test(start_paused = true)]
    async fn an_empty_answer_does_not_cut_off_the_other_address_family() {
        let found = authoritative_resolvers(
            "_autumn-challenge.app.clientco.com",
            &[resolver(1), resolver(2)],
            &SplitStackLookup,
        )
        .await;
        assert_eq!(
            found,
            vec![
                SocketAddr::from(([192, 0, 2, 1], 53)),
                "[2001:db8::1]:53".parse::<SocketAddr>().unwrap(),
            ]
        );
    }

    #[tokio::test]
    async fn propagation_succeeds_once_every_resolver_sees_every_value() {
        let visible = || {
            Ok(TxtAnswer {
                values: vec!["value-apex".to_owned(), "value-wildcard".to_owned()],
                rcode: 0,
            })
        };
        let lookup =
            ScriptedLookup::new(vec![(resolver(53), visible()), (resolver(5353), visible())]);
        let records = vec![
            TxtRecord::new("myapp.com", "value-apex"),
            TxtRecord::new("myapp.com", "value-wildcard"),
        ];
        wait_for_propagation(
            &records,
            &ProbeTargets::flat(&[resolver(53), resolver(5353)]),
            Duration::from_secs(5),
            Duration::from_millis(10),
            lookup.as_ref(),
        )
        .await
        .expect("both resolvers see both values");
        // One round, one query per resolver: the two values share a record name,
        // so they are checked as a set rather than queried separately.
        assert_eq!(lookup.call_count(), 2);
    }

    // AC5: "a bounded, documented wait for TXT record propagation whose timeout
    // error names the exact record that failed to propagate."
    #[tokio::test]
    async fn a_timeout_names_the_record_the_value_and_the_resolver() {
        let lookup = ScriptedLookup::new(Vec::new()); // every resolver: NXDOMAIN
        let records = vec![TxtRecord::new("myapp.com", "value-apex")];
        let err = wait_for_propagation(
            &records,
            &ProbeTargets::flat(&[resolver(5353)]),
            Duration::from_millis(60),
            Duration::from_millis(10),
            lookup.as_ref(),
        )
        .await
        .expect_err("an unpublished record must time out");
        assert!(err.contains("_acme-challenge.myapp.com"), "got: {err}");
        assert!(err.contains("value-apex"), "got: {err}");
        assert!(err.contains("127.0.0.1:5353"), "got: {err}");
        assert!(err.contains("does not exist there yet"), "got: {err}");
        assert!(
            err.contains("propagation_timeout_secs"),
            "the message must say which knob to turn: {err}"
        );
    }

    // One resolver lagging is enough to keep waiting: telling the CA to validate
    // while a resolver it might pick still 404s is how an authorization is burnt.
    #[tokio::test]
    async fn one_lagging_resolver_blocks_the_wait() {
        let lookup = ScriptedLookup::new(vec![
            (
                resolver(53),
                Ok(TxtAnswer {
                    values: vec!["v".to_owned()],
                    rcode: 0,
                }),
            ),
            (
                resolver(5353),
                Ok(TxtAnswer {
                    values: Vec::new(),
                    rcode: 0,
                }),
            ),
        ]);
        let err = wait_for_propagation(
            &[TxtRecord::new("myapp.com", "v")],
            &ProbeTargets::flat(&[resolver(53), resolver(5353)]),
            Duration::from_millis(40),
            Duration::from_millis(10),
            lookup.as_ref(),
        )
        .await
        .expect_err("the lagging resolver must hold the wait");
        assert!(err.contains("127.0.0.1:5353"), "got: {err}");
    }

    // A resolver error is a not-yet rather than a hard failure — but it is
    // reported verbatim if the budget then runs out, because SERVFAIL on
    // `_acme-challenge` is usually a broken delegation.
    #[tokio::test]
    async fn a_resolver_error_is_carried_into_the_timeout_message() {
        let lookup = ScriptedLookup::new(vec![(
            resolver(53),
            Err("SERVFAIL — the zone's nameservers did not answer".to_owned()),
        )]);
        let err = wait_for_propagation(
            &[TxtRecord::new("myapp.com", "v")],
            &ProbeTargets::flat(&[resolver(53)]),
            Duration::from_millis(20),
            Duration::from_millis(10),
            lookup.as_ref(),
        )
        .await
        .expect_err("times out");
        assert!(err.contains("SERVFAIL"), "got: {err}");
    }

    /// Regression (#1620): the recursion-desired bit must follow whether the
    /// probed server is authoritative for the name.
    ///
    /// An authoritative server is asked with RD=0 — it reads its own zone, and a
    /// non-recursive query cannot plant a negative cache entry. The *fallback*
    /// path probes public recursive resolvers, and RD=0 there means "answer from
    /// cache only": a `_acme-challenge` name nothing has ever looked up is not
    /// in any cache, so the resolver returns an empty answer or REFUSED however
    /// long the wait runs. That is the whole fallback path failing closed on a
    /// correctly published record — including the IPv6-only authoritative set,
    /// where discovery finds no A records and the fallback is all there is.
    #[tokio::test]
    async fn recursion_is_desired_only_when_probing_the_fallback_resolvers() {
        struct RecordingLookup {
            asked: Mutex<Vec<(SocketAddr, bool)>>,
        }

        impl DnsLookup for RecordingLookup {
            fn query<'a>(
                &'a self,
                server: SocketAddr,
                name: &'a str,
                _qtype: u16,
                recursion_desired: bool,
            ) -> BoxFuture<'a, Result<DnsAnswer, String>> {
                self.asked.lock().unwrap().push((server, recursion_desired));
                let name = normalize_name(name);
                Box::pin(async move {
                    Ok(DnsAnswer {
                        rcode: 0,
                        records: vec![ResourceRecord {
                            name,
                            rtype: QTYPE_TXT,
                            rdata: Rdata::Txt("v".to_owned()),
                        }],
                    })
                })
            }
        }

        let authoritative = resolver(53);
        let recursive = resolver(5353);
        let lookup = Arc::new(RecordingLookup {
            asked: Mutex::new(Vec::new()),
        });

        // `myapp.com` has discovered authoritative servers; `other.com` did not,
        // so it falls back to the configured recursive resolver.
        let mut targets = ProbeTargets::flat(&[recursive]);
        targets.set_authoritative("_acme-challenge.myapp.com", vec![authoritative]);

        wait_for_propagation(
            &[
                TxtRecord::new("myapp.com", "v"),
                TxtRecord::new("other.com", "v"),
            ],
            &targets,
            Duration::from_secs(1),
            Duration::from_millis(10),
            lookup.as_ref(),
        )
        .await
        .expect("both names are visible");

        let asked = lookup.asked.lock().unwrap().clone();
        assert!(
            asked.contains(&(authoritative, false)),
            "an authoritative server must be asked with RD=0: {asked:?}"
        );
        assert!(
            asked.contains(&(recursive, true)),
            "a fallback recursive resolver must be asked with RD=1, or it answers only from a \
             cache that cannot hold a name nothing has looked up: {asked:?}"
        );
    }

    #[tokio::test]
    async fn no_records_is_an_immediate_pass_and_no_resolvers_is_an_error() {
        let lookup = ScriptedLookup::new(Vec::new());
        assert!(
            wait_for_propagation(
                &[],
                &ProbeTargets::default(),
                Duration::from_secs(1),
                Duration::from_millis(1),
                lookup.as_ref()
            )
            .await
            .is_ok()
        );
        let err = wait_for_propagation(
            &[TxtRecord::new("myapp.com", "v")],
            &ProbeTargets::default(),
            Duration::from_secs(1),
            Duration::from_millis(1),
            lookup.as_ref(),
        )
        .await
        .expect_err("no resolvers cannot confirm anything");
        assert!(err.contains("resolvers"), "got: {err}");
    }

    // The real UDP path, against an in-process resolver: proves the wire format
    // autumn sends is one a server can answer, and that the answer round-trips.
    #[tokio::test]
    async fn udp_lookup_round_trips_against_a_real_socket() {
        let server = tokio::net::UdpSocket::bind("127.0.0.1:0")
            .await
            .expect("bind fake resolver");
        let addr = server.local_addr().expect("local addr");
        tokio::spawn(async move {
            let mut buf = vec![0_u8; 512];
            let Ok((read, peer)) = server.recv_from(&mut buf).await else {
                return;
            };
            let id = u16::from_be_bytes([buf[0], buf[1]]);
            let name = query_name(&buf[..read]);
            let response = txt_response(id, &name, &["propagated-value"], 0);
            let _ = server.send_to(&response, peer).await;
        });

        let lookup = UdpDnsLookup::new(Duration::from_secs(5));
        let answer = lookup
            .query(addr, "_acme-challenge.myapp.com", QTYPE_TXT, false)
            .await
            .expect("the fake resolver answers");
        assert_eq!(
            answer.txt_values("_acme-challenge.myapp.com"),
            vec!["propagated-value"]
        );
    }

    #[tokio::test]
    async fn udp_lookup_times_out_against_a_silent_resolver() {
        let server = tokio::net::UdpSocket::bind("127.0.0.1:0")
            .await
            .expect("bind silent resolver");
        let addr = server.local_addr().expect("local addr");
        let lookup = UdpDnsLookup::new(Duration::from_millis(50));
        let err = lookup
            .query(addr, "_acme-challenge.myapp.com", QTYPE_TXT, false)
            .await
            .expect_err("a silent resolver must time out, not hang");
        assert!(err.contains("did not answer"), "got: {err}");
    }
}