chore: fix data corruption bug

The bug happenned when an command with less than required arguments was processed during the squashing.

The test:
1. Reproduces the original bug
2. Reproduces the fact that the ERROR in squashing was not handled correctly

The fix: removes the ERROR state, adds a verification step that ensures we fail before accessing non-existing index.
Fixes #6165

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

Author: romange

src/server/main_service.cc

@@ -86,8 +86,6 @@ ABSL_FLAG(uint32_t, memcached_port, 0, "Memcached port");
 
 ABSL_FLAG(uint32_t, num_shards, 0, "Number of database shards, 0 - to choose automatically");
 
-ABSL_RETIRED_FLAG(uint32_t, multi_exec_mode, 2, "DEPRECATED. Sets multi exec atomicity mode");
-
 ABSL_FLAG(bool, multi_exec_squash, true,
           "Whether multi exec will squash single shard commands to optimize performance");
 

src/server/multi_command_squasher.cc

@@ -120,22 +120,27 @@ MultiCommandSquasher::ShardExecInfo& MultiCommandSquasher::PrepareShardInfo(Shar
 MultiCommandSquasher::SquashResult MultiCommandSquasher::TrySquash(const StoredCmd* cmd) {
   DCHECK(cmd->Cid());
 
-  if (!cmd->Cid()->IsTransactional() || (cmd->Cid()->opt_mask() & CO::BLOCKING) ||
-      (cmd->Cid()->opt_mask() & CO::GLOBAL_TRANS))
+  const CommandId& cid = *cmd->Cid();
+  if (!cid.IsTransactional() || (cid.opt_mask() & CO::BLOCKING) ||
+      (cid.opt_mask() & CO::GLOBAL_TRANS))
     return SquashResult::NOT_SQUASHED;
 
-  if (cmd->Cid()->name() == "CLIENT" || cntx_->conn_state.tracking_info_.IsTrackingOn()) {
+  if (cid.name() == "CLIENT" || cntx_->conn_state.tracking_info_.IsTrackingOn()) {
     return SquashResult::NOT_SQUASHED;
   }
 
   auto args = cmd->ArgList(&tmp_keylist_);
   if (args.empty())
     return SquashResult::NOT_SQUASHED;
 
-  auto keys = DetermineKeys(cmd->Cid(), args);
-  if (!keys.ok())
-    return SquashResult::ERROR;
-  if (keys->NumArgs() == 0)
+  // Instead of returning an error, we treat command as non-squashable, allowing the
+  // standalone execution path to handle it.
+  // Validate returns an optional ErrorReply
+  if (cid.Validate(args).has_value())
+    return SquashResult::NOT_SQUASHED;
+
+  auto keys = DetermineKeys(&cid, args);
+  if (!keys.ok() || keys->NumArgs() == 0)
     return SquashResult::NOT_SQUASHED;
 
   // Check if all command keys belong to one shard
@@ -382,9 +387,6 @@ void MultiCommandSquasher::Run(RedisReplyBuilder* rb) {
   for (auto& cmd : cmds_) {
     auto res = TrySquash(&cmd);
 
-    if (res == SquashResult::ERROR)
-      break;
-
     if (res == SquashResult::NOT_SQUASHED || res == SquashResult::SQUASHED_FULL) {
       if (!ExecuteSquashed(rb))
         break;

src/server/multi_command_squasher.h

@@ -66,7 +66,7 @@ class MultiCommandSquasher {
     boost::intrusive_ptr<Transaction> local_tx;  // stub-mode tx for use inside shard
   };
 
-  enum class SquashResult : uint8_t { SQUASHED, SQUASHED_FULL, NOT_SQUASHED, ERROR };
+  enum class SquashResult : uint8_t { SQUASHED, SQUASHED_FULL, NOT_SQUASHED };
 
   MultiCommandSquasher(absl::Span<StoredCmd> cmds, ConnectionContext* cntx, Service* Service,
                        const Opts& opts);

tests/dragonfly/connection_test.py

@@ -1416,7 +1416,6 @@ async def test_client_migrate(df_server: DflyInstance):
     assert resp == 1  # migrated successfully
 
 
-@dfly_args({})
 async def test_issue_5931_malformed_protocol_crash(df_server: DflyInstance):
     """
     Regression test for #5931
@@ -1461,7 +1460,6 @@ async def test_issue_5931_malformed_protocol_crash(df_server: DflyInstance):
     assert await client.ping() == True
 
 
-@dfly_args({})
 async def test_issue_5949_nil_bulk_string_crash(df_server: DflyInstance):
     """
     Regression test for #5949
@@ -1504,3 +1502,20 @@ async def test_issue_5949_nil_bulk_string_crash(df_server: DflyInstance):
     client = df_server.client()
     await client.ping()
     assert await client.ping() == True
+
+
+async def test_issue_6165_squash_invalid_syntax(async_client):
+    pipe = async_client.pipeline(transaction=False)
+    pipe.set("k", "v")
+    pipe.execute_command("RENAME bar")
+    res = await pipe.execute(raise_on_error=False)
+
+    assert res[0] == True  # SET key1
+    assert isinstance(res[1], aioredis.ResponseError)  # INVALID SYNTAX COMMAND
+
+    pip = async_client.pipeline(transaction=False)
+    pip.set("k", "v")
+    pip.execute_command("ZUNION 2 set1")
+    res = await pip.execute(raise_on_error=False)
+    assert res[0] == True  # SET key1
+    assert isinstance(res[1], aioredis.ResponseError)  # INVALID SYNTAX