Ensure process-internal message ID uniqueness

When a storage API command is created internally on a node it is
always assigned a strictly increasing message ID that is guaranteed
to be unique within the process. Some parts of the code use this
as a way to distinguish messages from another. However, uniqueness
(prior to this commit) did not necessarily hold, as the underlying
wire protocol would inherit message IDs _from other nodes_ and override
the generated ID with this. I.e. uniqueness no longer holds.

This had exciting consequences when the stars aligned and a remote
node sent the same ID as one generated at the same time internally
on the receiver node. Luckily, in practice this would only be used
in a potentially ambiguous context when sanity checking shared read
lock sets for the _same bucket_ in the persistence threads. Invariant
checks would detect this is as an attempted duplicate lock acquisition
and abort the process. This has been latent for many, many years,
but we've seen it happen exactly once.

This commit introduces an explicit domain separation between the
node-internal (locally unique) IDs and the ID used by the originator.
The originator ID is maintained and returned over the wire to the
caller when sending a response to the incoming request.

Curiously, we don't actually need this originator ID at all since
the caller maintains explicit state containing the sender command.
Unfortunately we can't simply remove it, since versions prior to
this commit will still use whatever's on the wire.

1 files changed, 18 insertions, 10 deletions

diff --git a/storage/src/vespa/storageapi/mbusprot/protocolserialization7.cpp b/storage/src/vespa/storageapi/mbusprot/protocolserialization7.cpp
index 0ede96179e8..f2dcf07c01a 100644
--- a/storage/src/vespa/storageapi/mbusprot/protocolserialization7.cpp
+++ b/storage/src/vespa/storageapi/mbusprot/protocolserialization7.cpp
@@ -88,7 +88,7 @@ std::shared_ptr<document::Document> get_document(const protobuf::Document& src_d
         vespalib::nbostream doc_buf(src_doc.payload().data(), src_doc.payload().size());
         return std::make_shared<document::Document>(type_repo, doc_buf);
     }
-    return std::shared_ptr<document::Document>();
+    return {};
 }
 
 void set_update(protobuf::Update& dest, const document::DocumentUpdate& src) {
@@ -104,12 +104,17 @@ std::shared_ptr<document::DocumentUpdate> get_update(const protobuf::Update& src
         return document::DocumentUpdate::createHEAD(
                 type_repo, vespalib::nbostream(src.payload().data(), src.payload().size()));
     }
-    return std::shared_ptr<document::DocumentUpdate>();
+    return {};
 }
 
 void write_request_header(vespalib::GrowableByteBuffer& buf, const api::StorageCommand& cmd) {
     protobuf::RequestHeader hdr; // Arena alloc not needed since there are no nested messages
-    hdr.set_message_id(cmd.getMsgId());
+    // TODO deprecate this field entirely; we already match replies with their originating command
+    //   on the sender and set the reply's message ID from that command directly. There's therefore
+    //   no need to redundantly carry this on the wire as well.
+    //   Important: cannot be deprecated until there are no nodes using this value verbatim as the
+    //              internal message ID!
+    hdr.set_message_id(cmd.getMsgId()); // This is _our_ process-internal message ID
     hdr.set_priority(cmd.getPriority());
     hdr.set_source_index(cmd.getSourceIndex());
 
@@ -129,7 +134,8 @@ void write_response_header(vespalib::GrowableByteBuffer& buf, const api::Storage
     if (!result.getMessage().empty()) {
         hdr.set_return_code_message(result.getMessage().data(), result.getMessage().size());
     }
-    hdr.set_message_id(reply.getMsgId());
+    // TODO deprecate this field entirely
+    hdr.set_message_id(reply.originator_msg_id()); // This is the _peer's_ process-internal message ID
     hdr.set_priority(reply.getPriority());
 
     const auto header_size = hdr.ByteSizeLong();
@@ -227,11 +233,11 @@ public:
 
 template <typename ProtobufType>
 class RequestDecoder {
-    protobuf::RequestHeader         _hdr;
-    ::google::protobuf::Arena       _arena;
-    ProtobufType*                   _proto_obj;
+    protobuf::RequestHeader   _hdr;
+    ::google::protobuf::Arena _arena;
+    ProtobufType*             _proto_obj;
 public:
-    RequestDecoder(document::ByteBuffer& in_buf)
+    explicit RequestDecoder(document::ByteBuffer& in_buf)
         : _arena(),
           _proto_obj(::google::protobuf::Arena::Create<ProtobufType>(&_arena))
     {
@@ -246,7 +252,7 @@ public:
     }
 
     void transfer_meta_information_to(api::StorageCommand& dest) {
-        dest.forceMsgId(_hdr.message_id());
+        dest.force_originator_msg_id(_hdr.message_id()); // TODO deprecate
         dest.setPriority(static_cast<uint8_t>(_hdr.priority()));
         dest.setSourceIndex(static_cast<uint16_t>(_hdr.source_index()));
     }
@@ -276,7 +282,9 @@ public:
     }
 
     void transfer_meta_information_to(api::StorageReply& dest) {
-        dest.forceMsgId(_hdr.message_id());
+        // TODO deprecate this; not currently used internally (message ID is explicitly set
+        //  from the originator command instance)
+        dest.force_originator_msg_id(_hdr.message_id());
         dest.setPriority(static_cast<uint8_t>(_hdr.priority()));
         dest.setResult(api::ReturnCode(static_cast<api::ReturnCode::Result>(_hdr.return_code_id()),
                                        _hdr.return_code_message()));