ruststream 0.7.5

//! What a message carries through the broker's own publisher: its place in the order, its
//! headers, its key and its per-message settings.
//!
//! [`harness::lifecycle`](super::harness::lifecycle) runs the part every broker owes with no input
//! of its own: every message published arrives on one subscription, in whatever order the
//! transport delivers unkeyed messages, and headers come back byte for byte, or the publish that
//! carries them is refused and the message never arrives. The rest needs something only the
//! broker can supply, so each part is a function the broker crate calls:
//!
//! * [`keyed_order`]: a keyed message reports its key on delivery, and one key keeps its order;
//!   the broker says where a key goes. It is the order check: the order between unkeyed messages
//!   is free.
//! * [`publish_options`]: [`Publisher::Options`] resolve over the policy's defaults, and a value
//!   the transport cannot honour is a publish error; the broker supplies the values and a way to
//!   read their effect off a delivery.
//! * [`publishes_without_credentials`] and [`describes_addresses_without_credentials`]: the
//!   credential scan of [`harness::describes_without_credentials`] applied to a publish policy's
//!   bindings and to a server configured with several addresses.
//!
//! Each runs live and, through [`InProcessBroker`](super::harness::InProcessBroker), in process.
//!
//! [`harness::describes_without_credentials`]: super::harness::describes_without_credentials

use std::{fmt, future::Future, time::Duration};

use bytes::Bytes;
use futures::{Stream, StreamExt};
#[cfg(feature = "asyncapi")]
use serde_json::Value;
use tokio::time::{Instant, timeout, timeout_at};

use super::helpers::unique_subject;
#[cfg(feature = "asyncapi")]
use crate::ConnectedBroker;
#[cfg(feature = "asyncapi")]
use crate::DescribeServer;
#[cfg(feature = "asyncapi")]
use crate::asyncapi::build_spec;
#[cfg(feature = "asyncapi")]
use crate::runtime::{AppInfo, RustStream};
use crate::{
    AckError, Broker, Connected, HeaderMap, IncomingMessage, OutgoingFor, OutgoingMessage,
    PublishPolicy, Publisher, Subscriber, SubscriptionSource, runtime::RETRY_COUNT_HEADER,
};

/// How long a check waits for a delivery it expects. Long enough for a live consumer that joins a
/// group on its first poll; a correct broker never comes near it, it only bounds a failing run.
const DELIVERY_TIMEOUT: Duration = Duration::from_secs(10);
/// How long a connect, a subscription, a pairing or a single publish may take before the check
/// fails it. As long as a delivery may take, for the same live server.
const STEP_TIMEOUT: Duration = DELIVERY_TIMEOUT;
/// How long a check watches, after the last delivery it expects, for a refused message arriving
/// late.
const QUIET_PERIOD: Duration = Duration::from_millis(500);
/// How many unkeyed messages [`publisher_carries`] publishes one after another.
const ORDERED: usize = 32;
/// How many entries the header check puts on one message.
const MANY_HEADERS: usize = 24;
/// How many messages [`keyed_order`] publishes under each key.
const PER_KEY: usize = 8;
/// The keys [`keyed_order`] interleaves.
const KEYS: [&str; 3] = [
    "conformance-key-a",
    "conformance-key-b",
    "conformance-key-c",
];
/// A W3C trace context as the tracing layer writes it.
const TRACEPARENT: (&str, &str) = (
    "traceparent",
    "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01",
);
/// Bytes that are not UTF-8 (a lone continuation byte, a truncated sequence, `0xff`), with a NUL
/// and line breaks a text encoding would mangle.
const BINARY_VALUE: &[u8] = &[0x00, 0x80, b'\r', b'\n', 0xfe, 0xff, 0xc3];

/// The message type a subscriber yields.
type SubscriberMessage<S> = <S as Subscriber>::Message;

