fix: reply_builder serializes \0 chars instead of eol

Fixes #4424

src/facade/reply_builder.cc

@@ -105,11 +105,13 @@ void SinkReplyBuilder::CloseConnection() {
 }
 
 template <typename... Ts> void SinkReplyBuilder::WritePieces(Ts&&... pieces) {
+  size_t required2 = (piece_size(pieces) + ...);
   if (size_t required = (piece_size(pieces) + ...); buffer_.AppendLen() <= required)
     Flush(required);
 
   // Ensure last iovec points to buffer segment
   char* dest = reinterpret_cast<char*>(buffer_.AppendBuffer().data());
+  size_t vs = vecs_.size();
   if (vecs_.empty() || ((char*)vecs_.back().iov_base) + vecs_.back().iov_len != dest)
     NextVec({dest, 0});
 
@@ -120,6 +122,12 @@ template <typename... Ts> void SinkReplyBuilder::WritePieces(Ts&&... pieces) {
   size_t written = ptr - dest;
   buffer_.CommitWrite(written);
   vecs_.back().iov_len += written;
+  if (vecs_.front().iov_len == 2) {
+    uint8_t* ptr = reinterpret_cast<uint8_t*>(vecs_.front().iov_base);
+    if (ptr[0] == 0 && ptr[1] == 0) {
+      LOG(DFATAL) << "Zero byte detected in the middle of the string";
+    }
+  }
   total_size_ += written;
 }
 
@@ -183,7 +191,7 @@ void SinkReplyBuilder::FinishScope() {
   if (ref_bytes > buffer_.AppendLen())
     return Flush(ref_bytes);
 
-  // Copy all extenral references to buffer to safely keep batching
+  // Copy all external references to buffer to safely keep batching
   for (size_t i = guaranteed_pieces_; i < vecs_.size(); i++) {
     auto ib = buffer_.InputBuffer();
     if (vecs_[i].iov_base >= ib.data() && vecs_[i].iov_base <= ib.data() + ib.size())

src/facade/reply_builder_test.cc

@@ -917,6 +917,19 @@ TEST_F(RedisReplyBuilderTest, Issue3449) {
   EXPECT_EQ(10000, parse_result.args.size());
 }
 
+TEST_F(RedisReplyBuilderTest, Issue4424) {
+  vector<string> records;
+  for (unsigned i = 0; i < 800; ++i) {
+    records.push_back(string(100, 'a'));
+  }
+  for (unsigned j = 0; j < 2; ++j) {
+    builder_->SendBulkStrArr(records);
+    ASSERT_TRUE(NoErrors());
+    ParsingResults parse_result = Parse();
+    ASSERT_FALSE(parse_result.IsError());
+  }
+}
+
 static void BM_FormatDouble(benchmark::State& state) {
   vector<double> values;
   char buf[64];