From 4ca2b4824a1787f8be43c3d87dd650f12d272fa8 Mon Sep 17 00:00:00 2001 From: Michael Bunsen Date: Wed, 15 Apr 2026 21:13:06 -0700 Subject: [PATCH 1/9] feat(jobs): make missing-Redis-state FAILUREs loud and self-diagnosing MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Before: when process_nats_pipeline_result found the job's total-images key missing, it failed the job with the hardcoded reason "Job state keys not found in Redis (likely cleaned up concurrently)". The reason string went to job.logger only — not progress.errors — and collapsed three distinct causes (DB-index drift across hosts, key eviction, never-initialized state) into one misleading line. The Redis target (host:port/db) was never logged, so operators couldn't tell a cache-DB split from a cleanup race without shelling into workers. This commit makes that path name the actual cause: 1. AsyncJobStateManager.diagnose_missing_state() returns a one-line snapshot: masked host:port, DB index, and SCAN output for job:{id}:* with SCARDs. "keys_for_job=" ⇒ never initialized or wrong DB; SCARDs present ⇒ partial cleanup / eviction. 2. update_state() emits a WARN with that snapshot immediately before returning None, so the trigger shows up in the worker log even if the caller's FAILURE log is filtered out. 3. process_nats_pipeline_result passes the live snapshot to _fail_job instead of the hardcoded string. 4. _fail_job appends the reason to job.progress.errors before save, so the UI surfaces FAILUREs with a cause instead of errors=[]. 5. run_job logs "Running job X on redis=HOST:PORT/dbN" at start. Cross- host DB drift becomes visible in every job's log without needing to already suspect it. Tests added: - diagnose_missing_state with never-initialized state (keys_for_job=) - diagnose_missing_state after partial cleanup (SCARDs of surviving sets listed) - Existing "genuinely missing state" test assertion updated to match the richer reason string. Co-Authored-By: Claude Opus 4.6 (1M context) --- ami/jobs/tasks.py | 51 +++++++++++++++++++++-- ami/jobs/tests/test_tasks.py | 10 +++-- ami/ml/orchestration/async_job_state.py | 54 +++++++++++++++++++++++++ ami/ml/tests.py | 29 +++++++++++++ 4 files changed, 137 insertions(+), 7 deletions(-) diff --git a/ami/jobs/tasks.py b/ami/jobs/tasks.py index b00213ebb..5e7108ef0 100644 --- a/ami/jobs/tasks.py +++ b/ami/jobs/tasks.py @@ -146,7 +146,13 @@ def run_job(self, job_id: int) -> None: raise e # self.retry(exc=e, countdown=1, max_retries=1) else: - job.logger.info(f"Running job {job}") + # Log the Redis target at task start so cross-host DB-index drift surfaces + # in every job's log without needing to know where to look. The cluster + # convention is DB 0 = cache, DB 1 = Celery results (see + # config/settings/base.py); the Job state manager also uses the "default" + # connection, so every worker in the pool must agree on the DB number or + # initialize_job writes to one DB while update_state reads from another. + job.logger.info(f"Running job {job} on {_describe_redis_target()}") try: job.run() except Exception as e: @@ -281,10 +287,18 @@ def process_nats_pipeline_result(self, job_id: int, result_data: dict, reply_sub if not progress_info: # State keys genuinely missing (the total-images key returned None). # Ack so NATS stops redelivering and fail the job — there's no state - # left to reconcile against. + # left to reconcile against. The reason string is built from a live + # Redis snapshot (DB index, keys present under job:{id}:*) so the + # FAILURE log and the UI progress.errors entry name the actual cause + # instead of the previous hardcoded "likely cleaned up concurrently" + # guess — which conflated DB-index misconfig, eviction, and genuine + # concurrent cleanup into a single misleading string. _log_missing_state_context(job_id, "process") _ack_task_via_nats(reply_subject, logger) - _fail_job(job_id, "Job state keys not found in Redis (likely cleaned up concurrently)") + _fail_job( + job_id, + f"Job state missing from Redis (stage=process): {state_manager.diagnose_missing_state()}", + ) return try: @@ -367,7 +381,10 @@ def process_nats_pipeline_result(self, job_id: int, result_data: dict, reply_sub # then fail the job. Mirrors the stage=process missing-state path. _log_missing_state_context(job_id, "results") _ack_task_via_nats(reply_subject, job.logger) - _fail_job(job_id, "Job state keys not found in Redis (likely cleaned up concurrently)") + _fail_job( + job_id, + f"Job state missing from Redis (stage=results): {state_manager.diagnose_missing_state()}", + ) return # update complete state based on latest progress info after saving results @@ -416,6 +433,24 @@ def process_nats_pipeline_result(self, job_id: int, result_data: dict, reply_sub job.logger.error(error) +def _describe_redis_target() -> str: + """Return a ``host:port/dbN`` string for the "default" Redis connection. + + Logged at the start of every ``run_job`` so DB-index drift across hosts + (the class of misconfig that manifests as silent ``process_nats_pipeline_result`` + FAILUREs on whichever worker happens to read state from the wrong DB) is + visible in each job's log without requiring a separate diagnostic. + """ + try: + from django_redis import get_redis_connection + + redis = get_redis_connection("default") + kwargs = getattr(redis.connection_pool, "connection_kwargs", {}) or {} + return f"redis={kwargs.get('host', '?')}:{kwargs.get('port', '?')}/db{kwargs.get('db', '?')}" + except Exception as e: + return f"redis=(unavailable: {e})" + + def _fail_job(job_id: int, reason: str) -> None: from ami.jobs.models import Job, JobState from ami.ml.orchestration.jobs import cleanup_async_job_resources @@ -425,6 +460,14 @@ def _fail_job(job_id: int, reason: str) -> None: job = Job.objects.select_for_update().get(pk=job_id) if job.status in (JobState.CANCELING, *JobState.final_states()): return + # Mirror the reason into progress.errors so the UI surfaces it + # alongside the FAILURE state. Previously the reason lived only in + # job.logger, which meant the UI showed errors=[] and operators had + # to dig into Celery worker logs to find out why a job died. + try: + job.progress.errors.append(reason) + except Exception: + pass job.update_status(JobState.FAILURE, save=False) job.finished_at = datetime.datetime.now() job.save(update_fields=["status", "progress", "finished_at"]) diff --git a/ami/jobs/tests/test_tasks.py b/ami/jobs/tests/test_tasks.py index 760af1809..80944895b 100644 --- a/ami/jobs/tests/test_tasks.py +++ b/ami/jobs/tests/test_tasks.py @@ -348,10 +348,14 @@ def test_genuinely_missing_state_acks_and_fails_job(self, mock_manager_class, mo mock_ack.assert_called_once() mock_fail.assert_called_once() - # New, accurate message — no longer the misleading "Redis state missing" - # that users saw in the UI for transient connection drops. + # Reason string now leads with the stage and embeds a live Redis + # snapshot (DB index + key listing from diagnose_missing_state) so the + # failure cause — DB-index drift, eviction, or never-initialized — + # is visible in the FAILURE log instead of the previous single + # hardcoded "likely cleaned up concurrently" guess. args, _ = mock_fail.call_args - self.assertIn("Job state keys not found in Redis", args[1]) + self.assertIn("Job state missing from Redis", args[1]) + self.assertIn("stage=process", args[1]) @patch("ami.jobs.tasks._ack_task_via_nats") @patch("ami.jobs.tasks.TaskQueueManager") diff --git a/ami/ml/orchestration/async_job_state.py b/ami/ml/orchestration/async_job_state.py index 26e7c3024..f208aa8af 100644 --- a/ami/ml/orchestration/async_job_state.py +++ b/ami/ml/orchestration/async_job_state.py @@ -163,6 +163,18 @@ def update_state( newly_removed = results[0] if processed_image_ids else 0 if total_raw is None: + # Loud diagnostic before the silent None return. The caller will mark + # the job FAILURE based on this result, so the operator needs to see + # *why* the total key is gone. Distinguishes three different causes + # that previously all surfaced as the same hardcoded "likely cleaned + # up concurrently" reason string: Redis DB mismatch across hosts, + # key eviction, and genuinely-never-initialized state. + logger.warning( + "Job %s state missing in Redis (stage=%s): %s", + self.job_id, + stage, + self.diagnose_missing_state(), + ) return None total = int(total_raw) @@ -182,6 +194,48 @@ def update_state( newly_removed=newly_removed, ) + def diagnose_missing_state(self) -> str: + """ + One-line snapshot of what Redis actually holds for this job. + + Called from the missing-state path in ``update_state`` and from + ``_fail_job`` so the FAILURE log and the UI ``progress.errors`` entry + distinguish the three common causes — DB mismatch across hosts, key + eviction, and never-initialized state — instead of a single hardcoded + "likely cleaned up concurrently" guess that all three collapse to. + + Intentionally defensive: any failure to collect diagnostics is + swallowed, because the caller is already about to fail the job and + an exception from diagnostics would mask the original cause. + """ + try: + redis = self._get_redis() + kwargs = getattr(redis.connection_pool, "connection_kwargs", {}) or {} + db = kwargs.get("db", "?") + host = kwargs.get("host", "?") + port = kwargs.get("port", "?") + # Cursor-safe SCAN over the job's keyspace — cheap even on a busy + # Redis because it's filtered server-side and the per-job fanout is + # at most a handful of keys (pending:process, pending:results, + # failed, total). + keys = sorted(k.decode() if isinstance(k, bytes) else k for k in redis.scan_iter(match=self._pattern())) + sizes: list[str] = [] + for key in keys: + if key == self._total_key: + sizes.append(f"{key}=") + continue + try: + sizes.append(f"{key}=SCARD:{redis.scard(key)}") + except RedisError: + sizes.append(f"{key}=") + keys_summary = ", ".join(sizes) if sizes else "" + return f"redis={host}:{port}/db{db} keys_for_job={keys_summary}" + except Exception as e: + return f"(diagnostics failed: {e})" + + def _pattern(self) -> str: + return f"job:{self.job_id}:*" + def get_progress(self, stage: str) -> "JobStateProgress | None": """Read-only progress snapshot for the given stage.""" try: diff --git a/ami/ml/tests.py b/ami/ml/tests.py index 898a69d11..49edfd6e2 100644 --- a/ami/ml/tests.py +++ b/ami/ml/tests.py @@ -1742,6 +1742,35 @@ def test_update_state_returns_none_when_state_genuinely_missing(self): progress = self.manager.update_state({"img1", "img2"}, "process") self.assertIsNone(progress) + def test_diagnose_missing_state_when_never_initialized(self): + """ + Diagnostic string for the "never initialized" case: no keys are + present under ``job:{id}:*``. Output must still name the Redis host + and DB so cross-host DB drift is distinguishable from eviction and + truly-never-initialized state in one log line. + """ + # initialize_job has NOT been called; nothing under job:123:*. + diagnosis = self.manager.diagnose_missing_state() + self.assertIn("redis=", diagnosis) + self.assertIn("/db", diagnosis) + self.assertIn("keys_for_job=", diagnosis) + + def test_diagnose_missing_state_lists_present_keys(self): + """ + Diagnostic string for the partial-cleanup / eviction case: some keys + remain under ``job:{id}:*`` and their SCARDs should appear so the + operator can tell "total key evicted but pending sets still present" + from "nothing here, this DB never saw the job". + """ + self.manager.initialize_job(self.image_ids) + # Drop the total key to simulate eviction while pending sets survive. + redis = self.manager._get_redis() + redis.delete(self.manager._total_key) + + diagnosis = self.manager.diagnose_missing_state() + self.assertIn(f"job:{self.job_id}:pending_images:process=SCARD:", diagnosis) + self.assertNotIn(self.manager._total_key, diagnosis) + class TestSaveResultsRefreshesDeploymentCounts(TestCase): """save_results must refresh Deployment cached counts, not just Event counts. From b9ae01c84a79c715e6e39ce530a3647077995a98 Mon Sep 17 00:00:00 2001 From: Michael Bunsen Date: Fri, 17 Apr 2026 00:02:55 -0700 Subject: [PATCH 2/9] test(jobs): cover _fail_job reason append + clarify defensive comments MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add TestFailJob with two TDD-confirmed cases: - ``test_fail_job_appends_reason_to_progress_errors`` — verifies the reason string lands in ``job.progress.errors`` (both in-memory and after refresh_from_db) so UI surfaces the cause of the FAILURE. The existing ``_fail_job`` call-site tests in ``TestProcessNatsPipelineResultError`` mock ``_fail_job`` entirely, so a regression that stops appending to ``progress.errors`` would slip through undetected. - ``test_fail_job_is_noop_on_already_final_job`` — regression guard for the early-return branch when the job is already in a final state. Also: - Comment the bare ``except Exception: pass`` around the ``progress.errors.append`` in ``_fail_job`` to explain why we swallow diagnostic-write failures. - Extend the ``AsyncJobStateManager.diagnose_missing_state`` docstring with a one-paragraph note about the SCAN cost (failure-path only, per-job fanout of at most four keys) to head off the obvious review question. Co-Authored-By: Claude --- ami/jobs/tasks.py | 1 + ami/jobs/tests/test_tasks.py | 89 +++++++++++++++++++++++++ ami/ml/orchestration/async_job_state.py | 5 ++ 3 files changed, 95 insertions(+) diff --git a/ami/jobs/tasks.py b/ami/jobs/tasks.py index 5e7108ef0..92a1a1933 100644 --- a/ami/jobs/tasks.py +++ b/ami/jobs/tasks.py @@ -467,6 +467,7 @@ def _fail_job(job_id: int, reason: str) -> None: try: job.progress.errors.append(reason) except Exception: + # Don't let diagnostic-write failures mask the original FAILURE. pass job.update_status(JobState.FAILURE, save=False) job.finished_at = datetime.datetime.now() diff --git a/ami/jobs/tests/test_tasks.py b/ami/jobs/tests/test_tasks.py index 80944895b..8663ac2ec 100644 --- a/ami/jobs/tests/test_tasks.py +++ b/ami/jobs/tests/test_tasks.py @@ -631,6 +631,95 @@ def test_task_failure_marks_sync_api_job_failure_and_cleans_up(self, mock_cleanu mock_cleanup.assert_called_once() +class TestFailJob(TransactionTestCase): + """ + Regression tests for ``_fail_job`` — specifically for the reason-string + mirroring into ``progress.errors`` that this PR adds. + + The FAILURE log line alone is not enough for operators; the UI reads + ``progress.errors``, and prior to this PR that list stayed empty on the + missing-Redis-state path. Any regression that stops appending the reason + (e.g. silently dropping it via the defensive ``try/except``) would put + operators back in the position of digging through Celery worker logs to + find out why a job died. + """ + + def setUp(self): + cache.clear() + self.project = Project.objects.create(name="FailJob Test Project") + self.pipeline = Pipeline.objects.create(name="FailJob Pipeline", slug="fail-job-pipeline") + self.pipeline.projects.add(self.project) + self.collection = SourceImageCollection.objects.create(name="FailJob Collection", project=self.project) + + def tearDown(self): + cache.clear() + + def _make_job(self, dispatch_mode: JobDispatchMode = JobDispatchMode.ASYNC_API) -> Job: + job = Job.objects.create( + job_type_key=MLJob.key, + project=self.project, + name=f"{dispatch_mode} fail-job test", + pipeline=self.pipeline, + source_image_collection=self.collection, + dispatch_mode=dispatch_mode, + ) + job.update_status(JobState.STARTED, save=True) + return job + + @patch("ami.ml.orchestration.jobs.cleanup_async_job_resources") + def test_fail_job_appends_reason_to_progress_errors(self, mock_cleanup): + """ + Reason string must end up in ``job.progress.errors`` (persisted) so the + UI shows the cause of the FAILURE alongside the status change. Before + this PR the reason lived only in ``job.logger`` and the UI showed + ``errors=[]``. A silent regression here would not be caught by the + ``_fail_job`` call-site tests in ``TestProcessNatsPipelineResultError`` + (they mock ``_fail_job`` entirely). + """ + from ami.jobs.tasks import _fail_job + + job = self._make_job() + reason = "Job state missing from Redis (stage=process): redis=host:6379/db1 keys_for_job=" + + _fail_job(job.pk, reason) + + job.refresh_from_db() + self.assertEqual(job.status, JobState.FAILURE) + self.assertIn( + reason, + job.progress.errors, + f"expected reason in progress.errors, got: {job.progress.errors!r}", + ) + # Sanity: the fix also propagates to the DB-persisted copy (i.e. the + # update_fields tuple on job.save includes 'progress'). Re-read from a + # fresh Job instance to prove the append wasn't only visible on the + # in-memory object returned by select_for_update. + reloaded = Job.objects.get(pk=job.pk) + self.assertIn(reason, reloaded.progress.errors) + mock_cleanup.assert_called_once_with(job.pk) + + @patch("ami.ml.orchestration.jobs.cleanup_async_job_resources") + def test_fail_job_is_noop_on_already_final_job(self, mock_cleanup): + """ + If the job is already in a final state (e.g. concurrent cleanup + beat us), ``_fail_job`` must return early without touching status + or progress. This protects against double-failing a job that has + already been reconciled to SUCCESS by the reconciler path. + """ + from ami.jobs.tasks import _fail_job + + job = self._make_job() + job.update_status(JobState.SUCCESS, save=True) + errors_before = list(job.progress.errors) + + _fail_job(job.pk, "should be ignored") + + job.refresh_from_db() + self.assertEqual(job.status, JobState.SUCCESS) + self.assertEqual(job.progress.errors, errors_before) + mock_cleanup.assert_not_called() + + class TestResultEndpointWithError(APITestCase): """Integration test for the result API endpoint with error results.""" diff --git a/ami/ml/orchestration/async_job_state.py b/ami/ml/orchestration/async_job_state.py index f208aa8af..3a8be27a1 100644 --- a/ami/ml/orchestration/async_job_state.py +++ b/ami/ml/orchestration/async_job_state.py @@ -204,6 +204,11 @@ def diagnose_missing_state(self) -> str: eviction, and never-initialized state — instead of a single hardcoded "likely cleaned up concurrently" guess that all three collapse to. + Cost: the internal ``SCAN`` runs only on the failure path (once per + job-lifetime FAILURE), and the per-job key fanout is at most four + (pending:process, pending:results, failed, total), so the cost is + negligible compared to the FAILURE branch it only helps diagnose. + Intentionally defensive: any failure to collect diagnostics is swallowed, because the caller is already about to fail the job and an exception from diagnostics would mask the original cause. From 6314e3743bff7537f115c62a3e1937cd361327a3 Mon Sep 17 00:00:00 2001 From: Michael Bunsen Date: Mon, 22 Jun 2026 10:29:07 -0700 Subject: [PATCH 3/9] docs(jobs): correct diagnose/redis-target docstrings per review Fix three inaccurate comments flagged in review: - diagnose_missing_state() is called from update_state and the result handler (not _fail_job); note it can run at most twice per FAILURE. - SCAN is O(keyspace) regardless of MATCH; acceptable only because it runs on the rare missing-state failure path. - _describe_redis_target() returns a redis=... prefixed string. --- ami/jobs/tasks.py | 2 +- ami/ml/orchestration/async_job_state.py | 33 ++++++++++++++----------- 2 files changed, 20 insertions(+), 15 deletions(-) diff --git a/ami/jobs/tasks.py b/ami/jobs/tasks.py index 92a1a1933..cc61773d7 100644 --- a/ami/jobs/tasks.py +++ b/ami/jobs/tasks.py @@ -434,7 +434,7 @@ def process_nats_pipeline_result(self, job_id: int, result_data: dict, reply_sub def _describe_redis_target() -> str: - """Return a ``host:port/dbN`` string for the "default" Redis connection. + """Return a ``redis=host:port/dbN`` string for the "default" Redis connection. Logged at the start of every ``run_job`` so DB-index drift across hosts (the class of misconfig that manifests as silent ``process_nats_pipeline_result`` diff --git a/ami/ml/orchestration/async_job_state.py b/ami/ml/orchestration/async_job_state.py index 3a8be27a1..6dffb3179 100644 --- a/ami/ml/orchestration/async_job_state.py +++ b/ami/ml/orchestration/async_job_state.py @@ -198,16 +198,20 @@ def diagnose_missing_state(self) -> str: """ One-line snapshot of what Redis actually holds for this job. - Called from the missing-state path in ``update_state`` and from - ``_fail_job`` so the FAILURE log and the UI ``progress.errors`` entry - distinguish the three common causes — DB mismatch across hosts, key - eviction, and never-initialized state — instead of a single hardcoded - "likely cleaned up concurrently" guess that all three collapse to. - - Cost: the internal ``SCAN`` runs only on the failure path (once per - job-lifetime FAILURE), and the per-job key fanout is at most four - (pending:process, pending:results, failed, total), so the cost is - negligible compared to the FAILURE branch it only helps diagnose. + Called from the missing-state path in ``update_state`` (the loud log) + and from the result handler in ``process_nats_pipeline_result``, which + folds the result into the reason it passes to ``_fail_job`` so the + FAILURE log and the UI ``progress.errors`` entry distinguish the three + common causes — DB mismatch across hosts, key eviction, and + never-initialized state — instead of a single hardcoded "likely cleaned + up concurrently" guess that all three collapse to. + + Cost: only ever runs on the missing-state failure path (at most twice + per job-lifetime FAILURE — once for the log, once for the reason + string). ``SCAN`` is O(keyspace) regardless of ``MATCH`` (MATCH filters + the returned keys, not the keys scanned), but on the rare failure path + that one extra full cursor walk is negligible next to the FAILURE it + helps diagnose. Intentionally defensive: any failure to collect diagnostics is swallowed, because the caller is already about to fail the job and @@ -219,10 +223,11 @@ def diagnose_missing_state(self) -> str: db = kwargs.get("db", "?") host = kwargs.get("host", "?") port = kwargs.get("port", "?") - # Cursor-safe SCAN over the job's keyspace — cheap even on a busy - # Redis because it's filtered server-side and the per-job fanout is - # at most a handful of keys (pending:process, pending:results, - # failed, total). + # Cursor-safe SCAN over the job's keyspace. SCAN walks the whole DB + # (MATCH only filters what's returned, not what's scanned), so this + # is acceptable precisely because it runs only on the rare + # missing-state failure path; the per-job fanout returned is at most + # a handful of keys (pending:process, pending:results, failed, total). keys = sorted(k.decode() if isinstance(k, bytes) else k for k in redis.scan_iter(match=self._pattern())) sizes: list[str] = [] for key in keys: From 877a6863dfa7d5d7b431113f1a88e8b78edca85c Mon Sep 17 00:00:00 2001 From: Michael Bunsen Date: Mon, 22 Jun 2026 17:21:20 -0700 Subject: [PATCH 4/9] docs(jobs): rewrite time-indexed comments to present-tense and remove duplicate rationale [skip ci] MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - async_job_state.py update_state: "previously all surfaced as..." → states the contract ("Distinguishes three causes that map to the same symptom") - async_job_state.py diagnose_missing_state: remove inline SCAN/MATCH cost duplicate that repeated the docstring verbatim; keep the docstring - tasks.py _fail_job: "Previously the reason lived only in job.logger..." → states what the code does ("Operators can see the cause in the job detail view") Co-Authored-By: Claude Fable 5 --- ami/jobs/tasks.py | 5 ++--- ami/ml/orchestration/async_job_state.py | 12 +++--------- 2 files changed, 5 insertions(+), 12 deletions(-) diff --git a/ami/jobs/tasks.py b/ami/jobs/tasks.py index cc61773d7..ccb46741a 100644 --- a/ami/jobs/tasks.py +++ b/ami/jobs/tasks.py @@ -461,9 +461,8 @@ def _fail_job(job_id: int, reason: str) -> None: if job.status in (JobState.CANCELING, *JobState.final_states()): return # Mirror the reason into progress.errors so the UI surfaces it - # alongside the FAILURE state. Previously the reason lived only in - # job.logger, which meant the UI showed errors=[] and operators had - # to dig into Celery worker logs to find out why a job died. + # alongside the FAILURE status. Operators can see the cause in the + # job detail view without digging through Celery worker logs. try: job.progress.errors.append(reason) except Exception: diff --git a/ami/ml/orchestration/async_job_state.py b/ami/ml/orchestration/async_job_state.py index 6dffb3179..9c93f5d9d 100644 --- a/ami/ml/orchestration/async_job_state.py +++ b/ami/ml/orchestration/async_job_state.py @@ -165,10 +165,9 @@ def update_state( if total_raw is None: # Loud diagnostic before the silent None return. The caller will mark # the job FAILURE based on this result, so the operator needs to see - # *why* the total key is gone. Distinguishes three different causes - # that previously all surfaced as the same hardcoded "likely cleaned - # up concurrently" reason string: Redis DB mismatch across hosts, - # key eviction, and genuinely-never-initialized state. + # *why* the total key is gone. Distinguishes three causes that map to + # the same symptom: DB-index mismatch across hosts, key eviction, and + # never-initialized state. logger.warning( "Job %s state missing in Redis (stage=%s): %s", self.job_id, @@ -223,11 +222,6 @@ def diagnose_missing_state(self) -> str: db = kwargs.get("db", "?") host = kwargs.get("host", "?") port = kwargs.get("port", "?") - # Cursor-safe SCAN over the job's keyspace. SCAN walks the whole DB - # (MATCH only filters what's returned, not what's scanned), so this - # is acceptable precisely because it runs only on the rare - # missing-state failure path; the per-job fanout returned is at most - # a handful of keys (pending:process, pending:results, failed, total). keys = sorted(k.decode() if isinstance(k, bytes) else k for k in redis.scan_iter(match=self._pattern())) sizes: list[str] = [] for key in keys: From 8edfb75e39198dd87e1d30e18b06d646e42c813c Mon Sep 17 00:00:00 2001 From: Michael Bunsen Date: Mon, 22 Jun 2026 17:21:29 -0700 Subject: [PATCH 5/9] test(jobs): cover stage=results missing-state path in process_nats_pipeline_result Adds test_genuinely_missing_state_results_stage_acks_and_fails_job to TestProcessNatsPipelineResultError, mirroring the existing process-branch test for the results-stage missing-state path (tasks.py lines 378-388). Verifies that when update_state returns None at stage=results, the task acks NATS (to stop redelivery) and calls _fail_job with a reason string containing "stage=results". Co-Authored-By: Claude Fable 5 --- ami/jobs/tests/test_tasks.py | 65 +++++++++++++++++++++++++++++++++--- 1 file changed, 60 insertions(+), 5 deletions(-) diff --git a/ami/jobs/tests/test_tasks.py b/ami/jobs/tests/test_tasks.py index 8663ac2ec..49f753f22 100644 --- a/ami/jobs/tests/test_tasks.py +++ b/ami/jobs/tests/test_tasks.py @@ -348,15 +348,70 @@ def test_genuinely_missing_state_acks_and_fails_job(self, mock_manager_class, mo mock_ack.assert_called_once() mock_fail.assert_called_once() - # Reason string now leads with the stage and embeds a live Redis - # snapshot (DB index + key listing from diagnose_missing_state) so the - # failure cause — DB-index drift, eviction, or never-initialized — - # is visible in the FAILURE log instead of the previous single - # hardcoded "likely cleaned up concurrently" guess. + # The reason string passed to _fail_job identifies the stage and embeds + # a live Redis snapshot (from diagnose_missing_state) so the FAILURE + # log and UI progress.errors distinguish DB-index drift, eviction, and + # never-initialized state rather than collapsing them into one message. args, _ = mock_fail.call_args self.assertIn("Job state missing from Redis", args[1]) self.assertIn("stage=process", args[1]) + @patch("ami.jobs.tasks._fail_job") + @patch("ami.jobs.tasks._ack_task_via_nats") + @patch("ami.jobs.tasks.TaskQueueManager") + def test_genuinely_missing_state_results_stage_acks_and_fails_job(self, mock_manager_class, mock_ack, mock_fail): + """ + Mirror of test_genuinely_missing_state_acks_and_fails_job for the + stage=results path (tasks.py lines 378-388). When the total-images key + is gone at the results-stage update_state call, the task must ack NATS + to stop redelivery and fail the job — there is no state to reconcile. + The reason string must identify stage=results. + """ + self._setup_mock_nats(mock_manager_class) + + # save_results requires at least one algorithm on the pipeline. + detection_algorithm = Algorithm.objects.create( + name="results-missing-state-detector", + key="results-missing-state-detector", + task_type=AlgorithmTaskType.LOCALIZATION, + ) + self.pipeline.algorithms.add(detection_algorithm) + + # Use a success result so the process-stage path succeeds and + # save_results runs before the results-stage update_state is reached. + success_data = PipelineResultsResponse( + pipeline="test-pipeline", + algorithms={}, + total_time=1.0, + source_images=[SourceImageResponse(id=str(self.images[0].pk), url="http://example.com/test_image_0.jpg")], + detections=[], + errors=None, + ).dict() + + real_update_state = AsyncJobStateManager.update_state + + def none_on_results_stage(self_inner, processed_image_ids, stage, failed_image_ids=None): + if stage == "results": + return None + return real_update_state(self_inner, processed_image_ids, stage, failed_image_ids) + + with patch.object(AsyncJobStateManager, "update_state", none_on_results_stage): + process_nats_pipeline_result( + job_id=self.job.pk, + result_data=success_data, + reply_subject="reply.missing-results", + ) + + mock_ack.assert_called_once() + mock_fail.assert_called_once() + # The reason string passed to _fail_job identifies the stage and embeds + # a live Redis snapshot (from diagnose_missing_state) so the FAILURE + # log and UI progress.errors distinguish DB-index drift, eviction, and + # never-initialized state rather than collapsing them into one message. + args, _ = mock_fail.call_args + self.assertIn("Job state missing from Redis", args[1]) + self.assertIn("stage=results", args[1]) + @patch("ami.jobs.tasks._ack_task_via_nats") @patch("ami.jobs.tasks.TaskQueueManager") def test_ack_deferred_until_after_results_stage_srem(self, mock_manager_class, mock_ack): From 7705816134c3daafec13ca5518870b83e66fbb9f Mon Sep 17 00:00:00 2001 From: Michael Bunsen Date: Mon, 22 Jun 2026 18:12:05 -0700 Subject: [PATCH 6/9] fix(jobs): keep the Redis host out of the public job-failure reason diagnose_missing_state() embedded the Redis host:port in the string that becomes the job's public progress.errors / failure reason, leaking the internal Redis hostname into a user-visible surface. The DB index is the load-bearing diagnostic (the missing-state incidents were a db0-vs-db1 mismatch across processes), so the public string now reports only "redis db{N}: keys_for_job=...". The host:port moves to a new connection_target() used solely in the server-side warning logged by update_state, where operators see it without it reaching the public job log. Co-Authored-By: Claude Opus 4.8 (1M context) --- ami/ml/orchestration/async_job_state.py | 68 ++++++++++++++++--------- ami/ml/tests.py | 30 +++++++++-- 2 files changed, 69 insertions(+), 29 deletions(-) diff --git a/ami/ml/orchestration/async_job_state.py b/ami/ml/orchestration/async_job_state.py index 9c93f5d9d..3d3d0ebdc 100644 --- a/ami/ml/orchestration/async_job_state.py +++ b/ami/ml/orchestration/async_job_state.py @@ -169,9 +169,10 @@ def update_state( # the same symptom: DB-index mismatch across hosts, key eviction, and # never-initialized state. logger.warning( - "Job %s state missing in Redis (stage=%s): %s", + "Job %s state missing in Redis (stage=%s, target=%s): %s", self.job_id, stage, + self.connection_target(), self.diagnose_missing_state(), ) return None @@ -195,33 +196,36 @@ def update_state( def diagnose_missing_state(self) -> str: """ - One-line snapshot of what Redis actually holds for this job. - - Called from the missing-state path in ``update_state`` (the loud log) - and from the result handler in ``process_nats_pipeline_result``, which - folds the result into the reason it passes to ``_fail_job`` so the - FAILURE log and the UI ``progress.errors`` entry distinguish the three - common causes — DB mismatch across hosts, key eviction, and - never-initialized state — instead of a single hardcoded "likely cleaned - up concurrently" guess that all three collapse to. - - Cost: only ever runs on the missing-state failure path (at most twice - per job-lifetime FAILURE — once for the log, once for the reason - string). ``SCAN`` is O(keyspace) regardless of ``MATCH`` (MATCH filters - the returned keys, not the keys scanned), but on the rare failure path - that one extra full cursor walk is negligible next to the FAILURE it - helps diagnose. - - Intentionally defensive: any failure to collect diagnostics is - swallowed, because the caller is already about to fail the job and - an exception from diagnostics would mask the original cause. + One-line, log-safe snapshot of what Redis holds for this job. + + Reports the Redis DB index and the per-job keys (with set cardinalities) so a + missing-state FAILURE distinguishes its three common causes — DB-index mismatch + across processes, key eviction, and never-initialized state — instead of a single + hardcoded "likely cleaned up concurrently" guess that all three collapse to. + + The Redis host/port is deliberately omitted here: this string is surfaced in the + job's public ``progress.errors`` (via the reason passed to ``_fail_job``), and the + host identifies internal infrastructure. The DB index is the load-bearing signal for + the mismatch case and is safe to expose. Operators who need the host see it in the + server-side warning ``update_state`` logs via ``connection_target()``. + + Called from the missing-state path in ``update_state`` (the loud log) and from the + result handler in ``process_nats_pipeline_result``. + + Cost: only ever runs on the missing-state failure path (at most twice per + job-lifetime FAILURE — once for the log, once for the reason string). ``SCAN`` is + O(keyspace) regardless of ``MATCH`` (MATCH filters the returned keys, not the keys + scanned), but on the rare failure path that one extra full cursor walk is negligible + next to the FAILURE it helps diagnose. + + Intentionally defensive: any failure to collect diagnostics is swallowed, because the + caller is already about to fail the job and an exception from diagnostics would mask + the original cause. """ try: redis = self._get_redis() kwargs = getattr(redis.connection_pool, "connection_kwargs", {}) or {} db = kwargs.get("db", "?") - host = kwargs.get("host", "?") - port = kwargs.get("port", "?") keys = sorted(k.decode() if isinstance(k, bytes) else k for k in redis.scan_iter(match=self._pattern())) sizes: list[str] = [] for key in keys: @@ -233,10 +237,26 @@ def diagnose_missing_state(self) -> str: except RedisError: sizes.append(f"{key}=") keys_summary = ", ".join(sizes) if sizes else "" - return f"redis={host}:{port}/db{db} keys_for_job={keys_summary}" + return f"redis db{db}: keys_for_job={keys_summary}" except Exception as e: return f"(diagnostics failed: {e})" + def connection_target(self) -> str: + """ + Redis target as ``host:port/dbN``, for server-side operator logs only. + + Names the internal Redis host, so this must NOT go into job logs or + ``progress.errors`` — use :meth:`diagnose_missing_state` for anything surfaced to + the user. Kept separate so the host stays in operator-facing logs (where the + cross-host DB-drift diagnosis needs it) without leaking to the public job view. + """ + try: + redis = self._get_redis() + kwargs = getattr(redis.connection_pool, "connection_kwargs", {}) or {} + return f"{kwargs.get('host', '?')}:{kwargs.get('port', '?')}/db{kwargs.get('db', '?')}" + except Exception as e: + return f"(unavailable: {e})" + def _pattern(self) -> str: return f"job:{self.job_id}:*" diff --git a/ami/ml/tests.py b/ami/ml/tests.py index 49edfd6e2..575fcbd45 100644 --- a/ami/ml/tests.py +++ b/ami/ml/tests.py @@ -1745,16 +1745,36 @@ def test_update_state_returns_none_when_state_genuinely_missing(self): def test_diagnose_missing_state_when_never_initialized(self): """ Diagnostic string for the "never initialized" case: no keys are - present under ``job:{id}:*``. Output must still name the Redis host - and DB so cross-host DB drift is distinguishable from eviction and - truly-never-initialized state in one log line. + present under ``job:{id}:*``. The public string names the Redis DB + index (so cross-process DB drift is distinguishable from eviction and + truly-never-initialized state) but must NOT name the Redis host: it is + surfaced in the job's public progress.errors. """ # initialize_job has NOT been called; nothing under job:123:*. diagnosis = self.manager.diagnose_missing_state() - self.assertIn("redis=", diagnosis) - self.assertIn("/db", diagnosis) + self.assertIn("db", diagnosis) self.assertIn("keys_for_job=", diagnosis) + def test_diagnose_missing_state_omits_host_but_connection_target_keeps_it(self): + """ + The public diagnostic must not leak the internal Redis host/port, but the + operator-only connection_target() must still carry host:port/db for the + server-side cross-host DB-drift diagnosis. + """ + kwargs = self.manager._get_redis().connection_pool.connection_kwargs + host = str(kwargs.get("host", "")) + + public = self.manager.diagnose_missing_state() + target = self.manager.connection_target() + + # The leak is the host:port connection detail, so assert the host:port pattern is + # absent from the public string (not the bare host substring — the public string + # legitimately contains the word "redis"). + if host: + self.assertNotIn(f"{host}:", public) + self.assertIn(f"{host}:", target) + self.assertIn("/db", target) + def test_diagnose_missing_state_lists_present_keys(self): """ Diagnostic string for the partial-cleanup / eviction case: some keys From eb8a2016307ce04a29e511593219b2e1deb36bb3 Mon Sep 17 00:00:00 2001 From: Michael Bunsen Date: Mon, 22 Jun 2026 18:23:16 -0700 Subject: [PATCH 7/9] fix(jobs): keep the Redis host out of the per-job public log too MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The earlier fix sanitized diagnose_missing_state()/progress.errors but missed the other public surface: run_job's start log line `Running job ... on redis=host:port/dbN` goes through job.logger into the JobLog table, which the UI shows — so the internal Redis host leaked into every job's public log on the success path, not just failures. Split the helpers: _redis_db_index() returns only the DB index for the public job log (the load-bearing signal for cross-host DB-drift), and _describe_redis_target() (host:port) now logs to the server logger only. Also stop _fail_job silently swallowing a failed progress.errors append — log a warning so the swallow is observable — and correct the run_job comment that overstated the DB-0-vs-DB-1 convention. Adds TestRedisTargetLogging pinning that the public string omits host:port while the server string keeps it. Co-Authored-By: Claude Opus 4.8 (1M context) --- ami/jobs/tasks.py | 48 +++++++++++++++++++++++++----------- ami/jobs/tests/test_tasks.py | 27 +++++++++++++++++++- 2 files changed, 60 insertions(+), 15 deletions(-) diff --git a/ami/jobs/tasks.py b/ami/jobs/tasks.py index ccb46741a..018f144ae 100644 --- a/ami/jobs/tasks.py +++ b/ami/jobs/tasks.py @@ -146,13 +146,15 @@ def run_job(self, job_id: int) -> None: raise e # self.retry(exc=e, countdown=1, max_retries=1) else: - # Log the Redis target at task start so cross-host DB-index drift surfaces - # in every job's log without needing to know where to look. The cluster - # convention is DB 0 = cache, DB 1 = Celery results (see - # config/settings/base.py); the Job state manager also uses the "default" - # connection, so every worker in the pool must agree on the DB number or - # initialize_job writes to one DB while update_state reads from another. - job.logger.info(f"Running job {job} on {_describe_redis_target()}") + # Log the Redis DB index at task start so cross-host DB-index drift — the misconfig + # that surfaces as silent process_nats_pipeline_result FAILUREs on whichever worker + # reads state from the wrong DB — is visible per job. Every host's "default" Redis + # connection (the one the job state manager uses) must select the same DB number, or + # initialize_job and update_state operate on different DBs. The public job log carries + # only the DB index; the full host:port goes to the server log, since the host names + # internal infrastructure. + job.logger.info(f"Running job {job} on Redis db {_redis_db_index()}") + logger.info("Job %s Redis target: %s", job.pk, _describe_redis_target()) try: job.run() except Exception as e: @@ -433,13 +435,29 @@ def process_nats_pipeline_result(self, job_id: int, result_data: dict, reply_sub job.logger.error(error) +def _redis_db_index() -> str: + """Return just the DB index of the "default" Redis connection (e.g. ``"1"``). + + Safe for the public job log: the DB index is the load-bearing signal for diagnosing + cross-host DB drift, without exposing the internal Redis host. Pair with + :func:`_describe_redis_target` (host:port, server logs only) when an operator needs the host. + """ + try: + from django_redis import get_redis_connection + + redis = get_redis_connection("default") + kwargs = getattr(redis.connection_pool, "connection_kwargs", {}) or {} + return str(kwargs.get("db", "?")) + except Exception as e: + return f"(unavailable: {e})" + + def _describe_redis_target() -> str: """Return a ``redis=host:port/dbN`` string for the "default" Redis connection. - Logged at the start of every ``run_job`` so DB-index drift across hosts - (the class of misconfig that manifests as silent ``process_nats_pipeline_result`` - FAILUREs on whichever worker happens to read state from the wrong DB) is - visible in each job's log without requiring a separate diagnostic. + For server-side logs only — it names the internal Redis host, so it must not reach the + public job log (use :func:`_redis_db_index` there). Logged server-side at ``run_job`` start + so an operator can resolve a DB-index drift to a specific host. """ try: from django_redis import get_redis_connection @@ -465,9 +483,11 @@ def _fail_job(job_id: int, reason: str) -> None: # job detail view without digging through Celery worker logs. try: job.progress.errors.append(reason) - except Exception: - # Don't let diagnostic-write failures mask the original FAILURE. - pass + except Exception as e: + # Don't let a diagnostic-write failure mask the original FAILURE, but record + # that the reason could not be attached so the swallow is observable — otherwise + # the UI silently loses the cause this PR exists to surface. + logger.warning("Job %s: could not append failure reason to progress.errors: %s", job_id, e) job.update_status(JobState.FAILURE, save=False) job.finished_at = datetime.datetime.now() job.save(update_fields=["status", "progress", "finished_at"]) diff --git a/ami/jobs/tests/test_tasks.py b/ami/jobs/tests/test_tasks.py index 49f753f22..90fb4103c 100644 --- a/ami/jobs/tests/test_tasks.py +++ b/ami/jobs/tests/test_tasks.py @@ -12,7 +12,7 @@ from unittest.mock import AsyncMock, MagicMock, patch from django.core.cache import cache -from django.test import TransactionTestCase +from django.test import SimpleTestCase, TransactionTestCase from rest_framework.test import APITestCase from ami.base.serializers import reverse_with_params @@ -1910,3 +1910,28 @@ def worker(): JobState.REVOKED.value, f"late completion resurrected a REVOKED job to {self.job.status!r}", ) + + +class TestRedisTargetLogging(SimpleTestCase): + """The per-job Redis-target log must not leak the internal Redis host. + + run_job logs the Redis DB index to the public job log (via _redis_db_index) and the full + host:port only to the server log (via _describe_redis_target). The host names internal + infrastructure, so it must never reach the public job log / progress.errors. This pins the + split that a prior leak (host:port in the public job log) slipped through without. + """ + + def test_public_db_index_omits_host_and_port(self): + from ami.jobs.tasks import _redis_db_index + + out = _redis_db_index() + self.assertNotIn(":", out, "DB index must not contain a host:port") + self.assertNotIn("redis=", out) + + def test_server_target_includes_host_and_port(self): + from ami.jobs.tasks import _describe_redis_target + + out = _describe_redis_target() + self.assertTrue(out.startswith("redis=")) + self.assertIn(":", out) + self.assertIn("/db", out) From a8e9c7fa3235be5500a1d93ffc01f6364f0cdac9 Mon Sep 17 00:00:00 2001 From: Michael Bunsen Date: Wed, 24 Jun 2026 14:46:42 -0700 Subject: [PATCH 8/9] docs(jobs): trim missing-state diagnostics comments to match precedent Cut the over-long diagnose_missing_state docstring (dropped the SCAN-cost and standalone defensive paragraphs) and the run_job DB-index comment, and removed changelog-style references to the prior hardcoded reason string. Behavior unchanged. Co-Authored-By: Claude Opus 4.8 (1M context) --- ami/jobs/tasks.py | 21 ++++++-------- ami/ml/orchestration/async_job_state.py | 38 +++++++++---------------- 2 files changed, 21 insertions(+), 38 deletions(-) diff --git a/ami/jobs/tasks.py b/ami/jobs/tasks.py index 018f144ae..f88330474 100644 --- a/ami/jobs/tasks.py +++ b/ami/jobs/tasks.py @@ -146,13 +146,10 @@ def run_job(self, job_id: int) -> None: raise e # self.retry(exc=e, countdown=1, max_retries=1) else: - # Log the Redis DB index at task start so cross-host DB-index drift — the misconfig - # that surfaces as silent process_nats_pipeline_result FAILUREs on whichever worker - # reads state from the wrong DB — is visible per job. Every host's "default" Redis - # connection (the one the job state manager uses) must select the same DB number, or - # initialize_job and update_state operate on different DBs. The public job log carries - # only the DB index; the full host:port goes to the server log, since the host names - # internal infrastructure. + # Log the Redis DB index at task start so cross-host DB-index drift is visible per job + # (workers reading state from a different DB than run_job wrote it to fail silently in + # process_nats_pipeline_result). Public log carries only the DB index; host:port goes + # to the server log, since the host names internal infrastructure. job.logger.info(f"Running job {job} on Redis db {_redis_db_index()}") logger.info("Job %s Redis target: %s", job.pk, _describe_redis_target()) try: @@ -289,12 +286,10 @@ def process_nats_pipeline_result(self, job_id: int, result_data: dict, reply_sub if not progress_info: # State keys genuinely missing (the total-images key returned None). # Ack so NATS stops redelivering and fail the job — there's no state - # left to reconcile against. The reason string is built from a live - # Redis snapshot (DB index, keys present under job:{id}:*) so the - # FAILURE log and the UI progress.errors entry name the actual cause - # instead of the previous hardcoded "likely cleaned up concurrently" - # guess — which conflated DB-index misconfig, eviction, and genuine - # concurrent cleanup into a single misleading string. + # left to reconcile against. The reason is built from a live Redis + # snapshot (DB index, keys present under job:{id}:*) so the FAILURE log + # and the UI progress.errors entry name the actual cause (DB-index + # misconfig vs eviction vs concurrent cleanup). _log_missing_state_context(job_id, "process") _ack_task_via_nats(reply_subject, logger) _fail_job( diff --git a/ami/ml/orchestration/async_job_state.py b/ami/ml/orchestration/async_job_state.py index 3d3d0ebdc..b83b13b12 100644 --- a/ami/ml/orchestration/async_job_state.py +++ b/ami/ml/orchestration/async_job_state.py @@ -196,31 +196,19 @@ def update_state( def diagnose_missing_state(self) -> str: """ - One-line, log-safe snapshot of what Redis holds for this job. - - Reports the Redis DB index and the per-job keys (with set cardinalities) so a - missing-state FAILURE distinguishes its three common causes — DB-index mismatch - across processes, key eviction, and never-initialized state — instead of a single - hardcoded "likely cleaned up concurrently" guess that all three collapse to. - - The Redis host/port is deliberately omitted here: this string is surfaced in the - job's public ``progress.errors`` (via the reason passed to ``_fail_job``), and the - host identifies internal infrastructure. The DB index is the load-bearing signal for - the mismatch case and is safe to expose. Operators who need the host see it in the - server-side warning ``update_state`` logs via ``connection_target()``. - - Called from the missing-state path in ``update_state`` (the loud log) and from the - result handler in ``process_nats_pipeline_result``. - - Cost: only ever runs on the missing-state failure path (at most twice per - job-lifetime FAILURE — once for the log, once for the reason string). ``SCAN`` is - O(keyspace) regardless of ``MATCH`` (MATCH filters the returned keys, not the keys - scanned), but on the rare failure path that one extra full cursor walk is negligible - next to the FAILURE it helps diagnose. - - Intentionally defensive: any failure to collect diagnostics is swallowed, because the - caller is already about to fail the job and an exception from diagnostics would mask - the original cause. + One-line, log-safe snapshot of what Redis holds for this job: the DB index and + the per-job keys with their set cardinalities. Lets a missing-state FAILURE name + its cause — DB-index mismatch across processes, key eviction, or never-initialized + state — rather than collapsing all three into one guess. + + The Redis host/port is omitted: this string is surfaced in the job's public + ``progress.errors`` (via the reason passed to ``_fail_job``), and the host names + internal infrastructure. The DB index is the load-bearing signal for the mismatch + case and is safe to expose; operators who need the host read it from the server-side + ``update_state`` warning via ``connection_target()``. + + Any failure to collect diagnostics is swallowed: the caller is already failing the + job, and an exception here would mask the original cause. """ try: redis = self._get_redis() From 4b007c85ba1c038bc2d04bc2fbb7806e19760f5a Mon Sep 17 00:00:00 2001 From: Michael Bunsen Date: Wed, 24 Jun 2026 16:42:05 -0700 Subject: [PATCH 9/9] test(jobs): drop diagnostics content/append tests, keep host-omission guards [skip ci] The missing-state diagnostics are a low-frequency logging path; the two diagnose_missing_state content assertions and the _fail_job progress.errors append test were disproportionate coverage for it. Kept the security-relevant guards (TestRedisTargetLogging and test_diagnose_missing_state_omits_host_*) that pin the Redis host out of user-facing strings, plus the missing-state ack+fail behavior tests. No production code changed. Co-Authored-By: Claude Opus 4.8 (1M context) --- ami/jobs/tests/test_tasks.py | 32 -------------------------------- ami/ml/tests.py | 29 ----------------------------- 2 files changed, 61 deletions(-) diff --git a/ami/jobs/tests/test_tasks.py b/ami/jobs/tests/test_tasks.py index 90fb4103c..013095aad 100644 --- a/ami/jobs/tests/test_tasks.py +++ b/ami/jobs/tests/test_tasks.py @@ -721,38 +721,6 @@ def _make_job(self, dispatch_mode: JobDispatchMode = JobDispatchMode.ASYNC_API) job.update_status(JobState.STARTED, save=True) return job - @patch("ami.ml.orchestration.jobs.cleanup_async_job_resources") - def test_fail_job_appends_reason_to_progress_errors(self, mock_cleanup): - """ - Reason string must end up in ``job.progress.errors`` (persisted) so the - UI shows the cause of the FAILURE alongside the status change. Before - this PR the reason lived only in ``job.logger`` and the UI showed - ``errors=[]``. A silent regression here would not be caught by the - ``_fail_job`` call-site tests in ``TestProcessNatsPipelineResultError`` - (they mock ``_fail_job`` entirely). - """ - from ami.jobs.tasks import _fail_job - - job = self._make_job() - reason = "Job state missing from Redis (stage=process): redis=host:6379/db1 keys_for_job=" - - _fail_job(job.pk, reason) - - job.refresh_from_db() - self.assertEqual(job.status, JobState.FAILURE) - self.assertIn( - reason, - job.progress.errors, - f"expected reason in progress.errors, got: {job.progress.errors!r}", - ) - # Sanity: the fix also propagates to the DB-persisted copy (i.e. the - # update_fields tuple on job.save includes 'progress'). Re-read from a - # fresh Job instance to prove the append wasn't only visible on the - # in-memory object returned by select_for_update. - reloaded = Job.objects.get(pk=job.pk) - self.assertIn(reason, reloaded.progress.errors) - mock_cleanup.assert_called_once_with(job.pk) - @patch("ami.ml.orchestration.jobs.cleanup_async_job_resources") def test_fail_job_is_noop_on_already_final_job(self, mock_cleanup): """ diff --git a/ami/ml/tests.py b/ami/ml/tests.py index 575fcbd45..de4f34ee0 100644 --- a/ami/ml/tests.py +++ b/ami/ml/tests.py @@ -1742,19 +1742,6 @@ def test_update_state_returns_none_when_state_genuinely_missing(self): progress = self.manager.update_state({"img1", "img2"}, "process") self.assertIsNone(progress) - def test_diagnose_missing_state_when_never_initialized(self): - """ - Diagnostic string for the "never initialized" case: no keys are - present under ``job:{id}:*``. The public string names the Redis DB - index (so cross-process DB drift is distinguishable from eviction and - truly-never-initialized state) but must NOT name the Redis host: it is - surfaced in the job's public progress.errors. - """ - # initialize_job has NOT been called; nothing under job:123:*. - diagnosis = self.manager.diagnose_missing_state() - self.assertIn("db", diagnosis) - self.assertIn("keys_for_job=", diagnosis) - def test_diagnose_missing_state_omits_host_but_connection_target_keeps_it(self): """ The public diagnostic must not leak the internal Redis host/port, but the @@ -1775,22 +1762,6 @@ def test_diagnose_missing_state_omits_host_but_connection_target_keeps_it(self): self.assertIn(f"{host}:", target) self.assertIn("/db", target) - def test_diagnose_missing_state_lists_present_keys(self): - """ - Diagnostic string for the partial-cleanup / eviction case: some keys - remain under ``job:{id}:*`` and their SCARDs should appear so the - operator can tell "total key evicted but pending sets still present" - from "nothing here, this DB never saw the job". - """ - self.manager.initialize_job(self.image_ids) - # Drop the total key to simulate eviction while pending sets survive. - redis = self.manager._get_redis() - redis.delete(self.manager._total_key) - - diagnosis = self.manager.diagnose_missing_state() - self.assertIn(f"job:{self.job_id}:pending_images:process=SCARD:", diagnosis) - self.assertNotIn(self.manager._total_key, diagnosis) - class TestSaveResultsRefreshesDeploymentCounts(TestCase): """save_results must refresh Deployment cached counts, not just Event counts.