Skip to main content

ruststream/runtime/
slot.rs

1//! Marker-identified `Out` slots: the type-level machinery pairing each
2//! [`Out`](super::Out) parameter with the publish policy attached at the include site.
3//!
4//! A handler declares `Out(out): Out<impl Publisher, MySlot>`; the marker (`MySlot`, or the
5//! implicit [`DefaultSlot`]) identifies the slot, and the concrete publisher type is inferred
6//! from the attachment's [`PublishPolicy::Live`](crate::PublishPolicy::Live) - fully monomorphized, no erasure. The pieces:
7//!
8//! - [`OutSlot`] / [`DefaultSlot`]: the marker vocabulary (`#[derive(OutSlot)]` for named ones).
9//! - [`PublishedThrough`]: membership in a marker's `#[publishes(..)]` dictionary, so what the
10//!   document reports as leaving a slot is exactly what the publish builder admits.
11//! - [`SlotPublisher`]: the transparent wrapper the handler actually receives; it delegates the
12//!   publisher capabilities and, under the `testing` feature, attributes publishes to the slot.
13//! - [`HasSlots`] / [`BindSlots`]: the macro-implemented contract on a definition (the marker
14//!   list, and the instantiation of the publisher-generic definition from the bound sources).
15//! - [`BindSlot`] / [`InitSlots`] / [`MissingSlot`]: the include-site builder machinery that
16//!   places each `.out(marker, policy)` attachment into its marker's position, in any order;
17//!   a registration commits only when no `MissingSlot` remains, so a forgotten binding is a
18//!   compile error naming the slot.
19
20use std::borrow::Cow;
21use std::marker::PhantomData;
22use std::time::Duration;
23
24use crate::codec::Codec;
25use crate::runtime::handle::{DeclaresReply, ReplyShape};
26use crate::runtime::metadata::{OutgoingMessageMetadata, PublishDescription};
27use crate::runtime::publish::{
28    AddBatchReplyTransform, AddReplyTransform, CallCodec, CodecSlotOpen, DestinationUse, FitsOffer,
29    ForReply, ForSlot, LowerOutTransforms, MapReplyPolicy, NameReplyCodec, Names, NarrowToUse,
30    PublishTransform, PublishTransformIdentity, PublishTransformStack, PublishingDirectly,
31    RawReplyWiring, Reads, ReplyWiring, SlotStackUse, TransactionalReply, UnnamedCodec,
32};
33use crate::runtime::router::{
34    BatchPublishInjectMount, BatchPublishMount, DefaultReply, PublishInjectMount, PublishMount,
35    RawReplyInjectMount, RawReplyMount, ReplyAttachment,
36};
37#[cfg(feature = "testing")]
38use crate::testing::coordinator::{PipelinePublish, record_slot_publish};
39use crate::{
40    Broker, CallerName, Connected, ConnectedBroker, FixedName, HeaderMap, NameTemplate,
41    OutgoingDestination, OutgoingFor, OwnedTransactions, PublishPolicy, Publisher, RequestReply,
42    TransactionalPublisher,
43};
44
45/// A slot marker: the identity of one [`Out`](super::Out) injection.
46///
47/// Markers decouple a handler signature from the concrete publisher type: the handler names the
48/// slot, the include site names the policy, and the runtime pairs them. Derive it on a unit
49/// struct with `#[derive(OutSlot)]` (`macros` feature); [`NAME`](Self::NAME) labels the slot in
50/// startup errors and test assertions.
51///
52/// # Examples
53///
54/// ```
55/// use ruststream::prelude::*;
56/// use ruststream::runtime::Reads;
57///
58/// #[derive(Clone, Copy, Debug)]
59/// struct Encoded;
60///
61/// impl OutSlot for Encoded {
62///     const NAME: &'static str = "Encoded";
63///     type Destination = Reads;
64/// }
65///
66/// assert_eq!(<Encoded as OutSlot>::NAME, "Encoded");
67/// ```
68pub trait OutSlot: 'static {
69    /// The human-readable slot name, used by diagnostics and test assertions.
70    const NAME: &'static str;
71
72    /// The most a transform on this slot may do to a message's destination.
73    ///
74    /// [`Names`] only where every type the slot publishes leaves its destination open, because a
75    /// type that fixes its own channel is published there and the generated document says so.
76    /// `#[derive(OutSlot)]` folds this out of the `#[publishes(..)]` list through
77    /// [`ListOffer`]; a marker with no list has nothing to fold and writes [`Reads`], which is
78    /// what the implicit [`DefaultSlot`] does - it admits every message, so it can promise
79    /// nothing. Stable Rust has no default for an associated type, so every marker writes the
80    /// line.
81    type Destination: DestinationUse;
82
83    /// The slot's publish dictionary as `AsyncAPI` metadata, one entry per
84    /// `#[publishes(..)]` type. The derive fills this in; the default (a marker without a
85    /// dictionary) publishes nothing declared. Called once at registration; never on the
86    /// publish path.
87    #[must_use]
88    fn outgoing() -> Vec<OutgoingMessageMetadata> {
89        Vec::new()
90    }
91}
92
93/// The implicit marker of a single unnamed `Out<impl Publisher>` parameter.
94///
95/// A handler with one `Out` parameter needs no marker: the parameter binds to this slot, and the
96/// include site attaches its policy by naming the marker like any other, `.out(DefaultSlot,
97/// policy)`.
98///
99/// # Examples
100///
101/// ```
102/// use ruststream::prelude::*;
103///
104/// assert_eq!(<DefaultSlot as OutSlot>::NAME, "default");
105/// ```
106#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Hash)]
107pub struct DefaultSlot;
108
109// The implicit slot admits every declared message, so it can promise nothing about their
110// destinations and lets no transform name one.
111impl OutSlot for DefaultSlot {
112    const NAME: &'static str = "default";
113    type Destination = Reads;
114}
115
116/// The marker of a handler's reply position: `.out(Reply, policy)` names the publish policy the
117/// value a `publish("dest")` handler returns leaves through.
118///
119/// A mount site attaches every publish policy with one verb, and this is the marker of the one
120/// position that is not an [`Out`](super::Out) slot. It carries no dictionary: what a reply may be
121/// is the reply type's own [`ReplyShape`](crate::runtime::ReplyShape), while what leaves a slot is
122/// the slot marker's `#[publishes(..)]` list.
123///
124/// Without the call the reply takes the broker's own
125/// [`DefaultPublish`](crate::DefaultPublish) policy, so naming it is the exception, not the rule.
126///
127/// # Examples
128///
129/// ```
130/// # #[cfg(all(feature = "memory", feature = "macros", feature = "json"))]
131/// # mod demo {
132/// use ruststream::memory::prelude::*;
133/// # use ruststream::subscriber;
134/// # #[derive(serde::Deserialize, schemars::JsonSchema)]
135/// # struct Order { id: u64 }
136/// # #[derive(serde::Serialize, schemars::JsonSchema, ruststream::Outgoing)]
137/// # struct Confirmation { id: u64 }
138///
139/// #[subscriber("orders", publish("confirmations"))]
140/// async fn confirm(order: &Order) -> Confirmation {
141///     Confirmation { id: order.id }
142/// }
143///
144/// fn app() -> RustStream {
145///     RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
146///         b.include(confirm).out(Reply, Publish);
147///     })
148/// }
149/// # }
150/// ```
151#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Hash)]
152pub struct Reply;
153
154/// The reply attachment of a registration whose handler declares no reply: there is no
155/// [`Reply`] position to bind, so `.out(Reply, ..)` on it does not compile.
156#[doc(hidden)]
157#[derive(Debug, Clone, Copy, Default)]
158pub struct NoReply;
159
160/// Membership of a message type in a slot marker's `#[publishes(..)]` dictionary: the type may
161/// leave through a slot identified by `Slot`.
162///
163/// `#[derive(OutSlot)]` emits one impl per listed type, and the publish builder's typed entry
164/// point ([`Slot::message`](crate::runtime::Slot::message)) requires it. That is what keeps
165/// the generated document honest: an unrestricted `Out<impl Publisher, Marker>` reports the
166/// marker's dictionary as what the handler sends, so a message outside it would be a publish the
167/// document never declared.
168///
169/// The membership is declared on the message type, with the slot as the parameter, so the
170/// compile error names the message that is not in the dictionary.
171///
172/// # Examples
173///
174/// ```
175/// use ruststream::prelude::*;
176///
177/// struct Progress;
178///
179/// struct Events;
180///
181/// impl OutSlot for Events {
182///     const NAME: &'static str = "Events";
183///     type Destination = Reads;
184/// }
185///
186/// // What `#[derive(OutSlot)]` + `#[publishes(Progress)]` generates:
187/// impl PublishedThrough<Events> for Progress {}
188///
189/// fn admits<T: PublishedThrough<Slot>, Slot>() {}
190/// admits::<Progress, Events>();
191/// ```
192#[diagnostic::on_unimplemented(
193    message = "the `{Slot}` slot does not publish `{Self}`",
194    note = "the marker lists what leaves through the slot, and the generated document reports \
195            that list: add the type to it (`#[derive(OutSlot)] #[publishes({Self}, ..)]`), or \
196            publish through a slot that lists it"
197)]
198pub trait PublishedThrough<Slot> {}
199
200// The implicit slot has no declaration site to list types on, so it admits every message.
201impl<T> PublishedThrough<DefaultSlot> for T {}
202
203/// What one destination form lets a transform do: [`Names`] where nothing is declared, [`Reads`]
204/// where something is.
205///
206/// A type deriving `Outgoing` without `#[outgoing(name = "..")]` is published where `.to(..)`
207/// says, so a transform that names the destination replaces a choice the call site was making
208/// anyway. A type that fixes its own channel is published there and the document says so, and
209/// nothing may move it.
210#[doc(hidden)]
211pub trait DestinationOffer {
212    /// The most a transform may do to a message declared this way.
213    type Offer: DestinationUse;
214}
215
216impl DestinationOffer for CallerName {
217    type Offer = Names;
218}
219
220impl DestinationOffer for FixedName {
221    type Offer = Reads;
222}
223
224impl DestinationOffer for NameTemplate {
225    type Offer = Reads;
226}
227
228/// The offer of a whole `#[publishes(..)]` dictionary: [`Names`] only while every member leaves
229/// its destination open, because one member that fixes its channel settles the whole slot.
230///
231/// The dual of [`Either`](crate::runtime::publish::Either), which combines what a transform stack
232/// does. Machinery behind [`OutSlot::Destination`].
233#[doc(hidden)]
234pub trait Both<Rhs> {
235    /// The combined offer.
236    type Out;
237}
238
239impl Both<Self> for Names {
240    type Out = Self;
241}
242
243impl Both<Reads> for Names {
244    type Out = Reads;
245}
246
247impl<Rhs> Both<Rhs> for Reads {
248    type Out = Self;
249}
250
251/// A slot marker's `#[publishes(..)]` dictionary as a type-level list, `(First, (Second, ()))`,
252/// folded into what the slot offers a transform.
253///
254/// [`OutSlot::outgoing`] reports the same list as data, for the generated document; this folds it
255/// as types, so `#[derive(OutSlot)]` computes [`OutSlot::Destination`] from the list rather than
256/// the author reasoning it out. A hand-written marker either projects the fold the same way or
257/// writes the offer directly.
258///
259/// # Examples
260///
261/// ```
262/// # #[cfg(feature = "macros")]
263/// # mod demo {
264/// use ruststream::prelude::*;
265/// use ruststream::runtime::ListOffer;
266/// use serde::Serialize;
267///
268/// #[derive(Outgoing, Serialize)]
269/// struct Progress {
270///     percent: u8,
271/// }
272///
273/// #[derive(Outgoing, Serialize)]
274/// #[outgoing(name = "chunks.done")]
275/// struct ChunkDone {
276///     key: String,
277/// }
278///
279/// struct Events;
280///
281/// // What `#[derive(OutSlot)] #[publishes(ChunkDone, Progress)]` writes. `ChunkDone` fixes its
282/// // channel, so the fold comes out `Reads` and the slot lets no transform name a destination.
283/// impl OutSlot for Events {
284///     const NAME: &'static str = "Events";
285///     type Destination = <(ChunkDone, (Progress, ())) as ListOffer>::Offer;
286/// }
287///
288/// fn offers_reads<T: OutSlot<Destination = Reads>>() {}
289/// # fn check() { offers_reads::<Events>(); }
290/// # }
291/// ```
292pub trait ListOffer {
293    /// What every member of the list allows, taken together.
294    type Offer: DestinationUse;
295}
296
297impl ListOffer for () {
298    type Offer = Names;
299}
300
301impl<Head, Tail> ListOffer for (Head, Tail)
302where
303    Head: OutgoingDestination<Form: DestinationOffer>,
304    Tail: ListOffer,
305    <<Head as OutgoingDestination>::Form as DestinationOffer>::Offer:
306        Both<Tail::Offer, Out: DestinationUse>,
307{
308    type Offer = <<<Head as OutgoingDestination>::Form as DestinationOffer>::Offer as Both<
309        Tail::Offer,
310    >>::Out;
311}
312
313/// The live publisher an [`Out`](super::Out) slot injects: the attachment's paired publisher,
314/// wrapped with the slot identity.
315///
316/// The wrapper is a zero-cost newtype (static dispatch, no boxing): every capability of the
317/// underlying publisher ([`Publisher`], [`TransactionalPublisher`], [`OwnedTransactions`],
318/// [`RequestReply`]) is delegated, so an `Out<impl OwnedTransactions, M>` bound holds exactly
319/// when the attached policy's live form supports it. Under the `testing` feature, publishes made
320/// through the wrapper inside a `TestApp`-driven handler are also recorded against the slot,
321/// which is what backs the harness's per-slot assertions.
322///
323/// You never name this type: the handler sees it as `impl Publisher` (or a capability
324/// refinement), and the include machinery constructs it.
325#[derive(Debug)]
326pub struct SlotPublisher<P, M> {
327    inner: P,
328    _slot: PhantomData<fn() -> M>,
329}
330
331impl<P, M> SlotPublisher<P, M> {
332    pub(crate) fn new(inner: P) -> Self {
333        Self {
334            inner,
335            _slot: PhantomData,
336        }
337    }
338
339    /// The paired value this wrapper delegates to.
340    ///
341    /// This is the wrapper's own delegation seam, not the extension point for broker-defined
342    /// capabilities: a handler body holds the arena entry
343    /// ([`Slot`](crate::runtime::Slot)), never this wrapper, so a trait grafted here is
344    /// unreachable from one. A broker whose paired value offers more than the core capability
345    /// vocabulary ([`Publisher`], [`TransactionalPublisher`], [`OwnedTransactions`],
346    /// [`RequestReply`]) - or is not a publisher at all (a per-partition producer cache, a shard
347    /// router) - implements its capability trait for the live value and grafts it onto the entry
348    /// instead, as [`Slot`](crate::runtime::Slot) documents.
349    ///
350    /// Publishes made through the value obtained here bypass the slot's test-capture
351    /// attribution, like a settled owned transaction's buffer: assert on the broker's publish
352    /// log for those.
353    ///
354    /// # Examples
355    ///
356    /// ```
357    /// use ruststream::runtime::{OutSlot, SlotPublisher};
358    ///
359    /// /// A broker-defined capability the core knows nothing about.
360    /// trait PartitionLanes {
361    ///     fn lane_id(&self, partition: i32) -> String;
362    /// }
363    ///
364    /// // The wrapper's own delegation: a slot entry reaches the live value through it. The
365    /// // impl a handler body needs is the one on `Slot`, not this one.
366    /// impl<P: PartitionLanes, M: OutSlot> PartitionLanes for SlotPublisher<P, M> {
367    ///     fn lane_id(&self, partition: i32) -> String {
368    ///         self.inner().lane_id(partition)
369    ///     }
370    /// }
371    /// ```
372    #[must_use]
373    pub fn inner(&self) -> &P {
374        &self.inner
375    }
376}
377
378impl<P: Publisher, M: OutSlot> Publisher for SlotPublisher<P, M> {
379    type Payload = P::Payload;
380    type Error = P::Error;
381    type Options = P::Options;
382
383    async fn publish(
384        &self,
385        msg: OutgoingFor<'_, Self::Payload>,
386        options: Option<&Self::Options>,
387    ) -> Result<(), Self::Error> {
388        #[cfg(feature = "testing")]
389        record_slot_publish(M::NAME, &msg, options);
390        self.inner.publish(msg, options).await
391    }
392
393    // The slot is attribution, not policy: whatever the broker's publisher contributes to every
394    // message it sends has to reach the builder through the wrapper unchanged.
395    fn base_headers(&self) -> Option<&HeaderMap> {
396        self.inner.base_headers()
397    }
398}
399
400impl<P: TransactionalPublisher, M: OutSlot> TransactionalPublisher for SlotPublisher<P, M> {
401    async fn begin_transaction(&self) -> Result<(), Self::Error> {
402        self.inner.begin_transaction().await
403    }
404
405    async fn commit(&self) -> Result<(), Self::Error> {
406        self.inner.commit().await
407    }
408
409    async fn abort(&self) -> Result<(), Self::Error> {
410        self.inner.abort().await
411    }
412}
413
414// Transaction values own their buffer and leave the slot's scope, so publishes made through
415// them are visible in the broker's publish log but are not attributed to the slot.
416impl<P: OwnedTransactions, M: OutSlot> OwnedTransactions for SlotPublisher<P, M> {
417    type Transaction = P::Transaction;
418
419    async fn transaction(&self) -> Result<Self::Transaction, Self::Error> {
420        self.inner.transaction().await
421    }
422}
423
424impl<P: RequestReply, M: OutSlot> RequestReply for SlotPublisher<P, M> {
425    type Reply = P::Reply;
426
427    async fn request(
428        &self,
429        msg: OutgoingFor<'_, Self::Payload>,
430        timeout: Duration,
431    ) -> Result<Self::Reply, Self::Error> {
432        // A request takes no call-site options, so what it carried is the policy's defaults.
433        #[cfg(feature = "testing")]
434        record_slot_publish::<P::Options, _>(M::NAME, &msg, None);
435        // A request reaches the broker directly rather than through the publish pipeline, so it
436        // is recorded here for the channel view as well.
437        #[cfg(feature = "testing")]
438        let recorded = PipelinePublish::capture(&msg);
439        let reply = self.inner.request(msg, timeout).await;
440        #[cfg(feature = "testing")]
441        if reply.is_ok() {
442            recorded.sent();
443        }
444        reply
445    }
446}
447
448/// Membership of a message type in an `Out` parameter's declared message list.
449///
450/// The declaration is a tuple listing types, a set-defining type (a `#[derive(MessageInfo)]` type
451/// declares itself, a `#[derive(OutMessages)]` enum declares its variants' models), or the
452/// unrestricted `()` (any dictionary type). The `Index` parameter is inferred per call, like
453/// the slot-binding machinery's positions; a duplicate type in a declaration is rejected where
454/// it is declared, so the index is always unambiguous.
455#[diagnostic::on_unimplemented(
456    message = "`{T}` is not in this Out parameter's declared message set `{Self}`",
457    note = "the third Out argument declares what the handler publishes: list the type there \
458            (`Out<impl Publisher, Marker, (A, B)>`), leave the slot unrestricted with `()`, or \
459            publish through another slot"
460)]
461pub trait ContainsMessage<T, Index> {}
462
463/// The [`ContainsMessage`] index of the unrestricted `()` declaration.
464#[doc(hidden)]
465#[derive(Debug, Clone, Copy)]
466pub struct Unrestricted;
467
468// The unrestricted declaration: any message the marker's dictionary admits.
469impl<T> ContainsMessage<T, Unrestricted> for () {}
470
471/// Implements [`ContainsMessage`] for each (arity, position) pair of the message-list tuple.
472macro_rules! impl_contains_message {
473    ($(($($before:ident,)* @ $pos:literal $(, $after:ident)*))+) => {$(
474        impl<T $(, $before)* $(, $after)*> ContainsMessage<T, SlotPos<$pos>>
475            for ($($before,)* T, $($after,)*)
476        {
477        }
478    )+};
479}
480
481impl_contains_message! {
482    (@ 0)
483    (@ 0, A1)
484    (A0, @ 1)
485    (@ 0, A1, A2)
486    (A0, @ 1, A2)
487    (A0, A1, @ 2)
488    (@ 0, A1, A2, A3)
489    (A0, @ 1, A2, A3)
490    (A0, A1, @ 2, A3)
491    (A0, A1, A2, @ 3)
492}
493
494/// A declared message set's `AsyncAPI` contribution: one
495/// [`OutgoingMessageMetadata`] entry per member, each read off the member's own declaration.
496///
497/// Implemented by `#[derive(Outgoing)]` (the type declares itself and where it goes) and by
498/// `#[derive(OutMessages)]` on a set enum (each variant's model); `()` falls back to the whole
499/// dictionary. A tuple declaration needs no impl - the `#[subscriber]` macro enumerates its
500/// elements itself.
501///
502/// # Examples
503///
504/// ```
505/// # #[cfg(all(feature = "macros", feature = "json"))]
506/// # mod demo {
507/// use ruststream::prelude::*;
508/// use serde::Serialize;
509///
510/// #[derive(Outgoing, Serialize)]
511/// #[outgoing(name = "chunks.progress")]
512/// struct Progress {
513///     percent: u8,
514/// }
515///
516/// #[derive(Outgoing, Serialize)]
517/// #[outgoing(name = "chunks.done")]
518/// struct ChunkDone {
519///     output_key: String,
520/// }
521///
522/// // A reusable named set: the variants' models are what a handler naming `ConvertSends` as
523/// // its third Out argument publishes. The enum is never constructed.
524/// #[derive(OutMessages)]
525/// enum ConvertSends {
526///     Progress(Progress),
527///     Done(ChunkDone),
528/// }
529/// # }
530/// ```
531#[diagnostic::on_unimplemented(
532    message = "`{Self}` does not define a message set for the `{M}` slot",
533    note = "the third Out argument is a tuple of types, `()` (unrestricted), a \
534            #[derive(Outgoing)] type (declares itself and where it goes), or a \
535            #[derive(OutMessages)] enum (declares its variants' models)"
536)]
537pub trait OutMessages<M: OutSlot> {
538    /// The set's declared outgoing messages, for the generated document. Called once at
539    /// registration; never on the publish path.
540    #[must_use]
541    fn outgoing() -> Vec<OutgoingMessageMetadata>;
542}
543
544// The unrestricted declaration documents the whole dictionary.
545impl<M: OutSlot> OutMessages<M> for () {
546    fn outgoing() -> Vec<OutgoingMessageMetadata> {
547        M::outgoing()
548    }
549}
550
551/// The "not bound yet" placeholder of the `Out` slot marked `M` at the include site.
552///
553/// The marker rides in the type so the compile error of an incomplete registration (a
554/// `.build()` whose attachment tuple still contains a `MissingSlot<..>`) names the slot that
555/// was forgotten. A value of this type never reaches the runtime: committing requires every
556/// position bound.
557#[doc(hidden)]
558#[derive(Debug, Clone, Copy, Default)]
559pub struct MissingSlot<M>(PhantomData<fn() -> M>);
560
561impl<M> MissingSlot<M> {
562    pub(crate) fn new() -> Self {
563        Self(PhantomData)
564    }
565}
566
567/// What one `.out(marker, policy)` call attaches: the slot's publish policy and the
568/// [`PublishTransform`](crate::runtime::PublishTransform) stack the `.transform(..)` steps after it
569/// compose.
570///
571/// The stack is pure declaration, like the policy: it lowers onto the app's publish pipeline at
572/// the mount ([`LowerOutTransforms`]), and the composed pipeline is what the slot entry publishes
573/// through. Machinery; the chain builds it and the mount consumes it, so it is never named in
574/// service code.
575#[doc(hidden)]
576#[derive(Debug)]
577pub struct OutAttachment<M, Policy, Layers = PublishTransformIdentity, Enc = UnnamedCodec> {
578    policy: Policy,
579    layers: Layers,
580    enc: Enc,
581    // The marker the `.out(marker, policy)` call bound, carried so the steps after it can ask what
582    // the slot's own declaration offers: the `.transform(..)` step asks it whether the slot lets a
583    // transform name the destination.
584    _marker: PhantomData<fn() -> M>,
585}
586
587impl<M, Policy> OutAttachment<M, Policy> {
588    /// The attachment a bare `.out(marker, policy)` produces: the policy, no transforms, the
589    /// surface's own codec.
590    pub(crate) fn new(policy: Policy) -> Self {
591        Self {
592            policy,
593            layers: PublishTransformIdentity,
594            enc: UnnamedCodec::new(),
595            _marker: PhantomData,
596        }
597    }
598}
599
600impl<M, Policy, Layers, Enc> OutAttachment<M, Policy, Layers, Enc> {
601    /// Composes one more transform on top of the stack: the `.transform(..)` step.
602    pub(crate) fn add_transform<N>(
603        self,
604        transform: N,
605    ) -> OutAttachment<M, Policy, PublishTransformStack<Layers, N>, Enc> {
606        OutAttachment {
607            policy: self.policy,
608            layers: PublishTransformStack {
609                inner: self.layers,
610                outer: transform,
611            },
612            enc: self.enc,
613            _marker: PhantomData,
614        }
615    }
616
617    /// Fills the slot's codec position: the `.codec(..)` step.
618    pub(crate) fn name_codec<C>(self, codec: C) -> OutAttachment<M, Policy, Layers, CallCodec<C>> {
619        OutAttachment {
620            policy: self.policy,
621            layers: self.layers,
622            enc: CallCodec(codec),
623            _marker: PhantomData,
624        }
625    }
626
627    /// Replaces the policy, keeping everything the chain already named: the `.map_publisher(..)`
628    /// step a broker's own settings trait layers on.
629    pub(crate) fn map_policy(self, f: impl FnOnce(Policy) -> Policy) -> Self {
630        Self {
631            policy: f(self.policy),
632            layers: self.layers,
633            enc: self.enc,
634            _marker: PhantomData,
635        }
636    }
637
638    /// What this slot contributes to the generated document: the destinations its marker
639    /// declares, and what its policy and codec say about each of them.
640    ///
641    /// The policy answers once per destination, because a binding may carry the channel's own
642    /// name and a dictionary names several.
643    ///
644    /// Read before [`wire`](Self::wire) consumes the attachment, because the slot's policy is
645    /// what answers and wiring moves it. Nothing here runs per message.
646    pub(crate) fn describe<C, Surface>(&self) -> Vec<(Cow<'static, str>, PublishDescription)>
647    where
648        M: OutSlot,
649        C: ConnectedBroker,
650        Policy: PublishPolicy<C>,
651        Enc: SlotCodec<Surface>,
652        <Enc as SlotCodec<Surface>>::Codec: Codec,
653    {
654        M::outgoing()
655            .into_iter()
656            .map(|entry| {
657                // A slot publishes where its dictionary says, so no transform on it names a reply
658                // address: that question belongs to the reply position.
659                let description = PublishDescription::of::<C, Policy>(
660                    &self.policy,
661                    Some(<<Enc as SlotCodec<Surface>>::Codec as Codec>::CONTENT_TYPE),
662                    false,
663                    entry.channel.as_ref(),
664                );
665                (entry.channel, description)
666            })
667            .collect()
668    }
669
670    /// Hands back the pieces without folding them: what the deferred-retry position resolves
671    /// from, which keeps the mount site's transform stack out of the publish pipeline so it can
672    /// be run against the delivery being retried.
673    pub(crate) fn into_parts(self) -> (Policy, Layers, Enc) {
674        (self.policy, self.layers, self.enc)
675    }
676
677    /// Splits the attachment into what one slot resolves from at startup: the policy the runtime
678    /// pairs (narrowed to plain sending where this slot's transforms name the destination), the
679    /// encode codec (this slot's own when it named one, the surface's otherwise), and the pipeline
680    /// the entry publishes through.
681    pub(crate) fn wire<Surface, Pipeline, Live>(
682        self,
683        surface: Surface,
684        pipeline: Pipeline,
685    ) -> (SlotPolicy<Layers, Policy, Live>, Enc::Codec, Layers::Out)
686    where
687        M: OutSlot,
688        Enc: SlotCodec<Surface>,
689        Layers: LowerOutTransforms<Pipeline> + SlotStackUse<Live>,
690        <Layers as SlotStackUse<Live>>::Destination: NarrowToUse<Policy>,
691    {
692        let codec = self.enc.resolve(surface);
693        let policy = <<Layers as SlotStackUse<Live>>::Destination as NarrowToUse<Policy>>::narrow(
694            self.policy,
695        );
696        (policy, codec, self.layers.lower(M::NAME, pipeline))
697    }
698}
699
700/// The policy one slot pairs: its own, narrowed where the stack names the destination. Machinery;
701/// the commit spells it so the bound reads.
702#[doc(hidden)]
703pub type SlotPolicy<Layers, Policy, Live> =
704    <<Layers as SlotStackUse<Live>>::Destination as NarrowToUse<Policy>>::Out;
705
706/// Resolves one slot's encode codec: the codec the chain named for that slot, or the registration
707/// surface's own when it named none. The slot counterpart of a reply wiring's
708/// [`NameReplyCodec`] position.
709#[doc(hidden)]
710pub trait SlotCodec<Surface> {
711    /// The resolved codec the slot entry encodes with.
712    type Codec;
713
714    /// Resolves it against the surface's codec.
715    fn resolve(self, surface: Surface) -> Self::Codec;
716}
717
718impl<Surface> SlotCodec<Surface> for UnnamedCodec {
719    type Codec = Surface;
720
721    fn resolve(self, surface: Surface) -> Surface {
722        surface
723    }
724}
725
726impl<Surface, C> SlotCodec<Surface> for CallCodec<C> {
727    type Codec = C;
728
729    fn resolve(self, _surface: Surface) -> C {
730        self.0
731    }
732}
733
734/// The step a mount site's chain has named so far, before it has named any.
735///
736/// It is not a [`NamedStep`], so a `.transform(..)` here fails on that bound rather than on a
737/// missing method, and the call site reads the guidance.
738#[doc(hidden)]
739#[derive(Debug, Clone, Copy, Default)]
740pub struct NoOutBound;
741
742/// A position a chain's steps can ride: what the chain named right before them.
743///
744/// The step traits below report it as an associated type ([`TransformLast::Step`] and its
745/// siblings) rather than refusing to apply, so the step's own bound is what fails and this note
746/// is what the call site reads.
747#[doc(hidden)]
748#[diagnostic::on_unimplemented(
749    message = "this step has no publish position to apply to",
750    label = "no `.out(marker, policy)` call precedes it in this chain",
751    note = "a step rides the position named right before it: `.out(Reply, Policy).transform(..)` \
752            for a reply, `.out(Marker, Policy).transform(..)` for a slot"
753)]
754pub trait NamedStep {}
755
756impl<const POS: usize> NamedStep for SlotPos<POS> {}
757
758/// The position a step applies to when the chain last named the reply rather than a slot.
759///
760/// A mount chain's steps read as "on the position before them": after `.out(Reply, ..)` they grow
761/// the reply's wiring, after `.out(marker, ..)` that slot's own.
762#[doc(hidden)]
763#[derive(Debug, Clone, Copy, Default)]
764pub struct ReplyLast;
765
766impl NamedStep for ReplyLast {}
767
768/// One `.out(marker, policy)` call, keyed by the marker: what the position it names becomes.
769///
770/// The dispatch is on the marker rather than on the attachment so that the call site's own
771/// argument settles which position is meant, and the error names that marker when the position is
772/// not there to bind. Every marker grows the attachment through [`BindAt`]: [`Reply`], the
773/// handler's own slot markers, and [`Retry`](crate::runtime::Retry), which is the slot the
774/// runtime publishes a deferred `retry_after` copy through.
775///
776/// Machinery; never named directly.
777#[doc(hidden)]
778#[diagnostic::on_unimplemented(
779    message = "this registration has no unbound publish position marked `{Self}`",
780    label = "`.out({Self}, ..)` has no position to bind here",
781    note = "`.out(marker, policy)` binds one position: `Reply` for the value a \
782            `publish(\"dest\")` handler returns, an `Out` slot's own marker for a slot, `Retry` \
783            for the deferred `retry_after` copy. Check the marker, that the handler declares it, \
784            and that it was not bound twice"
785)]
786pub trait OutPosition<Mount, B, Attach, Policy, Index> {
787    /// The grown attachment the chain carries after the call.
788    type Out;
789
790    /// Binds the position.
791    fn bind(attach: Attach, policy: Policy) -> Self::Out;
792}
793
794// The reply and the slots are positions of the attachment itself, so both delegate. The broker
795// takes no part: a policy for one of these positions pairs where the registration mounts.
796impl<Mount, B, Attach, Policy> OutPosition<Mount, B, Attach, Policy, ReplyLast> for Reply
797where
798    Attach: BindAt<Mount, Self, Policy, ReplyLast>,
799{
800    type Out = <Attach as BindAt<Mount, Self, Policy, ReplyLast>>::Out;
801
802    fn bind(attach: Attach, policy: Policy) -> Self::Out {
803        attach.bind_at(policy)
804    }
805}
806
807impl<Mount, B, M, Attach, Policy, Index> OutPosition<Mount, B, Attach, Policy, Index> for M
808where
809    M: OutSlot,
810    Attach: BindAt<Mount, M, Policy, Index>,
811{
812    type Out = <Attach as BindAt<Mount, M, Policy, Index>>::Out;
813
814    fn bind(attach: Attach, policy: Policy) -> Self::Out {
815        attach.bind_at(policy)
816    }
817}
818
819/// Binds one `.out(marker, policy)` call into a mount chain's attachment: the reply position for
820/// [`Reply`], one [`Out`](super::Out) slot for a slot marker, the deferred-retry slot for
821/// [`Retry`](crate::runtime::Retry).
822///
823/// `Index` is inferred per call - [`ReplyLast`] for the reply, [`SlotPos`] for a slot,
824/// [`RetryPos`](crate::runtime::RetryPos) for the deferred retry - which is what makes the calls
825/// order-independent and what the steps after the call ride. Machinery; never named directly.
826#[doc(hidden)]
827#[diagnostic::on_unimplemented(
828    message = "this registration has no unbound publish position marked `{M}`",
829    label = "`.out({M}, ..)` has no position to bind here",
830    note = "`.out(marker, policy)` binds one position: `Reply` for the value a \
831            `publish(\"dest\")` handler returns, an `Out` slot's own marker for a slot, `Retry` \
832            for the deferred `retry_after` copy. Check the marker, that the handler declares it, \
833            and that it was not bound twice"
834)]
835pub trait BindAt<Mount, M, Policy, Index> {
836    /// The attachment with that position bound.
837    type Out;
838
839    /// Binds it.
840    fn bind_at(self, policy: Policy) -> Self::Out;
841}
842
843/// A reply position still open: what `.out(Reply, policy)` binds.
844///
845/// The bind step states this about the position rather than about the whole attachment, so a
846/// second `.out(Reply, ..)` - and an `.out(Reply, ..)` on a handler that declares no reply -
847/// fails on this bound and the call site reads the note instead of a bare "no method".
848#[doc(hidden)]
849#[diagnostic::on_unimplemented(
850    message = "this registration has no open reply position",
851    label = "`.out(Reply, ..)` has nothing to bind here",
852    note = "`.out(Reply, policy)` names the publish policy of a `publish(\"dest\")` handler's \
853            returned value, once: a handler that declares no reply has no `Reply` position at \
854            all, and one whose policy is already named has none left open"
855)]
856pub trait ReplyOpen {}
857
858impl ReplyOpen for DefaultReply {}
859
860// The reply position, open exactly while it still carries the broker's default: the impl matches
861// on any reply attachment so that a second `.out(Reply, ..)` fails on `ReplyOpen` and reads its
862// note, rather than finding no impl at all.
863impl<Mount, Policy, Rep, Slots> BindAt<Mount, Reply, Policy, ReplyLast> for (Rep, Slots)
864where
865    Rep: ReplyOpen,
866    Mount: ReplyAttachment<Policy>,
867{
868    type Out = (WithSource<Mount::Wiring>, Slots);
869
870    fn bind_at(self, policy: Policy) -> Self::Out {
871        (WithSource::new(Mount::wire(policy)), self.1)
872    }
873}
874
875// One slot position, placed by marker through the positional machinery.
876impl<Mount, M, Policy, const POS: usize, Rep, Slots> BindAt<Mount, M, Policy, SlotPos<POS>>
877    for (Rep, Slots)
878where
879    M: OutSlot,
880    Slots: BindSlot<M, OutAttachment<M, Policy>, SlotPos<POS>>,
881{
882    type Out = (Rep, Slots::Out);
883
884    fn bind_at(self, policy: Policy) -> Self::Out {
885        (self.0, self.1.bind(OutAttachment::<M, Policy>::new(policy)))
886    }
887}
888
889/// Grows the transform stack of the slot bound at `Index`: the `.transform(..)` step of a mount
890/// chain, applied to a slot. Machinery; never named directly.
891///
892/// The index is the position the preceding `.out(marker, policy)` bound, carried by the builder,
893/// so the step reads as "on the slot just named" rather than repeating the marker.
894#[doc(hidden)]
895pub trait TransformAt<N, Index> {
896    /// The attachment tuple with that slot's stack grown.
897    type Out;
898
899    /// Composes it.
900    fn transform_at(self, transform: N) -> Self::Out;
901}
902
903/// Fills the codec position of the slot bound at `Index`: the `.codec(..)` step applied to a
904/// slot. Machinery; never named directly.
905#[doc(hidden)]
906pub trait CodecAt<Cd, Index> {
907    /// The attachment tuple with that slot's codec named.
908    type Out;
909
910    /// Names it.
911    fn codec_at(self, codec: Cd) -> Self::Out;
912}
913
914/// Replaces the publish policy of the slot bound at `Index`: the `.map_publisher(..)` step
915/// applied to a slot. Machinery; never named directly.
916#[doc(hidden)]
917pub trait MapPolicyAt<Index>: Sized {
918    /// The policy the slot carries.
919    type Policy;
920
921    /// Replaces it.
922    #[must_use]
923    fn map_policy_at(self, f: impl FnOnce(Self::Policy) -> Self::Policy) -> Self;
924}
925
926/// Composes a transform onto whatever a mount chain named last: the reply's wiring
927/// ([`ReplyLast`]) or one bound slot ([`SlotPos`]). Machinery; never named directly.
928#[doc(hidden)]
929pub trait TransformLast<N, Last> {
930    /// The position the transform rides, or [`NoOutBound`] when the chain has named none.
931    type Step;
932
933    /// The attachment after the step.
934    type Out;
935
936    /// Composes it.
937    fn transform_last(self, transform: N) -> Self::Out;
938}
939
940impl<N, W, Slots> TransformLast<N, ReplyLast> for (WithSource<W>, Slots)
941where
942    W: AddReplyTransform<N>,
943{
944    type Step = ReplyLast;
945    type Out = (WithSource<W::Out>, Slots);
946
947    fn transform_last(self, transform: N) -> Self::Out {
948        let (reply, slots) = self;
949        (reply.map(|wiring| wiring.add_transform(transform)), slots)
950    }
951}
952
953impl<N, const POS: usize, Rep, Slots> TransformLast<N, SlotPos<POS>> for (Rep, Slots)
954where
955    Slots: TransformAt<N, SlotPos<POS>>,
956{
957    type Step = SlotPos<POS>;
958    type Out = (Rep, Slots::Out);
959
960    fn transform_last(self, transform: N) -> Self::Out {
961        let (reply, slots) = self;
962        (reply, slots.transform_at(transform))
963    }
964}
965
966// The "nothing named yet" arm: it exists so the step resolves as a method and fails on
967// `Step: NamedStep`, which is where the guidance lives. It never runs.
968impl<N, Rep, Slots> TransformLast<N, NoOutBound> for (Rep, Slots) {
969    type Step = NoOutBound;
970    type Out = Self;
971
972    fn transform_last(self, _transform: N) -> Self {
973        self
974    }
975}
976
977/// Whether the position a chain named last admits the transform the `.transform(..)` step is
978/// about to compose, given what its stack has already declared.
979///
980/// This is where the six facts about naming a destination live. The position states what it
981/// offers - a reply type that leaves its destination open offers [`Names`], one that fixes it
982/// offers [`Reads`], a batch's replies offer [`Reads`] because they answer many deliveries, a slot
983/// offers what its `#[publishes(..)]` dictionary adds up to - and [`Admits`] settles it against
984/// what the transform declares and what the stack has taken. Machinery; never named directly.
985#[doc(hidden)]
986pub trait AdmitsAt<N, Last, Mount, Def, B> {}
987
988/// The broker's per-message options one position's live publisher takes, which is what a
989/// transform mounted there writes. Machinery behind the `.transform(..)` step's own bound.
990#[doc(hidden)]
991pub type PositionOptions<Policy, B> =
992    <<Policy as PublishPolicy<Connected<B>>>::Live as Publisher>::Options;
993
994// The reply position: the mount token turns the definition's reply type into an offer, and the
995// wiring's own stack says whether the right has been taken already.
996impl<N, Policy, Enc, PL, BL, Tx, Slots, Mount, Def, B> AdmitsAt<N, ReplyLast, Mount, Def, B>
997    for (WithSource<ReplyWiring<Policy, Enc, PL, BL, Tx>>, Slots)
998where
999    B: Broker,
1000    Policy: PublishPolicy<Connected<B>, Live: Publisher>,
1001    Mount: MountOffer<Def>,
1002    Def: DeclaresReply,
1003    PL: PublishTransform<ForReply<Def::Context>, PositionOptions<Policy, B>>,
1004    N: PublishTransform<ForReply<Def::Context>, PositionOptions<Policy, B>>,
1005    <N as PublishTransform<ForReply<Def::Context>, PositionOptions<Policy, B>>>::Destination:
1006        FitsOffer<
1007            Mount::Offer,
1008            <PL as PublishTransform<ForReply<Def::Context>, PositionOptions<Policy, B>>>::Destination,
1009            Def::Reply,
1010        >,
1011{
1012}
1013
1014// The byte-for-byte reply carries no transform stack at all, so `AddReplyTransform` is what has to
1015// fail there; this arm keeps this check from failing first and hiding it.
1016impl<N, Policy, Slots, Mount, Def, B> AdmitsAt<N, ReplyLast, Mount, Def, B>
1017    for (WithSource<RawReplyWiring<Policy>>, Slots)
1018{
1019}
1020
1021// One slot position, asked of the slot tuple through the positional machinery.
1022impl<N, const POS: usize, Rep, Slots, Mount, Def, B> AdmitsAt<N, SlotPos<POS>, Mount, Def, B>
1023    for (Rep, Slots)
1024where
1025    Slots: AdmitsSlotAt<N, SlotPos<POS>, B>,
1026{
1027}
1028
1029// See `TransformLast`'s own arm: `Step: NamedStep` is the bound that should fail with nothing
1030// named, so this one must not fail first.
1031impl<N, Rep, Slots, Mount, Def, B> AdmitsAt<N, NoOutBound, Mount, Def, B> for (Rep, Slots) {}
1032
1033/// Whether the slot bound at `Index` admits the transform the `.transform(..)` step is about to
1034/// compose. The positional half of [`AdmitsAt`]; machinery, never named directly.
1035#[doc(hidden)]
1036pub trait AdmitsSlotAt<N, Index, B> {}
1037
1038/// What one mount token offers a reply transform, read off the definition's reply type.
1039///
1040/// A one-by-one reply offers what its type's destination form offers; a batch's replies offer
1041/// [`Reads`] whatever their type says, because a batch answers many deliveries and carries none of
1042/// their headers, so there is nothing a transform could name a destination from. Machinery.
1043#[doc(hidden)]
1044pub trait MountOffer<Def> {
1045    /// The most a transform on this mount's reply may do to the destination.
1046    type Offer;
1047}
1048
1049/// What one reply type offers, folded out of its payload's destination declaration. Machinery
1050/// behind [`MountOffer`].
1051#[doc(hidden)]
1052pub trait ReplyOffer {
1053    /// The most a transform on this reply may do to the destination.
1054    type Offer;
1055}
1056
1057// The nested projection is the machinery of the resolution, not the user's mistake.
1058#[diagnostic::do_not_recommend]
1059impl<R> ReplyOffer for R
1060where
1061    R: ReplyShape<Body: OutgoingDestination<Form: DestinationOffer>>,
1062{
1063    type Offer = <<R::Body as OutgoingDestination>::Form as DestinationOffer>::Offer;
1064}
1065
1066/// Implements [`MountOffer`] for the mounts whose reply answers one delivery: the offer is the
1067/// reply type's own.
1068macro_rules! impl_solo_mount_offer {
1069    ($($mount:ident),+ $(,)?) => {$(
1070        impl<Def: DeclaresReply<Reply: ReplyOffer>> MountOffer<Def> for $mount {
1071            type Offer = <Def::Reply as ReplyOffer>::Offer;
1072        }
1073    )+};
1074}
1075
1076impl_solo_mount_offer!(PublishMount, PublishInjectMount);
1077
1078/// Implements [`MountOffer`] for the mounts that offer nothing: a batch answers many deliveries,
1079/// and a byte-for-byte reply has no transform stack to offer anything to.
1080macro_rules! impl_reads_only_mount_offer {
1081    ($($mount:ident),+ $(,)?) => {$(
1082        impl<Def> MountOffer<Def> for $mount {
1083            type Offer = Reads;
1084        }
1085    )+};
1086}
1087
1088impl_reads_only_mount_offer!(
1089    BatchPublishMount,
1090    BatchPublishInjectMount,
1091    RawReplyMount,
1092    RawReplyInjectMount,
1093);
1094
1095/// Names the codec of whatever a mount chain named last: the reply's encode codec after
1096/// `.out(Reply, ..)`, one slot's after `.out(marker, ..)`. Machinery; never named directly.
1097#[doc(hidden)]
1098pub trait CodecLast<Cd, Last> {
1099    /// See [`TransformLast::Step`].
1100    type Step;
1101
1102    /// The attachment after the step.
1103    type Out;
1104
1105    /// Names it.
1106    fn codec_last(self, codec: Cd) -> Self::Out;
1107}
1108
1109impl<Cd, W, Slots> CodecLast<Cd, ReplyLast> for (WithSource<W>, Slots)
1110where
1111    W: NameReplyCodec<Cd, Slot: CodecSlotOpen>,
1112{
1113    type Step = ReplyLast;
1114    type Out = (WithSource<W::Out>, Slots);
1115
1116    fn codec_last(self, codec: Cd) -> Self::Out {
1117        let (reply, slots) = self;
1118        (reply.map(|wiring| wiring.name_codec(codec)), slots)
1119    }
1120}
1121
1122impl<Cd, const POS: usize, Rep, Slots> CodecLast<Cd, SlotPos<POS>> for (Rep, Slots)
1123where
1124    Slots: CodecAt<Cd, SlotPos<POS>>,
1125{
1126    type Step = SlotPos<POS>;
1127    type Out = (Rep, Slots::Out);
1128
1129    fn codec_last(self, codec: Cd) -> Self::Out {
1130        let (reply, slots) = self;
1131        (reply, slots.codec_at(codec))
1132    }
1133}
1134
1135// See `TransformLast`'s own arm.
1136impl<Cd, Rep, Slots> CodecLast<Cd, NoOutBound> for (Rep, Slots) {
1137    type Step = NoOutBound;
1138    type Out = Self;
1139
1140    fn codec_last(self, _codec: Cd) -> Self {
1141        self
1142    }
1143}
1144
1145/// A position the batch-only steps can ride: the reply, and nothing else.
1146///
1147/// [`BatchTransformLast`] and [`TransactionalLast`] report the position as an associated type and
1148/// state this about it, so the step fails on this bound and the call site reads the note rather
1149/// than a bare "no method".
1150#[doc(hidden)]
1151#[diagnostic::on_unimplemented(
1152    message = "`.batch_transform(..)` and `.transactional()` ride the reply position",
1153    label = "the position this chain named before it is not `Reply`",
1154    note = "name the reply's publish policy first (`.out(Reply, policy)`); a slot publish is one \
1155            message with no batch, so a slot takes `.transform(..)` instead and a body opens its \
1156            own slot transaction with `entry.begin()`"
1157)]
1158pub trait ReplyStep {}
1159
1160impl ReplyStep for ReplyLast {}
1161
1162/// Composes a batch transform onto the reply a mount chain named last. Reply-only: a slot
1163/// publish is one message, so there is no batch for a batch transform to run over. Machinery;
1164/// never named directly.
1165#[doc(hidden)]
1166pub trait BatchTransformLast<N, Last> {
1167    /// See [`TransformLast::Step`].
1168    type Step;
1169
1170    /// The attachment after the step.
1171    type Out;
1172
1173    /// Composes it.
1174    fn batch_transform_last(self, transform: N) -> Self::Out;
1175}
1176
1177impl<N, W, Slots> BatchTransformLast<N, ReplyLast> for (WithSource<W>, Slots)
1178where
1179    W: AddBatchReplyTransform<N>,
1180{
1181    type Step = ReplyLast;
1182    type Out = (WithSource<W::Out>, Slots);
1183
1184    fn batch_transform_last(self, transform: N) -> Self::Out {
1185        let (reply, slots) = self;
1186        (
1187            reply.map(|wiring| wiring.add_batch_transform(transform)),
1188            slots,
1189        )
1190    }
1191}
1192
1193// The slot and the "nothing named yet" arms: present so the step resolves as a method and fails
1194// on `Step: ReplyStep`, which is where the guidance lives. Neither ever runs.
1195impl<N, const POS: usize, Rep, Slots> BatchTransformLast<N, SlotPos<POS>> for (Rep, Slots) {
1196    type Step = SlotPos<POS>;
1197    type Out = Self;
1198
1199    fn batch_transform_last(self, _transform: N) -> Self {
1200        self
1201    }
1202}
1203
1204impl<N, Rep, Slots> BatchTransformLast<N, NoOutBound> for (Rep, Slots) {
1205    type Step = NoOutBound;
1206    type Out = Self;
1207
1208    fn batch_transform_last(self, _transform: N) -> Self {
1209        self
1210    }
1211}
1212
1213/// Marks the reply a mount chain named last as publishing inside one broker transaction.
1214/// Reply-only, for the same reason [`BatchTransformLast`] is. Machinery; never named directly.
1215#[doc(hidden)]
1216pub trait TransactionalLast<Last> {
1217    /// See [`TransformLast::Step`].
1218    type Step;
1219
1220    /// The attachment after the step.
1221    type Out;
1222
1223    /// Marks it.
1224    fn transactional_last(self) -> Self::Out;
1225}
1226
1227impl<W, Slots> TransactionalLast<ReplyLast> for (WithSource<W>, Slots)
1228where
1229    W: TransactionalReply<State: PublishingDirectly>,
1230{
1231    type Step = ReplyLast;
1232    type Out = (WithSource<W::Out>, Slots);
1233
1234    fn transactional_last(self) -> Self::Out {
1235        let (reply, slots) = self;
1236        (reply.map(TransactionalReply::into_transactional), slots)
1237    }
1238}
1239
1240// See `BatchTransformLast`'s own arms: present so the step fails on `Step: ReplyStep`.
1241impl<const POS: usize, Rep, Slots> TransactionalLast<SlotPos<POS>> for (Rep, Slots) {
1242    type Step = SlotPos<POS>;
1243    type Out = Self;
1244
1245    fn transactional_last(self) -> Self {
1246        self
1247    }
1248}
1249
1250impl<Rep, Slots> TransactionalLast<NoOutBound> for (Rep, Slots) {
1251    type Step = NoOutBound;
1252    type Out = Self;
1253
1254    fn transactional_last(self) -> Self {
1255        self
1256    }
1257}
1258
1259/// Replaces the publish policy of whatever a mount chain named last: the hook a broker crate
1260/// layers its own publisher settings on. Machinery; never named directly.
1261#[doc(hidden)]
1262pub trait MapPolicyLast<Last>: Sized {
1263    /// See [`TransformLast::Step`].
1264    type Step;
1265
1266    /// The policy that position carries.
1267    type Policy;
1268
1269    /// Replaces it.
1270    #[must_use]
1271    fn map_policy_last(self, f: impl FnOnce(Self::Policy) -> Self::Policy) -> Self;
1272}
1273
1274impl<W: MapReplyPolicy, Slots> MapPolicyLast<ReplyLast> for (WithSource<W>, Slots) {
1275    type Step = ReplyLast;
1276    type Policy = W::Policy;
1277
1278    fn map_policy_last(self, f: impl FnOnce(W::Policy) -> W::Policy) -> Self {
1279        let (reply, slots) = self;
1280        (reply.map(|wiring| wiring.map_policy(f)), slots)
1281    }
1282}
1283
1284impl<const POS: usize, Rep, Slots: MapPolicyAt<SlotPos<POS>>> MapPolicyLast<SlotPos<POS>>
1285    for (Rep, Slots)
1286{
1287    type Step = SlotPos<POS>;
1288    type Policy = Slots::Policy;
1289
1290    fn map_policy_last(self, f: impl FnOnce(Slots::Policy) -> Slots::Policy) -> Self {
1291        let (reply, slots) = self;
1292        (reply, slots.map_policy_at(f))
1293    }
1294}
1295
1296// See `TransformLast`'s own arm.
1297impl<Rep, Slots> MapPolicyLast<NoOutBound> for (Rep, Slots) {
1298    type Step = NoOutBound;
1299    type Policy = ();
1300
1301    fn map_policy_last(self, _f: impl FnOnce(())) -> Self {
1302        self
1303    }
1304}
1305
1306/// A user-attached publisher source, wrapped so the bound and unbound attachment states live on
1307/// different type constructors (disjoint impls, no negative reasoning needed).
1308#[doc(hidden)]
1309#[derive(Debug)]
1310pub struct WithSource<Source>(Source);
1311
1312impl<Source> WithSource<Source> {
1313    pub(crate) fn new(source: Source) -> Self {
1314        Self(source)
1315    }
1316
1317    /// Borrows the wrapped source, so a mount can read what it declares before consuming it.
1318    pub(crate) const fn source(&self) -> &Source {
1319        &self.0
1320    }
1321
1322    /// Grows the wrapped source in place: how a mount site's reply chain adds one step to the
1323    /// wiring it already attached.
1324    pub(crate) fn map<NewSource>(
1325        self,
1326        f: impl FnOnce(Source) -> NewSource,
1327    ) -> WithSource<NewSource> {
1328        WithSource(f(self.0))
1329    }
1330}
1331
1332impl<Source> IntoSlotSource for WithSource<Source> {
1333    type Source = Source;
1334
1335    fn into_source(self) -> Self::Source {
1336        self.0
1337    }
1338}
1339
1340/// A definition whose handler carries marker-identified `Out` slots.
1341///
1342/// Implemented by `#[subscriber]` on the value you pass to `include`; [`Markers`](Self::Markers)
1343/// lists the slot markers in signature order, which is the order the positional machinery
1344/// ([`InitSlots`], [`BindSlot`], [`BindSlots`]) works in. You never implement this by hand
1345/// unless you hand-write a definition.
1346pub trait HasSlots {
1347    /// The slot markers, as a tuple in the handler signature's parameter order.
1348    type Markers;
1349}
1350
1351/// Builds the initial all-unbound attachment tuple for a definition's markers. Machinery behind
1352/// `include`; never named directly.
1353#[doc(hidden)]
1354pub trait InitSlots {
1355    /// One [`MissingSlot`] per marker, carrying it.
1356    type Init;
1357
1358    fn init() -> Self::Init;
1359}
1360
1361/// Unwraps one include-site attachment into the policy the runtime pairs: a bound
1362/// [`WithSource`] yields its policy. Machinery; never named directly.
1363#[doc(hidden)]
1364pub trait IntoSlotSource {
1365    /// The publish policy this attachment resolves to.
1366    type Source;
1367
1368    fn into_source(self) -> Self::Source;
1369}
1370
1371/// Positional placement of one `.out(marker, policy)` attachment: finds the (still unbound)
1372/// element carrying the marker and replaces it. The `Index` parameter is inferred per call,
1373/// which is what makes the calls order-independent; binding the same slot twice finds no
1374/// still-unbound element and fails to compile. Machinery; never named directly.
1375#[doc(hidden)]
1376#[diagnostic::on_unimplemented(
1377    message = "this handler has no unbound Out slot marked `{M}`",
1378    note = "each .out(marker, policy) call binds one still-unbound slot the handler declares; \
1379            check the marker, and that the slot was not bound twice"
1380)]
1381pub trait BindSlot<M, Src, Index> {
1382    /// The attachment tuple with the marker's position bound.
1383    type Out;
1384
1385    fn bind(self, src: Src) -> Self::Out;
1386}
1387
1388/// The position tokens inferred by [`BindSlot`].
1389#[doc(hidden)]
1390#[derive(Debug, Clone, Copy)]
1391pub struct SlotPos<const N: usize>;
1392
1393/// Instantiates a slot-carrying definition from the bound sources.
1394///
1395/// Implemented by `#[subscriber]` next to [`HasSlots`]: given the source tuple (in marker
1396/// order), it names the fully-applied definition type - each slot's publisher is
1397/// [`SlotPublisher`] over the source's [`PublishPolicy::Live`](crate::PublishPolicy::Live) - and pads the sources into the
1398/// definition's injection-extra tuple. The connected
1399/// broker type `C` is the pairing target the publisher types are computed against.
1400pub trait BindSlots<C: ConnectedBroker, Sources>: Sized {
1401    /// The publisher-applied definition type.
1402    type Bound;
1403
1404    /// The extra tuple the startup resolution (`FromStartup`) resolves the injections against:
1405    /// the sources, padded to the injection tuple's arity.
1406    type Extra;
1407
1408    /// Instantiates the definition and arranges the sources for resolution.
1409    fn bind(self, sources: Sources) -> (Self::Bound, Self::Extra);
1410}
1411
1412/// Implements [`InitSlots`] for each marker-tuple arity: every position starts as its marker's
1413/// [`MissingSlot`].
1414macro_rules! impl_init_slots {
1415    ($(($($marker:ident),+))+) => {$(
1416        impl<$($marker),+> InitSlots for ($($marker,)+) {
1417            type Init = ($(MissingSlot<$marker>,)+);
1418
1419            fn init() -> Self::Init {
1420                ($(MissingSlot::<$marker>::new(),)+)
1421            }
1422        }
1423    )+};
1424}
1425
1426impl_init_slots! {
1427    (M0)
1428    (M0, M1)
1429    (M0, M1, M2)
1430}
1431
1432/// Implements [`BindSlot`] for each (arity, position) pair: the element carrying the marker
1433/// (`@ <position>`) becomes [`WithSource`], the surrounding elements pass through unchanged.
1434macro_rules! impl_bind_slot {
1435    ($(($($before:ident,)* @ $pos:literal $(, $after:ident)*))+) => {$(
1436        impl<M, Src $(, $before)* $(, $after)*> BindSlot<M, Src, SlotPos<$pos>>
1437            for ($($before,)* MissingSlot<M>, $($after,)*)
1438        {
1439            type Out = ($($before,)* WithSource<Src>, $($after,)*);
1440
1441            fn bind(self, src: Src) -> Self::Out {
1442                #[allow(non_snake_case)]
1443                let ($($before,)* _missing, $($after,)*) = self;
1444                ($($before,)* WithSource::new(src), $($after,)*)
1445            }
1446        }
1447    )+};
1448}
1449
1450impl_bind_slot! {
1451    (@ 0)
1452    (@ 0, A1)
1453    (A0, @ 1)
1454    (@ 0, A1, A2)
1455    (A0, @ 1, A2)
1456    (A0, A1, @ 2)
1457}
1458
1459/// Implements [`TransformAt`] for each (arity, position) pair: the attachment at the position the
1460/// last `.out(..)` bound (`@ <position>`) grows its stack, the surrounding elements pass through.
1461macro_rules! impl_step_at {
1462    ($(($($before:ident,)* @ $pos:literal $(, $after:ident)*))+) => {$(
1463        impl<N, M, Policy, Layers, Enc $(, $before)* $(, $after)*> TransformAt<N, SlotPos<$pos>>
1464            for ($($before,)* WithSource<OutAttachment<M, Policy, Layers, Enc>>, $($after,)*)
1465        {
1466            type Out = (
1467                $($before,)*
1468                WithSource<OutAttachment<M, Policy, PublishTransformStack<Layers, N>, Enc>>,
1469                $($after,)*
1470            );
1471
1472            fn transform_at(self, transform: N) -> Self::Out {
1473                #[allow(non_snake_case)]
1474                let ($($before,)* bound, $($after,)*) = self;
1475                ($($before,)* bound.map(|slot| slot.add_transform(transform)), $($after,)*)
1476            }
1477        }
1478
1479        impl<N, M, Policy, Layers, Enc, B $(, $before)* $(, $after)*>
1480            AdmitsSlotAt<N, SlotPos<$pos>, B>
1481            for ($($before,)* WithSource<OutAttachment<M, Policy, Layers, Enc>>, $($after,)*)
1482        where
1483            B: Broker,
1484            M: OutSlot,
1485            Policy: PublishPolicy<Connected<B>, Live: Publisher>,
1486            Layers: PublishTransform<ForSlot, PositionOptions<Policy, B>>,
1487            N: PublishTransform<ForSlot, PositionOptions<Policy, B>>,
1488            <N as PublishTransform<ForSlot, PositionOptions<Policy, B>>>::Destination: FitsOffer<
1489                    M::Destination,
1490                    <Layers as PublishTransform<ForSlot, PositionOptions<Policy, B>>>::Destination,
1491                    M,
1492                >,
1493        {
1494        }
1495
1496        impl<Cd, M, Policy, Layers $(, $before)* $(, $after)*> CodecAt<Cd, SlotPos<$pos>>
1497            for (
1498                $($before,)*
1499                WithSource<OutAttachment<M, Policy, Layers, UnnamedCodec>>,
1500                $($after,)*
1501            )
1502        {
1503            type Out = (
1504                $($before,)*
1505                WithSource<OutAttachment<M, Policy, Layers, CallCodec<Cd>>>,
1506                $($after,)*
1507            );
1508
1509            fn codec_at(self, codec: Cd) -> Self::Out {
1510                #[allow(non_snake_case)]
1511                let ($($before,)* bound, $($after,)*) = self;
1512                ($($before,)* bound.map(|slot| slot.name_codec(codec)), $($after,)*)
1513            }
1514        }
1515
1516        impl<M, Policy, Layers, Enc $(, $before)* $(, $after)*> MapPolicyAt<SlotPos<$pos>>
1517            for ($($before,)* WithSource<OutAttachment<M, Policy, Layers, Enc>>, $($after,)*)
1518        {
1519            type Policy = Policy;
1520
1521            fn map_policy_at(self, f: impl FnOnce(Policy) -> Policy) -> Self {
1522                #[allow(non_snake_case)]
1523                let ($($before,)* bound, $($after,)*) = self;
1524                ($($before,)* bound.map(|slot| slot.map_policy(f)), $($after,)*)
1525            }
1526        }
1527    )+};
1528}
1529
1530impl_step_at! {
1531    (@ 0)
1532    (@ 0, A1)
1533    (A0, @ 1)
1534    (@ 0, A1, A2)
1535    (A0, @ 1, A2)
1536    (A0, A1, @ 2)
1537}
1538
1539#[cfg(test)]
1540mod tests;