[Networking] Do not spawn an RPC responder task if message_router dropped the message

This also adds a new metric/counter to count the bytes we dropped after receiving due to a lossy service.

Author: AhmedSoliman

crates/core/src/network/io/reactor.rs

@@ -34,7 +34,9 @@ use restate_types::{Version, Versioned};
 
 use crate::network::incoming::{RawRpc, RawUnary, RpcReplyPort};
 use crate::network::io::EgressMessage;
-use crate::network::metric_definitions::NETWORK_MESSAGE_RECEIVED_BYTES;
+use crate::network::metric_definitions::{
+    NETWORK_MESSAGE_RECEIVED_BYTES, NETWORK_MESSAGE_RECEIVED_DROPPED_BYTES,
+};
 use crate::network::protobuf::network::message::{Body, Signal};
 use crate::network::protobuf::network::{Datagram, RpcReply, datagram, rpc_reply};
 use crate::network::protobuf::network::{Header, Message};
@@ -332,7 +334,7 @@ impl ConnectionReactor {
                 };
                 let target_service = rpc_call.service();
 
-                let encoded_len = rpc_call.payload.len();
+                let encoded_len = rpc_call.payload.len() as u64;
                 let (reply_port, reply_rx) = RpcReplyPort::new();
                 let raw_rpc = RawRpc {
                     reply_port,
@@ -352,22 +354,21 @@ impl ConnectionReactor {
                     "Received RPC call: {target_service}::{}",
                     incoming.msg_type()
                 );
+
                 // ship to the service router, dropping the reply port will close the responder
                 // task.
                 match tokio::task::unconstrained(self.router.call_rpc(target_service, incoming))
                     .await
                 {
-                    Ok(()) => { /* spawn reply task */ }
+                    Ok(()) => {
+                        counter!(NETWORK_MESSAGE_RECEIVED_BYTES, "target" => target_service.as_str_name()).increment(encoded_len);
+                        spawn_rpc_responder(tx.clone(), rpc_call.id, reply_rx, target_service);
+                    }
                     Err(err) => {
                         send_rpc_error(tx, err, rpc_call.id);
+                        counter!(NETWORK_MESSAGE_RECEIVED_DROPPED_BYTES, "target" => target_service.as_str_name()).increment(encoded_len);
                     }
                 }
-
-                counter!(NETWORK_MESSAGE_RECEIVED_BYTES, "target" => target_service.as_str_name())
-                    .increment(encoded_len as u64);
-
-                spawn_rpc_responder(tx.clone(), rpc_call.id, reply_rx, target_service);
-
                 Decision::Continue
             }
             // UNARY MESSAGE
@@ -376,7 +377,7 @@ impl ConnectionReactor {
             }) => {
                 let metadata_versions = PeerMetadataVersion::from(header);
                 let target = unary.service();
-                let encoded_len = unary.payload.len();
+                let encoded_len = unary.payload.len() as u64;
                 let incoming = Incoming::new(
                     self.connection.protocol_version,
                     RawUnary {
@@ -392,7 +393,7 @@ impl ConnectionReactor {
                 let _ = tokio::task::unconstrained(self.router.call_unary(target, incoming)).await;
 
                 counter!(NETWORK_MESSAGE_RECEIVED_BYTES, "target" => target.as_str_name())
-                    .increment(encoded_len as u64);
+                    .increment(encoded_len);
                 Decision::Continue
             }
             // RPC REPLY

crates/core/src/network/metric_definitions.rs

@@ -13,6 +13,8 @@ use metrics::{Unit, describe_counter, describe_histogram};
 pub const NETWORK_CONNECTION_CREATED: &str = "restate.network.connection_created.total";
 pub const NETWORK_CONNECTION_DROPPED: &str = "restate.network.connection_dropped.total";
 pub const NETWORK_MESSAGE_RECEIVED_BYTES: &str = "restate.network.message_received_bytes.total";
+pub const NETWORK_MESSAGE_RECEIVED_DROPPED_BYTES: &str =
+    "restate.network.message_received_dropped_bytes.total";
 
 pub const NETWORK_MESSAGE_PROCESSING_DURATION: &str =
     "restate.network.message_processing_duration.seconds";
@@ -31,7 +33,13 @@ pub fn describe_metrics() {
     describe_counter!(
         NETWORK_MESSAGE_RECEIVED_BYTES,
         Unit::Bytes,
-        "Number of bytes received by message name"
+        "Number of bytes received by service name"
+    );
+
+    describe_counter!(
+        NETWORK_MESSAGE_RECEIVED_DROPPED_BYTES,
+        Unit::Bytes,
+        "Number of bytes received and dropped/rejected by service name"
     );
 
     describe_histogram!(