Enhance filtering logic for virtual and external endpoints (#6290)

MiguelCompany · raulojeda22 · web-flow · commit 3d2479e6b62a · 2026-02-06T08:16:10.000+01:00
* Enhance filtering logic for virtual and external endpoints

Signed-off-by: Miguel Company &lt;miguelcompany@eprosima.com&gt;

* Restore proper filtering for external endpoint discovery matching

Signed-off-by: Raül &lt;raulojeda@eprosima.com&gt;

* Uncrustify

Signed-off-by: Raül &lt;raulojeda@eprosima.com&gt;

* Add DS test checking external-to-external relay

Signed-off-by: Raül &lt;raulojeda@eprosima.com&gt;

* Add additional logic to reader_gives_info and uncrustify

Signed-off-by: Raül &lt;raulojeda@eprosima.com&gt;

* Remove unnecessary comment

Signed-off-by: Raül &lt;raulojeda@eprosima.com&gt;

* Remove leaseDuration config to trigger test error in previous code

Signed-off-by: Raül &lt;raulojeda@eprosima.com&gt;

* Add additional logic to writer_gives_info

Signed-off-by: Raül &lt;raulojeda@eprosima.com&gt;

* Remove unnecessary extra logic on match_writer_reader

Signed-off-by: Raül &lt;raulojeda@eprosima.com&gt;

* Apply revision

Signed-off-by: Raül &lt;raulojeda@eprosima.com&gt;

* Add explicit _not_virtual to _is_local variable name

Signed-off-by: Raül &lt;raulojeda@eprosima.com&gt;

* Fix DS test failing in windows CI

Signed-off-by: Raül &lt;raulojeda@eprosima.com&gt;

---------

Signed-off-by: Miguel Company &lt;miguelcompany@eprosima.com&gt;
Signed-off-by: Raül &lt;raulojeda@eprosima.com&gt;
Co-authored-by: Raül &lt;raulojeda@eprosima.com&gt;
diff --git a/src/cpp/rtps/builtin/discovery/database/DiscoveryDataBase.cpp b/src/cpp/rtps/builtin/discovery/database/DiscoveryDataBase.cpp
@@ -504,15 +504,17 @@ void DiscoveryDataBase::process_pdp_data_queue()
         if (data_queue_info.change()->kind == eprosima::fastdds::rtps::ALIVE)
         {
             // Update participants map
-            EPROSIMA_LOG_INFO(DISCOVERY_DATABASE, "DATA(p) of entity " << data_queue_info.change()->instanceHandle <<
-                    " received from: " << data_queue_info.change()->writerGUID);
+            EPROSIMA_LOG_INFO(DISCOVERY_DATABASE, "DATA(p) of entity " << data_queue_info.change()->instanceHandle
+                                                                       << " received from: "
+                                                                       << data_queue_info.change()->writerGUID);
             create_participant_from_change_(data_queue_info.change(), data_queue_info.participant_change_data());
         }
         // If the change is a DATA(Up)
         else
         {
-            EPROSIMA_LOG_INFO(DISCOVERY_DATABASE, "DATA(Up) of entity " << data_queue_info.change()->instanceHandle <<
-                    " received from: " << data_queue_info.change()->writerGUID);
+            EPROSIMA_LOG_INFO(DISCOVERY_DATABASE, "DATA(Up) of entity " << data_queue_info.change()->instanceHandle
+                                                                        << " received from: "
+                                                                        << data_queue_info.change()->writerGUID);
             process_dispose_participant_(data_queue_info.change());
         }
     }
@@ -1197,31 +1199,59 @@ void DiscoveryDataBase::match_writer_reader_(
     }
     DiscoveryParticipantInfo& reader_participant_info = p_rit->second;
 
-    // Always exchange all information between all participants (no filtering)
-    // Skip only if it's a virtual endpoint matching with another virtual endpoint
+    // Virtual endpoints do not exchange info
     if (writer_info.is_virtual() && reader_info.is_virtual())
     {
         return;
     }
 
