fix(namespace): didn't reload namespaces from DB after the full sync (#…

…2527)

Namespaces would store inside DB if the repl-namespace-enabled was set to yes.
Then replicas will intercept and parse the increment replication log to
see if it needs to reload namespaces. But it won't have the namespace update log
when doing the full sync, so we need to reload from DB once the full replication is done.

src/cluster/replication.cc

@@ -732,6 +732,13 @@ ReplicationThread::CBState ReplicationThread::fullSyncReadCB(bufferevent *bev) {
       LOG(INFO) << "[replication] Succeeded restoring the backup, fullsync was finish";
       post_fullsync_cb_();
 
+      // It needs to reload namespaces from DB after the full sync is done,
+      // or namespaces are not visible in the replica.
+      s = srv_->GetNamespace()->LoadAndRewrite();
+      if (!s.IsOK()) {
+        LOG(ERROR) << "[replication] Failed to load and rewrite namespace: " << s.Msg();
+      }
+
       // Switch to psync state machine again
       psync_steps_.Start();
       return CBState::QUIT;

src/server/namespace.cc

@@ -54,7 +54,8 @@ bool Namespace::IsAllowModify() const {
 Status Namespace::loadFromDB(std::map<std::string, std::string>* db_tokens) const {
   std::string value;
   engine::Context ctx(storage_);
-  auto s = storage_->Get(ctx, ctx.GetReadOptions(), cf_, kNamespaceDBKey, &value);
+  auto cf = storage_->GetCFHandle(ColumnFamilyID::Propagate);
+  auto s = storage_->Get(ctx, ctx.GetReadOptions(), cf, kNamespaceDBKey, &value);
   if (!s.ok()) {
     if (s.IsNotFound()) return Status::OK();
     return {Status::NotOK, s.ToString()};

src/server/namespace.h

@@ -26,8 +26,7 @@ constexpr const char *kNamespaceDBKey = "__namespace_keys__";
 
 class Namespace {
  public:
-  explicit Namespace(engine::Storage *storage)
-      : storage_(storage), cf_(storage_->GetCFHandle(ColumnFamilyID::Propagate)) {}
+  explicit Namespace(engine::Storage *storage) : storage_(storage) {}
 
   ~Namespace() = default;
   Namespace(const Namespace &) = delete;
@@ -45,7 +44,6 @@ class Namespace {
 
  private:
   engine::Storage *storage_;
-  rocksdb::ColumnFamilyHandle *cf_ = nullptr;
 
   std::shared_mutex tokens_mu_;
   // mapping from token to namespace name

tests/gocase/unit/namespace/namespace_test.go

@@ -21,6 +21,8 @@ package namespace
 
 import (
 	"context"
+	"fmt"
+	"strings"
 	"sync"
 	"testing"
 	"time"
@@ -252,6 +254,43 @@ func TestNamespaceReplicate(t *testing.T) {
 	})
 }
 
+func TestNamespaceReplicateWithFullSync(t *testing.T) {
+	config := map[string]string{
+		"rocksdb.write_buffer_size":       "4",
+		"rocksdb.target_file_size_base":   "16",
+		"rocksdb.max_write_buffer_number": "1",
+		"rocksdb.wal_ttl_seconds":         "0",
+		"rocksdb.wal_size_limit_mb":       "0",
+		"repl-namespace-enabled":          "yes",
+		"requirepass":                     "123",
+		"masterauth":                      "123",
+	}
+	master := util.StartServer(t, config)
+	defer master.Close()
+	masterClient := master.NewClientWithOption(&redis.Options{Password: "123"})
+	defer func() { require.NoError(t, masterClient.Close()) }()
+
+	slave := util.StartServer(t, config)
+	defer slave.Close()
+	slaveClient := slave.NewClientWithOption(&redis.Options{Password: "123"})
+	defer func() { require.NoError(t, slaveClient.Close()) }()
+
+	ctx := context.Background()
+	value := strings.Repeat("a", 128*1024)
+	for i := 0; i < 1024; i++ {
+		require.NoError(t, masterClient.Set(ctx, fmt.Sprintf("key%d", i), value, 0).Err())
+	}
+	require.NoError(t, masterClient.Do(ctx, "NAMESPACE", "ADD", "foo", "bar").Err())
+
+	util.SlaveOf(t, slaveClient, master)
+	util.WaitForOffsetSync(t, masterClient, slaveClient, 60*time.Second)
+
+	// Namespaces should be replicated after the full sync
+	token, err := slaveClient.Do(ctx, "NAMESPACE", "GET", "foo").Result()
+	require.NoError(t, err)
+	require.EqualValues(t, "bar", token)
+}
+
 func TestNamespaceRewrite(t *testing.T) {
 	password := "pwd"
 	srv := util.StartServerWithCLIOptions(t, false, map[string]string{