fix(core): Isolate auto-sync failures per object

An exception from any single object exited the auto_sync receiver, so
every record after it kept stale data. Each object now synchronizes in
its own transaction and all failures are reported together, with stale
records removed rather than failing the sync.

Closes: #23122

netbox/core/signals.py

@@ -4,15 +4,19 @@ from threading import local
 from django.contrib.contenttypes.models import ContentType
 from django.core.exceptions import ObjectDoesNotExist, ValidationError
 from django.core.signals import request_finished
+from django.db import transaction
 from django.db.models import CASCADE, RESTRICT
 from django.db.models.fields.reverse_related import ManyToManyRel, ManyToOneRel
 from django.db.models.signals import m2m_changed, post_migrate, post_save, pre_delete
 from django.dispatch import Signal, receiver
 from django.utils.translation import gettext_lazy as _
+from django.utils.translation import ngettext
 from django_prometheus.models import model_deletes, model_inserts, model_updates
+from rq.timeouts import JobTimeoutException
 
 from core.choices import JobStatusChoices, ObjectChangeActionChoices
 from core.events import *
+from core.exceptions import SyncError
 from core.models import ObjectType
 from extras.events import enqueue_event
 from extras.models import Tag
@@ -25,6 +29,8 @@ from utilities.exceptions import AbortRequest
 
 from .models import ConfigRevision, DataSource, ObjectChange
 
