Resonate setup

README.md

@@ -51,6 +51,26 @@ To enable the Django Debug Toolbar, set the `ENABLE_DJANGO_DEBUG_TOOLBAR` variab
 ENABLE_DJANGO_DEBUG_TOOLBAR=true
 ```
 
+## Durable workflows (Resonate)
+
+Long, multi-step operations run as [Resonate](https://www.resonatehq.io/)
+durable workflows, executed by the `resonate-worker` service (`docker-compose
+up` starts it, along with the Resonate server and its UI at
+http://localhost:8005).
+
+Workflows live in `<app>/workflows.py` (or `<app>/workflows/`); they are
+registered against the instance returned by `pycon.resonate_app.get_resonate()`
+and discovered automatically by the worker. Synchronous, ORM-using steps are
+wrapped with `pycon.resonate_app.database_step`.
+
+The worker restarts itself when the code changes, like `runserver` does, so
+editing a workflow is enough to see it run in its new shape. Pass `--no-reload`
+to turn that off; outside `DEBUG` it is off to begin with.
+
+To start a workflow from Django (a view, an admin action, a command), use
+`pycon.resonate_app.start_workflow(name, workflow_id, ...)`. The workflow id is
+the idempotency key: creating the same id twice joins the existing run.
+
 ## External repos
 
 Repos used by this project are in separate repositories.

backend/pycon/management/commands/resonate_worker.py

@@ -0,0 +1,94 @@
+"""Run the Resonate worker: the process that executes durable workflows."""
+
+import asyncio
+import atexit
+import signal
+import threading
+
+from django.conf import settings
+from django.core.management.base import BaseCommand
+from django.utils import autoreload
+
+from pycon.resonate_app import autodiscover_workflows, get_resonate, require_server
+
+# How long the worker is given to release its tasks when the autoreloader is
+# about to replace the process.
+SHUTDOWN_TIMEOUT_SECONDS = 5
+
+
+class Command(BaseCommand):
+    help = "Run the Resonate worker, executing durable workflows"
+
+    def add_arguments(self, parser):
+        parser.add_argument(
+            "--reload",
+            action="store_true",
+            dest="use_reloader",
+            default=settings.DEBUG,
+            help="Restart the worker when the code changes (the default with DEBUG)",
+        )
+        parser.add_argument(
+            "--no-reload",
+            action="store_false",
+            dest="use_reloader",
+            help="Do not restart the worker when the code changes",
+        )
+
+    def handle(self, *args, **options):
+        # Fail before the reloader spawns a child that could only fail there,
+        # where the error is a lot less visible.
+        require_server()
+
+        if options["use_reloader"]:
+            autoreload.run_with_reloader(self.run_worker)
+        else:
+            self.run_worker()
+
+    def run_worker(self):
+        asyncio.run(self._run())
+
+    async def _run(self):
+        autodiscover_workflows()
+
+        resonate = get_resonate()
+        resonate.start()
+
+        self.stdout.write(self.style.SUCCESS("Resonate worker started"))
+
+        stop = asyncio.Event()
+        stopped = threading.Event()
+        self._stop_on_shutdown(stop, stopped)
+
+        try:
+            await stop.wait()
+        finally:
+            self.stdout.write("Stopping Resonate worker")
+            await resonate.stop()
+            stopped.set()
+
+    def _stop_on_shutdown(self, stop: asyncio.Event, stopped: threading.Event):
+        """Ask the worker to stop when the process is going away.
+
+        Run directly, the worker owns the main thread and can take the signals
+        itself. Under the autoreloader it instead runs in a daemon thread,
+        while the reloader exits the process from the main thread on a code
+        change -- so the worker stops from an exit hook, which still runs while
+        the daemon thread is alive, rather than being killed mid-task.
+        """
+        loop = asyncio.get_running_loop()
+
+        if threading.current_thread() is threading.main_thread():
+            for sig in (signal.SIGINT, signal.SIGTERM):
+                loop.add_signal_handler(sig, stop.set)
+            return
+
+        def on_exit():
+            try:
+                loop.call_soon_threadsafe(stop.set)
+            except RuntimeError:
+                # The loop is already gone; nothing left to stop.
+                return
+
+            stopped.wait(timeout=SHUTDOWN_TIMEOUT_SECONDS)
+
+        atexit.register(on_exit)

backend/pycon/resonate_app.py

@@ -0,0 +1,138 @@
+"""Resonate (durable workflows) wiring for the backend.
+
+Two roles share one module:
+
+- **The worker** (``manage.py resonate_worker``) builds the singleton returned
+  by :func:`get_resonate`, which every workflow module registers against, and
+  starts it so the Resonate server can push work to it.
+- **The web/admin tier** never starts that singleton. It triggers work with
+  :func:`start_workflow`, which opens a short-lived, send-only client, creates
+  the durable promise and returns its id.
+
+Workflow modules live in ``<app>/workflows.py`` (or ``<app>/workflows/``) and
+are imported by :func:`autodiscover_workflows`, mirroring how Celery tasks are
+discovered.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import functools
+from typing import Any, Callable, TypeVar
+
+from asgiref.sync import sync_to_async
+from django.conf import settings
+from django.core.exceptions import ImproperlyConfigured
+from django.db import connections
+from django.utils.module_loading import autodiscover_modules
+from resonate.resonate import Resonate
+
+_resonate: Resonate | None = None
+
+T = TypeVar("T")
+
+
+def _build(*, send_only: bool) -> Resonate:
+    kwargs: dict[str, Any] = {
+        "group": settings.RESONATE_GROUP,
+        # Nothing here needs an event loop: connections are opened by
+        # ``start()``, so building at import time is safe.
+        "autostart": False,
+        # The SDK falls back to reading RESONATE_URL (and friends) from the
+        # environment; hiding it keeps Django settings the only source.
+        "env": {},
+    }
+
+    if settings.RESONATE_URL:
+        kwargs["url"] = settings.RESONATE_URL
+
+    if send_only:
+        # No source: this process only creates promises, it never picks work up.
+        kwargs["sources"] = []
+
+    return Resonate(**kwargs)
+
+
+def get_resonate() -> Resonate:
+    """The process-wide Resonate instance workflows register against."""
+    global _resonate
+
+    if _resonate is None:
+        _resonate = _build(send_only=False)
+
+    return _resonate
+
+
+def require_server() -> None:
+    """Refuse to run workflows without a Resonate server to run them on.
+
+    Without a URL the SDK falls back to its in-process connection, where a
+    workflow would be created and then never executed by anything.
+    """
+    if not settings.RESONATE_URL:
+        raise ImproperlyConfigured(
+            "RESONATE_URL is not set, so there is no Resonate server to run "
+            "workflows on"
+        )
+
+
+def autodiscover_workflows() -> None:
+    """Import every ``workflows`` module so its registrations happen."""
+    autodiscover_modules("workflows")
+
+
+def start_workflow(name: str, workflow_id: str, *args: Any, **kwargs: Any) -> str:
+    """Create a durable promise for ``name`` and return its id.
+
+    Fire-and-forget from synchronous Django code: the workflow itself runs on a
+    worker subscribed to ``settings.RESONATE_GROUP``. ``workflow_id`` is the
+    idempotency key -- creating the same id twice joins the existing run
+    instead of starting a second one.
+    """
+
+    require_server()
+
+    async def _dispatch() -> str:
+        client = _build(send_only=True)
+        client.start()
+
+        try:
+            handle = client.rpc(workflow_id, name, *args, **kwargs)
+            return await handle.id()
+        finally:
+            await client.stop()
+
+    return asyncio.run(_dispatch())
+
+
+def database_step(fn: Callable[..., T]) -> Callable[..., Any]:
+    """Adapt a synchronous, ORM-using step so it can run as a durable step.
+
+    The SDK executes durable functions on an event loop, where Django's ORM
+    refuses to run (``SynchronousOnlyOperation``). The body is therefore
+    handed to asgiref's thread-sensitive executor -- a single thread, so every
+    step reuses one database connection -- with stale connections recycled
+    first, the way a request or a Celery task would.
+    """
+
+    @functools.wraps(fn)
+    async def wrapper(*args: Any, **kwargs: Any) -> T:
+        def run() -> T:
+            _recycle_connections()
+            return fn(*args, **kwargs)
+
+        return await sync_to_async(run, thread_sensitive=True)()
+
+    return wrapper
+
+
+def _recycle_connections() -> None:
+    """``close_old_connections``, minus any connection we do not own.
+
+    A connection inside an atomic block belongs to whoever opened the
+    transaction (in tests, the one wrapping each test), and closing it there
+    breaks the caller.
+    """
+    for connection in connections.all(initialized_only=True):
+        if not connection.in_atomic_block:
+            connection.close_if_unusable_or_obsolete()

backend/pycon/settings/base.py

@@ -127,6 +127,9 @@
     "privacy_policy.apps.PrivacyPolicyConfig",
     "visa.apps.VisaConfig",
     "generic_forms.apps.GenericFormsConfig",
+    # Project package, registered as an app so its project-level management
+    # commands (e.g. resonate_worker) are discoverable. It has no models.
+    "pycon",
 ]
 
 MIDDLEWARE = [
@@ -402,6 +405,11 @@
 
 CELERY_TASK_IGNORE_RESULT = True
 
+# Resonate (durable workflows). Empty URL keeps the SDK on its in-process
+# local connection, which is what tests use.
+RESONATE_URL = env("RESONATE_URL", default="")
+RESONATE_GROUP = env("RESONATE_GROUP", default="pycon")
+
 AWS_STORAGE_BUCKET_NAME = env("AWS_MEDIA_BUCKET", default=None)
 AWS_REGION_NAME = AWS_SES_REGION_NAME = AWS_S3_REGION_NAME = env(
     "AWS_REGION_NAME", default="eu-central-1"

backend/pycon/tests/test_resonate_worker.py

@@ -0,0 +1,61 @@
+from django.core.exceptions import ImproperlyConfigured
+from django.core.management import call_command
+from pytest import raises
+
+from pycon.management.commands.resonate_worker import Command
+
+
+def test_worker_reloads_on_code_changes_by_default(mocker, settings):
+    settings.DEBUG = True
+    settings.RESONATE_URL = "http://resonate:8001"
+
+    mock_run_with_reloader = mocker.patch(
+        "pycon.management.commands.resonate_worker.autoreload.run_with_reloader"
+    )
+    mock_run_worker = mocker.patch.object(Command, "run_worker")
+
+    call_command("resonate_worker")
+
+    mock_run_with_reloader.assert_called_once()
+    mock_run_worker.assert_not_called()
+
+
+def test_worker_does_not_reload_outside_debug(mocker, settings):
+    settings.DEBUG = False
+    settings.RESONATE_URL = "http://resonate:8001"
+
+    mock_run_with_reloader = mocker.patch(
+        "pycon.management.commands.resonate_worker.autoreload.run_with_reloader"
+    )
+    mock_run_worker = mocker.patch.object(Command, "run_worker")
+
+    call_command("resonate_worker")
+
+    mock_run_worker.assert_called_once()
+    mock_run_with_reloader.assert_not_called()
+
+
+def test_worker_reload_can_be_turned_off(mocker, settings):
+    settings.DEBUG = True
+    settings.RESONATE_URL = "http://resonate:8001"
+
+    mock_run_with_reloader = mocker.patch(
+        "pycon.management.commands.resonate_worker.autoreload.run_with_reloader"
+    )
+    mock_run_worker = mocker.patch.object(Command, "run_worker")
+
+    call_command("resonate_worker", "--no-reload")
+
+    mock_run_worker.assert_called_once()
+    mock_run_with_reloader.assert_not_called()
+
+
+def test_worker_needs_a_resonate_server(mocker, settings):
+    settings.RESONATE_URL = ""
+
+    mock_run_worker = mocker.patch.object(Command, "run_worker")
+
+    with raises(ImproperlyConfigured):
+        call_command("resonate_worker", "--no-reload")
+
+    mock_run_worker.assert_not_called()

backend/pyproject.toml

@@ -112,6 +112,7 @@ dependencies = [
     "scikit-learn>=1.5.0",
     "bertopic>=0.16.0",
     "nltk>=3.9.4",
+    "resonate-sdk>=0.8.1",
 ]
 name = "backend"
 version = "0.1.0"