mirror of
https://github.com/tiennm99/DocsGPT.git
synced 2026-10-14 18:18:08 +00:00
fix(tasks): a deferred duplicate stands down instead of failing the task
A task that runs longer than the broker's visibility timeout is redelivered while its first run is still going. The idempotency lease correctly stops the duplicate from doing the work, but the duplicate then re-queued itself once per LEASE_TTL until celery ran out of retries and raised MaxRetriesExceededError — so a perfectly healthy long task (a large graph extraction is the one that found this) reported a task failure, with a traceback, while the real run was still making progress next to it. Catch the exhaustion and return a "deferred" result instead. A normal deferral still re-queues: only the give-up path changes, and the lease holder's dedup row is left untouched so its own completion still records.
This commit is contained in:
1 parent
e3d819d9fd
commit
6ea737e57b
2 files changed
+93
-4
No files matched your search
@@ -9,6 +9,8 @@ import threading
|
||||
import uuid
|
||||
from typing import Any, Callable, Optional
|
||||
|
||||
from celery.exceptions import MaxRetriesExceededError
|
||||
|
||||
from docsgpt.storage.db.repositories.idempotency import IdempotencyRepository
|
||||
from docsgpt.storage.db.session import db_readonly, db_session
|
||||
|
||||
@@ -81,10 +83,28 @@ def with_idempotency(
|
||||
"idempotency: live lease held; deferring task=%s key=%s",
|
||||
task_name, key,
|
||||
)
|
||||
raise self.retry(
|
||||
countdown=LEASE_TTL_SECONDS,
|
||||
max_retries=LEASE_RETRY_MAX,
|
||||
)
|
||||
try:
|
||||
raise self.retry(
|
||||
countdown=LEASE_TTL_SECONDS,
|
||||
max_retries=LEASE_RETRY_MAX,
|
||||
)
|
||||
except MaxRetriesExceededError:
|
||||
# The holder is simply slower than LEASE_RETRY_MAX
|
||||
# deferrals — a task that outruns the broker's visibility
|
||||
# timeout is redelivered while its first run is still
|
||||
# going. Standing down is the correct end state for the
|
||||
# duplicate; raising here would report a failure for a
|
||||
# task that is running normally somewhere else.
|
||||
logger.info(
|
||||
"idempotency: lease still held after %s deferrals; "
|
||||
"leaving task=%s key=%s to its holder",
|
||||
LEASE_RETRY_MAX, task_name, key,
|
||||
)
|
||||
return {
|
||||
"status": "deferred",
|
||||
"reason": "another worker holds the lease",
|
||||
"idempotency_key": key,
|
||||
}
|
||||
|
||||
if attempt > MAX_TASK_ATTEMPTS:
|
||||
logger.error(
|
||||
|
||||
Reference in new issue
Block a user