+logger = logging.getLogger('netbox.core.signals')
+
 __all__ = (
     'clear_events',
     'job_end',
@@ -299,6 +305,10 @@ def enqueue_sync_job(instance, created, **kwargs):
             job.delete()
 
 
+# Keeps the aggregated error readable when a whole source fails at once
+_AUTO_SYNC_DETAIL_LIMIT = 10
+
+
 @receiver(post_sync)
 def auto_sync(instance, **kwargs):
     """
@@ -306,8 +316,51 @@ def auto_sync(instance, **kwargs):
     """
     from .models import AutoSyncRecord
 
+    failure_count = 0
+    details = []
+    first_error = None
+
     for autosync in AutoSyncRecord.objects.filter(datafile__source=instance).prefetch_related('object'):
-        autosync.object.sync(save=True)
+        try:
+            # The try must stay outside this, so the savepoint is rolled back before the handler runs
+            with transaction.atomic():
+                if autosync.object is None:
+                    logger.warning(
+                        f"Deleting stale AutoSyncRecord for {autosync.object_type} ID {autosync.object_id}"
+                    )
+                    autosync.delete()
+                    continue
+                autosync.object.sync(save=True)
+        except JobTimeoutException:
+            # rq arms one alarm per job, so a timeout is not an ordinary per-object failure
+            raise
+        except Exception as e:
+            failure_count += 1
+            if first_error is None:
+                first_error = e
+            if autosync.object is None:
+                # Only reachable if deleting the stale record above also failed
+                target = f'{autosync.object_type} ID {autosync.object_id}'
+            else:
+                target = repr(autosync.object)
+            # Logged unconditionally, as the raised message below is capped
+            logger.error(f"Error auto-syncing {target}: {e}", exc_info=True)
+            if len(details) < _AUTO_SYNC_DETAIL_LIMIT:
+                details.append(f'- {target}: {type(e).__name__}: {e}')
+
+    if first_error is not None:
+        summary = ngettext(
+            'Automatic synchronization failed for {count} object:',
+            'Automatic synchronization failed for {count} objects:',
+            failure_count,
+        ).format(count=failure_count)
+        if omitted := failure_count - len(details):
+            details.append(ngettext(
+                'and {count} more failure.',
+                'and {count} more failures.',
+                omitted,
+            ).format(count=omitted))
+        raise SyncError('\n'.join([summary, *details])) from first_error
 
 
 @receiver(post_save, sender=ConfigRevision)

netbox/core/tests/test_signals.py

@@ -5,16 +5,19 @@ from unittest.mock import MagicMock, Mock, patch
 from django.contrib.contenttypes.models import ContentType
 from django.core.exceptions import ValidationError
 from django.core.signals import request_finished
-from django.db import transaction
+from django.db import IntegrityError, transaction
 from django.test import RequestFactory, SimpleTestCase, TestCase, override_settings
+from django.utils import timezone
+from rq.timeouts import JobTimeoutException
 
 from core import signals
 from core.choices import DataSourceStatusChoices, JobStatusChoices, ObjectChangeActionChoices
-from core.models import ConfigRevision, DataSource, ObjectChange, ObjectType
+from core.exceptions import SyncError
+from core.models import AutoSyncRecord, ConfigRevision, DataFile, DataSource, ObjectChange, ObjectType
 from core.signals import _signals_received, clear_events, post_sync
 from dcim.choices import InterfaceTypeChoices
 from dcim.models import Device, DeviceRole, DeviceType, Interface, Manufacturer, Site, SiteGroup
-from extras.models import Tag
+from extras.models import ConfigContext, Tag
 from extras.validators import CustomValidator
 from netbox.context import events_queue
 from netbox.context_managers import event_tracking
@@ -348,13 +351,16 @@ class AutoSyncSignalTestCase(TestCase):
     DataSource when post_sync fires.
     """
 
-    def test_post_sync_resyncs_dependent_records(self):
-        ds = DataSource.objects.create(
+    @classmethod
+    def setUpTestData(cls):
+        cls.datasource = DataSource.objects.create(
             name='DS 1',
             type='local',
             source_url='/tmp/ds1',
             status=DataSourceStatusChoices.COMPLETED,
         )
+
+    def test_post_sync_resyncs_dependent_records(self):
         record_a = SimpleNamespace(object=SimpleNamespace(synced=False))
         record_a.object.sync = lambda save: setattr(record_a.object, 'synced', save)
         record_b = SimpleNamespace(object=SimpleNamespace(synced=False))
@@ -365,11 +371,143 @@ class AutoSyncSignalTestCase(TestCase):
                 record_a,
                 record_b,
             ]
-            post_sync.send(sender=ds.__class__, instance=ds)
+            post_sync.send(sender=DataSource, instance=self.datasource)
 
         self.assertTrue(record_a.object.synced)
         self.assertTrue(record_b.object.synced)
 
+    def test_post_sync_continues_after_failed_record(self):
+        """A failing record leaves the connection usable for the records after it."""
+        record_a = SimpleNamespace(object=MagicMock())
+        record_b = SimpleNamespace(object=MagicMock())
+        record_a.object.__repr__ = lambda self: '<ConfigContext: Context A>'
+
+        def create_duplicate_datasource(save):
+            # Violates the unique constraint on DataSource.name
+            DataSource.objects.create(name=self.datasource.name, type='local', source_url='/tmp/duplicate')
+
+        def create_datasource(save):
+            DataSource.objects.create(name='DS 2', type='local', source_url='/tmp/ds2')
+
+        record_a.object.sync.side_effect = create_duplicate_datasource
+        record_b.object.sync.side_effect = create_datasource
+
+        with patch('core.models.AutoSyncRecord') as autosync_model:
+            autosync_model.objects.filter.return_value.prefetch_related.return_value = [record_a, record_b]
+            with self.assertLogs('netbox.core.signals', 'ERROR'), self.assertRaises(SyncError) as cm:
+                post_sync.send(sender=DataSource, instance=self.datasource)
+
+        record_a.object.sync.assert_called_once_with(save=True)
+        record_b.object.sync.assert_called_once_with(save=True)
+        self.assertIsInstance(cm.exception.__cause__, IntegrityError)
+        self.assertIn('Automatic synchronization failed for 1 object:', str(cm.exception))
+        self.assertIn('<ConfigContext: Context A>: IntegrityError:', str(cm.exception))
+        self.assertTrue(DataSource.objects.filter(name='DS 2').exists())
+
+    def test_post_sync_reports_every_failure(self):
+        """Each failed object is named in the aggregated error."""
+        record_a = SimpleNamespace(object=MagicMock())
+        record_b = SimpleNamespace(object=MagicMock())
+        record_a.object.__repr__ = lambda self: '<ConfigContext: Context A>'
+        record_b.object.__repr__ = lambda self: '<ConfigContext: Context B>'
+        record_a.object.sync.side_effect = ValueError('First failure')
+        record_b.object.sync.side_effect = RuntimeError('Second failure')
+
+        with patch('core.models.AutoSyncRecord') as autosync_model:
+            autosync_model.objects.filter.return_value.prefetch_related.return_value = [record_a, record_b]
+            with self.assertLogs('netbox.core.signals', 'ERROR'), self.assertRaises(SyncError) as cm:
+                post_sync.send(sender=DataSource, instance=self.datasource)
+
+        record_a.object.sync.assert_called_once_with(save=True)
+        record_b.object.sync.assert_called_once_with(save=True)
+        self.assertEqual(
+            str(cm.exception),
+            'Automatic synchronization failed for 2 objects:\n'
+            '- <ConfigContext: Context A>: ValueError: First failure\n'
+            '- <ConfigContext: Context B>: RuntimeError: Second failure',
+        )
+        self.assertIsInstance(cm.exception.__cause__, ValueError)
+
+    def test_post_sync_deletes_dangling_record(self):
+        """A record whose generic relation no longer resolves is removed instead of failing the sync."""
+        dangling = MagicMock(object=None)
+        record = SimpleNamespace(object=MagicMock())
+
+        with patch('core.models.AutoSyncRecord') as autosync_model:
+            autosync_model.objects.filter.return_value.prefetch_related.return_value = [dangling, record]
+            with self.assertLogs('netbox.core.signals', 'WARNING'):
+                post_sync.send(sender=DataSource, instance=self.datasource)
+
+        dangling.delete.assert_called_once_with()
+        record.object.sync.assert_called_once_with(save=True)
+
+    def test_post_sync_deletes_real_dangling_record(self):
+        """The stale record is removed from the database, exercising the unmocked queryset."""
+        datafile = DataFile.objects.create(
+            source=self.datasource,
+            path='dir1/context.yaml',
+            last_updated=timezone.now(),
+            size=1000,
+            hash='442da078f0111cbdf42f21903724f6597c692535f55bdfbbea758a1ae99ad9e1',
+            data=b'value: original',
+        )
+        AutoSyncRecord.objects.create(
+            datafile=datafile,
+            object_type=ObjectType.objects.get_for_model(ConfigContext),
+            object_id=99999,
+        )
+
+        with self.assertLogs('netbox.core.signals', 'WARNING'):
+            post_sync.send(sender=DataSource, instance=self.datasource)
+
+        self.assertFalse(AutoSyncRecord.objects.filter(datafile=datafile).exists())
+
+    def test_post_sync_reports_dangling_record_when_cleanup_is_refused(self):
+        """A stale record is reported by its target keys when its own cleanup is blocked."""
+        dangling = MagicMock(object=None, object_type='extras.configcontext', object_id=7)
+        dangling.delete.side_effect = AbortRequest('Deletion is prevented by a protection rule')
+
+        with patch('core.models.AutoSyncRecord') as autosync_model:
+            autosync_model.objects.filter.return_value.prefetch_related.return_value = [dangling]
+            with self.assertLogs('netbox.core.signals', 'ERROR'), self.assertRaises(SyncError) as cm:
+                post_sync.send(sender=DataSource, instance=self.datasource)
+
+        self.assertIn('- extras.configcontext ID 7: AbortRequest:', str(cm.exception))
+        self.assertNotIn('- None:', str(cm.exception))
+
+    def test_post_sync_caps_reported_failures(self):
+        """Only the first N failures are detailed, with the remainder counted."""
+        records = []
+        for i in range(signals._AUTO_SYNC_DETAIL_LIMIT + 3):
+            record = SimpleNamespace(object=MagicMock())
+            record.object.sync.side_effect = ValueError(f'Failure {i}')
+            records.append(record)
+
+        with patch('core.models.AutoSyncRecord') as autosync_model:
+            autosync_model.objects.filter.return_value.prefetch_related.return_value = records
+            with self.assertLogs('netbox.core.signals', 'ERROR') as logs, self.assertRaises(SyncError) as cm:
+                post_sync.send(sender=DataSource, instance=self.datasource)
+
+        message = str(cm.exception)
+        self.assertIn(f'failed for {len(records)} objects:', message)
+        self.assertEqual(message.count('ValueError: Failure'), signals._AUTO_SYNC_DETAIL_LIMIT)
+        self.assertIn('and 3 more failures.', message)
+        # Capping the message must not lose an identity, so every failure is still logged
+        self.assertEqual(len(logs.records), len(records))
+
+    def test_post_sync_propagates_job_timeout(self):
+        """An rq job timeout escapes the receiver instead of being recorded as a failure."""
+        record_a = SimpleNamespace(object=MagicMock())
+        record_b = SimpleNamespace(object=MagicMock())
+        record_a.object.sync.side_effect = JobTimeoutException('Task exceeded maximum timeout value')
+
+        with patch('core.models.AutoSyncRecord') as autosync_model:
+            autosync_model.objects.filter.return_value.prefetch_related.return_value = [record_a, record_b]
+            with self.assertRaises(JobTimeoutException):
+                post_sync.send(sender=DataSource, instance=self.datasource)
+
+        record_b.object.sync.assert_not_called()
+
 
 class UpdateConfigSignalTestCase(TestCase):
     """