subtle dsr processing issue (#6469)

Co-authored-by: Jade Wibbels <jade@ethyca.com>

Author: JadeCara

src/fides/api/service/privacy_request/request_service.py

@@ -191,7 +191,11 @@ def poll_for_exited_privacy_request_tasks(self: DatabaseTask) -> Set[str]:
             db.query(PrivacyRequest)
             .filter(
                 PrivacyRequest.status.in_(
-                    [PrivacyRequestStatus.in_processing, PrivacyRequestStatus.approved]
+                    [
+                        PrivacyRequestStatus.in_processing,
+                        PrivacyRequestStatus.approved,
+                        PrivacyRequestStatus.requires_input,
+                    ]
                 )
             )
             # Only look at Privacy Requests that haven't been deleted

src/fides/api/task/manual/manual_task_graph_task.py

@@ -15,6 +15,7 @@
     StatusType,
 )
 from fides.api.models.privacy_request import PrivacyRequest
+from fides.api.models.worker_task import ExecutionLogStatus
 from fides.api.schemas.policy import ActionType
 from fides.api.schemas.privacy_request import PrivacyRequestStatus
 from fides.api.task.graph_task import GraphTask, retry
@@ -40,6 +41,17 @@ def access_request(self, *inputs: list[Row]) -> list[Row]:
         db = self.resources.session
         collection_address = self.execution_node.address
 
+        if self.resources.request.policy.get_action_type() == ActionType.erasure:
+            # We're in an erasure privacy request's access phase - complete access task immediately
+            # since access is just for data collection to support erasure, not for user data access
+            self.update_status(
+                "Access task completed immediately for erasure privacy request (data collection only)",
+                [],
+                ActionType.access,
+                ExecutionLogStatus.complete,
+            )
+            return []
+
         # Verify this is a manual task address
         if not ManualTaskAddress.is_manual_task_address(collection_address):
             raise ValueError(f"Invalid manual task address: {collection_address}")

tests/api/task/manual/test_manual_task_graph_task.py

@@ -7,6 +7,9 @@
     ManualTaskInstance,
     ManualTaskSubmission,
 )
+from fides.api.models.worker_task import ExecutionLogStatus
+from fides.api.schemas.policy import ActionType
+from fides.api.schemas.privacy_request import PrivacyRequestStatus
 
 
 class TestManualTaskDataAggregation:
@@ -322,3 +325,25 @@ def test_aggregate_submission_data_attachment_field_no_attachments(
         # Should return None for attachment field with no attachments
         assert "user_email" in result
         assert result["user_email"] is None
+
+    def test_access_request_early_return_for_erasure_policy(
+        self, build_erasure_graph_task, db
+    ):
+        """Test that access_request returns early and completes immediately for erasure policy"""
+        manual_task, graph_task = build_erasure_graph_task
+        privacy_request = graph_task.resources.request
+
+        # Set privacy request to requires_input status
+        privacy_request.status = PrivacyRequestStatus.requires_input
+        privacy_request.save(db)
+
+        # Call access_request - should return early due to erasure policy
+        result = graph_task.access_request([])
+
+        # Should return empty list (early return for erasure policy)
+        assert result == []
+
+        # Privacy request status should remain requires_input since the early return path
+        # does not call _return_to_in_processing()
+        db.refresh(privacy_request)
+        assert privacy_request.status == PrivacyRequestStatus.requires_input

tests/api/task/manual/test_manual_task_integration.py

@@ -974,3 +974,118 @@ def test_manual_task_traversal_integration_with_nested_groups(
 
         # Verify traversal is valid
         assert traversal is not None
+
+
+@pytest.mark.integration
+class TestManualTaskIntegrationStatusUpdates:
+    """Test that manual task status updates correctly"""
+
+    @pytest.mark.usefixtures("erasure_privacy_request")
+    def test_erasure_request_updates_privacy_request_status_when_manual_task_completed(
+        self, build_erasure_graph_task, db
+    ):
+        """Test that erasure_request properly updates privacy request status when manual task is completed"""
+        manual_task, graph_task = build_erasure_graph_task
+        privacy_request = graph_task.resources.request
+
+        # Set privacy request to requires_input status to simulate the scenario
+        privacy_request.status = PrivacyRequestStatus.requires_input
+        privacy_request.save(db)
+
+        # Create a manual task instance for this privacy request
+        instance = ManualTaskInstance.create(
+            db=db,
+            data={
+                "task_id": manual_task.id,
+                "config_id": manual_task.configs[0].id,  # Use the first config
+                "entity_id": privacy_request.id,
+                "entity_type": ManualTaskEntityType.privacy_request.value,
+                "status": StatusType.pending.value,
+            },
+        )
+
+        # Create a submission to complete the manual task
+        field = manual_task.configs[0].field_definitions[0]  # Use the first field
+        submission = ManualTaskSubmission.create(
+            db=db,
+            data={
+                "task_id": manual_task.id,
+                "config_id": manual_task.configs[0].id,
+                "field_id": field.id,
+                "instance_id": instance.id,
+                "submitted_by": None,
+                "data": {
+                    "field_type": ManualTaskFieldType.text.value,
+                    "value": "test_value",
+                },
+            },
+        )
+
+        # Mark the instance as completed
+        instance.status = StatusType.completed.value
+        instance.save(db)
+
+        # Call erasure_request - should update status and return 0
+        result = graph_task.erasure_request([])
+
+        # Should return 0 (manual tasks don't mask data directly)
+        assert result == 0
+
+        # Privacy request status should remain requires_input since the early return path
+        db.refresh(privacy_request)
+        assert privacy_request.status == PrivacyRequestStatus.requires_input
+
+    def test_access_request_updates_privacy_request_status_when_manual_task_completed(
+        self, build_graph_task, db
+    ):
+        """Test that access_request properly updates privacy request status when manual task is completed"""
+        manual_task, graph_task = build_graph_task
+        privacy_request = graph_task.resources.request
+
+        # Set privacy request to requires_input status to simulate the scenario
+        privacy_request.status = PrivacyRequestStatus.requires_input
+        privacy_request.save(db)
+
+        # Create a manual task instance for this privacy request
+        instance = ManualTaskInstance.create(
+            db=db,
+            data={
+                "task_id": manual_task.id,
+                "config_id": manual_task.configs[0].id,  # Use the first config
+                "entity_id": privacy_request.id,
+                "entity_type": ManualTaskEntityType.privacy_request.value,
+                "status": StatusType.pending.value,
+            },
+        )
+
+        # Create a submission to complete the manual task
+        field = manual_task.configs[0].field_definitions[0]  # Use the first field
+        submission = ManualTaskSubmission.create(
+            db=db,
+            data={
+                "task_id": manual_task.id,
+                "config_id": manual_task.configs[0].id,
+                "field_id": field.id,
+                "instance_id": instance.id,
+                "submitted_by": None,
+                "data": {
+                    "field_type": ManualTaskFieldType.text.value,
+                    "value": "test_value",
+                },
+            },
+        )
+
+        # Mark the instance as completed
+        instance.status = StatusType.completed.value
+        instance.save(db)
+
+        # Call access_request - should update status and return data
+        result = graph_task.access_request([])
+
+        # Should return the data from the manual task
+        assert len(result) > 0
+        assert "user_email" in result[0]  # The field key from the fixture
+
+        # Privacy request status should remain requires_input since the early return path
+        db.refresh(privacy_request)
+        assert privacy_request.status == PrivacyRequestStatus.requires_input

tests/ops/service/privacy_request/test_request_service.py

@@ -187,6 +187,84 @@ def test_approved_privacy_request_task_with_errored_tasks(
         db.refresh(privacy_request)
         assert privacy_request.status == PrivacyRequestStatus.error
 
+    def test_requires_input_privacy_request_task_with_errored_tasks(
+        self, db, privacy_request_requires_input
+    ):
+        """Privacy requests in requires_input status should be monitored for task errors
+        and marked as errored if tasks fail.
+
+        The "poll_for_exited_privacy_request_tasks" task looks for Privacy Requests in
+        "approved", "in_processing", and "requires_input" states.
+        """
+
+        # Create the necessary tasks for this privacy request (similar to request_task fixture)
+        root_task = RequestTask.create(
+            db,
+            data={
+                "action_type": ActionType.access,
+                "status": "complete",
+                "privacy_request_id": privacy_request_requires_input.id,
+                "collection_address": "__ROOT__:__ROOT__",
+                "dataset_name": "__ROOT__",
+                "collection_name": "__ROOT__",
+                "upstream_tasks": [],
+                "downstream_tasks": ["test_dataset:test_collection"],
+                "all_descendant_tasks": [
+                    "test_dataset:test_collection",
+                    "__TERMINATE__:__TERMINATE__",
+                ],
+            },
+        )
+
+        request_task = RequestTask.create(
+            db,
+            data={
+                "action_type": ActionType.access,
+                "status": "pending",
+                "privacy_request_id": privacy_request_requires_input.id,
+                "collection_address": "test_dataset:test_collection",
+                "dataset_name": "test_dataset",
+                "collection_name": "test_collection",
+                "upstream_tasks": ["__ROOT__:__ROOT__"],
+                "downstream_tasks": ["__TERMINATE__:__TERMINATE__"],
+                "all_descendant_tasks": ["__TERMINATE__:__TERMINATE__"],
+            },
+        )
+
+        terminator_task = RequestTask.create(
+            db,
+            data={
+                "action_type": ActionType.access,
+                "status": "pending",
+                "privacy_request_id": privacy_request_requires_input.id,
+                "collection_address": "__TERMINATE__:__TERMINATE__",
+                "dataset_name": "__TERMINATE__",
+                "collection_name": "__TERMINATE__",
+                "upstream_tasks": ["test_dataset:test_collection"],
+                "downstream_tasks": [],
+                "all_descendant_tasks": [],
+            },
+        )
+
+        # Put all tasks in an exited state - completed, errored, or skipped
+        assert root_task.status == ExecutionLogStatus.complete
+        request_task.update_status(db, ExecutionLogStatus.error)
+        terminator_task.update_status(db, ExecutionLogStatus.error)
+
+        errored_prs = poll_for_exited_privacy_request_tasks.delay().get()
+        assert errored_prs == {privacy_request_requires_input.id}
+
+        db.refresh(privacy_request_requires_input)
+        assert privacy_request_requires_input.status == PrivacyRequestStatus.error
+
+        # Clean up created tasks
+        try:
+            root_task.delete(db)
+            request_task.delete(db)
+            terminator_task.delete(db)
+        except Exception:
+            pass
+
     def test_request_tasks_all_exited_none_errored(
         self, db, privacy_request, request_task
     ):