/// Every message published through the broker's own publisher arrives, and headers survive the
/// transport byte for byte, or the publish carrying them is refused and the message never arrives.
///
/// Run first by [`harness::lifecycle`](super::harness::lifecycle), on a connection of its own
/// that it closes before the ladder opens its own. The factories come back so the ladder goes on
/// with them: holding them by reference across the awaits would make the ladder's future `Send`
/// only for factories that are also `Sync`.
pub(crate) async fn publisher_carries<B, MkBroker, Src, MkSrc, Pub, MkPub>(
    make_broker: MkBroker,
    make_source: MkSrc,
    make_publisher: MkPub,
) -> (MkBroker, MkSrc, MkPub)
where
    B: Broker,
    MkBroker: Fn() -> B,
    Src: SubscriptionSource<Connected<B>> + Send,
    Src::Subscriber: Send,
    MkSrc: Fn(&str) -> Src,
    Pub: Publisher,
    MkPub: Fn(&Connected<B>) -> Pub,
{
    // Under the lifecycle prefix, because this runs inside `lifecycle` and a broker whose
    // subscriptions need a resource of their own (a stream, a queue) declares it for that prefix.
    let subject = unique_subject("conformance.lifecycle");
    let connected = within(make_broker().connect(), "message shape: connect")
        .await
        .expect("message shape: broker must connect after synchronous construction");
    let mut subscriber = within(
        make_source(&subject).subscribe(&connected),
        "message shape: subscribe",
    )
    .await
    .expect("message shape: subscription source must open against the connected form");
    let publisher = make_publisher(&connected);
    let sent = publish_every_shape(&publisher, &subject).await;
    {
        let mut stream = std::pin::pin!(subscriber.stream());
        receive_every_shape(&mut stream, &sent).await;
    }

    // Closed before the ladder subscribes: a consumer group that still counts this subscription
    // as a member would hold the ladder's subscription back until the member timed out.
    drop(subscriber);
    super::lifecycle::shutdown_within(connected, "lifecycle").await;
    (make_broker, make_source, make_publisher)
}

/// One message's worth of headers, and what the check calls it in a failure.
struct HeaderCase {
    label: &'static str,
    headers: HeaderMap,
}

fn header_cases() -> [HeaderCase; 4] {
    let mut many = HeaderMap::new();
    for index in 0..MANY_HEADERS {
        many.insert(
            format!("x-conformance-entry-{index:02}"),
            format!("value-{index}"),
        );
    }
    let mut empty = HeaderMap::new();
    empty.insert("x-conformance-empty", Bytes::new());
    empty.insert("x-conformance-present", "yes");
    let mut binary = HeaderMap::new();
    binary.insert("x-conformance-binary", Bytes::from_static(BINARY_VALUE));
    let mut framework = HeaderMap::new();
    framework.insert(RETRY_COUNT_HEADER, "2");
    framework.insert(TRACEPARENT.0, TRACEPARENT.1);
    [
        HeaderCase {
            label: "many entries",
            headers: many,
        },
        HeaderCase {
            label: "an empty value",
            headers: empty,
        },
        HeaderCase {
            label: "a non-UTF-8 value",
            headers: binary,
        },
        HeaderCase {
            label: "the framework's retry count and trace context",
            headers: framework,
        },
    ]
}

