diff --git a/celery/backends/database/session.py b/celery/backends/database/session.py index 047a9271d92..a3d5dfb15c5 100644 --- a/celery/backends/database/session.py +++ b/celery/backends/database/session.py @@ -1,6 +1,9 @@ """SQLAlchemy session.""" +import time + from kombu.utils.compat import register_after_fork from sqlalchemy import create_engine +from sqlalchemy.exc import DatabaseError from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker from sqlalchemy.pool import NullPool @@ -50,7 +53,26 @@ def create_session(self, dburi, short_lived_sessions=False, **kwargs): def prepare_models(self, engine): if not self.prepared: - ResultModelBase.metadata.create_all(engine) + # SQLAlchemy will check if the items exist before trying to + # create them, which is a race condition. If it raises an error + # in one iteration, the next may pass all the existence checks + # and the call will succeed. + max_retries = 10 + retries = 0 + while True: + try: + ResultModelBase.metadata.create_all(engine) + except DatabaseError: + if retries < max_retries: + sleep_amount_ms = get_exponential_backoff_interval( + 10, retries, 1000, True + ) + time.sleep(sleep_amount_ms / 1000) + retries += 1 + else: + raise + else: + break self.prepared = True def session_factory(self, dburi, **kwargs):