ruststream 0.7.0-rc.11

//! Publishing subscribers: a handler whose return value is encoded and published.
//!
//! Generated by `#[subscriber("name", publish("reply-name"))]` and mounted with
//! [`BrokerScope::include`](super::BrokerScope::include), chaining
//! [`.out(Reply, ..)`](super::RouterWith::out) to name the reply publish policy
//! (without it, the statement commits with the broker's default policy); on a
//! [`Router`](super::Router) the same chain ends in `.build()`, which is what commits the
//! registration. At startup the wiring pairs into the live reply sink (the connection plus the
//! reply codec). The destination name comes from the macro; the publisher and codec come from
//! the chain.

use std::future::Future;

use bytes::BytesMut;
use serde::Serialize;
use tracing::warn;

use crate::codec::Codec;
#[cfg(feature = "testing")]
use crate::testing::coordinator::{Origin, PipelinePublish, publishing_to};
use crate::{IncomingMessage, OutgoingMessage, PayloadForm, Publisher};

use super::context::Context;
use super::dispatch::Workers;
use super::failure::{FailurePolicies, FailurePolicy};
use super::handler::{Handler, HandlerOutcome};
use super::input::{DecodeWith, InputKind};
use super::metadata::{HandlerMetadata, OutgoingMessageMetadata};
use super::publish::{
    ForReply, PublishContext, PublishIdentity, PublishPipeline, PublishTransform, TypedPublisher,
};

/// The reply-wiring axis: how a handler's reply value leaves the service.
///
/// Selected by the wiring the include site's chain built: the encoded stack encodes the
/// reply with its codec and runs its transforms and the scope's publish pipeline, while a bare
/// [`Publisher`] sends the reply's bytes as-is (no codec, no transforms, no pipeline). One
/// [`PublishingHandler`] serves both, so the encoded and the byte reply forms differ only in
/// what pairs at the include site.
pub(crate) trait ReplySink<Reply, DeliveryCx, Pipeline>: Send + Sync {
    /// The payload form of the publisher this wiring ends in, so the reply encodes into what
    /// that form carries. See [`Publisher::Payload`].
    type Payload: PayloadForm;

    /// The error surfaced when the reply cannot be published.
    type Error: std::fmt::Display;

    /// Publishes `reply` to `name`.
    fn deliver<'e>(
        &self,
        name: &'e str,
        reply: &'e Reply,
        pipeline: &Pipeline,
        cx: &PublishContext<'_, DeliveryCx>,
        encode: <Self::Payload as PayloadForm>::Encode<'e>,
    ) -> impl Future<Output = Result<(), Self::Error>> + Send;
}

/// What an encoded reply value knows about leaving through a typed stack.
///
/// A `Serialize` reply encodes as the payload; a [`Message`](super::Message) pair additionally
/// serializes its header contract into the outgoing headers. The bound is pipeline-agnostic,
/// which is what lets the routes state it once per registration.
#[diagnostic::on_unimplemented(
    message = "`{Self}` cannot be encoded as a reply",
    note = "an encoded reply is a `serde::Serialize` value (derive it), or a \
            `Message<Headers, Payload>` pair whose halves are; reply bytes that must leave \
            unencoded ride a `#[derive(Serialized)]` reply type instead"
)]
pub(crate) trait EncodeReply: Send + Sync {
    /// Delivers `self` through the typed reply stack.
    #[doc(hidden)]
    fn deliver_typed<'e, Leaf, ReplyCodec, Transforms, Cx, PP>(
        &self,
        stack: &TypedPublisher<Leaf, ReplyCodec, Transforms>,
        name: &'e str,
        pipeline: &PP,
        cx: &PublishContext<'_, Cx>,
        encode: <Leaf::Payload as PayloadForm>::Encode<'e>,
    ) -> impl Future<Output = Result<(), Box<dyn std::error::Error + Send + Sync>>> + Send
    where
        Leaf: Publisher,
        ReplyCodec: Codec,
        Transforms: PublishTransform<ForReply<Cx>, Leaf::Options>,
        Cx: Sync,
        PP: PublishPipeline;
}

