fix(batch): survive compute-node loss after a task completes

CHANGELOG.md

@@ -19,6 +19,11 @@ Versioning follows the Docker image tags defined in the CI workflows (see [.gith
 
 - **A reused Batch job stayed pinned to the pool that created it** — job ids default to the configured pool id, and `create_job` only re-enabled an existing job, never re-pointing it. Capacity-aware spillover was therefore silently ineffective, and once a pool was renamed or deleted every task queued into a job bound to a pool that no longer existed. Job ids are now derived from the pool a task is routed to (one job per pool), so spillover works even while another task is running; `create_job` falls back to a pool-scoped job rather than failing the submission. Environments with a single candidate pool keep their existing job ids unchanged. Full design in [`spec/features/batch-pool-job-binding/`](spec/features/batch-pool-job-binding).
 - **Missing Batch settings now fail fast** — the runner validates its configuration before the first Batch call and names the specific application setting that is unset, instead of surfacing an opaque Azure error from deep inside pool creation.
+
+- **Losing the Batch compute node failed image layers that had already succeeded** — imagery preprocessing reads `imagery_manifest.json` and `imagery_friendly.log` back off the node that ran the task, but those APIs are served *by the node*, and autoscale pools deallocate it the moment the task completes (`$NodeDeallocationOption = taskcompletion`; low-priority nodes can also be preempted). The resulting `NodeNotReady` is an HTTP 409, which the runner's 5xx-only retry never covered, so a completed layer was marked FAILED and — because the queue trigger overwrote `statusMessage` with the raw SDK exception — its entire progress history was replaced by an unreadable object dump. The runner now retries transient node errors (`NodeNotReady`, `NodeStateInvalid`), treats a vanished node (`NodeNotFound`) as "file unavailable" rather than a failure, and never fails a workload on post-task cleanup; imagery falls back to the copy Azure Batch already uploaded to blob storage on task completion, and its progress log is now uploaded too (`logs/`) so it survives node loss. Failure messages are appended rather than assigned and are rendered as `NodeNotReady: <cause>`. Applies to all Batch workloads (imagery, training, inference, artifacts, embedding). Full design in [`spec/features/batch-node-loss-resilience/`](spec/features/batch-node-loss-resilience).
+
+  > **Operator note:** the `logs/` upload only applies to newly submitted tasks, so image layers that already failed this way must be re-run.
+
 - **Pool creation whitelisted only the training image** — pools created by the runner now list both the training and imageryprep images, matching [`infra/modules/batchPool.bicep`](infra/modules/batchPool.bicep).
 
 ### Changed

api/hastefuncqueues/README.md

@@ -54,12 +54,20 @@ Each pool runs tasks as Docker containers pulled from Azure Container Registry u
 
 Preprocesses a newly uploaded geospatial image layer. On each invocation:
 - **PENDING** → submits a Batch task via `ImageryPostProcessor`; task runs `prepare-imagery` CLI inside the container
-- **IN_PROGRESS** → reads `imagery_friendly.log` from the task working directory for progress, then re-queues
+- **IN_PROGRESS** → reads `imagery_friendly.log` for progress, then re-queues
 - **COMPLETED** → reads `imagery_manifest.json` for output paths (mosaics, COGs, building footprints, valid-area mask); generates label project files, converts to GeoJSON, stores artifacts
 - **FAILED** → saves the layer with a FAILED status and error message
 
 If the layer was deleted before processing starts, the message is silently skipped.
 
+Both files are read from the compute node first and, if that node is no longer
+able to serve them, from the copy Azure Batch uploaded to blob storage on task
+completion (`outputs/` and `logs/` under `<projectHash>/<taskId>/`). On
+autoscale pools the node is deallocated the moment the task completes — and
+low-priority nodes can be preempted — so the node-local copy is frequently gone
+by the time the trigger looks for it. The manifest is required (a layer with no
+manifest in either place fails); the progress log is best-effort.
+
 ---
 
 ### GetCreateModelRunQueueMessage

api/hastefuncqueues/function_app.py

@@ -29,6 +29,7 @@
 from hastegeo.core.processors.stats import StatsPostProcessor
 from hastegeo.core.processors.train import TrainPostprocessor
 from hastegeo.core.utils.data import convert_json_to_geojson
+from hastegeo.core.utils.errors import describe_exception
 from hastegeo.core.utils.logs import Logger
 from hastegeo.core.utils.metadata import MetadataUtils
 from pydantic import ValidationError  # type: ignore
@@ -215,7 +216,13 @@ async def GetProcessImageLayerQueueMessage(msg: func.QueueMessage) -> None:
         try:
             if "output" in locals():
                 output.status = config.get_status_types().FAILED.value
-                output.statusMessage = str(e)
+                # Append rather than assign: the status dialog shows this
+                # field, and overwriting it discards the whole progress
+                # history the user needs to see what actually ran.
+                output.statusMessage = MetadataUtils.append_status_message(
+                    output.statusMessage,
+                    f"Image layer processing failed: {describe_exception(e)}",
+                )
                 await asyncio.to_thread(
                     MetadataProcessor(
                         data_type=config.get_metadata_types().IMAGELAYER.value,
@@ -226,7 +233,10 @@ async def GetProcessImageLayerQueueMessage(msg: func.QueueMessage) -> None:
                 )
             elif "image_data" in locals():
                 image_data.status = config.get_status_types().FAILED.value
-                image_data.statusMessage = str(e)
+                image_data.statusMessage = MetadataUtils.append_status_message(
+                    image_data.statusMessage,
+                    f"Image layer processing failed: {describe_exception(e)}",
+                )
                 await asyncio.to_thread(
                     MetadataProcessor(
                         data_type=config.get_metadata_types().IMAGELAYER.value,
@@ -408,7 +418,7 @@ async def GetCreateModelRunQueueMessage(msg: func.QueueMessage) -> None:
             model_data.status = config.get_status_types().FAILED.value
             model_data.statusMessage = MetadataUtils.append_status_message(
                 model_data.statusMessage,
-                f"Training job failed: {str(e)}",
+                f"Training job failed: {describe_exception(e)}",
             )
             output = model_data
             await asyncio.to_thread(
@@ -571,7 +581,7 @@ async def GetRunEmbeddingQueueMessage(msg: func.QueueMessage) -> None:
                 model_data.status = config.get_status_types().FAILED.value
                 model_data.statusMessage = MetadataUtils.append_status_message(
                     model_data.statusMessage,
-                    f"Embedding job failed: {str(e)}",
+                    f"Embedding job failed: {describe_exception(e)}",
                 )
                 await asyncio.to_thread(
                     MetadataProcessor(
@@ -741,7 +751,7 @@ async def GetRunInferenceQueueMessage(msg: func.QueueMessage) -> None:
         model_data.inferenceStatusMessage = (
             MetadataUtils.append_status_message(
                 model_data.inferenceStatusMessage or "",
-                f"Inference job failed: {str(e)}",
+                f"Inference job failed: {describe_exception(e)}",
             )
         )
         output = model_data

docs/api/hastefuncqueues.md

@@ -53,12 +53,20 @@ Each pool runs tasks as Docker containers pulled from Azure Container Registry u
 
 Preprocesses a newly uploaded geospatial image layer. On each invocation:
 - **PENDING** → submits a Batch task via `ImageryPostProcessor`; task runs `prepare-imagery` CLI inside the container
-- **IN_PROGRESS** → reads `imagery_friendly.log` from the task working directory for progress, then re-queues
+- **IN_PROGRESS** → reads `imagery_friendly.log` for progress, then re-queues
 - **COMPLETED** → reads `imagery_manifest.json` for output paths (mosaics, COGs, building footprints, valid-area mask); generates label project files, converts to GeoJSON, stores artifacts
 - **FAILED** → saves the layer with a FAILED status and error message
 
 If the layer was deleted before processing starts, the message is silently skipped.
 
+Both files are read from the compute node first and, if that node is no longer
+able to serve them, from the copy Azure Batch uploaded to blob storage on task
+completion (`outputs/` and `logs/` under `<projectHash>/<taskId>/`). On
+autoscale pools the node is deallocated the moment the task completes — and
+low-priority nodes can be preempted — so the node-local copy is frequently gone
+by the time the trigger looks for it. The manifest is required (a layer with no
+manifest in either place fails); the progress log is best-effort.
+
 ---
 
 ### GetCreateModelRunQueueMessage

hastelib/src/hastegeo/core/processors/imagery.py

@@ -9,6 +9,7 @@
 from ..config import ArtifactTypes, Config
 from ..data_layer.unified import UnifiedDataLayer
 from ..models.projects import ImageLayer, ImageryPreprocessJob
+from ..utils.blob import fetch_url_text
 from ..utils.data import extract_from_url
 from ..utils.logs import Logger
 from ..utils.metadata import MetadataUtils
@@ -346,7 +347,12 @@ def _execute_image_preprocess(self):
             task_id=task_id,
             output_prefix=imagery_output_prefix,
             resource_files_for_upload=imagery_input_files,
-            file_pattern=f"${BATCH_JOB_WORKDIR}/outputs/*.*",
+            file_pattern=[
+                f"${BATCH_JOB_WORKDIR}/outputs/*.*",
+                # Progress log, so it survives the node being deallocated or
+                # preempted once the task completes.
+                f"${BATCH_JOB_WORKDIR}/logs/*.*",
+            ],
             command=command,
             # TODO: maybe this needs to be encapsulated in the batch runner and not be part of the processor
             image_name=self.config.get_azure_batch_config()[
@@ -376,12 +382,59 @@ def _execute_image_preprocess(self):
         )
         return self.image_data
 
-    def _get_image_preprocess_logs(self):
+    def _read_task_output(self, filename: str):
+        """Return the text of a task output file, or ``None``.
+
+        Reads the copy on the compute node first, then falls back to the copy
+        Azure Batch uploaded to blob storage on task completion. The node-local
+        copy disappears as soon as the node is deallocated or preempted — which
+        on autoscale pools happens the moment the task completes — so the blob
+        copy is often the only one left by the time we look.
+        """
         content = self.runner.get_filecontent_from_task(
             job_id=self.image_data.preprocessJob.jobId,
             task_id=self.image_data.preprocessJob.taskId,
-            filename="imagery_friendly.log",
+            filename=filename,
         )
+        if content:
+            return content
+
+        task_id = self.image_data.preprocessJob.taskId
+        self.logger.info(
+            "%s not available from the node for task %s; "
+            "falling back to the uploaded copy in storage.",
+            filename,
+            task_id,
+        )
+        try:
+            url = self.storage.get_file_remote_path(
+                identifier=filename,
+                extra_partition_keys=task_id,
+                data_format=os.path.splitext(filename)[1].strip("."),
+            )
+            content = fetch_url_text(url)
+        except Exception as e:
+            # The fallback must never replace the original reason the file was
+            # unreadable — callers decide whether a miss is fatal.
+            self.logger.warning(
+                "Fallback read of %s for task %s failed: %s",
+                filename,
+                task_id,
+                e,
+            )
+            return None
+        if not content:
+            self.logger.warning(
+                "%s for task %s is not available from the node or storage.",
+                filename,
+                task_id,
+            )
+        return content
+
+    def _get_image_preprocess_logs(self):
+        # Best-effort: the progress log is a nicety for the status dialog, so a
+        # node that went away must not fail an otherwise successful layer.
+        content = self._read_task_output("imagery_friendly.log")
         logs = []
         if content:
             try:
@@ -419,11 +472,7 @@ def _update_results_from_job(self):
         """
         Update the image layer object with imagery processing results from the processed manifest file
         """
-        content = self.runner.get_filecontent_from_task(
-            job_id=self.image_data.preprocessJob.jobId,
-            task_id=self.image_data.preprocessJob.taskId,
-            filename="imagery_manifest.json",
-        )
+        content = self._read_task_output("imagery_manifest.json")
         if not content:
             raise FileNotFoundError(
                 f"Processed manifest file not found for image layer id: {self.image_data.imageLayerId}"