fix(transaction): DCHECK fail in non-atomic transactions (#5217)

Also fixes occasional DCHECK fail with ACL message being sent into a closing connection.
(There is not test as its hard to reproduce with pytests).

The TX bug was that we did not reset references into kv_fp_ array when switching
commands within non-atomic transactions. This resulted in dcheck fails.

Fixes #5212.

The fix:
1. added test that reproduces the issue
2. Reset the fp_start and fp_count fields in each shard data.

Signed-off-by: Roman Gershman <[email protected]>

Author: romange

.clang-tidy

@@ -8,11 +8,13 @@ Checks: >
   -bugprone-easily-swappable-parameters,
   -bugprone-branch-clone,
   -bugprone-implicit-widening-of-multiplication-result,
+  -bugprone-too-small-loop-variable,
+  -bugprone-reserved-identifier,
   boost-use-to-string,
   performance*,
-  cert*,
   -cert-err58-cpp,
   -cert-dcl58-cpp,  # Ignore std changes
+  -cert-dcl51-cpp,  # bugprone-reserved-identifier
   # Doesn't work with abseil flags
   clang-analyzer*,
   google-*,

src/facade/dragonfly_connection.cc

@@ -1809,8 +1809,9 @@ void Connection::SendAsync(MessageHandle msg) {
   DCHECK_EQ(ProactorBase::me(), socket_->proactor());
 
   // "Closing" connections might be still processing commands, as we don't interrupt them.
-  // So we still want to deliver control messages to them (like checkpoints).
-  if (cc_->conn_closing && !msg.IsControl())
+  // So we still want to deliver control messages to them (like checkpoints) if
+  // async_fb_ is running (joinable).
+  if (cc_->conn_closing && (!msg.IsControl() || !async_fb_.IsJoinable()))
     return;
 
   // If we launch while closing, it won't be awaited. Control messages will be processed on cleanup.

src/server/multi_test.cc

@@ -998,6 +998,13 @@ TEST_F(MultiTest, NoKeyTransactional) {
   Run({"exec"});
 }
 
+TEST_F(MultiTest, NoKeyTransactionalMany) {
+  vector<vector<string>> cmds;
+  cmds.push_back({"rename", "x", "z"});
+  cmds.push_back({"ft._list"});
+  RunMany(cmds);
+}
+
 class MultiEvalTest : public BaseFamilyTest {
  protected:
   MultiEvalTest() : BaseFamilyTest() {

src/server/test_utils.cc

@@ -467,6 +467,25 @@ RespExpr BaseFamilyTest::Run(std::string_view id, ArgSlice slice) {
   return e;
 }
 
+void BaseFamilyTest::RunMany(const std::vector<std::vector<std::string>>& cmds) {
+  if (!ProactorBase::IsProactorThread()) {
+    return pp_->at(0)->Await([&] { return this->RunMany(cmds); });
+  }
+  TestConnWrapper* conn_wrapper = AddFindConn(Protocol::REDIS, GetId());
+  auto* context = conn_wrapper->cmd_cntx();
+  context->ns = &namespaces->GetDefaultNamespace();
+  vector<ArgSlice> args_vec(cmds.size());
+  vector<vector<string_view>> cmd_views(cmds.size());
+  for (size_t i = 0; i < cmds.size(); ++i) {
+    for (const auto& arg : cmds[i]) {
+      cmd_views[i].emplace_back(arg);
+    }
+    args_vec[i] = absl::MakeSpan(cmd_views[i]);
+  }
+  service_->DispatchManyCommands(absl::MakeSpan(args_vec), conn_wrapper->builder(), context);
+  DCHECK(context->transaction == nullptr);
+}
+
 auto BaseFamilyTest::RunMC(MP::CmdType cmd_type, string_view key, string_view value, uint32_t flags,
                            chrono::seconds ttl) -> MCResponse {
   if (!ProactorBase::IsProactorThread()) {

src/server/test_utils.h

@@ -85,6 +85,7 @@ class BaseFamilyTest : public ::testing::Test {
   RespExpr Run(absl::Span<std::string> list);
 
   RespExpr Run(std::string_view id, ArgSlice list);
+  void RunMany(const std::vector<std::vector<std::string>>& cmds);
 
   using MCResponse = std::vector<std::string>;
   MCResponse RunMC(MemcacheParser::CmdType cmd_type, std::string_view key, std::string_view value,

src/server/transaction.cc

@@ -512,6 +512,7 @@ void Transaction::MultiSwitchCmd(const CommandId* cid) {
 
   for (auto& sd : shard_data_) {
     sd.slice_count = sd.slice_start = 0;
+    sd.fp_start = sd.fp_count = 0;  // Reset fingerprints span as kv_fp_ was cleared above.
 
     if (multi_->mode == NON_ATOMIC) {
       sd.local_mask = 0;  // Non atomic transactions schedule each time, so remove all flags
@@ -1325,7 +1326,7 @@ OpStatus Transaction::WaitOnWatch(const time_point& tp, WaitKeys wkeys, KeyReady
 
   // If we don't follow up with an "action" hop, we must clean up manually on all shards.
   if (result != OpStatus::OK)
-    ExpireBlocking(std::move(wkeys));
+    ExpireBlocking(wkeys);
 
   return result;
 }
@@ -1515,7 +1516,7 @@ void Transaction::ReviveAutoJournal() {
   re_enabled_auto_journal_ = true;
 }
 
-void Transaction::CancelBlocking(std::function<OpStatus(ArgSlice)> status_cb) {
+void Transaction::CancelBlocking(const std::function<OpStatus(ArgSlice)>& status_cb) {
   // We're on the owning thread of this transaction, so we can safely access it's data below.
   // First, check if it makes sense to proceed.
   if (blocking_barrier_.IsClaimed() || cid_ == nullptr || (cid_->opt_mask() & CO::BLOCKING) == 0)
@@ -1610,7 +1611,7 @@ OpResult<KeyIndex> DetermineKeys(const CommandId* cid, CmdArgList args) {
 
   if (cid->first_key_pos() > 0) {
     start = cid->first_key_pos() - 1;
-    int last = cid->last_key_pos();
+    int8_t last = cid->last_key_pos();
 
     if (num_custom_keys >= 0) {
       end = start + num_custom_keys;

src/server/transaction.h

@@ -225,7 +225,7 @@ class Transaction {
 
   // Cancel all blocking watches. Set COORD_CANCELLED.
   // Must be called from coordinator thread.
-  void CancelBlocking(std::function<OpStatus(ArgSlice)>);
+  void CancelBlocking(const std::function<OpStatus(ArgSlice)>&);
 
   // Prepare a squashed hop on given shards.
   // Only compatible with multi modes that acquire all locks ahead - global and lock_ahead.