impl<Reply: Serialize + Send + Sync> EncodeReply for Reply {
    // The inner future is returned rather than awaited: this is a spelling of the reply's shape,
    // and a state machine of its own here would sit in every dispatch's future for nothing.
    fn deliver_typed<'e, Leaf, ReplyCodec, Transforms, Cx, PP>(
        &self,
        stack: &TypedPublisher<Leaf, ReplyCodec, Transforms>,
        name: &'e str,
        pipeline: &PP,
        cx: &PublishContext<'_, Cx>,
        encode: <Leaf::Payload as PayloadForm>::Encode<'e>,
    ) -> impl Future<Output = Result<(), Box<dyn std::error::Error + Send + Sync>>> + Send
    where
        Leaf: Publisher,
        ReplyCodec: Codec,
        Transforms: PublishTransform<ForReply<Cx>, Leaf::Options>,
        Cx: Sync,
        PP: PublishPipeline,
    {
        stack.publish(name, self, pipeline, cx, encode)
    }
}

impl<Hd, Pd> EncodeReply for super::Message<Hd, Pd>
where
    Hd: Serialize + Send + Sync,
    Pd: Serialize + Send + Sync,
{
    fn deliver_typed<'e, Leaf, ReplyCodec, Transforms, Cx, PP>(
        &self,
        stack: &TypedPublisher<Leaf, ReplyCodec, Transforms>,
        name: &'e str,
        pipeline: &PP,
        cx: &PublishContext<'_, Cx>,
        encode: <Leaf::Payload as PayloadForm>::Encode<'e>,
    ) -> impl Future<Output = Result<(), Box<dyn std::error::Error + Send + Sync>>> + Send
    where
        Leaf: Publisher,
        ReplyCodec: Codec,
        Transforms: PublishTransform<ForReply<Cx>, Leaf::Options>,
        Cx: Sync,
        PP: PublishPipeline,
    {
        stack.publish_pair(name, &self.headers, &self.body, pipeline, cx, encode)
    }
}

/// The encoded wiring: the reply serializes through the stack's reply codec, then travels the
/// stack's transforms and the scope's publish pipeline. A [`Message`](super::Message) reply
/// additionally carries its typed header contract into the outgoing headers.
impl<Reply, DeliveryCx, Pipeline, Leaf, ReplyCodec, Transforms>
    ReplySink<Reply, DeliveryCx, Pipeline> for TypedPublisher<Leaf, ReplyCodec, Transforms>
where
    Reply: EncodeReply,
    DeliveryCx: Sync,
    Pipeline: PublishPipeline,
    Leaf: Publisher,
    ReplyCodec: Codec,
    Transforms: PublishTransform<ForReply<DeliveryCx>, Leaf::Options>,
{
    type Payload = Leaf::Payload;
    type Error = Box<dyn std::error::Error + Send + Sync>;

    // As in `EncodeReply`: the wiring's own step is the choice of lane, and the future it
    // hands back is the publish itself.
    fn deliver<'e>(
        &self,
        name: &'e str,
        reply: &'e Reply,
        pipeline: &Pipeline,
        cx: &PublishContext<'_, DeliveryCx>,
        encode: <Self::Payload as PayloadForm>::Encode<'e>,
    ) -> impl Future<Output = Result<(), Self::Error>> + Send {
        reply.deliver_typed(self, name, pipeline, cx, encode)
    }
}

