pg_proto/
pipeline.rs

1//! Bounded, payload-free orchestration for proxy request pipelines.
2//!
3//! The ledger in this module records protocol obligations, not wire messages.
4//! Applications retain ownership of decoded messages until [`FrontendAction`] or
5//! [`BackendAction`] tells them to forward, emit, retry, or discard the value.
6//! [`Pipeline::accept_frontend`] and [`Pipeline::accept_backend`] are the canonical
7//! ledger interface. Their typed counterparts additionally dispatch accepted
8//! messages through phase-specific middleware before committing them.
9//! Callers receiving [`crate::demux::SessionItem`] values should first consume
10//! any pooling or attribution evidence they need, then convert the item with
11//! [`crate::demux::SessionItem::into_backend_message`] before backend acceptance.
12
13use std::{collections::VecDeque, convert::Infallible, sync::Arc};
14
15use tokio::sync::Notify;
16
17use crate::{
18    codec::{BackendMessage, FrontendMessage},
19    demux::Demux,
20    grammar::backend,
21    middleware::{
22        AsynchronousBackendMessage, ChainError, MessageMiddleware, Middleware,
23        ReconstructableMessage as _, Then,
24    },
25};
26
27/// Stable identity of one accepted frontend operation.
28///
29/// The identity is opaque: it permits application state to be correlated with
30/// ordered responses without exposing the intermediary's pipeline ledger or
31/// projection controls.
32#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
33pub struct OperationId(u64);
34
35/// Pipeline policy which preserves the historical lock-step behaviour.
36#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
37pub struct NoPipeline;
38
39/// Configuration for a bounded frontend operation pipeline.
40#[derive(Clone, Copy, Debug, Eq, PartialEq)]
41pub struct BoundedPipeline {
42    max_operations: usize,
43}
44
45impl BoundedPipeline {
46    /// Creates a pipeline with a non-zero operation-count limit.
47    ///
48    /// # Errors
49    ///
50    /// Returns an error when `max_operations` is zero.
51    pub fn new(max_operations: usize) -> Result<Self, PipelineConfigError> {
52        if max_operations == 0 {
53            return Err(PipelineConfigError);
54        }
55        Ok(Self { max_operations })
56    }
57
58    /// Returns the maximum number of incomplete operations.
59    #[must_use]
60    pub const fn max_operations(self) -> usize {
61        self.max_operations
62    }
63}
64
65/// A zero operation-count limit is not a usable pipeline configuration.
66#[derive(Clone, Copy, Debug, Eq, PartialEq)]
67pub struct PipelineConfigError;
68
69impl std::fmt::Display for PipelineConfigError {
70    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
71        formatter.write_str("pipeline operation limit must be non-zero")
72    }
73}
74
75impl std::error::Error for PipelineConfigError {}
76
77mod private {
78    pub trait Sealed {}
79}
80
81/// Sealed configuration accepted by the intermediary builder.
82pub trait PipelinePolicy: private::Sealed + Copy {
83    /// Maximum number of incomplete operation records.
84    fn operation_limit(self) -> usize;
85}
86
87impl private::Sealed for NoPipeline {}
88impl PipelinePolicy for NoPipeline {
89    fn operation_limit(self) -> usize {
90        1
91    }
92}
93
94impl private::Sealed for BoundedPipeline {}
95impl PipelinePolicy for BoundedPipeline {
96    fn operation_limit(self) -> usize {
97        self.max_operations
98    }
99}
100
101/// Whether an accepted frontend operation is locally handled or forwarded.
102#[derive(Clone, Copy, Debug, Eq, PartialEq)]
103pub(crate) enum FrontendHandling {
104    /// Send the returned message to the upstream connection.
105    Forward,
106    /// Do not send the request upstream; application code will synthesize its response.
107    Local,
108}
109
110/// Application action for one successfully projected frontend message.
111#[derive(Debug, Eq, PartialEq)]
112pub(crate) enum FrontendAction {
113    /// Forward the owned message upstream.
114    Forward {
115        /// Accepted operation identity.
116        id: OperationId,
117        /// Original, unretained frontend message.
118        message: FrontendMessage,
119    },
120    /// The operation is locally handled or belongs to a failed extended-query
121    /// cycle and the message must not be sent upstream.
122    Discard {
123        /// Accepted operation identity.
124        id: OperationId,
125    },
126}
127
128/// Position of a successfully accepted operation.
129#[derive(Debug, Eq, PartialEq)]
130pub(crate) enum FrontendAdmission {
131    /// Nothing earlier prevents this operation's response from being emitted.
132    Immediate(FrontendAction),
133    /// The operation was accepted but an earlier response must be emitted first.
134    Waiting(FrontendAction),
135}
136
137impl FrontendAdmission {
138    /// Returns the application action, discarding only the positional annotation.
139    #[must_use]
140    pub(crate) fn into_action(self) -> FrontendAction {
141        match self {
142            Self::Immediate(action) | Self::Waiting(action) => action,
143        }
144    }
145}
146
147/// Why a frontend message could not be accepted.
148#[derive(Debug, Eq, PartialEq)]
149pub enum FrontendProjectionError {
150    /// The bounded ledger is full; the unchanged message may be retried.
151    Capacity(Box<FrontendMessage>),
152    /// The message is not legal in the projected frontend protocol state.
153    Illegal {
154        /// Projected state at rejection.
155        state: PipelineState,
156        /// Unchanged illegal message.
157        message: Box<FrontendMessage>,
158    },
159}
160
161/// Application action for a backend message.
162#[derive(Debug, Eq, PartialEq)]
163pub(crate) enum BackendAction {
164    /// Emit this owned message to the downstream client now.
165    Emit(BackendMessage),
166    /// An earlier operation must complete; retry this unchanged message later.
167    Deferred(BackendMessage),
168}
169
170/// A backend message was not legal for any outstanding operation.
171#[derive(Debug, Eq, PartialEq)]
172pub struct BackendProjectionError {
173    /// Current response-side state.
174    pub state: PipelineState,
175    /// Unchanged illegal message.
176    pub message: BackendMessage,
177}
178
179#[derive(Debug, Eq, PartialEq)]
180pub(crate) enum BackendSequenceError {
181    Cardinality { expected: usize, actual: usize },
182    Source(BackendMessage),
183    Replacement(BackendMessage),
184    DifferentSpan,
185}
186
187/// Public summary of the pipeline's projected frontend state.
188#[derive(Clone, Copy, Debug, Eq, PartialEq)]
189pub enum PipelineState {
190    /// A simple or extended cycle may begin.
191    Ready,
192    /// Extended-query messages are being accepted.
193    Extended,
194    /// An extended error discards messages through `Sync`.
195    ExtendedError,
196    /// COPY IN accepts frontend data.
197    CopyIn,
198    /// COPY OUT accepts only backend data.
199    CopyOut,
200    /// COPY BOTH accepts data in both directions.
201    CopyBoth,
202    /// The connection has terminated.
203    Terminated,
204}
205
206/// Error returned while dispatching a pipeline message through typed middleware.
207#[derive(Debug)]
208pub(crate) enum PipelineMiddlewareError<MiddlewareError, ProjectionError> {
209    /// Middleware rejected the phase-typed message.
210    Middleware(MiddlewareError),
211    /// The pipeline rejected the original or rewritten message.
212    Projection(ProjectionError),
213}
214
215macro_rules! frontend_pipeline_phases {
216    ($consumer:ident) => {
217        $consumer! {
218            Ready => frontend_ready => backend::ReadyExternalMessage,
219            Building => frontend_building => backend::BuildingExternalMessage,
220            ExtendedError => frontend_extended_error => backend::ExtendedErrorExternalMessage,
221            SimpleCopyIn => frontend_simple_copy_in => backend::SimpleCopyInExternalMessage,
222            ExtendedCopyIn => frontend_extended_copy_in => backend::ExtendedCopyInExternalMessage,
223            SimpleCopyBoth => frontend_simple_copy_both => backend::SimpleCopyBothExternalMessage,
224            ExtendedCopyBoth => frontend_extended_copy_both => backend::ExtendedCopyBothExternalMessage,
225        }
226    };
227}
228
229macro_rules! backend_pipeline_phases {
230    ($consumer:ident) => {
231        $consumer! {
232            Asynchronous => backend_asynchronous => AsynchronousBackendMessage,
233            Simple => backend_simple => backend::SimpleInternalMessage,
234            SimpleError => backend_simple_error => backend::SimpleErrorInternalMessage,
235            ParseResponse => backend_parse_response => backend::ParseResponseInternalMessage,
236            BindResponse => backend_bind_response => backend::BindResponseInternalMessage,
237            DescribeResponse => backend_describe_response => backend::DescribeResponseInternalMessage,
238            ExecuteResponse => backend_execute_response => backend::ExecuteResponseInternalMessage,
239            CloseResponse => backend_close_response => backend::CloseResponseInternalMessage,
240            SyncResponse => backend_sync_response => backend::SyncResponseInternalMessage,
241            FunctionResponse => backend_function_response => backend::FunctionResponseInternalMessage,
242            FunctionReady => backend_function_ready => backend::FunctionReadyInternalMessage,
243            SimpleCopyInDone => backend_simple_copy_in_done => backend::SimpleCopyInDoneInternalMessage,
244            SimpleCopyInFailed => backend_simple_copy_in_failed => backend::SimpleCopyInFailedInternalMessage,
245            SimpleCopyOut => backend_simple_copy_out => backend::SimpleCopyOutInternalMessage,
246            SimpleCopyOutDone => backend_simple_copy_out_done => backend::SimpleCopyOutDoneInternalMessage,
247            SimpleCopyReady => backend_simple_copy_ready => backend::SimpleCopyReadyInternalMessage,
248            ExtendedCopyInDone => backend_extended_copy_in_done => backend::ExtendedCopyInDoneInternalMessage,
249            ExtendedCopyInFailed => backend_extended_copy_in_failed => backend::ExtendedCopyInFailedInternalMessage,
250            ExtendedCopyOut => backend_extended_copy_out => backend::ExtendedCopyOutInternalMessage,
251            ExtendedCopyOutDone => backend_extended_copy_out_done => backend::ExtendedCopyOutDoneInternalMessage,
252            SimpleCopyBoth => backend_simple_copy_both => backend::SimpleCopyBothInternalMessage,
253            SimpleCopyBothClientDone => backend_simple_copy_both_client_done => backend::SimpleCopyBothClientDoneInternalMessage,
254            SimpleCopyBothDone => backend_simple_copy_both_done => backend::SimpleCopyBothDoneInternalMessage,
255            SimpleCopyBothFailed => backend_simple_copy_both_failed => backend::SimpleCopyBothFailedInternalMessage,
256            ExtendedCopyBoth => backend_extended_copy_both => backend::ExtendedCopyBothInternalMessage,
257            ExtendedCopyBothClientDone => backend_extended_copy_both_client_done => backend::ExtendedCopyBothClientDoneInternalMessage,
258            ExtendedCopyBothDone => backend_extended_copy_both_done => backend::ExtendedCopyBothDoneInternalMessage,
259            ExtendedCopyBothFailed => backend_extended_copy_both_failed => backend::ExtendedCopyBothFailedInternalMessage,
260        }
261    };
262}
263
264macro_rules! declare_pipeline_hooks {
265    ($($phase:ident => $method:ident => $message:path),+ $(,)?) => {
266        $(
267            #[doc = concat!("Intercepts backend messages in generated `", stringify!($message), "` phase.")]
268            async fn $method(
269                &mut self,
270                _state: &mut State,
271                message: $message,
272            ) -> Result<$message, Self::Error> {
273                Ok(message)
274            }
275        )+
276    };
277}
278
279/// Async middleware for frontend messages selected from the runtime ledger phase.
280#[allow(async_fn_in_trait)]
281pub(crate) trait FrontendPipelineMiddleware<State> {
282    /// An error which prevents the message from continuing through the pipeline.
283    type Error;
284    frontend_pipeline_phases!(declare_pipeline_hooks);
285}
286
287/// Async middleware for backend messages selected from the runtime ledger phase.
288#[allow(async_fn_in_trait)]
289pub(crate) trait BackendPipelineMiddleware<State> {
290    /// An error which prevents the message from continuing through the pipeline.
291    type Error;
292    backend_pipeline_phases!(declare_pipeline_hooks);
293}
294
295impl<State> FrontendPipelineMiddleware<State> for crate::middleware::Identity {
296    type Error = Infallible;
297}
298
299impl<State> BackendPipelineMiddleware<State> for crate::middleware::Identity {
300    type Error = Infallible;
301}
302
303macro_rules! chained_pipeline_hooks {
304    ($($phase:ident => $method:ident => $message:ty),+ $(,)?) => {
305        $(
306            async fn $method(
307                &mut self,
308                state: &mut State,
309                message: $message,
310            ) -> Result<$message, Self::Error> {
311                let (first, second) = self.parts_mut();
312                let message = first
313                    .$method(state, message)
314                    .await
315                    .map_err(ChainError::First)?;
316                second
317                    .$method(state, message)
318                    .await
319                    .map_err(ChainError::Second)
320            }
321        )+
322    };
323}
324
325impl<State, First, Second> FrontendPipelineMiddleware<State> for Then<First, Second>
326where
327    First: FrontendPipelineMiddleware<State>,
328    Second: FrontendPipelineMiddleware<State>,
329{
330    type Error = ChainError<First::Error, Second::Error>;
331
332    frontend_pipeline_phases!(chained_pipeline_hooks);
333}
334
335impl<State, First, Second> BackendPipelineMiddleware<State> for Then<First, Second>
336where
337    First: BackendPipelineMiddleware<State>,
338    Second: BackendPipelineMiddleware<State>,
339{
340    type Error = ChainError<First::Error, Second::Error>;
341    backend_pipeline_phases!(chained_pipeline_hooks);
342}
343
344/// Adapts direction-wide async middleware to every typed pipeline hook.
345pub(crate) struct PipelineWireAdapter<Handler> {
346    handler: Handler,
347}
348
349impl<Handler> PipelineWireAdapter<Handler> {
350    /// Wraps direction-wide middleware for runtime phase dispatch.
351    pub(crate) const fn new(handler: Handler) -> Self {
352        Self { handler }
353    }
354
355    /// Returns the wrapped direction-wide middleware.
356    pub(crate) fn into_inner(self) -> Handler {
357        self.handler
358    }
359}
360
361/// Failure from direction-wide middleware adapted to typed pipeline dispatch.
362#[derive(Debug)]
363pub(crate) enum FrontendPipelineWireAdapterError<Error> {
364    /// The wrapped middleware rejected a message.
365    Middleware(Error),
366    /// The wrapped middleware returned a frontend message illegal in the selected phase.
367    IllegalFrontend(FrontendMessage),
368}
369
370/// Failure from backend wire middleware adapted to typed pipeline dispatch.
371#[derive(Debug)]
372pub(crate) enum BackendPipelineWireAdapterError<Error> {
373    /// The wrapped middleware rejected a message.
374    Middleware(Error),
375    /// The wrapped middleware returned a message illegal in the selected phase.
376    Illegal(BackendMessage),
377}
378
379macro_rules! pipeline_adapter_frontend_hooks {
380    ($($phase:ident => $method:ident => $message:ty),+ $(,)?) => {
381        $(
382            async fn $method(
383                &mut self,
384                state: &mut State,
385                message: $message,
386            ) -> Result<$message, Self::Error> {
387                let message: FrontendMessage = message.into();
388                let message = self
389                    .handler
390                    .intercept(state, message)
391                    .await
392                    .map_err(FrontendPipelineWireAdapterError::Middleware)?;
393                <$message>::try_from(message)
394                    .map_err(FrontendPipelineWireAdapterError::IllegalFrontend)
395            }
396        )+
397    };
398}
399
400macro_rules! pipeline_adapter_backend_hooks {
401    ($ignored:ident => $async_method:ident => AsynchronousBackendMessage, $($phase:ident => $method:ident => $message:ty),+ $(,)?) => {
402        async fn $async_method(
403            &mut self,
404            state: &mut State,
405            message: AsynchronousBackendMessage,
406        ) -> Result<AsynchronousBackendMessage, Self::Error> {
407            let message = self.handler.intercept(state, message.into_wire()).await
408                .map_err(BackendPipelineWireAdapterError::Middleware)?;
409            AsynchronousBackendMessage::try_from(message)
410                .map_err(BackendPipelineWireAdapterError::Illegal)
411        }
412        $(
413            async fn $method(
414                &mut self,
415                state: &mut State,
416                message: $message,
417            ) -> Result<$message, Self::Error> {
418                let message: BackendMessage = message.into();
419                let message = self
420                    .handler
421                    .intercept(state, message)
422                    .await
423                    .map_err(BackendPipelineWireAdapterError::Middleware)?;
424                <$message>::try_from(message).map_err(BackendPipelineWireAdapterError::Illegal)
425            }
426        )+
427    };
428}
429
430impl<State, Handler> FrontendPipelineMiddleware<State> for PipelineWireAdapter<Handler>
431where
432    Handler: MessageMiddleware<FrontendMessage, State>,
433{
434    type Error = FrontendPipelineWireAdapterError<Handler::Error>;
435    frontend_pipeline_phases!(pipeline_adapter_frontend_hooks);
436}
437
438impl<State, Handler> BackendPipelineMiddleware<State> for PipelineWireAdapter<Handler>
439where
440    Handler: MessageMiddleware<BackendMessage, State>,
441{
442    type Error = BackendPipelineWireAdapterError<Handler::Error>;
443    backend_pipeline_phases!(pipeline_adapter_backend_hooks);
444}
445
446#[derive(Clone, Copy, Debug, Eq, PartialEq)]
447enum RequestState {
448    Ready,
449    Extended { bound: bool },
450    ExtendedError,
451    CopyIn,
452    CopyOut,
453    CopyBoth,
454    Terminated,
455}
456
457impl RequestState {
458    const fn public(self) -> PipelineState {
459        match self {
460            Self::Ready => PipelineState::Ready,
461            Self::Extended { .. } => PipelineState::Extended,
462            Self::ExtendedError => PipelineState::ExtendedError,
463            Self::CopyIn => PipelineState::CopyIn,
464            Self::CopyOut => PipelineState::CopyOut,
465            Self::CopyBoth => PipelineState::CopyBoth,
466            Self::Terminated => PipelineState::Terminated,
467        }
468    }
469}
470
471#[derive(Clone, Copy, Debug, Eq, PartialEq)]
472enum Origin {
473    Forwarded,
474    Local,
475}
476
477#[derive(Clone, Copy, Debug, Eq, PartialEq)]
478enum OperationKind {
479    Query,
480    FunctionCall,
481    Parse,
482    Bind,
483    Describe,
484    Execute,
485    Close,
486    Flush,
487    Sync,
488    CopyData,
489    CopyDone,
490    CopyFail,
491    Terminate,
492}
493
494#[derive(Clone, Copy, Debug, Eq, PartialEq)]
495enum PreparedResponse {
496    Asynchronous,
497    Emit {
498        head: Operation,
499        response_state: backend::RuntimeState,
500    },
501    Deferred,
502    Illegal,
503}
504
505#[derive(Clone, Copy, Debug, Eq, PartialEq)]
506enum FrontendPhase {
507    Ready,
508    Building,
509    ExtendedError,
510    SimpleCopyIn,
511    ExtendedCopyIn,
512    SimpleCopyBoth,
513    ExtendedCopyBoth,
514}
515
516#[derive(Clone, Copy, Debug)]
517struct PreparedFrontend {
518    phase: FrontendPhase,
519    request_state: RequestState,
520}
521
522#[derive(Clone, Copy, Debug, Eq, PartialEq)]
523struct Operation {
524    id: OperationId,
525    kind: OperationKind,
526    origin: Origin,
527    discarded: bool,
528    response_state: backend::RuntimeState,
529}
530
531/// A bounded ledger coordinating independently owned frontend and backend values.
532#[derive(Debug)]
533pub(crate) struct Pipeline<P = NoPipeline> {
534    policy: P,
535    operations: VecDeque<Operation>,
536    request_state: RequestState,
537    response_state: Option<PipelineState>,
538    next_id: u64,
539    changed: Arc<Notify>,
540}
541
542impl Default for Pipeline<NoPipeline> {
543    fn default() -> Self {
544        Self::new(NoPipeline)
545    }
546}
547
548impl<P: PipelinePolicy> Pipeline<P> {
549    /// Returns the identity that will be assigned to the next accepted
550    /// frontend operation.
551    pub(crate) const fn next_operation_id(&self) -> OperationId {
552        OperationId(self.next_id)
553    }
554
555    /// Attributes a backend message to its forwarded frontend operation.
556    /// Asynchronous messages have no operation identity.
557    pub(crate) fn backend_operation_id(&self, message: &BackendMessage) -> Option<OperationId> {
558        if is_asynchronous(message) {
559            return None;
560        }
561        self.operations
562            .iter()
563            .find(|operation| {
564                operation.origin == Origin::Forwarded
565                    && !operation.discarded
566                    && response_fits(**operation, message)
567            })
568            .map(|operation| operation.id)
569    }
570
571    /// Attributes an ordered backend span while projecting each source against
572    /// a private ledger snapshot.
573    pub(crate) fn backend_operation_ids(
574        &self,
575        messages: &[BackendMessage],
576    ) -> Vec<Option<OperationId>> {
577        let mut projection = self.snapshot();
578        messages
579            .iter()
580            .map(|message| {
581                let id = projection.backend_operation_id(message);
582                let _ = projection.accept_backend(message.clone());
583                id
584            })
585            .collect()
586    }
587
588    fn snapshot(&self) -> Self {
589        Self {
590            policy: self.policy,
591            operations: self.operations.clone(),
592            request_state: self.request_state,
593            response_state: self.response_state,
594            next_id: self.next_id,
595            changed: Arc::clone(&self.changed),
596        }
597    }
598
599    pub(crate) fn prepare_backend_replacements(
600        &self,
601        sources: &[BackendMessage],
602        replacements: &[BackendMessage],
603    ) -> Result<Self, BackendSequenceError> {
604        if sources.len() != replacements.len() {
605            return Err(BackendSequenceError::Cardinality {
606                expected: sources.len(),
607                actual: replacements.len(),
608            });
609        }
610        let mut source_projection = self.snapshot();
611        for message in sources.iter().cloned() {
612            match source_projection.accept_backend(message) {
613                Ok(BackendAction::Emit(_)) => {}
614                Ok(BackendAction::Deferred(message))
615                | Err(BackendProjectionError { message, .. }) => {
616                    return Err(BackendSequenceError::Source(message));
617                }
618            }
619        }
620        let mut replacement_projection = self.snapshot();
621        for message in replacements.iter().cloned() {
622            if !message.is_reconstructable() {
623                return Err(BackendSequenceError::Replacement(message));
624            }
625            match replacement_projection.accept_backend(message) {
626                Ok(BackendAction::Emit(_)) => {}
627                Ok(BackendAction::Deferred(message))
628                | Err(BackendProjectionError { message, .. }) => {
629                    return Err(BackendSequenceError::Replacement(message));
630                }
631            }
632        }
633        if source_projection.operations != replacement_projection.operations
634            || source_projection.request_state != replacement_projection.request_state
635            || source_projection.response_state != replacement_projection.response_state
636        {
637            return Err(BackendSequenceError::DifferentSpan);
638        }
639        Ok(replacement_projection)
640    }
641    /// Creates an empty pipeline using `policy`.
642    #[must_use]
643    pub(crate) fn new(policy: P) -> Self {
644        Self {
645            policy,
646            operations: VecDeque::new(),
647            request_state: RequestState::Ready,
648            response_state: None,
649            next_id: 0,
650            changed: Arc::new(Notify::new()),
651        }
652    }
653
654    /// Returns the projected frontend protocol state.
655    #[must_use]
656    pub(crate) fn state(&self) -> PipelineState {
657        self.response_state
658            .unwrap_or_else(|| self.request_state.public())
659    }
660
661    /// Returns the number of incomplete lightweight operation records.
662    #[must_use]
663    pub(crate) fn len(&self) -> usize {
664        self.operations.len()
665    }
666
667    /// Reports whether no operations remain outstanding.
668    #[must_use]
669    pub(crate) fn is_empty(&self) -> bool {
670        self.operations.is_empty()
671    }
672
673    /// Projects and accepts one frontend message without retaining its payload.
674    ///
675    /// Capacity and legality failures return the original owned message. A
676    /// capacity failure does not mutate either projected state or the ledger.
677    ///
678    /// # Errors
679    ///
680    /// Returns the unchanged boxed message when capacity is exhausted or the
681    /// message is illegal in the projected state.
682    pub(crate) fn accept_frontend(
683        &mut self,
684        message: FrontendMessage,
685        handling: FrontendHandling,
686    ) -> Result<FrontendAdmission, FrontendProjectionError> {
687        let prepared = self.prepare_frontend(&message)?;
688        Ok(self.commit_frontend(prepared, message, handling))
689    }
690
691    fn prepare_frontend(
692        &self,
693        message: &FrontendMessage,
694    ) -> Result<PreparedFrontend, FrontendProjectionError> {
695        if self.operations.len() == self.policy.operation_limit()
696            && !matches!(
697                self.request_state,
698                RequestState::CopyIn | RequestState::CopyBoth
699            )
700        {
701            return Err(FrontendProjectionError::Capacity(Box::new(message.clone())));
702        }
703        if project_frontend(self.request_state, message).is_none() {
704            return Err(FrontendProjectionError::Illegal {
705                state: self.state(),
706                message: Box::new(message.clone()),
707            });
708        }
709        Ok(PreparedFrontend {
710            phase: frontend_phase(
711                self.request_state,
712                self.operations.front().map(|operation| operation.kind),
713            ),
714            request_state: self.request_state,
715        })
716    }
717
718    fn commit_frontend(
719        &mut self,
720        prepared: PreparedFrontend,
721        message: FrontendMessage,
722        handling: FrontendHandling,
723    ) -> FrontendAdmission {
724        let (kind, next_state) = classify_frontend(prepared.request_state, &message)
725            .expect("phase-typed frontend replacement has a ledger classification");
726        let waiting = !self.operations.is_empty();
727        let id = OperationId(self.next_id);
728        self.next_id = self.next_id.saturating_add(1);
729        self.request_state = next_state;
730        let origin = match handling {
731            FrontendHandling::Forward => Origin::Forwarded,
732            FrontendHandling::Local => Origin::Local,
733        };
734        if let Some(head) = self.operations.front_mut()
735            && let Some(event) = backend::project_external(head.response_state, &message)
736            && let Some(transition) = backend::transition(head.response_state, event)
737        {
738            head.response_state = transition.target;
739        }
740        let discarded = matches!(self.request_state, RequestState::ExtendedError)
741            && kind != OperationKind::Sync;
742        self.operations.push_back(Operation {
743            id,
744            kind,
745            origin,
746            discarded,
747            response_state: initial_response_state(kind),
748        });
749        let action = match (handling, discarded) {
750            (FrontendHandling::Forward, false) => FrontendAction::Forward { id, message },
751            (FrontendHandling::Forward | FrontendHandling::Local, _) => {
752                FrontendAction::Discard { id }
753            }
754        };
755        let admission = if waiting {
756            FrontendAdmission::Waiting(action)
757        } else {
758            FrontendAdmission::Immediate(action)
759        };
760        self.remove_inert_heads();
761        admission
762    }
763
764    /// Projects, asynchronously intercepts, and accepts one frontend message.
765    ///
766    /// The ledger selects the phase-specific middleware hook at runtime. The
767    /// selected hook can only return a message legal in that same phase.
768    /// Middleware is not invoked when capacity is exhausted.
769    ///
770    /// # Errors
771    ///
772    /// Returns a middleware error, an illegal original or replacement message,
773    /// or the unchanged message when capacity is exhausted.
774    pub(crate) async fn accept_frontend_typed<State, Handler>(
775        &mut self,
776        middleware: &mut Middleware<State, Handler>,
777        message: FrontendMessage,
778        handling: FrontendHandling,
779    ) -> Result<FrontendAdmission, PipelineMiddlewareError<Handler::Error, FrontendProjectionError>>
780    where
781        Handler: FrontendPipelineMiddleware<State>,
782    {
783        let prepared = self
784            .prepare_frontend(&message)
785            .map_err(PipelineMiddlewareError::Projection)?;
786
787        let message = self
788            .intercept_frontend(prepared.phase, middleware, message)
789            .await
790            .map_err(PipelineMiddlewareError::Middleware)?;
791        if !message.is_reconstructable() {
792            return Err(PipelineMiddlewareError::Projection(
793                FrontendProjectionError::Illegal {
794                    state: self.state(),
795                    message: Box::new(message),
796                },
797            ));
798        }
799        Ok(self.commit_frontend(prepared, message, handling))
800    }
801
802    /// Projects one upstream backend message and preserves response order.
803    ///
804    /// # Errors
805    ///
806    /// Returns an unchanged response which cannot belong to any outstanding operation.
807    pub(crate) fn accept_backend(
808        &mut self,
809        message: BackendMessage,
810    ) -> Result<BackendAction, BackendProjectionError> {
811        self.accept_response(None, message)
812    }
813
814    /// Intercepts an emittable backend response through its operation-typed hook.
815    ///
816    /// Responses belonging to a later operation are returned unchanged as
817    /// [`BackendAction::Deferred`] and are intercepted only when retried at the
818    /// response head. Asynchronous messages use their non-advancing hook.
819    ///
820    /// # Errors
821    ///
822    /// Returns a middleware error or an unchanged response which cannot belong
823    /// to any outstanding operation.
824    pub(crate) async fn accept_backend_typed<State, Handler>(
825        &mut self,
826        middleware: &mut Middleware<State, Handler>,
827        message: BackendMessage,
828    ) -> Result<BackendAction, PipelineMiddlewareError<Handler::Error, BackendProjectionError>>
829    where
830        Handler: BackendPipelineMiddleware<State>,
831    {
832        self.accept_response_typed(None, middleware, message).await
833    }
834
835    /// Attempts to register and emit a locally synthesized response.
836    ///
837    /// The message is returned as [`BackendAction::Deferred`] when `id` has not
838    /// reached the head. No backend payload is retained by the ledger.
839    ///
840    /// # Errors
841    ///
842    /// Returns the unchanged message when it is illegal for the named operation.
843    pub(crate) fn try_emit_local(
844        &mut self,
845        id: OperationId,
846        message: BackendMessage,
847    ) -> Result<BackendAction, BackendProjectionError> {
848        self.accept_response(Some(id), message)
849    }
850
851    /// Typed-middleware counterpart to [`Self::try_emit_local`].
852    ///
853    /// Deferred local responses are not intercepted until their operation reaches
854    /// the response head.
855    ///
856    /// # Errors
857    ///
858    /// Returns a middleware error or an illegal response for the named operation.
859    pub(crate) async fn try_emit_local_typed<State, Handler>(
860        &mut self,
861        middleware: &mut Middleware<State, Handler>,
862        id: OperationId,
863        message: BackendMessage,
864    ) -> Result<BackendAction, PipelineMiddlewareError<Handler::Error, BackendProjectionError>>
865    where
866        Handler: BackendPipelineMiddleware<State>,
867    {
868        self.accept_response_typed(Some(id), middleware, message)
869            .await
870    }
871
872    /// Waits until a local operation reaches the response head.
873    ///
874    /// Cancellation is safe: polling this future never reserves or removes a
875    /// ledger entry. The caller should then invoke [`Self::try_emit_local`].
876    pub(crate) async fn wait_until_emittable(&self, id: OperationId) {
877        loop {
878            let notified = self.changed.notified();
879            if self
880                .operations
881                .front()
882                .is_some_and(|operation| operation.id == id)
883            {
884                return;
885            }
886            notified.await;
887        }
888    }
889
890    async fn intercept_frontend<State, Handler>(
891        &self,
892        phase: FrontendPhase,
893        middleware: &mut Middleware<State, Handler>,
894        message: FrontendMessage,
895    ) -> Result<FrontendMessage, Handler::Error>
896    where
897        Handler: FrontendPipelineMiddleware<State>,
898    {
899        let (state, handler) = middleware.parts_mut();
900        macro_rules! dispatch {
901            ($message:expr, $type:path, $handler:ident, $state:ident, $method:ident) => {{
902                let Ok(typed) = <$type>::try_from($message) else {
903                    unreachable!("frontend message was prevalidated for pipeline phase")
904                };
905                $handler.$method($state, typed).await?.into()
906            }};
907        }
908
909        macro_rules! dispatch_catalogue {
910            ($($catalogue_phase:ident => $method:ident => $message_type:path),+ $(,)?) => {
911                match phase {
912                    $(
913                        FrontendPhase::$catalogue_phase =>
914                            dispatch!(message, $message_type, handler, state, $method),
915                    )+
916                }
917            };
918        }
919
920        Ok(frontend_pipeline_phases!(dispatch_catalogue))
921    }
922
923    #[allow(clippy::too_many_lines)]
924    async fn accept_response_typed<State, Handler>(
925        &mut self,
926        local_id: Option<OperationId>,
927        middleware: &mut Middleware<State, Handler>,
928        message: BackendMessage,
929    ) -> Result<BackendAction, PipelineMiddlewareError<Handler::Error, BackendProjectionError>>
930    where
931        Handler: BackendPipelineMiddleware<State>,
932    {
933        let prepared = self.prepare_response(local_id, &message);
934        if matches!(prepared, PreparedResponse::Deferred) {
935            return Ok(BackendAction::Deferred(message));
936        }
937        if matches!(prepared, PreparedResponse::Illegal) {
938            return Err(PipelineMiddlewareError::Projection(
939                BackendProjectionError {
940                    state: self.state(),
941                    message,
942                },
943            ));
944        }
945
946        let (state, handler) = middleware.parts_mut();
947        let message = match prepared {
948            PreparedResponse::Asynchronous => {
949                let Ok(typed) = AsynchronousBackendMessage::try_from(message) else {
950                    unreachable!("asynchronous response was prevalidated")
951                };
952                handler
953                    .backend_asynchronous(state, typed)
954                    .await
955                    .map_err(PipelineMiddlewareError::Middleware)?
956                    .into_wire()
957            }
958            PreparedResponse::Emit { response_state, .. } => {
959                macro_rules! dispatch {
960                    ($message:ty, $method:ident) => {{
961                        let typed = match <$message>::try_from(message) {
962                            Ok(typed) => typed,
963                            Err(message) => {
964                                return Err(PipelineMiddlewareError::Projection(
965                                    BackendProjectionError {
966                                        state: self.state(),
967                                        message,
968                                    },
969                                ));
970                            }
971                        };
972                        handler
973                            .$method(state, typed)
974                            .await
975                            .map_err(PipelineMiddlewareError::Middleware)?
976                            .into_wire()
977                    }};
978                }
979
980                macro_rules! dispatch_catalogue {
981                    ($ignored:ident => $ignored_method:ident => AsynchronousBackendMessage,
982                     $($catalogue_phase:ident => $method:ident => $message_type:path),+ $(,)?) => {
983                        match response_state {
984                            $(
985                                backend::RuntimeState::$catalogue_phase =>
986                                    dispatch!($message_type, $method),
987                            )+
988                            _ => unreachable!("response phase has no backend-selected transition"),
989                        }
990                    };
991                }
992
993                backend_pipeline_phases!(dispatch_catalogue)
994            }
995            PreparedResponse::Deferred | PreparedResponse::Illegal => unreachable!(),
996        };
997
998        if !message.is_reconstructable() {
999            return Err(PipelineMiddlewareError::Projection(
1000                BackendProjectionError {
1001                    state: self.state(),
1002                    message,
1003                },
1004            ));
1005        }
1006        self.commit_response(prepared, message)
1007            .map_err(PipelineMiddlewareError::Projection)
1008    }
1009
1010    fn prepare_response(
1011        &self,
1012        local_id: Option<OperationId>,
1013        message: &BackendMessage,
1014    ) -> PreparedResponse {
1015        if is_asynchronous(message) {
1016            return PreparedResponse::Asynchronous;
1017        }
1018        let Some(head) = self.operations.front().copied() else {
1019            return PreparedResponse::Illegal;
1020        };
1021        if let Some(id) = local_id {
1022            if head.id != id {
1023                return PreparedResponse::Deferred;
1024            }
1025            if head.origin != Origin::Local {
1026                return PreparedResponse::Illegal;
1027            }
1028        } else if head.origin == Origin::Local {
1029            return if self.operations.iter().skip(1).any(|operation| {
1030                operation.origin == Origin::Forwarded && response_fits(*operation, message)
1031            }) {
1032                PreparedResponse::Deferred
1033            } else {
1034                PreparedResponse::Illegal
1035            };
1036        }
1037        if head.discarded || !response_fits(head, message) {
1038            return if self
1039                .operations
1040                .iter()
1041                .skip(1)
1042                .any(|operation| response_fits(*operation, message))
1043            {
1044                PreparedResponse::Deferred
1045            } else {
1046                PreparedResponse::Illegal
1047            };
1048        }
1049        PreparedResponse::Emit {
1050            head,
1051            response_state: head.response_state,
1052        }
1053    }
1054
1055    fn accept_response(
1056        &mut self,
1057        local_id: Option<OperationId>,
1058        message: BackendMessage,
1059    ) -> Result<BackendAction, BackendProjectionError> {
1060        let prepared = self.prepare_response(local_id, &message);
1061        if prepared == PreparedResponse::Deferred
1062            && matches!(message, BackendMessage::ErrorResponse(_))
1063            && let Some(id) = local_id
1064        {
1065            self.project_deferred_local_error(id);
1066        }
1067        self.commit_response(prepared, message)
1068    }
1069
1070    fn project_deferred_local_error(&mut self, id: OperationId) {
1071        let Some(rejected) = self
1072            .operations
1073            .iter()
1074            .position(|operation| operation.id == id)
1075        else {
1076            return;
1077        };
1078        if self.operations[rejected].origin != Origin::Local
1079            || !is_extended_kind(self.operations[rejected].kind)
1080        {
1081            return;
1082        }
1083
1084        let mut sync_accepted = false;
1085        for operation in self.operations.iter_mut().skip(rejected + 1) {
1086            if operation.kind == OperationKind::Sync {
1087                sync_accepted = true;
1088                break;
1089            }
1090            operation.discarded = true;
1091        }
1092        if !sync_accepted {
1093            self.request_state = RequestState::ExtendedError;
1094        }
1095    }
1096
1097    fn commit_response(
1098        &mut self,
1099        prepared: PreparedResponse,
1100        message: BackendMessage,
1101    ) -> Result<BackendAction, BackendProjectionError> {
1102        let PreparedResponse::Emit {
1103            head,
1104            response_state,
1105        } = prepared
1106        else {
1107            return match prepared {
1108                PreparedResponse::Asynchronous => Ok(BackendAction::Emit(message)),
1109                PreparedResponse::Deferred => Ok(BackendAction::Deferred(message)),
1110                PreparedResponse::Illegal => Err(BackendProjectionError {
1111                    state: self.state(),
1112                    message,
1113                }),
1114                PreparedResponse::Emit { .. } => unreachable!(),
1115            };
1116        };
1117        let event = backend::project_internal(response_state, &message)
1118            .expect("response was validated against its generated backend phase");
1119        let next_response_state = backend::transition(response_state, event)
1120            .expect("projected backend event has a generated transition")
1121            .target;
1122        let terminal = response_is_terminal(head.kind, &message);
1123        let error = matches!(message, BackendMessage::ErrorResponse(_));
1124        let copy_state = response_copy_state(next_response_state);
1125        if terminal {
1126            self.operations.pop_front();
1127            if error && is_extended_kind(head.kind) {
1128                self.enter_extended_error();
1129            }
1130        } else if let Some(head) = self.operations.front_mut() {
1131            head.response_state = next_response_state;
1132        }
1133        if !terminal {
1134            self.response_state = copy_state.map(RequestState::public);
1135        }
1136        if let Some(state) = copy_state
1137            && matches!(state, RequestState::CopyIn | RequestState::CopyBoth)
1138        {
1139            self.request_state = state;
1140        }
1141        if terminal {
1142            self.response_state = None;
1143        }
1144        self.remove_inert_heads();
1145        self.changed.notify_waiters();
1146        Ok(BackendAction::Emit(message))
1147    }
1148
1149    fn enter_extended_error(&mut self) {
1150        self.response_state = None;
1151        let mut sync_accepted = false;
1152        for operation in &mut self.operations {
1153            if operation.kind == OperationKind::Sync {
1154                sync_accepted = true;
1155                break;
1156            }
1157            operation.discarded = true;
1158        }
1159        if !sync_accepted {
1160            self.request_state = RequestState::ExtendedError;
1161        }
1162    }
1163
1164    fn remove_inert_heads(&mut self) {
1165        let previous_len = self.operations.len();
1166        while self.operations.front().is_some_and(|operation| {
1167            operation.kind == OperationKind::Flush
1168                || operation.kind == OperationKind::CopyData
1169                || operation.kind == OperationKind::CopyDone
1170                || operation.kind == OperationKind::CopyFail
1171                || operation.kind == OperationKind::Terminate
1172                || operation.discarded
1173        }) {
1174            self.operations.pop_front();
1175        }
1176        if self.operations.len() != previous_len {
1177            self.changed.notify_waiters();
1178        }
1179    }
1180}
1181
1182fn project_frontend(
1183    state: RequestState,
1184    message: &FrontendMessage,
1185) -> Option<(OperationKind, RequestState)> {
1186    use RequestState as S;
1187    let generated_state = match state {
1188        S::Ready => backend::RuntimeState::Ready,
1189        S::Extended { .. } => backend::RuntimeState::Building,
1190        S::ExtendedError => backend::RuntimeState::ExtendedError,
1191        S::CopyIn => backend::RuntimeState::ExtendedCopyIn,
1192        S::CopyOut => backend::RuntimeState::ExtendedCopyOut,
1193        S::CopyBoth => backend::RuntimeState::ExtendedCopyBoth,
1194        S::Terminated => backend::RuntimeState::Terminated,
1195    };
1196    backend::project_external(generated_state, message)?;
1197    classify_frontend(state, message)
1198}
1199
1200fn classify_frontend(
1201    state: RequestState,
1202    message: &FrontendMessage,
1203) -> Option<(OperationKind, RequestState)> {
1204    use FrontendMessage as F;
1205    use OperationKind as O;
1206    use RequestState as S;
1207
1208    match (state, message) {
1209        (S::Ready, F::Query(_)) => Some((O::Query, S::Ready)),
1210        (S::Ready, F::FunctionCall(_)) => Some((O::FunctionCall, S::Ready)),
1211        (S::Ready, F::Parse(_)) => Some((O::Parse, S::Extended { bound: false })),
1212        (S::Ready | S::Extended { .. }, F::Bind(_)) => Some((O::Bind, S::Extended { bound: true })),
1213        (S::Ready, F::Describe(_)) => Some((O::Describe, S::Extended { bound: false })),
1214        (S::Ready | S::Extended { .. }, F::Execute(_)) => {
1215            Some((O::Execute, S::Extended { bound: true }))
1216        }
1217        (S::Ready, F::Close(_)) => Some((O::Close, S::Extended { bound: false })),
1218        (S::Ready, F::Terminate) => Some((O::Terminate, S::Terminated)),
1219        (S::Extended { bound }, F::Parse(_)) => Some((O::Parse, S::Extended { bound })),
1220        (S::Extended { bound }, F::Describe(_)) => Some((O::Describe, S::Extended { bound })),
1221        (S::Extended { bound }, F::Close(_)) => Some((O::Close, S::Extended { bound })),
1222        (S::Extended { bound }, F::Flush) => Some((O::Flush, S::Extended { bound })),
1223        (S::Extended { .. } | S::ExtendedError, F::Sync) => Some((O::Sync, S::Ready)),
1224        (S::ExtendedError, _) => Some((classify_discard(message)?, S::ExtendedError)),
1225        (S::CopyIn, F::CopyData(_)) => Some((O::CopyData, S::CopyIn)),
1226        (S::CopyIn, F::CopyDone) => Some((O::CopyDone, S::Extended { bound: true })),
1227        (S::CopyIn, F::CopyFail(_)) => Some((O::CopyFail, S::ExtendedError)),
1228        (S::CopyBoth, F::CopyData(_)) => Some((O::CopyData, S::CopyBoth)),
1229        (S::CopyBoth, F::CopyDone) => Some((O::CopyDone, S::CopyBoth)),
1230        _ => None,
1231    }
1232}
1233
1234fn frontend_phase(state: RequestState, response_head: Option<OperationKind>) -> FrontendPhase {
1235    match (state, response_head) {
1236        (RequestState::Ready, _) => FrontendPhase::Ready,
1237        (RequestState::Extended { .. }, _) => FrontendPhase::Building,
1238        (RequestState::ExtendedError, _) => FrontendPhase::ExtendedError,
1239        (RequestState::CopyIn, Some(OperationKind::Query)) => FrontendPhase::SimpleCopyIn,
1240        (RequestState::CopyIn, Some(OperationKind::Execute)) => FrontendPhase::ExtendedCopyIn,
1241        (RequestState::CopyBoth, Some(OperationKind::Query)) => FrontendPhase::SimpleCopyBoth,
1242        (RequestState::CopyBoth, Some(OperationKind::Execute)) => FrontendPhase::ExtendedCopyBoth,
1243        (RequestState::CopyIn | RequestState::CopyBoth, _) => {
1244            unreachable!("COPY phase must belong to Query or Execute")
1245        }
1246        (RequestState::CopyOut | RequestState::Terminated, _) => {
1247            unreachable!("non-accepting frontend phase cannot be prepared")
1248        }
1249    }
1250}
1251
1252fn classify_discard(message: &FrontendMessage) -> Option<OperationKind> {
1253    Some(match message {
1254        FrontendMessage::Parse(_) => OperationKind::Parse,
1255        FrontendMessage::Bind(_) => OperationKind::Bind,
1256        FrontendMessage::Describe(_) => OperationKind::Describe,
1257        FrontendMessage::Execute(_) => OperationKind::Execute,
1258        FrontendMessage::Close(_) => OperationKind::Close,
1259        FrontendMessage::Flush => OperationKind::Flush,
1260        FrontendMessage::Query(_) => OperationKind::Query,
1261        FrontendMessage::FunctionCall(_) => OperationKind::FunctionCall,
1262        FrontendMessage::CopyData(_) => OperationKind::CopyData,
1263        FrontendMessage::CopyDone => OperationKind::CopyDone,
1264        FrontendMessage::CopyFail(_) => OperationKind::CopyFail,
1265        FrontendMessage::Terminate => OperationKind::Terminate,
1266        FrontendMessage::PasswordResponse(_) => return None,
1267        FrontendMessage::Sync => unreachable!("Sync is classified before discard"),
1268    })
1269}
1270
1271const fn initial_response_state(kind: OperationKind) -> backend::RuntimeState {
1272    use OperationKind as O;
1273    match kind {
1274        O::Query => backend::RuntimeState::Simple,
1275        O::FunctionCall => backend::RuntimeState::FunctionResponse,
1276        O::Parse => backend::RuntimeState::ParseResponse,
1277        O::Bind => backend::RuntimeState::BindResponse,
1278        O::Describe => backend::RuntimeState::DescribeResponse,
1279        O::Execute => backend::RuntimeState::ExecuteResponse,
1280        O::Close => backend::RuntimeState::CloseResponse,
1281        O::Sync => backend::RuntimeState::SyncResponse,
1282        O::Flush | O::CopyData | O::CopyDone | O::CopyFail | O::Terminate => {
1283            backend::RuntimeState::Terminated
1284        }
1285    }
1286}
1287
1288fn response_fits(operation: Operation, message: &BackendMessage) -> bool {
1289    backend::project_internal(operation.response_state, message).is_some()
1290}
1291
1292fn response_is_terminal(kind: OperationKind, message: &BackendMessage) -> bool {
1293    use BackendMessage as B;
1294    use OperationKind as O;
1295    match kind {
1296        O::Query | O::FunctionCall | O::Sync => matches!(message, B::ReadyForQuery(_)),
1297        O::Parse => matches!(message, B::ParseComplete | B::ErrorResponse(_)),
1298        O::Bind => matches!(message, B::BindComplete | B::ErrorResponse(_)),
1299        O::Describe => matches!(
1300            message,
1301            B::RowDescription(_) | B::NoData | B::ErrorResponse(_)
1302        ),
1303        O::Execute => matches!(
1304            message,
1305            B::CommandComplete(_) | B::PortalSuspended | B::ErrorResponse(_)
1306        ),
1307        O::Close => matches!(message, B::CloseComplete | B::ErrorResponse(_)),
1308        O::CopyDone => matches!(
1309            message,
1310            B::CopyDone | B::CommandComplete(_) | B::ErrorResponse(_)
1311        ),
1312        O::CopyFail => matches!(message, B::ErrorResponse(_)),
1313        O::Flush | O::CopyData | O::Terminate => true,
1314    }
1315}
1316
1317fn is_extended_kind(kind: OperationKind) -> bool {
1318    !matches!(
1319        kind,
1320        OperationKind::Query | OperationKind::FunctionCall | OperationKind::Terminate
1321    )
1322}
1323
1324fn response_copy_state(state: backend::RuntimeState) -> Option<RequestState> {
1325    use backend::RuntimeState as S;
1326    match state {
1327        S::SimpleCopyIn | S::ExtendedCopyIn => Some(RequestState::CopyIn),
1328        S::SimpleCopyOut | S::ExtendedCopyOut => Some(RequestState::CopyOut),
1329        S::SimpleCopyBoth
1330        | S::SimpleCopyBothClientDone
1331        | S::SimpleCopyBothServerDone
1332        | S::ExtendedCopyBoth
1333        | S::ExtendedCopyBothClientDone
1334        | S::ExtendedCopyBothServerDone => Some(RequestState::CopyBoth),
1335        _ => None,
1336    }
1337}
1338
1339fn is_asynchronous(message: &BackendMessage) -> bool {
1340    Demux::is_asynchronous(message)
1341}
pg_proto/pipeline.rs

pg_proto/
pipeline.rs
