test_computation_object_code_dask_compute fails when using --reruns

从 distributed/tests/test_client.py 和 test_computation_object_code_dask_compute 测试开始。使用 --reruns 3 运行报告中的 pytest 命令，然后与不使用重试的运行进行比较，同时检查 scheduler 为什么会收到两个代码条目。完成的标准是该测试在两种模式下都能可靠通过，且不削弱其 assertions。

Describe the issue:

Trying to tackle flaky tests, I added --reruns 3 to pytest. That causes test_computation_object_code_dask_compute to fail consistently, while it happily succeeds without --reruns/

Minimal Complete Verifiable Example:

python3 -m venv .venv
. .venv/bin/activate
pip install --prefer-binary .
pip install --prefer-binary pytest pytest-rerunfailures pytest-repeat pytest-anyio pytest-timeout numpy
python3 -m pytest -k 'test_computation_object_code_dask_compute' --reruns 3 --timeout_method=signal distributed/tests/test_client.py

Result:

=================================================================================================================================== FAILURES ===================================================================================================================================
__________________________________________________________________________________________________________________ test_computation_object_code_dask_compute ___________________________________________________________________________________________________________________

client = <Client: 'tcp://127.0.0.1:36961' processes=2 threads=2, memory=62.50 GiB>

    def test_computation_object_code_dask_compute(client):
        pytest.importorskip("numpy")
        da = pytest.importorskip("dask.array")
    
        with dask.config.set({"distributed.diagnostics.computations.nframes": 2}):
            x = da.ones((10, 10), chunks=(3, 3))
            x.sum().compute()
    
        test_function_code = inspect.getsource(test_computation_object_code_dask_compute)
    
        def fetch_comp_code(dask_scheduler):
            computations = list(dask_scheduler.computations)
            assert len(computations) == 1
            comp = computations[0]
            assert len(comp.code) == 1
            return comp.code[0]
    
        code = client.run_on_scheduler(fetch_comp_code)
>       assert len(code) == 1
E       assert 2 == 1
E        +  where 2 = len((SourceCode(code='def pytest_runtest_protocol(item, nextitem):\n    """\n    Run the test protocol.\n\n    Note: when teardown fails, two reports are generated for the case, one for\n    the test case and the other for the teardown error.\n    """\n    reruns = get_reruns_count(item)\n    if reruns is None:\n        # global setting is not specified, and this test is not marked with\n        # flaky\n        return\n\n    # while this doesn\'t need to be run with every item, it will fail on the\n    # first item if necessary\n    check_options(item.session.config)\n    delay = get_reruns_delay(item)\n    parallel = not is_master(item.config)\n    db = item.session.config.failures_db\n    item.execution_count = db.get_test_failures(item.nodeid)\n    db.set_test_reruns(item.nodeid, reruns)\n\n    if item.execution_count > reruns:\n        return True\n\n    need_to_run = True\n    while need_to_run:\n        item.execution_count += 1\n        item.ihook.pytest_runtest_logstart(nodeid=item.nodeid, location=item.location)\n        reports = runtestprotocol(item, nextitem=nextitem, log=False)\n\n        for report in reports:  # 3 reports: setup, call, teardown\n            report.rer...trigger rerun\n        else:\n            need_to_run = False\n\n        item.ihook.pytest_runtest_logfinish(nodeid=item.nodeid, location=item.location)\n\n    return True\n', lineno_frame=566, lineno_relative=29, filename='/home/sandro/devel/distributed/.py313/lib64/python3.13/site-packages/pytest_rerunfailures.py'), SourceCode(code='def test_computation_object_code_dask_compute(client):\n    pytest.importorskip("numpy")\n    da = pytest.importorskip("dask.array")\n\n    with dask.config.set({"distributed.diagnostics.computations.nframes": 2}):\n        x = da.ones((10, 10), chunks=(3, 3))\n        x.sum().compute()\n\n    test_function_code = inspect.getsource(test_computation_object_code_dask_compute)\n\n    def fetch_comp_code(dask_scheduler):\n        computations = list(dask_scheduler.computations)\n        assert len(computations) == 1\n        comp = computations[0]\n        assert len(comp.code) == 1\n        return comp.code[0]\n\n    code = client.run_on_scheduler(fetch_comp_code)\n    assert len(code) == 1\n    assert code[0].code == test_function_code\n', lineno_frame=7097, lineno_relative=6, filename='/home/sandro/devel/distributed/distributed/tests/test_client.py')))

distributed/tests/test_client.py:7109: AssertionError

Anything else we need to know?:

Rerunning tackles the flaky tests we've seen.

Environment:

• Dask version: 2025.4.0
• Python version: 3.13.2
• Operating System: Fedora 41
• Install method (conda, pip, source): source