Method serve_with_peer_identity

crates/openshell-server/src/multiplex.rs:147–204

Serve a TLS connection with an optional mTLS peer identity.

(
        &self,
        stream: S,
        peer_identity: Option<Identity>,
    )

    /// Serve a TLS connection with an optional mTLS peer identity.
    pub async fn serve_with_peer_identity<S>(
        &self,
        stream: S,
        peer_identity: Option<Identity>,
    ) -> Result<(), Box<dyn std::error::Error + Send + Sync>>
    where
        S: AsyncRead + AsyncWrite + Unpin + Send + 'static,
    {
        let openshell = OpenShellServer::new(OpenShellService::new(self.state.clone()))
            .max_decoding_message_size(MAX_GRPC_DECODE_SIZE);
        let inference = InferenceServer::new(InferenceService::new(self.state.clone()))
            .max_decoding_message_size(MAX_GRPC_DECODE_SIZE);
        let authz_policy = self.state.config.oidc.as_ref().map(|oidc| AuthzPolicy {
            admin_role: oidc.admin_role.clone(),
            user_role: oidc.user_role.clone(),
            scopes_enabled: !oidc.scopes_claim.is_empty(),
        });
        let authenticator_chain = build_authenticator_chain(&self.state);
        let grpc_service = AuthGrpcRouter::with_peer_identity(
            GrpcRouter::new(openshell, inference),
            authenticator_chain,
            authz_policy,
            self.state
                .config
                .mtls_auth
                .enabled
                .then_some(peer_identity)
                .flatten(),
            self.state.config.mtls_auth.enabled,
            self.state.config.auth.allow_unauthenticated_users,
        );
        let grpc_service =
            GrpcRateLimitService::new(grpc_service, self.state.grpc_rate_limiter.clone());
        let http_service = http_router(self.state.clone());

        let grpc_service = request_id_middleware!(grpc_service);
        let http_service = request_id_middleware!(http_service);

        let service = MultiplexedService::new(grpc_service, http_service);

        let mut builder = Builder::new(TokioExecutor::new());
        // Server-side HTTP/2 keepalive: supervisors hold long-lived sessions, and without
        // it the gateway never PINGs them, so idle/half-dead connections linger and orphan
        // in-flight relay execs. The timer is required — hyper panics on the keepalive
        // interval without one.
        builder
            .http2()
            .timer(TokioTimer::new())
            .adaptive_window(true)
            .keep_alive_interval(Some(Duration::from_secs(20)))
            .keep_alive_timeout(Duration::from_secs(10));

        builder
            .serve_connection_with_upgrades(TokioIo::new(stream), service)
            .await?;

        Ok(())
    }