/// The byte wiring: a bare [`Publisher`] sends a [`Serialized`](super::Serialized) reply's own
/// bytes, unencoded.
impl<Reply, DeliveryCx, Pipeline, Bare> ReplySink<Reply, DeliveryCx, Pipeline> for Bare
where
    Reply: super::Serialized + Sync,
    DeliveryCx: Sync,
    Pipeline: Send + Sync,
    Bare: Publisher,
{
    type Payload = Bare::Payload;
    // Erased, because two unrelated failures reach this point: the reply's own encoder and the
    // publisher. The axis only ever reports its error, so nothing downstream matches on it.
    type Error = Box<dyn std::error::Error + Send + Sync>;

    async fn deliver<'e>(
        &self,
        name: &'e str,
        reply: &'e Reply,
        _pipeline: &Pipeline,
        _cx: &PublishContext<'_, DeliveryCx>,
        encode: <Self::Payload as PayloadForm>::Encode<'e>,
    ) -> Result<(), Self::Error> {
        // The dispatch loop's buffer, for a reply that writes its bytes rather than holding
        // them; a reply that already holds them leaves it untouched. Which of the two a
        // publisher is handed follows its own declaration, as on every other publish position.
        let payload = <Bare::Payload as PayloadForm>::serialized(reply, encode)?;
        let msg = OutgoingMessage::with_payload(name, payload);
        #[cfg(feature = "testing")]
        let recorded = PipelinePublish::capture(&msg);
        let sent = self.publish(msg, None).await;
        #[cfg(feature = "testing")]
        if sent.is_ok() {
            recorded.sent();
        }
        sent.map_err(Into::into)
    }
}

/// A subscriber definition that produces a reply to publish.
///
/// The generated type carries the subscribe name, the reply name, and the reply type. *How* the
/// reply is encoded (codec) and *through which* connection it is sent come from the wiring the
/// mount site's chain built.
#[diagnostic::on_unimplemented(
    message = "`{Self}` is not a mountable reply definition",
    note = "a reply needs a destination before it mounts: declare one on the reply type \
            (`#[derive(Outgoing)]` with `#[outgoing(name = \"..\")]`), or name the default with \
            `publish(\"subject\")` on the attribute and `.to(\"subject\")` on the chain"
)]
pub trait PublishingDef: Send + Sync {
    /// The input kind the handler consumes ([`Decoded<T>`](super::Decoded) for a typed `&T`
    /// parameter, [`RawBytes`](super::RawBytes) for a raw `&[u8]` one).
    type Input: InputKind;

    /// The tuple of startup-injected parameters ([`Out`](super::Out), ...; `()` when the
    /// signature carries none), resolved like
    /// [`InjectDef::Injections`](super::InjectDef::Injections).
    type Injections;

    /// The reply type the handler produces, encoded and published.
    type Reply;

    /// The broker's typed per-delivery context the handler reads by key, mirroring
    /// [`SubscriberDef::Context`](super::SubscriberDef::Context) (`()` when the handler names
    /// none). It is threaded onto the reply's static
    /// [`PublishTransform`](super::PublishTransform) so a publish transform can stamp the delivery's trace
    /// or correlation id on the reply.
    type Context;

    /// The subscription source this handler binds to (see
    /// [`SubscriberDef::Source`](super::SubscriberDef::Source)).
    type Source;

    /// Builds the subscription source (fresh each call).
    fn source(&self) -> Self::Source;

    /// The name (subject / channel) the reply is published to.
    fn reply_name(&self) -> &str;

    /// The concurrency policy for this subscriber's dispatch loop. The macro fills this in from
    /// the `workers(..)` argument; the default is sequential dispatch.
    fn workers(&self) -> Workers {
        Workers::sequential()
    }

    /// The failure policy for a handler panic and a decode failure. The macro fills this in from
    /// the `on_failure(panic = .., decode = ..)` argument; the default fails fast on a panic and
    /// drops on a decode failure.
    fn failure_policies(&self) -> FailurePolicies {
        FailurePolicies::default()
    }

    /// An optional human description (from the handler's doc comment), for `AsyncAPI`.
    fn description(&self) -> Option<&str> {
        None
    }

    /// The input type's serialized JSON Schema, when it implements [`schemars::JsonSchema`] and the
    /// `asyncapi` feature is on. The macro fills this in; the default omits it.
    fn input_schema(&self) -> Option<String> {
        None
    }

