Split heartbeat instance check-in into helpers

SonarCloud flagged _heartbeat_instance_management's cognitive complexity.
Pull the receptor status read, this instance's check-in and the missing
record registration into small functions, and cover each.

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

awx/main/tasks/system.py

@@ -788,17 +788,48 @@ def _mesh_all_ready_nodes_visible(mesh_status):
     return True
 
 
+def _receptor_mesh_status(ctl):
+    """Return receptor's status, or None when it can't be read."""
+    try:
+        return ctl.simple_command('status')
+    except (OSError, RuntimeError, ValueError) as exc:
+        logger.warning(f'Receptor status unavailable: {exc}')
+        return None
+
+
+def _check_in_this_instance(this_inst, nowtime):
+    """Run this instance's health check. Return False when it is rejoining the cluster."""
+    startup_event = this_inst.is_lost(ref_time=nowtime)
+    last_last_seen = this_inst.last_seen
+    this_inst.local_health_check()
+    if startup_event and this_inst.capacity != 0:
+        logger.warning(f'Rejoining the cluster as instance {this_inst.hostname}. Prior last_seen {last_last_seen}')
+        return False
+    if not last_last_seen:
+        logger.warning(f'Instance does not have recorded last_seen, updating to {nowtime}')
+    elif (nowtime - last_last_seen) > timedelta(seconds=settings.CLUSTER_NODE_HEARTBEAT_PERIOD + 2):
+        logger.warning(f'Heartbeat skew - interval={(nowtime - last_last_seen).total_seconds():.4f}, expected={settings.CLUSTER_NODE_HEARTBEAT_PERIOD}')
+    return True
+
+
+def _register_missing_this_instance():
+    """Recreate this instance's record when auto-deprovisioning removed it. Return None when it can't."""
+    if not settings.AWX_AUTO_DEPROVISION_INSTANCES:
+        logger.error("Cluster Host Not Found: {}".format(settings.CLUSTER_HOST_ID))
+        return None
+    changed, this_inst = Instance.objects.register(ip_address=os.environ.get('MY_POD_IP'), node_type='control', node_uuid=settings.SYSTEM_UUID)
+    if changed:
+        logger.warning(f'Recreated instance record {this_inst.hostname} after unexpected removal')
+    this_inst.local_health_check()
+    return this_inst
+
+
 def _heartbeat_instance_management(clock=None):
     """Common logic for heartbeat instance management."""
     logger.debug("Cluster node heartbeat task.")
     nowtime = (clock or now)()
     instance_list = list(Instance.objects.filter(node_state__in=(Instance.States.READY, Instance.States.UNAVAILABLE, Instance.States.INSTALLED)))
-    this_inst = None
-
-    for inst in instance_list:
-        if inst.hostname == settings.CLUSTER_HOST_ID:
-            this_inst = inst
-            break
+    this_inst = next((inst for inst in instance_list if inst.hostname == settings.CLUSTER_HOST_ID), None)
 
     try:
         ctl = get_receptor_ctl()
@@ -809,35 +840,18 @@ def _heartbeat_instance_management(clock=None):
             this_inst.mark_offline(errors='Receptor not available')
         return None, None, None, None
 
-    try:
-        mesh_status = ctl.simple_command('status')
-    except (OSError, RuntimeError, ValueError) as exc:
-        logger.warning(f'Receptor status unavailable: {exc}')
-        mesh_status = None
+    mesh_status = _receptor_mesh_status(ctl)
 
     inspect_execution_and_hop_nodes(instance_list, mesh_status)
 
     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)
-        last_last_seen = this_inst.last_seen
-        this_inst.local_health_check()
-        if startup_event and this_inst.capacity != 0:
-            logger.warning(f'Rejoining the cluster as instance {this_inst.hostname}. Prior last_seen {last_last_seen}')
+        if not _check_in_this_instance(this_inst, nowtime):
             return None, None, None, None  # Early return case
-        elif not last_last_seen:
-            logger.warning(f'Instance does not have recorded last_seen, updating to {nowtime}')
-        elif (nowtime - last_last_seen) > timedelta(seconds=settings.CLUSTER_NODE_HEARTBEAT_PERIOD + 2):
-            logger.warning(f'Heartbeat skew - interval={(nowtime - last_last_seen).total_seconds():.4f}, expected={settings.CLUSTER_NODE_HEARTBEAT_PERIOD}')
     else:
