diff --git a/.github/scripts/deploy_apps.sh b/.github/scripts/deploy_apps.sh index 9804d70..c439f48 100644 --- a/.github/scripts/deploy_apps.sh +++ b/.github/scripts/deploy_apps.sh @@ -134,7 +134,8 @@ deploy_function() { "STATIC_APP_NAME=$STATIC_WEB_APP" \ "STATIC_APP_DOMAIN=${STATIC_APP_DOMAIN}" \ "EMAIL_CONNECTION_STRING=${EMAIL_CONNECTION_STRING}" \ - "EMAIL_SENDER=${EMAIL_SENDER}" + "EMAIL_SENDER=${EMAIL_SENDER}" \ + --output none fi az functionapp restart --name "$FUNCTION_NAME" --resource-group "$RESOURCE_GROUP" diff --git a/.github/workflows/deploy-apps.yml b/.github/workflows/deploy-apps.yml index 5348363..d5dec0f 100644 --- a/.github/workflows/deploy-apps.yml +++ b/.github/workflows/deploy-apps.yml @@ -116,15 +116,16 @@ jobs: # on Azure's remote build. HASTEGEO_WHEEL_URL: ${{ steps.hastegeo.outputs.url }} # Batch pool wiring. All optional: unset reproduces the legacy - # single-pool behavior. Set these as GitHub Environment variables to - # point an environment at pre-created shared pools. - BATCH_TRAINING_POOL_ID: ${{ vars.BATCH_TRAINING_POOL_ID }} - BATCH_IMAGERYPREP_POOL_ID: ${{ vars.BATCH_IMAGERYPREP_POOL_ID }} - BATCH_TRAINING_POOL_IDS: ${{ vars.BATCH_TRAINING_POOL_IDS }} - BATCH_INFERENCE_POOL_IDS: ${{ vars.BATCH_INFERENCE_POOL_IDS }} - BATCH_IMAGERYPREP_POOL_IDS: ${{ vars.BATCH_IMAGERYPREP_POOL_IDS }} - BATCH_USE_SAS: ${{ vars.BATCH_USE_SAS }} - BATCH_MANAGE_POOLS: ${{ vars.BATCH_MANAGE_POOLS }} + # single-pool behavior. Stored as environment *secrets* (not + # variables) so the values are masked in this repo's public + # Actions logs, matching the other infra-naming config. + BATCH_TRAINING_POOL_ID: ${{ secrets.BATCH_TRAINING_POOL_ID }} + BATCH_IMAGERYPREP_POOL_ID: ${{ secrets.BATCH_IMAGERYPREP_POOL_ID }} + BATCH_TRAINING_POOL_IDS: ${{ secrets.BATCH_TRAINING_POOL_IDS }} + BATCH_INFERENCE_POOL_IDS: ${{ secrets.BATCH_INFERENCE_POOL_IDS }} + BATCH_IMAGERYPREP_POOL_IDS: ${{ secrets.BATCH_IMAGERYPREP_POOL_IDS }} + BATCH_USE_SAS: ${{ secrets.BATCH_USE_SAS }} + BATCH_MANAGE_POOLS: ${{ secrets.BATCH_MANAGE_POOLS }} run: | chmod +x .github/scripts/deploy_apps.sh .github/scripts/deploy_apps.sh \ diff --git a/CHANGELOG.md b/CHANGELOG.md index 24fc087..4bbdb98 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -17,7 +17,7 @@ Versioning follows the Docker image tags defined in the CI workflows (see [.gith > **Operator action:** rename the `AZURE_BATCH_REGISTRY_SERVER_URL` application setting to `AZURE_BATCH_REGISTRY_SERVER` (value: the bare login server, e.g. `myacr.azurecr.io`) on the `api` and `queues` Function Apps. Confirm `AZURE_BATCH_TRAINING_POOL_ID` / `AZURE_BATCH_IMAGERYPREP_POOL_ID` name pools that still exist and are **identical across both apps** — they double as the default Batch job ids. -- **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. The runner now rebinds the job to the pool selected for the submission, and fails loudly when Batch refuses. +- **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. - **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). diff --git a/docs/configuration.md b/docs/configuration.md index e278cb1..476b715 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -176,19 +176,29 @@ The runner picks a pool from the candidate list **at submit time** — the first with an idle node, otherwise the preferred (first) pool, which scales up / queues. These are emitted by both deploy paths: set them as GitHub Environment -variables (`BATCH_TRAINING_POOL_IDS`, `BATCH_INFERENCE_POOL_IDS`, +**secrets** (`BATCH_TRAINING_POOL_IDS`, `BATCH_INFERENCE_POOL_IDS`, `BATCH_IMAGERYPREP_POOL_IDS`, `BATCH_USE_SAS`, `BATCH_MANAGE_POOLS`, `BATCH_TRAINING_POOL_ID`, `BATCH_IMAGERYPREP_POOL_ID`) for [`deploy-apps.yml`](../.github/workflows/deploy-apps.yml), or as the corresponding `infra/main.bicepparam` values for the Bicep path. +> **Secrets, not variables.** Resource naming is treated as sensitive in this +> repo (as `RESOURCE_PREFIX`, `LOCATION` and friends already are). GitHub masks +> secret values in Actions logs but does **not** mask variables, and this repo's +> workflow logs are public. + > **`*_POOL_ID` (singular) also names the Batch job.** `TRAINING_BATCH_JOB_ID` > and friends default to the matching pool id, so the value must be **identical > on the api and queues apps** — the queues app submits under that job id and -> the api app reads status back from it. Pointing it at a pool that no longer -> exists leaves a job permanently bound to a deleted pool; the runner now -> rebinds such a job to the pool it selected, but the setting should still be -> corrected. +> the api app reads status back from it. + +**Job ids are scoped to the selected pool.** A Batch job is permanently bound to +the pool it was created against and can only be re-pointed while it has no +active tasks, so one static job id cannot span pools. When routing across +multiple candidates, the runner derives the job id from the pool the task was +routed to — one job per pool. Environments with a single candidate pool keep +their existing job id unchanged. This is why a task that spills over to a +second pool no longer collides with the job created on the preferred pool. ### Keeping settings in sync diff --git a/hastelib/src/hastegeo/core/processors/artifacts.py b/hastelib/src/hastegeo/core/processors/artifacts.py index 489b306..89710fa 100644 --- a/hastelib/src/hastegeo/core/processors/artifacts.py +++ b/hastelib/src/hastegeo/core/processors/artifacts.py @@ -310,7 +310,7 @@ def submit_zip_job(self): task_id = f"{ZIP_PREFIX}-{MetadataUtils.generate_id()}" zip_output_prefix = f"{MetadataUtils.hash_string(self.model_artifacts.projectId)}/{task_id}" - self.runner.add_task( + job_id, task_id = self.runner.add_task( job_id=job_id, task_id=task_id, output_prefix=zip_output_prefix, diff --git a/hastelib/src/hastegeo/core/processors/embedding.py b/hastelib/src/hastegeo/core/processors/embedding.py index cfd18c6..9afc8bc 100644 --- a/hastelib/src/hastegeo/core/processors/embedding.py +++ b/hastelib/src/hastegeo/core/processors/embedding.py @@ -169,7 +169,7 @@ def _execute_embedding(self): f"{MetadataUtils.hash_string(self.model_data.projectId)}" f"/{task_id}" ) - self.runner.add_task( + job_id, task_id = self.runner.add_task( job_id=job_id, task_id=task_id, output_prefix=output_prefix, diff --git a/hastelib/src/hastegeo/core/processors/imagery.py b/hastelib/src/hastegeo/core/processors/imagery.py index 24ad74f..5c8f9c4 100644 --- a/hastelib/src/hastegeo/core/processors/imagery.py +++ b/hastelib/src/hastegeo/core/processors/imagery.py @@ -338,7 +338,7 @@ def _execute_image_preprocess(self): imagery_output_prefix = ( f"{MetadataUtils.hash_string(self.image_data.projectId)}/{task_id}" ) - self.runner.add_task( + job_id, task_id = self.runner.add_task( job_id=job_id, task_id=task_id, output_prefix=imagery_output_prefix, diff --git a/hastelib/src/hastegeo/core/processors/inference.py b/hastelib/src/hastegeo/core/processors/inference.py index 5924517..903ba5f 100644 --- a/hastelib/src/hastegeo/core/processors/inference.py +++ b/hastelib/src/hastegeo/core/processors/inference.py @@ -289,7 +289,7 @@ def _execute_inference(self): task_id = f"{INFERENCE_PREFIX}-{MetadataUtils.generate_id()}" inference_output_prefix = f"{MetadataUtils.hash_string(self.model_data.projectId)}/{task_id}" - self.runner.add_task( + job_id, task_id = self.runner.add_task( job_id=job_id, task_id=task_id, output_prefix=inference_output_prefix, diff --git a/hastelib/src/hastegeo/core/processors/train.py b/hastelib/src/hastegeo/core/processors/train.py index fe18a0d..e062455 100644 --- a/hastelib/src/hastegeo/core/processors/train.py +++ b/hastelib/src/hastegeo/core/processors/train.py @@ -265,7 +265,7 @@ def _execute_training(self): job_id = job_id[:64] task_id = f"{TRAINING_PREFIX}-{MetadataUtils.generate_id()}" training_output_prefix = f"{MetadataUtils.hash_string(self.model_data.projectId)}/{task_id}" - self.runner.add_task( + job_id, task_id = self.runner.add_task( job_id=job_id, task_id=task_id, output_prefix=training_output_prefix, diff --git a/hastelib/src/hastegeo/core/runners/azure_batch.py b/hastelib/src/hastegeo/core/runners/azure_batch.py index f87a52c..cf131f7 100644 --- a/hastelib/src/hastegeo/core/runners/azure_batch.py +++ b/hastelib/src/hastegeo/core/runners/azure_batch.py @@ -49,7 +49,7 @@ ) from hastegeo.core.config import Config from hastegeo.core.utils.batch_config import ( - BatchJobPoolMismatchError, + resolve_job_id, validate_batch_config, ) from hastegeo.core.utils.logs import Logger @@ -163,6 +163,11 @@ def add_task( task_id, ) + # A Batch job is pinned to one pool, so the job id has to follow the + # pool this task was routed to; otherwise a second task that spills over + # to another pool collides with the job the first one created. + job_id = resolve_job_id(job_id, selected_pool, self.candidate_pool_ids) + # Pre-created IaC/autoscale pools manage their own lifecycle; only # auto-create/resize for legacy single-pool envs (manage_pools=True). if self.manage_pools: @@ -194,7 +199,7 @@ def add_task( self.logger.info( "Creating job for job_id: %s and task_id: %s", job_id, task_id ) - self.batch_cluster.create_job(job_id) + job_id = self.batch_cluster.create_job(job_id) self.batch_cluster.add_task( job_id=job_id, @@ -451,6 +456,12 @@ def create_pool_if_not_exists( raise def create_job(self, job_id): + """Ensure an active job bound to ``self.pool_id``. + + Returns the job id that was actually used, which may differ from the + requested one when an existing job is pinned to another pool and cannot + be re-pointed. + """ # Fixed pools we manage need a ready node before the job is created; # autoscale / pre-created pools scale up in response to queued tasks, so # waiting here would deadlock (0 nodes until tasks exist). @@ -458,48 +469,70 @@ def create_job(self, job_id): self.wait_for_pool_to_be_ready() try: job = self.batch_client.job.get(job_id) - self._rebind_job_pool(job) - if job.state != JobState.active: - self.batch_client.job.enable(job_id) - self.logger.info(f"Job {job_id} activated.") except BatchErrorException as e: if e.error.code == "JobNotFound": - job = JobAddParameter( - id=job_id, pool_info=PoolInformation(pool_id=self.pool_id) - ) - self.batch_client.job.add(job) - self.logger.info(f"Job {job_id} created.") - - def _rebind_job_pool(self, job): - # A job keeps the pool it was first created against, so reusing a job id - # would pin every later task to that original pool -- defeating - # capacity-aware routing, and failing outright once that pool is renamed - # or deleted. Re-point the job at the pool selected for this submission. + self._add_job(job_id) + return job_id + raise + bound_pool = getattr(job.pool_info, "pool_id", None) if not bound_pool or bound_pool == self.pool_id: - return + if job.state != JobState.active: + self.batch_client.job.enable(job_id) + self.logger.info(f"Job {job_id} activated.") + return job_id + + # The job belongs to a different pool. Re-pointing it only works while + # it has no active tasks, so fall back to a pool-scoped job rather than + # failing the submission. self.logger.info( "Job %s is bound to pool %s; rebinding to selected pool %s.", - job.id, + job_id, bound_pool, self.pool_id, ) try: self.batch_client.job.patch( - job.id, + job_id, JobPatchParameter( pool_info=PoolInformation(pool_id=self.pool_id) ), ) + if job.state != JobState.active: + self.batch_client.job.enable(job_id) + return job_id except BatchErrorException as e: - # Batch refuses the change while tasks are active. Surface it rather - # than silently submitting into a job on the wrong pool. - raise BatchJobPoolMismatchError( - f"Job {job.id} is bound to pool {bound_pool} but this task " - f"targets pool {self.pool_id}, and rebinding failed " - f"({getattr(e.error, 'code', e)}). Delete or drain the job, " - "or set a distinct *_BATCH_JOB_ID." - ) from e + fallback_id = resolve_job_id( + job_id, self.pool_id, [job_id, self.pool_id] + ) + self.logger.info( + "Rebinding job %s failed (%s); using pool-scoped job %s.", + job_id, + getattr(e.error, "code", e), + fallback_id, + ) + return self._ensure_job_on_pool(fallback_id) + + def _ensure_job_on_pool(self, job_id): + # Last step of the fallback: the id is derived from the pool, so it can + # only be missing or already bound to that same pool. + try: + job = self.batch_client.job.get(job_id) + except BatchErrorException as e: + if e.error.code == "JobNotFound": + self._add_job(job_id) + return job_id + raise + if job.state != JobState.active: + self.batch_client.job.enable(job_id) + return job_id + + def _add_job(self, job_id): + job = JobAddParameter( + id=job_id, pool_info=PoolInformation(pool_id=self.pool_id) + ) + self.batch_client.job.add(job) + self.logger.info(f"Job {job_id} created on pool {self.pool_id}.") def add_task( self, diff --git a/hastelib/src/hastegeo/core/utils/batch_config.py b/hastelib/src/hastegeo/core/utils/batch_config.py index fdb792c..4ca5e94 100644 --- a/hastelib/src/hastegeo/core/utils/batch_config.py +++ b/hastelib/src/hastegeo/core/utils/batch_config.py @@ -8,12 +8,19 @@ until Azure rejects it — surfacing as an opaque API error far from the cause (e.g. ``InvalidPropertyValue`` on ``registryServer``). Validating before the first Batch call lets the failure name the setting that is actually missing. + +This module also resolves the Batch *job* id, which has to account for +capacity-aware routing: a Batch job is permanently bound to one pool, but the +pool is chosen per task, so a single static job id cannot span pools. """ import re PLACEHOLDER_PATTERN = re.compile(r"<[^<>]+>") +# Azure Batch job ids are limited to 64 characters. +MAX_JOB_ID_LENGTH = 64 + # Settings the runner needs for any submission, mapped to the application # setting that supplies each one. ALWAYS_REQUIRED = { @@ -37,10 +44,6 @@ class BatchConfigurationError(RuntimeError): """Raised when a required Azure Batch application setting is missing.""" -class BatchJobPoolMismatchError(RuntimeError): - """Raised when a reused Batch job is pinned to a different pool.""" - - def has_placeholder(value): """True when ``value`` still contains an unresolved ``<...>`` default.""" return isinstance(value, str) and bool(PLACEHOLDER_PATTERN.search(value)) @@ -75,3 +78,41 @@ def validate_batch_config(batch_config, manage_pools=None): + "; ".join(unresolved) + ". Set these on the Function App and restart it." ) + + +def resolve_job_id(base_job_id, selected_pool, candidate_pool_ids=None): + """Return the Batch job id to use for a task routed to ``selected_pool``. + + A Batch job is permanently bound to the pool it was created against, and + can only be re-pointed while it has no active tasks. Capacity-aware routing + picks the pool per task, so reusing one static job id across pools breaks as + soon as two tasks are in flight on different pools. Scoping the job id to + the selected pool gives one job per pool and removes the conflict. + + Environments that are not routing across multiple pools keep their existing + job id, so their jobs are not renamed. + + Args: + base_job_id: The configured job id (e.g. ``IMAGERYPREP_BATCH_JOB_ID``). + selected_pool: The pool this task was routed to. + candidate_pool_ids: The pools routing may choose between. + """ + if not selected_pool: + return base_job_id + candidates = list(candidate_pool_ids or []) + + # The default convention is job id == pool id, so follow the selected pool + # and keep names clean rather than doubling the pool id up. + if base_job_id in candidates: + return selected_pool[:MAX_JOB_ID_LENGTH] + + # A single candidate cannot spill over, so leave custom ids untouched. + if len(candidates) <= 1: + return base_job_id + + # Reserve room for the suffix: truncating the base instead of the pool + # keeps ids for two different pools distinct. + room = MAX_JOB_ID_LENGTH - len(selected_pool) - 1 + if room <= 0: + return selected_pool[:MAX_JOB_ID_LENGTH] + return f"{base_job_id[:room]}-{selected_pool}" diff --git a/hastelib/tests/core/utils/test_batch_config.py b/hastelib/tests/core/utils/test_batch_config.py index aca64b0..a745d74 100644 --- a/hastelib/tests/core/utils/test_batch_config.py +++ b/hastelib/tests/core/utils/test_batch_config.py @@ -15,11 +15,11 @@ import pytest from azure.batch.models import BatchErrorException, JobState from hastegeo.core.config import REGISTRY_SERVER_PLACEHOLDER, Config -from hastegeo.core.runners.azure_batch import AzureBatchJob +from hastegeo.core.runners.azure_batch import AzureBatchJob, AzureBatchRunner from hastegeo.core.utils.batch_config import ( BatchConfigurationError, - BatchJobPoolMismatchError, has_placeholder, + resolve_job_id, validate_batch_config, ) @@ -144,7 +144,7 @@ def test_create_job_rebinds_a_job_pinned_to_a_stale_pool(): job = _job(pool_id="selected-pool") job.batch_client.job.get.return_value = _existing_job("deleted-pool") - job.create_job("job-1") + assert job.create_job("job-1") == "job-1" job.batch_client.job.patch.assert_called_once() patched = job.batch_client.job.patch.call_args[0][1] @@ -155,17 +155,118 @@ def test_create_job_does_not_rebind_when_pool_already_matches(): job = _job(pool_id="selected-pool") job.batch_client.job.get.return_value = _existing_job("selected-pool") - job.create_job("job-1") + assert job.create_job("job-1") == "job-1" job.batch_client.job.patch.assert_not_called() -def test_create_job_raises_when_rebinding_is_refused(): - job = _job(pool_id="selected-pool") - job.batch_client.job.get.return_value = _existing_job("deleted-pool") +def test_create_job_falls_back_to_pool_scoped_job_when_rebinding_refused(): + # Batch refuses to re-point a job that still has active tasks. Rather than + # failing the submission, fall back to a job scoped to the selected pool. + job = _job(pool_id="t4-pool") error = BatchErrorException(lambda *a, **k: None, MagicMock()) - error.error = MagicMock(code="JobStateInvalid") + error.error = MagicMock(code="OperationInvalidForCurrentState") job.batch_client.job.patch.side_effect = error - with pytest.raises(BatchJobPoolMismatchError): - job.create_job("job-1") + not_found = BatchErrorException(lambda *a, **k: None, MagicMock()) + not_found.error = MagicMock(code="JobNotFound") + job.batch_client.job.get.side_effect = [ + _existing_job("h100-pool"), + not_found, + ] + + used = job.create_job("h100-pool") + + assert used == "t4-pool" + added = job.batch_client.job.add.call_args[0][0] + assert added.id == "t4-pool" + assert added.pool_info.pool_id == "t4-pool" + + +def test_create_job_reuses_existing_pool_scoped_job_on_fallback(): + job = _job(pool_id="t4-pool") + error = BatchErrorException(lambda *a, **k: None, MagicMock()) + error.error = MagicMock(code="OperationInvalidForCurrentState") + job.batch_client.job.patch.side_effect = error + job.batch_client.job.get.side_effect = [ + _existing_job("h100-pool"), + _existing_job("t4-pool"), + ] + + assert job.create_job("h100-pool") == "t4-pool" + job.batch_client.job.add.assert_not_called() + + +def test_resolve_job_id_follows_selected_pool_by_default_convention(): + # Job ids default to the pool id, so track whichever pool was selected + # rather than doubling the pool id up. + assert ( + resolve_job_id("h100-pool", "t4-pool", ["h100-pool", "t4-pool"]) + == "t4-pool" + ) + + +def test_resolve_job_id_leaves_single_pool_environments_untouched(): + # No spillover is possible, so an existing custom job id is not renamed. + assert resolve_job_id("my-job", "only-pool", ["only-pool"]) == "my-job" + assert resolve_job_id("my-job", "only-pool", []) == "my-job" + + +def test_resolve_job_id_scopes_custom_job_ids_when_routing(): + assert ( + resolve_job_id("my-job", "t4-pool", ["h100-pool", "t4-pool"]) + == "my-job-t4-pool" + ) + + +def test_resolve_job_id_respects_the_64_character_batch_limit(): + base = "b" * 60 + pool = "p" * 20 + resolved = resolve_job_id(base, pool, ["other-pool", pool]) + assert len(resolved) <= 64 + assert resolved.endswith(pool) + + +def test_resolve_job_id_keeps_different_pools_distinct_when_truncating(): + # Truncating the base rather than the pool is what keeps two pools from + # collapsing onto the same job id. + base = "b" * 60 + a = resolve_job_id(base, "pool-aaaaaaaaaaaaaaaaaaaa", ["x", "y"]) + b = resolve_job_id(base, "pool-bbbbbbbbbbbbbbbbbbbb", ["x", "y"]) + assert a != b + assert len(a) <= 64 and len(b) <= 64 + + +def test_spillover_to_a_second_pool_uses_a_separate_job(monkeypatch): + # Regression: a task that spills over to another pool must not collide with + # the job the first task created on the preferred pool. + monkeypatch.setenv("AZURE_BATCH_ACCOUNT_NAME", "acct") + monkeypatch.setenv("AZURE_BATCH_ACCOUNT_KEY", "key") + monkeypatch.setenv( + "AZURE_BATCH_URL", "https://acct.westus2.batch.azure.com" + ) + monkeypatch.setenv( + "AZURE_BATCH_OUTPUT_CONTAINER_URL", + "https://sa.blob.core.windows.net/data", + ) + monkeypatch.setenv("AZURE_BATCH_MANAGE_POOLS", "false") + monkeypatch.setenv("AZURE_BATCH_IMAGERYPREP_POOL_ID", "h100-pool") + + runner = AzureBatchRunner( + pool_id="h100-pool", candidate_pool_ids=["h100-pool", "t4-pool"] + ) + runner.batch_cluster = MagicMock() + runner.batch_cluster.select_pool.return_value = "t4-pool" + runner.batch_cluster.create_job.side_effect = lambda jid: jid + + job_id, _ = runner.add_task( + job_id="h100-pool", + task_id="img-1", + image_name="acr.azurecr.io/img:1", + command="run", + arguments=[], + output_container_url="https://sa.blob.core.windows.net/data", + ) + + assert job_id == "t4-pool" + runner.batch_cluster.create_job.assert_called_once_with("t4-pool") diff --git a/spec/features/batch-config-drift/design.md b/spec/features/batch-config-drift/design.md index ba51d37..065df83 100644 --- a/spec/features/batch-config-drift/design.md +++ b/spec/features/batch-config-drift/design.md @@ -55,8 +55,12 @@ That silently defeats capacity-aware routing (the job stays on whichever pool created it, ignoring `select_pool`) and hard-fails once the original pool is renamed or deleted — tasks queue into a job bound to a pool that no longer exists. `_rebind_job_pool` now patches `pool_info` when the existing binding -differs from the selected pool, and raises `BatchJobPoolMismatchError` when -Batch refuses (e.g. active tasks) rather than submitting to the wrong pool. +differs from the selected pool, and falls back to a pool-scoped job id when +Batch refuses (which it does whenever the job still has active tasks). + +> Superseded by [`../batch-pool-job-binding/`](../batch-pool-job-binding/): +> rebinding cannot work while a job is running, so job ids are now scoped to +> the selected pool instead. ## 4. Deploy-path parity diff --git a/spec/features/batch-pool-job-binding/README.md b/spec/features/batch-pool-job-binding/README.md new file mode 100644 index 0000000..f501499 --- /dev/null +++ b/spec/features/batch-pool-job-binding/README.md @@ -0,0 +1,90 @@ +# Batch job / pool binding + +**Status:** in-progress +**Type:** modification (fix) +**Supersedes:** the `_rebind_job_pool` approach in +[`../batch-config-drift/`](../batch-config-drift/) +**Related:** [`../batch-compute-expansion/`](../batch-compute-expansion/) +(capacity-aware routing) + +## Problem + +Two image layers submitted close together. The first ran; the second failed +before reaching Batch: + +``` +Job is bound to pool but this task targets pool +, and rebinding failed (OperationInvalidForCurrentState). +``` + +Capacity-aware spillover therefore broke in exactly the concurrent case it +exists to serve. + +## Root cause + +Three properties collide: + +1. **The Batch job id is static** — it comes from configuration and defaults to + the singular pool id (`config.py`, `*_BATCH_JOB_ID`). +2. **The pool is chosen per task** — `select_pool` returns the first candidate + with an idle node. +3. **A Batch job is permanently bound to one pool** — it can only be re-pointed + while it has no active tasks. + +| Step | Result | +|---|---| +| Task A | `t4` has no idle node → routed to `h100`; job created **bound to h100** | +| Task B (while A runs) | `h100` busy, `t4` idle → routed to **t4** | +| | Same static job id, but that job is bound to `h100` | +| Rebind attempt | `OperationInvalidForCurrentState` — job has active tasks | +| | Submission fails | + +Rebinding was the wrong remedy: **one static job id cannot span multiple +pools**, and the moment concurrency exists it cannot be rebound. + +## Design + +**One job per pool.** `resolve_job_id(base, selected_pool, candidates)` in +`hastegeo.core.utils.batch_config` derives the job id from the routed pool: + +| Case | Result | +|---|---| +| `base` is one of the candidate pool ids (the default convention) | the selected pool id | +| single candidate | `base` unchanged — legacy environments are not renamed | +| multiple candidates + custom `base` | `"-"`, trimmed to 64 | + +Truncation removes characters from **`base`**, never the pool, so two different +pools can never collapse onto the same job id under the 64-character Batch +limit. + +`AzureBatchJob.create_job` is now **graceful and returns the job id it actually +used**: it reuses a job whose binding already matches, rebinds one that is idle, +and otherwise falls back to a pool-scoped job instead of raising. + +### Callers must capture the returned job id + +`add_task` returns `(job_id, task_id)`, and the returned id may differ from the +one passed in. All five processors persist `jobId` into metadata, which is later +used for status polling and log retrieval, so each captures the returned value: +`imagery`, `train`, `inference`, `artifacts`, `embedding`. + +## Non-goals + +- Changing `select_pool`'s routing policy. +- Job cleanup/retention: jobs left bound to deleted pools are simply no longer + selected, and Batch retention handles them. + +## Agent assignment + +| Area | Implements | Validates | +|---|---|---| +| `hastelib/` runner + processors | `backend-dev` | `backend-validation` | +| Docs + spec | `backend-dev` | `orchestrator` | + +## Acceptance criteria + +1. A task routed to a spillover pool uses a job scoped to that pool. +2. Single-candidate environments keep their existing job id. +3. Resolved job ids never exceed 64 characters, and remain distinct per pool. +4. `create_job` never fails a submission over a recoverable pool mismatch. +5. Processors persist the job id actually used, so status polling resolves.