MB-32573 [SR]: Include VBucket id in DCP seqno_ack/commit/abort

daverigby · daverigby · commit 371fc3587164 · 2019-01-11T20:16:19.000Z
The vBucket was incorrectly missed out from these new Sync Replication DCP messages. Add the vBucketID in, and wire up correcly in ep-engine. Change-Id: I484236d077eea66ef9e0e58e46cbe7c303b74c5c Reviewed-on: http://review.couchbase.org/103557 Tested-by: Build Bot <build@couchbase.com> Reviewed-by: Dave Rigby <daver@couchbase.com>
diff --git a/daemon/connection.cc b/daemon/connection.cc
@@ -2292,6 +2292,7 @@ ENGINE_ERROR_CODE Connection::prepare(uint32_t opaque,
 }
 
 ENGINE_ERROR_CODE Connection::seqno_acknowledged(uint32_t opaque,
+                                                 Vbid vbucket,
                                                  uint64_t in_memory_seqno,
                                                  uint64_t on_disk_seqno) {
     cb::mcbp::request::DcpSeqnoAcknowledgedPayload extras(in_memory_seqno,
@@ -2301,11 +2302,13 @@ ENGINE_ERROR_CODE Connection::seqno_acknowledged(uint32_t opaque,
     builder.setMagic(cb::mcbp::Magic::ClientRequest);
     builder.setOpcode(cb::mcbp::ClientOpcode::DcpSeqnoAcknowledged);
     builder.setOpaque(opaque);
+    builder.setVBucket(vbucket);
     builder.setExtras(extras.getBuffer());
     return add_packet_to_send_pipe(builder.getFrame()->getFrame());
 }
 
 ENGINE_ERROR_CODE Connection::commit(uint32_t opaque,
+                                     Vbid vbucket,
                                      const DocKey& key,
                                      uint64_t commit_seqno) {
     cb::mcbp::request::DcpCommitPayload extras(0, commit_seqno);
@@ -2317,12 +2320,14 @@ ENGINE_ERROR_CODE Connection::commit(uint32_t opaque,
     builder.setMagic(cb::mcbp::Magic::ClientRequest);
     builder.setOpcode(cb::mcbp::ClientOpcode::DcpCommit);
     builder.setOpaque(opaque);
+    builder.setVBucket(vbucket);
     builder.setExtras(extras.getBuffer());
     builder.setKey(cb::const_char_buffer(key));
     return add_packet_to_send_pipe(builder.getFrame()->getFrame());
 }
 
 ENGINE_ERROR_CODE Connection::abort(uint32_t opaque,
+                                    Vbid vbucket,
                                     uint64_t prepared_seqno,
                                     uint64_t abort_seqno) {
     cb::mcbp::request::DcpAbortPayload extras;
@@ -2333,6 +2338,7 @@ ENGINE_ERROR_CODE Connection::abort(uint32_t opaque,
     builder.setMagic(cb::mcbp::Magic::ClientRequest);
     builder.setOpcode(cb::mcbp::ClientOpcode::DcpAbort);
     builder.setOpaque(opaque);
+    builder.setVBucket(vbucket);
     builder.setExtras(extras.getBuffer());
     return add_packet_to_send_pipe(builder.getFrame()->getFrame());
 }
diff --git a/daemon/connection.h b/daemon/connection.h
@@ -932,14 +932,17 @@ class Connection : public dcp_message_producers {
                               cb::durability::Requirements durability) override;
 
     ENGINE_ERROR_CODE seqno_acknowledged(uint32_t opaque,
+                                         Vbid vbucket,
                                          uint64_t in_memory_seqno,
                                          uint64_t on_disk_seqno) override;
 
     ENGINE_ERROR_CODE commit(uint32_t opaque,
+                             Vbid vbucket,
                              const DocKey& key,
                              uint64_t commit_seqno) override;
 
     ENGINE_ERROR_CODE abort(uint32_t opaque,
+                            Vbid vbucket,
                             uint64_t prepared_seqno,
                             uint64_t abort_seqno) override;
 
diff --git a/engines/ep/management/mc_bin_client.py b/engines/ep/management/mc_bin_client.py
@@ -194,13 +194,13 @@ def _sendCmd(self, cmd, key, val, opaque, extraHeader='', cas=0, collection=None
                       vbucketId=self.vbucketId, collection=collection)
 
     def _sendAltCmd(self, cmd, flex, key, val, opaque, extras='', cas=0,
-                    dtype=0, vbucketId=0, collection=None):
+                    dtype=0, collection=None):
         """Send a request in the alternative format supporing flex framing extras"""
         if collection:
             key = self._encodeCollectionId(key, collection)
 
         msg = struct.pack(ALT_REQ_PKT_FMT, ALT_REQ_MAGIC_BYTE, cmd, len(flex),
-                          len(key), len(extras), dtype, vbucketId,
+                          len(key), len(extras), dtype, self.vbucketId,
                           len(flex) + len(key) + len(extras) + len(val),
                           opaque, cas)
         self.s.sendall(msg + flex + extras + key + val)
diff --git a/engines/ep/src/dcp/active_stream.cc b/engines/ep/src/dcp/active_stream.cc
@@ -896,8 +896,11 @@ std::unique_ptr<DcpResponse> ActiveStream::makeResponseFromItem(
     // as a Commit message - otherwise it's just sent as a Mutation.
     if ((item->getOperation() == queue_op::commit_sync_write) &&
         (syncReplication == SyncReplication::Yes)) {
-        return std::make_unique<CommitSyncWrite>(
-                opaque_, 0, item->getBySeqno(), item->getKey());
+        return std::make_unique<CommitSyncWrite>(opaque_,
+                                                 item->getVBucketId(),
+                                                 /*preparedSeqno*/ 0,
+                                                 item->getBySeqno(),
+                                                 item->getKey());
     }
 
     if (item->getOperation() != queue_op::system_event) {
diff --git a/engines/ep/src/dcp/consumer.cc b/engines/ep/src/dcp/consumer.cc
@@ -891,6 +891,7 @@ ENGINE_ERROR_CODE DcpConsumer::step(struct dcp_message_producers* producers) {
         case DcpResponse::Event::SeqnoAcknowledgement: {
             auto* ack = static_cast<SeqnoAcknowledgement*>(resp.get());
             ret = producers->seqno_acknowledged(ack->getOpaque(),
+                                                ack->getVbucket(),
                                                 ack->getInMemorySeqno(),
                                                 ack->getOnDiskSeqno());
             break;
@@ -1663,6 +1664,7 @@ ENGINE_ERROR_CODE DcpConsumer::commit(uint32_t opaque,
         try {
             err = stream->messageReceived(
                     std::make_unique<CommitSyncWrite>(opaque,
+                                                      vbucket,
                                                       /*prepared_seqno*/ 0,
                                                       commit_seqno,
                                                       key));
diff --git a/engines/ep/src/dcp/passive_stream.cc b/engines/ep/src/dcp/passive_stream.cc
@@ -620,7 +620,10 @@ ENGINE_ERROR_CODE PassiveStream::processPrepare(
         LockHolder lh(streamMutex);
         // @todo-durability add in the correct on-disk seqno.
         pushToReadyQ(std::make_unique<SeqnoAcknowledgement>(
-                opaque_, prepare->getItem()->getBySeqno(), 0));
+                opaque_,
+                prepare->getVBucket(),
+                prepare->getItem()->getBySeqno(),
+                0));
     }
     notifyStreamReady();
 
diff --git a/engines/ep/src/dcp/producer.cc b/engines/ep/src/dcp/producer.cc
@@ -635,8 +635,10 @@ ENGINE_ERROR_CODE DcpProducer::step(struct dcp_message_producers* producers) {
         }
         case DcpResponse::Event::Commit: {
             CommitSyncWrite* csr = static_cast<CommitSyncWrite*>(resp.get());
-            ret = producers->commit(
-                    csr->getOpaque(), csr->getKey(), csr->getCommitSeqno());
+            ret = producers->commit(csr->getOpaque(),
+                                    csr->getVbucket(),
+                                    csr->getKey(),
+                                    csr->getCommitSeqno());
             break;
         }
 
diff --git a/engines/ep/src/dcp/response.cc b/engines/ep/src/dcp/response.cc
@@ -123,11 +123,13 @@ std::ostream& operator<<(std::ostream& os, const DcpResponse& r) {
 }
 
 CommitSyncWrite::CommitSyncWrite(uint32_t opaque,
+                                 Vbid vbucket,
                                  uint64_t preparedSeqno,
                                  uint64_t commitSeqno,
                                  const DocKey& key)
     /// @todo-durability: Remove key, use pending seqno to identify instead.
     : DcpResponse(Event::Commit, opaque, cb::mcbp::DcpStreamId{}),
+      vbucket(vbucket),
       key(key),
       payload(preparedSeqno, commitSeqno) {
 }
diff --git a/engines/ep/src/dcp/response.h b/engines/ep/src/dcp/response.h
@@ -580,10 +580,12 @@ class MutationConsumerMessage : public MutationResponse {
 class SeqnoAcknowledgement : public DcpResponse {
 public:
     SeqnoAcknowledgement(uint32_t opaque,
+                         Vbid vbucket,
                          uint64_t inMemorySeqno,
                          uint64_t onDiskSeqno)
         : DcpResponse(
                   Event::SeqnoAcknowledgement, opaque, cb::mcbp::DcpStreamId{}),
+          vbucket(vbucket),
           payload(inMemorySeqno, onDiskSeqno) {
     }
 
@@ -592,6 +594,10 @@ class SeqnoAcknowledgement : public DcpResponse {
                sizeof(cb::mcbp::request::DcpSeqnoAcknowledgedPayload);
     }
 
+    Vbid getVbucket() const {
+        return vbucket;
+    }
+
     uint64_t getInMemorySeqno() const {
         return payload.getInMemorySeqno();
     }
@@ -601,6 +607,7 @@ class SeqnoAcknowledgement : public DcpResponse {
     }
 
 private:
+    Vbid vbucket;
     cb::mcbp::request::DcpSeqnoAcknowledgedPayload payload;
 };
 
@@ -610,6 +617,7 @@ class SeqnoAcknowledgement : public DcpResponse {
 class CommitSyncWrite : public DcpResponse {
 public:
     CommitSyncWrite(uint32_t opaque,
+                    Vbid vbucket,
                     uint64_t preparedSeqno,
                     uint64_t commitSeqno,
                     const DocKey& key);
@@ -624,6 +632,10 @@ class CommitSyncWrite : public DcpResponse {
         return key;
     }
 
+    Vbid getVbucket() const {
+        return vbucket;
+    }
+
     uint64_t getPreparedSeqno() const {
         return payload.getPreparedSeqno();
     }
@@ -637,6 +649,7 @@ class CommitSyncWrite : public DcpResponse {
             sizeof(cb::mcbp::request::DcpCommitPayload);
 
 private:
+    Vbid vbucket;
     StoredDocKey key;
     cb::mcbp::request::DcpCommitPayload payload;
 };
diff --git a/engines/ep/tests/mock/mock_dcp.h b/engines/ep/tests/mock/mock_dcp.h
@@ -139,16 +139,19 @@ class MockDcpMessageProducers : public dcp_message_producers {
                               cb::durability::Requirements durability) override;
 
     ENGINE_ERROR_CODE seqno_acknowledged(uint32_t opaque,
+                                         Vbid vbucket,
                                          uint64_t in_memory_seqno,
                                          uint64_t on_disk_seqno) override {
         return ENGINE_ENOTSUP;
     }
     ENGINE_ERROR_CODE commit(uint32_t opaque,
+                             Vbid vbucket,
                              const DocKey& key,
                              uint64_t commit_seqno) override {
         return ENGINE_ENOTSUP;
     }
     ENGINE_ERROR_CODE abort(uint32_t opaque,
+                            Vbid vbucket,
                             uint64_t prepared_seqno,
                             uint64_t abort_seqno) override {
         return ENGINE_ENOTSUP;
diff --git a/include/memcached/dcp.h b/include/memcached/dcp.h
@@ -321,10 +321,12 @@ struct dcp_message_producers {
      * messages up to the specified seqno.
      *
      * @param opaque identifying stream
+     * @param vbucket the vbucket the seqno ack is for
      * @param in_memory_seqno
      * @param on_disk_seqno
      */
     virtual ENGINE_ERROR_CODE seqno_acknowledged(uint32_t opaque,
+                                                 Vbid vbucket,
                                                  uint64_t in_memory_seqno,
                                                  uint64_t on_disk_seqno) = 0;
 
@@ -336,13 +338,15 @@ struct dcp_message_producers {
      * KV-Engine replicas of a committed Sync Write
      *
      * @param opaque
+     * @param vbucket the vbucket the event applies to
      * @param key The key of the committed mutation.
      * @todo-durability change from identifying by key to the pendingSeqno -
      * more efficient over the wire.
      * @param commit_seqno The sequence number to commit this mutation at.
      * @return
      */
     virtual ENGINE_ERROR_CODE commit(uint32_t opaque,
+                                     Vbid vbucket,
                                      const DocKey& key,
                                      uint64_t commit_seqno) = 0;
     /**
@@ -351,6 +355,7 @@ struct dcp_message_producers {
      * This is sent from the DCP Producer to the DCP Consumer.
      */
     virtual ENGINE_ERROR_CODE abort(uint32_t opaque,
+                                    Vbid vbucket,
                                     uint64_t prepared_seqno,
                                     uint64_t abort_seqno) = 0;
 };

Original file line number	Diff line number	Diff line change
`@@ -635,8 +635,10 @@ ENGINE_ERROR_CODE DcpProducer::step(struct dcp_message_producers* producers) {`
`635`	`635`	`}`
`636`	`636`	`case DcpResponse::Event::Commit: {`
`637`	`637`	`CommitSyncWrite* csr = static_cast<CommitSyncWrite*>(resp.get());`
`638`		`- ret = producers->commit(`
`639`		`- csr->getOpaque(), csr->getKey(), csr->getCommitSeqno());`
	`638`	`+ ret = producers->commit(csr->getOpaque(),`
	`639`	`+ csr->getVbucket(),`
	`640`	`+ csr->getKey(),`
	`641`	`+ csr->getCommitSeqno());`
`640`	`642`	`break;`
`641`	`643`	`}`
`642`	`644`
Original file line number	Diff line number	Diff line change
`@@ -123,11 +123,13 @@ std::ostream& operator<<(std::ostream& os, const DcpResponse& r) {`
`123`	`123`	`}`
`124`	`124`
`125`	`125`	`CommitSyncWrite::CommitSyncWrite(uint32_t opaque,`
	`126`	`+ Vbid vbucket,`
`126`	`127`	`uint64_t preparedSeqno,`
`127`	`128`	`uint64_t commitSeqno,`
`128`	`129`	`const DocKey& key)`
`129`	`130`	`/// @todo-durability: Remove key, use pending seqno to identify instead.`
`130`	`131`	`: DcpResponse(Event::Commit, opaque, cb::mcbp::DcpStreamId{}),`
	`132`	`+ vbucket(vbucket),`
`131`	`133`	`key(key),`
`132`	`134`	`payload(preparedSeqno, commitSeqno) {`
`133`	`135`	`}`