    /// The serialized JSON Schema of the handler's typed header contract (its
    /// [`Headers<T>`](super::Headers) parameter), when `T` implements
    /// [`schemars::JsonSchema`] and the `asyncapi` feature is on. The macro fills this in; the
    /// default omits it.
    fn headers_schema(&self) -> Option<String> {
        None
    }

    /// The messages this handler publishes, for the `AsyncAPI` `send` operations: the reply
    /// message, plus every `Out` slot dictionary entry. The macro fills this in; the default
    /// declares nothing. Called once at registration.
    fn outgoing(&self) -> Vec<OutgoingMessageMetadata> {
        Vec::new()
    }

    /// The input type's [`Message`](crate::MessageInfo) name, when it implements that trait. The macro
    /// fills this in; the default omits it.
    fn message_name(&self) -> Option<&'static str> {
        None
    }

    /// The input type's [`Message`](crate::MessageInfo) description, when it implements that trait.
    /// The macro fills this in; the default omits it.
    fn message_description(&self) -> Option<&'static str> {
        None
    }
}

/// Runs a [`PublishingDef`]'s handler body over an app state of type `S`.
///
/// Split from [`PublishingDef`] so a handler that ignores the app state is generic over `S` (mounts
/// on any app), while one that reads it via [`Context::state`](super::Context::state) implements
/// this only for its declared `S` - the same shape as [`Handler<M, C, S>`](super::Handler), so the
/// state match is checked at compile time without pinning a single `State` on the def.
pub(crate) trait PublishingCall<S>: PublishingDef {
    /// Runs the handler body.
    ///
    /// `Ok(reply)` is encoded and published to [`reply_name`](PublishingDef::reply_name), then the
    /// incoming message is acked. `Err(outcome)` skips publishing and the dispatcher settles by
    /// the returned [`HandlerOutcome`] (for example [`HandlerOutcome::retry`] to ask for
    /// redelivery), running its continuation after the settle.
    fn call(
        &self,
        input: &<Self::Input as InputKind>::Target,
        injections: &Self::Injections,
        ctx: &mut Context<'_, Self::Context, S>,
    ) -> impl Future<Output = Result<Self::Reply, HandlerOutcome>> + Send;
}

/// Builds the registration metadata for a publishing definition mounted under `name`.
pub(crate) fn publishing_metadata<D: PublishingDef>(name: String, def: &D) -> HandlerMetadata {
    let mut meta = HandlerMetadata::raw(name)
        .with_output_type(std::any::type_name::<D::Reply>())
        .with_def_details(
            def.description(),
            def.input_schema(),
            def.headers_schema(),
            def.message_name(),
            def.message_description(),
        );
    meta.input_type = <D::Input as InputKind>::input_label();
    meta.deserialized = <D::Input as InputKind>::DESERIALIZED;
    meta.outgoing = def.outgoing();
    meta
}

/// The [`Handler`] built from a [`PublishingDef`]: decode, run, deliver the reply, ack.
///
/// `DecodeCodec` decodes the incoming message; the reply leaves through the [`ReplySink`]
/// `Wiring` (encoded by the reply stack, or byte-for-byte through a bare publisher) to the
/// definition's [`reply_name`](PublishingDef::reply_name). A handler returning `Err(result)`
/// skips the publish; a failed reply publish nacks the incoming message with `requeue = true`,
/// so the broker redelivers it instead of silently losing the reply.
pub(crate) struct PublishingHandler<
    Def: PublishingDef,
    DecodeCodec,
    Wiring,
    Pipeline = PublishIdentity,
> {
    pub(crate) def: Def,
    pub(crate) codec: DecodeCodec,
    pub(crate) publisher: Wiring,
    pub(crate) pipeline: Pipeline,
    pub(crate) injections: Def::Injections,
    pub(crate) decode: FailurePolicy,
    /// The broker the reply publisher was paired against, which its replies go to.
    #[cfg(feature = "testing")]
    pub(crate) origin: Origin,
}

