AAP-90120/90121: mesh stability gate and task manager lock (#16628)

AAP-90120: add _mesh_all_ready_nodes_visible() to block peer-judgment
while KnownConnectionCosts is empty after a controller restart. Gate
fails open when mesh_status is unavailable. Eliminates double receptor
status call per heartbeat.

AAP-90121: wrap each instance's reap+mark_offline block in
advisory_lock('task_manager_lock', wait=False) so the task manager
cannot dispatch to a node being concurrently marked offline. Extracts
_reap_and_mark_lost_instance() helper.

Tests: rewrite new test classes to use real Django DB (functional/)
instead of MagicMock objects.

---------

Signed-off-by: Hui Song <hsong@redhat.com>
Co-authored-by: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>

awx/main/tasks/system.py

@@ -588,23 +588,21 @@ def inspect_established_receptor_connections(mesh_status):
     InstanceLink.objects.bulk_update(update_links, ['link_state'])
 
 
-def inspect_execution_and_hop_nodes(instance_list, receptor_ctl):
+def inspect_execution_and_hop_nodes(instance_list, mesh_status):
     with advisory_lock('inspect_execution_and_hop_nodes_lock', wait=False) as acquired:
         if not acquired:
             logger.debug("Not running inspect_execution_and_hop_nodes, another instance holds lock")
             return
+        if mesh_status is None:
+            logger.debug("Not running inspect_execution_and_hop_nodes, mesh status unavailable")
+            return
         start = time.monotonic()
         node_lookup = {inst.hostname: inst for inst in instance_list}
-        try:
-            mesh_status = receptor_ctl.simple_command('status')
-        except ValueError as exc:
-            logger.error(f'Error running receptorctl status command, error: {str(exc)}')
-            return
 
         inspect_established_receptor_connections(mesh_status)
 
         nowtime = now()
-        workers = mesh_status['Advertisements']
+        workers = mesh_status.get('Advertisements') or []
         updated_count = 0
 
         for ad in workers:
@@ -727,6 +725,27 @@ def _get_active_task_ids_from_dispatcherd(binder):
         return None
 
 
+def _mesh_all_ready_nodes_visible(mesh_status):
+    """Return False if the receptor mesh is still re-establishing after a controller restart.
+
+    Uses KnownConnectionCosts as the stability signal: receptor's routing protocol
+    populates this table as connections establish via gossip. An empty table means no
+    routing has propagated yet (Window A — typically the first ~10s after restart).
+
+    No DB state is consulted. KnownConnectionCosts is maintained entirely by the receptor
+    Go process, making it a reliable mesh-state signal free of stale DB records.
+
+    Fails open (returns True) when mesh_status is None so existing error paths are
+    not bypassed.
+    """
+    if mesh_status is None:
+        return True  # fail open: status unavailable, let normal peer-judgment proceed
+    if not (mesh_status.get('KnownConnectionCosts') or {}):
+        logger.info('Mesh stability gate: routing table empty, deferring peer-judgment (receptor re-establishing)')
+        return False
+    return True
+
+
 def _heartbeat_instance_management():
     """Common logic for heartbeat instance management."""
     logger.debug("Cluster node heartbeat task.")
@@ -749,7 +768,13 @@ def _heartbeat_instance_management():
             this_inst.mark_offline(errors='Receptor not available')
         return None, None, None
 
-    inspect_execution_and_hop_nodes(instance_list, ctl)
+    try:
+        mesh_status = ctl.simple_command('status')
+    except (OSError, RuntimeError, ValueError) as exc:
+        logger.warning(f'Receptor status unavailable: {exc}')
+        mesh_status = None
+
+    inspect_execution_and_hop_nodes(instance_list, mesh_status)
 
     for inst in list(instance_list):
         if inst == this_inst:
@@ -779,6 +804,9 @@ def _heartbeat_instance_management():
             logger.error("Cluster Host Not Found: {}".format(settings.CLUSTER_HOST_ID))
             return None, None, None
 
+    if lost_instances and not _mesh_all_ready_nodes_visible(mesh_status):
+        return this_inst, instance_list, []
+
     return this_inst, instance_list, lost_instances
 
 
@@ -801,39 +829,50 @@ def _heartbeat_check_versions(this_inst, instance_list):
             raise RuntimeError("Shutting down.")
 
 
+def _reap_and_mark_lost_instance(other_inst):
+    """Reap a lost instance's running jobs and mark it offline (or deprovision it)."""
+    try:
+        # Only reap jobs never dispatched to receptor; dispatched jobs are left for adoption (AAP-89602)
+        explanation = "Job reaped due to instance shutdown"
+        reaper.reap(other_inst, job_explanation=explanation)
+        # 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":
+            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:
+            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))
+
+    except DatabaseError as e:
+        cause = e.__cause__
+        if cause and hasattr(cause, 'sqlstate'):
+            sqlstate = cause.sqlstate
+            sqlstate_str = psycopg.errors.lookup(sqlstate)
+            logger.debug('SQL Error state: {} - {}'.format(sqlstate, sqlstate_str))
+
+            if sqlstate == psycopg.errors.NoData:
+                logger.debug('Another instance has marked {} as lost'.format(other_inst.hostname))
+            else:
+                logger.exception("Error marking {} as lost.".format(other_inst.hostname))
+        else:
+            logger.exception('No SQL state available.  Error marking {} as lost'.format(other_inst.hostname))
+
+
 def _heartbeat_handle_lost_instances(lost_instances, this_inst):
     """Handle lost instances by reaping their running jobs and marking them offline."""
     for other_inst in lost_instances:
-        try:
+        # Serialize offline handling against the task manager so we never reap jobs
-            # Any jobs marked as running will be marked as error
+        # mid-schedule; if the lock is held, retry this instance on the next heartbeat.
-            explanation = "Job reaped due to instance shutdown"
+        with advisory_lock('task_manager_lock', wait=False) as acquired:
-            reaper.reap(other_inst, job_explanation=explanation)
+            if not acquired:
-            # Any jobs that were waiting to be processed by this node will be handed back to task manager
+                logger.info(f'task_manager_lock held, deferring offline handling for {other_inst.hostname} to next heartbeat cycle')
-            UnifiedJob.objects.filter(status='waiting', controller_node=other_inst.hostname).update(status='pending', controller_node='', execution_node='')
+                continue
-        except Exception:
+            _reap_and_mark_lost_instance(other_inst)
-            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":
-                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:
-                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))
-
-        except DatabaseError as e:
-            cause = e.__cause__
-            if cause and hasattr(cause, 'sqlstate'):
-                sqlstate = cause.sqlstate
-                sqlstate_str = psycopg.errors.lookup(sqlstate)
-                logger.debug('SQL Error state: {} - {}'.format(sqlstate, sqlstate_str))
-
-                if sqlstate == psycopg.errors.NoData:
-                    logger.debug('Another instance has marked {} as lost'.format(other_inst.hostname))
-                else:
-                    logger.exception("Error marking {} as lost.".format(other_inst.hostname))
-            else:
-                logger.exception('No SQL state available.  Error marking {} as lost'.format(other_inst.hostname))
 
 
 @task(queue=get_task_queuename, timeout=1800, on_duplicate='queue_one')

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

@@ -4,6 +4,7 @@ import logging
 import os
 import tempfile
 import shutil
+from contextlib import contextmanager
 from unittest import mock
 
 import pytest
@@ -14,9 +15,15 @@ from awx.main.tasks.system import (
     inspect_established_receptor_connections,
     clear_setting_cache,
     _batched_delete_inventory,
+    _mesh_all_ready_nodes_visible,
+    _heartbeat_handle_lost_instances,
+    _reap_and_mark_lost_instance,
+    inspect_execution_and_hop_nodes,
+    _heartbeat_instance_management,
 )
 from awx.main.management.commands.dispatcherd import Command
 from django.db import DatabaseError