-    // Add writer's participant to reader's relevant list
-    if (!reader_participant_info.is_relevant_participant(writer_guid.guidPrefix))
-    {
-        reader_participant_info.add_or_update_ack_participant(writer_guid.guidPrefix);
-    }
-    if (!reader_info.is_relevant_participant(writer_guid.guidPrefix))
-    {
-        reader_info.add_or_update_ack_participant(writer_guid.guidPrefix);
-    }
+    // Classify writer and reader types
+    const bool writer_is_virtual = writer_info.is_virtual();
+    const bool writer_is_local_not_virtual = !writer_is_virtual && writer_participant_info.is_local();
+    const bool writer_is_external_not_virtual = !writer_is_virtual && !writer_is_local_not_virtual;
 
-    // Add reader's participant to writer's relevant list
-    if (!writer_participant_info.is_relevant_participant(reader_guid.guidPrefix))
+    const bool reader_is_virtual = reader_info.is_virtual();
+    const bool reader_is_local_not_virtual = !reader_is_virtual && reader_participant_info.is_local();
+    const bool reader_is_external_not_virtual = !reader_is_virtual && !reader_is_local_not_virtual;
+
+    // Determine info exchange based on endpoint types:
+    // - Local endpoints both need and give info
+    // - Virtual endpoints only need info (never give)
+    // - External endpoints only give info (never need)
+    // Additionally, servers do not redirect between remote clients,
+    // so external endpoints only exchange with local endpoints
+    const bool writer_needs_info = writer_is_virtual || writer_is_local_not_virtual;
+    const bool writer_gives_info = writer_is_local_not_virtual ||
+            (writer_is_external_not_virtual && reader_is_local_not_virtual);
+
+    const bool reader_needs_info = reader_is_virtual || reader_is_local_not_virtual;
+    const bool reader_gives_info = reader_is_local_not_virtual ||
+            (reader_is_external_not_virtual && writer_is_local_not_virtual);
+
+    // Writer needs info from reader: add writer's guid prefix to reader's ack lists
+    if (writer_needs_info && reader_gives_info)
     {
-        writer_participant_info.add_or_update_ack_participant(reader_guid.guidPrefix);
+        if (!reader_participant_info.is_relevant_participant(writer_guid.guidPrefix))
+        {
+            reader_participant_info.add_or_update_ack_participant(writer_guid.guidPrefix);
+        }
+        if (!reader_info.is_relevant_participant(writer_guid.guidPrefix))
+        {
+            reader_info.add_or_update_ack_participant(writer_guid.guidPrefix);
+        }
     }
-    if (!writer_info.is_relevant_participant(reader_guid.guidPrefix))
+
+    // Reader needs info from writer: add reader's guid prefix to writer's ack lists
+    if (reader_needs_info && writer_gives_info)
     {
-        writer_info.add_or_update_ack_participant(reader_guid.guidPrefix);
+        if (!writer_participant_info.is_relevant_participant(reader_guid.guidPrefix))
+        {
+            writer_participant_info.add_or_update_ack_participant(reader_guid.guidPrefix);
+        }
+        if (!writer_info.is_relevant_participant(reader_guid.guidPrefix))
+        {
+            writer_info.add_or_update_ack_participant(reader_guid.guidPrefix);
+        }
     }
 }
 
@@ -1775,8 +1805,9 @@ void DiscoveryDataBase::AckedFunctor::operator () (
                     auto remote_server_it = db_->participants_.find(*it);
                     if (remote_server_it == db_->participants_.end())
                     {
-                        EPROSIMA_LOG_INFO(DISCOVERY_DATABASE, "Change " << change_->instanceHandle <<
-                                "check as acked for " << reader_proxy->guid() << " as it has not answered pinging yet");
+                        EPROSIMA_LOG_INFO(DISCOVERY_DATABASE,
+                                "Change " << change_->instanceHandle << "check as acked for " << reader_proxy->guid()
+                                          << " as it has not answered pinging yet");
                         return;
                     }
 
@@ -2544,8 +2575,8 @@ bool DiscoveryDataBase::from_json(
             }
 
             EPROSIMA_LOG_INFO(DISCOVERY_DATABASE,
-                    "Writer " << guid_aux << " created with instance handle " <<
-                    wit.first->second.change()->instanceHandle);
+                    "Writer " << guid_aux << " created with instance handle "
+                              << wit.first->second.change()->instanceHandle);
 
             if (change->kind != fastdds::rtps::ALIVE)
             {
diff --git a/test/blackbox/common/BlackboxTestsDiscovery.cpp b/test/blackbox/common/BlackboxTestsDiscovery.cpp
@@ -2202,6 +2202,139 @@ TEST_P(Discovery, discovery_server_pdp_messages_sent)
     EXPECT_EQ(num_data_p_sends.load(std::memory_order::memory_order_seq_cst), 9u);
 }
 
