pipecircuit 0.0.18

//! Physical progress for a pipe, independent of the host that activates it.
//!
//! A driver chooses the scope of each received message and supplies a sender
//! valid in that scope. This accommodates locally staged scatter/gather batches
//! without requiring the sender, the batch, or the implementation to be static.

use crate::{Inline, Pipe, ReceiveFromOutcome, StepsPipeInputToOutput, TransportTo};

/// Physical intake permission for the next bounded decoding or dequeue step.
/// This is not a semantic refusal and does not reserve storage by itself.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum IntakeAdmission {
    Ready,
    Paused,
}

/// A circuit checks its reserved storage before allowing a source to advance.
/// The context is physical (for example a connection identity), not a field
/// added to the semantic protocol. A single-threaded handler must preserve the
/// capacity it checked until it settles the admitted output. `Err` refuses the
/// intake as a failure, not a retryable pause; drivers preserve that failure
/// through their conduction error path. Physical work before intake may already
/// have progressed. Report refusal before advancing the gated dequeue or decoding step.
pub trait IntakeGate<Context, Failure = core::convert::Infallible> {
    fn admit(&mut self, context: Context) -> Result<IntakeAdmission, Failure>;
}

/// A scoped handler that can attach an explicit intake decision. The policy and
/// peer types supply read-only inspection to the gate without erasing borrows.
/// Gate-owned state can mutate through its `FnMut` capture; the gate cannot
/// replace peer bindings or submit work through their exclusive references.
pub trait AdmissionHandler<Context>: Sized {
    type Policy;
    type Peers;

    fn admission_context(&self) -> (&Self::Policy, &Self::Peers);
    type WithAdmission<Gate>: IntakeGate<Context>
    where
        Gate: FnMut(&Self::Policy, &Self::Peers, Context) -> IntakeAdmission;

    fn with_fallible_admission<Gate, Failure>(self, gate: Gate) -> FallibleAdmission<Self, Gate>
    where
        Gate: FnMut(&Self::Policy, &Self::Peers, Context) -> Result<IntakeAdmission, Failure>,
    {
        FallibleAdmission::new(self, gate)
    }

    fn with_admission<Gate>(self, gate: Gate) -> Self::WithAdmission<Gate>
    where
        Gate: FnMut(&Self::Policy, &Self::Peers, Context) -> IntakeAdmission;
}

/// A send that finishes borrowing its input before returning. A carrier may
/// retain owned storage for a partial write, but cannot retain this reference.
pub trait PipeSender<Protocol: Pipe> {
    type Outcome<'message>;

    #[must_use = "submission must be admitted, retained, or explicitly refused"]
    fn send<'message>(&mut self, input: Protocol::In<'message>) -> Self::Outcome<'message>;
}

/// Sending within one message scope. The exact carrier outcome remains visible.
///
/// `submit` may stage borrowed input until the receive turn ends. `send` also
/// remains available for a reply built locally inside the output handler.
pub trait ScopedSender<'message, Protocol: Pipe>: PipeSender<Protocol> {
    #[must_use = "submission must be admitted, retained, or explicitly refused"]
    fn submit(&mut self, input: Protocol::In<'message>) -> Self::Outcome<'message>;
}

/// Consumes an output before the driver's receive storage can be recycled.
///
/// `Failure` is independent of the message lifetime: a borrowed message cannot
/// escape the turn through either a successful return or an error. The handler
/// must settle any borrowed refusal while its exact ownership is still present.
pub trait PipeOutputHandler<Protocol: Pipe, Driver: PipeDriver<Protocol> + ?Sized> {
    type Failure;

    fn output<'message, Sender>(
        &mut self,
        output: Protocol::Out<'message>,
        outgoing: &mut Sender,
    ) -> Result<(), Self::Failure>
    where
        Sender: ScopedSender<'message, Protocol>
            + for<'input> PipeSender<Protocol, Outcome<'input> = Driver::Submission<'input>>;
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum DriveFailure<Physical, Conduct> {
    Physical(Physical),
    Conduct(Conduct),
    PhysicalAndConduct {
        physical: Physical,
        conduct: Conduct,
    },
}