impl<Def: PublishingDef, DecodeCodec, Wiring, Pipeline> std::fmt::Debug
    for PublishingHandler<Def, DecodeCodec, Wiring, Pipeline>
{
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("PublishingHandler").finish_non_exhaustive()
    }
}

impl<Msg, Def, DecodeCodec, Wiring, Pipeline, State> Handler<Msg, Def::Context, State>
    for PublishingHandler<Def, DecodeCodec, Wiring, Pipeline>
where
    Msg: IncomingMessage,
    Def: PublishingCall<State>,
    Def::Input: DecodeWith<DecodeCodec>,
    Def::Injections: Send + Sync,
    Def::Reply: Send + Sync,
    Def::Context: Send + Sync,
    // `Send + Sync` rather than `Codec`: what the input needs of the codec is exactly what
    // `DecodeWith` asks for, and a raw input asks for nothing - so demanding a codec would
    // shut the byte-in/byte-out handler out of a build with no codec feature at all. The
    // thread bounds stay, because the handler is held across an await.
    DecodeCodec: Send + Sync,
    Wiring: ReplySink<Def::Reply, Def::Context, Pipeline>,
    Pipeline: Send + Sync,
    State: Send + Sync,
{
    async fn handle(
        &self,
        msg: &Msg,
        ctx: &mut Context<'_, Def::Context, State>,
    ) -> HandlerOutcome {
        // The publishing path: decode, run, publish the reply, then ack. A body's `Err` outcome
        // (with any `and_after` continuation it carries) settles the delivery directly. The
        // decode product lives on this stack frame and the handler borrows its view.
        let owned = match <Def::Input as DecodeWith<DecodeCodec>>::decode(&self.codec, msg) {
            Ok(value) => value,
            Err(err) => {
                warn!(
                    target: "ruststream::dispatch",
                    subscription = %ctx.name(),
                    message_type = <Def::Input as InputKind>::input_label(),
                    error = %err,
                    "codec decode failed",
                );
                #[cfg(any(feature = "testing", feature = "otel"))]
                ctx.mark_decode_failed();
                return match self.decode {
                    FailurePolicy::FailFast => {
                        ctx.fail_fast(&format!("decode failed: {err}"));
                        HandlerOutcome::drop()
                    }
                    other => other
                        .settlement()
                        .map_or_else(HandlerOutcome::drop, Into::into),
                };
            }
        };
        let view = <Def::Input as InputKind>::view(&owned, msg.payload());
        let reply = match self.def.call(view, &self.injections, ctx).await {
            Ok(reply) => reply,
            Err(outcome) => return outcome,
        };
        let name = self.def.reply_name();
        // The loop's buffer where the dispatch lent one, and one of this publish's own where it
        // did not; `BytesMut::new` allocates nothing, so the second costs the first nothing.
        let mut own = BytesMut::new();
        let encode = ctx.take_encode_buffer().unwrap_or(&mut own);
        let slot = <<Wiring as ReplySink<Def::Reply, Def::Context, Pipeline>>::Payload as PayloadForm>::encode_slot(
            encode,
        );
        let pubcx = PublishContext::new(ctx.name(), ctx.delivery_headers(), ctx.cx_ref());
        let publish = self
            .publisher
            .deliver(name, &reply, &self.pipeline, &pubcx, slot);
        #[cfg(feature = "testing")]
        let publish = publishing_to(self.origin, publish);
        if let Err(err) = publish.await {
            warn!(
                target: "ruststream::dispatch",
                subscription = %ctx.name(),
                reply = %name,
                reply_type = std::any::type_name::<Def::Reply>(),
                error = %err,
                "reply publish failed",
            );
            return HandlerOutcome::retry();
        }
        HandlerOutcome::ack()
    }
}

