Closes #18399: Refactor logic for marking data source syncing as queued (#19960)

netbox/core/api/views.py

@@ -1,29 +1,28 @@
 from django.http import Http404, HttpResponse
 from django.shortcuts import get_object_or_404
 from django.utils.translation import gettext_lazy as _
+from django_rq.queues import get_redis_connection
+from django_rq.settings import QUEUES_LIST
+from django_rq.utils import get_statistics
 from drf_spectacular.types import OpenApiTypes
 from drf_spectacular.utils import extend_schema
+from rest_framework import viewsets
 from rest_framework.decorators import action
 from rest_framework.exceptions import PermissionDenied
+from rest_framework.permissions import IsAdminUser
 from rest_framework.response import Response
 from rest_framework.routers import APIRootView
 from rest_framework.viewsets import ReadOnlyModelViewSet
+from rq.job import Job as RQ_Job
+from rq.worker import Worker
 
 from core import filtersets
-from core.choices import DataSourceStatusChoices
 from core.jobs import SyncDataSourceJob
 from core.models import *
 from core.utils import delete_rq_job, enqueue_rq_job, get_rq_jobs, requeue_rq_job, stop_rq_job
-from django_rq.queues import get_redis_connection
-from django_rq.utils import get_statistics
-from django_rq.settings import QUEUES_LIST
 from netbox.api.metadata import ContentTypeMetadata
 from netbox.api.pagination import LimitOffsetListPagination
 from netbox.api.viewsets import NetBoxModelViewSet, NetBoxReadOnlyModelViewSet
-from rest_framework import viewsets
-from rest_framework.permissions import IsAdminUser
-from rq.job import Job as RQ_Job
-from rq.worker import Worker
 from . import serializers
 
 
@@ -50,10 +49,8 @@ class DataSourceViewSet(NetBoxModelViewSet):
         if not request.user.has_perm('core.sync_datasource', obj=datasource):
             raise PermissionDenied(_("This user does not have permission to synchronize this data source."))
 
-        # Enqueue the sync job & update the DataSource's status
+        # Enqueue the sync job
         SyncDataSourceJob.enqueue(instance=datasource, user=request.user)
-        datasource.status = DataSourceStatusChoices.QUEUED
-        DataSource.objects.filter(pk=datasource.pk).update(status=datasource.status)
 
         serializer = serializers.DataSourceSerializer(datasource, context={'request': request})
 

netbox/core/jobs.py

@@ -21,6 +21,17 @@ class SyncDataSourceJob(JobRunner):
     class Meta:
         name = 'Synchronization'
 
+    @classmethod
+    def enqueue(cls, *args, **kwargs):
+        job = super().enqueue(*args, **kwargs)
+
+        # Update the DataSource's synchronization status to queued
+        if datasource := job.object:
+            datasource.status = DataSourceStatusChoices.QUEUED
+            DataSource.objects.filter(pk=datasource.pk).update(status=datasource.status)
+
+        return job
+
     def run(self, *args, **kwargs):
         datasource = DataSource.objects.get(pk=self.job.object_id)
 

netbox/core/signals.py

@@ -1,10 +1,12 @@
 import logging
+from threading import local
 
 from django.contrib.contenttypes.models import ContentType
 from django.core.exceptions import ValidationError
 from django.db.models.fields.reverse_related import ManyToManyRel, ManyToOneRel
 from django.db.models.signals import m2m_changed, post_save, pre_delete
 from django.dispatch import receiver, Signal
+from django.core.signals import request_finished
 from django.utils.translation import gettext_lazy as _
 from django_prometheus.models import model_deletes, model_inserts, model_updates
 
@@ -42,6 +44,10 @@ clear_events = Signal()
 # Change logging & event handling
 #
 
