map CachedArtifactNotFoundError to PreviousStepStillProcessingError (#…

…2093)

map CachedArtifactNotFoundError to customError

libs/libcommon/src/libcommon/exceptions.py

@@ -111,6 +111,7 @@ def as_response(self) -> ErrorResponse:
     "ParquetResponseEmptyError",
     "PreviousStepFormatError",
     "PreviousStepStatusError",
+    "PreviousStepStillProcessingError",
     "ResponseAlreadyComputedError",
     "RowsPostProcessingError",
     "SplitsNamesError",
@@ -441,6 +442,13 @@ def __init__(self, message: str, cause: Optional[BaseException] = None):
         super().__init__(message, HTTPStatus.INTERNAL_SERVER_ERROR, "PreviousStepStatusError", cause, False)
 
 
+class PreviousStepStillProcessingError(CacheableError):
+    """The previous steps are still being processed."""
+
+    def __init__(self, message: str, cause: Optional[BaseException] = None):
+        super().__init__(message, HTTPStatus.INTERNAL_SERVER_ERROR, "PreviousStepStillProcessingError", cause, False)
+
+
 class ResponseAlreadyComputedError(CacheableError):
     """The response has been already computed by another job runner."""
 

services/worker/src/worker/job_manager.py

@@ -13,6 +13,7 @@
     DatasetScriptError,
     JobManagerCrashedError,
     JobManagerExceededMaximumDurationError,
+    PreviousStepStillProcessingError,
     ResponseAlreadyComputedError,
     TooBigContentError,
     UnexpectedError,
@@ -21,6 +22,7 @@
 from libcommon.processing_graph import ProcessingGraph, ProcessingStep
 from libcommon.simple_cache import (
     CachedArtifactError,
+    CachedArtifactNotFoundError,
     CacheEntryDoesNotExistError,
     get_response_without_content_params,
 )
@@ -174,6 +176,10 @@ def process(
                         "The computed response content exceeds the supported size in bytes"
                         f" ({self.worker_config.content_max_bytes})."
                     )
+            except CachedArtifactNotFoundError as err:
+                raise PreviousStepStillProcessingError(
+                    message="The previous steps are still being processed", cause=err
+                )
             finally:
                 # ensure the post_compute hook is called even if the compute raises an exception
                 self.job_runner.post_compute()