impl<Physical, Conduct> DriveFailure<Physical, Conduct> {
    /// Attach physical source context without changing or dropping conduction failure.
    pub fn map_physical<Mapped>(
        self,
        map: impl FnOnce(Physical) -> Mapped,
    ) -> DriveFailure<Mapped, Conduct> {
        match self {
            Self::Physical(physical) => DriveFailure::Physical(map(physical)),
            Self::Conduct(conduct) => DriveFailure::Conduct(conduct),
            Self::PhysicalAndConduct { physical, conduct } => DriveFailure::PhysicalAndConduct {
                physical: map(physical),
                conduct,
            },
        }
    }
}

/// A concrete installed transport's activation and scoped conduction surface.
///
/// The host delivers physical activations; it does not decode messages. Drivers
/// may make physical progress without output or conduct several outputs in one
/// bounded turn. They never block waiting for another activation inside `drive`.
/// Registration and host selection remain separate from this portable contract.
pub trait PipeDriver<Protocol: Pipe> {
    type Activation;
    type Failure;
    type Submission<'message>;
    type Outgoing<'transport>: for<'message> ScopedSender<
            'message,
            Protocol,
            Outcome<'message> = Self::Submission<'message>,
        >
    where
        Self: 'transport;

    /// An outgoing view for inputs that do not borrow this driver's receive turn.
    fn outgoing(&mut self) -> Self::Outgoing<'_>;

    fn drive<Handler>(
        &mut self,
        activation: Self::Activation,
        handler: &mut Handler,
    ) -> ReceiveFromOutcome<(), Self::Activation, DriveFailure<Self::Failure, Handler::Failure>>
    where
        Handler: PipeOutputHandler<Protocol, Self>;
}

/// A driver with a documented admission boundary for bounded intake.
///
/// Admission controls only that boundary. Physical output progress and lifecycle
/// notifications may be independent; each implementation documents which outputs
/// require separate settlement capacity. A pause returns the activation so the
/// circuit can resume it on a capacity notification, without rerouting old work.
pub trait AdmittedPipeDriver<Protocol: Pipe>: PipeDriver<Protocol> {
    type IntakeContext;

    fn drive_admitted<Handler>(
        &mut self,
        activation: Self::Activation,
        handler: &mut Handler,
    ) -> ReceiveFromOutcome<(), Self::Activation, DriveFailure<Self::Failure, Handler::Failure>>
    where
        Handler:
            PipeOutputHandler<Protocol, Self> + IntakeGate<Self::IntakeContext, Handler::Failure>;
}

/// Inline steps have no independently driven source of activity.
pub enum NoActivation {}

impl<Protocol, Implementation> PipeSender<Protocol> for &mut Inline<Implementation, Protocol>
where
    Protocol: Pipe,
    Implementation: for<'input> StepsPipeInputToOutput<'input, Protocol>,
{
    type Outcome<'message> = Protocol::Out<'message>;

    fn send<'message>(&mut self, input: Protocol::In<'message>) -> Self::Outcome<'message> {
        TransportTo::send(*self, input)
    }
}

impl<'message, Protocol, Implementation> ScopedSender<'message, Protocol>
    for &mut Inline<Implementation, Protocol>
where
    Protocol: Pipe,
    Implementation: for<'input> StepsPipeInputToOutput<'input, Protocol>,
{
    fn submit(&mut self, input: Protocol::In<'message>) -> Self::Outcome<'message> {
        PipeSender::send(self, input)
    }
}

impl<Protocol, Implementation> PipeDriver<Protocol> for Inline<Implementation, Protocol>
where
    Protocol: Pipe,
    Implementation: for<'message> StepsPipeInputToOutput<'message, Protocol>,
{
    type Activation = NoActivation;
    type Failure = core::convert::Infallible;
    type Submission<'message> = Protocol::Out<'message>;
    type Outgoing<'transport>
        = &'transport mut Self
    where
        Self: 'transport;

    fn outgoing(&mut self) -> Self::Outgoing<'_> {
        self
    }

    fn drive<Handler>(
        &mut self,
        activation: NoActivation,
        _handler: &mut Handler,
    ) -> ReceiveFromOutcome<(), NoActivation, DriveFailure<Self::Failure, Handler::Failure>>
    where
        Handler: PipeOutputHandler<Protocol, Self>,
    {
        match activation {}
    }
}

mod admission;
pub use admission::{AdmissionFailure, FallibleAdmission};