Pretty print all MULTI requests in a raft entry with raft-fetch

src/Formatter.cc

@@ -25,6 +25,8 @@
 #include "raft/RaftCommon.hh"
 #include "Common.hh"
 #include "Formatter.hh"
+#include "redis/ArrayResponseBuilder.hh"
+#include "redis/MultiOp.hh"
 using namespace quarkdb;
 
 RedisEncodedResponse Formatter::moved(int64_t shardId, const RaftServer &location) {
@@ -102,18 +104,45 @@ RedisEncodedResponse Formatter::scan(const std::string &marker, const std::vecto
   return RedisEncodedResponse(ss.str());
 }
 
+RedisEncodedResponse Formatter::simpleRedisRequest(const RedisRequest &req) {
+  std::vector<std::string> vec;
+
+  for(size_t i = 0; i < req.size(); i++) {
+    vec.emplace_back(req[i]);
+  }
+
+  return Formatter::statusVector(vec);
+}
+
+RedisEncodedResponse Formatter::redisRequest(const RedisRequest &req) {
+  if(req.getCommand() == RedisCommand::MULTIOP_READWRITE || req.getCommand() == RedisCommand::MULTIOP_READ) {
+    MultiOp multiOp;
+    multiOp.deserialize(req[1]);
+
+    ArrayResponseBuilder builder(multiOp.size() + 1);
+    builder.push_back(Formatter::status(req[0]));
+
+    for(size_t i = 0; i < multiOp.size(); i++) {
+      builder.push_back(simpleRedisRequest(multiOp[i]));
+    }
+
+    return builder.buildResponse();
+  }
+
+  // Simple case, no transactions.
+  return simpleRedisRequest(req);
+}
+
 RedisEncodedResponse Formatter::raftEntry(const RaftEntry &entry) {
   // Very inefficient with copying, but this function is only to help
   // debugging, so we don't really mind.
 
-  std::vector<std::string> vec;
-  vec.emplace_back(std::to_string(entry.term));
+  ArrayResponseBuilder builder(2);
 
-  for(size_t i = 0; i < entry.request.size(); i++) {
-    vec.emplace_back(entry.request[i]);
-  }
+  builder.push_back(Formatter::status(SSTR(entry.term)));
+  builder.push_back(redisRequest(entry.request));
 
-  return Formatter::vector(vec);
+  return builder.buildResponse();
 }
 
 RedisEncodedResponse Formatter::raftEntries(const std::vector<RaftEntry> &entries) {

src/Formatter.hh

@@ -31,6 +31,7 @@
 namespace quarkdb {
 
 class RaftEntry;
+class RedisRequest;
 
 class Formatter {
 public:
@@ -49,6 +50,9 @@ public:
   static RedisEncodedResponse scan(const std::string &marker, const std::vector<std::string> &vec);
   static RedisEncodedResponse raftEntry(const RaftEntry &entry);
   static RedisEncodedResponse raftEntries(const std::vector<RaftEntry> &entries);
+
+  static RedisEncodedResponse simpleRedisRequest(const RedisRequest &req);
+  static RedisEncodedResponse redisRequest(const RedisRequest &req);
 };
 
 }

src/raft/RaftUtils.cc

@@ -265,12 +265,22 @@ bool RaftParser::voteResponse(const redisReplyPtr &source, RaftVoteResponse &des
 }
 
 bool RaftParser::fetchResponse(redisReply *source, RaftEntry &entry) {
-  if(source == nullptr || source->type != REDIS_REPLY_ARRAY || source->elements < 2) {
+  if(source == nullptr || source->type != REDIS_REPLY_ARRAY || source->elements != 2) {
     return false;
   }
 
-  for(size_t i = 0; i < source->elements; i++) {
-    if(source->element[i]->type != REDIS_REPLY_STRING) {
+  if(source->element[0]->type != REDIS_REPLY_STATUS) {
+    return false;
+  }
+
+  if(source->element[1]->type != REDIS_REPLY_ARRAY) {
+    return false;
+  }
+
+  redisReply *req = source->element[1];
+
+  for(size_t i = 0; i < req->elements; i++) {
+    if(req->element[i]->type != REDIS_REPLY_STATUS) {
       return false;
     }
   }
@@ -279,8 +289,8 @@ bool RaftParser::fetchResponse(redisReply *source, RaftEntry &entry) {
   if(!my_strtoll(tmp, entry.term)) return false;
 
   entry.request.clear();
-  for(size_t i = 1; i < source->elements; i++) {
-    entry.request.emplace_back(source->element[i]->str, source->element[i]->len);
+  for(size_t i = 0; i < req->elements; i++) {
+    entry.request.emplace_back(req->element[i]->str, req->element[i]->len);
   }
 
   return true;

test/e2e.cc

@@ -84,7 +84,7 @@ TEST_F(Raft_e2e, simultaneous_clients) {
   futures.emplace_back(tunnel(leaderID)->exec("ping"));
   futures.emplace_back(tunnel(leaderID)->exec("set", "asdf", "1234"));
   futures.emplace_back(tunnel(leaderID)->exec("get", "asdf"));
-  futures.emplace_back(tunnel(leaderID)->exec("raft_fetch", SSTR(lastEntry+1) ));
+  futures.emplace_back(tunnel(leaderID)->exec("raft-fetch", SSTR(lastEntry+1) ));
 
   ASSERT_REPLY(futures[0], "");
   ASSERT_REPLY(futures[1], "PONG");