pg_proto/
intermediary_component.rs

1//! Builder-centred composition of the client-facing and PostgreSQL-facing roles.
2
3use std::{collections::VecDeque, fmt, future::Future, io, pin::Pin};
4
5use tokio::io::{AsyncRead, AsyncWrite};
6
7use crate::{
8    ConnectTarget, NoPipeline, StartupParameters,
9    pipeline::{BackendAction, FrontendAction, FrontendHandling, Pipeline, PipelinePolicy},
10};
11
12/// Required posture for out-of-band cancellation connections.
13///
14/// Forwarding cancellation is implemented by issue #36. Until then the only
15/// safe operational posture is an explicit rejection.
16#[derive(Clone, Copy, Debug, Eq, PartialEq)]
17pub enum CancellationPolicy {
18    /// Reject cancellation packets instead of silently routing them.
19    Reject,
20    /// Resolve and forward cancellation using the configured registry.
21    Forward,
22}
23
24/// Disclosure-safe handling for failures after a downstream connection exists.
25#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
26pub enum EstablishmentFailurePolicy {
27    /// Silently close without exposing internal failure details.
28    #[default]
29    Close,
30    /// Send one fixed, non-disclosing PostgreSQL diagnostic and then close.
31    SafeDiagnostic,
32}
33
34fn safe_establishment_diagnostic() -> crate::codec::BackendMessage {
35    crate::codec::BackendMessage::ErrorResponse(crate::codec::DiagnosticResponse {
36        fields: vec![
37            crate::codec::DiagnosticField {
38                code: b'S',
39                value: bytes::Bytes::from_static(b"ERROR"),
40            },
41            crate::codec::DiagnosticField {
42                code: b'M',
43                value: bytes::Bytes::from_static(b"connection establishment failed"),
44            },
45        ],
46    })
47}
48
49/// A destination and upstream key retained independently of startup routing.
50#[derive(Clone, Debug, Eq, PartialEq)]
51pub struct CancellationRoute {
52    target: ConnectTarget,
53    upstream: crate::demux::CancelKey,
54}
55
56impl CancellationRoute {
57    /// Creates a cancellation route.
58    #[must_use]
59    pub const fn new(target: ConnectTarget, upstream: crate::demux::CancelKey) -> Self {
60        Self { target, upstream }
61    }
62    /// Returns the original destination, including application metadata.
63    #[must_use]
64    pub const fn target(&self) -> &ConnectTarget {
65        &self.target
66    }
67    /// Returns the upstream cancellation key.
68    #[must_use]
69    pub const fn upstream_key(&self) -> &crate::demux::CancelKey {
70        &self.upstream
71    }
72}
73
74/// Application-owned concurrent cancellation mapping and key allocator.
75///
76/// Methods take `&self` so implementations can use an application-selected
77/// lock, actor, shared store, or other concurrency mechanism. No global
78/// `Send`, `Sync`, or `'static` requirement is imposed.
79pub trait IntermediaryCancellationRegistry {
80    /// Collision, allocation, or storage failure.
81    type Error;
82    /// Records a live route and returns the proxy key exposed downstream.
83    ///
84    /// # Errors
85    ///
86    /// Returns an application-defined allocation, collision, or storage error.
87    fn register(&self, route: CancellationRoute) -> Result<crate::demux::CancelKey, Self::Error>;
88    /// Resolves a later out-of-band request without consulting startup routing.
89    fn resolve(&self, client: &crate::demux::CancelKey) -> Option<CancellationRoute>;
90    /// Explicitly detaches a live client key.
91    fn detach(&self, client: &crate::demux::CancelKey) -> Option<CancellationRoute>;
92}
93
94/// Marker registry used by explicit cancellation rejection.
95#[derive(Clone, Copy, Debug, Default)]
96pub struct RejectCancellation;
97impl IntermediaryCancellationRegistry for RejectCancellation {
98    type Error = std::convert::Infallible;
99    fn register(&self, _: CancellationRoute) -> Result<crate::demux::CancelKey, Self::Error> {
100        unreachable!()
101    }
102    fn resolve(&self, _: &crate::demux::CancelKey) -> Option<CancellationRoute> {
103        None
104    }
105    fn detach(&self, _: &crate::demux::CancelKey) -> Option<CancellationRoute> {
106        None
107    }
108}
109
110/// Deterministic failure while assembling an intermediary component.
111#[derive(Clone, Copy, Debug, Eq, PartialEq)]
112pub enum IntermediaryBuildError {
113    /// No complete client-facing server role was supplied.
114    MissingServer,
115    /// No complete PostgreSQL-facing client role was supplied.
116    MissingClient,
117    /// No asynchronous startup resolver was supplied.
118    MissingStartupResolver,
119    /// Cancellation behavior was not selected explicitly.
120    MissingCancellationPolicy,
121}
122
123impl fmt::Display for IntermediaryBuildError {
124    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
125        formatter.write_str(match self {
126            Self::MissingServer => "an intermediary server component is required",
127            Self::MissingClient => "an intermediary client component is required",
128            Self::MissingStartupResolver => "an asynchronous startup resolver is required",
129            Self::MissingCancellationPolicy => "an explicit cancellation policy is required",
130        })
131    }
132}
133
134impl std::error::Error for IntermediaryBuildError {}
135
136/// Immutable server-side facts available before authentication begins.
137#[derive(Clone, Copy, Debug)]
138pub struct InitialServerContext<'a, Peer> {
139    peer: &'a Peer,
140    tls: &'a crate::NegotiatedServerTls,
141}
142
143impl<'a, Peer> InitialServerContext<'a, Peer> {
144    pub(crate) const fn new(peer: &'a Peer, tls: &'a crate::NegotiatedServerTls) -> Self {
145        Self { peer, tls }
146    }
147
148    /// Returns application-supplied peer metadata.
149    #[must_use]
150    pub const fn peer(&self) -> &Peer {
151        self.peer
152    }
153
154    /// Returns transport security negotiated on the client-facing side.
155    #[must_use]
156    pub const fn tls(&self) -> &crate::NegotiatedServerTls {
157        self.tls
158    }
159}
160
161/// Required asynchronous startup routing policy.
162pub trait StartupRouteResolver<Peer> {
163    /// Application resolver failure.
164    type Error;
165
166    /// Selects a destination before client-facing authentication begins.
167    fn resolve<'a>(
168        &'a self,
169        startup: StartupParameters,
170        context: InitialServerContext<'a, Peer>,
171    ) -> Pin<Box<dyn Future<Output = Result<ConnectTarget, Self::Error>> + 'a>>;
172}
173
174/// Optional policy that validates or refines a destination after authentication.
175pub trait AuthenticatedRoutePolicy<Peer, Identity> {
176    /// Application policy failure.
177    type Error;
178    /// Validates or refines the startup-selected target using typed identity evidence.
179    fn route<'a>(
180        &'a self,
181        target: ConnectTarget,
182        context: AuthenticatedRouteContext<'a, Peer, Identity>,
183    ) -> Pin<Box<dyn Future<Output = Result<ConnectTarget, Self::Error>> + 'a>>;
184}
185
186/// Borrowed facts passed to authenticated route policy.
187#[derive(Clone, Copy, Debug)]
188pub struct AuthenticatedRouteContext<'a, Peer, Identity> {
189    peer: &'a Peer,
190    identity: &'a Identity,
191}
192
193impl<Peer, Identity> AuthenticatedRouteContext<'_, Peer, Identity> {
194    /// Returns application-supplied peer metadata.
195    #[must_use]
196    pub const fn peer(&self) -> &Peer {
197        self.peer
198    }
199
200    /// Returns independently verified client-facing identity evidence.
201    #[must_use]
202    pub const fn identity(&self) -> &Identity {
203        self.identity
204    }
205}
206
207/// Identity authenticated-route policy.
208#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
209pub struct AllowAuthenticatedRoute;
210
211impl<Peer, Identity> AuthenticatedRoutePolicy<Peer, Identity> for AllowAuthenticatedRoute {
212    type Error = std::convert::Infallible;
213    fn route<'a>(
214        &'a self,
215        target: ConnectTarget,
216        _context: AuthenticatedRouteContext<'a, Peer, Identity>,
217    ) -> Pin<Box<dyn Future<Output = Result<ConnectTarget, Self::Error>> + 'a>> {
218        Box::pin(async move { Ok(target) })
219    }
220}
221
222/// Result of asynchronously intercepting a client-originated message.
223#[derive(Debug, Eq, PartialEq)]
224pub enum FrontendMiddlewareOutput {
225    /// Forward this owned message to PostgreSQL.
226    Forward(crate::codec::FrontendMessage),
227    /// Consume the message without forwarding it or registering a response.
228    Suppress(crate::codec::FrontendMessage),
229    /// Handle the request locally and emit these responses in protocol order.
230    Respond {
231        /// Consumed client request.
232        request: crate::codec::FrontendMessage,
233        /// Responses generated for this request, in emission order.
234        responses: Vec<crate::codec::BackendMessage>,
235    },
236}
237
238/// Result of asynchronously intercepting a PostgreSQL-originated message.
239#[derive(Debug, Eq, PartialEq)]
240pub enum BackendMiddlewareOutput {
241    /// Forward this owned message to the client.
242    Forward(crate::codec::BackendMessage),
243    /// Replace one PostgreSQL response with one or more ordered client responses.
244    Expand(Vec<crate::codec::BackendMessage>),
245    /// Consume the response after advancing protocol and pipeline state.
246    Suppress(crate::codec::BackendMessage),
247}
248
249/// Asynchronous, fallible middleware at the forwarding boundary.
250pub trait IntermediaryMiddleware<State, ServerContext, ClientContext> {
251    /// Application-defined interception failure.
252    type Error;
253
254    /// Intercepts a client-originated message after server-role middleware and
255    /// before client-role middleware.
256    fn frontend<'a>(
257        &'a mut self,
258        _server: &'a ServerContext,
259        _client: &'a ClientContext,
260        _state: &'a mut State,
261        message: crate::codec::FrontendMessage,
262    ) -> Pin<Box<dyn Future<Output = Result<FrontendMiddlewareOutput, Self::Error>> + 'a>> {
263        Box::pin(async move { Ok(FrontendMiddlewareOutput::Forward(message)) })
264    }
265
266    /// Intercepts a PostgreSQL-originated message after client-role middleware
267    /// and before server-role middleware.
268    fn backend<'a>(
269        &'a mut self,
270        _server: &'a ServerContext,
271        _client: &'a ClientContext,
272        _state: &'a mut State,
273        message: crate::codec::BackendMessage,
274    ) -> Pin<Box<dyn Future<Output = Result<BackendMiddlewareOutput, Self::Error>> + 'a>> {
275        Box::pin(async move { Ok(BackendMiddlewareOutput::Forward(message)) })
276    }
277}
278
279/// Identity forwarding-boundary middleware.
280#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
281pub struct IdentityIntermediaryMiddleware;
282
283impl<State, ServerContext, ClientContext>
284    IntermediaryMiddleware<State, ServerContext, ClientContext> for IdentityIntermediaryMiddleware
285{
286    type Error = std::convert::Infallible;
287}
288
289/// Creates fresh forwarding-boundary middleware for one established pair.
290pub trait IntermediaryMiddlewareFactory<ServerContext, ClientContext> {
291    /// Per-connection boundary handler.
292    type Handler;
293    /// Creates an isolated handler from both distinct role contexts.
294    fn create(&self, server: &ServerContext, client: &ClientContext) -> Self::Handler;
295}
296
297impl<ServerContext, ClientContext, Handler, Factory>
298    IntermediaryMiddlewareFactory<ServerContext, ClientContext> for Factory
299where
300    Factory: Fn(&ServerContext, &ClientContext) -> Handler,
301{
302    type Handler = Handler;
303    fn create(&self, server: &ServerContext, client: &ClientContext) -> Handler {
304        self(server, client)
305    }
306}
307
308impl<ServerContext, ClientContext> IntermediaryMiddlewareFactory<ServerContext, ClientContext>
309    for IdentityIntermediaryMiddleware
310{
311    type Handler = Self;
312    fn create(&self, _server: &ServerContext, _client: &ClientContext) -> Self {
313        *self
314    }
315}
316
317/// A reusable operational intermediary configuration.
318pub struct Intermediary<
319    Server = (),
320    Client = (),
321    Resolver = (),
322    Route = AllowAuthenticatedRoute,
323    Policy = NoPipeline,
324    Boundary = IdentityIntermediaryMiddleware,
325    Cancellation = RejectCancellation,
326> {
327    pub(crate) server: Server,
328    pub(crate) client: Client,
329    pub(crate) resolver: Resolver,
330    pub(crate) route: Route,
331    pub(crate) pipeline: Policy,
332    pub(crate) boundary: Boundary,
333    pub(crate) cancellation: CancellationPolicy,
334    pub(crate) cancellation_registry: Cancellation,
335    pub(crate) failure_policy: EstablishmentFailurePolicy,
336}
337
338impl Intermediary<()> {
339    /// Starts composition of the two complete role configurations.
340    #[must_use]
341    pub fn builder() -> IntermediaryBuilder {
342        IntermediaryBuilder::default()
343    }
344}
345
346impl<S, C, R, A, P, B, K> fmt::Debug for Intermediary<S, C, R, A, P, B, K> {
347    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
348        formatter
349            .debug_struct("Intermediary")
350            .field("server", &"<configured>")
351            .field("client", &"<configured>")
352            .field("resolver", &"<redacted>")
353            .field("authenticated_route", &"<redacted>")
354            .field("cancellation", &self.cancellation)
355            .finish_non_exhaustive()
356    }
357}
358
359/// Progressive builder for [`Intermediary`].
360pub struct IntermediaryBuilder<
361    Server = (),
362    Client = (),
363    Resolver = (),
364    Route = AllowAuthenticatedRoute,
365    Policy = NoPipeline,
366    Boundary = IdentityIntermediaryMiddleware,
367    Cancellation = RejectCancellation,
368> {
369    server: Option<Server>,
370    client: Option<Client>,
371    resolver: Option<Resolver>,
372    route: Route,
373    pipeline: Policy,
374    boundary: Boundary,
375    cancellation: Option<CancellationPolicy>,
376    cancellation_registry: Cancellation,
377    failure_policy: EstablishmentFailurePolicy,
378}
379
380impl Default for IntermediaryBuilder {
381    fn default() -> Self {
382        Self {
383            server: None,
384            client: None,
385            resolver: None,
386            route: AllowAuthenticatedRoute,
387            pipeline: NoPipeline,
388            boundary: IdentityIntermediaryMiddleware,
389            cancellation: None,
390            cancellation_registry: RejectCancellation,
391            failure_policy: EstablishmentFailurePolicy::Close,
392        }
393    }
394}
395
396impl<S, C, R, A, P, B, K> IntermediaryBuilder<S, C, R, A, P, B, K> {
397    /// Supplies the complete client-facing role configuration.
398    #[must_use]
399    pub fn server<Next>(self, server: Next) -> IntermediaryBuilder<Next, C, R, A, P, B, K> {
400        IntermediaryBuilder {
401            server: Some(server),
402            client: self.client,
403            resolver: self.resolver,
404            route: self.route,
405            pipeline: self.pipeline,
406            boundary: self.boundary,
407            cancellation: self.cancellation,
408            cancellation_registry: self.cancellation_registry,
409            failure_policy: self.failure_policy,
410        }
411    }
412
413    /// Supplies the complete PostgreSQL-facing role configuration.
414    #[must_use]
415    pub fn client<Next>(self, client: Next) -> IntermediaryBuilder<S, Next, R, A, P, B, K> {
416        IntermediaryBuilder {
417            server: self.server,
418            client: Some(client),
419            resolver: self.resolver,
420            route: self.route,
421            pipeline: self.pipeline,
422            boundary: self.boundary,
423            cancellation: self.cancellation,
424            cancellation_registry: self.cancellation_registry,
425            failure_policy: self.failure_policy,
426        }
427    }
428
429    /// Supplies the required asynchronous startup resolver.
430    #[must_use]
431    pub fn startup_resolver<Next>(
432        self,
433        resolver: Next,
434    ) -> IntermediaryBuilder<S, C, Next, A, P, B, K> {
435        IntermediaryBuilder {
436            server: self.server,
437            client: self.client,
438            resolver: Some(resolver),
439            route: self.route,
440            pipeline: self.pipeline,
441            boundary: self.boundary,
442            cancellation: self.cancellation,
443            cancellation_registry: self.cancellation_registry,
444            failure_policy: self.failure_policy,
445        }
446    }
447
448    /// Supplies optional post-authentication routing policy.
449    #[must_use]
450    pub fn authenticated_route<Next>(
451        self,
452        route: Next,
453    ) -> IntermediaryBuilder<S, C, R, Next, P, B, K> {
454        IntermediaryBuilder {
455            server: self.server,
456            client: self.client,
457            resolver: self.resolver,
458            route,
459            pipeline: self.pipeline,
460            boundary: self.boundary,
461            cancellation: self.cancellation,
462            cancellation_registry: self.cancellation_registry,
463            failure_policy: self.failure_policy,
464        }
465    }
466
467    /// Selects lock-step or bounded request pipelining.
468    #[must_use]
469    pub fn pipeline<Next: PipelinePolicy>(
470        self,
471        pipeline: Next,
472    ) -> IntermediaryBuilder<S, C, R, A, Next, B, K> {
473        IntermediaryBuilder {
474            server: self.server,
475            client: self.client,
476            resolver: self.resolver,
477            route: self.route,
478            pipeline,
479            boundary: self.boundary,
480            cancellation: self.cancellation,
481            cancellation_registry: self.cancellation_registry,
482            failure_policy: self.failure_policy,
483        }
484    }
485
486    /// Supplies middleware for the forwarding boundary.
487    #[must_use]
488    pub fn middleware<Next>(self, boundary: Next) -> IntermediaryBuilder<S, C, R, A, P, Next, K> {
489        IntermediaryBuilder {
490            server: self.server,
491            client: self.client,
492            resolver: self.resolver,
493            route: self.route,
494            pipeline: self.pipeline,
495            boundary,
496            cancellation: self.cancellation,
497            cancellation_registry: self.cancellation_registry,
498            failure_policy: self.failure_policy,
499        }
500    }
501
502    /// Selects an explicit cancellation posture.
503    #[must_use]
504    pub fn cancellation(mut self, cancellation: CancellationPolicy) -> Self {
505        self.cancellation = match cancellation {
506            CancellationPolicy::Reject => Some(CancellationPolicy::Reject),
507            CancellationPolicy::Forward => None,
508        };
509        self
510    }
511
512    /// Selects conservative close or one fixed safe diagnostic on establishment failure.
513    #[must_use]
514    pub fn establishment_failure(mut self, policy: EstablishmentFailurePolicy) -> Self {
515        self.failure_policy = policy;
516        self
517    }
518
519    /// Enables forwarding through an application-owned concurrent registry.
520    #[must_use]
521    pub fn cancellation_registry<Next>(
522        self,
523        registry: Next,
524    ) -> IntermediaryBuilder<S, C, R, A, P, B, Next> {
525        IntermediaryBuilder {
526            server: self.server,
527            client: self.client,
528            resolver: self.resolver,
529            route: self.route,
530            pipeline: self.pipeline,
531            boundary: self.boundary,
532            cancellation: Some(CancellationPolicy::Forward),
533            cancellation_registry: registry,
534            failure_policy: self.failure_policy,
535        }
536    }
537
538    /// Validates composition and creates a reusable component.
539    ///
540    /// # Errors
541    ///
542    /// Returns the first missing mandatory role, resolver, or cancellation configuration.
543    #[allow(clippy::type_complexity)]
544    pub fn build(self) -> Result<Intermediary<S, C, R, A, P, B, K>, IntermediaryBuildError> {
545        Ok(Intermediary {
546            server: self.server.ok_or(IntermediaryBuildError::MissingServer)?,
547            client: self.client.ok_or(IntermediaryBuildError::MissingClient)?,
548            resolver: self
549                .resolver
550                .ok_or(IntermediaryBuildError::MissingStartupResolver)?,
551            route: self.route,
552            pipeline: self.pipeline,
553            boundary: self.boundary,
554            cancellation: self
555                .cancellation
556                .ok_or(IntermediaryBuildError::MissingCancellationPolicy)?,
557            cancellation_registry: self.cancellation_registry,
558            failure_policy: self.failure_policy,
559        })
560    }
561}
562
563struct StartupResolverAdapter<'a, Resolver> {
564    resolver: &'a Resolver,
565}
566
567/// Failure while decoding or resolving a startup route.
568#[derive(Debug)]
569pub enum StartupResolutionError<Error> {
570    /// A startup parameter was not representable by the structured facade.
571    Parameters(io::Error),
572    /// The application resolver rejected the route.
573    Resolver(Error),
574}
575
576impl<Error: fmt::Display> fmt::Display for StartupResolutionError<Error> {
577    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
578        match self {
579            Self::Parameters(error) => error.fmt(formatter),
580            Self::Resolver(error) => error.fmt(formatter),
581        }
582    }
583}
584
585impl<Error: std::error::Error + 'static> std::error::Error for StartupResolutionError<Error> {
586    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
587        match self {
588            Self::Parameters(error) => Some(error),
589            Self::Resolver(error) => Some(error),
590        }
591    }
592}
593
594impl<Resolver, State, Peer, Identity>
595    crate::server_component::StartupResolver<State, Peer, Identity>
596    for StartupResolverAdapter<'_, Resolver>
597where
598    Resolver: StartupRouteResolver<Peer>,
599{
600    type Route = ConnectTarget;
601    type Error = StartupResolutionError<Resolver::Error>;
602
603    fn defer_ready(&self) -> bool {
604        true
605    }
606
607    fn resolve<'a>(
608        &'a mut self,
609        startup: &'a crate::startup::StartupMessage,
610        context: &'a crate::ServerConnectionContext<Peer, Identity>,
611        _state: &'a mut State,
612    ) -> Pin<Box<dyn Future<Output = Result<Self::Route, Self::Error>> + 'a>> {
613        let parameters = StartupParameters::from_wire(startup);
614        let initial = context
615            .tls_if_known()
616            .map(|tls| InitialServerContext::new(context.peer(), tls));
617        let resolver = self.resolver;
618        Box::pin(async move {
619            let parameters = parameters.map_err(StartupResolutionError::Parameters)?;
620            let initial = initial.expect("startup routing runs after TLS negotiation");
621            resolver
622                .resolve(parameters, initial)
623                .await
624                .map_err(StartupResolutionError::Resolver)
625        })
626    }
627}
628
629/// Failure while establishing both independently authenticated roles.
630pub enum IntermediaryAcceptError<
631    ServerError,
632    ResolverError,
633    RouteError,
634    ClientError,
635    RegistryError = std::convert::Infallible,
636    CancellationError = std::convert::Infallible,
637    MiddlewareError = std::convert::Infallible,
638> {
639    /// Client-facing TLS, startup, or authentication failed.
640    Server(ServerError),
641    /// Startup routing failed before client-facing authentication.
642    StartupRoute(StartupResolutionError<ResolverError>),
643    /// The explicit cancellation posture rejected an out-of-band request.
644    CancellationRejected,
645    /// Authenticated routing rejected or failed to refine the destination.
646    AuthenticatedRoute(RouteError),
647    /// PostgreSQL-facing connection, TLS, startup, or authentication failed.
648    Client(ClientError),
649    /// Cancellation-key allocation, collision detection, or storage failed.
650    CancellationRegistry(RegistryError),
651    /// A generated establishment message could not be written downstream.
652    ServerOutput(io::Error),
653    /// Opening or writing the one-shot upstream cancellation connection failed.
654    Cancellation(CancellationError),
655    /// Forwarding-boundary middleware rejected generated establishment output.
656    Middleware(MiddlewareError),
657}
658
659impl<S, R, A, C, K, X, M> fmt::Debug for IntermediaryAcceptError<S, R, A, C, K, X, M> {
660    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
661        formatter.write_str(match self {
662            Self::Server(_) => "IntermediaryAcceptError::Server([REDACTED])",
663            Self::StartupRoute(_) => "IntermediaryAcceptError::StartupRoute([REDACTED])",
664            Self::CancellationRejected => "IntermediaryAcceptError::CancellationRejected",
665            Self::AuthenticatedRoute(_) => {
666                "IntermediaryAcceptError::AuthenticatedRoute([REDACTED])"
667            }
668            Self::Client(_) => "IntermediaryAcceptError::Client([REDACTED])",
669            Self::CancellationRegistry(_) => {
670                "IntermediaryAcceptError::CancellationRegistry([REDACTED])"
671            }
672            Self::ServerOutput(_) => "IntermediaryAcceptError::ServerOutput([REDACTED])",
673            Self::Cancellation(_) => "IntermediaryAcceptError::Cancellation([REDACTED])",
674            Self::Middleware(_) => "IntermediaryAcceptError::Middleware([REDACTED])",
675        })
676    }
677}
678
679impl<S, R, A, C, K, X, M> fmt::Display for IntermediaryAcceptError<S, R, A, C, K, X, M> {
680    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
681        match self {
682            Self::Server(_) => formatter.write_str("client-facing establishment failed"),
683            Self::StartupRoute(_) => formatter.write_str("startup routing failed"),
684            Self::CancellationRejected => {
685                formatter.write_str("cancellation is explicitly rejected")
686            }
687            Self::AuthenticatedRoute(_) => formatter.write_str("authenticated routing failed"),
688            Self::Client(_) => formatter.write_str("PostgreSQL-facing establishment failed"),
689            Self::CancellationRegistry(_) => {
690                formatter.write_str("cancellation registration failed")
691            }
692            Self::ServerOutput(_) => {
693                formatter.write_str("client-facing establishment output failed")
694            }
695            Self::Cancellation(_) => formatter.write_str("cancellation forwarding failed"),
696            Self::Middleware(_) => {
697                formatter.write_str("forwarding middleware rejected establishment output")
698            }
699        }
700    }
701}
702
703impl<S, R, A, C, K, X, M> std::error::Error for IntermediaryAcceptError<S, R, A, C, K, X, M>
704where
705    S: std::error::Error + 'static,
706    R: std::error::Error + 'static,
707    A: std::error::Error + 'static,
708    C: std::error::Error + 'static,
709    K: std::error::Error + 'static,
710    X: std::error::Error + 'static,
711    M: std::error::Error + 'static,
712{
713    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
714        match self {
715            Self::Server(error) => Some(error),
716            Self::StartupRoute(error) => Some(error),
717            Self::CancellationRejected => None,
718            Self::AuthenticatedRoute(error) => Some(error),
719            Self::Client(error) => Some(error),
720            Self::CancellationRegistry(error) => Some(error),
721            Self::ServerOutput(error) => Some(error),
722            Self::Cancellation(error) => Some(error),
723            Self::Middleware(error) => Some(error),
724        }
725    }
726}
727
728/// Both role contexts recovered during deliberate intermediary teardown.
729#[derive(Debug)]
730pub struct IntermediaryContexts<ServerContext, ClientContext> {
731    server: ServerContext,
732    client: ClientContext,
733}
734
735impl<ServerContext, ClientContext> IntermediaryContexts<ServerContext, ClientContext> {
736    /// Returns the client-facing role context.
737    #[must_use]
738    pub const fn server(&self) -> &ServerContext {
739        &self.server
740    }
741    /// Returns the PostgreSQL-facing role context.
742    #[must_use]
743    pub const fn client(&self) -> &ClientContext {
744        &self.client
745    }
746}
747
748/// One operational, independently authenticated intermediary session.
749pub struct IntermediaryConnection<
750    DT,
751    UT,
752    State,
753    Peer,
754    ServerIdentity,
755    ClientEvidence,
756    ServerHandler,
757    ClientHandler,
758    Boundary,
759    Policy,
760    Cancellation = RejectCancellation,
761> {
762    downstream:
763        crate::server_component::ServerConnectionCore<DT, Peer, ServerIdentity, ServerHandler>,
764    upstream: crate::client_component::ClientConnectionCore<
765        crate::ClientTransport<UT>,
766        crate::Pristine,
767        ClientEvidence,
768        ClientHandler,
769    >,
770    state: State,
771    boundary: Boundary,
772    pipeline: Pipeline<Policy>,
773    target: ConnectTarget,
774    pending_frontend: Option<crate::codec::FrontendMessage>,
775    pending_local: VecDeque<PendingLocalResponses>,
776    cancellation_registry: Cancellation,
777    client_cancel_key: Option<crate::demux::CancelKey>,
778}
779
780struct PendingLocalResponses {
781    operation: crate::pipeline::OperationId,
782    messages: VecDeque<crate::codec::BackendMessage>,
783}
784
785/// Result of accepting either an ordinary session or an out-of-band request.
786#[derive(Debug)]
787pub enum IntermediaryAccept<Connection> {
788    /// A fully established, independently authenticated session pair.
789    Session(Connection),
790    /// The resolved cancellation packet was rewritten and forwarded.
791    CancellationForwarded,
792}
793
794impl<Connection> IntermediaryAccept<Connection> {
795    /// Extracts the ordinary session branch.
796    ///
797    /// # Panics
798    /// Panics when the accepted connection was cancellation-only.
799    #[must_use]
800    pub fn into_session(self) -> Connection {
801        match self {
802            Self::Session(connection) => connection,
803            Self::CancellationForwarded => panic!("accepted cancellation has no session"),
804        }
805    }
806}
807
808/// Direction selected by one cancellation-safe duplex forwarding step.
809#[derive(Debug)]
810pub enum ForwardedMessage {
811    /// A client-originated message was forwarded to PostgreSQL.
812    Frontend(crate::codec::FrontendMessage),
813    /// A PostgreSQL-originated message was forwarded to the client.
814    Backend(crate::codec::BackendMessage),
815    /// One PostgreSQL response expanded into ordered client responses.
816    BackendExpanded {
817        /// Response received from PostgreSQL.
818        source: crate::codec::BackendMessage,
819        /// Responses emitted to the client.
820        messages: Vec<crate::codec::BackendMessage>,
821    },
822    /// A client-originated message was deliberately suppressed.
823    FrontendSuppressed(crate::codec::FrontendMessage),
824    /// A request was handled locally; responses were emitted or queued in order.
825    FrontendLocallyHandled(crate::codec::FrontendMessage),
826    /// A PostgreSQL-originated response advanced protocol state but was suppressed.
827    BackendSuppressed(crate::codec::BackendMessage),
828}
829
830/// Observable result of processing one client-originated message.
831#[derive(Debug, Eq, PartialEq)]
832pub enum FrontendForwarding {
833    /// The message was sent to PostgreSQL.
834    Forwarded(crate::codec::FrontendMessage),
835    /// The message was consumed without being sent.
836    Suppressed(crate::codec::FrontendMessage),
837    /// The request was admitted as local and its responses were emitted or queued.
838    LocallyHandled(crate::codec::FrontendMessage),
839}
840
841impl FrontendForwarding {
842    /// Returns the owned client message represented by this outcome.
843    #[must_use]
844    pub fn into_message(self) -> crate::codec::FrontendMessage {
845        match self {
846            Self::Forwarded(message)
847            | Self::Suppressed(message)
848            | Self::LocallyHandled(message) => message,
849        }
850    }
851}
852
853/// Observable result of processing one PostgreSQL-originated response.
854#[derive(Debug, Eq, PartialEq)]
855pub enum BackendForwarding {
856    /// The response was sent to the client.
857    Forwarded(crate::codec::BackendMessage),
858    /// One PostgreSQL response expanded into these ordered client responses.
859    Expanded {
860        /// Response received from PostgreSQL after client-role interception.
861        source: crate::codec::BackendMessage,
862        /// Responses emitted to the client after server-role interception.
863        messages: Vec<crate::codec::BackendMessage>,
864    },
865    /// The response advanced protocol state but was not sent.
866    Suppressed(crate::codec::BackendMessage),
867}
868
869impl BackendForwarding {
870    /// Returns the owned PostgreSQL response represented by this outcome.
871    #[must_use]
872    pub fn into_message(self) -> crate::codec::BackendMessage {
873        match self {
874            Self::Forwarded(message) | Self::Suppressed(message) => message,
875            Self::Expanded { source, .. } => source,
876        }
877    }
878}
879
880impl<DT, UT, State, Peer, SI, CE, SH, CH, Boundary, Policy, K>
881    IntermediaryConnection<DT, UT, State, Peer, SI, CE, SH, CH, Boundary, Policy, K>
882where
883    Policy: PipelinePolicy,
884{
885    /// Returns the authoritative destination selected for the client component.
886    #[must_use]
887    pub const fn target(&self) -> &ConnectTarget {
888        &self.target
889    }
890    /// Returns the single caller-owned state shared by all three middleware layers.
891    #[must_use]
892    pub const fn state(&self) -> &State {
893        &self.state
894    }
895    /// Returns the proxy-issued cancellation key for this live session.
896    #[must_use]
897    pub const fn cancellation_key(&self) -> Option<&crate::demux::CancelKey> {
898        self.client_cancel_key.as_ref()
899    }
900
901    /// Detaches this session's cancellation mapping explicitly.
902    pub fn detach_cancellation(&mut self) -> Option<CancellationRoute>
903    where
904        K: IntermediaryCancellationRegistry,
905    {
906        self.client_cancel_key
907            .take()
908            .and_then(|key| self.cancellation_registry.detach(&key))
909    }
910}
911
912impl<
913    DT,
914    UT,
915    State,
916    Peer,
917    ServerIdentity,
918    ClientEvidence,
919    ServerHandler,
920    ClientHandler,
921    Boundary,
922    Policy,
923    K,
924>
925    IntermediaryConnection<
926        DT,
927        UT,
928        State,
929        Peer,
930        ServerIdentity,
931        ClientEvidence,
932        ServerHandler,
933        ClientHandler,
934        Boundary,
935        Policy,
936        K,
937    >
938where
939    DT: AsyncRead + AsyncWrite + Unpin,
940    UT: AsyncRead + AsyncWrite + Unpin,
941    ServerHandler:
942        crate::ServerMiddleware<State, crate::ServerConnectionContext<Peer, ServerIdentity>>,
943    ClientHandler: crate::ClientMiddleware<State, crate::ClientConnectionContext<ClientEvidence>>,
944    Boundary: IntermediaryMiddleware<
945            State,
946            crate::ServerConnectionContext<Peer, ServerIdentity>,
947            crate::ClientConnectionContext<ClientEvidence>,
948        >,
949    Policy: PipelinePolicy,
950    K: IntermediaryCancellationRegistry,
951{
952    /// Receives one client message and reports whether it was forwarded,
953    /// suppressed, or handled with pipeline-ordered local responses.
954    ///
955    /// # Errors
956    ///
957    /// Returns middleware, transport, protocol-legality, or capacity failures.
958    pub async fn forward_frontend(
959        &mut self,
960    ) -> Result<FrontendForwarding, ForwardError<Boundary::Error>> {
961        if let Some(message) = self.pending_frontend.take() {
962            self.process_frontend(message, false).await
963        } else {
964            let message = self.downstream.receive_wire_raw().await?;
965            self.process_frontend(message, true).await
966        }
967    }
968
969    async fn process_frontend(
970        &mut self,
971        message: crate::codec::FrontendMessage,
972        intercept_source_and_boundary: bool,
973    ) -> Result<FrontendForwarding, ForwardError<Boundary::Error>> {
974        let decision = if intercept_source_and_boundary {
975            let message = self.downstream.intercept_frontend(&mut self.state, message);
976            self.boundary
977                .frontend(
978                    self.downstream.context(),
979                    self.upstream.context(),
980                    &mut self.state,
981                    message,
982                )
983                .await
984                .map_err(ForwardError::Middleware)?
985        } else {
986            FrontendMiddlewareOutput::Forward(message)
987        };
988        let (message, handling) = match decision {
989            FrontendMiddlewareOutput::Forward(message) => {
990                let message = if intercept_source_and_boundary {
991                    self.upstream.intercept_frontend(&mut self.state, message)
992                } else {
993                    message
994                };
995                (message, FrontendHandling::Forward)
996            }
997            FrontendMiddlewareOutput::Suppress(message) => {
998                return Ok(FrontendForwarding::Suppressed(message));
999            }
1000            FrontendMiddlewareOutput::Respond { request, responses } => {
1001                let admission = self
1002                    .pipeline
1003                    .accept_frontend(request.clone(), FrontendHandling::Local)
1004                    .map_err(ForwardError::Frontend)?;
1005                let FrontendAction::Discard { id } = admission.into_action() else {
1006                    unreachable!()
1007                };
1008                let messages = responses
1009                    .into_iter()
1010                    .map(|message| self.downstream.intercept_backend(&mut self.state, message))
1011                    .collect();
1012                self.pending_local.push_back(PendingLocalResponses {
1013                    operation: id,
1014                    messages,
1015                });
1016                self.flush_local_responses().await?;
1017                return Ok(FrontendForwarding::LocallyHandled(request));
1018            }
1019        };
1020        let admission = match self.pipeline.accept_frontend(message.clone(), handling) {
1021            Ok(admission) => admission,
1022            Err(error) => {
1023                self.pending_frontend = Some(message);
1024                return Err(ForwardError::Frontend(error));
1025            }
1026        };
1027        let FrontendAction::Forward { message, .. } = admission.into_action() else {
1028            unreachable!()
1029        };
1030        self.upstream.send_wire_raw(message.clone()).await?;
1031        Ok(FrontendForwarding::Forwarded(message))
1032    }
1033
1034    /// Receives one legal PostgreSQL response and forwards it downstream in
1035    /// source-role, boundary, destination-role middleware order.
1036    ///
1037    /// # Errors
1038    ///
1039    /// Returns transport, framing, ordering, or protocol-legality failures.
1040    /// Receives one PostgreSQL response and reports whether it was forwarded,
1041    /// expanded, or suppressed.
1042    ///
1043    /// # Errors
1044    ///
1045    /// Returns middleware, transport, ordering, or protocol-legality failures.
1046    pub async fn forward_backend(
1047        &mut self,
1048    ) -> Result<BackendForwarding, ForwardError<Boundary::Error>> {
1049        let message = self.upstream.receive_wire_raw().await?;
1050        self.process_backend(message).await
1051    }
1052
1053    async fn process_backend(
1054        &mut self,
1055        message: crate::codec::BackendMessage,
1056    ) -> Result<BackendForwarding, ForwardError<Boundary::Error>> {
1057        let message = self.upstream.intercept_backend(&mut self.state, message);
1058        let source = message.clone();
1059        let decision = self
1060            .boundary
1061            .backend(
1062                self.downstream.context(),
1063                self.upstream.context(),
1064                &mut self.state,
1065                message,
1066            )
1067            .await
1068            .map_err(ForwardError::Middleware)?;
1069        let outcome = match decision {
1070            BackendMiddlewareOutput::Forward(message) => {
1071                let message = self.downstream.intercept_backend(&mut self.state, message);
1072                let message = self.emit_backend(message).await?;
1073                BackendForwarding::Forwarded(message)
1074            }
1075            BackendMiddlewareOutput::Suppress(message) => {
1076                let message = self.advance_backend(message)?;
1077                BackendForwarding::Suppressed(message)
1078            }
1079            BackendMiddlewareOutput::Expand(messages) => {
1080                if messages.is_empty() {
1081                    return Err(ForwardError::EmptyExpansion(source));
1082                }
1083                let mut emitted = Vec::with_capacity(messages.len());
1084                for message in messages {
1085                    let message = self.downstream.intercept_backend(&mut self.state, message);
1086                    emitted.push(self.emit_backend(message).await?);
1087                }
1088                BackendForwarding::Expanded {
1089                    source,
1090                    messages: emitted,
1091                }
1092            }
1093        };
1094        self.flush_local_responses().await?;
1095        Ok(outcome)
1096    }
1097
1098    fn advance_backend(
1099        &mut self,
1100        message: crate::codec::BackendMessage,
1101    ) -> Result<crate::codec::BackendMessage, ForwardError<Boundary::Error>> {
1102        match self
1103            .pipeline
1104            .accept_backend(message)
1105            .map_err(ForwardError::Backend)?
1106        {
1107            BackendAction::Emit(message) => Ok(message),
1108            BackendAction::Deferred(message) => Err(ForwardError::Deferred(message)),
1109        }
1110    }
1111
1112    async fn emit_backend(
1113        &mut self,
1114        message: crate::codec::BackendMessage,
1115    ) -> Result<crate::codec::BackendMessage, ForwardError<Boundary::Error>> {
1116        let message = self.advance_backend(message)?;
1117        self.downstream.send_wire_raw(message.clone()).await?;
1118        Ok(message)
1119    }
1120
1121    async fn flush_local_responses(&mut self) -> Result<(), ForwardError<Boundary::Error>> {
1122        loop {
1123            let Some(pending) = self.pending_local.front_mut() else {
1124                return Ok(());
1125            };
1126            let Some(message) = pending.messages.pop_front() else {
1127                self.pending_local.pop_front();
1128                continue;
1129            };
1130            match self.pipeline.try_emit_local(pending.operation, message) {
1131                Ok(BackendAction::Emit(message)) => {
1132                    self.downstream.send_wire_raw(message).await?;
1133                }
1134                Ok(BackendAction::Deferred(message)) => {
1135                    pending.messages.push_front(message);
1136                    return Ok(());
1137                }
1138                Err(error) => return Err(ForwardError::Backend(error)),
1139            }
1140        }
1141    }
1142
1143    /// Waits on both transports and forwards whichever legal message becomes
1144    /// available first. This is the duplex driver for asynchronous traffic,
1145    /// COPY BOTH, and physical replication.
1146    ///
1147    /// When frontend capacity is exhausted, the unchanged pending request is
1148    /// retained and only backend progress is polled until capacity recovers.
1149    ///
1150    /// # Errors
1151    ///
1152    /// Returns transport, framing, ordering, protocol-legality, or capacity failures.
1153    pub async fn forward_next(
1154        &mut self,
1155    ) -> Result<ForwardedMessage, ForwardError<Boundary::Error>> {
1156        if self.pending_frontend.is_some() {
1157            let message = self.upstream.receive_wire_raw().await?;
1158            return self
1159                .process_backend(message)
1160                .await
1161                .map(|outcome| match outcome {
1162                    BackendForwarding::Forwarded(message) => ForwardedMessage::Backend(message),
1163                    BackendForwarding::Expanded { source, messages } => {
1164                        ForwardedMessage::BackendExpanded { source, messages }
1165                    }
1166                    BackendForwarding::Suppressed(message) => {
1167                        ForwardedMessage::BackendSuppressed(message)
1168                    }
1169                });
1170        }
1171        tokio::select! {
1172            result = self.downstream.receive_wire_raw() => {
1173                let message = result?;
1174                self.process_frontend(message, true).await.map(|outcome| match outcome {
1175                    FrontendForwarding::Forwarded(message) => ForwardedMessage::Frontend(message),
1176                    FrontendForwarding::Suppressed(message) => ForwardedMessage::FrontendSuppressed(message),
1177                    FrontendForwarding::LocallyHandled(message) => ForwardedMessage::FrontendLocallyHandled(message),
1178                })
1179            }
1180            result = self.upstream.receive_wire_raw() => {
1181                let message = result?;
1182                self.process_backend(message).await.map(|outcome| match outcome {
1183                    BackendForwarding::Forwarded(message) => ForwardedMessage::Backend(message),
1184                    BackendForwarding::Expanded { source, messages } => {
1185                        ForwardedMessage::BackendExpanded { source, messages }
1186                    }
1187                    BackendForwarding::Suppressed(message) => ForwardedMessage::BackendSuppressed(message),
1188                })
1189            }
1190        }
1191    }
1192
1193    /// Deliberately tears down both roles and recovers transports, handlers,
1194    /// contexts, boundary middleware, and the sole connection state.
1195    #[allow(clippy::type_complexity)]
1196    pub fn teardown(
1197        mut self,
1198    ) -> (
1199        crate::AcceptedServerTransport<DT>,
1200        crate::ClientTransport<UT>,
1201        State,
1202        Boundary,
1203        (ServerHandler, ClientHandler),
1204        IntermediaryContexts<
1205            crate::ServerConnectionContext<Peer, ServerIdentity>,
1206            crate::ClientConnectionContext<ClientEvidence>,
1207        >,
1208    ) {
1209        let _ = self.detach_cancellation();
1210        let (downstream, server_handler, server_context) = self.downstream.into_parts();
1211        let (upstream, client_handler, client_context) = self.upstream.into_parts();
1212        (
1213            downstream,
1214            upstream,
1215            self.state,
1216            self.boundary,
1217            (server_handler, client_handler),
1218            IntermediaryContexts {
1219                server: server_context,
1220                client: client_context,
1221            },
1222        )
1223    }
1224}
1225
1226/// Operational forwarding or pipeline projection failure.
1227#[derive(Debug)]
1228pub enum ForwardError<MiddlewareError = std::convert::Infallible> {
1229    /// Transport, decoding, or encoding failure.
1230    Io(io::Error),
1231    /// Frontend backpressure or protocol-legality rejection.
1232    Frontend(crate::pipeline::FrontendProjectionError),
1233    /// Backend protocol-legality rejection.
1234    Backend(crate::pipeline::BackendProjectionError),
1235    /// A bounded response arrived before its operation became emittable.
1236    Deferred(crate::codec::BackendMessage),
1237    /// Backend fan-out did not contain a replacement response.
1238    EmptyExpansion(crate::codec::BackendMessage),
1239    /// Forwarding-boundary middleware rejected a message.
1240    Middleware(MiddlewareError),
1241}
1242
1243impl<E> fmt::Display for ForwardError<E> {
1244    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
1245        match self {
1246            Self::Io(error) => error.fmt(formatter),
1247            Self::Frontend(_) => {
1248                formatter.write_str("frontend message violates pipeline legality or capacity")
1249            }
1250            Self::Backend(_) => formatter.write_str("backend message violates pipeline legality"),
1251            Self::Deferred(_) => formatter.write_str("backend response is not yet emittable"),
1252            Self::EmptyExpansion(_) => {
1253                formatter.write_str("backend expansion must contain at least one response")
1254            }
1255            Self::Middleware(_) => formatter.write_str("forwarding middleware rejected a message"),
1256        }
1257    }
1258}
1259
1260impl<E> std::error::Error for ForwardError<E>
1261where
1262    E: std::error::Error + 'static,
1263{
1264    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
1265        match self {
1266            Self::Io(error) => Some(error),
1267            Self::Middleware(error) => Some(error),
1268            Self::Frontend(_) | Self::Backend(_) | Self::Deferred(_) | Self::EmptyExpansion(_) => {
1269                None
1270            }
1271        }
1272    }
1273}
1274
1275impl<E> From<io::Error> for ForwardError<E> {
1276    fn from(error: io::Error) -> Self {
1277        Self::Io(error)
1278    }
1279}
1280
1281impl<ST, SA, SM, Connector, CT, CA, CM, Resolver, Route, Policy, Boundary, K>
1282    Intermediary<
1283        crate::Server<ST, SA, SM>,
1284        crate::Client<Connector, CT, CA, CM>,
1285        Resolver,
1286        Route,
1287        Policy,
1288        Boundary,
1289        K,
1290    >
1291where
1292    ST: crate::ServerTlsConfiguration,
1293    SA: crate::ServerAuthenticationProvider,
1294    CT: crate::client_component::ClientTlsConfiguration,
1295    CA: crate::ClientAuthentication,
1296    CM: crate::MiddlewareFactory<crate::ClientInitialContext>,
1297    Policy: PipelinePolicy,
1298    K: IntermediaryCancellationRegistry + Clone,
1299{
1300    /// Establishes both independently authenticated roles around one shared state.
1301    ///
1302    /// # Errors
1303    ///
1304    /// Returns the typed failure from either role or routing policy, or explicit
1305    /// cancellation rejection.
1306    #[allow(clippy::type_complexity, clippy::too_many_lines)]
1307    pub async fn accept<DT, State, Peer, CW, UT, CE>(
1308        &self,
1309        transport: DT,
1310        peer: Peer,
1311        state: State,
1312    ) -> Result<
1313        IntermediaryAccept<
1314            IntermediaryConnection<
1315                DT,
1316                UT,
1317                State,
1318                Peer,
1319                <SA::Authentication as crate::ServerAuthentication<Peer>>::Identity,
1320                CA::Evidence,
1321                <SM as crate::MiddlewareFactory<
1322                    crate::ServerConnectionContext<
1323                        Peer,
1324                        <SA::Authentication as crate::ServerAuthentication<Peer>>::Identity,
1325                    >,
1326                >>::Handler,
1327                <CM as crate::MiddlewareFactory<crate::ClientInitialContext>>::Handler,
1328                <Boundary as IntermediaryMiddlewareFactory<
1329                    crate::ServerConnectionContext<
1330                        Peer,
1331                        <SA::Authentication as crate::ServerAuthentication<Peer>>::Identity,
1332                    >,
1333                    crate::ClientConnectionContext<CA::Evidence>,
1334                >>::Handler,
1335                Policy,
1336                K,
1337            >,
1338        >,
1339        IntermediaryAcceptError<
1340            crate::AcceptError<
1341                <ST::Provider as crate::ServerIdentityProvider>::Error,
1342                <SA::Authentication as crate::ServerAuthentication<Peer>>::Error,
1343            >,
1344            Resolver::Error,
1345            Route::Error,
1346            crate::ConnectError<
1347                CE,
1348                crate::ClientTlsError<<CT::Provider as crate::ClientTlsProvider>::Error>,
1349                crate::ClientAuthenticationError<CA::Error>,
1350            >,
1351            K::Error,
1352            crate::CancelError<CE>,
1353            <<Boundary as IntermediaryMiddlewareFactory<
1354                crate::ServerConnectionContext<
1355                    Peer,
1356                    <SA::Authentication as crate::ServerAuthentication<Peer>>::Identity,
1357                >,
1358                crate::ClientConnectionContext<CA::Evidence>,
1359            >>::Handler as IntermediaryMiddleware<
1360                State,
1361                crate::ServerConnectionContext<
1362                    Peer,
1363                    <SA::Authentication as crate::ServerAuthentication<Peer>>::Identity,
1364                >,
1365                crate::ClientConnectionContext<CA::Evidence>,
1366            >>::Error,
1367        >,
1368    >
1369    where
1370        DT: AsyncRead + AsyncWrite + Unpin,
1371        UT: AsyncRead + AsyncWrite + Unpin,
1372        SA::Authentication: crate::ServerAuthentication<Peer>,
1373        SM: crate::MiddlewareFactory<
1374                crate::ServerConnectionContext<
1375                    Peer,
1376                    <SA::Authentication as crate::ServerAuthentication<Peer>>::Identity,
1377                >,
1378            >,
1379        <SM as crate::MiddlewareFactory<
1380            crate::ServerConnectionContext<
1381                Peer,
1382                <SA::Authentication as crate::ServerAuthentication<Peer>>::Identity,
1383            >,
1384        >>::Handler: crate::ServerMiddleware<
1385                State,
1386                crate::ServerConnectionContext<
1387                    Peer,
1388                    <SA::Authentication as crate::ServerAuthentication<Peer>>::Identity,
1389                >,
1390            >,
1391        Resolver: StartupRouteResolver<Peer>,
1392        Connector: Fn(&ConnectTarget) -> CW,
1393        CW: Future<Output = Result<UT, CE>>,
1394        <CM as crate::MiddlewareFactory<crate::ClientInitialContext>>::Handler:
1395            crate::ClientMiddleware<State, crate::ClientConnectionContext<CA::Evidence>>,
1396        Route: AuthenticatedRoutePolicy<
1397                Peer,
1398                <SA::Authentication as crate::ServerAuthentication<Peer>>::Identity,
1399            >,
1400        Boundary: IntermediaryMiddlewareFactory<
1401                crate::ServerConnectionContext<
1402                    Peer,
1403                    <SA::Authentication as crate::ServerAuthentication<Peer>>::Identity,
1404                >,
1405                crate::ClientConnectionContext<CA::Evidence>,
1406            >,
1407        <Boundary as IntermediaryMiddlewareFactory<
1408            crate::ServerConnectionContext<
1409                Peer,
1410                <SA::Authentication as crate::ServerAuthentication<Peer>>::Identity,
1411            >,
1412            crate::ClientConnectionContext<CA::Evidence>,
1413        >>::Handler: IntermediaryMiddleware<
1414                State,
1415                crate::ServerConnectionContext<
1416                    Peer,
1417                    <SA::Authentication as crate::ServerAuthentication<Peer>>::Identity,
1418                >,
1419                crate::ClientConnectionContext<CA::Evidence>,
1420            >,
1421    {
1422        let mut resolver = StartupResolverAdapter {
1423            resolver: &self.resolver,
1424        };
1425        let (accepted, selected) = self
1426            .server
1427            .accept_routed(transport, peer, state, &mut resolver)
1428            .await
1429            .map_err(|error| match error {
1430                crate::server_component::RoutedAcceptError::Accept(error) => {
1431                    IntermediaryAcceptError::Server(error)
1432                }
1433                crate::server_component::RoutedAcceptError::Route(error) => {
1434                    IntermediaryAcceptError::StartupRoute(error)
1435                }
1436            })?;
1437        let mut downstream = match accepted {
1438            crate::ServerAccept::Session(downstream) => downstream,
1439            crate::ServerAccept::Cancellation(cancellation) => {
1440                if self.cancellation == CancellationPolicy::Reject {
1441                    let _ = cancellation.teardown();
1442                    return Err(IntermediaryAcceptError::CancellationRejected);
1443                }
1444                let request = cancellation.request();
1445                let client_key = crate::demux::CancelKey {
1446                    process_id: request.process_id(),
1447                    secret_key: bytes::Bytes::copy_from_slice(request.secret_key()),
1448                };
1449                let Some(route) = self.cancellation_registry.resolve(&client_key) else {
1450                    let _ = cancellation.teardown();
1451                    return Err(IntermediaryAcceptError::CancellationRejected);
1452                };
1453                if let Err(error) = self
1454                    .client
1455                    .cancel(route.target(), route.upstream_key())
1456                    .await
1457                {
1458                    let _ = cancellation.teardown();
1459                    return Err(IntermediaryAcceptError::Cancellation(error));
1460                }
1461                let _ = cancellation.teardown();
1462                return Ok(IntermediaryAccept::CancellationForwarded);
1463            }
1464        };
1465        let startup = match StartupParameters::from_wire(downstream.startup()) {
1466            Ok(startup) => startup,
1467            Err(error) => {
1468                if self.failure_policy == EstablishmentFailurePolicy::SafeDiagnostic {
1469                    let _ = downstream
1470                        .send_generated_error(safe_establishment_diagnostic())
1471                        .await;
1472                }
1473                let _ = downstream.teardown();
1474                return Err(IntermediaryAcceptError::StartupRoute(
1475                    StartupResolutionError::Parameters(error),
1476                ));
1477            }
1478        };
1479        let context = AuthenticatedRouteContext {
1480            peer: downstream.context().peer(),
1481            identity: downstream.context().identity(),
1482        };
1483        let Some(selected) = selected else {
1484            let _ = downstream.teardown();
1485            return Err(IntermediaryAcceptError::CancellationRejected);
1486        };
1487        let selected = match self.route.route(selected, context).await {
1488            Ok(target) => target,
1489            Err(error) => {
1490                if self.failure_policy == EstablishmentFailurePolicy::SafeDiagnostic {
1491                    let _ = downstream
1492                        .send_generated_error(safe_establishment_diagnostic())
1493                        .await;
1494                }
1495                let _ = downstream.teardown();
1496                return Err(IntermediaryAcceptError::AuthenticatedRoute(error));
1497            }
1498        };
1499        let (mut downstream, mut state) = downstream.into_core_and_state();
1500        let upstream = match self
1501            .client
1502            .connect_core(selected.clone(), startup, &mut state)
1503            .await
1504        {
1505            Ok(upstream) => upstream,
1506            Err(error) => {
1507                if self.failure_policy == EstablishmentFailurePolicy::SafeDiagnostic {
1508                    let diagnostic = safe_establishment_diagnostic();
1509                    let diagnostic = downstream.intercept_backend(&mut state, diagnostic);
1510                    if matches!(diagnostic, crate::codec::BackendMessage::ErrorResponse(_)) {
1511                        // A failed encode/write is a terminal close; do not recursively
1512                        // invoke failure handling or middleware.
1513                        let _ = downstream.send_wire_raw(diagnostic).await;
1514                    }
1515                }
1516                let _ = downstream.into_parts();
1517                return Err(IntermediaryAcceptError::Client(error));
1518            }
1519        };
1520        let boundary = self
1521            .boundary
1522            .create(downstream.context(), upstream.context());
1523        let (client_cancel_key, backend_key_message) =
1524            match (self.cancellation, upstream.context().backend_key().cloned()) {
1525                (CancellationPolicy::Forward, Some(upstream_key)) => {
1526                    let client_key = match self
1527                        .cancellation_registry
1528                        .register(CancellationRoute::new(selected.clone(), upstream_key))
1529                    {
1530                        Ok(key) => key,
1531                        Err(error) => {
1532                            if self.failure_policy == EstablishmentFailurePolicy::SafeDiagnostic {
1533                                let diagnostic = downstream
1534                                    .intercept_backend(&mut state, safe_establishment_diagnostic());
1535                                if matches!(
1536                                    diagnostic,
1537                                    crate::codec::BackendMessage::ErrorResponse(_)
1538                                ) {
1539                                    let _ = downstream.send_wire_raw(diagnostic).await;
1540                                }
1541                            }
1542                            let _ = downstream.into_parts();
1543                            let _ = upstream.into_parts();
1544                            return Err(IntermediaryAcceptError::CancellationRegistry(error));
1545                        }
1546                    };
1547                    let message = crate::codec::BackendMessage::BackendKeyData {
1548                        process_id: client_key.process_id,
1549                        secret_key: client_key.secret_key.clone(),
1550                    };
1551                    (Some(client_key), Some(message))
1552                }
1553                _ => (None, None),
1554            };
1555        let mut connection = IntermediaryConnection {
1556            downstream,
1557            upstream,
1558            state,
1559            boundary,
1560            pipeline: Pipeline::new(self.pipeline),
1561            target: selected,
1562            pending_frontend: None,
1563            pending_local: VecDeque::new(),
1564            cancellation_registry: self.cancellation_registry.clone(),
1565            client_cancel_key,
1566        };
1567        if let Some(message) = backend_key_message {
1568            let expected = message.clone();
1569            let message = connection
1570                .boundary
1571                .backend(
1572                    connection.downstream.context(),
1573                    connection.upstream.context(),
1574                    &mut connection.state,
1575                    message,
1576                )
1577                .await;
1578            let message = match message {
1579                Ok(BackendMiddlewareOutput::Forward(message)) => message,
1580                Ok(BackendMiddlewareOutput::Suppress(_) | BackendMiddlewareOutput::Expand(_)) => {
1581                    let _ = connection.detach_cancellation();
1582                    let _ = connection.teardown();
1583                    return Err(IntermediaryAcceptError::ServerOutput(io::Error::new(
1584                        io::ErrorKind::InvalidData,
1585                        "middleware suppressed or expanded generated cancellation key",
1586                    )));
1587                }
1588                Err(error) => {
1589                    let _ = connection.detach_cancellation();
1590                    let _ = connection.teardown();
1591                    return Err(IntermediaryAcceptError::Middleware(error));
1592                }
1593            };
1594            let message = connection
1595                .downstream
1596                .intercept_backend(&mut connection.state, message);
1597            if message != expected {
1598                let _ = connection.detach_cancellation();
1599                let _ = connection.teardown();
1600                return Err(IntermediaryAcceptError::ServerOutput(io::Error::new(
1601                    io::ErrorKind::InvalidData,
1602                    "middleware rejected generated cancellation key",
1603                )));
1604            }
1605            if let Err(error) = connection.downstream.send_wire_raw(message).await {
1606                let _ = connection.detach_cancellation();
1607                let _ = connection.teardown();
1608                return Err(IntermediaryAcceptError::ServerOutput(error));
1609            }
1610        }
1611        let ready = connection.downstream.intercept_backend(
1612            &mut connection.state,
1613            crate::codec::BackendMessage::ReadyForQuery(crate::codec::TransactionStatus::Idle),
1614        );
1615        if !matches!(ready, crate::codec::BackendMessage::ReadyForQuery(_)) {
1616            let _ = connection.detach_cancellation();
1617            let _ = connection.teardown();
1618            return Err(IntermediaryAcceptError::ServerOutput(io::Error::new(
1619                io::ErrorKind::InvalidData,
1620                "middleware rejected generated readiness",
1621            )));
1622        }
1623        if let Err(error) = connection.downstream.send_wire_raw(ready).await {
1624            let _ = connection.detach_cancellation();
1625            let _ = connection.teardown();
1626            return Err(IntermediaryAcceptError::ServerOutput(error));
1627        }
1628        Ok(IntermediaryAccept::Session(connection))
1629    }
1630}
pg_proto/intermediary_component.rs

pg_proto/
intermediary_component.rs