+# Used to track received signals per object
+_signals_received = local()
+
+
 @receiver((post_save, m2m_changed))
 def handle_changed_object(sender, instance, **kwargs):
     """
@@ -130,6 +136,16 @@ def handle_deleted_object(sender, instance, **kwargs):
     if request is None:
         return
 
+    # Check whether we've already processed a pre_delete signal for this object. (This can
+    # happen e.g. when both a parent object and its child are deleted simultaneously, due
+    # to cascading deletion.)
+    if not hasattr(_signals_received, 'pre_delete'):
+        _signals_received.pre_delete = set()
+    signature = (ContentType.objects.get_for_model(instance), instance.pk)
+    if signature in _signals_received.pre_delete:
+        return
+    _signals_received.pre_delete.add(signature)
+
     # Record an ObjectChange if applicable
     if hasattr(instance, 'to_objectchange'):
         if hasattr(instance, 'snapshot') and not getattr(instance, '_prechange_snapshot', None):
@@ -179,6 +195,14 @@ def handle_deleted_object(sender, instance, **kwargs):
     model_deletes.labels(instance._meta.model_name).inc()
 
 
+@receiver(request_finished)
+def clear_signal_history(sender, **kwargs):
+    """
+    Clear out the signals history once the request is finished.
+    """
+    _signals_received.pre_delete = set()
+
+
 @receiver(clear_events)
 def clear_events_queue(sender, **kwargs):
     """

netbox/core/tests/test_changelog.py

@@ -346,6 +346,38 @@ class ChangeLogViewTest(ModelViewTestCase):
         self.assertEqual(changes[1].changed_object_type, ContentType.objects.get_for_model(Interface))
         self.assertEqual(changes[2].changed_object_type, ContentType.objects.get_for_model(Device))
 
+    def test_duplicate_deletions(self):
+        """
+        Check that a cascading deletion event does not generate multiple "deleted" ObjectChange records for
+        the same object.
+        """
+        role1 = DeviceRole(name='Role 1', slug='role-1')
+        role1.save()
+        role2 = DeviceRole(name='Role 2', slug='role-2', parent=role1)
+        role2.save()
+        pk_list = [role1.pk, role2.pk]
+
+        # Delete both objects simultaneously
+        form_data = {
+            'pk': pk_list,
+            'confirm': True,
+            '_confirm': True,
+        }
+        request = {
+            'path': reverse('dcim:devicerole_bulk_delete'),
+            'data': post_data(form_data),
+        }
+        self.add_permissions('dcim.delete_devicerole')
+        self.assertHttpStatus(self.client.post(**request), 302)
+
+        # This should result in exactly one change record per object
+        objectchanges = ObjectChange.objects.filter(
+            changed_object_type=ContentType.objects.get_for_model(DeviceRole),
+            changed_object_id__in=pk_list,
+            action=ObjectChangeActionChoices.ACTION_DELETE
+        )
+        self.assertEqual(objectchanges.count(), 2)
+
 
 class ChangeLogAPITest(APITestCase):
 

netbox/core/views.py

@@ -33,7 +33,6 @@ from utilities.json import ConfigJSONEncoder
 from utilities.query import count_related
 from utilities.views import ContentTypePermissionRequiredMixin, GetRelatedModelsMixin, register_model_view
 from . import filtersets, forms, tables
-from .choices import DataSourceStatusChoices
 from .jobs import SyncDataSourceJob
 from .models import *
 from .plugins import get_catalog_plugins, get_local_plugins
@@ -78,12 +77,8 @@ class DataSourceSyncView(BaseObjectView):
 
     def post(self, request, pk):
         datasource = get_object_or_404(self.queryset, pk=pk)
-
-        # Enqueue the sync job & update the DataSource's status
+        # Enqueue the sync job
         job = SyncDataSourceJob.enqueue(instance=datasource, user=request.user)
-        datasource.status = DataSourceStatusChoices.QUEUED
-        DataSource.objects.filter(pk=datasource.pk).update(status=datasource.status)
-
         messages.success(
             request,
             _("Queued job #{id} to sync {datasource}").format(id=job.pk, datasource=datasource)