fix(asan): heap-use-after-free in rpc_read_stream

include/dsn/cpp/rpc_stream.h

@@ -48,12 +48,13 @@ class rpc_read_stream : public binary_reader
     void set_read_msg(message_ex *msg)
     {
         _msg = msg;
+        if (nullptr != _msg) {
+            ::dsn::blob bb;
+            bool r = ((::dsn::message_ex *)_msg)->read_next(bb);
+            dassert(r, "read msg must have one segment of buffer ready");
 
-        ::dsn::blob bb;
-        bool r = ((::dsn::message_ex *)_msg)->read_next(bb);
-        dassert(r, "read msg must have one segment of buffer ready");
-
-        init(std::move(bb));
+            init(std::move(bb));
+        }
     }
 
     ~rpc_read_stream()

src/runtime/rpc/thrift_message_parser.cpp

@@ -111,6 +111,9 @@ static message_ex *create_message_from_request_blob(const blob &body_data)
     if (dsn_hdr->context.u.is_request != 1) {
         derror("invalid message type: %d", mtype);
         delete msg;
+        /// set set rpc_read_stream::_msg to nullptr,
+        /// to avoid the dstor to call read_commit of _msg, which is deleted here.
+        stream.set_read_msg(nullptr);
         return nullptr;
     }
     dsn_hdr->context.u.serialize_format = DSF_THRIFT_BINARY; // always serialize in thrift binary