Closes: #23122: Prevent an auto-sync failure from skipping subsequent objects (#23132)

netbox/core/models/data.py

@@ -6,7 +6,6 @@ from urllib.parse import urlparse
 
 import yaml
 from django.conf import settings
-from django.contrib.contenttypes.fields import GenericForeignKey
 from django.core.exceptions import ValidationError
 from django.core.validators import RegexValidator
 from django.db import models
@@ -18,6 +17,7 @@ from netbox.constants import CENSOR_TOKEN, CENSOR_TOKEN_CHANGED
 from netbox.models import PrimaryModel
 from netbox.models.features import JobsMixin
 from netbox.registry import registry
+from utilities.fields import RestrictedGenericForeignKey
 from utilities.querysets import RestrictedQuerySet
 
 from ..choices import *
@@ -396,7 +396,7 @@ class AutoSyncRecord(models.Model):
         related_name='+'
     )
     object_id = models.PositiveBigIntegerField()
-    object = GenericForeignKey(
+    object = RestrictedGenericForeignKey(
         ct_field='object_type',
         fk_field='object_id'
     )

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,56 @@ def auto_sync(instance, **kwargs):
     """
     from .models import AutoSyncRecord
 
-    for autosync in AutoSyncRecord.objects.filter(datafile__source=instance).prefetch_related('object'):
-        autosync.object.sync(save=True)
+    failure_count = 0
+    details = []
+    first_error = None
+
+    records = AutoSyncRecord.objects.filter(datafile__source=instance).order_by('pk')
+    for autosync in records.select_related('object_type').prefetch_related('object'):
+        # The object may be unresolvable or mid-failure, so identify by keys
+        target = f'{autosync.object_type.app_label}.{autosync.object_type.model} ID {autosync.object_id}'
+        if autosync.object_type.model_class() is None:
+            # Not an orphaned row, so leave it for remove_stale_contenttypes
+            logger.warning(f"Skipping AutoSyncRecord for uninstalled model {target}")
+            continue
+        try:
+            # The try must stay outside this, so the savepoint is rolled back before the handler runs
+            with transaction.atomic():
+                obj = autosync.object
+                if obj is None:
+                    # The prefetch resolves through the default manager, so recheck with the base manager
+                    try:
+                        obj = autosync.object_type.get_object_for_this_type(pk=autosync.object_id)
+                    except ObjectDoesNotExist:
+                        logger.warning(f"Deleting stale AutoSyncRecord for {target}")
+                        autosync.delete()
+                        continue
+                obj.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
+            # Not capped, unlike the raised message below
+            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(
+                '{count} additional failure is not shown.',
+                '{count} additional failures are not shown.',
+                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,28 +351,226 @@ 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,
         )
-        record_a = SimpleNamespace(object=SimpleNamespace(synced=False))
+        cls.object_type = ObjectType.objects.get_for_model(ConfigContext)
+
+    def make_record(self, object_id, obj):
+        """Stand in for an AutoSyncRecord, carrying the attributes the receiver reads."""
+        return SimpleNamespace(object=obj, object_type=self.object_type, object_id=object_id)
+
+    def patch_records(self, autosync_model, records):
+        """Point the patched manager's queryset chain at the given stand-in records."""
+        queryset = autosync_model.objects.filter.return_value.order_by.return_value.select_related.return_value
+        queryset.prefetch_related.return_value = records
+
+    def test_post_sync_resyncs_dependent_records(self):
+        record_a = self.make_record(1, SimpleNamespace(synced=False))
         record_a.object.sync = lambda save: setattr(record_a.object, 'synced', save)
-        record_b = SimpleNamespace(object=SimpleNamespace(synced=False))
+        record_b = self.make_record(2, SimpleNamespace(synced=False))
         record_b.object.sync = lambda save: setattr(record_b.object, 'synced', save)
 
         with patch('core.models.AutoSyncRecord') as autosync_model:
-            autosync_model.objects.filter.return_value.prefetch_related.return_value = [
-                record_a,
-                record_b,
-            ]
-            post_sync.send(sender=ds.__class__, instance=ds)
+            self.patch_records(autosync_model, [record_a, record_b])
+            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 = self.make_record(1, MagicMock())
+        record_b = self.make_record(2, MagicMock())
+
+        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:
+            self.patch_records(autosync_model, [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('- extras.configcontext ID 1: 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 = self.make_record(1, MagicMock())
+        record_b = self.make_record(2, MagicMock())
+        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:
+            self.patch_records(autosync_model, [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'
+            '- extras.configcontext ID 1: ValueError: First failure\n'
+            '- extras.configcontext ID 2: 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, object_type=self.object_type, object_id=7)
+        record = self.make_record(1, MagicMock())
+
+        with patch('core.models.AutoSyncRecord') as autosync_model:
+            self.patch_records(autosync_model, [dangling, record])
+            # Pin the absence, rather than relying on no ConfigContext holding this pk
+            with patch.object(self.object_type, 'get_object_for_this_type', side_effect=ConfigContext.DoesNotExist):
+                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 make_datafile(self, path='dir1/context.yaml'):
+        """Create a DataFile on the source the receiver filters by."""
+        return DataFile.objects.create(
+            source=self.datasource,
+            path=path,
+            last_updated=timezone.now(),
+            size=1000,
+            hash='442da078f0111cbdf42f21903724f6597c692535f55bdfbbea758a1ae99ad9e1',
+            data=b'value: original',
+        )
+
+    def test_post_sync_skips_record_for_uninstalled_model(self):
+        """A record whose model is no longer installed is skipped rather than deleted."""
+        # get_for_id() caches the ghost type in the manager, which the rollback does not undo
+        self.addCleanup(ContentType.objects.clear_cache)
+        stale_type = ContentType.objects.create(app_label='ghost_plugin', model='ghostmodel')
+        AutoSyncRecord.objects.create(
+            datafile=self.make_datafile('dir1/ghost.yaml'),
+            object_type=stale_type,
+            object_id=1,
+        )
+
+        with self.assertLogs('netbox.core.signals', 'WARNING') as logs:
+            post_sync.send(sender=DataSource, instance=self.datasource)
+
+        # remove_stale_contenttypes owns this cleanup and cascades to the record
+        self.assertTrue(AutoSyncRecord.objects.filter(object_type=stale_type).exists())
+        self.assertIn('ghost_plugin.ghostmodel ID 1', logs.output[0])
+
+    def test_post_sync_deletes_real_dangling_record(self):
+        """The stale record is removed from the database, exercising the unmocked queryset."""
+        datafile = self.make_datafile()
+        AutoSyncRecord.objects.create(
+            datafile=datafile,
+            object_type=ObjectType.objects.get_for_model(ConfigContext),
+            object_id=99999,
+        )
+
+        with self.assertLogs('netbox.core.signals', 'WARNING') as logs:
+            post_sync.send(sender=DataSource, instance=self.datasource)
+
+        self.assertFalse(AutoSyncRecord.objects.filter(datafile=datafile).exists())
+        # The mocked tests must not assert a format the real content type never emits
+        self.assertIn('extras.configcontext ID 99999', logs.output[0])
+
+    def test_post_sync_syncs_record_hidden_by_default_manager(self):
+        """A target the default manager excludes is synced through the base manager, not deleted."""
+        datafile = self.make_datafile('dir1/hidden.yaml')
+        context = ConfigContext.objects.create(
+            name='CC 1',
+            data={},
+            data_source=self.datasource,
+            data_file=datafile,
+            data_path=datafile.path,
+            auto_sync_enabled=True,
+        )
+        hidden = AutoSyncRecord.objects.get(object_type=self.object_type, object_id=context.pk)
+        missing = AutoSyncRecord.objects.create(
+            datafile=datafile,
+            object_type=self.object_type,
+            object_id=99999,
+        )
+
+        # The prefetch reads ConfigContext.objects, so an empty queryset hides a live target
+        with patch.object(ConfigContext, 'objects', ConfigContext.objects.none()):
+            with self.assertLogs('netbox.core.signals', 'WARNING') as logs:
+                post_sync.send(sender=DataSource, instance=self.datasource)
+
+        context.refresh_from_db()
+        self.assertEqual(context.data, {'value': 'original'})
+        self.assertTrue(context.is_synced)
+        self.assertTrue(AutoSyncRecord.objects.filter(pk=hidden.pk).exists())
+        self.assertFalse(AutoSyncRecord.objects.filter(pk=missing.pk).exists())
+        self.assertIn('extras.configcontext ID 99999', logs.output[0])
+
+    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=self.object_type, object_id=7)
+        dangling.delete.side_effect = AbortRequest('Deletion is prevented by a protection rule')
+
+        with patch('core.models.AutoSyncRecord') as autosync_model:
+            self.patch_records(autosync_model, [dangling])
+            with patch.object(self.object_type, 'get_object_for_this_type', side_effect=ConfigContext.DoesNotExist):
+                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 = self.make_record(i, MagicMock())
+            record.object.sync.side_effect = ValueError(f'Failure {i}')
+            records.append(record)
+
+        with patch('core.models.AutoSyncRecord') as autosync_model:
+            self.patch_records(autosync_model, 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('3 additional failures are not shown.', message)
+        # Capping the message must not lose an identity, so every failure is still logged
+        self.assertEqual(len(logs.records), len(records))
+        # A capped list is only deterministic if the subset is ordered
+        autosync_model.objects.filter.return_value.order_by.assert_called_once_with('pk')
+
+    def test_post_sync_propagates_job_timeout(self):
+        """An rq job timeout escapes the receiver instead of being recorded as a failure."""
+        record_a = self.make_record(1, MagicMock())
+        record_b = self.make_record(2, MagicMock())
+        record_a.object.sync.side_effect = JobTimeoutException('Task exceeded maximum timeout value')
+
+        with patch('core.models.AutoSyncRecord') as autosync_model:
+            self.patch_records(autosync_model, [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):
     """

netbox/utilities/fields.py

@@ -136,7 +136,10 @@ class RestrictedGenericForeignKey(GenericForeignKey):
             else:
                 instance = instance_dict[ct_id]
                 ct = self.get_content_type(id=ct_id, using=instance._state.db)
-                qs = ct.model_class().objects.filter(pk__in=fkeys)
+                model = ct.model_class()
+                # A model may declare its manager under a name other than `objects`
+                manager = getattr(model, 'objects', model._default_manager)
+                qs = manager.filter(pk__in=fkeys)
                 if restrict_params:
                     qs = qs.restrict(**restrict_params)
             # Carry the fetch mode of the objects being prefetched over to the objects prefetched

netbox/utilities/tests/test_prefetch.py

@@ -1,3 +1,5 @@
+from unittest.mock import PropertyMock, patch
+
 from django.db.models import FETCH_ONE, FETCH_RAISE
 
 from circuits.models import Circuit, Provider
@@ -62,3 +64,16 @@ class RestrictedGenericForeignKeyTestCase(TestCase):
 
         self.assertEqual(obj, self.provider)
         self.assertIs(obj._state.fetch_mode, FETCH_ONE)
+
+    def test_prefetch_resolves_target_without_objects_manager(self):
+        """
+        A target model which names its manager something other than `objects` resolves through
+        the default manager.
+        """
+        with patch.object(Provider, 'objects', new_callable=PropertyMock, side_effect=AttributeError):
+            cached_value = list(CachedValue.objects.prefetch_related('object'))[0]
+            # A resolved prefetch leaves nothing for the relation access to fetch
+            with self.assertNumQueries(0):
+                obj = cached_value.object
+
+        self.assertEqual(obj, self.provider)