fix: don't clear wait_set triggered flag without holding condition_mutex

[karmanyaahm]

Summary

check_and_attach_condition() clears wait_set_data->triggered without holding condition_mutex, which can swallow a wakeup and park rmw_wait() forever with messages already queued. This deletes that write.

Observed on ROS 2 Humble, rmw_zenoh_cpp 0.1.9 (arm64, Ubuntu 22.04, Jetson), against a long-lived rosbridge-style node subscribed to many topics. 0.1.9 already contains the ros2#1015 deadlock fix — this is a separate race introduced by that fix, not the bug it fixed.

Root cause

rmw_wait() since ros2#1005 / ros2#1015:

{
  // reset the trigger prior to attaching any entities
  std::unique_lock<std::mutex> lock(wait_set_data->condition_mutex);
  wait_set_data->triggered = false;
}

{
  // We explicitly do not lock the condition_mutex here
  // ...
  // Note taking the mutex here leads to a deadlock.
  bool skip_wait = check_and_attach_condition(...);

  if (!skip_wait) {
    std::unique_lock<std::mutex> lock(wait_set_data->condition_mutex);
    wait_set_data->condition_variable.wait(
      lock, [wait_set_data]() { return wait_set_data->triggered; });
  }
}

But check_and_attach_condition() still ends with:

  // No conditions are available. Set the triggered flag of the wait_set to false.
  // Note that wait_set_data->condition_mutex has been locked before calling
  // check_and_attach_condition. So it's safe to modify the wait_set_data triggered flag.
  wait_set_data->triggered = false;

That comment's precondition is no longer true. ros2#1005 deliberately dropped the lock around this call to break the ABBA cycle in ros2#998, but this trailing write was left behind. It now races with the notifier paths, e.g. SubscriptionData::add_new_message():

  std::lock_guard<std::mutex> lock(mutex_);
  ...
  message_queue_.emplace_back(std::move(msg));
  if (wait_set_data_ != nullptr) {
    std::lock_guard<std::mutex> wait_set_lock(wait_set_data_->condition_mutex);
    wait_set_data_->triggered = true;
    wait_set_data_->condition_variable.notify_one();
  }

Interleaving:

	executor thread	zenoh RX thread
1	triggered = false (under lock)
2	check_and_attach_condition() attaches wait set to sub S
3		message arrives on S → triggered = true (under lock), notify_one() — no waiter yet
4	falls through: triggered = false (no lock) — clobbers step 3
5	takes lock, wait(pred) → predicate false → blocks with data queued

The race window is the entire attach loop, so the probability scales with the number of entities in the wait set and the message rate — consistent with ros2#998's report that likelihood rises with topic count and rate.

The write is also redundant: rmw_wait() already performs the same reset under condition_mutex immediately before calling this function.

Evidence

Thread stacks from the wedged process (16 threads, captured with gdb at the moment of the stall):

Thread 13:  pthread_cond_wait
            → rmw_wait          (librmw_zenoh_cpp.so)
            → rcl_wait
Thread 1:   __pthread_mutex_lock
            → rclcpp::executors::MultiThreadedExecutor::run()
            → MultiThreadedExecutor::spin() → main
Threads 11,12,14,15:  identical to Thread 1 — all blocked behind the executor mutex
Thread 10:  epoll_pwait → asio::io_context::run  (websocketpp — still alive)

One executor thread parked in rmw_wait; every other executor thread queued behind wait_mutex_, which MultiThreadedExecutor::run() holds across get_next_executable(). All callback dispatch stops process-wide.

While wedged:

• the topic was genuinely publishing — ros2 topic hz /tf reported a steady 99.0 Hz from another process throughout;
• the process stayed alive at ~0 % CPU, all threads sleeping (matches Deadlock (ABBA) between rmw_wait/check_and_attach_condition and SubscriptionData::add_new_message ros2/rmw_zenoh#998's description);
• zenoh's own RUST_LOG=zenoh=debug,zenoh_transport=trace output went completely silent for the stalled session while other nodes kept receiving;
• re-issuing the subscription restored delivery immediately (a fresh attach re-triggers the wait set), which is why "restart the client and it works again" masked this in production.

Reproduction

A rosbridge-style bridge (C++, MultiThreadedExecutor) with a websocket client subscribed to /tf at ~99 Hz alongside the node's other subscriptions. A single subscriber on an otherwise idle graph did not reproduce it — the wait set has to hold enough entities to widen the window.

Before / after

	time to wedge	messages delivered
0.1.9 unpatched	13 s, 22 s, 85 s, 108 s (4 for 4)	512 / 1 951 / 10 566 / 12 363
0.1.9 + this patch	no wedge	> 76 000 and counting, 12+ min continuous at 99 msg/s

Same binary, same host, same workload; the only delta is this deletion. Every unpatched run wedged permanently and never recovered.

Risk

Removing the reset can at worst produce a spurious wakeup — rmw_wait() returning with nothing ready — which the existing per-entity readiness checks after the wait already handle, and which callers must tolerate regardless. That failure mode is strictly more benign than an unrecoverable hang.

Notes

• Present identically on rolling, humble, jazzy and kilted as of this writing, so the same deletion should apply upstream.
• Related: Deadlock (ABBA) between rmw_wait/check_and_attach_condition and SubscriptionData::add_new_message ros2/rmw_zenoh#998 (the ABBA this stems from), Fix lock order inversion in rmw_wait ros2/rmw_zenoh#1005 (rolling fix), fix: Fix lock order inversion / deadlock (backport #1010) ros2/rmw_zenoh#1015 (humble backport, in 0.1.9).
• Branch base here is humble @ 0307698, which is 0.1.9 plus When SHM enabled, enable transport_optimization (backport #1020) ros2/rmw_zenoh#1024 and return early when unable to find any topic endpoints. (backport #1017) ros2/rmw_zenoh#1025.

Reproduction scripts

Everything used to find, confirm and measure this. All are standalone (pip install websockets) and talk plain rosbridge protocol to the bridge under test.

tf_wedge_probe.py — detects the wedge and proves the socket is still alive

#!/usr/bin/env python3
"""Subscribe to /tf over the rws websocket and report when the stream wedges.

Distinguishes a dead socket from a live-but-silent one: the ping/pong keepalive
runs independently of the data path, so a wedge shows up as pongs still landing
while /tf messages stop.

Usage: tf_wedge_probe.py [--url ws://127.0.0.1:9090] [--stall 2.0]
"""

import argparse
import asyncio
import json
import time

import websockets


def ts() -> str:
    return time.strftime("%H:%M:%S")


async def probe(url: str, stall_after: float, topic: str) -> None:
    print(f"[{ts()}] connecting to {url}")
    async with websockets.connect(url, max_size=None, ping_interval=5, ping_timeout=None) as ws:
        await ws.send(json.dumps({"op": "subscribe", "topic": topic, "type": "tf2_msgs/msg/TFMessage"}))
        print(f"[{ts()}] subscribed to {topic}; watching for stalls > {stall_after}s")

        total = 0
        window = 0
        last_msg = time.monotonic()
        last_report = last_msg
        wedged_since = None

        while True:
            try:
                raw = await asyncio.wait_for(ws.recv(), timeout=1.0)
            except asyncio.TimeoutError:
                raw = None
            except websockets.ConnectionClosed as exc:
                print(f"[{ts()}] SOCKET CLOSED: {exc} -- not a wedge, the connection died")
                return

            now = time.monotonic()

            if raw is not None:
                total += 1
                window += 1
                last_msg = now
                if wedged_since is not None:
                    print(f"[{ts()}] RECOVERED after {now - wedged_since:.1f}s")
                    wedged_since = None

            gap = now - last_msg
            if gap > stall_after and wedged_since is None:
                wedged_since = last_msg
                # The socket is provably alive here: this round-trips a ping frame
                # and only resolves when the peer's pong comes back.
                pong = await ws.ping()
                rtt_start = time.monotonic()
                try:
                    await asyncio.wait_for(pong, timeout=5.0)
                    alive = f"pong in {(time.monotonic() - rtt_start) * 1000:.0f}ms"
                except asyncio.TimeoutError:
                    alive = "NO PONG (socket also dead)"
                print(f"[{ts()}] WEDGED: no {topic} for {gap:.1f}s, socket {alive}, {total} msgs so far")

            if now - last_report >= 5.0:
                state = "WEDGED" if wedged_since else "ok"
                print(f"[{ts()}] {state}: {window / (now - last_report):6.1f} msg/s  total={total}")
                window = 0
                last_report = now


def main() -> None:
    ap = argparse.ArgumentParser()
    ap.add_argument("--url", default="ws://127.0.0.1:9090")
    ap.add_argument("--stall", type=float, default=2.0, help="seconds of silence that counts as a wedge")
    ap.add_argument("--topic", default="/tf")
    args = ap.parse_args()

    try:
        asyncio.run(probe(args.url, args.stall, args.topic))
    except KeyboardInterrupt:
        print(f"\n[{ts()}] stopped")


if __name__ == "__main__":
    main()

tf_wedge_capture.py — same, but shells out to gdb the instant it stalls (this produced the stacks above)

#!/usr/bin/env python3
"""Watch rws /tf and dump all thread stacks the instant it wedges.

The wedge is intermittent, so catching it by hand is unreliable. This holds a
subscription open and, the moment /tf goes silent while the socket is still
alive, shells out to gdb and captures every thread's backtrace -- the one piece
of evidence that says where delivery is stuck.

The sudo password is read from $SUDO_PASS so it is never written to disk.
Usage: SUDO_PASS=... tf_wedge_capture.py [--url ...] [--stall 2.0]
"""

import argparse
import asyncio
import json
import os
import subprocess
import time

import websockets


def ts() -> str:
    return time.strftime("%H:%M:%S")


def find_rws_pid(node_name: str) -> int | None:
    try:
        out = subprocess.run(
            ["pgrep", "-f", f"rws_server.*{node_name}"], capture_output=True, text=True, timeout=10
        ).stdout.split()
        return int(out[0]) if out else None
    except Exception:
        return None


def capture_stacks(pid: int, out_path: str, password: str) -> str:
    """gdb-attach to `pid` and write every thread's backtrace to out_path."""
    cmd = [
        "sudo", "-S", "-p", "",
        "gdb", "-p", str(pid), "-batch",
        "-ex", "set pagination off",
        "-ex", "info threads",
        "-ex", "thread apply all bt 20",
    ]
    try:
        res = subprocess.run(
            cmd, input=password + "\n", capture_output=True, text=True, timeout=180
        )
    except subprocess.TimeoutExpired:
        return "gdb timed out"

    with open(out_path, "w") as fh:
        fh.write(res.stdout)
        fh.write("\n=== stderr ===\n")
        fh.write(res.stderr)
    return f"wrote {out_path} ({len(res.stdout)} bytes)"


async def main() -> None:
    ap = argparse.ArgumentParser()
    ap.add_argument("--url", default="ws://127.0.0.1:9090")
    ap.add_argument("--stall", type=float, default=2.0)
    ap.add_argument("--topic", default="/tf")
    ap.add_argument("--node", default="ros_websocket_server")
    ap.add_argument("--out", default="wedge_stacks.txt")
    ap.add_argument("--max-wait", type=float, default=0, help="give up after N seconds (0 = forever)")
    args = ap.parse_args()

    password = os.environ.get("SUDO_PASS", "")
    if not password:
        print("SUDO_PASS not set -- will detect the wedge but cannot capture stacks")

    pid = find_rws_pid(args.node)
    print(f"[{ts()}] watching {args.url} {args.topic}; rws pid = {pid}")

    async with websockets.connect(args.url, max_size=None, ping_interval=5, ping_timeout=None) as ws:
        await ws.send(json.dumps({"op": "subscribe", "topic": args.topic, "type": "tf2_msgs/msg/TFMessage"}))

        total = 0
        last_msg = time.monotonic()
        last_report = last_msg
        started = last_msg
        window = 0

        while True:
            if args.max_wait and time.monotonic() - started > args.max_wait:
                print(f"[{ts()}] no wedge within {args.max_wait}s, giving up ({total} msgs)")
                return
            try:
                await asyncio.wait_for(ws.recv(), timeout=1.0)
                total += 1
                window += 1
                last_msg = time.monotonic()
            except asyncio.TimeoutError:
                pass
            except websockets.ConnectionClosed as exc:
                print(f"[{ts()}] socket closed ({exc}) -- process likely respawned, not a wedge")
                return

            now = time.monotonic()
            gap = now - last_msg

            if gap > args.stall:
                # Prove the socket is alive before blaming delivery.
                try:
                    await asyncio.wait_for(await ws.ping(), timeout=5.0)
                    alive = "socket alive (pong ok)"
                except (asyncio.TimeoutError, websockets.ConnectionClosed):
                    alive = "socket DEAD"
                print(f"[{ts()}] WEDGED after {total} msgs, {gap:.1f}s silent, {alive}")

                live_pid = find_rws_pid(args.node) or pid
                if live_pid != pid:
                    print(f"[{ts()}] pid changed {pid} -> {live_pid}: it respawned, not wedged")
                    return
                if password and live_pid:
                    print(f"[{ts()}] capturing stacks from pid {live_pid}...")
                    print(f"[{ts()}] {capture_stacks(live_pid, args.out, password)}")
                return

            if now - last_report >= 10.0:
                print(f"[{ts()}] ok {window / (now - last_report):5.1f} msg/s  total={total}")
                window = 0
                last_report = now


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print(f"\n[{ts()}] stopped")

tf_scope_probe.py — establishes the stall is server-wide, not per-connection

#!/usr/bin/env python3
"""Determine whether an rws /tf wedge is per-connection or server-wide.

Client A subscribes and waits to wedge. The moment it does, client B opens a
fresh connection and subscribes to the same topic. If B receives data while A
stays silent, the fault is scoped to A's connection/subscription. If B is also
silent, the whole server (or its ROS side) has stalled.
"""

import asyncio
import json
import time

import websockets

URL = "ws://127.0.0.1:9090"
TOPIC = "/tf"
SUB = json.dumps({"op": "subscribe", "topic": TOPIC, "type": "tf2_msgs/msg/TFMessage"})
STALL = 2.0


def ts() -> str:
    return time.strftime("%H:%M:%S")


async def count_for(label: str, seconds: float) -> int:
    """Open a fresh connection, subscribe, and count messages for `seconds`."""
    got = 0
    try:
        async with websockets.connect(URL, max_size=None, ping_interval=5, ping_timeout=None) as ws:
            await ws.send(SUB)
            deadline = time.monotonic() + seconds
            while time.monotonic() < deadline:
                try:
                    await asyncio.wait_for(ws.recv(), timeout=deadline - time.monotonic())
                    got += 1
                except asyncio.TimeoutError:
                    break
    except Exception as exc:
        print(f"[{ts()}] {label} connection error: {exc}")
    return got


async def main() -> None:
    print(f"[{ts()}] client A connecting, waiting for a wedge...")
    async with websockets.connect(URL, max_size=None, ping_interval=5, ping_timeout=None) as a:
        await a.send(SUB)
        total = 0
        last = time.monotonic()

        while True:
            try:
                await asyncio.wait_for(a.recv(), timeout=1.0)
                total += 1
                last = time.monotonic()
                continue
            except asyncio.TimeoutError:
                pass
            except websockets.ConnectionClosed as exc:
                print(f"[{ts()}] A socket closed ({exc}) -- restart and retry")
                return

            if time.monotonic() - last <= STALL:
                continue

            print(f"[{ts()}] A WEDGED after {total} msgs. Opening client B on the same server...")
            b_count = await count_for("B", 5.0)

            # Re-check A over the same window B just used.
            a_before = total
            try:
                await asyncio.wait_for(a.recv(), timeout=2.0)
                total += 1
            except (asyncio.TimeoutError, websockets.ConnectionClosed):
                pass
            a_moved = total - a_before

            print(f"[{ts()}] RESULT: A got {a_moved} msg in the window, B got {b_count} msg in 5s")
            if b_count > 0 and a_moved == 0:
                print(f"[{ts()}] => PER-CONNECTION: server is fine, A's subscription is dead")
            elif b_count == 0 and a_moved == 0:
                print(f"[{ts()}] => SERVER-WIDE: every client is starved")
            else:
                print(f"[{ts()}] => inconclusive (A recovered mid-test)")
            return


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print(f"\n[{ts()}] stopped")

tf_pub.py — minimal 100 Hz /tf publisher for the isolated control (which did not reproduce)

#!/usr/bin/env python3
"""Minimal /tf publisher at 100 Hz -- stands in for the full robot stack."""

import rclpy
from geometry_msgs.msg import TransformStamped
from rclpy.node import Node
from tf2_msgs.msg import TFMessage


class TfPub(Node):
    def __init__(self) -> None:
        super().__init__("isolated_tf_pub")
        self.pub = self.create_publisher(TFMessage, "/tf", 10)
        self.timer = self.create_timer(0.01, self.tick)
        self.n = 0

    def tick(self) -> None:
        t = TransformStamped()
        t.header.stamp = self.get_clock().now().to_msg()
        t.header.frame_id = "odom"
        t.child_frame_id = "base_link"
        t.transform.translation.x = float(self.n % 100) * 0.01
        t.transform.rotation.w = 1.0
        self.pub.publish(TFMessage(transforms=[t]))
        self.n += 1
        if self.n % 1000 == 0:
            self.get_logger().info(f"published {self.n}")


def main() -> None:
    rclpy.init()
    node = TfPub()
    try:
        rclpy.spin(node)
    except KeyboardInterrupt:
        pass
    finally:
        node.destroy_node()


if __name__ == "__main__":
    main()

[karmanyaahm]

This is vibecoded

[karmanyaahm]

just yolo for now