Django-celery-beat: Beat scheduler crashes due to race condition

[alon-unifi]

Problem

Previously reported by @poloping on the celery project tracker here, and encountered by myself as well - if a schedule gets updated just around the time that the 3 minute sync cycle elapses, one of the sync operations gets a closed connection and the scheduler crashes.

I've traced it to a previous fix, in this pull request https://github.com/celery/django-celery-beat/pull/58, where a call to close_old_connections() was added to the sync method. This PR seems to fix an issue for MySQL but breaks postgresql in my case. I'm using CONN_MAX_AGE = 0

Using:

celery 4.2.0
django-celery-beat 1.1.1

Trace:

[2018-08-23 20:00:33,887: CRITICAL/MainProcess] beat raised exception <class 'django.db.utils.InterfaceError'>: InterfaceError('connection already closed',)
Traceback (most recent call last):
  File "/Users/alon/dev/unifi_virtenv/lib/python2.7/site-packages/celery/apps/beat.py", line 109, in start_scheduler
    service.start()
  File "/Users/alon/dev/unifi_virtenv/lib/python2.7/site-packages/celery/beat.py", line 595, in start
    self.scheduler._do_sync()
  File "/Users/alon/dev/unifi_virtenv/lib/python2.7/site-packages/celery/beat.py", line 374, in _do_sync
    self.sync()
  File "/Users/alon/dev/unifi_virtenv/lib/python2.7/site-packages/django_celery_beat/schedulers.py", line 243, in sync
    self.schedule[name].save()
  File "/Users/alon/dev/unifi_virtenv/lib/python2.7/site-packages/django_celery_beat/schedulers.py", line 289, in schedule
    self.sync()
  File "/Users/alon/dev/unifi_virtenv/lib/python2.7/site-packages/django_celery_beat/schedulers.py", line 243, in sync
    self.schedule[name].save()
  File "/Users/alon/dev/unifi_virtenv/lib/python2.7/site-packages/django_celery_beat/schedulers.py", line 284, in schedule
    elif self.schedule_changed():
  File "/Users/alon/dev/unifi_virtenv/lib/python2.7/site-packages/django_celery_beat/schedulers.py", line 215, in schedule_changed
    last, ts = self._last_timestamp, self.Changes.last_change()
  File "/Users/alon/dev/unifi_virtenv/lib/python2.7/site-packages/django_celery_beat/models.py", line 226, in last_change
    return cls.objects.get(ident=1).last_update
  File "/Users/alon/dev/unifi_virtenv/lib/python2.7/site-packages/django/db/models/manager.py", line 85, in manager_method
    return getattr(self.get_queryset(), name)(*args, **kwargs)
  File "/Users/alon/dev/unifi_virtenv/lib/python2.7/site-packages/django/db/models/query.py", line 374, in get
    num = len(clone)
  File "/Users/alon/dev/unifi_virtenv/lib/python2.7/site-packages/django/db/models/query.py", line 232, in __len__
    self._fetch_all()
  File "/Users/alon/dev/unifi_virtenv/lib/python2.7/site-packages/django/db/models/query.py", line 1118, in _fetch_all
    self._result_cache = list(self._iterable_class(self))
  File "/Users/alon/dev/unifi_virtenv/lib/python2.7/site-packages/django/db/models/query.py", line 53, in __iter__
    results = compiler.execute_sql(chunked_fetch=self.chunked_fetch)
  File "/Users/alon/dev/unifi_virtenv/lib/python2.7/site-packages/django/db/models/sql/compiler.py", line 887, in execute_sql
    cursor = self.connection.cursor()
  File "/Users/alon/dev/unifi_virtenv/lib/python2.7/site-packages/django/db/backends/base/base.py", line 254, in cursor
    return self._cursor()
  File "/Users/alon/dev/unifi_virtenv/lib/python2.7/site-packages/django/db/backends/base/base.py", line 231, in _cursor
    return self._prepare_cursor(self.create_cursor(name))
  File "/Users/alon/dev/unifi_virtenv/lib/python2.7/site-packages/django/db/utils.py", line 94, in __exit__
    six.reraise(dj_exc_type, dj_exc_value, traceback)
  File "/Users/alon/dev/unifi_virtenv/lib/python2.7/site-packages/django/db/backends/base/base.py", line 231, in _cursor
    return self._prepare_cursor(self.create_cursor(name))
  File "/Users/alon/dev/unifi_virtenv/lib/python2.7/site-packages/django/db/backends/postgresql/base.py", line 220, in create_cursor
    cursor = self.connection.cursor()
