Separate heartbeat adoption decisions from their side effects

Move the decisions the heartbeat makes about lost instances and orphaned
jobs into awx.main.tasks.adoption_decisions: which peers are lost, which
lost instances the mesh gate lets through, what happens to a lost
instance and its jobs, whether a running job is adopted, reaped or left
alone, and whether an orphan is past the adoption deadline. The
functions read plain attributes and touch no database, receptor or
dispatcher, so they can be tested with simple objects.

awx.main.tasks.system keeps the queries and side effects and calls the
decisions. Behavior is unchanged.

Co-Authored-By: Claude <noreply@anthropic.com>

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)

awx/main/tasks/system.py

@@ -71,6 +71,17 @@ from awx.main.models import (
     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

awx/main/tests/unit/tasks/test_adoption_decisions.py

@@ -0,0 +1,207 @@
+from datetime import datetime, timezone
+from types import SimpleNamespace
+
+import pytest
+from django.utils.timezone import timedelta
+
+from awx.main.models import Instance
+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,
+)
+
+NOW = datetime(2026, 1, 1, 12, 0, 0, tzinfo=timezone.utc)
+ME = 'awx-1'
+PEER = 'awx-2'
+EXEC = 'receptor-1'
+WORKFLOW_CTYPE = 99
+JOB_CTYPE = 1
+
+
+def make_job(**kwargs):
+    fields = dict(
+        status='running',
+        controller_node=ME,
+        execution_node=EXEC,
+        polymorphic_ctype_id=JOB_CTYPE,
+        celery_task_id='task-1',
+        started=NOW - timedelta(minutes=5),
+        work_unit_id='unit-1',
+    )
+    fields.update(kwargs)
+    return SimpleNamespace(**fields)
+
+
+def make_instance(hostname, node_type='control', seconds_since_seen=0, node_state='ready'):
+    last_seen = None if seconds_since_seen is None else NOW - timedelta(seconds=seconds_since_seen)
+    return Instance(hostname=hostname, node_type=node_type, node_state=node_state, last_seen=last_seen)
+
+
+@pytest.fixture
+def heartbeat_settings(settings):
+    settings.CLUSTER_NODE_HEARTBEAT_PERIOD = 60
+    settings.CLUSTER_NODE_MISSED_HEARTBEAT_TOLERANCE = 2
+    settings.RECEPTOR_SERVICE_ADVERTISEMENT_PERIOD = 60
+    return settings
+
+
+class TestFindLostInstances:
+    def test_peer_within_grace_period_is_not_lost(self, heartbeat_settings):
+        peer = make_instance(PEER, seconds_since_seen=119)
+        lost, remaining = find_lost_instances([peer], ME, NOW)
+        assert lost == []
+        assert remaining == [peer]
+
+    def test_peer_past_grace_period_is_lost(self, heartbeat_settings):
+        peer = make_instance(PEER, seconds_since_seen=121)
+        lost, remaining = find_lost_instances([peer], ME, NOW)
+        assert lost == [peer]
+        assert remaining == []
+
+    def test_peer_never_seen_is_lost(self, heartbeat_settings):
+        peer = make_instance(PEER, seconds_since_seen=None)
+        lost, _ = find_lost_instances([peer], ME, NOW)
+        assert lost == [peer]
+
+    def test_this_instance_is_never_lost_here(self, heartbeat_settings):
+        me = make_instance(ME, seconds_since_seen=10_000)
+        lost, remaining = find_lost_instances([me], ME, NOW)
+        assert lost == []
+        assert remaining == [me]
+
+    def test_execution_node_gets_advertisement_period_added(self, heartbeat_settings):
+        node = make_instance(EXEC, node_type='execution', seconds_since_seen=150)
+        lost, _ = find_lost_instances([node], ME, NOW)
+        assert lost == []
+
+    def test_slow_controller_is_judged_lost_while_still_running(self, heartbeat_settings):
+        # A controller whose heartbeat is delayed past the grace period looks exactly like
+        # a dead one. The decision has only last_seen to go on.
+        slow = make_instance(PEER, seconds_since_seen=125)
+        lost, _ = find_lost_instances([slow], ME, NOW)
+        assert lost == [slow]
+
+    def test_clock_skew_on_the_judge_makes_a_healthy_peer_look_lost(self, heartbeat_settings):
+        # The peer heartbeated just now by its own clock, but this instance's clock runs
+        # five minutes ahead.
+        peer = make_instance(PEER, seconds_since_seen=0)
+        lost, _ = find_lost_instances([peer], ME, NOW + timedelta(minutes=5))
+        assert lost == [peer]
+
+
+class TestGateLostInstances:
+    def test_mesh_ready_handles_every_lost_instance(self):
+        lost = [make_instance(PEER), make_instance(EXEC, node_type='execution'), make_instance('hop-1', node_type='hop')]
+        assert gate_lost_instances(lost, mesh_ready=True) == lost
+
+    def test_mesh_not_ready_defers_only_control_nodes(self):
+        control = make_instance(PEER)
+        execution = make_instance(EXEC, node_type='execution')
+        hop = make_instance('hop-1', node_type='hop')
+        hybrid = make_instance('hybrid-1', node_type='hybrid')
+        assert gate_lost_instances([control, execution, hop, hybrid], mesh_ready=False) == [execution, hop]
+
+
+class TestLostInstanceDecisions:
+    @pytest.mark.parametrize('work_unit_id', ['unit-1', ''])
+    def test_jobs_of_a_lost_instance_are_reaped(self, work_unit_id):
+        assert lost_instance_job_action(make_job(controller_node=PEER, work_unit_id=work_unit_id)) == JobAction.REAP
+
+    @pytest.mark.parametrize(
+        'node_type, node_state, auto_deprovision, expected',
+        [
+            ('control', 'ready', True, LostInstanceAction.DEPROVISION),
+            ('control', 'unavailable', True, LostInstanceAction.DEPROVISION),
+            ('execution', 'ready', True, LostInstanceAction.MARK_OFFLINE),
+            ('control', 'ready', False, LostInstanceAction.MARK_OFFLINE),
+            ('control', 'unavailable', False, LostInstanceAction.NONE),
+            ('control', 'installed', False, LostInstanceAction.NONE),
+            ('execution', 'unavailable', True, LostInstanceAction.NONE),
+        ],
+    )
+    def test_disposition(self, node_type, node_state, auto_deprovision, expected):
+        inst = make_instance(PEER, node_type=node_type, node_state=node_state)
+        assert lost_instance_disposition(inst, auto_deprovision) == expected
+
+
+class TestStartupJobAction:
+    def test_dispatched_job_is_adopted(self):
+        assert startup_job_action(make_job(work_unit_id='unit-1')) == JobAction.ADOPT
+
+    @pytest.mark.parametrize('work_unit_id', ['', None])
+    def test_undispatched_job_is_reaped(self, work_unit_id):
+        assert startup_job_action(make_job(work_unit_id=work_unit_id)) == JobAction.REAP
+
+
+class TestRunningJobAction:
+    def decide(self, job, active_task_ids=('other-task',), ref_time=NOW):
+        return running_job_action(job, ME, list(active_task_ids), ref_time, WORKFLOW_CTYPE)
+
+    def test_orphan_this_instance_controls_is_adopted(self):
+        assert self.decide(make_job()) == JobAction.ADOPT
+
+    def test_job_a_dispatcher_task_still_tracks_is_left_alone(self):
+        assert self.decide(make_job(celery_task_id='task-1'), active_task_ids=['task-1']) == JobAction.SKIP
+
+    def test_orphan_not_yet_dispatched_to_receptor_is_reaped(self):
+        assert self.decide(make_job(work_unit_id='')) == JobAction.REAP
+
+    def test_orphan_this_instance_only_executes_is_reaped(self):
+        # A hybrid node running a job for another controller: it can't adopt, so it reaps.
+        assert self.decide(make_job(controller_node=PEER, execution_node=ME)) == JobAction.REAP
+
+    def test_job_of_another_instance_is_left_alone(self):
+        assert self.decide(make_job(controller_node=PEER)) == JobAction.SKIP
+
+    @pytest.mark.parametrize('status', ['pending', 'waiting', 'successful', 'failed', 'canceled'])
+    def test_job_not_running_is_left_alone(self, status):
+        assert self.decide(make_job(status=status)) == JobAction.SKIP
+
+    def test_workflow_job_is_left_alone(self):
+        assert self.decide(make_job(polymorphic_ctype_id=WORKFLOW_CTYPE)) == JobAction.SKIP
+
+    def test_job_started_after_ref_time_is_left_alone(self):
+        assert self.decide(make_job(started=NOW + timedelta(seconds=1))) == JobAction.SKIP
+
+    def test_job_without_start_time_is_left_alone(self):
+        assert self.decide(make_job(started=None)) == JobAction.SKIP
+
+    def test_no_ref_time_considers_every_start_time(self):
+        assert self.decide(make_job(started=NOW + timedelta(days=1)), ref_time=None) == JobAction.ADOPT
+
+    def test_empty_active_list_treats_every_job_as_orphaned(self):
+        assert self.decide(make_job(), active_task_ids=()) == JobAction.ADOPT
+
+
+class TestAdoptionDeadline:
+    TIMEOUT = 300
+
+    def test_recent_event_is_within_deadline(self):
+        assert not adoption_deadline_passed(NOW - timedelta(seconds=299), NOW - timedelta(hours=1), NOW, self.TIMEOUT)
+
+    def test_old_last_event_is_past_deadline(self):
+        assert adoption_deadline_passed(NOW - timedelta(seconds=301), NOW - timedelta(hours=1), NOW, self.TIMEOUT)
+
+    def test_last_event_wins_over_start_time(self):
+        # A long job that is still producing events is not timed out by its age.
+        assert not adoption_deadline_passed(NOW - timedelta(seconds=10), NOW - timedelta(days=1), NOW, self.TIMEOUT)
+
+    def test_start_time_used_when_no_events(self):
+        assert adoption_deadline_passed(None, NOW - timedelta(seconds=301), NOW, self.TIMEOUT)
+        assert not adoption_deadline_passed(None, NOW - timedelta(seconds=299), NOW, self.TIMEOUT)
+
+    def test_job_with_no_events_or_start_is_never_past_deadline(self):
+        assert not adoption_deadline_passed(None, None, NOW, self.TIMEOUT)
+
+    def test_quiet_job_is_past_deadline_even_while_healthy(self):
+        # A playbook task that runs silently longer than the timeout (a long pause, a slow
+        # package install) looks the same as a dead one.
+        last_event = NOW - timedelta(seconds=self.TIMEOUT + 1)
+        assert adoption_deadline_passed(last_event, NOW - timedelta(seconds=self.TIMEOUT + 60), NOW, self.TIMEOUT)