fix(set_family): Don't replicate SPOP if nothing was removed (#5936)

* fix(set_family): Don't replicate SPOP if nothing was removed

If nothing is removed with SPOP command we are rewriting command as SREM
with invalid syntax. This replicated SREM command fails on replica side
and we don't increment LSN. Fixed by skipping replication if SPOP didn't
remove anything.

Enable logging on replica if command verification failed.

Signed-off-by: mkaruza <[email protected]>

* fix(zset_family): Disable AUTOJOURNAL for ZRANGESTORE

ZRANGESTORE is rewriten as SREM and ZADD so auto journal needs to be
disabled.

* feat(replica): Report DFATAL error only in ACTIVE and RUNNING state

---------

Signed-off-by: mkaruza <[email protected]>

Author: mkaruza

src/server/main_service.cc

@@ -1452,7 +1452,8 @@ DispatchResult Service::DispatchCommand(ArgSlice args, SinkReplyBuilder* builder
   }
 
   if (auto err = VerifyCommandState(cid, args_no_cmd, *dfly_cntx); err) {
-    LOG_IF(WARNING, cntx->replica_conn) << "VerifyCommandState error: " << err->ToSv();
+    LOG_IF(WARNING, cntx->replica_conn || !cntx->conn() /* no owner in replica context */)
+        << "VerifyCommandState error: " << err->ToSv();
     if (auto& exec_info = dfly_cntx->conn_state.exec_info; exec_info.IsCollecting())
       exec_info.state = ConnectionState::ExecInfo::EXEC_ERROR;
 

src/server/replica.cc

@@ -1006,6 +1006,13 @@ void DflyShardReplica::StableSyncDflyReadFb(ExecutionState* cntx) {
         // inconsistent data because the replica will resume from the next
         // lsn of the master and this lsn entry will be lost.
         journal_rec_executed_.fetch_add(1, std::memory_order_relaxed);
+      } else {
+        // We only report DFATAL:
+        // 1. Context is running
+        // 2. We are ACTIVE global state
+        if (cntx->IsRunning() && ((*ServerState::tlocal()).gstate() == GlobalState::ACTIVE)) {
+          LOG(DFATAL) << "ExecuteTx() on replica should be successful.";
+        }
       }
     }
 

src/server/set_family.cc

@@ -610,7 +610,7 @@ OpResult<uint32_t> OpRem(const OpArgs& op_args, string_view key, facade::ArgRang
   if (isempty) {
     db_slice.Del(op_args.db_cntx, find_res->it);
   }
-  if (journal_rewrite && op_args.shard->journal()) {
+  if (removed && journal_rewrite && op_args.shard->journal()) {
     vector<string_view> mapped(vals.Size() + 1);
     mapped[0] = key;
     std::copy(vals.begin(), vals.end(), mapped.begin() + 1);
@@ -950,13 +950,13 @@ OpResult<StringVec> OpPop(const OpArgs& op_args, string_view key, unsigned count
   StringVec result = RandMemberSet(db_cntx, co, generator, picks_count);
 
   // Remove selected members
-  bool is_empty = RemoveSet(db_cntx, result, &co).second;
+  auto [removed, is_empty] = RemoveSet(db_cntx, result, &co);
   find_res->post_updater.Run();
 
   CHECK(!is_empty);
 
   // Replicate as SREM with removed keys, because SPOP is not deterministic.
-  if (op_args.shard->journal()) {
+  if (removed && op_args.shard->journal()) {
     vector<string_view> mapped(result.size() + 1);
     mapped[0] = key;
     copy(result.begin(), result.end(), mapped.begin() + 1);

src/server/zset_family.cc

@@ -2855,7 +2855,8 @@ void ZSetFamily::Register(CommandRegistry* registry) {
       << CI{"ZRANK", CO::READONLY | CO::FAST, -3, 1, 1}.HFUNC(ZRank)
       << CI{"ZRANGEBYLEX", CO::READONLY, -4, 1, 1}.HFUNC(ZRangeByLex)
       << CI{"ZRANGEBYSCORE", CO::READONLY, -4, 1, 1}.HFUNC(ZRangeByScore)
-      << CI{"ZRANGESTORE", CO::WRITE | CO::DENYOOM, -5, 1, 2}.HFUNC(ZRangeStore)
+      << CI{"ZRANGESTORE", CO::WRITE | CO::DENYOOM | CO::NO_AUTOJOURNAL, -5, 1, 2}.HFUNC(
+             ZRangeStore)
       << CI{"ZSCORE", CO::READONLY | CO::FAST, 3, 1, 1}.HFUNC(ZScore)
       << CI{"ZMSCORE", CO::READONLY | CO::FAST, -3, 1, 1}.HFUNC(ZMScore)
       << CI{"ZREMRANGEBYRANK", CO::WRITE, 4, 1, 1}.HFUNC(ZRemRangeByRank)