From 371d7850d8c2ebc151441e63932ee1245aafb9cb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Carlos=20Ferreira=20Gonz=C3=A1lez?= Date: Thu, 25 Jun 2026 13:40:51 +0200 Subject: [PATCH] Fix volatile reader desync after initial positioning GAP MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * Refs #24631: Regression Test Signed-off-by: Carlos Ferreira González * Refs #24631: Limit GAP processing to HB announced value Signed-off-by: Carlos Ferreira González * Refs #24631: Send a HB before every GAP Signed-off-by: Carlos Ferreira González * Refs #24631: Update tests Signed-off-by: Carlos Ferreira González * Refs #24631: Fix writer's assumption of reader state after GAP sent Signed-off-by: Carlos Ferreira González * Refs #24631: Uncrustify Signed-off-by: Carlos Ferreira González * Refs #24631: Update versions.md Signed-off-by: Carlos Ferreira González * Refs #24631: Skip GAPs for changes removed from history Signed-off-by: Carlos Ferreira González * Refs #24631: Fix Datasharing GAP reception Signed-off-by: Carlos Ferreira González * Refs #24631: Handle sample processed as lost instead of GAP'ed Signed-off-by: Carlos Ferreira González * Refs #24631: Send HB in process_acknack iff GAP is required Signed-off-by: Carlos Ferreira González * Refs #24631: Coverage test for unknown sequence number comparison Signed-off-by: Carlos Ferreira González --------- Signed-off-by: Carlos Ferreira González --- src/cpp/rtps/reader/WriterProxy.h | 23 +- src/cpp/rtps/writer/StatefulWriter.cpp | 73 ++++- src/cpp/rtps/writer/StatefulWriter.hpp | 5 +- .../common/DDSBlackboxTestsContentFilter.cpp | 83 +++++- .../rtps/common/SequenceNumberTests.cpp | 19 ++ .../unittest/rtps/reader/WriterProxyTests.cpp | 264 ++++++------------ 6 files changed, 266 insertions(+), 201 deletions(-) diff --git a/src/cpp/rtps/reader/WriterProxy.h b/src/cpp/rtps/reader/WriterProxy.h index 95f8604eb45..806ef2470a8 100644 --- a/src/cpp/rtps/reader/WriterProxy.h +++ b/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; diff --git a/src/cpp/rtps/writer/StatefulWriter.cpp b/src/cpp/rtps/writer/StatefulWriter.cpp index 84b292e2ffb..681a3f5e108 100644 --- a/src/cpp/rtps/writer/StatefulWriter.cpp +++ b/src/cpp/rtps/writer/StatefulWriter.cpp @@ -19,6 +19,7 @@ #include "StatefulWriter.hpp" +#include #include #include #include @@ -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())) diff --git a/src/cpp/rtps/writer/StatefulWriter.hpp b/src/cpp/rtps/writer/StatefulWriter.hpp index 2a92a7f192e..069a5bef216 100644 --- a/src/cpp/rtps/writer/StatefulWriter.hpp +++ b/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. diff --git a/test/blackbox/common/DDSBlackboxTestsContentFilter.cpp b/test/blackbox/common/DDSBlackboxTestsContentFilter.cpp index 7cb92af1c0d..37a6b3ecf64 100644 --- a/test/blackbox/common/DDSBlackboxTestsContentFilter.cpp +++ b/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 writer("ShortT"); + std::shared_ptr udp_descriptor = + std::make_shared(); + 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>("ShortT", expression, + std::vector{}); + 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>("ShortT", expression_2, + std::vector{}); + 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 - diff --git a/test/unittest/rtps/common/SequenceNumberTests.cpp b/test/unittest/rtps/common/SequenceNumberTests.cpp index 941212c332a..94a8ca387c7 100644 --- a/test/unittest/rtps/common/SequenceNumberTests.cpp +++ b/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 #include #include @@ -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) diff --git a/test/unittest/rtps/reader/WriterProxyTests.cpp b/test/unittest/rtps/reader/WriterProxyTests.cpp index 9074406444c..3a8444c655f 100644 --- a/test/unittest/rtps/reader/WriterProxyTests.cpp +++ b/test/unittest/rtps/reader/WriterProxyTests.cpp @@ -21,7 +21,8 @@ FRIEND_TEST(WriterProxyTests, MissingChangesUpdate); \ FRIEND_TEST(WriterProxyTests, LostChangesUpdate); \ FRIEND_TEST(WriterProxyTests, ReceivedChangeSet); \ - FRIEND_TEST(WriterProxyTests, IrrelevantChangeSet); + FRIEND_TEST(WriterProxyTests, IrrelevantChangeSet); \ + FRIEND_TEST(WriterProxyTests, GapIgnoredUntilHeartbeat); #include @@ -559,6 +560,16 @@ TEST(WriterProxyTests, IrrelevantChangeSet) counted_sequence_numbers.insert(seq); }; + // A Heartbeat must precede every GAP for the reader to learn the writer's sequence number range + EXPECT_CALL(*wproxy.heartbeat_response_, restart_timer()).Times(::testing::AnyNumber()); + bool assert_liveliness = false; + uint32_t heartbeat_count = 1; + int32_t current_sample_lost = 0; + + // Heartbeat announcing the writer has data up to sequence number 3. + wproxy.process_heartbeat(heartbeat_count++, SequenceNumber_t(0, 1), SequenceNumber_t(0, 3), + false, false, false, assert_liveliness, current_sample_lost); + // 1. Simulate reception of a GAP message for sequence number 3. // Sequence number 1 should be UNKNOWN // Sequence number 2 should be UNKNOWN @@ -575,6 +586,10 @@ TEST(WriterProxyTests, IrrelevantChangeSet) ASSERT_EQ(wproxy.are_there_missing_changes(), true); ASSERT_EQ(wproxy.unknown_missing_changes_up_to(SequenceNumber_t(0, 4)), 2u); + // Heartbeat announcing the writer has data up to sequence number 6. + wproxy.process_heartbeat(heartbeat_count++, SequenceNumber_t(0, 1), SequenceNumber_t(0, 6), + false, false, false, assert_liveliness, current_sample_lost); + // 2. Simulate reception of a GAP message for sequence number 6. // Sequence number 1 should be UNKNOWN // Sequence number 2 should be UNKNOWN @@ -635,6 +650,10 @@ TEST(WriterProxyTests, IrrelevantChangeSet) ASSERT_EQ(wproxy.are_there_missing_changes(), true); ASSERT_EQ(wproxy.unknown_missing_changes_up_to(SequenceNumber_t(0, 7)), 2u); + // Heartbeat announcing the writer has data up to sequence number 8. + wproxy.process_heartbeat(heartbeat_count++, SequenceNumber_t(0, 1), SequenceNumber_t(0, 8), + false, false, false, assert_liveliness, current_sample_lost); + // 5. Simulate reception of a GAP message for sequence number 8. // Sequence number 4 should be UNKNOWN // Sequence number 5 should be UNKNOWN @@ -697,10 +716,6 @@ TEST(WriterProxyTests, IrrelevantChangeSet) ASSERT_EQ(wproxy.unknown_missing_changes_up_to(SequenceNumber_t(0, 9)), 0u); // 9. Simulate reception of a HEARTBEAT(1,10008) after all changes have been marked as irrelevant - bool assert_liveliness = false; - uint32_t heartbeat_count = 1; - int32_t current_sample_lost = 0; - EXPECT_CALL(*wproxy.heartbeat_response_, restart_timer()).Times(1u); wproxy.process_heartbeat( heartbeat_count++, SequenceNumber_t(0, 1), @@ -718,190 +733,91 @@ TEST(WriterProxyTests, IrrelevantChangeSet) ASSERT_EQ(wproxy.are_there_missing_changes(), true); ASSERT_EQ(wproxy.unknown_missing_changes_up_to(SequenceNumber_t(0, 10009)), 10000u); - // 10. Simulate reception of a GAP message for sequence numbers 1000 to 2000. GAP should be ignored - wproxy.process_gap(SequenceNumber_t(0, 1000), SequenceNumberSet_t(SequenceNumber_t(0, 2001)), validate_fn); + // From here on max_sequence_number_ == 10008 and changes_from_writer_low_mark_ == 8. + // GAP processing is capped to max_sequence_number_ (the last sequence announced by a Heartbeat) - ASSERT_THAT(t9, wproxy.missing_changes()); - ASSERT_EQ(wproxy.number_of_changes_from_writer(), 10000u); - ASSERT_EQ(wproxy.are_there_missing_changes(), true); - ASSERT_EQ(wproxy.unknown_missing_changes_up_to(SequenceNumber_t(0, 10009)), 10000u); + // 10. GAP for 5000..5002, far above the low mark but within the announced range + wproxy.process_gap(SequenceNumber_t(0, 5000), SequenceNumberSet_t(SequenceNumber_t(0, 5003)), validate_fn); - // 11. Simulate reception of a GAP for sequence numbers 10, 20, 30 - // Sequence numbers 10, 20, and 30 should be RECEIVED with is_relevant = false - { - SequenceNumberSet_t gap_set(SequenceNumber_t(0, 11)); - gap_set.add(SequenceNumber_t(0, 20)); - gap_set.add(SequenceNumber_t(0, 30)); - wproxy.process_gap(SequenceNumber_t(0, 10), gap_set, validate_fn); - } - - SequenceNumberSet_t t11(SequenceNumber_t(0, 9)); - t11.add_range(SequenceNumber_t(0, 9), SequenceNumber_t(0, 10008)); - t11.remove(SequenceNumber_t(0, 10)); - t11.remove(SequenceNumber_t(0, 20)); - t11.remove(SequenceNumber_t(0, 30)); - ASSERT_THAT(t11, wproxy.missing_changes()); ASSERT_EQ(wproxy.number_of_changes_from_writer(), 10000u); ASSERT_EQ(wproxy.are_there_missing_changes(), true); ASSERT_EQ(wproxy.unknown_missing_changes_up_to(SequenceNumber_t(0, 10009)), 9997u); - ASSERT_EQ(wproxy.unknown_missing_changes_up_to(SequenceNumber_t(0, 10)), 1u); - ASSERT_EQ(wproxy.unknown_missing_changes_up_to(SequenceNumber_t(0, 11)), 1u); - ASSERT_EQ(wproxy.unknown_missing_changes_up_to(SequenceNumber_t(0, 20)), 10u); - ASSERT_EQ(wproxy.unknown_missing_changes_up_to(SequenceNumber_t(0, 21)), 10u); - ASSERT_EQ(wproxy.unknown_missing_changes_up_to(SequenceNumber_t(0, 30)), 19u); - ASSERT_EQ(wproxy.unknown_missing_changes_up_to(SequenceNumber_t(0, 31)), 19u); - - // 12. Simulate reception of a GAP for sequence numbers 100, 200, 300 - // Sequence numbers 100, and 200 should be RECEIVED with is_relevant = false - // GAP for sequence number 300 should be ignored - { - SequenceNumberSet_t gap_set(SequenceNumber_t(0, 101)); - gap_set.add(SequenceNumber_t(0, 200)); - gap_set.add(SequenceNumber_t(0, 300)); - wproxy.process_gap(SequenceNumber_t(0, 100), gap_set, validate_fn); - } - - SequenceNumberSet_t t12(t11); - t12.remove(SequenceNumber_t(0, 100)); - t12.remove(SequenceNumber_t(0, 200)); - ASSERT_THAT(t12, wproxy.missing_changes()); + ASSERT_TRUE(wproxy.change_was_received(SequenceNumber_t(0, 5000))); + ASSERT_TRUE(wproxy.change_was_received(SequenceNumber_t(0, 5002))); + ASSERT_FALSE(wproxy.change_was_received(SequenceNumber_t(0, 5003))); + + // 11. GAP for 10007..10010, partially beyond the announced max (10008). + // Only 10007 and 10008 are marked irrelevant. 10009 and 10010 are capped out. + wproxy.process_gap(SequenceNumber_t(0, 10007), SequenceNumberSet_t(SequenceNumber_t(0, 10011)), validate_fn); + ASSERT_EQ(wproxy.number_of_changes_from_writer(), 10000u); - ASSERT_EQ(wproxy.are_there_missing_changes(), true); ASSERT_EQ(wproxy.unknown_missing_changes_up_to(SequenceNumber_t(0, 10009)), 9995u); - ASSERT_EQ(wproxy.unknown_missing_changes_up_to(SequenceNumber_t(0, 100)), 88u); - ASSERT_EQ(wproxy.unknown_missing_changes_up_to(SequenceNumber_t(0, 101)), 88u); - ASSERT_EQ(wproxy.unknown_missing_changes_up_to(SequenceNumber_t(0, 200)), 187u); - ASSERT_EQ(wproxy.unknown_missing_changes_up_to(SequenceNumber_t(0, 201)), 187u); - - // 13. Simulate reception of a GAP for sequence numbers 9 - 1008 - // All sequence numbers below 1009 considered RECEIVED, so first missing is 1009 - wproxy.process_gap(SequenceNumber_t(0, 9), SequenceNumberSet_t(SequenceNumber_t(0, 1009)), validate_fn); - - SequenceNumberSet_t t13(SequenceNumber_t(0, 1009)); - t13.add_range(SequenceNumber_t(0, 1009), SequenceNumber_t(0, 10008)); - ASSERT_THAT(t13, wproxy.missing_changes()); - ASSERT_EQ(wproxy.number_of_changes_from_writer(), 9000u); - ASSERT_EQ(wproxy.are_there_missing_changes(), true); - ASSERT_EQ(wproxy.unknown_missing_changes_up_to(SequenceNumber_t(0, 10009)), 9000u); + ASSERT_TRUE(wproxy.change_was_received(SequenceNumber_t(0, 10008))); + ASSERT_FALSE(wproxy.change_was_received(SequenceNumber_t(0, 10009))); - // 14. Simulate reception of a GAP for sequence numbers 1000 - 1008. GAP should be ignored - wproxy.process_gap(SequenceNumber_t(0, 1000), SequenceNumberSet_t(SequenceNumber_t(0, 1009)), validate_fn); + // 12. GAP entirely beyond the announced max. Ignored: gap_start > max_sequence_number_. + wproxy.process_gap(SequenceNumber_t(0, 20000), SequenceNumberSet_t(SequenceNumber_t(0, 20003)), validate_fn); - ASSERT_THAT(t13, wproxy.missing_changes()); - ASSERT_EQ(wproxy.number_of_changes_from_writer(), 9000u); - ASSERT_EQ(wproxy.are_there_missing_changes(), true); - ASSERT_EQ(wproxy.unknown_missing_changes_up_to(SequenceNumber_t(0, 10009)), 9000u); - - // 15. Simulate reception of a GAP for sequence numbers 1009 - 2008, 2010 - 3008 - // All sequence numbers below 2009 considered RECEIVED, so first missing is 2009 - // Sequence numbers 2010 - 2265 should be RECEIVED with is_relevant = false - { - SequenceNumberSet_t gap_set(SequenceNumber_t(0, 2009)); - gap_set.add_range(SequenceNumber_t(0, 2010), SequenceNumber_t(0, 3009)); - wproxy.process_gap(SequenceNumber_t(0, 1009), gap_set, validate_fn); - } - - SequenceNumberSet_t t15(SequenceNumber_t(0, 2009)); - t15.add(SequenceNumber_t(0, 2009)); - ASSERT_THAT(t15, wproxy.missing_changes()); - ASSERT_EQ(wproxy.number_of_changes_from_writer(), 8000u); - ASSERT_EQ(wproxy.are_there_missing_changes(), true); - ASSERT_EQ(wproxy.unknown_missing_changes_up_to(SequenceNumber_t(0, 10009)), 8000u - 255u); - ASSERT_EQ(wproxy.unknown_missing_changes_up_to(SequenceNumber_t(0, 2009)), 0u); - ASSERT_EQ(wproxy.unknown_missing_changes_up_to(SequenceNumber_t(0, 2010)), 1u); - ASSERT_EQ(wproxy.unknown_missing_changes_up_to(SequenceNumber_t(0, 2009 + 256)), 1u); - ASSERT_EQ(wproxy.unknown_missing_changes_up_to(SequenceNumber_t(0, 2009 + 257)), 2u); - ASSERT_EQ(wproxy.unknown_missing_changes_up_to(SequenceNumber_t(0, 3009)), 1000u - 255u); - - // 16. Simulate reception of a GAP for sequence 2009 + 256 - // Should be ignored because it exceeds the maximum allowed GAP limit - { - SequenceNumber_t gap_start(0, 2009 + 256); - wproxy.process_gap(gap_start, SequenceNumberSet_t(gap_start + 1), validate_fn); - } - - ASSERT_THAT(t15, wproxy.missing_changes()); - ASSERT_EQ(wproxy.number_of_changes_from_writer(), 8000u); - ASSERT_EQ(wproxy.are_there_missing_changes(), true); - ASSERT_EQ(wproxy.unknown_missing_changes_up_to(SequenceNumber_t(0, 10009)), 8000u - 255u); - ASSERT_EQ(wproxy.unknown_missing_changes_up_to(SequenceNumber_t(0, 2009)), 0u); - ASSERT_EQ(wproxy.unknown_missing_changes_up_to(SequenceNumber_t(0, 2010)), 1u); - ASSERT_EQ(wproxy.unknown_missing_changes_up_to(SequenceNumber_t(0, 2009 + 256)), 1u); - ASSERT_EQ(wproxy.unknown_missing_changes_up_to(SequenceNumber_t(0, 2009 + 257)), 2u); - ASSERT_EQ(wproxy.unknown_missing_changes_up_to(SequenceNumber_t(0, 3009)), 1000u - 255u); - - // 17. Simulate reception of a GAP for sequence numbers 2009 + 255 - // Should be ignored because it is exactly the maximum allowed GAP limit - { - SequenceNumber_t gap_start(0, 2009 + 255); - wproxy.process_gap(gap_start, SequenceNumberSet_t(gap_start + 1), validate_fn); - } - - ASSERT_THAT(t15, wproxy.missing_changes()); - ASSERT_EQ(wproxy.number_of_changes_from_writer(), 8000u); - ASSERT_EQ(wproxy.are_there_missing_changes(), true); - ASSERT_EQ(wproxy.unknown_missing_changes_up_to(SequenceNumber_t(0, 10009)), 8000u - 255u); - ASSERT_EQ(wproxy.unknown_missing_changes_up_to(SequenceNumber_t(0, 2009)), 0u); - ASSERT_EQ(wproxy.unknown_missing_changes_up_to(SequenceNumber_t(0, 2010)), 1u); - ASSERT_EQ(wproxy.unknown_missing_changes_up_to(SequenceNumber_t(0, 2009 + 256)), 1u); - ASSERT_EQ(wproxy.unknown_missing_changes_up_to(SequenceNumber_t(0, 2009 + 257)), 2u); - ASSERT_EQ(wproxy.unknown_missing_changes_up_to(SequenceNumber_t(0, 3009)), 1000u - 255u); - - // 18. Simulate reception of a GAP for sequence numbers 1 - 3008 - // All sequence numbers below 3009 considered RECEIVED, so first missing is 3009 - wproxy.process_gap(SequenceNumber_t(0, 1), SequenceNumberSet_t(SequenceNumber_t(0, 3009)), validate_fn); - - SequenceNumberSet_t t18(SequenceNumber_t(0, 3009)); - t18.add_range(SequenceNumber_t(0, 3009), SequenceNumber_t(0, 10008)); - ASSERT_THAT(t18, wproxy.missing_changes()); - ASSERT_EQ(wproxy.number_of_changes_from_writer(), 7000u); - ASSERT_EQ(wproxy.are_there_missing_changes(), true); - ASSERT_EQ(wproxy.unknown_missing_changes_up_to(SequenceNumber_t(0, 10009)), 7000u); + ASSERT_EQ(wproxy.number_of_changes_from_writer(), 10000u); + ASSERT_EQ(wproxy.unknown_missing_changes_up_to(SequenceNumber_t(0, 10009)), 9995u); + ASSERT_FALSE(wproxy.change_was_received(SequenceNumber_t(0, 20000))); - // 19. Simulate reception of a GAP for sequence 3009 + 256 - // Should be ignored because it exceeds the maximum allowed GAP limit - { - SequenceNumber_t gap_start(0, 3009 + 256); - wproxy.process_gap(gap_start, SequenceNumberSet_t(gap_start + 1), validate_fn); - } + // 13. A new Heartbeat raises the announced max to 20002. The same GAP is now processed. + wproxy.process_heartbeat(heartbeat_count++, SequenceNumber_t(0, 1), SequenceNumber_t(0, 20002), + false, false, false, assert_liveliness, current_sample_lost); + wproxy.process_gap(SequenceNumber_t(0, 20000), SequenceNumberSet_t(SequenceNumber_t(0, 20003)), validate_fn); - ASSERT_THAT(t18, wproxy.missing_changes()); - ASSERT_EQ(wproxy.number_of_changes_from_writer(), 7000u); - ASSERT_EQ(wproxy.are_there_missing_changes(), true); - ASSERT_EQ(wproxy.unknown_missing_changes_up_to(SequenceNumber_t(0, 10009)), 7000u); + ASSERT_EQ(wproxy.number_of_changes_from_writer(), 19994u); + ASSERT_EQ(wproxy.unknown_missing_changes_up_to(SequenceNumber_t(0, 20003)), 19986u); + ASSERT_TRUE(wproxy.change_was_received(SequenceNumber_t(0, 20000))); + ASSERT_TRUE(wproxy.change_was_received(SequenceNumber_t(0, 20002))); - // 20. Simulate reception of a GAP for sequence 3009 + 255 - // Should be ignored because it is exactly the maximum allowed GAP limit - { - SequenceNumber_t gap_start(0, 3009 + 255); - wproxy.process_gap(gap_start, SequenceNumberSet_t(gap_start + 1), validate_fn); - } + // 14. A contiguous GAP starting at the low mark advances it with O(1) storage and + // absorbs the previously stored out-of-order changes (5000..5002). + wproxy.process_gap(SequenceNumber_t(0, 9), SequenceNumberSet_t(SequenceNumber_t(0, 5000)), validate_fn); - ASSERT_THAT(t18, wproxy.missing_changes()); - ASSERT_EQ(wproxy.number_of_changes_from_writer(), 7000u); + ASSERT_EQ(wproxy.number_of_changes_from_writer(), 15000u); ASSERT_EQ(wproxy.are_there_missing_changes(), true); - ASSERT_EQ(wproxy.unknown_missing_changes_up_to(SequenceNumber_t(0, 10009)), 7000u); - - // 21. Simulate reception of a GAP for sequence 3009 + 254 - // Sequence number 3009 + 254 should be RECEIVED with is_relevant = false - { - SequenceNumber_t gap_start(0, 3009 + 254); - wproxy.process_gap(gap_start, SequenceNumberSet_t(gap_start + 1), validate_fn); - } - - t18.remove(SequenceNumber_t(0, 3009 + 254)); - ASSERT_THAT(t18, wproxy.missing_changes()); - ASSERT_EQ(wproxy.number_of_changes_from_writer(), 7000u); - ASSERT_EQ(wproxy.are_there_missing_changes(), true); - ASSERT_EQ(wproxy.unknown_missing_changes_up_to(SequenceNumber_t(0, 10009)), 6999u); + ASSERT_EQ(wproxy.unknown_missing_changes_up_to(SequenceNumber_t(0, 20003)), 14995u); + ASSERT_TRUE(wproxy.change_was_received(SequenceNumber_t(0, 9))); + ASSERT_TRUE(wproxy.change_was_received(SequenceNumber_t(0, 5002))); +} - // 22. Simulate reception of a GAP for the full range (1 - 10008) - // All sequence numbers considered RECEIVED, no changes from writer - wproxy.process_gap(SequenceNumber_t(0, 1), SequenceNumberSet_t(SequenceNumber_t(0, 10009)), validate_fn); - ASSERT_THAT(SequenceNumberSet_t(), wproxy.missing_changes()); - ASSERT_EQ(wproxy.number_of_changes_from_writer(), 0u); - ASSERT_EQ(wproxy.are_there_missing_changes(), false); - ASSERT_EQ(wproxy.unknown_missing_changes_up_to(SequenceNumber_t(0, 10009)), 0u); +// A GAP received before any Heartbeat advances nothing (the writer's range is unknown), +// and is fully applied once a Heartbeat announces the range +TEST(WriterProxyTests, GapIgnoredUntilHeartbeat) +{ + WriterProxyData wattr(4u, 1u); + StatefulReader readerMock; + EXPECT_CALL(readerMock, getEventResource()).Times(1u); + WriterProxy wproxy(&readerMock, RemoteLocatorsAllocationAttributes(), ResourceLimitedContainerConfig()); + EXPECT_CALL(*wproxy.initial_acknack_, update_interval(readerMock.getTimes().initial_acknack_delay)).Times(1u); + EXPECT_CALL(*wproxy.heartbeat_response_, update_interval(readerMock.getTimes().heartbeat_response_delay)).Times(1u); + EXPECT_CALL(*wproxy.initial_acknack_, restart_timer()).Times(1u); + EXPECT_CALL(*wproxy.heartbeat_response_, restart_timer()).Times(::testing::AnyNumber()); + wproxy.start(wattr, SequenceNumber_t()); + + auto noop = [](const SequenceNumber_t&) + { + }; + + // No Heartbeat processed yet: max_sequence_number_ == low mark, so the GAP is ignored. + wproxy.process_gap(SequenceNumber_t(0, 5), SequenceNumberSet_t(SequenceNumber_t(0, 20)), noop); + ASSERT_EQ(wproxy.available_changes_max(), SequenceNumber_t(0, 0)); + ASSERT_FALSE(wproxy.change_was_received(SequenceNumber_t(0, 5))); + ASSERT_FALSE(wproxy.change_was_received(SequenceNumber_t(0, 19))); + + // A Heartbeat announces the writer has data up to sequence number 30. + bool assert_liveliness = false; + int32_t current_sample_lost = 0; + wproxy.process_heartbeat(1u, SequenceNumber_t(0, 1), SequenceNumber_t(0, 30), + false, false, false, assert_liveliness, current_sample_lost); + + // The same GAP is now applied, capped to the announced max (sequence 20 is excluded). + wproxy.process_gap(SequenceNumber_t(0, 5), SequenceNumberSet_t(SequenceNumber_t(0, 20)), noop); + ASSERT_TRUE(wproxy.change_was_received(SequenceNumber_t(0, 5))); + ASSERT_TRUE(wproxy.change_was_received(SequenceNumber_t(0, 19))); + ASSERT_FALSE(wproxy.change_was_received(SequenceNumber_t(0, 20))); } } // namespace rtps