/// What the delivery of one published message must show.
enum Awaited {
    /// A message of the unkeyed run: it arrives.
    Plain,
    /// An accepted publish with headers: it arrives carrying them byte for byte.
    Carrying(HeaderCase),
    /// A refused publish: it never arrives.
    Refused(&'static str),
}

/// A message [`publisher_carries`] published, by payload.
struct Sent {
    payload: String,
    awaited: Awaited,
}

/// Publishes the header cases, then the unkeyed run, and returns what each delivery must show.
async fn publish_every_shape<Pub: Publisher>(publisher: &Pub, subject: &str) -> Vec<Sent> {
    let cases = header_cases();
    let mut sent = Vec::with_capacity(cases.len() + ORDERED);
    for (index, case) in cases.into_iter().enumerate() {
        let payload = format!("headers:{index}");
        let what = format!("header round trip: the publish carrying {}", case.label);
        let answer = publish_within(
            publisher,
            OutgoingMessage::new(subject, payload.as_bytes()).with_headers(case.headers.clone()),
            None,
            &what,
        )
        .await;
        // Refusing is the honest answer of a transport that cannot carry these headers; what the
        // check holds it to is that a refused message never arrives.
        let awaited = match answer {
            Ok(()) => Awaited::Carrying(case),
            Err(_) => Awaited::Refused(case.label),
        };
        sent.push(Sent { payload, awaited });
    }
    for index in 0..ORDERED {
        let payload = format!("unkeyed:{index:04}");
        let what = format!("unkeyed run: publish {index} through the broker's publisher");
        publish_within(
            publisher,
            OutgoingMessage::new(subject, payload.as_bytes()),
            None,
            &what,
        )
        .await
        .unwrap_or_else(|err| panic!("{what} failed: {err}"));
        sent.push(Sent {
            payload,
            awaited: Awaited::Plain,
        });
    }
    sent
}

/// Reads until every message that must arrive did, in whatever order the transport delivers them
/// (a partitioned one spreads unkeyed records over partitions), then watches for a refused one
/// arriving late. A redelivery of an accepted message is the transport's right and is settled.
async fn receive_every_shape<S, M, E>(stream: &mut S, sent: &[Sent])
where
    S: Stream<Item = Result<M, E>> + Unpin,
    M: IncomingMessage,
    E: fmt::Debug,
{
    let mut arrived = vec![false; sent.len()];
    // The wait restarts only when a message arrives for the first time: a transport that delivers
    // gradually keeps making progress, while a redelivery the check settles and skips buys the
    // missing message no fresh wait, or a transport that repeats one message never fails it.
    let mut deadline = Instant::now() + DELIVERY_TIMEOUT;
    while let Some(missing) = sent
        .iter()
        .zip(&arrived)
        .position(|(sent, arrived)| !arrived && !matches!(sent.awaited, Awaited::Refused(_)))
    {
        let label = match &sent[missing].awaited {
            Awaited::Carrying(case) => format!(
                "header round trip: waiting for the accepted publish carrying {}",
                case.label
            ),
            _ => format!("unkeyed run: waiting for {}", sent[missing].payload),
        };
        let msg = next_delivery(stream, deadline, &label)
            .await
            .unwrap_or_else(|| panic!("{label}: no delivery within {DELIVERY_TIMEOUT:?}"));
        let index = sent
            .iter()
            .position(|sent| sent.payload.as_bytes() == msg.payload())
            .unwrap_or_else(|| {
                panic!(
                    "message shape: a delivery nobody published arrived: {:?}; the payload must \
                     arrive as it was published",
                    String::from_utf8_lossy(msg.payload()),
                )
            });
        match &sent[index].awaited {
            Awaited::Refused(label) => {
                refused_arrived(
                    "header round trip",
                    &format!("the publish carrying {label}"),
                );
            }
            Awaited::Carrying(case) => {
                for (name, value) in case.headers.iter() {
                    assert_eq!(
                        msg.headers().get(name),
                        Some(value),
                        "header round trip ({}): header {name:?} must arrive byte for byte; a \
                         transport that cannot carry it must refuse the publish instead of \
                         dropping or rewriting it",
                        case.label,
                    );
                }
            }
            Awaited::Plain => {}
        }
        if !arrived[index] {
            arrived[index] = true;
            deadline = Instant::now() + DELIVERY_TIMEOUT;
        }
        settle(msg, "message shape").await;
    }
    nothing_refused_arrives(stream, "header round trip", |payload| {
        sent.iter()
            .find(|sent| sent.payload.as_bytes() == payload)
            .and_then(|sent| match sent.awaited {
                Awaited::Refused(label) => Some(format!("the publish carrying {label}")),
                _ => None,
            })
    })
    .await;
}

/// Fails the check for a message whose publish was refused and which arrived all the same.
fn refused_arrived(check: &str, what: &str) -> ! {
    panic!("{check}: {what} was refused, and the message arrived anyway")
}

/// Verifies that a keyed message reports its key on delivery and that one key keeps its order.
///
/// Three keys are interleaved, eight messages each, through the broker's own publisher. Every
/// delivery on the one subscription must report the key it was published under through
/// [`IncomingMessage::partition_key`], and the messages of one key must arrive in the order they
/// were published. The order between different keys is free: a partitioned transport spreads
/// keys over partitions and reads them back in any interleaving.
///
/// `subject` is the name the check subscribes and publishes under. Build it with
/// [`unique_subject`], so a rerun against one server reads only its own messages, in whatever form
/// the transport needs for its keys to exist (an SQS FIFO queue is a name ending in `.fifo`).
///
/// `key_with` is where the broker says how a publish carries a key: it writes the key into the
/// message's headers, or returns the [`Publisher::Options`] that carry it, or both. A transport
/// with no key has nothing to report here and does not run this check; one that has keys and
/// reports `None` fails it, because the runtime's keyed worker lanes (`workers(n, by_key)`) read
/// the key from there.
///
/// # Examples
///
/// ```no_run
/// # #[cfg(feature = "memory")]
/// # async fn run() {
/// use ruststream::Bytes;
/// use ruststream::conformance::{helpers::unique_subject, message_shape};
/// use ruststream::memory::{MemoryBroker, MemorySource, PARTITION_KEY_HEADER};
///
/// message_shape::keyed_order(
///     MemoryBroker::new,
///     &unique_subject("conformance.keyed"),
///     |name| MemorySource::new(name),
///     |connected| connected.publisher(),
///     // The in-memory broker carries a key in a header, and has no per-message options.
///     |key, headers| {
///         headers.insert(PARTITION_KEY_HEADER, Bytes::copy_from_slice(key));
///         None
///     },
/// )
/// .await;
/// # }
/// ```
///
/// # Panics
///
/// Panics with a descriptive message when a keyed publish fails, a delivery reports another key
/// or none, or a key's messages arrive out of order.
pub async fn keyed_order<B, MkBroker, Src, MkSrc, Pub, MkPub, KeyWith>(
    make_broker: MkBroker,
    subject: &str,
    make_source: MkSrc,
    make_publisher: MkPub,
    key_with: KeyWith,
) where
    B: Broker,
    MkBroker: Fn() -> B,
    Src: SubscriptionSource<Connected<B>> + Send,
    Src::Subscriber: Send,
    MkSrc: Fn(&str) -> Src,
    Pub: Publisher,
    MkPub: Fn(&Connected<B>) -> Pub,
    KeyWith: Fn(&[u8], &mut HeaderMap) -> Option<Pub::Options>,
{
    let connected = within(make_broker().connect(), "keyed order: connect")
        .await
        .expect("keyed order: broker must connect after synchronous construction");
    let mut subscriber = within(
        make_source(subject).subscribe(&connected),
        "keyed order: subscribe",
    )
    .await
    .expect("keyed order: subscription source must open against the connected form");
    let publisher = make_publisher(&connected);

    for seq in 0..PER_KEY {
        for key in KEYS {
            let payload = format!("{key}:{seq:02}");
            let mut headers = HeaderMap::new();
            let options = key_with(key.as_bytes(), &mut headers);
            let msg = OutgoingMessage::new(subject, payload.as_bytes()).with_headers(headers);
            let what = format!("keyed order: publishing {payload}");
            publish_within(&publisher, msg, options.as_ref(), &what)
                .await
                .unwrap_or_else(|err| panic!("{what} failed: {err}"));
        }
    }

    {
        let mut next = [0usize; KEYS.len()];
        let mut stream = std::pin::pin!(subscriber.stream());
        for _ in 0..PER_KEY * KEYS.len() {
            let msg = expect_delivery(&mut stream, "keyed order").await;
            let payload = String::from_utf8_lossy(msg.payload()).into_owned();
            let (slot, seq) = payload
                .rsplit_once(':')
                .and_then(|(key, seq)| {
                    Some((
                        KEYS.iter().position(|known| *known == key)?,
                        seq.parse::<usize>().ok()?,
                    ))
                })
                .unwrap_or_else(|| panic!("keyed order: a delivery nobody published: {payload:?}"));
            let key = KEYS[slot];
            assert_eq!(
                msg.partition_key(),
                Some(key.as_bytes()),
                "keyed order: {payload} was published under key {key:?}; the delivery must report \
                 that key from partition_key()",
            );
            assert_eq!(
                seq, next[slot],
                "keyed order: the messages of key {key:?} must arrive in publish order",
            );
            next[slot] += 1;
            settle(msg, "keyed order").await;
        }
    }
    drop(subscriber);
    super::lifecycle::shutdown_within(connected, "keyed order").await;
}

/// The per-message settings [`publish_options`] publishes with, and what each must look like on
/// the delivery.
///
/// `Options` is the broker's [`Publisher::Options`]; `Observed` is whatever the broker reads off
/// a delivery to see a setting's effect (a priority, a `QoS`, an ordering key).
///
/// # Examples
///
/// ```
/// use ruststream::conformance::message_shape::OptionCases;
///
/// /// A broker's options: every field optional, as the contract asks.
/// #[derive(Clone)]
/// struct Priority {
///     priority: Option<u8>,
/// }
///
/// /// What a broker crate's conformance test hands `publish_options`: the policy under test is
/// /// configured with priority 5, a call may lower it to 1, and 42 is past the transport's range.
/// fn priority_cases() -> OptionCases<Priority, Option<u8>> {
///     OptionCases::new(Some(5))
///         .overrides(Priority { priority: Some(1) }, Some(1))
///         .refuses(Priority { priority: Some(42) })
/// }
/// ```
#[derive(Debug, Clone)]
pub struct OptionCases<Options, Observed> {
    policy_default: Observed,
    overrides: Vec<(Options, Observed)>,
    refused: Vec<Options>,
}

impl<Options, Observed> OptionCases<Options, Observed> {
    /// What a publish with no options (`None`) must show: the setting the policy fixed.
    ///
    /// Configure the policy with a value the transport would not pick by itself, so a publisher
    /// that forgets the policy cannot pass by landing on the transport's own default.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # #[cfg(feature = "memory")]
    /// # async fn run() {
    /// use ruststream::conformance::helpers::unique_subject;
    /// use ruststream::conformance::message_shape::{self, OptionCases};
    /// use ruststream::memory::{MemoryBroker, MemoryPublish, MemorySource};
    ///
    /// // The in-memory broker has no per-message settings: the policy default and the one override
    /// // are both the unit value.
    /// message_shape::publish_options(
    ///     MemoryBroker::new,
    ///     &unique_subject("conformance.options"),
    ///     |name| MemorySource::new(name),
    ///     MemoryPublish,
    ///     OptionCases::new(()).overrides((), ()),
    ///     |_delivery| (),
    /// )
    /// .await;
    /// # }
    /// ```
    #[must_use]
    pub const fn new(policy_default: Observed) -> Self {
        Self {
            policy_default,
            overrides: Vec::new(),
            refused: Vec::new(),
        }
    }

