[20257] Add a keyed fragmented change to the reader data instance only when its completed (backport #4261)

src/cpp/fastdds/subscriber/history/DataReaderHistory.cpp

@@ -132,14 +132,14 @@ DataReaderHistory::DataReaderHistory(
         compute_key_for_change_fn_ =
                 [this](CacheChange_t* a_change)
                 {
-                    if (a_change->instanceHandle.isDefined())
+                    if (!a_change->is_fully_assembled())
                     {
-                        return true;
+                        return false;
                     }
 
-                    if (!a_change->is_fully_assembled())
+                    if (a_change->instanceHandle.isDefined())
                     {
-                        return false;
+                        return true;
                     }
 
                     if (type_ != nullptr)
@@ -708,27 +708,24 @@ bool DataReaderHistory::completed_change(
         size_t unknown_missing_changes_up_to,
         SampleRejectedStatusKind& rejection_reason)
 {
-    bool ret_value = true;
-    rejection_reason = NOT_REJECTED;
+    bool ret_value = false;
+    rejection_reason = REJECTED_BY_INSTANCES_LIMIT;
 
-    if (!change->instanceHandle.isDefined())
+    if (compute_key_for_change_fn_(change))
     {
-        ret_value = false;
-        if (compute_key_for_change_fn_(change))
+        InstanceCollection::iterator vit;
+        if (find_key(change->instanceHandle, vit))
         {
-            InstanceCollection::iterator vit;
-            if (find_key(change->instanceHandle, vit))
-            {
-                ret_value = !change->instanceHandle.isDefined() ||
-                        complete_fn_(change, *vit->second, unknown_missing_changes_up_to, rejection_reason);
-            }
-            else
-            {
-                rejection_reason = REJECTED_BY_INSTANCES_LIMIT;
-            }
+            ret_value = !change->instanceHandle.isDefined() ||
+                    complete_fn_(change, *vit->second, unknown_missing_changes_up_to, rejection_reason);
         }
     }
 
+    if (ret_value)
+    {
+        rejection_reason = NOT_REJECTED;
+    }
+
     return ret_value;
 }
 

src/cpp/rtps/reader/StatefulReader.cpp

@@ -695,6 +695,7 @@ bool StatefulReader::processDataFragMsg(
                     {
                         work_change->copy_not_memcpy(change_to_add);
                         work_change->serializedPayload.length = sampleSize;
+                        work_change->instanceHandle.clear();
                         work_change->setFragmentSize(change_to_add->getFragmentSize(), true);
                         change_created = work_change;
                     }
@@ -703,6 +704,12 @@ bool StatefulReader::processDataFragMsg(
 
             if (work_change != nullptr)
             {
+                // Set the instanceHandle only when fragment number 1 is received
+                if (!work_change->instanceHandle.isDefined() && fragmentStartingNum == 1)
+                {
+                    work_change->instanceHandle = change_to_add->instanceHandle;
+                }
+
                 work_change->add_fragments(change_to_add->serializedPayload, fragmentStartingNum,
                         fragmentsInSubmessage);
             }

src/cpp/rtps/reader/StatelessReader.cpp

@@ -645,6 +645,7 @@ bool StatelessReader::processDataFragMsg(
                             // Sample fits inside pending change. Reuse it.
                             work_change->copy_not_memcpy(change_to_add);
                             work_change->serializedPayload.length = sampleSize;
+                            work_change->instanceHandle.clear();
                             work_change->setFragmentSize(change_to_add->getFragmentSize(), true);
                         }
                         else
@@ -670,6 +671,7 @@ bool StatelessReader::processDataFragMsg(
                         {
                             work_change->copy_not_memcpy(change_to_add);
                             work_change->serializedPayload.length = sampleSize;
+                            work_change->instanceHandle.clear();
                             work_change->setFragmentSize(change_to_add->getFragmentSize(), true);
                         }
                     }
@@ -679,6 +681,12 @@ bool StatelessReader::processDataFragMsg(
                 CacheChange_t* change_completed = nullptr;
                 if (work_change != nullptr)
                 {
+                    // Set the instanceHandle only when fragment number 1 is received
+                    if (!work_change->instanceHandle.isDefined() && fragmentStartingNum == 1)
+                    {
+                        work_change->instanceHandle = change_to_add->instanceHandle;
+                    }
+
                     if (work_change->add_fragments(change_to_add->serializedPayload, fragmentStartingNum,
                             fragmentsInSubmessage))
                     {

test/blackbox/api/dds-pim/PubSubReader.hpp

@@ -1008,6 +1008,13 @@ class PubSubReader
         return *this;
     }
 
+    PubSubReader& expect_inline_qos(
+            bool expect)
+    {
+        datareader_qos_.expects_inline_qos(expect);
+        return *this;
+    }
+
     PubSubReader& heartbeatResponseDelay(
             const int32_t secs,
             const int32_t frac)

test/blackbox/api/fastrtps_deprecated/PubSubReader.hpp

@@ -794,6 +794,13 @@ class PubSubReader
         return *this;
     }
 
+    PubSubReader& expect_inline_qos(
+            bool expect)
+    {
+        subscriber_attr_.expectsInlineQos = expect;
+        return *this;
+    }
+
     PubSubReader& heartbeatResponseDelay(
             const int32_t secs,
             const int32_t frac)

test/blackbox/common/BlackboxTestsPubSubFragments.cpp

@@ -649,6 +649,67 @@ TEST_P(PubSubFragmentsLimited, AsyncPubSubAsReliableKeyedData300kbKeepLast1InLos
         testTransport->dropLogLength);
 }
 
+// Regression test for 20257
+// When a non existing change is removed, the change is also removed from the data instance changes sequence
+// For uncrustify sake *INDENT-OFF*
+TEST(PubSubFragmentsLimited, AsyncPubSubAsReliableKeyedData300kbKeepLast1LoosyConditionsSmallFragmentsCorrectlyBehavesWhenInlineQoSAreForced)
+// *INDENT-ON*
+{
+    PubSubReader<KeyedData1mbPubSubType> reader(TEST_TOPIC_NAME);
+    PubSubWriter<KeyedData1mbPubSubType> writer(TEST_TOPIC_NAME);
+
+    reader.history_depth(2)
+            .expect_inline_qos(true)
+            .reliability(eprosima::fastrtps::RELIABLE_RELIABILITY_QOS)
+            .init();
+
+    ASSERT_TRUE(reader.isInitialized());
+
+    // To simulate lossy conditions, we are going to remove the default
+    // builtin transport, and instead use a lossy shim layer variant.
+    auto testTransport = std::make_shared<test_UDPv4TransportDescriptor>();
+    testTransport->maxMessageSize = 1024;
+    // We drop 20% of all data frags
+    testTransport->dropDataFragMessagesPercentage = 20;
+    testTransport->dropLogLength = 1;
+    writer.disable_builtin_transport();
+    writer.add_user_transport_to_pparams(testTransport);
+
+    // When doing fragmentation, it is necessary to have some degree of
+    // flow control not to overrun the receive buffer.
+    uint32_t bytesPerPeriod = 153601;
+    uint32_t periodInMs = 100;
+    writer.add_throughput_controller_descriptor_to_pparams(
+        eprosima::fastdds::rtps::FlowControllerSchedulerPolicy::HIGH_PRIORITY, bytesPerPeriod, periodInMs)
+            .heartbeat_period_seconds(0)
+            .heartbeat_period_nanosec(1000000)
+            .history_depth(1)
+            .asynchronously(eprosima::fastrtps::ASYNCHRONOUS_PUBLISH_MODE).init();
+
+    ASSERT_TRUE(writer.isInitialized());
+
+    // Because its volatile the durability
+    // Wait for discovery.
+    writer.wait_discovery();
+    reader.wait_discovery();
+
+    auto data = default_keyeddata300kb_data_generator(5);
+
+    reader.startReception(data);
+
+    // Send data
+    writer.send(data, 100);
+    // In this test all data should be sent.
+    ASSERT_TRUE(data.empty());
+    // Block reader until reception finished or timeout.
+    reader.block_for_seq({ 0, 5 });
+
+    // Sanity check. Make sure we have dropped a few packets
+    ASSERT_EQ(
+        test_UDPv4Transport::test_UDPv4Transport_DropLog.size(),
+        testTransport->dropLogLength);
+}
+
 TEST_P(PubSubFragmentsLimited, AsyncPubSubAsReliableVolatileData300kbInLossyConditionsSmallFragments)
 {
     PubSubReader<Data1mbPubSubType> reader(TEST_TOPIC_NAME);