Fix concurrency with async rwlock, split command validation (#1081)

sdk/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "iggy"
-version = "0.5.0"
+version = "0.5.1"
 description = "Iggy is the persistent message streaming platform written in Rust, supporting QUIC, TCP and HTTP transport protocols, capable of processing millions of messages per second."
 edition = "2021"
 license = "MIT"

sdk/src/command.rs

@@ -1,8 +1,9 @@
 use crate::bytes_serializable::BytesSerializable;
 use crate::error::IggyError;
+use crate::validatable::Validatable;
 use std::fmt::Display;
 
-pub trait Command: BytesSerializable + Send + Sync + Display {
+pub trait Command: BytesSerializable + Validatable<IggyError> + Send + Sync + Display {
     fn code(&self) -> u32;
 }
 

sdk/src/consumer_groups/create_consumer_group.rs

@@ -104,7 +104,6 @@ impl BytesSerializable for CreateConsumerGroup {
             group_id,
             name,
         };
-        command.validate()?;
         Ok(command)
     }
 }

sdk/src/consumer_groups/delete_consumer_group.rs

@@ -67,7 +67,6 @@ impl BytesSerializable for DeleteConsumerGroup {
             topic_id,
             group_id,
         };
-        command.validate()?;
         Ok(command)
     }
 }

sdk/src/consumer_groups/get_consumer_group.rs

@@ -67,7 +67,6 @@ impl BytesSerializable for GetConsumerGroup {
             topic_id,
             group_id,
         };
-        command.validate()?;
         Ok(command)
     }
 }

sdk/src/consumer_groups/get_consumer_groups.rs

@@ -56,7 +56,6 @@ impl BytesSerializable for GetConsumerGroups {
             stream_id,
             topic_id,
         };
-        command.validate()?;
         Ok(command)
     }
 }

sdk/src/consumer_groups/join_consumer_group.rs

@@ -67,7 +67,6 @@ impl BytesSerializable for JoinConsumerGroup {
             topic_id,
             group_id,
         };
-        command.validate()?;
         Ok(command)
     }
 }

sdk/src/consumer_groups/leave_consumer_group.rs

@@ -67,7 +67,6 @@ impl BytesSerializable for LeaveConsumerGroup {
             topic_id,
             group_id,
         };
-        command.validate()?;
         Ok(command)
     }
 }

sdk/src/consumer_offsets/get_consumer_offset.rs

@@ -105,7 +105,6 @@ impl BytesSerializable for GetConsumerOffset {
             topic_id,
             partition_id,
         };
-        command.validate()?;
         Ok(command)
     }
 }

sdk/src/consumer_offsets/store_consumer_offset.rs

@@ -107,7 +107,6 @@ impl BytesSerializable for StoreConsumerOffset {
             partition_id,
             offset,
         };
-        command.validate()?;
         Ok(command)
     }
 }

sdk/src/messages/poll_messages.rs

@@ -291,7 +291,6 @@ impl BytesSerializable for PollMessages {
             count,
             auto_commit,
         };
-        command.validate()?;
         Ok(command)
     }
 }

sdk/src/messages/send_messages.rs

@@ -478,7 +478,6 @@ impl BytesSerializable for SendMessages {
             partitioning: key,
             messages,
         };
-        command.validate()?;
         Ok(command)
     }
 }

sdk/src/partitions/create_partitions.rs

@@ -78,7 +78,6 @@ impl BytesSerializable for CreatePartitions {
             topic_id,
             partitions_count,
         };
-        command.validate()?;
         Ok(command)
     }
 }

sdk/src/partitions/delete_partitions.rs

@@ -78,7 +78,6 @@ impl BytesSerializable for DeletePartitions {
             topic_id,
             partitions_count,
         };
-        command.validate()?;
         Ok(command)
     }
 }

sdk/src/personal_access_tokens/create_personal_access_token.rs

