Issue 1:
Checklist

I have included the output of celery -A proj report in the issue.

(if you are not able to do this, then at least specify the Celery

version affected).

I have included all related issues and possible duplicate issues in this issue.

I have included the contents of pip freeze in the issue.

I have verified that the issue exists against the master branch of Celery.

I have tried reproducing the issue on more than one message broker and/or result backend.

I have tried reproducing the issue on more than one workers pool.

I have tried reproducing the issue with retries, ETA/Countdown & rate limits disabled.

ISSUE

Working with celery 4.2.0 and Redis as broker and backend.

Having a task

@shared_task()
def add(a, b):
return a+b

And while a worker is active, running the fallowing command:

add.apply_async(countdown=60) Results in the task not being registered to the default celery queue, but still being executed after the period of time stated in countdown

Why is that, and how can I look for all pending tasks? Doing this would have worked if the task would be registered to the queue:

with celery_app.pool.acquire(block=True) as conn:
tasks = conn.default_channel.client.lrange('celery', 0, -1)

If I terminate the worker while task havent been started I get the fallowing:

[WARNING/MainProcess] Restoring 1 unacknowledged message(s)

This tells me the task is kept somewhere else other then the queue, but why ?

Issue 2:
We've seen this CI failure sporadically since we fixed the pypy3 CI job. It possible an attribute in a timer is being mocked/assigned incorrectly, or there's just some difference in how pypy3 implements Timer objects (I believe the complaint is about a threading.Timer() object).

Checklist

I have verified that the issue exists against the master branch of Celery.

This has already been asked to the discussion group first.

I have read the relevant section in the

contribution guide

on reporting bugs.

I have checked the issues list

for similar or identical bug reports.

I have checked the pull requests list

for existing proposed fixes.

I have checked the commit log

to find out if the bug was already fixed in the master branch.

I have included all related issues and possible duplicate issues

in this issue (If there are none, check this box anyway).

Mandatory Debugging Information

I have included the output of celery -A proj report in the issue.

(if you are not able to do this, then at least specify the Celery

version affected).

I have verified that the issue exists against the master branch of Celery.

I have included the contents of pip freeze in the issue.

I have included all the versions of all the external dependencies required

to reproduce this bug.

Optional Debugging Information

I have tried reproducing the issue on more than one Python version

and/or implementation.

I have tried reproducing the issue on more than one message broker and/or

result backend.

I have tried reproducing the issue on more than one version of the message

broker and/or result backend.

I have tried reproducing the issue on more than one operating system.

I have tried reproducing the issue on more than one workers pool.

I have tried reproducing the issue with autoscaling, retries,

ETA/Countdown & rate limits disabled.

I have tried reproducing the issue after downgrading

and/or upgrading Celery and its dependencies.

Related Issues and Possible Duplicates

Related Issues

None

Possible Duplicates

None

Environment & Settings

Celery version:

celery report Output:

Steps to Reproduce

Required Dependencies

Minimal Python Version: N/A or Unknown

Minimal Celery Version: N/A or Unknown

Minimal Kombu Version: N/A or Unknown

Minimal Broker Version: N/A or Unknown

Minimal Result Backend Version: N/A or Unknown

Minimal OS and/or Kernel Version: N/A or Unknown

Minimal Broker Client Version: N/A or Unknown

Minimal Result Backend Client Version: N/A or Unknown

Python Packages

pip freeze Output:

Other Dependencies

N/A

Minimally Reproducible Test Case

Expected Behavior

Actual Behavior

Output from failed unit tests on pypy

```
t/unit/worker/test_worker.py::test_Consumer::test_receieve_message_eta_isoformat FAILED [ 97%]

t/unit/worker/test_worker.py::test_Consumer::test_receieve_message_eta_isoformat ERROR [ 97%]

==================================== ERRORS ====================================

____ ERROR at teardown of test_Consumer.test_receieve_message_eta_isoformat ____

self = <t.unit.worker.test_worker.test_Consumer object at 0x0000000008936528>

def teardown(self):

self.timer.stop()

t/unit/worker/test_worker.py:92:

self = <Timer(Timer-106, stopped daemon 139665216587520)>

def stop(self):

self._is_shutdown.set()

if self.running:

self._is_stopped.wait()

E AttributeError: 'bool' object has no attribute 'wait'

celery/utils/timer2.py:96: AttributeError

=================================== FAILURES ===================================

______________ test_Consumer.test_receieve_message_eta_isoformat _______________

self = <t.unit.worker.test_worker.test_Consumer object at 0x0000000008936528>

???

t/unit/worker/test_worker.py:417:

self = <Timer(Timer-106, stopped daemon 139665216587520)>

def stop(self):

self._is_shutdown.set()

if self.running:

self._is_stopped.wait()

E AttributeError: 'bool' object has no attribute 'wait'

celery/utils/timer2.py:96: AttributeError

</details>