Add flaky to a queue related test.

[karajan1001]

This is a flaky, the queue tasks status might remain "Running" even if
it had already been completed, beside sometimes I also noticed it raise
a pytest fixture error.

• ❗ I have followed the Contributing to DVC checklist.

• 📖 If this PR requires documentation updates, I have created a separate PR (or issue, at least) in dvc.org and linked it here.

Thank you for the contribution - we'll try to review it as soon as possible. 🙏

[skshetry]

@karajan1001, do you know why this happens? Is there something that we can fix?

[karajan1001]

@karajan1001, do you know why this happens? Is there something that we can fix?

Currently not, this behavior happens after I add some other unit test (in which I mocked some methods). Need some detailed investigation.

[karajan1001]

One example of fixture setup error.

_________ ERROR at setup of test_modified_data_dep[False-foo: 2-True] __________
[gw0] linux -- Python 3.10.5 /opt/hostedtoolcache/Python/3.10.5/x64/bin/python

request = <FixtureRequest for <Function test_modified_data_dep[False-foo: 2-True]>>

    def fill(request):
        item = request._pyfuncitem
        fixturenames = getattr(item, "fixturenames", None)
        if fixturenames is None:
            fixturenames = request.fixturenames
    
        if hasattr(item, 'callspec'):
            for param, val in sorted_by_dependency(item.callspec.params, fixturenames):
                if val is not None and is_lazy_fixture(val):
                    item.callspec.params[param] = request.getfixturevalue(val.name)
                elif param not in item.funcargs:
>                   item.funcargs[param] = request.getfixturevalue(param)

self = <celery.backends.filesystem.FilesystemBackend object at 0x7fe8dded7880>
task_id = 'bdbf0ab3-eaef-4e8c-8dc5-0ac905a52d1a', timeout = 10.0, interval = 0.5
no_ack = True, on_interval = <promise@0x7fe8dde4c4c0>

    def wait_for(self, task_id,
                 timeout=None, interval=0.5, no_ack=True, on_interval=None):
        """Wait for task and return its result.
    
        If the task raises an exception, this exception
        will be re-raised by :func:`wait_for`.
    
        Raises:
            celery.exceptions.TimeoutError:
                If `timeout` is not :const:`None`, and the operation
                takes longer than `timeout` seconds.
        """
        self._ensure_not_eager()
    
        time_elapsed = 0.0
    
        while 1:
            meta = self.get_task_meta(task_id)
            if meta['status'] in states.READY_STATES:
                return meta
            if on_interval:
                on_interval()
            # avoid hammering the CPU checking status.
            time.sleep(interval)
            time_elapsed += interval
            if timeout and time_elapsed >= timeout:
>               raise TimeoutError('The operation timed out.')
E               celery.exceptions.TimeoutError: The operation timed out.

/opt/hostedtoolcache/Python/3.10.5/x64/lib/python3.10/site-packages/celery/backends/base.py:792: TimeoutError

This is a celery timeout error, and I guess it comes out when the celery starts when we have a heavy CPU load. It can't get enough resources?

[skshetry]

@karajan1001, is it possible to increase the timeout then?

[pmrowla]

We can probably disable the default pytest-celery worker ping timeout:

https://github.com/celery/celery/blob/128f0027005f2bf9d4b93082049d2c96c2bcd879/celery/contrib/testing/worker.py#L63

it should just need to be set in our fixture:
https://github.com/iterative/dvc/blob/7a0b3e93771c212fea3fa7d80f710877e9fdf958/tests/unit/repo/experiments/conftest.py#L15