Function is_task_lock_held

core/utils.py:314–320

Return True when another worker holds the task lock (read-only check).

(task_name, id)

def is_task_lock_held(task_name, id):
    """Return True when another worker holds the task lock (read-only check)."""
    redis_client = RedisClient.get_client()
    if redis_client is None:
        return False
    lock_id = f"task_lock_{task_name}_{id}"
    return bool(redis_client.exists(lock_id))


class TaskLockRenewer: