Skip to content

UN-2798 [FIX] Fix WorkflowFileExecution stuck in EXECUTING status due to database connection errors #1524

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
wants to merge 6 commits into
base: main
Choose a base branch
from
Draft

UN-2798 [FIX] Fix WorkflowFileExecution stuck in EXECUTING status due to database connection errors #1524

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
86cff59
UN-2798 [FIX] Fix WorkflowFileExecution stuck in EXECUTING status
muhammad-ali-e Sep 9, 2025
f0979aa
refactor to reduce complexity
muhammad-ali-e Sep 9, 2025
78a8660
reducing complexity of methods
muhammad-ali-e Sep 9, 2025
7c0d03c
reducing complexity of methods
muhammad-ali-e Sep 9, 2025
0eeae21
reducing complexity of methods
muhammad-ali-e Sep 9, 2025
e606f1e
use env values for configurations, rplace print with logger
muhammad-ali-e Sep 10, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions backend/account_v2/organization.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
import logging

from django.db import IntegrityError
from utils.db_retry import db_retry

from account_v2.models import Organization

Expand All @@ -12,13 +13,15 @@ def __init__(self): # type: ignore
pass

@staticmethod
@db_retry() # Add retry for connection drops during organization lookup
def get_organization_by_org_id(org_id: str) -> Organization | None:
try:
return Organization.objects.get(organization_id=org_id) # type: ignore
except Organization.DoesNotExist:
return None

@staticmethod
@db_retry() # Add retry for connection drops during organization creation
def create_organization(
name: str, display_name: str, organization_id: str
) -> Organization:
Expand Down
38 changes: 37 additions & 1 deletion backend/backend/celery_config.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,10 +1,19 @@
import logging
import os
from urllib.parse import quote_plus

from django.conf import settings

from backend.celery_db_retry import get_celery_db_engine_options, should_use_builtin_retry

Comment on lines +7 to +8
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue

Fix import path for retry helpers.

File resides under backend/backend/, so the import should reference backend.backend.celery_db_retry. Current path will fail at runtime.

Apply this diff:

-from backend.celery_db_retry import get_celery_db_engine_options, should_use_builtin_retry
+from backend.backend.celery_db_retry import (
+    get_celery_db_engine_options,
+    should_use_builtin_retry,
+)
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
from backend.celery_db_retry import get_celery_db_engine_options, should_use_builtin_retry
from backend.backend.celery_db_retry import (
get_celery_db_engine_options,
should_use_builtin_retry,
)
🤖 Prompt for AI Agents 
In backend/backend/celery_config.py around lines 6 to 7, the import path for the
retry helpers is incorrect; update the import to reference the actual module
location by changing the import to use backend.backend.celery_db_retry and
import get_celery_db_engine_options and should_use_builtin_retry from that
module so the runtime can resolve the symbols.

logger = logging.getLogger(__name__)


class CeleryConfig:
"""Specifies celery configuration
"""Specifies celery configuration with hybrid retry support.

Supports both custom retry (via patching) and Celery's built-in retry
based on CELERY_USE_BUILTIN_RETRY environment variable.

