Fix volatile reader desync after initial positioning GAP

src/cpp/rtps/reader/WriterProxy.h

@@ -163,22 +163,21 @@ class WriterProxy : public RTPSMessageSenderInterface
         // Cap start sequence number
         SequenceNumber_t initial_seq = std::max(gap_start, first_allowed_gap);
 
-        // Default maximum allowed GAP is low mark + 256
-        SequenceNumber_t max_allowed_gap = changes_from_writer_low_mark_ + 256u;
+        // Cap the GAP to the maximum sequence number the writer has announced through a
+        // Heartbeat, which is is guaranteed to precede every GAP.
+        // If no Heartbeat has been processed yet, max_sequence_number_ equals
+        // changes_from_writer_low_mark_, so initial_seq >= max_allowed_gap and nothing is
+        // processed (the GAP is ignored until a Heartbeat sets the range).
+        SequenceNumber_t max_allowed_gap = max_sequence_number_ + 1u;
 
-        // Special case when initial_seq is exactly the first allowed GAP
         if (initial_seq == first_allowed_gap)
         {
-            max_allowed_gap = gap_list.base() + 256u;
-
-            // Do not exceed max sequence number when known from a heartbeat
-            if (max_sequence_number_ > changes_from_writer_low_mark_)
-            {
-                max_allowed_gap = max_sequence_number_ + 1;
-            }
+            // Special case for datasharing where no GAP is emitted from the writer's side,
+            // but created locally by the reader for initial positioning
+            max_allowed_gap = std::max(max_allowed_gap, gap_list.base());
         }
 
-        // Early exit if gap_start is beyond allowed range
+        // Early exit if gap_start is beyond the announced range
         if (gap_start > max_allowed_gap)
         {
             return;
@@ -195,7 +194,7 @@ class WriterProxy : public RTPSMessageSenderInterface
             }
         }
 
-        // Early exit if the entire gap_list is beyond allowed range
+        // Early exit if the entire gap_list is beyond the announced range
         if (gap_list.base() > max_allowed_gap)
         {
             return;

src/cpp/rtps/writer/StatefulWriter.cpp

@@ -19,6 +19,7 @@
 
 #include "StatefulWriter.hpp"
 
+#include <algorithm>
 #include <mutex>
 #include <stdexcept>
 #include <vector>
@@ -442,7 +443,8 @@ bool StatefulWriter::intraprocess_gap(
 
 bool StatefulWriter::intraprocess_heartbeat(
         ReaderProxy* reader_proxy,
-        bool liveliness)
+        bool liveliness,
+        bool gap_preceded)
 {
     bool returned_value = false;
     LocalReaderPointer::Instance local_reader = reader_proxy->local_reader();
@@ -463,7 +465,7 @@ bool StatefulWriter::intraprocess_heartbeat(
         }
 
         if ((first_seq != c_SequenceNumber_Unknown && last_seq != c_SequenceNumber_Unknown) &&
-                (liveliness || reader_proxy->has_changes()))
+                (liveliness || gap_preceded || reader_proxy->has_changes()))
         {
             increment_hb_count();
             Count_t hb_count = heartbeat_count_;
@@ -495,6 +497,25 @@ bool StatefulWriter::change_removed_by_history(
             biggest_removed_sequence_number_ = sequence_number;
         }
 
+        // Flush pending irrelevant GAPs for any remote reader before the change is erased from history
+        for (ReaderProxy* reader : matched_remote_readers_)
+        {
+            if (SequenceNumber_t::unknown() != reader->first_irrelevant_removed() &&
+                    reader->first_irrelevant_removed() <= sequence_number)
+            {
+                try
+                {
+                    RTPSMessageGroup group(mp_RTPSParticipant, this, reader->message_sender());
+                    send_heartbeat_nts_(1u, group, true);  // Final Heartbeat
+                    add_gaps_for_removed_irrelevants(*reader, group);
+                }
+                catch (const RTPSMessageGroup::timeout&)
+                {
+                    EPROSIMA_LOG_ERROR(RTPS_WRITER, "Max blocking time reached");
+                }
+            }
+        }
+
         // Invalidate CacheChange pointer in ReaderProxies.
         for_matched_readers(matched_local_readers_, matched_datasharing_readers_, matched_remote_readers_,
                 [sequence_number](ReaderProxy* reader)
@@ -558,9 +579,9 @@ void StatefulWriter::send_heartbeat_to_all_readers(
                 (SequenceNumber_t::unknown() != get_seq_num_min() &&
                 SequenceNumber_t::unknown() != get_seq_num_max()));
 
-            add_gaps_for_holes_in_history(group);
-
             send_heartbeat_nts_(locator_selector_general_.all_remote_readers.size(), group, disable_positive_acks_);
+
+            add_gaps_for_holes_in_history(group);
         }
     }
 }