    /// Adds a call whose `options` must win over the policy, and what the delivery then shows.
    ///
    /// Set one field and leave the rest unset in some of them: a field the call leaves unset
    /// keeps the policy's value, and `observed` says so.
    ///
    /// # Examples
    ///
    /// ```
    /// use ruststream::conformance::message_shape::OptionCases;
    ///
    /// /// An MQTT publisher's options, and what a delivery shows of them.
    /// #[derive(Clone, Default)]
    /// struct MqttOptions {
    ///     qos: Option<u8>,
    ///     retain: Option<bool>,
    /// }
    ///
    /// /// The policy publishes at QoS 1 without retain. A call that names only the QoS keeps the
    /// /// policy's retain flag, and one that names only retain keeps its QoS.
    /// fn mqtt_cases() -> OptionCases<MqttOptions, (u8, bool)> {
    ///     OptionCases::new((1, false))
    ///         .overrides(
    ///             MqttOptions {
    ///                 qos: Some(2),
    ///                 ..MqttOptions::default()
    ///             },
    ///             (2, false),
    ///         )
    ///         .overrides(
    ///             MqttOptions {
    ///                 retain: Some(true),
    ///                 ..MqttOptions::default()
    ///             },
    ///             (1, true),
    ///         )
    /// }
    /// ```
    #[must_use]
    pub fn overrides(mut self, options: Options, observed: Observed) -> Self {
        self.overrides.push((options, observed));
        self
    }

