Task State Manager#

class ccat_data_transfer.task_state_manager.TaskStateManager(redis_client)[source]#

Bases: object

Manager for tracking and recovering task states across all operation types.

__init__(redis_client)[source]#
register_task(task_id, operation_type, operation_id, additional_info=None, max_retries=3)[source]#

Register a task in Redis with its metadata.

Parameters:

  • task_id (str) – Celery task ID

  • operation_type (str) – Type of operation (transfer, archive, package, delete, verify)

  • operation_id (int) – Database ID of the operation

  • additional_info (dict, optional) – Additional context about the operation

  • max_retries (int, optional) – Maximum retry count for this task

update_heartbeat(task_id)[source]#

Update task heartbeat to indicate it’s still running.

complete_task(task_id)[source]#

Mark task as completed and remove from tracking.

fail_task(task_id, error_message, is_retryable=True)[source]#

Mark task as failed.

Returns:

(can_retry, operation_type, operation_id)

Return type:

tuple

get_stalled_tasks(heartbeat_timeout=300)[source]#

Find tasks that haven’t updated their heartbeat recently.

Returns:

List of dicts with task information

Return type:

list

is_operation_alive(operation_type, operation_id, heartbeat_timeout=None)[source]#

Report whether an operation still has a live worker behind it.

Reverse-lookup of the per-operation index: read the sibling task_ids in tasks_for_operation:{operation_type}:{operation_id} and inspect each task:{task_id} hash. The operation is alive only if at least one sibling is RUNNING with a heartbeat fresher than heartbeat_timeout.

This is the liveness probe the archive reconciler gates on (#161): the primary dead-worker path is task_monitor heartbeat recovery, and this lets the reconciler stand down whenever a registered task is still beating.

Parameters:

  • operation_type – Operation identity — an OperationKind member (preferred) or its breadcrumb string (e.g. OperationKind.ARCHIVE / "long_term_archive"). The member’s value is the on-the-wire key, so both build the same Redis key.

  • operation_id (int) – Database ID of the operation.

  • heartbeat_timeout (int, optional) – Max heartbeat age in seconds to count a sibling as alive. Defaults to TASK_RECOVERY.heartbeat_timeout.

Returns:

True if a sibling task is RUNNING with a fresh heartbeat;

False if the index set is empty, the hashes are missing, or every sibling is stale or non-RUNNING.

Return type:

bool

Overview#

The Task State Manager manages task state transitions and persistence.

Key Functions#