diff --git a/eogrow/pipelines/download_batch.py b/eogrow/pipelines/download_batch.py index 4a0bc515..661d667a 100644 --- a/eogrow/pipelines/download_batch.py +++ b/eogrow/pipelines/download_batch.py @@ -49,7 +49,7 @@ def _retry_on_404(func: Callable[P, T]) -> Callable[P, T]: @wraps(func) def retrying_func(*args: P.args, **kwargs: P.kwargs) -> T: - for wait_time in [0, 10, 100]: + for wait_time in [0, 10, 20, 100]: time.sleep(wait_time) # if we start monitoring too soon we might hit a 404 try: return func(*args, **kwargs) @@ -178,6 +178,7 @@ def run_procedure(self) -> tuple[list[str], list[str]]: batch_request = self._create_or_collect_batch_request() user_action = self._trigger_user_action(batch_request) + self._wait_for_sh_db_sync(batch_request) if user_action is BatchUserAction.ANALYSE or ( user_action is BatchUserAction.START and batch_request.status is not BatchRequestStatus.ANALYSIS_DONE @@ -308,6 +309,11 @@ def _trigger_user_action(self, batch_request: BatchRequest) -> BatchUserAction: LOGGER.info("Didn't trigger batch job because current batch request status is %s", status) return BatchUserAction.NONE + @_retry_on_404 + def _wait_for_sh_db_sync(self, batch_request: BatchRequest) -> None: + """Wait for SH read/write databases to sync.""" + self.batch_client.get_request(batch_request) + def cache_batch_area_manager_grid(self, request_id: str) -> None: """This method ensures that area manager caches batch grid into the storage.""" if self.area_manager.config.batch_id and self.area_manager.config.batch_id != request_id: diff --git a/eogrow/utils/pipeline_chain.py b/eogrow/utils/pipeline_chain.py index f876c372..e9b052a1 100644 --- a/eogrow/utils/pipeline_chain.py +++ b/eogrow/utils/pipeline_chain.py @@ -40,8 +40,8 @@ def validate_pipeline_chain(pipeline_chain: list[RawConfig]) -> None: def run_pipeline_chain(pipeline_chain: list[RawConfig]) -> None: for run_config in pipeline_chain: run_schema = PipelineRunSchema.parse_obj(run_config) - runner = _pipeline_runner.options(**run_schema.pipeline_resources) # type: ignore[attr-defined] - ray.get(runner.remote(run_schema.pipeline_config)) + runner = _pipeline_runner.options(**run_schema.pipeline_resources) + ray.get(runner.remote(run_schema.pipeline_config)) # type: ignore [arg-type] @ray.remote(max_retries=0)