    /// Adds a value the transport cannot honour: the publish carrying it must fail.
    ///
    /// A broker whose every value is honourable adds none.
    ///
    /// # Examples
    ///
    /// ```
    /// use ruststream::conformance::message_shape::OptionCases;
    ///
    /// /// An MQTT publisher's options, and the QoS a delivery shows.
    /// #[derive(Clone, Default)]
    /// struct MqttOptions {
    ///     qos: Option<u8>,
    /// }
    ///
    /// /// MQTT knows QoS 0 to 2, so a publish asking for 3 must fail rather than be clamped.
    /// fn qos_cases() -> OptionCases<MqttOptions, u8> {
    ///     OptionCases::new(1)
    ///         .overrides(MqttOptions { qos: Some(0) }, 0)
    ///         .refuses(MqttOptions { qos: Some(3) })
    /// }
    /// ```
    #[must_use]
    pub fn refuses(mut self, options: Options) -> Self {
        self.refused.push(options);
        self
    }
}

/// Verifies that a publisher's per-message settings resolve over its policy, and that a value
/// the transport cannot honour fails the publish.
///
/// The live publisher comes from pairing `policy` with the connected broker, the way the runtime
/// builds it. Then, through that publisher:
///
/// * a publish with `None` shows the policy's setting ([`OptionCases::new`]);
/// * each override shows its own setting, and the next publish with `None` shows the policy's
///   setting again, so a call adjusts only itself ([`OptionCases::overrides`]);
/// * each refused value fails the publish, and the message never arrives, neither before the
///   publish that follows nor after it ([`OptionCases::refuses`]); substituting a default for it
///   would be a silent fallback.
///
/// `subject` is the name the check subscribes and publishes under. Build it with
/// [`unique_subject`], so a rerun against one server reads only its own messages, in whatever form
/// the transport needs for the settings to apply (an SQS FIFO queue is a name ending in `.fifo`).
///
/// `observe` reads the setting's effect off a delivery, in whatever form the broker exposes it.
///
/// # Examples
///
/// The in-memory broker has no per-message settings, so its run is a single unit case. A broker
/// with settings passes its own policy, configured away from the transport default, and cases
/// that exercise every field.
///
/// ```no_run
/// # #[cfg(feature = "memory")]
/// # async fn run() {
/// use ruststream::conformance::helpers::unique_subject;
/// use ruststream::conformance::message_shape::{self, OptionCases};
/// use ruststream::memory::{MemoryBroker, MemoryPublish, MemorySource};
///
/// message_shape::publish_options(
///     MemoryBroker::new,
///     &unique_subject("conformance.options"),
///     |name| MemorySource::new(name),
///     MemoryPublish,
///     OptionCases::new(()).overrides((), ()),
///     |_delivery| (),
/// )
/// .await;
/// # }
/// ```
///
/// # Panics
///
/// Panics with a descriptive message when `cases` has no override (a check with nothing to
/// override proves nothing), when pairing or a publish that must succeed fails, when a delivery
/// shows another setting than expected, or when a refused value is accepted or arrives.
pub async fn publish_options<B, MkBroker, Src, MkSrc, Policy, Observed, Observe>(
    make_broker: MkBroker,
    subject: &str,
    make_source: MkSrc,
    policy: Policy,
    cases: OptionCases<<Policy::Live as Publisher>::Options, Observed>,
    observe: Observe,
) where
    B: Broker,
    MkBroker: Fn() -> B,
    Src: SubscriptionSource<Connected<B>> + Send,
    Src::Subscriber: Send,
    MkSrc: Fn(&str) -> Src,
    Policy: PublishPolicy<Connected<B>>,
    Policy::Live: Publisher,
    Observed: PartialEq + fmt::Debug + Sync,
    Observe: Fn(&SubscriberMessage<Src::Subscriber>) -> Observed + Sync,
{
    assert!(
        !cases.overrides.is_empty(),
        "publish options: pass at least one override; with none the check cannot tell a \
         publisher that ignores the call's options from one that honours them",
    );
    let connected = within(make_broker().connect(), "publish options: connect")
        .await
        .expect("publish options: broker must connect after synchronous construction");
    let mut subscriber = within(
        make_source(subject).subscribe(&connected),
        "publish options: subscribe",
    )
    .await
    .expect("publish options: subscription source must open against the connected form");
    let publisher = within(
        policy.pair(&connected),
        "publish options: pairing the policy",
    )
    .await
    .expect("publish options: the policy must pair with the connected broker");
    {
        let mut stream = std::pin::pin!(subscriber.stream());

        let (publisher, observe, delivered) = (&publisher, &observe, &mut stream);
        let mut round = 0usize;
        let mut expect = async move |options: Option<&<Policy::Live as Publisher>::Options>,
                                     expected: &Observed,
                                     what: &str| {
            let payload = format!("options:{round}");
            round += 1;
            let what = format!("publish options: {what}");
            publish_within(
                publisher,
                OutgoingMessage::new(subject, payload.as_bytes()),
                options,
                &what,
            )
            .await
            .unwrap_or_else(|err| panic!("{what}: the publish failed: {err}"));
            let msg = expect_delivery(&mut *delivered, &what).await;
            assert_eq!(
                String::from_utf8_lossy(msg.payload()),
                payload,
                "{what}: a message arrived out of order, or one that was refused",
            );
            let observed = observe(&msg);
            assert_eq!(
                &observed, expected,
                "{what}: the delivery shows another setting",
            );
            settle(msg, "publish options").await;
        };

        expect(
            None,
            &cases.policy_default,
            "a publish with no options takes the policy's",
        )
        .await;
        for (index, (options, observed)) in cases.overrides.iter().enumerate() {
            let what = format!("override {index} must win over the policy");
            expect(Some(options), observed, &what).await;
            let what =
                format!("the publish after override {index} takes the policy's setting again");
            expect(None, &cases.policy_default, &what).await;
        }
        for (index, options) in cases.refused.iter().enumerate() {
            let payload = format!("options:refused:{index}");
            let refused = publish_within(
                publisher,
                OutgoingMessage::new(subject, payload.as_bytes()),
                Some(options),
                &format!("publish options: refused value {index}"),
            )
            .await;
            assert!(
                refused.is_err(),
                "publish options: refused value {index} was accepted; a value the transport cannot \
                 honour must fail the publish, never fall back to another setting",
            );
        }
        // The refused publishes sit before this one, so a refused message that arrived anyway is
        // the next delivery and fails the payload assertion; one that arrives later still is
        // caught by the quiet period after it.
        expect(
            None,
            &cases.policy_default,
            "the publish after the refused values",
        )
        .await;
        nothing_refused_arrives(&mut stream, "publish options", |payload| {
            let index = payload.strip_prefix(b"options:refused:")?;
            Some(format!("refused value {}", String::from_utf8_lossy(index)))
        })
        .await;
    }
    drop(subscriber);
    super::lifecycle::shutdown_within(connected, "publish options").await;
}

/// Fails when a publish policy's bindings carry `secret`.
///
/// A policy describes the positions it is bound on (a reply, an [`Out`](crate::runtime::Out)
/// slot, a dead-letter destination) in the generated document, which is published and shared.
/// Configure `policy` the way a deployment would, with a password you pass as `secret`, and this
/// scans its channel, operation and message bindings and its reply address location.
///
/// # Examples
///
/// ```
/// # #[cfg(all(feature = "asyncapi", feature = "memory"))]
/// # fn demo() {
/// use ruststream::conformance::message_shape;
/// use ruststream::memory::{ConnectedMemoryBroker, MemoryPublish};
///
/// message_shape::publishes_without_credentials::<ConnectedMemoryBroker, _>(
///     &MemoryPublish,
///     "hunter2",
/// );
/// # }
/// ```
///
/// # Panics
///
/// Panics when `secret` appears in any binding or the reply address location, and when `secret`
/// is empty (a scan for nothing passes for the wrong reason).
#[cfg(feature = "asyncapi")]
pub fn publishes_without_credentials<C, Policy>(policy: &Policy, secret: &str)
where
    C: ConnectedBroker,
    Policy: PublishPolicy<C>,
{
    assert!(
        !secret.is_empty(),
        "pass the password the policy was configured with; scanning for an empty string passes \
         whatever the policy does",
    );
    let channel = "conformance.credentials";
    let described = serde_json::json!({
        "channel": policy.channel_bindings(channel),
        "operation": policy.operation_bindings(channel),
        "message": policy.message_bindings(channel),
        "reply": policy.reply_address_location(),
    });
    // The failure names no binding: the text it would print is the password it found.
    assert!(
        !carries_secret(&described, secret),
        "the publish policy's bindings carry the password. A published document is shared: keep \
         credentials out of every binding body.",
    );
}

/// Whether any string in `value`, a key or a value, contains `secret`. Read off the decoded tree
/// rather than the serialized text, where a quote, a backslash or a control character in the
/// password is escaped and a plain search for it misses.
#[cfg(feature = "asyncapi")]
pub(crate) fn carries_secret(value: &Value, secret: &str) -> bool {
    match value {
        Value::String(text) => text.contains(secret),
        Value::Array(items) => items.iter().any(|item| carries_secret(item, secret)),
        Value::Object(fields) => fields
            .iter()
            .any(|(key, field)| key.contains(secret) || carries_secret(field, secret)),
        Value::Null | Value::Bool(_) | Value::Number(_) => false,
    }
}

/// Fails when a broker configured with several addresses describes any of them with its
/// credentials.
///
/// A broker that connects to a cluster describes its addresses in one server description, and
/// each address is a URL that may carry a user and a password. `make_broker` receives three such
/// URLs under `scheme`, each with the same user and password, and builds the broker the way a
/// deployment would from them (joining them into one setting where the broker takes a list).
/// The document built from its [`DescribeServer`] must carry none of the userinfo, whichever
/// addresses it names, and must name at least one of them with its port (what
/// [`ServerSpec::host_from_url`](crate::ServerSpec::host_from_url) returns for it): a
/// description that names no address passes the scan for the wrong reason.
///
/// # Examples
///
/// ```
/// # #[cfg(all(feature = "asyncapi", feature = "memory"))]
/// # fn demo() {
/// use std::future::Future;
///
/// use ruststream::conformance::message_shape;
/// use ruststream::memory::{ConnectedMemoryBroker, MemoryBroker, MemoryError};
/// use ruststream::{Broker, DescribeServer, ServerSpec};
///
/// /// A broker configured with a list of addresses, described the way the contract asks.
/// struct Cluster {
///     addrs: Vec<String>,
/// }
///
/// impl Broker for Cluster {
///     type Error = MemoryError;
///     type Connected = ConnectedMemoryBroker;
///
///     fn connect(self) -> impl Future<Output = Result<Self::Connected, Self::Error>> + Send {
///         MemoryBroker::new().connect()
///     }
/// }
///
/// impl DescribeServer for Cluster {
///     fn describe_server(&self) -> ServerSpec {
///         let hosts: Vec<String> =
///             self.addrs.iter().map(|addr| ServerSpec::host_from_url(addr)).collect();
///         ServerSpec::new(hosts.join(","), "nats")
///     }
/// }
///
/// message_shape::describes_addresses_without_credentials(
///     |addrs| Cluster { addrs: addrs.iter().map(|addr| (*addr).to_owned()).collect() },
///     "nats",
/// );
/// # }
/// ```
///
/// # Panics
///
/// Panics when the document carries the user or the password of any address, or names none of
/// the addresses.
#[cfg(feature = "asyncapi")]
pub fn describes_addresses_without_credentials<B, MkBroker>(make_broker: MkBroker, scheme: &str)
where
    B: DescribeServer,
    MkBroker: FnOnce(&[&str]) -> B,
{
    const USER: &str = "conformance-user";
    const SECRET: &str = "conformance-secret-7f3a";
    const HOSTS: [&str; 3] = [
        "conformance-a.invalid:4101",
        "conformance-b.invalid:4102",
        "conformance-c.invalid:4103",
    ];

    let urls = HOSTS.map(|host| format!("{scheme}://{USER}:{SECRET}@{host}"));
    let addrs = urls.each_ref().map(String::as_str);
    let broker = make_broker(&addrs);
    let app = RustStream::new(AppInfo::new("conformance", "0.0.0"))
        .server("broker", broker.describe_server());
    let document = build_spec(&app)
        .to_json()
        .expect("the generated document must serialize");

    for leaked in [SECRET, USER] {
        assert!(
            !document.contains(leaked),
            "the server description of a broker configured with several addresses carries their \
             userinfo. Describe each address through ServerSpec::host_from_url. Got: {document}",
        );
    }
    assert!(
        HOSTS.iter().any(|host| document.contains(host)),
        "the server description names none of the configured addresses, so the scan for their \
         credentials proves nothing. Describe each address through ServerSpec::host_from_url. \
         Got: {document}",
    );
}

/// Awaits `work` within [`STEP_TIMEOUT`], failing the check with `what` when it hangs.
async fn within<Work: Future>(work: Work, what: &str) -> Work::Output {
    timeout(STEP_TIMEOUT, work)
        .await
        .unwrap_or_else(|_| panic!("{what} hung for {STEP_TIMEOUT:?}"))
}

/// Publishes `msg` within [`STEP_TIMEOUT`], failing the check with `what` when the publish hangs.
async fn publish_within<Pub: Publisher>(
    publisher: &Pub,
    msg: OutgoingFor<'_, Pub::Payload>,
    options: Option<&Pub::Options>,
    what: &str,
) -> Result<(), Pub::Error> {
    within(publisher.publish(msg, options), what).await
}

/// Awaits the next delivery, failing the check on a timeout, an ended stream or an error.
async fn expect_delivery<S, M, E>(stream: &mut S, label: &str) -> M
where
    S: Stream<Item = Result<M, E>> + Unpin,
    M: IncomingMessage,
    E: fmt::Debug,
{
    next_delivery(stream, Instant::now() + DELIVERY_TIMEOUT, label)
        .await
        .unwrap_or_else(|| panic!("{label}: no delivery within {DELIVERY_TIMEOUT:?}"))
}

/// Awaits the next delivery until `deadline`, `None` when nothing arrived by then; fails the check
/// on an ended stream or an error.
async fn next_delivery<S, M, E>(stream: &mut S, deadline: Instant, label: &str) -> Option<M>
where
    S: Stream<Item = Result<M, E>> + Unpin,
    M: IncomingMessage,
    E: fmt::Debug,
{
    let item = timeout_at(deadline, stream.next()).await.ok()?;
    Some(
        item.unwrap_or_else(|| {
            panic!("{label}: the subscription ended before the expected delivery")
        })
        .unwrap_or_else(|err| panic!("{label}: the subscription yielded an error: {err:?}")),
    )
}

/// Reads what arrives within [`QUIET_PERIOD`] after the last expected delivery, failing the check
/// when `refused` names one of them; anything else (a redelivery of an accepted message, which an
/// at-least-once transport may make) is settled.
async fn nothing_refused_arrives<S, M, E, Refused>(stream: &mut S, check: &str, refused: Refused)
where
    S: Stream<Item = Result<M, E>> + Unpin,
    M: IncomingMessage,
    E: fmt::Debug,
    Refused: Fn(&[u8]) -> Option<String>,
{
    let until = Instant::now() + QUIET_PERIOD;
    while let Some(msg) = next_delivery(stream, until, check).await {
        if let Some(what) = refused(msg.payload()) {
            refused_arrived(check, &what);
        }
        settle(msg, check).await;
    }
}

/// Acknowledges `msg`, accepting a transport that cannot acknowledge.
async fn settle<M: IncomingMessage>(msg: M, label: &str) {
    match msg.ack().await {
        Ok(()) | Err(AckError::Unsupported) => {}
        Err(other) => panic!("{label}: ack must succeed or be unsupported, got: {other:?}"),
    }
}

#[cfg(all(test, feature = "memory"))]
mod tests;