+from django.utils.timezone import now, timedelta
 
 from awx.main.models import Instance, Inventory, Job, Organization, ReceptorAddress, InstanceLink
 from awx.main.models.inventory import Group, Host
@@ -250,3 +257,313 @@ def test_configure_dispatcher_logging_updates_level(settings):
 
     assert logging.getLogger('dispatcherd').level == logging.WARNING
     settings.LOGGING = original_logging_settings
+
+
+# ── Helpers ───────────────────────────────────────────────────────────────────
+
+_UNSET = object()
+
+
+def _mesh_status(known_costs=_UNSET, advertisements=_UNSET):
+    return {
+        'KnownConnectionCosts': {} if known_costs is _UNSET else known_costs,
+        'Advertisements': [] if advertisements is _UNSET else [{'NodeID': n} for n in (advertisements or [])],
+    }
+
+
+def _make_lock(acquired):
+    @contextmanager
+    def _cm(name, wait=True, **kwargs):
+        yield acquired
+
+    return _cm
+
+
+def _make_lock_sequence(sequence):
+    it = iter(sequence)
+
+    @contextmanager
+    def _cm(name, wait=True, **kwargs):
+        yield next(it)
+
+    return _cm
+
+
+# ── Tests: _mesh_all_ready_nodes_visible ──────────────────────────────────────
+
+
+@pytest.mark.django_db
+class TestMeshAllReadyNodesVisible:
+    """Gate uses KnownConnectionCosts (receptor routing table) as the stability signal.
+
+    No DB state is consulted — KnownConnectionCosts is maintained entirely by
+    receptor's routing protocol. Empty table = routing not yet established (Window A).
+    """
+
+    def test_defers_when_routing_table_empty(self):
+        status = _mesh_status(known_costs={})
+        assert _mesh_all_ready_nodes_visible(status) is False
+
+    def test_defers_when_routing_table_null(self):
+        """Receptor is a Go service; nil map marshals to JSON null → Python None."""
+        status = _mesh_status(known_costs=None)
+        assert _mesh_all_ready_nodes_visible(status) is False
+
+    def test_passes_when_routing_established(self):
+        status = _mesh_status(known_costs={'ctrl-0': {'ee-0': 1}})
+        assert _mesh_all_ready_nodes_visible(status) is True
+
+    def test_passes_when_routing_established_dead_ee_not_in_routing(self):
+        """Gate checks only whether routing exists, not which nodes appear."""
+        status = _mesh_status(known_costs={'ctrl-0': {'live-ee': 1}})
+        assert _mesh_all_ready_nodes_visible(status) is True
+
+    def test_fails_open_when_mesh_status_none(self):
+        """Fail open so existing peer-judgment error paths are not bypassed."""
+        assert _mesh_all_ready_nodes_visible(None) is True
+
+
+# ── Tests: _heartbeat_handle_lost_instances (task manager lock) ───────────────
+
+
+@pytest.mark.django_db
+class TestHeartbeatHandleLostInstancesLock:
+    def _run(self, lost_instances, lock_behavior):
+        with (
+            mock.patch('awx.main.tasks.system.reaper'),
+            mock.patch('awx.main.tasks.system.advisory_lock', lock_behavior),
+        ):
+            _heartbeat_handle_lost_instances(lost_instances, None)
+
+    def test_processes_instance_when_lock_acquired(self, settings):
+        settings.AWX_AUTO_DEPROVISION_INSTANCES = False
+        inst = Instance.objects.create(hostname='ctrl-1', node_type='control', node_state='ready')
+        self._run([inst], _make_lock(True))
+        inst.refresh_from_db()
+        assert inst.node_state == Instance.States.UNAVAILABLE
+
+    def test_skips_instance_when_lock_unavailable(self, settings):
+        settings.AWX_AUTO_DEPROVISION_INSTANCES = False
+        inst = Instance.objects.create(hostname='ctrl-1', node_type='control', node_state='ready')
+        self._run([inst], _make_lock(False))
+        inst.refresh_from_db()
+        assert inst.node_state == Instance.States.READY
+
+    def test_logs_when_instance_deferred(self, settings):
+        settings.AWX_AUTO_DEPROVISION_INSTANCES = False
+        inst = Instance.objects.create(hostname='ctrl-1', node_type='control', node_state='ready')
+        with (
+            mock.patch('awx.main.tasks.system.reaper'),
+            mock.patch('awx.main.tasks.system.advisory_lock', _make_lock(False)),
+            mock.patch('awx.main.tasks.system.logger') as mock_log,
+        ):
+            _heartbeat_handle_lost_instances([inst], None)
+        assert mock_log.info.called
+        assert 'ctrl-1' in mock_log.info.call_args[0][0]
+
+    def test_per_instance_lock_first_skipped_second_processed(self, settings):
+        settings.AWX_AUTO_DEPROVISION_INSTANCES = False
+        inst1 = Instance.objects.create(hostname='ctrl-1', node_type='control', node_state='ready')
+        inst2 = Instance.objects.create(hostname='ctrl-2', node_type='control', node_state='ready')
+        self._run([inst1, inst2], _make_lock_sequence([False, True]))
+        inst1.refresh_from_db()
+        inst2.refresh_from_db()
+        assert inst1.node_state == Instance.States.READY
+        assert inst2.node_state == Instance.States.UNAVAILABLE
+
+    def test_all_instances_processed_when_lock_always_acquired(self, settings):
+        settings.AWX_AUTO_DEPROVISION_INSTANCES = False
+        inst1 = Instance.objects.create(hostname='ctrl-1', node_type='control', node_state='ready')
+        inst2 = Instance.objects.create(hostname='ctrl-2', node_type='control', node_state='ready')
+        self._run([inst1, inst2], _make_lock(True))
+        inst1.refresh_from_db()
+        inst2.refresh_from_db()
+        assert inst1.node_state == Instance.States.UNAVAILABLE
+        assert inst2.node_state == Instance.States.UNAVAILABLE
+
+
+# ── Tests: _reap_and_mark_lost_instance (exception / branch coverage) ────────
+
+
+@pytest.mark.django_db
+class TestReapAndMarkLostInstance:
+    def _inst(self, **kwargs):
+        defaults = {'hostname': 'ctrl-1', 'node_type': 'control', 'node_state': 'ready'}
+        defaults.update(kwargs)
+        return Instance.objects.create(**defaults)
+
+    def test_reap_exception_does_not_prevent_mark_offline(self, settings):
+        settings.AWX_AUTO_DEPROVISION_INSTANCES = False
+        inst = self._inst()
+        with mock.patch('awx.main.tasks.system.reaper') as mock_reaper:
+            mock_reaper.reap.side_effect = Exception('receptor timeout')
+            _reap_and_mark_lost_instance(inst)
+        inst.refresh_from_db()
+        assert inst.node_state == Instance.States.UNAVAILABLE
+
+    def test_auto_deprovision_deletes_control_node(self, settings):
+        settings.AWX_AUTO_DEPROVISION_INSTANCES = True
+        inst = self._inst(hostname='ctrl-dep')
+        with mock.patch('awx.main.tasks.system.reaper'):
+            _reap_and_mark_lost_instance(inst)
+        assert not Instance.objects.filter(hostname='ctrl-dep').exists()
+
+    def test_skips_mark_offline_when_node_already_unavailable(self, settings):
+        settings.AWX_AUTO_DEPROVISION_INSTANCES = False
+        inst = self._inst(node_state='unavailable')
+        with mock.patch('awx.main.tasks.system.reaper'):
+            _reap_and_mark_lost_instance(inst)
+        inst.refresh_from_db()
+        assert inst.node_state == Instance.States.UNAVAILABLE
+
+    def test_database_error_without_sqlstate_logs_exception(self, settings):
+        settings.AWX_AUTO_DEPROVISION_INSTANCES = False
+        inst = self._inst()
+        err = DatabaseError('constraint violation')
+        err.__cause__ = None
+        with (
+            mock.patch('awx.main.tasks.system.reaper'),
+            mock.patch.object(inst, 'mark_offline', side_effect=err),
+            mock.patch('awx.main.tasks.system.logger') as mock_log,
+        ):
+            _reap_and_mark_lost_instance(inst)
+        assert mock_log.exception.called
+        assert 'No SQL state' in mock_log.exception.call_args[0][0]
+
+    def test_database_error_with_sqlstate_logs_details(self, settings):
+        settings.AWX_AUTO_DEPROVISION_INSTANCES = False
+        inst = self._inst()
+        err = DatabaseError('unique violation')
+
+        class _FakePsycopgError(Exception):
+            sqlstate = 'some_state'
+
+        err.__cause__ = _FakePsycopgError('underlying pg error')
+        with (
+            mock.patch('awx.main.tasks.system.reaper'),
+            mock.patch.object(inst, 'mark_offline', side_effect=err),
+            mock.patch('awx.main.tasks.system.logger') as mock_log,
+            mock.patch('awx.main.tasks.system.psycopg') as mock_psycopg,
+        ):
+            mock_psycopg.errors.lookup.return_value = 'SomeError'
+            mock_psycopg.errors.NoData = 'other_state'
+            _reap_and_mark_lost_instance(inst)
+        mock_psycopg.errors.lookup.assert_called_once_with('some_state')
+        assert mock_log.exception.called
+
+    def test_database_error_with_nodata_sqlstate_logs_debug(self, settings):
+        settings.AWX_AUTO_DEPROVISION_INSTANCES = False
+        inst = self._inst()
+        err = DatabaseError('nodata')
+
+        class _FakePsycopgError(Exception):
+            sqlstate = 'nodata_state'
+
+        err.__cause__ = _FakePsycopgError('nodata cause')
+        with (
+            mock.patch('awx.main.tasks.system.reaper'),
+            mock.patch.object(inst, 'mark_offline', side_effect=err),
+            mock.patch('awx.main.tasks.system.logger') as mock_log,
+            mock.patch('awx.main.tasks.system.psycopg') as mock_psycopg,
+        ):
+            mock_psycopg.errors.lookup.return_value = 'NoData'
+            mock_psycopg.errors.NoData = 'nodata_state'
+            _reap_and_mark_lost_instance(inst)
+        mock_log.exception.assert_not_called()
+        debug_messages = [str(call) for call in mock_log.debug.call_args_list]
+        assert any('marked' in m for m in debug_messages)
+
+
+# ── Tests: inspect_execution_and_hop_nodes ────────────────────────────────────
+
+
+@pytest.mark.django_db
+class TestInspectExecutionAndHopNodes:
+    @mock.patch('awx.main.tasks.system.inspect_established_receptor_connections')
+    def test_skips_when_lock_not_acquired(self, mock_inspect_conns):
+        with mock.patch('awx.main.tasks.system.advisory_lock', _make_lock(False)):
+            inspect_execution_and_hop_nodes([], _mesh_status())
+        mock_inspect_conns.assert_not_called()
+
+    @mock.patch('awx.main.tasks.system.inspect_established_receptor_connections')
+    def test_skips_when_mesh_status_none(self, mock_inspect_conns):
+        with mock.patch('awx.main.tasks.system.advisory_lock', _make_lock(True)):
+            inspect_execution_and_hop_nodes([], None)
+        mock_inspect_conns.assert_not_called()
+
+    @mock.patch('awx.main.tasks.system.inspect_established_receptor_connections')
+    def test_runs_when_lock_acquired(self, mock_inspect_conns):
+        with mock.patch('awx.main.tasks.system.advisory_lock', _make_lock(True)):
+            inspect_execution_and_hop_nodes([], _mesh_status())
+        mock_inspect_conns.assert_called_once()
+
+    @mock.patch('awx.main.tasks.system.inspect_established_receptor_connections')
+    def test_updates_last_seen_for_execution_nodes(self, mock_inspect_conns):
+        exec_node = Instance.objects.create(hostname='exec-1', node_type='execution', node_state='ready')
+        control_node = Instance.objects.create(hostname='control-1', node_type='control', node_state='ready')
+        status = {
+            'KnownConnectionCosts': {},
+            'Advertisements': [
+                {'NodeID': 'exec-1', 'Time': '2026-01-01T00:00:00+00:00'},
+                {'NodeID': 'control-1', 'Time': '2026-01-01T00:00:00+00:00'},
+            ],
+        }
+        with mock.patch('awx.main.tasks.system.advisory_lock', _make_lock(True)):
+            inspect_execution_and_hop_nodes([exec_node, control_node], status)
+        exec_node.refresh_from_db()
+        assert exec_node.last_seen is not None  # was None; function set it from Advertisements Time
+        control_node.refresh_from_db()
+        assert control_node.last_seen is None  # control nodes not updated by this function
+
+
+# ── Integration tests: _heartbeat_instance_management ────────────────────────
+
+
+@pytest.mark.django_db
+def test_heartbeat_defers_lost_instances_when_mesh_gate_blocks(settings):
+    """Gate returns False (empty KnownConnectionCosts) → lost_instances suppressed."""
+    settings.CLUSTER_HOST_ID = 'ctrl-0'
+    settings.AWX_AUTO_DEPROVISION_INSTANCES = False
+    settings.CLUSTER_NODE_HEARTBEAT_PERIOD = 60
+    settings.CLUSTER_NODE_MISSED_HEARTBEAT_TOLERANCE = 2
+
+    this_inst = Instance.objects.create(hostname='ctrl-0', node_type='control', node_state='ready')
+    this_inst.last_seen = now() - timedelta(seconds=30)
+    this_inst.save(update_fields=['last_seen'])
+
+    lost_peer = Instance.objects.create(hostname='ctrl-1', node_type='control', node_state='ready')
+    lost_peer.last_seen = now() - timedelta(seconds=200)  # > 120s grace → is_lost() True
+    lost_peer.save(update_fields=['last_seen'])
+
+    mock_ctl = mock.MagicMock()
+    mock_ctl.simple_command.return_value = {'KnownConnectionCosts': {}, 'Advertisements': []}
+
+    with (
+        mock.patch('awx.main.tasks.system.get_receptor_ctl', return_value=mock_ctl),
+        mock.patch('awx.main.tasks.system.inspect_execution_and_hop_nodes'),
+        mock.patch.object(Instance, 'local_health_check'),
+    ):
+        _, _, lost_result = _heartbeat_instance_management()
+
+    assert lost_result == []
+
+
+@pytest.mark.django_db
+def test_heartbeat_marks_offline_when_receptor_unavailable(settings):
+    """FileNotFoundError from get_receptor_ctl → this_inst marked offline, returns (None, None, None)."""
+    settings.CLUSTER_HOST_ID = 'ctrl-0'
+    settings.AWX_AUTO_DEPROVISION_INSTANCES = False
+
+    this_inst = Instance.objects.create(hostname='ctrl-0', node_type='control', node_state='ready')
+    this_inst.last_seen = now() - timedelta(seconds=30)
+    this_inst.save(update_fields=['last_seen'])
+
+    with (
+        mock.patch('awx.main.tasks.system.get_receptor_ctl', side_effect=FileNotFoundError),
+        mock.patch.object(Instance, 'local_health_check'),
+    ):
+        result = _heartbeat_instance_management()
+
+    assert result == (None, None, None)
+    this_inst.refresh_from_db()
+    assert this_inst.node_state == Instance.States.UNAVAILABLE

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

