chrismeyersfsu · Oct 11, 2026
diff --git a/awx/main/tasks/adoption_decisions.py b/awx/main/tasks/adoption_decisions.py
@@ -0,0 +1,110 @@
+"""Decisions the heartbeat makes about lost instances and orphaned jobs.
+
+Nothing in this module touches the database, receptor or the dispatcher. The heartbeat in
+awx.main.tasks.system gathers state, asks these functions what to do, and carries out the
+answer. Keeping the decisions separate lets them be tested with plain objects, without a
+cluster, and gives cross-controller adoption (AAP-89602) one place to change each rule.
+
+Jobs and instances are read only through attributes, so model instances and simple
+stand-ins in tests both work.
+"""
+
+from enum import Enum
+
+from django.utils.timezone import timedelta
+
+
+class JobAction(str, Enum):
+    SKIP = 'skip'
+    ADOPT = 'adopt'
+    REAP = 'reap'
+
+
+class LostInstanceAction(str, Enum):
+    DEPROVISION = 'deprovision'
+    MARK_OFFLINE = 'mark_offline'
+    NONE = 'none'
+
+
+def find_lost_instances(instance_list, this_hostname, ref_time):
+    """Split peers into (lost, remaining). This instance is never lost here; it is judged separately."""
+    lost = []
+    remaining = []
+    for inst in instance_list:
+        if inst.hostname != this_hostname and inst.is_lost(ref_time=ref_time):
+            lost.append(inst)
+        else:
+            remaining.append(inst)
+    return lost, remaining
+
+
+def gate_lost_instances(lost_instances, mesh_ready):
+    """Return the lost instances that may be handled now.
+
+    While the receptor mesh is re-establishing, a control node may only look lost, so its
+    handling waits for a later heartbeat. Execution and hop nodes don't depend on mesh
+    consensus and are handled either way.
+    """
+    if mesh_ready:
+        return list(lost_instances)
+    return [inst for inst in lost_instances if inst.node_type in ('execution', 'hop')]
+
+
+def lost_instance_job_action(job):
+    """What to do with a running job whose controller or execution node was lost.
+
+    AAP-89602: cross-controller adoption for dispatched jobs will return ADOPT here
+    once ansible/receptor#1564 merges. Until then every such job is reaped.
+    """
+    return JobAction.REAP
+
+
+def lost_instance_disposition(inst, auto_deprovision):
+    """What to do with the lost instance itself once its jobs are handled."""
+    if auto_deprovision and inst.node_type == 'control':
+        return LostInstanceAction.DEPROVISION
+    if inst.node_state == 'ready':
+        return LostInstanceAction.MARK_OFFLINE
+    return LostInstanceAction.NONE
+
+
+def startup_job_action(job):
+    """At startup, a running job this controller owns is adopted if it was dispatched to receptor, else reaped."""
+    if job.work_unit_id:
+        return JobAction.ADOPT
+    return JobAction.REAP
+
+
+def running_job_action(job, this_hostname, active_task_ids, ref_time, workflow_ctype_id):
+    """On a periodic heartbeat, what to do with a job that may have been orphaned on this instance.
+
+    A job is left alone unless it is running, belongs to this instance as controller or
+    execution node, is not a workflow, started before ref_time, and no dispatcher task is
+    tracking it. Such an orphan is adopted when this instance controls it and it was
+    dispatched to receptor; otherwise it is reaped.
+    """
+    if job.status != 'running':
+        return JobAction.SKIP
+    if this_hostname not in (job.controller_node, job.execution_node):
+        return JobAction.SKIP
+    if job.polymorphic_ctype_id == workflow_ctype_id:
+        return JobAction.SKIP
+    if active_task_ids and job.celery_task_id in active_task_ids:
+        return JobAction.SKIP
+    if ref_time and (job.started is None or job.started > ref_time):
+        return JobAction.SKIP
+    if job.work_unit_id and job.controller_node == this_hostname:
+        return JobAction.ADOPT
+    return JobAction.REAP
+
+
+def adoption_deadline_passed(last_event_time, started, ref_time, timeout_seconds):
+    """True if a job has been silent too long to adopt.
+
+    Silence is measured from the last saved event, or from the start when there are no
+    events. A job with neither is never past the deadline.
+    """
+    orphaned_since = last_event_time or started
+    if not orphaned_since:
+        return False
+    return orphaned_since < ref_time - timedelta(seconds=timeout_seconds)
diff --git a/awx/main/tasks/system.py b/awx/main/tasks/system.py
@@ -71,6 +71,17 @@
     convert_jsonfields,
 )
 from awx.main.models.credential import CredentialType