InterfaceError: connection already closed
celery beat v4.2.0 (windowlicker) is starting.
__    -    ... __   -        _
LocalTime -> 2018-08-23 19:58:28
Configuration ->
    . broker -> redis://localhost:6379//
    . loader -> celery.loaders.app.AppLoader
    . scheduler -> django_celery_beat.schedulers.DatabaseScheduler

    . logfile -> [stderr]@%DEBUG
    . maxinterval -> 5.00 seconds (5s)
[2018-08-23 20:00:33,895: WARNING/MainProcess] Traceback (most recent call last):
[2018-08-23 20:00:33,895: WARNING/MainProcess] File "/Users/alon/dev/unifi_virtenv/bin/celery", line 11, in <module>
[2018-08-23 20:00:33,896: WARNING/MainProcess] sys.exit(main())
[2018-08-23 20:00:33,896: WARNING/MainProcess] File "/Users/alon/dev/unifi_virtenv/lib/python2.7/site-packages/celery/__main__.py", line 16, in main
[2018-08-23 20:00:33,896: WARNING/MainProcess] _main()
[2018-08-23 20:00:33,897: WARNING/MainProcess] File "/Users/alon/dev/unifi_virtenv/lib/python2.7/site-packages/celery/bin/celery.py", line 322, in main
[2018-08-23 20:00:33,897: WARNING/MainProcess] cmd.execute_from_commandline(argv)
[2018-08-23 20:00:33,897: WARNING/MainProcess] File "/Users/alon/dev/unifi_virtenv/lib/python2.7/site-packages/celery/bin/celery.py", line 496, in execute_from_commandline
[2018-08-23 20:00:33,898: WARNING/MainProcess] super(CeleryCommand, self).execute_from_commandline(argv)))
[2018-08-23 20:00:33,898: WARNING/MainProcess] File "/Users/alon/dev/unifi_virtenv/lib/python2.7/site-packages/celery/bin/base.py", line 275, in execute_from_commandline
[2018-08-23 20:00:33,898: WARNING/MainProcess] return self.handle_argv(self.prog_name, argv[1:])
[2018-08-23 20:00:33,898: WARNING/MainProcess] File "/Users/alon/dev/unifi_virtenv/lib/python2.7/site-packages/celery/bin/celery.py", line 488, in handle_argv
[2018-08-23 20:00:33,898: WARNING/MainProcess] return self.execute(command, argv)
[2018-08-23 20:00:33,899: WARNING/MainProcess] File "/Users/alon/dev/unifi_virtenv/lib/python2.7/site-packages/celery/bin/celery.py", line 420, in execute
[2018-08-23 20:00:33,899: WARNING/MainProcess] ).run_from_argv(self.prog_name, argv[1:], command=argv[0])
[2018-08-23 20:00:33,899: WARNING/MainProcess] File "/Users/alon/dev/unifi_virtenv/lib/python2.7/site-packages/celery/bin/base.py", line 279, in run_from_argv
[2018-08-23 20:00:33,899: WARNING/MainProcess] sys.argv if argv is None else argv, command)
[2018-08-23 20:00:33,899: WARNING/MainProcess] File "/Users/alon/dev/unifi_virtenv/lib/python2.7/site-packages/celery/bin/base.py", line 363, in handle_argv
[2018-08-23 20:00:33,899: WARNING/MainProcess] return self(*args, **options)
[2018-08-23 20:00:33,899: WARNING/MainProcess] File "/Users/alon/dev/unifi_virtenv/lib/python2.7/site-packages/celery/bin/base.py", line 238, in __call__
[2018-08-23 20:00:33,899: WARNING/MainProcess] ret = self.run(*args, **kwargs)
[2018-08-23 20:00:33,900: WARNING/MainProcess] File "/Users/alon/dev/unifi_virtenv/lib/python2.7/site-packages/celery/bin/beat.py", line 109, in run
[2018-08-23 20:00:33,900: WARNING/MainProcess] return beat().run()
[2018-08-23 20:00:33,900: WARNING/MainProcess] File "/Users/alon/dev/unifi_virtenv/lib/python2.7/site-packages/celery/apps/beat.py", line 81, in run
[2018-08-23 20:00:33,900: WARNING/MainProcess] self.start_scheduler()
[2018-08-23 20:00:33,900: WARNING/MainProcess] File "/Users/alon/dev/unifi_virtenv/lib/python2.7/site-packages/celery/apps/beat.py", line 109, in start_scheduler
[2018-08-23 20:00:33,901: WARNING/MainProcess] service.start()
[2018-08-23 20:00:33,901: WARNING/MainProcess] File "/Users/alon/dev/unifi_virtenv/lib/python2.7/site-packages/celery/beat.py", line 595, in start
[2018-08-23 20:00:33,901: WARNING/MainProcess] self.scheduler._do_sync()
[2018-08-23 20:00:33,901: WARNING/MainProcess] File "/Users/alon/dev/unifi_virtenv/lib/python2.7/site-packages/celery/beat.py", line 374, in _do_sync
[2018-08-23 20:00:33,901: WARNING/MainProcess] self.sync()
[2018-08-23 20:00:33,901: WARNING/MainProcess] File "/Users/alon/dev/unifi_virtenv/lib/python2.7/site-packages/django_celery_beat/schedulers.py", line 243, in sync
[2018-08-23 20:00:33,901: WARNING/MainProcess] self.schedule[name].save()
[2018-08-23 20:00:33,901: WARNING/MainProcess] File "/Users/alon/dev/unifi_virtenv/lib/python2.7/site-packages/django_celery_beat/schedulers.py", line 289, in schedule
[2018-08-23 20:00:33,902: WARNING/MainProcess] self.sync()
[2018-08-23 20:00:33,902: WARNING/MainProcess] File "/Users/alon/dev/unifi_virtenv/lib/python2.7/site-packages/django_celery_beat/schedulers.py", line 243, in sync
[2018-08-23 20:00:33,902: WARNING/MainProcess] self.schedule[name].save()
[2018-08-23 20:00:33,902: WARNING/MainProcess] File "/Users/alon/dev/unifi_virtenv/lib/python2.7/site-packages/django_celery_beat/schedulers.py", line 284, in schedule
[2018-08-23 20:00:33,902: WARNING/MainProcess] elif self.schedule_changed():
[2018-08-23 20:00:33,902: WARNING/MainProcess] File "/Users/alon/dev/unifi_virtenv/lib/python2.7/site-packages/django_celery_beat/schedulers.py", line 215, in schedule_changed
[2018-08-23 20:00:33,902: WARNING/MainProcess] last, ts = self._last_timestamp, self.Changes.last_change()
[2018-08-23 20:00:33,902: WARNING/MainProcess] File "/Users/alon/dev/unifi_virtenv/lib/python2.7/site-packages/django_celery_beat/models.py", line 226, in last_change
[2018-08-23 20:00:33,902: WARNING/MainProcess] return cls.objects.get(ident=1).last_update
[2018-08-23 20:00:33,903: WARNING/MainProcess] File "/Users/alon/dev/unifi_virtenv/lib/python2.7/site-packages/django/db/models/manager.py", line 85, in manager_method
[2018-08-23 20:00:33,903: WARNING/MainProcess] return getattr(self.get_queryset(), name)(*args, **kwargs)
[2018-08-23 20:00:33,903: WARNING/MainProcess] File "/Users/alon/dev/unifi_virtenv/lib/python2.7/site-packages/django/db/models/query.py", line 374, in get
[2018-08-23 20:00:33,903: WARNING/MainProcess] num = len(clone)
[2018-08-23 20:00:33,903: WARNING/MainProcess] File "/Users/alon/dev/unifi_virtenv/lib/python2.7/site-packages/django/db/models/query.py", line 232, in __len__
[2018-08-23 20:00:33,903: WARNING/MainProcess] self._fetch_all()
[2018-08-23 20:00:33,903: WARNING/MainProcess] File "/Users/alon/dev/unifi_virtenv/lib/python2.7/site-packages/django/db/models/query.py", line 1118, in _fetch_all
[2018-08-23 20:00:33,904: WARNING/MainProcess] self._result_cache = list(self._iterable_class(self))
[2018-08-23 20:00:33,904: WARNING/MainProcess] File "/Users/alon/dev/unifi_virtenv/lib/python2.7/site-packages/django/db/models/query.py", line 53, in __iter__
[2018-08-23 20:00:33,904: WARNING/MainProcess] results = compiler.execute_sql(chunked_fetch=self.chunked_fetch)
[2018-08-23 20:00:33,904: WARNING/MainProcess] File "/Users/alon/dev/unifi_virtenv/lib/python2.7/site-packages/django/db/models/sql/compiler.py", line 887, in execute_sql
[2018-08-23 20:00:33,904: WARNING/MainProcess] cursor = self.connection.cursor()
[2018-08-23 20:00:33,904: WARNING/MainProcess] File "/Users/alon/dev/unifi_virtenv/lib/python2.7/site-packages/django/db/backends/base/base.py", line 254, in cursor
[2018-08-23 20:00:33,904: WARNING/MainProcess] return self._cursor()
[2018-08-23 20:00:33,904: WARNING/MainProcess] File "/Users/alon/dev/unifi_virtenv/lib/python2.7/site-packages/django/db/backends/base/base.py", line 231, in _cursor
[2018-08-23 20:00:33,905: WARNING/MainProcess] return self._prepare_cursor(self.create_cursor(name))
[2018-08-23 20:00:33,905: WARNING/MainProcess] File "/Users/alon/dev/unifi_virtenv/lib/python2.7/site-packages/django/db/utils.py", line 94, in __exit__
[2018-08-23 20:00:33,905: WARNING/MainProcess] six.reraise(dj_exc_type, dj_exc_value, traceback)
[2018-08-23 20:00:33,905: WARNING/MainProcess] File "/Users/alon/dev/unifi_virtenv/lib/python2.7/site-packages/django/db/backends/base/base.py", line 231, in _cursor
[2018-08-23 20:00:33,905: WARNING/MainProcess] return self._prepare_cursor(self.create_cursor(name))
[2018-08-23 20:00:33,905: WARNING/MainProcess] File "/Users/alon/dev/unifi_virtenv/lib/python2.7/site-packages/django/db/backends/postgresql/base.py", line 220, in create_cursor
[2018-08-23 20:00:33,905: WARNING/MainProcess] cursor = self.connection.cursor()
[2018-08-23 20:00:33,905: WARNING/MainProcess] django.db.utils
[2018-08-23 20:00:33,905: WARNING/MainProcess] .
[2018-08-23 20:00:33,905: WARNING/MainProcess] InterfaceError
[2018-08-23 20:00:33,906: WARNING/MainProcess] :
[2018-08-23 20:00:33,906: WARNING/MainProcess] connection already closed
[2018-08-23 20:00:33,906: INFO/MainProcess] Writing entries...
[2018-08-23 20:00:33,910: INFO/MainProcess] Writing entries...