+// This test checks that a client connected to a Discovery Server does not receive
+// discovery information about clients connected to a different server from a server
+// different than their own.
+TEST(Discovery, discovery_server_no_external_to_external_relay)
+{
+    using namespace eprosima::fastdds::dds;
+
+    // Define a specific port for client_1's metatraffic
+    uint32_t client_1_metatraffic_port = global_port + 10;
+
+    // Declare a test transport that will count messages sent from server_2 to client_1
+    std::atomic<size_t> num_messages_to_client_1{ 0 };
+
+    auto test_transport_server_2 = std::make_shared<test_UDPv4TransportDescriptor>();
+    test_transport_server_2->locator_filter_ = [&](
+        const eprosima::fastdds::rtps::Locator& destination, int32_t)
+            {
+                // Check if destination port matches client_1's metatraffic port
+                if (destination.port == client_1_metatraffic_port)
+                {
+                    std::cout << "Message from server_2 to client_1 detected on port "
+                              << destination.port << std::endl;
+                    num_messages_to_client_1.fetch_add(1u, std::memory_order_seq_cst);
+                }
+                return false; // Don't drop the message
+            };
+
+    // Create server 1
+    auto server_1 = std::make_shared<PubSubParticipant<HelloWorldPubSubType>>(0, 0, 0, 0);
+
+    Locator_t locator_server_1;  // UDPv4 locator by default
+    eprosima::fastdds::rtps::IPLocator::setIPv4(locator_server_1, 127, 0, 0, 1);
+    eprosima::fastdds::rtps::IPLocator::setPhysicalPort(locator_server_1, global_port);
+
+    WireProtocolConfigQos server_wp_qos_1;
+    server_wp_qos_1.builtin.discovery_config.discoveryProtocol = DiscoveryProtocol::SERVER;
+    server_wp_qos_1.builtin.metatrafficUnicastLocatorList.push_back(locator_server_1);
+    server_wp_qos_1.builtin.discovery_config.initial_announcements.count = 1;
+
+    server_1->wire_protocol(server_wp_qos_1)
+            .setup_transports(eprosima::fastdds::rtps::BuiltinTransports::UDPv4);
+
+    // Start server 1
+    ASSERT_TRUE(server_1->init_participant());
+
+    // Create server 2 (federated with server 1) with test transport
+    auto server_2 = std::make_shared<PubSubParticipant<HelloWorldPubSubType>>(0, 0, 0, 0);
+
+    Locator_t locator_server_2;  // UDPv4 locator by default
+    eprosima::fastdds::rtps::IPLocator::setIPv4(locator_server_2, 127, 0, 0, 1);
+    eprosima::fastdds::rtps::IPLocator::setPhysicalPort(locator_server_2, global_port + 1);
+
+    WireProtocolConfigQos server_wp_qos_2;
+    server_wp_qos_2.builtin.discovery_config.discoveryProtocol = DiscoveryProtocol::SERVER;
+    server_wp_qos_2.builtin.metatrafficUnicastLocatorList.push_back(locator_server_2);
+    server_wp_qos_2.builtin.discovery_config.initial_announcements.count = 1;
+    // Federate server 2 with server 1
+    server_wp_qos_2.builtin.discovery_config.m_DiscoveryServers.push_back(locator_server_1);
+
+    server_2->disable_builtin_transport()
+            .add_user_transport_to_pparams(test_transport_server_2)
+            .wire_protocol(server_wp_qos_2);
+
+    // Start server 2
+    ASSERT_TRUE(server_2->init_participant());
+
+    // Wait for servers to discover each other
+    server_1->wait_discovery(std::chrono::seconds(5), 1, true);
+    server_2->wait_discovery(std::chrono::seconds(5), 1, true);
+
+    // Record baseline message count after server federation
+    size_t baseline_count = num_messages_to_client_1.load(std::memory_order_seq_cst);
+    ASSERT_EQ(baseline_count, 0u); // No messages should ever be sent from server2 to client1
+
+    // Create client 1 connected ONLY to server 1 with a specific metatraffic port
+    PubSubWriter<HelloWorldPubSubType> client_1(TEST_TOPIC_NAME);
+
+    // Set up client_1's metatraffic unicast locator with a specific port
+    Locator_t client_1_metatraffic_locator;
+    eprosima::fastdds::rtps::IPLocator::setIPv4(client_1_metatraffic_locator, 127, 0, 0, 1);
+    client_1_metatraffic_locator.port = client_1_metatraffic_port;
+
+    WireProtocolConfigQos client_1_qos;
+    client_1_qos.builtin.discovery_config.discoveryProtocol = DiscoveryProtocol::CLIENT;
+    client_1_qos.builtin.discovery_config.m_DiscoveryServers.push_back(locator_server_1);
+    client_1_qos.builtin.discovery_config.initial_announcements.count = 1;
+    client_1_qos.builtin.metatrafficUnicastLocatorList.push_back(client_1_metatraffic_locator);
+
+    client_1.set_wire_protocol_qos(client_1_qos)
+            .setup_transports(eprosima::fastdds::rtps::BuiltinTransports::UDPv4)
+            .init();
+
+    ASSERT_TRUE(client_1.isInitialized());
+
+    // Wait for server_1 to discover client_1
+    server_1->wait_discovery(std::chrono::seconds(5), 2, true);  // server_2 + client_1
+
+    // Record message count after client_1 connects
+    size_t messages_after_client_1 = num_messages_to_client_1.load(std::memory_order_seq_cst);
+    ASSERT_EQ(messages_after_client_1, 0u); // No messages should ever be sent from server2 to client1
+
+    // Create client 2 connected ONLY to server 2
+    PubSubReader<HelloWorldPubSubType> client_2(TEST_TOPIC_NAME);
+
+    WireProtocolConfigQos client_2_qos;
+    client_2_qos.builtin.discovery_config.discoveryProtocol = DiscoveryProtocol::CLIENT;
+    client_2_qos.builtin.discovery_config.m_DiscoveryServers.push_back(locator_server_2);
+    client_2_qos.builtin.discovery_config.initial_announcements.count = 1;
+
+    client_2.set_wire_protocol_qos(client_2_qos)
+            .setup_transports(eprosima::fastdds::rtps::BuiltinTransports::UDPv4)
+            .init();
+
+    ASSERT_TRUE(client_2.isInitialized());
+
+    // Wait for server_2 to discover client_2
+    server_2->wait_discovery(std::chrono::seconds(5), 2, true);  // server_1 + client_2
+
+    // Give enough time for any potential (incorrect) relay of discovery information
+    std::this_thread::sleep_for(std::chrono::seconds(5));
+
+    // Record final message count
+    size_t messages_final = num_messages_to_client_1.load(std::memory_order_seq_cst);
+
+    // Verify that server_2 did NOT send messages directly to client_1 after client_2 connected
+    EXPECT_EQ(messages_final, baseline_count)
+        << "Server_2 sent " << (messages_final - baseline_count) << " messages directly to client_1. "
+        << "This suggests external-to-external relay is happening. "
+        << "Baseline: " << baseline_count
+        << ", After client_1: " << messages_after_client_1
+        << ", Final: " << messages_final;
+}
+
 // This test checks that a Discover Server does not send duplicated EDP messages when its routine
 // is triggered by EDP Listeners while it waits for ACKs
 TEST_P(Discovery, discovery_server_edp_messages_sent)
diff --git a/test/blackbox/common/DDSBlackboxTestsDiscovery.cpp b/test/blackbox/common/DDSBlackboxTestsDiscovery.cpp
@@ -2050,9 +2050,9 @@ TEST(DDSDiscovery, client_server_participants_with_different_domain_ids_discover
     client_domain_3.init();
     EXPECT_TRUE(client_domain_3.isInitialized());
 
-    server_domain_1.wait_discovery(std::chrono::seconds(2));
-    client_domain_2.wait_discovery(std::chrono::seconds(2));
-    client_domain_3.wait_discovery(std::chrono::seconds(2));
+    server_domain_1.wait_discovery(std::chrono::seconds(5));
+    client_domain_2.wait_discovery(2, std::chrono::seconds(5));
+    client_domain_3.wait_discovery(std::chrono::seconds(5));
 
     ASSERT_TRUE(client_domain_2.is_matched());
     ASSERT_TRUE(client_domain_3.is_matched());
