[15375] Skip writer_removed processing for unaccounted instances (backport #2905)

src/cpp/fastdds/subscriber/history/DataReaderInstance.hpp

@@ -75,9 +75,9 @@ struct DataReaderInstance
     {
         bool ret_val = false;
 
-        if (!has_been_accounted)
+        if (!has_been_accounted_)
         {
-            has_been_accounted = true;
+            has_been_accounted_ = true;
             assert(ViewStateKind::NEW_VIEW_STATE == view_state);
             ++counters.instances_new;
             assert(InstanceStateKind::ALIVE_INSTANCE_STATE == instance_state);
@@ -115,13 +115,13 @@ struct DataReaderInstance
             DataReaderHistoryCounters& counters,
             const fastrtps::rtps::GUID_t& writer_guid)
     {
-        return writer_unregister(counters, writer_guid);
+        return has_been_accounted_ && writer_unregister(counters, writer_guid);
     }
 
 private:
 
     //! Whether this instance has ever been included in the history counters
-    bool has_been_accounted = false;
+    bool has_been_accounted_ = false;
 
     bool writer_alive(
             DataReaderHistoryCounters& counters,

test/blackbox/common/BlackboxTestsPubSubHistory.cpp

@@ -12,6 +12,9 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+#include <atomic>
+#include <tuple>
+
 #include "BlackboxTests.hpp"
 
 #include "PubSubReader.hpp"
@@ -20,7 +23,6 @@
 #include <fastrtps/xmlparser/XMLProfileManager.h>
 #include <rtps/transport/test_UDPv4Transport.h>
 #include <gtest/gtest.h>
-#include <tuple>
 
 using namespace eprosima::fastrtps;
 using namespace eprosima::fastdds::rtps;
@@ -1279,6 +1281,120 @@ TEST_P(PubSubHistory, KeepAllWriterContinueSendingAfterReaderMatched)
     ASSERT_TRUE(writer.waitForAllAcked(std::chrono::seconds(3)));
 }
 
+// Regression test for redmine bug #15370
+/*!
+ * @fn TEST(PubSubHistory, ReliableUnmatchWithFutureChanges)
+ * @brief This test checks reader behavior when a writer for which only future changes have been received is unmatched.
+ *
+ * It uses a test transport to drop some DATA and HEARTBEAT messages, in order to force the reader to only know
+ * about changes in the future (i.e. with sequence number greater than 1).
+ *
+ * The test creates a Reliable, Transient Local, Keep Last (10) Writer and Reader.
+ *
+ * After waiting for them to match, 10 samples are sent with both heartbeats and data messages being dropped.
+ * Another 10 samples are then sent, with heartbeats still dropped.
+ * This way, the reader receives them as changes in the future.
+ *
+ * The Writer is then destroyed, and the reader waits for it to unmatch.
+ * The Writer is then created again, all messages are let through, and 10 samples are sent and expected to be received
+ * by the Reader.
+ */
+TEST(PubSubHistory, ReliableUnmatchWithFutureChanges)
+{
+    PubSubReader<HelloWorldPubSubType> reader(TEST_TOPIC_NAME);
+    PubSubWriter<HelloWorldPubSubType> writer(TEST_TOPIC_NAME);
+
+    const uint32_t depth = 10;
+
+    reader.reliability(eprosima::fastrtps::RELIABLE_RELIABILITY_QOS).
+            durability_kind(eprosima::fastrtps::TRANSIENT_LOCAL_DURABILITY_QOS).
+            history_kind(eprosima::fastrtps::KEEP_LAST_HISTORY_QOS).history_depth(depth).
+            init();
+
+    ASSERT_TRUE(reader.isInitialized());
+
+    std::atomic_bool drop_data {false};
+    std::atomic_bool drop_heartbeat {false};
+
+    auto testTransport = std::make_shared<test_UDPv4TransportDescriptor>();
+    testTransport->drop_data_messages_filter_ = [&drop_data](eprosima::fastrtps::rtps::CDRMessage_t& msg)
+            -> bool
+            {
+                auto old_pos = msg.pos;
+
+                // Jump to writer entity id
+                msg.pos += 2 + 2 + 4;
+
+                // Read writer entity id
+                eprosima::fastrtps::rtps::GUID_t writer_guid;
+                eprosima::fastrtps::rtps::CDRMessage::readEntityId(&msg, &writer_guid.entityId);
+                msg.pos = old_pos;
+
+                return drop_data && !writer_guid.is_builtin();
+            };
+    testTransport->drop_heartbeat_messages_filter_ = [&drop_heartbeat](eprosima::fastrtps::rtps::CDRMessage_t& msg)
+            -> bool
+            {
+                auto old_pos = msg.pos;
+                msg.pos += 4;
+                eprosima::fastrtps::rtps::GUID_t writer_guid;
+                eprosima::fastrtps::rtps::CDRMessage::readEntityId(&msg, &writer_guid.entityId);
+                msg.pos = old_pos;
+
+                return drop_heartbeat && !writer_guid.is_builtin();
+            };
+
+    writer.reliability(eprosima::fastrtps::RELIABLE_RELIABILITY_QOS).
+            history_kind(eprosima::fastrtps::KEEP_LAST_HISTORY_QOS).history_depth(depth).
+            disable_builtin_transport().add_user_transport_to_pparams(testTransport).
+            init();
+
+    ASSERT_TRUE(writer.isInitialized());
+
+    // Wait for discovery.
+    writer.wait_discovery();
+    reader.wait_discovery();
+
+    // Drop all heartbeat messages, so reader doesn't know the writer state.
+    drop_heartbeat = true;
+    // Drop data messages the first time, so reader starts receiving changes in the future
+    drop_data = true;
+
+    for (int i = 0; i < 2; ++i)
+    {
+        auto data = default_helloworld_data_generator(depth);
+
+        // Send data
+        writer.send(data);
+        ASSERT_TRUE(data.empty());
+
+        // Let data messages pass the second time, so the reader receive the 'future' changes
+        drop_data = false;
+    }
+
+    // Kill the writer and wait for the reader to unmatch
+    writer.destroy();
+    reader.wait_writer_undiscovery();
+    reader.wait_participant_undiscovery();
+
+    // Create writer again and wait for matching
+    writer.init();
+    reader.wait_discovery();
+
+    // Expect normal (non-dropping) behavior to work
+    drop_heartbeat = 0;
+    auto data = default_helloworld_data_generator(depth);
+    reader.startReception(data);
+
+    // Send data
+    writer.send(data);
+    ASSERT_TRUE(data.empty());
+
+    reader.block_for_all();
+}
+
+
+
 #ifdef INSTANTIATE_TEST_SUITE_P
 #define GTEST_INSTANTIATE_TEST_MACRO(x, y, z, w) INSTANTIATE_TEST_SUITE_P(x, y, z, w)
 #else