note that from this point on nothing gets scheduled until beat gets restarted.

@bigmars86 @auvipy - as you've been working on the previous PR do you have any ideas what may be causing this on postgresql and what I should be looking it to get to the root cause?

for now I've worked around the issue by using a custom scheduler class that extends the DatabaseScheduler class and overrides the sync() method to exclude the close_old_connections() call. This made the issue go away. This is also the version of sync() that was used by djcelery and that worked for me in the last few years with no hiccups.

Thank you.

[wrvdklooster]

I have the same issue and took same solution (using own sync() without closing connections).

[ukrutt]

I have a similar issue, with

celery==4.2.1
django-celery-beat==1.2.0
Django==2.0.9
psql (PostgreSQL) 10.5

Traceback:

2018-10-11 12:00:39,495 ERROR: Beat django_celery_beat.schedulers#L268: Database error while sync: InterfaceError('connection already closed',)
Traceback (most recent call last):
  File "/.../.venv/lib/python3.6/site-packages/django/db/backends/base/base.py", line 234, in _cursor
    return self._prepare_cursor(self.create_cursor(name))
  File "/.../.venv/lib/python3.6/site-packages/django/db/backends/postgresql/base.py", line 212, in create_cursor
    cursor = self.connection.cursor()
psycopg2.InterfaceError: connection already closed

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "/.../.venv/lib/python3.6/site-packages/django_celery_beat/schedulers.py", line 262, in sync
    self.schedule[name].save()
  File "/.../.venv/lib/python3.6/site-packages/django_celery_beat/schedulers.py", line 303, in schedule
    elif self.schedule_changed():
  File "/.../.venv/lib/python3.6/site-packages/django_celery_beat/schedulers.py", line 234, in schedule_changed
    last, ts = self._last_timestamp, self.Changes.last_change()
  File "/.../.venv/lib/python3.6/site-packages/django_celery_beat/models.py", line 240, in last_change
    return cls.objects.get(ident=1).last_update
  File "/.../.venv/lib/python3.6/site-packages/django/db/models/manager.py", line 82, in manager_method
    return getattr(self.get_queryset(), name)(*args, **kwargs)
  File "/.../.venv/lib/python3.6/site-packages/django/db/models/query.py", line 397, in get
    num = len(clone)
  File "/.../.venv/lib/python3.6/site-packages/django/db/models/query.py", line 254, in __len__
    self._fetch_all()
  File "/.../.venv/lib/python3.6/site-packages/django/db/models/query.py", line 1182, in _fetch_all
    self._result_cache = list(self._iterable_class(self))
  File "/.../.venv/lib/python3.6/site-packages/django/db/models/query.py", line 53, in __iter__
    results = compiler.execute_sql(chunked_fetch=self.chunked_fetch, chunk_size=self.chunk_size)
  File "/.../.venv/lib/python3.6/site-packages/django/db/models/sql/compiler.py", line 1066, in execute_sql
    cursor = self.connection.cursor()
  File "/.../.venv/lib/python3.6/site-packages/django/db/backends/base/base.py", line 255, in cursor
    return self._cursor()
  File "/.../.venv/lib/python3.6/site-packages/django/db/backends/base/base.py", line 234, in _cursor
    return self._prepare_cursor(self.create_cursor(name))
  File "/.../.venv/lib/python3.6/site-packages/django/db/utils.py", line 89, in __exit__
    raise dj_exc_value.with_traceback(traceback) from exc_value
  File "/.../.venv/lib/python3.6/site-packages/django/db/backends/base/base.py", line 234, in _cursor
    return self._prepare_cursor(self.create_cursor(name))
  File "/.../.venv/lib/python3.6/site-packages/django/db/backends/postgresql/base.py", line 212, in create_cursor
    cursor = self.connection.cursor()