+from awx.main.tasks.adoption_decisions import (
+    JobAction,
+    LostInstanceAction,
+    adoption_deadline_passed,
+    find_lost_instances,
+    gate_lost_instances,
+    lost_instance_disposition,
+    lost_instance_job_action,
+    running_job_action,
+    startup_job_action,
+)
 from awx.main.tasks.helpers import is_run_threshold_reached
 from awx.main.tasks.host_indirect import save_indirect_host_entries
 from awx.main.tasks.receptor import (
@@ -771,7 +782,6 @@ def _heartbeat_instance_management():
     nowtime = now()
     instance_list = list(Instance.objects.filter(node_state__in=(Instance.States.READY, Instance.States.UNAVAILABLE, Instance.States.INSTALLED)))
     this_inst = None
-    lost_instances = []
 
     for inst in instance_list:
         if inst.hostname == settings.CLUSTER_HOST_ID:
@@ -795,12 +805,7 @@ def _heartbeat_instance_management():
 
     inspect_execution_and_hop_nodes(instance_list, mesh_status)
 
-    for inst in list(instance_list):
-        if inst == this_inst:
-            continue
-        if inst.is_lost(ref_time=nowtime):
-            lost_instances.append(inst)
-            instance_list.remove(inst)
+    lost_instances, instance_list = find_lost_instances(instance_list, settings.CLUSTER_HOST_ID, nowtime)
 
     if this_inst:
         startup_event = this_inst.is_lost(ref_time=nowtime)
@@ -823,11 +828,8 @@ def _heartbeat_instance_management():
             logger.error("Cluster Host Not Found: {}".format(settings.CLUSTER_HOST_ID))
             return None, None, None, None
 
-    if lost_instances and not _mesh_all_ready_nodes_visible(mesh_status):
-        # Mesh gate blocks cleanup, but execution and hop nodes can still be reaped
-        # (they don't depend on mesh consensus). Defer only control nodes.
-        execution_hop_lost = [inst for inst in lost_instances if inst.node_type in ('execution', 'hop')]
-        return this_inst, instance_list, execution_hop_lost, ctl
+    if lost_instances:
+        lost_instances = gate_lost_instances(lost_instances, _mesh_all_ready_nodes_visible(mesh_status))
 
     return this_inst, instance_list, lost_instances, ctl
 
@@ -862,19 +864,19 @@ def _reap_and_mark_lost_instance(other_inst):
             ).exclude(polymorphic_ctype_id=workflow_ctype_id)
         )
         for j in running_jobs:
-            # AAP-89602: cross-controller adoption for dispatched jobs will be added here
-            # once ansible/receptor#1564 merges. Both paths reap identically for now.
-            reaper.reap_job(j, 'failed', job_explanation='Job reaped due to instance shutdown')
+            if lost_instance_job_action(j) == JobAction.REAP:
+                reaper.reap_job(j, 'failed', job_explanation='Job reaped due to instance shutdown')
         # Any jobs that were waiting to be processed by this node will be handed back to task manager
         UnifiedJob.objects.filter(status='waiting', controller_node=other_inst.hostname).update(status='pending', controller_node='', execution_node='')
     except Exception:
         logger.exception('failed to re-process jobs for lost instance {}'.format(other_inst.hostname))
     try:
-        if settings.AWX_AUTO_DEPROVISION_INSTANCES and other_inst.node_type == "control":
+        disposition = lost_instance_disposition(other_inst, settings.AWX_AUTO_DEPROVISION_INSTANCES)
+        if disposition == LostInstanceAction.DEPROVISION:
             deprovision_hostname = other_inst.hostname
             other_inst.delete()  # FIXME: what about associated inbound links?
             logger.info("Host {} Automatically Deprovisioned.".format(deprovision_hostname))
-        elif other_inst.node_state == Instance.States.READY:
+        elif disposition == LostInstanceAction.MARK_OFFLINE:
             other_inst.mark_offline(errors=_('Another cluster node has determined this instance to be unresponsive'))
             logger.error("Host {} last checked in at {}, marked as lost.".format(other_inst.hostname, other_inst.last_seen))
 
@@ -950,7 +952,7 @@ def _process_startup_jobs(this_inst):
     reaped_ids = []
     for j in jobs:
         try:
-            if j.work_unit_id:
+            if startup_job_action(j) == JobAction.ADOPT:
                 obj, _ = adopt_job_async.apply_async(args=[j.id], queue=get_task_queuename())
                 UnifiedJob.objects.filter(pk=j.id).update(celery_task_id=obj['uuid'])
             else:
@@ -975,6 +977,7 @@ def _process_running_jobs(this_inst, active_task_ids, ref_time):
     - Undispatched or not owned by this controller → reap.
 
     Cross-controller adoption (jobs from a dead peer controller) is deferred to AAP-89602.
+    The query narrows the candidates; running_job_action makes the decision for each one.
     """
     workflow_ctype_id = ContentType.objects.get_for_model(WorkflowJob).id
     base_q = Q(status='running') & (Q(execution_node=this_inst.hostname) | Q(controller_node=this_inst.hostname))
@@ -989,12 +992,13 @@ def _process_running_jobs(this_inst, active_task_ids, ref_time):
 
     for j in jobs:
         try:
-            if j.work_unit_id and j.controller_node == this_inst.hostname:
+            action = running_job_action(j, this_inst.hostname, active_task_ids, ref_time, workflow_ctype_id)
+            if action == JobAction.ADOPT:
                 obj, _ = adopt_job_async.apply_async(args=[j.id], queue=get_task_queuename())
                 # Record the adoption task UUID so subsequent heartbeats see this job
                 # as active and skip it — preventing redundant re-adoption dispatches.
                 UnifiedJob.objects.filter(pk=j.id).update(celery_task_id=obj['uuid'])
-            else:
+            elif action == JobAction.REAP:
                 reaper.reap_job(j, 'failed')
         except Exception:
             logger.exception(f'Failed processing job {j.id} in heartbeat job loop')
@@ -1014,10 +1018,8 @@ def adopt_job_async(job_id):
         return
 
     adoption_timeout = settings.HADR_JOB_ADOPTION_TIMEOUT
-    timeout_cutoff = now() - timedelta(seconds=adoption_timeout)
     last_event_time = job.get_event_queryset().aggregate(Max('created'))['created__max']
-    orphaned_since = last_event_time or job.started
-    if orphaned_since and orphaned_since < timeout_cutoff:
+    if adoption_deadline_passed(last_event_time, job.started, now(), adoption_timeout):
         logger.error(f'Job {job.id} (unit={job.work_unit_id}) orphaned for >{adoption_timeout}s, failing')
         reaper.reap_job(job, 'failed', job_explanation='Job exceeded HADR_JOB_ADOPTION_TIMEOUT during controller restart')
         return