feat: Add proper logging for dedup callback

Author: roma-valor

src/country_workspace/config/fragments/constance.py

@@ -1,5 +1,5 @@
 from .app import AURORA_API_TOKEN, AURORA_API_URL, HOPE_API_TOKEN, HOPE_API_URL, NEW_USER_DEFAULT_GROUP
-from .dedup import DEDUP_API_URL, DEDUP_API_TOKEN
+from .dedup import DEDUP_API_URL, DEDUP_API_TOKEN, APP_BASE_URL
 from .kobo import KOBO_API_TOKEN, KOBO_KF_URL, KOBO_MASTER_API_TOKEN, KOBO_PROJECT_VIEW_ID
 from .mail import MAILJET_API_KEY, MAILJET_SECRET_KEY
 
@@ -68,6 +68,7 @@
     "MAILJET_SECRET_KEY": (MAILJET_SECRET_KEY, "Mailjet secret key", "write_only_text_input"),
     "DEDUP_API_URL": (DEDUP_API_URL, "Dedup Engine server address", str),
     "DEDUP_API_TOKEN": (DEDUP_API_TOKEN, "Dedup Engine API access token", "write_only_text_input"),
+    "APP_BASE_URL": (APP_BASE_URL, "Current server address", str),
     "CHUNK_SIZE_FOR_VALIDATION_TASK": (500, "Number of records to process per chunk in validation tasks", int),
     "RDP_CLEANUP_DAYS": (
         30,
@@ -161,6 +162,7 @@
         "NEW_USER_IS_STAFF",
         "NEW_USER_DEFAULT_GROUP",
         "CONCURRENCY_GUARD",
+        "APP_BASE_URL",
     ),
 }
 

src/country_workspace/config/urls.py

@@ -18,7 +18,7 @@
     path("select2/", include(django_select2.urls)),
     path(r"__debug__/", include(debug_toolbar.urls)),
     path(
-        "hope/dedup/callback/<str:signed_token>/",
+        "api/dedup/callback/<str:signed_token>/",
         DeduplicationCallbackView.as_view(),
         name="dedup_callback",
     ),

src/country_workspace/contrib/hope/push/orchestration.py

@@ -1,9 +1,10 @@
+import logging
 from collections.abc import Callable, Iterator
 from functools import partial
 from typing import Any, NamedTuple
+from constance import config
 from uuid import UUID, uuid4
 
-from django.conf import settings
 from django.core import signing
 from django.db import IntegrityError, transaction
 from django.urls import reverse
@@ -43,6 +44,8 @@
 _DEDUP_CALLBACK_SIGN_KEY = "dedup_callback"
 _DEDUP_CALLBACK_MAX_AGE = 60 * 60 * 96  # 96 hours
 
+logger = logging.getLogger(__name__)
+
 
 class CloneDeduplicationContext(NamedTuple):
     dedup_source_pk: int
@@ -130,7 +133,7 @@ def _build_dedup_callback_url(rdp_id: int, job_id: int) -> str:
         key=_DEDUP_CALLBACK_SIGN_KEY,
     )
     path = reverse("dedup_callback", kwargs={"signed_token": token})
-    base = getattr(settings, "APP_BASE_URL", "").rstrip("/")
+    base = config.APP_BASE_URL.rstrip("/")
     return f"{base}{path}"
 
 
@@ -144,9 +147,10 @@ def create_rdp_and_start_dedup_core(job: AsyncJob) -> dict[str, Any]:
     create_result = create_rdp_core(job)
     rdp_id = create_result["rdp_id"]
 
-    _check, rdp = claim_rdp_deduplication(rdp_id)
-    if rdp is None:
-        raise HopePushError({"errors": ["RDP: could not claim deduplication set."], "rdp_id": rdp_id})
+    with transaction.atomic():
+        _check, rdp = claim_rdp_deduplication(rdp_id)
+        if rdp is None:
+            raise HopePushError({"errors": ["RDP: could not claim deduplication set."], "rdp_id": rdp_id})
 
     job.refresh_from_db()
 
@@ -179,11 +183,23 @@ def create_and_push_rdp_core(job: AsyncJob) -> dict[str, Any]:
     return create_rdp_and_start_dedup_core(job)
 
 
