holetron-lab/fleet-memory
1
0
Fork 0
Code Releases Packages Activity Actions

fix: resolve consolidation deadlock caused by zombie processing tasks on retry (#463)

Browse source 
* fix: resolve consolidation deadlock caused by zombie 'processing' tasks on retry

When a task failed and was rescheduled for retry, submit_task() only updated
task_payload without resetting status/worker_id/claimed_at. The task stayed
permanently in 'processing', blocking all future consolidation for that bank
via the NOT EXISTS guard in claim_batch().

Fix: remove the duplicate payload-based retry mechanism from execute_task().
Retryable failures now re-raise so the poller handles them via _retry_or_fail(),
which already correctly resets status='pending', worker_id=NULL, claimed_at=NULL
and uses the DB retry_count column as single source of truth.

Non-retryable tasks (file_convert_retain) continue to mark themselves failed
and return normally — no exception reaches the poller.

Tests: add regression tests for the retry path (status reset to pending) and
the max-retries exhaustion path (status set to failed).

* ci: re-trigger CI
This commit is contained in:
Nicolò Boschi 2026-03-02 11:45:41 +01:00 committed by GitHub
parent eaeaa1f24d
commit c2876490df
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 98 additions and 48 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

28
hindsight-api/hindsight_api/engine/memory_engine.py
View file

@ -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."""

15
hindsight-api/hindsight_api/worker/poller.py
View file

@ -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:
"""

103
hindsight-api/tests/test_worker.py
View file

@ -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