#[cfg(test)]
mod tests {
    use std::future::ready;

    use super::{PublishingCall, PublishingDef, publishing_metadata};
    use crate::HeaderMap;
    use crate::Name;
    use crate::runtime::context::Context;
    use crate::runtime::dispatch::{Delivery, Workers};
    use crate::runtime::failure::FailurePolicies;
    use crate::runtime::handler::HandlerOutcome;

    /// A hand-written publishing def overriding nothing optional, pinning the trait defaults that
    /// the macro always fills in.
    struct ManualPub;

    impl PublishingDef for ManualPub {
        type Input = crate::runtime::input::Decoded<u32>;
        type Injections = ();
        type Reply = u32;
        type Context = ();
        type Source = Name;

        fn source(&self) -> Name {
            Name::new("in")
        }

        // The trait signature returns `&str` (tied to `&self`); the macro-generated impls do the
        // same, so this hand-written one cannot narrow to `&'static str` without diverging.
        #[allow(clippy::unnecessary_literal_bound)]
        fn reply_name(&self) -> &str {
            "out"
        }
    }

    // Ignores the app state, so it is generic over it (mounts on any app).
    impl<S: Send + Sync> PublishingCall<S> for ManualPub {
        fn call(
            &self,
            input: &u32,
            (): &(),
            _ctx: &mut Context<'_, (), S>,
        ) -> impl Future<Output = Result<u32, HandlerOutcome>> {
            ready(Ok(*input))
        }
    }

    #[tokio::test]
    async fn defaults_metadata_and_call() {
        let def = ManualPub;
        assert_eq!(def.workers(), Workers::sequential());
        assert_eq!(def.failure_policies(), FailurePolicies::default());
        assert!(def.description().is_none());
        assert!(def.input_schema().is_none());
        assert!(def.headers_schema().is_none());
        assert!(def.outgoing().is_empty());
        assert!(def.message_name().is_none());
        assert!(def.message_description().is_none());
        assert_eq!(def.reply_name(), "out");
        let _source = def.source();

        let meta = publishing_metadata("in".to_owned(), &def);
        assert_eq!(meta.name, "in");

        let state = ();
        let delivery = Delivery::empty();
        let headers = HeaderMap::new();
        let mut ctx = Context::new("in", &headers, &state, (), &delivery);
        assert_eq!(def.call(&5, &(), &mut ctx).await.unwrap(), 5);
    }

    /// The two diagnostics of the publishing path. They are asserted on the handler itself
    /// because the subject is the warning's content, and a field value is only evaluated while a
    /// subscriber listens.
    #[cfg(all(feature = "memory", feature = "json", feature = "logging"))]
    mod diagnostics {
        use std::future::ready;

        use super::ManualPub;
        use crate::codec::JsonCodec;
        use crate::memory::{MemoryBroker, MemoryError};
        use crate::runtime::context::Context;
        use crate::runtime::dispatch::Delivery;
        use crate::runtime::failure::FailurePolicy;
        use crate::runtime::handler::Handler;
        use crate::runtime::publish::{PublishIdentity, TypedPublisher};
        use crate::runtime::publishing::PublishingHandler;
        #[cfg(feature = "testing")]
        use crate::testing::coordinator::Origin;
        use crate::testkit::delivery::one_delivery;
        use crate::testkit::log_capture::{find, start};
        use crate::{HeaderMap, Lend, OutgoingMessage, Publisher};

        /// A publisher that always refuses, modelling a broker rejecting the reply.
        struct Rejecting;

        impl Publisher for Rejecting {
            type Payload = Lend;
            type Error = MemoryError;
            type Options = ();

            fn publish(
                &self,
                _msg: OutgoingMessage<'_, &[u8]>,
                _options: Option<&Self::Options>,
            ) -> impl Future<Output = Result<(), Self::Error>> {
                ready(Err(MemoryError::ShutDown))
            }
        }