@@ -80,7 +80,6 @@ impl BytesSerializable for CreatePersonalAccessToken {
         let expiry: IggyExpiry = expiry.into();
 
         let command = CreatePersonalAccessToken { name, expiry };
-        command.validate()?;
         Ok(command)
     }
 }

sdk/src/personal_access_tokens/delete_personal_access_token.rs

@@ -70,7 +70,6 @@ impl BytesSerializable for DeletePersonalAccessToken {
         }
 
         let command = DeletePersonalAccessToken { name };
-        command.validate()?;
         Ok(command)
     }
 }

sdk/src/personal_access_tokens/get_personal_access_tokens.rs

@@ -28,13 +28,11 @@ impl BytesSerializable for GetPersonalAccessTokens {
         Bytes::new()
     }
 
-    fn from_bytes(bytes: Bytes) -> std::result::Result<GetPersonalAccessTokens, IggyError> {
+    fn from_bytes(bytes: Bytes) -> Result<GetPersonalAccessTokens, IggyError> {
         if !bytes.is_empty() {
             return Err(IggyError::InvalidCommand);
         }
 
-        let command = GetPersonalAccessTokens {};
-        command.validate()?;
         Ok(GetPersonalAccessTokens {})
     }
 }

sdk/src/personal_access_tokens/login_with_personal_access_token.rs

@@ -62,7 +62,6 @@ impl BytesSerializable for LoginWithPersonalAccessToken {
         }
 
         let command = LoginWithPersonalAccessToken { token };
-        command.validate()?;
         Ok(command)
     }
 }

sdk/src/quic/client.rs

@@ -64,6 +64,7 @@ impl BinaryTransport for QuicClient {
     }
 
     async fn send_with_response<T: Command>(&self, command: &T) -> Result<Bytes, IggyError> {
+        command.validate()?;
         self.send_raw_with_response(command.code(), command.to_bytes())
             .await
     }

sdk/src/streams/create_stream.rs

@@ -84,7 +84,6 @@ impl BytesSerializable for CreateStream {
         }
 
         let command = CreateStream { stream_id, name };
-        command.validate()?;
         Ok(command)
     }
 }

sdk/src/streams/delete_stream.rs

@@ -41,7 +41,6 @@ impl BytesSerializable for DeleteStream {
 
         let stream_id = Identifier::from_bytes(bytes)?;
         let command = DeleteStream { stream_id };
-        command.validate()?;
         Ok(command)
     }
 }

sdk/src/streams/get_stream.rs

@@ -41,7 +41,6 @@ impl BytesSerializable for GetStream {
 
         let stream_id = Identifier::from_bytes(bytes)?;
         let command = GetStream { stream_id };
-        command.validate()?;
         Ok(command)
     }
 }

sdk/src/streams/get_streams.rs

@@ -33,8 +33,6 @@ impl BytesSerializable for GetStreams {
             return Err(IggyError::InvalidCommand);
         }
 
-        let command = GetStreams {};
-        command.validate()?;
         Ok(GetStreams {})
     }
 }

sdk/src/streams/purge_stream.rs

@@ -44,7 +44,6 @@ impl BytesSerializable for PurgeStream {
 
         let stream_id = Identifier::from_bytes(bytes)?;
         let command = PurgeStream { stream_id };
-        command.validate()?;
         Ok(command)
     }
 }

sdk/src/streams/update_stream.rs

@@ -79,7 +79,6 @@ impl BytesSerializable for UpdateStream {
         }
 
         let command = UpdateStream { stream_id, name };
-        command.validate()?;
         Ok(command)
     }
 }

sdk/src/system/get_client.rs

@@ -51,7 +51,6 @@ impl BytesSerializable for GetClient {
 
         let client_id = u32::from_le_bytes(bytes.as_ref().try_into()?);
         let command = GetClient { client_id };
-        command.validate()?;
         Ok(command)
     }
 }

sdk/src/system/get_clients.rs

@@ -33,8 +33,6 @@ impl BytesSerializable for GetClients {
             return Err(IggyError::InvalidCommand);
         }
 
