zksync_consensus_network 0.13.0

ZKsync consensus network component
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144

#![allow(dead_code)]
use std::collections::BTreeMap;

use anyhow::Context as _;
use rand::Rng;
use zksync_concurrency::{ctx, limiter};
use zksync_consensus_roles::validator;

use super::*;
use crate::{frame, mux, noise, preface, rpc, testonly::make_config, Config};

/// Connection.
pub(super) struct Conn {
    accept: BTreeMap<mux::CapabilityId, Arc<mux::StreamQueue>>,
    connect: BTreeMap<mux::CapabilityId, Arc<mux::StreamQueue>>,
}

/// Background task of the connection.
pub(super) struct ConnRunner {
    mux: mux::Mux,
    stream: noise::Stream,
}

impl ConnRunner {
    /// Runs the background task of the connection.
    pub(super) async fn run(self, ctx: &ctx::Ctx) -> Result<(), mux::RunError> {
        self.mux.run(ctx, self.stream).await
    }
}

fn mux_entry<R: rpc::Rpc>(ctx: &ctx::Ctx) -> (mux::CapabilityId, Arc<mux::StreamQueue>) {
    (
        R::CAPABILITY.id(),
        mux::StreamQueue::new(ctx, R::INFLIGHT, limiter::Rate::INF),
    )
}

/// Establishes an anonymous gossipnet connection to a peer.
pub(super) async fn connect(
    ctx: &ctx::Ctx,
    peer: &Config,
    genesis: validator::GenesisHash,
) -> ctx::Result<(Conn, ConnRunner)> {
    assert!(peer.gossip.dynamic_inbound_limit > 0);
    let addr = peer
        .public_addr
        .resolve(ctx)
        .await?
        .context("peer.public_addr.resolve()")?[0];
    let mut stream = preface::connect(ctx, addr, preface::Endpoint::GossipNet)
        .await
        .context("preface::connect()")?;
    let cfg = make_config(ctx.rng().gen());
    handshake::outbound(ctx, &cfg, genesis, &mut stream, &peer.gossip.key.public())
        .await
        .context("handshake::outbound()")?;
    let conn = Conn {
        accept: [
            mux_entry::<rpc::get_block::Rpc>(ctx),
            mux_entry::<rpc::push_block_store_state::Rpc>(ctx),
        ]
        .into(),
        connect: [
            mux_entry::<rpc::get_block::Rpc>(ctx),
            mux_entry::<rpc::push_block_store_state::Rpc>(ctx),
        ]
        .into(),
    };
    let mux = mux::Mux {
        cfg: Arc::new(rpc::MUX_CONFIG.clone()),
        accept: conn.accept.clone(),
        connect: conn.connect.clone(),
    };
    Ok((conn, ConnRunner { mux, stream }))
}

impl Conn {
    /// Opens a server-side stream.
    pub(super) async fn open_server<R: rpc::Rpc>(
        &self,
        ctx: &ctx::Ctx,
    ) -> ctx::OrCanceled<ServerStream<R>> {
        Ok(ServerStream(
            self.connect
                .get(&R::CAPABILITY.id())
                .unwrap()
                .open(ctx)
                .await?,
            std::marker::PhantomData,
        ))
    }

    /// Opens a client-side stream.
    pub(super) async fn open_client<R: rpc::Rpc>(
        &self,
        ctx: &ctx::Ctx,
    ) -> ctx::OrCanceled<ClientStream<R>> {
        Ok(ClientStream(
            self.accept
                .get(&R::CAPABILITY.id())
                .unwrap()
                .open(ctx)
                .await?,
            std::marker::PhantomData,
        ))
    }
}

/// Client side stream.
pub(super) struct ClientStream<R>(mux::Stream, std::marker::PhantomData<R>);
/// Server side stream.
pub(super) struct ServerStream<R>(mux::Stream, std::marker::PhantomData<R>);

impl<R: rpc::Rpc> ClientStream<R> {
    /// Sends a request.
    pub(super) async fn send(&mut self, ctx: &ctx::Ctx, req: &R::Req) -> anyhow::Result<()> {
        frame::mux_send_proto(ctx, &mut self.0.write, req).await?;
        self.0.write.flush(ctx).await?;
        Ok(())
    }

    /// Receives a response.
    pub(super) async fn recv(mut self, ctx: &ctx::Ctx) -> anyhow::Result<R::Resp> {
        Ok(frame::mux_recv_proto(ctx, &mut self.0.read, usize::MAX)
            .await?
            .0)
    }
}

impl<R: rpc::Rpc> ServerStream<R> {
    /// Sends a response.
    pub(super) async fn send(mut self, ctx: &ctx::Ctx, resp: &R::Resp) -> anyhow::Result<()> {
        frame::mux_send_proto(ctx, &mut self.0.write, resp).await?;
        self.0.write.flush(ctx).await?;
        Ok(())
    }

    /// Receives a request.
    pub(super) async fn recv(&mut self, ctx: &ctx::Ctx) -> anyhow::Result<R::Req> {
        Ok(frame::mux_recv_proto(ctx, &mut self.0.read, usize::MAX)
            .await?
            .0)
    }
}