Integrate Zenoh 1.8.0 - Part 2/2

[caguero]

Summary

This patch completes the Zenoh 1.8 integration started in #867:

1. Cold start race in service requests. With interest driven routing (Zenoh > 1.6), the first Node::Request to a freshly spawned responder can time out silently. Requests now go through a per process cached zenoh::Querier whose interest declaration stays alive, so an in flight request reaches a queryable that appears late (INTEGRATION_twoProcsSrvCallLateResponder).

2. Racy shutdown at exit. A new explicit NodeShared::Shutdown() closes the session in a deterministic order, removing the 1.8 close race by construction.

3. No more teardown leaks. With the deterministic shutdown in place, cached Queriers and Publisher entities are undeclared normally instead of leaked, so phantom publishers disappear from discovery immediately (INTEGRATION_zenohPublisherLeak).

Also, CreateZenohGet no longer blocks while holding NodeShared::mutex (replies arrive through a weak_ptr, mirroring the ZeroMQ flow), and the minimum Zenoh version is relaxed to 1.7.2 after verifying the API surface is identical (drop 7176ddf if we prefer a 1.8.0 floor).

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 "Generated-by" to your commits. (See this policy for more info.)

Generated-by: Claude Opus 4.7

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 think this won't work, I tried to run this and traced a segfault to this line.
It seems what it tries to do is allocate a channel with a capacity of size_t-1 which is incredibly large, the allocation fails and the code segfaults.
This is wildly out of my comfort zone so I had Gemini suggest a fix that seems to simplify the code a fair bit using an API with a lambda instead of a channel that could grow indefinitely.
This seems to fix the segfault for me but again take it with a lot of salt since I am not familiar with Zenoh or gz-transport:

diff --git a/src/NodeShared.cc b/src/NodeShared.cc
index 2d174850..2b44b609 100644
--- a/src/NodeShared.cc
+++ b/src/NodeShared.cc
@@ -338,28 +338,20 @@ NodeShared::NodeShared()
       opts.timeout_ms = kZenohLivelinessGetTimeoutMs;
 
       zenoh::ZResult result = Z_OK;
-      auto replies = this->Session()->liveliness_get(
+      this->Session()->liveliness_get(
         zenoh::KeyExpr("@gz/**"),
-        zenoh::channels::FifoChannel(SIZE_MAX - 1),
+        [this](zenoh::Reply &reply)
+        {
+          if (reply.is_ok())
+          {
+            const auto &sample = reply.get_ok();
+            this->dataPtr->msgDiscovery->LivelinessMsgDataHandler(sample);
+            this->dataPtr->srvDiscovery->LivelinessSrvDataHandler(sample);
+          }
+        },
+        []() {},
         std::move(opts),
         &result);
-
-      if (result == Z_OK)
-      {
-        for (auto res = replies.recv();
-             std::holds_alternative<zenoh::Reply>(res);
-             res = replies.recv())
-        {
-          const auto &reply = std::get<zenoh::Reply>(res);
-          if (!reply.is_ok())
-            continue;
-          const auto &sample = reply.get_ok();
-          // Both handlers filter by entityType internally, so it is
-          // safe to dispatch every sample to both.
-          this->dataPtr->msgDiscovery->LivelinessMsgDataHandler(sample);
-          this->dataPtr->srvDiscovery->LivelinessSrvDataHandler(sample);
-        }
-      }
     }
     catch (const zenoh::ZException &e)
     {

[caguero]

I think this won't work, I tried to run this and traced a segfault to this line. It seems what it tries to do is allocate a channel with a capacity of size_t-1 which is incredibly large, the allocation fails and the code segfaults. This is wildly out of my comfort zone so I had Gemini suggest a fix that seems to simplify the code a fair bit using an API with a lambda instead of a channel that could grow indefinitely. This seems to fix the segfault for me but again take it with a lot of salt since I am not familiar with Zenoh or gz-transport:

Thanks!

I could not reproduce the segfault here. Did it happen in gz-transport or testing with gz-sim or something else? I thought that the FifoChannel was doing only as a logical bound, not an actual preallocation. And just to confirm, are you testing with Zenoh 1.8.0 or a different version?

That said, I'm testing the callback approach with one latch. Otherwise, your suggestion is fire and forget and reintroduces the cold start race this PR is trying to fix.

[luca-della-vedova]

I'm on 1.7.2, I was testing with the CLI (i.e. gz topic -l).
Yea again can't really tell what is happening. I dug a bit deeper and indeed you are right and it doesn't preallocate and it was some sort of hallucination, now it's pointing to some sort of double free? Again not sure but the callback based method seems to work.
How have you been testing this? I have been building zenoh-c / zenoh-cpp / zenoh from source on the 1.7.2 tag in a colcon workspace

[caguero]

I'm testing running all tests in gz-transport, and then, running gz-sim, and making sure that it doesn't crash when shutting down. I'm on Zenoh 1.8.0 right now but 1.7.2 should be fine as well.

[luca-della-vedova]

Actually hem apologies, I tried again to do a full clean build and actually it was working fine? I'll try to stress test this set of PRs a bit more but in general functionally it looks good

[cheriehu]

I’ve tried some functional tests on this branch with a few of the example worlds and things seem to work as expected. This is what I’ve tried.

1. Sending cmd_vel via the command line on diff_drive.sdf. The robot moves.
2. Pausing, resuming, and resetting the simulation via a gz service call from the command line works
3. In sensors_demo.sdf, started the server and GUI in separate terminals. Echoed some camera topics, there was data being published.
4. Started a world with server and GUI and separate terminals. Spawned a box from the GUI and saw a debug message on the server terminal reflecting the change. Calling the/world/world_name/create gz service to spawn a box also led to the box being spawned in the GUI
5. Tried gz log record and gz log playback and verified that the playback looks right (at least for the /stats topic)