django.db.utils.InterfaceError: connection already closed

[ukrutt]

I have a question about the advertised workaround: @alon-unifi suggests overriding the sync() method, while @norac89 suggests (in https://github.com/celery/django-celery-beat/issues/159#issuecomment-426412346) overriding the schedule_changed() method (either would remove the close_old_connections() call). Which one is "better"? (FWIW, it seems that https://github.com/celery/django-celery-beat/pull/157 / https://github.com/celery/django-celery-beat/commit/95ab2e403d4f46f283080807eafd68870a062cae added close_old_connections() to schedule_changed(), so I'm inclined to simply revert that...)

[alon-unifi]

Looks like this call was added to schedule_changed() in 1.2.0. I'm on 1.1.1 so don't have to worry about it just yet. I think this makes the problem now much worse since the schedule changes are checked more often than the per 3 minute sync call (may be wrong). So @ukrutt I'd think you need to remove it from both, or consider reverting to an older version of dcb - the last one before these changes were made (1.1.0?)
@andziak , @auvipy, @ask - there seem to be at least of couple of recent commits that fix mysql issues but seem to break postgresql. I'm assuming this is undesirable for this project, or are you targeting mysql specifically?

[ukrutt]

Thanks @alon-unifi -- I didn't realize that both might need to be removed. It seems that downgrading to 1.1.0 may be the way to go (yes, the call in sync() was added with 9001f0f, i.e., indeed after 1.1.0.)

[sdrew71]

I have the same issue with - my concern with reverting back to 1.1.0 is the following listed in the Change History:

"Fix task heap invalidation bug which prevented scheduled tasks from running when syncing tasks from the database."

From this comment it sounds like if I revert back and have scheduled tasks that run everyday - they will no longer run. Can anyone confirm this?

[rjw57]

As another data point, we were also being bitten by this same error. Reverting to 1.1.1 fixed the problem for us. We only have an hourly periodic task scheduled so I don't think you need much load to trigger this.

[rsichnyi]

so i got to the full testing, the fix seems to have resolved crashes, but since DatabaseError occurs on every sync, the tasks are never updated in DB

[ddahan]

After reading the whole conversation, I'm still not sure to understand what is the state of the resolution of this bug.

• I'm having the same error, but for now, my tasks seems to be triggered anyway... What are the actual consequences of this connection already closed error?
• When you're talking about overriding some methods like sync() I'm not sure what you need to do exactly. Does someone have a code example to show?
• When you're talking about reverting to older version, it seems it's not easy when you're already in production (because of removed features, maybe because of migrations), can someone elaborate about this?
• It seems that the written fix does not work (@rsichny last comment). Right?

In the end, does anyone know where are we now with this major issue?
If no, I guess it could be wise to replace DCB with another solution until it's officially fixed.

Thanks for your answers.

[rsichnyi]

@ddahan the fix prevents scheduler from crashing and tasks are properly scheduled in any case, but unfortunately the sync() will always be skipped and so tasks (i.e. last_run_at and other fields) will never get updated in database. I'll fix it as soon as i'll find time if no one gets to it first.

The general problem is that close_old_connections() is called twice in same iteration here (inside both tick() and _do_sync()):
https://github.com/celery/celery/blob/58e6219c12cd26a3ec92d26d21d0f5ac4cc20abd/celery/beat.py#L593

My suggestion for the fix is either to only call close_old_connections if mysql is used or just remove close_old_connections from sync since the same call was added to schedule_changed(). The second idea will probably still cause the issues to mysql users.