-        if settings.AWX_AUTO_DEPROVISION_INSTANCES:
-            changed, this_inst = Instance.objects.register(ip_address=os.environ.get('MY_POD_IP'), node_type='control', node_uuid=settings.SYSTEM_UUID)
-            if changed:
-                logger.warning(f'Recreated instance record {this_inst.hostname} after unexpected removal')
-            this_inst.local_health_check()
-        else:
-            logger.error("Cluster Host Not Found: {}".format(settings.CLUSTER_HOST_ID))
+        this_inst = _register_missing_this_instance()
+        if this_inst is None:
             return None, None, None, None
 
     lost_instances = gate_lost_instances(lost_instances, lambda: _mesh_all_ready_nodes_visible(mesh_status))

awx/main/tests/functional/tasks/test_tasks_system.py

@@ -676,6 +676,18 @@ def test_heartbeat_marks_offline_when_receptor_unavailable(settings):
     assert this_inst.node_state == Instance.States.UNAVAILABLE
 
 
+@pytest.mark.django_db
+def test_heartbeat_stops_when_this_instance_is_missing(settings):
+    """No record for this node and no auto-deprovision → returns (None, None, None, None)."""
+    settings.CLUSTER_HOST_ID = 'ctrl-0'
+    settings.AWX_AUTO_DEPROVISION_INSTANCES = False
+    ctl = mock.MagicMock()
+    ctl.simple_command.return_value = {'KnownConnectionCosts': {'ctrl-0': {}}}
+
+    with mock.patch('awx.main.tasks.system.get_receptor_ctl', return_value=ctl):
+        assert _heartbeat_instance_management() == (None, None, None, None)
+
+
 # ── AAP-89607: unified startup job processing ─────────────────────────────────
 
 

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

@@ -1,6 +1,12 @@
 import pytest
+from datetime import datetime, timedelta, timezone
 from unittest.mock import MagicMock, patch
-from awx.main.tasks.system import update_inventory_computed_fields
+from awx.main.tasks.system import (
+    _check_in_this_instance,
+    _receptor_mesh_status,
+    _register_missing_this_instance,
+    update_inventory_computed_fields,
+)
 from awx.main.models import Inventory
 from django.db import DatabaseError
 
@@ -62,3 +68,60 @@ def test_update_inventory_computed_fields_database_error_nosqlstate(mock_logger,
             mock_filter.assert_called_once_with(id=1)
             mock_update_computed_fields.assert_called_once()
             mock_inventory.update_computed_fields.assert_called_once()
+
+
+NOW = datetime(2026, 1, 1, tzinfo=timezone.utc)
+
+
+def test_receptor_mesh_status_returns_status():
+    ctl = MagicMock()
+    ctl.simple_command.return_value = {'KnownConnectionCosts': {}}
+
+    assert _receptor_mesh_status(ctl) == {'KnownConnectionCosts': {}}
+
+
+def test_receptor_mesh_status_unavailable_is_none(mock_logger):
+    ctl = MagicMock()
+    ctl.simple_command.side_effect = OSError('socket gone')
+
+    assert _receptor_mesh_status(ctl) is None
+
+
+def _this_inst(lost=False, capacity=100, last_seen=NOW):
+    inst = MagicMock(hostname='ctrl-0', capacity=capacity, last_seen=last_seen)
+    inst.is_lost.return_value = lost
+    return inst
+
+
+def test_check_in_rejoining_instance_returns_false(mock_logger):
+    inst = _this_inst(lost=True)
+
+    assert _check_in_this_instance(inst, NOW) is False
+    inst.local_health_check.assert_called_once_with()
+
+
+def test_check_in_lost_instance_without_capacity_continues(mock_logger):
+    assert _check_in_this_instance(_this_inst(lost=True, capacity=0), NOW) is True
+
+
+@pytest.mark.parametrize('last_seen, warns', [(None, True), (NOW - timedelta(seconds=600), True), (NOW, False)])
+def test_check_in_warns_on_missing_or_skewed_last_seen(mock_logger, last_seen, warns):
+    assert _check_in_this_instance(_this_inst(last_seen=last_seen), NOW) is True
+    assert mock_logger.warning.called is warns
+
+
+@pytest.mark.parametrize('changed', [True, False])
+def test_register_missing_this_instance_recreates_record(mock_logger, settings, changed):
+    settings.AWX_AUTO_DEPROVISION_INSTANCES = True
+    inst = MagicMock()
+    with patch('awx.main.tasks.system.Instance.objects.register', return_value=(changed, inst)):
+        assert _register_missing_this_instance() is inst
+    inst.local_health_check.assert_called_once_with()
+    assert mock_logger.warning.called is changed
+
+
+def test_register_missing_this_instance_without_auto_deprovision(mock_logger, settings):
+    settings.AWX_AUTO_DEPROVISION_INSTANCES = False
+
+    assert _register_missing_this_instance() is None
+    mock_logger.error.assert_called_once()