UN-2733 [FIX] Allow file processing to continue on cleanup related errors (Zipstack#1503)

chandrasekharan-zipstack · web-flow · commit a032813bd000 · 2025-08-25T14:45:44.000+05:30
* UN-2733 Error handling fix for file processing related celery task

* Minor improvement in logging traceback too
diff --git a/backend/workflow_manager/workflow_v2/file_execution_tasks.py b/backend/workflow_manager/workflow_v2/file_execution_tasks.py
@@ -557,7 +557,7 @@ def _process_file(
                 result=result,
                 workflow_file_execution=None,
                 error=error_msg,
-                is_api=destination.is_api,
+                is_api=destination.is_api if destination else False,
                 destination=destination,
             )
         except Exception as error:
@@ -578,7 +578,7 @@ def _process_file(
                 result=result,
                 workflow_file_execution=workflow_file_execution,
                 error=error_msg,
-                is_api=destination.is_api,
+                is_api=destination.is_api if destination else False,
                 destination=destination,
             )
 
@@ -1136,10 +1136,17 @@ def _build_final_result(
             )
 
         if destination:
-            logger.info(
-                f"Deleting file execution directory for file: '{file_hash.file_name}'"
-            )
-            destination.delete_file_execution_directory()
+            try:
+                logger.info(
+                    f"Deleting file execution directory for file: '{file_hash.file_name}'"
+                )
+                destination.delete_file_execution_directory()
+            except Exception as cleanup_error:
+                logger.exception(
+                    f"[Execution ID: {workflow_execution.id}, File Execution ID: {workflow_file_execution.id}] "
+                    f"Failed to delete file execution directory for '{file_hash.file_name}': {cleanup_error}. "
+                    f"This is non-critical and needs to be cleaned up later."
+                )
 
         status = (
             FileExecutionStageStatus.SUCCESS
diff --git a/unstract/core/src/unstract/core/file_execution_tracker.py b/unstract/core/src/unstract/core/file_execution_tracker.py
@@ -191,10 +191,12 @@ def get_cache_key(self, execution_id: str, file_execution_id: str) -> str:
     def set_data(self, data: FileExecutionData, ttl_in_second: int | None = None) -> None:
         data.validate()
         key = self.get_cache_key(data.execution_id, data.file_execution_id)
-        logger.info(f"Setting file execution data for {key}: {data}")
+        logger.info(f"Setting file execution data for '{key}': {data}")
         self.redis_client.hset(key, mapping=data.to_serializable())
         ttl = ttl_in_second or self.CACHE_TTL_IN_SECOND
-        logger.info(f"Setting file execution data for {key} to expire in {ttl} seconds")
+        logger.info(
+            f"Setting file execution data for '{key}' to expire in '{ttl}' seconds"
+        )
         self.redis_client.expire(key, ttl)
 
     def exists(self, execution_id: str, file_execution_id: str) -> bool:
@@ -229,7 +231,7 @@ def update_stage_status(
             execution_id=execution_id,
             file_execution_id=file_execution_id,
         )
-        logger.info(f"Existing data for {key}: {existing_data}")
+        logger.info(f"Existing data for '{key}': {existing_data}")
         if not existing_data:
             raise FileExecutionTrackerNotFound()
 