-def _lock_and_fail_rdp(rdp_id: int) -> None:
+def _lock_and_fail_rdp(rdp_id: int, *, reason: str) -> None:
     with transaction.atomic():
         locked = lock_rdp_for_update(pk=rdp_id)
         if locked.status == Rdp.PushStatus.DEDUP_PENDING:
             set_rdp_push_status(rdp=locked, status=Rdp.PushStatus.FAILURE, hope_rdi_id="N/A")
+            logger.warning(
+                "dedup_callback_handle: rdp_id=%s marked FAILURE (%s)",
+                rdp_id,
+                reason,
+            )
+        else:
+            logger.info(
+                "dedup_callback_handle: rdp_id=%s skip FAILURE (%s); status=%s",
+                rdp_id,
+                reason,
+                locked.status,
+            )
 
 
 def _handle_deduplicated(rdp: Rdp, rdp_id: int, findings_count: int) -> None:
@@ -193,13 +209,31 @@ def _handle_deduplicated(rdp: Rdp, rdp_id: int, findings_count: int) -> None:
     total_individuals = qs_individuals_for_rdp(rdp=rdp).count()
     findings_rate = findings_count / total_individuals * 100 if total_individuals > 0 else float(findings_count)
 
+    logger.info(
+        "dedup_callback_handle: rdp_id=%s DEDUPLICATED findings_count=%s total_individuals=%s "
+        "findings_rate=%.2f%% max_dedup_findings_percent=%s",
+        rdp_id,
+        findings_count,
+        total_individuals,
+        findings_rate,
+        max_findings_percent,
+    )
+
     if findings_rate > max_findings_percent:
-        _lock_and_fail_rdp(rdp_id)
+        _lock_and_fail_rdp(
+            rdp_id,
+            reason=f"findings_rate {findings_rate:.2f}% exceeds threshold {max_findings_percent}%",
+        )
         return
 
     with transaction.atomic():
         locked = lock_rdp_for_update(pk=rdp_id)
         if locked.status != Rdp.PushStatus.DEDUP_PENDING:
+            logger.warning(
+                "dedup_callback_handle: rdp_id=%s skip push queue; status changed to %s",
+                rdp_id,
+                locked.status,
+            )
             return
         locked.status = Rdp.PushStatus.PENDING
         locked.save(update_fields=["status"])
@@ -214,6 +248,11 @@ def _handle_deduplicated(rdp: Rdp, rdp_id: int, findings_count: int) -> None:
         config={"rdp_id": rdp_id},
     )
     push_job.queue()
+    logger.info(
+        "dedup_callback_handle: rdp_id=%s status DEDUP_PENDING→PENDING; queued push job_id=%s",
+        rdp_id,
+        push_job.pk,
+    )
 
 
 def dedup_callback_handle(rdp_id: int) -> None:
@@ -231,27 +270,64 @@ def dedup_callback_handle(rdp_id: int) -> None:
     try:
         rdp = rdp_for_push(pk=rdp_id)
     except Rdp.DoesNotExist:
+        logger.warning("dedup_callback_handle: rdp_id=%s not found", rdp_id)
         return
 
     if rdp.status != Rdp.PushStatus.DEDUP_PENDING:
+        logger.info(
+            "dedup_callback_handle: rdp_id=%s skip; status=%s (expected DEDUP_PENDING)",
+            rdp_id,
+            rdp.status,
+        )
         return
 
     if not rdp.deduplication_set_id:
-        _lock_and_fail_rdp(rdp_id)
+        logger.warning("dedup_callback_handle: rdp_id=%s missing deduplication_set_id", rdp_id)
+        _lock_and_fail_rdp(rdp_id, reason="missing deduplication_set_id")
         return
 
+    logger.info(
+        "dedup_callback_handle: rdp_id=%s deduplication_set_id=%s; fetching remote status",
+        rdp_id,
+        rdp.deduplication_set_id,
+    )
+
     status = get_rdp_policy(rdp).deduplication_status(rdp)
