diff --git a/hindsight-api/hindsight_api/engine/memory_engine.py b/hindsight-api/hindsight_api/engine/memory_engine.py index aaed0587..484d4809 100644 --- a/hindsight-api/hindsight_api/engine/memory_engine.py +++ b/hindsight-api/hindsight_api/engine/memory_engine.py @@ -925,8 +925,6 @@ class MemoryEngine(MemoryEngineInterface): """ task_type = task_dict.get("type") operation_id = task_dict.get("operation_id") - retry_count = task_dict.get("retry_count", 0) - max_retries = 3 # Set schema context for multi-tenant task execution schema = task_dict.pop("_schema", None) @@ -972,30 +970,22 @@ class MemoryEngine(MemoryEngineInterface): await self._mark_operation_completed(operation_id) except Exception as e: - # Task failed - check if we should retry - logger.error( - f"Task execution failed (attempt {retry_count + 1}/{max_retries + 1}): {task_type}, error: {e}" - ) + logger.error(f"Task execution failed: {task_type}, error: {e}") import traceback error_traceback = traceback.format_exc() traceback.print_exc() - # Don't retry file conversion - if conversion fails, it won't succeed on retry - # (missing OCR, corrupted file, unsupported format, etc.) - should_retry = retry_count < max_retries and task_type != "file_convert_retain" - - if should_retry: - # Reschedule with incremented retry count - task_dict["retry_count"] = retry_count + 1 - logger.info(f"Rescheduling task {task_type} (retry {retry_count + 1}/{max_retries})") - await self._task_backend.submit_task(task_dict) - else: - # Max retries exceeded or non-retryable task - mark operation as failed - reason = "non-retryable task type" if task_type == "file_convert_retain" else "max retries exceeded" - logger.error(f"Not retrying task {task_type} ({reason}), marking as failed") + if task_type == "file_convert_retain": + # Non-retryable: mark as failed immediately. + # Conversion failures won't improve on retry (missing OCR, corrupted file, etc.) + logger.error(f"Not retrying task {task_type} (non-retryable), marking as failed") if operation_id: await self._mark_operation_failed(operation_id, str(e), error_traceback) + else: + # Retryable: re-raise so the worker poller handles retry/fail via _retry_or_fail, + # which correctly resets status='pending' and increments the DB retry_count. + raise async def _delete_operation_record(self, operation_id: str): """Helper to delete an operation record from the database.""" diff --git a/hindsight-api/hindsight_api/worker/poller.py b/hindsight-api/hindsight_api/worker/poller.py index c7420278..b26405c8 100644 --- a/hindsight-api/hindsight_api/worker/poller.py +++ b/hindsight-api/hindsight_api/worker/poller.py @@ -376,11 +376,13 @@ class WorkerPoller: del self._in_flight_by_type[operation_type] async def _execute_task_inner(self, task: ClaimedTask): - """Inner task execution with error handling. + """Inner task execution with retry/fail handling. - Note: The executor (MemoryEngine.execute_task) handles status marking internally - (marking operations as completed/failed and handling retries). This method should - NOT override those status updates. + Retryable task failures are re-raised by the executor (MemoryEngine.execute_task) + and handled here via _retry_or_fail, which resets status='pending' (or marks as + 'failed' after max retries). Non-retryable failures (e.g., file_convert_retain) are + handled by the executor internally — it marks the operation as failed and returns + normally, so no exception reaches here. """ task_type = task.task_dict.get("type", "unknown") bank_id = task.task_dict.get("bank_id", "unknown") @@ -393,10 +395,9 @@ class WorkerPoller: await self._executor(task.task_dict) logger.debug(f"Task {task.operation_id} execution finished") except Exception as e: - # The executor should handle its own errors, but if an unexpected exception - # propagates (e.g., from schema setup), log it as a warning - logger.error(f"Task {task.operation_id} raised unexpected exception: {e}") + logger.error(f"Task {task.operation_id} failed: {e}") traceback.print_exc() + await self._retry_or_fail(task.operation_id, str(e), task.schema) async def recover_own_tasks(self) -> int: """ diff --git a/hindsight-api/tests/test_worker.py b/hindsight-api/tests/test_worker.py index 2f8d5d95..b4341a5b 100644 --- a/hindsight-api/tests/test_worker.py +++ b/hindsight-api/tests/test_worker.py @@ -268,24 +268,26 @@ class TestWorkerPoller: assert row["completed_at"] is not None @pytest.mark.asyncio - async def test_executor_exception_does_not_crash_poller(self, pool, clean_operations): - """Test that unexpected exceptions from executor are caught and don't crash the poller. + async def test_executor_exception_triggers_retry(self, pool, clean_operations): + """Test that exceptions from the executor trigger _retry_or_fail (not a crash). - If the executor raises an unexpected exception (which MemoryEngine.execute_task should NOT do, - but could happen from schema setup or other infrastructure issues), the poller should catch it - gracefully. Status remains 'processing' since neither executor nor poller handled it. + When the executor re-raises an exception (as MemoryEngine.execute_task does for + retryable task failures), the poller calls _retry_or_fail, which resets the task + back to 'pending' and increments retry_count so it can be reclaimed. + + This is the fix for the consolidation deadlock: previously submit_task was called + with only a task_payload update, leaving status='processing' forever. """ from hindsight_api.worker import WorkerPoller from hindsight_api.worker.poller import ClaimedTask - # Create a pending task bank_id = f"test-worker-{uuid.uuid4().hex[:8]}" op_id = uuid.uuid4() - payload = json.dumps({"type": "test_task", "operation_id": str(op_id), "bank_id": bank_id}) + payload = json.dumps({"type": "consolidation", "operation_id": str(op_id), "bank_id": bank_id}) await pool.execute( """ - INSERT INTO async_operations (operation_id, bank_id, operation_type, status, task_payload, worker_id) - VALUES ($1, $2, 'test', 'processing', $3::jsonb, 'test-worker-1') + INSERT INTO async_operations (operation_id, bank_id, operation_type, status, task_payload, worker_id, claimed_at) + VALUES ($1, $2, 'consolidation', 'processing', $3::jsonb, 'test-worker-1', now()) """, op_id, bank_id, @@ -293,43 +295,100 @@ class TestWorkerPoller: ) async def failing_executor(task_dict): - raise ValueError("Unexpected infrastructure failure") + raise ValueError("TimeoutError during recall") poller = WorkerPoller( pool=pool, worker_id="test-worker-1", executor=failing_executor, + max_retries=3, ) - # Execute - should catch exception without crashing task_dict = json.loads(payload) claimed_task = ClaimedTask(operation_id=str(op_id), task_dict=task_dict, schema=None) await poller.execute_task(claimed_task) - # Wait for background task to complete completed = await poller.wait_for_active_tasks(timeout=5.0) assert completed, "Task did not complete within timeout" - # Status stays 'processing' since the poller no longer manages status + # Task must be reset to 'pending' with worker_id/claimed_at cleared — not left as + # 'processing', which would cause a permanent deadlock via the NOT EXISTS guard. row = await pool.fetchrow( - "SELECT status FROM async_operations WHERE operation_id = $1", + "SELECT status, worker_id, claimed_at, retry_count FROM async_operations WHERE operation_id = $1", op_id, ) - assert row["status"] == "processing" + assert row["status"] == "pending", ( + f"REGRESSION: Task status is '{row['status']}' instead of 'pending'. " + "A task stuck in 'processing' after a retry causes a consolidation deadlock." + ) + assert row["worker_id"] is None, "worker_id must be cleared on retry" + assert row["claimed_at"] is None, "claimed_at must be cleared on retry" + assert row["retry_count"] == 1 + + @pytest.mark.asyncio + async def test_executor_exception_marks_failed_after_max_retries(self, pool, clean_operations): + """Test that a task is permanently marked 'failed' once retry_count hits max_retries. + + After max_retries exhaustion the task must NOT be reset to 'pending' — it should + be marked 'failed' with an error message so it stops consuming retry budget. + """ + from hindsight_api.worker import WorkerPoller + from hindsight_api.worker.poller import ClaimedTask + + max_retries = 3 + bank_id = f"test-worker-{uuid.uuid4().hex[:8]}" + op_id = uuid.uuid4() + payload = json.dumps({"type": "consolidation", "operation_id": str(op_id), "bank_id": bank_id}) + # Insert with retry_count already at the limit + await pool.execute( + """ + INSERT INTO async_operations (operation_id, bank_id, operation_type, status, task_payload, worker_id, claimed_at, retry_count) + VALUES ($1, $2, 'consolidation', 'processing', $3::jsonb, 'test-worker-1', now(), $4) + """, + op_id, + bank_id, + payload, + max_retries, + ) + + async def failing_executor(task_dict): + raise ValueError("Still failing after all retries") + + poller = WorkerPoller( + pool=pool, + worker_id="test-worker-1", + executor=failing_executor, + max_retries=max_retries, + ) + + task_dict = json.loads(payload) + claimed_task = ClaimedTask(operation_id=str(op_id), task_dict=task_dict, schema=None) + await poller.execute_task(claimed_task) + + completed = await poller.wait_for_active_tasks(timeout=5.0) + assert completed, "Task did not complete within timeout" + + row = await pool.fetchrow( + "SELECT status, error_message, retry_count FROM async_operations WHERE operation_id = $1", + op_id, + ) + assert row["status"] == "failed", ( + f"Expected 'failed' after max retries, got '{row['status']}'" + ) + assert row["error_message"] is not None + assert "Max retries" in row["error_message"] + assert row["retry_count"] == max_retries # not incremented further @pytest.mark.asyncio async def test_executor_failed_status_not_overridden(self, pool, clean_operations): """REGRESSION TEST: Verify poller does NOT overwrite executor's 'failed' status to 'completed'. - This test catches the bug where the poller always called _mark_completed() after executor - returned, overwriting the 'failed' status that the executor had already set. - - Scenario: - 1. Executor catches an internal error and marks the operation as 'failed' in the DB - 2. Executor returns normally (does NOT re-raise) - this is how MemoryEngine.execute_task works + This test covers the non-retryable failure path (e.g., file_convert_retain): + 1. Executor catches an internal error, marks the operation as 'failed' in the DB + 2. Executor returns normally (does NOT re-raise) — so no exception reaches the poller 3. The poller must NOT overwrite the 'failed' status to 'completed' - With the old buggy code, this test would FAIL (status would be 'completed'). + Retryable failures re-raise instead (see test_executor_exception_triggers_retry). """ from hindsight_api.worker import WorkerPoller from hindsight_api.worker.poller import ClaimedTask