Merge pull request #70324 from dwoz/dwoz/fix/3008x-stress-tests-branch-build

tests/stress: fix two 3008.x branch-build flakes

Author: dwoz

tests/pytests/stress/master_subprocess/mworker/test_mworker_stress.py

@@ -268,17 +268,27 @@ def test_requester_disconnect_midflight_leaves_worker_alive(
 
         # MWorker had already dispatched and queued the reply for the
         # "drop-me" request before we closed the DEALER; on reconnect
-        # libzmq redelivers that queued reply to our fresh DEALER
-        # first.  Drain it, then send + receive a fresh request.
-        try:
-            stale = handle.recv(timeout=2.0)
-            log.info("drained stale reply after reconnect: %r", stale)
-        except TimeoutError:
-            # Some libzmq versions do not redeliver buffered replies
-            # after a peer identity change; that is fine too.
-            log.info("no stale reply queued")
-
-        good = handle.send_recv(_ping("after-reconnect"), timeout=10.0)
+        # libzmq may redeliver that queued reply to our fresh DEALER.
+        # The timing of that redelivery is not deterministic across
+        # libzmq builds and load levels: sometimes it arrives inside
+        # the first poll window, sometimes after we've already sent
+        # the follow-up request.  Rather than draining with a fixed
+        # timeout and hoping, we send the fresh request first, then
+        # drain replies until we see the one keyed to
+        # ``after-reconnect``.  Any earlier reply keyed to ``drop-me``
+        # is the redelivered stale that we're intentionally skipping.
+        handle.send(_ping("after-reconnect"))
+        good = None
+        drain_deadline = time.monotonic() + 15.0
+        while time.monotonic() < drain_deadline:
+            try:
+                reply = handle.recv(timeout=5.0)
+            except TimeoutError:
+                continue
+            if isinstance(reply, dict) and reply.get("id") == "after-reconnect":
+                good = reply
+                break
+            log.info("drained stale reply after reconnect: %r", reply)
         assert good == {"cmd": "ping", "id": "after-reconnect"}
     finally:
         handle.stop()

tests/pytests/stress/master_subprocess/mworkerqueue/test_mworkerqueue_stress.py

@@ -326,8 +326,54 @@ def test_requester_churn_fd_bounded(mworkerqueue, proc_stats):
     A truly unbounded per-connection leak would grow phase 2 as much or
     more than phase 1.
     """
+    import zmq.utils.monitor
+
     pid = mworkerqueue.process.pid
-    worker = mworkerqueue.worker()
+
+    # ``mworkerqueue.worker()`` only issues ``connect()``; the ROUTER↔DEALER
+    # zmq handshake and the internal DEALER's peer-list update are
+    # asynchronous.  On the first churn cycle we would otherwise race:
+    # a REQ can complete its own handshake with the ROUTER and land a
+    # message in the inner DEALER before the worker REP has registered
+    # as a routable peer, so the DEALER queues the message internally
+    # and ``_poll_recv(worker, 2000)`` in the churn loop times out.
+    #
+    # Wait for the worker REP's HANDSHAKE_SUCCEEDED against the proxy's
+    # inner DEALER *before* we start the churn loop.  This is a passive
+    # readiness signal — no test traffic is injected — so it can't
+    # leave stale probe messages queued at the DEALER or leave the REP
+    # in "must send" state.  (An earlier attempt that used an active
+    # probe REQ→worker round-trip pileged stale ``b"probe"`` messages
+    # onto the worker's inbound queue, which the churn loop then
+    # mis-read as its own reply key and ended up hanging on ``m.recv()``
+    # in the ROUTER→REQ direction.)
+    worker = mworkerqueue.ctx.socket(zmq.REP)
+    worker.setsockopt(zmq.LINGER, 500)
+    worker.setsockopt(zmq.RCVTIMEO, 5000)
+    worker.setsockopt(zmq.SNDTIMEO, 5000)
+    monitor = worker.get_monitor_socket()
+    worker.connect(mworkerqueue.dealer_uri)
+    # Keep the socket alive for the fixture's teardown sweep.
+    mworkerqueue._sockets.append(worker)  # noqa: SLF001 — intentional
+
+    handshake_deadline = time.monotonic() + 15.0
+    while time.monotonic() < handshake_deadline:
+        if not monitor.poll(500):
+            continue
+        try:
+            ev = zmq.utils.monitor.recv_monitor_message(monitor, zmq.NOBLOCK)
+        except zmq.error.Again:
+            continue
+        if ev.get("event") == zmq.Event.HANDSHAKE_SUCCEEDED:
+            break
+    else:
+        monitor.close(linger=0)
+        raise AssertionError(
+            "worker REP never completed HANDSHAKE_SUCCEEDED against the "
+            "inner DEALER within 15s (test setup race, not the FD-bound "
+            "assertion)"
+        )
+    monitor.close(linger=0)
 
     def _churn(n_cycles: int, id_prefix: str) -> None:
         for i in range(n_cycles):