From e79577811e5864addc686850b035d592234847f6 Mon Sep 17 00:00:00 2001 From: Ryan Westfall Date: Mon, 24 Aug 2026 10:04:24 -0500 Subject: [PATCH] Poll dispatch_due in the worker so scheduled posts fire. Co-authored-by: Cursor --- .env.example | 3 ++ README.md | 9 ++-- docker-compose.prod.yml | 1 + docker-compose.yml | 3 +- scripts/worker-entrypoint.sh | 21 +++++++- site/core/management/commands/dispatch_due.py | 36 +++++++++++++- site/core/tests.py | 48 +++++++++++++++++++ 7 files changed, 113 insertions(+), 8 deletions(-) diff --git a/.env.example b/.env.example index d0a98a6..edfb813 100644 --- a/.env.example +++ b/.env.example @@ -92,4 +92,7 @@ TIANJI_ENABLED=false TIANJI_TRACKER_URL=https://tianji.aimloperations.com/tracker.js TIANJI_WEBSITE_ID=cmshzhxdf6gee10qzkzrfn9iw +# Worker poll interval for due scheduled campaigns / social posts (seconds). +# DISPATCH_DUE_INTERVAL_SECONDS=15 + GUNICORN_WORKERS=2 diff --git a/README.md b/README.md index 1fa75bd..67470e0 100644 --- a/README.md +++ b/README.md @@ -25,13 +25,14 @@ uv sync cd site uv run python manage.py migrate uv run python manage.py runserver -# optional: docker compose up # web + postgres (hot reload) -# optional worker: docker compose --profile worker up worker +# optional: docker compose up # web + postgres + worker (hot reload) ``` Docker compose mounts `./site` into the container and runs Django `runserver` when -`DJANGO_ENV=dev` (the default). Edit Python/templates → auto-reload; no image rebuild. -Rebuild only when Dockerfile / deps (`pyproject.toml`, `uv.lock`) change: +`DJANGO_ENV=dev` (the default). The **worker** service polls `dispatch_due` every 15s so +scheduled campaigns and social posts fire without a manual command. Edit Python/templates +→ auto-reload; no image rebuild. +Rebuild only when Dockerfile, entrypoint scripts, or deps (`pyproject.toml`, `uv.lock`) change: ```bash docker compose up --build diff --git a/docker-compose.prod.yml b/docker-compose.prod.yml index d636ff3..2edbeda 100644 --- a/docker-compose.prod.yml +++ b/docker-compose.prod.yml @@ -4,6 +4,7 @@ # Web runs on every app host (active/active). Start the dj-queue worker on # exactly ONE host via: # docker compose -f docker-compose.prod.yml --profile worker up -d +# Worker also polls dispatch_due so scheduled campaigns/social posts fire. services: web: build: . diff --git a/docker-compose.yml b/docker-compose.yml index 3565892..97b3dc9 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -67,7 +67,6 @@ services: worker: build: . - profiles: ["worker"] entrypoint: ["/worker-entrypoint.sh"] volumes: - ./site:/app/site @@ -77,6 +76,8 @@ services: DJANGO_DEBUG: ${DJANGO_DEBUG:-true} DJANGO_ALLOWED_HOSTS: ${DJANGO_ALLOWED_HOSTS:-localhost,127.0.0.1,0.0.0.0} DATABASE_URL: ${COMPOSE_DATABASE_URL:-postgres://monica_site:monica_site@db:5432/monica_site} + PUBLIC_SITE_URL: ${PUBLIC_SITE_URL:-http://127.0.0.1:8000} + DISPATCH_DUE_INTERVAL_SECONDS: ${DISPATCH_DUE_INTERVAL_SECONDS:-15} EMAIL_HOST: ${EMAIL_HOST:-mail.smtp2go.com} EMAIL_HOST_USER: ${EMAIL_HOST_USER:-} EMAIL_HOST_PASSWORD: ${EMAIL_HOST_PASSWORD:-} diff --git a/scripts/worker-entrypoint.sh b/scripts/worker-entrypoint.sh index 5f78b44..b22c97f 100755 --- a/scripts/worker-entrypoint.sh +++ b/scripts/worker-entrypoint.sh @@ -47,5 +47,24 @@ uv run python manage.py migrate --noinput # worker must too or hashed lookups raise ValueError. uv run python manage.py collectstatic --noinput +interval="${DISPATCH_DUE_INTERVAL_SECONDS:-15}" +echo "Starting dispatch_due poller (every ${interval}s)." +uv run python manage.py dispatch_due --loop --interval "${interval}" & +dispatch_pid=$! + +cleanup() { + echo "Stopping dispatch_due poller..." + kill "${dispatch_pid}" 2>/dev/null || true + wait "${dispatch_pid}" 2>/dev/null || true +} +trap cleanup EXIT INT TERM + +# Dev uses ImmediateBackend: enqueue runs inline in dispatch_due. No dj-queue. +if [[ "${DJANGO_ENV:-}" == "dev" ]]; then + echo "Dev worker: dispatch_due loop publishes scheduled work." + wait "${dispatch_pid}" + exit $? +fi + # dj-queue supervisor (workers + dispatcher + scheduler). Run on ONE host only. -exec uv run python manage.py dj_queue +uv run python manage.py dj_queue diff --git a/site/core/management/commands/dispatch_due.py b/site/core/management/commands/dispatch_due.py index 61a3e6b..33cc3e3 100644 --- a/site/core/management/commands/dispatch_due.py +++ b/site/core/management/commands/dispatch_due.py @@ -1,3 +1,5 @@ +import time + from django.core.management.base import BaseCommand from django.utils import timezone @@ -10,10 +12,37 @@ from social.tasks import publish_social_post class Command(BaseCommand): help = ( "Enqueue due scheduled campaign messages and social posts. " - "Optional when the task backend supports run_after defer; useful as a safety net." + "Pass --loop to poll until stopped (worker entrypoint)." ) + def add_arguments(self, parser): + parser.add_argument( + "--loop", + action="store_true", + help="Poll forever until SIGINT/SIGTERM.", + ) + parser.add_argument( + "--interval", + type=int, + default=15, + help="Seconds between polls when --loop is set (default 15).", + ) + def handle(self, *args, **options): + self.verbosity = int(options.get("verbosity", 1)) + if options["loop"]: + interval = max(1, int(options["interval"] or 15)) + self.stdout.write(f"Polling due scheduled work every {interval}s.") + try: + while True: + self._dispatch_once() + time.sleep(interval) + except KeyboardInterrupt: + self.stdout.write("dispatch_due loop stopped.") + return + self._dispatch_once() + + def _dispatch_once(self): now = timezone.now() enqueued = 0 @@ -35,4 +64,7 @@ class Command(BaseCommand): publish_social_post.enqueue(post_id=str(post.pk)) enqueued += 1 - self.stdout.write(self.style.SUCCESS(f"Enqueued {enqueued} due item(s).")) + if enqueued: + self.stdout.write(self.style.SUCCESS(f"Enqueued {enqueued} due item(s).")) + elif self.verbosity >= 2: + self.stdout.write("No due items.") diff --git a/site/core/tests.py b/site/core/tests.py index 8d2967c..e606058 100644 --- a/site/core/tests.py +++ b/site/core/tests.py @@ -1,5 +1,13 @@ +from datetime import timedelta +from io import StringIO +from unittest.mock import patch + +from django.core.management import call_command from django.test import Client, TestCase, override_settings from django.urls import reverse +from django.utils import timezone + +from social.models import SocialPost class HealthzTests(TestCase): @@ -27,3 +35,43 @@ class PublicSmokeTests(TestCase): client = Client() self.assertEqual(client.get(reverse("public:about")).status_code, 200) self.assertEqual(client.get(reverse("public:contact")).status_code, 200) + + +class DispatchDueTests(TestCase): + def test_enqueues_past_scheduled_social_post(self): + post = SocialPost.objects.create( + body="Open house", + status=SocialPost.Status.SCHEDULED, + scheduled_for=timezone.now() - timedelta(minutes=5), + ) + mock_task = patch( + "core.management.commands.dispatch_due.publish_social_post" + ).start() + self.addCleanup(patch.stopall) + out = StringIO() + call_command("dispatch_due", stdout=out) + post.refresh_from_db() + self.assertEqual(post.status, SocialPost.Status.QUEUED) + mock_task.enqueue.assert_called_once_with(post_id=str(post.pk)) + self.assertIn("Enqueued 1 due item(s).", out.getvalue()) + + def test_skips_future_scheduled_social_post(self): + post = SocialPost.objects.create( + body="Later", + status=SocialPost.Status.SCHEDULED, + scheduled_for=timezone.now() + timedelta(hours=1), + ) + mock_task = patch( + "core.management.commands.dispatch_due.publish_social_post" + ).start() + self.addCleanup(patch.stopall) + call_command("dispatch_due", stdout=StringIO()) + post.refresh_from_db() + self.assertEqual(post.status, SocialPost.Status.SCHEDULED) + mock_task.enqueue.assert_not_called() + + def test_quiet_when_nothing_due(self): + out = StringIO() + call_command("dispatch_due", stdout=out) + self.assertEqual(out.getvalue(), "") +