@@ -579,6 +600,8 @@ void StatefulWriter::deliver_sample_to_intraprocesses(
             // send it a personal GAP.
             if (SequenceNumber_t::unknown() != gap_seq)
             {
+                // Force the heartbeat. The reader may have no pending changes but still needs the HB to apply the GAP
+                intraprocess_heartbeat(remoteReader, false, true);
                 intraprocess_gap(remoteReader, gap_seq, change->sequenceNumber);
                 remoteReader->acked_changes_set(change->sequenceNumber);
             }
@@ -658,6 +681,7 @@ DeliveryRetCode StatefulWriter::deliver_sample_to_network(
             {
                 // Send GAP with irrelevant changes that are not in history.
                 group.sender(this, (*remote_reader)->message_sender());
+                send_heartbeat_nts_(1u, group, disable_positive_acks_);
                 add_gaps_for_removed_irrelevants(**remote_reader, group);
                 group.sender(this, &locator_selector);             // This makes the flush_and_reset().
             }
@@ -704,9 +728,9 @@ DeliveryRetCode StatefulWriter::deliver_sample_to_network(
                     if (gap_seq_for_all != gap_seq)     // If it is an individual GAP, sent it to repective reader.
                     {
                         group.sender(this, (*remote_reader)->message_sender());
+                        send_heartbeat_nts_(1u, group, disable_positive_acks_);
                         group.add_gap(gap_seq, SequenceNumberSet_t(change->sequenceNumber),
                                 (*remote_reader)->guid());
-                        send_heartbeat_nts_(1u, group, disable_positive_acks_);
                         group.sender(this, &locator_selector);     // This makes the flush_and_reset().
                     }
                 }
@@ -729,6 +753,7 @@ DeliveryRetCode StatefulWriter::deliver_sample_to_network(
 
         if (should_send_global_gap) // Send GAP for all readers
         {
+            send_heartbeat_nts_(locator_selector.all_remote_readers.size(), group, disable_positive_acks_);
             group.add_gap(gap_seq_for_all, SequenceNumberSet_t(change->sequenceNumber));
         }
 
@@ -1121,6 +1146,17 @@ bool StatefulWriter::matched_reader_add_edp(
         SequenceNumber_t last_seq = get_seq_num_max();
         RTPSMessageGroup group(mp_RTPSParticipant, this, rp->message_sender());
 
+        // Force the intraprocess Heartbeat in case a VOLATILE local reader has no pending changes, but it
+        // still needs the range to apply the positioning GAP.
+        if (rp->is_local_reader())
+        {
+            intraprocess_heartbeat(rp, false, true);
+        }
+        else
+        {
+            send_heartbeat_nts_(1u, group, disable_positive_acks_);
+        }
+
         // History not empty
         if (min_seq != SequenceNumber_t::unknown())
         {
@@ -1179,13 +1215,8 @@ bool StatefulWriter::matched_reader_add_edp(
             }
         }
 
-        if (rp->is_local_reader())
-        {
-            intraprocess_heartbeat(rp);
-        }
-        else
+        if (!rp->is_local_reader())
         {
-            send_heartbeat_nts_(1u, group, disable_positive_acks_);
             group.flush_and_reset();
         }
     }
@@ -1820,6 +1851,9 @@ void StatefulWriter::send_heartbeat_to_nts(
             try
             {
                 RTPSMessageGroup group(mp_RTPSParticipant, this, remoteReaderProxy.message_sender());
+
+                send_heartbeat_nts_(1u, group, disable_positive_acks_, liveliness);
+
                 SequenceNumber_t firstSeq = get_seq_num_min();
                 SequenceNumber_t lastSeq = get_seq_num_max();
 
@@ -1832,8 +1866,6 @@ void StatefulWriter::send_heartbeat_to_nts(
                         add_gaps_for_holes_in_history(group);
                     }
                 }
-
-                send_heartbeat_nts_(1u, group, disable_positive_acks_, liveliness);
             }
             catch (const RTPSMessageGroup::timeout&)
             {
@@ -1979,6 +2011,21 @@ bool StatefulWriter::process_acknack(
                                 {
                                     // Prepare GAP for requested  samples that are not in history or are irrelevants.
                                     RTPSMessageGroup group(mp_RTPSParticipant, this, remote_reader->message_sender());
+
+                                    // A VOLATILE reader may be requesting changes below the position we assumed for it when
+                                    // it matched if its one-shot positioning GAP was lost.
+                                    // Here we re-send a GAP covering [base, low_mark] on every ACKNACK that reveals the
+                                    // reader is still behind. Base is computed as the maximum of the requested base and
+                                    // the lowest sequence number in history to avoid GAPs for changes already removed
+                                    SequenceNumber_t assumed_low_mark = remote_reader->changes_low_mark();
+                                    SequenceNumber_t gap_from = std::max(sn_set.base(), get_seq_num_min());
+                                    if (gap_from <= assumed_low_mark)
+                                    {
+                                        send_heartbeat_nts_(1u, group, true);
+                                        group.add_gap(gap_from, SequenceNumberSet_t(assumed_low_mark + 1),
+                                        remote_reader->guid());
+                                    }
+
                                     RTPSGapBuilder gap_builder(group);
 
                                     if (remote_reader->requested_changes_set(sn_set, gap_builder, get_seq_num_min()))

src/cpp/rtps/writer/StatefulWriter.hpp

@@ -214,12 +214,15 @@ class StatefulWriter : public BaseWriter
      *
      * @param reader_proxy  Pointer to the proxy representing the reader to deliver the heartbeat to.
      * @param liveliness    True if the heartbeat is a liveliness one.
+     * @param gap_preceded  True to send the heartbeat even when the reader has no changes pending.
+     *                      Used to position a late-joining reader (e.g. VOLATILE) before a GAP.
      *
      * @return True on success.
      */
     bool intraprocess_heartbeat(
             ReaderProxy* reader_proxy,
-            bool liveliness = false);
+            bool liveliness = false,
+            bool gap_preceded = false);
 
     /**
      * @brief Increment the HB count.

test/blackbox/common/DDSBlackboxTestsContentFilter.cpp

@@ -1250,6 +1250,88 @@ TEST(DDSContentFilter, ShouldNotFailWithTooManySubExpressionsDiscovered)
     }
 }
 
+/*
+ * Regression test for Redmine issue #24631
+ *
+ * This test checks that a late-joining VOLATILE (by default) subscriber is able to properly synchronize its
+ * state with a reliable publisher that has a content filter applied, and receive the data that matches the filter.
+ */
+TEST(DDSContentFilter, LateJointerSubWithCFTMustReceiveData)
+{
+    // Create writer
+    PubSubWriter<HelloWorldPubSubType> writer("ShortT");
+    std::shared_ptr<eprosima::fastdds::rtps::UDPv4TransportDescriptor> udp_descriptor =
+            std::make_shared<eprosima::fastdds::rtps::UDPv4TransportDescriptor>();
+    writer.disable_builtin_transport().add_user_transport_to_pparams(udp_descriptor);
+    writer.datasharing_off();
+    writer.reliability(ReliabilityQosPolicyKind::RELIABLE_RELIABILITY_QOS);
+    writer.history_depth(1000).init();
+    ASSERT_TRUE(writer.isInitialized());
+
+    // Create reader
+    const std::string expression = "index BETWEEN " + std::to_string(1) + " AND " +
+            std::to_string(300);
+    auto reader = std::make_shared<PubSubReader<HelloWorldPubSubType>>("ShortT", expression,
+                    std::vector<std::string>{});
+    reader->reliability(ReliabilityQosPolicyKind::RELIABLE_RELIABILITY_QOS).durability_kind(
+        DurabilityQosPolicyKind::VOLATILE_DURABILITY_QOS);
+    reader->datasharing_off();
+    reader->history_depth(1000).init();
+    ASSERT_TRUE(reader->isInitialized());
+    reader->wait_discovery();
+
+    // Wait for discovery before start to publish
+    writer.wait_discovery(1u);
+
+    // Start samples reception
+    reader->startReception(300);
+
+    auto data = default_helloworld_data_generator(500);
+
+    // Send up to sample 300, which will be received by the first reader
+    uint16_t index = 0;
+    for (auto data_sample : data)
+    {
+        if (++index > 300)
+        {
+            break;
+        }
+        ASSERT_TRUE(writer.send_sample(data_sample));
+        std::this_thread::sleep_for(std::chrono::milliseconds(10));
+    }
+
+    ASSERT_EQ(300u, reader->block_for_all(std::chrono::seconds(10)));
+    // Do not destroy reader 1 to avoid the writer removing its history
+
+    // Create a fresh second reader with an expression to only receive new samples
+    const std::string expression_2 = "index > " + std::to_string(300);
+    auto reader_2 = std::make_shared<PubSubReader<HelloWorldPubSubType>>("ShortT", expression_2,
+                    std::vector<std::string>{});
+    reader_2->reliability(ReliabilityQosPolicyKind::RELIABLE_RELIABILITY_QOS);
+    reader_2->datasharing_off();
+    reader_2->history_depth(500).init();
+    ASSERT_TRUE(reader_2->isInitialized());
+
+    reader_2->wait_discovery();
+    writer.wait_discovery(2u);
+
+    reader_2->startReception(200);
+
+    // Send the remaining samples (index 301..500)
+    index = 0;
+    for (auto data_sample : data)
+    {
+        if (++index <= 300)
+        {
+            continue;
+        }
+        ASSERT_TRUE(writer.send_sample(data_sample));
+        std::this_thread::sleep_for(std::chrono::milliseconds(10));
+    }
+
+    ASSERT_EQ(200u, reader_2->block_for_all(std::chrono::seconds(10)));
+}
+
 #ifdef INSTANTIATE_TEST_SUITE_P
 #define GTEST_INSTANTIATE_TEST_MACRO(x, y, z, w) INSTANTIATE_TEST_SUITE_P(x, y, z, w)
 #else
@@ -1282,4 +1364,3 @@ GTEST_INSTANTIATE_TEST_MACRO(DDSContentFilter,
 } // namespace dds
 } // namespace fastdds
 } // namespace eprosima
-

test/unittest/rtps/common/SequenceNumberTests.cpp

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+#include <algorithm>
 #include <climits>
 
 #include <gtest/gtest.h>
@@ -491,6 +492,24 @@ TEST(SequenceNumberSet, GetMaxSeqNumOperation)
     ASSERT_EQ(set.max(), expected_seq);
 }
 
+/*!
+ * @fn TEST(SequenceNumberSet, ComparisonForUnknownSeqNum)
+ * @brief This test checks the unknown sequence number is always lower than any other sequence number.
+ */
+TEST(SequenceNumberSet, ComparisonForUnknownSeqNum)
+{
+    SequenceNumber_t seq = SequenceNumber_t::unknown();
+    SequenceNumber_t seq_higher;
+    ASSERT_GT(seq_higher, seq);
+    ASSERT_EQ(std::max(seq_higher, seq), seq_higher);
+    seq_higher.high = 1;
+    ASSERT_GT(seq_higher, seq);
+    ASSERT_EQ(std::max(seq_higher, seq), seq_higher);
+    SequenceNumber_t seq_higher_low(0, 1);
+    ASSERT_GT(seq_higher_low, seq);
+    ASSERT_EQ(std::max(seq_higher_low, seq), seq_higher_low);
+}
+
 int main(
         int argc,
         char** argv)