fix(server): flush trailing duplicate batch when last mutation is ignored

src/server/pegasus_mutation_duplicator.cpp

@@ -253,7 +253,6 @@ void pegasus_mutation_duplicator::duplicate(mutation_tuple_set muts, callback cb
         batch_count++;
         dsn::task_code rpc_code = std::get<1>(mut);
         dsn::blob raw_message = std::get<2>(mut);
-        auto dreq = std::make_unique<dsn::apps::duplicate_request>();
 
         if (gutil::ContainsKey(ingnored_rpc_code, rpc_code)) {
             // It it do not recommend to use bulkload and normal writing in the same app,
@@ -266,23 +265,34 @@ void pegasus_mutation_duplicator::duplicate(mutation_tuple_set muts, callback cb
             if (rpc_code == dsn::apps::RPC_RRDB_RRDB_BULK_LOAD) {
                 LOG_DEBUG_PREFIX("Ignore sending bulkload rpc when doing duplication");
             }
-            continue;
+            // NOTE: do not `continue` here. The batch-flush check below must still run, otherwise
+            // if the LAST mutation in `muts` is an ignored code the trailing batch of valid
+            // mutations would be silently dropped — the caller has already advanced last_decree
+            // past them, so they would never be re-read and duplicated. See #2284.
+        } else {
+            dsn::apps::duplicate_entry entry;
+            entry.__set_raw_message(raw_message);
+            entry.__set_task_code(rpc_code);
+            entry.__set_timestamp(std::get<0>(mut));
+            entry.__set_cluster_id(dsn::replication::get_current_dup_cluster_id());
+            batch_request->entries.emplace_back(std::move(entry));
+            batch_bytes += raw_message.length();
         }
 
-        dsn::apps::duplicate_entry entry;
-        entry.__set_raw_message(raw_message);
-        entry.__set_task_code(rpc_code);
-        entry.__set_timestamp(std::get<0>(mut));
-        entry.__set_cluster_id(dsn::replication::get_current_dup_cluster_id());
-        batch_request->entries.emplace_back(std::move(entry));
-        batch_bytes += raw_message.length();
-
         if (batch_count == muts.size() || batch_bytes >= FLAGS_duplicate_log_batch_bytes ||
             batch_bytes >= dsn::replication::FLAGS_dup_max_allowed_write_size) {
+            if (batch_request->entries.empty()) {
+                // nothing to flush (e.g. this mutation was ignored and the batch is empty)
+                continue;
+            }
             // since all the plog's mutations of replica belong to same gpid though the hash of
             // mutation is different, use the last mutation of one batch to get and represents the
-            // current hash value, it will still send to remote correct replica
-            uint64_t hash = get_hash_from_request(rpc_code, raw_message);
+            // current hash value, it will still send to remote correct replica.
+            // The hash is computed from the last ENTRY in the batch, which is always a
+            // duplicatable rpc code (PUT/REMOVE/MULTI_*); ignored codes are never added to the
+            // batch, and get_hash_from_request would LOG_FATAL on them.
+            const auto &last_entry = batch_request->entries.back();
+            uint64_t hash = get_hash_from_request(last_entry.task_code, last_entry.raw_message);
             duplicate_rpc rpc(std::move(batch_request),
                               dsn::apps::RPC_RRDB_RRDB_DUPLICATE,
                               100_s, // TODO(wutao1): configurable timeout.

src/server/test/pegasus_mutation_duplicator_test.cpp

@@ -407,5 +407,56 @@ TEST_P(pegasus_mutation_duplicator_test, duplicate_duplicate)
     _tracker.wait_outstanding_tasks();
 }
 
+// Regression test for #2284: when the mutation with the highest timestamp (iterated LAST,
+// since mutation_tuple_set is sorted by timestamp) has an ignored rpc code (BULK_LOAD or
+// DUPLICATE), the trailing batch of valid mutations must still be flushed. Before the fix the
+// `continue` for ignored codes skipped the batch-flush check, so the in-progress batch was
+// dropped on the floor — yet the caller had already advanced last_decree past those mutations,
+// meaning they were never duplicated (silent data loss).
+TEST_P(pegasus_mutation_duplicator_test, duplicate_ignored_last_mutation)
+{
+    replica_base replica(dsn::gpid(1, 1), "fake_replica", "temp");
+    auto duplicator = new_mutation_duplicator(&replica, "onebox2", "temp");
+    duplicator->set_task_environment(&_env);
+
+    const std::string hash_key("hash");
+    mutation_tuple_set muts;
+    const int valid_count = 3;
+    for (int i = 0; i < valid_count; i++) {
+        dsn::apps::update_request request;
+        pegasus::pegasus_generate_key(request.key, hash_key, fmt::format("sort_{}", i));
+        dsn::message_ptr msg =
+            dsn::from_thrift_request_to_received_message(request, dsn::apps::RPC_RRDB_RRDB_PUT);
+        auto data = dsn::move_message_to_blob(msg.get());
+        // timestamps 200, 201, 202
+        muts.insert(std::make_tuple(200 + i, dsn::apps::RPC_RRDB_RRDB_PUT, data));
+    }
+
+    // An ignored BULK_LOAD mutation with the HIGHEST timestamp (203), so it is iterated LAST.
+    // Its raw_message content is irrelevant: ignored codes are never parsed (and
+    // get_hash_from_request would LOG_FATAL on them).
+    {
+        dsn::apps::update_request request;
+        pegasus::pegasus_generate_key(request.key, hash_key, std::string("sort_bulkload"));
+        dsn::message_ptr msg =
+            dsn::from_thrift_request_to_received_message(request, dsn::apps::RPC_RRDB_RRDB_PUT);
+        auto data = dsn::move_message_to_blob(msg.get());
+        muts.insert(std::make_tuple(203, dsn::apps::RPC_RRDB_RRDB_BULK_LOAD, data));
+    }
+
+    RPC_MOCKING(duplicate_rpc)
+    {
+        duplicator->duplicate(muts, [](size_t) {});
+
+        // Before the fix, the trailing PUT batch was dropped: nothing would be shipped, so the
+        // mail_box would be empty. After the fix, the 3 PUTs are flushed as a single batch.
+        ASSERT_EQ(duplicate_rpc::mail_box().size(), 1u);
+
+        auto rpc = duplicate_rpc::mail_box().back();
+        ASSERT_EQ(rpc.request().entries.size(), static_cast<size_t>(valid_count));
+    }
+    _tracker.wait_outstanding_tasks();
+}
+
 } // namespace server
 } // namespace pegasus