diff --git a/backend/api_v2/deployment_helper.py b/backend/api_v2/deployment_helper.py index 5fbec7999c..143544f906 100644 --- a/backend/api_v2/deployment_helper.py +++ b/backend/api_v2/deployment_helper.py @@ -19,6 +19,7 @@ from utils.constants import Account, CeleryQueue from utils.local_context import StateStore from workflow_manager.endpoint_v2.destination import DestinationConnector +from workflow_manager.endpoint_v2.result_cache_utils import ResultCacheUtils from workflow_manager.endpoint_v2.source import SourceConnector from workflow_manager.workflow_v2.dto import ExecutionResponse from workflow_manager.workflow_v2.enums import ExecutionStatus @@ -293,7 +294,14 @@ def execute_workflow( logger.exception(f"Failed to mark execution {execution_id} as ERROR") # Async job never started — release the rate limit slot and clean up. - APIDeploymentRateLimiter.release_slot(api.organization, str(execution_id)) + # str(...organization_id), NOT the model instance: release_slot formats + # its argument into the Redis key, and acquire_slot built that key from + # str(organization.organization_id). Passing the instance ZREMs a + # non-member — it returns 0 and raises nothing, so the slot silently + # stays held for the full TTL. Same trap as undispatched_sweep.py:245. + APIDeploymentRateLimiter.release_slot( + str(api.organization.organization_id), str(execution_id) + ) DestinationConnector.delete_api_storage_dir( workflow_id=workflow_id, execution_id=execution_id ) @@ -306,6 +314,45 @@ def execute_workflow( ) ).data + # Staging rejected every file, so there is nothing to dispatch. The worker + # short-circuits an empty file set without writing a status back, which + # would strand this execution in PENDING — terminalise it here instead. + if not hash_values_of_files: + # Isolate the DB write the way the staging-failure path above does, so + # the rate limit slot and staging dir are released even if it raises. + execution = None + try: + execution = WorkflowExecutionServiceHelper.update_execution_completed( + str(execution_id), + total_files=len(file_objs), + failed_files=len(file_objs), + ) + except Exception: + logger.exception(f"Failed to mark execution {execution_id} as COMPLETED") + + APIDeploymentRateLimiter.release_slot( + str(api.organization.organization_id), str(execution_id) + ) + DestinationConnector.delete_api_storage_dir( + workflow_id=workflow_id, execution_id=execution_id + ) + # Report the stored status rather than asserting COMPLETED: the row may + # be missing, or the terminal guard may have refused the change. Claiming + # success here would only hide the stranded execution behind a 200 that a + # follow-up GET /status then contradicts. + return APIExecutionResponseSerializer( + ExecutionResponse( + workflow_id=workflow_id, + execution_id=execution_id, + execution_status=( + execution.status if execution else ExecutionStatus.ERROR.value + ), + result=ResultCacheUtils.get_api_results( + workflow_id=str(workflow_id), execution_id=str(execution_id) + ), + ) + ).data + try: result = WorkflowHelper.execute_workflow_async( workflow_id=workflow_id, @@ -352,7 +399,9 @@ def execute_workflow( # Dispatch failures are marked ERROR internally by execute_workflow_async; # post-dispatch failures (enrichment/config) must not overwrite a running # execution's status, so only release the slot and clean up storage here. - APIDeploymentRateLimiter.release_slot(api.organization, str(execution_id)) + APIDeploymentRateLimiter.release_slot( + str(api.organization.organization_id), str(execution_id) + ) # Clean up storage DestinationConnector.delete_api_storage_dir( diff --git a/backend/api_v2/tests/test_deployment_helper.py b/backend/api_v2/tests/test_deployment_helper.py index 39b23e5b16..e167dcbeb8 100644 --- a/backend/api_v2/tests/test_deployment_helper.py +++ b/backend/api_v2/tests/test_deployment_helper.py @@ -38,8 +38,8 @@ def collaborators(): mocks[ "WorkflowExecutionServiceHelper" ].create_workflow_execution.return_value = execution_row - mocks["SourceConnector"].add_input_file_to_api_storage.side_effect = ( - RuntimeError("boom") + mocks["SourceConnector"].add_input_file_to_api_storage.side_effect = RuntimeError( + "boom" ) yield mocks @@ -48,6 +48,7 @@ def _api() -> MagicMock: api = MagicMock() api.workflow.id = "wf-1" api.id = "pipe-1" + api.organization.organization_id = "org-uuid-1" return api @@ -74,9 +75,9 @@ def test_staging_failure_marks_execution_error(collaborators) -> None: def test_staging_failure_cleanup_survives_db_marking_error(collaborators) -> None: """If marking the row ERROR itself raises, cleanup must still run (not propagate).""" - collaborators["WorkflowExecutionServiceHelper"].update_execution_err.side_effect = ( - RuntimeError("db down") - ) + collaborators[ + "WorkflowExecutionServiceHelper" + ].update_execution_err.side_effect = RuntimeError("db down") # Must NOT raise — a failed error-marking should not break cleanup. dh.DeploymentHelper.execute_workflow( @@ -89,3 +90,124 @@ def test_staging_failure_cleanup_survives_db_marking_error(collaborators) -> Non # Cleanup still runs even though error-marking raised. collaborators["APIDeploymentRateLimiter"].release_slot.assert_called_once() collaborators["DestinationConnector"].delete_api_storage_dir.assert_called_once() + + +@pytest.fixture +def staging_rejects_everything(): + """Patch execute_workflow's collaborators; staging returns no dispatchable files.""" + with mock.patch.multiple( + dh, + WorkflowExecutionServiceHelper=mock.DEFAULT, + SourceConnector=mock.DEFAULT, + DestinationConnector=mock.DEFAULT, + APIDeploymentRateLimiter=mock.DEFAULT, + WorkflowHelper=mock.DEFAULT, + ResultCacheUtils=mock.DEFAULT, + Tag=mock.DEFAULT, + logger=mock.DEFAULT, + ) as mocks: + execution_row = MagicMock() + execution_row.id = "exec-123" + mocks[ + "WorkflowExecutionServiceHelper" + ].create_workflow_execution.return_value = execution_row + mocks["SourceConnector"].add_input_file_to_api_storage.return_value = {} + mocks["ResultCacheUtils"].get_api_results.return_value = [ + {"file": "evil.pdf", "status": "Failed", "error": "unsupported MIME type"} + ] + completed_row = MagicMock() + completed_row.status = "COMPLETED" + mocks[ + "WorkflowExecutionServiceHelper" + ].update_execution_completed.return_value = completed_row + yield mocks + + +def test_all_files_rejected_completes_without_dispatch( + staging_rejects_everything, +) -> None: + """A request whose every file is rejected must reach a terminal status. + + The worker short-circuits an empty file set without writing a status back, so + dispatching one strands the execution in PENDING and the caller polls forever. + """ + mocks = staging_rejects_everything + # A non-empty upload whose staging result is empty. Passing [] instead would + # leave the branch satisfied by `not file_objs` too, and the original bug - + # dispatching a request whose files were all rejected - would pass this test. + response = dh.DeploymentHelper.execute_workflow( + organization_name="org", + api=_api(), + file_objs=[MagicMock()], + timeout=-1, + ) + + # Nothing is dispatched... + mocks["WorkflowHelper"].execute_workflow_async.assert_not_called() + # ...the row is terminalised here instead of being left PENDING, and the + # counters are written so the run does not read back as a clean success... + mocks[ + "WorkflowExecutionServiceHelper" + ].update_execution_completed.assert_called_once_with( + "exec-123", total_files=1, failed_files=1 + ) + # ...the slot and staging dir are released. The slot must be released by org + # id string: release_slot formats its argument into the Redis key, so passing + # the model instance removes a non-member and silently holds the slot. + mocks["APIDeploymentRateLimiter"].release_slot.assert_called_once_with( + "org-uuid-1", "exec-123" + ) + mocks["DestinationConnector"].delete_api_storage_dir.assert_called_once() + # ...and the caller still sees why each file failed. + assert response["execution_status"] == "COMPLETED" + assert response["result"][0]["file"] == "evil.pdf" + assert response["result"][0]["status"] == "Failed" + + +def test_files_staged_successfully_are_dispatched(staging_rejects_everything) -> None: + """The short-circuit must not fire when staging did return files. + + Sibling to the test above: together they pin the branch to the staging result + rather than to the upload list. + """ + mocks = staging_rejects_everything + mocks["SourceConnector"].add_input_file_to_api_storage.return_value = { + "good.pdf": MagicMock() + } + + dh.DeploymentHelper.execute_workflow( + organization_name="org", + api=_api(), + file_objs=[MagicMock()], + timeout=-1, + ) + + mocks["WorkflowHelper"].execute_workflow_async.assert_called_once() + mocks["WorkflowExecutionServiceHelper"].update_execution_completed.assert_not_called() + + +def test_all_files_rejected_cleanup_survives_db_marking_error( + staging_rejects_everything, +) -> None: + """A failing status write must not strand the slot or the staging dir. + + update_execution_completed only catches DoesNotExist, so a lock timeout or a + dropped connection propagates; without isolation the org's rate limit slot + stays held for its full TTL and throttles every other call for that org. + """ + mocks = staging_rejects_everything + mocks[ + "WorkflowExecutionServiceHelper" + ].update_execution_completed.side_effect = Exception("db is down") + + response = dh.DeploymentHelper.execute_workflow( + organization_name="org", + api=_api(), + file_objs=[MagicMock()], + timeout=-1, + ) + + mocks["APIDeploymentRateLimiter"].release_slot.assert_called_once() + mocks["DestinationConnector"].delete_api_storage_dir.assert_called_once() + # The row never reached COMPLETED, so the response must not claim it did. + assert response["execution_status"] == "ERROR" diff --git a/backend/workflow_manager/endpoint_v2/source.py b/backend/workflow_manager/endpoint_v2/source.py index 23b0d04f2c..3c0e15da6a 100644 --- a/backend/workflow_manager/endpoint_v2/source.py +++ b/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 @@ -25,7 +24,11 @@ SourceConstant, SourceKey, ) -from workflow_manager.endpoint_v2.dto import FileHash, SourceConfig +from workflow_manager.endpoint_v2.dto import ( + FileExecutionResult, + FileHash, + SourceConfig, +) from workflow_manager.endpoint_v2.enums import AllowedFileTypes from workflow_manager.endpoint_v2.exceptions import ( InvalidInputDirectory, @@ -37,6 +40,7 @@ UnsupportedMimeTypeError, ) from workflow_manager.endpoint_v2.models import WorkflowEndpoint +from workflow_manager.endpoint_v2.result_cache_utils import ResultCacheUtils from workflow_manager.file_execution.models import WorkflowFileExecution from workflow_manager.utils.workflow_log import WorkflowLog from workflow_manager.workflow_v2.enums import ExecutionStatus @@ -69,6 +73,15 @@ class SourceConnector(BaseConnector): """ READ_CHUNK_SIZE = 4194304 # Chunk size for reading files + # Most formats are identifiable from their leading bytes, so a small sample + # keeps the common path cheap. + MIME_DETECT_CHUNK_SIZE = 8192 + # These two carry the real format in a structure libmagic can only reach by + # reading the whole file: the OLE2 directory sector and the zip central + # directory both sit at the end. A sample of any size reports the container + # rather than the .doc/.xls/.ppt or .docx/.xlsx/.pptx inside it, so these + # must never be resolved from the sample alone. + CONTAINER_MIME_TYPES = frozenset({"application/x-ole-storage", "application/zip"}) def __init__( self, @@ -1187,6 +1200,44 @@ def load_file(self, input_file_path: str) -> tuple[str, BytesIO]: return os.path.basename(input_file_path), file_stream + @classmethod + def _detect_uploaded_file_mime_type(cls, file: UploadedFile) -> str: + """Detect an uploaded file's MIME type from its own bytes. + + The multipart Content-Type is supplied by the caller and never verified, + so it cannot be used to decide what is allowed into API storage. + """ + sample = file.read(cls.MIME_DETECT_CHUNK_SIZE) + file.seek(0) + if not sample: + # libmagic reports "application/x-empty" here, which would reject the + # file as an unsupported type. An empty upload is a distinct failure + # and is reported as such once staging hands off, so let it pass. + return AllowedFileTypes.OCTET_STREAM.value + + mime_type = magic.from_buffer(sample, mime=True) + if mime_type not in cls.CONTAINER_MIME_TYPES: + return mime_type + return cls._detect_container_mime_type(file, fallback=mime_type) + + @classmethod + def _detect_container_mime_type(cls, file: UploadedFile, fallback: str) -> str: + """Resolve a container format by classifying the file in full. + + Django spills uploads over FILE_UPLOAD_MAX_MEMORY_SIZE to disk, so this + hands libmagic the path when there is one and only buffers the whole + upload for the in-memory case, where that ceiling already bounds it. + """ + temporary_file_path = getattr(file, "temporary_file_path", None) + if temporary_file_path is not None: + return magic.from_file(temporary_file_path(), mime=True) + + content = file.read() + file.seek(0) + if not content: + return fallback + return magic.from_buffer(content, mime=True) + @classmethod def add_input_file_to_api_storage( cls, @@ -1228,30 +1279,43 @@ def add_input_file_to_api_storage( file_name = file.name destination_path = os.path.join(api_storage_dir, file_name) - mime_type = file.content_type - logger.info(f"Detected MIME type: {mime_type} for file {file_name}") - if not mime_type: - logger.info( - f"MIME type not found for file {file_name}, using default MIME type: {AllowedFileTypes.OCTET_STREAM.value}" + try: + mime_type = cls._detect_uploaded_file_mime_type(file) + except Exception: + # Detection reads the upload, so a broken stream raises here. Fail + # this one file instead of the whole request, and say that detection + # failed rather than blaming the file's type - an I/O fault and an + # unsupported format need different follow-ups. + log_message = ( + f"Rejecting file '{file_name}': could not determine its type" ) - mime_type = AllowedFileTypes.OCTET_STREAM.value + logger.exception(log_message) + workflow_log.log_error(logger=logger, message=log_message) + ResultCacheUtils.update_api_results( + workflow_id=workflow_id, + execution_id=execution_id, + api_result=FileExecutionResult(file=file_name, error=log_message), + ) + continue + + logger.info(f"Detected MIME type: {mime_type} for file {file_name}") 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, + log_message = ( + f"Rejecting file '{file_name}' with unsupported MIME type " + f"'{mime_type}'" + ) + workflow_log.log_error(logger=logger, message=log_message) + # Rejected files are never dispatched, so nothing downstream will + # report on them - surface the failure in the API response here. + ResultCacheUtils.update_api_results( + workflow_id=workflow_id, + execution_id=execution_id, + api_result=FileExecutionResult( + file=file_name, + error=log_message, + ), ) - file_hashes.update({file_name: file_hash}) continue file_system = FileSystem(FileStorageType.API_EXECUTION) diff --git a/backend/workflow_manager/endpoint_v2/tests/test_api_storage_mime_validation.py b/backend/workflow_manager/endpoint_v2/tests/test_api_storage_mime_validation.py new file mode 100644 index 0000000000..8bd5412d32 --- /dev/null +++ b/backend/workflow_manager/endpoint_v2/tests/test_api_storage_mime_validation.py @@ -0,0 +1,243 @@ +"""MIME validation for files staged into API storage. + +``SourceConnector.add_input_file_to_api_storage`` is the single funnel through +which API-deployment uploads reach the API storage bucket, so an unsupported +file has to be rejected here or it reaches the extraction step and fails there +with an error that does not name the real cause. + +Unit tests: the real classmethod runs with its DB/storage-touching +collaborators patched on the imported module, so no database is needed. MIME +detection itself is deliberately *not* patched — sniffing the bytes with +libmagic is the behaviour under test. +""" + +from unittest import mock +from unittest.mock import MagicMock + +import pytest +from django.core.files.uploadedfile import SimpleUploadedFile + +import workflow_manager.endpoint_v2.source as src_mod +from workflow_manager.endpoint_v2.constants import ApiDeploymentResultStatus +from workflow_manager.endpoint_v2.source import SourceConnector + +# Bytes chosen from what libmagic actually reports (verified against the pinned +# python-magic): a PDF header sniffs application/pdf, an HTML document sniffs +# text/html, which is absent from AllowedFileTypes. +PDF_BYTES = b"%PDF-1.7\n1 0 obj\n<< /Type /Catalog >>\nendobj\n" +HTML_BYTES = b"hello" + + +API_STORAGE_DIR = "/api-storage/exec-1" + + +@pytest.fixture +def collaborators(): + """Patch everything the staging loop touches except MIME detection.""" + with ( + mock.patch.multiple( + src_mod, + UserContext=mock.DEFAULT, + WorkflowLog=mock.DEFAULT, + Workflow=mock.DEFAULT, + FileSystem=mock.DEFAULT, + FileHistoryHelper=mock.DEFAULT, + ResultCacheUtils=mock.DEFAULT, + ) as mocks, + mock.patch.object( + SourceConnector, + "get_api_storage_dir_path", + return_value=API_STORAGE_DIR, + ), + ): + storage = MagicMock() + mocks["FileSystem"].return_value.get_file_storage.return_value = storage + mocks["storage"] = storage + yield mocks + + +def _upload(name: str, content: bytes, declared: str) -> SimpleUploadedFile: + """An uploaded file whose declared Content-Type may not match its bytes.""" + return SimpleUploadedFile(name, content, content_type=declared) + + +def _stage(files): + return SourceConnector.add_input_file_to_api_storage( + pipeline_id="pipe-1", + workflow_id="wf-1", + execution_id="exec-1", + file_objs=files, + ) + + +def _staged_names(storage: MagicMock) -> set[str]: + """File names that actually had bytes written to API storage.""" + return { + call.kwargs["path"].rsplit("/", 1)[-1] for call in storage.write.call_args_list + } + + +def test_supported_file_is_staged(collaborators) -> None: + """A real PDF is staged and returned for dispatch.""" + result = _stage([_upload("doc.pdf", PDF_BYTES, "application/pdf")]) + + assert set(result) == {"doc.pdf"} + assert result["doc.pdf"].mime_type == "application/pdf" + assert _staged_names(collaborators["storage"]) == {"doc.pdf"} + + +def test_unsupported_bytes_rejected_despite_supported_declared_type( + collaborators, +) -> None: + """The declared Content-Type must not decide what reaches the bucket. + + An HTML file announced as application/pdf satisfies any header-based check, + so only sniffing the bytes keeps it out. + """ + result = _stage([_upload("evil.pdf", HTML_BYTES, "application/pdf")]) + + # Never dispatched... + assert result == {} + # ...and never written to the bucket. + collaborators["storage"].write.assert_not_called() + + +def test_rejection_is_reported_to_the_caller(collaborators) -> None: + """A rejected file gets its own failed entry in the API response.""" + _stage([_upload("evil.pdf", HTML_BYTES, "application/pdf")]) + + collaborators["ResultCacheUtils"].update_api_results.assert_called_once() + api_result = collaborators["ResultCacheUtils"].update_api_results.call_args.kwargs[ + "api_result" + ] + assert api_result.file == "evil.pdf" + # The message has to name the offending type, not a downstream symptom. + assert "text/html" in api_result.error + assert api_result.status == ApiDeploymentResultStatus.FAILED + + +def test_missing_declared_type_falls_back_to_sniffed_type(collaborators) -> None: + """A supported file with no declared Content-Type is still staged. + + The recorded type comes from the bytes, so an absent header neither blocks + the file nor degrades it to application/octet-stream. + """ + result = _stage([_upload("doc.pdf", PDF_BYTES, "")]) + + assert result["doc.pdf"].mime_type == "application/pdf" + + +def test_supported_files_survive_a_rejected_sibling(collaborators) -> None: + """One bad file does not fail the whole request.""" + result = _stage( + [ + _upload("good.pdf", PDF_BYTES, "application/pdf"), + _upload("evil.pdf", HTML_BYTES, "application/pdf"), + ] + ) + + assert set(result) == {"good.pdf"} + assert _staged_names(collaborators["storage"]) == {"good.pdf"} + + +def _ole2_like(total_size: int) -> bytes: + """An OLE2 compound file whose format markers sit past the sample window. + + libmagic resolves .doc/.xls/.ppt through the OLE2 directory sector, which + lives at the end of the file. Only the container signature is visible in the + leading bytes, which is exactly the shape that made these files unstageable. + """ + header = b"\xd0\xcf\x11\xe0\xa1\xb1\x1a\xe1" + b"\x00" * 504 + return header + b"\x00" * (total_size - len(header)) + + +def test_container_prefix_triggers_a_full_file_sniff(collaborators) -> None: + """A container type seen in the sample must not decide the verdict alone. + + Pins the regression directly: an OLE2 upload sniffs application/x-ole-storage + from its first bytes, which is absent from AllowedFileTypes, so resolving from + the sample alone rejects every legacy Office file bigger than the window. + + The sniff results are stubbed because libmagic's container reporting differs + between builds; what must hold everywhere is that an inconclusive sample is + escalated to the full file instead of being treated as a verdict. + """ + ole_bytes = _ole2_like(SourceConnector.MIME_DETECT_CHUNK_SIZE * 4) + sniffs = ["application/x-ole-storage", "application/msword"] + + with mock.patch.object(src_mod.magic, "from_buffer", side_effect=sniffs) as sniff: + result = _stage([_upload("legacy.doc", ole_bytes, "application/msword")]) + + # The sample verdict was inconclusive, so the whole file was classified... + assert sniff.call_count == 2 + assert len(sniff.call_args_list[0].args[0]) == SourceConnector.MIME_DETECT_CHUNK_SIZE + assert len(sniff.call_args_list[1].args[0]) == len(ole_bytes) + # ...and the answer from the full file is what decides. + assert set(result) == {"legacy.doc"} + assert result["legacy.doc"].mime_type == "application/msword" + + +def test_container_still_rejected_when_the_full_file_is_unsupported( + collaborators, +) -> None: + """The full-file re-sniff widens the evidence, not the allow-list.""" + ole_bytes = _ole2_like(SourceConnector.MIME_DETECT_CHUNK_SIZE * 4) + sniffs = ["application/x-ole-storage", "application/x-dosexec"] + + with mock.patch.object(src_mod.magic, "from_buffer", side_effect=sniffs): + result = _stage([_upload("legacy.doc", ole_bytes, "application/msword")]) + + assert result == {} + collaborators["storage"].write.assert_not_called() + + +def test_container_upload_is_not_consumed_by_detection(collaborators) -> None: + """Reading the whole file to classify it must still leave it stageable.""" + ole_bytes = _ole2_like(SourceConnector.MIME_DETECT_CHUNK_SIZE * 4) + sniffs = ["application/x-ole-storage", "application/msword"] + + with mock.patch.object(src_mod.magic, "from_buffer", side_effect=sniffs): + _stage([_upload("legacy.doc", ole_bytes, "application/msword")]) + + written = b"".join( + call.kwargs["data"] for call in collaborators["storage"].write.call_args_list + ) + assert written == ole_bytes + + +def test_undetectable_file_fails_alone(collaborators) -> None: + """A stream that cannot be read fails its own file, not the whole request.""" + with mock.patch.object( + SourceConnector, + "_detect_uploaded_file_mime_type", + side_effect=[OSError("stream is gone"), "application/pdf"], + ): + result = _stage( + [ + _upload("broken.pdf", PDF_BYTES, "application/pdf"), + _upload("good.pdf", PDF_BYTES, "application/pdf"), + ] + ) + + assert set(result) == {"good.pdf"} + api_result = collaborators["ResultCacheUtils"].update_api_results.call_args.kwargs[ + "api_result" + ] + assert api_result.file == "broken.pdf" + # An I/O fault and an unsupported format need different follow-ups, so the + # message must not blame the file's type. + assert "could not determine its type" in api_result.error + + +def test_empty_upload_is_staged_rather_than_called_unsupported(collaborators) -> None: + """An empty file must reach the downstream empty-file error, not a type error. + + libmagic calls zero bytes application/x-empty, which is absent from + AllowedFileTypes; without the short-circuit an empty upload would be reported + as an unsupported type, which names the wrong cause. + """ + result = _stage([_upload("empty.pdf", b"", "application/pdf")]) + + assert set(result) == {"empty.pdf"} + assert result["empty.pdf"].mime_type == "application/octet-stream" + collaborators["ResultCacheUtils"].update_api_results.assert_not_called() diff --git a/backend/workflow_manager/workflow_v2/execution.py b/backend/workflow_manager/workflow_v2/execution.py index 213c81302b..d84bccfa90 100644 --- a/backend/workflow_manager/workflow_v2/execution.py +++ b/backend/workflow_manager/workflow_v2/execution.py @@ -387,6 +387,46 @@ def update_execution_err(execution_id: str, err_msg: str = "") -> WorkflowExecut except WorkflowExecution.DoesNotExist: logger.error(f"execution doesn't exist {execution_id}") + @staticmethod + def update_execution_completed( + execution_id: str, total_files: int = 0, failed_files: int = 0 + ) -> WorkflowExecution | None: + """Terminalise an execution that finished without any work to dispatch. + + The counters must be written alongside the status: a terminal row whose + failed_files is NULL reads as a clean success to is_failure_run() and to + run history, which would hide a run whose files were all rejected. + + Returns the row as persisted, so callers can see whether the status + actually changed rather than assuming it did. + """ + try: + execution = WorkflowExecution.objects.get(pk=execution_id) + # Same reason as update_execution_err: on the PG transport the model + # method owns the terminal-one-way guard, so a row the callback already + # finalized cannot be reverted. The legacy transport has no such guard. + execution.update_execution(status=ExecutionStatus.COMPLETED) + execution.total_files = total_files + execution.successful_files = 0 + execution.failed_files = failed_files + # Field-scoped, matching update_execution, so this cannot clobber the + # status write or anything a concurrent writer touched. + execution.save( + update_fields=[ + "total_files", + "successful_files", + "failed_files", + "modified_at", + ] + ) + # The guard may have refused the status change without raising; re-read + # so the returned row reflects what is actually stored. + execution.refresh_from_db() + return execution + except WorkflowExecution.DoesNotExist: + logger.error(f"execution doesn't exist {execution_id}") + return None + @staticmethod def update_execution_task(execution_id: str, task_id: str) -> None: try: diff --git a/workers/api-deployment/tasks.py b/workers/api-deployment/tasks.py index a07ef7b7bd..e913b0441d 100644 --- a/workers/api-deployment/tasks.py +++ b/workers/api-deployment/tasks.py @@ -221,7 +221,39 @@ def _unified_api_execution( converted_files = FileProcessingUtils.convert_file_hash_data(hash_values_of_files) if not converted_files: - logger.warning("No valid files to process after conversion") + # convert_file_hash_data swallows per-file errors and returns only what + # converted, so {} means "nothing was dispatched" OR "every file failed + # to convert". Reporting the second as COMPLETED would turn a total + # failure into a silent success with no results and no error. + if hash_values_of_files: + error_message = ( + f"None of the {len(hash_values_of_files)} dispatched files could " + "be converted for processing" + ) + logger.error(error_message) + api_client.update_workflow_execution_status( + execution_id=execution_id, + status=ExecutionStatus.ERROR.value, + error_message=error_message, + total_files=len(hash_values_of_files), + successful_files=0, + failed_files=len(hash_values_of_files), + ) + return { + "execution_id": execution_id, + "status": "ERROR", + "message": error_message, + "files_processed": 0, + } + + logger.warning("No files dispatched for this execution") + # Returning COMPLETED is not enough: without this write the row keeps + # whatever status it was dispatched with, and the caller polls forever. + api_client.update_workflow_execution_status( + execution_id=execution_id, + status=ExecutionStatus.COMPLETED.value, + total_files=0, + ) return { "execution_id": execution_id, "status": "COMPLETED",