feat(eth-wire): Implement `p2p` stream #114

Rjected · 2022-10-21T17:21:02Z

WIP meant to implement #64, specifically the p2p connection part

TODO:

complete p2p handshake
finish Stream and Sink implementations
- transform message id based on shared capabilities
complete p2p state task
implement test on top of ECIESStream and another stream type
implement EthStream test on top of P2PStream
consider moving some types out of p2pstream.rs

mattsse

let's allow dead_code for now, otherwise, this is a bit hard to review in the browser.

crates/net/eth-wire/src/p2pstream.rs

gakonst · 2022-10-22T01:11:07Z

crates/net/eth-wire/src/p2pstream.rs

+// snappy::encode(ping): [0x01, 0x00, 0x80]
+// snappy::encode(pong): [0x01, 0x00, 0x80]
+// snappy::encode(reason): [0x01, 0x00, reason as u8]
+// the reason for doing this is so we don't need to run it through the encoder or decoder.


Can probably implement it as a constant and add a test that checks they are correct

gakonst · 2022-10-22T01:31:38Z

crates/net/eth-wire/src/p2pstream.rs

+// complete overhaul of the p2p protocol?
+// should we even accept protocol v4?
+
+// how should we accept arbitrary subprotocol messages?


Why do we need arbitrary subprotocols and how would we use them?

mattsse

nice progress.

I think the main loop is halfway there, but since Ping/Pong are part of the message protocol, we can simplify how they are tracked

mattsse · 2022-10-26T09:15:04Z

crates/net/eth-wire/src/p2pstream.rs

@@ -31,7 +34,8 @@ const MAX_P2P_MESSAGE_ID: u8 = P2PMessageID::Pong as u8;
 const HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(10);
 const PING_TIMEOUT: Duration = Duration::from_secs(15);
 const PING_INTERVAL: Duration = Duration::from_secs(60);
-const MAX_FAILED_PINGS: usize = 3;
+const GRACE_PERIOD: Duration = Duration::from_secs(2);
+const MAX_FAILED_PINGS: u8 = 3;

 /// A P2PStream wraps over any `Stream` that yields bytes and makes it compatible with `p2p`


this should mention what (part of) the protocol this represents.

mattsse · 2022-10-26T09:18:09Z

crates/net/eth-wire/src/p2pstream.rs

+    ///
+    /// This task must be created only after the [`Hello`] handshake has been completed, so this
+    /// method will return an error if the [`P2PStream`] is not yet authed.
+    pub fn ping_task(&mut self) -> Result<impl Future<Output = Result<(), P2PStreamError>> + '_, P2PStreamError> {


This looks like the actual session loop?

There seems to big a big emphasis on ping/pongs here, but I don't think there are that important? aren't they just regular requests that expect a pong response?

mattsse · 2022-10-26T09:33:50Z

crates/net/eth-wire/src/p2pstream.rs