-        let command = GetClients {};
-        command.validate()?;
         Ok(GetClients {})
     }
 }

sdk/src/system/get_me.rs

@@ -33,8 +33,6 @@ impl BytesSerializable for GetMe {
             return Err(IggyError::InvalidCommand);
         }
 
-        let command = GetMe {};
-        command.validate()?;
         Ok(GetMe {})
     }
 }

sdk/src/system/get_stats.rs

@@ -33,8 +33,6 @@ impl BytesSerializable for GetStats {
             return Err(IggyError::InvalidCommand);
         }
 
-        let command = GetStats {};
-        command.validate()?;
         Ok(GetStats {})
     }
 }

sdk/src/system/ping.rs

@@ -34,7 +34,6 @@ impl BytesSerializable for Ping {
         }
 
         let command = Ping {};
-        command.validate()?;
         Ok(command)
     }
 }

sdk/src/tcp/client.rs

@@ -141,6 +141,7 @@ impl BinaryTransport for TcpClient {
     }
 
     async fn send_with_response<T: Command>(&self, command: &T) -> Result<Bytes, IggyError> {
+        command.validate()?;
         self.send_raw_with_response(command.code(), command.to_bytes())
             .await
     }

sdk/src/topics/create_topic.rs

@@ -150,7 +150,6 @@ impl BytesSerializable for CreateTopic {
             replication_factor,
             name,
         };
-        command.validate()?;
         Ok(command)
     }
 }

sdk/src/topics/delete_topic.rs

@@ -56,7 +56,6 @@ impl BytesSerializable for DeleteTopic {
             stream_id,
             topic_id,
         };
-        command.validate()?;
         Ok(command)
     }
 }

sdk/src/topics/get_topic.rs

@@ -56,7 +56,6 @@ impl BytesSerializable for GetTopic {
             stream_id,
             topic_id,
         };
-        command.validate()?;
         Ok(command)
     }
 }

sdk/src/topics/get_topics.rs

@@ -41,7 +41,6 @@ impl BytesSerializable for GetTopics {
 
         let stream_id = Identifier::from_bytes(bytes)?;
         let command = GetTopics { stream_id };
-        command.validate()?;
         Ok(command)
     }
 }

sdk/src/topics/purge_topic.rs

@@ -56,7 +56,6 @@ impl BytesSerializable for PurgeTopic {
             stream_id,
             topic_id,
         };
-        command.validate()?;
         Ok(command)
     }
 }

sdk/src/topics/update_topic.rs

@@ -139,7 +139,6 @@ impl BytesSerializable for UpdateTopic {
             replication_factor,
             name,
         };
-        command.validate()?;
         Ok(command)
     }
 }

sdk/src/users/change_password.rs

@@ -97,18 +97,13 @@ impl BytesSerializable for ChangePassword {
             current_password,
             new_password,
         };
-        command.validate()?;
         Ok(command)
     }
 }
 
 impl Display for ChangePassword {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        write!(
-            f,
-            "{}|{}|{}",
-            self.user_id, self.current_password, self.new_password
-        )
+        write!(f, "{}|******|******", self.user_id)
     }
 }
 

sdk/src/users/create_user.rs

@@ -137,7 +137,6 @@ impl BytesSerializable for CreateUser {
             status,
             permissions,
         };
-        command.validate()?;
         Ok(command)
     }
 }
@@ -151,8 +150,8 @@ impl Display for CreateUser {
         };
         write!(
             f,
-            "{}|{}|{}|{}",
-            self.username, self.password, self.status, permissions
+            "{}|******|{}|{}",
+            self.username, self.status, permissions
         )
     }
 }

sdk/src/users/delete_user.rs

@@ -41,7 +41,6 @@ impl BytesSerializable for DeleteUser {
 
         let user_id = Identifier::from_bytes(bytes)?;
         let command = DeleteUser { user_id };
-        command.validate()?;
         Ok(command)
     }
 }