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

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

@@ -329,6 +329,51 @@ def test_requester_churn_fd_bounded(mworkerqueue, proc_stats):
     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, at which point the message sits in the
+    # DEALER's queue with nowhere to send and ``_poll_recv(worker, 2000)``
+    # times out.  Warm the outbound pipeline until the worker actually
+    # receives.  The churn loop below never reads the return path, so
+    # we don't wait for the probe reply either (avoids racing on the
+    # return leg being ready) — but we do complete the REP FSM cycle
+    # so ``worker`` isn't left in "must send" state going into churn.
+    probe = mworkerqueue.ctx.socket(zmq.REQ)
+    try:
+        probe.setsockopt(zmq.LINGER, 0)
+        probe.setsockopt(zmq.SNDTIMEO, 2000)
+        probe.connect(mworkerqueue.router_uri)
+        probe_deadline = time.monotonic() + 10.0
+        while time.monotonic() < probe_deadline:
+            probe.send(b"probe")
+            msg = _poll_recv(worker, 1000)
+            if msg is not None:
+                # Complete the REP FSM cycle so ``worker`` is ready to
+                # ``recv()`` again in the churn loop.  The reply may be
+                # dropped by ROUTER (probe closes before it routes) —
+                # that's fine; we only need the FSM state, not delivery.
+                try:
+                    worker.send(b"probe-ack")
+                except zmq.error.Again:
+                    pass
+                break
+            # REQ FSM stalls on double-send without recv; recycle.
+            probe.close(linger=0)
+            probe = mworkerqueue.ctx.socket(zmq.REQ)
+            probe.setsockopt(zmq.LINGER, 0)
+            probe.setsockopt(zmq.SNDTIMEO, 2000)
+            probe.connect(mworkerqueue.router_uri)
+        else:
+            raise AssertionError(
+                "worker never became a routable peer of the inner DEALER "
+                "within 10s (test setup race, not the FD-bound assertion)"
+            )
+    finally:
+        probe.close(linger=0)
+
     def _churn(n_cycles: int, id_prefix: str) -> None:
         for i in range(n_cycles):
             m = mworkerqueue.ctx.socket(zmq.REQ)