@@ -1,7 +1,7 @@
 import pytest
 from unittest.mock import MagicMock, patch
-from awx.main.tasks.system import _heartbeat_instance_management, update_inventory_computed_fields, inspect_execution_and_hop_nodes
+from awx.main.tasks.system import update_inventory_computed_fields
-from awx.main.models import Instance, Inventory
+from awx.main.models import Inventory
 from django.db import DatabaseError
 
 
@@ -62,82 +62,3 @@ 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()
-
-
-@patch('awx.main.tasks.system.get_receptor_ctl', side_effect=FileNotFoundError)
-@patch('awx.main.tasks.system.Instance.objects.filter')
-def test_heartbeat_marks_offline_when_receptor_unavailable(mock_filter, mock_get_ctl):
-    this_inst = MagicMock(spec=Instance)
-    this_inst.hostname = 'test-host'
-    mock_filter.return_value = [this_inst]
-
-    with patch('awx.main.tasks.system.settings') as mock_settings:
-        mock_settings.CLUSTER_HOST_ID = 'test-host'
-        result = _heartbeat_instance_management()
-
-    assert result == (None, None, None)
-    this_inst.local_health_check.assert_called_once()
-    this_inst.mark_offline.assert_called_once_with(errors='Receptor not available')
-
-
-class TestInspectExecutionAndHopNodes:
-    """Tests for the advisory_lock guard in inspect_execution_and_hop_nodes."""
-
-    @patch('awx.main.tasks.system.inspect_established_receptor_connections')
-    @patch('awx.main.tasks.system.advisory_lock')
-    def test_skips_when_lock_not_acquired(self, mock_lock, mock_inspect_conns):
-        mock_lock.return_value.__enter__ = MagicMock(return_value=False)
-        mock_lock.return_value.__exit__ = MagicMock(return_value=False)
-        receptor_ctl = MagicMock()
-
-        inspect_execution_and_hop_nodes([], receptor_ctl)
-
-        receptor_ctl.simple_command.assert_not_called()
-        mock_inspect_conns.assert_not_called()
-
-    @patch('awx.main.tasks.system.inspect_established_receptor_connections')
-    @patch('awx.main.tasks.system.advisory_lock')
-    def test_runs_when_lock_acquired(self, mock_lock, mock_inspect_conns):
-        mock_lock.return_value.__enter__ = MagicMock(return_value=True)
-        mock_lock.return_value.__exit__ = MagicMock(return_value=False)
-        receptor_ctl = MagicMock()
-        receptor_ctl.simple_command.return_value = {'Advertisements': [], 'KnownConnectionCosts': {}}
-
-        inspect_execution_and_hop_nodes([], receptor_ctl)
-
-        receptor_ctl.simple_command.assert_called_once_with('status')
-        mock_inspect_conns.assert_called_once()
-
-    @patch('awx.main.tasks.system.inspect_established_receptor_connections')
-    @patch('awx.main.tasks.system.advisory_lock')
-    def test_updates_last_seen_for_execution_nodes(self, mock_lock, mock_inspect_conns):
-        mock_lock.return_value.__enter__ = MagicMock(return_value=True)
-        mock_lock.return_value.__exit__ = MagicMock(return_value=False)
-
-        exec_node = MagicMock(spec=Instance)
-        exec_node.hostname = 'exec-1'
-        exec_node.node_type = 'execution'
-        exec_node.node_state = 'ready'
-        exec_node.last_seen = None
-        exec_node.capacity = 100
-        exec_node.enabled = True
-        exec_node.cpu = 4
-        exec_node.memory = 8000000000
-
-        control_node = MagicMock(spec=Instance)
-        control_node.hostname = 'control-1'
-        control_node.node_type = 'control'
-
-        receptor_ctl = MagicMock()
-        receptor_ctl.simple_command.return_value = {
-            'Advertisements': [
-                {'NodeID': 'exec-1', 'Time': '2026-01-01T00:00:00+00:00'},
-                {'NodeID': 'control-1', 'Time': '2026-01-01T00:00:00+00:00'},
-            ],
-            'KnownConnectionCosts': {},
-        }
-
-        inspect_execution_and_hop_nodes([exec_node, control_node], receptor_ctl)
-
-        exec_node.save.assert_called_once_with(update_fields=['last_seen'])
-        control_node.save.assert_not_called()