Redeliver same tasks repeatedly with redis broker and gevent

[wuqiallen]

Hi,

kombu==4.6.4
celery==4.3.0
redis==3.2.1

The visibility_timeout is ignored when using redis broker and gevent. As the result, when launching multiple workers with redis broker and gevent, the tasks will be re-delivered repeatedly.
Maybe the reason is #905.

def restore_visible(self, start=0, num=10, interval=10):
......
ceil = time() - self.visibility_timeout
......
env = _detect_environment()
if env == 'gevent':
    ceil = time()
visible = client.zrevrangebyscore(
    self.unacked_index_key, ceil, 0,
    start=num and start, num=num, withscores=True)
for tag, score in visible or []:
    self.restore_by_tag(tag, client)

When env==gevent, ceil will be changed to time() from time() - self.visibility_timeout.
As a result, all the tasks even the newly added ones in unacked set will be fetched out into visible, and then re-delivered by calling restore_by_tag, ignoring the functionality of visibility_timeout. The function restore_visible in QoS is called by maybe_restore_messages

    def maybe_restore_messages(self):
        for channel in self._channels:
            if channel.active_queues:
                # only need to do this once, as they are not local to channel.
                return channel.qos.restore_visible(
                    num=channel.unacked_restore_limit,
                )

But the function maybe_restore_messages is further called by other methods multiple times. Particularly in register_with_event_loop, where maybe_restore_messages is called every 10 seconds.

    def register_with_event_loop(self, connection, loop):
        ...
        loop.call_repeatedly(10, cycle.maybe_restore_messages)

So when launching multiple workers with redis broker and gevent will re-deliver all tasks in unacked set repeatedly.

[matteius]

@wuqiallen I will ask that you also try this with the master branch of kombu in place of 4.6.4 -- I did a bunch of work with Redis recently that was merged: https://github.com/celery/kombu/pull/1089/files

While your issue report may be different, we need to know if it is still an issue for you on the master branch as we are prepping for a soon release. If we know it is still an issue than a possible fix may be discovered.

[matteius]

Also try master celery branch with that please, or at least latest. Also for what its worth latest redis on pypi is redis==3.3.8

[wuqiallen]

@matteius Thanks for your reply. I will try to reproduce this issue with the env which you talked me. I will let you know the results as soon as I can.

[arikfr]

This still reproduces on latest versions, as the relevant code is still there:

kombu/kombu/transport/redis.py

Lines 197 to 199 in b51d1d6

	env = _detect_environment()
	if env == 'gevent':
	ceil = time()

Whenever restore_visible is called in a gevent environment it will restore all unacked messages, regardless of their visibility timeout. While #905 tried to fix an issue with not restoring messages when worker starts, they broke this functionality in all other cases (restore_visible is called periodically and not only when worker starts).

In our case, we don't even use gevent based workers. But our API services do use gevent. When we use celery.control to monitor the status of Celery (we have an API to return queues status), it triggers restore_visible with this broken code. It took us a while to find this 😢

[deepanjan-nsk]

Can confirm @arikfr 's statement that we are still seeing this problem on 3rd feb 2019 as the code in question is still present.

[arikfr]

@deepanjan-nsk we ended up patching this in our codebase (and later replacing Celery with RQ). If you want I can try and find the patch.

[deepanjan-nsk]

Thanks @arikfr, We ended up doing the same w.r.t. a patch in our codebase.

[calvin620707]

Also, we do a patch to remove ceil = time() to avoid duplicated tasks since we don't care too much regarding losing tasks after workers restart.

[auvipy]

celery==4.4.*?

[calvin620707]

I had reproduced this with celery==4.4.0 and kombu==4.6.7. Actually, it seems like a kombu's issue instead of celery.

[auvipy]

try celery==4.4.2 and report again please

[mayouzi]

replace gevent with eventlet

[mayouzi]

replace gevent with eventlet

it`s necessary to remind that make a HTTP request cautious when use eventlet + requests
which need to install pyopenssl with pip install pyOpenSSL

[wuqiallen]

fixed on #1259