[#28665] DocDB: Add CRC to RPC response

Summary:
It could happen that RPC call response could be corrupted on lower layers.
Which could lead to data actual corruption.
To avoid this we add CRC checksum like it was already done for RPC requests to filter out such responses.

**Upgrade/Rollback safety:**
Changed CRC field ids, so different versions would not see CRC from each other.
So CRC would not be checked between old and new version.
Jira: DB-18359

Test Plan: Jenkins

Reviewers: mlillibridge

Reviewed By: mlillibridge

Subscribers: kannan, hbhanawat, rthallam, ybase

Tags: #jenkins-ready

Differential Revision: https://phorge.dev.yugabyte.com/D46647

Author: spolitov

src/yb/rpc/constants.h

@@ -34,11 +34,9 @@
 
 #include <stdint.h>
 
-namespace yb {
-namespace rpc {
+namespace yb::rpc {
 
 // There is a 4-byte length prefix before any packet.
-const uint8_t kMsgLengthPrefixLength = 4;
+constexpr size_t kMsgLengthPrefixLength = 4;
 
-} // namespace rpc
-} // namespace yb
+} // namespace yb::rpc

src/yb/rpc/outbound_call.cc

@@ -403,13 +403,7 @@ Status OutboundCall::SetRequestParam(
   }
   RETURN_NOT_OK(SerializeMessage(req, req_size, buffer_, sidecars_size, header_size));
   if (use_crc) {
-    auto crc = crc::Crc64c(buffer_.udata() + header_size, message_size, 0);
-    if (sidecars_size) {
-      sidecars_->buffer().IterateBlocks([&crc](Slice block) {
-        crc = crc::Crc64c(block.data(), block.size(), crc);
-      });
-    }
-    LittleEndian::Store32(buffer_.udata() + header_size - 4, static_cast<uint32_t>(crc));
+    StoreCrc(buffer_, header_size, message_size, sidecars_.get());
   }
 
   if (method_metrics_) {

src/yb/rpc/rpc-test.cc

@@ -1184,6 +1184,7 @@ void TestMaxSizeRpcResponse(CalculatorServiceProxy* proxy) {
   ResponseHeader resp_header;
   resp_header.set_call_id(1);
   resp_header.set_is_error(false);
+  resp_header.set_crc(0);
 
   const size_t header_pb_len = resp_header.ByteSize();
   const size_t header_tot_len = OutboundCall::HeaderTotalLength(header_pb_len);

src/yb/rpc/rpc_header.proto

@@ -56,6 +56,8 @@ message RemoteMethodPB {
 
 // The header for the RPC request frame.
 message RequestHeader {
+  reserved 6;
+
   // A sequence number that is sent back in the Response. Hadoop specifies a uint32 and
   // casts it to a signed int. That is counterintuitive, so we use an int32 instead.
   // Allowed values (inherited from Hadoop):
@@ -78,11 +80,13 @@ message RequestHeader {
 
   optional AshMetadataPB metadata = 5;
 
-  // CRC32C for message body.
-  optional fixed32 crc = 6;
+  // CRC32C for message body and header without CRC field. Use 15 as max 1 byte tag.
+  optional fixed32 crc = 15;
 }
 
 message ResponseHeader {
+  reserved 4;
+
   required int32 call_id = 1;
 
   // If this is set, then this is an error response and the
@@ -95,7 +99,8 @@ message ResponseHeader {
   // is the first byte after the bytes for this protobuf.
   repeated uint32 sidecar_offsets = 3;
 
-  optional fixed32 crc = 4;
+  // CRC32C for message body and header without CRC field. Use 15 as max 1 byte tag.
+  optional fixed32 crc = 15;
 }
 
 // An emtpy message. Since CQL RPC server bypasses protobuf to handle requests and responses but

src/yb/rpc/serialization.cc

@@ -43,6 +43,7 @@
 
 #include "yb/rpc/call_data.h"
 #include "yb/rpc/rpc_header.pb.h"
+#include "yb/rpc/sidecars.h"
 
 #include "yb/util/crc.h"
 #include "yb/util/faststring.h"
@@ -57,8 +58,10 @@ using google::protobuf::MessageLite;
 using google::protobuf::io::CodedInputStream;
 using google::protobuf::io::CodedOutputStream;
 
-namespace yb {
-namespace rpc {
+namespace yb::rpc {
+
+// tag + 4 bytes values
+constexpr size_t kCrcFieldSize = 5;
 
 size_t SerializedMessageSize(size_t body_size, size_t additional_size) {
   auto full_size = body_size + additional_size;
@@ -133,7 +136,7 @@ Status SerializeHeader(const MessageLite& header,
   return Status::OK();
 }
 
-Result<RefCntBuffer> SerializeRequest(
+Result<std::pair<RefCntBuffer, size_t>> SerializeResponse(
     size_t body_size, size_t additional_size, const google::protobuf::Message& header,
     AnyMessageConstPtr body) {
   auto message_size = SerializedMessageSize(body_size, additional_size);
@@ -143,7 +146,7 @@ Result<RefCntBuffer> SerializeRequest(
       header, message_size + additional_size, &result, message_size, &header_size));
 
   RETURN_NOT_OK(SerializeMessage(body, body_size, result, additional_size, header_size));
-  return result;
+  return std::pair{result, header_size};
 }
 
 bool SkipField(uint8_t type, CodedInputStream* in) {
@@ -267,11 +270,12 @@ Result<Slice> ParseYBHeader(Slice buf, Header* parsed_header) {
   in.PopLimit(l);
 
   if (auto crc = GetCrc(*parsed_header)) {
-    auto offset = in.CurrentPosition();
-    auto msg_crc = crc::Crc32c(buf.data() + offset, buf.size() - offset);
-    if (msg_crc != *crc) {
+    crc::Crc32Accumulator msg_crc;
+    msg_crc.Feed(buf.Prefix(in.CurrentPosition() - kCrcFieldSize));
+    msg_crc.Feed(buf.WithoutPrefix(in.CurrentPosition()));
+    if (msg_crc.result() != *crc) {
       auto status = STATUS_FORMAT(
-          Corruption, "Invalid CRC $0 for request $1", msg_crc, *parsed_header);
+          Corruption, "Invalid CRC $0 for $1", msg_crc.result(), *parsed_header);
       LOG(DFATAL) << status;
       return status;
     }
@@ -411,5 +415,20 @@ Status ParseMetadataFromSharedMemory(
   return ParseMetadata(metadata, out);
 }
 
-}  // namespace rpc
-}  // namespace yb
+void StoreCrc(
+    const RefCntBuffer& buffer, size_t header_size, size_t message_size, Sidecars* sidecars) {
+  crc::Crc32Accumulator crc;
+  // 5 last bytes for CRC.
+  crc.Feed(
+      buffer.AsSlice().Prefix(header_size - kCrcFieldSize).WithoutPrefix(kMsgLengthPrefixLength));
+  crc.Feed(buffer.udata() + header_size, message_size);
+  if (sidecars) {
+    sidecars->buffer().IterateBlocks([&crc](Slice block) {
+      crc.Feed(block.data(), block.size());
+    });
+  }
+  // Expect CRC to be located at header end.
+  LittleEndian::Store32(buffer.udata() + header_size - 4, crc.result());
+}
+
+}  // namespace yb::rpc

src/yb/rpc/serialization.h

@@ -60,7 +60,7 @@ class Status;
 
 namespace rpc {
 
-Result<RefCntBuffer> SerializeRequest(
+Result<std::pair<RefCntBuffer, size_t>> SerializeResponse(
     size_t body_size, size_t additional_size, const google::protobuf::Message& header,
     AnyMessageConstPtr body);
 
@@ -105,5 +105,8 @@ Result<ParsedRemoteMethod> ParseRemoteMethod(const Slice& buf);
 Status ParseMetadata(Slice buf, AnyMessagePtr out);
 Status ParseMetadataFromSharedMemory(uint8_t** input, size_t length, AnyMessagePtr out);
 
+void StoreCrc(
+    const RefCntBuffer& buffer, size_t header_size, size_t message_size, Sidecars* sidecars);
+
 }  // namespace rpc
 }  // namespace yb

src/yb/rpc/yb_rpc.cc

@@ -29,6 +29,7 @@
 #include "yb/rpc/rpc_introspection.pb.h"
 #include "yb/rpc/serialization.h"
 
+#include "yb/util/crc.h"
 #include "yb/util/debug/trace_event.h"
 #include "yb/util/flags.h"
 #include "yb/util/format.h"
@@ -49,6 +50,7 @@ DEFINE_test_flag(uint64, yb_inbound_big_calls_parse_delay_ms, false,
                  "rpc_throttle_threshold_bytes");
 
 DECLARE_bool(rpc_dump_all_traces);
+DECLARE_bool(rpc_enable_crc);
 DECLARE_bool(ysql_yb_enable_ash);
 DECLARE_int32(print_trace_every);
 DECLARE_int32(rpc_slow_query_threshold_ms);
@@ -325,9 +327,19 @@ Status YBInboundCall::SerializeResponseBuffer(AnyMessageConstPtr response, bool
   ResponseHeader resp_hdr;
   resp_hdr.set_call_id(header_.call_id);
   resp_hdr.set_is_error(!is_success);
+  auto use_crc = FLAGS_rpc_enable_crc;
+  if (use_crc) {
+    resp_hdr.set_crc(0);
+  }
   sidecars_.MoveOffsetsTo(body_size, resp_hdr.mutable_sidecar_offsets());
 
-  response_buf_ = VERIFY_RESULT(SerializeRequest(body_size, sidecars_.size(), resp_hdr, response));
+  size_t header_size;
+  std::tie(response_buf_, header_size) = VERIFY_RESULT(SerializeResponse(
+      body_size, sidecars_.size(), resp_hdr, response));
+  if (use_crc) {
+    StoreCrc(response_buf_, header_size, response_buf_.size() - header_size, &sidecars_);
+  }
+
   response_data_memory_usage_ = response_buf_.size();
 
   return Status::OK();

src/yb/util/crc.cc

@@ -36,8 +36,7 @@
 #include "yb/gutil/once.h"
 #include "yb/util/debug/leakcheck_disabler.h"
 
-namespace yb {
-namespace crc {
+namespace yb::crc {
 
 using debug::ScopedLeakCheckDisabler;
 
@@ -58,13 +57,12 @@ Crc* GetCrc32cInstance() {
 uint32_t Crc32c(const void* data, size_t length) {
   uint64_t crc32 = 0;
   GetCrc32cInstance()->Compute(data, length, &crc32);
-  return static_cast<uint32_t>(crc32); // Only uses lower 32 bits.
+  // Only uses lower 32 bits, since top 32 bits are always zero.
+  return static_cast<uint32_t>(crc32);
 }
 
-uint64_t Crc64c(const void* data, size_t length, uint64_t start) {
-  GetCrc32cInstance()->Compute(data, length, &start);
-  return start;
+void Crc32Accumulator::Feed(const void* data, size_t length) {
+  GetCrc32cInstance()->Compute(data, length, &state_);
 }
 
-} // namespace crc
-} // namespace yb
+} // namespace yb::crc

src/yb/util/crc.h

@@ -36,8 +36,9 @@
 
 #include <crcutil/interface.h>
 
-namespace yb {
-namespace crc {
+#include "yb/util/slice.h"
+
+namespace yb::crc {
 
 typedef crcutil_interface::CRC Crc;
 
@@ -47,7 +48,27 @@ Crc* GetCrc32cInstance();
 // Helper function to simply calculate a CRC32C of the given data.
 uint32_t Crc32c(const void* data, size_t length);
 
-uint64_t Crc64c(const void* data, size_t length, uint64_t start);
+inline uint32_t Crc32c(Slice slice) {
+  return Crc32c(slice.data(), slice.size());
+}
+
+class Crc32Accumulator {
+ public:
+  explicit Crc32Accumulator(uint64_t state = 0) : state_(state) {}
+
+  void Feed(Slice slice) {
+    Feed(slice.data(), slice.size());
+  }
+
+  void Feed(const void* data, size_t length);
+
+  uint32_t result() const {
+    return static_cast<uint32_t>(state_);
+  }
+
+ private:
+  // CRC32C has 64 bits state, but top 32 bits are always zero.
+  uint64_t state_;
+};
 
-} // namespace crc
-} // namespace yb
+} // namespace yb::crc