+        // when we are not waiting for a pong, we will only be selecting on the ping interval.
+        // otherwise, we will be retrying the ping until we either get a pong or we have reached
+        // the maximum number of retries.
+        let task = async move {


we can simplify this by moving everything in the poll function and looping over messages

fn poll() { // here we need to poll the POLL_INTERVAL or `pinger` type to decide if a pong did not arrive, or whether to send a new ping while let Some(msg) = this.inner.poll_next(cx) { // this is where we handle all messages including pings } }

The pinger looks good, we can add a poll(cx) function that returns Poll<Result<Ping, Err>> where Ok(Ping) should trigger a new ping message and Err(Err) that a pong did not arrive as expected.

gakonst

love the direction

gakonst · 2022-10-27T08:03:28Z

crates/net/eth-wire/src/p2pstream.rs

+/// [`HANDSHAKE_TIMEOUT`] determines the amount of time to wait before determining that a `p2p`
+/// handshake has timed out.
+const HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(10);
+const PING_TIMEOUT: Duration = Duration::from_secs(15);
+const PING_INTERVAL: Duration = Duration::from_secs(60);
+const GRACE_PERIOD: Duration = Duration::from_secs(2);
+const MAX_FAILED_PINGS: u8 = 3;


Are these protocol-specific? Or should they be made configurable?

these are not protocol specific, they could definitely be made configurable (with these values as defaults)

gakonst · 2022-10-27T08:07:24Z

crates/net/eth-wire/src/p2pstream.rs

+    /// Whether the `Hello` handshake has been completed
+    authed: bool,


I think we should refactor the stream to UnauthenticatedStream and P2PStream and remove the auth: bool variable. The UnauthenticatedStream would only contain the handshake method and it'd return a P2PStream. This lets us remove the if self.authed error code path from the start_send function below: https://github.com/foundry-rs/reth/pull/114/files#diff-669cc8ac18463cf275fc4d66d3d4dff60e2ae6d4ff6e634098db97fc83dfdb47R427-R429.

If you think you can do apply this pattern in this PR, and do a follow-up refactor on the EthStream that'd be great.

yeah that makes sense, we can enforce that a stream is authenticated at the type level

gakonst · 2022-10-27T08:08:54Z

crates/net/eth-wire/src/p2pstream.rs

+#[pin_project]
+pub struct P2PStream<S> {


We may want to add examples on how to use these, and their various message types. For a follow-up.

mattsse

great progress.

some comments re Stream+Sink and codec.

mattsse · 2022-10-31T11:46:34Z

crates/net/eth-wire/src/p2pstream.rs

+/// An un-authenticated `P2PStream`. This is consumed and returns a [`P2PStream`] after the `Hello`
+/// handshake is completed.
+#[pin_project]
+pub struct UnauthedP2PStream<S> {
+    #[pin]
+    stream: P2PStream<S>,
+}


this is great.

This is essentially a future type that does the handshake.

mattsse · 2022-10-31T11:47:06Z

crates/net/eth-wire/src/p2pstream.rs

+{
+    /// Consumes the `UnauthedP2PStream` and returns a `P2PStream` after the `Hello` handshake is
+    /// completed.
+    pub async fn handshake(mut self, hello: HelloMessage) -> Result<P2PStream<S>, P2PStreamError> {


this also needs to return the contents of the received hello message.

tracking in #157

crates/net/eth-wire/src/p2pstream.rs

mattsse · 2022-10-31T11:59:05Z

crates/net/eth-wire/src/p2pstream.rs

+
+                let send_res = Pin::new(&mut this.inner).send(ping_bytes.into()).poll_unpin(cx)?;
+                ready!(send_res)


this is invalid usage of the Sink API here since this SinkExt::send creates a future that resolves when the item has been sent.

the underlying stream should be Framed, right? so we should use a codec for the stream I think.
But I think this could be tricky for the sending part since we need to handle the offset?

mattsse · 2022-10-31T12:04:24Z

crates/net/eth-wire/src/p2pstream.rs

+        // we should loop here to ensure we don't return Poll::Pending if we have a message to
+        // return behind any pings we need to respond to
+        while let Poll::Ready(res) = this.inner.as_mut().poll_next(cx) {
+            let mut bytes = match res {


is this guaranteed to be a framed message?

mattsse · 2022-10-31T12:07:06Z

crates/net/eth-wire/src/p2pstream.rs

+                let send_res = Pin::new(&mut this.inner).send(pong_bytes.into()).poll_unpin(cx)?;
+                ready!(send_res)


we need to use the actual sink api here which is always:

if poll_ready().is_ready() { start_send(msg) } else { // need to buffer until sink ready. }

see Sink trait docs.

tracking in #156

mattsse

this looks very good now, some parts are not quite complete yet, but I'd suggest merging this now and do smaller iterations moving forward? Because I think the stream-type interfaces are somewhat stable now.
we only really need Stream+Sinkand handshake function I believe.

wdyt @Rjected

Rjected · 2022-11-01T16:09:44Z

@mattsse sure, let me get tests passing with the handshake and we can merge. I'll make some issues for things left to do

* TODO: make it compile * TODO: test ping/pong/disconnect state machine * TODO: send subprotocol messages to stream * TODO: encode non-hello p2p messages as snappy encoding without using an encoder * TODO: create test comparing encoder to hand-written snappy encoding for ping, pong, disconnect messages

* restricts S to be Stream+Sink for P2PStream to implement Stream * start of a poll-based refactor

* add tests * TODO: make stream/sink types compatible * TODO: handshake message ids * TODO: inner poll fn * TODO: pinger interval * TODO: ethstream test * TODO: passthrough test

* it should be much easier to poll for pings and detect timeouts now

* change item produced by stream so it's compatible with EthStream * add note on pros/cons * shorten message sends in stream

* switch to snappy formatting for non-hello p2p messages

* remove check for `Hello` messages outside of the handshake because `P2PStream`s should assume messages sent in the sink are subprotocol messages, not `p2p` messages.

* disallow protocol versions other than v5

* todo: test shared capabilities * todo: determine how / when / why to support multiple capabilities * removes obsolete authed and offset fields

* should add protocols when necessary rather than name unsupported protocols

* add test for p2pstream over a passthrough codec which tests that peers agree on a single shared capability

Rjected · 2022-11-03T03:22:29Z

OK, I got tests to pass - we can probably merge this now, the TODOs / smaller suggestions can be converted to issues if that sounds good

gakonst

Really clean

gakonst · 2022-11-03T06:53:12Z

crates/net/eth-wire/src/p2pstream.rs

+use bytes::{Buf, Bytes, BytesMut};
+use futures::{ready, FutureExt, Sink, SinkExt, StreamExt};
+use pin_project::pin_project;
+use reth_primitives::H512 as PeerId;


Should we add PeerId as a re export in primitives and note why it's that size? It seems common enough

Yeah that's a good idea

mattsse reviewed Oct 21, 2022

View reviewed changes

onbjerg added C-enhancement New feature or request A-devp2p Related to the Ethereum P2P protocol labels Oct 21, 2022

gakonst reviewed Oct 21, 2022

View reviewed changes

crates/net/eth-wire/src/p2pstream.rs Show resolved Hide resolved

gakonst reviewed Oct 22, 2022

View reviewed changes

mattsse requested changes Oct 26, 2022

View reviewed changes

mattsse reviewed Oct 26, 2022

View reviewed changes

Rjected force-pushed the dan/p2p-stream branch 3 times, most recently from aed5093 to 78e41ab Compare October 27, 2022 01:37

gakonst reviewed Oct 27, 2022

View reviewed changes

mattsse requested changes Oct 31, 2022

View reviewed changes

mattsse reviewed Nov 1, 2022

View reviewed changes

Rjected force-pushed the dan/p2p-stream branch from 933bd73 to 31a6d22 Compare November 1, 2022 15:59

Rjected added 15 commits November 2, 2022 23:16

wip: p2pstream

81db48f

add comment for handshake timeout

4713cf0

temp allow some lint violations

55ac1bc

ignore unused_variables

9b26c4e

implement message handling in stream poll method

d8b19ab

* restricts S to be Stream+Sink for P2PStream to implement Stream * start of a poll-based refactor

impl Stream and Sink for P2PStream

ee653d4

* add tests * TODO: make stream/sink types compatible * TODO: handshake message ids * TODO: inner poll fn * TODO: pinger interval * TODO: ethstream test * TODO: passthrough test

create pingers and test

ac60d42

impl working timeout interval pinger

d07d4e0

* it should be much easier to poll for pings and detect timeouts now

use pinger in p2p stream

594a271

* change item produced by stream so it's compatible with EthStream * add note on pros/cons * shorten message sends in stream

improve errors and remove redundant methods

8442ae2

fix handshake

a2ae0d1

debugging printlns

b9298ca

fix encoding and decoding

b54481b

* switch to snappy formatting for non-hello p2p messages

cargo fmt

3a4bb84

Rjected added 13 commits November 2, 2022 23:16

perform handshake in ethstream over p2pstream test

2380a46

* remove check for `Hello` messages outside of the handshake because `P2PStream`s should assume messages sent in the sink are subprotocol messages, not `p2p` messages.

impl From<EthVersion> for CapabilityMessage

8e9d51f

remove printlns

eec5b8d

add total_message method to EthVersion

22359d3

decode Hello in handshake

220d055

* disallow protocol versions other than v5

Integrate snappy and implement message size limits

a213d50

document constants, move stream definition

1377ccb

fix missing hello message id

3684814

implement shared capabilities

3fe018d

* todo: test shared capabilities * todo: determine how / when / why to support multiple capabilities * removes obsolete authed and offset fields

add sink api TODOs

412c502

remove les

4d6ed82

* should add protocols when necessary rather than name unsupported protocols

fix snappy compression length

ae48b0d

* add test for p2pstream over a passthrough codec which tests that peers agree on a single shared capability

fix some clippy lints

2f2b250

Rjected force-pushed the dan/p2p-stream branch from 9a06fc6 to 2f2b250 Compare November 3, 2022 03:16

gakonst approved these changes Nov 3, 2022

View reviewed changes

gakonst merged commit 3c72a12 into main Nov 3, 2022

gakonst deleted the dan/p2p-stream branch November 3, 2022 07:01

This was referenced Nov 3, 2022

Fix Sink API usage in P2PStream #156

Closed

Tracking: P2P #64

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat(eth-wire): Implement `p2p` stream #114

feat(eth-wire): Implement `p2p` stream #114

Rjected commented Oct 21, 2022 •

edited

Loading

mattsse left a comment

gakonst Oct 22, 2022

gakonst Oct 22, 2022

mattsse left a comment

mattsse Oct 26, 2022

mattsse Oct 26, 2022

mattsse Oct 26, 2022

gakonst left a comment

gakonst Oct 27, 2022

Rjected Oct 27, 2022

gakonst Oct 27, 2022

Rjected Oct 27, 2022

gakonst Oct 27, 2022

mattsse left a comment

mattsse Oct 31, 2022

mattsse Oct 31, 2022

Rjected Nov 3, 2022

mattsse Oct 31, 2022

mattsse Oct 31, 2022

mattsse Oct 31, 2022

Rjected Nov 3, 2022

mattsse left a comment •

edited

Loading

Rjected commented Nov 1, 2022

Rjected commented Nov 3, 2022

gakonst left a comment

gakonst Nov 3, 2022

Rjected Nov 3, 2022

		/// Whether the `Hello` handshake has been completed
		authed: bool,


		let send_res = Pin::new(&mut this.inner).send(ping_bytes.into()).poll_unpin(cx)?;
		ready!(send_res)

feat(eth-wire): Implement p2p stream #114

feat(eth-wire): Implement p2p stream #114

Conversation

Rjected commented Oct 21, 2022 • edited Loading

mattsse left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mattsse left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

gakonst left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mattsse left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mattsse left a comment • edited Loading

Choose a reason for hiding this comment

Rjected commented Nov 1, 2022

Rjected commented Nov 3, 2022

gakonst left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

feat(eth-wire): Implement `p2p` stream #114

feat(eth-wire): Implement `p2p` stream #114

Rjected commented Oct 21, 2022 •

edited

Loading

mattsse left a comment •

edited

Loading