Fix unbounded memory growth with slow local subscribers

[caguero]

🦟 Bug fix

Fixes #926

Summary

The queue delivering messages to local (intraprocess) subscribers was unbounded, so callbacks slower than the publication rate grew it until out of memory. This patch bounds it per topic: when a topic already has GZ_TRANSPORT_LOCAL_HWM messages queued (default 1000), publishing a new one drops that topic's oldest queued message and warns once per overload episode. Setting the variable to 0 restores the old unbounded behavior. Mirrors
GZ_TRANSPORT_RCVHWM/SNDHWM and works on both backends.

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: 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.

[luca-della-vedova]

I'm not familiar with the history of gz-transport but is there a reason we should still commit to this strategy of using a global parameter set through environment variables for queue sizes, now with a new parameter for local queues that is different from the publish / subscribe queue size?

IMHO ROS does it in a sensible way that users are used to, with a publisher queue / subscriber queue parameter for each publisher and subscriber, and no distinction between local and remote pub/sub queue sizes.
Apart from converging towards the way ROS does things, a global parameter is not a great approach imho. Different topics might have very different requirements for queue sizes. For example users might want a larger queue size for data they don't want to drop, or a smaller queue size (as small as 1) for data that they will just parse once in a while and they only want the latest sample.

[caguero]

I'm not familiar with the history of gz-transport but is there a reason we should still commit to this strategy of using a global parameter set through environment variables for queue sizes, now with a new parameter for local queues that is different from the publish / subscribe queue size?

IMHO ROS does it in a sensible way that users are used to, with a publisher queue / subscriber queue parameter for each publisher and subscriber, and no distinction between local and remote pub/sub queue sizes. Apart from converging towards the way ROS does things, a global parameter is not a great approach imho. Different topics might have very different requirements for queue sizes. For example users might want a larger queue size for data they don't want to drop, or a smaller queue size (as small as 1) for data that they will just parse once in a while and they only want the latest sample.

Thanks for the suggestion! I agree that this is the way to go in the medium/long-term but in the short term we're trying to address the potential out of memory crash. The global local env variable decision is there because it's the logical continuation of how it started in ZeroMQ exposing global env vars as well. Creating queues for publishers/subscribers at the topic level with a programmatic API will need more architecture and not sure if it'll be all backportable. We'll need to think about it and prototype. I'll keep working on it in the background in a separate PR.