Refer https://docs.celeryq.dev/en/stable/userguide/configuration.html
"""
Expand All @@ -31,3 +40,30 @@ class CeleryConfig:
beat_scheduler = "django_celery_beat.schedulers:DatabaseScheduler"

task_acks_late = True

# Database backend engine options for PgBouncer compatibility
result_backend_transport_options = get_celery_db_engine_options()

Comment on lines +44 to +46
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

💡 Verification agent

🧩 Analysis chain

Confirm Celery picks up engine options for SQLAlchemy backend.

result_backend_transport_options is correct for Celery’s SQLAlchemy backend; please verify on your Celery version that these are passed to create_engine.

Run:

If your version differs, we may need to adjust the config key or option names.

🏁 Script executed:

#!/bin/bash
# Inspect Celery DatabaseBackend to confirm it uses `result_backend_transport_options`
python - <<'PY'
import inspect
from celery.backends import database
print("DatabaseBackend.create_backend method:\n")
print(inspect.getsource(database.DatabaseBackend.__init__))
PY

Length of output: 118

🏁 Script executed:

#!/bin/bash
# Locate Celery dependency and its pinned version in the repository
grep -R "celery" -n requirements*.txt pyproject.toml setup.py

Length of output: 255

🏁 Script executed:

#!/bin/bash
# Install the pinned Celery version and inspect DatabaseBackend for transport options handling
pip install celery==5.3.4 >/dev/null 2>&1
python - <<'PY'
import inspect
from celery.backends.database import DatabaseBackend

print("Signature of DatabaseBackend.__init__:\n")
print(inspect.signature(DatabaseBackend.__init__), "\n")
print("Source snippet around engine creation in DatabaseBackend:\n")
src = inspect.getsource(DatabaseBackend)
for line in src.splitlines():
    if "create_engine" in line or "transport_options" in line:
        print(line)
PY

Length of output: 437

🏁 Script executed:

#!/bin/bash
# Install Celery and SQLAlchemy to enable DatabaseBackend inspection
pip install celery==5.3.4 sqlalchemy >/dev/null 2>&1
python - <<'PY'
import inspect
from celery.backends.database import DatabaseBackend

print("Signature of DatabaseBackend.__init__:\n", inspect.signature(DatabaseBackend.__init__), "\n")
source = inspect.getsource(DatabaseBackend)
print("Lines containing engine creation or transport options:\n")
for line in source.splitlines():
    if "create_engine" in line or "transport_options" in line:
        print(line)
PY

Length of output: 157

🏁 Script executed:

#!/bin/bash
# Print the source of DatabaseBackend.__init__ to inspect engine_options usage
python - <<'PY'
import inspect
from celery.backends.database import DatabaseBackend
print(inspect.getsource(DatabaseBackend.__init__))
PY

Length of output: 1243

Use database_engine_options instead of result_backend_transport_options in backend/backend/celery_config.py
Celery’s SQLAlchemy result backend reads engine options from conf.database_engine_options (not result_backend_transport_options) and passes them into create_engine via SessionManager.session_factory (docs.celeryq.dev)

🤖 Prompt for AI Agents 
In backend/backend/celery_config.py around lines 41 to 43, the code sets Celery
SQLAlchemy engine options using result_backend_transport_options, but Celery
expects these under database_engine_options; change the assignment to set
conf.database_engine_options (i.e., assign get_celery_db_engine_options() to
database_engine_options) so the SQLAlchemy result backend receives the engine
options passed into create_engine via SessionManager.session_factory.

# Hybrid retry configuration - built-in vs custom
if should_use_builtin_retry():
# Use Celery's built-in database backend retry
result_backend_always_retry = (
os.environ.get("CELERY_RESULT_BACKEND_ALWAYS_RETRY", "true").lower() == "true"
)
result_backend_max_retries = int(
os.environ.get("CELERY_RESULT_BACKEND_MAX_RETRIES", "3")
)
result_backend_base_sleep_between_retries_ms = int(
os.environ.get("CELERY_RESULT_BACKEND_BASE_SLEEP_BETWEEN_RETRIES_MS", "1000")
)
result_backend_max_sleep_between_retries_ms = int(
os.environ.get("CELERY_RESULT_BACKEND_MAX_SLEEP_BETWEEN_RETRIES_MS", "30000")
)

logger.info(
f"[Celery Config] Using built-in retry: max_retries={result_backend_max_retries}, "
f"base_sleep={result_backend_base_sleep_between_retries_ms}ms, max_sleep={result_backend_max_sleep_between_retries_ms}ms"
)
else:
# Custom retry is handled by patch_celery_database_backend()
logger.info("[Celery Config] Using custom retry system (patching enabled)")
Loading