Issue 1:
We're using Celery with MongoDB both as broker and results backend in production with quite a success. However recently we found that Mongo database used for Celery queues grown up to 50+ GBs. After short investigation I found that 99.9% of all documents in "messages" collection are belonging to "celeryev" queue.

Our workers have events enabled in order for Celery Flower frontend to work. I always assumed that event got wiped after consuming, but looks like I was wrong.

Celery report is as follows:

[root@app1 nimble]# bin/celery report

software -> celery:3.0.9nimble (Chiastic Slide) kombu:2.4.7nimble py:2.7.3

billiard:2.7.3.12 pymongo:2.1.1

platform -> system:Linux arch:64bit, ELF imp:CPython

loader -> celery.loaders.default.Loader

settings -> transport:mongodb results:mongodb

I believe that we see exactly the same behavior as in #436. I see three different celeryev queues, but we only have one single listener active (Flower):

SECONDARY> db.messages.distinct("queue", {queue: /^celeryev/})
[
"celeryev.2a304cc3-78b8-451c-9728-99d0c09ccde1",
"celeryev.3cbbd9b5-b967-4297-8a50-f3313dae0650",
"celeryev.d94fb4c0-8ab4-4d12-9469-f93e090a7631"
]

I also see a lot of "worker.heartbeat" routing keys in documents in these queues. Probably that's the majority of all events.

Distribution between these three queues is quite uniform:

SECONDARY> db.messages.group({key: {queue: true}, cond: {queue: /^celeryev/}, reduce: function(obj,prev) { prev.csum += 1; }, initial: { csum: 0 }})
[
{
"queue" : "celeryev.2a304cc3-78b8-451c-9728-99d0c09ccde1",
"csum" : 21619432
},
{
"queue" : "celeryev.3cbbd9b5-b967-4297-8a50-f3313dae0650",
"csum" : 14117210
},
{
"queue" : "celeryev.d94fb4c0-8ab4-4d12-9469-f93e090a7631",
"csum" : 18428199
}
]

I'm wondering, is it possible to use same fix for MongoDB as was used for Redis in #436?

Issue 2:
I am using Django-celery and Redis backend to push the tasks in celery queue. I am pushing tasks with same custom task_id in the queue. The documentation says that task_id is UUID which should be unique. But I am able to add more than one task with the same task_id in the queue. It is expected that it should raise an exception if a task with the same task_id is already there in the queue.

@task
def register_task(a, b):
print a + b

@task
def deregister_task(a, b, c):
print a + b + c

taskid=str(UUID(md5('test').hexdigest()))
register_task.apply_async(args=('hello', 'world'), task_id=taskid, countdown=60)
deregister_task.apply_async(args=('sample', 'hello', 'world'), task_id=taskid, countdown=60)
i = inspect()
i.scheduled()

{u'celery@Suyashs-MacBook-Pro.local': [{u'eta': u'2017-06-05T17:15:33.947002',
u'priority': 6,
u'request': {u'acknowledged': False,
u'args': u"('sample', 'hello', 'world')",
u'delivery_info': {u'exchange': u'celery',
u'priority': 0,
u'redelivered': None,
u'routing_key': u'celery'},
u'hostname': u'celery@Suyashs-MacBook-Pro.local',
u'id': u'098f6bcd-4621-d373-cade-4e832627b4f6',
u'kwargs': u'{}',
u'name': u'flight.tasks.deregister_task',
u'time_start': None,
u'worker_pid': None}},
{u'eta': u'2017-06-05T17:15:44.234537',
u'priority': 6,
u'request': {u'acknowledged': False,
u'args': u"('hello', 'world')",
u'delivery_info': {u'exchange': u'celery',
u'priority': 0,
u'redelivered': None,
u'routing_key': u'celery'},
u'hostname': u'celery@Suyashs-MacBook-Pro.local',
u'id': u'098f6bcd-4621-d373-cade-4e832627b4f6',
u'kwargs': u'{}',
u'name': u'flight.tasks.register_task',
u'time_start': None,
u'worker_pid': None}}]}