Allow more configuration of async Task behaviour

api/app/settings/common.py

@@ -23,6 +23,8 @@
 from django.core.management.utils import get_random_secret_key
 from environs import Env
 
+from task_processor.task_run_method import TaskRunMethod
+
 env = Env()
 logger = logging.getLogger(__name__)
 
@@ -676,4 +678,7 @@
 MAX_SELF_MIGRATABLE_IDENTITIES = env.int("MAX_SELF_MIGRATABLE_IDENTITIES", 100000)
 
 # Setting to allow asynchronous tasks to be run synchronously for testing purposes
-RUN_TASKS_SYNCHRONOUSLY = env.bool("RUN_TASKS_SYNCHRONOUSLY", True)
+# or in a separate thread for self-hosted users
+TASK_RUN_METHOD = env.enum(
+    "TASK_RUN_METHOD", type=TaskRunMethod, default=TaskRunMethod.SEPARATE_THREAD.value
+)

api/task_processor/apps.py

@@ -2,14 +2,16 @@
 from django.conf import settings
 from health_check.plugins import plugin_dir
 
+from task_processor.task_run_method import TaskRunMethod
+
 
 class TaskProcessorAppConfig(AppConfig):
     name = "task_processor"
 
     def ready(self):
         from . import tasks  # noqa
 
-        if not settings.RUN_TASKS_SYNCHRONOUSLY:
+        if settings.TASK_RUN_METHOD == TaskRunMethod.TASK_PROCESSOR:
             from .health import TaskProcessorHealthCheckBackend
 
             plugin_dir.register(TaskProcessorHealthCheckBackend)

api/task_processor/decorators.py

@@ -7,6 +7,7 @@
 
 from task_processor.models import Task
 from task_processor.task_registry import register_task
+from task_processor.task_run_method import TaskRunMethod
 
 logger = logging.getLogger(__name__)
 
@@ -22,16 +23,18 @@ def decorator(f: typing.Callable):
         register_task(task_identifier, f)
 
         def delay(*args, **kwargs) -> typing.Optional[Task]:
-            if settings.RUN_TASKS_SYNCHRONOUSLY:
+            if settings.TASK_RUN_METHOD == TaskRunMethod.SYNCHRONOUSLY:
                 logger.debug("Running task '%s' synchronously", task_identifier)
                 f(*args, **kwargs)
+            elif settings.TASK_RUN_METHOD == TaskRunMethod.SEPARATE_THREAD:
+                logger.debug("Running task '%s' in separate thread", task_identifier)
+                run_in_thread(*args, **kwargs)
             else:
                 logger.debug("Creating task for function '%s'...", task_identifier)
                 task = Task.create(task_identifier, *args, **kwargs)
                 task.save()
                 return task
 
-        # TODO: remove this functionality and use delay in all scenarios
         def run_in_thread(*args, **kwargs):
             logger.info("Running function %s in unmanaged thread.", f.__name__)
             Thread(target=f, args=args, kwargs=kwargs, daemon=True).start()

api/task_processor/task_run_method.py

@@ -0,0 +1,7 @@
+from enum import Enum
+
+
+class TaskRunMethod(Enum):
+    SYNCHRONOUSLY = "SYNCHRONOUSLY"
+    SEPARATE_THREAD = "SEPARATE_THREAD"
+    TASK_PROCESSOR = "TASK_PROCESSOR"

api/tests/unit/task_processor/test_unit_task_processor_health.py

@@ -1,5 +1,6 @@
 from task_processor.health import is_processor_healthy
 from task_processor.models import HealthCheckModel
+from task_processor.task_run_method import TaskRunMethod
 
 
 def test_is_processor_healthy_returns_false_if_task_not_processed(mocker):
@@ -21,7 +22,7 @@ def test_is_processor_healthy_returns_false_if_task_not_processed(mocker):
 
 def test_is_processor_healthy_returns_true_if_task_processed(db, settings):
     # Given
-    settings.RUN_TASKS_SYNCHRONOUSLY = True
+    settings.TASK_RUN_METHOD = TaskRunMethod.SYNCHRONOUSLY
 
     # When
     result = is_processor_healthy(max_tries=3)

docker-compose.yml

@@ -47,6 +47,11 @@ services:
             # ENABLE_ADMIN_ACCESS_USER_PASS: True # set to True to enable access to the /admin/ Django backend via your username and password
             # ALLOW_REGISTRATION_WITHOUT_INVITE: True
 
+            # Enable Task Processor
+            # To use task processor service, uncomment line below and additional 'flagsmith_processor'
+            # container below
+            # TASK_RUN_METHOD: TASK_PROCESSOR  # other options are: SYNCHRONOUSLY, SEPARATE_THREAD (default)
+
         ports:
             - '8000:8000'
         depends_on:
@@ -56,19 +61,26 @@ services:
             - postgres
             # - influxdb:influxdb
 
-    flagsmith_processor:
-        build:
-            dockerfile: api/Dockerfile
-            context: .
-        environment:
-            DATABASE_URL: postgresql://postgres:password@postgres:5432/flagsmith
-        command:
-            - "python"
-            - "manage.py"
-            - "runprocessor"
-        depends_on:
-            - flagsmith
-            - postgres
+#    Run the asynchronous task processor as a separate container alongside the API.
+#    When enabled, the API will write tasks to a queue (in the PostgreSQL database) for
+#    the processor to consume asynchronously.
+#    Documentation on the processor can be found here:
+#    https://docs.flagsmith.com/advanced-use/task-processor
+#    flagsmith_processor:
+#        build:
+#            dockerfile: api/Dockerfile
+#            context: .
+#        environment:
+#            DATABASE_URL: postgresql://postgres:password@postgres:5432/flagsmith
+#        command:
+#            - "python"
+#            - "manage.py"
+#            - "runprocessor"
+#            - "--sleepintervalms"
+#            - "500"
+#        depends_on:
+#            - flagsmith
+#            - postgres
 
     # InfluxDB requires additional setup - please see https://docs.flagsmith.com/deployment-overview/#influxdb
     # Note that InfluxDB is optional, but enabling it will provide additional functionality to the Flagsmith platform

infrastructure/aws/staging/ecs-task-definition-web.json

@@ -152,8 +152,8 @@
                     "value": "60"
                 },
                 {
-                    "name": "RUN_TASKS_SYNCHRONOUSLY",
-                    "value": "False"
+                    "name": "TASK_RUN_METHOD",
+                    "value": "TASK_PROCESSOR"
                 },
                 {
                     "name": "CACHE_ENVIRONMENT_DOCUMENT_SECONDS",