GH-818 Chunked Responses Review 1

.gitignore

@@ -4,6 +4,7 @@
 **/*.rs.bk
 .idea/azure/
 .idea/inspectionProfiles/Project_Default.xml
+.idea/copilot.data.migration.*
 
 ### Node
 node_modules

multinode_integration_tests/tests/connection_termination_test.rs

@@ -86,13 +86,12 @@ fn reported_server_drop() {
     let (_, _, lcp) = mock_node
         .wait_for_package(&masquerader, Duration::from_secs(2))
         .unwrap();
-    let (stream_key, return_route_id) =
-        context_from_request_lcp(lcp, real_node.main_cryptde_null().unwrap(), &exit_cryptde);
+    let stream_key = stream_key_from_request_lcp(lcp, &exit_cryptde);
 
     mock_node
         .transmit_package(
             mock_node.port_list()[0],
-            create_server_drop_report(&mock_node, &real_node, stream_key, return_route_id),
+            create_server_drop_report(&mock_node, &real_node, stream_key),
             &masquerader,
             real_node.main_public_key(),
             real_node.socket_addr(PortSelector::First),
@@ -115,7 +114,7 @@ fn actual_server_drop() {
     let server_port = find_free_port();
     let mut server = real_node.make_server(server_port);
     let masquerader = JsonMasquerader::new();
-    let (stream_key, return_route_id) = arbitrary_context();
+    let stream_key = arbitrary_stream_key();
     let index: u64 = 0;
     request_server_payload(
         index,
@@ -125,7 +124,6 @@ fn actual_server_drop() {
         &mut server,
         &masquerader,
         stream_key,
-        return_route_id,
     );
     let index: u64 = 1;
     request_server_payload(
@@ -136,7 +134,6 @@ fn actual_server_drop() {
         &mut server,
         &masquerader,
         stream_key,
-        return_route_id,
     );
 
     server.shutdown();
@@ -174,7 +171,6 @@ fn request_server_payload(
     server: &mut MASQNodeServer,
     masquerader: &JsonMasquerader,
     stream_key: StreamKey,
-    return_route_id: u32,
 ) {
     mock_node
         .transmit_package(
@@ -184,7 +180,6 @@ fn request_server_payload(
                 &mock_node,
                 &real_node,
                 stream_key,
-                return_route_id,
                 &server,
                 cluster.chain,
             ),
@@ -212,7 +207,7 @@ fn reported_client_drop() {
     let server_port = find_free_port();
     let mut server = real_node.make_server(server_port);
     let masquerader = JsonMasquerader::new();
-    let (stream_key, return_route_id) = arbitrary_context();
+    let stream_key = arbitrary_stream_key();
     let index: u64 = 0;
     mock_node
         .transmit_package(
@@ -222,7 +217,6 @@ fn reported_client_drop() {
                 &mock_node,
                 &real_node,
                 stream_key,
-                return_route_id,
                 &server,
                 cluster.chain,
             ),
@@ -240,7 +234,7 @@ fn reported_client_drop() {
     mock_node
         .transmit_package(
             mock_node.port_list()[0],
-            create_client_drop_report(&mock_node, &real_node, stream_key, return_route_id),
+            create_client_drop_report(&mock_node, &real_node, stream_key),
             &masquerader,
             real_node.main_public_key(),
             real_node.socket_addr(PortSelector::First),
@@ -322,40 +316,32 @@ fn full_neighbor(one: &mut NodeRecord, another: &mut NodeRecord) {
         .unwrap();
 }
 
-fn context_from_request_lcp(
-    lcp: LiveCoresPackage,
-    originating_cryptde: &dyn CryptDE,
-    exit_cryptde: &dyn CryptDE,
-) -> (StreamKey, u32) {
+fn stream_key_from_request_lcp(lcp: LiveCoresPackage, exit_cryptde: &dyn CryptDE) -> StreamKey {
     let payload = match decodex::<MessageType>(exit_cryptde, &lcp.payload).unwrap() {
         MessageType::ClientRequest(vd) => vd
             .extract(&node_lib::sub_lib::migrations::client_request_payload::MIGRATIONS)
             .unwrap(),
         mt => panic!("Unexpected: {:?}", mt),
     };
     let stream_key = payload.stream_key;
-    let return_route_id = decodex::<u32>(originating_cryptde, &lcp.route.hops[6]).unwrap();
-    (stream_key, return_route_id)
+    stream_key
 }
 
-fn arbitrary_context() -> (StreamKey, u32) {
-    (
-        StreamKey::make_meaningful_stream_key("arbitrary_context"),
-        12345678,
-    )
+fn arbitrary_stream_key() -> StreamKey {
+    StreamKey::make_meaningful_stream_key("arbitrary_context")
 }
 
 fn create_request_icp(
     index: u64,
     originating_node: &MASQMockNode,
     exit_node: &MASQRealNode,
     stream_key: StreamKey,
-    return_route_id: u32,
     server: &MASQNodeServer,
     chain: Chain,
 ) -> IncipientCoresPackage {
+    let originating_main_cryptde = originating_node.main_cryptde_null().unwrap();
     IncipientCoresPackage::new(
-        originating_node.main_cryptde_null().unwrap(),
+        originating_main_cryptde,
         Route::round_trip(
             RouteSegment::new(
                 vec![
@@ -371,12 +357,12 @@ fn create_request_icp(
                 ],
                 Component::ProxyServer,
             ),
-            originating_node.main_cryptde_null().unwrap(),
+            originating_main_cryptde,
             originating_node.consuming_wallet(),
-            return_route_id,
             Some(chain.rec().contract),
         )
-        .unwrap(),
+        .unwrap()
+        .set_return_route_id(originating_main_cryptde, 0),
         MessageType::ClientRequest(VersionedData::new(
             &node_lib::sub_lib::migrations::client_request_payload::MIGRATIONS,
             &ClientRequestPayload_0v1 {
@@ -400,8 +386,9 @@ fn create_meaningless_icp(
     let socket_addr = SocketAddr::from_str("3.2.1.0:7654").unwrap();
     let stream_key =
         StreamKey::make_meaningful_stream_key("Chancellor on brink of second bailout for banks");
+    let main_cryptde = originating_node.main_cryptde_null().unwrap();
     IncipientCoresPackage::new(
-        originating_node.main_cryptde_null().unwrap(),
+        main_cryptde,
         Route::round_trip(
             RouteSegment::new(
                 vec![
@@ -417,9 +404,8 @@ fn create_meaningless_icp(
                 ],
                 Component::ProxyServer,
             ),
-            originating_node.main_cryptde_null().unwrap(),
+            main_cryptde,
             originating_node.consuming_wallet(),
-            1357,
             Some(TEST_DEFAULT_MULTINODE_CHAIN.rec().contract),
         )
         .unwrap(),
@@ -428,7 +414,7 @@ fn create_meaningless_icp(
             &ClientRequestPayload_0v1 {
                 stream_key,
                 sequenced_packet: SequencedPacket::new(Vec::from(HTTP_REQUEST), 0, false),
-                target_hostname: Some(format!("nowhere.com")),
+                target_hostname: Some("nowhere.com".to_string()),
                 target_port: socket_addr.port(),
                 protocol: ProxyProtocol::HTTP,
                 originator_public_key: originating_node.main_public_key().clone(),
@@ -443,8 +429,9 @@ fn create_server_drop_report(
     exit_node: &MASQMockNode,
     originating_node: &MASQRealNode,
     stream_key: StreamKey,
-    return_route_id: u32,
 ) -> IncipientCoresPackage {
+    let exit_main_cryptde = exit_node.main_cryptde_null().unwrap();
+    let originating_main_cryptde = originating_node.main_cryptde_null().unwrap();
     let mut route = Route::round_trip(
         RouteSegment::new(
             vec![
@@ -460,15 +447,13 @@ fn create_server_drop_report(
             ],
             Component::ProxyServer,
         ),
-        originating_node.main_cryptde_null().unwrap(),
+        originating_main_cryptde,
         originating_node.consuming_wallet(),
-        return_route_id,
         Some(TEST_DEFAULT_MULTINODE_CHAIN.rec().contract),
     )
-    .unwrap();
-    route
-        .shift(originating_node.main_cryptde_null().unwrap())
-        .unwrap();
+    .unwrap()
+    .set_return_route_id(originating_main_cryptde, 0);
+    route.shift(originating_main_cryptde).unwrap();
     let payload = MessageType::ClientResponse(VersionedData::new(
         &node_lib::sub_lib::migrations::client_response_payload::MIGRATIONS,
         &ClientResponsePayload_0v1 {
@@ -478,7 +463,7 @@ fn create_server_drop_report(
     ));
 
     IncipientCoresPackage::new(
-        exit_node.main_cryptde_null().unwrap(),
+        exit_main_cryptde,
         route,
         payload,
         originating_node.alias_public_key(),
@@ -490,8 +475,8 @@ fn create_client_drop_report(
     originating_node: &MASQMockNode,
     exit_node: &MASQRealNode,
     stream_key: StreamKey,
-    return_route_id: u32,
 ) -> IncipientCoresPackage {
+    let originating_main_cryptde = originating_node.main_cryptde_null().unwrap();
     let route = Route::round_trip(
         RouteSegment::new(
             vec![
@@ -507,12 +492,12 @@ fn create_client_drop_report(
             ],
             Component::ProxyServer,
         ),
-        originating_node.main_cryptde_null().unwrap(),
+        originating_main_cryptde,
         originating_node.consuming_wallet(),
-        return_route_id,
         Some(TEST_DEFAULT_MULTINODE_CHAIN.rec().contract),
     )
-    .unwrap();
+    .unwrap()
+    .set_return_route_id(originating_main_cryptde, 0);
     let payload = MessageType::ClientRequest(VersionedData::new(
         &node_lib::sub_lib::migrations::client_request_payload::MIGRATIONS,
         &ClientRequestPayload_0v1 {
@@ -526,7 +511,7 @@ fn create_client_drop_report(
     ));
 
     IncipientCoresPackage::new(
-        originating_node.main_cryptde_null().unwrap(),
+        originating_main_cryptde,
         route,
         payload,
         exit_node.main_public_key(),

multinode_integration_tests/tests/self_test.rs

@@ -14,6 +14,7 @@ use node_lib::sub_lib::dispatcher::Component;
 use node_lib::sub_lib::hopper::IncipientCoresPackage;
 use node_lib::sub_lib::route::Route;
 use node_lib::sub_lib::route::RouteSegment;
+use node_lib::sub_lib::stream_key::StreamKey;
 use node_lib::test_utils::{main_cryptde, make_meaningless_message_type, make_paying_wallet};
 use std::collections::HashSet;
 use std::io::ErrorKind;
@@ -66,6 +67,7 @@ fn server_relays_cores_package() {
     let masquerader = JsonMasquerader::new();
     let server = MASQCoresServer::new(cluster.chain);
     let cryptde = server.main_cryptde();
+    let stream_key = StreamKey::make_meaningless_stream_key();
     let mut client = MASQCoresClient::new(server.local_addr(), cryptde);
     let mut route = Route::one_way(
         RouteSegment::new(
@@ -80,7 +82,7 @@ fn server_relays_cores_package() {
     let incipient = IncipientCoresPackage::new(
         cryptde,
         route.clone(),
-        make_meaningless_message_type(),
+        make_meaningless_message_type(stream_key),
         &cryptde.public_key(),
     )
     .unwrap();
@@ -97,7 +99,7 @@ fn server_relays_cores_package() {
 
     route.shift(cryptde).unwrap();
     assert_eq!(expired.remaining_route, route);
-    assert_eq!(expired.payload, make_meaningless_message_type());
+    assert_eq!(expired.payload, make_meaningless_message_type(stream_key));
 }
 
 #[test]
@@ -109,6 +111,7 @@ fn one_mock_node_talks_to_another() {
     let mock_node_1 = cluster.get_mock_node_by_name("mock_node_1").unwrap();
     let mock_node_2 = cluster.get_mock_node_by_name("mock_node_2").unwrap();
     let cryptde = main_cryptde();
+    let stream_key = StreamKey::make_meaningless_stream_key();
     let route = Route::one_way(
         RouteSegment::new(
             vec![
@@ -125,7 +128,7 @@ fn one_mock_node_talks_to_another() {
     let incipient_cores_package = IncipientCoresPackage::new(
         cryptde,
         route,
-        make_meaningless_message_type(),
+        make_meaningless_message_type(stream_key),
         &mock_node_2.main_public_key(),
     )
     .unwrap();
@@ -154,7 +157,7 @@ fn one_mock_node_talks_to_another() {
     assert_eq!(package_to, mock_node_2.socket_addr(PortSelector::First));
     assert_eq!(
         expired_cores_package.payload,
-        make_meaningless_message_type()
+        make_meaningless_message_type(stream_key)
     );
 }
 

node/src/hopper/consuming_service.rs

@@ -138,6 +138,7 @@ mod tests {
     use crate::sub_lib::node_addr::NodeAddr;
     use crate::sub_lib::route::Route;
     use crate::sub_lib::route::RouteSegment;
+    use crate::sub_lib::stream_key::StreamKey;
     use crate::test_utils::recorder::make_recorder;
     use crate::test_utils::recorder::peer_actors_builder;
     use crate::test_utils::{main_cryptde, make_meaningless_message_type, make_paying_wallet};
@@ -158,7 +159,7 @@ mod tests {
             main_cryptde(),
             &target_key,
             &target_node_addr,
-            make_meaningless_message_type(),
+            make_meaningless_message_type(StreamKey::make_meaningless_stream_key()),
         )
         .unwrap();
         let system = System::new("");
@@ -228,7 +229,7 @@ mod tests {
             Some(TEST_DEFAULT_CHAIN.rec().contract),
         )
         .unwrap();
-        let payload = make_meaningless_message_type();
+        let payload = make_meaningless_message_type(StreamKey::make_meaningless_stream_key());
         let incipient_cores_package =
             IncipientCoresPackage::new(cryptde, route.clone(), payload, &destination_key).unwrap();
         let system = System::new("converts_incipient_message_to_live_and_sends_to_dispatcher");
@@ -275,7 +276,7 @@ mod tests {
             Some(TEST_DEFAULT_CHAIN.rec().contract),
         )
         .unwrap();
-        let payload = make_meaningless_message_type();
+        let payload = make_meaningless_message_type(StreamKey::make_meaningless_stream_key());
         let incipient_cores_package =
             IncipientCoresPackage::new(cryptde, route.clone(), payload, &destination_key).unwrap();
         let system = System::new("consume_sends_zero_hop_incipient_directly_to_hopper");
@@ -326,7 +327,7 @@ mod tests {
             IncipientCoresPackage::new(
                 main_cryptde(),
                 Route { hops: vec![] },
-                make_meaningless_message_type(),
+                make_meaningless_message_type(StreamKey::make_meaningless_stream_key()),
                 &PublicKey::new(&[1, 2]),
             )
             .unwrap(),