MB-47318: [BP] Add DcpConsumer related client connection methods

Add methods so a test can create and drive a DCP consumer with
snapshot/mutate/delete.

The formats of the snapshot and delete match the format used by
a current producer (the V2 variants).

Prepare/Commit not added as they're out of scope for the test
which will utilise these.

Change-Id: Ibd2ccce7c30b13ebb82b36de32273a2ea3500f21
Reviewed-on: https://review.couchbase.org/c/kv_engine/+/165017
Well-Formed: Restriction Checker
Tested-by: Build Bot <[email protected]>
Reviewed-by: Dave Rigby <[email protected]>

Author: jimwwalker

protocol/connection/client_connection.cc

@@ -19,6 +19,7 @@
 #include "frameinfo.h"
 
 #include <cbsasl/client.h>
+#include <mcbp/dcp_snapshot_marker_codec.h>
 #include <mcbp/mcbp.h>
 #include <mcbp/protocol/framebuilder.h>
 #include <memcached/protocol_binary.h>
@@ -741,7 +742,7 @@ void MemcachedConnection::recvFrame(Frame& frame) {
     }
 }
 
-void MemcachedConnection::sendCommand(const BinprotCommand& command) {
+size_t MemcachedConnection::sendCommand(const BinprotCommand& command) {
     traceData.reset();
 
     auto encoded = command.encode();
@@ -765,14 +766,17 @@ void MemcachedConnection::sendCommand(const BinprotCommand& command) {
     iovec iov{};
     iov.iov_base = encoded.header.data();
     iov.iov_len = encoded.header.size();
+    size_t sentBytes = iov.iov_len;
     message.push_back(iov);
     for (auto buf : encoded.bufs) {
         iov.iov_base = const_cast<uint8_t*>(buf.data());
         iov.iov_len = buf.size();
+        sentBytes += iov.iov_len;
         message.push_back(iov);
     }
 
     sendBuffer(message);
+    return sentBytes;
 }
 
 void MemcachedConnection::recvResponse(BinprotResponse& response) {
@@ -1484,6 +1488,158 @@ Document MemcachedConnection::getRandomKey(Vbid vbucket) {
     return ret;
 }
 
+void MemcachedConnection::dcpOpenProducer(const std::string& name) {
+    BinprotDcpOpenCommand open{name,
+                               cb::mcbp::request::DcpOpenPayload::Producer};
+    const auto response = BinprotResponse(execute(open));
+    if (!response.isSuccess()) {
+        throw ConnectionError("Failed dcpOpenProducer", response);
+    }
+}
+
+void MemcachedConnection::dcpOpenConsumer(const std::string& name) {
+    BinprotDcpOpenCommand open{name};
+    const auto response = BinprotResponse(execute(open));
+    if (!response.isSuccess()) {
+        throw ConnectionError("Failed dcpOpenConsumer", response);
+    }
+}
+
+void MemcachedConnection::dcpControl(const std::string& key,
+                                     const std::string& value) {
+    BinprotDcpControlCommand control;
+    control.setKey(key);
+    control.setValue(value);
+    const auto response = BinprotResponse(execute(control));
+    if (!response.isSuccess()) {
+        throw ConnectionError("Failed dcpControl", response);
+    }
+}
+
+void MemcachedConnection::dcpStreamRequest(Vbid vbid,
+                                           uint32_t flags,
+                                           uint64_t startSeq,
+                                           uint64_t endSeq,
+                                           uint64_t vbUuid,
+                                           uint64_t snapStart,
+                                           uint64_t snapEnd) {
+    BinprotDcpStreamRequestCommand stream(
+            vbid, flags, startSeq, endSeq, vbUuid, snapStart, snapEnd);
+    const auto response = BinprotResponse(execute(stream));
+    if (!response.isSuccess()) {
+        throw ConnectionError("Failed dcpStreamRequest", response);
+    }
+}
+
+void MemcachedConnection::dcpAddStream(Vbid vbid, uint32_t flags) {
+    sendCommand(BinprotDcpAddStreamCommand{flags}.setVBucket(vbid));
+}
+
+void MemcachedConnection::dcpStreamRequestResponse(
+        uint32_t opaque,
+        const std::vector<std::pair<uint64_t, uint64_t>>& failovers) {
+    BinprotCommandResponse rsp{cb::mcbp::ClientOpcode::DcpStreamReq, opaque};
+
+    // Turn the vector of pairs into a protocol failover table (in a string
+    // so we can attach to the response)
+    std::string table;
+    for (const auto& entry : failovers) {
+        auto wireUuid = htonll(entry.first);
+        auto wireSeqno = htonll(entry.second);
+
+        std::copy_n(reinterpret_cast<uint8_t*>(&wireUuid),
+                    sizeof(uint64_t),
+                    std::back_inserter(table));
+
+        std::copy_n(reinterpret_cast<uint8_t*>(&wireSeqno),
+                    sizeof(uint64_t),
+                    std::back_inserter(table));
+    }
+
+    rsp.setValue(table);
+    sendCommand(rsp);
+}
+
+size_t MemcachedConnection::dcpSnapshotMarkerV2(uint32_t opaque,
+                                                uint64_t start,
+                                                uint64_t end,
+                                                uint32_t flags) {
+    const auto size = sizeof(cb::mcbp::Request) +
+                      sizeof(cb::mcbp::request::DcpSnapshotMarkerV2xPayload) +
+                      sizeof(cb::mcbp::request::DcpSnapshotMarkerV2_0Value);
+    Frame buffer;
+    buffer.payload.resize(size);
+
+    cb::mcbp::FrameBuilder<cb::mcbp::Request> builder(
+            {buffer.payload.data(), buffer.payload.size()});
+    builder.setMagic(cb::mcbp::Magic::ClientRequest);
+    builder.setOpcode(cb::mcbp::ClientOpcode::DcpSnapshotMarker);
+    builder.setOpaque(opaque);
+
+    cb::mcbp::encodeDcpSnapshotMarker(builder, start, end, flags, {}, end);
+    sendFrame(buffer);
+    return buffer.payload.size();
+}
+
+size_t MemcachedConnection::dcpMutation(const Document& doc,
+                                        uint32_t opaque,
+                                        uint64_t seqno,
+                                        uint64_t revSeqno,
+                                        uint32_t lockTime,
+                                        uint8_t nru) {
+    // No reply expected
+    return sendCommand(BinprotDcpMutationCommand{doc.info.id,
+                                                 doc.value,
+                                                 opaque,
+                                                 uint8_t(doc.info.datatype),
+                                                 doc.info.expiration,
+                                                 doc.info.cas,
+                                                 seqno,
+                                                 revSeqno,
+                                                 doc.info.flags,
+                                                 lockTime,
+                                                 nru});
+}
+
+size_t MemcachedConnection::dcpDeletionV2(const Document& doc,
+                                          uint32_t opaque,
+                                          uint64_t seqno,
+                                          uint64_t revSeqno,
+                                          uint32_t deleteTime) {
+    // No reply expected
+    return sendCommand(BinprotDcpDeletionV2Command{doc.info.id,
+                                                   doc.value,
+                                                   opaque,
+                                                   uint8_t(doc.info.datatype),
+                                                   doc.info.cas,
+                                                   seqno,
+                                                   revSeqno,
+                                                   deleteTime});
+}
+
+void MemcachedConnection::recvDcpBufferAck(uint32_t expected) {
+    Frame frame;
+    recvFrame(frame);
+    const auto* request = frame.getRequest();
+    if (request->getClientOpcode() !=
+        cb::mcbp::ClientOpcode::DcpBufferAcknowledgement) {
+        throw std::logic_error(
+                "MemcachedConnection::recvDcpBufferAck not a buffer ack "
+                "opcode");
+    }
+    auto* dcpBufferAck =
+            reinterpret_cast<const cb::mcbp::request::DcpBufferAckPayload*>(
+                    request->getExtdata().data());
+
+    if (dcpBufferAck->getBufferBytes() != expected) {
+        throw std::logic_error(
+                "MemcachedConnection::recvDcpBufferAck: Unexpected buffer "
+                "bytes:" +
+                std::to_string(dcpBufferAck->getBufferBytes()) +
+                " expected:" + std::to_string(expected));
+    }
+}
+
 void MemcachedConnection::setUnorderedExecutionMode(ExecutionMode mode) {
     switch (mode) {
     case ExecutionMode::Ordered:

protocol/connection/client_connection.h

@@ -576,7 +576,7 @@ class MemcachedConnection {
      */
     void recvFrame(Frame& frame);
 
-    void sendCommand(const BinprotCommand& command);
+    size_t sendCommand(const BinprotCommand& command);
 
     void recvResponse(BinprotResponse& response);
 
@@ -860,6 +860,58 @@ class MemcachedConnection {
 
     Document getRandomKey(Vbid vbid);
 
+    void dcpOpenProducer(const std::string& name);
+    void dcpOpenConsumer(const std::string& name);
+    void dcpControl(const std::string& key, const std::string& value);
+    void dcpStreamRequest(Vbid vbid,
+                          uint32_t flags,
+                          uint64_t startSeq,
+                          uint64_t endSeq,
+                          uint64_t vbUuid,
+                          uint64_t snapStart,
+                          uint64_t snapEnd);
+    void dcpStreamRequest(Vbid vbid,
+                          uint32_t flags,
+                          uint64_t startSeq,
+                          uint64_t endSeq,
+                          uint64_t vbUuid,
+                          uint64_t snapStart,
+                          uint64_t snapEnd,
+                          const nlohmann::json& value);
+
+    /* following dcp functions are for working with a consumer */
+    void dcpAddStream(Vbid vbid, uint32_t flags = 0);
+
+    /**
+     * Send a success response for a DcpStreamRequest
+     * Includes a value encoding a failover table
+     * @param opaque request/response opaque
+     * @param failovers vector of pair representing failover table. The pair
+     *        encodes first = uuid, second = seqno
+     */
+    void dcpStreamRequestResponse(
+            uint32_t opaque,
+            const std::vector<std::pair<uint64_t, uint64_t>>& failovers);
+    /**
+     * Send the V2 marker with max visible seqno set to end
+     */
+    size_t dcpSnapshotMarkerV2(uint32_t opaque,
+                               uint64_t start,
+                               uint64_t end,
+                               uint32_t flags);
+    size_t dcpMutation(const Document& doc,
+                       uint32_t opaque,
+                       uint64_t seqno,
+                       uint64_t revSeqno = 0,
+                       uint32_t lockTime = 0,
+                       uint8_t nru = 0);
+    size_t dcpDeletionV2(const Document& doc,
+                         uint32_t opaque,
+                         uint64_t seqno,
+                         uint64_t revSeqno = 0,
+                         uint32_t deleteTime = 0);
+    void recvDcpBufferAck(uint32_t expected);
+
 protected:
     void read(Frame& frame, size_t bytes);