fix(stream): harden XACKDEL parsing and replication

kirito632 · kirito632 · commit f55dbb378ca7 · 2026-06-19T11:48:10.000+08:00
diff --git a/src/cluster/batch_sender.cc b/src/cluster/batch_sender.cc
@@ -38,6 +38,7 @@ Status BatchSender::Put(rocksdb::ColumnFamilyHandle *cf, const rocksdb::Slice &k
     return {Status::NotOK, fmt::format("failed to put key value to migration batch, {}", s.ToString())};
   }
 
+  pending_logdata_only_ = false;
   pending_entries_++;
   entries_num_++;
   return Status::OK();
@@ -48,6 +49,7 @@ Status BatchSender::Delete(rocksdb::ColumnFamilyHandle *cf, const rocksdb::Slice
   if (!s.ok()) {
     return {Status::NotOK, fmt::format("failed to delete key from migration batch, {}", s.ToString())};
   }
+  pending_logdata_only_ = false;
   pending_entries_++;
   entries_num_++;
   return Status::OK();
@@ -58,6 +60,7 @@ Status BatchSender::PutLogData(const rocksdb::Slice &blob) {
   if (!s.ok()) {
     return {Status::NotOK, fmt::format("failed to put log data to migration batch, {}", s.ToString())};
   }
+  pending_logdata_only_ = true;
   pending_entries_++;
   entries_num_++;
   return Status::OK();
@@ -89,6 +92,7 @@ Status BatchSender::Send() {
   sent_bytes_ += write_batch_.GetDataSize();
   sent_batches_num_++;
   pending_entries_ = 0;
+  pending_logdata_only_ = false;
   write_batch_.Clear();
   return Status::OK();
 }
diff --git a/src/cluster/batch_sender.h b/src/cluster/batch_sender.h
@@ -46,7 +46,9 @@ class BatchSender {
   void SetMaxBytes(size_t max_bytes) {
     if (max_bytes_ != max_bytes) max_bytes_ = max_bytes;
   }
-  bool IsFull() const { return write_batch_.GetDataSize() >= max_bytes_; }
+  // A log-data-only batch must wait for the first data record;
+  // otherwise the receiver would see metadata without matching writes.
+  bool IsFull() const { return !pending_logdata_only_ && write_batch_.GetDataSize() >= max_bytes_; }
   uint64_t GetSentBytes() const { return sent_bytes_; }
   uint32_t GetSentBatchesNum() const { return sent_batches_num_; }
   uint32_t GetEntriesNum() const { return entries_num_; }
@@ -62,6 +64,7 @@ class BatchSender {
   uint32_t sent_batches_num_ = 0;
   uint32_t entries_num_ = 0;
   uint32_t pending_entries_ = 0;
+  bool pending_logdata_only_ = false;
 
   int dst_fd_;
   size_t max_bytes_;
diff --git a/src/commands/cmd_stream.cc b/src/commands/cmd_stream.cc
@@ -22,6 +22,7 @@
 #include <limits>
 #include <memory>
 #include <stdexcept>
+#include <string_view>
 
 #include "command_parser.h"
 #include "commander.h"
@@ -49,6 +50,48 @@ CommandKeyRange ParseStreamReadRange(const std::vector<std::string> &args, uint3
   range.last_key = range.first_key + stream_size - 1;
   return range;
 }
+
+// Redis accepts only canonical positive decimal numids here;
+// reject forms like +1 or 01 before integer parsing.
+bool IsXAckDelNumIDs(std::string_view input) {
+  if (input.empty() || input[0] < '1' || input[0] > '9') return false;
+
+  return std::all_of(input.begin() + 1, input.end(), [](char c) { return c >= '0' && c <= '9'; });
+}
+
+StatusOr<uint64_t> ParseXAckDelStreamEntryIDComponent(std::string_view input, bool allow_negative_zero) {
+  if (input.empty()) return {Status::RedisParseErr, redis::kErrInvalidEntryIdSpecified};
+
+  if (input[0] == '+') {
+    input.remove_prefix(1);
+  } else if (input[0] == '-') {
+    if (!allow_negative_zero) return {Status::RedisParseErr, redis::kErrInvalidEntryIdSpecified};
+    input.remove_prefix(1);
+    if (input.empty() || !std::all_of(input.begin(), input.end(), [](char c) { return c == '0'; })) {
+      return {Status::RedisParseErr, redis::kErrInvalidEntryIdSpecified};
+    }
+    return 0;
+  }
+
+  auto parsed = ParseInt<uint64_t>(input, 10);
+  if (!parsed) return {Status::RedisParseErr, redis::kErrInvalidEntryIdSpecified};
+  return *parsed;
+}
+
+Status ParseXAckDelStreamEntryID(const std::string &input, redis::StreamEntryID *id) {
+  auto pos = input.find('-');
+  if (pos != std::string::npos) {
+    auto ms = GET_OR_RET(ParseXAckDelStreamEntryIDComponent(std::string_view(input).substr(0, pos), false));
+    auto seq = GET_OR_RET(ParseXAckDelStreamEntryIDComponent(std::string_view(input).substr(pos + 1), true));
+    id->ms = ms;
+    id->seq = seq;
+  } else {
+    auto ms = GET_OR_RET(ParseXAckDelStreamEntryIDComponent(input, false));
+    id->ms = ms;
+    id->seq = 0;
+  }
+  return Status::OK();
+}
 }  // namespace
 
 void AddStreamEntriesToResponse(std::string *output, const std::vector<StreamEntry> &entries) {
@@ -276,44 +319,57 @@ class CommandXAckDel : public Commander {
     group_name_ = GET_OR_RET(parser.TakeStr());
 
     option_ = redis::StreamDeleteOption::KeepRef;
+    bool has_option = false;
+    bool has_ids = false;
 
-    while (parser.Good() && !util::EqualICase(parser.RawPeek(), "IDS")) {
+    while (parser.Good()) {
       if (parser.EatEqICase("KEEPREF")) {
+        if (has_option) return parser.InvalidSyntax();
         option_ = redis::StreamDeleteOption::KeepRef;
+        has_option = true;
       } else if (parser.EatEqICase("DELREF")) {
+        if (has_option) return parser.InvalidSyntax();
         option_ = redis::StreamDeleteOption::DelRef;
+        has_option = true;
       } else if (parser.EatEqICase("ACKED")) {
+        if (has_option) return parser.InvalidSyntax();
         option_ = redis::StreamDeleteOption::Acked;
+        has_option = true;
+      } else if (parser.EatEqICase("IDS")) {
+        has_ids = true;
+
+        if (!parser.Good() || !IsXAckDelNumIDs(parser.RawPeek())) {
+          return {Status::RedisParseErr, "Number of IDs must be a positive integer"};
+        }
+        auto numids_result = parser.TakeInt<int64_t>();
+        if (!numids_result.IsOK()) {
+          return {Status::RedisParseErr, "Number of IDs must be a positive integer"};
+        }
+        int64_t numids = numids_result.GetValue();
+
+        if (parser.Remains() < static_cast<size_t>(numids)) {
+          return {Status::RedisParseErr, "The `numids` parameter must match the number of arguments"};
+        }
+
+        std::vector<redis::StreamEntryID> entry_ids;
+        entry_ids.reserve(static_cast<size_t>(numids));
+        for (int64_t i = 0; i < numids; i++) {
+          auto id_str = GET_OR_RET(parser.TakeStr());
+          redis::StreamEntryID id;
+          auto s = ParseXAckDelStreamEntryID(id_str, &id);
+          if (!s.IsOK()) return s;
+          entry_ids.emplace_back(id);
+        }
+        entry_ids_ = std::move(entry_ids);
       } else {
         return parser.InvalidSyntax();
       }
     }
 
-    if (!parser.EatEqICase("IDS")) {
+    if (!has_ids) {
       return {Status::RedisParseErr, "syntax error, expected IDS keyword"};
     }
 
-    auto numids_result = parser.TakeInt<int64_t>();
-    if (!numids_result.IsOK()) {
-      return {Status::RedisParseErr, errValueNotInteger};
-    }
-    int64_t numids = numids_result.GetValue();
-    if (numids <= 0) {
-      return {Status::RedisParseErr, "numids must be positive"};
-    }
-
-    for (int64_t i = 0; i < numids; i++) {
-      auto id_str = GET_OR_RET(parser.TakeStr());
-      redis::StreamEntryID id;
-      auto s = ParseStreamEntryID(id_str, &id);
-      if (!s.IsOK()) return s;
-      entry_ids_.emplace_back(id);
-    }
-
-    if (parser.Good()) {
-      return {Status::RedisParseErr, "syntax error, unexpected trailing arguments after IDs"};
-    }
-
     return Status::OK();
   }
 
diff --git a/src/storage/batch_extractor.cc b/src/storage/batch_extractor.cc
@@ -37,10 +37,13 @@ void WriteBatchExtractor::LogData(const rocksdb::Slice &blob) {
     }
   } else {
     // Redis type log data
+    // Reset state first so malformed log data cannot reuse XACKDEL dedup keys.
+    log_data_ = redis::WriteBatchLogData();
+    first_seen_ = true;
+    seen_xackdel_xack_keys_.clear();
+    seen_xackdel_xdel_keys_.clear();
     if (auto s = log_data_.Decode(blob); !s.IsOK()) {
       WARN("Failed to decode Redis type log: {}", s.Msg());
-    } else {
-      seen_xackdel_entry_keys_.clear();
     }
   }
 }
@@ -417,10 +420,11 @@ rocksdb::Status WriteBatchExtractor::DeleteCF(uint32_t column_family_id, const S
     GetFixed64(&encoded_id, &entry_id.ms);
 
     if (entry_id.ms == UINT64_MAX) {
-      // PEL / group / consumer metadata sub-key. Only PEL deletions
-      // produce XACKDEL commands for replication.
+      // XACKDEL physical replay emits XACK for each PEL deletion,
+      // using the actual group encoded in the PEL key.
       auto args = log_data_.GetArguments();
       if (!args->empty() && (*args)[0] == "XACKDEL" && args->size() >= 3) {
+        // Skip malformed internal stream keys instead of emitting partial replay commands.
         uint8_t type_delimiter = 0;
         if (!GetFixed8(&encoded_id, &type_delimiter)) {
           return rocksdb::Status::OK();
@@ -433,6 +437,7 @@ rocksdb::Status WriteBatchExtractor::DeleteCF(uint32_t column_family_id, const S
           if (group_name_len > encoded_id.size() || encoded_id.size() - group_name_len < 16) {
             return rocksdb::Status::OK();
           }
+          std::string group_name = encoded_id.ToString().substr(0, group_name_len);
           encoded_id.remove_prefix(group_name_len);
 
           if (!GetFixed64(&encoded_id, &entry_id.ms) || !GetFixed64(&encoded_id, &entry_id.seq)) {
@@ -446,9 +451,9 @@ rocksdb::Status WriteBatchExtractor::DeleteCF(uint32_t column_family_id, const S
             return rocksdb::Status::OK();
           }
           ns = ikey.GetNamespace().ToString();
-          std::string dedup_key = ns + '\0' + user_key + '\0' + (*args)[1] + '\0' + entry_id_str;
-          if (seen_xackdel_entry_keys_.insert(std::move(dedup_key)).second) {
-            command_args = {(*args)[0], user_key, (*args)[1], (*args)[2], "IDS", "1", entry_id_str};
+          std::string dedup_key = ns + '\0' + user_key + '\0' + group_name + '\0' + entry_id_str;
+          if (seen_xackdel_xack_keys_.insert(std::move(dedup_key)).second) {
+            command_args = {"XACK", user_key, group_name, entry_id_str};
             resp_commands_[ns].emplace_back(redis::ArrayOfBulkStrings(command_args));
           }
         }
@@ -469,9 +474,9 @@ rocksdb::Status WriteBatchExtractor::DeleteCF(uint32_t column_family_id, const S
     auto args = log_data_.GetArguments();
     if (!args->empty()) {
       if ((*args)[0] == "XACKDEL" && args->size() >= 3) {
-        std::string dedup_key = ns + '\0' + user_key + '\0' + (*args)[1] + '\0' + entry_id_str;
-        if (seen_xackdel_entry_keys_.insert(std::move(dedup_key)).second) {
-          command_args = {(*args)[0], user_key, (*args)[1], (*args)[2], "IDS", "1", entry_id_str};
+        std::string dedup_key = ns + '\0' + user_key + '\0' + entry_id_str;
+        if (seen_xackdel_xdel_keys_.insert(std::move(dedup_key)).second) {
+          command_args = {"XDEL", user_key, entry_id_str};
         }
       } else {
         command_args = {"XDEL", user_key, entry_id_str};
diff --git a/src/storage/batch_extractor.h b/src/storage/batch_extractor.h
@@ -55,5 +55,6 @@ class WriteBatchExtractor : public rocksdb::WriteBatch::Handler {
   bool is_slot_id_encoded_ = false;
   SlotRange slot_range_;
   bool to_redis_;
-  std::unordered_set<std::string> seen_xackdel_entry_keys_;
+  std::unordered_set<std::string> seen_xackdel_xack_keys_;
+  std::unordered_set<std::string> seen_xackdel_xdel_keys_;
 };
diff --git a/src/types/redis_stream.cc b/src/types/redis_stream.cc
@@ -28,6 +28,7 @@
 #include <vector>
 
 #include "db_util.h"
+#include "scope_exit.h"
 #include "string_util.h"
 #include "time_util.h"
 
@@ -466,6 +467,7 @@ rocksdb::Status Stream::flushPendingNumberUpdates(
     engine::Context &ctx, const std::string &ns_key, const StreamMetadata &metadata, rocksdb::WriteBatchBase *batch,
     const std::map<std::string, uint64_t> &group_pending_decrements,
     const std::map<std::string, std::map<std::string, uint64_t>> &consumer_pending_decrements) {
+  // Saturate pending counters to tolerate stale metadata without uint64_t underflow.
   for (const auto &[group_name, decrement] : group_pending_decrements) {
     auto group_key = internalKeyFromGroupName(ns_key, metadata, group_name);
     std::string group_value;
@@ -475,7 +477,7 @@ rocksdb::Status Stream::flushPendingNumberUpdates(
     }
     if (s.ok()) {
       auto group_meta = decodeStreamConsumerGroupMetadataValue(group_value);
-      group_meta.pending_number -= decrement;
+      group_meta.pending_number = group_meta.pending_number > decrement ? group_meta.pending_number - decrement : 0;
       s = batch->Put(stream_cf_handle_, group_key, encodeStreamConsumerGroupMetadataValue(group_meta));
       if (!s.ok()) return s;
     }
@@ -491,7 +493,8 @@ rocksdb::Status Stream::flushPendingNumberUpdates(
       }
       if (s.ok()) {
         auto consumer_meta = decodeStreamConsumerMetadataValue(consumer_value);
-        consumer_meta.pending_number -= decrement;
+        consumer_meta.pending_number =
+            consumer_meta.pending_number > decrement ? consumer_meta.pending_number - decrement : 0;
         s = batch->Put(stream_cf_handle_, consumer_key, encodeStreamConsumerMetadataValue(consumer_meta));
         if (!s.ok()) return s;
       }
@@ -560,6 +563,11 @@ rocksdb::Status Stream::DeleteEntriesAndAck(engine::Context &ctx, const Slice &s
   }
 
   auto batch = storage_->GetWriteBatchBase();
+  // PutLogData is added before knowing whether any ID mutates data;
+  // roll it back on no-op or error to avoid phantom replay.
+  batch->SetSavePoint();
+  auto rollback_batch = MakeScopeExit([&batch] { batch->RollbackToSavePoint().PermitUncheckedError(); });
+
   std::string option_str;
   switch (option) {
     case StreamDeleteOption::DelRef:
@@ -626,11 +634,21 @@ rocksdb::Status Stream::DeleteEntriesAndAck(engine::Context &ctx, const Slice &s
   StreamEntryID original_first_entry_id = metadata.first_entry_id;
   StreamEntryID original_last_entry_id = metadata.last_entry_id;
 
+  std::vector<std::string> other_groups;
+  if (need_groups) {
+    other_groups.reserve(all_groups.empty() ? 0 : all_groups.size() - 1);
+    for (const auto &candidate_group_name : all_groups) {
+      if (candidate_group_name != group_name) other_groups.push_back(candidate_group_name);
+    }
+  }
+
   for (size_t i = 0; i < ids.size(); i++) {
     const auto &id = ids[i];
 
     std::string entry_key = internalKeyFromEntryID(ns_key, metadata, id);
 
+    // Each ID can mutate the stream at most once per command;
+    // duplicates report not-found after the first attempt.
     if (!seen_entry_keys.insert(entry_key).second) {
       (*results)[i] = static_cast<int>(StreamEntryDeleteResult::kEntryNotFound);
       continue;
@@ -663,13 +681,6 @@ rocksdb::Status Stream::DeleteEntriesAndAck(engine::Context &ctx, const Slice &s
     }
     bool stream_entry_exists = s.ok();
 
-    std::vector<std::string> other_groups;
-    if (need_groups) {
-      for (const auto &candidate_group_name : all_groups) {
-        if (candidate_group_name != group_name) other_groups.push_back(candidate_group_name);
-      }
-    }
-
     if (!stream_entry_exists) {
       if (option == StreamDeleteOption::DelRef && !other_groups.empty()) {
         s = cleanPelFromAllGroups(ctx, ns_key, metadata, id, batch.Get(), &batch_modified, other_groups,
@@ -777,9 +788,11 @@ rocksdb::Status Stream::DeleteEntriesAndAck(engine::Context &ctx, const Slice &s
       if (!s.ok()) return s;
     }
 
+    // Saturate pending counters to tolerate stale metadata without uint64_t underflow.
     if (acknowledged_cnt > 0) {
       StreamConsumerGroupMetadata group_metadata = decodeStreamConsumerGroupMetadataValue(get_group_value);
-      group_metadata.pending_number -= acknowledged_cnt;
+      group_metadata.pending_number =
+          group_metadata.pending_number > acknowledged_cnt ? group_metadata.pending_number - acknowledged_cnt : 0;
       std::string group_value = encodeStreamConsumerGroupMetadataValue(group_metadata);
       s = batch->Put(stream_cf_handle_, group_key, group_value);
       if (!s.ok()) return s;
@@ -793,7 +806,8 @@ rocksdb::Status Stream::DeleteEntriesAndAck(engine::Context &ctx, const Slice &s
         }
         if (s.ok()) {
           auto consumer_metadata = decodeStreamConsumerMetadataValue(consumer_meta_original);
-          consumer_metadata.pending_number -= ack_count;
+          consumer_metadata.pending_number =
+              consumer_metadata.pending_number > ack_count ? consumer_metadata.pending_number - ack_count : 0;
           s = batch->Put(stream_cf_handle_, consumer_meta_key, encodeStreamConsumerMetadataValue(consumer_metadata));
           if (!s.ok()) return s;
         }
@@ -811,6 +825,10 @@ rocksdb::Status Stream::DeleteEntriesAndAck(engine::Context &ctx, const Slice &s
     return rocksdb::Status::OK();
   }
 
+  s = batch->PopSavePoint();
+  if (!s.ok()) return s;
+  rollback_batch.Disable();
+
   return storage_->Write(ctx, storage_->DefaultWriteOptions(), batch->GetWriteBatch());
 }
 
diff --git a/src/types/redis_stream_base.h b/src/types/redis_stream_base.h
@@ -32,6 +32,8 @@
 
 namespace redis {
 
+extern const char *kErrInvalidEntryIdSpecified;
+
 struct StreamEntryID {
   uint64_t ms = 0;
   uint64_t seq = 0;
diff --git a/tests/cppunit/writebatch_indexer_test.cc b/tests/cppunit/writebatch_indexer_test.cc
diff --git a/tests/gocase/unit/server/poll_updates_test.go b/tests/gocase/unit/server/poll_updates_test.go
diff --git a/tests/gocase/unit/type/stream/stream_test.go b/tests/gocase/unit/type/stream/stream_test.go
