Add binary protocol for adapter transports #117
Conversation
Introduces a binary protocol for inter-node communication using bincode and xxhash for efficient serialization and fast routing. Updates NATS, Redis, and Redis Cluster transports to use the new binary protocol instead of JSON for broadcast, request, and response messages. Adds comprehensive tests for binary protocol conversions and message parsing. Updates dependencies to include bincode and xxhash-rust.
Summary of ChangesHello @countradooku, I'm Gemini Code Assist1! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed! This pull request introduces a fundamental change to how nodes communicate within the system by implementing a new binary protocol. This strategic move from JSON to "bincode" serialization, coupled with "xxhash" for optimized routing, is designed to enhance the efficiency and speed of inter-node messaging. The change is applied across key adapter transports (NATS, Redis, Redis Cluster) to ensure a consistent and performant communication layer, ultimately leading to a more scalable and responsive distributed architecture. Highlights
Using Gemini Code AssistThe full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips. Invoking Gemini You can request assistance from Gemini at any point by creating a comment using either
Customization To customize Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a Limitations & Feedback Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counter productive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for Github and other Google products, sign up here. You can also get AI-powered code generation, chat, as well as code reviews directly in the IDE at no cost with the Gemini Code Assist IDE Extension. Footnotes
|
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces a binary protocol for inter-node communication using bincode and xxhash, which is a great performance enhancement over the previous JSON-based protocol. The implementation is well-structured, with a dedicated binary_protocol module and updates to the NATS and Redis transports. The addition of comprehensive tests is also commendable.
My review focuses on improving error handling and making the code more robust. I've identified a few places where errors are silently ignored, which could make debugging difficult. Specifically, I've suggested changes to propagate errors during message conversion and to add logging in the transport listeners when deserialization or conversion fails. I've also pointed out a couple of tests that could be improved by adding assertions to make them fully automated.
| impl From<BroadcastMessage> for BinaryBroadcastMessage { | ||
| fn from(msg: BroadcastMessage) -> Self { | ||
| let node_id_bytes = uuid_to_bytes(&msg.node_id).unwrap_or([0u8; 16]); | ||
| let channel_hash = calculate_channel_hash(&msg.channel); | ||
|
|
||
| // The message field contains the JSON string that should be sent to clients | ||
| let raw_client_json = msg.message.into_bytes(); | ||
|
|
||
| Self { | ||
| version: BINARY_PROTOCOL_VERSION, | ||
| channel_hash, | ||
| channel: msg.channel, | ||
| node_id_bytes, | ||
| app_id: msg.app_id, | ||
| raw_client_json, | ||
| except_socket_id: msg.except_socket_id, | ||
| timestamp_ms: msg.timestamp_ms, | ||
| } | ||
| } | ||
| } |
There was a problem hiding this comment.
The From implementation for BinaryBroadcastMessage uses .unwrap_or([0u8; 16]) when converting the node_id. This will silently replace an invalid UUID with a zeroed-out one, which could lead to hard-to-debug issues like message loss or mis-routing. For consistency with other fallible conversions in this module (like for RequestBody), this should be converted to an impl TryFrom, propagating any parsing errors. This change will require updating the call sites to use try_into()?.
impl TryFrom<BroadcastMessage> for BinaryBroadcastMessage {
type Error = Error;
fn try_from(msg: BroadcastMessage) -> Result<Self> {
let node_id_bytes = uuid_to_bytes(&msg.node_id)?;
let channel_hash = calculate_channel_hash(&msg.channel);
// The message field contains the JSON string that should be sent to clients
let raw_client_json = msg.message.into_bytes();
Ok(Self {
version: BINARY_PROTOCOL_VERSION,
channel_hash,
channel: msg.channel,
node_id_bytes,
app_id: msg.app_id,
raw_client_json,
except_socket_id: msg.except_socket_id,
timestamp_ms: msg.timestamp_ms,
})
}
}| if let Ok(binary_req) = bincode::deserialize::<BinaryRequestBody>(&msg.payload) { | ||
| if let Ok(request) = RequestBody::try_from(binary_req) { | ||
| let response_result = request_handler(request).await; | ||
|
|
||
| if let Ok(response) = response_result { | ||
| // Serialize response to binary | ||
| if let Ok(binary_resp) = BinaryResponseBody::try_from(response) { | ||
| if let Ok(response_data) = bincode::serialize(&binary_resp) { | ||
| let _ = response_client | ||
| .publish( | ||
| Subject::from(response_subject.clone()), | ||
| response_data.into(), | ||
| ) | ||
| .await; | ||
| } | ||
| } | ||
| } | ||
| } | ||
| } |
There was a problem hiding this comment.
The request handling logic contains deeply nested if let statements. This pattern makes the code hard to read and, more importantly, it silently swallows errors. If any of the try_from or serialize calls fail, the message is simply dropped without any logging, which can make debugging very difficult. The error handling should be explicit, and failures should be logged to aid in debugging.
let binary_req = match bincode::deserialize::<BinaryRequestBody>(&msg.payload) {
Ok(req) => req,
Err(e) => {
warn!("Failed to deserialize binary request: {}", e);
continue;
}
};
let request = match RequestBody::try_from(binary_req) {
Ok(req) => req,
Err(e) => {
warn!("Failed to convert binary request to request body: {}", e);
continue;
}
};
if let Ok(response) = request_handler(request).await {
let binary_resp = match BinaryResponseBody::try_from(response) {
Ok(resp) => resp,
Err(e) => {
warn!("Failed to convert response to binary format: {}", e);
continue;
}
};
let response_data = match bincode::serialize(&binary_resp) {
Ok(data) => data,
Err(e) => {
warn!("Failed to serialize binary response: {}", e);
continue;
}
};
let _ = response_client
.publish(
Subject::from(response_subject.clone()),
response_data.into(),
)
.await;
}| if let Ok(binary_req) = | ||
| bincode::deserialize::<BinaryRequestBody>(&payload_bytes) | ||
| { | ||
| if let Ok(request) = RequestBody::try_from(binary_req) { | ||
| let response_result = request_handler(request).await; | ||
|
|
||
| if let Ok(response) = response_result { | ||
| // Serialize response to binary | ||
| if let Ok(binary_resp) = BinaryResponseBody::try_from(response) | ||
| { | ||
| if let Ok(response_bytes) = bincode::serialize(&binary_resp) | ||
| { | ||
| // Use client's connection pooling for response publishing | ||
| if let Ok(mut conn) = | ||
| client_clone.get_async_connection().await | ||
| { | ||
| let _ = conn | ||
| .publish::<_, _, ()>( | ||
| &response_channel_clone, | ||
| response_bytes, | ||
| ) | ||
| .await; | ||
| } | ||
| } | ||
| } | ||
| } | ||
| } | ||
| } |
There was a problem hiding this comment.
The request handling logic contains deeply nested if let statements. This pattern makes the code hard to read and, more importantly, it silently swallows errors. If any of the try_from or serialize calls fail, the message is simply dropped without any logging, which can make debugging very difficult. The error handling should be explicit, and failures should be logged.
let binary_req = match bincode::deserialize::<BinaryRequestBody>(&payload_bytes) {
Ok(req) => req,
Err(e) => {
warn!("Failed to deserialize binary request: {}", e);
return;
}
};
let request = match RequestBody::try_from(binary_req) {
Ok(req) => req,
Err(e) => {
warn!("Failed to convert binary request to request body: {}", e);
return;
}
};
if let Ok(response) = request_handler(request).await {
let binary_resp = match BinaryResponseBody::try_from(response) {
Ok(resp) => resp,
Err(e) => {
warn!("Failed to convert response to binary format: {}", e);
return;
}
};
let response_bytes = match bincode::serialize(&binary_resp) {
Ok(data) => data,
Err(e) => {
warn!("Failed to serialize binary response: {}", e);
return;
}
};
if let Ok(mut conn) = client_clone.get_async_connection().await {
let _ = conn
.publish::<_, _, ()>(
&response_channel_clone,
response_bytes,
)
.await;
} else {
warn!("Failed to get redis connection to publish response");
}
}| if let Ok(binary_req) = | ||
| bincode::deserialize::<BinaryRequestBody>(payload.as_bytes()) | ||
| { | ||
| if let Ok(request) = RequestBody::try_from(binary_req) { | ||
| let response_result = request_handler(request).await; | ||
|
|
||
| if let Ok(response) = response_result { | ||
| // Serialize response to binary | ||
| if let Ok(binary_resp) = | ||
| BinaryResponseBody::try_from(response) | ||
| { | ||
| if let Ok(response_bytes) = | ||
| bincode::serialize(&binary_resp) | ||
| { | ||
| let mut conn = pub_connection_clone.clone(); | ||
| let _ = conn | ||
| .publish::<_, _, ()>( | ||
| &response_channel_clone, | ||
| response_bytes, | ||
| ) | ||
| .await; | ||
| } | ||
| } | ||
| } | ||
| } | ||
| } |
There was a problem hiding this comment.
The request handling logic contains deeply nested if let statements. This pattern makes the code hard to read and, more importantly, it silently swallows errors. If any of the try_from or serialize calls fail, the message is simply dropped without any logging, which can make debugging very difficult. The error handling should be explicit, and failures should be logged.
let binary_req =
match bincode::deserialize::<BinaryRequestBody>(payload.as_bytes()) {
Ok(req) => req,
Err(e) => {
warn!("Failed to deserialize binary request: {}", e);
return;
}
};
let request = match RequestBody::try_from(binary_req) {
Ok(req) => req,
Err(e) => {
warn!("Failed to convert binary request to request body: {}", e);
return;
}
};
if let Ok(response) = request_handler(request).await {
let binary_resp = match BinaryResponseBody::try_from(response) {
Ok(resp) => resp,
Err(e) => {
warn!("Failed to convert response to binary format: {}", e);
return;
}
};
let response_bytes = match bincode::serialize(&binary_resp) {
Ok(data) => data,
Err(e) => {
warn!("Failed to serialize binary response: {}", e);
return;
}
};
let mut conn = pub_connection_clone.clone();
let _ = conn
.publish::<_, _, ()>(
&response_channel_clone,
response_bytes,
)
.await;
}
src/adapter/binary_protocol.rs
Outdated
| let binary_size = bincode::serialize(&binary).unwrap().len(); | ||
|
|
||
| // Binary should be smaller or similar in size | ||
| println!("JSON size: {}, Binary size: {}", json_size, binary_size); |
There was a problem hiding this comment.
The test test_bincode_serialization_size currently only prints the JSON and binary sizes to the console. This makes it a manual check rather than an automated test. To make this a useful regression test, an assertion should be added to ensure the binary format remains efficient. For this test case, the binary size should be smaller than the JSON size.
println!("JSON size: {}, Binary size: {}", json_size, binary_size);
assert!(binary_size < json_size);| if let Ok(binary_resp) = bincode::deserialize::<BinaryResponseBody>(&msg.payload) { | ||
| if let Ok(response) = ResponseBody::try_from(binary_resp) { | ||
| response_handler(response).await; | ||
| } | ||
| } else { | ||
| warn!("Failed to parse binary response message"); | ||
| } |
There was a problem hiding this comment.
If bincode::deserialize succeeds but ResponseBody::try_from fails, the error is silently ignored. This can happen during rolling updates if there's a version mismatch or other data format issues. It's important to log this failure to aid in debugging.
if let Ok(binary_resp) = bincode::deserialize::<BinaryResponseBody>(&msg.payload) {
if let Ok(response) = ResponseBody::try_from(binary_resp) {
response_handler(response).await;
} else {
warn!("Failed to convert binary response to response body");
}
} else {
warn!("Failed to parse binary response message");
}| if let Ok(binary_resp) = | ||
| bincode::deserialize::<BinaryResponseBody>(&payload_bytes) | ||
| { | ||
| if let Ok(response) = ResponseBody::try_from(binary_resp) { | ||
| response_handler(response).await; | ||
| } | ||
| } else { | ||
| warn!("Failed to parse binary response message"); | ||
| } |
There was a problem hiding this comment.
If bincode::deserialize succeeds but ResponseBody::try_from fails, the error is silently ignored. This can happen during rolling updates if there's a version mismatch. It's important to log this failure to aid in debugging.
if let Ok(binary_resp) =
bincode::deserialize::<BinaryResponseBody>(&payload_bytes)
{
if let Ok(response) = ResponseBody::try_from(binary_resp) {
response_handler(response).await;
} else {
warn!("Failed to convert binary response to response body");
}
} else {
warn!("Failed to parse binary response message");
}| if let Ok(binary_resp) = | ||
| bincode::deserialize::<BinaryResponseBody>(payload.as_bytes()) | ||
| { | ||
| response_handler(response).await; | ||
| if let Ok(response) = ResponseBody::try_from(binary_resp) { | ||
| response_handler(response).await; | ||
| } | ||
| } else { | ||
| warn!("Failed to parse response message: {}", payload); | ||
| warn!("Failed to parse binary response message"); | ||
| } |
There was a problem hiding this comment.
If bincode::deserialize succeeds but ResponseBody::try_from fails, the error is silently ignored. This can happen during rolling updates if there's a version mismatch. It's important to log this failure to aid in debugging. Additionally, the warning for a failed deserialization no longer includes the error details, which is a regression from the previous implementation.
match bincode::deserialize::<BinaryResponseBody>(payload.as_bytes()) {
Ok(binary_resp) => {
if let Ok(response) = ResponseBody::try_from(binary_resp) {
response_handler(response).await;
} else {
warn!("Failed to convert binary response to response body");
}
}
Err(e) => {
warn!("Failed to parse binary response message: {}", e);
}
}| println!("JSON size: {} bytes", json_bytes.len()); | ||
| println!("Binary size: {} bytes", binary_bytes.len()); | ||
|
|
||
| // Binary should be competitive or better | ||
| // Note: For very small messages, overhead might make binary slightly larger, | ||
| // but for typical messages with longer payloads, binary will be smaller |
There was a problem hiding this comment.
The test test_bincode_size_reduction currently only prints the JSON and binary sizes to the console. This makes it a manual check rather than an automated test. To make this a useful regression test, an assertion should be added to ensure the binary format remains efficient.
| println!("JSON size: {} bytes", json_bytes.len()); | |
| println!("Binary size: {} bytes", binary_bytes.len()); | |
| // Binary should be competitive or better | |
| // Note: For very small messages, overhead might make binary slightly larger, | |
| // but for typical messages with longer payloads, binary will be smaller | |
| println!("JSON size: {} bytes", json_bytes.len()); | |
| println!("Binary size: {} bytes", binary_bytes.len()); | |
| // Binary should be competitive or better. For this payload, it should be smaller. | |
| assert!(binary_bytes.len() < json_bytes.len()); |
|
/gemini review |
There was a problem hiding this comment.
Code Review
This pull request introduces a binary protocol for inter-node communication, replacing JSON with bincode for serialization and xxhash for routing. This is a significant performance improvement. The changes are well-contained within a new binary_protocol.rs module and are consistently applied across the NATS, Redis, and Redis Cluster transports. The addition of comprehensive tests is also great.
My review focuses on a few key areas to enhance robustness and maintainability:
- Ensuring error propagation in data conversions.
- Maintaining consistency in the use of the new binary protocol configuration.
- Improving test structure and correctness.
Overall, this is a solid enhancement to the project's performance and scalability.
| impl From<BroadcastMessage> for BinaryBroadcastMessage { | ||
| fn from(msg: BroadcastMessage) -> Self { | ||
| let node_id_bytes = uuid_to_bytes(&msg.node_id).unwrap_or([0u8; 16]); | ||
| let channel_hash = calculate_channel_hash(&msg.channel); | ||
|
|
||
| // The message field contains the JSON string that should be sent to clients | ||
| let raw_client_json = msg.message.into_bytes(); | ||
|
|
||
| Self { | ||
| version: BINARY_PROTOCOL_VERSION, | ||
| channel_hash, | ||
| channel: msg.channel, | ||
| node_id_bytes, | ||
| app_id: msg.app_id, | ||
| raw_client_json, | ||
| except_socket_id: msg.except_socket_id, | ||
| timestamp_ms: msg.timestamp_ms, | ||
| } | ||
| } | ||
| } |
There was a problem hiding this comment.
Using unwrap_or here can silently hide errors. If uuid_to_bytes fails due to a malformed UUID string, the node_id will be replaced with a zero-UUID ([0u8; 16]). This could lead to hard-to-debug issues, such as messages being misattributed or lost. It's safer to propagate the error by using TryFrom instead of From. This will require updating the call sites from .into() to .try_into()?, which is a worthwhile trade-off for correctness.
impl TryFrom<BroadcastMessage> for BinaryBroadcastMessage {
type Error = Error;
fn try_from(msg: BroadcastMessage) -> Result<Self> {
let node_id_bytes = uuid_to_bytes(&msg.node_id)?;
let channel_hash = calculate_channel_hash(&msg.channel);
// The message field contains the JSON string that should be sent to clients
let raw_client_json = msg.message.into_bytes();
Ok(Self {
version: BINARY_PROTOCOL_VERSION,
channel_hash,
channel: msg.channel,
node_id_bytes,
app_id: msg.app_id,
raw_client_json,
except_socket_id: msg.except_socket_id,
timestamp_ms: msg.timestamp_ms,
})
}
}| #[test] | ||
| fn test_bincode_serialization_size() { | ||
| let msg = BroadcastMessage { | ||
| node_id: "550e8400-e29b-41d4-a716-446655440000".to_string(), | ||
| app_id: "test-app".to_string(), | ||
| channel: "test-channel".to_string(), | ||
| message: r#"{"event":"test","data":"payload"}"#.to_string(), | ||
| except_socket_id: None, | ||
| timestamp_ms: Some(1234567890.123), | ||
| }; | ||
|
|
||
| let _json_size = serde_json::to_vec(&msg).unwrap().len(); | ||
| let binary: BinaryBroadcastMessage = msg.into(); | ||
| let _binary_size = bincode::encode_to_vec(&binary, bincode_config()) | ||
| .unwrap() | ||
| .len(); | ||
| } |
There was a problem hiding this comment.
This test function calculates serialization sizes but doesn't perform any assertions. The variables _json_size and _binary_size are unused, which will result in compiler warnings. A test without assertions doesn't verify any behavior and can be misleading. Consider adding a meaningful assertion or removing this test if it was intended for manual debugging.
fn test_bincode_serialization_size() {
let msg = BroadcastMessage {
node_id: "550e8400-e29b-41d4-a716-446655440000".to_string(),
app_id: "test-app".to_string(),
channel: "test-channel".to_string(),
message: r#"{"event":"test","data":"payload"}"#.to_string(),
except_socket_id: None,
timestamp_ms: Some(1234567890.123),
};
let json_size = serde_json::to_vec(&msg).unwrap().len();
let binary: BinaryBroadcastMessage = msg.try_into().unwrap();
let binary_size = bincode::encode_to_vec(&binary, bincode_config())
.unwrap()
.len();
assert!(binary_size < json_size, "binary size ({}) should be less than json size ({})", binary_size, json_size);
}| let message_data = bincode::encode_to_vec(&binary_msg, bincode::config::standard()) | ||
| .map_err(|e| Error::Other(format!("Failed to serialize broadcast: {}", e)))?; |
There was a problem hiding this comment.
For consistency and to enforce the message size limit, it's better to use the shared bincode_config() function from the binary_protocol module instead of bincode::config::standard(). This ensures that all serialization and deserialization operations across transports follow the same rules, including the MAX_MESSAGE_SIZE limit for security. This applies to all bincode::encode_to_vec and bincode::decode_from_slice calls in this file.
| let message_data = bincode::encode_to_vec(&binary_msg, bincode::config::standard()) | |
| .map_err(|e| Error::Other(format!("Failed to serialize broadcast: {}", e)))?; | |
| let message_data = bincode::encode_to_vec(&binary_msg, crate::adapter::binary_protocol::bincode_config()) | |
| .map_err(|e| Error::Other(format!("Failed to serialize broadcast: {}", e)))?; |
| let broadcast_bytes = bincode::encode_to_vec(&binary_msg, bincode::config::standard()) | ||
| .map_err(|e| Error::Other(format!("Failed to serialize broadcast: {}", e)))?; |
There was a problem hiding this comment.
For consistency and to enforce the message size limit, it's better to use the shared bincode_config() function from the binary_protocol module instead of bincode::config::standard(). This ensures that all serialization and deserialization operations across transports follow the same rules, including the MAX_MESSAGE_SIZE limit for security. This applies to all bincode::encode_to_vec and bincode::decode_from_slice calls in this file.
| let broadcast_bytes = bincode::encode_to_vec(&binary_msg, bincode::config::standard()) | |
| .map_err(|e| Error::Other(format!("Failed to serialize broadcast: {}", e)))?; | |
| let broadcast_bytes = bincode::encode_to_vec(&binary_msg, crate::adapter::binary_protocol::bincode_config()) | |
| .map_err(|e| Error::Other(format!("Failed to serialize broadcast: {}", e)))?; |
| let broadcast_bytes = bincode::encode_to_vec(&binary_msg, bincode::config::standard()) | ||
| .map_err(|e| Error::Other(format!("Failed to serialize broadcast: {}", e)))?; |
There was a problem hiding this comment.
For consistency and to enforce the message size limit, it's better to use the shared bincode_config() function from the binary_protocol module instead of bincode::config::standard(). This ensures that all serialization and deserialization operations across transports follow the same rules, including the MAX_MESSAGE_SIZE limit for security. This applies to all bincode::encode_to_vec and bincode::decode_from_slice calls in this file.
| let broadcast_bytes = bincode::encode_to_vec(&binary_msg, bincode::config::standard()) | |
| .map_err(|e| Error::Other(format!("Failed to serialize broadcast: {}", e)))?; | |
| let broadcast_bytes = bincode::encode_to_vec(&binary_msg, crate::adapter::binary_protocol::bincode_config()) | |
| .map_err(|e| Error::Other(format!("Failed to serialize broadcast: {}", e)))?; |
| @@ -0,0 +1,230 @@ | |||
| use sockudo::adapter::binary_protocol::*; | |||
There was a problem hiding this comment.
This new test file contains comprehensive tests for the binary protocol conversions. However, these appear to be unit tests for the binary_protocol module. To improve maintainability and keep tests co-located with the code they test, it's a Rust convention to place unit tests inside the module file itself, under a #[cfg(test)] mod tests { ... } block.
Consider moving the tests from this file into src/adapter/binary_protocol.rs and deleting this file. This will consolidate all binary_protocol unit tests in one place, making them easier to find and maintain. The existing tests in binary_protocol.rs can be merged with or replaced by these more comprehensive ones.
Additionally, the test_bincode_size_reduction test currently only prints to the console and has no assertions. Tests should be verifiable. Please add an assertion (e.g., assert!(binary_bytes.len() < json_bytes.len())) or remove it if it's for debugging purposes.
Introduces a binary protocol for inter-node communication using bincode and xxhash for efficient serialization and fast routing. Updates NATS, Redis, and Redis Cluster transports to use the new binary protocol instead of JSON for broadcast, request, and response messages. Adds comprehensive tests for binary protocol conversions and message parsing. Updates dependencies to include bincode and xxhash-rust.
Description
Type of Change
Testing
Checklist
Build Artifacts
Additional Notes