is_pending_outbound fixes

[whereistejas]

is_pending_outbound returns true if the connection
to the mentioned peer hasn't been established, yet.

Closes issue: #1885

Signed-off-by: Tejas Sanap [email protected]

[whereistejas]

Hi @mxinden , this PR is ready to be merged. I would really appreciate a review. Thanks!

[mxinden]

I think you missed a small edge case here, right?

[mxinden]

Suggested change

	// check if request id is exists in already established connections
	// Check if request is already sent on established connection.

[mxinden]

Say you have two requests R1 and R2 all targeting peer P1. Say R1 is done and a result is returned to the upper layer. Say R2 is still in self.pending_outbound_requests.

If you call is_pending_outbound(P1, R1) you will get true even though R1 is already finished. You need to check the request_id in the RequestProtocol in pending_outbound_request.

[whereistejas]

Should we include this scenario in the test program, too? I think, we should.

[whereistejas]

I added a few more assert statements in ping.rs to cover this scenario. Thank you for bringing up this. :)

[whereistejas]

I'm trying to replicate this scenario. For some reason, I'm not able to do so.

I added a few print statements, to log stuff to the console, in the is_pending_outbound method.

let est_conn = self.connected.get(peer)
    .map(|cs| cs.iter().any(|c| c.pending_inbound_responses.contains(request_id)))
    .unwrap_or(false);

let pen_conn = self.pending_outbound_requests.contains_key(peer);

println!("request_id (from is_pending_outbound): {}", *request_id);
println!("est_conn || pen_conn: {} || {}", est_conn, pen_conn);
// print all the request ids in pending_outbound_requests for a given peer id
for value in self.pending_outbound_requests.get(peer) {
    for item in value {
        println!("request_id (from pending_outbound_requests): {}", item.request_id);
    }
}

est_conn || pen_conn

And, then I called the is_pending_outbound method in ping_protocol method in ping.rs as follows:

let mut req_id = swarm2.send_request(&peer1_id, ping.clone());
let mut req_id_org = req_id.clone();
assert!(swarm2.is_pending_outbound(&peer1_id, &req_id));

let peer2 = async move {
    let mut count = 0;
    let addr = rx.next().await.unwrap();
    swarm2.add_address(&peer1_id, addr.clone());
    let mut req_id = swarm2.send_request(&peer1_id, ping.clone());
    let mut req_id_org = req_id.clone();
    assert!(swarm2.is_pending_outbound(&peer1_id, &req_id));

    //let mut req_id2 = swarm2.send_request(&peer1_id, ping.clone());
    loop {
        match swarm2.next().await {
            RequestResponseEvent::Message {
                peer,
                message: RequestResponseMessage::Response { request_id, response }
            } => {
                count += 1;
                assert_eq!(&response, &expected_pong);
                assert_eq!(&peer, &peer1_id);
                assert_eq!(req_id, request_id);

                // check if is_pending_outbound returns false for a finished request when we have already issued a new one
                if req_id != req_id_org {
                    assert!(swarm2.is_pending_outbound(&peer1_id, &req_id_org) == false);
                }
                
                req_id_org = req_id.clone();
                if count >= num_pings {
                   break 
                } else {
                    req_id = swarm2.send_request(&peer1_id, ping.clone());
                    assert!(swarm2.is_pending_outbound(&peer1_id, &req_id));
                }
            },
            e => panic!("Peer2: Unexpected event: {:?}", e)
        }
    }

On running this code, I get the following output:

request_id (from is_pending_outbound): 2
--> est_conn || pen_conn: true || false

request_id (from is_pending_outbound): 1
--> est_conn || pen_conn: false || false

request_id (from is_pending_outbound): 3
--> est_conn || pen_conn: true || false

request_id (from is_pending_outbound): 2
--> est_conn || pen_conn: false || false

request_id (from is_pending_outbound): 4
--> est_conn || pen_conn: true || false

request_id (from is_pending_outbound): 3
--> est_conn || pen_conn: false || false

So, in every iteration of the loop, I'm checking if the previous request is outbound and it always returns a false statement. So, I guess, the edge case you mentioned doesn't occur? I highly suspect that I am doing something wrong.

[mxinden]

Suggested change

	// check if peer exists in pending connections
	// Check if request is still pending to be send.

[whereistejas]

hi @mxinden, could you give me some pointers on how to come up with a test for this scenario?

[whereistejas]

Hi @mxinden ,
I have removed the incorrect tests I had added to ping.rs. I think this PR is ready to be merged. I will create a new issue for the tests and work on it, separately.

Also, if you are looking for which commit to review then it is this:

1. Handled edge case.
2. Renamed variables for request response type.

[romanb]

Looks good to me. Thanks!

[mxinden]

Thanks for fixing this. 👍 for tackling the additional unit test in a follow up pull request.

[whereistejas]

Thanks for fixing this. for tackling the additional unit test in a follow up pull request.

Hi @mxinden, so we are done? Will the PR get merged on its own? This is my first time doing this, so I'm not aware of the processes. :)

[mxinden]

For the record, I pushed 25ea8b6 updating changelogs and cargo tomls.

[mxinden]

hi @mxinden, could you give me some pointers on how to come up with a test for this scenario?

Instead of extending an existing test, I would suggest creating a new one, only including a very minimal setup. Is the pseudo code below of some help @whereistejas?

diff --git a/protocols/request-response/tests/ping.rs b/protocols/request-response/tests/ping.rs
index b03931cd..5622fb06 100644
--- a/protocols/request-response/tests/ping.rs
+++ b/protocols/request-response/tests/ping.rs
@@ -274,6 +274,28 @@ fn ping_protocol_throttled() {
     pool.run_until(peer2);
 }
 
+#[test]
+fn is_pending_outbound() {
+    let ping = Ping("ping".to_string().into_bytes());
+    let offline_peer = PeerId::random();
+
+    let protocols = iter::once((PingProtocol(), ProtocolSupport::Full));
+    let cfg = RequestResponseConfig::default();
+
+    let (peer1_id, trans) = mk_transport();
+    let ping_proto1 = RequestResponse::throttled(PingCodec(), protocols.clone(), cfg.clone());
+    let mut swarm1 = Swarm::new(trans, ping_proto1, peer1_id.clone());
+
+    let request_id_1 = swarm1.send_request(offline_peer, ping.clone());
+
+    // Poll swarm until request 1 is reported as failed.
+
+    let request_id_2 = swarm1.send_request(offline_peer, ping.clone());
+
+    assert!(!swarm1.is_pending_outbound(offline_peers, request_id_1));
+    assert!(swarm1.is_pending_outbound(offline_peers, request_id_2));
+}
+