feat: let pika slave support Redis transaction

include/pika_dispatch_thread.h

@@ -23,6 +23,7 @@ class PikaDispatchThread {
   void SetQueueLimit(int queue_limit) { thread_rep_->SetQueueLimit(queue_limit); }
 
   void UnAuthUserAndKillClient(const std::set<std::string> &users, const std::shared_ptr<User>& defaultUser);
+  net::ServerThread* server_thread() { return thread_rep_; }
 
  private:
   class ClientConnFactory : public net::ConnFactory {

include/pika_server.h

@@ -105,6 +105,9 @@ class PikaServer : public pstd::noncopyable {
   void SetForceFullSync(bool v);
   void SetDispatchQueueLimit(int queue_limit);
   storage::StorageOptions storage_options();
+  std::unique_ptr<PikaDispatchThread>& pika_dispatch_thread() {
+    return pika_dispatch_thread_;
+  }
 
   /*
    * DB use

src/pika_client_conn.cc

@@ -114,6 +114,11 @@ std::shared_ptr<Cmd> PikaClientConn::DoCmd(const PikaCmdArgsType& argv, const st
   }
 
   if (IsInTxn() && opt != kCmdNameExec && opt != kCmdNameWatch && opt != kCmdNameDiscard && opt != kCmdNameMulti) {
+    if (c_ptr->is_write() && g_pika_server->readonly(current_db_)) {
+      SetTxnInitFailState(true);
+      c_ptr->res().SetRes(CmdRes::kErrOther, "READONLY You can't write against a read only replica.");
+      return c_ptr;
+    }
     PushCmdToQue(c_ptr);
     c_ptr->res().SetRes(CmdRes::kTxnQueued);
     return c_ptr;
@@ -158,8 +163,8 @@ std::shared_ptr<Cmd> PikaClientConn::DoCmd(const PikaCmdArgsType& argv, const st
       c_ptr->res().SetRes(CmdRes::kErrOther, "Internal ERROR");
       return c_ptr;
     }
-    if (g_pika_server->readonly(current_db_)) {
-      c_ptr->res().SetRes(CmdRes::kErrOther, "Server in read-only");
+    if (g_pika_server->readonly(current_db_) && opt != kCmdNameExec) {
+      c_ptr->res().SetRes(CmdRes::kErrOther, "READONLY You can't write against a read only replica.");
       return c_ptr;
     }
   } else if (c_ptr->is_read() && c_ptr->flag_ == 0) {

src/pika_repl_bgworker.cc

@@ -238,6 +238,24 @@ void PikaReplBgWorker::HandleBGWorkerWriteDB(void* arg) {
   if (!c_ptr->IsSuspend()) {
     c_ptr->GetDB()->DBUnlockShared();
   }
+
+  if (c_ptr->res().ok()
+      && c_ptr->is_write()
+      && c_ptr->name() != kCmdNameFlushdb
+      && c_ptr->name() != kCmdNameFlushall
+      && c_ptr->name() != kCmdNameExec) {
+    auto table_keys = c_ptr->current_key();
+    for (auto& key : table_keys) {
+      key = c_ptr->db_name().append(key);
+    }
+    auto dispatcher = dynamic_cast<net::DispatchThread*>(g_pika_server->pika_dispatch_thread()->server_thread());
+    auto involved_conns = dispatcher->GetInvolvedTxn(table_keys);
+    for (auto& conn : involved_conns) {
+      auto c = std::dynamic_pointer_cast<PikaClientConn>(conn);
+      c->SetTxnWatchFailState(true);
+    }
+  }
+
   record_lock.Unlock(c_ptr->current_key());
   if (g_pika_conf->slowlog_slower_than() >= 0) {
     auto start_time = static_cast<int32_t>(start_us / 1000000);

src/pika_transaction.cc

@@ -25,7 +25,7 @@ void MultiCmd::Do() {
     return;
   }
   if (client_conn->IsInTxn()) {
-    res_.SetRes(CmdRes::kErrOther, "ERR MULTI calls can not be nested");
+    res_.SetRes(CmdRes::kErrOther, "MULTI calls can not be nested");
     return;
   }
   client_conn->SetTxnStartState(true);
@@ -93,7 +93,7 @@ void ExecCmd::Execute() {
     return;
   }
   if (!client_conn->IsInTxn()) {
-    res_.SetRes(CmdRes::kErrOther, "ERR EXEC without MULTI");
+    res_.SetRes(CmdRes::kErrOther, "EXEC without MULTI");
     return;
   }
   if (IsTxnFailedAndSetState()) {
@@ -238,7 +238,7 @@ void WatchCmd::Do() {
     return;
   }
   if (client_conn->IsInTxn()) {
-    res_.SetRes(CmdRes::CmdRet::kErrOther, "ERR WATCH inside MULTI is not allowed");
+    res_.SetRes(CmdRes::CmdRet::kErrOther, "WATCH inside MULTI is not allowed");
     return;
   }
   client_conn->AddKeysToWatch(db_keys_);
@@ -301,7 +301,7 @@ void DiscardCmd::Do() {
     return;
   }
   if (!client_conn->IsInTxn()) {
-    res_.SetRes(CmdRes::kErrOther, "ERR DISCARD without MULTI");
+    res_.SetRes(CmdRes::kErrOther, "DISCARD without MULTI");
     return;
   }
   client_conn->ExitTxn();

tests/integration/replication_test.go

@@ -421,7 +421,7 @@ var _ = Describe("should replication ", func() {
 			Expect(infoRes.Val()).To(ContainSubstring("connected_slaves:1"))
 
 			slaveWrite := clientSlave.Set(ctx, "foo", "bar", 0)
-			Expect(slaveWrite.Err()).To(MatchError("ERR Server in read-only"))
+			Expect(slaveWrite.Err()).To(MatchError("ERR READONLY You can't write against a read only replica."))
 
 			log.Println("Replication test 1 start")
 			err1 := clientMaster.SetEx(ctx, "key", "hello", 60*time.Second).Err()
@@ -638,6 +638,99 @@ var _ = Describe("should replication ", func() {
 				Expect(clientMaster.LIndex(ctx, "blist0", i)).To(Equal(clientSlave.LIndex(ctx, "blist0", i)))
 			}
 			err = clientMaster.Del(ctx, lists...)
+
+			// transaction replication test
+			log.Println("transaction replication test start")
+			clientMaster.Set(ctx, "txkey1", "txvalue1", 0)
+			time.Sleep(time.Second)
+
+			// transaction: multi/get/get/exec
+			r := clientSlave.Do(ctx, "MULTI")
+			Expect(r.Err()).NotTo(HaveOccurred())
+			get := clientSlave.Get(ctx, "txkey1")
+			Expect(get.Err()).NotTo(HaveOccurred())
+			Expect(get.Val()).To(Equal("QUEUED"))
+			get = clientSlave.Get(ctx, "txkey2")
+			Expect(get.Err()).NotTo(HaveOccurred())
+			Expect(get.Val()).To(Equal("QUEUED"))
+			r = clientSlave.Do(ctx, "EXEC")
+			Expect(r.Err()).NotTo(HaveOccurred())
+			Expect(r.Val()).To(Equal([]interface{}{"txvalue1", nil}))
+
+			// transaction: multi/get/get/exec
+			pipeline := clientSlave.TxPipeline()
+			pipeline.Get(ctx, "txkey1")
+			pipeline.Get(ctx, "txkey2")
+			result, perr := pipeline.Exec(ctx)
+			Expect(perr).To(Equal(redis.Nil))
+			AssertEqualRedisString("txvalue1", result[0])
+			Expect(result[1].Err()).To(Equal(redis.Nil))
+
+			// transaction: multi/get/set/exec
+			r = clientSlave.Do(ctx, "MULTI")
+			Expect(r.Err()).NotTo(HaveOccurred())
+			get = clientSlave.Get(ctx, "txkey1")
+			Expect(get.Err()).NotTo(HaveOccurred())
+			Expect(get.Val()).To(Equal("QUEUED"))
+			set = clientSlave.Set(ctx, "txkey2", "txvalue2", 0)
+			Expect(set.Err().Error()).To(Equal("ERR READONLY You can't write against a read only replica."))
+			r = clientSlave.Do(ctx, "EXEC")
+			Expect(r.Err().Error()).To(Equal("EXECABORT Transaction discarded because of previous errors."))
+
+			// transaction: multi/get/set/exec
+			pipeline = clientSlave.TxPipeline()
+			pipeline.Get(ctx, "txkey1")
+			pipeline.Set(ctx, "txkey2", "txvalue2", 0)
+			result, perr = pipeline.Exec(ctx)
+			Expect(perr.Error()).To(Equal("EXECABORT Transaction discarded because of previous errors."))
+
+			// transaction: watch/multi/master-set/exec
+			r = clientSlave.Do(ctx, "WATCH", "txkey1")
+			Expect(r.Err()).NotTo(HaveOccurred())
+			r = clientSlave.Do(ctx, "MULTI")
+			Expect(r.Err()).NotTo(HaveOccurred())
+			set = clientMaster.Set(ctx, "txkey1", "txvalue11", 0)
+			Expect(set.Err()).NotTo(HaveOccurred())
+			Expect(set.Val()).To(Equal("OK"))
+			r = clientSlave.Do(ctx, "EXEC")
+			Expect(r.Err()).NotTo(HaveOccurred())
+			Expect(r.Val()).To(Equal([]interface{}{}))
+
+			// transaction: multi/get/discard
+			r = clientSlave.Do(ctx, "MULTI")
+			Expect(r.Err()).NotTo(HaveOccurred())
+			get = clientSlave.Get(ctx, "txkey1")
+			Expect(get.Err()).NotTo(HaveOccurred())
+			Expect(get.Val()).To(Equal("QUEUED"))
+			r = clientSlave.Do(ctx, "DISCARD")
+			Expect(r.Err()).NotTo(HaveOccurred())
+			Expect(r.Val()).To(Equal("OK"))
+
+			// transaction: watch/unwatch
+			r = clientSlave.Do(ctx, "WATCH", "txkey1")
+			Expect(r.Err()).NotTo(HaveOccurred())
+			Expect(r.Val()).To(Equal("OK"))
+			r = clientSlave.Do(ctx, "UNWATCH")
+			Expect(r.Err()).NotTo(HaveOccurred())
+			Expect(r.Val()).To(Equal("OK"))
+
+			// transaction: times-multi/get/exec
+			r = clientSlave.Do(ctx, "MULTI")
+			Expect(r.Err()).NotTo(HaveOccurred())
+			r = clientSlave.Do(ctx, "MULTI")
+			Expect(r.Err()).To(MatchError("ERR MULTI calls can not be nested"))
+			get = clientSlave.Get(ctx, "txkey1")
+			Expect(get.Err()).NotTo(HaveOccurred())
+			Expect(get.Val()).To(Equal("QUEUED"))
+			r = clientSlave.Do(ctx, "EXEC")
+			Expect(r.Err()).NotTo(HaveOccurred())
+			Expect(r.Val()).To(Equal([]interface{}{"txvalue11"}))
+
+			// transaction: exec without multi
+			r = clientSlave.Do(ctx, "EXEC")
+			Expect(r.Err()).To(MatchError("ERR EXEC without MULTI"))
+
+			err = clientMaster.Del(ctx, "txkey1")
 			log.Println("master-slave replication test success")
 		})