Make connection_id optional in ClientRequest::Send (#1798)

clients/native/examples/websocket_binarysend.rs

@@ -43,7 +43,7 @@ async fn send_file_with_reply() {
         recipient,
         message: read_data,
         with_reply_surb: true,
-        connection_id: 0,
+        connection_id: Some(0),
     };
 
     println!("sending content of 'dummy_file' over the mix network...");
@@ -92,7 +92,7 @@ async fn send_file_without_reply() {
         recipient,
         message: read_data,
         with_reply_surb: false,
-        connection_id: 0,
+        connection_id: Some(0),
     };
 
     println!("sending content of 'dummy_file' over the mix network...");

clients/native/src/websocket/handler.rs

@@ -92,26 +92,42 @@ impl Handler {
         recipient: &Recipient,
         message: Vec<u8>,
         with_reply_surb: bool,
-        connection_id: u64,
+        connection_id: Option<u64>,
     ) -> Option<ServerResponse> {
+        // We map the absence of a connection id as going into the general lane.
+        let lane = connection_id.map_or(TransmissionLane::General, |id| {
+            TransmissionLane::ConnectionId(id)
+        });
+
         // the ack control is now responsible for chunking, etc.
-        let lane = TransmissionLane::ConnectionId(connection_id);
         let input_msg = InputMessage::new_fresh(*recipient, message, with_reply_surb, lane);
         self.msg_input
             .send(input_msg)
             .await
             .expect("InputMessageReceiver has stopped receiving!");
 
+        // Only reply back with a `LaneQueueLength` if the sender providided a connection id
+        let connection_id = match lane {
+            TransmissionLane::General
+            | TransmissionLane::Reply
+            | TransmissionLane::Retransmission
+            | TransmissionLane::Control => return None,
+            TransmissionLane::ConnectionId(id) => id,
+        };
+
         // on receiving a send, we reply back the current lane queue length for that connection id.
         // Note that this does _NOT_ take into account the packets that have been received but not
         // yet reach `OutQueueControl`, so it might be a tad low.
-        if let Ok(lane_queue_lengths) = self.lane_queue_lengths.lock() {
-            let queue_length = lane_queue_lengths.get(&lane).unwrap_or(0);
-            return Some(ServerResponse::LaneQueueLength(connection_id, queue_length));
-        }
+        let Ok(lane_queue_lengths) = self.lane_queue_lengths.lock() else {
+            log::warn!(
+                "Failed to get the lane queue length lock, \
+                not responding back with the current queue length"
+            );
+            return None;
+        };
 
-        log::warn!("Failed to get the lane queue length lock, not responding back with the current queue length");
-        None
+        let queue_length = lane_queue_lengths.get(&lane).unwrap_or(0);
+        Some(ServerResponse::LaneQueueLength(connection_id, queue_length))
     }
 
     async fn handle_reply(
@@ -120,7 +136,13 @@ impl Handler {
         message: Vec<u8>,
     ) -> Option<ServerResponse> {
         if message.len() > ReplySurb::max_msg_len(Default::default()) {
-            return Some(ServerResponse::new_error(format!("too long message to put inside a reply SURB. Received: {} bytes and maximum is {} bytes", message.len(), ReplySurb::max_msg_len(Default::default()))));
+            return Some(
+                ServerResponse::new_error(
+                    format!(
+                        "too long message to put inside a reply SURB. Received: {} bytes and maximum is {} bytes",
+                        message.len(), ReplySurb::max_msg_len(Default::default()))
+                    )
+                );
         }
 
         let input_msg = InputMessage::new_reply(reply_surb, message);

clients/native/websocket-requests/src/requests.rs

@@ -34,7 +34,7 @@ pub enum ClientRequest {
         message: Vec<u8>,
         // Perhaps we could change it to a number to indicate how many reply_SURBs we want to include?
         with_reply_surb: bool,
-        connection_id: u64,
+        connection_id: Option<u64>,
     },
     Reply {
         message: Vec<u8>,
@@ -53,10 +53,10 @@ impl ClientRequest {
         recipient: Recipient,
         data: Vec<u8>,
         with_reply_surb: bool,
-        connection_id: u64,
+        connection_id: Option<u64>,
     ) -> Vec<u8> {
         let data_len_bytes = (data.len() as u64).to_be_bytes();
-        let conn_id_bytes = connection_id.to_be_bytes();
+        let conn_id_bytes = connection_id.unwrap_or(0).to_be_bytes();
         std::iter::once(SEND_REQUEST_TAG)
             .chain(std::iter::once(with_reply_surb as u8))
             .chain(recipient.to_bytes().iter().cloned()) // will not be length prefixed because the length is constant
@@ -106,6 +106,11 @@ impl ClientRequest {
         connection_id_bytes
             .copy_from_slice(&b[2 + Recipient::LEN..2 + Recipient::LEN + size_of::<u64>()]);
         let connection_id = u64::from_be_bytes(connection_id_bytes);
+        let connection_id = if connection_id == 0 {
+            None
+        } else {
+            Some(connection_id)
+        };
 
         let data_len_bytes =
             &b[2 + Recipient::LEN + size_of::<u64>()..2 + Recipient::LEN + 2 * size_of::<u64>()];
@@ -350,7 +355,7 @@ mod tests {
             recipient,
             message: b"foomp".to_vec(),
             with_reply_surb: false,
-            connection_id: 42,
+            connection_id: Some(42),
         };
 
         let bytes = send_request_no_surb.serialize();
@@ -365,7 +370,7 @@ mod tests {
                 assert_eq!(recipient.to_string(), recipient_string);
                 assert_eq!(message, b"foomp".to_vec());
                 assert!(!with_reply_surb);
-                assert_eq!(connection_id, 42)
+                assert_eq!(connection_id, Some(42))
             }
             _ => unreachable!(),
         }
@@ -374,7 +379,7 @@ mod tests {
             recipient,
             message: b"foomp".to_vec(),
             with_reply_surb: true,
-            connection_id: 213,
+            connection_id: None,
         };
 
         let bytes = send_request_surb.serialize();
@@ -389,7 +394,7 @@ mod tests {
                 assert_eq!(recipient.to_string(), recipient_string);
                 assert_eq!(message, b"foomp".to_vec());
                 assert!(with_reply_surb);
-                assert_eq!(connection_id, 213)
+                assert_eq!(connection_id, None)
             }
             _ => unreachable!(),
         }

clients/native/websocket-requests/src/text.rs

@@ -20,7 +20,7 @@ pub(super) enum ClientRequestText {
         message: String,
         recipient: String,
         with_reply_surb: bool,
-        connection_id: u64,
+        connection_id: Option<u64>,
     },
     SelfAddress,
     #[serde(rename_all = "camelCase")]

service-providers/network-requester/src/core.rs

@@ -102,7 +102,7 @@ impl ServiceProvider {
                             recipient: return_address,
                             message: msg.into_bytes(),
                             with_reply_surb: false,
-                            connection_id: conn_id,
+                            connection_id: Some(conn_id),
                         };
 
                         let message = Message::Binary(response_message.serialize());