diff --git a/tests/pytests/stress/master_subprocess/mworker/test_mworker_stress.py b/tests/pytests/stress/master_subprocess/mworker/test_mworker_stress.py index 3b0056da3d51..543150ccef85 100644 --- a/tests/pytests/stress/master_subprocess/mworker/test_mworker_stress.py +++ b/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() diff --git a/tests/pytests/stress/master_subprocess/mworkerqueue/test_mworkerqueue_stress.py b/tests/pytests/stress/master_subprocess/mworkerqueue/test_mworkerqueue_stress.py index 8c21f7a863b8..2e2be9ed41ec 100644 --- a/tests/pytests/stress/master_subprocess/mworkerqueue/test_mworkerqueue_stress.py +++ b/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)