Merge branch 'fix/duplicate-client-messages' into refactor/do-not-wait-for-update-call

Author: ilbertt

.github/workflows/rust-tests.yml

@@ -24,7 +24,7 @@ jobs:
         run: cargo fmt -- --check
 
       - name: Cargo clippy
-        run: cargo clippy --all --all-features -- -Dwarnings
+        run: cargo clippy --all-targets --all-features --workspace -- -Dwarnings
 
       - name: Run unit tests
         run: ./scripts/unit_test.sh

src/gateway-state/src/lib.rs

@@ -399,7 +399,6 @@ mod tests {
 
         {
             let gateway_state = gateway_state.clone();
-            let canister_id = canister_id.clone();
             thread::spawn(move || loop {
                 gateway_state.remove_canister_if_empty(canister_id);
             });
@@ -461,7 +460,6 @@ mod tests {
 
         {
             let gateway_state = gateway_state.clone();
-            let canister_id = canister_id.clone();
             thread::spawn(move || {
                 for i in 0.. {
                     let client_key = ClientKey::new(Principal::anonymous(), i);
@@ -516,7 +514,6 @@ mod tests {
 
         {
             let gateway_state = gateway_state.clone();
-            let canister_id = canister_id.clone();
             thread::spawn(move || {
                 for i in 0.. {
                     let client_key = ClientKey::new(Principal::anonymous(), i);

src/ic-websocket-gateway/src/client_session.rs

@@ -165,8 +165,7 @@ impl<S: AsyncRead + AsyncWrite + Unpin> ClientSession<S> {
         &mut self,
         client_update: Result<Message, Error>,
     ) -> Result<(), IcWsError> {
-        // upon receiving an update while the session is Setup, check if it is due to
-        // the client disconnecting without a closing handshake
+        // upon receiving an update, check if the message is valid and is not a close message
         let ws_message = self.check_client_update(client_update)?;
         if ws_message.is_close() && !self.session_state.is_closed() {
             trace!("Client disconnected while in {} state", self.session_state);
@@ -179,7 +178,7 @@ impl<S: AsyncRead + AsyncWrite + Unpin> ClientSession<S> {
                 // upon receiving a message while the session is Init, check if the message is valid
                 // if not return an error, otherwise set the session state to Setup
                 // if multiple messages are received while in Init state, 'handle_setup_transition' will
-                // return an error as the client shall not send more than one message while in Init state
+                // ignore them as the client shall not send more than one message while in Init state
                 let setup_state = self.handle_setup_transition(ws_message).await?;
                 self.session_state = setup_state;
             },
@@ -244,7 +243,7 @@ impl<S: AsyncRead + AsyncWrite + Unpin> ClientSession<S> {
                         // this implies a bug in the WS Gateway
                         error!("Received canister message while in Closed state");
                         Err(IcWsError::IcWsProtocol(String::from(
-                            "Poller shall not be able tosend messages while the session is in Closed state",
+                            "Poller shall not be able to send messages while the session is in Closed state",
                         )))
                     },
                 }
@@ -290,17 +289,14 @@ impl<S: AsyncRead + AsyncWrite + Unpin> ClientSession<S> {
             Ok((client_key, canister_id)) => {
                 // replace the field with the canister_id received in the first envelope
                 // this shall not be updated anymore
-                // if canister_id is already set in the struct, we return an error as inspect_ic_ws_open_message shall only be called once
+                // if canister_id is already set in the struct, we log a warning as inspect_ic_ws_open_message shall only be called once
                 if self.canister_id.replace(canister_id).is_some()
                     || self.client_key.replace(client_key.clone()).is_some()
                 {
                     // if the canister_id or client_key field was already set,
                     // it means that the client sent the WS open message twice,
                     // which it shall not do
-                    // therefore, return an error
-                    return Err(IcWsError::IcWsProtocol(String::from(
-                        "canister_id or client_key field was set twice",
-                    )));
+                    warn!("canister_id or client_key field was set twice");
                 }
                 trace!("Validated WS open message");
 
@@ -370,7 +366,7 @@ impl<S: AsyncRead + AsyncWrite + Unpin> ClientSession<S> {
             // relay the envelope to the IC
             self.relay_envelope_to_canister(serialized_envelope, canister_id);
 
-            // there is no need to relay the response back to the client as the response to a request to the /call enpoint is not certified by the canister
+            // there is no need to relay the response back to the client as the response to a request to the /call endpoint is not certified by the canister
             // and therefore could be manufactured by the gateway
 
             trace!("Relayed client message to canister");

src/ic-websocket-gateway/src/tests/canister_poller.rs

@@ -1,3 +1,10 @@
+// mockito::Server is behind a SYNC mutex so that only one test at the same time can access it
+// otherwise, as async tests are run on multiple threads, the mock response of one test might overwrite
+// the mock response of another, causing the test to fail
+// acquiring the mutex and the beginning of each test and dropping the guard only at the end,
+// ensures that only one test at the time can set the mock response
+// this enables running the tests without specifying each time "-- --test-threads=1"
+#[allow(clippy::await_holding_lock)]
 #[cfg(test)]
 mod test {
     use candid::Principal;
@@ -86,14 +93,12 @@ mod test {
     }
 
     lazy_static! {
-        // mockito::Server is behind a SYNC mutex so that only one test at the same time can access it
-        // otherwise, as async tests are run on multiple threads, the mock response of one test might overwrite
-        // the mock response of another, causing the test to fail
-        // acquiring the mutex and the beginning of each test and dropping the guard only at the end,
-        // ensures that only one test at the time can set the mock response
-        // this enables running the tests without specifying each time "-- --test-threads=1"
-        static ref MOCK_SERVER: Arc<Mutex<mockito::Server>> =
-            Arc::new(Mutex::new(mockito::Server::new_with_opts(mockito::ServerOpts { port: 51558, ..Default::default() })));
+        static ref MOCK_SERVER: Arc<Mutex<mockito::Server>> = Arc::new(Mutex::new(
+            mockito::Server::new_with_opts(mockito::ServerOpts {
+                port: 51558,
+                ..Default::default()
+            })
+        ));
     }
 
     fn create_poller(
@@ -332,7 +337,7 @@ mod test {
             .mock("GET", path)
             .with_chunked_body(|w| {
                 thread::sleep(Duration::from_millis(POLLING_TIMEOUT_MS + 10));
-                w.write_all(&vec![])
+                w.write_all(&[])
             })
             .expect(2)
             .create_async()