UN-3016 [FIX] Record why an execution failed; stop dispatching skipped files

backend/workflow_manager/endpoint_v2/source.py

@@ -2,7 +2,6 @@
 import logging
 import os
 import shutil
-import uuid
 from collections.abc import Collection
 from hashlib import sha256
 from io import BytesIO
@@ -1188,7 +1187,7 @@
         return os.path.basename(input_file_path), file_stream
 
     @classmethod
     def add_input_file_to_api_storage(
         cls,
         pipeline_id: str,
         workflow_id: str,
@@ -1223,6 +1222,9 @@
         workflow: Workflow = Workflow.objects.get(id=workflow_id)
         file_hashes: dict[str, FileHash] = {}
         unique_file_hashes: set[str] = set()
+        # UN-3016: files rejected for an unsupported MIME type, pre-formatted
+        # for the error below; they are never staged and never handed to a worker.
+        skipped_files: list[str] = []
         connection_type = WorkflowEndpoint.ConnectionType.API
         for file in file_objs:
             file_name = file.name
@@ -1237,21 +1239,20 @@
                 mime_type = AllowedFileTypes.OCTET_STREAM.value
 
             if not AllowedFileTypes.is_allowed(mime_type):
-                log_message = f"Skipping file '{file_name}' to stage due to unsupported MIME type '{mime_type}'"
-                workflow_log.log_info(logger=logger, message=log_message)
-                # Generate a clearly marked temporary hash to avoid reading the file content
-                # Helps to prevent duplicate entries in file executions
-                fake_hash = f"temp-hash-{uuid.uuid4().hex}"
-                file_hash = FileHash(
-                    file_path=destination_path,
-                    source_connection_type=connection_type,
-                    file_name=file_name,
-                    file_hash=fake_hash,
-                    is_executed=True,
-                    file_size=file.size,
-                    mime_type=mime_type,
+                # UN-3016: the file is deliberately NOT staged (its bytes are never
+                # written), so it must not be handed to a worker either. Previously
+                # it was returned with is_executed=True and a temp hash; nothing
+                # downstream filters on is_executed, so the worker ran anyway, failed
+                # on the missing file, and the execution died with an opaque
+                # "Execution: <path>; Destination: <path>" error instead of a clear
+                # "skipped, unsupported type" outcome. Excluding it here keeps the
+                # skip a skip.
+                log_message = (
+                    f"Skipping file '{file_name}': unsupported file type "
+                    f"'{mime_type}'. It will not be processed."
                 )
-                file_hashes.update({file_name: file_hash})
+                workflow_log.log_error(logger=logger, message=log_message)
+                skipped_files.append(f"'{file_name}' ({mime_type})")
                 continue
 
             file_system = FileSystem(FileStorageType.API_EXECUTION)
@@ -1285,6 +1286,17 @@
                 mime_type=mime_type,
             )
             file_hashes.update({file_name: file_hash})
+
+        # UN-3016: if every uploaded file was rejected there is nothing to run.
+        # Fail loudly with the reason instead of dispatching an empty execution,
+        # which would otherwise finish as a vacuous success and leave the user
+        # wondering why nothing happened.
+        if skipped_files and not file_hashes:
+            raise UnsupportedMimeTypeError(
+                "No files could be processed. Unsupported file type(s): "
+                + ", ".join(skipped_files)
+            )
+
         return file_hashes
 
     @classmethod

backend/workflow_manager/endpoint_v2/tests/test_un3016_unsupported_mime_skip.py

@@ -0,0 +1,131 @@
+"""UN-3016: a file rejected for an unsupported MIME type must not reach a worker.
+
+``SourceConnector.add_input_file_to_api_storage`` deliberately does not stage
+the bytes of a file whose MIME type is not allowed. It used to return that file
+anyway, with ``is_executed=True`` and a placeholder hash; nothing downstream
+filters on ``is_executed``, so a worker picked the file up, failed on the file
+that was never written, and the whole execution died with an opaque
+"Execution: <path>; Destination: <path>" message. These tests pin the fix: the
+rejected file is excluded from the returned mapping, the supported files in the
+same request still go through, and an all-rejected request fails loudly with a
+message that names what was rejected.
+
+DB-free by construction: the ORM boundary (``Workflow.objects.get``) and the
+file storage are patched, so nothing here needs Postgres. Files are real
+``SimpleUploadedFile`` objects rather than mocks — a mock's ``content_type`` is
+not in ``AllowedFileTypes``, so a mocked "supported" file would silently take
+the skip branch and the test would pass for the wrong reason.
+"""
+
+from __future__ import annotations
+
+import os
+from unittest.mock import MagicMock, patch
+
+import django
+import pytest
+from django.apps import apps
+
+os.environ.setdefault("DJANGO_SETTINGS_MODULE", "backend.settings.test")
+if not apps.ready:
+    django.setup()
+
+from django.core.files.uploadedfile import SimpleUploadedFile  # noqa: E402
+from workflow_manager.endpoint_v2.exceptions import (  # noqa: E402
+    UnsupportedMimeTypeError,
+)
+from workflow_manager.endpoint_v2.source import SourceConnector  # noqa: E402
+
+SUPPORTED_MIME = "application/pdf"
+UNSUPPORTED_MIME = "application/x-msdownload"
+
+
+def _upload(name: str, content_type: str) -> SimpleUploadedFile:
+    """A real uploaded file: gives .name, .content_type, .size and .chunks()."""
+    return SimpleUploadedFile(name, b"some bytes", content_type=content_type)
+
+
+def _stage(file_objs: list[SimpleUploadedFile]) -> dict:
+    """Call the staging helper with every external boundary patched out."""
+    with (
+        patch("workflow_manager.endpoint_v2.source.UserContext") as mock_user_context,
+        patch("workflow_manager.endpoint_v2.source.WorkflowLog"),
+        patch("workflow_manager.endpoint_v2.source.Workflow") as mock_workflow,
+        patch("workflow_manager.endpoint_v2.source.FileSystem"),
+        patch.object(
+            SourceConnector,
+            "get_api_storage_dir_path",
+            return_value="unstract/api/org/exec-1",
+        ),
+    ):
+        mock_user_context.get_organization_identifier.return_value = "org"
+        mock_workflow.objects.get.return_value = MagicMock()
+        return SourceConnector.add_input_file_to_api_storage(
+            pipeline_id="pipeline-1",
+            workflow_id="workflow-1",
+            execution_id="exec-1",
+            file_objs=file_objs,
+        )
+
+
+def test_partial_skip_returns_only_the_supported_files():
+    """The core fix: a rejected file is absent from the returned mapping, and
+    the supported file alongside it is unaffected.
+
+    Also pins that a partial skip does NOT raise — rejecting some files while
+    others are runnable proceeds with what can be run (UN-4055 tracks whether
+    that should stay the behaviour).
+    """
+    file_hashes = _stage(
+        [
+            _upload("good.pdf", SUPPORTED_MIME),
+            _upload("bad.exe", UNSUPPORTED_MIME),
+        ]
+    )
+
+    assert "bad.exe" not in file_hashes
+    assert "good.pdf" in file_hashes
+    assert len(file_hashes) == 1
+    assert file_hashes["good.pdf"].mime_type == SUPPORTED_MIME
+    assert file_hashes["good.pdf"].file_name == "good.pdf"
+
+
+def test_supported_file_is_staged_with_a_real_hash():
+    """An accepted file's FileHash carries the sha256 of the bytes that were
+    staged.
+
+    Characterisation only: this held before the fix too, and no mutation of the
+    fix makes it fail. It is here to document what a returned entry looks like,
+    which is what makes the rejected file's absence elsewhere meaningful.
+    """
+    file_hashes = _stage([_upload("good.pdf", SUPPORTED_MIME)])
+
+    assert set(file_hashes) == {"good.pdf"}
+    file_hash = file_hashes["good.pdf"].file_hash
+    assert len(file_hash) == 64
+    assert all(c in "0123456789abcdef" for c in file_hash)
+
+
+def test_total_skip_raises_naming_the_skipped_files():
+    """When nothing survives the filter there is nothing to run: fail with the
+    reason instead of dispatching an empty execution that reports success.
+    """
+    with pytest.raises(UnsupportedMimeTypeError) as excinfo:
+        _stage(
+            [
+                _upload("bad.exe", UNSUPPORTED_MIME),
+                _upload("worse.dll", UNSUPPORTED_MIME),
+            ]
+        )
+
+    message = str(excinfo.value)
+    assert "bad.exe" in message
+    assert "worse.dll" in message
+    assert UNSUPPORTED_MIME in message
+
+
+def test_no_files_at_all_returns_empty_without_raising():
+    """An empty request has no skipped files, so it is not an unsupported-type
+    failure — the raise is guarded on there being something skipped.
+    """
+    assert _stage([]) == {}

backend/workflow_manager/workflow_v2/tests/test_un3016_execute_staging_cleanup.py

@@ -0,0 +1,96 @@
+"""UN-3016: a staging failure must clean up the API storage directory.
+
+``WorkflowViewSet.execute`` stages uploaded files before running the workflow.
+Staging can now fail part-way — ``add_input_file_to_api_storage`` raises
+``UnsupportedMimeTypeError`` when every uploaded file is rejected, and it may
+have written some files before reaching that point. The staging call therefore
+sits inside the ``try`` whose handler calls ``delete_api_storage_dir``, and that
+handler's guard (``has_uploads``) is exactly the condition under which staging
+ran at all. These tests pin both halves: cleanup happens when staging fails,
+and cleanup is not attempted for a request that never staged anything.
+
+DB-free: the serializer, the workflow lookup and both connectors are patched,
+so ``execute`` is exercised as pure control flow.
+"""
+
+from __future__ import annotations
+
+import os
+from contextlib import ExitStack
+from unittest.mock import MagicMock, patch
+
+import django
+import pytest
+from django.apps import apps
+
+os.environ.setdefault("DJANGO_SETTINGS_MODULE", "backend.settings.test")
+if not apps.ready:
+    django.setup()
+
+from workflow_manager.endpoint_v2.exceptions import (  # noqa: E402
+    UnsupportedMimeTypeError,
+)
+from workflow_manager.workflow_v2.views import WorkflowViewSet  # noqa: E402
+
+WORKFLOW_ID = "workflow-1"
+EXECUTION_ID = "exec-1"
+
+VIEWS = "workflow_manager.workflow_v2.views"
+
+
+def _request(with_files: bool) -> MagicMock:
+    request = MagicMock()
+    request.FILES.getlist.return_value = [MagicMock()] if with_files else []
+    return request
+
+
+def _patched_serializer(stack):
+    """Patch the serializer so execute() gets ids without parsing a payload."""
+    serializer_cls = stack.enter_context(patch(f"{VIEWS}.ExecuteWorkflowSerializer"))
+    serializer = serializer_cls.return_value
+    serializer.get_workflow_id.return_value = WORKFLOW_ID
+    serializer.get_execution_id.return_value = EXECUTION_ID
+    serializer.get_execution_action.return_value = None
+    return serializer
+
+
+def test_staging_failure_deletes_the_api_storage_dir():
+    """A staging failure leaves already-written files behind unless the handler
+    cleans up, so the failure must reach ``delete_api_storage_dir``.
+    """
+    with ExitStack() as stack:
+        _patched_serializer(stack)
+        source = stack.enter_context(patch(f"{VIEWS}.SourceConnector"))
+        destination = stack.enter_context(patch(f"{VIEWS}.DestinationConnector"))
+        source.add_input_file_to_api_storage.side_effect = UnsupportedMimeTypeError(
+            "No files could be processed. Unsupported file type(s): 'bad.exe'"
+        )
+
+        with pytest.raises(UnsupportedMimeTypeError):
+            WorkflowViewSet().execute(_request(with_files=True))
+
+        destination.delete_api_storage_dir.assert_called_once_with(
+            workflow_id=WORKFLOW_ID, execution_id=EXECUTION_ID
+        )
+
+
+def test_no_cleanup_when_the_request_staged_nothing():
+    """A request with no uploads never created a storage dir; a later failure
+    must not try to delete one.
+    """
+    with ExitStack() as stack:
+        _patched_serializer(stack)
+        stack.enter_context(patch(f"{VIEWS}.SourceConnector"))
+        destination = stack.enter_context(patch(f"{VIEWS}.DestinationConnector"))
+        stack.enter_context(
+            patch.object(
+                WorkflowViewSet,
+                "get_workflow_by_id",
+                side_effect=RuntimeError("boom"),
+            )
+        )
+
+        with pytest.raises(RuntimeError):
+            WorkflowViewSet().execute(_request(with_files=False))
+
+        destination.delete_api_storage_dir.assert_not_called()

backend/workflow_manager/workflow_v2/views.py

@@ -258,17 +258,21 @@ def execute(
         use_file_history: bool = True
 
         hashes_of_files: dict[str, FileHash] = {}
-        if file_objs and execution_id and workflow_id:
+        has_uploads = bool(file_objs and execution_id and workflow_id)
+        if has_uploads:
             use_file_history = False
-            hashes_of_files = SourceConnector.add_input_file_to_api_storage(
-                pipeline_id=pipeline_guid,
-                workflow_id=workflow_id,
-                execution_id=execution_id,
-                file_objs=file_objs,
-                use_file_history=False,
-            )
 
         try:
+            # Staged inside this try so the handler below cleans up after a
+            # partial stage: its guard is exactly this staging condition.
+            if has_uploads:
+                hashes_of_files = SourceConnector.add_input_file_to_api_storage(
+                    pipeline_id=pipeline_guid,
+                    workflow_id=workflow_id,
+                    execution_id=execution_id,
+                    file_objs=file_objs,
+                    use_file_history=False,
+                )
             workflow = self.get_workflow_by_id(workflow_id=workflow_id)
             execution_response = self.execute_workflow(
                 workflow=workflow,
@@ -290,7 +294,7 @@ def execute(
             )
         except Exception as exception:
             logger.error(f"Error while executing workflow: {exception}", exc_info=True)
-            if file_objs and execution_id and workflow_id:
+            if has_uploads:
                 DestinationConnector.delete_api_storage_dir(
                     workflow_id=workflow_id, execution_id=execution_id
                 )