Convert a few low risk tasks to use procrastinate

.devcontainer/docker-compose.devcontainer.yml

@@ -12,3 +12,7 @@ services:
   celerybeat:
     # Celery beat will be started via launch.json or the terminal.
     profiles: ["celerybeat"]
+
+  procrastinate:
+    # The procrastinate worker will be started via launch.json or the terminal.
+    profiles: ["procrastinate"]

.greenmask/config.yml

@@ -60,6 +60,10 @@ dump:
       - "oauth2_provider_refreshtoken"
       - "socialaccount_socialtoken"
       - "socialaccount_socialaccount"
+      - "procrastinate_events"
+      - "procrastinate_jobs"
+      - "procrastinate_periodic_defers"
+      - "procrastinate_workers"
 
 
   transformation:

.vscode/launch.json

@@ -87,6 +87,20 @@
         "order": 3
       }
     },
+    {
+      "name": "Procrastinate: Worker",
+      "type": "debugpy",
+      "request": "launch",
+      "program": "${workspaceFolder}/manage.py",
+      "args": ["procrastinate", "worker"],
+      "django": true,
+      "console": "integratedTerminal",
+      "justMyCode": false,
+      "presentation": {
+        "group": "2-services",
+        "order": 4
+      }
+    },
     {
       "name": "Pytest: Debug",
       "type": "debugpy",
@@ -108,7 +122,7 @@
       "console": "integratedTerminal",
       "presentation": {
         "group": "2-services",
-        "order": 4
+        "order": 5
       }
     }
   ],

Procfile

@@ -10,3 +10,4 @@ web: gunicorn --bind 0.0.0.0:$PORT --graceful-timeout 120 --timeout 125 --limit-
 worker: REMAP_SIGTERM=SIGQUIT ./deploy/worker.sh
 low_priority_worker: REMAP_SIGTERM=SIGQUIT ./deploy/low-priority-worker.sh
 beat: REMAP_SIGTERM=SIGQUIT celery --app isic.celery beat --loglevel INFO
+procrastinate: ./deploy/procrastinate-worker.sh

deploy/procrastinate-worker.sh

@@ -0,0 +1,18 @@
+#!/bin/bash
+set -e
+
+# Procrastinate has no per-task timeout (no soft_time_limit/time_limit equivalent).
+# This is a coarse backstop bounding any single statement issued by this process.
+# libpq reads PGOPTIONS, and neither Django's connection params nor procrastinate's
+# worker pool set "options", so both inherit it.
+export PGOPTIONS="-c statement_timeout=30min"
+
+# Fail at boot on a misconfigured worker rather than idling silently.
+python ./manage.py procrastinate healthchecks
+
+# Heroku sends SIGTERM then SIGKILLs 30 seconds later, so the graceful window has
+# to fit inside that or it never actually runs.
+python ./manage.py procrastinate worker \
+    --queues default,stats-aggregation \
+    --concurrency 2 \
+    --shutdown-graceful-timeout 25

docker-compose.override.yml

@@ -82,6 +82,31 @@ services:
       rabbitmq:
         condition: service_healthy
 
+  procrastinate:
+    build:
+      context: .
+      dockerfile: ./dev/django.Dockerfile
+    command: [
+      "uv", "run",
+      "./manage.py", "procrastinate", "worker"
+    ]
+    # uv progress doesn't display properly with a Docker TTY
+    tty: false
+    working_dir: /home/vscode/isic
+    env_file: ./dev/.env.docker-compose
+    volumes:
+      - .:/home/vscode/isic
+      - pkg_cache:/home/vscode/pkg-cache
+    depends_on:
+      postgres:
+        condition: service_healthy
+      elasticsearch:
+        condition: service_healthy
+      minio:
+        condition: service_healthy
+      redis:
+        condition: service_healthy
+
   zipstreamer:
     image: ghcr.io/imagemarkup/isic-zipstreamer:master
     environment:

isic/conftest.py

