From b74c29e053fa649e0aa9cd08e2a00e90dfd0ac78 Mon Sep 17 00:00:00 2001 From: Carlos Aguero Date: Thu, 30 Jul 2026 18:50:29 +0200 Subject: [PATCH 01/13] Maintain the remote subscribers cache instead of clearing it on every TopicList() call Assisted-by: Claude Fable 5 Signed-off-by: Carlos Aguero --- src/Discovery.hh | 28 ++++++++++++++++++++-------- 1 file changed, 20 insertions(+), 8 deletions(-) diff --git a/src/Discovery.hh b/src/Discovery.hh index efdc4c71c..0037de487 100644 --- a/src/Discovery.hh +++ b/src/Discovery.hh @@ -66,6 +66,7 @@ #include #include #include +#include #include #include @@ -817,17 +818,20 @@ namespace gz /// \param[out] _topics List of advertised topics. public: void TopicList(std::vector &_topics) { - if (!this->useZenoh) + // Request the list of subscribers. This request is only meaningful + // for message discovery over UDP: the Zenoh backend keeps + // remoteSubscribers updated via liveliness tokens and nothing + // answers this request on the service discovery channel. + if constexpr (std::is_same_v) { - std::lock_guard lock(this->mutex); - this->remoteSubscribers.Clear(); + if (!this->useZenoh) + { + Publisher pub("", "", this->pUuid, "", AdvertiseOptions()); + this->SendMsg( + DestinationType::ALL, msgs::Discovery::SUBSCRIBERS_REQ, pub); + } } - // Request the list of subscribers. - Publisher pub("", "", this->pUuid, "", AdvertiseOptions()); - this->SendMsg( - DestinationType::ALL, msgs::Discovery::SUBSCRIBERS_REQ, pub); - this->WaitForInit(); std::lock_guard lock(this->mutex); this->info.TopicList(_topics); @@ -889,6 +893,7 @@ namespace gz { // Remove all the info entries for this process UUID. this->info.DelPublishersByProc(it->first); + this->remoteSubscribers.DelPublishersByProc(it->first); uuids.push_back(it->first); @@ -1305,6 +1310,12 @@ namespace gz Pub publisher; publisher.SetFromDiscovery(msg); + { + std::lock_guard lock(this->mutex); + this->remoteSubscribers.DelPublisherByNode( + publisher.Topic(), publisher.PUuid(), publisher.NUuid()); + } + if (unregisterCb) unregisterCb(publisher); @@ -1335,6 +1346,7 @@ namespace gz { std::lock_guard lock(this->mutex); this->info.DelPublishersByProc(recvPUuid); + this->remoteSubscribers.DelPublishersByProc(recvPUuid); } break; From b2c43a091deeff7ab962b5fe821df7671c4dabb7 Mon Sep 17 00:00:00 2001 From: Carlos Aguero Date: Thu, 30 Jul 2026 18:50:29 +0200 Subject: [PATCH 02/13] Include topics subscribed within this process in Node::TopicList() Assisted-by: Claude Fable 5 Signed-off-by: Carlos Aguero --- src/Node.cc | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) diff --git a/src/Node.cc b/src/Node.cc index ff150cfb4..e6985164e 100644 --- a/src/Node.cc +++ b/src/Node.cc @@ -824,6 +824,23 @@ void Node::TopicList(std::vector &_topics) const this->dataPtr->shared->dataPtr->msgDiscovery->TopicList(allTopics); + // Add the topics subscribed within this process. They are not part of the + // discovery information because a process discards its own discovery + // messages. + { + std::lock_guard lock(this->dataPtr->shared->mutex); + for (const auto &pub : this->dataPtr->shared->localSubscribers.Convert( + this->dataPtr->shared->dataPtr->myAddress, + this->dataPtr->shared->pUuid)) + { + if (std::find(allTopics.begin(), allTopics.end(), pub.Topic()) == + allTopics.end()) + { + allTopics.push_back(pub.Topic()); + } + } + } + for (const auto &fullyQualifiedTopic : allTopics) { std::string partition; From a17b00c7bb2a0f0af67f06886fee2f39fd9fd40d Mon Sep 17 00:00:00 2001 From: Carlos Aguero Date: Thu, 30 Jul 2026 18:50:29 +0200 Subject: [PATCH 03/13] Add regression tests for subscriber only topics in TopicList() Assisted-by: Claude Fable 5 Signed-off-by: Carlos Aguero --- src/Node_TEST.cc | 16 +++++ test/CMakeLists.txt | 1 + test/integration/CMakeLists.txt | 1 + .../test_executables/subscriberOnly_aux.cc | 59 +++++++++++++++++++ test/integration/twoProcsPubSub.cc | 25 ++++++++ test/test_config.hh.in | 4 ++ 6 files changed, 106 insertions(+) create mode 100644 test/integration/test_executables/subscriberOnly_aux.cc diff --git a/src/Node_TEST.cc b/src/Node_TEST.cc index 0c5337d55..f048cd7b4 100644 --- a/src/Node_TEST.cc +++ b/src/Node_TEST.cc @@ -2440,6 +2440,22 @@ TEST(NodeTest, TopicListRemap) EXPECT_EQ(g_topic_remap, topics.at(0)); } +////////////////////////////////////////////////// +/// \brief This test creates a node that subscribes to a topic without any +/// publisher. The test verifies that TopicList() includes topics that are +/// only subscribed within this process. +TEST(NodeTest, TopicListSubscriberOnly) +{ + std::vector topics; + transport::Node node; + + EXPECT_TRUE(node.Subscribe(g_topic, cb)); + + node.TopicList(topics); + ASSERT_EQ(1u, topics.size()); + EXPECT_EQ(g_topic, topics.at(0)); +} + ////////////////////////////////////////////////// /// \brief This test creates two nodes and advertises some services. The test /// verifies that ServiceList() returns the list of all the services advertised. diff --git a/test/CMakeLists.txt b/test/CMakeLists.txt index 0c1bb56e0..cddcdbfb0 100644 --- a/test/CMakeLists.txt +++ b/test/CMakeLists.txt @@ -22,6 +22,7 @@ target_compile_definitions(test_config INTERFACE "PUB_EXE=\"$\"" "PUB_THROTTLED_EXE=\"$\"" "SCOPED_TOPIC_SUBSCRIBER_EXE=\"$\"" + "SUBSCRIBER_ONLY_EXE=\"$\"" "TWO_PROCS_PUBLISHER_EXE=\"$\"" "TWO_PROCS_PUB_SUB_MIXED_SUBSCRIBERS_EXE=\"$\"" "TWO_PROCS_PUB_SUB_SINGLE_SUBSCRIBER_EXE=\"$\"" diff --git a/test/integration/CMakeLists.txt b/test/integration/CMakeLists.txt index 07ad49b43..b51dae7a9 100644 --- a/test/integration/CMakeLists.txt +++ b/test/integration/CMakeLists.txt @@ -44,6 +44,7 @@ set(auxiliary_files pub_aux pub_aux_throttled scopedTopicSubscriber_aux + subscriberOnly_aux twoProcsPublisher_aux twoProcsPubSubMixedSubscribers_aux twoProcsPubSubSingleSubscriber_aux diff --git a/test/integration/test_executables/subscriberOnly_aux.cc b/test/integration/test_executables/subscriberOnly_aux.cc new file mode 100644 index 000000000..8c2f04dc8 --- /dev/null +++ b/test/integration/test_executables/subscriberOnly_aux.cc @@ -0,0 +1,59 @@ +/* + * Copyright (C) 2026 Open Source Robotics Foundation + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * +*/ + +#include + +#include +#include +#include +#include + +#include "gz/transport/Node.hh" + +#include + +#include "test_config.hh" + +using namespace gz; + +static const std::string g_topic = "/subscriber_only"; // NOLINT(*) + +////////////////////////////////////////////////// +/// \brief A callback that is never expected to be executed because nobody +/// publishes on this topic. +void cb(const msgs::Vector3d &) +{ +} + +////////////////////////////////////////////////// +int main(int argc, char **argv) +{ + if (argc != 2) + { + std::cerr << "Partition name has not be passed as argument" << std::endl; + return -1; + } + + // Set the partition name for this test. + gz::utils::setenv("GZ_PARTITION", argv[1]); + + // Subscribe to a topic without any publisher and stay alive for a while, + // giving the test process time to discover this subscription. + transport::Node node; + node.Subscribe(g_topic, cb); + std::this_thread::sleep_for(std::chrono::seconds(10)); +} diff --git a/test/integration/twoProcsPubSub.cc b/test/integration/twoProcsPubSub.cc index df871f3fe..b8144ef6f 100644 --- a/test/integration/twoProcsPubSub.cc +++ b/test/integration/twoProcsPubSub.cc @@ -365,6 +365,31 @@ TEST(twoProcPubSub, TopicList) reset(); } +////////////////////////////////////////////////// +/// \brief This test spawns a process that only subscribes to a topic, without +/// any publisher involved. The test verifies that a node whose discovery is +/// already initialized eventually reports the topic in TopicList(). +TEST(twoProcPubSub, TopicListSubscriberOnly) +{ + reset(); + + transport::Node node; + std::vector topics; + + // Make sure that discovery is initialized before spawning the subscriber, + // so this test does not benefit from the initialization wait inside the + // first TopicList() call. + node.TopicList(topics); + + auto pi = testing::SubprocessJoinWrapper( + {test_executables::kSubscriberOnly, partition}); + + EXPECT_TRUE(transport::waitForTopic(node, "/subscriber_only", + std::chrono::milliseconds(5000))); + + reset(); +} + ////////////////////////////////////////////////// /// \brief This test spawns two nodes on different processes. One of the nodes /// advertises a topic and the other uses TopicInfo() for getting information diff --git a/test/test_config.hh.in b/test/test_config.hh.in index 406c9679d..b98683f52 100644 --- a/test/test_config.hh.in +++ b/test/test_config.hh.in @@ -59,6 +59,10 @@ constexpr const char * kPubThrottled = PUB_THROTTLED_EXE; constexpr const char * kScopedTopicSubscriber = SCOPED_TOPIC_SUBSCRIBER_EXE; #endif // SCOPED_TOPIC_SUBSCRIBER_EXE +#ifdef SUBSCRIBER_ONLY_EXE +constexpr const char * kSubscriberOnly = SUBSCRIBER_ONLY_EXE; +#endif // SUBSCRIBER_ONLY_EXE + #ifdef TWO_PROCS_PUBLISHER_EXE constexpr const char * kTwoProcsPublisher = TWO_PROCS_PUBLISHER_EXE; #endif // TWO_PROCS_PUBLISHER_EXE From 87cd014dcbdfeff30d7e7fd58926fa62436188ae Mon Sep 17 00:00:00 2001 From: Carlos Aguero Date: Thu, 30 Jul 2026 19:21:45 +0200 Subject: [PATCH 04/13] Include subscribers from this process in Node::TopicInfo() Assisted-by: Claude Fable 5 Signed-off-by: Carlos Aguero --- src/Node.cc | 15 +++++++++++++++ 1 file changed, 15 insertions(+) diff --git a/src/Node.cc b/src/Node.cc index e6985164e..17d4fec7e 100644 --- a/src/Node.cc +++ b/src/Node.cc @@ -1079,6 +1079,21 @@ bool Node::TopicInfo(const std::string &_topic, convert(subs, _subscribers); } + // Add the subscribers within this process. They are not part of the + // discovery information because a process discards its own discovery + // messages. + for (const auto &pub : this->dataPtr->shared->localSubscribers.Convert( + this->dataPtr->shared->dataPtr->myAddress, + this->dataPtr->shared->pUuid)) + { + if (pub.Topic() == fullyQualifiedTopic && + std::find(_subscribers.begin(), _subscribers.end(), pub) == + _subscribers.end()) + { + _subscribers.push_back(pub); + } + } + return true; } From ebad3e0c593d0d627b4eeb437271987fd2148a65 Mon Sep 17 00:00:00 2001 From: Carlos Aguero Date: Thu, 30 Jul 2026 19:21:45 +0200 Subject: [PATCH 05/13] Add tests for unsubscribe staleness and TopicInfo() subscribers Assisted-by: Claude Fable 5 Signed-off-by: Carlos Aguero --- src/Node_TEST.cc | 24 ++++++++++ .../test_executables/subscriberOnly_aux.cc | 27 +++++++++-- test/integration/twoProcsPubSub.cc | 47 +++++++++++++++++++ 3 files changed, 93 insertions(+), 5 deletions(-) diff --git a/src/Node_TEST.cc b/src/Node_TEST.cc index f048cd7b4..51d1a7937 100644 --- a/src/Node_TEST.cc +++ b/src/Node_TEST.cc @@ -2454,6 +2454,30 @@ TEST(NodeTest, TopicListSubscriberOnly) node.TopicList(topics); ASSERT_EQ(1u, topics.size()); EXPECT_EQ(g_topic, topics.at(0)); + + // After unsubscribing, the topic should not be listed anymore. + EXPECT_TRUE(node.Unsubscribe(g_topic)); + topics.clear(); + node.TopicList(topics); + EXPECT_TRUE(topics.empty()); +} + +////////////////////////////////////////////////// +/// \brief This test creates a node that subscribes to a topic without any +/// publisher. The test verifies that TopicInfo() includes the subscribers +/// from this process. +TEST(NodeTest, TopicInfoSubscriberOnly) +{ + transport::Node node; + + EXPECT_TRUE(node.Subscribe(g_topic, cb)); + + std::vector publishers; + std::vector subscribers; + EXPECT_TRUE(node.TopicInfo(g_topic, publishers, subscribers)); + EXPECT_EQ(publishers.size(), 0u); + ASSERT_EQ(subscribers.size(), 1u); + EXPECT_EQ(subscribers.front().MsgTypeName(), "gz.msgs.Int32"); } ////////////////////////////////////////////////// diff --git a/test/integration/test_executables/subscriberOnly_aux.cc b/test/integration/test_executables/subscriberOnly_aux.cc index 8c2f04dc8..df9a4834b 100644 --- a/test/integration/test_executables/subscriberOnly_aux.cc +++ b/test/integration/test_executables/subscriberOnly_aux.cc @@ -40,9 +40,15 @@ void cb(const msgs::Vector3d &) } ////////////////////////////////////////////////// +/// \brief Usage: subscriberOnly_aux [topic] [lifetimeSec] +/// [unsubscribeAfterSec]. +/// Subscribe to a topic without any publisher and stay alive for +/// lifetimeSec, giving the test process time to discover this subscription. +/// If unsubscribeAfterSec is positive, unsubscribe after that time while +/// keeping the process alive until lifetimeSec. int main(int argc, char **argv) { - if (argc != 2) + if (argc < 2 || argc > 5) { std::cerr << "Partition name has not be passed as argument" << std::endl; return -1; @@ -51,9 +57,20 @@ int main(int argc, char **argv) // Set the partition name for this test. gz::utils::setenv("GZ_PARTITION", argv[1]); - // Subscribe to a topic without any publisher and stay alive for a while, - // giving the test process time to discover this subscription. + const std::string topic = argc > 2 ? argv[2] : g_topic; + const int lifetimeSec = argc > 3 ? std::stoi(argv[3]) : 10; + const int unsubscribeAfterSec = argc > 4 ? std::stoi(argv[4]) : 0; + transport::Node node; - node.Subscribe(g_topic, cb); - std::this_thread::sleep_for(std::chrono::seconds(10)); + node.Subscribe(topic, cb); + + int elapsedSec = 0; + if (unsubscribeAfterSec > 0) + { + std::this_thread::sleep_for(std::chrono::seconds(unsubscribeAfterSec)); + node.Unsubscribe(topic); + elapsedSec = unsubscribeAfterSec; + } + + std::this_thread::sleep_for(std::chrono::seconds(lifetimeSec - elapsedSec)); } diff --git a/test/integration/twoProcsPubSub.cc b/test/integration/twoProcsPubSub.cc index b8144ef6f..cfd7eec55 100644 --- a/test/integration/twoProcsPubSub.cc +++ b/test/integration/twoProcsPubSub.cc @@ -17,6 +17,7 @@ #include #include +#include #include #include #include @@ -387,6 +388,52 @@ TEST(twoProcPubSub, TopicListSubscriberOnly) EXPECT_TRUE(transport::waitForTopic(node, "/subscriber_only", std::chrono::milliseconds(5000))); + // The remote subscription should also be visible in TopicInfo(). + std::vector publishers; + std::vector subscribers; + EXPECT_TRUE(node.TopicInfo("/subscriber_only", publishers, subscribers)); + EXPECT_EQ(publishers.size(), 0u); + EXPECT_EQ(subscribers.size(), 1u); + + reset(); +} + +////////////////////////////////////////////////// +/// \brief This test spawns a process that subscribes to a topic and +/// unsubscribes after a while, keeping the process alive. The test verifies +/// that TopicList() stops reporting the topic after the unsubscription. +TEST(twoProcPubSub, TopicListRemovesUnsubscribed) +{ + reset(); + + transport::Node node; + std::vector topics; + + // Make sure that discovery is initialized before spawning the subscriber. + node.TopicList(topics); + + // The remote process subscribes to /unsub_test, unsubscribes after 4 + // seconds and stays alive for 15 seconds. + auto pi = testing::SubprocessJoinWrapper( + {test_executables::kSubscriberOnly, partition, "/unsub_test", "15", "4"}); + + ASSERT_TRUE(transport::waitForTopic(node, "/unsub_test", + std::chrono::milliseconds(4000))); + + // After the remote node unsubscribes, the topic should disappear from + // TopicList() while its process is still alive. Note that the timeout is + // shorter than the remaining lifetime of the remote process, so a topic + // removal triggered by the process termination cannot make this pass. + auto topicAbsent = [&node]() + { + std::vector topicsNow; + node.TopicList(topicsNow); + return std::find(topicsNow.begin(), topicsNow.end(), "/unsub_test") == + topicsNow.end(); + }; + EXPECT_TRUE(transport::waitUntil(topicAbsent, + std::chrono::milliseconds(8000), std::chrono::milliseconds(100))); + reset(); } From 277c1599e9f90a3eb180563c70b2bc3d3b8f853b Mon Sep 17 00:00:00 2001 From: Carlos Aguero Date: Thu, 30 Jul 2026 19:54:57 +0200 Subject: [PATCH 06/13] [bazel] Add subscriberOnly_aux to the test build Assisted-by: Claude Fable 5 Signed-off-by: Carlos Aguero --- test/BUILD.bazel | 2 ++ 1 file changed, 2 insertions(+) diff --git a/test/BUILD.bazel b/test/BUILD.bazel index bf4122f36..1ada6a345 100644 --- a/test/BUILD.bazel +++ b/test/BUILD.bazel @@ -36,6 +36,7 @@ test_executables = [ "pub_aux", "pub_aux_throttled", "scopedTopicSubscriber_aux", + "subscriberOnly_aux", "twoProcsPublisher_aux", "twoProcsPubSubMixedSubscribers_aux", "twoProcsPubSubSingleSubscriber_aux", @@ -96,6 +97,7 @@ flaky_test_srcs = [ 'PUB_EXE=\\"./test/pub_aux\\"', 'PUB_THROTTLED_EXE=\\"./test/pub_aux_throttled\\"', 'SCOPED_TOPIC_SUBSCRIBER_EXE=\\"./test/scopedTopicSubscriber_aux\\"', + 'SUBSCRIBER_ONLY_EXE=\\"./test/subscriberOnly_aux\\"', 'TWO_PROCS_PUBLISHER_EXE=\\"./test/twoProcsPublisher_aux\\"', 'TWO_PROCS_PUB_SUB_MIXED_SUBSCRIBERS_EXE=\\"./test/twoProcsPubSubMixedSubscribers_aux\\"', 'TWO_PROCS_PUB_SUB_SINGLE_SUBSCRIBER_EXE=\\"./test/twoProcsPubSubSingleSubscriber_aux\\"', From deb2bd5570bf4f628e8dbe394265368634e90e84 Mon Sep 17 00:00:00 2001 From: Carlos Aguero Date: Thu, 30 Jul 2026 22:09:21 +0200 Subject: [PATCH 07/13] Announce subscriptions in SUBSCRIBE messages and sync subscribers at discovery startup Assisted-by: Claude Fable 5 Signed-off-by: Carlos Aguero --- src/Discovery.hh | 33 +++++++++- src/NodeShared.cc | 5 +- test/BUILD.bazel | 2 + test/integration/CMakeLists.txt | 1 + test/integration/topicListFirstCall.cc | 89 ++++++++++++++++++++++++++ 5 files changed, 127 insertions(+), 3 deletions(-) create mode 100644 test/integration/topicListFirstCall.cc diff --git a/src/Discovery.hh b/src/Discovery.hh index 0037de487..c48edca3c 100644 --- a/src/Discovery.hh +++ b/src/Discovery.hh @@ -457,6 +457,16 @@ namespace gz // Start the thread that receives discovery information. this->threadReception = std::thread(&Discovery::RecvMessages, this); + + // Request the list of current subscribers so that the cache is + // already complete when the initialization phase finishes. New + // subscriptions are tracked through the SUBSCRIBE announcements. + if constexpr (std::is_same_v) + { + Publisher pub("", "", this->pUuid, "", AdvertiseOptions()); + this->SendMsg( + DestinationType::ALL, msgs::Discovery::SUBSCRIBERS_REQ, pub); + } } #ifdef HAVE_ZENOH @@ -518,9 +528,13 @@ namespace gz /// \sa SetConnectionsCb. /// \sa SetDisconnectionsCb. /// \param[in] _topic Topic name requested. + /// \param[in] _nUuid Node UUID of the subscriber requesting the topic, + /// announced so that other processes can track this subscription. + /// Empty when the request is not tied to a subscription. /// \return True if the method succeeded or false otherwise /// (e.g. if the discovery has not been started). - public: bool Discover(const std::string &_topic) const + public: bool Discover(const std::string &_topic, + const std::string &_nUuid = "") const { DiscoveryCallback cb; bool found; @@ -538,6 +552,7 @@ namespace gz Pub pub; pub.SetTopic(_topic); pub.SetPUuid(this->pUuid); + pub.SetNUuid(_nUuid); // Send a discovery request. this->SendMsg(DestinationType::ALL, msgs::Discovery::SUBSCRIBE, pub); @@ -1244,6 +1259,21 @@ namespace gz break; } + // Register the remote subscriber. Subscribers running an older + // version do not announce their node UUID and are only tracked + // through the SUBSCRIBERS_REQ mechanism. + if constexpr (std::is_same_v) + { + if (!msg.sub().n_uuid().empty()) + { + Pub subscriber(recvTopic, "", "", recvPUuid, + msg.sub().n_uuid(), kGenericMessageType, + AdvertiseMessageOptions()); + std::lock_guard lock(this->mutex); + this->remoteSubscribers.AddPublisher(subscriber); + } + } + // Check if at least one of my nodes advertises the topic requested. Addresses_M addresses; { @@ -1418,6 +1448,7 @@ namespace gz case msgs::Discovery::SUBSCRIBE: { discoveryMsg.mutable_sub()->set_topic(_pub.Topic()); + discoveryMsg.mutable_sub()->set_n_uuid(_pub.NUuid()); break; } case msgs::Discovery::HEARTBEAT: diff --git a/src/NodeShared.cc b/src/NodeShared.cc index 64b192ee6..218e2d287 100644 --- a/src/NodeShared.cc +++ b/src/NodeShared.cc @@ -2022,11 +2022,12 @@ bool NodeShared::SubscribeHelper(const std::string &_fullyQualifiedTopic, this->dataPtr->topicsSubscribed[_nUuid].insert(_fullyQualifiedTopic); } - // Discover the list of nodes that publish on the topic. + // Discover the list of nodes that publish on the topic. The node UUID is + // announced so that other processes can track this subscription. std::string impl = this->GzImplementation(); if (impl == "zeromq") { - return this->dataPtr->msgDiscovery->Discover(_fullyQualifiedTopic); + return this->dataPtr->msgDiscovery->Discover(_fullyQualifiedTopic, _nUuid); } return true; } diff --git a/test/BUILD.bazel b/test/BUILD.bazel index 1ada6a345..9a374f9cf 100644 --- a/test/BUILD.bazel +++ b/test/BUILD.bazel @@ -66,6 +66,7 @@ integration_test_sources = [ "scopedTopic.cc", "callback_scope_TEST.cc", "statistics.cc", + "topicListFirstCall.cc", "twoProcsPubSub.cc", "twoProcsPubSubStats.cc", "twoProcsSrvCall.cc", @@ -81,6 +82,7 @@ integration_test_sources = [ # Found with # bazel test test:all --runs_per_test 10 flaky_test_srcs = [ + "topicListFirstCall.cc", "twoProcsPubSub.cc", "twoProcsSrvCall.cc", "twoProcsSrvCallWithoutInput.cc", diff --git a/test/integration/CMakeLists.txt b/test/integration/CMakeLists.txt index b51dae7a9..e318e7a33 100644 --- a/test/integration/CMakeLists.txt +++ b/test/integration/CMakeLists.txt @@ -5,6 +5,7 @@ set(tests scopedTopic.cc callback_scope_TEST.cc statistics.cc + topicListFirstCall.cc twoProcsPubSub.cc twoProcsPubSubStats.cc twoProcsSrvCall.cc diff --git a/test/integration/topicListFirstCall.cc b/test/integration/topicListFirstCall.cc new file mode 100644 index 000000000..0613ab8df --- /dev/null +++ b/test/integration/topicListFirstCall.cc @@ -0,0 +1,89 @@ +/* + * Copyright (C) 2026 Open Source Robotics Foundation + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * +*/ + +#include +#include +#include +#include +#include + +#include "gz/transport/Node.hh" + +#include +#include + +#include "gtest/gtest.h" +#include "test_config.hh" +#include "test_utils.hh" + +using namespace gz; + +static std::string partition; // NOLINT(*) + +////////////////////////////////////////////////// +/// \brief This test spawns a process that only subscribes to a topic. The +/// test verifies that the first TopicList() call of an initialized node +/// already includes the topic thanks to the reply collection window. This +/// test needs its own process because the collection window only applies to +/// the first TopicList() call of a process. +TEST(topicListFirstCall, SubscriberInFirstCall) +{ + transport::Node node; + + // Let discovery initialize without calling TopicList(). + std::this_thread::sleep_for(std::chrono::seconds(4)); + + auto pi = testing::SubprocessJoinWrapper( + {test_executables::kSubscriberOnly, partition}); + + // Give the remote process time to start and subscribe. + std::this_thread::sleep_for(std::chrono::seconds(2)); + + // The first call should collect the remote subscriber replies. + auto start = std::chrono::steady_clock::now(); + std::vector topics; + node.TopicList(topics); + auto elapsed = std::chrono::steady_clock::now() - start; + + EXPECT_TRUE(std::find(topics.begin(), topics.end(), "/subscriber_only") != + topics.end()); + + // The collection window is bounded. + EXPECT_LT(std::chrono::duration_cast( + elapsed).count(), 2000); + + // The second call should never block. + topics.clear(); + auto start2 = std::chrono::steady_clock::now(); + node.TopicList(topics); + auto elapsed2 = std::chrono::steady_clock::now() - start2; + EXPECT_LT(std::chrono::duration_cast( + elapsed2).count(), 2); +} + +////////////////////////////////////////////////// +int main(int argc, char **argv) +{ + // Get a random partition name. + partition = testing::getRandomNumber(); + + // Set the partition name for this process. + gz::utils::setenv("GZ_PARTITION", partition); + + ::testing::InitGoogleTest(&argc, argv); + return RUN_ALL_TESTS(); +} From 073c007fb0a924e3c6e6fb3cb7ada94b261f9421 Mon Sep 17 00:00:00 2001 From: Carlos Aguero Date: Thu, 30 Jul 2026 23:36:33 +0200 Subject: [PATCH 08/13] Report subscribers with generation tagged snapshots and wait for their completion in TopicList() Assisted-by: Claude Fable 5 Signed-off-by: Carlos Aguero --- src/Discovery.hh | 157 +++++++++++++++++++++++-- src/NodeShared.cc | 23 +++- src/NodeSharedPrivate.hh | 5 + src/Node_TEST.cc | 5 +- test/integration/topicListFirstCall.cc | 10 +- test/integration/twoProcsPubSub.cc | 5 +- 6 files changed, 185 insertions(+), 20 deletions(-) diff --git a/src/Discovery.hh b/src/Discovery.hh index c48edca3c..03578730a 100644 --- a/src/Discovery.hh +++ b/src/Discovery.hh @@ -60,6 +60,7 @@ #include #include +#include #include #include #include @@ -583,12 +584,35 @@ namespace gz return true; } - /// \brief Send the response to a SUBSCRIBERS_REQ message. - /// \param[in] _pub Information to send. - public: void SendSubscribersRep(const MessagePublisher &_pub) const + /// \brief Send one response of the burst answering a SUBSCRIBERS_REQ + /// message. The burst is a snapshot of all the subscriptions of this + /// process: the receiver knows that the snapshot is complete when + /// _count messages have been received. + /// \param[in] _pub Information to send. An empty topic is used to + /// answer when the process has no subscriptions (_count is zero). + /// \param[in] _count Number of messages in this burst. + /// \param[in] _generation Subscription generation of this process when + /// the snapshot was taken. + public: void SendSubscribersRep(const MessagePublisher &_pub, + const uint32_t _count, + const uint64_t _generation) const { - this->SendMsg( - DestinationType::ALL, msgs::Discovery::SUBSCRIBERS_REP, _pub); + gz::msgs::Discovery discoveryMsg; + discoveryMsg.set_version(this->Version()); + discoveryMsg.set_type(msgs::Discovery::SUBSCRIBERS_REP); + discoveryMsg.set_process_uuid(this->pUuid); + if (!_pub.Topic().empty()) + _pub.FillDiscovery(discoveryMsg); + + auto *snapshot = discoveryMsg.mutable_subscribers_snapshot(); + snapshot->set_count(_count); + snapshot->set_generation(_generation); + + this->SendMulticast(discoveryMsg); + + // Set the RELAY flag in the header and send to the unicast relays. + discoveryMsg.mutable_flags()->set_relay(true); + this->SendUnicast(discoveryMsg); } /// \brief Register a node from this process as a remote subscriber. @@ -829,10 +853,17 @@ namespace gz } /// \brief Get the list of topics currently advertised and subscribed - /// in the network. + /// in the network. The call blocks until every known process has + /// reported a complete snapshot of its subscribers or a short timeout + /// expires. The wait normally finishes in a few milliseconds, when + /// the last snapshot arrives. /// \param[out] _topics List of advertised topics. public: void TopicList(std::vector &_topics) { + [[maybe_unused]] Timestamp requestTime = + std::chrono::steady_clock::now(); + [[maybe_unused]] std::vector knownProcs; + // Request the list of subscribers. This request is only meaningful // for message discovery over UDP: the Zenoh backend keeps // remoteSubscribers updated via liveliness tokens and nothing @@ -841,6 +872,12 @@ namespace gz { if (!this->useZenoh) { + { + std::lock_guard lock(this->mutex); + for (const auto &proc : this->activity) + knownProcs.push_back(proc.first); + } + Publisher pub("", "", this->pUuid, "", AdvertiseOptions()); this->SendMsg( DestinationType::ALL, msgs::Discovery::SUBSCRIBERS_REQ, pub); @@ -848,7 +885,39 @@ namespace gz } this->WaitForInit(); - std::lock_guard lock(this->mutex); + std::unique_lock lock(this->mutex); + + if constexpr (std::is_same_v) + { + if (!this->useZenoh) + { + // Wait until every process known at request time has reported a + // snapshot after the request or the timeout expires. A lost + // reply or a process running an older version is covered by the + // timeout: the cached information is used instead. A process + // that expired while waiting is not expected to reply. + this->subscribersRepCv.wait_until(lock, + requestTime + + std::chrono::milliseconds(kDefSubscribersRepTimeout), + [&] + { + for (const auto &proc : knownProcs) + { + if (this->activity.find(proc) == this->activity.end()) + continue; + + auto it = this->subscribersSnapshots.find(proc); + if (it == this->subscribersSnapshots.end() || + it->second.completed < requestTime) + { + return false; + } + } + return true; + }); + } + } + this->info.TopicList(_topics); std::vector remoteSubs; @@ -909,6 +978,7 @@ namespace gz // Remove all the info entries for this process UUID. this->info.DelPublishersByProc(it->first); this->remoteSubscribers.DelPublishersByProc(it->first); + this->subscribersSnapshots.erase(it->first); uuids.push_back(it->first); @@ -1319,7 +1389,43 @@ namespace gz { std::lock_guard lock(this->mutex); - this->remoteSubscribers.AddPublisher(publisher); + + if (msg.has_subscribers_snapshot()) + { + const auto &snapshot = msg.subscribers_snapshot(); + auto &progress = this->subscribersSnapshots[recvPUuid]; + + // A new generation starts an authoritative snapshot: + // replace all the entries known for this process. + if (!progress.started || + progress.generation != snapshot.generation()) + { + progress.started = true; + progress.generation = snapshot.generation(); + progress.expected = snapshot.count(); + progress.received = 0; + this->remoteSubscribers.DelPublishersByProc(recvPUuid); + } + + if (!publisher.Topic().empty()) + { + this->remoteSubscribers.AddPublisher(publisher); + ++progress.received; + } + + if (progress.received >= progress.expected) + { + progress.completed = std::chrono::steady_clock::now(); + this->subscribersRepCv.notify_all(); + } + } + else + { + // A process running an older version does not attach the + // snapshot metadata. Its information is merged and its + // completion is covered by the TopicList() timeout. + this->remoteSubscribers.AddPublisher(publisher); + } } break; } @@ -1377,6 +1483,7 @@ namespace gz std::lock_guard lock(this->mutex); this->info.DelPublishersByProc(recvPUuid); this->remoteSubscribers.DelPublishersByProc(recvPUuid); + this->subscribersSnapshots.erase(recvPUuid); } break; @@ -1680,6 +1787,11 @@ namespace gz /// \sa SetHeartbeatInterval. private: static const unsigned int kDefHeartbeatInterval = 1000; + /// \brief Default maximum time waiting for the subscriber snapshots + /// in TopicList() (ms.). The wait normally finishes much earlier, + /// when every known process has answered. + private: static const unsigned int kDefSubscribersRepTimeout = 100; + /// \brief Default silence interval value (ms.). /// \sa MaxSilenceInterval. /// \sa SetMaxSilenceInterval. @@ -1747,6 +1859,35 @@ namespace gz /// \brief Remote subscribers. private: TopicStorage remoteSubscribers; + /// \brief Progress of the snapshot that a remote process reports in + /// a SUBSCRIBERS_REP burst. + private: struct SubscribersSnapshotProgress + { + /// \brief True when at least one burst has been received. + bool started = false; + + /// \brief Subscription generation of the last burst. + uint64_t generation = 0; + + /// \brief Number of messages expected in the burst. + uint32_t expected = 0; + + /// \brief Number of messages received from the burst. + uint32_t received = 0; + + /// \brief Last time a complete snapshot was received. + Timestamp completed = Timestamp::min(); + }; + + /// \brief Snapshot progress of each remote process, keyed by its + /// process UUID. + private: std::map + subscribersSnapshots; + + /// \brief Condition variable notified every time a remote process + /// completes a subscribers snapshot. + private: mutable std::condition_variable subscribersRepCv; + /// \brief Activity information. Every time there is a message from a /// remote node, its activity information is updated. If we do not hear /// from a node in a while, its entries in 'info' will be invalided. The diff --git a/src/NodeShared.cc b/src/NodeShared.cc index 218e2d287..26acc6462 100644 --- a/src/NodeShared.cc +++ b/src/NodeShared.cc @@ -1215,16 +1215,31 @@ void NodeShared::OnSubscribers() { // Get the list of local subscribers while holding the lock. std::vector pubs; + uint64_t generation; { std::lock_guard lock(this->mutex); pubs = this->localSubscribers.Convert( this->dataPtr->myAddress, this->pUuid); + generation = this->dataPtr->subscriptionsGeneration; } - // Reply to the SUBSCRIBERS_REQ with multiple SUBSCRIBERS_REP. + // Reply to the SUBSCRIBERS_REQ with a snapshot of multiple + // SUBSCRIBERS_REP. An empty snapshot is sent when there are no + // subscriptions, so that the requester can account this process. // Called outside the lock to avoid deadlocks with Discovery::mutex. - for (auto const &publisher : pubs) - this->dataPtr->msgDiscovery->SendSubscribersRep(publisher); + if (pubs.empty()) + { + this->dataPtr->msgDiscovery->SendSubscribersRep( + MessagePublisher(), 0, generation); + } + else + { + for (auto const &publisher : pubs) + { + this->dataPtr->msgDiscovery->SendSubscribersRep( + publisher, static_cast(pubs.size()), generation); + } + } } ////////////////////////////////////////////////// @@ -1960,6 +1975,7 @@ bool NodeShared::Unsubscribe(const std::string &_topic, shouldNotifyPublishers = true; localAddress = this->dataPtr->myAddress; localPUuid = this->pUuid; + ++this->dataPtr->subscriptionsGeneration; } // Notify to the publishers that I am no longer interested in the topic. @@ -2020,6 +2036,7 @@ bool NodeShared::SubscribeHelper(const std::string &_fullyQualifiedTopic, std::lock_guard lk(this->mutex); // Add the topic to the list of subscribed topics (if it was not before). this->dataPtr->topicsSubscribed[_nUuid].insert(_fullyQualifiedTopic); + ++this->dataPtr->subscriptionsGeneration; } // Discover the list of nodes that publish on the topic. The node UUID is diff --git a/src/NodeSharedPrivate.hh b/src/NodeSharedPrivate.hh index 689103dc1..1910a16f2 100644 --- a/src/NodeSharedPrivate.hh +++ b/src/NodeSharedPrivate.hh @@ -302,6 +302,11 @@ namespace gz::transport /// \brief When true, the reception thread will finish. public: std::atomic exit = false; + /// \brief Subscription generation of this process. Increased on every + /// subscription change, attached to the subscriber snapshots so that + /// remote processes can identify them. + public: std::atomic subscriptionsGeneration = 0; + /// \brief Timeout used for receiving messages (ms.). public: inline static const int Timeout = 250; diff --git a/src/Node_TEST.cc b/src/Node_TEST.cc index 51d1a7937..551cfae59 100644 --- a/src/Node_TEST.cc +++ b/src/Node_TEST.cc @@ -2414,10 +2414,11 @@ TEST(NodeTest, TopicList) // The first TopicList() call might block if the discovery is still // initializing (it may happen if we run this test alone). - // However, the second call should never block. + // However, the second call should finish as soon as all the known + // processes report their subscribers, well below the internal timeout. auto elapsed = end - start; EXPECT_LT(std::chrono::duration_cast - (elapsed).count(), 2); + (elapsed).count(), 50); } ////////////////////////////////////////////////// diff --git a/test/integration/topicListFirstCall.cc b/test/integration/topicListFirstCall.cc index 0613ab8df..272b098d2 100644 --- a/test/integration/topicListFirstCall.cc +++ b/test/integration/topicListFirstCall.cc @@ -37,9 +37,8 @@ static std::string partition; // NOLINT(*) ////////////////////////////////////////////////// /// \brief This test spawns a process that only subscribes to a topic. The /// test verifies that the first TopicList() call of an initialized node -/// already includes the topic thanks to the reply collection window. This -/// test needs its own process because the collection window only applies to -/// the first TopicList() call of a process. +/// already includes the topic, thanks to the subscription announcements and +/// the subscriber snapshots collected before returning. TEST(topicListFirstCall, SubscriberInFirstCall) { transport::Node node; @@ -66,13 +65,14 @@ TEST(topicListFirstCall, SubscriberInFirstCall) EXPECT_LT(std::chrono::duration_cast( elapsed).count(), 2000); - // The second call should never block. + // The second call should finish as soon as all the known processes + // report their subscribers, well below the internal timeout. topics.clear(); auto start2 = std::chrono::steady_clock::now(); node.TopicList(topics); auto elapsed2 = std::chrono::steady_clock::now() - start2; EXPECT_LT(std::chrono::duration_cast( - elapsed2).count(), 2); + elapsed2).count(), 50); } ////////////////////////////////////////////////// diff --git a/test/integration/twoProcsPubSub.cc b/test/integration/twoProcsPubSub.cc index cfd7eec55..cbff80a04 100644 --- a/test/integration/twoProcsPubSub.cc +++ b/test/integration/twoProcsPubSub.cc @@ -352,7 +352,8 @@ TEST(twoProcPubSub, TopicList) EXPECT_EQ(topics.at(0), g_topic); topics.clear(); - // The second call should never block since discovery already completed. + // The second call should finish as soon as all the known processes + // report their subscribers, well below the internal timeout. auto start2 = std::chrono::steady_clock::now(); node.TopicList(topics); auto end2 = std::chrono::steady_clock::now(); @@ -361,7 +362,7 @@ TEST(twoProcPubSub, TopicList) auto elapsed2 = std::chrono::duration_cast (end2 - start2).count(); - EXPECT_LT(elapsed2, 2); + EXPECT_LT(elapsed2, 50); reset(); } From 13b41bfe1af8d4c296960ad09b9e1559adcf4818 Mon Sep 17 00:00:00 2001 From: Carlos Aguero Date: Fri, 31 Jul 2026 00:41:44 +0200 Subject: [PATCH 09/13] Add startup sync, silent peer timeout, and traffic bound tests for the subscribers protocol Assisted-by: Claude Fable 5 Signed-off-by: Carlos Aguero --- test/BUILD.bazel | 4 + test/integration/CMakeLists.txt | 6 + test/integration/topicListStartupSync.cc | 83 ++++++++ test/integration/topicListTraffic.cc | 256 +++++++++++++++++++++++ 4 files changed, 349 insertions(+) create mode 100644 test/integration/topicListStartupSync.cc create mode 100644 test/integration/topicListTraffic.cc diff --git a/test/BUILD.bazel b/test/BUILD.bazel index 9a374f9cf..dc8702d1b 100644 --- a/test/BUILD.bazel +++ b/test/BUILD.bazel @@ -67,6 +67,8 @@ integration_test_sources = [ "callback_scope_TEST.cc", "statistics.cc", "topicListFirstCall.cc", + "topicListStartupSync.cc", + "topicListTraffic.cc", "twoProcsPubSub.cc", "twoProcsPubSubStats.cc", "twoProcsSrvCall.cc", @@ -83,6 +85,8 @@ integration_test_sources = [ # bazel test test:all --runs_per_test 10 flaky_test_srcs = [ "topicListFirstCall.cc", + "topicListStartupSync.cc", + "topicListTraffic.cc", "twoProcsPubSub.cc", "twoProcsSrvCall.cc", "twoProcsSrvCallWithoutInput.cc", diff --git a/test/integration/CMakeLists.txt b/test/integration/CMakeLists.txt index e318e7a33..c65990eca 100644 --- a/test/integration/CMakeLists.txt +++ b/test/integration/CMakeLists.txt @@ -6,6 +6,7 @@ set(tests callback_scope_TEST.cc statistics.cc topicListFirstCall.cc + topicListStartupSync.cc twoProcsPubSub.cc twoProcsPubSubStats.cc twoProcsSrvCall.cc @@ -18,6 +19,11 @@ set(tests twoProcsSrvCallWithoutOutputStress.cc ) +# The traffic test crafts raw discovery datagrams using POSIX sockets. +if (UNIX AND NOT APPLE) + list(APPEND tests topicListTraffic.cc) +endif() + # Test symbols having the right name on linux only. if (UNIX AND NOT APPLE) configure_file(all_symbols_have_version.bash.in diff --git a/test/integration/topicListStartupSync.cc b/test/integration/topicListStartupSync.cc new file mode 100644 index 000000000..8f3d49009 --- /dev/null +++ b/test/integration/topicListStartupSync.cc @@ -0,0 +1,83 @@ +/* + * Copyright (C) 2026 Open Source Robotics Foundation + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * +*/ + +#include +#include +#include +#include +#include + +#include "gz/transport/Node.hh" + +#include +#include + +#include "gtest/gtest.h" +#include "test_config.hh" +#include "test_utils.hh" + +using namespace gz; + +static std::string partition; // NOLINT(*) + +////////////////////////////////////////////////// +/// \brief This test spawns a process that only subscribes to a topic before +/// this process starts its discovery. The test verifies that the discovery +/// startup requests the existing subscribers, so the first TopicList() call +/// already includes the topic. This test needs its own process because the +/// discovery of a process only starts once. +TEST(topicListStartupSync, PreexistingSubscriberInFirstCall) +{ + // The remote subscriber exists before this process starts its discovery. + auto pi = testing::SubprocessJoinWrapper( + {test_executables::kSubscriberOnly, partition, "/subscriber_only", "12"}); + + std::this_thread::sleep_for(std::chrono::seconds(2)); + + // The first transport node starts the discovery of this process, which + // requests the current subscribers. + transport::Node node; + + std::vector topics; + node.TopicList(topics); + EXPECT_TRUE(std::find(topics.begin(), topics.end(), "/subscriber_only") != + topics.end()); + + // The second call should finish as soon as all the known processes + // report their subscribers, well below the internal timeout. + topics.clear(); + auto start = std::chrono::steady_clock::now(); + node.TopicList(topics); + auto elapsed = std::chrono::steady_clock::now() - start; + EXPECT_TRUE(std::find(topics.begin(), topics.end(), "/subscriber_only") != + topics.end()); + EXPECT_LT(std::chrono::duration_cast( + elapsed).count(), 50); +} + +////////////////////////////////////////////////// +int main(int argc, char **argv) +{ + // Get a random partition name. + partition = testing::getRandomNumber(); + + // Set the partition name for this process. + gz::utils::setenv("GZ_PARTITION", partition); + + ::testing::InitGoogleTest(&argc, argv); + return RUN_ALL_TESTS(); +} diff --git a/test/integration/topicListTraffic.cc b/test/integration/topicListTraffic.cc new file mode 100644 index 000000000..12b41769d --- /dev/null +++ b/test/integration/topicListTraffic.cc @@ -0,0 +1,256 @@ +/* + * Copyright (C) 2026 Open Source Robotics Foundation + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * +*/ + +#include + +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include + +#include "gz/transport/Node.hh" +#include "gz/transport/NodeShared.hh" + +#include +#include + +#include "gtest/gtest.h" +#include "test_config.hh" +#include "test_utils.hh" + +using namespace gz; + +static std::string partition; // NOLINT(*) + +// Private discovery port so that this test observes only its own traffic. +static const int kTestDiscPort = 11417; + +// Wire version of the discovery protocol. It must match +// Discovery::wireVersion or the crafted messages are discarded. +static const uint32_t kWireVersion = 10; + +////////////////////////////////////////////////// +/// \brief Helper joining the discovery multicast group with a raw UDP +/// socket. It can passively count discovery messages by type and send +/// crafted discovery messages, emulating a remote process. +class DiscoveryWire +{ + public: DiscoveryWire() + { + this->sock = socket(AF_INET, SOCK_DGRAM, 0); + EXPECT_GE(this->sock, 0); + + int reuse = 1; + setsockopt(this->sock, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse)); + setsockopt(this->sock, SOL_SOCKET, SO_REUSEPORT, &reuse, sizeof(reuse)); + + sockaddr_in local{}; + local.sin_family = AF_INET; + local.sin_port = htons(kTestDiscPort); + local.sin_addr.s_addr = htonl(INADDR_ANY); + EXPECT_EQ(bind(this->sock, + reinterpret_cast(&local), sizeof(local)), 0); + + ip_mreq mreq{}; + mreq.imr_multiaddr.s_addr = inet_addr("239.255.0.7"); + mreq.imr_interface.s_addr = inet_addr("127.0.0.1"); + EXPECT_EQ(setsockopt(this->sock, IPPROTO_IP, IP_ADD_MEMBERSHIP, + &mreq, sizeof(mreq)), 0); + + in_addr iface{}; + iface.s_addr = inet_addr("127.0.0.1"); + setsockopt(this->sock, IPPROTO_IP, IP_MULTICAST_IF, + &iface, sizeof(iface)); + int loop = 1; + setsockopt(this->sock, IPPROTO_IP, IP_MULTICAST_LOOP, + &loop, sizeof(loop)); + + fcntl(this->sock, F_SETFL, O_NONBLOCK); + + this->dst = {}; + this->dst.sin_family = AF_INET; + this->dst.sin_port = htons(kTestDiscPort); + this->dst.sin_addr.s_addr = inet_addr("239.255.0.7"); + } + + public: ~DiscoveryWire() + { + close(this->sock); + } + + /// \brief Send a discovery message to the multicast group, framed with + /// the 2 byte length prefix used by the discovery wire format. + public: void Send(const gz::msgs::Discovery &_msg) + { + const uint16_t msgSize = static_cast(_msg.ByteSizeLong()); + std::vector buffer(sizeof(msgSize) + msgSize); + memcpy(buffer.data(), &msgSize, sizeof(msgSize)); + ASSERT_TRUE(_msg.SerializeToArray( + buffer.data() + sizeof(msgSize), msgSize)); + sendto(this->sock, buffer.data(), buffer.size(), 0, + reinterpret_cast(&this->dst), sizeof(this->dst)); + } + + /// \brief Drain the pending datagrams, counting the parsed discovery + /// messages by type. + /// \param[in] _windowMs Extra time to keep draining (ms.). + /// \return Map of message type to number of messages observed. + public: std::map CountTypes(const int _windowMs) + { + std::map counts; + const auto deadline = std::chrono::steady_clock::now() + + std::chrono::milliseconds(_windowMs); + + do + { + char buffer[65536]; + ssize_t received; + while ((received = recvfrom(this->sock, buffer, sizeof(buffer), 0, + nullptr, nullptr)) > 0) + { + uint16_t msgSize; + if (received < static_cast(sizeof(msgSize))) + continue; + memcpy(&msgSize, buffer, sizeof(msgSize)); + + gz::msgs::Discovery msg; + if (msg.ParseFromArray(buffer + sizeof(msgSize), msgSize)) + ++counts[msg.type()]; + } + std::this_thread::sleep_for(std::chrono::milliseconds(5)); + } while (std::chrono::steady_clock::now() < deadline); + + return counts; + } + + /// \brief Drain and discard all the pending datagrams. + public: void Drain() + { + this->CountTypes(0); + } + + private: int sock = -1; + private: sockaddr_in dst; +}; + +////////////////////////////////////////////////// +/// \brief A known process that never answers a SUBSCRIBERS_REQ makes +/// TopicList() wait for the timeout, and the call recovers once the silent +/// process expires. +TEST(topicListTraffic, TimeoutWithSilentPeer) +{ + transport::Node node; + + // Initialize discovery. + std::vector topics; + node.TopicList(topics); + + // Emulate a remote process that heartbeats but never answers. + DiscoveryWire wire; + gz::msgs::Discovery heartbeat; + heartbeat.set_version(kWireVersion); + heartbeat.set_type(gz::msgs::Discovery::HEARTBEAT); + heartbeat.set_process_uuid("topicListTraffic-silent-peer"); + for (int i = 0; i < 3; ++i) + { + wire.Send(heartbeat); + std::this_thread::sleep_for(std::chrono::milliseconds(100)); + } + + // The silent peer is known but never reports its subscribers: the call + // returns when the timeout expires. + auto start = std::chrono::steady_clock::now(); + node.TopicList(topics); + auto elapsedMs = std::chrono::duration_cast( + std::chrono::steady_clock::now() - start).count(); + EXPECT_GE(elapsedMs, 90); + EXPECT_LT(elapsedMs, 250); + + // After the silence interval the peer expires and the calls are fast + // again. + std::this_thread::sleep_for(std::chrono::milliseconds(3500)); + start = std::chrono::steady_clock::now(); + node.TopicList(topics); + elapsedMs = std::chrono::duration_cast( + std::chrono::steady_clock::now() - start).count(); + EXPECT_LT(elapsedMs, 50); +} + +////////////////////////////////////////////////// +/// \brief The subscribers traffic is proportional to the demand: nothing is +/// requested while idle and a single TopicList() call produces one request +/// and one bounded reply burst. +TEST(topicListTraffic, TrafficBounds) +{ + transport::Node node; + + // A remote process with one subscription. + auto pi = testing::SubprocessJoinWrapper( + {test_executables::kSubscriberOnly, partition, "/subscriber_only", "15"}); + + // Let the remote process start and its discovery settle. + std::this_thread::sleep_for(std::chrono::seconds(3)); + + DiscoveryWire wire; + wire.Drain(); + + // While idle, no subscribers traffic flows. + auto counts = wire.CountTypes(2000); + EXPECT_EQ(counts[gz::msgs::Discovery::SUBSCRIBERS_REQ], 0); + EXPECT_EQ(counts[gz::msgs::Discovery::SUBSCRIBERS_REP], 0); + + // A single call produces one request and one reply from the remote + // process, which has a single subscription. + std::vector topics; + node.TopicList(topics); + std::this_thread::sleep_for(std::chrono::milliseconds(300)); + + counts = wire.CountTypes(0); + EXPECT_EQ(counts[gz::msgs::Discovery::SUBSCRIBERS_REQ], 1); + EXPECT_EQ(counts[gz::msgs::Discovery::SUBSCRIBERS_REP], 1); + + EXPECT_TRUE(std::find(topics.begin(), topics.end(), "/subscriber_only") != + topics.end()); +} + +////////////////////////////////////////////////// +int main(int argc, char **argv) +{ + // Get a random partition name. + partition = testing::getRandomNumber(); + + // Set the partition name for this process. + gz::utils::setenv("GZ_PARTITION", partition); + + // Use a private discovery port so that this test observes only its own + // traffic. + gz::utils::setenv("GZ_DISCOVERY_MSG_PORT", std::to_string(kTestDiscPort)); + gz::utils::setenv("GZ_DISCOVERY_SRV_PORT", + std::to_string(kTestDiscPort + 1)); + + ::testing::InitGoogleTest(&argc, argv); + return RUN_ALL_TESTS(); +} From 0a952c40f5327014faee9939af90b9ca4578e918 Mon Sep 17 00:00:00 2001 From: Carlos Aguero Date: Fri, 31 Jul 2026 17:18:54 +0200 Subject: [PATCH 10/13] Declare the subscribers reply timeout constexpr to fix non optimized builds Assisted-by: Claude Fable 5 Signed-off-by: Carlos Aguero --- src/Discovery.hh | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/Discovery.hh b/src/Discovery.hh index 03578730a..d9ace3800 100644 --- a/src/Discovery.hh +++ b/src/Discovery.hh @@ -1790,7 +1790,7 @@ namespace gz /// \brief Default maximum time waiting for the subscriber snapshots /// in TopicList() (ms.). The wait normally finishes much earlier, /// when every known process has answered. - private: static const unsigned int kDefSubscribersRepTimeout = 100; + private: static constexpr unsigned int kDefSubscribersRepTimeout = 100; /// \brief Default silence interval value (ms.). /// \sa MaxSilenceInterval. From 9eea0d9ebecbacd1d9ee1b88e5dd7903f0f9ba8c Mon Sep 17 00:00:00 2001 From: Carlos Aguero Date: Fri, 31 Jul 2026 17:34:17 +0200 Subject: [PATCH 11/13] Add legacy peer and BYE purge coverage to the traffic test Assisted-by: Claude Fable 5 Signed-off-by: Carlos Aguero --- test/integration/topicListTraffic.cc | 53 ++++++++++++++++++++++++++++ 1 file changed, 53 insertions(+) diff --git a/test/integration/topicListTraffic.cc b/test/integration/topicListTraffic.cc index 12b41769d..1618070da 100644 --- a/test/integration/topicListTraffic.cc +++ b/test/integration/topicListTraffic.cc @@ -254,3 +254,56 @@ int main(int argc, char **argv) ::testing::InitGoogleTest(&argc, argv); return RUN_ALL_TESTS(); } + +////////////////////////////////////////////////// +/// \brief A process running an older version reports subscribers without +/// the snapshot metadata: its information is merged into the results, and +/// a BYE message purges the process and its subscriptions. +TEST(topicListTraffic, LegacyPeerAndBye) +{ + transport::Node node; + std::vector topics; + node.TopicList(topics); + + DiscoveryWire wire; + + // The legacy peer heartbeats and reports one subscription without the + // snapshot metadata, like versions predating it. + gz::msgs::Discovery heartbeat; + heartbeat.set_version(kWireVersion); + heartbeat.set_type(gz::msgs::Discovery::HEARTBEAT); + heartbeat.set_process_uuid("topicListTraffic-legacy-peer"); + wire.Send(heartbeat); + + gz::msgs::Discovery rep; + rep.set_version(kWireVersion); + rep.set_type(gz::msgs::Discovery::SUBSCRIBERS_REP); + rep.set_process_uuid("topicListTraffic-legacy-peer"); + auto *pub = rep.mutable_pub(); + pub->set_topic("@/" + partition + "@/legacy_sub"); + pub->set_process_uuid("topicListTraffic-legacy-peer"); + pub->set_node_uuid("legacy-node"); + wire.Send(rep); + + std::this_thread::sleep_for(std::chrono::milliseconds(200)); + + // The topic is reported. The call pays the timeout because a legacy + // peer never reports a snapshot completion. + topics.clear(); + node.TopicList(topics); + EXPECT_TRUE(std::find(topics.begin(), topics.end(), "/legacy_sub") != + topics.end()); + + // BYE purges the process and its subscriptions. + gz::msgs::Discovery bye; + bye.set_version(kWireVersion); + bye.set_type(gz::msgs::Discovery::BYE); + bye.set_process_uuid("topicListTraffic-legacy-peer"); + wire.Send(bye); + std::this_thread::sleep_for(std::chrono::milliseconds(200)); + + topics.clear(); + node.TopicList(topics); + EXPECT_TRUE(std::find(topics.begin(), topics.end(), "/legacy_sub") == + topics.end()); +} From abced66206daa3702cfdf530e8ea88c765aa86b6 Mon Sep 17 00:00:00 2001 From: Carlos Aguero Date: Fri, 31 Jul 2026 18:27:49 +0200 Subject: [PATCH 12/13] Drop the raw socket traffic test to avoid coupling tests to the wire internals Assisted-by: Claude Fable 5 Signed-off-by: Carlos Aguero --- test/BUILD.bazel | 2 - test/integration/CMakeLists.txt | 5 - test/integration/topicListTraffic.cc | 309 --------------------------- 3 files changed, 316 deletions(-) delete mode 100644 test/integration/topicListTraffic.cc diff --git a/test/BUILD.bazel b/test/BUILD.bazel index dc8702d1b..d4734f501 100644 --- a/test/BUILD.bazel +++ b/test/BUILD.bazel @@ -68,7 +68,6 @@ integration_test_sources = [ "statistics.cc", "topicListFirstCall.cc", "topicListStartupSync.cc", - "topicListTraffic.cc", "twoProcsPubSub.cc", "twoProcsPubSubStats.cc", "twoProcsSrvCall.cc", @@ -86,7 +85,6 @@ integration_test_sources = [ flaky_test_srcs = [ "topicListFirstCall.cc", "topicListStartupSync.cc", - "topicListTraffic.cc", "twoProcsPubSub.cc", "twoProcsSrvCall.cc", "twoProcsSrvCallWithoutInput.cc", diff --git a/test/integration/CMakeLists.txt b/test/integration/CMakeLists.txt index c65990eca..035d83265 100644 --- a/test/integration/CMakeLists.txt +++ b/test/integration/CMakeLists.txt @@ -19,11 +19,6 @@ set(tests twoProcsSrvCallWithoutOutputStress.cc ) -# The traffic test crafts raw discovery datagrams using POSIX sockets. -if (UNIX AND NOT APPLE) - list(APPEND tests topicListTraffic.cc) -endif() - # Test symbols having the right name on linux only. if (UNIX AND NOT APPLE) configure_file(all_symbols_have_version.bash.in diff --git a/test/integration/topicListTraffic.cc b/test/integration/topicListTraffic.cc deleted file mode 100644 index 1618070da..000000000 --- a/test/integration/topicListTraffic.cc +++ /dev/null @@ -1,309 +0,0 @@ -/* - * Copyright (C) 2026 Open Source Robotics Foundation - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - * -*/ - -#include - -#include -#include -#include -#include -#include - -#include -#include -#include -#include -#include -#include -#include - -#include "gz/transport/Node.hh" -#include "gz/transport/NodeShared.hh" - -#include -#include - -#include "gtest/gtest.h" -#include "test_config.hh" -#include "test_utils.hh" - -using namespace gz; - -static std::string partition; // NOLINT(*) - -// Private discovery port so that this test observes only its own traffic. -static const int kTestDiscPort = 11417; - -// Wire version of the discovery protocol. It must match -// Discovery::wireVersion or the crafted messages are discarded. -static const uint32_t kWireVersion = 10; - -////////////////////////////////////////////////// -/// \brief Helper joining the discovery multicast group with a raw UDP -/// socket. It can passively count discovery messages by type and send -/// crafted discovery messages, emulating a remote process. -class DiscoveryWire -{ - public: DiscoveryWire() - { - this->sock = socket(AF_INET, SOCK_DGRAM, 0); - EXPECT_GE(this->sock, 0); - - int reuse = 1; - setsockopt(this->sock, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse)); - setsockopt(this->sock, SOL_SOCKET, SO_REUSEPORT, &reuse, sizeof(reuse)); - - sockaddr_in local{}; - local.sin_family = AF_INET; - local.sin_port = htons(kTestDiscPort); - local.sin_addr.s_addr = htonl(INADDR_ANY); - EXPECT_EQ(bind(this->sock, - reinterpret_cast(&local), sizeof(local)), 0); - - ip_mreq mreq{}; - mreq.imr_multiaddr.s_addr = inet_addr("239.255.0.7"); - mreq.imr_interface.s_addr = inet_addr("127.0.0.1"); - EXPECT_EQ(setsockopt(this->sock, IPPROTO_IP, IP_ADD_MEMBERSHIP, - &mreq, sizeof(mreq)), 0); - - in_addr iface{}; - iface.s_addr = inet_addr("127.0.0.1"); - setsockopt(this->sock, IPPROTO_IP, IP_MULTICAST_IF, - &iface, sizeof(iface)); - int loop = 1; - setsockopt(this->sock, IPPROTO_IP, IP_MULTICAST_LOOP, - &loop, sizeof(loop)); - - fcntl(this->sock, F_SETFL, O_NONBLOCK); - - this->dst = {}; - this->dst.sin_family = AF_INET; - this->dst.sin_port = htons(kTestDiscPort); - this->dst.sin_addr.s_addr = inet_addr("239.255.0.7"); - } - - public: ~DiscoveryWire() - { - close(this->sock); - } - - /// \brief Send a discovery message to the multicast group, framed with - /// the 2 byte length prefix used by the discovery wire format. - public: void Send(const gz::msgs::Discovery &_msg) - { - const uint16_t msgSize = static_cast(_msg.ByteSizeLong()); - std::vector buffer(sizeof(msgSize) + msgSize); - memcpy(buffer.data(), &msgSize, sizeof(msgSize)); - ASSERT_TRUE(_msg.SerializeToArray( - buffer.data() + sizeof(msgSize), msgSize)); - sendto(this->sock, buffer.data(), buffer.size(), 0, - reinterpret_cast(&this->dst), sizeof(this->dst)); - } - - /// \brief Drain the pending datagrams, counting the parsed discovery - /// messages by type. - /// \param[in] _windowMs Extra time to keep draining (ms.). - /// \return Map of message type to number of messages observed. - public: std::map CountTypes(const int _windowMs) - { - std::map counts; - const auto deadline = std::chrono::steady_clock::now() + - std::chrono::milliseconds(_windowMs); - - do - { - char buffer[65536]; - ssize_t received; - while ((received = recvfrom(this->sock, buffer, sizeof(buffer), 0, - nullptr, nullptr)) > 0) - { - uint16_t msgSize; - if (received < static_cast(sizeof(msgSize))) - continue; - memcpy(&msgSize, buffer, sizeof(msgSize)); - - gz::msgs::Discovery msg; - if (msg.ParseFromArray(buffer + sizeof(msgSize), msgSize)) - ++counts[msg.type()]; - } - std::this_thread::sleep_for(std::chrono::milliseconds(5)); - } while (std::chrono::steady_clock::now() < deadline); - - return counts; - } - - /// \brief Drain and discard all the pending datagrams. - public: void Drain() - { - this->CountTypes(0); - } - - private: int sock = -1; - private: sockaddr_in dst; -}; - -////////////////////////////////////////////////// -/// \brief A known process that never answers a SUBSCRIBERS_REQ makes -/// TopicList() wait for the timeout, and the call recovers once the silent -/// process expires. -TEST(topicListTraffic, TimeoutWithSilentPeer) -{ - transport::Node node; - - // Initialize discovery. - std::vector topics; - node.TopicList(topics); - - // Emulate a remote process that heartbeats but never answers. - DiscoveryWire wire; - gz::msgs::Discovery heartbeat; - heartbeat.set_version(kWireVersion); - heartbeat.set_type(gz::msgs::Discovery::HEARTBEAT); - heartbeat.set_process_uuid("topicListTraffic-silent-peer"); - for (int i = 0; i < 3; ++i) - { - wire.Send(heartbeat); - std::this_thread::sleep_for(std::chrono::milliseconds(100)); - } - - // The silent peer is known but never reports its subscribers: the call - // returns when the timeout expires. - auto start = std::chrono::steady_clock::now(); - node.TopicList(topics); - auto elapsedMs = std::chrono::duration_cast( - std::chrono::steady_clock::now() - start).count(); - EXPECT_GE(elapsedMs, 90); - EXPECT_LT(elapsedMs, 250); - - // After the silence interval the peer expires and the calls are fast - // again. - std::this_thread::sleep_for(std::chrono::milliseconds(3500)); - start = std::chrono::steady_clock::now(); - node.TopicList(topics); - elapsedMs = std::chrono::duration_cast( - std::chrono::steady_clock::now() - start).count(); - EXPECT_LT(elapsedMs, 50); -} - -////////////////////////////////////////////////// -/// \brief The subscribers traffic is proportional to the demand: nothing is -/// requested while idle and a single TopicList() call produces one request -/// and one bounded reply burst. -TEST(topicListTraffic, TrafficBounds) -{ - transport::Node node; - - // A remote process with one subscription. - auto pi = testing::SubprocessJoinWrapper( - {test_executables::kSubscriberOnly, partition, "/subscriber_only", "15"}); - - // Let the remote process start and its discovery settle. - std::this_thread::sleep_for(std::chrono::seconds(3)); - - DiscoveryWire wire; - wire.Drain(); - - // While idle, no subscribers traffic flows. - auto counts = wire.CountTypes(2000); - EXPECT_EQ(counts[gz::msgs::Discovery::SUBSCRIBERS_REQ], 0); - EXPECT_EQ(counts[gz::msgs::Discovery::SUBSCRIBERS_REP], 0); - - // A single call produces one request and one reply from the remote - // process, which has a single subscription. - std::vector topics; - node.TopicList(topics); - std::this_thread::sleep_for(std::chrono::milliseconds(300)); - - counts = wire.CountTypes(0); - EXPECT_EQ(counts[gz::msgs::Discovery::SUBSCRIBERS_REQ], 1); - EXPECT_EQ(counts[gz::msgs::Discovery::SUBSCRIBERS_REP], 1); - - EXPECT_TRUE(std::find(topics.begin(), topics.end(), "/subscriber_only") != - topics.end()); -} - -////////////////////////////////////////////////// -int main(int argc, char **argv) -{ - // Get a random partition name. - partition = testing::getRandomNumber(); - - // Set the partition name for this process. - gz::utils::setenv("GZ_PARTITION", partition); - - // Use a private discovery port so that this test observes only its own - // traffic. - gz::utils::setenv("GZ_DISCOVERY_MSG_PORT", std::to_string(kTestDiscPort)); - gz::utils::setenv("GZ_DISCOVERY_SRV_PORT", - std::to_string(kTestDiscPort + 1)); - - ::testing::InitGoogleTest(&argc, argv); - return RUN_ALL_TESTS(); -} - -////////////////////////////////////////////////// -/// \brief A process running an older version reports subscribers without -/// the snapshot metadata: its information is merged into the results, and -/// a BYE message purges the process and its subscriptions. -TEST(topicListTraffic, LegacyPeerAndBye) -{ - transport::Node node; - std::vector topics; - node.TopicList(topics); - - DiscoveryWire wire; - - // The legacy peer heartbeats and reports one subscription without the - // snapshot metadata, like versions predating it. - gz::msgs::Discovery heartbeat; - heartbeat.set_version(kWireVersion); - heartbeat.set_type(gz::msgs::Discovery::HEARTBEAT); - heartbeat.set_process_uuid("topicListTraffic-legacy-peer"); - wire.Send(heartbeat); - - gz::msgs::Discovery rep; - rep.set_version(kWireVersion); - rep.set_type(gz::msgs::Discovery::SUBSCRIBERS_REP); - rep.set_process_uuid("topicListTraffic-legacy-peer"); - auto *pub = rep.mutable_pub(); - pub->set_topic("@/" + partition + "@/legacy_sub"); - pub->set_process_uuid("topicListTraffic-legacy-peer"); - pub->set_node_uuid("legacy-node"); - wire.Send(rep); - - std::this_thread::sleep_for(std::chrono::milliseconds(200)); - - // The topic is reported. The call pays the timeout because a legacy - // peer never reports a snapshot completion. - topics.clear(); - node.TopicList(topics); - EXPECT_TRUE(std::find(topics.begin(), topics.end(), "/legacy_sub") != - topics.end()); - - // BYE purges the process and its subscriptions. - gz::msgs::Discovery bye; - bye.set_version(kWireVersion); - bye.set_type(gz::msgs::Discovery::BYE); - bye.set_process_uuid("topicListTraffic-legacy-peer"); - wire.Send(bye); - std::this_thread::sleep_for(std::chrono::milliseconds(200)); - - topics.clear(); - node.TopicList(topics); - EXPECT_TRUE(std::find(topics.begin(), topics.end(), "/legacy_sub") == - topics.end()); -} From 02d5424457975246383d80ed67eacddf55ad56ba Mon Sep 17 00:00:00 2001 From: Carlos Aguero Date: Fri, 31 Jul 2026 18:27:49 +0200 Subject: [PATCH 13/13] Synchronize with the subscriber aux through a ready file instead of fixed sleeps Assisted-by: Claude Fable 5 Signed-off-by: Carlos Aguero --- .../test_executables/subscriberOnly_aux.cc | 13 +++++++++--- test/integration/topicListFirstCall.cc | 20 ++++++++++++++----- test/integration/topicListStartupSync.cc | 14 ++++++++++--- 3 files changed, 36 insertions(+), 11 deletions(-) diff --git a/test/integration/test_executables/subscriberOnly_aux.cc b/test/integration/test_executables/subscriberOnly_aux.cc index df9a4834b..22a98629a 100644 --- a/test/integration/test_executables/subscriberOnly_aux.cc +++ b/test/integration/test_executables/subscriberOnly_aux.cc @@ -18,6 +18,7 @@ #include #include +#include #include #include #include @@ -41,14 +42,16 @@ void cb(const msgs::Vector3d &) ////////////////////////////////////////////////// /// \brief Usage: subscriberOnly_aux [topic] [lifetimeSec] -/// [unsubscribeAfterSec]. +/// [unsubscribeAfterSec] [readyFile]. /// Subscribe to a topic without any publisher and stay alive for /// lifetimeSec, giving the test process time to discover this subscription. /// If unsubscribeAfterSec is positive, unsubscribe after that time while -/// keeping the process alive until lifetimeSec. +/// keeping the process alive until lifetimeSec. If readyFile is provided, +/// create that file right after subscribing, so that the test can +/// synchronize without waiting a fixed time. int main(int argc, char **argv) { - if (argc < 2 || argc > 5) + if (argc < 2 || argc > 6) { std::cerr << "Partition name has not be passed as argument" << std::endl; return -1; @@ -60,10 +63,14 @@ int main(int argc, char **argv) const std::string topic = argc > 2 ? argv[2] : g_topic; const int lifetimeSec = argc > 3 ? std::stoi(argv[3]) : 10; const int unsubscribeAfterSec = argc > 4 ? std::stoi(argv[4]) : 0; + const std::string readyFile = argc > 5 ? argv[5] : ""; transport::Node node; node.Subscribe(topic, cb); + if (!readyFile.empty()) + std::ofstream(readyFile) << "ready"; + int elapsedSec = 0; if (unsubscribeAfterSec > 0) { diff --git a/test/integration/topicListFirstCall.cc b/test/integration/topicListFirstCall.cc index 272b098d2..e936fd3cf 100644 --- a/test/integration/topicListFirstCall.cc +++ b/test/integration/topicListFirstCall.cc @@ -17,11 +17,13 @@ #include #include +#include #include #include #include #include "gz/transport/Node.hh" +#include "gz/transport/WaitHelpers.hh" #include #include @@ -43,14 +45,22 @@ TEST(topicListFirstCall, SubscriberInFirstCall) { transport::Node node; - // Let discovery initialize without calling TopicList(). + // Let discovery initialize without calling TopicList(). This wait cannot + // poll: any query would exercise the API under test. The value covers + // the two heartbeat initialization phase with margin. std::this_thread::sleep_for(std::chrono::seconds(4)); + const std::string readyFile = "subscriberOnly_" + partition + ".ready"; + std::filesystem::remove(readyFile); auto pi = testing::SubprocessJoinWrapper( - {test_executables::kSubscriberOnly, partition}); - - // Give the remote process time to start and subscribe. - std::this_thread::sleep_for(std::chrono::seconds(2)); + {test_executables::kSubscriberOnly, partition, "/subscriber_only", "12", + "0", readyFile}); + + // Wait until the remote process is subscribed. + ASSERT_TRUE(transport::waitUntil([&readyFile] + { + return std::filesystem::exists(readyFile); + })); // The first call should collect the remote subscriber replies. auto start = std::chrono::steady_clock::now(); diff --git a/test/integration/topicListStartupSync.cc b/test/integration/topicListStartupSync.cc index 8f3d49009..1112e6394 100644 --- a/test/integration/topicListStartupSync.cc +++ b/test/integration/topicListStartupSync.cc @@ -17,11 +17,12 @@ #include #include +#include #include -#include #include #include "gz/transport/Node.hh" +#include "gz/transport/WaitHelpers.hh" #include #include @@ -43,10 +44,17 @@ static std::string partition; // NOLINT(*) TEST(topicListStartupSync, PreexistingSubscriberInFirstCall) { // The remote subscriber exists before this process starts its discovery. + const std::string readyFile = "subscriberOnly_" + partition + ".ready"; + std::filesystem::remove(readyFile); auto pi = testing::SubprocessJoinWrapper( - {test_executables::kSubscriberOnly, partition, "/subscriber_only", "12"}); + {test_executables::kSubscriberOnly, partition, "/subscriber_only", "12", + "0", readyFile}); - std::this_thread::sleep_for(std::chrono::seconds(2)); + // Wait until the remote process is subscribed. + ASSERT_TRUE(transport::waitUntil([&readyFile] + { + return std::filesystem::exists(readyFile); + })); // The first transport node starts the discovery of this process, which // requests the current subscribers.