        /// The decode diagnostic names the subscription and the type that was expected, so the
        /// offending producer is findable from the logs; `fail_fast` still settles the message
        /// out of the way (the teardown is what makes the failure loud).
        #[tokio::test]
        async fn a_decode_failure_names_the_subscription_and_the_expected_type() {
            let broker = MemoryBroker::new();
            let msg = one_delivery(&broker, "in", b"not json").await;
            let handler = PublishingHandler {
                def: ManualPub,
                codec: JsonCodec,
                publisher: TypedPublisher::with_codec(broker.publisher(), JsonCodec),
                pipeline: PublishIdentity,
                injections: (),
                decode: FailurePolicy::FailFast,
                #[cfg(feature = "testing")]
                origin: Origin::default(),
            };

            let state = ();
            let delivery = Delivery::empty();
            let headers = HeaderMap::new();
            let mut ctx = Context::new("in", &headers, &state, (), &delivery);

            let (events, guard) = start();
            let settle = handler.handle(&msg, &mut ctx).await;
            drop(guard);

            let failure = find(&events, "codec decode failed");
            assert_eq!(failure.get("subscription").map(String::as_str), Some("in"));
            assert_eq!(failure.get("message_type").map(String::as_str), Some("u32"));
            assert!(
                failure.get("error").is_some_and(|e| e.contains("decode")),
                "the diagnostic must carry the codec error: {failure:?}",
            );
            assert!(settle.is_drop());
        }

        /// A reply the broker rejects is diagnosed with the reply channel and asks for a
        /// redelivery: the reply is retried, never silently lost.
        #[tokio::test]
        async fn a_failed_reply_publish_names_the_reply_channel_and_retries() {
            let broker = MemoryBroker::new();
            let msg = one_delivery(&broker, "in", b"5").await;
            let handler = PublishingHandler {
                def: ManualPub,
                codec: JsonCodec,
                publisher: TypedPublisher::with_codec(Rejecting, JsonCodec),
                pipeline: PublishIdentity,
                injections: (),
                decode: FailurePolicy::Drop,
                #[cfg(feature = "testing")]
                origin: Origin::default(),
            };

            let state = ();
            let delivery = Delivery::empty();
            let headers = HeaderMap::new();
            let mut ctx = Context::new("in", &headers, &state, (), &delivery);

            let (events, guard) = start();
            let settle = handler.handle(&msg, &mut ctx).await;
            drop(guard);

            let failure = find(&events, "reply publish failed");
            assert_eq!(failure.get("subscription").map(String::as_str), Some("in"));
            assert_eq!(failure.get("reply").map(String::as_str), Some("out"));
            assert_eq!(failure.get("reply_type").map(String::as_str), Some("u32"));
            assert!(
                failure
                    .get("error")
                    .is_some_and(|e| e.contains("shut down")),
                "the diagnostic must carry the broker error: {failure:?}",
            );
            assert!(settle.is_retry());
        }
    }

    /// The mounted handler keeps its wiring out of Debug: it holds a live connection, and a
    /// registration dump must not print one.
    #[cfg(all(feature = "memory", feature = "json"))]
    #[test]
    fn a_mounted_publishing_handler_hides_its_wiring() {
        use crate::codec::JsonCodec;
        use crate::memory::MemoryPublish;
        use crate::runtime::failure::FailurePolicy;
        use crate::runtime::publish::{PublishIdentity, TypedPublisher};
        use crate::runtime::publishing::PublishingHandler;
        #[cfg(feature = "testing")]
        use crate::testing::coordinator::Origin;

        let handler = PublishingHandler {
            def: ManualPub,
            codec: JsonCodec,
            publisher: TypedPublisher::with_codec(MemoryPublish, JsonCodec),
            pipeline: PublishIdentity,
            injections: (),
            decode: FailurePolicy::Drop,
            #[cfg(feature = "testing")]
            origin: Origin::default(),
        };

        assert_eq!(format!("{handler:?}"), "PublishingHandler { .. }");
    }
}