@@ -7,8 +7,10 @@
 from django.core.cache import cache
 from django.core.files.base import ContentFile
 from django.core.files.storage import default_storage
+from django.db import connection
 from django.test.client import Client
 from playwright.sync_api import expect
+from procrastinate.contrib.django import DjangoApp, procrastinate_app
 import pytest
 from pytest_factoryboy import register
 
@@ -86,6 +88,45 @@ def _clear_cache():
     cache.clear()
 
 
+@pytest.fixture
+def run_procrastinate_jobs():
+    """
+    Run every queued procrastinate job to completion.
+
+    Procrastinate has no equivalent of CELERY_TASK_ALWAYS_EAGER, so a test that
+    triggers a deferral has to run a worker to observe what the job did. Needs
+    django_db(transaction=True), since the worker reads jobs the test committed.
+    """
+
+    def _run():
+        app = procrastinate_app.current_app
+        # A throwaway app, so the worker doesn't also schedule every periodic task
+        # into this test's database. It gets the worker connector rather than the
+        # default Django one, which hands back jsonb as a string and so can't
+        # decode job arguments.
+        worker_app = DjangoApp(connector=app.connector.get_worker_connector())
+        worker_app.tasks.update(app.tasks)
+        worker_app.run_worker(wait=False, install_signal_handlers=False, listen_notify=False)
+
+    return _run
+
+
+@pytest.fixture(autouse=True)
+def _clear_procrastinate_jobs(request):
+    # Procrastinate's models are managed=False, so django_table_names() skips them
+    # and TransactionTestCase never truncates them. Without this, a job deferred by
+    # a transaction=True test survives into the next one.
+    yield
+
+    if "django_db_setup" in request.fixturenames:
+        with connection.cursor() as cursor:
+            cursor.execute(
+                "TRUNCATE procrastinate_jobs, procrastinate_events,"
+                " procrastinate_periodic_defers, procrastinate_workers"
+                " RESTART IDENTITY CASCADE"
+            )
+
+
 @pytest.fixture
 def authenticated_client(user):
     # Do not use the client fixture, to prevent mutating its state

isic/core/tasks.py

@@ -14,10 +14,13 @@
 from django.core.files.storage import default_storage, storages
 from django.core.mail import send_mail
 from django.db import connection, transaction
-from django.db.models import Prefetch
+from django.db.models import Prefetch, Q
 from django.template.loader import render_to_string
 from oauth2_provider.models import clear_expired as clear_expired_oauth_tokens
+from procrastinate.contrib.django import app as procrastinate_app
+from procrastinate.contrib.django.models import ProcrastinateJob
 from resonant_utils.storages import expiring_url
+import sentry_sdk
 from urllib3.exceptions import ConnectionError as Urllib3ConnectionError
 from urllib3.exceptions import TimeoutError as Urllib3TimeoutError
 
@@ -42,6 +45,11 @@
 
 DoiType = Literal["Doi", "DraftDoi"]
 
