fix(Task processor): Task processor silently fails every task when started with run-task-processor

src/common/core/cli/run.py

@@ -8,6 +8,7 @@
 import shlex
 
 from django.conf import settings
+from django.core.exceptions import ImproperlyConfigured
 from django.core.management import (
     execute_from_command_line as django_execute_from_command_line,
 )
@@ -57,6 +58,11 @@ def serve(argv: list[str], *, prog: str) -> None:
 
 def run_task_processor(argv: list[str], *, prog: str) -> None:
     """Migrate, wait for migrations to be applied, then start the task processor."""
+    if not getattr(settings, "TASK_PROCESSOR_MODE", False):
+        raise ImproperlyConfigured(
+            f"{prog} {argv} is not supported as an entrypoint. "
+            "Please use `flagsmith start task-processor` to start the Task processor."
+        )
     _migrate()
     databases: list[str] = getattr(
         settings, "FLAGSMITH_WAIT_FOR_MIGRATIONS_DATABASES", ["default"]

src/common/core/main.py

@@ -21,6 +21,15 @@
 logger = logging.getLogger(__name__)
 
 
+def _is_task_processor(argv: list[str]) -> bool:
+    """
+    Returns True if the current running process is inferred to be the task processor,
+    i.e. that it was run by either `flagsmith start task-processor` or
+    `flagsmith run-task-processor`.
+    """
+    return not {"task-processor", "run-task-processor"}.isdisjoint(argv)
+
+
 @contextlib.contextmanager
 def ensure_cli_env() -> typing.Generator[None, None, None]:
     """
@@ -50,7 +59,9 @@ def ensure_cli_env() -> typing.Generator[None, None, None]:
     if "docgen" in sys.argv:
         os.environ["DOCGEN_MODE"] = "true"
 
-    if "task-processor" in sys.argv:
+    task_processor_mode = _is_task_processor(sys.argv)
+
+    if task_processor_mode:
         # A hacky way to signal we're not running the API
         os.environ["RUN_BY_PROCESSOR"] = "true"
 
@@ -70,9 +81,7 @@ def ensure_cli_env() -> typing.Generator[None, None, None]:
         )
 
         default_service_name = (
-            "flagsmith-task-processor"
-            if "task-processor" in sys.argv
-            else "flagsmith-api"
+            "flagsmith-task-processor" if task_processor_mode else "flagsmith-api"
         )
         service_name = env.str("OTEL_SERVICE_NAME", default_service_name)
         otel_protocol = typing.cast(

tests/unit/common/core/test_main.py

@@ -163,25 +163,51 @@ def test_ensure_cli_env__docgen_in_argv__sets_docgen_mode(
         assert os.environ.get("DOCGEN_MODE") == "true"
 
 
+@pytest.mark.parametrize(
+    "argv",
+    [
+        pytest.param(["flagsmith", "start", "task-processor"], id="start_verb"),
+        pytest.param(["flagsmith", "run-task-processor"], id="composite_verb"),
+    ],
+)
 def test_ensure_cli_env__task_processor_in_argv__sets_run_by_processor(
     monkeypatch: pytest.MonkeyPatch,
+    argv: list[str],
 ) -> None:
     # Given
-    monkeypatch.setattr("sys.argv", ["flagsmith", "task-processor"])
+    monkeypatch.delenv("RUN_BY_PROCESSOR", raising=False)
+    monkeypatch.setattr("sys.argv", argv)
 
     # When / Then
     with ensure_cli_env():
         assert os.environ.get("RUN_BY_PROCESSOR") == "true"
 
 
+def test_ensure_cli_env__api_argv__does_not_set_run_by_processor(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    # Given
+    monkeypatch.delenv("RUN_BY_PROCESSOR", raising=False)
+    monkeypatch.setattr("sys.argv", ["flagsmith", "migrate-and-serve"])
+
+    # When / Then
+    with ensure_cli_env():
+        assert os.environ.get("RUN_BY_PROCESSOR") is None
+
+
 @pytest.mark.parametrize(
     "argv,expected_otel_service_name",
     [
         pytest.param(
-            ["flagsmith", "task-processor"],
+            ["flagsmith", "start", "task-processor"],
             "flagsmith-task-processor",
             id="task_processor",
         ),
+        pytest.param(
+            ["flagsmith", "run-task-processor"],
+            "flagsmith-task-processor",
+            id="task_processor_composite_verb",
+        ),
         pytest.param(
             ["flagsmith", "anything-else"],
             "flagsmith-api",

tests/unit/common/core/test_run.py

@@ -1,6 +1,7 @@
 from unittest.mock import MagicMock
 
 import pytest
+from django.core.exceptions import ImproperlyConfigured
 from django.test import override_settings
 from pytest_mock import MockerFixture
 
@@ -102,6 +103,7 @@ def test_migrate__with_args__defers_to_django_migrate(
     ]
 
 
+@override_settings(TASK_PROCESSOR_MODE=True)
 def test_run_task_processor__default__migrates_waits_then_starts(
     mock_run: MagicMock,
 ) -> None:
@@ -127,6 +129,7 @@ def test_run_task_processor__default__migrates_waits_then_starts(
 
 
 @override_settings(
+    TASK_PROCESSOR_MODE=True,
     FLAGSMITH_MIGRATE_DATABASES=["default", "analytics"],
     FLAGSMITH_WAIT_FOR_MIGRATIONS_DATABASES=["default", "analytics"],
 )
@@ -164,6 +167,20 @@ def test_run_task_processor__configured_databases__migrates_and_waits_for_each(
     ]
 
 
+@override_settings(TASK_PROCESSOR_MODE=False)
+def test_run_task_processor__task_processor_mode_unset__refuses_to_start(
+    mock_run: MagicMock,
+) -> None:
+    # Given / When
+    with pytest.raises(ImproperlyConfigured) as exc_info:
+        run.run_task_processor([], prog="flagsmith run-task-processor")
+
+    # Then — fails before any work, rather than claiming tasks it cannot run,
+    # and points at an entrypoint that does set the mode
+    assert "flagsmith start task-processor" in str(exc_info.value)
+    assert mock_run.call_args_list == []
+
+
 def test_migrate_and_serve__no_startup_commands__migrates_then_serves(
     mock_run: MagicMock,
 ) -> None: