Fix DataReader history enforcement to respect max_samples_per_instance

Signed-off-by: Raül <raulojeda@eprosima.com>

Author: raulojeda22

src/cpp/fastdds/publisher/DataWriterHistory.cpp

@@ -48,7 +48,7 @@ HistoryAttributes DataWriterHistory::to_history_attributes(
 
     if (history_qos.kind != KEEP_ALL_HISTORY_QOS)
     {
-        max_samples = history_qos.depth;
+        max_samples = std::min(history_qos.depth, resource_limits_qos.max_samples_per_instance);
         if (topic_kind != NO_KEY)
         {
             max_samples *= resource_limits_qos.max_instances;

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

@@ -95,7 +95,10 @@ DataReaderHistory::DataReaderHistory(
     else
     {
         resource_limited_qos_.max_instances = 1;
-        resource_limited_qos_.max_samples_per_instance = resource_limited_qos_.max_samples;
+        resource_limited_qos_.max_samples_per_instance = std::min(
+            resource_limited_qos_.max_samples_per_instance,
+            resource_limited_qos_.max_samples
+            );
         key_changes_allocation_.initial = resource_limited_qos_.allocated_samples;
         key_changes_allocation_.maximum = resource_limited_qos_.max_samples;
 
@@ -274,7 +277,8 @@ bool DataReaderHistory::received_change_keep_last(
     if (find_key(a_change->instanceHandle, vit))
     {
         DataReaderInstance::ChangeCollection& instance_changes = vit->second->cache_changes;
-        if (instance_changes.size() < static_cast<size_t>(history_qos_.depth))
+        auto effective_depth = std::min(history_qos_.depth, resource_limited_qos_.max_samples_per_instance);
+        if (instance_changes.size() < static_cast<size_t>(effective_depth))
         {
             ret_value = true;
         }
@@ -809,7 +813,8 @@ bool DataReaderHistory::completed_change_keep_last(
 {
     bool ret_value = false;
     DataReaderInstance::ChangeCollection& instance_changes = instance.cache_changes;
-    if (instance_changes.size() < static_cast<size_t>(history_qos_.depth))
+    auto effective_depth = std::min(history_qos_.depth, resource_limited_qos_.max_samples_per_instance);
+    if (instance_changes.size() < static_cast<size_t>(effective_depth))
     {
         ret_value = true;
     }

test/unittest/dds/subscriber/DataReaderTests.cpp

@@ -3948,6 +3948,126 @@ TEST_F(DataReaderTests, set_related_datawriter)
     DomainParticipantFactory::get_instance()->delete_participant(another_participant);
 }
 
+TEST_F(DataReaderTests, history_depth_vs_max_samples_per_instance)
+{
+    static constexpr int32_t depth = 10;
+    static constexpr int32_t max_samples_per_instance = 5;
+    static constexpr int32_t num_samples_to_write = 12;
+    static constexpr int32_t expected_samples = 5;
+
+    DataWriterQos writer_qos = DATAWRITER_QOS_DEFAULT;
+    writer_qos.history().kind = KEEP_LAST_HISTORY_QOS;
+    writer_qos.history().depth = depth;
+    writer_qos.resource_limits().max_samples = 1000;
+    writer_qos.resource_limits().max_instances = 10;
+    writer_qos.resource_limits().max_samples_per_instance = max_samples_per_instance;
+    writer_qos.reliability().kind = RELIABLE_RELIABILITY_QOS;
+
+    DataReaderQos reader_qos = DATAREADER_QOS_DEFAULT;
+    reader_qos.history().kind = KEEP_LAST_HISTORY_QOS;
+    reader_qos.history().depth = depth;
+    reader_qos.resource_limits().max_samples = 1000;
+    reader_qos.resource_limits().max_instances = 10;
+    reader_qos.resource_limits().max_samples_per_instance = max_samples_per_instance;
+    reader_qos.reliability().kind = RELIABLE_RELIABILITY_QOS;
+
+    create_instance_handles();
+    create_entities(nullptr, reader_qos, SUBSCRIBER_QOS_DEFAULT, writer_qos);
+
+    FooType data;
+    data.index(0);
+
+    // Write more samples than max_samples_per_instance
+    for (int32_t i = 0; i < num_samples_to_write; ++i)
+    {
+        data.message()[0] = static_cast<char>('0' + i);
+        data.message()[1] = '\0';
+        ASSERT_EQ(RETCODE_OK, data_writer_->write(&data, handle_ok_));
+        std::this_thread::sleep_for(std::chrono::milliseconds(10));
+    }
+
+    std::this_thread::sleep_for(std::chrono::milliseconds(500));
+
+    FooSeq data_seq;
+    SampleInfoSeq info_seq;
+    ASSERT_EQ(RETCODE_OK, data_reader_->take(data_seq, info_seq, LENGTH_UNLIMITED));
+
+    int valid_samples = 0;
+    for (LoanableCollection::size_type i = 0; i < info_seq.length(); ++i)
+    {
+        if (info_seq[i].valid_data)
+        {
+            valid_samples++;
+        }
+    }
+
+    ASSERT_EQ(expected_samples, valid_samples)
+        << "Expected max_samples_per_instance (" << expected_samples
+        << ") samples, not depth (" << depth << ")";
+
+    ASSERT_EQ(RETCODE_OK, data_reader_->return_loan(data_seq, info_seq));
+}
+
+TEST_F(DataReaderTests, history_depth_vs_max_samples_per_instance_no_key)
+{
+    static constexpr int32_t depth = 10;
+    static constexpr int32_t max_samples_per_instance = 5;
+    static constexpr int32_t max_samples = 100;
+    static constexpr int32_t num_samples_to_write = 12;
+    static constexpr int32_t expected_samples = 5;
+
+    type_.reset(new FooBoundedTypeSupport());
+
+    DataWriterQos writer_qos = DATAWRITER_QOS_DEFAULT;
+    writer_qos.history().kind = KEEP_LAST_HISTORY_QOS;
+    writer_qos.history().depth = depth;
+    writer_qos.resource_limits().max_samples = max_samples;
+    writer_qos.resource_limits().max_instances = 1;
+    writer_qos.resource_limits().max_samples_per_instance = max_samples_per_instance;
+    writer_qos.reliability().kind = RELIABLE_RELIABILITY_QOS;
+
+    DataReaderQos reader_qos = DATAREADER_QOS_DEFAULT;
+    reader_qos.history().kind = KEEP_LAST_HISTORY_QOS;
+    reader_qos.history().depth = depth;
+    reader_qos.resource_limits().max_samples = max_samples;
+    reader_qos.resource_limits().max_instances = 1;
+    reader_qos.resource_limits().max_samples_per_instance = max_samples_per_instance;
+    reader_qos.reliability().kind = RELIABLE_RELIABILITY_QOS;
+
+    create_entities(nullptr, reader_qos, SUBSCRIBER_QOS_DEFAULT, writer_qos);
+
+    FooBoundedType data;
+
+    // Write more samples than max_samples_per_instance
+    for (int32_t i = 0; i < num_samples_to_write; ++i)
+    {
+        data.index(i);
+        ASSERT_EQ(RETCODE_OK, data_writer_->write(&data, HANDLE_NIL));
+        std::this_thread::sleep_for(std::chrono::milliseconds(10));
+    }
+
+    std::this_thread::sleep_for(std::chrono::milliseconds(500));
+
+    FooBoundedSeq data_seq;
+    SampleInfoSeq info_seq;
+    ASSERT_EQ(RETCODE_OK, data_reader_->take(data_seq, info_seq, LENGTH_UNLIMITED));
+
+    int valid_samples = 0;
+    for (LoanableCollection::size_type i = 0; i < info_seq.length(); ++i)
+    {
+        if (info_seq[i].valid_data)
+        {
+            valid_samples++;
+        }
+    }
+
+    ASSERT_EQ(expected_samples, valid_samples)
+        << "NO_KEY topic should respect max_samples_per_instance (" << expected_samples
+        << ") not max_samples (" << max_samples << ") or depth (" << depth << ")";
+
+    ASSERT_EQ(RETCODE_OK, data_reader_->return_loan(data_seq, info_seq));
+}
+
 int main(
         int argc,
         char** argv)