+MATERIALIZED_VIEW_REFRESH_EXPIRY = timedelta(seconds=60)
+
+# matches the worker's own stalled_worker_timeout default
+STALLED_WORKER_TIMEOUT = timedelta(seconds=30)
+
 
 @shared_task(soft_time_limit=600, time_limit=610)
 def populate_collection_from_search_task(
@@ -139,7 +147,7 @@ def sync_elasticsearch_indices_task():
         cache.delete_pattern("es:*")
 
 
-@shared_task(soft_time_limit=1800, time_limit=1810)
+@procrastinate_app.task()
 def generate_staff_image_list_metadata_csv_task(user_id: int) -> None:
     user = User.objects.get(pk=user_id, is_staff=True)
 
@@ -256,13 +264,39 @@ def generate_archive_snapshot_task() -> None:
         Path(metadata_filename).unlink()
 
 
-@shared_task(soft_time_limit=10, time_limit=15)
-def prune_expired_oauth_tokens_task():
+@procrastinate_app.periodic(cron="*/10 * * * *")
+@procrastinate_app.task(queueing_lock="report_stalled_jobs")
+@sentry_sdk.monitor(monitor_slug="report-stalled-jobs")
+def report_stalled_jobs_task(timestamp: int = 0):
+    stalled = ProcrastinateJob.objects.filter(status="doing").filter(
+        Q(worker__isnull=True)
+        | Q(worker__last_heartbeat__lt=datetime.now(tz=UTC) - STALLED_WORKER_TIMEOUT)
+    )
+
+    for job_id, task_name in stalled.values_list("id", "task_name"):
+        logger.error("Procrastinate job %s is stalled, task %s", job_id, task_name)
+
+
+@procrastinate_app.periodic(cron="0 0 * * *")
+@procrastinate_app.task(queueing_lock="prune_expired_oauth_tokens")
+@sentry_sdk.monitor(monitor_slug="prune-expired-oauth-tokens")
+def prune_expired_oauth_tokens_task(timestamp: int = 0):
     clear_expired_oauth_tokens()
 
 
-@shared_task(soft_time_limit=90, time_limit=120)
-def refresh_materialized_view_collection_counts_task():
+# queueing_lock keeps at most one refresh waiting to run, replacing half of celery's
+# expires=60. The other half is the staleness check below: refreshing is expensive,
+# and a job scheduled for a period that has already passed has nothing to add over
+# the next one.
+@procrastinate_app.periodic(cron="*/15 * * * *")
+@procrastinate_app.task(queueing_lock="refresh_materialized_view_collection_counts")
+@sentry_sdk.monitor(monitor_slug="refresh-materialized-view-collection-counts")
+def refresh_materialized_view_collection_counts_task(timestamp: int = 0):
+    scheduled_at = datetime.fromtimestamp(timestamp, tz=UTC) if timestamp else None
+    if scheduled_at and datetime.now(tz=UTC) - scheduled_at > MATERIALIZED_VIEW_REFRESH_EXPIRY:
+        logger.info("Skipping collection counts refresh scheduled for %s", scheduled_at)
+        return
+
     with connection.cursor() as cursor:
         cursor.execute("REFRESH MATERIALIZED VIEW CONCURRENTLY materialized_collection_counts;")
 

isic/core/tests/test_view_image_list.py

@@ -7,7 +7,9 @@
 
 # needs a real transaction due to setting the isolation level
 @pytest.mark.django_db(transaction=True)
-def test_image_list_metadata_download_view(mocker, staff_client, mailoutbox, user, image: Image):
+def test_image_list_metadata_download_view(
+    mocker, staff_client, mailoutbox, user, image: Image, run_procrastinate_jobs
+):
     image.accession.update_metadata(
         user,
         {
@@ -26,6 +28,8 @@ def test_image_list_metadata_download_view(mocker, staff_client, mailoutbox, use
     r = staff_client.get(reverse("core/image-list-metadata-download"), follow=True)
     assert r.status_code == 200
 
+    run_procrastinate_jobs()
+
     assert len(mailoutbox) == 1
     assert spy.call_count == 1
     storage, key, _ = spy.call_args[0]

isic/core/views/images.py

@@ -216,7 +216,7 @@ def staff_image_list_export(request: AuthenticatedHttpRequest) -> HttpResponse:
 
 @staff_member_required
 def staff_image_list_metadata_download(request: AuthenticatedHttpRequest):
-    generate_staff_image_list_metadata_csv_task.delay_on_commit(request.user.id)
+    generate_staff_image_list_metadata_csv_task.defer(user_id=request.user.id)
 
     messages.add_message(
         request,

isic/settings/base.py

@@ -5,6 +5,7 @@
 from pathlib import Path
 from typing import TYPE_CHECKING
 
+from cachalot.settings import Settings as CachalotSettings
 from celery.schedules import crontab
 import django_stubs_ext
 from environ import Env
@@ -68,6 +69,7 @@
     # Install "ninja" to force Swagger to be served locally, so it can be overridden
     "ninja",
     "oauth2_provider",
+    "procrastinate.contrib.django",
     "resonant_utils",
     "s3_file_field",
     "template_partials",
@@ -178,33 +180,19 @@
 CORS_ALLOW_ALL_ORIGINS = True
 
 CELERY_BEAT_SCHEDULE = {
-    "collect-google-analytics-stats": {
-        "task": "isic.stats.tasks.collect_google_analytics_metrics_task",
-        "schedule": timedelta(hours=6),
-    },
     "sync-elasticsearch-index": {
         "task": "isic.core.tasks.sync_elasticsearch_indices_task",
         "schedule": crontab(minute="0", hour="0"),
     },
-    "prune-expired-oauth-tokens": {
-        "task": "isic.core.tasks.prune_expired_oauth_tokens_task",
-        "schedule": crontab(minute="0", hour="0"),
-    },
-    "refresh-materialized-view-collection-counts": {
-        "task": "isic.core.tasks.refresh_materialized_view_collection_counts_task",
-        "schedule": crontab(minute="*/15", hour="*"),
-        "options": {
-            # to avoid overcomputing, the message should expire 60 seconds after created
-            "expires": timedelta(seconds=60).total_seconds(),
-        },
-    },
     "run-health-checks": {
         "task": "isic.core.tasks.run_health_checks_task",
         "schedule": crontab(minute="0", hour="0"),
     },
 }
 CELERY_WORKER_MAX_MEMORY_PER_CHILD = 256 * 1024
 
+PROCRASTINATE_WORKER_DEFAULTS = {"delete_jobs": "never"}
+
 CACHES = {
     # use django-redis instead of the builtin backend. the builtin redis backend
     # doesn't support deleting keys by prefix, which is important for invalidating
@@ -217,6 +205,16 @@
 # This seems like an essential setting for correctness, see
 # https://github.com/noripyt/django-cachalot/issues/266
 CACHALOT_FINAL_SQL_CHECK = True
+# Procrastinate writes to its tables with raw cursor SQL, which never reaches
+# cachalot's SQLCompiler hooks. Without this, an ORM read of ProcrastinateJob
+# would be cached and never invalidated.
+CACHALOT_UNCACHABLE_TABLES = [
+    *CachalotSettings.CACHALOT_UNCACHABLE_TABLES,
+    "procrastinate_events",
+    "procrastinate_jobs",
+    "procrastinate_periodic_defers",
+    "procrastinate_workers",
+]
 
 MARKDOWNIFY = {
     "default": {
@@ -278,6 +276,9 @@
 ISIC_USE_ELASTICSEARCH_COUNTS = True
 # opensearch logs every single request, which is too verbose
 logging.getLogger("elastic_transport").setLevel(logging.WARNING)
+# procrastinate logs every task registration at import time, in every process
+logging.getLogger("procrastinate.blueprints").setLevel(logging.WARNING)
+logging.getLogger("procrastinate.periodic").setLevel(logging.WARNING)
 
 ISIC_DATACITE_API_URL: ParseResult | None = env.url("DJANGO_ISIC_DATACITE_API_URL", default=None)
 # These are the default styles with their proper names that are used by the

isic/settings/testing.py

@@ -63,6 +63,12 @@
 CELERY_TASK_ALWAYS_EAGER = True
 CELERY_TASK_EAGER_PROPAGATES = True
 
+# Don't hold persistent connections in tests. The procrastinate worker runs tasks
+# in an asgiref thread pool, and close_old_connections() respects CONN_MAX_AGE, so
+# a healthy persistent connection in one of those threads would outlive the test
+# and block the test database from being dropped.
+DATABASES["default"]["CONN_MAX_AGE"] = 0
+
 ISIC_ELASTICSEARCH_IMAGES_INDEX = "test-isic-images"
 ISIC_ELASTICSEARCH_LESIONS_INDEX = "test-isic-lesions"
 ISIC_USE_ELASTICSEARCH_COUNTS = False