[ros2-eloquent] Simulate initial acknack on ReaderProxy::start() [7974]

src/cpp/rtps/writer/ReaderProxy.cpp

@@ -107,6 +107,8 @@ void ReaderProxy::start(
     is_active_ = true;
     reader_attributes_ = reader_attributes;
 
+    acked_changes_set(SequenceNumber_t());  // Simulate initial acknack to set low mark
+
     timers_enabled_.store(is_remote_and_reliable());
     initial_heartbeat_event_->restart_timer();
 
@@ -262,7 +264,8 @@ void ReaderProxy::acked_changes_set(
         if (seq_num == SequenceNumber_t())
         {
             // Special case. Currently only used on Builtin StatefulWriters
-            // after losing lease duration.
+            // after losing lease duration, and on late joiners to set
+            // changes_low_mark_ to match that of the writer.
             SequenceNumber_t min_sequence = writer_->get_seq_num_min();
 
             if (min_sequence != SequenceNumber_t::unknown())
@@ -307,6 +310,10 @@ void ReaderProxy::acked_changes_set(
                 }
             }
         }
+        else if (!is_local_reader())
+        {
+            future_low_mark = writer_->next_sequence_number();
+        }
     }
 
     changes_low_mark_ = future_low_mark - 1;

test/mock/rtps/StatefulWriter/fastrtps/rtps/writer/StatefulWriter.h

@@ -75,6 +75,11 @@ class StatefulWriter : public RTPSWriter
         return SequenceNumber_t(0, 0);
     }
 
+    SequenceNumber_t next_sequence_number() const
+    {
+        return mp_history->next_sequence_number();
+    }
+
 private:
 
     friend class ReaderProxy;

test/mock/rtps/WriterHistory/fastrtps/rtps/history/WriterHistory.h

@@ -77,6 +77,11 @@ class WriterHistory
             }
         }
 
+        SequenceNumber_t next_sequence_number() const
+        {
+            return last_sequence_number_ + 1;
+        }
+
         HistoryAttributes m_att;
 
     private: