Issue 1:
Checklist

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: 4.2.1 (windowlicker)

celery report Output:

... traceback removed ...
KeyError: 'test'

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

# settings.py
raise KeyError('test')

Expected Behavior

Traceback with KeyError: 'test'

Actual Behavior

Actual traceback when running celery -A proj beat -l DEBUG:

Traceback (most recent call last):
File "/usr/local/lib/python3.6/dist-packages/celery/utils/collections.py", line 134, in __getattr__
return self[k]
File "/usr/local/lib/python3.6/dist-packages/celery/utils/collections.py", line 442, in __getitem__
return self.__missing__(key)
File "/usr/local/lib/python3.6/dist-packages/celery/utils/collections.py", line 275, in __missing__
raise KeyError(key)
KeyError: 'beat_schedule_filename'

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
File "/usr/local/bin/celery", line 11, in <module>
sys.exit(main())
File "/usr/local/lib/python3.6/dist-packages/celery/__main__.py", line 16, in main
_main()
File "/usr/local/lib/python3.6/dist-packages/celery/bin/celery.py", line 322, in main
cmd.execute_from_commandline(argv)
File "/usr/local/lib/python3.6/dist-packages/celery/bin/celery.py", line 496, in execute_from_commandline
super(CeleryCommand, self).execute_from_commandline(argv)))
File "/usr/local/lib/python3.6/dist-packages/celery/bin/base.py", line 298, in execute_from_commandline
return self.handle_argv(self.prog_name, argv[1:])
File "/usr/local/lib/python3.6/dist-packages/celery/bin/celery.py", line 488, in handle_argv
return self.execute(command, argv)
File "/usr/local/lib/python3.6/dist-packages/celery/bin/celery.py", line 420, in execute
).run_from_argv(self.prog_name, argv[1:], command=argv[0])
File "/usr/local/lib/python3.6/dist-packages/celery/bin/base.py", line 302, in run_from_argv
sys.argv if argv is None else argv, command)
File "/usr/local/lib/python3.6/dist-packages/celery/bin/base.py", line 385, in handle_argv
*self.parse_options(prog_name, argv, command))
File "/usr/local/lib/python3.6/dist-packages/celery/bin/base.py", line 421, in parse_options
self.parser = self.create_parser(prog_name, command)
File "/usr/local/lib/python3.6/dist-packages/celery/bin/base.py", line 437, in create_parser
self.add_arguments(parser)
File "/usr/local/lib/python3.6/dist-packages/celery/bin/beat.py", line 116, in add_arguments
'-s', '--schedule', default=c.beat_schedule_filename)
File "/usr/local/lib/python3.6/dist-packages/celery/utils/collections.py", line 138, in __getattr__
type(self).__name__, k))
AttributeError: 'Settings' object has no attribute 'beat_schedule_filename'

Issue 2:
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.

Environment & Settings

Celery version: 4.3.0rc3 (rhubarb)

celery report Output:

software -> celery:4.3.0rc3 (rhubarb) kombu:4.4.0 py:3.5.3
billiard:3.6.0.0 py-amqp:2.4.2
platform -> system:Linux arch:64bit
kernel version:4.9.0-8-amd64 imp:CPython
loader -> celery.loaders.app.AppLoader
settings -> transport:pyamqp results:rpc:///

broker_url: 'amqp://guest:********@localhost:5672//'
result_backend: 'rpc:///'
include: ['pyodhean.interface']

Steps to Reproduce

Python Packages

pip freeze Output:

amqp==2.4.2
appdirs==1.4.3
astroid==2.2.5
autopep8==1.4.3
Babel==2.6.0
billiard==3.6.0.0
-e git+ssh://git@github.com/celery/celery@e1f84b06dc812d1a4520f717948328fbae7486be#egg=celery
flake8==3.6.0
flower==0.9.3
isort==4.3.4
kombu==4.4.0
lazy-object-proxy==1.3.1
mccabe==0.6.1
nose==1.3.7
pkg-resources==0.0.0
ply==3.11
pycodestyle==2.4.0
pyflakes==2.0.0
pylint==2.3.1
Pyomo==5.5.1
pytz==2018.9
PyUtilib==5.6.3
six==1.11.0
tornado==5.1.1
typed-ast==1.3.1
vine==1.3.0
wrapt==1.10.11

Minimally Reproducible Test Case

import time
from celery import Celery

app = Celery(
'test',
backend='rpc://',
broker='pyamqp://',
)

@app.task
def task():
return True

task_id = task.delay()
dummy_task_id = '00000000-0000-0000-0000-000000000000'

time.sleep(1)
assert app.AsyncResult(dummy_task_id).state == 'PENDING'
assert app.AsyncResult(task_id).state == 'SUCCESS'
assert app.AsyncResult(task_id).state == 'SUCCESS' # This fails

Expected Behavior

assert app.AsyncResult(task_id).state should always pass

Actual Behavior

The second assert app.AsyncResult(task_id).state fails.

I believe the issues resides in get_task_meta:

def get_task_meta(self, task_id, backlog_limit=1000):
buffered = self._out_of_band.pop(task_id, None)
if buffered:
return self._set_cache_by_message(task_id, buffered)

# Polling and using basic_get
latest_by_id = {}
prev = None
for acc in self._slurp_from_queue(task_id, self.accept, backlog_limit):
tid = self._get_message_task_id(acc)
prev, latest_by_id[tid] = latest_by_id.get(tid), acc
if prev:
# backends aren't expected to keep history,
# so we delete everything except the most recent state.
prev.ack()
prev = None

latest = latest_by_id.pop(task_id, None)
for tid, msg in items(latest_by_id):
self.on_out_of_band_result(tid, msg)

if latest:
latest.requeue()
return self._set_cache_by_message(task_id, latest)
else:
# no new state, use previous
try:
return self._cache[task_id]
except KeyError:
# result probably pending.
return {'status': states.PENDING, 'result': None}

When polling only task_id, the out_of_band buffer is never used, and latest is always requeued.

However, when another task is polled, the return message of the task_id task is buffered in out_of_band and never requeued, so it is returned correctly on the next call but then it is lost because it is popped from out_of_band.

I get the correct behaviour with any of those two fixes:

def on_out_of_band_result(self, task_id, message):
if self.result_consumer:
self.result_consumer.on_out_of_band_result(message)
self._out_of_band[task_id] = message
message.requeue() # <-- requeue

def get_task_meta(self, task_id, backlog_limit=1000):
buffered = self._out_of_band.pop(task_id, None)
if buffered:
buffered.requeue() # <-- requeue
return self._set_cache_by_message(task_id, buffered)
[...]

This is as far as I could get. It took me a few hours to figure that out. Hopefully this report provides enough information to be helpful.