Skip to content

Announce subscriptions and report subscriber snapshots in discovery - #916

Draft
caguero wants to merge 13 commits into
mainfrom
caguero_914_subscribe_announce
Draft

Announce subscriptions and report subscriber snapshots in discovery#916
caguero wants to merge 13 commits into
mainfrom
caguero_914_subscribe_announce

Conversation

@caguero

@caguero caguero commented Jul 31, 2026 •
edited
Loading

Copy link
Copy Markdown
Collaborator

🎉 New feature

Follow up of #915. Fixes #914.
Requires gazebosim/gz-msgs#615

Summary

Completes the subscriber discovery so that TopicList() returns complete information on any call, including the first one:

  • SUBSCRIBE messages announce the subscription (node UUID) and the receivers record it.
  • Discovery requests the current subscribers once at startup.
  • SUBSCRIBERS_REP bursts are tagged as snapshots (count, generation). TopicList() waits until every known process reports a complete snapshot, with a 100 ms timeout.

Slightly behavior change: TopicList() now blocks a few ms (up to 100 ms when a process does not answer).

Backport Policy

  • This is safe to backport to the following versions:
    • Jetty
    • Ionic
    • Harmonic
    • Fortress
  • This should not be backported
  • I am not sure
  • Other (fill in yourself)

Checklist

  • Signed all commits for DCO
  • Added a screen capture or video to the PR description that demonstrates the fix (as needed)
  • Added tests
  • Updated documentation (as needed)
  • Updated migration guide (as needed)
  • Consider updating Python bindings (if the library has them)
  • codecheck passed (See contributing)
  • All tests passed (See test coverage)
  • Updated Bazel files (if adding new files). Created an issue otherwise.
  • While waiting for a review on your PR, please help review another open pull request to support the maintainers
  • Was GenAI used to generate this PR? If so, make sure to add "Assisted-by" to your commits. (See this policy for more info.)

Assisted-by: Claude Fable 5

Note to maintainers: Remember to use Squash-Merge and edit the commit message to match the pull request summary while retaining Signed-off-by and Generated-by messages.

Backports: If this is a backport, please use Rebase and Merge instead.

caguero added 11 commits July 30, 2026 18:50
@caguero
Maintain the remote subscribers cache instead of clearing it on every…
b74c29e 
… TopicList() call

Assisted-by: Claude Fable 5
Signed-off-by: Carlos Aguero <caguero@honurobotics.com>
@caguero
Include topics subscribed within this process in Node::TopicList()
b2c43a0 
Assisted-by: Claude Fable 5
Signed-off-by: Carlos Aguero <caguero@honurobotics.com>
@caguero
Add regression tests for subscriber only topics in TopicList()
a17b00c 
Assisted-by: Claude Fable 5
Signed-off-by: Carlos Aguero <caguero@honurobotics.com>
@caguero
Include subscribers from this process in Node::TopicInfo()
87cd014 
Assisted-by: Claude Fable 5
Signed-off-by: Carlos Aguero <caguero@honurobotics.com>
@caguero
Add tests for unsubscribe staleness and TopicInfo() subscribers
ebad3e0 
Assisted-by: Claude Fable 5
Signed-off-by: Carlos Aguero <caguero@honurobotics.com>
@caguero
[bazel] Add subscriberOnly_aux to the test build
277c159 
Assisted-by: Claude Fable 5
Signed-off-by: Carlos Aguero <caguero@honurobotics.com>
@caguero
Announce subscriptions in SUBSCRIBE messages and sync subscribers at …
deb2bd5 
…discovery startup

Assisted-by: Claude Fable 5
Signed-off-by: Carlos Aguero <caguero@honurobotics.com>
@caguero
Report subscribers with generation tagged snapshots and wait for thei…
073c007 
…r completion in TopicList()

Assisted-by: Claude Fable 5
Signed-off-by: Carlos Aguero <caguero@honurobotics.com>
@caguero
Add startup sync, silent peer timeout, and traffic bound tests for th…
13b41bf 
…e subscribers protocol

Assisted-by: Claude Fable 5
Signed-off-by: Carlos Aguero <caguero@honurobotics.com>
@caguero
Declare the subscribers reply timeout constexpr to fix non optimized …
0a952c4 
…builds

Assisted-by: Claude Fable 5
Signed-off-by: Carlos Aguero <caguero@honurobotics.com>
@caguero
Add legacy peer and BYE purge coverage to the traffic test
9eea0d9 
Assisted-by: Claude Fable 5
Signed-off-by: Carlos Aguero <caguero@honurobotics.com>
@caguero caguero added the needs upstream release Blocked by a release of an upstream library label Jul 31, 2026
@caguero caguero mentioned this pull request Jul 31, 2026
Open
caguero added 2 commits July 31, 2026 18:27
@caguero
Drop the raw socket traffic test to avoid coupling tests to the wire …
abced66 
…internals

Assisted-by: Claude Fable 5
Signed-off-by: Carlos Aguero <caguero@honurobotics.com>
@caguero
Synchronize with the subscriber aux through a ready file instead of f…
02d5424 
…ixed sleeps

Assisted-by: Claude Fable 5
Signed-off-by: Carlos Aguero <caguero@honurobotics.com>
@caguero
caguero force-pushed the caguero_914_subscribe_announce branch from fae3f53 to 02d5424 Compare July 31, 2026 16:27
C88-YQ
C88-YQ reviewed Aug 16, 2026
View reviewed changes

@C88-YQ C88-YQ left a comment •
edited
Loading

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Otherwise, LGTM!
I also did some behavior tests, and the results were consistent with #915 (comment)
Thanks for the fix, and sorry for the late reply.

Comment thread src/Discovery.hh
Comment on lines +1400 to +1414
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;
}

@C88-YQ C88-YQ Aug 16, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think we may need a way to identify each snapshot response burst.

Because, if the subscriber generation does not change, and TopicList() is called concurrently, replies from different SUBSCRIBERS_REQ calls can be mixed.

Also, if one TopicList() call times out, late replies from that older SUBSCRIBERS_REQ can arrive during a later TopicList() call.

In both cases, SUBSCRIBERS_REP only carries generation and count, so the receiver cannot tell which request a reply belongs to. Since completion is decided by received >= expected, mixed replies may make the receiver mark the snapshot as complete even when it did not receive one full response burst.

Comment thread src/NodeShared.cc
std::lock_guard<std::recursive_mutex> 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;

@C88-YQ C88-YQ Aug 16, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should subscriptionsGeneration be updated for every subscription handler change, or only when the topic/node membership changes?
Currently, it is incremented even when topicsSubscribed already contains the topic.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Reviewers

1 more reviewer
@C88-YQ C88-YQ C88-YQ left review comments
Reviewers whose approvals may not affect merge requirements

At least 1 approving review is required to merge this pull request.

Assignees

No one assigned

Labels

needs upstream release Blocked by a release of an upstream library 🎡 rotary

Projects

Core development
Status: Inbox

Milestone

No milestone

Development

Successfully merging this pull request may close these issues.

TopicList() may return incomplete topic information

3 participants

@caguero @C88-YQ @azeey