-    if status is None or status.response_status != DedupResponseStatus.OK:
-        # Dedup engine unavailable — leave in DEDUP_PENDING for the next callback.
+    if status is None:
+        logger.warning(
+            "dedup_callback_handle: rdp_id=%s dedup engine returned no status; leaving DEDUP_PENDING",
+            rdp_id,
+        )
+        return
+
+    if status.response_status != DedupResponseStatus.OK:
+        logger.warning(
+            "dedup_callback_handle: rdp_id=%s dedup engine status unavailable (%s); leaving DEDUP_PENDING",
+            rdp_id,
+            status.response_status,
+        )
         return
 
     dedup_state = status.deduplication_set_status
     terminal_failure_states = {DeduplicationSetState.DEDUPLICATION_FAILED, DeduplicationSetState.ENCODING_FAILED}
 
+    logger.info(
+        "dedup_callback_handle: rdp_id=%s remote_state=%s findings_count=%s",
+        rdp_id,
+        dedup_state,
+        status.findings_count,
+    )
+
     if dedup_state in terminal_failure_states:
-        _lock_and_fail_rdp(rdp_id)
+        _lock_and_fail_rdp(rdp_id, reason=f"dedup engine state {dedup_state}")
     elif dedup_state == DeduplicationSetState.DEDUPLICATED:
         _handle_deduplicated(rdp, rdp_id, status.findings_count)
+    else:
+        logger.info(
+            "dedup_callback_handle: rdp_id=%s intermediate state %s; waiting for next callback",
+            rdp_id,
+            dedup_state,
+        )
 
 
 def claim_rdp_deduplication(rdp_id: int) -> tuple[ActionCheck, Rdp | None]:

tests/contrib/hope/push/test_dedup_callback_flow.py

@@ -8,6 +8,7 @@
 """
 
 import pytest
+from constance.test import override_config
 from pytest_mock import MockerFixture
 
 from country_workspace.contrib.dedup_engine import (
@@ -35,11 +36,10 @@
 # ---------------------------------------------------------------------------
 
 
-def test_build_dedup_callback_url_contains_signed_token(settings) -> None:
-    settings.APP_BASE_URL = "https://cw.example.org"
+@override_config(APP_BASE_URL="https://cw.example.org")
+def test_build_dedup_callback_url_contains_signed_token() -> None:
     url = _build_dedup_callback_url(rdp_id=7, job_id=42)
-    assert url.startswith("https://cw.example.org")
-    assert "hope/dedup/callback/" in url
+    assert url.startswith("https://cw.example.org/api/dedup/callback/")
 
 
 def test_build_dedup_callback_url_token_verifiable() -> None:
@@ -342,6 +342,34 @@ def test_dedup_callback_handle_deduplicated_within_threshold_queues_push(mocker:
     assert locked_pending.status == Rdp.PushStatus.PENDING
 
 
+def test_dedup_callback_handle_deduplicated_skips_push_when_status_changed_under_lock(
+    mocker: MockerFixture,
+) -> None:
+    rdp = _make_rdp(mocker)
+    locked = mocker.MagicMock()
+    locked.status = Rdp.PushStatus.FAILURE
+    _patch_rdp_for_push(mocker, rdp)
+    _patch_dedup_status(mocker, state=DeduplicationSetState.DEDUPLICATED, findings_count=5)
+    origin_job = mocker.MagicMock(config={"max_dedup_findings_percent": 10})
+    mocker.patch.object(
+        AsyncJob.objects,
+        "filter",
+        return_value=mocker.MagicMock(
+            order_by=mocker.MagicMock(return_value=mocker.MagicMock(first=mocker.MagicMock(return_value=origin_job)))
+        ),
+    )
+    mocker.patch(
+        f"{MOD}.qs_individuals_for_rdp", return_value=mocker.MagicMock(count=mocker.MagicMock(return_value=100))
+    )
+    mocker.patch(f"{MOD}.lock_rdp_for_update", return_value=locked)
+    create_job = mocker.patch.object(AsyncJob.objects, "create")
+
+    dedup_callback_handle(rdp_id=99)
+
+    create_job.assert_not_called()
+    locked.save.assert_not_called()
+
+
 def test_dedup_callback_handle_deduplicated_exceeds_threshold_marks_failure(mocker: MockerFixture) -> None:
     rdp = _make_rdp(mocker)
     locked = mocker.MagicMock()