From 80dab965a36c2d51f0b321cc04187d092fcbe7e0 Mon Sep 17 00:00:00 2001 From: jQuinRivero Date: Wed, 12 Aug 2026 18:01:56 -0300 Subject: [PATCH 1/2] fix(batch): survive compute-node loss after a task completes Processing new imagery on dev1 failed with the raw Azure SDK error dump as the image layer's entire status history: Request encountered an exception. Code: NodeNotReady Message: {'additional_properties': {}, 'lang': 'en-US', 'value': 'Node is not able to perform the requested operations in its current state... The Batch task had succeeded and its outputs were already in blob storage. The layer was still marked FAILED. ImageryPostProcessor reacts to a task reaching a terminal state by reading files back off the node that ran it -- imagery_manifest.json, imagery_friendly.log, then a working-directory delete. Those file APIs (file.list/get/delete_from_task) are served *by the compute node*, and the node is being torn down at exactly that moment: infra/modules/batchPool.bicep sets `$NodeDeallocationOption = taskcompletion`, and the shared-dev pools dev1 targets are autoscale with minNodes=0 on low-priority nodes, which can also be preempted. A deallocating, rebooting or preempted node answers NodeNotReady. Three properties then turned a transient race into a permanent failure: - NodeNotReady is HTTP 409, and is_server_error only retried 5xx, so it was never retried. - The imagery queue trigger did `statusMessage = str(e)`, replacing the whole progress history with the SDK repr -- the unreadable dialog above. - host.json sets maxDequeueCount: 1, so the message is never redelivered. Classify node errors and retry the transient ones. NodeNotReady and NodeStateInvalid describe a node that is mid-transition and may still answer, so they join the retry predicate; NodeNotFound describes a node that is gone, so it is surfaced as unavailable instead. The retry budget is unchanged (5 attempts, exponential 4-10s) and reraise=True was added so an exhausted budget still yields the BatchErrorException rather than tenacity's RetryError, keeping it classifiable by callers. Degrade instead of failing. get_filecontent_from_task now returns None when the node cannot serve a file -- matching its existing "file not found" contract, which every caller already branches on -- and cleanup_task skips the working-directory delete while still disabling the job (task retention_time=P2D reclaims the disk). Every other Batch error still propagates. This covers all five workloads: imagery, training, inference, artifacts and embedding. Recover the outputs from blob. Batch already uploads outputs/*.* on task completion, so the manifest was sitting at // the whole time -- the same path _generate_imagery_url already resolves in production. ImageryPostProcessor._read_task_output reads the node copy first and falls back to that blob copy. The manifest stays required; the progress log is best-effort. A failing fallback returns None rather than raising, so it can never mask the original reason the node read failed. To make the log recoverable at all, add_task now accepts file_pattern as a string or a list, emitting one OutputFile per pattern, and imagery submits both outputs/*.* and logs/*.* -- the log is written to logs/, which no pattern previously covered, so it only ever existed on the node. A single blanket **/* was rejected: it would also upload the raw downloaded imagery. Stop destroying the status history. The imagery trigger appends via append_status_message instead of assigning, and renders the cause through a new hastegeo.core.utils.errors.describe_exception, which reduces an Azure-style error to "NodeNotReady: Node is not able to perform the requested operations in its current state" and strips the RequestId/Time trailer. It matches the error shape (.error.code / .error.message.value) rather than importing azure.batch, so utils stays a leaf package, and it lives in hastegeo rather than function_app.py per the repository's function-app boundary rule. The train, embedding and inference triggers already appended; they now use the same formatter. The pool configuration is deliberately untouched. This makes the race survivable, not impossible -- preemption of a spot node can happen regardless of the deallocation policy. Whether the pool should hold a node floor is an infrastructure decision that needs an ADR and a quota discussion; it is recorded as an open question in the spec. Note the fix is not retroactive: the logs/ upload only applies to newly submitted tasks, so image layers that already failed this way must be re-run. Adds 31 unit tests across error classification, retry policy, runner degradation, the imagery blob fallback, upload patterns and message formatting. Also repairs a pre-existing fixture gap in test_imagery_preprocess_config.py, which covers the same _execute_image_preprocess method this change touches and had been failing since clipBbox was added to the submitted config -- MagicMock(spec=ImageLayer) does not expose pydantic field names. Suite goes from 155 passed / 2 failed to 188 passed / 0 failed. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 158a23ee-bb84-40cf-9974-98b097e7c6a1 --- api/hastefuncqueues/function_app.py | 20 +- .../src/hastegeo/core/processors/imagery.py | 65 +++++- .../src/hastegeo/core/runners/azure_batch.py | 155 +++++++++---- hastelib/src/hastegeo/core/runners/local.py | 32 ++- hastelib/src/hastegeo/core/utils/blob.py | 26 +++ hastelib/src/hastegeo/core/utils/errors.py | 67 ++++++ .../test_imagery_output_fallback.py | 199 +++++++++++++++++ .../test_imagery_preprocess_config.py | 1 + .../runners/test_azure_batch_node_errors.py | 204 ++++++++++++++++++ hastelib/tests/core/utils/test_errors.py | 91 ++++++++ 10 files changed, 797 insertions(+), 63 deletions(-) create mode 100644 hastelib/src/hastegeo/core/utils/errors.py create mode 100644 hastelib/tests/core/processors/test_imagery_output_fallback.py create mode 100644 hastelib/tests/core/runners/test_azure_batch_node_errors.py create mode 100644 hastelib/tests/core/utils/test_errors.py diff --git a/api/hastefuncqueues/function_app.py b/api/hastefuncqueues/function_app.py index d3bdc86b..26798632 100644 --- a/api/hastefuncqueues/function_app.py +++ b/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 diff --git a/hastelib/src/hastegeo/core/processors/imagery.py b/hastelib/src/hastegeo/core/processors/imagery.py index e233cb6c..e9a68a5d 100644 --- a/hastelib/src/hastegeo/core/processors/imagery.py +++ b/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}" diff --git a/hastelib/src/hastegeo/core/runners/azure_batch.py b/hastelib/src/hastegeo/core/runners/azure_batch.py index cf131f72..dc5db2c4 100644 --- a/hastelib/src/hastegeo/core/runners/azure_batch.py +++ b/hastelib/src/hastegeo/core/runners/azure_batch.py @@ -62,6 +62,18 @@ from .base import BaseRunner +# Node-scoped Batch file APIs (list/get/delete_from_task) are answered by the +# compute node that ran the task, so they fail once that node goes away. On +# autoscale pools with `$NodeDeallocationOption = taskcompletion` (and on +# low-priority/spot nodes that get preempted) the node is torn down at exactly +# the moment the caller reads the task's outputs back. +# +# NodeNotReady is a 409, not a 5xx, so the server-error retry never covered it. +# A node that is starting/rebooting recovers, so those codes are retried; a node +# that is gone never will, so those are surfaced as a non-fatal "unavailable". +TRANSIENT_NODE_ERROR_CODES = frozenset({"NodeNotReady", "NodeStateInvalid"}) +TERMINAL_NODE_ERROR_CODES = frozenset({"NodeNotFound"}) + class AzureBatchRunner(BaseRunner): def __init__( @@ -92,14 +104,32 @@ def __init__( def get_filecontent_from_task( self, job_id, task_id, filename, as_chunk=False ): - full_file_name = self.batch_cluster.get_file_by_match_from_task( - job_id, task_id, filename - ) - if full_file_name is None: + try: + full_file_name = self.batch_cluster.get_file_by_match_from_task( + job_id, task_id, filename + ) + if full_file_name is None: + return None + output = self.batch_cluster.get_file_from_task( + job_id, task_id, full_file_name + ) + except BatchErrorException as e: + # The node that ran the task is gone (deallocated, preempted or + # rebooting), so its copy of the file is unreachable. Report it as + # a missing file rather than a failure: the task's outputs were + # already uploaded to blob on completion, so callers can recover + # from there. + if not is_node_unavailable_error(e): + raise + self.logger.warning( + "Node serving task %s (job %s) is unavailable (%s); " + "cannot read %s from the node.", + task_id, + job_id, + batch_error_code(e), + filename, + ) return None - output = self.batch_cluster.get_file_from_task( - job_id, task_id, full_file_name - ) if output is None: return None if as_chunk: @@ -218,7 +248,21 @@ def add_task( return job_id, task_id def cleanup_task(self, job_id, task_id): - self.batch_cluster.delete_files_from_task(job_id, task_id) + try: + self.batch_cluster.delete_files_from_task(job_id, task_id) + except BatchErrorException as e: + # Deleting the working directory of a node that no longer exists is + # already a no-op, and the task's `retention_time` reclaims the disk + # anyway — never fail the workload over it. + if not is_node_unavailable_error(e): + raise + self.logger.warning( + "Node serving task %s (job %s) is unavailable (%s); " + "skipping working-directory cleanup.", + task_id, + job_id, + batch_error_code(e), + ) self.batch_cluster.disable_job(job_id) def cancel_task(self, job_id, task_id): @@ -233,11 +277,47 @@ def is_server_error(exception): return False +def batch_error_code(exception): + """Return the Batch error code of ``exception``, or None.""" + if not isinstance(exception, BatchErrorException): + return None + return getattr(getattr(exception, "error", None), "code", None) + + +def is_transient_node_error(exception): + """True when the node may still recover and serve the request. + + The node-scoped file APIs are answered by the compute node itself, so a + node that is starting, rebooting or otherwise mid-transition rejects them + with a 409 rather than a 5xx. Those are worth another attempt. + """ + return batch_error_code(exception) in TRANSIENT_NODE_ERROR_CODES + + +def is_terminal_node_error(exception): + """True when the node is gone for good and retrying cannot help.""" + return batch_error_code(exception) in TERMINAL_NODE_ERROR_CODES + + +def is_node_unavailable_error(exception): + """True for any node-loss error, transient or terminal.""" + return is_transient_node_error(exception) or is_terminal_node_error( + exception + ) + + +def is_retryable_batch_error(exception): + return is_server_error(exception) or is_transient_node_error(exception) + + def retry_on_server_error(): return retry( - retry=retry_if_exception(is_server_error), + retry=retry_if_exception(is_retryable_batch_error), wait=wait_exponential(multiplier=1, min=4, max=10), stop=stop_after_attempt(5), + # Surface the underlying BatchErrorException once the budget is spent, + # instead of tenacity's RetryError, so callers can still classify it. + reraise=True, ) @@ -562,6 +642,13 @@ def add_task( output_prefix = "output" if file_pattern is None: file_pattern = "$AZ_BATCH_TASK_WORKING_DIR/**/*" + # A pattern only ever matches one directory level, so a workload whose + # outputs live in more than one directory (e.g. imagery writes results + # to outputs/ and its progress log to logs/) has to supply several. + if isinstance(file_pattern, str): + file_patterns = [file_pattern] + else: + file_patterns = [p for p in file_pattern if p] if resource_files_for_upload is not None: resource_files = [ @@ -596,6 +683,25 @@ def add_task( output_sas_url = self._maybe_sas(output_container_url, "racwl") output_identity = self._blob_identity() + def _output_file(pattern): + return OutputFile( + file_pattern=pattern, + destination=OutputFileDestination( + container=OutputFileBlobContainerDestination( + container_url=output_sas_url, + path=output_prefix, + identity_reference=output_identity, + ) + ), + upload_options=OutputFileUploadOptions( + upload_condition=OutputFileUploadCondition.task_completion + ), + ) + + # Data files, then stdout, stderr, fileuploadout and fileuploaderr. + output_files = [_output_file(p) for p in file_patterns] + output_files.append(_output_file("../*.txt")) + task = TaskAddParameter( id=task_id, constraints=task_constraints, @@ -609,36 +715,7 @@ def add_task( elevation_level=ElevationLevel.admin, ) ), - output_files=[ - # Data files - OutputFile( - file_pattern=file_pattern, - destination=OutputFileDestination( - container=OutputFileBlobContainerDestination( - container_url=output_sas_url, - path=output_prefix, - identity_reference=output_identity, - ) - ), - upload_options=OutputFileUploadOptions( - upload_condition=OutputFileUploadCondition.task_completion - ), - ), - # grabs stdout, stderr, fileuploadout and fileuploaderr - OutputFile( - file_pattern="../*.txt", - destination=OutputFileDestination( - container=OutputFileBlobContainerDestination( - container_url=output_sas_url, - path=output_prefix, - identity_reference=output_identity, - ) - ), - upload_options=OutputFileUploadOptions( - upload_condition=OutputFileUploadCondition.task_completion - ), - ), - ], + output_files=output_files, ) self.batch_client.task.add(job_id, task) diff --git a/hastelib/src/hastegeo/core/runners/local.py b/hastelib/src/hastegeo/core/runners/local.py index bc39b378..e602a84d 100644 --- a/hastelib/src/hastegeo/core/runners/local.py +++ b/hastelib/src/hastegeo/core/runners/local.py @@ -1279,7 +1279,7 @@ def _upload_task_outputs( output_container_url: str, output_prefix: str, resource_files: list, - file_pattern: str, + file_pattern, ): """Upload task output files to blob storage.""" if not self.blob_client: @@ -1290,16 +1290,26 @@ def _upload_task_outputs( files_to_upload = [] if file_pattern: - normalized_pattern = file_pattern.replace("\\", "/") - try: - files_to_upload = [ - Path(p) - for p in glob.glob(normalized_pattern, recursive=True) - ] - except Exception as e: - self.logger.warning( - f"Failed globbing output pattern {file_pattern}: {e}. Falling back to task_dir search" - ) + # Mirror AzureBatchJob.add_task: a workload may supply several + # patterns when its outputs span more than one directory. + patterns = ( + [file_pattern] + if isinstance(file_pattern, str) + else [p for p in file_pattern if p] + ) + for pattern in patterns: + normalized_pattern = pattern.replace("\\", "/") + try: + files_to_upload += [ + Path(p) + for p in glob.glob( + normalized_pattern, recursive=True + ) + ] + except Exception as e: + self.logger.warning( + f"Failed globbing output pattern {pattern}: {e}. Falling back to task_dir search" + ) if not files_to_upload: outputs_dir = task_dir / "outputs" diff --git a/hastelib/src/hastegeo/core/utils/blob.py b/hastelib/src/hastegeo/core/utils/blob.py index c000946b..2ecdb3fa 100644 --- a/hastelib/src/hastegeo/core/utils/blob.py +++ b/hastelib/src/hastegeo/core/utils/blob.py @@ -60,6 +60,32 @@ def split_blob_url(url: str) -> Tuple[str, str]: return parts[1], "/".join(parts[2:]) +def fetch_url_text(url: str, timeout: int = 30) -> Optional[str]: + """Return the text body at ``url``, or ``None`` if it cannot be read. + + Used to recover a task's output files from the copy Azure Batch uploaded to + blob storage when the compute node that ran the task is no longer able to + serve them. Callers treat this as a best-effort fallback, so transport and + HTTP errors are reported as ``None`` rather than raised — a failure here + must never mask the original reason the node read failed. + + Only ``http(s)`` locations are fetched; a data layer that resolves to a + local filesystem path (the docker dev stack) returns ``None``. + """ + if not url or not urlparse(url).scheme.startswith("http"): + return None + # Imported lazily to keep this module cheap for callers that only need + # split_blob_url(). + import requests + + try: + response = requests.get(url, timeout=timeout) + response.raise_for_status() + return response.text + except Exception: + return None + + async def download_blob_to_tempfile(url: str, suffix: str = "") -> str: """Download the blob at ``url`` to a NamedTemporaryFile and return the path. diff --git a/hastelib/src/hastegeo/core/utils/errors.py b/hastelib/src/hastegeo/core/utils/errors.py new file mode 100644 index 00000000..176130ac --- /dev/null +++ b/hastelib/src/hastegeo/core/utils/errors.py @@ -0,0 +1,67 @@ +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. + +"""Human-readable rendering of exceptions surfaced to end users. + +Azure SDK errors stringify to a debug dump — an ``azure.batch`` +``BatchErrorException`` renders as:: + + Request encountered an exception. + Code: NodeNotReady + Message: {'additional_properties': {}, 'lang': 'en-US', 'value': 'Node is + not able to perform the requested operations in its current + state\\nRequestId:...\\nTime:...'} + +That text ends up in ``statusMessage`` and then verbatim in the UI. This module +reduces such errors to ``Code: message`` so the status dialog stays readable. + +The Azure error shape is matched structurally (``.error.code`` / +``.error.message.value``) rather than by ``isinstance``, so this stays a leaf +utility with no SDK import. +""" + +# RequestId/Time are appended by the service to the message body; they are +# useful in logs but noise in a UI status message. +_SERVICE_TRAILER_MARKERS = ("\nRequestId:", "\nTime:") + + +def _unwrap_message(message): + """Return the human-readable text of an Azure ``ErrorMessage``-like value.""" + value = getattr(message, "value", None) + if isinstance(value, str): + return value + return message if isinstance(message, str) else None + + +def _strip_service_trailer(text): + for marker in _SERVICE_TRAILER_MARKERS: + index = text.find(marker) + if index != -1: + text = text[:index] + return text.strip() + + +def describe_exception(exc): + """Return a short, user-facing description of ``exc``. + + Azure-style errors become ``": "``; everything else falls + back to ``str(exc)``. The exception type is used when there is no message + at all, so the result is never an empty string. + """ + error = getattr(exc, "error", None) + code = getattr(error, "code", None) if error is not None else None + message = ( + _unwrap_message(getattr(error, "message", None)) + if error is not None + else None + ) + + if isinstance(code, str) and code: + if message: + return f"{code}: {_strip_service_trailer(message)}" + return code + if message: + return _strip_service_trailer(message) + + text = str(exc).strip() + return text or type(exc).__name__ diff --git a/hastelib/tests/core/processors/test_imagery_output_fallback.py b/hastelib/tests/core/processors/test_imagery_output_fallback.py new file mode 100644 index 00000000..96f053d9 --- /dev/null +++ b/hastelib/tests/core/processors/test_imagery_output_fallback.py @@ -0,0 +1,199 @@ +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. + +"""Unit tests for ImageryPostProcessor's tolerance of Batch node loss. + +Azure Batch uploads a task's ``outputs/`` (and now ``logs/``) to blob storage on +completion, so the results survive the compute node being deallocated or +preempted. These tests pin that the processor actually falls back to that copy +instead of failing the image layer, and that both directories are submitted for +upload in the first place. +""" + +import json +import unittest +from unittest.mock import MagicMock, patch + +from hastegeo.core.models.projects import ImageLayer + + +def _build_processor(): + image_data = MagicMock(spec=ImageLayer) + image_data.dict.return_value = {} + image_data.projectId = "proj-1" + image_data.imageLayerId = "layer-9" + image_data.preEventImageryUrls = ["https://example/pre.tif"] + image_data.postEventImageryUrls = ["https://example/post.tif"] + image_data.sourceTypePreEvent = "url" + image_data.sourceTypePostEvent = "url" + image_data.autoFineTune = False + image_data.userBuildingFootprintsUrl = None + image_data.clipBbox = None + image_data.currentStep = 0 + image_data.totalSteps = 4 + image_data.progressPct = 0.0 + image_data.statusMessage = "" + image_data.preprocessJob = MagicMock(jobId="job-1", taskId="img-123") + + with patch( + "hastegeo.core.processors.imagery.UnifiedDataLayer", autospec=True + ), patch( + "hastegeo.core.processors.imagery.UnifiedRunner", autospec=True + ), patch( + "hastegeo.core.processors.imagery.AzureQueueHandler", autospec=True + ): + from hastegeo.core.processors.imagery import ImageryPostProcessor + + return ImageryPostProcessor(image_data=image_data) + + +class TestReadTaskOutputFallback(unittest.TestCase): + def test_prefers_the_node_copy(self): + processor = _build_processor() + processor.runner.get_filecontent_from_task.return_value = "from-node" + + with patch("hastegeo.core.processors.imagery.fetch_url_text") as fetch: + self.assertEqual( + processor._read_task_output("imagery_manifest.json"), + "from-node", + ) + fetch.assert_not_called() + + def test_falls_back_to_the_uploaded_blob_copy(self): + processor = _build_processor() + # The node is gone, so the runner reports the file as unavailable. + processor.runner.get_filecontent_from_task.return_value = None + processor.storage.get_file_remote_path.return_value = ( + "https://acct.blob.core.windows.net/c/hash/img-123/" + "imagery_manifest.json?sas" + ) + + with patch( + "hastegeo.core.processors.imagery.fetch_url_text", + return_value="from-blob", + ) as fetch: + self.assertEqual( + processor._read_task_output("imagery_manifest.json"), + "from-blob", + ) + + fetch.assert_called_once() + # The blob copy lives under the task id, matching the Batch + # OutputFile prefix. + _, kwargs = processor.storage.get_file_remote_path.call_args + self.assertEqual(kwargs["identifier"], "imagery_manifest.json") + self.assertEqual(kwargs["extra_partition_keys"], "img-123") + self.assertEqual(kwargs["data_format"], "json") + + def test_returns_none_when_neither_copy_is_available(self): + processor = _build_processor() + processor.runner.get_filecontent_from_task.return_value = None + processor.storage.get_file_remote_path.return_value = None + + with patch( + "hastegeo.core.processors.imagery.fetch_url_text", + return_value=None, + ): + self.assertIsNone( + processor._read_task_output("imagery_manifest.json") + ) + + def test_a_failing_fallback_does_not_raise(self): + processor = _build_processor() + processor.runner.get_filecontent_from_task.return_value = None + processor.storage.get_file_remote_path.side_effect = RuntimeError( + "storage down" + ) + + self.assertIsNone(processor._read_task_output("imagery_manifest.json")) + + +class TestPreprocessLogsAreBestEffort(unittest.TestCase): + def test_unreachable_log_yields_no_records_instead_of_failing(self): + processor = _build_processor() + processor.runner.get_filecontent_from_task.return_value = None + processor.storage.get_file_remote_path.return_value = None + + with patch( + "hastegeo.core.processors.imagery.fetch_url_text", + return_value=None, + ): + self.assertEqual(processor._get_image_preprocess_logs(), []) + + def test_log_is_parsed_from_the_blob_copy(self): + processor = _build_processor() + processor.runner.get_filecontent_from_task.return_value = None + processor.storage.get_file_remote_path.return_value = "https://b/log" + + with patch( + "hastegeo.core.processors.imagery.fetch_url_text", + return_value="2026-08-12T20:00:00|Downloading imagery\n", + ): + logs = processor._get_image_preprocess_logs() + + self.assertEqual(len(logs), 1) + self.assertEqual(logs[0].message, "Downloading imagery") + + +class TestUpdateResultsFromJob(unittest.TestCase): + def test_uses_the_blob_manifest_when_the_node_is_gone(self): + processor = _build_processor() + processor.runner.get_filecontent_from_task.return_value = None + processor.storage.get_file_remote_path.return_value = "https://b/m" + manifest = { + "preview_pre_event_filenames": [], + "preview_post_event_filenames": [], + "pre_event_mosaic_filename": "", + "pre_event_processed_filename": "", + "post_event_mosaic_filename": "", + "post_event_processed_filename": "", + "normalization_means": [1.0], + "normalization_stds": [2.0], + "building_footprints_filename": "", + "building_footprints_error": "", + "valid_area_mask_filename": "", + "valid_area_mask_error": "", + } + + with patch( + "hastegeo.core.processors.imagery.fetch_url_text", + return_value=json.dumps(manifest), + ): + processor._update_results_from_job() + + self.assertEqual(processor.image_data.normalizationMeans, [1.0]) + self.assertEqual(processor.image_data.normalizationStds, [2.0]) + + def test_raises_when_the_manifest_is_lost_everywhere(self): + processor = _build_processor() + processor.runner.get_filecontent_from_task.return_value = None + processor.storage.get_file_remote_path.return_value = None + + with patch( + "hastegeo.core.processors.imagery.fetch_url_text", + return_value=None, + ): + with self.assertRaises(FileNotFoundError): + processor._update_results_from_job() + + +class TestSubmittedOutputPatterns(unittest.TestCase): + def test_uploads_both_outputs_and_logs(self): + processor = _build_processor() + processor.storage.get_file_remote_path.return_value = ( + "https://acct/c/hash/config.yaml?sig=x" + ) + processor.runner.add_task.return_value = ("job-1", "img-123") + + processor._execute_image_preprocess() + + _, kwargs = processor.runner.add_task.call_args + patterns = kwargs["file_pattern"] + self.assertIsInstance(patterns, list) + self.assertIn("$AZ_BATCH_TASK_WORKING_DIR/outputs/*.*", patterns) + # Without this the progress log only ever exists on the node. + self.assertIn("$AZ_BATCH_TASK_WORKING_DIR/logs/*.*", patterns) + + +if __name__ == "__main__": + unittest.main() diff --git a/hastelib/tests/core/processors/test_imagery_preprocess_config.py b/hastelib/tests/core/processors/test_imagery_preprocess_config.py index 2fad81b1..2cebeefc 100644 --- a/hastelib/tests/core/processors/test_imagery_preprocess_config.py +++ b/hastelib/tests/core/processors/test_imagery_preprocess_config.py @@ -31,6 +31,7 @@ def _build_processor(self, *, user_url): image_data.sourceTypePostEvent = "url" image_data.autoFineTune = False image_data.userBuildingFootprintsUrl = user_url + image_data.clipBbox = None with patch( "hastegeo.core.processors.imagery.UnifiedDataLayer", diff --git a/hastelib/tests/core/runners/test_azure_batch_node_errors.py b/hastelib/tests/core/runners/test_azure_batch_node_errors.py new file mode 100644 index 00000000..6b0983e3 --- /dev/null +++ b/hastelib/tests/core/runners/test_azure_batch_node_errors.py @@ -0,0 +1,204 @@ +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. + +"""Unit tests for Azure Batch node-loss handling. + +The node-scoped file APIs (``file.list_from_task`` / ``get_from_task`` / +``delete_from_task``) are answered by the compute node that ran the task, so +they start failing the moment that node is deallocated or preempted — which on +an autoscale pool is exactly when the task completes. These tests pin the +resulting behavior: transient node errors are retried, terminal ones degrade to +"file unavailable" instead of failing the workload, and unrelated Batch errors +still propagate. +""" + +import unittest +from unittest.mock import MagicMock + +from azure.batch.models import BatchError, BatchErrorException, ErrorMessage +from hastegeo.core.runners.azure_batch import ( + AzureBatchRunner, + batch_error_code, + is_node_unavailable_error, + is_retryable_batch_error, + is_server_error, + is_terminal_node_error, + is_transient_node_error, + retry_on_server_error, +) +from tenacity import wait_none + + +def _batch_error(code, status_code=409, value="node is busy"): + """Build a BatchErrorException without going through the deserializer.""" + exc = BatchErrorException.__new__(BatchErrorException) + exc.error = BatchError(code=code, message=ErrorMessage(value=value)) + exc.response = MagicMock(status_code=status_code) + return exc + + +def _runner(): + """An AzureBatchRunner with only the collaborators these tests touch.""" + runner = AzureBatchRunner.__new__(AzureBatchRunner) + runner.batch_cluster = MagicMock() + runner.logger = MagicMock() + return runner + + +class TestBatchErrorClassification(unittest.TestCase): + def test_node_not_ready_is_transient_and_retryable(self): + # The failure reported from dev1: a 409, not a 5xx. + exc = _batch_error("NodeNotReady", status_code=409) + self.assertTrue(is_transient_node_error(exc)) + self.assertFalse(is_terminal_node_error(exc)) + self.assertTrue(is_node_unavailable_error(exc)) + self.assertTrue(is_retryable_batch_error(exc)) + self.assertFalse(is_server_error(exc)) + + def test_node_state_invalid_is_transient(self): + exc = _batch_error("NodeStateInvalid", status_code=409) + self.assertTrue(is_transient_node_error(exc)) + self.assertTrue(is_retryable_batch_error(exc)) + + def test_node_not_found_is_terminal_but_unavailable(self): + exc = _batch_error("NodeNotFound", status_code=404) + self.assertTrue(is_terminal_node_error(exc)) + self.assertFalse(is_transient_node_error(exc)) + self.assertTrue(is_node_unavailable_error(exc)) + # Retrying cannot bring a deallocated node back. + self.assertFalse(is_retryable_batch_error(exc)) + + def test_server_errors_remain_retryable(self): + exc = _batch_error("InternalServerError", status_code=500) + self.assertTrue(is_server_error(exc)) + self.assertTrue(is_retryable_batch_error(exc)) + + def test_unrelated_client_error_is_not_retryable(self): + exc = _batch_error("TaskNotFound", status_code=404) + self.assertFalse(is_retryable_batch_error(exc)) + self.assertFalse(is_node_unavailable_error(exc)) + + def test_non_batch_exception_is_not_retryable(self): + exc = ValueError("boom") + self.assertIsNone(batch_error_code(exc)) + self.assertFalse(is_retryable_batch_error(exc)) + self.assertFalse(is_node_unavailable_error(exc)) + + +class TestRetryOnServerError(unittest.TestCase): + def _no_wait(self, func): + # Keep the real predicate/stop policy, drop only the backoff sleep. + return func.retry_with(wait=wait_none()) + + def test_retries_node_not_ready_until_it_succeeds(self): + calls = {"n": 0} + + @retry_on_server_error() + def flaky(): + calls["n"] += 1 + if calls["n"] < 3: + raise _batch_error("NodeNotReady") + return "ok" + + self.assertEqual(self._no_wait(flaky)(), "ok") + self.assertEqual(calls["n"], 3) + + def test_reraises_the_batch_error_not_a_retry_error(self): + @retry_on_server_error() + def always_failing(): + raise _batch_error("NodeNotReady") + + # reraise=True keeps the error classifiable by callers instead of + # burying it in tenacity's RetryError. + with self.assertRaises(BatchErrorException) as ctx: + self._no_wait(always_failing)() + self.assertEqual(ctx.exception.error.code, "NodeNotReady") + + def test_does_not_retry_unrelated_errors(self): + calls = {"n": 0} + + @retry_on_server_error() + def failing(): + calls["n"] += 1 + raise _batch_error("TaskNotFound", status_code=404) + + with self.assertRaises(BatchErrorException): + self._no_wait(failing)() + self.assertEqual(calls["n"], 1) + + +class TestGetFileContentFromTask(unittest.TestCase): + def test_returns_none_when_node_is_gone(self): + runner = _runner() + runner.batch_cluster.get_file_by_match_from_task.side_effect = ( + _batch_error("NodeNotFound", status_code=404) + ) + self.assertIsNone( + runner.get_filecontent_from_task( + "job-1", "task-1", "imagery_manifest.json" + ) + ) + runner.logger.warning.assert_called_once() + + def test_returns_none_when_node_never_became_ready(self): + runner = _runner() + runner.batch_cluster.get_file_by_match_from_task.return_value = ( + "wd/logs/imagery_friendly.log" + ) + runner.batch_cluster.get_file_from_task.side_effect = _batch_error( + "NodeNotReady" + ) + self.assertIsNone( + runner.get_filecontent_from_task( + "job-1", "task-1", "imagery_friendly.log" + ) + ) + + def test_propagates_unrelated_batch_errors(self): + runner = _runner() + runner.batch_cluster.get_file_by_match_from_task.side_effect = ( + _batch_error("JobNotFound", status_code=404) + ) + with self.assertRaises(BatchErrorException): + runner.get_filecontent_from_task("job-1", "task-1", "any.json") + + def test_reads_content_when_the_node_answers(self): + runner = _runner() + runner.batch_cluster.get_file_by_match_from_task.return_value = ( + "wd/outputs/imagery_manifest.json" + ) + runner.batch_cluster.get_file_from_task.return_value = [ + b'{"a": ', + b"1}", + ] + self.assertEqual( + runner.get_filecontent_from_task( + "job-1", "task-1", "imagery_manifest.json" + ), + '{"a": 1}', + ) + + +class TestCleanupTask(unittest.TestCase): + def test_skips_working_directory_cleanup_when_node_is_gone(self): + runner = _runner() + runner.batch_cluster.delete_files_from_task.side_effect = _batch_error( + "NodeNotReady" + ) + runner.cleanup_task("job-1", "task-1") + # The job still has to be disabled — cleanup of a dead node's disk is + # handled by the task retention time. + runner.batch_cluster.disable_job.assert_called_once_with("job-1") + + def test_propagates_unrelated_batch_errors(self): + runner = _runner() + runner.batch_cluster.delete_files_from_task.side_effect = _batch_error( + "OperationTimedOut", status_code=408 + ) + with self.assertRaises(BatchErrorException): + runner.cleanup_task("job-1", "task-1") + runner.batch_cluster.disable_job.assert_not_called() + + +if __name__ == "__main__": + unittest.main() diff --git a/hastelib/tests/core/utils/test_errors.py b/hastelib/tests/core/utils/test_errors.py new file mode 100644 index 00000000..618ecd0e --- /dev/null +++ b/hastelib/tests/core/utils/test_errors.py @@ -0,0 +1,91 @@ +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. + +"""Unit tests for describe_exception.""" + +import unittest + +from hastegeo.core.utils.errors import describe_exception + + +class _Message: + def __init__(self, value): + self.value = value + self.lang = "en-US" + + +class _BatchError: + def __init__(self, code, message=None): + self.code = code + self.message = message + self.additional_properties = {} + + +class _AzureStyleException(Exception): + def __init__(self, error): + super().__init__("Request encountered an exception.") + self.error = error + + +class TestDescribeException(unittest.TestCase): + def test_renders_batch_error_as_code_and_message(self): + exc = _AzureStyleException( + _BatchError( + "NodeNotReady", + _Message( + "Node is not able to perform the requested operations " + "in its current state" + ), + ) + ) + self.assertEqual( + describe_exception(exc), + "NodeNotReady: Node is not able to perform the requested " + "operations in its current state", + ) + + def test_strips_the_request_id_and_time_trailer(self): + exc = _AzureStyleException( + _BatchError( + "NodeNotReady", + _Message( + "Node is not able to perform the requested operations " + "in its current state\n" + "RequestId:a61bf14a-dcf1-4a68-b0ab-dddb878b4951\n" + "Time:2026-08-12T20:19:59.1266709Z" + ), + ) + ) + described = describe_exception(exc) + self.assertNotIn("RequestId", described) + self.assertNotIn("Time:", described) + self.assertTrue(described.startswith("NodeNotReady: Node is not able")) + + def test_never_leaks_the_error_model_repr(self): + exc = _AzureStyleException( + _BatchError("NodeNotReady", _Message("node is busy")) + ) + described = describe_exception(exc) + self.assertNotIn("additional_properties", described) + self.assertNotIn("'lang'", described) + + def test_falls_back_to_the_code_when_there_is_no_message(self): + exc = _AzureStyleException(_BatchError("PoolNotFound")) + self.assertEqual(describe_exception(exc), "PoolNotFound") + + def test_falls_back_to_str_for_plain_exceptions(self): + self.assertEqual( + describe_exception(ValueError("bad imagery url")), + "bad imagery url", + ) + + def test_never_returns_an_empty_string(self): + self.assertEqual(describe_exception(RuntimeError()), "RuntimeError") + + def test_handles_a_plain_string_message(self): + exc = _AzureStyleException(_BatchError("NodeNotFound", "node is gone")) + self.assertEqual(describe_exception(exc), "NodeNotFound: node is gone") + + +if __name__ == "__main__": + unittest.main() From 64642fbbe24d450b000fc658ed7d5294d3991b57 Mon Sep 17 00:00:00 2001 From: jQuinRivero Date: Wed, 12 Aug 2026 18:02:18 -0300 Subject: [PATCH 2/2] docs(spec): add batch-node-loss-resilience specification Record the root cause, the design, and the alternatives that were rejected, per the repository rule that specs are the source of truth. Two documentation claims were also wrong once the read path changed. Both api/hastefuncqueues/README.md and docs/api/hastefuncqueues.md stated that imagery_friendly.log and imagery_manifest.json are read "from the task working directory", which is now only the first of two sources -- and, on an autoscale pool, usually the one that no longer exists. They now describe the node-then-blob order, which of the two files is required, and why the node copy disappears. The spec set follows spec/_templates/feature/ and includes the Agent Assignment Map required by the templates. It records the pool configuration as an explicit non-goal: this change makes node loss survivable rather than impossible, and whether the pool should hold a node floor is left as an open question needing an ADR and a quota discussion. The CHANGELOG entry sits alongside the other Batch fixes under Unreleased, and carries the operator note that the logs/ upload is not retroactive -- image layers that already failed this way have to be re-run. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 158a23ee-bb84-40cf-9974-98b097e7c6a1 --- CHANGELOG.md | 5 + api/hastefuncqueues/README.md | 10 +- docs/api/hastefuncqueues.md | 10 +- .../batch-node-loss-resilience/README.md | 167 +++++++++++++++ .../batch-node-loss-resilience/design.md | 144 +++++++++++++ .../impact-analysis.md | 90 ++++++++ .../batch-node-loss-resilience/plan.md | 93 ++++++++ .../batch-node-loss-resilience/rollout.md | 98 +++++++++ .../batch-node-loss-resilience/test-plan.md | 130 +++++++++++ .../user-stories.md | 202 ++++++++++++++++++ 10 files changed, 947 insertions(+), 2 deletions(-) create mode 100644 spec/features/batch-node-loss-resilience/README.md create mode 100644 spec/features/batch-node-loss-resilience/design.md create mode 100644 spec/features/batch-node-loss-resilience/impact-analysis.md create mode 100644 spec/features/batch-node-loss-resilience/plan.md create mode 100644 spec/features/batch-node-loss-resilience/rollout.md create mode 100644 spec/features/batch-node-loss-resilience/test-plan.md create mode 100644 spec/features/batch-node-loss-resilience/user-stories.md diff --git a/CHANGELOG.md b/CHANGELOG.md index 4bbdb98f..2a739974 100644 --- a/CHANGELOG.md +++ b/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: `. 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 diff --git a/api/hastefuncqueues/README.md b/api/hastefuncqueues/README.md index d5e4d695..ec18b159 100644 --- a/api/hastefuncqueues/README.md +++ b/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 `//`). 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 diff --git a/docs/api/hastefuncqueues.md b/docs/api/hastefuncqueues.md index 336dc3e5..b5a90122 100644 --- a/docs/api/hastefuncqueues.md +++ b/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 `//`). 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 diff --git a/spec/features/batch-node-loss-resilience/README.md b/spec/features/batch-node-loss-resilience/README.md new file mode 100644 index 00000000..924facab --- /dev/null +++ b/spec/features/batch-node-loss-resilience/README.md @@ -0,0 +1,167 @@ +# Batch node-loss resilience + +**Status:** implemented +**Type:** modification (fix) +**Related:** [`../batch-compute-expansion/`](../batch-compute-expansion/) +(shared autoscale/spot pools), [`../batch-pool-job-binding/`](../batch-pool-job-binding/) +(job/pool binding) + +## Problem + +Processing new imagery on **dev1** failed, and the UI status dialog showed a raw +Azure SDK dump as the entire status history for the layer: + +``` +Request encountered an exception. +Code: NodeNotReady +Message: {'additional_properties': {}, 'lang': 'en-US', 'value': 'Node is not +able to perform the requested operations in its current state +RequestId:a61bf14a-dcf1-4a68-b0ab-dddb878b4951 +Time:2026-08-12T20:19:59.1266709Z'} +``` + +The Batch task itself ran fine. The imagery it produced was already in blob +storage. The layer was still marked FAILED. + +## Root cause + +`ImageryPostProcessor.process` reacts to a task reaching a terminal state by +reading files **back off the compute node**: + +| Call | Batch API | Purpose | +|---|---|---| +| `_get_image_preprocess_logs` | `file.list/get_from_task` | `imagery_friendly.log` | +| `_update_results_from_job` | `file.list/get_from_task` | `imagery_manifest.json` | +| `runner.cleanup_task` | `file.delete_from_task` | working-directory cleanup | + +Those APIs are served **by the node**, and the node is being torn down at exactly +that moment: + +1. `infra/modules/batchPool.bicep` sets `$NodeDeallocationOption = taskcompletion`. +2. The shared-dev pools dev1 targets are autoscale with `minNodes = 0` and + low-priority (spot) nodes, so they can also be preempted + (see [`../batch-compute-expansion/design.md`](../batch-compute-expansion/design.md)). + +A deallocating, rebooting or preempted node answers `NodeNotReady`. + +Three properties then turned a transient race into a permanent failure: + +| Property | Effect | +|---|---| +| `NodeNotReady` is HTTP **409**, and `is_server_error` only retried 5xx | never retried | +| The imagery queue trigger did `statusMessage = str(e)` | wiped the entire progress history with the SDK repr | +| `host.json` sets `maxDequeueCount: 1` | the message is never redelivered | + +**The outputs were never lost.** Batch uploads `outputs/*.*` to blob on task +completion (`OutputFileUploadCondition.task_completion`), so the manifest was +sitting at `//imagery_manifest.json` — the same path +`_generate_imagery_url` already resolves. The processor simply never looked +there. `imagery_friendly.log` is written to `logs/`, which no upload pattern +covered, so it only ever existed on the node. + +## Design + +Three layers, all in code — the pool configuration is unchanged. + +### 1. Classify node errors, and retry the transient ones + +`hastegeo.core.runners.azure_batch` gains: + +| Set | Codes | Treatment | +|---|---|---| +| `TRANSIENT_NODE_ERROR_CODES` | `NodeNotReady`, `NodeStateInvalid` | retried — a starting/rebooting node recovers | +| `TERMINAL_NODE_ERROR_CODES` | `NodeNotFound` | not retried — a deallocated node never comes back | + +`is_retryable_batch_error` = 5xx **or** transient node code, and is now the +predicate behind `retry_on_server_error()`. The retry budget is unchanged +(5 attempts, exponential 4–10s) and `reraise=True` was added so the original +`BatchErrorException` survives an exhausted budget instead of being buried in +tenacity's `RetryError`. + +### 2. Degrade instead of failing + +| Method | Behavior on node loss | +|---|---| +| `AzureBatchRunner.get_filecontent_from_task` | logs a warning, returns `None` — matching its existing "file not found" contract | +| `AzureBatchRunner.cleanup_task` | skips the working-directory delete, still disables the job (`task_retention_time=P2D` reclaims the disk) | + +This applies to every workload on the runner: imagery, train, inference, +artifacts, embedding. + +### 3. Recover the outputs from blob (imagery) + +`ImageryPostProcessor._read_task_output(filename)` reads the node copy first, +then falls back to the uploaded copy via +`storage.get_file_remote_path(identifier=filename, extra_partition_keys=taskId)` +and `hastegeo.core.utils.blob.fetch_url_text`. A failing fallback returns `None` +rather than raising, so it can never mask the original reason the node read +failed. + +The manifest is required — losing it in both places still raises +`FileNotFoundError`. The progress log is best-effort. + +To make the log recoverable at all, `add_task` now accepts `file_pattern` as a +string **or a list**, emitting one `OutputFile` per pattern, and imagery submits +both `outputs/*.*` and `logs/*.*`. A single blanket `**/*` was rejected: it would +also upload the raw downloaded imagery. + +### 4. Stop destroying the status history + +The imagery queue trigger appends via `MetadataUtils.append_status_message` +instead of assigning, and renders the cause through a new +`hastegeo.core.utils.errors.describe_exception`, which reduces an Azure-style +error to `NodeNotReady: Node is not able to perform the requested operations in +its current state`. It matches the error *shape* (`.error.code` / +`.error.message.value`) rather than importing `azure.batch`, so it stays a leaf +utility. + +## Non-goals + +- **Pool configuration.** Deallocation policy, `minNodes` and spot-vs-dedicated + are untouched; this change makes the race survivable, not impossible. +- **Raising `maxDequeueCount`.** Redelivery would re-run whole tasks. +- **Blob fallback for other workloads.** The runner-level fixes cover them; only + imagery reads a required output file back. + +## Agent assignment + +| Area | Implements | Validates | +|---|---|---| +| `hastelib/` runner, processor, utils | `backend-dev` | `backend-validation` | +| `api/hastefuncqueues/` status messages | `backend-dev` | `backend-validation` | +| Docs + spec | `backend-dev` | `orchestrator` | + +## Acceptance criteria + +1. A `NodeNotReady` response is retried rather than failing the layer. +2. When the node is gone for good, the manifest is recovered from blob and the + layer completes normally. +3. An unreachable progress log never fails an otherwise successful layer. +4. Cleanup against a dead node does not fail the workload, and the job is still + disabled. +5. Losing the manifest in both places still fails the layer, with a readable + message. +6. A failure appends to `statusMessage`; it never replaces the history. +7. Unrelated Batch errors still propagate unchanged. + +## Document index + +| Document | Purpose | +|---|---| +| [design.md](design.md) | Technical design, call paths, contracts | +| [impact-analysis.md](impact-analysis.md) | Risk, dependencies, blast radius | +| [user-stories.md](user-stories.md) | Stories, acceptance criteria, agent map | +| [test-plan.md](test-plan.md) | Test strategy and coverage matrix | +| [plan.md](plan.md) | Execution plan and task status | +| [rollout.md](rollout.md) | Rollout, verification, rollback | + +## Decision log + +| Decision | Rationale | +|---|---| +| Widen the retry predicate rather than add a second decorator | `apply_retry_to_methods` already wraps every `AzureBatchJob` method | +| `reraise=True` | callers must still be able to classify the error after the budget is spent | +| Return `None` for an unreachable file | matches the existing contract; callers already branch on falsy content | +| Fall back via `get_file_remote_path` + HTTP | resolves the exact path Batch uploaded to, with no data-layer signature changes | +| `file_pattern` accepts a list | targeted; avoids a `**/*` upload of raw imagery | +| Duck-type the error shape in `utils.errors` | keeps a leaf utility free of an `azure.batch` import | diff --git a/spec/features/batch-node-loss-resilience/design.md b/spec/features/batch-node-loss-resilience/design.md new file mode 100644 index 00000000..6d9adf87 --- /dev/null +++ b/spec/features/batch-node-loss-resilience/design.md @@ -0,0 +1,144 @@ +# Design: Batch node-loss resilience + +## Context + +Azure Batch exposes two different classes of API: + +| Class | Examples | Served by | Survives node loss | +|---|---|---|---| +| Job/task scope | `task.get`, `job.disable`, `task.add` | Batch service | yes | +| Node scope | `file.list_from_task`, `file.get_from_task`, `file.delete_from_task` | the compute node | **no** | + +HASTE's post-task bookkeeping uses the second class. On a fixed, always-on pool +that is safe. On an autoscale pool with `$NodeDeallocationOption = taskcompletion` +— or on preemptible low-priority nodes — the node disappears the moment the task +finishes, which is precisely when the bookkeeping runs. + +## Call path + +``` +GetProcessImageLayerQueueTrigger api/hastefuncqueues/function_app.py +└── ImageryPostProcessor.process hastelib/.../processors/imagery.py + ├── runner.get_task_status (job scope — safe) + ├── _update_results_from_job + │ └── _read_task_output("imagery_manifest.json") ← node scope + ├── _get_image_preprocess_logs + │ └── _read_task_output("imagery_friendly.log") ← node scope + └── runner.cleanup_task + └── delete_files_from_task ← node scope +``` + +## Contracts + +### `hastegeo.core.runners.azure_batch` + +```python +TRANSIENT_NODE_ERROR_CODES = frozenset({"NodeNotReady", "NodeStateInvalid"}) +TERMINAL_NODE_ERROR_CODES = frozenset({"NodeNotFound"}) + +batch_error_code(exc) -> str | None # None for non-Batch exceptions +is_transient_node_error(exc) -> bool +is_terminal_node_error(exc) -> bool +is_node_unavailable_error(exc) -> bool # transient or terminal +is_retryable_batch_error(exc) -> bool # 5xx or transient node error +``` + +`retry_on_server_error()` keeps its name (it is applied to every `AzureBatchJob` +method by `apply_retry_to_methods`) but now retries on +`is_retryable_batch_error` and sets `reraise=True`. + +| Aspect | Before | After | +|---|---|---| +| Retried | 5xx only | 5xx + transient node errors | +| Budget | 5 attempts, exponential 4–10s | unchanged | +| On exhaustion | `tenacity.RetryError` | the original `BatchErrorException` | + +### `AzureBatchRunner` + +| Method | Contract | +|---|---| +| `get_filecontent_from_task` | returns file text, or `None` when the file is missing **or** the node cannot serve it. Unrelated `BatchErrorException`s propagate. | +| `cleanup_task` | best-effort working-directory delete; always disables the job. Unrelated `BatchErrorException`s propagate. | + +### `AzureBatchJob.add_task` + +`file_pattern` accepts `str | list[str]`. Each pattern becomes its own +`OutputFile` against the same destination prefix; the `../*.txt` stdout/stderr +capture is appended as before. Existing string callers are unaffected. + +| Workload | Patterns | +|---|---| +| imagery | `$AZ_BATCH_TASK_WORKING_DIR/outputs/*.*`, `$AZ_BATCH_TASK_WORKING_DIR/logs/*.*` | +| artifacts | `$AZ_BATCH_TASK_WORKING_DIR/outputs/*.*` | +| inference | `$AZ_BATCH_TASK_WORKING_DIR/inference/**/*` | +| train | `$AZ_BATCH_TASK_WORKING_DIR/**/*` | + +`LocalRunner._upload_task_outputs` normalizes the same way so the docker dev +stack stays consistent. + +### `ImageryPostProcessor._read_task_output(filename) -> str | None` + +``` +node copy (runner.get_filecontent_from_task) + └─ falsy? → storage.get_file_remote_path( + identifier=filename, + extra_partition_keys=, + data_format=) + → fetch_url_text(url) + └─ any exception in the fallback → warn, return None +``` + +The blob path resolves to `//` because +`UnifiedDataLayer` hashes the partition key with `MetadataUtils.hash_string` — +the same transformation `_execute_image_preprocess` uses to build +`output_prefix`. This is the path `_generate_imagery_url` already reads from in +production, so no new path convention is introduced. + +| Caller | Missing file | +|---|---| +| `_update_results_from_job` | raises `FileNotFoundError` — the layer cannot complete without a manifest | +| `_get_image_preprocess_logs` | returns `[]` — progress detail only | + +### `hastegeo.core.utils.blob.fetch_url_text(url, timeout=30) -> str | None` + +Best-effort HTTP GET. Returns `None` for a falsy URL, a non-`http(s)` scheme +(a data layer resolving to a local filesystem path), or any transport/HTTP +error. It never raises: a failure here must not replace the original reason the +node read failed. + +### `hastegeo.core.utils.errors.describe_exception(exc) -> str` + +| Input | Output | +|---|---| +| `.error.code` + `.error.message.value` | `": "`, service `RequestId:`/`Time:` trailer stripped | +| `.error.code` only | `""` | +| anything else | `str(exc)`, or the type name when empty | + +Matched structurally, not by `isinstance`, so `hastegeo.core.utils` needs no +`azure.batch` import. + +## Queue trigger + +```python +output.statusMessage = MetadataUtils.append_status_message( + output.statusMessage, + f"Image layer processing failed: {describe_exception(e)}", +) +``` + +Appending preserves the progress history the status dialog renders. The +train/embedding/inference triggers already appended; they now use +`describe_exception` for the same readability. + +No business logic was added to `function_app.py` — the formatter lives in +`hastegeo`, per the repository's function-app boundary rule. + +## Rejected alternatives + +| Alternative | Why not | +|---|---| +| Fix the pool (keep nodes alive after task completion) | Burns scarce GPU quota, and preemption of spot nodes remains possible regardless | +| Raise `maxDequeueCount` so the message is redelivered | Redelivery re-runs whole tasks; the fix belongs in-process | +| Add `extra_partition_keys` to `load()` across all five data layers | Wide blast radius for a fallback path; `load()` also returns `None` for non-json/yaml formats, so it cannot read the `.log` | +| A single `**/*` upload pattern for imagery | Would upload the raw downloaded imagery alongside the outputs | +| Poll node state before reading | Racy — the node can go away between the check and the read | diff --git a/spec/features/batch-node-loss-resilience/impact-analysis.md b/spec/features/batch-node-loss-resilience/impact-analysis.md new file mode 100644 index 00000000..dd05f3a3 --- /dev/null +++ b/spec/features/batch-node-loss-resilience/impact-analysis.md @@ -0,0 +1,90 @@ +# Impact Analysis: Batch node-loss resilience + +## Scope of change + +| Component | Path | Type of Change | Severity | +|---|---|---|---| +| Core library — runners | `hastelib/src/hastegeo/core/runners/azure_batch.py` | modified | medium | +| Core library — runners | `hastelib/src/hastegeo/core/runners/local.py` | modified | low | +| Core library — processors | `hastelib/src/hastegeo/core/processors/imagery.py` | modified | medium | +| Core library — utils | `hastelib/src/hastegeo/core/utils/errors.py` | new | low | +| Core library — utils | `hastelib/src/hastegeo/core/utils/blob.py` | modified (additive) | low | +| Queue workers | `api/hastefuncqueues/function_app.py` | modified | low | +| Docs | `docs/api/hastefuncqueues.md`, `api/hastefuncqueues/README.md` | modified | low | + +No infrastructure, Bicep, UI, CI or dependency changes. + +## Azure service impact + +| Service | Change | Cost Impact | +|---|---|---| +| Azure Batch | None to pool or account configuration. Tasks now register one extra `OutputFile` pattern (`logs/*.*`). | negligible | +| Blob Storage | Imagery tasks additionally upload `logs/imagery_friendly.log` (a few KB per layer). One extra GET per layer on the fallback path. | negligible | +| Azure Functions | Node-file reads may now retry up to 5 times with 4–10s backoff (worst case ~40s added per failing read). `functionTimeout` is `23:59:59`, so no timeout risk. | negligible | + +## Dependency analysis + +### Upstream + +| Dependency | Type | Status | Risk if unavailable | +|---|---|---|---| +| `azure-batch==14.2.0` error model (`.error.code`) | library | pinned, available | Classification degrades to "not retryable" — i.e. today's behavior | +| `tenacity` | library | already a dependency | none | +| `requests` | library | already a dependency | fallback returns `None`; manifest miss fails the layer as before | +| Batch `OutputFile` upload on task completion | platform | in use since before this change | fallback finds nothing; behaves as today | + +### Downstream + +| Consumer | How affected | Breaking? | Migration needed? | +|---|---|---|---| +| `ImageryPostProcessor` | gains a blob fallback | no | no | +| `train` / `inference` / `artifacts` / `embedding` processors | inherit the runner-level retry and tolerant cleanup; `get_filecontent_from_task` can now return `None` where it previously raised | no — all callers already branch on falsy content | no | +| `LocalRunner` | `file_pattern` may be a list | no — string callers unchanged | no | +| Existing Cosmos/Blob documents | none — no schema change | no | no | +| Already-failed dev1 layers | not repaired retroactively | no | re-run the layer | + +## Risk assessment + +| Risk | Likelihood | Impact | Mitigation | +|---|---|---|---| +| A returned `None` masks a real failure in a non-imagery workload | low | medium | Only node-unavailability is converted; every other Batch error still propagates. Covered by `test_propagates_unrelated_batch_errors`. | +| Retrying widens the window in which a genuinely broken node stalls a message | low | low | Budget unchanged (5 attempts / 4–10s); `NodeNotFound` is not retried at all | +| Blob fallback reads a stale manifest from a previous task | very low | medium | The path is scoped by `taskId`, which is unique per submission | +| `logs/*.*` upload leaks unintended files | very low | low | The imageryprep container writes only `imagery_friendly.log` there | +| Fallback failure hides the original error | low | medium | `fetch_url_text` and the fallback block never raise; both log and return `None` | +| A layer completes on the blob copy while the node copy was more recent | very low | low | Batch uploads on task completion, so the blob copy is the final state | + +## Performance impact + +- **Happy path:** unchanged — the node copy is still read first, and the fallback only runs when that returns nothing. +- **Failure path:** up to ~40s of backoff per unreadable file before falling back, replacing an immediate permanent failure. +- **Batch compute:** unchanged. No pool, node-count or VM SKU changes. + +## Security impact + +- [ ] New API endpoints exposed? — **no** +- [ ] New data classification handled? — **no** +- [ ] MSAL/Entra ID auth changes? — **no** +- [ ] New secrets or connection strings? — **no**. The fallback reuses the SAS URL the data layer already issues (`get_file_remote_path`), the same one handed to the UI. +- [ ] CORS changes? — **no** +- [ ] New dependencies? — **no** + +`describe_exception` strips the service `RequestId`/`Time` trailer from +user-facing status text; the full exception is still written to the function +logs via `traceback.format_exc()`. + +## Compliance & data impact + +- No change to data residency, retention or partner sharing. +- One additional small log file per imagery layer in the existing outputs + container, under the existing retention policy. +- No new Python or npm dependencies, so no Component Governance implications. + +## Rollback assessment + +- **Reversibility:** fully reversible — code-only, no state or schema migration. +- **Cosmos data:** unaffected. +- **Blob data:** the extra `imagery_friendly.log` blobs are harmless if the + change is reverted; nothing reads them unless the fallback exists. +- **API:** no contract change. +- **Estimated rollback time:** one revert + redeploy (<15 min). diff --git a/spec/features/batch-node-loss-resilience/plan.md b/spec/features/batch-node-loss-resilience/plan.md new file mode 100644 index 00000000..ec6ba304 --- /dev/null +++ b/spec/features/batch-node-loss-resilience/plan.md @@ -0,0 +1,93 @@ +# Execution Plan: Batch node-loss resilience + +## Phases + +### Phase 1: Core Library + +**Goal:** Make the Batch runner tolerate node loss, and let imagery recover its +outputs from blob storage. + +| Task | Agent | Dependencies | Story Ref | Status | +|---|---|---|---|---| +| Classify transient vs terminal node errors in `runners/azure_batch.py` | `backend-dev` | — | US-001 | done | +| Retry transient node errors; `reraise=True` on exhaustion | `backend-dev` | above | US-001 | done | +| `get_filecontent_from_task` returns `None` on node loss | `backend-dev` | above | US-001 | done | +| `cleanup_task` tolerates node loss, still disables the job | `backend-dev` | above | US-004 | done | +| Accept `file_pattern` as `str \| list[str]` in `add_task` | `backend-dev` | — | US-002 | done | +| Mirror list handling in `runners/local.py` | `backend-dev` | above | US-002 | done | +| Add `utils/errors.py::describe_exception` | `backend-dev` | — | US-003 | done | +| Add `utils/blob.py::fetch_url_text` | `backend-dev` | — | US-001 | done | +| Add `ImageryPostProcessor._read_task_output` with blob fallback | `backend-dev` | runner tasks | US-001, US-002 | done | +| Submit `outputs/` **and** `logs/` for upload from imagery | `backend-dev` | `add_task` change | US-002 | done | +| Unit tests: runner node errors | `backend-dev` | runner tasks | US-001, US-004 | done | +| Unit tests: imagery fallback + upload patterns | `backend-dev` | imagery tasks | US-001, US-002 | done | +| Unit tests: exception formatter | `backend-dev` | `utils/errors.py` | US-003 | done | +| Repair the `clipBbox` fixture gap in `test_imagery_preprocess_config.py` | `backend-dev` | — | — | done | + +**Exit Criteria:** +- [x] All new unit tests pass (31) +- [x] No new failures in the `hastelib` suite versus a clean worktree at HEAD +- [x] Core logic works without any API-layer involvement + +### Phase 2: Queue workers + +**Goal:** Stop the imagery trigger from destroying the status history, and make +the recorded cause readable. + +| Task | Agent | Dependencies | Story Ref | Status | +|---|---|---|---|---| +| Append instead of assign `statusMessage` in the imagery trigger | `backend-dev` | Phase 1 | US-003 | done | +| Render the cause via `describe_exception` | `backend-dev` | Phase 1 | US-003 | done | +| Apply the formatter to train / embedding / inference messages | `backend-dev` | Phase 1 | US-003 | done | + +**Exit Criteria:** +- [x] No business logic added to `function_app.py` (formatter lives in `hastegeo`) +- [x] flake8 / black / isort clean on the touched files + +### Phase 3: Docs & spec + +**Goal:** Bring documentation in line with the new read path. + +| Task | Agent | Dependencies | Story Ref | Status | +|---|---|---|---|---| +| Update `api/hastefuncqueues/README.md` (manifest/log source) | `backend-dev` | Phase 1 | US-001 | done | +| Update `docs/api/hastefuncqueues.md` (manifest/log source) | `backend-dev` | Phase 1 | US-001 | done | +| Write this spec set | `backend-dev` | Phases 1–2 | — | done | + +**Exit Criteria:** +- [x] Docs no longer claim the files are read only from the task working directory +- [x] Spec records root cause, design, and rejected alternatives + +## Milestones + +| Milestone | Deliverable | Status | +|---|---|---| +| Root cause confirmed | Traced from the dev1 status message to the node-scoped file APIs | done | +| Core library done | `hastelib` runner + processor + utils changes | done | +| Queue workers done | Readable, appended status messages | done | +| Docs & spec done | Spec set + updated queue docs | done | +| Verified on dev1 | Manual verification per [test-plan.md](test-plan.md#manual-verification-dev1) | pending | + +## Agent Summary + +| Agent | Tasks Owned | Phases | +|---|---|---| +| `backend-dev` | 19 | 1, 2, 3 | +| `backend-validation` | validation of all phases | 1, 2 | +| `orchestrator` | spec status tracking | 3 | + +## Resource Requirements + +- **Agents:** `backend-dev` implements, `backend-validation` validates. +- **Azure services:** none new. No pool, quota or GPU changes. +- **External data:** none. + +## Open Questions + +- [ ] Should the pool stop deallocating on task completion (or hold a floor of + one node) so the race disappears rather than being absorbed? That is an + infrastructure decision, deliberately out of scope here — it would need an + ADR and a quota discussion. +- [ ] Should node loss *during* a task (rather than after it) be surfaced + distinctly? Batch requeues such tasks itself today; no evidence yet that it + needs handling. diff --git a/spec/features/batch-node-loss-resilience/rollout.md b/spec/features/batch-node-loss-resilience/rollout.md new file mode 100644 index 00000000..bb859e56 --- /dev/null +++ b/spec/features/batch-node-loss-resilience/rollout.md @@ -0,0 +1,98 @@ +# Rollout Plan: Batch node-loss resilience + +## Rollout Strategy + +**Type:** big-bang (no feature flag) + +The change is a strict widening of what the system tolerates: the node copy of a +file is still read first, and every non-node Batch error still propagates +exactly as before. There is no configuration surface worth flagging, and a flag +would leave the failing path in place for whoever forgot to turn it on. + +## Deployment Targets + +| Component | Deployment Method | Target | +|---|---|---| +| `hastelib` (wheel) | `hatch build -t wheel` → Function App rebuild | All Function Apps | +| `hastefuncqueues` | GitHub Actions `deploy-apps.yml` | Azure Functions | +| `hastefuncapi` | GitHub Actions `deploy-apps.yml` | Azure Functions (picks up the wheel; no code change) | + +No infrastructure deployment. No Bicep, pool, or app-setting changes. + +## Feature Flags + +None. + +## Rollout Phases + +### Phase 1: dev1 + +- **Target:** dev1 Function Apps + SWA +- **Deployment:** merge to `main` (triggers `deploy-apps.yml`) +- **Success criteria:** + - [ ] A new image layer on a scale-to-zero pool reaches COMPLETED + - [ ] Function logs show the node-unavailable warning followed by a successful + fallback, when the node does disappear + - [ ] `//` in the outputs container contains both + `imagery_manifest.json` and `imagery_friendly.log` + - [ ] A deliberately broken layer (bad imagery URL) shows a readable cause + appended below its prior progress, not a raw SDK dump + - [ ] Training / inference / artifact jobs still complete normally +- **Rollback trigger:** any workload failing with an error that previously + propagated, or a layer completing with an empty/incorrect manifest + +### Phase 2: testing + +- **Target:** testing environment +- **Success criteria:** + - [ ] No new failure modes across a full imagery → label → train → inference run + - [ ] Batch job/task failure rate unchanged or lower + +### Phase 3: production + +- **Target:** production Function Apps +- **Success criteria:** + - [ ] Imagery layer failure rate unchanged or lower + - [ ] No increase in queue-message processing time beyond the retry budget + +## Rollback Plan + +| Step | Action | ETA | +|---|---|---| +| 1 | Revert the PR on `main` | immediate | +| 2 | Redeploy via `deploy-apps.yml` | <15 min | +| 3 | Confirm imagery submission still works | <5 min | + +**Cosmos data rollback required?** No — no schema or document changes. +**Blob artifacts cleanup needed?** No — the extra `imagery_friendly.log` blobs +are inert if the reader is reverted. + +Reverting restores the previous behavior exactly, including the original +failure mode. Nothing written while the change was live becomes unreadable. + +## Monitoring & Alerting + +### Key Metrics to Watch + +| Metric | Source | Expectation | +|---|---|---| +| Image layer FAILED rate | Cosmos / UI | decreases | +| Occurrences of `NodeNotReady` in function logs | Application Insights | may appear as warnings; should no longer coincide with FAILED layers | +| `is unavailable` warning count | Application Insights | indicator of how often the race actually fires — informs whether the pool itself should change | +| Queue message processing duration | Azure Functions metrics | may rise by up to the retry budget on affected messages | +| Batch task failure rate | Azure Batch metrics | unchanged (this change does not affect task execution) | + +### Alerts to Configure + +None new. The existing failure-rate monitoring is sufficient; this change is +expected to reduce, not add, failure signal. + +## Post-Rollout Checklist + +- [ ] Re-run the dev1 image layers that failed with `NodeNotReady` (they are not + repaired retroactively — the `logs/` upload only applies to newly + submitted tasks) +- [ ] Record the observed frequency of the node-unavailable warning, and decide + whether the pool's deallocation policy warrants an ADR +- [ ] `docs/` updated (done as part of this change) +- [ ] Spec status moved to `released` diff --git a/spec/features/batch-node-loss-resilience/test-plan.md b/spec/features/batch-node-loss-resilience/test-plan.md new file mode 100644 index 00000000..47fc8cf5 --- /dev/null +++ b/spec/features/batch-node-loss-resilience/test-plan.md @@ -0,0 +1,130 @@ +# Test Plan: Batch node-loss resilience + +## Strategy + +| Level | Scope | Mechanism | +|---|---|---| +| Unit | Error classification, retry policy, runner degradation, imagery fallback, message formatting | `hastelib/tests`, `unittest.TestCase`, mocked Batch client | +| Manual | End-to-end recovery on a real autoscale/spot pool | dev1 imagery run | + +Live Batch behavior is not simulated in CI: `BatchErrorException` instances are +constructed directly with the codes the service returns, and the Batch client is +mocked. What is verified is HASTE's *reaction* to those codes. + +## Test files + +| File | Covers | +|---|---| +| `hastelib/tests/core/runners/test_azure_batch_node_errors.py` | US-001 (retry), US-004 (cleanup), runner degradation | +| `hastelib/tests/core/processors/test_imagery_output_fallback.py` | US-001 (fallback), US-002 (log + upload patterns) | +| `hastelib/tests/core/utils/test_errors.py` | US-003 (message formatting) | + +## Coverage matrix + +### Error classification + +| ID | Case | Expected | Story | +|---|---|---|---| +| UT-001 | `NodeNotReady` (409) | transient, unavailable, retryable, **not** a server error | US-001 | +| UT-002 | `NodeStateInvalid` (409) | transient, retryable | US-001 | +| UT-003 | `NodeNotFound` (404) | terminal, unavailable, **not** retryable | US-001 | +| UT-004 | 500 `InternalServerError` | still retryable (no regression) | US-001 | +| UT-005 | `TaskNotFound` (404) | neither retryable nor unavailable | US-001 | +| UT-006 | `ValueError` | no error code; not retryable | US-001 | + +### Retry policy + +| ID | Case | Expected | Story | +|---|---|---|---| +| UT-007 | `NodeNotReady` twice, then success | succeeds; 3 attempts | US-001 | +| UT-008 | `NodeNotReady` beyond the budget | raises `BatchErrorException`, **not** `RetryError` | US-001 | +| UT-009 | `TaskNotFound` | raises after exactly 1 attempt | US-001 | + +> The backoff is removed with `retry_with(wait=wait_none())`, so the real +> predicate and stop policy are exercised without the 4–10s sleeps. + +### Runner degradation + +| ID | Case | Expected | Story | +|---|---|---|---| +| UT-010 | Node gone during file listing | `get_filecontent_from_task` returns `None` and warns | US-001 | +| UT-011 | Node gone during file download | returns `None` | US-001 | +| UT-012 | Unrelated Batch error (`JobNotFound`) | propagates | US-001 | +| UT-013 | Node answers normally | chunks are decoded and joined | US-001 | +| UT-014 | Cleanup against a dead node | delete skipped; `disable_job` still called | US-004 | +| UT-015 | Cleanup fails with `OperationTimedOut` | propagates; `disable_job` not called | US-004 | + +### Imagery fallback + +| ID | Case | Expected | Story | +|---|---|---|---| +| UT-016 | Node copy available | node copy used; blob never fetched | US-001 | +| UT-017 | Node copy unavailable | blob copy used; identifier/`taskId`/format passed correctly | US-001 | +| UT-018 | Neither copy available | returns `None` | US-001 | +| UT-019 | Fallback itself raises | returns `None`, does not propagate | US-001 | +| UT-020 | Manifest recovered from blob | `_update_results_from_job` populates the layer | US-001 | +| UT-021 | Manifest lost everywhere | raises `FileNotFoundError` | US-001 | +| UT-022 | Log unavailable everywhere | returns `[]`; no exception | US-002 | +| UT-023 | Log recovered from blob | parsed into `ImageryLogRecord`s | US-002 | +| UT-024 | Submitted output patterns | list containing both `outputs/*.*` and `logs/*.*` | US-002 | + +### Message formatting + +| ID | Case | Expected | Story | +|---|---|---|---| +| UT-025 | Azure-style error with code + message | `": "` | US-003 | +| UT-026 | Message with `RequestId:`/`Time:` trailer | trailer stripped | US-003 | +| UT-027 | Any Azure-style error | no `additional_properties` / `'lang'` leakage | US-003 | +| UT-028 | Code with no message | just the code | US-003 | +| UT-029 | Plain `ValueError` | `str(exc)` | US-003 | +| UT-030 | Exception with no text | the type name; never empty | US-003 | +| UT-031 | Plain-string message | `": "` | US-003 | + +## Regression guard + +The full `hastelib` suite must show no new failures. Measured against a clean +worktree at HEAD, excluding two modules that cannot run outside the conda test +env (`tests/workflows/test_prepare_imagery.py` needs `osgeo`; +`tests/core/processors/test_artifacts.py` needs `pytest-mock`): + +| Run | Passed | Failed | +|---|---|---| +| Baseline (HEAD) | 155 | 2 | +| With this change | 188 | 0 | + +The two baseline failures were in +`test_imagery_preprocess_config.py`, which exercises +`_execute_image_preprocess` — the same method this change touches. They were a +fixture gap, not a product bug: `MagicMock(spec=ImageLayer)` does not expose +pydantic field names, so the mock had no `clipBbox` attribute once that field +was added to the submitted config. Fixed here with a one-line fixture addition +so the imagery suite is green and future readers are not left wondering whether +this change broke it. + +## Commands + +```bash +# Targeted +cd hastelib && hatch run test:pytest \ + tests/core/runners/test_azure_batch_node_errors.py \ + tests/core/processors/test_imagery_output_fallback.py \ + tests/core/utils/test_errors.py -v + +# Full suite +cd hastelib && hatch run test:pytest +``` + +## Manual verification (dev1) + +| Step | Expectation | +|---|---| +| 1. Submit an image layer on a pool that scales to zero | Task runs and completes | +| 2. Let the node deallocate before the trigger reads outputs | Function logs warn that the node is unavailable | +| 3. Observe the layer | COMPLETED, with the manifest recovered from blob | +| 4. Inspect `//` in the outputs container | Contains `imagery_manifest.json` **and** `imagery_friendly.log` | +| 5. Force a genuine failure (bad imagery URL) | Status dialog shows the readable cause appended below prior progress | + +## Out of scope + +- Simulating real node preemption in CI. +- Load/performance testing — the change adds at most one HTTP GET per layer. diff --git a/spec/features/batch-node-loss-resilience/user-stories.md b/spec/features/batch-node-loss-resilience/user-stories.md new file mode 100644 index 00000000..093e9d92 --- /dev/null +++ b/spec/features/batch-node-loss-resilience/user-stories.md @@ -0,0 +1,202 @@ +# User Stories: Batch node-loss resilience + +## Personas + +| Persona | Description | Key Goals | +|---|---|---| +| Disaster Analyst | Domain expert who uploads imagery and interprets damage maps | Imagery processing that completes, and status text that explains itself | +| ML Engineer | Runs training and inference on Batch pools | Jobs that are not lost to infrastructure churn | + +--- + +## Stories + +### US-001: Imagery processing survives the node going away + +**As a** Disaster Analyst, +**I want to** have my image layer finish processing even when the Batch node +that ran it is deallocated or preempted, +**So that** I am not forced to re-upload and re-run imagery that already +processed successfully. + +**Priority:** P0 +**Estimate:** M +**Component(s):** `hastelib/core/runners/azure_batch.py`, +`hastelib/core/processors/imagery.py` + +**Acceptance Criteria:** + +```gherkin +Given an imagery preprocessing task that completed successfully +And the compute node that ran it is transitioning (starting, rebooting) +When the processor reads imagery_manifest.json from the node +Then the read is retried +And the layer completes normally once the node answers +``` + +```gherkin +Given an imagery preprocessing task that completed successfully +And the node that ran it has been deallocated or preempted +When the processor reads imagery_manifest.json from the node +Then the manifest is read from the copy Azure Batch uploaded to blob storage +And the image layer is marked COMPLETED +``` + +```gherkin +Given an imagery preprocessing task whose manifest is available nowhere +When the processor tries the node and then blob storage +Then the layer is marked FAILED with a readable cause +``` + +**Notes:** The manifest is required for the layer to complete; the fallback path +is the recovery, not a silent skip. + +--- + +### US-002: A lost progress log never fails a good layer + +**As a** Disaster Analyst, +**I want** a missing progress log to cost me detail, not the whole layer, +**So that** infrastructure churn cannot discard imagery that was produced +correctly. + +**Priority:** P1 +**Estimate:** S +**Component(s):** `hastelib/core/processors/imagery.py` + +**Acceptance Criteria:** + +```gherkin +Given a completed imagery task +And imagery_friendly.log cannot be read from the node or from blob storage +When the processor collects progress records +Then no records are added +And the layer still completes +``` + +```gherkin +Given a completed imagery task whose node is gone +When the processor collects progress records +Then the log is read from blob storage +And its entries appear in the layer status history +``` + +**Notes:** Requires `logs/` to be included in the task's `OutputFile` patterns; +before this change the log existed only on the node. + +--- + +### US-003: Failure messages stay readable and keep their history + +**As a** Disaster Analyst, +**I want** the status dialog to show what went wrong in plain language, on top +of the progress that already happened, +**So that** I can tell whether the failure was mine (bad imagery) or the +platform's. + +**Priority:** P1 +**Estimate:** S +**Component(s):** `api/hastefuncqueues/function_app.py`, +`hastelib/core/utils/errors.py` + +**Acceptance Criteria:** + +```gherkin +Given an image layer that recorded progress messages +When processing fails with an Azure Batch error +Then the failure is appended to the existing status history +And it is rendered as ": " without the SDK object dump +And without the RequestId/Time trailer +``` + +```gherkin +Given processing fails with a non-Azure exception +When the status message is written +Then it falls back to the exception's own text +And is never empty +``` + +--- + +### US-004: Post-task cleanup never fails a workload + +**As an** ML Engineer, +**I want** working-directory cleanup against a dead node to be a no-op, +**So that** a completed training, inference, artifact or imagery job is not +marked failed by its own cleanup step. + +**Priority:** P1 +**Estimate:** S +**Component(s):** `hastelib/core/runners/azure_batch.py` + +**Acceptance Criteria:** + +```gherkin +Given a task whose node is no longer available +When cleanup_task runs +Then the working-directory delete is skipped with a warning +And the job is still disabled +``` + +```gherkin +Given cleanup fails with an error unrelated to node availability +When cleanup_task runs +Then the error propagates +``` + +**Notes:** Disk on a dead node is reclaimed by Batch; `task_retention_time` is +`P2D`. + +--- + +## Agent Assignment Map + +### Available Agents + +| Agent | Scope | Touches Code? | +|---|---|---| +| `backend-dev` | Python backend, API, processors, data layers, runners | Yes | +| `backend-validation` | Validates backend code against specs, conventions, tests | No (validates only) | +| `orchestrator` | Records what agents did, when, why. Tracks spec status. | No (observes only) | + +### Story → Agent Mapping + +| Story | Implementing Agent(s) | Validating Agent(s) | Notes | +|---|---|---|---| +| US-001 | `backend-dev` | `backend-validation` | `hastelib/` runner + imagery processor | +| US-002 | `backend-dev` | `backend-validation` | Output-upload patterns + best-effort log read | +| US-003 | `backend-dev` | `backend-validation` | `api/hastefuncqueues/` + new `utils/errors.py` | +| US-004 | `backend-dev` | `backend-validation` | Runner cleanup path, all workloads | + +> `gis` is **not** assigned: no imagery, GDAL or provider-adapter logic changes — +> only how already-produced outputs are retrieved. +> `ui` is **not** assigned: the status dialog renders `statusMessage` unchanged; +> only the text written into it improves. +> `security` is **not** assigned: no new dependencies. + +### Agent Workflow Per Phase + +| Phase | Lead Agent | Supporting Agents | Validation | +|---|---|---|---| +| Phase 1 — Core Library | `backend-dev` | — | `backend-validation` | +| Phase 2 — Queue workers | `backend-dev` | — | `backend-validation` | +| Phase 3 — Docs & spec | `backend-dev` | — | `orchestrator` | + +## Story Map + +| Priority | Story | Phase | Implementing Agent | Component | +|---|---|---|---|---| +| P0 | US-001 | Phase 1 — Core Library | `backend-dev` | `hastelib` | +| P1 | US-002 | Phase 1 — Core Library | `backend-dev` | `hastelib` | +| P1 | US-003 | Phase 2 — Queue workers | `backend-dev` | `hastefuncqueues` | +| P1 | US-004 | Phase 1 — Core Library | `backend-dev` | `hastelib` | + +## Out of Scope + +- [ ] Changing the Batch pool's deallocation policy, `minNodes`, or + dedicated-vs-spot node type — the race is made survivable, not impossible. +- [ ] Raising `maxDequeueCount` so failed queue messages are redelivered — + redelivery would re-run whole tasks. +- [ ] A blob fallback for train/inference/artifacts/embedding — they inherit the + runner-level fixes; only imagery reads a required output file back. +- [ ] Recovering image layers that already failed on dev1 — they must be re-run.