Function handle_stream

nodedb-cluster/src/transport/server.rs:246–354

(
    handler: Arc<H>,
    auth: Arc<AuthContext>,
    identity_store: Arc<S>,
    peer_cert_der: Option<Vec<u8>>,
    conn: quinn::Connection,
    mut send: quinn::SendStream,
    mut recv: quinn::Re

/// module docstring for the rationale.
#[allow(clippy::too_many_arguments)]
async fn handle_stream<H: RaftRpcHandler, S: PeerIdentityStore>(
    handler: Arc<H>,
    auth: Arc<AuthContext>,
    identity_store: Arc<S>,
    peer_cert_der: Option<Vec<u8>>,
    conn: quinn::Connection,
    mut send: quinn::SendStream,
    mut recv: quinn::RecvStream,
    mut shutdown: watch::Receiver<bool>,
) -> Result<()> {
    let work = async {
        // 1. Read one envelope.
        let envelope = read_envelope(&mut recv).await?;
        let (fields, inner_frame) = auth_envelope::parse_envelope(&envelope, &auth.mac_key)?;

        // 2. Replay window — under the advertised from_node_id (MAC-verified).
        //    Self-addressed frames skip the window: when a node dispatches
        //    an RPC to itself over the transport, the shared `AuthContext`
        //    means one window is updated by both the server-side request
        //    accept (here) and the client-side response accept (in
        //    `send.rs::parse_inbound`). Skipping when `from == local`
        //    keeps the two flows from tripping on each other's entries —
        //    a self-addressed frame can't have been replayed by an
        //    external attacker by definition.
        if fields.from_node_id != auth.local_node_id {
            auth.peer_seq_in.accept(fields.from_node_id, fields.seq)?;
        }

        // 3b. Peer identity check — binds the MAC-verified node_id to the
        //     TLS certificate.  Self-addressed frames skip the check by the
        //     same reasoning as the replay window above.
        if fields.from_node_id != auth.local_node_id
            && let Some(cert_der) = &peer_cert_der
        {
            let node_info = identity_store.get_node_info(fields.from_node_id);
            match node_info {
                Some(ref info) => match verify_peer_identity(info, cert_der) {
                    VerifyOutcome::Accepted { method } => {
                        debug!(
                            node_id = fields.from_node_id,
                            ?method,
                            "peer identity verified"
                        );
                    }
                    VerifyOutcome::BootstrapAccepted => {
                        warn!(
                            node_id = fields.from_node_id,
                            "peer identity not pinned — bootstrap window accepted"
                        );
                    }
                    VerifyOutcome::Rejected => {
                        warn!(
                            node_id = fields.from_node_id,
                            "peer identity mismatch — closing connection"
                        );
                        conn.close(IDENTITY_MISMATCH_QUIC_ERROR, b"peer identity mismatch");
                        return Err(ClusterError::Transport {
                            detail: format!(