diff --git a/apps/notifications/announcement_worker.py b/apps/notifications/announcement_worker.py index 6724dd7..83d7132 100644 --- a/apps/notifications/announcement_worker.py +++ b/apps/notifications/announcement_worker.py @@ -94,7 +94,8 @@ def process_due_announcements(*, job_id=None): ).update(status=AnnouncementJob.Status.PENDING, next_attempt_at=now) with transaction.atomic(): queryset = ( - AnnouncementJob.objects.select_for_update(skip_locked=True) + # Nullable message/ticket joins must not be included in FOR UPDATE on PostgreSQL. + AnnouncementJob.objects.select_for_update(skip_locked=True, of=("self",)) .filter(status=AnnouncementJob.Status.PENDING, next_attempt_at__lte=now) .select_related("message__ticket__customer", "ticket__customer") .annotate(priority=Case(When(source=AnnouncementJob.Source.MANUAL, then=Value(0)), default=Value(1), output_field=IntegerField())) diff --git a/tests/test_announcements.py b/tests/test_announcements.py index 0ec4c0b..593ff34 100644 --- a/tests/test_announcements.py +++ b/tests/test_announcements.py @@ -84,6 +84,28 @@ def test_due_unread_message_calls_once_with_optional_api_key(announcement_data, assert post.call_args.kwargs["headers"] == ({"Idempotency-Key": job.idempotency_key, "X-API-Key": api_key} if api_key else {"Idempotency-Key": job.idempotency_key}) +def test_worker_locks_only_announcement_rows_with_nullable_relations(announcement_data, monkeypatch): + category, staff, customer, _ = announcement_data + ticket = create_ticket(customer=customer, created_by=staff, category=category, subject="پیگیری تماس", body="پیام نخست") + job = AnnouncementJob.objects.get(message__ticket=ticket) + due(job) + manager = AnnouncementJob.objects + original = manager.select_for_update + lock_targets = [] + + def capture_lock(*args, **kwargs): + queryset = original(*args, **kwargs) + lock_targets.append(queryset.query.select_for_update_of) + return queryset + + monkeypatch.setattr(manager, "select_for_update", capture_lock) + with patch("apps.integrations.announcements.requests.post", return_value=Mock(content=b"")): + call_command("process_notifications", once=True) + job.refresh_from_db() + assert job.status == AnnouncementJob.Status.SENT + assert lock_targets == [("self",), ("self",)] + + def test_customer_read_cancels_only_that_message(announcement_data): category, staff, customer, _ = announcement_data ticket = create_ticket(customer=customer, created_by=staff, category=category, subject="پیگیری مشتری", body="پیام نخست")